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