md-cluster.c

   1// SPDX-License-Identifier: GPL-2.0-or-later
   2/*
   3 * Copyright (C) 2015, SUSE
   4 */
   5
   6
   7#include <linux/module.h>
   8#include <linux/kthread.h>
   9#include <linux/dlm.h>
  10#include <linux/sched.h>
  11#include <linux/raid/md_p.h>
  12#include "md.h"
  13#include "md-bitmap.h"
  14#include "md-cluster.h"
  15
  16#define LVB_SIZE	64
  17#define NEW_DEV_TIMEOUT 5000
  18#define WAIT_DLM_LOCK_TIMEOUT (30 * HZ)
  19
  20struct dlm_lock_resource {
  21	dlm_lockspace_t *ls;
  22	struct dlm_lksb lksb;
  23	char *name; /* lock name. */
  24	uint32_t flags; /* flags to pass to dlm_lock() */
  25	wait_queue_head_t sync_locking; /* wait queue for synchronized locking */
  26	bool sync_locking_done;
  27	void (*bast)(void *arg, int mode); /* blocking AST function pointer*/
  28	struct mddev *mddev; /* pointing back to mddev. */
  29	int mode;
  30};
  31
  32struct resync_info {
  33	__le64 lo;
  34	__le64 hi;
  35};
  36
  37/* md_cluster_info flags */
  38#define		MD_CLUSTER_WAITING_FOR_NEWDISK		1
  39#define		MD_CLUSTER_SUSPEND_READ_BALANCING	2
  40#define		MD_CLUSTER_BEGIN_JOIN_CLUSTER		3
  41
  42/* Lock the send communication. This is done through
  43 * bit manipulation as opposed to a mutex in order to
  44 * accommodate lock and hold. See next comment.
  45 */
  46#define		MD_CLUSTER_SEND_LOCK			4
  47/* If cluster operations (such as adding a disk) must lock the
  48 * communication channel, so as to perform extra operations
  49 * (update metadata) and no other operation is allowed on the
  50 * MD. Token needs to be locked and held until the operation
  51 * completes witha md_update_sb(), which would eventually release
  52 * the lock.
  53 */
  54#define		MD_CLUSTER_SEND_LOCKED_ALREADY		5
  55/* We should receive message after node joined cluster and
  56 * set up all the related infos such as bitmap and personality */
  57#define		MD_CLUSTER_ALREADY_IN_CLUSTER		6
  58#define		MD_CLUSTER_PENDING_RECV_EVENT		7
  59#define 	MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD		8
  60#define		MD_CLUSTER_WAITING_FOR_SYNC		9
  61
  62struct md_cluster_info {
  63	struct mddev *mddev; /* the md device which md_cluster_info belongs to */
  64	/* dlm lock space and resources for clustered raid. */
  65	dlm_lockspace_t *lockspace;
  66	int slot_number;
  67	struct completion completion;
  68	struct mutex recv_mutex;
  69	struct dlm_lock_resource *bitmap_lockres;
  70	struct dlm_lock_resource **other_bitmap_lockres;
  71	struct dlm_lock_resource *resync_lockres;
  72	struct list_head suspend_list;
  73
  74	spinlock_t suspend_lock;
  75	/* record the region which write should be suspended */
  76	sector_t suspend_lo;
  77	sector_t suspend_hi;
  78	int suspend_from; /* the slot which broadcast suspend_lo/hi */
  79
  80	struct md_thread __rcu *recovery_thread;
  81	unsigned long recovery_map;
  82	/* communication loc resources */
  83	struct dlm_lock_resource *ack_lockres;
  84	struct dlm_lock_resource *message_lockres;
  85	struct dlm_lock_resource *token_lockres;
  86	struct dlm_lock_resource *no_new_dev_lockres;
  87	struct md_thread __rcu *recv_thread;
  88	struct completion newdisk_completion;
  89	wait_queue_head_t wait;
  90	unsigned long state;
  91	/* record the region in RESYNCING message */
  92	sector_t sync_low;
  93	sector_t sync_hi;
  94};
  95
  96/* For compatibility, add the new msg_type at the end. */
  97enum msg_type {
  98	METADATA_UPDATED = 0,
  99	RESYNCING,
 100	NEWDISK,
 101	REMOVE,
 102	RE_ADD,
 103	BITMAP_NEEDS_SYNC,
 104	CHANGE_CAPACITY,
 105	BITMAP_RESIZE,
 106	RESYNCING_START,
 107};
 108
 109struct cluster_msg {
 110	__le32 type;
 111	__le32 slot;
 112	/* TODO: Unionize this for smaller footprint */
 113	__le64 low;
 114	__le64 high;
 115	char uuid[16];
 116	__le32 raid_slot;
 117};
 118
 119static void sync_ast(void *arg)
 120{
 121	struct dlm_lock_resource *res;
 122
 123	res = arg;
 124	res->sync_locking_done = true;
 125	wake_up(&res->sync_locking);
 126}
 127
 128static int dlm_lock_sync(struct dlm_lock_resource *res, int mode)
 129{
 130	int ret = 0;
 131
 132	ret = dlm_lock(res->ls, mode, &res->lksb,
 133			res->flags, res->name, strlen(res->name),
 134			0, sync_ast, res, res->bast);
 135	if (ret)
 136		return ret;
 137	ret = wait_event_timeout(res->sync_locking, res->sync_locking_done,
 138				WAIT_DLM_LOCK_TIMEOUT);
 139	res->sync_locking_done = false;
 140	if (!ret) {
 141		pr_err("locking DLM '%s' timeout!\n", res->name);
 142		return -EBUSY;
 143	}
 144	if (res->lksb.sb_status == 0)
 145		res->mode = mode;
 146	return res->lksb.sb_status;
 147}
 148
 149static int dlm_unlock_sync(struct dlm_lock_resource *res)
 150{
 151	return dlm_lock_sync(res, DLM_LOCK_NL);
 152}
 153
 154/*
 155 * An variation of dlm_lock_sync, which make lock request could
 156 * be interrupted
 157 */
 158static int dlm_lock_sync_interruptible(struct dlm_lock_resource *res, int mode,
 159				       struct mddev *mddev)
 160{
 161	int ret = 0;
 162
 163	ret = dlm_lock(res->ls, mode, &res->lksb,
 164			res->flags, res->name, strlen(res->name),
 165			0, sync_ast, res, res->bast);
 166	if (ret)
 167		return ret;
 168
 169	wait_event(res->sync_locking, res->sync_locking_done
 170				      || kthread_should_stop()
 171				      || test_bit(MD_CLOSING, &mddev->flags));
 172	if (!res->sync_locking_done) {
 173		/*
 174		 * the convert queue contains the lock request when request is
 175		 * interrupted, and sync_ast could still be run, so need to
 176		 * cancel the request and reset completion
 177		 */
 178		ret = dlm_unlock(res->ls, res->lksb.sb_lkid, DLM_LKF_CANCEL,
 179			&res->lksb, res);
 180		res->sync_locking_done = false;
 181		if (unlikely(ret != 0))
 182			pr_info("failed to cancel previous lock request "
 183				 "%s return %d\n", res->name, ret);
 184		return -EPERM;
 185	} else
 186		res->sync_locking_done = false;
 187	if (res->lksb.sb_status == 0)
 188		res->mode = mode;
 189	return res->lksb.sb_status;
 190}
 191
 192static struct dlm_lock_resource *lockres_init(struct mddev *mddev,
 193		char *name, void (*bastfn)(void *arg, int mode), int with_lvb)
 194{
 195	struct dlm_lock_resource *res = NULL;
 196	int ret, namelen;
 197	struct md_cluster_info *cinfo = mddev->cluster_info;
 198
 199	res = kzalloc(sizeof(struct dlm_lock_resource), GFP_KERNEL);
 200	if (!res)
 201		return NULL;
 202	init_waitqueue_head(&res->sync_locking);
 203	res->sync_locking_done = false;
 204	res->ls = cinfo->lockspace;
 205	res->mddev = mddev;
 206	res->mode = DLM_LOCK_IV;
 207	namelen = strlen(name);
 208	res->name = kzalloc(namelen + 1, GFP_KERNEL);
 209	if (!res->name) {
 210		pr_err("md-cluster: Unable to allocate resource name for resource %s\n", name);
 211		goto out_err;
 212	}
 213	strscpy(res->name, name, namelen + 1);
 214	if (with_lvb) {
 215		res->lksb.sb_lvbptr = kzalloc(LVB_SIZE, GFP_KERNEL);
 216		if (!res->lksb.sb_lvbptr) {
 217			pr_err("md-cluster: Unable to allocate LVB for resource %s\n", name);
 218			goto out_err;
 219		}
 220		res->flags = DLM_LKF_VALBLK;
 221	}
 222
 223	if (bastfn)
 224		res->bast = bastfn;
 225
 226	res->flags |= DLM_LKF_EXPEDITE;
 227
 228	ret = dlm_lock_sync(res, DLM_LOCK_NL);
 229	if (ret) {
 230		pr_err("md-cluster: Unable to lock NL on new lock resource %s\n", name);
 231		goto out_err;
 232	}
 233	res->flags &= ~DLM_LKF_EXPEDITE;
 234	res->flags |= DLM_LKF_CONVERT;
 235
 236	return res;
 237out_err:
 238	kfree(res->lksb.sb_lvbptr);
 239	kfree(res->name);
 240	kfree(res);
 241	return NULL;
 242}
 243
 244static void lockres_free(struct dlm_lock_resource *res)
 245{
 246	int ret = 0;
 247
 248	if (!res)
 249		return;
 250
 251	/*
 252	 * use FORCEUNLOCK flag, so we can unlock even the lock is on the
 253	 * waiting or convert queue
 254	 */
 255	ret = dlm_unlock(res->ls, res->lksb.sb_lkid, DLM_LKF_FORCEUNLOCK,
 256		&res->lksb, res);
 257	if (unlikely(ret != 0))
 258		pr_err("failed to unlock %s return %d\n", res->name, ret);
 259	else
 260		wait_event(res->sync_locking, res->sync_locking_done);
 261
 262	kfree(res->name);
 263	kfree(res->lksb.sb_lvbptr);
 264	kfree(res);
 265}
 266
 267static void add_resync_info(struct dlm_lock_resource *lockres,
 268			    sector_t lo, sector_t hi)
 269{
 270	struct resync_info *ri;
 271
 272	ri = (struct resync_info *)lockres->lksb.sb_lvbptr;
 273	ri->lo = cpu_to_le64(lo);
 274	ri->hi = cpu_to_le64(hi);
 275}
 276
 277static int read_resync_info(struct mddev *mddev,
 278			    struct dlm_lock_resource *lockres)
 279{
 280	struct resync_info ri;
 281	struct md_cluster_info *cinfo = mddev->cluster_info;
 282	int ret = 0;
 283
 284	dlm_lock_sync(lockres, DLM_LOCK_CR);
 285	memcpy(&ri, lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
 286	if (le64_to_cpu(ri.hi) > 0) {
 287		cinfo->suspend_hi = le64_to_cpu(ri.hi);
 288		cinfo->suspend_lo = le64_to_cpu(ri.lo);
 289		ret = 1;
 290	}
 291	dlm_unlock_sync(lockres);
 292	return ret;
 293}
 294
 295static void recover_bitmaps(struct md_thread *thread)
 296{
 297	struct mddev *mddev = thread->mddev;
 298	struct md_cluster_info *cinfo = mddev->cluster_info;
 299	struct dlm_lock_resource *bm_lockres;
 300	char str[64];
 301	int slot, ret;
 302	sector_t lo, hi;
 303
 304	while (cinfo->recovery_map) {
 305		slot = fls64((u64)cinfo->recovery_map) - 1;
 306
 307		snprintf(str, 64, "bitmap%04d", slot);
 308		bm_lockres = lockres_init(mddev, str, NULL, 1);
 309		if (!bm_lockres) {
 310			pr_err("md-cluster: Cannot initialize bitmaps\n");
 311			goto clear_bit;
 312		}
 313
 314		ret = dlm_lock_sync_interruptible(bm_lockres, DLM_LOCK_PW, mddev);
 315		if (ret) {
 316			pr_err("md-cluster: Could not DLM lock %s: %d\n",
 317					str, ret);
 318			goto clear_bit;
 319		}
 320		ret = mddev->bitmap_ops->copy_from_slot(mddev, slot, &lo, &hi, true);
 321		if (ret) {
 322			pr_err("md-cluster: Could not copy data from bitmap %d\n", slot);
 323			goto clear_bit;
 324		}
 325
 326		/* Clear suspend_area associated with the bitmap */
 327		spin_lock_irq(&cinfo->suspend_lock);
 328		cinfo->suspend_hi = 0;
 329		cinfo->suspend_lo = 0;
 330		cinfo->suspend_from = -1;
 331		spin_unlock_irq(&cinfo->suspend_lock);
 332
 333		/* Kick off a reshape if needed */
 334		if (test_bit(MD_RESYNCING_REMOTE, &mddev->recovery) &&
 335		    test_bit(MD_RECOVERY_RESHAPE, &mddev->recovery) &&
 336		    mddev->reshape_position != MaxSector)
 337			md_wakeup_thread(mddev->sync_thread);
 338
 339		if (hi > 0) {
 340			if (lo < mddev->resync_offset)
 341				mddev->resync_offset = lo;
 342			/* wake up thread to continue resync in case resync
 343			 * is not finished */
 344			if (mddev->resync_offset != MaxSector) {
 345				/*
 346				 * clear the REMOTE flag since we will launch
 347				 * resync thread in current node.
 348				 */
 349				clear_bit(MD_RESYNCING_REMOTE,
 350					  &mddev->recovery);
 351				set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
 352				md_wakeup_thread(mddev->thread);
 353			}
 354		}
 355clear_bit:
 356		lockres_free(bm_lockres);
 357		clear_bit(slot, &cinfo->recovery_map);
 358	}
 359}
 360
 361static void recover_prep(void *arg)
 362{
 363	struct mddev *mddev = arg;
 364	struct md_cluster_info *cinfo = mddev->cluster_info;
 365	set_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state);
 366}
 367
 368static void __recover_slot(struct mddev *mddev, int slot)
 369{
 370	struct md_cluster_info *cinfo = mddev->cluster_info;
 371
 372	set_bit(slot, &cinfo->recovery_map);
 373	if (!cinfo->recovery_thread) {
 374		rcu_assign_pointer(cinfo->recovery_thread,
 375			md_register_thread(recover_bitmaps, mddev, "recover"));
 376		if (!cinfo->recovery_thread) {
 377			pr_warn("md-cluster: Could not create recovery thread\n");
 378			return;
 379		}
 380	}
 381	md_wakeup_thread(cinfo->recovery_thread);
 382}
 383
 384static void recover_slot(void *arg, struct dlm_slot *slot)
 385{
 386	struct mddev *mddev = arg;
 387	struct md_cluster_info *cinfo = mddev->cluster_info;
 388
 389	pr_info("md-cluster: %s Node %d/%d down. My slot: %d. Initiating recovery.\n",
 390			mddev->bitmap_info.cluster_name,
 391			slot->nodeid, slot->slot,
 392			cinfo->slot_number);
 393	/* deduct one since dlm slot starts from one while the num of
 394	 * cluster-md begins with 0 */
 395	__recover_slot(mddev, slot->slot - 1);
 396}
 397
 398static void recover_done(void *arg, struct dlm_slot *slots,
 399		int num_slots, int our_slot,
 400		uint32_t generation)
 401{
 402	struct mddev *mddev = arg;
 403	struct md_cluster_info *cinfo = mddev->cluster_info;
 404
 405	cinfo->slot_number = our_slot;
 406	/* completion is only need to be complete when node join cluster,
 407	 * it doesn't need to run during another node's failure */
 408	if (test_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state)) {
 409		complete(&cinfo->completion);
 410		clear_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state);
 411	}
 412	clear_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state);
 413}
 414
 415/* the ops is called when node join the cluster, and do lock recovery
 416 * if node failure occurs */
 417static const struct dlm_lockspace_ops md_ls_ops = {
 418	.recover_prep = recover_prep,
 419	.recover_slot = recover_slot,
 420	.recover_done = recover_done,
 421};
 422
 423/*
 424 * The BAST function for the ack lock resource
 425 * This function wakes up the receive thread in
 426 * order to receive and process the message.
 427 */
 428static void ack_bast(void *arg, int mode)
 429{
 430	struct dlm_lock_resource *res = arg;
 431	struct md_cluster_info *cinfo = res->mddev->cluster_info;
 432
 433	if (mode == DLM_LOCK_EX) {
 434		if (test_bit(MD_CLUSTER_ALREADY_IN_CLUSTER, &cinfo->state))
 435			md_wakeup_thread(cinfo->recv_thread);
 436		else
 437			set_bit(MD_CLUSTER_PENDING_RECV_EVENT, &cinfo->state);
 438	}
 439}
 440
 441static void remove_suspend_info(struct mddev *mddev, int slot)
 442{
 443	struct md_cluster_info *cinfo = mddev->cluster_info;
 444	mddev->pers->quiesce(mddev, 1);
 445	spin_lock_irq(&cinfo->suspend_lock);
 446	cinfo->suspend_hi = 0;
 447	cinfo->suspend_lo = 0;
 448	spin_unlock_irq(&cinfo->suspend_lock);
 449	mddev->pers->quiesce(mddev, 0);
 450}
 451
 452static void process_suspend_info(struct mddev *mddev,
 453		int slot, sector_t lo, sector_t hi)
 454{
 455	struct md_cluster_info *cinfo = mddev->cluster_info;
 456	struct mdp_superblock_1 *sb = NULL;
 457	struct md_rdev *rdev;
 458
 459	if (!hi) {
 460		/*
 461		 * clear the REMOTE flag since resync or recovery is finished
 462		 * in remote node.
 463		 */
 464		clear_bit(MD_RESYNCING_REMOTE, &mddev->recovery);
 465		remove_suspend_info(mddev, slot);
 466		set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
 467		clear_bit(MD_CLUSTER_WAITING_FOR_SYNC, &cinfo->state);
 468		md_wakeup_thread(mddev->thread);
 469		return;
 470	}
 471
 472	rdev_for_each(rdev, mddev)
 473		if (rdev->raid_disk > -1 && !test_bit(Faulty, &rdev->flags)) {
 474			sb = page_address(rdev->sb_page);
 475			break;
 476		}
 477
 478	/*
 479	 * The bitmaps are not same for different nodes
 480	 * if RESYNCING is happening in one node, then
 481	 * the node which received the RESYNCING message
 482	 * probably will perform resync with the region
 483	 * [lo, hi] again, so we could reduce resync time
 484	 * a lot if we can ensure that the bitmaps among
 485	 * different nodes are match up well.
 486	 *
 487	 * sync_low/hi is used to record the region which
 488	 * arrived in the previous RESYNCING message,
 489	 *
 490	 * Call md_bitmap_sync_with_cluster to clear NEEDED_MASK
 491	 * and set RESYNC_MASK since  resync thread is running
 492	 * in another node, so we don't need to do the resync
 493	 * again with the same section.
 494	 *
 495	 * Skip md_bitmap_sync_with_cluster in case reshape
 496	 * happening, because reshaping region is small and
 497	 * we don't want to trigger lots of WARN.
 498	 */
 499	if (sb && !(le32_to_cpu(sb->feature_map) & MD_FEATURE_RESHAPE_ACTIVE))
 500		mddev->bitmap_ops->sync_with_cluster(mddev, cinfo->sync_low,
 501						     cinfo->sync_hi, lo, hi);
 502	cinfo->sync_low = lo;
 503	cinfo->sync_hi = hi;
 504
 505	mddev->pers->quiesce(mddev, 1);
 506	spin_lock_irq(&cinfo->suspend_lock);
 507	cinfo->suspend_from = slot;
 508	cinfo->suspend_lo = lo;
 509	cinfo->suspend_hi = hi;
 510	spin_unlock_irq(&cinfo->suspend_lock);
 511	mddev->pers->quiesce(mddev, 0);
 512}
 513
 514static int process_add_new_disk(struct mddev *mddev, struct cluster_msg *cmsg)
 515{
 516	char disk_uuid[64];
 517	struct md_cluster_info *cinfo = mddev->cluster_info;
 518	char event_name[] = "EVENT=ADD_DEVICE";
 519	char raid_slot[16];
 520	char *envp[] = {event_name, disk_uuid, raid_slot, NULL};
 521	int len;
 522	int res = 0;
 523
 524	len = snprintf(disk_uuid, 64, "DEVICE_UUID=");
 525	sprintf(disk_uuid + len, "%pU", cmsg->uuid);
 526	snprintf(raid_slot, 16, "RAID_DISK=%d", le32_to_cpu(cmsg->raid_slot));
 527	pr_info("%s:%d Sending kobject change with %s and %s\n", __func__, __LINE__, disk_uuid, raid_slot);
 528	init_completion(&cinfo->newdisk_completion);
 529	set_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state);
 530	kobject_uevent_env(&disk_to_dev(mddev->gendisk)->kobj, KOBJ_CHANGE, envp);
 531	if (!wait_for_completion_timeout(&cinfo->newdisk_completion,
 532					NEW_DEV_TIMEOUT)) {
 533		pr_err("md-cluster(%s:%d): timeout on a new disk adding\n",
 534			__func__, __LINE__);
 535		res = -1;
 536	}
 537	clear_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state);
 538	set_bit(MD_CLUSTER_WAITING_FOR_SYNC, &cinfo->state);
 539	return res;
 540}
 541
 542
 543static void process_metadata_update(struct mddev *mddev, struct cluster_msg *msg)
 544{
 545	int got_lock = 0;
 546	struct md_thread *thread;
 547	struct md_cluster_info *cinfo = mddev->cluster_info;
 548	mddev->good_device_nr = le32_to_cpu(msg->raid_slot);
 549
 550	dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR);
 551
 552	thread = rcu_dereference_protected(mddev->thread, true);
 553	if (!thread) {
 554		pr_warn("md-cluster: Received metadata update but MD thread is not ready\n");
 555		dlm_unlock_sync(cinfo->no_new_dev_lockres);
 556		return;
 557	}
 558
 559	wait_event(thread->wqueue,
 560		   (got_lock = mddev_trylock(mddev)) ||
 561		    test_bit(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state));
 562	md_reload_sb(mddev, mddev->good_device_nr);
 563	if (got_lock)
 564		mddev_unlock(mddev);
 565}
 566
 567static void process_remove_disk(struct mddev *mddev, struct cluster_msg *msg)
 568{
 569	struct md_rdev *rdev;
 570
 571	rcu_read_lock();
 572	rdev = md_find_rdev_nr_rcu(mddev, le32_to_cpu(msg->raid_slot));
 573	if (rdev) {
 574		set_bit(ClusterRemove, &rdev->flags);
 575		set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
 576		md_wakeup_thread(mddev->thread);
 577	}
 578	else
 579		pr_warn("%s: %d Could not find disk(%d) to REMOVE\n",
 580			__func__, __LINE__, le32_to_cpu(msg->raid_slot));
 581	rcu_read_unlock();
 582}
 583
 584static void process_readd_disk(struct mddev *mddev, struct cluster_msg *msg)
 585{
 586	struct md_rdev *rdev;
 587
 588	rcu_read_lock();
 589	rdev = md_find_rdev_nr_rcu(mddev, le32_to_cpu(msg->raid_slot));
 590	if (rdev && test_bit(Faulty, &rdev->flags))
 591		clear_bit(Faulty, &rdev->flags);
 592	else
 593		pr_warn("%s: %d Could not find disk(%d) which is faulty",
 594			__func__, __LINE__, le32_to_cpu(msg->raid_slot));
 595	rcu_read_unlock();
 596}
 597
 598static int process_recvd_msg(struct mddev *mddev, struct cluster_msg *msg)
 599{
 600	int ret = 0;
 601
 602	if (WARN(mddev->cluster_info->slot_number - 1 == le32_to_cpu(msg->slot),
 603		"node %d received its own msg\n", le32_to_cpu(msg->slot)))
 604		return -1;
 605	switch (le32_to_cpu(msg->type)) {
 606	case METADATA_UPDATED:
 607		process_metadata_update(mddev, msg);
 608		break;
 609	case CHANGE_CAPACITY:
 610		set_capacity_and_notify(mddev->gendisk, mddev->array_sectors);
 611		break;
 612	case RESYNCING_START:
 613		clear_bit(MD_CLUSTER_WAITING_FOR_SYNC, &mddev->cluster_info->state);
 614		break;
 615	case RESYNCING:
 616		set_bit(MD_RESYNCING_REMOTE, &mddev->recovery);
 617		process_suspend_info(mddev, le32_to_cpu(msg->slot),
 618				     le64_to_cpu(msg->low),
 619				     le64_to_cpu(msg->high));
 620		break;
 621	case NEWDISK:
 622		if (process_add_new_disk(mddev, msg))
 623			ret = -1;
 624		break;
 625	case REMOVE:
 626		process_remove_disk(mddev, msg);
 627		break;
 628	case RE_ADD:
 629		process_readd_disk(mddev, msg);
 630		break;
 631	case BITMAP_NEEDS_SYNC:
 632		__recover_slot(mddev, le32_to_cpu(msg->slot));
 633		break;
 634	case BITMAP_RESIZE:
 635		if (le64_to_cpu(msg->high) != mddev->pers->size(mddev, 0, 0))
 636			ret = mddev->bitmap_ops->resize(mddev,
 637							le64_to_cpu(msg->high),
 638							0);
 639		break;
 640	default:
 641		ret = -1;
 642		pr_warn("%s:%d Received unknown message from %d\n",
 643			__func__, __LINE__, msg->slot);
 644	}
 645	return ret;
 646}
 647
 648/*
 649 * thread for receiving message
 650 */
 651static void recv_daemon(struct md_thread *thread)
 652{
 653	struct md_cluster_info *cinfo = thread->mddev->cluster_info;
 654	struct dlm_lock_resource *ack_lockres = cinfo->ack_lockres;
 655	struct dlm_lock_resource *message_lockres = cinfo->message_lockres;
 656	struct cluster_msg msg;
 657	int ret;
 658
 659	mutex_lock(&cinfo->recv_mutex);
 660	/*get CR on Message*/
 661	if (dlm_lock_sync(message_lockres, DLM_LOCK_CR)) {
 662		pr_err("md/raid1:failed to get CR on MESSAGE\n");
 663		mutex_unlock(&cinfo->recv_mutex);
 664		return;
 665	}
 666
 667	/* read lvb and wake up thread to process this message_lockres */
 668	memcpy(&msg, message_lockres->lksb.sb_lvbptr, sizeof(struct cluster_msg));
 669	ret = process_recvd_msg(thread->mddev, &msg);
 670	if (ret)
 671		goto out;
 672
 673	/*release CR on ack_lockres*/
 674	ret = dlm_unlock_sync(ack_lockres);
 675	if (unlikely(ret != 0))
 676		pr_info("unlock ack failed return %d\n", ret);
 677	/*up-convert to PR on message_lockres*/
 678	ret = dlm_lock_sync(message_lockres, DLM_LOCK_PR);
 679	if (unlikely(ret != 0))
 680		pr_info("lock PR on msg failed return %d\n", ret);
 681	/*get CR on ack_lockres again*/
 682	ret = dlm_lock_sync(ack_lockres, DLM_LOCK_CR);
 683	if (unlikely(ret != 0))
 684		pr_info("lock CR on ack failed return %d\n", ret);
 685out:
 686	/*release CR on message_lockres*/
 687	ret = dlm_unlock_sync(message_lockres);
 688	if (unlikely(ret != 0))
 689		pr_info("unlock msg failed return %d\n", ret);
 690	mutex_unlock(&cinfo->recv_mutex);
 691}
 692
 693/* lock_token()
 694 * Takes the lock on the TOKEN lock resource so no other
 695 * node can communicate while the operation is underway.
 696 */
 697static int lock_token(struct md_cluster_info *cinfo)
 698{
 699	int error;
 700
 701	error = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
 702	if (error) {
 703		pr_err("md-cluster(%s:%d): failed to get EX on TOKEN (%d)\n",
 704				__func__, __LINE__, error);
 705	} else {
 706		/* Lock the receive sequence */
 707		mutex_lock(&cinfo->recv_mutex);
 708	}
 709	return error;
 710}
 711
 712/* lock_comm()
 713 * Sets the MD_CLUSTER_SEND_LOCK bit to lock the send channel.
 714 */
 715static int lock_comm(struct md_cluster_info *cinfo, bool mddev_locked)
 716{
 717	int rv, set_bit = 0;
 718	struct mddev *mddev = cinfo->mddev;
 719
 720	/*
 721	 * If resync thread run after raid1d thread, then process_metadata_update
 722	 * could not continue if raid1d held reconfig_mutex (and raid1d is blocked
 723	 * since another node already got EX on Token and waiting the EX of Ack),
 724	 * so let resync wake up thread in case flag is set.
 725	 */
 726	if (mddev_locked && !test_bit(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD,
 727				      &cinfo->state)) {
 728		rv = test_and_set_bit_lock(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD,
 729					      &cinfo->state);
 730		WARN_ON_ONCE(rv);
 731		md_wakeup_thread(mddev->thread);
 732		set_bit = 1;
 733	}
 734
 735	wait_event(cinfo->wait,
 736		   !test_and_set_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state));
 737	rv = lock_token(cinfo);
 738	if (set_bit)
 739		clear_bit_unlock(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state);
 740	return rv;
 741}
 742
 743static void unlock_comm(struct md_cluster_info *cinfo)
 744{
 745	WARN_ON(cinfo->token_lockres->mode != DLM_LOCK_EX);
 746	mutex_unlock(&cinfo->recv_mutex);
 747	dlm_unlock_sync(cinfo->token_lockres);
 748	clear_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state);
 749	wake_up(&cinfo->wait);
 750}
 751
 752/* __sendmsg()
 753 * This function performs the actual sending of the message. This function is
 754 * usually called after performing the encompassing operation
 755 * The function:
 756 * 1. Grabs the message lockresource in EX mode
 757 * 2. Copies the message to the message LVB
 758 * 3. Downconverts message lockresource to CW
 759 * 4. Upconverts ack lock resource from CR to EX. This forces the BAST on other nodes
 760 *    and the other nodes read the message. The thread will wait here until all other
 761 *    nodes have released ack lock resource.
 762 * 5. Downconvert ack lockresource to CR
 763 */
 764static int __sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
 765{
 766	int error, unlock_error;
 767	int slot = cinfo->slot_number - 1;
 768
 769	cmsg->slot = cpu_to_le32(slot);
 770	/*get EX on Message*/
 771	error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_EX);
 772	if (error) {
 773		pr_err("md-cluster: failed to get EX on MESSAGE (%d)\n", error);
 774		return error;
 775	}
 776
 777	memcpy(cinfo->message_lockres->lksb.sb_lvbptr, (void *)cmsg,
 778			sizeof(struct cluster_msg));
 779	/*down-convert EX to CW on Message*/
 780	error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_CW);
 781	if (error) {
 782		pr_err("md-cluster: failed to convert EX to CW on MESSAGE(%d)\n",
 783				error);
 784		goto failed_ack;
 785	}
 786
 787	/*up-convert CR to EX on Ack*/
 788	error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_EX);
 789	if (error) {
 790		pr_err("md-cluster: failed to convert CR to EX on ACK(%d)\n",
 791				error);
 792		goto failed_ack;
 793	}
 794
 795	/*down-convert EX to CR on Ack*/
 796	error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR);
 797	if (error) {
 798		pr_err("md-cluster: failed to convert EX to CR on ACK(%d)\n",
 799				error);
 800		goto failed_ack;
 801	}
 802
 803failed_ack:
 804	while ((unlock_error = dlm_unlock_sync(cinfo->message_lockres)))
 805		pr_err("md-cluster: failed convert to NL on MESSAGE(%d)\n",
 806			unlock_error);
 807
 808	return error;
 809}
 810
 811static int sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg,
 812		   bool mddev_locked)
 813{
 814	int ret;
 815
 816	ret = lock_comm(cinfo, mddev_locked);
 817	if (!ret) {
 818		ret = __sendmsg(cinfo, cmsg);
 819		unlock_comm(cinfo);
 820	}
 821	return ret;
 822}
 823
 824static int gather_all_resync_info(struct mddev *mddev, int total_slots)
 825{
 826	struct md_cluster_info *cinfo = mddev->cluster_info;
 827	int i, ret = 0;
 828	struct dlm_lock_resource *bm_lockres;
 829	char str[64];
 830	sector_t lo, hi;
 831
 832
 833	for (i = 0; i < total_slots; i++) {
 834		memset(str, '\0', 64);
 835		snprintf(str, 64, "bitmap%04d", i);
 836		bm_lockres = lockres_init(mddev, str, NULL, 1);
 837		if (!bm_lockres)
 838			return -ENOMEM;
 839		if (i == (cinfo->slot_number - 1)) {
 840			lockres_free(bm_lockres);
 841			continue;
 842		}
 843
 844		bm_lockres->flags |= DLM_LKF_NOQUEUE;
 845		ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
 846		if (ret == -EAGAIN) {
 847			if (read_resync_info(mddev, bm_lockres)) {
 848				pr_info("%s:%d Resync[%llu..%llu] in progress on %d\n",
 849						__func__, __LINE__,
 850					(unsigned long long) cinfo->suspend_lo,
 851					(unsigned long long) cinfo->suspend_hi,
 852					i);
 853				cinfo->suspend_from = i;
 854			}
 855			ret = 0;
 856			lockres_free(bm_lockres);
 857			continue;
 858		}
 859		if (ret) {
 860			lockres_free(bm_lockres);
 861			goto out;
 862		}
 863
 864		/* Read the disk bitmap sb and check if it needs recovery */
 865		ret = mddev->bitmap_ops->copy_from_slot(mddev, i, &lo, &hi, false);
 866		if (ret) {
 867			pr_warn("md-cluster: Could not gather bitmaps from slot %d", i);
 868			lockres_free(bm_lockres);
 869			continue;
 870		}
 871		if ((hi > 0) && (lo < mddev->resync_offset)) {
 872			set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
 873			mddev->resync_offset = lo;
 874			md_check_recovery(mddev);
 875		}
 876
 877		lockres_free(bm_lockres);
 878	}
 879out:
 880	return ret;
 881}
 882
 883static int join(struct mddev *mddev, int nodes)
 884{
 885	struct md_cluster_info *cinfo;
 886	int ret, ops_rv;
 887	char str[64];
 888
 889	cinfo = kzalloc(sizeof(struct md_cluster_info), GFP_KERNEL);
 890	if (!cinfo)
 891		return -ENOMEM;
 892
 893	INIT_LIST_HEAD(&cinfo->suspend_list);
 894	spin_lock_init(&cinfo->suspend_lock);
 895	init_completion(&cinfo->completion);
 896	set_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state);
 897	init_waitqueue_head(&cinfo->wait);
 898	mutex_init(&cinfo->recv_mutex);
 899
 900	mddev->cluster_info = cinfo;
 901	cinfo->mddev = mddev;
 902
 903	memset(str, 0, 64);
 904	sprintf(str, "%pU", mddev->uuid);
 905	ret = dlm_new_lockspace(str, mddev->bitmap_info.cluster_name,
 906				DLM_LSFL_SOFTIRQ, LVB_SIZE, &md_ls_ops, mddev,
 907				&ops_rv, &cinfo->lockspace);
 908	if (ret)
 909		goto err;
 910	wait_for_completion(&cinfo->completion);
 911	if (nodes < cinfo->slot_number) {
 912		pr_err("md-cluster: Slot allotted(%d) is greater than available slots(%d).",
 913			cinfo->slot_number, nodes);
 914		ret = -ERANGE;
 915		goto err;
 916	}
 917	/* Initiate the communication resources */
 918	ret = -ENOMEM;
 919	rcu_assign_pointer(cinfo->recv_thread,
 920			md_register_thread(recv_daemon, mddev, "cluster_recv"));
 921	if (!cinfo->recv_thread) {
 922		pr_err("md-cluster: cannot allocate memory for recv_thread!\n");
 923		goto err;
 924	}
 925	cinfo->message_lockres = lockres_init(mddev, "message", NULL, 1);
 926	if (!cinfo->message_lockres)
 927		goto err;
 928	cinfo->token_lockres = lockres_init(mddev, "token", NULL, 0);
 929	if (!cinfo->token_lockres)
 930		goto err;
 931	cinfo->no_new_dev_lockres = lockres_init(mddev, "no-new-dev", NULL, 0);
 932	if (!cinfo->no_new_dev_lockres)
 933		goto err;
 934
 935	ret = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
 936	if (ret) {
 937		ret = -EAGAIN;
 938		pr_err("md-cluster: can't join cluster to avoid lock issue\n");
 939		goto err;
 940	}
 941	cinfo->ack_lockres = lockres_init(mddev, "ack", ack_bast, 0);
 942	if (!cinfo->ack_lockres) {
 943		ret = -ENOMEM;
 944		goto err;
 945	}
 946	/* get sync CR lock on ACK. */
 947	if (dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR))
 948		pr_err("md-cluster: failed to get a sync CR lock on ACK!(%d)\n",
 949				ret);
 950	dlm_unlock_sync(cinfo->token_lockres);
 951	/* get sync CR lock on no-new-dev. */
 952	if (dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR))
 953		pr_err("md-cluster: failed to get a sync CR lock on no-new-dev!(%d)\n", ret);
 954
 955
 956	pr_info("md-cluster: Joined cluster %s slot %d\n", str, cinfo->slot_number);
 957	snprintf(str, 64, "bitmap%04d", cinfo->slot_number - 1);
 958	cinfo->bitmap_lockres = lockres_init(mddev, str, NULL, 1);
 959	if (!cinfo->bitmap_lockres) {
 960		ret = -ENOMEM;
 961		goto err;
 962	}
 963	if (dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW)) {
 964		pr_err("Failed to get bitmap lock\n");
 965		ret = -EINVAL;
 966		goto err;
 967	}
 968
 969	cinfo->resync_lockres = lockres_init(mddev, "resync", NULL, 0);
 970	if (!cinfo->resync_lockres) {
 971		ret = -ENOMEM;
 972		goto err;
 973	}
 974
 975	return 0;
 976err:
 977	set_bit(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state);
 978	md_unregister_thread(mddev, &cinfo->recovery_thread);
 979	md_unregister_thread(mddev, &cinfo->recv_thread);
 980	lockres_free(cinfo->message_lockres);
 981	lockres_free(cinfo->token_lockres);
 982	lockres_free(cinfo->ack_lockres);
 983	lockres_free(cinfo->no_new_dev_lockres);
 984	lockres_free(cinfo->resync_lockres);
 985	lockres_free(cinfo->bitmap_lockres);
 986	if (cinfo->lockspace)
 987		dlm_release_lockspace(cinfo->lockspace, DLM_RELEASE_NORMAL);
 988	mddev->cluster_info = NULL;
 989	kfree(cinfo);
 990	return ret;
 991}
 992
 993static void load_bitmaps(struct mddev *mddev, int total_slots)
 994{
 995	struct md_cluster_info *cinfo = mddev->cluster_info;
 996
 997	/* load all the node's bitmap info for resync */
 998	if (gather_all_resync_info(mddev, total_slots))
 999		pr_err("md-cluster: failed to gather all resyn infos\n");
1000	set_bit(MD_CLUSTER_ALREADY_IN_CLUSTER, &cinfo->state);
1001	/* wake up recv thread in case something need to be handled */
1002	if (test_and_clear_bit(MD_CLUSTER_PENDING_RECV_EVENT, &cinfo->state))
1003		md_wakeup_thread(cinfo->recv_thread);
1004}
1005
1006static void resync_bitmap(struct mddev *mddev)
1007{
1008	struct md_cluster_info *cinfo = mddev->cluster_info;
1009	struct cluster_msg cmsg = {0};
1010	int err;
1011
1012	cmsg.type = cpu_to_le32(BITMAP_NEEDS_SYNC);
1013	err = sendmsg(cinfo, &cmsg, 1);
1014	if (err)
1015		pr_err("%s:%d: failed to send BITMAP_NEEDS_SYNC message (%d)\n",
1016			__func__, __LINE__, err);
1017}
1018
1019static void unlock_all_bitmaps(struct mddev *mddev);
1020static int leave(struct mddev *mddev)
1021{
1022	struct md_cluster_info *cinfo = mddev->cluster_info;
1023
1024	if (!cinfo)
1025		return 0;
1026
1027	/*
1028	 * BITMAP_NEEDS_SYNC message should be sent when node
1029	 * is leaving the cluster with dirty bitmap, also we
1030	 * can only deliver it when dlm connection is available.
1031	 *
1032	 * Also, we should send BITMAP_NEEDS_SYNC message in
1033	 * case reshaping is interrupted.
1034	 */
1035	if ((cinfo->slot_number > 0 && mddev->resync_offset != MaxSector) ||
1036	    (mddev->reshape_position != MaxSector &&
1037	     test_bit(MD_CLOSING, &mddev->flags)))
1038		resync_bitmap(mddev);
1039
1040	set_bit(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state);
1041	md_unregister_thread(mddev, &cinfo->recovery_thread);
1042	md_unregister_thread(mddev, &cinfo->recv_thread);
1043	lockres_free(cinfo->message_lockres);
1044	lockres_free(cinfo->token_lockres);
1045	lockres_free(cinfo->ack_lockres);
1046	lockres_free(cinfo->no_new_dev_lockres);
1047	lockres_free(cinfo->resync_lockres);
1048	lockres_free(cinfo->bitmap_lockres);
1049	unlock_all_bitmaps(mddev);
1050	dlm_release_lockspace(cinfo->lockspace, DLM_RELEASE_NORMAL);
1051	kfree(cinfo);
1052	return 0;
1053}
1054
1055/* slot_number(): Returns the MD slot number to use
1056 * DLM starts the slot numbers from 1, wheras cluster-md
1057 * wants the number to be from zero, so we deduct one
1058 */
1059static int slot_number(struct mddev *mddev)
1060{
1061	struct md_cluster_info *cinfo = mddev->cluster_info;
1062
1063	return cinfo->slot_number - 1;
1064}
1065
1066/*
1067 * Check if the communication is already locked, else lock the communication
1068 * channel.
1069 * If it is already locked, token is in EX mode, and hence lock_token()
1070 * should not be called.
1071 */
1072static int metadata_update_start(struct mddev *mddev)
1073{
1074	struct md_cluster_info *cinfo = mddev->cluster_info;
1075	int ret;
1076
1077	/*
1078	 * metadata_update_start is always called with the protection of
1079	 * reconfig_mutex, so set WAITING_FOR_TOKEN here.
1080	 */
1081	ret = test_and_set_bit_lock(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD,
1082				    &cinfo->state);
1083	WARN_ON_ONCE(ret);
1084	md_wakeup_thread(mddev->thread);
1085
1086	wait_event(cinfo->wait,
1087		   !test_and_set_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state) ||
1088		   test_and_clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state));
1089
1090	/* If token is already locked, return 0 */
1091	if (cinfo->token_lockres->mode == DLM_LOCK_EX) {
1092		clear_bit_unlock(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state);
1093		return 0;
1094	}
1095
1096	ret = lock_token(cinfo);
1097	clear_bit_unlock(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state);
1098	return ret;
1099}
1100
1101static int metadata_update_finish(struct mddev *mddev)
1102{
1103	struct md_cluster_info *cinfo = mddev->cluster_info;
1104	struct cluster_msg cmsg;
1105	struct md_rdev *rdev;
1106	int ret = 0;
1107	int raid_slot = -1;
1108
1109	memset(&cmsg, 0, sizeof(cmsg));
1110	cmsg.type = cpu_to_le32(METADATA_UPDATED);
1111	/* Pick up a good active device number to send.
1112	 */
1113	rdev_for_each(rdev, mddev)
1114		if (rdev->raid_disk > -1 && !test_bit(Faulty, &rdev->flags)) {
1115			raid_slot = rdev->desc_nr;
1116			break;
1117		}
1118	if (raid_slot >= 0) {
1119		cmsg.raid_slot = cpu_to_le32(raid_slot);
1120		ret = __sendmsg(cinfo, &cmsg);
1121	} else
1122		pr_warn("md-cluster: No good device id found to send\n");
1123	clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
1124	unlock_comm(cinfo);
1125	return ret;
1126}
1127
1128static void metadata_update_cancel(struct mddev *mddev)
1129{
1130	struct md_cluster_info *cinfo = mddev->cluster_info;
1131	clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
1132	unlock_comm(cinfo);
1133}
1134
1135static int update_bitmap_size(struct mddev *mddev, sector_t size)
1136{
1137	struct md_cluster_info *cinfo = mddev->cluster_info;
1138	struct cluster_msg cmsg = {0};
1139	int ret;
1140
1141	cmsg.type = cpu_to_le32(BITMAP_RESIZE);
1142	cmsg.high = cpu_to_le64(size);
1143	ret = sendmsg(cinfo, &cmsg, 0);
1144	if (ret)
1145		pr_err("%s:%d: failed to send BITMAP_RESIZE message (%d)\n",
1146			__func__, __LINE__, ret);
1147	return ret;
1148}
1149
1150static int resize_bitmaps(struct mddev *mddev, sector_t newsize, sector_t oldsize)
1151{
1152	void *bitmap = mddev->bitmap;
1153	struct md_bitmap_stats stats;
1154	unsigned long my_pages;
1155	int i, rv;
1156
1157	rv = mddev->bitmap_ops->get_stats(bitmap, &stats);
1158	if (rv)
1159		return rv;
1160
1161	my_pages = stats.pages;
1162	/*
1163	 * We need to ensure all the nodes can grow to a larger
1164	 * bitmap size before make the reshaping.
1165	 */
1166	rv = update_bitmap_size(mddev, newsize);
1167	if (rv)
1168		return rv;
1169
1170	for (i = 0; i < mddev->bitmap_info.nodes; i++) {
1171		struct dlm_lock_resource *bm_lockres;
1172		char str[64];
1173
1174		if (i == slot_number(mddev))
1175			continue;
1176
1177		bitmap = mddev->bitmap_ops->get_from_slot(mddev, i);
1178		if (IS_ERR(bitmap)) {
1179			pr_err("can't get bitmap from slot %d\n", i);
1180			bitmap = NULL;
1181			goto out;
1182		}
1183
1184		rv = mddev->bitmap_ops->get_stats(bitmap, &stats);
1185		if (rv)
1186			goto out;
1187		/*
1188		 * If we can hold the bitmap lock of one node then
1189		 * the slot is not occupied, update the pages.
1190		 */
1191		snprintf(str, 64, "bitmap%04d", i);
1192		bm_lockres = lockres_init(mddev, str, NULL, 1);
1193		if (!bm_lockres) {
1194			pr_err("Cannot initialize %s lock\n", str);
1195			goto out;
1196		}
1197		bm_lockres->flags |= DLM_LKF_NOQUEUE;
1198		rv = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
1199		if (!rv)
1200			mddev->bitmap_ops->set_pages(bitmap, my_pages);
1201		lockres_free(bm_lockres);
1202
1203		if (my_pages != stats.pages)
1204			/*
1205			 * Let's revert the bitmap size if one node
1206			 * can't resize bitmap
1207			 */
1208			goto out;
1209		mddev->bitmap_ops->free(bitmap);
1210	}
1211
1212	return 0;
1213out:
1214	mddev->bitmap_ops->free(bitmap);
1215	update_bitmap_size(mddev, oldsize);
1216	return -1;
1217}
1218
1219/*
1220 * return 0 if all the bitmaps have the same sync_size
1221 */
1222static int cluster_check_sync_size(struct mddev *mddev)
1223{
1224	int current_slot = slot_number(mddev);
1225	int node_num = mddev->bitmap_info.nodes;
1226	struct dlm_lock_resource *bm_lockres;
1227	struct md_bitmap_stats stats;
1228	void *bitmap = mddev->bitmap;
1229	unsigned long sync_size = 0;
1230	unsigned long my_sync_size;
1231	char str[64];
1232	int i, rv;
1233
1234	rv = mddev->bitmap_ops->get_stats(bitmap, &stats);
1235	if (rv)
1236		return rv;
1237
1238	my_sync_size = stats.sync_size;
1239
1240	for (i = 0; i < node_num; i++) {
1241		if (i == current_slot)
1242			continue;
1243
1244		bitmap = mddev->bitmap_ops->get_from_slot(mddev, i);
1245		if (IS_ERR(bitmap)) {
1246			pr_err("can't get bitmap from slot %d\n", i);
1247			return -1;
1248		}
1249
1250		/*
1251		 * If we can hold the bitmap lock of one node then
1252		 * the slot is not occupied, update the sb.
1253		 */
1254		snprintf(str, 64, "bitmap%04d", i);
1255		bm_lockres = lockres_init(mddev, str, NULL, 1);
1256		if (!bm_lockres) {
1257			pr_err("md-cluster: Cannot initialize %s\n", str);
1258			mddev->bitmap_ops->free(bitmap);
1259			return -1;
1260		}
1261		bm_lockres->flags |= DLM_LKF_NOQUEUE;
1262		rv = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
1263		if (!rv)
1264			mddev->bitmap_ops->update_sb(bitmap);
1265		lockres_free(bm_lockres);
1266
1267		rv = mddev->bitmap_ops->get_stats(bitmap, &stats);
1268		if (rv) {
1269			mddev->bitmap_ops->free(bitmap);
1270			return rv;
1271		}
1272
1273		if (sync_size == 0) {
1274			sync_size = stats.sync_size;
1275		} else if (sync_size != stats.sync_size) {
1276			mddev->bitmap_ops->free(bitmap);
1277			return -1;
1278		}
1279		mddev->bitmap_ops->free(bitmap);
1280	}
1281
1282	return (my_sync_size == sync_size) ? 0 : -1;
1283}
1284
1285/*
1286 * Update the size for cluster raid is a little more complex, we perform it
1287 * by the steps:
1288 * 1. hold token lock and update superblock in initiator node.
1289 * 2. send METADATA_UPDATED msg to other nodes.
1290 * 3. The initiator node continues to check each bitmap's sync_size, if all
1291 *    bitmaps have the same value of sync_size, then we can set capacity and
1292 *    let other nodes to perform it. If one node can't update sync_size
1293 *    accordingly, we need to revert to previous value.
1294 */
1295static void update_size(struct mddev *mddev, sector_t old_dev_sectors)
1296{
1297	struct md_cluster_info *cinfo = mddev->cluster_info;
1298	struct cluster_msg cmsg;
1299	struct md_rdev *rdev;
1300	int ret = 0;
1301	int raid_slot = -1;
1302
1303	md_update_sb(mddev, 1);
1304	if (lock_comm(cinfo, 1)) {
1305		pr_err("%s: lock_comm failed\n", __func__);
1306		return;
1307	}
1308
1309	memset(&cmsg, 0, sizeof(cmsg));
1310	cmsg.type = cpu_to_le32(METADATA_UPDATED);
1311	rdev_for_each(rdev, mddev)
1312		if (rdev->raid_disk >= 0 && !test_bit(Faulty, &rdev->flags)) {
1313			raid_slot = rdev->desc_nr;
1314			break;
1315		}
1316	if (raid_slot >= 0) {
1317		cmsg.raid_slot = cpu_to_le32(raid_slot);
1318		/*
1319		 * We can only change capiticy after all the nodes can do it,
1320		 * so need to wait after other nodes already received the msg
1321		 * and handled the change
1322		 */
1323		ret = __sendmsg(cinfo, &cmsg);
1324		if (ret) {
1325			pr_err("%s:%d: failed to send METADATA_UPDATED msg\n",
1326			       __func__, __LINE__);
1327			unlock_comm(cinfo);
1328			return;
1329		}
1330	} else {
1331		pr_err("md-cluster: No good device id found to send\n");
1332		unlock_comm(cinfo);
1333		return;
1334	}
1335
1336	/*
1337	 * check the sync_size from other node's bitmap, if sync_size
1338	 * have already updated in other nodes as expected, send an
1339	 * empty metadata msg to permit the change of capacity
1340	 */
1341	if (cluster_check_sync_size(mddev) == 0) {
1342		memset(&cmsg, 0, sizeof(cmsg));
1343		cmsg.type = cpu_to_le32(CHANGE_CAPACITY);
1344		ret = __sendmsg(cinfo, &cmsg);
1345		if (ret)
1346			pr_err("%s:%d: failed to send CHANGE_CAPACITY msg\n",
1347			       __func__, __LINE__);
1348		set_capacity_and_notify(mddev->gendisk, mddev->array_sectors);
1349	} else {
1350		/* revert to previous sectors */
1351		ret = mddev->pers->resize(mddev, old_dev_sectors);
1352		ret = __sendmsg(cinfo, &cmsg);
1353		if (ret)
1354			pr_err("%s:%d: failed to send METADATA_UPDATED msg\n",
1355			       __func__, __LINE__);
1356	}
1357	unlock_comm(cinfo);
1358}
1359
1360static int resync_start(struct mddev *mddev)
1361{
1362	struct md_cluster_info *cinfo = mddev->cluster_info;
1363	return dlm_lock_sync_interruptible(cinfo->resync_lockres, DLM_LOCK_EX, mddev);
1364}
1365
1366static void resync_info_get(struct mddev *mddev, sector_t *lo, sector_t *hi)
1367{
1368	struct md_cluster_info *cinfo = mddev->cluster_info;
1369
1370	spin_lock_irq(&cinfo->suspend_lock);
1371	*lo = cinfo->suspend_lo;
1372	*hi = cinfo->suspend_hi;
1373	spin_unlock_irq(&cinfo->suspend_lock);
1374}
1375
1376static int resync_status_get(struct mddev *mddev)
1377{
1378	struct md_cluster_info *cinfo = mddev->cluster_info;
1379
1380	return test_bit(MD_CLUSTER_WAITING_FOR_SYNC, &cinfo->state);
1381}
1382
1383static int resync_start_notify(struct mddev *mddev)
1384{
1385	struct md_cluster_info *cinfo = mddev->cluster_info;
1386	struct cluster_msg cmsg = {0};
1387
1388	cmsg.type = cpu_to_le32(RESYNCING_START);
1389
1390	return sendmsg(cinfo, &cmsg, 0);
1391}
1392
1393static int resync_info_update(struct mddev *mddev, sector_t lo, sector_t hi)
1394{
1395	struct md_cluster_info *cinfo = mddev->cluster_info;
1396	struct resync_info ri;
1397	struct cluster_msg cmsg = {0};
1398
1399	/* do not send zero again, if we have sent before */
1400	if (hi == 0) {
1401		memcpy(&ri, cinfo->bitmap_lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
1402		if (le64_to_cpu(ri.hi) == 0)
1403			return 0;
1404	}
1405
1406	add_resync_info(cinfo->bitmap_lockres, lo, hi);
1407	/* Re-acquire the lock to refresh LVB */
1408	dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW);
1409	cmsg.type = cpu_to_le32(RESYNCING);
1410	cmsg.low = cpu_to_le64(lo);
1411	cmsg.high = cpu_to_le64(hi);
1412
1413	/*
1414	 * mddev_lock is held if resync_info_update is called from
1415	 * resync_finish (md_reap_sync_thread -> resync_finish)
1416	 */
1417	if (lo == 0 && hi == 0)
1418		return sendmsg(cinfo, &cmsg, 1);
1419	else
1420		return sendmsg(cinfo, &cmsg, 0);
1421}
1422
1423static int resync_finish(struct mddev *mddev)
1424{
1425	struct md_cluster_info *cinfo = mddev->cluster_info;
1426	int ret = 0;
1427
1428	clear_bit(MD_RESYNCING_REMOTE, &mddev->recovery);
1429
1430	/*
1431	 * If resync thread is interrupted so we can't say resync is finished,
1432	 * another node will launch resync thread to continue.
1433	 */
1434	if (!test_bit(MD_CLOSING, &mddev->flags))
1435		ret = resync_info_update(mddev, 0, 0);
1436	dlm_unlock_sync(cinfo->resync_lockres);
1437	return ret;
1438}
1439
1440static int area_resyncing(struct mddev *mddev, int direction,
1441		sector_t lo, sector_t hi)
1442{
1443	struct md_cluster_info *cinfo = mddev->cluster_info;
1444	int ret = 0;
1445
1446	if ((direction == READ) &&
1447		test_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state))
1448		return 1;
1449
1450	spin_lock_irq(&cinfo->suspend_lock);
1451	if (hi > cinfo->suspend_lo && lo < cinfo->suspend_hi)
1452		ret = 1;
1453	spin_unlock_irq(&cinfo->suspend_lock);
1454	return ret;
1455}
1456
1457/* add_new_disk() - initiates a disk add
1458 * However, if this fails before writing md_update_sb(),
1459 * add_new_disk_cancel() must be called to release token lock
1460 */
1461static int add_new_disk(struct mddev *mddev, struct md_rdev *rdev)
1462{
1463	struct md_cluster_info *cinfo = mddev->cluster_info;
1464	struct cluster_msg cmsg;
1465	int ret = 0;
1466	struct mdp_superblock_1 *sb = page_address(rdev->sb_page);
1467	char *uuid = sb->device_uuid;
1468
1469	memset(&cmsg, 0, sizeof(cmsg));
1470	cmsg.type = cpu_to_le32(NEWDISK);
1471	memcpy(cmsg.uuid, uuid, 16);
1472	cmsg.raid_slot = cpu_to_le32(rdev->desc_nr);
1473	if (lock_comm(cinfo, 1))
1474		return -EAGAIN;
1475	ret = __sendmsg(cinfo, &cmsg);
1476	if (ret) {
1477		unlock_comm(cinfo);
1478		return ret;
1479	}
1480	cinfo->no_new_dev_lockres->flags |= DLM_LKF_NOQUEUE;
1481	ret = dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_EX);
1482	cinfo->no_new_dev_lockres->flags &= ~DLM_LKF_NOQUEUE;
1483	/* Some node does not "see" the device */
1484	if (ret == -EAGAIN)
1485		ret = -ENOENT;
1486	if (ret)
1487		unlock_comm(cinfo);
1488	else {
1489		dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR);
1490		/* Since MD_CHANGE_DEVS will be set in add_bound_rdev which
1491		 * will run soon after add_new_disk, the below path will be
1492		 * invoked:
1493		 *   md_wakeup_thread(mddev->thread)
1494		 *	-> conf->thread (raid1d)
1495		 *	-> md_check_recovery -> md_update_sb
1496		 *	-> metadata_update_start/finish
1497		 * MD_CLUSTER_SEND_LOCKED_ALREADY will be cleared eventually.
1498		 *
1499		 * For other failure cases, metadata_update_cancel and
1500		 * add_new_disk_cancel also clear below bit as well.
1501		 * */
1502		set_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
1503		wake_up(&cinfo->wait);
1504	}
1505	return ret;
1506}
1507
1508static void add_new_disk_cancel(struct mddev *mddev)
1509{
1510	struct md_cluster_info *cinfo = mddev->cluster_info;
1511	clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
1512	unlock_comm(cinfo);
1513}
1514
1515static int new_disk_ack(struct mddev *mddev, bool ack)
1516{
1517	struct md_cluster_info *cinfo = mddev->cluster_info;
1518
1519	if (!test_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state)) {
1520		pr_warn("md-cluster(%s): Spurious cluster confirmation\n", mdname(mddev));
1521		return -EINVAL;
1522	}
1523
1524	if (ack)
1525		dlm_unlock_sync(cinfo->no_new_dev_lockres);
1526	complete(&cinfo->newdisk_completion);
1527	return 0;
1528}
1529
1530static int remove_disk(struct mddev *mddev, struct md_rdev *rdev)
1531{
1532	struct cluster_msg cmsg = {0};
1533	struct md_cluster_info *cinfo = mddev->cluster_info;
1534	cmsg.type = cpu_to_le32(REMOVE);
1535	cmsg.raid_slot = cpu_to_le32(rdev->desc_nr);
1536	return sendmsg(cinfo, &cmsg, 1);
1537}
1538
1539static int lock_all_bitmaps(struct mddev *mddev)
1540{
1541	int slot, my_slot, ret, held = 1, i = 0;
1542	char str[64];
1543	struct md_cluster_info *cinfo = mddev->cluster_info;
1544
1545	cinfo->other_bitmap_lockres =
1546		kcalloc(mddev->bitmap_info.nodes - 1,
1547			sizeof(struct dlm_lock_resource *), GFP_KERNEL);
1548	if (!cinfo->other_bitmap_lockres) {
1549		pr_err("md: can't alloc mem for other bitmap locks\n");
1550		return 0;
1551	}
1552
1553	my_slot = slot_number(mddev);
1554	for (slot = 0; slot < mddev->bitmap_info.nodes; slot++) {
1555		if (slot == my_slot)
1556			continue;
1557
1558		memset(str, '\0', 64);
1559		snprintf(str, 64, "bitmap%04d", slot);
1560		cinfo->other_bitmap_lockres[i] = lockres_init(mddev, str, NULL, 1);
1561		if (!cinfo->other_bitmap_lockres[i])
1562			return -ENOMEM;
1563
1564		cinfo->other_bitmap_lockres[i]->flags |= DLM_LKF_NOQUEUE;
1565		ret = dlm_lock_sync(cinfo->other_bitmap_lockres[i], DLM_LOCK_PW);
1566		if (ret)
1567			held = -1;
1568		i++;
1569	}
1570
1571	return held;
1572}
1573
1574static void unlock_all_bitmaps(struct mddev *mddev)
1575{
1576	struct md_cluster_info *cinfo = mddev->cluster_info;
1577	int i;
1578
1579	/* release other node's bitmap lock if they are existed */
1580	if (cinfo->other_bitmap_lockres) {
1581		for (i = 0; i < mddev->bitmap_info.nodes - 1; i++) {
1582			if (cinfo->other_bitmap_lockres[i]) {
1583				lockres_free(cinfo->other_bitmap_lockres[i]);
1584			}
1585		}
1586		kfree(cinfo->other_bitmap_lockres);
1587		cinfo->other_bitmap_lockres = NULL;
1588	}
1589}
1590
1591static int gather_bitmaps(struct md_rdev *rdev)
1592{
1593	int sn, err;
1594	sector_t lo, hi;
1595	struct cluster_msg cmsg = {0};
1596	struct mddev *mddev = rdev->mddev;
1597	struct md_cluster_info *cinfo = mddev->cluster_info;
1598
1599	cmsg.type = cpu_to_le32(RE_ADD);
1600	cmsg.raid_slot = cpu_to_le32(rdev->desc_nr);
1601	err = sendmsg(cinfo, &cmsg, 1);
1602	if (err)
1603		goto out;
1604
1605	for (sn = 0; sn < mddev->bitmap_info.nodes; sn++) {
1606		if (sn == (cinfo->slot_number - 1))
1607			continue;
1608		err = mddev->bitmap_ops->copy_from_slot(mddev, sn, &lo, &hi, false);
1609		if (err) {
1610			pr_warn("md-cluster: Could not gather bitmaps from slot %d", sn);
1611			goto out;
1612		}
1613		if ((hi > 0) && (lo < mddev->resync_offset))
1614			mddev->resync_offset = lo;
1615	}
1616out:
1617	return err;
1618}
1619
1620static struct md_cluster_operations cluster_ops = {
1621	.head = {
1622		.type	= MD_CLUSTER,
1623		.id	= ID_CLUSTER,
1624		.name	= "cluster",
1625		.owner	= THIS_MODULE,
1626	},
1627
1628	.join   = join,
1629	.leave  = leave,
1630	.slot_number = slot_number,
1631	.resync_start = resync_start,
1632	.resync_finish = resync_finish,
1633	.resync_info_update = resync_info_update,
1634	.resync_start_notify = resync_start_notify,
1635	.resync_status_get = resync_status_get,
1636	.resync_info_get = resync_info_get,
1637	.metadata_update_start = metadata_update_start,
1638	.metadata_update_finish = metadata_update_finish,
1639	.metadata_update_cancel = metadata_update_cancel,
1640	.area_resyncing = area_resyncing,
1641	.add_new_disk = add_new_disk,
1642	.add_new_disk_cancel = add_new_disk_cancel,
1643	.new_disk_ack = new_disk_ack,
1644	.remove_disk = remove_disk,
1645	.load_bitmaps = load_bitmaps,
1646	.gather_bitmaps = gather_bitmaps,
1647	.resize_bitmaps = resize_bitmaps,
1648	.lock_all_bitmaps = lock_all_bitmaps,
1649	.unlock_all_bitmaps = unlock_all_bitmaps,
1650	.update_size = update_size,
1651};
1652
1653static int __init cluster_init(void)
1654{
1655	pr_warn("md-cluster: support raid1 and raid10 (limited support)\n");
1656	pr_info("Registering Cluster MD functions\n");
1657	return register_md_submodule(&cluster_ops.head);
1658}
1659
1660static void cluster_exit(void)
1661{
1662	unregister_md_submodule(&cluster_ops.head);
1663}
1664
1665module_init(cluster_init);
1666module_exit(cluster_exit);
1667MODULE_AUTHOR("SUSE");
1668MODULE_LICENSE("GPL");
1669MODULE_DESCRIPTION("Clustering support for MD");