drivers/md/md-cluster.c

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