md-cluster.c 30.0 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12
/*
 * Copyright (C) 2015, SUSE
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2, or (at your option)
 * any later version.
 *
 */


#include <linux/module.h>
13 14
#include <linux/dlm.h>
#include <linux/sched.h>
15
#include <linux/raid/md_p.h>
16
#include "md.h"
17
#include "bitmap.h"
18
#include "md-cluster.h"
19 20

#define LVB_SIZE	64
21
#define NEW_DEV_TIMEOUT 5000
22 23 24 25 26 27 28

struct dlm_lock_resource {
	dlm_lockspace_t *ls;
	struct dlm_lksb lksb;
	char *name; /* lock name. */
	uint32_t flags; /* flags to pass to dlm_lock() */
	struct completion completion; /* completion for synchronized locking */
G
Goldwyn Rodrigues 已提交
29 30
	void (*bast)(void *arg, int mode); /* blocking AST function pointer*/
	struct mddev *mddev; /* pointing back to mddev. */
31
	int mode;
G
Goldwyn Rodrigues 已提交
32 33
};

34 35 36 37 38 39 40 41 42 43 44 45
struct suspend_info {
	int slot;
	sector_t lo;
	sector_t hi;
	struct list_head list;
};

struct resync_info {
	__le64 lo;
	__le64 hi;
};

46 47
/* md_cluster_info flags */
#define		MD_CLUSTER_WAITING_FOR_NEWDISK		1
48
#define		MD_CLUSTER_SUSPEND_READ_BALANCING	2
49
#define		MD_CLUSTER_BEGIN_JOIN_CLUSTER		3
50 51


G
Goldwyn Rodrigues 已提交
52 53 54
struct md_cluster_info {
	/* dlm lock space and resources for clustered raid. */
	dlm_lockspace_t *lockspace;
G
Goldwyn Rodrigues 已提交
55 56
	int slot_number;
	struct completion completion;
57
	struct dlm_lock_resource *bitmap_lockres;
58
	struct dlm_lock_resource **other_bitmap_lockres;
59
	struct dlm_lock_resource *resync_lockres;
60 61
	struct list_head suspend_list;
	spinlock_t suspend_lock;
62 63
	struct md_thread *recovery_thread;
	unsigned long recovery_map;
64 65 66 67
	/* communication loc resources */
	struct dlm_lock_resource *ack_lockres;
	struct dlm_lock_resource *message_lockres;
	struct dlm_lock_resource *token_lockres;
68
	struct dlm_lock_resource *no_new_dev_lockres;
69
	struct md_thread *recv_thread;
70
	struct completion newdisk_completion;
71
	unsigned long state;
72 73 74 75 76
};

enum msg_type {
	METADATA_UPDATED = 0,
	RESYNCING,
77
	NEWDISK,
78
	REMOVE,
79
	RE_ADD,
80
	BITMAP_NEEDS_SYNC,
81 82 83
};

struct cluster_msg {
84 85
	__le32 type;
	__le32 slot;
86
	/* TODO: Unionize this for smaller footprint */
87 88
	__le64 low;
	__le64 high;
89
	char uuid[16];
90
	__le32 raid_slot;
91 92 93 94 95 96
};

static void sync_ast(void *arg)
{
	struct dlm_lock_resource *res;

97
	res = arg;
98 99 100 101 102 103 104 105 106 107 108 109 110
	complete(&res->completion);
}

static int dlm_lock_sync(struct dlm_lock_resource *res, int mode)
{
	int ret = 0;

	ret = dlm_lock(res->ls, mode, &res->lksb,
			res->flags, res->name, strlen(res->name),
			0, sync_ast, res, res->bast);
	if (ret)
		return ret;
	wait_for_completion(&res->completion);
111 112
	if (res->lksb.sb_status == 0)
		res->mode = mode;
113 114 115 116 117 118 119 120
	return res->lksb.sb_status;
}

static int dlm_unlock_sync(struct dlm_lock_resource *res)
{
	return dlm_lock_sync(res, DLM_LOCK_NL);
}

G
Goldwyn Rodrigues 已提交
121
static struct dlm_lock_resource *lockres_init(struct mddev *mddev,
122 123 124 125
		char *name, void (*bastfn)(void *arg, int mode), int with_lvb)
{
	struct dlm_lock_resource *res = NULL;
	int ret, namelen;
G
Goldwyn Rodrigues 已提交
126
	struct md_cluster_info *cinfo = mddev->cluster_info;
127 128 129 130

	res = kzalloc(sizeof(struct dlm_lock_resource), GFP_KERNEL);
	if (!res)
		return NULL;
131
	init_completion(&res->completion);
G
Goldwyn Rodrigues 已提交
132 133
	res->ls = cinfo->lockspace;
	res->mddev = mddev;
134
	res->mode = DLM_LOCK_IV;
135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173
	namelen = strlen(name);
	res->name = kzalloc(namelen + 1, GFP_KERNEL);
	if (!res->name) {
		pr_err("md-cluster: Unable to allocate resource name for resource %s\n", name);
		goto out_err;
	}
	strlcpy(res->name, name, namelen + 1);
	if (with_lvb) {
		res->lksb.sb_lvbptr = kzalloc(LVB_SIZE, GFP_KERNEL);
		if (!res->lksb.sb_lvbptr) {
			pr_err("md-cluster: Unable to allocate LVB for resource %s\n", name);
			goto out_err;
		}
		res->flags = DLM_LKF_VALBLK;
	}

	if (bastfn)
		res->bast = bastfn;

	res->flags |= DLM_LKF_EXPEDITE;

	ret = dlm_lock_sync(res, DLM_LOCK_NL);
	if (ret) {
		pr_err("md-cluster: Unable to lock NL on new lock resource %s\n", name);
		goto out_err;
	}
	res->flags &= ~DLM_LKF_EXPEDITE;
	res->flags |= DLM_LKF_CONVERT;

	return res;
out_err:
	kfree(res->lksb.sb_lvbptr);
	kfree(res->name);
	kfree(res);
	return NULL;
}

