drivers/block/rbd.c

   1
   2 /*
   3    rbd.c -- Export ceph rados objects as a Linux block device
   4
   5
   6    based on drivers/block/osdblk.c:
   7
   8    Copyright 2009 Red Hat, Inc.
   9
  10    This program is free software; you can redistribute it and/or modify
  11    it under the terms of the GNU General Public License as published by
  12    the Free Software Foundation.
  13
  14    This program is distributed in the hope that it will be useful,
  15    but WITHOUT ANY WARRANTY; without even the implied warranty of
  16    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17    GNU General Public License for more details.
  18
  19    You should have received a copy of the GNU General Public License
  20    along with this program; see the file COPYING.  If not, write to
  21    the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
  22
  23
  24
  25    For usage instructions, please refer to:
  26
  27                  Documentation/ABI/testing/sysfs-bus-rbd
  28
  29  */
  30
  31 #include <linux/ceph/libceph.h>
  32 #include <linux/ceph/osd_client.h>
  33 #include <linux/ceph/mon_client.h>
  34 #include <linux/ceph/decode.h>
  35 #include <linux/parser.h>
  36 #include <linux/bsearch.h>
  37
  38 #include <linux/kernel.h>
  39 #include <linux/device.h>
  40 #include <linux/module.h>
  41 #include <linux/blk-mq.h>
  42 #include <linux/fs.h>
  43 #include <linux/blkdev.h>
  44 #include <linux/slab.h>
  45 #include <linux/idr.h>
  46 #include <linux/workqueue.h>
  47
  48 #include "rbd_types.h"
  49
  50 #define RBD_DEBUG       /* Activate rbd_assert() calls */
  51
  52 /*
  53  * The basic unit of block I/O is a sector.  It is interpreted in a
  54  * number of contexts in Linux (blk, bio, genhd), but the default is
  55  * universally 512 bytes.  These symbols are just slightly more
  56  * meaningful than the bare numbers they represent.
  57  */
  58 #define SECTOR_SHIFT    9
  59 #define SECTOR_SIZE     (1ULL << SECTOR_SHIFT)
  60
  61 /*
  62  * Increment the given counter and return its updated value.
  63  * If the counter is already 0 it will not be incremented.
  64  * If the counter is already at its maximum value returns
  65  * -EINVAL without updating it.
  66  */
  67 static int atomic_inc_return_safe(atomic_t *v)
  68 {
  69         unsigned int counter;
  70
  71         counter = (unsigned int)__atomic_add_unless(v, 1, 0);
  72         if (counter <= (unsigned int)INT_MAX)
  73                 return (int)counter;
  74
  75         atomic_dec(v);
  76
  77         return -EINVAL;
  78 }
  79
  80 /* Decrement the counter.  Return the resulting value, or -EINVAL */
  81 static int atomic_dec_return_safe(atomic_t *v)
  82 {
  83         int counter;
  84
  85         counter = atomic_dec_return(v);
  86         if (counter >= 0)
  87                 return counter;
  88
  89         atomic_inc(v);
  90
  91         return -EINVAL;
  92 }
  93
  94 #define RBD_DRV_NAME "rbd"
  95
  96 #define RBD_MINORS_PER_MAJOR            256
  97 #define RBD_SINGLE_MAJOR_PART_SHIFT     4
  98
  99 #define RBD_MAX_PARENT_CHAIN_LEN        16
 100
 101 #define RBD_SNAP_DEV_NAME_PREFIX        "snap_"
 102 #define RBD_MAX_SNAP_NAME_LEN   \
 103                         (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
 104
 105 #define RBD_MAX_SNAP_COUNT      510     /* allows max snapc to fit in 4KB */
 106
 107 #define RBD_SNAP_HEAD_NAME      "-"
 108
 109 #define BAD_SNAP_INDEX  U32_MAX         /* invalid index into snap array */
 110
 111 /* This allows a single page to hold an image name sent by OSD */
 112 #define RBD_IMAGE_NAME_LEN_MAX  (PAGE_SIZE - sizeof (__le32) - 1)
 113 #define RBD_IMAGE_ID_LEN_MAX    64
 114
 115 #define RBD_OBJ_PREFIX_LEN_MAX  64
 116
 117 /* Feature bits */
 118
 119 #define RBD_FEATURE_LAYERING    (1<<0)
 120 #define RBD_FEATURE_STRIPINGV2  (1<<1)
 121 #define RBD_FEATURES_ALL \
 122             (RBD_FEATURE_LAYERING | RBD_FEATURE_STRIPINGV2)
 123
 124 /* Features supported by this (client software) implementation. */
 125
 126 #define RBD_FEATURES_SUPPORTED  (RBD_FEATURES_ALL)
 127
 128 /*
 129  * An RBD device name will be "rbd#", where the "rbd" comes from
 130  * RBD_DRV_NAME above, and # is a unique integer identifier.
 131  * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
 132  * enough to hold all possible device names.
 133  */
 134 #define DEV_NAME_LEN            32
 135 #define MAX_INT_FORMAT_WIDTH    ((5 * sizeof (int)) / 2 + 1)
 136
 137 /*
 138  * block device image metadata (in-memory version)
 139  */
 140 struct rbd_image_header {
 141         /* These six fields never change for a given rbd image */
 142         char *object_prefix;
 143         __u8 obj_order;
 144         __u8 crypt_type;
 145         __u8 comp_type;
 146         u64 stripe_unit;
 147         u64 stripe_count;
 148         u64 features;           /* Might be changeable someday? */
 149
 150         /* The remaining fields need to be updated occasionally */
 151         u64 image_size;
 152         struct ceph_snap_context *snapc;
 153         char *snap_names;       /* format 1 only */
 154         u64 *snap_sizes;        /* format 1 only */
 155 };
 156
 157 /*
 158  * An rbd image specification.
 159  *
 160  * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
 161  * identify an image.  Each rbd_dev structure includes a pointer to
 162  * an rbd_spec structure that encapsulates this identity.
 163  *
 164  * Each of the id's in an rbd_spec has an associated name.  For a
 165  * user-mapped image, the names are supplied and the id's associated
 166  * with them are looked up.  For a layered image, a parent image is
 167  * defined by the tuple, and the names are looked up.
 168  *
 169  * An rbd_dev structure contains a parent_spec pointer which is
 170  * non-null if the image it represents is a child in a layered
 171  * image.  This pointer will refer to the rbd_spec structure used
 172  * by the parent rbd_dev for its own identity (i.e., the structure
 173  * is shared between the parent and child).
 174  *
 175  * Since these structures are populated once, during the discovery
 176  * phase of image construction, they are effectively immutable so
 177  * we make no effort to synchronize access to them.
 178  *
 179  * Note that code herein does not assume the image name is known (it
 180  * could be a null pointer).
 181  */
 182 struct rbd_spec {
 183         u64             pool_id;
 184         const char      *pool_name;
 185
 186         const char      *image_id;
 187         const char      *image_name;
 188
 189         u64             snap_id;
 190         const char      *snap_name;
 191
 192         struct kref     kref;
 193 };
 194
 195 /*
 196  * an instance of the client.  multiple devices may share an rbd client.
 197  */
 198 struct rbd_client {
 199         struct ceph_client      *client;
 200         struct kref             kref;
 201         struct list_head        node;
 202 };
 203
 204 struct rbd_img_request;
 205 typedef void (*rbd_img_callback_t)(struct rbd_img_request *);
 206
 207 #define BAD_WHICH       U32_MAX         /* Good which or bad which, which? */
 208
 209 struct rbd_obj_request;
 210 typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);
 211
 212 enum obj_request_type {
 213         OBJ_REQUEST_NODATA, OBJ_REQUEST_BIO, OBJ_REQUEST_PAGES
 214 };
 215
 216 enum obj_operation_type {
 217         OBJ_OP_WRITE,
 218         OBJ_OP_READ,
 219         OBJ_OP_DISCARD,
 220 };
 221
 222 enum obj_req_flags {
 223         OBJ_REQ_DONE,           /* completion flag: not done = 0, done = 1 */
 224         OBJ_REQ_IMG_DATA,       /* object usage: standalone = 0, image = 1 */
 225         OBJ_REQ_KNOWN,          /* EXISTS flag valid: no = 0, yes = 1 */
 226         OBJ_REQ_EXISTS,         /* target exists: no = 0, yes = 1 */
 227 };
 228
 229 struct rbd_obj_request {
 230         const char              *object_name;
 231         u64                     offset;         /* object start byte */
 232         u64                     length;         /* bytes from offset */
 233         unsigned long           flags;
 234
 235         /*
 236          * An object request associated with an image will have its
 237          * img_data flag set; a standalone object request will not.
 238          *
 239          * A standalone object request will have which == BAD_WHICH
 240          * and a null obj_request pointer.
 241          *
 242          * An object request initiated in support of a layered image
 243          * object (to check for its existence before a write) will
 244          * have which == BAD_WHICH and a non-null obj_request pointer.
 245          *
 246          * Finally, an object request for rbd image data will have
 247          * which != BAD_WHICH, and will have a non-null img_request
 248          * pointer.  The value of which will be in the range
 249          * 0..(img_request->obj_request_count-1).
 250          */
 251         union {
 252                 struct rbd_obj_request  *obj_request;   /* STAT op */
 253                 struct {
 254                         struct rbd_img_request  *img_request;
 255                         u64                     img_offset;
 256                         /* links for img_request->obj_requests list */
 257                         struct list_head        links;
 258                 };
 259         };
 260         u32                     which;          /* posn image request list */
 261
 262         enum obj_request_type   type;
 263         union {
 264                 struct bio      *bio_list;
 265                 struct {
 266                         struct page     **pages;
 267                         u32             page_count;
 268                 };
 269         };
 270         struct page             **copyup_pages;
 271         u32                     copyup_page_count;
 272
 273         struct ceph_osd_request *osd_req;
 274
 275         u64                     xferred;        /* bytes transferred */
 276         int                     result;
 277
 278         rbd_obj_callback_t      callback;
 279         struct completion       completion;
 280
 281         struct kref             kref;
 282 };
 283
 284 enum img_req_flags {
 285         IMG_REQ_WRITE,          /* I/O direction: read = 0, write = 1 */
 286         IMG_REQ_CHILD,          /* initiator: block = 0, child image = 1 */
 287         IMG_REQ_LAYERED,        /* ENOENT handling: normal = 0, layered = 1 */
 288         IMG_REQ_DISCARD,        /* discard: normal = 0, discard request = 1 */
 289 };
 290
 291 struct rbd_img_request {
 292         struct rbd_device       *rbd_dev;
 293         u64                     offset; /* starting image byte offset */
 294         u64                     length; /* byte count from offset */
 295         unsigned long           flags;
 296         union {
 297                 u64                     snap_id;        /* for reads */
 298                 struct ceph_snap_context *snapc;        /* for writes */
 299         };
 300         union {
 301                 struct request          *rq;            /* block request */
 302                 struct rbd_obj_request  *obj_request;   /* obj req initiator */
 303         };
 304         struct page             **copyup_pages;
 305         u32                     copyup_page_count;
 306         spinlock_t              completion_lock;/* protects next_completion */
 307         u32                     next_completion;
 308         rbd_img_callback_t      callback;
 309         u64                     xferred;/* aggregate bytes transferred */
 310         int                     result; /* first nonzero obj_request result */
 311
 312         u32                     obj_request_count;
 313         struct list_head        obj_requests;   /* rbd_obj_request structs */
 314
 315         struct kref             kref;
 316 };
 317
 318 #define for_each_obj_request(ireq, oreq) \
 319         list_for_each_entry(oreq, &(ireq)->obj_requests, links)
 320 #define for_each_obj_request_from(ireq, oreq) \
 321         list_for_each_entry_from(oreq, &(ireq)->obj_requests, links)
 322 #define for_each_obj_request_safe(ireq, oreq, n) \
 323         list_for_each_entry_safe_reverse(oreq, n, &(ireq)->obj_requests, links)
 324
 325 struct rbd_mapping {
 326         u64                     size;
 327         u64                     features;
 328         bool                    read_only;
 329 };
 330
 331 /*
 332  * a single device
 333  */
 334 struct rbd_device {
 335         int                     dev_id;         /* blkdev unique id */
 336
 337         int                     major;          /* blkdev assigned major */
 338         int                     minor;
 339         struct gendisk          *disk;          /* blkdev's gendisk and rq */
 340
 341         u32                     image_format;   /* Either 1 or 2 */
 342         struct rbd_client       *rbd_client;
 343
 344         char                    name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
 345
 346         spinlock_t              lock;           /* queue, flags, open_count */
 347
 348         struct rbd_image_header header;
 349         unsigned long           flags;          /* possibly lock protected */
 350         struct rbd_spec         *spec;
 351         struct rbd_options      *opts;
 352
 353         struct ceph_object_id   header_oid;
 354         struct ceph_object_locator header_oloc;
 355
 356         struct ceph_file_layout layout;
 357
 358         struct ceph_osd_linger_request *watch_handle;
 359
 360         struct rbd_spec         *parent_spec;
 361         u64                     parent_overlap;
 362         atomic_t                parent_ref;
 363         struct rbd_device       *parent;
 364
 365         /* Block layer tags. */
 366         struct blk_mq_tag_set   tag_set;
 367
 368         /* protects updating the header */
 369         struct rw_semaphore     header_rwsem;
 370
 371         struct rbd_mapping      mapping;
 372
 373         struct list_head        node;
 374
 375         /* sysfs related */
 376         struct device           dev;
 377         unsigned long           open_count;     /* protected by lock */
 378 };
 379
 380 /*
 381  * Flag bits for rbd_dev->flags.  If atomicity is required,
 382  * rbd_dev->lock is used to protect access.
 383  *
 384  * Currently, only the "removing" flag (which is coupled with the
 385  * "open_count" field) requires atomic access.
 386  */
 387 enum rbd_dev_flags {
 388         RBD_DEV_FLAG_EXISTS,    /* mapped snapshot has not been deleted */
 389         RBD_DEV_FLAG_REMOVING,  /* this mapping is being removed */
 390 };
 391
 392 static DEFINE_MUTEX(client_mutex);      /* Serialize client creation */
 393
 394 static LIST_HEAD(rbd_dev_list);    /* devices */
 395 static DEFINE_SPINLOCK(rbd_dev_list_lock);
 396
 397 static LIST_HEAD(rbd_client_list);              /* clients */
 398 static DEFINE_SPINLOCK(rbd_client_list_lock);
 399
 400 /* Slab caches for frequently-allocated structures */
 401
 402 static struct kmem_cache        *rbd_img_request_cache;
 403 static struct kmem_cache        *rbd_obj_request_cache;
 404 static struct kmem_cache        *rbd_segment_name_cache;
 405
 406 static int rbd_major;
 407 static DEFINE_IDA(rbd_dev_id_ida);
 408
 409 static struct workqueue_struct *rbd_wq;
 410
 411 /*
 412  * Default to false for now, as single-major requires >= 0.75 version of
 413  * userspace rbd utility.
 414  */
 415 static bool single_major = false;
 416 module_param(single_major, bool, S_IRUGO);
 417 MODULE_PARM_DESC(single_major, "Use a single major number for all rbd devices (default: false)");
 418
 419 static int rbd_img_request_submit(struct rbd_img_request *img_request);
 420
 421 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
 422                        size_t count);
 423 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
 424                           size_t count);
 425 static ssize_t rbd_add_single_major(struct bus_type *bus, const char *buf,
 426                                     size_t count);
 427 static ssize_t rbd_remove_single_major(struct bus_type *bus, const char *buf,
 428                                        size_t count);
 429 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth);
 430 static void rbd_spec_put(struct rbd_spec *spec);
 431
 432 static int rbd_dev_id_to_minor(int dev_id)
 433 {
 434         return dev_id << RBD_SINGLE_MAJOR_PART_SHIFT;
 435 }
 436
 437 static int minor_to_rbd_dev_id(int minor)
 438 {
 439         return minor >> RBD_SINGLE_MAJOR_PART_SHIFT;
 440 }
 441
 442 static BUS_ATTR(add, S_IWUSR, NULL, rbd_add);
 443 static BUS_ATTR(remove, S_IWUSR, NULL, rbd_remove);
 444 static BUS_ATTR(add_single_major, S_IWUSR, NULL, rbd_add_single_major);
 445 static BUS_ATTR(remove_single_major, S_IWUSR, NULL, rbd_remove_single_major);
 446
 447 static struct attribute *rbd_bus_attrs[] = {
 448         &bus_attr_add.attr,
 449         &bus_attr_remove.attr,
 450         &bus_attr_add_single_major.attr,
 451         &bus_attr_remove_single_major.attr,
 452         NULL,
 453 };
 454
 455 static umode_t rbd_bus_is_visible(struct kobject *kobj,
 456                                   struct attribute *attr, int index)
 457 {
 458         if (!single_major &&
 459             (attr == &bus_attr_add_single_major.attr ||
 460              attr == &bus_attr_remove_single_major.attr))
 461                 return 0;
 462
 463         return attr->mode;
 464 }
 465
 466 static const struct attribute_group rbd_bus_group = {
 467         .attrs = rbd_bus_attrs,
 468         .is_visible = rbd_bus_is_visible,
 469 };
 470 __ATTRIBUTE_GROUPS(rbd_bus);
 471
 472 static struct bus_type rbd_bus_type = {
 473         .name           = "rbd",
 474         .bus_groups     = rbd_bus_groups,
 475 };
 476
 477 static void rbd_root_dev_release(struct device *dev)
 478 {
 479 }
 480
 481 static struct device rbd_root_dev = {
 482         .init_name =    "rbd",
 483         .release =      rbd_root_dev_release,
 484 };
 485
 486 static __printf(2, 3)
 487 void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
 488 {
 489         struct va_format vaf;
 490         va_list args;
 491
 492         va_start(args, fmt);
 493         vaf.fmt = fmt;
 494         vaf.va = &args;
 495
 496         if (!rbd_dev)
 497                 printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
 498         else if (rbd_dev->disk)
 499                 printk(KERN_WARNING "%s: %s: %pV\n",
 500                         RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
 501         else if (rbd_dev->spec && rbd_dev->spec->image_name)
 502                 printk(KERN_WARNING "%s: image %s: %pV\n",
 503                         RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
 504         else if (rbd_dev->spec && rbd_dev->spec->image_id)
 505                 printk(KERN_WARNING "%s: id %s: %pV\n",
 506                         RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
 507         else    /* punt */
 508                 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
 509                         RBD_DRV_NAME, rbd_dev, &vaf);
 510         va_end(args);
 511 }
 512
 513 #ifdef RBD_DEBUG
 514 #define rbd_assert(expr)                                                \
 515                 if (unlikely(!(expr))) {                                \
 516                         printk(KERN_ERR "\nAssertion failure in %s() "  \
 517                                                 "at line %d:\n\n"       \
 518                                         "\trbd_assert(%s);\n\n",        \
 519                                         __func__, __LINE__, #expr);     \
 520                         BUG();                                          \
 521                 }
 522 #else /* !RBD_DEBUG */
 523 #  define rbd_assert(expr)      ((void) 0)
 524 #endif /* !RBD_DEBUG */
 525
 526 static void rbd_osd_copyup_callback(struct rbd_obj_request *obj_request);
 527 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request);
 528 static void rbd_img_parent_read(struct rbd_obj_request *obj_request);
 529 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev);
 530
 531 static int rbd_dev_refresh(struct rbd_device *rbd_dev);
 532 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev);
 533 static int rbd_dev_header_info(struct rbd_device *rbd_dev);
 534 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev);
 535 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
 536                                         u64 snap_id);
 537 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
 538                                 u8 *order, u64 *snap_size);
 539 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
 540                 u64 *snap_features);
 541
 542 static int rbd_open(struct block_device *bdev, fmode_t mode)
 543 {
 544         struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
 545         bool removing = false;
 546
 547         if ((mode & FMODE_WRITE) && rbd_dev->mapping.read_only)
 548                 return -EROFS;
 549
 550         spin_lock_irq(&rbd_dev->lock);
 551         if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
 552                 removing = true;
 553         else
 554                 rbd_dev->open_count++;
 555         spin_unlock_irq(&rbd_dev->lock);
 556         if (removing)
 557                 return -ENOENT;
 558
 559         (void) get_device(&rbd_dev->dev);
 560
 561         return 0;
 562 }
 563
 564 static void rbd_release(struct gendisk *disk, fmode_t mode)
 565 {
 566         struct rbd_device *rbd_dev = disk->private_data;
 567         unsigned long open_count_before;
 568
 569         spin_lock_irq(&rbd_dev->lock);
 570         open_count_before = rbd_dev->open_count--;
 571         spin_unlock_irq(&rbd_dev->lock);
 572         rbd_assert(open_count_before > 0);
 573
 574         put_device(&rbd_dev->dev);
 575 }
 576
 577 static int rbd_ioctl_set_ro(struct rbd_device *rbd_dev, unsigned long arg)
 578 {
 579         int ret = 0;
 580         int val;
 581         bool ro;
 582         bool ro_changed = false;
 583
 584         /* get_user() may sleep, so call it before taking rbd_dev->lock */
 585         if (get_user(val, (int __user *)(arg)))
 586                 return -EFAULT;
 587
 588         ro = val ? true : false;
 589         /* Snapshot doesn't allow to write*/
 590         if (rbd_dev->spec->snap_id != CEPH_NOSNAP && !ro)
 591                 return -EROFS;
 592
 593         spin_lock_irq(&rbd_dev->lock);
 594         /* prevent others open this device */
 595         if (rbd_dev->open_count > 1) {
 596                 ret = -EBUSY;
 597                 goto out;
 598         }
 599
 600         if (rbd_dev->mapping.read_only != ro) {
 601                 rbd_dev->mapping.read_only = ro;
 602                 ro_changed = true;
 603         }
 604
 605 out:
 606         spin_unlock_irq(&rbd_dev->lock);
 607         /* set_disk_ro() may sleep, so call it after releasing rbd_dev->lock */
 608         if (ret == 0 && ro_changed)
 609                 set_disk_ro(rbd_dev->disk, ro ? 1 : 0);
 610
 611         return ret;
 612 }
 613
 614 static int rbd_ioctl(struct block_device *bdev, fmode_t mode,
 615                         unsigned int cmd, unsigned long arg)
 616 {
 617         struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
 618         int ret = 0;
 619
 620         switch (cmd) {
 621         case BLKROSET:
 622                 ret = rbd_ioctl_set_ro(rbd_dev, arg);
 623                 break;
 624         default:
 625                 ret = -ENOTTY;
 626         }
 627
 628         return ret;
 629 }
 630
 631 #ifdef CONFIG_COMPAT
 632 static int rbd_compat_ioctl(struct block_device *bdev, fmode_t mode,
 633                                 unsigned int cmd, unsigned long arg)
 634 {
 635         return rbd_ioctl(bdev, mode, cmd, arg);
 636 }
 637 #endif /* CONFIG_COMPAT */
 638
 639 static const struct block_device_operations rbd_bd_ops = {
 640         .owner                  = THIS_MODULE,
 641         .open                   = rbd_open,
 642         .release                = rbd_release,
 643         .ioctl                  = rbd_ioctl,
 644 #ifdef CONFIG_COMPAT
 645         .compat_ioctl           = rbd_compat_ioctl,
 646 #endif
 647 };
 648
 649 /*
 650  * Initialize an rbd client instance.  Success or not, this function
 651  * consumes ceph_opts.  Caller holds client_mutex.
 652  */
 653 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
 654 {
 655         struct rbd_client *rbdc;
 656         int ret = -ENOMEM;
 657
 658         dout("%s:\n", __func__);
 659         rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
 660         if (!rbdc)
 661                 goto out_opt;
 662
 663         kref_init(&rbdc->kref);
 664         INIT_LIST_HEAD(&rbdc->node);
 665
 666         rbdc->client = ceph_create_client(ceph_opts, rbdc, 0, 0);
 667         if (IS_ERR(rbdc->client))
 668                 goto out_rbdc;
 669         ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
 670
 671         ret = ceph_open_session(rbdc->client);
 672         if (ret < 0)
 673                 goto out_client;
 674
 675         spin_lock(&rbd_client_list_lock);
 676         list_add_tail(&rbdc->node, &rbd_client_list);
 677         spin_unlock(&rbd_client_list_lock);
 678
 679         dout("%s: rbdc %p\n", __func__, rbdc);
 680
 681         return rbdc;
 682 out_client:
 683         ceph_destroy_client(rbdc->client);
 684 out_rbdc:
 685         kfree(rbdc);
 686 out_opt:
 687         if (ceph_opts)
 688                 ceph_destroy_options(ceph_opts);
 689         dout("%s: error %d\n", __func__, ret);
 690
 691         return ERR_PTR(ret);
 692 }
 693
 694 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
 695 {
 696         kref_get(&rbdc->kref);
 697
 698         return rbdc;
 699 }
 700
 701 /*
 702  * Find a ceph client with specific addr and configuration.  If
 703  * found, bump its reference count.
 704  */
 705 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
 706 {
 707         struct rbd_client *client_node;
 708         bool found = false;
 709
 710         if (ceph_opts->flags & CEPH_OPT_NOSHARE)
 711                 return NULL;
 712
 713         spin_lock(&rbd_client_list_lock);
 714         list_for_each_entry(client_node, &rbd_client_list, node) {
 715                 if (!ceph_compare_options(ceph_opts, client_node->client)) {
 716                         __rbd_get_client(client_node);
 717
 718                         found = true;
 719                         break;
 720                 }
 721         }
 722         spin_unlock(&rbd_client_list_lock);
 723
 724         return found ? client_node : NULL;
 725 }
 726
 727 /*
 728  * (Per device) rbd map options
 729  */
 730 enum {
 731         Opt_queue_depth,
 732         Opt_last_int,
 733         /* int args above */
 734         Opt_last_string,
 735         /* string args above */
 736         Opt_read_only,
 737         Opt_read_write,
 738         Opt_err
 739 };
 740
 741 static match_table_t rbd_opts_tokens = {
 742         {Opt_queue_depth, "queue_depth=%d"},
 743         /* int args above */
 744         /* string args above */
 745         {Opt_read_only, "read_only"},
 746         {Opt_read_only, "ro"},          /* Alternate spelling */
 747         {Opt_read_write, "read_write"},
 748         {Opt_read_write, "rw"},         /* Alternate spelling */
 749         {Opt_err, NULL}
 750 };
 751
 752 struct rbd_options {
 753         int     queue_depth;
 754         bool    read_only;
 755 };
 756
 757 #define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_MAX_RQ
 758 #define RBD_READ_ONLY_DEFAULT   false
 759
 760 static int parse_rbd_opts_token(char *c, void *private)
 761 {
 762         struct rbd_options *rbd_opts = private;
 763         substring_t argstr[MAX_OPT_ARGS];
 764         int token, intval, ret;
 765
 766         token = match_token(c, rbd_opts_tokens, argstr);
 767         if (token < Opt_last_int) {
 768                 ret = match_int(&argstr[0], &intval);
 769                 if (ret < 0) {
 770                         pr_err("bad mount option arg (not int) at '%s'\n", c);
 771                         return ret;
 772                 }
 773                 dout("got int token %d val %d\n", token, intval);
 774         } else if (token > Opt_last_int && token < Opt_last_string) {
 775                 dout("got string token %d val %s\n", token, argstr[0].from);
 776         } else {
 777                 dout("got token %d\n", token);
 778         }
 779
 780         switch (token) {
 781         case Opt_queue_depth:
 782                 if (intval < 1) {
 783                         pr_err("queue_depth out of range\n");
 784                         return -EINVAL;
 785                 }
 786                 rbd_opts->queue_depth = intval;
 787                 break;
 788         case Opt_read_only:
 789                 rbd_opts->read_only = true;
 790                 break;
 791         case Opt_read_write:
 792                 rbd_opts->read_only = false;
 793                 break;
 794         default:
 795                 /* libceph prints "bad option" msg */
 796                 return -EINVAL;
 797         }
 798
 799         return 0;
 800 }
 801
 802 static char* obj_op_name(enum obj_operation_type op_type)
 803 {
 804         switch (op_type) {
 805         case OBJ_OP_READ:
 806                 return "read";
 807         case OBJ_OP_WRITE:
 808                 return "write";
 809         case OBJ_OP_DISCARD:
 810                 return "discard";
 811         default:
 812                 return "???";
 813         }
 814 }
 815
 816 /*
 817  * Get a ceph client with specific addr and configuration, if one does
 818  * not exist create it.  Either way, ceph_opts is consumed by this
 819  * function.
 820  */
 821 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
 822 {
 823         struct rbd_client *rbdc;
 824
 825         mutex_lock_nested(&client_mutex, SINGLE_DEPTH_NESTING);
 826         rbdc = rbd_client_find(ceph_opts);
 827         if (rbdc)       /* using an existing client */
 828                 ceph_destroy_options(ceph_opts);
 829         else
 830                 rbdc = rbd_client_create(ceph_opts);
 831         mutex_unlock(&client_mutex);
 832
 833         return rbdc;
 834 }
 835
 836 /*
 837  * Destroy ceph client
 838  *
 839  * Caller must hold rbd_client_list_lock.
 840  */
 841 static void rbd_client_release(struct kref *kref)
 842 {
 843         struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
 844
 845         dout("%s: rbdc %p\n", __func__, rbdc);
 846         spin_lock(&rbd_client_list_lock);
 847         list_del(&rbdc->node);
 848         spin_unlock(&rbd_client_list_lock);
 849
 850         ceph_destroy_client(rbdc->client);
 851         kfree(rbdc);
 852 }
 853
 854 /*
 855  * Drop reference to ceph client node. If it's not referenced anymore, release
 856  * it.
 857  */
 858 static void rbd_put_client(struct rbd_client *rbdc)
 859 {
 860         if (rbdc)
 861                 kref_put(&rbdc->kref, rbd_client_release);
 862 }
 863
 864 static bool rbd_image_format_valid(u32 image_format)
 865 {
 866         return image_format == 1 || image_format == 2;
 867 }
 868
 869 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
 870 {
 871         size_t size;
 872         u32 snap_count;
 873
 874         /* The header has to start with the magic rbd header text */
 875         if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
 876                 return false;
 877
 878         /* The bio layer requires at least sector-sized I/O */
 879
 880         if (ondisk->options.order < SECTOR_SHIFT)
 881                 return false;
 882
 883         /* If we use u64 in a few spots we may be able to loosen this */
 884
 885         if (ondisk->options.order > 8 * sizeof (int) - 1)
 886                 return false;
 887
 888         /*
 889          * The size of a snapshot header has to fit in a size_t, and
 890          * that limits the number of snapshots.
 891          */
 892         snap_count = le32_to_cpu(ondisk->snap_count);
 893         size = SIZE_MAX - sizeof (struct ceph_snap_context);
 894         if (snap_count > size / sizeof (__le64))
 895                 return false;
 896
 897         /*
 898          * Not only that, but the size of the entire the snapshot
 899          * header must also be representable in a size_t.
 900          */
 901         size -= snap_count * sizeof (__le64);
 902         if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
 903                 return false;
 904
 905         return true;
 906 }
 907
 908 /*
 909  * Fill an rbd image header with information from the given format 1
 910  * on-disk header.
 911  */
 912 static int rbd_header_from_disk(struct rbd_device *rbd_dev,
 913                                  struct rbd_image_header_ondisk *ondisk)
 914 {
 915         struct rbd_image_header *header = &rbd_dev->header;
 916         bool first_time = header->object_prefix == NULL;
 917         struct ceph_snap_context *snapc;
 918         char *object_prefix = NULL;
 919         char *snap_names = NULL;
 920         u64 *snap_sizes = NULL;
 921         u32 snap_count;
 922         size_t size;
 923         int ret = -ENOMEM;
 924         u32 i;
 925
 926         /* Allocate this now to avoid having to handle failure below */
 927
 928         if (first_time) {
 929                 size_t len;
 930
 931                 len = strnlen(ondisk->object_prefix,
 932                                 sizeof (ondisk->object_prefix));
 933                 object_prefix = kmalloc(len + 1, GFP_KERNEL);
 934                 if (!object_prefix)
 935                         return -ENOMEM;
 936                 memcpy(object_prefix, ondisk->object_prefix, len);
 937                 object_prefix[len] = '\0';
 938         }
 939
 940         /* Allocate the snapshot context and fill it in */
 941
 942         snap_count = le32_to_cpu(ondisk->snap_count);
 943         snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
 944         if (!snapc)
 945                 goto out_err;
 946         snapc->seq = le64_to_cpu(ondisk->snap_seq);
 947         if (snap_count) {
 948                 struct rbd_image_snap_ondisk *snaps;
 949                 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
 950
 951                 /* We'll keep a copy of the snapshot names... */
 952
 953                 if (snap_names_len > (u64)SIZE_MAX)
 954                         goto out_2big;
 955                 snap_names = kmalloc(snap_names_len, GFP_KERNEL);
 956                 if (!snap_names)
 957                         goto out_err;
 958
 959                 /* ...as well as the array of their sizes. */
 960
 961                 size = snap_count * sizeof (*header->snap_sizes);
 962                 snap_sizes = kmalloc(size, GFP_KERNEL);
 963                 if (!snap_sizes)
 964                         goto out_err;
 965
 966                 /*
 967                  * Copy the names, and fill in each snapshot's id
 968                  * and size.
 969                  *
 970                  * Note that rbd_dev_v1_header_info() guarantees the
 971                  * ondisk buffer we're working with has
 972                  * snap_names_len bytes beyond the end of the
 973                  * snapshot id array, this memcpy() is safe.
 974                  */
 975                 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
 976                 snaps = ondisk->snaps;
 977                 for (i = 0; i < snap_count; i++) {
 978                         snapc->snaps[i] = le64_to_cpu(snaps[i].id);
 979                         snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
 980                 }
 981         }
 982
 983         /* We won't fail any more, fill in the header */
 984
 985         if (first_time) {
 986                 header->object_prefix = object_prefix;
 987                 header->obj_order = ondisk->options.order;
 988                 header->crypt_type = ondisk->options.crypt_type;
 989                 header->comp_type = ondisk->options.comp_type;
 990                 /* The rest aren't used for format 1 images */
 991                 header->stripe_unit = 0;
 992                 header->stripe_count = 0;
 993                 header->features = 0;
 994         } else {
 995                 ceph_put_snap_context(header->snapc);
 996                 kfree(header->snap_names);
 997                 kfree(header->snap_sizes);
 998         }
 999
1000         /* The remaining fields always get updated (when we refresh) */
1001
1002         header->image_size = le64_to_cpu(ondisk->image_size);
1003         header->snapc = snapc;
1004         header->snap_names = snap_names;
1005         header->snap_sizes = snap_sizes;
1006
1007         return 0;
1008 out_2big:
1009         ret = -EIO;
1010 out_err:
1011         kfree(snap_sizes);
1012         kfree(snap_names);
1013         ceph_put_snap_context(snapc);
1014         kfree(object_prefix);
1015
1016         return ret;
1017 }
1018
1019 static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1020 {
1021         const char *snap_name;
1022
1023         rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1024
1025         /* Skip over names until we find the one we are looking for */
1026
1027         snap_name = rbd_dev->header.snap_names;
1028         while (which--)
1029                 snap_name += strlen(snap_name) + 1;
1030
1031         return kstrdup(snap_name, GFP_KERNEL);
1032 }
1033
1034 /*
1035  * Snapshot id comparison function for use with qsort()/bsearch().
1036  * Note that result is for snapshots in *descending* order.
1037  */
1038 static int snapid_compare_reverse(const void *s1, const void *s2)
1039 {
1040         u64 snap_id1 = *(u64 *)s1;
1041         u64 snap_id2 = *(u64 *)s2;
1042
1043         if (snap_id1 < snap_id2)
1044                 return 1;
1045         return snap_id1 == snap_id2 ? 0 : -1;
1046 }
1047
1048 /*
1049  * Search a snapshot context to see if the given snapshot id is
1050  * present.
1051  *
1052  * Returns the position of the snapshot id in the array if it's found,
1053  * or BAD_SNAP_INDEX otherwise.
1054  *
1055  * Note: The snapshot array is in kept sorted (by the osd) in
1056  * reverse order, highest snapshot id first.
1057  */
1058 static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1059 {
1060         struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1061         u64 *found;
1062
1063         found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1064                                 sizeof (snap_id), snapid_compare_reverse);
1065
1066         return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1067 }
1068
1069 static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1070                                         u64 snap_id)
1071 {
1072         u32 which;
1073         const char *snap_name;
1074
1075         which = rbd_dev_snap_index(rbd_dev, snap_id);
1076         if (which == BAD_SNAP_INDEX)
1077                 return ERR_PTR(-ENOENT);
1078
1079         snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1080         return snap_name ? snap_name : ERR_PTR(-ENOMEM);
1081 }
1082
1083 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1084 {
1085         if (snap_id == CEPH_NOSNAP)
1086                 return RBD_SNAP_HEAD_NAME;
1087
1088         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1089         if (rbd_dev->image_format == 1)
1090                 return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1091
1092         return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1093 }
1094
1095 static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1096                                 u64 *snap_size)
1097 {
1098         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1099         if (snap_id == CEPH_NOSNAP) {
1100                 *snap_size = rbd_dev->header.image_size;
1101         } else if (rbd_dev->image_format == 1) {
1102                 u32 which;
1103
1104                 which = rbd_dev_snap_index(rbd_dev, snap_id);
1105                 if (which == BAD_SNAP_INDEX)
1106                         return -ENOENT;
1107
1108                 *snap_size = rbd_dev->header.snap_sizes[which];
1109         } else {
1110                 u64 size = 0;
1111                 int ret;
1112
1113                 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1114                 if (ret)
1115                         return ret;
1116
1117                 *snap_size = size;
1118         }
1119         return 0;
1120 }
1121
1122 static int rbd_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
1123                         u64 *snap_features)
1124 {
1125         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1126         if (snap_id == CEPH_NOSNAP) {
1127                 *snap_features = rbd_dev->header.features;
1128         } else if (rbd_dev->image_format == 1) {
1129                 *snap_features = 0;     /* No features for format 1 */
1130         } else {
1131                 u64 features = 0;
1132                 int ret;
1133
1134                 ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, &features);
1135                 if (ret)
1136                         return ret;
1137
1138                 *snap_features = features;
1139         }
1140         return 0;
1141 }
1142
1143 static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1144 {
1145         u64 snap_id = rbd_dev->spec->snap_id;
1146         u64 size = 0;
1147         u64 features = 0;
1148         int ret;
1149
1150         ret = rbd_snap_size(rbd_dev, snap_id, &size);
1151         if (ret)
1152                 return ret;
1153         ret = rbd_snap_features(rbd_dev, snap_id, &features);
1154         if (ret)
1155                 return ret;
1156
1157         rbd_dev->mapping.size = size;
1158         rbd_dev->mapping.features = features;
1159
1160         return 0;
1161 }
1162
1163 static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1164 {
1165         rbd_dev->mapping.size = 0;
1166         rbd_dev->mapping.features = 0;
1167 }
1168
1169 static void rbd_segment_name_free(const char *name)
1170 {
1171         /* The explicit cast here is needed to drop the const qualifier */
1172
1173         kmem_cache_free(rbd_segment_name_cache, (void *)name);
1174 }
1175
1176 static const char *rbd_segment_name(struct rbd_device *rbd_dev, u64 offset)
1177 {
1178         char *name;
1179         u64 segment;
1180         int ret;
1181         char *name_format;
1182
1183         name = kmem_cache_alloc(rbd_segment_name_cache, GFP_NOIO);
1184         if (!name)
1185                 return NULL;
1186         segment = offset >> rbd_dev->header.obj_order;
1187         name_format = "%s.%012llx";
1188         if (rbd_dev->image_format == 2)
1189                 name_format = "%s.%016llx";
1190         ret = snprintf(name, CEPH_MAX_OID_NAME_LEN + 1, name_format,
1191                         rbd_dev->header.object_prefix, segment);
1192         if (ret < 0 || ret > CEPH_MAX_OID_NAME_LEN) {
1193                 pr_err("error formatting segment name for #%llu (%d)\n",
1194                         segment, ret);
1195                 rbd_segment_name_free(name);
1196                 name = NULL;
1197         }
1198
1199         return name;
1200 }
1201
1202 static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset)
1203 {
1204         u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
1205
1206         return offset & (segment_size - 1);
1207 }
1208
1209 static u64 rbd_segment_length(struct rbd_device *rbd_dev,
1210                                 u64 offset, u64 length)
1211 {
1212         u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
1213
1214         offset &= segment_size - 1;
1215
1216         rbd_assert(length <= U64_MAX - offset);
1217         if (offset + length > segment_size)
1218                 length = segment_size - offset;
1219
1220         return length;
1221 }
1222
1223 /*
1224  * returns the size of an object in the image
1225  */
1226 static u64 rbd_obj_bytes(struct rbd_image_header *header)
1227 {
1228         return 1 << header->obj_order;
1229 }
1230
1231 /*
1232  * bio helpers
1233  */
1234
1235 static void bio_chain_put(struct bio *chain)
1236 {
1237         struct bio *tmp;
1238
1239         while (chain) {
1240                 tmp = chain;
1241                 chain = chain->bi_next;
1242                 bio_put(tmp);
1243         }
1244 }
1245
1246 /*
1247  * zeros a bio chain, starting at specific offset
1248  */
1249 static void zero_bio_chain(struct bio *chain, int start_ofs)
1250 {
1251         struct bio_vec bv;
1252         struct bvec_iter iter;
1253         unsigned long flags;
1254         void *buf;
1255         int pos = 0;
1256
1257         while (chain) {
1258                 bio_for_each_segment(bv, chain, iter) {
1259                         if (pos + bv.bv_len > start_ofs) {
1260                                 int remainder = max(start_ofs - pos, 0);
1261                                 buf = bvec_kmap_irq(&bv, &flags);
1262                                 memset(buf + remainder, 0,
1263                                        bv.bv_len - remainder);
1264                                 flush_dcache_page(bv.bv_page);
1265                                 bvec_kunmap_irq(buf, &flags);
1266                         }
1267                         pos += bv.bv_len;
1268                 }
1269
1270                 chain = chain->bi_next;
1271         }
1272 }
1273
1274 /*
1275  * similar to zero_bio_chain(), zeros data defined by a page array,
1276  * starting at the given byte offset from the start of the array and
1277  * continuing up to the given end offset.  The pages array is
1278  * assumed to be big enough to hold all bytes up to the end.
1279  */
1280 static void zero_pages(struct page **pages, u64 offset, u64 end)
1281 {
1282         struct page **page = &pages[offset >> PAGE_SHIFT];
1283
1284         rbd_assert(end > offset);
1285         rbd_assert(end - offset <= (u64)SIZE_MAX);
1286         while (offset < end) {
1287                 size_t page_offset;
1288                 size_t length;
1289                 unsigned long flags;
1290                 void *kaddr;
1291
1292                 page_offset = offset & ~PAGE_MASK;
1293                 length = min_t(size_t, PAGE_SIZE - page_offset, end - offset);
1294                 local_irq_save(flags);
1295                 kaddr = kmap_atomic(*page);
1296                 memset(kaddr + page_offset, 0, length);
1297                 flush_dcache_page(*page);
1298                 kunmap_atomic(kaddr);
1299                 local_irq_restore(flags);
1300
1301                 offset += length;
1302                 page++;
1303         }
1304 }
1305
1306 /*
1307  * Clone a portion of a bio, starting at the given byte offset
1308  * and continuing for the number of bytes indicated.
1309  */
1310 static struct bio *bio_clone_range(struct bio *bio_src,
1311                                         unsigned int offset,
1312                                         unsigned int len,
1313                                         gfp_t gfpmask)
1314 {
1315         struct bio *bio;
1316
1317         bio = bio_clone(bio_src, gfpmask);
1318         if (!bio)
1319                 return NULL;    /* ENOMEM */
1320
1321         bio_advance(bio, offset);
1322         bio->bi_iter.bi_size = len;
1323
1324         return bio;
1325 }
1326
1327 /*
1328  * Clone a portion of a bio chain, starting at the given byte offset
1329  * into the first bio in the source chain and continuing for the
1330  * number of bytes indicated.  The result is another bio chain of
1331  * exactly the given length, or a null pointer on error.
1332  *
1333  * The bio_src and offset parameters are both in-out.  On entry they
1334  * refer to the first source bio and the offset into that bio where
1335  * the start of data to be cloned is located.
1336  *
1337  * On return, bio_src is updated to refer to the bio in the source
1338  * chain that contains first un-cloned byte, and *offset will
1339  * contain the offset of that byte within that bio.
1340  */
1341 static struct bio *bio_chain_clone_range(struct bio **bio_src,
1342                                         unsigned int *offset,
1343                                         unsigned int len,
1344                                         gfp_t gfpmask)
1345 {
1346         struct bio *bi = *bio_src;
1347         unsigned int off = *offset;
1348         struct bio *chain = NULL;
1349         struct bio **end;
1350
1351         /* Build up a chain of clone bios up to the limit */
1352
1353         if (!bi || off >= bi->bi_iter.bi_size || !len)
1354                 return NULL;            /* Nothing to clone */
1355
1356         end = &chain;
1357         while (len) {
1358                 unsigned int bi_size;
1359                 struct bio *bio;
1360
1361                 if (!bi) {
1362                         rbd_warn(NULL, "bio_chain exhausted with %u left", len);
1363                         goto out_err;   /* EINVAL; ran out of bio's */
1364                 }
1365                 bi_size = min_t(unsigned int, bi->bi_iter.bi_size - off, len);
1366                 bio = bio_clone_range(bi, off, bi_size, gfpmask);
1367                 if (!bio)
1368                         goto out_err;   /* ENOMEM */
1369
1370                 *end = bio;
1371                 end = &bio->bi_next;
1372
1373                 off += bi_size;
1374                 if (off == bi->bi_iter.bi_size) {
1375                         bi = bi->bi_next;
1376                         off = 0;
1377                 }
1378                 len -= bi_size;
1379         }
1380         *bio_src = bi;
1381         *offset = off;
1382
1383         return chain;
1384 out_err:
1385         bio_chain_put(chain);
1386
1387         return NULL;
1388 }
1389
1390 /*
1391  * The default/initial value for all object request flags is 0.  For
1392  * each flag, once its value is set to 1 it is never reset to 0
1393  * again.
1394  */
1395 static void obj_request_img_data_set(struct rbd_obj_request *obj_request)
1396 {
1397         if (test_and_set_bit(OBJ_REQ_IMG_DATA, &obj_request->flags)) {
1398                 struct rbd_device *rbd_dev;
1399
1400                 rbd_dev = obj_request->img_request->rbd_dev;
1401                 rbd_warn(rbd_dev, "obj_request %p already marked img_data",
1402                         obj_request);
1403         }
1404 }
1405
1406 static bool obj_request_img_data_test(struct rbd_obj_request *obj_request)
1407 {
1408         smp_mb();
1409         return test_bit(OBJ_REQ_IMG_DATA, &obj_request->flags) != 0;
1410 }
1411
1412 static void obj_request_done_set(struct rbd_obj_request *obj_request)
1413 {
1414         if (test_and_set_bit(OBJ_REQ_DONE, &obj_request->flags)) {
1415                 struct rbd_device *rbd_dev = NULL;
1416
1417                 if (obj_request_img_data_test(obj_request))
1418                         rbd_dev = obj_request->img_request->rbd_dev;
1419                 rbd_warn(rbd_dev, "obj_request %p already marked done",
1420                         obj_request);
1421         }
1422 }
1423
1424 static bool obj_request_done_test(struct rbd_obj_request *obj_request)
1425 {
1426         smp_mb();
1427         return test_bit(OBJ_REQ_DONE, &obj_request->flags) != 0;
1428 }
1429
1430 /*
1431  * This sets the KNOWN flag after (possibly) setting the EXISTS
1432  * flag.  The latter is set based on the "exists" value provided.
1433  *
1434  * Note that for our purposes once an object exists it never goes
1435  * away again.  It's possible that the response from two existence
1436  * checks are separated by the creation of the target object, and
1437  * the first ("doesn't exist") response arrives *after* the second
1438  * ("does exist").  In that case we ignore the second one.
1439  */
1440 static void obj_request_existence_set(struct rbd_obj_request *obj_request,
1441                                 bool exists)
1442 {
1443         if (exists)
1444                 set_bit(OBJ_REQ_EXISTS, &obj_request->flags);
1445         set_bit(OBJ_REQ_KNOWN, &obj_request->flags);
1446         smp_mb();
1447 }
1448
1449 static bool obj_request_known_test(struct rbd_obj_request *obj_request)
1450 {
1451         smp_mb();
1452         return test_bit(OBJ_REQ_KNOWN, &obj_request->flags) != 0;
1453 }
1454
1455 static bool obj_request_exists_test(struct rbd_obj_request *obj_request)
1456 {
1457         smp_mb();
1458         return test_bit(OBJ_REQ_EXISTS, &obj_request->flags) != 0;
1459 }
1460
1461 static bool obj_request_overlaps_parent(struct rbd_obj_request *obj_request)
1462 {
1463         struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1464
1465         return obj_request->img_offset <
1466             round_up(rbd_dev->parent_overlap, rbd_obj_bytes(&rbd_dev->header));
1467 }
1468
1469 static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
1470 {
1471         dout("%s: obj %p (was %d)\n", __func__, obj_request,
1472                 atomic_read(&obj_request->kref.refcount));
1473         kref_get(&obj_request->kref);
1474 }
1475
1476 static void rbd_obj_request_destroy(struct kref *kref);
1477 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1478 {
1479         rbd_assert(obj_request != NULL);
1480         dout("%s: obj %p (was %d)\n", __func__, obj_request,
1481                 atomic_read(&obj_request->kref.refcount));
1482         kref_put(&obj_request->kref, rbd_obj_request_destroy);
1483 }
1484
1485 static void rbd_img_request_get(struct rbd_img_request *img_request)
1486 {
1487         dout("%s: img %p (was %d)\n", __func__, img_request,
1488              atomic_read(&img_request->kref.refcount));
1489         kref_get(&img_request->kref);
1490 }
1491
1492 static bool img_request_child_test(struct rbd_img_request *img_request);
1493 static void rbd_parent_request_destroy(struct kref *kref);
1494 static void rbd_img_request_destroy(struct kref *kref);
1495 static void rbd_img_request_put(struct rbd_img_request *img_request)
1496 {
1497         rbd_assert(img_request != NULL);
1498         dout("%s: img %p (was %d)\n", __func__, img_request,
1499                 atomic_read(&img_request->kref.refcount));
1500         if (img_request_child_test(img_request))
1501                 kref_put(&img_request->kref, rbd_parent_request_destroy);
1502         else
1503                 kref_put(&img_request->kref, rbd_img_request_destroy);
1504 }
1505
1506 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1507                                         struct rbd_obj_request *obj_request)
1508 {
1509         rbd_assert(obj_request->img_request == NULL);
1510
1511         /* Image request now owns object's original reference */
1512         obj_request->img_request = img_request;
1513         obj_request->which = img_request->obj_request_count;
1514         rbd_assert(!obj_request_img_data_test(obj_request));
1515         obj_request_img_data_set(obj_request);
1516         rbd_assert(obj_request->which != BAD_WHICH);
1517         img_request->obj_request_count++;
1518         list_add_tail(&obj_request->links, &img_request->obj_requests);
1519         dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1520                 obj_request->which);
1521 }
1522
1523 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1524                                         struct rbd_obj_request *obj_request)
1525 {
1526         rbd_assert(obj_request->which != BAD_WHICH);
1527
1528         dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1529                 obj_request->which);
1530         list_del(&obj_request->links);
1531         rbd_assert(img_request->obj_request_count > 0);
1532         img_request->obj_request_count--;
1533         rbd_assert(obj_request->which == img_request->obj_request_count);
1534         obj_request->which = BAD_WHICH;
1535         rbd_assert(obj_request_img_data_test(obj_request));
1536         rbd_assert(obj_request->img_request == img_request);
1537         obj_request->img_request = NULL;
1538         obj_request->callback = NULL;
1539         rbd_obj_request_put(obj_request);
1540 }
1541
1542 static bool obj_request_type_valid(enum obj_request_type type)
1543 {
1544         switch (type) {
1545         case OBJ_REQUEST_NODATA:
1546         case OBJ_REQUEST_BIO:
1547         case OBJ_REQUEST_PAGES:
1548                 return true;
1549         default:
1550                 return false;
1551         }
1552 }
1553
1554 static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
1555                                 struct rbd_obj_request *obj_request)
1556 {
1557         dout("%s %p\n", __func__, obj_request);
1558         return ceph_osdc_start_request(osdc, obj_request->osd_req, false);
1559 }
1560
1561 static void rbd_obj_request_end(struct rbd_obj_request *obj_request)
1562 {
1563         dout("%s %p\n", __func__, obj_request);
1564         ceph_osdc_cancel_request(obj_request->osd_req);
1565 }
1566
1567 /*
1568  * Wait for an object request to complete.  If interrupted, cancel the
1569  * underlying osd request.
1570  *
1571  * @timeout: in jiffies, 0 means "wait forever"
1572  */
1573 static int __rbd_obj_request_wait(struct rbd_obj_request *obj_request,
1574                                   unsigned long timeout)
1575 {
1576         long ret;
1577
1578         dout("%s %p\n", __func__, obj_request);
1579         ret = wait_for_completion_interruptible_timeout(
1580                                         &obj_request->completion,
1581                                         ceph_timeout_jiffies(timeout));
1582         if (ret <= 0) {
1583                 if (ret == 0)
1584                         ret = -ETIMEDOUT;
1585                 rbd_obj_request_end(obj_request);
1586         } else {
1587                 ret = 0;
1588         }
1589
1590         dout("%s %p ret %d\n", __func__, obj_request, (int)ret);
1591         return ret;
1592 }
1593
1594 static int rbd_obj_request_wait(struct rbd_obj_request *obj_request)
1595 {
1596         return __rbd_obj_request_wait(obj_request, 0);
1597 }
1598
1599 static void rbd_img_request_complete(struct rbd_img_request *img_request)
1600 {
1601
1602         dout("%s: img %p\n", __func__, img_request);
1603
1604         /*
1605          * If no error occurred, compute the aggregate transfer
1606          * count for the image request.  We could instead use
1607          * atomic64_cmpxchg() to update it as each object request
1608          * completes; not clear which way is better off hand.
1609          */
1610         if (!img_request->result) {
1611                 struct rbd_obj_request *obj_request;
1612                 u64 xferred = 0;
1613
1614                 for_each_obj_request(img_request, obj_request)
1615                         xferred += obj_request->xferred;
1616                 img_request->xferred = xferred;
1617         }
1618
1619         if (img_request->callback)
1620                 img_request->callback(img_request);
1621         else
1622                 rbd_img_request_put(img_request);
1623 }
1624
1625 /*
1626  * The default/initial value for all image request flags is 0.  Each
1627  * is conditionally set to 1 at image request initialization time
1628  * and currently never change thereafter.
1629  */
1630 static void img_request_write_set(struct rbd_img_request *img_request)
1631 {
1632         set_bit(IMG_REQ_WRITE, &img_request->flags);
1633         smp_mb();
1634 }
1635
1636 static bool img_request_write_test(struct rbd_img_request *img_request)
1637 {
1638         smp_mb();
1639         return test_bit(IMG_REQ_WRITE, &img_request->flags) != 0;
1640 }
1641
1642 /*
1643  * Set the discard flag when the img_request is an discard request
1644  */
1645 static void img_request_discard_set(struct rbd_img_request *img_request)
1646 {
1647         set_bit(IMG_REQ_DISCARD, &img_request->flags);
1648         smp_mb();
1649 }
1650
1651 static bool img_request_discard_test(struct rbd_img_request *img_request)
1652 {
1653         smp_mb();
1654         return test_bit(IMG_REQ_DISCARD, &img_request->flags) != 0;
1655 }
1656
1657 static void img_request_child_set(struct rbd_img_request *img_request)
1658 {
1659         set_bit(IMG_REQ_CHILD, &img_request->flags);
1660         smp_mb();
1661 }
1662
1663 static void img_request_child_clear(struct rbd_img_request *img_request)
1664 {
1665         clear_bit(IMG_REQ_CHILD, &img_request->flags);
1666         smp_mb();
1667 }
1668
1669 static bool img_request_child_test(struct rbd_img_request *img_request)
1670 {
1671         smp_mb();
1672         return test_bit(IMG_REQ_CHILD, &img_request->flags) != 0;
1673 }
1674
1675 static void img_request_layered_set(struct rbd_img_request *img_request)
1676 {
1677         set_bit(IMG_REQ_LAYERED, &img_request->flags);
1678         smp_mb();
1679 }
1680
1681 static void img_request_layered_clear(struct rbd_img_request *img_request)
1682 {
1683         clear_bit(IMG_REQ_LAYERED, &img_request->flags);
1684         smp_mb();
1685 }
1686
1687 static bool img_request_layered_test(struct rbd_img_request *img_request)
1688 {
1689         smp_mb();
1690         return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1691 }
1692
1693 static enum obj_operation_type
1694 rbd_img_request_op_type(struct rbd_img_request *img_request)
1695 {
1696         if (img_request_write_test(img_request))
1697                 return OBJ_OP_WRITE;
1698         else if (img_request_discard_test(img_request))
1699                 return OBJ_OP_DISCARD;
1700         else
1701                 return OBJ_OP_READ;
1702 }
1703
1704 static void
1705 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
1706 {
1707         u64 xferred = obj_request->xferred;
1708         u64 length = obj_request->length;
1709
1710         dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1711                 obj_request, obj_request->img_request, obj_request->result,
1712                 xferred, length);
1713         /*
1714          * ENOENT means a hole in the image.  We zero-fill the entire
1715          * length of the request.  A short read also implies zero-fill
1716          * to the end of the request.  An error requires the whole
1717          * length of the request to be reported finished with an error
1718          * to the block layer.  In each case we update the xferred
1719          * count to indicate the whole request was satisfied.
1720          */
1721         rbd_assert(obj_request->type != OBJ_REQUEST_NODATA);
1722         if (obj_request->result == -ENOENT) {
1723                 if (obj_request->type == OBJ_REQUEST_BIO)
1724                         zero_bio_chain(obj_request->bio_list, 0);
1725                 else
1726                         zero_pages(obj_request->pages, 0, length);
1727                 obj_request->result = 0;
1728         } else if (xferred < length && !obj_request->result) {
1729                 if (obj_request->type == OBJ_REQUEST_BIO)
1730                         zero_bio_chain(obj_request->bio_list, xferred);
1731                 else
1732                         zero_pages(obj_request->pages, xferred, length);
1733         }
1734         obj_request->xferred = length;
1735         obj_request_done_set(obj_request);
1736 }
1737
1738 static void rbd_obj_request_complete(struct rbd_obj_request *obj_request)
1739 {
1740         dout("%s: obj %p cb %p\n", __func__, obj_request,
1741                 obj_request->callback);
1742         if (obj_request->callback)
1743                 obj_request->callback(obj_request);
1744         else
1745                 complete_all(&obj_request->completion);
1746 }
1747
1748 static void rbd_osd_read_callback(struct rbd_obj_request *obj_request)
1749 {
1750         struct rbd_img_request *img_request = NULL;
1751         struct rbd_device *rbd_dev = NULL;
1752         bool layered = false;
1753
1754         if (obj_request_img_data_test(obj_request)) {
1755                 img_request = obj_request->img_request;
1756                 layered = img_request && img_request_layered_test(img_request);
1757                 rbd_dev = img_request->rbd_dev;
1758         }
1759
1760         dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1761                 obj_request, img_request, obj_request->result,
1762                 obj_request->xferred, obj_request->length);
1763         if (layered && obj_request->result == -ENOENT &&
1764                         obj_request->img_offset < rbd_dev->parent_overlap)
1765                 rbd_img_parent_read(obj_request);
1766         else if (img_request)
1767                 rbd_img_obj_request_read_callback(obj_request);
1768         else
1769                 obj_request_done_set(obj_request);
1770 }
1771
1772 static void rbd_osd_write_callback(struct rbd_obj_request *obj_request)
1773 {
1774         dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1775                 obj_request->result, obj_request->length);
1776         /*
1777          * There is no such thing as a successful short write.  Set
1778          * it to our originally-requested length.
1779          */
1780         obj_request->xferred = obj_request->length;
1781         obj_request_done_set(obj_request);
1782 }
1783
1784 static void rbd_osd_discard_callback(struct rbd_obj_request *obj_request)
1785 {
1786         dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1787                 obj_request->result, obj_request->length);
1788         /*
1789          * There is no such thing as a successful short discard.  Set
1790          * it to our originally-requested length.
1791          */
1792         obj_request->xferred = obj_request->length;
1793         /* discarding a non-existent object is not a problem */
1794         if (obj_request->result == -ENOENT)
1795                 obj_request->result = 0;
1796         obj_request_done_set(obj_request);
1797 }
1798
1799 /*
1800  * For a simple stat call there's nothing to do.  We'll do more if
1801  * this is part of a write sequence for a layered image.
1802  */
1803 static void rbd_osd_stat_callback(struct rbd_obj_request *obj_request)
1804 {
1805         dout("%s: obj %p\n", __func__, obj_request);
1806         obj_request_done_set(obj_request);
1807 }
1808
1809 static void rbd_osd_call_callback(struct rbd_obj_request *obj_request)
1810 {
1811         dout("%s: obj %p\n", __func__, obj_request);
1812
1813         if (obj_request_img_data_test(obj_request))
1814                 rbd_osd_copyup_callback(obj_request);
1815         else
1816                 obj_request_done_set(obj_request);
1817 }
1818
1819 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
1820 {
1821         struct rbd_obj_request *obj_request = osd_req->r_priv;
1822         u16 opcode;
1823
1824         dout("%s: osd_req %p\n", __func__, osd_req);
1825         rbd_assert(osd_req == obj_request->osd_req);
1826         if (obj_request_img_data_test(obj_request)) {
1827                 rbd_assert(obj_request->img_request);
1828                 rbd_assert(obj_request->which != BAD_WHICH);
1829         } else {
1830                 rbd_assert(obj_request->which == BAD_WHICH);
1831         }
1832
1833         if (osd_req->r_result < 0)
1834                 obj_request->result = osd_req->r_result;
1835
1836         /*
1837          * We support a 64-bit length, but ultimately it has to be
1838          * passed to the block layer, which just supports a 32-bit
1839          * length field.
1840          */
1841         obj_request->xferred = osd_req->r_ops[0].outdata_len;
1842         rbd_assert(obj_request->xferred < (u64)UINT_MAX);
1843
1844         opcode = osd_req->r_ops[0].op;
1845         switch (opcode) {
1846         case CEPH_OSD_OP_READ:
1847                 rbd_osd_read_callback(obj_request);
1848                 break;
1849         case CEPH_OSD_OP_SETALLOCHINT:
1850                 rbd_assert(osd_req->r_ops[1].op == CEPH_OSD_OP_WRITE ||
1851                            osd_req->r_ops[1].op == CEPH_OSD_OP_WRITEFULL);
1852                 /* fall through */
1853         case CEPH_OSD_OP_WRITE:
1854         case CEPH_OSD_OP_WRITEFULL:
1855                 rbd_osd_write_callback(obj_request);
1856                 break;
1857         case CEPH_OSD_OP_STAT:
1858                 rbd_osd_stat_callback(obj_request);
1859                 break;
1860         case CEPH_OSD_OP_DELETE:
1861         case CEPH_OSD_OP_TRUNCATE:
1862         case CEPH_OSD_OP_ZERO:
1863                 rbd_osd_discard_callback(obj_request);
1864                 break;
1865         case CEPH_OSD_OP_CALL:
1866                 rbd_osd_call_callback(obj_request);
1867                 break;
1868         default:
1869                 rbd_warn(NULL, "%s: unsupported op %hu",
1870                         obj_request->object_name, (unsigned short) opcode);
1871                 break;
1872         }
1873
1874         if (obj_request_done_test(obj_request))
1875                 rbd_obj_request_complete(obj_request);
1876 }
1877
1878 static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request)
1879 {
1880         struct rbd_img_request *img_request = obj_request->img_request;
1881         struct ceph_osd_request *osd_req = obj_request->osd_req;
1882
1883         if (img_request)
1884                 osd_req->r_snapid = img_request->snap_id;
1885 }
1886
1887 static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request)
1888 {
1889         struct ceph_osd_request *osd_req = obj_request->osd_req;
1890
1891         osd_req->r_mtime = CURRENT_TIME;
1892         osd_req->r_data_offset = obj_request->offset;
1893 }
1894
1895 /*
1896  * Create an osd request.  A read request has one osd op (read).
1897  * A write request has either one (watch) or two (hint+write) osd ops.
1898  * (All rbd data writes are prefixed with an allocation hint op, but
1899  * technically osd watch is a write request, hence this distinction.)
1900  */
1901 static struct ceph_osd_request *rbd_osd_req_create(
1902                                         struct rbd_device *rbd_dev,
1903                                         enum obj_operation_type op_type,
1904                                         unsigned int num_ops,
1905                                         struct rbd_obj_request *obj_request)
1906 {
1907         struct ceph_snap_context *snapc = NULL;
1908         struct ceph_osd_client *osdc;
1909         struct ceph_osd_request *osd_req;
1910
1911         if (obj_request_img_data_test(obj_request) &&
1912                 (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_WRITE)) {
1913                 struct rbd_img_request *img_request = obj_request->img_request;
1914                 if (op_type == OBJ_OP_WRITE) {
1915                         rbd_assert(img_request_write_test(img_request));
1916                 } else {
1917                         rbd_assert(img_request_discard_test(img_request));
1918                 }
1919                 snapc = img_request->snapc;
1920         }
1921
1922         rbd_assert(num_ops == 1 || ((op_type == OBJ_OP_WRITE) && num_ops == 2));
1923
1924         /* Allocate and initialize the request, for the num_ops ops */
1925
1926         osdc = &rbd_dev->rbd_client->client->osdc;
1927         osd_req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false,
1928                                           GFP_NOIO);
1929         if (!osd_req)
1930                 goto fail;
1931
1932         if (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD)
1933                 osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
1934         else
1935                 osd_req->r_flags = CEPH_OSD_FLAG_READ;
1936
1937         osd_req->r_callback = rbd_osd_req_callback;
1938         osd_req->r_priv = obj_request;
1939
1940         osd_req->r_base_oloc.pool = ceph_file_layout_pg_pool(rbd_dev->layout);
1941         if (ceph_oid_aprintf(&osd_req->r_base_oid, GFP_NOIO, "%s",
1942                              obj_request->object_name))
1943                 goto fail;
1944
1945         if (ceph_osdc_alloc_messages(osd_req, GFP_NOIO))
1946                 goto fail;
1947
1948         return osd_req;
1949
1950 fail:
1951         ceph_osdc_put_request(osd_req);
1952         return NULL;
1953 }
1954
1955 /*
1956  * Create a copyup osd request based on the information in the object
1957  * request supplied.  A copyup request has two or three osd ops, a
1958  * copyup method call, potentially a hint op, and a write or truncate
1959  * or zero op.
1960  */
1961 static struct ceph_osd_request *
1962 rbd_osd_req_create_copyup(struct rbd_obj_request *obj_request)
1963 {
1964         struct rbd_img_request *img_request;
1965         struct ceph_snap_context *snapc;
1966         struct rbd_device *rbd_dev;
1967         struct ceph_osd_client *osdc;
1968         struct ceph_osd_request *osd_req;
1969         int num_osd_ops = 3;
1970
1971         rbd_assert(obj_request_img_data_test(obj_request));
1972         img_request = obj_request->img_request;
1973         rbd_assert(img_request);
1974         rbd_assert(img_request_write_test(img_request) ||
1975                         img_request_discard_test(img_request));
1976
1977         if (img_request_discard_test(img_request))
1978                 num_osd_ops = 2;
1979
1980         /* Allocate and initialize the request, for all the ops */
1981
1982         snapc = img_request->snapc;
1983         rbd_dev = img_request->rbd_dev;
1984         osdc = &rbd_dev->rbd_client->client->osdc;
1985         osd_req = ceph_osdc_alloc_request(osdc, snapc, num_osd_ops,
1986                                                 false, GFP_NOIO);
1987         if (!osd_req)
1988                 goto fail;
1989
1990         osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
1991         osd_req->r_callback = rbd_osd_req_callback;
1992         osd_req->r_priv = obj_request;
1993
1994         osd_req->r_base_oloc.pool = ceph_file_layout_pg_pool(rbd_dev->layout);
1995         if (ceph_oid_aprintf(&osd_req->r_base_oid, GFP_NOIO, "%s",
1996                              obj_request->object_name))
1997                 goto fail;
1998
1999         if (ceph_osdc_alloc_messages(osd_req, GFP_NOIO))
2000                 goto fail;
2001
2002         return osd_req;
2003
2004 fail:
2005         ceph_osdc_put_request(osd_req);
2006         return NULL;
2007 }
2008
2009
2010 static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req)
2011 {
2012         ceph_osdc_put_request(osd_req);
2013 }
2014
2015 /* object_name is assumed to be a non-null pointer and NUL-terminated */
2016
2017 static struct rbd_obj_request *rbd_obj_request_create(const char *object_name,
2018                                                 u64 offset, u64 length,
2019                                                 enum obj_request_type type)
2020 {
2021         struct rbd_obj_request *obj_request;
2022         size_t size;
2023         char *name;
2024
2025         rbd_assert(obj_request_type_valid(type));
2026
2027         size = strlen(object_name) + 1;
2028         name = kmalloc(size, GFP_NOIO);
2029         if (!name)
2030                 return NULL;
2031
2032         obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_NOIO);
2033         if (!obj_request) {
2034                 kfree(name);
2035                 return NULL;
2036         }
2037
2038         obj_request->object_name = memcpy(name, object_name, size);
2039         obj_request->offset = offset;
2040         obj_request->length = length;
2041         obj_request->flags = 0;
2042         obj_request->which = BAD_WHICH;
2043         obj_request->type = type;
2044         INIT_LIST_HEAD(&obj_request->links);
2045         init_completion(&obj_request->completion);
2046         kref_init(&obj_request->kref);
2047
2048         dout("%s: \"%s\" %llu/%llu %d -> obj %p\n", __func__, object_name,
2049                 offset, length, (int)type, obj_request);
2050
2051         return obj_request;
2052 }
2053
2054 static void rbd_obj_request_destroy(struct kref *kref)
2055 {
2056         struct rbd_obj_request *obj_request;
2057
2058         obj_request = container_of(kref, struct rbd_obj_request, kref);
2059
2060         dout("%s: obj %p\n", __func__, obj_request);
2061
2062         rbd_assert(obj_request->img_request == NULL);
2063         rbd_assert(obj_request->which == BAD_WHICH);
2064
2065         if (obj_request->osd_req)
2066                 rbd_osd_req_destroy(obj_request->osd_req);
2067
2068         rbd_assert(obj_request_type_valid(obj_request->type));
2069         switch (obj_request->type) {
2070         case OBJ_REQUEST_NODATA:
2071                 break;          /* Nothing to do */
2072         case OBJ_REQUEST_BIO:
2073                 if (obj_request->bio_list)
2074                         bio_chain_put(obj_request->bio_list);
2075                 break;
2076         case OBJ_REQUEST_PAGES:
2077                 if (obj_request->pages)
2078                         ceph_release_page_vector(obj_request->pages,
2079                                                 obj_request->page_count);
2080                 break;
2081         }
2082
2083         kfree(obj_request->object_name);
2084         obj_request->object_name = NULL;
2085         kmem_cache_free(rbd_obj_request_cache, obj_request);
2086 }
2087
2088 /* It's OK to call this for a device with no parent */
2089
2090 static void rbd_spec_put(struct rbd_spec *spec);
2091 static void rbd_dev_unparent(struct rbd_device *rbd_dev)
2092 {
2093         rbd_dev_remove_parent(rbd_dev);
2094         rbd_spec_put(rbd_dev->parent_spec);
2095         rbd_dev->parent_spec = NULL;
2096         rbd_dev->parent_overlap = 0;
2097 }
2098
2099 /*
2100  * Parent image reference counting is used to determine when an
2101  * image's parent fields can be safely torn down--after there are no
2102  * more in-flight requests to the parent image.  When the last
2103  * reference is dropped, cleaning them up is safe.
2104  */
2105 static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
2106 {
2107         int counter;
2108
2109         if (!rbd_dev->parent_spec)
2110                 return;
2111
2112         counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
2113         if (counter > 0)
2114                 return;
2115
2116         /* Last reference; clean up parent data structures */
2117
2118         if (!counter)
2119                 rbd_dev_unparent(rbd_dev);
2120         else
2121                 rbd_warn(rbd_dev, "parent reference underflow");
2122 }
2123
2124 /*
2125  * If an image has a non-zero parent overlap, get a reference to its
2126  * parent.
2127  *
2128  * Returns true if the rbd device has a parent with a non-zero
2129  * overlap and a reference for it was successfully taken, or
2130  * false otherwise.
2131  */
2132 static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
2133 {
2134         int counter = 0;
2135
2136         if (!rbd_dev->parent_spec)
2137                 return false;
2138
2139         down_read(&rbd_dev->header_rwsem);
2140         if (rbd_dev->parent_overlap)
2141                 counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
2142         up_read(&rbd_dev->header_rwsem);
2143
2144         if (counter < 0)
2145                 rbd_warn(rbd_dev, "parent reference overflow");
2146
2147         return counter > 0;
2148 }
2149
2150 /*
2151  * Caller is responsible for filling in the list of object requests
2152  * that comprises the image request, and the Linux request pointer
2153  * (if there is one).
2154  */
2155 static struct rbd_img_request *rbd_img_request_create(
2156                                         struct rbd_device *rbd_dev,
2157                                         u64 offset, u64 length,
2158                                         enum obj_operation_type op_type,
2159                                         struct ceph_snap_context *snapc)
2160 {
2161         struct rbd_img_request *img_request;
2162
2163         img_request = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
2164         if (!img_request)
2165                 return NULL;
2166
2167         img_request->rq = NULL;
2168         img_request->rbd_dev = rbd_dev;
2169         img_request->offset = offset;
2170         img_request->length = length;
2171         img_request->flags = 0;
2172         if (op_type == OBJ_OP_DISCARD) {
2173                 img_request_discard_set(img_request);
2174                 img_request->snapc = snapc;
2175         } else if (op_type == OBJ_OP_WRITE) {
2176                 img_request_write_set(img_request);
2177                 img_request->snapc = snapc;
2178         } else {
2179                 img_request->snap_id = rbd_dev->spec->snap_id;
2180         }
2181         if (rbd_dev_parent_get(rbd_dev))
2182                 img_request_layered_set(img_request);
2183         spin_lock_init(&img_request->completion_lock);
2184         img_request->next_completion = 0;
2185         img_request->callback = NULL;
2186         img_request->result = 0;
2187         img_request->obj_request_count = 0;
2188         INIT_LIST_HEAD(&img_request->obj_requests);
2189         kref_init(&img_request->kref);
2190
2191         dout("%s: rbd_dev %p %s %llu/%llu -> img %p\n", __func__, rbd_dev,
2192                 obj_op_name(op_type), offset, length, img_request);
2193
2194         return img_request;
2195 }
2196
2197 static void rbd_img_request_destroy(struct kref *kref)
2198 {
2199         struct rbd_img_request *img_request;
2200         struct rbd_obj_request *obj_request;
2201         struct rbd_obj_request *next_obj_request;
2202
2203         img_request = container_of(kref, struct rbd_img_request, kref);
2204
2205         dout("%s: img %p\n", __func__, img_request);
2206
2207         for_each_obj_request_safe(img_request, obj_request, next_obj_request)
2208                 rbd_img_obj_request_del(img_request, obj_request);
2209         rbd_assert(img_request->obj_request_count == 0);
2210
2211         if (img_request_layered_test(img_request)) {
2212                 img_request_layered_clear(img_request);
2213                 rbd_dev_parent_put(img_request->rbd_dev);
2214         }
2215
2216         if (img_request_write_test(img_request) ||
2217                 img_request_discard_test(img_request))
2218                 ceph_put_snap_context(img_request->snapc);
2219
2220         kmem_cache_free(rbd_img_request_cache, img_request);
2221 }
2222
2223 static struct rbd_img_request *rbd_parent_request_create(
2224                                         struct rbd_obj_request *obj_request,
2225                                         u64 img_offset, u64 length)
2226 {
2227         struct rbd_img_request *parent_request;
2228         struct rbd_device *rbd_dev;
2229
2230         rbd_assert(obj_request->img_request);
2231         rbd_dev = obj_request->img_request->rbd_dev;
2232
2233         parent_request = rbd_img_request_create(rbd_dev->parent, img_offset,
2234                                                 length, OBJ_OP_READ, NULL);
2235         if (!parent_request)
2236                 return NULL;
2237
2238         img_request_child_set(parent_request);
2239         rbd_obj_request_get(obj_request);
2240         parent_request->obj_request = obj_request;
2241
2242         return parent_request;
2243 }
2244
2245 static void rbd_parent_request_destroy(struct kref *kref)
2246 {
2247         struct rbd_img_request *parent_request;
2248         struct rbd_obj_request *orig_request;
2249
2250         parent_request = container_of(kref, struct rbd_img_request, kref);
2251         orig_request = parent_request->obj_request;
2252
2253         parent_request->obj_request = NULL;
2254         rbd_obj_request_put(orig_request);
2255         img_request_child_clear(parent_request);
2256
2257         rbd_img_request_destroy(kref);
2258 }
2259
2260 static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request)
2261 {
2262         struct rbd_img_request *img_request;
2263         unsigned int xferred;
2264         int result;
2265         bool more;
2266
2267         rbd_assert(obj_request_img_data_test(obj_request));
2268         img_request = obj_request->img_request;
2269
2270         rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
2271         xferred = (unsigned int)obj_request->xferred;
2272         result = obj_request->result;
2273         if (result) {
2274                 struct rbd_device *rbd_dev = img_request->rbd_dev;
2275                 enum obj_operation_type op_type;
2276
2277                 if (img_request_discard_test(img_request))
2278                         op_type = OBJ_OP_DISCARD;
2279                 else if (img_request_write_test(img_request))
2280                         op_type = OBJ_OP_WRITE;
2281                 else
2282                         op_type = OBJ_OP_READ;
2283
2284                 rbd_warn(rbd_dev, "%s %llx at %llx (%llx)",
2285                         obj_op_name(op_type), obj_request->length,
2286                         obj_request->img_offset, obj_request->offset);
2287                 rbd_warn(rbd_dev, "  result %d xferred %x",
2288                         result, xferred);
2289                 if (!img_request->result)
2290                         img_request->result = result;
2291                 /*
2292                  * Need to end I/O on the entire obj_request worth of
2293                  * bytes in case of error.
2294                  */
2295                 xferred = obj_request->length;
2296         }
2297
2298         /* Image object requests don't own their page array */
2299
2300         if (obj_request->type == OBJ_REQUEST_PAGES) {
2301                 obj_request->pages = NULL;
2302                 obj_request->page_count = 0;
2303         }
2304
2305         if (img_request_child_test(img_request)) {
2306                 rbd_assert(img_request->obj_request != NULL);
2307                 more = obj_request->which < img_request->obj_request_count - 1;
2308         } else {
2309                 rbd_assert(img_request->rq != NULL);
2310
2311                 more = blk_update_request(img_request->rq, result, xferred);
2312                 if (!more)
2313                         __blk_mq_end_request(img_request->rq, result);
2314         }
2315
2316         return more;
2317 }
2318
2319 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request)
2320 {
2321         struct rbd_img_request *img_request;
2322         u32 which = obj_request->which;
2323         bool more = true;
2324
2325         rbd_assert(obj_request_img_data_test(obj_request));
2326         img_request = obj_request->img_request;
2327
2328         dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
2329         rbd_assert(img_request != NULL);
2330         rbd_assert(img_request->obj_request_count > 0);
2331         rbd_assert(which != BAD_WHICH);
2332         rbd_assert(which < img_request->obj_request_count);
2333
2334         spin_lock_irq(&img_request->completion_lock);
2335         if (which != img_request->next_completion)
2336                 goto out;
2337
2338         for_each_obj_request_from(img_request, obj_request) {
2339                 rbd_assert(more);
2340                 rbd_assert(which < img_request->obj_request_count);
2341
2342                 if (!obj_request_done_test(obj_request))
2343                         break;
2344                 more = rbd_img_obj_end_request(obj_request);
2345                 which++;
2346         }
2347
2348         rbd_assert(more ^ (which == img_request->obj_request_count));
2349         img_request->next_completion = which;
2350 out:
2351         spin_unlock_irq(&img_request->completion_lock);
2352         rbd_img_request_put(img_request);
2353
2354         if (!more)
2355                 rbd_img_request_complete(img_request);
2356 }
2357
2358 /*
2359  * Add individual osd ops to the given ceph_osd_request and prepare
2360  * them for submission. num_ops is the current number of
2361  * osd operations already to the object request.
2362  */
2363 static void rbd_img_obj_request_fill(struct rbd_obj_request *obj_request,
2364                                 struct ceph_osd_request *osd_request,
2365                                 enum obj_operation_type op_type,
2366                                 unsigned int num_ops)
2367 {
2368         struct rbd_img_request *img_request = obj_request->img_request;
2369         struct rbd_device *rbd_dev = img_request->rbd_dev;
2370         u64 object_size = rbd_obj_bytes(&rbd_dev->header);
2371         u64 offset = obj_request->offset;
2372         u64 length = obj_request->length;
2373         u64 img_end;
2374         u16 opcode;
2375
2376         if (op_type == OBJ_OP_DISCARD) {
2377                 if (!offset && length == object_size &&
2378                     (!img_request_layered_test(img_request) ||
2379                      !obj_request_overlaps_parent(obj_request))) {
2380                         opcode = CEPH_OSD_OP_DELETE;
2381                 } else if ((offset + length == object_size)) {
2382                         opcode = CEPH_OSD_OP_TRUNCATE;
2383                 } else {
2384                         down_read(&rbd_dev->header_rwsem);
2385                         img_end = rbd_dev->header.image_size;
2386                         up_read(&rbd_dev->header_rwsem);
2387
2388                         if (obj_request->img_offset + length == img_end)
2389                                 opcode = CEPH_OSD_OP_TRUNCATE;
2390                         else
2391                                 opcode = CEPH_OSD_OP_ZERO;
2392                 }
2393         } else if (op_type == OBJ_OP_WRITE) {
2394                 if (!offset && length == object_size)
2395                         opcode = CEPH_OSD_OP_WRITEFULL;
2396                 else
2397                         opcode = CEPH_OSD_OP_WRITE;
2398                 osd_req_op_alloc_hint_init(osd_request, num_ops,
2399                                         object_size, object_size);
2400                 num_ops++;
2401         } else {
2402                 opcode = CEPH_OSD_OP_READ;
2403         }
2404
2405         if (opcode == CEPH_OSD_OP_DELETE)
2406                 osd_req_op_init(osd_request, num_ops, opcode, 0);
2407         else
2408                 osd_req_op_extent_init(osd_request, num_ops, opcode,
2409                                        offset, length, 0, 0);
2410
2411         if (obj_request->type == OBJ_REQUEST_BIO)
2412                 osd_req_op_extent_osd_data_bio(osd_request, num_ops,
2413                                         obj_request->bio_list, length);
2414         else if (obj_request->type == OBJ_REQUEST_PAGES)
2415                 osd_req_op_extent_osd_data_pages(osd_request, num_ops,
2416                                         obj_request->pages, length,
2417                                         offset & ~PAGE_MASK, false, false);
2418
2419         /* Discards are also writes */
2420         if (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD)
2421                 rbd_osd_req_format_write(obj_request);
2422         else
2423                 rbd_osd_req_format_read(obj_request);
2424 }
2425
2426 /*
2427  * Split up an image request into one or more object requests, each
2428  * to a different object.  The "type" parameter indicates whether
2429  * "data_desc" is the pointer to the head of a list of bio
2430  * structures, or the base of a page array.  In either case this
2431  * function assumes data_desc describes memory sufficient to hold
2432  * all data described by the image request.
2433  */
2434 static int rbd_img_request_fill(struct rbd_img_request *img_request,
2435                                         enum obj_request_type type,
2436                                         void *data_desc)
2437 {
2438         struct rbd_device *rbd_dev = img_request->rbd_dev;
2439         struct rbd_obj_request *obj_request = NULL;
2440         struct rbd_obj_request *next_obj_request;
2441         struct bio *bio_list = NULL;
2442         unsigned int bio_offset = 0;
2443         struct page **pages = NULL;
2444         enum obj_operation_type op_type;
2445         u64 img_offset;
2446         u64 resid;
2447
2448         dout("%s: img %p type %d data_desc %p\n", __func__, img_request,
2449                 (int)type, data_desc);
2450
2451         img_offset = img_request->offset;
2452         resid = img_request->length;
2453         rbd_assert(resid > 0);
2454         op_type = rbd_img_request_op_type(img_request);
2455
2456         if (type == OBJ_REQUEST_BIO) {
2457                 bio_list = data_desc;
2458                 rbd_assert(img_offset ==
2459                            bio_list->bi_iter.bi_sector << SECTOR_SHIFT);
2460         } else if (type == OBJ_REQUEST_PAGES) {
2461                 pages = data_desc;
2462         }
2463
2464         while (resid) {
2465                 struct ceph_osd_request *osd_req;
2466                 const char *object_name;
2467                 u64 offset;
2468                 u64 length;
2469
2470                 object_name = rbd_segment_name(rbd_dev, img_offset);
2471                 if (!object_name)
2472                         goto out_unwind;
2473                 offset = rbd_segment_offset(rbd_dev, img_offset);
2474                 length = rbd_segment_length(rbd_dev, img_offset, resid);
2475                 obj_request = rbd_obj_request_create(object_name,
2476                                                 offset, length, type);
2477                 /* object request has its own copy of the object name */
2478                 rbd_segment_name_free(object_name);
2479                 if (!obj_request)
2480                         goto out_unwind;
2481
2482                 /*
2483                  * set obj_request->img_request before creating the
2484                  * osd_request so that it gets the right snapc
2485                  */
2486                 rbd_img_obj_request_add(img_request, obj_request);
2487
2488                 if (type == OBJ_REQUEST_BIO) {
2489                         unsigned int clone_size;
2490
2491                         rbd_assert(length <= (u64)UINT_MAX);
2492                         clone_size = (unsigned int)length;
2493                         obj_request->bio_list =
2494                                         bio_chain_clone_range(&bio_list,
2495                                                                 &bio_offset,
2496                                                                 clone_size,
2497                                                                 GFP_NOIO);
2498                         if (!obj_request->bio_list)
2499                                 goto out_unwind;
2500                 } else if (type == OBJ_REQUEST_PAGES) {
2501                         unsigned int page_count;
2502
2503                         obj_request->pages = pages;
2504                         page_count = (u32)calc_pages_for(offset, length);
2505                         obj_request->page_count = page_count;
2506                         if ((offset + length) & ~PAGE_MASK)
2507                                 page_count--;   /* more on last page */
2508                         pages += page_count;
2509                 }
2510
2511                 osd_req = rbd_osd_req_create(rbd_dev, op_type,
2512                                         (op_type == OBJ_OP_WRITE) ? 2 : 1,
2513                                         obj_request);
2514                 if (!osd_req)
2515                         goto out_unwind;
2516
2517                 obj_request->osd_req = osd_req;
2518                 obj_request->callback = rbd_img_obj_callback;
2519                 obj_request->img_offset = img_offset;
2520
2521                 rbd_img_obj_request_fill(obj_request, osd_req, op_type, 0);
2522
2523                 rbd_img_request_get(img_request);
2524
2525                 img_offset += length;
2526                 resid -= length;
2527         }
2528
2529         return 0;
2530
2531 out_unwind:
2532         for_each_obj_request_safe(img_request, obj_request, next_obj_request)
2533                 rbd_img_obj_request_del(img_request, obj_request);
2534
2535         return -ENOMEM;
2536 }
2537
2538 static void
2539 rbd_osd_copyup_callback(struct rbd_obj_request *obj_request)
2540 {
2541         struct rbd_img_request *img_request;
2542         struct rbd_device *rbd_dev;
2543         struct page **pages;
2544         u32 page_count;
2545
2546         dout("%s: obj %p\n", __func__, obj_request);
2547
2548         rbd_assert(obj_request->type == OBJ_REQUEST_BIO ||
2549                 obj_request->type == OBJ_REQUEST_NODATA);
2550         rbd_assert(obj_request_img_data_test(obj_request));
2551         img_request = obj_request->img_request;
2552         rbd_assert(img_request);
2553
2554         rbd_dev = img_request->rbd_dev;
2555         rbd_assert(rbd_dev);
2556
2557         pages = obj_request->copyup_pages;
2558         rbd_assert(pages != NULL);
2559         obj_request->copyup_pages = NULL;
2560         page_count = obj_request->copyup_page_count;
2561         rbd_assert(page_count);
2562         obj_request->copyup_page_count = 0;
2563         ceph_release_page_vector(pages, page_count);
2564
2565         /*
2566          * We want the transfer count to reflect the size of the
2567          * original write request.  There is no such thing as a
2568          * successful short write, so if the request was successful
2569          * we can just set it to the originally-requested length.
2570          */
2571         if (!obj_request->result)
2572                 obj_request->xferred = obj_request->length;
2573
2574         obj_request_done_set(obj_request);
2575 }
2576
2577 static void
2578 rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request)
2579 {
2580         struct rbd_obj_request *orig_request;
2581         struct ceph_osd_request *osd_req;
2582         struct ceph_osd_client *osdc;
2583         struct rbd_device *rbd_dev;
2584         struct page **pages;
2585         enum obj_operation_type op_type;
2586         u32 page_count;
2587         int img_result;
2588         u64 parent_length;
2589
2590         rbd_assert(img_request_child_test(img_request));
2591
2592         /* First get what we need from the image request */
2593
2594         pages = img_request->copyup_pages;
2595         rbd_assert(pages != NULL);
2596         img_request->copyup_pages = NULL;
2597         page_count = img_request->copyup_page_count;
2598         rbd_assert(page_count);
2599         img_request->copyup_page_count = 0;
2600
2601         orig_request = img_request->obj_request;
2602         rbd_assert(orig_request != NULL);
2603         rbd_assert(obj_request_type_valid(orig_request->type));
2604         img_result = img_request->result;
2605         parent_length = img_request->length;
2606         rbd_assert(parent_length == img_request->xferred);
2607         rbd_img_request_put(img_request);
2608
2609         rbd_assert(orig_request->img_request);
2610         rbd_dev = orig_request->img_request->rbd_dev;
2611         rbd_assert(rbd_dev);
2612
2613         /*
2614          * If the overlap has become 0 (most likely because the
2615          * image has been flattened) we need to free the pages
2616          * and re-submit the original write request.
2617          */
2618         if (!rbd_dev->parent_overlap) {
2619                 struct ceph_osd_client *osdc;
2620
2621                 ceph_release_page_vector(pages, page_count);
2622                 osdc = &rbd_dev->rbd_client->client->osdc;
2623                 img_result = rbd_obj_request_submit(osdc, orig_request);
2624                 if (!img_result)
2625                         return;
2626         }
2627
2628         if (img_result)
2629                 goto out_err;
2630
2631         /*
2632          * The original osd request is of no use to use any more.
2633          * We need a new one that can hold the three ops in a copyup
2634          * request.  Allocate the new copyup osd request for the
2635          * original request, and release the old one.
2636          */
2637         img_result = -ENOMEM;
2638         osd_req = rbd_osd_req_create_copyup(orig_request);
2639         if (!osd_req)
2640                 goto out_err;
2641         rbd_osd_req_destroy(orig_request->osd_req);
2642         orig_request->osd_req = osd_req;
2643         orig_request->copyup_pages = pages;
2644         orig_request->copyup_page_count = page_count;
2645
2646         /* Initialize the copyup op */
2647
2648         osd_req_op_cls_init(osd_req, 0, CEPH_OSD_OP_CALL, "rbd", "copyup");
2649         osd_req_op_cls_request_data_pages(osd_req, 0, pages, parent_length, 0,
2650                                                 false, false);
2651
2652         /* Add the other op(s) */
2653
2654         op_type = rbd_img_request_op_type(orig_request->img_request);
2655         rbd_img_obj_request_fill(orig_request, osd_req, op_type, 1);
2656
2657         /* All set, send it off. */
2658
2659         osdc = &rbd_dev->rbd_client->client->osdc;
2660         img_result = rbd_obj_request_submit(osdc, orig_request);
2661         if (!img_result)
2662                 return;
2663 out_err:
2664         /* Record the error code and complete the request */
2665
2666         orig_request->result = img_result;
2667         orig_request->xferred = 0;
2668         obj_request_done_set(orig_request);
2669         rbd_obj_request_complete(orig_request);
2670 }
2671
2672 /*
2673  * Read from the parent image the range of data that covers the
2674  * entire target of the given object request.  This is used for
2675  * satisfying a layered image write request when the target of an
2676  * object request from the image request does not exist.
2677  *
2678  * A page array big enough to hold the returned data is allocated
2679  * and supplied to rbd_img_request_fill() as the "data descriptor."
2680  * When the read completes, this page array will be transferred to
2681  * the original object request for the copyup operation.
2682  *
2683  * If an error occurs, record it as the result of the original
2684  * object request and mark it done so it gets completed.
2685  */
2686 static int rbd_img_obj_parent_read_full(struct rbd_obj_request *obj_request)
2687 {
2688         struct rbd_img_request *img_request = NULL;
2689         struct rbd_img_request *parent_request = NULL;
2690         struct rbd_device *rbd_dev;
2691         u64 img_offset;
2692         u64 length;
2693         struct page **pages = NULL;
2694         u32 page_count;
2695         int result;
2696
2697         rbd_assert(obj_request_img_data_test(obj_request));
2698         rbd_assert(obj_request_type_valid(obj_request->type));
2699
2700         img_request = obj_request->img_request;
2701         rbd_assert(img_request != NULL);
2702         rbd_dev = img_request->rbd_dev;
2703         rbd_assert(rbd_dev->parent != NULL);
2704
2705         /*
2706          * Determine the byte range covered by the object in the
2707          * child image to which the original request was to be sent.
2708          */
2709         img_offset = obj_request->img_offset - obj_request->offset;
2710         length = (u64)1 << rbd_dev->header.obj_order;
2711
2712         /*
2713          * There is no defined parent data beyond the parent
2714          * overlap, so limit what we read at that boundary if
2715          * necessary.
2716          */
2717         if (img_offset + length > rbd_dev->parent_overlap) {
2718                 rbd_assert(img_offset < rbd_dev->parent_overlap);
2719                 length = rbd_dev->parent_overlap - img_offset;
2720         }
2721
2722         /*
2723          * Allocate a page array big enough to receive the data read
2724          * from the parent.
2725          */
2726         page_count = (u32)calc_pages_for(0, length);
2727         pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2728         if (IS_ERR(pages)) {
2729                 result = PTR_ERR(pages);
2730                 pages = NULL;
2731                 goto out_err;
2732         }
2733
2734         result = -ENOMEM;
2735         parent_request = rbd_parent_request_create(obj_request,
2736                                                 img_offset, length);
2737         if (!parent_request)
2738                 goto out_err;
2739
2740         result = rbd_img_request_fill(parent_request, OBJ_REQUEST_PAGES, pages);
2741         if (result)
2742                 goto out_err;
2743         parent_request->copyup_pages = pages;
2744         parent_request->copyup_page_count = page_count;
2745
2746         parent_request->callback = rbd_img_obj_parent_read_full_callback;
2747         result = rbd_img_request_submit(parent_request);
2748         if (!result)
2749                 return 0;
2750
2751         parent_request->copyup_pages = NULL;
2752         parent_request->copyup_page_count = 0;
2753         parent_request->obj_request = NULL;
2754         rbd_obj_request_put(obj_request);
2755 out_err:
2756         if (pages)
2757                 ceph_release_page_vector(pages, page_count);
2758         if (parent_request)
2759                 rbd_img_request_put(parent_request);
2760         obj_request->result = result;
2761         obj_request->xferred = 0;
2762         obj_request_done_set(obj_request);
2763
2764         return result;
2765 }
2766
2767 static void rbd_img_obj_exists_callback(struct rbd_obj_request *obj_request)
2768 {
2769         struct rbd_obj_request *orig_request;
2770         struct rbd_device *rbd_dev;
2771         int result;
2772
2773         rbd_assert(!obj_request_img_data_test(obj_request));
2774
2775         /*
2776          * All we need from the object request is the original
2777          * request and the result of the STAT op.  Grab those, then
2778          * we're done with the request.
2779          */
2780         orig_request = obj_request->obj_request;
2781         obj_request->obj_request = NULL;
2782         rbd_obj_request_put(orig_request);
2783         rbd_assert(orig_request);
2784         rbd_assert(orig_request->img_request);
2785
2786         result = obj_request->result;
2787         obj_request->result = 0;
2788
2789         dout("%s: obj %p for obj %p result %d %llu/%llu\n", __func__,
2790                 obj_request, orig_request, result,
2791                 obj_request->xferred, obj_request->length);
2792         rbd_obj_request_put(obj_request);
2793
2794         /*
2795          * If the overlap has become 0 (most likely because the
2796          * image has been flattened) we need to free the pages
2797          * and re-submit the original write request.
2798          */
2799         rbd_dev = orig_request->img_request->rbd_dev;
2800         if (!rbd_dev->parent_overlap) {
2801                 struct ceph_osd_client *osdc;
2802
2803                 osdc = &rbd_dev->rbd_client->client->osdc;
2804                 result = rbd_obj_request_submit(osdc, orig_request);
2805                 if (!result)
2806                         return;
2807         }
2808
2809         /*
2810          * Our only purpose here is to determine whether the object
2811          * exists, and we don't want to treat the non-existence as
2812          * an error.  If something else comes back, transfer the
2813          * error to the original request and complete it now.
2814          */
2815         if (!result) {
2816                 obj_request_existence_set(orig_request, true);
2817         } else if (result == -ENOENT) {
2818                 obj_request_existence_set(orig_request, false);
2819         } else if (result) {
2820                 orig_request->result = result;
2821                 goto out;
2822         }
2823
2824         /*
2825          * Resubmit the original request now that we have recorded
2826          * whether the target object exists.
2827          */
2828         orig_request->result = rbd_img_obj_request_submit(orig_request);
2829 out:
2830         if (orig_request->result)
2831                 rbd_obj_request_complete(orig_request);
2832 }
2833
2834 static int rbd_img_obj_exists_submit(struct rbd_obj_request *obj_request)
2835 {
2836         struct rbd_obj_request *stat_request;
2837         struct rbd_device *rbd_dev;
2838         struct ceph_osd_client *osdc;
2839         struct page **pages = NULL;
2840         u32 page_count;
2841         size_t size;
2842         int ret;
2843
2844         /*
2845          * The response data for a STAT call consists of:
2846          *     le64 length;
2847          *     struct {
2848          *         le32 tv_sec;
2849          *         le32 tv_nsec;
2850          *     } mtime;
2851          */
2852         size = sizeof (__le64) + sizeof (__le32) + sizeof (__le32);
2853         page_count = (u32)calc_pages_for(0, size);
2854         pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2855         if (IS_ERR(pages))
2856                 return PTR_ERR(pages);
2857
2858         ret = -ENOMEM;
2859         stat_request = rbd_obj_request_create(obj_request->object_name, 0, 0,
2860                                                         OBJ_REQUEST_PAGES);
2861         if (!stat_request)
2862                 goto out;
2863
2864         rbd_obj_request_get(obj_request);
2865         stat_request->obj_request = obj_request;
2866         stat_request->pages = pages;
2867         stat_request->page_count = page_count;
2868
2869         rbd_assert(obj_request->img_request);
2870         rbd_dev = obj_request->img_request->rbd_dev;
2871         stat_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1,
2872                                                    stat_request);
2873         if (!stat_request->osd_req)
2874                 goto out;
2875         stat_request->callback = rbd_img_obj_exists_callback;
2876
2877         osd_req_op_init(stat_request->osd_req, 0, CEPH_OSD_OP_STAT, 0);
2878         osd_req_op_raw_data_in_pages(stat_request->osd_req, 0, pages, size, 0,
2879                                         false, false);
2880         rbd_osd_req_format_read(stat_request);
2881
2882         osdc = &rbd_dev->rbd_client->client->osdc;
2883         ret = rbd_obj_request_submit(osdc, stat_request);
2884 out:
2885         if (ret)
2886                 rbd_obj_request_put(obj_request);
2887
2888         return ret;
2889 }
2890
2891 static bool img_obj_request_simple(struct rbd_obj_request *obj_request)
2892 {
2893         struct rbd_img_request *img_request;
2894         struct rbd_device *rbd_dev;
2895
2896         rbd_assert(obj_request_img_data_test(obj_request));
2897
2898         img_request = obj_request->img_request;
2899         rbd_assert(img_request);
2900         rbd_dev = img_request->rbd_dev;
2901
2902         /* Reads */
2903         if (!img_request_write_test(img_request) &&
2904             !img_request_discard_test(img_request))
2905                 return true;
2906
2907         /* Non-layered writes */
2908         if (!img_request_layered_test(img_request))
2909                 return true;
2910
2911         /*
2912          * Layered writes outside of the parent overlap range don't
2913          * share any data with the parent.
2914          */
2915         if (!obj_request_overlaps_parent(obj_request))
2916                 return true;
2917
2918         /*
2919          * Entire-object layered writes - we will overwrite whatever
2920          * parent data there is anyway.
2921          */
2922         if (!obj_request->offset &&
2923             obj_request->length == rbd_obj_bytes(&rbd_dev->header))
2924                 return true;
2925
2926         /*
2927          * If the object is known to already exist, its parent data has
2928          * already been copied.
2929          */
2930         if (obj_request_known_test(obj_request) &&
2931             obj_request_exists_test(obj_request))
2932                 return true;
2933
2934         return false;
2935 }
2936
2937 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request)
2938 {
2939         if (img_obj_request_simple(obj_request)) {
2940                 struct rbd_device *rbd_dev;
2941                 struct ceph_osd_client *osdc;
2942
2943                 rbd_dev = obj_request->img_request->rbd_dev;
2944                 osdc = &rbd_dev->rbd_client->client->osdc;
2945
2946                 return rbd_obj_request_submit(osdc, obj_request);
2947         }
2948
2949         /*
2950          * It's a layered write.  The target object might exist but
2951          * we may not know that yet.  If we know it doesn't exist,
2952          * start by reading the data for the full target object from
2953          * the parent so we can use it for a copyup to the target.
2954          */
2955         if (obj_request_known_test(obj_request))
2956                 return rbd_img_obj_parent_read_full(obj_request);
2957
2958         /* We don't know whether the target exists.  Go find out. */
2959
2960         return rbd_img_obj_exists_submit(obj_request);
2961 }
2962
2963 static int rbd_img_request_submit(struct rbd_img_request *img_request)
2964 {
2965         struct rbd_obj_request *obj_request;
2966         struct rbd_obj_request *next_obj_request;
2967         int ret = 0;
2968
2969         dout("%s: img %p\n", __func__, img_request);
2970
2971         rbd_img_request_get(img_request);
2972         for_each_obj_request_safe(img_request, obj_request, next_obj_request) {
2973                 ret = rbd_img_obj_request_submit(obj_request);
2974                 if (ret)
2975                         goto out_put_ireq;
2976         }
2977
2978 out_put_ireq:
2979         rbd_img_request_put(img_request);
2980         return ret;
2981 }
2982
2983 static void rbd_img_parent_read_callback(struct rbd_img_request *img_request)
2984 {
2985         struct rbd_obj_request *obj_request;
2986         struct rbd_device *rbd_dev;
2987         u64 obj_end;
2988         u64 img_xferred;
2989         int img_result;
2990
2991         rbd_assert(img_request_child_test(img_request));
2992
2993         /* First get what we need from the image request and release it */
2994
2995         obj_request = img_request->obj_request;
2996         img_xferred = img_request->xferred;
2997         img_result = img_request->result;
2998         rbd_img_request_put(img_request);
2999
3000         /*
3001          * If the overlap has become 0 (most likely because the
3002          * image has been flattened) we need to re-submit the
3003          * original request.
3004          */
3005         rbd_assert(obj_request);
3006         rbd_assert(obj_request->img_request);
3007         rbd_dev = obj_request->img_request->rbd_dev;
3008         if (!rbd_dev->parent_overlap) {
3009                 struct ceph_osd_client *osdc;
3010
3011                 osdc = &rbd_dev->rbd_client->client->osdc;
3012                 img_result = rbd_obj_request_submit(osdc, obj_request);
3013                 if (!img_result)
3014                         return;
3015         }
3016
3017         obj_request->result = img_result;
3018         if (obj_request->result)
3019                 goto out;
3020
3021         /*
3022          * We need to zero anything beyond the parent overlap
3023          * boundary.  Since rbd_img_obj_request_read_callback()
3024          * will zero anything beyond the end of a short read, an
3025          * easy way to do this is to pretend the data from the
3026          * parent came up short--ending at the overlap boundary.
3027          */
3028         rbd_assert(obj_request->img_offset < U64_MAX - obj_request->length);
3029         obj_end = obj_request->img_offset + obj_request->length;
3030         if (obj_end > rbd_dev->parent_overlap) {
3031                 u64 xferred = 0;
3032
3033                 if (obj_request->img_offset < rbd_dev->parent_overlap)
3034                         xferred = rbd_dev->parent_overlap -
3035                                         obj_request->img_offset;
3036
3037                 obj_request->xferred = min(img_xferred, xferred);
3038         } else {
3039                 obj_request->xferred = img_xferred;
3040         }
3041 out:
3042         rbd_img_obj_request_read_callback(obj_request);
3043         rbd_obj_request_complete(obj_request);
3044 }
3045
3046 static void rbd_img_parent_read(struct rbd_obj_request *obj_request)
3047 {
3048         struct rbd_img_request *img_request;
3049         int result;
3050
3051         rbd_assert(obj_request_img_data_test(obj_request));
3052         rbd_assert(obj_request->img_request != NULL);
3053         rbd_assert(obj_request->result == (s32) -ENOENT);
3054         rbd_assert(obj_request_type_valid(obj_request->type));
3055
3056         /* rbd_read_finish(obj_request, obj_request->length); */
3057         img_request = rbd_parent_request_create(obj_request,
3058                                                 obj_request->img_offset,
3059                                                 obj_request->length);
3060         result = -ENOMEM;
3061         if (!img_request)
3062                 goto out_err;
3063
3064         if (obj_request->type == OBJ_REQUEST_BIO)
3065                 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
3066                                                 obj_request->bio_list);
3067         else
3068                 result = rbd_img_request_fill(img_request, OBJ_REQUEST_PAGES,
3069                                                 obj_request->pages);
3070         if (result)
3071                 goto out_err;
3072
3073         img_request->callback = rbd_img_parent_read_callback;
3074         result = rbd_img_request_submit(img_request);
3075         if (result)
3076                 goto out_err;
3077
3078         return;
3079 out_err:
3080         if (img_request)
3081                 rbd_img_request_put(img_request);
3082         obj_request->result = result;
3083         obj_request->xferred = 0;
3084         obj_request_done_set(obj_request);
3085 }
3086
3087 static int rbd_dev_header_watch_sync(struct rbd_device *rbd_dev);
3088 static void __rbd_dev_header_unwatch_sync(struct rbd_device *rbd_dev);
3089
3090 static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie,
3091                          u64 notifier_id, void *data, size_t data_len)
3092 {
3093         struct rbd_device *rbd_dev = arg;
3094         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3095         int ret;
3096
3097         dout("%s rbd_dev %p cookie %llu notify_id %llu\n", __func__, rbd_dev,
3098              cookie, notify_id);
3099
3100         /*
3101          * Until adequate refresh error handling is in place, there is
3102          * not much we can do here, except warn.
3103          *
3104          * See http://tracker.ceph.com/issues/5040
3105          */
3106         ret = rbd_dev_refresh(rbd_dev);
3107         if (ret)
3108                 rbd_warn(rbd_dev, "refresh failed: %d", ret);
3109
3110         ret = ceph_osdc_notify_ack(osdc, &rbd_dev->header_oid,
3111                                    &rbd_dev->header_oloc, notify_id, cookie,
3112                                    NULL, 0);
3113         if (ret)
3114                 rbd_warn(rbd_dev, "notify_ack ret %d", ret);
3115 }
3116
3117 static void rbd_watch_errcb(void *arg, u64 cookie, int err)
3118 {
3119         struct rbd_device *rbd_dev = arg;
3120         int ret;
3121
3122         rbd_warn(rbd_dev, "encountered watch error: %d", err);
3123
3124         __rbd_dev_header_unwatch_sync(rbd_dev);
3125
3126         ret = rbd_dev_header_watch_sync(rbd_dev);
3127         if (ret) {
3128                 rbd_warn(rbd_dev, "failed to reregister watch: %d", ret);
3129                 return;
3130         }
3131
3132         ret = rbd_dev_refresh(rbd_dev);
3133         if (ret)
3134                 rbd_warn(rbd_dev, "reregisteration refresh failed: %d", ret);
3135 }
3136
3137 /*
3138  * Initiate a watch request, synchronously.
3139  */
3140 static int rbd_dev_header_watch_sync(struct rbd_device *rbd_dev)
3141 {
3142         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3143         struct ceph_osd_linger_request *handle;
3144
3145         rbd_assert(!rbd_dev->watch_handle);
3146
3147         handle = ceph_osdc_watch(osdc, &rbd_dev->header_oid,
3148                                  &rbd_dev->header_oloc, rbd_watch_cb,
3149                                  rbd_watch_errcb, rbd_dev);
3150         if (IS_ERR(handle))
3151                 return PTR_ERR(handle);
3152
3153         rbd_dev->watch_handle = handle;
3154         return 0;
3155 }
3156
3157 static void __rbd_dev_header_unwatch_sync(struct rbd_device *rbd_dev)
3158 {
3159         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3160         int ret;
3161
3162         if (!rbd_dev->watch_handle)
3163                 return;
3164
3165         ret = ceph_osdc_unwatch(osdc, rbd_dev->watch_handle);
3166         if (ret)
3167                 rbd_warn(rbd_dev, "failed to unwatch: %d", ret);
3168
3169         rbd_dev->watch_handle = NULL;
3170 }
3171
3172 /*
3173  * Tear down a watch request, synchronously.
3174  */
3175 static void rbd_dev_header_unwatch_sync(struct rbd_device *rbd_dev)
3176 {
3177         __rbd_dev_header_unwatch_sync(rbd_dev);
3178
3179         dout("%s flushing notifies\n", __func__);
3180         ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
3181 }
3182
3183 /*
3184  * Synchronous osd object method call.  Returns the number of bytes
3185  * returned in the outbound buffer, or a negative error code.
3186  */
3187 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
3188                              const char *object_name,
3189                              const char *class_name,
3190                              const char *method_name,
3191                              const void *outbound,
3192                              size_t outbound_size,
3193                              void *inbound,
3194                              size_t inbound_size)
3195 {
3196         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3197         struct rbd_obj_request *obj_request;
3198         struct page **pages;
3199         u32 page_count;
3200         int ret;
3201
3202         /*
3203          * Method calls are ultimately read operations.  The result
3204          * should placed into the inbound buffer provided.  They
3205          * also supply outbound data--parameters for the object
3206          * method.  Currently if this is present it will be a
3207          * snapshot id.
3208          */
3209         page_count = (u32)calc_pages_for(0, inbound_size);
3210         pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
3211         if (IS_ERR(pages))
3212                 return PTR_ERR(pages);
3213
3214         ret = -ENOMEM;
3215         obj_request = rbd_obj_request_create(object_name, 0, inbound_size,
3216                                                         OBJ_REQUEST_PAGES);
3217         if (!obj_request)
3218                 goto out;
3219
3220         obj_request->pages = pages;
3221         obj_request->page_count = page_count;
3222
3223         obj_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1,
3224                                                   obj_request);
3225         if (!obj_request->osd_req)
3226                 goto out;
3227
3228         osd_req_op_cls_init(obj_request->osd_req, 0, CEPH_OSD_OP_CALL,
3229                                         class_name, method_name);
3230         if (outbound_size) {
3231                 struct ceph_pagelist *pagelist;
3232
3233                 pagelist = kmalloc(sizeof (*pagelist), GFP_NOFS);
3234                 if (!pagelist)
3235                         goto out;
3236
3237                 ceph_pagelist_init(pagelist);
3238                 ceph_pagelist_append(pagelist, outbound, outbound_size);
3239                 osd_req_op_cls_request_data_pagelist(obj_request->osd_req, 0,
3240                                                 pagelist);
3241         }
3242         osd_req_op_cls_response_data_pages(obj_request->osd_req, 0,
3243                                         obj_request->pages, inbound_size,
3244                                         0, false, false);
3245         rbd_osd_req_format_read(obj_request);
3246
3247         ret = rbd_obj_request_submit(osdc, obj_request);
3248         if (ret)
3249                 goto out;
3250         ret = rbd_obj_request_wait(obj_request);
3251         if (ret)
3252                 goto out;
3253
3254         ret = obj_request->result;
3255         if (ret < 0)
3256                 goto out;
3257
3258         rbd_assert(obj_request->xferred < (u64)INT_MAX);
3259         ret = (int)obj_request->xferred;
3260         ceph_copy_from_page_vector(pages, inbound, 0, obj_request->xferred);
3261 out:
3262         if (obj_request)
3263                 rbd_obj_request_put(obj_request);
3264         else
3265                 ceph_release_page_vector(pages, page_count);
3266
3267         return ret;
3268 }
3269
3270 static void rbd_queue_workfn(struct work_struct *work)
3271 {
3272         struct request *rq = blk_mq_rq_from_pdu(work);
3273         struct rbd_device *rbd_dev = rq->q->queuedata;
3274         struct rbd_img_request *img_request;
3275         struct ceph_snap_context *snapc = NULL;
3276         u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
3277         u64 length = blk_rq_bytes(rq);
3278         enum obj_operation_type op_type;
3279         u64 mapping_size;
3280         int result;
3281
3282         if (rq->cmd_type != REQ_TYPE_FS) {
3283                 dout("%s: non-fs request type %d\n", __func__,
3284                         (int) rq->cmd_type);
3285                 result = -EIO;
3286                 goto err;
3287         }
3288
3289         if (req_op(rq) == REQ_OP_DISCARD)
3290                 op_type = OBJ_OP_DISCARD;
3291         else if (req_op(rq) == REQ_OP_WRITE)
3292                 op_type = OBJ_OP_WRITE;
3293         else
3294                 op_type = OBJ_OP_READ;
3295
3296         /* Ignore/skip any zero-length requests */
3297
3298         if (!length) {
3299                 dout("%s: zero-length request\n", __func__);
3300                 result = 0;
3301                 goto err_rq;
3302         }
3303
3304         /* Only reads are allowed to a read-only device */
3305
3306         if (op_type != OBJ_OP_READ) {
3307                 if (rbd_dev->mapping.read_only) {
3308                         result = -EROFS;
3309                         goto err_rq;
3310                 }
3311                 rbd_assert(rbd_dev->spec->snap_id == CEPH_NOSNAP);
3312         }
3313
3314         /*
3315          * Quit early if the mapped snapshot no longer exists.  It's
3316          * still possible the snapshot will have disappeared by the
3317          * time our request arrives at the osd, but there's no sense in
3318          * sending it if we already know.
3319          */
3320         if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags)) {
3321                 dout("request for non-existent snapshot");
3322                 rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
3323                 result = -ENXIO;
3324                 goto err_rq;
3325         }
3326
3327         if (offset && length > U64_MAX - offset + 1) {
3328                 rbd_warn(rbd_dev, "bad request range (%llu~%llu)", offset,
3329                          length);
3330                 result = -EINVAL;
3331                 goto err_rq;    /* Shouldn't happen */
3332         }
3333
3334         blk_mq_start_request(rq);
3335
3336         down_read(&rbd_dev->header_rwsem);
3337         mapping_size = rbd_dev->mapping.size;
3338         if (op_type != OBJ_OP_READ) {
3339                 snapc = rbd_dev->header.snapc;
3340                 ceph_get_snap_context(snapc);
3341         }
3342         up_read(&rbd_dev->header_rwsem);
3343
3344         if (offset + length > mapping_size) {
3345                 rbd_warn(rbd_dev, "beyond EOD (%llu~%llu > %llu)", offset,
3346                          length, mapping_size);
3347                 result = -EIO;
3348                 goto err_rq;
3349         }
3350
3351         img_request = rbd_img_request_create(rbd_dev, offset, length, op_type,
3352                                              snapc);
3353         if (!img_request) {
3354                 result = -ENOMEM;
3355                 goto err_rq;
3356         }
3357         img_request->rq = rq;
3358         snapc = NULL; /* img_request consumes a ref */
3359
3360         if (op_type == OBJ_OP_DISCARD)
3361                 result = rbd_img_request_fill(img_request, OBJ_REQUEST_NODATA,
3362                                               NULL);
3363         else
3364                 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
3365                                               rq->bio);
3366         if (result)
3367                 goto err_img_request;
3368
3369         result = rbd_img_request_submit(img_request);
3370         if (result)
3371                 goto err_img_request;
3372
3373         return;
3374
3375 err_img_request:
3376         rbd_img_request_put(img_request);
3377 err_rq:
3378         if (result)
3379                 rbd_warn(rbd_dev, "%s %llx at %llx result %d",
3380                          obj_op_name(op_type), length, offset, result);
3381         ceph_put_snap_context(snapc);
3382 err:
3383         blk_mq_end_request(rq, result);
3384 }
3385
3386 static int rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
3387                 const struct blk_mq_queue_data *bd)
3388 {
3389         struct request *rq = bd->rq;
3390         struct work_struct *work = blk_mq_rq_to_pdu(rq);
3391
3392         queue_work(rbd_wq, work);
3393         return BLK_MQ_RQ_QUEUE_OK;
3394 }
3395
3396 static void rbd_free_disk(struct rbd_device *rbd_dev)
3397 {
3398         struct gendisk *disk = rbd_dev->disk;
3399
3400         if (!disk)
3401                 return;
3402
3403         rbd_dev->disk = NULL;
3404         if (disk->flags & GENHD_FL_UP) {
3405                 del_gendisk(disk);
3406                 if (disk->queue)
3407                         blk_cleanup_queue(disk->queue);
3408                 blk_mq_free_tag_set(&rbd_dev->tag_set);
3409         }
3410         put_disk(disk);
3411 }
3412
3413 static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
3414                                 const char *object_name,
3415                                 u64 offset, u64 length, void *buf)
3416
3417 {
3418         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3419         struct rbd_obj_request *obj_request;
3420         struct page **pages = NULL;
3421         u32 page_count;
3422         size_t size;
3423         int ret;
3424
3425         page_count = (u32) calc_pages_for(offset, length);
3426         pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
3427         if (IS_ERR(pages))
3428                 return PTR_ERR(pages);
3429
3430         ret = -ENOMEM;
3431         obj_request = rbd_obj_request_create(object_name, offset, length,
3432                                                         OBJ_REQUEST_PAGES);
3433         if (!obj_request)
3434                 goto out;
3435
3436         obj_request->pages = pages;
3437         obj_request->page_count = page_count;
3438
3439         obj_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1,
3440                                                   obj_request);
3441         if (!obj_request->osd_req)
3442                 goto out;
3443
3444         osd_req_op_extent_init(obj_request->osd_req, 0, CEPH_OSD_OP_READ,
3445                                         offset, length, 0, 0);
3446         osd_req_op_extent_osd_data_pages(obj_request->osd_req, 0,
3447                                         obj_request->pages,
3448                                         obj_request->length,
3449                                         obj_request->offset & ~PAGE_MASK,
3450                                         false, false);
3451         rbd_osd_req_format_read(obj_request);
3452
3453         ret = rbd_obj_request_submit(osdc, obj_request);
3454         if (ret)
3455                 goto out;
3456         ret = rbd_obj_request_wait(obj_request);
3457         if (ret)
3458                 goto out;
3459
3460         ret = obj_request->result;
3461         if (ret < 0)
3462                 goto out;
3463
3464         rbd_assert(obj_request->xferred <= (u64) SIZE_MAX);
3465         size = (size_t) obj_request->xferred;
3466         ceph_copy_from_page_vector(pages, buf, 0, size);
3467         rbd_assert(size <= (size_t)INT_MAX);
3468         ret = (int)size;
3469 out:
3470         if (obj_request)
3471                 rbd_obj_request_put(obj_request);
3472         else
3473                 ceph_release_page_vector(pages, page_count);
3474
3475         return ret;
3476 }
3477
3478 /*
3479  * Read the complete header for the given rbd device.  On successful
3480  * return, the rbd_dev->header field will contain up-to-date
3481  * information about the image.
3482  */
3483 static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev)
3484 {
3485         struct rbd_image_header_ondisk *ondisk = NULL;
3486         u32 snap_count = 0;
3487         u64 names_size = 0;
3488         u32 want_count;
3489         int ret;
3490
3491         /*
3492          * The complete header will include an array of its 64-bit
3493          * snapshot ids, followed by the names of those snapshots as
3494          * a contiguous block of NUL-terminated strings.  Note that
3495          * the number of snapshots could change by the time we read
3496          * it in, in which case we re-read it.
3497          */
3498         do {
3499                 size_t size;
3500
3501                 kfree(ondisk);
3502
3503                 size = sizeof (*ondisk);
3504                 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
3505                 size += names_size;
3506                 ondisk = kmalloc(size, GFP_KERNEL);
3507                 if (!ondisk)
3508                         return -ENOMEM;
3509
3510                 ret = rbd_obj_read_sync(rbd_dev, rbd_dev->header_oid.name,
3511                                        0, size, ondisk);
3512                 if (ret < 0)
3513                         goto out;
3514                 if ((size_t)ret < size) {
3515                         ret = -ENXIO;
3516                         rbd_warn(rbd_dev, "short header read (want %zd got %d)",
3517                                 size, ret);
3518                         goto out;
3519                 }
3520                 if (!rbd_dev_ondisk_valid(ondisk)) {
3521                         ret = -ENXIO;
3522                         rbd_warn(rbd_dev, "invalid header");
3523                         goto out;
3524                 }
3525
3526                 names_size = le64_to_cpu(ondisk->snap_names_len);
3527                 want_count = snap_count;
3528                 snap_count = le32_to_cpu(ondisk->snap_count);
3529         } while (snap_count != want_count);
3530
3531         ret = rbd_header_from_disk(rbd_dev, ondisk);
3532 out:
3533         kfree(ondisk);
3534
3535         return ret;
3536 }
3537
3538 /*
3539  * Clear the rbd device's EXISTS flag if the snapshot it's mapped to
3540  * has disappeared from the (just updated) snapshot context.
3541  */
3542 static void rbd_exists_validate(struct rbd_device *rbd_dev)
3543 {
3544         u64 snap_id;
3545
3546         if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags))
3547                 return;
3548
3549         snap_id = rbd_dev->spec->snap_id;
3550         if (snap_id == CEPH_NOSNAP)
3551                 return;
3552
3553         if (rbd_dev_snap_index(rbd_dev, snap_id) == BAD_SNAP_INDEX)
3554                 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
3555 }
3556
3557 static void rbd_dev_update_size(struct rbd_device *rbd_dev)
3558 {
3559         sector_t size;
3560
3561         /*
3562          * If EXISTS is not set, rbd_dev->disk may be NULL, so don't
3563          * try to update its size.  If REMOVING is set, updating size
3564          * is just useless work since the device can't be opened.
3565          */
3566         if (test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags) &&
3567             !test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) {
3568                 size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
3569                 dout("setting size to %llu sectors", (unsigned long long)size);
3570                 set_capacity(rbd_dev->disk, size);
3571                 revalidate_disk(rbd_dev->disk);
3572         }
3573 }
3574
3575 static int rbd_dev_refresh(struct rbd_device *rbd_dev)
3576 {
3577         u64 mapping_size;
3578         int ret;
3579
3580         down_write(&rbd_dev->header_rwsem);
3581         mapping_size = rbd_dev->mapping.size;
3582
3583         ret = rbd_dev_header_info(rbd_dev);
3584         if (ret)
3585                 goto out;
3586
3587         /*
3588          * If there is a parent, see if it has disappeared due to the
3589          * mapped image getting flattened.
3590          */
3591         if (rbd_dev->parent) {
3592                 ret = rbd_dev_v2_parent_info(rbd_dev);
3593                 if (ret)
3594                         goto out;
3595         }
3596
3597         if (rbd_dev->spec->snap_id == CEPH_NOSNAP) {
3598                 rbd_dev->mapping.size = rbd_dev->header.image_size;
3599         } else {
3600                 /* validate mapped snapshot's EXISTS flag */
3601                 rbd_exists_validate(rbd_dev);
3602         }
3603
3604 out:
3605         up_write(&rbd_dev->header_rwsem);
3606         if (!ret && mapping_size != rbd_dev->mapping.size)
3607                 rbd_dev_update_size(rbd_dev);
3608
3609         return ret;
3610 }
3611
3612 static int rbd_init_request(void *data, struct request *rq,
3613                 unsigned int hctx_idx, unsigned int request_idx,
3614                 unsigned int numa_node)
3615 {
3616         struct work_struct *work = blk_mq_rq_to_pdu(rq);
3617
3618         INIT_WORK(work, rbd_queue_workfn);
3619         return 0;
3620 }
3621
3622 static struct blk_mq_ops rbd_mq_ops = {
3623         .queue_rq       = rbd_queue_rq,
3624         .map_queue      = blk_mq_map_queue,
3625         .init_request   = rbd_init_request,
3626 };
3627
3628 static int rbd_init_disk(struct rbd_device *rbd_dev)
3629 {
3630         struct gendisk *disk;
3631         struct request_queue *q;
3632         u64 segment_size;
3633         int err;
3634
3635         /* create gendisk info */
3636         disk = alloc_disk(single_major ?
3637                           (1 << RBD_SINGLE_MAJOR_PART_SHIFT) :
3638                           RBD_MINORS_PER_MAJOR);
3639         if (!disk)
3640                 return -ENOMEM;
3641
3642         snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
3643                  rbd_dev->dev_id);
3644         disk->major = rbd_dev->major;
3645         disk->first_minor = rbd_dev->minor;
3646         if (single_major)
3647                 disk->flags |= GENHD_FL_EXT_DEVT;
3648         disk->fops = &rbd_bd_ops;
3649         disk->private_data = rbd_dev;
3650
3651         memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
3652         rbd_dev->tag_set.ops = &rbd_mq_ops;
3653         rbd_dev->tag_set.queue_depth = rbd_dev->opts->queue_depth;
3654         rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
3655         rbd_dev->tag_set.flags = BLK_MQ_F_SHOULD_MERGE | BLK_MQ_F_SG_MERGE;
3656         rbd_dev->tag_set.nr_hw_queues = 1;
3657         rbd_dev->tag_set.cmd_size = sizeof(struct work_struct);
3658
3659         err = blk_mq_alloc_tag_set(&rbd_dev->tag_set);
3660         if (err)
3661                 goto out_disk;
3662
3663         q = blk_mq_init_queue(&rbd_dev->tag_set);
3664         if (IS_ERR(q)) {
3665                 err = PTR_ERR(q);
3666                 goto out_tag_set;
3667         }
3668
3669         queue_flag_set_unlocked(QUEUE_FLAG_NONROT, q);
3670         /* QUEUE_FLAG_ADD_RANDOM is off by default for blk-mq */
3671
3672         /* set io sizes to object size */
3673         segment_size = rbd_obj_bytes(&rbd_dev->header);
3674         blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE);
3675         q->limits.max_sectors = queue_max_hw_sectors(q);
3676         blk_queue_max_segments(q, segment_size / SECTOR_SIZE);
3677         blk_queue_max_segment_size(q, segment_size);
3678         blk_queue_io_min(q, segment_size);
3679         blk_queue_io_opt(q, segment_size);
3680
3681         /* enable the discard support */
3682         queue_flag_set_unlocked(QUEUE_FLAG_DISCARD, q);
3683         q->limits.discard_granularity = segment_size;
3684         q->limits.discard_alignment = segment_size;
3685         blk_queue_max_discard_sectors(q, segment_size / SECTOR_SIZE);
3686         q->limits.discard_zeroes_data = 1;
3687
3688         if (!ceph_test_opt(rbd_dev->rbd_client->client, NOCRC))
3689                 q->backing_dev_info.capabilities |= BDI_CAP_STABLE_WRITES;
3690
3691         disk->queue = q;
3692
3693         q->queuedata = rbd_dev;
3694
3695         rbd_dev->disk = disk;
3696
3697         return 0;
3698 out_tag_set:
3699         blk_mq_free_tag_set(&rbd_dev->tag_set);
3700 out_disk:
3701         put_disk(disk);
3702         return err;
3703 }
3704
3705 /*
3706   sysfs
3707 */
3708
3709 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
3710 {
3711         return container_of(dev, struct rbd_device, dev);
3712 }
3713
3714 static ssize_t rbd_size_show(struct device *dev,
3715                              struct device_attribute *attr, char *buf)
3716 {
3717         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3718
3719         return sprintf(buf, "%llu\n",
3720                 (unsigned long long)rbd_dev->mapping.size);
3721 }
3722
3723 /*
3724  * Note this shows the features for whatever's mapped, which is not
3725  * necessarily the base image.
3726  */
3727 static ssize_t rbd_features_show(struct device *dev,
3728                              struct device_attribute *attr, char *buf)
3729 {
3730         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3731
3732         return sprintf(buf, "0x%016llx\n",
3733                         (unsigned long long)rbd_dev->mapping.features);
3734 }
3735
3736 static ssize_t rbd_major_show(struct device *dev,
3737                               struct device_attribute *attr, char *buf)
3738 {
3739         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3740
3741         if (rbd_dev->major)
3742                 return sprintf(buf, "%d\n", rbd_dev->major);
3743
3744         return sprintf(buf, "(none)\n");
3745 }
3746
3747 static ssize_t rbd_minor_show(struct device *dev,
3748                               struct device_attribute *attr, char *buf)
3749 {
3750         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3751
3752         return sprintf(buf, "%d\n", rbd_dev->minor);
3753 }
3754
3755 static ssize_t rbd_client_id_show(struct device *dev,
3756                                   struct device_attribute *attr, char *buf)
3757 {
3758         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3759
3760         return sprintf(buf, "client%lld\n",
3761                         ceph_client_id(rbd_dev->rbd_client->client));
3762 }
3763
3764 static ssize_t rbd_pool_show(struct device *dev,
3765                              struct device_attribute *attr, char *buf)
3766 {
3767         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3768
3769         return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
3770 }
3771
3772 static ssize_t rbd_pool_id_show(struct device *dev,
3773                              struct device_attribute *attr, char *buf)
3774 {
3775         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3776
3777         return sprintf(buf, "%llu\n",
3778                         (unsigned long long) rbd_dev->spec->pool_id);
3779 }
3780
3781 static ssize_t rbd_name_show(struct device *dev,
3782                              struct device_attribute *attr, char *buf)
3783 {
3784         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3785
3786         if (rbd_dev->spec->image_name)
3787                 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
3788
3789         return sprintf(buf, "(unknown)\n");
3790 }
3791
3792 static ssize_t rbd_image_id_show(struct device *dev,
3793                              struct device_attribute *attr, char *buf)
3794 {
3795         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3796
3797         return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
3798 }
3799
3800 /*
3801  * Shows the name of the currently-mapped snapshot (or
3802  * RBD_SNAP_HEAD_NAME for the base image).
3803  */
3804 static ssize_t rbd_snap_show(struct device *dev,
3805                              struct device_attribute *attr,
3806                              char *buf)
3807 {
3808         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3809
3810         return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
3811 }
3812
3813 /*
3814  * For a v2 image, shows the chain of parent images, separated by empty
3815  * lines.  For v1 images or if there is no parent, shows "(no parent
3816  * image)".
3817  */
3818 static ssize_t rbd_parent_show(struct device *dev,
3819                                struct device_attribute *attr,
3820                                char *buf)
3821 {
3822         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3823         ssize_t count = 0;
3824
3825         if (!rbd_dev->parent)
3826                 return sprintf(buf, "(no parent image)\n");
3827
3828         for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) {
3829                 struct rbd_spec *spec = rbd_dev->parent_spec;
3830
3831                 count += sprintf(&buf[count], "%s"
3832                             "pool_id %llu\npool_name %s\n"
3833                             "image_id %s\nimage_name %s\n"
3834                             "snap_id %llu\nsnap_name %s\n"
3835                             "overlap %llu\n",
3836                             !count ? "" : "\n", /* first? */
3837                             spec->pool_id, spec->pool_name,
3838                             spec->image_id, spec->image_name ?: "(unknown)",
3839                             spec->snap_id, spec->snap_name,
3840                             rbd_dev->parent_overlap);
3841         }
3842
3843         return count;
3844 }
3845
3846 static ssize_t rbd_image_refresh(struct device *dev,
3847                                  struct device_attribute *attr,
3848                                  const char *buf,
3849                                  size_t size)
3850 {
3851         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3852         int ret;
3853
3854         ret = rbd_dev_refresh(rbd_dev);
3855         if (ret)
3856                 return ret;
3857
3858         return size;
3859 }
3860
3861 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
3862 static DEVICE_ATTR(features, S_IRUGO, rbd_features_show, NULL);
3863 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
3864 static DEVICE_ATTR(minor, S_IRUGO, rbd_minor_show, NULL);
3865 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
3866 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
3867 static DEVICE_ATTR(pool_id, S_IRUGO, rbd_pool_id_show, NULL);
3868 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
3869 static DEVICE_ATTR(image_id, S_IRUGO, rbd_image_id_show, NULL);
3870 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
3871 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
3872 static DEVICE_ATTR(parent, S_IRUGO, rbd_parent_show, NULL);
3873
3874 static struct attribute *rbd_attrs[] = {
3875         &dev_attr_size.attr,
3876         &dev_attr_features.attr,
3877         &dev_attr_major.attr,
3878         &dev_attr_minor.attr,
3879         &dev_attr_client_id.attr,
3880         &dev_attr_pool.attr,
3881         &dev_attr_pool_id.attr,
3882         &dev_attr_name.attr,
3883         &dev_attr_image_id.attr,
3884         &dev_attr_current_snap.attr,
3885         &dev_attr_parent.attr,
3886         &dev_attr_refresh.attr,
3887         NULL
3888 };
3889
3890 static struct attribute_group rbd_attr_group = {
3891         .attrs = rbd_attrs,
3892 };
3893
3894 static const struct attribute_group *rbd_attr_groups[] = {
3895         &rbd_attr_group,
3896         NULL
3897 };
3898
3899 static void rbd_dev_release(struct device *dev);
3900
3901 static struct device_type rbd_device_type = {
3902         .name           = "rbd",
3903         .groups         = rbd_attr_groups,
3904         .release        = rbd_dev_release,
3905 };
3906
3907 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
3908 {
3909         kref_get(&spec->kref);
3910
3911         return spec;
3912 }
3913
3914 static void rbd_spec_free(struct kref *kref);
3915 static void rbd_spec_put(struct rbd_spec *spec)
3916 {
3917         if (spec)
3918                 kref_put(&spec->kref, rbd_spec_free);
3919 }
3920
3921 static struct rbd_spec *rbd_spec_alloc(void)
3922 {
3923         struct rbd_spec *spec;
3924
3925         spec = kzalloc(sizeof (*spec), GFP_KERNEL);
3926         if (!spec)
3927                 return NULL;
3928
3929         spec->pool_id = CEPH_NOPOOL;
3930         spec->snap_id = CEPH_NOSNAP;
3931         kref_init(&spec->kref);
3932
3933         return spec;
3934 }
3935
3936 static void rbd_spec_free(struct kref *kref)
3937 {
3938         struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
3939
3940         kfree(spec->pool_name);
3941         kfree(spec->image_id);
3942         kfree(spec->image_name);
3943         kfree(spec->snap_name);
3944         kfree(spec);
3945 }
3946
3947 static void rbd_dev_release(struct device *dev)
3948 {
3949         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3950         bool need_put = !!rbd_dev->opts;
3951
3952         ceph_oid_destroy(&rbd_dev->header_oid);
3953
3954         rbd_put_client(rbd_dev->rbd_client);
3955         rbd_spec_put(rbd_dev->spec);
3956         kfree(rbd_dev->opts);
3957         kfree(rbd_dev);
3958
3959         /*
3960          * This is racy, but way better than putting module outside of
3961          * the release callback.  The race window is pretty small, so
3962          * doing something similar to dm (dm-builtin.c) is overkill.
3963          */
3964         if (need_put)
3965                 module_put(THIS_MODULE);
3966 }
3967
3968 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
3969                                          struct rbd_spec *spec,
3970                                          struct rbd_options *opts)
3971 {
3972         struct rbd_device *rbd_dev;
3973
3974         rbd_dev = kzalloc(sizeof (*rbd_dev), GFP_KERNEL);
3975         if (!rbd_dev)
3976                 return NULL;
3977
3978         spin_lock_init(&rbd_dev->lock);
3979         rbd_dev->flags = 0;
3980         atomic_set(&rbd_dev->parent_ref, 0);
3981         INIT_LIST_HEAD(&rbd_dev->node);
3982         init_rwsem(&rbd_dev->header_rwsem);
3983
3984         ceph_oid_init(&rbd_dev->header_oid);
3985         ceph_oloc_init(&rbd_dev->header_oloc);
3986
3987         rbd_dev->dev.bus = &rbd_bus_type;
3988         rbd_dev->dev.type = &rbd_device_type;
3989         rbd_dev->dev.parent = &rbd_root_dev;
3990         device_initialize(&rbd_dev->dev);
3991
3992         rbd_dev->rbd_client = rbdc;
3993         rbd_dev->spec = spec;
3994         rbd_dev->opts = opts;
3995
3996         /* Initialize the layout used for all rbd requests */
3997
3998         rbd_dev->layout.fl_stripe_unit = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
3999         rbd_dev->layout.fl_stripe_count = cpu_to_le32(1);
4000         rbd_dev->layout.fl_object_size = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
4001         rbd_dev->layout.fl_pg_pool = cpu_to_le32((u32) spec->pool_id);
4002
4003         /*
4004          * If this is a mapping rbd_dev (as opposed to a parent one),
4005          * pin our module.  We have a ref from do_rbd_add(), so use
4006          * __module_get().
4007          */
4008         if (rbd_dev->opts)
4009                 __module_get(THIS_MODULE);
4010
4011         return rbd_dev;
4012 }
4013
4014 static void rbd_dev_destroy(struct rbd_device *rbd_dev)
4015 {
4016         if (rbd_dev)
4017                 put_device(&rbd_dev->dev);
4018 }
4019
4020 /*
4021  * Get the size and object order for an image snapshot, or if
4022  * snap_id is CEPH_NOSNAP, gets this information for the base
4023  * image.
4024  */
4025 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
4026                                 u8 *order, u64 *snap_size)
4027 {
4028         __le64 snapid = cpu_to_le64(snap_id);
4029         int ret;
4030         struct {
4031                 u8 order;
4032                 __le64 size;
4033         } __attribute__ ((packed)) size_buf = { 0 };
4034
4035         ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_oid.name,
4036                                 "rbd", "get_size",
4037                                 &snapid, sizeof (snapid),
4038                                 &size_buf, sizeof (size_buf));
4039         dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4040         if (ret < 0)
4041                 return ret;
4042         if (ret < sizeof (size_buf))
4043                 return -ERANGE;
4044
4045         if (order) {
4046                 *order = size_buf.order;
4047                 dout("  order %u", (unsigned int)*order);
4048         }
4049         *snap_size = le64_to_cpu(size_buf.size);
4050
4051         dout("  snap_id 0x%016llx snap_size = %llu\n",
4052                 (unsigned long long)snap_id,
4053                 (unsigned long long)*snap_size);
4054
4055         return 0;
4056 }
4057
4058 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
4059 {
4060         return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
4061                                         &rbd_dev->header.obj_order,
4062                                         &rbd_dev->header.image_size);
4063 }
4064
4065 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
4066 {
4067         void *reply_buf;
4068         int ret;
4069         void *p;
4070
4071         reply_buf = kzalloc(RBD_OBJ_PREFIX_LEN_MAX, GFP_KERNEL);
4072         if (!reply_buf)
4073                 return -ENOMEM;
4074
4075         ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_oid.name,
4076                                 "rbd", "get_object_prefix", NULL, 0,
4077                                 reply_buf, RBD_OBJ_PREFIX_LEN_MAX);
4078         dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4079         if (ret < 0)
4080                 goto out;
4081
4082         p = reply_buf;
4083         rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
4084                                                 p + ret, NULL, GFP_NOIO);
4085         ret = 0;
4086
4087         if (IS_ERR(rbd_dev->header.object_prefix)) {
4088                 ret = PTR_ERR(rbd_dev->header.object_prefix);
4089                 rbd_dev->header.object_prefix = NULL;
4090         } else {
4091                 dout("  object_prefix = %s\n", rbd_dev->header.object_prefix);
4092         }
4093 out:
4094         kfree(reply_buf);
4095
4096         return ret;
4097 }
4098
4099 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
4100                 u64 *snap_features)
4101 {
4102         __le64 snapid = cpu_to_le64(snap_id);
4103         struct {
4104                 __le64 features;
4105                 __le64 incompat;
4106         } __attribute__ ((packed)) features_buf = { 0 };
4107         u64 unsup;
4108         int ret;
4109
4110         ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_oid.name,
4111                                 "rbd", "get_features",
4112                                 &snapid, sizeof (snapid),
4113                                 &features_buf, sizeof (features_buf));
4114         dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4115         if (ret < 0)
4116                 return ret;
4117         if (ret < sizeof (features_buf))
4118                 return -ERANGE;
4119
4120         unsup = le64_to_cpu(features_buf.incompat) & ~RBD_FEATURES_SUPPORTED;
4121         if (unsup) {
4122                 rbd_warn(rbd_dev, "image uses unsupported features: 0x%llx",
4123                          unsup);
4124                 return -ENXIO;
4125         }
4126
4127         *snap_features = le64_to_cpu(features_buf.features);
4128
4129         dout("  snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
4130                 (unsigned long long)snap_id,
4131                 (unsigned long long)*snap_features,
4132                 (unsigned long long)le64_to_cpu(features_buf.incompat));
4133
4134         return 0;
4135 }
4136
4137 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
4138 {
4139         return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
4140                                                 &rbd_dev->header.features);
4141 }
4142
4143 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
4144 {
4145         struct rbd_spec *parent_spec;
4146         size_t size;
4147         void *reply_buf = NULL;
4148         __le64 snapid;
4149         void *p;
4150         void *end;
4151         u64 pool_id;
4152         char *image_id;
4153         u64 snap_id;
4154         u64 overlap;
4155         int ret;
4156
4157         parent_spec = rbd_spec_alloc();
4158         if (!parent_spec)
4159                 return -ENOMEM;
4160
4161         size = sizeof (__le64) +                                /* pool_id */
4162                 sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX +        /* image_id */
4163                 sizeof (__le64) +                               /* snap_id */
4164                 sizeof (__le64);                                /* overlap */
4165         reply_buf = kmalloc(size, GFP_KERNEL);
4166         if (!reply_buf) {
4167                 ret = -ENOMEM;
4168                 goto out_err;
4169         }
4170
4171         snapid = cpu_to_le64(rbd_dev->spec->snap_id);
4172         ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_oid.name,
4173                                 "rbd", "get_parent",
4174                                 &snapid, sizeof (snapid),
4175                                 reply_buf, size);
4176         dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4177         if (ret < 0)
4178                 goto out_err;
4179
4180         p = reply_buf;
4181         end = reply_buf + ret;
4182         ret = -ERANGE;
4183         ceph_decode_64_safe(&p, end, pool_id, out_err);
4184         if (pool_id == CEPH_NOPOOL) {
4185                 /*
4186                  * Either the parent never existed, or we have
4187                  * record of it but the image got flattened so it no
4188                  * longer has a parent.  When the parent of a
4189                  * layered image disappears we immediately set the
4190                  * overlap to 0.  The effect of this is that all new
4191                  * requests will be treated as if the image had no
4192                  * parent.
4193                  */
4194                 if (rbd_dev->parent_overlap) {
4195                         rbd_dev->parent_overlap = 0;
4196                         rbd_dev_parent_put(rbd_dev);
4197                         pr_info("%s: clone image has been flattened\n",
4198                                 rbd_dev->disk->disk_name);
4199                 }
4200
4201                 goto out;       /* No parent?  No problem. */
4202         }
4203
4204         /* The ceph file layout needs to fit pool id in 32 bits */
4205
4206         ret = -EIO;
4207         if (pool_id > (u64)U32_MAX) {
4208                 rbd_warn(NULL, "parent pool id too large (%llu > %u)",
4209                         (unsigned long long)pool_id, U32_MAX);
4210                 goto out_err;
4211         }
4212
4213         image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
4214         if (IS_ERR(image_id)) {
4215                 ret = PTR_ERR(image_id);
4216                 goto out_err;
4217         }
4218         ceph_decode_64_safe(&p, end, snap_id, out_err);
4219         ceph_decode_64_safe(&p, end, overlap, out_err);
4220
4221         /*
4222          * The parent won't change (except when the clone is
4223          * flattened, already handled that).  So we only need to
4224          * record the parent spec we have not already done so.
4225          */
4226         if (!rbd_dev->parent_spec) {
4227                 parent_spec->pool_id = pool_id;
4228                 parent_spec->image_id = image_id;
4229                 parent_spec->snap_id = snap_id;
4230                 rbd_dev->parent_spec = parent_spec;
4231                 parent_spec = NULL;     /* rbd_dev now owns this */
4232         } else {
4233                 kfree(image_id);
4234         }
4235
4236         /*
4237          * We always update the parent overlap.  If it's zero we issue
4238          * a warning, as we will proceed as if there was no parent.
4239          */
4240         if (!overlap) {
4241                 if (parent_spec) {
4242                         /* refresh, careful to warn just once */
4243                         if (rbd_dev->parent_overlap)
4244                                 rbd_warn(rbd_dev,
4245                                     "clone now standalone (overlap became 0)");
4246                 } else {
4247                         /* initial probe */
4248                         rbd_warn(rbd_dev, "clone is standalone (overlap 0)");
4249                 }
4250         }
4251         rbd_dev->parent_overlap = overlap;
4252
4253 out:
4254         ret = 0;
4255 out_err:
4256         kfree(reply_buf);
4257         rbd_spec_put(parent_spec);
4258
4259         return ret;
4260 }
4261
4262 static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev)
4263 {
4264         struct {
4265                 __le64 stripe_unit;
4266                 __le64 stripe_count;
4267         } __attribute__ ((packed)) striping_info_buf = { 0 };
4268         size_t size = sizeof (striping_info_buf);
4269         void *p;
4270         u64 obj_size;
4271         u64 stripe_unit;
4272         u64 stripe_count;
4273         int ret;
4274
4275         ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_oid.name,
4276                                 "rbd", "get_stripe_unit_count", NULL, 0,
4277                                 (char *)&striping_info_buf, size);
4278         dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4279         if (ret < 0)
4280                 return ret;
4281         if (ret < size)
4282                 return -ERANGE;
4283
4284         /*
4285          * We don't actually support the "fancy striping" feature
4286          * (STRIPINGV2) yet, but if the striping sizes are the
4287          * defaults the behavior is the same as before.  So find
4288          * out, and only fail if the image has non-default values.
4289          */
4290         ret = -EINVAL;
4291         obj_size = (u64)1 << rbd_dev->header.obj_order;
4292         p = &striping_info_buf;
4293         stripe_unit = ceph_decode_64(&p);
4294         if (stripe_unit != obj_size) {
4295                 rbd_warn(rbd_dev, "unsupported stripe unit "
4296                                 "(got %llu want %llu)",
4297                                 stripe_unit, obj_size);
4298                 return -EINVAL;
4299         }
4300         stripe_count = ceph_decode_64(&p);
4301         if (stripe_count != 1) {
4302                 rbd_warn(rbd_dev, "unsupported stripe count "
4303                                 "(got %llu want 1)", stripe_count);
4304                 return -EINVAL;
4305         }
4306         rbd_dev->header.stripe_unit = stripe_unit;
4307         rbd_dev->header.stripe_count = stripe_count;
4308
4309         return 0;
4310 }
4311
4312 static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
4313 {
4314         size_t image_id_size;
4315         char *image_id;
4316         void *p;
4317         void *end;
4318         size_t size;
4319         void *reply_buf = NULL;
4320         size_t len = 0;
4321         char *image_name = NULL;
4322         int ret;
4323
4324         rbd_assert(!rbd_dev->spec->image_name);
4325
4326         len = strlen(rbd_dev->spec->image_id);
4327         image_id_size = sizeof (__le32) + len;
4328         image_id = kmalloc(image_id_size, GFP_KERNEL);
4329         if (!image_id)
4330                 return NULL;
4331
4332         p = image_id;
4333         end = image_id + image_id_size;
4334         ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len);
4335
4336         size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
4337         reply_buf = kmalloc(size, GFP_KERNEL);
4338         if (!reply_buf)
4339                 goto out;
4340
4341         ret = rbd_obj_method_sync(rbd_dev, RBD_DIRECTORY,
4342                                 "rbd", "dir_get_name",
4343                                 image_id, image_id_size,
4344                                 reply_buf, size);
4345         if (ret < 0)
4346                 goto out;
4347         p = reply_buf;
4348         end = reply_buf + ret;
4349
4350         image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
4351         if (IS_ERR(image_name))
4352                 image_name = NULL;
4353         else
4354                 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
4355 out:
4356         kfree(reply_buf);
4357         kfree(image_id);
4358
4359         return image_name;
4360 }
4361
4362 static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
4363 {
4364         struct ceph_snap_context *snapc = rbd_dev->header.snapc;
4365         const char *snap_name;
4366         u32 which = 0;
4367
4368         /* Skip over names until we find the one we are looking for */
4369
4370         snap_name = rbd_dev->header.snap_names;
4371         while (which < snapc->num_snaps) {
4372                 if (!strcmp(name, snap_name))
4373                         return snapc->snaps[which];
4374                 snap_name += strlen(snap_name) + 1;
4375                 which++;
4376         }
4377         return CEPH_NOSNAP;
4378 }
4379
4380 static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
4381 {
4382         struct ceph_snap_context *snapc = rbd_dev->header.snapc;
4383         u32 which;
4384         bool found = false;
4385         u64 snap_id;
4386
4387         for (which = 0; !found && which < snapc->num_snaps; which++) {
4388                 const char *snap_name;
4389
4390                 snap_id = snapc->snaps[which];
4391                 snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
4392                 if (IS_ERR(snap_name)) {
4393                         /* ignore no-longer existing snapshots */
4394                         if (PTR_ERR(snap_name) == -ENOENT)
4395                                 continue;
4396                         else
4397                                 break;
4398                 }
4399                 found = !strcmp(name, snap_name);
4400                 kfree(snap_name);
4401         }
4402         return found ? snap_id : CEPH_NOSNAP;
4403 }
4404
4405 /*
4406  * Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if
4407  * no snapshot by that name is found, or if an error occurs.
4408  */
4409 static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
4410 {
4411         if (rbd_dev->image_format == 1)
4412                 return rbd_v1_snap_id_by_name(rbd_dev, name);
4413
4414         return rbd_v2_snap_id_by_name(rbd_dev, name);
4415 }
4416
4417 /*
4418  * An image being mapped will have everything but the snap id.
4419  */
4420 static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev)
4421 {
4422         struct rbd_spec *spec = rbd_dev->spec;
4423
4424         rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name);
4425         rbd_assert(spec->image_id && spec->image_name);
4426         rbd_assert(spec->snap_name);
4427
4428         if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) {
4429                 u64 snap_id;
4430
4431                 snap_id = rbd_snap_id_by_name(rbd_dev, spec->snap_name);
4432                 if (snap_id == CEPH_NOSNAP)
4433                         return -ENOENT;
4434
4435                 spec->snap_id = snap_id;
4436         } else {
4437                 spec->snap_id = CEPH_NOSNAP;
4438         }
4439
4440         return 0;
4441 }
4442
4443 /*
4444  * A parent image will have all ids but none of the names.
4445  *
4446  * All names in an rbd spec are dynamically allocated.  It's OK if we
4447  * can't figure out the name for an image id.
4448  */
4449 static int rbd_spec_fill_names(struct rbd_device *rbd_dev)
4450 {
4451         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4452         struct rbd_spec *spec = rbd_dev->spec;
4453         const char *pool_name;
4454         const char *image_name;
4455         const char *snap_name;
4456         int ret;
4457
4458         rbd_assert(spec->pool_id != CEPH_NOPOOL);
4459         rbd_assert(spec->image_id);
4460         rbd_assert(spec->snap_id != CEPH_NOSNAP);
4461
4462         /* Get the pool name; we have to make our own copy of this */
4463
4464         pool_name = ceph_pg_pool_name_by_id(osdc->osdmap, spec->pool_id);
4465         if (!pool_name) {
4466                 rbd_warn(rbd_dev, "no pool with id %llu", spec->pool_id);
4467                 return -EIO;
4468         }
4469         pool_name = kstrdup(pool_name, GFP_KERNEL);
4470         if (!pool_name)
4471                 return -ENOMEM;
4472
4473         /* Fetch the image name; tolerate failure here */
4474
4475         image_name = rbd_dev_image_name(rbd_dev);
4476         if (!image_name)
4477                 rbd_warn(rbd_dev, "unable to get image name");
4478
4479         /* Fetch the snapshot name */
4480
4481         snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
4482         if (IS_ERR(snap_name)) {
4483                 ret = PTR_ERR(snap_name);
4484                 goto out_err;
4485         }
4486
4487         spec->pool_name = pool_name;
4488         spec->image_name = image_name;
4489         spec->snap_name = snap_name;
4490
4491         return 0;
4492
4493 out_err:
4494         kfree(image_name);
4495         kfree(pool_name);
4496         return ret;
4497 }
4498
4499 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev)
4500 {
4501         size_t size;
4502         int ret;
4503         void *reply_buf;
4504         void *p;
4505         void *end;
4506         u64 seq;
4507         u32 snap_count;
4508         struct ceph_snap_context *snapc;
4509         u32 i;
4510
4511         /*
4512          * We'll need room for the seq value (maximum snapshot id),
4513          * snapshot count, and array of that many snapshot ids.
4514          * For now we have a fixed upper limit on the number we're
4515          * prepared to receive.
4516          */
4517         size = sizeof (__le64) + sizeof (__le32) +
4518                         RBD_MAX_SNAP_COUNT * sizeof (__le64);
4519         reply_buf = kzalloc(size, GFP_KERNEL);
4520         if (!reply_buf)
4521                 return -ENOMEM;
4522
4523         ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_oid.name,
4524                                 "rbd", "get_snapcontext", NULL, 0,
4525                                 reply_buf, size);
4526         dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4527         if (ret < 0)
4528                 goto out;
4529
4530         p = reply_buf;
4531         end = reply_buf + ret;
4532         ret = -ERANGE;
4533         ceph_decode_64_safe(&p, end, seq, out);
4534         ceph_decode_32_safe(&p, end, snap_count, out);
4535
4536         /*
4537          * Make sure the reported number of snapshot ids wouldn't go
4538          * beyond the end of our buffer.  But before checking that,
4539          * make sure the computed size of the snapshot context we
4540          * allocate is representable in a size_t.
4541          */
4542         if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
4543                                  / sizeof (u64)) {
4544                 ret = -EINVAL;
4545                 goto out;
4546         }
4547         if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
4548                 goto out;
4549         ret = 0;
4550
4551         snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
4552         if (!snapc) {
4553                 ret = -ENOMEM;
4554                 goto out;
4555         }
4556         snapc->seq = seq;
4557         for (i = 0; i < snap_count; i++)
4558                 snapc->snaps[i] = ceph_decode_64(&p);
4559
4560         ceph_put_snap_context(rbd_dev->header.snapc);
4561         rbd_dev->header.snapc = snapc;
4562
4563         dout("  snap context seq = %llu, snap_count = %u\n",
4564                 (unsigned long long)seq, (unsigned int)snap_count);
4565 out:
4566         kfree(reply_buf);
4567
4568         return ret;
4569 }
4570
4571 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
4572                                         u64 snap_id)
4573 {
4574         size_t size;
4575         void *reply_buf;
4576         __le64 snapid;
4577         int ret;
4578         void *p;
4579         void *end;
4580         char *snap_name;
4581
4582         size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
4583         reply_buf = kmalloc(size, GFP_KERNEL);
4584         if (!reply_buf)
4585                 return ERR_PTR(-ENOMEM);
4586
4587         snapid = cpu_to_le64(snap_id);
4588         ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_oid.name,
4589                                 "rbd", "get_snapshot_name",
4590                                 &snapid, sizeof (snapid),
4591                                 reply_buf, size);
4592         dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4593         if (ret < 0) {
4594                 snap_name = ERR_PTR(ret);
4595                 goto out;
4596         }
4597
4598         p = reply_buf;
4599         end = reply_buf + ret;
4600         snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
4601         if (IS_ERR(snap_name))
4602                 goto out;
4603
4604         dout("  snap_id 0x%016llx snap_name = %s\n",
4605                 (unsigned long long)snap_id, snap_name);
4606 out:
4607         kfree(reply_buf);
4608
4609         return snap_name;
4610 }
4611
4612 static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev)
4613 {
4614         bool first_time = rbd_dev->header.object_prefix == NULL;
4615         int ret;
4616
4617         ret = rbd_dev_v2_image_size(rbd_dev);
4618         if (ret)
4619                 return ret;
4620
4621         if (first_time) {
4622                 ret = rbd_dev_v2_header_onetime(rbd_dev);
4623                 if (ret)
4624                         return ret;
4625         }
4626
4627         ret = rbd_dev_v2_snap_context(rbd_dev);
4628         if (ret && first_time) {
4629                 kfree(rbd_dev->header.object_prefix);
4630                 rbd_dev->header.object_prefix = NULL;
4631         }
4632
4633         return ret;
4634 }
4635
4636 static int rbd_dev_header_info(struct rbd_device *rbd_dev)
4637 {
4638         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
4639
4640         if (rbd_dev->image_format == 1)
4641                 return rbd_dev_v1_header_info(rbd_dev);
4642
4643         return rbd_dev_v2_header_info(rbd_dev);
4644 }
4645
4646 /*
4647  * Get a unique rbd identifier for the given new rbd_dev, and add
4648  * the rbd_dev to the global list.
4649  */
4650 static int rbd_dev_id_get(struct rbd_device *rbd_dev)
4651 {
4652         int new_dev_id;
4653
4654         new_dev_id = ida_simple_get(&rbd_dev_id_ida,
4655                                     0, minor_to_rbd_dev_id(1 << MINORBITS),
4656                                     GFP_KERNEL);
4657         if (new_dev_id < 0)
4658                 return new_dev_id;
4659
4660         rbd_dev->dev_id = new_dev_id;
4661
4662         spin_lock(&rbd_dev_list_lock);
4663         list_add_tail(&rbd_dev->node, &rbd_dev_list);
4664         spin_unlock(&rbd_dev_list_lock);
4665
4666         dout("rbd_dev %p given dev id %d\n", rbd_dev, rbd_dev->dev_id);
4667
4668         return 0;
4669 }
4670
4671 /*
4672  * Remove an rbd_dev from the global list, and record that its
4673  * identifier is no longer in use.
4674  */
4675 static void rbd_dev_id_put(struct rbd_device *rbd_dev)
4676 {
4677         spin_lock(&rbd_dev_list_lock);
4678         list_del_init(&rbd_dev->node);
4679         spin_unlock(&rbd_dev_list_lock);
4680
4681         ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
4682
4683         dout("rbd_dev %p released dev id %d\n", rbd_dev, rbd_dev->dev_id);
4684 }
4685
4686 /*
4687  * Skips over white space at *buf, and updates *buf to point to the
4688  * first found non-space character (if any). Returns the length of
4689  * the token (string of non-white space characters) found.  Note
4690  * that *buf must be terminated with '\0'.
4691  */
4692 static inline size_t next_token(const char **buf)
4693 {
4694         /*
4695         * These are the characters that produce nonzero for
4696         * isspace() in the "C" and "POSIX" locales.
4697         */
4698         const char *spaces = " \f\n\r\t\v";
4699
4700         *buf += strspn(*buf, spaces);   /* Find start of token */
4701
4702         return strcspn(*buf, spaces);   /* Return token length */
4703 }
4704
4705 /*
4706  * Finds the next token in *buf, dynamically allocates a buffer big
4707  * enough to hold a copy of it, and copies the token into the new
4708  * buffer.  The copy is guaranteed to be terminated with '\0'.  Note
4709  * that a duplicate buffer is created even for a zero-length token.
4710  *
4711  * Returns a pointer to the newly-allocated duplicate, or a null
4712  * pointer if memory for the duplicate was not available.  If
4713  * the lenp argument is a non-null pointer, the length of the token
4714  * (not including the '\0') is returned in *lenp.
4715  *
4716  * If successful, the *buf pointer will be updated to point beyond
4717  * the end of the found token.
4718  *
4719  * Note: uses GFP_KERNEL for allocation.
4720  */
4721 static inline char *dup_token(const char **buf, size_t *lenp)
4722 {
4723         char *dup;
4724         size_t len;
4725
4726         len = next_token(buf);
4727         dup = kmemdup(*buf, len + 1, GFP_KERNEL);
4728         if (!dup)
4729                 return NULL;
4730         *(dup + len) = '\0';
4731         *buf += len;
4732
4733         if (lenp)
4734                 *lenp = len;
4735
4736         return dup;
4737 }
4738
4739 /*
4740  * Parse the options provided for an "rbd add" (i.e., rbd image
4741  * mapping) request.  These arrive via a write to /sys/bus/rbd/add,
4742  * and the data written is passed here via a NUL-terminated buffer.
4743  * Returns 0 if successful or an error code otherwise.
4744  *
4745  * The information extracted from these options is recorded in
4746  * the other parameters which return dynamically-allocated
4747  * structures:
4748  *  ceph_opts
4749  *      The address of a pointer that will refer to a ceph options
4750  *      structure.  Caller must release the returned pointer using
4751  *      ceph_destroy_options() when it is no longer needed.
4752  *  rbd_opts
4753  *      Address of an rbd options pointer.  Fully initialized by
4754  *      this function; caller must release with kfree().
4755  *  spec
4756  *      Address of an rbd image specification pointer.  Fully
4757  *      initialized by this function based on parsed options.
4758  *      Caller must release with rbd_spec_put().
4759  *
4760  * The options passed take this form:
4761  *  <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
4762  * where:
4763  *  <mon_addrs>
4764  *      A comma-separated list of one or more monitor addresses.
4765  *      A monitor address is an ip address, optionally followed
4766  *      by a port number (separated by a colon).
4767  *        I.e.:  ip1[:port1][,ip2[:port2]...]
4768  *  <options>
4769  *      A comma-separated list of ceph and/or rbd options.
4770  *  <pool_name>
4771  *      The name of the rados pool containing the rbd image.
4772  *  <image_name>
4773  *      The name of the image in that pool to map.
4774  *  <snap_id>
4775  *      An optional snapshot id.  If provided, the mapping will
4776  *      present data from the image at the time that snapshot was
4777  *      created.  The image head is used if no snapshot id is
4778  *      provided.  Snapshot mappings are always read-only.
4779  */
4780 static int rbd_add_parse_args(const char *buf,
4781                                 struct ceph_options **ceph_opts,
4782                                 struct rbd_options **opts,
4783                                 struct rbd_spec **rbd_spec)
4784 {
4785         size_t len;
4786         char *options;
4787         const char *mon_addrs;
4788         char *snap_name;
4789         size_t mon_addrs_size;
4790         struct rbd_spec *spec = NULL;
4791         struct rbd_options *rbd_opts = NULL;
4792         struct ceph_options *copts;
4793         int ret;
4794
4795         /* The first four tokens are required */
4796
4797         len = next_token(&buf);
4798         if (!len) {
4799                 rbd_warn(NULL, "no monitor address(es) provided");
4800                 return -EINVAL;
4801         }
4802         mon_addrs = buf;
4803         mon_addrs_size = len + 1;
4804         buf += len;
4805
4806         ret = -EINVAL;
4807         options = dup_token(&buf, NULL);
4808         if (!options)
4809                 return -ENOMEM;
4810         if (!*options) {
4811                 rbd_warn(NULL, "no options provided");
4812                 goto out_err;
4813         }
4814
4815         spec = rbd_spec_alloc();
4816         if (!spec)
4817                 goto out_mem;
4818
4819         spec->pool_name = dup_token(&buf, NULL);
4820         if (!spec->pool_name)
4821                 goto out_mem;
4822         if (!*spec->pool_name) {
4823                 rbd_warn(NULL, "no pool name provided");
4824                 goto out_err;
4825         }
4826
4827         spec->image_name = dup_token(&buf, NULL);
4828         if (!spec->image_name)
4829                 goto out_mem;
4830         if (!*spec->image_name) {
4831                 rbd_warn(NULL, "no image name provided");
4832                 goto out_err;
4833         }
4834
4835         /*
4836          * Snapshot name is optional; default is to use "-"
4837          * (indicating the head/no snapshot).
4838          */
4839         len = next_token(&buf);
4840         if (!len) {
4841                 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
4842                 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
4843         } else if (len > RBD_MAX_SNAP_NAME_LEN) {
4844                 ret = -ENAMETOOLONG;
4845                 goto out_err;
4846         }
4847         snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
4848         if (!snap_name)
4849                 goto out_mem;
4850         *(snap_name + len) = '\0';
4851         spec->snap_name = snap_name;
4852
4853         /* Initialize all rbd options to the defaults */
4854
4855         rbd_opts = kzalloc(sizeof (*rbd_opts), GFP_KERNEL);
4856         if (!rbd_opts)
4857                 goto out_mem;
4858
4859         rbd_opts->read_only = RBD_READ_ONLY_DEFAULT;
4860         rbd_opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
4861
4862         copts = ceph_parse_options(options, mon_addrs,
4863                                         mon_addrs + mon_addrs_size - 1,
4864                                         parse_rbd_opts_token, rbd_opts);
4865         if (IS_ERR(copts)) {
4866                 ret = PTR_ERR(copts);
4867                 goto out_err;
4868         }
4869         kfree(options);
4870
4871         *ceph_opts = copts;
4872         *opts = rbd_opts;
4873         *rbd_spec = spec;
4874
4875         return 0;
4876 out_mem:
4877         ret = -ENOMEM;
4878 out_err:
4879         kfree(rbd_opts);
4880         rbd_spec_put(spec);
4881         kfree(options);
4882
4883         return ret;
4884 }
4885
4886 /*
4887  * Return pool id (>= 0) or a negative error code.
4888  */
4889 static int rbd_add_get_pool_id(struct rbd_client *rbdc, const char *pool_name)
4890 {
4891         struct ceph_options *opts = rbdc->client->options;
4892         u64 newest_epoch;
4893         int tries = 0;
4894         int ret;
4895
4896 again:
4897         ret = ceph_pg_poolid_by_name(rbdc->client->osdc.osdmap, pool_name);
4898         if (ret == -ENOENT && tries++ < 1) {
4899                 ret = ceph_monc_get_version(&rbdc->client->monc, "osdmap",
4900                                             &newest_epoch);
4901                 if (ret < 0)
4902                         return ret;
4903
4904                 if (rbdc->client->osdc.osdmap->epoch < newest_epoch) {
4905                         ceph_osdc_maybe_request_map(&rbdc->client->osdc);
4906                         (void) ceph_monc_wait_osdmap(&rbdc->client->monc,
4907                                                      newest_epoch,
4908                                                      opts->mount_timeout);
4909                         goto again;
4910                 } else {
4911                         /* the osdmap we have is new enough */
4912                         return -ENOENT;
4913                 }
4914         }
4915
4916         return ret;
4917 }
4918
4919 /*
4920  * An rbd format 2 image has a unique identifier, distinct from the
4921  * name given to it by the user.  Internally, that identifier is
4922  * what's used to specify the names of objects related to the image.
4923  *
4924  * A special "rbd id" object is used to map an rbd image name to its
4925  * id.  If that object doesn't exist, then there is no v2 rbd image
4926  * with the supplied name.
4927  *
4928  * This function will record the given rbd_dev's image_id field if
4929  * it can be determined, and in that case will return 0.  If any
4930  * errors occur a negative errno will be returned and the rbd_dev's
4931  * image_id field will be unchanged (and should be NULL).
4932  */
4933 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
4934 {
4935         int ret;
4936         size_t size;
4937         char *object_name;
4938         void *response;
4939         char *image_id;
4940
4941         /*
4942          * When probing a parent image, the image id is already
4943          * known (and the image name likely is not).  There's no
4944          * need to fetch the image id again in this case.  We
4945          * do still need to set the image format though.
4946          */
4947         if (rbd_dev->spec->image_id) {
4948                 rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
4949
4950                 return 0;
4951         }
4952
4953         /*
4954          * First, see if the format 2 image id file exists, and if
4955          * so, get the image's persistent id from it.
4956          */
4957         size = sizeof (RBD_ID_PREFIX) + strlen(rbd_dev->spec->image_name);
4958         object_name = kmalloc(size, GFP_NOIO);
4959         if (!object_name)
4960                 return -ENOMEM;
4961         sprintf(object_name, "%s%s", RBD_ID_PREFIX, rbd_dev->spec->image_name);
4962         dout("rbd id object name is %s\n", object_name);
4963
4964         /* Response will be an encoded string, which includes a length */
4965
4966         size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
4967         response = kzalloc(size, GFP_NOIO);
4968         if (!response) {
4969                 ret = -ENOMEM;
4970                 goto out;
4971         }
4972
4973         /* If it doesn't exist we'll assume it's a format 1 image */
4974
4975         ret = rbd_obj_method_sync(rbd_dev, object_name,
4976                                 "rbd", "get_id", NULL, 0,
4977                                 response, RBD_IMAGE_ID_LEN_MAX);
4978         dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4979         if (ret == -ENOENT) {
4980                 image_id = kstrdup("", GFP_KERNEL);
4981                 ret = image_id ? 0 : -ENOMEM;
4982                 if (!ret)
4983                         rbd_dev->image_format = 1;
4984         } else if (ret >= 0) {
4985                 void *p = response;
4986
4987                 image_id = ceph_extract_encoded_string(&p, p + ret,
4988                                                 NULL, GFP_NOIO);
4989                 ret = PTR_ERR_OR_ZERO(image_id);
4990                 if (!ret)
4991                         rbd_dev->image_format = 2;
4992         }
4993
4994         if (!ret) {
4995                 rbd_dev->spec->image_id = image_id;
4996                 dout("image_id is %s\n", image_id);
4997         }
4998 out:
4999         kfree(response);
5000         kfree(object_name);
5001
5002         return ret;
5003 }
5004
5005 /*
5006  * Undo whatever state changes are made by v1 or v2 header info
5007  * call.
5008  */
5009 static void rbd_dev_unprobe(struct rbd_device *rbd_dev)
5010 {
5011         struct rbd_image_header *header;
5012
5013         rbd_dev_parent_put(rbd_dev);
5014
5015         /* Free dynamic fields from the header, then zero it out */
5016
5017         header = &rbd_dev->header;
5018         ceph_put_snap_context(header->snapc);
5019         kfree(header->snap_sizes);
5020         kfree(header->snap_names);
5021         kfree(header->object_prefix);
5022         memset(header, 0, sizeof (*header));
5023 }
5024
5025 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev)
5026 {
5027         int ret;
5028
5029         ret = rbd_dev_v2_object_prefix(rbd_dev);
5030         if (ret)
5031                 goto out_err;
5032
5033         /*
5034          * Get the and check features for the image.  Currently the
5035          * features are assumed to never change.
5036          */
5037         ret = rbd_dev_v2_features(rbd_dev);
5038         if (ret)
5039                 goto out_err;
5040
5041         /* If the image supports fancy striping, get its parameters */
5042
5043         if (rbd_dev->header.features & RBD_FEATURE_STRIPINGV2) {
5044                 ret = rbd_dev_v2_striping_info(rbd_dev);
5045                 if (ret < 0)
5046                         goto out_err;
5047         }
5048         /* No support for crypto and compression type format 2 images */
5049
5050         return 0;
5051 out_err:
5052         rbd_dev->header.features = 0;
5053         kfree(rbd_dev->header.object_prefix);
5054         rbd_dev->header.object_prefix = NULL;
5055
5056         return ret;
5057 }
5058
5059 /*
5060  * @depth is rbd_dev_image_probe() -> rbd_dev_probe_parent() ->
5061  * rbd_dev_image_probe() recursion depth, which means it's also the
5062  * length of the already discovered part of the parent chain.
5063  */
5064 static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
5065 {
5066         struct rbd_device *parent = NULL;
5067         int ret;
5068
5069         if (!rbd_dev->parent_spec)
5070                 return 0;
5071
5072         if (++depth > RBD_MAX_PARENT_CHAIN_LEN) {
5073                 pr_info("parent chain is too long (%d)\n", depth);
5074                 ret = -EINVAL;
5075                 goto out_err;
5076         }
5077
5078         parent = rbd_dev_create(rbd_dev->rbd_client, rbd_dev->parent_spec,
5079                                 NULL);
5080         if (!parent) {
5081                 ret = -ENOMEM;
5082                 goto out_err;
5083         }
5084
5085         /*
5086          * Images related by parent/child relationships always share
5087          * rbd_client and spec/parent_spec, so bump their refcounts.
5088          */
5089         __rbd_get_client(rbd_dev->rbd_client);
5090         rbd_spec_get(rbd_dev->parent_spec);
5091
5092         ret = rbd_dev_image_probe(parent, depth);
5093         if (ret < 0)
5094                 goto out_err;
5095
5096         rbd_dev->parent = parent;
5097         atomic_set(&rbd_dev->parent_ref, 1);
5098         return 0;
5099
5100 out_err:
5101         rbd_dev_unparent(rbd_dev);
5102         rbd_dev_destroy(parent);
5103         return ret;
5104 }
5105
5106 /*
5107  * rbd_dev->header_rwsem must be locked for write and will be unlocked
5108  * upon return.
5109  */
5110 static int rbd_dev_device_setup(struct rbd_device *rbd_dev)
5111 {
5112         int ret;
5113
5114         /* Get an id and fill in device name. */
5115
5116         ret = rbd_dev_id_get(rbd_dev);
5117         if (ret)
5118                 goto err_out_unlock;
5119
5120         BUILD_BUG_ON(DEV_NAME_LEN
5121                         < sizeof (RBD_DRV_NAME) + MAX_INT_FORMAT_WIDTH);
5122         sprintf(rbd_dev->name, "%s%d", RBD_DRV_NAME, rbd_dev->dev_id);
5123
5124         /* Record our major and minor device numbers. */
5125
5126         if (!single_major) {
5127                 ret = register_blkdev(0, rbd_dev->name);
5128                 if (ret < 0)
5129                         goto err_out_id;
5130
5131                 rbd_dev->major = ret;
5132                 rbd_dev->minor = 0;
5133         } else {
5134                 rbd_dev->major = rbd_major;
5135                 rbd_dev->minor = rbd_dev_id_to_minor(rbd_dev->dev_id);
5136         }
5137
5138         /* Set up the blkdev mapping. */
5139
5140         ret = rbd_init_disk(rbd_dev);
5141         if (ret)
5142                 goto err_out_blkdev;
5143
5144         ret = rbd_dev_mapping_set(rbd_dev);
5145         if (ret)
5146                 goto err_out_disk;
5147
5148         set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
5149         set_disk_ro(rbd_dev->disk, rbd_dev->mapping.read_only);
5150
5151         dev_set_name(&rbd_dev->dev, "%d", rbd_dev->dev_id);
5152         ret = device_add(&rbd_dev->dev);
5153         if (ret)
5154                 goto err_out_mapping;
5155
5156         /* Everything's ready.  Announce the disk to the world. */
5157
5158         set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
5159         up_write(&rbd_dev->header_rwsem);
5160
5161         add_disk(rbd_dev->disk);
5162         pr_info("%s: added with size 0x%llx\n", rbd_dev->disk->disk_name,
5163                 (unsigned long long) rbd_dev->mapping.size);
5164
5165         return ret;
5166
5167 err_out_mapping:
5168         rbd_dev_mapping_clear(rbd_dev);
5169 err_out_disk:
5170         rbd_free_disk(rbd_dev);
5171 err_out_blkdev:
5172         if (!single_major)
5173                 unregister_blkdev(rbd_dev->major, rbd_dev->name);
5174 err_out_id:
5175         rbd_dev_id_put(rbd_dev);
5176 err_out_unlock:
5177         up_write(&rbd_dev->header_rwsem);
5178         return ret;
5179 }
5180
5181 static int rbd_dev_header_name(struct rbd_device *rbd_dev)
5182 {
5183         struct rbd_spec *spec = rbd_dev->spec;
5184         int ret;
5185
5186         /* Record the header object name for this rbd image. */
5187
5188         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
5189
5190         rbd_dev->header_oloc.pool = ceph_file_layout_pg_pool(rbd_dev->layout);
5191         if (rbd_dev->image_format == 1)
5192                 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
5193                                        spec->image_name, RBD_SUFFIX);
5194         else
5195                 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
5196                                        RBD_HEADER_PREFIX, spec->image_id);
5197
5198         return ret;
5199 }
5200
5201 static void rbd_dev_image_release(struct rbd_device *rbd_dev)
5202 {
5203         rbd_dev_unprobe(rbd_dev);
5204         rbd_dev->image_format = 0;
5205         kfree(rbd_dev->spec->image_id);
5206         rbd_dev->spec->image_id = NULL;
5207
5208         rbd_dev_destroy(rbd_dev);
5209 }
5210
5211 /*
5212  * Probe for the existence of the header object for the given rbd
5213  * device.  If this image is the one being mapped (i.e., not a
5214  * parent), initiate a watch on its header object before using that
5215  * object to get detailed information about the rbd image.
5216  */
5217 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
5218 {
5219         int ret;
5220
5221         /*
5222          * Get the id from the image id object.  Unless there's an
5223          * error, rbd_dev->spec->image_id will be filled in with
5224          * a dynamically-allocated string, and rbd_dev->image_format
5225          * will be set to either 1 or 2.
5226          */
5227         ret = rbd_dev_image_id(rbd_dev);
5228         if (ret)
5229                 return ret;
5230
5231         ret = rbd_dev_header_name(rbd_dev);
5232         if (ret)
5233                 goto err_out_format;
5234
5235         if (!depth) {
5236                 ret = rbd_dev_header_watch_sync(rbd_dev);
5237                 if (ret) {
5238                         if (ret == -ENOENT)
5239                                 pr_info("image %s/%s does not exist\n",
5240                                         rbd_dev->spec->pool_name,
5241                                         rbd_dev->spec->image_name);
5242                         goto err_out_format;
5243                 }
5244         }
5245
5246         ret = rbd_dev_header_info(rbd_dev);
5247         if (ret)
5248                 goto err_out_watch;
5249
5250         /*
5251          * If this image is the one being mapped, we have pool name and
5252          * id, image name and id, and snap name - need to fill snap id.
5253          * Otherwise this is a parent image, identified by pool, image
5254          * and snap ids - need to fill in names for those ids.
5255          */
5256         if (!depth)
5257                 ret = rbd_spec_fill_snap_id(rbd_dev);
5258         else
5259                 ret = rbd_spec_fill_names(rbd_dev);
5260         if (ret) {
5261                 if (ret == -ENOENT)
5262                         pr_info("snap %s/%s@%s does not exist\n",
5263                                 rbd_dev->spec->pool_name,
5264                                 rbd_dev->spec->image_name,
5265                                 rbd_dev->spec->snap_name);
5266                 goto err_out_probe;
5267         }
5268
5269         if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
5270                 ret = rbd_dev_v2_parent_info(rbd_dev);
5271                 if (ret)
5272                         goto err_out_probe;
5273
5274                 /*
5275                  * Need to warn users if this image is the one being
5276                  * mapped and has a parent.
5277                  */
5278                 if (!depth && rbd_dev->parent_spec)
5279                         rbd_warn(rbd_dev,
5280                                  "WARNING: kernel layering is EXPERIMENTAL!");
5281         }
5282
5283         ret = rbd_dev_probe_parent(rbd_dev, depth);
5284         if (ret)
5285                 goto err_out_probe;
5286
5287         dout("discovered format %u image, header name is %s\n",
5288                 rbd_dev->image_format, rbd_dev->header_oid.name);
5289         return 0;
5290
5291 err_out_probe:
5292         rbd_dev_unprobe(rbd_dev);
5293 err_out_watch:
5294         if (!depth)
5295                 rbd_dev_header_unwatch_sync(rbd_dev);
5296 err_out_format:
5297         rbd_dev->image_format = 0;
5298         kfree(rbd_dev->spec->image_id);
5299         rbd_dev->spec->image_id = NULL;
5300         return ret;
5301 }
5302
5303 static ssize_t do_rbd_add(struct bus_type *bus,
5304                           const char *buf,
5305                           size_t count)
5306 {
5307         struct rbd_device *rbd_dev = NULL;
5308         struct ceph_options *ceph_opts = NULL;
5309         struct rbd_options *rbd_opts = NULL;
5310         struct rbd_spec *spec = NULL;
5311         struct rbd_client *rbdc;
5312         bool read_only;
5313         int rc;
5314
5315         if (!try_module_get(THIS_MODULE))
5316                 return -ENODEV;
5317
5318         /* parse add command */
5319         rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
5320         if (rc < 0)
5321                 goto out;
5322
5323         rbdc = rbd_get_client(ceph_opts);
5324         if (IS_ERR(rbdc)) {
5325                 rc = PTR_ERR(rbdc);
5326                 goto err_out_args;
5327         }
5328
5329         /* pick the pool */
5330         rc = rbd_add_get_pool_id(rbdc, spec->pool_name);
5331         if (rc < 0) {
5332                 if (rc == -ENOENT)
5333                         pr_info("pool %s does not exist\n", spec->pool_name);
5334                 goto err_out_client;
5335         }
5336         spec->pool_id = (u64)rc;
5337
5338         /* The ceph file layout needs to fit pool id in 32 bits */
5339
5340         if (spec->pool_id > (u64)U32_MAX) {
5341                 rbd_warn(NULL, "pool id too large (%llu > %u)",
5342                                 (unsigned long long)spec->pool_id, U32_MAX);
5343                 rc = -EIO;
5344                 goto err_out_client;
5345         }
5346
5347         rbd_dev = rbd_dev_create(rbdc, spec, rbd_opts);
5348         if (!rbd_dev) {
5349                 rc = -ENOMEM;
5350                 goto err_out_client;
5351         }
5352         rbdc = NULL;            /* rbd_dev now owns this */
5353         spec = NULL;            /* rbd_dev now owns this */
5354         rbd_opts = NULL;        /* rbd_dev now owns this */
5355
5356         down_write(&rbd_dev->header_rwsem);
5357         rc = rbd_dev_image_probe(rbd_dev, 0);
5358         if (rc < 0)
5359                 goto err_out_rbd_dev;
5360
5361         /* If we are mapping a snapshot it must be marked read-only */
5362
5363         read_only = rbd_dev->opts->read_only;
5364         if (rbd_dev->spec->snap_id != CEPH_NOSNAP)
5365                 read_only = true;
5366         rbd_dev->mapping.read_only = read_only;
5367
5368         rc = rbd_dev_device_setup(rbd_dev);
5369         if (rc) {
5370                 /*
5371                  * rbd_dev_header_unwatch_sync() can't be moved into
5372                  * rbd_dev_image_release() without refactoring, see
5373                  * commit 1f3ef78861ac.
5374                  */
5375                 rbd_dev_header_unwatch_sync(rbd_dev);
5376                 rbd_dev_image_release(rbd_dev);
5377                 goto out;
5378         }
5379
5380         rc = count;
5381 out:
5382         module_put(THIS_MODULE);
5383         return rc;
5384
5385 err_out_rbd_dev:
5386         up_write(&rbd_dev->header_rwsem);
5387         rbd_dev_destroy(rbd_dev);
5388 err_out_client:
5389         rbd_put_client(rbdc);
5390 err_out_args:
5391         rbd_spec_put(spec);
5392         kfree(rbd_opts);
5393         goto out;
5394 }
5395
5396 static ssize_t rbd_add(struct bus_type *bus,
5397                        const char *buf,
5398                        size_t count)
5399 {
5400         if (single_major)
5401                 return -EINVAL;
5402
5403         return do_rbd_add(bus, buf, count);
5404 }
5405
5406 static ssize_t rbd_add_single_major(struct bus_type *bus,
5407                                     const char *buf,
5408                                     size_t count)
5409 {
5410         return do_rbd_add(bus, buf, count);
5411 }
5412
5413 static void rbd_dev_device_release(struct rbd_device *rbd_dev)
5414 {
5415         rbd_free_disk(rbd_dev);
5416         clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
5417         device_del(&rbd_dev->dev);
5418         rbd_dev_mapping_clear(rbd_dev);
5419         if (!single_major)
5420                 unregister_blkdev(rbd_dev->major, rbd_dev->name);
5421         rbd_dev_id_put(rbd_dev);
5422 }
5423
5424 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev)
5425 {
5426         while (rbd_dev->parent) {
5427                 struct rbd_device *first = rbd_dev;
5428                 struct rbd_device *second = first->parent;
5429                 struct rbd_device *third;
5430
5431                 /*
5432                  * Follow to the parent with no grandparent and
5433                  * remove it.
5434                  */
5435                 while (second && (third = second->parent)) {
5436                         first = second;
5437                         second = third;
5438                 }
5439                 rbd_assert(second);
5440                 rbd_dev_image_release(second);
5441                 first->parent = NULL;
5442                 first->parent_overlap = 0;
5443
5444                 rbd_assert(first->parent_spec);
5445                 rbd_spec_put(first->parent_spec);
5446                 first->parent_spec = NULL;
5447         }
5448 }
5449
5450 static ssize_t do_rbd_remove(struct bus_type *bus,
5451                              const char *buf,
5452                              size_t count)
5453 {
5454         struct rbd_device *rbd_dev = NULL;
5455         struct list_head *tmp;
5456         int dev_id;
5457         unsigned long ul;
5458         bool already = false;
5459         int ret;
5460
5461         ret = kstrtoul(buf, 10, &ul);
5462         if (ret)
5463                 return ret;
5464
5465         /* convert to int; abort if we lost anything in the conversion */
5466         dev_id = (int)ul;
5467         if (dev_id != ul)
5468                 return -EINVAL;
5469
5470         ret = -ENOENT;
5471         spin_lock(&rbd_dev_list_lock);
5472         list_for_each(tmp, &rbd_dev_list) {
5473                 rbd_dev = list_entry(tmp, struct rbd_device, node);
5474                 if (rbd_dev->dev_id == dev_id) {
5475                         ret = 0;
5476                         break;
5477                 }
5478         }
5479         if (!ret) {
5480                 spin_lock_irq(&rbd_dev->lock);
5481                 if (rbd_dev->open_count)
5482                         ret = -EBUSY;
5483                 else
5484                         already = test_and_set_bit(RBD_DEV_FLAG_REMOVING,
5485                                                         &rbd_dev->flags);
5486                 spin_unlock_irq(&rbd_dev->lock);
5487         }
5488         spin_unlock(&rbd_dev_list_lock);
5489         if (ret < 0 || already)
5490                 return ret;
5491
5492         rbd_dev_header_unwatch_sync(rbd_dev);
5493
5494         /*
5495          * Don't free anything from rbd_dev->disk until after all
5496          * notifies are completely processed. Otherwise
5497          * rbd_bus_del_dev() will race with rbd_watch_cb(), resulting
5498          * in a potential use after free of rbd_dev->disk or rbd_dev.
5499          */
5500         rbd_dev_device_release(rbd_dev);
5501         rbd_dev_image_release(rbd_dev);
5502
5503         return count;
5504 }
5505
5506 static ssize_t rbd_remove(struct bus_type *bus,
5507                           const char *buf,
5508                           size_t count)
5509 {
5510         if (single_major)
5511                 return -EINVAL;
5512
5513         return do_rbd_remove(bus, buf, count);
5514 }
5515
5516 static ssize_t rbd_remove_single_major(struct bus_type *bus,
5517                                        const char *buf,
5518                                        size_t count)
5519 {
5520         return do_rbd_remove(bus, buf, count);
5521 }
5522
5523 /*
5524  * create control files in sysfs
5525  * /sys/bus/rbd/...
5526  */
5527 static int rbd_sysfs_init(void)
5528 {
5529         int ret;
5530
5531         ret = device_register(&rbd_root_dev);
5532         if (ret < 0)
5533                 return ret;
5534
5535         ret = bus_register(&rbd_bus_type);
5536         if (ret < 0)
5537                 device_unregister(&rbd_root_dev);
5538
5539         return ret;
5540 }
5541
5542 static void rbd_sysfs_cleanup(void)
5543 {
5544         bus_unregister(&rbd_bus_type);
5545         device_unregister(&rbd_root_dev);
5546 }
5547
5548 static int rbd_slab_init(void)
5549 {
5550         rbd_assert(!rbd_img_request_cache);
5551         rbd_img_request_cache = KMEM_CACHE(rbd_img_request, 0);
5552         if (!rbd_img_request_cache)
5553                 return -ENOMEM;
5554
5555         rbd_assert(!rbd_obj_request_cache);
5556         rbd_obj_request_cache = KMEM_CACHE(rbd_obj_request, 0);
5557         if (!rbd_obj_request_cache)
5558                 goto out_err;
5559
5560         rbd_assert(!rbd_segment_name_cache);
5561         rbd_segment_name_cache = kmem_cache_create("rbd_segment_name",
5562                                         CEPH_MAX_OID_NAME_LEN + 1, 1, 0, NULL);
5563         if (rbd_segment_name_cache)
5564                 return 0;
5565 out_err:
5566         kmem_cache_destroy(rbd_obj_request_cache);
5567         rbd_obj_request_cache = NULL;
5568
5569         kmem_cache_destroy(rbd_img_request_cache);
5570         rbd_img_request_cache = NULL;
5571
5572         return -ENOMEM;
5573 }
5574
5575 static void rbd_slab_exit(void)
5576 {
5577         rbd_assert(rbd_segment_name_cache);
5578         kmem_cache_destroy(rbd_segment_name_cache);
5579         rbd_segment_name_cache = NULL;
5580
5581         rbd_assert(rbd_obj_request_cache);
5582         kmem_cache_destroy(rbd_obj_request_cache);
5583         rbd_obj_request_cache = NULL;
5584
5585         rbd_assert(rbd_img_request_cache);
5586         kmem_cache_destroy(rbd_img_request_cache);
5587         rbd_img_request_cache = NULL;
5588 }
5589
5590 static int __init rbd_init(void)
5591 {
5592         int rc;
5593
5594         if (!libceph_compatible(NULL)) {
5595                 rbd_warn(NULL, "libceph incompatibility (quitting)");
5596                 return -EINVAL;
5597         }
5598
5599         rc = rbd_slab_init();
5600         if (rc)
5601                 return rc;
5602
5603         /*
5604          * The number of active work items is limited by the number of
5605          * rbd devices * queue depth, so leave @max_active at default.
5606          */
5607         rbd_wq = alloc_workqueue(RBD_DRV_NAME, WQ_MEM_RECLAIM, 0);
5608         if (!rbd_wq) {
5609                 rc = -ENOMEM;
5610                 goto err_out_slab;
5611         }
5612
5613         if (single_major) {
5614                 rbd_major = register_blkdev(0, RBD_DRV_NAME);
5615                 if (rbd_major < 0) {
5616                         rc = rbd_major;
5617                         goto err_out_wq;
5618                 }
5619         }
5620
5621         rc = rbd_sysfs_init();
5622         if (rc)
5623                 goto err_out_blkdev;
5624
5625         if (single_major)
5626                 pr_info("loaded (major %d)\n", rbd_major);
5627         else
5628                 pr_info("loaded\n");
5629
5630         return 0;
5631
5632 err_out_blkdev:
5633         if (single_major)
5634                 unregister_blkdev(rbd_major, RBD_DRV_NAME);
5635 err_out_wq:
5636         destroy_workqueue(rbd_wq);
5637 err_out_slab:
5638         rbd_slab_exit();
5639         return rc;
5640 }
5641
5642 static void __exit rbd_exit(void)
5643 {
5644         ida_destroy(&rbd_dev_id_ida);
5645         rbd_sysfs_cleanup();
5646         if (single_major)
5647                 unregister_blkdev(rbd_major, RBD_DRV_NAME);
5648         destroy_workqueue(rbd_wq);
5649         rbd_slab_exit();
5650 }
5651
5652 module_init(rbd_init);
5653 module_exit(rbd_exit);
5654
5655 MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
5656 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
5657 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
5658 /* following authorship retained from original osdblk.c */
5659 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
5660
5661 MODULE_DESCRIPTION("RADOS Block Device (RBD) driver");
5662 MODULE_LICENSE("GPL");