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 /* Write requests need a reference to the snapshot context */
1667 if (rq_data_dir(rq
) == WRITE
) {
1669 if (read_only
) /* Can't write to a read-only device */
1670 goto out_end_request
;
1673 * Note that each osd request will take its
1674 * own reference to the snapshot context
1675 * supplied. The reference we take here
1676 * just guarantees the one we provide stays
1679 down_read(&rbd_dev
->header_rwsem
);
1680 snapc
= ceph_get_snap_context(rbd_dev
->header
.snapc
);
1681 up_read(&rbd_dev
->header_rwsem
);
1682 rbd_assert(snapc
!= NULL
);
1683 } else if (!atomic_read(&rbd_dev
->exists
)) {
1684 rbd_assert(rbd_dev
->spec
->snap_id
!= CEPH_NOSNAP
);
1685 dout("request for non-existent snapshot");
1687 goto out_end_request
;
1690 size
= blk_rq_bytes(rq
);
1691 result
= rbd_dev_do_request(rq
, rbd_dev
, snapc
,
1692 blk_rq_pos(rq
) * SECTOR_SIZE
,
1696 ceph_put_snap_context(snapc
);
1697 spin_lock_irq(q
->queue_lock
);
1698 if (!size
|| result
< 0)
1699 __blk_end_request_all(rq
, result
);
1704 * a queue callback. Makes sure that we don't create a bio that spans across
1705 * multiple osd objects. One exception would be with a single page bios,
1706 * which we handle later at bio_chain_clone_range()
1708 static int rbd_merge_bvec(struct request_queue
*q
, struct bvec_merge_data
*bmd
,
1709 struct bio_vec
*bvec
)
1711 struct rbd_device
*rbd_dev
= q
->queuedata
;
1712 sector_t sector_offset
;
1713 sector_t sectors_per_obj
;
1714 sector_t obj_sector_offset
;
1718 * Find how far into its rbd object the partition-relative
1719 * bio start sector is to offset relative to the enclosing
1722 sector_offset
= get_start_sect(bmd
->bi_bdev
) + bmd
->bi_sector
;
1723 sectors_per_obj
= 1 << (rbd_dev
->header
.obj_order
- SECTOR_SHIFT
);
1724 obj_sector_offset
= sector_offset
& (sectors_per_obj
- 1);
1727 * Compute the number of bytes from that offset to the end
1728 * of the object. Account for what's already used by the bio.
1730 ret
= (int) (sectors_per_obj
- obj_sector_offset
) << SECTOR_SHIFT
;
1731 if (ret
> bmd
->bi_size
)
1732 ret
-= bmd
->bi_size
;
1737 * Don't send back more than was asked for. And if the bio
1738 * was empty, let the whole thing through because: "Note
1739 * that a block device *must* allow a single page to be
1740 * added to an empty bio."
1742 rbd_assert(bvec
->bv_len
<= PAGE_SIZE
);
1743 if (ret
> (int) bvec
->bv_len
|| !bmd
->bi_size
)
1744 ret
= (int) bvec
->bv_len
;
1749 static void rbd_free_disk(struct rbd_device
*rbd_dev
)
1751 struct gendisk
*disk
= rbd_dev
->disk
;
1756 if (disk
->flags
& GENHD_FL_UP
)
1759 blk_cleanup_queue(disk
->queue
);
1764 * Read the complete header for the given rbd device.
1766 * Returns a pointer to a dynamically-allocated buffer containing
1767 * the complete and validated header. Caller can pass the address
1768 * of a variable that will be filled in with the version of the
1769 * header object at the time it was read.
1771 * Returns a pointer-coded errno if a failure occurs.
1773 static struct rbd_image_header_ondisk
*
1774 rbd_dev_v1_header_read(struct rbd_device
*rbd_dev
, u64
*version
)
1776 struct rbd_image_header_ondisk
*ondisk
= NULL
;
1783 * The complete header will include an array of its 64-bit
1784 * snapshot ids, followed by the names of those snapshots as
1785 * a contiguous block of NUL-terminated strings. Note that
1786 * the number of snapshots could change by the time we read
1787 * it in, in which case we re-read it.
1794 size
= sizeof (*ondisk
);
1795 size
+= snap_count
* sizeof (struct rbd_image_snap_ondisk
);
1797 ondisk
= kmalloc(size
, GFP_KERNEL
);
1799 return ERR_PTR(-ENOMEM
);
1801 ret
= rbd_req_sync_read(rbd_dev
, CEPH_NOSNAP
,
1802 rbd_dev
->header_name
,
1804 (char *) ondisk
, version
);
1808 if (WARN_ON((size_t) ret
< size
)) {
1810 rbd_warn(rbd_dev
, "short header read (want %zd got %d)",
1814 if (!rbd_dev_ondisk_valid(ondisk
)) {
1816 rbd_warn(rbd_dev
, "invalid header");
1820 names_size
= le64_to_cpu(ondisk
->snap_names_len
);
1821 want_count
= snap_count
;
1822 snap_count
= le32_to_cpu(ondisk
->snap_count
);
1823 } while (snap_count
!= want_count
);
1830 return ERR_PTR(ret
);
1834 * reload the ondisk the header
1836 static int rbd_read_header(struct rbd_device
*rbd_dev
,
1837 struct rbd_image_header
*header
)
1839 struct rbd_image_header_ondisk
*ondisk
;
1843 ondisk
= rbd_dev_v1_header_read(rbd_dev
, &ver
);
1845 return PTR_ERR(ondisk
);
1846 ret
= rbd_header_from_disk(header
, ondisk
);
1848 header
->obj_version
= ver
;
1854 static void rbd_remove_all_snaps(struct rbd_device
*rbd_dev
)
1856 struct rbd_snap
*snap
;
1857 struct rbd_snap
*next
;
1859 list_for_each_entry_safe(snap
, next
, &rbd_dev
->snaps
, node
)
1860 rbd_remove_snap_dev(snap
);
1863 static void rbd_update_mapping_size(struct rbd_device
*rbd_dev
)
1867 if (rbd_dev
->spec
->snap_id
!= CEPH_NOSNAP
)
1870 size
= (sector_t
) rbd_dev
->header
.image_size
/ SECTOR_SIZE
;
1871 dout("setting size to %llu sectors", (unsigned long long) size
);
1872 rbd_dev
->mapping
.size
= (u64
) size
;
1873 set_capacity(rbd_dev
->disk
, size
);
1877 * only read the first part of the ondisk header, without the snaps info
1879 static int rbd_dev_v1_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
1882 struct rbd_image_header h
;
1884 ret
= rbd_read_header(rbd_dev
, &h
);
1888 down_write(&rbd_dev
->header_rwsem
);
1890 /* Update image size, and check for resize of mapped image */
1891 rbd_dev
->header
.image_size
= h
.image_size
;
1892 rbd_update_mapping_size(rbd_dev
);
1894 /* rbd_dev->header.object_prefix shouldn't change */
1895 kfree(rbd_dev
->header
.snap_sizes
);
1896 kfree(rbd_dev
->header
.snap_names
);
1897 /* osd requests may still refer to snapc */
1898 ceph_put_snap_context(rbd_dev
->header
.snapc
);
1901 *hver
= h
.obj_version
;
1902 rbd_dev
->header
.obj_version
= h
.obj_version
;
1903 rbd_dev
->header
.image_size
= h
.image_size
;
1904 rbd_dev
->header
.snapc
= h
.snapc
;
1905 rbd_dev
->header
.snap_names
= h
.snap_names
;
1906 rbd_dev
->header
.snap_sizes
= h
.snap_sizes
;
1907 /* Free the extra copy of the object prefix */
1908 WARN_ON(strcmp(rbd_dev
->header
.object_prefix
, h
.object_prefix
));
1909 kfree(h
.object_prefix
);
1911 ret
= rbd_dev_snaps_update(rbd_dev
);
1913 ret
= rbd_dev_snaps_register(rbd_dev
);
1915 up_write(&rbd_dev
->header_rwsem
);
1920 static int rbd_dev_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
1924 rbd_assert(rbd_image_format_valid(rbd_dev
->image_format
));
1925 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1926 if (rbd_dev
->image_format
== 1)
1927 ret
= rbd_dev_v1_refresh(rbd_dev
, hver
);
1929 ret
= rbd_dev_v2_refresh(rbd_dev
, hver
);
1930 mutex_unlock(&ctl_mutex
);
1935 static int rbd_init_disk(struct rbd_device
*rbd_dev
)
1937 struct gendisk
*disk
;
1938 struct request_queue
*q
;
1941 /* create gendisk info */
1942 disk
= alloc_disk(RBD_MINORS_PER_MAJOR
);
1946 snprintf(disk
->disk_name
, sizeof(disk
->disk_name
), RBD_DRV_NAME
"%d",
1948 disk
->major
= rbd_dev
->major
;
1949 disk
->first_minor
= 0;
1950 disk
->fops
= &rbd_bd_ops
;
1951 disk
->private_data
= rbd_dev
;
1954 q
= blk_init_queue(rbd_rq_fn
, &rbd_dev
->lock
);
1958 /* We use the default size, but let's be explicit about it. */
1959 blk_queue_physical_block_size(q
, SECTOR_SIZE
);
1961 /* set io sizes to object size */
1962 segment_size
= rbd_obj_bytes(&rbd_dev
->header
);
1963 blk_queue_max_hw_sectors(q
, segment_size
/ SECTOR_SIZE
);
1964 blk_queue_max_segment_size(q
, segment_size
);
1965 blk_queue_io_min(q
, segment_size
);
1966 blk_queue_io_opt(q
, segment_size
);
1968 blk_queue_merge_bvec(q
, rbd_merge_bvec
);
1971 q
->queuedata
= rbd_dev
;
1973 rbd_dev
->disk
= disk
;
1975 set_capacity(rbd_dev
->disk
, rbd_dev
->mapping
.size
/ SECTOR_SIZE
);
1988 static struct rbd_device
*dev_to_rbd_dev(struct device
*dev
)
1990 return container_of(dev
, struct rbd_device
, dev
);
1993 static ssize_t
rbd_size_show(struct device
*dev
,
1994 struct device_attribute
*attr
, char *buf
)
1996 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1999 down_read(&rbd_dev
->header_rwsem
);
2000 size
= get_capacity(rbd_dev
->disk
);
2001 up_read(&rbd_dev
->header_rwsem
);
2003 return sprintf(buf
, "%llu\n", (unsigned long long) size
* SECTOR_SIZE
);
2007 * Note this shows the features for whatever's mapped, which is not
2008 * necessarily the base image.
2010 static ssize_t
rbd_features_show(struct device
*dev
,
2011 struct device_attribute
*attr
, char *buf
)
2013 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2015 return sprintf(buf
, "0x%016llx\n",
2016 (unsigned long long) rbd_dev
->mapping
.features
);
2019 static ssize_t
rbd_major_show(struct device
*dev
,
2020 struct device_attribute
*attr
, char *buf
)
2022 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2024 return sprintf(buf
, "%d\n", rbd_dev
->major
);
2027 static ssize_t
rbd_client_id_show(struct device
*dev
,
2028 struct device_attribute
*attr
, char *buf
)
2030 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2032 return sprintf(buf
, "client%lld\n",
2033 ceph_client_id(rbd_dev
->rbd_client
->client
));
2036 static ssize_t
rbd_pool_show(struct device
*dev
,
2037 struct device_attribute
*attr
, char *buf
)
2039 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2041 return sprintf(buf
, "%s\n", rbd_dev
->spec
->pool_name
);
2044 static ssize_t
rbd_pool_id_show(struct device
*dev
,
2045 struct device_attribute
*attr
, char *buf
)
2047 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2049 return sprintf(buf
, "%llu\n",
2050 (unsigned long long) rbd_dev
->spec
->pool_id
);
2053 static ssize_t
rbd_name_show(struct device
*dev
,
2054 struct device_attribute
*attr
, char *buf
)
2056 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2058 if (rbd_dev
->spec
->image_name
)
2059 return sprintf(buf
, "%s\n", rbd_dev
->spec
->image_name
);
2061 return sprintf(buf
, "(unknown)\n");
2064 static ssize_t
rbd_image_id_show(struct device
*dev
,
2065 struct device_attribute
*attr
, char *buf
)
2067 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2069 return sprintf(buf
, "%s\n", rbd_dev
->spec
->image_id
);
2073 * Shows the name of the currently-mapped snapshot (or
2074 * RBD_SNAP_HEAD_NAME for the base image).
2076 static ssize_t
rbd_snap_show(struct device
*dev
,
2077 struct device_attribute
*attr
,
2080 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2082 return sprintf(buf
, "%s\n", rbd_dev
->spec
->snap_name
);
2086 * For an rbd v2 image, shows the pool id, image id, and snapshot id
2087 * for the parent image. If there is no parent, simply shows
2088 * "(no parent image)".
2090 static ssize_t
rbd_parent_show(struct device
*dev
,
2091 struct device_attribute
*attr
,
2094 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2095 struct rbd_spec
*spec
= rbd_dev
->parent_spec
;
2100 return sprintf(buf
, "(no parent image)\n");
2102 count
= sprintf(bufp
, "pool_id %llu\npool_name %s\n",
2103 (unsigned long long) spec
->pool_id
, spec
->pool_name
);
2108 count
= sprintf(bufp
, "image_id %s\nimage_name %s\n", spec
->image_id
,
2109 spec
->image_name
? spec
->image_name
: "(unknown)");
2114 count
= sprintf(bufp
, "snap_id %llu\nsnap_name %s\n",
2115 (unsigned long long) spec
->snap_id
, spec
->snap_name
);
2120 count
= sprintf(bufp
, "overlap %llu\n", rbd_dev
->parent_overlap
);
2125 return (ssize_t
) (bufp
- buf
);
2128 static ssize_t
rbd_image_refresh(struct device
*dev
,
2129 struct device_attribute
*attr
,
2133 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2136 ret
= rbd_dev_refresh(rbd_dev
, NULL
);
2138 return ret
< 0 ? ret
: size
;
2141 static DEVICE_ATTR(size
, S_IRUGO
, rbd_size_show
, NULL
);
2142 static DEVICE_ATTR(features
, S_IRUGO
, rbd_features_show
, NULL
);
2143 static DEVICE_ATTR(major
, S_IRUGO
, rbd_major_show
, NULL
);
2144 static DEVICE_ATTR(client_id
, S_IRUGO
, rbd_client_id_show
, NULL
);
2145 static DEVICE_ATTR(pool
, S_IRUGO
, rbd_pool_show
, NULL
);
2146 static DEVICE_ATTR(pool_id
, S_IRUGO
, rbd_pool_id_show
, NULL
);
2147 static DEVICE_ATTR(name
, S_IRUGO
, rbd_name_show
, NULL
);
2148 static DEVICE_ATTR(image_id
, S_IRUGO
, rbd_image_id_show
, NULL
);
2149 static DEVICE_ATTR(refresh
, S_IWUSR
, NULL
, rbd_image_refresh
);
2150 static DEVICE_ATTR(current_snap
, S_IRUGO
, rbd_snap_show
, NULL
);
2151 static DEVICE_ATTR(parent
, S_IRUGO
, rbd_parent_show
, NULL
);
2153 static struct attribute
*rbd_attrs
[] = {
2154 &dev_attr_size
.attr
,
2155 &dev_attr_features
.attr
,
2156 &dev_attr_major
.attr
,
2157 &dev_attr_client_id
.attr
,
2158 &dev_attr_pool
.attr
,
2159 &dev_attr_pool_id
.attr
,
2160 &dev_attr_name
.attr
,
2161 &dev_attr_image_id
.attr
,
2162 &dev_attr_current_snap
.attr
,
2163 &dev_attr_parent
.attr
,
2164 &dev_attr_refresh
.attr
,
2168 static struct attribute_group rbd_attr_group
= {
2172 static const struct attribute_group
*rbd_attr_groups
[] = {
2177 static void rbd_sysfs_dev_release(struct device
*dev
)
2181 static struct device_type rbd_device_type
= {
2183 .groups
= rbd_attr_groups
,
2184 .release
= rbd_sysfs_dev_release
,
2192 static ssize_t
rbd_snap_size_show(struct device
*dev
,
2193 struct device_attribute
*attr
,
2196 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2198 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->size
);
2201 static ssize_t
rbd_snap_id_show(struct device
*dev
,
2202 struct device_attribute
*attr
,
2205 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2207 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->id
);
2210 static ssize_t
rbd_snap_features_show(struct device
*dev
,
2211 struct device_attribute
*attr
,
2214 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2216 return sprintf(buf
, "0x%016llx\n",
2217 (unsigned long long) snap
->features
);
2220 static DEVICE_ATTR(snap_size
, S_IRUGO
, rbd_snap_size_show
, NULL
);
2221 static DEVICE_ATTR(snap_id
, S_IRUGO
, rbd_snap_id_show
, NULL
);
2222 static DEVICE_ATTR(snap_features
, S_IRUGO
, rbd_snap_features_show
, NULL
);
2224 static struct attribute
*rbd_snap_attrs
[] = {
2225 &dev_attr_snap_size
.attr
,
2226 &dev_attr_snap_id
.attr
,
2227 &dev_attr_snap_features
.attr
,
2231 static struct attribute_group rbd_snap_attr_group
= {
2232 .attrs
= rbd_snap_attrs
,
2235 static void rbd_snap_dev_release(struct device
*dev
)
2237 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2242 static const struct attribute_group
*rbd_snap_attr_groups
[] = {
2243 &rbd_snap_attr_group
,
2247 static struct device_type rbd_snap_device_type
= {
2248 .groups
= rbd_snap_attr_groups
,
2249 .release
= rbd_snap_dev_release
,
2252 static struct rbd_spec
*rbd_spec_get(struct rbd_spec
*spec
)
2254 kref_get(&spec
->kref
);
2259 static void rbd_spec_free(struct kref
*kref
);
2260 static void rbd_spec_put(struct rbd_spec
*spec
)
2263 kref_put(&spec
->kref
, rbd_spec_free
);
2266 static struct rbd_spec
*rbd_spec_alloc(void)
2268 struct rbd_spec
*spec
;
2270 spec
= kzalloc(sizeof (*spec
), GFP_KERNEL
);
2273 kref_init(&spec
->kref
);
2275 rbd_spec_put(rbd_spec_get(spec
)); /* TEMPORARY */
2280 static void rbd_spec_free(struct kref
*kref
)
2282 struct rbd_spec
*spec
= container_of(kref
, struct rbd_spec
, kref
);
2284 kfree(spec
->pool_name
);
2285 kfree(spec
->image_id
);
2286 kfree(spec
->image_name
);
2287 kfree(spec
->snap_name
);
2291 struct rbd_device
*rbd_dev_create(struct rbd_client
*rbdc
,
2292 struct rbd_spec
*spec
)
2294 struct rbd_device
*rbd_dev
;
2296 rbd_dev
= kzalloc(sizeof (*rbd_dev
), GFP_KERNEL
);
2300 spin_lock_init(&rbd_dev
->lock
);
2301 atomic_set(&rbd_dev
->exists
, 0);
2302 INIT_LIST_HEAD(&rbd_dev
->node
);
2303 INIT_LIST_HEAD(&rbd_dev
->snaps
);
2304 init_rwsem(&rbd_dev
->header_rwsem
);
2306 rbd_dev
->spec
= spec
;
2307 rbd_dev
->rbd_client
= rbdc
;
2312 static void rbd_dev_destroy(struct rbd_device
*rbd_dev
)
2314 rbd_spec_put(rbd_dev
->parent_spec
);
2315 kfree(rbd_dev
->header_name
);
2316 rbd_put_client(rbd_dev
->rbd_client
);
2317 rbd_spec_put(rbd_dev
->spec
);
2321 static bool rbd_snap_registered(struct rbd_snap
*snap
)
2323 bool ret
= snap
->dev
.type
== &rbd_snap_device_type
;
2324 bool reg
= device_is_registered(&snap
->dev
);
2326 rbd_assert(!ret
^ reg
);
2331 static void rbd_remove_snap_dev(struct rbd_snap
*snap
)
2333 list_del(&snap
->node
);
2334 if (device_is_registered(&snap
->dev
))
2335 device_unregister(&snap
->dev
);
2338 static int rbd_register_snap_dev(struct rbd_snap
*snap
,
2339 struct device
*parent
)
2341 struct device
*dev
= &snap
->dev
;
2344 dev
->type
= &rbd_snap_device_type
;
2345 dev
->parent
= parent
;
2346 dev
->release
= rbd_snap_dev_release
;
2347 dev_set_name(dev
, "%s%s", RBD_SNAP_DEV_NAME_PREFIX
, snap
->name
);
2348 dout("%s: registering device for snapshot %s\n", __func__
, snap
->name
);
2350 ret
= device_register(dev
);
2355 static struct rbd_snap
*__rbd_add_snap_dev(struct rbd_device
*rbd_dev
,
2356 const char *snap_name
,
2357 u64 snap_id
, u64 snap_size
,
2360 struct rbd_snap
*snap
;
2363 snap
= kzalloc(sizeof (*snap
), GFP_KERNEL
);
2365 return ERR_PTR(-ENOMEM
);
2368 snap
->name
= kstrdup(snap_name
, GFP_KERNEL
);
2373 snap
->size
= snap_size
;
2374 snap
->features
= snap_features
;
2382 return ERR_PTR(ret
);
2385 static char *rbd_dev_v1_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2386 u64
*snap_size
, u64
*snap_features
)
2390 rbd_assert(which
< rbd_dev
->header
.snapc
->num_snaps
);
2392 *snap_size
= rbd_dev
->header
.snap_sizes
[which
];
2393 *snap_features
= 0; /* No features for v1 */
2395 /* Skip over names until we find the one we are looking for */
2397 snap_name
= rbd_dev
->header
.snap_names
;
2399 snap_name
+= strlen(snap_name
) + 1;
2405 * Get the size and object order for an image snapshot, or if
2406 * snap_id is CEPH_NOSNAP, gets this information for the base
2409 static int _rbd_dev_v2_snap_size(struct rbd_device
*rbd_dev
, u64 snap_id
,
2410 u8
*order
, u64
*snap_size
)
2412 __le64 snapid
= cpu_to_le64(snap_id
);
2417 } __attribute__ ((packed
)) size_buf
= { 0 };
2419 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2421 (char *) &snapid
, sizeof (snapid
),
2422 (char *) &size_buf
, sizeof (size_buf
),
2423 CEPH_OSD_FLAG_READ
, NULL
);
2424 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2428 *order
= size_buf
.order
;
2429 *snap_size
= le64_to_cpu(size_buf
.size
);
2431 dout(" snap_id 0x%016llx order = %u, snap_size = %llu\n",
2432 (unsigned long long) snap_id
, (unsigned int) *order
,
2433 (unsigned long long) *snap_size
);
2438 static int rbd_dev_v2_image_size(struct rbd_device
*rbd_dev
)
2440 return _rbd_dev_v2_snap_size(rbd_dev
, CEPH_NOSNAP
,
2441 &rbd_dev
->header
.obj_order
,
2442 &rbd_dev
->header
.image_size
);
2445 static int rbd_dev_v2_object_prefix(struct rbd_device
*rbd_dev
)
2451 reply_buf
= kzalloc(RBD_OBJ_PREFIX_LEN_MAX
, GFP_KERNEL
);
2455 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2456 "rbd", "get_object_prefix",
2458 reply_buf
, RBD_OBJ_PREFIX_LEN_MAX
,
2459 CEPH_OSD_FLAG_READ
, NULL
);
2460 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2463 ret
= 0; /* rbd_req_sync_exec() can return positive */
2466 rbd_dev
->header
.object_prefix
= ceph_extract_encoded_string(&p
,
2467 p
+ RBD_OBJ_PREFIX_LEN_MAX
,
2470 if (IS_ERR(rbd_dev
->header
.object_prefix
)) {
2471 ret
= PTR_ERR(rbd_dev
->header
.object_prefix
);
2472 rbd_dev
->header
.object_prefix
= NULL
;
2474 dout(" object_prefix = %s\n", rbd_dev
->header
.object_prefix
);
2483 static int _rbd_dev_v2_snap_features(struct rbd_device
*rbd_dev
, u64 snap_id
,
2486 __le64 snapid
= cpu_to_le64(snap_id
);
2490 } features_buf
= { 0 };
2494 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2495 "rbd", "get_features",
2496 (char *) &snapid
, sizeof (snapid
),
2497 (char *) &features_buf
, sizeof (features_buf
),
2498 CEPH_OSD_FLAG_READ
, NULL
);
2499 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2503 incompat
= le64_to_cpu(features_buf
.incompat
);
2504 if (incompat
& ~RBD_FEATURES_ALL
)
2507 *snap_features
= le64_to_cpu(features_buf
.features
);
2509 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
2510 (unsigned long long) snap_id
,
2511 (unsigned long long) *snap_features
,
2512 (unsigned long long) le64_to_cpu(features_buf
.incompat
));
2517 static int rbd_dev_v2_features(struct rbd_device
*rbd_dev
)
2519 return _rbd_dev_v2_snap_features(rbd_dev
, CEPH_NOSNAP
,
2520 &rbd_dev
->header
.features
);
2523 static int rbd_dev_v2_parent_info(struct rbd_device
*rbd_dev
)
2525 struct rbd_spec
*parent_spec
;
2527 void *reply_buf
= NULL
;
2535 parent_spec
= rbd_spec_alloc();
2539 size
= sizeof (__le64
) + /* pool_id */
2540 sizeof (__le32
) + RBD_IMAGE_ID_LEN_MAX
+ /* image_id */
2541 sizeof (__le64
) + /* snap_id */
2542 sizeof (__le64
); /* overlap */
2543 reply_buf
= kmalloc(size
, GFP_KERNEL
);
2549 snapid
= cpu_to_le64(CEPH_NOSNAP
);
2550 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2551 "rbd", "get_parent",
2552 (char *) &snapid
, sizeof (snapid
),
2553 (char *) reply_buf
, size
,
2554 CEPH_OSD_FLAG_READ
, NULL
);
2555 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2561 end
= (char *) reply_buf
+ size
;
2562 ceph_decode_64_safe(&p
, end
, parent_spec
->pool_id
, out_err
);
2563 if (parent_spec
->pool_id
== CEPH_NOPOOL
)
2564 goto out
; /* No parent? No problem. */
2566 image_id
= ceph_extract_encoded_string(&p
, end
, NULL
, GFP_KERNEL
);
2567 if (IS_ERR(image_id
)) {
2568 ret
= PTR_ERR(image_id
);
2571 parent_spec
->image_id
= image_id
;
2572 ceph_decode_64_safe(&p
, end
, parent_spec
->snap_id
, out_err
);
2573 ceph_decode_64_safe(&p
, end
, overlap
, out_err
);
2575 rbd_dev
->parent_overlap
= overlap
;
2576 rbd_dev
->parent_spec
= parent_spec
;
2577 parent_spec
= NULL
; /* rbd_dev now owns this */
2582 rbd_spec_put(parent_spec
);
2587 static char *rbd_dev_image_name(struct rbd_device
*rbd_dev
)
2589 size_t image_id_size
;
2594 void *reply_buf
= NULL
;
2596 char *image_name
= NULL
;
2599 rbd_assert(!rbd_dev
->spec
->image_name
);
2601 len
= strlen(rbd_dev
->spec
->image_id
);
2602 image_id_size
= sizeof (__le32
) + len
;
2603 image_id
= kmalloc(image_id_size
, GFP_KERNEL
);
2608 end
= (char *) image_id
+ image_id_size
;
2609 ceph_encode_string(&p
, end
, rbd_dev
->spec
->image_id
, (u32
) len
);
2611 size
= sizeof (__le32
) + RBD_IMAGE_NAME_LEN_MAX
;
2612 reply_buf
= kmalloc(size
, GFP_KERNEL
);
2616 ret
= rbd_req_sync_exec(rbd_dev
, RBD_DIRECTORY
,
2617 "rbd", "dir_get_name",
2618 image_id
, image_id_size
,
2619 (char *) reply_buf
, size
,
2620 CEPH_OSD_FLAG_READ
, NULL
);
2624 end
= (char *) reply_buf
+ size
;
2625 image_name
= ceph_extract_encoded_string(&p
, end
, &len
, GFP_KERNEL
);
2626 if (IS_ERR(image_name
))
2629 dout("%s: name is %s len is %zd\n", __func__
, image_name
, len
);
2638 * When a parent image gets probed, we only have the pool, image,
2639 * and snapshot ids but not the names of any of them. This call
2640 * is made later to fill in those names. It has to be done after
2641 * rbd_dev_snaps_update() has completed because some of the
2642 * information (in particular, snapshot name) is not available
2645 static int rbd_dev_probe_update_spec(struct rbd_device
*rbd_dev
)
2647 struct ceph_osd_client
*osdc
;
2649 void *reply_buf
= NULL
;
2652 if (rbd_dev
->spec
->pool_name
)
2653 return 0; /* Already have the names */
2655 /* Look up the pool name */
2657 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
2658 name
= ceph_pg_pool_name_by_id(osdc
->osdmap
, rbd_dev
->spec
->pool_id
);
2660 rbd_warn(rbd_dev
, "there is no pool with id %llu",
2661 rbd_dev
->spec
->pool_id
); /* Really a BUG() */
2665 rbd_dev
->spec
->pool_name
= kstrdup(name
, GFP_KERNEL
);
2666 if (!rbd_dev
->spec
->pool_name
)
2669 /* Fetch the image name; tolerate failure here */
2671 name
= rbd_dev_image_name(rbd_dev
);
2673 rbd_dev
->spec
->image_name
= (char *) name
;
2675 rbd_warn(rbd_dev
, "unable to get image name");
2677 /* Look up the snapshot name. */
2679 name
= rbd_snap_name(rbd_dev
, rbd_dev
->spec
->snap_id
);
2681 rbd_warn(rbd_dev
, "no snapshot with id %llu",
2682 rbd_dev
->spec
->snap_id
); /* Really a BUG() */
2686 rbd_dev
->spec
->snap_name
= kstrdup(name
, GFP_KERNEL
);
2687 if(!rbd_dev
->spec
->snap_name
)
2693 kfree(rbd_dev
->spec
->pool_name
);
2694 rbd_dev
->spec
->pool_name
= NULL
;
2699 static int rbd_dev_v2_snap_context(struct rbd_device
*rbd_dev
, u64
*ver
)
2708 struct ceph_snap_context
*snapc
;
2712 * We'll need room for the seq value (maximum snapshot id),
2713 * snapshot count, and array of that many snapshot ids.
2714 * For now we have a fixed upper limit on the number we're
2715 * prepared to receive.
2717 size
= sizeof (__le64
) + sizeof (__le32
) +
2718 RBD_MAX_SNAP_COUNT
* sizeof (__le64
);
2719 reply_buf
= kzalloc(size
, GFP_KERNEL
);
2723 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2724 "rbd", "get_snapcontext",
2727 CEPH_OSD_FLAG_READ
, ver
);
2728 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2734 end
= (char *) reply_buf
+ size
;
2735 ceph_decode_64_safe(&p
, end
, seq
, out
);
2736 ceph_decode_32_safe(&p
, end
, snap_count
, out
);
2739 * Make sure the reported number of snapshot ids wouldn't go
2740 * beyond the end of our buffer. But before checking that,
2741 * make sure the computed size of the snapshot context we
2742 * allocate is representable in a size_t.
2744 if (snap_count
> (SIZE_MAX
- sizeof (struct ceph_snap_context
))
2749 if (!ceph_has_room(&p
, end
, snap_count
* sizeof (__le64
)))
2752 size
= sizeof (struct ceph_snap_context
) +
2753 snap_count
* sizeof (snapc
->snaps
[0]);
2754 snapc
= kmalloc(size
, GFP_KERNEL
);
2760 atomic_set(&snapc
->nref
, 1);
2762 snapc
->num_snaps
= snap_count
;
2763 for (i
= 0; i
< snap_count
; i
++)
2764 snapc
->snaps
[i
] = ceph_decode_64(&p
);
2766 rbd_dev
->header
.snapc
= snapc
;
2768 dout(" snap context seq = %llu, snap_count = %u\n",
2769 (unsigned long long) seq
, (unsigned int) snap_count
);
2777 static char *rbd_dev_v2_snap_name(struct rbd_device
*rbd_dev
, u32 which
)
2787 size
= sizeof (__le32
) + RBD_MAX_SNAP_NAME_LEN
;
2788 reply_buf
= kmalloc(size
, GFP_KERNEL
);
2790 return ERR_PTR(-ENOMEM
);
2792 snap_id
= cpu_to_le64(rbd_dev
->header
.snapc
->snaps
[which
]);
2793 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
2794 "rbd", "get_snapshot_name",
2795 (char *) &snap_id
, sizeof (snap_id
),
2797 CEPH_OSD_FLAG_READ
, NULL
);
2798 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
2803 end
= (char *) reply_buf
+ size
;
2804 snap_name
= ceph_extract_encoded_string(&p
, end
, NULL
, GFP_KERNEL
);
2805 if (IS_ERR(snap_name
)) {
2806 ret
= PTR_ERR(snap_name
);
2809 dout(" snap_id 0x%016llx snap_name = %s\n",
2810 (unsigned long long) le64_to_cpu(snap_id
), snap_name
);
2818 return ERR_PTR(ret
);
2821 static char *rbd_dev_v2_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2822 u64
*snap_size
, u64
*snap_features
)
2828 snap_id
= rbd_dev
->header
.snapc
->snaps
[which
];
2829 ret
= _rbd_dev_v2_snap_size(rbd_dev
, snap_id
, &order
, snap_size
);
2831 return ERR_PTR(ret
);
2832 ret
= _rbd_dev_v2_snap_features(rbd_dev
, snap_id
, snap_features
);
2834 return ERR_PTR(ret
);
2836 return rbd_dev_v2_snap_name(rbd_dev
, which
);
2839 static char *rbd_dev_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2840 u64
*snap_size
, u64
*snap_features
)
2842 if (rbd_dev
->image_format
== 1)
2843 return rbd_dev_v1_snap_info(rbd_dev
, which
,
2844 snap_size
, snap_features
);
2845 if (rbd_dev
->image_format
== 2)
2846 return rbd_dev_v2_snap_info(rbd_dev
, which
,
2847 snap_size
, snap_features
);
2848 return ERR_PTR(-EINVAL
);
2851 static int rbd_dev_v2_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
2856 down_write(&rbd_dev
->header_rwsem
);
2858 /* Grab old order first, to see if it changes */
2860 obj_order
= rbd_dev
->header
.obj_order
,
2861 ret
= rbd_dev_v2_image_size(rbd_dev
);
2864 if (rbd_dev
->header
.obj_order
!= obj_order
) {
2868 rbd_update_mapping_size(rbd_dev
);
2870 ret
= rbd_dev_v2_snap_context(rbd_dev
, hver
);
2871 dout("rbd_dev_v2_snap_context returned %d\n", ret
);
2874 ret
= rbd_dev_snaps_update(rbd_dev
);
2875 dout("rbd_dev_snaps_update returned %d\n", ret
);
2878 ret
= rbd_dev_snaps_register(rbd_dev
);
2879 dout("rbd_dev_snaps_register returned %d\n", ret
);
2881 up_write(&rbd_dev
->header_rwsem
);
2887 * Scan the rbd device's current snapshot list and compare it to the
2888 * newly-received snapshot context. Remove any existing snapshots
2889 * not present in the new snapshot context. Add a new snapshot for
2890 * any snaphots in the snapshot context not in the current list.
2891 * And verify there are no changes to snapshots we already know
2894 * Assumes the snapshots in the snapshot context are sorted by
2895 * snapshot id, highest id first. (Snapshots in the rbd_dev's list
2896 * are also maintained in that order.)
2898 static int rbd_dev_snaps_update(struct rbd_device
*rbd_dev
)
2900 struct ceph_snap_context
*snapc
= rbd_dev
->header
.snapc
;
2901 const u32 snap_count
= snapc
->num_snaps
;
2902 struct list_head
*head
= &rbd_dev
->snaps
;
2903 struct list_head
*links
= head
->next
;
2906 dout("%s: snap count is %u\n", __func__
, (unsigned int) snap_count
);
2907 while (index
< snap_count
|| links
!= head
) {
2909 struct rbd_snap
*snap
;
2912 u64 snap_features
= 0;
2914 snap_id
= index
< snap_count
? snapc
->snaps
[index
]
2916 snap
= links
!= head
? list_entry(links
, struct rbd_snap
, node
)
2918 rbd_assert(!snap
|| snap
->id
!= CEPH_NOSNAP
);
2920 if (snap_id
== CEPH_NOSNAP
|| (snap
&& snap
->id
> snap_id
)) {
2921 struct list_head
*next
= links
->next
;
2923 /* Existing snapshot not in the new snap context */
2925 if (rbd_dev
->spec
->snap_id
== snap
->id
)
2926 atomic_set(&rbd_dev
->exists
, 0);
2927 rbd_remove_snap_dev(snap
);
2928 dout("%ssnap id %llu has been removed\n",
2929 rbd_dev
->spec
->snap_id
== snap
->id
?
2931 (unsigned long long) snap
->id
);
2933 /* Done with this list entry; advance */
2939 snap_name
= rbd_dev_snap_info(rbd_dev
, index
,
2940 &snap_size
, &snap_features
);
2941 if (IS_ERR(snap_name
))
2942 return PTR_ERR(snap_name
);
2944 dout("entry %u: snap_id = %llu\n", (unsigned int) snap_count
,
2945 (unsigned long long) snap_id
);
2946 if (!snap
|| (snap_id
!= CEPH_NOSNAP
&& snap
->id
< snap_id
)) {
2947 struct rbd_snap
*new_snap
;
2949 /* We haven't seen this snapshot before */
2951 new_snap
= __rbd_add_snap_dev(rbd_dev
, snap_name
,
2952 snap_id
, snap_size
, snap_features
);
2953 if (IS_ERR(new_snap
)) {
2954 int err
= PTR_ERR(new_snap
);
2956 dout(" failed to add dev, error %d\n", err
);
2961 /* New goes before existing, or at end of list */
2963 dout(" added dev%s\n", snap
? "" : " at end\n");
2965 list_add_tail(&new_snap
->node
, &snap
->node
);
2967 list_add_tail(&new_snap
->node
, head
);
2969 /* Already have this one */
2971 dout(" already present\n");
2973 rbd_assert(snap
->size
== snap_size
);
2974 rbd_assert(!strcmp(snap
->name
, snap_name
));
2975 rbd_assert(snap
->features
== snap_features
);
2977 /* Done with this list entry; advance */
2979 links
= links
->next
;
2982 /* Advance to the next entry in the snapshot context */
2986 dout("%s: done\n", __func__
);
2992 * Scan the list of snapshots and register the devices for any that
2993 * have not already been registered.
2995 static int rbd_dev_snaps_register(struct rbd_device
*rbd_dev
)
2997 struct rbd_snap
*snap
;
3000 dout("%s called\n", __func__
);
3001 if (WARN_ON(!device_is_registered(&rbd_dev
->dev
)))
3004 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
3005 if (!rbd_snap_registered(snap
)) {
3006 ret
= rbd_register_snap_dev(snap
, &rbd_dev
->dev
);
3011 dout("%s: returning %d\n", __func__
, ret
);
3016 static int rbd_bus_add_dev(struct rbd_device
*rbd_dev
)
3021 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
3023 dev
= &rbd_dev
->dev
;
3024 dev
->bus
= &rbd_bus_type
;
3025 dev
->type
= &rbd_device_type
;
3026 dev
->parent
= &rbd_root_dev
;
3027 dev
->release
= rbd_dev_release
;
3028 dev_set_name(dev
, "%d", rbd_dev
->dev_id
);
3029 ret
= device_register(dev
);
3031 mutex_unlock(&ctl_mutex
);
3036 static void rbd_bus_del_dev(struct rbd_device
*rbd_dev
)
3038 device_unregister(&rbd_dev
->dev
);
3041 static int rbd_init_watch_dev(struct rbd_device
*rbd_dev
)
3046 ret
= rbd_req_sync_watch(rbd_dev
);
3047 if (ret
== -ERANGE
) {
3048 rc
= rbd_dev_refresh(rbd_dev
, NULL
);
3052 } while (ret
== -ERANGE
);
3057 static atomic64_t rbd_dev_id_max
= ATOMIC64_INIT(0);
3060 * Get a unique rbd identifier for the given new rbd_dev, and add
3061 * the rbd_dev to the global list. The minimum rbd id is 1.
3063 static void rbd_dev_id_get(struct rbd_device
*rbd_dev
)
3065 rbd_dev
->dev_id
= atomic64_inc_return(&rbd_dev_id_max
);
3067 spin_lock(&rbd_dev_list_lock
);
3068 list_add_tail(&rbd_dev
->node
, &rbd_dev_list
);
3069 spin_unlock(&rbd_dev_list_lock
);
3070 dout("rbd_dev %p given dev id %llu\n", rbd_dev
,
3071 (unsigned long long) rbd_dev
->dev_id
);
3075 * Remove an rbd_dev from the global list, and record that its
3076 * identifier is no longer in use.
3078 static void rbd_dev_id_put(struct rbd_device
*rbd_dev
)
3080 struct list_head
*tmp
;
3081 int rbd_id
= rbd_dev
->dev_id
;
3084 rbd_assert(rbd_id
> 0);
3086 dout("rbd_dev %p released dev id %llu\n", rbd_dev
,
3087 (unsigned long long) rbd_dev
->dev_id
);
3088 spin_lock(&rbd_dev_list_lock
);
3089 list_del_init(&rbd_dev
->node
);
3092 * If the id being "put" is not the current maximum, there
3093 * is nothing special we need to do.
3095 if (rbd_id
!= atomic64_read(&rbd_dev_id_max
)) {
3096 spin_unlock(&rbd_dev_list_lock
);
3101 * We need to update the current maximum id. Search the
3102 * list to find out what it is. We're more likely to find
3103 * the maximum at the end, so search the list backward.
3106 list_for_each_prev(tmp
, &rbd_dev_list
) {
3107 struct rbd_device
*rbd_dev
;
3109 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
3110 if (rbd_dev
->dev_id
> max_id
)
3111 max_id
= rbd_dev
->dev_id
;
3113 spin_unlock(&rbd_dev_list_lock
);
3116 * The max id could have been updated by rbd_dev_id_get(), in
3117 * which case it now accurately reflects the new maximum.
3118 * Be careful not to overwrite the maximum value in that
3121 atomic64_cmpxchg(&rbd_dev_id_max
, rbd_id
, max_id
);
3122 dout(" max dev id has been reset\n");
3126 * Skips over white space at *buf, and updates *buf to point to the
3127 * first found non-space character (if any). Returns the length of
3128 * the token (string of non-white space characters) found. Note
3129 * that *buf must be terminated with '\0'.
3131 static inline size_t next_token(const char **buf
)
3134 * These are the characters that produce nonzero for
3135 * isspace() in the "C" and "POSIX" locales.
3137 const char *spaces
= " \f\n\r\t\v";
3139 *buf
+= strspn(*buf
, spaces
); /* Find start of token */
3141 return strcspn(*buf
, spaces
); /* Return token length */
3145 * Finds the next token in *buf, and if the provided token buffer is
3146 * big enough, copies the found token into it. The result, if
3147 * copied, is guaranteed to be terminated with '\0'. Note that *buf
3148 * must be terminated with '\0' on entry.
3150 * Returns the length of the token found (not including the '\0').
3151 * Return value will be 0 if no token is found, and it will be >=
3152 * token_size if the token would not fit.
3154 * The *buf pointer will be updated to point beyond the end of the
3155 * found token. Note that this occurs even if the token buffer is
3156 * too small to hold it.
3158 static inline size_t copy_token(const char **buf
,
3164 len
= next_token(buf
);
3165 if (len
< token_size
) {
3166 memcpy(token
, *buf
, len
);
3167 *(token
+ len
) = '\0';
3175 * Finds the next token in *buf, dynamically allocates a buffer big
3176 * enough to hold a copy of it, and copies the token into the new
3177 * buffer. The copy is guaranteed to be terminated with '\0'. Note
3178 * that a duplicate buffer is created even for a zero-length token.
3180 * Returns a pointer to the newly-allocated duplicate, or a null
3181 * pointer if memory for the duplicate was not available. If
3182 * the lenp argument is a non-null pointer, the length of the token
3183 * (not including the '\0') is returned in *lenp.
3185 * If successful, the *buf pointer will be updated to point beyond
3186 * the end of the found token.
3188 * Note: uses GFP_KERNEL for allocation.
3190 static inline char *dup_token(const char **buf
, size_t *lenp
)
3195 len
= next_token(buf
);
3196 dup
= kmemdup(*buf
, len
+ 1, GFP_KERNEL
);
3199 *(dup
+ len
) = '\0';
3209 * Parse the options provided for an "rbd add" (i.e., rbd image
3210 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
3211 * and the data written is passed here via a NUL-terminated buffer.
3212 * Returns 0 if successful or an error code otherwise.
3214 * The information extracted from these options is recorded in
3215 * the other parameters which return dynamically-allocated
3218 * The address of a pointer that will refer to a ceph options
3219 * structure. Caller must release the returned pointer using
3220 * ceph_destroy_options() when it is no longer needed.
3222 * Address of an rbd options pointer. Fully initialized by
3223 * this function; caller must release with kfree().
3225 * Address of an rbd image specification pointer. Fully
3226 * initialized by this function based on parsed options.
3227 * Caller must release with rbd_spec_put().
3229 * The options passed take this form:
3230 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
3233 * A comma-separated list of one or more monitor addresses.
3234 * A monitor address is an ip address, optionally followed
3235 * by a port number (separated by a colon).
3236 * I.e.: ip1[:port1][,ip2[:port2]...]
3238 * A comma-separated list of ceph and/or rbd options.
3240 * The name of the rados pool containing the rbd image.
3242 * The name of the image in that pool to map.
3244 * An optional snapshot id. If provided, the mapping will
3245 * present data from the image at the time that snapshot was
3246 * created. The image head is used if no snapshot id is
3247 * provided. Snapshot mappings are always read-only.
3249 static int rbd_add_parse_args(const char *buf
,
3250 struct ceph_options
**ceph_opts
,
3251 struct rbd_options
**opts
,
3252 struct rbd_spec
**rbd_spec
)
3256 const char *mon_addrs
;
3257 size_t mon_addrs_size
;
3258 struct rbd_spec
*spec
= NULL
;
3259 struct rbd_options
*rbd_opts
= NULL
;
3260 struct ceph_options
*copts
;
3263 /* The first four tokens are required */
3265 len
= next_token(&buf
);
3267 rbd_warn(NULL
, "no monitor address(es) provided");
3271 mon_addrs_size
= len
+ 1;
3275 options
= dup_token(&buf
, NULL
);
3279 rbd_warn(NULL
, "no options provided");
3283 spec
= rbd_spec_alloc();
3287 spec
->pool_name
= dup_token(&buf
, NULL
);
3288 if (!spec
->pool_name
)
3290 if (!*spec
->pool_name
) {
3291 rbd_warn(NULL
, "no pool name provided");
3295 spec
->image_name
= dup_token(&buf
, NULL
);
3296 if (!spec
->image_name
)
3298 if (!*spec
->image_name
) {
3299 rbd_warn(NULL
, "no image name provided");
3304 * Snapshot name is optional; default is to use "-"
3305 * (indicating the head/no snapshot).
3307 len
= next_token(&buf
);
3309 buf
= RBD_SNAP_HEAD_NAME
; /* No snapshot supplied */
3310 len
= sizeof (RBD_SNAP_HEAD_NAME
) - 1;
3311 } else if (len
> RBD_MAX_SNAP_NAME_LEN
) {
3312 ret
= -ENAMETOOLONG
;
3315 spec
->snap_name
= kmemdup(buf
, len
+ 1, GFP_KERNEL
);
3316 if (!spec
->snap_name
)
3318 *(spec
->snap_name
+ len
) = '\0';
3320 /* Initialize all rbd options to the defaults */
3322 rbd_opts
= kzalloc(sizeof (*rbd_opts
), GFP_KERNEL
);
3326 rbd_opts
->read_only
= RBD_READ_ONLY_DEFAULT
;
3328 copts
= ceph_parse_options(options
, mon_addrs
,
3329 mon_addrs
+ mon_addrs_size
- 1,
3330 parse_rbd_opts_token
, rbd_opts
);
3331 if (IS_ERR(copts
)) {
3332 ret
= PTR_ERR(copts
);
3353 * An rbd format 2 image has a unique identifier, distinct from the
3354 * name given to it by the user. Internally, that identifier is
3355 * what's used to specify the names of objects related to the image.
3357 * A special "rbd id" object is used to map an rbd image name to its
3358 * id. If that object doesn't exist, then there is no v2 rbd image
3359 * with the supplied name.
3361 * This function will record the given rbd_dev's image_id field if
3362 * it can be determined, and in that case will return 0. If any
3363 * errors occur a negative errno will be returned and the rbd_dev's
3364 * image_id field will be unchanged (and should be NULL).
3366 static int rbd_dev_image_id(struct rbd_device
*rbd_dev
)
3375 * When probing a parent image, the image id is already
3376 * known (and the image name likely is not). There's no
3377 * need to fetch the image id again in this case.
3379 if (rbd_dev
->spec
->image_id
)
3383 * First, see if the format 2 image id file exists, and if
3384 * so, get the image's persistent id from it.
3386 size
= sizeof (RBD_ID_PREFIX
) + strlen(rbd_dev
->spec
->image_name
);
3387 object_name
= kmalloc(size
, GFP_NOIO
);
3390 sprintf(object_name
, "%s%s", RBD_ID_PREFIX
, rbd_dev
->spec
->image_name
);
3391 dout("rbd id object name is %s\n", object_name
);
3393 /* Response will be an encoded string, which includes a length */
3395 size
= sizeof (__le32
) + RBD_IMAGE_ID_LEN_MAX
;
3396 response
= kzalloc(size
, GFP_NOIO
);
3402 ret
= rbd_req_sync_exec(rbd_dev
, object_name
,
3405 response
, RBD_IMAGE_ID_LEN_MAX
,
3406 CEPH_OSD_FLAG_READ
, NULL
);
3407 dout("%s: rbd_req_sync_exec returned %d\n", __func__
, ret
);
3410 ret
= 0; /* rbd_req_sync_exec() can return positive */
3413 rbd_dev
->spec
->image_id
= ceph_extract_encoded_string(&p
,
3414 p
+ RBD_IMAGE_ID_LEN_MAX
,
3416 if (IS_ERR(rbd_dev
->spec
->image_id
)) {
3417 ret
= PTR_ERR(rbd_dev
->spec
->image_id
);
3418 rbd_dev
->spec
->image_id
= NULL
;
3420 dout("image_id is %s\n", rbd_dev
->spec
->image_id
);
3429 static int rbd_dev_v1_probe(struct rbd_device
*rbd_dev
)
3434 /* Version 1 images have no id; empty string is used */
3436 rbd_dev
->spec
->image_id
= kstrdup("", GFP_KERNEL
);
3437 if (!rbd_dev
->spec
->image_id
)
3440 /* Record the header object name for this rbd image. */
3442 size
= strlen(rbd_dev
->spec
->image_name
) + sizeof (RBD_SUFFIX
);
3443 rbd_dev
->header_name
= kmalloc(size
, GFP_KERNEL
);
3444 if (!rbd_dev
->header_name
) {
3448 sprintf(rbd_dev
->header_name
, "%s%s",
3449 rbd_dev
->spec
->image_name
, RBD_SUFFIX
);
3451 /* Populate rbd image metadata */
3453 ret
= rbd_read_header(rbd_dev
, &rbd_dev
->header
);
3457 /* Version 1 images have no parent (no layering) */
3459 rbd_dev
->parent_spec
= NULL
;
3460 rbd_dev
->parent_overlap
= 0;
3462 rbd_dev
->image_format
= 1;
3464 dout("discovered version 1 image, header name is %s\n",
3465 rbd_dev
->header_name
);
3470 kfree(rbd_dev
->header_name
);
3471 rbd_dev
->header_name
= NULL
;
3472 kfree(rbd_dev
->spec
->image_id
);
3473 rbd_dev
->spec
->image_id
= NULL
;
3478 static int rbd_dev_v2_probe(struct rbd_device
*rbd_dev
)
3485 * Image id was filled in by the caller. Record the header
3486 * object name for this rbd image.
3488 size
= sizeof (RBD_HEADER_PREFIX
) + strlen(rbd_dev
->spec
->image_id
);
3489 rbd_dev
->header_name
= kmalloc(size
, GFP_KERNEL
);
3490 if (!rbd_dev
->header_name
)
3492 sprintf(rbd_dev
->header_name
, "%s%s",
3493 RBD_HEADER_PREFIX
, rbd_dev
->spec
->image_id
);
3495 /* Get the size and object order for the image */
3497 ret
= rbd_dev_v2_image_size(rbd_dev
);
3501 /* Get the object prefix (a.k.a. block_name) for the image */
3503 ret
= rbd_dev_v2_object_prefix(rbd_dev
);
3507 /* Get the and check features for the image */
3509 ret
= rbd_dev_v2_features(rbd_dev
);
3513 /* If the image supports layering, get the parent info */
3515 if (rbd_dev
->header
.features
& RBD_FEATURE_LAYERING
) {
3516 ret
= rbd_dev_v2_parent_info(rbd_dev
);
3521 /* crypto and compression type aren't (yet) supported for v2 images */
3523 rbd_dev
->header
.crypt_type
= 0;
3524 rbd_dev
->header
.comp_type
= 0;
3526 /* Get the snapshot context, plus the header version */
3528 ret
= rbd_dev_v2_snap_context(rbd_dev
, &ver
);
3531 rbd_dev
->header
.obj_version
= ver
;
3533 rbd_dev
->image_format
= 2;
3535 dout("discovered version 2 image, header name is %s\n",
3536 rbd_dev
->header_name
);
3540 rbd_dev
->parent_overlap
= 0;
3541 rbd_spec_put(rbd_dev
->parent_spec
);
3542 rbd_dev
->parent_spec
= NULL
;
3543 kfree(rbd_dev
->header_name
);
3544 rbd_dev
->header_name
= NULL
;
3545 kfree(rbd_dev
->header
.object_prefix
);
3546 rbd_dev
->header
.object_prefix
= NULL
;
3551 static int rbd_dev_probe_finish(struct rbd_device
*rbd_dev
)
3555 /* no need to lock here, as rbd_dev is not registered yet */
3556 ret
= rbd_dev_snaps_update(rbd_dev
);
3560 ret
= rbd_dev_probe_update_spec(rbd_dev
);
3564 ret
= rbd_dev_set_mapping(rbd_dev
);
3568 /* generate unique id: find highest unique id, add one */
3569 rbd_dev_id_get(rbd_dev
);
3571 /* Fill in the device name, now that we have its id. */
3572 BUILD_BUG_ON(DEV_NAME_LEN
3573 < sizeof (RBD_DRV_NAME
) + MAX_INT_FORMAT_WIDTH
);
3574 sprintf(rbd_dev
->name
, "%s%d", RBD_DRV_NAME
, rbd_dev
->dev_id
);
3576 /* Get our block major device number. */
3578 ret
= register_blkdev(0, rbd_dev
->name
);
3581 rbd_dev
->major
= ret
;
3583 /* Set up the blkdev mapping. */
3585 ret
= rbd_init_disk(rbd_dev
);
3587 goto err_out_blkdev
;
3589 ret
= rbd_bus_add_dev(rbd_dev
);
3594 * At this point cleanup in the event of an error is the job
3595 * of the sysfs code (initiated by rbd_bus_del_dev()).
3597 down_write(&rbd_dev
->header_rwsem
);
3598 ret
= rbd_dev_snaps_register(rbd_dev
);
3599 up_write(&rbd_dev
->header_rwsem
);
3603 ret
= rbd_init_watch_dev(rbd_dev
);
3607 /* Everything's ready. Announce the disk to the world. */
3609 add_disk(rbd_dev
->disk
);
3611 pr_info("%s: added with size 0x%llx\n", rbd_dev
->disk
->disk_name
,
3612 (unsigned long long) rbd_dev
->mapping
.size
);
3616 /* this will also clean up rest of rbd_dev stuff */
3618 rbd_bus_del_dev(rbd_dev
);
3622 rbd_free_disk(rbd_dev
);
3624 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
3626 rbd_dev_id_put(rbd_dev
);
3628 rbd_remove_all_snaps(rbd_dev
);
3634 * Probe for the existence of the header object for the given rbd
3635 * device. For format 2 images this includes determining the image
3638 static int rbd_dev_probe(struct rbd_device
*rbd_dev
)
3643 * Get the id from the image id object. If it's not a
3644 * format 2 image, we'll get ENOENT back, and we'll assume
3645 * it's a format 1 image.
3647 ret
= rbd_dev_image_id(rbd_dev
);
3649 ret
= rbd_dev_v1_probe(rbd_dev
);
3651 ret
= rbd_dev_v2_probe(rbd_dev
);
3653 dout("probe failed, returning %d\n", ret
);
3658 ret
= rbd_dev_probe_finish(rbd_dev
);
3660 rbd_header_free(&rbd_dev
->header
);
3665 static ssize_t
rbd_add(struct bus_type
*bus
,
3669 struct rbd_device
*rbd_dev
= NULL
;
3670 struct ceph_options
*ceph_opts
= NULL
;
3671 struct rbd_options
*rbd_opts
= NULL
;
3672 struct rbd_spec
*spec
= NULL
;
3673 struct rbd_client
*rbdc
;
3674 struct ceph_osd_client
*osdc
;
3677 if (!try_module_get(THIS_MODULE
))
3680 /* parse add command */
3681 rc
= rbd_add_parse_args(buf
, &ceph_opts
, &rbd_opts
, &spec
);
3683 goto err_out_module
;
3685 rbdc
= rbd_get_client(ceph_opts
);
3690 ceph_opts
= NULL
; /* rbd_dev client now owns this */
3693 osdc
= &rbdc
->client
->osdc
;
3694 rc
= ceph_pg_poolid_by_name(osdc
->osdmap
, spec
->pool_name
);
3696 goto err_out_client
;
3697 spec
->pool_id
= (u64
) rc
;
3699 rbd_dev
= rbd_dev_create(rbdc
, spec
);
3701 goto err_out_client
;
3702 rbdc
= NULL
; /* rbd_dev now owns this */
3703 spec
= NULL
; /* rbd_dev now owns this */
3705 rbd_dev
->mapping
.read_only
= rbd_opts
->read_only
;
3707 rbd_opts
= NULL
; /* done with this */
3709 rc
= rbd_dev_probe(rbd_dev
);
3711 goto err_out_rbd_dev
;
3715 rbd_dev_destroy(rbd_dev
);
3717 rbd_put_client(rbdc
);
3720 ceph_destroy_options(ceph_opts
);
3724 module_put(THIS_MODULE
);
3726 dout("Error adding device %s\n", buf
);
3728 return (ssize_t
) rc
;
3731 static struct rbd_device
*__rbd_get_dev(unsigned long dev_id
)
3733 struct list_head
*tmp
;
3734 struct rbd_device
*rbd_dev
;
3736 spin_lock(&rbd_dev_list_lock
);
3737 list_for_each(tmp
, &rbd_dev_list
) {
3738 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
3739 if (rbd_dev
->dev_id
== dev_id
) {
3740 spin_unlock(&rbd_dev_list_lock
);
3744 spin_unlock(&rbd_dev_list_lock
);
3748 static void rbd_dev_release(struct device
*dev
)
3750 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
3752 if (rbd_dev
->watch_request
) {
3753 struct ceph_client
*client
= rbd_dev
->rbd_client
->client
;
3755 ceph_osdc_unregister_linger_request(&client
->osdc
,
3756 rbd_dev
->watch_request
);
3758 if (rbd_dev
->watch_event
)
3759 rbd_req_sync_unwatch(rbd_dev
);
3762 /* clean up and free blkdev */
3763 rbd_free_disk(rbd_dev
);
3764 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
3766 /* release allocated disk header fields */
3767 rbd_header_free(&rbd_dev
->header
);
3769 /* done with the id, and with the rbd_dev */
3770 rbd_dev_id_put(rbd_dev
);
3771 rbd_assert(rbd_dev
->rbd_client
!= NULL
);
3772 rbd_dev_destroy(rbd_dev
);
3774 /* release module ref */
3775 module_put(THIS_MODULE
);
3778 static ssize_t
rbd_remove(struct bus_type
*bus
,
3782 struct rbd_device
*rbd_dev
= NULL
;
3787 rc
= strict_strtoul(buf
, 10, &ul
);
3791 /* convert to int; abort if we lost anything in the conversion */
3792 target_id
= (int) ul
;
3793 if (target_id
!= ul
)
3796 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
3798 rbd_dev
= __rbd_get_dev(target_id
);
3804 if (rbd_dev
->open_count
) {
3809 rbd_remove_all_snaps(rbd_dev
);
3810 rbd_bus_del_dev(rbd_dev
);
3813 mutex_unlock(&ctl_mutex
);
3819 * create control files in sysfs
3822 static int rbd_sysfs_init(void)
3826 ret
= device_register(&rbd_root_dev
);
3830 ret
= bus_register(&rbd_bus_type
);
3832 device_unregister(&rbd_root_dev
);
3837 static void rbd_sysfs_cleanup(void)
3839 bus_unregister(&rbd_bus_type
);
3840 device_unregister(&rbd_root_dev
);
3843 int __init
rbd_init(void)
3847 rc
= rbd_sysfs_init();
3850 pr_info("loaded " RBD_DRV_NAME_LONG
"\n");
3854 void __exit
rbd_exit(void)
3856 rbd_sysfs_cleanup();
3859 module_init(rbd_init
);
3860 module_exit(rbd_exit
);
3862 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
3863 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
3864 MODULE_DESCRIPTION("rados block device");
3866 /* following authorship retained from original osdblk.c */
3867 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
3869 MODULE_LICENSE("GPL");