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
;
99 * an instance of the client. multiple devices may share an rbd client.
102 struct ceph_client
*client
;
103 struct rbd_options
*rbd_opts
;
105 struct list_head node
;
109 * a request completion status
111 struct rbd_req_status
{
118 * a collection of requests
120 struct rbd_req_coll
{
124 struct rbd_req_status status
[0];
128 * a single io request
131 struct request
*rq
; /* blk layer request */
132 struct bio
*bio
; /* cloned bio */
133 struct page
**pages
; /* list of used pages */
136 struct rbd_req_coll
*coll
;
143 struct list_head node
;
151 int id
; /* blkdev unique id */
153 int major
; /* blkdev assigned major */
154 struct gendisk
*disk
; /* blkdev's gendisk and rq */
155 struct request_queue
*q
;
157 struct rbd_client
*rbd_client
;
159 char name
[DEV_NAME_LEN
]; /* blkdev name, e.g. rbd3 */
161 spinlock_t lock
; /* queue lock */
163 struct rbd_image_header header
;
165 size_t image_name_len
;
170 struct ceph_osd_event
*watch_event
;
171 struct ceph_osd_request
*watch_request
;
173 /* protects updating the header */
174 struct rw_semaphore header_rwsem
;
175 /* name of the snapshot this device reads from */
177 /* id of the snapshot this device reads from */
178 u64 snap_id
; /* current snapshot id */
179 /* whether the snap_id this device reads from still exists */
183 struct list_head node
;
185 /* list of snapshots */
186 struct list_head snaps
;
192 static DEFINE_MUTEX(ctl_mutex
); /* Serialize open/close/setup/teardown */
194 static LIST_HEAD(rbd_dev_list
); /* devices */
195 static DEFINE_SPINLOCK(rbd_dev_list_lock
);
197 static LIST_HEAD(rbd_client_list
); /* clients */
198 static DEFINE_SPINLOCK(rbd_client_list_lock
);
200 static int __rbd_init_snaps_header(struct rbd_device
*rbd_dev
);
201 static void rbd_dev_release(struct device
*dev
);
202 static ssize_t
rbd_snap_add(struct device
*dev
,
203 struct device_attribute
*attr
,
206 static void __rbd_remove_snap_dev(struct rbd_device
*rbd_dev
,
207 struct rbd_snap
*snap
);
209 static ssize_t
rbd_add(struct bus_type
*bus
, const char *buf
,
211 static ssize_t
rbd_remove(struct bus_type
*bus
, const char *buf
,
214 static struct bus_attribute rbd_bus_attrs
[] = {
215 __ATTR(add
, S_IWUSR
, NULL
, rbd_add
),
216 __ATTR(remove
, S_IWUSR
, NULL
, rbd_remove
),
220 static struct bus_type rbd_bus_type
= {
222 .bus_attrs
= rbd_bus_attrs
,
225 static void rbd_root_dev_release(struct device
*dev
)
229 static struct device rbd_root_dev
= {
231 .release
= rbd_root_dev_release
,
235 static struct device
*rbd_get_dev(struct rbd_device
*rbd_dev
)
237 return get_device(&rbd_dev
->dev
);
240 static void rbd_put_dev(struct rbd_device
*rbd_dev
)
242 put_device(&rbd_dev
->dev
);
245 static int __rbd_refresh_header(struct rbd_device
*rbd_dev
);
247 static int rbd_open(struct block_device
*bdev
, fmode_t mode
)
249 struct rbd_device
*rbd_dev
= bdev
->bd_disk
->private_data
;
251 rbd_get_dev(rbd_dev
);
253 set_device_ro(bdev
, rbd_dev
->read_only
);
255 if ((mode
& FMODE_WRITE
) && rbd_dev
->read_only
)
261 static int rbd_release(struct gendisk
*disk
, fmode_t mode
)
263 struct rbd_device
*rbd_dev
= disk
->private_data
;
265 rbd_put_dev(rbd_dev
);
270 static const struct block_device_operations rbd_bd_ops
= {
271 .owner
= THIS_MODULE
,
273 .release
= rbd_release
,
277 * Initialize an rbd client instance.
280 static struct rbd_client
*rbd_client_create(struct ceph_options
*ceph_opts
,
281 struct rbd_options
*rbd_opts
)
283 struct rbd_client
*rbdc
;
286 dout("rbd_client_create\n");
287 rbdc
= kmalloc(sizeof(struct rbd_client
), GFP_KERNEL
);
291 kref_init(&rbdc
->kref
);
292 INIT_LIST_HEAD(&rbdc
->node
);
294 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
296 rbdc
->client
= ceph_create_client(ceph_opts
, rbdc
, 0, 0);
297 if (IS_ERR(rbdc
->client
))
299 ceph_opts
= NULL
; /* Now rbdc->client is responsible for ceph_opts */
301 ret
= ceph_open_session(rbdc
->client
);
305 rbdc
->rbd_opts
= rbd_opts
;
307 spin_lock(&rbd_client_list_lock
);
308 list_add_tail(&rbdc
->node
, &rbd_client_list
);
309 spin_unlock(&rbd_client_list_lock
);
311 mutex_unlock(&ctl_mutex
);
313 dout("rbd_client_create created %p\n", rbdc
);
317 ceph_destroy_client(rbdc
->client
);
319 mutex_unlock(&ctl_mutex
);
323 ceph_destroy_options(ceph_opts
);
328 * Find a ceph client with specific addr and configuration.
330 static struct rbd_client
*__rbd_client_find(struct ceph_options
*ceph_opts
)
332 struct rbd_client
*client_node
;
334 if (ceph_opts
->flags
& CEPH_OPT_NOSHARE
)
337 list_for_each_entry(client_node
, &rbd_client_list
, node
)
338 if (!ceph_compare_options(ceph_opts
, client_node
->client
))
351 /* string args above */
354 static match_table_t rbd_opts_tokens
= {
355 {Opt_notify_timeout
, "notify_timeout=%d"},
357 /* string args above */
361 static int parse_rbd_opts_token(char *c
, void *private)
363 struct rbd_options
*rbd_opts
= private;
364 substring_t argstr
[MAX_OPT_ARGS
];
365 int token
, intval
, ret
;
367 token
= match_token(c
, rbd_opts_tokens
, argstr
);
371 if (token
< Opt_last_int
) {
372 ret
= match_int(&argstr
[0], &intval
);
374 pr_err("bad mount option arg (not int) "
378 dout("got int token %d val %d\n", token
, intval
);
379 } else if (token
> Opt_last_int
&& token
< Opt_last_string
) {
380 dout("got string token %d val %s\n", token
,
383 dout("got token %d\n", token
);
387 case Opt_notify_timeout
:
388 rbd_opts
->notify_timeout
= intval
;
397 * Get a ceph client with specific addr and configuration, if one does
398 * not exist create it.
400 static struct rbd_client
*rbd_get_client(const char *mon_addr
,
404 struct rbd_client
*rbdc
;
405 struct ceph_options
*ceph_opts
;
406 struct rbd_options
*rbd_opts
;
408 rbd_opts
= kzalloc(sizeof(*rbd_opts
), GFP_KERNEL
);
410 return ERR_PTR(-ENOMEM
);
412 rbd_opts
->notify_timeout
= RBD_NOTIFY_TIMEOUT_DEFAULT
;
414 ceph_opts
= ceph_parse_options(options
, mon_addr
,
415 mon_addr
+ mon_addr_len
,
416 parse_rbd_opts_token
, rbd_opts
);
417 if (IS_ERR(ceph_opts
)) {
419 return ERR_CAST(ceph_opts
);
422 spin_lock(&rbd_client_list_lock
);
423 rbdc
= __rbd_client_find(ceph_opts
);
425 /* using an existing client */
426 kref_get(&rbdc
->kref
);
427 spin_unlock(&rbd_client_list_lock
);
429 ceph_destroy_options(ceph_opts
);
434 spin_unlock(&rbd_client_list_lock
);
436 rbdc
= rbd_client_create(ceph_opts
, rbd_opts
);
445 * Destroy ceph client
447 * Caller must hold rbd_client_list_lock.
449 static void rbd_client_release(struct kref
*kref
)
451 struct rbd_client
*rbdc
= container_of(kref
, struct rbd_client
, kref
);
453 dout("rbd_release_client %p\n", rbdc
);
454 spin_lock(&rbd_client_list_lock
);
455 list_del(&rbdc
->node
);
456 spin_unlock(&rbd_client_list_lock
);
458 ceph_destroy_client(rbdc
->client
);
459 kfree(rbdc
->rbd_opts
);
464 * Drop reference to ceph client node. If it's not referenced anymore, release
467 static void rbd_put_client(struct rbd_device
*rbd_dev
)
469 kref_put(&rbd_dev
->rbd_client
->kref
, rbd_client_release
);
470 rbd_dev
->rbd_client
= NULL
;
474 * Destroy requests collection
476 static void rbd_coll_release(struct kref
*kref
)
478 struct rbd_req_coll
*coll
=
479 container_of(kref
, struct rbd_req_coll
, kref
);
481 dout("rbd_coll_release %p\n", coll
);
486 * Create a new header structure, translate header format from the on-disk
489 static int rbd_header_from_disk(struct rbd_image_header
*header
,
490 struct rbd_image_header_ondisk
*ondisk
,
496 if (memcmp(ondisk
, RBD_HEADER_TEXT
, sizeof(RBD_HEADER_TEXT
)))
499 snap_count
= le32_to_cpu(ondisk
->snap_count
);
500 if (snap_count
> (UINT_MAX
- sizeof(struct ceph_snap_context
))
503 header
->snapc
= kmalloc(sizeof(struct ceph_snap_context
) +
504 snap_count
* sizeof(u64
),
509 header
->snap_names_len
= le64_to_cpu(ondisk
->snap_names_len
);
511 header
->snap_names
= kmalloc(header
->snap_names_len
,
513 if (!header
->snap_names
)
515 header
->snap_sizes
= kmalloc(snap_count
* sizeof(u64
),
517 if (!header
->snap_sizes
)
520 header
->snap_names
= NULL
;
521 header
->snap_sizes
= NULL
;
524 header
->object_prefix
= kmalloc(sizeof (ondisk
->block_name
) + 1,
526 if (!header
->object_prefix
)
529 memcpy(header
->object_prefix
, ondisk
->block_name
,
530 sizeof(ondisk
->block_name
));
531 header
->object_prefix
[sizeof (ondisk
->block_name
)] = '\0';
533 header
->image_size
= le64_to_cpu(ondisk
->image_size
);
534 header
->obj_order
= ondisk
->options
.order
;
535 header
->crypt_type
= ondisk
->options
.crypt_type
;
536 header
->comp_type
= ondisk
->options
.comp_type
;
538 atomic_set(&header
->snapc
->nref
, 1);
539 header
->snap_seq
= le64_to_cpu(ondisk
->snap_seq
);
540 header
->snapc
->seq
= le64_to_cpu(ondisk
->snap_seq
);
541 header
->snapc
->num_snaps
= snap_count
;
542 header
->total_snaps
= snap_count
;
544 if (snap_count
&& allocated_snaps
== snap_count
) {
545 for (i
= 0; i
< snap_count
; i
++) {
546 header
->snapc
->snaps
[i
] =
547 le64_to_cpu(ondisk
->snaps
[i
].id
);
548 header
->snap_sizes
[i
] =
549 le64_to_cpu(ondisk
->snaps
[i
].image_size
);
552 /* copy snapshot names */
553 memcpy(header
->snap_names
, &ondisk
->snaps
[i
],
554 header
->snap_names_len
);
560 kfree(header
->snap_sizes
);
562 kfree(header
->snap_names
);
564 kfree(header
->snapc
);
568 static int snap_by_name(struct rbd_image_header
*header
, const char *snap_name
,
572 char *p
= header
->snap_names
;
574 for (i
= 0; i
< header
->total_snaps
; i
++) {
575 if (!strcmp(snap_name
, p
)) {
577 /* Found it. Pass back its id and/or size */
580 *seq
= header
->snapc
->snaps
[i
];
582 *size
= header
->snap_sizes
[i
];
585 p
+= strlen(p
) + 1; /* Skip ahead to the next name */
590 static int rbd_header_set_snap(struct rbd_device
*rbd_dev
, u64
*size
)
594 down_write(&rbd_dev
->header_rwsem
);
596 if (!memcmp(rbd_dev
->snap_name
, RBD_SNAP_HEAD_NAME
,
597 sizeof (RBD_SNAP_HEAD_NAME
))) {
598 rbd_dev
->snap_id
= CEPH_NOSNAP
;
599 rbd_dev
->snap_exists
= false;
600 rbd_dev
->read_only
= 0;
602 *size
= rbd_dev
->header
.image_size
;
606 ret
= snap_by_name(&rbd_dev
->header
, rbd_dev
->snap_name
,
610 rbd_dev
->snap_id
= snap_id
;
611 rbd_dev
->snap_exists
= true;
612 rbd_dev
->read_only
= 1;
617 up_write(&rbd_dev
->header_rwsem
);
621 static void rbd_header_free(struct rbd_image_header
*header
)
623 kfree(header
->object_prefix
);
624 kfree(header
->snap_sizes
);
625 kfree(header
->snap_names
);
626 ceph_put_snap_context(header
->snapc
);
630 * get the actual striped segment name, offset and length
632 static u64
rbd_get_segment(struct rbd_image_header
*header
,
633 const char *object_prefix
,
635 char *seg_name
, u64
*segofs
)
637 u64 seg
= ofs
>> header
->obj_order
;
640 snprintf(seg_name
, RBD_MAX_SEG_NAME_LEN
,
641 "%s.%012llx", object_prefix
, seg
);
643 ofs
= ofs
& ((1 << header
->obj_order
) - 1);
644 len
= min_t(u64
, len
, (1 << header
->obj_order
) - ofs
);
652 static int rbd_get_num_segments(struct rbd_image_header
*header
,
655 u64 start_seg
= ofs
>> header
->obj_order
;
656 u64 end_seg
= (ofs
+ len
- 1) >> header
->obj_order
;
657 return end_seg
- start_seg
+ 1;
661 * returns the size of an object in the image
663 static u64
rbd_obj_bytes(struct rbd_image_header
*header
)
665 return 1 << header
->obj_order
;
672 static void bio_chain_put(struct bio
*chain
)
678 chain
= chain
->bi_next
;
684 * zeros a bio chain, starting at specific offset
686 static void zero_bio_chain(struct bio
*chain
, int start_ofs
)
695 bio_for_each_segment(bv
, chain
, i
) {
696 if (pos
+ bv
->bv_len
> start_ofs
) {
697 int remainder
= max(start_ofs
- pos
, 0);
698 buf
= bvec_kmap_irq(bv
, &flags
);
699 memset(buf
+ remainder
, 0,
700 bv
->bv_len
- remainder
);
701 bvec_kunmap_irq(buf
, &flags
);
706 chain
= chain
->bi_next
;
711 * bio_chain_clone - clone a chain of bios up to a certain length.
712 * might return a bio_pair that will need to be released.
714 static struct bio
*bio_chain_clone(struct bio
**old
, struct bio
**next
,
715 struct bio_pair
**bp
,
716 int len
, gfp_t gfpmask
)
718 struct bio
*tmp
, *old_chain
= *old
, *new_chain
= NULL
, *tail
= NULL
;
722 bio_pair_release(*bp
);
726 while (old_chain
&& (total
< len
)) {
727 tmp
= bio_kmalloc(gfpmask
, old_chain
->bi_max_vecs
);
731 if (total
+ old_chain
->bi_size
> len
) {
735 * this split can only happen with a single paged bio,
736 * split_bio will BUG_ON if this is not the case
738 dout("bio_chain_clone split! total=%d remaining=%d"
740 (int)total
, (int)len
-total
,
741 (int)old_chain
->bi_size
);
743 /* split the bio. We'll release it either in the next
744 call, or it will have to be released outside */
745 bp
= bio_split(old_chain
, (len
- total
) / SECTOR_SIZE
);
749 __bio_clone(tmp
, &bp
->bio1
);
753 __bio_clone(tmp
, old_chain
);
754 *next
= old_chain
->bi_next
;
758 gfpmask
&= ~__GFP_WAIT
;
762 new_chain
= tail
= tmp
;
767 old_chain
= old_chain
->bi_next
;
769 total
+= tmp
->bi_size
;
775 tail
->bi_next
= NULL
;
782 dout("bio_chain_clone with err\n");
783 bio_chain_put(new_chain
);
788 * helpers for osd request op vectors.
790 static int rbd_create_rw_ops(struct ceph_osd_req_op
**ops
,
795 *ops
= kzalloc(sizeof(struct ceph_osd_req_op
) * (num_ops
+ 1),
799 (*ops
)[0].op
= opcode
;
801 * op extent offset and length will be set later on
802 * in calc_raw_layout()
804 (*ops
)[0].payload_len
= payload_len
;
808 static void rbd_destroy_ops(struct ceph_osd_req_op
*ops
)
813 static void rbd_coll_end_req_index(struct request
*rq
,
814 struct rbd_req_coll
*coll
,
818 struct request_queue
*q
;
821 dout("rbd_coll_end_req_index %p index %d ret %d len %lld\n",
822 coll
, index
, ret
, len
);
828 blk_end_request(rq
, ret
, len
);
834 spin_lock_irq(q
->queue_lock
);
835 coll
->status
[index
].done
= 1;
836 coll
->status
[index
].rc
= ret
;
837 coll
->status
[index
].bytes
= len
;
838 max
= min
= coll
->num_done
;
839 while (max
< coll
->total
&& coll
->status
[max
].done
)
842 for (i
= min
; i
<max
; i
++) {
843 __blk_end_request(rq
, coll
->status
[i
].rc
,
844 coll
->status
[i
].bytes
);
846 kref_put(&coll
->kref
, rbd_coll_release
);
848 spin_unlock_irq(q
->queue_lock
);
851 static void rbd_coll_end_req(struct rbd_request
*req
,
854 rbd_coll_end_req_index(req
->rq
, req
->coll
, req
->coll_index
, ret
, len
);
858 * Send ceph osd request
860 static int rbd_do_request(struct request
*rq
,
861 struct rbd_device
*rbd_dev
,
862 struct ceph_snap_context
*snapc
,
864 const char *object_name
, u64 ofs
, u64 len
,
869 struct ceph_osd_req_op
*ops
,
870 struct rbd_req_coll
*coll
,
872 void (*rbd_cb
)(struct ceph_osd_request
*req
,
873 struct ceph_msg
*msg
),
874 struct ceph_osd_request
**linger_req
,
877 struct ceph_osd_request
*req
;
878 struct ceph_file_layout
*layout
;
881 struct timespec mtime
= CURRENT_TIME
;
882 struct rbd_request
*req_data
;
883 struct ceph_osd_request_head
*reqhead
;
884 struct ceph_osd_client
*osdc
;
886 req_data
= kzalloc(sizeof(*req_data
), GFP_NOIO
);
889 rbd_coll_end_req_index(rq
, coll
, coll_index
,
895 req_data
->coll
= coll
;
896 req_data
->coll_index
= coll_index
;
899 dout("rbd_do_request object_name=%s ofs=%lld len=%lld\n",
900 object_name
, len
, ofs
);
902 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
903 req
= ceph_osdc_alloc_request(osdc
, flags
, snapc
, ops
,
904 false, GFP_NOIO
, pages
, bio
);
910 req
->r_callback
= rbd_cb
;
914 req_data
->pages
= pages
;
917 req
->r_priv
= req_data
;
919 reqhead
= req
->r_request
->front
.iov_base
;
920 reqhead
->snapid
= cpu_to_le64(CEPH_NOSNAP
);
922 strncpy(req
->r_oid
, object_name
, sizeof(req
->r_oid
));
923 req
->r_oid_len
= strlen(req
->r_oid
);
925 layout
= &req
->r_file_layout
;
926 memset(layout
, 0, sizeof(*layout
));
927 layout
->fl_stripe_unit
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
928 layout
->fl_stripe_count
= cpu_to_le32(1);
929 layout
->fl_object_size
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
930 layout
->fl_pg_pool
= cpu_to_le32(rbd_dev
->pool_id
);
931 ceph_calc_raw_layout(osdc
, layout
, snapid
, ofs
, &len
, &bno
,
934 ceph_osdc_build_request(req
, ofs
, &len
,
938 req
->r_oid
, req
->r_oid_len
);
941 ceph_osdc_set_request_linger(osdc
, req
);
945 ret
= ceph_osdc_start_request(osdc
, req
, false);
950 ret
= ceph_osdc_wait_request(osdc
, req
);
952 *ver
= le64_to_cpu(req
->r_reassert_version
.version
);
953 dout("reassert_ver=%lld\n",
954 le64_to_cpu(req
->r_reassert_version
.version
));
955 ceph_osdc_put_request(req
);
960 bio_chain_put(req_data
->bio
);
961 ceph_osdc_put_request(req
);
963 rbd_coll_end_req(req_data
, ret
, len
);
969 * Ceph osd op callback
971 static void rbd_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
973 struct rbd_request
*req_data
= req
->r_priv
;
974 struct ceph_osd_reply_head
*replyhead
;
975 struct ceph_osd_op
*op
;
981 replyhead
= msg
->front
.iov_base
;
982 WARN_ON(le32_to_cpu(replyhead
->num_ops
) == 0);
983 op
= (void *)(replyhead
+ 1);
984 rc
= le32_to_cpu(replyhead
->result
);
985 bytes
= le64_to_cpu(op
->extent
.length
);
986 read_op
= (le16_to_cpu(op
->op
) == CEPH_OSD_OP_READ
);
988 dout("rbd_req_cb bytes=%lld readop=%d rc=%d\n", bytes
, read_op
, rc
);
990 if (rc
== -ENOENT
&& read_op
) {
991 zero_bio_chain(req_data
->bio
, 0);
993 } else if (rc
== 0 && read_op
&& bytes
< req_data
->len
) {
994 zero_bio_chain(req_data
->bio
, bytes
);
995 bytes
= req_data
->len
;
998 rbd_coll_end_req(req_data
, rc
, bytes
);
1001 bio_chain_put(req_data
->bio
);
1003 ceph_osdc_put_request(req
);
1007 static void rbd_simple_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
1009 ceph_osdc_put_request(req
);
1013 * Do a synchronous ceph osd operation
1015 static int rbd_req_sync_op(struct rbd_device
*rbd_dev
,
1016 struct ceph_snap_context
*snapc
,
1020 struct ceph_osd_req_op
*orig_ops
,
1021 const char *object_name
,
1024 struct ceph_osd_request
**linger_req
,
1028 struct page
**pages
;
1030 struct ceph_osd_req_op
*ops
= orig_ops
;
1033 num_pages
= calc_pages_for(ofs
, len
);
1034 pages
= ceph_alloc_page_vector(num_pages
, GFP_KERNEL
);
1036 return PTR_ERR(pages
);
1039 payload_len
= (flags
& CEPH_OSD_FLAG_WRITE
? len
: 0);
1040 ret
= rbd_create_rw_ops(&ops
, 1, opcode
, payload_len
);
1044 if ((flags
& CEPH_OSD_FLAG_WRITE
) && buf
) {
1045 ret
= ceph_copy_to_page_vector(pages
, buf
, ofs
, len
);
1051 ret
= rbd_do_request(NULL
, rbd_dev
, snapc
, snapid
,
1052 object_name
, ofs
, len
, NULL
,
1062 if ((flags
& CEPH_OSD_FLAG_READ
) && buf
)
1063 ret
= ceph_copy_from_page_vector(pages
, buf
, ofs
, ret
);
1067 rbd_destroy_ops(ops
);
1069 ceph_release_page_vector(pages
, num_pages
);
1074 * Do an asynchronous ceph osd operation
1076 static int rbd_do_op(struct request
*rq
,
1077 struct rbd_device
*rbd_dev
,
1078 struct ceph_snap_context
*snapc
,
1080 int opcode
, int flags
,
1083 struct rbd_req_coll
*coll
,
1090 struct ceph_osd_req_op
*ops
;
1093 seg_name
= kmalloc(RBD_MAX_SEG_NAME_LEN
+ 1, GFP_NOIO
);
1097 seg_len
= rbd_get_segment(&rbd_dev
->header
,
1098 rbd_dev
->header
.object_prefix
,
1100 seg_name
, &seg_ofs
);
1102 payload_len
= (flags
& CEPH_OSD_FLAG_WRITE
? seg_len
: 0);
1104 ret
= rbd_create_rw_ops(&ops
, 1, opcode
, payload_len
);
1108 /* we've taken care of segment sizes earlier when we
1109 cloned the bios. We should never have a segment
1110 truncated at this point */
1111 BUG_ON(seg_len
< len
);
1113 ret
= rbd_do_request(rq
, rbd_dev
, snapc
, snapid
,
1114 seg_name
, seg_ofs
, seg_len
,
1120 rbd_req_cb
, 0, NULL
);
1122 rbd_destroy_ops(ops
);
1129 * Request async osd write
1131 static int rbd_req_write(struct request
*rq
,
1132 struct rbd_device
*rbd_dev
,
1133 struct ceph_snap_context
*snapc
,
1136 struct rbd_req_coll
*coll
,
1139 return rbd_do_op(rq
, rbd_dev
, snapc
, CEPH_NOSNAP
,
1141 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1142 ofs
, len
, bio
, coll
, coll_index
);
1146 * Request async osd read
1148 static int rbd_req_read(struct request
*rq
,
1149 struct rbd_device
*rbd_dev
,
1153 struct rbd_req_coll
*coll
,
1156 return rbd_do_op(rq
, rbd_dev
, NULL
,
1160 ofs
, len
, bio
, coll
, coll_index
);
1164 * Request sync osd read
1166 static int rbd_req_sync_read(struct rbd_device
*rbd_dev
,
1167 struct ceph_snap_context
*snapc
,
1169 const char *object_name
,
1174 return rbd_req_sync_op(rbd_dev
, NULL
,
1179 object_name
, ofs
, len
, buf
, NULL
, ver
);
1183 * Request sync osd watch
1185 static int rbd_req_sync_notify_ack(struct rbd_device
*rbd_dev
,
1188 const char *object_name
)
1190 struct ceph_osd_req_op
*ops
;
1193 ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_NOTIFY_ACK
, 0);
1197 ops
[0].watch
.ver
= cpu_to_le64(ver
);
1198 ops
[0].watch
.cookie
= notify_id
;
1199 ops
[0].watch
.flag
= 0;
1201 ret
= rbd_do_request(NULL
, rbd_dev
, NULL
, CEPH_NOSNAP
,
1202 object_name
, 0, 0, NULL
,
1207 rbd_simple_req_cb
, 0, NULL
);
1209 rbd_destroy_ops(ops
);
1213 static void rbd_watch_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1215 struct rbd_device
*rbd_dev
= (struct rbd_device
*)data
;
1222 dout("rbd_watch_cb %s notify_id=%lld opcode=%d\n",
1223 rbd_dev
->header_name
, notify_id
, (int) opcode
);
1224 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1225 rc
= __rbd_refresh_header(rbd_dev
);
1226 hver
= rbd_dev
->header
.obj_version
;
1227 mutex_unlock(&ctl_mutex
);
1229 pr_warning(RBD_DRV_NAME
"%d got notification but failed to "
1230 " update snaps: %d\n", rbd_dev
->major
, rc
);
1232 rbd_req_sync_notify_ack(rbd_dev
, hver
, notify_id
, rbd_dev
->header_name
);
1236 * Request sync osd watch
1238 static int rbd_req_sync_watch(struct rbd_device
*rbd_dev
,
1239 const char *object_name
,
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(ver
);
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 object_name
, 0, 0, NULL
,
1264 &rbd_dev
->watch_request
, NULL
);
1269 rbd_destroy_ops(ops
);
1273 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1274 rbd_dev
->watch_event
= NULL
;
1276 rbd_destroy_ops(ops
);
1281 * Request sync osd unwatch
1283 static int rbd_req_sync_unwatch(struct rbd_device
*rbd_dev
,
1284 const char *object_name
)
1286 struct ceph_osd_req_op
*ops
;
1288 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_WATCH
, 0);
1292 ops
[0].watch
.ver
= 0;
1293 ops
[0].watch
.cookie
= cpu_to_le64(rbd_dev
->watch_event
->cookie
);
1294 ops
[0].watch
.flag
= 0;
1296 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1299 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1301 object_name
, 0, 0, NULL
, NULL
, NULL
);
1303 rbd_destroy_ops(ops
);
1304 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1305 rbd_dev
->watch_event
= NULL
;
1309 struct rbd_notify_info
{
1310 struct rbd_device
*rbd_dev
;
1313 static void rbd_notify_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1315 struct rbd_device
*rbd_dev
= (struct rbd_device
*)data
;
1319 dout("rbd_notify_cb %s notify_id=%lld opcode=%d\n",
1320 rbd_dev
->header_name
,
1321 notify_id
, (int)opcode
);
1325 * Request sync osd notify
1327 static int rbd_req_sync_notify(struct rbd_device
*rbd_dev
,
1328 const char *object_name
)
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 object_name
, 0, 0, NULL
, NULL
, NULL
);
1363 ret
= ceph_osdc_wait_event(event
, CEPH_OSD_TIMEOUT_DEFAULT
);
1364 dout("ceph_osdc_wait_event returned %d\n", ret
);
1365 rbd_destroy_ops(ops
);
1369 ceph_osdc_cancel_event(event
);
1371 rbd_destroy_ops(ops
);
1376 * Request sync osd read
1378 static int rbd_req_sync_exec(struct rbd_device
*rbd_dev
,
1379 const char *object_name
,
1380 const char *class_name
,
1381 const char *method_name
,
1386 struct ceph_osd_req_op
*ops
;
1387 int class_name_len
= strlen(class_name
);
1388 int method_name_len
= strlen(method_name
);
1389 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_CALL
,
1390 class_name_len
+ method_name_len
+ len
);
1394 ops
[0].cls
.class_name
= class_name
;
1395 ops
[0].cls
.class_len
= (__u8
) class_name_len
;
1396 ops
[0].cls
.method_name
= method_name
;
1397 ops
[0].cls
.method_len
= (__u8
) method_name_len
;
1398 ops
[0].cls
.argc
= 0;
1399 ops
[0].cls
.indata
= data
;
1400 ops
[0].cls
.indata_len
= len
;
1402 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1405 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1407 object_name
, 0, 0, NULL
, NULL
, ver
);
1409 rbd_destroy_ops(ops
);
1411 dout("cls_exec returned %d\n", ret
);
1415 static struct rbd_req_coll
*rbd_alloc_coll(int num_reqs
)
1417 struct rbd_req_coll
*coll
=
1418 kzalloc(sizeof(struct rbd_req_coll
) +
1419 sizeof(struct rbd_req_status
) * num_reqs
,
1424 coll
->total
= num_reqs
;
1425 kref_init(&coll
->kref
);
1430 * block device queue callback
1432 static void rbd_rq_fn(struct request_queue
*q
)
1434 struct rbd_device
*rbd_dev
= q
->queuedata
;
1436 struct bio_pair
*bp
= NULL
;
1438 while ((rq
= blk_fetch_request(q
))) {
1440 struct bio
*rq_bio
, *next_bio
= NULL
;
1442 int size
, op_size
= 0;
1444 int num_segs
, cur_seg
= 0;
1445 struct rbd_req_coll
*coll
;
1446 struct ceph_snap_context
*snapc
;
1448 /* peek at request from block layer */
1452 dout("fetched request\n");
1454 /* filter out block requests we don't understand */
1455 if ((rq
->cmd_type
!= REQ_TYPE_FS
)) {
1456 __blk_end_request_all(rq
, 0);
1460 /* deduce our operation (read, write) */
1461 do_write
= (rq_data_dir(rq
) == WRITE
);
1463 size
= blk_rq_bytes(rq
);
1464 ofs
= blk_rq_pos(rq
) * SECTOR_SIZE
;
1466 if (do_write
&& rbd_dev
->read_only
) {
1467 __blk_end_request_all(rq
, -EROFS
);
1471 spin_unlock_irq(q
->queue_lock
);
1473 down_read(&rbd_dev
->header_rwsem
);
1475 if (rbd_dev
->snap_id
!= CEPH_NOSNAP
&& !rbd_dev
->snap_exists
) {
1476 up_read(&rbd_dev
->header_rwsem
);
1477 dout("request for non-existent snapshot");
1478 spin_lock_irq(q
->queue_lock
);
1479 __blk_end_request_all(rq
, -ENXIO
);
1483 snapc
= ceph_get_snap_context(rbd_dev
->header
.snapc
);
1485 up_read(&rbd_dev
->header_rwsem
);
1487 dout("%s 0x%x bytes at 0x%llx\n",
1488 do_write
? "write" : "read",
1489 size
, blk_rq_pos(rq
) * SECTOR_SIZE
);
1491 num_segs
= rbd_get_num_segments(&rbd_dev
->header
, ofs
, size
);
1492 coll
= rbd_alloc_coll(num_segs
);
1494 spin_lock_irq(q
->queue_lock
);
1495 __blk_end_request_all(rq
, -ENOMEM
);
1496 ceph_put_snap_context(snapc
);
1501 /* a bio clone to be passed down to OSD req */
1502 dout("rq->bio->bi_vcnt=%d\n", rq
->bio
->bi_vcnt
);
1503 op_size
= rbd_get_segment(&rbd_dev
->header
,
1504 rbd_dev
->header
.object_prefix
,
1507 kref_get(&coll
->kref
);
1508 bio
= bio_chain_clone(&rq_bio
, &next_bio
, &bp
,
1509 op_size
, GFP_ATOMIC
);
1511 rbd_coll_end_req_index(rq
, coll
, cur_seg
,
1517 /* init OSD command: write or read */
1519 rbd_req_write(rq
, rbd_dev
,
1525 rbd_req_read(rq
, rbd_dev
,
1538 kref_put(&coll
->kref
, rbd_coll_release
);
1541 bio_pair_release(bp
);
1542 spin_lock_irq(q
->queue_lock
);
1544 ceph_put_snap_context(snapc
);
1549 * a queue callback. Makes sure that we don't create a bio that spans across
1550 * multiple osd objects. One exception would be with a single page bios,
1551 * which we handle later at bio_chain_clone
1553 static int rbd_merge_bvec(struct request_queue
*q
, struct bvec_merge_data
*bmd
,
1554 struct bio_vec
*bvec
)
1556 struct rbd_device
*rbd_dev
= q
->queuedata
;
1557 unsigned int chunk_sectors
;
1559 unsigned int bio_sectors
;
1562 chunk_sectors
= 1 << (rbd_dev
->header
.obj_order
- SECTOR_SHIFT
);
1563 sector
= bmd
->bi_sector
+ get_start_sect(bmd
->bi_bdev
);
1564 bio_sectors
= bmd
->bi_size
>> SECTOR_SHIFT
;
1566 max
= (chunk_sectors
- ((sector
& (chunk_sectors
- 1))
1567 + bio_sectors
)) << SECTOR_SHIFT
;
1569 max
= 0; /* bio_add cannot handle a negative return */
1570 if (max
<= bvec
->bv_len
&& bio_sectors
== 0)
1571 return bvec
->bv_len
;
1575 static void rbd_free_disk(struct rbd_device
*rbd_dev
)
1577 struct gendisk
*disk
= rbd_dev
->disk
;
1582 rbd_header_free(&rbd_dev
->header
);
1584 if (disk
->flags
& GENHD_FL_UP
)
1587 blk_cleanup_queue(disk
->queue
);
1592 * reload the ondisk the header
1594 static int rbd_read_header(struct rbd_device
*rbd_dev
,
1595 struct rbd_image_header
*header
)
1598 struct rbd_image_header_ondisk
*dh
;
1604 * First reads the fixed-size header to determine the number
1605 * of snapshots, then re-reads it, along with all snapshot
1606 * records as well as their stored names.
1610 dh
= kmalloc(len
, GFP_KERNEL
);
1614 rc
= rbd_req_sync_read(rbd_dev
,
1616 rbd_dev
->header_name
,
1622 rc
= rbd_header_from_disk(header
, dh
, snap_count
, GFP_KERNEL
);
1625 pr_warning("unrecognized header format"
1627 rbd_dev
->image_name
);
1631 if (snap_count
== header
->total_snaps
)
1634 snap_count
= header
->total_snaps
;
1635 len
= sizeof (*dh
) +
1636 snap_count
* sizeof(struct rbd_image_snap_ondisk
) +
1637 header
->snap_names_len
;
1639 rbd_header_free(header
);
1642 header
->obj_version
= ver
;
1652 static int rbd_header_add_snap(struct rbd_device
*rbd_dev
,
1653 const char *snap_name
,
1656 int name_len
= strlen(snap_name
);
1661 struct ceph_mon_client
*monc
;
1663 /* we should create a snapshot only if we're pointing at the head */
1664 if (rbd_dev
->snap_id
!= CEPH_NOSNAP
)
1667 monc
= &rbd_dev
->rbd_client
->client
->monc
;
1668 ret
= ceph_monc_create_snapid(monc
, rbd_dev
->pool_id
, &new_snapid
);
1669 dout("created snapid=%lld\n", new_snapid
);
1673 data
= kmalloc(name_len
+ 16, gfp_flags
);
1678 e
= data
+ name_len
+ 16;
1680 ceph_encode_string_safe(&p
, e
, snap_name
, name_len
, bad
);
1681 ceph_encode_64_safe(&p
, e
, new_snapid
, bad
);
1683 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
1685 data
, p
- data
, &ver
);
1689 return ret
< 0 ? ret
: 0;
1694 static void __rbd_remove_all_snaps(struct rbd_device
*rbd_dev
)
1696 struct rbd_snap
*snap
;
1698 while (!list_empty(&rbd_dev
->snaps
)) {
1699 snap
= list_first_entry(&rbd_dev
->snaps
, struct rbd_snap
, node
);
1700 __rbd_remove_snap_dev(rbd_dev
, snap
);
1705 * only read the first part of the ondisk header, without the snaps info
1707 static int __rbd_refresh_header(struct rbd_device
*rbd_dev
)
1710 struct rbd_image_header h
;
1712 ret
= rbd_read_header(rbd_dev
, &h
);
1716 down_write(&rbd_dev
->header_rwsem
);
1719 if (rbd_dev
->snap_id
== CEPH_NOSNAP
) {
1720 sector_t size
= (sector_t
) h
.image_size
/ SECTOR_SIZE
;
1722 dout("setting size to %llu sectors", (unsigned long long) size
);
1723 set_capacity(rbd_dev
->disk
, size
);
1726 /* rbd_dev->header.object_prefix shouldn't change */
1727 kfree(rbd_dev
->header
.snap_sizes
);
1728 kfree(rbd_dev
->header
.snap_names
);
1729 /* osd requests may still refer to snapc */
1730 ceph_put_snap_context(rbd_dev
->header
.snapc
);
1732 rbd_dev
->header
.obj_version
= h
.obj_version
;
1733 rbd_dev
->header
.image_size
= h
.image_size
;
1734 rbd_dev
->header
.total_snaps
= h
.total_snaps
;
1735 rbd_dev
->header
.snapc
= h
.snapc
;
1736 rbd_dev
->header
.snap_names
= h
.snap_names
;
1737 rbd_dev
->header
.snap_names_len
= h
.snap_names_len
;
1738 rbd_dev
->header
.snap_sizes
= h
.snap_sizes
;
1739 /* Free the extra copy of the object prefix */
1740 WARN_ON(strcmp(rbd_dev
->header
.object_prefix
, h
.object_prefix
));
1741 kfree(h
.object_prefix
);
1743 ret
= __rbd_init_snaps_header(rbd_dev
);
1745 up_write(&rbd_dev
->header_rwsem
);
1750 static int rbd_init_disk(struct rbd_device
*rbd_dev
)
1752 struct gendisk
*disk
;
1753 struct request_queue
*q
;
1758 /* contact OSD, request size info about the object being mapped */
1759 rc
= rbd_read_header(rbd_dev
, &rbd_dev
->header
);
1763 /* no need to lock here, as rbd_dev is not registered yet */
1764 rc
= __rbd_init_snaps_header(rbd_dev
);
1768 rc
= rbd_header_set_snap(rbd_dev
, &total_size
);
1772 /* create gendisk info */
1774 disk
= alloc_disk(RBD_MINORS_PER_MAJOR
);
1778 snprintf(disk
->disk_name
, sizeof(disk
->disk_name
), RBD_DRV_NAME
"%d",
1780 disk
->major
= rbd_dev
->major
;
1781 disk
->first_minor
= 0;
1782 disk
->fops
= &rbd_bd_ops
;
1783 disk
->private_data
= rbd_dev
;
1787 q
= blk_init_queue(rbd_rq_fn
, &rbd_dev
->lock
);
1791 /* We use the default size, but let's be explicit about it. */
1792 blk_queue_physical_block_size(q
, SECTOR_SIZE
);
1794 /* set io sizes to object size */
1795 segment_size
= rbd_obj_bytes(&rbd_dev
->header
);
1796 blk_queue_max_hw_sectors(q
, segment_size
/ SECTOR_SIZE
);
1797 blk_queue_max_segment_size(q
, segment_size
);
1798 blk_queue_io_min(q
, segment_size
);
1799 blk_queue_io_opt(q
, segment_size
);
1801 blk_queue_merge_bvec(q
, rbd_merge_bvec
);
1804 q
->queuedata
= rbd_dev
;
1806 rbd_dev
->disk
= disk
;
1809 /* finally, announce the disk to the world */
1810 set_capacity(disk
, total_size
/ SECTOR_SIZE
);
1813 pr_info("%s: added with size 0x%llx\n",
1814 disk
->disk_name
, (unsigned long long)total_size
);
1827 static struct rbd_device
*dev_to_rbd_dev(struct device
*dev
)
1829 return container_of(dev
, struct rbd_device
, dev
);
1832 static ssize_t
rbd_size_show(struct device
*dev
,
1833 struct device_attribute
*attr
, char *buf
)
1835 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1838 down_read(&rbd_dev
->header_rwsem
);
1839 size
= get_capacity(rbd_dev
->disk
);
1840 up_read(&rbd_dev
->header_rwsem
);
1842 return sprintf(buf
, "%llu\n", (unsigned long long) size
* SECTOR_SIZE
);
1845 static ssize_t
rbd_major_show(struct device
*dev
,
1846 struct device_attribute
*attr
, char *buf
)
1848 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1850 return sprintf(buf
, "%d\n", rbd_dev
->major
);
1853 static ssize_t
rbd_client_id_show(struct device
*dev
,
1854 struct device_attribute
*attr
, char *buf
)
1856 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1858 return sprintf(buf
, "client%lld\n",
1859 ceph_client_id(rbd_dev
->rbd_client
->client
));
1862 static ssize_t
rbd_pool_show(struct device
*dev
,
1863 struct device_attribute
*attr
, char *buf
)
1865 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1867 return sprintf(buf
, "%s\n", rbd_dev
->pool_name
);
1870 static ssize_t
rbd_pool_id_show(struct device
*dev
,
1871 struct device_attribute
*attr
, char *buf
)
1873 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1875 return sprintf(buf
, "%d\n", rbd_dev
->pool_id
);
1878 static ssize_t
rbd_name_show(struct device
*dev
,
1879 struct device_attribute
*attr
, char *buf
)
1881 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1883 return sprintf(buf
, "%s\n", rbd_dev
->image_name
);
1886 static ssize_t
rbd_snap_show(struct device
*dev
,
1887 struct device_attribute
*attr
,
1890 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1892 return sprintf(buf
, "%s\n", rbd_dev
->snap_name
);
1895 static ssize_t
rbd_image_refresh(struct device
*dev
,
1896 struct device_attribute
*attr
,
1900 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1904 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1906 rc
= __rbd_refresh_header(rbd_dev
);
1910 mutex_unlock(&ctl_mutex
);
1914 static DEVICE_ATTR(size
, S_IRUGO
, rbd_size_show
, NULL
);
1915 static DEVICE_ATTR(major
, S_IRUGO
, rbd_major_show
, NULL
);
1916 static DEVICE_ATTR(client_id
, S_IRUGO
, rbd_client_id_show
, NULL
);
1917 static DEVICE_ATTR(pool
, S_IRUGO
, rbd_pool_show
, NULL
);
1918 static DEVICE_ATTR(pool_id
, S_IRUGO
, rbd_pool_id_show
, NULL
);
1919 static DEVICE_ATTR(name
, S_IRUGO
, rbd_name_show
, NULL
);
1920 static DEVICE_ATTR(refresh
, S_IWUSR
, NULL
, rbd_image_refresh
);
1921 static DEVICE_ATTR(current_snap
, S_IRUGO
, rbd_snap_show
, NULL
);
1922 static DEVICE_ATTR(create_snap
, S_IWUSR
, NULL
, rbd_snap_add
);
1924 static struct attribute
*rbd_attrs
[] = {
1925 &dev_attr_size
.attr
,
1926 &dev_attr_major
.attr
,
1927 &dev_attr_client_id
.attr
,
1928 &dev_attr_pool
.attr
,
1929 &dev_attr_pool_id
.attr
,
1930 &dev_attr_name
.attr
,
1931 &dev_attr_current_snap
.attr
,
1932 &dev_attr_refresh
.attr
,
1933 &dev_attr_create_snap
.attr
,
1937 static struct attribute_group rbd_attr_group
= {
1941 static const struct attribute_group
*rbd_attr_groups
[] = {
1946 static void rbd_sysfs_dev_release(struct device
*dev
)
1950 static struct device_type rbd_device_type
= {
1952 .groups
= rbd_attr_groups
,
1953 .release
= rbd_sysfs_dev_release
,
1961 static ssize_t
rbd_snap_size_show(struct device
*dev
,
1962 struct device_attribute
*attr
,
1965 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1967 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->size
);
1970 static ssize_t
rbd_snap_id_show(struct device
*dev
,
1971 struct device_attribute
*attr
,
1974 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1976 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->id
);
1979 static DEVICE_ATTR(snap_size
, S_IRUGO
, rbd_snap_size_show
, NULL
);
1980 static DEVICE_ATTR(snap_id
, S_IRUGO
, rbd_snap_id_show
, NULL
);
1982 static struct attribute
*rbd_snap_attrs
[] = {
1983 &dev_attr_snap_size
.attr
,
1984 &dev_attr_snap_id
.attr
,
1988 static struct attribute_group rbd_snap_attr_group
= {
1989 .attrs
= rbd_snap_attrs
,
1992 static void rbd_snap_dev_release(struct device
*dev
)
1994 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1999 static const struct attribute_group
*rbd_snap_attr_groups
[] = {
2000 &rbd_snap_attr_group
,
2004 static struct device_type rbd_snap_device_type
= {
2005 .groups
= rbd_snap_attr_groups
,
2006 .release
= rbd_snap_dev_release
,
2009 static void __rbd_remove_snap_dev(struct rbd_device
*rbd_dev
,
2010 struct rbd_snap
*snap
)
2012 list_del(&snap
->node
);
2013 device_unregister(&snap
->dev
);
2016 static int rbd_register_snap_dev(struct rbd_device
*rbd_dev
,
2017 struct rbd_snap
*snap
,
2018 struct device
*parent
)
2020 struct device
*dev
= &snap
->dev
;
2023 dev
->type
= &rbd_snap_device_type
;
2024 dev
->parent
= parent
;
2025 dev
->release
= rbd_snap_dev_release
;
2026 dev_set_name(dev
, "snap_%s", snap
->name
);
2027 ret
= device_register(dev
);
2032 static int __rbd_add_snap_dev(struct rbd_device
*rbd_dev
,
2033 int i
, const char *name
,
2034 struct rbd_snap
**snapp
)
2037 struct rbd_snap
*snap
= kzalloc(sizeof(*snap
), GFP_KERNEL
);
2040 snap
->name
= kstrdup(name
, GFP_KERNEL
);
2041 snap
->size
= rbd_dev
->header
.snap_sizes
[i
];
2042 snap
->id
= rbd_dev
->header
.snapc
->snaps
[i
];
2043 if (device_is_registered(&rbd_dev
->dev
)) {
2044 ret
= rbd_register_snap_dev(rbd_dev
, snap
,
2058 * search for the previous snap in a null delimited string list
2060 const char *rbd_prev_snap_name(const char *name
, const char *start
)
2062 if (name
< start
+ 2)
2075 * compare the old list of snapshots that we have to what's in the header
2076 * and update it accordingly. Note that the header holds the snapshots
2077 * in a reverse order (from newest to oldest) and we need to go from
2078 * older to new so that we don't get a duplicate snap name when
2079 * doing the process (e.g., removed snapshot and recreated a new
2080 * one with the same name.
2082 static int __rbd_init_snaps_header(struct rbd_device
*rbd_dev
)
2084 const char *name
, *first_name
;
2085 int i
= rbd_dev
->header
.total_snaps
;
2086 struct rbd_snap
*snap
, *old_snap
= NULL
;
2088 struct list_head
*p
, *n
;
2090 first_name
= rbd_dev
->header
.snap_names
;
2091 name
= first_name
+ rbd_dev
->header
.snap_names_len
;
2093 list_for_each_prev_safe(p
, n
, &rbd_dev
->snaps
) {
2096 old_snap
= list_entry(p
, struct rbd_snap
, node
);
2099 cur_id
= rbd_dev
->header
.snapc
->snaps
[i
- 1];
2101 if (!i
|| old_snap
->id
< cur_id
) {
2103 * old_snap->id was skipped, thus was
2104 * removed. If this rbd_dev is mapped to
2105 * the removed snapshot, record that it no
2106 * longer exists, to prevent further I/O.
2108 if (rbd_dev
->snap_id
== old_snap
->id
)
2109 rbd_dev
->snap_exists
= false;
2110 __rbd_remove_snap_dev(rbd_dev
, old_snap
);
2113 if (old_snap
->id
== cur_id
) {
2114 /* we have this snapshot already */
2116 name
= rbd_prev_snap_name(name
, first_name
);
2120 i
--, name
= rbd_prev_snap_name(name
, first_name
)) {
2125 cur_id
= rbd_dev
->header
.snapc
->snaps
[i
];
2126 /* snapshot removal? handle it above */
2127 if (cur_id
>= old_snap
->id
)
2129 /* a new snapshot */
2130 ret
= __rbd_add_snap_dev(rbd_dev
, i
- 1, name
, &snap
);
2134 /* note that we add it backward so using n and not p */
2135 list_add(&snap
->node
, n
);
2139 /* we're done going over the old snap list, just add what's left */
2140 for (; i
> 0; i
--) {
2141 name
= rbd_prev_snap_name(name
, first_name
);
2146 ret
= __rbd_add_snap_dev(rbd_dev
, i
- 1, name
, &snap
);
2149 list_add(&snap
->node
, &rbd_dev
->snaps
);
2155 static int rbd_bus_add_dev(struct rbd_device
*rbd_dev
)
2159 struct rbd_snap
*snap
;
2161 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2162 dev
= &rbd_dev
->dev
;
2164 dev
->bus
= &rbd_bus_type
;
2165 dev
->type
= &rbd_device_type
;
2166 dev
->parent
= &rbd_root_dev
;
2167 dev
->release
= rbd_dev_release
;
2168 dev_set_name(dev
, "%d", rbd_dev
->id
);
2169 ret
= device_register(dev
);
2173 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
2174 ret
= rbd_register_snap_dev(rbd_dev
, snap
,
2180 mutex_unlock(&ctl_mutex
);
2184 static void rbd_bus_del_dev(struct rbd_device
*rbd_dev
)
2186 device_unregister(&rbd_dev
->dev
);
2189 static int rbd_init_watch_dev(struct rbd_device
*rbd_dev
)
2194 ret
= rbd_req_sync_watch(rbd_dev
, rbd_dev
->header_name
,
2195 rbd_dev
->header
.obj_version
);
2196 if (ret
== -ERANGE
) {
2197 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2198 rc
= __rbd_refresh_header(rbd_dev
);
2199 mutex_unlock(&ctl_mutex
);
2203 } while (ret
== -ERANGE
);
2208 static atomic64_t rbd_id_max
= ATOMIC64_INIT(0);
2211 * Get a unique rbd identifier for the given new rbd_dev, and add
2212 * the rbd_dev to the global list. The minimum rbd id is 1.
2214 static void rbd_id_get(struct rbd_device
*rbd_dev
)
2216 rbd_dev
->id
= atomic64_inc_return(&rbd_id_max
);
2218 spin_lock(&rbd_dev_list_lock
);
2219 list_add_tail(&rbd_dev
->node
, &rbd_dev_list
);
2220 spin_unlock(&rbd_dev_list_lock
);
2224 * Remove an rbd_dev from the global list, and record that its
2225 * identifier is no longer in use.
2227 static void rbd_id_put(struct rbd_device
*rbd_dev
)
2229 struct list_head
*tmp
;
2230 int rbd_id
= rbd_dev
->id
;
2235 spin_lock(&rbd_dev_list_lock
);
2236 list_del_init(&rbd_dev
->node
);
2239 * If the id being "put" is not the current maximum, there
2240 * is nothing special we need to do.
2242 if (rbd_id
!= atomic64_read(&rbd_id_max
)) {
2243 spin_unlock(&rbd_dev_list_lock
);
2248 * We need to update the current maximum id. Search the
2249 * list to find out what it is. We're more likely to find
2250 * the maximum at the end, so search the list backward.
2253 list_for_each_prev(tmp
, &rbd_dev_list
) {
2254 struct rbd_device
*rbd_dev
;
2256 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
2257 if (rbd_id
> max_id
)
2260 spin_unlock(&rbd_dev_list_lock
);
2263 * The max id could have been updated by rbd_id_get(), in
2264 * which case it now accurately reflects the new maximum.
2265 * Be careful not to overwrite the maximum value in that
2268 atomic64_cmpxchg(&rbd_id_max
, rbd_id
, max_id
);
2272 * Skips over white space at *buf, and updates *buf to point to the
2273 * first found non-space character (if any). Returns the length of
2274 * the token (string of non-white space characters) found. Note
2275 * that *buf must be terminated with '\0'.
2277 static inline size_t next_token(const char **buf
)
2280 * These are the characters that produce nonzero for
2281 * isspace() in the "C" and "POSIX" locales.
2283 const char *spaces
= " \f\n\r\t\v";
2285 *buf
+= strspn(*buf
, spaces
); /* Find start of token */
2287 return strcspn(*buf
, spaces
); /* Return token length */
2291 * Finds the next token in *buf, and if the provided token buffer is
2292 * big enough, copies the found token into it. The result, if
2293 * copied, is guaranteed to be terminated with '\0'. Note that *buf
2294 * must be terminated with '\0' on entry.
2296 * Returns the length of the token found (not including the '\0').
2297 * Return value will be 0 if no token is found, and it will be >=
2298 * token_size if the token would not fit.
2300 * The *buf pointer will be updated to point beyond the end of the
2301 * found token. Note that this occurs even if the token buffer is
2302 * too small to hold it.
2304 static inline size_t copy_token(const char **buf
,
2310 len
= next_token(buf
);
2311 if (len
< token_size
) {
2312 memcpy(token
, *buf
, len
);
2313 *(token
+ len
) = '\0';
2321 * Finds the next token in *buf, dynamically allocates a buffer big
2322 * enough to hold a copy of it, and copies the token into the new
2323 * buffer. The copy is guaranteed to be terminated with '\0'. Note
2324 * that a duplicate buffer is created even for a zero-length token.
2326 * Returns a pointer to the newly-allocated duplicate, or a null
2327 * pointer if memory for the duplicate was not available. If
2328 * the lenp argument is a non-null pointer, the length of the token
2329 * (not including the '\0') is returned in *lenp.
2331 * If successful, the *buf pointer will be updated to point beyond
2332 * the end of the found token.
2334 * Note: uses GFP_KERNEL for allocation.
2336 static inline char *dup_token(const char **buf
, size_t *lenp
)
2341 len
= next_token(buf
);
2342 dup
= kmalloc(len
+ 1, GFP_KERNEL
);
2346 memcpy(dup
, *buf
, len
);
2347 *(dup
+ len
) = '\0';
2357 * This fills in the pool_name, image_name, image_name_len, snap_name,
2358 * rbd_dev, rbd_md_name, and name fields of the given rbd_dev, based
2359 * on the list of monitor addresses and other options provided via
2362 * Note: rbd_dev is assumed to have been initially zero-filled.
2364 static int rbd_add_parse_args(struct rbd_device
*rbd_dev
,
2366 const char **mon_addrs
,
2367 size_t *mon_addrs_size
,
2369 size_t options_size
)
2374 /* The first four tokens are required */
2376 len
= next_token(&buf
);
2379 *mon_addrs_size
= len
+ 1;
2384 len
= copy_token(&buf
, options
, options_size
);
2385 if (!len
|| len
>= options_size
)
2389 rbd_dev
->pool_name
= dup_token(&buf
, NULL
);
2390 if (!rbd_dev
->pool_name
)
2393 rbd_dev
->image_name
= dup_token(&buf
, &rbd_dev
->image_name_len
);
2394 if (!rbd_dev
->image_name
)
2397 /* Create the name of the header object */
2399 rbd_dev
->header_name
= kmalloc(rbd_dev
->image_name_len
2400 + sizeof (RBD_SUFFIX
),
2402 if (!rbd_dev
->header_name
)
2404 sprintf(rbd_dev
->header_name
, "%s%s", rbd_dev
->image_name
, RBD_SUFFIX
);
2407 * The snapshot name is optional. If none is is supplied,
2408 * we use the default value.
2410 rbd_dev
->snap_name
= dup_token(&buf
, &len
);
2411 if (!rbd_dev
->snap_name
)
2414 /* Replace the empty name with the default */
2415 kfree(rbd_dev
->snap_name
);
2417 = kmalloc(sizeof (RBD_SNAP_HEAD_NAME
), GFP_KERNEL
);
2418 if (!rbd_dev
->snap_name
)
2421 memcpy(rbd_dev
->snap_name
, RBD_SNAP_HEAD_NAME
,
2422 sizeof (RBD_SNAP_HEAD_NAME
));
2428 kfree(rbd_dev
->header_name
);
2429 kfree(rbd_dev
->image_name
);
2430 kfree(rbd_dev
->pool_name
);
2431 rbd_dev
->pool_name
= NULL
;
2436 static ssize_t
rbd_add(struct bus_type
*bus
,
2441 struct rbd_device
*rbd_dev
= NULL
;
2442 const char *mon_addrs
= NULL
;
2443 size_t mon_addrs_size
= 0;
2444 struct ceph_osd_client
*osdc
;
2447 if (!try_module_get(THIS_MODULE
))
2450 options
= kmalloc(count
, GFP_KERNEL
);
2453 rbd_dev
= kzalloc(sizeof(*rbd_dev
), GFP_KERNEL
);
2457 /* static rbd_device initialization */
2458 spin_lock_init(&rbd_dev
->lock
);
2459 INIT_LIST_HEAD(&rbd_dev
->node
);
2460 INIT_LIST_HEAD(&rbd_dev
->snaps
);
2461 init_rwsem(&rbd_dev
->header_rwsem
);
2463 init_rwsem(&rbd_dev
->header_rwsem
);
2465 /* generate unique id: find highest unique id, add one */
2466 rbd_id_get(rbd_dev
);
2468 /* Fill in the device name, now that we have its id. */
2469 BUILD_BUG_ON(DEV_NAME_LEN
2470 < sizeof (RBD_DRV_NAME
) + MAX_INT_FORMAT_WIDTH
);
2471 sprintf(rbd_dev
->name
, "%s%d", RBD_DRV_NAME
, rbd_dev
->id
);
2473 /* parse add command */
2474 rc
= rbd_add_parse_args(rbd_dev
, buf
, &mon_addrs
, &mon_addrs_size
,
2479 rbd_dev
->rbd_client
= rbd_get_client(mon_addrs
, mon_addrs_size
- 1,
2481 if (IS_ERR(rbd_dev
->rbd_client
)) {
2482 rc
= PTR_ERR(rbd_dev
->rbd_client
);
2487 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
2488 rc
= ceph_pg_poolid_by_name(osdc
->osdmap
, rbd_dev
->pool_name
);
2490 goto err_out_client
;
2491 rbd_dev
->pool_id
= rc
;
2493 /* register our block device */
2494 rc
= register_blkdev(0, rbd_dev
->name
);
2496 goto err_out_client
;
2497 rbd_dev
->major
= rc
;
2499 rc
= rbd_bus_add_dev(rbd_dev
);
2501 goto err_out_blkdev
;
2504 * At this point cleanup in the event of an error is the job
2505 * of the sysfs code (initiated by rbd_bus_del_dev()).
2507 * Set up and announce blkdev mapping.
2509 rc
= rbd_init_disk(rbd_dev
);
2513 rc
= rbd_init_watch_dev(rbd_dev
);
2520 /* this will also clean up rest of rbd_dev stuff */
2522 rbd_bus_del_dev(rbd_dev
);
2527 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
2529 rbd_put_client(rbd_dev
);
2531 if (rbd_dev
->pool_name
) {
2532 kfree(rbd_dev
->snap_name
);
2533 kfree(rbd_dev
->header_name
);
2534 kfree(rbd_dev
->image_name
);
2535 kfree(rbd_dev
->pool_name
);
2537 rbd_id_put(rbd_dev
);
2542 dout("Error adding device %s\n", buf
);
2543 module_put(THIS_MODULE
);
2545 return (ssize_t
) rc
;
2548 static struct rbd_device
*__rbd_get_dev(unsigned long id
)
2550 struct list_head
*tmp
;
2551 struct rbd_device
*rbd_dev
;
2553 spin_lock(&rbd_dev_list_lock
);
2554 list_for_each(tmp
, &rbd_dev_list
) {
2555 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
2556 if (rbd_dev
->id
== id
) {
2557 spin_unlock(&rbd_dev_list_lock
);
2561 spin_unlock(&rbd_dev_list_lock
);
2565 static void rbd_dev_release(struct device
*dev
)
2567 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2569 if (rbd_dev
->watch_request
) {
2570 struct ceph_client
*client
= rbd_dev
->rbd_client
->client
;
2572 ceph_osdc_unregister_linger_request(&client
->osdc
,
2573 rbd_dev
->watch_request
);
2575 if (rbd_dev
->watch_event
)
2576 rbd_req_sync_unwatch(rbd_dev
, rbd_dev
->header_name
);
2578 rbd_put_client(rbd_dev
);
2580 /* clean up and free blkdev */
2581 rbd_free_disk(rbd_dev
);
2582 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
2584 /* done with the id, and with the rbd_dev */
2585 kfree(rbd_dev
->snap_name
);
2586 kfree(rbd_dev
->header_name
);
2587 kfree(rbd_dev
->pool_name
);
2588 kfree(rbd_dev
->image_name
);
2589 rbd_id_put(rbd_dev
);
2592 /* release module ref */
2593 module_put(THIS_MODULE
);
2596 static ssize_t
rbd_remove(struct bus_type
*bus
,
2600 struct rbd_device
*rbd_dev
= NULL
;
2605 rc
= strict_strtoul(buf
, 10, &ul
);
2609 /* convert to int; abort if we lost anything in the conversion */
2610 target_id
= (int) ul
;
2611 if (target_id
!= ul
)
2614 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2616 rbd_dev
= __rbd_get_dev(target_id
);
2622 __rbd_remove_all_snaps(rbd_dev
);
2623 rbd_bus_del_dev(rbd_dev
);
2626 mutex_unlock(&ctl_mutex
);
2630 static ssize_t
rbd_snap_add(struct device
*dev
,
2631 struct device_attribute
*attr
,
2635 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2637 char *name
= kmalloc(count
+ 1, GFP_KERNEL
);
2641 snprintf(name
, count
, "%s", buf
);
2643 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2645 ret
= rbd_header_add_snap(rbd_dev
,
2650 ret
= __rbd_refresh_header(rbd_dev
);
2654 /* shouldn't hold ctl_mutex when notifying.. notify might
2655 trigger a watch callback that would need to get that mutex */
2656 mutex_unlock(&ctl_mutex
);
2658 /* make a best effort, don't error if failed */
2659 rbd_req_sync_notify(rbd_dev
, rbd_dev
->header_name
);
2666 mutex_unlock(&ctl_mutex
);
2672 * create control files in sysfs
2675 static int rbd_sysfs_init(void)
2679 ret
= device_register(&rbd_root_dev
);
2683 ret
= bus_register(&rbd_bus_type
);
2685 device_unregister(&rbd_root_dev
);
2690 static void rbd_sysfs_cleanup(void)
2692 bus_unregister(&rbd_bus_type
);
2693 device_unregister(&rbd_root_dev
);
2696 int __init
rbd_init(void)
2700 rc
= rbd_sysfs_init();
2703 pr_info("loaded " RBD_DRV_NAME_LONG
"\n");
2707 void __exit
rbd_exit(void)
2709 rbd_sysfs_cleanup();
2712 module_init(rbd_init
);
2713 module_exit(rbd_exit
);
2715 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
2716 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
2717 MODULE_DESCRIPTION("rados block device");
2719 /* following authorship retained from original osdblk.c */
2720 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
2722 MODULE_LICENSE("GPL");