2 rbd.c -- Export ceph rados objects as a Linux block device
5 based on drivers/block/osdblk.c:
7 Copyright 2009 Red Hat, Inc.
9 This program is free software; you can redistribute it and/or modify
10 it under the terms of the GNU General Public License as published by
11 the Free Software Foundation.
13 This program is distributed in the hope that it will be useful,
14 but WITHOUT ANY WARRANTY; without even the implied warranty of
15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 GNU General Public License for more details.
18 You should have received a copy of the GNU General Public License
19 along with this program; see the file COPYING. If not, write to
20 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
24 For usage instructions, please refer to:
26 Documentation/ABI/testing/sysfs-bus-rbd
30 #include <linux/ceph/libceph.h>
31 #include <linux/ceph/osd_client.h>
32 #include <linux/ceph/mon_client.h>
33 #include <linux/ceph/decode.h>
34 #include <linux/parser.h>
36 #include <linux/kernel.h>
37 #include <linux/device.h>
38 #include <linux/module.h>
40 #include <linux/blkdev.h>
42 #include "rbd_types.h"
44 #define RBD_DEBUG /* Activate rbd_assert() calls */
47 * The basic unit of block I/O is a sector. It is interpreted in a
48 * number of contexts in Linux (blk, bio, genhd), but the default is
49 * universally 512 bytes. These symbols are just slightly more
50 * meaningful than the bare numbers they represent.
52 #define SECTOR_SHIFT 9
53 #define SECTOR_SIZE (1ULL << SECTOR_SHIFT)
55 #define RBD_DRV_NAME "rbd"
56 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
58 #define RBD_MINORS_PER_MAJOR 256 /* max minors per blkdev */
60 #define RBD_SNAP_DEV_NAME_PREFIX "snap_"
61 #define RBD_MAX_SNAP_NAME_LEN \
62 (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
64 #define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */
66 #define RBD_SNAP_HEAD_NAME "-"
68 /* This allows a single page to hold an image name sent by OSD */
69 #define RBD_IMAGE_NAME_LEN_MAX (PAGE_SIZE - sizeof (__le32) - 1)
70 #define RBD_IMAGE_ID_LEN_MAX 64
72 #define RBD_OBJ_PREFIX_LEN_MAX 64
76 #define RBD_FEATURE_LAYERING (1<<0)
77 #define RBD_FEATURE_STRIPINGV2 (1<<1)
78 #define RBD_FEATURES_ALL \
79 (RBD_FEATURE_LAYERING | RBD_FEATURE_STRIPINGV2)
81 /* Features supported by this (client software) implementation. */
83 #define RBD_FEATURES_SUPPORTED (0)
86 * An RBD device name will be "rbd#", where the "rbd" comes from
87 * RBD_DRV_NAME above, and # is a unique integer identifier.
88 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
89 * enough to hold all possible device names.
91 #define DEV_NAME_LEN 32
92 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1)
95 * block device image metadata (in-memory version)
97 struct rbd_image_header {
98 /* These four fields never change for a given rbd image */
105 /* The remaining fields need to be updated occasionally */
107 struct ceph_snap_context *snapc;
115 * An rbd image specification.
117 * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
118 * identify an image. Each rbd_dev structure includes a pointer to
119 * an rbd_spec structure that encapsulates this identity.
121 * Each of the id's in an rbd_spec has an associated name. For a
122 * user-mapped image, the names are supplied and the id's associated
123 * with them are looked up. For a layered image, a parent image is
124 * defined by the tuple, and the names are looked up.
126 * An rbd_dev structure contains a parent_spec pointer which is
127 * non-null if the image it represents is a child in a layered
128 * image. This pointer will refer to the rbd_spec structure used
129 * by the parent rbd_dev for its own identity (i.e., the structure
130 * is shared between the parent and child).
132 * Since these structures are populated once, during the discovery
133 * phase of image construction, they are effectively immutable so
134 * we make no effort to synchronize access to them.
136 * Note that code herein does not assume the image name is known (it
137 * could be a null pointer).
153 * an instance of the client. multiple devices may share an rbd client.
156 struct ceph_client *client;
158 struct list_head node;
161 struct rbd_img_request;
162 typedef void (*rbd_img_callback_t)(struct rbd_img_request *);
164 #define BAD_WHICH U32_MAX /* Good which or bad which, which? */
166 struct rbd_obj_request;
167 typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);
169 enum obj_request_type {
170 OBJ_REQUEST_NODATA, OBJ_REQUEST_BIO, OBJ_REQUEST_PAGES
174 OBJ_REQ_DONE, /* completion flag: not done = 0, done = 1 */
175 OBJ_REQ_IMG_DATA, /* object usage: standalone = 0, image = 1 */
176 OBJ_REQ_KNOWN, /* EXISTS flag valid: no = 0, yes = 1 */
177 OBJ_REQ_EXISTS, /* target exists: no = 0, yes = 1 */
180 struct rbd_obj_request {
181 const char *object_name;
182 u64 offset; /* object start byte */
183 u64 length; /* bytes from offset */
187 * An object request associated with an image will have its
188 * img_data flag set; a standalone object request will not.
190 * A standalone object request will have which == BAD_WHICH
191 * and a null obj_request pointer.
193 * An object request initiated in support of a layered image
194 * object (to check for its existence before a write) will
195 * have which == BAD_WHICH and a non-null obj_request pointer.
197 * Finally, an object request for rbd image data will have
198 * which != BAD_WHICH, and will have a non-null img_request
199 * pointer. The value of which will be in the range
200 * 0..(img_request->obj_request_count-1).
203 struct rbd_obj_request *obj_request; /* STAT op */
205 struct rbd_img_request *img_request;
207 /* links for img_request->obj_requests list */
208 struct list_head links;
211 u32 which; /* posn image request list */
213 enum obj_request_type type;
215 struct bio *bio_list;
222 struct ceph_osd_request *osd_req;
224 u64 xferred; /* bytes transferred */
228 rbd_obj_callback_t callback;
229 struct completion completion;
235 IMG_REQ_WRITE, /* I/O direction: read = 0, write = 1 */
236 IMG_REQ_CHILD, /* initiator: block = 0, child image = 1 */
237 IMG_REQ_LAYERED, /* ENOENT handling: normal = 0, layered = 1 */
240 struct rbd_img_request {
241 struct rbd_device *rbd_dev;
242 u64 offset; /* starting image byte offset */
243 u64 length; /* byte count from offset */
246 u64 snap_id; /* for reads */
247 struct ceph_snap_context *snapc; /* for writes */
250 struct request *rq; /* block request */
251 struct rbd_obj_request *obj_request; /* obj req initiator */
253 struct page **copyup_pages;
254 spinlock_t completion_lock;/* protects next_completion */
256 rbd_img_callback_t callback;
257 u64 xferred;/* aggregate bytes transferred */
258 int result; /* first nonzero obj_request result */
260 u32 obj_request_count;
261 struct list_head obj_requests; /* rbd_obj_request structs */
266 #define for_each_obj_request(ireq, oreq) \
267 list_for_each_entry(oreq, &(ireq)->obj_requests, links)
268 #define for_each_obj_request_from(ireq, oreq) \
269 list_for_each_entry_from(oreq, &(ireq)->obj_requests, links)
270 #define for_each_obj_request_safe(ireq, oreq, n) \
271 list_for_each_entry_safe_reverse(oreq, n, &(ireq)->obj_requests, links)
277 struct list_head node;
292 int dev_id; /* blkdev unique id */
294 int major; /* blkdev assigned major */
295 struct gendisk *disk; /* blkdev's gendisk and rq */
297 u32 image_format; /* Either 1 or 2 */
298 struct rbd_client *rbd_client;
300 char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
302 spinlock_t lock; /* queue, flags, open_count */
304 struct rbd_image_header header;
305 unsigned long flags; /* possibly lock protected */
306 struct rbd_spec *spec;
310 struct ceph_file_layout layout;
312 struct ceph_osd_event *watch_event;
313 struct rbd_obj_request *watch_request;
315 struct rbd_spec *parent_spec;
317 struct rbd_device *parent;
319 /* protects updating the header */
320 struct rw_semaphore header_rwsem;
322 struct rbd_mapping mapping;
324 struct list_head node;
326 /* list of snapshots */
327 struct list_head snaps;
331 unsigned long open_count; /* protected by lock */
335 * Flag bits for rbd_dev->flags. If atomicity is required,
336 * rbd_dev->lock is used to protect access.
338 * Currently, only the "removing" flag (which is coupled with the
339 * "open_count" field) requires atomic access.
342 RBD_DEV_FLAG_EXISTS, /* mapped snapshot has not been deleted */
343 RBD_DEV_FLAG_REMOVING, /* this mapping is being removed */
346 static DEFINE_MUTEX(ctl_mutex); /* Serialize open/close/setup/teardown */
348 static LIST_HEAD(rbd_dev_list); /* devices */
349 static DEFINE_SPINLOCK(rbd_dev_list_lock);
351 static LIST_HEAD(rbd_client_list); /* clients */
352 static DEFINE_SPINLOCK(rbd_client_list_lock);
354 static int rbd_img_request_submit(struct rbd_img_request *img_request);
356 static int rbd_dev_snaps_update(struct rbd_device *rbd_dev);
357 static int rbd_dev_snaps_register(struct rbd_device *rbd_dev);
359 static void rbd_dev_release(struct device *dev);
360 static void rbd_remove_snap_dev(struct rbd_snap *snap);
362 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
364 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
366 static int rbd_dev_probe(struct rbd_device *rbd_dev);
368 static struct bus_attribute rbd_bus_attrs[] = {
369 __ATTR(add, S_IWUSR, NULL, rbd_add),
370 __ATTR(remove, S_IWUSR, NULL, rbd_remove),
374 static struct bus_type rbd_bus_type = {
376 .bus_attrs = rbd_bus_attrs,
379 static void rbd_root_dev_release(struct device *dev)
383 static struct device rbd_root_dev = {
385 .release = rbd_root_dev_release,
388 static __printf(2, 3)
389 void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
391 struct va_format vaf;
399 printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
400 else if (rbd_dev->disk)
401 printk(KERN_WARNING "%s: %s: %pV\n",
402 RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
403 else if (rbd_dev->spec && rbd_dev->spec->image_name)
404 printk(KERN_WARNING "%s: image %s: %pV\n",
405 RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
406 else if (rbd_dev->spec && rbd_dev->spec->image_id)
407 printk(KERN_WARNING "%s: id %s: %pV\n",
408 RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
410 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
411 RBD_DRV_NAME, rbd_dev, &vaf);
416 #define rbd_assert(expr) \
417 if (unlikely(!(expr))) { \
418 printk(KERN_ERR "\nAssertion failure in %s() " \
420 "\trbd_assert(%s);\n\n", \
421 __func__, __LINE__, #expr); \
424 #else /* !RBD_DEBUG */
425 # define rbd_assert(expr) ((void) 0)
426 #endif /* !RBD_DEBUG */
428 static void rbd_img_parent_read(struct rbd_obj_request *obj_request);
429 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request);
431 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver);
432 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver);
434 static int rbd_open(struct block_device *bdev, fmode_t mode)
436 struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
437 bool removing = false;
439 if ((mode & FMODE_WRITE) && rbd_dev->mapping.read_only)
442 spin_lock_irq(&rbd_dev->lock);
443 if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
446 rbd_dev->open_count++;
447 spin_unlock_irq(&rbd_dev->lock);
451 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
452 (void) get_device(&rbd_dev->dev);
453 set_device_ro(bdev, rbd_dev->mapping.read_only);
454 mutex_unlock(&ctl_mutex);
459 static int rbd_release(struct gendisk *disk, fmode_t mode)
461 struct rbd_device *rbd_dev = disk->private_data;
462 unsigned long open_count_before;
464 spin_lock_irq(&rbd_dev->lock);
465 open_count_before = rbd_dev->open_count--;
466 spin_unlock_irq(&rbd_dev->lock);
467 rbd_assert(open_count_before > 0);
469 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
470 put_device(&rbd_dev->dev);
471 mutex_unlock(&ctl_mutex);
476 static const struct block_device_operations rbd_bd_ops = {
477 .owner = THIS_MODULE,
479 .release = rbd_release,
483 * Initialize an rbd client instance.
486 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
488 struct rbd_client *rbdc;
491 dout("%s:\n", __func__);
492 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
496 kref_init(&rbdc->kref);
497 INIT_LIST_HEAD(&rbdc->node);
499 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
501 rbdc->client = ceph_create_client(ceph_opts, rbdc, 0, 0);
502 if (IS_ERR(rbdc->client))
504 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
506 ret = ceph_open_session(rbdc->client);
510 spin_lock(&rbd_client_list_lock);
511 list_add_tail(&rbdc->node, &rbd_client_list);
512 spin_unlock(&rbd_client_list_lock);
514 mutex_unlock(&ctl_mutex);
515 dout("%s: rbdc %p\n", __func__, rbdc);
520 ceph_destroy_client(rbdc->client);
522 mutex_unlock(&ctl_mutex);
526 ceph_destroy_options(ceph_opts);
527 dout("%s: error %d\n", __func__, ret);
532 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
534 kref_get(&rbdc->kref);
540 * Find a ceph client with specific addr and configuration. If
541 * found, bump its reference count.
543 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
545 struct rbd_client *client_node;
548 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
551 spin_lock(&rbd_client_list_lock);
552 list_for_each_entry(client_node, &rbd_client_list, node) {
553 if (!ceph_compare_options(ceph_opts, client_node->client)) {
554 __rbd_get_client(client_node);
560 spin_unlock(&rbd_client_list_lock);
562 return found ? client_node : NULL;
572 /* string args above */
575 /* Boolean args above */
579 static match_table_t rbd_opts_tokens = {
581 /* string args above */
582 {Opt_read_only, "read_only"},
583 {Opt_read_only, "ro"}, /* Alternate spelling */
584 {Opt_read_write, "read_write"},
585 {Opt_read_write, "rw"}, /* Alternate spelling */
586 /* Boolean args above */
594 #define RBD_READ_ONLY_DEFAULT false
596 static int parse_rbd_opts_token(char *c, void *private)
598 struct rbd_options *rbd_opts = private;
599 substring_t argstr[MAX_OPT_ARGS];
600 int token, intval, ret;
602 token = match_token(c, rbd_opts_tokens, argstr);
606 if (token < Opt_last_int) {
607 ret = match_int(&argstr[0], &intval);
609 pr_err("bad mount option arg (not int) "
613 dout("got int token %d val %d\n", token, intval);
614 } else if (token > Opt_last_int && token < Opt_last_string) {
615 dout("got string token %d val %s\n", token,
617 } else if (token > Opt_last_string && token < Opt_last_bool) {
618 dout("got Boolean token %d\n", token);
620 dout("got token %d\n", token);
625 rbd_opts->read_only = true;
628 rbd_opts->read_only = false;
638 * Get a ceph client with specific addr and configuration, if one does
639 * not exist create it.
641 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
643 struct rbd_client *rbdc;
645 rbdc = rbd_client_find(ceph_opts);
646 if (rbdc) /* using an existing client */
647 ceph_destroy_options(ceph_opts);
649 rbdc = rbd_client_create(ceph_opts);
655 * Destroy ceph client
657 * Caller must hold rbd_client_list_lock.
659 static void rbd_client_release(struct kref *kref)
661 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
663 dout("%s: rbdc %p\n", __func__, rbdc);
664 spin_lock(&rbd_client_list_lock);
665 list_del(&rbdc->node);
666 spin_unlock(&rbd_client_list_lock);
668 ceph_destroy_client(rbdc->client);
673 * Drop reference to ceph client node. If it's not referenced anymore, release
676 static void rbd_put_client(struct rbd_client *rbdc)
679 kref_put(&rbdc->kref, rbd_client_release);
682 static bool rbd_image_format_valid(u32 image_format)
684 return image_format == 1 || image_format == 2;
687 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
692 /* The header has to start with the magic rbd header text */
693 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
696 /* The bio layer requires at least sector-sized I/O */
698 if (ondisk->options.order < SECTOR_SHIFT)
701 /* If we use u64 in a few spots we may be able to loosen this */
703 if (ondisk->options.order > 8 * sizeof (int) - 1)
707 * The size of a snapshot header has to fit in a size_t, and
708 * that limits the number of snapshots.
710 snap_count = le32_to_cpu(ondisk->snap_count);
711 size = SIZE_MAX - sizeof (struct ceph_snap_context);
712 if (snap_count > size / sizeof (__le64))
716 * Not only that, but the size of the entire the snapshot
717 * header must also be representable in a size_t.
719 size -= snap_count * sizeof (__le64);
720 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
727 * Create a new header structure, translate header format from the on-disk
730 static int rbd_header_from_disk(struct rbd_image_header *header,
731 struct rbd_image_header_ondisk *ondisk)
738 memset(header, 0, sizeof (*header));
740 snap_count = le32_to_cpu(ondisk->snap_count);
742 len = strnlen(ondisk->object_prefix, sizeof (ondisk->object_prefix));
743 header->object_prefix = kmalloc(len + 1, GFP_KERNEL);
744 if (!header->object_prefix)
746 memcpy(header->object_prefix, ondisk->object_prefix, len);
747 header->object_prefix[len] = '\0';
750 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
752 /* Save a copy of the snapshot names */
754 if (snap_names_len > (u64) SIZE_MAX)
756 header->snap_names = kmalloc(snap_names_len, GFP_KERNEL);
757 if (!header->snap_names)
760 * Note that rbd_dev_v1_header_read() guarantees
761 * the ondisk buffer we're working with has
762 * snap_names_len bytes beyond the end of the
763 * snapshot id array, this memcpy() is safe.
765 memcpy(header->snap_names, &ondisk->snaps[snap_count],
768 /* Record each snapshot's size */
770 size = snap_count * sizeof (*header->snap_sizes);
771 header->snap_sizes = kmalloc(size, GFP_KERNEL);
772 if (!header->snap_sizes)
774 for (i = 0; i < snap_count; i++)
775 header->snap_sizes[i] =
776 le64_to_cpu(ondisk->snaps[i].image_size);
778 WARN_ON(ondisk->snap_names_len);
779 header->snap_names = NULL;
780 header->snap_sizes = NULL;
783 header->features = 0; /* No features support in v1 images */
784 header->obj_order = ondisk->options.order;
785 header->crypt_type = ondisk->options.crypt_type;
786 header->comp_type = ondisk->options.comp_type;
788 /* Allocate and fill in the snapshot context */
790 header->image_size = le64_to_cpu(ondisk->image_size);
791 size = sizeof (struct ceph_snap_context);
792 size += snap_count * sizeof (header->snapc->snaps[0]);
793 header->snapc = kzalloc(size, GFP_KERNEL);
797 atomic_set(&header->snapc->nref, 1);
798 header->snapc->seq = le64_to_cpu(ondisk->snap_seq);
799 header->snapc->num_snaps = snap_count;
800 for (i = 0; i < snap_count; i++)
801 header->snapc->snaps[i] =
802 le64_to_cpu(ondisk->snaps[i].id);
807 kfree(header->snap_sizes);
808 header->snap_sizes = NULL;
809 kfree(header->snap_names);
810 header->snap_names = NULL;
811 kfree(header->object_prefix);
812 header->object_prefix = NULL;
817 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
819 struct rbd_snap *snap;
821 if (snap_id == CEPH_NOSNAP)
822 return RBD_SNAP_HEAD_NAME;
824 list_for_each_entry(snap, &rbd_dev->snaps, node)
825 if (snap_id == snap->id)
831 static int snap_by_name(struct rbd_device *rbd_dev, const char *snap_name)
834 struct rbd_snap *snap;
836 list_for_each_entry(snap, &rbd_dev->snaps, node) {
837 if (!strcmp(snap_name, snap->name)) {
838 rbd_dev->spec->snap_id = snap->id;
839 rbd_dev->mapping.size = snap->size;
840 rbd_dev->mapping.features = snap->features;
849 static int rbd_dev_set_mapping(struct rbd_device *rbd_dev)
853 if (!memcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME,
854 sizeof (RBD_SNAP_HEAD_NAME))) {
855 rbd_dev->spec->snap_id = CEPH_NOSNAP;
856 rbd_dev->mapping.size = rbd_dev->header.image_size;
857 rbd_dev->mapping.features = rbd_dev->header.features;
860 ret = snap_by_name(rbd_dev, rbd_dev->spec->snap_name);
863 rbd_dev->mapping.read_only = true;
865 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
871 static void rbd_header_free(struct rbd_image_header *header)
873 kfree(header->object_prefix);
874 header->object_prefix = NULL;
875 kfree(header->snap_sizes);
876 header->snap_sizes = NULL;
877 kfree(header->snap_names);
878 header->snap_names = NULL;
879 ceph_put_snap_context(header->snapc);
880 header->snapc = NULL;
883 static const char *rbd_segment_name(struct rbd_device *rbd_dev, u64 offset)
889 name = kmalloc(MAX_OBJ_NAME_SIZE + 1, GFP_NOIO);
892 segment = offset >> rbd_dev->header.obj_order;
893 ret = snprintf(name, MAX_OBJ_NAME_SIZE + 1, "%s.%012llx",
894 rbd_dev->header.object_prefix, segment);
895 if (ret < 0 || ret > MAX_OBJ_NAME_SIZE) {
896 pr_err("error formatting segment name for #%llu (%d)\n",
905 static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset)
907 u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
909 return offset & (segment_size - 1);
912 static u64 rbd_segment_length(struct rbd_device *rbd_dev,
913 u64 offset, u64 length)
915 u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
917 offset &= segment_size - 1;
919 rbd_assert(length <= U64_MAX - offset);
920 if (offset + length > segment_size)
921 length = segment_size - offset;
927 * returns the size of an object in the image
929 static u64 rbd_obj_bytes(struct rbd_image_header *header)
931 return 1 << header->obj_order;
938 static void bio_chain_put(struct bio *chain)
944 chain = chain->bi_next;
950 * zeros a bio chain, starting at specific offset
952 static void zero_bio_chain(struct bio *chain, int start_ofs)
961 bio_for_each_segment(bv, chain, i) {
962 if (pos + bv->bv_len > start_ofs) {
963 int remainder = max(start_ofs - pos, 0);
964 buf = bvec_kmap_irq(bv, &flags);
965 memset(buf + remainder, 0,
966 bv->bv_len - remainder);
967 bvec_kunmap_irq(buf, &flags);
972 chain = chain->bi_next;
977 * similar to zero_bio_chain(), zeros data defined by a page array,
978 * starting at the given byte offset from the start of the array and
979 * continuing up to the given end offset. The pages array is
980 * assumed to be big enough to hold all bytes up to the end.
982 static void zero_pages(struct page **pages, u64 offset, u64 end)
984 struct page **page = &pages[offset >> PAGE_SHIFT];
986 rbd_assert(end > offset);
987 rbd_assert(end - offset <= (u64)SIZE_MAX);
988 while (offset < end) {
994 page_offset = (size_t)(offset & ~PAGE_MASK);
995 length = min(PAGE_SIZE - page_offset, (size_t)(end - offset));
996 local_irq_save(flags);
997 kaddr = kmap_atomic(*page);
998 memset(kaddr + page_offset, 0, length);
999 kunmap_atomic(kaddr);
1000 local_irq_restore(flags);
1008 * Clone a portion of a bio, starting at the given byte offset
1009 * and continuing for the number of bytes indicated.
1011 static struct bio *bio_clone_range(struct bio *bio_src,
1012 unsigned int offset,
1020 unsigned short end_idx;
1021 unsigned short vcnt;
1024 /* Handle the easy case for the caller */
1026 if (!offset && len == bio_src->bi_size)
1027 return bio_clone(bio_src, gfpmask);
1029 if (WARN_ON_ONCE(!len))
1031 if (WARN_ON_ONCE(len > bio_src->bi_size))
1033 if (WARN_ON_ONCE(offset > bio_src->bi_size - len))
1036 /* Find first affected segment... */
1039 __bio_for_each_segment(bv, bio_src, idx, 0) {
1040 if (resid < bv->bv_len)
1042 resid -= bv->bv_len;
1046 /* ...and the last affected segment */
1049 __bio_for_each_segment(bv, bio_src, end_idx, idx) {
1050 if (resid <= bv->bv_len)
1052 resid -= bv->bv_len;
1054 vcnt = end_idx - idx + 1;
1056 /* Build the clone */
1058 bio = bio_alloc(gfpmask, (unsigned int) vcnt);
1060 return NULL; /* ENOMEM */
1062 bio->bi_bdev = bio_src->bi_bdev;
1063 bio->bi_sector = bio_src->bi_sector + (offset >> SECTOR_SHIFT);
1064 bio->bi_rw = bio_src->bi_rw;
1065 bio->bi_flags |= 1 << BIO_CLONED;
1068 * Copy over our part of the bio_vec, then update the first
1069 * and last (or only) entries.
1071 memcpy(&bio->bi_io_vec[0], &bio_src->bi_io_vec[idx],
1072 vcnt * sizeof (struct bio_vec));
1073 bio->bi_io_vec[0].bv_offset += voff;
1075 bio->bi_io_vec[0].bv_len -= voff;
1076 bio->bi_io_vec[vcnt - 1].bv_len = resid;
1078 bio->bi_io_vec[0].bv_len = len;
1081 bio->bi_vcnt = vcnt;
1089 * Clone a portion of a bio chain, starting at the given byte offset
1090 * into the first bio in the source chain and continuing for the
1091 * number of bytes indicated. The result is another bio chain of
1092 * exactly the given length, or a null pointer on error.
1094 * The bio_src and offset parameters are both in-out. On entry they
1095 * refer to the first source bio and the offset into that bio where
1096 * the start of data to be cloned is located.
1098 * On return, bio_src is updated to refer to the bio in the source
1099 * chain that contains first un-cloned byte, and *offset will
1100 * contain the offset of that byte within that bio.
1102 static struct bio *bio_chain_clone_range(struct bio **bio_src,
1103 unsigned int *offset,
1107 struct bio *bi = *bio_src;
1108 unsigned int off = *offset;
1109 struct bio *chain = NULL;
1112 /* Build up a chain of clone bios up to the limit */
1114 if (!bi || off >= bi->bi_size || !len)
1115 return NULL; /* Nothing to clone */
1119 unsigned int bi_size;
1123 rbd_warn(NULL, "bio_chain exhausted with %u left", len);
1124 goto out_err; /* EINVAL; ran out of bio's */
1126 bi_size = min_t(unsigned int, bi->bi_size - off, len);
1127 bio = bio_clone_range(bi, off, bi_size, gfpmask);
1129 goto out_err; /* ENOMEM */
1132 end = &bio->bi_next;
1135 if (off == bi->bi_size) {
1146 bio_chain_put(chain);
1152 * The default/initial value for all object request flags is 0. For
1153 * each flag, once its value is set to 1 it is never reset to 0
1156 static void obj_request_img_data_set(struct rbd_obj_request *obj_request)
1158 if (test_and_set_bit(OBJ_REQ_IMG_DATA, &obj_request->flags)) {
1159 struct rbd_device *rbd_dev;
1161 rbd_dev = obj_request->img_request->rbd_dev;
1162 rbd_warn(rbd_dev, "obj_request %p already marked img_data\n",
1167 static bool obj_request_img_data_test(struct rbd_obj_request *obj_request)
1170 return test_bit(OBJ_REQ_IMG_DATA, &obj_request->flags) != 0;
1173 static void obj_request_done_set(struct rbd_obj_request *obj_request)
1175 if (test_and_set_bit(OBJ_REQ_DONE, &obj_request->flags)) {
1176 struct rbd_device *rbd_dev = NULL;
1178 if (obj_request_img_data_test(obj_request))
1179 rbd_dev = obj_request->img_request->rbd_dev;
1180 rbd_warn(rbd_dev, "obj_request %p already marked done\n",
1185 static bool obj_request_done_test(struct rbd_obj_request *obj_request)
1188 return test_bit(OBJ_REQ_DONE, &obj_request->flags) != 0;
1192 * This sets the KNOWN flag after (possibly) setting the EXISTS
1193 * flag. The latter is set based on the "exists" value provided.
1195 * Note that for our purposes once an object exists it never goes
1196 * away again. It's possible that the response from two existence
1197 * checks are separated by the creation of the target object, and
1198 * the first ("doesn't exist") response arrives *after* the second
1199 * ("does exist"). In that case we ignore the second one.
1201 static void obj_request_existence_set(struct rbd_obj_request *obj_request,
1205 set_bit(OBJ_REQ_EXISTS, &obj_request->flags);
1206 set_bit(OBJ_REQ_KNOWN, &obj_request->flags);
1210 static bool obj_request_known_test(struct rbd_obj_request *obj_request)
1213 return test_bit(OBJ_REQ_KNOWN, &obj_request->flags) != 0;
1216 static bool obj_request_exists_test(struct rbd_obj_request *obj_request)
1219 return test_bit(OBJ_REQ_EXISTS, &obj_request->flags) != 0;
1222 static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
1224 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1225 atomic_read(&obj_request->kref.refcount));
1226 kref_get(&obj_request->kref);
1229 static void rbd_obj_request_destroy(struct kref *kref);
1230 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1232 rbd_assert(obj_request != NULL);
1233 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1234 atomic_read(&obj_request->kref.refcount));
1235 kref_put(&obj_request->kref, rbd_obj_request_destroy);
1238 static void rbd_img_request_get(struct rbd_img_request *img_request)
1240 dout("%s: img %p (was %d)\n", __func__, img_request,
1241 atomic_read(&img_request->kref.refcount));
1242 kref_get(&img_request->kref);
1245 static void rbd_img_request_destroy(struct kref *kref);
1246 static void rbd_img_request_put(struct rbd_img_request *img_request)
1248 rbd_assert(img_request != NULL);
1249 dout("%s: img %p (was %d)\n", __func__, img_request,
1250 atomic_read(&img_request->kref.refcount));
1251 kref_put(&img_request->kref, rbd_img_request_destroy);
1254 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1255 struct rbd_obj_request *obj_request)
1257 rbd_assert(obj_request->img_request == NULL);
1259 /* Image request now owns object's original reference */
1260 obj_request->img_request = img_request;
1261 obj_request->which = img_request->obj_request_count;
1262 rbd_assert(!obj_request_img_data_test(obj_request));
1263 obj_request_img_data_set(obj_request);
1264 rbd_assert(obj_request->which != BAD_WHICH);
1265 img_request->obj_request_count++;
1266 list_add_tail(&obj_request->links, &img_request->obj_requests);
1267 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1268 obj_request->which);
1271 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1272 struct rbd_obj_request *obj_request)
1274 rbd_assert(obj_request->which != BAD_WHICH);
1276 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1277 obj_request->which);
1278 list_del(&obj_request->links);
1279 rbd_assert(img_request->obj_request_count > 0);
1280 img_request->obj_request_count--;
1281 rbd_assert(obj_request->which == img_request->obj_request_count);
1282 obj_request->which = BAD_WHICH;
1283 rbd_assert(obj_request_img_data_test(obj_request));
1284 rbd_assert(obj_request->img_request == img_request);
1285 obj_request->img_request = NULL;
1286 obj_request->callback = NULL;
1287 rbd_obj_request_put(obj_request);
1290 static bool obj_request_type_valid(enum obj_request_type type)
1293 case OBJ_REQUEST_NODATA:
1294 case OBJ_REQUEST_BIO:
1295 case OBJ_REQUEST_PAGES:
1302 static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
1303 struct rbd_obj_request *obj_request)
1305 dout("%s: osdc %p obj %p\n", __func__, osdc, obj_request);
1307 return ceph_osdc_start_request(osdc, obj_request->osd_req, false);
1310 static void rbd_img_request_complete(struct rbd_img_request *img_request)
1313 dout("%s: img %p\n", __func__, img_request);
1316 * If no error occurred, compute the aggregate transfer
1317 * count for the image request. We could instead use
1318 * atomic64_cmpxchg() to update it as each object request
1319 * completes; not clear which way is better off hand.
1321 if (!img_request->result) {
1322 struct rbd_obj_request *obj_request;
1325 for_each_obj_request(img_request, obj_request)
1326 xferred += obj_request->xferred;
1327 img_request->xferred = xferred;
1330 if (img_request->callback)
1331 img_request->callback(img_request);
1333 rbd_img_request_put(img_request);
1336 /* Caller is responsible for rbd_obj_request_destroy(obj_request) */
1338 static int rbd_obj_request_wait(struct rbd_obj_request *obj_request)
1340 dout("%s: obj %p\n", __func__, obj_request);
1342 return wait_for_completion_interruptible(&obj_request->completion);
1346 * The default/initial value for all image request flags is 0. Each
1347 * is conditionally set to 1 at image request initialization time
1348 * and currently never change thereafter.
1350 static void img_request_write_set(struct rbd_img_request *img_request)
1352 set_bit(IMG_REQ_WRITE, &img_request->flags);
1356 static bool img_request_write_test(struct rbd_img_request *img_request)
1359 return test_bit(IMG_REQ_WRITE, &img_request->flags) != 0;
1362 static void img_request_child_set(struct rbd_img_request *img_request)
1364 set_bit(IMG_REQ_CHILD, &img_request->flags);
1368 static bool img_request_child_test(struct rbd_img_request *img_request)
1371 return test_bit(IMG_REQ_CHILD, &img_request->flags) != 0;
1374 static void img_request_layered_set(struct rbd_img_request *img_request)
1376 set_bit(IMG_REQ_LAYERED, &img_request->flags);
1380 static bool img_request_layered_test(struct rbd_img_request *img_request)
1383 return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1387 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
1389 u64 xferred = obj_request->xferred;
1390 u64 length = obj_request->length;
1392 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1393 obj_request, obj_request->img_request, obj_request->result,
1396 * ENOENT means a hole in the image. We zero-fill the
1397 * entire length of the request. A short read also implies
1398 * zero-fill to the end of the request. Either way we
1399 * update the xferred count to indicate the whole request
1402 rbd_assert(obj_request->type != OBJ_REQUEST_NODATA);
1403 if (obj_request->result == -ENOENT) {
1404 if (obj_request->type == OBJ_REQUEST_BIO)
1405 zero_bio_chain(obj_request->bio_list, 0);
1407 zero_pages(obj_request->pages, 0, length);
1408 obj_request->result = 0;
1409 obj_request->xferred = length;
1410 } else if (xferred < length && !obj_request->result) {
1411 if (obj_request->type == OBJ_REQUEST_BIO)
1412 zero_bio_chain(obj_request->bio_list, xferred);
1414 zero_pages(obj_request->pages, xferred, length);
1415 obj_request->xferred = length;
1417 obj_request_done_set(obj_request);
1420 static void rbd_obj_request_complete(struct rbd_obj_request *obj_request)
1422 dout("%s: obj %p cb %p\n", __func__, obj_request,
1423 obj_request->callback);
1424 if (obj_request->callback)
1425 obj_request->callback(obj_request);
1427 complete_all(&obj_request->completion);
1430 static void rbd_osd_trivial_callback(struct rbd_obj_request *obj_request)
1432 dout("%s: obj %p\n", __func__, obj_request);
1433 obj_request_done_set(obj_request);
1436 static void rbd_osd_read_callback(struct rbd_obj_request *obj_request)
1438 struct rbd_img_request *img_request = NULL;
1439 bool layered = false;
1441 if (obj_request_img_data_test(obj_request)) {
1442 img_request = obj_request->img_request;
1443 layered = img_request && img_request_layered_test(img_request);
1449 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1450 obj_request, img_request, obj_request->result,
1451 obj_request->xferred, obj_request->length);
1452 if (layered && obj_request->result == -ENOENT)
1453 rbd_img_parent_read(obj_request);
1454 else if (img_request)
1455 rbd_img_obj_request_read_callback(obj_request);
1457 obj_request_done_set(obj_request);
1460 static void rbd_osd_write_callback(struct rbd_obj_request *obj_request)
1462 dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1463 obj_request->result, obj_request->length);
1465 * There is no such thing as a successful short write. Set
1466 * it to our originally-requested length.
1468 obj_request->xferred = obj_request->length;
1469 obj_request_done_set(obj_request);
1473 * For a simple stat call there's nothing to do. We'll do more if
1474 * this is part of a write sequence for a layered image.
1476 static void rbd_osd_stat_callback(struct rbd_obj_request *obj_request)
1478 dout("%s: obj %p\n", __func__, obj_request);
1479 obj_request_done_set(obj_request);
1482 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
1483 struct ceph_msg *msg)
1485 struct rbd_obj_request *obj_request = osd_req->r_priv;
1488 dout("%s: osd_req %p msg %p\n", __func__, osd_req, msg);
1489 rbd_assert(osd_req == obj_request->osd_req);
1490 if (obj_request_img_data_test(obj_request)) {
1491 rbd_assert(obj_request->img_request);
1492 rbd_assert(obj_request->which != BAD_WHICH);
1494 rbd_assert(obj_request->which == BAD_WHICH);
1497 if (osd_req->r_result < 0)
1498 obj_request->result = osd_req->r_result;
1499 obj_request->version = le64_to_cpu(osd_req->r_reassert_version.version);
1501 WARN_ON(osd_req->r_num_ops != 1); /* For now */
1504 * We support a 64-bit length, but ultimately it has to be
1505 * passed to blk_end_request(), which takes an unsigned int.
1507 obj_request->xferred = osd_req->r_reply_op_len[0];
1508 rbd_assert(obj_request->xferred < (u64)UINT_MAX);
1509 opcode = osd_req->r_ops[0].op;
1511 case CEPH_OSD_OP_READ:
1512 rbd_osd_read_callback(obj_request);
1514 case CEPH_OSD_OP_WRITE:
1515 rbd_osd_write_callback(obj_request);
1517 case CEPH_OSD_OP_STAT:
1518 rbd_osd_stat_callback(obj_request);
1520 case CEPH_OSD_OP_CALL:
1521 case CEPH_OSD_OP_NOTIFY_ACK:
1522 case CEPH_OSD_OP_WATCH:
1523 rbd_osd_trivial_callback(obj_request);
1526 rbd_warn(NULL, "%s: unsupported op %hu\n",
1527 obj_request->object_name, (unsigned short) opcode);
1531 if (obj_request_done_test(obj_request))
1532 rbd_obj_request_complete(obj_request);
1535 static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request)
1537 struct rbd_img_request *img_request = obj_request->img_request;
1538 struct ceph_osd_request *osd_req = obj_request->osd_req;
1541 rbd_assert(osd_req != NULL);
1543 snap_id = img_request ? img_request->snap_id : CEPH_NOSNAP;
1544 ceph_osdc_build_request(osd_req, obj_request->offset,
1545 NULL, snap_id, NULL);
1548 static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request)
1550 struct rbd_img_request *img_request = obj_request->img_request;
1551 struct ceph_osd_request *osd_req = obj_request->osd_req;
1552 struct ceph_snap_context *snapc;
1553 struct timespec mtime = CURRENT_TIME;
1555 rbd_assert(osd_req != NULL);
1557 snapc = img_request ? img_request->snapc : NULL;
1558 ceph_osdc_build_request(osd_req, obj_request->offset,
1559 snapc, CEPH_NOSNAP, &mtime);
1562 static struct ceph_osd_request *rbd_osd_req_create(
1563 struct rbd_device *rbd_dev,
1565 struct rbd_obj_request *obj_request)
1567 struct ceph_snap_context *snapc = NULL;
1568 struct ceph_osd_client *osdc;
1569 struct ceph_osd_request *osd_req;
1571 if (obj_request_img_data_test(obj_request)) {
1572 struct rbd_img_request *img_request = obj_request->img_request;
1574 rbd_assert(write_request ==
1575 img_request_write_test(img_request));
1577 snapc = img_request->snapc;
1580 /* Allocate and initialize the request, for the single op */
1582 osdc = &rbd_dev->rbd_client->client->osdc;
1583 osd_req = ceph_osdc_alloc_request(osdc, snapc, 1, false, GFP_ATOMIC);
1585 return NULL; /* ENOMEM */
1588 osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
1590 osd_req->r_flags = CEPH_OSD_FLAG_READ;
1592 osd_req->r_callback = rbd_osd_req_callback;
1593 osd_req->r_priv = obj_request;
1595 osd_req->r_oid_len = strlen(obj_request->object_name);
1596 rbd_assert(osd_req->r_oid_len < sizeof (osd_req->r_oid));
1597 memcpy(osd_req->r_oid, obj_request->object_name, osd_req->r_oid_len);
1599 osd_req->r_file_layout = rbd_dev->layout; /* struct */
1604 static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req)
1606 ceph_osdc_put_request(osd_req);
1609 /* object_name is assumed to be a non-null pointer and NUL-terminated */
1611 static struct rbd_obj_request *rbd_obj_request_create(const char *object_name,
1612 u64 offset, u64 length,
1613 enum obj_request_type type)
1615 struct rbd_obj_request *obj_request;
1619 rbd_assert(obj_request_type_valid(type));
1621 size = strlen(object_name) + 1;
1622 obj_request = kzalloc(sizeof (*obj_request) + size, GFP_KERNEL);
1626 name = (char *)(obj_request + 1);
1627 obj_request->object_name = memcpy(name, object_name, size);
1628 obj_request->offset = offset;
1629 obj_request->length = length;
1630 obj_request->flags = 0;
1631 obj_request->which = BAD_WHICH;
1632 obj_request->type = type;
1633 INIT_LIST_HEAD(&obj_request->links);
1634 init_completion(&obj_request->completion);
1635 kref_init(&obj_request->kref);
1637 dout("%s: \"%s\" %llu/%llu %d -> obj %p\n", __func__, object_name,
1638 offset, length, (int)type, obj_request);
1643 static void rbd_obj_request_destroy(struct kref *kref)
1645 struct rbd_obj_request *obj_request;
1647 obj_request = container_of(kref, struct rbd_obj_request, kref);
1649 dout("%s: obj %p\n", __func__, obj_request);
1651 rbd_assert(obj_request->img_request == NULL);
1652 rbd_assert(obj_request->which == BAD_WHICH);
1654 if (obj_request->osd_req)
1655 rbd_osd_req_destroy(obj_request->osd_req);
1657 rbd_assert(obj_request_type_valid(obj_request->type));
1658 switch (obj_request->type) {
1659 case OBJ_REQUEST_NODATA:
1660 break; /* Nothing to do */
1661 case OBJ_REQUEST_BIO:
1662 if (obj_request->bio_list)
1663 bio_chain_put(obj_request->bio_list);
1665 case OBJ_REQUEST_PAGES:
1666 if (obj_request->pages)
1667 ceph_release_page_vector(obj_request->pages,
1668 obj_request->page_count);
1676 * Caller is responsible for filling in the list of object requests
1677 * that comprises the image request, and the Linux request pointer
1678 * (if there is one).
1680 static struct rbd_img_request *rbd_img_request_create(
1681 struct rbd_device *rbd_dev,
1682 u64 offset, u64 length,
1686 struct rbd_img_request *img_request;
1687 struct ceph_snap_context *snapc = NULL;
1689 img_request = kmalloc(sizeof (*img_request), GFP_ATOMIC);
1693 if (write_request) {
1694 down_read(&rbd_dev->header_rwsem);
1695 snapc = ceph_get_snap_context(rbd_dev->header.snapc);
1696 up_read(&rbd_dev->header_rwsem);
1697 if (WARN_ON(!snapc)) {
1699 return NULL; /* Shouldn't happen */
1704 img_request->rq = NULL;
1705 img_request->rbd_dev = rbd_dev;
1706 img_request->offset = offset;
1707 img_request->length = length;
1708 img_request->flags = 0;
1709 if (write_request) {
1710 img_request_write_set(img_request);
1711 img_request->snapc = snapc;
1713 img_request->snap_id = rbd_dev->spec->snap_id;
1716 img_request_child_set(img_request);
1717 if (rbd_dev->parent_spec)
1718 img_request_layered_set(img_request);
1719 spin_lock_init(&img_request->completion_lock);
1720 img_request->next_completion = 0;
1721 img_request->callback = NULL;
1722 img_request->result = 0;
1723 img_request->obj_request_count = 0;
1724 INIT_LIST_HEAD(&img_request->obj_requests);
1725 kref_init(&img_request->kref);
1727 rbd_img_request_get(img_request); /* Avoid a warning */
1728 rbd_img_request_put(img_request); /* TEMPORARY */
1730 dout("%s: rbd_dev %p %s %llu/%llu -> img %p\n", __func__, rbd_dev,
1731 write_request ? "write" : "read", offset, length,
1737 static void rbd_img_request_destroy(struct kref *kref)
1739 struct rbd_img_request *img_request;
1740 struct rbd_obj_request *obj_request;
1741 struct rbd_obj_request *next_obj_request;
1743 img_request = container_of(kref, struct rbd_img_request, kref);
1745 dout("%s: img %p\n", __func__, img_request);
1747 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1748 rbd_img_obj_request_del(img_request, obj_request);
1749 rbd_assert(img_request->obj_request_count == 0);
1751 if (img_request_write_test(img_request))
1752 ceph_put_snap_context(img_request->snapc);
1754 if (img_request_child_test(img_request))
1755 rbd_obj_request_put(img_request->obj_request);
1760 static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request)
1762 struct rbd_img_request *img_request;
1763 unsigned int xferred;
1767 rbd_assert(obj_request_img_data_test(obj_request));
1768 img_request = obj_request->img_request;
1770 rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
1771 xferred = (unsigned int)obj_request->xferred;
1772 result = obj_request->result;
1774 struct rbd_device *rbd_dev = img_request->rbd_dev;
1776 rbd_warn(rbd_dev, "%s %llx at %llx (%llx)\n",
1777 img_request_write_test(img_request) ? "write" : "read",
1778 obj_request->length, obj_request->img_offset,
1779 obj_request->offset);
1780 rbd_warn(rbd_dev, " result %d xferred %x\n",
1782 if (!img_request->result)
1783 img_request->result = result;
1786 /* Image object requests don't own their page array */
1788 if (obj_request->type == OBJ_REQUEST_PAGES) {
1789 obj_request->pages = NULL;
1790 obj_request->page_count = 0;
1793 if (img_request_child_test(img_request)) {
1794 rbd_assert(img_request->obj_request != NULL);
1795 more = obj_request->which < img_request->obj_request_count - 1;
1797 rbd_assert(img_request->rq != NULL);
1798 more = blk_end_request(img_request->rq, result, xferred);
1804 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request)
1806 struct rbd_img_request *img_request;
1807 u32 which = obj_request->which;
1810 rbd_assert(obj_request_img_data_test(obj_request));
1811 img_request = obj_request->img_request;
1813 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1814 rbd_assert(img_request != NULL);
1815 rbd_assert(img_request->obj_request_count > 0);
1816 rbd_assert(which != BAD_WHICH);
1817 rbd_assert(which < img_request->obj_request_count);
1818 rbd_assert(which >= img_request->next_completion);
1820 spin_lock_irq(&img_request->completion_lock);
1821 if (which != img_request->next_completion)
1824 for_each_obj_request_from(img_request, obj_request) {
1826 rbd_assert(which < img_request->obj_request_count);
1828 if (!obj_request_done_test(obj_request))
1830 more = rbd_img_obj_end_request(obj_request);
1834 rbd_assert(more ^ (which == img_request->obj_request_count));
1835 img_request->next_completion = which;
1837 spin_unlock_irq(&img_request->completion_lock);
1840 rbd_img_request_complete(img_request);
1844 * Split up an image request into one or more object requests, each
1845 * to a different object. The "type" parameter indicates whether
1846 * "data_desc" is the pointer to the head of a list of bio
1847 * structures, or the base of a page array. In either case this
1848 * function assumes data_desc describes memory sufficient to hold
1849 * all data described by the image request.
1851 static int rbd_img_request_fill(struct rbd_img_request *img_request,
1852 enum obj_request_type type,
1855 struct rbd_device *rbd_dev = img_request->rbd_dev;
1856 struct rbd_obj_request *obj_request = NULL;
1857 struct rbd_obj_request *next_obj_request;
1858 bool write_request = img_request_write_test(img_request);
1859 struct bio *bio_list;
1860 unsigned int bio_offset = 0;
1861 struct page **pages;
1866 dout("%s: img %p type %d data_desc %p\n", __func__, img_request,
1867 (int)type, data_desc);
1869 opcode = write_request ? CEPH_OSD_OP_WRITE : CEPH_OSD_OP_READ;
1870 img_offset = img_request->offset;
1871 resid = img_request->length;
1872 rbd_assert(resid > 0);
1874 if (type == OBJ_REQUEST_BIO) {
1875 bio_list = data_desc;
1876 rbd_assert(img_offset == bio_list->bi_sector << SECTOR_SHIFT);
1878 rbd_assert(type == OBJ_REQUEST_PAGES);
1883 struct ceph_osd_request *osd_req;
1884 const char *object_name;
1888 object_name = rbd_segment_name(rbd_dev, img_offset);
1891 offset = rbd_segment_offset(rbd_dev, img_offset);
1892 length = rbd_segment_length(rbd_dev, img_offset, resid);
1893 obj_request = rbd_obj_request_create(object_name,
1894 offset, length, type);
1895 kfree(object_name); /* object request has its own copy */
1899 if (type == OBJ_REQUEST_BIO) {
1900 unsigned int clone_size;
1902 rbd_assert(length <= (u64)UINT_MAX);
1903 clone_size = (unsigned int)length;
1904 obj_request->bio_list =
1905 bio_chain_clone_range(&bio_list,
1909 if (!obj_request->bio_list)
1912 unsigned int page_count;
1914 obj_request->pages = pages;
1915 page_count = (u32)calc_pages_for(offset, length);
1916 obj_request->page_count = page_count;
1917 if ((offset + length) & ~PAGE_MASK)
1918 page_count--; /* more on last page */
1919 pages += page_count;
1922 osd_req = rbd_osd_req_create(rbd_dev, write_request,
1926 obj_request->osd_req = osd_req;
1927 obj_request->callback = rbd_img_obj_callback;
1929 osd_req_op_extent_init(osd_req, 0, opcode, offset, length,
1931 if (type == OBJ_REQUEST_BIO)
1932 osd_req_op_extent_osd_data_bio(osd_req, 0,
1933 obj_request->bio_list, length);
1935 osd_req_op_extent_osd_data_pages(osd_req, 0,
1936 obj_request->pages, length,
1937 offset & ~PAGE_MASK, false, false);
1940 rbd_osd_req_format_write(obj_request);
1942 rbd_osd_req_format_read(obj_request);
1944 obj_request->img_offset = img_offset;
1945 rbd_img_obj_request_add(img_request, obj_request);
1947 img_offset += length;
1954 rbd_obj_request_put(obj_request);
1956 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1957 rbd_obj_request_put(obj_request);
1963 rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request)
1965 struct rbd_obj_request *orig_request;
1966 struct page **pages;
1972 rbd_assert(img_request_child_test(img_request));
1974 /* First get what we need from the image request */
1976 pages = img_request->copyup_pages;
1977 rbd_assert(pages != NULL);
1978 img_request->copyup_pages = NULL;
1980 orig_request = img_request->obj_request;
1981 rbd_assert(orig_request != NULL);
1983 result = img_request->result;
1984 obj_size = img_request->length;
1985 xferred = img_request->xferred;
1987 rbd_img_request_put(img_request);
1989 obj_request_existence_set(orig_request, true);
1991 page_count = (u32)calc_pages_for(0, obj_size);
1992 ceph_release_page_vector(pages, page_count);
1994 /* Resubmit the original request (for now). */
1996 orig_request->result = rbd_img_obj_request_submit(orig_request);
1997 if (orig_request->result) {
1998 obj_request_done_set(orig_request);
1999 rbd_obj_request_complete(orig_request);
2004 * Read from the parent image the range of data that covers the
2005 * entire target of the given object request. This is used for
2006 * satisfying a layered image write request when the target of an
2007 * object request from the image request does not exist.
2009 * A page array big enough to hold the returned data is allocated
2010 * and supplied to rbd_img_request_fill() as the "data descriptor."
2011 * When the read completes, this page array will be transferred to
2012 * the original object request for the copyup operation.
2014 * If an error occurs, record it as the result of the original
2015 * object request and mark it done so it gets completed.
2017 static int rbd_img_obj_parent_read_full(struct rbd_obj_request *obj_request)
2019 struct rbd_img_request *img_request = NULL;
2020 struct rbd_img_request *parent_request = NULL;
2021 struct rbd_device *rbd_dev;
2024 struct page **pages = NULL;
2028 rbd_assert(obj_request_img_data_test(obj_request));
2029 rbd_assert(obj_request->type == OBJ_REQUEST_BIO);
2031 img_request = obj_request->img_request;
2032 rbd_assert(img_request != NULL);
2033 rbd_dev = img_request->rbd_dev;
2034 rbd_assert(rbd_dev->parent != NULL);
2037 * Determine the byte range covered by the object in the
2038 * child image to which the original request was to be sent.
2040 img_offset = obj_request->img_offset - obj_request->offset;
2041 length = (u64)1 << rbd_dev->header.obj_order;
2044 * Allocate a page array big enough to receive the data read
2047 page_count = (u32)calc_pages_for(0, length);
2048 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2049 if (IS_ERR(pages)) {
2050 result = PTR_ERR(pages);
2056 parent_request = rbd_img_request_create(rbd_dev->parent,
2059 if (!parent_request)
2061 rbd_obj_request_get(obj_request);
2062 parent_request->obj_request = obj_request;
2064 result = rbd_img_request_fill(parent_request, OBJ_REQUEST_PAGES, pages);
2067 parent_request->copyup_pages = pages;
2069 parent_request->callback = rbd_img_obj_parent_read_full_callback;
2070 result = rbd_img_request_submit(parent_request);
2074 parent_request->copyup_pages = NULL;
2075 parent_request->obj_request = NULL;
2076 rbd_obj_request_put(obj_request);
2079 ceph_release_page_vector(pages, page_count);
2081 rbd_img_request_put(parent_request);
2082 obj_request->result = result;
2083 obj_request->xferred = 0;
2084 obj_request_done_set(obj_request);
2089 static void rbd_img_obj_exists_callback(struct rbd_obj_request *obj_request)
2091 struct rbd_obj_request *orig_request;
2094 rbd_assert(!obj_request_img_data_test(obj_request));
2097 * All we need from the object request is the original
2098 * request and the result of the STAT op. Grab those, then
2099 * we're done with the request.
2101 orig_request = obj_request->obj_request;
2102 obj_request->obj_request = NULL;
2103 rbd_assert(orig_request);
2104 rbd_assert(orig_request->img_request);
2106 result = obj_request->result;
2107 obj_request->result = 0;
2109 dout("%s: obj %p for obj %p result %d %llu/%llu\n", __func__,
2110 obj_request, orig_request, result,
2111 obj_request->xferred, obj_request->length);
2112 rbd_obj_request_put(obj_request);
2114 rbd_assert(orig_request);
2115 rbd_assert(orig_request->img_request);
2118 * Our only purpose here is to determine whether the object
2119 * exists, and we don't want to treat the non-existence as
2120 * an error. If something else comes back, transfer the
2121 * error to the original request and complete it now.
2124 obj_request_existence_set(orig_request, true);
2125 } else if (result == -ENOENT) {
2126 obj_request_existence_set(orig_request, false);
2127 } else if (result) {
2128 orig_request->result = result;
2133 * Resubmit the original request now that we have recorded
2134 * whether the target object exists.
2136 orig_request->result = rbd_img_obj_request_submit(orig_request);
2138 if (orig_request->result)
2139 rbd_obj_request_complete(orig_request);
2140 rbd_obj_request_put(orig_request);
2143 static int rbd_img_obj_exists_submit(struct rbd_obj_request *obj_request)
2145 struct rbd_obj_request *stat_request;
2146 struct rbd_device *rbd_dev;
2147 struct ceph_osd_client *osdc;
2148 struct page **pages = NULL;
2154 * The response data for a STAT call consists of:
2161 size = sizeof (__le64) + sizeof (__le32) + sizeof (__le32);
2162 page_count = (u32)calc_pages_for(0, size);
2163 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2165 return PTR_ERR(pages);
2168 stat_request = rbd_obj_request_create(obj_request->object_name, 0, 0,
2173 rbd_obj_request_get(obj_request);
2174 stat_request->obj_request = obj_request;
2175 stat_request->pages = pages;
2176 stat_request->page_count = page_count;
2178 rbd_assert(obj_request->img_request);
2179 rbd_dev = obj_request->img_request->rbd_dev;
2180 stat_request->osd_req = rbd_osd_req_create(rbd_dev, false,
2182 if (!stat_request->osd_req)
2184 stat_request->callback = rbd_img_obj_exists_callback;
2186 osd_req_op_init(stat_request->osd_req, 0, CEPH_OSD_OP_STAT);
2187 osd_req_op_raw_data_in_pages(stat_request->osd_req, 0, pages, size, 0,
2189 rbd_osd_req_format_read(stat_request);
2191 osdc = &rbd_dev->rbd_client->client->osdc;
2192 ret = rbd_obj_request_submit(osdc, stat_request);
2195 rbd_obj_request_put(obj_request);
2200 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request)
2202 struct rbd_img_request *img_request;
2205 rbd_assert(obj_request_img_data_test(obj_request));
2207 img_request = obj_request->img_request;
2208 rbd_assert(img_request);
2211 * Only layered writes need special handling. If it's not a
2212 * layered write, or it is a layered write but we know the
2213 * target object exists, it's no different from any other
2216 if (!img_request_write_test(img_request) ||
2217 !img_request_layered_test(img_request) ||
2218 ((known = obj_request_known_test(obj_request)) &&
2219 obj_request_exists_test(obj_request))) {
2221 struct rbd_device *rbd_dev;
2222 struct ceph_osd_client *osdc;
2224 rbd_dev = obj_request->img_request->rbd_dev;
2225 osdc = &rbd_dev->rbd_client->client->osdc;
2227 return rbd_obj_request_submit(osdc, obj_request);
2231 * It's a layered write. The target object might exist but
2232 * we may not know that yet. If we know it doesn't exist,
2233 * start by reading the data for the full target object from
2234 * the parent so we can use it for a copyup to the target.
2237 return rbd_img_obj_parent_read_full(obj_request);
2239 /* We don't know whether the target exists. Go find out. */
2241 return rbd_img_obj_exists_submit(obj_request);
2244 static int rbd_img_request_submit(struct rbd_img_request *img_request)
2246 struct rbd_obj_request *obj_request;
2247 struct rbd_obj_request *next_obj_request;
2249 dout("%s: img %p\n", __func__, img_request);
2250 for_each_obj_request_safe(img_request, obj_request, next_obj_request) {
2253 ret = rbd_img_obj_request_submit(obj_request);
2261 static void rbd_img_parent_read_callback(struct rbd_img_request *img_request)
2263 struct rbd_obj_request *obj_request;
2265 rbd_assert(img_request_child_test(img_request));
2267 obj_request = img_request->obj_request;
2268 rbd_assert(obj_request != NULL);
2269 obj_request->result = img_request->result;
2270 obj_request->xferred = img_request->xferred;
2272 rbd_img_obj_request_read_callback(obj_request);
2273 rbd_obj_request_complete(obj_request);
2276 static void rbd_img_parent_read(struct rbd_obj_request *obj_request)
2278 struct rbd_device *rbd_dev;
2279 struct rbd_img_request *img_request;
2282 rbd_assert(obj_request_img_data_test(obj_request));
2283 rbd_assert(obj_request->img_request != NULL);
2284 rbd_assert(obj_request->result == (s32) -ENOENT);
2285 rbd_assert(obj_request->type == OBJ_REQUEST_BIO);
2287 rbd_dev = obj_request->img_request->rbd_dev;
2288 rbd_assert(rbd_dev->parent != NULL);
2289 /* rbd_read_finish(obj_request, obj_request->length); */
2290 img_request = rbd_img_request_create(rbd_dev->parent,
2291 obj_request->img_offset,
2292 obj_request->length,
2298 rbd_obj_request_get(obj_request);
2299 img_request->obj_request = obj_request;
2301 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
2302 obj_request->bio_list);
2306 img_request->callback = rbd_img_parent_read_callback;
2307 result = rbd_img_request_submit(img_request);
2314 rbd_img_request_put(img_request);
2315 obj_request->result = result;
2316 obj_request->xferred = 0;
2317 obj_request_done_set(obj_request);
2320 static int rbd_obj_notify_ack(struct rbd_device *rbd_dev,
2321 u64 ver, u64 notify_id)
2323 struct rbd_obj_request *obj_request;
2324 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2327 obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0,
2328 OBJ_REQUEST_NODATA);
2333 obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
2334 if (!obj_request->osd_req)
2336 obj_request->callback = rbd_obj_request_put;
2338 osd_req_op_watch_init(obj_request->osd_req, 0, CEPH_OSD_OP_NOTIFY_ACK,
2340 rbd_osd_req_format_read(obj_request);
2342 ret = rbd_obj_request_submit(osdc, obj_request);
2345 rbd_obj_request_put(obj_request);
2350 static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
2352 struct rbd_device *rbd_dev = (struct rbd_device *)data;
2359 dout("%s: \"%s\" notify_id %llu opcode %u\n", __func__,
2360 rbd_dev->header_name, (unsigned long long) notify_id,
2361 (unsigned int) opcode);
2362 rc = rbd_dev_refresh(rbd_dev, &hver);
2364 rbd_warn(rbd_dev, "got notification but failed to "
2365 " update snaps: %d\n", rc);
2367 rbd_obj_notify_ack(rbd_dev, hver, notify_id);
2371 * Request sync osd watch/unwatch. The value of "start" determines
2372 * whether a watch request is being initiated or torn down.
2374 static int rbd_dev_header_watch_sync(struct rbd_device *rbd_dev, int start)
2376 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2377 struct rbd_obj_request *obj_request;
2380 rbd_assert(start ^ !!rbd_dev->watch_event);
2381 rbd_assert(start ^ !!rbd_dev->watch_request);
2384 ret = ceph_osdc_create_event(osdc, rbd_watch_cb, rbd_dev,
2385 &rbd_dev->watch_event);
2388 rbd_assert(rbd_dev->watch_event != NULL);
2392 obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0,
2393 OBJ_REQUEST_NODATA);
2397 obj_request->osd_req = rbd_osd_req_create(rbd_dev, true, obj_request);
2398 if (!obj_request->osd_req)
2402 ceph_osdc_set_request_linger(osdc, obj_request->osd_req);
2404 ceph_osdc_unregister_linger_request(osdc,
2405 rbd_dev->watch_request->osd_req);
2407 osd_req_op_watch_init(obj_request->osd_req, 0, CEPH_OSD_OP_WATCH,
2408 rbd_dev->watch_event->cookie,
2409 rbd_dev->header.obj_version, start);
2410 rbd_osd_req_format_write(obj_request);
2412 ret = rbd_obj_request_submit(osdc, obj_request);
2415 ret = rbd_obj_request_wait(obj_request);
2418 ret = obj_request->result;
2423 * A watch request is set to linger, so the underlying osd
2424 * request won't go away until we unregister it. We retain
2425 * a pointer to the object request during that time (in
2426 * rbd_dev->watch_request), so we'll keep a reference to
2427 * it. We'll drop that reference (below) after we've
2431 rbd_dev->watch_request = obj_request;
2436 /* We have successfully torn down the watch request */
2438 rbd_obj_request_put(rbd_dev->watch_request);
2439 rbd_dev->watch_request = NULL;
2441 /* Cancel the event if we're tearing down, or on error */
2442 ceph_osdc_cancel_event(rbd_dev->watch_event);
2443 rbd_dev->watch_event = NULL;
2445 rbd_obj_request_put(obj_request);
2451 * Synchronous osd object method call
2453 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
2454 const char *object_name,
2455 const char *class_name,
2456 const char *method_name,
2457 const char *outbound,
2458 size_t outbound_size,
2460 size_t inbound_size,
2463 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2464 struct rbd_obj_request *obj_request;
2465 struct page **pages;
2470 * Method calls are ultimately read operations. The result
2471 * should placed into the inbound buffer provided. They
2472 * also supply outbound data--parameters for the object
2473 * method. Currently if this is present it will be a
2476 page_count = (u32) calc_pages_for(0, inbound_size);
2477 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2479 return PTR_ERR(pages);
2482 obj_request = rbd_obj_request_create(object_name, 0, inbound_size,
2487 obj_request->pages = pages;
2488 obj_request->page_count = page_count;
2490 obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
2491 if (!obj_request->osd_req)
2494 osd_req_op_cls_init(obj_request->osd_req, 0, CEPH_OSD_OP_CALL,
2495 class_name, method_name);
2496 if (outbound_size) {
2497 struct ceph_pagelist *pagelist;
2499 pagelist = kmalloc(sizeof (*pagelist), GFP_NOFS);
2503 ceph_pagelist_init(pagelist);
2504 ceph_pagelist_append(pagelist, outbound, outbound_size);
2505 osd_req_op_cls_request_data_pagelist(obj_request->osd_req, 0,
2508 osd_req_op_cls_response_data_pages(obj_request->osd_req, 0,
2509 obj_request->pages, inbound_size,
2511 rbd_osd_req_format_read(obj_request);
2513 ret = rbd_obj_request_submit(osdc, obj_request);
2516 ret = rbd_obj_request_wait(obj_request);
2520 ret = obj_request->result;
2524 ceph_copy_from_page_vector(pages, inbound, 0, obj_request->xferred);
2526 *version = obj_request->version;
2529 rbd_obj_request_put(obj_request);
2531 ceph_release_page_vector(pages, page_count);
2536 static void rbd_request_fn(struct request_queue *q)
2537 __releases(q->queue_lock) __acquires(q->queue_lock)
2539 struct rbd_device *rbd_dev = q->queuedata;
2540 bool read_only = rbd_dev->mapping.read_only;
2544 while ((rq = blk_fetch_request(q))) {
2545 bool write_request = rq_data_dir(rq) == WRITE;
2546 struct rbd_img_request *img_request;
2550 /* Ignore any non-FS requests that filter through. */
2552 if (rq->cmd_type != REQ_TYPE_FS) {
2553 dout("%s: non-fs request type %d\n", __func__,
2554 (int) rq->cmd_type);
2555 __blk_end_request_all(rq, 0);
2559 /* Ignore/skip any zero-length requests */
2561 offset = (u64) blk_rq_pos(rq) << SECTOR_SHIFT;
2562 length = (u64) blk_rq_bytes(rq);
2565 dout("%s: zero-length request\n", __func__);
2566 __blk_end_request_all(rq, 0);
2570 spin_unlock_irq(q->queue_lock);
2572 /* Disallow writes to a read-only device */
2574 if (write_request) {
2578 rbd_assert(rbd_dev->spec->snap_id == CEPH_NOSNAP);
2582 * Quit early if the mapped snapshot no longer
2583 * exists. It's still possible the snapshot will
2584 * have disappeared by the time our request arrives
2585 * at the osd, but there's no sense in sending it if
2588 if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags)) {
2589 dout("request for non-existent snapshot");
2590 rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
2596 if (WARN_ON(offset && length > U64_MAX - offset + 1))
2597 goto end_request; /* Shouldn't happen */
2600 img_request = rbd_img_request_create(rbd_dev, offset, length,
2601 write_request, false);
2605 img_request->rq = rq;
2607 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
2610 result = rbd_img_request_submit(img_request);
2612 rbd_img_request_put(img_request);
2614 spin_lock_irq(q->queue_lock);
2616 rbd_warn(rbd_dev, "%s %llx at %llx result %d\n",
2617 write_request ? "write" : "read",
2618 length, offset, result);
2620 __blk_end_request_all(rq, result);
2626 * a queue callback. Makes sure that we don't create a bio that spans across
2627 * multiple osd objects. One exception would be with a single page bios,
2628 * which we handle later at bio_chain_clone_range()
2630 static int rbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bmd,
2631 struct bio_vec *bvec)
2633 struct rbd_device *rbd_dev = q->queuedata;
2634 sector_t sector_offset;
2635 sector_t sectors_per_obj;
2636 sector_t obj_sector_offset;
2640 * Find how far into its rbd object the partition-relative
2641 * bio start sector is to offset relative to the enclosing
2644 sector_offset = get_start_sect(bmd->bi_bdev) + bmd->bi_sector;
2645 sectors_per_obj = 1 << (rbd_dev->header.obj_order - SECTOR_SHIFT);
2646 obj_sector_offset = sector_offset & (sectors_per_obj - 1);
2649 * Compute the number of bytes from that offset to the end
2650 * of the object. Account for what's already used by the bio.
2652 ret = (int) (sectors_per_obj - obj_sector_offset) << SECTOR_SHIFT;
2653 if (ret > bmd->bi_size)
2654 ret -= bmd->bi_size;
2659 * Don't send back more than was asked for. And if the bio
2660 * was empty, let the whole thing through because: "Note
2661 * that a block device *must* allow a single page to be
2662 * added to an empty bio."
2664 rbd_assert(bvec->bv_len <= PAGE_SIZE);
2665 if (ret > (int) bvec->bv_len || !bmd->bi_size)
2666 ret = (int) bvec->bv_len;
2671 static void rbd_free_disk(struct rbd_device *rbd_dev)
2673 struct gendisk *disk = rbd_dev->disk;
2678 if (disk->flags & GENHD_FL_UP)
2681 blk_cleanup_queue(disk->queue);
2685 static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
2686 const char *object_name,
2687 u64 offset, u64 length,
2688 char *buf, u64 *version)
2691 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2692 struct rbd_obj_request *obj_request;
2693 struct page **pages = NULL;
2698 page_count = (u32) calc_pages_for(offset, length);
2699 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2701 ret = PTR_ERR(pages);
2704 obj_request = rbd_obj_request_create(object_name, offset, length,
2709 obj_request->pages = pages;
2710 obj_request->page_count = page_count;
2712 obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
2713 if (!obj_request->osd_req)
2716 osd_req_op_extent_init(obj_request->osd_req, 0, CEPH_OSD_OP_READ,
2717 offset, length, 0, 0);
2718 osd_req_op_extent_osd_data_pages(obj_request->osd_req, 0,
2720 obj_request->length,
2721 obj_request->offset & ~PAGE_MASK,
2723 rbd_osd_req_format_read(obj_request);
2725 ret = rbd_obj_request_submit(osdc, obj_request);
2728 ret = rbd_obj_request_wait(obj_request);
2732 ret = obj_request->result;
2736 rbd_assert(obj_request->xferred <= (u64) SIZE_MAX);
2737 size = (size_t) obj_request->xferred;
2738 ceph_copy_from_page_vector(pages, buf, 0, size);
2739 rbd_assert(size <= (size_t) INT_MAX);
2742 *version = obj_request->version;
2745 rbd_obj_request_put(obj_request);
2747 ceph_release_page_vector(pages, page_count);
2753 * Read the complete header for the given rbd device.
2755 * Returns a pointer to a dynamically-allocated buffer containing
2756 * the complete and validated header. Caller can pass the address
2757 * of a variable that will be filled in with the version of the
2758 * header object at the time it was read.
2760 * Returns a pointer-coded errno if a failure occurs.
2762 static struct rbd_image_header_ondisk *
2763 rbd_dev_v1_header_read(struct rbd_device *rbd_dev, u64 *version)
2765 struct rbd_image_header_ondisk *ondisk = NULL;
2772 * The complete header will include an array of its 64-bit
2773 * snapshot ids, followed by the names of those snapshots as
2774 * a contiguous block of NUL-terminated strings. Note that
2775 * the number of snapshots could change by the time we read
2776 * it in, in which case we re-read it.
2783 size = sizeof (*ondisk);
2784 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
2786 ondisk = kmalloc(size, GFP_KERNEL);
2788 return ERR_PTR(-ENOMEM);
2790 ret = rbd_obj_read_sync(rbd_dev, rbd_dev->header_name,
2792 (char *) ondisk, version);
2795 if (WARN_ON((size_t) ret < size)) {
2797 rbd_warn(rbd_dev, "short header read (want %zd got %d)",
2801 if (!rbd_dev_ondisk_valid(ondisk)) {
2803 rbd_warn(rbd_dev, "invalid header");
2807 names_size = le64_to_cpu(ondisk->snap_names_len);
2808 want_count = snap_count;
2809 snap_count = le32_to_cpu(ondisk->snap_count);
2810 } while (snap_count != want_count);
2817 return ERR_PTR(ret);
2821 * reload the ondisk the header
2823 static int rbd_read_header(struct rbd_device *rbd_dev,
2824 struct rbd_image_header *header)
2826 struct rbd_image_header_ondisk *ondisk;
2830 ondisk = rbd_dev_v1_header_read(rbd_dev, &ver);
2832 return PTR_ERR(ondisk);
2833 ret = rbd_header_from_disk(header, ondisk);
2835 header->obj_version = ver;
2841 static void rbd_remove_all_snaps(struct rbd_device *rbd_dev)
2843 struct rbd_snap *snap;
2844 struct rbd_snap *next;
2846 list_for_each_entry_safe(snap, next, &rbd_dev->snaps, node)
2847 rbd_remove_snap_dev(snap);
2850 static void rbd_update_mapping_size(struct rbd_device *rbd_dev)
2854 if (rbd_dev->spec->snap_id != CEPH_NOSNAP)
2857 size = (sector_t) rbd_dev->header.image_size / SECTOR_SIZE;
2858 dout("setting size to %llu sectors", (unsigned long long) size);
2859 rbd_dev->mapping.size = (u64) size;
2860 set_capacity(rbd_dev->disk, size);
2864 * only read the first part of the ondisk header, without the snaps info
2866 static int rbd_dev_v1_refresh(struct rbd_device *rbd_dev, u64 *hver)
2869 struct rbd_image_header h;
2871 ret = rbd_read_header(rbd_dev, &h);
2875 down_write(&rbd_dev->header_rwsem);
2877 /* Update image size, and check for resize of mapped image */
2878 rbd_dev->header.image_size = h.image_size;
2879 rbd_update_mapping_size(rbd_dev);
2881 /* rbd_dev->header.object_prefix shouldn't change */
2882 kfree(rbd_dev->header.snap_sizes);
2883 kfree(rbd_dev->header.snap_names);
2884 /* osd requests may still refer to snapc */
2885 ceph_put_snap_context(rbd_dev->header.snapc);
2888 *hver = h.obj_version;
2889 rbd_dev->header.obj_version = h.obj_version;
2890 rbd_dev->header.image_size = h.image_size;
2891 rbd_dev->header.snapc = h.snapc;
2892 rbd_dev->header.snap_names = h.snap_names;
2893 rbd_dev->header.snap_sizes = h.snap_sizes;
2894 /* Free the extra copy of the object prefix */
2895 WARN_ON(strcmp(rbd_dev->header.object_prefix, h.object_prefix));
2896 kfree(h.object_prefix);
2898 ret = rbd_dev_snaps_update(rbd_dev);
2900 ret = rbd_dev_snaps_register(rbd_dev);
2902 up_write(&rbd_dev->header_rwsem);
2907 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver)
2911 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
2912 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2913 if (rbd_dev->image_format == 1)
2914 ret = rbd_dev_v1_refresh(rbd_dev, hver);
2916 ret = rbd_dev_v2_refresh(rbd_dev, hver);
2917 mutex_unlock(&ctl_mutex);
2918 revalidate_disk(rbd_dev->disk);
2923 static int rbd_init_disk(struct rbd_device *rbd_dev)
2925 struct gendisk *disk;
2926 struct request_queue *q;
2929 /* create gendisk info */
2930 disk = alloc_disk(RBD_MINORS_PER_MAJOR);
2934 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
2936 disk->major = rbd_dev->major;
2937 disk->first_minor = 0;
2938 disk->fops = &rbd_bd_ops;
2939 disk->private_data = rbd_dev;
2941 q = blk_init_queue(rbd_request_fn, &rbd_dev->lock);
2945 /* We use the default size, but let's be explicit about it. */
2946 blk_queue_physical_block_size(q, SECTOR_SIZE);
2948 /* set io sizes to object size */
2949 segment_size = rbd_obj_bytes(&rbd_dev->header);
2950 blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE);
2951 blk_queue_max_segment_size(q, segment_size);
2952 blk_queue_io_min(q, segment_size);
2953 blk_queue_io_opt(q, segment_size);
2955 blk_queue_merge_bvec(q, rbd_merge_bvec);
2958 q->queuedata = rbd_dev;
2960 rbd_dev->disk = disk;
2962 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
2975 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
2977 return container_of(dev, struct rbd_device, dev);
2980 static ssize_t rbd_size_show(struct device *dev,
2981 struct device_attribute *attr, char *buf)
2983 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2986 down_read(&rbd_dev->header_rwsem);
2987 size = get_capacity(rbd_dev->disk);
2988 up_read(&rbd_dev->header_rwsem);
2990 return sprintf(buf, "%llu\n", (unsigned long long) size * SECTOR_SIZE);
2994 * Note this shows the features for whatever's mapped, which is not
2995 * necessarily the base image.
2997 static ssize_t rbd_features_show(struct device *dev,
2998 struct device_attribute *attr, char *buf)
3000 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3002 return sprintf(buf, "0x%016llx\n",
3003 (unsigned long long) rbd_dev->mapping.features);
3006 static ssize_t rbd_major_show(struct device *dev,
3007 struct device_attribute *attr, char *buf)
3009 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3011 return sprintf(buf, "%d\n", rbd_dev->major);
3014 static ssize_t rbd_client_id_show(struct device *dev,
3015 struct device_attribute *attr, char *buf)
3017 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3019 return sprintf(buf, "client%lld\n",
3020 ceph_client_id(rbd_dev->rbd_client->client));
3023 static ssize_t rbd_pool_show(struct device *dev,
3024 struct device_attribute *attr, char *buf)
3026 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3028 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
3031 static ssize_t rbd_pool_id_show(struct device *dev,
3032 struct device_attribute *attr, char *buf)
3034 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3036 return sprintf(buf, "%llu\n",
3037 (unsigned long long) rbd_dev->spec->pool_id);
3040 static ssize_t rbd_name_show(struct device *dev,
3041 struct device_attribute *attr, char *buf)
3043 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3045 if (rbd_dev->spec->image_name)
3046 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
3048 return sprintf(buf, "(unknown)\n");
3051 static ssize_t rbd_image_id_show(struct device *dev,
3052 struct device_attribute *attr, char *buf)
3054 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3056 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
3060 * Shows the name of the currently-mapped snapshot (or
3061 * RBD_SNAP_HEAD_NAME for the base image).
3063 static ssize_t rbd_snap_show(struct device *dev,
3064 struct device_attribute *attr,
3067 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3069 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
3073 * For an rbd v2 image, shows the pool id, image id, and snapshot id
3074 * for the parent image. If there is no parent, simply shows
3075 * "(no parent image)".
3077 static ssize_t rbd_parent_show(struct device *dev,
3078 struct device_attribute *attr,
3081 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3082 struct rbd_spec *spec = rbd_dev->parent_spec;
3087 return sprintf(buf, "(no parent image)\n");
3089 count = sprintf(bufp, "pool_id %llu\npool_name %s\n",
3090 (unsigned long long) spec->pool_id, spec->pool_name);
3095 count = sprintf(bufp, "image_id %s\nimage_name %s\n", spec->image_id,
3096 spec->image_name ? spec->image_name : "(unknown)");
3101 count = sprintf(bufp, "snap_id %llu\nsnap_name %s\n",
3102 (unsigned long long) spec->snap_id, spec->snap_name);
3107 count = sprintf(bufp, "overlap %llu\n", rbd_dev->parent_overlap);
3112 return (ssize_t) (bufp - buf);
3115 static ssize_t rbd_image_refresh(struct device *dev,
3116 struct device_attribute *attr,
3120 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3123 ret = rbd_dev_refresh(rbd_dev, NULL);
3125 return ret < 0 ? ret : size;
3128 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
3129 static DEVICE_ATTR(features, S_IRUGO, rbd_features_show, NULL);
3130 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
3131 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
3132 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
3133 static DEVICE_ATTR(pool_id, S_IRUGO, rbd_pool_id_show, NULL);
3134 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
3135 static DEVICE_ATTR(image_id, S_IRUGO, rbd_image_id_show, NULL);
3136 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
3137 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
3138 static DEVICE_ATTR(parent, S_IRUGO, rbd_parent_show, NULL);
3140 static struct attribute *rbd_attrs[] = {
3141 &dev_attr_size.attr,
3142 &dev_attr_features.attr,
3143 &dev_attr_major.attr,
3144 &dev_attr_client_id.attr,
3145 &dev_attr_pool.attr,
3146 &dev_attr_pool_id.attr,
3147 &dev_attr_name.attr,
3148 &dev_attr_image_id.attr,
3149 &dev_attr_current_snap.attr,
3150 &dev_attr_parent.attr,
3151 &dev_attr_refresh.attr,
3155 static struct attribute_group rbd_attr_group = {
3159 static const struct attribute_group *rbd_attr_groups[] = {
3164 static void rbd_sysfs_dev_release(struct device *dev)
3168 static struct device_type rbd_device_type = {
3170 .groups = rbd_attr_groups,
3171 .release = rbd_sysfs_dev_release,
3179 static ssize_t rbd_snap_size_show(struct device *dev,
3180 struct device_attribute *attr,
3183 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
3185 return sprintf(buf, "%llu\n", (unsigned long long)snap->size);
3188 static ssize_t rbd_snap_id_show(struct device *dev,
3189 struct device_attribute *attr,
3192 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
3194 return sprintf(buf, "%llu\n", (unsigned long long)snap->id);
3197 static ssize_t rbd_snap_features_show(struct device *dev,
3198 struct device_attribute *attr,
3201 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
3203 return sprintf(buf, "0x%016llx\n",
3204 (unsigned long long) snap->features);
3207 static DEVICE_ATTR(snap_size, S_IRUGO, rbd_snap_size_show, NULL);
3208 static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
3209 static DEVICE_ATTR(snap_features, S_IRUGO, rbd_snap_features_show, NULL);
3211 static struct attribute *rbd_snap_attrs[] = {
3212 &dev_attr_snap_size.attr,
3213 &dev_attr_snap_id.attr,
3214 &dev_attr_snap_features.attr,
3218 static struct attribute_group rbd_snap_attr_group = {
3219 .attrs = rbd_snap_attrs,
3222 static void rbd_snap_dev_release(struct device *dev)
3224 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
3229 static const struct attribute_group *rbd_snap_attr_groups[] = {
3230 &rbd_snap_attr_group,
3234 static struct device_type rbd_snap_device_type = {
3235 .groups = rbd_snap_attr_groups,
3236 .release = rbd_snap_dev_release,
3239 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
3241 kref_get(&spec->kref);
3246 static void rbd_spec_free(struct kref *kref);
3247 static void rbd_spec_put(struct rbd_spec *spec)
3250 kref_put(&spec->kref, rbd_spec_free);
3253 static struct rbd_spec *rbd_spec_alloc(void)
3255 struct rbd_spec *spec;
3257 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
3260 kref_init(&spec->kref);
3265 static void rbd_spec_free(struct kref *kref)
3267 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
3269 kfree(spec->pool_name);
3270 kfree(spec->image_id);
3271 kfree(spec->image_name);
3272 kfree(spec->snap_name);
3276 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
3277 struct rbd_spec *spec)
3279 struct rbd_device *rbd_dev;
3281 rbd_dev = kzalloc(sizeof (*rbd_dev), GFP_KERNEL);
3285 spin_lock_init(&rbd_dev->lock);
3287 INIT_LIST_HEAD(&rbd_dev->node);
3288 INIT_LIST_HEAD(&rbd_dev->snaps);
3289 init_rwsem(&rbd_dev->header_rwsem);
3291 rbd_dev->spec = spec;
3292 rbd_dev->rbd_client = rbdc;
3294 /* Initialize the layout used for all rbd requests */
3296 rbd_dev->layout.fl_stripe_unit = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
3297 rbd_dev->layout.fl_stripe_count = cpu_to_le32(1);
3298 rbd_dev->layout.fl_object_size = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
3299 rbd_dev->layout.fl_pg_pool = cpu_to_le32((u32) spec->pool_id);
3304 static void rbd_dev_destroy(struct rbd_device *rbd_dev)
3306 rbd_spec_put(rbd_dev->parent_spec);
3307 kfree(rbd_dev->header_name);
3308 rbd_put_client(rbd_dev->rbd_client);
3309 rbd_spec_put(rbd_dev->spec);
3313 static bool rbd_snap_registered(struct rbd_snap *snap)
3315 bool ret = snap->dev.type == &rbd_snap_device_type;
3316 bool reg = device_is_registered(&snap->dev);
3318 rbd_assert(!ret ^ reg);
3323 static void rbd_remove_snap_dev(struct rbd_snap *snap)
3325 list_del(&snap->node);
3326 if (device_is_registered(&snap->dev))
3327 device_unregister(&snap->dev);
3330 static int rbd_register_snap_dev(struct rbd_snap *snap,
3331 struct device *parent)
3333 struct device *dev = &snap->dev;
3336 dev->type = &rbd_snap_device_type;
3337 dev->parent = parent;
3338 dev->release = rbd_snap_dev_release;
3339 dev_set_name(dev, "%s%s", RBD_SNAP_DEV_NAME_PREFIX, snap->name);
3340 dout("%s: registering device for snapshot %s\n", __func__, snap->name);
3342 ret = device_register(dev);
3347 static struct rbd_snap *__rbd_add_snap_dev(struct rbd_device *rbd_dev,
3348 const char *snap_name,
3349 u64 snap_id, u64 snap_size,
3352 struct rbd_snap *snap;
3355 snap = kzalloc(sizeof (*snap), GFP_KERNEL);
3357 return ERR_PTR(-ENOMEM);
3360 snap->name = kstrdup(snap_name, GFP_KERNEL);
3365 snap->size = snap_size;
3366 snap->features = snap_features;
3374 return ERR_PTR(ret);
3377 static char *rbd_dev_v1_snap_info(struct rbd_device *rbd_dev, u32 which,
3378 u64 *snap_size, u64 *snap_features)
3382 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
3384 *snap_size = rbd_dev->header.snap_sizes[which];
3385 *snap_features = 0; /* No features for v1 */
3387 /* Skip over names until we find the one we are looking for */
3389 snap_name = rbd_dev->header.snap_names;
3391 snap_name += strlen(snap_name) + 1;
3397 * Get the size and object order for an image snapshot, or if
3398 * snap_id is CEPH_NOSNAP, gets this information for the base
3401 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
3402 u8 *order, u64 *snap_size)
3404 __le64 snapid = cpu_to_le64(snap_id);
3409 } __attribute__ ((packed)) size_buf = { 0 };
3411 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3413 (char *) &snapid, sizeof (snapid),
3414 (char *) &size_buf, sizeof (size_buf), NULL);
3415 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3419 *order = size_buf.order;
3420 *snap_size = le64_to_cpu(size_buf.size);
3422 dout(" snap_id 0x%016llx order = %u, snap_size = %llu\n",
3423 (unsigned long long) snap_id, (unsigned int) *order,
3424 (unsigned long long) *snap_size);
3429 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
3431 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
3432 &rbd_dev->header.obj_order,
3433 &rbd_dev->header.image_size);
3436 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
3442 reply_buf = kzalloc(RBD_OBJ_PREFIX_LEN_MAX, GFP_KERNEL);
3446 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3447 "rbd", "get_object_prefix",
3449 reply_buf, RBD_OBJ_PREFIX_LEN_MAX, NULL);
3450 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3455 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
3456 p + RBD_OBJ_PREFIX_LEN_MAX,
3459 if (IS_ERR(rbd_dev->header.object_prefix)) {
3460 ret = PTR_ERR(rbd_dev->header.object_prefix);
3461 rbd_dev->header.object_prefix = NULL;
3463 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix);
3472 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
3475 __le64 snapid = cpu_to_le64(snap_id);
3479 } features_buf = { 0 };
3483 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3484 "rbd", "get_features",
3485 (char *) &snapid, sizeof (snapid),
3486 (char *) &features_buf, sizeof (features_buf),
3488 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3492 incompat = le64_to_cpu(features_buf.incompat);
3493 if (incompat & ~RBD_FEATURES_SUPPORTED)
3496 *snap_features = le64_to_cpu(features_buf.features);
3498 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
3499 (unsigned long long) snap_id,
3500 (unsigned long long) *snap_features,
3501 (unsigned long long) le64_to_cpu(features_buf.incompat));
3506 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
3508 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
3509 &rbd_dev->header.features);
3512 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
3514 struct rbd_spec *parent_spec;
3516 void *reply_buf = NULL;
3524 parent_spec = rbd_spec_alloc();
3528 size = sizeof (__le64) + /* pool_id */
3529 sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX + /* image_id */
3530 sizeof (__le64) + /* snap_id */
3531 sizeof (__le64); /* overlap */
3532 reply_buf = kmalloc(size, GFP_KERNEL);
3538 snapid = cpu_to_le64(CEPH_NOSNAP);
3539 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3540 "rbd", "get_parent",
3541 (char *) &snapid, sizeof (snapid),
3542 (char *) reply_buf, size, NULL);
3543 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3549 end = (char *) reply_buf + size;
3550 ceph_decode_64_safe(&p, end, parent_spec->pool_id, out_err);
3551 if (parent_spec->pool_id == CEPH_NOPOOL)
3552 goto out; /* No parent? No problem. */
3554 /* The ceph file layout needs to fit pool id in 32 bits */
3557 if (WARN_ON(parent_spec->pool_id > (u64) U32_MAX))
3560 image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
3561 if (IS_ERR(image_id)) {
3562 ret = PTR_ERR(image_id);
3565 parent_spec->image_id = image_id;
3566 ceph_decode_64_safe(&p, end, parent_spec->snap_id, out_err);
3567 ceph_decode_64_safe(&p, end, overlap, out_err);
3569 rbd_dev->parent_overlap = overlap;
3570 rbd_dev->parent_spec = parent_spec;
3571 parent_spec = NULL; /* rbd_dev now owns this */
3576 rbd_spec_put(parent_spec);
3581 static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
3583 size_t image_id_size;
3588 void *reply_buf = NULL;
3590 char *image_name = NULL;
3593 rbd_assert(!rbd_dev->spec->image_name);
3595 len = strlen(rbd_dev->spec->image_id);
3596 image_id_size = sizeof (__le32) + len;
3597 image_id = kmalloc(image_id_size, GFP_KERNEL);
3602 end = (char *) image_id + image_id_size;
3603 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32) len);
3605 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
3606 reply_buf = kmalloc(size, GFP_KERNEL);
3610 ret = rbd_obj_method_sync(rbd_dev, RBD_DIRECTORY,
3611 "rbd", "dir_get_name",
3612 image_id, image_id_size,
3613 (char *) reply_buf, size, NULL);
3617 end = (char *) reply_buf + size;
3618 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
3619 if (IS_ERR(image_name))
3622 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
3631 * When a parent image gets probed, we only have the pool, image,
3632 * and snapshot ids but not the names of any of them. This call
3633 * is made later to fill in those names. It has to be done after
3634 * rbd_dev_snaps_update() has completed because some of the
3635 * information (in particular, snapshot name) is not available
3638 static int rbd_dev_probe_update_spec(struct rbd_device *rbd_dev)
3640 struct ceph_osd_client *osdc;
3642 void *reply_buf = NULL;
3645 if (rbd_dev->spec->pool_name)
3646 return 0; /* Already have the names */
3648 /* Look up the pool name */
3650 osdc = &rbd_dev->rbd_client->client->osdc;
3651 name = ceph_pg_pool_name_by_id(osdc->osdmap, rbd_dev->spec->pool_id);
3653 rbd_warn(rbd_dev, "there is no pool with id %llu",
3654 rbd_dev->spec->pool_id); /* Really a BUG() */
3658 rbd_dev->spec->pool_name = kstrdup(name, GFP_KERNEL);
3659 if (!rbd_dev->spec->pool_name)
3662 /* Fetch the image name; tolerate failure here */
3664 name = rbd_dev_image_name(rbd_dev);
3666 rbd_dev->spec->image_name = (char *) name;
3668 rbd_warn(rbd_dev, "unable to get image name");
3670 /* Look up the snapshot name. */
3672 name = rbd_snap_name(rbd_dev, rbd_dev->spec->snap_id);
3674 rbd_warn(rbd_dev, "no snapshot with id %llu",
3675 rbd_dev->spec->snap_id); /* Really a BUG() */
3679 rbd_dev->spec->snap_name = kstrdup(name, GFP_KERNEL);
3680 if(!rbd_dev->spec->snap_name)
3686 kfree(rbd_dev->spec->pool_name);
3687 rbd_dev->spec->pool_name = NULL;
3692 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev, u64 *ver)
3701 struct ceph_snap_context *snapc;
3705 * We'll need room for the seq value (maximum snapshot id),
3706 * snapshot count, and array of that many snapshot ids.
3707 * For now we have a fixed upper limit on the number we're
3708 * prepared to receive.
3710 size = sizeof (__le64) + sizeof (__le32) +
3711 RBD_MAX_SNAP_COUNT * sizeof (__le64);
3712 reply_buf = kzalloc(size, GFP_KERNEL);
3716 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3717 "rbd", "get_snapcontext",
3719 reply_buf, size, ver);
3720 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3726 end = (char *) reply_buf + size;
3727 ceph_decode_64_safe(&p, end, seq, out);
3728 ceph_decode_32_safe(&p, end, snap_count, out);
3731 * Make sure the reported number of snapshot ids wouldn't go
3732 * beyond the end of our buffer. But before checking that,
3733 * make sure the computed size of the snapshot context we
3734 * allocate is representable in a size_t.
3736 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
3741 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
3744 size = sizeof (struct ceph_snap_context) +
3745 snap_count * sizeof (snapc->snaps[0]);
3746 snapc = kmalloc(size, GFP_KERNEL);
3752 atomic_set(&snapc->nref, 1);
3754 snapc->num_snaps = snap_count;
3755 for (i = 0; i < snap_count; i++)
3756 snapc->snaps[i] = ceph_decode_64(&p);
3758 rbd_dev->header.snapc = snapc;
3760 dout(" snap context seq = %llu, snap_count = %u\n",
3761 (unsigned long long) seq, (unsigned int) snap_count);
3769 static char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev, u32 which)
3779 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
3780 reply_buf = kmalloc(size, GFP_KERNEL);
3782 return ERR_PTR(-ENOMEM);
3784 snap_id = cpu_to_le64(rbd_dev->header.snapc->snaps[which]);
3785 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3786 "rbd", "get_snapshot_name",
3787 (char *) &snap_id, sizeof (snap_id),
3788 reply_buf, size, NULL);
3789 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3794 end = (char *) reply_buf + size;
3795 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
3796 if (IS_ERR(snap_name)) {
3797 ret = PTR_ERR(snap_name);
3800 dout(" snap_id 0x%016llx snap_name = %s\n",
3801 (unsigned long long) le64_to_cpu(snap_id), snap_name);
3809 return ERR_PTR(ret);
3812 static char *rbd_dev_v2_snap_info(struct rbd_device *rbd_dev, u32 which,
3813 u64 *snap_size, u64 *snap_features)
3819 snap_id = rbd_dev->header.snapc->snaps[which];
3820 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, &order, snap_size);
3822 return ERR_PTR(ret);
3823 ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, snap_features);
3825 return ERR_PTR(ret);
3827 return rbd_dev_v2_snap_name(rbd_dev, which);
3830 static char *rbd_dev_snap_info(struct rbd_device *rbd_dev, u32 which,
3831 u64 *snap_size, u64 *snap_features)
3833 if (rbd_dev->image_format == 1)
3834 return rbd_dev_v1_snap_info(rbd_dev, which,
3835 snap_size, snap_features);
3836 if (rbd_dev->image_format == 2)
3837 return rbd_dev_v2_snap_info(rbd_dev, which,
3838 snap_size, snap_features);
3839 return ERR_PTR(-EINVAL);
3842 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver)
3847 down_write(&rbd_dev->header_rwsem);
3849 /* Grab old order first, to see if it changes */
3851 obj_order = rbd_dev->header.obj_order,
3852 ret = rbd_dev_v2_image_size(rbd_dev);
3855 if (rbd_dev->header.obj_order != obj_order) {
3859 rbd_update_mapping_size(rbd_dev);
3861 ret = rbd_dev_v2_snap_context(rbd_dev, hver);
3862 dout("rbd_dev_v2_snap_context returned %d\n", ret);
3865 ret = rbd_dev_snaps_update(rbd_dev);
3866 dout("rbd_dev_snaps_update returned %d\n", ret);
3869 ret = rbd_dev_snaps_register(rbd_dev);
3870 dout("rbd_dev_snaps_register returned %d\n", ret);
3872 up_write(&rbd_dev->header_rwsem);
3878 * Scan the rbd device's current snapshot list and compare it to the
3879 * newly-received snapshot context. Remove any existing snapshots
3880 * not present in the new snapshot context. Add a new snapshot for
3881 * any snaphots in the snapshot context not in the current list.
3882 * And verify there are no changes to snapshots we already know
3885 * Assumes the snapshots in the snapshot context are sorted by
3886 * snapshot id, highest id first. (Snapshots in the rbd_dev's list
3887 * are also maintained in that order.)
3889 static int rbd_dev_snaps_update(struct rbd_device *rbd_dev)
3891 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
3892 const u32 snap_count = snapc->num_snaps;
3893 struct list_head *head = &rbd_dev->snaps;
3894 struct list_head *links = head->next;
3897 dout("%s: snap count is %u\n", __func__, (unsigned int) snap_count);
3898 while (index < snap_count || links != head) {
3900 struct rbd_snap *snap;
3903 u64 snap_features = 0;
3905 snap_id = index < snap_count ? snapc->snaps[index]
3907 snap = links != head ? list_entry(links, struct rbd_snap, node)
3909 rbd_assert(!snap || snap->id != CEPH_NOSNAP);
3911 if (snap_id == CEPH_NOSNAP || (snap && snap->id > snap_id)) {
3912 struct list_head *next = links->next;
3915 * A previously-existing snapshot is not in
3916 * the new snap context.
3918 * If the now missing snapshot is the one the
3919 * image is mapped to, clear its exists flag
3920 * so we can avoid sending any more requests
3923 if (rbd_dev->spec->snap_id == snap->id)
3924 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
3925 rbd_remove_snap_dev(snap);
3926 dout("%ssnap id %llu has been removed\n",
3927 rbd_dev->spec->snap_id == snap->id ?
3929 (unsigned long long) snap->id);
3931 /* Done with this list entry; advance */
3937 snap_name = rbd_dev_snap_info(rbd_dev, index,
3938 &snap_size, &snap_features);
3939 if (IS_ERR(snap_name))
3940 return PTR_ERR(snap_name);
3942 dout("entry %u: snap_id = %llu\n", (unsigned int) snap_count,
3943 (unsigned long long) snap_id);
3944 if (!snap || (snap_id != CEPH_NOSNAP && snap->id < snap_id)) {
3945 struct rbd_snap *new_snap;
3947 /* We haven't seen this snapshot before */
3949 new_snap = __rbd_add_snap_dev(rbd_dev, snap_name,
3950 snap_id, snap_size, snap_features);
3951 if (IS_ERR(new_snap)) {
3952 int err = PTR_ERR(new_snap);
3954 dout(" failed to add dev, error %d\n", err);
3959 /* New goes before existing, or at end of list */
3961 dout(" added dev%s\n", snap ? "" : " at end\n");
3963 list_add_tail(&new_snap->node, &snap->node);
3965 list_add_tail(&new_snap->node, head);
3967 /* Already have this one */
3969 dout(" already present\n");
3971 rbd_assert(snap->size == snap_size);
3972 rbd_assert(!strcmp(snap->name, snap_name));
3973 rbd_assert(snap->features == snap_features);
3975 /* Done with this list entry; advance */
3977 links = links->next;
3980 /* Advance to the next entry in the snapshot context */
3984 dout("%s: done\n", __func__);
3990 * Scan the list of snapshots and register the devices for any that
3991 * have not already been registered.
3993 static int rbd_dev_snaps_register(struct rbd_device *rbd_dev)
3995 struct rbd_snap *snap;
3998 dout("%s:\n", __func__);
3999 if (WARN_ON(!device_is_registered(&rbd_dev->dev)))
4002 list_for_each_entry(snap, &rbd_dev->snaps, node) {
4003 if (!rbd_snap_registered(snap)) {
4004 ret = rbd_register_snap_dev(snap, &rbd_dev->dev);
4009 dout("%s: returning %d\n", __func__, ret);
4014 static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
4019 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
4021 dev = &rbd_dev->dev;
4022 dev->bus = &rbd_bus_type;
4023 dev->type = &rbd_device_type;
4024 dev->parent = &rbd_root_dev;
4025 dev->release = rbd_dev_release;
4026 dev_set_name(dev, "%d", rbd_dev->dev_id);
4027 ret = device_register(dev);
4029 mutex_unlock(&ctl_mutex);
4034 static void rbd_bus_del_dev(struct rbd_device *rbd_dev)
4036 device_unregister(&rbd_dev->dev);
4039 static atomic64_t rbd_dev_id_max = ATOMIC64_INIT(0);
4042 * Get a unique rbd identifier for the given new rbd_dev, and add
4043 * the rbd_dev to the global list. The minimum rbd id is 1.
4045 static void rbd_dev_id_get(struct rbd_device *rbd_dev)
4047 rbd_dev->dev_id = atomic64_inc_return(&rbd_dev_id_max);
4049 spin_lock(&rbd_dev_list_lock);
4050 list_add_tail(&rbd_dev->node, &rbd_dev_list);
4051 spin_unlock(&rbd_dev_list_lock);
4052 dout("rbd_dev %p given dev id %llu\n", rbd_dev,
4053 (unsigned long long) rbd_dev->dev_id);
4057 * Remove an rbd_dev from the global list, and record that its
4058 * identifier is no longer in use.
4060 static void rbd_dev_id_put(struct rbd_device *rbd_dev)
4062 struct list_head *tmp;
4063 int rbd_id = rbd_dev->dev_id;
4066 rbd_assert(rbd_id > 0);
4068 dout("rbd_dev %p released dev id %llu\n", rbd_dev,
4069 (unsigned long long) rbd_dev->dev_id);
4070 spin_lock(&rbd_dev_list_lock);
4071 list_del_init(&rbd_dev->node);
4074 * If the id being "put" is not the current maximum, there
4075 * is nothing special we need to do.
4077 if (rbd_id != atomic64_read(&rbd_dev_id_max)) {
4078 spin_unlock(&rbd_dev_list_lock);
4083 * We need to update the current maximum id. Search the
4084 * list to find out what it is. We're more likely to find
4085 * the maximum at the end, so search the list backward.
4088 list_for_each_prev(tmp, &rbd_dev_list) {
4089 struct rbd_device *rbd_dev;
4091 rbd_dev = list_entry(tmp, struct rbd_device, node);
4092 if (rbd_dev->dev_id > max_id)
4093 max_id = rbd_dev->dev_id;
4095 spin_unlock(&rbd_dev_list_lock);
4098 * The max id could have been updated by rbd_dev_id_get(), in
4099 * which case it now accurately reflects the new maximum.
4100 * Be careful not to overwrite the maximum value in that
4103 atomic64_cmpxchg(&rbd_dev_id_max, rbd_id, max_id);
4104 dout(" max dev id has been reset\n");
4108 * Skips over white space at *buf, and updates *buf to point to the
4109 * first found non-space character (if any). Returns the length of
4110 * the token (string of non-white space characters) found. Note
4111 * that *buf must be terminated with '\0'.
4113 static inline size_t next_token(const char **buf)
4116 * These are the characters that produce nonzero for
4117 * isspace() in the "C" and "POSIX" locales.
4119 const char *spaces = " \f\n\r\t\v";
4121 *buf += strspn(*buf, spaces); /* Find start of token */
4123 return strcspn(*buf, spaces); /* Return token length */
4127 * Finds the next token in *buf, and if the provided token buffer is
4128 * big enough, copies the found token into it. The result, if
4129 * copied, is guaranteed to be terminated with '\0'. Note that *buf
4130 * must be terminated with '\0' on entry.
4132 * Returns the length of the token found (not including the '\0').
4133 * Return value will be 0 if no token is found, and it will be >=
4134 * token_size if the token would not fit.
4136 * The *buf pointer will be updated to point beyond the end of the
4137 * found token. Note that this occurs even if the token buffer is
4138 * too small to hold it.
4140 static inline size_t copy_token(const char **buf,
4146 len = next_token(buf);
4147 if (len < token_size) {
4148 memcpy(token, *buf, len);
4149 *(token + len) = '\0';
4157 * Finds the next token in *buf, dynamically allocates a buffer big
4158 * enough to hold a copy of it, and copies the token into the new
4159 * buffer. The copy is guaranteed to be terminated with '\0'. Note
4160 * that a duplicate buffer is created even for a zero-length token.
4162 * Returns a pointer to the newly-allocated duplicate, or a null
4163 * pointer if memory for the duplicate was not available. If
4164 * the lenp argument is a non-null pointer, the length of the token
4165 * (not including the '\0') is returned in *lenp.
4167 * If successful, the *buf pointer will be updated to point beyond
4168 * the end of the found token.
4170 * Note: uses GFP_KERNEL for allocation.
4172 static inline char *dup_token(const char **buf, size_t *lenp)
4177 len = next_token(buf);
4178 dup = kmemdup(*buf, len + 1, GFP_KERNEL);
4181 *(dup + len) = '\0';
4191 * Parse the options provided for an "rbd add" (i.e., rbd image
4192 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
4193 * and the data written is passed here via a NUL-terminated buffer.
4194 * Returns 0 if successful or an error code otherwise.
4196 * The information extracted from these options is recorded in
4197 * the other parameters which return dynamically-allocated
4200 * The address of a pointer that will refer to a ceph options
4201 * structure. Caller must release the returned pointer using
4202 * ceph_destroy_options() when it is no longer needed.
4204 * Address of an rbd options pointer. Fully initialized by
4205 * this function; caller must release with kfree().
4207 * Address of an rbd image specification pointer. Fully
4208 * initialized by this function based on parsed options.
4209 * Caller must release with rbd_spec_put().
4211 * The options passed take this form:
4212 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
4215 * A comma-separated list of one or more monitor addresses.
4216 * A monitor address is an ip address, optionally followed
4217 * by a port number (separated by a colon).
4218 * I.e.: ip1[:port1][,ip2[:port2]...]
4220 * A comma-separated list of ceph and/or rbd options.
4222 * The name of the rados pool containing the rbd image.
4224 * The name of the image in that pool to map.
4226 * An optional snapshot id. If provided, the mapping will
4227 * present data from the image at the time that snapshot was
4228 * created. The image head is used if no snapshot id is
4229 * provided. Snapshot mappings are always read-only.
4231 static int rbd_add_parse_args(const char *buf,
4232 struct ceph_options **ceph_opts,
4233 struct rbd_options **opts,
4234 struct rbd_spec **rbd_spec)
4238 const char *mon_addrs;
4239 size_t mon_addrs_size;
4240 struct rbd_spec *spec = NULL;
4241 struct rbd_options *rbd_opts = NULL;
4242 struct ceph_options *copts;
4245 /* The first four tokens are required */
4247 len = next_token(&buf);
4249 rbd_warn(NULL, "no monitor address(es) provided");
4253 mon_addrs_size = len + 1;
4257 options = dup_token(&buf, NULL);
4261 rbd_warn(NULL, "no options provided");
4265 spec = rbd_spec_alloc();
4269 spec->pool_name = dup_token(&buf, NULL);
4270 if (!spec->pool_name)
4272 if (!*spec->pool_name) {
4273 rbd_warn(NULL, "no pool name provided");
4277 spec->image_name = dup_token(&buf, NULL);
4278 if (!spec->image_name)
4280 if (!*spec->image_name) {
4281 rbd_warn(NULL, "no image name provided");
4286 * Snapshot name is optional; default is to use "-"
4287 * (indicating the head/no snapshot).
4289 len = next_token(&buf);
4291 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
4292 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
4293 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
4294 ret = -ENAMETOOLONG;
4297 spec->snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
4298 if (!spec->snap_name)
4300 *(spec->snap_name + len) = '\0';
4302 /* Initialize all rbd options to the defaults */
4304 rbd_opts = kzalloc(sizeof (*rbd_opts), GFP_KERNEL);
4308 rbd_opts->read_only = RBD_READ_ONLY_DEFAULT;
4310 copts = ceph_parse_options(options, mon_addrs,
4311 mon_addrs + mon_addrs_size - 1,
4312 parse_rbd_opts_token, rbd_opts);
4313 if (IS_ERR(copts)) {
4314 ret = PTR_ERR(copts);
4335 * An rbd format 2 image has a unique identifier, distinct from the
4336 * name given to it by the user. Internally, that identifier is
4337 * what's used to specify the names of objects related to the image.
4339 * A special "rbd id" object is used to map an rbd image name to its
4340 * id. If that object doesn't exist, then there is no v2 rbd image
4341 * with the supplied name.
4343 * This function will record the given rbd_dev's image_id field if
4344 * it can be determined, and in that case will return 0. If any
4345 * errors occur a negative errno will be returned and the rbd_dev's
4346 * image_id field will be unchanged (and should be NULL).
4348 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
4356 /* If we already have it we don't need to look it up */
4358 if (rbd_dev->spec->image_id)
4362 * When probing a parent image, the image id is already
4363 * known (and the image name likely is not). There's no
4364 * need to fetch the image id again in this case.
4366 if (rbd_dev->spec->image_id)
4370 * First, see if the format 2 image id file exists, and if
4371 * so, get the image's persistent id from it.
4373 size = sizeof (RBD_ID_PREFIX) + strlen(rbd_dev->spec->image_name);
4374 object_name = kmalloc(size, GFP_NOIO);
4377 sprintf(object_name, "%s%s", RBD_ID_PREFIX, rbd_dev->spec->image_name);
4378 dout("rbd id object name is %s\n", object_name);
4380 /* Response will be an encoded string, which includes a length */
4382 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
4383 response = kzalloc(size, GFP_NOIO);
4389 ret = rbd_obj_method_sync(rbd_dev, object_name,
4392 response, RBD_IMAGE_ID_LEN_MAX, NULL);
4393 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4398 rbd_dev->spec->image_id = ceph_extract_encoded_string(&p,
4399 p + RBD_IMAGE_ID_LEN_MAX,
4401 if (IS_ERR(rbd_dev->spec->image_id)) {
4402 ret = PTR_ERR(rbd_dev->spec->image_id);
4403 rbd_dev->spec->image_id = NULL;
4405 dout("image_id is %s\n", rbd_dev->spec->image_id);
4414 static int rbd_dev_v1_probe(struct rbd_device *rbd_dev)
4419 /* Version 1 images have no id; empty string is used */
4421 rbd_dev->spec->image_id = kstrdup("", GFP_KERNEL);
4422 if (!rbd_dev->spec->image_id)
4425 /* Record the header object name for this rbd image. */
4427 size = strlen(rbd_dev->spec->image_name) + sizeof (RBD_SUFFIX);
4428 rbd_dev->header_name = kmalloc(size, GFP_KERNEL);
4429 if (!rbd_dev->header_name) {
4433 sprintf(rbd_dev->header_name, "%s%s",
4434 rbd_dev->spec->image_name, RBD_SUFFIX);
4436 /* Populate rbd image metadata */
4438 ret = rbd_read_header(rbd_dev, &rbd_dev->header);
4442 /* Version 1 images have no parent (no layering) */
4444 rbd_dev->parent_spec = NULL;
4445 rbd_dev->parent_overlap = 0;
4447 rbd_dev->image_format = 1;
4449 dout("discovered version 1 image, header name is %s\n",
4450 rbd_dev->header_name);
4455 kfree(rbd_dev->header_name);
4456 rbd_dev->header_name = NULL;
4457 kfree(rbd_dev->spec->image_id);
4458 rbd_dev->spec->image_id = NULL;
4463 static int rbd_dev_v2_probe(struct rbd_device *rbd_dev)
4470 * Image id was filled in by the caller. Record the header
4471 * object name for this rbd image.
4473 size = sizeof (RBD_HEADER_PREFIX) + strlen(rbd_dev->spec->image_id);
4474 rbd_dev->header_name = kmalloc(size, GFP_KERNEL);
4475 if (!rbd_dev->header_name)
4477 sprintf(rbd_dev->header_name, "%s%s",
4478 RBD_HEADER_PREFIX, rbd_dev->spec->image_id);
4480 /* Get the size and object order for the image */
4482 ret = rbd_dev_v2_image_size(rbd_dev);
4486 /* Get the object prefix (a.k.a. block_name) for the image */
4488 ret = rbd_dev_v2_object_prefix(rbd_dev);
4492 /* Get the and check features for the image */
4494 ret = rbd_dev_v2_features(rbd_dev);
4498 /* If the image supports layering, get the parent info */
4500 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
4501 ret = rbd_dev_v2_parent_info(rbd_dev);
4506 /* crypto and compression type aren't (yet) supported for v2 images */
4508 rbd_dev->header.crypt_type = 0;
4509 rbd_dev->header.comp_type = 0;
4511 /* Get the snapshot context, plus the header version */
4513 ret = rbd_dev_v2_snap_context(rbd_dev, &ver);
4516 rbd_dev->header.obj_version = ver;
4518 rbd_dev->image_format = 2;
4520 dout("discovered version 2 image, header name is %s\n",
4521 rbd_dev->header_name);
4525 rbd_dev->parent_overlap = 0;
4526 rbd_spec_put(rbd_dev->parent_spec);
4527 rbd_dev->parent_spec = NULL;
4528 kfree(rbd_dev->header_name);
4529 rbd_dev->header_name = NULL;
4530 kfree(rbd_dev->header.object_prefix);
4531 rbd_dev->header.object_prefix = NULL;
4536 static int rbd_dev_probe_finish(struct rbd_device *rbd_dev)
4538 struct rbd_device *parent = NULL;
4539 struct rbd_spec *parent_spec = NULL;
4540 struct rbd_client *rbdc = NULL;
4543 /* no need to lock here, as rbd_dev is not registered yet */
4544 ret = rbd_dev_snaps_update(rbd_dev);
4548 ret = rbd_dev_probe_update_spec(rbd_dev);
4552 ret = rbd_dev_set_mapping(rbd_dev);
4556 /* generate unique id: find highest unique id, add one */
4557 rbd_dev_id_get(rbd_dev);
4559 /* Fill in the device name, now that we have its id. */
4560 BUILD_BUG_ON(DEV_NAME_LEN
4561 < sizeof (RBD_DRV_NAME) + MAX_INT_FORMAT_WIDTH);
4562 sprintf(rbd_dev->name, "%s%d", RBD_DRV_NAME, rbd_dev->dev_id);
4564 /* Get our block major device number. */
4566 ret = register_blkdev(0, rbd_dev->name);
4569 rbd_dev->major = ret;
4571 /* Set up the blkdev mapping. */
4573 ret = rbd_init_disk(rbd_dev);
4575 goto err_out_blkdev;
4577 ret = rbd_bus_add_dev(rbd_dev);
4582 * At this point cleanup in the event of an error is the job
4583 * of the sysfs code (initiated by rbd_bus_del_dev()).
4585 /* Probe the parent if there is one */
4587 if (rbd_dev->parent_spec) {
4589 * We need to pass a reference to the client and the
4590 * parent spec when creating the parent rbd_dev.
4591 * Images related by parent/child relationships
4592 * always share both.
4594 parent_spec = rbd_spec_get(rbd_dev->parent_spec);
4595 rbdc = __rbd_get_client(rbd_dev->rbd_client);
4597 parent = rbd_dev_create(rbdc, parent_spec);
4602 rbdc = NULL; /* parent now owns reference */
4603 parent_spec = NULL; /* parent now owns reference */
4604 ret = rbd_dev_probe(parent);
4606 goto err_out_parent;
4607 rbd_dev->parent = parent;
4610 down_write(&rbd_dev->header_rwsem);
4611 ret = rbd_dev_snaps_register(rbd_dev);
4612 up_write(&rbd_dev->header_rwsem);
4616 ret = rbd_dev_header_watch_sync(rbd_dev, 1);
4620 /* Everything's ready. Announce the disk to the world. */
4622 add_disk(rbd_dev->disk);
4624 pr_info("%s: added with size 0x%llx\n", rbd_dev->disk->disk_name,
4625 (unsigned long long) rbd_dev->mapping.size);
4630 rbd_dev_destroy(parent);
4632 rbd_spec_put(parent_spec);
4633 rbd_put_client(rbdc);
4635 /* this will also clean up rest of rbd_dev stuff */
4637 rbd_bus_del_dev(rbd_dev);
4641 rbd_free_disk(rbd_dev);
4643 unregister_blkdev(rbd_dev->major, rbd_dev->name);
4645 rbd_dev_id_put(rbd_dev);
4647 rbd_remove_all_snaps(rbd_dev);
4653 * Probe for the existence of the header object for the given rbd
4654 * device. For format 2 images this includes determining the image
4657 static int rbd_dev_probe(struct rbd_device *rbd_dev)
4662 * Get the id from the image id object. If it's not a
4663 * format 2 image, we'll get ENOENT back, and we'll assume
4664 * it's a format 1 image.
4666 ret = rbd_dev_image_id(rbd_dev);
4668 ret = rbd_dev_v1_probe(rbd_dev);
4670 ret = rbd_dev_v2_probe(rbd_dev);
4672 dout("probe failed, returning %d\n", ret);
4677 ret = rbd_dev_probe_finish(rbd_dev);
4679 rbd_header_free(&rbd_dev->header);
4684 static ssize_t rbd_add(struct bus_type *bus,
4688 struct rbd_device *rbd_dev = NULL;
4689 struct ceph_options *ceph_opts = NULL;
4690 struct rbd_options *rbd_opts = NULL;
4691 struct rbd_spec *spec = NULL;
4692 struct rbd_client *rbdc;
4693 struct ceph_osd_client *osdc;
4696 if (!try_module_get(THIS_MODULE))
4699 /* parse add command */
4700 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
4702 goto err_out_module;
4704 rbdc = rbd_get_client(ceph_opts);
4709 ceph_opts = NULL; /* rbd_dev client now owns this */
4712 osdc = &rbdc->client->osdc;
4713 rc = ceph_pg_poolid_by_name(osdc->osdmap, spec->pool_name);
4715 goto err_out_client;
4716 spec->pool_id = (u64) rc;
4718 /* The ceph file layout needs to fit pool id in 32 bits */
4720 if (WARN_ON(spec->pool_id > (u64) U32_MAX)) {
4722 goto err_out_client;
4725 rbd_dev = rbd_dev_create(rbdc, spec);
4727 goto err_out_client;
4728 rbdc = NULL; /* rbd_dev now owns this */
4729 spec = NULL; /* rbd_dev now owns this */
4731 rbd_dev->mapping.read_only = rbd_opts->read_only;
4733 rbd_opts = NULL; /* done with this */
4735 rc = rbd_dev_probe(rbd_dev);
4737 goto err_out_rbd_dev;
4741 rbd_dev_destroy(rbd_dev);
4743 rbd_put_client(rbdc);
4746 ceph_destroy_options(ceph_opts);
4750 module_put(THIS_MODULE);
4752 dout("Error adding device %s\n", buf);
4754 return (ssize_t) rc;
4757 static struct rbd_device *__rbd_get_dev(unsigned long dev_id)
4759 struct list_head *tmp;
4760 struct rbd_device *rbd_dev;
4762 spin_lock(&rbd_dev_list_lock);
4763 list_for_each(tmp, &rbd_dev_list) {
4764 rbd_dev = list_entry(tmp, struct rbd_device, node);
4765 if (rbd_dev->dev_id == dev_id) {
4766 spin_unlock(&rbd_dev_list_lock);
4770 spin_unlock(&rbd_dev_list_lock);
4774 static void rbd_dev_release(struct device *dev)
4776 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4778 if (rbd_dev->watch_event)
4779 rbd_dev_header_watch_sync(rbd_dev, 0);
4781 /* clean up and free blkdev */
4782 rbd_free_disk(rbd_dev);
4783 unregister_blkdev(rbd_dev->major, rbd_dev->name);
4785 /* release allocated disk header fields */
4786 rbd_header_free(&rbd_dev->header);
4788 /* done with the id, and with the rbd_dev */
4789 rbd_dev_id_put(rbd_dev);
4790 rbd_assert(rbd_dev->rbd_client != NULL);
4791 rbd_dev_destroy(rbd_dev);
4793 /* release module ref */
4794 module_put(THIS_MODULE);
4797 static void __rbd_remove(struct rbd_device *rbd_dev)
4799 rbd_remove_all_snaps(rbd_dev);
4800 rbd_bus_del_dev(rbd_dev);
4803 static ssize_t rbd_remove(struct bus_type *bus,
4807 struct rbd_device *rbd_dev = NULL;
4812 rc = strict_strtoul(buf, 10, &ul);
4816 /* convert to int; abort if we lost anything in the conversion */
4817 target_id = (int) ul;
4818 if (target_id != ul)
4821 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
4823 rbd_dev = __rbd_get_dev(target_id);
4829 spin_lock_irq(&rbd_dev->lock);
4830 if (rbd_dev->open_count)
4833 set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
4834 spin_unlock_irq(&rbd_dev->lock);
4838 while (rbd_dev->parent_spec) {
4839 struct rbd_device *first = rbd_dev;
4840 struct rbd_device *second = first->parent;
4841 struct rbd_device *third;
4844 * Follow to the parent with no grandparent and
4847 while (second && (third = second->parent)) {
4851 __rbd_remove(second);
4852 rbd_spec_put(first->parent_spec);
4853 first->parent_spec = NULL;
4854 first->parent_overlap = 0;
4855 first->parent = NULL;
4857 __rbd_remove(rbd_dev);
4860 mutex_unlock(&ctl_mutex);
4866 * create control files in sysfs
4869 static int rbd_sysfs_init(void)
4873 ret = device_register(&rbd_root_dev);
4877 ret = bus_register(&rbd_bus_type);
4879 device_unregister(&rbd_root_dev);
4884 static void rbd_sysfs_cleanup(void)
4886 bus_unregister(&rbd_bus_type);
4887 device_unregister(&rbd_root_dev);
4890 static int __init rbd_init(void)
4894 if (!libceph_compatible(NULL)) {
4895 rbd_warn(NULL, "libceph incompatibility (quitting)");
4899 rc = rbd_sysfs_init();
4902 pr_info("loaded " RBD_DRV_NAME_LONG "\n");
4906 static void __exit rbd_exit(void)
4908 rbd_sysfs_cleanup();
4911 module_init(rbd_init);
4912 module_exit(rbd_exit);
4914 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
4915 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
4916 MODULE_DESCRIPTION("rados block device");
4918 /* following authorship retained from original osdblk.c */
4919 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
4921 MODULE_LICENSE("GPL");