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