static void lockres_free(struct dlm_lock_resource *res)
{
174 175
	int ret;

176 177 178
	if (!res)
		return;

179 180 181 182 183 184 185 186 187 188 189 190 191
	/* cancel a lock request or a conversion request that is blocked */
	res->flags |= DLM_LKF_CANCEL;
retry:
	ret = dlm_unlock(res->ls, res->lksb.sb_lkid, 0, &res->lksb, res);
	if (unlikely(ret != 0)) {
		pr_info("%s: failed to unlock %s return %d\n", __func__, res->name, ret);

		/* if a lock conversion is cancelled, then the lock is put
		 * back to grant queue, need to ensure it is unlocked */
		if (ret == -DLM_ECANCEL)
			goto retry;
	}
	res->flags &= ~DLM_LKF_CANCEL;
192 193 194 195 196 197
	wait_for_completion(&res->completion);

	kfree(res->name);
	kfree(res->lksb.sb_lvbptr);
	kfree(res);
}
198

199 200
static void add_resync_info(struct dlm_lock_resource *lockres,
			    sector_t lo, sector_t hi)
201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217
{
	struct resync_info *ri;

	ri = (struct resync_info *)lockres->lksb.sb_lvbptr;
	ri->lo = cpu_to_le64(lo);
	ri->hi = cpu_to_le64(hi);
}

static struct suspend_info *read_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres)
{
	struct resync_info ri;
	struct suspend_info *s = NULL;
	sector_t hi = 0;

	dlm_lock_sync(lockres, DLM_LOCK_CR);
	memcpy(&ri, lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
	hi = le64_to_cpu(ri.hi);
218
	if (hi > 0) {
219 220 221 222 223 224 225 226 227 228 229
		s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL);
		if (!s)
			goto out;
		s->hi = hi;
		s->lo = le64_to_cpu(ri.lo);
	}
	dlm_unlock_sync(lockres);
out:
	return s;
}

230
static void recover_bitmaps(struct md_thread *thread)
231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264
{
	struct mddev *mddev = thread->mddev;
	struct md_cluster_info *cinfo = mddev->cluster_info;
	struct dlm_lock_resource *bm_lockres;
	char str[64];
	int slot, ret;
	struct suspend_info *s, *tmp;
	sector_t lo, hi;

	while (cinfo->recovery_map) {
		slot = fls64((u64)cinfo->recovery_map) - 1;

		/* Clear suspend_area associated with the bitmap */
		spin_lock_irq(&cinfo->suspend_lock);
		list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list)
			if (slot == s->slot) {
				list_del(&s->list);
				kfree(s);
			}
		spin_unlock_irq(&cinfo->suspend_lock);

		snprintf(str, 64, "bitmap%04d", slot);
		bm_lockres = lockres_init(mddev, str, NULL, 1);
		if (!bm_lockres) {
			pr_err("md-cluster: Cannot initialize bitmaps\n");
			goto clear_bit;
		}

		ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
		if (ret) {
			pr_err("md-cluster: Could not DLM lock %s: %d\n",
					str, ret);
			goto clear_bit;
		}
265
		ret = bitmap_copy_from_slot(mddev, slot, &lo, &hi, true);
266
		if (ret) {
267
			pr_err("md-cluster: Could not copy data from bitmap %d\n", slot);
268 269 270 271 272 273 274 275 276 277
			goto dlm_unlock;
		}
		if (hi > 0) {
			/* TODO:Wait for current resync to get over */
			set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
			if (lo < mddev->recovery_cp)
				mddev->recovery_cp = lo;
			md_check_recovery(mddev);
		}
dlm_unlock:
278 279 280 281 282 283
		dlm_unlock_sync(bm_lockres);
clear_bit:
		clear_bit(slot, &cinfo->recovery_map);
	}
}

G
Goldwyn Rodrigues 已提交
284 285
static void recover_prep(void *arg)
{
286 287 288
	struct mddev *mddev = arg;
	struct md_cluster_info *cinfo = mddev->cluster_info;
	set_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state);
G
Goldwyn Rodrigues 已提交
289 290
}

291
static void __recover_slot(struct mddev *mddev, int slot)
G
Goldwyn Rodrigues 已提交
292 293 294
{
	struct md_cluster_info *cinfo = mddev->cluster_info;

295
	set_bit(slot, &cinfo->recovery_map);
296 297 298 299 300 301 302 303 304
	if (!cinfo->recovery_thread) {
		cinfo->recovery_thread = md_register_thread(recover_bitmaps,
				mddev, "recover");
		if (!cinfo->recovery_thread) {
			pr_warn("md-cluster: Could not create recovery thread\n");
			return;
		}
	}
	md_wakeup_thread(cinfo->recovery_thread);
G
Goldwyn Rodrigues 已提交
305 306
}

