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