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 int mode; 32 }; 33 34 struct suspend_info { 35 int slot; 36 sector_t lo; 37 sector_t hi; 38 struct list_head list; 39 }; 40 41 struct resync_info { 42 __le64 lo; 43 __le64 hi; 44 }; 45 46 /* md_cluster_info flags */ 47 #define MD_CLUSTER_WAITING_FOR_NEWDISK 1 48 #define MD_CLUSTER_SUSPEND_READ_BALANCING 2 49 #define MD_CLUSTER_BEGIN_JOIN_CLUSTER 3 50 51 /* Lock the send communication. This is done through 52 * bit manipulation as opposed to a mutex in order to 53 * accomodate lock and hold. See next comment. 54 */ 55 #define MD_CLUSTER_SEND_LOCK 4 56 /* If cluster operations (such as adding a disk) must lock the 57 * communication channel, so as to perform extra operations 58 * (update metadata) and no other operation is allowed on the 59 * MD. Token needs to be locked and held until the operation 60 * completes witha md_update_sb(), which would eventually release 61 * the lock. 62 */ 63 #define MD_CLUSTER_SEND_LOCKED_ALREADY 5 64 /* We should receive message after node joined cluster and 65 * set up all the related infos such as bitmap and personality */ 66 #define MD_CLUSTER_ALREADY_IN_CLUSTER 6 67 #define MD_CLUSTER_PENDING_RECV_EVENT 7 68 69 70 struct md_cluster_info { 71 /* dlm lock space and resources for clustered raid. */ 72 dlm_lockspace_t *lockspace; 73 int slot_number; 74 struct completion completion; 75 struct mutex recv_mutex; 76 struct dlm_lock_resource *bitmap_lockres; 77 struct dlm_lock_resource **other_bitmap_lockres; 78 struct dlm_lock_resource *resync_lockres; 79 struct list_head suspend_list; 80 spinlock_t suspend_lock; 81 struct md_thread *recovery_thread; 82 unsigned long recovery_map; 83 /* communication loc resources */ 84 struct dlm_lock_resource *ack_lockres; 85 struct dlm_lock_resource *message_lockres; 86 struct dlm_lock_resource *token_lockres; 87 struct dlm_lock_resource *no_new_dev_lockres; 88 struct md_thread *recv_thread; 89 struct completion newdisk_completion; 90 wait_queue_head_t wait; 91 unsigned long state; 92 /* record the region in RESYNCING message */ 93 sector_t sync_low; 94 sector_t sync_hi; 95 }; 96 97 enum msg_type { 98 METADATA_UPDATED = 0, 99 RESYNCING, 100 NEWDISK, 101 REMOVE, 102 RE_ADD, 103 BITMAP_NEEDS_SYNC, 104 }; 105 106 struct cluster_msg { 107 __le32 type; 108 __le32 slot; 109 /* TODO: Unionize this for smaller footprint */ 110 __le64 low; 111 __le64 high; 112 char uuid[16]; 113 __le32 raid_slot; 114 }; 115 116 static void sync_ast(void *arg) 117 { 118 struct dlm_lock_resource *res; 119 120 res = arg; 121 complete(&res->completion); 122 } 123 124 static int dlm_lock_sync(struct dlm_lock_resource *res, int mode) 125 { 126 int ret = 0; 127 128 ret = dlm_lock(res->ls, mode, &res->lksb, 129 res->flags, res->name, strlen(res->name), 130 0, sync_ast, res, res->bast); 131 if (ret) 132 return ret; 133 wait_for_completion(&res->completion); 134 if (res->lksb.sb_status == 0) 135 res->mode = mode; 136 return res->lksb.sb_status; 137 } 138 139 static int dlm_unlock_sync(struct dlm_lock_resource *res) 140 { 141 return dlm_lock_sync(res, DLM_LOCK_NL); 142 } 143 144 static struct dlm_lock_resource *lockres_init(struct mddev *mddev, 145 char *name, void (*bastfn)(void *arg, int mode), int with_lvb) 146 { 147 struct dlm_lock_resource *res = NULL; 148 int ret, namelen; 149 struct md_cluster_info *cinfo = mddev->cluster_info; 150 151 res = kzalloc(sizeof(struct dlm_lock_resource), GFP_KERNEL); 152 if (!res) 153 return NULL; 154 init_completion(&res->completion); 155 res->ls = cinfo->lockspace; 156 res->mddev = mddev; 157 res->mode = DLM_LOCK_IV; 158 namelen = strlen(name); 159 res->name = kzalloc(namelen + 1, GFP_KERNEL); 160 if (!res->name) { 161 pr_err("md-cluster: Unable to allocate resource name for resource %s\n", name); 162 goto out_err; 163 } 164 strlcpy(res->name, name, namelen + 1); 165 if (with_lvb) { 166 res->lksb.sb_lvbptr = kzalloc(LVB_SIZE, GFP_KERNEL); 167 if (!res->lksb.sb_lvbptr) { 168 pr_err("md-cluster: Unable to allocate LVB for resource %s\n", name); 169 goto out_err; 170 } 171 res->flags = DLM_LKF_VALBLK; 172 } 173 174 if (bastfn) 175 res->bast = bastfn; 176 177 res->flags |= DLM_LKF_EXPEDITE; 178 179 ret = dlm_lock_sync(res, DLM_LOCK_NL); 180 if (ret) { 181 pr_err("md-cluster: Unable to lock NL on new lock resource %s\n", name); 182 goto out_err; 183 } 184 res->flags &= ~DLM_LKF_EXPEDITE; 185 res->flags |= DLM_LKF_CONVERT; 186 187 return res; 188 out_err: 189 kfree(res->lksb.sb_lvbptr); 190 kfree(res->name); 191 kfree(res); 192 return NULL; 193 } 194 195 static void lockres_free(struct dlm_lock_resource *res) 196 { 197 int ret = 0; 198 199 if (!res) 200 return; 201 202 /* 203 * use FORCEUNLOCK flag, so we can unlock even the lock is on the 204 * waiting or convert queue 205 */ 206 ret = dlm_unlock(res->ls, res->lksb.sb_lkid, DLM_LKF_FORCEUNLOCK, 207 &res->lksb, res); 208 if (unlikely(ret != 0)) 209 pr_err("failed to unlock %s return %d\n", res->name, ret); 210 else 211 wait_for_completion(&res->completion); 212 213 kfree(res->name); 214 kfree(res->lksb.sb_lvbptr); 215 kfree(res); 216 } 217 218 static void add_resync_info(struct dlm_lock_resource *lockres, 219 sector_t lo, sector_t hi) 220 { 221 struct resync_info *ri; 222 223 ri = (struct resync_info *)lockres->lksb.sb_lvbptr; 224 ri->lo = cpu_to_le64(lo); 225 ri->hi = cpu_to_le64(hi); 226 } 227 228 static struct suspend_info *read_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres) 229 { 230 struct resync_info ri; 231 struct suspend_info *s = NULL; 232 sector_t hi = 0; 233 234 dlm_lock_sync(lockres, DLM_LOCK_CR); 235 memcpy(&ri, lockres->lksb.sb_lvbptr, sizeof(struct resync_info)); 236 hi = le64_to_cpu(ri.hi); 237 if (hi > 0) { 238 s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL); 239 if (!s) 240 goto out; 241 s->hi = hi; 242 s->lo = le64_to_cpu(ri.lo); 243 } 244 dlm_unlock_sync(lockres); 245 out: 246 return s; 247 } 248 249 static void recover_bitmaps(struct md_thread *thread) 250 { 251 struct mddev *mddev = thread->mddev; 252 struct md_cluster_info *cinfo = mddev->cluster_info; 253 struct dlm_lock_resource *bm_lockres; 254 char str[64]; 255 int slot, ret; 256 struct suspend_info *s, *tmp; 257 sector_t lo, hi; 258 259 while (cinfo->recovery_map) { 260 slot = fls64((u64)cinfo->recovery_map) - 1; 261 262 /* Clear suspend_area associated with the bitmap */ 263 spin_lock_irq(&cinfo->suspend_lock); 264 list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list) 265 if (slot == s->slot) { 266 list_del(&s->list); 267 kfree(s); 268 } 269 spin_unlock_irq(&cinfo->suspend_lock); 270 271 snprintf(str, 64, "bitmap%04d", slot); 272 bm_lockres = lockres_init(mddev, str, NULL, 1); 273 if (!bm_lockres) { 274 pr_err("md-cluster: Cannot initialize bitmaps\n"); 275 goto clear_bit; 276 } 277 278 ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW); 279 if (ret) { 280 pr_err("md-cluster: Could not DLM lock %s: %d\n", 281 str, ret); 282 goto clear_bit; 283 } 284 ret = bitmap_copy_from_slot(mddev, slot, &lo, &hi, true); 285 if (ret) { 286 pr_err("md-cluster: Could not copy data from bitmap %d\n", slot); 287 goto dlm_unlock; 288 } 289 if (hi > 0) { 290 if (lo < mddev->recovery_cp) 291 mddev->recovery_cp = lo; 292 /* wake up thread to continue resync in case resync 293 * is not finished */ 294 if (mddev->recovery_cp != MaxSector) { 295 set_bit(MD_RECOVERY_NEEDED, &mddev->recovery); 296 md_wakeup_thread(mddev->thread); 297 } 298 } 299 dlm_unlock: 300 dlm_unlock_sync(bm_lockres); 301 clear_bit: 302 lockres_free(bm_lockres); 303 clear_bit(slot, &cinfo->recovery_map); 304 } 305 } 306 307 static void recover_prep(void *arg) 308 { 309 struct mddev *mddev = arg; 310 struct md_cluster_info *cinfo = mddev->cluster_info; 311 set_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state); 312 } 313 314 static void __recover_slot(struct mddev *mddev, int slot) 315 { 316 struct md_cluster_info *cinfo = mddev->cluster_info; 317 318 set_bit(slot, &cinfo->recovery_map); 319 if (!cinfo->recovery_thread) { 320 cinfo->recovery_thread = md_register_thread(recover_bitmaps, 321 mddev, "recover"); 322 if (!cinfo->recovery_thread) { 323 pr_warn("md-cluster: Could not create recovery thread\n"); 324 return; 325 } 326 } 327 md_wakeup_thread(cinfo->recovery_thread); 328 } 329 330 static void recover_slot(void *arg, struct dlm_slot *slot) 331 { 332 struct mddev *mddev = arg; 333 struct md_cluster_info *cinfo = mddev->cluster_info; 334 335 pr_info("md-cluster: %s Node %d/%d down. My slot: %d. Initiating recovery.\n", 336 mddev->bitmap_info.cluster_name, 337 slot->nodeid, slot->slot, 338 cinfo->slot_number); 339 /* deduct one since dlm slot starts from one while the num of 340 * cluster-md begins with 0 */ 341 __recover_slot(mddev, slot->slot - 1); 342 } 343 344 static void recover_done(void *arg, struct dlm_slot *slots, 345 int num_slots, int our_slot, 346 uint32_t generation) 347 { 348 struct mddev *mddev = arg; 349 struct md_cluster_info *cinfo = mddev->cluster_info; 350 351 cinfo->slot_number = our_slot; 352 /* completion is only need to be complete when node join cluster, 353 * it doesn't need to run during another node's failure */ 354 if (test_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state)) { 355 complete(&cinfo->completion); 356 clear_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state); 357 } 358 clear_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state); 359 } 360 361 /* the ops is called when node join the cluster, and do lock recovery 362 * if node failure occurs */ 363 static const struct dlm_lockspace_ops md_ls_ops = { 364 .recover_prep = recover_prep, 365 .recover_slot = recover_slot, 366 .recover_done = recover_done, 367 }; 368 369 /* 370 * The BAST function for the ack lock resource 371 * This function wakes up the receive thread in 372 * order to receive and process the message. 373 */ 374 static void ack_bast(void *arg, int mode) 375 { 376 struct dlm_lock_resource *res = arg; 377 struct md_cluster_info *cinfo = res->mddev->cluster_info; 378 379 if (mode == DLM_LOCK_EX) { 380 if (test_bit(MD_CLUSTER_ALREADY_IN_CLUSTER, &cinfo->state)) 381 md_wakeup_thread(cinfo->recv_thread); 382 else 383 set_bit(MD_CLUSTER_PENDING_RECV_EVENT, &cinfo->state); 384 } 385 } 386 387 static void __remove_suspend_info(struct md_cluster_info *cinfo, int slot) 388 { 389 struct suspend_info *s, *tmp; 390 391 list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list) 392 if (slot == s->slot) { 393 list_del(&s->list); 394 kfree(s); 395 break; 396 } 397 } 398 399 static void remove_suspend_info(struct mddev *mddev, int slot) 400 { 401 struct md_cluster_info *cinfo = mddev->cluster_info; 402 spin_lock_irq(&cinfo->suspend_lock); 403 __remove_suspend_info(cinfo, slot); 404 spin_unlock_irq(&cinfo->suspend_lock); 405 mddev->pers->quiesce(mddev, 2); 406 } 407 408 409 static void process_suspend_info(struct mddev *mddev, 410 int slot, sector_t lo, sector_t hi) 411 { 412 struct md_cluster_info *cinfo = mddev->cluster_info; 413 struct suspend_info *s; 414 415 if (!hi) { 416 remove_suspend_info(mddev, slot); 417 set_bit(MD_RECOVERY_NEEDED, &mddev->recovery); 418 md_wakeup_thread(mddev->thread); 419 return; 420 } 421 422 /* 423 * The bitmaps are not same for different nodes 424 * if RESYNCING is happening in one node, then 425 * the node which received the RESYNCING message 426 * probably will perform resync with the region 427 * [lo, hi] again, so we could reduce resync time 428 * a lot if we can ensure that the bitmaps among 429 * different nodes are match up well. 430 * 431 * sync_low/hi is used to record the region which 432 * arrived in the previous RESYNCING message, 433 * 434 * Call bitmap_sync_with_cluster to clear 435 * NEEDED_MASK and set RESYNC_MASK since 436 * resync thread is running in another node, 437 * so we don't need to do the resync again 438 * with the same section */ 439 bitmap_sync_with_cluster(mddev, cinfo->sync_low, 440 cinfo->sync_hi, 441 lo, hi); 442 cinfo->sync_low = lo; 443 cinfo->sync_hi = hi; 444 445 s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL); 446 if (!s) 447 return; 448 s->slot = slot; 449 s->lo = lo; 450 s->hi = hi; 451 mddev->pers->quiesce(mddev, 1); 452 mddev->pers->quiesce(mddev, 0); 453 spin_lock_irq(&cinfo->suspend_lock); 454 /* Remove existing entry (if exists) before adding */ 455 __remove_suspend_info(cinfo, slot); 456 list_add(&s->list, &cinfo->suspend_list); 457 spin_unlock_irq(&cinfo->suspend_lock); 458 mddev->pers->quiesce(mddev, 2); 459 } 460 461 static void process_add_new_disk(struct mddev *mddev, struct cluster_msg *cmsg) 462 { 463 char disk_uuid[64]; 464 struct md_cluster_info *cinfo = mddev->cluster_info; 465 char event_name[] = "EVENT=ADD_DEVICE"; 466 char raid_slot[16]; 467 char *envp[] = {event_name, disk_uuid, raid_slot, NULL}; 468 int len; 469 470 len = snprintf(disk_uuid, 64, "DEVICE_UUID="); 471 sprintf(disk_uuid + len, "%pU", cmsg->uuid); 472 snprintf(raid_slot, 16, "RAID_DISK=%d", le32_to_cpu(cmsg->raid_slot)); 473 pr_info("%s:%d Sending kobject change with %s and %s\n", __func__, __LINE__, disk_uuid, raid_slot); 474 init_completion(&cinfo->newdisk_completion); 475 set_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state); 476 kobject_uevent_env(&disk_to_dev(mddev->gendisk)->kobj, KOBJ_CHANGE, envp); 477 wait_for_completion_timeout(&cinfo->newdisk_completion, 478 NEW_DEV_TIMEOUT); 479 clear_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state); 480 } 481 482 483 static void process_metadata_update(struct mddev *mddev, struct cluster_msg *msg) 484 { 485 struct md_cluster_info *cinfo = mddev->cluster_info; 486 mddev->good_device_nr = le32_to_cpu(msg->raid_slot); 487 set_bit(MD_RELOAD_SB, &mddev->flags); 488 dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR); 489 md_wakeup_thread(mddev->thread); 490 } 491 492 static void process_remove_disk(struct mddev *mddev, struct cluster_msg *msg) 493 { 494 struct md_rdev *rdev = md_find_rdev_nr_rcu(mddev, 495 le32_to_cpu(msg->raid_slot)); 496 497 if (rdev) { 498 set_bit(ClusterRemove, &rdev->flags); 499 set_bit(MD_RECOVERY_NEEDED, &mddev->recovery); 500 md_wakeup_thread(mddev->thread); 501 } 502 else 503 pr_warn("%s: %d Could not find disk(%d) to REMOVE\n", 504 __func__, __LINE__, le32_to_cpu(msg->raid_slot)); 505 } 506 507 static void process_readd_disk(struct mddev *mddev, struct cluster_msg *msg) 508 { 509 struct md_rdev *rdev = md_find_rdev_nr_rcu(mddev, 510 le32_to_cpu(msg->raid_slot)); 511 512 if (rdev && test_bit(Faulty, &rdev->flags)) 513 clear_bit(Faulty, &rdev->flags); 514 else 515 pr_warn("%s: %d Could not find disk(%d) which is faulty", 516 __func__, __LINE__, le32_to_cpu(msg->raid_slot)); 517 } 518 519 static int process_recvd_msg(struct mddev *mddev, struct cluster_msg *msg) 520 { 521 int ret = 0; 522 523 if (WARN(mddev->cluster_info->slot_number - 1 == le32_to_cpu(msg->slot), 524 "node %d received it's own msg\n", le32_to_cpu(msg->slot))) 525 return -1; 526 switch (le32_to_cpu(msg->type)) { 527 case METADATA_UPDATED: 528 process_metadata_update(mddev, msg); 529 break; 530 case RESYNCING: 531 process_suspend_info(mddev, le32_to_cpu(msg->slot), 532 le64_to_cpu(msg->low), 533 le64_to_cpu(msg->high)); 534 break; 535 case NEWDISK: 536 process_add_new_disk(mddev, msg); 537 break; 538 case REMOVE: 539 process_remove_disk(mddev, msg); 540 break; 541 case RE_ADD: 542 process_readd_disk(mddev, msg); 543 break; 544 case BITMAP_NEEDS_SYNC: 545 __recover_slot(mddev, le32_to_cpu(msg->slot)); 546 break; 547 default: 548 ret = -1; 549 pr_warn("%s:%d Received unknown message from %d\n", 550 __func__, __LINE__, msg->slot); 551 } 552 return ret; 553 } 554 555 /* 556 * thread for receiving message 557 */ 558 static void recv_daemon(struct md_thread *thread) 559 { 560 struct md_cluster_info *cinfo = thread->mddev->cluster_info; 561 struct dlm_lock_resource *ack_lockres = cinfo->ack_lockres; 562 struct dlm_lock_resource *message_lockres = cinfo->message_lockres; 563 struct cluster_msg msg; 564 int ret; 565 566 mutex_lock(&cinfo->recv_mutex); 567 /*get CR on Message*/ 568 if (dlm_lock_sync(message_lockres, DLM_LOCK_CR)) { 569 pr_err("md/raid1:failed to get CR on MESSAGE\n"); 570 mutex_unlock(&cinfo->recv_mutex); 571 return; 572 } 573 574 /* read lvb and wake up thread to process this message_lockres */ 575 memcpy(&msg, message_lockres->lksb.sb_lvbptr, sizeof(struct cluster_msg)); 576 ret = process_recvd_msg(thread->mddev, &msg); 577 if (ret) 578 goto out; 579 580 /*release CR on ack_lockres*/ 581 ret = dlm_unlock_sync(ack_lockres); 582 if (unlikely(ret != 0)) 583 pr_info("unlock ack failed return %d\n", ret); 584 /*up-convert to PR on message_lockres*/ 585 ret = dlm_lock_sync(message_lockres, DLM_LOCK_PR); 586 if (unlikely(ret != 0)) 587 pr_info("lock PR on msg failed return %d\n", ret); 588 /*get CR on ack_lockres again*/ 589 ret = dlm_lock_sync(ack_lockres, DLM_LOCK_CR); 590 if (unlikely(ret != 0)) 591 pr_info("lock CR on ack failed return %d\n", ret); 592 out: 593 /*release CR on message_lockres*/ 594 ret = dlm_unlock_sync(message_lockres); 595 if (unlikely(ret != 0)) 596 pr_info("unlock msg failed return %d\n", ret); 597 mutex_unlock(&cinfo->recv_mutex); 598 } 599 600 /* lock_token() 601 * Takes the lock on the TOKEN lock resource so no other 602 * node can communicate while the operation is underway. 603 */ 604 static int lock_token(struct md_cluster_info *cinfo) 605 { 606 int error; 607 608 error = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX); 609 if (error) 610 pr_err("md-cluster(%s:%d): failed to get EX on TOKEN (%d)\n", 611 __func__, __LINE__, error); 612 613 /* Lock the receive sequence */ 614 mutex_lock(&cinfo->recv_mutex); 615 return error; 616 } 617 618 /* lock_comm() 619 * Sets the MD_CLUSTER_SEND_LOCK bit to lock the send channel. 620 */ 621 static int lock_comm(struct md_cluster_info *cinfo) 622 { 623 wait_event(cinfo->wait, 624 !test_and_set_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state)); 625 626 return lock_token(cinfo); 627 } 628 629 static void unlock_comm(struct md_cluster_info *cinfo) 630 { 631 WARN_ON(cinfo->token_lockres->mode != DLM_LOCK_EX); 632 mutex_unlock(&cinfo->recv_mutex); 633 dlm_unlock_sync(cinfo->token_lockres); 634 clear_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state); 635 wake_up(&cinfo->wait); 636 } 637 638 /* __sendmsg() 639 * This function performs the actual sending of the message. This function is 640 * usually called after performing the encompassing operation 641 * The function: 642 * 1. Grabs the message lockresource in EX mode 643 * 2. Copies the message to the message LVB 644 * 3. Downconverts message lockresource to CW 645 * 4. Upconverts ack lock resource from CR to EX. This forces the BAST on other nodes 646 * and the other nodes read the message. The thread will wait here until all other 647 * nodes have released ack lock resource. 648 * 5. Downconvert ack lockresource to CR 649 */ 650 static int __sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg) 651 { 652 int error; 653 int slot = cinfo->slot_number - 1; 654 655 cmsg->slot = cpu_to_le32(slot); 656 /*get EX on Message*/ 657 error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_EX); 658 if (error) { 659 pr_err("md-cluster: failed to get EX on MESSAGE (%d)\n", error); 660 goto failed_message; 661 } 662 663 memcpy(cinfo->message_lockres->lksb.sb_lvbptr, (void *)cmsg, 664 sizeof(struct cluster_msg)); 665 /*down-convert EX to CW on Message*/ 666 error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_CW); 667 if (error) { 668 pr_err("md-cluster: failed to convert EX to CW on MESSAGE(%d)\n", 669 error); 670 goto failed_ack; 671 } 672 673 /*up-convert CR to EX on Ack*/ 674 error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_EX); 675 if (error) { 676 pr_err("md-cluster: failed to convert CR to EX on ACK(%d)\n", 677 error); 678 goto failed_ack; 679 } 680 681 /*down-convert EX to CR on Ack*/ 682 error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR); 683 if (error) { 684 pr_err("md-cluster: failed to convert EX to CR on ACK(%d)\n", 685 error); 686 goto failed_ack; 687 } 688 689 failed_ack: 690 error = dlm_unlock_sync(cinfo->message_lockres); 691 if (unlikely(error != 0)) { 692 pr_err("md-cluster: failed convert to NL on MESSAGE(%d)\n", 693 error); 694 /* in case the message can't be released due to some reason */ 695 goto failed_ack; 696 } 697 failed_message: 698 return error; 699 } 700 701 static int sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg) 702 { 703 int ret; 704 705 lock_comm(cinfo); 706 ret = __sendmsg(cinfo, cmsg); 707 unlock_comm(cinfo); 708 return ret; 709 } 710 711 static int gather_all_resync_info(struct mddev *mddev, int total_slots) 712 { 713 struct md_cluster_info *cinfo = mddev->cluster_info; 714 int i, ret = 0; 715 struct dlm_lock_resource *bm_lockres; 716 struct suspend_info *s; 717 char str[64]; 718 sector_t lo, hi; 719 720 721 for (i = 0; i < total_slots; i++) { 722 memset(str, '\0', 64); 723 snprintf(str, 64, "bitmap%04d", i); 724 bm_lockres = lockres_init(mddev, str, NULL, 1); 725 if (!bm_lockres) 726 return -ENOMEM; 727 if (i == (cinfo->slot_number - 1)) { 728 lockres_free(bm_lockres); 729 continue; 730 } 731 732 bm_lockres->flags |= DLM_LKF_NOQUEUE; 733 ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW); 734 if (ret == -EAGAIN) { 735 memset(bm_lockres->lksb.sb_lvbptr, '\0', LVB_SIZE); 736 s = read_resync_info(mddev, bm_lockres); 737 if (s) { 738 pr_info("%s:%d Resync[%llu..%llu] in progress on %d\n", 739 __func__, __LINE__, 740 (unsigned long long) s->lo, 741 (unsigned long long) s->hi, i); 742 spin_lock_irq(&cinfo->suspend_lock); 743 s->slot = i; 744 list_add(&s->list, &cinfo->suspend_list); 745 spin_unlock_irq(&cinfo->suspend_lock); 746 } 747 ret = 0; 748 lockres_free(bm_lockres); 749 continue; 750 } 751 if (ret) { 752 lockres_free(bm_lockres); 753 goto out; 754 } 755 756 /* Read the disk bitmap sb and check if it needs recovery */ 757 ret = bitmap_copy_from_slot(mddev, i, &lo, &hi, false); 758 if (ret) { 759 pr_warn("md-cluster: Could not gather bitmaps from slot %d", i); 760 lockres_free(bm_lockres); 761 continue; 762 } 763 if ((hi > 0) && (lo < mddev->recovery_cp)) { 764 set_bit(MD_RECOVERY_NEEDED, &mddev->recovery); 765 mddev->recovery_cp = lo; 766 md_check_recovery(mddev); 767 } 768 769 dlm_unlock_sync(bm_lockres); 770 lockres_free(bm_lockres); 771 } 772 out: 773 return ret; 774 } 775 776 static int join(struct mddev *mddev, int nodes) 777 { 778 struct md_cluster_info *cinfo; 779 int ret, ops_rv; 780 char str[64]; 781 782 cinfo = kzalloc(sizeof(struct md_cluster_info), GFP_KERNEL); 783 if (!cinfo) 784 return -ENOMEM; 785 786 INIT_LIST_HEAD(&cinfo->suspend_list); 787 spin_lock_init(&cinfo->suspend_lock); 788 init_completion(&cinfo->completion); 789 set_bit(MD_CLUSTER_BEGIN_JOIN_CLUSTER, &cinfo->state); 790 init_waitqueue_head(&cinfo->wait); 791 mutex_init(&cinfo->recv_mutex); 792 793 mddev->cluster_info = cinfo; 794 795 memset(str, 0, 64); 796 sprintf(str, "%pU", mddev->uuid); 797 ret = dlm_new_lockspace(str, mddev->bitmap_info.cluster_name, 798 DLM_LSFL_FS, LVB_SIZE, 799 &md_ls_ops, mddev, &ops_rv, &cinfo->lockspace); 800 if (ret) 801 goto err; 802 wait_for_completion(&cinfo->completion); 803 if (nodes < cinfo->slot_number) { 804 pr_err("md-cluster: Slot allotted(%d) is greater than available slots(%d).", 805 cinfo->slot_number, nodes); 806 ret = -ERANGE; 807 goto err; 808 } 809 /* Initiate the communication resources */ 810 ret = -ENOMEM; 811 cinfo->recv_thread = md_register_thread(recv_daemon, mddev, "cluster_recv"); 812 if (!cinfo->recv_thread) { 813 pr_err("md-cluster: cannot allocate memory for recv_thread!\n"); 814 goto err; 815 } 816 cinfo->message_lockres = lockres_init(mddev, "message", NULL, 1); 817 if (!cinfo->message_lockres) 818 goto err; 819 cinfo->token_lockres = lockres_init(mddev, "token", NULL, 0); 820 if (!cinfo->token_lockres) 821 goto err; 822 cinfo->no_new_dev_lockres = lockres_init(mddev, "no-new-dev", NULL, 0); 823 if (!cinfo->no_new_dev_lockres) 824 goto err; 825 826 ret = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX); 827 if (ret) { 828 ret = -EAGAIN; 829 pr_err("md-cluster: can't join cluster to avoid lock issue\n"); 830 goto err; 831 } 832 cinfo->ack_lockres = lockres_init(mddev, "ack", ack_bast, 0); 833 if (!cinfo->ack_lockres) { 834 ret = -ENOMEM; 835 goto err; 836 } 837 /* get sync CR lock on ACK. */ 838 if (dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR)) 839 pr_err("md-cluster: failed to get a sync CR lock on ACK!(%d)\n", 840 ret); 841 dlm_unlock_sync(cinfo->token_lockres); 842 /* get sync CR lock on no-new-dev. */ 843 if (dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR)) 844 pr_err("md-cluster: failed to get a sync CR lock on no-new-dev!(%d)\n", ret); 845 846 847 pr_info("md-cluster: Joined cluster %s slot %d\n", str, cinfo->slot_number); 848 snprintf(str, 64, "bitmap%04d", cinfo->slot_number - 1); 849 cinfo->bitmap_lockres = lockres_init(mddev, str, NULL, 1); 850 if (!cinfo->bitmap_lockres) { 851 ret = -ENOMEM; 852 goto err; 853 } 854 if (dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW)) { 855 pr_err("Failed to get bitmap lock\n"); 856 ret = -EINVAL; 857 goto err; 858 } 859 860 cinfo->resync_lockres = lockres_init(mddev, "resync", NULL, 0); 861 if (!cinfo->resync_lockres) { 862 ret = -ENOMEM; 863 goto err; 864 } 865 866 return 0; 867 err: 868 md_unregister_thread(&cinfo->recovery_thread); 869 md_unregister_thread(&cinfo->recv_thread); 870 lockres_free(cinfo->message_lockres); 871 lockres_free(cinfo->token_lockres); 872 lockres_free(cinfo->ack_lockres); 873 lockres_free(cinfo->no_new_dev_lockres); 874 lockres_free(cinfo->resync_lockres); 875 lockres_free(cinfo->bitmap_lockres); 876 if (cinfo->lockspace) 877 dlm_release_lockspace(cinfo->lockspace, 2); 878 mddev->cluster_info = NULL; 879 kfree(cinfo); 880 return ret; 881 } 882 883 static void load_bitmaps(struct mddev *mddev, int total_slots) 884 { 885 struct md_cluster_info *cinfo = mddev->cluster_info; 886 887 /* load all the node's bitmap info for resync */ 888 if (gather_all_resync_info(mddev, total_slots)) 889 pr_err("md-cluster: failed to gather all resyn infos\n"); 890 set_bit(MD_CLUSTER_ALREADY_IN_CLUSTER, &cinfo->state); 891 /* wake up recv thread in case something need to be handled */ 892 if (test_and_clear_bit(MD_CLUSTER_PENDING_RECV_EVENT, &cinfo->state)) 893 md_wakeup_thread(cinfo->recv_thread); 894 } 895 896 static void resync_bitmap(struct mddev *mddev) 897 { 898 struct md_cluster_info *cinfo = mddev->cluster_info; 899 struct cluster_msg cmsg = {0}; 900 int err; 901 902 cmsg.type = cpu_to_le32(BITMAP_NEEDS_SYNC); 903 err = sendmsg(cinfo, &cmsg); 904 if (err) 905 pr_err("%s:%d: failed to send BITMAP_NEEDS_SYNC message (%d)\n", 906 __func__, __LINE__, err); 907 } 908 909 static void unlock_all_bitmaps(struct mddev *mddev); 910 static int leave(struct mddev *mddev) 911 { 912 struct md_cluster_info *cinfo = mddev->cluster_info; 913 914 if (!cinfo) 915 return 0; 916 917 /* BITMAP_NEEDS_SYNC message should be sent when node 918 * is leaving the cluster with dirty bitmap, also we 919 * can only deliver it when dlm connection is available */ 920 if (cinfo->slot_number > 0 && mddev->recovery_cp != MaxSector) 921 resync_bitmap(mddev); 922 923 md_unregister_thread(&cinfo->recovery_thread); 924 md_unregister_thread(&cinfo->recv_thread); 925 lockres_free(cinfo->message_lockres); 926 lockres_free(cinfo->token_lockres); 927 lockres_free(cinfo->ack_lockres); 928 lockres_free(cinfo->no_new_dev_lockres); 929 lockres_free(cinfo->resync_lockres); 930 lockres_free(cinfo->bitmap_lockres); 931 unlock_all_bitmaps(mddev); 932 dlm_release_lockspace(cinfo->lockspace, 2); 933 return 0; 934 } 935 936 /* slot_number(): Returns the MD slot number to use 937 * DLM starts the slot numbers from 1, wheras cluster-md 938 * wants the number to be from zero, so we deduct one 939 */ 940 static int slot_number(struct mddev *mddev) 941 { 942 struct md_cluster_info *cinfo = mddev->cluster_info; 943 944 return cinfo->slot_number - 1; 945 } 946 947 /* 948 * Check if the communication is already locked, else lock the communication 949 * channel. 950 * If it is already locked, token is in EX mode, and hence lock_token() 951 * should not be called. 952 */ 953 static int metadata_update_start(struct mddev *mddev) 954 { 955 struct md_cluster_info *cinfo = mddev->cluster_info; 956 957 wait_event(cinfo->wait, 958 !test_and_set_bit(MD_CLUSTER_SEND_LOCK, &cinfo->state) || 959 test_and_clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state)); 960 961 /* If token is already locked, return 0 */ 962 if (cinfo->token_lockres->mode == DLM_LOCK_EX) 963 return 0; 964 965 return lock_token(cinfo); 966 } 967 968 static int metadata_update_finish(struct mddev *mddev) 969 { 970 struct md_cluster_info *cinfo = mddev->cluster_info; 971 struct cluster_msg cmsg; 972 struct md_rdev *rdev; 973 int ret = 0; 974 int raid_slot = -1; 975 976 memset(&cmsg, 0, sizeof(cmsg)); 977 cmsg.type = cpu_to_le32(METADATA_UPDATED); 978 /* Pick up a good active device number to send. 979 */ 980 rdev_for_each(rdev, mddev) 981 if (rdev->raid_disk > -1 && !test_bit(Faulty, &rdev->flags)) { 982 raid_slot = rdev->desc_nr; 983 break; 984 } 985 if (raid_slot >= 0) { 986 cmsg.raid_slot = cpu_to_le32(raid_slot); 987 ret = __sendmsg(cinfo, &cmsg); 988 } else 989 pr_warn("md-cluster: No good device id found to send\n"); 990 clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state); 991 unlock_comm(cinfo); 992 return ret; 993 } 994 995 static void metadata_update_cancel(struct mddev *mddev) 996 { 997 struct md_cluster_info *cinfo = mddev->cluster_info; 998 clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state); 999 unlock_comm(cinfo); 1000 } 1001 1002 static int resync_start(struct mddev *mddev) 1003 { 1004 struct md_cluster_info *cinfo = mddev->cluster_info; 1005 return dlm_lock_sync(cinfo->resync_lockres, DLM_LOCK_EX); 1006 } 1007 1008 static int resync_info_update(struct mddev *mddev, sector_t lo, sector_t hi) 1009 { 1010 struct md_cluster_info *cinfo = mddev->cluster_info; 1011 struct resync_info ri; 1012 struct cluster_msg cmsg = {0}; 1013 1014 /* do not send zero again, if we have sent before */ 1015 if (hi == 0) { 1016 memcpy(&ri, cinfo->bitmap_lockres->lksb.sb_lvbptr, sizeof(struct resync_info)); 1017 if (le64_to_cpu(ri.hi) == 0) 1018 return 0; 1019 } 1020 1021 add_resync_info(cinfo->bitmap_lockres, lo, hi); 1022 /* Re-acquire the lock to refresh LVB */ 1023 dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW); 1024 cmsg.type = cpu_to_le32(RESYNCING); 1025 cmsg.low = cpu_to_le64(lo); 1026 cmsg.high = cpu_to_le64(hi); 1027 1028 return sendmsg(cinfo, &cmsg); 1029 } 1030 1031 static int resync_finish(struct mddev *mddev) 1032 { 1033 struct md_cluster_info *cinfo = mddev->cluster_info; 1034 dlm_unlock_sync(cinfo->resync_lockres); 1035 return resync_info_update(mddev, 0, 0); 1036 } 1037 1038 static int area_resyncing(struct mddev *mddev, int direction, 1039 sector_t lo, sector_t hi) 1040 { 1041 struct md_cluster_info *cinfo = mddev->cluster_info; 1042 int ret = 0; 1043 struct suspend_info *s; 1044 1045 if ((direction == READ) && 1046 test_bit(MD_CLUSTER_SUSPEND_READ_BALANCING, &cinfo->state)) 1047 return 1; 1048 1049 spin_lock_irq(&cinfo->suspend_lock); 1050 if (list_empty(&cinfo->suspend_list)) 1051 goto out; 1052 list_for_each_entry(s, &cinfo->suspend_list, list) 1053 if (hi > s->lo && lo < s->hi) { 1054 ret = 1; 1055 break; 1056 } 1057 out: 1058 spin_unlock_irq(&cinfo->suspend_lock); 1059 return ret; 1060 } 1061 1062 /* add_new_disk() - initiates a disk add 1063 * However, if this fails before writing md_update_sb(), 1064 * add_new_disk_cancel() must be called to release token lock 1065 */ 1066 static int add_new_disk(struct mddev *mddev, struct md_rdev *rdev) 1067 { 1068 struct md_cluster_info *cinfo = mddev->cluster_info; 1069 struct cluster_msg cmsg; 1070 int ret = 0; 1071 struct mdp_superblock_1 *sb = page_address(rdev->sb_page); 1072 char *uuid = sb->device_uuid; 1073 1074 memset(&cmsg, 0, sizeof(cmsg)); 1075 cmsg.type = cpu_to_le32(NEWDISK); 1076 memcpy(cmsg.uuid, uuid, 16); 1077 cmsg.raid_slot = cpu_to_le32(rdev->desc_nr); 1078 lock_comm(cinfo); 1079 ret = __sendmsg(cinfo, &cmsg); 1080 if (ret) 1081 return ret; 1082 cinfo->no_new_dev_lockres->flags |= DLM_LKF_NOQUEUE; 1083 ret = dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_EX); 1084 cinfo->no_new_dev_lockres->flags &= ~DLM_LKF_NOQUEUE; 1085 /* Some node does not "see" the device */ 1086 if (ret == -EAGAIN) 1087 ret = -ENOENT; 1088 if (ret) 1089 unlock_comm(cinfo); 1090 else { 1091 dlm_lock_sync(cinfo->no_new_dev_lockres, DLM_LOCK_CR); 1092 /* Since MD_CHANGE_DEVS will be set in add_bound_rdev which 1093 * will run soon after add_new_disk, the below path will be 1094 * invoked: 1095 * md_wakeup_thread(mddev->thread) 1096 * -> conf->thread (raid1d) 1097 * -> md_check_recovery -> md_update_sb 1098 * -> metadata_update_start/finish 1099 * MD_CLUSTER_SEND_LOCKED_ALREADY will be cleared eventually. 1100 * 1101 * For other failure cases, metadata_update_cancel and 1102 * add_new_disk_cancel also clear below bit as well. 1103 * */ 1104 set_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state); 1105 wake_up(&cinfo->wait); 1106 } 1107 return ret; 1108 } 1109 1110 static void add_new_disk_cancel(struct mddev *mddev) 1111 { 1112 struct md_cluster_info *cinfo = mddev->cluster_info; 1113 clear_bit(MD_CLUSTER_SEND_LOCKED_ALREADY, &cinfo->state); 1114 unlock_comm(cinfo); 1115 } 1116 1117 static int new_disk_ack(struct mddev *mddev, bool ack) 1118 { 1119 struct md_cluster_info *cinfo = mddev->cluster_info; 1120 1121 if (!test_bit(MD_CLUSTER_WAITING_FOR_NEWDISK, &cinfo->state)) { 1122 pr_warn("md-cluster(%s): Spurious cluster confirmation\n", mdname(mddev)); 1123 return -EINVAL; 1124 } 1125 1126 if (ack) 1127 dlm_unlock_sync(cinfo->no_new_dev_lockres); 1128 complete(&cinfo->newdisk_completion); 1129 return 0; 1130 } 1131 1132 static int remove_disk(struct mddev *mddev, struct md_rdev *rdev) 1133 { 1134 struct cluster_msg cmsg = {0}; 1135 struct md_cluster_info *cinfo = mddev->cluster_info; 1136 cmsg.type = cpu_to_le32(REMOVE); 1137 cmsg.raid_slot = cpu_to_le32(rdev->desc_nr); 1138 return sendmsg(cinfo, &cmsg); 1139 } 1140 1141 static int lock_all_bitmaps(struct mddev *mddev) 1142 { 1143 int slot, my_slot, ret, held = 1, i = 0; 1144 char str[64]; 1145 struct md_cluster_info *cinfo = mddev->cluster_info; 1146 1147 cinfo->other_bitmap_lockres = kzalloc((mddev->bitmap_info.nodes - 1) * 1148 sizeof(struct dlm_lock_resource *), 1149 GFP_KERNEL); 1150 if (!cinfo->other_bitmap_lockres) { 1151 pr_err("md: can't alloc mem for other bitmap locks\n"); 1152 return 0; 1153 } 1154 1155 my_slot = slot_number(mddev); 1156 for (slot = 0; slot < mddev->bitmap_info.nodes; slot++) { 1157 if (slot == my_slot) 1158 continue; 1159 1160 memset(str, '\0', 64); 1161 snprintf(str, 64, "bitmap%04d", slot); 1162 cinfo->other_bitmap_lockres[i] = lockres_init(mddev, str, NULL, 1); 1163 if (!cinfo->other_bitmap_lockres[i]) 1164 return -ENOMEM; 1165 1166 cinfo->other_bitmap_lockres[i]->flags |= DLM_LKF_NOQUEUE; 1167 ret = dlm_lock_sync(cinfo->other_bitmap_lockres[i], DLM_LOCK_PW); 1168 if (ret) 1169 held = -1; 1170 i++; 1171 } 1172 1173 return held; 1174 } 1175 1176 static void unlock_all_bitmaps(struct mddev *mddev) 1177 { 1178 struct md_cluster_info *cinfo = mddev->cluster_info; 1179 int i; 1180 1181 /* release other node's bitmap lock if they are existed */ 1182 if (cinfo->other_bitmap_lockres) { 1183 for (i = 0; i < mddev->bitmap_info.nodes - 1; i++) { 1184 if (cinfo->other_bitmap_lockres[i]) { 1185 dlm_unlock_sync(cinfo->other_bitmap_lockres[i]); 1186 lockres_free(cinfo->other_bitmap_lockres[i]); 1187 } 1188 } 1189 kfree(cinfo->other_bitmap_lockres); 1190 } 1191 } 1192 1193 static int gather_bitmaps(struct md_rdev *rdev) 1194 { 1195 int sn, err; 1196 sector_t lo, hi; 1197 struct cluster_msg cmsg = {0}; 1198 struct mddev *mddev = rdev->mddev; 1199 struct md_cluster_info *cinfo = mddev->cluster_info; 1200 1201 cmsg.type = cpu_to_le32(RE_ADD); 1202 cmsg.raid_slot = cpu_to_le32(rdev->desc_nr); 1203 err = sendmsg(cinfo, &cmsg); 1204 if (err) 1205 goto out; 1206 1207 for (sn = 0; sn < mddev->bitmap_info.nodes; sn++) { 1208 if (sn == (cinfo->slot_number - 1)) 1209 continue; 1210 err = bitmap_copy_from_slot(mddev, sn, &lo, &hi, false); 1211 if (err) { 1212 pr_warn("md-cluster: Could not gather bitmaps from slot %d", sn); 1213 goto out; 1214 } 1215 if ((hi > 0) && (lo < mddev->recovery_cp)) 1216 mddev->recovery_cp = lo; 1217 } 1218 out: 1219 return err; 1220 } 1221 1222 static struct md_cluster_operations cluster_ops = { 1223 .join = join, 1224 .leave = leave, 1225 .slot_number = slot_number, 1226 .resync_start = resync_start, 1227 .resync_finish = resync_finish, 1228 .resync_info_update = resync_info_update, 1229 .metadata_update_start = metadata_update_start, 1230 .metadata_update_finish = metadata_update_finish, 1231 .metadata_update_cancel = metadata_update_cancel, 1232 .area_resyncing = area_resyncing, 1233 .add_new_disk = add_new_disk, 1234 .add_new_disk_cancel = add_new_disk_cancel, 1235 .new_disk_ack = new_disk_ack, 1236 .remove_disk = remove_disk, 1237 .load_bitmaps = load_bitmaps, 1238 .gather_bitmaps = gather_bitmaps, 1239 .lock_all_bitmaps = lock_all_bitmaps, 1240 .unlock_all_bitmaps = unlock_all_bitmaps, 1241 }; 1242 1243 static int __init cluster_init(void) 1244 { 1245 pr_warn("md-cluster: EXPERIMENTAL. Use with caution\n"); 1246 pr_info("Registering Cluster MD functions\n"); 1247 register_md_cluster_operations(&cluster_ops, THIS_MODULE); 1248 return 0; 1249 } 1250 1251 static void cluster_exit(void) 1252 { 1253 unregister_md_cluster_operations(); 1254 } 1255 1256 module_init(cluster_init); 1257 module_exit(cluster_exit); 1258 MODULE_AUTHOR("SUSE"); 1259 MODULE_LICENSE("GPL"); 1260 MODULE_DESCRIPTION("Clustering support for MD"); 1261