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 /* This allows a single page to hold an image name sent by OSD */
74 #define RBD_IMAGE_NAME_LEN_MAX (PAGE_SIZE - sizeof (__le32) - 1)
75 #define RBD_IMAGE_ID_LEN_MAX 64
77 #define RBD_OBJ_PREFIX_LEN_MAX 64
81 #define RBD_FEATURE_LAYERING 1
83 /* Features supported by this (client software) implementation. */
85 #define RBD_FEATURES_ALL (0)
88 * An RBD device name will be "rbd#", where the "rbd" comes from
89 * RBD_DRV_NAME above, and # is a unique integer identifier.
90 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
91 * enough to hold all possible device names.
93 #define DEV_NAME_LEN 32
94 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1)
96 #define RBD_READ_ONLY_DEFAULT false
99 * block device image metadata (in-memory version)
101 struct rbd_image_header
{
102 /* These four fields never change for a given rbd image */
109 /* The remaining fields need to be updated occasionally */
111 struct ceph_snap_context
*snapc
;
119 * An rbd image specification.
121 * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
122 * identify an image. Each rbd_dev structure includes a pointer to
123 * an rbd_spec structure that encapsulates this identity.
125 * Each of the id's in an rbd_spec has an associated name. For a
126 * user-mapped image, the names are supplied and the id's associated
127 * with them are looked up. For a layered image, a parent image is
128 * defined by the tuple, and the names are looked up.
130 * An rbd_dev structure contains a parent_spec pointer which is
131 * non-null if the image it represents is a child in a layered
132 * image. This pointer will refer to the rbd_spec structure used
133 * by the parent rbd_dev for its own identity (i.e., the structure
134 * is shared between the parent and child).
136 * Since these structures are populated once, during the discovery
137 * phase of image construction, they are effectively immutable so
138 * we make no effort to synchronize access to them.
140 * Note that code herein does not assume the image name is known (it
141 * could be a null pointer).
161 * an instance of the client. multiple devices may share an rbd client.
164 struct ceph_client
*client
;
166 struct list_head node
;
170 * a request completion status
172 struct rbd_req_status
{
179 * a collection of requests
181 struct rbd_req_coll
{
185 struct rbd_req_status status
[0];
189 * a single io request
192 struct request
*rq
; /* blk layer request */
193 struct bio
*bio
; /* cloned bio */
194 struct page
**pages
; /* list of used pages */
197 struct rbd_req_coll
*coll
;
204 struct list_head node
;
219 int dev_id
; /* blkdev unique id */
221 int major
; /* blkdev assigned major */
222 struct gendisk
*disk
; /* blkdev's gendisk and rq */
224 u32 image_format
; /* Either 1 or 2 */
225 struct rbd_client
*rbd_client
;
227 char name
[DEV_NAME_LEN
]; /* blkdev name, e.g. rbd3 */
229 spinlock_t lock
; /* queue lock */
231 struct rbd_image_header header
;
233 struct rbd_spec
*spec
;
237 struct ceph_osd_event
*watch_event
;
238 struct ceph_osd_request
*watch_request
;
240 struct rbd_spec
*parent_spec
;
243 /* protects updating the header */
244 struct rw_semaphore header_rwsem
;
246 struct rbd_mapping mapping
;
248 struct list_head node
;
250 /* list of snapshots */
251 struct list_head snaps
;
255 unsigned long open_count
;
258 static DEFINE_MUTEX(ctl_mutex
); /* Serialize open/close/setup/teardown */
260 static LIST_HEAD(rbd_dev_list
); /* devices */
261 static DEFINE_SPINLOCK(rbd_dev_list_lock
);
263 static LIST_HEAD(rbd_client_list
); /* clients */
264 static DEFINE_SPINLOCK(rbd_client_list_lock
);
266 static int rbd_dev_snaps_update(struct rbd_device
*rbd_dev
);
267 static int rbd_dev_snaps_register(struct rbd_device
*rbd_dev
);
269 static void rbd_dev_release(struct device
*dev
);
270 static void rbd_remove_snap_dev(struct rbd_snap
*snap
);
272 static ssize_t
rbd_add(struct bus_type
*bus
, const char *buf
,
274 static ssize_t
rbd_remove(struct bus_type
*bus
, const char *buf
,
277 static struct bus_attribute rbd_bus_attrs
[] = {
278 __ATTR(add
, S_IWUSR
, NULL
, rbd_add
),
279 __ATTR(remove
, S_IWUSR
, NULL
, rbd_remove
),
283 static struct bus_type rbd_bus_type
= {
285 .bus_attrs
= rbd_bus_attrs
,
288 static void rbd_root_dev_release(struct device
*dev
)
292 static struct device rbd_root_dev
= {
294 .release
= rbd_root_dev_release
,
297 static __printf(2, 3)
298 void rbd_warn(struct rbd_device
*rbd_dev
, const char *fmt
, ...)
300 struct va_format vaf
;
308 printk(KERN_WARNING
"%s: %pV\n", RBD_DRV_NAME
, &vaf
);
309 else if (rbd_dev
->disk
)
310 printk(KERN_WARNING
"%s: %s: %pV\n",
311 RBD_DRV_NAME
, rbd_dev
->disk
->disk_name
, &vaf
);
312 else if (rbd_dev
->spec
&& rbd_dev
->spec
->image_name
)
313 printk(KERN_WARNING
"%s: image %s: %pV\n",
314 RBD_DRV_NAME
, rbd_dev
->spec
->image_name
, &vaf
);
315 else if (rbd_dev
->spec
&& rbd_dev
->spec
->image_id
)
316 printk(KERN_WARNING
"%s: id %s: %pV\n",
317 RBD_DRV_NAME
, rbd_dev
->spec
->image_id
, &vaf
);
319 printk(KERN_WARNING
"%s: rbd_dev %p: %pV\n",
320 RBD_DRV_NAME
, rbd_dev
, &vaf
);
325 #define rbd_assert(expr) \
326 if (unlikely(!(expr))) { \
327 printk(KERN_ERR "\nAssertion failure in %s() " \
329 "\trbd_assert(%s);\n\n", \
330 __func__, __LINE__, #expr); \
333 #else /* !RBD_DEBUG */
334 # define rbd_assert(expr) ((void) 0)
335 #endif /* !RBD_DEBUG */
337 static int rbd_dev_refresh(struct rbd_device
*rbd_dev
, u64
*hver
);
338 static int rbd_dev_v2_refresh(struct rbd_device
*rbd_dev
, u64
*hver
);
340 static int rbd_open(struct block_device
*bdev
, fmode_t mode
)
342 struct rbd_device
*rbd_dev
= bdev
->bd_disk
->private_data
;
344 if ((mode
& FMODE_WRITE
) && rbd_dev
->mapping
.read_only
)
347 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
348 (void) get_device(&rbd_dev
->dev
);
349 set_device_ro(bdev
, rbd_dev
->mapping
.read_only
);
350 rbd_dev
->open_count
++;
351 mutex_unlock(&ctl_mutex
);
356 static int rbd_release(struct gendisk
*disk
, fmode_t mode
)
358 struct rbd_device
*rbd_dev
= disk
->private_data
;
360 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
361 rbd_assert(rbd_dev
->open_count
> 0);
362 rbd_dev
->open_count
--;
363 put_device(&rbd_dev
->dev
);
364 mutex_unlock(&ctl_mutex
);
369 static const struct block_device_operations rbd_bd_ops
= {
370 .owner
= THIS_MODULE
,
372 .release
= rbd_release
,
376 * Initialize an rbd client instance.
379 static struct rbd_client
*rbd_client_create(struct ceph_options
*ceph_opts
)
381 struct rbd_client
*rbdc
;
384 dout("rbd_client_create\n");
385 rbdc
= kmalloc(sizeof(struct rbd_client
), GFP_KERNEL
);
389 kref_init(&rbdc
->kref
);
390 INIT_LIST_HEAD(&rbdc
->node
);
392 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
394 rbdc
->client
= ceph_create_client(ceph_opts
, rbdc
, 0, 0);
395 if (IS_ERR(rbdc
->client
))
397 ceph_opts
= NULL
; /* Now rbdc->client is responsible for ceph_opts */
399 ret
= ceph_open_session(rbdc
->client
);
403 spin_lock(&rbd_client_list_lock
);
404 list_add_tail(&rbdc
->node
, &rbd_client_list
);
405 spin_unlock(&rbd_client_list_lock
);
407 mutex_unlock(&ctl_mutex
);
409 dout("rbd_client_create created %p\n", rbdc
);
413 ceph_destroy_client(rbdc
->client
);
415 mutex_unlock(&ctl_mutex
);
419 ceph_destroy_options(ceph_opts
);
424 * Find a ceph client with specific addr and configuration. If
425 * found, bump its reference count.
427 static struct rbd_client
*rbd_client_find(struct ceph_options
*ceph_opts
)
429 struct rbd_client
*client_node
;
432 if (ceph_opts
->flags
& CEPH_OPT_NOSHARE
)
435 spin_lock(&rbd_client_list_lock
);
436 list_for_each_entry(client_node
, &rbd_client_list
, node
) {
437 if (!ceph_compare_options(ceph_opts
, client_node
->client
)) {
438 kref_get(&client_node
->kref
);
443 spin_unlock(&rbd_client_list_lock
);
445 return found
? client_node
: NULL
;
455 /* string args above */
458 /* Boolean args above */
462 static match_table_t rbd_opts_tokens
= {
464 /* string args above */
465 {Opt_read_only
, "read_only"},
466 {Opt_read_only
, "ro"}, /* Alternate spelling */
467 {Opt_read_write
, "read_write"},
468 {Opt_read_write
, "rw"}, /* Alternate spelling */
469 /* Boolean args above */
473 static int parse_rbd_opts_token(char *c
, void *private)
475 struct rbd_options
*rbd_opts
= private;
476 substring_t argstr
[MAX_OPT_ARGS
];
477 int token
, intval
, ret
;
479 token
= match_token(c
, rbd_opts_tokens
, argstr
);
483 if (token
< Opt_last_int
) {
484 ret
= match_int(&argstr
[0], &intval
);
486 pr_err("bad mount option arg (not int) "
490 dout("got int token %d val %d\n", token
, intval
);
491 } else if (token
> Opt_last_int
&& token
< Opt_last_string
) {
492 dout("got string token %d val %s\n", token
,
494 } else if (token
> Opt_last_string
&& token
< Opt_last_bool
) {
495 dout("got Boolean token %d\n", token
);
497 dout("got token %d\n", token
);
502 rbd_opts
->read_only
= true;
505 rbd_opts
->read_only
= false;
515 * Get a ceph client with specific addr and configuration, if one does
516 * not exist create it.
518 static struct rbd_client
*rbd_get_client(struct ceph_options
*ceph_opts
)
520 struct rbd_client
*rbdc
;
522 rbdc
= rbd_client_find(ceph_opts
);
523 if (rbdc
) /* using an existing client */
524 ceph_destroy_options(ceph_opts
);
526 rbdc
= rbd_client_create(ceph_opts
);
532 * Destroy ceph client
534 * Caller must hold rbd_client_list_lock.
536 static void rbd_client_release(struct kref
*kref
)
538 struct rbd_client
*rbdc
= container_of(kref
, struct rbd_client
, kref
);
540 dout("rbd_release_client %p\n", rbdc
);
541 spin_lock(&rbd_client_list_lock
);
542 list_del(&rbdc
->node
);
543 spin_unlock(&rbd_client_list_lock
);
545 ceph_destroy_client(rbdc
->client
);
550 * Drop reference to ceph client node. If it's not referenced anymore, release
553 static void rbd_put_client(struct rbd_client
*rbdc
)
556 kref_put(&rbdc
->kref
, rbd_client_release
);
560 * Destroy requests collection
562 static void rbd_coll_release(struct kref
*kref
)
564 struct rbd_req_coll
*coll
=
565 container_of(kref
, struct rbd_req_coll
, kref
);
567 dout("rbd_coll_release %p\n", coll
);
571 static bool rbd_image_format_valid(u32 image_format
)
573 return image_format
== 1 || image_format
== 2;
576 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk
*ondisk
)
581 /* The header has to start with the magic rbd header text */
582 if (memcmp(&ondisk
->text
, RBD_HEADER_TEXT
, sizeof (RBD_HEADER_TEXT
)))
585 /* The bio layer requires at least sector-sized I/O */
587 if (ondisk
->options
.order
< SECTOR_SHIFT
)
590 /* If we use u64 in a few spots we may be able to loosen this */
592 if (ondisk
->options
.order
> 8 * sizeof (int) - 1)
596 * The size of a snapshot header has to fit in a size_t, and
597 * that limits the number of snapshots.
599 snap_count
= le32_to_cpu(ondisk
->snap_count
);
600 size
= SIZE_MAX
- sizeof (struct ceph_snap_context
);
601 if (snap_count
> size
/ sizeof (__le64
))
605 * Not only that, but the size of the entire the snapshot
606 * header must also be representable in a size_t.
608 size
-= snap_count
* sizeof (__le64
);
609 if ((u64
) size
< le64_to_cpu(ondisk
->snap_names_len
))
616 * Create a new header structure, translate header format from the on-disk
619 static int rbd_header_from_disk(struct rbd_image_header
*header
,
620 struct rbd_image_header_ondisk
*ondisk
)
627 memset(header
, 0, sizeof (*header
));
629 snap_count
= le32_to_cpu(ondisk
->snap_count
);
631 len
= strnlen(ondisk
->object_prefix
, sizeof (ondisk
->object_prefix
));
632 header
->object_prefix
= kmalloc(len
+ 1, GFP_KERNEL
);
633 if (!header
->object_prefix
)
635 memcpy(header
->object_prefix
, ondisk
->object_prefix
, len
);
636 header
->object_prefix
[len
] = '\0';
639 u64 snap_names_len
= le64_to_cpu(ondisk
->snap_names_len
);
641 /* Save a copy of the snapshot names */
643 if (snap_names_len
> (u64
) SIZE_MAX
)
645 header
->snap_names
= kmalloc(snap_names_len
, GFP_KERNEL
);
646 if (!header
->snap_names
)
649 * Note that rbd_dev_v1_header_read() guarantees
650 * the ondisk buffer we're working with has
651 * snap_names_len bytes beyond the end of the
652 * snapshot id array, this memcpy() is safe.
654 memcpy(header
->snap_names
, &ondisk
->snaps
[snap_count
],
657 /* Record each snapshot's size */
659 size
= snap_count
* sizeof (*header
->snap_sizes
);
660 header
->snap_sizes
= kmalloc(size
, GFP_KERNEL
);
661 if (!header
->snap_sizes
)
663 for (i
= 0; i
< snap_count
; i
++)
664 header
->snap_sizes
[i
] =
665 le64_to_cpu(ondisk
->snaps
[i
].image_size
);
667 WARN_ON(ondisk
->snap_names_len
);
668 header
->snap_names
= NULL
;
669 header
->snap_sizes
= NULL
;
672 header
->features
= 0; /* No features support in v1 images */
673 header
->obj_order
= ondisk
->options
.order
;
674 header
->crypt_type
= ondisk
->options
.crypt_type
;
675 header
->comp_type
= ondisk
->options
.comp_type
;
677 /* Allocate and fill in the snapshot context */
679 header
->image_size
= le64_to_cpu(ondisk
->image_size
);
680 size
= sizeof (struct ceph_snap_context
);
681 size
+= snap_count
* sizeof (header
->snapc
->snaps
[0]);
682 header
->snapc
= kzalloc(size
, GFP_KERNEL
);
686 atomic_set(&header
->snapc
->nref
, 1);
687 header
->snapc
->seq
= le64_to_cpu(ondisk
->snap_seq
);
688 header
->snapc
->num_snaps
= snap_count
;
689 for (i
= 0; i
< snap_count
; i
++)
690 header
->snapc
->snaps
[i
] =
691 le64_to_cpu(ondisk
->snaps
[i
].id
);
696 kfree(header
->snap_sizes
);
697 header
->snap_sizes
= NULL
;
698 kfree(header
->snap_names
);
699 header
->snap_names
= NULL
;
700 kfree(header
->object_prefix
);
701 header
->object_prefix
= NULL
;
706 static const char *rbd_snap_name(struct rbd_device
*rbd_dev
, u64 snap_id
)
708 struct rbd_snap
*snap
;
710 if (snap_id
== CEPH_NOSNAP
)
711 return RBD_SNAP_HEAD_NAME
;
713 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
)
714 if (snap_id
== snap
->id
)
720 static int snap_by_name(struct rbd_device
*rbd_dev
, const char *snap_name
)
723 struct rbd_snap
*snap
;
725 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
726 if (!strcmp(snap_name
, snap
->name
)) {
727 rbd_dev
->spec
->snap_id
= snap
->id
;
728 rbd_dev
->mapping
.size
= snap
->size
;
729 rbd_dev
->mapping
.features
= snap
->features
;
738 static int rbd_dev_set_mapping(struct rbd_device
*rbd_dev
)
742 if (!memcmp(rbd_dev
->spec
->snap_name
, RBD_SNAP_HEAD_NAME
,
743 sizeof (RBD_SNAP_HEAD_NAME
))) {
744 rbd_dev
->spec
->snap_id
= CEPH_NOSNAP
;
745 rbd_dev
->mapping
.size
= rbd_dev
->header
.image_size
;
746 rbd_dev
->mapping
.features
= rbd_dev
->header
.features
;
749 ret
= snap_by_name(rbd_dev
, rbd_dev
->spec
->snap_name
);
752 rbd_dev
->mapping
.read_only
= true;
754 atomic_set(&rbd_dev
->exists
, 1);
759 static void rbd_header_free(struct rbd_image_header
*header
)
761 kfree(header
->object_prefix
);
762 header
->object_prefix
= NULL
;
763 kfree(header
->snap_sizes
);
764 header
->snap_sizes
= NULL
;
765 kfree(header
->snap_names
);
766 header
->snap_names
= NULL
;
767 ceph_put_snap_context(header
->snapc
);
768 header
->snapc
= NULL
;
771 static char *rbd_segment_name(struct rbd_device
*rbd_dev
, u64 offset
)
777 name
= kmalloc(MAX_OBJ_NAME_SIZE
+ 1, GFP_NOIO
);
780 segment
= offset
>> rbd_dev
->header
.obj_order
;
781 ret
= snprintf(name
, MAX_OBJ_NAME_SIZE
+ 1, "%s.%012llx",
782 rbd_dev
->header
.object_prefix
, segment
);
783 if (ret
< 0 || ret
> MAX_OBJ_NAME_SIZE
) {
784 pr_err("error formatting segment name for #%llu (%d)\n",
793 static u64
rbd_segment_offset(struct rbd_device
*rbd_dev
, u64 offset
)
795 u64 segment_size
= (u64
) 1 << rbd_dev
->header
.obj_order
;
797 return offset
& (segment_size
- 1);
800 static u64
rbd_segment_length(struct rbd_device
*rbd_dev
,
801 u64 offset
, u64 length
)
803 u64 segment_size
= (u64
) 1 << rbd_dev
->header
.obj_order
;
805 offset
&= segment_size
- 1;
807 rbd_assert(length
<= U64_MAX
- offset
);
808 if (offset
+ length
> segment_size
)
809 length
= segment_size
- offset
;
814 static int rbd_get_num_segments(struct rbd_image_header
*header
,
822 if (len
- 1 > U64_MAX
- ofs
)
825 start_seg
= ofs
>> header
->obj_order
;
826 end_seg
= (ofs
+ len
- 1) >> header
->obj_order
;
828 return end_seg
- start_seg
+ 1;
832 * returns the size of an object in the image
834 static u64
rbd_obj_bytes(struct rbd_image_header
*header
)
836 return 1 << header
->obj_order
;
843 static void bio_chain_put(struct bio
*chain
)
849 chain
= chain
->bi_next
;
855 * zeros a bio chain, starting at specific offset
857 static void zero_bio_chain(struct bio
*chain
, int start_ofs
)
866 bio_for_each_segment(bv
, chain
, i
) {
867 if (pos
+ bv
->bv_len
> start_ofs
) {
868 int remainder
= max(start_ofs
- pos
, 0);
869 buf
= bvec_kmap_irq(bv
, &flags
);
870 memset(buf
+ remainder
, 0,
871 bv
->bv_len
- remainder
);
872 bvec_kunmap_irq(buf
, &flags
);
877 chain
= chain
->bi_next
;
882 * Clone a portion of a bio, starting at the given byte offset
883 * and continuing for the number of bytes indicated.
885 static struct bio
*bio_clone_range(struct bio
*bio_src
,
894 unsigned short end_idx
;
898 /* Handle the easy case for the caller */
900 if (!offset
&& len
== bio_src
->bi_size
)
901 return bio_clone(bio_src
, gfpmask
);
903 if (WARN_ON_ONCE(!len
))
905 if (WARN_ON_ONCE(len
> bio_src
->bi_size
))
907 if (WARN_ON_ONCE(offset
> bio_src
->bi_size
- len
))
910 /* Find first affected segment... */
913 __bio_for_each_segment(bv
, bio_src
, idx
, 0) {
914 if (resid
< bv
->bv_len
)
920 /* ...and the last affected segment */
923 __bio_for_each_segment(bv
, bio_src
, end_idx
, idx
) {
924 if (resid
<= bv
->bv_len
)
928 vcnt
= end_idx
- idx
+ 1;
930 /* Build the clone */
932 bio
= bio_alloc(gfpmask
, (unsigned int) vcnt
);
934 return NULL
; /* ENOMEM */
936 bio
->bi_bdev
= bio_src
->bi_bdev
;
937 bio
->bi_sector
= bio_src
->bi_sector
+ (offset
>> SECTOR_SHIFT
);
938 bio
->bi_rw
= bio_src
->bi_rw
;
939 bio
->bi_flags
|= 1 << BIO_CLONED
;
942 * Copy over our part of the bio_vec, then update the first
943 * and last (or only) entries.
945 memcpy(&bio
->bi_io_vec
[0], &bio_src
->bi_io_vec
[idx
],
946 vcnt
* sizeof (struct bio_vec
));
947 bio
->bi_io_vec
[0].bv_offset
+= voff
;
949 bio
->bi_io_vec
[0].bv_len
-= voff
;
950 bio
->bi_io_vec
[vcnt
- 1].bv_len
= resid
;
952 bio
->bi_io_vec
[0].bv_len
= len
;
963 * Clone a portion of a bio chain, starting at the given byte offset
964 * into the first bio in the source chain and continuing for the
965 * number of bytes indicated. The result is another bio chain of
966 * exactly the given length, or a null pointer on error.
968 * The bio_src and offset parameters are both in-out. On entry they
969 * refer to the first source bio and the offset into that bio where
970 * the start of data to be cloned is located.
972 * On return, bio_src is updated to refer to the bio in the source
973 * chain that contains first un-cloned byte, and *offset will
974 * contain the offset of that byte within that bio.
976 static struct bio
*bio_chain_clone_range(struct bio
**bio_src
,
977 unsigned int *offset
,
981 struct bio
*bi
= *bio_src
;
982 unsigned int off
= *offset
;
983 struct bio
*chain
= NULL
;
986 /* Build up a chain of clone bios up to the limit */
988 if (!bi
|| off
>= bi
->bi_size
|| !len
)
989 return NULL
; /* Nothing to clone */
993 unsigned int bi_size
;
997 rbd_warn(NULL
, "bio_chain exhausted with %u left", len
);
998 goto out_err
; /* EINVAL; ran out of bio's */
1000 bi_size
= min_t(unsigned int, bi
->bi_size
- off
, len
);
1001 bio
= bio_clone_range(bi
, off
, bi_size
, gfpmask
);
1003 goto out_err
; /* ENOMEM */
1006 end
= &bio
->bi_next
;
1009 if (off
== bi
->bi_size
) {
1020 bio_chain_put(chain
);
1026 * helpers for osd request op vectors.
1028 static struct ceph_osd_req_op
*rbd_create_rw_ops(int num_ops
,
1029 int opcode
, u32 payload_len
)
1031 struct ceph_osd_req_op
*ops
;
1033 ops
= kzalloc(sizeof (*ops
) * (num_ops
+ 1), GFP_NOIO
);
1040 * op extent offset and length will be set later on
1041 * in calc_raw_layout()
1043 ops
[0].payload_len
= payload_len
;
1048 static void rbd_destroy_ops(struct ceph_osd_req_op
*ops
)
1053 static void rbd_coll_end_req_index(struct request
*rq
,
1054 struct rbd_req_coll
*coll
,
1058 struct request_queue
*q
;
1061 dout("rbd_coll_end_req_index %p index %d ret %d len %llu\n",
1062 coll
, index
, (int)ret
, (unsigned long long)len
);
1068 blk_end_request(rq
, ret
, len
);
1074 spin_lock_irq(q
->queue_lock
);
1075 coll
->status
[index
].done
= 1;
1076 coll
->status
[index
].rc
= ret
;
1077 coll
->status
[index
].bytes
= len
;
1078 max
= min
= coll
->num_done
;
1079 while (max
< coll
->total
&& coll
->status
[max
].done
)
1082 for (i
= min
; i
<max
; i
++) {
1083 __blk_end_request(rq
, (int)coll
->status
[i
].rc
,
1084 coll
->status
[i
].bytes
);
1086 kref_put(&coll
->kref
, rbd_coll_release
);
1088 spin_unlock_irq(q
->queue_lock
);
1091 static void rbd_coll_end_req(struct rbd_request
*rbd_req
,
1094 rbd_coll_end_req_index(rbd_req
->rq
,
1095 rbd_req
->coll
, rbd_req
->coll_index
,
1100 * Send ceph osd request
1102 static int rbd_do_request(struct request
*rq
,
1103 struct rbd_device
*rbd_dev
,
1104 struct ceph_snap_context
*snapc
,
1106 const char *object_name
, u64 ofs
, u64 len
,
1108 struct page
**pages
,
1111 struct ceph_osd_req_op
*ops
,
1112 struct rbd_req_coll
*coll
,
1114 void (*rbd_cb
)(struct ceph_osd_request
*,
1116 struct ceph_osd_request
**linger_req
,
1119 struct ceph_osd_request
*osd_req
;
1120 struct ceph_file_layout
*layout
;
1123 struct timespec mtime
= CURRENT_TIME
;
1124 struct rbd_request
*rbd_req
;
1125 struct ceph_osd_request_head
*reqhead
;
1126 struct ceph_osd_client
*osdc
;
1128 rbd_req
= kzalloc(sizeof(*rbd_req
), GFP_NOIO
);
1133 rbd_req
->coll
= coll
;
1134 rbd_req
->coll_index
= coll_index
;
1137 dout("rbd_do_request object_name=%s ofs=%llu len=%llu coll=%p[%d]\n",
1138 object_name
, (unsigned long long) ofs
,
1139 (unsigned long long) len
, coll
, coll_index
);
1141 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1142 osd_req
= ceph_osdc_alloc_request(osdc
, flags
, snapc
, ops
,
1143 false, GFP_NOIO
, pages
, bio
);
1149 osd_req
->r_callback
= rbd_cb
;
1153 rbd_req
->pages
= pages
;
1156 osd_req
->r_priv
= rbd_req
;
1158 reqhead
= osd_req
->r_request
->front
.iov_base
;
1159 reqhead
->snapid
= cpu_to_le64(CEPH_NOSNAP
);
1161 strncpy(osd_req
->r_oid
, object_name
, sizeof(osd_req
->r_oid
));
1162 osd_req
->r_oid_len
= strlen(osd_req
->r_oid
);
1164 layout
= &osd_req
->r_file_layout
;
1165 memset(layout
, 0, sizeof(*layout
));
1166 layout
->fl_stripe_unit
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
1167 layout
->fl_stripe_count
= cpu_to_le32(1);
1168 layout
->fl_object_size
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
1169 layout
->fl_pg_pool
= cpu_to_le32((int) rbd_dev
->spec
->pool_id
);
1170 ret
= ceph_calc_raw_layout(osdc
, layout
, snapid
, ofs
, &len
, &bno
,
1172 rbd_assert(ret
== 0);
1174 ceph_osdc_build_request(osd_req
, ofs
, &len
,
1178 osd_req
->r_oid
, osd_req
->r_oid_len
);
1181 ceph_osdc_set_request_linger(osdc
, osd_req
);
1182 *linger_req
= osd_req
;
1185 ret
= ceph_osdc_start_request(osdc
, osd_req
, false);
1192 ret
= ceph_osdc_wait_request(osdc
, osd_req
);
1193 version
= le64_to_cpu(osd_req
->r_reassert_version
.version
);
1196 dout("reassert_ver=%llu\n", (unsigned long long) version
);
1197 ceph_osdc_put_request(osd_req
);
1202 bio_chain_put(rbd_req
->bio
);
1203 ceph_osdc_put_request(osd_req
);
1210 * Ceph osd op callback
1212 static void rbd_req_cb(struct ceph_osd_request
*osd_req
, struct ceph_msg
*msg
)
1214 struct rbd_request
*rbd_req
= osd_req
->r_priv
;
1215 struct ceph_osd_reply_head
*replyhead
;
1216 struct ceph_osd_op
*op
;
1222 replyhead
= msg
->front
.iov_base
;
1223 WARN_ON(le32_to_cpu(replyhead
->num_ops
) == 0);
1224 op
= (void *)(replyhead
+ 1);
1225 rc
= (s32
)le32_to_cpu(replyhead
->result
);
1226 bytes
= le64_to_cpu(op
->extent
.length
);
1227 read_op
= (le16_to_cpu(op
->op
) == CEPH_OSD_OP_READ
);
1229 dout("rbd_req_cb bytes=%llu readop=%d rc=%d\n",
1230 (unsigned long long) bytes
, read_op
, (int) rc
);
1232 if (rc
== (s32
)-ENOENT
&& read_op
) {
1233 zero_bio_chain(rbd_req
->bio
, 0);
1235 } else if (rc
== 0 && read_op
&& bytes
< rbd_req
->len
) {
1236 zero_bio_chain(rbd_req
->bio
, bytes
);
1237 bytes
= rbd_req
->len
;
1240 rbd_coll_end_req(rbd_req
, rc
, bytes
);
1243 bio_chain_put(rbd_req
->bio
);
1245 ceph_osdc_put_request(osd_req
);
1249 static void rbd_simple_req_cb(struct ceph_osd_request
*osd_req
,
1250 struct ceph_msg
*msg
)
1252 ceph_osdc_put_request(osd_req
);
1256 * Do a synchronous ceph osd operation
1258 static int rbd_req_sync_op(struct rbd_device
*rbd_dev
,
1259 struct ceph_snap_context
*snapc
,
1262 struct ceph_osd_req_op
*ops
,
1263 const char *object_name
,
1264 u64 ofs
, u64 inbound_size
,
1266 struct ceph_osd_request
**linger_req
,
1270 struct page
**pages
;
1273 rbd_assert(ops
!= NULL
);
1275 num_pages
= calc_pages_for(ofs
, inbound_size
);
1276 pages
= ceph_alloc_page_vector(num_pages
, GFP_KERNEL
);
1278 return PTR_ERR(pages
);
1280 ret
= rbd_do_request(NULL
, rbd_dev
, snapc
, snapid
,
1281 object_name
, ofs
, inbound_size
, NULL
,
1291 if ((flags
& CEPH_OSD_FLAG_READ
) && inbound
)
1292 ret
= ceph_copy_from_page_vector(pages
, inbound
, ofs
, ret
);
1295 ceph_release_page_vector(pages
, num_pages
);
1300 * Do an asynchronous ceph osd operation
1302 static int rbd_do_op(struct request
*rq
,
1303 struct rbd_device
*rbd_dev
,
1304 struct ceph_snap_context
*snapc
,
1307 struct rbd_req_coll
*coll
,
1314 struct ceph_osd_req_op
*ops
;
1320 seg_name
= rbd_segment_name(rbd_dev
, ofs
);
1323 seg_len
= rbd_segment_length(rbd_dev
, ofs
, len
);
1324 seg_ofs
= rbd_segment_offset(rbd_dev
, ofs
);
1326 if (rq_data_dir(rq
) == WRITE
) {
1327 opcode
= CEPH_OSD_OP_WRITE
;
1328 flags
= CEPH_OSD_FLAG_WRITE
|CEPH_OSD_FLAG_ONDISK
;
1329 snapid
= CEPH_NOSNAP
;
1330 payload_len
= seg_len
;
1332 opcode
= CEPH_OSD_OP_READ
;
1333 flags
= CEPH_OSD_FLAG_READ
;
1335 snapid
= rbd_dev
->spec
->snap_id
;
1340 ops
= rbd_create_rw_ops(1, opcode
, payload_len
);
1344 /* we've taken care of segment sizes earlier when we
1345 cloned the bios. We should never have a segment
1346 truncated at this point */
1347 rbd_assert(seg_len
== len
);
1349 ret
= rbd_do_request(rq
, rbd_dev
, snapc
, snapid
,
1350 seg_name
, seg_ofs
, seg_len
,
1356 rbd_req_cb
, 0, NULL
);
1358 rbd_coll_end_req_index(rq
, coll
, coll_index
,
1360 rbd_destroy_ops(ops
);
1367 * Request sync osd read
1369 static int rbd_req_sync_read(struct rbd_device
*rbd_dev
,
1371 const char *object_name
,
1376 struct ceph_osd_req_op
*ops
;
1379 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_READ
, 0);
1383 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1386 ops
, object_name
, ofs
, len
, buf
, NULL
, ver
);
1387 rbd_destroy_ops(ops
);
1393 * Request sync osd watch
1395 static int rbd_req_sync_notify_ack(struct rbd_device
*rbd_dev
,
1399 struct ceph_osd_req_op
*ops
;
1402 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_NOTIFY_ACK
, 0);
1406 ops
[0].watch
.ver
= cpu_to_le64(ver
);
1407 ops
[0].watch
.cookie
= notify_id
;
1408 ops
[0].watch
.flag
= 0;
1410 ret
= rbd_do_request(NULL
, rbd_dev
, NULL
, CEPH_NOSNAP
,
1411 rbd_dev
->header_name
, 0, 0, NULL
,
1416 rbd_simple_req_cb
, 0, NULL
);
1418 rbd_destroy_ops(ops
);
1422 static void rbd_watch_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1424 struct rbd_device
*rbd_dev
= (struct rbd_device
*)data
;
1431 dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
1432 rbd_dev
->header_name
, (unsigned long long) notify_id
,
1433 (unsigned int) opcode
);
1434 rc
= rbd_dev_refresh(rbd_dev
, &hver
);
1436 rbd_warn(rbd_dev
, "got notification but failed to "
1437 " update snaps: %d\n", rc
);
1439 rbd_req_sync_notify_ack(rbd_dev
, hver
, notify_id
);
1443 * Request sync osd watch
1445 static int rbd_req_sync_watch(struct rbd_device
*rbd_dev
)
1447 struct ceph_osd_req_op
*ops
;
1448 struct ceph_osd_client
*osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1451 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_WATCH
, 0);
1455 ret
= ceph_osdc_create_event(osdc
, rbd_watch_cb
, 0,
1456 (void *)rbd_dev
, &rbd_dev
->watch_event
);
1460 ops
[0].watch
.ver
= cpu_to_le64(rbd_dev
->header
.obj_version
);
1461 ops
[0].watch
.cookie
= cpu_to_le64(rbd_dev
->watch_event
->cookie
);
1462 ops
[0].watch
.flag
= 1;
1464 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1466 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1468 rbd_dev
->header_name
,
1470 &rbd_dev
->watch_request
, NULL
);
1475 rbd_destroy_ops(ops
);
1479 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1480 rbd_dev
->watch_event
= NULL
;
1482 rbd_destroy_ops(ops
);
1487 * Request sync osd unwatch
1489 static int rbd_req_sync_unwatch(struct rbd_device
*rbd_dev
)
1491 struct ceph_osd_req_op
*ops
;
1494 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_WATCH
, 0);
1498 ops
[0].watch
.ver
= 0;
1499 ops
[0].watch
.cookie
= cpu_to_le64(rbd_dev
->watch_event
->cookie
);
1500 ops
[0].watch
.flag
= 0;
1502 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1504 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1506 rbd_dev
->header_name
,
1507 0, 0, NULL
, NULL
, NULL
);
1510 rbd_destroy_ops(ops
);
1511 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1512 rbd_dev
->watch_event
= NULL
;
1517 * Synchronous osd object method call
1519 static int rbd_req_sync_exec(struct rbd_device
*rbd_dev
,
1520 const char *object_name
,
1521 const char *class_name
,
1522 const char *method_name
,
1523 const char *outbound
,
1524 size_t outbound_size
,
1526 size_t inbound_size
,
1530 struct ceph_osd_req_op
*ops
;
1531 int class_name_len
= strlen(class_name
);
1532 int method_name_len
= strlen(method_name
);
1537 * Any input parameters required by the method we're calling
1538 * will be sent along with the class and method names as
1539 * part of the message payload. That data and its size are
1540 * supplied via the indata and indata_len fields (named from
1541 * the perspective of the server side) in the OSD request
1544 payload_size
= class_name_len
+ method_name_len
+ outbound_size
;
1545 ops
= rbd_create_rw_ops(1, CEPH_OSD_OP_CALL
, payload_size
);
1549 ops
[0].cls
.class_name
= class_name
;
1550 ops
[0].cls
.class_len
= (__u8
) class_name_len
;
1551 ops
[0].cls
.method_name
= method_name
;
1552 ops
[0].cls
.method_len
= (__u8
) method_name_len
;
1553 ops
[0].cls
.argc
= 0;
1554 ops
[0].cls
.indata
= outbound
;
1555 ops
[0].cls
.indata_len
= outbound_size
;
1557 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1560 object_name
, 0, inbound_size
, inbound
,
1563 rbd_destroy_ops(ops
);
1565 dout("cls_exec returned %d\n", ret
);
1569 static struct rbd_req_coll
*rbd_alloc_coll(int num_reqs
)
1571 struct rbd_req_coll
*coll
=
1572 kzalloc(sizeof(struct rbd_req_coll
) +
1573 sizeof(struct rbd_req_status
) * num_reqs
,
1578 coll
->total
= num_reqs
;
1579 kref_init(&coll
->kref
);
1583 static int rbd_dev_do_request(struct request
*rq
,
1584 struct rbd_device
*rbd_dev
,
1585 struct ceph_snap_context
*snapc
,
1586 u64 ofs
, unsigned int size
,
1587 struct bio
*bio_chain
)
1590 struct rbd_req_coll
*coll
;
1591 unsigned int bio_offset
;
1594 dout("%s 0x%x bytes at 0x%llx\n",
1595 rq_data_dir(rq
) == WRITE
? "write" : "read",
1596 size
, (unsigned long long) blk_rq_pos(rq
) * SECTOR_SIZE
);
1598 num_segs
= rbd_get_num_segments(&rbd_dev
->header
, ofs
, size
);
1602 coll
= rbd_alloc_coll(num_segs
);
1608 u64 limit
= rbd_segment_length(rbd_dev
, ofs
, size
);
1609 unsigned int clone_size
;
1610 struct bio
*bio_clone
;
1612 BUG_ON(limit
> (u64
)UINT_MAX
);
1613 clone_size
= (unsigned int)limit
;
1614 dout("bio_chain->bi_vcnt=%hu\n", bio_chain
->bi_vcnt
);
1616 kref_get(&coll
->kref
);
1618 /* Pass a cloned bio chain via an osd request */
1620 bio_clone
= bio_chain_clone_range(&bio_chain
,
1621 &bio_offset
, clone_size
,
1624 (void)rbd_do_op(rq
, rbd_dev
, snapc
,
1626 bio_clone
, coll
, cur_seg
);
1628 rbd_coll_end_req_index(rq
, coll
, cur_seg
,
1636 kref_put(&coll
->kref
, rbd_coll_release
);
1642 * block device queue callback
1644 static void rbd_rq_fn(struct request_queue
*q
)
1646 struct rbd_device
*rbd_dev
= q
->queuedata
;
1647 bool read_only
= rbd_dev
->mapping
.read_only
;
1650 while ((rq
= blk_fetch_request(q
))) {
1651 struct ceph_snap_context
*snapc
= NULL
;
1652 unsigned int size
= 0;
1655 dout("fetched request\n");
1657 /* Filter out block requests we don't understand */
1659 if ((rq
->cmd_type
!= REQ_TYPE_FS
)) {
1660 __blk_end_request_all(rq
, 0);
1663 spin_unlock_irq(q
->queue_lock
);
1665 /* Stop writes to a read-only device */
1668 if (read_only
&& rq_data_dir(rq
) == WRITE
)
1669 goto out_end_request
;
1671 /* Grab a reference to the snapshot context */
1673 down_read(&rbd_dev
->header_rwsem
);
1674 if (atomic_read(&rbd_dev
->exists
)) {
1675 snapc
= ceph_get_snap_context(rbd_dev
->header
.snapc
);
1676 rbd_assert(snapc
!= NULL
);
1678 up_read(&rbd_dev
->header_rwsem
);
1681 rbd_assert(rbd_dev
->spec
->snap_id
!= CEPH_NOSNAP
);
1682 dout("request for non-existent snapshot");
1684 goto out_end_request
;
1687 size
= blk_rq_bytes(rq
);
1688 result
= rbd_dev_do_request(rq
, rbd_dev
, snapc
,
1689 blk_rq_pos(rq
) * SECTOR_SIZE
,
1692 ceph_put_snap_context(snapc
);
1693 spin_lock_irq(q
->queue_lock
);
1694 if (!size
|| result
< 0)
1695 __blk_end_request_all(rq
, result
);
1700 * a queue callback. Makes sure that we don't create a bio that spans across
1701 * multiple osd objects. One exception would be with a single page bios,
1702 * which we handle later at bio_chain_clone_range()
1704 static int rbd_merge_bvec(struct request_queue
*q
, struct bvec_merge_data
*bmd
,
1705 struct bio_vec
*bvec
)
1707 struct rbd_device
*rbd_dev
= q
->queuedata
;
1708 sector_t sector_offset
;
1709 sector_t sectors_per_obj
;
1710 sector_t obj_sector_offset
;
1714 * Find how far into its rbd object the partition-relative
1715 * bio start sector is to offset relative to the enclosing
1718 sector_offset
= get_start_sect(bmd
->bi_bdev
) + bmd
->bi_sector
;
1719 sectors_per_obj
= 1 << (rbd_dev
->header
.obj_order
- SECTOR_SHIFT
);
1720 obj_sector_offset
= sector_offset
& (sectors_per_obj
- 1);
1723 * Compute the number of bytes from that offset to the end
1724 * of the object. Account for what's already used by the bio.
1726 ret
= (int) (sectors_per_obj
- obj_sector_offset
) << SECTOR_SHIFT
;
1727 if (ret
> bmd
->bi_size
)
1728 ret
-= bmd
->bi_size
;
1733 * Don't send back more than was asked for. And if the bio
1734 * was empty, let the whole thing through because: "Note
1735 * that a block device *must* allow a single page to be
1736 * added to an empty bio."
1738 rbd_assert(bvec
->bv_len
<= PAGE_SIZE
);
1739 if (ret
> (int) bvec
->bv_len
|| !bmd
->bi_size
)
1740 ret
= (int) bvec
->bv_len
;
1745 static void rbd_free_disk(struct rbd_device
*rbd_dev
)
1747 struct gendisk
*disk
= rbd_dev
->disk
;
1752 if (disk
->flags
& GENHD_FL_UP
)
1755 blk_cleanup_queue(disk
->queue
);
1760 * Read the complete header for the given rbd device.
1762 * Returns a pointer to a dynamically-allocated buffer containing
1763 * the complete and validated header. Caller can pass the address
1764 * of a variable that will be filled in with the version of the
1765 * header object at the time it was read.
1767 * Returns a pointer-coded errno if a failure occurs.
1769 static struct rbd_image_header_ondisk
*
1770 rbd_dev_v1_header_read(struct rbd_device
*rbd_dev
, u64
*version
)
1772 struct rbd_image_header_ondisk
*ondisk
= NULL
;
1779 * The complete header will include an array of its 64-bit
1780 * snapshot ids, followed by the names of those snapshots as
1781 * a contiguous block of NUL-terminated strings. Note that
1782 * the number of snapshots could change by the time we read
1783 * it in, in which case we re-read it.
1790 size
= sizeof (*ondisk
);
1791 size
+= snap_count
* sizeof (struct rbd_image_snap_ondisk
);
1793 ondisk
= kmalloc(size
, GFP_KERNEL
);
1795 return ERR_PTR(-ENOMEM
);
1797 ret
= rbd_req_sync_read(rbd_dev
, CEPH_NOSNAP
,
1798 rbd_dev
->header_name
,
1800 (char *) ondisk
, version
);
1804 if (WARN_ON((size_t) ret
< size
)) {
1806 rbd_warn(rbd_dev
, "short header read (want %zd got %d)",
1810 if (!rbd_dev_ondisk_valid(ondisk
)) {
1812 rbd_warn(rbd_dev
, "invalid header");
1816 names_size
= le64_to_cpu(ondisk
->snap_names_len
);
1817 want_count
= snap_count
;
1818 snap_count
= le32_to_cpu(ondisk
->snap_count
);
1819 } while (snap_count
!= want_count
);
1826 return ERR_PTR(ret
);
1830 * reload the ondisk the header
1832 static int rbd_read_header(struct rbd_device
*rbd_dev
,
1833 struct rbd_image_header
*header
)
1835 struct rbd_image_header_ondisk
*ondisk
;
1839 ondisk
= rbd_dev_v1_header_read(rbd_dev
, &ver
);
1841 return PTR_ERR(ondisk
);
1842 ret
= rbd_header_from_disk(header
, ondisk
);
1844 header
->obj_version
= ver
;
1850 static void rbd_remove_all_snaps(struct rbd_device
*rbd_dev
)
1852 struct rbd_snap
*snap
;
1853 struct rbd_snap
*next
;
1855 list_for_each_entry_safe(snap
, next
, &rbd_dev
->snaps
, node
)
1856 rbd_remove_snap_dev(snap
);
1859 static void rbd_update_mapping_size(struct rbd_device
*rbd_dev
)
1863 if (rbd_dev
->spec
->snap_id
!= CEPH_NOSNAP
)
1866 size
= (sector_t
) rbd_dev
->header
.image_size
/ SECTOR_SIZE
;
1867 dout("setting size to %llu sectors", (unsigned long long) size
);
1868 rbd_dev
->mapping
.size
= (u64
) size
;
1869 set_capacity(rbd_dev
->disk
, size
);
1873 * only read the first part of the ondisk header, without the snaps info
1875 static int rbd_dev_v1_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
1878 struct rbd_image_header h
;
1880 ret
= rbd_read_header(rbd_dev
, &h
);
1884 down_write(&rbd_dev
->header_rwsem
);
1886 /* Update image size, and check for resize of mapped image */
1887 rbd_dev
->header
.image_size
= h
.image_size
;
1888 rbd_update_mapping_size(rbd_dev
);
1890 /* rbd_dev->header.object_prefix shouldn't change */
1891 kfree(rbd_dev
->header
.snap_sizes
);
1892 kfree(rbd_dev
->header
.snap_names
);
1893 /* osd requests may still refer to snapc */
1894 ceph_put_snap_context(rbd_dev
->header
.snapc
);
1897 *hver
= h
.obj_version
;
1898 rbd_dev
->header
.obj_version
= h
.obj_version
;
1899 rbd_dev
->header
.image_size
= h
.image_size
;
1900 rbd_dev
->header
.snapc
= h
.snapc
;
1901 rbd_dev
->header
.snap_names
= h
.snap_names
;
1902 rbd_dev
->header
.snap_sizes
= h
.snap_sizes
;
1903 /* Free the extra copy of the object prefix */
1904 WARN_ON(strcmp(rbd_dev
->header
.object_prefix
, h
.object_prefix
));
1905 kfree(h
.object_prefix
);
1907 ret
= rbd_dev_snaps_update(rbd_dev
);
1909 ret
= rbd_dev_snaps_register(rbd_dev
);
1911 up_write(&rbd_dev
->header_rwsem
);
1916 static int rbd_dev_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
1920 rbd_assert(rbd_image_format_valid(rbd_dev
->image_format
));
1921 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1922 if (rbd_dev
->image_format
== 1)
1923 ret
= rbd_dev_v1_refresh(rbd_dev
, hver
);
1925 ret
= rbd_dev_v2_refresh(rbd_dev
, hver
);
1926 mutex_unlock(&ctl_mutex
);
1931 static int rbd_init_disk(struct rbd_device
*rbd_dev
)
1933 struct gendisk
*disk
;
1934 struct request_queue
*q
;
1937 /* create gendisk info */
1938 disk
= alloc_disk(RBD_MINORS_PER_MAJOR
);
1942 snprintf(disk
->disk_name
, sizeof(disk
->disk_name
), RBD_DRV_NAME
"%d",
1944 disk
->major
= rbd_dev
->major
;
1945 disk
->first_minor
= 0;
1946 disk
->fops
= &rbd_bd_ops
;
1947 disk
->private_data
= rbd_dev
;
1950 q
= blk_init_queue(rbd_rq_fn
, &rbd_dev
->lock
);
1954 /* We use the default size, but let's be explicit about it. */
1955 blk_queue_physical_block_size(q
, SECTOR_SIZE
);
1957 /* set io sizes to object size */
1958 segment_size
= rbd_obj_bytes(&rbd_dev
->header
);
1959 blk_queue_max_hw_sectors(q
, segment_size
/ SECTOR_SIZE
);
1960 blk_queue_max_segment_size(q
, segment_size
);
1961 blk_queue_io_min(q
, segment_size
);
1962 blk_queue_io_opt(q
, segment_size
);
1964 blk_queue_merge_bvec(q
, rbd_merge_bvec
);
1967 q
->queuedata
= rbd_dev
;
1969 rbd_dev
->disk
= disk
;
1971 set_capacity(rbd_dev
->disk
, rbd_dev
->mapping
.size
/ SECTOR_SIZE
);
1984 static struct rbd_device
*dev_to_rbd_dev(struct device
*dev
)
1986 return container_of(dev
, struct rbd_device
, dev
);
1989 static ssize_t
rbd_size_show(struct device
*dev
,
1990 struct device_attribute
*attr
, char *buf
)
1992 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1995 down_read(&rbd_dev
->header_rwsem
);
1996 size
= get_capacity(rbd_dev
->disk
);
1997 up_read(&rbd_dev
->header_rwsem
);
1999 return sprintf(buf
, "%llu\n", (unsigned long long) size
* SECTOR_SIZE
);
2003 * Note this shows the features for whatever's mapped, which is not
2004 * necessarily the base image.
2006 static ssize_t
rbd_features_show(struct device
*dev
,
2007 struct device_attribute
*attr
, char *buf
)
2009 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2011 return sprintf(buf
, "0x%016llx\n",
2012 (unsigned long long) rbd_dev
->mapping
.features
);
2015 static ssize_t
rbd_major_show(struct device
*dev
,
2016 struct device_attribute
*attr
, char *buf
)
2018 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2020 return sprintf(buf
, "%d\n", rbd_dev
->major
);
2023 static ssize_t
rbd_client_id_show(struct device
*dev
,
2024 struct device_attribute
*attr
, char *buf
)
2026 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2028 return sprintf(buf
, "client%lld\n",
2029 ceph_client_id(rbd_dev
->rbd_client
->client
));
2032 static ssize_t
rbd_pool_show(struct device
*dev
,
2033 struct device_attribute
*attr
, char *buf
)
2035 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2037 return sprintf(buf
, "%s\n", rbd_dev
->spec
->pool_name
);
2040 static ssize_t
rbd_pool_id_show(struct device
*dev
,
2041 struct device_attribute
*attr
, char *buf
)
2043 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2045 return sprintf(buf
, "%llu\n",
2046 (unsigned long long) rbd_dev
->spec
->pool_id
);
2049 static ssize_t
rbd_name_show(struct device
*dev
,
2050 struct device_attribute
*attr
, char *buf
)
2052 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2054 if (rbd_dev
->spec
->image_name
)
2055 return sprintf(buf
, "%s\n", rbd_dev
->spec
->image_name
);
2057 return sprintf(buf
, "(unknown)\n");
2060 static ssize_t
rbd_image_id_show(struct device
*dev
,
2061 struct device_attribute
*attr
, char *buf
)
2063 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2065 return sprintf(buf
, "%s\n", rbd_dev
->spec
->image_id
);
2069 * Shows the name of the currently-mapped snapshot (or
2070 * RBD_SNAP_HEAD_NAME for the base image).
2072 static ssize_t
rbd_snap_show(struct device
*dev
,
2073 struct device_attribute
*attr
,
2076 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2078 return sprintf(buf
, "%s\n", rbd_dev
->spec
->snap_name
);
2082 * For an rbd v2 image, shows the pool id, image id, and snapshot id
2083 * for the parent image. If there is no parent, simply shows
2084 * "(no parent image)".
2086 static ssize_t
rbd_parent_show(struct device
*dev
,
2087 struct device_attribute
*attr
,
2090 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2091 struct rbd_spec
*spec
= rbd_dev
->parent_spec
;
2096 return sprintf(buf
, "(no parent image)\n");
2098 count
= sprintf(bufp
, "pool_id %llu\npool_name %s\n",
2099 (unsigned long long) spec
->pool_id
, spec
->pool_name
);
2104 count
= sprintf(bufp
, "image_id %s\nimage_name %s\n", spec
->image_id
,
2105 spec
->image_name
? spec
->image_name
: "(unknown)");
2110 count
= sprintf(bufp
, "snap_id %llu\nsnap_name %s\n",
2111 (unsigned long long) spec
->snap_id
, spec
->snap_name
);
2116 count
= sprintf(bufp
, "overlap %llu\n", rbd_dev
->parent_overlap
);
2121 return (ssize_t
) (bufp
- buf
);
2124 static ssize_t
rbd_image_refresh(struct device
*dev
,
2125 struct device_attribute
*attr
,
2129 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2132 ret
= rbd_dev_refresh(rbd_dev
, NULL
);
2134 return ret
< 0 ? ret
: size
;
2137 static DEVICE_ATTR(size
, S_IRUGO
, rbd_size_show
, NULL
);
2138 static DEVICE_ATTR(features
, S_IRUGO
, rbd_features_show
, NULL
);
2139 static DEVICE_ATTR(major
, S_IRUGO
, rbd_major_show
, NULL
);
2140 static DEVICE_ATTR(client_id
, S_IRUGO
, rbd_client_id_show
, NULL
);
2141 static DEVICE_ATTR(pool
, S_IRUGO
, rbd_pool_show
, NULL
);
2142 static DEVICE_ATTR(pool_id
, S_IRUGO
, rbd_pool_id_show
, NULL
);
2143 static DEVICE_ATTR(name
, S_IRUGO
, rbd_name_show
, NULL
);
2144 static DEVICE_ATTR(image_id
, S_IRUGO
, rbd_image_id_show
, NULL
);
2145 static DEVICE_ATTR(refresh
, S_IWUSR
, NULL
, rbd_image_refresh
);
2146 static DEVICE_ATTR(current_snap
, S_IRUGO
, rbd_snap_show
, NULL
);
2147 static DEVICE_ATTR(parent
, S_IRUGO
, rbd_parent_show
, NULL
);
2149 static struct attribute
*rbd_attrs
[] = {
2150 &dev_attr_size
.attr
,
2151 &dev_attr_features
.attr
,
2152 &dev_attr_major
.attr
,
2153 &dev_attr_client_id
.attr
,
2154 &dev_attr_pool
.attr
,
2155 &dev_attr_pool_id
.attr
,
2156 &dev_attr_name
.attr
,
2157 &dev_attr_image_id
.attr
,
2158 &dev_attr_current_snap
.attr
,
2159 &dev_attr_parent
.attr
,
2160 &dev_attr_refresh
.attr
,
2164 static struct attribute_group rbd_attr_group
= {
2168 static const struct attribute_group
*rbd_attr_groups
[] = {
2173 static void rbd_sysfs_dev_release(struct device
*dev
)
2177 static struct device_type rbd_device_type
= {
2179 .groups
= rbd_attr_groups
,
2180 .release
= rbd_sysfs_dev_release
,
2188 static ssize_t
rbd_snap_size_show(struct device
*dev
,
2189 struct device_attribute
*attr
,
2192 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2194 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->size
);
2197 static ssize_t
rbd_snap_id_show(struct device
*dev
,
2198 struct device_attribute
*attr
,
2201 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2203 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->id
);
2206 static ssize_t
rbd_snap_features_show(struct device
*dev
,
2207 struct device_attribute
*attr
,
2210 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2212 return sprintf(buf
, "0x%016llx\n",
2213 (unsigned long long) snap
->features
);
2216 static DEVICE_ATTR(snap_size
, S_IRUGO
, rbd_snap_size_show
, NULL
);
2217 static DEVICE_ATTR(snap_id
, S_IRUGO
, rbd_snap_id_show
, NULL
);
2218 static DEVICE_ATTR(snap_features
, S_IRUGO
, rbd_snap_features_show
, NULL
);
2220 static struct attribute
*rbd_snap_attrs
[] = {
2221 &dev_attr_snap_size
.attr
,
2222 &dev_attr_snap_id
.attr
,
2223 &dev_attr_snap_features
.attr
,
2227 static struct attribute_group rbd_snap_attr_group
= {
2228 .attrs
= rbd_snap_attrs
,
2231 static void rbd_snap_dev_release(struct device
*dev
)
2233 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2238 static const struct attribute_group
*rbd_snap_attr_groups
[] = {
2239 &rbd_snap_attr_group
,
2243 static struct device_type rbd_snap_device_type
= {
2244 .groups
= rbd_snap_attr_groups
,
2245 .release
= rbd_snap_dev_release
,
2248 static struct rbd_spec
*rbd_spec_get(struct rbd_spec
*spec
)
2250 kref_get(&spec
->kref
);
2255 static void rbd_spec_free(struct kref
*kref
);
2256 static void rbd_spec_put(struct rbd_spec
*spec
)
2259 kref_put(&spec
->kref
, rbd_spec_free
);
2262 static struct rbd_spec
*rbd_spec_alloc(void)
2264 struct rbd_spec
*spec
;
2266 spec
= kzalloc(sizeof (*spec
), GFP_KERNEL
);
2269 kref_init(&spec
->kref
);
2271 rbd_spec_put(rbd_spec_get(spec
)); /* TEMPORARY */
2276 static void rbd_spec_free(struct kref
*kref
)
2278 struct rbd_spec
*spec
= container_of(kref
, struct rbd_spec
, kref
);
2280 kfree(spec
->pool_name
);
2281 kfree(spec
->image_id
);
2282 kfree(spec
->image_name
);
2283 kfree(spec
->snap_name
);
2287 struct rbd_device
*rbd_dev_create(struct rbd_client
*rbdc
,
2288 struct rbd_spec
*spec
)
2290 struct rbd_device
*rbd_dev
;
2292 rbd_dev
= kzalloc(sizeof (*rbd_dev
), GFP_KERNEL
);
2296 spin_lock_init(&rbd_dev
->lock
);
2297 atomic_set(&rbd_dev
->exists
, 0);
2298 INIT_LIST_HEAD(&rbd_dev
->node
);
2299 INIT_LIST_HEAD(&rbd_dev
->snaps
);
2300 init_rwsem(&rbd_dev
->header_rwsem
);
2302 rbd_dev
->spec
= spec
;
2303 rbd_dev
->rbd_client
= rbdc
;
2308 static void rbd_dev_destroy(struct rbd_device
*rbd_dev
)
2310 rbd_spec_put(rbd_dev
->parent_spec
);
2311 kfree(rbd_dev
->header_name
);
2312 rbd_put_client(rbd_dev
->rbd_client
);
2313 rbd_spec_put(rbd_dev
->spec
);
2317 static bool rbd_snap_registered(struct rbd_snap
*snap
)
2319 bool ret
= snap
->dev
.type
== &rbd_snap_device_type
;
2320 bool reg
= device_is_registered(&snap
->dev
);
2322 rbd_assert(!ret
^ reg
);
2327 static void rbd_remove_snap_dev(struct rbd_snap
*snap
)
2329 list_del(&snap
->node
);
2330 if (device_is_registered(&snap
->dev
))
2331 device_unregister(&snap
->dev
);
2334 static int rbd_register_snap_dev(struct rbd_snap
*snap
,
2335 struct device
*parent
)
2337 struct device
*dev
= &snap
->dev
;
2340 dev
->type
= &rbd_snap_device_type
;
2341 dev
->parent
= parent
;
2342 dev
->release
= rbd_snap_dev_release
;
2343 dev_set_name(dev
, "%s%s", RBD_SNAP_DEV_NAME_PREFIX
, snap
->name
);
2344 dout("%s: registering device for snapshot %s\n", __func__
, snap
->name
);
2346 ret
= device_register(dev
);
2351 static struct rbd_snap
*__rbd_add_snap_dev(struct rbd_device
*rbd_dev
,
2352 const char *snap_name
,
2353 u64 snap_id
, u64 snap_size
,
2356 struct rbd_snap
*snap
;
2359 snap
= kzalloc(sizeof (*snap
), GFP_KERNEL
);
2361 return ERR_PTR(-ENOMEM
);
2364 snap
->name
= kstrdup(snap_name
, GFP_KERNEL
);
2369 snap
->size
= snap_size
;
2370 snap
->features
= snap_features
;
2378 return ERR_PTR(ret
);
2381 static char *rbd_dev_v1_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2382 u64
*snap_size
, u64
*snap_features
)
2386 rbd_assert(which
< rbd_dev
->header
.snapc
->num_snaps
);
2388 *snap_size
= rbd_dev
->header
.snap_sizes
[which
];
2389 *snap_features
= 0; /* No features for v1 */
2391 /* Skip over names until we find the one we are looking for */
2393 snap_name
= rbd_dev
->header
.snap_names
;
2395 snap_name
+= strlen(snap_name
) + 1;
2401 * Get the size and object order for an image snapshot, or if
2402 * snap_id is CEPH_NOSNAP, gets this information for the base
2405 static int _rbd_dev_v2_snap_size(struct rbd_device
*rbd_dev
, u64 snap_id
,
2406 u8
*order
, u64
*snap_size
)
2408 __le64 snapid
= cpu_to_le64(snap_id
);
2413 } __attribute__ ((packed
)) size_buf
= { 0 };
2415 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2417 (char *) &snapid
, sizeof (snapid
),
2418 (char *) &size_buf
, sizeof (size_buf
),
2419 CEPH_OSD_FLAG_READ
, NULL
);
2420 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2424 *order
= size_buf
.order
;
2425 *snap_size
= le64_to_cpu(size_buf
.size
);
2427 dout(" snap_id 0x%016llx order = %u, snap_size = %llu\n",
2428 (unsigned long long) snap_id
, (unsigned int) *order
,
2429 (unsigned long long) *snap_size
);
2434 static int rbd_dev_v2_image_size(struct rbd_device
*rbd_dev
)
2436 return _rbd_dev_v2_snap_size(rbd_dev
, CEPH_NOSNAP
,
2437 &rbd_dev
->header
.obj_order
,
2438 &rbd_dev
->header
.image_size
);
2441 static int rbd_dev_v2_object_prefix(struct rbd_device
*rbd_dev
)
2447 reply_buf
= kzalloc(RBD_OBJ_PREFIX_LEN_MAX
, GFP_KERNEL
);
2451 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2452 "rbd", "get_object_prefix",
2454 reply_buf
, RBD_OBJ_PREFIX_LEN_MAX
,
2455 CEPH_OSD_FLAG_READ
, NULL
);
2456 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2459 ret
= 0; /* rbd_req_sync_exec() can return positive */
2462 rbd_dev
->header
.object_prefix
= ceph_extract_encoded_string(&p
,
2463 p
+ RBD_OBJ_PREFIX_LEN_MAX
,
2466 if (IS_ERR(rbd_dev
->header
.object_prefix
)) {
2467 ret
= PTR_ERR(rbd_dev
->header
.object_prefix
);
2468 rbd_dev
->header
.object_prefix
= NULL
;
2470 dout(" object_prefix = %s\n", rbd_dev
->header
.object_prefix
);
2479 static int _rbd_dev_v2_snap_features(struct rbd_device
*rbd_dev
, u64 snap_id
,
2482 __le64 snapid
= cpu_to_le64(snap_id
);
2486 } features_buf
= { 0 };
2490 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2491 "rbd", "get_features",
2492 (char *) &snapid
, sizeof (snapid
),
2493 (char *) &features_buf
, sizeof (features_buf
),
2494 CEPH_OSD_FLAG_READ
, NULL
);
2495 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2499 incompat
= le64_to_cpu(features_buf
.incompat
);
2500 if (incompat
& ~RBD_FEATURES_ALL
)
2503 *snap_features
= le64_to_cpu(features_buf
.features
);
2505 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
2506 (unsigned long long) snap_id
,
2507 (unsigned long long) *snap_features
,
2508 (unsigned long long) le64_to_cpu(features_buf
.incompat
));
2513 static int rbd_dev_v2_features(struct rbd_device
*rbd_dev
)
2515 return _rbd_dev_v2_snap_features(rbd_dev
, CEPH_NOSNAP
,
2516 &rbd_dev
->header
.features
);
2519 static int rbd_dev_v2_parent_info(struct rbd_device
*rbd_dev
)
2521 struct rbd_spec
*parent_spec
;
2523 void *reply_buf
= NULL
;
2531 parent_spec
= rbd_spec_alloc();
2535 size
= sizeof (__le64
) + /* pool_id */
2536 sizeof (__le32
) + RBD_IMAGE_ID_LEN_MAX
+ /* image_id */
2537 sizeof (__le64
) + /* snap_id */
2538 sizeof (__le64
); /* overlap */
2539 reply_buf
= kmalloc(size
, GFP_KERNEL
);
2545 snapid
= cpu_to_le64(CEPH_NOSNAP
);
2546 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2547 "rbd", "get_parent",
2548 (char *) &snapid
, sizeof (snapid
),
2549 (char *) reply_buf
, size
,
2550 CEPH_OSD_FLAG_READ
, NULL
);
2551 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2557 end
= (char *) reply_buf
+ size
;
2558 ceph_decode_64_safe(&p
, end
, parent_spec
->pool_id
, out_err
);
2559 if (parent_spec
->pool_id
== CEPH_NOPOOL
)
2560 goto out
; /* No parent? No problem. */
2562 image_id
= ceph_extract_encoded_string(&p
, end
, NULL
, GFP_KERNEL
);
2563 if (IS_ERR(image_id
)) {
2564 ret
= PTR_ERR(image_id
);
2567 parent_spec
->image_id
= image_id
;
2568 ceph_decode_64_safe(&p
, end
, parent_spec
->snap_id
, out_err
);
2569 ceph_decode_64_safe(&p
, end
, overlap
, out_err
);
2571 rbd_dev
->parent_overlap
= overlap
;
2572 rbd_dev
->parent_spec
= parent_spec
;
2573 parent_spec
= NULL
; /* rbd_dev now owns this */
2578 rbd_spec_put(parent_spec
);
2583 static char *rbd_dev_image_name(struct rbd_device
*rbd_dev
)
2585 size_t image_id_size
;
2590 void *reply_buf
= NULL
;
2592 char *image_name
= NULL
;
2595 rbd_assert(!rbd_dev
->spec
->image_name
);
2597 len
= strlen(rbd_dev
->spec
->image_id
);
2598 image_id_size
= sizeof (__le32
) + len
;
2599 image_id
= kmalloc(image_id_size
, GFP_KERNEL
);
2604 end
= (char *) image_id
+ image_id_size
;
2605 ceph_encode_string(&p
, end
, rbd_dev
->spec
->image_id
, (u32
) len
);
2607 size
= sizeof (__le32
) + RBD_IMAGE_NAME_LEN_MAX
;
2608 reply_buf
= kmalloc(size
, GFP_KERNEL
);
2612 ret
= rbd_req_sync_exec(rbd_dev
, RBD_DIRECTORY
,
2613 "rbd", "dir_get_name",
2614 image_id
, image_id_size
,
2615 (char *) reply_buf
, size
,
2616 CEPH_OSD_FLAG_READ
, NULL
);
2620 end
= (char *) reply_buf
+ size
;
2621 image_name
= ceph_extract_encoded_string(&p
, end
, &len
, GFP_KERNEL
);
2622 if (IS_ERR(image_name
))
2625 dout("%s: name is %s len is %zd\n", __func__
, image_name
, len
);
2634 * When a parent image gets probed, we only have the pool, image,
2635 * and snapshot ids but not the names of any of them. This call
2636 * is made later to fill in those names. It has to be done after
2637 * rbd_dev_snaps_update() has completed because some of the
2638 * information (in particular, snapshot name) is not available
2641 static int rbd_dev_probe_update_spec(struct rbd_device
*rbd_dev
)
2643 struct ceph_osd_client
*osdc
;
2645 void *reply_buf
= NULL
;
2648 if (rbd_dev
->spec
->pool_name
)
2649 return 0; /* Already have the names */
2651 /* Look up the pool name */
2653 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
2654 name
= ceph_pg_pool_name_by_id(osdc
->osdmap
, rbd_dev
->spec
->pool_id
);
2656 rbd_warn(rbd_dev
, "there is no pool with id %llu",
2657 rbd_dev
->spec
->pool_id
); /* Really a BUG() */
2661 rbd_dev
->spec
->pool_name
= kstrdup(name
, GFP_KERNEL
);
2662 if (!rbd_dev
->spec
->pool_name
)
2665 /* Fetch the image name; tolerate failure here */
2667 name
= rbd_dev_image_name(rbd_dev
);
2669 rbd_dev
->spec
->image_name
= (char *) name
;
2671 rbd_warn(rbd_dev
, "unable to get image name");
2673 /* Look up the snapshot name. */
2675 name
= rbd_snap_name(rbd_dev
, rbd_dev
->spec
->snap_id
);
2677 rbd_warn(rbd_dev
, "no snapshot with id %llu",
2678 rbd_dev
->spec
->snap_id
); /* Really a BUG() */
2682 rbd_dev
->spec
->snap_name
= kstrdup(name
, GFP_KERNEL
);
2683 if(!rbd_dev
->spec
->snap_name
)
2689 kfree(rbd_dev
->spec
->pool_name
);
2690 rbd_dev
->spec
->pool_name
= NULL
;
2695 static int rbd_dev_v2_snap_context(struct rbd_device
*rbd_dev
, u64
*ver
)
2704 struct ceph_snap_context
*snapc
;
2708 * We'll need room for the seq value (maximum snapshot id),
2709 * snapshot count, and array of that many snapshot ids.
2710 * For now we have a fixed upper limit on the number we're
2711 * prepared to receive.
2713 size
= sizeof (__le64
) + sizeof (__le32
) +
2714 RBD_MAX_SNAP_COUNT
* sizeof (__le64
);
2715 reply_buf
= kzalloc(size
, GFP_KERNEL
);
2719 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2720 "rbd", "get_snapcontext",
2723 CEPH_OSD_FLAG_READ
, ver
);
2724 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2730 end
= (char *) reply_buf
+ size
;
2731 ceph_decode_64_safe(&p
, end
, seq
, out
);
2732 ceph_decode_32_safe(&p
, end
, snap_count
, out
);
2735 * Make sure the reported number of snapshot ids wouldn't go
2736 * beyond the end of our buffer. But before checking that,
2737 * make sure the computed size of the snapshot context we
2738 * allocate is representable in a size_t.
2740 if (snap_count
> (SIZE_MAX
- sizeof (struct ceph_snap_context
))
2745 if (!ceph_has_room(&p
, end
, snap_count
* sizeof (__le64
)))
2748 size
= sizeof (struct ceph_snap_context
) +
2749 snap_count
* sizeof (snapc
->snaps
[0]);
2750 snapc
= kmalloc(size
, GFP_KERNEL
);
2756 atomic_set(&snapc
->nref
, 1);
2758 snapc
->num_snaps
= snap_count
;
2759 for (i
= 0; i
< snap_count
; i
++)
2760 snapc
->snaps
[i
] = ceph_decode_64(&p
);
2762 rbd_dev
->header
.snapc
= snapc
;
2764 dout(" snap context seq = %llu, snap_count = %u\n",
2765 (unsigned long long) seq
, (unsigned int) snap_count
);
2773 static char *rbd_dev_v2_snap_name(struct rbd_device
*rbd_dev
, u32 which
)
2783 size
= sizeof (__le32
) + RBD_MAX_SNAP_NAME_LEN
;
2784 reply_buf
= kmalloc(size
, GFP_KERNEL
);
2786 return ERR_PTR(-ENOMEM
);
2788 snap_id
= cpu_to_le64(rbd_dev
->header
.snapc
->snaps
[which
]);
2789 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2790 "rbd", "get_snapshot_name",
2791 (char *) &snap_id
, sizeof (snap_id
),
2793 CEPH_OSD_FLAG_READ
, NULL
);
2794 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2799 end
= (char *) reply_buf
+ size
;
2800 snap_name
= ceph_extract_encoded_string(&p
, end
, NULL
, GFP_KERNEL
);
2801 if (IS_ERR(snap_name
)) {
2802 ret
= PTR_ERR(snap_name
);
2805 dout(" snap_id 0x%016llx snap_name = %s\n",
2806 (unsigned long long) le64_to_cpu(snap_id
), snap_name
);
2814 return ERR_PTR(ret
);
2817 static char *rbd_dev_v2_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2818 u64
*snap_size
, u64
*snap_features
)
2824 snap_id
= rbd_dev
->header
.snapc
->snaps
[which
];
2825 ret
= _rbd_dev_v2_snap_size(rbd_dev
, snap_id
, &order
, snap_size
);
2827 return ERR_PTR(ret
);
2828 ret
= _rbd_dev_v2_snap_features(rbd_dev
, snap_id
, snap_features
);
2830 return ERR_PTR(ret
);
2832 return rbd_dev_v2_snap_name(rbd_dev
, which
);
2835 static char *rbd_dev_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2836 u64
*snap_size
, u64
*snap_features
)
2838 if (rbd_dev
->image_format
== 1)
2839 return rbd_dev_v1_snap_info(rbd_dev
, which
,
2840 snap_size
, snap_features
);
2841 if (rbd_dev
->image_format
== 2)
2842 return rbd_dev_v2_snap_info(rbd_dev
, which
,
2843 snap_size
, snap_features
);
2844 return ERR_PTR(-EINVAL
);
2847 static int rbd_dev_v2_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
2852 down_write(&rbd_dev
->header_rwsem
);
2854 /* Grab old order first, to see if it changes */
2856 obj_order
= rbd_dev
->header
.obj_order
,
2857 ret
= rbd_dev_v2_image_size(rbd_dev
);
2860 if (rbd_dev
->header
.obj_order
!= obj_order
) {
2864 rbd_update_mapping_size(rbd_dev
);
2866 ret
= rbd_dev_v2_snap_context(rbd_dev
, hver
);
2867 dout("rbd_dev_v2_snap_context returned %d\n", ret
);
2870 ret
= rbd_dev_snaps_update(rbd_dev
);
2871 dout("rbd_dev_snaps_update returned %d\n", ret
);
2874 ret
= rbd_dev_snaps_register(rbd_dev
);
2875 dout("rbd_dev_snaps_register returned %d\n", ret
);
2877 up_write(&rbd_dev
->header_rwsem
);
2883 * Scan the rbd device's current snapshot list and compare it to the
2884 * newly-received snapshot context. Remove any existing snapshots
2885 * not present in the new snapshot context. Add a new snapshot for
2886 * any snaphots in the snapshot context not in the current list.
2887 * And verify there are no changes to snapshots we already know
2890 * Assumes the snapshots in the snapshot context are sorted by
2891 * snapshot id, highest id first. (Snapshots in the rbd_dev's list
2892 * are also maintained in that order.)
2894 static int rbd_dev_snaps_update(struct rbd_device
*rbd_dev
)
2896 struct ceph_snap_context
*snapc
= rbd_dev
->header
.snapc
;
2897 const u32 snap_count
= snapc
->num_snaps
;
2898 struct list_head
*head
= &rbd_dev
->snaps
;
2899 struct list_head
*links
= head
->next
;
2902 dout("%s: snap count is %u\n", __func__
, (unsigned int) snap_count
);
2903 while (index
< snap_count
|| links
!= head
) {
2905 struct rbd_snap
*snap
;
2908 u64 snap_features
= 0;
2910 snap_id
= index
< snap_count
? snapc
->snaps
[index
]
2912 snap
= links
!= head
? list_entry(links
, struct rbd_snap
, node
)
2914 rbd_assert(!snap
|| snap
->id
!= CEPH_NOSNAP
);
2916 if (snap_id
== CEPH_NOSNAP
|| (snap
&& snap
->id
> snap_id
)) {
2917 struct list_head
*next
= links
->next
;
2919 /* Existing snapshot not in the new snap context */
2921 if (rbd_dev
->spec
->snap_id
== snap
->id
)
2922 atomic_set(&rbd_dev
->exists
, 0);
2923 rbd_remove_snap_dev(snap
);
2924 dout("%ssnap id %llu has been removed\n",
2925 rbd_dev
->spec
->snap_id
== snap
->id
?
2927 (unsigned long long) snap
->id
);
2929 /* Done with this list entry; advance */
2935 snap_name
= rbd_dev_snap_info(rbd_dev
, index
,
2936 &snap_size
, &snap_features
);
2937 if (IS_ERR(snap_name
))
2938 return PTR_ERR(snap_name
);
2940 dout("entry %u: snap_id = %llu\n", (unsigned int) snap_count
,
2941 (unsigned long long) snap_id
);
2942 if (!snap
|| (snap_id
!= CEPH_NOSNAP
&& snap
->id
< snap_id
)) {
2943 struct rbd_snap
*new_snap
;
2945 /* We haven't seen this snapshot before */
2947 new_snap
= __rbd_add_snap_dev(rbd_dev
, snap_name
,
2948 snap_id
, snap_size
, snap_features
);
2949 if (IS_ERR(new_snap
)) {
2950 int err
= PTR_ERR(new_snap
);
2952 dout(" failed to add dev, error %d\n", err
);
2957 /* New goes before existing, or at end of list */
2959 dout(" added dev%s\n", snap
? "" : " at end\n");
2961 list_add_tail(&new_snap
->node
, &snap
->node
);
2963 list_add_tail(&new_snap
->node
, head
);
2965 /* Already have this one */
2967 dout(" already present\n");
2969 rbd_assert(snap
->size
== snap_size
);
2970 rbd_assert(!strcmp(snap
->name
, snap_name
));
2971 rbd_assert(snap
->features
== snap_features
);
2973 /* Done with this list entry; advance */
2975 links
= links
->next
;
2978 /* Advance to the next entry in the snapshot context */
2982 dout("%s: done\n", __func__
);
2988 * Scan the list of snapshots and register the devices for any that
2989 * have not already been registered.
2991 static int rbd_dev_snaps_register(struct rbd_device
*rbd_dev
)
2993 struct rbd_snap
*snap
;
2996 dout("%s called\n", __func__
);
2997 if (WARN_ON(!device_is_registered(&rbd_dev
->dev
)))
3000 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
3001 if (!rbd_snap_registered(snap
)) {
3002 ret
= rbd_register_snap_dev(snap
, &rbd_dev
->dev
);
3007 dout("%s: returning %d\n", __func__
, ret
);
3012 static int rbd_bus_add_dev(struct rbd_device
*rbd_dev
)
3017 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
3019 dev
= &rbd_dev
->dev
;
3020 dev
->bus
= &rbd_bus_type
;
3021 dev
->type
= &rbd_device_type
;
3022 dev
->parent
= &rbd_root_dev
;
3023 dev
->release
= rbd_dev_release
;
3024 dev_set_name(dev
, "%d", rbd_dev
->dev_id
);
3025 ret
= device_register(dev
);
3027 mutex_unlock(&ctl_mutex
);
3032 static void rbd_bus_del_dev(struct rbd_device
*rbd_dev
)
3034 device_unregister(&rbd_dev
->dev
);
3037 static int rbd_init_watch_dev(struct rbd_device
*rbd_dev
)
3042 ret
= rbd_req_sync_watch(rbd_dev
);
3043 if (ret
== -ERANGE
) {
3044 rc
= rbd_dev_refresh(rbd_dev
, NULL
);
3048 } while (ret
== -ERANGE
);
3053 static atomic64_t rbd_dev_id_max
= ATOMIC64_INIT(0);
3056 * Get a unique rbd identifier for the given new rbd_dev, and add
3057 * the rbd_dev to the global list. The minimum rbd id is 1.
3059 static void rbd_dev_id_get(struct rbd_device
*rbd_dev
)
3061 rbd_dev
->dev_id
= atomic64_inc_return(&rbd_dev_id_max
);
3063 spin_lock(&rbd_dev_list_lock
);
3064 list_add_tail(&rbd_dev
->node
, &rbd_dev_list
);
3065 spin_unlock(&rbd_dev_list_lock
);
3066 dout("rbd_dev %p given dev id %llu\n", rbd_dev
,
3067 (unsigned long long) rbd_dev
->dev_id
);
3071 * Remove an rbd_dev from the global list, and record that its
3072 * identifier is no longer in use.
3074 static void rbd_dev_id_put(struct rbd_device
*rbd_dev
)
3076 struct list_head
*tmp
;
3077 int rbd_id
= rbd_dev
->dev_id
;
3080 rbd_assert(rbd_id
> 0);
3082 dout("rbd_dev %p released dev id %llu\n", rbd_dev
,
3083 (unsigned long long) rbd_dev
->dev_id
);
3084 spin_lock(&rbd_dev_list_lock
);
3085 list_del_init(&rbd_dev
->node
);
3088 * If the id being "put" is not the current maximum, there
3089 * is nothing special we need to do.
3091 if (rbd_id
!= atomic64_read(&rbd_dev_id_max
)) {
3092 spin_unlock(&rbd_dev_list_lock
);
3097 * We need to update the current maximum id. Search the
3098 * list to find out what it is. We're more likely to find
3099 * the maximum at the end, so search the list backward.
3102 list_for_each_prev(tmp
, &rbd_dev_list
) {
3103 struct rbd_device
*rbd_dev
;
3105 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
3106 if (rbd_dev
->dev_id
> max_id
)
3107 max_id
= rbd_dev
->dev_id
;
3109 spin_unlock(&rbd_dev_list_lock
);
3112 * The max id could have been updated by rbd_dev_id_get(), in
3113 * which case it now accurately reflects the new maximum.
3114 * Be careful not to overwrite the maximum value in that
3117 atomic64_cmpxchg(&rbd_dev_id_max
, rbd_id
, max_id
);
3118 dout(" max dev id has been reset\n");
3122 * Skips over white space at *buf, and updates *buf to point to the
3123 * first found non-space character (if any). Returns the length of
3124 * the token (string of non-white space characters) found. Note
3125 * that *buf must be terminated with '\0'.
3127 static inline size_t next_token(const char **buf
)
3130 * These are the characters that produce nonzero for
3131 * isspace() in the "C" and "POSIX" locales.
3133 const char *spaces
= " \f\n\r\t\v";
3135 *buf
+= strspn(*buf
, spaces
); /* Find start of token */
3137 return strcspn(*buf
, spaces
); /* Return token length */
3141 * Finds the next token in *buf, and if the provided token buffer is
3142 * big enough, copies the found token into it. The result, if
3143 * copied, is guaranteed to be terminated with '\0'. Note that *buf
3144 * must be terminated with '\0' on entry.
3146 * Returns the length of the token found (not including the '\0').
3147 * Return value will be 0 if no token is found, and it will be >=
3148 * token_size if the token would not fit.
3150 * The *buf pointer will be updated to point beyond the end of the
3151 * found token. Note that this occurs even if the token buffer is
3152 * too small to hold it.
3154 static inline size_t copy_token(const char **buf
,
3160 len
= next_token(buf
);
3161 if (len
< token_size
) {
3162 memcpy(token
, *buf
, len
);
3163 *(token
+ len
) = '\0';
3171 * Finds the next token in *buf, dynamically allocates a buffer big
3172 * enough to hold a copy of it, and copies the token into the new
3173 * buffer. The copy is guaranteed to be terminated with '\0'. Note
3174 * that a duplicate buffer is created even for a zero-length token.
3176 * Returns a pointer to the newly-allocated duplicate, or a null
3177 * pointer if memory for the duplicate was not available. If
3178 * the lenp argument is a non-null pointer, the length of the token
3179 * (not including the '\0') is returned in *lenp.
3181 * If successful, the *buf pointer will be updated to point beyond
3182 * the end of the found token.
3184 * Note: uses GFP_KERNEL for allocation.
3186 static inline char *dup_token(const char **buf
, size_t *lenp
)
3191 len
= next_token(buf
);
3192 dup
= kmemdup(*buf
, len
+ 1, GFP_KERNEL
);
3195 *(dup
+ len
) = '\0';
3205 * Parse the options provided for an "rbd add" (i.e., rbd image
3206 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
3207 * and the data written is passed here via a NUL-terminated buffer.
3208 * Returns 0 if successful or an error code otherwise.
3210 * The information extracted from these options is recorded in
3211 * the other parameters which return dynamically-allocated
3214 * The address of a pointer that will refer to a ceph options
3215 * structure. Caller must release the returned pointer using
3216 * ceph_destroy_options() when it is no longer needed.
3218 * Address of an rbd options pointer. Fully initialized by
3219 * this function; caller must release with kfree().
3221 * Address of an rbd image specification pointer. Fully
3222 * initialized by this function based on parsed options.
3223 * Caller must release with rbd_spec_put().
3225 * The options passed take this form:
3226 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
3229 * A comma-separated list of one or more monitor addresses.
3230 * A monitor address is an ip address, optionally followed
3231 * by a port number (separated by a colon).
3232 * I.e.: ip1[:port1][,ip2[:port2]...]
3234 * A comma-separated list of ceph and/or rbd options.
3236 * The name of the rados pool containing the rbd image.
3238 * The name of the image in that pool to map.
3240 * An optional snapshot id. If provided, the mapping will
3241 * present data from the image at the time that snapshot was
3242 * created. The image head is used if no snapshot id is
3243 * provided. Snapshot mappings are always read-only.
3245 static int rbd_add_parse_args(const char *buf
,
3246 struct ceph_options
**ceph_opts
,
3247 struct rbd_options
**opts
,
3248 struct rbd_spec
**rbd_spec
)
3252 const char *mon_addrs
;
3253 size_t mon_addrs_size
;
3254 struct rbd_spec
*spec
= NULL
;
3255 struct rbd_options
*rbd_opts
= NULL
;
3256 struct ceph_options
*copts
;
3259 /* The first four tokens are required */
3261 len
= next_token(&buf
);
3263 rbd_warn(NULL
, "no monitor address(es) provided");
3267 mon_addrs_size
= len
+ 1;
3271 options
= dup_token(&buf
, NULL
);
3275 rbd_warn(NULL
, "no options provided");
3279 spec
= rbd_spec_alloc();
3283 spec
->pool_name
= dup_token(&buf
, NULL
);
3284 if (!spec
->pool_name
)
3286 if (!*spec
->pool_name
) {
3287 rbd_warn(NULL
, "no pool name provided");
3291 spec
->image_name
= dup_token(&buf
, NULL
);
3292 if (!spec
->image_name
)
3294 if (!*spec
->image_name
) {
3295 rbd_warn(NULL
, "no image name provided");
3300 * Snapshot name is optional; default is to use "-"
3301 * (indicating the head/no snapshot).
3303 len
= next_token(&buf
);
3305 buf
= RBD_SNAP_HEAD_NAME
; /* No snapshot supplied */
3306 len
= sizeof (RBD_SNAP_HEAD_NAME
) - 1;
3307 } else if (len
> RBD_MAX_SNAP_NAME_LEN
) {
3308 ret
= -ENAMETOOLONG
;
3311 spec
->snap_name
= kmemdup(buf
, len
+ 1, GFP_KERNEL
);
3312 if (!spec
->snap_name
)
3314 *(spec
->snap_name
+ len
) = '\0';
3316 /* Initialize all rbd options to the defaults */
3318 rbd_opts
= kzalloc(sizeof (*rbd_opts
), GFP_KERNEL
);
3322 rbd_opts
->read_only
= RBD_READ_ONLY_DEFAULT
;
3324 copts
= ceph_parse_options(options
, mon_addrs
,
3325 mon_addrs
+ mon_addrs_size
- 1,
3326 parse_rbd_opts_token
, rbd_opts
);
3327 if (IS_ERR(copts
)) {
3328 ret
= PTR_ERR(copts
);
3349 * An rbd format 2 image has a unique identifier, distinct from the
3350 * name given to it by the user. Internally, that identifier is
3351 * what's used to specify the names of objects related to the image.
3353 * A special "rbd id" object is used to map an rbd image name to its
3354 * id. If that object doesn't exist, then there is no v2 rbd image
3355 * with the supplied name.
3357 * This function will record the given rbd_dev's image_id field if
3358 * it can be determined, and in that case will return 0. If any
3359 * errors occur a negative errno will be returned and the rbd_dev's
3360 * image_id field will be unchanged (and should be NULL).
3362 static int rbd_dev_image_id(struct rbd_device
*rbd_dev
)
3371 * When probing a parent image, the image id is already
3372 * known (and the image name likely is not). There's no
3373 * need to fetch the image id again in this case.
3375 if (rbd_dev
->spec
->image_id
)
3379 * First, see if the format 2 image id file exists, and if
3380 * so, get the image's persistent id from it.
3382 size
= sizeof (RBD_ID_PREFIX
) + strlen(rbd_dev
->spec
->image_name
);
3383 object_name
= kmalloc(size
, GFP_NOIO
);
3386 sprintf(object_name
, "%s%s", RBD_ID_PREFIX
, rbd_dev
->spec
->image_name
);
3387 dout("rbd id object name is %s\n", object_name
);
3389 /* Response will be an encoded string, which includes a length */
3391 size
= sizeof (__le32
) + RBD_IMAGE_ID_LEN_MAX
;
3392 response
= kzalloc(size
, GFP_NOIO
);
3398 ret
= rbd_req_sync_exec(rbd_dev
, object_name
,
3401 response
, RBD_IMAGE_ID_LEN_MAX
,
3402 CEPH_OSD_FLAG_READ
, NULL
);
3403 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
3406 ret
= 0; /* rbd_req_sync_exec() can return positive */
3409 rbd_dev
->spec
->image_id
= ceph_extract_encoded_string(&p
,
3410 p
+ RBD_IMAGE_ID_LEN_MAX
,
3412 if (IS_ERR(rbd_dev
->spec
->image_id
)) {
3413 ret
= PTR_ERR(rbd_dev
->spec
->image_id
);
3414 rbd_dev
->spec
->image_id
= NULL
;
3416 dout("image_id is %s\n", rbd_dev
->spec
->image_id
);
3425 static int rbd_dev_v1_probe(struct rbd_device
*rbd_dev
)
3430 /* Version 1 images have no id; empty string is used */
3432 rbd_dev
->spec
->image_id
= kstrdup("", GFP_KERNEL
);
3433 if (!rbd_dev
->spec
->image_id
)
3436 /* Record the header object name for this rbd image. */
3438 size
= strlen(rbd_dev
->spec
->image_name
) + sizeof (RBD_SUFFIX
);
3439 rbd_dev
->header_name
= kmalloc(size
, GFP_KERNEL
);
3440 if (!rbd_dev
->header_name
) {
3444 sprintf(rbd_dev
->header_name
, "%s%s",
3445 rbd_dev
->spec
->image_name
, RBD_SUFFIX
);
3447 /* Populate rbd image metadata */
3449 ret
= rbd_read_header(rbd_dev
, &rbd_dev
->header
);
3453 /* Version 1 images have no parent (no layering) */
3455 rbd_dev
->parent_spec
= NULL
;
3456 rbd_dev
->parent_overlap
= 0;
3458 rbd_dev
->image_format
= 1;
3460 dout("discovered version 1 image, header name is %s\n",
3461 rbd_dev
->header_name
);
3466 kfree(rbd_dev
->header_name
);
3467 rbd_dev
->header_name
= NULL
;
3468 kfree(rbd_dev
->spec
->image_id
);
3469 rbd_dev
->spec
->image_id
= NULL
;
3474 static int rbd_dev_v2_probe(struct rbd_device
*rbd_dev
)
3481 * Image id was filled in by the caller. Record the header
3482 * object name for this rbd image.
3484 size
= sizeof (RBD_HEADER_PREFIX
) + strlen(rbd_dev
->spec
->image_id
);
3485 rbd_dev
->header_name
= kmalloc(size
, GFP_KERNEL
);
3486 if (!rbd_dev
->header_name
)
3488 sprintf(rbd_dev
->header_name
, "%s%s",
3489 RBD_HEADER_PREFIX
, rbd_dev
->spec
->image_id
);
3491 /* Get the size and object order for the image */
3493 ret
= rbd_dev_v2_image_size(rbd_dev
);
3497 /* Get the object prefix (a.k.a. block_name) for the image */
3499 ret
= rbd_dev_v2_object_prefix(rbd_dev
);
3503 /* Get the and check features for the image */
3505 ret
= rbd_dev_v2_features(rbd_dev
);
3509 /* If the image supports layering, get the parent info */
3511 if (rbd_dev
->header
.features
& RBD_FEATURE_LAYERING
) {
3512 ret
= rbd_dev_v2_parent_info(rbd_dev
);
3517 /* crypto and compression type aren't (yet) supported for v2 images */
3519 rbd_dev
->header
.crypt_type
= 0;
3520 rbd_dev
->header
.comp_type
= 0;
3522 /* Get the snapshot context, plus the header version */
3524 ret
= rbd_dev_v2_snap_context(rbd_dev
, &ver
);
3527 rbd_dev
->header
.obj_version
= ver
;
3529 rbd_dev
->image_format
= 2;
3531 dout("discovered version 2 image, header name is %s\n",
3532 rbd_dev
->header_name
);
3536 rbd_dev
->parent_overlap
= 0;
3537 rbd_spec_put(rbd_dev
->parent_spec
);
3538 rbd_dev
->parent_spec
= NULL
;
3539 kfree(rbd_dev
->header_name
);
3540 rbd_dev
->header_name
= NULL
;
3541 kfree(rbd_dev
->header
.object_prefix
);
3542 rbd_dev
->header
.object_prefix
= NULL
;
3547 static int rbd_dev_probe_finish(struct rbd_device
*rbd_dev
)
3551 /* no need to lock here, as rbd_dev is not registered yet */
3552 ret
= rbd_dev_snaps_update(rbd_dev
);
3556 ret
= rbd_dev_probe_update_spec(rbd_dev
);
3560 ret
= rbd_dev_set_mapping(rbd_dev
);
3564 /* generate unique id: find highest unique id, add one */
3565 rbd_dev_id_get(rbd_dev
);
3567 /* Fill in the device name, now that we have its id. */
3568 BUILD_BUG_ON(DEV_NAME_LEN
3569 < sizeof (RBD_DRV_NAME
) + MAX_INT_FORMAT_WIDTH
);
3570 sprintf(rbd_dev
->name
, "%s%d", RBD_DRV_NAME
, rbd_dev
->dev_id
);
3572 /* Get our block major device number. */
3574 ret
= register_blkdev(0, rbd_dev
->name
);
3577 rbd_dev
->major
= ret
;
3579 /* Set up the blkdev mapping. */
3581 ret
= rbd_init_disk(rbd_dev
);
3583 goto err_out_blkdev
;
3585 ret
= rbd_bus_add_dev(rbd_dev
);
3590 * At this point cleanup in the event of an error is the job
3591 * of the sysfs code (initiated by rbd_bus_del_dev()).
3593 down_write(&rbd_dev
->header_rwsem
);
3594 ret
= rbd_dev_snaps_register(rbd_dev
);
3595 up_write(&rbd_dev
->header_rwsem
);
3599 ret
= rbd_init_watch_dev(rbd_dev
);
3603 /* Everything's ready. Announce the disk to the world. */
3605 add_disk(rbd_dev
->disk
);
3607 pr_info("%s: added with size 0x%llx\n", rbd_dev
->disk
->disk_name
,
3608 (unsigned long long) rbd_dev
->mapping
.size
);
3612 /* this will also clean up rest of rbd_dev stuff */
3614 rbd_bus_del_dev(rbd_dev
);
3618 rbd_free_disk(rbd_dev
);
3620 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
3622 rbd_dev_id_put(rbd_dev
);
3624 rbd_remove_all_snaps(rbd_dev
);
3630 * Probe for the existence of the header object for the given rbd
3631 * device. For format 2 images this includes determining the image
3634 static int rbd_dev_probe(struct rbd_device
*rbd_dev
)
3639 * Get the id from the image id object. If it's not a
3640 * format 2 image, we'll get ENOENT back, and we'll assume
3641 * it's a format 1 image.
3643 ret
= rbd_dev_image_id(rbd_dev
);
3645 ret
= rbd_dev_v1_probe(rbd_dev
);
3647 ret
= rbd_dev_v2_probe(rbd_dev
);
3649 dout("probe failed, returning %d\n", ret
);
3654 ret
= rbd_dev_probe_finish(rbd_dev
);
3656 rbd_header_free(&rbd_dev
->header
);
3661 static ssize_t
rbd_add(struct bus_type
*bus
,
3665 struct rbd_device
*rbd_dev
= NULL
;
3666 struct ceph_options
*ceph_opts
= NULL
;
3667 struct rbd_options
*rbd_opts
= NULL
;
3668 struct rbd_spec
*spec
= NULL
;
3669 struct rbd_client
*rbdc
;
3670 struct ceph_osd_client
*osdc
;
3673 if (!try_module_get(THIS_MODULE
))
3676 /* parse add command */
3677 rc
= rbd_add_parse_args(buf
, &ceph_opts
, &rbd_opts
, &spec
);
3679 goto err_out_module
;
3681 rbdc
= rbd_get_client(ceph_opts
);
3686 ceph_opts
= NULL
; /* rbd_dev client now owns this */
3689 osdc
= &rbdc
->client
->osdc
;
3690 rc
= ceph_pg_poolid_by_name(osdc
->osdmap
, spec
->pool_name
);
3692 goto err_out_client
;
3693 spec
->pool_id
= (u64
) rc
;
3695 rbd_dev
= rbd_dev_create(rbdc
, spec
);
3697 goto err_out_client
;
3698 rbdc
= NULL
; /* rbd_dev now owns this */
3699 spec
= NULL
; /* rbd_dev now owns this */
3701 rbd_dev
->mapping
.read_only
= rbd_opts
->read_only
;
3703 rbd_opts
= NULL
; /* done with this */
3705 rc
= rbd_dev_probe(rbd_dev
);
3707 goto err_out_rbd_dev
;
3711 rbd_dev_destroy(rbd_dev
);
3713 rbd_put_client(rbdc
);
3716 ceph_destroy_options(ceph_opts
);
3720 module_put(THIS_MODULE
);
3722 dout("Error adding device %s\n", buf
);
3724 return (ssize_t
) rc
;
3727 static struct rbd_device
*__rbd_get_dev(unsigned long dev_id
)
3729 struct list_head
*tmp
;
3730 struct rbd_device
*rbd_dev
;
3732 spin_lock(&rbd_dev_list_lock
);
3733 list_for_each(tmp
, &rbd_dev_list
) {
3734 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
3735 if (rbd_dev
->dev_id
== dev_id
) {
3736 spin_unlock(&rbd_dev_list_lock
);
3740 spin_unlock(&rbd_dev_list_lock
);
3744 static void rbd_dev_release(struct device
*dev
)
3746 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
3748 if (rbd_dev
->watch_request
) {
3749 struct ceph_client
*client
= rbd_dev
->rbd_client
->client
;
3751 ceph_osdc_unregister_linger_request(&client
->osdc
,
3752 rbd_dev
->watch_request
);
3754 if (rbd_dev
->watch_event
)
3755 rbd_req_sync_unwatch(rbd_dev
);
3758 /* clean up and free blkdev */
3759 rbd_free_disk(rbd_dev
);
3760 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
3762 /* release allocated disk header fields */
3763 rbd_header_free(&rbd_dev
->header
);
3765 /* done with the id, and with the rbd_dev */
3766 rbd_dev_id_put(rbd_dev
);
3767 rbd_assert(rbd_dev
->rbd_client
!= NULL
);
3768 rbd_dev_destroy(rbd_dev
);
3770 /* release module ref */
3771 module_put(THIS_MODULE
);
3774 static ssize_t
rbd_remove(struct bus_type
*bus
,
3778 struct rbd_device
*rbd_dev
= NULL
;
3783 rc
= strict_strtoul(buf
, 10, &ul
);
3787 /* convert to int; abort if we lost anything in the conversion */
3788 target_id
= (int) ul
;
3789 if (target_id
!= ul
)
3792 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
3794 rbd_dev
= __rbd_get_dev(target_id
);
3800 if (rbd_dev
->open_count
) {
3805 rbd_remove_all_snaps(rbd_dev
);
3806 rbd_bus_del_dev(rbd_dev
);
3809 mutex_unlock(&ctl_mutex
);
3815 * create control files in sysfs
3818 static int rbd_sysfs_init(void)
3822 ret
= device_register(&rbd_root_dev
);
3826 ret
= bus_register(&rbd_bus_type
);
3828 device_unregister(&rbd_root_dev
);
3833 static void rbd_sysfs_cleanup(void)
3835 bus_unregister(&rbd_bus_type
);
3836 device_unregister(&rbd_root_dev
);
3839 int __init
rbd_init(void)
3843 rc
= rbd_sysfs_init();
3846 pr_info("loaded " RBD_DRV_NAME_LONG
"\n");
3850 void __exit
rbd_exit(void)
3852 rbd_sysfs_cleanup();
3855 module_init(rbd_init
);
3856 module_exit(rbd_exit
);
3858 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
3859 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
3860 MODULE_DESCRIPTION("rados block device");
3862 /* following authorship retained from original osdblk.c */
3863 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
3865 MODULE_LICENSE("GPL");