307 308 309 310 311 312 313 314 315 316 317 318 319 320
static void recover_slot(void *arg, struct dlm_slot *slot)
{
	struct mddev *mddev = arg;
	struct md_cluster_info *cinfo = mddev->cluster_info;

	pr_info("md-cluster: %s Node %d/%d down. My slot: %d. Initiating recovery.\n",
			mddev->bitmap_info.cluster_name,
			slot->nodeid, slot->slot,
			cinfo->slot_number);
	/* deduct one since dlm slot starts from one while the num of
	 * cluster-md begins with 0 */
	__recover_slot(mddev, slot->slot - 1);
}

G
Goldwyn Rodrigues 已提交
321 322 323 324 325 326 327 328
static void recover_done(void *arg, struct dlm_slot *slots,
		int num_slots, int our_slot,
		uint32_t generation)
{
	struct mddev *mddev = arg;
	struct md_cluster_info *cinfo = mddev->cluster_info;

	cinfo->slot_number = our_slot;
329 330 331 332 333 334
	/* completion is only need to be complete when node join cluster,
	 * it doesn't need to run during another node's failure */
	if (test_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state)) {
		complete(&cinfo->completion);
		clear_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state);
	}
335
	clear_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state);
G
Goldwyn Rodrigues 已提交
336 337
}

338 339
/* the ops is called when node join the cluster, and do lock recovery
 * if node failure occurs */
G
Goldwyn Rodrigues 已提交
340 341 342 343 344 345
static const struct dlm_lockspace_ops md_ls_ops = {
	.recover_prep = recover_prep,
	.recover_slot = recover_slot,
	.recover_done = recover_done,
};

346 347 348 349 350 351 352
/*
 * The BAST function for the ack lock resource
 * This function wakes up the receive thread in
 * order to receive and process the message.
 */
static void ack_bast(void *arg, int mode)
{
353
	struct dlm_lock_resource *res = arg;
354 355 356 357 358 359
	struct md_cluster_info *cinfo = res->mddev->cluster_info;

	if (mode == DLM_LOCK_EX)
		md_wakeup_thread(cinfo->recv_thread);
}

G
Goldwyn Rodrigues 已提交
360 361 362 363 364 365 366 367 368 369 370 371
static void __remove_suspend_info(struct md_cluster_info *cinfo, int slot)
{
	struct suspend_info *s, *tmp;

	list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list)
		if (slot == s->slot) {
			list_del(&s->list);
			kfree(s);
			break;
		}
}

372
static void remove_suspend_info(struct mddev *mddev, int slot)
G
Goldwyn Rodrigues 已提交
373
{
374
	struct md_cluster_info *cinfo = mddev->cluster_info;
G
Goldwyn Rodrigues 已提交
375 376 377
	spin_lock_irq(&cinfo->suspend_lock);
	__remove_suspend_info(cinfo, slot);
	spin_unlock_irq(&cinfo->suspend_lock);
378
	mddev->pers->quiesce(mddev, 2);
G
Goldwyn Rodrigues 已提交
379 380 381
}


382
static void process_suspend_info(struct mddev *mddev,
G
Goldwyn Rodrigues 已提交
383 384
		int slot, sector_t lo, sector_t hi)
{
385
	struct md_cluster_info *cinfo = mddev->cluster_info;
G
Goldwyn Rodrigues 已提交
386 387 388
	struct suspend_info *s;

	if (!hi) {
389
		remove_suspend_info(mddev, slot);
390 391
		set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
		md_wakeup_thread(mddev->thread);
G
Goldwyn Rodrigues 已提交
392 393 394 395 396 397 398 399
		return;
	}
	s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL);
	if (!s)
		return;
	s->slot = slot;
	s->lo = lo;
	s->hi = hi;
400 401
	mddev->pers->quiesce(mddev, 1);
	mddev->pers->quiesce(mddev, 0);
G
Goldwyn Rodrigues 已提交
402 403 404 405 406
	spin_lock_irq(&cinfo->suspend_lock);
	/* Remove existing entry (if exists) before adding */
	__remove_suspend_info(cinfo, slot);
	list_add(&s->list, &cinfo->suspend_list);
	spin_unlock_irq(&cinfo->suspend_lock);
407
	mddev->pers->quiesce(mddev, 2);
G
Goldwyn Rodrigues 已提交
408 409
}

410 411 412 413 414 415 416 417 418 419
static void process_add_new_disk(struct mddev *mddev, struct cluster_msg *cmsg)
{
	char disk_uuid[64];
	struct md_cluster_info *cinfo = mddev->cluster_info;
	char event_name[] = "EVENT=ADD_DEVICE";
	char raid_slot[16];
	char *envp[] = {event_name, disk_uuid, raid_slot, NULL};
	int len;

	len = snprintf(disk_uuid, 64, "DEVICE_UUID=");
420
	sprintf(disk_uuid + len, "%pU", cmsg->uuid);
421
	snprintf(raid_slot, 16, "RAID_DISK=%d", le32_to_cpu(cmsg->raid_slot));
422 423
	pr_info("%s:%d Sending kobject change with %s and %s\n", __func__, __LINE__, disk_uuid, raid_slot);
	init_completion(&cinfo->newdisk_completion);
424
	set_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state);
425 426 427
	kobject_uevent_env(&disk_to_dev(mddev->gendisk)->kobj, KOBJ_CHANGE, envp);
	wait_for_completion_timeout(&cinfo->newdisk_completion,
			NEW_DEV_TIMEOUT);
428
	clear_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state);
429 430 431 432 433 434
}


