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