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"
45 * The basic unit of block I/O is a sector. It is interpreted in a
46 * number of contexts in Linux (blk, bio, genhd), but the default is
47 * universally 512 bytes. These symbols are just slightly more
48 * meaningful than the bare numbers they represent.
50 #define SECTOR_SHIFT 9
51 #define SECTOR_SIZE (1ULL << SECTOR_SHIFT)
53 #define RBD_DRV_NAME "rbd"
54 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
56 #define RBD_MINORS_PER_MAJOR 256 /* max minors per blkdev */
58 #define RBD_MAX_SNAP_NAME_LEN 32
59 #define RBD_MAX_OPT_LEN 1024
61 #define RBD_SNAP_HEAD_NAME "-"
64 * An RBD device name will be "rbd#", where the "rbd" comes from
65 * RBD_DRV_NAME above, and # is a unique integer identifier.
66 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
67 * enough to hold all possible device names.
69 #define DEV_NAME_LEN 32
70 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1)
72 #define RBD_NOTIFY_TIMEOUT_DEFAULT 10
75 * block device image metadata (in-memory version)
77 struct rbd_image_header {
83 struct ceph_snap_context *snapc;
84 size_t snap_names_len;
98 * an instance of the client. multiple devices may share an rbd client.
101 struct ceph_client *client;
102 struct rbd_options *rbd_opts;
104 struct list_head node;
108 * a request completion status
110 struct rbd_req_status {
117 * a collection of requests
119 struct rbd_req_coll {
123 struct rbd_req_status status[0];
127 * a single io request
130 struct request *rq; /* blk layer request */
131 struct bio *bio; /* cloned bio */
132 struct page **pages; /* list of used pages */
135 struct rbd_req_coll *coll;
142 struct list_head node;
150 int dev_id; /* blkdev unique id */
152 int major; /* blkdev assigned major */
153 struct gendisk *disk; /* blkdev's gendisk and rq */
154 struct request_queue *q;
156 struct rbd_client *rbd_client;
158 char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
160 spinlock_t lock; /* queue lock */
162 struct rbd_image_header header;
164 size_t image_name_len;
169 struct ceph_osd_event *watch_event;
170 struct ceph_osd_request *watch_request;
172 /* protects updating the header */
173 struct rw_semaphore header_rwsem;
174 /* name of the snapshot this device reads from */
176 /* id of the snapshot this device reads from */
177 u64 snap_id; /* current snapshot id */
178 /* whether the snap_id this device reads from still exists */
182 struct list_head node;
184 /* list of snapshots */
185 struct list_head snaps;
191 static DEFINE_MUTEX(ctl_mutex); /* Serialize open/close/setup/teardown */
193 static LIST_HEAD(rbd_dev_list); /* devices */
194 static DEFINE_SPINLOCK(rbd_dev_list_lock);
196 static LIST_HEAD(rbd_client_list); /* clients */
197 static DEFINE_SPINLOCK(rbd_client_list_lock);
199 static int __rbd_init_snaps_header(struct rbd_device *rbd_dev);
200 static void rbd_dev_release(struct device *dev);
201 static ssize_t rbd_snap_add(struct device *dev,
202 struct device_attribute *attr,
205 static void __rbd_remove_snap_dev(struct rbd_snap *snap);
207 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
209 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
212 static struct bus_attribute rbd_bus_attrs[] = {
213 __ATTR(add, S_IWUSR, NULL, rbd_add),
214 __ATTR(remove, S_IWUSR, NULL, rbd_remove),
218 static struct bus_type rbd_bus_type = {
220 .bus_attrs = rbd_bus_attrs,
223 static void rbd_root_dev_release(struct device *dev)
227 static struct device rbd_root_dev = {
229 .release = rbd_root_dev_release,
233 static struct device *rbd_get_dev(struct rbd_device *rbd_dev)
235 return get_device(&rbd_dev->dev);
238 static void rbd_put_dev(struct rbd_device *rbd_dev)
240 put_device(&rbd_dev->dev);
243 static int __rbd_refresh_header(struct rbd_device *rbd_dev);
245 static int rbd_open(struct block_device *bdev, fmode_t mode)
247 struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
249 rbd_get_dev(rbd_dev);
251 set_device_ro(bdev, rbd_dev->read_only);
253 if ((mode & FMODE_WRITE) && rbd_dev->read_only)
259 static int rbd_release(struct gendisk *disk, fmode_t mode)
261 struct rbd_device *rbd_dev = disk->private_data;
263 rbd_put_dev(rbd_dev);
268 static const struct block_device_operations rbd_bd_ops = {
269 .owner = THIS_MODULE,
271 .release = rbd_release,
275 * Initialize an rbd client instance.
278 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts,
279 struct rbd_options *rbd_opts)
281 struct rbd_client *rbdc;
284 dout("rbd_client_create\n");
285 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
289 kref_init(&rbdc->kref);
290 INIT_LIST_HEAD(&rbdc->node);
292 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
294 rbdc->client = ceph_create_client(ceph_opts, rbdc, 0, 0);
295 if (IS_ERR(rbdc->client))
297 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
299 ret = ceph_open_session(rbdc->client);
303 rbdc->rbd_opts = rbd_opts;
305 spin_lock(&rbd_client_list_lock);
306 list_add_tail(&rbdc->node, &rbd_client_list);
307 spin_unlock(&rbd_client_list_lock);
309 mutex_unlock(&ctl_mutex);
311 dout("rbd_client_create created %p\n", rbdc);
315 ceph_destroy_client(rbdc->client);
317 mutex_unlock(&ctl_mutex);
321 ceph_destroy_options(ceph_opts);
326 * Find a ceph client with specific addr and configuration.
328 static struct rbd_client *__rbd_client_find(struct ceph_options *ceph_opts)
330 struct rbd_client *client_node;
332 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
335 list_for_each_entry(client_node, &rbd_client_list, node)
336 if (!ceph_compare_options(ceph_opts, client_node->client))
349 /* string args above */
352 static match_table_t rbd_opts_tokens = {
353 {Opt_notify_timeout, "notify_timeout=%d"},
355 /* string args above */
359 static int parse_rbd_opts_token(char *c, void *private)
361 struct rbd_options *rbd_opts = private;
362 substring_t argstr[MAX_OPT_ARGS];
363 int token, intval, ret;
365 token = match_token(c, rbd_opts_tokens, argstr);
369 if (token < Opt_last_int) {
370 ret = match_int(&argstr[0], &intval);
372 pr_err("bad mount option arg (not int) "
376 dout("got int token %d val %d\n", token, intval);
377 } else if (token > Opt_last_int && token < Opt_last_string) {
378 dout("got string token %d val %s\n", token,
381 dout("got token %d\n", token);
385 case Opt_notify_timeout:
386 rbd_opts->notify_timeout = intval;
395 * Get a ceph client with specific addr and configuration, if one does
396 * not exist create it.
398 static struct rbd_client *rbd_get_client(const char *mon_addr,
402 struct rbd_client *rbdc;
403 struct ceph_options *ceph_opts;
404 struct rbd_options *rbd_opts;
406 rbd_opts = kzalloc(sizeof(*rbd_opts), GFP_KERNEL);
408 return ERR_PTR(-ENOMEM);
410 rbd_opts->notify_timeout = RBD_NOTIFY_TIMEOUT_DEFAULT;
412 ceph_opts = ceph_parse_options(options, mon_addr,
413 mon_addr + mon_addr_len,
414 parse_rbd_opts_token, rbd_opts);
415 if (IS_ERR(ceph_opts)) {
417 return ERR_CAST(ceph_opts);
420 spin_lock(&rbd_client_list_lock);
421 rbdc = __rbd_client_find(ceph_opts);
423 /* using an existing client */
424 kref_get(&rbdc->kref);
425 spin_unlock(&rbd_client_list_lock);
427 ceph_destroy_options(ceph_opts);
432 spin_unlock(&rbd_client_list_lock);
434 rbdc = rbd_client_create(ceph_opts, rbd_opts);
443 * Destroy ceph client
445 * Caller must hold rbd_client_list_lock.
447 static void rbd_client_release(struct kref *kref)
449 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
451 dout("rbd_release_client %p\n", rbdc);
452 spin_lock(&rbd_client_list_lock);
453 list_del(&rbdc->node);
454 spin_unlock(&rbd_client_list_lock);
456 ceph_destroy_client(rbdc->client);
457 kfree(rbdc->rbd_opts);
462 * Drop reference to ceph client node. If it's not referenced anymore, release
465 static void rbd_put_client(struct rbd_device *rbd_dev)
467 kref_put(&rbd_dev->rbd_client->kref, rbd_client_release);
468 rbd_dev->rbd_client = NULL;
472 * Destroy requests collection
474 static void rbd_coll_release(struct kref *kref)
476 struct rbd_req_coll *coll =
477 container_of(kref, struct rbd_req_coll, kref);
479 dout("rbd_coll_release %p\n", coll);
483 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
485 return !memcmp(&ondisk->text,
486 RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT));
490 * Create a new header structure, translate header format from the on-disk
493 static int rbd_header_from_disk(struct rbd_image_header *header,
494 struct rbd_image_header_ondisk *ondisk,
499 if (!rbd_dev_ondisk_valid(ondisk))
502 snap_count = le32_to_cpu(ondisk->snap_count);
503 if (snap_count > (UINT_MAX - sizeof(struct ceph_snap_context))
506 header->snapc = kmalloc(sizeof(struct ceph_snap_context) +
507 snap_count * sizeof(u64),
512 header->snap_names_len = le64_to_cpu(ondisk->snap_names_len);
514 header->snap_names = kmalloc(header->snap_names_len,
516 if (!header->snap_names)
518 header->snap_sizes = kmalloc(snap_count * sizeof(u64),
520 if (!header->snap_sizes)
523 header->snap_names = NULL;
524 header->snap_sizes = NULL;
527 header->object_prefix = kmalloc(sizeof (ondisk->block_name) + 1,
529 if (!header->object_prefix)
532 memcpy(header->object_prefix, ondisk->block_name,
533 sizeof(ondisk->block_name));
534 header->object_prefix[sizeof (ondisk->block_name)] = '\0';
536 header->image_size = le64_to_cpu(ondisk->image_size);
537 header->obj_order = ondisk->options.order;
538 header->crypt_type = ondisk->options.crypt_type;
539 header->comp_type = ondisk->options.comp_type;
541 atomic_set(&header->snapc->nref, 1);
542 header->snapc->seq = le64_to_cpu(ondisk->snap_seq);
543 header->snapc->num_snaps = snap_count;
544 header->total_snaps = snap_count;
546 if (snap_count && allocated_snaps == snap_count) {
547 for (i = 0; i < snap_count; i++) {
548 header->snapc->snaps[i] =
549 le64_to_cpu(ondisk->snaps[i].id);
550 header->snap_sizes[i] =
551 le64_to_cpu(ondisk->snaps[i].image_size);
554 /* copy snapshot names */
555 memcpy(header->snap_names, &ondisk->snaps[i],
556 header->snap_names_len);
562 kfree(header->snap_sizes);
564 kfree(header->snap_names);
566 kfree(header->snapc);
570 static int snap_by_name(struct rbd_image_header *header, const char *snap_name,
574 char *p = header->snap_names;
576 for (i = 0; i < header->total_snaps; i++) {
577 if (!strcmp(snap_name, p)) {
579 /* Found it. Pass back its id and/or size */
582 *seq = header->snapc->snaps[i];
584 *size = header->snap_sizes[i];
587 p += strlen(p) + 1; /* Skip ahead to the next name */
592 static int rbd_header_set_snap(struct rbd_device *rbd_dev, u64 *size)
596 down_write(&rbd_dev->header_rwsem);
598 if (!memcmp(rbd_dev->snap_name, RBD_SNAP_HEAD_NAME,
599 sizeof (RBD_SNAP_HEAD_NAME))) {
600 rbd_dev->snap_id = CEPH_NOSNAP;
601 rbd_dev->snap_exists = false;
602 rbd_dev->read_only = 0;
604 *size = rbd_dev->header.image_size;
608 ret = snap_by_name(&rbd_dev->header, rbd_dev->snap_name,
612 rbd_dev->snap_id = snap_id;
613 rbd_dev->snap_exists = true;
614 rbd_dev->read_only = 1;
619 up_write(&rbd_dev->header_rwsem);
623 static void rbd_header_free(struct rbd_image_header *header)
625 kfree(header->object_prefix);
626 kfree(header->snap_sizes);
627 kfree(header->snap_names);
628 ceph_put_snap_context(header->snapc);
632 * get the actual striped segment name, offset and length
634 static u64 rbd_get_segment(struct rbd_image_header *header,
635 const char *object_prefix,
637 char *seg_name, u64 *segofs)
639 u64 seg = ofs >> header->obj_order;
642 snprintf(seg_name, RBD_MAX_SEG_NAME_LEN,
643 "%s.%012llx", object_prefix, seg);
645 ofs = ofs & ((1 << header->obj_order) - 1);
646 len = min_t(u64, len, (1 << header->obj_order) - ofs);
654 static int rbd_get_num_segments(struct rbd_image_header *header,
657 u64 start_seg = ofs >> header->obj_order;
658 u64 end_seg = (ofs + len - 1) >> header->obj_order;
659 return end_seg - start_seg + 1;
663 * returns the size of an object in the image
665 static u64 rbd_obj_bytes(struct rbd_image_header *header)
667 return 1 << header->obj_order;
674 static void bio_chain_put(struct bio *chain)
680 chain = chain->bi_next;
686 * zeros a bio chain, starting at specific offset
688 static void zero_bio_chain(struct bio *chain, int start_ofs)
697 bio_for_each_segment(bv, chain, i) {
698 if (pos + bv->bv_len > start_ofs) {
699 int remainder = max(start_ofs - pos, 0);
700 buf = bvec_kmap_irq(bv, &flags);
701 memset(buf + remainder, 0,
702 bv->bv_len - remainder);
703 bvec_kunmap_irq(buf, &flags);
708 chain = chain->bi_next;
713 * bio_chain_clone - clone a chain of bios up to a certain length.
714 * might return a bio_pair that will need to be released.
716 static struct bio *bio_chain_clone(struct bio **old, struct bio **next,
717 struct bio_pair **bp,
718 int len, gfp_t gfpmask)
720 struct bio *tmp, *old_chain = *old, *new_chain = NULL, *tail = NULL;
724 bio_pair_release(*bp);
728 while (old_chain && (total < len)) {
729 tmp = bio_kmalloc(gfpmask, old_chain->bi_max_vecs);
733 if (total + old_chain->bi_size > len) {
737 * this split can only happen with a single paged bio,
738 * split_bio will BUG_ON if this is not the case
740 dout("bio_chain_clone split! total=%d remaining=%d"
742 total, len - total, old_chain->bi_size);
744 /* split the bio. We'll release it either in the next
745 call, or it will have to be released outside */
746 bp = bio_split(old_chain, (len - total) / SECTOR_SIZE);
750 __bio_clone(tmp, &bp->bio1);
754 __bio_clone(tmp, old_chain);
755 *next = old_chain->bi_next;
759 gfpmask &= ~__GFP_WAIT;
763 new_chain = tail = tmp;
768 old_chain = old_chain->bi_next;
770 total += tmp->bi_size;
776 tail->bi_next = NULL;
783 dout("bio_chain_clone with err\n");
784 bio_chain_put(new_chain);
789 * helpers for osd request op vectors.
791 static int rbd_create_rw_ops(struct ceph_osd_req_op **ops,
796 *ops = kzalloc(sizeof(struct ceph_osd_req_op) * (num_ops + 1),
800 (*ops)[0].op = opcode;
802 * op extent offset and length will be set later on
803 * in calc_raw_layout()
805 (*ops)[0].payload_len = payload_len;
809 static void rbd_destroy_ops(struct ceph_osd_req_op *ops)
814 static void rbd_coll_end_req_index(struct request *rq,
815 struct rbd_req_coll *coll,
819 struct request_queue *q;
822 dout("rbd_coll_end_req_index %p index %d ret %d len %llu\n",
823 coll, index, ret, (unsigned long long) len);
829 blk_end_request(rq, ret, len);
835 spin_lock_irq(q->queue_lock);
836 coll->status[index].done = 1;
837 coll->status[index].rc = ret;
838 coll->status[index].bytes = len;
839 max = min = coll->num_done;
840 while (max < coll->total && coll->status[max].done)
843 for (i = min; i<max; i++) {
844 __blk_end_request(rq, coll->status[i].rc,
845 coll->status[i].bytes);
847 kref_put(&coll->kref, rbd_coll_release);
849 spin_unlock_irq(q->queue_lock);
852 static void rbd_coll_end_req(struct rbd_request *req,
855 rbd_coll_end_req_index(req->rq, req->coll, req->coll_index, ret, len);
859 * Send ceph osd request
861 static int rbd_do_request(struct request *rq,
862 struct rbd_device *rbd_dev,
863 struct ceph_snap_context *snapc,
865 const char *object_name, u64 ofs, u64 len,
870 struct ceph_osd_req_op *ops,
871 struct rbd_req_coll *coll,
873 void (*rbd_cb)(struct ceph_osd_request *req,
874 struct ceph_msg *msg),
875 struct ceph_osd_request **linger_req,
878 struct ceph_osd_request *req;
879 struct ceph_file_layout *layout;
882 struct timespec mtime = CURRENT_TIME;
883 struct rbd_request *req_data;
884 struct ceph_osd_request_head *reqhead;
885 struct ceph_osd_client *osdc;
887 req_data = kzalloc(sizeof(*req_data), GFP_NOIO);
890 rbd_coll_end_req_index(rq, coll, coll_index,
896 req_data->coll = coll;
897 req_data->coll_index = coll_index;
900 dout("rbd_do_request object_name=%s ofs=%llu len=%llu\n", object_name,
901 (unsigned long long) ofs, (unsigned long long) len);
903 osdc = &rbd_dev->rbd_client->client->osdc;
904 req = ceph_osdc_alloc_request(osdc, flags, snapc, ops,
905 false, GFP_NOIO, pages, bio);
911 req->r_callback = rbd_cb;
915 req_data->pages = pages;
918 req->r_priv = req_data;
920 reqhead = req->r_request->front.iov_base;
921 reqhead->snapid = cpu_to_le64(CEPH_NOSNAP);
923 strncpy(req->r_oid, object_name, sizeof(req->r_oid));
924 req->r_oid_len = strlen(req->r_oid);
926 layout = &req->r_file_layout;
927 memset(layout, 0, sizeof(*layout));
928 layout->fl_stripe_unit = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
929 layout->fl_stripe_count = cpu_to_le32(1);
930 layout->fl_object_size = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
931 layout->fl_pg_pool = cpu_to_le32(rbd_dev->pool_id);
932 ceph_calc_raw_layout(osdc, layout, snapid, ofs, &len, &bno,
935 ceph_osdc_build_request(req, ofs, &len,
939 req->r_oid, req->r_oid_len);
942 ceph_osdc_set_request_linger(osdc, req);
946 ret = ceph_osdc_start_request(osdc, req, false);
951 ret = ceph_osdc_wait_request(osdc, req);
953 *ver = le64_to_cpu(req->r_reassert_version.version);
954 dout("reassert_ver=%llu\n",
956 le64_to_cpu(req->r_reassert_version.version));
957 ceph_osdc_put_request(req);
962 bio_chain_put(req_data->bio);
963 ceph_osdc_put_request(req);
965 rbd_coll_end_req(req_data, ret, len);
971 * Ceph osd op callback
973 static void rbd_req_cb(struct ceph_osd_request *req, struct ceph_msg *msg)
975 struct rbd_request *req_data = req->r_priv;
976 struct ceph_osd_reply_head *replyhead;
977 struct ceph_osd_op *op;
983 replyhead = msg->front.iov_base;
984 WARN_ON(le32_to_cpu(replyhead->num_ops) == 0);
985 op = (void *)(replyhead + 1);
986 rc = le32_to_cpu(replyhead->result);
987 bytes = le64_to_cpu(op->extent.length);
988 read_op = (le16_to_cpu(op->op) == CEPH_OSD_OP_READ);
990 dout("rbd_req_cb bytes=%llu readop=%d rc=%d\n",
991 (unsigned long long) bytes, read_op, (int) rc);
993 if (rc == -ENOENT && read_op) {
994 zero_bio_chain(req_data->bio, 0);
996 } else if (rc == 0 && read_op && bytes < req_data->len) {
997 zero_bio_chain(req_data->bio, bytes);
998 bytes = req_data->len;
1001 rbd_coll_end_req(req_data, rc, bytes);
1004 bio_chain_put(req_data->bio);
1006 ceph_osdc_put_request(req);
1010 static void rbd_simple_req_cb(struct ceph_osd_request *req, struct ceph_msg *msg)
1012 ceph_osdc_put_request(req);
1016 * Do a synchronous ceph osd operation
1018 static int rbd_req_sync_op(struct rbd_device *rbd_dev,
1019 struct ceph_snap_context *snapc,
1023 struct ceph_osd_req_op *orig_ops,
1024 const char *object_name,
1027 struct ceph_osd_request **linger_req,
1031 struct page **pages;
1033 struct ceph_osd_req_op *ops = orig_ops;
1036 num_pages = calc_pages_for(ofs , len);
1037 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1039 return PTR_ERR(pages);
1042 payload_len = (flags & CEPH_OSD_FLAG_WRITE ? len : 0);
1043 ret = rbd_create_rw_ops(&ops, 1, opcode, payload_len);
1047 if ((flags & CEPH_OSD_FLAG_WRITE) && buf) {
1048 ret = ceph_copy_to_page_vector(pages, buf, ofs, len);
1054 ret = rbd_do_request(NULL, rbd_dev, snapc, snapid,
1055 object_name, ofs, len, NULL,
1065 if ((flags & CEPH_OSD_FLAG_READ) && buf)
1066 ret = ceph_copy_from_page_vector(pages, buf, ofs, ret);
1070 rbd_destroy_ops(ops);
1072 ceph_release_page_vector(pages, num_pages);
1077 * Do an asynchronous ceph osd operation
1079 static int rbd_do_op(struct request *rq,
1080 struct rbd_device *rbd_dev,
1081 struct ceph_snap_context *snapc,
1083 int opcode, int flags,
1086 struct rbd_req_coll *coll,
1093 struct ceph_osd_req_op *ops;
1096 seg_name = kmalloc(RBD_MAX_SEG_NAME_LEN + 1, GFP_NOIO);
1100 seg_len = rbd_get_segment(&rbd_dev->header,
1101 rbd_dev->header.object_prefix,
1103 seg_name, &seg_ofs);
1105 payload_len = (flags & CEPH_OSD_FLAG_WRITE ? seg_len : 0);
1107 ret = rbd_create_rw_ops(&ops, 1, opcode, payload_len);
1111 /* we've taken care of segment sizes earlier when we
1112 cloned the bios. We should never have a segment
1113 truncated at this point */
1114 BUG_ON(seg_len < len);
1116 ret = rbd_do_request(rq, rbd_dev, snapc, snapid,
1117 seg_name, seg_ofs, seg_len,
1123 rbd_req_cb, 0, NULL);
1125 rbd_destroy_ops(ops);
1132 * Request async osd write
1134 static int rbd_req_write(struct request *rq,
1135 struct rbd_device *rbd_dev,
1136 struct ceph_snap_context *snapc,
1139 struct rbd_req_coll *coll,
1142 return rbd_do_op(rq, rbd_dev, snapc, CEPH_NOSNAP,
1144 CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1145 ofs, len, bio, coll, coll_index);
1149 * Request async osd read
1151 static int rbd_req_read(struct request *rq,
1152 struct rbd_device *rbd_dev,
1156 struct rbd_req_coll *coll,
1159 return rbd_do_op(rq, rbd_dev, NULL,
1163 ofs, len, bio, coll, coll_index);
1167 * Request sync osd read
1169 static int rbd_req_sync_read(struct rbd_device *rbd_dev,
1171 const char *object_name,
1176 return rbd_req_sync_op(rbd_dev, NULL,
1181 object_name, ofs, len, buf, NULL, ver);
1185 * Request sync osd watch
1187 static int rbd_req_sync_notify_ack(struct rbd_device *rbd_dev,
1191 struct ceph_osd_req_op *ops;
1194 ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_NOTIFY_ACK, 0);
1198 ops[0].watch.ver = cpu_to_le64(ver);
1199 ops[0].watch.cookie = notify_id;
1200 ops[0].watch.flag = 0;
1202 ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,
1203 rbd_dev->header_name, 0, 0, NULL,
1208 rbd_simple_req_cb, 0, NULL);
1210 rbd_destroy_ops(ops);
1214 static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
1216 struct rbd_device *rbd_dev = (struct rbd_device *)data;
1223 dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
1224 rbd_dev->header_name, (unsigned long long) notify_id,
1225 (unsigned int) opcode);
1226 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1227 rc = __rbd_refresh_header(rbd_dev);
1228 hver = rbd_dev->header.obj_version;
1229 mutex_unlock(&ctl_mutex);
1231 pr_warning(RBD_DRV_NAME "%d got notification but failed to "
1232 " update snaps: %d\n", rbd_dev->major, rc);
1234 rbd_req_sync_notify_ack(rbd_dev, hver, notify_id);
1238 * Request sync osd watch
1240 static int rbd_req_sync_watch(struct rbd_device *rbd_dev)
1242 struct ceph_osd_req_op *ops;
1243 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1245 int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_WATCH, 0);
1249 ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0,
1250 (void *)rbd_dev, &rbd_dev->watch_event);
1254 ops[0].watch.ver = cpu_to_le64(rbd_dev->header.obj_version);
1255 ops[0].watch.cookie = cpu_to_le64(rbd_dev->watch_event->cookie);
1256 ops[0].watch.flag = 1;
1258 ret = rbd_req_sync_op(rbd_dev, NULL,
1261 CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1263 rbd_dev->header_name,
1265 &rbd_dev->watch_request, NULL);
1270 rbd_destroy_ops(ops);
1274 ceph_osdc_cancel_event(rbd_dev->watch_event);
1275 rbd_dev->watch_event = NULL;
1277 rbd_destroy_ops(ops);
1282 * Request sync osd unwatch
1284 static int rbd_req_sync_unwatch(struct rbd_device *rbd_dev,
1285 const char *object_name)
1287 struct ceph_osd_req_op *ops;
1289 int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_WATCH, 0);
1293 ops[0].watch.ver = 0;
1294 ops[0].watch.cookie = cpu_to_le64(rbd_dev->watch_event->cookie);
1295 ops[0].watch.flag = 0;
1297 ret = rbd_req_sync_op(rbd_dev, NULL,
1300 CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1302 object_name, 0, 0, NULL, NULL, NULL);
1304 rbd_destroy_ops(ops);
1305 ceph_osdc_cancel_event(rbd_dev->watch_event);
1306 rbd_dev->watch_event = NULL;
1310 struct rbd_notify_info {
1311 struct rbd_device *rbd_dev;
1314 static void rbd_notify_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
1316 struct rbd_device *rbd_dev = (struct rbd_device *)data;
1320 dout("rbd_notify_cb %s notify_id=%llu opcode=%u\n",
1321 rbd_dev->header_name, (unsigned long long) notify_id,
1322 (unsigned int) opcode);
1326 * Request sync osd notify
1328 static int rbd_req_sync_notify(struct rbd_device *rbd_dev)
1330 struct ceph_osd_req_op *ops;
1331 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1332 struct ceph_osd_event *event;
1333 struct rbd_notify_info info;
1334 int payload_len = sizeof(u32) + sizeof(u32);
1337 ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_NOTIFY, payload_len);
1341 info.rbd_dev = rbd_dev;
1343 ret = ceph_osdc_create_event(osdc, rbd_notify_cb, 1,
1344 (void *)&info, &event);
1348 ops[0].watch.ver = 1;
1349 ops[0].watch.flag = 1;
1350 ops[0].watch.cookie = event->cookie;
1351 ops[0].watch.prot_ver = RADOS_NOTIFY_VER;
1352 ops[0].watch.timeout = 12;
1354 ret = rbd_req_sync_op(rbd_dev, NULL,
1357 CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1359 rbd_dev->header_name,
1360 0, 0, NULL, NULL, NULL);
1364 ret = ceph_osdc_wait_event(event, CEPH_OSD_TIMEOUT_DEFAULT);
1365 dout("ceph_osdc_wait_event returned %d\n", ret);
1366 rbd_destroy_ops(ops);
1370 ceph_osdc_cancel_event(event);
1372 rbd_destroy_ops(ops);
1377 * Request sync osd read
1379 static int rbd_req_sync_exec(struct rbd_device *rbd_dev,
1380 const char *object_name,
1381 const char *class_name,
1382 const char *method_name,
1387 struct ceph_osd_req_op *ops;
1388 int class_name_len = strlen(class_name);
1389 int method_name_len = strlen(method_name);
1390 int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_CALL,
1391 class_name_len + method_name_len + len);
1395 ops[0].cls.class_name = class_name;
1396 ops[0].cls.class_len = (__u8) class_name_len;
1397 ops[0].cls.method_name = method_name;
1398 ops[0].cls.method_len = (__u8) method_name_len;
1399 ops[0].cls.argc = 0;
1400 ops[0].cls.indata = data;
1401 ops[0].cls.indata_len = len;
1403 ret = rbd_req_sync_op(rbd_dev, NULL,
1406 CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1408 object_name, 0, 0, NULL, NULL, ver);
1410 rbd_destroy_ops(ops);
1412 dout("cls_exec returned %d\n", ret);
1416 static struct rbd_req_coll *rbd_alloc_coll(int num_reqs)
1418 struct rbd_req_coll *coll =
1419 kzalloc(sizeof(struct rbd_req_coll) +
1420 sizeof(struct rbd_req_status) * num_reqs,
1425 coll->total = num_reqs;
1426 kref_init(&coll->kref);
1431 * block device queue callback
1433 static void rbd_rq_fn(struct request_queue *q)
1435 struct rbd_device *rbd_dev = q->queuedata;
1437 struct bio_pair *bp = NULL;
1439 while ((rq = blk_fetch_request(q))) {
1441 struct bio *rq_bio, *next_bio = NULL;
1446 int num_segs, cur_seg = 0;
1447 struct rbd_req_coll *coll;
1448 struct ceph_snap_context *snapc;
1450 /* peek at request from block layer */
1454 dout("fetched request\n");
1456 /* filter out block requests we don't understand */
1457 if ((rq->cmd_type != REQ_TYPE_FS)) {
1458 __blk_end_request_all(rq, 0);
1462 /* deduce our operation (read, write) */
1463 do_write = (rq_data_dir(rq) == WRITE);
1465 size = blk_rq_bytes(rq);
1466 ofs = blk_rq_pos(rq) * SECTOR_SIZE;
1468 if (do_write && rbd_dev->read_only) {
1469 __blk_end_request_all(rq, -EROFS);
1473 spin_unlock_irq(q->queue_lock);
1475 down_read(&rbd_dev->header_rwsem);
1477 if (rbd_dev->snap_id != CEPH_NOSNAP && !rbd_dev->snap_exists) {
1478 up_read(&rbd_dev->header_rwsem);
1479 dout("request for non-existent snapshot");
1480 spin_lock_irq(q->queue_lock);
1481 __blk_end_request_all(rq, -ENXIO);
1485 snapc = ceph_get_snap_context(rbd_dev->header.snapc);
1487 up_read(&rbd_dev->header_rwsem);
1489 dout("%s 0x%x bytes at 0x%llx\n",
1490 do_write ? "write" : "read",
1491 size, (unsigned long long) blk_rq_pos(rq) * SECTOR_SIZE);
1493 num_segs = rbd_get_num_segments(&rbd_dev->header, ofs, size);
1494 coll = rbd_alloc_coll(num_segs);
1496 spin_lock_irq(q->queue_lock);
1497 __blk_end_request_all(rq, -ENOMEM);
1498 ceph_put_snap_context(snapc);
1503 /* a bio clone to be passed down to OSD req */
1504 dout("rq->bio->bi_vcnt=%hu\n", rq->bio->bi_vcnt);
1505 op_size = rbd_get_segment(&rbd_dev->header,
1506 rbd_dev->header.object_prefix,
1509 kref_get(&coll->kref);
1510 bio = bio_chain_clone(&rq_bio, &next_bio, &bp,
1511 op_size, GFP_ATOMIC);
1513 rbd_coll_end_req_index(rq, coll, cur_seg,
1519 /* init OSD command: write or read */
1521 rbd_req_write(rq, rbd_dev,
1527 rbd_req_read(rq, rbd_dev,
1540 kref_put(&coll->kref, rbd_coll_release);
1543 bio_pair_release(bp);
1544 spin_lock_irq(q->queue_lock);
1546 ceph_put_snap_context(snapc);
1551 * a queue callback. Makes sure that we don't create a bio that spans across
1552 * multiple osd objects. One exception would be with a single page bios,
1553 * which we handle later at bio_chain_clone
1555 static int rbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bmd,
1556 struct bio_vec *bvec)
1558 struct rbd_device *rbd_dev = q->queuedata;
1559 unsigned int chunk_sectors;
1561 unsigned int bio_sectors;
1564 chunk_sectors = 1 << (rbd_dev->header.obj_order - SECTOR_SHIFT);
1565 sector = bmd->bi_sector + get_start_sect(bmd->bi_bdev);
1566 bio_sectors = bmd->bi_size >> SECTOR_SHIFT;
1568 max = (chunk_sectors - ((sector & (chunk_sectors - 1))
1569 + bio_sectors)) << SECTOR_SHIFT;
1571 max = 0; /* bio_add cannot handle a negative return */
1572 if (max <= bvec->bv_len && bio_sectors == 0)
1573 return bvec->bv_len;
1577 static void rbd_free_disk(struct rbd_device *rbd_dev)
1579 struct gendisk *disk = rbd_dev->disk;
1584 rbd_header_free(&rbd_dev->header);
1586 if (disk->flags & GENHD_FL_UP)
1589 blk_cleanup_queue(disk->queue);
1594 * reload the ondisk the header
1596 static int rbd_read_header(struct rbd_device *rbd_dev,
1597 struct rbd_image_header *header)
1600 struct rbd_image_header_ondisk *dh;
1606 * First reads the fixed-size header to determine the number
1607 * of snapshots, then re-reads it, along with all snapshot
1608 * records as well as their stored names.
1612 dh = kmalloc(len, GFP_KERNEL);
1616 rc = rbd_req_sync_read(rbd_dev,
1618 rbd_dev->header_name,
1624 rc = rbd_header_from_disk(header, dh, snap_count);
1627 pr_warning("unrecognized header format"
1629 rbd_dev->image_name);
1633 if (snap_count == header->total_snaps)
1636 snap_count = header->total_snaps;
1637 len = sizeof (*dh) +
1638 snap_count * sizeof(struct rbd_image_snap_ondisk) +
1639 header->snap_names_len;
1641 rbd_header_free(header);
1644 header->obj_version = ver;
1654 static int rbd_header_add_snap(struct rbd_device *rbd_dev,
1655 const char *snap_name,
1658 int name_len = strlen(snap_name);
1663 struct ceph_mon_client *monc;
1665 /* we should create a snapshot only if we're pointing at the head */
1666 if (rbd_dev->snap_id != CEPH_NOSNAP)
1669 monc = &rbd_dev->rbd_client->client->monc;
1670 ret = ceph_monc_create_snapid(monc, rbd_dev->pool_id, &new_snapid);
1671 dout("created snapid=%llu\n", (unsigned long long) new_snapid);
1675 data = kmalloc(name_len + 16, gfp_flags);
1680 e = data + name_len + 16;
1682 ceph_encode_string_safe(&p, e, snap_name, name_len, bad);
1683 ceph_encode_64_safe(&p, e, new_snapid, bad);
1685 ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
1687 data, p - data, &ver);
1691 return ret < 0 ? ret : 0;
1696 static void __rbd_remove_all_snaps(struct rbd_device *rbd_dev)
1698 struct rbd_snap *snap;
1699 struct rbd_snap *next;
1701 list_for_each_entry_safe(snap, next, &rbd_dev->snaps, node)
1702 __rbd_remove_snap_dev(snap);
1706 * only read the first part of the ondisk header, without the snaps info
1708 static int __rbd_refresh_header(struct rbd_device *rbd_dev)
1711 struct rbd_image_header h;
1713 ret = rbd_read_header(rbd_dev, &h);
1717 down_write(&rbd_dev->header_rwsem);
1720 if (rbd_dev->snap_id == CEPH_NOSNAP) {
1721 sector_t size = (sector_t) h.image_size / SECTOR_SIZE;
1723 dout("setting size to %llu sectors", (unsigned long long) size);
1724 set_capacity(rbd_dev->disk, size);
1727 /* rbd_dev->header.object_prefix shouldn't change */
1728 kfree(rbd_dev->header.snap_sizes);
1729 kfree(rbd_dev->header.snap_names);
1730 /* osd requests may still refer to snapc */
1731 ceph_put_snap_context(rbd_dev->header.snapc);
1733 rbd_dev->header.obj_version = h.obj_version;
1734 rbd_dev->header.image_size = h.image_size;
1735 rbd_dev->header.total_snaps = h.total_snaps;
1736 rbd_dev->header.snapc = h.snapc;
1737 rbd_dev->header.snap_names = h.snap_names;
1738 rbd_dev->header.snap_names_len = h.snap_names_len;
1739 rbd_dev->header.snap_sizes = h.snap_sizes;
1740 /* Free the extra copy of the object prefix */
1741 WARN_ON(strcmp(rbd_dev->header.object_prefix, h.object_prefix));
1742 kfree(h.object_prefix);
1744 ret = __rbd_init_snaps_header(rbd_dev);
1746 up_write(&rbd_dev->header_rwsem);
1751 static int rbd_init_disk(struct rbd_device *rbd_dev)
1753 struct gendisk *disk;
1754 struct request_queue *q;
1759 /* contact OSD, request size info about the object being mapped */
1760 rc = rbd_read_header(rbd_dev, &rbd_dev->header);
1764 /* no need to lock here, as rbd_dev is not registered yet */
1765 rc = __rbd_init_snaps_header(rbd_dev);
1769 rc = rbd_header_set_snap(rbd_dev, &total_size);
1773 /* create gendisk info */
1775 disk = alloc_disk(RBD_MINORS_PER_MAJOR);
1779 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
1781 disk->major = rbd_dev->major;
1782 disk->first_minor = 0;
1783 disk->fops = &rbd_bd_ops;
1784 disk->private_data = rbd_dev;
1788 q = blk_init_queue(rbd_rq_fn, &rbd_dev->lock);
1792 /* We use the default size, but let's be explicit about it. */
1793 blk_queue_physical_block_size(q, SECTOR_SIZE);
1795 /* set io sizes to object size */
1796 segment_size = rbd_obj_bytes(&rbd_dev->header);
1797 blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE);
1798 blk_queue_max_segment_size(q, segment_size);
1799 blk_queue_io_min(q, segment_size);
1800 blk_queue_io_opt(q, segment_size);
1802 blk_queue_merge_bvec(q, rbd_merge_bvec);
1805 q->queuedata = rbd_dev;
1807 rbd_dev->disk = disk;
1810 /* finally, announce the disk to the world */
1811 set_capacity(disk, total_size / SECTOR_SIZE);
1814 pr_info("%s: added with size 0x%llx\n",
1815 disk->disk_name, (unsigned long long)total_size);
1828 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
1830 return container_of(dev, struct rbd_device, dev);
1833 static ssize_t rbd_size_show(struct device *dev,
1834 struct device_attribute *attr, char *buf)
1836 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1839 down_read(&rbd_dev->header_rwsem);
1840 size = get_capacity(rbd_dev->disk);
1841 up_read(&rbd_dev->header_rwsem);
1843 return sprintf(buf, "%llu\n", (unsigned long long) size * SECTOR_SIZE);
1846 static ssize_t rbd_major_show(struct device *dev,
1847 struct device_attribute *attr, char *buf)
1849 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1851 return sprintf(buf, "%d\n", rbd_dev->major);
1854 static ssize_t rbd_client_id_show(struct device *dev,
1855 struct device_attribute *attr, char *buf)
1857 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1859 return sprintf(buf, "client%lld\n",
1860 ceph_client_id(rbd_dev->rbd_client->client));
1863 static ssize_t rbd_pool_show(struct device *dev,
1864 struct device_attribute *attr, char *buf)
1866 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1868 return sprintf(buf, "%s\n", rbd_dev->pool_name);
1871 static ssize_t rbd_pool_id_show(struct device *dev,
1872 struct device_attribute *attr, char *buf)
1874 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1876 return sprintf(buf, "%d\n", rbd_dev->pool_id);
1879 static ssize_t rbd_name_show(struct device *dev,
1880 struct device_attribute *attr, char *buf)
1882 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1884 return sprintf(buf, "%s\n", rbd_dev->image_name);
1887 static ssize_t rbd_snap_show(struct device *dev,
1888 struct device_attribute *attr,
1891 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1893 return sprintf(buf, "%s\n", rbd_dev->snap_name);
1896 static ssize_t rbd_image_refresh(struct device *dev,
1897 struct device_attribute *attr,
1901 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1905 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1907 rc = __rbd_refresh_header(rbd_dev);
1911 mutex_unlock(&ctl_mutex);
1915 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
1916 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
1917 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
1918 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
1919 static DEVICE_ATTR(pool_id, S_IRUGO, rbd_pool_id_show, NULL);
1920 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
1921 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
1922 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
1923 static DEVICE_ATTR(create_snap, S_IWUSR, NULL, rbd_snap_add);
1925 static struct attribute *rbd_attrs[] = {
1926 &dev_attr_size.attr,
1927 &dev_attr_major.attr,
1928 &dev_attr_client_id.attr,
1929 &dev_attr_pool.attr,
1930 &dev_attr_pool_id.attr,
1931 &dev_attr_name.attr,
1932 &dev_attr_current_snap.attr,
1933 &dev_attr_refresh.attr,
1934 &dev_attr_create_snap.attr,
1938 static struct attribute_group rbd_attr_group = {
1942 static const struct attribute_group *rbd_attr_groups[] = {
1947 static void rbd_sysfs_dev_release(struct device *dev)
1951 static struct device_type rbd_device_type = {
1953 .groups = rbd_attr_groups,
1954 .release = rbd_sysfs_dev_release,
1962 static ssize_t rbd_snap_size_show(struct device *dev,
1963 struct device_attribute *attr,
1966 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1968 return sprintf(buf, "%llu\n", (unsigned long long)snap->size);
1971 static ssize_t rbd_snap_id_show(struct device *dev,
1972 struct device_attribute *attr,
1975 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1977 return sprintf(buf, "%llu\n", (unsigned long long)snap->id);
1980 static DEVICE_ATTR(snap_size, S_IRUGO, rbd_snap_size_show, NULL);
1981 static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
1983 static struct attribute *rbd_snap_attrs[] = {
1984 &dev_attr_snap_size.attr,
1985 &dev_attr_snap_id.attr,
1989 static struct attribute_group rbd_snap_attr_group = {
1990 .attrs = rbd_snap_attrs,
1993 static void rbd_snap_dev_release(struct device *dev)
1995 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2000 static const struct attribute_group *rbd_snap_attr_groups[] = {
2001 &rbd_snap_attr_group,
2005 static struct device_type rbd_snap_device_type = {
2006 .groups = rbd_snap_attr_groups,
2007 .release = rbd_snap_dev_release,
2010 static void __rbd_remove_snap_dev(struct rbd_snap *snap)
2012 list_del(&snap->node);
2013 device_unregister(&snap->dev);
2016 static int rbd_register_snap_dev(struct rbd_snap *snap,
2017 struct device *parent)
2019 struct device *dev = &snap->dev;
2022 dev->type = &rbd_snap_device_type;
2023 dev->parent = parent;
2024 dev->release = rbd_snap_dev_release;
2025 dev_set_name(dev, "snap_%s", snap->name);
2026 ret = device_register(dev);
2031 static int __rbd_add_snap_dev(struct rbd_device *rbd_dev,
2032 int i, const char *name,
2033 struct rbd_snap **snapp)
2036 struct rbd_snap *snap = kzalloc(sizeof(*snap), GFP_KERNEL);
2039 snap->name = kstrdup(name, GFP_KERNEL);
2040 snap->size = rbd_dev->header.snap_sizes[i];
2041 snap->id = rbd_dev->header.snapc->snaps[i];
2042 if (device_is_registered(&rbd_dev->dev)) {
2043 ret = rbd_register_snap_dev(snap, &rbd_dev->dev);
2056 * search for the previous snap in a null delimited string list
2058 const char *rbd_prev_snap_name(const char *name, const char *start)
2060 if (name < start + 2)
2073 * compare the old list of snapshots that we have to what's in the header
2074 * and update it accordingly. Note that the header holds the snapshots
2075 * in a reverse order (from newest to oldest) and we need to go from
2076 * older to new so that we don't get a duplicate snap name when
2077 * doing the process (e.g., removed snapshot and recreated a new
2078 * one with the same name.
2080 static int __rbd_init_snaps_header(struct rbd_device *rbd_dev)
2082 const char *name, *first_name;
2083 int i = rbd_dev->header.total_snaps;
2084 struct rbd_snap *snap, *old_snap = NULL;
2086 struct list_head *p, *n;
2088 first_name = rbd_dev->header.snap_names;
2089 name = first_name + rbd_dev->header.snap_names_len;
2091 list_for_each_prev_safe(p, n, &rbd_dev->snaps) {
2094 old_snap = list_entry(p, struct rbd_snap, node);
2097 cur_id = rbd_dev->header.snapc->snaps[i - 1];
2099 if (!i || old_snap->id < cur_id) {
2101 * old_snap->id was skipped, thus was
2102 * removed. If this rbd_dev is mapped to
2103 * the removed snapshot, record that it no
2104 * longer exists, to prevent further I/O.
2106 if (rbd_dev->snap_id == old_snap->id)
2107 rbd_dev->snap_exists = false;
2108 __rbd_remove_snap_dev(old_snap);
2111 if (old_snap->id == cur_id) {
2112 /* we have this snapshot already */
2114 name = rbd_prev_snap_name(name, first_name);
2118 i--, name = rbd_prev_snap_name(name, first_name)) {
2123 cur_id = rbd_dev->header.snapc->snaps[i];
2124 /* snapshot removal? handle it above */
2125 if (cur_id >= old_snap->id)
2127 /* a new snapshot */
2128 ret = __rbd_add_snap_dev(rbd_dev, i - 1, name, &snap);
2132 /* note that we add it backward so using n and not p */
2133 list_add(&snap->node, n);
2137 /* we're done going over the old snap list, just add what's left */
2138 for (; i > 0; i--) {
2139 name = rbd_prev_snap_name(name, first_name);
2144 ret = __rbd_add_snap_dev(rbd_dev, i - 1, name, &snap);
2147 list_add(&snap->node, &rbd_dev->snaps);
2153 static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
2157 struct rbd_snap *snap;
2159 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2160 dev = &rbd_dev->dev;
2162 dev->bus = &rbd_bus_type;
2163 dev->type = &rbd_device_type;
2164 dev->parent = &rbd_root_dev;
2165 dev->release = rbd_dev_release;
2166 dev_set_name(dev, "%d", rbd_dev->dev_id);
2167 ret = device_register(dev);
2171 list_for_each_entry(snap, &rbd_dev->snaps, node) {
2172 ret = rbd_register_snap_dev(snap, &rbd_dev->dev);
2177 mutex_unlock(&ctl_mutex);
2181 static void rbd_bus_del_dev(struct rbd_device *rbd_dev)
2183 device_unregister(&rbd_dev->dev);
2186 static int rbd_init_watch_dev(struct rbd_device *rbd_dev)
2191 ret = rbd_req_sync_watch(rbd_dev);
2192 if (ret == -ERANGE) {
2193 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2194 rc = __rbd_refresh_header(rbd_dev);
2195 mutex_unlock(&ctl_mutex);
2199 } while (ret == -ERANGE);
2204 static atomic64_t rbd_id_max = ATOMIC64_INIT(0);
2207 * Get a unique rbd identifier for the given new rbd_dev, and add
2208 * the rbd_dev to the global list. The minimum rbd id is 1.
2210 static void rbd_id_get(struct rbd_device *rbd_dev)
2212 rbd_dev->dev_id = atomic64_inc_return(&rbd_id_max);
2214 spin_lock(&rbd_dev_list_lock);
2215 list_add_tail(&rbd_dev->node, &rbd_dev_list);
2216 spin_unlock(&rbd_dev_list_lock);
2220 * Remove an rbd_dev from the global list, and record that its
2221 * identifier is no longer in use.
2223 static void rbd_id_put(struct rbd_device *rbd_dev)
2225 struct list_head *tmp;
2226 int rbd_id = rbd_dev->dev_id;
2231 spin_lock(&rbd_dev_list_lock);
2232 list_del_init(&rbd_dev->node);
2235 * If the id being "put" is not the current maximum, there
2236 * is nothing special we need to do.
2238 if (rbd_id != atomic64_read(&rbd_id_max)) {
2239 spin_unlock(&rbd_dev_list_lock);
2244 * We need to update the current maximum id. Search the
2245 * list to find out what it is. We're more likely to find
2246 * the maximum at the end, so search the list backward.
2249 list_for_each_prev(tmp, &rbd_dev_list) {
2250 struct rbd_device *rbd_dev;
2252 rbd_dev = list_entry(tmp, struct rbd_device, node);
2253 if (rbd_id > max_id)
2256 spin_unlock(&rbd_dev_list_lock);
2259 * The max id could have been updated by rbd_id_get(), in
2260 * which case it now accurately reflects the new maximum.
2261 * Be careful not to overwrite the maximum value in that
2264 atomic64_cmpxchg(&rbd_id_max, rbd_id, max_id);
2268 * Skips over white space at *buf, and updates *buf to point to the
2269 * first found non-space character (if any). Returns the length of
2270 * the token (string of non-white space characters) found. Note
2271 * that *buf must be terminated with '\0'.
2273 static inline size_t next_token(const char **buf)
2276 * These are the characters that produce nonzero for
2277 * isspace() in the "C" and "POSIX" locales.
2279 const char *spaces = " \f\n\r\t\v";
2281 *buf += strspn(*buf, spaces); /* Find start of token */
2283 return strcspn(*buf, spaces); /* Return token length */
2287 * Finds the next token in *buf, and if the provided token buffer is
2288 * big enough, copies the found token into it. The result, if
2289 * copied, is guaranteed to be terminated with '\0'. Note that *buf
2290 * must be terminated with '\0' on entry.
2292 * Returns the length of the token found (not including the '\0').
2293 * Return value will be 0 if no token is found, and it will be >=
2294 * token_size if the token would not fit.
2296 * The *buf pointer will be updated to point beyond the end of the
2297 * found token. Note that this occurs even if the token buffer is
2298 * too small to hold it.
2300 static inline size_t copy_token(const char **buf,
2306 len = next_token(buf);
2307 if (len < token_size) {
2308 memcpy(token, *buf, len);
2309 *(token + len) = '\0';
2317 * Finds the next token in *buf, dynamically allocates a buffer big
2318 * enough to hold a copy of it, and copies the token into the new
2319 * buffer. The copy is guaranteed to be terminated with '\0'. Note
2320 * that a duplicate buffer is created even for a zero-length token.
2322 * Returns a pointer to the newly-allocated duplicate, or a null
2323 * pointer if memory for the duplicate was not available. If
2324 * the lenp argument is a non-null pointer, the length of the token
2325 * (not including the '\0') is returned in *lenp.
2327 * If successful, the *buf pointer will be updated to point beyond
2328 * the end of the found token.
2330 * Note: uses GFP_KERNEL for allocation.
2332 static inline char *dup_token(const char **buf, size_t *lenp)
2337 len = next_token(buf);
2338 dup = kmalloc(len + 1, GFP_KERNEL);
2342 memcpy(dup, *buf, len);
2343 *(dup + len) = '\0';
2353 * This fills in the pool_name, image_name, image_name_len, snap_name,
2354 * rbd_dev, rbd_md_name, and name fields of the given rbd_dev, based
2355 * on the list of monitor addresses and other options provided via
2358 * Note: rbd_dev is assumed to have been initially zero-filled.
2360 static int rbd_add_parse_args(struct rbd_device *rbd_dev,
2362 const char **mon_addrs,
2363 size_t *mon_addrs_size,
2365 size_t options_size)
2370 /* The first four tokens are required */
2372 len = next_token(&buf);
2375 *mon_addrs_size = len + 1;
2380 len = copy_token(&buf, options, options_size);
2381 if (!len || len >= options_size)
2385 rbd_dev->pool_name = dup_token(&buf, NULL);
2386 if (!rbd_dev->pool_name)
2389 rbd_dev->image_name = dup_token(&buf, &rbd_dev->image_name_len);
2390 if (!rbd_dev->image_name)
2393 /* Create the name of the header object */
2395 rbd_dev->header_name = kmalloc(rbd_dev->image_name_len
2396 + sizeof (RBD_SUFFIX),
2398 if (!rbd_dev->header_name)
2400 sprintf(rbd_dev->header_name, "%s%s", rbd_dev->image_name, RBD_SUFFIX);
2403 * The snapshot name is optional. If none is is supplied,
2404 * we use the default value.
2406 rbd_dev->snap_name = dup_token(&buf, &len);
2407 if (!rbd_dev->snap_name)
2410 /* Replace the empty name with the default */
2411 kfree(rbd_dev->snap_name);
2413 = kmalloc(sizeof (RBD_SNAP_HEAD_NAME), GFP_KERNEL);
2414 if (!rbd_dev->snap_name)
2417 memcpy(rbd_dev->snap_name, RBD_SNAP_HEAD_NAME,
2418 sizeof (RBD_SNAP_HEAD_NAME));
2424 kfree(rbd_dev->header_name);
2425 kfree(rbd_dev->image_name);
2426 kfree(rbd_dev->pool_name);
2427 rbd_dev->pool_name = NULL;
2432 static ssize_t rbd_add(struct bus_type *bus,
2437 struct rbd_device *rbd_dev = NULL;
2438 const char *mon_addrs = NULL;
2439 size_t mon_addrs_size = 0;
2440 struct ceph_osd_client *osdc;
2443 if (!try_module_get(THIS_MODULE))
2446 options = kmalloc(count, GFP_KERNEL);
2449 rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
2453 /* static rbd_device initialization */
2454 spin_lock_init(&rbd_dev->lock);
2455 INIT_LIST_HEAD(&rbd_dev->node);
2456 INIT_LIST_HEAD(&rbd_dev->snaps);
2457 init_rwsem(&rbd_dev->header_rwsem);
2459 /* generate unique id: find highest unique id, add one */
2460 rbd_id_get(rbd_dev);
2462 /* Fill in the device name, now that we have its id. */
2463 BUILD_BUG_ON(DEV_NAME_LEN
2464 < sizeof (RBD_DRV_NAME) + MAX_INT_FORMAT_WIDTH);
2465 sprintf(rbd_dev->name, "%s%d", RBD_DRV_NAME, rbd_dev->dev_id);
2467 /* parse add command */
2468 rc = rbd_add_parse_args(rbd_dev, buf, &mon_addrs, &mon_addrs_size,
2473 rbd_dev->rbd_client = rbd_get_client(mon_addrs, mon_addrs_size - 1,
2475 if (IS_ERR(rbd_dev->rbd_client)) {
2476 rc = PTR_ERR(rbd_dev->rbd_client);
2481 osdc = &rbd_dev->rbd_client->client->osdc;
2482 rc = ceph_pg_poolid_by_name(osdc->osdmap, rbd_dev->pool_name);
2484 goto err_out_client;
2485 rbd_dev->pool_id = rc;
2487 /* register our block device */
2488 rc = register_blkdev(0, rbd_dev->name);
2490 goto err_out_client;
2491 rbd_dev->major = rc;
2493 rc = rbd_bus_add_dev(rbd_dev);
2495 goto err_out_blkdev;
2498 * At this point cleanup in the event of an error is the job
2499 * of the sysfs code (initiated by rbd_bus_del_dev()).
2501 * Set up and announce blkdev mapping.
2503 rc = rbd_init_disk(rbd_dev);
2507 rc = rbd_init_watch_dev(rbd_dev);
2514 /* this will also clean up rest of rbd_dev stuff */
2516 rbd_bus_del_dev(rbd_dev);
2521 unregister_blkdev(rbd_dev->major, rbd_dev->name);
2523 rbd_put_client(rbd_dev);
2525 if (rbd_dev->pool_name) {
2526 kfree(rbd_dev->snap_name);
2527 kfree(rbd_dev->header_name);
2528 kfree(rbd_dev->image_name);
2529 kfree(rbd_dev->pool_name);
2531 rbd_id_put(rbd_dev);
2536 dout("Error adding device %s\n", buf);
2537 module_put(THIS_MODULE);
2539 return (ssize_t) rc;
2542 static struct rbd_device *__rbd_get_dev(unsigned long dev_id)
2544 struct list_head *tmp;
2545 struct rbd_device *rbd_dev;
2547 spin_lock(&rbd_dev_list_lock);
2548 list_for_each(tmp, &rbd_dev_list) {
2549 rbd_dev = list_entry(tmp, struct rbd_device, node);
2550 if (rbd_dev->dev_id == dev_id) {
2551 spin_unlock(&rbd_dev_list_lock);
2555 spin_unlock(&rbd_dev_list_lock);
2559 static void rbd_dev_release(struct device *dev)
2561 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2563 if (rbd_dev->watch_request) {
2564 struct ceph_client *client = rbd_dev->rbd_client->client;
2566 ceph_osdc_unregister_linger_request(&client->osdc,
2567 rbd_dev->watch_request);
2569 if (rbd_dev->watch_event)
2570 rbd_req_sync_unwatch(rbd_dev, rbd_dev->header_name);
2572 rbd_put_client(rbd_dev);
2574 /* clean up and free blkdev */
2575 rbd_free_disk(rbd_dev);
2576 unregister_blkdev(rbd_dev->major, rbd_dev->name);
2578 /* done with the id, and with the rbd_dev */
2579 kfree(rbd_dev->snap_name);
2580 kfree(rbd_dev->header_name);
2581 kfree(rbd_dev->pool_name);
2582 kfree(rbd_dev->image_name);
2583 rbd_id_put(rbd_dev);
2586 /* release module ref */
2587 module_put(THIS_MODULE);
2590 static ssize_t rbd_remove(struct bus_type *bus,
2594 struct rbd_device *rbd_dev = NULL;
2599 rc = strict_strtoul(buf, 10, &ul);
2603 /* convert to int; abort if we lost anything in the conversion */
2604 target_id = (int) ul;
2605 if (target_id != ul)
2608 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2610 rbd_dev = __rbd_get_dev(target_id);
2616 __rbd_remove_all_snaps(rbd_dev);
2617 rbd_bus_del_dev(rbd_dev);
2620 mutex_unlock(&ctl_mutex);
2624 static ssize_t rbd_snap_add(struct device *dev,
2625 struct device_attribute *attr,
2629 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2631 char *name = kmalloc(count + 1, GFP_KERNEL);
2635 snprintf(name, count, "%s", buf);
2637 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2639 ret = rbd_header_add_snap(rbd_dev,
2644 ret = __rbd_refresh_header(rbd_dev);
2648 /* shouldn't hold ctl_mutex when notifying.. notify might
2649 trigger a watch callback that would need to get that mutex */
2650 mutex_unlock(&ctl_mutex);
2652 /* make a best effort, don't error if failed */
2653 rbd_req_sync_notify(rbd_dev);
2660 mutex_unlock(&ctl_mutex);
2666 * create control files in sysfs
2669 static int rbd_sysfs_init(void)
2673 ret = device_register(&rbd_root_dev);
2677 ret = bus_register(&rbd_bus_type);
2679 device_unregister(&rbd_root_dev);
2684 static void rbd_sysfs_cleanup(void)
2686 bus_unregister(&rbd_bus_type);
2687 device_unregister(&rbd_root_dev);
2690 int __init rbd_init(void)
2694 rc = rbd_sysfs_init();
2697 pr_info("loaded " RBD_DRV_NAME_LONG "\n");
2701 void __exit rbd_exit(void)
2703 rbd_sysfs_cleanup();
2706 module_init(rbd_init);
2707 module_exit(rbd_exit);
2709 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
2710 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
2711 MODULE_DESCRIPTION("rados block device");
2713 /* following authorship retained from original osdblk.c */
2714 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
2716 MODULE_LICENSE("GPL");