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 /* It might be useful to have this defined elsewhere too */
57 #define U64_MAX ((u64) (~0ULL))
59 #define RBD_DRV_NAME "rbd"
60 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
62 #define RBD_MINORS_PER_MAJOR 256 /* max minors per blkdev */
64 #define RBD_SNAP_DEV_NAME_PREFIX "snap_"
65 #define RBD_MAX_SNAP_NAME_LEN \
66 (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
68 #define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */
69 #define RBD_MAX_OPT_LEN 1024
71 #define RBD_SNAP_HEAD_NAME "-"
73 #define RBD_IMAGE_ID_LEN_MAX 64
74 #define RBD_OBJ_PREFIX_LEN_MAX 64
78 #define RBD_FEATURE_LAYERING 1
80 /* Features supported by this (client software) implementation. */
82 #define RBD_FEATURES_ALL (0)
85 * An RBD device name will be "rbd#", where the "rbd" comes from
86 * RBD_DRV_NAME above, and # is a unique integer identifier.
87 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
88 * enough to hold all possible device names.
90 #define DEV_NAME_LEN 32
91 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1)
93 #define RBD_READ_ONLY_DEFAULT false
96 * block device image metadata (in-memory version)
98 struct rbd_image_header {
99 /* These four fields never change for a given rbd image */
106 /* The remaining fields need to be updated occasionally */
108 struct ceph_snap_context *snapc;
116 * An rbd image specification.
118 * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
128 size_t image_name_len;
141 * an instance of the client. multiple devices may share an rbd client.
144 struct ceph_client *client;
146 struct list_head node;
150 * a request completion status
152 struct rbd_req_status {
159 * a collection of requests
161 struct rbd_req_coll {
165 struct rbd_req_status status[0];
169 * a single io request
172 struct request *rq; /* blk layer request */
173 struct bio *bio; /* cloned bio */
174 struct page **pages; /* list of used pages */
177 struct rbd_req_coll *coll;
184 struct list_head node;
199 int dev_id; /* blkdev unique id */
201 int major; /* blkdev assigned major */
202 struct gendisk *disk; /* blkdev's gendisk and rq */
204 u32 image_format; /* Either 1 or 2 */
205 struct rbd_client *rbd_client;
207 char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
209 spinlock_t lock; /* queue lock */
211 struct rbd_image_header header;
213 struct rbd_spec *spec;
217 struct ceph_osd_event *watch_event;
218 struct ceph_osd_request *watch_request;
220 /* protects updating the header */
221 struct rw_semaphore header_rwsem;
223 struct rbd_mapping mapping;
225 struct list_head node;
227 /* list of snapshots */
228 struct list_head snaps;
234 static DEFINE_MUTEX(ctl_mutex); /* Serialize open/close/setup/teardown */
236 static LIST_HEAD(rbd_dev_list); /* devices */
237 static DEFINE_SPINLOCK(rbd_dev_list_lock);
239 static LIST_HEAD(rbd_client_list); /* clients */
240 static DEFINE_SPINLOCK(rbd_client_list_lock);
242 static int rbd_dev_snaps_update(struct rbd_device *rbd_dev);
243 static int rbd_dev_snaps_register(struct rbd_device *rbd_dev);
245 static void rbd_dev_release(struct device *dev);
246 static void rbd_remove_snap_dev(struct rbd_snap *snap);
248 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
250 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
253 static struct bus_attribute rbd_bus_attrs[] = {
254 __ATTR(add, S_IWUSR, NULL, rbd_add),
255 __ATTR(remove, S_IWUSR, NULL, rbd_remove),
259 static struct bus_type rbd_bus_type = {
261 .bus_attrs = rbd_bus_attrs,
264 static void rbd_root_dev_release(struct device *dev)
268 static struct device rbd_root_dev = {
270 .release = rbd_root_dev_release,
274 #define rbd_assert(expr) \
275 if (unlikely(!(expr))) { \
276 printk(KERN_ERR "\nAssertion failure in %s() " \
278 "\trbd_assert(%s);\n\n", \
279 __func__, __LINE__, #expr); \
282 #else /* !RBD_DEBUG */
283 # define rbd_assert(expr) ((void) 0)
284 #endif /* !RBD_DEBUG */
286 static struct device *rbd_get_dev(struct rbd_device *rbd_dev)
288 return get_device(&rbd_dev->dev);
291 static void rbd_put_dev(struct rbd_device *rbd_dev)
293 put_device(&rbd_dev->dev);
296 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver);
297 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver);
299 static int rbd_open(struct block_device *bdev, fmode_t mode)
301 struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
303 if ((mode & FMODE_WRITE) && rbd_dev->mapping.read_only)
306 rbd_get_dev(rbd_dev);
307 set_device_ro(bdev, rbd_dev->mapping.read_only);
312 static int rbd_release(struct gendisk *disk, fmode_t mode)
314 struct rbd_device *rbd_dev = disk->private_data;
316 rbd_put_dev(rbd_dev);
321 static const struct block_device_operations rbd_bd_ops = {
322 .owner = THIS_MODULE,
324 .release = rbd_release,
328 * Initialize an rbd client instance.
331 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
333 struct rbd_client *rbdc;
336 dout("rbd_client_create\n");
337 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
341 kref_init(&rbdc->kref);
342 INIT_LIST_HEAD(&rbdc->node);
344 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
346 rbdc->client = ceph_create_client(ceph_opts, rbdc, 0, 0);
347 if (IS_ERR(rbdc->client))
349 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
351 ret = ceph_open_session(rbdc->client);
355 spin_lock(&rbd_client_list_lock);
356 list_add_tail(&rbdc->node, &rbd_client_list);
357 spin_unlock(&rbd_client_list_lock);
359 mutex_unlock(&ctl_mutex);
361 dout("rbd_client_create created %p\n", rbdc);
365 ceph_destroy_client(rbdc->client);
367 mutex_unlock(&ctl_mutex);
371 ceph_destroy_options(ceph_opts);
376 * Find a ceph client with specific addr and configuration. If
377 * found, bump its reference count.
379 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
381 struct rbd_client *client_node;
384 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
387 spin_lock(&rbd_client_list_lock);
388 list_for_each_entry(client_node, &rbd_client_list, node) {
389 if (!ceph_compare_options(ceph_opts, client_node->client)) {
390 kref_get(&client_node->kref);
395 spin_unlock(&rbd_client_list_lock);
397 return found ? client_node : NULL;
407 /* string args above */
410 /* Boolean args above */
414 static match_table_t rbd_opts_tokens = {
416 /* string args above */
417 {Opt_read_only, "read_only"},
418 {Opt_read_only, "ro"}, /* Alternate spelling */
419 {Opt_read_write, "read_write"},
420 {Opt_read_write, "rw"}, /* Alternate spelling */
421 /* Boolean args above */
425 static int parse_rbd_opts_token(char *c, void *private)
427 struct rbd_options *rbd_opts = private;
428 substring_t argstr[MAX_OPT_ARGS];
429 int token, intval, ret;
431 token = match_token(c, rbd_opts_tokens, argstr);
435 if (token < Opt_last_int) {
436 ret = match_int(&argstr[0], &intval);
438 pr_err("bad mount option arg (not int) "
442 dout("got int token %d val %d\n", token, intval);
443 } else if (token > Opt_last_int && token < Opt_last_string) {
444 dout("got string token %d val %s\n", token,
446 } else if (token > Opt_last_string && token < Opt_last_bool) {
447 dout("got Boolean token %d\n", token);
449 dout("got token %d\n", token);
454 rbd_opts->read_only = true;
457 rbd_opts->read_only = false;
467 * Get a ceph client with specific addr and configuration, if one does
468 * not exist create it.
470 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
472 struct rbd_client *rbdc;
474 rbdc = rbd_client_find(ceph_opts);
475 if (rbdc) /* using an existing client */
476 ceph_destroy_options(ceph_opts);
478 rbdc = rbd_client_create(ceph_opts);
484 * Destroy ceph client
486 * Caller must hold rbd_client_list_lock.
488 static void rbd_client_release(struct kref *kref)
490 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
492 dout("rbd_release_client %p\n", rbdc);
493 spin_lock(&rbd_client_list_lock);
494 list_del(&rbdc->node);
495 spin_unlock(&rbd_client_list_lock);
497 ceph_destroy_client(rbdc->client);
502 * Drop reference to ceph client node. If it's not referenced anymore, release
505 static void rbd_put_client(struct rbd_client *rbdc)
507 kref_put(&rbdc->kref, rbd_client_release);
511 * Destroy requests collection
513 static void rbd_coll_release(struct kref *kref)
515 struct rbd_req_coll *coll =
516 container_of(kref, struct rbd_req_coll, kref);
518 dout("rbd_coll_release %p\n", coll);
522 static bool rbd_image_format_valid(u32 image_format)
524 return image_format == 1 || image_format == 2;
527 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
532 /* The header has to start with the magic rbd header text */
533 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
536 /* The bio layer requires at least sector-sized I/O */
538 if (ondisk->options.order < SECTOR_SHIFT)
541 /* If we use u64 in a few spots we may be able to loosen this */
543 if (ondisk->options.order > 8 * sizeof (int) - 1)
547 * The size of a snapshot header has to fit in a size_t, and
548 * that limits the number of snapshots.
550 snap_count = le32_to_cpu(ondisk->snap_count);
551 size = SIZE_MAX - sizeof (struct ceph_snap_context);
552 if (snap_count > size / sizeof (__le64))
556 * Not only that, but the size of the entire the snapshot
557 * header must also be representable in a size_t.
559 size -= snap_count * sizeof (__le64);
560 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
567 * Create a new header structure, translate header format from the on-disk
570 static int rbd_header_from_disk(struct rbd_image_header *header,
571 struct rbd_image_header_ondisk *ondisk)
578 memset(header, 0, sizeof (*header));
580 snap_count = le32_to_cpu(ondisk->snap_count);
582 len = strnlen(ondisk->object_prefix, sizeof (ondisk->object_prefix));
583 header->object_prefix = kmalloc(len + 1, GFP_KERNEL);
584 if (!header->object_prefix)
586 memcpy(header->object_prefix, ondisk->object_prefix, len);
587 header->object_prefix[len] = '\0';
590 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
592 /* Save a copy of the snapshot names */
594 if (snap_names_len > (u64) SIZE_MAX)
596 header->snap_names = kmalloc(snap_names_len, GFP_KERNEL);
597 if (!header->snap_names)
600 * Note that rbd_dev_v1_header_read() guarantees
601 * the ondisk buffer we're working with has
602 * snap_names_len bytes beyond the end of the
603 * snapshot id array, this memcpy() is safe.
605 memcpy(header->snap_names, &ondisk->snaps[snap_count],
608 /* Record each snapshot's size */
610 size = snap_count * sizeof (*header->snap_sizes);
611 header->snap_sizes = kmalloc(size, GFP_KERNEL);
612 if (!header->snap_sizes)
614 for (i = 0; i < snap_count; i++)
615 header->snap_sizes[i] =
616 le64_to_cpu(ondisk->snaps[i].image_size);
618 WARN_ON(ondisk->snap_names_len);
619 header->snap_names = NULL;
620 header->snap_sizes = NULL;
623 header->features = 0; /* No features support in v1 images */
624 header->obj_order = ondisk->options.order;
625 header->crypt_type = ondisk->options.crypt_type;
626 header->comp_type = ondisk->options.comp_type;
628 /* Allocate and fill in the snapshot context */
630 header->image_size = le64_to_cpu(ondisk->image_size);
631 size = sizeof (struct ceph_snap_context);
632 size += snap_count * sizeof (header->snapc->snaps[0]);
633 header->snapc = kzalloc(size, GFP_KERNEL);
637 atomic_set(&header->snapc->nref, 1);
638 header->snapc->seq = le64_to_cpu(ondisk->snap_seq);
639 header->snapc->num_snaps = snap_count;
640 for (i = 0; i < snap_count; i++)
641 header->snapc->snaps[i] =
642 le64_to_cpu(ondisk->snaps[i].id);
647 kfree(header->snap_sizes);
648 header->snap_sizes = NULL;
649 kfree(header->snap_names);
650 header->snap_names = NULL;
651 kfree(header->object_prefix);
652 header->object_prefix = NULL;
657 static int snap_by_name(struct rbd_device *rbd_dev, const char *snap_name)
660 struct rbd_snap *snap;
662 list_for_each_entry(snap, &rbd_dev->snaps, node) {
663 if (!strcmp(snap_name, snap->name)) {
664 rbd_dev->spec->snap_id = snap->id;
665 rbd_dev->mapping.size = snap->size;
666 rbd_dev->mapping.features = snap->features;
675 static int rbd_dev_set_mapping(struct rbd_device *rbd_dev)
679 if (!memcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME,
680 sizeof (RBD_SNAP_HEAD_NAME))) {
681 rbd_dev->spec->snap_id = CEPH_NOSNAP;
682 rbd_dev->mapping.size = rbd_dev->header.image_size;
683 rbd_dev->mapping.features = rbd_dev->header.features;
686 ret = snap_by_name(rbd_dev, rbd_dev->spec->snap_name);
689 rbd_dev->mapping.read_only = true;
691 rbd_dev->exists = true;
696 static void rbd_header_free(struct rbd_image_header *header)
698 kfree(header->object_prefix);
699 header->object_prefix = NULL;
700 kfree(header->snap_sizes);
701 header->snap_sizes = NULL;
702 kfree(header->snap_names);
703 header->snap_names = NULL;
704 ceph_put_snap_context(header->snapc);
705 header->snapc = NULL;
708 static char *rbd_segment_name(struct rbd_device *rbd_dev, u64 offset)
714 name = kmalloc(RBD_MAX_SEG_NAME_LEN + 1, GFP_NOIO);
717 segment = offset >> rbd_dev->header.obj_order;
718 ret = snprintf(name, RBD_MAX_SEG_NAME_LEN, "%s.%012llx",
719 rbd_dev->header.object_prefix, segment);
720 if (ret < 0 || ret >= RBD_MAX_SEG_NAME_LEN) {
721 pr_err("error formatting segment name for #%llu (%d)\n",
730 static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset)
732 u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
734 return offset & (segment_size - 1);
737 static u64 rbd_segment_length(struct rbd_device *rbd_dev,
738 u64 offset, u64 length)
740 u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
742 offset &= segment_size - 1;
744 rbd_assert(length <= U64_MAX - offset);
745 if (offset + length > segment_size)
746 length = segment_size - offset;
751 static int rbd_get_num_segments(struct rbd_image_header *header,
759 if (len - 1 > U64_MAX - ofs)
762 start_seg = ofs >> header->obj_order;
763 end_seg = (ofs + len - 1) >> header->obj_order;
765 return end_seg - start_seg + 1;
769 * returns the size of an object in the image
771 static u64 rbd_obj_bytes(struct rbd_image_header *header)
773 return 1 << header->obj_order;
780 static void bio_chain_put(struct bio *chain)
786 chain = chain->bi_next;
792 * zeros a bio chain, starting at specific offset
794 static void zero_bio_chain(struct bio *chain, int start_ofs)
803 bio_for_each_segment(bv, chain, i) {
804 if (pos + bv->bv_len > start_ofs) {
805 int remainder = max(start_ofs - pos, 0);
806 buf = bvec_kmap_irq(bv, &flags);
807 memset(buf + remainder, 0,
808 bv->bv_len - remainder);
809 bvec_kunmap_irq(buf, &flags);
814 chain = chain->bi_next;
819 * Clone a portion of a bio, starting at the given byte offset
820 * and continuing for the number of bytes indicated.
822 static struct bio *bio_clone_range(struct bio *bio_src,
831 unsigned short end_idx;
835 /* Handle the easy case for the caller */
837 if (!offset && len == bio_src->bi_size)
838 return bio_clone(bio_src, gfpmask);
840 if (WARN_ON_ONCE(!len))
842 if (WARN_ON_ONCE(len > bio_src->bi_size))
844 if (WARN_ON_ONCE(offset > bio_src->bi_size - len))
847 /* Find first affected segment... */
850 __bio_for_each_segment(bv, bio_src, idx, 0) {
851 if (resid < bv->bv_len)
857 /* ...and the last affected segment */
860 __bio_for_each_segment(bv, bio_src, end_idx, idx) {
861 if (resid <= bv->bv_len)
865 vcnt = end_idx - idx + 1;
867 /* Build the clone */
869 bio = bio_alloc(gfpmask, (unsigned int) vcnt);
871 return NULL; /* ENOMEM */
873 bio->bi_bdev = bio_src->bi_bdev;
874 bio->bi_sector = bio_src->bi_sector + (offset >> SECTOR_SHIFT);
875 bio->bi_rw = bio_src->bi_rw;
876 bio->bi_flags |= 1 << BIO_CLONED;
879 * Copy over our part of the bio_vec, then update the first
880 * and last (or only) entries.
882 memcpy(&bio->bi_io_vec[0], &bio_src->bi_io_vec[idx],
883 vcnt * sizeof (struct bio_vec));
884 bio->bi_io_vec[0].bv_offset += voff;
886 bio->bi_io_vec[0].bv_len -= voff;
887 bio->bi_io_vec[vcnt - 1].bv_len = resid;
889 bio->bi_io_vec[0].bv_len = len;
900 * Clone a portion of a bio chain, starting at the given byte offset
901 * into the first bio in the source chain and continuing for the
902 * number of bytes indicated. The result is another bio chain of
903 * exactly the given length, or a null pointer on error.
905 * The bio_src and offset parameters are both in-out. On entry they
906 * refer to the first source bio and the offset into that bio where
907 * the start of data to be cloned is located.
909 * On return, bio_src is updated to refer to the bio in the source
910 * chain that contains first un-cloned byte, and *offset will
911 * contain the offset of that byte within that bio.
913 static struct bio *bio_chain_clone_range(struct bio **bio_src,
914 unsigned int *offset,
918 struct bio *bi = *bio_src;
919 unsigned int off = *offset;
920 struct bio *chain = NULL;
923 /* Build up a chain of clone bios up to the limit */
925 if (!bi || off >= bi->bi_size || !len)
926 return NULL; /* Nothing to clone */
930 unsigned int bi_size;
934 goto out_err; /* EINVAL; ran out of bio's */
935 bi_size = min_t(unsigned int, bi->bi_size - off, len);
936 bio = bio_clone_range(bi, off, bi_size, gfpmask);
938 goto out_err; /* ENOMEM */
944 if (off == bi->bi_size) {
955 bio_chain_put(chain);
961 * helpers for osd request op vectors.
963 static struct ceph_osd_req_op *rbd_create_rw_ops(int num_ops,
964 int opcode, u32 payload_len)
966 struct ceph_osd_req_op *ops;
968 ops = kzalloc(sizeof (*ops) * (num_ops + 1), GFP_NOIO);
975 * op extent offset and length will be set later on
976 * in calc_raw_layout()
978 ops[0].payload_len = payload_len;
983 static void rbd_destroy_ops(struct ceph_osd_req_op *ops)
988 static void rbd_coll_end_req_index(struct request *rq,
989 struct rbd_req_coll *coll,
993 struct request_queue *q;
996 dout("rbd_coll_end_req_index %p index %d ret %d len %llu\n",
997 coll, index, ret, (unsigned long long) len);
1003 blk_end_request(rq, ret, len);
1009 spin_lock_irq(q->queue_lock);
1010 coll->status[index].done = 1;
1011 coll->status[index].rc = ret;
1012 coll->status[index].bytes = len;
1013 max = min = coll->num_done;
1014 while (max < coll->total && coll->status[max].done)
1017 for (i = min; i<max; i++) {
1018 __blk_end_request(rq, coll->status[i].rc,
1019 coll->status[i].bytes);
1021 kref_put(&coll->kref, rbd_coll_release);
1023 spin_unlock_irq(q->queue_lock);
1026 static void rbd_coll_end_req(struct rbd_request *req,
1029 rbd_coll_end_req_index(req->rq, req->coll, req->coll_index, ret, len);
1033 * Send ceph osd request
1035 static int rbd_do_request(struct request *rq,
1036 struct rbd_device *rbd_dev,
1037 struct ceph_snap_context *snapc,
1039 const char *object_name, u64 ofs, u64 len,
1041 struct page **pages,
1044 struct ceph_osd_req_op *ops,
1045 struct rbd_req_coll *coll,
1047 void (*rbd_cb)(struct ceph_osd_request *req,
1048 struct ceph_msg *msg),
1049 struct ceph_osd_request **linger_req,
1052 struct ceph_osd_request *req;
1053 struct ceph_file_layout *layout;
1056 struct timespec mtime = CURRENT_TIME;
1057 struct rbd_request *req_data;
1058 struct ceph_osd_request_head *reqhead;
1059 struct ceph_osd_client *osdc;
1061 req_data = kzalloc(sizeof(*req_data), GFP_NOIO);
1064 rbd_coll_end_req_index(rq, coll, coll_index,
1070 req_data->coll = coll;
1071 req_data->coll_index = coll_index;
1074 dout("rbd_do_request object_name=%s ofs=%llu len=%llu coll=%p[%d]\n",
1075 object_name, (unsigned long long) ofs,
1076 (unsigned long long) len, coll, coll_index);
1078 osdc = &rbd_dev->rbd_client->client->osdc;
1079 req = ceph_osdc_alloc_request(osdc, flags, snapc, ops,
1080 false, GFP_NOIO, pages, bio);
1086 req->r_callback = rbd_cb;
1089 req_data->bio = bio;
1090 req_data->pages = pages;
1091 req_data->len = len;
1093 req->r_priv = req_data;
1095 reqhead = req->r_request->front.iov_base;
1096 reqhead->snapid = cpu_to_le64(CEPH_NOSNAP);
1098 strncpy(req->r_oid, object_name, sizeof(req->r_oid));
1099 req->r_oid_len = strlen(req->r_oid);
1101 layout = &req->r_file_layout;
1102 memset(layout, 0, sizeof(*layout));
1103 layout->fl_stripe_unit = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
1104 layout->fl_stripe_count = cpu_to_le32(1);
1105 layout->fl_object_size = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
1106 layout->fl_pg_pool = cpu_to_le32((int) rbd_dev->spec->pool_id);
1107 ret = ceph_calc_raw_layout(osdc, layout, snapid, ofs, &len, &bno,
1109 rbd_assert(ret == 0);
1111 ceph_osdc_build_request(req, ofs, &len,
1115 req->r_oid, req->r_oid_len);
1118 ceph_osdc_set_request_linger(osdc, req);
1122 ret = ceph_osdc_start_request(osdc, req, false);
1127 ret = ceph_osdc_wait_request(osdc, req);
1129 *ver = le64_to_cpu(req->r_reassert_version.version);
1130 dout("reassert_ver=%llu\n",
1131 (unsigned long long)
1132 le64_to_cpu(req->r_reassert_version.version));
1133 ceph_osdc_put_request(req);
1138 bio_chain_put(req_data->bio);
1139 ceph_osdc_put_request(req);
1141 rbd_coll_end_req(req_data, ret, len);
1147 * Ceph osd op callback
1149 static void rbd_req_cb(struct ceph_osd_request *req, struct ceph_msg *msg)
1151 struct rbd_request *req_data = req->r_priv;
1152 struct ceph_osd_reply_head *replyhead;
1153 struct ceph_osd_op *op;
1159 replyhead = msg->front.iov_base;
1160 WARN_ON(le32_to_cpu(replyhead->num_ops) == 0);
1161 op = (void *)(replyhead + 1);
1162 rc = le32_to_cpu(replyhead->result);
1163 bytes = le64_to_cpu(op->extent.length);
1164 read_op = (le16_to_cpu(op->op) == CEPH_OSD_OP_READ);
1166 dout("rbd_req_cb bytes=%llu readop=%d rc=%d\n",
1167 (unsigned long long) bytes, read_op, (int) rc);
1169 if (rc == -ENOENT && read_op) {
1170 zero_bio_chain(req_data->bio, 0);
1172 } else if (rc == 0 && read_op && bytes < req_data->len) {
1173 zero_bio_chain(req_data->bio, bytes);
1174 bytes = req_data->len;
1177 rbd_coll_end_req(req_data, rc, bytes);
1180 bio_chain_put(req_data->bio);
1182 ceph_osdc_put_request(req);
1186 static void rbd_simple_req_cb(struct ceph_osd_request *req, struct ceph_msg *msg)
1188 ceph_osdc_put_request(req);
1192 * Do a synchronous ceph osd operation
1194 static int rbd_req_sync_op(struct rbd_device *rbd_dev,
1195 struct ceph_snap_context *snapc,
1198 struct ceph_osd_req_op *ops,
1199 const char *object_name,
1200 u64 ofs, u64 inbound_size,
1202 struct ceph_osd_request **linger_req,
1206 struct page **pages;
1209 rbd_assert(ops != NULL);
1211 num_pages = calc_pages_for(ofs, inbound_size);
1212 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1214 return PTR_ERR(pages);
1216 ret = rbd_do_request(NULL, rbd_dev, snapc, snapid,
1217 object_name, ofs, inbound_size, NULL,
1227 if ((flags & CEPH_OSD_FLAG_READ) && inbound)
1228 ret = ceph_copy_from_page_vector(pages, inbound, ofs, ret);
1231 ceph_release_page_vector(pages, num_pages);
1236 * Do an asynchronous ceph osd operation
1238 static int rbd_do_op(struct request *rq,
1239 struct rbd_device *rbd_dev,
1240 struct ceph_snap_context *snapc,
1243 struct rbd_req_coll *coll,
1250 struct ceph_osd_req_op *ops;
1256 seg_name = rbd_segment_name(rbd_dev, ofs);
1259 seg_len = rbd_segment_length(rbd_dev, ofs, len);
1260 seg_ofs = rbd_segment_offset(rbd_dev, ofs);
1262 if (rq_data_dir(rq) == WRITE) {
1263 opcode = CEPH_OSD_OP_WRITE;
1264 flags = CEPH_OSD_FLAG_WRITE|CEPH_OSD_FLAG_ONDISK;
1265 snapid = CEPH_NOSNAP;
1266 payload_len = seg_len;
1268 opcode = CEPH_OSD_OP_READ;
1269 flags = CEPH_OSD_FLAG_READ;
1271 snapid = rbd_dev->spec->snap_id;
1276 ops = rbd_create_rw_ops(1, opcode, payload_len);
1280 /* we've taken care of segment sizes earlier when we
1281 cloned the bios. We should never have a segment
1282 truncated at this point */
1283 rbd_assert(seg_len == len);
1285 ret = rbd_do_request(rq, rbd_dev, snapc, snapid,
1286 seg_name, seg_ofs, seg_len,
1292 rbd_req_cb, 0, NULL);
1294 rbd_destroy_ops(ops);
1301 * Request sync osd read
1303 static int rbd_req_sync_read(struct rbd_device *rbd_dev,
1305 const char *object_name,
1310 struct ceph_osd_req_op *ops;
1313 ops = rbd_create_rw_ops(1, CEPH_OSD_OP_READ, 0);
1317 ret = rbd_req_sync_op(rbd_dev, NULL,
1320 ops, object_name, ofs, len, buf, NULL, ver);
1321 rbd_destroy_ops(ops);
1327 * Request sync osd watch
1329 static int rbd_req_sync_notify_ack(struct rbd_device *rbd_dev,
1333 struct ceph_osd_req_op *ops;
1336 ops = rbd_create_rw_ops(1, CEPH_OSD_OP_NOTIFY_ACK, 0);
1340 ops[0].watch.ver = cpu_to_le64(ver);
1341 ops[0].watch.cookie = notify_id;
1342 ops[0].watch.flag = 0;
1344 ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,
1345 rbd_dev->header_name, 0, 0, NULL,
1350 rbd_simple_req_cb, 0, NULL);
1352 rbd_destroy_ops(ops);
1356 static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
1358 struct rbd_device *rbd_dev = (struct rbd_device *)data;
1365 dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
1366 rbd_dev->header_name, (unsigned long long) notify_id,
1367 (unsigned int) opcode);
1368 rc = rbd_dev_refresh(rbd_dev, &hver);
1370 pr_warning(RBD_DRV_NAME "%d got notification but failed to "
1371 " update snaps: %d\n", rbd_dev->major, rc);
1373 rbd_req_sync_notify_ack(rbd_dev, hver, notify_id);
1377 * Request sync osd watch
1379 static int rbd_req_sync_watch(struct rbd_device *rbd_dev)
1381 struct ceph_osd_req_op *ops;
1382 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1385 ops = rbd_create_rw_ops(1, CEPH_OSD_OP_WATCH, 0);
1389 ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0,
1390 (void *)rbd_dev, &rbd_dev->watch_event);
1394 ops[0].watch.ver = cpu_to_le64(rbd_dev->header.obj_version);
1395 ops[0].watch.cookie = cpu_to_le64(rbd_dev->watch_event->cookie);
1396 ops[0].watch.flag = 1;
1398 ret = rbd_req_sync_op(rbd_dev, NULL,
1400 CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1402 rbd_dev->header_name,
1404 &rbd_dev->watch_request, NULL);
1409 rbd_destroy_ops(ops);
1413 ceph_osdc_cancel_event(rbd_dev->watch_event);
1414 rbd_dev->watch_event = NULL;
1416 rbd_destroy_ops(ops);
1421 * Request sync osd unwatch
1423 static int rbd_req_sync_unwatch(struct rbd_device *rbd_dev)
1425 struct ceph_osd_req_op *ops;
1428 ops = rbd_create_rw_ops(1, CEPH_OSD_OP_WATCH, 0);
1432 ops[0].watch.ver = 0;
1433 ops[0].watch.cookie = cpu_to_le64(rbd_dev->watch_event->cookie);
1434 ops[0].watch.flag = 0;
1436 ret = rbd_req_sync_op(rbd_dev, NULL,
1438 CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1440 rbd_dev->header_name,
1441 0, 0, NULL, NULL, NULL);
1444 rbd_destroy_ops(ops);
1445 ceph_osdc_cancel_event(rbd_dev->watch_event);
1446 rbd_dev->watch_event = NULL;
1451 * Synchronous osd object method call
1453 static int rbd_req_sync_exec(struct rbd_device *rbd_dev,
1454 const char *object_name,
1455 const char *class_name,
1456 const char *method_name,
1457 const char *outbound,
1458 size_t outbound_size,
1460 size_t inbound_size,
1464 struct ceph_osd_req_op *ops;
1465 int class_name_len = strlen(class_name);
1466 int method_name_len = strlen(method_name);
1471 * Any input parameters required by the method we're calling
1472 * will be sent along with the class and method names as
1473 * part of the message payload. That data and its size are
1474 * supplied via the indata and indata_len fields (named from
1475 * the perspective of the server side) in the OSD request
1478 payload_size = class_name_len + method_name_len + outbound_size;
1479 ops = rbd_create_rw_ops(1, CEPH_OSD_OP_CALL, payload_size);
1483 ops[0].cls.class_name = class_name;
1484 ops[0].cls.class_len = (__u8) class_name_len;
1485 ops[0].cls.method_name = method_name;
1486 ops[0].cls.method_len = (__u8) method_name_len;
1487 ops[0].cls.argc = 0;
1488 ops[0].cls.indata = outbound;
1489 ops[0].cls.indata_len = outbound_size;
1491 ret = rbd_req_sync_op(rbd_dev, NULL,
1494 object_name, 0, inbound_size, inbound,
1497 rbd_destroy_ops(ops);
1499 dout("cls_exec returned %d\n", ret);
1503 static struct rbd_req_coll *rbd_alloc_coll(int num_reqs)
1505 struct rbd_req_coll *coll =
1506 kzalloc(sizeof(struct rbd_req_coll) +
1507 sizeof(struct rbd_req_status) * num_reqs,
1512 coll->total = num_reqs;
1513 kref_init(&coll->kref);
1518 * block device queue callback
1520 static void rbd_rq_fn(struct request_queue *q)
1522 struct rbd_device *rbd_dev = q->queuedata;
1525 while ((rq = blk_fetch_request(q))) {
1530 int num_segs, cur_seg = 0;
1531 struct rbd_req_coll *coll;
1532 struct ceph_snap_context *snapc;
1533 unsigned int bio_offset;
1535 dout("fetched request\n");
1537 /* filter out block requests we don't understand */
1538 if ((rq->cmd_type != REQ_TYPE_FS)) {
1539 __blk_end_request_all(rq, 0);
1543 /* deduce our operation (read, write) */
1544 do_write = (rq_data_dir(rq) == WRITE);
1545 if (do_write && rbd_dev->mapping.read_only) {
1546 __blk_end_request_all(rq, -EROFS);
1550 spin_unlock_irq(q->queue_lock);
1552 down_read(&rbd_dev->header_rwsem);
1554 if (!rbd_dev->exists) {
1555 rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
1556 up_read(&rbd_dev->header_rwsem);
1557 dout("request for non-existent snapshot");
1558 spin_lock_irq(q->queue_lock);
1559 __blk_end_request_all(rq, -ENXIO);
1563 snapc = ceph_get_snap_context(rbd_dev->header.snapc);
1565 up_read(&rbd_dev->header_rwsem);
1567 size = blk_rq_bytes(rq);
1568 ofs = blk_rq_pos(rq) * SECTOR_SIZE;
1571 dout("%s 0x%x bytes at 0x%llx\n",
1572 do_write ? "write" : "read",
1573 size, (unsigned long long) blk_rq_pos(rq) * SECTOR_SIZE);
1575 num_segs = rbd_get_num_segments(&rbd_dev->header, ofs, size);
1576 if (num_segs <= 0) {
1577 spin_lock_irq(q->queue_lock);
1578 __blk_end_request_all(rq, num_segs);
1579 ceph_put_snap_context(snapc);
1582 coll = rbd_alloc_coll(num_segs);
1584 spin_lock_irq(q->queue_lock);
1585 __blk_end_request_all(rq, -ENOMEM);
1586 ceph_put_snap_context(snapc);
1592 u64 limit = rbd_segment_length(rbd_dev, ofs, size);
1593 unsigned int chain_size;
1594 struct bio *bio_chain;
1596 BUG_ON(limit > (u64) UINT_MAX);
1597 chain_size = (unsigned int) limit;
1598 dout("rq->bio->bi_vcnt=%hu\n", rq->bio->bi_vcnt);
1600 kref_get(&coll->kref);
1602 /* Pass a cloned bio chain via an osd request */
1604 bio_chain = bio_chain_clone_range(&bio,
1605 &bio_offset, chain_size,
1608 (void) rbd_do_op(rq, rbd_dev, snapc,
1610 bio_chain, coll, cur_seg);
1612 rbd_coll_end_req_index(rq, coll, cur_seg,
1613 -ENOMEM, chain_size);
1619 kref_put(&coll->kref, rbd_coll_release);
1621 spin_lock_irq(q->queue_lock);
1623 ceph_put_snap_context(snapc);
1628 * a queue callback. Makes sure that we don't create a bio that spans across
1629 * multiple osd objects. One exception would be with a single page bios,
1630 * which we handle later at bio_chain_clone_range()
1632 static int rbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bmd,
1633 struct bio_vec *bvec)
1635 struct rbd_device *rbd_dev = q->queuedata;
1636 sector_t sector_offset;
1637 sector_t sectors_per_obj;
1638 sector_t obj_sector_offset;
1642 * Find how far into its rbd object the partition-relative
1643 * bio start sector is to offset relative to the enclosing
1646 sector_offset = get_start_sect(bmd->bi_bdev) + bmd->bi_sector;
1647 sectors_per_obj = 1 << (rbd_dev->header.obj_order - SECTOR_SHIFT);
1648 obj_sector_offset = sector_offset & (sectors_per_obj - 1);
1651 * Compute the number of bytes from that offset to the end
1652 * of the object. Account for what's already used by the bio.
1654 ret = (int) (sectors_per_obj - obj_sector_offset) << SECTOR_SHIFT;
1655 if (ret > bmd->bi_size)
1656 ret -= bmd->bi_size;
1661 * Don't send back more than was asked for. And if the bio
1662 * was empty, let the whole thing through because: "Note
1663 * that a block device *must* allow a single page to be
1664 * added to an empty bio."
1666 rbd_assert(bvec->bv_len <= PAGE_SIZE);
1667 if (ret > (int) bvec->bv_len || !bmd->bi_size)
1668 ret = (int) bvec->bv_len;
1673 static void rbd_free_disk(struct rbd_device *rbd_dev)
1675 struct gendisk *disk = rbd_dev->disk;
1680 if (disk->flags & GENHD_FL_UP)
1683 blk_cleanup_queue(disk->queue);
1688 * Read the complete header for the given rbd device.
1690 * Returns a pointer to a dynamically-allocated buffer containing
1691 * the complete and validated header. Caller can pass the address
1692 * of a variable that will be filled in with the version of the
1693 * header object at the time it was read.
1695 * Returns a pointer-coded errno if a failure occurs.
1697 static struct rbd_image_header_ondisk *
1698 rbd_dev_v1_header_read(struct rbd_device *rbd_dev, u64 *version)
1700 struct rbd_image_header_ondisk *ondisk = NULL;
1707 * The complete header will include an array of its 64-bit
1708 * snapshot ids, followed by the names of those snapshots as
1709 * a contiguous block of NUL-terminated strings. Note that
1710 * the number of snapshots could change by the time we read
1711 * it in, in which case we re-read it.
1718 size = sizeof (*ondisk);
1719 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
1721 ondisk = kmalloc(size, GFP_KERNEL);
1723 return ERR_PTR(-ENOMEM);
1725 ret = rbd_req_sync_read(rbd_dev, CEPH_NOSNAP,
1726 rbd_dev->header_name,
1728 (char *) ondisk, version);
1732 if (WARN_ON((size_t) ret < size)) {
1734 pr_warning("short header read for image %s"
1735 " (want %zd got %d)\n",
1736 rbd_dev->spec->image_name, size, ret);
1739 if (!rbd_dev_ondisk_valid(ondisk)) {
1741 pr_warning("invalid header for image %s\n",
1742 rbd_dev->spec->image_name);
1746 names_size = le64_to_cpu(ondisk->snap_names_len);
1747 want_count = snap_count;
1748 snap_count = le32_to_cpu(ondisk->snap_count);
1749 } while (snap_count != want_count);
1756 return ERR_PTR(ret);
1760 * reload the ondisk the header
1762 static int rbd_read_header(struct rbd_device *rbd_dev,
1763 struct rbd_image_header *header)
1765 struct rbd_image_header_ondisk *ondisk;
1769 ondisk = rbd_dev_v1_header_read(rbd_dev, &ver);
1771 return PTR_ERR(ondisk);
1772 ret = rbd_header_from_disk(header, ondisk);
1774 header->obj_version = ver;
1780 static void rbd_remove_all_snaps(struct rbd_device *rbd_dev)
1782 struct rbd_snap *snap;
1783 struct rbd_snap *next;
1785 list_for_each_entry_safe(snap, next, &rbd_dev->snaps, node)
1786 rbd_remove_snap_dev(snap);
1789 static void rbd_update_mapping_size(struct rbd_device *rbd_dev)
1793 if (rbd_dev->spec->snap_id != CEPH_NOSNAP)
1796 size = (sector_t) rbd_dev->header.image_size / SECTOR_SIZE;
1797 dout("setting size to %llu sectors", (unsigned long long) size);
1798 rbd_dev->mapping.size = (u64) size;
1799 set_capacity(rbd_dev->disk, size);
1803 * only read the first part of the ondisk header, without the snaps info
1805 static int rbd_dev_v1_refresh(struct rbd_device *rbd_dev, u64 *hver)
1808 struct rbd_image_header h;
1810 ret = rbd_read_header(rbd_dev, &h);
1814 down_write(&rbd_dev->header_rwsem);
1816 /* Update image size, and check for resize of mapped image */
1817 rbd_dev->header.image_size = h.image_size;
1818 rbd_update_mapping_size(rbd_dev);
1820 /* rbd_dev->header.object_prefix shouldn't change */
1821 kfree(rbd_dev->header.snap_sizes);
1822 kfree(rbd_dev->header.snap_names);
1823 /* osd requests may still refer to snapc */
1824 ceph_put_snap_context(rbd_dev->header.snapc);
1827 *hver = h.obj_version;
1828 rbd_dev->header.obj_version = h.obj_version;
1829 rbd_dev->header.image_size = h.image_size;
1830 rbd_dev->header.snapc = h.snapc;
1831 rbd_dev->header.snap_names = h.snap_names;
1832 rbd_dev->header.snap_sizes = h.snap_sizes;
1833 /* Free the extra copy of the object prefix */
1834 WARN_ON(strcmp(rbd_dev->header.object_prefix, h.object_prefix));
1835 kfree(h.object_prefix);
1837 ret = rbd_dev_snaps_update(rbd_dev);
1839 ret = rbd_dev_snaps_register(rbd_dev);
1841 up_write(&rbd_dev->header_rwsem);
1846 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver)
1850 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1851 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1852 if (rbd_dev->image_format == 1)
1853 ret = rbd_dev_v1_refresh(rbd_dev, hver);
1855 ret = rbd_dev_v2_refresh(rbd_dev, hver);
1856 mutex_unlock(&ctl_mutex);
1861 static int rbd_init_disk(struct rbd_device *rbd_dev)
1863 struct gendisk *disk;
1864 struct request_queue *q;
1867 /* create gendisk info */
1868 disk = alloc_disk(RBD_MINORS_PER_MAJOR);
1872 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
1874 disk->major = rbd_dev->major;
1875 disk->first_minor = 0;
1876 disk->fops = &rbd_bd_ops;
1877 disk->private_data = rbd_dev;
1880 q = blk_init_queue(rbd_rq_fn, &rbd_dev->lock);
1884 /* We use the default size, but let's be explicit about it. */
1885 blk_queue_physical_block_size(q, SECTOR_SIZE);
1887 /* set io sizes to object size */
1888 segment_size = rbd_obj_bytes(&rbd_dev->header);
1889 blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE);
1890 blk_queue_max_segment_size(q, segment_size);
1891 blk_queue_io_min(q, segment_size);
1892 blk_queue_io_opt(q, segment_size);
1894 blk_queue_merge_bvec(q, rbd_merge_bvec);
1897 q->queuedata = rbd_dev;
1899 rbd_dev->disk = disk;
1901 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
1914 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
1916 return container_of(dev, struct rbd_device, dev);
1919 static ssize_t rbd_size_show(struct device *dev,
1920 struct device_attribute *attr, char *buf)
1922 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1925 down_read(&rbd_dev->header_rwsem);
1926 size = get_capacity(rbd_dev->disk);
1927 up_read(&rbd_dev->header_rwsem);
1929 return sprintf(buf, "%llu\n", (unsigned long long) size * SECTOR_SIZE);
1933 * Note this shows the features for whatever's mapped, which is not
1934 * necessarily the base image.
1936 static ssize_t rbd_features_show(struct device *dev,
1937 struct device_attribute *attr, char *buf)
1939 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1941 return sprintf(buf, "0x%016llx\n",
1942 (unsigned long long) rbd_dev->mapping.features);
1945 static ssize_t rbd_major_show(struct device *dev,
1946 struct device_attribute *attr, char *buf)
1948 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1950 return sprintf(buf, "%d\n", rbd_dev->major);
1953 static ssize_t rbd_client_id_show(struct device *dev,
1954 struct device_attribute *attr, char *buf)
1956 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1958 return sprintf(buf, "client%lld\n",
1959 ceph_client_id(rbd_dev->rbd_client->client));
1962 static ssize_t rbd_pool_show(struct device *dev,
1963 struct device_attribute *attr, char *buf)
1965 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1967 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
1970 static ssize_t rbd_pool_id_show(struct device *dev,
1971 struct device_attribute *attr, char *buf)
1973 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1975 return sprintf(buf, "%llu\n",
1976 (unsigned long long) rbd_dev->spec->pool_id);
1979 static ssize_t rbd_name_show(struct device *dev,
1980 struct device_attribute *attr, char *buf)
1982 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1984 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
1987 static ssize_t rbd_image_id_show(struct device *dev,
1988 struct device_attribute *attr, char *buf)
1990 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1992 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
1996 * Shows the name of the currently-mapped snapshot (or
1997 * RBD_SNAP_HEAD_NAME for the base image).
1999 static ssize_t rbd_snap_show(struct device *dev,
2000 struct device_attribute *attr,
2003 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2005 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
2008 static ssize_t rbd_image_refresh(struct device *dev,
2009 struct device_attribute *attr,
2013 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2016 ret = rbd_dev_refresh(rbd_dev, NULL);
2018 return ret < 0 ? ret : size;
2021 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
2022 static DEVICE_ATTR(features, S_IRUGO, rbd_features_show, NULL);
2023 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
2024 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
2025 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
2026 static DEVICE_ATTR(pool_id, S_IRUGO, rbd_pool_id_show, NULL);
2027 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
2028 static DEVICE_ATTR(image_id, S_IRUGO, rbd_image_id_show, NULL);
2029 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
2030 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
2032 static struct attribute *rbd_attrs[] = {
2033 &dev_attr_size.attr,
2034 &dev_attr_features.attr,
2035 &dev_attr_major.attr,
2036 &dev_attr_client_id.attr,
2037 &dev_attr_pool.attr,
2038 &dev_attr_pool_id.attr,
2039 &dev_attr_name.attr,
2040 &dev_attr_image_id.attr,
2041 &dev_attr_current_snap.attr,
2042 &dev_attr_refresh.attr,
2046 static struct attribute_group rbd_attr_group = {
2050 static const struct attribute_group *rbd_attr_groups[] = {
2055 static void rbd_sysfs_dev_release(struct device *dev)
2059 static struct device_type rbd_device_type = {
2061 .groups = rbd_attr_groups,
2062 .release = rbd_sysfs_dev_release,
2070 static ssize_t rbd_snap_size_show(struct device *dev,
2071 struct device_attribute *attr,
2074 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2076 return sprintf(buf, "%llu\n", (unsigned long long)snap->size);
2079 static ssize_t rbd_snap_id_show(struct device *dev,
2080 struct device_attribute *attr,
2083 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2085 return sprintf(buf, "%llu\n", (unsigned long long)snap->id);
2088 static ssize_t rbd_snap_features_show(struct device *dev,
2089 struct device_attribute *attr,
2092 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2094 return sprintf(buf, "0x%016llx\n",
2095 (unsigned long long) snap->features);
2098 static DEVICE_ATTR(snap_size, S_IRUGO, rbd_snap_size_show, NULL);
2099 static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
2100 static DEVICE_ATTR(snap_features, S_IRUGO, rbd_snap_features_show, NULL);
2102 static struct attribute *rbd_snap_attrs[] = {
2103 &dev_attr_snap_size.attr,
2104 &dev_attr_snap_id.attr,
2105 &dev_attr_snap_features.attr,
2109 static struct attribute_group rbd_snap_attr_group = {
2110 .attrs = rbd_snap_attrs,
2113 static void rbd_snap_dev_release(struct device *dev)
2115 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2120 static const struct attribute_group *rbd_snap_attr_groups[] = {
2121 &rbd_snap_attr_group,
2125 static struct device_type rbd_snap_device_type = {
2126 .groups = rbd_snap_attr_groups,
2127 .release = rbd_snap_dev_release,
2130 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
2132 kref_get(&spec->kref);
2137 static void rbd_spec_free(struct kref *kref);
2138 static void rbd_spec_put(struct rbd_spec *spec)
2141 kref_put(&spec->kref, rbd_spec_free);
2144 static struct rbd_spec *rbd_spec_alloc(void)
2146 struct rbd_spec *spec;
2148 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
2151 kref_init(&spec->kref);
2153 rbd_spec_put(rbd_spec_get(spec)); /* TEMPORARY */
2158 static void rbd_spec_free(struct kref *kref)
2160 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
2162 kfree(spec->pool_name);
2163 kfree(spec->image_id);
2164 kfree(spec->image_name);
2165 kfree(spec->snap_name);
2169 static bool rbd_snap_registered(struct rbd_snap *snap)
2171 bool ret = snap->dev.type == &rbd_snap_device_type;
2172 bool reg = device_is_registered(&snap->dev);
2174 rbd_assert(!ret ^ reg);
2179 static void rbd_remove_snap_dev(struct rbd_snap *snap)
2181 list_del(&snap->node);
2182 if (device_is_registered(&snap->dev))
2183 device_unregister(&snap->dev);
2186 static int rbd_register_snap_dev(struct rbd_snap *snap,
2187 struct device *parent)
2189 struct device *dev = &snap->dev;
2192 dev->type = &rbd_snap_device_type;
2193 dev->parent = parent;
2194 dev->release = rbd_snap_dev_release;
2195 dev_set_name(dev, "%s%s", RBD_SNAP_DEV_NAME_PREFIX, snap->name);
2196 dout("%s: registering device for snapshot %s\n", __func__, snap->name);
2198 ret = device_register(dev);
2203 static struct rbd_snap *__rbd_add_snap_dev(struct rbd_device *rbd_dev,
2204 const char *snap_name,
2205 u64 snap_id, u64 snap_size,
2208 struct rbd_snap *snap;
2211 snap = kzalloc(sizeof (*snap), GFP_KERNEL);
2213 return ERR_PTR(-ENOMEM);
2216 snap->name = kstrdup(snap_name, GFP_KERNEL);
2221 snap->size = snap_size;
2222 snap->features = snap_features;
2230 return ERR_PTR(ret);
2233 static char *rbd_dev_v1_snap_info(struct rbd_device *rbd_dev, u32 which,
2234 u64 *snap_size, u64 *snap_features)
2238 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
2240 *snap_size = rbd_dev->header.snap_sizes[which];
2241 *snap_features = 0; /* No features for v1 */
2243 /* Skip over names until we find the one we are looking for */
2245 snap_name = rbd_dev->header.snap_names;
2247 snap_name += strlen(snap_name) + 1;
2253 * Get the size and object order for an image snapshot, or if
2254 * snap_id is CEPH_NOSNAP, gets this information for the base
2257 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
2258 u8 *order, u64 *snap_size)
2260 __le64 snapid = cpu_to_le64(snap_id);
2265 } __attribute__ ((packed)) size_buf = { 0 };
2267 ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
2269 (char *) &snapid, sizeof (snapid),
2270 (char *) &size_buf, sizeof (size_buf),
2271 CEPH_OSD_FLAG_READ, NULL);
2272 dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2276 *order = size_buf.order;
2277 *snap_size = le64_to_cpu(size_buf.size);
2279 dout(" snap_id 0x%016llx order = %u, snap_size = %llu\n",
2280 (unsigned long long) snap_id, (unsigned int) *order,
2281 (unsigned long long) *snap_size);
2286 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
2288 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
2289 &rbd_dev->header.obj_order,
2290 &rbd_dev->header.image_size);
2293 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
2299 reply_buf = kzalloc(RBD_OBJ_PREFIX_LEN_MAX, GFP_KERNEL);
2303 ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
2304 "rbd", "get_object_prefix",
2306 reply_buf, RBD_OBJ_PREFIX_LEN_MAX,
2307 CEPH_OSD_FLAG_READ, NULL);
2308 dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2311 ret = 0; /* rbd_req_sync_exec() can return positive */
2314 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
2315 p + RBD_OBJ_PREFIX_LEN_MAX,
2318 if (IS_ERR(rbd_dev->header.object_prefix)) {
2319 ret = PTR_ERR(rbd_dev->header.object_prefix);
2320 rbd_dev->header.object_prefix = NULL;
2322 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix);
2331 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
2334 __le64 snapid = cpu_to_le64(snap_id);
2338 } features_buf = { 0 };
2342 ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
2343 "rbd", "get_features",
2344 (char *) &snapid, sizeof (snapid),
2345 (char *) &features_buf, sizeof (features_buf),
2346 CEPH_OSD_FLAG_READ, NULL);
2347 dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2351 incompat = le64_to_cpu(features_buf.incompat);
2352 if (incompat & ~RBD_FEATURES_ALL)
2355 *snap_features = le64_to_cpu(features_buf.features);
2357 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
2358 (unsigned long long) snap_id,
2359 (unsigned long long) *snap_features,
2360 (unsigned long long) le64_to_cpu(features_buf.incompat));
2365 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
2367 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
2368 &rbd_dev->header.features);
2371 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev, u64 *ver)
2380 struct ceph_snap_context *snapc;
2384 * We'll need room for the seq value (maximum snapshot id),
2385 * snapshot count, and array of that many snapshot ids.
2386 * For now we have a fixed upper limit on the number we're
2387 * prepared to receive.
2389 size = sizeof (__le64) + sizeof (__le32) +
2390 RBD_MAX_SNAP_COUNT * sizeof (__le64);
2391 reply_buf = kzalloc(size, GFP_KERNEL);
2395 ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
2396 "rbd", "get_snapcontext",
2399 CEPH_OSD_FLAG_READ, ver);
2400 dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2406 end = (char *) reply_buf + size;
2407 ceph_decode_64_safe(&p, end, seq, out);
2408 ceph_decode_32_safe(&p, end, snap_count, out);
2411 * Make sure the reported number of snapshot ids wouldn't go
2412 * beyond the end of our buffer. But before checking that,
2413 * make sure the computed size of the snapshot context we
2414 * allocate is representable in a size_t.
2416 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
2421 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
2424 size = sizeof (struct ceph_snap_context) +
2425 snap_count * sizeof (snapc->snaps[0]);
2426 snapc = kmalloc(size, GFP_KERNEL);
2432 atomic_set(&snapc->nref, 1);
2434 snapc->num_snaps = snap_count;
2435 for (i = 0; i < snap_count; i++)
2436 snapc->snaps[i] = ceph_decode_64(&p);
2438 rbd_dev->header.snapc = snapc;
2440 dout(" snap context seq = %llu, snap_count = %u\n",
2441 (unsigned long long) seq, (unsigned int) snap_count);
2449 static char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev, u32 which)
2459 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
2460 reply_buf = kmalloc(size, GFP_KERNEL);
2462 return ERR_PTR(-ENOMEM);
2464 snap_id = cpu_to_le64(rbd_dev->header.snapc->snaps[which]);
2465 ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
2466 "rbd", "get_snapshot_name",
2467 (char *) &snap_id, sizeof (snap_id),
2469 CEPH_OSD_FLAG_READ, NULL);
2470 dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2475 end = (char *) reply_buf + size;
2476 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
2477 if (IS_ERR(snap_name)) {
2478 ret = PTR_ERR(snap_name);
2481 dout(" snap_id 0x%016llx snap_name = %s\n",
2482 (unsigned long long) le64_to_cpu(snap_id), snap_name);
2490 return ERR_PTR(ret);
2493 static char *rbd_dev_v2_snap_info(struct rbd_device *rbd_dev, u32 which,
2494 u64 *snap_size, u64 *snap_features)
2500 snap_id = rbd_dev->header.snapc->snaps[which];
2501 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, &order, snap_size);
2503 return ERR_PTR(ret);
2504 ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, snap_features);
2506 return ERR_PTR(ret);
2508 return rbd_dev_v2_snap_name(rbd_dev, which);
2511 static char *rbd_dev_snap_info(struct rbd_device *rbd_dev, u32 which,
2512 u64 *snap_size, u64 *snap_features)
2514 if (rbd_dev->image_format == 1)
2515 return rbd_dev_v1_snap_info(rbd_dev, which,
2516 snap_size, snap_features);
2517 if (rbd_dev->image_format == 2)
2518 return rbd_dev_v2_snap_info(rbd_dev, which,
2519 snap_size, snap_features);
2520 return ERR_PTR(-EINVAL);
2523 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver)
2528 down_write(&rbd_dev->header_rwsem);
2530 /* Grab old order first, to see if it changes */
2532 obj_order = rbd_dev->header.obj_order,
2533 ret = rbd_dev_v2_image_size(rbd_dev);
2536 if (rbd_dev->header.obj_order != obj_order) {
2540 rbd_update_mapping_size(rbd_dev);
2542 ret = rbd_dev_v2_snap_context(rbd_dev, hver);
2543 dout("rbd_dev_v2_snap_context returned %d\n", ret);
2546 ret = rbd_dev_snaps_update(rbd_dev);
2547 dout("rbd_dev_snaps_update returned %d\n", ret);
2550 ret = rbd_dev_snaps_register(rbd_dev);
2551 dout("rbd_dev_snaps_register returned %d\n", ret);
2553 up_write(&rbd_dev->header_rwsem);
2559 * Scan the rbd device's current snapshot list and compare it to the
2560 * newly-received snapshot context. Remove any existing snapshots
2561 * not present in the new snapshot context. Add a new snapshot for
2562 * any snaphots in the snapshot context not in the current list.
2563 * And verify there are no changes to snapshots we already know
2566 * Assumes the snapshots in the snapshot context are sorted by
2567 * snapshot id, highest id first. (Snapshots in the rbd_dev's list
2568 * are also maintained in that order.)
2570 static int rbd_dev_snaps_update(struct rbd_device *rbd_dev)
2572 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
2573 const u32 snap_count = snapc->num_snaps;
2574 struct list_head *head = &rbd_dev->snaps;
2575 struct list_head *links = head->next;
2578 dout("%s: snap count is %u\n", __func__, (unsigned int) snap_count);
2579 while (index < snap_count || links != head) {
2581 struct rbd_snap *snap;
2584 u64 snap_features = 0;
2586 snap_id = index < snap_count ? snapc->snaps[index]
2588 snap = links != head ? list_entry(links, struct rbd_snap, node)
2590 rbd_assert(!snap || snap->id != CEPH_NOSNAP);
2592 if (snap_id == CEPH_NOSNAP || (snap && snap->id > snap_id)) {
2593 struct list_head *next = links->next;
2595 /* Existing snapshot not in the new snap context */
2597 if (rbd_dev->spec->snap_id == snap->id)
2598 rbd_dev->exists = false;
2599 rbd_remove_snap_dev(snap);
2600 dout("%ssnap id %llu has been removed\n",
2601 rbd_dev->spec->snap_id == snap->id ?
2603 (unsigned long long) snap->id);
2605 /* Done with this list entry; advance */
2611 snap_name = rbd_dev_snap_info(rbd_dev, index,
2612 &snap_size, &snap_features);
2613 if (IS_ERR(snap_name))
2614 return PTR_ERR(snap_name);
2616 dout("entry %u: snap_id = %llu\n", (unsigned int) snap_count,
2617 (unsigned long long) snap_id);
2618 if (!snap || (snap_id != CEPH_NOSNAP && snap->id < snap_id)) {
2619 struct rbd_snap *new_snap;
2621 /* We haven't seen this snapshot before */
2623 new_snap = __rbd_add_snap_dev(rbd_dev, snap_name,
2624 snap_id, snap_size, snap_features);
2625 if (IS_ERR(new_snap)) {
2626 int err = PTR_ERR(new_snap);
2628 dout(" failed to add dev, error %d\n", err);
2633 /* New goes before existing, or at end of list */
2635 dout(" added dev%s\n", snap ? "" : " at end\n");
2637 list_add_tail(&new_snap->node, &snap->node);
2639 list_add_tail(&new_snap->node, head);
2641 /* Already have this one */
2643 dout(" already present\n");
2645 rbd_assert(snap->size == snap_size);
2646 rbd_assert(!strcmp(snap->name, snap_name));
2647 rbd_assert(snap->features == snap_features);
2649 /* Done with this list entry; advance */
2651 links = links->next;
2654 /* Advance to the next entry in the snapshot context */
2658 dout("%s: done\n", __func__);
2664 * Scan the list of snapshots and register the devices for any that
2665 * have not already been registered.
2667 static int rbd_dev_snaps_register(struct rbd_device *rbd_dev)
2669 struct rbd_snap *snap;
2672 dout("%s called\n", __func__);
2673 if (WARN_ON(!device_is_registered(&rbd_dev->dev)))
2676 list_for_each_entry(snap, &rbd_dev->snaps, node) {
2677 if (!rbd_snap_registered(snap)) {
2678 ret = rbd_register_snap_dev(snap, &rbd_dev->dev);
2683 dout("%s: returning %d\n", __func__, ret);
2688 static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
2693 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2695 dev = &rbd_dev->dev;
2696 dev->bus = &rbd_bus_type;
2697 dev->type = &rbd_device_type;
2698 dev->parent = &rbd_root_dev;
2699 dev->release = rbd_dev_release;
2700 dev_set_name(dev, "%d", rbd_dev->dev_id);
2701 ret = device_register(dev);
2703 mutex_unlock(&ctl_mutex);
2708 static void rbd_bus_del_dev(struct rbd_device *rbd_dev)
2710 device_unregister(&rbd_dev->dev);
2713 static int rbd_init_watch_dev(struct rbd_device *rbd_dev)
2718 ret = rbd_req_sync_watch(rbd_dev);
2719 if (ret == -ERANGE) {
2720 rc = rbd_dev_refresh(rbd_dev, NULL);
2724 } while (ret == -ERANGE);
2729 static atomic64_t rbd_dev_id_max = ATOMIC64_INIT(0);
2732 * Get a unique rbd identifier for the given new rbd_dev, and add
2733 * the rbd_dev to the global list. The minimum rbd id is 1.
2735 static void rbd_dev_id_get(struct rbd_device *rbd_dev)
2737 rbd_dev->dev_id = atomic64_inc_return(&rbd_dev_id_max);
2739 spin_lock(&rbd_dev_list_lock);
2740 list_add_tail(&rbd_dev->node, &rbd_dev_list);
2741 spin_unlock(&rbd_dev_list_lock);
2742 dout("rbd_dev %p given dev id %llu\n", rbd_dev,
2743 (unsigned long long) rbd_dev->dev_id);
2747 * Remove an rbd_dev from the global list, and record that its
2748 * identifier is no longer in use.
2750 static void rbd_dev_id_put(struct rbd_device *rbd_dev)
2752 struct list_head *tmp;
2753 int rbd_id = rbd_dev->dev_id;
2756 rbd_assert(rbd_id > 0);
2758 dout("rbd_dev %p released dev id %llu\n", rbd_dev,
2759 (unsigned long long) rbd_dev->dev_id);
2760 spin_lock(&rbd_dev_list_lock);
2761 list_del_init(&rbd_dev->node);
2764 * If the id being "put" is not the current maximum, there
2765 * is nothing special we need to do.
2767 if (rbd_id != atomic64_read(&rbd_dev_id_max)) {
2768 spin_unlock(&rbd_dev_list_lock);
2773 * We need to update the current maximum id. Search the
2774 * list to find out what it is. We're more likely to find
2775 * the maximum at the end, so search the list backward.
2778 list_for_each_prev(tmp, &rbd_dev_list) {
2779 struct rbd_device *rbd_dev;
2781 rbd_dev = list_entry(tmp, struct rbd_device, node);
2782 if (rbd_dev->dev_id > max_id)
2783 max_id = rbd_dev->dev_id;
2785 spin_unlock(&rbd_dev_list_lock);
2788 * The max id could have been updated by rbd_dev_id_get(), in
2789 * which case it now accurately reflects the new maximum.
2790 * Be careful not to overwrite the maximum value in that
2793 atomic64_cmpxchg(&rbd_dev_id_max, rbd_id, max_id);
2794 dout(" max dev id has been reset\n");
2798 * Skips over white space at *buf, and updates *buf to point to the
2799 * first found non-space character (if any). Returns the length of
2800 * the token (string of non-white space characters) found. Note
2801 * that *buf must be terminated with '\0'.
2803 static inline size_t next_token(const char **buf)
2806 * These are the characters that produce nonzero for
2807 * isspace() in the "C" and "POSIX" locales.
2809 const char *spaces = " \f\n\r\t\v";
2811 *buf += strspn(*buf, spaces); /* Find start of token */
2813 return strcspn(*buf, spaces); /* Return token length */
2817 * Finds the next token in *buf, and if the provided token buffer is
2818 * big enough, copies the found token into it. The result, if
2819 * copied, is guaranteed to be terminated with '\0'. Note that *buf
2820 * must be terminated with '\0' on entry.
2822 * Returns the length of the token found (not including the '\0').
2823 * Return value will be 0 if no token is found, and it will be >=
2824 * token_size if the token would not fit.
2826 * The *buf pointer will be updated to point beyond the end of the
2827 * found token. Note that this occurs even if the token buffer is
2828 * too small to hold it.
2830 static inline size_t copy_token(const char **buf,
2836 len = next_token(buf);
2837 if (len < token_size) {
2838 memcpy(token, *buf, len);
2839 *(token + len) = '\0';
2847 * Finds the next token in *buf, dynamically allocates a buffer big
2848 * enough to hold a copy of it, and copies the token into the new
2849 * buffer. The copy is guaranteed to be terminated with '\0'. Note
2850 * that a duplicate buffer is created even for a zero-length token.
2852 * Returns a pointer to the newly-allocated duplicate, or a null
2853 * pointer if memory for the duplicate was not available. If
2854 * the lenp argument is a non-null pointer, the length of the token
2855 * (not including the '\0') is returned in *lenp.
2857 * If successful, the *buf pointer will be updated to point beyond
2858 * the end of the found token.
2860 * Note: uses GFP_KERNEL for allocation.
2862 static inline char *dup_token(const char **buf, size_t *lenp)
2867 len = next_token(buf);
2868 dup = kmalloc(len + 1, GFP_KERNEL);
2872 memcpy(dup, *buf, len);
2873 *(dup + len) = '\0';
2883 * Parse the options provided for an "rbd add" (i.e., rbd image
2884 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
2885 * and the data written is passed here via a NUL-terminated buffer.
2886 * Returns 0 if successful or an error code otherwise.
2888 * The information extracted from these options is recorded in
2889 * the other parameters which return dynamically-allocated
2892 * The address of a pointer that will refer to a ceph options
2893 * structure. Caller must release the returned pointer using
2894 * ceph_destroy_options() when it is no longer needed.
2896 * Address of an rbd options pointer. Fully initialized by
2897 * this function; caller must release with kfree().
2899 * Address of an rbd image specification pointer. Fully
2900 * initialized by this function based on parsed options.
2901 * Caller must release with rbd_spec_put().
2903 * The options passed take this form:
2904 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
2907 * A comma-separated list of one or more monitor addresses.
2908 * A monitor address is an ip address, optionally followed
2909 * by a port number (separated by a colon).
2910 * I.e.: ip1[:port1][,ip2[:port2]...]
2912 * A comma-separated list of ceph and/or rbd options.
2914 * The name of the rados pool containing the rbd image.
2916 * The name of the image in that pool to map.
2918 * An optional snapshot id. If provided, the mapping will
2919 * present data from the image at the time that snapshot was
2920 * created. The image head is used if no snapshot id is
2921 * provided. Snapshot mappings are always read-only.
2923 static int rbd_add_parse_args(const char *buf,
2924 struct ceph_options **ceph_opts,
2925 struct rbd_options **opts,
2926 struct rbd_spec **rbd_spec)
2930 const char *mon_addrs;
2931 size_t mon_addrs_size;
2932 struct rbd_spec *spec = NULL;
2933 struct rbd_options *rbd_opts = NULL;
2934 struct ceph_options *copts;
2937 /* The first four tokens are required */
2939 len = next_token(&buf);
2941 return -EINVAL; /* Missing monitor address(es) */
2943 mon_addrs_size = len + 1;
2947 options = dup_token(&buf, NULL);
2951 goto out_err; /* Missing options */
2953 spec = rbd_spec_alloc();
2957 spec->pool_name = dup_token(&buf, NULL);
2958 if (!spec->pool_name)
2960 if (!*spec->pool_name)
2961 goto out_err; /* Missing pool name */
2963 spec->image_name = dup_token(&buf, &spec->image_name_len);
2964 if (!spec->image_name)
2966 if (!*spec->image_name)
2967 goto out_err; /* Missing image name */
2970 * Snapshot name is optional; default is to use "-"
2971 * (indicating the head/no snapshot).
2973 len = next_token(&buf);
2975 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
2976 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
2977 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
2978 ret = -ENAMETOOLONG;
2981 spec->snap_name = kmalloc(len + 1, GFP_KERNEL);
2982 if (!spec->snap_name)
2984 memcpy(spec->snap_name, buf, len);
2985 *(spec->snap_name + len) = '\0';
2987 /* Initialize all rbd options to the defaults */
2989 rbd_opts = kzalloc(sizeof (*rbd_opts), GFP_KERNEL);
2993 rbd_opts->read_only = RBD_READ_ONLY_DEFAULT;
2995 copts = ceph_parse_options(options, mon_addrs,
2996 mon_addrs + mon_addrs_size - 1,
2997 parse_rbd_opts_token, rbd_opts);
2998 if (IS_ERR(copts)) {
2999 ret = PTR_ERR(copts);
3020 * An rbd format 2 image has a unique identifier, distinct from the
3021 * name given to it by the user. Internally, that identifier is
3022 * what's used to specify the names of objects related to the image.
3024 * A special "rbd id" object is used to map an rbd image name to its
3025 * id. If that object doesn't exist, then there is no v2 rbd image
3026 * with the supplied name.
3028 * This function will record the given rbd_dev's image_id field if
3029 * it can be determined, and in that case will return 0. If any
3030 * errors occur a negative errno will be returned and the rbd_dev's
3031 * image_id field will be unchanged (and should be NULL).
3033 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
3042 * First, see if the format 2 image id file exists, and if
3043 * so, get the image's persistent id from it.
3045 size = sizeof (RBD_ID_PREFIX) + rbd_dev->spec->image_name_len;
3046 object_name = kmalloc(size, GFP_NOIO);
3049 sprintf(object_name, "%s%s", RBD_ID_PREFIX, rbd_dev->spec->image_name);
3050 dout("rbd id object name is %s\n", object_name);
3052 /* Response will be an encoded string, which includes a length */
3054 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
3055 response = kzalloc(size, GFP_NOIO);
3061 ret = rbd_req_sync_exec(rbd_dev, object_name,
3064 response, RBD_IMAGE_ID_LEN_MAX,
3065 CEPH_OSD_FLAG_READ, NULL);
3066 dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
3069 ret = 0; /* rbd_req_sync_exec() can return positive */
3072 rbd_dev->spec->image_id = ceph_extract_encoded_string(&p,
3073 p + RBD_IMAGE_ID_LEN_MAX,
3074 &rbd_dev->spec->image_id_len,
3076 if (IS_ERR(rbd_dev->spec->image_id)) {
3077 ret = PTR_ERR(rbd_dev->spec->image_id);
3078 rbd_dev->spec->image_id = NULL;
3080 dout("image_id is %s\n", rbd_dev->spec->image_id);
3089 static int rbd_dev_v1_probe(struct rbd_device *rbd_dev)
3094 /* Version 1 images have no id; empty string is used */
3096 rbd_dev->spec->image_id = kstrdup("", GFP_KERNEL);
3097 if (!rbd_dev->spec->image_id)
3099 rbd_dev->spec->image_id_len = 0;
3101 /* Record the header object name for this rbd image. */
3103 size = rbd_dev->spec->image_name_len + sizeof (RBD_SUFFIX);
3104 rbd_dev->header_name = kmalloc(size, GFP_KERNEL);
3105 if (!rbd_dev->header_name) {
3109 sprintf(rbd_dev->header_name, "%s%s",
3110 rbd_dev->spec->image_name, RBD_SUFFIX);
3112 /* Populate rbd image metadata */
3114 ret = rbd_read_header(rbd_dev, &rbd_dev->header);
3117 rbd_dev->image_format = 1;
3119 dout("discovered version 1 image, header name is %s\n",
3120 rbd_dev->header_name);
3125 kfree(rbd_dev->header_name);
3126 rbd_dev->header_name = NULL;
3127 kfree(rbd_dev->spec->image_id);
3128 rbd_dev->spec->image_id = NULL;
3133 static int rbd_dev_v2_probe(struct rbd_device *rbd_dev)
3140 * Image id was filled in by the caller. Record the header
3141 * object name for this rbd image.
3143 size = sizeof (RBD_HEADER_PREFIX) + rbd_dev->spec->image_id_len;
3144 rbd_dev->header_name = kmalloc(size, GFP_KERNEL);
3145 if (!rbd_dev->header_name)
3147 sprintf(rbd_dev->header_name, "%s%s",
3148 RBD_HEADER_PREFIX, rbd_dev->spec->image_id);
3150 /* Get the size and object order for the image */
3152 ret = rbd_dev_v2_image_size(rbd_dev);
3156 /* Get the object prefix (a.k.a. block_name) for the image */
3158 ret = rbd_dev_v2_object_prefix(rbd_dev);
3162 /* Get the and check features for the image */
3164 ret = rbd_dev_v2_features(rbd_dev);
3168 /* crypto and compression type aren't (yet) supported for v2 images */
3170 rbd_dev->header.crypt_type = 0;
3171 rbd_dev->header.comp_type = 0;
3173 /* Get the snapshot context, plus the header version */
3175 ret = rbd_dev_v2_snap_context(rbd_dev, &ver);
3178 rbd_dev->header.obj_version = ver;
3180 rbd_dev->image_format = 2;
3182 dout("discovered version 2 image, header name is %s\n",
3183 rbd_dev->header_name);
3187 kfree(rbd_dev->header_name);
3188 rbd_dev->header_name = NULL;
3189 kfree(rbd_dev->header.object_prefix);
3190 rbd_dev->header.object_prefix = NULL;
3196 * Probe for the existence of the header object for the given rbd
3197 * device. For format 2 images this includes determining the image
3200 static int rbd_dev_probe(struct rbd_device *rbd_dev)
3205 * Get the id from the image id object. If it's not a
3206 * format 2 image, we'll get ENOENT back, and we'll assume
3207 * it's a format 1 image.
3209 ret = rbd_dev_image_id(rbd_dev);
3211 ret = rbd_dev_v1_probe(rbd_dev);
3213 ret = rbd_dev_v2_probe(rbd_dev);
3215 dout("probe failed, returning %d\n", ret);
3220 static ssize_t rbd_add(struct bus_type *bus,
3224 struct rbd_device *rbd_dev = NULL;
3225 struct ceph_options *ceph_opts = NULL;
3226 struct rbd_options *rbd_opts = NULL;
3227 struct rbd_spec *spec = NULL;
3228 struct rbd_client *rbdc;
3229 struct ceph_osd_client *osdc;
3232 if (!try_module_get(THIS_MODULE))
3235 rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
3239 /* static rbd_device initialization */
3240 spin_lock_init(&rbd_dev->lock);
3241 INIT_LIST_HEAD(&rbd_dev->node);
3242 INIT_LIST_HEAD(&rbd_dev->snaps);
3243 init_rwsem(&rbd_dev->header_rwsem);
3245 /* parse add command */
3246 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
3250 rbd_dev->mapping.read_only = rbd_opts->read_only;
3252 rbdc = rbd_get_client(ceph_opts);
3257 rbd_dev->rbd_client = rbdc;
3258 ceph_opts = NULL; /* ceph_opts now owned by rbd_dev client */
3261 osdc = &rbdc->client->osdc;
3262 rc = ceph_pg_poolid_by_name(osdc->osdmap, spec->pool_name);
3264 goto err_out_client;
3265 spec->pool_id = (u64) rc;
3267 rbd_dev->spec = spec;
3269 rc = rbd_dev_probe(rbd_dev);
3271 goto err_out_client;
3273 /* no need to lock here, as rbd_dev is not registered yet */
3274 rc = rbd_dev_snaps_update(rbd_dev);
3278 rc = rbd_dev_set_mapping(rbd_dev);
3282 /* generate unique id: find highest unique id, add one */
3283 rbd_dev_id_get(rbd_dev);
3285 /* Fill in the device name, now that we have its id. */
3286 BUILD_BUG_ON(DEV_NAME_LEN
3287 < sizeof (RBD_DRV_NAME) + MAX_INT_FORMAT_WIDTH);
3288 sprintf(rbd_dev->name, "%s%d", RBD_DRV_NAME, rbd_dev->dev_id);
3290 /* Get our block major device number. */
3292 rc = register_blkdev(0, rbd_dev->name);
3295 rbd_dev->major = rc;
3297 /* Set up the blkdev mapping. */
3299 rc = rbd_init_disk(rbd_dev);
3301 goto err_out_blkdev;
3303 rc = rbd_bus_add_dev(rbd_dev);
3308 * At this point cleanup in the event of an error is the job
3309 * of the sysfs code (initiated by rbd_bus_del_dev()).
3312 down_write(&rbd_dev->header_rwsem);
3313 rc = rbd_dev_snaps_register(rbd_dev);
3314 up_write(&rbd_dev->header_rwsem);
3318 rc = rbd_init_watch_dev(rbd_dev);
3324 /* Everything's ready. Announce the disk to the world. */
3326 add_disk(rbd_dev->disk);
3328 pr_info("%s: added with size 0x%llx\n", rbd_dev->disk->disk_name,
3329 (unsigned long long) rbd_dev->mapping.size);
3334 /* this will also clean up rest of rbd_dev stuff */
3336 rbd_bus_del_dev(rbd_dev);
3342 rbd_free_disk(rbd_dev);
3344 unregister_blkdev(rbd_dev->major, rbd_dev->name);
3346 rbd_dev_id_put(rbd_dev);
3348 rbd_remove_all_snaps(rbd_dev);
3350 rbd_header_free(&rbd_dev->header);
3352 kfree(rbd_dev->header_name);
3353 rbd_put_client(rbdc);
3356 ceph_destroy_options(ceph_opts);
3362 dout("Error adding device %s\n", buf);
3363 module_put(THIS_MODULE);
3365 return (ssize_t) rc;
3368 static struct rbd_device *__rbd_get_dev(unsigned long dev_id)
3370 struct list_head *tmp;
3371 struct rbd_device *rbd_dev;
3373 spin_lock(&rbd_dev_list_lock);
3374 list_for_each(tmp, &rbd_dev_list) {
3375 rbd_dev = list_entry(tmp, struct rbd_device, node);
3376 if (rbd_dev->dev_id == dev_id) {
3377 spin_unlock(&rbd_dev_list_lock);
3381 spin_unlock(&rbd_dev_list_lock);
3385 static void rbd_dev_release(struct device *dev)
3387 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3389 if (rbd_dev->watch_request) {
3390 struct ceph_client *client = rbd_dev->rbd_client->client;
3392 ceph_osdc_unregister_linger_request(&client->osdc,
3393 rbd_dev->watch_request);
3395 if (rbd_dev->watch_event)
3396 rbd_req_sync_unwatch(rbd_dev);
3398 rbd_put_client(rbd_dev->rbd_client);
3400 /* clean up and free blkdev */
3401 rbd_free_disk(rbd_dev);
3402 unregister_blkdev(rbd_dev->major, rbd_dev->name);
3404 /* release allocated disk header fields */
3405 rbd_header_free(&rbd_dev->header);
3407 /* done with the id, and with the rbd_dev */
3408 kfree(rbd_dev->header_name);
3409 rbd_dev_id_put(rbd_dev);
3410 rbd_spec_put(rbd_dev->spec);
3413 /* release module ref */
3414 module_put(THIS_MODULE);
3417 static ssize_t rbd_remove(struct bus_type *bus,
3421 struct rbd_device *rbd_dev = NULL;
3426 rc = strict_strtoul(buf, 10, &ul);
3430 /* convert to int; abort if we lost anything in the conversion */
3431 target_id = (int) ul;
3432 if (target_id != ul)
3435 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
3437 rbd_dev = __rbd_get_dev(target_id);
3443 rbd_remove_all_snaps(rbd_dev);
3444 rbd_bus_del_dev(rbd_dev);
3447 mutex_unlock(&ctl_mutex);
3453 * create control files in sysfs
3456 static int rbd_sysfs_init(void)
3460 ret = device_register(&rbd_root_dev);
3464 ret = bus_register(&rbd_bus_type);
3466 device_unregister(&rbd_root_dev);
3471 static void rbd_sysfs_cleanup(void)
3473 bus_unregister(&rbd_bus_type);
3474 device_unregister(&rbd_root_dev);
3477 int __init rbd_init(void)
3481 rc = rbd_sysfs_init();
3484 pr_info("loaded " RBD_DRV_NAME_LONG "\n");
3488 void __exit rbd_exit(void)
3490 rbd_sysfs_cleanup();
3493 module_init(rbd_init);
3494 module_exit(rbd_exit);
3496 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
3497 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
3498 MODULE_DESCRIPTION("rados block device");
3500 /* following authorship retained from original osdblk.c */
3501 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
3503 MODULE_LICENSE("GPL");