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