static void process_metadata_update(struct mddev *mddev, struct cluster_msg *msg)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
435 436
	mddev->good_device_nr = le32_to_cpu(msg->raid_slot);
	set_bit(MD_RELOAD_SB, &mddev->flags);
437
	dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR);
438
	md_wakeup_thread(mddev->thread);
439 440
}

441 442
static void process_remove_disk(struct mddev *mddev, struct cluster_msg *msg)
{
443 444
	struct md_rdev *rdev = md_find_rdev_nr_rcu(mddev,
						   le32_to_cpu(msg->raid_slot));
445

446 447 448 449 450
	if (rdev) {
		set_bit(ClusterRemove, &rdev->flags);
		set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
		md_wakeup_thread(mddev->thread);
	}
451
	else
452 453
		pr_warn("%s: %d Could not find disk(%d) to REMOVE\n",
			__func__, __LINE__, le32_to_cpu(msg->raid_slot));
454 455
}

456 457
static void process_readd_disk(struct mddev *mddev, struct cluster_msg *msg)
{
458 459
	struct md_rdev *rdev = md_find_rdev_nr_rcu(mddev,
						   le32_to_cpu(msg->raid_slot));
460 461 462 463

	if (rdev && test_bit(Faulty, &rdev->flags))
		clear_bit(Faulty, &rdev->flags);
	else
464 465
		pr_warn("%s: %d Could not find disk(%d) which is faulty",
			__func__, __LINE__, le32_to_cpu(msg->raid_slot));
466 467
}

468 469
static void process_recvd_msg(struct mddev *mddev, struct cluster_msg *msg)
{
470 471 472
	if (WARN(mddev->cluster_info->slot_number - 1 == le32_to_cpu(msg->slot),
		"node %d received it's own msg\n", le32_to_cpu(msg->slot)))
		return;
473
	switch (le32_to_cpu(msg->type)) {
474
	case METADATA_UPDATED:
475
		process_metadata_update(mddev, msg);
476 477
		break;
	case RESYNCING:
478 479 480
		process_suspend_info(mddev, le32_to_cpu(msg->slot),
				     le64_to_cpu(msg->low),
				     le64_to_cpu(msg->high));
481
		break;
482 483
	case NEWDISK:
		process_add_new_disk(mddev, msg);
484 485 486 487
		break;
	case REMOVE:
		process_remove_disk(mddev, msg);
		break;
488 489 490
	case RE_ADD:
		process_readd_disk(mddev, msg);
		break;
491
	case BITMAP_NEEDS_SYNC:
492
		__recover_slot(mddev, le32_to_cpu(msg->slot));
493
		break;
494 495 496
	default:
		pr_warn("%s:%d Received unknown message from %d\n",
			__func__, __LINE__, msg->slot);
497
	}
498 499 500 501 502 503 504 505 506 507 508
}

/*
 * thread for receiving message
 */
static void recv_daemon(struct md_thread *thread)
{
	struct md_cluster_info *cinfo = thread->mddev->cluster_info;
	struct dlm_lock_resource *ack_lockres = cinfo->ack_lockres;
	struct dlm_lock_resource *message_lockres = cinfo->message_lockres;
	struct cluster_msg msg;
509
	int ret;
510 511 512 513 514 515 516 517 518 519 520 521

	/*get CR on Message*/
	if (dlm_lock_sync(message_lockres, DLM_LOCK_CR)) {
		pr_err("md/raid1:failed to get CR on MESSAGE\n");
		return;
	}

	/* read lvb and wake up thread to process this message_lockres */
	memcpy(&msg, message_lockres->lksb.sb_lvbptr, sizeof(struct cluster_msg));
	process_recvd_msg(thread->mddev, &msg);

	/*release CR on ack_lockres*/
522 523 524
	ret = dlm_unlock_sync(ack_lockres);
	if (unlikely(ret != 0))
		pr_info("unlock ack failed return %d\n", ret);
525
	/*up-convert to PR on message_lockres*/
526 527 528
	ret = dlm_lock_sync(message_lockres, DLM_LOCK_PR);
	if (unlikely(ret != 0))
		pr_info("lock PR on msg failed return %d\n", ret);
529
	/*get CR on ack_lockres again*/
530 531 532
	ret = dlm_lock_sync(ack_lockres, DLM_LOCK_CR);
	if (unlikely(ret != 0))
		pr_info("lock CR on ack failed return %d\n", ret);
533
	/*release CR on message_lockres*/
534 535 536
	ret = dlm_unlock_sync(message_lockres);
	if (unlikely(ret != 0))
		pr_info("unlock msg failed return %d\n", ret);
537 538
}

539 540 541
/* lock_comm()
 * Takes the lock on the TOKEN lock resource so no other
 * node can communicate while the operation is underway.
542 543 544
 * If called again, and the TOKEN lock is alread in EX mode
 * return success. However, care must be taken that unlock_comm()
 * is called only once.
545 546 547 548 549
 */
static int lock_comm(struct md_cluster_info *cinfo)
{
	int error;

550 551 552
	if (cinfo->token_lockres->mode == DLM_LOCK_EX)
		return 0;

553 554 555 556 557 558 559 560 561
	error = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
	if (error)
		pr_err("md-cluster(%s:%d): failed to get EX on TOKEN (%d)\n",
				__func__, __LINE__, error);
	return error;
}

static void unlock_comm(struct md_cluster_info *cinfo)
{
562
	WARN_ON(cinfo->token_lockres->mode != DLM_LOCK_EX);
563 564 565 566 567 568 569 570 571
	dlm_unlock_sync(cinfo->token_lockres);
}

