drivers/block/rbd.c

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