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 (RBD_FEATURES_ALL)
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).
141 const char *pool_name;
143 const char *image_id;
144 const char *image_name;
147 const char *snap_name;
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;
221 struct page **copyup_pages;
223 struct ceph_osd_request *osd_req;
225 u64 xferred; /* bytes transferred */
229 rbd_obj_callback_t callback;
230 struct completion completion;
236 IMG_REQ_WRITE, /* I/O direction: read = 0, write = 1 */
237 IMG_REQ_CHILD, /* initiator: block = 0, child image = 1 */
238 IMG_REQ_LAYERED, /* ENOENT handling: normal = 0, layered = 1 */
241 struct rbd_img_request {
242 struct rbd_device *rbd_dev;
243 u64 offset; /* starting image byte offset */
244 u64 length; /* byte count from offset */
247 u64 snap_id; /* for reads */
248 struct ceph_snap_context *snapc; /* for writes */
251 struct request *rq; /* block request */
252 struct rbd_obj_request *obj_request; /* obj req initiator */
254 struct page **copyup_pages;
255 spinlock_t completion_lock;/* protects next_completion */
257 rbd_img_callback_t callback;
258 u64 xferred;/* aggregate bytes transferred */
259 int result; /* first nonzero obj_request result */
261 u32 obj_request_count;
262 struct list_head obj_requests; /* rbd_obj_request structs */
267 #define for_each_obj_request(ireq, oreq) \
268 list_for_each_entry(oreq, &(ireq)->obj_requests, links)
269 #define for_each_obj_request_from(ireq, oreq) \
270 list_for_each_entry_from(oreq, &(ireq)->obj_requests, links)
271 #define for_each_obj_request_safe(ireq, oreq, n) \
272 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;
322 /* protects updating the header */
323 struct rw_semaphore header_rwsem;
325 struct rbd_mapping mapping;
327 struct list_head node;
329 /* list of snapshots */
330 struct list_head snaps;
334 unsigned long open_count; /* protected by lock */
338 * Flag bits for rbd_dev->flags. If atomicity is required,
339 * rbd_dev->lock is used to protect access.
341 * Currently, only the "removing" flag (which is coupled with the
342 * "open_count" field) requires atomic access.
345 RBD_DEV_FLAG_EXISTS, /* mapped snapshot has not been deleted */
346 RBD_DEV_FLAG_REMOVING, /* this mapping is being removed */
349 static DEFINE_MUTEX(ctl_mutex); /* Serialize open/close/setup/teardown */
351 static LIST_HEAD(rbd_dev_list); /* devices */
352 static DEFINE_SPINLOCK(rbd_dev_list_lock);
354 static LIST_HEAD(rbd_client_list); /* clients */
355 static DEFINE_SPINLOCK(rbd_client_list_lock);
357 static int rbd_img_request_submit(struct rbd_img_request *img_request);
359 static int rbd_dev_snaps_update(struct rbd_device *rbd_dev);
361 static void rbd_dev_release(struct device *dev);
362 static void rbd_snap_destroy(struct rbd_snap *snap);
364 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
366 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
368 static int rbd_dev_probe(struct rbd_device *rbd_dev);
370 static struct bus_attribute rbd_bus_attrs[] = {
371 __ATTR(add, S_IWUSR, NULL, rbd_add),
372 __ATTR(remove, S_IWUSR, NULL, rbd_remove),
376 static struct bus_type rbd_bus_type = {
378 .bus_attrs = rbd_bus_attrs,
381 static void rbd_root_dev_release(struct device *dev)
385 static struct device rbd_root_dev = {
387 .release = rbd_root_dev_release,
390 static __printf(2, 3)
391 void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
393 struct va_format vaf;
401 printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
402 else if (rbd_dev->disk)
403 printk(KERN_WARNING "%s: %s: %pV\n",
404 RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
405 else if (rbd_dev->spec && rbd_dev->spec->image_name)
406 printk(KERN_WARNING "%s: image %s: %pV\n",
407 RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
408 else if (rbd_dev->spec && rbd_dev->spec->image_id)
409 printk(KERN_WARNING "%s: id %s: %pV\n",
410 RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
412 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
413 RBD_DRV_NAME, rbd_dev, &vaf);
418 #define rbd_assert(expr) \
419 if (unlikely(!(expr))) { \
420 printk(KERN_ERR "\nAssertion failure in %s() " \
422 "\trbd_assert(%s);\n\n", \
423 __func__, __LINE__, #expr); \
426 #else /* !RBD_DEBUG */
427 # define rbd_assert(expr) ((void) 0)
428 #endif /* !RBD_DEBUG */
430 static void rbd_img_parent_read(struct rbd_obj_request *obj_request);
431 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request);
433 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver);
434 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver);
436 static int rbd_open(struct block_device *bdev, fmode_t mode)
438 struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
439 bool removing = false;
441 if ((mode & FMODE_WRITE) && rbd_dev->mapping.read_only)
444 spin_lock_irq(&rbd_dev->lock);
445 if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
448 rbd_dev->open_count++;
449 spin_unlock_irq(&rbd_dev->lock);
453 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
454 (void) get_device(&rbd_dev->dev);
455 set_device_ro(bdev, rbd_dev->mapping.read_only);
456 mutex_unlock(&ctl_mutex);
461 static int rbd_release(struct gendisk *disk, fmode_t mode)
463 struct rbd_device *rbd_dev = disk->private_data;
464 unsigned long open_count_before;
466 spin_lock_irq(&rbd_dev->lock);
467 open_count_before = rbd_dev->open_count--;
468 spin_unlock_irq(&rbd_dev->lock);
469 rbd_assert(open_count_before > 0);
471 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
472 put_device(&rbd_dev->dev);
473 mutex_unlock(&ctl_mutex);
478 static const struct block_device_operations rbd_bd_ops = {
479 .owner = THIS_MODULE,
481 .release = rbd_release,
485 * Initialize an rbd client instance.
488 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
490 struct rbd_client *rbdc;
493 dout("%s:\n", __func__);
494 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
498 kref_init(&rbdc->kref);
499 INIT_LIST_HEAD(&rbdc->node);
501 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
503 rbdc->client = ceph_create_client(ceph_opts, rbdc, 0, 0);
504 if (IS_ERR(rbdc->client))
506 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
508 ret = ceph_open_session(rbdc->client);
512 spin_lock(&rbd_client_list_lock);
513 list_add_tail(&rbdc->node, &rbd_client_list);
514 spin_unlock(&rbd_client_list_lock);
516 mutex_unlock(&ctl_mutex);
517 dout("%s: rbdc %p\n", __func__, rbdc);
522 ceph_destroy_client(rbdc->client);
524 mutex_unlock(&ctl_mutex);
528 ceph_destroy_options(ceph_opts);
529 dout("%s: error %d\n", __func__, ret);
534 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
536 kref_get(&rbdc->kref);
542 * Find a ceph client with specific addr and configuration. If
543 * found, bump its reference count.
545 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
547 struct rbd_client *client_node;
550 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
553 spin_lock(&rbd_client_list_lock);
554 list_for_each_entry(client_node, &rbd_client_list, node) {
555 if (!ceph_compare_options(ceph_opts, client_node->client)) {
556 __rbd_get_client(client_node);
562 spin_unlock(&rbd_client_list_lock);
564 return found ? client_node : NULL;
574 /* string args above */
577 /* Boolean args above */
581 static match_table_t rbd_opts_tokens = {
583 /* string args above */
584 {Opt_read_only, "read_only"},
585 {Opt_read_only, "ro"}, /* Alternate spelling */
586 {Opt_read_write, "read_write"},
587 {Opt_read_write, "rw"}, /* Alternate spelling */
588 /* Boolean args above */
596 #define RBD_READ_ONLY_DEFAULT false
598 static int parse_rbd_opts_token(char *c, void *private)
600 struct rbd_options *rbd_opts = private;
601 substring_t argstr[MAX_OPT_ARGS];
602 int token, intval, ret;
604 token = match_token(c, rbd_opts_tokens, argstr);
608 if (token < Opt_last_int) {
609 ret = match_int(&argstr[0], &intval);
611 pr_err("bad mount option arg (not int) "
615 dout("got int token %d val %d\n", token, intval);
616 } else if (token > Opt_last_int && token < Opt_last_string) {
617 dout("got string token %d val %s\n", token,
619 } else if (token > Opt_last_string && token < Opt_last_bool) {
620 dout("got Boolean token %d\n", token);
622 dout("got token %d\n", token);
627 rbd_opts->read_only = true;
630 rbd_opts->read_only = false;
640 * Get a ceph client with specific addr and configuration, if one does
641 * not exist create it.
643 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
645 struct rbd_client *rbdc;
647 rbdc = rbd_client_find(ceph_opts);
648 if (rbdc) /* using an existing client */
649 ceph_destroy_options(ceph_opts);
651 rbdc = rbd_client_create(ceph_opts);
657 * Destroy ceph client
659 * Caller must hold rbd_client_list_lock.
661 static void rbd_client_release(struct kref *kref)
663 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
665 dout("%s: rbdc %p\n", __func__, rbdc);
666 spin_lock(&rbd_client_list_lock);
667 list_del(&rbdc->node);
668 spin_unlock(&rbd_client_list_lock);
670 ceph_destroy_client(rbdc->client);
675 * Drop reference to ceph client node. If it's not referenced anymore, release
678 static void rbd_put_client(struct rbd_client *rbdc)
681 kref_put(&rbdc->kref, rbd_client_release);
684 static bool rbd_image_format_valid(u32 image_format)
686 return image_format == 1 || image_format == 2;
689 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
694 /* The header has to start with the magic rbd header text */
695 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
698 /* The bio layer requires at least sector-sized I/O */
700 if (ondisk->options.order < SECTOR_SHIFT)
703 /* If we use u64 in a few spots we may be able to loosen this */
705 if (ondisk->options.order > 8 * sizeof (int) - 1)
709 * The size of a snapshot header has to fit in a size_t, and
710 * that limits the number of snapshots.
712 snap_count = le32_to_cpu(ondisk->snap_count);
713 size = SIZE_MAX - sizeof (struct ceph_snap_context);
714 if (snap_count > size / sizeof (__le64))
718 * Not only that, but the size of the entire the snapshot
719 * header must also be representable in a size_t.
721 size -= snap_count * sizeof (__le64);
722 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
729 * Create a new header structure, translate header format from the on-disk
732 static int rbd_header_from_disk(struct rbd_image_header *header,
733 struct rbd_image_header_ondisk *ondisk)
740 memset(header, 0, sizeof (*header));
742 snap_count = le32_to_cpu(ondisk->snap_count);
744 len = strnlen(ondisk->object_prefix, sizeof (ondisk->object_prefix));
745 header->object_prefix = kmalloc(len + 1, GFP_KERNEL);
746 if (!header->object_prefix)
748 memcpy(header->object_prefix, ondisk->object_prefix, len);
749 header->object_prefix[len] = '\0';
752 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
754 /* Save a copy of the snapshot names */
756 if (snap_names_len > (u64) SIZE_MAX)
758 header->snap_names = kmalloc(snap_names_len, GFP_KERNEL);
759 if (!header->snap_names)
762 * Note that rbd_dev_v1_header_read() guarantees
763 * the ondisk buffer we're working with has
764 * snap_names_len bytes beyond the end of the
765 * snapshot id array, this memcpy() is safe.
767 memcpy(header->snap_names, &ondisk->snaps[snap_count],
770 /* Record each snapshot's size */
772 size = snap_count * sizeof (*header->snap_sizes);
773 header->snap_sizes = kmalloc(size, GFP_KERNEL);
774 if (!header->snap_sizes)
776 for (i = 0; i < snap_count; i++)
777 header->snap_sizes[i] =
778 le64_to_cpu(ondisk->snaps[i].image_size);
780 WARN_ON(ondisk->snap_names_len);
781 header->snap_names = NULL;
782 header->snap_sizes = NULL;
785 header->features = 0; /* No features support in v1 images */
786 header->obj_order = ondisk->options.order;
787 header->crypt_type = ondisk->options.crypt_type;
788 header->comp_type = ondisk->options.comp_type;
790 /* Allocate and fill in the snapshot context */
792 header->image_size = le64_to_cpu(ondisk->image_size);
793 size = sizeof (struct ceph_snap_context);
794 size += snap_count * sizeof (header->snapc->snaps[0]);
795 header->snapc = kzalloc(size, GFP_KERNEL);
799 atomic_set(&header->snapc->nref, 1);
800 header->snapc->seq = le64_to_cpu(ondisk->snap_seq);
801 header->snapc->num_snaps = snap_count;
802 for (i = 0; i < snap_count; i++)
803 header->snapc->snaps[i] =
804 le64_to_cpu(ondisk->snaps[i].id);
809 kfree(header->snap_sizes);
810 header->snap_sizes = NULL;
811 kfree(header->snap_names);
812 header->snap_names = NULL;
813 kfree(header->object_prefix);
814 header->object_prefix = NULL;
819 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
821 struct rbd_snap *snap;
823 if (snap_id == CEPH_NOSNAP)
824 return RBD_SNAP_HEAD_NAME;
826 list_for_each_entry(snap, &rbd_dev->snaps, node)
827 if (snap_id == snap->id)
833 static struct rbd_snap *snap_by_name(struct rbd_device *rbd_dev,
834 const char *snap_name)
836 struct rbd_snap *snap;
838 list_for_each_entry(snap, &rbd_dev->snaps, node)
839 if (!strcmp(snap_name, snap->name))
845 static int rbd_dev_set_mapping(struct rbd_device *rbd_dev)
847 if (!memcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME,
848 sizeof (RBD_SNAP_HEAD_NAME))) {
849 rbd_dev->mapping.size = rbd_dev->header.image_size;
850 rbd_dev->mapping.features = rbd_dev->header.features;
852 struct rbd_snap *snap;
854 snap = snap_by_name(rbd_dev, rbd_dev->spec->snap_name);
857 rbd_dev->mapping.size = snap->size;
858 rbd_dev->mapping.features = snap->features;
859 rbd_dev->mapping.read_only = true;
861 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
866 static void rbd_header_free(struct rbd_image_header *header)
868 kfree(header->object_prefix);
869 header->object_prefix = NULL;
870 kfree(header->snap_sizes);
871 header->snap_sizes = NULL;
872 kfree(header->snap_names);
873 header->snap_names = NULL;
874 ceph_put_snap_context(header->snapc);
875 header->snapc = NULL;
878 static const char *rbd_segment_name(struct rbd_device *rbd_dev, u64 offset)
884 name = kmalloc(MAX_OBJ_NAME_SIZE + 1, GFP_NOIO);
887 segment = offset >> rbd_dev->header.obj_order;
888 ret = snprintf(name, MAX_OBJ_NAME_SIZE + 1, "%s.%012llx",
889 rbd_dev->header.object_prefix, segment);
890 if (ret < 0 || ret > MAX_OBJ_NAME_SIZE) {
891 pr_err("error formatting segment name for #%llu (%d)\n",
900 static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset)
902 u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
904 return offset & (segment_size - 1);
907 static u64 rbd_segment_length(struct rbd_device *rbd_dev,
908 u64 offset, u64 length)
910 u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
912 offset &= segment_size - 1;
914 rbd_assert(length <= U64_MAX - offset);
915 if (offset + length > segment_size)
916 length = segment_size - offset;
922 * returns the size of an object in the image
924 static u64 rbd_obj_bytes(struct rbd_image_header *header)
926 return 1 << header->obj_order;
933 static void bio_chain_put(struct bio *chain)
939 chain = chain->bi_next;
945 * zeros a bio chain, starting at specific offset
947 static void zero_bio_chain(struct bio *chain, int start_ofs)
956 bio_for_each_segment(bv, chain, i) {
957 if (pos + bv->bv_len > start_ofs) {
958 int remainder = max(start_ofs - pos, 0);
959 buf = bvec_kmap_irq(bv, &flags);
960 memset(buf + remainder, 0,
961 bv->bv_len - remainder);
962 bvec_kunmap_irq(buf, &flags);
967 chain = chain->bi_next;
972 * similar to zero_bio_chain(), zeros data defined by a page array,
973 * starting at the given byte offset from the start of the array and
974 * continuing up to the given end offset. The pages array is
975 * assumed to be big enough to hold all bytes up to the end.
977 static void zero_pages(struct page **pages, u64 offset, u64 end)
979 struct page **page = &pages[offset >> PAGE_SHIFT];
981 rbd_assert(end > offset);
982 rbd_assert(end - offset <= (u64)SIZE_MAX);
983 while (offset < end) {
989 page_offset = (size_t)(offset & ~PAGE_MASK);
990 length = min(PAGE_SIZE - page_offset, (size_t)(end - offset));
991 local_irq_save(flags);
992 kaddr = kmap_atomic(*page);
993 memset(kaddr + page_offset, 0, length);
994 kunmap_atomic(kaddr);
995 local_irq_restore(flags);
1003 * Clone a portion of a bio, starting at the given byte offset
1004 * and continuing for the number of bytes indicated.
1006 static struct bio *bio_clone_range(struct bio *bio_src,
1007 unsigned int offset,
1015 unsigned short end_idx;
1016 unsigned short vcnt;
1019 /* Handle the easy case for the caller */
1021 if (!offset && len == bio_src->bi_size)
1022 return bio_clone(bio_src, gfpmask);
1024 if (WARN_ON_ONCE(!len))
1026 if (WARN_ON_ONCE(len > bio_src->bi_size))
1028 if (WARN_ON_ONCE(offset > bio_src->bi_size - len))
1031 /* Find first affected segment... */
1034 __bio_for_each_segment(bv, bio_src, idx, 0) {
1035 if (resid < bv->bv_len)
1037 resid -= bv->bv_len;
1041 /* ...and the last affected segment */
1044 __bio_for_each_segment(bv, bio_src, end_idx, idx) {
1045 if (resid <= bv->bv_len)
1047 resid -= bv->bv_len;
1049 vcnt = end_idx - idx + 1;
1051 /* Build the clone */
1053 bio = bio_alloc(gfpmask, (unsigned int) vcnt);
1055 return NULL; /* ENOMEM */
1057 bio->bi_bdev = bio_src->bi_bdev;
1058 bio->bi_sector = bio_src->bi_sector + (offset >> SECTOR_SHIFT);
1059 bio->bi_rw = bio_src->bi_rw;
1060 bio->bi_flags |= 1 << BIO_CLONED;
1063 * Copy over our part of the bio_vec, then update the first
1064 * and last (or only) entries.
1066 memcpy(&bio->bi_io_vec[0], &bio_src->bi_io_vec[idx],
1067 vcnt * sizeof (struct bio_vec));
1068 bio->bi_io_vec[0].bv_offset += voff;
1070 bio->bi_io_vec[0].bv_len -= voff;
1071 bio->bi_io_vec[vcnt - 1].bv_len = resid;
1073 bio->bi_io_vec[0].bv_len = len;
1076 bio->bi_vcnt = vcnt;
1084 * Clone a portion of a bio chain, starting at the given byte offset
1085 * into the first bio in the source chain and continuing for the
1086 * number of bytes indicated. The result is another bio chain of
1087 * exactly the given length, or a null pointer on error.
1089 * The bio_src and offset parameters are both in-out. On entry they
1090 * refer to the first source bio and the offset into that bio where
1091 * the start of data to be cloned is located.
1093 * On return, bio_src is updated to refer to the bio in the source
1094 * chain that contains first un-cloned byte, and *offset will
1095 * contain the offset of that byte within that bio.
1097 static struct bio *bio_chain_clone_range(struct bio **bio_src,
1098 unsigned int *offset,
1102 struct bio *bi = *bio_src;
1103 unsigned int off = *offset;
1104 struct bio *chain = NULL;
1107 /* Build up a chain of clone bios up to the limit */
1109 if (!bi || off >= bi->bi_size || !len)
1110 return NULL; /* Nothing to clone */
1114 unsigned int bi_size;
1118 rbd_warn(NULL, "bio_chain exhausted with %u left", len);
1119 goto out_err; /* EINVAL; ran out of bio's */
1121 bi_size = min_t(unsigned int, bi->bi_size - off, len);
1122 bio = bio_clone_range(bi, off, bi_size, gfpmask);
1124 goto out_err; /* ENOMEM */
1127 end = &bio->bi_next;
1130 if (off == bi->bi_size) {
1141 bio_chain_put(chain);
1147 * The default/initial value for all object request flags is 0. For
1148 * each flag, once its value is set to 1 it is never reset to 0
1151 static void obj_request_img_data_set(struct rbd_obj_request *obj_request)
1153 if (test_and_set_bit(OBJ_REQ_IMG_DATA, &obj_request->flags)) {
1154 struct rbd_device *rbd_dev;
1156 rbd_dev = obj_request->img_request->rbd_dev;
1157 rbd_warn(rbd_dev, "obj_request %p already marked img_data\n",
1162 static bool obj_request_img_data_test(struct rbd_obj_request *obj_request)
1165 return test_bit(OBJ_REQ_IMG_DATA, &obj_request->flags) != 0;
1168 static void obj_request_done_set(struct rbd_obj_request *obj_request)
1170 if (test_and_set_bit(OBJ_REQ_DONE, &obj_request->flags)) {
1171 struct rbd_device *rbd_dev = NULL;
1173 if (obj_request_img_data_test(obj_request))
1174 rbd_dev = obj_request->img_request->rbd_dev;
1175 rbd_warn(rbd_dev, "obj_request %p already marked done\n",
1180 static bool obj_request_done_test(struct rbd_obj_request *obj_request)
1183 return test_bit(OBJ_REQ_DONE, &obj_request->flags) != 0;
1187 * This sets the KNOWN flag after (possibly) setting the EXISTS
1188 * flag. The latter is set based on the "exists" value provided.
1190 * Note that for our purposes once an object exists it never goes
1191 * away again. It's possible that the response from two existence
1192 * checks are separated by the creation of the target object, and
1193 * the first ("doesn't exist") response arrives *after* the second
1194 * ("does exist"). In that case we ignore the second one.
1196 static void obj_request_existence_set(struct rbd_obj_request *obj_request,
1200 set_bit(OBJ_REQ_EXISTS, &obj_request->flags);
1201 set_bit(OBJ_REQ_KNOWN, &obj_request->flags);
1205 static bool obj_request_known_test(struct rbd_obj_request *obj_request)
1208 return test_bit(OBJ_REQ_KNOWN, &obj_request->flags) != 0;
1211 static bool obj_request_exists_test(struct rbd_obj_request *obj_request)
1214 return test_bit(OBJ_REQ_EXISTS, &obj_request->flags) != 0;
1217 static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
1219 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1220 atomic_read(&obj_request->kref.refcount));
1221 kref_get(&obj_request->kref);
1224 static void rbd_obj_request_destroy(struct kref *kref);
1225 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1227 rbd_assert(obj_request != NULL);
1228 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1229 atomic_read(&obj_request->kref.refcount));
1230 kref_put(&obj_request->kref, rbd_obj_request_destroy);
1233 static void rbd_img_request_get(struct rbd_img_request *img_request)
1235 dout("%s: img %p (was %d)\n", __func__, img_request,
1236 atomic_read(&img_request->kref.refcount));
1237 kref_get(&img_request->kref);
1240 static void rbd_img_request_destroy(struct kref *kref);
1241 static void rbd_img_request_put(struct rbd_img_request *img_request)
1243 rbd_assert(img_request != NULL);
1244 dout("%s: img %p (was %d)\n", __func__, img_request,
1245 atomic_read(&img_request->kref.refcount));
1246 kref_put(&img_request->kref, rbd_img_request_destroy);
1249 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1250 struct rbd_obj_request *obj_request)
1252 rbd_assert(obj_request->img_request == NULL);
1254 /* Image request now owns object's original reference */
1255 obj_request->img_request = img_request;
1256 obj_request->which = img_request->obj_request_count;
1257 rbd_assert(!obj_request_img_data_test(obj_request));
1258 obj_request_img_data_set(obj_request);
1259 rbd_assert(obj_request->which != BAD_WHICH);
1260 img_request->obj_request_count++;
1261 list_add_tail(&obj_request->links, &img_request->obj_requests);
1262 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1263 obj_request->which);
1266 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1267 struct rbd_obj_request *obj_request)
1269 rbd_assert(obj_request->which != BAD_WHICH);
1271 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1272 obj_request->which);
1273 list_del(&obj_request->links);
1274 rbd_assert(img_request->obj_request_count > 0);
1275 img_request->obj_request_count--;
1276 rbd_assert(obj_request->which == img_request->obj_request_count);
1277 obj_request->which = BAD_WHICH;
1278 rbd_assert(obj_request_img_data_test(obj_request));
1279 rbd_assert(obj_request->img_request == img_request);
1280 obj_request->img_request = NULL;
1281 obj_request->callback = NULL;
1282 rbd_obj_request_put(obj_request);
1285 static bool obj_request_type_valid(enum obj_request_type type)
1288 case OBJ_REQUEST_NODATA:
1289 case OBJ_REQUEST_BIO:
1290 case OBJ_REQUEST_PAGES:
1297 static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
1298 struct rbd_obj_request *obj_request)
1300 dout("%s: osdc %p obj %p\n", __func__, osdc, obj_request);
1302 return ceph_osdc_start_request(osdc, obj_request->osd_req, false);
1305 static void rbd_img_request_complete(struct rbd_img_request *img_request)
1308 dout("%s: img %p\n", __func__, img_request);
1311 * If no error occurred, compute the aggregate transfer
1312 * count for the image request. We could instead use
1313 * atomic64_cmpxchg() to update it as each object request
1314 * completes; not clear which way is better off hand.
1316 if (!img_request->result) {
1317 struct rbd_obj_request *obj_request;
1320 for_each_obj_request(img_request, obj_request)
1321 xferred += obj_request->xferred;
1322 img_request->xferred = xferred;
1325 if (img_request->callback)
1326 img_request->callback(img_request);
1328 rbd_img_request_put(img_request);
1331 /* Caller is responsible for rbd_obj_request_destroy(obj_request) */
1333 static int rbd_obj_request_wait(struct rbd_obj_request *obj_request)
1335 dout("%s: obj %p\n", __func__, obj_request);
1337 return wait_for_completion_interruptible(&obj_request->completion);
1341 * The default/initial value for all image request flags is 0. Each
1342 * is conditionally set to 1 at image request initialization time
1343 * and currently never change thereafter.
1345 static void img_request_write_set(struct rbd_img_request *img_request)
1347 set_bit(IMG_REQ_WRITE, &img_request->flags);
1351 static bool img_request_write_test(struct rbd_img_request *img_request)
1354 return test_bit(IMG_REQ_WRITE, &img_request->flags) != 0;
1357 static void img_request_child_set(struct rbd_img_request *img_request)
1359 set_bit(IMG_REQ_CHILD, &img_request->flags);
1363 static bool img_request_child_test(struct rbd_img_request *img_request)
1366 return test_bit(IMG_REQ_CHILD, &img_request->flags) != 0;
1369 static void img_request_layered_set(struct rbd_img_request *img_request)
1371 set_bit(IMG_REQ_LAYERED, &img_request->flags);
1375 static bool img_request_layered_test(struct rbd_img_request *img_request)
1378 return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1382 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
1384 u64 xferred = obj_request->xferred;
1385 u64 length = obj_request->length;
1387 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1388 obj_request, obj_request->img_request, obj_request->result,
1391 * ENOENT means a hole in the image. We zero-fill the
1392 * entire length of the request. A short read also implies
1393 * zero-fill to the end of the request. Either way we
1394 * update the xferred count to indicate the whole request
1397 rbd_assert(obj_request->type != OBJ_REQUEST_NODATA);
1398 if (obj_request->result == -ENOENT) {
1399 if (obj_request->type == OBJ_REQUEST_BIO)
1400 zero_bio_chain(obj_request->bio_list, 0);
1402 zero_pages(obj_request->pages, 0, length);
1403 obj_request->result = 0;
1404 obj_request->xferred = length;
1405 } else if (xferred < length && !obj_request->result) {
1406 if (obj_request->type == OBJ_REQUEST_BIO)
1407 zero_bio_chain(obj_request->bio_list, xferred);
1409 zero_pages(obj_request->pages, xferred, length);
1410 obj_request->xferred = length;
1412 obj_request_done_set(obj_request);
1415 static void rbd_obj_request_complete(struct rbd_obj_request *obj_request)
1417 dout("%s: obj %p cb %p\n", __func__, obj_request,
1418 obj_request->callback);
1419 if (obj_request->callback)
1420 obj_request->callback(obj_request);
1422 complete_all(&obj_request->completion);
1425 static void rbd_osd_trivial_callback(struct rbd_obj_request *obj_request)
1427 dout("%s: obj %p\n", __func__, obj_request);
1428 obj_request_done_set(obj_request);
1431 static void rbd_osd_read_callback(struct rbd_obj_request *obj_request)
1433 struct rbd_img_request *img_request = NULL;
1434 struct rbd_device *rbd_dev = NULL;
1435 bool layered = false;
1437 if (obj_request_img_data_test(obj_request)) {
1438 img_request = obj_request->img_request;
1439 layered = img_request && img_request_layered_test(img_request);
1440 rbd_dev = img_request->rbd_dev;
1443 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1444 obj_request, img_request, obj_request->result,
1445 obj_request->xferred, obj_request->length);
1446 if (layered && obj_request->result == -ENOENT &&
1447 obj_request->img_offset < rbd_dev->parent_overlap)
1448 rbd_img_parent_read(obj_request);
1449 else if (img_request)
1450 rbd_img_obj_request_read_callback(obj_request);
1452 obj_request_done_set(obj_request);
1455 static void rbd_osd_write_callback(struct rbd_obj_request *obj_request)
1457 dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1458 obj_request->result, obj_request->length);
1460 * There is no such thing as a successful short write. Set
1461 * it to our originally-requested length.
1463 obj_request->xferred = obj_request->length;
1464 obj_request_done_set(obj_request);
1468 * For a simple stat call there's nothing to do. We'll do more if
1469 * this is part of a write sequence for a layered image.
1471 static void rbd_osd_stat_callback(struct rbd_obj_request *obj_request)
1473 dout("%s: obj %p\n", __func__, obj_request);
1474 obj_request_done_set(obj_request);
1477 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
1478 struct ceph_msg *msg)
1480 struct rbd_obj_request *obj_request = osd_req->r_priv;
1483 dout("%s: osd_req %p msg %p\n", __func__, osd_req, msg);
1484 rbd_assert(osd_req == obj_request->osd_req);
1485 if (obj_request_img_data_test(obj_request)) {
1486 rbd_assert(obj_request->img_request);
1487 rbd_assert(obj_request->which != BAD_WHICH);
1489 rbd_assert(obj_request->which == BAD_WHICH);
1492 if (osd_req->r_result < 0)
1493 obj_request->result = osd_req->r_result;
1494 obj_request->version = le64_to_cpu(osd_req->r_reassert_version.version);
1496 BUG_ON(osd_req->r_num_ops > 2);
1499 * We support a 64-bit length, but ultimately it has to be
1500 * passed to blk_end_request(), which takes an unsigned int.
1502 obj_request->xferred = osd_req->r_reply_op_len[0];
1503 rbd_assert(obj_request->xferred < (u64)UINT_MAX);
1504 opcode = osd_req->r_ops[0].op;
1506 case CEPH_OSD_OP_READ:
1507 rbd_osd_read_callback(obj_request);
1509 case CEPH_OSD_OP_WRITE:
1510 rbd_osd_write_callback(obj_request);
1512 case CEPH_OSD_OP_STAT:
1513 rbd_osd_stat_callback(obj_request);
1515 case CEPH_OSD_OP_CALL:
1516 case CEPH_OSD_OP_NOTIFY_ACK:
1517 case CEPH_OSD_OP_WATCH:
1518 rbd_osd_trivial_callback(obj_request);
1521 rbd_warn(NULL, "%s: unsupported op %hu\n",
1522 obj_request->object_name, (unsigned short) opcode);
1526 if (obj_request_done_test(obj_request))
1527 rbd_obj_request_complete(obj_request);
1530 static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request)
1532 struct rbd_img_request *img_request = obj_request->img_request;
1533 struct ceph_osd_request *osd_req = obj_request->osd_req;
1536 rbd_assert(osd_req != NULL);
1538 snap_id = img_request ? img_request->snap_id : CEPH_NOSNAP;
1539 ceph_osdc_build_request(osd_req, obj_request->offset,
1540 NULL, snap_id, NULL);
1543 static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request)
1545 struct rbd_img_request *img_request = obj_request->img_request;
1546 struct ceph_osd_request *osd_req = obj_request->osd_req;
1547 struct ceph_snap_context *snapc;
1548 struct timespec mtime = CURRENT_TIME;
1550 rbd_assert(osd_req != NULL);
1552 snapc = img_request ? img_request->snapc : NULL;
1553 ceph_osdc_build_request(osd_req, obj_request->offset,
1554 snapc, CEPH_NOSNAP, &mtime);
1557 static struct ceph_osd_request *rbd_osd_req_create(
1558 struct rbd_device *rbd_dev,
1560 struct rbd_obj_request *obj_request)
1562 struct ceph_snap_context *snapc = NULL;
1563 struct ceph_osd_client *osdc;
1564 struct ceph_osd_request *osd_req;
1566 if (obj_request_img_data_test(obj_request)) {
1567 struct rbd_img_request *img_request = obj_request->img_request;
1569 rbd_assert(write_request ==
1570 img_request_write_test(img_request));
1572 snapc = img_request->snapc;
1575 /* Allocate and initialize the request, for the single op */
1577 osdc = &rbd_dev->rbd_client->client->osdc;
1578 osd_req = ceph_osdc_alloc_request(osdc, snapc, 1, false, GFP_ATOMIC);
1580 return NULL; /* ENOMEM */
1583 osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
1585 osd_req->r_flags = CEPH_OSD_FLAG_READ;
1587 osd_req->r_callback = rbd_osd_req_callback;
1588 osd_req->r_priv = obj_request;
1590 osd_req->r_oid_len = strlen(obj_request->object_name);
1591 rbd_assert(osd_req->r_oid_len < sizeof (osd_req->r_oid));
1592 memcpy(osd_req->r_oid, obj_request->object_name, osd_req->r_oid_len);
1594 osd_req->r_file_layout = rbd_dev->layout; /* struct */
1600 * Create a copyup osd request based on the information in the
1601 * object request supplied. A copyup request has two osd ops,
1602 * a copyup method call, and a "normal" write request.
1604 static struct ceph_osd_request *
1605 rbd_osd_req_create_copyup(struct rbd_obj_request *obj_request)
1607 struct rbd_img_request *img_request;
1608 struct ceph_snap_context *snapc;
1609 struct rbd_device *rbd_dev;
1610 struct ceph_osd_client *osdc;
1611 struct ceph_osd_request *osd_req;
1613 rbd_assert(obj_request_img_data_test(obj_request));
1614 img_request = obj_request->img_request;
1615 rbd_assert(img_request);
1616 rbd_assert(img_request_write_test(img_request));
1618 /* Allocate and initialize the request, for the two ops */
1620 snapc = img_request->snapc;
1621 rbd_dev = img_request->rbd_dev;
1622 osdc = &rbd_dev->rbd_client->client->osdc;
1623 osd_req = ceph_osdc_alloc_request(osdc, snapc, 2, false, GFP_ATOMIC);
1625 return NULL; /* ENOMEM */
1627 osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
1628 osd_req->r_callback = rbd_osd_req_callback;
1629 osd_req->r_priv = obj_request;
1631 osd_req->r_oid_len = strlen(obj_request->object_name);
1632 rbd_assert(osd_req->r_oid_len < sizeof (osd_req->r_oid));
1633 memcpy(osd_req->r_oid, obj_request->object_name, osd_req->r_oid_len);
1635 osd_req->r_file_layout = rbd_dev->layout; /* struct */
1641 static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req)
1643 ceph_osdc_put_request(osd_req);
1646 /* object_name is assumed to be a non-null pointer and NUL-terminated */
1648 static struct rbd_obj_request *rbd_obj_request_create(const char *object_name,
1649 u64 offset, u64 length,
1650 enum obj_request_type type)
1652 struct rbd_obj_request *obj_request;
1656 rbd_assert(obj_request_type_valid(type));
1658 size = strlen(object_name) + 1;
1659 obj_request = kzalloc(sizeof (*obj_request) + size, GFP_KERNEL);
1663 name = (char *)(obj_request + 1);
1664 obj_request->object_name = memcpy(name, object_name, size);
1665 obj_request->offset = offset;
1666 obj_request->length = length;
1667 obj_request->flags = 0;
1668 obj_request->which = BAD_WHICH;
1669 obj_request->type = type;
1670 INIT_LIST_HEAD(&obj_request->links);
1671 init_completion(&obj_request->completion);
1672 kref_init(&obj_request->kref);
1674 dout("%s: \"%s\" %llu/%llu %d -> obj %p\n", __func__, object_name,
1675 offset, length, (int)type, obj_request);
1680 static void rbd_obj_request_destroy(struct kref *kref)
1682 struct rbd_obj_request *obj_request;
1684 obj_request = container_of(kref, struct rbd_obj_request, kref);
1686 dout("%s: obj %p\n", __func__, obj_request);
1688 rbd_assert(obj_request->img_request == NULL);
1689 rbd_assert(obj_request->which == BAD_WHICH);
1691 if (obj_request->osd_req)
1692 rbd_osd_req_destroy(obj_request->osd_req);
1694 rbd_assert(obj_request_type_valid(obj_request->type));
1695 switch (obj_request->type) {
1696 case OBJ_REQUEST_NODATA:
1697 break; /* Nothing to do */
1698 case OBJ_REQUEST_BIO:
1699 if (obj_request->bio_list)
1700 bio_chain_put(obj_request->bio_list);
1702 case OBJ_REQUEST_PAGES:
1703 if (obj_request->pages)
1704 ceph_release_page_vector(obj_request->pages,
1705 obj_request->page_count);
1713 * Caller is responsible for filling in the list of object requests
1714 * that comprises the image request, and the Linux request pointer
1715 * (if there is one).
1717 static struct rbd_img_request *rbd_img_request_create(
1718 struct rbd_device *rbd_dev,
1719 u64 offset, u64 length,
1723 struct rbd_img_request *img_request;
1724 struct ceph_snap_context *snapc = NULL;
1726 img_request = kmalloc(sizeof (*img_request), GFP_ATOMIC);
1730 if (write_request) {
1731 down_read(&rbd_dev->header_rwsem);
1732 snapc = ceph_get_snap_context(rbd_dev->header.snapc);
1733 up_read(&rbd_dev->header_rwsem);
1734 if (WARN_ON(!snapc)) {
1736 return NULL; /* Shouldn't happen */
1741 img_request->rq = NULL;
1742 img_request->rbd_dev = rbd_dev;
1743 img_request->offset = offset;
1744 img_request->length = length;
1745 img_request->flags = 0;
1746 if (write_request) {
1747 img_request_write_set(img_request);
1748 img_request->snapc = snapc;
1750 img_request->snap_id = rbd_dev->spec->snap_id;
1753 img_request_child_set(img_request);
1754 if (rbd_dev->parent_spec)
1755 img_request_layered_set(img_request);
1756 spin_lock_init(&img_request->completion_lock);
1757 img_request->next_completion = 0;
1758 img_request->callback = NULL;
1759 img_request->result = 0;
1760 img_request->obj_request_count = 0;
1761 INIT_LIST_HEAD(&img_request->obj_requests);
1762 kref_init(&img_request->kref);
1764 rbd_img_request_get(img_request); /* Avoid a warning */
1765 rbd_img_request_put(img_request); /* TEMPORARY */
1767 dout("%s: rbd_dev %p %s %llu/%llu -> img %p\n", __func__, rbd_dev,
1768 write_request ? "write" : "read", offset, length,
1774 static void rbd_img_request_destroy(struct kref *kref)
1776 struct rbd_img_request *img_request;
1777 struct rbd_obj_request *obj_request;
1778 struct rbd_obj_request *next_obj_request;
1780 img_request = container_of(kref, struct rbd_img_request, kref);
1782 dout("%s: img %p\n", __func__, img_request);
1784 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1785 rbd_img_obj_request_del(img_request, obj_request);
1786 rbd_assert(img_request->obj_request_count == 0);
1788 if (img_request_write_test(img_request))
1789 ceph_put_snap_context(img_request->snapc);
1791 if (img_request_child_test(img_request))
1792 rbd_obj_request_put(img_request->obj_request);
1797 static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request)
1799 struct rbd_img_request *img_request;
1800 unsigned int xferred;
1804 rbd_assert(obj_request_img_data_test(obj_request));
1805 img_request = obj_request->img_request;
1807 rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
1808 xferred = (unsigned int)obj_request->xferred;
1809 result = obj_request->result;
1811 struct rbd_device *rbd_dev = img_request->rbd_dev;
1813 rbd_warn(rbd_dev, "%s %llx at %llx (%llx)\n",
1814 img_request_write_test(img_request) ? "write" : "read",
1815 obj_request->length, obj_request->img_offset,
1816 obj_request->offset);
1817 rbd_warn(rbd_dev, " result %d xferred %x\n",
1819 if (!img_request->result)
1820 img_request->result = result;
1823 /* Image object requests don't own their page array */
1825 if (obj_request->type == OBJ_REQUEST_PAGES) {
1826 obj_request->pages = NULL;
1827 obj_request->page_count = 0;
1830 if (img_request_child_test(img_request)) {
1831 rbd_assert(img_request->obj_request != NULL);
1832 more = obj_request->which < img_request->obj_request_count - 1;
1834 rbd_assert(img_request->rq != NULL);
1835 more = blk_end_request(img_request->rq, result, xferred);
1841 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request)
1843 struct rbd_img_request *img_request;
1844 u32 which = obj_request->which;
1847 rbd_assert(obj_request_img_data_test(obj_request));
1848 img_request = obj_request->img_request;
1850 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1851 rbd_assert(img_request != NULL);
1852 rbd_assert(img_request->obj_request_count > 0);
1853 rbd_assert(which != BAD_WHICH);
1854 rbd_assert(which < img_request->obj_request_count);
1855 rbd_assert(which >= img_request->next_completion);
1857 spin_lock_irq(&img_request->completion_lock);
1858 if (which != img_request->next_completion)
1861 for_each_obj_request_from(img_request, obj_request) {
1863 rbd_assert(which < img_request->obj_request_count);
1865 if (!obj_request_done_test(obj_request))
1867 more = rbd_img_obj_end_request(obj_request);
1871 rbd_assert(more ^ (which == img_request->obj_request_count));
1872 img_request->next_completion = which;
1874 spin_unlock_irq(&img_request->completion_lock);
1877 rbd_img_request_complete(img_request);
1881 * Split up an image request into one or more object requests, each
1882 * to a different object. The "type" parameter indicates whether
1883 * "data_desc" is the pointer to the head of a list of bio
1884 * structures, or the base of a page array. In either case this
1885 * function assumes data_desc describes memory sufficient to hold
1886 * all data described by the image request.
1888 static int rbd_img_request_fill(struct rbd_img_request *img_request,
1889 enum obj_request_type type,
1892 struct rbd_device *rbd_dev = img_request->rbd_dev;
1893 struct rbd_obj_request *obj_request = NULL;
1894 struct rbd_obj_request *next_obj_request;
1895 bool write_request = img_request_write_test(img_request);
1896 struct bio *bio_list;
1897 unsigned int bio_offset = 0;
1898 struct page **pages;
1903 dout("%s: img %p type %d data_desc %p\n", __func__, img_request,
1904 (int)type, data_desc);
1906 opcode = write_request ? CEPH_OSD_OP_WRITE : CEPH_OSD_OP_READ;
1907 img_offset = img_request->offset;
1908 resid = img_request->length;
1909 rbd_assert(resid > 0);
1911 if (type == OBJ_REQUEST_BIO) {
1912 bio_list = data_desc;
1913 rbd_assert(img_offset == bio_list->bi_sector << SECTOR_SHIFT);
1915 rbd_assert(type == OBJ_REQUEST_PAGES);
1920 struct ceph_osd_request *osd_req;
1921 const char *object_name;
1925 object_name = rbd_segment_name(rbd_dev, img_offset);
1928 offset = rbd_segment_offset(rbd_dev, img_offset);
1929 length = rbd_segment_length(rbd_dev, img_offset, resid);
1930 obj_request = rbd_obj_request_create(object_name,
1931 offset, length, type);
1932 kfree(object_name); /* object request has its own copy */
1936 if (type == OBJ_REQUEST_BIO) {
1937 unsigned int clone_size;
1939 rbd_assert(length <= (u64)UINT_MAX);
1940 clone_size = (unsigned int)length;
1941 obj_request->bio_list =
1942 bio_chain_clone_range(&bio_list,
1946 if (!obj_request->bio_list)
1949 unsigned int page_count;
1951 obj_request->pages = pages;
1952 page_count = (u32)calc_pages_for(offset, length);
1953 obj_request->page_count = page_count;
1954 if ((offset + length) & ~PAGE_MASK)
1955 page_count--; /* more on last page */
1956 pages += page_count;
1959 osd_req = rbd_osd_req_create(rbd_dev, write_request,
1963 obj_request->osd_req = osd_req;
1964 obj_request->callback = rbd_img_obj_callback;
1966 osd_req_op_extent_init(osd_req, 0, opcode, offset, length,
1968 if (type == OBJ_REQUEST_BIO)
1969 osd_req_op_extent_osd_data_bio(osd_req, 0,
1970 obj_request->bio_list, length);
1972 osd_req_op_extent_osd_data_pages(osd_req, 0,
1973 obj_request->pages, length,
1974 offset & ~PAGE_MASK, false, false);
1977 rbd_osd_req_format_write(obj_request);
1979 rbd_osd_req_format_read(obj_request);
1981 obj_request->img_offset = img_offset;
1982 rbd_img_obj_request_add(img_request, obj_request);
1984 img_offset += length;
1991 rbd_obj_request_put(obj_request);
1993 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1994 rbd_obj_request_put(obj_request);
2000 rbd_img_obj_copyup_callback(struct rbd_obj_request *obj_request)
2002 struct rbd_img_request *img_request;
2003 struct rbd_device *rbd_dev;
2007 rbd_assert(obj_request->type == OBJ_REQUEST_BIO);
2008 rbd_assert(obj_request_img_data_test(obj_request));
2009 img_request = obj_request->img_request;
2010 rbd_assert(img_request);
2012 rbd_dev = img_request->rbd_dev;
2013 rbd_assert(rbd_dev);
2014 length = (u64)1 << rbd_dev->header.obj_order;
2015 page_count = (u32)calc_pages_for(0, length);
2017 rbd_assert(obj_request->copyup_pages);
2018 ceph_release_page_vector(obj_request->copyup_pages, page_count);
2019 obj_request->copyup_pages = NULL;
2022 * We want the transfer count to reflect the size of the
2023 * original write request. There is no such thing as a
2024 * successful short write, so if the request was successful
2025 * we can just set it to the originally-requested length.
2027 if (!obj_request->result)
2028 obj_request->xferred = obj_request->length;
2030 /* Finish up with the normal image object callback */
2032 rbd_img_obj_callback(obj_request);
2036 rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request)
2038 struct rbd_obj_request *orig_request;
2039 struct ceph_osd_request *osd_req;
2040 struct ceph_osd_client *osdc;
2041 struct rbd_device *rbd_dev;
2042 struct page **pages;
2047 rbd_assert(img_request_child_test(img_request));
2049 /* First get what we need from the image request */
2051 pages = img_request->copyup_pages;
2052 rbd_assert(pages != NULL);
2053 img_request->copyup_pages = NULL;
2055 orig_request = img_request->obj_request;
2056 rbd_assert(orig_request != NULL);
2057 rbd_assert(orig_request->type == OBJ_REQUEST_BIO);
2058 result = img_request->result;
2059 obj_size = img_request->length;
2060 xferred = img_request->xferred;
2062 rbd_dev = img_request->rbd_dev;
2063 rbd_assert(rbd_dev);
2064 rbd_assert(obj_size == (u64)1 << rbd_dev->header.obj_order);
2066 rbd_img_request_put(img_request);
2071 /* Allocate the new copyup osd request for the original request */
2074 rbd_assert(!orig_request->osd_req);
2075 osd_req = rbd_osd_req_create_copyup(orig_request);
2078 orig_request->osd_req = osd_req;
2079 orig_request->copyup_pages = pages;
2081 /* Initialize the copyup op */
2083 osd_req_op_cls_init(osd_req, 0, CEPH_OSD_OP_CALL, "rbd", "copyup");
2084 osd_req_op_cls_request_data_pages(osd_req, 0, pages, obj_size, 0,
2087 /* Then the original write request op */
2089 osd_req_op_extent_init(osd_req, 1, CEPH_OSD_OP_WRITE,
2090 orig_request->offset,
2091 orig_request->length, 0, 0);
2092 osd_req_op_extent_osd_data_bio(osd_req, 1, orig_request->bio_list,
2093 orig_request->length);
2095 rbd_osd_req_format_write(orig_request);
2097 /* All set, send it off. */
2099 orig_request->callback = rbd_img_obj_copyup_callback;
2100 osdc = &rbd_dev->rbd_client->client->osdc;
2101 result = rbd_obj_request_submit(osdc, orig_request);
2105 /* Record the error code and complete the request */
2107 orig_request->result = result;
2108 orig_request->xferred = 0;
2109 obj_request_done_set(orig_request);
2110 rbd_obj_request_complete(orig_request);
2114 * Read from the parent image the range of data that covers the
2115 * entire target of the given object request. This is used for
2116 * satisfying a layered image write request when the target of an
2117 * object request from the image request does not exist.
2119 * A page array big enough to hold the returned data is allocated
2120 * and supplied to rbd_img_request_fill() as the "data descriptor."
2121 * When the read completes, this page array will be transferred to
2122 * the original object request for the copyup operation.
2124 * If an error occurs, record it as the result of the original
2125 * object request and mark it done so it gets completed.
2127 static int rbd_img_obj_parent_read_full(struct rbd_obj_request *obj_request)
2129 struct rbd_img_request *img_request = NULL;
2130 struct rbd_img_request *parent_request = NULL;
2131 struct rbd_device *rbd_dev;
2134 struct page **pages = NULL;
2138 rbd_assert(obj_request_img_data_test(obj_request));
2139 rbd_assert(obj_request->type == OBJ_REQUEST_BIO);
2141 img_request = obj_request->img_request;
2142 rbd_assert(img_request != NULL);
2143 rbd_dev = img_request->rbd_dev;
2144 rbd_assert(rbd_dev->parent != NULL);
2147 * First things first. The original osd request is of no
2148 * use to use any more, we'll need a new one that can hold
2149 * the two ops in a copyup request. We'll get that later,
2150 * but for now we can release the old one.
2152 rbd_osd_req_destroy(obj_request->osd_req);
2153 obj_request->osd_req = NULL;
2156 * Determine the byte range covered by the object in the
2157 * child image to which the original request was to be sent.
2159 img_offset = obj_request->img_offset - obj_request->offset;
2160 length = (u64)1 << rbd_dev->header.obj_order;
2163 * There is no defined parent data beyond the parent
2164 * overlap, so limit what we read at that boundary if
2167 if (img_offset + length > rbd_dev->parent_overlap) {
2168 rbd_assert(img_offset < rbd_dev->parent_overlap);
2169 length = rbd_dev->parent_overlap - img_offset;
2173 * Allocate a page array big enough to receive the data read
2176 page_count = (u32)calc_pages_for(0, length);
2177 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2178 if (IS_ERR(pages)) {
2179 result = PTR_ERR(pages);
2185 parent_request = rbd_img_request_create(rbd_dev->parent,
2188 if (!parent_request)
2190 rbd_obj_request_get(obj_request);
2191 parent_request->obj_request = obj_request;
2193 result = rbd_img_request_fill(parent_request, OBJ_REQUEST_PAGES, pages);
2196 parent_request->copyup_pages = pages;
2198 parent_request->callback = rbd_img_obj_parent_read_full_callback;
2199 result = rbd_img_request_submit(parent_request);
2203 parent_request->copyup_pages = NULL;
2204 parent_request->obj_request = NULL;
2205 rbd_obj_request_put(obj_request);
2208 ceph_release_page_vector(pages, page_count);
2210 rbd_img_request_put(parent_request);
2211 obj_request->result = result;
2212 obj_request->xferred = 0;
2213 obj_request_done_set(obj_request);
2218 static void rbd_img_obj_exists_callback(struct rbd_obj_request *obj_request)
2220 struct rbd_obj_request *orig_request;
2223 rbd_assert(!obj_request_img_data_test(obj_request));
2226 * All we need from the object request is the original
2227 * request and the result of the STAT op. Grab those, then
2228 * we're done with the request.
2230 orig_request = obj_request->obj_request;
2231 obj_request->obj_request = NULL;
2232 rbd_assert(orig_request);
2233 rbd_assert(orig_request->img_request);
2235 result = obj_request->result;
2236 obj_request->result = 0;
2238 dout("%s: obj %p for obj %p result %d %llu/%llu\n", __func__,
2239 obj_request, orig_request, result,
2240 obj_request->xferred, obj_request->length);
2241 rbd_obj_request_put(obj_request);
2243 rbd_assert(orig_request);
2244 rbd_assert(orig_request->img_request);
2247 * Our only purpose here is to determine whether the object
2248 * exists, and we don't want to treat the non-existence as
2249 * an error. If something else comes back, transfer the
2250 * error to the original request and complete it now.
2253 obj_request_existence_set(orig_request, true);
2254 } else if (result == -ENOENT) {
2255 obj_request_existence_set(orig_request, false);
2256 } else if (result) {
2257 orig_request->result = result;
2262 * Resubmit the original request now that we have recorded
2263 * whether the target object exists.
2265 orig_request->result = rbd_img_obj_request_submit(orig_request);
2267 if (orig_request->result)
2268 rbd_obj_request_complete(orig_request);
2269 rbd_obj_request_put(orig_request);
2272 static int rbd_img_obj_exists_submit(struct rbd_obj_request *obj_request)
2274 struct rbd_obj_request *stat_request;
2275 struct rbd_device *rbd_dev;
2276 struct ceph_osd_client *osdc;
2277 struct page **pages = NULL;
2283 * The response data for a STAT call consists of:
2290 size = sizeof (__le64) + sizeof (__le32) + sizeof (__le32);
2291 page_count = (u32)calc_pages_for(0, size);
2292 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2294 return PTR_ERR(pages);
2297 stat_request = rbd_obj_request_create(obj_request->object_name, 0, 0,
2302 rbd_obj_request_get(obj_request);
2303 stat_request->obj_request = obj_request;
2304 stat_request->pages = pages;
2305 stat_request->page_count = page_count;
2307 rbd_assert(obj_request->img_request);
2308 rbd_dev = obj_request->img_request->rbd_dev;
2309 stat_request->osd_req = rbd_osd_req_create(rbd_dev, false,
2311 if (!stat_request->osd_req)
2313 stat_request->callback = rbd_img_obj_exists_callback;
2315 osd_req_op_init(stat_request->osd_req, 0, CEPH_OSD_OP_STAT);
2316 osd_req_op_raw_data_in_pages(stat_request->osd_req, 0, pages, size, 0,
2318 rbd_osd_req_format_read(stat_request);
2320 osdc = &rbd_dev->rbd_client->client->osdc;
2321 ret = rbd_obj_request_submit(osdc, stat_request);
2324 rbd_obj_request_put(obj_request);
2329 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request)
2331 struct rbd_img_request *img_request;
2332 struct rbd_device *rbd_dev;
2335 rbd_assert(obj_request_img_data_test(obj_request));
2337 img_request = obj_request->img_request;
2338 rbd_assert(img_request);
2339 rbd_dev = img_request->rbd_dev;
2342 * Only writes to layered images need special handling.
2343 * Reads and non-layered writes are simple object requests.
2344 * Layered writes that start beyond the end of the overlap
2345 * with the parent have no parent data, so they too are
2346 * simple object requests. Finally, if the target object is
2347 * known to already exist, its parent data has already been
2348 * copied, so a write to the object can also be handled as a
2349 * simple object request.
2351 if (!img_request_write_test(img_request) ||
2352 !img_request_layered_test(img_request) ||
2353 rbd_dev->parent_overlap <= obj_request->img_offset ||
2354 ((known = obj_request_known_test(obj_request)) &&
2355 obj_request_exists_test(obj_request))) {
2357 struct rbd_device *rbd_dev;
2358 struct ceph_osd_client *osdc;
2360 rbd_dev = obj_request->img_request->rbd_dev;
2361 osdc = &rbd_dev->rbd_client->client->osdc;
2363 return rbd_obj_request_submit(osdc, obj_request);
2367 * It's a layered write. The target object might exist but
2368 * we may not know that yet. If we know it doesn't exist,
2369 * start by reading the data for the full target object from
2370 * the parent so we can use it for a copyup to the target.
2373 return rbd_img_obj_parent_read_full(obj_request);
2375 /* We don't know whether the target exists. Go find out. */
2377 return rbd_img_obj_exists_submit(obj_request);
2380 static int rbd_img_request_submit(struct rbd_img_request *img_request)
2382 struct rbd_obj_request *obj_request;
2383 struct rbd_obj_request *next_obj_request;
2385 dout("%s: img %p\n", __func__, img_request);
2386 for_each_obj_request_safe(img_request, obj_request, next_obj_request) {
2389 ret = rbd_img_obj_request_submit(obj_request);
2397 static void rbd_img_parent_read_callback(struct rbd_img_request *img_request)
2399 struct rbd_obj_request *obj_request;
2400 struct rbd_device *rbd_dev;
2403 rbd_assert(img_request_child_test(img_request));
2405 obj_request = img_request->obj_request;
2406 rbd_assert(obj_request);
2407 rbd_assert(obj_request->img_request);
2409 obj_request->result = img_request->result;
2410 if (obj_request->result)
2414 * We need to zero anything beyond the parent overlap
2415 * boundary. Since rbd_img_obj_request_read_callback()
2416 * will zero anything beyond the end of a short read, an
2417 * easy way to do this is to pretend the data from the
2418 * parent came up short--ending at the overlap boundary.
2420 rbd_assert(obj_request->img_offset < U64_MAX - obj_request->length);
2421 obj_end = obj_request->img_offset + obj_request->length;
2422 rbd_dev = obj_request->img_request->rbd_dev;
2423 if (obj_end > rbd_dev->parent_overlap) {
2426 if (obj_request->img_offset < rbd_dev->parent_overlap)
2427 xferred = rbd_dev->parent_overlap -
2428 obj_request->img_offset;
2430 obj_request->xferred = min(img_request->xferred, xferred);
2432 obj_request->xferred = img_request->xferred;
2435 rbd_img_obj_request_read_callback(obj_request);
2436 rbd_obj_request_complete(obj_request);
2439 static void rbd_img_parent_read(struct rbd_obj_request *obj_request)
2441 struct rbd_device *rbd_dev;
2442 struct rbd_img_request *img_request;
2445 rbd_assert(obj_request_img_data_test(obj_request));
2446 rbd_assert(obj_request->img_request != NULL);
2447 rbd_assert(obj_request->result == (s32) -ENOENT);
2448 rbd_assert(obj_request->type == OBJ_REQUEST_BIO);
2450 rbd_dev = obj_request->img_request->rbd_dev;
2451 rbd_assert(rbd_dev->parent != NULL);
2452 /* rbd_read_finish(obj_request, obj_request->length); */
2453 img_request = rbd_img_request_create(rbd_dev->parent,
2454 obj_request->img_offset,
2455 obj_request->length,
2461 rbd_obj_request_get(obj_request);
2462 img_request->obj_request = obj_request;
2464 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
2465 obj_request->bio_list);
2469 img_request->callback = rbd_img_parent_read_callback;
2470 result = rbd_img_request_submit(img_request);
2477 rbd_img_request_put(img_request);
2478 obj_request->result = result;
2479 obj_request->xferred = 0;
2480 obj_request_done_set(obj_request);
2483 static int rbd_obj_notify_ack(struct rbd_device *rbd_dev,
2484 u64 ver, u64 notify_id)
2486 struct rbd_obj_request *obj_request;
2487 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2490 obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0,
2491 OBJ_REQUEST_NODATA);
2496 obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
2497 if (!obj_request->osd_req)
2499 obj_request->callback = rbd_obj_request_put;
2501 osd_req_op_watch_init(obj_request->osd_req, 0, CEPH_OSD_OP_NOTIFY_ACK,
2503 rbd_osd_req_format_read(obj_request);
2505 ret = rbd_obj_request_submit(osdc, obj_request);
2508 rbd_obj_request_put(obj_request);
2513 static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
2515 struct rbd_device *rbd_dev = (struct rbd_device *)data;
2521 dout("%s: \"%s\" notify_id %llu opcode %u\n", __func__,
2522 rbd_dev->header_name, (unsigned long long) notify_id,
2523 (unsigned int) opcode);
2524 (void)rbd_dev_refresh(rbd_dev, &hver);
2526 rbd_obj_notify_ack(rbd_dev, hver, notify_id);
2530 * Request sync osd watch/unwatch. The value of "start" determines
2531 * whether a watch request is being initiated or torn down.
2533 static int rbd_dev_header_watch_sync(struct rbd_device *rbd_dev, int start)
2535 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2536 struct rbd_obj_request *obj_request;
2539 rbd_assert(start ^ !!rbd_dev->watch_event);
2540 rbd_assert(start ^ !!rbd_dev->watch_request);
2543 ret = ceph_osdc_create_event(osdc, rbd_watch_cb, rbd_dev,
2544 &rbd_dev->watch_event);
2547 rbd_assert(rbd_dev->watch_event != NULL);
2551 obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0,
2552 OBJ_REQUEST_NODATA);
2556 obj_request->osd_req = rbd_osd_req_create(rbd_dev, true, obj_request);
2557 if (!obj_request->osd_req)
2561 ceph_osdc_set_request_linger(osdc, obj_request->osd_req);
2563 ceph_osdc_unregister_linger_request(osdc,
2564 rbd_dev->watch_request->osd_req);
2566 osd_req_op_watch_init(obj_request->osd_req, 0, CEPH_OSD_OP_WATCH,
2567 rbd_dev->watch_event->cookie,
2568 rbd_dev->header.obj_version, start);
2569 rbd_osd_req_format_write(obj_request);
2571 ret = rbd_obj_request_submit(osdc, obj_request);
2574 ret = rbd_obj_request_wait(obj_request);
2577 ret = obj_request->result;
2582 * A watch request is set to linger, so the underlying osd
2583 * request won't go away until we unregister it. We retain
2584 * a pointer to the object request during that time (in
2585 * rbd_dev->watch_request), so we'll keep a reference to
2586 * it. We'll drop that reference (below) after we've
2590 rbd_dev->watch_request = obj_request;
2595 /* We have successfully torn down the watch request */
2597 rbd_obj_request_put(rbd_dev->watch_request);
2598 rbd_dev->watch_request = NULL;
2600 /* Cancel the event if we're tearing down, or on error */
2601 ceph_osdc_cancel_event(rbd_dev->watch_event);
2602 rbd_dev->watch_event = NULL;
2604 rbd_obj_request_put(obj_request);
2610 * Synchronous osd object method call. Returns the number of bytes
2611 * returned in the outbound buffer, or a negative error code.
2613 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
2614 const char *object_name,
2615 const char *class_name,
2616 const char *method_name,
2617 const void *outbound,
2618 size_t outbound_size,
2620 size_t inbound_size,
2623 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2624 struct rbd_obj_request *obj_request;
2625 struct page **pages;
2630 * Method calls are ultimately read operations. The result
2631 * should placed into the inbound buffer provided. They
2632 * also supply outbound data--parameters for the object
2633 * method. Currently if this is present it will be a
2636 page_count = (u32)calc_pages_for(0, inbound_size);
2637 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2639 return PTR_ERR(pages);
2642 obj_request = rbd_obj_request_create(object_name, 0, inbound_size,
2647 obj_request->pages = pages;
2648 obj_request->page_count = page_count;
2650 obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
2651 if (!obj_request->osd_req)
2654 osd_req_op_cls_init(obj_request->osd_req, 0, CEPH_OSD_OP_CALL,
2655 class_name, method_name);
2656 if (outbound_size) {
2657 struct ceph_pagelist *pagelist;
2659 pagelist = kmalloc(sizeof (*pagelist), GFP_NOFS);
2663 ceph_pagelist_init(pagelist);
2664 ceph_pagelist_append(pagelist, outbound, outbound_size);
2665 osd_req_op_cls_request_data_pagelist(obj_request->osd_req, 0,
2668 osd_req_op_cls_response_data_pages(obj_request->osd_req, 0,
2669 obj_request->pages, inbound_size,
2671 rbd_osd_req_format_read(obj_request);
2673 ret = rbd_obj_request_submit(osdc, obj_request);
2676 ret = rbd_obj_request_wait(obj_request);
2680 ret = obj_request->result;
2684 rbd_assert(obj_request->xferred < (u64)INT_MAX);
2685 ret = (int)obj_request->xferred;
2686 ceph_copy_from_page_vector(pages, inbound, 0, obj_request->xferred);
2688 *version = obj_request->version;
2691 rbd_obj_request_put(obj_request);
2693 ceph_release_page_vector(pages, page_count);
2698 static void rbd_request_fn(struct request_queue *q)
2699 __releases(q->queue_lock) __acquires(q->queue_lock)
2701 struct rbd_device *rbd_dev = q->queuedata;
2702 bool read_only = rbd_dev->mapping.read_only;
2706 while ((rq = blk_fetch_request(q))) {
2707 bool write_request = rq_data_dir(rq) == WRITE;
2708 struct rbd_img_request *img_request;
2712 /* Ignore any non-FS requests that filter through. */
2714 if (rq->cmd_type != REQ_TYPE_FS) {
2715 dout("%s: non-fs request type %d\n", __func__,
2716 (int) rq->cmd_type);
2717 __blk_end_request_all(rq, 0);
2721 /* Ignore/skip any zero-length requests */
2723 offset = (u64) blk_rq_pos(rq) << SECTOR_SHIFT;
2724 length = (u64) blk_rq_bytes(rq);
2727 dout("%s: zero-length request\n", __func__);
2728 __blk_end_request_all(rq, 0);
2732 spin_unlock_irq(q->queue_lock);
2734 /* Disallow writes to a read-only device */
2736 if (write_request) {
2740 rbd_assert(rbd_dev->spec->snap_id == CEPH_NOSNAP);
2744 * Quit early if the mapped snapshot no longer
2745 * exists. It's still possible the snapshot will
2746 * have disappeared by the time our request arrives
2747 * at the osd, but there's no sense in sending it if
2750 if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags)) {
2751 dout("request for non-existent snapshot");
2752 rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
2758 if (WARN_ON(offset && length > U64_MAX - offset + 1))
2759 goto end_request; /* Shouldn't happen */
2762 img_request = rbd_img_request_create(rbd_dev, offset, length,
2763 write_request, false);
2767 img_request->rq = rq;
2769 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
2772 result = rbd_img_request_submit(img_request);
2774 rbd_img_request_put(img_request);
2776 spin_lock_irq(q->queue_lock);
2778 rbd_warn(rbd_dev, "%s %llx at %llx result %d\n",
2779 write_request ? "write" : "read",
2780 length, offset, result);
2782 __blk_end_request_all(rq, result);
2788 * a queue callback. Makes sure that we don't create a bio that spans across
2789 * multiple osd objects. One exception would be with a single page bios,
2790 * which we handle later at bio_chain_clone_range()
2792 static int rbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bmd,
2793 struct bio_vec *bvec)
2795 struct rbd_device *rbd_dev = q->queuedata;
2796 sector_t sector_offset;
2797 sector_t sectors_per_obj;
2798 sector_t obj_sector_offset;
2802 * Find how far into its rbd object the partition-relative
2803 * bio start sector is to offset relative to the enclosing
2806 sector_offset = get_start_sect(bmd->bi_bdev) + bmd->bi_sector;
2807 sectors_per_obj = 1 << (rbd_dev->header.obj_order - SECTOR_SHIFT);
2808 obj_sector_offset = sector_offset & (sectors_per_obj - 1);
2811 * Compute the number of bytes from that offset to the end
2812 * of the object. Account for what's already used by the bio.
2814 ret = (int) (sectors_per_obj - obj_sector_offset) << SECTOR_SHIFT;
2815 if (ret > bmd->bi_size)
2816 ret -= bmd->bi_size;
2821 * Don't send back more than was asked for. And if the bio
2822 * was empty, let the whole thing through because: "Note
2823 * that a block device *must* allow a single page to be
2824 * added to an empty bio."
2826 rbd_assert(bvec->bv_len <= PAGE_SIZE);
2827 if (ret > (int) bvec->bv_len || !bmd->bi_size)
2828 ret = (int) bvec->bv_len;
2833 static void rbd_free_disk(struct rbd_device *rbd_dev)
2835 struct gendisk *disk = rbd_dev->disk;
2840 rbd_dev->disk = NULL;
2841 if (disk->flags & GENHD_FL_UP) {
2844 blk_cleanup_queue(disk->queue);
2849 static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
2850 const char *object_name,
2851 u64 offset, u64 length,
2852 void *buf, u64 *version)
2855 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2856 struct rbd_obj_request *obj_request;
2857 struct page **pages = NULL;
2862 page_count = (u32) calc_pages_for(offset, length);
2863 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2865 ret = PTR_ERR(pages);
2868 obj_request = rbd_obj_request_create(object_name, offset, length,
2873 obj_request->pages = pages;
2874 obj_request->page_count = page_count;
2876 obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
2877 if (!obj_request->osd_req)
2880 osd_req_op_extent_init(obj_request->osd_req, 0, CEPH_OSD_OP_READ,
2881 offset, length, 0, 0);
2882 osd_req_op_extent_osd_data_pages(obj_request->osd_req, 0,
2884 obj_request->length,
2885 obj_request->offset & ~PAGE_MASK,
2887 rbd_osd_req_format_read(obj_request);
2889 ret = rbd_obj_request_submit(osdc, obj_request);
2892 ret = rbd_obj_request_wait(obj_request);
2896 ret = obj_request->result;
2900 rbd_assert(obj_request->xferred <= (u64) SIZE_MAX);
2901 size = (size_t) obj_request->xferred;
2902 ceph_copy_from_page_vector(pages, buf, 0, size);
2903 rbd_assert(size <= (size_t) INT_MAX);
2906 *version = obj_request->version;
2909 rbd_obj_request_put(obj_request);
2911 ceph_release_page_vector(pages, page_count);
2917 * Read the complete header for the given rbd device.
2919 * Returns a pointer to a dynamically-allocated buffer containing
2920 * the complete and validated header. Caller can pass the address
2921 * of a variable that will be filled in with the version of the
2922 * header object at the time it was read.
2924 * Returns a pointer-coded errno if a failure occurs.
2926 static struct rbd_image_header_ondisk *
2927 rbd_dev_v1_header_read(struct rbd_device *rbd_dev, u64 *version)
2929 struct rbd_image_header_ondisk *ondisk = NULL;
2936 * The complete header will include an array of its 64-bit
2937 * snapshot ids, followed by the names of those snapshots as
2938 * a contiguous block of NUL-terminated strings. Note that
2939 * the number of snapshots could change by the time we read
2940 * it in, in which case we re-read it.
2947 size = sizeof (*ondisk);
2948 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
2950 ondisk = kmalloc(size, GFP_KERNEL);
2952 return ERR_PTR(-ENOMEM);
2954 ret = rbd_obj_read_sync(rbd_dev, rbd_dev->header_name,
2955 0, size, ondisk, version);
2958 if (WARN_ON((size_t) ret < size)) {
2960 rbd_warn(rbd_dev, "short header read (want %zd got %d)",
2964 if (!rbd_dev_ondisk_valid(ondisk)) {
2966 rbd_warn(rbd_dev, "invalid header");
2970 names_size = le64_to_cpu(ondisk->snap_names_len);
2971 want_count = snap_count;
2972 snap_count = le32_to_cpu(ondisk->snap_count);
2973 } while (snap_count != want_count);
2980 return ERR_PTR(ret);
2984 * reload the ondisk the header
2986 static int rbd_read_header(struct rbd_device *rbd_dev,
2987 struct rbd_image_header *header)
2989 struct rbd_image_header_ondisk *ondisk;
2993 ondisk = rbd_dev_v1_header_read(rbd_dev, &ver);
2995 return PTR_ERR(ondisk);
2996 ret = rbd_header_from_disk(header, ondisk);
2998 header->obj_version = ver;
3004 static void rbd_remove_all_snaps(struct rbd_device *rbd_dev)
3006 struct rbd_snap *snap;
3007 struct rbd_snap *next;
3009 list_for_each_entry_safe(snap, next, &rbd_dev->snaps, node) {
3010 list_del(&snap->node);
3011 rbd_snap_destroy(snap);
3015 static void rbd_update_mapping_size(struct rbd_device *rbd_dev)
3019 if (rbd_dev->spec->snap_id != CEPH_NOSNAP)
3022 size = (sector_t) rbd_dev->header.image_size / SECTOR_SIZE;
3023 dout("setting size to %llu sectors", (unsigned long long) size);
3024 rbd_dev->mapping.size = (u64) size;
3025 set_capacity(rbd_dev->disk, size);
3029 * only read the first part of the ondisk header, without the snaps info
3031 static int rbd_dev_v1_refresh(struct rbd_device *rbd_dev, u64 *hver)
3034 struct rbd_image_header h;
3036 ret = rbd_read_header(rbd_dev, &h);
3040 down_write(&rbd_dev->header_rwsem);
3042 /* Update image size, and check for resize of mapped image */
3043 rbd_dev->header.image_size = h.image_size;
3044 rbd_update_mapping_size(rbd_dev);
3046 /* rbd_dev->header.object_prefix shouldn't change */
3047 kfree(rbd_dev->header.snap_sizes);
3048 kfree(rbd_dev->header.snap_names);
3049 /* osd requests may still refer to snapc */
3050 ceph_put_snap_context(rbd_dev->header.snapc);
3053 *hver = h.obj_version;
3054 rbd_dev->header.obj_version = h.obj_version;
3055 rbd_dev->header.image_size = h.image_size;
3056 rbd_dev->header.snapc = h.snapc;
3057 rbd_dev->header.snap_names = h.snap_names;
3058 rbd_dev->header.snap_sizes = h.snap_sizes;
3059 /* Free the extra copy of the object prefix */
3060 WARN_ON(strcmp(rbd_dev->header.object_prefix, h.object_prefix));
3061 kfree(h.object_prefix);
3063 ret = rbd_dev_snaps_update(rbd_dev);
3065 up_write(&rbd_dev->header_rwsem);
3070 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver)
3074 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
3075 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
3076 if (rbd_dev->image_format == 1)
3077 ret = rbd_dev_v1_refresh(rbd_dev, hver);
3079 ret = rbd_dev_v2_refresh(rbd_dev, hver);
3080 mutex_unlock(&ctl_mutex);
3081 revalidate_disk(rbd_dev->disk);
3083 rbd_warn(rbd_dev, "got notification but failed to "
3084 " update snaps: %d\n", ret);
3089 static int rbd_init_disk(struct rbd_device *rbd_dev)
3091 struct gendisk *disk;
3092 struct request_queue *q;
3095 /* create gendisk info */
3096 disk = alloc_disk(RBD_MINORS_PER_MAJOR);
3100 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
3102 disk->major = rbd_dev->major;
3103 disk->first_minor = 0;
3104 disk->fops = &rbd_bd_ops;
3105 disk->private_data = rbd_dev;
3107 q = blk_init_queue(rbd_request_fn, &rbd_dev->lock);
3111 /* We use the default size, but let's be explicit about it. */
3112 blk_queue_physical_block_size(q, SECTOR_SIZE);
3114 /* set io sizes to object size */
3115 segment_size = rbd_obj_bytes(&rbd_dev->header);
3116 blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE);
3117 blk_queue_max_segment_size(q, segment_size);
3118 blk_queue_io_min(q, segment_size);
3119 blk_queue_io_opt(q, segment_size);
3121 blk_queue_merge_bvec(q, rbd_merge_bvec);
3124 q->queuedata = rbd_dev;
3126 rbd_dev->disk = disk;
3128 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
3141 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
3143 return container_of(dev, struct rbd_device, dev);
3146 static ssize_t rbd_size_show(struct device *dev,
3147 struct device_attribute *attr, char *buf)
3149 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3152 down_read(&rbd_dev->header_rwsem);
3153 size = get_capacity(rbd_dev->disk);
3154 up_read(&rbd_dev->header_rwsem);
3156 return sprintf(buf, "%llu\n", (unsigned long long) size * SECTOR_SIZE);
3160 * Note this shows the features for whatever's mapped, which is not
3161 * necessarily the base image.
3163 static ssize_t rbd_features_show(struct device *dev,
3164 struct device_attribute *attr, char *buf)
3166 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3168 return sprintf(buf, "0x%016llx\n",
3169 (unsigned long long) rbd_dev->mapping.features);
3172 static ssize_t rbd_major_show(struct device *dev,
3173 struct device_attribute *attr, char *buf)
3175 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3177 return sprintf(buf, "%d\n", rbd_dev->major);
3180 static ssize_t rbd_client_id_show(struct device *dev,
3181 struct device_attribute *attr, char *buf)
3183 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3185 return sprintf(buf, "client%lld\n",
3186 ceph_client_id(rbd_dev->rbd_client->client));
3189 static ssize_t rbd_pool_show(struct device *dev,
3190 struct device_attribute *attr, char *buf)
3192 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3194 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
3197 static ssize_t rbd_pool_id_show(struct device *dev,
3198 struct device_attribute *attr, char *buf)
3200 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3202 return sprintf(buf, "%llu\n",
3203 (unsigned long long) rbd_dev->spec->pool_id);
3206 static ssize_t rbd_name_show(struct device *dev,
3207 struct device_attribute *attr, char *buf)
3209 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3211 if (rbd_dev->spec->image_name)
3212 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
3214 return sprintf(buf, "(unknown)\n");
3217 static ssize_t rbd_image_id_show(struct device *dev,
3218 struct device_attribute *attr, char *buf)
3220 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3222 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
3226 * Shows the name of the currently-mapped snapshot (or
3227 * RBD_SNAP_HEAD_NAME for the base image).
3229 static ssize_t rbd_snap_show(struct device *dev,
3230 struct device_attribute *attr,
3233 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3235 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
3239 * For an rbd v2 image, shows the pool id, image id, and snapshot id
3240 * for the parent image. If there is no parent, simply shows
3241 * "(no parent image)".
3243 static ssize_t rbd_parent_show(struct device *dev,
3244 struct device_attribute *attr,
3247 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3248 struct rbd_spec *spec = rbd_dev->parent_spec;
3253 return sprintf(buf, "(no parent image)\n");
3255 count = sprintf(bufp, "pool_id %llu\npool_name %s\n",
3256 (unsigned long long) spec->pool_id, spec->pool_name);
3261 count = sprintf(bufp, "image_id %s\nimage_name %s\n", spec->image_id,
3262 spec->image_name ? spec->image_name : "(unknown)");
3267 count = sprintf(bufp, "snap_id %llu\nsnap_name %s\n",
3268 (unsigned long long) spec->snap_id, spec->snap_name);
3273 count = sprintf(bufp, "overlap %llu\n", rbd_dev->parent_overlap);
3278 return (ssize_t) (bufp - buf);
3281 static ssize_t rbd_image_refresh(struct device *dev,
3282 struct device_attribute *attr,
3286 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3289 ret = rbd_dev_refresh(rbd_dev, NULL);
3291 return ret < 0 ? ret : size;
3294 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
3295 static DEVICE_ATTR(features, S_IRUGO, rbd_features_show, NULL);
3296 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
3297 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
3298 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
3299 static DEVICE_ATTR(pool_id, S_IRUGO, rbd_pool_id_show, NULL);
3300 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
3301 static DEVICE_ATTR(image_id, S_IRUGO, rbd_image_id_show, NULL);
3302 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
3303 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
3304 static DEVICE_ATTR(parent, S_IRUGO, rbd_parent_show, NULL);
3306 static struct attribute *rbd_attrs[] = {
3307 &dev_attr_size.attr,
3308 &dev_attr_features.attr,
3309 &dev_attr_major.attr,
3310 &dev_attr_client_id.attr,
3311 &dev_attr_pool.attr,
3312 &dev_attr_pool_id.attr,
3313 &dev_attr_name.attr,
3314 &dev_attr_image_id.attr,
3315 &dev_attr_current_snap.attr,
3316 &dev_attr_parent.attr,
3317 &dev_attr_refresh.attr,
3321 static struct attribute_group rbd_attr_group = {
3325 static const struct attribute_group *rbd_attr_groups[] = {
3330 static void rbd_sysfs_dev_release(struct device *dev)
3334 static struct device_type rbd_device_type = {
3336 .groups = rbd_attr_groups,
3337 .release = rbd_sysfs_dev_release,
3340 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
3342 kref_get(&spec->kref);
3347 static void rbd_spec_free(struct kref *kref);
3348 static void rbd_spec_put(struct rbd_spec *spec)
3351 kref_put(&spec->kref, rbd_spec_free);
3354 static struct rbd_spec *rbd_spec_alloc(void)
3356 struct rbd_spec *spec;
3358 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
3361 kref_init(&spec->kref);
3366 static void rbd_spec_free(struct kref *kref)
3368 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
3370 kfree(spec->pool_name);
3371 kfree(spec->image_id);
3372 kfree(spec->image_name);
3373 kfree(spec->snap_name);
3377 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
3378 struct rbd_spec *spec)
3380 struct rbd_device *rbd_dev;
3382 rbd_dev = kzalloc(sizeof (*rbd_dev), GFP_KERNEL);
3386 spin_lock_init(&rbd_dev->lock);
3388 INIT_LIST_HEAD(&rbd_dev->node);
3389 INIT_LIST_HEAD(&rbd_dev->snaps);
3390 init_rwsem(&rbd_dev->header_rwsem);
3392 rbd_dev->spec = spec;
3393 rbd_dev->rbd_client = rbdc;
3395 /* Initialize the layout used for all rbd requests */
3397 rbd_dev->layout.fl_stripe_unit = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
3398 rbd_dev->layout.fl_stripe_count = cpu_to_le32(1);
3399 rbd_dev->layout.fl_object_size = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
3400 rbd_dev->layout.fl_pg_pool = cpu_to_le32((u32) spec->pool_id);
3405 static void rbd_dev_destroy(struct rbd_device *rbd_dev)
3407 rbd_spec_put(rbd_dev->parent_spec);
3408 kfree(rbd_dev->header_name);
3409 rbd_put_client(rbd_dev->rbd_client);
3410 rbd_spec_put(rbd_dev->spec);
3414 static void rbd_snap_destroy(struct rbd_snap *snap)
3420 static struct rbd_snap *rbd_snap_create(struct rbd_device *rbd_dev,
3421 const char *snap_name,
3422 u64 snap_id, u64 snap_size,
3425 struct rbd_snap *snap;
3427 snap = kzalloc(sizeof (*snap), GFP_KERNEL);
3429 return ERR_PTR(-ENOMEM);
3431 snap->name = snap_name;
3433 snap->size = snap_size;
3434 snap->features = snap_features;
3440 * Returns a dynamically-allocated snapshot name if successful, or a
3441 * pointer-coded error otherwise.
3443 static char *rbd_dev_v1_snap_info(struct rbd_device *rbd_dev, u32 which,
3444 u64 *snap_size, u64 *snap_features)
3449 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
3451 /* Skip over names until we find the one we are looking for */
3453 snap_name = rbd_dev->header.snap_names;
3454 for (i = 0; i < which; i++)
3455 snap_name += strlen(snap_name) + 1;
3457 snap_name = kstrdup(snap_name, GFP_KERNEL);
3459 return ERR_PTR(-ENOMEM);
3461 *snap_size = rbd_dev->header.snap_sizes[which];
3462 *snap_features = 0; /* No features for v1 */
3468 * Get the size and object order for an image snapshot, or if
3469 * snap_id is CEPH_NOSNAP, gets this information for the base
3472 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
3473 u8 *order, u64 *snap_size)
3475 __le64 snapid = cpu_to_le64(snap_id);
3480 } __attribute__ ((packed)) size_buf = { 0 };
3482 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3484 &snapid, sizeof (snapid),
3485 &size_buf, sizeof (size_buf), NULL);
3486 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3489 if (ret < sizeof (size_buf))
3493 *order = size_buf.order;
3494 *snap_size = le64_to_cpu(size_buf.size);
3496 dout(" snap_id 0x%016llx order = %u, snap_size = %llu\n",
3497 (unsigned long long)snap_id, (unsigned int)*order,
3498 (unsigned long long)*snap_size);
3503 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
3505 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
3506 &rbd_dev->header.obj_order,
3507 &rbd_dev->header.image_size);
3510 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
3516 reply_buf = kzalloc(RBD_OBJ_PREFIX_LEN_MAX, GFP_KERNEL);
3520 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3521 "rbd", "get_object_prefix", NULL, 0,
3522 reply_buf, RBD_OBJ_PREFIX_LEN_MAX, NULL);
3523 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3528 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
3529 p + ret, NULL, GFP_NOIO);
3532 if (IS_ERR(rbd_dev->header.object_prefix)) {
3533 ret = PTR_ERR(rbd_dev->header.object_prefix);
3534 rbd_dev->header.object_prefix = NULL;
3536 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix);
3544 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
3547 __le64 snapid = cpu_to_le64(snap_id);
3551 } __attribute__ ((packed)) features_buf = { 0 };
3555 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3556 "rbd", "get_features",
3557 &snapid, sizeof (snapid),
3558 &features_buf, sizeof (features_buf), NULL);
3559 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3562 if (ret < sizeof (features_buf))
3565 incompat = le64_to_cpu(features_buf.incompat);
3566 if (incompat & ~RBD_FEATURES_SUPPORTED)
3569 *snap_features = le64_to_cpu(features_buf.features);
3571 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
3572 (unsigned long long)snap_id,
3573 (unsigned long long)*snap_features,
3574 (unsigned long long)le64_to_cpu(features_buf.incompat));
3579 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
3581 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
3582 &rbd_dev->header.features);
3585 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
3587 struct rbd_spec *parent_spec;
3589 void *reply_buf = NULL;
3597 parent_spec = rbd_spec_alloc();
3601 size = sizeof (__le64) + /* pool_id */
3602 sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX + /* image_id */
3603 sizeof (__le64) + /* snap_id */
3604 sizeof (__le64); /* overlap */
3605 reply_buf = kmalloc(size, GFP_KERNEL);
3611 snapid = cpu_to_le64(CEPH_NOSNAP);
3612 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3613 "rbd", "get_parent",
3614 &snapid, sizeof (snapid),
3615 reply_buf, size, NULL);
3616 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3621 end = reply_buf + ret;
3623 ceph_decode_64_safe(&p, end, parent_spec->pool_id, out_err);
3624 if (parent_spec->pool_id == CEPH_NOPOOL)
3625 goto out; /* No parent? No problem. */
3627 /* The ceph file layout needs to fit pool id in 32 bits */
3630 if (WARN_ON(parent_spec->pool_id > (u64)U32_MAX))
3633 image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
3634 if (IS_ERR(image_id)) {
3635 ret = PTR_ERR(image_id);
3638 parent_spec->image_id = image_id;
3639 ceph_decode_64_safe(&p, end, parent_spec->snap_id, out_err);
3640 ceph_decode_64_safe(&p, end, overlap, out_err);
3642 rbd_dev->parent_overlap = overlap;
3643 rbd_dev->parent_spec = parent_spec;
3644 parent_spec = NULL; /* rbd_dev now owns this */
3649 rbd_spec_put(parent_spec);
3654 static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev)
3658 __le64 stripe_count;
3659 } __attribute__ ((packed)) striping_info_buf = { 0 };
3660 size_t size = sizeof (striping_info_buf);
3667 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3668 "rbd", "get_stripe_unit_count", NULL, 0,
3669 (char *)&striping_info_buf, size, NULL);
3670 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3677 * We don't actually support the "fancy striping" feature
3678 * (STRIPINGV2) yet, but if the striping sizes are the
3679 * defaults the behavior is the same as before. So find
3680 * out, and only fail if the image has non-default values.
3683 obj_size = (u64)1 << rbd_dev->header.obj_order;
3684 p = &striping_info_buf;
3685 stripe_unit = ceph_decode_64(&p);
3686 if (stripe_unit != obj_size) {
3687 rbd_warn(rbd_dev, "unsupported stripe unit "
3688 "(got %llu want %llu)",
3689 stripe_unit, obj_size);
3692 stripe_count = ceph_decode_64(&p);
3693 if (stripe_count != 1) {
3694 rbd_warn(rbd_dev, "unsupported stripe count "
3695 "(got %llu want 1)", stripe_count);
3698 rbd_dev->stripe_unit = stripe_unit;
3699 rbd_dev->stripe_count = stripe_count;
3704 static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
3706 size_t image_id_size;
3711 void *reply_buf = NULL;
3713 char *image_name = NULL;
3716 rbd_assert(!rbd_dev->spec->image_name);
3718 len = strlen(rbd_dev->spec->image_id);
3719 image_id_size = sizeof (__le32) + len;
3720 image_id = kmalloc(image_id_size, GFP_KERNEL);
3725 end = image_id + image_id_size;
3726 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len);
3728 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
3729 reply_buf = kmalloc(size, GFP_KERNEL);
3733 ret = rbd_obj_method_sync(rbd_dev, RBD_DIRECTORY,
3734 "rbd", "dir_get_name",
3735 image_id, image_id_size,
3736 reply_buf, size, NULL);
3740 end = reply_buf + ret;
3742 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
3743 if (IS_ERR(image_name))
3746 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
3755 * When a parent image gets probed, we only have the pool, image,
3756 * and snapshot ids but not the names of any of them. This call
3757 * is made later to fill in those names. It has to be done after
3758 * rbd_dev_snaps_update() has completed because some of the
3759 * information (in particular, snapshot name) is not available
3762 * When an image being mapped (not a parent) is probed, we have the
3763 * pool name and pool id, image name and image id, and the snapshot
3764 * name. The only thing we're missing is the snapshot id.
3766 static int rbd_dev_probe_update_spec(struct rbd_device *rbd_dev)
3768 struct ceph_osd_client *osdc;
3770 void *reply_buf = NULL;
3774 * An image being mapped will have the pool name (etc.), but
3775 * we need to look up the snapshot id.
3777 if (rbd_dev->spec->pool_name) {
3778 if (strcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME)) {
3779 struct rbd_snap *snap;
3781 snap = snap_by_name(rbd_dev, rbd_dev->spec->snap_name);
3784 rbd_dev->spec->snap_id = snap->id;
3786 rbd_dev->spec->snap_id = CEPH_NOSNAP;
3792 /* Look up the pool name */
3794 osdc = &rbd_dev->rbd_client->client->osdc;
3795 name = ceph_pg_pool_name_by_id(osdc->osdmap, rbd_dev->spec->pool_id);
3797 rbd_warn(rbd_dev, "there is no pool with id %llu",
3798 rbd_dev->spec->pool_id); /* Really a BUG() */
3802 rbd_dev->spec->pool_name = kstrdup(name, GFP_KERNEL);
3803 if (!rbd_dev->spec->pool_name)
3806 /* Fetch the image name; tolerate failure here */
3808 name = rbd_dev_image_name(rbd_dev);
3810 rbd_dev->spec->image_name = (char *)name;
3812 rbd_warn(rbd_dev, "unable to get image name");
3814 /* Look up the snapshot name. */
3816 name = rbd_snap_name(rbd_dev, rbd_dev->spec->snap_id);
3818 rbd_warn(rbd_dev, "no snapshot with id %llu",
3819 rbd_dev->spec->snap_id); /* Really a BUG() */
3823 rbd_dev->spec->snap_name = kstrdup(name, GFP_KERNEL);
3824 if(!rbd_dev->spec->snap_name)
3830 kfree(rbd_dev->spec->pool_name);
3831 rbd_dev->spec->pool_name = NULL;
3836 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev, u64 *ver)
3845 struct ceph_snap_context *snapc;
3849 * We'll need room for the seq value (maximum snapshot id),
3850 * snapshot count, and array of that many snapshot ids.
3851 * For now we have a fixed upper limit on the number we're
3852 * prepared to receive.
3854 size = sizeof (__le64) + sizeof (__le32) +
3855 RBD_MAX_SNAP_COUNT * sizeof (__le64);
3856 reply_buf = kzalloc(size, GFP_KERNEL);
3860 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3861 "rbd", "get_snapcontext", NULL, 0,
3862 reply_buf, size, ver);
3863 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3868 end = reply_buf + ret;
3870 ceph_decode_64_safe(&p, end, seq, out);
3871 ceph_decode_32_safe(&p, end, snap_count, out);
3874 * Make sure the reported number of snapshot ids wouldn't go
3875 * beyond the end of our buffer. But before checking that,
3876 * make sure the computed size of the snapshot context we
3877 * allocate is representable in a size_t.
3879 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
3884 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
3887 size = sizeof (struct ceph_snap_context) +
3888 snap_count * sizeof (snapc->snaps[0]);
3889 snapc = kmalloc(size, GFP_KERNEL);
3896 atomic_set(&snapc->nref, 1);
3898 snapc->num_snaps = snap_count;
3899 for (i = 0; i < snap_count; i++)
3900 snapc->snaps[i] = ceph_decode_64(&p);
3902 rbd_dev->header.snapc = snapc;
3904 dout(" snap context seq = %llu, snap_count = %u\n",
3905 (unsigned long long)seq, (unsigned int)snap_count);
3912 static char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev, u32 which)
3922 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
3923 reply_buf = kmalloc(size, GFP_KERNEL);
3925 return ERR_PTR(-ENOMEM);
3927 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
3928 snap_id = cpu_to_le64(rbd_dev->header.snapc->snaps[which]);
3929 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3930 "rbd", "get_snapshot_name",
3931 &snap_id, sizeof (snap_id),
3932 reply_buf, size, NULL);
3933 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3935 snap_name = ERR_PTR(ret);
3940 end = reply_buf + ret;
3941 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
3942 if (IS_ERR(snap_name))
3945 dout(" snap_id 0x%016llx snap_name = %s\n",
3946 (unsigned long long)le64_to_cpu(snap_id), snap_name);
3953 static char *rbd_dev_v2_snap_info(struct rbd_device *rbd_dev, u32 which,
3954 u64 *snap_size, u64 *snap_features)
3962 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
3963 snap_id = rbd_dev->header.snapc->snaps[which];
3964 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
3968 ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, &features);
3972 snap_name = rbd_dev_v2_snap_name(rbd_dev, which);
3973 if (!IS_ERR(snap_name)) {
3975 *snap_features = features;
3980 return ERR_PTR(ret);
3983 static char *rbd_dev_snap_info(struct rbd_device *rbd_dev, u32 which,
3984 u64 *snap_size, u64 *snap_features)
3986 if (rbd_dev->image_format == 1)
3987 return rbd_dev_v1_snap_info(rbd_dev, which,
3988 snap_size, snap_features);
3989 if (rbd_dev->image_format == 2)
3990 return rbd_dev_v2_snap_info(rbd_dev, which,
3991 snap_size, snap_features);
3992 return ERR_PTR(-EINVAL);
3995 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver)
4000 down_write(&rbd_dev->header_rwsem);
4002 /* Grab old order first, to see if it changes */
4004 obj_order = rbd_dev->header.obj_order,
4005 ret = rbd_dev_v2_image_size(rbd_dev);
4008 if (rbd_dev->header.obj_order != obj_order) {
4012 rbd_update_mapping_size(rbd_dev);
4014 ret = rbd_dev_v2_snap_context(rbd_dev, hver);
4015 dout("rbd_dev_v2_snap_context returned %d\n", ret);
4018 ret = rbd_dev_snaps_update(rbd_dev);
4019 dout("rbd_dev_snaps_update returned %d\n", ret);
4023 up_write(&rbd_dev->header_rwsem);
4029 * Scan the rbd device's current snapshot list and compare it to the
4030 * newly-received snapshot context. Remove any existing snapshots
4031 * not present in the new snapshot context. Add a new snapshot for
4032 * any snaphots in the snapshot context not in the current list.
4033 * And verify there are no changes to snapshots we already know
4036 * Assumes the snapshots in the snapshot context are sorted by
4037 * snapshot id, highest id first. (Snapshots in the rbd_dev's list
4038 * are also maintained in that order.)
4040 * Note that any error occurs while updating the snapshot list
4041 * aborts the update, and the entire list is cleared. The snapshot
4042 * list becomes inconsistent at that point anyway, so it might as
4045 static int rbd_dev_snaps_update(struct rbd_device *rbd_dev)
4047 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
4048 const u32 snap_count = snapc->num_snaps;
4049 struct list_head *head = &rbd_dev->snaps;
4050 struct list_head *links = head->next;
4054 dout("%s: snap count is %u\n", __func__, (unsigned int)snap_count);
4055 while (index < snap_count || links != head) {
4057 struct rbd_snap *snap;
4060 u64 snap_features = 0;
4062 snap_id = index < snap_count ? snapc->snaps[index]
4064 snap = links != head ? list_entry(links, struct rbd_snap, node)
4066 rbd_assert(!snap || snap->id != CEPH_NOSNAP);
4068 if (snap_id == CEPH_NOSNAP || (snap && snap->id > snap_id)) {
4069 struct list_head *next = links->next;
4072 * A previously-existing snapshot is not in
4073 * the new snap context.
4075 * If the now-missing snapshot is the one
4076 * the image represents, clear its existence
4077 * flag so we can avoid sending any more
4080 if (rbd_dev->spec->snap_id == snap->id)
4081 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
4082 dout("removing %ssnap id %llu\n",
4083 rbd_dev->spec->snap_id == snap->id ?
4085 (unsigned long long)snap->id);
4087 list_del(&snap->node);
4088 rbd_snap_destroy(snap);
4090 /* Done with this list entry; advance */
4096 snap_name = rbd_dev_snap_info(rbd_dev, index,
4097 &snap_size, &snap_features);
4098 if (IS_ERR(snap_name)) {
4099 ret = PTR_ERR(snap_name);
4100 dout("failed to get snap info, error %d\n", ret);
4104 dout("entry %u: snap_id = %llu\n", (unsigned int)snap_count,
4105 (unsigned long long)snap_id);
4106 if (!snap || (snap_id != CEPH_NOSNAP && snap->id < snap_id)) {
4107 struct rbd_snap *new_snap;
4109 /* We haven't seen this snapshot before */
4111 new_snap = rbd_snap_create(rbd_dev, snap_name,
4112 snap_id, snap_size, snap_features);
4113 if (IS_ERR(new_snap)) {
4114 ret = PTR_ERR(new_snap);
4115 dout(" failed to add dev, error %d\n", ret);
4119 /* New goes before existing, or at end of list */
4121 dout(" added dev%s\n", snap ? "" : " at end\n");
4123 list_add_tail(&new_snap->node, &snap->node);
4125 list_add_tail(&new_snap->node, head);
4127 /* Already have this one */
4129 dout(" already present\n");
4131 rbd_assert(snap->size == snap_size);
4132 rbd_assert(!strcmp(snap->name, snap_name));
4133 rbd_assert(snap->features == snap_features);
4135 /* Done with this list entry; advance */
4137 links = links->next;
4140 /* Advance to the next entry in the snapshot context */
4144 dout("%s: done\n", __func__);
4148 rbd_remove_all_snaps(rbd_dev);
4153 static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
4158 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
4160 dev = &rbd_dev->dev;
4161 dev->bus = &rbd_bus_type;
4162 dev->type = &rbd_device_type;
4163 dev->parent = &rbd_root_dev;
4164 dev->release = rbd_dev_release;
4165 dev_set_name(dev, "%d", rbd_dev->dev_id);
4166 ret = device_register(dev);
4168 mutex_unlock(&ctl_mutex);
4173 static void rbd_bus_del_dev(struct rbd_device *rbd_dev)
4175 device_unregister(&rbd_dev->dev);
4178 static atomic64_t rbd_dev_id_max = ATOMIC64_INIT(0);
4181 * Get a unique rbd identifier for the given new rbd_dev, and add
4182 * the rbd_dev to the global list. The minimum rbd id is 1.
4184 static void rbd_dev_id_get(struct rbd_device *rbd_dev)
4186 rbd_dev->dev_id = atomic64_inc_return(&rbd_dev_id_max);
4188 spin_lock(&rbd_dev_list_lock);
4189 list_add_tail(&rbd_dev->node, &rbd_dev_list);
4190 spin_unlock(&rbd_dev_list_lock);
4191 dout("rbd_dev %p given dev id %llu\n", rbd_dev,
4192 (unsigned long long) rbd_dev->dev_id);
4196 * Remove an rbd_dev from the global list, and record that its
4197 * identifier is no longer in use.
4199 static void rbd_dev_id_put(struct rbd_device *rbd_dev)
4201 struct list_head *tmp;
4202 int rbd_id = rbd_dev->dev_id;
4205 rbd_assert(rbd_id > 0);
4207 dout("rbd_dev %p released dev id %llu\n", rbd_dev,
4208 (unsigned long long) rbd_dev->dev_id);
4209 spin_lock(&rbd_dev_list_lock);
4210 list_del_init(&rbd_dev->node);
4213 * If the id being "put" is not the current maximum, there
4214 * is nothing special we need to do.
4216 if (rbd_id != atomic64_read(&rbd_dev_id_max)) {
4217 spin_unlock(&rbd_dev_list_lock);
4222 * We need to update the current maximum id. Search the
4223 * list to find out what it is. We're more likely to find
4224 * the maximum at the end, so search the list backward.
4227 list_for_each_prev(tmp, &rbd_dev_list) {
4228 struct rbd_device *rbd_dev;
4230 rbd_dev = list_entry(tmp, struct rbd_device, node);
4231 if (rbd_dev->dev_id > max_id)
4232 max_id = rbd_dev->dev_id;
4234 spin_unlock(&rbd_dev_list_lock);
4237 * The max id could have been updated by rbd_dev_id_get(), in
4238 * which case it now accurately reflects the new maximum.
4239 * Be careful not to overwrite the maximum value in that
4242 atomic64_cmpxchg(&rbd_dev_id_max, rbd_id, max_id);
4243 dout(" max dev id has been reset\n");
4247 * Skips over white space at *buf, and updates *buf to point to the
4248 * first found non-space character (if any). Returns the length of
4249 * the token (string of non-white space characters) found. Note
4250 * that *buf must be terminated with '\0'.
4252 static inline size_t next_token(const char **buf)
4255 * These are the characters that produce nonzero for
4256 * isspace() in the "C" and "POSIX" locales.
4258 const char *spaces = " \f\n\r\t\v";
4260 *buf += strspn(*buf, spaces); /* Find start of token */
4262 return strcspn(*buf, spaces); /* Return token length */
4266 * Finds the next token in *buf, and if the provided token buffer is
4267 * big enough, copies the found token into it. The result, if
4268 * copied, is guaranteed to be terminated with '\0'. Note that *buf
4269 * must be terminated with '\0' on entry.
4271 * Returns the length of the token found (not including the '\0').
4272 * Return value will be 0 if no token is found, and it will be >=
4273 * token_size if the token would not fit.
4275 * The *buf pointer will be updated to point beyond the end of the
4276 * found token. Note that this occurs even if the token buffer is
4277 * too small to hold it.
4279 static inline size_t copy_token(const char **buf,
4285 len = next_token(buf);
4286 if (len < token_size) {
4287 memcpy(token, *buf, len);
4288 *(token + len) = '\0';
4296 * Finds the next token in *buf, dynamically allocates a buffer big
4297 * enough to hold a copy of it, and copies the token into the new
4298 * buffer. The copy is guaranteed to be terminated with '\0'. Note
4299 * that a duplicate buffer is created even for a zero-length token.
4301 * Returns a pointer to the newly-allocated duplicate, or a null
4302 * pointer if memory for the duplicate was not available. If
4303 * the lenp argument is a non-null pointer, the length of the token
4304 * (not including the '\0') is returned in *lenp.
4306 * If successful, the *buf pointer will be updated to point beyond
4307 * the end of the found token.
4309 * Note: uses GFP_KERNEL for allocation.
4311 static inline char *dup_token(const char **buf, size_t *lenp)
4316 len = next_token(buf);
4317 dup = kmemdup(*buf, len + 1, GFP_KERNEL);
4320 *(dup + len) = '\0';
4330 * Parse the options provided for an "rbd add" (i.e., rbd image
4331 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
4332 * and the data written is passed here via a NUL-terminated buffer.
4333 * Returns 0 if successful or an error code otherwise.
4335 * The information extracted from these options is recorded in
4336 * the other parameters which return dynamically-allocated
4339 * The address of a pointer that will refer to a ceph options
4340 * structure. Caller must release the returned pointer using
4341 * ceph_destroy_options() when it is no longer needed.
4343 * Address of an rbd options pointer. Fully initialized by
4344 * this function; caller must release with kfree().
4346 * Address of an rbd image specification pointer. Fully
4347 * initialized by this function based on parsed options.
4348 * Caller must release with rbd_spec_put().
4350 * The options passed take this form:
4351 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
4354 * A comma-separated list of one or more monitor addresses.
4355 * A monitor address is an ip address, optionally followed
4356 * by a port number (separated by a colon).
4357 * I.e.: ip1[:port1][,ip2[:port2]...]
4359 * A comma-separated list of ceph and/or rbd options.
4361 * The name of the rados pool containing the rbd image.
4363 * The name of the image in that pool to map.
4365 * An optional snapshot id. If provided, the mapping will
4366 * present data from the image at the time that snapshot was
4367 * created. The image head is used if no snapshot id is
4368 * provided. Snapshot mappings are always read-only.
4370 static int rbd_add_parse_args(const char *buf,
4371 struct ceph_options **ceph_opts,
4372 struct rbd_options **opts,
4373 struct rbd_spec **rbd_spec)
4377 const char *mon_addrs;
4379 size_t mon_addrs_size;
4380 struct rbd_spec *spec = NULL;
4381 struct rbd_options *rbd_opts = NULL;
4382 struct ceph_options *copts;
4385 /* The first four tokens are required */
4387 len = next_token(&buf);
4389 rbd_warn(NULL, "no monitor address(es) provided");
4393 mon_addrs_size = len + 1;
4397 options = dup_token(&buf, NULL);
4401 rbd_warn(NULL, "no options provided");
4405 spec = rbd_spec_alloc();
4409 spec->pool_name = dup_token(&buf, NULL);
4410 if (!spec->pool_name)
4412 if (!*spec->pool_name) {
4413 rbd_warn(NULL, "no pool name provided");
4417 spec->image_name = dup_token(&buf, NULL);
4418 if (!spec->image_name)
4420 if (!*spec->image_name) {
4421 rbd_warn(NULL, "no image name provided");
4426 * Snapshot name is optional; default is to use "-"
4427 * (indicating the head/no snapshot).
4429 len = next_token(&buf);
4431 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
4432 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
4433 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
4434 ret = -ENAMETOOLONG;
4437 snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
4440 *(snap_name + len) = '\0';
4441 spec->snap_name = snap_name;
4443 /* Initialize all rbd options to the defaults */
4445 rbd_opts = kzalloc(sizeof (*rbd_opts), GFP_KERNEL);
4449 rbd_opts->read_only = RBD_READ_ONLY_DEFAULT;
4451 copts = ceph_parse_options(options, mon_addrs,
4452 mon_addrs + mon_addrs_size - 1,
4453 parse_rbd_opts_token, rbd_opts);
4454 if (IS_ERR(copts)) {
4455 ret = PTR_ERR(copts);
4476 * An rbd format 2 image has a unique identifier, distinct from the
4477 * name given to it by the user. Internally, that identifier is
4478 * what's used to specify the names of objects related to the image.
4480 * A special "rbd id" object is used to map an rbd image name to its
4481 * id. If that object doesn't exist, then there is no v2 rbd image
4482 * with the supplied name.
4484 * This function will record the given rbd_dev's image_id field if
4485 * it can be determined, and in that case will return 0. If any
4486 * errors occur a negative errno will be returned and the rbd_dev's
4487 * image_id field will be unchanged (and should be NULL).
4489 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
4498 * When probing a parent image, the image id is already
4499 * known (and the image name likely is not). There's no
4500 * need to fetch the image id again in this case. We
4501 * do still need to set the image format though.
4503 if (rbd_dev->spec->image_id) {
4504 rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
4510 * First, see if the format 2 image id file exists, and if
4511 * so, get the image's persistent id from it.
4513 size = sizeof (RBD_ID_PREFIX) + strlen(rbd_dev->spec->image_name);
4514 object_name = kmalloc(size, GFP_NOIO);
4517 sprintf(object_name, "%s%s", RBD_ID_PREFIX, rbd_dev->spec->image_name);
4518 dout("rbd id object name is %s\n", object_name);
4520 /* Response will be an encoded string, which includes a length */
4522 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
4523 response = kzalloc(size, GFP_NOIO);
4529 /* If it doesn't exist we'll assume it's a format 1 image */
4531 ret = rbd_obj_method_sync(rbd_dev, object_name,
4532 "rbd", "get_id", NULL, 0,
4533 response, RBD_IMAGE_ID_LEN_MAX, NULL);
4534 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4535 if (ret == -ENOENT) {
4536 image_id = kstrdup("", GFP_KERNEL);
4537 ret = image_id ? 0 : -ENOMEM;
4539 rbd_dev->image_format = 1;
4540 } else if (ret > sizeof (__le32)) {
4543 image_id = ceph_extract_encoded_string(&p, p + ret,
4545 ret = IS_ERR(image_id) ? PTR_ERR(image_id) : 0;
4547 rbd_dev->image_format = 2;
4553 rbd_dev->spec->image_id = image_id;
4554 dout("image_id is %s\n", image_id);
4563 static int rbd_dev_v1_probe(struct rbd_device *rbd_dev)
4568 /* Record the header object name for this rbd image. */
4570 size = strlen(rbd_dev->spec->image_name) + sizeof (RBD_SUFFIX);
4571 rbd_dev->header_name = kmalloc(size, GFP_KERNEL);
4572 if (!rbd_dev->header_name) {
4576 sprintf(rbd_dev->header_name, "%s%s",
4577 rbd_dev->spec->image_name, RBD_SUFFIX);
4579 /* Populate rbd image metadata */
4581 ret = rbd_read_header(rbd_dev, &rbd_dev->header);
4585 /* Version 1 images have no parent (no layering) */
4587 rbd_dev->parent_spec = NULL;
4588 rbd_dev->parent_overlap = 0;
4590 dout("discovered version 1 image, header name is %s\n",
4591 rbd_dev->header_name);
4596 kfree(rbd_dev->header_name);
4597 rbd_dev->header_name = NULL;
4598 kfree(rbd_dev->spec->image_id);
4599 rbd_dev->spec->image_id = NULL;
4604 static int rbd_dev_v2_probe(struct rbd_device *rbd_dev)
4611 * Image id was filled in by the caller. Record the header
4612 * object name for this rbd image.
4614 size = sizeof (RBD_HEADER_PREFIX) + strlen(rbd_dev->spec->image_id);
4615 rbd_dev->header_name = kmalloc(size, GFP_KERNEL);
4616 if (!rbd_dev->header_name)
4618 sprintf(rbd_dev->header_name, "%s%s",
4619 RBD_HEADER_PREFIX, rbd_dev->spec->image_id);
4621 /* Get the size and object order for the image */
4622 ret = rbd_dev_v2_image_size(rbd_dev);
4626 /* Get the object prefix (a.k.a. block_name) for the image */
4628 ret = rbd_dev_v2_object_prefix(rbd_dev);
4632 /* Get the and check features for the image */
4634 ret = rbd_dev_v2_features(rbd_dev);
4638 /* If the image supports layering, get the parent info */
4640 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
4641 ret = rbd_dev_v2_parent_info(rbd_dev);
4644 rbd_warn(rbd_dev, "WARNING: kernel support for "
4645 "layered rbd images is EXPERIMENTAL!");
4648 /* If the image supports fancy striping, get its parameters */
4650 if (rbd_dev->header.features & RBD_FEATURE_STRIPINGV2) {
4651 ret = rbd_dev_v2_striping_info(rbd_dev);
4656 /* crypto and compression type aren't (yet) supported for v2 images */
4658 rbd_dev->header.crypt_type = 0;
4659 rbd_dev->header.comp_type = 0;
4661 /* Get the snapshot context, plus the header version */
4663 ret = rbd_dev_v2_snap_context(rbd_dev, &ver);
4666 rbd_dev->header.obj_version = ver;
4668 dout("discovered version 2 image, header name is %s\n",
4669 rbd_dev->header_name);
4673 rbd_dev->parent_overlap = 0;
4674 rbd_spec_put(rbd_dev->parent_spec);
4675 rbd_dev->parent_spec = NULL;
4676 kfree(rbd_dev->header_name);
4677 rbd_dev->header_name = NULL;
4678 kfree(rbd_dev->header.object_prefix);
4679 rbd_dev->header.object_prefix = NULL;
4684 static int rbd_dev_probe_finish(struct rbd_device *rbd_dev)
4686 struct rbd_device *parent = NULL;
4687 struct rbd_spec *parent_spec = NULL;
4688 struct rbd_client *rbdc = NULL;
4691 /* no need to lock here, as rbd_dev is not registered yet */
4692 ret = rbd_dev_snaps_update(rbd_dev);
4696 ret = rbd_dev_probe_update_spec(rbd_dev);
4700 ret = rbd_dev_set_mapping(rbd_dev);
4704 /* generate unique id: find highest unique id, add one */
4705 rbd_dev_id_get(rbd_dev);
4707 /* Fill in the device name, now that we have its id. */
4708 BUILD_BUG_ON(DEV_NAME_LEN
4709 < sizeof (RBD_DRV_NAME) + MAX_INT_FORMAT_WIDTH);
4710 sprintf(rbd_dev->name, "%s%d", RBD_DRV_NAME, rbd_dev->dev_id);
4712 /* Get our block major device number. */
4714 ret = register_blkdev(0, rbd_dev->name);
4717 rbd_dev->major = ret;
4719 /* Set up the blkdev mapping. */
4721 ret = rbd_init_disk(rbd_dev);
4723 goto err_out_blkdev;
4725 ret = rbd_bus_add_dev(rbd_dev);
4730 * At this point cleanup in the event of an error is the job
4731 * of the sysfs code (initiated by rbd_bus_del_dev()).
4733 /* Probe the parent if there is one */
4735 if (rbd_dev->parent_spec) {
4737 * We need to pass a reference to the client and the
4738 * parent spec when creating the parent rbd_dev.
4739 * Images related by parent/child relationships
4740 * always share both.
4742 parent_spec = rbd_spec_get(rbd_dev->parent_spec);
4743 rbdc = __rbd_get_client(rbd_dev->rbd_client);
4745 parent = rbd_dev_create(rbdc, parent_spec);
4750 rbdc = NULL; /* parent now owns reference */
4751 parent_spec = NULL; /* parent now owns reference */
4752 ret = rbd_dev_probe(parent);
4754 goto err_out_parent;
4755 rbd_dev->parent = parent;
4758 ret = rbd_dev_header_watch_sync(rbd_dev, 1);
4762 /* Everything's ready. Announce the disk to the world. */
4764 add_disk(rbd_dev->disk);
4766 pr_info("%s: added with size 0x%llx\n", rbd_dev->disk->disk_name,
4767 (unsigned long long) rbd_dev->mapping.size);
4772 rbd_dev_destroy(parent);
4774 rbd_spec_put(parent_spec);
4775 rbd_put_client(rbdc);
4777 /* this will also clean up rest of rbd_dev stuff */
4779 rbd_bus_del_dev(rbd_dev);
4783 rbd_free_disk(rbd_dev);
4785 unregister_blkdev(rbd_dev->major, rbd_dev->name);
4787 rbd_dev_id_put(rbd_dev);
4789 rbd_remove_all_snaps(rbd_dev);
4795 * Probe for the existence of the header object for the given rbd
4796 * device. For format 2 images this includes determining the image
4799 static int rbd_dev_probe(struct rbd_device *rbd_dev)
4804 * Get the id from the image id object. If it's not a
4805 * format 2 image, we'll get ENOENT back, and we'll assume
4806 * it's a format 1 image.
4808 ret = rbd_dev_image_id(rbd_dev);
4811 rbd_assert(rbd_dev->spec->image_id);
4812 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
4814 if (rbd_dev->image_format == 1)
4815 ret = rbd_dev_v1_probe(rbd_dev);
4817 ret = rbd_dev_v2_probe(rbd_dev);
4821 ret = rbd_dev_probe_finish(rbd_dev);
4823 rbd_header_free(&rbd_dev->header);
4827 kfree(rbd_dev->spec->image_id);
4828 rbd_dev->spec->image_id = NULL;
4830 dout("probe failed, returning %d\n", ret);
4835 static ssize_t rbd_add(struct bus_type *bus,
4839 struct rbd_device *rbd_dev = NULL;
4840 struct ceph_options *ceph_opts = NULL;
4841 struct rbd_options *rbd_opts = NULL;
4842 struct rbd_spec *spec = NULL;
4843 struct rbd_client *rbdc;
4844 struct ceph_osd_client *osdc;
4847 if (!try_module_get(THIS_MODULE))
4850 /* parse add command */
4851 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
4853 goto err_out_module;
4855 rbdc = rbd_get_client(ceph_opts);
4860 ceph_opts = NULL; /* rbd_dev client now owns this */
4863 osdc = &rbdc->client->osdc;
4864 rc = ceph_pg_poolid_by_name(osdc->osdmap, spec->pool_name);
4866 goto err_out_client;
4867 spec->pool_id = (u64) rc;
4869 /* The ceph file layout needs to fit pool id in 32 bits */
4871 if (WARN_ON(spec->pool_id > (u64) U32_MAX)) {
4873 goto err_out_client;
4876 rbd_dev = rbd_dev_create(rbdc, spec);
4878 goto err_out_client;
4879 rbdc = NULL; /* rbd_dev now owns this */
4880 spec = NULL; /* rbd_dev now owns this */
4882 rbd_dev->mapping.read_only = rbd_opts->read_only;
4884 rbd_opts = NULL; /* done with this */
4886 rc = rbd_dev_probe(rbd_dev);
4888 goto err_out_rbd_dev;
4892 rbd_dev_destroy(rbd_dev);
4894 rbd_put_client(rbdc);
4897 ceph_destroy_options(ceph_opts);
4901 module_put(THIS_MODULE);
4903 dout("Error adding device %s\n", buf);
4905 return (ssize_t) rc;
4908 static struct rbd_device *__rbd_get_dev(unsigned long dev_id)
4910 struct list_head *tmp;
4911 struct rbd_device *rbd_dev;
4913 spin_lock(&rbd_dev_list_lock);
4914 list_for_each(tmp, &rbd_dev_list) {
4915 rbd_dev = list_entry(tmp, struct rbd_device, node);
4916 if (rbd_dev->dev_id == dev_id) {
4917 spin_unlock(&rbd_dev_list_lock);
4921 spin_unlock(&rbd_dev_list_lock);
4925 static void rbd_dev_release(struct device *dev)
4927 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4929 if (rbd_dev->watch_event)
4930 rbd_dev_header_watch_sync(rbd_dev, 0);
4932 /* clean up and free blkdev */
4933 rbd_free_disk(rbd_dev);
4934 unregister_blkdev(rbd_dev->major, rbd_dev->name);
4936 /* release allocated disk header fields */
4937 rbd_header_free(&rbd_dev->header);
4939 /* done with the id, and with the rbd_dev */
4940 rbd_dev_id_put(rbd_dev);
4941 rbd_assert(rbd_dev->rbd_client != NULL);
4942 rbd_dev_destroy(rbd_dev);
4944 /* release module ref */
4945 module_put(THIS_MODULE);
4948 static void __rbd_remove(struct rbd_device *rbd_dev)
4950 rbd_remove_all_snaps(rbd_dev);
4951 rbd_bus_del_dev(rbd_dev);
4954 static ssize_t rbd_remove(struct bus_type *bus,
4958 struct rbd_device *rbd_dev = NULL;
4963 rc = strict_strtoul(buf, 10, &ul);
4967 /* convert to int; abort if we lost anything in the conversion */
4968 target_id = (int) ul;
4969 if (target_id != ul)
4972 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
4974 rbd_dev = __rbd_get_dev(target_id);
4980 spin_lock_irq(&rbd_dev->lock);
4981 if (rbd_dev->open_count)
4984 set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
4985 spin_unlock_irq(&rbd_dev->lock);
4989 while (rbd_dev->parent_spec) {
4990 struct rbd_device *first = rbd_dev;
4991 struct rbd_device *second = first->parent;
4992 struct rbd_device *third;
4995 * Follow to the parent with no grandparent and
4998 while (second && (third = second->parent)) {
5002 __rbd_remove(second);
5003 rbd_spec_put(first->parent_spec);
5004 first->parent_spec = NULL;
5005 first->parent_overlap = 0;
5006 first->parent = NULL;
5008 __rbd_remove(rbd_dev);
5011 mutex_unlock(&ctl_mutex);
5017 * create control files in sysfs
5020 static int rbd_sysfs_init(void)
5024 ret = device_register(&rbd_root_dev);
5028 ret = bus_register(&rbd_bus_type);
5030 device_unregister(&rbd_root_dev);
5035 static void rbd_sysfs_cleanup(void)
5037 bus_unregister(&rbd_bus_type);
5038 device_unregister(&rbd_root_dev);
5041 static int __init rbd_init(void)
5045 if (!libceph_compatible(NULL)) {
5046 rbd_warn(NULL, "libceph incompatibility (quitting)");
5050 rc = rbd_sysfs_init();
5053 pr_info("loaded " RBD_DRV_NAME_LONG "\n");
5057 static void __exit rbd_exit(void)
5059 rbd_sysfs_cleanup();
5062 module_init(rbd_init);
5063 module_exit(rbd_exit);
5065 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
5066 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
5067 MODULE_DESCRIPTION("rados block device");
5069 /* following authorship retained from original osdblk.c */
5070 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
5072 MODULE_LICENSE("GPL");