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