drivers/md/md-cluster.c

188 	struct md_cluster_info *cinfo = mddev->cluster_info;
195 	res->ls = cinfo->lockspace;
272 	struct md_cluster_info *cinfo = mddev->cluster_info;
278 		cinfo->suspend_hi = le64_to_cpu(ri.hi);
279 		cinfo->suspend_lo = le64_to_cpu(ri.lo);
289 	struct md_cluster_info *cinfo = mddev->cluster_info;
295 	while (cinfo->recovery_map) {
296 		slot = fls64((u64)cinfo->recovery_map) - 1;
318 		spin_lock_irq(&cinfo->suspend_lock);
319 		cinfo->suspend_hi = 0;
320 		cinfo->suspend_lo = 0;
321 		cinfo->suspend_from = -1;
322 		spin_unlock_irq(&cinfo->suspend_lock);
348 		clear_bit(slot, &cinfo->recovery_map);
355 	struct md_cluster_info *cinfo = mddev->cluster_info;
356 	set_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state);
361 	struct md_cluster_info *cinfo = mddev->cluster_info;
363 	set_bit(slot, &cinfo->recovery_map);
364 	if (!cinfo->recovery_thread) {
365 		rcu_assign_pointer(cinfo->recovery_thread,
367 		if (!cinfo->recovery_thread) {
372 	md_wakeup_thread(cinfo->recovery_thread);
378 	struct md_cluster_info *cinfo = mddev->cluster_info;
383 			cinfo->slot_number);
394 	struct md_cluster_info *cinfo = mddev->cluster_info;
396 	cinfo->slot_number = our_slot;
399 	if (test_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state)) {
400 		complete(&cinfo->completion);
401 		clear_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state);
403 	clear_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state);
422 	struct md_cluster_info *cinfo = res->mddev->cluster_info;
425 		if (test_bit(MD_CLUSTER_ALREADY_IN_CLUSTER, &cinfo->state))
426 			md_wakeup_thread(cinfo->recv_thread);
428 			set_bit(MD_CLUSTER_PENDING_RECV_EVENT, &cinfo->state);
434 	struct md_cluster_info *cinfo = mddev->cluster_info;
436 	spin_lock_irq(&cinfo->suspend_lock);
437 	cinfo->suspend_hi = 0;
438 	cinfo->suspend_lo = 0;
439 	spin_unlock_irq(&cinfo->suspend_lock);
446 	struct md_cluster_info *cinfo = mddev->cluster_info;
490 		md_bitmap_sync_with_cluster(mddev, cinfo->sync_low,
491 					    cinfo->sync_hi, lo, hi);
492 	cinfo->sync_low = lo;
493 	cinfo->sync_hi = hi;
496 	spin_lock_irq(&cinfo->suspend_lock);
497 	cinfo->suspend_from = slot;
498 	cinfo->suspend_lo = lo;
499 	cinfo->suspend_hi = hi;
500 	spin_unlock_irq(&cinfo->suspend_lock);
507 	struct md_cluster_info *cinfo = mddev->cluster_info;
518 	init_completion(&cinfo->newdisk_completion);
519 	set_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state);
521 	if (!wait_for_completion_timeout(&cinfo->newdisk_completion,
527 	clear_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state);
536 	struct md_cluster_info *cinfo = mddev->cluster_info;
539 	dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR);
545 		    test_bit(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state));
633 	struct md_cluster_info *cinfo = thread->mddev->cluster_info;
634 	struct dlm_lock_resource *ack_lockres = cinfo->ack_lockres;
635 	struct dlm_lock_resource *message_lockres = cinfo->message_lockres;
639 	mutex_lock(&cinfo->recv_mutex);
643 		mutex_unlock(&cinfo->recv_mutex);
670 	mutex_unlock(&cinfo->recv_mutex);
677 static int lock_token(struct md_cluster_info *cinfo)
681 	error = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
687 		mutex_lock(&cinfo->recv_mutex);
695 static int lock_comm(struct md_cluster_info *cinfo, bool mddev_locked)
698 	struct mddev *mddev = cinfo->mddev;
707 				      &cinfo->state)) {
709 					      &cinfo->state);
715 	wait_event(cinfo->wait,
716 		   !test_and_set_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state));
717 	rv = lock_token(cinfo);
719 		clear_bit_unlock(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state);
723 static void unlock_comm(struct md_cluster_info *cinfo)
725 	WARN_ON(cinfo->token_lockres->mode != DLM_LOCK_EX);
726 	mutex_unlock(&cinfo->recv_mutex);
727 	dlm_unlock_sync(cinfo->token_lockres);
728 	clear_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state);
729 	wake_up(&cinfo->wait);
744 static int __sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
747 	int slot = cinfo->slot_number - 1;
751 	error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_EX);
757 	memcpy(cinfo->message_lockres->lksb.sb_lvbptr, (void *)cmsg,
760 	error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_CW);
768 	error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_EX);
776 	error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR);
784 	error = dlm_unlock_sync(cinfo->message_lockres);
795 static int sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg,
800 	ret = lock_comm(cinfo, mddev_locked);
802 		ret = __sendmsg(cinfo, cmsg);
803 		unlock_comm(cinfo);
810 	struct md_cluster_info *cinfo = mddev->cluster_info;
823 		if (i == (cinfo->slot_number - 1)) {
834 					(unsigned long long) cinfo->suspend_lo,
835 					(unsigned long long) cinfo->suspend_hi,
837 				cinfo->suspend_from = i;
869 	struct md_cluster_info *cinfo;
873 	cinfo = kzalloc(sizeof(struct md_cluster_info), GFP_KERNEL);
874 	if (!cinfo)
877 	INIT_LIST_HEAD(&cinfo->suspend_list);
878 	spin_lock_init(&cinfo->suspend_lock);
879 	init_completion(&cinfo->completion);
880 	set_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state);
881 	init_waitqueue_head(&cinfo->wait);
882 	mutex_init(&cinfo->recv_mutex);
884 	mddev->cluster_info = cinfo;
885 	cinfo->mddev = mddev;
891 				&ops_rv, &cinfo->lockspace);
894 	wait_for_completion(&cinfo->completion);
895 	if (nodes < cinfo->slot_number) {
897 			cinfo->slot_number, nodes);
903 	rcu_assign_pointer(cinfo->recv_thread,
905 	if (!cinfo->recv_thread) {
909 	cinfo->message_lockres = lockres_init(mddev, "message", NULL, 1);
910 	if (!cinfo->message_lockres)
912 	cinfo->token_lockres = lockres_init(mddev, "token", NULL, 0);
913 	if (!cinfo->token_lockres)
915 	cinfo->no_new_dev_lockres = lockres_init(mddev, "no-new-dev", NULL, 0);
916 	if (!cinfo->no_new_dev_lockres)
919 	ret = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
925 	cinfo->ack_lockres = lockres_init(mddev, "ack", ack_bast, 0);
926 	if (!cinfo->ack_lockres) {
931 	if (dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR))
934 	dlm_unlock_sync(cinfo->token_lockres);
936 	if (dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR))
940 	pr_info("md-cluster: Joined cluster %s slot %d\n", str, cinfo->slot_number);
941 	snprintf(str, 64, "bitmap%04d", cinfo->slot_number - 1);
942 	cinfo->bitmap_lockres = lockres_init(mddev, str, NULL, 1);
943 	if (!cinfo->bitmap_lockres) {
947 	if (dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW)) {
953 	cinfo->resync_lockres = lockres_init(mddev, "resync", NULL, 0);
954 	if (!cinfo->resync_lockres) {
961 	set_bit(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state);
962 	md_unregister_thread(mddev, &cinfo->recovery_thread);
963 	md_unregister_thread(mddev, &cinfo->recv_thread);
964 	lockres_free(cinfo->message_lockres);
965 	lockres_free(cinfo->token_lockres);
966 	lockres_free(cinfo->ack_lockres);
967 	lockres_free(cinfo->no_new_dev_lockres);
968 	lockres_free(cinfo->resync_lockres);
969 	lockres_free(cinfo->bitmap_lockres);
970 	if (cinfo->lockspace)
971 		dlm_release_lockspace(cinfo->lockspace, 2);
973 	kfree(cinfo);
979 	struct md_cluster_info *cinfo = mddev->cluster_info;
984 	set_bit(MD_CLUSTER_ALREADY_IN_CLUSTER, &cinfo->state);
986 	if (test_and_clear_bit(MD_CLUSTER_PENDING_RECV_EVENT, &cinfo->state))
987 		md_wakeup_thread(cinfo->recv_thread);
992 	struct md_cluster_info *cinfo = mddev->cluster_info;
997 	err = sendmsg(cinfo, &cmsg, 1);
1006 	struct md_cluster_info *cinfo = mddev->cluster_info;
1008 	if (!cinfo)
1019 	if ((cinfo->slot_number > 0 && mddev->recovery_cp != MaxSector) ||
1024 	set_bit(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state);
1025 	md_unregister_thread(mddev, &cinfo->recovery_thread);
1026 	md_unregister_thread(mddev, &cinfo->recv_thread);
1027 	lockres_free(cinfo->message_lockres);
1028 	lockres_free(cinfo->token_lockres);
1029 	lockres_free(cinfo->ack_lockres);
1030 	lockres_free(cinfo->no_new_dev_lockres);
1031 	lockres_free(cinfo->resync_lockres);
1032 	lockres_free(cinfo->bitmap_lockres);
1034 	dlm_release_lockspace(cinfo->lockspace, 2);
1035 	kfree(cinfo);
1045 	struct md_cluster_info *cinfo = mddev->cluster_info;
1047 	return cinfo->slot_number - 1;
1058 	struct md_cluster_info *cinfo = mddev->cluster_info;
1066 				    &cinfo->state);
1070 	wait_event(cinfo->wait,
1071 		   !test_and_set_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state) ||
1072 		   test_and_clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state));
1075 	if (cinfo->token_lockres->mode == DLM_LOCK_EX) {
1076 		clear_bit_unlock(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state);
1080 	ret = lock_token(cinfo);
1081 	clear_bit_unlock(MD_CLUSTER_HOLDING_MUTEX_FOR_RECVD, &cinfo->state);
1087 	struct md_cluster_info *cinfo = mddev->cluster_info;
1104 		ret = __sendmsg(cinfo, &cmsg);
1107 	clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
1108 	unlock_comm(cinfo);
1114 	struct md_cluster_info *cinfo = mddev->cluster_info;
1115 	clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
1116 	unlock_comm(cinfo);
1121 	struct md_cluster_info *cinfo = mddev->cluster_info;
1127 	ret = sendmsg(cinfo, &cmsg, 0);
1267 	struct md_cluster_info *cinfo = mddev->cluster_info;
1274 	if (lock_comm(cinfo, 1)) {
1293 		ret = __sendmsg(cinfo, &cmsg);
1297 			unlock_comm(cinfo);
1302 		unlock_comm(cinfo);
1314 		ret = __sendmsg(cinfo, &cmsg);
1322 		ret = __sendmsg(cinfo, &cmsg);
1327 	unlock_comm(cinfo);
1332 	struct md_cluster_info *cinfo = mddev->cluster_info;
1333 	return dlm_lock_sync_interruptible(cinfo->resync_lockres, DLM_LOCK_EX, mddev);
1338 	struct md_cluster_info *cinfo = mddev->cluster_info;
1340 	spin_lock_irq(&cinfo->suspend_lock);
1341 	*lo = cinfo->suspend_lo;
1342 	*hi = cinfo->suspend_hi;
1343 	spin_unlock_irq(&cinfo->suspend_lock);
1348 	struct md_cluster_info *cinfo = mddev->cluster_info;
1354 		memcpy(&ri, cinfo->bitmap_lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
1359 	add_resync_info(cinfo->bitmap_lockres, lo, hi);
1361 	dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW);
1371 		return sendmsg(cinfo, &cmsg, 1);
1373 		return sendmsg(cinfo, &cmsg, 0);
1378 	struct md_cluster_info *cinfo = mddev->cluster_info;
1389 	dlm_unlock_sync(cinfo->resync_lockres);
1396 	struct md_cluster_info *cinfo = mddev->cluster_info;
1400 		test_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state))
1403 	spin_lock_irq(&cinfo->suspend_lock);
1404 	if (hi > cinfo->suspend_lo && lo < cinfo->suspend_hi)
1406 	spin_unlock_irq(&cinfo->suspend_lock);
1416 	struct md_cluster_info *cinfo = mddev->cluster_info;
1426 	if (lock_comm(cinfo, 1))
1428 	ret = __sendmsg(cinfo, &cmsg);
1430 		unlock_comm(cinfo);
1433 	cinfo->no_new_dev_lockres->flags |= DLM_LKF_NOQUEUE;
1434 	ret = dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_EX);
1435 	cinfo->no_new_dev_lockres->flags &= ~DLM_LKF_NOQUEUE;
1440 		unlock_comm(cinfo);
1442 		dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR);
1455 		set_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
1456 		wake_up(&cinfo->wait);
1463 	struct md_cluster_info *cinfo = mddev->cluster_info;
1464 	clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state);
1465 	unlock_comm(cinfo);
1470 	struct md_cluster_info *cinfo = mddev->cluster_info;
1472 	if (!test_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state)) {
1478 		dlm_unlock_sync(cinfo->no_new_dev_lockres);
1479 	complete(&cinfo->newdisk_completion);
1486 	struct md_cluster_info *cinfo = mddev->cluster_info;
1489 	return sendmsg(cinfo, &cmsg, 1);
1496 	struct md_cluster_info *cinfo = mddev->cluster_info;
1498 	cinfo->other_bitmap_lockres =
1501 	if (!cinfo->other_bitmap_lockres) {
1513 		cinfo->other_bitmap_lockres[i] = lockres_init(mddev, str, NULL, 1);
1514 		if (!cinfo->other_bitmap_lockres[i])
1517 		cinfo->other_bitmap_lockres[i]->flags |= DLM_LKF_NOQUEUE;
1518 		ret = dlm_lock_sync(cinfo->other_bitmap_lockres[i], DLM_LOCK_PW);
1529 	struct md_cluster_info *cinfo = mddev->cluster_info;
1533 	if (cinfo->other_bitmap_lockres) {
1535 			if (cinfo->other_bitmap_lockres[i]) {
1536 				lockres_free(cinfo->other_bitmap_lockres[i]);
1539 		kfree(cinfo->other_bitmap_lockres);
1540 		cinfo->other_bitmap_lockres = NULL;
1550 	struct md_cluster_info *cinfo = mddev->cluster_info;
1554 	err = sendmsg(cinfo, &cmsg, 1);
1559 		if (sn == (cinfo->slot_number - 1))