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/fs.h>
  42 #include <linux/blkdev.h>
  43 #include <linux/slab.h>
  44 #include <linux/idr.h>
  45 #include <linux/workqueue.h>
  46
  47 #include "rbd_types.h"
  48
  49 #define RBD_DEBUG       /* Activate rbd_assert() calls */
  50
  51 /*
  52  * The basic unit of block I/O is a sector.  It is interpreted in a
  53  * number of contexts in Linux (blk, bio, genhd), but the default is
  54  * universally 512 bytes.  These symbols are just slightly more
  55  * meaningful than the bare numbers they represent.
  56  */
  57 #define SECTOR_SHIFT    9
  58 #define SECTOR_SIZE     (1ULL << SECTOR_SHIFT)
  59
  60 /*
  61  * Increment the given counter and return its updated value.
  62  * If the counter is already 0 it will not be incremented.
  63  * If the counter is already at its maximum value returns
  64  * -EINVAL without updating it.
  65  */
  66 static int atomic_inc_return_safe(atomic_t *v)
  67 {
  68         unsigned int counter;
  69
  70         counter = (unsigned int)__atomic_add_unless(v, 1, 0);
  71         if (counter <= (unsigned int)INT_MAX)
  72                 return (int)counter;
  73
  74         atomic_dec(v);
  75
  76         return -EINVAL;
  77 }
  78
  79 /* Decrement the counter.  Return the resulting value, or -EINVAL */
  80 static int atomic_dec_return_safe(atomic_t *v)
  81 {
  82         int counter;
  83
  84         counter = atomic_dec_return(v);
  85         if (counter >= 0)
  86                 return counter;
  87
  88         atomic_inc(v);
  89
  90         return -EINVAL;
  91 }
  92
  93 #define RBD_DRV_NAME "rbd"
  94
  95 #define RBD_MINORS_PER_MAJOR            256
  96 #define RBD_SINGLE_MAJOR_PART_SHIFT     4
  97
  98 #define RBD_SNAP_DEV_NAME_PREFIX        "snap_"
  99 #define RBD_MAX_SNAP_NAME_LEN   \
 100                         (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
 101
 102 #define RBD_MAX_SNAP_COUNT      510     /* allows max snapc to fit in 4KB */
 103
 104 #define RBD_SNAP_HEAD_NAME      "-"
 105
 106 #define BAD_SNAP_INDEX  U32_MAX         /* invalid index into snap array */
 107
 108 /* This allows a single page to hold an image name sent by OSD */
 109 #define RBD_IMAGE_NAME_LEN_MAX  (PAGE_SIZE - sizeof (__le32) - 1)
 110 #define RBD_IMAGE_ID_LEN_MAX    64
 111
 112 #define RBD_OBJ_PREFIX_LEN_MAX  64
 113
 114 /* Feature bits */
 115
 116 #define RBD_FEATURE_LAYERING    (1<<0)
 117 #define RBD_FEATURE_STRIPINGV2  (1<<1)
 118 #define RBD_FEATURES_ALL \
 119             (RBD_FEATURE_LAYERING | RBD_FEATURE_STRIPINGV2)
 120
 121 /* Features supported by this (client software) implementation. */
 122
 123 #define RBD_FEATURES_SUPPORTED  (RBD_FEATURES_ALL)
 124
 125 /*
 126  * An RBD device name will be "rbd#", where the "rbd" comes from
 127  * RBD_DRV_NAME above, and # is a unique integer identifier.
 128  * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
 129  * enough to hold all possible device names.
 130  */
 131 #define DEV_NAME_LEN            32
 132 #define MAX_INT_FORMAT_WIDTH    ((5 * sizeof (int)) / 2 + 1)
 133
 134 /*
 135  * block device image metadata (in-memory version)
 136  */
 137 struct rbd_image_header {
 138         /* These six fields never change for a given rbd image */
 139         char *object_prefix;
 140         __u8 obj_order;
 141         __u8 crypt_type;
 142         __u8 comp_type;
 143         u64 stripe_unit;
 144         u64 stripe_count;
 145         u64 features;           /* Might be changeable someday? */
 146
 147         /* The remaining fields need to be updated occasionally */
 148         u64 image_size;
 149         struct ceph_snap_context *snapc;
 150         char *snap_names;       /* format 1 only */
 151         u64 *snap_sizes;        /* format 1 only */
 152 };
 153
 154 /*
 155  * An rbd image specification.
 156  *
 157  * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
 158  * identify an image.  Each rbd_dev structure includes a pointer to
 159  * an rbd_spec structure that encapsulates this identity.
 160  *
 161  * Each of the id's in an rbd_spec has an associated name.  For a
 162  * user-mapped image, the names are supplied and the id's associated
 163  * with them are looked up.  For a layered image, a parent image is
 164  * defined by the tuple, and the names are looked up.
 165  *
 166  * An rbd_dev structure contains a parent_spec pointer which is
 167  * non-null if the image it represents is a child in a layered
 168  * image.  This pointer will refer to the rbd_spec structure used
 169  * by the parent rbd_dev for its own identity (i.e., the structure
 170  * is shared between the parent and child).
 171  *
 172  * Since these structures are populated once, during the discovery
 173  * phase of image construction, they are effectively immutable so
 174  * we make no effort to synchronize access to them.
 175  *
 176  * Note that code herein does not assume the image name is known (it
 177  * could be a null pointer).
 178  */
 179 struct rbd_spec {
 180         u64             pool_id;
 181         const char      *pool_name;
 182
 183         const char      *image_id;
 184         const char      *image_name;
 185
 186         u64             snap_id;
 187         const char      *snap_name;
 188
 189         struct kref     kref;
 190 };
 191
 192 /*
 193  * an instance of the client.  multiple devices may share an rbd client.
 194  */
 195 struct rbd_client {
 196         struct ceph_client      *client;
 197         struct kref             kref;
 198         struct list_head        node;
 199 };
 200
 201 struct rbd_img_request;
 202 typedef void (*rbd_img_callback_t)(struct rbd_img_request *);
 203
 204 #define BAD_WHICH       U32_MAX         /* Good which or bad which, which? */
 205
 206 struct rbd_obj_request;
 207 typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);
 208
 209 enum obj_request_type {
 210         OBJ_REQUEST_NODATA, OBJ_REQUEST_BIO, OBJ_REQUEST_PAGES
 211 };
 212
 213 enum obj_operation_type {
 214         OBJ_OP_WRITE,
 215         OBJ_OP_READ,
 216         OBJ_OP_DISCARD,
 217 };
 218
 219 enum obj_req_flags {
 220         OBJ_REQ_DONE,           /* completion flag: not done = 0, done = 1 */
 221         OBJ_REQ_IMG_DATA,       /* object usage: standalone = 0, image = 1 */
 222         OBJ_REQ_KNOWN,          /* EXISTS flag valid: no = 0, yes = 1 */
 223         OBJ_REQ_EXISTS,         /* target exists: no = 0, yes = 1 */
 224 };
 225
 226 struct rbd_obj_request {
 227         const char              *object_name;
 228         u64                     offset;         /* object start byte */
 229         u64                     length;         /* bytes from offset */
 230         unsigned long           flags;
 231
 232         /*
 233          * An object request associated with an image will have its
 234          * img_data flag set; a standalone object request will not.
 235          *
 236          * A standalone object request will have which == BAD_WHICH
 237          * and a null obj_request pointer.
 238          *
 239          * An object request initiated in support of a layered image
 240          * object (to check for its existence before a write) will
 241          * have which == BAD_WHICH and a non-null obj_request pointer.
 242          *
 243          * Finally, an object request for rbd image data will have
 244          * which != BAD_WHICH, and will have a non-null img_request
 245          * pointer.  The value of which will be in the range
 246          * 0..(img_request->obj_request_count-1).
 247          */
 248         union {
 249                 struct rbd_obj_request  *obj_request;   /* STAT op */
 250                 struct {
 251                         struct rbd_img_request  *img_request;
 252                         u64                     img_offset;
 253                         /* links for img_request->obj_requests list */
 254                         struct list_head        links;
 255                 };
 256         };
 257         u32                     which;          /* posn image request list */
 258
 259         enum obj_request_type   type;
 260         union {
 261                 struct bio      *bio_list;
 262                 struct {
 263                         struct page     **pages;
 264                         u32             page_count;
 265                 };
 266         };
 267         struct page             **copyup_pages;
 268         u32                     copyup_page_count;
 269
 270         struct ceph_osd_request *osd_req;
 271
 272         u64                     xferred;        /* bytes transferred */
 273         int                     result;
 274
 275         rbd_obj_callback_t      callback;
 276         struct completion       completion;
 277
 278         struct kref             kref;
 279 };
 280
 281 enum img_req_flags {
 282         IMG_REQ_WRITE,          /* I/O direction: read = 0, write = 1 */
 283         IMG_REQ_CHILD,          /* initiator: block = 0, child image = 1 */
 284         IMG_REQ_LAYERED,        /* ENOENT handling: normal = 0, layered = 1 */
 285         IMG_REQ_DISCARD,        /* discard: normal = 0, discard request = 1 */
 286 };
 287
 288 struct rbd_img_request {
 289         struct rbd_device       *rbd_dev;
 290         u64                     offset; /* starting image byte offset */
 291         u64                     length; /* byte count from offset */
 292         unsigned long           flags;
 293         union {
 294                 u64                     snap_id;        /* for reads */
 295                 struct ceph_snap_context *snapc;        /* for writes */
 296         };
 297         union {
 298                 struct request          *rq;            /* block request */
 299                 struct rbd_obj_request  *obj_request;   /* obj req initiator */
 300         };
 301         struct page             **copyup_pages;
 302         u32                     copyup_page_count;
 303         spinlock_t              completion_lock;/* protects next_completion */
 304         u32                     next_completion;
 305         rbd_img_callback_t      callback;
 306         u64                     xferred;/* aggregate bytes transferred */
 307         int                     result; /* first nonzero obj_request result */
 308
 309         u32                     obj_request_count;
 310         struct list_head        obj_requests;   /* rbd_obj_request structs */
 311
 312         struct kref             kref;
 313 };
 314
 315 #define for_each_obj_request(ireq, oreq) \
 316         list_for_each_entry(oreq, &(ireq)->obj_requests, links)
 317 #define for_each_obj_request_from(ireq, oreq) \
 318         list_for_each_entry_from(oreq, &(ireq)->obj_requests, links)
 319 #define for_each_obj_request_safe(ireq, oreq, n) \
 320         list_for_each_entry_safe_reverse(oreq, n, &(ireq)->obj_requests, links)
 321
 322 struct rbd_mapping {
 323         u64                     size;
 324         u64                     features;
 325         bool                    read_only;
 326 };
 327
 328 /*
 329  * a single device
 330  */
 331 struct rbd_device {
 332         int                     dev_id;         /* blkdev unique id */
 333
 334         int                     major;          /* blkdev assigned major */
 335         int                     minor;
 336         struct gendisk          *disk;          /* blkdev's gendisk and rq */
 337
 338         u32                     image_format;   /* Either 1 or 2 */
 339         struct rbd_client       *rbd_client;
 340
 341         char                    name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
 342
 343         struct list_head        rq_queue;       /* incoming rq queue */
 344         spinlock_t              lock;           /* queue, flags, open_count */
 345         struct workqueue_struct *rq_wq;
 346         struct work_struct      rq_work;
 347
 348         struct rbd_image_header header;
 349         unsigned long           flags;          /* possibly lock protected */
 350         struct rbd_spec         *spec;
 351
 352         char                    *header_name;
 353
 354         struct ceph_file_layout layout;
 355
 356         struct ceph_osd_event   *watch_event;
 357         struct rbd_obj_request  *watch_request;
 358
 359         struct rbd_spec         *parent_spec;
 360         u64                     parent_overlap;
 361         atomic_t                parent_ref;
 362         struct rbd_device       *parent;
 363
 364         /* protects updating the header */
 365         struct rw_semaphore     header_rwsem;
 366
 367         struct rbd_mapping      mapping;
 368
 369         struct list_head        node;
 370
 371         /* sysfs related */
 372         struct device           dev;
 373         unsigned long           open_count;     /* protected by lock */
 374 };
 375
 376 /*
 377  * Flag bits for rbd_dev->flags.  If atomicity is required,
 378  * rbd_dev->lock is used to protect access.
 379  *
 380  * Currently, only the "removing" flag (which is coupled with the
 381  * "open_count" field) requires atomic access.
 382  */
 383 enum rbd_dev_flags {
 384         RBD_DEV_FLAG_EXISTS,    /* mapped snapshot has not been deleted */
 385         RBD_DEV_FLAG_REMOVING,  /* this mapping is being removed */
 386 };
 387
 388 static DEFINE_MUTEX(client_mutex);      /* Serialize client creation */
 389
 390 static LIST_HEAD(rbd_dev_list);    /* devices */
 391 static DEFINE_SPINLOCK(rbd_dev_list_lock);
 392
 393 static LIST_HEAD(rbd_client_list);              /* clients */
 394 static DEFINE_SPINLOCK(rbd_client_list_lock);
 395
 396 /* Slab caches for frequently-allocated structures */
 397
 398 static struct kmem_cache        *rbd_img_request_cache;
 399 static struct kmem_cache        *rbd_obj_request_cache;
 400 static struct kmem_cache        *rbd_segment_name_cache;
 401
 402 static int rbd_major;
 403 static DEFINE_IDA(rbd_dev_id_ida);
 404
 405 /*
 406  * Default to false for now, as single-major requires >= 0.75 version of
 407  * userspace rbd utility.
 408  */
 409 static bool single_major = false;
 410 module_param(single_major, bool, S_IRUGO);
 411 MODULE_PARM_DESC(single_major, "Use a single major number for all rbd devices (default: false)");
 412
 413 static int rbd_img_request_submit(struct rbd_img_request *img_request);
 414
 415 static void rbd_dev_device_release(struct device *dev);
 416
 417 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
 418                        size_t count);
 419 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
 420                           size_t count);
 421 static ssize_t rbd_add_single_major(struct bus_type *bus, const char *buf,
 422                                     size_t count);
 423 static ssize_t rbd_remove_single_major(struct bus_type *bus, const char *buf,
 424                                        size_t count);
 425 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, bool mapping);
 426 static void rbd_spec_put(struct rbd_spec *spec);
 427
 428 static int rbd_dev_id_to_minor(int dev_id)
 429 {
 430         return dev_id << RBD_SINGLE_MAJOR_PART_SHIFT;
 431 }
 432
 433 static int minor_to_rbd_dev_id(int minor)
 434 {
 435         return minor >> RBD_SINGLE_MAJOR_PART_SHIFT;
 436 }
 437
 438 static BUS_ATTR(add, S_IWUSR, NULL, rbd_add);
 439 static BUS_ATTR(remove, S_IWUSR, NULL, rbd_remove);
 440 static BUS_ATTR(add_single_major, S_IWUSR, NULL, rbd_add_single_major);
 441 static BUS_ATTR(remove_single_major, S_IWUSR, NULL, rbd_remove_single_major);
 442
 443 static struct attribute *rbd_bus_attrs[] = {
 444         &bus_attr_add.attr,
 445         &bus_attr_remove.attr,
 446         &bus_attr_add_single_major.attr,
 447         &bus_attr_remove_single_major.attr,
 448         NULL,
 449 };
 450
 451 static umode_t rbd_bus_is_visible(struct kobject *kobj,
 452                                   struct attribute *attr, int index)
 453 {
 454         if (!single_major &&
 455             (attr == &bus_attr_add_single_major.attr ||
 456              attr == &bus_attr_remove_single_major.attr))
 457                 return 0;
 458
 459         return attr->mode;
 460 }
 461
 462 static const struct attribute_group rbd_bus_group = {
 463         .attrs = rbd_bus_attrs,
 464         .is_visible = rbd_bus_is_visible,
 465 };
 466 __ATTRIBUTE_GROUPS(rbd_bus);
 467
 468 static struct bus_type rbd_bus_type = {
 469         .name           = "rbd",
 470         .bus_groups     = rbd_bus_groups,
 471 };
 472
 473 static void rbd_root_dev_release(struct device *dev)
 474 {
 475 }
 476
 477 static struct device rbd_root_dev = {
 478         .init_name =    "rbd",
 479         .release =      rbd_root_dev_release,
 480 };
 481
 482 static __printf(2, 3)
 483 void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
 484 {
 485         struct va_format vaf;
 486         va_list args;
 487
 488         va_start(args, fmt);
 489         vaf.fmt = fmt;
 490         vaf.va = &args;
 491
 492         if (!rbd_dev)
 493                 printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
 494         else if (rbd_dev->disk)
 495                 printk(KERN_WARNING "%s: %s: %pV\n",
 496                         RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
 497         else if (rbd_dev->spec && rbd_dev->spec->image_name)
 498                 printk(KERN_WARNING "%s: image %s: %pV\n",
 499                         RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
 500         else if (rbd_dev->spec && rbd_dev->spec->image_id)
 501                 printk(KERN_WARNING "%s: id %s: %pV\n",
 502                         RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
 503         else    /* punt */
 504                 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
 505                         RBD_DRV_NAME, rbd_dev, &vaf);
 506         va_end(args);
 507 }
 508
 509 #ifdef RBD_DEBUG
 510 #define rbd_assert(expr)                                                \
 511                 if (unlikely(!(expr))) {                                \
 512                         printk(KERN_ERR "\nAssertion failure in %s() "  \
 513                                                 "at line %d:\n\n"       \
 514                                         "\trbd_assert(%s);\n\n",        \
 515                                         __func__, __LINE__, #expr);     \
 516                         BUG();                                          \
 517                 }
 518 #else /* !RBD_DEBUG */
 519 #  define rbd_assert(expr)      ((void) 0)
 520 #endif /* !RBD_DEBUG */
 521
 522 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request);
 523 static void rbd_img_parent_read(struct rbd_obj_request *obj_request);
 524 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev);
 525
 526 static int rbd_dev_refresh(struct rbd_device *rbd_dev);
 527 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev);
 528 static int rbd_dev_header_info(struct rbd_device *rbd_dev);
 529 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev);
 530 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
 531                                         u64 snap_id);
 532 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
 533                                 u8 *order, u64 *snap_size);
 534 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
 535                 u64 *snap_features);
 536 static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name);
 537
 538 static int rbd_open(struct block_device *bdev, fmode_t mode)
 539 {
 540         struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
 541         bool removing = false;
 542
 543         if ((mode & FMODE_WRITE) && rbd_dev->mapping.read_only)
 544                 return -EROFS;
 545
 546         spin_lock_irq(&rbd_dev->lock);
 547         if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
 548                 removing = true;
 549         else
 550                 rbd_dev->open_count++;
 551         spin_unlock_irq(&rbd_dev->lock);
 552         if (removing)
 553                 return -ENOENT;
 554
 555         (void) get_device(&rbd_dev->dev);
 556
 557         return 0;
 558 }
 559
 560 static void rbd_release(struct gendisk *disk, fmode_t mode)
 561 {
 562         struct rbd_device *rbd_dev = disk->private_data;
 563         unsigned long open_count_before;
 564
 565         spin_lock_irq(&rbd_dev->lock);
 566         open_count_before = rbd_dev->open_count--;
 567         spin_unlock_irq(&rbd_dev->lock);
 568         rbd_assert(open_count_before > 0);
 569
 570         put_device(&rbd_dev->dev);
 571 }
 572
 573 static int rbd_ioctl_set_ro(struct rbd_device *rbd_dev, unsigned long arg)
 574 {
 575         int ret = 0;
 576         int val;
 577         bool ro;
 578         bool ro_changed = false;
 579
 580         /* get_user() may sleep, so call it before taking rbd_dev->lock */
 581         if (get_user(val, (int __user *)(arg)))
 582                 return -EFAULT;
 583
 584         ro = val ? true : false;
 585         /* Snapshot doesn't allow to write*/
 586         if (rbd_dev->spec->snap_id != CEPH_NOSNAP && !ro)
 587                 return -EROFS;
 588
 589         spin_lock_irq(&rbd_dev->lock);
 590         /* prevent others open this device */
 591         if (rbd_dev->open_count > 1) {
 592                 ret = -EBUSY;
 593                 goto out;
 594         }
 595
 596         if (rbd_dev->mapping.read_only != ro) {
 597                 rbd_dev->mapping.read_only = ro;
 598                 ro_changed = true;
 599         }
 600
 601 out:
 602         spin_unlock_irq(&rbd_dev->lock);
 603         /* set_disk_ro() may sleep, so call it after releasing rbd_dev->lock */
 604         if (ret == 0 && ro_changed)
 605                 set_disk_ro(rbd_dev->disk, ro ? 1 : 0);
 606
 607         return ret;
 608 }
 609
 610 static int rbd_ioctl(struct block_device *bdev, fmode_t mode,
 611                         unsigned int cmd, unsigned long arg)
 612 {
 613         struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
 614         int ret = 0;
 615
 616         switch (cmd) {
 617         case BLKROSET:
 618                 ret = rbd_ioctl_set_ro(rbd_dev, arg);
 619                 break;
 620         default:
 621                 ret = -ENOTTY;
 622         }
 623
 624         return ret;
 625 }
 626
 627 #ifdef CONFIG_COMPAT
 628 static int rbd_compat_ioctl(struct block_device *bdev, fmode_t mode,
 629                                 unsigned int cmd, unsigned long arg)
 630 {
 631         return rbd_ioctl(bdev, mode, cmd, arg);
 632 }
 633 #endif /* CONFIG_COMPAT */
 634
 635 static const struct block_device_operations rbd_bd_ops = {
 636         .owner                  = THIS_MODULE,
 637         .open                   = rbd_open,
 638         .release                = rbd_release,
 639         .ioctl                  = rbd_ioctl,
 640 #ifdef CONFIG_COMPAT
 641         .compat_ioctl           = rbd_compat_ioctl,
 642 #endif
 643 };
 644
 645 /*
 646  * Initialize an rbd client instance.  Success or not, this function
 647  * consumes ceph_opts.  Caller holds client_mutex.
 648  */
 649 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
 650 {
 651         struct rbd_client *rbdc;
 652         int ret = -ENOMEM;
 653
 654         dout("%s:\n", __func__);
 655         rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
 656         if (!rbdc)
 657                 goto out_opt;
 658
 659         kref_init(&rbdc->kref);
 660         INIT_LIST_HEAD(&rbdc->node);
 661
 662         rbdc->client = ceph_create_client(ceph_opts, rbdc, 0, 0);
 663         if (IS_ERR(rbdc->client))
 664                 goto out_rbdc;
 665         ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
 666
 667         ret = ceph_open_session(rbdc->client);
 668         if (ret < 0)
 669                 goto out_client;
 670
 671         spin_lock(&rbd_client_list_lock);
 672         list_add_tail(&rbdc->node, &rbd_client_list);
 673         spin_unlock(&rbd_client_list_lock);
 674
 675         dout("%s: rbdc %p\n", __func__, rbdc);
 676
 677         return rbdc;
 678 out_client:
 679         ceph_destroy_client(rbdc->client);
 680 out_rbdc:
 681         kfree(rbdc);
 682 out_opt:
 683         if (ceph_opts)
 684                 ceph_destroy_options(ceph_opts);
 685         dout("%s: error %d\n", __func__, ret);
 686
 687         return ERR_PTR(ret);
 688 }
 689
 690 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
 691 {
 692         kref_get(&rbdc->kref);
 693
 694         return rbdc;
 695 }
 696
 697 /*
 698  * Find a ceph client with specific addr and configuration.  If
 699  * found, bump its reference count.
 700  */
 701 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
 702 {
 703         struct rbd_client *client_node;
 704         bool found = false;
 705
 706         if (ceph_opts->flags & CEPH_OPT_NOSHARE)
 707                 return NULL;
 708
 709         spin_lock(&rbd_client_list_lock);
 710         list_for_each_entry(client_node, &rbd_client_list, node) {
 711                 if (!ceph_compare_options(ceph_opts, client_node->client)) {
 712                         __rbd_get_client(client_node);
 713
 714                         found = true;
 715                         break;
 716                 }
 717         }
 718         spin_unlock(&rbd_client_list_lock);
 719
 720         return found ? client_node : NULL;
 721 }
 722
 723 /*
 724  * mount options
 725  */
 726 enum {
 727         Opt_last_int,
 728         /* int args above */
 729         Opt_last_string,
 730         /* string args above */
 731         Opt_read_only,
 732         Opt_read_write,
 733         /* Boolean args above */
 734         Opt_last_bool,
 735 };
 736
 737 static match_table_t rbd_opts_tokens = {
 738         /* int args above */
 739         /* string args above */
 740         {Opt_read_only, "read_only"},
 741         {Opt_read_only, "ro"},          /* Alternate spelling */
 742         {Opt_read_write, "read_write"},
 743         {Opt_read_write, "rw"},         /* Alternate spelling */
 744         /* Boolean args above */
 745         {-1, NULL}
 746 };
 747
 748 struct rbd_options {
 749         bool    read_only;
 750 };
 751
 752 #define RBD_READ_ONLY_DEFAULT   false
 753
 754 static int parse_rbd_opts_token(char *c, void *private)
 755 {
 756         struct rbd_options *rbd_opts = private;
 757         substring_t argstr[MAX_OPT_ARGS];
 758         int token, intval, ret;
 759
 760         token = match_token(c, rbd_opts_tokens, argstr);
 761         if (token < 0)
 762                 return -EINVAL;
 763
 764         if (token < Opt_last_int) {
 765                 ret = match_int(&argstr[0], &intval);
 766                 if (ret < 0) {
 767                         pr_err("bad mount option arg (not int) "
 768                                "at '%s'\n", c);
 769                         return ret;
 770                 }
 771                 dout("got int token %d val %d\n", token, intval);
 772         } else if (token > Opt_last_int && token < Opt_last_string) {
 773                 dout("got string token %d val %s\n", token,
 774                      argstr[0].from);
 775         } else if (token > Opt_last_string && token < Opt_last_bool) {
 776                 dout("got Boolean token %d\n", token);
 777         } else {
 778                 dout("got token %d\n", token);
 779         }
 780
 781         switch (token) {
 782         case Opt_read_only:
 783                 rbd_opts->read_only = true;
 784                 break;
 785         case Opt_read_write:
 786                 rbd_opts->read_only = false;
 787                 break;
 788         default:
 789                 rbd_assert(false);
 790                 break;
 791         }
 792         return 0;
 793 }
 794
 795 static char* obj_op_name(enum obj_operation_type op_type)
 796 {
 797         switch (op_type) {
 798         case OBJ_OP_READ:
 799                 return "read";
 800         case OBJ_OP_WRITE:
 801                 return "write";
 802         case OBJ_OP_DISCARD:
 803                 return "discard";
 804         default:
 805                 return "???";
 806         }
 807 }
 808
 809 /*
 810  * Get a ceph client with specific addr and configuration, if one does
 811  * not exist create it.  Either way, ceph_opts is consumed by this
 812  * function.
 813  */
 814 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
 815 {
 816         struct rbd_client *rbdc;
 817
 818         mutex_lock_nested(&client_mutex, SINGLE_DEPTH_NESTING);
 819         rbdc = rbd_client_find(ceph_opts);
 820         if (rbdc)       /* using an existing client */
 821                 ceph_destroy_options(ceph_opts);
 822         else
 823                 rbdc = rbd_client_create(ceph_opts);
 824         mutex_unlock(&client_mutex);
 825
 826         return rbdc;
 827 }
 828
 829 /*
 830  * Destroy ceph client
 831  *
 832  * Caller must hold rbd_client_list_lock.
 833  */
 834 static void rbd_client_release(struct kref *kref)
 835 {
 836         struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
 837
 838         dout("%s: rbdc %p\n", __func__, rbdc);
 839         spin_lock(&rbd_client_list_lock);
 840         list_del(&rbdc->node);
 841         spin_unlock(&rbd_client_list_lock);
 842
 843         ceph_destroy_client(rbdc->client);
 844         kfree(rbdc);
 845 }
 846
 847 /*
 848  * Drop reference to ceph client node. If it's not referenced anymore, release
 849  * it.
 850  */
 851 static void rbd_put_client(struct rbd_client *rbdc)
 852 {
 853         if (rbdc)
 854                 kref_put(&rbdc->kref, rbd_client_release);
 855 }
 856
 857 static bool rbd_image_format_valid(u32 image_format)
 858 {
 859         return image_format == 1 || image_format == 2;
 860 }
 861
 862 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
 863 {
 864         size_t size;
 865         u32 snap_count;
 866
 867         /* The header has to start with the magic rbd header text */
 868         if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
 869                 return false;
 870
 871         /* The bio layer requires at least sector-sized I/O */
 872
 873         if (ondisk->options.order < SECTOR_SHIFT)
 874                 return false;
 875
 876         /* If we use u64 in a few spots we may be able to loosen this */
 877
 878         if (ondisk->options.order > 8 * sizeof (int) - 1)
 879                 return false;
 880
 881         /*
 882          * The size of a snapshot header has to fit in a size_t, and
 883          * that limits the number of snapshots.
 884          */
 885         snap_count = le32_to_cpu(ondisk->snap_count);
 886         size = SIZE_MAX - sizeof (struct ceph_snap_context);
 887         if (snap_count > size / sizeof (__le64))
 888                 return false;
 889
 890         /*
 891          * Not only that, but the size of the entire the snapshot
 892          * header must also be representable in a size_t.
 893          */
 894         size -= snap_count * sizeof (__le64);
 895         if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
 896                 return false;
 897
 898         return true;
 899 }
 900
 901 /*
 902  * Fill an rbd image header with information from the given format 1
 903  * on-disk header.
 904  */
 905 static int rbd_header_from_disk(struct rbd_device *rbd_dev,
 906                                  struct rbd_image_header_ondisk *ondisk)
 907 {
 908         struct rbd_image_header *header = &rbd_dev->header;
 909         bool first_time = header->object_prefix == NULL;
 910         struct ceph_snap_context *snapc;
 911         char *object_prefix = NULL;
 912         char *snap_names = NULL;
 913         u64 *snap_sizes = NULL;
 914         u32 snap_count;
 915         size_t size;
 916         int ret = -ENOMEM;
 917         u32 i;
 918
 919         /* Allocate this now to avoid having to handle failure below */
 920
 921         if (first_time) {
 922                 size_t len;
 923
 924                 len = strnlen(ondisk->object_prefix,
 925                                 sizeof (ondisk->object_prefix));
 926                 object_prefix = kmalloc(len + 1, GFP_KERNEL);
 927                 if (!object_prefix)
 928                         return -ENOMEM;
 929                 memcpy(object_prefix, ondisk->object_prefix, len);
 930                 object_prefix[len] = '\0';
 931         }
 932
 933         /* Allocate the snapshot context and fill it in */
 934
 935         snap_count = le32_to_cpu(ondisk->snap_count);
 936         snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
 937         if (!snapc)
 938                 goto out_err;
 939         snapc->seq = le64_to_cpu(ondisk->snap_seq);
 940         if (snap_count) {
 941                 struct rbd_image_snap_ondisk *snaps;
 942                 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
 943
 944                 /* We'll keep a copy of the snapshot names... */
 945
 946                 if (snap_names_len > (u64)SIZE_MAX)
 947                         goto out_2big;
 948                 snap_names = kmalloc(snap_names_len, GFP_KERNEL);
 949                 if (!snap_names)
 950                         goto out_err;
 951
 952                 /* ...as well as the array of their sizes. */
 953
 954                 size = snap_count * sizeof (*header->snap_sizes);
 955                 snap_sizes = kmalloc(size, GFP_KERNEL);
 956                 if (!snap_sizes)
 957                         goto out_err;
 958
 959                 /*
 960                  * Copy the names, and fill in each snapshot's id
 961                  * and size.
 962                  *
 963                  * Note that rbd_dev_v1_header_info() guarantees the
 964                  * ondisk buffer we're working with has
 965                  * snap_names_len bytes beyond the end of the
 966                  * snapshot id array, this memcpy() is safe.
 967                  */
 968                 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
 969                 snaps = ondisk->snaps;
 970                 for (i = 0; i < snap_count; i++) {
 971                         snapc->snaps[i] = le64_to_cpu(snaps[i].id);
 972                         snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
 973                 }
 974         }
 975
 976         /* We won't fail any more, fill in the header */
 977
 978         if (first_time) {
 979                 header->object_prefix = object_prefix;
 980                 header->obj_order = ondisk->options.order;
 981                 header->crypt_type = ondisk->options.crypt_type;
 982                 header->comp_type = ondisk->options.comp_type;
 983                 /* The rest aren't used for format 1 images */
 984                 header->stripe_unit = 0;
 985                 header->stripe_count = 0;
 986                 header->features = 0;
 987         } else {
 988                 ceph_put_snap_context(header->snapc);
 989                 kfree(header->snap_names);
 990                 kfree(header->snap_sizes);
 991         }
 992
 993         /* The remaining fields always get updated (when we refresh) */
 994
 995         header->image_size = le64_to_cpu(ondisk->image_size);
 996         header->snapc = snapc;
 997         header->snap_names = snap_names;
 998         header->snap_sizes = snap_sizes;
 999
1000         return 0;
1001 out_2big:
1002         ret = -EIO;
1003 out_err:
1004         kfree(snap_sizes);
1005         kfree(snap_names);
1006         ceph_put_snap_context(snapc);
1007         kfree(object_prefix);
1008
1009         return ret;
1010 }
1011
1012 static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1013 {
1014         const char *snap_name;
1015
1016         rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1017
1018         /* Skip over names until we find the one we are looking for */
1019
1020         snap_name = rbd_dev->header.snap_names;
1021         while (which--)
1022                 snap_name += strlen(snap_name) + 1;
1023
1024         return kstrdup(snap_name, GFP_KERNEL);
1025 }
1026
1027 /*
1028  * Snapshot id comparison function for use with qsort()/bsearch().
1029  * Note that result is for snapshots in *descending* order.
1030  */
1031 static int snapid_compare_reverse(const void *s1, const void *s2)
1032 {
1033         u64 snap_id1 = *(u64 *)s1;
1034         u64 snap_id2 = *(u64 *)s2;
1035
1036         if (snap_id1 < snap_id2)
1037                 return 1;
1038         return snap_id1 == snap_id2 ? 0 : -1;
1039 }
1040
1041 /*
1042  * Search a snapshot context to see if the given snapshot id is
1043  * present.
1044  *
1045  * Returns the position of the snapshot id in the array if it's found,
1046  * or BAD_SNAP_INDEX otherwise.
1047  *
1048  * Note: The snapshot array is in kept sorted (by the osd) in
1049  * reverse order, highest snapshot id first.
1050  */
1051 static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1052 {
1053         struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1054         u64 *found;
1055
1056         found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1057                                 sizeof (snap_id), snapid_compare_reverse);
1058
1059         return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1060 }
1061
1062 static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1063                                         u64 snap_id)
1064 {
1065         u32 which;
1066         const char *snap_name;
1067
1068         which = rbd_dev_snap_index(rbd_dev, snap_id);
1069         if (which == BAD_SNAP_INDEX)
1070                 return ERR_PTR(-ENOENT);
1071
1072         snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1073         return snap_name ? snap_name : ERR_PTR(-ENOMEM);
1074 }
1075
1076 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1077 {
1078         if (snap_id == CEPH_NOSNAP)
1079                 return RBD_SNAP_HEAD_NAME;
1080
1081         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1082         if (rbd_dev->image_format == 1)
1083                 return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1084
1085         return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1086 }
1087
1088 static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1089                                 u64 *snap_size)
1090 {
1091         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1092         if (snap_id == CEPH_NOSNAP) {
1093                 *snap_size = rbd_dev->header.image_size;
1094         } else if (rbd_dev->image_format == 1) {
1095                 u32 which;
1096
1097                 which = rbd_dev_snap_index(rbd_dev, snap_id);
1098                 if (which == BAD_SNAP_INDEX)
1099                         return -ENOENT;
1100
1101                 *snap_size = rbd_dev->header.snap_sizes[which];
1102         } else {
1103                 u64 size = 0;
1104                 int ret;
1105
1106                 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1107                 if (ret)
1108                         return ret;
1109
1110                 *snap_size = size;
1111         }
1112         return 0;
1113 }
1114
1115 static int rbd_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
1116                         u64 *snap_features)
1117 {
1118         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1119         if (snap_id == CEPH_NOSNAP) {
1120                 *snap_features = rbd_dev->header.features;
1121         } else if (rbd_dev->image_format == 1) {
1122                 *snap_features = 0;     /* No features for format 1 */
1123         } else {
1124                 u64 features = 0;
1125                 int ret;
1126
1127                 ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, &features);
1128                 if (ret)
1129                         return ret;
1130
1131                 *snap_features = features;
1132         }
1133         return 0;
1134 }
1135
1136 static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1137 {
1138         u64 snap_id = rbd_dev->spec->snap_id;
1139         u64 size = 0;
1140         u64 features = 0;
1141         int ret;
1142
1143         ret = rbd_snap_size(rbd_dev, snap_id, &size);
1144         if (ret)
1145                 return ret;
1146         ret = rbd_snap_features(rbd_dev, snap_id, &features);
1147         if (ret)
1148                 return ret;
1149
1150         rbd_dev->mapping.size = size;
1151         rbd_dev->mapping.features = features;
1152
1153         return 0;
1154 }
1155
1156 static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1157 {
1158         rbd_dev->mapping.size = 0;
1159         rbd_dev->mapping.features = 0;
1160 }
1161
1162 static void rbd_segment_name_free(const char *name)
1163 {
1164         /* The explicit cast here is needed to drop the const qualifier */
1165
1166         kmem_cache_free(rbd_segment_name_cache, (void *)name);
1167 }
1168
1169 static const char *rbd_segment_name(struct rbd_device *rbd_dev, u64 offset)
1170 {
1171         char *name;
1172         u64 segment;
1173         int ret;
1174         char *name_format;
1175
1176         name = kmem_cache_alloc(rbd_segment_name_cache, GFP_NOIO);
1177         if (!name)
1178                 return NULL;
1179         segment = offset >> rbd_dev->header.obj_order;
1180         name_format = "%s.%012llx";
1181         if (rbd_dev->image_format == 2)
1182                 name_format = "%s.%016llx";
1183         ret = snprintf(name, CEPH_MAX_OID_NAME_LEN + 1, name_format,
1184                         rbd_dev->header.object_prefix, segment);
1185         if (ret < 0 || ret > CEPH_MAX_OID_NAME_LEN) {
1186                 pr_err("error formatting segment name for #%llu (%d)\n",
1187                         segment, ret);
1188                 rbd_segment_name_free(name);
1189                 name = NULL;
1190         }
1191
1192         return name;
1193 }
1194
1195 static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset)
1196 {
1197         u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
1198
1199         return offset & (segment_size - 1);
1200 }
1201
1202 static u64 rbd_segment_length(struct rbd_device *rbd_dev,
1203                                 u64 offset, u64 length)
1204 {
1205         u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
1206
1207         offset &= segment_size - 1;
1208
1209         rbd_assert(length <= U64_MAX - offset);
1210         if (offset + length > segment_size)
1211                 length = segment_size - offset;
1212
1213         return length;
1214 }
1215
1216 /*
1217  * returns the size of an object in the image
1218  */
1219 static u64 rbd_obj_bytes(struct rbd_image_header *header)
1220 {
1221         return 1 << header->obj_order;
1222 }
1223
1224 /*
1225  * bio helpers
1226  */
1227
1228 static void bio_chain_put(struct bio *chain)
1229 {
1230         struct bio *tmp;
1231
1232         while (chain) {
1233                 tmp = chain;
1234                 chain = chain->bi_next;
1235                 bio_put(tmp);
1236         }
1237 }
1238
1239 /*
1240  * zeros a bio chain, starting at specific offset
1241  */
1242 static void zero_bio_chain(struct bio *chain, int start_ofs)
1243 {
1244         struct bio_vec bv;
1245         struct bvec_iter iter;
1246         unsigned long flags;
1247         void *buf;
1248         int pos = 0;
1249
1250         while (chain) {
1251                 bio_for_each_segment(bv, chain, iter) {
1252                         if (pos + bv.bv_len > start_ofs) {
1253                                 int remainder = max(start_ofs - pos, 0);
1254                                 buf = bvec_kmap_irq(&bv, &flags);
1255                                 memset(buf + remainder, 0,
1256                                        bv.bv_len - remainder);
1257                                 flush_dcache_page(bv.bv_page);
1258                                 bvec_kunmap_irq(buf, &flags);
1259                         }
1260                         pos += bv.bv_len;
1261                 }
1262
1263                 chain = chain->bi_next;
1264         }
1265 }
1266
1267 /*
1268  * similar to zero_bio_chain(), zeros data defined by a page array,
1269  * starting at the given byte offset from the start of the array and
1270  * continuing up to the given end offset.  The pages array is
1271  * assumed to be big enough to hold all bytes up to the end.
1272  */
1273 static void zero_pages(struct page **pages, u64 offset, u64 end)
1274 {
1275         struct page **page = &pages[offset >> PAGE_SHIFT];
1276
1277         rbd_assert(end > offset);
1278         rbd_assert(end - offset <= (u64)SIZE_MAX);
1279         while (offset < end) {
1280                 size_t page_offset;
1281                 size_t length;
1282                 unsigned long flags;
1283                 void *kaddr;
1284
1285                 page_offset = offset & ~PAGE_MASK;
1286                 length = min_t(size_t, PAGE_SIZE - page_offset, end - offset);
1287                 local_irq_save(flags);
1288                 kaddr = kmap_atomic(*page);
1289                 memset(kaddr + page_offset, 0, length);
1290                 flush_dcache_page(*page);
1291                 kunmap_atomic(kaddr);
1292                 local_irq_restore(flags);
1293
1294                 offset += length;
1295                 page++;
1296         }
1297 }
1298
1299 /*
1300  * Clone a portion of a bio, starting at the given byte offset
1301  * and continuing for the number of bytes indicated.
1302  */
1303 static struct bio *bio_clone_range(struct bio *bio_src,
1304                                         unsigned int offset,
1305                                         unsigned int len,
1306                                         gfp_t gfpmask)
1307 {
1308         struct bio *bio;
1309
1310         bio = bio_clone(bio_src, gfpmask);
1311         if (!bio)
1312                 return NULL;    /* ENOMEM */
1313
1314         bio_advance(bio, offset);
1315         bio->bi_iter.bi_size = len;
1316
1317         return bio;
1318 }
1319
1320 /*
1321  * Clone a portion of a bio chain, starting at the given byte offset
1322  * into the first bio in the source chain and continuing for the
1323  * number of bytes indicated.  The result is another bio chain of
1324  * exactly the given length, or a null pointer on error.
1325  *
1326  * The bio_src and offset parameters are both in-out.  On entry they
1327  * refer to the first source bio and the offset into that bio where
1328  * the start of data to be cloned is located.
1329  *
1330  * On return, bio_src is updated to refer to the bio in the source
1331  * chain that contains first un-cloned byte, and *offset will
1332  * contain the offset of that byte within that bio.
1333  */
1334 static struct bio *bio_chain_clone_range(struct bio **bio_src,
1335                                         unsigned int *offset,
1336                                         unsigned int len,
1337                                         gfp_t gfpmask)
1338 {
1339         struct bio *bi = *bio_src;
1340         unsigned int off = *offset;
1341         struct bio *chain = NULL;
1342         struct bio **end;
1343
1344         /* Build up a chain of clone bios up to the limit */
1345
1346         if (!bi || off >= bi->bi_iter.bi_size || !len)
1347                 return NULL;            /* Nothing to clone */
1348
1349         end = &chain;
1350         while (len) {
1351                 unsigned int bi_size;
1352                 struct bio *bio;
1353
1354                 if (!bi) {
1355                         rbd_warn(NULL, "bio_chain exhausted with %u left", len);
1356                         goto out_err;   /* EINVAL; ran out of bio's */
1357                 }
1358                 bi_size = min_t(unsigned int, bi->bi_iter.bi_size - off, len);
1359                 bio = bio_clone_range(bi, off, bi_size, gfpmask);
1360                 if (!bio)
1361                         goto out_err;   /* ENOMEM */
1362
1363                 *end = bio;
1364                 end = &bio->bi_next;
1365
1366                 off += bi_size;
1367                 if (off == bi->bi_iter.bi_size) {
1368                         bi = bi->bi_next;
1369                         off = 0;
1370                 }
1371                 len -= bi_size;
1372         }
1373         *bio_src = bi;
1374         *offset = off;
1375
1376         return chain;
1377 out_err:
1378         bio_chain_put(chain);
1379
1380         return NULL;
1381 }
1382
1383 /*
1384  * The default/initial value for all object request flags is 0.  For
1385  * each flag, once its value is set to 1 it is never reset to 0
1386  * again.
1387  */
1388 static void obj_request_img_data_set(struct rbd_obj_request *obj_request)
1389 {
1390         if (test_and_set_bit(OBJ_REQ_IMG_DATA, &obj_request->flags)) {
1391                 struct rbd_device *rbd_dev;
1392
1393                 rbd_dev = obj_request->img_request->rbd_dev;
1394                 rbd_warn(rbd_dev, "obj_request %p already marked img_data",
1395                         obj_request);
1396         }
1397 }
1398
1399 static bool obj_request_img_data_test(struct rbd_obj_request *obj_request)
1400 {
1401         smp_mb();
1402         return test_bit(OBJ_REQ_IMG_DATA, &obj_request->flags) != 0;
1403 }
1404
1405 static void obj_request_done_set(struct rbd_obj_request *obj_request)
1406 {
1407         if (test_and_set_bit(OBJ_REQ_DONE, &obj_request->flags)) {
1408                 struct rbd_device *rbd_dev = NULL;
1409
1410                 if (obj_request_img_data_test(obj_request))
1411                         rbd_dev = obj_request->img_request->rbd_dev;
1412                 rbd_warn(rbd_dev, "obj_request %p already marked done",
1413                         obj_request);
1414         }
1415 }
1416
1417 static bool obj_request_done_test(struct rbd_obj_request *obj_request)
1418 {
1419         smp_mb();
1420         return test_bit(OBJ_REQ_DONE, &obj_request->flags) != 0;
1421 }
1422
1423 /*
1424  * This sets the KNOWN flag after (possibly) setting the EXISTS
1425  * flag.  The latter is set based on the "exists" value provided.
1426  *
1427  * Note that for our purposes once an object exists it never goes
1428  * away again.  It's possible that the response from two existence
1429  * checks are separated by the creation of the target object, and
1430  * the first ("doesn't exist") response arrives *after* the second
1431  * ("does exist").  In that case we ignore the second one.
1432  */
1433 static void obj_request_existence_set(struct rbd_obj_request *obj_request,
1434                                 bool exists)
1435 {
1436         if (exists)
1437                 set_bit(OBJ_REQ_EXISTS, &obj_request->flags);
1438         set_bit(OBJ_REQ_KNOWN, &obj_request->flags);
1439         smp_mb();
1440 }
1441
1442 static bool obj_request_known_test(struct rbd_obj_request *obj_request)
1443 {
1444         smp_mb();
1445         return test_bit(OBJ_REQ_KNOWN, &obj_request->flags) != 0;
1446 }
1447
1448 static bool obj_request_exists_test(struct rbd_obj_request *obj_request)
1449 {
1450         smp_mb();
1451         return test_bit(OBJ_REQ_EXISTS, &obj_request->flags) != 0;
1452 }
1453
1454 static bool obj_request_overlaps_parent(struct rbd_obj_request *obj_request)
1455 {
1456         struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1457
1458         return obj_request->img_offset <
1459             round_up(rbd_dev->parent_overlap, rbd_obj_bytes(&rbd_dev->header));
1460 }
1461
1462 static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
1463 {
1464         dout("%s: obj %p (was %d)\n", __func__, obj_request,
1465                 atomic_read(&obj_request->kref.refcount));
1466         kref_get(&obj_request->kref);
1467 }
1468
1469 static void rbd_obj_request_destroy(struct kref *kref);
1470 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1471 {
1472         rbd_assert(obj_request != NULL);
1473         dout("%s: obj %p (was %d)\n", __func__, obj_request,
1474                 atomic_read(&obj_request->kref.refcount));
1475         kref_put(&obj_request->kref, rbd_obj_request_destroy);
1476 }
1477
1478 static void rbd_img_request_get(struct rbd_img_request *img_request)
1479 {
1480         dout("%s: img %p (was %d)\n", __func__, img_request,
1481              atomic_read(&img_request->kref.refcount));
1482         kref_get(&img_request->kref);
1483 }
1484
1485 static bool img_request_child_test(struct rbd_img_request *img_request);
1486 static void rbd_parent_request_destroy(struct kref *kref);
1487 static void rbd_img_request_destroy(struct kref *kref);
1488 static void rbd_img_request_put(struct rbd_img_request *img_request)
1489 {
1490         rbd_assert(img_request != NULL);
1491         dout("%s: img %p (was %d)\n", __func__, img_request,
1492                 atomic_read(&img_request->kref.refcount));
1493         if (img_request_child_test(img_request))
1494                 kref_put(&img_request->kref, rbd_parent_request_destroy);
1495         else
1496                 kref_put(&img_request->kref, rbd_img_request_destroy);
1497 }
1498
1499 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1500                                         struct rbd_obj_request *obj_request)
1501 {
1502         rbd_assert(obj_request->img_request == NULL);
1503
1504         /* Image request now owns object's original reference */
1505         obj_request->img_request = img_request;
1506         obj_request->which = img_request->obj_request_count;
1507         rbd_assert(!obj_request_img_data_test(obj_request));
1508         obj_request_img_data_set(obj_request);
1509         rbd_assert(obj_request->which != BAD_WHICH);
1510         img_request->obj_request_count++;
1511         list_add_tail(&obj_request->links, &img_request->obj_requests);
1512         dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1513                 obj_request->which);
1514 }
1515
1516 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1517                                         struct rbd_obj_request *obj_request)
1518 {
1519         rbd_assert(obj_request->which != BAD_WHICH);
1520
1521         dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1522                 obj_request->which);
1523         list_del(&obj_request->links);
1524         rbd_assert(img_request->obj_request_count > 0);
1525         img_request->obj_request_count--;
1526         rbd_assert(obj_request->which == img_request->obj_request_count);
1527         obj_request->which = BAD_WHICH;
1528         rbd_assert(obj_request_img_data_test(obj_request));
1529         rbd_assert(obj_request->img_request == img_request);
1530         obj_request->img_request = NULL;
1531         obj_request->callback = NULL;
1532         rbd_obj_request_put(obj_request);
1533 }
1534
1535 static bool obj_request_type_valid(enum obj_request_type type)
1536 {
1537         switch (type) {
1538         case OBJ_REQUEST_NODATA:
1539         case OBJ_REQUEST_BIO:
1540         case OBJ_REQUEST_PAGES:
1541                 return true;
1542         default:
1543                 return false;
1544         }
1545 }
1546
1547 static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
1548                                 struct rbd_obj_request *obj_request)
1549 {
1550         dout("%s %p\n", __func__, obj_request);
1551         return ceph_osdc_start_request(osdc, obj_request->osd_req, false);
1552 }
1553
1554 static void rbd_obj_request_end(struct rbd_obj_request *obj_request)
1555 {
1556         dout("%s %p\n", __func__, obj_request);
1557         ceph_osdc_cancel_request(obj_request->osd_req);
1558 }
1559
1560 /*
1561  * Wait for an object request to complete.  If interrupted, cancel the
1562  * underlying osd request.
1563  */
1564 static int rbd_obj_request_wait(struct rbd_obj_request *obj_request)
1565 {
1566         int ret;
1567
1568         dout("%s %p\n", __func__, obj_request);
1569
1570         ret = wait_for_completion_interruptible(&obj_request->completion);
1571         if (ret < 0) {
1572                 dout("%s %p interrupted\n", __func__, obj_request);
1573                 rbd_obj_request_end(obj_request);
1574                 return ret;
1575         }
1576
1577         dout("%s %p done\n", __func__, obj_request);
1578         return 0;
1579 }
1580
1581 static void rbd_img_request_complete(struct rbd_img_request *img_request)
1582 {
1583
1584         dout("%s: img %p\n", __func__, img_request);
1585
1586         /*
1587          * If no error occurred, compute the aggregate transfer
1588          * count for the image request.  We could instead use
1589          * atomic64_cmpxchg() to update it as each object request
1590          * completes; not clear which way is better off hand.
1591          */
1592         if (!img_request->result) {
1593                 struct rbd_obj_request *obj_request;
1594                 u64 xferred = 0;
1595
1596                 for_each_obj_request(img_request, obj_request)
1597                         xferred += obj_request->xferred;
1598                 img_request->xferred = xferred;
1599         }
1600
1601         if (img_request->callback)
1602                 img_request->callback(img_request);
1603         else
1604                 rbd_img_request_put(img_request);
1605 }
1606
1607 /*
1608  * The default/initial value for all image request flags is 0.  Each
1609  * is conditionally set to 1 at image request initialization time
1610  * and currently never change thereafter.
1611  */
1612 static void img_request_write_set(struct rbd_img_request *img_request)
1613 {
1614         set_bit(IMG_REQ_WRITE, &img_request->flags);
1615         smp_mb();
1616 }
1617
1618 static bool img_request_write_test(struct rbd_img_request *img_request)
1619 {
1620         smp_mb();
1621         return test_bit(IMG_REQ_WRITE, &img_request->flags) != 0;
1622 }
1623
1624 /*
1625  * Set the discard flag when the img_request is an discard request
1626  */
1627 static void img_request_discard_set(struct rbd_img_request *img_request)
1628 {
1629         set_bit(IMG_REQ_DISCARD, &img_request->flags);
1630         smp_mb();
1631 }
1632
1633 static bool img_request_discard_test(struct rbd_img_request *img_request)
1634 {
1635         smp_mb();
1636         return test_bit(IMG_REQ_DISCARD, &img_request->flags) != 0;
1637 }
1638
1639 static void img_request_child_set(struct rbd_img_request *img_request)
1640 {
1641         set_bit(IMG_REQ_CHILD, &img_request->flags);
1642         smp_mb();
1643 }
1644
1645 static void img_request_child_clear(struct rbd_img_request *img_request)
1646 {
1647         clear_bit(IMG_REQ_CHILD, &img_request->flags);
1648         smp_mb();
1649 }
1650
1651 static bool img_request_child_test(struct rbd_img_request *img_request)
1652 {
1653         smp_mb();
1654         return test_bit(IMG_REQ_CHILD, &img_request->flags) != 0;
1655 }
1656
1657 static void img_request_layered_set(struct rbd_img_request *img_request)
1658 {
1659         set_bit(IMG_REQ_LAYERED, &img_request->flags);
1660         smp_mb();
1661 }
1662
1663 static void img_request_layered_clear(struct rbd_img_request *img_request)
1664 {
1665         clear_bit(IMG_REQ_LAYERED, &img_request->flags);
1666         smp_mb();
1667 }
1668
1669 static bool img_request_layered_test(struct rbd_img_request *img_request)
1670 {
1671         smp_mb();
1672         return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1673 }
1674
1675 static enum obj_operation_type
1676 rbd_img_request_op_type(struct rbd_img_request *img_request)
1677 {
1678         if (img_request_write_test(img_request))
1679                 return OBJ_OP_WRITE;
1680         else if (img_request_discard_test(img_request))
1681                 return OBJ_OP_DISCARD;
1682         else
1683                 return OBJ_OP_READ;
1684 }
1685
1686 static void
1687 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
1688 {
1689         u64 xferred = obj_request->xferred;
1690         u64 length = obj_request->length;
1691
1692         dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1693                 obj_request, obj_request->img_request, obj_request->result,
1694                 xferred, length);
1695         /*
1696          * ENOENT means a hole in the image.  We zero-fill the entire
1697          * length of the request.  A short read also implies zero-fill
1698          * to the end of the request.  An error requires the whole
1699          * length of the request to be reported finished with an error
1700          * to the block layer.  In each case we update the xferred
1701          * count to indicate the whole request was satisfied.
1702          */
1703         rbd_assert(obj_request->type != OBJ_REQUEST_NODATA);
1704         if (obj_request->result == -ENOENT) {
1705                 if (obj_request->type == OBJ_REQUEST_BIO)
1706                         zero_bio_chain(obj_request->bio_list, 0);
1707                 else
1708                         zero_pages(obj_request->pages, 0, length);
1709                 obj_request->result = 0;
1710         } else if (xferred < length && !obj_request->result) {
1711                 if (obj_request->type == OBJ_REQUEST_BIO)
1712                         zero_bio_chain(obj_request->bio_list, xferred);
1713                 else
1714                         zero_pages(obj_request->pages, xferred, length);
1715         }
1716         obj_request->xferred = length;
1717         obj_request_done_set(obj_request);
1718 }
1719
1720 static void rbd_obj_request_complete(struct rbd_obj_request *obj_request)
1721 {
1722         dout("%s: obj %p cb %p\n", __func__, obj_request,
1723                 obj_request->callback);
1724         if (obj_request->callback)
1725                 obj_request->callback(obj_request);
1726         else
1727                 complete_all(&obj_request->completion);
1728 }
1729
1730 static void rbd_osd_trivial_callback(struct rbd_obj_request *obj_request)
1731 {
1732         dout("%s: obj %p\n", __func__, obj_request);
1733         obj_request_done_set(obj_request);
1734 }
1735
1736 static void rbd_osd_read_callback(struct rbd_obj_request *obj_request)
1737 {
1738         struct rbd_img_request *img_request = NULL;
1739         struct rbd_device *rbd_dev = NULL;
1740         bool layered = false;
1741
1742         if (obj_request_img_data_test(obj_request)) {
1743                 img_request = obj_request->img_request;
1744                 layered = img_request && img_request_layered_test(img_request);
1745                 rbd_dev = img_request->rbd_dev;
1746         }
1747
1748         dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1749                 obj_request, img_request, obj_request->result,
1750                 obj_request->xferred, obj_request->length);
1751         if (layered && obj_request->result == -ENOENT &&
1752                         obj_request->img_offset < rbd_dev->parent_overlap)
1753                 rbd_img_parent_read(obj_request);
1754         else if (img_request)
1755                 rbd_img_obj_request_read_callback(obj_request);
1756         else
1757                 obj_request_done_set(obj_request);
1758 }
1759
1760 static void rbd_osd_write_callback(struct rbd_obj_request *obj_request)
1761 {
1762         dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1763                 obj_request->result, obj_request->length);
1764         /*
1765          * There is no such thing as a successful short write.  Set
1766          * it to our originally-requested length.
1767          */
1768         obj_request->xferred = obj_request->length;
1769         obj_request_done_set(obj_request);
1770 }
1771
1772 static void rbd_osd_discard_callback(struct rbd_obj_request *obj_request)
1773 {
1774         dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1775                 obj_request->result, obj_request->length);
1776         /*
1777          * There is no such thing as a successful short discard.  Set
1778          * it to our originally-requested length.
1779          */
1780         obj_request->xferred = obj_request->length;
1781         /* discarding a non-existent object is not a problem */
1782         if (obj_request->result == -ENOENT)
1783                 obj_request->result = 0;
1784         obj_request_done_set(obj_request);
1785 }
1786
1787 /*
1788  * For a simple stat call there's nothing to do.  We'll do more if
1789  * this is part of a write sequence for a layered image.
1790  */
1791 static void rbd_osd_stat_callback(struct rbd_obj_request *obj_request)
1792 {
1793         dout("%s: obj %p\n", __func__, obj_request);
1794         obj_request_done_set(obj_request);
1795 }
1796
1797 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
1798                                 struct ceph_msg *msg)
1799 {
1800         struct rbd_obj_request *obj_request = osd_req->r_priv;
1801         u16 opcode;
1802
1803         dout("%s: osd_req %p msg %p\n", __func__, osd_req, msg);
1804         rbd_assert(osd_req == obj_request->osd_req);
1805         if (obj_request_img_data_test(obj_request)) {
1806                 rbd_assert(obj_request->img_request);
1807                 rbd_assert(obj_request->which != BAD_WHICH);
1808         } else {
1809                 rbd_assert(obj_request->which == BAD_WHICH);
1810         }
1811
1812         if (osd_req->r_result < 0)
1813                 obj_request->result = osd_req->r_result;
1814
1815         rbd_assert(osd_req->r_num_ops <= CEPH_OSD_MAX_OP);
1816
1817         /*
1818          * We support a 64-bit length, but ultimately it has to be
1819          * passed to blk_end_request(), which takes an unsigned int.
1820          */
1821         obj_request->xferred = osd_req->r_reply_op_len[0];
1822         rbd_assert(obj_request->xferred < (u64)UINT_MAX);
1823
1824         opcode = osd_req->r_ops[0].op;
1825         switch (opcode) {
1826         case CEPH_OSD_OP_READ:
1827                 rbd_osd_read_callback(obj_request);
1828                 break;
1829         case CEPH_OSD_OP_SETALLOCHINT:
1830                 rbd_assert(osd_req->r_ops[1].op == CEPH_OSD_OP_WRITE);
1831                 /* fall through */
1832         case CEPH_OSD_OP_WRITE:
1833                 rbd_osd_write_callback(obj_request);
1834                 break;
1835         case CEPH_OSD_OP_STAT:
1836                 rbd_osd_stat_callback(obj_request);
1837                 break;
1838         case CEPH_OSD_OP_DELETE:
1839         case CEPH_OSD_OP_TRUNCATE:
1840         case CEPH_OSD_OP_ZERO:
1841                 rbd_osd_discard_callback(obj_request);
1842                 break;
1843         case CEPH_OSD_OP_CALL:
1844         case CEPH_OSD_OP_NOTIFY_ACK:
1845         case CEPH_OSD_OP_WATCH:
1846                 rbd_osd_trivial_callback(obj_request);
1847                 break;
1848         default:
1849                 rbd_warn(NULL, "%s: unsupported op %hu",
1850                         obj_request->object_name, (unsigned short) opcode);
1851                 break;
1852         }
1853
1854         if (obj_request_done_test(obj_request))
1855                 rbd_obj_request_complete(obj_request);
1856 }
1857
1858 static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request)
1859 {
1860         struct rbd_img_request *img_request = obj_request->img_request;
1861         struct ceph_osd_request *osd_req = obj_request->osd_req;
1862         u64 snap_id;
1863
1864         rbd_assert(osd_req != NULL);
1865
1866         snap_id = img_request ? img_request->snap_id : CEPH_NOSNAP;
1867         ceph_osdc_build_request(osd_req, obj_request->offset,
1868                         NULL, snap_id, NULL);
1869 }
1870
1871 static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request)
1872 {
1873         struct rbd_img_request *img_request = obj_request->img_request;
1874         struct ceph_osd_request *osd_req = obj_request->osd_req;
1875         struct ceph_snap_context *snapc;
1876         struct timespec mtime = CURRENT_TIME;
1877
1878         rbd_assert(osd_req != NULL);
1879
1880         snapc = img_request ? img_request->snapc : NULL;
1881         ceph_osdc_build_request(osd_req, obj_request->offset,
1882                         snapc, CEPH_NOSNAP, &mtime);
1883 }
1884
1885 /*
1886  * Create an osd request.  A read request has one osd op (read).
1887  * A write request has either one (watch) or two (hint+write) osd ops.
1888  * (All rbd data writes are prefixed with an allocation hint op, but
1889  * technically osd watch is a write request, hence this distinction.)
1890  */
1891 static struct ceph_osd_request *rbd_osd_req_create(
1892                                         struct rbd_device *rbd_dev,
1893                                         enum obj_operation_type op_type,
1894                                         unsigned int num_ops,
1895                                         struct rbd_obj_request *obj_request)
1896 {
1897         struct ceph_snap_context *snapc = NULL;
1898         struct ceph_osd_client *osdc;
1899         struct ceph_osd_request *osd_req;
1900
1901         if (obj_request_img_data_test(obj_request) &&
1902                 (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_WRITE)) {
1903                 struct rbd_img_request *img_request = obj_request->img_request;
1904                 if (op_type == OBJ_OP_WRITE) {
1905                         rbd_assert(img_request_write_test(img_request));
1906                 } else {
1907                         rbd_assert(img_request_discard_test(img_request));
1908                 }
1909                 snapc = img_request->snapc;
1910         }
1911
1912         rbd_assert(num_ops == 1 || ((op_type == OBJ_OP_WRITE) && num_ops == 2));
1913
1914         /* Allocate and initialize the request, for the num_ops ops */
1915
1916         osdc = &rbd_dev->rbd_client->client->osdc;
1917         osd_req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false,
1918                                           GFP_ATOMIC);
1919         if (!osd_req)
1920                 return NULL;    /* ENOMEM */
1921
1922         if (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD)
1923                 osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
1924         else
1925                 osd_req->r_flags = CEPH_OSD_FLAG_READ;
1926
1927         osd_req->r_callback = rbd_osd_req_callback;
1928         osd_req->r_priv = obj_request;
1929
1930         osd_req->r_base_oloc.pool = ceph_file_layout_pg_pool(rbd_dev->layout);
1931         ceph_oid_set_name(&osd_req->r_base_oid, obj_request->object_name);
1932
1933         return osd_req;
1934 }
1935
1936 /*
1937  * Create a copyup osd request based on the information in the object
1938  * request supplied.  A copyup request has two or three osd ops, a
1939  * copyup method call, potentially a hint op, and a write or truncate
1940  * or zero op.
1941  */
1942 static struct ceph_osd_request *
1943 rbd_osd_req_create_copyup(struct rbd_obj_request *obj_request)
1944 {
1945         struct rbd_img_request *img_request;
1946         struct ceph_snap_context *snapc;
1947         struct rbd_device *rbd_dev;
1948         struct ceph_osd_client *osdc;
1949         struct ceph_osd_request *osd_req;
1950         int num_osd_ops = 3;
1951
1952         rbd_assert(obj_request_img_data_test(obj_request));
1953         img_request = obj_request->img_request;
1954         rbd_assert(img_request);
1955         rbd_assert(img_request_write_test(img_request) ||
1956                         img_request_discard_test(img_request));
1957
1958         if (img_request_discard_test(img_request))
1959                 num_osd_ops = 2;
1960
1961         /* Allocate and initialize the request, for all the ops */
1962
1963         snapc = img_request->snapc;
1964         rbd_dev = img_request->rbd_dev;
1965         osdc = &rbd_dev->rbd_client->client->osdc;
1966         osd_req = ceph_osdc_alloc_request(osdc, snapc, num_osd_ops,
1967                                                 false, GFP_ATOMIC);
1968         if (!osd_req)
1969                 return NULL;    /* ENOMEM */
1970
1971         osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
1972         osd_req->r_callback = rbd_osd_req_callback;
1973         osd_req->r_priv = obj_request;
1974
1975         osd_req->r_base_oloc.pool = ceph_file_layout_pg_pool(rbd_dev->layout);
1976         ceph_oid_set_name(&osd_req->r_base_oid, obj_request->object_name);
1977
1978         return osd_req;
1979 }
1980
1981
1982 static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req)
1983 {
1984         ceph_osdc_put_request(osd_req);
1985 }
1986
1987 /* object_name is assumed to be a non-null pointer and NUL-terminated */
1988
1989 static struct rbd_obj_request *rbd_obj_request_create(const char *object_name,
1990                                                 u64 offset, u64 length,
1991                                                 enum obj_request_type type)
1992 {
1993         struct rbd_obj_request *obj_request;
1994         size_t size;
1995         char *name;
1996
1997         rbd_assert(obj_request_type_valid(type));
1998
1999         size = strlen(object_name) + 1;
2000         name = kmalloc(size, GFP_KERNEL);
2001         if (!name)
2002                 return NULL;
2003
2004         obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_KERNEL);
2005         if (!obj_request) {
2006                 kfree(name);
2007                 return NULL;
2008         }
2009
2010         obj_request->object_name = memcpy(name, object_name, size);
2011         obj_request->offset = offset;
2012         obj_request->length = length;
2013         obj_request->flags = 0;
2014         obj_request->which = BAD_WHICH;
2015         obj_request->type = type;
2016         INIT_LIST_HEAD(&obj_request->links);
2017         init_completion(&obj_request->completion);
2018         kref_init(&obj_request->kref);
2019
2020         dout("%s: \"%s\" %llu/%llu %d -> obj %p\n", __func__, object_name,
2021                 offset, length, (int)type, obj_request);
2022
2023         return obj_request;
2024 }
2025
2026 static void rbd_obj_request_destroy(struct kref *kref)
2027 {
2028         struct rbd_obj_request *obj_request;
2029
2030         obj_request = container_of(kref, struct rbd_obj_request, kref);
2031
2032         dout("%s: obj %p\n", __func__, obj_request);
2033
2034         rbd_assert(obj_request->img_request == NULL);
2035         rbd_assert(obj_request->which == BAD_WHICH);
2036
2037         if (obj_request->osd_req)
2038                 rbd_osd_req_destroy(obj_request->osd_req);
2039
2040         rbd_assert(obj_request_type_valid(obj_request->type));
2041         switch (obj_request->type) {
2042         case OBJ_REQUEST_NODATA:
2043                 break;          /* Nothing to do */
2044         case OBJ_REQUEST_BIO:
2045                 if (obj_request->bio_list)
2046                         bio_chain_put(obj_request->bio_list);
2047                 break;
2048         case OBJ_REQUEST_PAGES:
2049                 if (obj_request->pages)
2050                         ceph_release_page_vector(obj_request->pages,
2051                                                 obj_request->page_count);
2052                 break;
2053         }
2054
2055         kfree(obj_request->object_name);
2056         obj_request->object_name = NULL;
2057         kmem_cache_free(rbd_obj_request_cache, obj_request);
2058 }
2059
2060 /* It's OK to call this for a device with no parent */
2061
2062 static void rbd_spec_put(struct rbd_spec *spec);
2063 static void rbd_dev_unparent(struct rbd_device *rbd_dev)
2064 {
2065         rbd_dev_remove_parent(rbd_dev);
2066         rbd_spec_put(rbd_dev->parent_spec);
2067         rbd_dev->parent_spec = NULL;
2068         rbd_dev->parent_overlap = 0;
2069 }
2070
2071 /*
2072  * Parent image reference counting is used to determine when an
2073  * image's parent fields can be safely torn down--after there are no
2074  * more in-flight requests to the parent image.  When the last
2075  * reference is dropped, cleaning them up is safe.
2076  */
2077 static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
2078 {
2079         int counter;
2080
2081         if (!rbd_dev->parent_spec)
2082                 return;
2083
2084         counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
2085         if (counter > 0)
2086                 return;
2087
2088         /* Last reference; clean up parent data structures */
2089
2090         if (!counter)
2091                 rbd_dev_unparent(rbd_dev);
2092         else
2093                 rbd_warn(rbd_dev, "parent reference underflow");
2094 }
2095
2096 /*
2097  * If an image has a non-zero parent overlap, get a reference to its
2098  * parent.
2099  *
2100  * We must get the reference before checking for the overlap to
2101  * coordinate properly with zeroing the parent overlap in
2102  * rbd_dev_v2_parent_info() when an image gets flattened.  We
2103  * drop it again if there is no overlap.
2104  *
2105  * Returns true if the rbd device has a parent with a non-zero
2106  * overlap and a reference for it was successfully taken, or
2107  * false otherwise.
2108  */
2109 static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
2110 {
2111         int counter;
2112
2113         if (!rbd_dev->parent_spec)
2114                 return false;
2115
2116         counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
2117         if (counter > 0 && rbd_dev->parent_overlap)
2118                 return true;
2119
2120         /* Image was flattened, but parent is not yet torn down */
2121
2122         if (counter < 0)
2123                 rbd_warn(rbd_dev, "parent reference overflow");
2124
2125         return false;
2126 }
2127
2128 /*
2129  * Caller is responsible for filling in the list of object requests
2130  * that comprises the image request, and the Linux request pointer
2131  * (if there is one).
2132  */
2133 static struct rbd_img_request *rbd_img_request_create(
2134                                         struct rbd_device *rbd_dev,
2135                                         u64 offset, u64 length,
2136                                         enum obj_operation_type op_type,
2137                                         struct ceph_snap_context *snapc)
2138 {
2139         struct rbd_img_request *img_request;
2140
2141         img_request = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
2142         if (!img_request)
2143                 return NULL;
2144
2145         img_request->rq = NULL;
2146         img_request->rbd_dev = rbd_dev;
2147         img_request->offset = offset;
2148         img_request->length = length;
2149         img_request->flags = 0;
2150         if (op_type == OBJ_OP_DISCARD) {
2151                 img_request_discard_set(img_request);
2152                 img_request->snapc = snapc;
2153         } else if (op_type == OBJ_OP_WRITE) {
2154                 img_request_write_set(img_request);
2155                 img_request->snapc = snapc;
2156         } else {
2157                 img_request->snap_id = rbd_dev->spec->snap_id;
2158         }
2159         if (rbd_dev_parent_get(rbd_dev))
2160                 img_request_layered_set(img_request);
2161         spin_lock_init(&img_request->completion_lock);
2162         img_request->next_completion = 0;
2163         img_request->callback = NULL;
2164         img_request->result = 0;
2165         img_request->obj_request_count = 0;
2166         INIT_LIST_HEAD(&img_request->obj_requests);
2167         kref_init(&img_request->kref);
2168
2169         dout("%s: rbd_dev %p %s %llu/%llu -> img %p\n", __func__, rbd_dev,
2170                 obj_op_name(op_type), offset, length, img_request);
2171
2172         return img_request;
2173 }
2174
2175 static void rbd_img_request_destroy(struct kref *kref)
2176 {
2177         struct rbd_img_request *img_request;
2178         struct rbd_obj_request *obj_request;
2179         struct rbd_obj_request *next_obj_request;
2180
2181         img_request = container_of(kref, struct rbd_img_request, kref);
2182
2183         dout("%s: img %p\n", __func__, img_request);
2184
2185         for_each_obj_request_safe(img_request, obj_request, next_obj_request)
2186                 rbd_img_obj_request_del(img_request, obj_request);
2187         rbd_assert(img_request->obj_request_count == 0);
2188
2189         if (img_request_layered_test(img_request)) {
2190                 img_request_layered_clear(img_request);
2191                 rbd_dev_parent_put(img_request->rbd_dev);
2192         }
2193
2194         if (img_request_write_test(img_request) ||
2195                 img_request_discard_test(img_request))
2196                 ceph_put_snap_context(img_request->snapc);
2197
2198         kmem_cache_free(rbd_img_request_cache, img_request);
2199 }
2200
2201 static struct rbd_img_request *rbd_parent_request_create(
2202                                         struct rbd_obj_request *obj_request,
2203                                         u64 img_offset, u64 length)
2204 {
2205         struct rbd_img_request *parent_request;
2206         struct rbd_device *rbd_dev;
2207
2208         rbd_assert(obj_request->img_request);
2209         rbd_dev = obj_request->img_request->rbd_dev;
2210
2211         parent_request = rbd_img_request_create(rbd_dev->parent, img_offset,
2212                                                 length, OBJ_OP_READ, NULL);
2213         if (!parent_request)
2214                 return NULL;
2215
2216         img_request_child_set(parent_request);
2217         rbd_obj_request_get(obj_request);
2218         parent_request->obj_request = obj_request;
2219
2220         return parent_request;
2221 }
2222
2223 static void rbd_parent_request_destroy(struct kref *kref)
2224 {
2225         struct rbd_img_request *parent_request;
2226         struct rbd_obj_request *orig_request;
2227
2228         parent_request = container_of(kref, struct rbd_img_request, kref);
2229         orig_request = parent_request->obj_request;
2230
2231         parent_request->obj_request = NULL;
2232         rbd_obj_request_put(orig_request);
2233         img_request_child_clear(parent_request);
2234
2235         rbd_img_request_destroy(kref);
2236 }
2237
2238 static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request)
2239 {
2240         struct rbd_img_request *img_request;
2241         unsigned int xferred;
2242         int result;
2243         bool more;
2244
2245         rbd_assert(obj_request_img_data_test(obj_request));
2246         img_request = obj_request->img_request;
2247
2248         rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
2249         xferred = (unsigned int)obj_request->xferred;
2250         result = obj_request->result;
2251         if (result) {
2252                 struct rbd_device *rbd_dev = img_request->rbd_dev;
2253                 enum obj_operation_type op_type;
2254
2255                 if (img_request_discard_test(img_request))
2256                         op_type = OBJ_OP_DISCARD;
2257                 else if (img_request_write_test(img_request))
2258                         op_type = OBJ_OP_WRITE;
2259                 else
2260                         op_type = OBJ_OP_READ;
2261
2262                 rbd_warn(rbd_dev, "%s %llx at %llx (%llx)",
2263                         obj_op_name(op_type), obj_request->length,
2264                         obj_request->img_offset, obj_request->offset);
2265                 rbd_warn(rbd_dev, "  result %d xferred %x",
2266                         result, xferred);
2267                 if (!img_request->result)
2268                         img_request->result = result;
2269         }
2270
2271         /* Image object requests don't own their page array */
2272
2273         if (obj_request->type == OBJ_REQUEST_PAGES) {
2274                 obj_request->pages = NULL;
2275                 obj_request->page_count = 0;
2276         }
2277
2278         if (img_request_child_test(img_request)) {
2279                 rbd_assert(img_request->obj_request != NULL);
2280                 more = obj_request->which < img_request->obj_request_count - 1;
2281         } else {
2282                 rbd_assert(img_request->rq != NULL);
2283                 more = blk_end_request(img_request->rq, result, xferred);
2284         }
2285
2286         return more;
2287 }
2288
2289 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request)
2290 {
2291         struct rbd_img_request *img_request;
2292         u32 which = obj_request->which;
2293         bool more = true;
2294
2295         rbd_assert(obj_request_img_data_test(obj_request));
2296         img_request = obj_request->img_request;
2297
2298         dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
2299         rbd_assert(img_request != NULL);
2300         rbd_assert(img_request->obj_request_count > 0);
2301         rbd_assert(which != BAD_WHICH);
2302         rbd_assert(which < img_request->obj_request_count);
2303
2304         spin_lock_irq(&img_request->completion_lock);
2305         if (which != img_request->next_completion)
2306                 goto out;
2307
2308         for_each_obj_request_from(img_request, obj_request) {
2309                 rbd_assert(more);
2310                 rbd_assert(which < img_request->obj_request_count);
2311
2312                 if (!obj_request_done_test(obj_request))
2313                         break;
2314                 more = rbd_img_obj_end_request(obj_request);
2315                 which++;
2316         }
2317
2318         rbd_assert(more ^ (which == img_request->obj_request_count));
2319         img_request->next_completion = which;
2320 out:
2321         spin_unlock_irq(&img_request->completion_lock);
2322         rbd_img_request_put(img_request);
2323
2324         if (!more)
2325                 rbd_img_request_complete(img_request);
2326 }
2327
2328 /*
2329  * Add individual osd ops to the given ceph_osd_request and prepare
2330  * them for submission. num_ops is the current number of
2331  * osd operations already to the object request.
2332  */
2333 static void rbd_img_obj_request_fill(struct rbd_obj_request *obj_request,
2334                                 struct ceph_osd_request *osd_request,
2335                                 enum obj_operation_type op_type,
2336                                 unsigned int num_ops)
2337 {
2338         struct rbd_img_request *img_request = obj_request->img_request;
2339         struct rbd_device *rbd_dev = img_request->rbd_dev;
2340         u64 object_size = rbd_obj_bytes(&rbd_dev->header);
2341         u64 offset = obj_request->offset;
2342         u64 length = obj_request->length;
2343         u64 img_end;
2344         u16 opcode;
2345
2346         if (op_type == OBJ_OP_DISCARD) {
2347                 if (!offset && length == object_size &&
2348                     (!img_request_layered_test(img_request) ||
2349                      !obj_request_overlaps_parent(obj_request))) {
2350                         opcode = CEPH_OSD_OP_DELETE;
2351                 } else if ((offset + length == object_size)) {
2352                         opcode = CEPH_OSD_OP_TRUNCATE;
2353                 } else {
2354                         down_read(&rbd_dev->header_rwsem);
2355                         img_end = rbd_dev->header.image_size;
2356                         up_read(&rbd_dev->header_rwsem);
2357
2358                         if (obj_request->img_offset + length == img_end)
2359                                 opcode = CEPH_OSD_OP_TRUNCATE;
2360                         else
2361                                 opcode = CEPH_OSD_OP_ZERO;
2362                 }
2363         } else if (op_type == OBJ_OP_WRITE) {
2364                 opcode = CEPH_OSD_OP_WRITE;
2365                 osd_req_op_alloc_hint_init(osd_request, num_ops,
2366                                         object_size, object_size);
2367                 num_ops++;
2368         } else {
2369                 opcode = CEPH_OSD_OP_READ;
2370         }
2371
2372         osd_req_op_extent_init(osd_request, num_ops, opcode, offset, length,
2373                                 0, 0);
2374         if (obj_request->type == OBJ_REQUEST_BIO)
2375                 osd_req_op_extent_osd_data_bio(osd_request, num_ops,
2376                                         obj_request->bio_list, length);
2377         else if (obj_request->type == OBJ_REQUEST_PAGES)
2378                 osd_req_op_extent_osd_data_pages(osd_request, num_ops,
2379                                         obj_request->pages, length,
2380                                         offset & ~PAGE_MASK, false, false);
2381
2382         /* Discards are also writes */
2383         if (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD)
2384                 rbd_osd_req_format_write(obj_request);
2385         else
2386                 rbd_osd_req_format_read(obj_request);
2387 }
2388
2389 /*
2390  * Split up an image request into one or more object requests, each
2391  * to a different object.  The "type" parameter indicates whether
2392  * "data_desc" is the pointer to the head of a list of bio
2393  * structures, or the base of a page array.  In either case this
2394  * function assumes data_desc describes memory sufficient to hold
2395  * all data described by the image request.
2396  */
2397 static int rbd_img_request_fill(struct rbd_img_request *img_request,
2398                                         enum obj_request_type type,
2399                                         void *data_desc)
2400 {
2401         struct rbd_device *rbd_dev = img_request->rbd_dev;
2402         struct rbd_obj_request *obj_request = NULL;
2403         struct rbd_obj_request *next_obj_request;
2404         struct bio *bio_list = NULL;
2405         unsigned int bio_offset = 0;
2406         struct page **pages = NULL;
2407         enum obj_operation_type op_type;
2408         u64 img_offset;
2409         u64 resid;
2410
2411         dout("%s: img %p type %d data_desc %p\n", __func__, img_request,
2412                 (int)type, data_desc);
2413
2414         img_offset = img_request->offset;
2415         resid = img_request->length;
2416         rbd_assert(resid > 0);
2417         op_type = rbd_img_request_op_type(img_request);
2418
2419         if (type == OBJ_REQUEST_BIO) {
2420                 bio_list = data_desc;
2421                 rbd_assert(img_offset ==
2422                            bio_list->bi_iter.bi_sector << SECTOR_SHIFT);
2423         } else if (type == OBJ_REQUEST_PAGES) {
2424                 pages = data_desc;
2425         }
2426
2427         while (resid) {
2428                 struct ceph_osd_request *osd_req;
2429                 const char *object_name;
2430                 u64 offset;
2431                 u64 length;
2432
2433                 object_name = rbd_segment_name(rbd_dev, img_offset);
2434                 if (!object_name)
2435                         goto out_unwind;
2436                 offset = rbd_segment_offset(rbd_dev, img_offset);
2437                 length = rbd_segment_length(rbd_dev, img_offset, resid);
2438                 obj_request = rbd_obj_request_create(object_name,
2439                                                 offset, length, type);
2440                 /* object request has its own copy of the object name */
2441                 rbd_segment_name_free(object_name);
2442                 if (!obj_request)
2443                         goto out_unwind;
2444
2445                 /*
2446                  * set obj_request->img_request before creating the
2447                  * osd_request so that it gets the right snapc
2448                  */
2449                 rbd_img_obj_request_add(img_request, obj_request);
2450
2451                 if (type == OBJ_REQUEST_BIO) {
2452                         unsigned int clone_size;
2453
2454                         rbd_assert(length <= (u64)UINT_MAX);
2455                         clone_size = (unsigned int)length;
2456                         obj_request->bio_list =
2457                                         bio_chain_clone_range(&bio_list,
2458                                                                 &bio_offset,
2459                                                                 clone_size,
2460                                                                 GFP_ATOMIC);
2461                         if (!obj_request->bio_list)
2462                                 goto out_unwind;
2463                 } else if (type == OBJ_REQUEST_PAGES) {
2464                         unsigned int page_count;
2465
2466                         obj_request->pages = pages;
2467                         page_count = (u32)calc_pages_for(offset, length);
2468                         obj_request->page_count = page_count;
2469                         if ((offset + length) & ~PAGE_MASK)
2470                                 page_count--;   /* more on last page */
2471                         pages += page_count;
2472                 }
2473
2474                 osd_req = rbd_osd_req_create(rbd_dev, op_type,
2475                                         (op_type == OBJ_OP_WRITE) ? 2 : 1,
2476                                         obj_request);
2477                 if (!osd_req)
2478                         goto out_unwind;
2479
2480                 obj_request->osd_req = osd_req;
2481                 obj_request->callback = rbd_img_obj_callback;
2482                 obj_request->img_offset = img_offset;
2483
2484                 rbd_img_obj_request_fill(obj_request, osd_req, op_type, 0);
2485
2486                 rbd_img_request_get(img_request);
2487
2488                 img_offset += length;
2489                 resid -= length;
2490         }
2491
2492         return 0;
2493
2494 out_unwind:
2495         for_each_obj_request_safe(img_request, obj_request, next_obj_request)
2496                 rbd_img_obj_request_del(img_request, obj_request);
2497
2498         return -ENOMEM;
2499 }
2500
2501 static void
2502 rbd_img_obj_copyup_callback(struct rbd_obj_request *obj_request)
2503 {
2504         struct rbd_img_request *img_request;
2505         struct rbd_device *rbd_dev;
2506         struct page **pages;
2507         u32 page_count;
2508
2509         rbd_assert(obj_request->type == OBJ_REQUEST_BIO ||
2510                 obj_request->type == OBJ_REQUEST_NODATA);
2511         rbd_assert(obj_request_img_data_test(obj_request));
2512         img_request = obj_request->img_request;
2513         rbd_assert(img_request);
2514
2515         rbd_dev = img_request->rbd_dev;
2516         rbd_assert(rbd_dev);
2517
2518         pages = obj_request->copyup_pages;
2519         rbd_assert(pages != NULL);
2520         obj_request->copyup_pages = NULL;
2521         page_count = obj_request->copyup_page_count;
2522         rbd_assert(page_count);
2523         obj_request->copyup_page_count = 0;
2524         ceph_release_page_vector(pages, page_count);
2525
2526         /*
2527          * We want the transfer count to reflect the size of the
2528          * original write request.  There is no such thing as a
2529          * successful short write, so if the request was successful
2530          * we can just set it to the originally-requested length.
2531          */
2532         if (!obj_request->result)
2533                 obj_request->xferred = obj_request->length;
2534
2535         /* Finish up with the normal image object callback */
2536
2537         rbd_img_obj_callback(obj_request);
2538 }
2539
2540 static void
2541 rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request)
2542 {
2543         struct rbd_obj_request *orig_request;
2544         struct ceph_osd_request *osd_req;
2545         struct ceph_osd_client *osdc;
2546         struct rbd_device *rbd_dev;
2547         struct page **pages;
2548         enum obj_operation_type op_type;
2549         u32 page_count;
2550         int img_result;
2551         u64 parent_length;
2552
2553         rbd_assert(img_request_child_test(img_request));
2554
2555         /* First get what we need from the image request */
2556
2557         pages = img_request->copyup_pages;
2558         rbd_assert(pages != NULL);
2559         img_request->copyup_pages = NULL;
2560         page_count = img_request->copyup_page_count;
2561         rbd_assert(page_count);
2562         img_request->copyup_page_count = 0;
2563
2564         orig_request = img_request->obj_request;
2565         rbd_assert(orig_request != NULL);
2566         rbd_assert(obj_request_type_valid(orig_request->type));
2567         img_result = img_request->result;
2568         parent_length = img_request->length;
2569         rbd_assert(parent_length == img_request->xferred);
2570         rbd_img_request_put(img_request);
2571
2572         rbd_assert(orig_request->img_request);
2573         rbd_dev = orig_request->img_request->rbd_dev;
2574         rbd_assert(rbd_dev);
2575
2576         /*
2577          * If the overlap has become 0 (most likely because the
2578          * image has been flattened) we need to free the pages
2579          * and re-submit the original write request.
2580          */
2581         if (!rbd_dev->parent_overlap) {
2582                 struct ceph_osd_client *osdc;
2583
2584                 ceph_release_page_vector(pages, page_count);
2585                 osdc = &rbd_dev->rbd_client->client->osdc;
2586                 img_result = rbd_obj_request_submit(osdc, orig_request);
2587                 if (!img_result)
2588                         return;
2589         }
2590
2591         if (img_result)
2592                 goto out_err;
2593
2594         /*
2595          * The original osd request is of no use to use any more.
2596          * We need a new one that can hold the three ops in a copyup
2597          * request.  Allocate the new copyup osd request for the
2598          * original request, and release the old one.
2599          */
2600         img_result = -ENOMEM;
2601         osd_req = rbd_osd_req_create_copyup(orig_request);
2602         if (!osd_req)
2603                 goto out_err;
2604         rbd_osd_req_destroy(orig_request->osd_req);
2605         orig_request->osd_req = osd_req;
2606         orig_request->copyup_pages = pages;
2607         orig_request->copyup_page_count = page_count;
2608
2609         /* Initialize the copyup op */
2610
2611         osd_req_op_cls_init(osd_req, 0, CEPH_OSD_OP_CALL, "rbd", "copyup");
2612         osd_req_op_cls_request_data_pages(osd_req, 0, pages, parent_length, 0,
2613                                                 false, false);
2614
2615         /* Add the other op(s) */
2616
2617         op_type = rbd_img_request_op_type(orig_request->img_request);
2618         rbd_img_obj_request_fill(orig_request, osd_req, op_type, 1);
2619
2620         /* All set, send it off. */
2621
2622         orig_request->callback = rbd_img_obj_copyup_callback;
2623         osdc = &rbd_dev->rbd_client->client->osdc;
2624         img_result = rbd_obj_request_submit(osdc, orig_request);
2625         if (!img_result)
2626                 return;
2627 out_err:
2628         /* Record the error code and complete the request */
2629
2630         orig_request->result = img_result;
2631         orig_request->xferred = 0;
2632         obj_request_done_set(orig_request);
2633         rbd_obj_request_complete(orig_request);
2634 }
2635
2636 /*
2637  * Read from the parent image the range of data that covers the
2638  * entire target of the given object request.  This is used for
2639  * satisfying a layered image write request when the target of an
2640  * object request from the image request does not exist.
2641  *
2642  * A page array big enough to hold the returned data is allocated
2643  * and supplied to rbd_img_request_fill() as the "data descriptor."
2644  * When the read completes, this page array will be transferred to
2645  * the original object request for the copyup operation.
2646  *
2647  * If an error occurs, record it as the result of the original
2648  * object request and mark it done so it gets completed.
2649  */
2650 static int rbd_img_obj_parent_read_full(struct rbd_obj_request *obj_request)
2651 {
2652         struct rbd_img_request *img_request = NULL;
2653         struct rbd_img_request *parent_request = NULL;
2654         struct rbd_device *rbd_dev;
2655         u64 img_offset;
2656         u64 length;
2657         struct page **pages = NULL;
2658         u32 page_count;
2659         int result;
2660
2661         rbd_assert(obj_request_img_data_test(obj_request));
2662         rbd_assert(obj_request_type_valid(obj_request->type));
2663
2664         img_request = obj_request->img_request;
2665         rbd_assert(img_request != NULL);
2666         rbd_dev = img_request->rbd_dev;
2667         rbd_assert(rbd_dev->parent != NULL);
2668
2669         /*
2670          * Determine the byte range covered by the object in the
2671          * child image to which the original request was to be sent.
2672          */
2673         img_offset = obj_request->img_offset - obj_request->offset;
2674         length = (u64)1 << rbd_dev->header.obj_order;
2675
2676         /*
2677          * There is no defined parent data beyond the parent
2678          * overlap, so limit what we read at that boundary if
2679          * necessary.
2680          */
2681         if (img_offset + length > rbd_dev->parent_overlap) {
2682                 rbd_assert(img_offset < rbd_dev->parent_overlap);
2683                 length = rbd_dev->parent_overlap - img_offset;
2684         }
2685
2686         /*
2687          * Allocate a page array big enough to receive the data read
2688          * from the parent.
2689          */
2690         page_count = (u32)calc_pages_for(0, length);
2691         pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2692         if (IS_ERR(pages)) {
2693                 result = PTR_ERR(pages);
2694                 pages = NULL;
2695                 goto out_err;
2696         }
2697
2698         result = -ENOMEM;
2699         parent_request = rbd_parent_request_create(obj_request,
2700                                                 img_offset, length);
2701         if (!parent_request)
2702                 goto out_err;
2703
2704         result = rbd_img_request_fill(parent_request, OBJ_REQUEST_PAGES, pages);
2705         if (result)
2706                 goto out_err;
2707         parent_request->copyup_pages = pages;
2708         parent_request->copyup_page_count = page_count;
2709
2710         parent_request->callback = rbd_img_obj_parent_read_full_callback;
2711         result = rbd_img_request_submit(parent_request);
2712         if (!result)
2713                 return 0;
2714
2715         parent_request->copyup_pages = NULL;
2716         parent_request->copyup_page_count = 0;
2717         parent_request->obj_request = NULL;
2718         rbd_obj_request_put(obj_request);
2719 out_err:
2720         if (pages)
2721                 ceph_release_page_vector(pages, page_count);
2722         if (parent_request)
2723                 rbd_img_request_put(parent_request);
2724         obj_request->result = result;
2725         obj_request->xferred = 0;
2726         obj_request_done_set(obj_request);
2727
2728         return result;
2729 }
2730
2731 static void rbd_img_obj_exists_callback(struct rbd_obj_request *obj_request)
2732 {
2733         struct rbd_obj_request *orig_request;
2734         struct rbd_device *rbd_dev;
2735         int result;
2736
2737         rbd_assert(!obj_request_img_data_test(obj_request));
2738
2739         /*
2740          * All we need from the object request is the original
2741          * request and the result of the STAT op.  Grab those, then
2742          * we're done with the request.
2743          */
2744         orig_request = obj_request->obj_request;
2745         obj_request->obj_request = NULL;
2746         rbd_obj_request_put(orig_request);
2747         rbd_assert(orig_request);
2748         rbd_assert(orig_request->img_request);
2749
2750         result = obj_request->result;
2751         obj_request->result = 0;
2752
2753         dout("%s: obj %p for obj %p result %d %llu/%llu\n", __func__,
2754                 obj_request, orig_request, result,
2755                 obj_request->xferred, obj_request->length);
2756         rbd_obj_request_put(obj_request);
2757
2758         /*
2759          * If the overlap has become 0 (most likely because the
2760          * image has been flattened) we need to free the pages
2761          * and re-submit the original write request.
2762          */
2763         rbd_dev = orig_request->img_request->rbd_dev;
2764         if (!rbd_dev->parent_overlap) {
2765                 struct ceph_osd_client *osdc;
2766
2767                 osdc = &rbd_dev->rbd_client->client->osdc;
2768                 result = rbd_obj_request_submit(osdc, orig_request);
2769                 if (!result)
2770                         return;
2771         }
2772
2773         /*
2774          * Our only purpose here is to determine whether the object
2775          * exists, and we don't want to treat the non-existence as
2776          * an error.  If something else comes back, transfer the
2777          * error to the original request and complete it now.
2778          */
2779         if (!result) {
2780                 obj_request_existence_set(orig_request, true);
2781         } else if (result == -ENOENT) {
2782                 obj_request_existence_set(orig_request, false);
2783         } else if (result) {
2784                 orig_request->result = result;
2785                 goto out;
2786         }
2787
2788         /*
2789          * Resubmit the original request now that we have recorded
2790          * whether the target object exists.
2791          */
2792         orig_request->result = rbd_img_obj_request_submit(orig_request);
2793 out:
2794         if (orig_request->result)
2795                 rbd_obj_request_complete(orig_request);
2796 }
2797
2798 static int rbd_img_obj_exists_submit(struct rbd_obj_request *obj_request)
2799 {
2800         struct rbd_obj_request *stat_request;
2801         struct rbd_device *rbd_dev;
2802         struct ceph_osd_client *osdc;
2803         struct page **pages = NULL;
2804         u32 page_count;
2805         size_t size;
2806         int ret;
2807
2808         /*
2809          * The response data for a STAT call consists of:
2810          *     le64 length;
2811          *     struct {
2812          *         le32 tv_sec;
2813          *         le32 tv_nsec;
2814          *     } mtime;
2815          */
2816         size = sizeof (__le64) + sizeof (__le32) + sizeof (__le32);
2817         page_count = (u32)calc_pages_for(0, size);
2818         pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2819         if (IS_ERR(pages))
2820                 return PTR_ERR(pages);
2821
2822         ret = -ENOMEM;
2823         stat_request = rbd_obj_request_create(obj_request->object_name, 0, 0,
2824                                                         OBJ_REQUEST_PAGES);
2825         if (!stat_request)
2826                 goto out;
2827
2828         rbd_obj_request_get(obj_request);
2829         stat_request->obj_request = obj_request;
2830         stat_request->pages = pages;
2831         stat_request->page_count = page_count;
2832
2833         rbd_assert(obj_request->img_request);
2834         rbd_dev = obj_request->img_request->rbd_dev;
2835         stat_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1,
2836                                                    stat_request);
2837         if (!stat_request->osd_req)
2838                 goto out;
2839         stat_request->callback = rbd_img_obj_exists_callback;
2840
2841         osd_req_op_init(stat_request->osd_req, 0, CEPH_OSD_OP_STAT);
2842         osd_req_op_raw_data_in_pages(stat_request->osd_req, 0, pages, size, 0,
2843                                         false, false);
2844         rbd_osd_req_format_read(stat_request);
2845
2846         osdc = &rbd_dev->rbd_client->client->osdc;
2847         ret = rbd_obj_request_submit(osdc, stat_request);
2848 out:
2849         if (ret)
2850                 rbd_obj_request_put(obj_request);
2851
2852         return ret;
2853 }
2854
2855 static bool img_obj_request_simple(struct rbd_obj_request *obj_request)
2856 {
2857         struct rbd_img_request *img_request;
2858         struct rbd_device *rbd_dev;
2859
2860         rbd_assert(obj_request_img_data_test(obj_request));
2861
2862         img_request = obj_request->img_request;
2863         rbd_assert(img_request);
2864         rbd_dev = img_request->rbd_dev;
2865
2866         /* Reads */
2867         if (!img_request_write_test(img_request) &&
2868             !img_request_discard_test(img_request))
2869                 return true;
2870
2871         /* Non-layered writes */
2872         if (!img_request_layered_test(img_request))
2873                 return true;
2874
2875         /*
2876          * Layered writes outside of the parent overlap range don't
2877          * share any data with the parent.
2878          */
2879         if (!obj_request_overlaps_parent(obj_request))
2880                 return true;
2881
2882         /*
2883          * Entire-object layered writes - we will overwrite whatever
2884          * parent data there is anyway.
2885          */
2886         if (!obj_request->offset &&
2887             obj_request->length == rbd_obj_bytes(&rbd_dev->header))
2888                 return true;
2889
2890         /*
2891          * If the object is known to already exist, its parent data has
2892          * already been copied.
2893          */
2894         if (obj_request_known_test(obj_request) &&
2895             obj_request_exists_test(obj_request))
2896                 return true;
2897
2898         return false;
2899 }
2900
2901 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request)
2902 {
2903         if (img_obj_request_simple(obj_request)) {
2904                 struct rbd_device *rbd_dev;
2905                 struct ceph_osd_client *osdc;
2906
2907                 rbd_dev = obj_request->img_request->rbd_dev;
2908                 osdc = &rbd_dev->rbd_client->client->osdc;
2909
2910                 return rbd_obj_request_submit(osdc, obj_request);
2911         }
2912
2913         /*
2914          * It's a layered write.  The target object might exist but
2915          * we may not know that yet.  If we know it doesn't exist,
2916          * start by reading the data for the full target object from
2917          * the parent so we can use it for a copyup to the target.
2918          */
2919         if (obj_request_known_test(obj_request))
2920                 return rbd_img_obj_parent_read_full(obj_request);
2921
2922         /* We don't know whether the target exists.  Go find out. */
2923
2924         return rbd_img_obj_exists_submit(obj_request);
2925 }
2926
2927 static int rbd_img_request_submit(struct rbd_img_request *img_request)
2928 {
2929         struct rbd_obj_request *obj_request;
2930         struct rbd_obj_request *next_obj_request;
2931
2932         dout("%s: img %p\n", __func__, img_request);
2933         for_each_obj_request_safe(img_request, obj_request, next_obj_request) {
2934                 int ret;
2935
2936                 ret = rbd_img_obj_request_submit(obj_request);
2937                 if (ret)
2938                         return ret;
2939         }
2940
2941         return 0;
2942 }
2943
2944 static void rbd_img_parent_read_callback(struct rbd_img_request *img_request)
2945 {
2946         struct rbd_obj_request *obj_request;
2947         struct rbd_device *rbd_dev;
2948         u64 obj_end;
2949         u64 img_xferred;
2950         int img_result;
2951
2952         rbd_assert(img_request_child_test(img_request));
2953
2954         /* First get what we need from the image request and release it */
2955
2956         obj_request = img_request->obj_request;
2957         img_xferred = img_request->xferred;
2958         img_result = img_request->result;
2959         rbd_img_request_put(img_request);
2960
2961         /*
2962          * If the overlap has become 0 (most likely because the
2963          * image has been flattened) we need to re-submit the
2964          * original request.
2965          */
2966         rbd_assert(obj_request);
2967         rbd_assert(obj_request->img_request);
2968         rbd_dev = obj_request->img_request->rbd_dev;
2969         if (!rbd_dev->parent_overlap) {
2970                 struct ceph_osd_client *osdc;
2971
2972                 osdc = &rbd_dev->rbd_client->client->osdc;
2973                 img_result = rbd_obj_request_submit(osdc, obj_request);
2974                 if (!img_result)
2975                         return;
2976         }
2977
2978         obj_request->result = img_result;
2979         if (obj_request->result)
2980                 goto out;
2981
2982         /*
2983          * We need to zero anything beyond the parent overlap
2984          * boundary.  Since rbd_img_obj_request_read_callback()
2985          * will zero anything beyond the end of a short read, an
2986          * easy way to do this is to pretend the data from the
2987          * parent came up short--ending at the overlap boundary.
2988          */
2989         rbd_assert(obj_request->img_offset < U64_MAX - obj_request->length);
2990         obj_end = obj_request->img_offset + obj_request->length;
2991         if (obj_end > rbd_dev->parent_overlap) {
2992                 u64 xferred = 0;
2993
2994                 if (obj_request->img_offset < rbd_dev->parent_overlap)
2995                         xferred = rbd_dev->parent_overlap -
2996                                         obj_request->img_offset;
2997
2998                 obj_request->xferred = min(img_xferred, xferred);
2999         } else {
3000                 obj_request->xferred = img_xferred;
3001         }
3002 out:
3003         rbd_img_obj_request_read_callback(obj_request);
3004         rbd_obj_request_complete(obj_request);
3005 }
3006
3007 static void rbd_img_parent_read(struct rbd_obj_request *obj_request)
3008 {
3009         struct rbd_img_request *img_request;
3010         int result;
3011
3012         rbd_assert(obj_request_img_data_test(obj_request));
3013         rbd_assert(obj_request->img_request != NULL);
3014         rbd_assert(obj_request->result == (s32) -ENOENT);
3015         rbd_assert(obj_request_type_valid(obj_request->type));
3016
3017         /* rbd_read_finish(obj_request, obj_request->length); */
3018         img_request = rbd_parent_request_create(obj_request,
3019                                                 obj_request->img_offset,
3020                                                 obj_request->length);
3021         result = -ENOMEM;
3022         if (!img_request)
3023                 goto out_err;
3024
3025         if (obj_request->type == OBJ_REQUEST_BIO)
3026                 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
3027                                                 obj_request->bio_list);
3028         else
3029                 result = rbd_img_request_fill(img_request, OBJ_REQUEST_PAGES,
3030                                                 obj_request->pages);
3031         if (result)
3032                 goto out_err;
3033
3034         img_request->callback = rbd_img_parent_read_callback;
3035         result = rbd_img_request_submit(img_request);
3036         if (result)
3037                 goto out_err;
3038
3039         return;
3040 out_err:
3041         if (img_request)
3042                 rbd_img_request_put(img_request);
3043         obj_request->result = result;
3044         obj_request->xferred = 0;
3045         obj_request_done_set(obj_request);
3046 }
3047
3048 static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev, u64 notify_id)
3049 {
3050         struct rbd_obj_request *obj_request;
3051         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3052         int ret;
3053
3054         obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0,
3055                                                         OBJ_REQUEST_NODATA);
3056         if (!obj_request)
3057                 return -ENOMEM;
3058
3059         ret = -ENOMEM;
3060         obj_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1,
3061                                                   obj_request);
3062         if (!obj_request->osd_req)
3063                 goto out;
3064
3065         osd_req_op_watch_init(obj_request->osd_req, 0, CEPH_OSD_OP_NOTIFY_ACK,
3066                                         notify_id, 0, 0);
3067         rbd_osd_req_format_read(obj_request);
3068
3069         ret = rbd_obj_request_submit(osdc, obj_request);
3070         if (ret)
3071                 goto out;
3072         ret = rbd_obj_request_wait(obj_request);
3073 out:
3074         rbd_obj_request_put(obj_request);
3075
3076         return ret;
3077 }
3078
3079 static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
3080 {
3081         struct rbd_device *rbd_dev = (struct rbd_device *)data;
3082         int ret;
3083
3084         if (!rbd_dev)
3085                 return;
3086
3087         dout("%s: \"%s\" notify_id %llu opcode %u\n", __func__,
3088                 rbd_dev->header_name, (unsigned long long)notify_id,
3089                 (unsigned int)opcode);
3090
3091         /*
3092          * Until adequate refresh error handling is in place, there is
3093          * not much we can do here, except warn.
3094          *
3095          * See http://tracker.ceph.com/issues/5040
3096          */
3097         ret = rbd_dev_refresh(rbd_dev);
3098         if (ret)
3099                 rbd_warn(rbd_dev, "refresh failed: %d", ret);
3100
3101         ret = rbd_obj_notify_ack_sync(rbd_dev, notify_id);
3102         if (ret)
3103                 rbd_warn(rbd_dev, "notify_ack ret %d", ret);
3104 }
3105
3106 /*
3107  * Send a (un)watch request and wait for the ack.  Return a request
3108  * with a ref held on success or error.
3109  */
3110 static struct rbd_obj_request *rbd_obj_watch_request_helper(
3111                                                 struct rbd_device *rbd_dev,
3112                                                 bool watch)
3113 {
3114         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3115         struct rbd_obj_request *obj_request;
3116         int ret;
3117
3118         obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0,
3119                                              OBJ_REQUEST_NODATA);
3120         if (!obj_request)
3121                 return ERR_PTR(-ENOMEM);
3122
3123         obj_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_WRITE, 1,
3124                                                   obj_request);
3125         if (!obj_request->osd_req) {
3126                 ret = -ENOMEM;
3127                 goto out;
3128         }
3129
3130         osd_req_op_watch_init(obj_request->osd_req, 0, CEPH_OSD_OP_WATCH,
3131                               rbd_dev->watch_event->cookie, 0, watch);
3132         rbd_osd_req_format_write(obj_request);
3133
3134         if (watch)
3135                 ceph_osdc_set_request_linger(osdc, obj_request->osd_req);
3136
3137         ret = rbd_obj_request_submit(osdc, obj_request);
3138         if (ret)
3139                 goto out;
3140
3141         ret = rbd_obj_request_wait(obj_request);
3142         if (ret)
3143                 goto out;
3144
3145         ret = obj_request->result;
3146         if (ret) {
3147                 if (watch)
3148                         rbd_obj_request_end(obj_request);
3149                 goto out;
3150         }
3151
3152         return obj_request;
3153
3154 out:
3155         rbd_obj_request_put(obj_request);
3156         return ERR_PTR(ret);
3157 }
3158
3159 /*
3160  * Initiate a watch request, synchronously.
3161  */
3162 static int rbd_dev_header_watch_sync(struct rbd_device *rbd_dev)
3163 {
3164         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3165         struct rbd_obj_request *obj_request;
3166         int ret;
3167
3168         rbd_assert(!rbd_dev->watch_event);
3169         rbd_assert(!rbd_dev->watch_request);
3170
3171         ret = ceph_osdc_create_event(osdc, rbd_watch_cb, rbd_dev,
3172                                      &rbd_dev->watch_event);
3173         if (ret < 0)
3174                 return ret;
3175
3176         obj_request = rbd_obj_watch_request_helper(rbd_dev, true);
3177         if (IS_ERR(obj_request)) {
3178                 ceph_osdc_cancel_event(rbd_dev->watch_event);
3179                 rbd_dev->watch_event = NULL;
3180                 return PTR_ERR(obj_request);
3181         }
3182
3183         /*
3184          * A watch request is set to linger, so the underlying osd
3185          * request won't go away until we unregister it.  We retain
3186          * a pointer to the object request during that time (in
3187          * rbd_dev->watch_request), so we'll keep a reference to it.
3188          * We'll drop that reference after we've unregistered it in
3189          * rbd_dev_header_unwatch_sync().
3190          */
3191         rbd_dev->watch_request = obj_request;
3192
3193         return 0;
3194 }
3195
3196 /*
3197  * Tear down a watch request, synchronously.
3198  */
3199 static void rbd_dev_header_unwatch_sync(struct rbd_device *rbd_dev)
3200 {
3201         struct rbd_obj_request *obj_request;
3202
3203         rbd_assert(rbd_dev->watch_event);
3204         rbd_assert(rbd_dev->watch_request);
3205
3206         rbd_obj_request_end(rbd_dev->watch_request);
3207         rbd_obj_request_put(rbd_dev->watch_request);
3208         rbd_dev->watch_request = NULL;
3209
3210         obj_request = rbd_obj_watch_request_helper(rbd_dev, false);
3211         if (!IS_ERR(obj_request))
3212                 rbd_obj_request_put(obj_request);
3213         else
3214                 rbd_warn(rbd_dev, "unable to tear down watch request (%ld)",
3215                          PTR_ERR(obj_request));
3216
3217         ceph_osdc_cancel_event(rbd_dev->watch_event);
3218         rbd_dev->watch_event = NULL;
3219 }
3220
3221 /*
3222  * Synchronous osd object method call.  Returns the number of bytes
3223  * returned in the outbound buffer, or a negative error code.
3224  */
3225 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
3226                              const char *object_name,
3227                              const char *class_name,
3228                              const char *method_name,
3229                              const void *outbound,
3230                              size_t outbound_size,
3231                              void *inbound,
3232                              size_t inbound_size)
3233 {
3234         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3235         struct rbd_obj_request *obj_request;
3236         struct page **pages;
3237         u32 page_count;
3238         int ret;
3239
3240         /*
3241          * Method calls are ultimately read operations.  The result
3242          * should placed into the inbound buffer provided.  They
3243          * also supply outbound data--parameters for the object
3244          * method.  Currently if this is present it will be a
3245          * snapshot id.
3246          */
3247         page_count = (u32)calc_pages_for(0, inbound_size);
3248         pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
3249         if (IS_ERR(pages))
3250                 return PTR_ERR(pages);
3251
3252         ret = -ENOMEM;
3253         obj_request = rbd_obj_request_create(object_name, 0, inbound_size,
3254                                                         OBJ_REQUEST_PAGES);
3255         if (!obj_request)
3256                 goto out;
3257
3258         obj_request->pages = pages;
3259         obj_request->page_count = page_count;
3260
3261         obj_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1,
3262                                                   obj_request);
3263         if (!obj_request->osd_req)
3264                 goto out;
3265
3266         osd_req_op_cls_init(obj_request->osd_req, 0, CEPH_OSD_OP_CALL,
3267                                         class_name, method_name);
3268         if (outbound_size) {
3269                 struct ceph_pagelist *pagelist;
3270
3271                 pagelist = kmalloc(sizeof (*pagelist), GFP_NOFS);
3272                 if (!pagelist)
3273                         goto out;
3274
3275                 ceph_pagelist_init(pagelist);
3276                 ceph_pagelist_append(pagelist, outbound, outbound_size);
3277                 osd_req_op_cls_request_data_pagelist(obj_request->osd_req, 0,
3278                                                 pagelist);
3279         }
3280         osd_req_op_cls_response_data_pages(obj_request->osd_req, 0,
3281                                         obj_request->pages, inbound_size,
3282                                         0, false, false);
3283         rbd_osd_req_format_read(obj_request);
3284
3285         ret = rbd_obj_request_submit(osdc, obj_request);
3286         if (ret)
3287                 goto out;
3288         ret = rbd_obj_request_wait(obj_request);
3289         if (ret)
3290                 goto out;
3291
3292         ret = obj_request->result;
3293         if (ret < 0)
3294                 goto out;
3295
3296         rbd_assert(obj_request->xferred < (u64)INT_MAX);
3297         ret = (int)obj_request->xferred;
3298         ceph_copy_from_page_vector(pages, inbound, 0, obj_request->xferred);
3299 out:
3300         if (obj_request)
3301                 rbd_obj_request_put(obj_request);
3302         else
3303                 ceph_release_page_vector(pages, page_count);
3304
3305         return ret;
3306 }
3307
3308 static void rbd_handle_request(struct rbd_device *rbd_dev, struct request *rq)
3309 {
3310         struct rbd_img_request *img_request;
3311         struct ceph_snap_context *snapc = NULL;
3312         u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
3313         u64 length = blk_rq_bytes(rq);
3314         enum obj_operation_type op_type;
3315         u64 mapping_size;
3316         int result;
3317
3318         if (rq->cmd_flags & REQ_DISCARD)
3319                 op_type = OBJ_OP_DISCARD;
3320         else if (rq->cmd_flags & REQ_WRITE)
3321                 op_type = OBJ_OP_WRITE;
3322         else
3323                 op_type = OBJ_OP_READ;
3324
3325         /* Ignore/skip any zero-length requests */
3326
3327         if (!length) {
3328                 dout("%s: zero-length request\n", __func__);
3329                 result = 0;
3330                 goto err_rq;
3331         }
3332
3333         /* Only reads are allowed to a read-only device */
3334
3335         if (op_type != OBJ_OP_READ) {
3336                 if (rbd_dev->mapping.read_only) {
3337                         result = -EROFS;
3338                         goto err_rq;
3339                 }
3340                 rbd_assert(rbd_dev->spec->snap_id == CEPH_NOSNAP);
3341         }
3342
3343         /*
3344          * Quit early if the mapped snapshot no longer exists.  It's
3345          * still possible the snapshot will have disappeared by the
3346          * time our request arrives at the osd, but there's no sense in
3347          * sending it if we already know.
3348          */
3349         if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags)) {
3350                 dout("request for non-existent snapshot");
3351                 rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
3352                 result = -ENXIO;
3353                 goto err_rq;
3354         }
3355
3356         if (offset && length > U64_MAX - offset + 1) {
3357                 rbd_warn(rbd_dev, "bad request range (%llu~%llu)", offset,
3358                          length);
3359                 result = -EINVAL;
3360                 goto err_rq;    /* Shouldn't happen */
3361         }
3362
3363         down_read(&rbd_dev->header_rwsem);
3364         mapping_size = rbd_dev->mapping.size;
3365         if (op_type != OBJ_OP_READ) {