/* __sendmsg()
 * This function performs the actual sending of the message. This function is
 * usually called after performing the encompassing operation
 * The function:
 * 1. Grabs the message lockresource in EX mode
 * 2. Copies the message to the message LVB
572
 * 3. Downconverts message lockresource to CW
573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592
 * 4. Upconverts ack lock resource from CR to EX. This forces the BAST on other nodes
 *    and the other nodes read the message. The thread will wait here until all other
 *    nodes have released ack lock resource.
 * 5. Downconvert ack lockresource to CR
 */
static int __sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
{
	int error;
	int slot = cinfo->slot_number - 1;

	cmsg->slot = cpu_to_le32(slot);
	/*get EX on Message*/
	error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_EX);
	if (error) {
		pr_err("md-cluster: failed to get EX on MESSAGE (%d)\n", error);
		goto failed_message;
	}

	memcpy(cinfo->message_lockres->lksb.sb_lvbptr, (void *)cmsg,
			sizeof(struct cluster_msg));
593 594
	/*down-convert EX to CW on Message*/
	error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_CW);
595
	if (error) {
596
		pr_err("md-cluster: failed to convert EX to CW on MESSAGE(%d)\n",
597
				error);
598
		goto failed_ack;
599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617
	}

	/*up-convert CR to EX on Ack*/
	error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_EX);
	if (error) {
		pr_err("md-cluster: failed to convert CR to EX on ACK(%d)\n",
				error);
		goto failed_ack;
	}

	/*down-convert EX to CR on Ack*/
	error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR);
	if (error) {
		pr_err("md-cluster: failed to convert EX to CR on ACK(%d)\n",
				error);
		goto failed_ack;
	}

failed_ack:
618 619 620 621 622 623 624
	error = dlm_unlock_sync(cinfo->message_lockres);
	if (unlikely(error != 0)) {
		pr_err("md-cluster: failed convert to NL on MESSAGE(%d)\n",
			error);
		/* in case the message can't be released due to some reason */
		goto failed_ack;
	}
625 626 627 628 629 630 631 632 633 634 635 636 637 638
failed_message:
	return error;
}

static int sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
{
	int ret;

	lock_comm(cinfo);
	ret = __sendmsg(cinfo, cmsg);
	unlock_comm(cinfo);
	return ret;
}

639 640 641 642 643 644 645
static int gather_all_resync_info(struct mddev *mddev, int total_slots)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
	int i, ret = 0;
	struct dlm_lock_resource *bm_lockres;
	struct suspend_info *s;
	char str[64];
646
	sector_t lo, hi;
647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676


	for (i = 0; i < total_slots; i++) {
		memset(str, '\0', 64);
		snprintf(str, 64, "bitmap%04d", i);
		bm_lockres = lockres_init(mddev, str, NULL, 1);
		if (!bm_lockres)
			return -ENOMEM;
		if (i == (cinfo->slot_number - 1))
			continue;

		bm_lockres->flags |= DLM_LKF_NOQUEUE;
		ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
		if (ret == -EAGAIN) {
			memset(bm_lockres->lksb.sb_lvbptr, '\0', LVB_SIZE);
			s = read_resync_info(mddev, bm_lockres);
			if (s) {
				pr_info("%s:%d Resync[%llu..%llu] in progress on %d\n",
						__func__, __LINE__,
						(unsigned long long) s->lo,
						(unsigned long long) s->hi, i);
				spin_lock_irq(&cinfo->suspend_lock);
				s->slot = i;
				list_add(&s->list, &cinfo->suspend_list);
				spin_unlock_irq(&cinfo->suspend_lock);
			}
			ret = 0;
			lockres_free(bm_lockres);
			continue;
		}
677 678
		if (ret) {
			lockres_free(bm_lockres);
679
			goto out;
680
		}
681 682 683 684 685 686 687 688 689 690 691 692 693 694

		/* Read the disk bitmap sb and check if it needs recovery */
		ret = bitmap_copy_from_slot(mddev, i, &lo, &hi, false);
		if (ret) {
			pr_warn("md-cluster: Could not gather bitmaps from slot %d", i);
			lockres_free(bm_lockres);
			continue;
		}
		if ((hi > 0) && (lo < mddev->recovery_cp)) {
			set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
			mddev->recovery_cp = lo;
			md_check_recovery(mddev);
		}

695 696 697 698 699 700 701
		dlm_unlock_sync(bm_lockres);
		lockres_free(bm_lockres);
	}
out:
	return ret;
}

702 703
static int join(struct mddev *mddev, int nodes)
{
G
Goldwyn Rodrigues 已提交
704
	struct md_cluster_info *cinfo;
G
Goldwyn Rodrigues 已提交
705
	int ret, ops_rv;
G
Goldwyn Rodrigues 已提交
706 707 708 709 710 711
	char str[64];

	cinfo = kzalloc(sizeof(struct md_cluster_info), GFP_KERNEL);
	if (!cinfo)
		return -ENOMEM;

712 713
	INIT_LIST_HEAD(&cinfo->suspend_list);
	spin_lock_init(&cinfo->suspend_lock);
G
Goldwyn Rodrigues 已提交
714
	init_completion(&cinfo->completion);
715
	set_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state);
G
Goldwyn Rodrigues 已提交
716 717 718

	mddev->cluster_info = cinfo;

G
Goldwyn Rodrigues 已提交
719
	memset(str, 0, 64);
720
	sprintf(str, "%pU", mddev->uuid);
