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