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 "md.h"
  16 #include "bitmap.h"
  17 #include "md-cluster.h"
  18
  19 #define LVB_SIZE        64
  20
  21 struct dlm_lock_resource {
  22         dlm_lockspace_t *ls;
  23         struct dlm_lksb lksb;
  24         char *name; /* lock name. */
  25         uint32_t flags; /* flags to pass to dlm_lock() */
  26         struct completion completion; /* completion for synchronized locking */
  27         void (*bast)(void *arg, int mode); /* blocking AST function pointer*/
  28         struct mddev *mddev; /* pointing back to mddev. */
  29 };
  30
  31 struct suspend_info {
  32         int slot;
  33         sector_t lo;
  34         sector_t hi;
  35         struct list_head list;
  36 };
  37
  38 struct resync_info {
  39         __le64 lo;
  40         __le64 hi;
  41 };
  42
  43 struct md_cluster_info {
  44         /* dlm lock space and resources for clustered raid. */
  45         dlm_lockspace_t *lockspace;
  46         int slot_number;
  47         struct completion completion;
  48         struct dlm_lock_resource *sb_lock;
  49         struct mutex sb_mutex;
  50         struct dlm_lock_resource *bitmap_lockres;
  51         struct list_head suspend_list;
  52         spinlock_t suspend_lock;
  53         struct md_thread *recovery_thread;
  54         unsigned long recovery_map;
  55         /* communication loc resources */
  56         struct dlm_lock_resource *ack_lockres;
  57         struct dlm_lock_resource *message_lockres;
  58         struct dlm_lock_resource *token_lockres;
  59         struct md_thread *recv_thread;
  60 };
  61
  62 enum msg_type {
  63         METADATA_UPDATED = 0,
  64         RESYNCING,
  65 };
  66
  67 struct cluster_msg {
  68         int type;
  69         int slot;
  70         sector_t low;
  71         sector_t high;
  72 };
  73
  74 static void sync_ast(void *arg)
  75 {
  76         struct dlm_lock_resource *res;
  77
  78         res = (struct dlm_lock_resource *) arg;
  79         complete(&res->completion);
  80 }
  81
  82 static int dlm_lock_sync(struct dlm_lock_resource *res, int mode)
  83 {
  84         int ret = 0;
  85
  86         init_completion(&res->completion);
  87         ret = dlm_lock(res->ls, mode, &res->lksb,
  88                         res->flags, res->name, strlen(res->name),
  89                         0, sync_ast, res, res->bast);
  90         if (ret)
  91                 return ret;
  92         wait_for_completion(&res->completion);
  93         return res->lksb.sb_status;
  94 }
  95
  96 static int dlm_unlock_sync(struct dlm_lock_resource *res)
  97 {
  98         return dlm_lock_sync(res, DLM_LOCK_NL);
  99 }
 100
 101 static struct dlm_lock_resource *lockres_init(struct mddev *mddev,
 102                 char *name, void (*bastfn)(void *arg, int mode), int with_lvb)
 103 {
 104         struct dlm_lock_resource *res = NULL;
 105         int ret, namelen;
 106         struct md_cluster_info *cinfo = mddev->cluster_info;
 107
 108         res = kzalloc(sizeof(struct dlm_lock_resource), GFP_KERNEL);
 109         if (!res)
 110                 return NULL;
 111         res->ls = cinfo->lockspace;
 112         res->mddev = mddev;
 113         namelen = strlen(name);
 114         res->name = kzalloc(namelen + 1, GFP_KERNEL);
 115         if (!res->name) {
 116                 pr_err("md-cluster: Unable to allocate resource name for resource %s\n", name);
 117                 goto out_err;
 118         }
 119         strlcpy(res->name, name, namelen + 1);
 120         if (with_lvb) {
 121                 res->lksb.sb_lvbptr = kzalloc(LVB_SIZE, GFP_KERNEL);
 122                 if (!res->lksb.sb_lvbptr) {
 123                         pr_err("md-cluster: Unable to allocate LVB for resource %s\n", name);
 124                         goto out_err;
 125                 }
 126                 res->flags = DLM_LKF_VALBLK;
 127         }
 128
 129         if (bastfn)
 130                 res->bast = bastfn;
 131
 132         res->flags |= DLM_LKF_EXPEDITE;
 133
 134         ret = dlm_lock_sync(res, DLM_LOCK_NL);
 135         if (ret) {
 136                 pr_err("md-cluster: Unable to lock NL on new lock resource %s\n", name);
 137                 goto out_err;
 138         }
 139         res->flags &= ~DLM_LKF_EXPEDITE;
 140         res->flags |= DLM_LKF_CONVERT;
 141
 142         return res;
 143 out_err:
 144         kfree(res->lksb.sb_lvbptr);
 145         kfree(res->name);
 146         kfree(res);
 147         return NULL;
 148 }
 149
 150 static void lockres_free(struct dlm_lock_resource *res)
 151 {
 152         if (!res)
 153                 return;
 154
 155         init_completion(&res->completion);
 156         dlm_unlock(res->ls, res->lksb.sb_lkid, 0, &res->lksb, res);
 157         wait_for_completion(&res->completion);
 158
 159         kfree(res->name);
 160         kfree(res->lksb.sb_lvbptr);
 161         kfree(res);
 162 }
 163
 164 static char *pretty_uuid(char *dest, char *src)
 165 {
 166         int i, len = 0;
 167
 168         for (i = 0; i < 16; i++) {
 169                 if (i == 4 || i == 6 || i == 8 || i == 10)
 170                         len += sprintf(dest + len, "-");
 171                 len += sprintf(dest + len, "%02x", (__u8)src[i]);
 172         }
 173         return dest;
 174 }
 175
 176 static void add_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres,
 177                 sector_t lo, sector_t hi)
 178 {
 179         struct resync_info *ri;
 180
 181         ri = (struct resync_info *)lockres->lksb.sb_lvbptr;
 182         ri->lo = cpu_to_le64(lo);
 183         ri->hi = cpu_to_le64(hi);
 184 }
 185
 186 static struct suspend_info *read_resync_info(struct mddev *mddev, struct dlm_lock_resource *lockres)
 187 {
 188         struct resync_info ri;
 189         struct suspend_info *s = NULL;
 190         sector_t hi = 0;
 191
 192         dlm_lock_sync(lockres, DLM_LOCK_CR);
 193         memcpy(&ri, lockres->lksb.sb_lvbptr, sizeof(struct resync_info));
 194         hi = le64_to_cpu(ri.hi);
 195         if (ri.hi > 0) {
 196                 s = kzalloc(sizeof(struct suspend_info), GFP_KERNEL);
 197                 if (!s)
 198                         goto out;
 199                 s->hi = hi;
 200                 s->lo = le64_to_cpu(ri.lo);
 201         }
 202         dlm_unlock_sync(lockres);
 203 out:
 204         return s;
 205 }
 206
 207 void recover_bitmaps(struct md_thread *thread)
 208 {
 209         struct mddev *mddev = thread->mddev;
 210         struct md_cluster_info *cinfo = mddev->cluster_info;
 211         struct dlm_lock_resource *bm_lockres;
 212         char str[64];
 213         int slot, ret;
 214         struct suspend_info *s, *tmp;
 215         sector_t lo, hi;
 216
 217         while (cinfo->recovery_map) {
 218                 slot = fls64((u64)cinfo->recovery_map) - 1;
 219
 220                 /* Clear suspend_area associated with the bitmap */
 221                 spin_lock_irq(&cinfo->suspend_lock);
 222                 list_for_each_entry_safe(s, tmp, &cinfo->suspend_list, list)
 223                         if (slot == s->slot) {
 224                                 list_del(&s->list);
 225                                 kfree(s);
 226                         }
 227                 spin_unlock_irq(&cinfo->suspend_lock);
 228
 229                 snprintf(str, 64, "bitmap%04d", slot);
 230                 bm_lockres = lockres_init(mddev, str, NULL, 1);
 231                 if (!bm_lockres) {
 232                         pr_err("md-cluster: Cannot initialize bitmaps\n");
 233                         goto clear_bit;
 234                 }
 235
 236                 ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
 237                 if (ret) {
 238                         pr_err("md-cluster: Could not DLM lock %s: %d\n",
 239                                         str, ret);
 240                         goto clear_bit;
 241                 }
 242                 ret = bitmap_copy_from_slot(mddev, slot, &lo, &hi);
 243                 if (ret) {
 244                         pr_err("md-cluster: Could not copy data from bitmap %d\n", slot);
 245                         goto dlm_unlock;
 246                 }
 247                 if (hi > 0) {
 248                         /* TODO:Wait for current resync to get over */
 249                         set_bit(MD_RECOVERY_NEEDED, &mddev->recovery);
 250                         if (lo < mddev->recovery_cp)
 251                                 mddev->recovery_cp = lo;
 252                         md_check_recovery(mddev);
 253                 }
 254 dlm_unlock:
 255                 dlm_unlock_sync(bm_lockres);
 256 clear_bit:
 257                 clear_bit(slot, &cinfo->recovery_map);
 258         }
 259 }
 260
 261 static void recover_prep(void *arg)
 262 {
 263 }
 264
 265 static void recover_slot(void *arg, struct dlm_slot *slot)
 266 {
 267         struct mddev *mddev = arg;
 268         struct md_cluster_info *cinfo = mddev->cluster_info;
 269
 270         pr_info("md-cluster: %s Node %d/%d down. My slot: %d. Initiating recovery.\n",
 271                         mddev->bitmap_info.cluster_name,
 272                         slot->nodeid, slot->slot,
 273                         cinfo->slot_number);
 274         set_bit(slot->slot - 1, &cinfo->recovery_map);
 275         if (!cinfo->recovery_thread) {
 276                 cinfo->recovery_thread = md_register_thread(recover_bitmaps,
 277                                 mddev, "recover");
 278                 if (!cinfo->recovery_thread) {
 279                         pr_warn("md-cluster: Could not create recovery thread\n");
 280                         return;
 281                 }
 282         }
 283         md_wakeup_thread(cinfo->recovery_thread);
 284 }
 285
 286 static void recover_done(void *arg, struct dlm_slot *slots,
 287                 int num_slots, int our_slot,
 288                 uint32_t generation)
 289 {
 290         struct mddev *mddev = arg;
 291         struct md_cluster_info *cinfo = mddev->cluster_info;
 292
 293         cinfo->slot_number = our_slot;
 294         complete(&cinfo->completion);
 295 }
 296
 297 static const struct dlm_lockspace_ops md_ls_ops = {
 298         .recover_prep = recover_prep,
 299         .recover_slot = recover_slot,
 300         .recover_done = recover_done,
 301 };
 302
 303 /*
 304  * The BAST function for the ack lock resource
 305  * This function wakes up the receive thread in
 306  * order to receive and process the message.
 307  */
 308 static void ack_bast(void *arg, int mode)
 309 {
 310         struct dlm_lock_resource *res = (struct dlm_lock_resource *)arg;
 311         struct md_cluster_info *cinfo = res->mddev->cluster_info;
 312
 313         if (mode == DLM_LOCK_EX)
 314                 md_wakeup_thread(cinfo->recv_thread);
 315 }
 316
 317 static void process_recvd_msg(struct mddev *mddev, struct cluster_msg *msg)
 318 {
 319         switch (msg->type) {
 320         case METADATA_UPDATED:
 321                 pr_info("%s: %d Received message: METADATA_UPDATE from %d\n",
 322                         __func__, __LINE__, msg->slot);
 323                 md_reload_sb(mddev);
 324                 break;
 325         case RESYNCING:
 326                 pr_info("%s: %d Received message: RESYNCING from %d\n",
 327                         __func__, __LINE__, msg->slot);
 328                 break;
 329         };
 330 }
 331
 332 /*
 333  * thread for receiving message
 334  */
 335 static void recv_daemon(struct md_thread *thread)
 336 {
 337         struct md_cluster_info *cinfo = thread->mddev->cluster_info;
 338         struct dlm_lock_resource *ack_lockres = cinfo->ack_lockres;
 339         struct dlm_lock_resource *message_lockres = cinfo->message_lockres;
 340         struct cluster_msg msg;
 341
 342         /*get CR on Message*/
 343         if (dlm_lock_sync(message_lockres, DLM_LOCK_CR)) {
 344                 pr_err("md/raid1:failed to get CR on MESSAGE\n");
 345                 return;
 346         }
 347
 348         /* read lvb and wake up thread to process this message_lockres */
 349         memcpy(&msg, message_lockres->lksb.sb_lvbptr, sizeof(struct cluster_msg));
 350         process_recvd_msg(thread->mddev, &msg);
 351
 352         /*release CR on ack_lockres*/
 353         dlm_unlock_sync(ack_lockres);
 354         /*up-convert to EX on message_lockres*/
 355         dlm_lock_sync(message_lockres, DLM_LOCK_EX);
 356         /*get CR on ack_lockres again*/
 357         dlm_lock_sync(ack_lockres, DLM_LOCK_CR);
 358         /*release CR on message_lockres*/
 359         dlm_unlock_sync(message_lockres);
 360 }
 361
 362 /* lock_comm()
 363  * Takes the lock on the TOKEN lock resource so no other
 364  * node can communicate while the operation is underway.
 365  */
 366 static int lock_comm(struct md_cluster_info *cinfo)
 367 {
 368         int error;
 369
 370         error = dlm_lock_sync(cinfo->token_lockres, DLM_LOCK_EX);
 371         if (error)
 372                 pr_err("md-cluster(%s:%d): failed to get EX on TOKEN (%d)\n",
 373                                 __func__, __LINE__, error);
 374         return error;
 375 }
 376
 377 static void unlock_comm(struct md_cluster_info *cinfo)
 378 {
 379         dlm_unlock_sync(cinfo->token_lockres);
 380 }
 381
 382 /* __sendmsg()
 383  * This function performs the actual sending of the message. This function is
 384  * usually called after performing the encompassing operation
 385  * The function:
 386  * 1. Grabs the message lockresource in EX mode
 387  * 2. Copies the message to the message LVB
 388  * 3. Downconverts message lockresource to CR
 389  * 4. Upconverts ack lock resource from CR to EX. This forces the BAST on other nodes
 390  *    and the other nodes read the message. The thread will wait here until all other
 391  *    nodes have released ack lock resource.
 392  * 5. Downconvert ack lockresource to CR
 393  */
 394 static int __sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
 395 {
 396         int error;
 397         int slot = cinfo->slot_number - 1;
 398
 399         cmsg->slot = cpu_to_le32(slot);
 400         /*get EX on Message*/
 401         error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_EX);
 402         if (error) {
 403                 pr_err("md-cluster: failed to get EX on MESSAGE (%d)\n", error);
 404                 goto failed_message;
 405         }
 406
 407         memcpy(cinfo->message_lockres->lksb.sb_lvbptr, (void *)cmsg,
 408                         sizeof(struct cluster_msg));
 409         /*down-convert EX to CR on Message*/
 410         error = dlm_lock_sync(cinfo->message_lockres, DLM_LOCK_CR);
 411         if (error) {
 412                 pr_err("md-cluster: failed to convert EX to CR on MESSAGE(%d)\n",
 413                                 error);
 414                 goto failed_message;
 415         }
 416
 417         /*up-convert CR to EX on Ack*/
 418         error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_EX);
 419         if (error) {
 420                 pr_err("md-cluster: failed to convert CR to EX on ACK(%d)\n",
 421                                 error);
 422                 goto failed_ack;
 423         }
 424
 425         /*down-convert EX to CR on Ack*/
 426         error = dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR);
 427         if (error) {
 428                 pr_err("md-cluster: failed to convert EX to CR on ACK(%d)\n",
 429                                 error);
 430                 goto failed_ack;
 431         }
 432
 433 failed_ack:
 434         dlm_unlock_sync(cinfo->message_lockres);
 435 failed_message:
 436         return error;
 437 }
 438
 439 static int sendmsg(struct md_cluster_info *cinfo, struct cluster_msg *cmsg)
 440 {
 441         int ret;
 442
 443         lock_comm(cinfo);
 444         ret = __sendmsg(cinfo, cmsg);
 445         unlock_comm(cinfo);
 446         return ret;
 447 }
 448
 449 static int gather_all_resync_info(struct mddev *mddev, int total_slots)
 450 {
 451         struct md_cluster_info *cinfo = mddev->cluster_info;
 452         int i, ret = 0;
 453         struct dlm_lock_resource *bm_lockres;
 454         struct suspend_info *s;
 455         char str[64];
 456
 457
 458         for (i = 0; i < total_slots; i++) {
 459                 memset(str, '\0', 64);
 460                 snprintf(str, 64, "bitmap%04d", i);
 461                 bm_lockres = lockres_init(mddev, str, NULL, 1);
 462                 if (!bm_lockres)
 463                         return -ENOMEM;
 464                 if (i == (cinfo->slot_number - 1))
 465                         continue;
 466
 467                 bm_lockres->flags |= DLM_LKF_NOQUEUE;
 468                 ret = dlm_lock_sync(bm_lockres, DLM_LOCK_PW);
 469                 if (ret == -EAGAIN) {
 470                         memset(bm_lockres->lksb.sb_lvbptr, '\0', LVB_SIZE);
 471                         s = read_resync_info(mddev, bm_lockres);
 472                         if (s) {
 473                                 pr_info("%s:%d Resync[%llu..%llu] in progress on %d\n",
 474                                                 __func__, __LINE__,
 475                                                 (unsigned long long) s->lo,
 476                                                 (unsigned long long) s->hi, i);
 477                                 spin_lock_irq(&cinfo->suspend_lock);
 478                                 s->slot = i;
 479                                 list_add(&s->list, &cinfo->suspend_list);
 480                                 spin_unlock_irq(&cinfo->suspend_lock);
 481                         }
 482                         ret = 0;
 483                         lockres_free(bm_lockres);
 484                         continue;
 485                 }
 486                 if (ret)
 487                         goto out;
 488                 /* TODO: Read the disk bitmap sb and check if it needs recovery */
 489                 dlm_unlock_sync(bm_lockres);
 490                 lockres_free(bm_lockres);
 491         }
 492 out:
 493         return ret;
 494 }
 495
 496 static int join(struct mddev *mddev, int nodes)
 497 {
 498         struct md_cluster_info *cinfo;
 499         int ret, ops_rv;
 500         char str[64];
 501
 502         if (!try_module_get(THIS_MODULE))
 503                 return -ENOENT;
 504
 505         cinfo = kzalloc(sizeof(struct md_cluster_info), GFP_KERNEL);
 506         if (!cinfo)
 507                 return -ENOMEM;
 508
 509         init_completion(&cinfo->completion);
 510
 511         mutex_init(&cinfo->sb_mutex);
 512         mddev->cluster_info = cinfo;
 513
 514         memset(str, 0, 64);
 515         pretty_uuid(str, mddev->uuid);
 516         ret = dlm_new_lockspace(str, mddev->bitmap_info.cluster_name,
 517                                 DLM_LSFL_FS, LVB_SIZE,
 518                                 &md_ls_ops, mddev, &ops_rv, &cinfo->lockspace);
 519         if (ret)
 520                 goto err;
 521         wait_for_completion(&cinfo->completion);
 522         if (nodes <= cinfo->slot_number) {
 523                 pr_err("md-cluster: Slot allotted(%d) greater than available slots(%d)", cinfo->slot_number - 1,
 524                         nodes);
 525                 ret = -ERANGE;
 526                 goto err;
 527         }
 528         cinfo->sb_lock = lockres_init(mddev, "cmd-super",
 529                                         NULL, 0);
 530         if (!cinfo->sb_lock) {
 531                 ret = -ENOMEM;
 532                 goto err;
 533         }
 534         /* Initiate the communication resources */
 535         ret = -ENOMEM;
 536         cinfo->recv_thread = md_register_thread(recv_daemon, mddev, "cluster_recv");
 537         if (!cinfo->recv_thread) {
 538                 pr_err("md-cluster: cannot allocate memory for recv_thread!\n");
 539                 goto err;
 540         }
 541         cinfo->message_lockres = lockres_init(mddev, "message", NULL, 1);
 542         if (!cinfo->message_lockres)
 543                 goto err;
 544         cinfo->token_lockres = lockres_init(mddev, "token", NULL, 0);
 545         if (!cinfo->token_lockres)
 546                 goto err;
 547         cinfo->ack_lockres = lockres_init(mddev, "ack", ack_bast, 0);
 548         if (!cinfo->ack_lockres)
 549                 goto err;
 550         /* get sync CR lock on ACK. */
 551         if (dlm_lock_sync(cinfo->ack_lockres, DLM_LOCK_CR))
 552                 pr_err("md-cluster: failed to get a sync CR lock on ACK!(%d)\n",
 553                                 ret);
 554
 555         pr_info("md-cluster: Joined cluster %s slot %d\n", str, cinfo->slot_number);
 556         snprintf(str, 64, "bitmap%04d", cinfo->slot_number - 1);
 557         cinfo->bitmap_lockres = lockres_init(mddev, str, NULL, 1);
 558         if (!cinfo->bitmap_lockres)
 559                 goto err;
 560         if (dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW)) {
 561                 pr_err("Failed to get bitmap lock\n");
 562                 ret = -EINVAL;
 563                 goto err;
 564         }
 565
 566         INIT_LIST_HEAD(&cinfo->suspend_list);
 567         spin_lock_init(&cinfo->suspend_lock);
 568
 569         ret = gather_all_resync_info(mddev, nodes);
 570         if (ret)
 571                 goto err;
 572
 573         return 0;
 574 err:
 575         lockres_free(cinfo->message_lockres);
 576         lockres_free(cinfo->token_lockres);
 577         lockres_free(cinfo->ack_lockres);
 578         lockres_free(cinfo->bitmap_lockres);
 579         lockres_free(cinfo->sb_lock);
 580         if (cinfo->lockspace)
 581                 dlm_release_lockspace(cinfo->lockspace, 2);
 582         mddev->cluster_info = NULL;
 583         kfree(cinfo);
 584         module_put(THIS_MODULE);
 585         return ret;
 586 }
 587
 588 static int leave(struct mddev *mddev)
 589 {
 590         struct md_cluster_info *cinfo = mddev->cluster_info;
 591
 592         if (!cinfo)
 593                 return 0;
 594         md_unregister_thread(&cinfo->recovery_thread);
 595         md_unregister_thread(&cinfo->recv_thread);
 596         lockres_free(cinfo->message_lockres);
 597         lockres_free(cinfo->token_lockres);
 598         lockres_free(cinfo->ack_lockres);
 599         lockres_free(cinfo->sb_lock);
 600         lockres_free(cinfo->bitmap_lockres);
 601         dlm_release_lockspace(cinfo->lockspace, 2);
 602         return 0;
 603 }
 604
 605 /* slot_number(): Returns the MD slot number to use
 606  * DLM starts the slot numbers from 1, wheras cluster-md
 607  * wants the number to be from zero, so we deduct one
 608  */
 609 static int slot_number(struct mddev *mddev)
 610 {
 611         struct md_cluster_info *cinfo = mddev->cluster_info;
 612
 613         return cinfo->slot_number - 1;
 614 }
 615
 616 static void resync_info_update(struct mddev *mddev, sector_t lo, sector_t hi)
 617 {
 618         struct md_cluster_info *cinfo = mddev->cluster_info;
 619
 620         add_resync_info(mddev, cinfo->bitmap_lockres, lo, hi);
 621         /* Re-acquire the lock to refresh LVB */
 622         dlm_lock_sync(cinfo->bitmap_lockres, DLM_LOCK_PW);
 623 }
 624
 625 static int metadata_update_start(struct mddev *mddev)
 626 {
 627         return lock_comm(mddev->cluster_info);
 628 }
 629
 630 static int metadata_update_finish(struct mddev *mddev)
 631 {
 632         struct md_cluster_info *cinfo = mddev->cluster_info;
 633         struct cluster_msg cmsg;
 634         int ret;
 635
 636         memset(&cmsg, 0, sizeof(cmsg));
 637         cmsg.type = cpu_to_le32(METADATA_UPDATED);
 638         ret = __sendmsg(cinfo, &cmsg);
 639         unlock_comm(cinfo);
 640         return ret;
 641 }
 642
 643 static int metadata_update_cancel(struct mddev *mddev)
 644 {
 645         struct md_cluster_info *cinfo = mddev->cluster_info;
 646
 647         return dlm_unlock_sync(cinfo->token_lockres);
 648 }
 649
 650 static struct md_cluster_operations cluster_ops = {
 651         .join   = join,
 652         .leave  = leave,
 653         .slot_number = slot_number,
 654         .resync_info_update = resync_info_update,
 655         .metadata_update_start = metadata_update_start,
 656         .metadata_update_finish = metadata_update_finish,
 657         .metadata_update_cancel = metadata_update_cancel,
 658 };
 659
 660 static int __init cluster_init(void)
 661 {
 662         pr_warn("md-cluster: EXPERIMENTAL. Use with caution\n");
 663         pr_info("Registering Cluster MD functions\n");
 664         register_md_cluster_operations(&cluster_ops, THIS_MODULE);
 665         return 0;
 666 }
 667
 668 static void cluster_exit(void)
 669 {
 670         unregister_md_cluster_operations();
 671 }
 672
 673 module_init(cluster_init);
 674 module_exit(cluster_exit);
 675 MODULE_LICENSE("GPL");
 676 MODULE_DESCRIPTION("Clustering support for MD");