rbd: set max_segments to USHRT_MAX
[sfrench/cifs-2.6.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/decode.h>
36 #include <linux/parser.h>
37 #include <linux/bsearch.h>
38
39 #include <linux/kernel.h>
40 #include <linux/device.h>
41 #include <linux/module.h>
42 #include <linux/blk-mq.h>
43 #include <linux/fs.h>
44 #include <linux/blkdev.h>
45 #include <linux/slab.h>
46 #include <linux/idr.h>
47 #include <linux/workqueue.h>
48
49 #include "rbd_types.h"
50
51 #define RBD_DEBUG       /* Activate rbd_assert() calls */
52
53 /*
54  * The basic unit of block I/O is a sector.  It is interpreted in a
55  * number of contexts in Linux (blk, bio, genhd), but the default is
56  * universally 512 bytes.  These symbols are just slightly more
57  * meaningful than the bare numbers they represent.
58  */
59 #define SECTOR_SHIFT    9
60 #define SECTOR_SIZE     (1ULL << SECTOR_SHIFT)
61
62 /*
63  * Increment the given counter and return its updated value.
64  * If the counter is already 0 it will not be incremented.
65  * If the counter is already at its maximum value returns
66  * -EINVAL without updating it.
67  */
68 static int atomic_inc_return_safe(atomic_t *v)
69 {
70         unsigned int counter;
71
72         counter = (unsigned int)__atomic_add_unless(v, 1, 0);
73         if (counter <= (unsigned int)INT_MAX)
74                 return (int)counter;
75
76         atomic_dec(v);
77
78         return -EINVAL;
79 }
80
81 /* Decrement the counter.  Return the resulting value, or -EINVAL */
82 static int atomic_dec_return_safe(atomic_t *v)
83 {
84         int counter;
85
86         counter = atomic_dec_return(v);
87         if (counter >= 0)
88                 return counter;
89
90         atomic_inc(v);
91
92         return -EINVAL;
93 }
94
95 #define RBD_DRV_NAME "rbd"
96
97 #define RBD_MINORS_PER_MAJOR            256
98 #define RBD_SINGLE_MAJOR_PART_SHIFT     4
99
100 #define RBD_MAX_PARENT_CHAIN_LEN        16
101
102 #define RBD_SNAP_DEV_NAME_PREFIX        "snap_"
103 #define RBD_MAX_SNAP_NAME_LEN   \
104                         (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
105
106 #define RBD_MAX_SNAP_COUNT      510     /* allows max snapc to fit in 4KB */
107
108 #define RBD_SNAP_HEAD_NAME      "-"
109
110 #define BAD_SNAP_INDEX  U32_MAX         /* invalid index into snap array */
111
112 /* This allows a single page to hold an image name sent by OSD */
113 #define RBD_IMAGE_NAME_LEN_MAX  (PAGE_SIZE - sizeof (__le32) - 1)
114 #define RBD_IMAGE_ID_LEN_MAX    64
115
116 #define RBD_OBJ_PREFIX_LEN_MAX  64
117
118 #define RBD_NOTIFY_TIMEOUT      5       /* seconds */
119 #define RBD_RETRY_DELAY         msecs_to_jiffies(1000)
120
121 /* Feature bits */
122
123 #define RBD_FEATURE_LAYERING            (1ULL<<0)
124 #define RBD_FEATURE_STRIPINGV2          (1ULL<<1)
125 #define RBD_FEATURE_EXCLUSIVE_LOCK      (1ULL<<2)
126 #define RBD_FEATURE_DATA_POOL           (1ULL<<7)
127
128 #define RBD_FEATURES_ALL        (RBD_FEATURE_LAYERING |         \
129                                  RBD_FEATURE_STRIPINGV2 |       \
130                                  RBD_FEATURE_EXCLUSIVE_LOCK |   \
131                                  RBD_FEATURE_DATA_POOL)
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
191         const char      *image_id;
192         const char      *image_name;
193
194         u64             snap_id;
195         const char      *snap_name;
196
197         struct kref     kref;
198 };
199
200 /*
201  * an instance of the client.  multiple devices may share an rbd client.
202  */
203 struct rbd_client {
204         struct ceph_client      *client;
205         struct kref             kref;
206         struct list_head        node;
207 };
208
209 struct rbd_img_request;
210 typedef void (*rbd_img_callback_t)(struct rbd_img_request *);
211
212 #define BAD_WHICH       U32_MAX         /* Good which or bad which, which? */
213
214 struct rbd_obj_request;
215 typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);
216
217 enum obj_request_type {
218         OBJ_REQUEST_NODATA, OBJ_REQUEST_BIO, OBJ_REQUEST_PAGES
219 };
220
221 enum obj_operation_type {
222         OBJ_OP_WRITE,
223         OBJ_OP_READ,
224         OBJ_OP_DISCARD,
225 };
226
227 enum obj_req_flags {
228         OBJ_REQ_DONE,           /* completion flag: not done = 0, done = 1 */
229         OBJ_REQ_IMG_DATA,       /* object usage: standalone = 0, image = 1 */
230         OBJ_REQ_KNOWN,          /* EXISTS flag valid: no = 0, yes = 1 */
231         OBJ_REQ_EXISTS,         /* target exists: no = 0, yes = 1 */
232 };
233
234 struct rbd_obj_request {
235         u64                     object_no;
236         u64                     offset;         /* object start byte */
237         u64                     length;         /* bytes from offset */
238         unsigned long           flags;
239
240         /*
241          * An object request associated with an image will have its
242          * img_data flag set; a standalone object request will not.
243          *
244          * A standalone object request will have which == BAD_WHICH
245          * and a null obj_request pointer.
246          *
247          * An object request initiated in support of a layered image
248          * object (to check for its existence before a write) will
249          * have which == BAD_WHICH and a non-null obj_request pointer.
250          *
251          * Finally, an object request for rbd image data will have
252          * which != BAD_WHICH, and will have a non-null img_request
253          * pointer.  The value of which will be in the range
254          * 0..(img_request->obj_request_count-1).
255          */
256         union {
257                 struct rbd_obj_request  *obj_request;   /* STAT op */
258                 struct {
259                         struct rbd_img_request  *img_request;
260                         u64                     img_offset;
261                         /* links for img_request->obj_requests list */
262                         struct list_head        links;
263                 };
264         };
265         u32                     which;          /* posn image request list */
266
267         enum obj_request_type   type;
268         union {
269                 struct bio      *bio_list;
270                 struct {
271                         struct page     **pages;
272                         u32             page_count;
273                 };
274         };
275         struct page             **copyup_pages;
276         u32                     copyup_page_count;
277
278         struct ceph_osd_request *osd_req;
279
280         u64                     xferred;        /* bytes transferred */
281         int                     result;
282
283         rbd_obj_callback_t      callback;
284         struct completion       completion;
285
286         struct kref             kref;
287 };
288
289 enum img_req_flags {
290         IMG_REQ_WRITE,          /* I/O direction: read = 0, write = 1 */
291         IMG_REQ_CHILD,          /* initiator: block = 0, child image = 1 */
292         IMG_REQ_LAYERED,        /* ENOENT handling: normal = 0, layered = 1 */
293         IMG_REQ_DISCARD,        /* discard: normal = 0, discard request = 1 */
294 };
295
296 struct rbd_img_request {
297         struct rbd_device       *rbd_dev;
298         u64                     offset; /* starting image byte offset */
299         u64                     length; /* byte count from offset */
300         unsigned long           flags;
301         union {
302                 u64                     snap_id;        /* for reads */
303                 struct ceph_snap_context *snapc;        /* for writes */
304         };
305         union {
306                 struct request          *rq;            /* block request */
307                 struct rbd_obj_request  *obj_request;   /* obj req initiator */
308         };
309         struct page             **copyup_pages;
310         u32                     copyup_page_count;
311         spinlock_t              completion_lock;/* protects next_completion */
312         u32                     next_completion;
313         rbd_img_callback_t      callback;
314         u64                     xferred;/* aggregate bytes transferred */
315         int                     result; /* first nonzero obj_request result */
316
317         u32                     obj_request_count;
318         struct list_head        obj_requests;   /* rbd_obj_request structs */
319
320         struct kref             kref;
321 };
322
323 #define for_each_obj_request(ireq, oreq) \
324         list_for_each_entry(oreq, &(ireq)->obj_requests, links)
325 #define for_each_obj_request_from(ireq, oreq) \
326         list_for_each_entry_from(oreq, &(ireq)->obj_requests, links)
327 #define for_each_obj_request_safe(ireq, oreq, n) \
328         list_for_each_entry_safe_reverse(oreq, n, &(ireq)->obj_requests, links)
329
330 enum rbd_watch_state {
331         RBD_WATCH_STATE_UNREGISTERED,
332         RBD_WATCH_STATE_REGISTERED,
333         RBD_WATCH_STATE_ERROR,
334 };
335
336 enum rbd_lock_state {
337         RBD_LOCK_STATE_UNLOCKED,
338         RBD_LOCK_STATE_LOCKED,
339         RBD_LOCK_STATE_RELEASING,
340 };
341
342 /* WatchNotify::ClientId */
343 struct rbd_client_id {
344         u64 gid;
345         u64 handle;
346 };
347
348 struct rbd_mapping {
349         u64                     size;
350         u64                     features;
351 };
352
353 /*
354  * a single device
355  */
356 struct rbd_device {
357         int                     dev_id;         /* blkdev unique id */
358
359         int                     major;          /* blkdev assigned major */
360         int                     minor;
361         struct gendisk          *disk;          /* blkdev's gendisk and rq */
362
363         u32                     image_format;   /* Either 1 or 2 */
364         struct rbd_client       *rbd_client;
365
366         char                    name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
367
368         spinlock_t              lock;           /* queue, flags, open_count */
369
370         struct rbd_image_header header;
371         unsigned long           flags;          /* possibly lock protected */
372         struct rbd_spec         *spec;
373         struct rbd_options      *opts;
374         char                    *config_info;   /* add{,_single_major} string */
375
376         struct ceph_object_id   header_oid;
377         struct ceph_object_locator header_oloc;
378
379         struct ceph_file_layout layout;         /* used for all rbd requests */
380
381         struct mutex            watch_mutex;
382         enum rbd_watch_state    watch_state;
383         struct ceph_osd_linger_request *watch_handle;
384         u64                     watch_cookie;
385         struct delayed_work     watch_dwork;
386
387         struct rw_semaphore     lock_rwsem;
388         enum rbd_lock_state     lock_state;
389         char                    lock_cookie[32];
390         struct rbd_client_id    owner_cid;
391         struct work_struct      acquired_lock_work;
392         struct work_struct      released_lock_work;
393         struct delayed_work     lock_dwork;
394         struct work_struct      unlock_work;
395         wait_queue_head_t       lock_waitq;
396
397         struct workqueue_struct *task_wq;
398
399         struct rbd_spec         *parent_spec;
400         u64                     parent_overlap;
401         atomic_t                parent_ref;
402         struct rbd_device       *parent;
403
404         /* Block layer tags. */
405         struct blk_mq_tag_set   tag_set;
406
407         /* protects updating the header */
408         struct rw_semaphore     header_rwsem;
409
410         struct rbd_mapping      mapping;
411
412         struct list_head        node;
413
414         /* sysfs related */
415         struct device           dev;
416         unsigned long           open_count;     /* protected by lock */
417 };
418
419 /*
420  * Flag bits for rbd_dev->flags:
421  * - REMOVING (which is coupled with rbd_dev->open_count) is protected
422  *   by rbd_dev->lock
423  * - BLACKLISTED is protected by rbd_dev->lock_rwsem
424  */
425 enum rbd_dev_flags {
426         RBD_DEV_FLAG_EXISTS,    /* mapped snapshot has not been deleted */
427         RBD_DEV_FLAG_REMOVING,  /* this mapping is being removed */
428         RBD_DEV_FLAG_BLACKLISTED, /* our ceph_client is blacklisted */
429 };
430
431 static DEFINE_MUTEX(client_mutex);      /* Serialize client creation */
432
433 static LIST_HEAD(rbd_dev_list);    /* devices */
434 static DEFINE_SPINLOCK(rbd_dev_list_lock);
435
436 static LIST_HEAD(rbd_client_list);              /* clients */
437 static DEFINE_SPINLOCK(rbd_client_list_lock);
438
439 /* Slab caches for frequently-allocated structures */
440
441 static struct kmem_cache        *rbd_img_request_cache;
442 static struct kmem_cache        *rbd_obj_request_cache;
443
444 static struct bio_set           *rbd_bio_clone;
445
446 static int rbd_major;
447 static DEFINE_IDA(rbd_dev_id_ida);
448
449 static struct workqueue_struct *rbd_wq;
450
451 /*
452  * single-major requires >= 0.75 version of userspace rbd utility.
453  */
454 static bool single_major = true;
455 module_param(single_major, bool, S_IRUGO);
456 MODULE_PARM_DESC(single_major, "Use a single major number for all rbd devices (default: true)");
457
458 static int rbd_img_request_submit(struct rbd_img_request *img_request);
459
460 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
461                        size_t count);
462 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
463                           size_t count);
464 static ssize_t rbd_add_single_major(struct bus_type *bus, const char *buf,
465                                     size_t count);
466 static ssize_t rbd_remove_single_major(struct bus_type *bus, const char *buf,
467                                        size_t count);
468 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth);
469 static void rbd_spec_put(struct rbd_spec *spec);
470
471 static int rbd_dev_id_to_minor(int dev_id)
472 {
473         return dev_id << RBD_SINGLE_MAJOR_PART_SHIFT;
474 }
475
476 static int minor_to_rbd_dev_id(int minor)
477 {
478         return minor >> RBD_SINGLE_MAJOR_PART_SHIFT;
479 }
480
481 static bool __rbd_is_lock_owner(struct rbd_device *rbd_dev)
482 {
483         return rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED ||
484                rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING;
485 }
486
487 static bool rbd_is_lock_owner(struct rbd_device *rbd_dev)
488 {
489         bool is_lock_owner;
490
491         down_read(&rbd_dev->lock_rwsem);
492         is_lock_owner = __rbd_is_lock_owner(rbd_dev);
493         up_read(&rbd_dev->lock_rwsem);
494         return is_lock_owner;
495 }
496
497 static ssize_t rbd_supported_features_show(struct bus_type *bus, char *buf)
498 {
499         return sprintf(buf, "0x%llx\n", RBD_FEATURES_SUPPORTED);
500 }
501
502 static BUS_ATTR(add, S_IWUSR, NULL, rbd_add);
503 static BUS_ATTR(remove, S_IWUSR, NULL, rbd_remove);
504 static BUS_ATTR(add_single_major, S_IWUSR, NULL, rbd_add_single_major);
505 static BUS_ATTR(remove_single_major, S_IWUSR, NULL, rbd_remove_single_major);
506 static BUS_ATTR(supported_features, S_IRUGO, rbd_supported_features_show, NULL);
507
508 static struct attribute *rbd_bus_attrs[] = {
509         &bus_attr_add.attr,
510         &bus_attr_remove.attr,
511         &bus_attr_add_single_major.attr,
512         &bus_attr_remove_single_major.attr,
513         &bus_attr_supported_features.attr,
514         NULL,
515 };
516
517 static umode_t rbd_bus_is_visible(struct kobject *kobj,
518                                   struct attribute *attr, int index)
519 {
520         if (!single_major &&
521             (attr == &bus_attr_add_single_major.attr ||
522              attr == &bus_attr_remove_single_major.attr))
523                 return 0;
524
525         return attr->mode;
526 }
527
528 static const struct attribute_group rbd_bus_group = {
529         .attrs = rbd_bus_attrs,
530         .is_visible = rbd_bus_is_visible,
531 };
532 __ATTRIBUTE_GROUPS(rbd_bus);
533
534 static struct bus_type rbd_bus_type = {
535         .name           = "rbd",
536         .bus_groups     = rbd_bus_groups,
537 };
538
539 static void rbd_root_dev_release(struct device *dev)
540 {
541 }
542
543 static struct device rbd_root_dev = {
544         .init_name =    "rbd",
545         .release =      rbd_root_dev_release,
546 };
547
548 static __printf(2, 3)
549 void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
550 {
551         struct va_format vaf;
552         va_list args;
553
554         va_start(args, fmt);
555         vaf.fmt = fmt;
556         vaf.va = &args;
557
558         if (!rbd_dev)
559                 printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
560         else if (rbd_dev->disk)
561                 printk(KERN_WARNING "%s: %s: %pV\n",
562                         RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
563         else if (rbd_dev->spec && rbd_dev->spec->image_name)
564                 printk(KERN_WARNING "%s: image %s: %pV\n",
565                         RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
566         else if (rbd_dev->spec && rbd_dev->spec->image_id)
567                 printk(KERN_WARNING "%s: id %s: %pV\n",
568                         RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
569         else    /* punt */
570                 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
571                         RBD_DRV_NAME, rbd_dev, &vaf);
572         va_end(args);
573 }
574
575 #ifdef RBD_DEBUG
576 #define rbd_assert(expr)                                                \
577                 if (unlikely(!(expr))) {                                \
578                         printk(KERN_ERR "\nAssertion failure in %s() "  \
579                                                 "at line %d:\n\n"       \
580                                         "\trbd_assert(%s);\n\n",        \
581                                         __func__, __LINE__, #expr);     \
582                         BUG();                                          \
583                 }
584 #else /* !RBD_DEBUG */
585 #  define rbd_assert(expr)      ((void) 0)
586 #endif /* !RBD_DEBUG */
587
588 static void rbd_osd_copyup_callback(struct rbd_obj_request *obj_request);
589 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request);
590 static void rbd_img_parent_read(struct rbd_obj_request *obj_request);
591 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev);
592
593 static int rbd_dev_refresh(struct rbd_device *rbd_dev);
594 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev);
595 static int rbd_dev_header_info(struct rbd_device *rbd_dev);
596 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev);
597 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
598                                         u64 snap_id);
599 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
600                                 u8 *order, u64 *snap_size);
601 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
602                 u64 *snap_features);
603
604 static int rbd_open(struct block_device *bdev, fmode_t mode)
605 {
606         struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
607         bool removing = false;
608
609         spin_lock_irq(&rbd_dev->lock);
610         if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
611                 removing = true;
612         else
613                 rbd_dev->open_count++;
614         spin_unlock_irq(&rbd_dev->lock);
615         if (removing)
616                 return -ENOENT;
617
618         (void) get_device(&rbd_dev->dev);
619
620         return 0;
621 }
622
623 static void rbd_release(struct gendisk *disk, fmode_t mode)
624 {
625         struct rbd_device *rbd_dev = disk->private_data;
626         unsigned long open_count_before;
627
628         spin_lock_irq(&rbd_dev->lock);
629         open_count_before = rbd_dev->open_count--;
630         spin_unlock_irq(&rbd_dev->lock);
631         rbd_assert(open_count_before > 0);
632
633         put_device(&rbd_dev->dev);
634 }
635
636 static int rbd_ioctl_set_ro(struct rbd_device *rbd_dev, unsigned long arg)
637 {
638         int ro;
639
640         if (get_user(ro, (int __user *)arg))
641                 return -EFAULT;
642
643         /* Snapshots can't be marked read-write */
644         if (rbd_dev->spec->snap_id != CEPH_NOSNAP && !ro)
645                 return -EROFS;
646
647         /* Let blkdev_roset() handle it */
648         return -ENOTTY;
649 }
650
651 static int rbd_ioctl(struct block_device *bdev, fmode_t mode,
652                         unsigned int cmd, unsigned long arg)
653 {
654         struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
655         int ret;
656
657         switch (cmd) {
658         case BLKROSET:
659                 ret = rbd_ioctl_set_ro(rbd_dev, arg);
660                 break;
661         default:
662                 ret = -ENOTTY;
663         }
664
665         return ret;
666 }
667
668 #ifdef CONFIG_COMPAT
669 static int rbd_compat_ioctl(struct block_device *bdev, fmode_t mode,
670                                 unsigned int cmd, unsigned long arg)
671 {
672         return rbd_ioctl(bdev, mode, cmd, arg);
673 }
674 #endif /* CONFIG_COMPAT */
675
676 static const struct block_device_operations rbd_bd_ops = {
677         .owner                  = THIS_MODULE,
678         .open                   = rbd_open,
679         .release                = rbd_release,
680         .ioctl                  = rbd_ioctl,
681 #ifdef CONFIG_COMPAT
682         .compat_ioctl           = rbd_compat_ioctl,
683 #endif
684 };
685
686 /*
687  * Initialize an rbd client instance.  Success or not, this function
688  * consumes ceph_opts.  Caller holds client_mutex.
689  */
690 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
691 {
692         struct rbd_client *rbdc;
693         int ret = -ENOMEM;
694
695         dout("%s:\n", __func__);
696         rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
697         if (!rbdc)
698                 goto out_opt;
699
700         kref_init(&rbdc->kref);
701         INIT_LIST_HEAD(&rbdc->node);
702
703         rbdc->client = ceph_create_client(ceph_opts, rbdc);
704         if (IS_ERR(rbdc->client))
705                 goto out_rbdc;
706         ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
707
708         ret = ceph_open_session(rbdc->client);
709         if (ret < 0)
710                 goto out_client;
711
712         spin_lock(&rbd_client_list_lock);
713         list_add_tail(&rbdc->node, &rbd_client_list);
714         spin_unlock(&rbd_client_list_lock);
715
716         dout("%s: rbdc %p\n", __func__, rbdc);
717
718         return rbdc;
719 out_client:
720         ceph_destroy_client(rbdc->client);
721 out_rbdc:
722         kfree(rbdc);
723 out_opt:
724         if (ceph_opts)
725                 ceph_destroy_options(ceph_opts);
726         dout("%s: error %d\n", __func__, ret);
727
728         return ERR_PTR(ret);
729 }
730
731 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
732 {
733         kref_get(&rbdc->kref);
734
735         return rbdc;
736 }
737
738 /*
739  * Find a ceph client with specific addr and configuration.  If
740  * found, bump its reference count.
741  */
742 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
743 {
744         struct rbd_client *client_node;
745         bool found = false;
746
747         if (ceph_opts->flags & CEPH_OPT_NOSHARE)
748                 return NULL;
749
750         spin_lock(&rbd_client_list_lock);
751         list_for_each_entry(client_node, &rbd_client_list, node) {
752                 if (!ceph_compare_options(ceph_opts, client_node->client)) {
753                         __rbd_get_client(client_node);
754
755                         found = true;
756                         break;
757                 }
758         }
759         spin_unlock(&rbd_client_list_lock);
760
761         return found ? client_node : NULL;
762 }
763
764 /*
765  * (Per device) rbd map options
766  */
767 enum {
768         Opt_queue_depth,
769         Opt_last_int,
770         /* int args above */
771         Opt_last_string,
772         /* string args above */
773         Opt_read_only,
774         Opt_read_write,
775         Opt_lock_on_read,
776         Opt_exclusive,
777         Opt_err
778 };
779
780 static match_table_t rbd_opts_tokens = {
781         {Opt_queue_depth, "queue_depth=%d"},
782         /* int args above */
783         /* string args above */
784         {Opt_read_only, "read_only"},
785         {Opt_read_only, "ro"},          /* Alternate spelling */
786         {Opt_read_write, "read_write"},
787         {Opt_read_write, "rw"},         /* Alternate spelling */
788         {Opt_lock_on_read, "lock_on_read"},
789         {Opt_exclusive, "exclusive"},
790         {Opt_err, NULL}
791 };
792
793 struct rbd_options {
794         int     queue_depth;
795         bool    read_only;
796         bool    lock_on_read;
797         bool    exclusive;
798 };
799
800 #define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_MAX_RQ
801 #define RBD_READ_ONLY_DEFAULT   false
802 #define RBD_LOCK_ON_READ_DEFAULT false
803 #define RBD_EXCLUSIVE_DEFAULT   false
804
805 static int parse_rbd_opts_token(char *c, void *private)
806 {
807         struct rbd_options *rbd_opts = private;
808         substring_t argstr[MAX_OPT_ARGS];
809         int token, intval, ret;
810
811         token = match_token(c, rbd_opts_tokens, argstr);
812         if (token < Opt_last_int) {
813                 ret = match_int(&argstr[0], &intval);
814                 if (ret < 0) {
815                         pr_err("bad mount option arg (not int) at '%s'\n", c);
816                         return ret;
817                 }
818                 dout("got int token %d val %d\n", token, intval);
819         } else if (token > Opt_last_int && token < Opt_last_string) {
820                 dout("got string token %d val %s\n", token, argstr[0].from);
821         } else {
822                 dout("got token %d\n", token);
823         }
824
825         switch (token) {
826         case Opt_queue_depth:
827                 if (intval < 1) {
828                         pr_err("queue_depth out of range\n");
829                         return -EINVAL;
830                 }
831                 rbd_opts->queue_depth = intval;
832                 break;
833         case Opt_read_only:
834                 rbd_opts->read_only = true;
835                 break;
836         case Opt_read_write:
837                 rbd_opts->read_only = false;
838                 break;
839         case Opt_lock_on_read:
840                 rbd_opts->lock_on_read = true;
841                 break;
842         case Opt_exclusive:
843                 rbd_opts->exclusive = true;
844                 break;
845         default:
846                 /* libceph prints "bad option" msg */
847                 return -EINVAL;
848         }
849
850         return 0;
851 }
852
853 static char* obj_op_name(enum obj_operation_type op_type)
854 {
855         switch (op_type) {
856         case OBJ_OP_READ:
857                 return "read";
858         case OBJ_OP_WRITE:
859                 return "write";
860         case OBJ_OP_DISCARD:
861                 return "discard";
862         default:
863                 return "???";
864         }
865 }
866
867 /*
868  * Get a ceph client with specific addr and configuration, if one does
869  * not exist create it.  Either way, ceph_opts is consumed by this
870  * function.
871  */
872 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
873 {
874         struct rbd_client *rbdc;
875
876         mutex_lock_nested(&client_mutex, SINGLE_DEPTH_NESTING);
877         rbdc = rbd_client_find(ceph_opts);
878         if (rbdc)       /* using an existing client */
879                 ceph_destroy_options(ceph_opts);
880         else
881                 rbdc = rbd_client_create(ceph_opts);
882         mutex_unlock(&client_mutex);
883
884         return rbdc;
885 }
886
887 /*
888  * Destroy ceph client
889  *
890  * Caller must hold rbd_client_list_lock.
891  */
892 static void rbd_client_release(struct kref *kref)
893 {
894         struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
895
896         dout("%s: rbdc %p\n", __func__, rbdc);
897         spin_lock(&rbd_client_list_lock);
898         list_del(&rbdc->node);
899         spin_unlock(&rbd_client_list_lock);
900
901         ceph_destroy_client(rbdc->client);
902         kfree(rbdc);
903 }
904
905 /*
906  * Drop reference to ceph client node. If it's not referenced anymore, release
907  * it.
908  */
909 static void rbd_put_client(struct rbd_client *rbdc)
910 {
911         if (rbdc)
912                 kref_put(&rbdc->kref, rbd_client_release);
913 }
914
915 static bool rbd_image_format_valid(u32 image_format)
916 {
917         return image_format == 1 || image_format == 2;
918 }
919
920 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
921 {
922         size_t size;
923         u32 snap_count;
924
925         /* The header has to start with the magic rbd header text */
926         if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
927                 return false;
928
929         /* The bio layer requires at least sector-sized I/O */
930
931         if (ondisk->options.order < SECTOR_SHIFT)
932                 return false;
933
934         /* If we use u64 in a few spots we may be able to loosen this */
935
936         if (ondisk->options.order > 8 * sizeof (int) - 1)
937                 return false;
938
939         /*
940          * The size of a snapshot header has to fit in a size_t, and
941          * that limits the number of snapshots.
942          */
943         snap_count = le32_to_cpu(ondisk->snap_count);
944         size = SIZE_MAX - sizeof (struct ceph_snap_context);
945         if (snap_count > size / sizeof (__le64))
946                 return false;
947
948         /*
949          * Not only that, but the size of the entire the snapshot
950          * header must also be representable in a size_t.
951          */
952         size -= snap_count * sizeof (__le64);
953         if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
954                 return false;
955
956         return true;
957 }
958
959 /*
960  * returns the size of an object in the image
961  */
962 static u32 rbd_obj_bytes(struct rbd_image_header *header)
963 {
964         return 1U << header->obj_order;
965 }
966
967 static void rbd_init_layout(struct rbd_device *rbd_dev)
968 {
969         if (rbd_dev->header.stripe_unit == 0 ||
970             rbd_dev->header.stripe_count == 0) {
971                 rbd_dev->header.stripe_unit = rbd_obj_bytes(&rbd_dev->header);
972                 rbd_dev->header.stripe_count = 1;
973         }
974
975         rbd_dev->layout.stripe_unit = rbd_dev->header.stripe_unit;
976         rbd_dev->layout.stripe_count = rbd_dev->header.stripe_count;
977         rbd_dev->layout.object_size = rbd_obj_bytes(&rbd_dev->header);
978         rbd_dev->layout.pool_id = rbd_dev->header.data_pool_id == CEPH_NOPOOL ?
979                           rbd_dev->spec->pool_id : rbd_dev->header.data_pool_id;
980         RCU_INIT_POINTER(rbd_dev->layout.pool_ns, NULL);
981 }
982
983 /*
984  * Fill an rbd image header with information from the given format 1
985  * on-disk header.
986  */
987 static int rbd_header_from_disk(struct rbd_device *rbd_dev,
988                                  struct rbd_image_header_ondisk *ondisk)
989 {
990         struct rbd_image_header *header = &rbd_dev->header;
991         bool first_time = header->object_prefix == NULL;
992         struct ceph_snap_context *snapc;
993         char *object_prefix = NULL;
994         char *snap_names = NULL;
995         u64 *snap_sizes = NULL;
996         u32 snap_count;
997         int ret = -ENOMEM;
998         u32 i;
999
1000         /* Allocate this now to avoid having to handle failure below */
1001
1002         if (first_time) {
1003                 object_prefix = kstrndup(ondisk->object_prefix,
1004                                          sizeof(ondisk->object_prefix),
1005                                          GFP_KERNEL);
1006                 if (!object_prefix)
1007                         return -ENOMEM;
1008         }
1009
1010         /* Allocate the snapshot context and fill it in */
1011
1012         snap_count = le32_to_cpu(ondisk->snap_count);
1013         snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
1014         if (!snapc)
1015                 goto out_err;
1016         snapc->seq = le64_to_cpu(ondisk->snap_seq);
1017         if (snap_count) {
1018                 struct rbd_image_snap_ondisk *snaps;
1019                 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
1020
1021                 /* We'll keep a copy of the snapshot names... */
1022
1023                 if (snap_names_len > (u64)SIZE_MAX)
1024                         goto out_2big;
1025                 snap_names = kmalloc(snap_names_len, GFP_KERNEL);
1026                 if (!snap_names)
1027                         goto out_err;
1028
1029                 /* ...as well as the array of their sizes. */
1030                 snap_sizes = kmalloc_array(snap_count,
1031                                            sizeof(*header->snap_sizes),
1032                                            GFP_KERNEL);
1033                 if (!snap_sizes)
1034                         goto out_err;
1035
1036                 /*
1037                  * Copy the names, and fill in each snapshot's id
1038                  * and size.
1039                  *
1040                  * Note that rbd_dev_v1_header_info() guarantees the
1041                  * ondisk buffer we're working with has
1042                  * snap_names_len bytes beyond the end of the
1043                  * snapshot id array, this memcpy() is safe.
1044                  */
1045                 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
1046                 snaps = ondisk->snaps;
1047                 for (i = 0; i < snap_count; i++) {
1048                         snapc->snaps[i] = le64_to_cpu(snaps[i].id);
1049                         snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
1050                 }
1051         }
1052
1053         /* We won't fail any more, fill in the header */
1054
1055         if (first_time) {
1056                 header->object_prefix = object_prefix;
1057                 header->obj_order = ondisk->options.order;
1058                 rbd_init_layout(rbd_dev);
1059         } else {
1060                 ceph_put_snap_context(header->snapc);
1061                 kfree(header->snap_names);
1062                 kfree(header->snap_sizes);
1063         }
1064
1065         /* The remaining fields always get updated (when we refresh) */
1066
1067         header->image_size = le64_to_cpu(ondisk->image_size);
1068         header->snapc = snapc;
1069         header->snap_names = snap_names;
1070         header->snap_sizes = snap_sizes;
1071
1072         return 0;
1073 out_2big:
1074         ret = -EIO;
1075 out_err:
1076         kfree(snap_sizes);
1077         kfree(snap_names);
1078         ceph_put_snap_context(snapc);
1079         kfree(object_prefix);
1080
1081         return ret;
1082 }
1083
1084 static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1085 {
1086         const char *snap_name;
1087
1088         rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1089
1090         /* Skip over names until we find the one we are looking for */
1091
1092         snap_name = rbd_dev->header.snap_names;
1093         while (which--)
1094                 snap_name += strlen(snap_name) + 1;
1095
1096         return kstrdup(snap_name, GFP_KERNEL);
1097 }
1098
1099 /*
1100  * Snapshot id comparison function for use with qsort()/bsearch().
1101  * Note that result is for snapshots in *descending* order.
1102  */
1103 static int snapid_compare_reverse(const void *s1, const void *s2)
1104 {
1105         u64 snap_id1 = *(u64 *)s1;
1106         u64 snap_id2 = *(u64 *)s2;
1107
1108         if (snap_id1 < snap_id2)
1109                 return 1;
1110         return snap_id1 == snap_id2 ? 0 : -1;
1111 }
1112
1113 /*
1114  * Search a snapshot context to see if the given snapshot id is
1115  * present.
1116  *
1117  * Returns the position of the snapshot id in the array if it's found,
1118  * or BAD_SNAP_INDEX otherwise.
1119  *
1120  * Note: The snapshot array is in kept sorted (by the osd) in
1121  * reverse order, highest snapshot id first.
1122  */
1123 static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1124 {
1125         struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1126         u64 *found;
1127
1128         found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1129                                 sizeof (snap_id), snapid_compare_reverse);
1130
1131         return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1132 }
1133
1134 static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1135                                         u64 snap_id)
1136 {
1137         u32 which;
1138         const char *snap_name;
1139
1140         which = rbd_dev_snap_index(rbd_dev, snap_id);
1141         if (which == BAD_SNAP_INDEX)
1142                 return ERR_PTR(-ENOENT);
1143
1144         snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1145         return snap_name ? snap_name : ERR_PTR(-ENOMEM);
1146 }
1147
1148 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1149 {
1150         if (snap_id == CEPH_NOSNAP)
1151                 return RBD_SNAP_HEAD_NAME;
1152
1153         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1154         if (rbd_dev->image_format == 1)
1155                 return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1156
1157         return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1158 }
1159
1160 static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1161                                 u64 *snap_size)
1162 {
1163         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1164         if (snap_id == CEPH_NOSNAP) {
1165                 *snap_size = rbd_dev->header.image_size;
1166         } else if (rbd_dev->image_format == 1) {
1167                 u32 which;
1168
1169                 which = rbd_dev_snap_index(rbd_dev, snap_id);
1170                 if (which == BAD_SNAP_INDEX)
1171                         return -ENOENT;
1172
1173                 *snap_size = rbd_dev->header.snap_sizes[which];
1174         } else {
1175                 u64 size = 0;
1176                 int ret;
1177
1178                 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1179                 if (ret)
1180                         return ret;
1181
1182                 *snap_size = size;
1183         }
1184         return 0;
1185 }
1186
1187 static int rbd_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
1188                         u64 *snap_features)
1189 {
1190         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1191         if (snap_id == CEPH_NOSNAP) {
1192                 *snap_features = rbd_dev->header.features;
1193         } else if (rbd_dev->image_format == 1) {
1194                 *snap_features = 0;     /* No features for format 1 */
1195         } else {
1196                 u64 features = 0;
1197                 int ret;
1198
1199                 ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, &features);
1200                 if (ret)
1201                         return ret;
1202
1203                 *snap_features = features;
1204         }
1205         return 0;
1206 }
1207
1208 static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1209 {
1210         u64 snap_id = rbd_dev->spec->snap_id;
1211         u64 size = 0;
1212         u64 features = 0;
1213         int ret;
1214
1215         ret = rbd_snap_size(rbd_dev, snap_id, &size);
1216         if (ret)
1217                 return ret;
1218         ret = rbd_snap_features(rbd_dev, snap_id, &features);
1219         if (ret)
1220                 return ret;
1221
1222         rbd_dev->mapping.size = size;
1223         rbd_dev->mapping.features = features;
1224
1225         return 0;
1226 }
1227
1228 static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1229 {
1230         rbd_dev->mapping.size = 0;
1231         rbd_dev->mapping.features = 0;
1232 }
1233
1234 static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset)
1235 {
1236         u64 segment_size = rbd_obj_bytes(&rbd_dev->header);
1237
1238         return offset & (segment_size - 1);
1239 }
1240
1241 static u64 rbd_segment_length(struct rbd_device *rbd_dev,
1242                                 u64 offset, u64 length)
1243 {
1244         u64 segment_size = rbd_obj_bytes(&rbd_dev->header);
1245
1246         offset &= segment_size - 1;
1247
1248         rbd_assert(length <= U64_MAX - offset);
1249         if (offset + length > segment_size)
1250                 length = segment_size - offset;
1251
1252         return length;
1253 }
1254
1255 /*
1256  * bio helpers
1257  */
1258
1259 static void bio_chain_put(struct bio *chain)
1260 {
1261         struct bio *tmp;
1262
1263         while (chain) {
1264                 tmp = chain;
1265                 chain = chain->bi_next;
1266                 bio_put(tmp);
1267         }
1268 }
1269
1270 /*
1271  * zeros a bio chain, starting at specific offset
1272  */
1273 static void zero_bio_chain(struct bio *chain, int start_ofs)
1274 {
1275         struct bio_vec bv;
1276         struct bvec_iter iter;
1277         unsigned long flags;
1278         void *buf;
1279         int pos = 0;
1280
1281         while (chain) {
1282                 bio_for_each_segment(bv, chain, iter) {
1283                         if (pos + bv.bv_len > start_ofs) {
1284                                 int remainder = max(start_ofs - pos, 0);
1285                                 buf = bvec_kmap_irq(&bv, &flags);
1286                                 memset(buf + remainder, 0,
1287                                        bv.bv_len - remainder);
1288                                 flush_dcache_page(bv.bv_page);
1289                                 bvec_kunmap_irq(buf, &flags);
1290                         }
1291                         pos += bv.bv_len;
1292                 }
1293
1294                 chain = chain->bi_next;
1295         }
1296 }
1297
1298 /*
1299  * similar to zero_bio_chain(), zeros data defined by a page array,
1300  * starting at the given byte offset from the start of the array and
1301  * continuing up to the given end offset.  The pages array is
1302  * assumed to be big enough to hold all bytes up to the end.
1303  */
1304 static void zero_pages(struct page **pages, u64 offset, u64 end)
1305 {
1306         struct page **page = &pages[offset >> PAGE_SHIFT];
1307
1308         rbd_assert(end > offset);
1309         rbd_assert(end - offset <= (u64)SIZE_MAX);
1310         while (offset < end) {
1311                 size_t page_offset;
1312                 size_t length;
1313                 unsigned long flags;
1314                 void *kaddr;
1315
1316                 page_offset = offset & ~PAGE_MASK;
1317                 length = min_t(size_t, PAGE_SIZE - page_offset, end - offset);
1318                 local_irq_save(flags);
1319                 kaddr = kmap_atomic(*page);
1320                 memset(kaddr + page_offset, 0, length);
1321                 flush_dcache_page(*page);
1322                 kunmap_atomic(kaddr);
1323                 local_irq_restore(flags);
1324
1325                 offset += length;
1326                 page++;
1327         }
1328 }
1329
1330 /*
1331  * Clone a portion of a bio, starting at the given byte offset
1332  * and continuing for the number of bytes indicated.
1333  */
1334 static struct bio *bio_clone_range(struct bio *bio_src,
1335                                         unsigned int offset,
1336                                         unsigned int len,
1337                                         gfp_t gfpmask)
1338 {
1339         struct bio *bio;
1340
1341         bio = bio_clone_fast(bio_src, gfpmask, rbd_bio_clone);
1342         if (!bio)
1343                 return NULL;    /* ENOMEM */
1344
1345         bio_advance(bio, offset);
1346         bio->bi_iter.bi_size = len;
1347
1348         return bio;
1349 }
1350
1351 /*
1352  * Clone a portion of a bio chain, starting at the given byte offset
1353  * into the first bio in the source chain and continuing for the
1354  * number of bytes indicated.  The result is another bio chain of
1355  * exactly the given length, or a null pointer on error.
1356  *
1357  * The bio_src and offset parameters are both in-out.  On entry they
1358  * refer to the first source bio and the offset into that bio where
1359  * the start of data to be cloned is located.
1360  *
1361  * On return, bio_src is updated to refer to the bio in the source
1362  * chain that contains first un-cloned byte, and *offset will
1363  * contain the offset of that byte within that bio.
1364  */
1365 static struct bio *bio_chain_clone_range(struct bio **bio_src,
1366                                         unsigned int *offset,
1367                                         unsigned int len,
1368                                         gfp_t gfpmask)
1369 {
1370         struct bio *bi = *bio_src;
1371         unsigned int off = *offset;
1372         struct bio *chain = NULL;
1373         struct bio **end;
1374
1375         /* Build up a chain of clone bios up to the limit */
1376
1377         if (!bi || off >= bi->bi_iter.bi_size || !len)
1378                 return NULL;            /* Nothing to clone */
1379
1380         end = &chain;
1381         while (len) {
1382                 unsigned int bi_size;
1383                 struct bio *bio;
1384
1385                 if (!bi) {
1386                         rbd_warn(NULL, "bio_chain exhausted with %u left", len);
1387                         goto out_err;   /* EINVAL; ran out of bio's */
1388                 }
1389                 bi_size = min_t(unsigned int, bi->bi_iter.bi_size - off, len);
1390                 bio = bio_clone_range(bi, off, bi_size, gfpmask);
1391                 if (!bio)
1392                         goto out_err;   /* ENOMEM */
1393
1394                 *end = bio;
1395                 end = &bio->bi_next;
1396
1397                 off += bi_size;
1398                 if (off == bi->bi_iter.bi_size) {
1399                         bi = bi->bi_next;
1400                         off = 0;
1401                 }
1402                 len -= bi_size;
1403         }
1404         *bio_src = bi;
1405         *offset = off;
1406
1407         return chain;
1408 out_err:
1409         bio_chain_put(chain);
1410
1411         return NULL;
1412 }
1413
1414 /*
1415  * The default/initial value for all object request flags is 0.  For
1416  * each flag, once its value is set to 1 it is never reset to 0
1417  * again.
1418  */
1419 static void obj_request_img_data_set(struct rbd_obj_request *obj_request)
1420 {
1421         if (test_and_set_bit(OBJ_REQ_IMG_DATA, &obj_request->flags)) {
1422                 struct rbd_device *rbd_dev;
1423
1424                 rbd_dev = obj_request->img_request->rbd_dev;
1425                 rbd_warn(rbd_dev, "obj_request %p already marked img_data",
1426                         obj_request);
1427         }
1428 }
1429
1430 static bool obj_request_img_data_test(struct rbd_obj_request *obj_request)
1431 {
1432         smp_mb();
1433         return test_bit(OBJ_REQ_IMG_DATA, &obj_request->flags) != 0;
1434 }
1435
1436 static void obj_request_done_set(struct rbd_obj_request *obj_request)
1437 {
1438         if (test_and_set_bit(OBJ_REQ_DONE, &obj_request->flags)) {
1439                 struct rbd_device *rbd_dev = NULL;
1440
1441                 if (obj_request_img_data_test(obj_request))
1442                         rbd_dev = obj_request->img_request->rbd_dev;
1443                 rbd_warn(rbd_dev, "obj_request %p already marked done",
1444                         obj_request);
1445         }
1446 }
1447
1448 static bool obj_request_done_test(struct rbd_obj_request *obj_request)
1449 {
1450         smp_mb();
1451         return test_bit(OBJ_REQ_DONE, &obj_request->flags) != 0;
1452 }
1453
1454 /*
1455  * This sets the KNOWN flag after (possibly) setting the EXISTS
1456  * flag.  The latter is set based on the "exists" value provided.
1457  *
1458  * Note that for our purposes once an object exists it never goes
1459  * away again.  It's possible that the response from two existence
1460  * checks are separated by the creation of the target object, and
1461  * the first ("doesn't exist") response arrives *after* the second
1462  * ("does exist").  In that case we ignore the second one.
1463  */
1464 static void obj_request_existence_set(struct rbd_obj_request *obj_request,
1465                                 bool exists)
1466 {
1467         if (exists)
1468                 set_bit(OBJ_REQ_EXISTS, &obj_request->flags);
1469         set_bit(OBJ_REQ_KNOWN, &obj_request->flags);
1470         smp_mb();
1471 }
1472
1473 static bool obj_request_known_test(struct rbd_obj_request *obj_request)
1474 {
1475         smp_mb();
1476         return test_bit(OBJ_REQ_KNOWN, &obj_request->flags) != 0;
1477 }
1478
1479 static bool obj_request_exists_test(struct rbd_obj_request *obj_request)
1480 {
1481         smp_mb();
1482         return test_bit(OBJ_REQ_EXISTS, &obj_request->flags) != 0;
1483 }
1484
1485 static bool obj_request_overlaps_parent(struct rbd_obj_request *obj_request)
1486 {
1487         struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1488
1489         return obj_request->img_offset <
1490             round_up(rbd_dev->parent_overlap, rbd_obj_bytes(&rbd_dev->header));
1491 }
1492
1493 static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
1494 {
1495         dout("%s: obj %p (was %d)\n", __func__, obj_request,
1496                 kref_read(&obj_request->kref));
1497         kref_get(&obj_request->kref);
1498 }
1499
1500 static void rbd_obj_request_destroy(struct kref *kref);
1501 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1502 {
1503         rbd_assert(obj_request != NULL);
1504         dout("%s: obj %p (was %d)\n", __func__, obj_request,
1505                 kref_read(&obj_request->kref));
1506         kref_put(&obj_request->kref, rbd_obj_request_destroy);
1507 }
1508
1509 static void rbd_img_request_get(struct rbd_img_request *img_request)
1510 {
1511         dout("%s: img %p (was %d)\n", __func__, img_request,
1512              kref_read(&img_request->kref));
1513         kref_get(&img_request->kref);
1514 }
1515
1516 static bool img_request_child_test(struct rbd_img_request *img_request);
1517 static void rbd_parent_request_destroy(struct kref *kref);
1518 static void rbd_img_request_destroy(struct kref *kref);
1519 static void rbd_img_request_put(struct rbd_img_request *img_request)
1520 {
1521         rbd_assert(img_request != NULL);
1522         dout("%s: img %p (was %d)\n", __func__, img_request,
1523                 kref_read(&img_request->kref));
1524         if (img_request_child_test(img_request))
1525                 kref_put(&img_request->kref, rbd_parent_request_destroy);
1526         else
1527                 kref_put(&img_request->kref, rbd_img_request_destroy);
1528 }
1529
1530 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1531                                         struct rbd_obj_request *obj_request)
1532 {
1533         rbd_assert(obj_request->img_request == NULL);
1534
1535         /* Image request now owns object's original reference */
1536         obj_request->img_request = img_request;
1537         obj_request->which = img_request->obj_request_count;
1538         rbd_assert(!obj_request_img_data_test(obj_request));
1539         obj_request_img_data_set(obj_request);
1540         rbd_assert(obj_request->which != BAD_WHICH);
1541         img_request->obj_request_count++;
1542         list_add_tail(&obj_request->links, &img_request->obj_requests);
1543         dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1544                 obj_request->which);
1545 }
1546
1547 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1548                                         struct rbd_obj_request *obj_request)
1549 {
1550         rbd_assert(obj_request->which != BAD_WHICH);
1551
1552         dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1553                 obj_request->which);
1554         list_del(&obj_request->links);
1555         rbd_assert(img_request->obj_request_count > 0);
1556         img_request->obj_request_count--;
1557         rbd_assert(obj_request->which == img_request->obj_request_count);
1558         obj_request->which = BAD_WHICH;
1559         rbd_assert(obj_request_img_data_test(obj_request));
1560         rbd_assert(obj_request->img_request == img_request);
1561         obj_request->img_request = NULL;
1562         obj_request->callback = NULL;
1563         rbd_obj_request_put(obj_request);
1564 }
1565
1566 static bool obj_request_type_valid(enum obj_request_type type)
1567 {
1568         switch (type) {
1569         case OBJ_REQUEST_NODATA:
1570         case OBJ_REQUEST_BIO:
1571         case OBJ_REQUEST_PAGES:
1572                 return true;
1573         default:
1574                 return false;
1575         }
1576 }
1577
1578 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request);
1579
1580 static void rbd_obj_request_submit(struct rbd_obj_request *obj_request)
1581 {
1582         struct ceph_osd_request *osd_req = obj_request->osd_req;
1583
1584         dout("%s %p object_no %016llx %llu~%llu osd_req %p\n", __func__,
1585              obj_request, obj_request->object_no, obj_request->offset,
1586              obj_request->length, osd_req);
1587         if (obj_request_img_data_test(obj_request)) {
1588                 WARN_ON(obj_request->callback != rbd_img_obj_callback);
1589                 rbd_img_request_get(obj_request->img_request);
1590         }
1591         ceph_osdc_start_request(osd_req->r_osdc, osd_req, false);
1592 }
1593
1594 static void rbd_img_request_complete(struct rbd_img_request *img_request)
1595 {
1596
1597         dout("%s: img %p\n", __func__, img_request);
1598
1599         /*
1600          * If no error occurred, compute the aggregate transfer
1601          * count for the image request.  We could instead use
1602          * atomic64_cmpxchg() to update it as each object request
1603          * completes; not clear which way is better off hand.
1604          */
1605         if (!img_request->result) {
1606                 struct rbd_obj_request *obj_request;
1607                 u64 xferred = 0;
1608
1609                 for_each_obj_request(img_request, obj_request)
1610                         xferred += obj_request->xferred;
1611                 img_request->xferred = xferred;
1612         }
1613
1614         if (img_request->callback)
1615                 img_request->callback(img_request);
1616         else
1617                 rbd_img_request_put(img_request);
1618 }
1619
1620 /*
1621  * The default/initial value for all image request flags is 0.  Each
1622  * is conditionally set to 1 at image request initialization time
1623  * and currently never change thereafter.
1624  */
1625 static void img_request_write_set(struct rbd_img_request *img_request)
1626 {
1627         set_bit(IMG_REQ_WRITE, &img_request->flags);
1628         smp_mb();
1629 }
1630
1631 static bool img_request_write_test(struct rbd_img_request *img_request)
1632 {
1633         smp_mb();
1634         return test_bit(IMG_REQ_WRITE, &img_request->flags) != 0;
1635 }
1636
1637 /*
1638  * Set the discard flag when the img_request is an discard request
1639  */
1640 static void img_request_discard_set(struct rbd_img_request *img_request)
1641 {
1642         set_bit(IMG_REQ_DISCARD, &img_request->flags);
1643         smp_mb();
1644 }
1645
1646 static bool img_request_discard_test(struct rbd_img_request *img_request)
1647 {
1648         smp_mb();
1649         return test_bit(IMG_REQ_DISCARD, &img_request->flags) != 0;
1650 }
1651
1652 static void img_request_child_set(struct rbd_img_request *img_request)
1653 {
1654         set_bit(IMG_REQ_CHILD, &img_request->flags);
1655         smp_mb();
1656 }
1657
1658 static void img_request_child_clear(struct rbd_img_request *img_request)
1659 {
1660         clear_bit(IMG_REQ_CHILD, &img_request->flags);
1661         smp_mb();
1662 }
1663
1664 static bool img_request_child_test(struct rbd_img_request *img_request)
1665 {
1666         smp_mb();
1667         return test_bit(IMG_REQ_CHILD, &img_request->flags) != 0;
1668 }
1669
1670 static void img_request_layered_set(struct rbd_img_request *img_request)
1671 {
1672         set_bit(IMG_REQ_LAYERED, &img_request->flags);
1673         smp_mb();
1674 }
1675
1676 static void img_request_layered_clear(struct rbd_img_request *img_request)
1677 {
1678         clear_bit(IMG_REQ_LAYERED, &img_request->flags);
1679         smp_mb();
1680 }
1681
1682 static bool img_request_layered_test(struct rbd_img_request *img_request)
1683 {
1684         smp_mb();
1685         return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1686 }
1687
1688 static enum obj_operation_type
1689 rbd_img_request_op_type(struct rbd_img_request *img_request)
1690 {
1691         if (img_request_write_test(img_request))
1692                 return OBJ_OP_WRITE;
1693         else if (img_request_discard_test(img_request))
1694                 return OBJ_OP_DISCARD;
1695         else
1696                 return OBJ_OP_READ;
1697 }
1698
1699 static void
1700 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
1701 {
1702         u64 xferred = obj_request->xferred;
1703         u64 length = obj_request->length;
1704
1705         dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1706                 obj_request, obj_request->img_request, obj_request->result,
1707                 xferred, length);
1708         /*
1709          * ENOENT means a hole in the image.  We zero-fill the entire
1710          * length of the request.  A short read also implies zero-fill
1711          * to the end of the request.  An error requires the whole
1712          * length of the request to be reported finished with an error
1713          * to the block layer.  In each case we update the xferred
1714          * count to indicate the whole request was satisfied.
1715          */
1716         rbd_assert(obj_request->type != OBJ_REQUEST_NODATA);
1717         if (obj_request->result == -ENOENT) {
1718                 if (obj_request->type == OBJ_REQUEST_BIO)
1719                         zero_bio_chain(obj_request->bio_list, 0);
1720                 else
1721                         zero_pages(obj_request->pages, 0, length);
1722                 obj_request->result = 0;
1723         } else if (xferred < length && !obj_request->result) {
1724                 if (obj_request->type == OBJ_REQUEST_BIO)
1725                         zero_bio_chain(obj_request->bio_list, xferred);
1726                 else
1727                         zero_pages(obj_request->pages, xferred, length);
1728         }
1729         obj_request->xferred = length;
1730         obj_request_done_set(obj_request);
1731 }
1732
1733 static void rbd_obj_request_complete(struct rbd_obj_request *obj_request)
1734 {
1735         dout("%s: obj %p cb %p\n", __func__, obj_request,
1736                 obj_request->callback);
1737         if (obj_request->callback)
1738                 obj_request->callback(obj_request);
1739         else
1740                 complete_all(&obj_request->completion);
1741 }
1742
1743 static void rbd_obj_request_error(struct rbd_obj_request *obj_request, int err)
1744 {
1745         obj_request->result = err;
1746         obj_request->xferred = 0;
1747         /*
1748          * kludge - mirror rbd_obj_request_submit() to match a put in
1749          * rbd_img_obj_callback()
1750          */
1751         if (obj_request_img_data_test(obj_request)) {
1752                 WARN_ON(obj_request->callback != rbd_img_obj_callback);
1753                 rbd_img_request_get(obj_request->img_request);
1754         }
1755         obj_request_done_set(obj_request);
1756         rbd_obj_request_complete(obj_request);
1757 }
1758
1759 static void rbd_osd_read_callback(struct rbd_obj_request *obj_request)
1760 {
1761         struct rbd_img_request *img_request = NULL;
1762         struct rbd_device *rbd_dev = NULL;
1763         bool layered = false;
1764
1765         if (obj_request_img_data_test(obj_request)) {
1766                 img_request = obj_request->img_request;
1767                 layered = img_request && img_request_layered_test(img_request);
1768                 rbd_dev = img_request->rbd_dev;
1769         }
1770
1771         dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1772                 obj_request, img_request, obj_request->result,
1773                 obj_request->xferred, obj_request->length);
1774         if (layered && obj_request->result == -ENOENT &&
1775                         obj_request->img_offset < rbd_dev->parent_overlap)
1776                 rbd_img_parent_read(obj_request);
1777         else if (img_request)
1778                 rbd_img_obj_request_read_callback(obj_request);
1779         else
1780                 obj_request_done_set(obj_request);
1781 }
1782
1783 static void rbd_osd_write_callback(struct rbd_obj_request *obj_request)
1784 {
1785         dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1786                 obj_request->result, obj_request->length);
1787         /*
1788          * There is no such thing as a successful short write.  Set
1789          * it to our originally-requested length.
1790          */
1791         obj_request->xferred = obj_request->length;
1792         obj_request_done_set(obj_request);
1793 }
1794
1795 static void rbd_osd_discard_callback(struct rbd_obj_request *obj_request)
1796 {
1797         dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1798                 obj_request->result, obj_request->length);
1799         /*
1800          * There is no such thing as a successful short discard.  Set
1801          * it to our originally-requested length.
1802          */
1803         obj_request->xferred = obj_request->length;
1804         /* discarding a non-existent object is not a problem */
1805         if (obj_request->result == -ENOENT)
1806                 obj_request->result = 0;
1807         obj_request_done_set(obj_request);
1808 }
1809
1810 /*
1811  * For a simple stat call there's nothing to do.  We'll do more if
1812  * this is part of a write sequence for a layered image.
1813  */
1814 static void rbd_osd_stat_callback(struct rbd_obj_request *obj_request)
1815 {
1816         dout("%s: obj %p\n", __func__, obj_request);
1817         obj_request_done_set(obj_request);
1818 }
1819
1820 static void rbd_osd_call_callback(struct rbd_obj_request *obj_request)
1821 {
1822         dout("%s: obj %p\n", __func__, obj_request);
1823
1824         if (obj_request_img_data_test(obj_request))
1825                 rbd_osd_copyup_callback(obj_request);
1826         else
1827                 obj_request_done_set(obj_request);
1828 }
1829
1830 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
1831 {
1832         struct rbd_obj_request *obj_request = osd_req->r_priv;
1833         u16 opcode;
1834
1835         dout("%s: osd_req %p\n", __func__, osd_req);
1836         rbd_assert(osd_req == obj_request->osd_req);
1837         if (obj_request_img_data_test(obj_request)) {
1838                 rbd_assert(obj_request->img_request);
1839                 rbd_assert(obj_request->which != BAD_WHICH);
1840         } else {
1841                 rbd_assert(obj_request->which == BAD_WHICH);
1842         }
1843
1844         if (osd_req->r_result < 0)
1845                 obj_request->result = osd_req->r_result;
1846
1847         /*
1848          * We support a 64-bit length, but ultimately it has to be
1849          * passed to the block layer, which just supports a 32-bit
1850          * length field.
1851          */
1852         obj_request->xferred = osd_req->r_ops[0].outdata_len;
1853         rbd_assert(obj_request->xferred < (u64)UINT_MAX);
1854
1855         opcode = osd_req->r_ops[0].op;
1856         switch (opcode) {
1857         case CEPH_OSD_OP_READ:
1858                 rbd_osd_read_callback(obj_request);
1859                 break;
1860         case CEPH_OSD_OP_SETALLOCHINT:
1861                 rbd_assert(osd_req->r_ops[1].op == CEPH_OSD_OP_WRITE ||
1862                            osd_req->r_ops[1].op == CEPH_OSD_OP_WRITEFULL);
1863                 /* fall through */
1864         case CEPH_OSD_OP_WRITE:
1865         case CEPH_OSD_OP_WRITEFULL:
1866                 rbd_osd_write_callback(obj_request);
1867                 break;
1868         case CEPH_OSD_OP_STAT:
1869                 rbd_osd_stat_callback(obj_request);
1870                 break;
1871         case CEPH_OSD_OP_DELETE:
1872         case CEPH_OSD_OP_TRUNCATE:
1873         case CEPH_OSD_OP_ZERO:
1874                 rbd_osd_discard_callback(obj_request);
1875                 break;
1876         case CEPH_OSD_OP_CALL:
1877                 rbd_osd_call_callback(obj_request);
1878                 break;
1879         default:
1880                 rbd_warn(NULL, "unexpected OSD op: object_no %016llx opcode %d",
1881                          obj_request->object_no, opcode);
1882                 break;
1883         }
1884
1885         if (obj_request_done_test(obj_request))
1886                 rbd_obj_request_complete(obj_request);
1887 }
1888
1889 static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request)
1890 {
1891         struct ceph_osd_request *osd_req = obj_request->osd_req;
1892
1893         rbd_assert(obj_request_img_data_test(obj_request));
1894         osd_req->r_snapid = obj_request->img_request->snap_id;
1895 }
1896
1897 static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request)
1898 {
1899         struct ceph_osd_request *osd_req = obj_request->osd_req;
1900
1901         ktime_get_real_ts(&osd_req->r_mtime);
1902         osd_req->r_data_offset = obj_request->offset;
1903 }
1904
1905 static struct ceph_osd_request *
1906 __rbd_osd_req_create(struct rbd_device *rbd_dev,
1907                      struct ceph_snap_context *snapc,
1908                      int num_ops, unsigned int flags,
1909                      struct rbd_obj_request *obj_request)
1910 {
1911         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1912         struct ceph_osd_request *req;
1913         const char *name_format = rbd_dev->image_format == 1 ?
1914                                       RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
1915
1916         req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, GFP_NOIO);
1917         if (!req)
1918                 return NULL;
1919
1920         req->r_flags = flags;
1921         req->r_callback = rbd_osd_req_callback;
1922         req->r_priv = obj_request;
1923
1924         req->r_base_oloc.pool = rbd_dev->layout.pool_id;
1925         if (ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
1926                         rbd_dev->header.object_prefix, obj_request->object_no))
1927                 goto err_req;
1928
1929         if (ceph_osdc_alloc_messages(req, GFP_NOIO))
1930                 goto err_req;
1931
1932         return req;
1933
1934 err_req:
1935         ceph_osdc_put_request(req);
1936         return NULL;
1937 }
1938
1939 /*
1940  * Create an osd request.  A read request has one osd op (read).
1941  * A write request has either one (watch) or two (hint+write) osd ops.
1942  * (All rbd data writes are prefixed with an allocation hint op, but
1943  * technically osd watch is a write request, hence this distinction.)
1944  */
1945 static struct ceph_osd_request *rbd_osd_req_create(
1946                                         struct rbd_device *rbd_dev,
1947                                         enum obj_operation_type op_type,
1948                                         unsigned int num_ops,
1949                                         struct rbd_obj_request *obj_request)
1950 {
1951         struct ceph_snap_context *snapc = NULL;
1952
1953         if (obj_request_img_data_test(obj_request) &&
1954                 (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_WRITE)) {
1955                 struct rbd_img_request *img_request = obj_request->img_request;
1956                 if (op_type == OBJ_OP_WRITE) {
1957                         rbd_assert(img_request_write_test(img_request));
1958                 } else {
1959                         rbd_assert(img_request_discard_test(img_request));
1960                 }
1961                 snapc = img_request->snapc;
1962         }
1963
1964         rbd_assert(num_ops == 1 || ((op_type == OBJ_OP_WRITE) && num_ops == 2));
1965
1966         return __rbd_osd_req_create(rbd_dev, snapc, num_ops,
1967             (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD) ?
1968             CEPH_OSD_FLAG_WRITE : CEPH_OSD_FLAG_READ, obj_request);
1969 }
1970
1971 /*
1972  * Create a copyup osd request based on the information in the object
1973  * request supplied.  A copyup request has two or three osd ops, a
1974  * copyup method call, potentially a hint op, and a write or truncate
1975  * or zero op.
1976  */
1977 static struct ceph_osd_request *
1978 rbd_osd_req_create_copyup(struct rbd_obj_request *obj_request)
1979 {
1980         struct rbd_img_request *img_request;
1981         int num_osd_ops = 3;
1982
1983         rbd_assert(obj_request_img_data_test(obj_request));
1984         img_request = obj_request->img_request;
1985         rbd_assert(img_request);
1986         rbd_assert(img_request_write_test(img_request) ||
1987                         img_request_discard_test(img_request));
1988
1989         if (img_request_discard_test(img_request))
1990                 num_osd_ops = 2;
1991
1992         return __rbd_osd_req_create(img_request->rbd_dev,
1993                                     img_request->snapc, num_osd_ops,
1994                                     CEPH_OSD_FLAG_WRITE, obj_request);
1995 }
1996
1997 static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req)
1998 {
1999         ceph_osdc_put_request(osd_req);
2000 }
2001
2002 static struct rbd_obj_request *
2003 rbd_obj_request_create(enum obj_request_type type)
2004 {
2005         struct rbd_obj_request *obj_request;
2006
2007         rbd_assert(obj_request_type_valid(type));
2008
2009         obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_NOIO);
2010         if (!obj_request)
2011                 return NULL;
2012
2013         obj_request->which = BAD_WHICH;
2014         obj_request->type = type;
2015         INIT_LIST_HEAD(&obj_request->links);
2016         init_completion(&obj_request->completion);
2017         kref_init(&obj_request->kref);
2018
2019         dout("%s %p\n", __func__, obj_request);
2020         return obj_request;
2021 }
2022
2023 static void rbd_obj_request_destroy(struct kref *kref)
2024 {
2025         struct rbd_obj_request *obj_request;
2026
2027         obj_request = container_of(kref, struct rbd_obj_request, kref);
2028
2029         dout("%s: obj %p\n", __func__, obj_request);
2030
2031         rbd_assert(obj_request->img_request == NULL);
2032         rbd_assert(obj_request->which == BAD_WHICH);
2033
2034         if (obj_request->osd_req)
2035                 rbd_osd_req_destroy(obj_request->osd_req);
2036
2037         rbd_assert(obj_request_type_valid(obj_request->type));
2038         switch (obj_request->type) {
2039         case OBJ_REQUEST_NODATA:
2040                 break;          /* Nothing to do */
2041         case OBJ_REQUEST_BIO:
2042                 if (obj_request->bio_list)
2043                         bio_chain_put(obj_request->bio_list);
2044                 break;
2045         case OBJ_REQUEST_PAGES:
2046                 /* img_data requests don't own their page array */
2047                 if (obj_request->pages &&
2048                     !obj_request_img_data_test(obj_request))
2049                         ceph_release_page_vector(obj_request->pages,
2050                                                 obj_request->page_count);
2051                 break;
2052         }
2053
2054         kmem_cache_free(rbd_obj_request_cache, obj_request);
2055 }
2056
2057 /* It's OK to call this for a device with no parent */
2058
2059 static void rbd_spec_put(struct rbd_spec *spec);
2060 static void rbd_dev_unparent(struct rbd_device *rbd_dev)
2061 {
2062         rbd_dev_remove_parent(rbd_dev);
2063         rbd_spec_put(rbd_dev->parent_spec);
2064         rbd_dev->parent_spec = NULL;
2065         rbd_dev->parent_overlap = 0;
2066 }
2067
2068 /*
2069  * Parent image reference counting is used to determine when an
2070  * image's parent fields can be safely torn down--after there are no
2071  * more in-flight requests to the parent image.  When the last
2072  * reference is dropped, cleaning them up is safe.
2073  */
2074 static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
2075 {
2076         int counter;
2077
2078         if (!rbd_dev->parent_spec)
2079                 return;
2080
2081         counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
2082         if (counter > 0)
2083                 return;
2084
2085         /* Last reference; clean up parent data structures */
2086
2087         if (!counter)
2088                 rbd_dev_unparent(rbd_dev);
2089         else
2090                 rbd_warn(rbd_dev, "parent reference underflow");
2091 }
2092
2093 /*
2094  * If an image has a non-zero parent overlap, get a reference to its
2095  * parent.
2096  *
2097  * Returns true if the rbd device has a parent with a non-zero
2098  * overlap and a reference for it was successfully taken, or
2099  * false otherwise.
2100  */
2101 static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
2102 {
2103         int counter = 0;
2104
2105         if (!rbd_dev->parent_spec)
2106                 return false;
2107
2108         down_read(&rbd_dev->header_rwsem);
2109         if (rbd_dev->parent_overlap)
2110                 counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
2111         up_read(&rbd_dev->header_rwsem);
2112
2113         if (counter < 0)
2114                 rbd_warn(rbd_dev, "parent reference overflow");
2115
2116         return counter > 0;
2117 }
2118
2119 /*
2120  * Caller is responsible for filling in the list of object requests
2121  * that comprises the image request, and the Linux request pointer
2122  * (if there is one).
2123  */
2124 static struct rbd_img_request *rbd_img_request_create(
2125                                         struct rbd_device *rbd_dev,
2126                                         u64 offset, u64 length,
2127                                         enum obj_operation_type op_type,
2128                                         struct ceph_snap_context *snapc)
2129 {
2130         struct rbd_img_request *img_request;
2131
2132         img_request = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
2133         if (!img_request)
2134                 return NULL;
2135
2136         img_request->rq = NULL;
2137         img_request->rbd_dev = rbd_dev;
2138         img_request->offset = offset;
2139         img_request->length = length;
2140         img_request->flags = 0;
2141         if (op_type == OBJ_OP_DISCARD) {
2142                 img_request_discard_set(img_request);
2143                 img_request->snapc = snapc;
2144         } else if (op_type == OBJ_OP_WRITE) {
2145                 img_request_write_set(img_request);
2146                 img_request->snapc = snapc;
2147         } else {
2148                 img_request->snap_id = rbd_dev->spec->snap_id;
2149         }
2150         if (rbd_dev_parent_get(rbd_dev))
2151                 img_request_layered_set(img_request);
2152         spin_lock_init(&img_request->completion_lock);
2153         img_request->next_completion = 0;
2154         img_request->callback = NULL;
2155         img_request->result = 0;
2156         img_request->obj_request_count = 0;
2157         INIT_LIST_HEAD(&img_request->obj_requests);
2158         kref_init(&img_request->kref);
2159
2160         dout("%s: rbd_dev %p %s %llu/%llu -> img %p\n", __func__, rbd_dev,
2161                 obj_op_name(op_type), offset, length, img_request);
2162
2163         return img_request;
2164 }
2165
2166 static void rbd_img_request_destroy(struct kref *kref)
2167 {
2168         struct rbd_img_request *img_request;
2169         struct rbd_obj_request *obj_request;
2170         struct rbd_obj_request *next_obj_request;
2171
2172         img_request = container_of(kref, struct rbd_img_request, kref);
2173
2174         dout("%s: img %p\n", __func__, img_request);
2175
2176         for_each_obj_request_safe(img_request, obj_request, next_obj_request)
2177                 rbd_img_obj_request_del(img_request, obj_request);
2178         rbd_assert(img_request->obj_request_count == 0);
2179
2180         if (img_request_layered_test(img_request)) {
2181                 img_request_layered_clear(img_request);
2182                 rbd_dev_parent_put(img_request->rbd_dev);
2183         }
2184
2185         if (img_request_write_test(img_request) ||
2186                 img_request_discard_test(img_request))
2187                 ceph_put_snap_context(img_request->snapc);
2188
2189         kmem_cache_free(rbd_img_request_cache, img_request);
2190 }
2191
2192 static struct rbd_img_request *rbd_parent_request_create(
2193                                         struct rbd_obj_request *obj_request,
2194                                         u64 img_offset, u64 length)
2195 {
2196         struct rbd_img_request *parent_request;
2197         struct rbd_device *rbd_dev;
2198
2199         rbd_assert(obj_request->img_request);
2200         rbd_dev = obj_request->img_request->rbd_dev;
2201
2202         parent_request = rbd_img_request_create(rbd_dev->parent, img_offset,
2203                                                 length, OBJ_OP_READ, NULL);
2204         if (!parent_request)
2205                 return NULL;
2206
2207         img_request_child_set(parent_request);
2208         rbd_obj_request_get(obj_request);
2209         parent_request->obj_request = obj_request;
2210
2211         return parent_request;
2212 }
2213
2214 static void rbd_parent_request_destroy(struct kref *kref)
2215 {
2216         struct rbd_img_request *parent_request;
2217         struct rbd_obj_request *orig_request;
2218
2219         parent_request = container_of(kref, struct rbd_img_request, kref);
2220         orig_request = parent_request->obj_request;
2221
2222         parent_request->obj_request = NULL;
2223         rbd_obj_request_put(orig_request);
2224         img_request_child_clear(parent_request);
2225
2226         rbd_img_request_destroy(kref);
2227 }
2228
2229 static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request)
2230 {
2231         struct rbd_img_request *img_request;
2232         unsigned int xferred;
2233         int result;
2234         bool more;
2235
2236         rbd_assert(obj_request_img_data_test(obj_request));
2237         img_request = obj_request->img_request;
2238
2239         rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
2240         xferred = (unsigned int)obj_request->xferred;
2241         result = obj_request->result;
2242         if (result) {
2243                 struct rbd_device *rbd_dev = img_request->rbd_dev;
2244                 enum obj_operation_type op_type;
2245
2246                 if (img_request_discard_test(img_request))
2247                         op_type = OBJ_OP_DISCARD;
2248                 else if (img_request_write_test(img_request))
2249                         op_type = OBJ_OP_WRITE;
2250                 else
2251                         op_type = OBJ_OP_READ;
2252
2253                 rbd_warn(rbd_dev, "%s %llx at %llx (%llx)",
2254                         obj_op_name(op_type), obj_request->length,
2255                         obj_request->img_offset, obj_request->offset);
2256                 rbd_warn(rbd_dev, "  result %d xferred %x",
2257                         result, xferred);
2258                 if (!img_request->result)
2259                         img_request->result = result;
2260                 /*
2261                  * Need to end I/O on the entire obj_request worth of
2262                  * bytes in case of error.
2263                  */
2264                 xferred = obj_request->length;
2265         }
2266
2267         if (img_request_child_test(img_request)) {
2268                 rbd_assert(img_request->obj_request != NULL);
2269                 more = obj_request->which < img_request->obj_request_count - 1;
2270         } else {
2271                 blk_status_t status = errno_to_blk_status(result);
2272
2273                 rbd_assert(img_request->rq != NULL);
2274
2275                 more = blk_update_request(img_request->rq, status, xferred);
2276                 if (!more)
2277                         __blk_mq_end_request(img_request->rq, status);
2278         }
2279
2280         return more;
2281 }
2282
2283 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request)
2284 {
2285         struct rbd_img_request *img_request;
2286         u32 which = obj_request->which;
2287         bool more = true;
2288
2289         rbd_assert(obj_request_img_data_test(obj_request));
2290         img_request = obj_request->img_request;
2291
2292         dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
2293         rbd_assert(img_request != NULL);
2294         rbd_assert(img_request->obj_request_count > 0);
2295         rbd_assert(which != BAD_WHICH);
2296         rbd_assert(which < img_request->obj_request_count);
2297
2298         spin_lock_irq(&img_request->completion_lock);
2299         if (which != img_request->next_completion)
2300                 goto out;
2301
2302         for_each_obj_request_from(img_request, obj_request) {
2303                 rbd_assert(more);
2304                 rbd_assert(which < img_request->obj_request_count);
2305
2306                 if (!obj_request_done_test(obj_request))
2307                         break;
2308                 more = rbd_img_obj_end_request(obj_request);
2309                 which++;
2310         }
2311
2312         rbd_assert(more ^ (which == img_request->obj_request_count));
2313         img_request->next_completion = which;
2314 out:
2315         spin_unlock_irq(&img_request->completion_lock);
2316         rbd_img_request_put(img_request);
2317
2318         if (!more)
2319                 rbd_img_request_complete(img_request);
2320 }
2321
2322 /*
2323  * Add individual osd ops to the given ceph_osd_request and prepare
2324  * them for submission. num_ops is the current number of
2325  * osd operations already to the object request.
2326  */
2327 static void rbd_img_obj_request_fill(struct rbd_obj_request *obj_request,
2328                                 struct ceph_osd_request *osd_request,
2329                                 enum obj_operation_type op_type,
2330                                 unsigned int num_ops)
2331 {
2332         struct rbd_img_request *img_request = obj_request->img_request;
2333         struct rbd_device *rbd_dev = img_request->rbd_dev;
2334         u64 object_size = rbd_obj_bytes(&rbd_dev->header);
2335         u64 offset = obj_request->offset;
2336         u64 length = obj_request->length;
2337         u64 img_end;
2338         u16 opcode;
2339
2340         if (op_type == OBJ_OP_DISCARD) {
2341                 if (!offset && length == object_size &&
2342                     (!img_request_layered_test(img_request) ||
2343                      !obj_request_overlaps_parent(obj_request))) {
2344                         opcode = CEPH_OSD_OP_DELETE;
2345                 } else if ((offset + length == object_size)) {
2346                         opcode = CEPH_OSD_OP_TRUNCATE;
2347                 } else {
2348                         down_read(&rbd_dev->header_rwsem);
2349                         img_end = rbd_dev->header.image_size;
2350                         up_read(&rbd_dev->header_rwsem);
2351
2352                         if (obj_request->img_offset + length == img_end)
2353                                 opcode = CEPH_OSD_OP_TRUNCATE;
2354                         else
2355                                 opcode = CEPH_OSD_OP_ZERO;
2356                 }
2357         } else if (op_type == OBJ_OP_WRITE) {
2358                 if (!offset && length == object_size)
2359                         opcode = CEPH_OSD_OP_WRITEFULL;
2360                 else
2361                         opcode = CEPH_OSD_OP_WRITE;
2362                 osd_req_op_alloc_hint_init(osd_request, num_ops,
2363                                         object_size, object_size);
2364                 num_ops++;
2365         } else {
2366                 opcode = CEPH_OSD_OP_READ;
2367         }
2368
2369         if (opcode == CEPH_OSD_OP_DELETE)
2370                 osd_req_op_init(osd_request, num_ops, opcode, 0);
2371         else
2372                 osd_req_op_extent_init(osd_request, num_ops, opcode,
2373                                        offset, length, 0, 0);
2374
2375         if (obj_request->type == OBJ_REQUEST_BIO)
2376                 osd_req_op_extent_osd_data_bio(osd_request, num_ops,
2377                                         obj_request->bio_list, length);
2378         else if (obj_request->type == OBJ_REQUEST_PAGES)
2379                 osd_req_op_extent_osd_data_pages(osd_request, num_ops,
2380                                         obj_request->pages, length,
2381                                         offset & ~PAGE_MASK, false, false);
2382
2383         /* Discards are also writes */
2384         if (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD)
2385                 rbd_osd_req_format_write(obj_request);
2386         else
2387                 rbd_osd_req_format_read(obj_request);
2388 }
2389
2390 /*
2391  * Split up an image request into one or more object requests, each
2392  * to a different object.  The "type" parameter indicates whether
2393  * "data_desc" is the pointer to the head of a list of bio
2394  * structures, or the base of a page array.  In either case this
2395  * function assumes data_desc describes memory sufficient to hold
2396  * all data described by the image request.
2397  */
2398 static int rbd_img_request_fill(struct rbd_img_request *img_request,
2399                                         enum obj_request_type type,
2400                                         void *data_desc)
2401 {
2402         struct rbd_device *rbd_dev = img_request->rbd_dev;
2403         struct rbd_obj_request *obj_request = NULL;
2404         struct rbd_obj_request *next_obj_request;
2405         struct bio *bio_list = NULL;
2406         unsigned int bio_offset = 0;
2407         struct page **pages = NULL;
2408         enum obj_operation_type op_type;
2409         u64 img_offset;
2410         u64 resid;
2411
2412         dout("%s: img %p type %d data_desc %p\n", __func__, img_request,
2413                 (int)type, data_desc);
2414
2415         img_offset = img_request->offset;
2416         resid = img_request->length;
2417         rbd_assert(resid > 0);
2418         op_type = rbd_img_request_op_type(img_request);
2419
2420         if (type == OBJ_REQUEST_BIO) {
2421                 bio_list = data_desc;
2422                 rbd_assert(img_offset ==
2423                            bio_list->bi_iter.bi_sector << SECTOR_SHIFT);
2424         } else if (type == OBJ_REQUEST_PAGES) {
2425                 pages = data_desc;
2426         }
2427
2428         while (resid) {
2429                 struct ceph_osd_request *osd_req;
2430                 u64 object_no = img_offset >> rbd_dev->header.obj_order;
2431                 u64 offset = rbd_segment_offset(rbd_dev, img_offset);
2432                 u64 length = rbd_segment_length(rbd_dev, img_offset, resid);
2433
2434                 obj_request = rbd_obj_request_create(type);
2435                 if (!obj_request)
2436                         goto out_unwind;
2437
2438                 obj_request->object_no = object_no;
2439                 obj_request->offset = offset;
2440                 obj_request->length = length;
2441
2442                 /*
2443                  * set obj_request->img_request before creating the
2444                  * osd_request so that it gets the right snapc
2445                  */
2446                 rbd_img_obj_request_add(img_request, obj_request);
2447
2448                 if (type == OBJ_REQUEST_BIO) {
2449                         unsigned int clone_size;
2450
2451                         rbd_assert(length <= (u64)UINT_MAX);
2452                         clone_size = (unsigned int)length;
2453                         obj_request->bio_list =
2454                                         bio_chain_clone_range(&bio_list,
2455                                                                 &bio_offset,
2456                                                                 clone_size,
2457                                                                 GFP_NOIO);
2458                         if (!obj_request->bio_list)
2459                                 goto out_unwind;
2460                 } else if (type == OBJ_REQUEST_PAGES) {
2461                         unsigned int page_count;
2462
2463                         obj_request->pages = pages;
2464                         page_count = (u32)calc_pages_for(offset, length);
2465                         obj_request->page_count = page_count;
2466                         if ((offset + length) & ~PAGE_MASK)
2467                                 page_count--;   /* more on last page */
2468                         pages += page_count;
2469                 }
2470
2471                 osd_req = rbd_osd_req_create(rbd_dev, op_type,
2472                                         (op_type == OBJ_OP_WRITE) ? 2 : 1,
2473                                         obj_request);
2474                 if (!osd_req)
2475                         goto out_unwind;
2476
2477                 obj_request->osd_req = osd_req;
2478                 obj_request->callback = rbd_img_obj_callback;
2479                 obj_request->img_offset = img_offset;
2480
2481                 rbd_img_obj_request_fill(obj_request, osd_req, op_type, 0);
2482
2483                 img_offset += length;
2484                 resid -= length;
2485         }
2486
2487         return 0;
2488
2489 out_unwind:
2490         for_each_obj_request_safe(img_request, obj_request, next_obj_request)
2491                 rbd_img_obj_request_del(img_request, obj_request);
2492
2493         return -ENOMEM;
2494 }
2495
2496 static void
2497 rbd_osd_copyup_callback(struct rbd_obj_request *obj_request)
2498 {
2499         struct rbd_img_request *img_request;
2500         struct rbd_device *rbd_dev;
2501         struct page **pages;
2502         u32 page_count;
2503
2504         dout("%s: obj %p\n", __func__, obj_request);
2505
2506         rbd_assert(obj_request->type == OBJ_REQUEST_BIO ||
2507                 obj_request->type == OBJ_REQUEST_NODATA);
2508         rbd_assert(obj_request_img_data_test(obj_request));
2509         img_request = obj_request->img_request;
2510         rbd_assert(img_request);
2511
2512         rbd_dev = img_request->rbd_dev;
2513         rbd_assert(rbd_dev);
2514
2515         pages = obj_request->copyup_pages;
2516         rbd_assert(pages != NULL);
2517         obj_request->copyup_pages = NULL;
2518         page_count = obj_request->copyup_page_count;
2519         rbd_assert(page_count);
2520         obj_request->copyup_page_count = 0;
2521         ceph_release_page_vector(pages, page_count);
2522
2523         /*
2524          * We want the transfer count to reflect the size of the
2525          * original write request.  There is no such thing as a
2526          * successful short write, so if the request was successful
2527          * we can just set it to the originally-requested length.
2528          */
2529         if (!obj_request->result)
2530                 obj_request->xferred = obj_request->length;
2531
2532         obj_request_done_set(obj_request);
2533 }
2534
2535 static void
2536 rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request)
2537 {
2538         struct rbd_obj_request *orig_request;
2539         struct ceph_osd_request *osd_req;
2540         struct rbd_device *rbd_dev;
2541         struct page **pages;
2542         enum obj_operation_type op_type;
2543         u32 page_count;
2544         int img_result;
2545         u64 parent_length;
2546
2547         rbd_assert(img_request_child_test(img_request));
2548
2549         /* First get what we need from the image request */
2550
2551         pages = img_request->copyup_pages;
2552         rbd_assert(pages != NULL);
2553         img_request->copyup_pages = NULL;
2554         page_count = img_request->copyup_page_count;
2555         rbd_assert(page_count);
2556         img_request->copyup_page_count = 0;
2557
2558         orig_request = img_request->obj_request;
2559         rbd_assert(orig_request != NULL);
2560         rbd_assert(obj_request_type_valid(orig_request->type));
2561         img_result = img_request->result;
2562         parent_length = img_request->length;
2563         rbd_assert(img_result || parent_length == img_request->xferred);
2564         rbd_img_request_put(img_request);
2565
2566         rbd_assert(orig_request->img_request);
2567         rbd_dev = orig_request->img_request->rbd_dev;
2568         rbd_assert(rbd_dev);
2569
2570         /*
2571          * If the overlap has become 0 (most likely because the
2572          * image has been flattened) we need to free the pages
2573          * and re-submit the original write request.
2574          */
2575         if (!rbd_dev->parent_overlap) {
2576                 ceph_release_page_vector(pages, page_count);
2577                 rbd_obj_request_submit(orig_request);
2578                 return;
2579         }
2580
2581         if (img_result)
2582                 goto out_err;
2583
2584         /*
2585          * The original osd request is of no use to use any more.
2586          * We need a new one that can hold the three ops in a copyup
2587          * request.  Allocate the new copyup osd request for the
2588          * original request, and release the old one.
2589          */
2590         img_result = -ENOMEM;
2591         osd_req = rbd_osd_req_create_copyup(orig_request);
2592         if (!osd_req)
2593                 goto out_err;
2594         rbd_osd_req_destroy(orig_request->osd_req);
2595         orig_request->osd_req = osd_req;
2596         orig_request->copyup_pages = pages;
2597         orig_request->copyup_page_count = page_count;
2598
2599         /* Initialize the copyup op */
2600
2601         osd_req_op_cls_init(osd_req, 0, CEPH_OSD_OP_CALL, "rbd", "copyup");
2602         osd_req_op_cls_request_data_pages(osd_req, 0, pages, parent_length, 0,
2603                                                 false, false);
2604
2605         /* Add the other op(s) */
2606
2607         op_type = rbd_img_request_op_type(orig_request->img_request);
2608         rbd_img_obj_request_fill(orig_request, osd_req, op_type, 1);
2609
2610         /* All set, send it off. */
2611
2612         rbd_obj_request_submit(orig_request);
2613         return;
2614
2615 out_err:
2616         ceph_release_page_vector(pages, page_count);
2617         rbd_obj_request_error(orig_request, img_result);
2618 }
2619
2620 /*
2621  * Read from the parent image the range of data that covers the
2622  * entire target of the given object request.  This is used for
2623  * satisfying a layered image write request when the target of an
2624  * object request from the image request does not exist.
2625  *
2626  * A page array big enough to hold the returned data is allocated
2627  * and supplied to rbd_img_request_fill() as the "data descriptor."
2628  * When the read completes, this page array will be transferred to
2629  * the original object request for the copyup operation.
2630  *
2631  * If an error occurs, it is recorded as the result of the original
2632  * object request in rbd_img_obj_exists_callback().
2633  */
2634 static int rbd_img_obj_parent_read_full(struct rbd_obj_request *obj_request)
2635 {
2636         struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
2637         struct rbd_img_request *parent_request = NULL;
2638         u64 img_offset;
2639         u64 length;
2640         struct page **pages = NULL;
2641         u32 page_count;
2642         int result;
2643
2644         rbd_assert(rbd_dev->parent != NULL);
2645
2646         /*
2647          * Determine the byte range covered by the object in the
2648          * child image to which the original request was to be sent.
2649          */
2650         img_offset = obj_request->img_offset - obj_request->offset;
2651         length = rbd_obj_bytes(&rbd_dev->header);
2652
2653         /*
2654          * There is no defined parent data beyond the parent
2655          * overlap, so limit what we read at that boundary if
2656          * necessary.
2657          */
2658         if (img_offset + length > rbd_dev->parent_overlap) {
2659                 rbd_assert(img_offset < rbd_dev->parent_overlap);
2660                 length = rbd_dev->parent_overlap - img_offset;
2661         }
2662
2663         /*
2664          * Allocate a page array big enough to receive the data read
2665          * from the parent.
2666          */
2667         page_count = (u32)calc_pages_for(0, length);
2668         pages = ceph_alloc_page_vector(page_count, GFP_NOIO);
2669         if (IS_ERR(pages)) {
2670                 result = PTR_ERR(pages);
2671                 pages = NULL;
2672                 goto out_err;
2673         }
2674
2675         result = -ENOMEM;
2676         parent_request = rbd_parent_request_create(obj_request,
2677                                                 img_offset, length);
2678         if (!parent_request)
2679                 goto out_err;
2680
2681         result = rbd_img_request_fill(parent_request, OBJ_REQUEST_PAGES, pages);
2682         if (result)
2683                 goto out_err;
2684
2685         parent_request->copyup_pages = pages;
2686         parent_request->copyup_page_count = page_count;
2687         parent_request->callback = rbd_img_obj_parent_read_full_callback;
2688
2689         result = rbd_img_request_submit(parent_request);
2690         if (!result)
2691                 return 0;
2692
2693         parent_request->copyup_pages = NULL;
2694         parent_request->copyup_page_count = 0;
2695         parent_request->obj_request = NULL;
2696         rbd_obj_request_put(obj_request);
2697 out_err:
2698         if (pages)
2699                 ceph_release_page_vector(pages, page_count);
2700         if (parent_request)
2701                 rbd_img_request_put(parent_request);
2702         return result;
2703 }
2704
2705 static void rbd_img_obj_exists_callback(struct rbd_obj_request *obj_request)
2706 {
2707         struct rbd_obj_request *orig_request;
2708         struct rbd_device *rbd_dev;
2709         int result;
2710
2711         rbd_assert(!obj_request_img_data_test(obj_request));
2712
2713         /*
2714          * All we need from the object request is the original
2715          * request and the result of the STAT op.  Grab those, then
2716          * we're done with the request.
2717          */
2718         orig_request = obj_request->obj_request;
2719         obj_request->obj_request = NULL;
2720         rbd_obj_request_put(orig_request);
2721         rbd_assert(orig_request);
2722         rbd_assert(orig_request->img_request);
2723
2724         result = obj_request->result;
2725         obj_request->result = 0;
2726
2727         dout("%s: obj %p for obj %p result %d %llu/%llu\n", __func__,
2728                 obj_request, orig_request, result,
2729                 obj_request->xferred, obj_request->length);
2730         rbd_obj_request_put(obj_request);
2731
2732         /*
2733          * If the overlap has become 0 (most likely because the
2734          * image has been flattened) we need to re-submit the
2735          * original request.
2736          */
2737         rbd_dev = orig_request->img_request->rbd_dev;
2738         if (!rbd_dev->parent_overlap) {
2739                 rbd_obj_request_submit(orig_request);
2740                 return;
2741         }
2742
2743         /*
2744          * Our only purpose here is to determine whether the object
2745          * exists, and we don't want to treat the non-existence as
2746          * an error.  If something else comes back, transfer the
2747          * error to the original request and complete it now.
2748          */
2749         if (!result) {
2750                 obj_request_existence_set(orig_request, true);
2751         } else if (result == -ENOENT) {
2752                 obj_request_existence_set(orig_request, false);
2753         } else {
2754                 goto fail_orig_request;
2755         }
2756
2757         /*
2758          * Resubmit the original request now that we have recorded
2759          * whether the target object exists.
2760          */
2761         result = rbd_img_obj_request_submit(orig_request);
2762         if (result)
2763                 goto fail_orig_request;
2764
2765         return;
2766
2767 fail_orig_request:
2768         rbd_obj_request_error(orig_request, result);
2769 }
2770
2771 static int rbd_img_obj_exists_submit(struct rbd_obj_request *obj_request)
2772 {
2773         struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
2774         struct rbd_obj_request *stat_request;
2775         struct page **pages;
2776         u32 page_count;
2777         size_t size;
2778         int ret;
2779
2780         stat_request = rbd_obj_request_create(OBJ_REQUEST_PAGES);
2781         if (!stat_request)
2782                 return -ENOMEM;
2783
2784         stat_request->object_no = obj_request->object_no;
2785
2786         stat_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1,
2787                                                    stat_request);
2788         if (!stat_request->osd_req) {
2789                 ret = -ENOMEM;
2790                 goto fail_stat_request;
2791         }
2792
2793         /*
2794          * The response data for a STAT call consists of:
2795          *     le64 length;
2796          *     struct {
2797          *         le32 tv_sec;
2798          *         le32 tv_nsec;
2799          *     } mtime;
2800          */
2801         size = sizeof (__le64) + sizeof (__le32) + sizeof (__le32);
2802         page_count = (u32)calc_pages_for(0, size);
2803         pages = ceph_alloc_page_vector(page_count, GFP_NOIO);
2804         if (IS_ERR(pages)) {
2805                 ret = PTR_ERR(pages);
2806                 goto fail_stat_request;
2807         }
2808
2809         osd_req_op_init(stat_request->osd_req, 0, CEPH_OSD_OP_STAT, 0);
2810         osd_req_op_raw_data_in_pages(stat_request->osd_req, 0, pages, size, 0,
2811                                      false, false);
2812
2813         rbd_obj_request_get(obj_request);
2814         stat_request->obj_request = obj_request;
2815         stat_request->pages = pages;
2816         stat_request->page_count = page_count;
2817         stat_request->callback = rbd_img_obj_exists_callback;
2818
2819         rbd_obj_request_submit(stat_request);
2820         return 0;
2821
2822 fail_stat_request:
2823         rbd_obj_request_put(stat_request);
2824         return ret;
2825 }
2826
2827 static bool img_obj_request_simple(struct rbd_obj_request *obj_request)
2828 {
2829         struct rbd_img_request *img_request = obj_request->img_request;
2830         struct rbd_device *rbd_dev = img_request->rbd_dev;
2831
2832         /* Reads */
2833         if (!img_request_write_test(img_request) &&
2834             !img_request_discard_test(img_request))
2835                 return true;
2836
2837         /* Non-layered writes */
2838         if (!img_request_layered_test(img_request))
2839                 return true;
2840
2841         /*
2842          * Layered writes outside of the parent overlap range don't
2843          * share any data with the parent.
2844          */
2845         if (!obj_request_overlaps_parent(obj_request))
2846                 return true;
2847
2848         /*
2849          * Entire-object layered writes - we will overwrite whatever
2850          * parent data there is anyway.
2851          */
2852         if (!obj_request->offset &&
2853             obj_request->length == rbd_obj_bytes(&rbd_dev->header))
2854                 return true;
2855
2856         /*
2857          * If the object is known to already exist, its parent data has
2858          * already been copied.
2859          */
2860         if (obj_request_known_test(obj_request) &&
2861             obj_request_exists_test(obj_request))
2862                 return true;
2863
2864         return false;
2865 }
2866
2867 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request)
2868 {
2869         rbd_assert(obj_request_img_data_test(obj_request));
2870         rbd_assert(obj_request_type_valid(obj_request->type));
2871         rbd_assert(obj_request->img_request);
2872
2873         if (img_obj_request_simple(obj_request)) {
2874                 rbd_obj_request_submit(obj_request);
2875                 return 0;
2876         }
2877
2878         /*
2879          * It's a layered write.  The target object might exist but
2880          * we may not know that yet.  If we know it doesn't exist,
2881          * start by reading the data for the full target object from
2882          * the parent so we can use it for a copyup to the target.
2883          */
2884         if (obj_request_known_test(obj_request))
2885                 return rbd_img_obj_parent_read_full(obj_request);
2886
2887         /* We don't know whether the target exists.  Go find out. */
2888
2889         return rbd_img_obj_exists_submit(obj_request);
2890 }
2891
2892 static int rbd_img_request_submit(struct rbd_img_request *img_request)
2893 {
2894         struct rbd_obj_request *obj_request;
2895         struct rbd_obj_request *next_obj_request;
2896         int ret = 0;
2897
2898         dout("%s: img %p\n", __func__, img_request);
2899
2900         rbd_img_request_get(img_request);
2901         for_each_obj_request_safe(img_request, obj_request, next_obj_request) {
2902                 ret = rbd_img_obj_request_submit(obj_request);
2903                 if (ret)
2904                         goto out_put_ireq;
2905         }
2906
2907 out_put_ireq:
2908         rbd_img_request_put(img_request);
2909         return ret;
2910 }
2911
2912 static void rbd_img_parent_read_callback(struct rbd_img_request *img_request)
2913 {
2914         struct rbd_obj_request *obj_request;
2915         struct rbd_device *rbd_dev;
2916         u64 obj_end;
2917         u64 img_xferred;
2918         int img_result;
2919
2920         rbd_assert(img_request_child_test(img_request));
2921
2922         /* First get what we need from the image request and release it */
2923
2924         obj_request = img_request->obj_request;
2925         img_xferred = img_request->xferred;
2926         img_result = img_request->result;
2927         rbd_img_request_put(img_request);
2928
2929         /*
2930          * If the overlap has become 0 (most likely because the
2931          * image has been flattened) we need to re-submit the
2932          * original request.
2933          */
2934         rbd_assert(obj_request);
2935         rbd_assert(obj_request->img_request);
2936         rbd_dev = obj_request->img_request->rbd_dev;
2937         if (!rbd_dev->parent_overlap) {
2938                 rbd_obj_request_submit(obj_request);
2939                 return;
2940         }
2941
2942         obj_request->result = img_result;
2943         if (obj_request->result)
2944                 goto out;
2945
2946         /*
2947          * We need to zero anything beyond the parent overlap
2948          * boundary.  Since rbd_img_obj_request_read_callback()
2949          * will zero anything beyond the end of a short read, an
2950          * easy way to do this is to pretend the data from the
2951          * parent came up short--ending at the overlap boundary.
2952          */
2953         rbd_assert(obj_request->img_offset < U64_MAX - obj_request->length);
2954         obj_end = obj_request->img_offset + obj_request->length;
2955         if (obj_end > rbd_dev->parent_overlap) {
2956                 u64 xferred = 0;
2957
2958                 if (obj_request->img_offset < rbd_dev->parent_overlap)
2959                         xferred = rbd_dev->parent_overlap -
2960                                         obj_request->img_offset;
2961
2962                 obj_request->xferred = min(img_xferred, xferred);
2963         } else {
2964                 obj_request->xferred = img_xferred;
2965         }
2966 out:
2967         rbd_img_obj_request_read_callback(obj_request);
2968         rbd_obj_request_complete(obj_request);
2969 }
2970
2971 static void rbd_img_parent_read(struct rbd_obj_request *obj_request)
2972 {
2973         struct rbd_img_request *img_request;
2974         int result;
2975
2976         rbd_assert(obj_request_img_data_test(obj_request));
2977         rbd_assert(obj_request->img_request != NULL);
2978         rbd_assert(obj_request->result == (s32) -ENOENT);
2979         rbd_assert(obj_request_type_valid(obj_request->type));
2980
2981         /* rbd_read_finish(obj_request, obj_request->length); */
2982         img_request = rbd_parent_request_create(obj_request,
2983                                                 obj_request->img_offset,
2984                                                 obj_request->length);
2985         result = -ENOMEM;
2986         if (!img_request)
2987                 goto out_err;
2988
2989         if (obj_request->type == OBJ_REQUEST_BIO)
2990                 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
2991                                                 obj_request->bio_list);
2992         else
2993                 result = rbd_img_request_fill(img_request, OBJ_REQUEST_PAGES,
2994                                                 obj_request->pages);
2995         if (result)
2996                 goto out_err;
2997
2998         img_request->callback = rbd_img_parent_read_callback;
2999         result = rbd_img_request_submit(img_request);
3000         if (result)
3001                 goto out_err;
3002
3003         return;
3004 out_err:
3005         if (img_request)
3006                 rbd_img_request_put(img_request);
3007         obj_request->result = result;
3008         obj_request->xferred = 0;
3009         obj_request_done_set(obj_request);
3010 }
3011
3012 static const struct rbd_client_id rbd_empty_cid;
3013
3014 static bool rbd_cid_equal(const struct rbd_client_id *lhs,
3015                           const struct rbd_client_id *rhs)
3016 {
3017         return lhs->gid == rhs->gid && lhs->handle == rhs->handle;
3018 }
3019
3020 static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev)
3021 {
3022         struct rbd_client_id cid;
3023
3024         mutex_lock(&rbd_dev->watch_mutex);
3025         cid.gid = ceph_client_gid(rbd_dev->rbd_client->client);
3026         cid.handle = rbd_dev->watch_cookie;
3027         mutex_unlock(&rbd_dev->watch_mutex);
3028         return cid;
3029 }
3030
3031 /*
3032  * lock_rwsem must be held for write
3033  */
3034 static void rbd_set_owner_cid(struct rbd_device *rbd_dev,
3035                               const struct rbd_client_id *cid)
3036 {
3037         dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev,
3038              rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle,
3039              cid->gid, cid->handle);
3040         rbd_dev->owner_cid = *cid; /* struct */
3041 }
3042
3043 static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf)
3044 {
3045         mutex_lock(&rbd_dev->watch_mutex);
3046         sprintf(buf, "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie);
3047         mutex_unlock(&rbd_dev->watch_mutex);
3048 }
3049
3050 static void __rbd_lock(struct rbd_device *rbd_dev, const char *cookie)
3051 {
3052         struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3053
3054         strcpy(rbd_dev->lock_cookie, cookie);
3055         rbd_set_owner_cid(rbd_dev, &cid);
3056         queue_work(rbd_dev->task_wq, &rbd_dev->acquired_lock_work);
3057 }
3058
3059 /*
3060  * lock_rwsem must be held for write
3061  */
3062 static int rbd_lock(struct rbd_device *rbd_dev)
3063 {
3064         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3065         char cookie[32];
3066         int ret;
3067
3068         WARN_ON(__rbd_is_lock_owner(rbd_dev) ||
3069                 rbd_dev->lock_cookie[0] != '\0');
3070
3071         format_lock_cookie(rbd_dev, cookie);
3072         ret = ceph_cls_lock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3073                             RBD_LOCK_NAME, CEPH_CLS_LOCK_EXCLUSIVE, cookie,
3074                             RBD_LOCK_TAG, "", 0);
3075         if (ret)
3076                 return ret;
3077
3078         rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED;
3079         __rbd_lock(rbd_dev, cookie);
3080         return 0;
3081 }
3082
3083 /*
3084  * lock_rwsem must be held for write
3085  */
3086 static void rbd_unlock(struct rbd_device *rbd_dev)
3087 {
3088         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3089         int ret;
3090
3091         WARN_ON(!__rbd_is_lock_owner(rbd_dev) ||
3092                 rbd_dev->lock_cookie[0] == '\0');
3093
3094         ret = ceph_cls_unlock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3095                               RBD_LOCK_NAME, rbd_dev->lock_cookie);
3096         if (ret && ret != -ENOENT)
3097                 rbd_warn(rbd_dev, "failed to unlock: %d", ret);
3098
3099         /* treat errors as the image is unlocked */
3100         rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
3101         rbd_dev->lock_cookie[0] = '\0';
3102         rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3103         queue_work(rbd_dev->task_wq, &rbd_dev->released_lock_work);
3104 }
3105
3106 static int __rbd_notify_op_lock(struct rbd_device *rbd_dev,
3107                                 enum rbd_notify_op notify_op,
3108                                 struct page ***preply_pages,
3109                                 size_t *preply_len)
3110 {
3111         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3112         struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3113         int buf_size = 4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN;
3114         char buf[buf_size];
3115         void *p = buf;
3116
3117         dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op);
3118
3119         /* encode *LockPayload NotifyMessage (op + ClientId) */
3120         ceph_start_encoding(&p, 2, 1, buf_size - CEPH_ENCODING_START_BLK_LEN);
3121         ceph_encode_32(&p, notify_op);
3122         ceph_encode_64(&p, cid.gid);
3123         ceph_encode_64(&p, cid.handle);
3124
3125         return ceph_osdc_notify(osdc, &rbd_dev->header_oid,
3126                                 &rbd_dev->header_oloc, buf, buf_size,
3127                                 RBD_NOTIFY_TIMEOUT, preply_pages, preply_len);
3128 }
3129
3130 static void rbd_notify_op_lock(struct rbd_device *rbd_dev,
3131                                enum rbd_notify_op notify_op)
3132 {
3133         struct page **reply_pages;
3134         size_t reply_len;
3135
3136         __rbd_notify_op_lock(rbd_dev, notify_op, &reply_pages, &reply_len);
3137         ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3138 }
3139
3140 static void rbd_notify_acquired_lock(struct work_struct *work)
3141 {
3142         struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3143                                                   acquired_lock_work);
3144
3145         rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_ACQUIRED_LOCK);
3146 }
3147
3148 static void rbd_notify_released_lock(struct work_struct *work)
3149 {
3150         struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3151                                                   released_lock_work);
3152
3153         rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_RELEASED_LOCK);
3154 }
3155
3156 static int rbd_request_lock(struct rbd_device *rbd_dev)
3157 {
3158         struct page **reply_pages;
3159         size_t reply_len;
3160         bool lock_owner_responded = false;
3161         int ret;
3162
3163         dout("%s rbd_dev %p\n", __func__, rbd_dev);
3164
3165         ret = __rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_REQUEST_LOCK,
3166                                    &reply_pages, &reply_len);
3167         if (ret && ret != -ETIMEDOUT) {
3168                 rbd_warn(rbd_dev, "failed to request lock: %d", ret);
3169                 goto out;
3170         }
3171
3172         if (reply_len > 0 && reply_len <= PAGE_SIZE) {
3173                 void *p = page_address(reply_pages[0]);
3174                 void *const end = p + reply_len;
3175                 u32 n;
3176
3177                 ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */
3178                 while (n--) {
3179                         u8 struct_v;
3180                         u32 len;
3181
3182                         ceph_decode_need(&p, end, 8 + 8, e_inval);
3183                         p += 8 + 8; /* skip gid and cookie */
3184
3185                         ceph_decode_32_safe(&p, end, len, e_inval);
3186                         if (!len)
3187                                 continue;
3188
3189                         if (lock_owner_responded) {
3190                                 rbd_warn(rbd_dev,
3191                                          "duplicate lock owners detected");
3192                                 ret = -EIO;
3193                                 goto out;
3194                         }
3195
3196                         lock_owner_responded = true;
3197                         ret = ceph_start_decoding(&p, end, 1, "ResponseMessage",
3198                                                   &struct_v, &len);
3199                         if (ret) {
3200                                 rbd_warn(rbd_dev,
3201                                          "failed to decode ResponseMessage: %d",
3202                                          ret);
3203                                 goto e_inval;
3204                         }
3205
3206                         ret = ceph_decode_32(&p);
3207                 }
3208         }
3209
3210         if (!lock_owner_responded) {
3211                 rbd_warn(rbd_dev, "no lock owners detected");
3212                 ret = -ETIMEDOUT;
3213         }
3214
3215 out:
3216         ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3217         return ret;
3218
3219 e_inval:
3220         ret = -EINVAL;
3221         goto out;
3222 }
3223
3224 static void wake_requests(struct rbd_device *rbd_dev, bool wake_all)
3225 {
3226         dout("%s rbd_dev %p wake_all %d\n", __func__, rbd_dev, wake_all);
3227
3228         cancel_delayed_work(&rbd_dev->lock_dwork);
3229         if (wake_all)
3230                 wake_up_all(&rbd_dev->lock_waitq);
3231         else
3232                 wake_up(&rbd_dev->lock_waitq);
3233 }
3234
3235 static int get_lock_owner_info(struct rbd_device *rbd_dev,
3236                                struct ceph_locker **lockers, u32 *num_lockers)
3237 {
3238         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3239         u8 lock_type;
3240         char *lock_tag;
3241         int ret;
3242
3243         dout("%s rbd_dev %p\n", __func__, rbd_dev);
3244
3245         ret = ceph_cls_lock_info(osdc, &rbd_dev->header_oid,
3246                                  &rbd_dev->header_oloc, RBD_LOCK_NAME,
3247                                  &lock_type, &lock_tag, lockers, num_lockers);
3248         if (ret)
3249                 return ret;
3250
3251         if (*num_lockers == 0) {
3252                 dout("%s rbd_dev %p no lockers detected\n", __func__, rbd_dev);
3253                 goto out;
3254         }
3255
3256         if (strcmp(lock_tag, RBD_LOCK_TAG)) {
3257                 rbd_warn(rbd_dev, "locked by external mechanism, tag %s",
3258                          lock_tag);
3259                 ret = -EBUSY;
3260                 goto out;
3261         }
3262
3263         if (lock_type == CEPH_CLS_LOCK_SHARED) {
3264                 rbd_warn(rbd_dev, "shared lock type detected");
3265                 ret = -EBUSY;
3266                 goto out;
3267         }
3268
3269         if (strncmp((*lockers)[0].id.cookie, RBD_LOCK_COOKIE_PREFIX,
3270                     strlen(RBD_LOCK_COOKIE_PREFIX))) {
3271                 rbd_warn(rbd_dev, "locked by external mechanism, cookie %s",
3272                          (*lockers)[0].id.cookie);
3273                 ret = -EBUSY;
3274                 goto out;
3275         }
3276
3277 out:
3278         kfree(lock_tag);
3279         return ret;
3280 }
3281
3282 static int find_watcher(struct rbd_device *rbd_dev,
3283                         const struct ceph_locker *locker)
3284 {
3285         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3286         struct ceph_watch_item *watchers;
3287         u32 num_watchers;
3288         u64 cookie;
3289         int i;
3290         int ret;
3291
3292         ret = ceph_osdc_list_watchers(osdc, &rbd_dev->header_oid,
3293                                       &rbd_dev->header_oloc, &watchers,
3294                                       &num_watchers);
3295         if (ret)
3296                 return ret;
3297
3298         sscanf(locker->id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu", &cookie);
3299         for (i = 0; i < num_watchers; i++) {
3300                 if (!memcmp(&watchers[i].addr, &locker->info.addr,
3301                             sizeof(locker->info.addr)) &&
3302                     watchers[i].cookie == cookie) {
3303                         struct rbd_client_id cid = {
3304                                 .gid = le64_to_cpu(watchers[i].name.num),
3305                                 .handle = cookie,
3306                         };
3307
3308                         dout("%s rbd_dev %p found cid %llu-%llu\n", __func__,
3309                              rbd_dev, cid.gid, cid.handle);
3310                         rbd_set_owner_cid(rbd_dev, &cid);
3311                         ret = 1;
3312                         goto out;
3313                 }
3314         }
3315
3316         dout("%s rbd_dev %p no watchers\n", __func__, rbd_dev);
3317         ret = 0;
3318 out:
3319         kfree(watchers);
3320         return ret;
3321 }
3322
3323 /*
3324  * lock_rwsem must be held for write
3325  */
3326 static int rbd_try_lock(struct rbd_device *rbd_dev)
3327 {
3328         struct ceph_client *client = rbd_dev->rbd_client->client;
3329         struct ceph_locker *lockers;
3330         u32 num_lockers;
3331         int ret;
3332
3333         for (;;) {
3334                 ret = rbd_lock(rbd_dev);
3335                 if (ret != -EBUSY)
3336                         return ret;
3337
3338                 /* determine if the current lock holder is still alive */
3339                 ret = get_lock_owner_info(rbd_dev, &lockers, &num_lockers);
3340                 if (ret)
3341                         return ret;
3342
3343                 if (num_lockers == 0)
3344                         goto again;
3345
3346                 ret = find_watcher(rbd_dev, lockers);
3347                 if (ret) {
3348                         if (ret > 0)
3349                                 ret = 0; /* have to request lock */
3350                         goto out;
3351                 }
3352
3353                 rbd_warn(rbd_dev, "%s%llu seems dead, breaking lock",
3354                          ENTITY_NAME(lockers[0].id.name));
3355
3356                 ret = ceph_monc_blacklist_add(&client->monc,
3357                                               &lockers[0].info.addr);
3358                 if (ret) {
3359                         rbd_warn(rbd_dev, "blacklist of %s%llu failed: %d",
3360                                  ENTITY_NAME(lockers[0].id.name), ret);
3361                         goto out;
3362                 }
3363
3364                 ret = ceph_cls_break_lock(&client->osdc, &rbd_dev->header_oid,
3365                                    &nb