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