drivers/block/rbd.c

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