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 these defined elsewhere */
57 #define U8_MAX ((u8) (~0U))
58 #define U16_MAX ((u16) (~0U))
59 #define U32_MAX ((u32) (~0U))
60 #define U64_MAX ((u64) (~0ULL))
62 #define RBD_DRV_NAME "rbd"
63 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
65 #define RBD_MINORS_PER_MAJOR 256 /* max minors per blkdev */
67 #define RBD_SNAP_DEV_NAME_PREFIX "snap_"
68 #define RBD_MAX_SNAP_NAME_LEN \
69 (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
71 #define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */
73 #define RBD_SNAP_HEAD_NAME "-"
75 /* This allows a single page to hold an image name sent by OSD */
76 #define RBD_IMAGE_NAME_LEN_MAX (PAGE_SIZE - sizeof (__le32) - 1)
77 #define RBD_IMAGE_ID_LEN_MAX 64
79 #define RBD_OBJ_PREFIX_LEN_MAX 64
83 #define RBD_FEATURE_LAYERING 1
85 /* Features supported by this (client software) implementation. */
87 #define RBD_FEATURES_ALL (0)
90 * An RBD device name will be "rbd#", where the "rbd" comes from
91 * RBD_DRV_NAME above, and # is a unique integer identifier.
92 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
93 * enough to hold all possible device names.
95 #define DEV_NAME_LEN 32
96 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1)
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).
157 * an instance of the client. multiple devices may share an rbd client.
160 struct ceph_client
*client
;
162 struct list_head node
;
165 struct rbd_img_request
;
166 typedef void (*rbd_img_callback_t
)(struct rbd_img_request
*);
168 #define BAD_WHICH U32_MAX /* Good which or bad which, which? */
170 struct rbd_obj_request
;
171 typedef void (*rbd_obj_callback_t
)(struct rbd_obj_request
*);
173 enum obj_request_type
{
174 OBJ_REQUEST_NODATA
, OBJ_REQUEST_BIO
, OBJ_REQUEST_PAGES
177 struct rbd_obj_request
{
178 const char *object_name
;
179 u64 offset
; /* object start byte */
180 u64 length
; /* bytes from offset */
182 struct rbd_img_request
*img_request
;
183 struct list_head links
; /* img_request->obj_requests */
184 u32 which
; /* posn image request list */
186 enum obj_request_type type
;
188 struct bio
*bio_list
;
195 struct ceph_osd_request
*osd_req
;
197 u64 xferred
; /* bytes transferred */
202 rbd_obj_callback_t callback
;
203 struct completion completion
;
208 struct rbd_img_request
{
210 struct rbd_device
*rbd_dev
;
211 u64 offset
; /* starting image byte offset */
212 u64 length
; /* byte count from offset */
213 bool write_request
; /* false for read */
215 struct ceph_snap_context
*snapc
; /* for writes */
216 u64 snap_id
; /* for reads */
218 spinlock_t completion_lock
;/* protects next_completion */
220 rbd_img_callback_t callback
;
222 u32 obj_request_count
;
223 struct list_head obj_requests
; /* rbd_obj_request structs */
228 #define for_each_obj_request(ireq, oreq) \
229 list_for_each_entry(oreq, &(ireq)->obj_requests, links)
230 #define for_each_obj_request_from(ireq, oreq) \
231 list_for_each_entry_from(oreq, &(ireq)->obj_requests, links)
232 #define for_each_obj_request_safe(ireq, oreq, n) \
233 list_for_each_entry_safe_reverse(oreq, n, &(ireq)->obj_requests, links)
239 struct list_head node
;
254 int dev_id
; /* blkdev unique id */
256 int major
; /* blkdev assigned major */
257 struct gendisk
*disk
; /* blkdev's gendisk and rq */
259 u32 image_format
; /* Either 1 or 2 */
260 struct rbd_client
*rbd_client
;
262 char name
[DEV_NAME_LEN
]; /* blkdev name, e.g. rbd3 */
264 spinlock_t lock
; /* queue, flags, open_count */
266 struct rbd_image_header header
;
267 unsigned long flags
; /* possibly lock protected */
268 struct rbd_spec
*spec
;
272 struct ceph_file_layout layout
;
274 struct ceph_osd_event
*watch_event
;
275 struct rbd_obj_request
*watch_request
;
277 struct rbd_spec
*parent_spec
;
280 /* protects updating the header */
281 struct rw_semaphore header_rwsem
;
283 struct rbd_mapping mapping
;
285 struct list_head node
;
287 /* list of snapshots */
288 struct list_head snaps
;
292 unsigned long open_count
; /* protected by lock */
296 * Flag bits for rbd_dev->flags. If atomicity is required,
297 * rbd_dev->lock is used to protect access.
299 * Currently, only the "removing" flag (which is coupled with the
300 * "open_count" field) requires atomic access.
303 RBD_DEV_FLAG_EXISTS
, /* mapped snapshot has not been deleted */
304 RBD_DEV_FLAG_REMOVING
, /* this mapping is being removed */
307 static DEFINE_MUTEX(ctl_mutex
); /* Serialize open/close/setup/teardown */
309 static LIST_HEAD(rbd_dev_list
); /* devices */
310 static DEFINE_SPINLOCK(rbd_dev_list_lock
);
312 static LIST_HEAD(rbd_client_list
); /* clients */
313 static DEFINE_SPINLOCK(rbd_client_list_lock
);
315 static int rbd_dev_snaps_update(struct rbd_device
*rbd_dev
);
316 static int rbd_dev_snaps_register(struct rbd_device
*rbd_dev
);
318 static void rbd_dev_release(struct device
*dev
);
319 static void rbd_remove_snap_dev(struct rbd_snap
*snap
);
321 static ssize_t
rbd_add(struct bus_type
*bus
, const char *buf
,
323 static ssize_t
rbd_remove(struct bus_type
*bus
, const char *buf
,
326 static struct bus_attribute rbd_bus_attrs
[] = {
327 __ATTR(add
, S_IWUSR
, NULL
, rbd_add
),
328 __ATTR(remove
, S_IWUSR
, NULL
, rbd_remove
),
332 static struct bus_type rbd_bus_type
= {
334 .bus_attrs
= rbd_bus_attrs
,
337 static void rbd_root_dev_release(struct device
*dev
)
341 static struct device rbd_root_dev
= {
343 .release
= rbd_root_dev_release
,
346 static __printf(2, 3)
347 void rbd_warn(struct rbd_device
*rbd_dev
, const char *fmt
, ...)
349 struct va_format vaf
;
357 printk(KERN_WARNING
"%s: %pV\n", RBD_DRV_NAME
, &vaf
);
358 else if (rbd_dev
->disk
)
359 printk(KERN_WARNING
"%s: %s: %pV\n",
360 RBD_DRV_NAME
, rbd_dev
->disk
->disk_name
, &vaf
);
361 else if (rbd_dev
->spec
&& rbd_dev
->spec
->image_name
)
362 printk(KERN_WARNING
"%s: image %s: %pV\n",
363 RBD_DRV_NAME
, rbd_dev
->spec
->image_name
, &vaf
);
364 else if (rbd_dev
->spec
&& rbd_dev
->spec
->image_id
)
365 printk(KERN_WARNING
"%s: id %s: %pV\n",
366 RBD_DRV_NAME
, rbd_dev
->spec
->image_id
, &vaf
);
368 printk(KERN_WARNING
"%s: rbd_dev %p: %pV\n",
369 RBD_DRV_NAME
, rbd_dev
, &vaf
);
374 #define rbd_assert(expr) \
375 if (unlikely(!(expr))) { \
376 printk(KERN_ERR "\nAssertion failure in %s() " \
378 "\trbd_assert(%s);\n\n", \
379 __func__, __LINE__, #expr); \
382 #else /* !RBD_DEBUG */
383 # define rbd_assert(expr) ((void) 0)
384 #endif /* !RBD_DEBUG */
386 static int rbd_dev_refresh(struct rbd_device
*rbd_dev
, u64
*hver
);
387 static int rbd_dev_v2_refresh(struct rbd_device
*rbd_dev
, u64
*hver
);
389 static int rbd_open(struct block_device
*bdev
, fmode_t mode
)
391 struct rbd_device
*rbd_dev
= bdev
->bd_disk
->private_data
;
392 bool removing
= false;
394 if ((mode
& FMODE_WRITE
) && rbd_dev
->mapping
.read_only
)
397 spin_lock_irq(&rbd_dev
->lock
);
398 if (test_bit(RBD_DEV_FLAG_REMOVING
, &rbd_dev
->flags
))
401 rbd_dev
->open_count
++;
402 spin_unlock_irq(&rbd_dev
->lock
);
406 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
407 (void) get_device(&rbd_dev
->dev
);
408 set_device_ro(bdev
, rbd_dev
->mapping
.read_only
);
409 mutex_unlock(&ctl_mutex
);
414 static int rbd_release(struct gendisk
*disk
, fmode_t mode
)
416 struct rbd_device
*rbd_dev
= disk
->private_data
;
417 unsigned long open_count_before
;
419 spin_lock_irq(&rbd_dev
->lock
);
420 open_count_before
= rbd_dev
->open_count
--;
421 spin_unlock_irq(&rbd_dev
->lock
);
422 rbd_assert(open_count_before
> 0);
424 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
425 put_device(&rbd_dev
->dev
);
426 mutex_unlock(&ctl_mutex
);
431 static const struct block_device_operations rbd_bd_ops
= {
432 .owner
= THIS_MODULE
,
434 .release
= rbd_release
,
438 * Initialize an rbd client instance.
441 static struct rbd_client
*rbd_client_create(struct ceph_options
*ceph_opts
)
443 struct rbd_client
*rbdc
;
446 dout("rbd_client_create\n");
447 rbdc
= kmalloc(sizeof(struct rbd_client
), GFP_KERNEL
);
451 kref_init(&rbdc
->kref
);
452 INIT_LIST_HEAD(&rbdc
->node
);
454 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
456 rbdc
->client
= ceph_create_client(ceph_opts
, rbdc
, 0, 0);
457 if (IS_ERR(rbdc
->client
))
459 ceph_opts
= NULL
; /* Now rbdc->client is responsible for ceph_opts */
461 ret
= ceph_open_session(rbdc
->client
);
465 spin_lock(&rbd_client_list_lock
);
466 list_add_tail(&rbdc
->node
, &rbd_client_list
);
467 spin_unlock(&rbd_client_list_lock
);
469 mutex_unlock(&ctl_mutex
);
471 dout("rbd_client_create created %p\n", rbdc
);
475 ceph_destroy_client(rbdc
->client
);
477 mutex_unlock(&ctl_mutex
);
481 ceph_destroy_options(ceph_opts
);
486 * Find a ceph client with specific addr and configuration. If
487 * found, bump its reference count.
489 static struct rbd_client
*rbd_client_find(struct ceph_options
*ceph_opts
)
491 struct rbd_client
*client_node
;
494 if (ceph_opts
->flags
& CEPH_OPT_NOSHARE
)
497 spin_lock(&rbd_client_list_lock
);
498 list_for_each_entry(client_node
, &rbd_client_list
, node
) {
499 if (!ceph_compare_options(ceph_opts
, client_node
->client
)) {
500 kref_get(&client_node
->kref
);
505 spin_unlock(&rbd_client_list_lock
);
507 return found
? client_node
: NULL
;
517 /* string args above */
520 /* Boolean args above */
524 static match_table_t rbd_opts_tokens
= {
526 /* string args above */
527 {Opt_read_only
, "read_only"},
528 {Opt_read_only
, "ro"}, /* Alternate spelling */
529 {Opt_read_write
, "read_write"},
530 {Opt_read_write
, "rw"}, /* Alternate spelling */
531 /* Boolean args above */
539 #define RBD_READ_ONLY_DEFAULT false
541 static int parse_rbd_opts_token(char *c
, void *private)
543 struct rbd_options
*rbd_opts
= private;
544 substring_t argstr
[MAX_OPT_ARGS
];
545 int token
, intval
, ret
;
547 token
= match_token(c
, rbd_opts_tokens
, argstr
);
551 if (token
< Opt_last_int
) {
552 ret
= match_int(&argstr
[0], &intval
);
554 pr_err("bad mount option arg (not int) "
558 dout("got int token %d val %d\n", token
, intval
);
559 } else if (token
> Opt_last_int
&& token
< Opt_last_string
) {
560 dout("got string token %d val %s\n", token
,
562 } else if (token
> Opt_last_string
&& token
< Opt_last_bool
) {
563 dout("got Boolean token %d\n", token
);
565 dout("got token %d\n", token
);
570 rbd_opts
->read_only
= true;
573 rbd_opts
->read_only
= false;
583 * Get a ceph client with specific addr and configuration, if one does
584 * not exist create it.
586 static struct rbd_client
*rbd_get_client(struct ceph_options
*ceph_opts
)
588 struct rbd_client
*rbdc
;
590 rbdc
= rbd_client_find(ceph_opts
);
591 if (rbdc
) /* using an existing client */
592 ceph_destroy_options(ceph_opts
);
594 rbdc
= rbd_client_create(ceph_opts
);
600 * Destroy ceph client
602 * Caller must hold rbd_client_list_lock.
604 static void rbd_client_release(struct kref
*kref
)
606 struct rbd_client
*rbdc
= container_of(kref
, struct rbd_client
, kref
);
608 dout("rbd_release_client %p\n", rbdc
);
609 spin_lock(&rbd_client_list_lock
);
610 list_del(&rbdc
->node
);
611 spin_unlock(&rbd_client_list_lock
);
613 ceph_destroy_client(rbdc
->client
);
618 * Drop reference to ceph client node. If it's not referenced anymore, release
621 static void rbd_put_client(struct rbd_client
*rbdc
)
624 kref_put(&rbdc
->kref
, rbd_client_release
);
627 static bool rbd_image_format_valid(u32 image_format
)
629 return image_format
== 1 || image_format
== 2;
632 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk
*ondisk
)
637 /* The header has to start with the magic rbd header text */
638 if (memcmp(&ondisk
->text
, RBD_HEADER_TEXT
, sizeof (RBD_HEADER_TEXT
)))
641 /* The bio layer requires at least sector-sized I/O */
643 if (ondisk
->options
.order
< SECTOR_SHIFT
)
646 /* If we use u64 in a few spots we may be able to loosen this */
648 if (ondisk
->options
.order
> 8 * sizeof (int) - 1)
652 * The size of a snapshot header has to fit in a size_t, and
653 * that limits the number of snapshots.
655 snap_count
= le32_to_cpu(ondisk
->snap_count
);
656 size
= SIZE_MAX
- sizeof (struct ceph_snap_context
);
657 if (snap_count
> size
/ sizeof (__le64
))
661 * Not only that, but the size of the entire the snapshot
662 * header must also be representable in a size_t.
664 size
-= snap_count
* sizeof (__le64
);
665 if ((u64
) size
< le64_to_cpu(ondisk
->snap_names_len
))
672 * Create a new header structure, translate header format from the on-disk
675 static int rbd_header_from_disk(struct rbd_image_header
*header
,
676 struct rbd_image_header_ondisk
*ondisk
)
683 memset(header
, 0, sizeof (*header
));
685 snap_count
= le32_to_cpu(ondisk
->snap_count
);
687 len
= strnlen(ondisk
->object_prefix
, sizeof (ondisk
->object_prefix
));
688 header
->object_prefix
= kmalloc(len
+ 1, GFP_KERNEL
);
689 if (!header
->object_prefix
)
691 memcpy(header
->object_prefix
, ondisk
->object_prefix
, len
);
692 header
->object_prefix
[len
] = '\0';
695 u64 snap_names_len
= le64_to_cpu(ondisk
->snap_names_len
);
697 /* Save a copy of the snapshot names */
699 if (snap_names_len
> (u64
) SIZE_MAX
)
701 header
->snap_names
= kmalloc(snap_names_len
, GFP_KERNEL
);
702 if (!header
->snap_names
)
705 * Note that rbd_dev_v1_header_read() guarantees
706 * the ondisk buffer we're working with has
707 * snap_names_len bytes beyond the end of the
708 * snapshot id array, this memcpy() is safe.
710 memcpy(header
->snap_names
, &ondisk
->snaps
[snap_count
],
713 /* Record each snapshot's size */
715 size
= snap_count
* sizeof (*header
->snap_sizes
);
716 header
->snap_sizes
= kmalloc(size
, GFP_KERNEL
);
717 if (!header
->snap_sizes
)
719 for (i
= 0; i
< snap_count
; i
++)
720 header
->snap_sizes
[i
] =
721 le64_to_cpu(ondisk
->snaps
[i
].image_size
);
723 WARN_ON(ondisk
->snap_names_len
);
724 header
->snap_names
= NULL
;
725 header
->snap_sizes
= NULL
;
728 header
->features
= 0; /* No features support in v1 images */
729 header
->obj_order
= ondisk
->options
.order
;
730 header
->crypt_type
= ondisk
->options
.crypt_type
;
731 header
->comp_type
= ondisk
->options
.comp_type
;
733 /* Allocate and fill in the snapshot context */
735 header
->image_size
= le64_to_cpu(ondisk
->image_size
);
736 size
= sizeof (struct ceph_snap_context
);
737 size
+= snap_count
* sizeof (header
->snapc
->snaps
[0]);
738 header
->snapc
= kzalloc(size
, GFP_KERNEL
);
742 atomic_set(&header
->snapc
->nref
, 1);
743 header
->snapc
->seq
= le64_to_cpu(ondisk
->snap_seq
);
744 header
->snapc
->num_snaps
= snap_count
;
745 for (i
= 0; i
< snap_count
; i
++)
746 header
->snapc
->snaps
[i
] =
747 le64_to_cpu(ondisk
->snaps
[i
].id
);
752 kfree(header
->snap_sizes
);
753 header
->snap_sizes
= NULL
;
754 kfree(header
->snap_names
);
755 header
->snap_names
= NULL
;
756 kfree(header
->object_prefix
);
757 header
->object_prefix
= NULL
;
762 static const char *rbd_snap_name(struct rbd_device
*rbd_dev
, u64 snap_id
)
764 struct rbd_snap
*snap
;
766 if (snap_id
== CEPH_NOSNAP
)
767 return RBD_SNAP_HEAD_NAME
;
769 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
)
770 if (snap_id
== snap
->id
)
776 static int snap_by_name(struct rbd_device
*rbd_dev
, const char *snap_name
)
779 struct rbd_snap
*snap
;
781 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
782 if (!strcmp(snap_name
, snap
->name
)) {
783 rbd_dev
->spec
->snap_id
= snap
->id
;
784 rbd_dev
->mapping
.size
= snap
->size
;
785 rbd_dev
->mapping
.features
= snap
->features
;
794 static int rbd_dev_set_mapping(struct rbd_device
*rbd_dev
)
798 if (!memcmp(rbd_dev
->spec
->snap_name
, RBD_SNAP_HEAD_NAME
,
799 sizeof (RBD_SNAP_HEAD_NAME
))) {
800 rbd_dev
->spec
->snap_id
= CEPH_NOSNAP
;
801 rbd_dev
->mapping
.size
= rbd_dev
->header
.image_size
;
802 rbd_dev
->mapping
.features
= rbd_dev
->header
.features
;
805 ret
= snap_by_name(rbd_dev
, rbd_dev
->spec
->snap_name
);
808 rbd_dev
->mapping
.read_only
= true;
810 set_bit(RBD_DEV_FLAG_EXISTS
, &rbd_dev
->flags
);
816 static void rbd_header_free(struct rbd_image_header
*header
)
818 kfree(header
->object_prefix
);
819 header
->object_prefix
= NULL
;
820 kfree(header
->snap_sizes
);
821 header
->snap_sizes
= NULL
;
822 kfree(header
->snap_names
);
823 header
->snap_names
= NULL
;
824 ceph_put_snap_context(header
->snapc
);
825 header
->snapc
= NULL
;
828 static const char *rbd_segment_name(struct rbd_device
*rbd_dev
, u64 offset
)
834 name
= kmalloc(MAX_OBJ_NAME_SIZE
+ 1, GFP_NOIO
);
837 segment
= offset
>> rbd_dev
->header
.obj_order
;
838 ret
= snprintf(name
, MAX_OBJ_NAME_SIZE
+ 1, "%s.%012llx",
839 rbd_dev
->header
.object_prefix
, segment
);
840 if (ret
< 0 || ret
> MAX_OBJ_NAME_SIZE
) {
841 pr_err("error formatting segment name for #%llu (%d)\n",
850 static u64
rbd_segment_offset(struct rbd_device
*rbd_dev
, u64 offset
)
852 u64 segment_size
= (u64
) 1 << rbd_dev
->header
.obj_order
;
854 return offset
& (segment_size
- 1);
857 static u64
rbd_segment_length(struct rbd_device
*rbd_dev
,
858 u64 offset
, u64 length
)
860 u64 segment_size
= (u64
) 1 << rbd_dev
->header
.obj_order
;
862 offset
&= segment_size
- 1;
864 rbd_assert(length
<= U64_MAX
- offset
);
865 if (offset
+ length
> segment_size
)
866 length
= segment_size
- offset
;
872 * returns the size of an object in the image
874 static u64
rbd_obj_bytes(struct rbd_image_header
*header
)
876 return 1 << header
->obj_order
;
883 static void bio_chain_put(struct bio
*chain
)
889 chain
= chain
->bi_next
;
895 * zeros a bio chain, starting at specific offset
897 static void zero_bio_chain(struct bio
*chain
, int start_ofs
)
906 bio_for_each_segment(bv
, chain
, i
) {
907 if (pos
+ bv
->bv_len
> start_ofs
) {
908 int remainder
= max(start_ofs
- pos
, 0);
909 buf
= bvec_kmap_irq(bv
, &flags
);
910 memset(buf
+ remainder
, 0,
911 bv
->bv_len
- remainder
);
912 bvec_kunmap_irq(buf
, &flags
);
917 chain
= chain
->bi_next
;
922 * Clone a portion of a bio, starting at the given byte offset
923 * and continuing for the number of bytes indicated.
925 static struct bio
*bio_clone_range(struct bio
*bio_src
,
934 unsigned short end_idx
;
938 /* Handle the easy case for the caller */
940 if (!offset
&& len
== bio_src
->bi_size
)
941 return bio_clone(bio_src
, gfpmask
);
943 if (WARN_ON_ONCE(!len
))
945 if (WARN_ON_ONCE(len
> bio_src
->bi_size
))
947 if (WARN_ON_ONCE(offset
> bio_src
->bi_size
- len
))
950 /* Find first affected segment... */
953 __bio_for_each_segment(bv
, bio_src
, idx
, 0) {
954 if (resid
< bv
->bv_len
)
960 /* ...and the last affected segment */
963 __bio_for_each_segment(bv
, bio_src
, end_idx
, idx
) {
964 if (resid
<= bv
->bv_len
)
968 vcnt
= end_idx
- idx
+ 1;
970 /* Build the clone */
972 bio
= bio_alloc(gfpmask
, (unsigned int) vcnt
);
974 return NULL
; /* ENOMEM */
976 bio
->bi_bdev
= bio_src
->bi_bdev
;
977 bio
->bi_sector
= bio_src
->bi_sector
+ (offset
>> SECTOR_SHIFT
);
978 bio
->bi_rw
= bio_src
->bi_rw
;
979 bio
->bi_flags
|= 1 << BIO_CLONED
;
982 * Copy over our part of the bio_vec, then update the first
983 * and last (or only) entries.
985 memcpy(&bio
->bi_io_vec
[0], &bio_src
->bi_io_vec
[idx
],
986 vcnt
* sizeof (struct bio_vec
));
987 bio
->bi_io_vec
[0].bv_offset
+= voff
;
989 bio
->bi_io_vec
[0].bv_len
-= voff
;
990 bio
->bi_io_vec
[vcnt
- 1].bv_len
= resid
;
992 bio
->bi_io_vec
[0].bv_len
= len
;
1003 * Clone a portion of a bio chain, starting at the given byte offset
1004 * into the first bio in the source chain and continuing for the
1005 * number of bytes indicated. The result is another bio chain of
1006 * exactly the given length, or a null pointer on error.
1008 * The bio_src and offset parameters are both in-out. On entry they
1009 * refer to the first source bio and the offset into that bio where
1010 * the start of data to be cloned is located.
1012 * On return, bio_src is updated to refer to the bio in the source
1013 * chain that contains first un-cloned byte, and *offset will
1014 * contain the offset of that byte within that bio.
1016 static struct bio
*bio_chain_clone_range(struct bio
**bio_src
,
1017 unsigned int *offset
,
1021 struct bio
*bi
= *bio_src
;
1022 unsigned int off
= *offset
;
1023 struct bio
*chain
= NULL
;
1026 /* Build up a chain of clone bios up to the limit */
1028 if (!bi
|| off
>= bi
->bi_size
|| !len
)
1029 return NULL
; /* Nothing to clone */
1033 unsigned int bi_size
;
1037 rbd_warn(NULL
, "bio_chain exhausted with %u left", len
);
1038 goto out_err
; /* EINVAL; ran out of bio's */
1040 bi_size
= min_t(unsigned int, bi
->bi_size
- off
, len
);
1041 bio
= bio_clone_range(bi
, off
, bi_size
, gfpmask
);
1043 goto out_err
; /* ENOMEM */
1046 end
= &bio
->bi_next
;
1049 if (off
== bi
->bi_size
) {
1060 bio_chain_put(chain
);
1065 static void rbd_obj_request_get(struct rbd_obj_request
*obj_request
)
1067 kref_get(&obj_request
->kref
);
1070 static void rbd_obj_request_destroy(struct kref
*kref
);
1071 static void rbd_obj_request_put(struct rbd_obj_request
*obj_request
)
1073 rbd_assert(obj_request
!= NULL
);
1074 kref_put(&obj_request
->kref
, rbd_obj_request_destroy
);
1077 static void rbd_img_request_get(struct rbd_img_request
*img_request
)
1079 kref_get(&img_request
->kref
);
1082 static void rbd_img_request_destroy(struct kref
*kref
);
1083 static void rbd_img_request_put(struct rbd_img_request
*img_request
)
1085 rbd_assert(img_request
!= NULL
);
1086 kref_put(&img_request
->kref
, rbd_img_request_destroy
);
1089 static inline void rbd_img_obj_request_add(struct rbd_img_request
*img_request
,
1090 struct rbd_obj_request
*obj_request
)
1092 rbd_assert(obj_request
->img_request
== NULL
);
1094 rbd_obj_request_get(obj_request
);
1095 obj_request
->img_request
= img_request
;
1096 obj_request
->which
= img_request
->obj_request_count
;
1097 rbd_assert(obj_request
->which
!= BAD_WHICH
);
1098 img_request
->obj_request_count
++;
1099 list_add_tail(&obj_request
->links
, &img_request
->obj_requests
);
1102 static inline void rbd_img_obj_request_del(struct rbd_img_request
*img_request
,
1103 struct rbd_obj_request
*obj_request
)
1105 rbd_assert(obj_request
->which
!= BAD_WHICH
);
1107 list_del(&obj_request
->links
);
1108 rbd_assert(img_request
->obj_request_count
> 0);
1109 img_request
->obj_request_count
--;
1110 rbd_assert(obj_request
->which
== img_request
->obj_request_count
);
1111 obj_request
->which
= BAD_WHICH
;
1112 rbd_assert(obj_request
->img_request
== img_request
);
1113 obj_request
->img_request
= NULL
;
1114 obj_request
->callback
= NULL
;
1115 rbd_obj_request_put(obj_request
);
1118 static bool obj_request_type_valid(enum obj_request_type type
)
1121 case OBJ_REQUEST_NODATA
:
1122 case OBJ_REQUEST_BIO
:
1123 case OBJ_REQUEST_PAGES
:
1130 struct ceph_osd_req_op
*rbd_osd_req_op_create(u16 opcode
, ...)
1132 struct ceph_osd_req_op
*op
;
1136 op
= kzalloc(sizeof (*op
), GFP_NOIO
);
1140 va_start(args
, opcode
);
1142 case CEPH_OSD_OP_READ
:
1143 case CEPH_OSD_OP_WRITE
:
1144 /* rbd_osd_req_op_create(READ, offset, length) */
1145 /* rbd_osd_req_op_create(WRITE, offset, length) */
1146 op
->extent
.offset
= va_arg(args
, u64
);
1147 op
->extent
.length
= va_arg(args
, u64
);
1148 if (opcode
== CEPH_OSD_OP_WRITE
)
1149 op
->payload_len
= op
->extent
.length
;
1151 case CEPH_OSD_OP_STAT
:
1153 case CEPH_OSD_OP_CALL
:
1154 /* rbd_osd_req_op_create(CALL, class, method, data, datalen) */
1155 op
->cls
.class_name
= va_arg(args
, char *);
1156 size
= strlen(op
->cls
.class_name
);
1157 rbd_assert(size
<= (size_t) U8_MAX
);
1158 op
->cls
.class_len
= size
;
1159 op
->payload_len
= size
;
1161 op
->cls
.method_name
= va_arg(args
, char *);
1162 size
= strlen(op
->cls
.method_name
);
1163 rbd_assert(size
<= (size_t) U8_MAX
);
1164 op
->cls
.method_len
= size
;
1165 op
->payload_len
+= size
;
1168 op
->cls
.indata
= va_arg(args
, void *);
1169 size
= va_arg(args
, size_t);
1170 rbd_assert(size
<= (size_t) U32_MAX
);
1171 op
->cls
.indata_len
= (u32
) size
;
1172 op
->payload_len
+= size
;
1174 case CEPH_OSD_OP_NOTIFY_ACK
:
1175 case CEPH_OSD_OP_WATCH
:
1176 /* rbd_osd_req_op_create(NOTIFY_ACK, cookie, version) */
1177 /* rbd_osd_req_op_create(WATCH, cookie, version, flag) */
1178 op
->watch
.cookie
= va_arg(args
, u64
);
1179 op
->watch
.ver
= va_arg(args
, u64
);
1180 op
->watch
.ver
= cpu_to_le64(op
->watch
.ver
);
1181 if (opcode
== CEPH_OSD_OP_WATCH
&& va_arg(args
, int))
1182 op
->watch
.flag
= (u8
) 1;
1185 rbd_warn(NULL
, "unsupported opcode %hu\n", opcode
);
1195 static void rbd_osd_req_op_destroy(struct ceph_osd_req_op
*op
)
1200 static int rbd_obj_request_submit(struct ceph_osd_client
*osdc
,
1201 struct rbd_obj_request
*obj_request
)
1203 return ceph_osdc_start_request(osdc
, obj_request
->osd_req
, false);
1206 static void rbd_img_request_complete(struct rbd_img_request
*img_request
)
1208 if (img_request
->callback
)
1209 img_request
->callback(img_request
);
1211 rbd_img_request_put(img_request
);
1214 /* Caller is responsible for rbd_obj_request_destroy(obj_request) */
1216 static int rbd_obj_request_wait(struct rbd_obj_request
*obj_request
)
1218 return wait_for_completion_interruptible(&obj_request
->completion
);
1221 static void obj_request_done_init(struct rbd_obj_request
*obj_request
)
1223 atomic_set(&obj_request
->done
, 0);
1227 static void obj_request_done_set(struct rbd_obj_request
*obj_request
)
1229 atomic_set(&obj_request
->done
, 1);
1233 static bool obj_request_done_test(struct rbd_obj_request
*obj_request
)
1236 return atomic_read(&obj_request
->done
) != 0;
1239 static void rbd_osd_trivial_callback(struct rbd_obj_request
*obj_request
,
1240 struct ceph_osd_op
*op
)
1242 obj_request_done_set(obj_request
);
1245 static void rbd_obj_request_complete(struct rbd_obj_request
*obj_request
)
1247 if (obj_request
->callback
)
1248 obj_request
->callback(obj_request
);
1250 complete_all(&obj_request
->completion
);
1253 static void rbd_osd_read_callback(struct rbd_obj_request
*obj_request
,
1254 struct ceph_osd_op
*op
)
1259 * We support a 64-bit length, but ultimately it has to be
1260 * passed to blk_end_request(), which takes an unsigned int.
1262 xferred
= le64_to_cpu(op
->extent
.length
);
1263 rbd_assert(xferred
< (u64
) UINT_MAX
);
1264 if (obj_request
->result
== (s32
) -ENOENT
) {
1265 zero_bio_chain(obj_request
->bio_list
, 0);
1266 obj_request
->result
= 0;
1267 } else if (xferred
< obj_request
->length
&& !obj_request
->result
) {
1268 zero_bio_chain(obj_request
->bio_list
, xferred
);
1269 xferred
= obj_request
->length
;
1271 obj_request
->xferred
= xferred
;
1272 obj_request_done_set(obj_request
);
1275 static void rbd_osd_write_callback(struct rbd_obj_request
*obj_request
,
1276 struct ceph_osd_op
*op
)
1278 obj_request
->xferred
= le64_to_cpu(op
->extent
.length
);
1279 obj_request_done_set(obj_request
);
1283 * For a simple stat call there's nothing to do. We'll do more if
1284 * this is part of a write sequence for a layered image.
1286 static void rbd_osd_stat_callback(struct rbd_obj_request
*obj_request
,
1287 struct ceph_osd_op
*op
)
1289 obj_request_done_set(obj_request
);
1292 static void rbd_osd_req_callback(struct ceph_osd_request
*osd_req
,
1293 struct ceph_msg
*msg
)
1295 struct rbd_obj_request
*obj_request
= osd_req
->r_priv
;
1296 struct ceph_osd_reply_head
*reply_head
;
1297 struct ceph_osd_op
*op
;
1301 rbd_assert(osd_req
== obj_request
->osd_req
);
1302 rbd_assert(!!obj_request
->img_request
^
1303 (obj_request
->which
== BAD_WHICH
));
1305 obj_request
->xferred
= le32_to_cpu(msg
->hdr
.data_len
);
1306 reply_head
= msg
->front
.iov_base
;
1307 obj_request
->result
= (s32
) le32_to_cpu(reply_head
->result
);
1308 obj_request
->version
= le64_to_cpu(osd_req
->r_reassert_version
.version
);
1310 num_ops
= le32_to_cpu(reply_head
->num_ops
);
1311 WARN_ON(num_ops
!= 1); /* For now */
1313 op
= &reply_head
->ops
[0];
1314 opcode
= le16_to_cpu(op
->op
);
1316 case CEPH_OSD_OP_READ
:
1317 rbd_osd_read_callback(obj_request
, op
);
1319 case CEPH_OSD_OP_WRITE
:
1320 rbd_osd_write_callback(obj_request
, op
);
1322 case CEPH_OSD_OP_STAT
:
1323 rbd_osd_stat_callback(obj_request
, op
);
1325 case CEPH_OSD_OP_CALL
:
1326 case CEPH_OSD_OP_NOTIFY_ACK
:
1327 case CEPH_OSD_OP_WATCH
:
1328 rbd_osd_trivial_callback(obj_request
, op
);
1331 rbd_warn(NULL
, "%s: unsupported op %hu\n",
1332 obj_request
->object_name
, (unsigned short) opcode
);
1336 if (obj_request_done_test(obj_request
))
1337 rbd_obj_request_complete(obj_request
);
1340 static struct ceph_osd_request
*rbd_osd_req_create(
1341 struct rbd_device
*rbd_dev
,
1343 struct rbd_obj_request
*obj_request
,
1344 struct ceph_osd_req_op
*op
)
1346 struct rbd_img_request
*img_request
= obj_request
->img_request
;
1347 struct ceph_snap_context
*snapc
= NULL
;
1348 struct ceph_osd_client
*osdc
;
1349 struct ceph_osd_request
*osd_req
;
1350 struct timespec now
;
1351 struct timespec
*mtime
;
1352 u64 snap_id
= CEPH_NOSNAP
;
1353 u64 offset
= obj_request
->offset
;
1354 u64 length
= obj_request
->length
;
1357 rbd_assert(img_request
->write_request
== write_request
);
1358 if (img_request
->write_request
)
1359 snapc
= img_request
->snapc
;
1361 snap_id
= img_request
->snap_id
;
1364 /* Allocate and initialize the request, for the single op */
1366 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1367 osd_req
= ceph_osdc_alloc_request(osdc
, snapc
, 1, false, GFP_ATOMIC
);
1369 return NULL
; /* ENOMEM */
1371 rbd_assert(obj_request_type_valid(obj_request
->type
));
1372 switch (obj_request
->type
) {
1373 case OBJ_REQUEST_NODATA
:
1374 break; /* Nothing to do */
1375 case OBJ_REQUEST_BIO
:
1376 rbd_assert(obj_request
->bio_list
!= NULL
);
1377 osd_req
->r_bio
= obj_request
->bio_list
;
1379 case OBJ_REQUEST_PAGES
:
1380 osd_req
->r_pages
= obj_request
->pages
;
1381 osd_req
->r_num_pages
= obj_request
->page_count
;
1382 osd_req
->r_page_alignment
= offset
& ~PAGE_MASK
;
1386 if (write_request
) {
1387 osd_req
->r_flags
= CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
;
1391 osd_req
->r_flags
= CEPH_OSD_FLAG_READ
;
1392 mtime
= NULL
; /* not needed for reads */
1393 offset
= 0; /* These are not used... */
1394 length
= 0; /* ...for osd read requests */
1397 osd_req
->r_callback
= rbd_osd_req_callback
;
1398 osd_req
->r_priv
= obj_request
;
1400 osd_req
->r_oid_len
= strlen(obj_request
->object_name
);
1401 rbd_assert(osd_req
->r_oid_len
< sizeof (osd_req
->r_oid
));
1402 memcpy(osd_req
->r_oid
, obj_request
->object_name
, osd_req
->r_oid_len
);
1404 osd_req
->r_file_layout
= rbd_dev
->layout
; /* struct */
1406 /* osd_req will get its own reference to snapc (if non-null) */
1408 ceph_osdc_build_request(osd_req
, offset
, length
, 1, op
,
1409 snapc
, snap_id
, mtime
);
1414 static void rbd_osd_req_destroy(struct ceph_osd_request
*osd_req
)
1416 ceph_osdc_put_request(osd_req
);
1419 /* object_name is assumed to be a non-null pointer and NUL-terminated */
1421 static struct rbd_obj_request
*rbd_obj_request_create(const char *object_name
,
1422 u64 offset
, u64 length
,
1423 enum obj_request_type type
)
1425 struct rbd_obj_request
*obj_request
;
1429 rbd_assert(obj_request_type_valid(type
));
1431 size
= strlen(object_name
) + 1;
1432 obj_request
= kzalloc(sizeof (*obj_request
) + size
, GFP_KERNEL
);
1436 name
= (char *)(obj_request
+ 1);
1437 obj_request
->object_name
= memcpy(name
, object_name
, size
);
1438 obj_request
->offset
= offset
;
1439 obj_request
->length
= length
;
1440 obj_request
->which
= BAD_WHICH
;
1441 obj_request
->type
= type
;
1442 INIT_LIST_HEAD(&obj_request
->links
);
1443 obj_request_done_init(obj_request
);
1444 init_completion(&obj_request
->completion
);
1445 kref_init(&obj_request
->kref
);
1450 static void rbd_obj_request_destroy(struct kref
*kref
)
1452 struct rbd_obj_request
*obj_request
;
1454 obj_request
= container_of(kref
, struct rbd_obj_request
, kref
);
1456 rbd_assert(obj_request
->img_request
== NULL
);
1457 rbd_assert(obj_request
->which
== BAD_WHICH
);
1459 if (obj_request
->osd_req
)
1460 rbd_osd_req_destroy(obj_request
->osd_req
);
1462 rbd_assert(obj_request_type_valid(obj_request
->type
));
1463 switch (obj_request
->type
) {
1464 case OBJ_REQUEST_NODATA
:
1465 break; /* Nothing to do */
1466 case OBJ_REQUEST_BIO
:
1467 if (obj_request
->bio_list
)
1468 bio_chain_put(obj_request
->bio_list
);
1470 case OBJ_REQUEST_PAGES
:
1471 if (obj_request
->pages
)
1472 ceph_release_page_vector(obj_request
->pages
,
1473 obj_request
->page_count
);
1481 * Caller is responsible for filling in the list of object requests
1482 * that comprises the image request, and the Linux request pointer
1483 * (if there is one).
1485 struct rbd_img_request
*rbd_img_request_create(struct rbd_device
*rbd_dev
,
1486 u64 offset
, u64 length
,
1489 struct rbd_img_request
*img_request
;
1490 struct ceph_snap_context
*snapc
= NULL
;
1492 img_request
= kmalloc(sizeof (*img_request
), GFP_ATOMIC
);
1496 if (write_request
) {
1497 down_read(&rbd_dev
->header_rwsem
);
1498 snapc
= ceph_get_snap_context(rbd_dev
->header
.snapc
);
1499 up_read(&rbd_dev
->header_rwsem
);
1500 if (WARN_ON(!snapc
)) {
1502 return NULL
; /* Shouldn't happen */
1506 img_request
->rq
= NULL
;
1507 img_request
->rbd_dev
= rbd_dev
;
1508 img_request
->offset
= offset
;
1509 img_request
->length
= length
;
1510 img_request
->write_request
= write_request
;
1512 img_request
->snapc
= snapc
;
1514 img_request
->snap_id
= rbd_dev
->spec
->snap_id
;
1515 spin_lock_init(&img_request
->completion_lock
);
1516 img_request
->next_completion
= 0;
1517 img_request
->callback
= NULL
;
1518 img_request
->obj_request_count
= 0;
1519 INIT_LIST_HEAD(&img_request
->obj_requests
);
1520 kref_init(&img_request
->kref
);
1522 rbd_img_request_get(img_request
); /* Avoid a warning */
1523 rbd_img_request_put(img_request
); /* TEMPORARY */
1528 static void rbd_img_request_destroy(struct kref
*kref
)
1530 struct rbd_img_request
*img_request
;
1531 struct rbd_obj_request
*obj_request
;
1532 struct rbd_obj_request
*next_obj_request
;
1534 img_request
= container_of(kref
, struct rbd_img_request
, kref
);
1536 for_each_obj_request_safe(img_request
, obj_request
, next_obj_request
)
1537 rbd_img_obj_request_del(img_request
, obj_request
);
1538 rbd_assert(img_request
->obj_request_count
== 0);
1540 if (img_request
->write_request
)
1541 ceph_put_snap_context(img_request
->snapc
);
1546 static int rbd_img_request_fill_bio(struct rbd_img_request
*img_request
,
1547 struct bio
*bio_list
)
1549 struct rbd_device
*rbd_dev
= img_request
->rbd_dev
;
1550 struct rbd_obj_request
*obj_request
= NULL
;
1551 struct rbd_obj_request
*next_obj_request
;
1552 unsigned int bio_offset
;
1557 opcode
= img_request
->write_request
? CEPH_OSD_OP_WRITE
1560 image_offset
= img_request
->offset
;
1561 rbd_assert(image_offset
== bio_list
->bi_sector
<< SECTOR_SHIFT
);
1562 resid
= img_request
->length
;
1563 rbd_assert(resid
> 0);
1565 const char *object_name
;
1566 unsigned int clone_size
;
1567 struct ceph_osd_req_op
*op
;
1571 object_name
= rbd_segment_name(rbd_dev
, image_offset
);
1574 offset
= rbd_segment_offset(rbd_dev
, image_offset
);
1575 length
= rbd_segment_length(rbd_dev
, image_offset
, resid
);
1576 obj_request
= rbd_obj_request_create(object_name
,
1579 kfree(object_name
); /* object request has its own copy */
1583 rbd_assert(length
<= (u64
) UINT_MAX
);
1584 clone_size
= (unsigned int) length
;
1585 obj_request
->bio_list
= bio_chain_clone_range(&bio_list
,
1586 &bio_offset
, clone_size
,
1588 if (!obj_request
->bio_list
)
1592 * Build up the op to use in building the osd
1593 * request. Note that the contents of the op are
1594 * copied by rbd_osd_req_create().
1596 op
= rbd_osd_req_op_create(opcode
, offset
, length
);
1599 obj_request
->osd_req
= rbd_osd_req_create(rbd_dev
,
1600 img_request
->write_request
,
1602 rbd_osd_req_op_destroy(op
);
1603 if (!obj_request
->osd_req
)
1605 /* status and version are initially zero-filled */
1607 rbd_img_obj_request_add(img_request
, obj_request
);
1609 image_offset
+= length
;
1616 rbd_obj_request_put(obj_request
);
1618 for_each_obj_request_safe(img_request
, obj_request
, next_obj_request
)
1619 rbd_obj_request_put(obj_request
);
1624 static void rbd_img_obj_callback(struct rbd_obj_request
*obj_request
)
1626 struct rbd_img_request
*img_request
;
1627 u32 which
= obj_request
->which
;
1630 img_request
= obj_request
->img_request
;
1632 rbd_assert(img_request
!= NULL
);
1633 rbd_assert(img_request
->rq
!= NULL
);
1634 rbd_assert(img_request
->obj_request_count
> 0);
1635 rbd_assert(which
!= BAD_WHICH
);
1636 rbd_assert(which
< img_request
->obj_request_count
);
1637 rbd_assert(which
>= img_request
->next_completion
);
1639 spin_lock_irq(&img_request
->completion_lock
);
1640 if (which
!= img_request
->next_completion
)
1643 for_each_obj_request_from(img_request
, obj_request
) {
1644 unsigned int xferred
;
1648 rbd_assert(which
< img_request
->obj_request_count
);
1650 if (!obj_request_done_test(obj_request
))
1653 rbd_assert(obj_request
->xferred
<= (u64
) UINT_MAX
);
1654 xferred
= (unsigned int) obj_request
->xferred
;
1655 result
= (int) obj_request
->result
;
1657 rbd_warn(NULL
, "obj_request %s result %d xferred %u\n",
1658 img_request
->write_request
? "write" : "read",
1661 more
= blk_end_request(img_request
->rq
, result
, xferred
);
1664 rbd_assert(more
^ (which
== img_request
->obj_request_count
));
1665 img_request
->next_completion
= which
;
1667 spin_unlock_irq(&img_request
->completion_lock
);
1670 rbd_img_request_complete(img_request
);
1673 static int rbd_img_request_submit(struct rbd_img_request
*img_request
)
1675 struct rbd_device
*rbd_dev
= img_request
->rbd_dev
;
1676 struct ceph_osd_client
*osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1677 struct rbd_obj_request
*obj_request
;
1679 for_each_obj_request(img_request
, obj_request
) {
1682 obj_request
->callback
= rbd_img_obj_callback
;
1683 ret
= rbd_obj_request_submit(osdc
, obj_request
);
1687 * The image request has its own reference to each
1688 * of its object requests, so we can safely drop the
1691 rbd_obj_request_put(obj_request
);
1697 static int rbd_obj_notify_ack(struct rbd_device
*rbd_dev
,
1698 u64 ver
, u64 notify_id
)
1700 struct rbd_obj_request
*obj_request
;
1701 struct ceph_osd_req_op
*op
;
1702 struct ceph_osd_client
*osdc
;
1705 obj_request
= rbd_obj_request_create(rbd_dev
->header_name
, 0, 0,
1706 OBJ_REQUEST_NODATA
);
1711 op
= rbd_osd_req_op_create(CEPH_OSD_OP_NOTIFY_ACK
, notify_id
, ver
);
1714 obj_request
->osd_req
= rbd_osd_req_create(rbd_dev
, false,
1716 rbd_osd_req_op_destroy(op
);
1717 if (!obj_request
->osd_req
)
1720 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1721 obj_request
->callback
= rbd_obj_request_put
;
1722 ret
= rbd_obj_request_submit(osdc
, obj_request
);
1725 rbd_obj_request_put(obj_request
);
1730 static void rbd_watch_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1732 struct rbd_device
*rbd_dev
= (struct rbd_device
*)data
;
1739 dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
1740 rbd_dev
->header_name
, (unsigned long long) notify_id
,
1741 (unsigned int) opcode
);
1742 rc
= rbd_dev_refresh(rbd_dev
, &hver
);
1744 rbd_warn(rbd_dev
, "got notification but failed to "
1745 " update snaps: %d\n", rc
);
1747 rbd_obj_notify_ack(rbd_dev
, hver
, notify_id
);
1751 * Request sync osd watch/unwatch. The value of "start" determines
1752 * whether a watch request is being initiated or torn down.
1754 static int rbd_dev_header_watch_sync(struct rbd_device
*rbd_dev
, int start
)
1756 struct ceph_osd_client
*osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1757 struct rbd_obj_request
*obj_request
;
1758 struct ceph_osd_req_op
*op
;
1761 rbd_assert(start
^ !!rbd_dev
->watch_event
);
1762 rbd_assert(start
^ !!rbd_dev
->watch_request
);
1765 ret
= ceph_osdc_create_event(osdc
, rbd_watch_cb
, rbd_dev
,
1766 &rbd_dev
->watch_event
);
1769 rbd_assert(rbd_dev
->watch_event
!= NULL
);
1773 obj_request
= rbd_obj_request_create(rbd_dev
->header_name
, 0, 0,
1774 OBJ_REQUEST_NODATA
);
1778 op
= rbd_osd_req_op_create(CEPH_OSD_OP_WATCH
,
1779 rbd_dev
->watch_event
->cookie
,
1780 rbd_dev
->header
.obj_version
, start
);
1783 obj_request
->osd_req
= rbd_osd_req_create(rbd_dev
, true,
1785 rbd_osd_req_op_destroy(op
);
1786 if (!obj_request
->osd_req
)
1790 ceph_osdc_set_request_linger(osdc
, obj_request
->osd_req
);
1792 ceph_osdc_unregister_linger_request(osdc
,
1793 rbd_dev
->watch_request
->osd_req
);
1794 ret
= rbd_obj_request_submit(osdc
, obj_request
);
1797 ret
= rbd_obj_request_wait(obj_request
);
1800 ret
= obj_request
->result
;
1805 * A watch request is set to linger, so the underlying osd
1806 * request won't go away until we unregister it. We retain
1807 * a pointer to the object request during that time (in
1808 * rbd_dev->watch_request), so we'll keep a reference to
1809 * it. We'll drop that reference (below) after we've
1813 rbd_dev
->watch_request
= obj_request
;
1818 /* We have successfully torn down the watch request */
1820 rbd_obj_request_put(rbd_dev
->watch_request
);
1821 rbd_dev
->watch_request
= NULL
;
1823 /* Cancel the event if we're tearing down, or on error */
1824 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1825 rbd_dev
->watch_event
= NULL
;
1827 rbd_obj_request_put(obj_request
);
1833 * Synchronous osd object method call
1835 static int rbd_obj_method_sync(struct rbd_device
*rbd_dev
,
1836 const char *object_name
,
1837 const char *class_name
,
1838 const char *method_name
,
1839 const char *outbound
,
1840 size_t outbound_size
,
1842 size_t inbound_size
,
1845 struct rbd_obj_request
*obj_request
;
1846 struct ceph_osd_client
*osdc
;
1847 struct ceph_osd_req_op
*op
;
1848 struct page
**pages
;
1853 * Method calls are ultimately read operations but they
1854 * don't involve object data (so no offset or length).
1855 * The result should placed into the inbound buffer
1856 * provided. They also supply outbound data--parameters for
1857 * the object method. Currently if this is present it will
1860 page_count
= (u32
) calc_pages_for(0, inbound_size
);
1861 pages
= ceph_alloc_page_vector(page_count
, GFP_KERNEL
);
1863 return PTR_ERR(pages
);
1866 obj_request
= rbd_obj_request_create(object_name
, 0, 0,
1871 obj_request
->pages
= pages
;
1872 obj_request
->page_count
= page_count
;
1874 op
= rbd_osd_req_op_create(CEPH_OSD_OP_CALL
, class_name
,
1875 method_name
, outbound
, outbound_size
);
1878 obj_request
->osd_req
= rbd_osd_req_create(rbd_dev
, false,
1880 rbd_osd_req_op_destroy(op
);
1881 if (!obj_request
->osd_req
)
1884 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1885 ret
= rbd_obj_request_submit(osdc
, obj_request
);
1888 ret
= rbd_obj_request_wait(obj_request
);
1892 ret
= obj_request
->result
;
1896 ceph_copy_from_page_vector(pages
, inbound
, 0, obj_request
->xferred
);
1898 *version
= obj_request
->version
;
1901 rbd_obj_request_put(obj_request
);
1903 ceph_release_page_vector(pages
, page_count
);
1908 static void rbd_request_fn(struct request_queue
*q
)
1910 struct rbd_device
*rbd_dev
= q
->queuedata
;
1911 bool read_only
= rbd_dev
->mapping
.read_only
;
1915 while ((rq
= blk_fetch_request(q
))) {
1916 bool write_request
= rq_data_dir(rq
) == WRITE
;
1917 struct rbd_img_request
*img_request
;
1921 /* Ignore any non-FS requests that filter through. */
1923 if (rq
->cmd_type
!= REQ_TYPE_FS
) {
1924 dout("%s: non-fs request type %d\n", __func__
,
1925 (int) rq
->cmd_type
);
1926 __blk_end_request_all(rq
, 0);
1930 /* Ignore/skip any zero-length requests */
1932 offset
= (u64
) blk_rq_pos(rq
) << SECTOR_SHIFT
;
1933 length
= (u64
) blk_rq_bytes(rq
);
1936 dout("%s: zero-length request\n", __func__
);
1937 __blk_end_request_all(rq
, 0);
1941 spin_unlock_irq(q
->queue_lock
);
1943 /* Disallow writes to a read-only device */
1945 if (write_request
) {
1949 rbd_assert(rbd_dev
->spec
->snap_id
== CEPH_NOSNAP
);
1953 * Quit early if the mapped snapshot no longer
1954 * exists. It's still possible the snapshot will
1955 * have disappeared by the time our request arrives
1956 * at the osd, but there's no sense in sending it if
1959 if (!test_bit(RBD_DEV_FLAG_EXISTS
, &rbd_dev
->flags
)) {
1960 dout("request for non-existent snapshot");
1961 rbd_assert(rbd_dev
->spec
->snap_id
!= CEPH_NOSNAP
);
1967 if (WARN_ON(offset
&& length
> U64_MAX
- offset
+ 1))
1968 goto end_request
; /* Shouldn't happen */
1971 img_request
= rbd_img_request_create(rbd_dev
, offset
, length
,
1976 img_request
->rq
= rq
;
1978 result
= rbd_img_request_fill_bio(img_request
, rq
->bio
);
1980 result
= rbd_img_request_submit(img_request
);
1982 rbd_img_request_put(img_request
);
1984 spin_lock_irq(q
->queue_lock
);
1986 rbd_warn(rbd_dev
, "obj_request %s result %d\n",
1987 write_request
? "write" : "read", result
);
1988 __blk_end_request_all(rq
, result
);
1994 * a queue callback. Makes sure that we don't create a bio that spans across
1995 * multiple osd objects. One exception would be with a single page bios,
1996 * which we handle later at bio_chain_clone_range()
1998 static int rbd_merge_bvec(struct request_queue
*q
, struct bvec_merge_data
*bmd
,
1999 struct bio_vec
*bvec
)
2001 struct rbd_device
*rbd_dev
= q
->queuedata
;
2002 sector_t sector_offset
;
2003 sector_t sectors_per_obj
;
2004 sector_t obj_sector_offset
;
2008 * Find how far into its rbd object the partition-relative
2009 * bio start sector is to offset relative to the enclosing
2012 sector_offset
= get_start_sect(bmd
->bi_bdev
) + bmd
->bi_sector
;
2013 sectors_per_obj
= 1 << (rbd_dev
->header
.obj_order
- SECTOR_SHIFT
);
2014 obj_sector_offset
= sector_offset
& (sectors_per_obj
- 1);
2017 * Compute the number of bytes from that offset to the end
2018 * of the object. Account for what's already used by the bio.
2020 ret
= (int) (sectors_per_obj
- obj_sector_offset
) << SECTOR_SHIFT
;
2021 if (ret
> bmd
->bi_size
)
2022 ret
-= bmd
->bi_size
;
2027 * Don't send back more than was asked for. And if the bio
2028 * was empty, let the whole thing through because: "Note
2029 * that a block device *must* allow a single page to be
2030 * added to an empty bio."
2032 rbd_assert(bvec
->bv_len
<= PAGE_SIZE
);
2033 if (ret
> (int) bvec
->bv_len
|| !bmd
->bi_size
)
2034 ret
= (int) bvec
->bv_len
;
2039 static void rbd_free_disk(struct rbd_device
*rbd_dev
)
2041 struct gendisk
*disk
= rbd_dev
->disk
;
2046 if (disk
->flags
& GENHD_FL_UP
)
2049 blk_cleanup_queue(disk
->queue
);
2053 static int rbd_obj_read_sync(struct rbd_device
*rbd_dev
,
2054 const char *object_name
,
2055 u64 offset
, u64 length
,
2056 char *buf
, u64
*version
)
2059 struct ceph_osd_req_op
*op
;
2060 struct rbd_obj_request
*obj_request
;
2061 struct ceph_osd_client
*osdc
;
2062 struct page
**pages
= NULL
;
2067 page_count
= (u32
) calc_pages_for(offset
, length
);
2068 pages
= ceph_alloc_page_vector(page_count
, GFP_KERNEL
);
2070 ret
= PTR_ERR(pages
);
2073 obj_request
= rbd_obj_request_create(object_name
, offset
, length
,
2078 obj_request
->pages
= pages
;
2079 obj_request
->page_count
= page_count
;
2081 op
= rbd_osd_req_op_create(CEPH_OSD_OP_READ
, offset
, length
);
2084 obj_request
->osd_req
= rbd_osd_req_create(rbd_dev
, false,
2086 rbd_osd_req_op_destroy(op
);
2087 if (!obj_request
->osd_req
)
2090 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
2091 ret
= rbd_obj_request_submit(osdc
, obj_request
);
2094 ret
= rbd_obj_request_wait(obj_request
);
2098 ret
= obj_request
->result
;
2102 rbd_assert(obj_request
->xferred
<= (u64
) SIZE_MAX
);
2103 size
= (size_t) obj_request
->xferred
;
2104 ceph_copy_from_page_vector(pages
, buf
, 0, size
);
2105 rbd_assert(size
<= (size_t) INT_MAX
);
2108 *version
= obj_request
->version
;
2111 rbd_obj_request_put(obj_request
);
2113 ceph_release_page_vector(pages
, page_count
);
2119 * Read the complete header for the given rbd device.
2121 * Returns a pointer to a dynamically-allocated buffer containing
2122 * the complete and validated header. Caller can pass the address
2123 * of a variable that will be filled in with the version of the
2124 * header object at the time it was read.
2126 * Returns a pointer-coded errno if a failure occurs.
2128 static struct rbd_image_header_ondisk
*
2129 rbd_dev_v1_header_read(struct rbd_device
*rbd_dev
, u64
*version
)
2131 struct rbd_image_header_ondisk
*ondisk
= NULL
;
2138 * The complete header will include an array of its 64-bit
2139 * snapshot ids, followed by the names of those snapshots as
2140 * a contiguous block of NUL-terminated strings. Note that
2141 * the number of snapshots could change by the time we read
2142 * it in, in which case we re-read it.
2149 size
= sizeof (*ondisk
);
2150 size
+= snap_count
* sizeof (struct rbd_image_snap_ondisk
);
2152 ondisk
= kmalloc(size
, GFP_KERNEL
);
2154 return ERR_PTR(-ENOMEM
);
2156 ret
= rbd_obj_read_sync(rbd_dev
, rbd_dev
->header_name
,
2158 (char *) ondisk
, version
);
2161 if (WARN_ON((size_t) ret
< size
)) {
2163 rbd_warn(rbd_dev
, "short header read (want %zd got %d)",
2167 if (!rbd_dev_ondisk_valid(ondisk
)) {
2169 rbd_warn(rbd_dev
, "invalid header");
2173 names_size
= le64_to_cpu(ondisk
->snap_names_len
);
2174 want_count
= snap_count
;
2175 snap_count
= le32_to_cpu(ondisk
->snap_count
);
2176 } while (snap_count
!= want_count
);
2183 return ERR_PTR(ret
);
2187 * reload the ondisk the header
2189 static int rbd_read_header(struct rbd_device
*rbd_dev
,
2190 struct rbd_image_header
*header
)
2192 struct rbd_image_header_ondisk
*ondisk
;
2196 ondisk
= rbd_dev_v1_header_read(rbd_dev
, &ver
);
2198 return PTR_ERR(ondisk
);
2199 ret
= rbd_header_from_disk(header
, ondisk
);
2201 header
->obj_version
= ver
;
2207 static void rbd_remove_all_snaps(struct rbd_device
*rbd_dev
)
2209 struct rbd_snap
*snap
;
2210 struct rbd_snap
*next
;
2212 list_for_each_entry_safe(snap
, next
, &rbd_dev
->snaps
, node
)
2213 rbd_remove_snap_dev(snap
);
2216 static void rbd_update_mapping_size(struct rbd_device
*rbd_dev
)
2220 if (rbd_dev
->spec
->snap_id
!= CEPH_NOSNAP
)
2223 size
= (sector_t
) rbd_dev
->header
.image_size
/ SECTOR_SIZE
;
2224 dout("setting size to %llu sectors", (unsigned long long) size
);
2225 rbd_dev
->mapping
.size
= (u64
) size
;
2226 set_capacity(rbd_dev
->disk
, size
);
2230 * only read the first part of the ondisk header, without the snaps info
2232 static int rbd_dev_v1_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
2235 struct rbd_image_header h
;
2237 ret
= rbd_read_header(rbd_dev
, &h
);
2241 down_write(&rbd_dev
->header_rwsem
);
2243 /* Update image size, and check for resize of mapped image */
2244 rbd_dev
->header
.image_size
= h
.image_size
;
2245 rbd_update_mapping_size(rbd_dev
);
2247 /* rbd_dev->header.object_prefix shouldn't change */
2248 kfree(rbd_dev
->header
.snap_sizes
);
2249 kfree(rbd_dev
->header
.snap_names
);
2250 /* osd requests may still refer to snapc */
2251 ceph_put_snap_context(rbd_dev
->header
.snapc
);
2254 *hver
= h
.obj_version
;
2255 rbd_dev
->header
.obj_version
= h
.obj_version
;
2256 rbd_dev
->header
.image_size
= h
.image_size
;
2257 rbd_dev
->header
.snapc
= h
.snapc
;
2258 rbd_dev
->header
.snap_names
= h
.snap_names
;
2259 rbd_dev
->header
.snap_sizes
= h
.snap_sizes
;
2260 /* Free the extra copy of the object prefix */
2261 WARN_ON(strcmp(rbd_dev
->header
.object_prefix
, h
.object_prefix
));
2262 kfree(h
.object_prefix
);
2264 ret
= rbd_dev_snaps_update(rbd_dev
);
2266 ret
= rbd_dev_snaps_register(rbd_dev
);
2268 up_write(&rbd_dev
->header_rwsem
);
2273 static int rbd_dev_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
2277 rbd_assert(rbd_image_format_valid(rbd_dev
->image_format
));
2278 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2279 if (rbd_dev
->image_format
== 1)
2280 ret
= rbd_dev_v1_refresh(rbd_dev
, hver
);
2282 ret
= rbd_dev_v2_refresh(rbd_dev
, hver
);
2283 mutex_unlock(&ctl_mutex
);
2288 static int rbd_init_disk(struct rbd_device
*rbd_dev
)
2290 struct gendisk
*disk
;
2291 struct request_queue
*q
;
2294 /* create gendisk info */
2295 disk
= alloc_disk(RBD_MINORS_PER_MAJOR
);
2299 snprintf(disk
->disk_name
, sizeof(disk
->disk_name
), RBD_DRV_NAME
"%d",
2301 disk
->major
= rbd_dev
->major
;
2302 disk
->first_minor
= 0;
2303 disk
->fops
= &rbd_bd_ops
;
2304 disk
->private_data
= rbd_dev
;
2306 q
= blk_init_queue(rbd_request_fn
, &rbd_dev
->lock
);
2310 /* We use the default size, but let's be explicit about it. */
2311 blk_queue_physical_block_size(q
, SECTOR_SIZE
);
2313 /* set io sizes to object size */
2314 segment_size
= rbd_obj_bytes(&rbd_dev
->header
);
2315 blk_queue_max_hw_sectors(q
, segment_size
/ SECTOR_SIZE
);
2316 blk_queue_max_segment_size(q
, segment_size
);
2317 blk_queue_io_min(q
, segment_size
);
2318 blk_queue_io_opt(q
, segment_size
);
2320 blk_queue_merge_bvec(q
, rbd_merge_bvec
);
2323 q
->queuedata
= rbd_dev
;
2325 rbd_dev
->disk
= disk
;
2327 set_capacity(rbd_dev
->disk
, rbd_dev
->mapping
.size
/ SECTOR_SIZE
);
2340 static struct rbd_device
*dev_to_rbd_dev(struct device
*dev
)
2342 return container_of(dev
, struct rbd_device
, dev
);
2345 static ssize_t
rbd_size_show(struct device
*dev
,
2346 struct device_attribute
*attr
, char *buf
)
2348 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2351 down_read(&rbd_dev
->header_rwsem
);
2352 size
= get_capacity(rbd_dev
->disk
);
2353 up_read(&rbd_dev
->header_rwsem
);
2355 return sprintf(buf
, "%llu\n", (unsigned long long) size
* SECTOR_SIZE
);
2359 * Note this shows the features for whatever's mapped, which is not
2360 * necessarily the base image.
2362 static ssize_t
rbd_features_show(struct device
*dev
,
2363 struct device_attribute
*attr
, char *buf
)
2365 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2367 return sprintf(buf
, "0x%016llx\n",
2368 (unsigned long long) rbd_dev
->mapping
.features
);
2371 static ssize_t
rbd_major_show(struct device
*dev
,
2372 struct device_attribute
*attr
, char *buf
)
2374 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2376 return sprintf(buf
, "%d\n", rbd_dev
->major
);
2379 static ssize_t
rbd_client_id_show(struct device
*dev
,
2380 struct device_attribute
*attr
, char *buf
)
2382 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2384 return sprintf(buf
, "client%lld\n",
2385 ceph_client_id(rbd_dev
->rbd_client
->client
));
2388 static ssize_t
rbd_pool_show(struct device
*dev
,
2389 struct device_attribute
*attr
, char *buf
)
2391 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2393 return sprintf(buf
, "%s\n", rbd_dev
->spec
->pool_name
);
2396 static ssize_t
rbd_pool_id_show(struct device
*dev
,
2397 struct device_attribute
*attr
, char *buf
)
2399 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2401 return sprintf(buf
, "%llu\n",
2402 (unsigned long long) rbd_dev
->spec
->pool_id
);
2405 static ssize_t
rbd_name_show(struct device
*dev
,
2406 struct device_attribute
*attr
, char *buf
)
2408 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2410 if (rbd_dev
->spec
->image_name
)
2411 return sprintf(buf
, "%s\n", rbd_dev
->spec
->image_name
);
2413 return sprintf(buf
, "(unknown)\n");
2416 static ssize_t
rbd_image_id_show(struct device
*dev
,
2417 struct device_attribute
*attr
, char *buf
)
2419 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2421 return sprintf(buf
, "%s\n", rbd_dev
->spec
->image_id
);
2425 * Shows the name of the currently-mapped snapshot (or
2426 * RBD_SNAP_HEAD_NAME for the base image).
2428 static ssize_t
rbd_snap_show(struct device
*dev
,
2429 struct device_attribute
*attr
,
2432 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2434 return sprintf(buf
, "%s\n", rbd_dev
->spec
->snap_name
);
2438 * For an rbd v2 image, shows the pool id, image id, and snapshot id
2439 * for the parent image. If there is no parent, simply shows
2440 * "(no parent image)".
2442 static ssize_t
rbd_parent_show(struct device
*dev
,
2443 struct device_attribute
*attr
,
2446 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2447 struct rbd_spec
*spec
= rbd_dev
->parent_spec
;
2452 return sprintf(buf
, "(no parent image)\n");
2454 count
= sprintf(bufp
, "pool_id %llu\npool_name %s\n",
2455 (unsigned long long) spec
->pool_id
, spec
->pool_name
);
2460 count
= sprintf(bufp
, "image_id %s\nimage_name %s\n", spec
->image_id
,
2461 spec
->image_name
? spec
->image_name
: "(unknown)");
2466 count
= sprintf(bufp
, "snap_id %llu\nsnap_name %s\n",
2467 (unsigned long long) spec
->snap_id
, spec
->snap_name
);
2472 count
= sprintf(bufp
, "overlap %llu\n", rbd_dev
->parent_overlap
);
2477 return (ssize_t
) (bufp
- buf
);
2480 static ssize_t
rbd_image_refresh(struct device
*dev
,
2481 struct device_attribute
*attr
,
2485 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2488 ret
= rbd_dev_refresh(rbd_dev
, NULL
);
2490 return ret
< 0 ? ret
: size
;
2493 static DEVICE_ATTR(size
, S_IRUGO
, rbd_size_show
, NULL
);
2494 static DEVICE_ATTR(features
, S_IRUGO
, rbd_features_show
, NULL
);
2495 static DEVICE_ATTR(major
, S_IRUGO
, rbd_major_show
, NULL
);
2496 static DEVICE_ATTR(client_id
, S_IRUGO
, rbd_client_id_show
, NULL
);
2497 static DEVICE_ATTR(pool
, S_IRUGO
, rbd_pool_show
, NULL
);
2498 static DEVICE_ATTR(pool_id
, S_IRUGO
, rbd_pool_id_show
, NULL
);
2499 static DEVICE_ATTR(name
, S_IRUGO
, rbd_name_show
, NULL
);
2500 static DEVICE_ATTR(image_id
, S_IRUGO
, rbd_image_id_show
, NULL
);
2501 static DEVICE_ATTR(refresh
, S_IWUSR
, NULL
, rbd_image_refresh
);
2502 static DEVICE_ATTR(current_snap
, S_IRUGO
, rbd_snap_show
, NULL
);
2503 static DEVICE_ATTR(parent
, S_IRUGO
, rbd_parent_show
, NULL
);
2505 static struct attribute
*rbd_attrs
[] = {
2506 &dev_attr_size
.attr
,
2507 &dev_attr_features
.attr
,
2508 &dev_attr_major
.attr
,
2509 &dev_attr_client_id
.attr
,
2510 &dev_attr_pool
.attr
,
2511 &dev_attr_pool_id
.attr
,
2512 &dev_attr_name
.attr
,
2513 &dev_attr_image_id
.attr
,
2514 &dev_attr_current_snap
.attr
,
2515 &dev_attr_parent
.attr
,
2516 &dev_attr_refresh
.attr
,
2520 static struct attribute_group rbd_attr_group
= {
2524 static const struct attribute_group
*rbd_attr_groups
[] = {
2529 static void rbd_sysfs_dev_release(struct device
*dev
)
2533 static struct device_type rbd_device_type
= {
2535 .groups
= rbd_attr_groups
,
2536 .release
= rbd_sysfs_dev_release
,
2544 static ssize_t
rbd_snap_size_show(struct device
*dev
,
2545 struct device_attribute
*attr
,
2548 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2550 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->size
);
2553 static ssize_t
rbd_snap_id_show(struct device
*dev
,
2554 struct device_attribute
*attr
,
2557 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2559 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->id
);
2562 static ssize_t
rbd_snap_features_show(struct device
*dev
,
2563 struct device_attribute
*attr
,
2566 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2568 return sprintf(buf
, "0x%016llx\n",
2569 (unsigned long long) snap
->features
);
2572 static DEVICE_ATTR(snap_size
, S_IRUGO
, rbd_snap_size_show
, NULL
);
2573 static DEVICE_ATTR(snap_id
, S_IRUGO
, rbd_snap_id_show
, NULL
);
2574 static DEVICE_ATTR(snap_features
, S_IRUGO
, rbd_snap_features_show
, NULL
);
2576 static struct attribute
*rbd_snap_attrs
[] = {
2577 &dev_attr_snap_size
.attr
,
2578 &dev_attr_snap_id
.attr
,
2579 &dev_attr_snap_features
.attr
,
2583 static struct attribute_group rbd_snap_attr_group
= {
2584 .attrs
= rbd_snap_attrs
,
2587 static void rbd_snap_dev_release(struct device
*dev
)
2589 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2594 static const struct attribute_group
*rbd_snap_attr_groups
[] = {
2595 &rbd_snap_attr_group
,
2599 static struct device_type rbd_snap_device_type
= {
2600 .groups
= rbd_snap_attr_groups
,
2601 .release
= rbd_snap_dev_release
,
2604 static struct rbd_spec
*rbd_spec_get(struct rbd_spec
*spec
)
2606 kref_get(&spec
->kref
);
2611 static void rbd_spec_free(struct kref
*kref
);
2612 static void rbd_spec_put(struct rbd_spec
*spec
)
2615 kref_put(&spec
->kref
, rbd_spec_free
);
2618 static struct rbd_spec
*rbd_spec_alloc(void)
2620 struct rbd_spec
*spec
;
2622 spec
= kzalloc(sizeof (*spec
), GFP_KERNEL
);
2625 kref_init(&spec
->kref
);
2627 rbd_spec_put(rbd_spec_get(spec
)); /* TEMPORARY */
2632 static void rbd_spec_free(struct kref
*kref
)
2634 struct rbd_spec
*spec
= container_of(kref
, struct rbd_spec
, kref
);
2636 kfree(spec
->pool_name
);
2637 kfree(spec
->image_id
);
2638 kfree(spec
->image_name
);
2639 kfree(spec
->snap_name
);
2643 struct rbd_device
*rbd_dev_create(struct rbd_client
*rbdc
,
2644 struct rbd_spec
*spec
)
2646 struct rbd_device
*rbd_dev
;
2648 rbd_dev
= kzalloc(sizeof (*rbd_dev
), GFP_KERNEL
);
2652 spin_lock_init(&rbd_dev
->lock
);
2654 INIT_LIST_HEAD(&rbd_dev
->node
);
2655 INIT_LIST_HEAD(&rbd_dev
->snaps
);
2656 init_rwsem(&rbd_dev
->header_rwsem
);
2658 rbd_dev
->spec
= spec
;
2659 rbd_dev
->rbd_client
= rbdc
;
2661 /* Initialize the layout used for all rbd requests */
2663 rbd_dev
->layout
.fl_stripe_unit
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
2664 rbd_dev
->layout
.fl_stripe_count
= cpu_to_le32(1);
2665 rbd_dev
->layout
.fl_object_size
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
2666 rbd_dev
->layout
.fl_pg_pool
= cpu_to_le32((u32
) spec
->pool_id
);
2671 static void rbd_dev_destroy(struct rbd_device
*rbd_dev
)
2673 rbd_spec_put(rbd_dev
->parent_spec
);
2674 kfree(rbd_dev
->header_name
);
2675 rbd_put_client(rbd_dev
->rbd_client
);
2676 rbd_spec_put(rbd_dev
->spec
);
2680 static bool rbd_snap_registered(struct rbd_snap
*snap
)
2682 bool ret
= snap
->dev
.type
== &rbd_snap_device_type
;
2683 bool reg
= device_is_registered(&snap
->dev
);
2685 rbd_assert(!ret
^ reg
);
2690 static void rbd_remove_snap_dev(struct rbd_snap
*snap
)
2692 list_del(&snap
->node
);
2693 if (device_is_registered(&snap
->dev
))
2694 device_unregister(&snap
->dev
);
2697 static int rbd_register_snap_dev(struct rbd_snap
*snap
,
2698 struct device
*parent
)
2700 struct device
*dev
= &snap
->dev
;
2703 dev
->type
= &rbd_snap_device_type
;
2704 dev
->parent
= parent
;
2705 dev
->release
= rbd_snap_dev_release
;
2706 dev_set_name(dev
, "%s%s", RBD_SNAP_DEV_NAME_PREFIX
, snap
->name
);
2707 dout("%s: registering device for snapshot %s\n", __func__
, snap
->name
);
2709 ret
= device_register(dev
);
2714 static struct rbd_snap
*__rbd_add_snap_dev(struct rbd_device
*rbd_dev
,
2715 const char *snap_name
,
2716 u64 snap_id
, u64 snap_size
,
2719 struct rbd_snap
*snap
;
2722 snap
= kzalloc(sizeof (*snap
), GFP_KERNEL
);
2724 return ERR_PTR(-ENOMEM
);
2727 snap
->name
= kstrdup(snap_name
, GFP_KERNEL
);
2732 snap
->size
= snap_size
;
2733 snap
->features
= snap_features
;
2741 return ERR_PTR(ret
);
2744 static char *rbd_dev_v1_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
2745 u64
*snap_size
, u64
*snap_features
)
2749 rbd_assert(which
< rbd_dev
->header
.snapc
->num_snaps
);
2751 *snap_size
= rbd_dev
->header
.snap_sizes
[which
];
2752 *snap_features
= 0; /* No features for v1 */
2754 /* Skip over names until we find the one we are looking for */
2756 snap_name
= rbd_dev
->header
.snap_names
;
2758 snap_name
+= strlen(snap_name
) + 1;
2764 * Get the size and object order for an image snapshot, or if
2765 * snap_id is CEPH_NOSNAP, gets this information for the base
2768 static int _rbd_dev_v2_snap_size(struct rbd_device
*rbd_dev
, u64 snap_id
,
2769 u8
*order
, u64
*snap_size
)
2771 __le64 snapid
= cpu_to_le64(snap_id
);
2776 } __attribute__ ((packed
)) size_buf
= { 0 };
2778 ret
= rbd_obj_method_sync(rbd_dev
, rbd_dev
->header_name
,
2780 (char *) &snapid
, sizeof (snapid
),
2781 (char *) &size_buf
, sizeof (size_buf
), NULL
);
2782 dout("%s: rbd_obj_method_sync returned %d\n", __func__
, ret
);
2786 *order
= size_buf
.order
;
2787 *snap_size
= le64_to_cpu(size_buf
.size
);
2789 dout(" snap_id 0x%016llx order = %u, snap_size = %llu\n",
2790 (unsigned long long) snap_id
, (unsigned int) *order
,
2791 (unsigned long long) *snap_size
);
2796 static int rbd_dev_v2_image_size(struct rbd_device
*rbd_dev
)
2798 return _rbd_dev_v2_snap_size(rbd_dev
, CEPH_NOSNAP
,
2799 &rbd_dev
->header
.obj_order
,
2800 &rbd_dev
->header
.image_size
);
2803 static int rbd_dev_v2_object_prefix(struct rbd_device
*rbd_dev
)
2809 reply_buf
= kzalloc(RBD_OBJ_PREFIX_LEN_MAX
, GFP_KERNEL
);
2813 ret
= rbd_obj_method_sync(rbd_dev
, rbd_dev
->header_name
,
2814 "rbd", "get_object_prefix",
2816 reply_buf
, RBD_OBJ_PREFIX_LEN_MAX
, NULL
);
2817 dout("%s: rbd_obj_method_sync returned %d\n", __func__
, ret
);
2822 rbd_dev
->header
.object_prefix
= ceph_extract_encoded_string(&p
,
2823 p
+ RBD_OBJ_PREFIX_LEN_MAX
,
2826 if (IS_ERR(rbd_dev
->header
.object_prefix
)) {
2827 ret
= PTR_ERR(rbd_dev
->header
.object_prefix
);
2828 rbd_dev
->header
.object_prefix
= NULL
;
2830 dout(" object_prefix = %s\n", rbd_dev
->header
.object_prefix
);
2839 static int _rbd_dev_v2_snap_features(struct rbd_device
*rbd_dev
, u64 snap_id
,
2842 __le64 snapid
= cpu_to_le64(snap_id
);
2846 } features_buf
= { 0 };
2850 ret
= rbd_obj_method_sync(rbd_dev
, rbd_dev
->header_name
,
2851 "rbd", "get_features",
2852 (char *) &snapid
, sizeof (snapid
),
2853 (char *) &features_buf
, sizeof (features_buf
),
2855 dout("%s: rbd_obj_method_sync returned %d\n", __func__
, ret
);
2859 incompat
= le64_to_cpu(features_buf
.incompat
);
2860 if (incompat
& ~RBD_FEATURES_ALL
)
2863 *snap_features
= le64_to_cpu(features_buf
.features
);
2865 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
2866 (unsigned long long) snap_id
,
2867 (unsigned long long) *snap_features
,
2868 (unsigned long long) le64_to_cpu(features_buf
.incompat
));
2873 static int rbd_dev_v2_features(struct rbd_device
*rbd_dev
)
2875 return _rbd_dev_v2_snap_features(rbd_dev
, CEPH_NOSNAP
,
2876 &rbd_dev
->header
.features
);
2879 static int rbd_dev_v2_parent_info(struct rbd_device
*rbd_dev
)
2881 struct rbd_spec
*parent_spec
;
2883 void *reply_buf
= NULL
;
2891 parent_spec
= rbd_spec_alloc();
2895 size
= sizeof (__le64
) + /* pool_id */
2896 sizeof (__le32
) + RBD_IMAGE_ID_LEN_MAX
+ /* image_id */
2897 sizeof (__le64
) + /* snap_id */
2898 sizeof (__le64
); /* overlap */
2899 reply_buf
= kmalloc(size
, GFP_KERNEL
);
2905 snapid
= cpu_to_le64(CEPH_NOSNAP
);
2906 ret
= rbd_obj_method_sync(rbd_dev
, rbd_dev
->header_name
,
2907 "rbd", "get_parent",
2908 (char *) &snapid
, sizeof (snapid
),
2909 (char *) reply_buf
, size
, NULL
);
2910 dout("%s: rbd_obj_method_sync returned %d\n", __func__
, ret
);
2916 end
= (char *) reply_buf
+ size
;
2917 ceph_decode_64_safe(&p
, end
, parent_spec
->pool_id
, out_err
);
2918 if (parent_spec
->pool_id
== CEPH_NOPOOL
)
2919 goto out
; /* No parent? No problem. */
2921 /* The ceph file layout needs to fit pool id in 32 bits */
2924 if (WARN_ON(parent_spec
->pool_id
> (u64
) U32_MAX
))
2927 image_id
= ceph_extract_encoded_string(&p
, end
, NULL
, GFP_KERNEL
);
2928 if (IS_ERR(image_id
)) {
2929 ret
= PTR_ERR(image_id
);
2932 parent_spec
->image_id
= image_id
;
2933 ceph_decode_64_safe(&p
, end
, parent_spec
->snap_id
, out_err
);
2934 ceph_decode_64_safe(&p
, end
, overlap
, out_err
);
2936 rbd_dev
->parent_overlap
= overlap
;
2937 rbd_dev
->parent_spec
= parent_spec
;
2938 parent_spec
= NULL
; /* rbd_dev now owns this */
2943 rbd_spec_put(parent_spec
);
2948 static char *rbd_dev_image_name(struct rbd_device
*rbd_dev
)
2950 size_t image_id_size
;
2955 void *reply_buf
= NULL
;
2957 char *image_name
= NULL
;
2960 rbd_assert(!rbd_dev
->spec
->image_name
);
2962 len
= strlen(rbd_dev
->spec
->image_id
);
2963 image_id_size
= sizeof (__le32
) + len
;
2964 image_id
= kmalloc(image_id_size
, GFP_KERNEL
);
2969 end
= (char *) image_id
+ image_id_size
;
2970 ceph_encode_string(&p
, end
, rbd_dev
->spec
->image_id
, (u32
) len
);
2972 size
= sizeof (__le32
) + RBD_IMAGE_NAME_LEN_MAX
;
2973 reply_buf
= kmalloc(size
, GFP_KERNEL
);
2977 ret
= rbd_obj_method_sync(rbd_dev
, RBD_DIRECTORY
,
2978 "rbd", "dir_get_name",
2979 image_id
, image_id_size
,
2980 (char *) reply_buf
, size
, NULL
);
2984 end
= (char *) reply_buf
+ size
;
2985 image_name
= ceph_extract_encoded_string(&p
, end
, &len
, GFP_KERNEL
);
2986 if (IS_ERR(image_name
))
2989 dout("%s: name is %s len is %zd\n", __func__
, image_name
, len
);
2998 * When a parent image gets probed, we only have the pool, image,
2999 * and snapshot ids but not the names of any of them. This call
3000 * is made later to fill in those names. It has to be done after
3001 * rbd_dev_snaps_update() has completed because some of the
3002 * information (in particular, snapshot name) is not available
3005 static int rbd_dev_probe_update_spec(struct rbd_device
*rbd_dev
)
3007 struct ceph_osd_client
*osdc
;
3009 void *reply_buf
= NULL
;
3012 if (rbd_dev
->spec
->pool_name
)
3013 return 0; /* Already have the names */
3015 /* Look up the pool name */
3017 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
3018 name
= ceph_pg_pool_name_by_id(osdc
->osdmap
, rbd_dev
->spec
->pool_id
);
3020 rbd_warn(rbd_dev
, "there is no pool with id %llu",
3021 rbd_dev
->spec
->pool_id
); /* Really a BUG() */
3025 rbd_dev
->spec
->pool_name
= kstrdup(name
, GFP_KERNEL
);
3026 if (!rbd_dev
->spec
->pool_name
)
3029 /* Fetch the image name; tolerate failure here */
3031 name
= rbd_dev_image_name(rbd_dev
);
3033 rbd_dev
->spec
->image_name
= (char *) name
;
3035 rbd_warn(rbd_dev
, "unable to get image name");
3037 /* Look up the snapshot name. */
3039 name
= rbd_snap_name(rbd_dev
, rbd_dev
->spec
->snap_id
);
3041 rbd_warn(rbd_dev
, "no snapshot with id %llu",
3042 rbd_dev
->spec
->snap_id
); /* Really a BUG() */
3046 rbd_dev
->spec
->snap_name
= kstrdup(name
, GFP_KERNEL
);
3047 if(!rbd_dev
->spec
->snap_name
)
3053 kfree(rbd_dev
->spec
->pool_name
);
3054 rbd_dev
->spec
->pool_name
= NULL
;
3059 static int rbd_dev_v2_snap_context(struct rbd_device
*rbd_dev
, u64
*ver
)
3068 struct ceph_snap_context
*snapc
;
3072 * We'll need room for the seq value (maximum snapshot id),
3073 * snapshot count, and array of that many snapshot ids.
3074 * For now we have a fixed upper limit on the number we're
3075 * prepared to receive.
3077 size
= sizeof (__le64
) + sizeof (__le32
) +
3078 RBD_MAX_SNAP_COUNT
* sizeof (__le64
);
3079 reply_buf
= kzalloc(size
, GFP_KERNEL
);
3083 ret
= rbd_obj_method_sync(rbd_dev
, rbd_dev
->header_name
,
3084 "rbd", "get_snapcontext",
3086 reply_buf
, size
, ver
);
3087 dout("%s: rbd_obj_method_sync returned %d\n", __func__
, ret
);
3093 end
= (char *) reply_buf
+ size
;
3094 ceph_decode_64_safe(&p
, end
, seq
, out
);
3095 ceph_decode_32_safe(&p
, end
, snap_count
, out
);
3098 * Make sure the reported number of snapshot ids wouldn't go
3099 * beyond the end of our buffer. But before checking that,
3100 * make sure the computed size of the snapshot context we
3101 * allocate is representable in a size_t.
3103 if (snap_count
> (SIZE_MAX
- sizeof (struct ceph_snap_context
))
3108 if (!ceph_has_room(&p
, end
, snap_count
* sizeof (__le64
)))
3111 size
= sizeof (struct ceph_snap_context
) +
3112 snap_count
* sizeof (snapc
->snaps
[0]);
3113 snapc
= kmalloc(size
, GFP_KERNEL
);
3119 atomic_set(&snapc
->nref
, 1);
3121 snapc
->num_snaps
= snap_count
;
3122 for (i
= 0; i
< snap_count
; i
++)
3123 snapc
->snaps
[i
] = ceph_decode_64(&p
);
3125 rbd_dev
->header
.snapc
= snapc
;
3127 dout(" snap context seq = %llu, snap_count = %u\n",
3128 (unsigned long long) seq
, (unsigned int) snap_count
);
3136 static char *rbd_dev_v2_snap_name(struct rbd_device
*rbd_dev
, u32 which
)
3146 size
= sizeof (__le32
) + RBD_MAX_SNAP_NAME_LEN
;
3147 reply_buf
= kmalloc(size
, GFP_KERNEL
);
3149 return ERR_PTR(-ENOMEM
);
3151 snap_id
= cpu_to_le64(rbd_dev
->header
.snapc
->snaps
[which
]);
3152 ret
= rbd_obj_method_sync(rbd_dev
, rbd_dev
->header_name
,
3153 "rbd", "get_snapshot_name",
3154 (char *) &snap_id
, sizeof (snap_id
),
3155 reply_buf
, size
, NULL
);
3156 dout("%s: rbd_obj_method_sync returned %d\n", __func__
, ret
);
3161 end
= (char *) reply_buf
+ size
;
3162 snap_name
= ceph_extract_encoded_string(&p
, end
, NULL
, GFP_KERNEL
);
3163 if (IS_ERR(snap_name
)) {
3164 ret
= PTR_ERR(snap_name
);
3167 dout(" snap_id 0x%016llx snap_name = %s\n",
3168 (unsigned long long) le64_to_cpu(snap_id
), snap_name
);
3176 return ERR_PTR(ret
);
3179 static char *rbd_dev_v2_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
3180 u64
*snap_size
, u64
*snap_features
)
3186 snap_id
= rbd_dev
->header
.snapc
->snaps
[which
];
3187 ret
= _rbd_dev_v2_snap_size(rbd_dev
, snap_id
, &order
, snap_size
);
3189 return ERR_PTR(ret
);
3190 ret
= _rbd_dev_v2_snap_features(rbd_dev
, snap_id
, snap_features
);
3192 return ERR_PTR(ret
);
3194 return rbd_dev_v2_snap_name(rbd_dev
, which
);
3197 static char *rbd_dev_snap_info(struct rbd_device
*rbd_dev
, u32 which
,
3198 u64
*snap_size
, u64
*snap_features
)
3200 if (rbd_dev
->image_format
== 1)
3201 return rbd_dev_v1_snap_info(rbd_dev
, which
,
3202 snap_size
, snap_features
);
3203 if (rbd_dev
->image_format
== 2)
3204 return rbd_dev_v2_snap_info(rbd_dev
, which
,
3205 snap_size
, snap_features
);
3206 return ERR_PTR(-EINVAL
);
3209 static int rbd_dev_v2_refresh(struct rbd_device
*rbd_dev
, u64
*hver
)
3214 down_write(&rbd_dev
->header_rwsem
);
3216 /* Grab old order first, to see if it changes */
3218 obj_order
= rbd_dev
->header
.obj_order
,
3219 ret
= rbd_dev_v2_image_size(rbd_dev
);
3222 if (rbd_dev
->header
.obj_order
!= obj_order
) {
3226 rbd_update_mapping_size(rbd_dev
);
3228 ret
= rbd_dev_v2_snap_context(rbd_dev
, hver
);
3229 dout("rbd_dev_v2_snap_context returned %d\n", ret
);
3232 ret
= rbd_dev_snaps_update(rbd_dev
);
3233 dout("rbd_dev_snaps_update returned %d\n", ret
);
3236 ret
= rbd_dev_snaps_register(rbd_dev
);
3237 dout("rbd_dev_snaps_register returned %d\n", ret
);
3239 up_write(&rbd_dev
->header_rwsem
);
3245 * Scan the rbd device's current snapshot list and compare it to the
3246 * newly-received snapshot context. Remove any existing snapshots
3247 * not present in the new snapshot context. Add a new snapshot for
3248 * any snaphots in the snapshot context not in the current list.
3249 * And verify there are no changes to snapshots we already know
3252 * Assumes the snapshots in the snapshot context are sorted by
3253 * snapshot id, highest id first. (Snapshots in the rbd_dev's list
3254 * are also maintained in that order.)
3256 static int rbd_dev_snaps_update(struct rbd_device
*rbd_dev
)
3258 struct ceph_snap_context
*snapc
= rbd_dev
->header
.snapc
;
3259 const u32 snap_count
= snapc
->num_snaps
;
3260 struct list_head
*head
= &rbd_dev
->snaps
;
3261 struct list_head
*links
= head
->next
;
3264 dout("%s: snap count is %u\n", __func__
, (unsigned int) snap_count
);
3265 while (index
< snap_count
|| links
!= head
) {
3267 struct rbd_snap
*snap
;
3270 u64 snap_features
= 0;
3272 snap_id
= index
< snap_count
? snapc
->snaps
[index
]
3274 snap
= links
!= head
? list_entry(links
, struct rbd_snap
, node
)
3276 rbd_assert(!snap
|| snap
->id
!= CEPH_NOSNAP
);
3278 if (snap_id
== CEPH_NOSNAP
|| (snap
&& snap
->id
> snap_id
)) {
3279 struct list_head
*next
= links
->next
;
3282 * A previously-existing snapshot is not in
3283 * the new snap context.
3285 * If the now missing snapshot is the one the
3286 * image is mapped to, clear its exists flag
3287 * so we can avoid sending any more requests
3290 if (rbd_dev
->spec
->snap_id
== snap
->id
)
3291 clear_bit(RBD_DEV_FLAG_EXISTS
, &rbd_dev
->flags
);
3292 rbd_remove_snap_dev(snap
);
3293 dout("%ssnap id %llu has been removed\n",
3294 rbd_dev
->spec
->snap_id
== snap
->id
?
3296 (unsigned long long) snap
->id
);
3298 /* Done with this list entry; advance */
3304 snap_name
= rbd_dev_snap_info(rbd_dev
, index
,
3305 &snap_size
, &snap_features
);
3306 if (IS_ERR(snap_name
))
3307 return PTR_ERR(snap_name
);
3309 dout("entry %u: snap_id = %llu\n", (unsigned int) snap_count
,
3310 (unsigned long long) snap_id
);
3311 if (!snap
|| (snap_id
!= CEPH_NOSNAP
&& snap
->id
< snap_id
)) {
3312 struct rbd_snap
*new_snap
;
3314 /* We haven't seen this snapshot before */
3316 new_snap
= __rbd_add_snap_dev(rbd_dev
, snap_name
,
3317 snap_id
, snap_size
, snap_features
);
3318 if (IS_ERR(new_snap
)) {
3319 int err
= PTR_ERR(new_snap
);
3321 dout(" failed to add dev, error %d\n", err
);
3326 /* New goes before existing, or at end of list */
3328 dout(" added dev%s\n", snap
? "" : " at end\n");
3330 list_add_tail(&new_snap
->node
, &snap
->node
);
3332 list_add_tail(&new_snap
->node
, head
);
3334 /* Already have this one */
3336 dout(" already present\n");
3338 rbd_assert(snap
->size
== snap_size
);
3339 rbd_assert(!strcmp(snap
->name
, snap_name
));
3340 rbd_assert(snap
->features
== snap_features
);
3342 /* Done with this list entry; advance */
3344 links
= links
->next
;
3347 /* Advance to the next entry in the snapshot context */
3351 dout("%s: done\n", __func__
);
3357 * Scan the list of snapshots and register the devices for any that
3358 * have not already been registered.
3360 static int rbd_dev_snaps_register(struct rbd_device
*rbd_dev
)
3362 struct rbd_snap
*snap
;
3365 dout("%s called\n", __func__
);
3366 if (WARN_ON(!device_is_registered(&rbd_dev
->dev
)))
3369 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
3370 if (!rbd_snap_registered(snap
)) {
3371 ret
= rbd_register_snap_dev(snap
, &rbd_dev
->dev
);
3376 dout("%s: returning %d\n", __func__
, ret
);
3381 static int rbd_bus_add_dev(struct rbd_device
*rbd_dev
)
3386 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
3388 dev
= &rbd_dev
->dev
;
3389 dev
->bus
= &rbd_bus_type
;
3390 dev
->type
= &rbd_device_type
;
3391 dev
->parent
= &rbd_root_dev
;
3392 dev
->release
= rbd_dev_release
;
3393 dev_set_name(dev
, "%d", rbd_dev
->dev_id
);
3394 ret
= device_register(dev
);
3396 mutex_unlock(&ctl_mutex
);
3401 static void rbd_bus_del_dev(struct rbd_device
*rbd_dev
)
3403 device_unregister(&rbd_dev
->dev
);
3406 static atomic64_t rbd_dev_id_max
= ATOMIC64_INIT(0);
3409 * Get a unique rbd identifier for the given new rbd_dev, and add
3410 * the rbd_dev to the global list. The minimum rbd id is 1.
3412 static void rbd_dev_id_get(struct rbd_device
*rbd_dev
)
3414 rbd_dev
->dev_id
= atomic64_inc_return(&rbd_dev_id_max
);
3416 spin_lock(&rbd_dev_list_lock
);
3417 list_add_tail(&rbd_dev
->node
, &rbd_dev_list
);
3418 spin_unlock(&rbd_dev_list_lock
);
3419 dout("rbd_dev %p given dev id %llu\n", rbd_dev
,
3420 (unsigned long long) rbd_dev
->dev_id
);
3424 * Remove an rbd_dev from the global list, and record that its
3425 * identifier is no longer in use.
3427 static void rbd_dev_id_put(struct rbd_device
*rbd_dev
)
3429 struct list_head
*tmp
;
3430 int rbd_id
= rbd_dev
->dev_id
;
3433 rbd_assert(rbd_id
> 0);
3435 dout("rbd_dev %p released dev id %llu\n", rbd_dev
,
3436 (unsigned long long) rbd_dev
->dev_id
);
3437 spin_lock(&rbd_dev_list_lock
);
3438 list_del_init(&rbd_dev
->node
);
3441 * If the id being "put" is not the current maximum, there
3442 * is nothing special we need to do.
3444 if (rbd_id
!= atomic64_read(&rbd_dev_id_max
)) {
3445 spin_unlock(&rbd_dev_list_lock
);
3450 * We need to update the current maximum id. Search the
3451 * list to find out what it is. We're more likely to find
3452 * the maximum at the end, so search the list backward.
3455 list_for_each_prev(tmp
, &rbd_dev_list
) {
3456 struct rbd_device
*rbd_dev
;
3458 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
3459 if (rbd_dev
->dev_id
> max_id
)
3460 max_id
= rbd_dev
->dev_id
;
3462 spin_unlock(&rbd_dev_list_lock
);
3465 * The max id could have been updated by rbd_dev_id_get(), in
3466 * which case it now accurately reflects the new maximum.
3467 * Be careful not to overwrite the maximum value in that
3470 atomic64_cmpxchg(&rbd_dev_id_max
, rbd_id
, max_id
);
3471 dout(" max dev id has been reset\n");
3475 * Skips over white space at *buf, and updates *buf to point to the
3476 * first found non-space character (if any). Returns the length of
3477 * the token (string of non-white space characters) found. Note
3478 * that *buf must be terminated with '\0'.
3480 static inline size_t next_token(const char **buf
)
3483 * These are the characters that produce nonzero for
3484 * isspace() in the "C" and "POSIX" locales.
3486 const char *spaces
= " \f\n\r\t\v";
3488 *buf
+= strspn(*buf
, spaces
); /* Find start of token */
3490 return strcspn(*buf
, spaces
); /* Return token length */
3494 * Finds the next token in *buf, and if the provided token buffer is
3495 * big enough, copies the found token into it. The result, if
3496 * copied, is guaranteed to be terminated with '\0'. Note that *buf
3497 * must be terminated with '\0' on entry.
3499 * Returns the length of the token found (not including the '\0').
3500 * Return value will be 0 if no token is found, and it will be >=
3501 * token_size if the token would not fit.
3503 * The *buf pointer will be updated to point beyond the end of the
3504 * found token. Note that this occurs even if the token buffer is
3505 * too small to hold it.
3507 static inline size_t copy_token(const char **buf
,
3513 len
= next_token(buf
);
3514 if (len
< token_size
) {
3515 memcpy(token
, *buf
, len
);
3516 *(token
+ len
) = '\0';
3524 * Finds the next token in *buf, dynamically allocates a buffer big
3525 * enough to hold a copy of it, and copies the token into the new
3526 * buffer. The copy is guaranteed to be terminated with '\0'. Note
3527 * that a duplicate buffer is created even for a zero-length token.
3529 * Returns a pointer to the newly-allocated duplicate, or a null
3530 * pointer if memory for the duplicate was not available. If
3531 * the lenp argument is a non-null pointer, the length of the token
3532 * (not including the '\0') is returned in *lenp.
3534 * If successful, the *buf pointer will be updated to point beyond
3535 * the end of the found token.
3537 * Note: uses GFP_KERNEL for allocation.
3539 static inline char *dup_token(const char **buf
, size_t *lenp
)
3544 len
= next_token(buf
);
3545 dup
= kmemdup(*buf
, len
+ 1, GFP_KERNEL
);
3548 *(dup
+ len
) = '\0';
3558 * Parse the options provided for an "rbd add" (i.e., rbd image
3559 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
3560 * and the data written is passed here via a NUL-terminated buffer.
3561 * Returns 0 if successful or an error code otherwise.
3563 * The information extracted from these options is recorded in
3564 * the other parameters which return dynamically-allocated
3567 * The address of a pointer that will refer to a ceph options
3568 * structure. Caller must release the returned pointer using
3569 * ceph_destroy_options() when it is no longer needed.
3571 * Address of an rbd options pointer. Fully initialized by
3572 * this function; caller must release with kfree().
3574 * Address of an rbd image specification pointer. Fully
3575 * initialized by this function based on parsed options.
3576 * Caller must release with rbd_spec_put().
3578 * The options passed take this form:
3579 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
3582 * A comma-separated list of one or more monitor addresses.
3583 * A monitor address is an ip address, optionally followed
3584 * by a port number (separated by a colon).
3585 * I.e.: ip1[:port1][,ip2[:port2]...]
3587 * A comma-separated list of ceph and/or rbd options.
3589 * The name of the rados pool containing the rbd image.
3591 * The name of the image in that pool to map.
3593 * An optional snapshot id. If provided, the mapping will
3594 * present data from the image at the time that snapshot was
3595 * created. The image head is used if no snapshot id is
3596 * provided. Snapshot mappings are always read-only.
3598 static int rbd_add_parse_args(const char *buf
,
3599 struct ceph_options
**ceph_opts
,
3600 struct rbd_options
**opts
,
3601 struct rbd_spec
**rbd_spec
)
3605 const char *mon_addrs
;
3606 size_t mon_addrs_size
;
3607 struct rbd_spec
*spec
= NULL
;
3608 struct rbd_options
*rbd_opts
= NULL
;
3609 struct ceph_options
*copts
;
3612 /* The first four tokens are required */
3614 len
= next_token(&buf
);
3616 rbd_warn(NULL
, "no monitor address(es) provided");
3620 mon_addrs_size
= len
+ 1;
3624 options
= dup_token(&buf
, NULL
);
3628 rbd_warn(NULL
, "no options provided");
3632 spec
= rbd_spec_alloc();
3636 spec
->pool_name
= dup_token(&buf
, NULL
);
3637 if (!spec
->pool_name
)
3639 if (!*spec
->pool_name
) {
3640 rbd_warn(NULL
, "no pool name provided");
3644 spec
->image_name
= dup_token(&buf
, NULL
);
3645 if (!spec
->image_name
)
3647 if (!*spec
->image_name
) {
3648 rbd_warn(NULL
, "no image name provided");
3653 * Snapshot name is optional; default is to use "-"
3654 * (indicating the head/no snapshot).
3656 len
= next_token(&buf
);
3658 buf
= RBD_SNAP_HEAD_NAME
; /* No snapshot supplied */
3659 len
= sizeof (RBD_SNAP_HEAD_NAME
) - 1;
3660 } else if (len
> RBD_MAX_SNAP_NAME_LEN
) {
3661 ret
= -ENAMETOOLONG
;
3664 spec
->snap_name
= kmemdup(buf
, len
+ 1, GFP_KERNEL
);
3665 if (!spec
->snap_name
)
3667 *(spec
->snap_name
+ len
) = '\0';
3669 /* Initialize all rbd options to the defaults */
3671 rbd_opts
= kzalloc(sizeof (*rbd_opts
), GFP_KERNEL
);
3675 rbd_opts
->read_only
= RBD_READ_ONLY_DEFAULT
;
3677 copts
= ceph_parse_options(options
, mon_addrs
,
3678 mon_addrs
+ mon_addrs_size
- 1,
3679 parse_rbd_opts_token
, rbd_opts
);
3680 if (IS_ERR(copts
)) {
3681 ret
= PTR_ERR(copts
);
3702 * An rbd format 2 image has a unique identifier, distinct from the
3703 * name given to it by the user. Internally, that identifier is
3704 * what's used to specify the names of objects related to the image.
3706 * A special "rbd id" object is used to map an rbd image name to its
3707 * id. If that object doesn't exist, then there is no v2 rbd image
3708 * with the supplied name.
3710 * This function will record the given rbd_dev's image_id field if
3711 * it can be determined, and in that case will return 0. If any
3712 * errors occur a negative errno will be returned and the rbd_dev's
3713 * image_id field will be unchanged (and should be NULL).
3715 static int rbd_dev_image_id(struct rbd_device
*rbd_dev
)
3724 * When probing a parent image, the image id is already
3725 * known (and the image name likely is not). There's no
3726 * need to fetch the image id again in this case.
3728 if (rbd_dev
->spec
->image_id
)
3732 * First, see if the format 2 image id file exists, and if
3733 * so, get the image's persistent id from it.
3735 size
= sizeof (RBD_ID_PREFIX
) + strlen(rbd_dev
->spec
->image_name
);
3736 object_name
= kmalloc(size
, GFP_NOIO
);
3739 sprintf(object_name
, "%s%s", RBD_ID_PREFIX
, rbd_dev
->spec
->image_name
);
3740 dout("rbd id object name is %s\n", object_name
);
3742 /* Response will be an encoded string, which includes a length */
3744 size
= sizeof (__le32
) + RBD_IMAGE_ID_LEN_MAX
;
3745 response
= kzalloc(size
, GFP_NOIO
);
3751 ret
= rbd_obj_method_sync(rbd_dev
, object_name
,
3754 response
, RBD_IMAGE_ID_LEN_MAX
, NULL
);
3755 dout("%s: rbd_obj_method_sync returned %d\n", __func__
, ret
);
3760 rbd_dev
->spec
->image_id
= ceph_extract_encoded_string(&p
,
3761 p
+ RBD_IMAGE_ID_LEN_MAX
,
3763 if (IS_ERR(rbd_dev
->spec
->image_id
)) {
3764 ret
= PTR_ERR(rbd_dev
->spec
->image_id
);
3765 rbd_dev
->spec
->image_id
= NULL
;
3767 dout("image_id is %s\n", rbd_dev
->spec
->image_id
);
3776 static int rbd_dev_v1_probe(struct rbd_device
*rbd_dev
)
3781 /* Version 1 images have no id; empty string is used */
3783 rbd_dev
->spec
->image_id
= kstrdup("", GFP_KERNEL
);
3784 if (!rbd_dev
->spec
->image_id
)
3787 /* Record the header object name for this rbd image. */
3789 size
= strlen(rbd_dev
->spec
->image_name
) + sizeof (RBD_SUFFIX
);
3790 rbd_dev
->header_name
= kmalloc(size
, GFP_KERNEL
);
3791 if (!rbd_dev
->header_name
) {
3795 sprintf(rbd_dev
->header_name
, "%s%s",
3796 rbd_dev
->spec
->image_name
, RBD_SUFFIX
);
3798 /* Populate rbd image metadata */
3800 ret
= rbd_read_header(rbd_dev
, &rbd_dev
->header
);
3804 /* Version 1 images have no parent (no layering) */
3806 rbd_dev
->parent_spec
= NULL
;
3807 rbd_dev
->parent_overlap
= 0;
3809 rbd_dev
->image_format
= 1;
3811 dout("discovered version 1 image, header name is %s\n",
3812 rbd_dev
->header_name
);
3817 kfree(rbd_dev
->header_name
);
3818 rbd_dev
->header_name
= NULL
;
3819 kfree(rbd_dev
->spec
->image_id
);
3820 rbd_dev
->spec
->image_id
= NULL
;
3825 static int rbd_dev_v2_probe(struct rbd_device
*rbd_dev
)
3832 * Image id was filled in by the caller. Record the header
3833 * object name for this rbd image.
3835 size
= sizeof (RBD_HEADER_PREFIX
) + strlen(rbd_dev
->spec
->image_id
);
3836 rbd_dev
->header_name
= kmalloc(size
, GFP_KERNEL
);
3837 if (!rbd_dev
->header_name
)
3839 sprintf(rbd_dev
->header_name
, "%s%s",
3840 RBD_HEADER_PREFIX
, rbd_dev
->spec
->image_id
);
3842 /* Get the size and object order for the image */
3844 ret
= rbd_dev_v2_image_size(rbd_dev
);
3848 /* Get the object prefix (a.k.a. block_name) for the image */
3850 ret
= rbd_dev_v2_object_prefix(rbd_dev
);
3854 /* Get the and check features for the image */
3856 ret
= rbd_dev_v2_features(rbd_dev
);
3860 /* If the image supports layering, get the parent info */
3862 if (rbd_dev
->header
.features
& RBD_FEATURE_LAYERING
) {
3863 ret
= rbd_dev_v2_parent_info(rbd_dev
);
3868 /* crypto and compression type aren't (yet) supported for v2 images */
3870 rbd_dev
->header
.crypt_type
= 0;
3871 rbd_dev
->header
.comp_type
= 0;
3873 /* Get the snapshot context, plus the header version */
3875 ret
= rbd_dev_v2_snap_context(rbd_dev
, &ver
);
3878 rbd_dev
->header
.obj_version
= ver
;
3880 rbd_dev
->image_format
= 2;
3882 dout("discovered version 2 image, header name is %s\n",
3883 rbd_dev
->header_name
);
3887 rbd_dev
->parent_overlap
= 0;
3888 rbd_spec_put(rbd_dev
->parent_spec
);
3889 rbd_dev
->parent_spec
= NULL
;
3890 kfree(rbd_dev
->header_name
);
3891 rbd_dev
->header_name
= NULL
;
3892 kfree(rbd_dev
->header
.object_prefix
);
3893 rbd_dev
->header
.object_prefix
= NULL
;
3898 static int rbd_dev_probe_finish(struct rbd_device
*rbd_dev
)
3902 /* no need to lock here, as rbd_dev is not registered yet */
3903 ret
= rbd_dev_snaps_update(rbd_dev
);
3907 ret
= rbd_dev_probe_update_spec(rbd_dev
);
3911 ret
= rbd_dev_set_mapping(rbd_dev
);
3915 /* generate unique id: find highest unique id, add one */
3916 rbd_dev_id_get(rbd_dev
);
3918 /* Fill in the device name, now that we have its id. */
3919 BUILD_BUG_ON(DEV_NAME_LEN
3920 < sizeof (RBD_DRV_NAME
) + MAX_INT_FORMAT_WIDTH
);
3921 sprintf(rbd_dev
->name
, "%s%d", RBD_DRV_NAME
, rbd_dev
->dev_id
);
3923 /* Get our block major device number. */
3925 ret
= register_blkdev(0, rbd_dev
->name
);
3928 rbd_dev
->major
= ret
;
3930 /* Set up the blkdev mapping. */
3932 ret
= rbd_init_disk(rbd_dev
);
3934 goto err_out_blkdev
;
3936 ret
= rbd_bus_add_dev(rbd_dev
);
3941 * At this point cleanup in the event of an error is the job
3942 * of the sysfs code (initiated by rbd_bus_del_dev()).
3944 down_write(&rbd_dev
->header_rwsem
);
3945 ret
= rbd_dev_snaps_register(rbd_dev
);
3946 up_write(&rbd_dev
->header_rwsem
);
3950 ret
= rbd_dev_header_watch_sync(rbd_dev
, 1);
3954 /* Everything's ready. Announce the disk to the world. */
3956 add_disk(rbd_dev
->disk
);
3958 pr_info("%s: added with size 0x%llx\n", rbd_dev
->disk
->disk_name
,
3959 (unsigned long long) rbd_dev
->mapping
.size
);
3963 /* this will also clean up rest of rbd_dev stuff */
3965 rbd_bus_del_dev(rbd_dev
);
3969 rbd_free_disk(rbd_dev
);
3971 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
3973 rbd_dev_id_put(rbd_dev
);
3975 rbd_remove_all_snaps(rbd_dev
);
3981 * Probe for the existence of the header object for the given rbd
3982 * device. For format 2 images this includes determining the image
3985 static int rbd_dev_probe(struct rbd_device
*rbd_dev
)
3990 * Get the id from the image id object. If it's not a
3991 * format 2 image, we'll get ENOENT back, and we'll assume
3992 * it's a format 1 image.
3994 ret
= rbd_dev_image_id(rbd_dev
);
3996 ret
= rbd_dev_v1_probe(rbd_dev
);
3998 ret
= rbd_dev_v2_probe(rbd_dev
);
4000 dout("probe failed, returning %d\n", ret
);
4005 ret
= rbd_dev_probe_finish(rbd_dev
);
4007 rbd_header_free(&rbd_dev
->header
);
4012 static ssize_t
rbd_add(struct bus_type
*bus
,
4016 struct rbd_device
*rbd_dev
= NULL
;
4017 struct ceph_options
*ceph_opts
= NULL
;
4018 struct rbd_options
*rbd_opts
= NULL
;
4019 struct rbd_spec
*spec
= NULL
;
4020 struct rbd_client
*rbdc
;
4021 struct ceph_osd_client
*osdc
;
4024 if (!try_module_get(THIS_MODULE
))
4027 /* parse add command */
4028 rc
= rbd_add_parse_args(buf
, &ceph_opts
, &rbd_opts
, &spec
);
4030 goto err_out_module
;
4032 rbdc
= rbd_get_client(ceph_opts
);
4037 ceph_opts
= NULL
; /* rbd_dev client now owns this */
4040 osdc
= &rbdc
->client
->osdc
;
4041 rc
= ceph_pg_poolid_by_name(osdc
->osdmap
, spec
->pool_name
);
4043 goto err_out_client
;
4044 spec
->pool_id
= (u64
) rc
;
4046 /* The ceph file layout needs to fit pool id in 32 bits */
4048 if (WARN_ON(spec
->pool_id
> (u64
) U32_MAX
)) {
4050 goto err_out_client
;
4053 rbd_dev
= rbd_dev_create(rbdc
, spec
);
4055 goto err_out_client
;
4056 rbdc
= NULL
; /* rbd_dev now owns this */
4057 spec
= NULL
; /* rbd_dev now owns this */
4059 rbd_dev
->mapping
.read_only
= rbd_opts
->read_only
;
4061 rbd_opts
= NULL
; /* done with this */
4063 rc
= rbd_dev_probe(rbd_dev
);
4065 goto err_out_rbd_dev
;
4069 rbd_dev_destroy(rbd_dev
);
4071 rbd_put_client(rbdc
);
4074 ceph_destroy_options(ceph_opts
);
4078 module_put(THIS_MODULE
);
4080 dout("Error adding device %s\n", buf
);
4082 return (ssize_t
) rc
;
4085 static struct rbd_device
*__rbd_get_dev(unsigned long dev_id
)
4087 struct list_head
*tmp
;
4088 struct rbd_device
*rbd_dev
;
4090 spin_lock(&rbd_dev_list_lock
);
4091 list_for_each(tmp
, &rbd_dev_list
) {
4092 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
4093 if (rbd_dev
->dev_id
== dev_id
) {
4094 spin_unlock(&rbd_dev_list_lock
);
4098 spin_unlock(&rbd_dev_list_lock
);
4102 static void rbd_dev_release(struct device
*dev
)
4104 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
4106 if (rbd_dev
->watch_event
)
4107 rbd_dev_header_watch_sync(rbd_dev
, 0);
4109 /* clean up and free blkdev */
4110 rbd_free_disk(rbd_dev
);
4111 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
4113 /* release allocated disk header fields */
4114 rbd_header_free(&rbd_dev
->header
);
4116 /* done with the id, and with the rbd_dev */
4117 rbd_dev_id_put(rbd_dev
);
4118 rbd_assert(rbd_dev
->rbd_client
!= NULL
);
4119 rbd_dev_destroy(rbd_dev
);
4121 /* release module ref */
4122 module_put(THIS_MODULE
);
4125 static ssize_t
rbd_remove(struct bus_type
*bus
,
4129 struct rbd_device
*rbd_dev
= NULL
;
4134 rc
= strict_strtoul(buf
, 10, &ul
);
4138 /* convert to int; abort if we lost anything in the conversion */
4139 target_id
= (int) ul
;
4140 if (target_id
!= ul
)
4143 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
4145 rbd_dev
= __rbd_get_dev(target_id
);
4151 spin_lock_irq(&rbd_dev
->lock
);
4152 if (rbd_dev
->open_count
)
4155 set_bit(RBD_DEV_FLAG_REMOVING
, &rbd_dev
->flags
);
4156 spin_unlock_irq(&rbd_dev
->lock
);
4160 rbd_remove_all_snaps(rbd_dev
);
4161 rbd_bus_del_dev(rbd_dev
);
4164 mutex_unlock(&ctl_mutex
);
4170 * create control files in sysfs
4173 static int rbd_sysfs_init(void)
4177 ret
= device_register(&rbd_root_dev
);
4181 ret
= bus_register(&rbd_bus_type
);
4183 device_unregister(&rbd_root_dev
);
4188 static void rbd_sysfs_cleanup(void)
4190 bus_unregister(&rbd_bus_type
);
4191 device_unregister(&rbd_root_dev
);
4194 int __init
rbd_init(void)
4198 if (!libceph_compatible(NULL
)) {
4199 rbd_warn(NULL
, "libceph incompatibility (quitting)");
4203 rc
= rbd_sysfs_init();
4206 pr_info("loaded " RBD_DRV_NAME_LONG
"\n");
4210 void __exit
rbd_exit(void)
4212 rbd_sysfs_cleanup();
4215 module_init(rbd_init
);
4216 module_exit(rbd_exit
);
4218 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
4219 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
4220 MODULE_DESCRIPTION("rados block device");
4222 /* following authorship retained from original osdblk.c */
4223 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
4225 MODULE_LICENSE("GPL");