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