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