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