2 rbd.c -- Export ceph rados objects as a Linux block device
5 based on drivers/block/osdblk.c:
7 Copyright 2009 Red Hat, Inc.
9 This program is free software; you can redistribute it and/or modify
10 it under the terms of the GNU General Public License as published by
11 the Free Software Foundation.
13 This program is distributed in the hope that it will be useful,
14 but WITHOUT ANY WARRANTY; without even the implied warranty of
15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 GNU General Public License for more details.
18 You should have received a copy of the GNU General Public License
19 along with this program; see the file COPYING. If not, write to
20 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
24 For usage instructions, please refer to:
26 Documentation/ABI/testing/sysfs-bus-rbd
30 #include <linux/ceph/libceph.h>
31 #include <linux/ceph/osd_client.h>
32 #include <linux/ceph/mon_client.h>
33 #include <linux/ceph/decode.h>
34 #include <linux/parser.h>
36 #include <linux/kernel.h>
37 #include <linux/device.h>
38 #include <linux/module.h>
40 #include <linux/blkdev.h>
42 #include "rbd_types.h"
44 #define DRV_NAME "rbd"
45 #define DRV_NAME_LONG "rbd (rados block device)"
47 #define RBD_MINORS_PER_MAJOR 256 /* max minors per blkdev */
49 #define RBD_MAX_MD_NAME_LEN (96 + sizeof(RBD_SUFFIX))
50 #define RBD_MAX_POOL_NAME_LEN 64
51 #define RBD_MAX_SNAP_NAME_LEN 32
52 #define RBD_MAX_OPT_LEN 1024
54 #define RBD_SNAP_HEAD_NAME "-"
56 #define DEV_NAME_LEN 32
58 #define RBD_NOTIFY_TIMEOUT_DEFAULT 10
61 * block device image metadata (in-memory version)
63 struct rbd_image_header
{
69 struct rw_semaphore snap_rwsem
;
70 struct ceph_snap_context
*snapc
;
71 size_t snap_names_len
;
86 * an instance of the client. multiple devices may share a client.
89 struct ceph_client
*client
;
90 struct rbd_options
*rbd_opts
;
92 struct list_head node
;
101 struct request
*rq
; /* blk layer request */
102 struct bio
*bio
; /* cloned bio */
103 struct page
**pages
; /* list of used pages */
106 struct rbd_req_coll
*coll
;
109 struct rbd_req_status
{
116 * a collection of requests
118 struct rbd_req_coll
{
122 struct rbd_req_status status
[0];
129 struct list_head node
;
137 int id
; /* blkdev unique id */
139 int major
; /* blkdev assigned major */
140 struct gendisk
*disk
; /* blkdev's gendisk and rq */
141 struct request_queue
*q
;
143 struct ceph_client
*client
;
144 struct rbd_client
*rbd_client
;
146 char name
[DEV_NAME_LEN
]; /* blkdev name, e.g. rbd3 */
148 spinlock_t lock
; /* queue lock */
150 struct rbd_image_header header
;
151 char obj
[RBD_MAX_OBJ_NAME_LEN
]; /* rbd image name */
153 char obj_md_name
[RBD_MAX_MD_NAME_LEN
]; /* hdr nm. */
154 char pool_name
[RBD_MAX_POOL_NAME_LEN
];
157 struct ceph_osd_event
*watch_event
;
158 struct ceph_osd_request
*watch_request
;
160 char snap_name
[RBD_MAX_SNAP_NAME_LEN
];
161 u32 cur_snap
; /* index+1 of current snapshot within snap context
165 struct list_head node
;
167 /* list of snapshots */
168 struct list_head snaps
;
174 static struct bus_type rbd_bus_type
= {
178 static spinlock_t node_lock
; /* protects client get/put */
180 static DEFINE_MUTEX(ctl_mutex
); /* Serialize open/close/setup/teardown */
181 static LIST_HEAD(rbd_dev_list
); /* devices */
182 static LIST_HEAD(rbd_client_list
); /* clients */
184 static int __rbd_init_snaps_header(struct rbd_device
*rbd_dev
);
185 static void rbd_dev_release(struct device
*dev
);
186 static ssize_t
rbd_snap_rollback(struct device
*dev
,
187 struct device_attribute
*attr
,
190 static ssize_t
rbd_snap_add(struct device
*dev
,
191 struct device_attribute
*attr
,
194 static void __rbd_remove_snap_dev(struct rbd_device
*rbd_dev
,
195 struct rbd_snap
*snap
);
198 static struct rbd_device
*dev_to_rbd(struct device
*dev
)
200 return container_of(dev
, struct rbd_device
, dev
);
203 static struct device
*rbd_get_dev(struct rbd_device
*rbd_dev
)
205 return get_device(&rbd_dev
->dev
);
208 static void rbd_put_dev(struct rbd_device
*rbd_dev
)
210 put_device(&rbd_dev
->dev
);
213 static int __rbd_update_snaps(struct rbd_device
*rbd_dev
);
215 static int rbd_open(struct block_device
*bdev
, fmode_t mode
)
217 struct gendisk
*disk
= bdev
->bd_disk
;
218 struct rbd_device
*rbd_dev
= disk
->private_data
;
220 rbd_get_dev(rbd_dev
);
222 set_device_ro(bdev
, rbd_dev
->read_only
);
224 if ((mode
& FMODE_WRITE
) && rbd_dev
->read_only
)
230 static int rbd_release(struct gendisk
*disk
, fmode_t mode
)
232 struct rbd_device
*rbd_dev
= disk
->private_data
;
234 rbd_put_dev(rbd_dev
);
239 static const struct block_device_operations rbd_bd_ops
= {
240 .owner
= THIS_MODULE
,
242 .release
= rbd_release
,
246 * Initialize an rbd client instance.
249 static struct rbd_client
*rbd_client_create(struct ceph_options
*opt
,
250 struct rbd_options
*rbd_opts
)
252 struct rbd_client
*rbdc
;
255 dout("rbd_client_create\n");
256 rbdc
= kmalloc(sizeof(struct rbd_client
), GFP_KERNEL
);
260 kref_init(&rbdc
->kref
);
261 INIT_LIST_HEAD(&rbdc
->node
);
263 rbdc
->client
= ceph_create_client(opt
, rbdc
, 0, 0);
264 if (IS_ERR(rbdc
->client
))
266 opt
= NULL
; /* Now rbdc->client is responsible for opt */
268 ret
= ceph_open_session(rbdc
->client
);
272 rbdc
->rbd_opts
= rbd_opts
;
274 spin_lock(&node_lock
);
275 list_add_tail(&rbdc
->node
, &rbd_client_list
);
276 spin_unlock(&node_lock
);
278 dout("rbd_client_create created %p\n", rbdc
);
282 ceph_destroy_client(rbdc
->client
);
287 ceph_destroy_options(opt
);
292 * Find a ceph client with specific addr and configuration.
294 static struct rbd_client
*__rbd_client_find(struct ceph_options
*opt
)
296 struct rbd_client
*client_node
;
298 if (opt
->flags
& CEPH_OPT_NOSHARE
)
301 list_for_each_entry(client_node
, &rbd_client_list
, node
)
302 if (ceph_compare_options(opt
, client_node
->client
) == 0)
315 /* string args above */
318 static match_table_t rbdopt_tokens
= {
319 {Opt_notify_timeout
, "notify_timeout=%d"},
321 /* string args above */
325 static int parse_rbd_opts_token(char *c
, void *private)
327 struct rbd_options
*rbdopt
= private;
328 substring_t argstr
[MAX_OPT_ARGS
];
329 int token
, intval
, ret
;
331 token
= match_token((char *)c
, rbdopt_tokens
, argstr
);
335 if (token
< Opt_last_int
) {
336 ret
= match_int(&argstr
[0], &intval
);
338 pr_err("bad mount option arg (not int) "
342 dout("got int token %d val %d\n", token
, intval
);
343 } else if (token
> Opt_last_int
&& token
< Opt_last_string
) {
344 dout("got string token %d val %s\n", token
,
347 dout("got token %d\n", token
);
351 case Opt_notify_timeout
:
352 rbdopt
->notify_timeout
= intval
;
361 * Get a ceph client with specific addr and configuration, if one does
362 * not exist create it.
364 static int rbd_get_client(struct rbd_device
*rbd_dev
, const char *mon_addr
,
367 struct rbd_client
*rbdc
;
368 struct ceph_options
*opt
;
370 struct rbd_options
*rbd_opts
;
372 rbd_opts
= kzalloc(sizeof(*rbd_opts
), GFP_KERNEL
);
376 rbd_opts
->notify_timeout
= RBD_NOTIFY_TIMEOUT_DEFAULT
;
378 ret
= ceph_parse_options(&opt
, options
, mon_addr
,
379 mon_addr
+ strlen(mon_addr
), parse_rbd_opts_token
, rbd_opts
);
383 spin_lock(&node_lock
);
384 rbdc
= __rbd_client_find(opt
);
386 ceph_destroy_options(opt
);
388 /* using an existing client */
389 kref_get(&rbdc
->kref
);
390 rbd_dev
->rbd_client
= rbdc
;
391 rbd_dev
->client
= rbdc
->client
;
392 spin_unlock(&node_lock
);
395 spin_unlock(&node_lock
);
397 rbdc
= rbd_client_create(opt
, rbd_opts
);
403 rbd_dev
->rbd_client
= rbdc
;
404 rbd_dev
->client
= rbdc
->client
;
412 * Destroy ceph client
414 static void rbd_client_release(struct kref
*kref
)
416 struct rbd_client
*rbdc
= container_of(kref
, struct rbd_client
, kref
);
418 dout("rbd_release_client %p\n", rbdc
);
419 spin_lock(&node_lock
);
420 list_del(&rbdc
->node
);
421 spin_unlock(&node_lock
);
423 ceph_destroy_client(rbdc
->client
);
424 kfree(rbdc
->rbd_opts
);
429 * Drop reference to ceph client node. If it's not referenced anymore, release
432 static void rbd_put_client(struct rbd_device
*rbd_dev
)
434 kref_put(&rbd_dev
->rbd_client
->kref
, rbd_client_release
);
435 rbd_dev
->rbd_client
= NULL
;
436 rbd_dev
->client
= NULL
;
440 * Destroy requests collection
442 static void rbd_coll_release(struct kref
*kref
)
444 struct rbd_req_coll
*coll
=
445 container_of(kref
, struct rbd_req_coll
, kref
);
447 dout("rbd_coll_release %p\n", coll
);
452 * Create a new header structure, translate header format from the on-disk
455 static int rbd_header_from_disk(struct rbd_image_header
*header
,
456 struct rbd_image_header_ondisk
*ondisk
,
461 u32 snap_count
= le32_to_cpu(ondisk
->snap_count
);
464 init_rwsem(&header
->snap_rwsem
);
465 header
->snap_names_len
= le64_to_cpu(ondisk
->snap_names_len
);
466 header
->snapc
= kmalloc(sizeof(struct ceph_snap_context
) +
468 sizeof(struct rbd_image_snap_ondisk
),
473 header
->snap_names
= kmalloc(header
->snap_names_len
,
475 if (!header
->snap_names
)
477 header
->snap_sizes
= kmalloc(snap_count
* sizeof(u64
),
479 if (!header
->snap_sizes
)
482 header
->snap_names
= NULL
;
483 header
->snap_sizes
= NULL
;
485 memcpy(header
->block_name
, ondisk
->block_name
,
486 sizeof(ondisk
->block_name
));
488 header
->image_size
= le64_to_cpu(ondisk
->image_size
);
489 header
->obj_order
= ondisk
->options
.order
;
490 header
->crypt_type
= ondisk
->options
.crypt_type
;
491 header
->comp_type
= ondisk
->options
.comp_type
;
493 atomic_set(&header
->snapc
->nref
, 1);
494 header
->snap_seq
= le64_to_cpu(ondisk
->snap_seq
);
495 header
->snapc
->num_snaps
= snap_count
;
496 header
->total_snaps
= snap_count
;
499 allocated_snaps
== snap_count
) {
500 for (i
= 0; i
< snap_count
; i
++) {
501 header
->snapc
->snaps
[i
] =
502 le64_to_cpu(ondisk
->snaps
[i
].id
);
503 header
->snap_sizes
[i
] =
504 le64_to_cpu(ondisk
->snaps
[i
].image_size
);
507 /* copy snapshot names */
508 memcpy(header
->snap_names
, &ondisk
->snaps
[i
],
509 header
->snap_names_len
);
515 kfree(header
->snap_names
);
517 kfree(header
->snapc
);
521 static int snap_index(struct rbd_image_header
*header
, int snap_num
)
523 return header
->total_snaps
- snap_num
;
526 static u64
cur_snap_id(struct rbd_device
*rbd_dev
)
528 struct rbd_image_header
*header
= &rbd_dev
->header
;
530 if (!rbd_dev
->cur_snap
)
533 return header
->snapc
->snaps
[snap_index(header
, rbd_dev
->cur_snap
)];
536 static int snap_by_name(struct rbd_image_header
*header
, const char *snap_name
,
540 char *p
= header
->snap_names
;
542 for (i
= 0; i
< header
->total_snaps
; i
++, p
+= strlen(p
) + 1) {
543 if (strcmp(snap_name
, p
) == 0)
546 if (i
== header
->total_snaps
)
549 *seq
= header
->snapc
->snaps
[i
];
552 *size
= header
->snap_sizes
[i
];
557 static int rbd_header_set_snap(struct rbd_device
*dev
,
558 const char *snap_name
,
561 struct rbd_image_header
*header
= &dev
->header
;
562 struct ceph_snap_context
*snapc
= header
->snapc
;
565 down_write(&header
->snap_rwsem
);
569 strcmp(snap_name
, "-") == 0 ||
570 strcmp(snap_name
, RBD_SNAP_HEAD_NAME
) == 0) {
571 if (header
->total_snaps
)
572 snapc
->seq
= header
->snap_seq
;
578 *size
= header
->image_size
;
580 ret
= snap_by_name(header
, snap_name
, &snapc
->seq
, size
);
584 dev
->cur_snap
= header
->total_snaps
- ret
;
590 up_write(&header
->snap_rwsem
);
594 static void rbd_header_free(struct rbd_image_header
*header
)
596 kfree(header
->snapc
);
597 kfree(header
->snap_names
);
598 kfree(header
->snap_sizes
);
602 * get the actual striped segment name, offset and length
604 static u64
rbd_get_segment(struct rbd_image_header
*header
,
605 const char *block_name
,
607 char *seg_name
, u64
*segofs
)
609 u64 seg
= ofs
>> header
->obj_order
;
612 snprintf(seg_name
, RBD_MAX_SEG_NAME_LEN
,
613 "%s.%012llx", block_name
, seg
);
615 ofs
= ofs
& ((1 << header
->obj_order
) - 1);
616 len
= min_t(u64
, len
, (1 << header
->obj_order
) - ofs
);
624 static int rbd_get_num_segments(struct rbd_image_header
*header
,
627 u64 start_seg
= ofs
>> header
->obj_order
;
628 u64 end_seg
= (ofs
+ len
- 1) >> header
->obj_order
;
629 return end_seg
- start_seg
+ 1;
633 * returns the size of an object in the image
635 static u64
rbd_obj_bytes(struct rbd_image_header
*header
)
637 return 1 << header
->obj_order
;
644 static void bio_chain_put(struct bio
*chain
)
650 chain
= chain
->bi_next
;
656 * zeros a bio chain, starting at specific offset
658 static void zero_bio_chain(struct bio
*chain
, int start_ofs
)
667 bio_for_each_segment(bv
, chain
, i
) {
668 if (pos
+ bv
->bv_len
> start_ofs
) {
669 int remainder
= max(start_ofs
- pos
, 0);
670 buf
= bvec_kmap_irq(bv
, &flags
);
671 memset(buf
+ remainder
, 0,
672 bv
->bv_len
- remainder
);
673 bvec_kunmap_irq(buf
, &flags
);
678 chain
= chain
->bi_next
;
683 * bio_chain_clone - clone a chain of bios up to a certain length.
684 * might return a bio_pair that will need to be released.
686 static struct bio
*bio_chain_clone(struct bio
**old
, struct bio
**next
,
687 struct bio_pair
**bp
,
688 int len
, gfp_t gfpmask
)
690 struct bio
*tmp
, *old_chain
= *old
, *new_chain
= NULL
, *tail
= NULL
;
694 bio_pair_release(*bp
);
698 while (old_chain
&& (total
< len
)) {
699 tmp
= bio_kmalloc(gfpmask
, old_chain
->bi_max_vecs
);
703 if (total
+ old_chain
->bi_size
> len
) {
707 * this split can only happen with a single paged bio,
708 * split_bio will BUG_ON if this is not the case
710 dout("bio_chain_clone split! total=%d remaining=%d"
712 (int)total
, (int)len
-total
,
713 (int)old_chain
->bi_size
);
715 /* split the bio. We'll release it either in the next
716 call, or it will have to be released outside */
717 bp
= bio_split(old_chain
, (len
- total
) / 512ULL);
721 __bio_clone(tmp
, &bp
->bio1
);
725 __bio_clone(tmp
, old_chain
);
726 *next
= old_chain
->bi_next
;
730 gfpmask
&= ~__GFP_WAIT
;
734 new_chain
= tail
= tmp
;
739 old_chain
= old_chain
->bi_next
;
741 total
+= tmp
->bi_size
;
747 tail
->bi_next
= NULL
;
754 dout("bio_chain_clone with err\n");
755 bio_chain_put(new_chain
);
760 * helpers for osd request op vectors.
762 static int rbd_create_rw_ops(struct ceph_osd_req_op
**ops
,
767 *ops
= kzalloc(sizeof(struct ceph_osd_req_op
) * (num_ops
+ 1),
771 (*ops
)[0].op
= opcode
;
773 * op extent offset and length will be set later on
774 * in calc_raw_layout()
776 (*ops
)[0].payload_len
= payload_len
;
780 static void rbd_destroy_ops(struct ceph_osd_req_op
*ops
)
785 static void rbd_coll_end_req_index(struct request
*rq
,
786 struct rbd_req_coll
*coll
,
790 struct request_queue
*q
;
793 dout("rbd_coll_end_req_index %p index %d ret %d len %lld\n",
794 coll
, index
, ret
, len
);
800 blk_end_request(rq
, ret
, len
);
806 spin_lock_irq(q
->queue_lock
);
807 coll
->status
[index
].done
= 1;
808 coll
->status
[index
].rc
= ret
;
809 coll
->status
[index
].bytes
= len
;
810 max
= min
= coll
->num_done
;
811 while (max
< coll
->total
&& coll
->status
[max
].done
)
814 for (i
= min
; i
<max
; i
++) {
815 __blk_end_request(rq
, coll
->status
[i
].rc
,
816 coll
->status
[i
].bytes
);
818 kref_put(&coll
->kref
, rbd_coll_release
);
820 spin_unlock_irq(q
->queue_lock
);
823 static void rbd_coll_end_req(struct rbd_request
*req
,
826 rbd_coll_end_req_index(req
->rq
, req
->coll
, req
->coll_index
, ret
, len
);
830 * Send ceph osd request
832 static int rbd_do_request(struct request
*rq
,
833 struct rbd_device
*dev
,
834 struct ceph_snap_context
*snapc
,
836 const char *obj
, u64 ofs
, u64 len
,
841 struct ceph_osd_req_op
*ops
,
843 struct rbd_req_coll
*coll
,
845 void (*rbd_cb
)(struct ceph_osd_request
*req
,
846 struct ceph_msg
*msg
),
847 struct ceph_osd_request
**linger_req
,
850 struct ceph_osd_request
*req
;
851 struct ceph_file_layout
*layout
;
854 struct timespec mtime
= CURRENT_TIME
;
855 struct rbd_request
*req_data
;
856 struct ceph_osd_request_head
*reqhead
;
857 struct rbd_image_header
*header
= &dev
->header
;
859 req_data
= kzalloc(sizeof(*req_data
), GFP_NOIO
);
862 rbd_coll_end_req_index(rq
, coll
, coll_index
,
868 req_data
->coll
= coll
;
869 req_data
->coll_index
= coll_index
;
872 dout("rbd_do_request obj=%s ofs=%lld len=%lld\n", obj
, len
, ofs
);
874 down_read(&header
->snap_rwsem
);
876 req
= ceph_osdc_alloc_request(&dev
->client
->osdc
, flags
,
880 GFP_NOIO
, pages
, bio
);
882 up_read(&header
->snap_rwsem
);
887 req
->r_callback
= rbd_cb
;
891 req_data
->pages
= pages
;
894 req
->r_priv
= req_data
;
896 reqhead
= req
->r_request
->front
.iov_base
;
897 reqhead
->snapid
= cpu_to_le64(CEPH_NOSNAP
);
899 strncpy(req
->r_oid
, obj
, sizeof(req
->r_oid
));
900 req
->r_oid_len
= strlen(req
->r_oid
);
902 layout
= &req
->r_file_layout
;
903 memset(layout
, 0, sizeof(*layout
));
904 layout
->fl_stripe_unit
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
905 layout
->fl_stripe_count
= cpu_to_le32(1);
906 layout
->fl_object_size
= cpu_to_le32(1 << RBD_MAX_OBJ_ORDER
);
907 layout
->fl_pg_preferred
= cpu_to_le32(-1);
908 layout
->fl_pg_pool
= cpu_to_le32(dev
->poolid
);
909 ceph_calc_raw_layout(&dev
->client
->osdc
, layout
, snapid
,
910 ofs
, &len
, &bno
, req
, ops
);
912 ceph_osdc_build_request(req
, ofs
, &len
,
916 req
->r_oid
, req
->r_oid_len
);
917 up_read(&header
->snap_rwsem
);
920 ceph_osdc_set_request_linger(&dev
->client
->osdc
, req
);
924 ret
= ceph_osdc_start_request(&dev
->client
->osdc
, req
, false);
929 ret
= ceph_osdc_wait_request(&dev
->client
->osdc
, req
);
931 *ver
= le64_to_cpu(req
->r_reassert_version
.version
);
932 dout("reassert_ver=%lld\n",
933 le64_to_cpu(req
->r_reassert_version
.version
));
934 ceph_osdc_put_request(req
);
939 bio_chain_put(req_data
->bio
);
940 ceph_osdc_put_request(req
);
942 rbd_coll_end_req(req_data
, ret
, len
);
948 * Ceph osd op callback
950 static void rbd_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
952 struct rbd_request
*req_data
= req
->r_priv
;
953 struct ceph_osd_reply_head
*replyhead
;
954 struct ceph_osd_op
*op
;
960 replyhead
= msg
->front
.iov_base
;
961 WARN_ON(le32_to_cpu(replyhead
->num_ops
) == 0);
962 op
= (void *)(replyhead
+ 1);
963 rc
= le32_to_cpu(replyhead
->result
);
964 bytes
= le64_to_cpu(op
->extent
.length
);
965 read_op
= (le32_to_cpu(op
->op
) == CEPH_OSD_OP_READ
);
967 dout("rbd_req_cb bytes=%lld readop=%d rc=%d\n", bytes
, read_op
, rc
);
969 if (rc
== -ENOENT
&& read_op
) {
970 zero_bio_chain(req_data
->bio
, 0);
972 } else if (rc
== 0 && read_op
&& bytes
< req_data
->len
) {
973 zero_bio_chain(req_data
->bio
, bytes
);
974 bytes
= req_data
->len
;
977 rbd_coll_end_req(req_data
, rc
, bytes
);
980 bio_chain_put(req_data
->bio
);
982 ceph_osdc_put_request(req
);
986 static void rbd_simple_req_cb(struct ceph_osd_request
*req
, struct ceph_msg
*msg
)
988 ceph_osdc_put_request(req
);
992 * Do a synchronous ceph osd operation
994 static int rbd_req_sync_op(struct rbd_device
*dev
,
995 struct ceph_snap_context
*snapc
,
999 struct ceph_osd_req_op
*orig_ops
,
1004 struct ceph_osd_request
**linger_req
,
1008 struct page
**pages
;
1010 struct ceph_osd_req_op
*ops
= orig_ops
;
1013 num_pages
= calc_pages_for(ofs
, len
);
1014 pages
= ceph_alloc_page_vector(num_pages
, GFP_KERNEL
);
1016 return PTR_ERR(pages
);
1019 payload_len
= (flags
& CEPH_OSD_FLAG_WRITE
? len
: 0);
1020 ret
= rbd_create_rw_ops(&ops
, 1, opcode
, payload_len
);
1024 if ((flags
& CEPH_OSD_FLAG_WRITE
) && buf
) {
1025 ret
= ceph_copy_to_page_vector(pages
, buf
, ofs
, len
);
1031 ret
= rbd_do_request(NULL
, dev
, snapc
, snapid
,
1032 obj
, ofs
, len
, NULL
,
1043 if ((flags
& CEPH_OSD_FLAG_READ
) && buf
)
1044 ret
= ceph_copy_from_page_vector(pages
, buf
, ofs
, ret
);
1048 rbd_destroy_ops(ops
);
1050 ceph_release_page_vector(pages
, num_pages
);
1055 * Do an asynchronous ceph osd operation
1057 static int rbd_do_op(struct request
*rq
,
1058 struct rbd_device
*rbd_dev
,
1059 struct ceph_snap_context
*snapc
,
1061 int opcode
, int flags
, int num_reply
,
1064 struct rbd_req_coll
*coll
,
1071 struct ceph_osd_req_op
*ops
;
1074 seg_name
= kmalloc(RBD_MAX_SEG_NAME_LEN
+ 1, GFP_NOIO
);
1078 seg_len
= rbd_get_segment(&rbd_dev
->header
,
1079 rbd_dev
->header
.block_name
,
1081 seg_name
, &seg_ofs
);
1083 payload_len
= (flags
& CEPH_OSD_FLAG_WRITE
? seg_len
: 0);
1085 ret
= rbd_create_rw_ops(&ops
, 1, opcode
, payload_len
);
1089 /* we've taken care of segment sizes earlier when we
1090 cloned the bios. We should never have a segment
1091 truncated at this point */
1092 BUG_ON(seg_len
< len
);
1094 ret
= rbd_do_request(rq
, rbd_dev
, snapc
, snapid
,
1095 seg_name
, seg_ofs
, seg_len
,
1102 rbd_req_cb
, 0, NULL
);
1104 rbd_destroy_ops(ops
);
1111 * Request async osd write
1113 static int rbd_req_write(struct request
*rq
,
1114 struct rbd_device
*rbd_dev
,
1115 struct ceph_snap_context
*snapc
,
1118 struct rbd_req_coll
*coll
,
1121 return rbd_do_op(rq
, rbd_dev
, snapc
, CEPH_NOSNAP
,
1123 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1125 ofs
, len
, bio
, coll
, coll_index
);
1129 * Request async osd read
1131 static int rbd_req_read(struct request
*rq
,
1132 struct rbd_device
*rbd_dev
,
1136 struct rbd_req_coll
*coll
,
1139 return rbd_do_op(rq
, rbd_dev
, NULL
,
1140 (snapid
? snapid
: CEPH_NOSNAP
),
1144 ofs
, len
, bio
, coll
, coll_index
);
1148 * Request sync osd read
1150 static int rbd_req_sync_read(struct rbd_device
*dev
,
1151 struct ceph_snap_context
*snapc
,
1158 return rbd_req_sync_op(dev
, NULL
,
1159 (snapid
? snapid
: CEPH_NOSNAP
),
1163 1, obj
, ofs
, len
, buf
, NULL
, ver
);
1167 * Request sync osd watch
1169 static int rbd_req_sync_notify_ack(struct rbd_device
*dev
,
1174 struct ceph_osd_req_op
*ops
;
1175 struct page
**pages
= NULL
;
1178 ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_NOTIFY_ACK
, 0);
1182 ops
[0].watch
.ver
= cpu_to_le64(dev
->header
.obj_version
);
1183 ops
[0].watch
.cookie
= notify_id
;
1184 ops
[0].watch
.flag
= 0;
1186 ret
= rbd_do_request(NULL
, dev
, NULL
, CEPH_NOSNAP
,
1193 rbd_simple_req_cb
, 0, NULL
);
1195 rbd_destroy_ops(ops
);
1199 static void rbd_watch_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1201 struct rbd_device
*dev
= (struct rbd_device
*)data
;
1207 dout("rbd_watch_cb %s notify_id=%lld opcode=%d\n", dev
->obj_md_name
,
1208 notify_id
, (int)opcode
);
1209 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1210 rc
= __rbd_update_snaps(dev
);
1211 mutex_unlock(&ctl_mutex
);
1213 pr_warning(DRV_NAME
"%d got notification but failed to update"
1214 " snaps: %d\n", dev
->major
, rc
);
1216 rbd_req_sync_notify_ack(dev
, ver
, notify_id
, dev
->obj_md_name
);
1220 * Request sync osd watch
1222 static int rbd_req_sync_watch(struct rbd_device
*dev
,
1226 struct ceph_osd_req_op
*ops
;
1227 struct ceph_osd_client
*osdc
= &dev
->client
->osdc
;
1229 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_WATCH
, 0);
1233 ret
= ceph_osdc_create_event(osdc
, rbd_watch_cb
, 0,
1234 (void *)dev
, &dev
->watch_event
);
1238 ops
[0].watch
.ver
= cpu_to_le64(ver
);
1239 ops
[0].watch
.cookie
= cpu_to_le64(dev
->watch_event
->cookie
);
1240 ops
[0].watch
.flag
= 1;
1242 ret
= rbd_req_sync_op(dev
, NULL
,
1245 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1248 &dev
->watch_request
, NULL
);
1253 rbd_destroy_ops(ops
);
1257 ceph_osdc_cancel_event(dev
->watch_event
);
1258 dev
->watch_event
= NULL
;
1260 rbd_destroy_ops(ops
);
1265 * Request sync osd unwatch
1267 static int rbd_req_sync_unwatch(struct rbd_device
*dev
,
1270 struct ceph_osd_req_op
*ops
;
1272 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_WATCH
, 0);
1276 ops
[0].watch
.ver
= 0;
1277 ops
[0].watch
.cookie
= cpu_to_le64(dev
->watch_event
->cookie
);
1278 ops
[0].watch
.flag
= 0;
1280 ret
= rbd_req_sync_op(dev
, NULL
,
1283 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1285 1, obj
, 0, 0, NULL
, NULL
, NULL
);
1287 rbd_destroy_ops(ops
);
1288 ceph_osdc_cancel_event(dev
->watch_event
);
1289 dev
->watch_event
= NULL
;
1293 struct rbd_notify_info
{
1294 struct rbd_device
*dev
;
1297 static void rbd_notify_cb(u64 ver
, u64 notify_id
, u8 opcode
, void *data
)
1299 struct rbd_device
*dev
= (struct rbd_device
*)data
;
1303 dout("rbd_notify_cb %s notify_id=%lld opcode=%d\n", dev
->obj_md_name
,
1304 notify_id
, (int)opcode
);
1308 * Request sync osd notify
1310 static int rbd_req_sync_notify(struct rbd_device
*dev
,
1313 struct ceph_osd_req_op
*ops
;
1314 struct ceph_osd_client
*osdc
= &dev
->client
->osdc
;
1315 struct ceph_osd_event
*event
;
1316 struct rbd_notify_info info
;
1317 int payload_len
= sizeof(u32
) + sizeof(u32
);
1320 ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_NOTIFY
, payload_len
);
1326 ret
= ceph_osdc_create_event(osdc
, rbd_notify_cb
, 1,
1327 (void *)&info
, &event
);
1331 ops
[0].watch
.ver
= 1;
1332 ops
[0].watch
.flag
= 1;
1333 ops
[0].watch
.cookie
= event
->cookie
;
1334 ops
[0].watch
.prot_ver
= RADOS_NOTIFY_VER
;
1335 ops
[0].watch
.timeout
= 12;
1337 ret
= rbd_req_sync_op(dev
, NULL
,
1340 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1342 1, obj
, 0, 0, NULL
, NULL
, NULL
);
1346 ret
= ceph_osdc_wait_event(event
, CEPH_OSD_TIMEOUT_DEFAULT
);
1347 dout("ceph_osdc_wait_event returned %d\n", ret
);
1348 rbd_destroy_ops(ops
);
1352 ceph_osdc_cancel_event(event
);
1354 rbd_destroy_ops(ops
);
1359 * Request sync osd rollback
1361 static int rbd_req_sync_rollback_obj(struct rbd_device
*dev
,
1365 struct ceph_osd_req_op
*ops
;
1366 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_ROLLBACK
, 0);
1370 ops
[0].snap
.snapid
= snapid
;
1372 ret
= rbd_req_sync_op(dev
, NULL
,
1375 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1377 1, obj
, 0, 0, NULL
, NULL
, NULL
);
1379 rbd_destroy_ops(ops
);
1385 * Request sync osd read
1387 static int rbd_req_sync_exec(struct rbd_device
*dev
,
1395 struct ceph_osd_req_op
*ops
;
1396 int cls_len
= strlen(cls
);
1397 int method_len
= strlen(method
);
1398 int ret
= rbd_create_rw_ops(&ops
, 1, CEPH_OSD_OP_CALL
,
1399 cls_len
+ method_len
+ len
);
1403 ops
[0].cls
.class_name
= cls
;
1404 ops
[0].cls
.class_len
= (__u8
)cls_len
;
1405 ops
[0].cls
.method_name
= method
;
1406 ops
[0].cls
.method_len
= (__u8
)method_len
;
1407 ops
[0].cls
.argc
= 0;
1408 ops
[0].cls
.indata
= data
;
1409 ops
[0].cls
.indata_len
= len
;
1411 ret
= rbd_req_sync_op(dev
, NULL
,
1414 CEPH_OSD_FLAG_WRITE
| CEPH_OSD_FLAG_ONDISK
,
1416 1, obj
, 0, 0, NULL
, NULL
, ver
);
1418 rbd_destroy_ops(ops
);
1420 dout("cls_exec returned %d\n", ret
);
1424 static struct rbd_req_coll
*rbd_alloc_coll(int num_reqs
)
1426 struct rbd_req_coll
*coll
=
1427 kzalloc(sizeof(struct rbd_req_coll
) +
1428 sizeof(struct rbd_req_status
) * num_reqs
,
1433 coll
->total
= num_reqs
;
1434 kref_init(&coll
->kref
);
1439 * block device queue callback
1441 static void rbd_rq_fn(struct request_queue
*q
)
1443 struct rbd_device
*rbd_dev
= q
->queuedata
;
1445 struct bio_pair
*bp
= NULL
;
1447 rq
= blk_fetch_request(q
);
1451 struct bio
*rq_bio
, *next_bio
= NULL
;
1453 int size
, op_size
= 0;
1455 int num_segs
, cur_seg
= 0;
1456 struct rbd_req_coll
*coll
;
1458 /* peek at request from block layer */
1462 dout("fetched request\n");
1464 /* filter out block requests we don't understand */
1465 if ((rq
->cmd_type
!= REQ_TYPE_FS
)) {
1466 __blk_end_request_all(rq
, 0);
1470 /* deduce our operation (read, write) */
1471 do_write
= (rq_data_dir(rq
) == WRITE
);
1473 size
= blk_rq_bytes(rq
);
1474 ofs
= blk_rq_pos(rq
) * 512ULL;
1476 if (do_write
&& rbd_dev
->read_only
) {
1477 __blk_end_request_all(rq
, -EROFS
);
1481 spin_unlock_irq(q
->queue_lock
);
1483 dout("%s 0x%x bytes at 0x%llx\n",
1484 do_write
? "write" : "read",
1485 size
, blk_rq_pos(rq
) * 512ULL);
1487 num_segs
= rbd_get_num_segments(&rbd_dev
->header
, ofs
, size
);
1488 coll
= rbd_alloc_coll(num_segs
);
1490 spin_lock_irq(q
->queue_lock
);
1491 __blk_end_request_all(rq
, -ENOMEM
);
1496 /* a bio clone to be passed down to OSD req */
1497 dout("rq->bio->bi_vcnt=%d\n", rq
->bio
->bi_vcnt
);
1498 op_size
= rbd_get_segment(&rbd_dev
->header
,
1499 rbd_dev
->header
.block_name
,
1502 kref_get(&coll
->kref
);
1503 bio
= bio_chain_clone(&rq_bio
, &next_bio
, &bp
,
1504 op_size
, GFP_ATOMIC
);
1506 rbd_coll_end_req_index(rq
, coll
, cur_seg
,
1512 /* init OSD command: write or read */
1514 rbd_req_write(rq
, rbd_dev
,
1515 rbd_dev
->header
.snapc
,
1520 rbd_req_read(rq
, rbd_dev
,
1521 cur_snap_id(rbd_dev
),
1533 kref_put(&coll
->kref
, rbd_coll_release
);
1536 bio_pair_release(bp
);
1537 spin_lock_irq(q
->queue_lock
);
1539 rq
= blk_fetch_request(q
);
1544 * a queue callback. Makes sure that we don't create a bio that spans across
1545 * multiple osd objects. One exception would be with a single page bios,
1546 * which we handle later at bio_chain_clone
1548 static int rbd_merge_bvec(struct request_queue
*q
, struct bvec_merge_data
*bmd
,
1549 struct bio_vec
*bvec
)
1551 struct rbd_device
*rbd_dev
= q
->queuedata
;
1552 unsigned int chunk_sectors
= 1 << (rbd_dev
->header
.obj_order
- 9);
1553 sector_t sector
= bmd
->bi_sector
+ get_start_sect(bmd
->bi_bdev
);
1554 unsigned int bio_sectors
= bmd
->bi_size
>> 9;
1557 max
= (chunk_sectors
- ((sector
& (chunk_sectors
- 1))
1558 + bio_sectors
)) << 9;
1560 max
= 0; /* bio_add cannot handle a negative return */
1561 if (max
<= bvec
->bv_len
&& bio_sectors
== 0)
1562 return bvec
->bv_len
;
1566 static void rbd_free_disk(struct rbd_device
*rbd_dev
)
1568 struct gendisk
*disk
= rbd_dev
->disk
;
1573 rbd_header_free(&rbd_dev
->header
);
1575 if (disk
->flags
& GENHD_FL_UP
)
1578 blk_cleanup_queue(disk
->queue
);
1583 * reload the ondisk the header
1585 static int rbd_read_header(struct rbd_device
*rbd_dev
,
1586 struct rbd_image_header
*header
)
1589 struct rbd_image_header_ondisk
*dh
;
1591 u64 snap_names_len
= 0;
1595 int len
= sizeof(*dh
) +
1596 snap_count
* sizeof(struct rbd_image_snap_ondisk
) +
1600 dh
= kmalloc(len
, GFP_KERNEL
);
1604 rc
= rbd_req_sync_read(rbd_dev
,
1606 rbd_dev
->obj_md_name
,
1612 rc
= rbd_header_from_disk(header
, dh
, snap_count
, GFP_KERNEL
);
1616 if (snap_count
!= header
->total_snaps
) {
1617 snap_count
= header
->total_snaps
;
1618 snap_names_len
= header
->snap_names_len
;
1619 rbd_header_free(header
);
1625 header
->obj_version
= ver
;
1635 static int rbd_header_add_snap(struct rbd_device
*dev
,
1636 const char *snap_name
,
1639 int name_len
= strlen(snap_name
);
1645 /* we should create a snapshot only if we're pointing at the head */
1649 ret
= ceph_monc_create_snapid(&dev
->client
->monc
, dev
->poolid
,
1651 dout("created snapid=%lld\n", new_snapid
);
1655 data
= kmalloc(name_len
+ 16, gfp_flags
);
1660 e
= data
+ name_len
+ 16;
1662 ceph_encode_string_safe(&p
, e
, snap_name
, name_len
, bad
);
1663 ceph_encode_64_safe(&p
, e
, new_snapid
, bad
);
1665 ret
= rbd_req_sync_exec(dev
, dev
->obj_md_name
, "rbd", "snap_add",
1666 data
, p
- data
, &ver
);
1673 dev
->header
.snapc
->seq
= new_snapid
;
1680 static void __rbd_remove_all_snaps(struct rbd_device
*rbd_dev
)
1682 struct rbd_snap
*snap
;
1684 while (!list_empty(&rbd_dev
->snaps
)) {
1685 snap
= list_first_entry(&rbd_dev
->snaps
, struct rbd_snap
, node
);
1686 __rbd_remove_snap_dev(rbd_dev
, snap
);
1691 * only read the first part of the ondisk header, without the snaps info
1693 static int __rbd_update_snaps(struct rbd_device
*rbd_dev
)
1696 struct rbd_image_header h
;
1700 ret
= rbd_read_header(rbd_dev
, &h
);
1705 set_capacity(rbd_dev
->disk
, h
.image_size
/ 512ULL);
1707 down_write(&rbd_dev
->header
.snap_rwsem
);
1709 snap_seq
= rbd_dev
->header
.snapc
->seq
;
1710 if (rbd_dev
->header
.total_snaps
&&
1711 rbd_dev
->header
.snapc
->snaps
[0] == snap_seq
)
1712 /* pointing at the head, will need to follow that
1716 kfree(rbd_dev
->header
.snapc
);
1717 kfree(rbd_dev
->header
.snap_names
);
1718 kfree(rbd_dev
->header
.snap_sizes
);
1720 rbd_dev
->header
.total_snaps
= h
.total_snaps
;
1721 rbd_dev
->header
.snapc
= h
.snapc
;
1722 rbd_dev
->header
.snap_names
= h
.snap_names
;
1723 rbd_dev
->header
.snap_names_len
= h
.snap_names_len
;
1724 rbd_dev
->header
.snap_sizes
= h
.snap_sizes
;
1726 rbd_dev
->header
.snapc
->seq
= rbd_dev
->header
.snapc
->snaps
[0];
1728 rbd_dev
->header
.snapc
->seq
= snap_seq
;
1730 ret
= __rbd_init_snaps_header(rbd_dev
);
1732 up_write(&rbd_dev
->header
.snap_rwsem
);
1737 static int rbd_init_disk(struct rbd_device
*rbd_dev
)
1739 struct gendisk
*disk
;
1740 struct request_queue
*q
;
1744 /* contact OSD, request size info about the object being mapped */
1745 rc
= rbd_read_header(rbd_dev
, &rbd_dev
->header
);
1749 /* no need to lock here, as rbd_dev is not registered yet */
1750 rc
= __rbd_init_snaps_header(rbd_dev
);
1754 rc
= rbd_header_set_snap(rbd_dev
, rbd_dev
->snap_name
, &total_size
);
1758 /* create gendisk info */
1760 disk
= alloc_disk(RBD_MINORS_PER_MAJOR
);
1764 snprintf(disk
->disk_name
, sizeof(disk
->disk_name
), DRV_NAME
"%d",
1766 disk
->major
= rbd_dev
->major
;
1767 disk
->first_minor
= 0;
1768 disk
->fops
= &rbd_bd_ops
;
1769 disk
->private_data
= rbd_dev
;
1773 q
= blk_init_queue(rbd_rq_fn
, &rbd_dev
->lock
);
1777 /* set io sizes to object size */
1778 blk_queue_max_hw_sectors(q
, rbd_obj_bytes(&rbd_dev
->header
) / 512ULL);
1779 blk_queue_max_segment_size(q
, rbd_obj_bytes(&rbd_dev
->header
));
1780 blk_queue_io_min(q
, rbd_obj_bytes(&rbd_dev
->header
));
1781 blk_queue_io_opt(q
, rbd_obj_bytes(&rbd_dev
->header
));
1783 blk_queue_merge_bvec(q
, rbd_merge_bvec
);
1786 q
->queuedata
= rbd_dev
;
1788 rbd_dev
->disk
= disk
;
1791 /* finally, announce the disk to the world */
1792 set_capacity(disk
, total_size
/ 512ULL);
1795 pr_info("%s: added with size 0x%llx\n",
1796 disk
->disk_name
, (unsigned long long)total_size
);
1809 static ssize_t
rbd_size_show(struct device
*dev
,
1810 struct device_attribute
*attr
, char *buf
)
1812 struct rbd_device
*rbd_dev
= dev_to_rbd(dev
);
1814 return sprintf(buf
, "%llu\n", (unsigned long long)rbd_dev
->header
.image_size
);
1817 static ssize_t
rbd_major_show(struct device
*dev
,
1818 struct device_attribute
*attr
, char *buf
)
1820 struct rbd_device
*rbd_dev
= dev_to_rbd(dev
);
1822 return sprintf(buf
, "%d\n", rbd_dev
->major
);
1825 static ssize_t
rbd_client_id_show(struct device
*dev
,
1826 struct device_attribute
*attr
, char *buf
)
1828 struct rbd_device
*rbd_dev
= dev_to_rbd(dev
);
1830 return sprintf(buf
, "client%lld\n", ceph_client_id(rbd_dev
->client
));
1833 static ssize_t
rbd_pool_show(struct device
*dev
,
1834 struct device_attribute
*attr
, char *buf
)
1836 struct rbd_device
*rbd_dev
= dev_to_rbd(dev
);
1838 return sprintf(buf
, "%s\n", rbd_dev
->pool_name
);
1841 static ssize_t
rbd_name_show(struct device
*dev
,
1842 struct device_attribute
*attr
, char *buf
)
1844 struct rbd_device
*rbd_dev
= dev_to_rbd(dev
);
1846 return sprintf(buf
, "%s\n", rbd_dev
->obj
);
1849 static ssize_t
rbd_snap_show(struct device
*dev
,
1850 struct device_attribute
*attr
,
1853 struct rbd_device
*rbd_dev
= dev_to_rbd(dev
);
1855 return sprintf(buf
, "%s\n", rbd_dev
->snap_name
);
1858 static ssize_t
rbd_image_refresh(struct device
*dev
,
1859 struct device_attribute
*attr
,
1863 struct rbd_device
*rbd_dev
= dev_to_rbd(dev
);
1867 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
1869 rc
= __rbd_update_snaps(rbd_dev
);
1873 mutex_unlock(&ctl_mutex
);
1877 static DEVICE_ATTR(size
, S_IRUGO
, rbd_size_show
, NULL
);
1878 static DEVICE_ATTR(major
, S_IRUGO
, rbd_major_show
, NULL
);
1879 static DEVICE_ATTR(client_id
, S_IRUGO
, rbd_client_id_show
, NULL
);
1880 static DEVICE_ATTR(pool
, S_IRUGO
, rbd_pool_show
, NULL
);
1881 static DEVICE_ATTR(name
, S_IRUGO
, rbd_name_show
, NULL
);
1882 static DEVICE_ATTR(refresh
, S_IWUSR
, NULL
, rbd_image_refresh
);
1883 static DEVICE_ATTR(current_snap
, S_IRUGO
, rbd_snap_show
, NULL
);
1884 static DEVICE_ATTR(create_snap
, S_IWUSR
, NULL
, rbd_snap_add
);
1885 static DEVICE_ATTR(rollback_snap
, S_IWUSR
, NULL
, rbd_snap_rollback
);
1887 static struct attribute
*rbd_attrs
[] = {
1888 &dev_attr_size
.attr
,
1889 &dev_attr_major
.attr
,
1890 &dev_attr_client_id
.attr
,
1891 &dev_attr_pool
.attr
,
1892 &dev_attr_name
.attr
,
1893 &dev_attr_current_snap
.attr
,
1894 &dev_attr_refresh
.attr
,
1895 &dev_attr_create_snap
.attr
,
1896 &dev_attr_rollback_snap
.attr
,
1900 static struct attribute_group rbd_attr_group
= {
1904 static const struct attribute_group
*rbd_attr_groups
[] = {
1909 static void rbd_sysfs_dev_release(struct device
*dev
)
1913 static struct device_type rbd_device_type
= {
1915 .groups
= rbd_attr_groups
,
1916 .release
= rbd_sysfs_dev_release
,
1924 static ssize_t
rbd_snap_size_show(struct device
*dev
,
1925 struct device_attribute
*attr
,
1928 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1930 return sprintf(buf
, "%lld\n", (long long)snap
->size
);
1933 static ssize_t
rbd_snap_id_show(struct device
*dev
,
1934 struct device_attribute
*attr
,
1937 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1939 return sprintf(buf
, "%lld\n", (long long)snap
->id
);
1942 static DEVICE_ATTR(snap_size
, S_IRUGO
, rbd_snap_size_show
, NULL
);
1943 static DEVICE_ATTR(snap_id
, S_IRUGO
, rbd_snap_id_show
, NULL
);
1945 static struct attribute
*rbd_snap_attrs
[] = {
1946 &dev_attr_snap_size
.attr
,
1947 &dev_attr_snap_id
.attr
,
1951 static struct attribute_group rbd_snap_attr_group
= {
1952 .attrs
= rbd_snap_attrs
,
1955 static void rbd_snap_dev_release(struct device
*dev
)
1957 struct rbd_snap
*snap
= container_of(dev
, struct rbd_snap
, dev
);
1962 static const struct attribute_group
*rbd_snap_attr_groups
[] = {
1963 &rbd_snap_attr_group
,
1967 static struct device_type rbd_snap_device_type
= {
1968 .groups
= rbd_snap_attr_groups
,
1969 .release
= rbd_snap_dev_release
,
1972 static void __rbd_remove_snap_dev(struct rbd_device
*rbd_dev
,
1973 struct rbd_snap
*snap
)
1975 list_del(&snap
->node
);
1976 device_unregister(&snap
->dev
);
1979 static int rbd_register_snap_dev(struct rbd_device
*rbd_dev
,
1980 struct rbd_snap
*snap
,
1981 struct device
*parent
)
1983 struct device
*dev
= &snap
->dev
;
1986 dev
->type
= &rbd_snap_device_type
;
1987 dev
->parent
= parent
;
1988 dev
->release
= rbd_snap_dev_release
;
1989 dev_set_name(dev
, "snap_%s", snap
->name
);
1990 ret
= device_register(dev
);
1995 static int __rbd_add_snap_dev(struct rbd_device
*rbd_dev
,
1996 int i
, const char *name
,
1997 struct rbd_snap
**snapp
)
2000 struct rbd_snap
*snap
= kzalloc(sizeof(*snap
), GFP_KERNEL
);
2003 snap
->name
= kstrdup(name
, GFP_KERNEL
);
2004 snap
->size
= rbd_dev
->header
.snap_sizes
[i
];
2005 snap
->id
= rbd_dev
->header
.snapc
->snaps
[i
];
2006 if (device_is_registered(&rbd_dev
->dev
)) {
2007 ret
= rbd_register_snap_dev(rbd_dev
, snap
,
2021 * search for the previous snap in a null delimited string list
2023 const char *rbd_prev_snap_name(const char *name
, const char *start
)
2025 if (name
< start
+ 2)
2038 * compare the old list of snapshots that we have to what's in the header
2039 * and update it accordingly. Note that the header holds the snapshots
2040 * in a reverse order (from newest to oldest) and we need to go from
2041 * older to new so that we don't get a duplicate snap name when
2042 * doing the process (e.g., removed snapshot and recreated a new
2043 * one with the same name.
2045 static int __rbd_init_snaps_header(struct rbd_device
*rbd_dev
)
2047 const char *name
, *first_name
;
2048 int i
= rbd_dev
->header
.total_snaps
;
2049 struct rbd_snap
*snap
, *old_snap
= NULL
;
2051 struct list_head
*p
, *n
;
2053 first_name
= rbd_dev
->header
.snap_names
;
2054 name
= first_name
+ rbd_dev
->header
.snap_names_len
;
2056 list_for_each_prev_safe(p
, n
, &rbd_dev
->snaps
) {
2059 old_snap
= list_entry(p
, struct rbd_snap
, node
);
2062 cur_id
= rbd_dev
->header
.snapc
->snaps
[i
- 1];
2064 if (!i
|| old_snap
->id
< cur_id
) {
2065 /* old_snap->id was skipped, thus was removed */
2066 __rbd_remove_snap_dev(rbd_dev
, old_snap
);
2069 if (old_snap
->id
== cur_id
) {
2070 /* we have this snapshot already */
2072 name
= rbd_prev_snap_name(name
, first_name
);
2076 i
--, name
= rbd_prev_snap_name(name
, first_name
)) {
2081 cur_id
= rbd_dev
->header
.snapc
->snaps
[i
];
2082 /* snapshot removal? handle it above */
2083 if (cur_id
>= old_snap
->id
)
2085 /* a new snapshot */
2086 ret
= __rbd_add_snap_dev(rbd_dev
, i
- 1, name
, &snap
);
2090 /* note that we add it backward so using n and not p */
2091 list_add(&snap
->node
, n
);
2095 /* we're done going over the old snap list, just add what's left */
2096 for (; i
> 0; i
--) {
2097 name
= rbd_prev_snap_name(name
, first_name
);
2102 ret
= __rbd_add_snap_dev(rbd_dev
, i
- 1, name
, &snap
);
2105 list_add(&snap
->node
, &rbd_dev
->snaps
);
2112 static void rbd_root_dev_release(struct device
*dev
)
2116 static struct device rbd_root_dev
= {
2118 .release
= rbd_root_dev_release
,
2121 static int rbd_bus_add_dev(struct rbd_device
*rbd_dev
)
2125 struct rbd_snap
*snap
;
2127 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2128 dev
= &rbd_dev
->dev
;
2130 dev
->bus
= &rbd_bus_type
;
2131 dev
->type
= &rbd_device_type
;
2132 dev
->parent
= &rbd_root_dev
;
2133 dev
->release
= rbd_dev_release
;
2134 dev_set_name(dev
, "%d", rbd_dev
->id
);
2135 ret
= device_register(dev
);
2139 list_for_each_entry(snap
, &rbd_dev
->snaps
, node
) {
2140 ret
= rbd_register_snap_dev(rbd_dev
, snap
,
2146 mutex_unlock(&ctl_mutex
);
2149 mutex_unlock(&ctl_mutex
);
2153 static void rbd_bus_del_dev(struct rbd_device
*rbd_dev
)
2155 device_unregister(&rbd_dev
->dev
);
2158 static int rbd_init_watch_dev(struct rbd_device
*rbd_dev
)
2163 ret
= rbd_req_sync_watch(rbd_dev
, rbd_dev
->obj_md_name
,
2164 rbd_dev
->header
.obj_version
);
2165 if (ret
== -ERANGE
) {
2166 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2167 rc
= __rbd_update_snaps(rbd_dev
);
2168 mutex_unlock(&ctl_mutex
);
2172 } while (ret
== -ERANGE
);
2177 static ssize_t
rbd_add(struct bus_type
*bus
,
2181 struct ceph_osd_client
*osdc
;
2182 struct rbd_device
*rbd_dev
;
2183 ssize_t rc
= -ENOMEM
;
2184 int irc
, new_id
= 0;
2185 struct list_head
*tmp
;
2189 if (!try_module_get(THIS_MODULE
))
2192 mon_dev_name
= kmalloc(RBD_MAX_OPT_LEN
, GFP_KERNEL
);
2196 options
= kmalloc(RBD_MAX_OPT_LEN
, GFP_KERNEL
);
2200 /* new rbd_device object */
2201 rbd_dev
= kzalloc(sizeof(*rbd_dev
), GFP_KERNEL
);
2205 /* static rbd_device initialization */
2206 spin_lock_init(&rbd_dev
->lock
);
2207 INIT_LIST_HEAD(&rbd_dev
->node
);
2208 INIT_LIST_HEAD(&rbd_dev
->snaps
);
2210 /* generate unique id: find highest unique id, add one */
2211 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2213 list_for_each(tmp
, &rbd_dev_list
) {
2214 struct rbd_device
*rbd_dev
;
2216 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
2217 if (rbd_dev
->id
>= new_id
)
2218 new_id
= rbd_dev
->id
+ 1;
2221 rbd_dev
->id
= new_id
;
2223 /* add to global list */
2224 list_add_tail(&rbd_dev
->node
, &rbd_dev_list
);
2226 /* parse add command */
2227 if (sscanf(buf
, "%" __stringify(RBD_MAX_OPT_LEN
) "s "
2228 "%" __stringify(RBD_MAX_OPT_LEN
) "s "
2229 "%" __stringify(RBD_MAX_POOL_NAME_LEN
) "s "
2230 "%" __stringify(RBD_MAX_OBJ_NAME_LEN
) "s"
2231 "%" __stringify(RBD_MAX_SNAP_NAME_LEN
) "s",
2232 mon_dev_name
, options
, rbd_dev
->pool_name
,
2233 rbd_dev
->obj
, rbd_dev
->snap_name
) < 4) {
2238 if (rbd_dev
->snap_name
[0] == 0)
2239 rbd_dev
->snap_name
[0] = '-';
2241 rbd_dev
->obj_len
= strlen(rbd_dev
->obj
);
2242 snprintf(rbd_dev
->obj_md_name
, sizeof(rbd_dev
->obj_md_name
), "%s%s",
2243 rbd_dev
->obj
, RBD_SUFFIX
);
2245 /* initialize rest of new object */
2246 snprintf(rbd_dev
->name
, DEV_NAME_LEN
, DRV_NAME
"%d", rbd_dev
->id
);
2247 rc
= rbd_get_client(rbd_dev
, mon_dev_name
, options
);
2251 mutex_unlock(&ctl_mutex
);
2254 osdc
= &rbd_dev
->client
->osdc
;
2255 rc
= ceph_pg_poolid_by_name(osdc
->osdmap
, rbd_dev
->pool_name
);
2257 goto err_out_client
;
2258 rbd_dev
->poolid
= rc
;
2260 /* register our block device */
2261 irc
= register_blkdev(0, rbd_dev
->name
);
2264 goto err_out_client
;
2266 rbd_dev
->major
= irc
;
2268 rc
= rbd_bus_add_dev(rbd_dev
);
2270 goto err_out_blkdev
;
2272 /* set up and announce blkdev mapping */
2273 rc
= rbd_init_disk(rbd_dev
);
2277 rc
= rbd_init_watch_dev(rbd_dev
);
2284 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2285 list_del_init(&rbd_dev
->node
);
2286 mutex_unlock(&ctl_mutex
);
2288 /* this will also clean up rest of rbd_dev stuff */
2290 rbd_bus_del_dev(rbd_dev
);
2292 kfree(mon_dev_name
);
2296 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
2298 rbd_put_client(rbd_dev
);
2299 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2301 list_del_init(&rbd_dev
->node
);
2302 mutex_unlock(&ctl_mutex
);
2308 kfree(mon_dev_name
);
2310 dout("Error adding device %s\n", buf
);
2311 module_put(THIS_MODULE
);
2315 static struct rbd_device
*__rbd_get_dev(unsigned long id
)
2317 struct list_head
*tmp
;
2318 struct rbd_device
*rbd_dev
;
2320 list_for_each(tmp
, &rbd_dev_list
) {
2321 rbd_dev
= list_entry(tmp
, struct rbd_device
, node
);
2322 if (rbd_dev
->id
== id
)
2328 static void rbd_dev_release(struct device
*dev
)
2330 struct rbd_device
*rbd_dev
=
2331 container_of(dev
, struct rbd_device
, dev
);
2333 if (rbd_dev
->watch_request
)
2334 ceph_osdc_unregister_linger_request(&rbd_dev
->client
->osdc
,
2335 rbd_dev
->watch_request
);
2336 if (rbd_dev
->watch_event
)
2337 rbd_req_sync_unwatch(rbd_dev
, rbd_dev
->obj_md_name
);
2339 rbd_put_client(rbd_dev
);
2341 /* clean up and free blkdev */
2342 rbd_free_disk(rbd_dev
);
2343 unregister_blkdev(rbd_dev
->major
, rbd_dev
->name
);
2346 /* release module ref */
2347 module_put(THIS_MODULE
);
2350 static ssize_t
rbd_remove(struct bus_type
*bus
,
2354 struct rbd_device
*rbd_dev
= NULL
;
2359 rc
= strict_strtoul(buf
, 10, &ul
);
2363 /* convert to int; abort if we lost anything in the conversion */
2364 target_id
= (int) ul
;
2365 if (target_id
!= ul
)
2368 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2370 rbd_dev
= __rbd_get_dev(target_id
);
2376 list_del_init(&rbd_dev
->node
);
2378 __rbd_remove_all_snaps(rbd_dev
);
2379 rbd_bus_del_dev(rbd_dev
);
2382 mutex_unlock(&ctl_mutex
);
2386 static ssize_t
rbd_snap_add(struct device
*dev
,
2387 struct device_attribute
*attr
,
2391 struct rbd_device
*rbd_dev
= dev_to_rbd(dev
);
2393 char *name
= kmalloc(count
+ 1, GFP_KERNEL
);
2397 snprintf(name
, count
, "%s", buf
);
2399 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2401 ret
= rbd_header_add_snap(rbd_dev
,
2406 ret
= __rbd_update_snaps(rbd_dev
);
2410 /* shouldn't hold ctl_mutex when notifying.. notify might
2411 trigger a watch callback that would need to get that mutex */
2412 mutex_unlock(&ctl_mutex
);
2414 /* make a best effort, don't error if failed */
2415 rbd_req_sync_notify(rbd_dev
, rbd_dev
->obj_md_name
);
2422 mutex_unlock(&ctl_mutex
);
2427 static ssize_t
rbd_snap_rollback(struct device
*dev
,
2428 struct device_attribute
*attr
,
2432 struct rbd_device
*rbd_dev
= dev_to_rbd(dev
);
2436 char *seg_name
= NULL
;
2437 char *snap_name
= kmalloc(count
+ 1, GFP_KERNEL
);
2442 /* parse snaps add command */
2443 snprintf(snap_name
, count
, "%s", buf
);
2444 seg_name
= kmalloc(RBD_MAX_SEG_NAME_LEN
+ 1, GFP_NOIO
);
2448 mutex_lock_nested(&ctl_mutex
, SINGLE_DEPTH_NESTING
);
2450 ret
= snap_by_name(&rbd_dev
->header
, snap_name
, &snapid
, NULL
);
2454 dout("snapid=%lld\n", snapid
);
2457 while (cur_ofs
< rbd_dev
->header
.image_size
) {
2458 cur_ofs
+= rbd_get_segment(&rbd_dev
->header
,
2462 dout("seg_name=%s\n", seg_name
);
2464 ret
= rbd_req_sync_rollback_obj(rbd_dev
, snapid
, seg_name
);
2466 pr_warning("could not roll back obj %s err=%d\n",
2470 ret
= __rbd_update_snaps(rbd_dev
);
2477 mutex_unlock(&ctl_mutex
);
2485 static struct bus_attribute rbd_bus_attrs
[] = {
2486 __ATTR(add
, S_IWUSR
, NULL
, rbd_add
),
2487 __ATTR(remove
, S_IWUSR
, NULL
, rbd_remove
),
2492 * create control files in sysfs
2495 static int rbd_sysfs_init(void)
2499 rbd_bus_type
.bus_attrs
= rbd_bus_attrs
;
2501 ret
= bus_register(&rbd_bus_type
);
2505 ret
= device_register(&rbd_root_dev
);
2510 static void rbd_sysfs_cleanup(void)
2512 device_unregister(&rbd_root_dev
);
2513 bus_unregister(&rbd_bus_type
);
2516 int __init
rbd_init(void)
2520 rc
= rbd_sysfs_init();
2523 spin_lock_init(&node_lock
);
2524 pr_info("loaded " DRV_NAME_LONG
"\n");
2528 void __exit
rbd_exit(void)
2530 rbd_sysfs_cleanup();
2533 module_init(rbd_init
);
2534 module_exit(rbd_exit
);
2536 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
2537 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
2538 MODULE_DESCRIPTION("rados block device");
2540 /* following authorship retained from original osdblk.c */
2541 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
2543 MODULE_LICENSE("GPL");