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