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