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