G
Goldwyn Rodrigues 已提交
721 722 723
	ret = dlm_new_lockspace(str, mddev->bitmap_info.cluster_name,
				DLM_LSFL_FS, LVB_SIZE,
				&md_ls_ops, mddev, &ops_rv, &cinfo->lockspace);
G
Goldwyn Rodrigues 已提交
724 725
	if (ret)
		goto err;
G
Goldwyn Rodrigues 已提交
726
	wait_for_completion(&cinfo->completion);
727 728 729
	if (nodes < cinfo->slot_number) {
		pr_err("md-cluster: Slot allotted(%d) is greater than available slots(%d).",
			cinfo->slot_number, nodes);
730 731 732
		ret = -ERANGE;
		goto err;
	}
733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748
	/* Initiate the communication resources */
	ret = -ENOMEM;
	cinfo->recv_thread = md_register_thread(recv_daemon, mddev, "cluster_recv");
	if (!cinfo->recv_thread) {
		pr_err("md-cluster: cannot allocate memory for recv_thread!\n");
		goto err;
	}
	cinfo->message_lockres = lockres_init(mddev, "message", NULL, 1);
	if (!cinfo->message_lockres)
		goto err;
	cinfo->token_lockres = lockres_init(mddev, "token", NULL, 0);
	if (!cinfo->token_lockres)
		goto err;
	cinfo->ack_lockres = lockres_init(mddev, "ack", ack_bast, 0);
	if (!cinfo->ack_lockres)
		goto err;
749 750 751 752
	cinfo->no_new_dev_lockres = lockres_init(mddev, "no-new-dev", NULL, 0);
	if (!cinfo->no_new_dev_lockres)
		goto err;

753 754 755 756
	/* get sync CR lock on ACK. */
	if (dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR))
		pr_err("md-cluster: failed to get a sync CR lock on ACK!(%d)\n",
				ret);
757 758 759 760
	/* get sync CR lock on no-new-dev. */
	if (dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR))
		pr_err("md-cluster: failed to get a sync CR lock on no-new-dev!(%d)\n", ret);

761 762 763 764 765 766 767 768 769 770 771 772

	pr_info("md-cluster: Joined cluster %s slot %d\n", str, cinfo->slot_number);
	snprintf(str, 64, "bitmap%04d", cinfo->slot_number - 1);
	cinfo->bitmap_lockres = lockres_init(mddev, str, NULL, 1);
	if (!cinfo->bitmap_lockres)
		goto err;
	if (dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW)) {
		pr_err("Failed to get bitmap lock\n");
		ret = -EINVAL;
		goto err;
	}

773 774 775 776
	cinfo->resync_lockres = lockres_init(mddev, "resync", NULL, 0);
	if (!cinfo->resync_lockres)
		goto err;

777 778 779 780
	ret = gather_all_resync_info(mddev, nodes);
	if (ret)
		goto err;

781
	return 0;
G
Goldwyn Rodrigues 已提交
782
err:
783 784 785
	lockres_free(cinfo->message_lockres);
	lockres_free(cinfo->token_lockres);
	lockres_free(cinfo->ack_lockres);
786
	lockres_free(cinfo->no_new_dev_lockres);
787
	lockres_free(cinfo->resync_lockres);
788
	lockres_free(cinfo->bitmap_lockres);
G
Goldwyn Rodrigues 已提交
789 790
	if (cinfo->lockspace)
		dlm_release_lockspace(cinfo->lockspace, 2);
G
Goldwyn Rodrigues 已提交
791
	mddev->cluster_info = NULL;
G
Goldwyn Rodrigues 已提交
792 793
	kfree(cinfo);
	return ret;
794 795
}

796 797 798 799 800 801 802 803 804 805 806 807 808
static void resync_bitmap(struct mddev *mddev)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
	struct cluster_msg cmsg = {0};
	int err;

	cmsg.type = cpu_to_le32(BITMAP_NEEDS_SYNC);
	err = sendmsg(cinfo, &cmsg);
	if (err)
		pr_err("%s:%d: failed to send BITMAP_NEEDS_SYNC message (%d)\n",
			__func__, __LINE__, err);
}

809
static void unlock_all_bitmaps(struct mddev *mddev);
810 811
static int leave(struct mddev *mddev)
{
G
Goldwyn Rodrigues 已提交
812 813 814 815
	struct md_cluster_info *cinfo = mddev->cluster_info;

	if (!cinfo)
		return 0;
816 817 818 819 820 821 822

	/* BITMAP_NEEDS_SYNC message should be sent when node
	 * is leaving the cluster with dirty bitmap, also we
	 * can only deliver it when dlm connection is available */
	if (cinfo->slot_number > 0 && mddev->recovery_cp != MaxSector)
		resync_bitmap(mddev);

823
	md_unregister_thread(&cinfo->recovery_thread);
824 825 826 827
	md_unregister_thread(&cinfo->recv_thread);
	lockres_free(cinfo->message_lockres);
	lockres_free(cinfo->token_lockres);
	lockres_free(cinfo->ack_lockres);
828
	lockres_free(cinfo->no_new_dev_lockres);
829
	lockres_free(cinfo->bitmap_lockres);
830
	unlock_all_bitmaps(mddev);
G
Goldwyn Rodrigues 已提交
831
	dlm_release_lockspace(cinfo->lockspace, 2);
832 833 834
	return 0;
}

