1 /* 2 * Copyright (C) 2015, SUSE 3 * 4 * This program is free software; you can redistribute it and/or modify 5 * it under the terms of the GNU General Public License as published by 6 * the Free Software Foundation; either version 2, or (at your option) 7 * any later version. 8 * 9 */ 10 11 12 #include <linux/module.h> 13 #include <linux/dlm.h> 14 #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 struct dlm_lock_resource { 24 dlm_lockspace_t *ls; 25 struct dlm_lksb lksb; 26 char *name; /* lock name. */ 27 uint32_t flags; /* flags to pass to dlm_lock() */ 28 struct completion completion; /* completion for synchronized locking */ 29 void (*bast)(void *arg, int mode); /* blocking AST function pointer*/ 30 struct mddev *mddev; /* pointing back to mddev. */ 31 }; 32 33 struct suspend_info { 34 int slot; 35 sector_t lo; 36 sector_t hi; 37 struct list_head list; 38 }; 39 40 struct resync_info { 41 __le64 lo; 42 __le64 hi; 43 }; 44 45 /* md_cluster_info flags */ 46 #define MD_CLUSTER_WAITING_FOR_NEWDISK 1 47 #define MD_CLUSTER_SUSPEND_READ_BALANCING 2 48 49 50 struct md_cluster_info { 51 /* dlm lock space and resources for clustered raid. */ 52 dlm_lockspace_t *lockspace; 53 int slot_number; 54 struct completion completion; 55 struct mutex sb_mutex; 56 struct dlm_lock_resource *bitmap_lockres; 57 struct list_head suspend_list; 58 spinlock_t suspend_lock; 59 struct md_thread *recovery_thread; 60 unsigned long recovery_map; 61 /* communication loc resources */ 62 struct dlm_lock_resource *ack_lockres; 63 struct dlm_lock_resource *message_lockres; 64 struct dlm_lock_resource *token_lockres; 65 struct dlm_lock_resource *no_new_dev_lockres; 66 struct md_thread *recv_thread; 67 struct completion newdisk_completion; 68 unsigned long state; 69 }; 70 71 enum msg_type { 72 METADATA_UPDATED = 0, 73 RESYNCING, 74 NEWDISK, 75 REMOVE, 76 RE_ADD, 77 BITMAP_NEEDS_SYNC, 78 }; 79 80 struct cluster_msg { 81 int type; 82 int slot; 83 /* TODO: Unionize this for smaller footprint */ 84 sector_t low; 85 sector_t high; 86 char uuid[16]; 87 int raid_slot; 88 }; 89 90 static void sync_ast(void *arg) 91 { 92 struct dlm_lock_resource *res; 93 94 res = (struct dlm_lock_resource *) arg; 95 complete(&res->completion); 96 } 97 98 static int dlm_lock_sync(struct dlm_lock_resource *res, int mode) 99 { 100 int ret = 0; 101 102 ret = dlm_lock(res->ls, mode, &res->lksb, 103 res->flags, res->name, strlen(res->name), 104 0, sync_ast, res, res->bast); 105 if (ret) 106 return ret; 107 wait_for_completion(&res->completion); 108 return res->lksb.sb_status; 109 } 110 111 static int dlm_unlock_sync(struct dlm_lock_resource *res) 112 { 113 return dlm_lock_sync(res, DLM_LOCK_NL); 114 } 115 116 static struct dlm_lock_resource *lockres_init(struct mddev *mddev, 117 char *name, void (*bastfn)(void *arg, int mode), int with_lvb) 118 { 119 struct dlm_lock_resource *res = NULL; 120 int ret, namelen; 121 struct md_cluster_info *cinfo = mddev->cluster_info; 122 123 res = kzalloc(sizeof(struct dlm_lock_resource), GFP_KERNEL); 124 if (!res) 125 return NULL; 126 init_completion(&res->completion); 127 res->ls = cinfo->lockspace; 128 res->mddev = mddev; 129 namelen = strlen(name); 130 res->name = kzalloc(namelen + 1, GFP_KERNEL); 131 if (!res->name) { 132 pr_err("md-cluster: Unable to allocate resource name for resource %s\n", name); 133 goto out_err; 134 } 135 strlcpy(res->name, name, namelen + 1); 136 if (with_lvb) { 137 res->lksb.sb_lvbptr = kzalloc(LVB_SIZE, GFP_KERNEL); 138 if (!res->lksb.sb_lvbptr) { 139 pr_err("md-cluster: Unable to allocate LVB for resource %s\n", name); 140 goto out_err; 141 } 142 res->flags = DLM_LKF_VALBLK; 143 } 144 145 if (bastfn) 146 res->bast = bastfn; 147 148 res->flags |= DLM_LKF_EXPEDITE; 149 150 ret = dlm_lock_sync(res, DLM_LOCK_NL); 151 if (ret) { 152 pr_err("md-cluster: Unable to lock NL on new lock resource %s\n", name); 153 goto out_err; 154 } 155 res->flags &= ~DLM_LKF_EXPEDITE; 156 res->flags |= DLM_LKF_CONVERT; 157 158 return res; 159 out_err: 160 kfree(res->lksb.sb_lvbptr); 161 kfree(res->name); 162 kfree(res); 163 return NULL; 164 } 165 166 static void lockres_free(struct dlm_lock_resource *res) 167 { 168 int ret; 169 170 if (!res) 171 return; 172 173 /* cancel a lock request or a conversion request that is blocked */ 174 res->flags |= DLM_LKF_CANCEL; 175 retry: 176 ret = dlm_unlock(res->ls, res->lksb.sb_lkid, 0, &res->lksb, res); 177 if (unlikely(ret != 0)) { 178 pr_info("%s: failed to unlock %s return %d\n", __func__, res->name, ret); 179 180 /* if a lock conversion is cancelled, then the lock is put 181 * back to grant queue, need to ensure it is unlocked */ 182 if (ret == -DLM_ECANCEL) 183 goto retry; 184 } 185 res->flags &= ~DLM_LKF_CANCEL; 186 wait_for_completion(&res->completion); 187 188 kfree(res->name); 189 kfree(res->lksb.sb_lvbptr); 190 kfree(res); 191 } 192 193 static void add_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres, 194 sector_t lo, sector_t hi) 195 { 196 struct resync_info *ri; 197 198 ri = (struct resync_info *)lockres->lksb.sb_lvbptr; 199 ri->lo = cpu_to_le64(lo); 200 ri->hi = cpu_to_le64(hi); 201 } 202 203 static struct suspend_info *read_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres) 204 { 205 struct resync_info ri; 206 struct suspend_info *s = NULL; 207 sector_t hi = 0; 208 209 dlm_lock_sync(lockres, DLM_LOCK_CR); 210 memcpy(&ri, lockres->lksb.sb_lvbptr, sizeof(struct resync_info)); 211 hi = le64_to_cpu(ri.hi); 212 if (ri.hi > 0) { 213 s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL); 214 if (!s) 215 goto out; 216 s->hi = hi; 217 s->lo = le64_to_cpu(ri.lo); 218 } 219 dlm_unlock_sync(lockres); 220 out: 221 return s; 222 } 223 224 static void recover_bitmaps(struct md_thread *thread) 225 { 226 struct mddev *mddev = thread->mddev; 227 struct md_cluster_info *cinfo = mddev->cluster_info; 228 struct dlm_lock_resource *bm_lockres; 229 char str[64]; 230 int slot, ret; 231 struct suspend_info *s, *tmp; 232 sector_t lo, hi; 233 234 while (cinfo->recovery_map) { 235 slot = fls64((u64)cinfo->recovery_map) - 1; 236 237 /* Clear suspend_area associated with the bitmap */ 238 spin_lock_irq(&cinfo->suspend_lock); 239 list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list) 240 if (slot == s->slot) { 241 list_del(&s->list); 242 kfree(s); 243 } 244 spin_unlock_irq(&cinfo->suspend_lock); 245 246 snprintf(str, 64, "bitmap%04d", slot); 247 bm_lockres = lockres_init(mddev, str, NULL, 1); 248 if (!bm_lockres) { 249 pr_err("md-cluster: Cannot initialize bitmaps\n"); 250 goto clear_bit; 251 } 252 253 ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW); 254 if (ret) { 255 pr_err("md-cluster: Could not DLM lock %s: %d\n", 256 str, ret); 257 goto clear_bit; 258 } 259 ret = bitmap_copy_from_slot(mddev, slot, &lo, &hi, true); 260 if (ret) { 261 pr_err("md-cluster: Could not copy data from bitmap %d\n", slot); 262 goto dlm_unlock; 263 } 264 if (hi > 0) { 265 /* TODO:Wait for current resync to get over */ 266 set_bit(MD_RECOVERY_NEEDED, &mddev->recovery); 267 if (lo < mddev->recovery_cp) 268 mddev->recovery_cp = lo; 269 md_check_recovery(mddev); 270 } 271 dlm_unlock: 272 dlm_unlock_sync(bm_lockres); 273 clear_bit: 274 clear_bit(slot, &cinfo->recovery_map); 275 } 276 } 277 278 static void recover_prep(void *arg) 279 { 280 struct mddev *mddev = arg; 281 struct md_cluster_info *cinfo = mddev->cluster_info; 282 set_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state); 283 } 284 285 static void __recover_slot(struct mddev *mddev, int slot) 286 { 287 struct md_cluster_info *cinfo = mddev->cluster_info; 288 289 set_bit(slot, &cinfo->recovery_map); 290 if (!cinfo->recovery_thread) { 291 cinfo->recovery_thread = md_register_thread(recover_bitmaps, 292 mddev, "recover"); 293 if (!cinfo->recovery_thread) { 294 pr_warn("md-cluster: Could not create recovery thread\n"); 295 return; 296 } 297 } 298 md_wakeup_thread(cinfo->recovery_thread); 299 } 300 301 static void recover_slot(void *arg, struct dlm_slot *slot) 302 { 303 struct mddev *mddev = arg; 304 struct md_cluster_info *cinfo = mddev->cluster_info; 305 306 pr_info("md-cluster: %s Node %d/%d down. My slot: %d. Initiating recovery.\n", 307 mddev->bitmap_info.cluster_name, 308 slot->nodeid, slot->slot, 309 cinfo->slot_number); 310 /* deduct one since dlm slot starts from one while the num of 311 * cluster-md begins with 0 */ 312 __recover_slot(mddev, slot->slot - 1); 313 } 314 315 static void recover_done(void *arg, struct dlm_slot *slots, 316 int num_slots, int our_slot, 317 uint32_t generation) 318 { 319 struct mddev *mddev = arg; 320 struct md_cluster_info *cinfo = mddev->cluster_info; 321 322 cinfo->slot_number = our_slot; 323 complete(&cinfo->completion); 324 clear_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state); 325 } 326 327 static const struct dlm_lockspace_ops md_ls_ops = { 328 .recover_prep = recover_prep, 329 .recover_slot = recover_slot, 330 .recover_done = recover_done, 331 }; 332 333 /* 334 * The BAST function for the ack lock resource 335 * This function wakes up the receive thread in 336 * order to receive and process the message. 337 */ 338 static void ack_bast(void *arg, int mode) 339 { 340 struct dlm_lock_resource *res = (struct dlm_lock_resource *)arg; 341 struct md_cluster_info *cinfo = res->mddev->cluster_info; 342 343 if (mode == DLM_LOCK_EX) 344 md_wakeup_thread(cinfo->recv_thread); 345 } 346 347 static void __remove_suspend_info(struct md_cluster_info *cinfo, int slot) 348 { 349 struct suspend_info *s, *tmp; 350 351 list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list) 352 if (slot == s->slot) { 353 pr_info("%s:%d Deleting suspend_info: %d\n", 354 __func__, __LINE__, slot); 355 list_del(&s->list); 356 kfree(s); 357 break; 358 } 359 } 360 361 static void remove_suspend_info(struct md_cluster_info *cinfo, int slot) 362 { 363 spin_lock_irq(&cinfo->suspend_lock); 364 __remove_suspend_info(cinfo, slot); 365 spin_unlock_irq(&cinfo->suspend_lock); 366 } 367 368 369 static void process_suspend_info(struct md_cluster_info *cinfo, 370 int slot, sector_t lo, sector_t hi) 371 { 372 struct suspend_info *s; 373 374 if (!hi) { 375 remove_suspend_info(cinfo, slot); 376 return; 377 } 378 s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL); 379 if (!s) 380 return; 381 s->slot = slot; 382 s->lo = lo; 383 s->hi = hi; 384 spin_lock_irq(&cinfo->suspend_lock); 385 /* Remove existing entry (if exists) before adding */ 386 __remove_suspend_info(cinfo, slot); 387 list_add(&s->list, &cinfo->suspend_list); 388 spin_unlock_irq(&cinfo->suspend_lock); 389 } 390 391 static void process_add_new_disk(struct mddev *mddev, struct cluster_msg *cmsg) 392 { 393 char disk_uuid[64]; 394 struct md_cluster_info *cinfo = mddev->cluster_info; 395 char event_name[] = "EVENT=ADD_DEVICE"; 396 char raid_slot[16]; 397 char *envp[] = {event_name, disk_uuid, raid_slot, NULL}; 398 int len; 399 400 len = snprintf(disk_uuid, 64, "DEVICE_UUID="); 401 sprintf(disk_uuid + len, "%pU", cmsg->uuid); 402 snprintf(raid_slot, 16, "RAID_DISK=%d", cmsg->raid_slot); 403 pr_info("%s:%d Sending kobject change with %s and %s\n", __func__, __LINE__, disk_uuid, raid_slot); 404 init_completion(&cinfo->newdisk_completion); 405 set_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state); 406 kobject_uevent_env(&disk_to_dev(mddev->gendisk)->kobj, KOBJ_CHANGE, envp); 407 wait_for_completion_timeout(&cinfo->newdisk_completion, 408 NEW_DEV_TIMEOUT); 409 clear_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state); 410 } 411 412 413 static void process_metadata_update(struct mddev *mddev, struct cluster_msg *msg) 414 { 415 struct md_cluster_info *cinfo = mddev->cluster_info; 416 417 md_reload_sb(mddev); 418 dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR); 419 } 420 421 static void process_remove_disk(struct mddev *mddev, struct cluster_msg *msg) 422 { 423 struct md_rdev *rdev = md_find_rdev_nr_rcu(mddev, msg->raid_slot); 424 425 if (rdev) 426 md_kick_rdev_from_array(rdev); 427 else 428 pr_warn("%s: %d Could not find disk(%d) to REMOVE\n", __func__, __LINE__, msg->raid_slot); 429 } 430 431 static void process_readd_disk(struct mddev *mddev, struct cluster_msg *msg) 432 { 433 struct md_rdev *rdev = md_find_rdev_nr_rcu(mddev, msg->raid_slot); 434 435 if (rdev && test_bit(Faulty, &rdev->flags)) 436 clear_bit(Faulty, &rdev->flags); 437 else 438 pr_warn("%s: %d Could not find disk(%d) which is faulty", __func__, __LINE__, msg->raid_slot); 439 } 440 441 static void process_recvd_msg(struct mddev *mddev, struct cluster_msg *msg) 442 { 443 switch (msg->type) { 444 case METADATA_UPDATED: 445 pr_info("%s: %d Received message: METADATA_UPDATE from %d\n", 446 __func__, __LINE__, msg->slot); 447 process_metadata_update(mddev, msg); 448 break; 449 case RESYNCING: 450 pr_info("%s: %d Received message: RESYNCING from %d\n", 451 __func__, __LINE__, msg->slot); 452 process_suspend_info(mddev->cluster_info, msg->slot, 453 msg->low, msg->high); 454 break; 455 case NEWDISK: 456 pr_info("%s: %d Received message: NEWDISK from %d\n", 457 __func__, __LINE__, msg->slot); 458 process_add_new_disk(mddev, msg); 459 break; 460 case REMOVE: 461 pr_info("%s: %d Received REMOVE from %d\n", 462 __func__, __LINE__, msg->slot); 463 process_remove_disk(mddev, msg); 464 break; 465 case RE_ADD: 466 pr_info("%s: %d Received RE_ADD from %d\n", 467 __func__, __LINE__, msg->slot); 468 process_readd_disk(mddev, msg); 469 break; 470 case BITMAP_NEEDS_SYNC: 471 pr_info("%s: %d Received BITMAP_NEEDS_SYNC from %d\n", 472 __func__, __LINE__, msg->slot); 473 __recover_slot(mddev, msg->slot); 474 break; 475 default: 476 pr_warn("%s:%d Received unknown message from %d\n", 477 __func__, __LINE__, msg->slot); 478 } 479 } 480 481 /* 482 * thread for receiving message 483 */ 484 static void recv_daemon(struct md_thread *thread) 485 { 486 struct md_cluster_info *cinfo = thread->mddev->cluster_info; 487 struct dlm_lock_resource *ack_lockres = cinfo->ack_lockres; 488 struct dlm_lock_resource *message_lockres = cinfo->message_lockres; 489 struct cluster_msg msg; 490 int ret; 491 492 /*get CR on Message*/ 493 if (dlm_lock_sync(message_lockres, DLM_LOCK_CR)) { 494 pr_err("md/raid1:failed to get CR on MESSAGE\n"); 495 return; 496 } 497 498 /* read lvb and wake up thread to process this message_lockres */ 499 memcpy(&msg, message_lockres->lksb.sb_lvbptr, sizeof(struct cluster_msg)); 500 process_recvd_msg(thread->mddev, &msg); 501 502 /*release CR on ack_lockres*/ 503 ret = dlm_unlock_sync(ack_lockres); 504 if (unlikely(ret != 0)) 505 pr_info("unlock ack failed return %d\n", ret); 506 /*up-convert to PR on message_lockres*/ 507 ret = dlm_lock_sync(message_lockres, DLM_LOCK_PR); 508 if (unlikely(ret != 0)) 509 pr_info("lock PR on msg failed return %d\n", ret); 510 /*get CR on ack_lockres again*/ 511 ret = dlm_lock_sync(ack_lockres, DLM_LOCK_CR); 512 if (unlikely(ret != 0)) 513 pr_info("lock CR on ack failed return %d\n", ret); 514 /*release CR on message_lockres*/ 515 ret = dlm_unlock_sync(message_lockres); 516 if (unlikely(ret != 0)) 517 pr_info("unlock msg failed return %d\n", ret); 518 } 519 520 /* lock_comm() 521 * Takes the lock on the TOKEN lock resource so no other 522 * node can communicate while the operation is underway. 523 */ 524 static int lock_comm(struct md_cluster_info *cinfo) 525 { 526 int error; 527 528 error = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX); 529 if (error) 530 pr_err("md-cluster(%s:%d): failed to get EX on TOKEN (%d)\n", 531 __func__, __LINE__, error); 532 return error; 533 } 534 535 static void unlock_comm(struct md_cluster_info *cinfo) 536 { 537 dlm_unlock_sync(cinfo->token_lockres); 538 } 539 540 /* __sendmsg() 541 * This function performs the actual sending of the message. This function is 542 * usually called after performing the encompassing operation 543 * The function: 544 * 1. Grabs the message lockresource in EX mode 545 * 2. Copies the message to the message LVB 546 * 3. Downconverts message lockresource to CW 547 * 4. Upconverts ack lock resource from CR to EX. This forces the BAST on other nodes 548 * and the other nodes read the message. The thread will wait here until all other 549 * nodes have released ack lock resource. 550 * 5. Downconvert ack lockresource to CR 551 */ 552 static int __sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg) 553 { 554 int error; 555 int slot = cinfo->slot_number - 1; 556 557 cmsg->slot = cpu_to_le32(slot); 558 /*get EX on Message*/ 559 error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_EX); 560 if (error) { 561 pr_err("md-cluster: failed to get EX on MESSAGE (%d)\n", error); 562 goto failed_message; 563 } 564 565 memcpy(cinfo->message_lockres->lksb.sb_lvbptr, (void *)cmsg, 566 sizeof(struct cluster_msg)); 567 /*down-convert EX to CW on Message*/ 568 error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_CW); 569 if (error) { 570 pr_err("md-cluster: failed to convert EX to CW on MESSAGE(%d)\n", 571 error); 572 goto failed_ack; 573 } 574 575 /*up-convert CR to EX on Ack*/ 576 error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_EX); 577 if (error) { 578 pr_err("md-cluster: failed to convert CR to EX on ACK(%d)\n", 579 error); 580 goto failed_ack; 581 } 582 583 /*down-convert EX to CR on Ack*/ 584 error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR); 585 if (error) { 586 pr_err("md-cluster: failed to convert EX to CR on ACK(%d)\n", 587 error); 588 goto failed_ack; 589 } 590 591 failed_ack: 592 error = dlm_unlock_sync(cinfo->message_lockres); 593 if (unlikely(error != 0)) { 594 pr_err("md-cluster: failed convert to NL on MESSAGE(%d)\n", 595 error); 596 /* in case the message can't be released due to some reason */ 597 goto failed_ack; 598 } 599 failed_message: 600 return error; 601 } 602 603 static int sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg) 604 { 605 int ret; 606 607 lock_comm(cinfo); 608 ret = __sendmsg(cinfo, cmsg); 609 unlock_comm(cinfo); 610 return ret; 611 } 612 613 static int gather_all_resync_info(struct mddev *mddev, int total_slots) 614 { 615 struct md_cluster_info *cinfo = mddev->cluster_info; 616 int i, ret = 0; 617 struct dlm_lock_resource *bm_lockres; 618 struct suspend_info *s; 619 char str[64]; 620 621 622 for (i = 0; i < total_slots; i++) { 623 memset(str, '\0', 64); 624 snprintf(str, 64, "bitmap%04d", i); 625 bm_lockres = lockres_init(mddev, str, NULL, 1); 626 if (!bm_lockres) 627 return -ENOMEM; 628 if (i == (cinfo->slot_number - 1)) 629 continue; 630 631 bm_lockres->flags |= DLM_LKF_NOQUEUE; 632 ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW); 633 if (ret == -EAGAIN) { 634 memset(bm_lockres->lksb.sb_lvbptr, '\0', LVB_SIZE); 635 s = read_resync_info(mddev, bm_lockres); 636 if (s) { 637 pr_info("%s:%d Resync[%llu..%llu] in progress on %d\n", 638 __func__, __LINE__, 639 (unsigned long long) s->lo, 640 (unsigned long long) s->hi, i); 641 spin_lock_irq(&cinfo->suspend_lock); 642 s->slot = i; 643 list_add(&s->list, &cinfo->suspend_list); 644 spin_unlock_irq(&cinfo->suspend_lock); 645 } 646 ret = 0; 647 lockres_free(bm_lockres); 648 continue; 649 } 650 if (ret) { 651 lockres_free(bm_lockres); 652 goto out; 653 } 654 /* TODO: Read the disk bitmap sb and check if it needs recovery */ 655 dlm_unlock_sync(bm_lockres); 656 lockres_free(bm_lockres); 657 } 658 out: 659 return ret; 660 } 661 662 static int join(struct mddev *mddev, int nodes) 663 { 664 struct md_cluster_info *cinfo; 665 int ret, ops_rv; 666 char str[64]; 667 668 if (!try_module_get(THIS_MODULE)) 669 return -ENOENT; 670 671 cinfo = kzalloc(sizeof(struct md_cluster_info), GFP_KERNEL); 672 if (!cinfo) 673 return -ENOMEM; 674 675 INIT_LIST_HEAD(&cinfo->suspend_list); 676 spin_lock_init(&cinfo->suspend_lock); 677 init_completion(&cinfo->completion); 678 679 mutex_init(&cinfo->sb_mutex); 680 mddev->cluster_info = cinfo; 681 682 memset(str, 0, 64); 683 sprintf(str, "%pU", mddev->uuid); 684 ret = dlm_new_lockspace(str, mddev->bitmap_info.cluster_name, 685 DLM_LSFL_FS, LVB_SIZE, 686 &md_ls_ops, mddev, &ops_rv, &cinfo->lockspace); 687 if (ret) 688 goto err; 689 wait_for_completion(&cinfo->completion); 690 if (nodes < cinfo->slot_number) { 691 pr_err("md-cluster: Slot allotted(%d) is greater than available slots(%d).", 692 cinfo->slot_number, nodes); 693 ret = -ERANGE; 694 goto err; 695 } 696 /* Initiate the communication resources */ 697 ret = -ENOMEM; 698 cinfo->recv_thread = md_register_thread(recv_daemon, mddev, "cluster_recv"); 699 if (!cinfo->recv_thread) { 700 pr_err("md-cluster: cannot allocate memory for recv_thread!\n"); 701 goto err; 702 } 703 cinfo->message_lockres = lockres_init(mddev, "message", NULL, 1); 704 if (!cinfo->message_lockres) 705 goto err; 706 cinfo->token_lockres = lockres_init(mddev, "token", NULL, 0); 707 if (!cinfo->token_lockres) 708 goto err; 709 cinfo->ack_lockres = lockres_init(mddev, "ack", ack_bast, 0); 710 if (!cinfo->ack_lockres) 711 goto err; 712 cinfo->no_new_dev_lockres = lockres_init(mddev, "no-new-dev", NULL, 0); 713 if (!cinfo->no_new_dev_lockres) 714 goto err; 715 716 /* get sync CR lock on ACK. */ 717 if (dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR)) 718 pr_err("md-cluster: failed to get a sync CR lock on ACK!(%d)\n", 719 ret); 720 /* get sync CR lock on no-new-dev. */ 721 if (dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR)) 722 pr_err("md-cluster: failed to get a sync CR lock on no-new-dev!(%d)\n", ret); 723 724 725 pr_info("md-cluster: Joined cluster %s slot %d\n", str, cinfo->slot_number); 726 snprintf(str, 64, "bitmap%04d", cinfo->slot_number - 1); 727 cinfo->bitmap_lockres = lockres_init(mddev, str, NULL, 1); 728 if (!cinfo->bitmap_lockres) 729 goto err; 730 if (dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW)) { 731 pr_err("Failed to get bitmap lock\n"); 732 ret = -EINVAL; 733 goto err; 734 } 735 736 ret = gather_all_resync_info(mddev, nodes); 737 if (ret) 738 goto err; 739 740 return 0; 741 err: 742 lockres_free(cinfo->message_lockres); 743 lockres_free(cinfo->token_lockres); 744 lockres_free(cinfo->ack_lockres); 745 lockres_free(cinfo->no_new_dev_lockres); 746 lockres_free(cinfo->bitmap_lockres); 747 if (cinfo->lockspace) 748 dlm_release_lockspace(cinfo->lockspace, 2); 749 mddev->cluster_info = NULL; 750 kfree(cinfo); 751 module_put(THIS_MODULE); 752 return ret; 753 } 754 755 static int leave(struct mddev *mddev) 756 { 757 struct md_cluster_info *cinfo = mddev->cluster_info; 758 759 if (!cinfo) 760 return 0; 761 md_unregister_thread(&cinfo->recovery_thread); 762 md_unregister_thread(&cinfo->recv_thread); 763 lockres_free(cinfo->message_lockres); 764 lockres_free(cinfo->token_lockres); 765 lockres_free(cinfo->ack_lockres); 766 lockres_free(cinfo->no_new_dev_lockres); 767 lockres_free(cinfo->bitmap_lockres); 768 dlm_release_lockspace(cinfo->lockspace, 2); 769 return 0; 770 } 771 772 /* slot_number(): Returns the MD slot number to use 773 * DLM starts the slot numbers from 1, wheras cluster-md 774 * wants the number to be from zero, so we deduct one 775 */ 776 static int slot_number(struct mddev *mddev) 777 { 778 struct md_cluster_info *cinfo = mddev->cluster_info; 779 780 return cinfo->slot_number - 1; 781 } 782 783 static void resync_info_update(struct mddev *mddev, sector_t lo, sector_t hi) 784 { 785 struct md_cluster_info *cinfo = mddev->cluster_info; 786 787 add_resync_info(mddev, cinfo->bitmap_lockres, lo, hi); 788 /* Re-acquire the lock to refresh LVB */ 789 dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW); 790 } 791 792 static int metadata_update_start(struct mddev *mddev) 793 { 794 return lock_comm(mddev->cluster_info); 795 } 796 797 static int metadata_update_finish(struct mddev *mddev) 798 { 799 struct md_cluster_info *cinfo = mddev->cluster_info; 800 struct cluster_msg cmsg; 801 int ret; 802 803 memset(&cmsg, 0, sizeof(cmsg)); 804 cmsg.type = cpu_to_le32(METADATA_UPDATED); 805 ret = __sendmsg(cinfo, &cmsg); 806 unlock_comm(cinfo); 807 return ret; 808 } 809 810 static int metadata_update_cancel(struct mddev *mddev) 811 { 812 struct md_cluster_info *cinfo = mddev->cluster_info; 813 814 return dlm_unlock_sync(cinfo->token_lockres); 815 } 816 817 static int resync_send(struct mddev *mddev, enum msg_type type, 818 sector_t lo, sector_t hi) 819 { 820 struct md_cluster_info *cinfo = mddev->cluster_info; 821 struct cluster_msg cmsg; 822 int slot = cinfo->slot_number - 1; 823 824 pr_info("%s:%d lo: %llu hi: %llu\n", __func__, __LINE__, 825 (unsigned long long)lo, 826 (unsigned long long)hi); 827 resync_info_update(mddev, lo, hi); 828 cmsg.type = cpu_to_le32(type); 829 cmsg.slot = cpu_to_le32(slot); 830 cmsg.low = cpu_to_le64(lo); 831 cmsg.high = cpu_to_le64(hi); 832 return sendmsg(cinfo, &cmsg); 833 } 834 835 static int resync_start(struct mddev *mddev, sector_t lo, sector_t hi) 836 { 837 pr_info("%s:%d\n", __func__, __LINE__); 838 return resync_send(mddev, RESYNCING, lo, hi); 839 } 840 841 static void resync_finish(struct mddev *mddev) 842 { 843 struct md_cluster_info *cinfo = mddev->cluster_info; 844 struct cluster_msg cmsg; 845 int slot = cinfo->slot_number - 1; 846 847 pr_info("%s:%d\n", __func__, __LINE__); 848 resync_send(mddev, RESYNCING, 0, 0); 849 if (test_bit(MD_RECOVERY_INTR, &mddev->recovery)) { 850 cmsg.type = cpu_to_le32(BITMAP_NEEDS_SYNC); 851 cmsg.slot = cpu_to_le32(slot); 852 sendmsg(cinfo, &cmsg); 853 } 854 } 855 856 static int area_resyncing(struct mddev *mddev, int direction, 857 sector_t lo, sector_t hi) 858 { 859 struct md_cluster_info *cinfo = mddev->cluster_info; 860 int ret = 0; 861 struct suspend_info *s; 862 863 if ((direction == READ) && 864 test_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state)) 865 return 1; 866 867 spin_lock_irq(&cinfo->suspend_lock); 868 if (list_empty(&cinfo->suspend_list)) 869 goto out; 870 list_for_each_entry(s, &cinfo->suspend_list, list) 871 if (hi > s->lo && lo < s->hi) { 872 ret = 1; 873 break; 874 } 875 out: 876 spin_unlock_irq(&cinfo->suspend_lock); 877 return ret; 878 } 879 880 static int add_new_disk_start(struct mddev *mddev, struct md_rdev *rdev) 881 { 882 struct md_cluster_info *cinfo = mddev->cluster_info; 883 struct cluster_msg cmsg; 884 int ret = 0; 885 struct mdp_superblock_1 *sb = page_address(rdev->sb_page); 886 char *uuid = sb->device_uuid; 887 888 memset(&cmsg, 0, sizeof(cmsg)); 889 cmsg.type = cpu_to_le32(NEWDISK); 890 memcpy(cmsg.uuid, uuid, 16); 891 cmsg.raid_slot = rdev->desc_nr; 892 lock_comm(cinfo); 893 ret = __sendmsg(cinfo, &cmsg); 894 if (ret) 895 return ret; 896 cinfo->no_new_dev_lockres->flags |= DLM_LKF_NOQUEUE; 897 ret = dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_EX); 898 cinfo->no_new_dev_lockres->flags &= ~DLM_LKF_NOQUEUE; 899 /* Some node does not "see" the device */ 900 if (ret == -EAGAIN) 901 ret = -ENOENT; 902 else 903 dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR); 904 return ret; 905 } 906 907 static int add_new_disk_finish(struct mddev *mddev) 908 { 909 struct cluster_msg cmsg; 910 struct md_cluster_info *cinfo = mddev->cluster_info; 911 int ret; 912 /* Write sb and inform others */ 913 md_update_sb(mddev, 1); 914 cmsg.type = METADATA_UPDATED; 915 ret = __sendmsg(cinfo, &cmsg); 916 unlock_comm(cinfo); 917 return ret; 918 } 919 920 static int new_disk_ack(struct mddev *mddev, bool ack) 921 { 922 struct md_cluster_info *cinfo = mddev->cluster_info; 923 924 if (!test_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state)) { 925 pr_warn("md-cluster(%s): Spurious cluster confirmation\n", mdname(mddev)); 926 return -EINVAL; 927 } 928 929 if (ack) 930 dlm_unlock_sync(cinfo->no_new_dev_lockres); 931 complete(&cinfo->newdisk_completion); 932 return 0; 933 } 934 935 static int remove_disk(struct mddev *mddev, struct md_rdev *rdev) 936 { 937 struct cluster_msg cmsg; 938 struct md_cluster_info *cinfo = mddev->cluster_info; 939 cmsg.type = REMOVE; 940 cmsg.raid_slot = rdev->desc_nr; 941 return __sendmsg(cinfo, &cmsg); 942 } 943 944 static int gather_bitmaps(struct md_rdev *rdev) 945 { 946 int sn, err; 947 sector_t lo, hi; 948 struct cluster_msg cmsg; 949 struct mddev *mddev = rdev->mddev; 950 struct md_cluster_info *cinfo = mddev->cluster_info; 951 952 cmsg.type = RE_ADD; 953 cmsg.raid_slot = rdev->desc_nr; 954 err = sendmsg(cinfo, &cmsg); 955 if (err) 956 goto out; 957 958 for (sn = 0; sn < mddev->bitmap_info.nodes; sn++) { 959 if (sn == (cinfo->slot_number - 1)) 960 continue; 961 err = bitmap_copy_from_slot(mddev, sn, &lo, &hi, false); 962 if (err) { 963 pr_warn("md-cluster: Could not gather bitmaps from slot %d", sn); 964 goto out; 965 } 966 if ((hi > 0) && (lo < mddev->recovery_cp)) 967 mddev->recovery_cp = lo; 968 } 969 out: 970 return err; 971 } 972 973 static struct md_cluster_operations cluster_ops = { 974 .join = join, 975 .leave = leave, 976 .slot_number = slot_number, 977 .resync_info_update = resync_info_update, 978 .resync_start = resync_start, 979 .resync_finish = resync_finish, 980 .metadata_update_start = metadata_update_start, 981 .metadata_update_finish = metadata_update_finish, 982 .metadata_update_cancel = metadata_update_cancel, 983 .area_resyncing = area_resyncing, 984 .add_new_disk_start = add_new_disk_start, 985 .add_new_disk_finish = add_new_disk_finish, 986 .new_disk_ack = new_disk_ack, 987 .remove_disk = remove_disk, 988 .gather_bitmaps = gather_bitmaps, 989 }; 990 991 static int __init cluster_init(void) 992 { 993 pr_warn("md-cluster: EXPERIMENTAL. Use with caution\n"); 994 pr_info("Registering Cluster MD functions\n"); 995 register_md_cluster_operations(&cluster_ops, THIS_MODULE); 996 return 0; 997 } 998 999 static void cluster_exit(void) 1000 { 1001 unregister_md_cluster_operations(); 1002 } 1003 1004 module_init(cluster_init); 1005 module_exit(cluster_exit); 1006 MODULE_LICENSE("GPL"); 1007 MODULE_DESCRIPTION("Clustering support for MD"); 1008