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