G
Goldwyn Rodrigues 已提交
835 836 837 838 839 840 841 842 843 844 845
/* slot_number(): Returns the MD slot number to use
 * DLM starts the slot numbers from 1, wheras cluster-md
 * wants the number to be from zero, so we deduct one
 */
static int slot_number(struct mddev *mddev)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;

	return cinfo->slot_number - 1;
}

846 847 848 849 850 851 852 853 854
static int metadata_update_start(struct mddev *mddev)
{
	return lock_comm(mddev->cluster_info);
}

static int metadata_update_finish(struct mddev *mddev)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
	struct cluster_msg cmsg;
855 856
	struct md_rdev *rdev;
	int ret = 0;
857
	int raid_slot = -1;
858 859 860

	memset(&cmsg, 0, sizeof(cmsg));
	cmsg.type = cpu_to_le32(METADATA_UPDATED);
861 862 863 864
	/* Pick up a good active device number to send.
	 */
	rdev_for_each(rdev, mddev)
		if (rdev->raid_disk > -1 && !test_bit(Faulty, &rdev->flags)) {
865
			raid_slot = rdev->desc_nr;
866 867
			break;
		}
868 869
	if (raid_slot >= 0) {
		cmsg.raid_slot = cpu_to_le32(raid_slot);
870
		ret = __sendmsg(cinfo, &cmsg);
871
	} else
872
		pr_warn("md-cluster: No good device id found to send\n");
873 874 875 876
	unlock_comm(cinfo);
	return ret;
}

877
static void metadata_update_cancel(struct mddev *mddev)
878 879
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
880
	unlock_comm(cinfo);
881 882
}

883 884 885 886 887 888 889
static int resync_start(struct mddev *mddev)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
	cinfo->resync_lockres->flags |= DLM_LKF_NOQUEUE;
	return dlm_lock_sync(cinfo->resync_lockres, DLM_LOCK_EX);
}

890
static int resync_info_update(struct mddev *mddev, sector_t lo, sector_t hi)
891 892
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
893
	struct resync_info ri;
894
	struct cluster_msg cmsg = {0};
895

896 897 898 899 900 901 902
	/* do not send zero again, if we have sent before */
	if (hi == 0) {
		memcpy(&ri, cinfo->bitmap_lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
		if (le64_to_cpu(ri.hi) == 0)
			return 0;
	}

903
	add_resync_info(cinfo->bitmap_lockres, lo, hi);
904 905 906
	/* Re-acquire the lock to refresh LVB */
	dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW);
	cmsg.type = cpu_to_le32(RESYNCING);
907 908
	cmsg.low = cpu_to_le64(lo);
	cmsg.high = cpu_to_le64(hi);
909

910 911 912
	return sendmsg(cinfo, &cmsg);
}

913 914 915 916 917 918 919 920
static int resync_finish(struct mddev *mddev)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
	cinfo->resync_lockres->flags &= ~DLM_LKF_NOQUEUE;
	dlm_unlock_sync(cinfo->resync_lockres);
	return resync_info_update(mddev, 0, 0);
}

921 922
static int area_resyncing(struct mddev *mddev, int direction,
		sector_t lo, sector_t hi)
923 924 925 926 927
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
	int ret = 0;
	struct suspend_info *s;

928 929 930 931
	if ((direction == READ) &&
		test_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state))
		return 1;

932 933 934 935 936 937 938 939 940 941 942 943 944
	spin_lock_irq(&cinfo->suspend_lock);
	if (list_empty(&cinfo->suspend_list))
		goto out;
	list_for_each_entry(s, &cinfo->suspend_list, list)
		if (hi > s->lo && lo < s->hi) {
			ret = 1;
			break;
		}
out:
	spin_unlock_irq(&cinfo->suspend_lock);
	return ret;
}

945 946 947 948 949
/* add_new_disk() - initiates a disk add
 * However, if this fails before writing md_update_sb(),
 * add_new_disk_cancel() must be called to release token lock
 */
static int add_new_disk(struct mddev *mddev, struct md_rdev *rdev)
950 951 952 953 954 955 956 957 958 959
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
	struct cluster_msg cmsg;
	int ret = 0;
	struct mdp_superblock_1 *sb = page_address(rdev->sb_page);
	char *uuid = sb->device_uuid;

	memset(&cmsg, 0, sizeof(cmsg));
	cmsg.type = cpu_to_le32(NEWDISK);
	memcpy(cmsg.uuid, uuid, 16);
960
	cmsg.raid_slot = cpu_to_le32(rdev->desc_nr);
961 962 963 964 965 966 967 968 969 970
	lock_comm(cinfo);
	ret = __sendmsg(cinfo, &cmsg);
	if (ret)
		return ret;
	cinfo->no_new_dev_lockres->flags |= DLM_LKF_NOQUEUE;
	ret = dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_EX);
	cinfo->no_new_dev_lockres->flags &= ~DLM_LKF_NOQUEUE;
	/* Some node does not "see" the device */
	if (ret == -EAGAIN)
		ret = -ENOENT;
971 972
	if (ret)
		unlock_comm(cinfo);
973 974 975 976 977
	else
		dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR);
	return ret;
}

978
static void add_new_disk_cancel(struct mddev *mddev)
979
{
980 981
	struct md_cluster_info *cinfo = mddev->cluster_info;
	unlock_comm(cinfo);
982 983
}

984
static int new_disk_ack(struct mddev *mddev, bool ack)
985 986 987
{
	struct md_cluster_info *cinfo = mddev->cluster_info;

988 989 990 991 992
	if (!test_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state)) {
		pr_warn("md-cluster(%s): Spurious cluster confirmation\n", mdname(mddev));
		return -EINVAL;
	}

