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