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