993 994 995
	if (ack)
		dlm_unlock_sync(cinfo->no_new_dev_lockres);
	complete(&cinfo->newdisk_completion);
996
	return 0;
997 998
}

999 1000
static int remove_disk(struct mddev *mddev, struct md_rdev *rdev)
{
1001
	struct cluster_msg cmsg = {0};
1002
	struct md_cluster_info *cinfo = mddev->cluster_info;
1003 1004
	cmsg.type = cpu_to_le32(REMOVE);
	cmsg.raid_slot = cpu_to_le32(rdev->desc_nr);
1005
	return sendmsg(cinfo, &cmsg);
1006 1007
}

1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059
static int lock_all_bitmaps(struct mddev *mddev)
{
	int slot, my_slot, ret, held = 1, i = 0;
	char str[64];
	struct md_cluster_info *cinfo = mddev->cluster_info;

	cinfo->other_bitmap_lockres = kzalloc((mddev->bitmap_info.nodes - 1) *
					     sizeof(struct dlm_lock_resource *),
					     GFP_KERNEL);
	if (!cinfo->other_bitmap_lockres) {
		pr_err("md: can't alloc mem for other bitmap locks\n");
		return 0;
	}

	my_slot = slot_number(mddev);
	for (slot = 0; slot < mddev->bitmap_info.nodes; slot++) {
		if (slot == my_slot)
			continue;

		memset(str, '\0', 64);
		snprintf(str, 64, "bitmap%04d", slot);
		cinfo->other_bitmap_lockres[i] = lockres_init(mddev, str, NULL, 1);
		if (!cinfo->other_bitmap_lockres[i])
			return -ENOMEM;

		cinfo->other_bitmap_lockres[i]->flags |= DLM_LKF_NOQUEUE;
		ret = dlm_lock_sync(cinfo->other_bitmap_lockres[i], DLM_LOCK_PW);
		if (ret)
			held = -1;
		i++;
	}

	return held;
}

static void unlock_all_bitmaps(struct mddev *mddev)
{
	struct md_cluster_info *cinfo = mddev->cluster_info;
	int i;

	/* release other node's bitmap lock if they are existed */
	if (cinfo->other_bitmap_lockres) {
		for (i = 0; i < mddev->bitmap_info.nodes - 1; i++) {
			if (cinfo->other_bitmap_lockres[i]) {
				dlm_unlock_sync(cinfo->other_bitmap_lockres[i]);
				lockres_free(cinfo->other_bitmap_lockres[i]);
			}
		}
		kfree(cinfo->other_bitmap_lockres);
	}
}

1060 1061 1062 1063
static int gather_bitmaps(struct md_rdev *rdev)
{
	int sn, err;
	sector_t lo, hi;
1064
	struct cluster_msg cmsg = {0};
1065 1066 1067
	struct mddev *mddev = rdev->mddev;
	struct md_cluster_info *cinfo = mddev->cluster_info;

1068 1069
	cmsg.type = cpu_to_le32(RE_ADD);
	cmsg.raid_slot = cpu_to_le32(rdev->desc_nr);
1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088
	err = sendmsg(cinfo, &cmsg);
	if (err)
		goto out;

	for (sn = 0; sn < mddev->bitmap_info.nodes; sn++) {
		if (sn == (cinfo->slot_number - 1))
			continue;
		err = bitmap_copy_from_slot(mddev, sn, &lo, &hi, false);
		if (err) {
			pr_warn("md-cluster: Could not gather bitmaps from slot %d", sn);
			goto out;
		}
		if ((hi > 0) && (lo < mddev->recovery_cp))
			mddev->recovery_cp = lo;
	}
out:
	return err;
}

1089 1090 1091
static struct md_cluster_operations cluster_ops = {
	.join   = join,
	.leave  = leave,
G
Goldwyn Rodrigues 已提交
1092
	.slot_number = slot_number,
1093 1094
	.resync_start = resync_start,
	.resync_finish = resync_finish,
1095
	.resync_info_update = resync_info_update,
1096 1097 1098
	.metadata_update_start = metadata_update_start,
	.metadata_update_finish = metadata_update_finish,
	.metadata_update_cancel = metadata_update_cancel,
1099
	.area_resyncing = area_resyncing,
1100 1101
	.add_new_disk = add_new_disk,
	.add_new_disk_cancel = add_new_disk_cancel,
1102
	.new_disk_ack = new_disk_ack,
1103
	.remove_disk = remove_disk,
1104
	.gather_bitmaps = gather_bitmaps,
1105 1106
	.lock_all_bitmaps = lock_all_bitmaps,
	.unlock_all_bitmaps = unlock_all_bitmaps,
1107 1108
};

1109 1110 1111 1112
static int __init cluster_init(void)
{
	pr_warn("md-cluster: EXPERIMENTAL. Use with caution\n");
	pr_info("Registering Cluster MD functions\n");
1113
	register_md_cluster_operations(&cluster_ops, THIS_MODULE);
1114 1115 1116 1117 1118
	return 0;
}

static void cluster_exit(void)
{
1119
	unregister_md_cluster_operations();
1120 1121 1122 1123
}

module_init(cluster_init);
module_exit(cluster_exit);
1124
MODULE_AUTHOR("SUSE");
1125 1126
MODULE_LICENSE("GPL");
MODULE_DESCRIPTION("Clustering support for MD");