kernel/drivers/block/rbd.c

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