drivers/block/rbd.c

   1 /*
   2    rbd.c -- Export ceph rados objects as a Linux block device
   3
   4
   5    based on drivers/block/osdblk.c:
   6
   7    Copyright 2009 Red Hat, Inc.
   8
   9    This program is free software; you can redistribute it and/or modify
  10    it under the terms of the GNU General Public License as published by
  11    the Free Software Foundation.
  12
  13    This program is distributed in the hope that it will be useful,
  14    but WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16    GNU General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; see the file COPYING.  If not, write to
  20    the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
  21
  22
  23
  24    For usage instructions, please refer to:
  25
  26                  Documentation/ABI/testing/sysfs-bus-rbd
  27
  28  */
  29
  30 #include <linux/ceph/libceph.h>
  31 #include <linux/ceph/osd_client.h>
  32 #include <linux/ceph/mon_client.h>
  33 #include <linux/ceph/decode.h>
  34 #include <linux/parser.h>
  35
  36 #include <linux/kernel.h>
  37 #include <linux/device.h>
  38 #include <linux/module.h>
  39 #include <linux/fs.h>
  40 #include <linux/blkdev.h>
  41
  42 #include "rbd_types.h"
  43
  44 #define DRV_NAME "rbd"
  45 #define DRV_NAME_LONG "rbd (rados block device)"
  46
  47 #define RBD_MINORS_PER_MAJOR    256             /* max minors per blkdev */
  48
  49 #define RBD_MAX_MD_NAME_LEN     (96 + sizeof(RBD_SUFFIX))
  50 #define RBD_MAX_POOL_NAME_LEN   64
  51 #define RBD_MAX_SNAP_NAME_LEN   32
  52 #define RBD_MAX_OPT_LEN         1024
  53
  54 #define RBD_SNAP_HEAD_NAME      "-"
  55
  56 #define DEV_NAME_LEN            32
  57
  58 #define RBD_NOTIFY_TIMEOUT_DEFAULT 10
  59
  60 /*
  61  * block device image metadata (in-memory version)
  62  */
  63 struct rbd_image_header {
  64         u64 image_size;
  65         char block_name[32];
  66         __u8 obj_order;
  67         __u8 crypt_type;
  68         __u8 comp_type;
  69         struct rw_semaphore snap_rwsem;
  70         struct ceph_snap_context *snapc;
  71         size_t snap_names_len;
  72         u64 snap_seq;
  73         u32 total_snaps;
  74
  75         char *snap_names;
  76         u64 *snap_sizes;
  77
  78         u64 obj_version;
  79 };
  80
  81 struct rbd_options {
  82         int     notify_timeout;
  83 };
  84
  85 /*
  86  * an instance of the client.  multiple devices may share a client.
  87  */
  88 struct rbd_client {
  89         struct ceph_client      *client;
  90         struct rbd_options      *rbd_opts;
  91         struct kref             kref;
  92         struct list_head        node;
  93 };
  94
  95 /*
  96  * a single io request
  97  */
  98 struct rbd_request {
  99         struct request          *rq;            /* blk layer request */
 100         struct bio              *bio;           /* cloned bio */
 101         struct page             **pages;        /* list of used pages */
 102         u64                     len;
 103 };
 104
 105 struct rbd_snap {
 106         struct  device          dev;
 107         const char              *name;
 108         size_t                  size;
 109         struct list_head        node;
 110         u64                     id;
 111 };
 112
 113 /*
 114  * a single device
 115  */
 116 struct rbd_device {
 117         int                     id;             /* blkdev unique id */
 118
 119         int                     major;          /* blkdev assigned major */
 120         struct gendisk          *disk;          /* blkdev's gendisk and rq */
 121         struct request_queue    *q;
 122
 123         struct ceph_client      *client;
 124         struct rbd_client       *rbd_client;
 125
 126         char                    name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
 127
 128         spinlock_t              lock;           /* queue lock */
 129
 130         struct rbd_image_header header;
 131         char                    obj[RBD_MAX_OBJ_NAME_LEN]; /* rbd image name */
 132         int                     obj_len;
 133         char                    obj_md_name[RBD_MAX_MD_NAME_LEN]; /* hdr nm. */
 134         char                    pool_name[RBD_MAX_POOL_NAME_LEN];
 135         int                     poolid;
 136
 137         struct ceph_osd_event   *watch_event;
 138         struct ceph_osd_request *watch_request;
 139
 140         char                    snap_name[RBD_MAX_SNAP_NAME_LEN];
 141         u32 cur_snap;   /* index+1 of current snapshot within snap context
 142                            0 - for the head */
 143         int read_only;
 144
 145         struct list_head        node;
 146
 147         /* list of snapshots */
 148         struct list_head        snaps;
 149
 150         /* sysfs related */
 151         struct device           dev;
 152 };
 153
 154 static struct bus_type rbd_bus_type = {
 155         .name           = "rbd",
 156 };
 157
 158 static spinlock_t node_lock;      /* protects client get/put */
 159
 160 static DEFINE_MUTEX(ctl_mutex);   /* Serialize open/close/setup/teardown */
 161 static LIST_HEAD(rbd_dev_list);    /* devices */
 162 static LIST_HEAD(rbd_client_list);      /* clients */
 163
 164 static int __rbd_init_snaps_header(struct rbd_device *rbd_dev);
 165 static void rbd_dev_release(struct device *dev);
 166 static ssize_t rbd_snap_rollback(struct device *dev,
 167                                  struct device_attribute *attr,
 168                                  const char *buf,
 169                                  size_t size);
 170 static ssize_t rbd_snap_add(struct device *dev,
 171                             struct device_attribute *attr,
 172                             const char *buf,
 173                             size_t count);
 174 static void __rbd_remove_snap_dev(struct rbd_device *rbd_dev,
 175                                   struct rbd_snap *snap);;
 176
 177
 178 static struct rbd_device *dev_to_rbd(struct device *dev)
 179 {
 180         return container_of(dev, struct rbd_device, dev);
 181 }
 182
 183 static struct device *rbd_get_dev(struct rbd_device *rbd_dev)
 184 {
 185         return get_device(&rbd_dev->dev);
 186 }
 187
 188 static void rbd_put_dev(struct rbd_device *rbd_dev)
 189 {
 190         put_device(&rbd_dev->dev);
 191 }
 192
 193 static int __rbd_update_snaps(struct rbd_device *rbd_dev);
 194
 195 static int rbd_open(struct block_device *bdev, fmode_t mode)
 196 {
 197         struct gendisk *disk = bdev->bd_disk;
 198         struct rbd_device *rbd_dev = disk->private_data;
 199
 200         rbd_get_dev(rbd_dev);
 201
 202         set_device_ro(bdev, rbd_dev->read_only);
 203
 204         if ((mode & FMODE_WRITE) && rbd_dev->read_only)
 205                 return -EROFS;
 206
 207         return 0;
 208 }
 209
 210 static int rbd_release(struct gendisk *disk, fmode_t mode)
 211 {
 212         struct rbd_device *rbd_dev = disk->private_data;
 213
 214         rbd_put_dev(rbd_dev);
 215
 216         return 0;
 217 }
 218
 219 static const struct block_device_operations rbd_bd_ops = {
 220         .owner                  = THIS_MODULE,
 221         .open                   = rbd_open,
 222         .release                = rbd_release,
 223 };
 224
 225 /*
 226  * Initialize an rbd client instance.
 227  * We own *opt.
 228  */
 229 static struct rbd_client *rbd_client_create(struct ceph_options *opt,
 230                                             struct rbd_options *rbd_opts)
 231 {
 232         struct rbd_client *rbdc;
 233         int ret = -ENOMEM;
 234
 235         dout("rbd_client_create\n");
 236         rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
 237         if (!rbdc)
 238                 goto out_opt;
 239
 240         kref_init(&rbdc->kref);
 241         INIT_LIST_HEAD(&rbdc->node);
 242
 243         rbdc->client = ceph_create_client(opt, rbdc);
 244         if (IS_ERR(rbdc->client))
 245                 goto out_rbdc;
 246         opt = NULL; /* Now rbdc->client is responsible for opt */
 247
 248         ret = ceph_open_session(rbdc->client);
 249         if (ret < 0)
 250                 goto out_err;
 251
 252         rbdc->rbd_opts = rbd_opts;
 253
 254         spin_lock(&node_lock);
 255         list_add_tail(&rbdc->node, &rbd_client_list);
 256         spin_unlock(&node_lock);
 257
 258         dout("rbd_client_create created %p\n", rbdc);
 259         return rbdc;
 260
 261 out_err:
 262         ceph_destroy_client(rbdc->client);
 263 out_rbdc:
 264         kfree(rbdc);
 265 out_opt:
 266         if (opt)
 267                 ceph_destroy_options(opt);
 268         return ERR_PTR(ret);
 269 }
 270
 271 /*
 272  * Find a ceph client with specific addr and configuration.
 273  */
 274 static struct rbd_client *__rbd_client_find(struct ceph_options *opt)
 275 {
 276         struct rbd_client *client_node;
 277
 278         if (opt->flags & CEPH_OPT_NOSHARE)
 279                 return NULL;
 280
 281         list_for_each_entry(client_node, &rbd_client_list, node)
 282                 if (ceph_compare_options(opt, client_node->client) == 0)
 283                         return client_node;
 284         return NULL;
 285 }
 286
 287 /*
 288  * mount options
 289  */
 290 enum {
 291         Opt_notify_timeout,
 292         Opt_last_int,
 293         /* int args above */
 294         Opt_last_string,
 295         /* string args above */
 296 };
 297
 298 static match_table_t rbdopt_tokens = {
 299         {Opt_notify_timeout, "notify_timeout=%d"},
 300         /* int args above */
 301         /* string args above */
 302         {-1, NULL}
 303 };
 304
 305 static int parse_rbd_opts_token(char *c, void *private)
 306 {
 307         struct rbd_options *rbdopt = private;
 308         substring_t argstr[MAX_OPT_ARGS];
 309         int token, intval, ret;
 310
 311         token = match_token((char *)c, rbdopt_tokens, argstr);
 312         if (token < 0)
 313                 return -EINVAL;
 314
 315         if (token < Opt_last_int) {
 316                 ret = match_int(&argstr[0], &intval);
 317                 if (ret < 0) {
 318                         pr_err("bad mount option arg (not int) "
 319                                "at '%s'\n", c);
 320                         return ret;
 321                 }
 322                 dout("got int token %d val %d\n", token, intval);
 323         } else if (token > Opt_last_int && token < Opt_last_string) {
 324                 dout("got string token %d val %s\n", token,
 325                      argstr[0].from);
 326         } else {
 327                 dout("got token %d\n", token);
 328         }
 329
 330         switch (token) {
 331         case Opt_notify_timeout:
 332                 rbdopt->notify_timeout = intval;
 333                 break;
 334         default:
 335                 BUG_ON(token);
 336         }
 337         return 0;
 338 }
 339
 340 /*
 341  * Get a ceph client with specific addr and configuration, if one does
 342  * not exist create it.
 343  */
 344 static int rbd_get_client(struct rbd_device *rbd_dev, const char *mon_addr,
 345                           char *options)
 346 {
 347         struct rbd_client *rbdc;
 348         struct ceph_options *opt;
 349         int ret;
 350         struct rbd_options *rbd_opts;
 351
 352         rbd_opts = kzalloc(sizeof(*rbd_opts), GFP_KERNEL);
 353         if (!rbd_opts)
 354                 return -ENOMEM;
 355
 356         rbd_opts->notify_timeout = RBD_NOTIFY_TIMEOUT_DEFAULT;
 357
 358         ret = ceph_parse_options(&opt, options, mon_addr,
 359                                  mon_addr + strlen(mon_addr), parse_rbd_opts_token, rbd_opts);
 360         if (ret < 0)
 361                 goto done_err;
 362
 363         spin_lock(&node_lock);
 364         rbdc = __rbd_client_find(opt);
 365         if (rbdc) {
 366                 ceph_destroy_options(opt);
 367
 368                 /* using an existing client */
 369                 kref_get(&rbdc->kref);
 370                 rbd_dev->rbd_client = rbdc;
 371                 rbd_dev->client = rbdc->client;
 372                 spin_unlock(&node_lock);
 373                 return 0;
 374         }
 375         spin_unlock(&node_lock);
 376
 377         rbdc = rbd_client_create(opt, rbd_opts);
 378         if (IS_ERR(rbdc)) {
 379                 ret = PTR_ERR(rbdc);
 380                 goto done_err;
 381         }
 382
 383         rbd_dev->rbd_client = rbdc;
 384         rbd_dev->client = rbdc->client;
 385         return 0;
 386 done_err:
 387         kfree(rbd_opts);
 388         return ret;
 389 }
 390
 391 /*
 392  * Destroy ceph client
 393  */
 394 static void rbd_client_release(struct kref *kref)
 395 {
 396         struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
 397
 398         dout("rbd_release_client %p\n", rbdc);
 399         spin_lock(&node_lock);
 400         list_del(&rbdc->node);
 401         spin_unlock(&node_lock);
 402
 403         ceph_destroy_client(rbdc->client);
 404         kfree(rbdc->rbd_opts);
 405         kfree(rbdc);
 406 }
 407
 408 /*
 409  * Drop reference to ceph client node. If it's not referenced anymore, release
 410  * it.
 411  */
 412 static void rbd_put_client(struct rbd_device *rbd_dev)
 413 {
 414         kref_put(&rbd_dev->rbd_client->kref, rbd_client_release);
 415         rbd_dev->rbd_client = NULL;
 416         rbd_dev->client = NULL;
 417 }
 418
 419
 420 /*
 421  * Create a new header structure, translate header format from the on-disk
 422  * header.
 423  */
 424 static int rbd_header_from_disk(struct rbd_image_header *header,
 425                                  struct rbd_image_header_ondisk *ondisk,
 426                                  int allocated_snaps,
 427                                  gfp_t gfp_flags)
 428 {
 429         int i;
 430         u32 snap_count = le32_to_cpu(ondisk->snap_count);
 431         int ret = -ENOMEM;
 432
 433         init_rwsem(&header->snap_rwsem);
 434         header->snap_names_len = le64_to_cpu(ondisk->snap_names_len);
 435         header->snapc = kmalloc(sizeof(struct ceph_snap_context) +
 436                                 snap_count *
 437                                  sizeof(struct rbd_image_snap_ondisk),
 438                                 gfp_flags);
 439         if (!header->snapc)
 440                 return -ENOMEM;
 441         if (snap_count) {
 442                 header->snap_names = kmalloc(header->snap_names_len,
 443                                              GFP_KERNEL);
 444                 if (!header->snap_names)
 445                         goto err_snapc;
 446                 header->snap_sizes = kmalloc(snap_count * sizeof(u64),
 447                                              GFP_KERNEL);
 448                 if (!header->snap_sizes)
 449                         goto err_names;
 450         } else {
 451                 header->snap_names = NULL;
 452                 header->snap_sizes = NULL;
 453         }
 454         memcpy(header->block_name, ondisk->block_name,
 455                sizeof(ondisk->block_name));
 456
 457         header->image_size = le64_to_cpu(ondisk->image_size);
 458         header->obj_order = ondisk->options.order;
 459         header->crypt_type = ondisk->options.crypt_type;
 460         header->comp_type = ondisk->options.comp_type;
 461
 462         atomic_set(&header->snapc->nref, 1);
 463         header->snap_seq = le64_to_cpu(ondisk->snap_seq);
 464         header->snapc->num_snaps = snap_count;
 465         header->total_snaps = snap_count;
 466
 467         if (snap_count &&
 468             allocated_snaps == snap_count) {
 469                 for (i = 0; i < snap_count; i++) {
 470                         header->snapc->snaps[i] =
 471                                 le64_to_cpu(ondisk->snaps[i].id);
 472                         header->snap_sizes[i] =
 473                                 le64_to_cpu(ondisk->snaps[i].image_size);
 474                 }
 475
 476                 /* copy snapshot names */
 477                 memcpy(header->snap_names, &ondisk->snaps[i],
 478                         header->snap_names_len);
 479         }
 480
 481         return 0;
 482
 483 err_names:
 484         kfree(header->snap_names);
 485 err_snapc:
 486         kfree(header->snapc);
 487         return ret;
 488 }
 489
 490 static int snap_index(struct rbd_image_header *header, int snap_num)
 491 {
 492         return header->total_snaps - snap_num;
 493 }
 494
 495 static u64 cur_snap_id(struct rbd_device *rbd_dev)
 496 {
 497         struct rbd_image_header *header = &rbd_dev->header;
 498
 499         if (!rbd_dev->cur_snap)
 500                 return 0;
 501
 502         return header->snapc->snaps[snap_index(header, rbd_dev->cur_snap)];
 503 }
 504
 505 static int snap_by_name(struct rbd_image_header *header, const char *snap_name,
 506                         u64 *seq, u64 *size)
 507 {
 508         int i;
 509         char *p = header->snap_names;
 510
 511         for (i = 0; i < header->total_snaps; i++, p += strlen(p) + 1) {
 512                 if (strcmp(snap_name, p) == 0)
 513                         break;
 514         }
 515         if (i == header->total_snaps)
 516                 return -ENOENT;
 517         if (seq)
 518                 *seq = header->snapc->snaps[i];
 519
 520         if (size)
 521                 *size = header->snap_sizes[i];
 522
 523         return i;
 524 }
 525
 526 static int rbd_header_set_snap(struct rbd_device *dev,
 527                                const char *snap_name,
 528                                u64 *size)
 529 {
 530         struct rbd_image_header *header = &dev->header;
 531         struct ceph_snap_context *snapc = header->snapc;
 532         int ret = -ENOENT;
 533
 534         down_write(&header->snap_rwsem);
 535
 536         if (!snap_name ||
 537             !*snap_name ||
 538             strcmp(snap_name, "-") == 0 ||
 539             strcmp(snap_name, RBD_SNAP_HEAD_NAME) == 0) {
 540                 if (header->total_snaps)
 541                         snapc->seq = header->snap_seq;
 542                 else
 543                         snapc->seq = 0;
 544                 dev->cur_snap = 0;
 545                 dev->read_only = 0;
 546                 if (size)
 547                         *size = header->image_size;
 548         } else {
 549                 ret = snap_by_name(header, snap_name, &snapc->seq, size);
 550                 if (ret < 0)
 551                         goto done;
 552
 553                 dev->cur_snap = header->total_snaps - ret;
 554                 dev->read_only = 1;
 555         }
 556
 557         ret = 0;
 558 done:
 559         up_write(&header->snap_rwsem);
 560         return ret;
 561 }
 562
 563 static void rbd_header_free(struct rbd_image_header *header)
 564 {
 565         kfree(header->snapc);
 566         kfree(header->snap_names);
 567         kfree(header->snap_sizes);
 568 }
 569
 570 /*
 571  * get the actual striped segment name, offset and length
 572  */
 573 static u64 rbd_get_segment(struct rbd_image_header *header,
 574                            const char *block_name,
 575                            u64 ofs, u64 len,
 576                            char *seg_name, u64 *segofs)
 577 {
 578         u64 seg = ofs >> header->obj_order;
 579
 580         if (seg_name)
 581                 snprintf(seg_name, RBD_MAX_SEG_NAME_LEN,
 582                          "%s.%012llx", block_name, seg);
 583
 584         ofs = ofs & ((1 << header->obj_order) - 1);
 585         len = min_t(u64, len, (1 << header->obj_order) - ofs);
 586
 587         if (segofs)
 588                 *segofs = ofs;
 589
 590         return len;
 591 }
 592
 593 /*
 594  * bio helpers
 595  */
 596
 597 static void bio_chain_put(struct bio *chain)
 598 {
 599         struct bio *tmp;
 600
 601         while (chain) {
 602                 tmp = chain;
 603                 chain = chain->bi_next;
 604                 bio_put(tmp);
 605         }
 606 }
 607
 608 /*
 609  * zeros a bio chain, starting at specific offset
 610  */
 611 static void zero_bio_chain(struct bio *chain, int start_ofs)
 612 {
 613         struct bio_vec *bv;
 614         unsigned long flags;
 615         void *buf;
 616         int i;
 617         int pos = 0;
 618
 619         while (chain) {
 620                 bio_for_each_segment(bv, chain, i) {
 621                         if (pos + bv->bv_len > start_ofs) {
 622                                 int remainder = max(start_ofs - pos, 0);
 623                                 buf = bvec_kmap_irq(bv, &flags);
 624                                 memset(buf + remainder, 0,
 625                                        bv->bv_len - remainder);
 626                                 bvec_kunmap_irq(buf, &flags);
 627                         }
 628                         pos += bv->bv_len;
 629                 }
 630
 631                 chain = chain->bi_next;
 632         }
 633 }
 634
 635 /*
 636  * bio_chain_clone - clone a chain of bios up to a certain length.
 637  * might return a bio_pair that will need to be released.
 638  */
 639 static struct bio *bio_chain_clone(struct bio **old, struct bio **next,
 640                                    struct bio_pair **bp,
 641                                    int len, gfp_t gfpmask)
 642 {
 643         struct bio *tmp, *old_chain = *old, *new_chain = NULL, *tail = NULL;
 644         int total = 0;
 645
 646         if (*bp) {
 647                 bio_pair_release(*bp);
 648                 *bp = NULL;
 649         }
 650
 651         while (old_chain && (total < len)) {
 652                 tmp = bio_kmalloc(gfpmask, old_chain->bi_max_vecs);
 653                 if (!tmp)
 654                         goto err_out;
 655
 656                 if (total + old_chain->bi_size > len) {
 657                         struct bio_pair *bp;
 658
 659                         /*
 660                          * this split can only happen with a single paged bio,
 661                          * split_bio will BUG_ON if this is not the case
 662                          */
 663                         dout("bio_chain_clone split! total=%d remaining=%d"
 664                              "bi_size=%d\n",
 665                              (int)total, (int)len-total,
 666                              (int)old_chain->bi_size);
 667
 668                         /* split the bio. We'll release it either in the next
 669                            call, or it will have to be released outside */
 670                         bp = bio_split(old_chain, (len - total) / 512ULL);
 671                         if (!bp)
 672                                 goto err_out;
 673
 674                         __bio_clone(tmp, &bp->bio1);
 675
 676                         *next = &bp->bio2;
 677                 } else {
 678                         __bio_clone(tmp, old_chain);
 679                         *next = old_chain->bi_next;
 680                 }
 681
 682                 tmp->bi_bdev = NULL;
 683                 gfpmask &= ~__GFP_WAIT;
 684                 tmp->bi_next = NULL;
 685
 686                 if (!new_chain) {
 687                         new_chain = tail = tmp;
 688                 } else {
 689                         tail->bi_next = tmp;
 690                         tail = tmp;
 691                 }
 692                 old_chain = old_chain->bi_next;
 693
 694                 total += tmp->bi_size;
 695         }
 696
 697         BUG_ON(total < len);
 698
 699         if (tail)
 700                 tail->bi_next = NULL;
 701
 702         *old = old_chain;
 703
 704         return new_chain;
 705
 706 err_out:
 707         dout("bio_chain_clone with err\n");
 708         bio_chain_put(new_chain);
 709         return NULL;
 710 }
 711
 712 /*
 713  * helpers for osd request op vectors.
 714  */
 715 static int rbd_create_rw_ops(struct ceph_osd_req_op **ops,
 716                             int num_ops,
 717                             int opcode,
 718                             u32 payload_len)
 719 {
 720         *ops = kzalloc(sizeof(struct ceph_osd_req_op) * (num_ops + 1),
 721                        GFP_NOIO);
 722         if (!*ops)
 723                 return -ENOMEM;
 724         (*ops)[0].op = opcode;
 725         /*
 726          * op extent offset and length will be set later on
 727          * in calc_raw_layout()
 728          */
 729         (*ops)[0].payload_len = payload_len;
 730         return 0;
 731 }
 732
 733 static void rbd_destroy_ops(struct ceph_osd_req_op *ops)
 734 {
 735         kfree(ops);
 736 }
 737
 738 /*
 739  * Send ceph osd request
 740  */
 741 static int rbd_do_request(struct request *rq,
 742                           struct rbd_device *dev,
 743                           struct ceph_snap_context *snapc,
 744                           u64 snapid,
 745                           const char *obj, u64 ofs, u64 len,
 746                           struct bio *bio,
 747                           struct page **pages,
 748                           int num_pages,
 749                           int flags,
 750                           struct ceph_osd_req_op *ops,
 751                           int num_reply,
 752                           void (*rbd_cb)(struct ceph_osd_request *req,
 753                                          struct ceph_msg *msg),
 754                           struct ceph_osd_request **linger_req,
 755                           u64 *ver)
 756 {
 757         struct ceph_osd_request *req;
 758         struct ceph_file_layout *layout;
 759         int ret;
 760         u64 bno;
 761         struct timespec mtime = CURRENT_TIME;
 762         struct rbd_request *req_data;
 763         struct ceph_osd_request_head *reqhead;
 764         struct rbd_image_header *header = &dev->header;
 765
 766         ret = -ENOMEM;
 767         req_data = kzalloc(sizeof(*req_data), GFP_NOIO);
 768         if (!req_data)
 769                 goto done;
 770
 771         dout("rbd_do_request len=%lld ofs=%lld\n", len, ofs);
 772
 773         down_read(&header->snap_rwsem);
 774
 775         req = ceph_osdc_alloc_request(&dev->client->osdc, flags,
 776                                       snapc,
 777                                       ops,
 778                                       false,
 779                                       GFP_NOIO, pages, bio);
 780         if (IS_ERR(req)) {
 781                 up_read(&header->snap_rwsem);
 782                 ret = PTR_ERR(req);
 783                 goto done_pages;
 784         }
 785
 786         req->r_callback = rbd_cb;
 787
 788         req_data->rq = rq;
 789         req_data->bio = bio;
 790         req_data->pages = pages;
 791         req_data->len = len;
 792
 793         req->r_priv = req_data;
 794
 795         reqhead = req->r_request->front.iov_base;
 796         reqhead->snapid = cpu_to_le64(CEPH_NOSNAP);
 797
 798         strncpy(req->r_oid, obj, sizeof(req->r_oid));
 799         req->r_oid_len = strlen(req->r_oid);
 800
 801         layout = &req->r_file_layout;
 802         memset(layout, 0, sizeof(*layout));
 803         layout->fl_stripe_unit = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
 804         layout->fl_stripe_count = cpu_to_le32(1);
 805         layout->fl_object_size = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
 806         layout->fl_pg_preferred = cpu_to_le32(-1);
 807         layout->fl_pg_pool = cpu_to_le32(dev->poolid);
 808         ceph_calc_raw_layout(&dev->client->osdc, layout, snapid,
 809                              ofs, &len, &bno, req, ops);
 810
 811         ceph_osdc_build_request(req, ofs, &len,
 812                                 ops,
 813                                 snapc,
 814                                 &mtime,
 815                                 req->r_oid, req->r_oid_len);
 816         up_read(&header->snap_rwsem);
 817
 818         if (linger_req) {
 819                 ceph_osdc_set_request_linger(&dev->client->osdc, req);
 820                 *linger_req = req;
 821         }
 822
 823         ret = ceph_osdc_start_request(&dev->client->osdc, req, false);
 824         if (ret < 0)
 825                 goto done_err;
 826
 827         if (!rbd_cb) {
 828                 ret = ceph_osdc_wait_request(&dev->client->osdc, req);
 829                 if (ver)
 830                         *ver = le64_to_cpu(req->r_reassert_version.version);
 831                 dout("reassert_ver=%lld\n", le64_to_cpu(req->r_reassert_version.version));
 832                 ceph_osdc_put_request(req);
 833         }
 834         return ret;
 835
 836 done_err:
 837         bio_chain_put(req_data->bio);
 838         ceph_osdc_put_request(req);
 839 done_pages:
 840         kfree(req_data);
 841 done:
 842         if (rq)
 843                 blk_end_request(rq, ret, len);
 844         return ret;
 845 }
 846
 847 /*
 848  * Ceph osd op callback
 849  */
 850 static void rbd_req_cb(struct ceph_osd_request *req, struct ceph_msg *msg)
 851 {
 852         struct rbd_request *req_data = req->r_priv;
 853         struct ceph_osd_reply_head *replyhead;
 854         struct ceph_osd_op *op;
 855         __s32 rc;
 856         u64 bytes;
 857         int read_op;
 858
 859         /* parse reply */
 860         replyhead = msg->front.iov_base;
 861         WARN_ON(le32_to_cpu(replyhead->num_ops) == 0);
 862         op = (void *)(replyhead + 1);
 863         rc = le32_to_cpu(replyhead->result);
 864         bytes = le64_to_cpu(op->extent.length);
 865         read_op = (le32_to_cpu(op->op) == CEPH_OSD_OP_READ);
 866
 867         dout("rbd_req_cb bytes=%lld readop=%d rc=%d\n", bytes, read_op, rc);
 868
 869         if (rc == -ENOENT && read_op) {
 870                 zero_bio_chain(req_data->bio, 0);
 871                 rc = 0;
 872         } else if (rc == 0 && read_op && bytes < req_data->len) {
 873                 zero_bio_chain(req_data->bio, bytes);
 874                 bytes = req_data->len;
 875         }
 876
 877         blk_end_request(req_data->rq, rc, bytes);
 878
 879         if (req_data->bio)
 880                 bio_chain_put(req_data->bio);
 881
 882         ceph_osdc_put_request(req);
 883         kfree(req_data);
 884 }
 885
 886 static void rbd_simple_req_cb(struct ceph_osd_request *req, struct ceph_msg *msg)
 887 {
 888         ceph_osdc_put_request(req);
 889 }
 890
 891 /*
 892  * Do a synchronous ceph osd operation
 893  */
 894 static int rbd_req_sync_op(struct rbd_device *dev,
 895                            struct ceph_snap_context *snapc,
 896                            u64 snapid,
 897                            int opcode,
 898                            int flags,
 899                            struct ceph_osd_req_op *orig_ops,
 900                            int num_reply,
 901                            const char *obj,
 902                            u64 ofs, u64 len,
 903                            char *buf,
 904                            struct ceph_osd_request **linger_req,
 905                            u64 *ver)
 906 {
 907         int ret;
 908         struct page **pages;
 909         int num_pages;
 910         struct ceph_osd_req_op *ops = orig_ops;
 911         u32 payload_len;
 912
 913         num_pages = calc_pages_for(ofs , len);
 914         pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
 915         if (IS_ERR(pages))
 916                 return PTR_ERR(pages);
 917
 918         if (!orig_ops) {
 919                 payload_len = (flags & CEPH_OSD_FLAG_WRITE ? len : 0);
 920                 ret = rbd_create_rw_ops(&ops, 1, opcode, payload_len);
 921                 if (ret < 0)
 922                         goto done;
 923
 924                 if ((flags & CEPH_OSD_FLAG_WRITE) && buf) {
 925                         ret = ceph_copy_to_page_vector(pages, buf, ofs, len);
 926                         if (ret < 0)
 927                                 goto done_ops;
 928                 }
 929         }
 930
 931         ret = rbd_do_request(NULL, dev, snapc, snapid,
 932                           obj, ofs, len, NULL,
 933                           pages, num_pages,
 934                           flags,
 935                           ops,
 936                           2,
 937                           NULL,
 938                           linger_req, ver);
 939         if (ret < 0)
 940                 goto done_ops;
 941
 942         if ((flags & CEPH_OSD_FLAG_READ) && buf)
 943                 ret = ceph_copy_from_page_vector(pages, buf, ofs, ret);
 944
 945 done_ops:
 946         if (!orig_ops)
 947                 rbd_destroy_ops(ops);
 948 done:
 949         ceph_release_page_vector(pages, num_pages);
 950         return ret;
 951 }
 952
 953 /*
 954  * Do an asynchronous ceph osd operation
 955  */
 956 static int rbd_do_op(struct request *rq,
 957                      struct rbd_device *rbd_dev ,
 958                      struct ceph_snap_context *snapc,
 959                      u64 snapid,
 960                      int opcode, int flags, int num_reply,
 961                      u64 ofs, u64 len,
 962                      struct bio *bio)
 963 {
 964         char *seg_name;
 965         u64 seg_ofs;
 966         u64 seg_len;
 967         int ret;
 968         struct ceph_osd_req_op *ops;
 969         u32 payload_len;
 970
 971         seg_name = kmalloc(RBD_MAX_SEG_NAME_LEN + 1, GFP_NOIO);
 972         if (!seg_name)
 973                 return -ENOMEM;
 974
 975         seg_len = rbd_get_segment(&rbd_dev->header,
 976                                   rbd_dev->header.block_name,
 977                                   ofs, len,
 978                                   seg_name, &seg_ofs);
 979
 980         payload_len = (flags & CEPH_OSD_FLAG_WRITE ? seg_len : 0);
 981
 982         ret = rbd_create_rw_ops(&ops, 1, opcode, payload_len);
 983         if (ret < 0)
 984                 goto done;
 985
 986         /* we've taken care of segment sizes earlier when we
 987            cloned the bios. We should never have a segment
 988            truncated at this point */
 989         BUG_ON(seg_len < len);
 990
 991         ret = rbd_do_request(rq, rbd_dev, snapc, snapid,
 992                              seg_name, seg_ofs, seg_len,
 993                              bio,
 994                              NULL, 0,
 995                              flags,
 996                              ops,
 997                              num_reply,
 998                              rbd_req_cb, 0, NULL);
 999 done:
1000         kfree(seg_name);
1001         return ret;
1002 }
1003
1004 /*
1005  * Request async osd write
1006  */
1007 static int rbd_req_write(struct request *rq,
1008                          struct rbd_device *rbd_dev,
1009                          struct ceph_snap_context *snapc,
1010                          u64 ofs, u64 len,
1011                          struct bio *bio)
1012 {
1013         return rbd_do_op(rq, rbd_dev, snapc, CEPH_NOSNAP,
1014                          CEPH_OSD_OP_WRITE,
1015                          CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1016                          2,
1017                          ofs, len, bio);
1018 }
1019
1020 /*
1021  * Request async osd read
1022  */
1023 static int rbd_req_read(struct request *rq,
1024                          struct rbd_device *rbd_dev,
1025                          u64 snapid,
1026                          u64 ofs, u64 len,
1027                          struct bio *bio)
1028 {
1029         return rbd_do_op(rq, rbd_dev, NULL,
1030                          (snapid ? snapid : CEPH_NOSNAP),
1031                          CEPH_OSD_OP_READ,
1032                          CEPH_OSD_FLAG_READ,
1033                          2,
1034                          ofs, len, bio);
1035 }
1036
1037 /*
1038  * Request sync osd read
1039  */
1040 static int rbd_req_sync_read(struct rbd_device *dev,
1041                           struct ceph_snap_context *snapc,
1042                           u64 snapid,
1043                           const char *obj,
1044                           u64 ofs, u64 len,
1045                           char *buf,
1046                           u64 *ver)
1047 {
1048         return rbd_req_sync_op(dev, NULL,
1049                                (snapid ? snapid : CEPH_NOSNAP),
1050                                CEPH_OSD_OP_READ,
1051                                CEPH_OSD_FLAG_READ,
1052                                NULL,
1053                                1, obj, ofs, len, buf, NULL, ver);
1054 }
1055
1056 /*
1057  * Request sync osd watch
1058  */
1059 static int rbd_req_sync_notify_ack(struct rbd_device *dev,
1060                                    u64 ver,
1061                                    u64 notify_id,
1062                                    const char *obj)
1063 {
1064         struct ceph_osd_req_op *ops;
1065         struct page **pages = NULL;
1066         int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_NOTIFY_ACK, 0);
1067         if (ret < 0)
1068                 return ret;
1069
1070         ops[0].watch.ver = cpu_to_le64(dev->header.obj_version);
1071         ops[0].watch.cookie = notify_id;
1072         ops[0].watch.flag = 0;
1073
1074         ret = rbd_do_request(NULL, dev, NULL, CEPH_NOSNAP,
1075                           obj, 0, 0, NULL,
1076                           pages, 0,
1077                           CEPH_OSD_FLAG_READ,
1078                           ops,
1079                           1,
1080                           rbd_simple_req_cb, 0, NULL);
1081
1082         rbd_destroy_ops(ops);
1083         return ret;
1084 }
1085
1086 static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
1087 {
1088         struct rbd_device *dev = (struct rbd_device *)data;
1089         if (!dev)
1090                 return;
1091
1092         dout("rbd_watch_cb %s notify_id=%lld opcode=%d\n", dev->obj_md_name,
1093                 notify_id, (int)opcode);
1094         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1095         __rbd_update_snaps(dev);
1096         mutex_unlock(&ctl_mutex);
1097
1098         rbd_req_sync_notify_ack(dev, ver, notify_id, dev->obj_md_name);
1099 }
1100
1101 /*
1102  * Request sync osd watch
1103  */
1104 static int rbd_req_sync_watch(struct rbd_device *dev,
1105                               const char *obj,
1106                               u64 ver)
1107 {
1108         struct ceph_osd_req_op *ops;
1109         struct ceph_osd_client *osdc = &dev->client->osdc;
1110
1111         int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_WATCH, 0);
1112         if (ret < 0)
1113                 return ret;
1114
1115         ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0,
1116                                      (void *)dev, &dev->watch_event);
1117         if (ret < 0)
1118                 goto fail;
1119
1120         ops[0].watch.ver = cpu_to_le64(ver);
1121         ops[0].watch.cookie = cpu_to_le64(dev->watch_event->cookie);
1122         ops[0].watch.flag = 1;
1123
1124         ret = rbd_req_sync_op(dev, NULL,
1125                               CEPH_NOSNAP,
1126                               0,
1127                               CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1128                               ops,
1129                               1, obj, 0, 0, NULL,
1130                               &dev->watch_request, NULL);
1131
1132         if (ret < 0)
1133                 goto fail_event;
1134
1135         rbd_destroy_ops(ops);
1136         return 0;
1137
1138 fail_event:
1139         ceph_osdc_cancel_event(dev->watch_event);
1140         dev->watch_event = NULL;
1141 fail:
1142         rbd_destroy_ops(ops);
1143         return ret;
1144 }
1145
1146 struct rbd_notify_info {
1147         struct rbd_device *dev;
1148 };
1149
1150 static void rbd_notify_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
1151 {
1152         struct rbd_device *dev = (struct rbd_device *)data;
1153         if (!dev)
1154                 return;
1155
1156         dout("rbd_notify_cb %s notify_id=%lld opcode=%d\n", dev->obj_md_name,
1157                 notify_id, (int)opcode);
1158 }
1159
1160 /*
1161  * Request sync osd notify
1162  */
1163 static int rbd_req_sync_notify(struct rbd_device *dev,
1164                           const char *obj)
1165 {
1166         struct ceph_osd_req_op *ops;
1167         struct ceph_osd_client *osdc = &dev->client->osdc;
1168         struct ceph_osd_event *event;
1169         struct rbd_notify_info info;
1170         int payload_len = sizeof(u32) + sizeof(u32);
1171         int ret;
1172
1173         ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_NOTIFY, payload_len);
1174         if (ret < 0)
1175                 return ret;
1176
1177         info.dev = dev;
1178
1179         ret = ceph_osdc_create_event(osdc, rbd_notify_cb, 1,
1180                                      (void *)&info, &event);
1181         if (ret < 0)
1182                 goto fail;
1183
1184         ops[0].watch.ver = 1;
1185         ops[0].watch.flag = 1;
1186         ops[0].watch.cookie = event->cookie;
1187         ops[0].watch.prot_ver = RADOS_NOTIFY_VER;
1188         ops[0].watch.timeout = 12;
1189
1190         ret = rbd_req_sync_op(dev, NULL,
1191                                CEPH_NOSNAP,
1192                                0,
1193                                CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1194                                ops,
1195                                1, obj, 0, 0, NULL, NULL, NULL);
1196         if (ret < 0)
1197                 goto fail_event;
1198
1199         ret = ceph_osdc_wait_event(event, CEPH_OSD_TIMEOUT_DEFAULT);
1200         dout("ceph_osdc_wait_event returned %d\n", ret);
1201         rbd_destroy_ops(ops);
1202         return 0;
1203
1204 fail_event:
1205         ceph_osdc_cancel_event(event);
1206 fail:
1207         rbd_destroy_ops(ops);
1208         return ret;
1209 }
1210
1211 /*
1212  * Request sync osd rollback
1213  */
1214 static int rbd_req_sync_rollback_obj(struct rbd_device *dev,
1215                                      u64 snapid,
1216                                      const char *obj)
1217 {
1218         struct ceph_osd_req_op *ops;
1219         int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_ROLLBACK, 0);
1220         if (ret < 0)
1221                 return ret;
1222
1223         ops[0].snap.snapid = snapid;
1224
1225         ret = rbd_req_sync_op(dev, NULL,
1226                                CEPH_NOSNAP,
1227                                0,
1228                                CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1229                                ops,
1230                                1, obj, 0, 0, NULL, NULL, NULL);
1231
1232         rbd_destroy_ops(ops);
1233
1234         return ret;
1235 }
1236
1237 /*
1238  * Request sync osd read
1239  */
1240 static int rbd_req_sync_exec(struct rbd_device *dev,
1241                              const char *obj,
1242                              const char *cls,
1243                              const char *method,
1244                              const char *data,
1245                              int len,
1246                              u64 *ver)
1247 {
1248         struct ceph_osd_req_op *ops;
1249         int cls_len = strlen(cls);
1250         int method_len = strlen(method);
1251         int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_CALL,
1252                                     cls_len + method_len + len);
1253         if (ret < 0)
1254                 return ret;
1255
1256         ops[0].cls.class_name = cls;
1257         ops[0].cls.class_len = (__u8)cls_len;
1258         ops[0].cls.method_name = method;
1259         ops[0].cls.method_len = (__u8)method_len;
1260         ops[0].cls.argc = 0;
1261         ops[0].cls.indata = data;
1262         ops[0].cls.indata_len = len;
1263
1264         ret = rbd_req_sync_op(dev, NULL,
1265                                CEPH_NOSNAP,
1266                                0,
1267                                CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1268                                ops,
1269                                1, obj, 0, 0, NULL, NULL, ver);
1270
1271         rbd_destroy_ops(ops);
1272
1273         dout("cls_exec returned %d\n", ret);
1274         return ret;
1275 }
1276
1277 /*
1278  * block device queue callback
1279  */
1280 static void rbd_rq_fn(struct request_queue *q)
1281 {
1282         struct rbd_device *rbd_dev = q->queuedata;
1283         struct request *rq;
1284         struct bio_pair *bp = NULL;
1285
1286         rq = blk_fetch_request(q);
1287
1288         while (1) {
1289                 struct bio *bio;
1290                 struct bio *rq_bio, *next_bio = NULL;
1291                 bool do_write;
1292                 int size, op_size = 0;
1293                 u64 ofs;
1294
1295                 /* peek at request from block layer */
1296                 if (!rq)
1297                         break;
1298
1299                 dout("fetched request\n");
1300
1301                 /* filter out block requests we don't understand */
1302                 if ((rq->cmd_type != REQ_TYPE_FS)) {
1303                         __blk_end_request_all(rq, 0);
1304                         goto next;
1305                 }
1306
1307                 /* deduce our operation (read, write) */
1308                 do_write = (rq_data_dir(rq) == WRITE);
1309
1310                 size = blk_rq_bytes(rq);
1311                 ofs = blk_rq_pos(rq) * 512ULL;
1312                 rq_bio = rq->bio;
1313                 if (do_write && rbd_dev->read_only) {
1314                         __blk_end_request_all(rq, -EROFS);
1315                         goto next;
1316                 }
1317
1318                 spin_unlock_irq(q->queue_lock);
1319
1320                 dout("%s 0x%x bytes at 0x%llx\n",
1321                      do_write ? "write" : "read",
1322                      size, blk_rq_pos(rq) * 512ULL);
1323
1324                 do {
1325                         /* a bio clone to be passed down to OSD req */
1326                         dout("rq->bio->bi_vcnt=%d\n", rq->bio->bi_vcnt);
1327                         op_size = rbd_get_segment(&rbd_dev->header,
1328                                                   rbd_dev->header.block_name,
1329                                                   ofs, size,
1330                                                   NULL, NULL);
1331                         bio = bio_chain_clone(&rq_bio, &next_bio, &bp,
1332                                               op_size, GFP_ATOMIC);
1333                         if (!bio) {
1334                                 spin_lock_irq(q->queue_lock);
1335                                 __blk_end_request_all(rq, -ENOMEM);
1336                                 goto next;
1337                         }
1338
1339                         /* init OSD command: write or read */
1340                         if (do_write)
1341                                 rbd_req_write(rq, rbd_dev,
1342                                               rbd_dev->header.snapc,
1343                                               ofs,
1344                                               op_size, bio);
1345                         else
1346                                 rbd_req_read(rq, rbd_dev,
1347                                              cur_snap_id(rbd_dev),
1348                                              ofs,
1349                                              op_size, bio);
1350
1351                         size -= op_size;
1352                         ofs += op_size;
1353
1354                         rq_bio = next_bio;
1355                 } while (size > 0);
1356
1357                 if (bp)
1358                         bio_pair_release(bp);
1359
1360                 spin_lock_irq(q->queue_lock);
1361 next:
1362                 rq = blk_fetch_request(q);
1363         }
1364 }
1365
1366 /*
1367  * a queue callback. Makes sure that we don't create a bio that spans across
1368  * multiple osd objects. One exception would be with a single page bios,
1369  * which we handle later at bio_chain_clone
1370  */
1371 static int rbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bmd,
1372                           struct bio_vec *bvec)
1373 {
1374         struct rbd_device *rbd_dev = q->queuedata;
1375         unsigned int chunk_sectors = 1 << (rbd_dev->header.obj_order - 9);
1376         sector_t sector = bmd->bi_sector + get_start_sect(bmd->bi_bdev);
1377         unsigned int bio_sectors = bmd->bi_size >> 9;
1378         int max;
1379
1380         max =  (chunk_sectors - ((sector & (chunk_sectors - 1))
1381                                  + bio_sectors)) << 9;
1382         if (max < 0)
1383                 max = 0; /* bio_add cannot handle a negative return */
1384         if (max <= bvec->bv_len && bio_sectors == 0)
1385                 return bvec->bv_len;
1386         return max;
1387 }
1388
1389 static void rbd_free_disk(struct rbd_device *rbd_dev)
1390 {
1391         struct gendisk *disk = rbd_dev->disk;
1392
1393         if (!disk)
1394                 return;
1395
1396         rbd_header_free(&rbd_dev->header);
1397
1398         if (disk->flags & GENHD_FL_UP)
1399                 del_gendisk(disk);
1400         if (disk->queue)
1401                 blk_cleanup_queue(disk->queue);
1402         put_disk(disk);
1403 }
1404
1405 /*
1406  * reload the ondisk the header
1407  */
1408 static int rbd_read_header(struct rbd_device *rbd_dev,
1409                            struct rbd_image_header *header)
1410 {
1411         ssize_t rc;
1412         struct rbd_image_header_ondisk *dh;
1413         int snap_count = 0;
1414         u64 snap_names_len = 0;
1415         u64 ver;
1416
1417         while (1) {
1418                 int len = sizeof(*dh) +
1419                           snap_count * sizeof(struct rbd_image_snap_ondisk) +
1420                           snap_names_len;
1421
1422                 rc = -ENOMEM;
1423                 dh = kmalloc(len, GFP_KERNEL);
1424                 if (!dh)
1425                         return -ENOMEM;
1426
1427                 rc = rbd_req_sync_read(rbd_dev,
1428                                        NULL, CEPH_NOSNAP,
1429                                        rbd_dev->obj_md_name,
1430                                        0, len,
1431                                        (char *)dh, &ver);
1432                 if (rc < 0)
1433                         goto out_dh;
1434
1435                 rc = rbd_header_from_disk(header, dh, snap_count, GFP_KERNEL);
1436                 if (rc < 0)
1437                         goto out_dh;
1438
1439                 if (snap_count != header->total_snaps) {
1440                         snap_count = header->total_snaps;
1441                         snap_names_len = header->snap_names_len;
1442                         rbd_header_free(header);
1443                         kfree(dh);
1444                         continue;
1445                 }
1446                 break;
1447         }
1448         header->obj_version = ver;
1449
1450 out_dh:
1451         kfree(dh);
1452         return rc;
1453 }
1454
1455 /*
1456  * create a snapshot
1457  */
1458 static int rbd_header_add_snap(struct rbd_device *dev,
1459                                const char *snap_name,
1460                                gfp_t gfp_flags)
1461 {
1462         int name_len = strlen(snap_name);
1463         u64 new_snapid;
1464         int ret;
1465         void *data, *data_start, *data_end;
1466         u64 ver;
1467
1468         /* we should create a snapshot only if we're pointing at the head */
1469         if (dev->cur_snap)
1470                 return -EINVAL;
1471
1472         ret = ceph_monc_create_snapid(&dev->client->monc, dev->poolid,
1473                                       &new_snapid);
1474         dout("created snapid=%lld\n", new_snapid);
1475         if (ret < 0)
1476                 return ret;
1477
1478         data = kmalloc(name_len + 16, gfp_flags);
1479         if (!data)
1480                 return -ENOMEM;
1481
1482         data_start = data;
1483         data_end = data + name_len + 16;
1484
1485         ceph_encode_string_safe(&data, data_end, snap_name, name_len, bad);
1486         ceph_encode_64_safe(&data, data_end, new_snapid, bad);
1487
1488         ret = rbd_req_sync_exec(dev, dev->obj_md_name, "rbd", "snap_add",
1489                                 data_start, data - data_start, &ver);
1490
1491         kfree(data_start);
1492
1493         if (ret < 0)
1494                 return ret;
1495
1496         dev->header.snapc->seq =  new_snapid;
1497
1498         return 0;
1499 bad:
1500         return -ERANGE;
1501 }
1502
1503 static void __rbd_remove_all_snaps(struct rbd_device *rbd_dev)
1504 {
1505         struct rbd_snap *snap;
1506
1507         while (!list_empty(&rbd_dev->snaps)) {
1508                 snap = list_first_entry(&rbd_dev->snaps, struct rbd_snap, node);
1509                 __rbd_remove_snap_dev(rbd_dev, snap);
1510         }
1511 }
1512
1513 /*
1514  * only read the first part of the ondisk header, without the snaps info
1515  */
1516 static int __rbd_update_snaps(struct rbd_device *rbd_dev)
1517 {
1518         int ret;
1519         struct rbd_image_header h;
1520         u64 snap_seq;
1521         int follow_seq = 0;
1522
1523         ret = rbd_read_header(rbd_dev, &h);
1524         if (ret < 0)
1525                 return ret;
1526
1527         down_write(&rbd_dev->header.snap_rwsem);
1528
1529         snap_seq = rbd_dev->header.snapc->seq;
1530         if (rbd_dev->header.total_snaps &&
1531             rbd_dev->header.snapc->snaps[0] == snap_seq)
1532                 /* pointing at the head, will need to follow that
1533                    if head moves */
1534                 follow_seq = 1;
1535
1536         kfree(rbd_dev->header.snapc);
1537         kfree(rbd_dev->header.snap_names);
1538         kfree(rbd_dev->header.snap_sizes);
1539
1540         rbd_dev->header.total_snaps = h.total_snaps;
1541         rbd_dev->header.snapc = h.snapc;
1542         rbd_dev->header.snap_names = h.snap_names;
1543         rbd_dev->header.snap_names_len = h.snap_names_len;
1544         rbd_dev->header.snap_sizes = h.snap_sizes;
1545         if (follow_seq)
1546                 rbd_dev->header.snapc->seq = rbd_dev->header.snapc->snaps[0];
1547         else
1548                 rbd_dev->header.snapc->seq = snap_seq;
1549
1550         ret = __rbd_init_snaps_header(rbd_dev);
1551
1552         up_write(&rbd_dev->header.snap_rwsem);
1553
1554         return ret;
1555 }
1556
1557 static int rbd_init_disk(struct rbd_device *rbd_dev)
1558 {
1559         struct gendisk *disk;
1560         struct request_queue *q;
1561         int rc;
1562         u64 total_size = 0;
1563
1564         /* contact OSD, request size info about the object being mapped */
1565         rc = rbd_read_header(rbd_dev, &rbd_dev->header);
1566         if (rc)
1567                 return rc;
1568
1569         /* no need to lock here, as rbd_dev is not registered yet */
1570         rc = __rbd_init_snaps_header(rbd_dev);
1571         if (rc)
1572                 return rc;
1573
1574         rc = rbd_header_set_snap(rbd_dev, rbd_dev->snap_name, &total_size);
1575         if (rc)
1576                 return rc;
1577
1578         /* create gendisk info */
1579         rc = -ENOMEM;
1580         disk = alloc_disk(RBD_MINORS_PER_MAJOR);
1581         if (!disk)
1582                 goto out;
1583
1584         sprintf(disk->disk_name, DRV_NAME "%d", rbd_dev->id);
1585         disk->major = rbd_dev->major;
1586         disk->first_minor = 0;
1587         disk->fops = &rbd_bd_ops;
1588         disk->private_data = rbd_dev;
1589
1590         /* init rq */
1591         rc = -ENOMEM;
1592         q = blk_init_queue(rbd_rq_fn, &rbd_dev->lock);
1593         if (!q)
1594                 goto out_disk;
1595         blk_queue_merge_bvec(q, rbd_merge_bvec);
1596         disk->queue = q;
1597
1598         q->queuedata = rbd_dev;
1599
1600         rbd_dev->disk = disk;
1601         rbd_dev->q = q;
1602
1603         /* finally, announce the disk to the world */
1604         set_capacity(disk, total_size / 512ULL);
1605         add_disk(disk);
1606
1607         pr_info("%s: added with size 0x%llx\n",
1608                 disk->disk_name, (unsigned long long)total_size);
1609         return 0;
1610
1611 out_disk:
1612         put_disk(disk);
1613 out:
1614         return rc;
1615 }
1616
1617 /*
1618   sysfs
1619 */
1620
1621 static ssize_t rbd_size_show(struct device *dev,
1622                              struct device_attribute *attr, char *buf)
1623 {
1624         struct rbd_device *rbd_dev = dev_to_rbd(dev);
1625
1626         return sprintf(buf, "%llu\n", (unsigned long long)rbd_dev->header.image_size);
1627 }
1628
1629 static ssize_t rbd_major_show(struct device *dev,
1630                               struct device_attribute *attr, char *buf)
1631 {
1632         struct rbd_device *rbd_dev = dev_to_rbd(dev);
1633
1634         return sprintf(buf, "%d\n", rbd_dev->major);
1635 }
1636
1637 static ssize_t rbd_client_id_show(struct device *dev,
1638                                   struct device_attribute *attr, char *buf)
1639 {
1640         struct rbd_device *rbd_dev = dev_to_rbd(dev);
1641
1642         return sprintf(buf, "client%lld\n", ceph_client_id(rbd_dev->client));
1643 }
1644
1645 static ssize_t rbd_pool_show(struct device *dev,
1646                              struct device_attribute *attr, char *buf)
1647 {
1648         struct rbd_device *rbd_dev = dev_to_rbd(dev);
1649
1650         return sprintf(buf, "%s\n", rbd_dev->pool_name);
1651 }
1652
1653 static ssize_t rbd_name_show(struct device *dev,
1654                              struct device_attribute *attr, char *buf)
1655 {
1656         struct rbd_device *rbd_dev = dev_to_rbd(dev);
1657
1658         return sprintf(buf, "%s\n", rbd_dev->obj);
1659 }
1660
1661 static ssize_t rbd_snap_show(struct device *dev,
1662                              struct device_attribute *attr,
1663                              char *buf)
1664 {
1665         struct rbd_device *rbd_dev = dev_to_rbd(dev);
1666
1667         return sprintf(buf, "%s\n", rbd_dev->snap_name);
1668 }
1669
1670 static ssize_t rbd_image_refresh(struct device *dev,
1671                                  struct device_attribute *attr,
1672                                  const char *buf,
1673                                  size_t size)
1674 {
1675         struct rbd_device *rbd_dev = dev_to_rbd(dev);
1676         int rc;
1677         int ret = size;
1678
1679         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1680
1681         rc = __rbd_update_snaps(rbd_dev);
1682         if (rc < 0)
1683                 ret = rc;
1684
1685         mutex_unlock(&ctl_mutex);
1686         return ret;
1687 }
1688
1689 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
1690 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
1691 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
1692 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
1693 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
1694 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
1695 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
1696 static DEVICE_ATTR(create_snap, S_IWUSR, NULL, rbd_snap_add);
1697 static DEVICE_ATTR(rollback_snap, S_IWUSR, NULL, rbd_snap_rollback);
1698
1699 static struct attribute *rbd_attrs[] = {
1700         &dev_attr_size.attr,
1701         &dev_attr_major.attr,
1702         &dev_attr_client_id.attr,
1703         &dev_attr_pool.attr,
1704         &dev_attr_name.attr,
1705         &dev_attr_current_snap.attr,
1706         &dev_attr_refresh.attr,
1707         &dev_attr_create_snap.attr,
1708         &dev_attr_rollback_snap.attr,
1709         NULL
1710 };
1711
1712 static struct attribute_group rbd_attr_group = {
1713         .attrs = rbd_attrs,
1714 };
1715
1716 static const struct attribute_group *rbd_attr_groups[] = {
1717         &rbd_attr_group,
1718         NULL
1719 };
1720
1721 static void rbd_sysfs_dev_release(struct device *dev)
1722 {
1723 }
1724
1725 static struct device_type rbd_device_type = {
1726         .name           = "rbd",
1727         .groups         = rbd_attr_groups,
1728         .release        = rbd_sysfs_dev_release,
1729 };
1730
1731
1732 /*
1733   sysfs - snapshots
1734 */
1735
1736 static ssize_t rbd_snap_size_show(struct device *dev,
1737                                   struct device_attribute *attr,
1738                                   char *buf)
1739 {
1740         struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1741
1742         return sprintf(buf, "%lld\n", (long long)snap->size);
1743 }
1744
1745 static ssize_t rbd_snap_id_show(struct device *dev,
1746                                 struct device_attribute *attr,
1747                                 char *buf)
1748 {
1749         struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1750
1751         return sprintf(buf, "%lld\n", (long long)snap->id);
1752 }
1753
1754 static DEVICE_ATTR(snap_size, S_IRUGO, rbd_snap_size_show, NULL);
1755 static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
1756
1757 static struct attribute *rbd_snap_attrs[] = {
1758         &dev_attr_snap_size.attr,
1759         &dev_attr_snap_id.attr,
1760         NULL,
1761 };
1762
1763 static struct attribute_group rbd_snap_attr_group = {
1764         .attrs = rbd_snap_attrs,
1765 };
1766
1767 static void rbd_snap_dev_release(struct device *dev)
1768 {
1769         struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1770         kfree(snap->name);
1771         kfree(snap);
1772 }
1773
1774 static const struct attribute_group *rbd_snap_attr_groups[] = {
1775         &rbd_snap_attr_group,
1776         NULL
1777 };
1778
1779 static struct device_type rbd_snap_device_type = {
1780         .groups         = rbd_snap_attr_groups,
1781         .release        = rbd_snap_dev_release,
1782 };
1783
1784 static void __rbd_remove_snap_dev(struct rbd_device *rbd_dev,
1785                                   struct rbd_snap *snap)
1786 {
1787         list_del(&snap->node);
1788         device_unregister(&snap->dev);
1789 }
1790
1791 static int rbd_register_snap_dev(struct rbd_device *rbd_dev,
1792                                   struct rbd_snap *snap,
1793                                   struct device *parent)
1794 {
1795         struct device *dev = &snap->dev;
1796         int ret;
1797
1798         dev->type = &rbd_snap_device_type;
1799         dev->parent = parent;
1800         dev->release = rbd_snap_dev_release;
1801         dev_set_name(dev, "snap_%s", snap->name);
1802         ret = device_register(dev);
1803
1804         return ret;
1805 }
1806
1807 static int __rbd_add_snap_dev(struct rbd_device *rbd_dev,
1808                               int i, const char *name,
1809                               struct rbd_snap **snapp)
1810 {
1811         int ret;
1812         struct rbd_snap *snap = kzalloc(sizeof(*snap), GFP_KERNEL);
1813         if (!snap)
1814                 return -ENOMEM;
1815         snap->name = kstrdup(name, GFP_KERNEL);
1816         snap->size = rbd_dev->header.snap_sizes[i];
1817         snap->id = rbd_dev->header.snapc->snaps[i];
1818         if (device_is_registered(&rbd_dev->dev)) {
1819                 ret = rbd_register_snap_dev(rbd_dev, snap,
1820                                              &rbd_dev->dev);
1821                 if (ret < 0)
1822                         goto err;
1823         }
1824         *snapp = snap;
1825         return 0;
1826 err:
1827         kfree(snap->name);
1828         kfree(snap);
1829         return ret;
1830 }
1831
1832 /*
1833  * search for the previous snap in a null delimited string list
1834  */
1835 const char *rbd_prev_snap_name(const char *name, const char *start)
1836 {
1837         if (name < start + 2)
1838                 return NULL;
1839
1840         name -= 2;
1841         while (*name) {
1842                 if (name == start)
1843                         return start;
1844                 name--;
1845         }
1846         return name + 1;
1847 }
1848
1849 /*
1850  * compare the old list of snapshots that we have to what's in the header
1851  * and update it accordingly. Note that the header holds the snapshots
1852  * in a reverse order (from newest to oldest) and we need to go from
1853  * older to new so that we don't get a duplicate snap name when
1854  * doing the process (e.g., removed snapshot and recreated a new
1855  * one with the same name.
1856  */
1857 static int __rbd_init_snaps_header(struct rbd_device *rbd_dev)
1858 {
1859         const char *name, *first_name;
1860         int i = rbd_dev->header.total_snaps;
1861         struct rbd_snap *snap, *old_snap = NULL;
1862         int ret;
1863         struct list_head *p, *n;
1864
1865         first_name = rbd_dev->header.snap_names;
1866         name = first_name + rbd_dev->header.snap_names_len;
1867
1868         list_for_each_prev_safe(p, n, &rbd_dev->snaps) {
1869                 u64 cur_id;
1870
1871                 old_snap = list_entry(p, struct rbd_snap, node);
1872
1873                 if (i)
1874                         cur_id = rbd_dev->header.snapc->snaps[i - 1];
1875
1876                 if (!i || old_snap->id < cur_id) {
1877                         /* old_snap->id was skipped, thus was removed */
1878                         __rbd_remove_snap_dev(rbd_dev, old_snap);
1879                         continue;
1880                 }
1881                 if (old_snap->id == cur_id) {
1882                         /* we have this snapshot already */
1883                         i--;
1884                         name = rbd_prev_snap_name(name, first_name);
1885                         continue;
1886                 }
1887                 for (; i > 0;
1888                      i--, name = rbd_prev_snap_name(name, first_name)) {
1889                         if (!name) {
1890                                 WARN_ON(1);
1891                                 return -EINVAL;
1892                         }
1893                         cur_id = rbd_dev->header.snapc->snaps[i];
1894                         /* snapshot removal? handle it above */
1895                         if (cur_id >= old_snap->id)
1896                                 break;
1897                         /* a new snapshot */
1898                         ret = __rbd_add_snap_dev(rbd_dev, i - 1, name, &snap);
1899                         if (ret < 0)
1900                                 return ret;
1901
1902                         /* note that we add it backward so using n and not p */
1903                         list_add(&snap->node, n);
1904                         p = &snap->node;
1905                 }
1906         }
1907         /* we're done going over the old snap list, just add what's left */
1908         for (; i > 0; i--) {
1909                 name = rbd_prev_snap_name(name, first_name);
1910                 if (!name) {
1911                         WARN_ON(1);
1912                         return -EINVAL;
1913                 }
1914                 ret = __rbd_add_snap_dev(rbd_dev, i - 1, name, &snap);
1915                 if (ret < 0)
1916                         return ret;
1917                 list_add(&snap->node, &rbd_dev->snaps);
1918         }
1919
1920         return 0;
1921 }
1922
1923
1924 static void rbd_root_dev_release(struct device *dev)
1925 {
1926 }
1927
1928 static struct device rbd_root_dev = {
1929         .init_name =    "rbd",
1930         .release =      rbd_root_dev_release,
1931 };
1932
1933 static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
1934 {
1935         int ret = -ENOMEM;
1936         struct device *dev;
1937         struct rbd_snap *snap;
1938
1939         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1940         dev = &rbd_dev->dev;
1941
1942         dev->bus = &rbd_bus_type;
1943         dev->type = &rbd_device_type;
1944         dev->parent = &rbd_root_dev;
1945         dev->release = rbd_dev_release;
1946         dev_set_name(dev, "%d", rbd_dev->id);
1947         ret = device_register(dev);
1948         if (ret < 0)
1949                 goto done_free;
1950
1951         list_for_each_entry(snap, &rbd_dev->snaps, node) {
1952                 ret = rbd_register_snap_dev(rbd_dev, snap,
1953                                              &rbd_dev->dev);
1954                 if (ret < 0)
1955                         break;
1956         }
1957
1958         mutex_unlock(&ctl_mutex);
1959         return 0;
1960 done_free:
1961         mutex_unlock(&ctl_mutex);
1962         return ret;
1963 }
1964
1965 static void rbd_bus_del_dev(struct rbd_device *rbd_dev)
1966 {
1967         device_unregister(&rbd_dev->dev);
1968 }
1969
1970 static int rbd_init_watch_dev(struct rbd_device *rbd_dev)
1971 {
1972         int ret, rc;
1973
1974         do {
1975                 ret = rbd_req_sync_watch(rbd_dev, rbd_dev->obj_md_name,
1976                                          rbd_dev->header.obj_version);
1977                 if (ret == -ERANGE) {
1978                         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1979                         rc = __rbd_update_snaps(rbd_dev);
1980                         mutex_unlock(&ctl_mutex);
1981                         if (rc < 0)
1982                                 return rc;
1983                 }
1984         } while (ret == -ERANGE);
1985
1986         return ret;
1987 }
1988
1989 static ssize_t rbd_add(struct bus_type *bus,
1990                        const char *buf,
1991                        size_t count)
1992 {
1993         struct ceph_osd_client *osdc;
1994         struct rbd_device *rbd_dev;
1995         ssize_t rc = -ENOMEM;
1996         int irc, new_id = 0;
1997         struct list_head *tmp;
1998         char *mon_dev_name;
1999         char *options;
2000
2001         if (!try_module_get(THIS_MODULE))
2002                 return -ENODEV;
2003
2004         mon_dev_name = kmalloc(RBD_MAX_OPT_LEN, GFP_KERNEL);
2005         if (!mon_dev_name)
2006                 goto err_out_mod;
2007
2008         options = kmalloc(RBD_MAX_OPT_LEN, GFP_KERNEL);
2009         if (!options)
2010                 goto err_mon_dev;
2011
2012         /* new rbd_device object */
2013         rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
2014         if (!rbd_dev)
2015                 goto err_out_opt;
2016
2017         /* static rbd_device initialization */
2018         spin_lock_init(&rbd_dev->lock);
2019         INIT_LIST_HEAD(&rbd_dev->node);
2020         INIT_LIST_HEAD(&rbd_dev->snaps);
2021
2022         /* generate unique id: find highest unique id, add one */
2023         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2024
2025         list_for_each(tmp, &rbd_dev_list) {
2026                 struct rbd_device *rbd_dev;
2027
2028                 rbd_dev = list_entry(tmp, struct rbd_device, node);
2029                 if (rbd_dev->id >= new_id)
2030                         new_id = rbd_dev->id + 1;
2031         }
2032
2033         rbd_dev->id = new_id;
2034
2035         /* add to global list */
2036         list_add_tail(&rbd_dev->node, &rbd_dev_list);
2037
2038         /* parse add command */
2039         if (sscanf(buf, "%" __stringify(RBD_MAX_OPT_LEN) "s "
2040                    "%" __stringify(RBD_MAX_OPT_LEN) "s "
2041                    "%" __stringify(RBD_MAX_POOL_NAME_LEN) "s "
2042                    "%" __stringify(RBD_MAX_OBJ_NAME_LEN) "s"
2043                    "%" __stringify(RBD_MAX_SNAP_NAME_LEN) "s",
2044                    mon_dev_name, options, rbd_dev->pool_name,
2045                    rbd_dev->obj, rbd_dev->snap_name) < 4) {
2046                 rc = -EINVAL;
2047                 goto err_out_slot;
2048         }
2049
2050         if (rbd_dev->snap_name[0] == 0)
2051                 rbd_dev->snap_name[0] = '-';
2052
2053         rbd_dev->obj_len = strlen(rbd_dev->obj);
2054         snprintf(rbd_dev->obj_md_name, sizeof(rbd_dev->obj_md_name), "%s%s",
2055                  rbd_dev->obj, RBD_SUFFIX);
2056
2057         /* initialize rest of new object */
2058         snprintf(rbd_dev->name, DEV_NAME_LEN, DRV_NAME "%d", rbd_dev->id);
2059         rc = rbd_get_client(rbd_dev, mon_dev_name, options);
2060         if (rc < 0)
2061                 goto err_out_slot;
2062
2063         mutex_unlock(&ctl_mutex);
2064
2065         /* pick the pool */
2066         osdc = &rbd_dev->client->osdc;
2067         rc = ceph_pg_poolid_by_name(osdc->osdmap, rbd_dev->pool_name);
2068         if (rc < 0)
2069                 goto err_out_client;
2070         rbd_dev->poolid = rc;
2071
2072         /* register our block device */
2073         irc = register_blkdev(0, rbd_dev->name);
2074         if (irc < 0) {
2075                 rc = irc;
2076                 goto err_out_client;
2077         }
2078         rbd_dev->major = irc;
2079
2080         rc = rbd_bus_add_dev(rbd_dev);
2081         if (rc)
2082                 goto err_out_blkdev;
2083
2084         /* set up and announce blkdev mapping */
2085         rc = rbd_init_disk(rbd_dev);
2086         if (rc)
2087                 goto err_out_bus;
2088
2089         rc = rbd_init_watch_dev(rbd_dev);
2090         if (rc)
2091                 goto err_out_bus;
2092
2093         return count;
2094
2095 err_out_bus:
2096         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2097         list_del_init(&rbd_dev->node);
2098         mutex_unlock(&ctl_mutex);
2099
2100         /* this will also clean up rest of rbd_dev stuff */
2101
2102         rbd_bus_del_dev(rbd_dev);
2103         kfree(options);
2104         kfree(mon_dev_name);
2105         return rc;
2106
2107 err_out_blkdev:
2108         unregister_blkdev(rbd_dev->major, rbd_dev->name);
2109 err_out_client:
2110         rbd_put_client(rbd_dev);
2111         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2112 err_out_slot:
2113         list_del_init(&rbd_dev->node);
2114         mutex_unlock(&ctl_mutex);
2115
2116         kfree(rbd_dev);
2117 err_out_opt:
2118         kfree(options);
2119 err_mon_dev:
2120         kfree(mon_dev_name);
2121 err_out_mod:
2122         dout("Error adding device %s\n", buf);
2123         module_put(THIS_MODULE);
2124         return rc;
2125 }
2126
2127 static struct rbd_device *__rbd_get_dev(unsigned long id)
2128 {
2129         struct list_head *tmp;
2130         struct rbd_device *rbd_dev;
2131
2132         list_for_each(tmp, &rbd_dev_list) {
2133                 rbd_dev = list_entry(tmp, struct rbd_device, node);
2134                 if (rbd_dev->id == id)
2135                         return rbd_dev;
2136         }
2137         return NULL;
2138 }
2139
2140 static void rbd_dev_release(struct device *dev)
2141 {
2142         struct rbd_device *rbd_dev =
2143                         container_of(dev, struct rbd_device, dev);
2144
2145         if (rbd_dev->watch_request)
2146                 ceph_osdc_unregister_linger_request(&rbd_dev->client->osdc,
2147                                                     rbd_dev->watch_request);
2148         if (rbd_dev->watch_event)
2149                 ceph_osdc_cancel_event(rbd_dev->watch_event);
2150
2151         rbd_put_client(rbd_dev);
2152
2153         /* clean up and free blkdev */
2154         rbd_free_disk(rbd_dev);
2155         unregister_blkdev(rbd_dev->major, rbd_dev->name);
2156         kfree(rbd_dev);
2157
2158         /* release module ref */
2159         module_put(THIS_MODULE);
2160 }
2161
2162 static ssize_t rbd_remove(struct bus_type *bus,
2163                           const char *buf,
2164                           size_t count)
2165 {
2166         struct rbd_device *rbd_dev = NULL;
2167         int target_id, rc;
2168         unsigned long ul;
2169         int ret = count;
2170
2171         rc = strict_strtoul(buf, 10, &ul);
2172         if (rc)
2173                 return rc;
2174
2175         /* convert to int; abort if we lost anything in the conversion */
2176         target_id = (int) ul;
2177         if (target_id != ul)
2178                 return -EINVAL;
2179
2180         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2181
2182         rbd_dev = __rbd_get_dev(target_id);
2183         if (!rbd_dev) {
2184                 ret = -ENOENT;
2185                 goto done;
2186         }
2187
2188         list_del_init(&rbd_dev->node);
2189
2190         __rbd_remove_all_snaps(rbd_dev);
2191         rbd_bus_del_dev(rbd_dev);
2192
2193 done:
2194         mutex_unlock(&ctl_mutex);
2195         return ret;
2196 }
2197
2198 static ssize_t rbd_snap_add(struct device *dev,
2199                             struct device_attribute *attr,
2200                             const char *buf,
2201                             size_t count)
2202 {
2203         struct rbd_device *rbd_dev = dev_to_rbd(dev);
2204         int ret;
2205         char *name = kmalloc(count + 1, GFP_KERNEL);
2206         if (!name)
2207                 return -ENOMEM;
2208
2209         snprintf(name, count, "%s", buf);
2210
2211         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2212
2213         ret = rbd_header_add_snap(rbd_dev,
2214                                   name, GFP_KERNEL);
2215         if (ret < 0)
2216                 goto err_unlock;
2217
2218         ret = __rbd_update_snaps(rbd_dev);
2219         if (ret < 0)
2220                 goto err_unlock;
2221
2222         /* shouldn't hold ctl_mutex when notifying.. notify might
2223            trigger a watch callback that would need to get that mutex */
2224         mutex_unlock(&ctl_mutex);
2225
2226         /* make a best effort, don't error if failed */
2227         rbd_req_sync_notify(rbd_dev, rbd_dev->obj_md_name);
2228
2229         ret = count;
2230         kfree(name);
2231         return ret;
2232
2233 err_unlock:
2234         mutex_unlock(&ctl_mutex);
2235         kfree(name);
2236         return ret;
2237 }
2238
2239 static ssize_t rbd_snap_rollback(struct device *dev,
2240                                  struct device_attribute *attr,
2241                                  const char *buf,
2242                                  size_t count)
2243 {
2244         struct rbd_device *rbd_dev = dev_to_rbd(dev);
2245         int ret;
2246         u64 snapid;
2247         u64 cur_ofs;
2248         char *seg_name = NULL;
2249         char *snap_name = kmalloc(count + 1, GFP_KERNEL);
2250         ret = -ENOMEM;
2251         if (!snap_name)
2252                 return ret;
2253
2254         /* parse snaps add command */
2255         snprintf(snap_name, count, "%s", buf);
2256         seg_name = kmalloc(RBD_MAX_SEG_NAME_LEN + 1, GFP_NOIO);
2257         if (!seg_name)
2258                 goto done;
2259
2260         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2261
2262         ret = snap_by_name(&rbd_dev->header, snap_name, &snapid, NULL);
2263         if (ret < 0)
2264                 goto done_unlock;
2265
2266         dout("snapid=%lld\n", snapid);
2267
2268         cur_ofs = 0;
2269         while (cur_ofs < rbd_dev->header.image_size) {
2270                 cur_ofs += rbd_get_segment(&rbd_dev->header,
2271                                            rbd_dev->obj,
2272                                            cur_ofs, (u64)-1,
2273                                            seg_name, NULL);
2274                 dout("seg_name=%s\n", seg_name);
2275
2276                 ret = rbd_req_sync_rollback_obj(rbd_dev, snapid, seg_name);
2277                 if (ret < 0)
2278                         pr_warning("could not roll back obj %s err=%d\n",
2279                                    seg_name, ret);
2280         }
2281
2282         ret = __rbd_update_snaps(rbd_dev);
2283         if (ret < 0)
2284                 goto done_unlock;
2285
2286         ret = count;
2287
2288 done_unlock:
2289         mutex_unlock(&ctl_mutex);
2290 done:
2291         kfree(seg_name);
2292         kfree(snap_name);
2293
2294         return ret;
2295 }
2296
2297 static struct bus_attribute rbd_bus_attrs[] = {
2298         __ATTR(add, S_IWUSR, NULL, rbd_add),
2299         __ATTR(remove, S_IWUSR, NULL, rbd_remove),
2300         __ATTR_NULL
2301 };
2302
2303 /*
2304  * create control files in sysfs
2305  * /sys/bus/rbd/...
2306  */
2307 static int rbd_sysfs_init(void)
2308 {
2309         int ret;
2310
2311         rbd_bus_type.bus_attrs = rbd_bus_attrs;
2312
2313         ret = bus_register(&rbd_bus_type);
2314          if (ret < 0)
2315                 return ret;
2316
2317         ret = device_register(&rbd_root_dev);
2318
2319         return ret;
2320 }
2321
2322 static void rbd_sysfs_cleanup(void)
2323 {
2324         device_unregister(&rbd_root_dev);
2325         bus_unregister(&rbd_bus_type);
2326 }
2327
2328 int __init rbd_init(void)
2329 {
2330         int rc;
2331
2332         rc = rbd_sysfs_init();
2333         if (rc)
2334                 return rc;
2335         spin_lock_init(&node_lock);
2336         pr_info("loaded " DRV_NAME_LONG "\n");
2337         return 0;
2338 }
2339
2340 void __exit rbd_exit(void)
2341 {
2342         rbd_sysfs_cleanup();
2343 }
2344
2345 module_init(rbd_init);
2346 module_exit(rbd_exit);
2347
2348 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
2349 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
2350 MODULE_DESCRIPTION("rados block device");
2351
2352 /* following authorship retained from original osdblk.c */
2353 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
2354
2355 MODULE_LICENSE("GPL");