2 rbd.c -- Export ceph rados objects as a Linux block device
5 based on drivers/block/osdblk.c:
7 Copyright 2009 Red Hat, Inc.
9 This program is free software; you can redistribute it and/or modify
10 it under the terms of the GNU General Public License as published by
11 the Free Software Foundation.
13 This program is distributed in the hope that it will be useful,
14 but WITHOUT ANY WARRANTY; without even the implied warranty of
15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 GNU General Public License for more details.
18 You should have received a copy of the GNU General Public License
19 along with this program; see the file COPYING. If not, write to
20 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
24 For usage instructions, please refer to:
26 Documentation/ABI/testing/sysfs-bus-rbd
30 #include <linux/ceph/libceph.h>
31 #include <linux/ceph/osd_client.h>
32 #include <linux/ceph/mon_client.h>
33 #include <linux/ceph/decode.h>
34 #include <linux/parser.h>
36 #include <linux/kernel.h>
37 #include <linux/device.h>
38 #include <linux/module.h>
40 #include <linux/blkdev.h>
42 #include "rbd_types.h"
44 #define RBD_DEBUG /* Activate rbd_assert() calls */
47 * The basic unit of block I/O is a sector. It is interpreted in a
48 * number of contexts in Linux (blk, bio, genhd), but the default is
49 * universally 512 bytes. These symbols are just slightly more
50 * meaningful than the bare numbers they represent.
52 #define SECTOR_SHIFT 9
53 #define SECTOR_SIZE (1ULL << SECTOR_SHIFT)
55 /* It might be useful to have this defined elsewhere too */
57 #define U64_MAX ((u64) (~0ULL))
59 #define RBD_DRV_NAME "rbd"
60 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
62 #define RBD_MINORS_PER_MAJOR 256 /* max minors per blkdev */
64 #define RBD_MAX_SNAP_NAME_LEN 32
65 #define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */
66 #define RBD_MAX_OPT_LEN 1024
68 #define RBD_SNAP_HEAD_NAME "-"
70 #define RBD_IMAGE_ID_LEN_MAX 64
71 #define RBD_OBJ_PREFIX_LEN_MAX 64
75 #define RBD_FEATURE_LAYERING 1
77 /* Features supported by this (client software) implementation. */
79 #define RBD_FEATURES_ALL (0)
82 * An RBD device name will be "rbd#", where the "rbd" comes from
83 * RBD_DRV_NAME above, and # is a unique integer identifier.
84 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
85 * enough to hold all possible device names.
87 #define DEV_NAME_LEN 32
88 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1)
90 #define RBD_READ_ONLY_DEFAULT false
93 * block device image metadata (in-memory version)
95 struct rbd_image_header
{
96 /* These four fields never change for a given rbd image */
103 /* The remaining fields need to be updated occasionally */
105 struct ceph_snap_context
*snapc
;
117 * an instance of the client. multiple devices may share an rbd client.
120 struct ceph_client
*client
;
122 struct list_head node
;
126 * a request completion status
128 struct rbd_req_status
{
135 * a collection of requests
137 struct rbd_req_coll
{
141 struct rbd_req_status status
[0];
145 * a single io request
148 struct request
*rq
; /* blk layer request */
149 struct bio
*bio
; /* cloned bio */
150 struct page
**pages
; /* list of used pages */
153 struct rbd_req_coll
*coll
;
160 struct list_head node
;
178 int dev_id
; /* blkdev unique id */
180 int major
; /* blkdev assigned major */
181 struct gendisk
*disk
; /* blkdev's gendisk and rq */
183 u32 image_format
; /* Either 1 or 2 */
184 struct rbd_options rbd_opts
;
185 struct rbd_client
*rbd_client
;
187 char name
[DEV_NAME_LEN
]; /* blkdev name, e.g. rbd3 */
189 spinlock_t lock
; /* queue lock */
191 struct rbd_image_header header
;
195 size_t image_name_len
;
200 struct ceph_osd_event
*watch_event
;
201 struct ceph_osd_request
*watch_request
;
203 /* protects updating the header */
204 struct rw_semaphore header_rwsem
;
206 struct rbd_mapping mapping
;
208 struct list_head node
;
210 /* list of snapshots */
211 struct list_head snaps
;
217 static DEFINE_MUTEX(ctl_mutex
); /* Serialize open/close/setup/teardown */
219 static LIST_HEAD(rbd_dev_list
); /* devices */
220 static DEFINE_SPINLOCK(rbd_dev_list_lock
);
222 static LIST_HEAD(rbd_client_list
); /* clients */
223 static DEFINE_SPINLOCK(rbd_client_list_lock
);
225 static int rbd_dev_snaps_update(struct rbd_device
*rbd_dev
);
226 static int rbd_dev_snaps_register(struct rbd_device
*rbd_dev
);
228 static void rbd_dev_release(struct device
*dev
);
229 static void __rbd_remove_snap_dev(struct rbd_snap
*snap
);
231 static ssize_t
rbd_add(struct bus_type
*bus
, const char *buf
,
233 static ssize_t
rbd_remove(struct bus_type
*bus
, const char *buf
,
236 static struct bus_attribute rbd_bus_attrs
[] = {
237 __ATTR(add
, S_IWUSR
, NULL
, rbd_add
),
238 __ATTR(remove
, S_IWUSR
, NULL
, rbd_remove
),
242 static struct bus_type rbd_bus_type
= {
244 .bus_attrs
= rbd_bus_attrs
,
247 static void rbd_root_dev_release(struct device
*dev
)
251 static struct device rbd_root_dev
= {
253 .release
= rbd_root_dev_release
,
257 #define rbd_assert(expr) \
258 if (unlikely(!(expr))) { \
259 printk(KERN_ERR "\nAssertion failure in %s() " \
261 "\trbd_assert(%s);\n\n", \
262 __func__, __LINE__, #expr); \
265 #else /* !RBD_DEBUG */
266 # define rbd_assert(expr) ((void) 0)
267 #endif /* !RBD_DEBUG */
269 static struct device
*rbd_get_dev(struct rbd_device
*rbd_dev
)
271 return get_device(&rbd_dev
->dev
);
274 static void rbd_put_dev(struct rbd_device
*rbd_dev
)
276 put_device(&rbd_dev
->dev
);
279 static int rbd_dev_refresh(struct rbd_device
*rbd_dev
, u64
*hver
);
280 static int rbd_dev_v2_refresh(struct rbd_device
*rbd_dev
, u64
*hver
);
282 static int rbd_open(struct block_device
*bdev
, fmode_t mode
)
284 struct rbd_device
*rbd_dev
= bdev
->bd_disk
->private_data
;
286 if ((mode
& FMODE_WRITE
) && rbd_dev
->mapping
.read_only
)
289 rbd_get_dev(rbd_dev
);
290 set_device_ro(bdev
, rbd_dev
->mapping
.read_only
);
295 static int rbd_release(struct gendisk
*disk
, fmode_t mode
)
297 struct rbd_device
*rbd_dev
= disk
->private_data
;
299 rbd_put_dev(rbd_dev
);
304 static const struct block_device_operations rbd_bd_ops
= {
305 .owner
= THIS_MODULE
,
307 .release
= rbd_release
,
311 * Initialize an rbd client instance.
314 static struct rbd_client
*rbd_client_create(struct ceph_options
*ceph_opts
)
316 struct rbd_client
*rbdc
;
319 dout("rbd_client_create\n");
320 rbdc
= kmalloc(sizeof(struct rbd_client
), GFP_KERNEL
);
324 kref_init(&rbdc
->kref
);
325 INIT_LIST_HEAD(&rbdc
->node
);
327 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
329 rbdc
->client
= ceph_create_client(ceph_opts
, rbdc
, 0, 0);
330 if (IS_ERR(rbdc
->client
))
332 ceph_opts
= NULL
; /* Now rbdc->client is responsible for ceph_opts */
334 ret
= ceph_open_session(rbdc
->client
);
338 spin_lock(&rbd_client_list_lock
);
339 list_add_tail(&rbdc
->node
, &rbd_client_list
);
340 spin_unlock(&rbd_client_list_lock
);
342 mutex_unlock(&ctl_mutex
);
344 dout("rbd_client_create created %p\n", rbdc
);
348 ceph_destroy_client(rbdc
->client
);
350 mutex_unlock(&ctl_mutex
);
354 ceph_destroy_options(ceph_opts
);
359 * Find a ceph client with specific addr and configuration. If
360 * found, bump its reference count.
362 static struct rbd_client
*rbd_client_find(struct ceph_options
*ceph_opts
)
364 struct rbd_client
*client_node
;
367 if (ceph_opts
->flags
& CEPH_OPT_NOSHARE
)
370 spin_lock(&rbd_client_list_lock
);
371 list_for_each_entry(client_node
, &rbd_client_list
, node
) {
372 if (!ceph_compare_options(ceph_opts
, client_node
->client
)) {
373 kref_get(&client_node
->kref
);
378 spin_unlock(&rbd_client_list_lock
);
380 return found
? client_node
: NULL
;
390 /* string args above */
393 /* Boolean args above */
397 static match_table_t rbd_opts_tokens
= {
399 /* string args above */
400 {Opt_read_only
, "mapping.read_only"},
401 {Opt_read_only
, "ro"}, /* Alternate spelling */
402 {Opt_read_write
, "read_write"},
403 {Opt_read_write
, "rw"}, /* Alternate spelling */
404 /* Boolean args above */
408 static int parse_rbd_opts_token(char *c
, void *private)
410 struct rbd_options
*rbd_opts
= private;
411 substring_t argstr
[MAX_OPT_ARGS
];
412 int token
, intval
, ret
;
414 token
= match_token(c
, rbd_opts_tokens
, argstr
);
418 if (token
< Opt_last_int
) {
419 ret
= match_int(&argstr
[0], &intval
);
421 pr_err("bad mount option arg (not int) "
425 dout("got int token %d val %d\n", token
, intval
);
426 } else if (token
> Opt_last_int
&& token
< Opt_last_string
) {
427 dout("got string token %d val %s\n", token
,
429 } else if (token
> Opt_last_string
&& token
< Opt_last_bool
) {
430 dout("got Boolean token %d\n", token
);
432 dout("got token %d\n", token
);
437 rbd_opts
->read_only
= true;
440 rbd_opts
->read_only
= false;
450 * Get a ceph client with specific addr and configuration, if one does
451 * not exist create it.
453 static int rbd_get_client(struct rbd_device
*rbd_dev
, const char *mon_addr
,
454 size_t mon_addr_len
, char *options
)
456 struct rbd_options
*rbd_opts
= &rbd_dev
->rbd_opts
;
457 struct ceph_options
*ceph_opts
;
458 struct rbd_client
*rbdc
;
460 rbd_opts
->read_only
= RBD_READ_ONLY_DEFAULT
;
462 ceph_opts
= ceph_parse_options(options
, mon_addr
,
463 mon_addr
+ mon_addr_len
,
464 parse_rbd_opts_token
, rbd_opts
);
465 if (IS_ERR(ceph_opts
))
466 return PTR_ERR(ceph_opts
);
468 rbdc
= rbd_client_find(ceph_opts
);
470 /* using an existing client */
471 ceph_destroy_options(ceph_opts
);
473 rbdc
= rbd_client_create(ceph_opts
);
475 return PTR_ERR(rbdc
);
477 rbd_dev
->rbd_client
= rbdc
;
483 * Destroy ceph client
485 * Caller must hold rbd_client_list_lock.
487 static void rbd_client_release(struct kref
*kref
)
489 struct rbd_client
*rbdc
= container_of(kref
, struct rbd_client
, kref
);
491 dout("rbd_release_client %p\n", rbdc
);
492 spin_lock(&rbd_client_list_lock
);
493 list_del(&rbdc
->node
);
494 spin_unlock(&rbd_client_list_lock
);
496 ceph_destroy_client(rbdc
->client
);
501 * Drop reference to ceph client node. If it's not referenced anymore, release
504 static void rbd_put_client(struct rbd_device
*rbd_dev
)
506 kref_put(&rbd_dev
->rbd_client
->kref
, rbd_client_release
);
507 rbd_dev
->rbd_client
= NULL
;
511 * Destroy requests collection
513 static void rbd_coll_release(struct kref
*kref
)
515 struct rbd_req_coll
*coll
=
516 container_of(kref
, struct rbd_req_coll
, kref
);
518 dout("rbd_coll_release %p\n", coll
);
522 static bool rbd_image_format_valid(u32 image_format
)
524 return image_format
== 1 || image_format
== 2;
527 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk
*ondisk
)
532 /* The header has to start with the magic rbd header text */
533 if (memcmp(&ondisk
->text
, RBD_HEADER_TEXT
, sizeof (RBD_HEADER_TEXT
)))
537 * The size of a snapshot header has to fit in a size_t, and
538 * that limits the number of snapshots.
540 snap_count
= le32_to_cpu(ondisk
->snap_count
);
541 size
= SIZE_MAX
- sizeof (struct ceph_snap_context
);
542 if (snap_count
> size
/ sizeof (__le64
))
546 * Not only that, but the size of the entire the snapshot
547 * header must also be representable in a size_t.
549 size
-= snap_count
* sizeof (__le64
);
550 if ((u64
) size
< le64_to_cpu(ondisk
->snap_names_len
))
557 * Create a new header structure, translate header format from the on-disk
560 static int rbd_header_from_disk(struct rbd_image_header
*header
,
561 struct rbd_image_header_ondisk
*ondisk
)
568 memset(header
, 0, sizeof (*header
));
570 snap_count
= le32_to_cpu(ondisk
->snap_count
);
572 len
= strnlen(ondisk
->object_prefix
, sizeof (ondisk
->object_prefix
));
573 header
->object_prefix
= kmalloc(len
+ 1, GFP_KERNEL
);
574 if (!header
->object_prefix
)
576 memcpy(header
->object_prefix
, ondisk
->object_prefix
, len
);
577 header
->object_prefix
[len
] = '\0';
580 u64 snap_names_len
= le64_to_cpu(ondisk
->snap_names_len
);
582 /* Save a copy of the snapshot names */
584 if (snap_names_len
> (u64
) SIZE_MAX
)
586 header
->snap_names
= kmalloc(snap_names_len
, GFP_KERNEL
);
587 if (!header
->snap_names
)
590 * Note that rbd_dev_v1_header_read() guarantees
591 * the ondisk buffer we're working with has
592 * snap_names_len bytes beyond the end of the
593 * snapshot id array, this memcpy() is safe.
595 memcpy(header
->snap_names
, &ondisk
->snaps
[snap_count
],
598 /* Record each snapshot's size */
600 size
= snap_count
* sizeof (*header
->snap_sizes
);
601 header
->snap_sizes
= kmalloc(size
, GFP_KERNEL
);
602 if (!header
->snap_sizes
)
604 for (i
= 0; i
< snap_count
; i
++)
605 header
->snap_sizes
[i
] =
606 le64_to_cpu(ondisk
->snaps
[i
].image_size
);
608 WARN_ON(ondisk
->snap_names_len
);
609 header
->snap_names
= NULL
;
610 header
->snap_sizes
= NULL
;
613 header
->features
= 0; /* No features support in v1 images */
614 header
->obj_order
= ondisk
->options
.order
;
615 header
->crypt_type
= ondisk
->options
.crypt_type
;
616 header
->comp_type
= ondisk
->options
.comp_type
;
618 /* Allocate and fill in the snapshot context */
620 header
->image_size
= le64_to_cpu(ondisk
->image_size
);
621 size
= sizeof (struct ceph_snap_context
);
622 size
+= snap_count
* sizeof (header
->snapc
->snaps
[0]);
623 header
->snapc
= kzalloc(size
, GFP_KERNEL
);
627 atomic_set(&header
->snapc
->nref
, 1);
628 header
->snapc
->seq
= le64_to_cpu(ondisk
->snap_seq
);
629 header
->snapc
->num_snaps
= snap_count
;
630 for (i
= 0; i
< snap_count
; i
++)
631 header
->snapc
->snaps
[i
] =
632 le64_to_cpu(ondisk
->snaps
[i
].id
);
637 kfree(header
->snap_sizes
);
638 header
->snap_sizes
= NULL
;
639 kfree(header
->snap_names
);
640 header
->snap_names
= NULL
;
641 kfree(header
->object_prefix
);
642 header
->object_prefix
= NULL
;
647 static int snap_by_name(struct rbd_device
*rbd_dev
, const char *snap_name
)
650 struct rbd_snap
*snap
;
652 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
653 if (!strcmp(snap_name
, snap
->name
)) {
654 rbd_dev
->mapping
.snap_id
= snap
->id
;
655 rbd_dev
->mapping
.size
= snap
->size
;
656 rbd_dev
->mapping
.features
= snap
->features
;
665 static int rbd_dev_set_mapping(struct rbd_device
*rbd_dev
, char *snap_name
)
669 if (!memcmp(snap_name
, RBD_SNAP_HEAD_NAME
,
670 sizeof (RBD_SNAP_HEAD_NAME
))) {
671 rbd_dev
->mapping
.snap_id
= CEPH_NOSNAP
;
672 rbd_dev
->mapping
.size
= rbd_dev
->header
.image_size
;
673 rbd_dev
->mapping
.features
= rbd_dev
->header
.features
;
674 rbd_dev
->mapping
.snap_exists
= false;
675 rbd_dev
->mapping
.read_only
= rbd_dev
->rbd_opts
.read_only
;
678 ret
= snap_by_name(rbd_dev
, snap_name
);
681 rbd_dev
->mapping
.snap_exists
= true;
682 rbd_dev
->mapping
.read_only
= true;
684 rbd_dev
->mapping
.snap_name
= snap_name
;
689 static void rbd_header_free(struct rbd_image_header
*header
)
691 kfree(header
->object_prefix
);
692 header
->object_prefix
= NULL
;
693 kfree(header
->snap_sizes
);
694 header
->snap_sizes
= NULL
;
695 kfree(header
->snap_names
);
696 header
->snap_names
= NULL
;
697 ceph_put_snap_context(header
->snapc
);
698 header
->snapc
= NULL
;
701 static char *rbd_segment_name(struct rbd_device
*rbd_dev
, u64 offset
)
707 name
= kmalloc(RBD_MAX_SEG_NAME_LEN
+ 1, GFP_NOIO
);
710 segment
= offset
>> rbd_dev
->header
.obj_order
;
711 ret
= snprintf(name
, RBD_MAX_SEG_NAME_LEN
, "%s.%012llx",
712 rbd_dev
->header
.object_prefix
, segment
);
713 if (ret
< 0 || ret
>= RBD_MAX_SEG_NAME_LEN
) {
714 pr_err("error formatting segment name for #%llu (%d)\n",
723 static u64
rbd_segment_offset(struct rbd_device
*rbd_dev
, u64 offset
)
725 u64 segment_size
= (u64
) 1 << rbd_dev
->header
.obj_order
;
727 return offset
& (segment_size
- 1);
730 static u64
rbd_segment_length(struct rbd_device
*rbd_dev
,
731 u64 offset
, u64 length
)
733 u64 segment_size
= (u64
) 1 << rbd_dev
->header
.obj_order
;
735 offset
&= segment_size
- 1;
737 rbd_assert(length
<= U64_MAX
- offset
);
738 if (offset
+ length
> segment_size
)
739 length
= segment_size
- offset
;
744 static int rbd_get_num_segments(struct rbd_image_header
*header
,
752 if (len
- 1 > U64_MAX
- ofs
)
755 start_seg
= ofs
>> header
->obj_order
;
756 end_seg
= (ofs
+ len
- 1) >> header
->obj_order
;
758 return end_seg
- start_seg
+ 1;
762 * returns the size of an object in the image
764 static u64
rbd_obj_bytes(struct rbd_image_header
*header
)
766 return 1 << header
->obj_order
;
773 static void bio_chain_put(struct bio
*chain
)
779 chain
= chain
->bi_next
;
785 * zeros a bio chain, starting at specific offset
787 static void zero_bio_chain(struct bio
*chain
, int start_ofs
)
796 bio_for_each_segment(bv
, chain
, i
) {
797 if (pos
+ bv
->bv_len
> start_ofs
) {
798 int remainder
= max(start_ofs
- pos
, 0);
799 buf
= bvec_kmap_irq(bv
, &flags
);
800 memset(buf
+ remainder
, 0,
801 bv
->bv_len
- remainder
);
802 bvec_kunmap_irq(buf
, &flags
);
807 chain
= chain
->bi_next
;
812 * bio_chain_clone - clone a chain of bios up to a certain length.
813 * might return a bio_pair that will need to be released.
815 static struct bio
*bio_chain_clone(struct bio
**old
, struct bio
**next
,
816 struct bio_pair
**bp
,
817 int len
, gfp_t gfpmask
)
819 struct bio
*old_chain
= *old
;
820 struct bio
*new_chain
= NULL
;
825 bio_pair_release(*bp
);
829 while (old_chain
&& (total
< len
)) {
832 tmp
= bio_kmalloc(gfpmask
, old_chain
->bi_max_vecs
);
835 gfpmask
&= ~__GFP_WAIT
; /* can't wait after the first */
837 if (total
+ old_chain
->bi_size
> len
) {
841 * this split can only happen with a single paged bio,
842 * split_bio will BUG_ON if this is not the case
844 dout("bio_chain_clone split! total=%d remaining=%d"
846 total
, len
- total
, old_chain
->bi_size
);
848 /* split the bio. We'll release it either in the next
849 call, or it will have to be released outside */
850 bp
= bio_split(old_chain
, (len
- total
) / SECTOR_SIZE
);
854 __bio_clone(tmp
, &bp
->bio1
);
858 __bio_clone(tmp
, old_chain
);
859 *next
= old_chain
->bi_next
;
869 old_chain
= old_chain
->bi_next
;
871 total
+= tmp
->bi_size
;
874 rbd_assert(total
== len
);
881 dout("bio_chain_clone with err\n");
882 bio_chain_put(new_chain
);
887 * helpers for osd request op vectors.
889 static struct ceph_osd_req_op
*rbd_create_rw_ops(int num_ops
,
890 int opcode
, u32 payload_len
)
892 struct ceph_osd_req_op
*ops
;
894 ops
= kzalloc(sizeof (*ops
) * (num_ops
+ 1), GFP_NOIO
);
901 * op extent offset and length will be set later on
902 * in calc_raw_layout()
904 ops
[0].payload_len
= payload_len
;
909 static void rbd_destroy_ops(struct ceph_osd_req_op
*ops
)
914 static void rbd_coll_end_req_index(struct request
*rq
,
915 struct rbd_req_coll
*coll
,
919 struct request_queue
*q
;
922 dout("rbd_coll_end_req_index %p index %d ret %d len %llu\n",
923 coll
, index
, ret
, (unsigned long long) len
);
929 blk_end_request(rq
, ret
, len
);
935 spin_lock_irq(q
->queue_lock
);
936 coll
->status
[index
].done
= 1;
937 coll
->status
[index
].rc
= ret
;
938 coll
->status
[index
].bytes
= len
;
939 max
= min
= coll
->num_done
;
940 while (max
< coll
->total
&& coll
->status
[max
].done
)
943 for (i
= min
; i
<max
; i
++) {
944 __blk_end_request(rq
, coll
->status
[i
].rc
,
945 coll
->status
[i
].bytes
);
947 kref_put(&coll
->kref
, rbd_coll_release
);
949 spin_unlock_irq(q
->queue_lock
);
952 static void rbd_coll_end_req(struct rbd_request
*req
,
955 rbd_coll_end_req_index(req
->rq
, req
->coll
, req
->coll_index
, ret
, len
);
959 * Send ceph osd request
961 static int rbd_do_request(struct request
*rq
,
962 struct rbd_device
*rbd_dev
,
963 struct ceph_snap_context
*snapc
,
965 const char *object_name
, u64 ofs
, u64 len
,
970 struct ceph_osd_req_op
*ops
,
971 struct rbd_req_coll
*coll
,
973 void (*rbd_cb
)(struct ceph_osd_request
*req
,
974 struct ceph_msg
*msg
),
975 struct ceph_osd_request
**linger_req
,
978 struct ceph_osd_request
*req
;
979 struct ceph_file_layout
*layout
;
982 struct timespec mtime
= CURRENT_TIME
;
983 struct rbd_request
*req_data
;
984 struct ceph_osd_request_head
*reqhead
;
985 struct ceph_osd_client
*osdc
;
987 req_data
= kzalloc(sizeof(*req_data
), GFP_NOIO
);
990 rbd_coll_end_req_index(rq
, coll
, coll_index
,
996 req_data
->coll
= coll
;
997 req_data
->coll_index
= coll_index
;
1000 dout("rbd_do_request object_name=%s ofs=%llu len=%llu\n", object_name
,
1001 (unsigned long long) ofs
, (unsigned long long) len
);
1003 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1004 req
= ceph_osdc_alloc_request(osdc
, flags
, snapc
, ops
,
1005 false, GFP_NOIO
, pages
, bio
);
1011 req
->r_callback
= rbd_cb
;
1014 req_data
->bio
= bio
;
1015 req_data
->pages
= pages
;
1016 req_data
->len
= len
;
1018 req
->r_priv
= req_data
;
1020 reqhead
= req
->r_request
->front
.iov_base
;
1021 reqhead
->snapid
= cpu_to_le64(CEPH_NOSNAP
);
1023 strncpy(req
->r_oid
, object_name
, sizeof(req
->r_oid
));
1024 req
->r_oid_len
= strlen(req
->r_oid
);
1026 layout
= &req
->r_file_layout
;
1027 memset(layout
, 0, sizeof(*layout
));
1028 layout
->fl_stripe_unit
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
1029 layout
->fl_stripe_count
= cpu_to_le32(1);
1030 layout
->fl_object_size
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
1031 layout
->fl_pg_pool
= cpu_to_le32(rbd_dev
->pool_id
);
1032 ret
= ceph_calc_raw_layout(osdc
, layout
, snapid
, ofs
, &len
, &bno
,
1034 rbd_assert(ret
== 0);
1036 ceph_osdc_build_request(req
, ofs
, &len
,
1040 req
->r_oid
, req
->r_oid_len
);
1043 ceph_osdc_set_request_linger(osdc
, req
);
1047 ret
= ceph_osdc_start_request(osdc
, req
, false);
1052 ret
= ceph_osdc_wait_request(osdc
, req
);
1054 *ver
= le64_to_cpu(req
->r_reassert_version
.version
);
1055 dout("reassert_ver=%llu\n",
1056 (unsigned long long)
1057 le64_to_cpu(req
->r_reassert_version
.version
));
1058 ceph_osdc_put_request(req
);
1063 bio_chain_put(req_data
->bio
);
1064 ceph_osdc_put_request(req
);
1066 rbd_coll_end_req(req_data
, ret
, len
);
1072 * Ceph osd op callback
1074 static void rbd_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
1076 struct rbd_request
*req_data
= req
->r_priv
;
1077 struct ceph_osd_reply_head
*replyhead
;
1078 struct ceph_osd_op
*op
;
1084 replyhead
= msg
->front
.iov_base
;
1085 WARN_ON(le32_to_cpu(replyhead
->num_ops
) == 0);
1086 op
= (void *)(replyhead
+ 1);
1087 rc
= le32_to_cpu(replyhead
->result
);
1088 bytes
= le64_to_cpu(op
->extent
.length
);
1089 read_op
= (le16_to_cpu(op
->op
) == CEPH_OSD_OP_READ
);
1091 dout("rbd_req_cb bytes=%llu readop=%d rc=%d\n",
1092 (unsigned long long) bytes
, read_op
, (int) rc
);
1094 if (rc
== -ENOENT
&& read_op
) {
1095 zero_bio_chain(req_data
->bio
, 0);
1097 } else if (rc
== 0 && read_op
&& bytes
< req_data
->len
) {
1098 zero_bio_chain(req_data
->bio
, bytes
);
1099 bytes
= req_data
->len
;
1102 rbd_coll_end_req(req_data
, rc
, bytes
);
1105 bio_chain_put(req_data
->bio
);
1107 ceph_osdc_put_request(req
);
1111 static void rbd_simple_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
1113 ceph_osdc_put_request(req
);
1117 * Do a synchronous ceph osd operation
1119 static int rbd_req_sync_op(struct rbd_device
*rbd_dev
,
1120 struct ceph_snap_context
*snapc
,
1123 struct ceph_osd_req_op
*ops
,
1124 const char *object_name
,
1125 u64 ofs
, u64 inbound_size
,
1127 struct ceph_osd_request
**linger_req
,
1131 struct page
**pages
;
1134 rbd_assert(ops
!= NULL
);
1136 num_pages
= calc_pages_for(ofs
, inbound_size
);
1137 pages
= ceph_alloc_page_vector(num_pages
, GFP_KERNEL
);
1139 return PTR_ERR(pages
);
1141 ret
= rbd_do_request(NULL
, rbd_dev
, snapc
, snapid
,
1142 object_name
, ofs
, inbound_size
, NULL
,
1152 if ((flags
& CEPH_OSD_FLAG_READ
) && inbound
)
1153 ret
= ceph_copy_from_page_vector(pages
, inbound
, ofs
, ret
);
1156 ceph_release_page_vector(pages
, num_pages
);
1161 * Do an asynchronous ceph osd operation
1163 static int rbd_do_op(struct request
*rq
,
1164 struct rbd_device
*rbd_dev
,
1165 struct ceph_snap_context
*snapc
,
1167 int opcode
, int flags
,
1170 struct rbd_req_coll
*coll
,
1177 struct ceph_osd_req_op
*ops
;
1180 seg_name
= rbd_segment_name(rbd_dev
, ofs
);
1183 seg_len
= rbd_segment_length(rbd_dev
, ofs
, len
);
1184 seg_ofs
= rbd_segment_offset(rbd_dev
, ofs
);
1186 payload_len
= (flags
& CEPH_OSD_FLAG_WRITE
? seg_len
: 0);
1189 ops
= rbd_create_rw_ops(1, opcode
, payload_len
);
1193 /* we've taken care of segment sizes earlier when we
1194 cloned the bios. We should never have a segment
1195 truncated at this point */
1196 rbd_assert(seg_len
== len
);
1198 ret
= rbd_do_request(rq
, rbd_dev
, snapc
, snapid
,
1199 seg_name
, seg_ofs
, seg_len
,
1205 rbd_req_cb
, 0, NULL
);
1207 rbd_destroy_ops(ops
);
1214 * Request async osd write
1216 static int rbd_req_write(struct request
*rq
,
1217 struct rbd_device
*rbd_dev
,
1218 struct ceph_snap_context
*snapc
,
1221 struct rbd_req_coll
*coll
,
1224 return rbd_do_op(rq
, rbd_dev
, snapc
, CEPH_NOSNAP
,
1226 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1227 ofs
, len
, bio
, coll
, coll_index
);
1231 * Request async osd read
1233 static int rbd_req_read(struct request
*rq
,
1234 struct rbd_device
*rbd_dev
,
1238 struct rbd_req_coll
*coll
,
1241 return rbd_do_op(rq
, rbd_dev
, NULL
,
1245 ofs
, len
, bio
, coll
, coll_index
);
1249 * Request sync osd read
1251 static int rbd_req_sync_read(struct rbd_device
*rbd_dev
,
1253 const char *object_name
,
1258 struct ceph_osd_req_op
*ops
;
1261 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_READ
, 0);
1265 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1268 ops
, object_name
, ofs
, len
, buf
, NULL
, ver
);
1269 rbd_destroy_ops(ops
);
1275 * Request sync osd watch
1277 static int rbd_req_sync_notify_ack(struct rbd_device
*rbd_dev
,
1281 struct ceph_osd_req_op
*ops
;
1284 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_NOTIFY_ACK
, 0);
1288 ops
[0].watch
.ver
= cpu_to_le64(ver
);
1289 ops
[0].watch
.cookie
= notify_id
;
1290 ops
[0].watch
.flag
= 0;
1292 ret
= rbd_do_request(NULL
, rbd_dev
, NULL
, CEPH_NOSNAP
,
1293 rbd_dev
->header_name
, 0, 0, NULL
,
1298 rbd_simple_req_cb
, 0, NULL
);
1300 rbd_destroy_ops(ops
);
1304 static void rbd_watch_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1306 struct rbd_device
*rbd_dev
= (struct rbd_device
*)data
;
1313 dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
1314 rbd_dev
->header_name
, (unsigned long long) notify_id
,
1315 (unsigned int) opcode
);
1316 rc
= rbd_dev_refresh(rbd_dev
, &hver
);
1318 pr_warning(RBD_DRV_NAME
"%d got notification but failed to "
1319 " update snaps: %d\n", rbd_dev
->major
, rc
);
1321 rbd_req_sync_notify_ack(rbd_dev
, hver
, notify_id
);
1325 * Request sync osd watch
1327 static int rbd_req_sync_watch(struct rbd_device
*rbd_dev
)
1329 struct ceph_osd_req_op
*ops
;
1330 struct ceph_osd_client
*osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1333 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_WATCH
, 0);
1337 ret
= ceph_osdc_create_event(osdc
, rbd_watch_cb
, 0,
1338 (void *)rbd_dev
, &rbd_dev
->watch_event
);
1342 ops
[0].watch
.ver
= cpu_to_le64(rbd_dev
->header
.obj_version
);
1343 ops
[0].watch
.cookie
= cpu_to_le64(rbd_dev
->watch_event
->cookie
);
1344 ops
[0].watch
.flag
= 1;
1346 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1348 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1350 rbd_dev
->header_name
,
1352 &rbd_dev
->watch_request
, NULL
);
1357 rbd_destroy_ops(ops
);
1361 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1362 rbd_dev
->watch_event
= NULL
;
1364 rbd_destroy_ops(ops
);
1369 * Request sync osd unwatch
1371 static int rbd_req_sync_unwatch(struct rbd_device
*rbd_dev
)
1373 struct ceph_osd_req_op
*ops
;
1376 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_WATCH
, 0);
1380 ops
[0].watch
.ver
= 0;
1381 ops
[0].watch
.cookie
= cpu_to_le64(rbd_dev
->watch_event
->cookie
);
1382 ops
[0].watch
.flag
= 0;
1384 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1386 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1388 rbd_dev
->header_name
,
1389 0, 0, NULL
, NULL
, NULL
);
1392 rbd_destroy_ops(ops
);
1393 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1394 rbd_dev
->watch_event
= NULL
;
1399 * Synchronous osd object method call
1401 static int rbd_req_sync_exec(struct rbd_device
*rbd_dev
,
1402 const char *object_name
,
1403 const char *class_name
,
1404 const char *method_name
,
1405 const char *outbound
,
1406 size_t outbound_size
,
1408 size_t inbound_size
,
1412 struct ceph_osd_req_op
*ops
;
1413 int class_name_len
= strlen(class_name
);
1414 int method_name_len
= strlen(method_name
);
1419 * Any input parameters required by the method we're calling
1420 * will be sent along with the class and method names as
1421 * part of the message payload. That data and its size are
1422 * supplied via the indata and indata_len fields (named from
1423 * the perspective of the server side) in the OSD request
1426 payload_size
= class_name_len
+ method_name_len
+ outbound_size
;
1427 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_CALL
, payload_size
);
1431 ops
[0].cls
.class_name
= class_name
;
1432 ops
[0].cls
.class_len
= (__u8
) class_name_len
;
1433 ops
[0].cls
.method_name
= method_name
;
1434 ops
[0].cls
.method_len
= (__u8
) method_name_len
;
1435 ops
[0].cls
.argc
= 0;
1436 ops
[0].cls
.indata
= outbound
;
1437 ops
[0].cls
.indata_len
= outbound_size
;
1439 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1442 object_name
, 0, inbound_size
, inbound
,
1445 rbd_destroy_ops(ops
);
1447 dout("cls_exec returned %d\n", ret
);
1451 static struct rbd_req_coll
*rbd_alloc_coll(int num_reqs
)
1453 struct rbd_req_coll
*coll
=
1454 kzalloc(sizeof(struct rbd_req_coll
) +
1455 sizeof(struct rbd_req_status
) * num_reqs
,
1460 coll
->total
= num_reqs
;
1461 kref_init(&coll
->kref
);
1466 * block device queue callback
1468 static void rbd_rq_fn(struct request_queue
*q
)
1470 struct rbd_device
*rbd_dev
= q
->queuedata
;
1472 struct bio_pair
*bp
= NULL
;
1474 while ((rq
= blk_fetch_request(q
))) {
1476 struct bio
*rq_bio
, *next_bio
= NULL
;
1481 int num_segs
, cur_seg
= 0;
1482 struct rbd_req_coll
*coll
;
1483 struct ceph_snap_context
*snapc
;
1485 dout("fetched request\n");
1487 /* filter out block requests we don't understand */
1488 if ((rq
->cmd_type
!= REQ_TYPE_FS
)) {
1489 __blk_end_request_all(rq
, 0);
1493 /* deduce our operation (read, write) */
1494 do_write
= (rq_data_dir(rq
) == WRITE
);
1496 size
= blk_rq_bytes(rq
);
1497 ofs
= blk_rq_pos(rq
) * SECTOR_SIZE
;
1499 if (do_write
&& rbd_dev
->mapping
.read_only
) {
1500 __blk_end_request_all(rq
, -EROFS
);
1504 spin_unlock_irq(q
->queue_lock
);
1506 down_read(&rbd_dev
->header_rwsem
);
1508 if (rbd_dev
->mapping
.snap_id
!= CEPH_NOSNAP
&&
1509 !rbd_dev
->mapping
.snap_exists
) {
1510 up_read(&rbd_dev
->header_rwsem
);
1511 dout("request for non-existent snapshot");
1512 spin_lock_irq(q
->queue_lock
);
1513 __blk_end_request_all(rq
, -ENXIO
);
1517 snapc
= ceph_get_snap_context(rbd_dev
->header
.snapc
);
1519 up_read(&rbd_dev
->header_rwsem
);
1521 dout("%s 0x%x bytes at 0x%llx\n",
1522 do_write
? "write" : "read",
1523 size
, (unsigned long long) blk_rq_pos(rq
) * SECTOR_SIZE
);
1525 num_segs
= rbd_get_num_segments(&rbd_dev
->header
, ofs
, size
);
1526 if (num_segs
<= 0) {
1527 spin_lock_irq(q
->queue_lock
);
1528 __blk_end_request_all(rq
, num_segs
);
1529 ceph_put_snap_context(snapc
);
1532 coll
= rbd_alloc_coll(num_segs
);
1534 spin_lock_irq(q
->queue_lock
);
1535 __blk_end_request_all(rq
, -ENOMEM
);
1536 ceph_put_snap_context(snapc
);
1541 /* a bio clone to be passed down to OSD req */
1542 dout("rq->bio->bi_vcnt=%hu\n", rq
->bio
->bi_vcnt
);
1543 op_size
= rbd_segment_length(rbd_dev
, ofs
, size
);
1544 kref_get(&coll
->kref
);
1545 bio
= bio_chain_clone(&rq_bio
, &next_bio
, &bp
,
1546 op_size
, GFP_ATOMIC
);
1548 rbd_coll_end_req_index(rq
, coll
, cur_seg
,
1554 /* init OSD command: write or read */
1556 rbd_req_write(rq
, rbd_dev
,
1562 rbd_req_read(rq
, rbd_dev
,
1563 rbd_dev
->mapping
.snap_id
,
1575 kref_put(&coll
->kref
, rbd_coll_release
);
1578 bio_pair_release(bp
);
1579 spin_lock_irq(q
->queue_lock
);
1581 ceph_put_snap_context(snapc
);
1586 * a queue callback. Makes sure that we don't create a bio that spans across
1587 * multiple osd objects. One exception would be with a single page bios,
1588 * which we handle later at bio_chain_clone
1590 static int rbd_merge_bvec(struct request_queue
*q
, struct bvec_merge_data
*bmd
,
1591 struct bio_vec
*bvec
)
1593 struct rbd_device
*rbd_dev
= q
->queuedata
;
1594 unsigned int chunk_sectors
;
1596 unsigned int bio_sectors
;
1599 chunk_sectors
= 1 << (rbd_dev
->header
.obj_order
- SECTOR_SHIFT
);
1600 sector
= bmd
->bi_sector
+ get_start_sect(bmd
->bi_bdev
);
1601 bio_sectors
= bmd
->bi_size
>> SECTOR_SHIFT
;
1603 max
= (chunk_sectors
- ((sector
& (chunk_sectors
- 1))
1604 + bio_sectors
)) << SECTOR_SHIFT
;
1606 max
= 0; /* bio_add cannot handle a negative return */
1607 if (max
<= bvec
->bv_len
&& bio_sectors
== 0)
1608 return bvec
->bv_len
;
1612 static void rbd_free_disk(struct rbd_device
*rbd_dev
)
1614 struct gendisk
*disk
= rbd_dev
->disk
;
1619 if (disk
->flags
& GENHD_FL_UP
)
1622 blk_cleanup_queue(disk
->queue
);
1627 * Read the complete header for the given rbd device.
1629 * Returns a pointer to a dynamically-allocated buffer containing
1630 * the complete and validated header. Caller can pass the address
1631 * of a variable that will be filled in with the version of the
1632 * header object at the time it was read.
1634 * Returns a pointer-coded errno if a failure occurs.
1636 static struct rbd_image_header_ondisk
*
1637 rbd_dev_v1_header_read(struct rbd_device
*rbd_dev
, u64
*version
)
1639 struct rbd_image_header_ondisk
*ondisk
= NULL
;
1646 * The complete header will include an array of its 64-bit
1647 * snapshot ids, followed by the names of those snapshots as
1648 * a contiguous block of NUL-terminated strings. Note that
1649 * the number of snapshots could change by the time we read
1650 * it in, in which case we re-read it.
1657 size
= sizeof (*ondisk
);
1658 size
+= snap_count
* sizeof (struct rbd_image_snap_ondisk
);
1660 ondisk
= kmalloc(size
, GFP_KERNEL
);
1662 return ERR_PTR(-ENOMEM
);
1664 ret
= rbd_req_sync_read(rbd_dev
, CEPH_NOSNAP
,
1665 rbd_dev
->header_name
,
1667 (char *) ondisk
, version
);
1671 if (WARN_ON((size_t) ret
< size
)) {
1673 pr_warning("short header read for image %s"
1674 " (want %zd got %d)\n",
1675 rbd_dev
->image_name
, size
, ret
);
1678 if (!rbd_dev_ondisk_valid(ondisk
)) {
1680 pr_warning("invalid header for image %s\n",
1681 rbd_dev
->image_name
);
1685 names_size
= le64_to_cpu(ondisk
->snap_names_len
);
1686 want_count
= snap_count
;
1687 snap_count
= le32_to_cpu(ondisk
->snap_count
);
1688 } while (snap_count
!= want_count
);
1695 return ERR_PTR(ret
);
1699 * reload the ondisk the header
1701 static int rbd_read_header(struct rbd_device
*rbd_dev
,
1702 struct rbd_image_header
*header
)
1704 struct rbd_image_header_ondisk
*ondisk
;
1708 ondisk
= rbd_dev_v1_header_read(rbd_dev
, &ver
);
1710 return PTR_ERR(ondisk
);
1711 ret
= rbd_header_from_disk(header
, ondisk
);
1713 header
->obj_version
= ver
;
1719 static void __rbd_remove_all_snaps(struct rbd_device
*rbd_dev
)
1721 struct rbd_snap
*snap
;
1722 struct rbd_snap
*next
;
1724 list_for_each_entry_safe(snap
, next
, &rbd_dev
->snaps
, node
)
1725 __rbd_remove_snap_dev(snap
);
1728 static void rbd_update_mapping_size(struct rbd_device
*rbd_dev
)
1732 if (rbd_dev
->mapping
.snap_id
!= CEPH_NOSNAP
)
1735 size
= (sector_t
) rbd_dev
->header
.image_size
/ SECTOR_SIZE
;
1736 dout("setting size to %llu sectors", (unsigned long long) size
);
1737 rbd_dev
->mapping
.size
= (u64
) size
;
1738 set_capacity(rbd_dev
->disk
, size
);
1742 * only read the first part of the ondisk header, without the snaps info
1744 static int rbd_dev_v1_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
1747 struct rbd_image_header h
;
1749 ret
= rbd_read_header(rbd_dev
, &h
);
1753 down_write(&rbd_dev
->header_rwsem
);
1755 /* Update image size, and check for resize of mapped image */
1756 rbd_dev
->header
.image_size
= h
.image_size
;
1757 rbd_update_mapping_size(rbd_dev
);
1759 /* rbd_dev->header.object_prefix shouldn't change */
1760 kfree(rbd_dev
->header
.snap_sizes
);
1761 kfree(rbd_dev
->header
.snap_names
);
1762 /* osd requests may still refer to snapc */
1763 ceph_put_snap_context(rbd_dev
->header
.snapc
);
1766 *hver
= h
.obj_version
;
1767 rbd_dev
->header
.obj_version
= h
.obj_version
;
1768 rbd_dev
->header
.image_size
= h
.image_size
;
1769 rbd_dev
->header
.snapc
= h
.snapc
;
1770 rbd_dev
->header
.snap_names
= h
.snap_names
;
1771 rbd_dev
->header
.snap_sizes
= h
.snap_sizes
;
1772 /* Free the extra copy of the object prefix */
1773 WARN_ON(strcmp(rbd_dev
->header
.object_prefix
, h
.object_prefix
));
1774 kfree(h
.object_prefix
);
1776 ret
= rbd_dev_snaps_update(rbd_dev
);
1778 ret
= rbd_dev_snaps_register(rbd_dev
);
1780 up_write(&rbd_dev
->header_rwsem
);
1785 static int rbd_dev_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
1789 rbd_assert(rbd_image_format_valid(rbd_dev
->image_format
));
1790 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1791 if (rbd_dev
->image_format
== 1)
1792 ret
= rbd_dev_v1_refresh(rbd_dev
, hver
);
1794 ret
= rbd_dev_v2_refresh(rbd_dev
, hver
);
1795 mutex_unlock(&ctl_mutex
);
1800 static int rbd_init_disk(struct rbd_device
*rbd_dev
)
1802 struct gendisk
*disk
;
1803 struct request_queue
*q
;
1806 /* create gendisk info */
1807 disk
= alloc_disk(RBD_MINORS_PER_MAJOR
);
1811 snprintf(disk
->disk_name
, sizeof(disk
->disk_name
), RBD_DRV_NAME
"%d",
1813 disk
->major
= rbd_dev
->major
;
1814 disk
->first_minor
= 0;
1815 disk
->fops
= &rbd_bd_ops
;
1816 disk
->private_data
= rbd_dev
;
1819 q
= blk_init_queue(rbd_rq_fn
, &rbd_dev
->lock
);
1823 /* We use the default size, but let's be explicit about it. */
1824 blk_queue_physical_block_size(q
, SECTOR_SIZE
);
1826 /* set io sizes to object size */
1827 segment_size
= rbd_obj_bytes(&rbd_dev
->header
);
1828 blk_queue_max_hw_sectors(q
, segment_size
/ SECTOR_SIZE
);
1829 blk_queue_max_segment_size(q
, segment_size
);
1830 blk_queue_io_min(q
, segment_size
);
1831 blk_queue_io_opt(q
, segment_size
);
1833 blk_queue_merge_bvec(q
, rbd_merge_bvec
);
1836 q
->queuedata
= rbd_dev
;
1838 rbd_dev
->disk
= disk
;
1840 set_capacity(rbd_dev
->disk
, rbd_dev
->mapping
.size
/ SECTOR_SIZE
);
1853 static struct rbd_device
*dev_to_rbd_dev(struct device
*dev
)
1855 return container_of(dev
, struct rbd_device
, dev
);
1858 static ssize_t
rbd_size_show(struct device
*dev
,
1859 struct device_attribute
*attr
, char *buf
)
1861 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1864 down_read(&rbd_dev
->header_rwsem
);
1865 size
= get_capacity(rbd_dev
->disk
);
1866 up_read(&rbd_dev
->header_rwsem
);
1868 return sprintf(buf
, "%llu\n", (unsigned long long) size
* SECTOR_SIZE
);
1872 * Note this shows the features for whatever's mapped, which is not
1873 * necessarily the base image.
1875 static ssize_t
rbd_features_show(struct device
*dev
,
1876 struct device_attribute
*attr
, char *buf
)
1878 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1880 return sprintf(buf
, "0x%016llx\n",
1881 (unsigned long long) rbd_dev
->mapping
.features
);
1884 static ssize_t
rbd_major_show(struct device
*dev
,
1885 struct device_attribute
*attr
, char *buf
)
1887 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1889 return sprintf(buf
, "%d\n", rbd_dev
->major
);
1892 static ssize_t
rbd_client_id_show(struct device
*dev
,
1893 struct device_attribute
*attr
, char *buf
)
1895 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1897 return sprintf(buf
, "client%lld\n",
1898 ceph_client_id(rbd_dev
->rbd_client
->client
));
1901 static ssize_t
rbd_pool_show(struct device
*dev
,
1902 struct device_attribute
*attr
, char *buf
)
1904 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1906 return sprintf(buf
, "%s\n", rbd_dev
->pool_name
);
1909 static ssize_t
rbd_pool_id_show(struct device
*dev
,
1910 struct device_attribute
*attr
, char *buf
)
1912 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1914 return sprintf(buf
, "%d\n", rbd_dev
->pool_id
);
1917 static ssize_t
rbd_name_show(struct device
*dev
,
1918 struct device_attribute
*attr
, char *buf
)
1920 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1922 return sprintf(buf
, "%s\n", rbd_dev
->image_name
);
1925 static ssize_t
rbd_image_id_show(struct device
*dev
,
1926 struct device_attribute
*attr
, char *buf
)
1928 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1930 return sprintf(buf
, "%s\n", rbd_dev
->image_id
);
1934 * Shows the name of the currently-mapped snapshot (or
1935 * RBD_SNAP_HEAD_NAME for the base image).
1937 static ssize_t
rbd_snap_show(struct device
*dev
,
1938 struct device_attribute
*attr
,
1941 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1943 return sprintf(buf
, "%s\n", rbd_dev
->mapping
.snap_name
);
1946 static ssize_t
rbd_image_refresh(struct device
*dev
,
1947 struct device_attribute
*attr
,
1951 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1954 ret
= rbd_dev_refresh(rbd_dev
, NULL
);
1956 return ret
< 0 ? ret
: size
;
1959 static DEVICE_ATTR(size
, S_IRUGO
, rbd_size_show
, NULL
);
1960 static DEVICE_ATTR(features
, S_IRUGO
, rbd_features_show
, NULL
);
1961 static DEVICE_ATTR(major
, S_IRUGO
, rbd_major_show
, NULL
);
1962 static DEVICE_ATTR(client_id
, S_IRUGO
, rbd_client_id_show
, NULL
);
1963 static DEVICE_ATTR(pool
, S_IRUGO
, rbd_pool_show
, NULL
);
1964 static DEVICE_ATTR(pool_id
, S_IRUGO
, rbd_pool_id_show
, NULL
);
1965 static DEVICE_ATTR(name
, S_IRUGO
, rbd_name_show
, NULL
);
1966 static DEVICE_ATTR(image_id
, S_IRUGO
, rbd_image_id_show
, NULL
);
1967 static DEVICE_ATTR(refresh
, S_IWUSR
, NULL
, rbd_image_refresh
);
1968 static DEVICE_ATTR(current_snap
, S_IRUGO
, rbd_snap_show
, NULL
);
1970 static struct attribute
*rbd_attrs
[] = {
1971 &dev_attr_size
.attr
,
1972 &dev_attr_features
.attr
,
1973 &dev_attr_major
.attr
,
1974 &dev_attr_client_id
.attr
,
1975 &dev_attr_pool
.attr
,
1976 &dev_attr_pool_id
.attr
,
1977 &dev_attr_name
.attr
,
1978 &dev_attr_image_id
.attr
,
1979 &dev_attr_current_snap
.attr
,
1980 &dev_attr_refresh
.attr
,
1984 static struct attribute_group rbd_attr_group
= {
1988 static const struct attribute_group
*rbd_attr_groups
[] = {
1993 static void rbd_sysfs_dev_release(struct device
*dev
)
1997 static struct device_type rbd_device_type
= {
1999 .groups
= rbd_attr_groups
,
2000 .release
= rbd_sysfs_dev_release
,
2008 static ssize_t
rbd_snap_size_show(struct device
*dev
,
2009 struct device_attribute
*attr
,
2012 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2014 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->size
);
2017 static ssize_t
rbd_snap_id_show(struct device
*dev
,
2018 struct device_attribute
*attr
,
2021 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2023 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->id
);
2026 static ssize_t
rbd_snap_features_show(struct device
*dev
,
2027 struct device_attribute
*attr
,
2030 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2032 return sprintf(buf
, "0x%016llx\n",
2033 (unsigned long long) snap
->features
);
2036 static DEVICE_ATTR(snap_size
, S_IRUGO
, rbd_snap_size_show
, NULL
);
2037 static DEVICE_ATTR(snap_id
, S_IRUGO
, rbd_snap_id_show
, NULL
);
2038 static DEVICE_ATTR(snap_features
, S_IRUGO
, rbd_snap_features_show
, NULL
);
2040 static struct attribute
*rbd_snap_attrs
[] = {
2041 &dev_attr_snap_size
.attr
,
2042 &dev_attr_snap_id
.attr
,
2043 &dev_attr_snap_features
.attr
,
2047 static struct attribute_group rbd_snap_attr_group
= {
2048 .attrs
= rbd_snap_attrs
,
2051 static void rbd_snap_dev_release(struct device
*dev
)
2053 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2058 static const struct attribute_group
*rbd_snap_attr_groups
[] = {
2059 &rbd_snap_attr_group
,
2063 static struct device_type rbd_snap_device_type
= {
2064 .groups
= rbd_snap_attr_groups
,
2065 .release
= rbd_snap_dev_release
,
2068 static bool rbd_snap_registered(struct rbd_snap
*snap
)
2070 bool ret
= snap
->dev
.type
== &rbd_snap_device_type
;
2071 bool reg
= device_is_registered(&snap
->dev
);
2073 rbd_assert(!ret
^ reg
);
2078 static void __rbd_remove_snap_dev(struct rbd_snap
*snap
)
2080 list_del(&snap
->node
);
2081 if (device_is_registered(&snap
->dev
))
2082 device_unregister(&snap
->dev
);
2085 static int rbd_register_snap_dev(struct rbd_snap
*snap
,
2086 struct device
*parent
)
2088 struct device
*dev
= &snap
->dev
;
2091 dev
->type
= &rbd_snap_device_type
;
2092 dev
->parent
= parent
;
2093 dev
->release
= rbd_snap_dev_release
;
2094 dev_set_name(dev
, "snap_%s", snap
->name
);
2095 dout("%s: registering device for snapshot %s\n", __func__
, snap
->name
);
2097 ret
= device_register(dev
);
2102 static struct rbd_snap
*__rbd_add_snap_dev(struct rbd_device
*rbd_dev
,
2103 const char *snap_name
,
2104 u64 snap_id
, u64 snap_size
,
2107 struct rbd_snap
*snap
;
2110 snap
= kzalloc(sizeof (*snap
), GFP_KERNEL
);
2112 return ERR_PTR(-ENOMEM
);
2115 snap
->name
= kstrdup(snap_name
, GFP_KERNEL
);
2120 snap
->size
= snap_size
;
2121 snap
->features
= snap_features
;
2129 return ERR_PTR(ret
);
2132 static char *rbd_dev_v1_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2133 u64
*snap_size
, u64
*snap_features
)
2137 rbd_assert(which
< rbd_dev
->header
.snapc
->num_snaps
);
2139 *snap_size
= rbd_dev
->header
.snap_sizes
[which
];
2140 *snap_features
= 0; /* No features for v1 */
2142 /* Skip over names until we find the one we are looking for */
2144 snap_name
= rbd_dev
->header
.snap_names
;
2146 snap_name
+= strlen(snap_name
) + 1;
2152 * Get the size and object order for an image snapshot, or if
2153 * snap_id is CEPH_NOSNAP, gets this information for the base
2156 static int _rbd_dev_v2_snap_size(struct rbd_device
*rbd_dev
, u64 snap_id
,
2157 u8
*order
, u64
*snap_size
)
2159 __le64 snapid
= cpu_to_le64(snap_id
);
2164 } __attribute__ ((packed
)) size_buf
= { 0 };
2166 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2168 (char *) &snapid
, sizeof (snapid
),
2169 (char *) &size_buf
, sizeof (size_buf
),
2170 CEPH_OSD_FLAG_READ
, NULL
);
2171 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2175 *order
= size_buf
.order
;
2176 *snap_size
= le64_to_cpu(size_buf
.size
);
2178 dout(" snap_id 0x%016llx order = %u, snap_size = %llu\n",
2179 (unsigned long long) snap_id
, (unsigned int) *order
,
2180 (unsigned long long) *snap_size
);
2185 static int rbd_dev_v2_image_size(struct rbd_device
*rbd_dev
)
2187 return _rbd_dev_v2_snap_size(rbd_dev
, CEPH_NOSNAP
,
2188 &rbd_dev
->header
.obj_order
,
2189 &rbd_dev
->header
.image_size
);
2192 static int rbd_dev_v2_object_prefix(struct rbd_device
*rbd_dev
)
2198 reply_buf
= kzalloc(RBD_OBJ_PREFIX_LEN_MAX
, GFP_KERNEL
);
2202 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2203 "rbd", "get_object_prefix",
2205 reply_buf
, RBD_OBJ_PREFIX_LEN_MAX
,
2206 CEPH_OSD_FLAG_READ
, NULL
);
2207 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2212 rbd_dev
->header
.object_prefix
= ceph_extract_encoded_string(&p
,
2213 p
+ RBD_OBJ_PREFIX_LEN_MAX
,
2216 if (IS_ERR(rbd_dev
->header
.object_prefix
)) {
2217 ret
= PTR_ERR(rbd_dev
->header
.object_prefix
);
2218 rbd_dev
->header
.object_prefix
= NULL
;
2220 dout(" object_prefix = %s\n", rbd_dev
->header
.object_prefix
);
2229 static int _rbd_dev_v2_snap_features(struct rbd_device
*rbd_dev
, u64 snap_id
,
2232 __le64 snapid
= cpu_to_le64(snap_id
);
2236 } features_buf
= { 0 };
2240 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2241 "rbd", "get_features",
2242 (char *) &snapid
, sizeof (snapid
),
2243 (char *) &features_buf
, sizeof (features_buf
),
2244 CEPH_OSD_FLAG_READ
, NULL
);
2245 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2249 incompat
= le64_to_cpu(features_buf
.incompat
);
2250 if (incompat
& ~RBD_FEATURES_ALL
)
2253 *snap_features
= le64_to_cpu(features_buf
.features
);
2255 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
2256 (unsigned long long) snap_id
,
2257 (unsigned long long) *snap_features
,
2258 (unsigned long long) le64_to_cpu(features_buf
.incompat
));
2263 static int rbd_dev_v2_features(struct rbd_device
*rbd_dev
)
2265 return _rbd_dev_v2_snap_features(rbd_dev
, CEPH_NOSNAP
,
2266 &rbd_dev
->header
.features
);
2269 static int rbd_dev_v2_snap_context(struct rbd_device
*rbd_dev
, u64
*ver
)
2278 struct ceph_snap_context
*snapc
;
2282 * We'll need room for the seq value (maximum snapshot id),
2283 * snapshot count, and array of that many snapshot ids.
2284 * For now we have a fixed upper limit on the number we're
2285 * prepared to receive.
2287 size
= sizeof (__le64
) + sizeof (__le32
) +
2288 RBD_MAX_SNAP_COUNT
* sizeof (__le64
);
2289 reply_buf
= kzalloc(size
, GFP_KERNEL
);
2293 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2294 "rbd", "get_snapcontext",
2297 CEPH_OSD_FLAG_READ
, ver
);
2298 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2304 end
= (char *) reply_buf
+ size
;
2305 ceph_decode_64_safe(&p
, end
, seq
, out
);
2306 ceph_decode_32_safe(&p
, end
, snap_count
, out
);
2309 * Make sure the reported number of snapshot ids wouldn't go
2310 * beyond the end of our buffer. But before checking that,
2311 * make sure the computed size of the snapshot context we
2312 * allocate is representable in a size_t.
2314 if (snap_count
> (SIZE_MAX
- sizeof (struct ceph_snap_context
))
2319 if (!ceph_has_room(&p
, end
, snap_count
* sizeof (__le64
)))
2322 size
= sizeof (struct ceph_snap_context
) +
2323 snap_count
* sizeof (snapc
->snaps
[0]);
2324 snapc
= kmalloc(size
, GFP_KERNEL
);
2330 atomic_set(&snapc
->nref
, 1);
2332 snapc
->num_snaps
= snap_count
;
2333 for (i
= 0; i
< snap_count
; i
++)
2334 snapc
->snaps
[i
] = ceph_decode_64(&p
);
2336 rbd_dev
->header
.snapc
= snapc
;
2338 dout(" snap context seq = %llu, snap_count = %u\n",
2339 (unsigned long long) seq
, (unsigned int) snap_count
);
2347 static char *rbd_dev_v2_snap_name(struct rbd_device
*rbd_dev
, u32 which
)
2355 size_t snap_name_len
;
2358 size
= sizeof (__le32
) + RBD_MAX_SNAP_NAME_LEN
;
2359 reply_buf
= kmalloc(size
, GFP_KERNEL
);
2361 return ERR_PTR(-ENOMEM
);
2363 snap_id
= cpu_to_le64(rbd_dev
->header
.snapc
->snaps
[which
]);
2364 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2365 "rbd", "get_snapshot_name",
2366 (char *) &snap_id
, sizeof (snap_id
),
2368 CEPH_OSD_FLAG_READ
, NULL
);
2369 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2374 end
= (char *) reply_buf
+ size
;
2376 snap_name
= ceph_extract_encoded_string(&p
, end
, &snap_name_len
,
2378 if (IS_ERR(snap_name
)) {
2379 ret
= PTR_ERR(snap_name
);
2382 dout(" snap_id 0x%016llx snap_name = %s\n",
2383 (unsigned long long) le64_to_cpu(snap_id
), snap_name
);
2391 return ERR_PTR(ret
);
2394 static char *rbd_dev_v2_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2395 u64
*snap_size
, u64
*snap_features
)
2401 snap_id
= rbd_dev
->header
.snapc
->snaps
[which
];
2402 ret
= _rbd_dev_v2_snap_size(rbd_dev
, snap_id
, &order
, snap_size
);
2404 return ERR_PTR(ret
);
2405 ret
= _rbd_dev_v2_snap_features(rbd_dev
, snap_id
, snap_features
);
2407 return ERR_PTR(ret
);
2409 return rbd_dev_v2_snap_name(rbd_dev
, which
);
2412 static char *rbd_dev_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2413 u64
*snap_size
, u64
*snap_features
)
2415 if (rbd_dev
->image_format
== 1)
2416 return rbd_dev_v1_snap_info(rbd_dev
, which
,
2417 snap_size
, snap_features
);
2418 if (rbd_dev
->image_format
== 2)
2419 return rbd_dev_v2_snap_info(rbd_dev
, which
,
2420 snap_size
, snap_features
);
2421 return ERR_PTR(-EINVAL
);
2424 static int rbd_dev_v2_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
2429 down_write(&rbd_dev
->header_rwsem
);
2431 /* Grab old order first, to see if it changes */
2433 obj_order
= rbd_dev
->header
.obj_order
,
2434 ret
= rbd_dev_v2_image_size(rbd_dev
);
2437 if (rbd_dev
->header
.obj_order
!= obj_order
) {
2441 rbd_update_mapping_size(rbd_dev
);
2443 ret
= rbd_dev_v2_snap_context(rbd_dev
, hver
);
2444 dout("rbd_dev_v2_snap_context returned %d\n", ret
);
2447 ret
= rbd_dev_snaps_update(rbd_dev
);
2448 dout("rbd_dev_snaps_update returned %d\n", ret
);
2451 ret
= rbd_dev_snaps_register(rbd_dev
);
2452 dout("rbd_dev_snaps_register returned %d\n", ret
);
2454 up_write(&rbd_dev
->header_rwsem
);
2460 * Scan the rbd device's current snapshot list and compare it to the
2461 * newly-received snapshot context. Remove any existing snapshots
2462 * not present in the new snapshot context. Add a new snapshot for
2463 * any snaphots in the snapshot context not in the current list.
2464 * And verify there are no changes to snapshots we already know
2467 * Assumes the snapshots in the snapshot context are sorted by
2468 * snapshot id, highest id first. (Snapshots in the rbd_dev's list
2469 * are also maintained in that order.)
2471 static int rbd_dev_snaps_update(struct rbd_device
*rbd_dev
)
2473 struct ceph_snap_context
*snapc
= rbd_dev
->header
.snapc
;
2474 const u32 snap_count
= snapc
->num_snaps
;
2475 struct list_head
*head
= &rbd_dev
->snaps
;
2476 struct list_head
*links
= head
->next
;
2479 dout("%s: snap count is %u\n", __func__
, (unsigned int) snap_count
);
2480 while (index
< snap_count
|| links
!= head
) {
2482 struct rbd_snap
*snap
;
2485 u64 snap_features
= 0;
2487 snap_id
= index
< snap_count
? snapc
->snaps
[index
]
2489 snap
= links
!= head
? list_entry(links
, struct rbd_snap
, node
)
2491 rbd_assert(!snap
|| snap
->id
!= CEPH_NOSNAP
);
2493 if (snap_id
== CEPH_NOSNAP
|| (snap
&& snap
->id
> snap_id
)) {
2494 struct list_head
*next
= links
->next
;
2496 /* Existing snapshot not in the new snap context */
2498 if (rbd_dev
->mapping
.snap_id
== snap
->id
)
2499 rbd_dev
->mapping
.snap_exists
= false;
2500 __rbd_remove_snap_dev(snap
);
2501 dout("%ssnap id %llu has been removed\n",
2502 rbd_dev
->mapping
.snap_id
== snap
->id
?
2504 (unsigned long long) snap
->id
);
2506 /* Done with this list entry; advance */
2512 snap_name
= rbd_dev_snap_info(rbd_dev
, index
,
2513 &snap_size
, &snap_features
);
2514 if (IS_ERR(snap_name
))
2515 return PTR_ERR(snap_name
);
2517 dout("entry %u: snap_id = %llu\n", (unsigned int) snap_count
,
2518 (unsigned long long) snap_id
);
2519 if (!snap
|| (snap_id
!= CEPH_NOSNAP
&& snap
->id
< snap_id
)) {
2520 struct rbd_snap
*new_snap
;
2522 /* We haven't seen this snapshot before */
2524 new_snap
= __rbd_add_snap_dev(rbd_dev
, snap_name
,
2525 snap_id
, snap_size
, snap_features
);
2526 if (IS_ERR(new_snap
)) {
2527 int err
= PTR_ERR(new_snap
);
2529 dout(" failed to add dev, error %d\n", err
);
2534 /* New goes before existing, or at end of list */
2536 dout(" added dev%s\n", snap
? "" : " at end\n");
2538 list_add_tail(&new_snap
->node
, &snap
->node
);
2540 list_add_tail(&new_snap
->node
, head
);
2542 /* Already have this one */
2544 dout(" already present\n");
2546 rbd_assert(snap
->size
== snap_size
);
2547 rbd_assert(!strcmp(snap
->name
, snap_name
));
2548 rbd_assert(snap
->features
== snap_features
);
2550 /* Done with this list entry; advance */
2552 links
= links
->next
;
2555 /* Advance to the next entry in the snapshot context */
2559 dout("%s: done\n", __func__
);
2565 * Scan the list of snapshots and register the devices for any that
2566 * have not already been registered.
2568 static int rbd_dev_snaps_register(struct rbd_device
*rbd_dev
)
2570 struct rbd_snap
*snap
;
2573 dout("%s called\n", __func__
);
2574 if (WARN_ON(!device_is_registered(&rbd_dev
->dev
)))
2577 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
2578 if (!rbd_snap_registered(snap
)) {
2579 ret
= rbd_register_snap_dev(snap
, &rbd_dev
->dev
);
2584 dout("%s: returning %d\n", __func__
, ret
);
2589 static int rbd_bus_add_dev(struct rbd_device
*rbd_dev
)
2594 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2596 dev
= &rbd_dev
->dev
;
2597 dev
->bus
= &rbd_bus_type
;
2598 dev
->type
= &rbd_device_type
;
2599 dev
->parent
= &rbd_root_dev
;
2600 dev
->release
= rbd_dev_release
;
2601 dev_set_name(dev
, "%d", rbd_dev
->dev_id
);
2602 ret
= device_register(dev
);
2604 mutex_unlock(&ctl_mutex
);
2609 static void rbd_bus_del_dev(struct rbd_device
*rbd_dev
)
2611 device_unregister(&rbd_dev
->dev
);
2614 static int rbd_init_watch_dev(struct rbd_device
*rbd_dev
)
2619 ret
= rbd_req_sync_watch(rbd_dev
);
2620 if (ret
== -ERANGE
) {
2621 rc
= rbd_dev_refresh(rbd_dev
, NULL
);
2625 } while (ret
== -ERANGE
);
2630 static atomic64_t rbd_dev_id_max
= ATOMIC64_INIT(0);
2633 * Get a unique rbd identifier for the given new rbd_dev, and add
2634 * the rbd_dev to the global list. The minimum rbd id is 1.
2636 static void rbd_dev_id_get(struct rbd_device
*rbd_dev
)
2638 rbd_dev
->dev_id
= atomic64_inc_return(&rbd_dev_id_max
);
2640 spin_lock(&rbd_dev_list_lock
);
2641 list_add_tail(&rbd_dev
->node
, &rbd_dev_list
);
2642 spin_unlock(&rbd_dev_list_lock
);
2643 dout("rbd_dev %p given dev id %llu\n", rbd_dev
,
2644 (unsigned long long) rbd_dev
->dev_id
);
2648 * Remove an rbd_dev from the global list, and record that its
2649 * identifier is no longer in use.
2651 static void rbd_dev_id_put(struct rbd_device
*rbd_dev
)
2653 struct list_head
*tmp
;
2654 int rbd_id
= rbd_dev
->dev_id
;
2657 rbd_assert(rbd_id
> 0);
2659 dout("rbd_dev %p released dev id %llu\n", rbd_dev
,
2660 (unsigned long long) rbd_dev
->dev_id
);
2661 spin_lock(&rbd_dev_list_lock
);
2662 list_del_init(&rbd_dev
->node
);
2665 * If the id being "put" is not the current maximum, there
2666 * is nothing special we need to do.
2668 if (rbd_id
!= atomic64_read(&rbd_dev_id_max
)) {
2669 spin_unlock(&rbd_dev_list_lock
);
2674 * We need to update the current maximum id. Search the
2675 * list to find out what it is. We're more likely to find
2676 * the maximum at the end, so search the list backward.
2679 list_for_each_prev(tmp
, &rbd_dev_list
) {
2680 struct rbd_device
*rbd_dev
;
2682 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
2683 if (rbd_id
> max_id
)
2686 spin_unlock(&rbd_dev_list_lock
);
2689 * The max id could have been updated by rbd_dev_id_get(), in
2690 * which case it now accurately reflects the new maximum.
2691 * Be careful not to overwrite the maximum value in that
2694 atomic64_cmpxchg(&rbd_dev_id_max
, rbd_id
, max_id
);
2695 dout(" max dev id has been reset\n");
2699 * Skips over white space at *buf, and updates *buf to point to the
2700 * first found non-space character (if any). Returns the length of
2701 * the token (string of non-white space characters) found. Note
2702 * that *buf must be terminated with '\0'.
2704 static inline size_t next_token(const char **buf
)
2707 * These are the characters that produce nonzero for
2708 * isspace() in the "C" and "POSIX" locales.
2710 const char *spaces
= " \f\n\r\t\v";
2712 *buf
+= strspn(*buf
, spaces
); /* Find start of token */
2714 return strcspn(*buf
, spaces
); /* Return token length */
2718 * Finds the next token in *buf, and if the provided token buffer is
2719 * big enough, copies the found token into it. The result, if
2720 * copied, is guaranteed to be terminated with '\0'. Note that *buf
2721 * must be terminated with '\0' on entry.
2723 * Returns the length of the token found (not including the '\0').
2724 * Return value will be 0 if no token is found, and it will be >=
2725 * token_size if the token would not fit.
2727 * The *buf pointer will be updated to point beyond the end of the
2728 * found token. Note that this occurs even if the token buffer is
2729 * too small to hold it.
2731 static inline size_t copy_token(const char **buf
,
2737 len
= next_token(buf
);
2738 if (len
< token_size
) {
2739 memcpy(token
, *buf
, len
);
2740 *(token
+ len
) = '\0';
2748 * Finds the next token in *buf, dynamically allocates a buffer big
2749 * enough to hold a copy of it, and copies the token into the new
2750 * buffer. The copy is guaranteed to be terminated with '\0'. Note
2751 * that a duplicate buffer is created even for a zero-length token.
2753 * Returns a pointer to the newly-allocated duplicate, or a null
2754 * pointer if memory for the duplicate was not available. If
2755 * the lenp argument is a non-null pointer, the length of the token
2756 * (not including the '\0') is returned in *lenp.
2758 * If successful, the *buf pointer will be updated to point beyond
2759 * the end of the found token.
2761 * Note: uses GFP_KERNEL for allocation.
2763 static inline char *dup_token(const char **buf
, size_t *lenp
)
2768 len
= next_token(buf
);
2769 dup
= kmalloc(len
+ 1, GFP_KERNEL
);
2773 memcpy(dup
, *buf
, len
);
2774 *(dup
+ len
) = '\0';
2784 * This fills in the pool_name, image_name, image_name_len, rbd_dev,
2785 * rbd_md_name, and name fields of the given rbd_dev, based on the
2786 * list of monitor addresses and other options provided via
2787 * /sys/bus/rbd/add. Returns a pointer to a dynamically-allocated
2788 * copy of the snapshot name to map if successful, or a
2789 * pointer-coded error otherwise.
2791 * Note: rbd_dev is assumed to have been initially zero-filled.
2793 static char *rbd_add_parse_args(struct rbd_device
*rbd_dev
,
2795 const char **mon_addrs
,
2796 size_t *mon_addrs_size
,
2798 size_t options_size
)
2801 char *err_ptr
= ERR_PTR(-EINVAL
);
2804 /* The first four tokens are required */
2806 len
= next_token(&buf
);
2809 *mon_addrs_size
= len
+ 1;
2814 len
= copy_token(&buf
, options
, options_size
);
2815 if (!len
|| len
>= options_size
)
2818 err_ptr
= ERR_PTR(-ENOMEM
);
2819 rbd_dev
->pool_name
= dup_token(&buf
, NULL
);
2820 if (!rbd_dev
->pool_name
)
2823 rbd_dev
->image_name
= dup_token(&buf
, &rbd_dev
->image_name_len
);
2824 if (!rbd_dev
->image_name
)
2827 /* Snapshot name is optional */
2828 len
= next_token(&buf
);
2830 buf
= RBD_SNAP_HEAD_NAME
; /* No snapshot supplied */
2831 len
= sizeof (RBD_SNAP_HEAD_NAME
) - 1;
2833 snap_name
= kmalloc(len
+ 1, GFP_KERNEL
);
2836 memcpy(snap_name
, buf
, len
);
2837 *(snap_name
+ len
) = '\0';
2839 dout(" SNAP_NAME is <%s>, len is %zd\n", snap_name
, len
);
2844 kfree(rbd_dev
->image_name
);
2845 rbd_dev
->image_name
= NULL
;
2846 rbd_dev
->image_name_len
= 0;
2847 kfree(rbd_dev
->pool_name
);
2848 rbd_dev
->pool_name
= NULL
;
2854 * An rbd format 2 image has a unique identifier, distinct from the
2855 * name given to it by the user. Internally, that identifier is
2856 * what's used to specify the names of objects related to the image.
2858 * A special "rbd id" object is used to map an rbd image name to its
2859 * id. If that object doesn't exist, then there is no v2 rbd image
2860 * with the supplied name.
2862 * This function will record the given rbd_dev's image_id field if
2863 * it can be determined, and in that case will return 0. If any
2864 * errors occur a negative errno will be returned and the rbd_dev's
2865 * image_id field will be unchanged (and should be NULL).
2867 static int rbd_dev_image_id(struct rbd_device
*rbd_dev
)
2876 * First, see if the format 2 image id file exists, and if
2877 * so, get the image's persistent id from it.
2879 size
= sizeof (RBD_ID_PREFIX
) + rbd_dev
->image_name_len
;
2880 object_name
= kmalloc(size
, GFP_NOIO
);
2883 sprintf(object_name
, "%s%s", RBD_ID_PREFIX
, rbd_dev
->image_name
);
2884 dout("rbd id object name is %s\n", object_name
);
2886 /* Response will be an encoded string, which includes a length */
2888 size
= sizeof (__le32
) + RBD_IMAGE_ID_LEN_MAX
;
2889 response
= kzalloc(size
, GFP_NOIO
);
2895 ret
= rbd_req_sync_exec(rbd_dev
, object_name
,
2898 response
, RBD_IMAGE_ID_LEN_MAX
,
2899 CEPH_OSD_FLAG_READ
, NULL
);
2900 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2905 rbd_dev
->image_id
= ceph_extract_encoded_string(&p
,
2906 p
+ RBD_IMAGE_ID_LEN_MAX
,
2907 &rbd_dev
->image_id_len
,
2909 if (IS_ERR(rbd_dev
->image_id
)) {
2910 ret
= PTR_ERR(rbd_dev
->image_id
);
2911 rbd_dev
->image_id
= NULL
;
2913 dout("image_id is %s\n", rbd_dev
->image_id
);
2922 static int rbd_dev_v1_probe(struct rbd_device
*rbd_dev
)
2927 /* Version 1 images have no id; empty string is used */
2929 rbd_dev
->image_id
= kstrdup("", GFP_KERNEL
);
2930 if (!rbd_dev
->image_id
)
2932 rbd_dev
->image_id_len
= 0;
2934 /* Record the header object name for this rbd image. */
2936 size
= rbd_dev
->image_name_len
+ sizeof (RBD_SUFFIX
);
2937 rbd_dev
->header_name
= kmalloc(size
, GFP_KERNEL
);
2938 if (!rbd_dev
->header_name
) {
2942 sprintf(rbd_dev
->header_name
, "%s%s", rbd_dev
->image_name
, RBD_SUFFIX
);
2944 /* Populate rbd image metadata */
2946 ret
= rbd_read_header(rbd_dev
, &rbd_dev
->header
);
2949 rbd_dev
->image_format
= 1;
2951 dout("discovered version 1 image, header name is %s\n",
2952 rbd_dev
->header_name
);
2957 kfree(rbd_dev
->header_name
);
2958 rbd_dev
->header_name
= NULL
;
2959 kfree(rbd_dev
->image_id
);
2960 rbd_dev
->image_id
= NULL
;
2965 static int rbd_dev_v2_probe(struct rbd_device
*rbd_dev
)
2972 * Image id was filled in by the caller. Record the header
2973 * object name for this rbd image.
2975 size
= sizeof (RBD_HEADER_PREFIX
) + rbd_dev
->image_id_len
;
2976 rbd_dev
->header_name
= kmalloc(size
, GFP_KERNEL
);
2977 if (!rbd_dev
->header_name
)
2979 sprintf(rbd_dev
->header_name
, "%s%s",
2980 RBD_HEADER_PREFIX
, rbd_dev
->image_id
);
2982 /* Get the size and object order for the image */
2984 ret
= rbd_dev_v2_image_size(rbd_dev
);
2988 /* Get the object prefix (a.k.a. block_name) for the image */
2990 ret
= rbd_dev_v2_object_prefix(rbd_dev
);
2994 /* Get the and check features for the image */
2996 ret
= rbd_dev_v2_features(rbd_dev
);
3000 /* crypto and compression type aren't (yet) supported for v2 images */
3002 rbd_dev
->header
.crypt_type
= 0;
3003 rbd_dev
->header
.comp_type
= 0;
3005 /* Get the snapshot context, plus the header version */
3007 ret
= rbd_dev_v2_snap_context(rbd_dev
, &ver
);
3010 rbd_dev
->header
.obj_version
= ver
;
3012 rbd_dev
->image_format
= 2;
3014 dout("discovered version 2 image, header name is %s\n",
3015 rbd_dev
->header_name
);
3019 kfree(rbd_dev
->header_name
);
3020 rbd_dev
->header_name
= NULL
;
3021 kfree(rbd_dev
->header
.object_prefix
);
3022 rbd_dev
->header
.object_prefix
= NULL
;
3028 * Probe for the existence of the header object for the given rbd
3029 * device. For format 2 images this includes determining the image
3032 static int rbd_dev_probe(struct rbd_device
*rbd_dev
)
3037 * Get the id from the image id object. If it's not a
3038 * format 2 image, we'll get ENOENT back, and we'll assume
3039 * it's a format 1 image.
3041 ret
= rbd_dev_image_id(rbd_dev
);
3043 ret
= rbd_dev_v1_probe(rbd_dev
);
3045 ret
= rbd_dev_v2_probe(rbd_dev
);
3047 dout("probe failed, returning %d\n", ret
);
3052 static ssize_t
rbd_add(struct bus_type
*bus
,
3057 struct rbd_device
*rbd_dev
= NULL
;
3058 const char *mon_addrs
= NULL
;
3059 size_t mon_addrs_size
= 0;
3060 struct ceph_osd_client
*osdc
;
3064 if (!try_module_get(THIS_MODULE
))
3067 options
= kmalloc(count
, GFP_KERNEL
);
3070 rbd_dev
= kzalloc(sizeof(*rbd_dev
), GFP_KERNEL
);
3074 /* static rbd_device initialization */
3075 spin_lock_init(&rbd_dev
->lock
);
3076 INIT_LIST_HEAD(&rbd_dev
->node
);
3077 INIT_LIST_HEAD(&rbd_dev
->snaps
);
3078 init_rwsem(&rbd_dev
->header_rwsem
);
3080 /* parse add command */
3081 snap_name
= rbd_add_parse_args(rbd_dev
, buf
,
3082 &mon_addrs
, &mon_addrs_size
, options
, count
);
3083 if (IS_ERR(snap_name
)) {
3084 rc
= PTR_ERR(snap_name
);
3088 rc
= rbd_get_client(rbd_dev
, mon_addrs
, mon_addrs_size
- 1, options
);
3093 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
3094 rc
= ceph_pg_poolid_by_name(osdc
->osdmap
, rbd_dev
->pool_name
);
3096 goto err_out_client
;
3097 rbd_dev
->pool_id
= rc
;
3099 rc
= rbd_dev_probe(rbd_dev
);
3101 goto err_out_client
;
3103 /* no need to lock here, as rbd_dev is not registered yet */
3104 rc
= rbd_dev_snaps_update(rbd_dev
);
3106 goto err_out_header
;
3108 rc
= rbd_dev_set_mapping(rbd_dev
, snap_name
);
3110 goto err_out_header
;
3112 /* generate unique id: find highest unique id, add one */
3113 rbd_dev_id_get(rbd_dev
);
3115 /* Fill in the device name, now that we have its id. */
3116 BUILD_BUG_ON(DEV_NAME_LEN
3117 < sizeof (RBD_DRV_NAME
) + MAX_INT_FORMAT_WIDTH
);
3118 sprintf(rbd_dev
->name
, "%s%d", RBD_DRV_NAME
, rbd_dev
->dev_id
);
3120 /* Get our block major device number. */
3122 rc
= register_blkdev(0, rbd_dev
->name
);
3125 rbd_dev
->major
= rc
;
3127 /* Set up the blkdev mapping. */
3129 rc
= rbd_init_disk(rbd_dev
);
3131 goto err_out_blkdev
;
3133 rc
= rbd_bus_add_dev(rbd_dev
);
3138 * At this point cleanup in the event of an error is the job
3139 * of the sysfs code (initiated by rbd_bus_del_dev()).
3142 down_write(&rbd_dev
->header_rwsem
);
3143 rc
= rbd_dev_snaps_register(rbd_dev
);
3144 up_write(&rbd_dev
->header_rwsem
);
3148 rc
= rbd_init_watch_dev(rbd_dev
);
3152 /* Everything's ready. Announce the disk to the world. */
3154 add_disk(rbd_dev
->disk
);
3156 pr_info("%s: added with size 0x%llx\n", rbd_dev
->disk
->disk_name
,
3157 (unsigned long long) rbd_dev
->mapping
.size
);
3162 /* this will also clean up rest of rbd_dev stuff */
3164 rbd_bus_del_dev(rbd_dev
);
3169 rbd_free_disk(rbd_dev
);
3171 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
3173 rbd_dev_id_put(rbd_dev
);
3175 rbd_header_free(&rbd_dev
->header
);
3177 kfree(rbd_dev
->header_name
);
3178 rbd_put_client(rbd_dev
);
3179 kfree(rbd_dev
->image_id
);
3181 kfree(rbd_dev
->mapping
.snap_name
);
3182 kfree(rbd_dev
->image_name
);
3183 kfree(rbd_dev
->pool_name
);
3188 dout("Error adding device %s\n", buf
);
3189 module_put(THIS_MODULE
);
3191 return (ssize_t
) rc
;
3194 static struct rbd_device
*__rbd_get_dev(unsigned long dev_id
)
3196 struct list_head
*tmp
;
3197 struct rbd_device
*rbd_dev
;
3199 spin_lock(&rbd_dev_list_lock
);
3200 list_for_each(tmp
, &rbd_dev_list
) {
3201 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
3202 if (rbd_dev
->dev_id
== dev_id
) {
3203 spin_unlock(&rbd_dev_list_lock
);
3207 spin_unlock(&rbd_dev_list_lock
);
3211 static void rbd_dev_release(struct device
*dev
)
3213 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
3215 if (rbd_dev
->watch_request
) {
3216 struct ceph_client
*client
= rbd_dev
->rbd_client
->client
;
3218 ceph_osdc_unregister_linger_request(&client
->osdc
,
3219 rbd_dev
->watch_request
);
3221 if (rbd_dev
->watch_event
)
3222 rbd_req_sync_unwatch(rbd_dev
);
3224 rbd_put_client(rbd_dev
);
3226 /* clean up and free blkdev */
3227 rbd_free_disk(rbd_dev
);
3228 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
3230 /* release allocated disk header fields */
3231 rbd_header_free(&rbd_dev
->header
);
3233 /* done with the id, and with the rbd_dev */
3234 kfree(rbd_dev
->mapping
.snap_name
);
3235 kfree(rbd_dev
->image_id
);
3236 kfree(rbd_dev
->header_name
);
3237 kfree(rbd_dev
->pool_name
);
3238 kfree(rbd_dev
->image_name
);
3239 rbd_dev_id_put(rbd_dev
);
3242 /* release module ref */
3243 module_put(THIS_MODULE
);
3246 static ssize_t
rbd_remove(struct bus_type
*bus
,
3250 struct rbd_device
*rbd_dev
= NULL
;
3255 rc
= strict_strtoul(buf
, 10, &ul
);
3259 /* convert to int; abort if we lost anything in the conversion */
3260 target_id
= (int) ul
;
3261 if (target_id
!= ul
)
3264 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
3266 rbd_dev
= __rbd_get_dev(target_id
);
3272 __rbd_remove_all_snaps(rbd_dev
);
3273 rbd_bus_del_dev(rbd_dev
);
3276 mutex_unlock(&ctl_mutex
);
3282 * create control files in sysfs
3285 static int rbd_sysfs_init(void)
3289 ret
= device_register(&rbd_root_dev
);
3293 ret
= bus_register(&rbd_bus_type
);
3295 device_unregister(&rbd_root_dev
);
3300 static void rbd_sysfs_cleanup(void)
3302 bus_unregister(&rbd_bus_type
);
3303 device_unregister(&rbd_root_dev
);
3306 int __init
rbd_init(void)
3310 rc
= rbd_sysfs_init();
3313 pr_info("loaded " RBD_DRV_NAME_LONG
"\n");
3317 void __exit
rbd_exit(void)
3319 rbd_sysfs_cleanup();
3322 module_init(rbd_init
);
3323 module_exit(rbd_exit
);
3325 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
3326 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
3327 MODULE_DESCRIPTION("rados block device");
3329 /* following authorship retained from original osdblk.c */
3330 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
3332 MODULE_LICENSE("GPL");