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