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_MD_NAME_LEN (RBD_MAX_OBJ_NAME_LEN + sizeof(RBD_SUFFIX))
59 #define RBD_MAX_POOL_NAME_LEN 64
60 #define RBD_MAX_SNAP_NAME_LEN 32
61 #define RBD_MAX_OPT_LEN 1024
63 #define RBD_SNAP_HEAD_NAME "-"
66 * An RBD device name will be "rbd#", where the "rbd" comes from
67 * RBD_DRV_NAME above, and # is a unique integer identifier.
68 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
69 * enough to hold all possible device names.
71 #define DEV_NAME_LEN 32
72 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1)
74 #define RBD_NOTIFY_TIMEOUT_DEFAULT 10
77 * block device image metadata (in-memory version)
79 struct rbd_image_header
{
85 struct ceph_snap_context
*snapc
;
86 size_t snap_names_len
;
101 * an instance of the client. multiple devices may share an rbd client.
104 struct ceph_client
*client
;
105 struct rbd_options
*rbd_opts
;
107 struct list_head node
;
111 * a request completion status
113 struct rbd_req_status
{
120 * a collection of requests
122 struct rbd_req_coll
{
126 struct rbd_req_status status
[0];
130 * a single io request
133 struct request
*rq
; /* blk layer request */
134 struct bio
*bio
; /* cloned bio */
135 struct page
**pages
; /* list of used pages */
138 struct rbd_req_coll
*coll
;
145 struct list_head node
;
153 int id
; /* blkdev unique id */
155 int major
; /* blkdev assigned major */
156 struct gendisk
*disk
; /* blkdev's gendisk and rq */
157 struct request_queue
*q
;
159 struct rbd_client
*rbd_client
;
161 char name
[DEV_NAME_LEN
]; /* blkdev name, e.g. rbd3 */
163 spinlock_t lock
; /* queue lock */
165 struct rbd_image_header header
;
166 char obj
[RBD_MAX_OBJ_NAME_LEN
]; /* rbd image name */
168 char obj_md_name
[RBD_MAX_MD_NAME_LEN
]; /* hdr nm. */
169 char pool_name
[RBD_MAX_POOL_NAME_LEN
];
172 struct ceph_osd_event
*watch_event
;
173 struct ceph_osd_request
*watch_request
;
175 /* protects updating the header */
176 struct rw_semaphore header_rwsem
;
177 char snap_name
[RBD_MAX_SNAP_NAME_LEN
];
178 u64 snap_id
; /* current snapshot id */
181 struct list_head node
;
183 /* list of snapshots */
184 struct list_head snaps
;
190 static DEFINE_MUTEX(ctl_mutex
); /* Serialize open/close/setup/teardown */
192 static LIST_HEAD(rbd_dev_list
); /* devices */
193 static DEFINE_SPINLOCK(rbd_dev_list_lock
);
195 static LIST_HEAD(rbd_client_list
); /* clients */
196 static DEFINE_SPINLOCK(rbd_client_list_lock
);
198 static int __rbd_init_snaps_header(struct rbd_device
*rbd_dev
);
199 static void rbd_dev_release(struct device
*dev
);
200 static ssize_t
rbd_snap_add(struct device
*dev
,
201 struct device_attribute
*attr
,
204 static void __rbd_remove_snap_dev(struct rbd_device
*rbd_dev
,
205 struct rbd_snap
*snap
);
207 static ssize_t
rbd_add(struct bus_type
*bus
, const char *buf
,
209 static ssize_t
rbd_remove(struct bus_type
*bus
, const char *buf
,
212 static struct bus_attribute rbd_bus_attrs
[] = {
213 __ATTR(add
, S_IWUSR
, NULL
, rbd_add
),
214 __ATTR(remove
, S_IWUSR
, NULL
, rbd_remove
),
218 static struct bus_type rbd_bus_type
= {
220 .bus_attrs
= rbd_bus_attrs
,
223 static void rbd_root_dev_release(struct device
*dev
)
227 static struct device rbd_root_dev
= {
229 .release
= rbd_root_dev_release
,
233 static struct device
*rbd_get_dev(struct rbd_device
*rbd_dev
)
235 return get_device(&rbd_dev
->dev
);
238 static void rbd_put_dev(struct rbd_device
*rbd_dev
)
240 put_device(&rbd_dev
->dev
);
243 static int __rbd_refresh_header(struct rbd_device
*rbd_dev
);
245 static int rbd_open(struct block_device
*bdev
, fmode_t mode
)
247 struct rbd_device
*rbd_dev
= bdev
->bd_disk
->private_data
;
249 rbd_get_dev(rbd_dev
);
251 set_device_ro(bdev
, rbd_dev
->read_only
);
253 if ((mode
& FMODE_WRITE
) && rbd_dev
->read_only
)
259 static int rbd_release(struct gendisk
*disk
, fmode_t mode
)
261 struct rbd_device
*rbd_dev
= disk
->private_data
;
263 rbd_put_dev(rbd_dev
);
268 static const struct block_device_operations rbd_bd_ops
= {
269 .owner
= THIS_MODULE
,
271 .release
= rbd_release
,
275 * Initialize an rbd client instance.
278 static struct rbd_client
*rbd_client_create(struct ceph_options
*opt
,
279 struct rbd_options
*rbd_opts
)
281 struct rbd_client
*rbdc
;
284 dout("rbd_client_create\n");
285 rbdc
= kmalloc(sizeof(struct rbd_client
), GFP_KERNEL
);
289 kref_init(&rbdc
->kref
);
290 INIT_LIST_HEAD(&rbdc
->node
);
292 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
294 rbdc
->client
= ceph_create_client(opt
, rbdc
, 0, 0);
295 if (IS_ERR(rbdc
->client
))
297 opt
= NULL
; /* Now rbdc->client is responsible for opt */
299 ret
= ceph_open_session(rbdc
->client
);
303 rbdc
->rbd_opts
= rbd_opts
;
305 spin_lock(&rbd_client_list_lock
);
306 list_add_tail(&rbdc
->node
, &rbd_client_list
);
307 spin_unlock(&rbd_client_list_lock
);
309 mutex_unlock(&ctl_mutex
);
311 dout("rbd_client_create created %p\n", rbdc
);
315 ceph_destroy_client(rbdc
->client
);
317 mutex_unlock(&ctl_mutex
);
321 ceph_destroy_options(opt
);
326 * Find a ceph client with specific addr and configuration.
328 static struct rbd_client
*__rbd_client_find(struct ceph_options
*opt
)
330 struct rbd_client
*client_node
;
332 if (opt
->flags
& CEPH_OPT_NOSHARE
)
335 list_for_each_entry(client_node
, &rbd_client_list
, node
)
336 if (ceph_compare_options(opt
, client_node
->client
) == 0)
349 /* string args above */
352 static match_table_t rbdopt_tokens
= {
353 {Opt_notify_timeout
, "notify_timeout=%d"},
355 /* string args above */
359 static int parse_rbd_opts_token(char *c
, void *private)
361 struct rbd_options
*rbdopt
= private;
362 substring_t argstr
[MAX_OPT_ARGS
];
363 int token
, intval
, ret
;
365 token
= match_token(c
, rbdopt_tokens
, argstr
);
369 if (token
< Opt_last_int
) {
370 ret
= match_int(&argstr
[0], &intval
);
372 pr_err("bad mount option arg (not int) "
376 dout("got int token %d val %d\n", token
, intval
);
377 } else if (token
> Opt_last_int
&& token
< Opt_last_string
) {
378 dout("got string token %d val %s\n", token
,
381 dout("got token %d\n", token
);
385 case Opt_notify_timeout
:
386 rbdopt
->notify_timeout
= intval
;
395 * Get a ceph client with specific addr and configuration, if one does
396 * not exist create it.
398 static struct rbd_client
*rbd_get_client(const char *mon_addr
,
402 struct rbd_client
*rbdc
;
403 struct ceph_options
*opt
;
404 struct rbd_options
*rbd_opts
;
406 rbd_opts
= kzalloc(sizeof(*rbd_opts
), GFP_KERNEL
);
408 return ERR_PTR(-ENOMEM
);
410 rbd_opts
->notify_timeout
= RBD_NOTIFY_TIMEOUT_DEFAULT
;
412 opt
= ceph_parse_options(options
, mon_addr
,
413 mon_addr
+ mon_addr_len
,
414 parse_rbd_opts_token
, rbd_opts
);
417 return ERR_CAST(opt
);
420 spin_lock(&rbd_client_list_lock
);
421 rbdc
= __rbd_client_find(opt
);
423 /* using an existing client */
424 kref_get(&rbdc
->kref
);
425 spin_unlock(&rbd_client_list_lock
);
427 ceph_destroy_options(opt
);
432 spin_unlock(&rbd_client_list_lock
);
434 rbdc
= rbd_client_create(opt
, rbd_opts
);
443 * Destroy ceph client
445 * Caller must hold rbd_client_list_lock.
447 static void rbd_client_release(struct kref
*kref
)
449 struct rbd_client
*rbdc
= container_of(kref
, struct rbd_client
, kref
);
451 dout("rbd_release_client %p\n", rbdc
);
452 spin_lock(&rbd_client_list_lock
);
453 list_del(&rbdc
->node
);
454 spin_unlock(&rbd_client_list_lock
);
456 ceph_destroy_client(rbdc
->client
);
457 kfree(rbdc
->rbd_opts
);
462 * Drop reference to ceph client node. If it's not referenced anymore, release
465 static void rbd_put_client(struct rbd_device
*rbd_dev
)
467 kref_put(&rbd_dev
->rbd_client
->kref
, rbd_client_release
);
468 rbd_dev
->rbd_client
= NULL
;
472 * Destroy requests collection
474 static void rbd_coll_release(struct kref
*kref
)
476 struct rbd_req_coll
*coll
=
477 container_of(kref
, struct rbd_req_coll
, kref
);
479 dout("rbd_coll_release %p\n", coll
);
484 * Create a new header structure, translate header format from the on-disk
487 static int rbd_header_from_disk(struct rbd_image_header
*header
,
488 struct rbd_image_header_ondisk
*ondisk
,
494 if (memcmp(ondisk
, RBD_HEADER_TEXT
, sizeof(RBD_HEADER_TEXT
)))
497 snap_count
= le32_to_cpu(ondisk
->snap_count
);
498 if (snap_count
> (UINT_MAX
- sizeof(struct ceph_snap_context
))
501 header
->snapc
= kmalloc(sizeof(struct ceph_snap_context
) +
502 snap_count
* sizeof(u64
),
507 header
->snap_names_len
= le64_to_cpu(ondisk
->snap_names_len
);
509 header
->snap_names
= kmalloc(header
->snap_names_len
,
511 if (!header
->snap_names
)
513 header
->snap_sizes
= kmalloc(snap_count
* sizeof(u64
),
515 if (!header
->snap_sizes
)
518 header
->snap_names
= NULL
;
519 header
->snap_sizes
= NULL
;
521 memcpy(header
->block_name
, ondisk
->block_name
,
522 sizeof(ondisk
->block_name
));
524 header
->image_size
= le64_to_cpu(ondisk
->image_size
);
525 header
->obj_order
= ondisk
->options
.order
;
526 header
->crypt_type
= ondisk
->options
.crypt_type
;
527 header
->comp_type
= ondisk
->options
.comp_type
;
529 atomic_set(&header
->snapc
->nref
, 1);
530 header
->snap_seq
= le64_to_cpu(ondisk
->snap_seq
);
531 header
->snapc
->num_snaps
= snap_count
;
532 header
->total_snaps
= snap_count
;
534 if (snap_count
&& allocated_snaps
== snap_count
) {
535 for (i
= 0; i
< snap_count
; i
++) {
536 header
->snapc
->snaps
[i
] =
537 le64_to_cpu(ondisk
->snaps
[i
].id
);
538 header
->snap_sizes
[i
] =
539 le64_to_cpu(ondisk
->snaps
[i
].image_size
);
542 /* copy snapshot names */
543 memcpy(header
->snap_names
, &ondisk
->snaps
[i
],
544 header
->snap_names_len
);
550 kfree(header
->snap_names
);
552 kfree(header
->snapc
);
556 static int snap_by_name(struct rbd_image_header
*header
, const char *snap_name
,
560 char *p
= header
->snap_names
;
562 for (i
= 0; i
< header
->total_snaps
; i
++) {
563 if (!strcmp(snap_name
, p
)) {
565 /* Found it. Pass back its id and/or size */
568 *seq
= header
->snapc
->snaps
[i
];
570 *size
= header
->snap_sizes
[i
];
573 p
+= strlen(p
) + 1; /* Skip ahead to the next name */
578 static int rbd_header_set_snap(struct rbd_device
*dev
, u64
*size
)
580 struct rbd_image_header
*header
= &dev
->header
;
581 struct ceph_snap_context
*snapc
= header
->snapc
;
584 BUILD_BUG_ON(sizeof (dev
->snap_name
) < sizeof (RBD_SNAP_HEAD_NAME
));
586 down_write(&dev
->header_rwsem
);
588 if (!memcmp(dev
->snap_name
, RBD_SNAP_HEAD_NAME
,
589 sizeof (RBD_SNAP_HEAD_NAME
))) {
590 if (header
->total_snaps
)
591 snapc
->seq
= header
->snap_seq
;
594 dev
->snap_id
= CEPH_NOSNAP
;
597 *size
= header
->image_size
;
599 ret
= snap_by_name(header
, dev
->snap_name
, &snapc
->seq
, size
);
602 dev
->snap_id
= snapc
->seq
;
608 up_write(&dev
->header_rwsem
);
612 static void rbd_header_free(struct rbd_image_header
*header
)
614 kfree(header
->snapc
);
615 kfree(header
->snap_names
);
616 kfree(header
->snap_sizes
);
620 * get the actual striped segment name, offset and length
622 static u64
rbd_get_segment(struct rbd_image_header
*header
,
623 const char *block_name
,
625 char *seg_name
, u64
*segofs
)
627 u64 seg
= ofs
>> header
->obj_order
;
630 snprintf(seg_name
, RBD_MAX_SEG_NAME_LEN
,
631 "%s.%012llx", block_name
, seg
);
633 ofs
= ofs
& ((1 << header
->obj_order
) - 1);
634 len
= min_t(u64
, len
, (1 << header
->obj_order
) - ofs
);
642 static int rbd_get_num_segments(struct rbd_image_header
*header
,
645 u64 start_seg
= ofs
>> header
->obj_order
;
646 u64 end_seg
= (ofs
+ len
- 1) >> header
->obj_order
;
647 return end_seg
- start_seg
+ 1;
651 * returns the size of an object in the image
653 static u64
rbd_obj_bytes(struct rbd_image_header
*header
)
655 return 1 << header
->obj_order
;
662 static void bio_chain_put(struct bio
*chain
)
668 chain
= chain
->bi_next
;
674 * zeros a bio chain, starting at specific offset
676 static void zero_bio_chain(struct bio
*chain
, int start_ofs
)
685 bio_for_each_segment(bv
, chain
, i
) {
686 if (pos
+ bv
->bv_len
> start_ofs
) {
687 int remainder
= max(start_ofs
- pos
, 0);
688 buf
= bvec_kmap_irq(bv
, &flags
);
689 memset(buf
+ remainder
, 0,
690 bv
->bv_len
- remainder
);
691 bvec_kunmap_irq(buf
, &flags
);
696 chain
= chain
->bi_next
;
701 * bio_chain_clone - clone a chain of bios up to a certain length.
702 * might return a bio_pair that will need to be released.
704 static struct bio
*bio_chain_clone(struct bio
**old
, struct bio
**next
,
705 struct bio_pair
**bp
,
706 int len
, gfp_t gfpmask
)
708 struct bio
*tmp
, *old_chain
= *old
, *new_chain
= NULL
, *tail
= NULL
;
712 bio_pair_release(*bp
);
716 while (old_chain
&& (total
< len
)) {
717 tmp
= bio_kmalloc(gfpmask
, old_chain
->bi_max_vecs
);
721 if (total
+ old_chain
->bi_size
> len
) {
725 * this split can only happen with a single paged bio,
726 * split_bio will BUG_ON if this is not the case
728 dout("bio_chain_clone split! total=%d remaining=%d"
730 (int)total
, (int)len
-total
,
731 (int)old_chain
->bi_size
);
733 /* split the bio. We'll release it either in the next
734 call, or it will have to be released outside */
735 bp
= bio_split(old_chain
, (len
- total
) / SECTOR_SIZE
);
739 __bio_clone(tmp
, &bp
->bio1
);
743 __bio_clone(tmp
, old_chain
);
744 *next
= old_chain
->bi_next
;
748 gfpmask
&= ~__GFP_WAIT
;
752 new_chain
= tail
= tmp
;
757 old_chain
= old_chain
->bi_next
;
759 total
+= tmp
->bi_size
;
765 tail
->bi_next
= NULL
;
772 dout("bio_chain_clone with err\n");
773 bio_chain_put(new_chain
);
778 * helpers for osd request op vectors.
780 static int rbd_create_rw_ops(struct ceph_osd_req_op
**ops
,
785 *ops
= kzalloc(sizeof(struct ceph_osd_req_op
) * (num_ops
+ 1),
789 (*ops
)[0].op
= opcode
;
791 * op extent offset and length will be set later on
792 * in calc_raw_layout()
794 (*ops
)[0].payload_len
= payload_len
;
798 static void rbd_destroy_ops(struct ceph_osd_req_op
*ops
)
803 static void rbd_coll_end_req_index(struct request
*rq
,
804 struct rbd_req_coll
*coll
,
808 struct request_queue
*q
;
811 dout("rbd_coll_end_req_index %p index %d ret %d len %lld\n",
812 coll
, index
, ret
, len
);
818 blk_end_request(rq
, ret
, len
);
824 spin_lock_irq(q
->queue_lock
);
825 coll
->status
[index
].done
= 1;
826 coll
->status
[index
].rc
= ret
;
827 coll
->status
[index
].bytes
= len
;
828 max
= min
= coll
->num_done
;
829 while (max
< coll
->total
&& coll
->status
[max
].done
)
832 for (i
= min
; i
<max
; i
++) {
833 __blk_end_request(rq
, coll
->status
[i
].rc
,
834 coll
->status
[i
].bytes
);
836 kref_put(&coll
->kref
, rbd_coll_release
);
838 spin_unlock_irq(q
->queue_lock
);
841 static void rbd_coll_end_req(struct rbd_request
*req
,
844 rbd_coll_end_req_index(req
->rq
, req
->coll
, req
->coll_index
, ret
, len
);
848 * Send ceph osd request
850 static int rbd_do_request(struct request
*rq
,
851 struct rbd_device
*dev
,
852 struct ceph_snap_context
*snapc
,
854 const char *obj
, u64 ofs
, u64 len
,
859 struct ceph_osd_req_op
*ops
,
861 struct rbd_req_coll
*coll
,
863 void (*rbd_cb
)(struct ceph_osd_request
*req
,
864 struct ceph_msg
*msg
),
865 struct ceph_osd_request
**linger_req
,
868 struct ceph_osd_request
*req
;
869 struct ceph_file_layout
*layout
;
872 struct timespec mtime
= CURRENT_TIME
;
873 struct rbd_request
*req_data
;
874 struct ceph_osd_request_head
*reqhead
;
875 struct ceph_osd_client
*osdc
;
877 req_data
= kzalloc(sizeof(*req_data
), GFP_NOIO
);
880 rbd_coll_end_req_index(rq
, coll
, coll_index
,
886 req_data
->coll
= coll
;
887 req_data
->coll_index
= coll_index
;
890 dout("rbd_do_request obj=%s ofs=%lld len=%lld\n", obj
, len
, ofs
);
892 down_read(&dev
->header_rwsem
);
894 osdc
= &dev
->rbd_client
->client
->osdc
;
895 req
= ceph_osdc_alloc_request(osdc
, flags
, snapc
, ops
,
896 false, GFP_NOIO
, pages
, bio
);
898 up_read(&dev
->header_rwsem
);
903 req
->r_callback
= rbd_cb
;
907 req_data
->pages
= pages
;
910 req
->r_priv
= req_data
;
912 reqhead
= req
->r_request
->front
.iov_base
;
913 reqhead
->snapid
= cpu_to_le64(CEPH_NOSNAP
);
915 strncpy(req
->r_oid
, obj
, sizeof(req
->r_oid
));
916 req
->r_oid_len
= strlen(req
->r_oid
);
918 layout
= &req
->r_file_layout
;
919 memset(layout
, 0, sizeof(*layout
));
920 layout
->fl_stripe_unit
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
921 layout
->fl_stripe_count
= cpu_to_le32(1);
922 layout
->fl_object_size
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
923 layout
->fl_pg_pool
= cpu_to_le32(dev
->poolid
);
924 ceph_calc_raw_layout(osdc
, layout
, snapid
, ofs
, &len
, &bno
,
927 ceph_osdc_build_request(req
, ofs
, &len
,
931 req
->r_oid
, req
->r_oid_len
);
932 up_read(&dev
->header_rwsem
);
935 ceph_osdc_set_request_linger(osdc
, req
);
939 ret
= ceph_osdc_start_request(osdc
, req
, false);
944 ret
= ceph_osdc_wait_request(osdc
, req
);
946 *ver
= le64_to_cpu(req
->r_reassert_version
.version
);
947 dout("reassert_ver=%lld\n",
948 le64_to_cpu(req
->r_reassert_version
.version
));
949 ceph_osdc_put_request(req
);
954 bio_chain_put(req_data
->bio
);
955 ceph_osdc_put_request(req
);
957 rbd_coll_end_req(req_data
, ret
, len
);
963 * Ceph osd op callback
965 static void rbd_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
967 struct rbd_request
*req_data
= req
->r_priv
;
968 struct ceph_osd_reply_head
*replyhead
;
969 struct ceph_osd_op
*op
;
975 replyhead
= msg
->front
.iov_base
;
976 WARN_ON(le32_to_cpu(replyhead
->num_ops
) == 0);
977 op
= (void *)(replyhead
+ 1);
978 rc
= le32_to_cpu(replyhead
->result
);
979 bytes
= le64_to_cpu(op
->extent
.length
);
980 read_op
= (le16_to_cpu(op
->op
) == CEPH_OSD_OP_READ
);
982 dout("rbd_req_cb bytes=%lld readop=%d rc=%d\n", bytes
, read_op
, rc
);
984 if (rc
== -ENOENT
&& read_op
) {
985 zero_bio_chain(req_data
->bio
, 0);
987 } else if (rc
== 0 && read_op
&& bytes
< req_data
->len
) {
988 zero_bio_chain(req_data
->bio
, bytes
);
989 bytes
= req_data
->len
;
992 rbd_coll_end_req(req_data
, rc
, bytes
);
995 bio_chain_put(req_data
->bio
);
997 ceph_osdc_put_request(req
);
1001 static void rbd_simple_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
1003 ceph_osdc_put_request(req
);
1007 * Do a synchronous ceph osd operation
1009 static int rbd_req_sync_op(struct rbd_device
*dev
,
1010 struct ceph_snap_context
*snapc
,
1014 struct ceph_osd_req_op
*orig_ops
,
1019 struct ceph_osd_request
**linger_req
,
1023 struct page
**pages
;
1025 struct ceph_osd_req_op
*ops
= orig_ops
;
1028 num_pages
= calc_pages_for(ofs
, len
);
1029 pages
= ceph_alloc_page_vector(num_pages
, GFP_KERNEL
);
1031 return PTR_ERR(pages
);
1034 payload_len
= (flags
& CEPH_OSD_FLAG_WRITE
? len
: 0);
1035 ret
= rbd_create_rw_ops(&ops
, 1, opcode
, payload_len
);
1039 if ((flags
& CEPH_OSD_FLAG_WRITE
) && buf
) {
1040 ret
= ceph_copy_to_page_vector(pages
, buf
, ofs
, len
);
1046 ret
= rbd_do_request(NULL
, dev
, snapc
, snapid
,
1047 obj
, ofs
, len
, NULL
,
1058 if ((flags
& CEPH_OSD_FLAG_READ
) && buf
)
1059 ret
= ceph_copy_from_page_vector(pages
, buf
, ofs
, ret
);
1063 rbd_destroy_ops(ops
);
1065 ceph_release_page_vector(pages
, num_pages
);
1070 * Do an asynchronous ceph osd operation
1072 static int rbd_do_op(struct request
*rq
,
1073 struct rbd_device
*rbd_dev
,
1074 struct ceph_snap_context
*snapc
,
1076 int opcode
, int flags
, int num_reply
,
1079 struct rbd_req_coll
*coll
,
1086 struct ceph_osd_req_op
*ops
;
1089 seg_name
= kmalloc(RBD_MAX_SEG_NAME_LEN
+ 1, GFP_NOIO
);
1093 seg_len
= rbd_get_segment(&rbd_dev
->header
,
1094 rbd_dev
->header
.block_name
,
1096 seg_name
, &seg_ofs
);
1098 payload_len
= (flags
& CEPH_OSD_FLAG_WRITE
? seg_len
: 0);
1100 ret
= rbd_create_rw_ops(&ops
, 1, opcode
, payload_len
);
1104 /* we've taken care of segment sizes earlier when we
1105 cloned the bios. We should never have a segment
1106 truncated at this point */
1107 BUG_ON(seg_len
< len
);
1109 ret
= rbd_do_request(rq
, rbd_dev
, snapc
, snapid
,
1110 seg_name
, seg_ofs
, seg_len
,
1117 rbd_req_cb
, 0, NULL
);
1119 rbd_destroy_ops(ops
);
1126 * Request async osd write
1128 static int rbd_req_write(struct request
*rq
,
1129 struct rbd_device
*rbd_dev
,
1130 struct ceph_snap_context
*snapc
,
1133 struct rbd_req_coll
*coll
,
1136 return rbd_do_op(rq
, rbd_dev
, snapc
, CEPH_NOSNAP
,
1138 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1140 ofs
, len
, bio
, coll
, coll_index
);
1144 * Request async osd read
1146 static int rbd_req_read(struct request
*rq
,
1147 struct rbd_device
*rbd_dev
,
1151 struct rbd_req_coll
*coll
,
1154 return rbd_do_op(rq
, rbd_dev
, NULL
,
1159 ofs
, len
, bio
, coll
, coll_index
);
1163 * Request sync osd read
1165 static int rbd_req_sync_read(struct rbd_device
*dev
,
1166 struct ceph_snap_context
*snapc
,
1173 return rbd_req_sync_op(dev
, NULL
,
1178 1, obj
, ofs
, len
, buf
, NULL
, ver
);
1182 * Request sync osd watch
1184 static int rbd_req_sync_notify_ack(struct rbd_device
*dev
,
1189 struct ceph_osd_req_op
*ops
;
1192 ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_NOTIFY_ACK
, 0);
1196 ops
[0].watch
.ver
= cpu_to_le64(dev
->header
.obj_version
);
1197 ops
[0].watch
.cookie
= notify_id
;
1198 ops
[0].watch
.flag
= 0;
1200 ret
= rbd_do_request(NULL
, dev
, NULL
, CEPH_NOSNAP
,
1207 rbd_simple_req_cb
, 0, NULL
);
1209 rbd_destroy_ops(ops
);
1213 static void rbd_watch_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1215 struct rbd_device
*dev
= (struct rbd_device
*)data
;
1221 dout("rbd_watch_cb %s notify_id=%lld opcode=%d\n", dev
->obj_md_name
,
1222 notify_id
, (int)opcode
);
1223 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1224 rc
= __rbd_refresh_header(dev
);
1225 mutex_unlock(&ctl_mutex
);
1227 pr_warning(RBD_DRV_NAME
"%d got notification but failed to "
1228 " update snaps: %d\n", dev
->major
, rc
);
1230 rbd_req_sync_notify_ack(dev
, ver
, notify_id
, dev
->obj_md_name
);
1234 * Request sync osd watch
1236 static int rbd_req_sync_watch(struct rbd_device
*dev
,
1240 struct ceph_osd_req_op
*ops
;
1241 struct ceph_osd_client
*osdc
= &dev
->rbd_client
->client
->osdc
;
1243 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_WATCH
, 0);
1247 ret
= ceph_osdc_create_event(osdc
, rbd_watch_cb
, 0,
1248 (void *)dev
, &dev
->watch_event
);
1252 ops
[0].watch
.ver
= cpu_to_le64(ver
);
1253 ops
[0].watch
.cookie
= cpu_to_le64(dev
->watch_event
->cookie
);
1254 ops
[0].watch
.flag
= 1;
1256 ret
= rbd_req_sync_op(dev
, NULL
,
1259 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1262 &dev
->watch_request
, NULL
);
1267 rbd_destroy_ops(ops
);
1271 ceph_osdc_cancel_event(dev
->watch_event
);
1272 dev
->watch_event
= NULL
;
1274 rbd_destroy_ops(ops
);
1279 * Request sync osd unwatch
1281 static int rbd_req_sync_unwatch(struct rbd_device
*dev
,
1284 struct ceph_osd_req_op
*ops
;
1286 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_WATCH
, 0);
1290 ops
[0].watch
.ver
= 0;
1291 ops
[0].watch
.cookie
= cpu_to_le64(dev
->watch_event
->cookie
);
1292 ops
[0].watch
.flag
= 0;
1294 ret
= rbd_req_sync_op(dev
, NULL
,
1297 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1299 1, obj
, 0, 0, NULL
, NULL
, NULL
);
1301 rbd_destroy_ops(ops
);
1302 ceph_osdc_cancel_event(dev
->watch_event
);
1303 dev
->watch_event
= NULL
;
1307 struct rbd_notify_info
{
1308 struct rbd_device
*dev
;
1311 static void rbd_notify_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1313 struct rbd_device
*dev
= (struct rbd_device
*)data
;
1317 dout("rbd_notify_cb %s notify_id=%lld opcode=%d\n", dev
->obj_md_name
,
1318 notify_id
, (int)opcode
);
1322 * Request sync osd notify
1324 static int rbd_req_sync_notify(struct rbd_device
*dev
,
1327 struct ceph_osd_req_op
*ops
;
1328 struct ceph_osd_client
*osdc
= &dev
->rbd_client
->client
->osdc
;
1329 struct ceph_osd_event
*event
;
1330 struct rbd_notify_info info
;
1331 int payload_len
= sizeof(u32
) + sizeof(u32
);
1334 ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_NOTIFY
, payload_len
);
1340 ret
= ceph_osdc_create_event(osdc
, rbd_notify_cb
, 1,
1341 (void *)&info
, &event
);
1345 ops
[0].watch
.ver
= 1;
1346 ops
[0].watch
.flag
= 1;
1347 ops
[0].watch
.cookie
= event
->cookie
;
1348 ops
[0].watch
.prot_ver
= RADOS_NOTIFY_VER
;
1349 ops
[0].watch
.timeout
= 12;
1351 ret
= rbd_req_sync_op(dev
, NULL
,
1354 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1356 1, obj
, 0, 0, NULL
, NULL
, NULL
);
1360 ret
= ceph_osdc_wait_event(event
, CEPH_OSD_TIMEOUT_DEFAULT
);
1361 dout("ceph_osdc_wait_event returned %d\n", ret
);
1362 rbd_destroy_ops(ops
);
1366 ceph_osdc_cancel_event(event
);
1368 rbd_destroy_ops(ops
);
1373 * Request sync osd read
1375 static int rbd_req_sync_exec(struct rbd_device
*dev
,
1383 struct ceph_osd_req_op
*ops
;
1384 int cls_len
= strlen(cls
);
1385 int method_len
= strlen(method
);
1386 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_CALL
,
1387 cls_len
+ method_len
+ len
);
1391 ops
[0].cls
.class_name
= cls
;
1392 ops
[0].cls
.class_len
= (__u8
)cls_len
;
1393 ops
[0].cls
.method_name
= method
;
1394 ops
[0].cls
.method_len
= (__u8
)method_len
;
1395 ops
[0].cls
.argc
= 0;
1396 ops
[0].cls
.indata
= data
;
1397 ops
[0].cls
.indata_len
= len
;
1399 ret
= rbd_req_sync_op(dev
, NULL
,
1402 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1404 1, obj
, 0, 0, NULL
, NULL
, ver
);
1406 rbd_destroy_ops(ops
);
1408 dout("cls_exec returned %d\n", ret
);
1412 static struct rbd_req_coll
*rbd_alloc_coll(int num_reqs
)
1414 struct rbd_req_coll
*coll
=
1415 kzalloc(sizeof(struct rbd_req_coll
) +
1416 sizeof(struct rbd_req_status
) * num_reqs
,
1421 coll
->total
= num_reqs
;
1422 kref_init(&coll
->kref
);
1427 * block device queue callback
1429 static void rbd_rq_fn(struct request_queue
*q
)
1431 struct rbd_device
*rbd_dev
= q
->queuedata
;
1433 struct bio_pair
*bp
= NULL
;
1435 while ((rq
= blk_fetch_request(q
))) {
1437 struct bio
*rq_bio
, *next_bio
= NULL
;
1439 int size
, op_size
= 0;
1441 int num_segs
, cur_seg
= 0;
1442 struct rbd_req_coll
*coll
;
1444 /* peek at request from block layer */
1448 dout("fetched request\n");
1450 /* filter out block requests we don't understand */
1451 if ((rq
->cmd_type
!= REQ_TYPE_FS
)) {
1452 __blk_end_request_all(rq
, 0);
1456 /* deduce our operation (read, write) */
1457 do_write
= (rq_data_dir(rq
) == WRITE
);
1459 size
= blk_rq_bytes(rq
);
1460 ofs
= blk_rq_pos(rq
) * SECTOR_SIZE
;
1462 if (do_write
&& rbd_dev
->read_only
) {
1463 __blk_end_request_all(rq
, -EROFS
);
1467 spin_unlock_irq(q
->queue_lock
);
1469 dout("%s 0x%x bytes at 0x%llx\n",
1470 do_write
? "write" : "read",
1471 size
, blk_rq_pos(rq
) * SECTOR_SIZE
);
1473 num_segs
= rbd_get_num_segments(&rbd_dev
->header
, ofs
, size
);
1474 coll
= rbd_alloc_coll(num_segs
);
1476 spin_lock_irq(q
->queue_lock
);
1477 __blk_end_request_all(rq
, -ENOMEM
);
1482 /* a bio clone to be passed down to OSD req */
1483 dout("rq->bio->bi_vcnt=%d\n", rq
->bio
->bi_vcnt
);
1484 op_size
= rbd_get_segment(&rbd_dev
->header
,
1485 rbd_dev
->header
.block_name
,
1488 kref_get(&coll
->kref
);
1489 bio
= bio_chain_clone(&rq_bio
, &next_bio
, &bp
,
1490 op_size
, GFP_ATOMIC
);
1492 rbd_coll_end_req_index(rq
, coll
, cur_seg
,
1498 /* init OSD command: write or read */
1500 rbd_req_write(rq
, rbd_dev
,
1501 rbd_dev
->header
.snapc
,
1506 rbd_req_read(rq
, rbd_dev
,
1519 kref_put(&coll
->kref
, rbd_coll_release
);
1522 bio_pair_release(bp
);
1523 spin_lock_irq(q
->queue_lock
);
1528 * a queue callback. Makes sure that we don't create a bio that spans across
1529 * multiple osd objects. One exception would be with a single page bios,
1530 * which we handle later at bio_chain_clone
1532 static int rbd_merge_bvec(struct request_queue
*q
, struct bvec_merge_data
*bmd
,
1533 struct bio_vec
*bvec
)
1535 struct rbd_device
*rbd_dev
= q
->queuedata
;
1536 unsigned int chunk_sectors
;
1538 unsigned int bio_sectors
;
1541 chunk_sectors
= 1 << (rbd_dev
->header
.obj_order
- SECTOR_SHIFT
);
1542 sector
= bmd
->bi_sector
+ get_start_sect(bmd
->bi_bdev
);
1543 bio_sectors
= bmd
->bi_size
>> SECTOR_SHIFT
;
1545 max
= (chunk_sectors
- ((sector
& (chunk_sectors
- 1))
1546 + bio_sectors
)) << SECTOR_SHIFT
;
1548 max
= 0; /* bio_add cannot handle a negative return */
1549 if (max
<= bvec
->bv_len
&& bio_sectors
== 0)
1550 return bvec
->bv_len
;
1554 static void rbd_free_disk(struct rbd_device
*rbd_dev
)
1556 struct gendisk
*disk
= rbd_dev
->disk
;
1561 rbd_header_free(&rbd_dev
->header
);
1563 if (disk
->flags
& GENHD_FL_UP
)
1566 blk_cleanup_queue(disk
->queue
);
1571 * reload the ondisk the header
1573 static int rbd_read_header(struct rbd_device
*rbd_dev
,
1574 struct rbd_image_header
*header
)
1577 struct rbd_image_header_ondisk
*dh
;
1583 * First reads the fixed-size header to determine the number
1584 * of snapshots, then re-reads it, along with all snapshot
1585 * records as well as their stored names.
1589 dh
= kmalloc(len
, GFP_KERNEL
);
1593 rc
= rbd_req_sync_read(rbd_dev
,
1595 rbd_dev
->obj_md_name
,
1601 rc
= rbd_header_from_disk(header
, dh
, snap_count
, GFP_KERNEL
);
1604 pr_warning("unrecognized header format"
1605 " for image %s", rbd_dev
->obj
);
1609 if (snap_count
== header
->total_snaps
)
1612 snap_count
= header
->total_snaps
;
1613 len
= sizeof (*dh
) +
1614 snap_count
* sizeof(struct rbd_image_snap_ondisk
) +
1615 header
->snap_names_len
;
1617 rbd_header_free(header
);
1620 header
->obj_version
= ver
;
1630 static int rbd_header_add_snap(struct rbd_device
*dev
,
1631 const char *snap_name
,
1634 int name_len
= strlen(snap_name
);
1639 struct ceph_mon_client
*monc
;
1641 /* we should create a snapshot only if we're pointing at the head */
1642 if (dev
->snap_id
!= CEPH_NOSNAP
)
1645 monc
= &dev
->rbd_client
->client
->monc
;
1646 ret
= ceph_monc_create_snapid(monc
, dev
->poolid
, &new_snapid
);
1647 dout("created snapid=%lld\n", new_snapid
);
1651 data
= kmalloc(name_len
+ 16, gfp_flags
);
1656 e
= data
+ name_len
+ 16;
1658 ceph_encode_string_safe(&p
, e
, snap_name
, name_len
, bad
);
1659 ceph_encode_64_safe(&p
, e
, new_snapid
, bad
);
1661 ret
= rbd_req_sync_exec(dev
, dev
->obj_md_name
, "rbd", "snap_add",
1662 data
, p
- data
, &ver
);
1669 down_write(&dev
->header_rwsem
);
1670 dev
->header
.snapc
->seq
= new_snapid
;
1671 up_write(&dev
->header_rwsem
);
1678 static void __rbd_remove_all_snaps(struct rbd_device
*rbd_dev
)
1680 struct rbd_snap
*snap
;
1682 while (!list_empty(&rbd_dev
->snaps
)) {
1683 snap
= list_first_entry(&rbd_dev
->snaps
, struct rbd_snap
, node
);
1684 __rbd_remove_snap_dev(rbd_dev
, snap
);
1689 * only read the first part of the ondisk header, without the snaps info
1691 static int __rbd_refresh_header(struct rbd_device
*rbd_dev
)
1694 struct rbd_image_header h
;
1698 ret
= rbd_read_header(rbd_dev
, &h
);
1703 set_capacity(rbd_dev
->disk
, h
.image_size
/ SECTOR_SIZE
);
1705 down_write(&rbd_dev
->header_rwsem
);
1707 snap_seq
= rbd_dev
->header
.snapc
->seq
;
1708 if (rbd_dev
->header
.total_snaps
&&
1709 rbd_dev
->header
.snapc
->snaps
[0] == snap_seq
)
1710 /* pointing at the head, will need to follow that
1714 kfree(rbd_dev
->header
.snapc
);
1715 kfree(rbd_dev
->header
.snap_names
);
1716 kfree(rbd_dev
->header
.snap_sizes
);
1718 rbd_dev
->header
.total_snaps
= h
.total_snaps
;
1719 rbd_dev
->header
.snapc
= h
.snapc
;
1720 rbd_dev
->header
.snap_names
= h
.snap_names
;
1721 rbd_dev
->header
.snap_names_len
= h
.snap_names_len
;
1722 rbd_dev
->header
.snap_sizes
= h
.snap_sizes
;
1724 rbd_dev
->header
.snapc
->seq
= rbd_dev
->header
.snapc
->snaps
[0];
1726 rbd_dev
->header
.snapc
->seq
= snap_seq
;
1728 ret
= __rbd_init_snaps_header(rbd_dev
);
1730 up_write(&rbd_dev
->header_rwsem
);
1735 static int rbd_init_disk(struct rbd_device
*rbd_dev
)
1737 struct gendisk
*disk
;
1738 struct request_queue
*q
;
1743 /* contact OSD, request size info about the object being mapped */
1744 rc
= rbd_read_header(rbd_dev
, &rbd_dev
->header
);
1748 /* no need to lock here, as rbd_dev is not registered yet */
1749 rc
= __rbd_init_snaps_header(rbd_dev
);
1753 rc
= rbd_header_set_snap(rbd_dev
, &total_size
);
1757 /* create gendisk info */
1759 disk
= alloc_disk(RBD_MINORS_PER_MAJOR
);
1763 snprintf(disk
->disk_name
, sizeof(disk
->disk_name
), RBD_DRV_NAME
"%d",
1765 disk
->major
= rbd_dev
->major
;
1766 disk
->first_minor
= 0;
1767 disk
->fops
= &rbd_bd_ops
;
1768 disk
->private_data
= rbd_dev
;
1772 q
= blk_init_queue(rbd_rq_fn
, &rbd_dev
->lock
);
1776 /* We use the default size, but let's be explicit about it. */
1777 blk_queue_physical_block_size(q
, SECTOR_SIZE
);
1779 /* set io sizes to object size */
1780 segment_size
= rbd_obj_bytes(&rbd_dev
->header
);
1781 blk_queue_max_hw_sectors(q
, segment_size
/ SECTOR_SIZE
);
1782 blk_queue_max_segment_size(q
, segment_size
);
1783 blk_queue_io_min(q
, segment_size
);
1784 blk_queue_io_opt(q
, segment_size
);
1786 blk_queue_merge_bvec(q
, rbd_merge_bvec
);
1789 q
->queuedata
= rbd_dev
;
1791 rbd_dev
->disk
= disk
;
1794 /* finally, announce the disk to the world */
1795 set_capacity(disk
, total_size
/ SECTOR_SIZE
);
1798 pr_info("%s: added with size 0x%llx\n",
1799 disk
->disk_name
, (unsigned long long)total_size
);
1812 static struct rbd_device
*dev_to_rbd_dev(struct device
*dev
)
1814 return container_of(dev
, struct rbd_device
, dev
);
1817 static ssize_t
rbd_size_show(struct device
*dev
,
1818 struct device_attribute
*attr
, char *buf
)
1820 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1822 return sprintf(buf
, "%llu\n", (unsigned long long)rbd_dev
->header
.image_size
);
1825 static ssize_t
rbd_major_show(struct device
*dev
,
1826 struct device_attribute
*attr
, char *buf
)
1828 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1830 return sprintf(buf
, "%d\n", rbd_dev
->major
);
1833 static ssize_t
rbd_client_id_show(struct device
*dev
,
1834 struct device_attribute
*attr
, char *buf
)
1836 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1838 return sprintf(buf
, "client%lld\n",
1839 ceph_client_id(rbd_dev
->rbd_client
->client
));
1842 static ssize_t
rbd_pool_show(struct device
*dev
,
1843 struct device_attribute
*attr
, char *buf
)
1845 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1847 return sprintf(buf
, "%s\n", rbd_dev
->pool_name
);
1850 static ssize_t
rbd_name_show(struct device
*dev
,
1851 struct device_attribute
*attr
, char *buf
)
1853 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1855 return sprintf(buf
, "%s\n", rbd_dev
->obj
);
1858 static ssize_t
rbd_snap_show(struct device
*dev
,
1859 struct device_attribute
*attr
,
1862 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1864 return sprintf(buf
, "%s\n", rbd_dev
->snap_name
);
1867 static ssize_t
rbd_image_refresh(struct device
*dev
,
1868 struct device_attribute
*attr
,
1872 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
1876 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1878 rc
= __rbd_refresh_header(rbd_dev
);
1882 mutex_unlock(&ctl_mutex
);
1886 static DEVICE_ATTR(size
, S_IRUGO
, rbd_size_show
, NULL
);
1887 static DEVICE_ATTR(major
, S_IRUGO
, rbd_major_show
, NULL
);
1888 static DEVICE_ATTR(client_id
, S_IRUGO
, rbd_client_id_show
, NULL
);
1889 static DEVICE_ATTR(pool
, S_IRUGO
, rbd_pool_show
, NULL
);
1890 static DEVICE_ATTR(name
, S_IRUGO
, rbd_name_show
, NULL
);
1891 static DEVICE_ATTR(refresh
, S_IWUSR
, NULL
, rbd_image_refresh
);
1892 static DEVICE_ATTR(current_snap
, S_IRUGO
, rbd_snap_show
, NULL
);
1893 static DEVICE_ATTR(create_snap
, S_IWUSR
, NULL
, rbd_snap_add
);
1895 static struct attribute
*rbd_attrs
[] = {
1896 &dev_attr_size
.attr
,
1897 &dev_attr_major
.attr
,
1898 &dev_attr_client_id
.attr
,
1899 &dev_attr_pool
.attr
,
1900 &dev_attr_name
.attr
,
1901 &dev_attr_current_snap
.attr
,
1902 &dev_attr_refresh
.attr
,
1903 &dev_attr_create_snap
.attr
,
1907 static struct attribute_group rbd_attr_group
= {
1911 static const struct attribute_group
*rbd_attr_groups
[] = {
1916 static void rbd_sysfs_dev_release(struct device
*dev
)
1920 static struct device_type rbd_device_type
= {
1922 .groups
= rbd_attr_groups
,
1923 .release
= rbd_sysfs_dev_release
,
1931 static ssize_t
rbd_snap_size_show(struct device
*dev
,
1932 struct device_attribute
*attr
,
1935 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1937 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->size
);
1940 static ssize_t
rbd_snap_id_show(struct device
*dev
,
1941 struct device_attribute
*attr
,
1944 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1946 return sprintf(buf
, "%llu\n", (unsigned long long)snap
->id
);
1949 static DEVICE_ATTR(snap_size
, S_IRUGO
, rbd_snap_size_show
, NULL
);
1950 static DEVICE_ATTR(snap_id
, S_IRUGO
, rbd_snap_id_show
, NULL
);
1952 static struct attribute
*rbd_snap_attrs
[] = {
1953 &dev_attr_snap_size
.attr
,
1954 &dev_attr_snap_id
.attr
,
1958 static struct attribute_group rbd_snap_attr_group
= {
1959 .attrs
= rbd_snap_attrs
,
1962 static void rbd_snap_dev_release(struct device
*dev
)
1964 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1969 static const struct attribute_group
*rbd_snap_attr_groups
[] = {
1970 &rbd_snap_attr_group
,
1974 static struct device_type rbd_snap_device_type
= {
1975 .groups
= rbd_snap_attr_groups
,
1976 .release
= rbd_snap_dev_release
,
1979 static void __rbd_remove_snap_dev(struct rbd_device
*rbd_dev
,
1980 struct rbd_snap
*snap
)
1982 list_del(&snap
->node
);
1983 device_unregister(&snap
->dev
);
1986 static int rbd_register_snap_dev(struct rbd_device
*rbd_dev
,
1987 struct rbd_snap
*snap
,
1988 struct device
*parent
)
1990 struct device
*dev
= &snap
->dev
;
1993 dev
->type
= &rbd_snap_device_type
;
1994 dev
->parent
= parent
;
1995 dev
->release
= rbd_snap_dev_release
;
1996 dev_set_name(dev
, "snap_%s", snap
->name
);
1997 ret
= device_register(dev
);
2002 static int __rbd_add_snap_dev(struct rbd_device
*rbd_dev
,
2003 int i
, const char *name
,
2004 struct rbd_snap
**snapp
)
2007 struct rbd_snap
*snap
= kzalloc(sizeof(*snap
), GFP_KERNEL
);
2010 snap
->name
= kstrdup(name
, GFP_KERNEL
);
2011 snap
->size
= rbd_dev
->header
.snap_sizes
[i
];
2012 snap
->id
= rbd_dev
->header
.snapc
->snaps
[i
];
2013 if (device_is_registered(&rbd_dev
->dev
)) {
2014 ret
= rbd_register_snap_dev(rbd_dev
, snap
,
2028 * search for the previous snap in a null delimited string list
2030 const char *rbd_prev_snap_name(const char *name
, const char *start
)
2032 if (name
< start
+ 2)
2045 * compare the old list of snapshots that we have to what's in the header
2046 * and update it accordingly. Note that the header holds the snapshots
2047 * in a reverse order (from newest to oldest) and we need to go from
2048 * older to new so that we don't get a duplicate snap name when
2049 * doing the process (e.g., removed snapshot and recreated a new
2050 * one with the same name.
2052 static int __rbd_init_snaps_header(struct rbd_device
*rbd_dev
)
2054 const char *name
, *first_name
;
2055 int i
= rbd_dev
->header
.total_snaps
;
2056 struct rbd_snap
*snap
, *old_snap
= NULL
;
2058 struct list_head
*p
, *n
;
2060 first_name
= rbd_dev
->header
.snap_names
;
2061 name
= first_name
+ rbd_dev
->header
.snap_names_len
;
2063 list_for_each_prev_safe(p
, n
, &rbd_dev
->snaps
) {
2066 old_snap
= list_entry(p
, struct rbd_snap
, node
);
2069 cur_id
= rbd_dev
->header
.snapc
->snaps
[i
- 1];
2071 if (!i
|| old_snap
->id
< cur_id
) {
2072 /* old_snap->id was skipped, thus was removed */
2073 __rbd_remove_snap_dev(rbd_dev
, old_snap
);
2076 if (old_snap
->id
== cur_id
) {
2077 /* we have this snapshot already */
2079 name
= rbd_prev_snap_name(name
, first_name
);
2083 i
--, name
= rbd_prev_snap_name(name
, first_name
)) {
2088 cur_id
= rbd_dev
->header
.snapc
->snaps
[i
];
2089 /* snapshot removal? handle it above */
2090 if (cur_id
>= old_snap
->id
)
2092 /* a new snapshot */
2093 ret
= __rbd_add_snap_dev(rbd_dev
, i
- 1, name
, &snap
);
2097 /* note that we add it backward so using n and not p */
2098 list_add(&snap
->node
, n
);
2102 /* we're done going over the old snap list, just add what's left */
2103 for (; i
> 0; i
--) {
2104 name
= rbd_prev_snap_name(name
, first_name
);
2109 ret
= __rbd_add_snap_dev(rbd_dev
, i
- 1, name
, &snap
);
2112 list_add(&snap
->node
, &rbd_dev
->snaps
);
2118 static int rbd_bus_add_dev(struct rbd_device
*rbd_dev
)
2122 struct rbd_snap
*snap
;
2124 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2125 dev
= &rbd_dev
->dev
;
2127 dev
->bus
= &rbd_bus_type
;
2128 dev
->type
= &rbd_device_type
;
2129 dev
->parent
= &rbd_root_dev
;
2130 dev
->release
= rbd_dev_release
;
2131 dev_set_name(dev
, "%d", rbd_dev
->id
);
2132 ret
= device_register(dev
);
2136 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
2137 ret
= rbd_register_snap_dev(rbd_dev
, snap
,
2143 mutex_unlock(&ctl_mutex
);
2147 static void rbd_bus_del_dev(struct rbd_device
*rbd_dev
)
2149 device_unregister(&rbd_dev
->dev
);
2152 static int rbd_init_watch_dev(struct rbd_device
*rbd_dev
)
2157 ret
= rbd_req_sync_watch(rbd_dev
, rbd_dev
->obj_md_name
,
2158 rbd_dev
->header
.obj_version
);
2159 if (ret
== -ERANGE
) {
2160 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2161 rc
= __rbd_refresh_header(rbd_dev
);
2162 mutex_unlock(&ctl_mutex
);
2166 } while (ret
== -ERANGE
);
2171 static atomic64_t rbd_id_max
= ATOMIC64_INIT(0);
2174 * Get a unique rbd identifier for the given new rbd_dev, and add
2175 * the rbd_dev to the global list. The minimum rbd id is 1.
2177 static void rbd_id_get(struct rbd_device
*rbd_dev
)
2179 rbd_dev
->id
= atomic64_inc_return(&rbd_id_max
);
2181 spin_lock(&rbd_dev_list_lock
);
2182 list_add_tail(&rbd_dev
->node
, &rbd_dev_list
);
2183 spin_unlock(&rbd_dev_list_lock
);
2187 * Remove an rbd_dev from the global list, and record that its
2188 * identifier is no longer in use.
2190 static void rbd_id_put(struct rbd_device
*rbd_dev
)
2192 struct list_head
*tmp
;
2193 int rbd_id
= rbd_dev
->id
;
2198 spin_lock(&rbd_dev_list_lock
);
2199 list_del_init(&rbd_dev
->node
);
2202 * If the id being "put" is not the current maximum, there
2203 * is nothing special we need to do.
2205 if (rbd_id
!= atomic64_read(&rbd_id_max
)) {
2206 spin_unlock(&rbd_dev_list_lock
);
2211 * We need to update the current maximum id. Search the
2212 * list to find out what it is. We're more likely to find
2213 * the maximum at the end, so search the list backward.
2216 list_for_each_prev(tmp
, &rbd_dev_list
) {
2217 struct rbd_device
*rbd_dev
;
2219 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
2220 if (rbd_id
> max_id
)
2223 spin_unlock(&rbd_dev_list_lock
);
2226 * The max id could have been updated by rbd_id_get(), in
2227 * which case it now accurately reflects the new maximum.
2228 * Be careful not to overwrite the maximum value in that
2231 atomic64_cmpxchg(&rbd_id_max
, rbd_id
, max_id
);
2235 * Skips over white space at *buf, and updates *buf to point to the
2236 * first found non-space character (if any). Returns the length of
2237 * the token (string of non-white space characters) found. Note
2238 * that *buf must be terminated with '\0'.
2240 static inline size_t next_token(const char **buf
)
2243 * These are the characters that produce nonzero for
2244 * isspace() in the "C" and "POSIX" locales.
2246 const char *spaces
= " \f\n\r\t\v";
2248 *buf
+= strspn(*buf
, spaces
); /* Find start of token */
2250 return strcspn(*buf
, spaces
); /* Return token length */
2254 * Finds the next token in *buf, and if the provided token buffer is
2255 * big enough, copies the found token into it. The result, if
2256 * copied, is guaranteed to be terminated with '\0'. Note that *buf
2257 * must be terminated with '\0' on entry.
2259 * Returns the length of the token found (not including the '\0').
2260 * Return value will be 0 if no token is found, and it will be >=
2261 * token_size if the token would not fit.
2263 * The *buf pointer will be updated to point beyond the end of the
2264 * found token. Note that this occurs even if the token buffer is
2265 * too small to hold it.
2267 static inline size_t copy_token(const char **buf
,
2273 len
= next_token(buf
);
2274 if (len
< token_size
) {
2275 memcpy(token
, *buf
, len
);
2276 *(token
+ len
) = '\0';
2284 * This fills in the pool_name, obj, obj_len, snap_name, obj_len,
2285 * rbd_dev, rbd_md_name, and name fields of the given rbd_dev, based
2286 * on the list of monitor addresses and other options provided via
2289 static int rbd_add_parse_args(struct rbd_device
*rbd_dev
,
2291 const char **mon_addrs
,
2292 size_t *mon_addrs_size
,
2294 size_t options_size
)
2298 /* The first four tokens are required */
2300 len
= next_token(&buf
);
2303 *mon_addrs_size
= len
+ 1;
2308 len
= copy_token(&buf
, options
, options_size
);
2309 if (!len
|| len
>= options_size
)
2312 len
= copy_token(&buf
, rbd_dev
->pool_name
, sizeof (rbd_dev
->pool_name
));
2313 if (!len
|| len
>= sizeof (rbd_dev
->pool_name
))
2316 len
= copy_token(&buf
, rbd_dev
->obj
, sizeof (rbd_dev
->obj
));
2317 if (!len
|| len
>= sizeof (rbd_dev
->obj
))
2320 /* We have the object length in hand, save it. */
2322 rbd_dev
->obj_len
= len
;
2324 BUILD_BUG_ON(RBD_MAX_MD_NAME_LEN
2325 < RBD_MAX_OBJ_NAME_LEN
+ sizeof (RBD_SUFFIX
));
2326 sprintf(rbd_dev
->obj_md_name
, "%s%s", rbd_dev
->obj
, RBD_SUFFIX
);
2329 * The snapshot name is optional, but it's an error if it's
2330 * too long. If no snapshot is supplied, fill in the default.
2332 len
= copy_token(&buf
, rbd_dev
->snap_name
, sizeof (rbd_dev
->snap_name
));
2334 memcpy(rbd_dev
->snap_name
, RBD_SNAP_HEAD_NAME
,
2335 sizeof (RBD_SNAP_HEAD_NAME
));
2336 else if (len
>= sizeof (rbd_dev
->snap_name
))
2342 static ssize_t
rbd_add(struct bus_type
*bus
,
2346 struct rbd_device
*rbd_dev
;
2347 const char *mon_addrs
= NULL
;
2348 size_t mon_addrs_size
= 0;
2349 char *options
= NULL
;
2350 struct ceph_osd_client
*osdc
;
2353 if (!try_module_get(THIS_MODULE
))
2356 rbd_dev
= kzalloc(sizeof(*rbd_dev
), GFP_KERNEL
);
2359 options
= kmalloc(count
, GFP_KERNEL
);
2363 /* static rbd_device initialization */
2364 spin_lock_init(&rbd_dev
->lock
);
2365 INIT_LIST_HEAD(&rbd_dev
->node
);
2366 INIT_LIST_HEAD(&rbd_dev
->snaps
);
2367 init_rwsem(&rbd_dev
->header_rwsem
);
2369 init_rwsem(&rbd_dev
->header_rwsem
);
2371 /* generate unique id: find highest unique id, add one */
2372 rbd_id_get(rbd_dev
);
2374 /* Fill in the device name, now that we have its id. */
2375 BUILD_BUG_ON(DEV_NAME_LEN
2376 < sizeof (RBD_DRV_NAME
) + MAX_INT_FORMAT_WIDTH
);
2377 sprintf(rbd_dev
->name
, "%s%d", RBD_DRV_NAME
, rbd_dev
->id
);
2379 /* parse add command */
2380 rc
= rbd_add_parse_args(rbd_dev
, buf
, &mon_addrs
, &mon_addrs_size
,
2385 rbd_dev
->rbd_client
= rbd_get_client(mon_addrs
, mon_addrs_size
- 1,
2387 if (IS_ERR(rbd_dev
->rbd_client
)) {
2388 rc
= PTR_ERR(rbd_dev
->rbd_client
);
2393 osdc
= &rbd_dev
->rbd_client
->client
->osdc
;
2394 rc
= ceph_pg_poolid_by_name(osdc
->osdmap
, rbd_dev
->pool_name
);
2396 goto err_out_client
;
2397 rbd_dev
->poolid
= rc
;
2399 /* register our block device */
2400 rc
= register_blkdev(0, rbd_dev
->name
);
2402 goto err_out_client
;
2403 rbd_dev
->major
= rc
;
2405 rc
= rbd_bus_add_dev(rbd_dev
);
2407 goto err_out_blkdev
;
2410 * At this point cleanup in the event of an error is the job
2411 * of the sysfs code (initiated by rbd_bus_del_dev()).
2413 * Set up and announce blkdev mapping.
2415 rc
= rbd_init_disk(rbd_dev
);
2419 rc
= rbd_init_watch_dev(rbd_dev
);
2426 /* this will also clean up rest of rbd_dev stuff */
2428 rbd_bus_del_dev(rbd_dev
);
2433 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
2435 rbd_put_client(rbd_dev
);
2437 rbd_id_put(rbd_dev
);
2442 dout("Error adding device %s\n", buf
);
2443 module_put(THIS_MODULE
);
2445 return (ssize_t
) rc
;
2448 static struct rbd_device
*__rbd_get_dev(unsigned long id
)
2450 struct list_head
*tmp
;
2451 struct rbd_device
*rbd_dev
;
2453 spin_lock(&rbd_dev_list_lock
);
2454 list_for_each(tmp
, &rbd_dev_list
) {
2455 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
2456 if (rbd_dev
->id
== id
) {
2457 spin_unlock(&rbd_dev_list_lock
);
2461 spin_unlock(&rbd_dev_list_lock
);
2465 static void rbd_dev_release(struct device
*dev
)
2467 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2469 if (rbd_dev
->watch_request
) {
2470 struct ceph_client
*client
= rbd_dev
->rbd_client
->client
;
2472 ceph_osdc_unregister_linger_request(&client
->osdc
,
2473 rbd_dev
->watch_request
);
2475 if (rbd_dev
->watch_event
)
2476 rbd_req_sync_unwatch(rbd_dev
, rbd_dev
->obj_md_name
);
2478 rbd_put_client(rbd_dev
);
2480 /* clean up and free blkdev */
2481 rbd_free_disk(rbd_dev
);
2482 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
2484 /* done with the id, and with the rbd_dev */
2485 rbd_id_put(rbd_dev
);
2488 /* release module ref */
2489 module_put(THIS_MODULE
);
2492 static ssize_t
rbd_remove(struct bus_type
*bus
,
2496 struct rbd_device
*rbd_dev
= NULL
;
2501 rc
= strict_strtoul(buf
, 10, &ul
);
2505 /* convert to int; abort if we lost anything in the conversion */
2506 target_id
= (int) ul
;
2507 if (target_id
!= ul
)
2510 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2512 rbd_dev
= __rbd_get_dev(target_id
);
2518 __rbd_remove_all_snaps(rbd_dev
);
2519 rbd_bus_del_dev(rbd_dev
);
2522 mutex_unlock(&ctl_mutex
);
2526 static ssize_t
rbd_snap_add(struct device
*dev
,
2527 struct device_attribute
*attr
,
2531 struct rbd_device
*rbd_dev
= dev_to_rbd_dev(dev
);
2533 char *name
= kmalloc(count
+ 1, GFP_KERNEL
);
2537 snprintf(name
, count
, "%s", buf
);
2539 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2541 ret
= rbd_header_add_snap(rbd_dev
,
2546 ret
= __rbd_refresh_header(rbd_dev
);
2550 /* shouldn't hold ctl_mutex when notifying.. notify might
2551 trigger a watch callback that would need to get that mutex */
2552 mutex_unlock(&ctl_mutex
);
2554 /* make a best effort, don't error if failed */
2555 rbd_req_sync_notify(rbd_dev
, rbd_dev
->obj_md_name
);
2562 mutex_unlock(&ctl_mutex
);
2568 * create control files in sysfs
2571 static int rbd_sysfs_init(void)
2575 ret
= device_register(&rbd_root_dev
);
2579 ret
= bus_register(&rbd_bus_type
);
2581 device_unregister(&rbd_root_dev
);
2586 static void rbd_sysfs_cleanup(void)
2588 bus_unregister(&rbd_bus_type
);
2589 device_unregister(&rbd_root_dev
);
2592 int __init
rbd_init(void)
2596 rc
= rbd_sysfs_init();
2599 pr_info("loaded " RBD_DRV_NAME_LONG
"\n");
2603 void __exit
rbd_exit(void)
2605 rbd_sysfs_cleanup();
2608 module_init(rbd_init
);
2609 module_exit(rbd_exit
);
2611 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
2612 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
2613 MODULE_DESCRIPTION("rados block device");
2615 /* following authorship retained from original osdblk.c */
2616 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
2618 MODULE_LICENSE("GPL");