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"
45 * The basic unit of block I/O is a sector. It is interpreted in a
46 * number of contexts in Linux (blk, bio, genhd), but the default is
47 * universally 512 bytes. These symbols are just slightly more
48 * meaningful than the bare numbers they represent.
50 #define SECTOR_SHIFT 9
51 #define SECTOR_SIZE (1ULL << SECTOR_SHIFT)
53 #define RBD_DRV_NAME "rbd"
54 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
56 #define RBD_MINORS_PER_MAJOR 256 /* max minors per blkdev */
58 #define RBD_MAX_SNAP_NAME_LEN 32
59 #define RBD_MAX_OPT_LEN 1024
61 #define RBD_SNAP_HEAD_NAME "-"
64 * An RBD device name will be "rbd#", where the "rbd" comes from
65 * RBD_DRV_NAME above, and # is a unique integer identifier.
66 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
67 * enough to hold all possible device names.
69 #define DEV_NAME_LEN 32
70 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1)
72 #define RBD_NOTIFY_TIMEOUT_DEFAULT 10
75 * block device image metadata (in-memory version)
77 struct rbd_image_header
{
83 struct ceph_snap_context
*snapc
;
84 size_t snap_names_len
;
98 * an instance of the client. multiple devices may share an rbd client.
101 struct ceph_client
*client
;
102 struct rbd_options
*rbd_opts
;
104 struct list_head node
;
108 * a request completion status
110 struct rbd_req_status
{
117 * a collection of requests
119 struct rbd_req_coll
{
123 struct rbd_req_status status
[0];
127 * a single io request
130 struct request
*rq
; /* blk layer request */
131 struct bio
*bio
; /* cloned bio */
132 struct page
**pages
; /* list of used pages */
135 struct rbd_req_coll
*coll
;
142 struct list_head node
;
150 int id
; /* blkdev unique id */
152 int major
; /* blkdev assigned major */
153 struct gendisk
*disk
; /* blkdev's gendisk and rq */
154 struct request_queue
*q
;
156 struct rbd_client
*rbd_client
;
158 char name
[DEV_NAME_LEN
]; /* blkdev name, e.g. rbd3 */
160 spinlock_t lock
; /* queue lock */
162 struct rbd_image_header header
;
164 size_t image_name_len
;
169 struct ceph_osd_event
*watch_event
;
170 struct ceph_osd_request
*watch_request
;
172 /* protects updating the header */
173 struct rw_semaphore header_rwsem
;
174 /* name of the snapshot this device reads from */
176 /* id of the snapshot this device reads from */
177 u64 snap_id
; /* current snapshot id */
178 /* whether the snap_id this device reads from still exists */
182 struct list_head node
;
184 /* list of snapshots */
185 struct list_head snaps
;
191 static DEFINE_MUTEX(ctl_mutex
); /* Serialize open/close/setup/teardown */
193 static LIST_HEAD(rbd_dev_list
); /* devices */
194 static DEFINE_SPINLOCK(rbd_dev_list_lock
);
196 static LIST_HEAD(rbd_client_list
); /* clients */
197 static DEFINE_SPINLOCK(rbd_client_list_lock
);
199 static int __rbd_init_snaps_header(struct rbd_device
*rbd_dev
);
200 static void rbd_dev_release(struct device
*dev
);
201 static ssize_t
rbd_snap_add(struct device
*dev
,
202 struct device_attribute
*attr
,
205 static void __rbd_remove_snap_dev(struct rbd_device
*rbd_dev
,
206 struct rbd_snap
*snap
);
208 static ssize_t
rbd_add(struct bus_type
*bus
, const char *buf
,
210 static ssize_t
rbd_remove(struct bus_type
*bus
, const char *buf
,
213 static struct bus_attribute rbd_bus_attrs
[] = {
214 __ATTR(add
, S_IWUSR
, NULL
, rbd_add
),
215 __ATTR(remove
, S_IWUSR
, NULL
, rbd_remove
),
219 static struct bus_type rbd_bus_type
= {
221 .bus_attrs
= rbd_bus_attrs
,
224 static void rbd_root_dev_release(struct device
*dev
)
228 static struct device rbd_root_dev
= {
230 .release
= rbd_root_dev_release
,
234 static struct device
*rbd_get_dev(struct rbd_device
*rbd_dev
)
236 return get_device(&rbd_dev
->dev
);
239 static void rbd_put_dev(struct rbd_device
*rbd_dev
)
241 put_device(&rbd_dev
->dev
);
244 static int __rbd_refresh_header(struct rbd_device
*rbd_dev
);
246 static int rbd_open(struct block_device
*bdev
, fmode_t mode
)
248 struct rbd_device
*rbd_dev
= bdev
->bd_disk
->private_data
;
250 rbd_get_dev(rbd_dev
);
252 set_device_ro(bdev
, rbd_dev
->read_only
);
254 if ((mode
& FMODE_WRITE
) && rbd_dev
->read_only
)
260 static int rbd_release(struct gendisk
*disk
, fmode_t mode
)
262 struct rbd_device
*rbd_dev
= disk
->private_data
;
264 rbd_put_dev(rbd_dev
);
269 static const struct block_device_operations rbd_bd_ops
= {
270 .owner
= THIS_MODULE
,
272 .release
= rbd_release
,
276 * Initialize an rbd client instance.
279 static struct rbd_client
*rbd_client_create(struct ceph_options
*ceph_opts
,
280 struct rbd_options
*rbd_opts
)
282 struct rbd_client
*rbdc
;
285 dout("rbd_client_create\n");
286 rbdc
= kmalloc(sizeof(struct rbd_client
), GFP_KERNEL
);
290 kref_init(&rbdc
->kref
);
291 INIT_LIST_HEAD(&rbdc
->node
);
293 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
295 rbdc
->client
= ceph_create_client(ceph_opts
, rbdc
, 0, 0);
296 if (IS_ERR(rbdc
->client
))
298 ceph_opts
= NULL
; /* Now rbdc->client is responsible for ceph_opts */
300 ret
= ceph_open_session(rbdc
->client
);
304 rbdc
->rbd_opts
= rbd_opts
;
306 spin_lock(&rbd_client_list_lock
);
307 list_add_tail(&rbdc
->node
, &rbd_client_list
);
308 spin_unlock(&rbd_client_list_lock
);
310 mutex_unlock(&ctl_mutex
);
312 dout("rbd_client_create created %p\n", rbdc
);
316 ceph_destroy_client(rbdc
->client
);
318 mutex_unlock(&ctl_mutex
);
322 ceph_destroy_options(ceph_opts
);
327 * Find a ceph client with specific addr and configuration.
329 static struct rbd_client
*__rbd_client_find(struct ceph_options
*ceph_opts
)
331 struct rbd_client
*client_node
;
333 if (ceph_opts
->flags
& CEPH_OPT_NOSHARE
)
336 list_for_each_entry(client_node
, &rbd_client_list
, node
)
337 if (!ceph_compare_options(ceph_opts
, client_node
->client
))
350 /* string args above */
353 static match_table_t rbd_opts_tokens
= {
354 {Opt_notify_timeout
, "notify_timeout=%d"},
356 /* string args above */
360 static int parse_rbd_opts_token(char *c
, void *private)
362 struct rbd_options
*rbd_opts
= private;
363 substring_t argstr
[MAX_OPT_ARGS
];
364 int token
, intval
, ret
;
366 token
= match_token(c
, rbd_opts_tokens
, argstr
);
370 if (token
< Opt_last_int
) {
371 ret
= match_int(&argstr
[0], &intval
);
373 pr_err("bad mount option arg (not int) "
377 dout("got int token %d val %d\n", token
, intval
);
378 } else if (token
> Opt_last_int
&& token
< Opt_last_string
) {
379 dout("got string token %d val %s\n", token
,
382 dout("got token %d\n", token
);
386 case Opt_notify_timeout
:
387 rbd_opts
->notify_timeout
= intval
;
396 * Get a ceph client with specific addr and configuration, if one does
397 * not exist create it.
399 static struct rbd_client
*rbd_get_client(const char *mon_addr
,
403 struct rbd_client
*rbdc
;
404 struct ceph_options
*ceph_opts
;
405 struct rbd_options
*rbd_opts
;
407 rbd_opts
= kzalloc(sizeof(*rbd_opts
), GFP_KERNEL
);
409 return ERR_PTR(-ENOMEM
);
411 rbd_opts
->notify_timeout
= RBD_NOTIFY_TIMEOUT_DEFAULT
;
413 ceph_opts
= ceph_parse_options(options
, mon_addr
,
414 mon_addr
+ mon_addr_len
,
415 parse_rbd_opts_token
, rbd_opts
);
416 if (IS_ERR(ceph_opts
)) {
418 return ERR_CAST(ceph_opts
);
421 spin_lock(&rbd_client_list_lock
);
422 rbdc
= __rbd_client_find(ceph_opts
);
424 /* using an existing client */
425 kref_get(&rbdc
->kref
);
426 spin_unlock(&rbd_client_list_lock
);
428 ceph_destroy_options(ceph_opts
);
433 spin_unlock(&rbd_client_list_lock
);
435 rbdc
= rbd_client_create(ceph_opts
, rbd_opts
);
444 * Destroy ceph client
446 * Caller must hold rbd_client_list_lock.
448 static void rbd_client_release(struct kref
*kref
)
450 struct rbd_client
*rbdc
= container_of(kref
, struct rbd_client
, kref
);
452 dout("rbd_release_client %p\n", rbdc
);
453 spin_lock(&rbd_client_list_lock
);
454 list_del(&rbdc
->node
);
455 spin_unlock(&rbd_client_list_lock
);
457 ceph_destroy_client(rbdc
->client
);
458 kfree(rbdc
->rbd_opts
);
463 * Drop reference to ceph client node. If it's not referenced anymore, release
466 static void rbd_put_client(struct rbd_device
*rbd_dev
)
468 kref_put(&rbd_dev
->rbd_client
->kref
, rbd_client_release
);
469 rbd_dev
->rbd_client
= NULL
;
473 * Destroy requests collection
475 static void rbd_coll_release(struct kref
*kref
)
477 struct rbd_req_coll
*coll
=
478 container_of(kref
, struct rbd_req_coll
, kref
);
480 dout("rbd_coll_release %p\n", coll
);
484 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk
*ondisk
)
486 return !memcmp(&ondisk
->text
,
487 RBD_HEADER_TEXT
, sizeof (RBD_HEADER_TEXT
));
491 * Create a new header structure, translate header format from the on-disk
494 static int rbd_header_from_disk(struct rbd_image_header
*header
,
495 struct rbd_image_header_ondisk
*ondisk
,
501 if (!rbd_dev_ondisk_valid(ondisk
))
504 snap_count
= le32_to_cpu(ondisk
->snap_count
);
505 if (snap_count
> (UINT_MAX
- sizeof(struct ceph_snap_context
))
508 header
->snapc
= kmalloc(sizeof(struct ceph_snap_context
) +
509 snap_count
* sizeof(u64
),
514 header
->snap_names_len
= le64_to_cpu(ondisk
->snap_names_len
);
516 header
->snap_names
= kmalloc(header
->snap_names_len
,
518 if (!header
->snap_names
)
520 header
->snap_sizes
= kmalloc(snap_count
* sizeof(u64
),
522 if (!header
->snap_sizes
)
525 header
->snap_names
= NULL
;
526 header
->snap_sizes
= NULL
;
529 header
->object_prefix
= kmalloc(sizeof (ondisk
->block_name
) + 1,
531 if (!header
->object_prefix
)
534 memcpy(header
->object_prefix
, ondisk
->block_name
,
535 sizeof(ondisk
->block_name
));
536 header
->object_prefix
[sizeof (ondisk
->block_name
)] = '\0';
538 header
->image_size
= le64_to_cpu(ondisk
->image_size
);
539 header
->obj_order
= ondisk
->options
.order
;
540 header
->crypt_type
= ondisk
->options
.crypt_type
;
541 header
->comp_type
= ondisk
->options
.comp_type
;
543 atomic_set(&header
->snapc
->nref
, 1);
544 header
->snapc
->seq
= le64_to_cpu(ondisk
->snap_seq
);
545 header
->snapc
->num_snaps
= snap_count
;
546 header
->total_snaps
= snap_count
;
548 if (snap_count
&& allocated_snaps
== snap_count
) {
549 for (i
= 0; i
< snap_count
; i
++) {
550 header
->snapc
->snaps
[i
] =
551 le64_to_cpu(ondisk
->snaps
[i
].id
);
552 header
->snap_sizes
[i
] =
553 le64_to_cpu(ondisk
->snaps
[i
].image_size
);
556 /* copy snapshot names */
557 memcpy(header
->snap_names
, &ondisk
->snaps
[i
],
558 header
->snap_names_len
);
564 kfree(header
->snap_sizes
);
566 kfree(header
->snap_names
);
568 kfree(header
->snapc
);
572 static int snap_by_name(struct rbd_image_header
*header
, const char *snap_name
,
576 char *p
= header
->snap_names
;
578 for (i
= 0; i
< header
->total_snaps
; i
++) {
579 if (!strcmp(snap_name
, p
)) {
581 /* Found it. Pass back its id and/or size */
584 *seq
= header
->snapc
->snaps
[i
];
586 *size
= header
->snap_sizes
[i
];
589 p
+= strlen(p
) + 1; /* Skip ahead to the next name */
594 static int rbd_header_set_snap(struct rbd_device
*rbd_dev
, u64
*size
)
598 down_write(&rbd_dev
->header_rwsem
);
600 if (!memcmp(rbd_dev
->snap_name
, RBD_SNAP_HEAD_NAME
,
601 sizeof (RBD_SNAP_HEAD_NAME
))) {
602 rbd_dev
->snap_id
= CEPH_NOSNAP
;
603 rbd_dev
->snap_exists
= false;
604 rbd_dev
->read_only
= 0;
606 *size
= rbd_dev
->header
.image_size
;
610 ret
= snap_by_name(&rbd_dev
->header
, rbd_dev
->snap_name
,
614 rbd_dev
->snap_id
= snap_id
;
615 rbd_dev
->snap_exists
= true;
616 rbd_dev
->read_only
= 1;
621 up_write(&rbd_dev
->header_rwsem
);
625 static void rbd_header_free(struct rbd_image_header
*header
)
627 kfree(header
->object_prefix
);
628 kfree(header
->snap_sizes
);
629 kfree(header
->snap_names
);
630 ceph_put_snap_context(header
->snapc
);
634 * get the actual striped segment name, offset and length
636 static u64
rbd_get_segment(struct rbd_image_header
*header
,
637 const char *object_prefix
,
639 char *seg_name
, u64
*segofs
)
641 u64 seg
= ofs
>> header
->obj_order
;
644 snprintf(seg_name
, RBD_MAX_SEG_NAME_LEN
,
645 "%s.%012llx", object_prefix
, seg
);
647 ofs
= ofs
& ((1 << header
->obj_order
) - 1);
648 len
= min_t(u64
, len
, (1 << header
->obj_order
) - ofs
);
656 static int rbd_get_num_segments(struct rbd_image_header
*header
,
659 u64 start_seg
= ofs
>> header
->obj_order
;
660 u64 end_seg
= (ofs
+ len
- 1) >> header
->obj_order
;
661 return end_seg
- start_seg
+ 1;
665 * returns the size of an object in the image
667 static u64
rbd_obj_bytes(struct rbd_image_header
*header
)
669 return 1 << header
->obj_order
;
676 static void bio_chain_put(struct bio
*chain
)
682 chain
= chain
->bi_next
;
688 * zeros a bio chain, starting at specific offset
690 static void zero_bio_chain(struct bio
*chain
, int start_ofs
)
699 bio_for_each_segment(bv
, chain
, i
) {
700 if (pos
+ bv
->bv_len
> start_ofs
) {
701 int remainder
= max(start_ofs
- pos
, 0);
702 buf
= bvec_kmap_irq(bv
, &flags
);
703 memset(buf
+ remainder
, 0,
704 bv
->bv_len
- remainder
);
705 bvec_kunmap_irq(buf
, &flags
);
710 chain
= chain
->bi_next
;
715 * bio_chain_clone - clone a chain of bios up to a certain length.
716 * might return a bio_pair that will need to be released.
718 static struct bio
*bio_chain_clone(struct bio
**old
, struct bio
**next
,
719 struct bio_pair
**bp
,
720 int len
, gfp_t gfpmask
)
722 struct bio
*tmp
, *old_chain
= *old
, *new_chain
= NULL
, *tail
= NULL
;
726 bio_pair_release(*bp
);
730 while (old_chain
&& (total
< len
)) {
731 tmp
= bio_kmalloc(gfpmask
, old_chain
->bi_max_vecs
);
735 if (total
+ old_chain
->bi_size
> len
) {
739 * this split can only happen with a single paged bio,
740 * split_bio will BUG_ON if this is not the case
742 dout("bio_chain_clone split! total=%d remaining=%d"
744 total
, len
- total
, old_chain
->bi_size
);
746 /* split the bio. We'll release it either in the next
747 call, or it will have to be released outside */
748 bp
= bio_split(old_chain
, (len
- total
) / SECTOR_SIZE
);
752 __bio_clone(tmp
, &bp
->bio1
);
756 __bio_clone(tmp
, old_chain
);
757 *next
= old_chain
->bi_next
;
761 gfpmask
&= ~__GFP_WAIT
;
765 new_chain
= tail
= tmp
;
770 old_chain
= old_chain
->bi_next
;
772 total
+= tmp
->bi_size
;
778 tail
->bi_next
= NULL
;
785 dout("bio_chain_clone with err\n");
786 bio_chain_put(new_chain
);
791 * helpers for osd request op vectors.
793 static int rbd_create_rw_ops(struct ceph_osd_req_op
**ops
,
798 *ops
= kzalloc(sizeof(struct ceph_osd_req_op
) * (num_ops
+ 1),
802 (*ops
)[0].op
= opcode
;
804 * op extent offset and length will be set later on
805 * in calc_raw_layout()
807 (*ops
)[0].payload_len
= payload_len
;
811 static void rbd_destroy_ops(struct ceph_osd_req_op
*ops
)
816 static void rbd_coll_end_req_index(struct request
*rq
,
817 struct rbd_req_coll
*coll
,
821 struct request_queue
*q
;
824 dout("rbd_coll_end_req_index %p index %d ret %d len %llu\n",
825 coll
, index
, ret
, (unsigned long long) len
);
831 blk_end_request(rq
, ret
, len
);
837 spin_lock_irq(q
->queue_lock
);
838 coll
->status
[index
].done
= 1;
839 coll
->status
[index
].rc
= ret
;
840 coll
->status
[index
].bytes
= len
;
841 max
= min
= coll
->num_done
;
842 while (max
< coll
->total
&& coll
->status
[max
].done
)
845 for (i
= min
; i
<max
; i
++) {
846 __blk_end_request(rq
, coll
->status
[i
].rc
,
847 coll
->status
[i
].bytes
);
849 kref_put(&coll
->kref
, rbd_coll_release
);
851 spin_unlock_irq(q
->queue_lock
);
854 static void rbd_coll_end_req(struct rbd_request
*req
,
857 rbd_coll_end_req_index(req
->rq
, req
->coll
, req
->coll_index
, ret
, len
);
861 * Send ceph osd request
863 static int rbd_do_request(struct request
*rq
,
864 struct rbd_device
*rbd_dev
,
865 struct ceph_snap_context
*snapc
,
867 const char *object_name
, u64 ofs
, u64 len
,
872 struct ceph_osd_req_op
*ops
,
873 struct rbd_req_coll
*coll
,
875 void (*rbd_cb
)(struct ceph_osd_request
*req
,
876 struct ceph_msg
*msg
),
877 struct ceph_osd_request
**linger_req
,
880 struct ceph_osd_request
*req
;
881 struct ceph_file_layout
*layout
;
884 struct timespec mtime
= CURRENT_TIME
;
885 struct rbd_request
*req_data
;
886 struct ceph_osd_request_head
*reqhead
;
887 struct ceph_osd_client
*osdc
;
889 req_data
= kzalloc(sizeof(*req_data
), GFP_NOIO
);
892 rbd_coll_end_req_index(rq
, coll
, coll_index
,
898 req_data
->coll
= coll
;
899 req_data
->coll_index
= coll_index
;
902 dout("rbd_do_request object_name=%s ofs=%llu len=%llu\n", object_name
,
903 (unsigned long long) ofs
, (unsigned long long) len
);
905 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
906 req
= ceph_osdc_alloc_request(osdc
, flags
, snapc
, ops
,
907 false, GFP_NOIO
, pages
, bio
);
913 req
->r_callback
= rbd_cb
;
917 req_data
->pages
= pages
;
920 req
->r_priv
= req_data
;
922 reqhead
= req
->r_request
->front
.iov_base
;
923 reqhead
->snapid
= cpu_to_le64(CEPH_NOSNAP
);
925 strncpy(req
->r_oid
, object_name
, sizeof(req
->r_oid
));
926 req
->r_oid_len
= strlen(req
->r_oid
);
928 layout
= &req
->r_file_layout
;
929 memset(layout
, 0, sizeof(*layout
));
930 layout
->fl_stripe_unit
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
931 layout
->fl_stripe_count
= cpu_to_le32(1);
932 layout
->fl_object_size
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
933 layout
->fl_pg_pool
= cpu_to_le32(rbd_dev
->pool_id
);
934 ceph_calc_raw_layout(osdc
, layout
, snapid
, ofs
, &len
, &bno
,
937 ceph_osdc_build_request(req
, ofs
, &len
,
941 req
->r_oid
, req
->r_oid_len
);
944 ceph_osdc_set_request_linger(osdc
, req
);
948 ret
= ceph_osdc_start_request(osdc
, req
, false);
953 ret
= ceph_osdc_wait_request(osdc
, req
);
955 *ver
= le64_to_cpu(req
->r_reassert_version
.version
);
956 dout("reassert_ver=%llu\n",
958 le64_to_cpu(req
->r_reassert_version
.version
));
959 ceph_osdc_put_request(req
);
964 bio_chain_put(req_data
->bio
);
965 ceph_osdc_put_request(req
);
967 rbd_coll_end_req(req_data
, ret
, len
);
973 * Ceph osd op callback
975 static void rbd_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
977 struct rbd_request
*req_data
= req
->r_priv
;
978 struct ceph_osd_reply_head
*replyhead
;
979 struct ceph_osd_op
*op
;
985 replyhead
= msg
->front
.iov_base
;
986 WARN_ON(le32_to_cpu(replyhead
->num_ops
) == 0);
987 op
= (void *)(replyhead
+ 1);
988 rc
= le32_to_cpu(replyhead
->result
);
989 bytes
= le64_to_cpu(op
->extent
.length
);
990 read_op
= (le16_to_cpu(op
->op
) == CEPH_OSD_OP_READ
);
992 dout("rbd_req_cb bytes=%llu readop=%d rc=%d\n",
993 (unsigned long long) bytes
, read_op
, (int) rc
);
995 if (rc
== -ENOENT
&& read_op
) {
996 zero_bio_chain(req_data
->bio
, 0);
998 } else if (rc
== 0 && read_op
&& bytes
< req_data
->len
) {
999 zero_bio_chain(req_data
->bio
, bytes
);
1000 bytes
= req_data
->len
;
1003 rbd_coll_end_req(req_data
, rc
, bytes
);
1006 bio_chain_put(req_data
->bio
);
1008 ceph_osdc_put_request(req
);
1012 static void rbd_simple_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
1014 ceph_osdc_put_request(req
);
1018 * Do a synchronous ceph osd operation
1020 static int rbd_req_sync_op(struct rbd_device
*rbd_dev
,
1021 struct ceph_snap_context
*snapc
,
1025 struct ceph_osd_req_op
*orig_ops
,
1026 const char *object_name
,
1029 struct ceph_osd_request
**linger_req
,
1033 struct page
**pages
;
1035 struct ceph_osd_req_op
*ops
= orig_ops
;
1038 num_pages
= calc_pages_for(ofs
, len
);
1039 pages
= ceph_alloc_page_vector(num_pages
, GFP_KERNEL
);
1041 return PTR_ERR(pages
);
1044 payload_len
= (flags
& CEPH_OSD_FLAG_WRITE
? len
: 0);
1045 ret
= rbd_create_rw_ops(&ops
, 1, opcode
, payload_len
);
1049 if ((flags
& CEPH_OSD_FLAG_WRITE
) && buf
) {
1050 ret
= ceph_copy_to_page_vector(pages
, buf
, ofs
, len
);
1056 ret
= rbd_do_request(NULL
, rbd_dev
, snapc
, snapid
,
1057 object_name
, ofs
, len
, NULL
,
1067 if ((flags
& CEPH_OSD_FLAG_READ
) && buf
)
1068 ret
= ceph_copy_from_page_vector(pages
, buf
, ofs
, ret
);
1072 rbd_destroy_ops(ops
);
1074 ceph_release_page_vector(pages
, num_pages
);
1079 * Do an asynchronous ceph osd operation
1081 static int rbd_do_op(struct request
*rq
,
1082 struct rbd_device
*rbd_dev
,
1083 struct ceph_snap_context
*snapc
,
1085 int opcode
, int flags
,
1088 struct rbd_req_coll
*coll
,
1095 struct ceph_osd_req_op
*ops
;
1098 seg_name
= kmalloc(RBD_MAX_SEG_NAME_LEN
+ 1, GFP_NOIO
);
1102 seg_len
= rbd_get_segment(&rbd_dev
->header
,
1103 rbd_dev
->header
.object_prefix
,
1105 seg_name
, &seg_ofs
);
1107 payload_len
= (flags
& CEPH_OSD_FLAG_WRITE
? seg_len
: 0);
1109 ret
= rbd_create_rw_ops(&ops
, 1, opcode
, payload_len
);
1113 /* we've taken care of segment sizes earlier when we
1114 cloned the bios. We should never have a segment
1115 truncated at this point */
1116 BUG_ON(seg_len
< len
);
1118 ret
= rbd_do_request(rq
, rbd_dev
, snapc
, snapid
,
1119 seg_name
, seg_ofs
, seg_len
,
1125 rbd_req_cb
, 0, NULL
);
1127 rbd_destroy_ops(ops
);
1134 * Request async osd write
1136 static int rbd_req_write(struct request
*rq
,
1137 struct rbd_device
*rbd_dev
,
1138 struct ceph_snap_context
*snapc
,
1141 struct rbd_req_coll
*coll
,
1144 return rbd_do_op(rq
, rbd_dev
, snapc
, CEPH_NOSNAP
,
1146 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1147 ofs
, len
, bio
, coll
, coll_index
);
1151 * Request async osd read
1153 static int rbd_req_read(struct request
*rq
,
1154 struct rbd_device
*rbd_dev
,
1158 struct rbd_req_coll
*coll
,
1161 return rbd_do_op(rq
, rbd_dev
, NULL
,
1165 ofs
, len
, bio
, coll
, coll_index
);
1169 * Request sync osd read
1171 static int rbd_req_sync_read(struct rbd_device
*rbd_dev
,
1172 struct ceph_snap_context
*snapc
,
1174 const char *object_name
,
1179 return rbd_req_sync_op(rbd_dev
, NULL
,
1184 object_name
, ofs
, len
, buf
, NULL
, ver
);
1188 * Request sync osd watch
1190 static int rbd_req_sync_notify_ack(struct rbd_device
*rbd_dev
,
1193 const char *object_name
)
1195 struct ceph_osd_req_op
*ops
;
1198 ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_NOTIFY_ACK
, 0);
1202 ops
[0].watch
.ver
= cpu_to_le64(ver
);
1203 ops
[0].watch
.cookie
= notify_id
;
1204 ops
[0].watch
.flag
= 0;
1206 ret
= rbd_do_request(NULL
, rbd_dev
, NULL
, CEPH_NOSNAP
,
1207 object_name
, 0, 0, NULL
,
1212 rbd_simple_req_cb
, 0, NULL
);
1214 rbd_destroy_ops(ops
);
1218 static void rbd_watch_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1220 struct rbd_device
*rbd_dev
= (struct rbd_device
*)data
;
1227 dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
1228 rbd_dev
->header_name
, (unsigned long long) notify_id
,
1229 (unsigned int) opcode
);
1230 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1231 rc
= __rbd_refresh_header(rbd_dev
);
1232 hver
= rbd_dev
->header
.obj_version
;
1233 mutex_unlock(&ctl_mutex
);
1235 pr_warning(RBD_DRV_NAME
"%d got notification but failed to "
1236 " update snaps: %d\n", rbd_dev
->major
, rc
);
1238 rbd_req_sync_notify_ack(rbd_dev
, hver
, notify_id
, rbd_dev
->header_name
);
1242 * Request sync osd watch
1244 static int rbd_req_sync_watch(struct rbd_device
*rbd_dev
,
1245 const char *object_name
,
1248 struct ceph_osd_req_op
*ops
;
1249 struct ceph_osd_client
*osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1251 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_WATCH
, 0);
1255 ret
= ceph_osdc_create_event(osdc
, rbd_watch_cb
, 0,
1256 (void *)rbd_dev
, &rbd_dev
->watch_event
);
1260 ops
[0].watch
.ver
= cpu_to_le64(ver
);
1261 ops
[0].watch
.cookie
= cpu_to_le64(rbd_dev
->watch_event
->cookie
);
1262 ops
[0].watch
.flag
= 1;
1264 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1267 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1269 object_name
, 0, 0, NULL
,
1270 &rbd_dev
->watch_request
, NULL
);
1275 rbd_destroy_ops(ops
);
1279 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1280 rbd_dev
->watch_event
= NULL
;
1282 rbd_destroy_ops(ops
);
1287 * Request sync osd unwatch
1289 static int rbd_req_sync_unwatch(struct rbd_device
*rbd_dev
,
1290 const char *object_name
)
1292 struct ceph_osd_req_op
*ops
;
1294 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_WATCH
, 0);
1298 ops
[0].watch
.ver
= 0;
1299 ops
[0].watch
.cookie
= cpu_to_le64(rbd_dev
->watch_event
->cookie
);
1300 ops
[0].watch
.flag
= 0;
1302 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1305 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1307 object_name
, 0, 0, NULL
, NULL
, NULL
);
1309 rbd_destroy_ops(ops
);
1310 ceph_osdc_cancel_event(rbd_dev
->watch_event
);
1311 rbd_dev
->watch_event
= NULL
;
1315 struct rbd_notify_info
{
1316 struct rbd_device
*rbd_dev
;
1319 static void rbd_notify_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1321 struct rbd_device
*rbd_dev
= (struct rbd_device
*)data
;
1325 dout("rbd_notify_cb %s notify_id=%llu opcode=%u\n",
1326 rbd_dev
->header_name
, (unsigned long long) notify_id
,
1327 (unsigned int) opcode
);
1331 * Request sync osd notify
1333 static int rbd_req_sync_notify(struct rbd_device
*rbd_dev
,
1334 const char *object_name
)
1336 struct ceph_osd_req_op
*ops
;
1337 struct ceph_osd_client
*osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
1338 struct ceph_osd_event
*event
;
1339 struct rbd_notify_info info
;
1340 int payload_len
= sizeof(u32
) + sizeof(u32
);
1343 ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_NOTIFY
, payload_len
);
1347 info
.rbd_dev
= rbd_dev
;
1349 ret
= ceph_osdc_create_event(osdc
, rbd_notify_cb
, 1,
1350 (void *)&info
, &event
);
1354 ops
[0].watch
.ver
= 1;
1355 ops
[0].watch
.flag
= 1;
1356 ops
[0].watch
.cookie
= event
->cookie
;
1357 ops
[0].watch
.prot_ver
= RADOS_NOTIFY_VER
;
1358 ops
[0].watch
.timeout
= 12;
1360 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1363 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1365 object_name
, 0, 0, NULL
, NULL
, NULL
);
1369 ret
= ceph_osdc_wait_event(event
, CEPH_OSD_TIMEOUT_DEFAULT
);
1370 dout("ceph_osdc_wait_event returned %d\n", ret
);
1371 rbd_destroy_ops(ops
);
1375 ceph_osdc_cancel_event(event
);
1377 rbd_destroy_ops(ops
);
1382 * Request sync osd read
1384 static int rbd_req_sync_exec(struct rbd_device
*rbd_dev
,
1385 const char *object_name
,
1386 const char *class_name
,
1387 const char *method_name
,
1392 struct ceph_osd_req_op
*ops
;
1393 int class_name_len
= strlen(class_name
);
1394 int method_name_len
= strlen(method_name
);
1395 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_CALL
,
1396 class_name_len
+ method_name_len
+ len
);
1400 ops
[0].cls
.class_name
= class_name
;
1401 ops
[0].cls
.class_len
= (__u8
) class_name_len
;
1402 ops
[0].cls
.method_name
= method_name
;
1403 ops
[0].cls
.method_len
= (__u8
) method_name_len
;
1404 ops
[0].cls
.argc
= 0;
1405 ops
[0].cls
.indata
= data
;
1406 ops
[0].cls
.indata_len
= len
;
1408 ret
= rbd_req_sync_op(rbd_dev
, NULL
,
1411 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1413 object_name
, 0, 0, NULL
, NULL
, ver
);
1415 rbd_destroy_ops(ops
);
1417 dout("cls_exec returned %d\n", ret
);
1421 static struct rbd_req_coll
*rbd_alloc_coll(int num_reqs
)
1423 struct rbd_req_coll
*coll
=
1424 kzalloc(sizeof(struct rbd_req_coll
) +
1425 sizeof(struct rbd_req_status
) * num_reqs
,
1430 coll
->total
= num_reqs
;
1431 kref_init(&coll
->kref
);
1436 * block device queue callback
1438 static void rbd_rq_fn(struct request_queue
*q
)
1440 struct rbd_device
*rbd_dev
= q
->queuedata
;
1442 struct bio_pair
*bp
= NULL
;
1444 while ((rq
= blk_fetch_request(q
))) {
1446 struct bio
*rq_bio
, *next_bio
= NULL
;
1451 int num_segs
, cur_seg
= 0;
1452 struct rbd_req_coll
*coll
;
1453 struct ceph_snap_context
*snapc
;
1455 /* peek at request from block layer */
1459 dout("fetched request\n");
1461 /* filter out block requests we don't understand */
1462 if ((rq
->cmd_type
!= REQ_TYPE_FS
)) {
1463 __blk_end_request_all(rq
, 0);
1467 /* deduce our operation (read, write) */
1468 do_write
= (rq_data_dir(rq
) == WRITE
);
1470 size
= blk_rq_bytes(rq
);
1471 ofs
= blk_rq_pos(rq
) * SECTOR_SIZE
;
1473 if (do_write
&& rbd_dev
->read_only
) {
1474 __blk_end_request_all(rq
, -EROFS
);
1478 spin_unlock_irq(q
->queue_lock
);
1480 down_read(&rbd_dev
->header_rwsem
);
1482 if (rbd_dev
->snap_id
!= CEPH_NOSNAP
&& !rbd_dev
->snap_exists
) {
1483 up_read(&rbd_dev
->header_rwsem
);
1484 dout("request for non-existent snapshot");
1485 spin_lock_irq(q
->queue_lock
);
1486 __blk_end_request_all(rq
, -ENXIO
);
1490 snapc
= ceph_get_snap_context(rbd_dev
->header
.snapc
);
1492 up_read(&rbd_dev
->header_rwsem
);
1494 dout("%s 0x%x bytes at 0x%llx\n",
1495 do_write
? "write" : "read",
1496 size
, (unsigned long long) blk_rq_pos(rq
) * SECTOR_SIZE
);
1498 num_segs
= rbd_get_num_segments(&rbd_dev
->header
, ofs
, size
);
1499 coll
= rbd_alloc_coll(num_segs
);
1501 spin_lock_irq(q
->queue_lock
);
1502 __blk_end_request_all(rq
, -ENOMEM
);
1503 ceph_put_snap_context(snapc
);
1508 /* a bio clone to be passed down to OSD req */
1509 dout("rq->bio->bi_vcnt=%hu\n", rq
->bio
->bi_vcnt
);
1510 op_size
= rbd_get_segment(&rbd_dev
->header
,
1511 rbd_dev
->header
.object_prefix
,
1514 kref_get(&coll
->kref
);
1515 bio
= bio_chain_clone(&rq_bio
, &next_bio
, &bp
,
1516 op_size
, GFP_ATOMIC
);
1518 rbd_coll_end_req_index(rq
, coll
, cur_seg
,
1524 /* init OSD command: write or read */
1526 rbd_req_write(rq
, rbd_dev
,
1532 rbd_req_read(rq
, rbd_dev
,
1545 kref_put(&coll
->kref
, rbd_coll_release
);
1548 bio_pair_release(bp
);
1549 spin_lock_irq(q
->queue_lock
);
1551 ceph_put_snap_context(snapc
);
1556 * a queue callback. Makes sure that we don't create a bio that spans across
1557 * multiple osd objects. One exception would be with a single page bios,
1558 * which we handle later at bio_chain_clone
1560 static int rbd_merge_bvec(struct request_queue
*q
, struct bvec_merge_data
*bmd
,
1561 struct bio_vec
*bvec
)
1563 struct rbd_device
*rbd_dev
= q
->queuedata
;
1564 unsigned int chunk_sectors
;
1566 unsigned int bio_sectors
;
1569 chunk_sectors
= 1 << (rbd_dev
->header
.obj_order
- SECTOR_SHIFT
);
1570 sector
= bmd
->bi_sector
+ get_start_sect(bmd
->bi_bdev
);
1571 bio_sectors
= bmd
->bi_size
>> SECTOR_SHIFT
;
1573 max
= (chunk_sectors
- ((sector
& (chunk_sectors
- 1))
1574 + bio_sectors
)) << SECTOR_SHIFT
;
1576 max
= 0; /* bio_add cannot handle a negative return */
1577 if (max
<= bvec
->bv_len
&& bio_sectors
== 0)
1578 return bvec
->bv_len
;
1582 static void rbd_free_disk(struct rbd_device
*rbd_dev
)
1584 struct gendisk
*disk
= rbd_dev
->disk
;
1589 rbd_header_free(&rbd_dev
->header
);
1591 if (disk
->flags
& GENHD_FL_UP
)
1594 blk_cleanup_queue(disk
->queue
);
1599 * reload the ondisk the header
1601 static int rbd_read_header(struct rbd_device
*rbd_dev
,
1602 struct rbd_image_header
*header
)
1605 struct rbd_image_header_ondisk
*dh
;
1611 * First reads the fixed-size header to determine the number
1612 * of snapshots, then re-reads it, along with all snapshot
1613 * records as well as their stored names.
1617 dh
= kmalloc(len
, GFP_KERNEL
);
1621 rc
= rbd_req_sync_read(rbd_dev
,
1623 rbd_dev
->header_name
,
1629 rc
= rbd_header_from_disk(header
, dh
, snap_count
, GFP_KERNEL
);
1632 pr_warning("unrecognized header format"
1634 rbd_dev
->image_name
);
1638 if (snap_count
== header
->total_snaps
)
1641 snap_count
= header
->total_snaps
;
1642 len
= sizeof (*dh
) +
1643 snap_count
* sizeof(struct rbd_image_snap_ondisk
) +
1644 header
->snap_names_len
;
1646 rbd_header_free(header
);
1649 header
->obj_version
= ver
;
1659 static int rbd_header_add_snap(struct rbd_device
*rbd_dev
,
1660 const char *snap_name
,
1663 int name_len
= strlen(snap_name
);
1668 struct ceph_mon_client
*monc
;
1670 /* we should create a snapshot only if we're pointing at the head */
1671 if (rbd_dev
->snap_id
!= CEPH_NOSNAP
)
1674 monc
= &rbd_dev
->rbd_client
->client
->monc
;
1675 ret
= ceph_monc_create_snapid(monc
, rbd_dev
->pool_id
, &new_snapid
);
1676 dout("created snapid=%llu\n", (unsigned long long) new_snapid
);
1680 data
= kmalloc(name_len
+ 16, gfp_flags
);
1685 e
= data
+ name_len
+ 16;
1687 ceph_encode_string_safe(&p
, e
, snap_name
, name_len
, bad
);
1688 ceph_encode_64_safe(&p
, e
, new_snapid
, bad
);
1690 ret
= rbd_req_sync_exec(rbd_dev
, rbd_dev
->header_name
,
1692 data
, p
- data
, &ver
);
1696 return ret
< 0 ? ret
: 0;
1701 static void __rbd_remove_all_snaps(struct rbd_device
*rbd_dev
)
1703 struct rbd_snap
*snap
;
1704 struct rbd_snap
*next
;
1706 list_for_each_entry_safe(snap
, next
, &rbd_dev
->snaps
, node
)
1707 __rbd_remove_snap_dev(rbd_dev
, snap
);
1711 * only read the first part of the ondisk header, without the snaps info
1713 static int __rbd_refresh_header(struct rbd_device
*rbd_dev
)
1716 struct rbd_image_header h
;
1718 ret
= rbd_read_header(rbd_dev
, &h
);
1722 down_write(&rbd_dev
->header_rwsem
);
1725 if (rbd_dev
->snap_id
== CEPH_NOSNAP
) {
1726 sector_t size
= (sector_t
) h
.image_size
/ SECTOR_SIZE
;
1728 dout("setting size to %llu sectors", (unsigned long long) size
);
1729 set_capacity(rbd_dev
->disk
, size
);
1732 /* rbd_dev->header.object_prefix shouldn't change */
1733 kfree(rbd_dev
->header
.snap_sizes
);
1734 kfree(rbd_dev
->header
.snap_names
);
1735 /* osd requests may still refer to snapc */
1736 ceph_put_snap_context(rbd_dev
->header
.snapc
);
1738 rbd_dev
->header
.obj_version
= h
.obj_version
;
1739 rbd_dev
->header
.image_size
= h
.image_size
;
1740 rbd_dev
->header
.total_snaps
= h
.total_snaps
;
1741 rbd_dev
->header
.snapc
= h
.snapc
;
1742 rbd_dev
->header
.snap_names
= h
.snap_names
;
1743 rbd_dev
->header
.snap_names_len
= h
.snap_names_len
;
1744 rbd_dev
->header
.snap_sizes
= h
.snap_sizes
;
1745 /* Free the extra copy of the object prefix */
1746 WARN_ON(strcmp(rbd_dev
->header
.object_prefix
, h
.object_prefix
));
1747 kfree(h
.object_prefix
);
1749 ret
= __rbd_init_snaps_header(rbd_dev
);
1751 up_write(&rbd_dev
->header_rwsem
);
1756 static int rbd_init_disk(struct rbd_device
*rbd_dev
)
1758 struct gendisk
*disk
;
1759 struct request_queue
*q
;
1764 /* contact OSD, request size info about the object being mapped */
1765 rc
= rbd_read_header(rbd_dev
, &rbd_dev
->header
);
1769 /* no need to lock here, as rbd_dev is not registered yet */
1770 rc
= __rbd_init_snaps_header(rbd_dev
);
1774 rc
= rbd_header_set_snap(rbd_dev
, &total_size
);
1778 /* create gendisk info */
1780 disk
= alloc_disk(RBD_MINORS_PER_MAJOR
);
1784 snprintf(disk
->disk_name
, sizeof(disk
->disk_name
), RBD_DRV_NAME
"%d",
1786 disk
->major
= rbd_dev
->major
;
1787 disk
->first_minor
= 0;
1788 disk
->fops
= &rbd_bd_ops
;
1789 disk
->private_data
= rbd_dev
;
1793 q
= blk_init_queue(rbd_rq_fn
, &rbd_dev
->lock
);
1797 /* We use the default size, but let's be explicit about it. */
1798 blk_queue_physical_block_size(q
, SECTOR_SIZE
);
1800 /* set io sizes to object size */
1801 segment_size
= rbd_obj_bytes(&rbd_dev
->header
);
1802 blk_queue_max_hw_sectors(q
, segment_size
/ SECTOR_SIZE
);
1803 blk_queue_max_segment_size(q
, segment_size
);
1804 blk_queue_io_min(q
, segment_size
);
1805 blk_queue_io_opt(q
, segment_size
);
1807 blk_queue_merge_bvec(q
, rbd_merge_bvec
);
1810 q
->queuedata
= rbd_dev
;
1812 rbd_dev
->disk
= disk
;
1815 /* finally, announce the disk to the world */
1816 set_capacity(disk
, total_size
/ SECTOR_SIZE
);
1819 pr_info("%s: added with size 0x%llx\n",
1820 disk
->disk_name
, (unsigned long long)total_size
);
1833 static struct rbd_device
*dev_to_rbd_dev(struct device
*dev
)
1835 return container_of(dev
, struct rbd_device
, dev
);
1838 static ssize_t
rbd_size_show(struct device
*dev
,
1839 struct device_attribute
*attr
, char *buf
)
1841 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1844 down_read(&rbd_dev
->header_rwsem
);
1845 size
= get_capacity(rbd_dev
->disk
);
1846 up_read(&rbd_dev
->header_rwsem
);
1848 return sprintf(buf
, "%llu\n", (unsigned long long) size
* SECTOR_SIZE
);
1851 static ssize_t
rbd_major_show(struct device
*dev
,
1852 struct device_attribute
*attr
, char *buf
)
1854 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1856 return sprintf(buf
, "%d\n", rbd_dev
->major
);
1859 static ssize_t
rbd_client_id_show(struct device
*dev
,
1860 struct device_attribute
*attr
, char *buf
)
1862 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1864 return sprintf(buf
, "client%lld\n",
1865 ceph_client_id(rbd_dev
->rbd_client
->client
));
1868 static ssize_t
rbd_pool_show(struct device
*dev
,
1869 struct device_attribute
*attr
, char *buf
)
1871 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1873 return sprintf(buf
, "%s\n", rbd_dev
->pool_name
);
1876 static ssize_t
rbd_pool_id_show(struct device
*dev
,
1877 struct device_attribute
*attr
, char *buf
)
1879 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1881 return sprintf(buf
, "%d\n", rbd_dev
->pool_id
);
1884 static ssize_t
rbd_name_show(struct device
*dev
,
1885 struct device_attribute
*attr
, char *buf
)
1887 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1889 return sprintf(buf
, "%s\n", rbd_dev
->image_name
);
1892 static ssize_t
rbd_snap_show(struct device
*dev
,
1893 struct device_attribute
*attr
,
1896 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1898 return sprintf(buf
, "%s\n", rbd_dev
->snap_name
);
1901 static ssize_t
rbd_image_refresh(struct device
*dev
,
1902 struct device_attribute
*attr
,
1906 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1910 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1912 rc
= __rbd_refresh_header(rbd_dev
);
1916 mutex_unlock(&ctl_mutex
);
1920 static DEVICE_ATTR(size
, S_IRUGO
, rbd_size_show
, NULL
);
1921 static DEVICE_ATTR(major
, S_IRUGO
, rbd_major_show
, NULL
);
1922 static DEVICE_ATTR(client_id
, S_IRUGO
, rbd_client_id_show
, NULL
);
1923 static DEVICE_ATTR(pool
, S_IRUGO
, rbd_pool_show
, NULL
);
1924 static DEVICE_ATTR(pool_id
, S_IRUGO
, rbd_pool_id_show
, NULL
);
1925 static DEVICE_ATTR(name
, S_IRUGO
, rbd_name_show
, NULL
);
1926 static DEVICE_ATTR(refresh
, S_IWUSR
, NULL
, rbd_image_refresh
);
1927 static DEVICE_ATTR(current_snap
, S_IRUGO
, rbd_snap_show
, NULL
);
1928 static DEVICE_ATTR(create_snap
, S_IWUSR
, NULL
, rbd_snap_add
);
1930 static struct attribute
*rbd_attrs
[] = {
1931 &dev_attr_size
.attr
,
1932 &dev_attr_major
.attr
,
1933 &dev_attr_client_id
.attr
,
1934 &dev_attr_pool
.attr
,
1935 &dev_attr_pool_id
.attr
,
1936 &dev_attr_name
.attr
,
1937 &dev_attr_current_snap
.attr
,
1938 &dev_attr_refresh
.attr
,
1939 &dev_attr_create_snap
.attr
,
1943 static struct attribute_group rbd_attr_group
= {
1947 static const struct attribute_group
*rbd_attr_groups
[] = {
1952 static void rbd_sysfs_dev_release(struct device
*dev
)
1956 static struct device_type rbd_device_type
= {
1958 .groups
= rbd_attr_groups
,
1959 .release
= rbd_sysfs_dev_release
,
1967 static ssize_t
rbd_snap_size_show(struct device
*dev
,
1968 struct device_attribute
*attr
,
1971 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1973 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->size
);
1976 static ssize_t
rbd_snap_id_show(struct device
*dev
,
1977 struct device_attribute
*attr
,
1980 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1982 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->id
);
1985 static DEVICE_ATTR(snap_size
, S_IRUGO
, rbd_snap_size_show
, NULL
);
1986 static DEVICE_ATTR(snap_id
, S_IRUGO
, rbd_snap_id_show
, NULL
);
1988 static struct attribute
*rbd_snap_attrs
[] = {
1989 &dev_attr_snap_size
.attr
,
1990 &dev_attr_snap_id
.attr
,
1994 static struct attribute_group rbd_snap_attr_group
= {
1995 .attrs
= rbd_snap_attrs
,
1998 static void rbd_snap_dev_release(struct device
*dev
)
2000 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
2005 static const struct attribute_group
*rbd_snap_attr_groups
[] = {
2006 &rbd_snap_attr_group
,
2010 static struct device_type rbd_snap_device_type
= {
2011 .groups
= rbd_snap_attr_groups
,
2012 .release
= rbd_snap_dev_release
,
2015 static void __rbd_remove_snap_dev(struct rbd_device
*rbd_dev
,
2016 struct rbd_snap
*snap
)
2018 list_del(&snap
->node
);
2019 device_unregister(&snap
->dev
);
2022 static int rbd_register_snap_dev(struct rbd_device
*rbd_dev
,
2023 struct rbd_snap
*snap
,
2024 struct device
*parent
)
2026 struct device
*dev
= &snap
->dev
;
2029 dev
->type
= &rbd_snap_device_type
;
2030 dev
->parent
= parent
;
2031 dev
->release
= rbd_snap_dev_release
;
2032 dev_set_name(dev
, "snap_%s", snap
->name
);
2033 ret
= device_register(dev
);
2038 static int __rbd_add_snap_dev(struct rbd_device
*rbd_dev
,
2039 int i
, const char *name
,
2040 struct rbd_snap
**snapp
)
2043 struct rbd_snap
*snap
= kzalloc(sizeof(*snap
), GFP_KERNEL
);
2046 snap
->name
= kstrdup(name
, GFP_KERNEL
);
2047 snap
->size
= rbd_dev
->header
.snap_sizes
[i
];
2048 snap
->id
= rbd_dev
->header
.snapc
->snaps
[i
];
2049 if (device_is_registered(&rbd_dev
->dev
)) {
2050 ret
= rbd_register_snap_dev(rbd_dev
, snap
,
2064 * search for the previous snap in a null delimited string list
2066 const char *rbd_prev_snap_name(const char *name
, const char *start
)
2068 if (name
< start
+ 2)
2081 * compare the old list of snapshots that we have to what's in the header
2082 * and update it accordingly. Note that the header holds the snapshots
2083 * in a reverse order (from newest to oldest) and we need to go from
2084 * older to new so that we don't get a duplicate snap name when
2085 * doing the process (e.g., removed snapshot and recreated a new
2086 * one with the same name.
2088 static int __rbd_init_snaps_header(struct rbd_device
*rbd_dev
)
2090 const char *name
, *first_name
;
2091 int i
= rbd_dev
->header
.total_snaps
;
2092 struct rbd_snap
*snap
, *old_snap
= NULL
;
2094 struct list_head
*p
, *n
;
2096 first_name
= rbd_dev
->header
.snap_names
;
2097 name
= first_name
+ rbd_dev
->header
.snap_names_len
;
2099 list_for_each_prev_safe(p
, n
, &rbd_dev
->snaps
) {
2102 old_snap
= list_entry(p
, struct rbd_snap
, node
);
2105 cur_id
= rbd_dev
->header
.snapc
->snaps
[i
- 1];
2107 if (!i
|| old_snap
->id
< cur_id
) {
2109 * old_snap->id was skipped, thus was
2110 * removed. If this rbd_dev is mapped to
2111 * the removed snapshot, record that it no
2112 * longer exists, to prevent further I/O.
2114 if (rbd_dev
->snap_id
== old_snap
->id
)
2115 rbd_dev
->snap_exists
= false;
2116 __rbd_remove_snap_dev(rbd_dev
, old_snap
);
2119 if (old_snap
->id
== cur_id
) {
2120 /* we have this snapshot already */
2122 name
= rbd_prev_snap_name(name
, first_name
);
2126 i
--, name
= rbd_prev_snap_name(name
, first_name
)) {
2131 cur_id
= rbd_dev
->header
.snapc
->snaps
[i
];
2132 /* snapshot removal? handle it above */
2133 if (cur_id
>= old_snap
->id
)
2135 /* a new snapshot */
2136 ret
= __rbd_add_snap_dev(rbd_dev
, i
- 1, name
, &snap
);
2140 /* note that we add it backward so using n and not p */
2141 list_add(&snap
->node
, n
);
2145 /* we're done going over the old snap list, just add what's left */
2146 for (; i
> 0; i
--) {
2147 name
= rbd_prev_snap_name(name
, first_name
);
2152 ret
= __rbd_add_snap_dev(rbd_dev
, i
- 1, name
, &snap
);
2155 list_add(&snap
->node
, &rbd_dev
->snaps
);
2161 static int rbd_bus_add_dev(struct rbd_device
*rbd_dev
)
2165 struct rbd_snap
*snap
;
2167 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2168 dev
= &rbd_dev
->dev
;
2170 dev
->bus
= &rbd_bus_type
;
2171 dev
->type
= &rbd_device_type
;
2172 dev
->parent
= &rbd_root_dev
;
2173 dev
->release
= rbd_dev_release
;
2174 dev_set_name(dev
, "%d", rbd_dev
->id
);
2175 ret
= device_register(dev
);
2179 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
2180 ret
= rbd_register_snap_dev(rbd_dev
, snap
,
2186 mutex_unlock(&ctl_mutex
);
2190 static void rbd_bus_del_dev(struct rbd_device
*rbd_dev
)
2192 device_unregister(&rbd_dev
->dev
);
2195 static int rbd_init_watch_dev(struct rbd_device
*rbd_dev
)
2200 ret
= rbd_req_sync_watch(rbd_dev
, rbd_dev
->header_name
,
2201 rbd_dev
->header
.obj_version
);
2202 if (ret
== -ERANGE
) {
2203 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2204 rc
= __rbd_refresh_header(rbd_dev
);
2205 mutex_unlock(&ctl_mutex
);
2209 } while (ret
== -ERANGE
);
2214 static atomic64_t rbd_id_max
= ATOMIC64_INIT(0);
2217 * Get a unique rbd identifier for the given new rbd_dev, and add
2218 * the rbd_dev to the global list. The minimum rbd id is 1.
2220 static void rbd_id_get(struct rbd_device
*rbd_dev
)
2222 rbd_dev
->id
= atomic64_inc_return(&rbd_id_max
);
2224 spin_lock(&rbd_dev_list_lock
);
2225 list_add_tail(&rbd_dev
->node
, &rbd_dev_list
);
2226 spin_unlock(&rbd_dev_list_lock
);
2230 * Remove an rbd_dev from the global list, and record that its
2231 * identifier is no longer in use.
2233 static void rbd_id_put(struct rbd_device
*rbd_dev
)
2235 struct list_head
*tmp
;
2236 int rbd_id
= rbd_dev
->id
;
2241 spin_lock(&rbd_dev_list_lock
);
2242 list_del_init(&rbd_dev
->node
);
2245 * If the id being "put" is not the current maximum, there
2246 * is nothing special we need to do.
2248 if (rbd_id
!= atomic64_read(&rbd_id_max
)) {
2249 spin_unlock(&rbd_dev_list_lock
);
2254 * We need to update the current maximum id. Search the
2255 * list to find out what it is. We're more likely to find
2256 * the maximum at the end, so search the list backward.
2259 list_for_each_prev(tmp
, &rbd_dev_list
) {
2260 struct rbd_device
*rbd_dev
;
2262 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
2263 if (rbd_id
> max_id
)
2266 spin_unlock(&rbd_dev_list_lock
);
2269 * The max id could have been updated by rbd_id_get(), in
2270 * which case it now accurately reflects the new maximum.
2271 * Be careful not to overwrite the maximum value in that
2274 atomic64_cmpxchg(&rbd_id_max
, rbd_id
, max_id
);
2278 * Skips over white space at *buf, and updates *buf to point to the
2279 * first found non-space character (if any). Returns the length of
2280 * the token (string of non-white space characters) found. Note
2281 * that *buf must be terminated with '\0'.
2283 static inline size_t next_token(const char **buf
)
2286 * These are the characters that produce nonzero for
2287 * isspace() in the "C" and "POSIX" locales.
2289 const char *spaces
= " \f\n\r\t\v";
2291 *buf
+= strspn(*buf
, spaces
); /* Find start of token */
2293 return strcspn(*buf
, spaces
); /* Return token length */
2297 * Finds the next token in *buf, and if the provided token buffer is
2298 * big enough, copies the found token into it. The result, if
2299 * copied, is guaranteed to be terminated with '\0'. Note that *buf
2300 * must be terminated with '\0' on entry.
2302 * Returns the length of the token found (not including the '\0').
2303 * Return value will be 0 if no token is found, and it will be >=
2304 * token_size if the token would not fit.
2306 * The *buf pointer will be updated to point beyond the end of the
2307 * found token. Note that this occurs even if the token buffer is
2308 * too small to hold it.
2310 static inline size_t copy_token(const char **buf
,
2316 len
= next_token(buf
);
2317 if (len
< token_size
) {
2318 memcpy(token
, *buf
, len
);
2319 *(token
+ len
) = '\0';
2327 * Finds the next token in *buf, dynamically allocates a buffer big
2328 * enough to hold a copy of it, and copies the token into the new
2329 * buffer. The copy is guaranteed to be terminated with '\0'. Note
2330 * that a duplicate buffer is created even for a zero-length token.
2332 * Returns a pointer to the newly-allocated duplicate, or a null
2333 * pointer if memory for the duplicate was not available. If
2334 * the lenp argument is a non-null pointer, the length of the token
2335 * (not including the '\0') is returned in *lenp.
2337 * If successful, the *buf pointer will be updated to point beyond
2338 * the end of the found token.
2340 * Note: uses GFP_KERNEL for allocation.
2342 static inline char *dup_token(const char **buf
, size_t *lenp
)
2347 len
= next_token(buf
);
2348 dup
= kmalloc(len
+ 1, GFP_KERNEL
);
2352 memcpy(dup
, *buf
, len
);
2353 *(dup
+ len
) = '\0';
2363 * This fills in the pool_name, image_name, image_name_len, snap_name,
2364 * rbd_dev, rbd_md_name, and name fields of the given rbd_dev, based
2365 * on the list of monitor addresses and other options provided via
2368 * Note: rbd_dev is assumed to have been initially zero-filled.
2370 static int rbd_add_parse_args(struct rbd_device
*rbd_dev
,
2372 const char **mon_addrs
,
2373 size_t *mon_addrs_size
,
2375 size_t options_size
)
2380 /* The first four tokens are required */
2382 len
= next_token(&buf
);
2385 *mon_addrs_size
= len
+ 1;
2390 len
= copy_token(&buf
, options
, options_size
);
2391 if (!len
|| len
>= options_size
)
2395 rbd_dev
->pool_name
= dup_token(&buf
, NULL
);
2396 if (!rbd_dev
->pool_name
)
2399 rbd_dev
->image_name
= dup_token(&buf
, &rbd_dev
->image_name_len
);
2400 if (!rbd_dev
->image_name
)
2403 /* Create the name of the header object */
2405 rbd_dev
->header_name
= kmalloc(rbd_dev
->image_name_len
2406 + sizeof (RBD_SUFFIX
),
2408 if (!rbd_dev
->header_name
)
2410 sprintf(rbd_dev
->header_name
, "%s%s", rbd_dev
->image_name
, RBD_SUFFIX
);
2413 * The snapshot name is optional. If none is is supplied,
2414 * we use the default value.
2416 rbd_dev
->snap_name
= dup_token(&buf
, &len
);
2417 if (!rbd_dev
->snap_name
)
2420 /* Replace the empty name with the default */
2421 kfree(rbd_dev
->snap_name
);
2423 = kmalloc(sizeof (RBD_SNAP_HEAD_NAME
), GFP_KERNEL
);
2424 if (!rbd_dev
->snap_name
)
2427 memcpy(rbd_dev
->snap_name
, RBD_SNAP_HEAD_NAME
,
2428 sizeof (RBD_SNAP_HEAD_NAME
));
2434 kfree(rbd_dev
->header_name
);
2435 kfree(rbd_dev
->image_name
);
2436 kfree(rbd_dev
->pool_name
);
2437 rbd_dev
->pool_name
= NULL
;
2442 static ssize_t
rbd_add(struct bus_type
*bus
,
2447 struct rbd_device
*rbd_dev
= NULL
;
2448 const char *mon_addrs
= NULL
;
2449 size_t mon_addrs_size
= 0;
2450 struct ceph_osd_client
*osdc
;
2453 if (!try_module_get(THIS_MODULE
))
2456 options
= kmalloc(count
, GFP_KERNEL
);
2459 rbd_dev
= kzalloc(sizeof(*rbd_dev
), GFP_KERNEL
);
2463 /* static rbd_device initialization */
2464 spin_lock_init(&rbd_dev
->lock
);
2465 INIT_LIST_HEAD(&rbd_dev
->node
);
2466 INIT_LIST_HEAD(&rbd_dev
->snaps
);
2467 init_rwsem(&rbd_dev
->header_rwsem
);
2469 /* generate unique id: find highest unique id, add one */
2470 rbd_id_get(rbd_dev
);
2472 /* Fill in the device name, now that we have its id. */
2473 BUILD_BUG_ON(DEV_NAME_LEN
2474 < sizeof (RBD_DRV_NAME
) + MAX_INT_FORMAT_WIDTH
);
2475 sprintf(rbd_dev
->name
, "%s%d", RBD_DRV_NAME
, rbd_dev
->id
);
2477 /* parse add command */
2478 rc
= rbd_add_parse_args(rbd_dev
, buf
, &mon_addrs
, &mon_addrs_size
,
2483 rbd_dev
->rbd_client
= rbd_get_client(mon_addrs
, mon_addrs_size
- 1,
2485 if (IS_ERR(rbd_dev
->rbd_client
)) {
2486 rc
= PTR_ERR(rbd_dev
->rbd_client
);
2491 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
2492 rc
= ceph_pg_poolid_by_name(osdc
->osdmap
, rbd_dev
->pool_name
);
2494 goto err_out_client
;
2495 rbd_dev
->pool_id
= rc
;
2497 /* register our block device */
2498 rc
= register_blkdev(0, rbd_dev
->name
);
2500 goto err_out_client
;
2501 rbd_dev
->major
= rc
;
2503 rc
= rbd_bus_add_dev(rbd_dev
);
2505 goto err_out_blkdev
;
2508 * At this point cleanup in the event of an error is the job
2509 * of the sysfs code (initiated by rbd_bus_del_dev()).
2511 * Set up and announce blkdev mapping.
2513 rc
= rbd_init_disk(rbd_dev
);
2517 rc
= rbd_init_watch_dev(rbd_dev
);
2524 /* this will also clean up rest of rbd_dev stuff */
2526 rbd_bus_del_dev(rbd_dev
);
2531 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
2533 rbd_put_client(rbd_dev
);
2535 if (rbd_dev
->pool_name
) {
2536 kfree(rbd_dev
->snap_name
);
2537 kfree(rbd_dev
->header_name
);
2538 kfree(rbd_dev
->image_name
);
2539 kfree(rbd_dev
->pool_name
);
2541 rbd_id_put(rbd_dev
);
2546 dout("Error adding device %s\n", buf
);
2547 module_put(THIS_MODULE
);
2549 return (ssize_t
) rc
;
2552 static struct rbd_device
*__rbd_get_dev(unsigned long id
)
2554 struct list_head
*tmp
;
2555 struct rbd_device
*rbd_dev
;
2557 spin_lock(&rbd_dev_list_lock
);
2558 list_for_each(tmp
, &rbd_dev_list
) {
2559 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
2560 if (rbd_dev
->id
== id
) {
2561 spin_unlock(&rbd_dev_list_lock
);
2565 spin_unlock(&rbd_dev_list_lock
);
2569 static void rbd_dev_release(struct device
*dev
)
2571 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2573 if (rbd_dev
->watch_request
) {
2574 struct ceph_client
*client
= rbd_dev
->rbd_client
->client
;
2576 ceph_osdc_unregister_linger_request(&client
->osdc
,
2577 rbd_dev
->watch_request
);
2579 if (rbd_dev
->watch_event
)
2580 rbd_req_sync_unwatch(rbd_dev
, rbd_dev
->header_name
);
2582 rbd_put_client(rbd_dev
);
2584 /* clean up and free blkdev */
2585 rbd_free_disk(rbd_dev
);
2586 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
2588 /* done with the id, and with the rbd_dev */
2589 kfree(rbd_dev
->snap_name
);
2590 kfree(rbd_dev
->header_name
);
2591 kfree(rbd_dev
->pool_name
);
2592 kfree(rbd_dev
->image_name
);
2593 rbd_id_put(rbd_dev
);
2596 /* release module ref */
2597 module_put(THIS_MODULE
);
2600 static ssize_t
rbd_remove(struct bus_type
*bus
,
2604 struct rbd_device
*rbd_dev
= NULL
;
2609 rc
= strict_strtoul(buf
, 10, &ul
);
2613 /* convert to int; abort if we lost anything in the conversion */
2614 target_id
= (int) ul
;
2615 if (target_id
!= ul
)
2618 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2620 rbd_dev
= __rbd_get_dev(target_id
);
2626 __rbd_remove_all_snaps(rbd_dev
);
2627 rbd_bus_del_dev(rbd_dev
);
2630 mutex_unlock(&ctl_mutex
);
2634 static ssize_t
rbd_snap_add(struct device
*dev
,
2635 struct device_attribute
*attr
,
2639 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2641 char *name
= kmalloc(count
+ 1, GFP_KERNEL
);
2645 snprintf(name
, count
, "%s", buf
);
2647 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2649 ret
= rbd_header_add_snap(rbd_dev
,
2654 ret
= __rbd_refresh_header(rbd_dev
);
2658 /* shouldn't hold ctl_mutex when notifying.. notify might
2659 trigger a watch callback that would need to get that mutex */
2660 mutex_unlock(&ctl_mutex
);
2662 /* make a best effort, don't error if failed */
2663 rbd_req_sync_notify(rbd_dev
, rbd_dev
->header_name
);
2670 mutex_unlock(&ctl_mutex
);
2676 * create control files in sysfs
2679 static int rbd_sysfs_init(void)
2683 ret
= device_register(&rbd_root_dev
);
2687 ret
= bus_register(&rbd_bus_type
);
2689 device_unregister(&rbd_root_dev
);
2694 static void rbd_sysfs_cleanup(void)
2696 bus_unregister(&rbd_bus_type
);
2697 device_unregister(&rbd_root_dev
);
2700 int __init
rbd_init(void)
2704 rc
= rbd_sysfs_init();
2707 pr_info("loaded " RBD_DRV_NAME_LONG
"\n");
2711 void __exit
rbd_exit(void)
2713 rbd_sysfs_cleanup();
2716 module_init(rbd_init
);
2717 module_exit(rbd_exit
);
2719 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
2720 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
2721 MODULE_DESCRIPTION("rados block device");
2723 /* following authorship retained from original osdblk.c */
2724 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
2726 MODULE_LICENSE("GPL");