2 rbd.c -- Export ceph rados objects as a Linux block device
5 based on drivers/block/osdblk.c:
7 Copyright 2009 Red Hat, Inc.
9 This program is free software; you can redistribute it and/or modify
10 it under the terms of the GNU General Public License as published by
11 the Free Software Foundation.
13 This program is distributed in the hope that it will be useful,
14 but WITHOUT ANY WARRANTY; without even the implied warranty of
15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 GNU General Public License for more details.
18 You should have received a copy of the GNU General Public License
19 along with this program; see the file COPYING. If not, write to
20 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
24 For usage instructions, please refer to:
26 Documentation/ABI/testing/sysfs-bus-rbd
30 #include <linux/ceph/libceph.h>
31 #include <linux/ceph/osd_client.h>
32 #include <linux/ceph/mon_client.h>
33 #include <linux/ceph/decode.h>
34 #include <linux/parser.h>
36 #include <linux/kernel.h>
37 #include <linux/device.h>
38 #include <linux/module.h>
40 #include <linux/blkdev.h>
42 #include "rbd_types.h"
44 #define RBD_DEBUG /* Activate rbd_assert() calls */
47 * The basic unit of block I/O is a sector. It is interpreted in a
48 * number of contexts in Linux (blk, bio, genhd), but the default is
49 * universally 512 bytes. These symbols are just slightly more
50 * meaningful than the bare numbers they represent.
52 #define SECTOR_SHIFT 9
53 #define SECTOR_SIZE (1ULL << SECTOR_SHIFT)
55 /* It might be useful to have this defined elsewhere too */
57 #define U64_MAX ((u64) (~0ULL))
59 #define RBD_DRV_NAME "rbd"
60 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
62 #define RBD_MINORS_PER_MAJOR 256 /* max minors per blkdev */
64 #define RBD_SNAP_DEV_NAME_PREFIX "snap_"
65 #define RBD_MAX_SNAP_NAME_LEN \
66 (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
68 #define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */
69 #define RBD_MAX_OPT_LEN 1024
71 #define RBD_SNAP_HEAD_NAME "-"
73 #define RBD_IMAGE_ID_LEN_MAX 64
74 #define RBD_OBJ_PREFIX_LEN_MAX 64
78 #define RBD_FEATURE_LAYERING 1
80 /* Features supported by this (client software) implementation. */
82 #define RBD_FEATURES_ALL (0)
85 * An RBD device name will be "rbd#", where the "rbd" comes from
86 * RBD_DRV_NAME above, and # is a unique integer identifier.
87 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
88 * enough to hold all possible device names.
90 #define DEV_NAME_LEN 32
91 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1)
93 #define RBD_READ_ONLY_DEFAULT false
96 * block device image metadata (in-memory version)
98 struct rbd_image_header
{
99 /* These four fields never change for a given rbd image */
106 /* The remaining fields need to be updated occasionally */
108 struct ceph_snap_context
*snapc
;
120 * an instance of the client. multiple devices may share an rbd client.
123 struct ceph_client
*client
;
125 struct list_head node
;
129 * a request completion status
131 struct rbd_req_status
{
138 * a collection of requests
140 struct rbd_req_coll
{
144 struct rbd_req_status status
[0];
148 * a single io request
151 struct request
*rq
; /* blk layer request */
152 struct bio
*bio
; /* cloned bio */
153 struct page
**pages
; /* list of used pages */
156 struct rbd_req_coll
*coll
;
163 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_client
*rbd_client
;
186 char name
[DEV_NAME_LEN
]; /* blkdev name, e.g. rbd3 */
188 spinlock_t lock
; /* queue lock */
190 struct rbd_image_header header
;
195 size_t image_name_len
;
203 struct ceph_osd_event
*watch_event
;
204 struct ceph_osd_request
*watch_request
;
206 /* protects updating the header */
207 struct rw_semaphore header_rwsem
;
209 struct rbd_mapping mapping
;
211 struct list_head node
;
213 /* list of snapshots */
214 struct list_head snaps
;
220 static DEFINE_MUTEX(ctl_mutex
); /* Serialize open/close/setup/teardown */
222 static LIST_HEAD(rbd_dev_list
); /* devices */
223 static DEFINE_SPINLOCK(rbd_dev_list_lock
);
225 static LIST_HEAD(rbd_client_list
); /* clients */
226 static DEFINE_SPINLOCK(rbd_client_list_lock
);
228 static int rbd_dev_snaps_update(struct rbd_device
*rbd_dev
);
229 static int rbd_dev_snaps_register(struct rbd_device
*rbd_dev
);
231 static void rbd_dev_release(struct device
*dev
);
232 static void rbd_remove_snap_dev(struct rbd_snap
*snap
);
234 static ssize_t
rbd_add(struct bus_type
*bus
, const char *buf
,
236 static ssize_t
rbd_remove(struct bus_type
*bus
, const char *buf
,
239 static struct bus_attribute rbd_bus_attrs
[] = {
240 __ATTR(add
, S_IWUSR
, NULL
, rbd_add
),
241 __ATTR(remove
, S_IWUSR
, NULL
, rbd_remove
),
245 static struct bus_type rbd_bus_type
= {
247 .bus_attrs
= rbd_bus_attrs
,
250 static void rbd_root_dev_release(struct device
*dev
)
254 static struct device rbd_root_dev
= {
256 .release
= rbd_root_dev_release
,
260 #define rbd_assert(expr) \
261 if (unlikely(!(expr))) { \
262 printk(KERN_ERR "\nAssertion failure in %s() " \
264 "\trbd_assert(%s);\n\n", \
265 __func__, __LINE__, #expr); \
268 #else /* !RBD_DEBUG */
269 # define rbd_assert(expr) ((void) 0)
270 #endif /* !RBD_DEBUG */
272 static struct device
*rbd_get_dev(struct rbd_device
*rbd_dev
)
274 return get_device(&rbd_dev
->dev
);
277 static void rbd_put_dev(struct rbd_device
*rbd_dev
)
279 put_device(&rbd_dev
->dev
);
282 static int rbd_dev_refresh(struct rbd_device
*rbd_dev
, u64
*hver
);
283 static int rbd_dev_v2_refresh(struct rbd_device
*rbd_dev
, u64
*hver
);
285 static int rbd_open(struct block_device
*bdev
, fmode_t mode
)
287 struct rbd_device
*rbd_dev
= bdev
->bd_disk
->private_data
;
289 if ((mode
& FMODE_WRITE
) && rbd_dev
->mapping
.read_only
)
292 rbd_get_dev(rbd_dev
);
293 set_device_ro(bdev
, rbd_dev
->mapping
.read_only
);
298 static int rbd_release(struct gendisk
*disk
, fmode_t mode
)
300 struct rbd_device
*rbd_dev
= disk
->private_data
;
302 rbd_put_dev(rbd_dev
);
307 static const struct block_device_operations rbd_bd_ops
= {
308 .owner
= THIS_MODULE
,
310 .release
= rbd_release
,
314 * Initialize an rbd client instance.
317 static struct rbd_client
*rbd_client_create(struct ceph_options
*ceph_opts
)
319 struct rbd_client
*rbdc
;
322 dout("rbd_client_create\n");
323 rbdc
= kmalloc(sizeof(struct rbd_client
), GFP_KERNEL
);
327 kref_init(&rbdc
->kref
);
328 INIT_LIST_HEAD(&rbdc
->node
);
330 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
332 rbdc
->client
= ceph_create_client(ceph_opts
, rbdc
, 0, 0);
333 if (IS_ERR(rbdc
->client
))
335 ceph_opts
= NULL
; /* Now rbdc->client is responsible for ceph_opts */
337 ret
= ceph_open_session(rbdc
->client
);
341 spin_lock(&rbd_client_list_lock
);
342 list_add_tail(&rbdc
->node
, &rbd_client_list
);
343 spin_unlock(&rbd_client_list_lock
);
345 mutex_unlock(&ctl_mutex
);
347 dout("rbd_client_create created %p\n", rbdc
);
351 ceph_destroy_client(rbdc
->client
);
353 mutex_unlock(&ctl_mutex
);
357 ceph_destroy_options(ceph_opts
);
362 * Find a ceph client with specific addr and configuration. If
363 * found, bump its reference count.
365 static struct rbd_client
*rbd_client_find(struct ceph_options
*ceph_opts
)
367 struct rbd_client
*client_node
;
370 if (ceph_opts
->flags
& CEPH_OPT_NOSHARE
)
373 spin_lock(&rbd_client_list_lock
);
374 list_for_each_entry(client_node
, &rbd_client_list
, node
) {
375 if (!ceph_compare_options(ceph_opts
, client_node
->client
)) {
376 kref_get(&client_node
->kref
);
381 spin_unlock(&rbd_client_list_lock
);
383 return found
? client_node
: NULL
;
393 /* string args above */
396 /* Boolean args above */
400 static match_table_t rbd_opts_tokens
= {
402 /* string args above */
403 {Opt_read_only
, "read_only"},
404 {Opt_read_only
, "ro"}, /* Alternate spelling */
405 {Opt_read_write
, "read_write"},
406 {Opt_read_write
, "rw"}, /* Alternate spelling */
407 /* Boolean args above */
411 static int parse_rbd_opts_token(char *c
, void *private)
413 struct rbd_options
*rbd_opts
= private;
414 substring_t argstr
[MAX_OPT_ARGS
];
415 int token
, intval
, ret
;
417 token
= match_token(c
, rbd_opts_tokens
, argstr
);
421 if (token
< Opt_last_int
) {
422 ret
= match_int(&argstr
[0], &intval
);
424 pr_err("bad mount option arg (not int) "
428 dout("got int token %d val %d\n", token
, intval
);
429 } else if (token
> Opt_last_int
&& token
< Opt_last_string
) {
430 dout("got string token %d val %s\n", token
,
432 } else if (token
> Opt_last_string
&& token
< Opt_last_bool
) {
433 dout("got Boolean token %d\n", token
);
435 dout("got token %d\n", token
);
440 rbd_opts
->read_only
= true;
443 rbd_opts
->read_only
= false;
453 * Get a ceph client with specific addr and configuration, if one does
454 * not exist create it.
456 static int rbd_get_client(struct rbd_device
*rbd_dev
,
457 struct ceph_options
*ceph_opts
)
459 struct rbd_client
*rbdc
;
461 rbdc
= rbd_client_find(ceph_opts
);
463 /* using an existing client */
464 ceph_destroy_options(ceph_opts
);
466 rbdc
= rbd_client_create(ceph_opts
);
468 return PTR_ERR(rbdc
);
470 rbd_dev
->rbd_client
= rbdc
;
476 * Destroy ceph client
478 * Caller must hold rbd_client_list_lock.
480 static void rbd_client_release(struct kref
*kref
)
482 struct rbd_client
*rbdc
= container_of(kref
, struct rbd_client
, kref
);
484 dout("rbd_release_client %p\n", rbdc
);
485 spin_lock(&rbd_client_list_lock
);
486 list_del(&rbdc
->node
);
487 spin_unlock(&rbd_client_list_lock
);
489 ceph_destroy_client(rbdc
->client
);
494 * Drop reference to ceph client node. If it's not referenced anymore, release
497 static void rbd_put_client(struct rbd_device
*rbd_dev
)
499 kref_put(&rbd_dev
->rbd_client
->kref
, rbd_client_release
);
500 rbd_dev
->rbd_client
= NULL
;
504 * Destroy requests collection
506 static void rbd_coll_release(struct kref
*kref
)
508 struct rbd_req_coll
*coll
=
509 container_of(kref
, struct rbd_req_coll
, kref
);
511 dout("rbd_coll_release %p\n", coll
);
515 static bool rbd_image_format_valid(u32 image_format
)
517 return image_format
== 1 || image_format
== 2;
520 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk
*ondisk
)
525 /* The header has to start with the magic rbd header text */
526 if (memcmp(&ondisk
->text
, RBD_HEADER_TEXT
, sizeof (RBD_HEADER_TEXT
)))
529 /* The bio layer requires at least sector-sized I/O */
531 if (ondisk
->options
.order
< SECTOR_SHIFT
)
534 /* If we use u64 in a few spots we may be able to loosen this */
536 if (ondisk
->options
.order
> 8 * sizeof (int) - 1)
540 * The size of a snapshot header has to fit in a size_t, and
541 * that limits the number of snapshots.
543 snap_count
= le32_to_cpu(ondisk
->snap_count
);
544 size
= SIZE_MAX
- sizeof (struct ceph_snap_context
);
545 if (snap_count
> size
/ sizeof (__le64
))
549 * Not only that, but the size of the entire the snapshot
550 * header must also be representable in a size_t.
552 size
-= snap_count
* sizeof (__le64
);
553 if ((u64
) size
< le64_to_cpu(ondisk
->snap_names_len
))
560 * Create a new header structure, translate header format from the on-disk
563 static int rbd_header_from_disk(struct rbd_image_header
*header
,
564 struct rbd_image_header_ondisk
*ondisk
)
571 memset(header
, 0, sizeof (*header
));
573 snap_count
= le32_to_cpu(ondisk
->snap_count
);
575 len
= strnlen(ondisk
->object_prefix
, sizeof (ondisk
->object_prefix
));
576 header
->object_prefix
= kmalloc(len
+ 1, GFP_KERNEL
);
577 if (!header
->object_prefix
)
579 memcpy(header
->object_prefix
, ondisk
->object_prefix
, len
);
580 header
->object_prefix
[len
] = '\0';
583 u64 snap_names_len
= le64_to_cpu(ondisk
->snap_names_len
);
585 /* Save a copy of the snapshot names */
587 if (snap_names_len
> (u64
) SIZE_MAX
)
589 header
->snap_names
= kmalloc(snap_names_len
, GFP_KERNEL
);
590 if (!header
->snap_names
)
593 * Note that rbd_dev_v1_header_read() guarantees
594 * the ondisk buffer we're working with has
595 * snap_names_len bytes beyond the end of the
596 * snapshot id array, this memcpy() is safe.
598 memcpy(header
->snap_names
, &ondisk
->snaps
[snap_count
],
601 /* Record each snapshot's size */
603 size
= snap_count
* sizeof (*header
->snap_sizes
);
604 header
->snap_sizes
= kmalloc(size
, GFP_KERNEL
);
605 if (!header
->snap_sizes
)
607 for (i
= 0; i
< snap_count
; i
++)
608 header
->snap_sizes
[i
] =
609 le64_to_cpu(ondisk
->snaps
[i
].image_size
);
611 WARN_ON(ondisk
->snap_names_len
);
612 header
->snap_names
= NULL
;
613 header
->snap_sizes
= NULL
;
616 header
->features
= 0; /* No features support in v1 images */
617 header
->obj_order
= ondisk
->options
.order
;
618 header
->crypt_type
= ondisk
->options
.crypt_type
;
619 header
->comp_type
= ondisk
->options
.comp_type
;
621 /* Allocate and fill in the snapshot context */
623 header
->image_size
= le64_to_cpu(ondisk
->image_size
);
624 size
= sizeof (struct ceph_snap_context
);
625 size
+= snap_count
* sizeof (header
->snapc
->snaps
[0]);
626 header
->snapc
= kzalloc(size
, GFP_KERNEL
);
630 atomic_set(&header
->snapc
->nref
, 1);
631 header
->snapc
->seq
= le64_to_cpu(ondisk
->snap_seq
);
632 header
->snapc
->num_snaps
= snap_count
;
633 for (i
= 0; i
< snap_count
; i
++)
634 header
->snapc
->snaps
[i
] =
635 le64_to_cpu(ondisk
->snaps
[i
].id
);
640 kfree(header
->snap_sizes
);
641 header
->snap_sizes
= NULL
;
642 kfree(header
->snap_names
);
643 header
->snap_names
= NULL
;
644 kfree(header
->object_prefix
);
645 header
->object_prefix
= NULL
;
650 static int snap_by_name(struct rbd_device
*rbd_dev
, const char *snap_name
)
653 struct rbd_snap
*snap
;
655 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
656 if (!strcmp(snap_name
, snap
->name
)) {
657 rbd_dev
->snap_id
= snap
->id
;
658 rbd_dev
->mapping
.size
= snap
->size
;
659 rbd_dev
->mapping
.features
= snap
->features
;
668 static int rbd_dev_set_mapping(struct rbd_device
*rbd_dev
, char *snap_name
)
672 if (!memcmp(snap_name
, RBD_SNAP_HEAD_NAME
,
673 sizeof (RBD_SNAP_HEAD_NAME
))) {
674 rbd_dev
->snap_id
= CEPH_NOSNAP
;
675 rbd_dev
->mapping
.size
= rbd_dev
->header
.image_size
;
676 rbd_dev
->mapping
.features
= rbd_dev
->header
.features
;
679 ret
= snap_by_name(rbd_dev
, snap_name
);
682 rbd_dev
->mapping
.read_only
= true;
684 rbd_dev
->snap_name
= snap_name
;
685 rbd_dev
->exists
= true;
690 static void rbd_header_free(struct rbd_image_header
*header
)
692 kfree(header
->object_prefix
);
693 header
->object_prefix
= NULL
;
694 kfree(header
->snap_sizes
);
695 header
->snap_sizes
= NULL
;
696 kfree(header
->snap_names
);
697 header
->snap_names
= NULL
;
698 ceph_put_snap_context(header
->snapc
);
699 header
->snapc
= NULL
;
702 static char *rbd_segment_name(struct rbd_device
*rbd_dev
, u64 offset
)
708 name
= kmalloc(RBD_MAX_SEG_NAME_LEN
+ 1, GFP_NOIO
);
711 segment
= offset
>> rbd_dev
->header
.obj_order
;
712 ret
= snprintf(name
, RBD_MAX_SEG_NAME_LEN
, "%s.%012llx",
713 rbd_dev
->header
.object_prefix
, segment
);
714 if (ret
< 0 || ret
>= RBD_MAX_SEG_NAME_LEN
) {
715 pr_err("error formatting segment name for #%llu (%d)\n",
724 static u64
rbd_segment_offset(struct rbd_device
*rbd_dev
, u64 offset
)
726 u64 segment_size
= (u64
) 1 << rbd_dev
->header
.obj_order
;
728 return offset
& (segment_size
- 1);
731 static u64
rbd_segment_length(struct rbd_device
*rbd_dev
,
732 u64 offset
, u64 length
)
734 u64 segment_size
= (u64
) 1 << rbd_dev
->header
.obj_order
;
736 offset
&= segment_size
- 1;
738 rbd_assert(length
<= U64_MAX
- offset
);
739 if (offset
+ length
> segment_size
)
740 length
= segment_size
- offset
;
745 static int rbd_get_num_segments(struct rbd_image_header
*header
,
753 if (len
- 1 > U64_MAX
- ofs
)
756 start_seg
= ofs
>> header
->obj_order
;
757 end_seg
= (ofs
+ len
- 1) >> header
->obj_order
;
759 return end_seg
- start_seg
+ 1;
763 * returns the size of an object in the image
765 static u64
rbd_obj_bytes(struct rbd_image_header
*header
)
767 return 1 << header
->obj_order
;
774 static void bio_chain_put(struct bio
*chain
)
780 chain
= chain
->bi_next
;
786 * zeros a bio chain, starting at specific offset
788 static void zero_bio_chain(struct bio
*chain
, int start_ofs
)
797 bio_for_each_segment(bv
, chain
, i
) {
798 if (pos
+ bv
->bv_len
> start_ofs
) {
799 int remainder
= max(start_ofs
- pos
, 0);
800 buf
= bvec_kmap_irq(bv
, &flags
);
801 memset(buf
+ remainder
, 0,
802 bv
->bv_len
- remainder
);
803 bvec_kunmap_irq(buf
, &flags
);
808 chain
= chain
->bi_next
;
813 * Clone a portion of a bio, starting at the given byte offset
814 * and continuing for the number of bytes indicated.
816 static struct bio
*bio_clone_range(struct bio
*bio_src
,
825 unsigned short end_idx
;
829 /* Handle the easy case for the caller */
831 if (!offset
&& len
== bio_src
->bi_size
)
832 return bio_clone(bio_src
, gfpmask
);
834 if (WARN_ON_ONCE(!len
))
836 if (WARN_ON_ONCE(len
> bio_src
->bi_size
))
838 if (WARN_ON_ONCE(offset
> bio_src
->bi_size
- len
))
841 /* Find first affected segment... */
844 __bio_for_each_segment(bv
, bio_src
, idx
, 0) {
845 if (resid
< bv
->bv_len
)
851 /* ...and the last affected segment */
854 __bio_for_each_segment(bv
, bio_src
, end_idx
, idx
) {
855 if (resid
<= bv
->bv_len
)
859 vcnt
= end_idx
- idx
+ 1;
861 /* Build the clone */
863 bio
= bio_alloc(gfpmask
, (unsigned int) vcnt
);
865 return NULL
; /* ENOMEM */
867 bio
->bi_bdev
= bio_src
->bi_bdev
;
868 bio
->bi_sector
= bio_src
->bi_sector
+ (offset
>> SECTOR_SHIFT
);
869 bio
->bi_rw
= bio_src
->bi_rw
;
870 bio
->bi_flags
|= 1 << BIO_CLONED
;
873 * Copy over our part of the bio_vec, then update the first
874 * and last (or only) entries.
876 memcpy(&bio
->bi_io_vec
[0], &bio_src
->bi_io_vec
[idx
],
877 vcnt
* sizeof (struct bio_vec
));
878 bio
->bi_io_vec
[0].bv_offset
+= voff
;
880 bio
->bi_io_vec
[0].bv_len
-= voff
;
881 bio
->bi_io_vec
[vcnt
- 1].bv_len
= resid
;
883 bio
->bi_io_vec
[0].bv_len
= len
;
894 * Clone a portion of a bio chain, starting at the given byte offset
895 * into the first bio in the source chain and continuing for the
896 * number of bytes indicated. The result is another bio chain of
897 * exactly the given length, or a null pointer on error.
899 * The bio_src and offset parameters are both in-out. On entry they
900 * refer to the first source bio and the offset into that bio where
901 * the start of data to be cloned is located.
903 * On return, bio_src is updated to refer to the bio in the source
904 * chain that contains first un-cloned byte, and *offset will
905 * contain the offset of that byte within that bio.
907 static struct bio
*bio_chain_clone_range(struct bio
**bio_src
,
908 unsigned int *offset
,
912 struct bio
*bi
= *bio_src
;
913 unsigned int off
= *offset
;
914 struct bio
*chain
= NULL
;
917 /* Build up a chain of clone bios up to the limit */
919 if (!bi
|| off
>= bi
->bi_size
|| !len
)
920 return NULL
; /* Nothing to clone */
924 unsigned int bi_size
;
928 goto out_err
; /* EINVAL; ran out of bio's */
929 bi_size
= min_t(unsigned int, bi
->bi_size
- off
, len
);
930 bio
= bio_clone_range(bi
, off
, bi_size
, gfpmask
);
932 goto out_err
; /* ENOMEM */
938 if (off
== bi
->bi_size
) {
949 bio_chain_put(chain
);
955 * helpers for osd request op vectors.
957 static struct ceph_osd_req_op
*rbd_create_rw_ops(int num_ops
,
958 int opcode
, u32 payload_len
)
960 struct ceph_osd_req_op
*ops
;
962 ops
= kzalloc(sizeof (*ops
) * (num_ops
+ 1), GFP_NOIO
);
969 * op extent offset and length will be set later on
970 * in calc_raw_layout()
972 ops
[0].payload_len
= payload_len
;
977 static void rbd_destroy_ops(struct ceph_osd_req_op
*ops
)
982 static void rbd_coll_end_req_index(struct request
*rq
,
983 struct rbd_req_coll
*coll
,
987 struct request_queue
*q
;
990 dout("rbd_coll_end_req_index %p index %d ret %d len %llu\n",
991 coll
, index
, ret
, (unsigned long long) len
);
997 blk_end_request(rq
, ret
, len
);
1003 spin_lock_irq(q
->queue_lock
);
1004 coll
->status
[index
].done
= 1;
1005 coll
->status
[index
].rc
= ret
;
1006 coll
->status
[index
].bytes
= len
;
1007 max
= min
= coll
->num_done
;
1008 while (max
< coll
->total
&& coll
->status
[max
].done
)
1011 for (i
= min
; i
<max
; i
++) {
1012 __blk_end_request(rq
, coll
->status
[i
].rc
,
1013 coll
->status
[i
].bytes
);
1015 kref_put(&coll
->kref
, rbd_coll_release
);
1017 spin_unlock_irq(q
->queue_lock
);
1020 static void rbd_coll_end_req(struct rbd_request
*req
,
1023 rbd_coll_end_req_index(req
->rq
, req
->coll
, req
->coll_index
, ret
, len
);
1027 * Send ceph osd request
1029 static int rbd_do_request(struct request
*rq
,
1030 struct rbd_device
*rbd_dev
,
1031 struct ceph_snap_context
*snapc
,
1033 const char *object_name
, u64 ofs
, u64 len
,
1035 struct page
**pages
,
1038 struct ceph_osd_req_op
*ops
,
1039 struct rbd_req_coll
*coll
,
1041 void (*rbd_cb
)(struct ceph_osd_request
*req
,
1042 struct ceph_msg
*msg
),
1043 struct ceph_osd_request
**linger_req
,
1046 struct ceph_osd_request
*req
;
1047 struct ceph_file_layout
*layout
;
1050 struct timespec mtime
= CURRENT_TIME
;
1051 struct rbd_request
*req_data
;
1052 struct ceph_osd_request_head
*reqhead
;
1053 struct ceph_osd_client
*osdc
;
1055 req_data
= kzalloc(sizeof(*req_data
), GFP_NOIO
);
1058 rbd_coll_end_req_index(rq
, coll
, coll_index
,
1064 req_data
->coll
= coll
;
1065 req_data
->coll_index
= coll_index
;
1068 dout("rbd_do_request object_name=%s ofs=%llu len=%llu coll=%p[%d]\n",
1069 object_name
, (unsigned long long) ofs
,
1070 (unsigned long long) len
, coll
, coll_index
);
1072 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1073 req
= ceph_osdc_alloc_request(osdc
, flags
, snapc
, ops
,
1074 false, GFP_NOIO
, pages
, bio
);
1080 req
->r_callback
= rbd_cb
;
1083 req_data
->bio
= bio
;
1084 req_data
->pages
= pages
;
1085 req_data
->len
= len
;
1087 req
->r_priv
= req_data
;
1089 reqhead
= req
->r_request
->front
.iov_base
;
1090 reqhead
->snapid
= cpu_to_le64(CEPH_NOSNAP
);
1092 strncpy(req
->r_oid
, object_name
, sizeof(req
->r_oid
));
1093 req
->r_oid_len
= strlen(req
->r_oid
);
1095 layout
= &req
->r_file_layout
;
1096 memset(layout
, 0, sizeof(*layout
));
1097 layout
->fl_stripe_unit
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
1098 layout
->fl_stripe_count
= cpu_to_le32(1);
1099 layout
->fl_object_size
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
1100 layout
->fl_pg_pool
= cpu_to_le32((int) rbd_dev
->pool_id
);
1101 ret
= ceph_calc_raw_layout(osdc
, layout
, snapid
, ofs
, &len
, &bno
,
1103 rbd_assert(ret
== 0);
1105 ceph_osdc_build_request(req
, ofs
, &len
,
1109 req
->r_oid
, req
->r_oid_len
);
1112 ceph_osdc_set_request_linger(osdc
, req
);
1116 ret
= ceph_osdc_start_request(osdc
, req
, false);
1121 ret
= ceph_osdc_wait_request(osdc
, req
);
1123 *ver
= le64_to_cpu(req
->r_reassert_version
.version
);
1124 dout("reassert_ver=%llu\n",
1125 (unsigned long long)
1126 le64_to_cpu(req
->r_reassert_version
.version
));
1127 ceph_osdc_put_request(req
);
1132 bio_chain_put(req_data
->bio
);
1133 ceph_osdc_put_request(req
);
1135 rbd_coll_end_req(req_data
, ret
, len
);
1141 * Ceph osd op callback
1143 static void rbd_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
1145 struct rbd_request
*req_data
= req
->r_priv
;
1146 struct ceph_osd_reply_head
*replyhead
;
1147 struct ceph_osd_op
*op
;
1153 replyhead
= msg
->front
.iov_base
;
1154 WARN_ON(le32_to_cpu(replyhead
->num_ops
) == 0);
1155 op
= (void *)(replyhead
+ 1);
1156 rc
= le32_to_cpu(replyhead
->result
);
1157 bytes
= le64_to_cpu(op
->extent
.length
);
1158 read_op
= (le16_to_cpu(op
->op
) == CEPH_OSD_OP_READ
);
1160 dout("rbd_req_cb bytes=%llu readop=%d rc=%d\n",
1161 (unsigned long long) bytes
, read_op
, (int) rc
);
1163 if (rc
== -ENOENT
&& read_op
) {
1164 zero_bio_chain(req_data
->bio
, 0);
1166 } else if (rc
== 0 && read_op
&& bytes
< req_data
->len
) {
1167 zero_bio_chain(req_data
->bio
, bytes
);
1168 bytes
= req_data
->len
;
1171 rbd_coll_end_req(req_data
, rc
, bytes
);
1174 bio_chain_put(req_data
->bio
);
1176 ceph_osdc_put_request(req
);
1180 static void rbd_simple_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
1182 ceph_osdc_put_request(req
);
1186 * Do a synchronous ceph osd operation
1188 static int rbd_req_sync_op(struct rbd_device
*rbd_dev
,
1189 struct ceph_snap_context
*snapc
,
1192 struct ceph_osd_req_op
*ops
,
1193 const char *object_name
,
1194 u64 ofs
, u64 inbound_size
,
1196 struct ceph_osd_request
**linger_req
,
1200 struct page
**pages
;
1203 rbd_assert(ops
!= NULL
);
1205 num_pages
= calc_pages_for(ofs
, inbound_size
);
1206 pages
= ceph_alloc_page_vector(num_pages
, GFP_KERNEL
);
1208 return PTR_ERR(pages
);
1210 ret
= rbd_do_request(NULL
, rbd_dev
, snapc
, snapid
,
1211 object_name
, ofs
, inbound_size
, NULL
,
1221 if ((flags
& CEPH_OSD_FLAG_READ
) && inbound
)
1222 ret
= ceph_copy_from_page_vector(pages
, inbound
, ofs
, ret
);
1225 ceph_release_page_vector(pages
, num_pages
);
1230 * Do an asynchronous ceph osd operation
1232 static int rbd_do_op(struct request
*rq
,
1233 struct rbd_device
*rbd_dev
,
1234 struct ceph_snap_context
*snapc
,
1237 struct rbd_req_coll
*coll
,
1244 struct ceph_osd_req_op
*ops
;
1250 seg_name
= rbd_segment_name(rbd_dev
, ofs
);
1253 seg_len
= rbd_segment_length(rbd_dev
, ofs
, len
);
1254 seg_ofs
= rbd_segment_offset(rbd_dev
, ofs
);
1256 if (rq_data_dir(rq
) == WRITE
) {
1257 opcode
= CEPH_OSD_OP_WRITE
;
1258 flags
= CEPH_OSD_FLAG_WRITE
|CEPH_OSD_FLAG_ONDISK
;
1259 snapid
= CEPH_NOSNAP
;
1260 payload_len
= seg_len
;
1262 opcode
= CEPH_OSD_OP_READ
;
1263 flags
= CEPH_OSD_FLAG_READ
;
1265 snapid
= rbd_dev
->snap_id
;
1270 ops
= rbd_create_rw_ops(1, opcode
, payload_len
);
1274 /* we've taken care of segment sizes earlier when we
1275 cloned the bios. We should never have a segment
1276 truncated at this point */
1277 rbd_assert(seg_len
== len
);
1279 ret
= rbd_do_request(rq
, rbd_dev
, snapc
, snapid
,
1280 seg_name
, seg_ofs
, seg_len
,
1286 rbd_req_cb
, 0, NULL
);
1288 rbd_destroy_ops(ops
);
1295 * Request sync osd read
1297 static int rbd_req_sync_read(struct rbd_device
*rbd_dev
,
1299 const char *object_name
,
1304 struct ceph_osd_req_op
*ops
;
1307 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_READ
, 0);
1311 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1314 ops
, object_name
, ofs
, len
, buf
, NULL
, ver
);
1315 rbd_destroy_ops(ops
);
1321 * Request sync osd watch
1323 static int rbd_req_sync_notify_ack(struct rbd_device
*rbd_dev
,
1327 struct ceph_osd_req_op
*ops
;
1330 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_NOTIFY_ACK
, 0);
1334 ops
[0].watch
.ver
= cpu_to_le64(ver
);
1335 ops
[0].watch
.cookie
= notify_id
;
1336 ops
[0].watch
.flag
= 0;
1338 ret
= rbd_do_request(NULL
, rbd_dev
, NULL
, CEPH_NOSNAP
,
1339 rbd_dev
->header_name
, 0, 0, NULL
,
1344 rbd_simple_req_cb
, 0, NULL
);
1346 rbd_destroy_ops(ops
);
1350 static void rbd_watch_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1352 struct rbd_device
*rbd_dev
= (struct rbd_device
*)data
;
1359 dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
1360 rbd_dev
->header_name
, (unsigned long long) notify_id
,
1361 (unsigned int) opcode
);
1362 rc
= rbd_dev_refresh(rbd_dev
, &hver
);
1364 pr_warning(RBD_DRV_NAME
"%d got notification but failed to "
1365 " update snaps: %d\n", rbd_dev
->major
, rc
);
1367 rbd_req_sync_notify_ack(rbd_dev
, hver
, notify_id
);
1371 * Request sync osd watch
1373 static int rbd_req_sync_watch(struct rbd_device
*rbd_dev
)
1375 struct ceph_osd_req_op
*ops
;
1376 struct ceph_osd_client
*osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1379 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_WATCH
, 0);
1383 ret
= ceph_osdc_create_event(osdc
, rbd_watch_cb
, 0,
1384 (void *)rbd_dev
, &rbd_dev
->watch_event
);
1388 ops
[0].watch
.ver
= cpu_to_le64(rbd_dev
->header
.obj_version
);
1389 ops
[0].watch
.cookie
= cpu_to_le64(rbd_dev
->watch_event
->cookie
);
1390 ops
[0].watch
.flag
= 1;
1392 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1394 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1396 rbd_dev
->header_name
,
1398 &rbd_dev
->watch_request
, NULL
);
1403 rbd_destroy_ops(ops
);
1407 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1408 rbd_dev
->watch_event
= NULL
;
1410 rbd_destroy_ops(ops
);
1415 * Request sync osd unwatch
1417 static int rbd_req_sync_unwatch(struct rbd_device
*rbd_dev
)
1419 struct ceph_osd_req_op
*ops
;
1422 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_WATCH
, 0);
1426 ops
[0].watch
.ver
= 0;
1427 ops
[0].watch
.cookie
= cpu_to_le64(rbd_dev
->watch_event
->cookie
);
1428 ops
[0].watch
.flag
= 0;
1430 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1432 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1434 rbd_dev
->header_name
,
1435 0, 0, NULL
, NULL
, NULL
);
1438 rbd_destroy_ops(ops
);
1439 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1440 rbd_dev
->watch_event
= NULL
;
1445 * Synchronous osd object method call
1447 static int rbd_req_sync_exec(struct rbd_device
*rbd_dev
,
1448 const char *object_name
,
1449 const char *class_name
,
1450 const char *method_name
,
1451 const char *outbound
,
1452 size_t outbound_size
,
1454 size_t inbound_size
,
1458 struct ceph_osd_req_op
*ops
;
1459 int class_name_len
= strlen(class_name
);
1460 int method_name_len
= strlen(method_name
);
1465 * Any input parameters required by the method we're calling
1466 * will be sent along with the class and method names as
1467 * part of the message payload. That data and its size are
1468 * supplied via the indata and indata_len fields (named from
1469 * the perspective of the server side) in the OSD request
1472 payload_size
= class_name_len
+ method_name_len
+ outbound_size
;
1473 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_CALL
, payload_size
);
1477 ops
[0].cls
.class_name
= class_name
;
1478 ops
[0].cls
.class_len
= (__u8
) class_name_len
;
1479 ops
[0].cls
.method_name
= method_name
;
1480 ops
[0].cls
.method_len
= (__u8
) method_name_len
;
1481 ops
[0].cls
.argc
= 0;
1482 ops
[0].cls
.indata
= outbound
;
1483 ops
[0].cls
.indata_len
= outbound_size
;
1485 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1488 object_name
, 0, inbound_size
, inbound
,
1491 rbd_destroy_ops(ops
);
1493 dout("cls_exec returned %d\n", ret
);
1497 static struct rbd_req_coll
*rbd_alloc_coll(int num_reqs
)
1499 struct rbd_req_coll
*coll
=
1500 kzalloc(sizeof(struct rbd_req_coll
) +
1501 sizeof(struct rbd_req_status
) * num_reqs
,
1506 coll
->total
= num_reqs
;
1507 kref_init(&coll
->kref
);
1512 * block device queue callback
1514 static void rbd_rq_fn(struct request_queue
*q
)
1516 struct rbd_device
*rbd_dev
= q
->queuedata
;
1519 while ((rq
= blk_fetch_request(q
))) {
1524 int num_segs
, cur_seg
= 0;
1525 struct rbd_req_coll
*coll
;
1526 struct ceph_snap_context
*snapc
;
1527 unsigned int bio_offset
;
1529 dout("fetched request\n");
1531 /* filter out block requests we don't understand */
1532 if ((rq
->cmd_type
!= REQ_TYPE_FS
)) {
1533 __blk_end_request_all(rq
, 0);
1537 /* deduce our operation (read, write) */
1538 do_write
= (rq_data_dir(rq
) == WRITE
);
1539 if (do_write
&& rbd_dev
->mapping
.read_only
) {
1540 __blk_end_request_all(rq
, -EROFS
);
1544 spin_unlock_irq(q
->queue_lock
);
1546 down_read(&rbd_dev
->header_rwsem
);
1548 if (!rbd_dev
->exists
) {
1549 rbd_assert(rbd_dev
->snap_id
!= CEPH_NOSNAP
);
1550 up_read(&rbd_dev
->header_rwsem
);
1551 dout("request for non-existent snapshot");
1552 spin_lock_irq(q
->queue_lock
);
1553 __blk_end_request_all(rq
, -ENXIO
);
1557 snapc
= ceph_get_snap_context(rbd_dev
->header
.snapc
);
1559 up_read(&rbd_dev
->header_rwsem
);
1561 size
= blk_rq_bytes(rq
);
1562 ofs
= blk_rq_pos(rq
) * SECTOR_SIZE
;
1565 dout("%s 0x%x bytes at 0x%llx\n",
1566 do_write
? "write" : "read",
1567 size
, (unsigned long long) blk_rq_pos(rq
) * SECTOR_SIZE
);
1569 num_segs
= rbd_get_num_segments(&rbd_dev
->header
, ofs
, size
);
1570 if (num_segs
<= 0) {
1571 spin_lock_irq(q
->queue_lock
);
1572 __blk_end_request_all(rq
, num_segs
);
1573 ceph_put_snap_context(snapc
);
1576 coll
= rbd_alloc_coll(num_segs
);
1578 spin_lock_irq(q
->queue_lock
);
1579 __blk_end_request_all(rq
, -ENOMEM
);
1580 ceph_put_snap_context(snapc
);
1586 u64 limit
= rbd_segment_length(rbd_dev
, ofs
, size
);
1587 unsigned int chain_size
;
1588 struct bio
*bio_chain
;
1590 BUG_ON(limit
> (u64
) UINT_MAX
);
1591 chain_size
= (unsigned int) limit
;
1592 dout("rq->bio->bi_vcnt=%hu\n", rq
->bio
->bi_vcnt
);
1594 kref_get(&coll
->kref
);
1596 /* Pass a cloned bio chain via an osd request */
1598 bio_chain
= bio_chain_clone_range(&bio
,
1599 &bio_offset
, chain_size
,
1602 (void) rbd_do_op(rq
, rbd_dev
, snapc
,
1604 bio_chain
, coll
, cur_seg
);
1606 rbd_coll_end_req_index(rq
, coll
, cur_seg
,
1607 -ENOMEM
, chain_size
);
1613 kref_put(&coll
->kref
, rbd_coll_release
);
1615 spin_lock_irq(q
->queue_lock
);
1617 ceph_put_snap_context(snapc
);
1622 * a queue callback. Makes sure that we don't create a bio that spans across
1623 * multiple osd objects. One exception would be with a single page bios,
1624 * which we handle later at bio_chain_clone_range()
1626 static int rbd_merge_bvec(struct request_queue
*q
, struct bvec_merge_data
*bmd
,
1627 struct bio_vec
*bvec
)
1629 struct rbd_device
*rbd_dev
= q
->queuedata
;
1630 sector_t sector_offset
;
1631 sector_t sectors_per_obj
;
1632 sector_t obj_sector_offset
;
1636 * Find how far into its rbd object the partition-relative
1637 * bio start sector is to offset relative to the enclosing
1640 sector_offset
= get_start_sect(bmd
->bi_bdev
) + bmd
->bi_sector
;
1641 sectors_per_obj
= 1 << (rbd_dev
->header
.obj_order
- SECTOR_SHIFT
);
1642 obj_sector_offset
= sector_offset
& (sectors_per_obj
- 1);
1645 * Compute the number of bytes from that offset to the end
1646 * of the object. Account for what's already used by the bio.
1648 ret
= (int) (sectors_per_obj
- obj_sector_offset
) << SECTOR_SHIFT
;
1649 if (ret
> bmd
->bi_size
)
1650 ret
-= bmd
->bi_size
;
1655 * Don't send back more than was asked for. And if the bio
1656 * was empty, let the whole thing through because: "Note
1657 * that a block device *must* allow a single page to be
1658 * added to an empty bio."
1660 rbd_assert(bvec
->bv_len
<= PAGE_SIZE
);
1661 if (ret
> (int) bvec
->bv_len
|| !bmd
->bi_size
)
1662 ret
= (int) bvec
->bv_len
;
1667 static void rbd_free_disk(struct rbd_device
*rbd_dev
)
1669 struct gendisk
*disk
= rbd_dev
->disk
;
1674 if (disk
->flags
& GENHD_FL_UP
)
1677 blk_cleanup_queue(disk
->queue
);
1682 * Read the complete header for the given rbd device.
1684 * Returns a pointer to a dynamically-allocated buffer containing
1685 * the complete and validated header. Caller can pass the address
1686 * of a variable that will be filled in with the version of the
1687 * header object at the time it was read.
1689 * Returns a pointer-coded errno if a failure occurs.
1691 static struct rbd_image_header_ondisk
*
1692 rbd_dev_v1_header_read(struct rbd_device
*rbd_dev
, u64
*version
)
1694 struct rbd_image_header_ondisk
*ondisk
= NULL
;
1701 * The complete header will include an array of its 64-bit
1702 * snapshot ids, followed by the names of those snapshots as
1703 * a contiguous block of NUL-terminated strings. Note that
1704 * the number of snapshots could change by the time we read
1705 * it in, in which case we re-read it.
1712 size
= sizeof (*ondisk
);
1713 size
+= snap_count
* sizeof (struct rbd_image_snap_ondisk
);
1715 ondisk
= kmalloc(size
, GFP_KERNEL
);
1717 return ERR_PTR(-ENOMEM
);
1719 ret
= rbd_req_sync_read(rbd_dev
, CEPH_NOSNAP
,
1720 rbd_dev
->header_name
,
1722 (char *) ondisk
, version
);
1726 if (WARN_ON((size_t) ret
< size
)) {
1728 pr_warning("short header read for image %s"
1729 " (want %zd got %d)\n",
1730 rbd_dev
->image_name
, size
, ret
);
1733 if (!rbd_dev_ondisk_valid(ondisk
)) {
1735 pr_warning("invalid header for image %s\n",
1736 rbd_dev
->image_name
);
1740 names_size
= le64_to_cpu(ondisk
->snap_names_len
);
1741 want_count
= snap_count
;
1742 snap_count
= le32_to_cpu(ondisk
->snap_count
);
1743 } while (snap_count
!= want_count
);
1750 return ERR_PTR(ret
);
1754 * reload the ondisk the header
1756 static int rbd_read_header(struct rbd_device
*rbd_dev
,
1757 struct rbd_image_header
*header
)
1759 struct rbd_image_header_ondisk
*ondisk
;
1763 ondisk
= rbd_dev_v1_header_read(rbd_dev
, &ver
);
1765 return PTR_ERR(ondisk
);
1766 ret
= rbd_header_from_disk(header
, ondisk
);
1768 header
->obj_version
= ver
;
1774 static void rbd_remove_all_snaps(struct rbd_device
*rbd_dev
)
1776 struct rbd_snap
*snap
;
1777 struct rbd_snap
*next
;
1779 list_for_each_entry_safe(snap
, next
, &rbd_dev
->snaps
, node
)
1780 rbd_remove_snap_dev(snap
);
1783 static void rbd_update_mapping_size(struct rbd_device
*rbd_dev
)
1787 if (rbd_dev
->snap_id
!= CEPH_NOSNAP
)
1790 size
= (sector_t
) rbd_dev
->header
.image_size
/ SECTOR_SIZE
;
1791 dout("setting size to %llu sectors", (unsigned long long) size
);
1792 rbd_dev
->mapping
.size
= (u64
) size
;
1793 set_capacity(rbd_dev
->disk
, size
);
1797 * only read the first part of the ondisk header, without the snaps info
1799 static int rbd_dev_v1_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
1802 struct rbd_image_header h
;
1804 ret
= rbd_read_header(rbd_dev
, &h
);
1808 down_write(&rbd_dev
->header_rwsem
);
1810 /* Update image size, and check for resize of mapped image */
1811 rbd_dev
->header
.image_size
= h
.image_size
;
1812 rbd_update_mapping_size(rbd_dev
);
1814 /* rbd_dev->header.object_prefix shouldn't change */
1815 kfree(rbd_dev
->header
.snap_sizes
);
1816 kfree(rbd_dev
->header
.snap_names
);
1817 /* osd requests may still refer to snapc */
1818 ceph_put_snap_context(rbd_dev
->header
.snapc
);
1821 *hver
= h
.obj_version
;
1822 rbd_dev
->header
.obj_version
= h
.obj_version
;
1823 rbd_dev
->header
.image_size
= h
.image_size
;
1824 rbd_dev
->header
.snapc
= h
.snapc
;
1825 rbd_dev
->header
.snap_names
= h
.snap_names
;
1826 rbd_dev
->header
.snap_sizes
= h
.snap_sizes
;
1827 /* Free the extra copy of the object prefix */
1828 WARN_ON(strcmp(rbd_dev
->header
.object_prefix
, h
.object_prefix
));
1829 kfree(h
.object_prefix
);
1831 ret
= rbd_dev_snaps_update(rbd_dev
);
1833 ret
= rbd_dev_snaps_register(rbd_dev
);
1835 up_write(&rbd_dev
->header_rwsem
);
1840 static int rbd_dev_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
1844 rbd_assert(rbd_image_format_valid(rbd_dev
->image_format
));
1845 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1846 if (rbd_dev
->image_format
== 1)
1847 ret
= rbd_dev_v1_refresh(rbd_dev
, hver
);
1849 ret
= rbd_dev_v2_refresh(rbd_dev
, hver
);
1850 mutex_unlock(&ctl_mutex
);
1855 static int rbd_init_disk(struct rbd_device
*rbd_dev
)
1857 struct gendisk
*disk
;
1858 struct request_queue
*q
;
1861 /* create gendisk info */
1862 disk
= alloc_disk(RBD_MINORS_PER_MAJOR
);
1866 snprintf(disk
->disk_name
, sizeof(disk
->disk_name
), RBD_DRV_NAME
"%d",
1868 disk
->major
= rbd_dev
->major
;
1869 disk
->first_minor
= 0;
1870 disk
->fops
= &rbd_bd_ops
;
1871 disk
->private_data
= rbd_dev
;
1874 q
= blk_init_queue(rbd_rq_fn
, &rbd_dev
->lock
);
1878 /* We use the default size, but let's be explicit about it. */
1879 blk_queue_physical_block_size(q
, SECTOR_SIZE
);
1881 /* set io sizes to object size */
1882 segment_size
= rbd_obj_bytes(&rbd_dev
->header
);
1883 blk_queue_max_hw_sectors(q
, segment_size
/ SECTOR_SIZE
);
1884 blk_queue_max_segment_size(q
, segment_size
);
1885 blk_queue_io_min(q
, segment_size
);
1886 blk_queue_io_opt(q
, segment_size
);
1888 blk_queue_merge_bvec(q
, rbd_merge_bvec
);
1891 q
->queuedata
= rbd_dev
;
1893 rbd_dev
->disk
= disk
;
1895 set_capacity(rbd_dev
->disk
, rbd_dev
->mapping
.size
/ SECTOR_SIZE
);
1908 static struct rbd_device
*dev_to_rbd_dev(struct device
*dev
)
1910 return container_of(dev
, struct rbd_device
, dev
);
1913 static ssize_t
rbd_size_show(struct device
*dev
,
1914 struct device_attribute
*attr
, char *buf
)
1916 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1919 down_read(&rbd_dev
->header_rwsem
);
1920 size
= get_capacity(rbd_dev
->disk
);
1921 up_read(&rbd_dev
->header_rwsem
);
1923 return sprintf(buf
, "%llu\n", (unsigned long long) size
* SECTOR_SIZE
);
1927 * Note this shows the features for whatever's mapped, which is not
1928 * necessarily the base image.
1930 static ssize_t
rbd_features_show(struct device
*dev
,
1931 struct device_attribute
*attr
, char *buf
)
1933 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1935 return sprintf(buf
, "0x%016llx\n",
1936 (unsigned long long) rbd_dev
->mapping
.features
);
1939 static ssize_t
rbd_major_show(struct device
*dev
,
1940 struct device_attribute
*attr
, char *buf
)
1942 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1944 return sprintf(buf
, "%d\n", rbd_dev
->major
);
1947 static ssize_t
rbd_client_id_show(struct device
*dev
,
1948 struct device_attribute
*attr
, char *buf
)
1950 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1952 return sprintf(buf
, "client%lld\n",
1953 ceph_client_id(rbd_dev
->rbd_client
->client
));
1956 static ssize_t
rbd_pool_show(struct device
*dev
,
1957 struct device_attribute
*attr
, char *buf
)
1959 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1961 return sprintf(buf
, "%s\n", rbd_dev
->pool_name
);
1964 static ssize_t
rbd_pool_id_show(struct device
*dev
,
1965 struct device_attribute
*attr
, char *buf
)
1967 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1969 return sprintf(buf
, "%llu\n", (unsigned long long) rbd_dev
->pool_id
);
1972 static ssize_t
rbd_name_show(struct device
*dev
,
1973 struct device_attribute
*attr
, char *buf
)
1975 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1977 return sprintf(buf
, "%s\n", rbd_dev
->image_name
);
1980 static ssize_t
rbd_image_id_show(struct device
*dev
,
1981 struct device_attribute
*attr
, char *buf
)
1983 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1985 return sprintf(buf
, "%s\n", rbd_dev
->image_id
);
1989 * Shows the name of the currently-mapped snapshot (or
1990 * RBD_SNAP_HEAD_NAME for the base image).
1992 static ssize_t
rbd_snap_show(struct device
*dev
,
1993 struct device_attribute
*attr
,
1996 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1998 return sprintf(buf
, "%s\n", rbd_dev
->snap_name
);
2001 static ssize_t
rbd_image_refresh(struct device
*dev
,
2002 struct device_attribute
*attr
,
2006 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2009 ret
= rbd_dev_refresh(rbd_dev
, NULL
);
2011 return ret
< 0 ? ret
: size
;
2014 static DEVICE_ATTR(size
, S_IRUGO
, rbd_size_show
, NULL
);
2015 static DEVICE_ATTR(features
, S_IRUGO
, rbd_features_show
, NULL
);
2016 static DEVICE_ATTR(major
, S_IRUGO
, rbd_major_show
, NULL
);
2017 static DEVICE_ATTR(client_id
, S_IRUGO
, rbd_client_id_show
, NULL
);
2018 static DEVICE_ATTR(pool
, S_IRUGO
, rbd_pool_show
, NULL
);
2019 static DEVICE_ATTR(pool_id
, S_IRUGO
, rbd_pool_id_show
, NULL
);
2020 static DEVICE_ATTR(name
, S_IRUGO
, rbd_name_show
, NULL
);
2021 static DEVICE_ATTR(image_id
, S_IRUGO
, rbd_image_id_show
, NULL
);
2022 static DEVICE_ATTR(refresh
, S_IWUSR
, NULL
, rbd_image_refresh
);
2023 static DEVICE_ATTR(current_snap
, S_IRUGO
, rbd_snap_show
, NULL
);
2025 static struct attribute
*rbd_attrs
[] = {
2026 &dev_attr_size
.attr
,
2027 &dev_attr_features
.attr
,
2028 &dev_attr_major
.attr
,
2029 &dev_attr_client_id
.attr
,
2030 &dev_attr_pool
.attr
,
2031 &dev_attr_pool_id
.attr
,
2032 &dev_attr_name
.attr
,
2033 &dev_attr_image_id
.attr
,
2034 &dev_attr_current_snap
.attr
,
2035 &dev_attr_refresh
.attr
,
2039 static struct attribute_group rbd_attr_group
= {
2043 static const struct attribute_group
*rbd_attr_groups
[] = {
2048 static void rbd_sysfs_dev_release(struct device
*dev
)
2052 static struct device_type rbd_device_type
= {
2054 .groups
= rbd_attr_groups
,
2055 .release
= rbd_sysfs_dev_release
,
2063 static ssize_t
rbd_snap_size_show(struct device
*dev
,
2064 struct device_attribute
*attr
,
2067 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2069 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->size
);
2072 static ssize_t
rbd_snap_id_show(struct device
*dev
,
2073 struct device_attribute
*attr
,
2076 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2078 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->id
);
2081 static ssize_t
rbd_snap_features_show(struct device
*dev
,
2082 struct device_attribute
*attr
,
2085 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2087 return sprintf(buf
, "0x%016llx\n",
2088 (unsigned long long) snap
->features
);
2091 static DEVICE_ATTR(snap_size
, S_IRUGO
, rbd_snap_size_show
, NULL
);
2092 static DEVICE_ATTR(snap_id
, S_IRUGO
, rbd_snap_id_show
, NULL
);
2093 static DEVICE_ATTR(snap_features
, S_IRUGO
, rbd_snap_features_show
, NULL
);
2095 static struct attribute
*rbd_snap_attrs
[] = {
2096 &dev_attr_snap_size
.attr
,
2097 &dev_attr_snap_id
.attr
,
2098 &dev_attr_snap_features
.attr
,
2102 static struct attribute_group rbd_snap_attr_group
= {
2103 .attrs
= rbd_snap_attrs
,
2106 static void rbd_snap_dev_release(struct device
*dev
)
2108 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2113 static const struct attribute_group
*rbd_snap_attr_groups
[] = {
2114 &rbd_snap_attr_group
,
2118 static struct device_type rbd_snap_device_type
= {
2119 .groups
= rbd_snap_attr_groups
,
2120 .release
= rbd_snap_dev_release
,
2123 static bool rbd_snap_registered(struct rbd_snap
*snap
)
2125 bool ret
= snap
->dev
.type
== &rbd_snap_device_type
;
2126 bool reg
= device_is_registered(&snap
->dev
);
2128 rbd_assert(!ret
^ reg
);
2133 static void rbd_remove_snap_dev(struct rbd_snap
*snap
)
2135 list_del(&snap
->node
);
2136 if (device_is_registered(&snap
->dev
))
2137 device_unregister(&snap
->dev
);
2140 static int rbd_register_snap_dev(struct rbd_snap
*snap
,
2141 struct device
*parent
)
2143 struct device
*dev
= &snap
->dev
;
2146 dev
->type
= &rbd_snap_device_type
;
2147 dev
->parent
= parent
;
2148 dev
->release
= rbd_snap_dev_release
;
2149 dev_set_name(dev
, "%s%s", RBD_SNAP_DEV_NAME_PREFIX
, snap
->name
);
2150 dout("%s: registering device for snapshot %s\n", __func__
, snap
->name
);
2152 ret
= device_register(dev
);
2157 static struct rbd_snap
*__rbd_add_snap_dev(struct rbd_device
*rbd_dev
,
2158 const char *snap_name
,
2159 u64 snap_id
, u64 snap_size
,
2162 struct rbd_snap
*snap
;
2165 snap
= kzalloc(sizeof (*snap
), GFP_KERNEL
);
2167 return ERR_PTR(-ENOMEM
);
2170 snap
->name
= kstrdup(snap_name
, GFP_KERNEL
);
2175 snap
->size
= snap_size
;
2176 snap
->features
= snap_features
;
2184 return ERR_PTR(ret
);
2187 static char *rbd_dev_v1_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2188 u64
*snap_size
, u64
*snap_features
)
2192 rbd_assert(which
< rbd_dev
->header
.snapc
->num_snaps
);
2194 *snap_size
= rbd_dev
->header
.snap_sizes
[which
];
2195 *snap_features
= 0; /* No features for v1 */
2197 /* Skip over names until we find the one we are looking for */
2199 snap_name
= rbd_dev
->header
.snap_names
;
2201 snap_name
+= strlen(snap_name
) + 1;
2207 * Get the size and object order for an image snapshot, or if
2208 * snap_id is CEPH_NOSNAP, gets this information for the base
2211 static int _rbd_dev_v2_snap_size(struct rbd_device
*rbd_dev
, u64 snap_id
,
2212 u8
*order
, u64
*snap_size
)
2214 __le64 snapid
= cpu_to_le64(snap_id
);
2219 } __attribute__ ((packed
)) size_buf
= { 0 };
2221 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2223 (char *) &snapid
, sizeof (snapid
),
2224 (char *) &size_buf
, sizeof (size_buf
),
2225 CEPH_OSD_FLAG_READ
, NULL
);
2226 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2230 *order
= size_buf
.order
;
2231 *snap_size
= le64_to_cpu(size_buf
.size
);
2233 dout(" snap_id 0x%016llx order = %u, snap_size = %llu\n",
2234 (unsigned long long) snap_id
, (unsigned int) *order
,
2235 (unsigned long long) *snap_size
);
2240 static int rbd_dev_v2_image_size(struct rbd_device
*rbd_dev
)
2242 return _rbd_dev_v2_snap_size(rbd_dev
, CEPH_NOSNAP
,
2243 &rbd_dev
->header
.obj_order
,
2244 &rbd_dev
->header
.image_size
);
2247 static int rbd_dev_v2_object_prefix(struct rbd_device
*rbd_dev
)
2253 reply_buf
= kzalloc(RBD_OBJ_PREFIX_LEN_MAX
, GFP_KERNEL
);
2257 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2258 "rbd", "get_object_prefix",
2260 reply_buf
, RBD_OBJ_PREFIX_LEN_MAX
,
2261 CEPH_OSD_FLAG_READ
, NULL
);
2262 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2265 ret
= 0; /* rbd_req_sync_exec() can return positive */
2268 rbd_dev
->header
.object_prefix
= ceph_extract_encoded_string(&p
,
2269 p
+ RBD_OBJ_PREFIX_LEN_MAX
,
2272 if (IS_ERR(rbd_dev
->header
.object_prefix
)) {
2273 ret
= PTR_ERR(rbd_dev
->header
.object_prefix
);
2274 rbd_dev
->header
.object_prefix
= NULL
;
2276 dout(" object_prefix = %s\n", rbd_dev
->header
.object_prefix
);
2285 static int _rbd_dev_v2_snap_features(struct rbd_device
*rbd_dev
, u64 snap_id
,
2288 __le64 snapid
= cpu_to_le64(snap_id
);
2292 } features_buf
= { 0 };
2296 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2297 "rbd", "get_features",
2298 (char *) &snapid
, sizeof (snapid
),
2299 (char *) &features_buf
, sizeof (features_buf
),
2300 CEPH_OSD_FLAG_READ
, NULL
);
2301 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2305 incompat
= le64_to_cpu(features_buf
.incompat
);
2306 if (incompat
& ~RBD_FEATURES_ALL
)
2309 *snap_features
= le64_to_cpu(features_buf
.features
);
2311 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
2312 (unsigned long long) snap_id
,
2313 (unsigned long long) *snap_features
,
2314 (unsigned long long) le64_to_cpu(features_buf
.incompat
));
2319 static int rbd_dev_v2_features(struct rbd_device
*rbd_dev
)
2321 return _rbd_dev_v2_snap_features(rbd_dev
, CEPH_NOSNAP
,
2322 &rbd_dev
->header
.features
);
2325 static int rbd_dev_v2_snap_context(struct rbd_device
*rbd_dev
, u64
*ver
)
2334 struct ceph_snap_context
*snapc
;
2338 * We'll need room for the seq value (maximum snapshot id),
2339 * snapshot count, and array of that many snapshot ids.
2340 * For now we have a fixed upper limit on the number we're
2341 * prepared to receive.
2343 size
= sizeof (__le64
) + sizeof (__le32
) +
2344 RBD_MAX_SNAP_COUNT
* sizeof (__le64
);
2345 reply_buf
= kzalloc(size
, GFP_KERNEL
);
2349 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2350 "rbd", "get_snapcontext",
2353 CEPH_OSD_FLAG_READ
, ver
);
2354 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2360 end
= (char *) reply_buf
+ size
;
2361 ceph_decode_64_safe(&p
, end
, seq
, out
);
2362 ceph_decode_32_safe(&p
, end
, snap_count
, out
);
2365 * Make sure the reported number of snapshot ids wouldn't go
2366 * beyond the end of our buffer. But before checking that,
2367 * make sure the computed size of the snapshot context we
2368 * allocate is representable in a size_t.
2370 if (snap_count
> (SIZE_MAX
- sizeof (struct ceph_snap_context
))
2375 if (!ceph_has_room(&p
, end
, snap_count
* sizeof (__le64
)))
2378 size
= sizeof (struct ceph_snap_context
) +
2379 snap_count
* sizeof (snapc
->snaps
[0]);
2380 snapc
= kmalloc(size
, GFP_KERNEL
);
2386 atomic_set(&snapc
->nref
, 1);
2388 snapc
->num_snaps
= snap_count
;
2389 for (i
= 0; i
< snap_count
; i
++)
2390 snapc
->snaps
[i
] = ceph_decode_64(&p
);
2392 rbd_dev
->header
.snapc
= snapc
;
2394 dout(" snap context seq = %llu, snap_count = %u\n",
2395 (unsigned long long) seq
, (unsigned int) snap_count
);
2403 static char *rbd_dev_v2_snap_name(struct rbd_device
*rbd_dev
, u32 which
)
2411 size_t snap_name_len
;
2414 size
= sizeof (__le32
) + RBD_MAX_SNAP_NAME_LEN
;
2415 reply_buf
= kmalloc(size
, GFP_KERNEL
);
2417 return ERR_PTR(-ENOMEM
);
2419 snap_id
= cpu_to_le64(rbd_dev
->header
.snapc
->snaps
[which
]);
2420 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2421 "rbd", "get_snapshot_name",
2422 (char *) &snap_id
, sizeof (snap_id
),
2424 CEPH_OSD_FLAG_READ
, NULL
);
2425 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2430 end
= (char *) reply_buf
+ size
;
2432 snap_name
= ceph_extract_encoded_string(&p
, end
, &snap_name_len
,
2434 if (IS_ERR(snap_name
)) {
2435 ret
= PTR_ERR(snap_name
);
2438 dout(" snap_id 0x%016llx snap_name = %s\n",
2439 (unsigned long long) le64_to_cpu(snap_id
), snap_name
);
2447 return ERR_PTR(ret
);
2450 static char *rbd_dev_v2_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2451 u64
*snap_size
, u64
*snap_features
)
2457 snap_id
= rbd_dev
->header
.snapc
->snaps
[which
];
2458 ret
= _rbd_dev_v2_snap_size(rbd_dev
, snap_id
, &order
, snap_size
);
2460 return ERR_PTR(ret
);
2461 ret
= _rbd_dev_v2_snap_features(rbd_dev
, snap_id
, snap_features
);
2463 return ERR_PTR(ret
);
2465 return rbd_dev_v2_snap_name(rbd_dev
, which
);
2468 static char *rbd_dev_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2469 u64
*snap_size
, u64
*snap_features
)
2471 if (rbd_dev
->image_format
== 1)
2472 return rbd_dev_v1_snap_info(rbd_dev
, which
,
2473 snap_size
, snap_features
);
2474 if (rbd_dev
->image_format
== 2)
2475 return rbd_dev_v2_snap_info(rbd_dev
, which
,
2476 snap_size
, snap_features
);
2477 return ERR_PTR(-EINVAL
);
2480 static int rbd_dev_v2_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
2485 down_write(&rbd_dev
->header_rwsem
);
2487 /* Grab old order first, to see if it changes */
2489 obj_order
= rbd_dev
->header
.obj_order
,
2490 ret
= rbd_dev_v2_image_size(rbd_dev
);
2493 if (rbd_dev
->header
.obj_order
!= obj_order
) {
2497 rbd_update_mapping_size(rbd_dev
);
2499 ret
= rbd_dev_v2_snap_context(rbd_dev
, hver
);
2500 dout("rbd_dev_v2_snap_context returned %d\n", ret
);
2503 ret
= rbd_dev_snaps_update(rbd_dev
);
2504 dout("rbd_dev_snaps_update returned %d\n", ret
);
2507 ret
= rbd_dev_snaps_register(rbd_dev
);
2508 dout("rbd_dev_snaps_register returned %d\n", ret
);
2510 up_write(&rbd_dev
->header_rwsem
);
2516 * Scan the rbd device's current snapshot list and compare it to the
2517 * newly-received snapshot context. Remove any existing snapshots
2518 * not present in the new snapshot context. Add a new snapshot for
2519 * any snaphots in the snapshot context not in the current list.
2520 * And verify there are no changes to snapshots we already know
2523 * Assumes the snapshots in the snapshot context are sorted by
2524 * snapshot id, highest id first. (Snapshots in the rbd_dev's list
2525 * are also maintained in that order.)
2527 static int rbd_dev_snaps_update(struct rbd_device
*rbd_dev
)
2529 struct ceph_snap_context
*snapc
= rbd_dev
->header
.snapc
;
2530 const u32 snap_count
= snapc
->num_snaps
;
2531 struct list_head
*head
= &rbd_dev
->snaps
;
2532 struct list_head
*links
= head
->next
;
2535 dout("%s: snap count is %u\n", __func__
, (unsigned int) snap_count
);
2536 while (index
< snap_count
|| links
!= head
) {
2538 struct rbd_snap
*snap
;
2541 u64 snap_features
= 0;
2543 snap_id
= index
< snap_count
? snapc
->snaps
[index
]
2545 snap
= links
!= head
? list_entry(links
, struct rbd_snap
, node
)
2547 rbd_assert(!snap
|| snap
->id
!= CEPH_NOSNAP
);
2549 if (snap_id
== CEPH_NOSNAP
|| (snap
&& snap
->id
> snap_id
)) {
2550 struct list_head
*next
= links
->next
;
2552 /* Existing snapshot not in the new snap context */
2554 if (rbd_dev
->snap_id
== snap
->id
)
2555 rbd_dev
->exists
= false;
2556 rbd_remove_snap_dev(snap
);
2557 dout("%ssnap id %llu has been removed\n",
2558 rbd_dev
->snap_id
== snap
->id
? "mapped " : "",
2559 (unsigned long long) snap
->id
);
2561 /* Done with this list entry; advance */
2567 snap_name
= rbd_dev_snap_info(rbd_dev
, index
,
2568 &snap_size
, &snap_features
);
2569 if (IS_ERR(snap_name
))
2570 return PTR_ERR(snap_name
);
2572 dout("entry %u: snap_id = %llu\n", (unsigned int) snap_count
,
2573 (unsigned long long) snap_id
);
2574 if (!snap
|| (snap_id
!= CEPH_NOSNAP
&& snap
->id
< snap_id
)) {
2575 struct rbd_snap
*new_snap
;
2577 /* We haven't seen this snapshot before */
2579 new_snap
= __rbd_add_snap_dev(rbd_dev
, snap_name
,
2580 snap_id
, snap_size
, snap_features
);
2581 if (IS_ERR(new_snap
)) {
2582 int err
= PTR_ERR(new_snap
);
2584 dout(" failed to add dev, error %d\n", err
);
2589 /* New goes before existing, or at end of list */
2591 dout(" added dev%s\n", snap
? "" : " at end\n");
2593 list_add_tail(&new_snap
->node
, &snap
->node
);
2595 list_add_tail(&new_snap
->node
, head
);
2597 /* Already have this one */
2599 dout(" already present\n");
2601 rbd_assert(snap
->size
== snap_size
);
2602 rbd_assert(!strcmp(snap
->name
, snap_name
));
2603 rbd_assert(snap
->features
== snap_features
);
2605 /* Done with this list entry; advance */
2607 links
= links
->next
;
2610 /* Advance to the next entry in the snapshot context */
2614 dout("%s: done\n", __func__
);
2620 * Scan the list of snapshots and register the devices for any that
2621 * have not already been registered.
2623 static int rbd_dev_snaps_register(struct rbd_device
*rbd_dev
)
2625 struct rbd_snap
*snap
;
2628 dout("%s called\n", __func__
);
2629 if (WARN_ON(!device_is_registered(&rbd_dev
->dev
)))
2632 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
2633 if (!rbd_snap_registered(snap
)) {
2634 ret
= rbd_register_snap_dev(snap
, &rbd_dev
->dev
);
2639 dout("%s: returning %d\n", __func__
, ret
);
2644 static int rbd_bus_add_dev(struct rbd_device
*rbd_dev
)
2649 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2651 dev
= &rbd_dev
->dev
;
2652 dev
->bus
= &rbd_bus_type
;
2653 dev
->type
= &rbd_device_type
;
2654 dev
->parent
= &rbd_root_dev
;
2655 dev
->release
= rbd_dev_release
;
2656 dev_set_name(dev
, "%d", rbd_dev
->dev_id
);
2657 ret
= device_register(dev
);
2659 mutex_unlock(&ctl_mutex
);
2664 static void rbd_bus_del_dev(struct rbd_device
*rbd_dev
)
2666 device_unregister(&rbd_dev
->dev
);
2669 static int rbd_init_watch_dev(struct rbd_device
*rbd_dev
)
2674 ret
= rbd_req_sync_watch(rbd_dev
);
2675 if (ret
== -ERANGE
) {
2676 rc
= rbd_dev_refresh(rbd_dev
, NULL
);
2680 } while (ret
== -ERANGE
);
2685 static atomic64_t rbd_dev_id_max
= ATOMIC64_INIT(0);
2688 * Get a unique rbd identifier for the given new rbd_dev, and add
2689 * the rbd_dev to the global list. The minimum rbd id is 1.
2691 static void rbd_dev_id_get(struct rbd_device
*rbd_dev
)
2693 rbd_dev
->dev_id
= atomic64_inc_return(&rbd_dev_id_max
);
2695 spin_lock(&rbd_dev_list_lock
);
2696 list_add_tail(&rbd_dev
->node
, &rbd_dev_list
);
2697 spin_unlock(&rbd_dev_list_lock
);
2698 dout("rbd_dev %p given dev id %llu\n", rbd_dev
,
2699 (unsigned long long) rbd_dev
->dev_id
);
2703 * Remove an rbd_dev from the global list, and record that its
2704 * identifier is no longer in use.
2706 static void rbd_dev_id_put(struct rbd_device
*rbd_dev
)
2708 struct list_head
*tmp
;
2709 int rbd_id
= rbd_dev
->dev_id
;
2712 rbd_assert(rbd_id
> 0);
2714 dout("rbd_dev %p released dev id %llu\n", rbd_dev
,
2715 (unsigned long long) rbd_dev
->dev_id
);
2716 spin_lock(&rbd_dev_list_lock
);
2717 list_del_init(&rbd_dev
->node
);
2720 * If the id being "put" is not the current maximum, there
2721 * is nothing special we need to do.
2723 if (rbd_id
!= atomic64_read(&rbd_dev_id_max
)) {
2724 spin_unlock(&rbd_dev_list_lock
);
2729 * We need to update the current maximum id. Search the
2730 * list to find out what it is. We're more likely to find
2731 * the maximum at the end, so search the list backward.
2734 list_for_each_prev(tmp
, &rbd_dev_list
) {
2735 struct rbd_device
*rbd_dev
;
2737 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
2738 if (rbd_dev
->dev_id
> max_id
)
2739 max_id
= rbd_dev
->dev_id
;
2741 spin_unlock(&rbd_dev_list_lock
);
2744 * The max id could have been updated by rbd_dev_id_get(), in
2745 * which case it now accurately reflects the new maximum.
2746 * Be careful not to overwrite the maximum value in that
2749 atomic64_cmpxchg(&rbd_dev_id_max
, rbd_id
, max_id
);
2750 dout(" max dev id has been reset\n");
2754 * Skips over white space at *buf, and updates *buf to point to the
2755 * first found non-space character (if any). Returns the length of
2756 * the token (string of non-white space characters) found. Note
2757 * that *buf must be terminated with '\0'.
2759 static inline size_t next_token(const char **buf
)
2762 * These are the characters that produce nonzero for
2763 * isspace() in the "C" and "POSIX" locales.
2765 const char *spaces
= " \f\n\r\t\v";
2767 *buf
+= strspn(*buf
, spaces
); /* Find start of token */
2769 return strcspn(*buf
, spaces
); /* Return token length */
2773 * Finds the next token in *buf, and if the provided token buffer is
2774 * big enough, copies the found token into it. The result, if
2775 * copied, is guaranteed to be terminated with '\0'. Note that *buf
2776 * must be terminated with '\0' on entry.
2778 * Returns the length of the token found (not including the '\0').
2779 * Return value will be 0 if no token is found, and it will be >=
2780 * token_size if the token would not fit.
2782 * The *buf pointer will be updated to point beyond the end of the
2783 * found token. Note that this occurs even if the token buffer is
2784 * too small to hold it.
2786 static inline size_t copy_token(const char **buf
,
2792 len
= next_token(buf
);
2793 if (len
< token_size
) {
2794 memcpy(token
, *buf
, len
);
2795 *(token
+ len
) = '\0';
2803 * Finds the next token in *buf, dynamically allocates a buffer big
2804 * enough to hold a copy of it, and copies the token into the new
2805 * buffer. The copy is guaranteed to be terminated with '\0'. Note
2806 * that a duplicate buffer is created even for a zero-length token.
2808 * Returns a pointer to the newly-allocated duplicate, or a null
2809 * pointer if memory for the duplicate was not available. If
2810 * the lenp argument is a non-null pointer, the length of the token
2811 * (not including the '\0') is returned in *lenp.
2813 * If successful, the *buf pointer will be updated to point beyond
2814 * the end of the found token.
2816 * Note: uses GFP_KERNEL for allocation.
2818 static inline char *dup_token(const char **buf
, size_t *lenp
)
2823 len
= next_token(buf
);
2824 dup
= kmalloc(len
+ 1, GFP_KERNEL
);
2828 memcpy(dup
, *buf
, len
);
2829 *(dup
+ len
) = '\0';
2839 * This fills in the pool_name, image_name, image_name_len, rbd_dev,
2840 * rbd_md_name, and name fields of the given rbd_dev, based on the
2841 * list of monitor addresses and other options provided via
2842 * /sys/bus/rbd/add. Returns a pointer to a dynamically-allocated
2843 * copy of the snapshot name to map if successful, or a
2844 * pointer-coded error otherwise.
2846 * Note: rbd_dev is assumed to have been initially zero-filled.
2848 static char *rbd_add_parse_args(struct rbd_device
*rbd_dev
,
2850 const char **mon_addrs
,
2851 size_t *mon_addrs_size
,
2853 size_t options_size
)
2856 char *err_ptr
= ERR_PTR(-EINVAL
);
2859 /* The first four tokens are required */
2861 len
= next_token(&buf
);
2864 *mon_addrs_size
= len
+ 1;
2869 len
= copy_token(&buf
, options
, options_size
);
2870 if (!len
|| len
>= options_size
)
2873 err_ptr
= ERR_PTR(-ENOMEM
);
2874 rbd_dev
->pool_name
= dup_token(&buf
, NULL
);
2875 if (!rbd_dev
->pool_name
)
2878 rbd_dev
->image_name
= dup_token(&buf
, &rbd_dev
->image_name_len
);
2879 if (!rbd_dev
->image_name
)
2882 /* Snapshot name is optional; default is to use "head" */
2884 len
= next_token(&buf
);
2885 if (len
> RBD_MAX_SNAP_NAME_LEN
) {
2886 err_ptr
= ERR_PTR(-ENAMETOOLONG
);
2890 buf
= RBD_SNAP_HEAD_NAME
; /* No snapshot supplied */
2891 len
= sizeof (RBD_SNAP_HEAD_NAME
) - 1;
2893 snap_name
= kmalloc(len
+ 1, GFP_KERNEL
);
2896 memcpy(snap_name
, buf
, len
);
2897 *(snap_name
+ len
) = '\0';
2902 kfree(rbd_dev
->image_name
);
2903 rbd_dev
->image_name
= NULL
;
2904 rbd_dev
->image_name_len
= 0;
2905 kfree(rbd_dev
->pool_name
);
2906 rbd_dev
->pool_name
= NULL
;
2912 * An rbd format 2 image has a unique identifier, distinct from the
2913 * name given to it by the user. Internally, that identifier is
2914 * what's used to specify the names of objects related to the image.
2916 * A special "rbd id" object is used to map an rbd image name to its
2917 * id. If that object doesn't exist, then there is no v2 rbd image
2918 * with the supplied name.
2920 * This function will record the given rbd_dev's image_id field if
2921 * it can be determined, and in that case will return 0. If any
2922 * errors occur a negative errno will be returned and the rbd_dev's
2923 * image_id field will be unchanged (and should be NULL).
2925 static int rbd_dev_image_id(struct rbd_device
*rbd_dev
)
2934 * First, see if the format 2 image id file exists, and if
2935 * so, get the image's persistent id from it.
2937 size
= sizeof (RBD_ID_PREFIX
) + rbd_dev
->image_name_len
;
2938 object_name
= kmalloc(size
, GFP_NOIO
);
2941 sprintf(object_name
, "%s%s", RBD_ID_PREFIX
, rbd_dev
->image_name
);
2942 dout("rbd id object name is %s\n", object_name
);
2944 /* Response will be an encoded string, which includes a length */
2946 size
= sizeof (__le32
) + RBD_IMAGE_ID_LEN_MAX
;
2947 response
= kzalloc(size
, GFP_NOIO
);
2953 ret
= rbd_req_sync_exec(rbd_dev
, object_name
,
2956 response
, RBD_IMAGE_ID_LEN_MAX
,
2957 CEPH_OSD_FLAG_READ
, NULL
);
2958 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2961 ret
= 0; /* rbd_req_sync_exec() can return positive */
2964 rbd_dev
->image_id
= ceph_extract_encoded_string(&p
,
2965 p
+ RBD_IMAGE_ID_LEN_MAX
,
2966 &rbd_dev
->image_id_len
,
2968 if (IS_ERR(rbd_dev
->image_id
)) {
2969 ret
= PTR_ERR(rbd_dev
->image_id
);
2970 rbd_dev
->image_id
= NULL
;
2972 dout("image_id is %s\n", rbd_dev
->image_id
);
2981 static int rbd_dev_v1_probe(struct rbd_device
*rbd_dev
)
2986 /* Version 1 images have no id; empty string is used */
2988 rbd_dev
->image_id
= kstrdup("", GFP_KERNEL
);
2989 if (!rbd_dev
->image_id
)
2991 rbd_dev
->image_id_len
= 0;
2993 /* Record the header object name for this rbd image. */
2995 size
= rbd_dev
->image_name_len
+ sizeof (RBD_SUFFIX
);
2996 rbd_dev
->header_name
= kmalloc(size
, GFP_KERNEL
);
2997 if (!rbd_dev
->header_name
) {
3001 sprintf(rbd_dev
->header_name
, "%s%s", rbd_dev
->image_name
, RBD_SUFFIX
);
3003 /* Populate rbd image metadata */
3005 ret
= rbd_read_header(rbd_dev
, &rbd_dev
->header
);
3008 rbd_dev
->image_format
= 1;
3010 dout("discovered version 1 image, header name is %s\n",
3011 rbd_dev
->header_name
);
3016 kfree(rbd_dev
->header_name
);
3017 rbd_dev
->header_name
= NULL
;
3018 kfree(rbd_dev
->image_id
);
3019 rbd_dev
->image_id
= NULL
;
3024 static int rbd_dev_v2_probe(struct rbd_device
*rbd_dev
)
3031 * Image id was filled in by the caller. Record the header
3032 * object name for this rbd image.
3034 size
= sizeof (RBD_HEADER_PREFIX
) + rbd_dev
->image_id_len
;
3035 rbd_dev
->header_name
= kmalloc(size
, GFP_KERNEL
);
3036 if (!rbd_dev
->header_name
)
3038 sprintf(rbd_dev
->header_name
, "%s%s",
3039 RBD_HEADER_PREFIX
, rbd_dev
->image_id
);
3041 /* Get the size and object order for the image */
3043 ret
= rbd_dev_v2_image_size(rbd_dev
);
3047 /* Get the object prefix (a.k.a. block_name) for the image */
3049 ret
= rbd_dev_v2_object_prefix(rbd_dev
);
3053 /* Get the and check features for the image */
3055 ret
= rbd_dev_v2_features(rbd_dev
);
3059 /* crypto and compression type aren't (yet) supported for v2 images */
3061 rbd_dev
->header
.crypt_type
= 0;
3062 rbd_dev
->header
.comp_type
= 0;
3064 /* Get the snapshot context, plus the header version */
3066 ret
= rbd_dev_v2_snap_context(rbd_dev
, &ver
);
3069 rbd_dev
->header
.obj_version
= ver
;
3071 rbd_dev
->image_format
= 2;
3073 dout("discovered version 2 image, header name is %s\n",
3074 rbd_dev
->header_name
);
3078 kfree(rbd_dev
->header_name
);
3079 rbd_dev
->header_name
= NULL
;
3080 kfree(rbd_dev
->header
.object_prefix
);
3081 rbd_dev
->header
.object_prefix
= NULL
;
3087 * Probe for the existence of the header object for the given rbd
3088 * device. For format 2 images this includes determining the image
3091 static int rbd_dev_probe(struct rbd_device
*rbd_dev
)
3096 * Get the id from the image id object. If it's not a
3097 * format 2 image, we'll get ENOENT back, and we'll assume
3098 * it's a format 1 image.
3100 ret
= rbd_dev_image_id(rbd_dev
);
3102 ret
= rbd_dev_v1_probe(rbd_dev
);
3104 ret
= rbd_dev_v2_probe(rbd_dev
);
3106 dout("probe failed, returning %d\n", ret
);
3111 static ssize_t
rbd_add(struct bus_type
*bus
,
3116 struct rbd_device
*rbd_dev
= NULL
;
3117 const char *mon_addrs
= NULL
;
3118 size_t mon_addrs_size
= 0;
3120 struct rbd_options rbd_opts
;
3121 struct ceph_options
*ceph_opts
;
3122 struct ceph_osd_client
*osdc
;
3125 if (!try_module_get(THIS_MODULE
))
3128 options
= kmalloc(count
, GFP_KERNEL
);
3131 rbd_dev
= kzalloc(sizeof(*rbd_dev
), GFP_KERNEL
);
3135 /* static rbd_device initialization */
3136 spin_lock_init(&rbd_dev
->lock
);
3137 INIT_LIST_HEAD(&rbd_dev
->node
);
3138 INIT_LIST_HEAD(&rbd_dev
->snaps
);
3139 init_rwsem(&rbd_dev
->header_rwsem
);
3141 /* parse add command */
3142 snap_name
= rbd_add_parse_args(rbd_dev
, buf
,
3143 &mon_addrs
, &mon_addrs_size
, options
, count
);
3144 if (IS_ERR(snap_name
)) {
3145 rc
= PTR_ERR(snap_name
);
3149 /* Initialize all rbd options to the defaults */
3151 rbd_opts
.read_only
= RBD_READ_ONLY_DEFAULT
;
3153 ceph_opts
= ceph_parse_options(options
, mon_addrs
,
3154 mon_addrs
+ mon_addrs_size
- 1,
3155 parse_rbd_opts_token
, &rbd_opts
);
3156 if (IS_ERR(ceph_opts
)) {
3157 rc
= PTR_ERR(ceph_opts
);
3161 /* Record the parsed rbd options */
3163 rbd_dev
->mapping
.read_only
= rbd_opts
.read_only
;
3165 rc
= rbd_get_client(rbd_dev
, ceph_opts
);
3168 ceph_opts
= NULL
; /* ceph_opts now owned by rbd_dev client */
3171 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
3172 rc
= ceph_pg_poolid_by_name(osdc
->osdmap
, rbd_dev
->pool_name
);
3174 goto err_out_client
;
3175 rbd_dev
->pool_id
= (u64
) rc
;
3177 rc
= rbd_dev_probe(rbd_dev
);
3179 goto err_out_client
;
3181 /* no need to lock here, as rbd_dev is not registered yet */
3182 rc
= rbd_dev_snaps_update(rbd_dev
);
3186 rc
= rbd_dev_set_mapping(rbd_dev
, snap_name
);
3190 /* generate unique id: find highest unique id, add one */
3191 rbd_dev_id_get(rbd_dev
);
3193 /* Fill in the device name, now that we have its id. */
3194 BUILD_BUG_ON(DEV_NAME_LEN
3195 < sizeof (RBD_DRV_NAME
) + MAX_INT_FORMAT_WIDTH
);
3196 sprintf(rbd_dev
->name
, "%s%d", RBD_DRV_NAME
, rbd_dev
->dev_id
);
3198 /* Get our block major device number. */
3200 rc
= register_blkdev(0, rbd_dev
->name
);
3203 rbd_dev
->major
= rc
;
3205 /* Set up the blkdev mapping. */
3207 rc
= rbd_init_disk(rbd_dev
);
3209 goto err_out_blkdev
;
3211 rc
= rbd_bus_add_dev(rbd_dev
);
3216 * At this point cleanup in the event of an error is the job
3217 * of the sysfs code (initiated by rbd_bus_del_dev()).
3220 down_write(&rbd_dev
->header_rwsem
);
3221 rc
= rbd_dev_snaps_register(rbd_dev
);
3222 up_write(&rbd_dev
->header_rwsem
);
3226 rc
= rbd_init_watch_dev(rbd_dev
);
3230 /* Everything's ready. Announce the disk to the world. */
3232 add_disk(rbd_dev
->disk
);
3234 pr_info("%s: added with size 0x%llx\n", rbd_dev
->disk
->disk_name
,
3235 (unsigned long long) rbd_dev
->mapping
.size
);
3240 /* this will also clean up rest of rbd_dev stuff */
3242 rbd_bus_del_dev(rbd_dev
);
3247 rbd_free_disk(rbd_dev
);
3249 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
3251 rbd_dev_id_put(rbd_dev
);
3253 rbd_remove_all_snaps(rbd_dev
);
3255 rbd_header_free(&rbd_dev
->header
);
3257 kfree(rbd_dev
->header_name
);
3258 rbd_put_client(rbd_dev
);
3259 kfree(rbd_dev
->image_id
);
3262 ceph_destroy_options(ceph_opts
);
3264 kfree(rbd_dev
->snap_name
);
3265 kfree(rbd_dev
->image_name
);
3266 kfree(rbd_dev
->pool_name
);
3271 dout("Error adding device %s\n", buf
);
3272 module_put(THIS_MODULE
);
3274 return (ssize_t
) rc
;
3277 static struct rbd_device
*__rbd_get_dev(unsigned long dev_id
)
3279 struct list_head
*tmp
;
3280 struct rbd_device
*rbd_dev
;
3282 spin_lock(&rbd_dev_list_lock
);
3283 list_for_each(tmp
, &rbd_dev_list
) {
3284 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
3285 if (rbd_dev
->dev_id
== dev_id
) {
3286 spin_unlock(&rbd_dev_list_lock
);
3290 spin_unlock(&rbd_dev_list_lock
);
3294 static void rbd_dev_release(struct device
*dev
)
3296 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
3298 if (rbd_dev
->watch_request
) {
3299 struct ceph_client
*client
= rbd_dev
->rbd_client
->client
;
3301 ceph_osdc_unregister_linger_request(&client
->osdc
,
3302 rbd_dev
->watch_request
);
3304 if (rbd_dev
->watch_event
)
3305 rbd_req_sync_unwatch(rbd_dev
);
3307 rbd_put_client(rbd_dev
);
3309 /* clean up and free blkdev */
3310 rbd_free_disk(rbd_dev
);
3311 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
3313 /* release allocated disk header fields */
3314 rbd_header_free(&rbd_dev
->header
);
3316 /* done with the id, and with the rbd_dev */
3317 kfree(rbd_dev
->snap_name
);
3318 kfree(rbd_dev
->image_id
);
3319 kfree(rbd_dev
->header_name
);
3320 kfree(rbd_dev
->pool_name
);
3321 kfree(rbd_dev
->image_name
);
3322 rbd_dev_id_put(rbd_dev
);
3325 /* release module ref */
3326 module_put(THIS_MODULE
);
3329 static ssize_t
rbd_remove(struct bus_type
*bus
,
3333 struct rbd_device
*rbd_dev
= NULL
;
3338 rc
= strict_strtoul(buf
, 10, &ul
);
3342 /* convert to int; abort if we lost anything in the conversion */
3343 target_id
= (int) ul
;
3344 if (target_id
!= ul
)
3347 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
3349 rbd_dev
= __rbd_get_dev(target_id
);
3355 rbd_remove_all_snaps(rbd_dev
);
3356 rbd_bus_del_dev(rbd_dev
);
3359 mutex_unlock(&ctl_mutex
);
3365 * create control files in sysfs
3368 static int rbd_sysfs_init(void)
3372 ret
= device_register(&rbd_root_dev
);
3376 ret
= bus_register(&rbd_bus_type
);
3378 device_unregister(&rbd_root_dev
);
3383 static void rbd_sysfs_cleanup(void)
3385 bus_unregister(&rbd_bus_type
);
3386 device_unregister(&rbd_root_dev
);
3389 int __init
rbd_init(void)
3393 rc
= rbd_sysfs_init();
3396 pr_info("loaded " RBD_DRV_NAME_LONG
"\n");
3400 void __exit
rbd_exit(void)
3402 rbd_sysfs_cleanup();
3405 module_init(rbd_init
);
3406 module_exit(rbd_exit
);
3408 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
3409 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
3410 MODULE_DESCRIPTION("rados block device");
3412 /* following authorship retained from original osdblk.c */
3413 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
3415 MODULE_LICENSE("GPL");