3 rbd.c -- Export ceph rados objects as a Linux block device
6 based on drivers/block/osdblk.c:
8 Copyright 2009 Red Hat, Inc.
10 This program is free software; you can redistribute it and/or modify
11 it under the terms of the GNU General Public License as published by
12 the Free Software Foundation.
14 This program is distributed in the hope that it will be useful,
15 but WITHOUT ANY WARRANTY; without even the implied warranty of
16 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 GNU General Public License for more details.
19 You should have received a copy of the GNU General Public License
20 along with this program; see the file COPYING. If not, write to
21 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
25 For usage instructions, please refer to:
27 Documentation/ABI/testing/sysfs-bus-rbd
31 #include <linux/ceph/libceph.h>
32 #include <linux/ceph/osd_client.h>
33 #include <linux/ceph/mon_client.h>
34 #include <linux/ceph/cls_lock_client.h>
35 #include <linux/ceph/striper.h>
36 #include <linux/ceph/decode.h>
37 #include <linux/fs_parser.h>
38 #include <linux/bsearch.h>
40 #include <linux/kernel.h>
41 #include <linux/device.h>
42 #include <linux/module.h>
43 #include <linux/blk-mq.h>
45 #include <linux/blkdev.h>
46 #include <linux/slab.h>
47 #include <linux/idr.h>
48 #include <linux/workqueue.h>
50 #include "rbd_types.h"
52 #define RBD_DEBUG /* Activate rbd_assert() calls */
55 * Increment the given counter and return its updated value.
56 * If the counter is already 0 it will not be incremented.
57 * If the counter is already at its maximum value returns
58 * -EINVAL without updating it.
60 static int atomic_inc_return_safe(atomic_t *v)
64 counter = (unsigned int)atomic_fetch_add_unless(v, 1, 0);
65 if (counter <= (unsigned int)INT_MAX)
73 /* Decrement the counter. Return the resulting value, or -EINVAL */
74 static int atomic_dec_return_safe(atomic_t *v)
78 counter = atomic_dec_return(v);
87 #define RBD_DRV_NAME "rbd"
89 #define RBD_MINORS_PER_MAJOR 256
90 #define RBD_SINGLE_MAJOR_PART_SHIFT 4
92 #define RBD_MAX_PARENT_CHAIN_LEN 16
94 #define RBD_SNAP_DEV_NAME_PREFIX "snap_"
95 #define RBD_MAX_SNAP_NAME_LEN \
96 (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
98 #define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */
100 #define RBD_SNAP_HEAD_NAME "-"
102 #define BAD_SNAP_INDEX U32_MAX /* invalid index into snap array */
104 /* This allows a single page to hold an image name sent by OSD */
105 #define RBD_IMAGE_NAME_LEN_MAX (PAGE_SIZE - sizeof (__le32) - 1)
106 #define RBD_IMAGE_ID_LEN_MAX 64
108 #define RBD_OBJ_PREFIX_LEN_MAX 64
110 #define RBD_NOTIFY_TIMEOUT 5 /* seconds */
111 #define RBD_RETRY_DELAY msecs_to_jiffies(1000)
115 #define RBD_FEATURE_LAYERING (1ULL<<0)
116 #define RBD_FEATURE_STRIPINGV2 (1ULL<<1)
117 #define RBD_FEATURE_EXCLUSIVE_LOCK (1ULL<<2)
118 #define RBD_FEATURE_OBJECT_MAP (1ULL<<3)
119 #define RBD_FEATURE_FAST_DIFF (1ULL<<4)
120 #define RBD_FEATURE_DEEP_FLATTEN (1ULL<<5)
121 #define RBD_FEATURE_DATA_POOL (1ULL<<7)
122 #define RBD_FEATURE_OPERATIONS (1ULL<<8)
124 #define RBD_FEATURES_ALL (RBD_FEATURE_LAYERING | \
125 RBD_FEATURE_STRIPINGV2 | \
126 RBD_FEATURE_EXCLUSIVE_LOCK | \
127 RBD_FEATURE_OBJECT_MAP | \
128 RBD_FEATURE_FAST_DIFF | \
129 RBD_FEATURE_DEEP_FLATTEN | \
130 RBD_FEATURE_DATA_POOL | \
131 RBD_FEATURE_OPERATIONS)
133 /* Features supported by this (client software) implementation. */
135 #define RBD_FEATURES_SUPPORTED (RBD_FEATURES_ALL)
138 * An RBD device name will be "rbd#", where the "rbd" comes from
139 * RBD_DRV_NAME above, and # is a unique integer identifier.
141 #define DEV_NAME_LEN 32
144 * block device image metadata (in-memory version)
146 struct rbd_image_header {
147 /* These six fields never change for a given rbd image */
153 u64 features; /* Might be changeable someday? */
155 /* The remaining fields need to be updated occasionally */
157 struct ceph_snap_context *snapc;
158 char *snap_names; /* format 1 only */
159 u64 *snap_sizes; /* format 1 only */
163 * An rbd image specification.
165 * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
166 * identify an image. Each rbd_dev structure includes a pointer to
167 * an rbd_spec structure that encapsulates this identity.
169 * Each of the id's in an rbd_spec has an associated name. For a
170 * user-mapped image, the names are supplied and the id's associated
171 * with them are looked up. For a layered image, a parent image is
172 * defined by the tuple, and the names are looked up.
174 * An rbd_dev structure contains a parent_spec pointer which is
175 * non-null if the image it represents is a child in a layered
176 * image. This pointer will refer to the rbd_spec structure used
177 * by the parent rbd_dev for its own identity (i.e., the structure
178 * is shared between the parent and child).
180 * Since these structures are populated once, during the discovery
181 * phase of image construction, they are effectively immutable so
182 * we make no effort to synchronize access to them.
184 * Note that code herein does not assume the image name is known (it
185 * could be a null pointer).
189 const char *pool_name;
190 const char *pool_ns; /* NULL if default, never "" */
192 const char *image_id;
193 const char *image_name;
196 const char *snap_name;
202 * an instance of the client. multiple devices may share an rbd client.
205 struct ceph_client *client;
207 struct list_head node;
210 struct pending_result {
211 int result; /* first nonzero result */
215 struct rbd_img_request;
217 enum obj_request_type {
218 OBJ_REQUEST_NODATA = 1,
219 OBJ_REQUEST_BIO, /* pointer into provided bio (list) */
220 OBJ_REQUEST_BVECS, /* pointer into provided bio_vec array */
221 OBJ_REQUEST_OWN_BVECS, /* private bio_vec array, doesn't own pages */
224 enum obj_operation_type {
231 #define RBD_OBJ_FLAG_DELETION (1U << 0)
232 #define RBD_OBJ_FLAG_COPYUP_ENABLED (1U << 1)
233 #define RBD_OBJ_FLAG_COPYUP_ZEROS (1U << 2)
234 #define RBD_OBJ_FLAG_MAY_EXIST (1U << 3)
235 #define RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT (1U << 4)
237 enum rbd_obj_read_state {
238 RBD_OBJ_READ_START = 1,
244 * Writes go through the following state machine to deal with
247 * . . . . . RBD_OBJ_WRITE_GUARD. . . . . . . . . . . . . .
250 * . RBD_OBJ_WRITE_READ_FROM_PARENT. . . .
252 * . v v (deep-copyup .
253 * (image . RBD_OBJ_WRITE_COPYUP_EMPTY_SNAPC . not needed) .
256 * . . . .RBD_OBJ_WRITE_COPYUP_OPS. . . . . (copyup .
259 * done . . . . . . . . . . . . . . . . . .
264 * Writes start in RBD_OBJ_WRITE_GUARD or _FLAT, depending on whether
265 * assert_exists guard is needed or not (in some cases it's not needed
266 * even if there is a parent).
268 enum rbd_obj_write_state {
269 RBD_OBJ_WRITE_START = 1,
270 RBD_OBJ_WRITE_PRE_OBJECT_MAP,
271 RBD_OBJ_WRITE_OBJECT,
272 __RBD_OBJ_WRITE_COPYUP,
273 RBD_OBJ_WRITE_COPYUP,
274 RBD_OBJ_WRITE_POST_OBJECT_MAP,
277 enum rbd_obj_copyup_state {
278 RBD_OBJ_COPYUP_START = 1,
279 RBD_OBJ_COPYUP_READ_PARENT,
280 __RBD_OBJ_COPYUP_OBJECT_MAPS,
281 RBD_OBJ_COPYUP_OBJECT_MAPS,
282 __RBD_OBJ_COPYUP_WRITE_OBJECT,
283 RBD_OBJ_COPYUP_WRITE_OBJECT,
286 struct rbd_obj_request {
287 struct ceph_object_extent ex;
288 unsigned int flags; /* RBD_OBJ_FLAG_* */
290 enum rbd_obj_read_state read_state; /* for reads */
291 enum rbd_obj_write_state write_state; /* for writes */
294 struct rbd_img_request *img_request;
295 struct ceph_file_extent *img_extents;
299 struct ceph_bio_iter bio_pos;
301 struct ceph_bvec_iter bvec_pos;
307 enum rbd_obj_copyup_state copyup_state;
308 struct bio_vec *copyup_bvecs;
309 u32 copyup_bvec_count;
311 struct list_head osd_reqs; /* w/ r_private_item */
313 struct mutex state_mutex;
314 struct pending_result pending;
319 IMG_REQ_CHILD, /* initiator: block = 0, child image = 1 */
320 IMG_REQ_LAYERED, /* ENOENT handling: normal = 0, layered = 1 */
325 RBD_IMG_EXCLUSIVE_LOCK,
326 __RBD_IMG_OBJECT_REQUESTS,
327 RBD_IMG_OBJECT_REQUESTS,
330 struct rbd_img_request {
331 struct rbd_device *rbd_dev;
332 enum obj_operation_type op_type;
333 enum obj_request_type data_type;
335 enum rbd_img_state state;
337 u64 snap_id; /* for reads */
338 struct ceph_snap_context *snapc; /* for writes */
340 struct rbd_obj_request *obj_request; /* obj req initiator */
342 struct list_head lock_item;
343 struct list_head object_extents; /* obj_req.ex structs */
345 struct mutex state_mutex;
346 struct pending_result pending;
347 struct work_struct work;
351 #define for_each_obj_request(ireq, oreq) \
352 list_for_each_entry(oreq, &(ireq)->object_extents, ex.oe_item)
353 #define for_each_obj_request_safe(ireq, oreq, n) \
354 list_for_each_entry_safe(oreq, n, &(ireq)->object_extents, ex.oe_item)
356 enum rbd_watch_state {
357 RBD_WATCH_STATE_UNREGISTERED,
358 RBD_WATCH_STATE_REGISTERED,
359 RBD_WATCH_STATE_ERROR,
362 enum rbd_lock_state {
363 RBD_LOCK_STATE_UNLOCKED,
364 RBD_LOCK_STATE_LOCKED,
365 RBD_LOCK_STATE_RELEASING,
368 /* WatchNotify::ClientId */
369 struct rbd_client_id {
382 int dev_id; /* blkdev unique id */
384 int major; /* blkdev assigned major */
386 struct gendisk *disk; /* blkdev's gendisk and rq */
388 u32 image_format; /* Either 1 or 2 */
389 struct rbd_client *rbd_client;
391 char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
393 spinlock_t lock; /* queue, flags, open_count */
395 struct rbd_image_header header;
396 unsigned long flags; /* possibly lock protected */
397 struct rbd_spec *spec;
398 struct rbd_options *opts;
399 char *config_info; /* add{,_single_major} string */
401 struct ceph_object_id header_oid;
402 struct ceph_object_locator header_oloc;
404 struct ceph_file_layout layout; /* used for all rbd requests */
406 struct mutex watch_mutex;
407 enum rbd_watch_state watch_state;
408 struct ceph_osd_linger_request *watch_handle;
410 struct delayed_work watch_dwork;
412 struct rw_semaphore lock_rwsem;
413 enum rbd_lock_state lock_state;
414 char lock_cookie[32];
415 struct rbd_client_id owner_cid;
416 struct work_struct acquired_lock_work;
417 struct work_struct released_lock_work;
418 struct delayed_work lock_dwork;
419 struct work_struct unlock_work;
420 spinlock_t lock_lists_lock;
421 struct list_head acquiring_list;
422 struct list_head running_list;
423 struct completion acquire_wait;
425 struct completion releasing_wait;
427 spinlock_t object_map_lock;
429 u64 object_map_size; /* in objects */
430 u64 object_map_flags;
432 struct workqueue_struct *task_wq;
434 struct rbd_spec *parent_spec;
437 struct rbd_device *parent;
439 /* Block layer tags. */
440 struct blk_mq_tag_set tag_set;
442 /* protects updating the header */
443 struct rw_semaphore header_rwsem;
445 struct rbd_mapping mapping;
447 struct list_head node;
451 unsigned long open_count; /* protected by lock */
455 * Flag bits for rbd_dev->flags:
456 * - REMOVING (which is coupled with rbd_dev->open_count) is protected
460 RBD_DEV_FLAG_EXISTS, /* rbd_dev_device_setup() ran */
461 RBD_DEV_FLAG_REMOVING, /* this mapping is being removed */
462 RBD_DEV_FLAG_READONLY, /* -o ro or snapshot */
465 static DEFINE_MUTEX(client_mutex); /* Serialize client creation */
467 static LIST_HEAD(rbd_dev_list); /* devices */
468 static DEFINE_SPINLOCK(rbd_dev_list_lock);
470 static LIST_HEAD(rbd_client_list); /* clients */
471 static DEFINE_SPINLOCK(rbd_client_list_lock);
473 /* Slab caches for frequently-allocated structures */
475 static struct kmem_cache *rbd_img_request_cache;
476 static struct kmem_cache *rbd_obj_request_cache;
478 static int rbd_major;
479 static DEFINE_IDA(rbd_dev_id_ida);
481 static struct workqueue_struct *rbd_wq;
483 static struct ceph_snap_context rbd_empty_snapc = {
484 .nref = REFCOUNT_INIT(1),
488 * single-major requires >= 0.75 version of userspace rbd utility.
490 static bool single_major = true;
491 module_param(single_major, bool, 0444);
492 MODULE_PARM_DESC(single_major, "Use a single major number for all rbd devices (default: true)");
494 static ssize_t add_store(const struct bus_type *bus, const char *buf, size_t count);
495 static ssize_t remove_store(const struct bus_type *bus, const char *buf,
497 static ssize_t add_single_major_store(const struct bus_type *bus, const char *buf,
499 static ssize_t remove_single_major_store(const struct bus_type *bus, const char *buf,
501 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth);
503 static int rbd_dev_id_to_minor(int dev_id)
505 return dev_id << RBD_SINGLE_MAJOR_PART_SHIFT;
508 static int minor_to_rbd_dev_id(int minor)
510 return minor >> RBD_SINGLE_MAJOR_PART_SHIFT;
513 static bool rbd_is_ro(struct rbd_device *rbd_dev)
515 return test_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
518 static bool rbd_is_snap(struct rbd_device *rbd_dev)
520 return rbd_dev->spec->snap_id != CEPH_NOSNAP;
523 static bool __rbd_is_lock_owner(struct rbd_device *rbd_dev)
525 lockdep_assert_held(&rbd_dev->lock_rwsem);
527 return rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED ||
528 rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING;
531 static bool rbd_is_lock_owner(struct rbd_device *rbd_dev)
535 down_read(&rbd_dev->lock_rwsem);
536 is_lock_owner = __rbd_is_lock_owner(rbd_dev);
537 up_read(&rbd_dev->lock_rwsem);
538 return is_lock_owner;
541 static ssize_t supported_features_show(const struct bus_type *bus, char *buf)
543 return sprintf(buf, "0x%llx\n", RBD_FEATURES_SUPPORTED);
546 static BUS_ATTR_WO(add);
547 static BUS_ATTR_WO(remove);
548 static BUS_ATTR_WO(add_single_major);
549 static BUS_ATTR_WO(remove_single_major);
550 static BUS_ATTR_RO(supported_features);
552 static struct attribute *rbd_bus_attrs[] = {
554 &bus_attr_remove.attr,
555 &bus_attr_add_single_major.attr,
556 &bus_attr_remove_single_major.attr,
557 &bus_attr_supported_features.attr,
561 static umode_t rbd_bus_is_visible(struct kobject *kobj,
562 struct attribute *attr, int index)
565 (attr == &bus_attr_add_single_major.attr ||
566 attr == &bus_attr_remove_single_major.attr))
572 static const struct attribute_group rbd_bus_group = {
573 .attrs = rbd_bus_attrs,
574 .is_visible = rbd_bus_is_visible,
576 __ATTRIBUTE_GROUPS(rbd_bus);
578 static struct bus_type rbd_bus_type = {
580 .bus_groups = rbd_bus_groups,
583 static void rbd_root_dev_release(struct device *dev)
587 static struct device rbd_root_dev = {
589 .release = rbd_root_dev_release,
592 static __printf(2, 3)
593 void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
595 struct va_format vaf;
603 printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
604 else if (rbd_dev->disk)
605 printk(KERN_WARNING "%s: %s: %pV\n",
606 RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
607 else if (rbd_dev->spec && rbd_dev->spec->image_name)
608 printk(KERN_WARNING "%s: image %s: %pV\n",
609 RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
610 else if (rbd_dev->spec && rbd_dev->spec->image_id)
611 printk(KERN_WARNING "%s: id %s: %pV\n",
612 RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
614 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
615 RBD_DRV_NAME, rbd_dev, &vaf);
620 #define rbd_assert(expr) \
621 if (unlikely(!(expr))) { \
622 printk(KERN_ERR "\nAssertion failure in %s() " \
624 "\trbd_assert(%s);\n\n", \
625 __func__, __LINE__, #expr); \
628 #else /* !RBD_DEBUG */
629 # define rbd_assert(expr) ((void) 0)
630 #endif /* !RBD_DEBUG */
632 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev);
634 static int rbd_dev_refresh(struct rbd_device *rbd_dev);
635 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev,
636 struct rbd_image_header *header);
637 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
639 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
640 u8 *order, u64 *snap_size);
641 static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev);
643 static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result);
644 static void rbd_img_handle_request(struct rbd_img_request *img_req, int result);
647 * Return true if nothing else is pending.
649 static bool pending_result_dec(struct pending_result *pending, int *result)
651 rbd_assert(pending->num_pending > 0);
653 if (*result && !pending->result)
654 pending->result = *result;
655 if (--pending->num_pending)
658 *result = pending->result;
662 static int rbd_open(struct gendisk *disk, blk_mode_t mode)
664 struct rbd_device *rbd_dev = disk->private_data;
665 bool removing = false;
667 spin_lock_irq(&rbd_dev->lock);
668 if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
671 rbd_dev->open_count++;
672 spin_unlock_irq(&rbd_dev->lock);
676 (void) get_device(&rbd_dev->dev);
681 static void rbd_release(struct gendisk *disk)
683 struct rbd_device *rbd_dev = disk->private_data;
684 unsigned long open_count_before;
686 spin_lock_irq(&rbd_dev->lock);
687 open_count_before = rbd_dev->open_count--;
688 spin_unlock_irq(&rbd_dev->lock);
689 rbd_assert(open_count_before > 0);
691 put_device(&rbd_dev->dev);
694 static const struct block_device_operations rbd_bd_ops = {
695 .owner = THIS_MODULE,
697 .release = rbd_release,
701 * Initialize an rbd client instance. Success or not, this function
702 * consumes ceph_opts. Caller holds client_mutex.
704 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
706 struct rbd_client *rbdc;
709 dout("%s:\n", __func__);
710 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
714 kref_init(&rbdc->kref);
715 INIT_LIST_HEAD(&rbdc->node);
717 rbdc->client = ceph_create_client(ceph_opts, rbdc);
718 if (IS_ERR(rbdc->client))
720 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
722 ret = ceph_open_session(rbdc->client);
726 spin_lock(&rbd_client_list_lock);
727 list_add_tail(&rbdc->node, &rbd_client_list);
728 spin_unlock(&rbd_client_list_lock);
730 dout("%s: rbdc %p\n", __func__, rbdc);
734 ceph_destroy_client(rbdc->client);
739 ceph_destroy_options(ceph_opts);
740 dout("%s: error %d\n", __func__, ret);
745 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
747 kref_get(&rbdc->kref);
753 * Find a ceph client with specific addr and configuration. If
754 * found, bump its reference count.
756 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
758 struct rbd_client *rbdc = NULL, *iter;
760 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
763 spin_lock(&rbd_client_list_lock);
764 list_for_each_entry(iter, &rbd_client_list, node) {
765 if (!ceph_compare_options(ceph_opts, iter->client)) {
766 __rbd_get_client(iter);
772 spin_unlock(&rbd_client_list_lock);
778 * (Per device) rbd map options
786 Opt_compression_hint,
787 /* string args above */
796 Opt_compression_hint_none,
797 Opt_compression_hint_compressible,
798 Opt_compression_hint_incompressible,
801 static const struct constant_table rbd_param_compression_hint[] = {
802 {"none", Opt_compression_hint_none},
803 {"compressible", Opt_compression_hint_compressible},
804 {"incompressible", Opt_compression_hint_incompressible},
808 static const struct fs_parameter_spec rbd_parameters[] = {
809 fsparam_u32 ("alloc_size", Opt_alloc_size),
810 fsparam_enum ("compression_hint", Opt_compression_hint,
811 rbd_param_compression_hint),
812 fsparam_flag ("exclusive", Opt_exclusive),
813 fsparam_flag ("lock_on_read", Opt_lock_on_read),
814 fsparam_u32 ("lock_timeout", Opt_lock_timeout),
815 fsparam_flag ("notrim", Opt_notrim),
816 fsparam_string ("_pool_ns", Opt_pool_ns),
817 fsparam_u32 ("queue_depth", Opt_queue_depth),
818 fsparam_flag ("read_only", Opt_read_only),
819 fsparam_flag ("read_write", Opt_read_write),
820 fsparam_flag ("ro", Opt_read_only),
821 fsparam_flag ("rw", Opt_read_write),
828 unsigned long lock_timeout;
834 u32 alloc_hint_flags; /* CEPH_OSD_OP_ALLOC_HINT_FLAG_* */
837 #define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_DEFAULT_RQ
838 #define RBD_ALLOC_SIZE_DEFAULT (64 * 1024)
839 #define RBD_LOCK_TIMEOUT_DEFAULT 0 /* no timeout */
840 #define RBD_READ_ONLY_DEFAULT false
841 #define RBD_LOCK_ON_READ_DEFAULT false
842 #define RBD_EXCLUSIVE_DEFAULT false
843 #define RBD_TRIM_DEFAULT true
845 struct rbd_parse_opts_ctx {
846 struct rbd_spec *spec;
847 struct ceph_options *copts;
848 struct rbd_options *opts;
851 static char* obj_op_name(enum obj_operation_type op_type)
868 * Destroy ceph client
870 * Caller must hold rbd_client_list_lock.
872 static void rbd_client_release(struct kref *kref)
874 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
876 dout("%s: rbdc %p\n", __func__, rbdc);
877 spin_lock(&rbd_client_list_lock);
878 list_del(&rbdc->node);
879 spin_unlock(&rbd_client_list_lock);
881 ceph_destroy_client(rbdc->client);
886 * Drop reference to ceph client node. If it's not referenced anymore, release
889 static void rbd_put_client(struct rbd_client *rbdc)
892 kref_put(&rbdc->kref, rbd_client_release);
896 * Get a ceph client with specific addr and configuration, if one does
897 * not exist create it. Either way, ceph_opts is consumed by this
900 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
902 struct rbd_client *rbdc;
905 mutex_lock(&client_mutex);
906 rbdc = rbd_client_find(ceph_opts);
908 ceph_destroy_options(ceph_opts);
911 * Using an existing client. Make sure ->pg_pools is up to
912 * date before we look up the pool id in do_rbd_add().
914 ret = ceph_wait_for_latest_osdmap(rbdc->client,
915 rbdc->client->options->mount_timeout);
917 rbd_warn(NULL, "failed to get latest osdmap: %d", ret);
918 rbd_put_client(rbdc);
922 rbdc = rbd_client_create(ceph_opts);
924 mutex_unlock(&client_mutex);
929 static bool rbd_image_format_valid(u32 image_format)
931 return image_format == 1 || image_format == 2;
934 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
939 /* The header has to start with the magic rbd header text */
940 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
943 /* The bio layer requires at least sector-sized I/O */
945 if (ondisk->options.order < SECTOR_SHIFT)
948 /* If we use u64 in a few spots we may be able to loosen this */
950 if (ondisk->options.order > 8 * sizeof (int) - 1)
954 * The size of a snapshot header has to fit in a size_t, and
955 * that limits the number of snapshots.
957 snap_count = le32_to_cpu(ondisk->snap_count);
958 size = SIZE_MAX - sizeof (struct ceph_snap_context);
959 if (snap_count > size / sizeof (__le64))
963 * Not only that, but the size of the entire the snapshot
964 * header must also be representable in a size_t.
966 size -= snap_count * sizeof (__le64);
967 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
974 * returns the size of an object in the image
976 static u32 rbd_obj_bytes(struct rbd_image_header *header)
978 return 1U << header->obj_order;
981 static void rbd_init_layout(struct rbd_device *rbd_dev)
983 if (rbd_dev->header.stripe_unit == 0 ||
984 rbd_dev->header.stripe_count == 0) {
985 rbd_dev->header.stripe_unit = rbd_obj_bytes(&rbd_dev->header);
986 rbd_dev->header.stripe_count = 1;
989 rbd_dev->layout.stripe_unit = rbd_dev->header.stripe_unit;
990 rbd_dev->layout.stripe_count = rbd_dev->header.stripe_count;
991 rbd_dev->layout.object_size = rbd_obj_bytes(&rbd_dev->header);
992 rbd_dev->layout.pool_id = rbd_dev->header.data_pool_id == CEPH_NOPOOL ?
993 rbd_dev->spec->pool_id : rbd_dev->header.data_pool_id;
994 RCU_INIT_POINTER(rbd_dev->layout.pool_ns, NULL);
997 static void rbd_image_header_cleanup(struct rbd_image_header *header)
999 kfree(header->object_prefix);
1000 ceph_put_snap_context(header->snapc);
1001 kfree(header->snap_sizes);
1002 kfree(header->snap_names);
1004 memset(header, 0, sizeof(*header));
1008 * Fill an rbd image header with information from the given format 1
1011 static int rbd_header_from_disk(struct rbd_image_header *header,
1012 struct rbd_image_header_ondisk *ondisk,
1015 struct ceph_snap_context *snapc;
1016 char *object_prefix = NULL;
1017 char *snap_names = NULL;
1018 u64 *snap_sizes = NULL;
1023 /* Allocate this now to avoid having to handle failure below */
1026 object_prefix = kstrndup(ondisk->object_prefix,
1027 sizeof(ondisk->object_prefix),
1033 /* Allocate the snapshot context and fill it in */
1035 snap_count = le32_to_cpu(ondisk->snap_count);
1036 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
1039 snapc->seq = le64_to_cpu(ondisk->snap_seq);
1041 struct rbd_image_snap_ondisk *snaps;
1042 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
1044 /* We'll keep a copy of the snapshot names... */
1046 if (snap_names_len > (u64)SIZE_MAX)
1048 snap_names = kmalloc(snap_names_len, GFP_KERNEL);
1052 /* ...as well as the array of their sizes. */
1053 snap_sizes = kmalloc_array(snap_count,
1054 sizeof(*header->snap_sizes),
1060 * Copy the names, and fill in each snapshot's id
1063 * Note that rbd_dev_v1_header_info() guarantees the
1064 * ondisk buffer we're working with has
1065 * snap_names_len bytes beyond the end of the
1066 * snapshot id array, this memcpy() is safe.
1068 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
1069 snaps = ondisk->snaps;
1070 for (i = 0; i < snap_count; i++) {
1071 snapc->snaps[i] = le64_to_cpu(snaps[i].id);
1072 snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
1076 /* We won't fail any more, fill in the header */
1079 header->object_prefix = object_prefix;
1080 header->obj_order = ondisk->options.order;
1083 /* The remaining fields always get updated (when we refresh) */
1085 header->image_size = le64_to_cpu(ondisk->image_size);
1086 header->snapc = snapc;
1087 header->snap_names = snap_names;
1088 header->snap_sizes = snap_sizes;
1096 ceph_put_snap_context(snapc);
1097 kfree(object_prefix);
1102 static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1104 const char *snap_name;
1106 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1108 /* Skip over names until we find the one we are looking for */
1110 snap_name = rbd_dev->header.snap_names;
1112 snap_name += strlen(snap_name) + 1;
1114 return kstrdup(snap_name, GFP_KERNEL);
1118 * Snapshot id comparison function for use with qsort()/bsearch().
1119 * Note that result is for snapshots in *descending* order.
1121 static int snapid_compare_reverse(const void *s1, const void *s2)
1123 u64 snap_id1 = *(u64 *)s1;
1124 u64 snap_id2 = *(u64 *)s2;
1126 if (snap_id1 < snap_id2)
1128 return snap_id1 == snap_id2 ? 0 : -1;
1132 * Search a snapshot context to see if the given snapshot id is
1135 * Returns the position of the snapshot id in the array if it's found,
1136 * or BAD_SNAP_INDEX otherwise.
1138 * Note: The snapshot array is in kept sorted (by the osd) in
1139 * reverse order, highest snapshot id first.
1141 static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1143 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1146 found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1147 sizeof (snap_id), snapid_compare_reverse);
1149 return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1152 static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1156 const char *snap_name;
1158 which = rbd_dev_snap_index(rbd_dev, snap_id);
1159 if (which == BAD_SNAP_INDEX)
1160 return ERR_PTR(-ENOENT);
1162 snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1163 return snap_name ? snap_name : ERR_PTR(-ENOMEM);
1166 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1168 if (snap_id == CEPH_NOSNAP)
1169 return RBD_SNAP_HEAD_NAME;
1171 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1172 if (rbd_dev->image_format == 1)
1173 return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1175 return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1178 static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1181 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1182 if (snap_id == CEPH_NOSNAP) {
1183 *snap_size = rbd_dev->header.image_size;
1184 } else if (rbd_dev->image_format == 1) {
1187 which = rbd_dev_snap_index(rbd_dev, snap_id);
1188 if (which == BAD_SNAP_INDEX)
1191 *snap_size = rbd_dev->header.snap_sizes[which];
1196 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1205 static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1207 u64 snap_id = rbd_dev->spec->snap_id;
1211 ret = rbd_snap_size(rbd_dev, snap_id, &size);
1215 rbd_dev->mapping.size = size;
1219 static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1221 rbd_dev->mapping.size = 0;
1224 static void zero_bios(struct ceph_bio_iter *bio_pos, u32 off, u32 bytes)
1226 struct ceph_bio_iter it = *bio_pos;
1228 ceph_bio_iter_advance(&it, off);
1229 ceph_bio_iter_advance_step(&it, bytes, ({
1234 static void zero_bvecs(struct ceph_bvec_iter *bvec_pos, u32 off, u32 bytes)
1236 struct ceph_bvec_iter it = *bvec_pos;
1238 ceph_bvec_iter_advance(&it, off);
1239 ceph_bvec_iter_advance_step(&it, bytes, ({
1245 * Zero a range in @obj_req data buffer defined by a bio (list) or
1246 * (private) bio_vec array.
1248 * @off is relative to the start of the data buffer.
1250 static void rbd_obj_zero_range(struct rbd_obj_request *obj_req, u32 off,
1253 dout("%s %p data buf %u~%u\n", __func__, obj_req, off, bytes);
1255 switch (obj_req->img_request->data_type) {
1256 case OBJ_REQUEST_BIO:
1257 zero_bios(&obj_req->bio_pos, off, bytes);
1259 case OBJ_REQUEST_BVECS:
1260 case OBJ_REQUEST_OWN_BVECS:
1261 zero_bvecs(&obj_req->bvec_pos, off, bytes);
1268 static void rbd_obj_request_destroy(struct kref *kref);
1269 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1271 rbd_assert(obj_request != NULL);
1272 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1273 kref_read(&obj_request->kref));
1274 kref_put(&obj_request->kref, rbd_obj_request_destroy);
1277 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1278 struct rbd_obj_request *obj_request)
1280 rbd_assert(obj_request->img_request == NULL);
1282 /* Image request now owns object's original reference */
1283 obj_request->img_request = img_request;
1284 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1287 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1288 struct rbd_obj_request *obj_request)
1290 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1291 list_del(&obj_request->ex.oe_item);
1292 rbd_assert(obj_request->img_request == img_request);
1293 rbd_obj_request_put(obj_request);
1296 static void rbd_osd_submit(struct ceph_osd_request *osd_req)
1298 struct rbd_obj_request *obj_req = osd_req->r_priv;
1300 dout("%s osd_req %p for obj_req %p objno %llu %llu~%llu\n",
1301 __func__, osd_req, obj_req, obj_req->ex.oe_objno,
1302 obj_req->ex.oe_off, obj_req->ex.oe_len);
1303 ceph_osdc_start_request(osd_req->r_osdc, osd_req);
1307 * The default/initial value for all image request flags is 0. Each
1308 * is conditionally set to 1 at image request initialization time
1309 * and currently never change thereafter.
1311 static void img_request_layered_set(struct rbd_img_request *img_request)
1313 set_bit(IMG_REQ_LAYERED, &img_request->flags);
1316 static bool img_request_layered_test(struct rbd_img_request *img_request)
1318 return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1321 static bool rbd_obj_is_entire(struct rbd_obj_request *obj_req)
1323 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1325 return !obj_req->ex.oe_off &&
1326 obj_req->ex.oe_len == rbd_dev->layout.object_size;
1329 static bool rbd_obj_is_tail(struct rbd_obj_request *obj_req)
1331 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1333 return obj_req->ex.oe_off + obj_req->ex.oe_len ==
1334 rbd_dev->layout.object_size;
1338 * Must be called after rbd_obj_calc_img_extents().
1340 static void rbd_obj_set_copyup_enabled(struct rbd_obj_request *obj_req)
1342 rbd_assert(obj_req->img_request->snapc);
1344 if (obj_req->img_request->op_type == OBJ_OP_DISCARD) {
1345 dout("%s %p objno %llu discard\n", __func__, obj_req,
1346 obj_req->ex.oe_objno);
1350 if (!obj_req->num_img_extents) {
1351 dout("%s %p objno %llu not overlapping\n", __func__, obj_req,
1352 obj_req->ex.oe_objno);
1356 if (rbd_obj_is_entire(obj_req) &&
1357 !obj_req->img_request->snapc->num_snaps) {
1358 dout("%s %p objno %llu entire\n", __func__, obj_req,
1359 obj_req->ex.oe_objno);
1363 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
1366 static u64 rbd_obj_img_extents_bytes(struct rbd_obj_request *obj_req)
1368 return ceph_file_extents_bytes(obj_req->img_extents,
1369 obj_req->num_img_extents);
1372 static bool rbd_img_is_write(struct rbd_img_request *img_req)
1374 switch (img_req->op_type) {
1378 case OBJ_OP_DISCARD:
1379 case OBJ_OP_ZEROOUT:
1386 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
1388 struct rbd_obj_request *obj_req = osd_req->r_priv;
1391 dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
1392 osd_req->r_result, obj_req);
1395 * Writes aren't allowed to return a data payload. In some
1396 * guarded write cases (e.g. stat + zero on an empty object)
1397 * a stat response makes it through, but we don't care.
1399 if (osd_req->r_result > 0 && rbd_img_is_write(obj_req->img_request))
1402 result = osd_req->r_result;
1404 rbd_obj_handle_request(obj_req, result);
1407 static void rbd_osd_format_read(struct ceph_osd_request *osd_req)
1409 struct rbd_obj_request *obj_request = osd_req->r_priv;
1410 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1411 struct ceph_options *opt = rbd_dev->rbd_client->client->options;
1413 osd_req->r_flags = CEPH_OSD_FLAG_READ | opt->read_from_replica;
1414 osd_req->r_snapid = obj_request->img_request->snap_id;
1417 static void rbd_osd_format_write(struct ceph_osd_request *osd_req)
1419 struct rbd_obj_request *obj_request = osd_req->r_priv;
1421 osd_req->r_flags = CEPH_OSD_FLAG_WRITE;
1422 ktime_get_real_ts64(&osd_req->r_mtime);
1423 osd_req->r_data_offset = obj_request->ex.oe_off;
1426 static struct ceph_osd_request *
1427 __rbd_obj_add_osd_request(struct rbd_obj_request *obj_req,
1428 struct ceph_snap_context *snapc, int num_ops)
1430 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1431 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1432 struct ceph_osd_request *req;
1433 const char *name_format = rbd_dev->image_format == 1 ?
1434 RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
1437 req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, GFP_NOIO);
1439 return ERR_PTR(-ENOMEM);
1441 list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
1442 req->r_callback = rbd_osd_req_callback;
1443 req->r_priv = obj_req;
1446 * Data objects may be stored in a separate pool, but always in
1447 * the same namespace in that pool as the header in its pool.
1449 ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
1450 req->r_base_oloc.pool = rbd_dev->layout.pool_id;
1452 ret = ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
1453 rbd_dev->header.object_prefix,
1454 obj_req->ex.oe_objno);
1456 return ERR_PTR(ret);
1461 static struct ceph_osd_request *
1462 rbd_obj_add_osd_request(struct rbd_obj_request *obj_req, int num_ops)
1464 rbd_assert(obj_req->img_request->snapc);
1465 return __rbd_obj_add_osd_request(obj_req, obj_req->img_request->snapc,
1469 static struct rbd_obj_request *rbd_obj_request_create(void)
1471 struct rbd_obj_request *obj_request;
1473 obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_NOIO);
1477 ceph_object_extent_init(&obj_request->ex);
1478 INIT_LIST_HEAD(&obj_request->osd_reqs);
1479 mutex_init(&obj_request->state_mutex);
1480 kref_init(&obj_request->kref);
1482 dout("%s %p\n", __func__, obj_request);
1486 static void rbd_obj_request_destroy(struct kref *kref)
1488 struct rbd_obj_request *obj_request;
1489 struct ceph_osd_request *osd_req;
1492 obj_request = container_of(kref, struct rbd_obj_request, kref);
1494 dout("%s: obj %p\n", __func__, obj_request);
1496 while (!list_empty(&obj_request->osd_reqs)) {
1497 osd_req = list_first_entry(&obj_request->osd_reqs,
1498 struct ceph_osd_request, r_private_item);
1499 list_del_init(&osd_req->r_private_item);
1500 ceph_osdc_put_request(osd_req);
1503 switch (obj_request->img_request->data_type) {
1504 case OBJ_REQUEST_NODATA:
1505 case OBJ_REQUEST_BIO:
1506 case OBJ_REQUEST_BVECS:
1507 break; /* Nothing to do */
1508 case OBJ_REQUEST_OWN_BVECS:
1509 kfree(obj_request->bvec_pos.bvecs);
1515 kfree(obj_request->img_extents);
1516 if (obj_request->copyup_bvecs) {
1517 for (i = 0; i < obj_request->copyup_bvec_count; i++) {
1518 if (obj_request->copyup_bvecs[i].bv_page)
1519 __free_page(obj_request->copyup_bvecs[i].bv_page);
1521 kfree(obj_request->copyup_bvecs);
1524 kmem_cache_free(rbd_obj_request_cache, obj_request);
1527 /* It's OK to call this for a device with no parent */
1529 static void rbd_spec_put(struct rbd_spec *spec);
1530 static void rbd_dev_unparent(struct rbd_device *rbd_dev)
1532 rbd_dev_remove_parent(rbd_dev);
1533 rbd_spec_put(rbd_dev->parent_spec);
1534 rbd_dev->parent_spec = NULL;
1535 rbd_dev->parent_overlap = 0;
1539 * Parent image reference counting is used to determine when an
1540 * image's parent fields can be safely torn down--after there are no
1541 * more in-flight requests to the parent image. When the last
1542 * reference is dropped, cleaning them up is safe.
1544 static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
1548 if (!rbd_dev->parent_spec)
1551 counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
1555 /* Last reference; clean up parent data structures */
1558 rbd_dev_unparent(rbd_dev);
1560 rbd_warn(rbd_dev, "parent reference underflow");
1564 * If an image has a non-zero parent overlap, get a reference to its
1567 * Returns true if the rbd device has a parent with a non-zero
1568 * overlap and a reference for it was successfully taken, or
1571 static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
1575 if (!rbd_dev->parent_spec)
1578 if (rbd_dev->parent_overlap)
1579 counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
1582 rbd_warn(rbd_dev, "parent reference overflow");
1587 static void rbd_img_request_init(struct rbd_img_request *img_request,
1588 struct rbd_device *rbd_dev,
1589 enum obj_operation_type op_type)
1591 memset(img_request, 0, sizeof(*img_request));
1593 img_request->rbd_dev = rbd_dev;
1594 img_request->op_type = op_type;
1596 INIT_LIST_HEAD(&img_request->lock_item);
1597 INIT_LIST_HEAD(&img_request->object_extents);
1598 mutex_init(&img_request->state_mutex);
1602 * Only snap_id is captured here, for reads. For writes, snapshot
1603 * context is captured in rbd_img_object_requests() after exclusive
1604 * lock is ensured to be held.
1606 static void rbd_img_capture_header(struct rbd_img_request *img_req)
1608 struct rbd_device *rbd_dev = img_req->rbd_dev;
1610 lockdep_assert_held(&rbd_dev->header_rwsem);
1612 if (!rbd_img_is_write(img_req))
1613 img_req->snap_id = rbd_dev->spec->snap_id;
1615 if (rbd_dev_parent_get(rbd_dev))
1616 img_request_layered_set(img_req);
1619 static void rbd_img_request_destroy(struct rbd_img_request *img_request)
1621 struct rbd_obj_request *obj_request;
1622 struct rbd_obj_request *next_obj_request;
1624 dout("%s: img %p\n", __func__, img_request);
1626 WARN_ON(!list_empty(&img_request->lock_item));
1627 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1628 rbd_img_obj_request_del(img_request, obj_request);
1630 if (img_request_layered_test(img_request))
1631 rbd_dev_parent_put(img_request->rbd_dev);
1633 if (rbd_img_is_write(img_request))
1634 ceph_put_snap_context(img_request->snapc);
1636 if (test_bit(IMG_REQ_CHILD, &img_request->flags))
1637 kmem_cache_free(rbd_img_request_cache, img_request);
1640 #define BITS_PER_OBJ 2
1641 #define OBJS_PER_BYTE (BITS_PER_BYTE / BITS_PER_OBJ)
1642 #define OBJ_MASK ((1 << BITS_PER_OBJ) - 1)
1644 static void __rbd_object_map_index(struct rbd_device *rbd_dev, u64 objno,
1645 u64 *index, u8 *shift)
1649 rbd_assert(objno < rbd_dev->object_map_size);
1650 *index = div_u64_rem(objno, OBJS_PER_BYTE, &off);
1651 *shift = (OBJS_PER_BYTE - off - 1) * BITS_PER_OBJ;
1654 static u8 __rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1659 lockdep_assert_held(&rbd_dev->object_map_lock);
1660 __rbd_object_map_index(rbd_dev, objno, &index, &shift);
1661 return (rbd_dev->object_map[index] >> shift) & OBJ_MASK;
1664 static void __rbd_object_map_set(struct rbd_device *rbd_dev, u64 objno, u8 val)
1670 lockdep_assert_held(&rbd_dev->object_map_lock);
1671 rbd_assert(!(val & ~OBJ_MASK));
1673 __rbd_object_map_index(rbd_dev, objno, &index, &shift);
1674 p = &rbd_dev->object_map[index];
1675 *p = (*p & ~(OBJ_MASK << shift)) | (val << shift);
1678 static u8 rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1682 spin_lock(&rbd_dev->object_map_lock);
1683 state = __rbd_object_map_get(rbd_dev, objno);
1684 spin_unlock(&rbd_dev->object_map_lock);
1688 static bool use_object_map(struct rbd_device *rbd_dev)
1691 * An image mapped read-only can't use the object map -- it isn't
1692 * loaded because the header lock isn't acquired. Someone else can
1693 * write to the image and update the object map behind our back.
1695 * A snapshot can't be written to, so using the object map is always
1698 if (!rbd_is_snap(rbd_dev) && rbd_is_ro(rbd_dev))
1701 return ((rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) &&
1702 !(rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID));
1705 static bool rbd_object_map_may_exist(struct rbd_device *rbd_dev, u64 objno)
1709 /* fall back to default logic if object map is disabled or invalid */
1710 if (!use_object_map(rbd_dev))
1713 state = rbd_object_map_get(rbd_dev, objno);
1714 return state != OBJECT_NONEXISTENT;
1717 static void rbd_object_map_name(struct rbd_device *rbd_dev, u64 snap_id,
1718 struct ceph_object_id *oid)
1720 if (snap_id == CEPH_NOSNAP)
1721 ceph_oid_printf(oid, "%s%s", RBD_OBJECT_MAP_PREFIX,
1722 rbd_dev->spec->image_id);
1724 ceph_oid_printf(oid, "%s%s.%016llx", RBD_OBJECT_MAP_PREFIX,
1725 rbd_dev->spec->image_id, snap_id);
1728 static int rbd_object_map_lock(struct rbd_device *rbd_dev)
1730 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1731 CEPH_DEFINE_OID_ONSTACK(oid);
1734 struct ceph_locker *lockers;
1736 bool broke_lock = false;
1739 rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1742 ret = ceph_cls_lock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1743 CEPH_CLS_LOCK_EXCLUSIVE, "", "", "", 0);
1744 if (ret != -EBUSY || broke_lock) {
1746 ret = 0; /* already locked by myself */
1748 rbd_warn(rbd_dev, "failed to lock object map: %d", ret);
1752 ret = ceph_cls_lock_info(osdc, &oid, &rbd_dev->header_oloc,
1753 RBD_LOCK_NAME, &lock_type, &lock_tag,
1754 &lockers, &num_lockers);
1759 rbd_warn(rbd_dev, "failed to get object map lockers: %d", ret);
1764 if (num_lockers == 0)
1767 rbd_warn(rbd_dev, "breaking object map lock owned by %s%llu",
1768 ENTITY_NAME(lockers[0].id.name));
1770 ret = ceph_cls_break_lock(osdc, &oid, &rbd_dev->header_oloc,
1771 RBD_LOCK_NAME, lockers[0].id.cookie,
1772 &lockers[0].id.name);
1773 ceph_free_lockers(lockers, num_lockers);
1778 rbd_warn(rbd_dev, "failed to break object map lock: %d", ret);
1786 static void rbd_object_map_unlock(struct rbd_device *rbd_dev)
1788 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1789 CEPH_DEFINE_OID_ONSTACK(oid);
1792 rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1794 ret = ceph_cls_unlock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1796 if (ret && ret != -ENOENT)
1797 rbd_warn(rbd_dev, "failed to unlock object map: %d", ret);
1800 static int decode_object_map_header(void **p, void *end, u64 *object_map_size)
1808 ceph_decode_32_safe(p, end, header_len, e_inval);
1809 header_end = *p + header_len;
1811 ret = ceph_start_decoding(p, end, 1, "BitVector header", &struct_v,
1816 ceph_decode_64_safe(p, end, *object_map_size, e_inval);
1825 static int __rbd_object_map_load(struct rbd_device *rbd_dev)
1827 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1828 CEPH_DEFINE_OID_ONSTACK(oid);
1829 struct page **pages;
1833 u64 object_map_bytes;
1834 u64 object_map_size;
1838 rbd_assert(!rbd_dev->object_map && !rbd_dev->object_map_size);
1840 num_objects = ceph_get_num_objects(&rbd_dev->layout,
1841 rbd_dev->mapping.size);
1842 object_map_bytes = DIV_ROUND_UP_ULL(num_objects * BITS_PER_OBJ,
1844 num_pages = calc_pages_for(0, object_map_bytes) + 1;
1845 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1847 return PTR_ERR(pages);
1849 reply_len = num_pages * PAGE_SIZE;
1850 rbd_object_map_name(rbd_dev, rbd_dev->spec->snap_id, &oid);
1851 ret = ceph_osdc_call(osdc, &oid, &rbd_dev->header_oloc,
1852 "rbd", "object_map_load", CEPH_OSD_FLAG_READ,
1853 NULL, 0, pages, &reply_len);
1857 p = page_address(pages[0]);
1858 end = p + min(reply_len, (size_t)PAGE_SIZE);
1859 ret = decode_object_map_header(&p, end, &object_map_size);
1863 if (object_map_size != num_objects) {
1864 rbd_warn(rbd_dev, "object map size mismatch: %llu vs %llu",
1865 object_map_size, num_objects);
1870 if (offset_in_page(p) + object_map_bytes > reply_len) {
1875 rbd_dev->object_map = kvmalloc(object_map_bytes, GFP_KERNEL);
1876 if (!rbd_dev->object_map) {
1881 rbd_dev->object_map_size = object_map_size;
1882 ceph_copy_from_page_vector(pages, rbd_dev->object_map,
1883 offset_in_page(p), object_map_bytes);
1886 ceph_release_page_vector(pages, num_pages);
1890 static void rbd_object_map_free(struct rbd_device *rbd_dev)
1892 kvfree(rbd_dev->object_map);
1893 rbd_dev->object_map = NULL;
1894 rbd_dev->object_map_size = 0;
1897 static int rbd_object_map_load(struct rbd_device *rbd_dev)
1901 ret = __rbd_object_map_load(rbd_dev);
1905 ret = rbd_dev_v2_get_flags(rbd_dev);
1907 rbd_object_map_free(rbd_dev);
1911 if (rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID)
1912 rbd_warn(rbd_dev, "object map is invalid");
1917 static int rbd_object_map_open(struct rbd_device *rbd_dev)
1921 ret = rbd_object_map_lock(rbd_dev);
1925 ret = rbd_object_map_load(rbd_dev);
1927 rbd_object_map_unlock(rbd_dev);
1934 static void rbd_object_map_close(struct rbd_device *rbd_dev)
1936 rbd_object_map_free(rbd_dev);
1937 rbd_object_map_unlock(rbd_dev);
1941 * This function needs snap_id (or more precisely just something to
1942 * distinguish between HEAD and snapshot object maps), new_state and
1943 * current_state that were passed to rbd_object_map_update().
1945 * To avoid allocating and stashing a context we piggyback on the OSD
1946 * request. A HEAD update has two ops (assert_locked). For new_state
1947 * and current_state we decode our own object_map_update op, encoded in
1948 * rbd_cls_object_map_update().
1950 static int rbd_object_map_update_finish(struct rbd_obj_request *obj_req,
1951 struct ceph_osd_request *osd_req)
1953 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1954 struct ceph_osd_data *osd_data;
1956 u8 state, new_state, current_state;
1957 bool has_current_state;
1960 if (osd_req->r_result)
1961 return osd_req->r_result;
1964 * Nothing to do for a snapshot object map.
1966 if (osd_req->r_num_ops == 1)
1970 * Update in-memory HEAD object map.
1972 rbd_assert(osd_req->r_num_ops == 2);
1973 osd_data = osd_req_op_data(osd_req, 1, cls, request_data);
1974 rbd_assert(osd_data->type == CEPH_OSD_DATA_TYPE_PAGES);
1976 p = page_address(osd_data->pages[0]);
1977 objno = ceph_decode_64(&p);
1978 rbd_assert(objno == obj_req->ex.oe_objno);
1979 rbd_assert(ceph_decode_64(&p) == objno + 1);
1980 new_state = ceph_decode_8(&p);
1981 has_current_state = ceph_decode_8(&p);
1982 if (has_current_state)
1983 current_state = ceph_decode_8(&p);
1985 spin_lock(&rbd_dev->object_map_lock);
1986 state = __rbd_object_map_get(rbd_dev, objno);
1987 if (!has_current_state || current_state == state ||
1988 (current_state == OBJECT_EXISTS && state == OBJECT_EXISTS_CLEAN))
1989 __rbd_object_map_set(rbd_dev, objno, new_state);
1990 spin_unlock(&rbd_dev->object_map_lock);
1995 static void rbd_object_map_callback(struct ceph_osd_request *osd_req)
1997 struct rbd_obj_request *obj_req = osd_req->r_priv;
2000 dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
2001 osd_req->r_result, obj_req);
2003 result = rbd_object_map_update_finish(obj_req, osd_req);
2004 rbd_obj_handle_request(obj_req, result);
2007 static bool update_needed(struct rbd_device *rbd_dev, u64 objno, u8 new_state)
2009 u8 state = rbd_object_map_get(rbd_dev, objno);
2011 if (state == new_state ||
2012 (new_state == OBJECT_PENDING && state == OBJECT_NONEXISTENT) ||
2013 (new_state == OBJECT_NONEXISTENT && state != OBJECT_PENDING))
2019 static int rbd_cls_object_map_update(struct ceph_osd_request *req,
2020 int which, u64 objno, u8 new_state,
2021 const u8 *current_state)
2023 struct page **pages;
2027 ret = osd_req_op_cls_init(req, which, "rbd", "object_map_update");
2031 pages = ceph_alloc_page_vector(1, GFP_NOIO);
2033 return PTR_ERR(pages);
2035 p = start = page_address(pages[0]);
2036 ceph_encode_64(&p, objno);
2037 ceph_encode_64(&p, objno + 1);
2038 ceph_encode_8(&p, new_state);
2039 if (current_state) {
2040 ceph_encode_8(&p, 1);
2041 ceph_encode_8(&p, *current_state);
2043 ceph_encode_8(&p, 0);
2046 osd_req_op_cls_request_data_pages(req, which, pages, p - start, 0,
2053 * 0 - object map update sent
2054 * 1 - object map update isn't needed
2057 static int rbd_object_map_update(struct rbd_obj_request *obj_req, u64 snap_id,
2058 u8 new_state, const u8 *current_state)
2060 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2061 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2062 struct ceph_osd_request *req;
2067 if (snap_id == CEPH_NOSNAP) {
2068 if (!update_needed(rbd_dev, obj_req->ex.oe_objno, new_state))
2071 num_ops++; /* assert_locked */
2074 req = ceph_osdc_alloc_request(osdc, NULL, num_ops, false, GFP_NOIO);
2078 list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
2079 req->r_callback = rbd_object_map_callback;
2080 req->r_priv = obj_req;
2082 rbd_object_map_name(rbd_dev, snap_id, &req->r_base_oid);
2083 ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
2084 req->r_flags = CEPH_OSD_FLAG_WRITE;
2085 ktime_get_real_ts64(&req->r_mtime);
2087 if (snap_id == CEPH_NOSNAP) {
2089 * Protect against possible race conditions during lock
2090 * ownership transitions.
2092 ret = ceph_cls_assert_locked(req, which++, RBD_LOCK_NAME,
2093 CEPH_CLS_LOCK_EXCLUSIVE, "", "");
2098 ret = rbd_cls_object_map_update(req, which, obj_req->ex.oe_objno,
2099 new_state, current_state);
2103 ret = ceph_osdc_alloc_messages(req, GFP_NOIO);
2107 ceph_osdc_start_request(osdc, req);
2111 static void prune_extents(struct ceph_file_extent *img_extents,
2112 u32 *num_img_extents, u64 overlap)
2114 u32 cnt = *num_img_extents;
2116 /* drop extents completely beyond the overlap */
2117 while (cnt && img_extents[cnt - 1].fe_off >= overlap)
2121 struct ceph_file_extent *ex = &img_extents[cnt - 1];
2123 /* trim final overlapping extent */
2124 if (ex->fe_off + ex->fe_len > overlap)
2125 ex->fe_len = overlap - ex->fe_off;
2128 *num_img_extents = cnt;
2132 * Determine the byte range(s) covered by either just the object extent
2133 * or the entire object in the parent image.
2135 static int rbd_obj_calc_img_extents(struct rbd_obj_request *obj_req,
2138 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2141 if (!rbd_dev->parent_overlap)
2144 ret = ceph_extent_to_file(&rbd_dev->layout, obj_req->ex.oe_objno,
2145 entire ? 0 : obj_req->ex.oe_off,
2146 entire ? rbd_dev->layout.object_size :
2148 &obj_req->img_extents,
2149 &obj_req->num_img_extents);
2153 prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
2154 rbd_dev->parent_overlap);
2158 static void rbd_osd_setup_data(struct ceph_osd_request *osd_req, int which)
2160 struct rbd_obj_request *obj_req = osd_req->r_priv;
2162 switch (obj_req->img_request->data_type) {
2163 case OBJ_REQUEST_BIO:
2164 osd_req_op_extent_osd_data_bio(osd_req, which,
2166 obj_req->ex.oe_len);
2168 case OBJ_REQUEST_BVECS:
2169 case OBJ_REQUEST_OWN_BVECS:
2170 rbd_assert(obj_req->bvec_pos.iter.bi_size ==
2171 obj_req->ex.oe_len);
2172 rbd_assert(obj_req->bvec_idx == obj_req->bvec_count);
2173 osd_req_op_extent_osd_data_bvec_pos(osd_req, which,
2174 &obj_req->bvec_pos);
2181 static int rbd_osd_setup_stat(struct ceph_osd_request *osd_req, int which)
2183 struct page **pages;
2186 * The response data for a STAT call consists of:
2193 pages = ceph_alloc_page_vector(1, GFP_NOIO);
2195 return PTR_ERR(pages);
2197 osd_req_op_init(osd_req, which, CEPH_OSD_OP_STAT, 0);
2198 osd_req_op_raw_data_in_pages(osd_req, which, pages,
2199 8 + sizeof(struct ceph_timespec),
2204 static int rbd_osd_setup_copyup(struct ceph_osd_request *osd_req, int which,
2207 struct rbd_obj_request *obj_req = osd_req->r_priv;
2210 ret = osd_req_op_cls_init(osd_req, which, "rbd", "copyup");
2214 osd_req_op_cls_request_data_bvecs(osd_req, which, obj_req->copyup_bvecs,
2215 obj_req->copyup_bvec_count, bytes);
2219 static int rbd_obj_init_read(struct rbd_obj_request *obj_req)
2221 obj_req->read_state = RBD_OBJ_READ_START;
2225 static void __rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2228 struct rbd_obj_request *obj_req = osd_req->r_priv;
2229 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2232 if (!use_object_map(rbd_dev) ||
2233 !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST)) {
2234 osd_req_op_alloc_hint_init(osd_req, which++,
2235 rbd_dev->layout.object_size,
2236 rbd_dev->layout.object_size,
2237 rbd_dev->opts->alloc_hint_flags);
2240 if (rbd_obj_is_entire(obj_req))
2241 opcode = CEPH_OSD_OP_WRITEFULL;
2243 opcode = CEPH_OSD_OP_WRITE;
2245 osd_req_op_extent_init(osd_req, which, opcode,
2246 obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2247 rbd_osd_setup_data(osd_req, which);
2250 static int rbd_obj_init_write(struct rbd_obj_request *obj_req)
2254 /* reverse map the entire object onto the parent */
2255 ret = rbd_obj_calc_img_extents(obj_req, true);
2259 obj_req->write_state = RBD_OBJ_WRITE_START;
2263 static u16 truncate_or_zero_opcode(struct rbd_obj_request *obj_req)
2265 return rbd_obj_is_tail(obj_req) ? CEPH_OSD_OP_TRUNCATE :
2269 static void __rbd_osd_setup_discard_ops(struct ceph_osd_request *osd_req,
2272 struct rbd_obj_request *obj_req = osd_req->r_priv;
2274 if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
2275 rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2276 osd_req_op_init(osd_req, which, CEPH_OSD_OP_DELETE, 0);
2278 osd_req_op_extent_init(osd_req, which,
2279 truncate_or_zero_opcode(obj_req),
2280 obj_req->ex.oe_off, obj_req->ex.oe_len,
2285 static int rbd_obj_init_discard(struct rbd_obj_request *obj_req)
2287 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2292 * Align the range to alloc_size boundary and punt on discards
2293 * that are too small to free up any space.
2295 * alloc_size == object_size && is_tail() is a special case for
2296 * filestore with filestore_punch_hole = false, needed to allow
2297 * truncate (in addition to delete).
2299 if (rbd_dev->opts->alloc_size != rbd_dev->layout.object_size ||
2300 !rbd_obj_is_tail(obj_req)) {
2301 off = round_up(obj_req->ex.oe_off, rbd_dev->opts->alloc_size);
2302 next_off = round_down(obj_req->ex.oe_off + obj_req->ex.oe_len,
2303 rbd_dev->opts->alloc_size);
2304 if (off >= next_off)
2307 dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
2308 obj_req, obj_req->ex.oe_off, obj_req->ex.oe_len,
2309 off, next_off - off);
2310 obj_req->ex.oe_off = off;
2311 obj_req->ex.oe_len = next_off - off;
2314 /* reverse map the entire object onto the parent */
2315 ret = rbd_obj_calc_img_extents(obj_req, true);
2319 obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2320 if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents)
2321 obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2323 obj_req->write_state = RBD_OBJ_WRITE_START;
2327 static void __rbd_osd_setup_zeroout_ops(struct ceph_osd_request *osd_req,
2330 struct rbd_obj_request *obj_req = osd_req->r_priv;
2333 if (rbd_obj_is_entire(obj_req)) {
2334 if (obj_req->num_img_extents) {
2335 if (!(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2336 osd_req_op_init(osd_req, which++,
2337 CEPH_OSD_OP_CREATE, 0);
2338 opcode = CEPH_OSD_OP_TRUNCATE;
2340 rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2341 osd_req_op_init(osd_req, which++,
2342 CEPH_OSD_OP_DELETE, 0);
2346 opcode = truncate_or_zero_opcode(obj_req);
2350 osd_req_op_extent_init(osd_req, which, opcode,
2351 obj_req->ex.oe_off, obj_req->ex.oe_len,
2355 static int rbd_obj_init_zeroout(struct rbd_obj_request *obj_req)
2359 /* reverse map the entire object onto the parent */
2360 ret = rbd_obj_calc_img_extents(obj_req, true);
2364 if (!obj_req->num_img_extents) {
2365 obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2366 if (rbd_obj_is_entire(obj_req))
2367 obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2370 obj_req->write_state = RBD_OBJ_WRITE_START;
2374 static int count_write_ops(struct rbd_obj_request *obj_req)
2376 struct rbd_img_request *img_req = obj_req->img_request;
2378 switch (img_req->op_type) {
2380 if (!use_object_map(img_req->rbd_dev) ||
2381 !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST))
2382 return 2; /* setallochint + write/writefull */
2384 return 1; /* write/writefull */
2385 case OBJ_OP_DISCARD:
2386 return 1; /* delete/truncate/zero */
2387 case OBJ_OP_ZEROOUT:
2388 if (rbd_obj_is_entire(obj_req) && obj_req->num_img_extents &&
2389 !(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2390 return 2; /* create + truncate */
2392 return 1; /* delete/truncate/zero */
2398 static void rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2401 struct rbd_obj_request *obj_req = osd_req->r_priv;
2403 switch (obj_req->img_request->op_type) {
2405 __rbd_osd_setup_write_ops(osd_req, which);
2407 case OBJ_OP_DISCARD:
2408 __rbd_osd_setup_discard_ops(osd_req, which);
2410 case OBJ_OP_ZEROOUT:
2411 __rbd_osd_setup_zeroout_ops(osd_req, which);
2419 * Prune the list of object requests (adjust offset and/or length, drop
2420 * redundant requests). Prepare object request state machines and image
2421 * request state machine for execution.
2423 static int __rbd_img_fill_request(struct rbd_img_request *img_req)
2425 struct rbd_obj_request *obj_req, *next_obj_req;
2428 for_each_obj_request_safe(img_req, obj_req, next_obj_req) {
2429 switch (img_req->op_type) {
2431 ret = rbd_obj_init_read(obj_req);
2434 ret = rbd_obj_init_write(obj_req);
2436 case OBJ_OP_DISCARD:
2437 ret = rbd_obj_init_discard(obj_req);
2439 case OBJ_OP_ZEROOUT:
2440 ret = rbd_obj_init_zeroout(obj_req);
2448 rbd_img_obj_request_del(img_req, obj_req);
2453 img_req->state = RBD_IMG_START;
2457 union rbd_img_fill_iter {
2458 struct ceph_bio_iter bio_iter;
2459 struct ceph_bvec_iter bvec_iter;
2462 struct rbd_img_fill_ctx {
2463 enum obj_request_type pos_type;
2464 union rbd_img_fill_iter *pos;
2465 union rbd_img_fill_iter iter;
2466 ceph_object_extent_fn_t set_pos_fn;
2467 ceph_object_extent_fn_t count_fn;
2468 ceph_object_extent_fn_t copy_fn;
2471 static struct ceph_object_extent *alloc_object_extent(void *arg)
2473 struct rbd_img_request *img_req = arg;
2474 struct rbd_obj_request *obj_req;
2476 obj_req = rbd_obj_request_create();
2480 rbd_img_obj_request_add(img_req, obj_req);
2481 return &obj_req->ex;
2485 * While su != os && sc == 1 is technically not fancy (it's the same
2486 * layout as su == os && sc == 1), we can't use the nocopy path for it
2487 * because ->set_pos_fn() should be called only once per object.
2488 * ceph_file_to_extents() invokes action_fn once per stripe unit, so
2489 * treat su != os && sc == 1 as fancy.
2491 static bool rbd_layout_is_fancy(struct ceph_file_layout *l)
2493 return l->stripe_unit != l->object_size;
2496 static int rbd_img_fill_request_nocopy(struct rbd_img_request *img_req,
2497 struct ceph_file_extent *img_extents,
2498 u32 num_img_extents,
2499 struct rbd_img_fill_ctx *fctx)
2504 img_req->data_type = fctx->pos_type;
2507 * Create object requests and set each object request's starting
2508 * position in the provided bio (list) or bio_vec array.
2510 fctx->iter = *fctx->pos;
2511 for (i = 0; i < num_img_extents; i++) {
2512 ret = ceph_file_to_extents(&img_req->rbd_dev->layout,
2513 img_extents[i].fe_off,
2514 img_extents[i].fe_len,
2515 &img_req->object_extents,
2516 alloc_object_extent, img_req,
2517 fctx->set_pos_fn, &fctx->iter);
2522 return __rbd_img_fill_request(img_req);
2526 * Map a list of image extents to a list of object extents, create the
2527 * corresponding object requests (normally each to a different object,
2528 * but not always) and add them to @img_req. For each object request,
2529 * set up its data descriptor to point to the corresponding chunk(s) of
2530 * @fctx->pos data buffer.
2532 * Because ceph_file_to_extents() will merge adjacent object extents
2533 * together, each object request's data descriptor may point to multiple
2534 * different chunks of @fctx->pos data buffer.
2536 * @fctx->pos data buffer is assumed to be large enough.
2538 static int rbd_img_fill_request(struct rbd_img_request *img_req,
2539 struct ceph_file_extent *img_extents,
2540 u32 num_img_extents,
2541 struct rbd_img_fill_ctx *fctx)
2543 struct rbd_device *rbd_dev = img_req->rbd_dev;
2544 struct rbd_obj_request *obj_req;
2548 if (fctx->pos_type == OBJ_REQUEST_NODATA ||
2549 !rbd_layout_is_fancy(&rbd_dev->layout))
2550 return rbd_img_fill_request_nocopy(img_req, img_extents,
2551 num_img_extents, fctx);
2553 img_req->data_type = OBJ_REQUEST_OWN_BVECS;
2556 * Create object requests and determine ->bvec_count for each object
2557 * request. Note that ->bvec_count sum over all object requests may
2558 * be greater than the number of bio_vecs in the provided bio (list)
2559 * or bio_vec array because when mapped, those bio_vecs can straddle
2560 * stripe unit boundaries.
2562 fctx->iter = *fctx->pos;
2563 for (i = 0; i < num_img_extents; i++) {
2564 ret = ceph_file_to_extents(&rbd_dev->layout,
2565 img_extents[i].fe_off,
2566 img_extents[i].fe_len,
2567 &img_req->object_extents,
2568 alloc_object_extent, img_req,
2569 fctx->count_fn, &fctx->iter);
2574 for_each_obj_request(img_req, obj_req) {
2575 obj_req->bvec_pos.bvecs = kmalloc_array(obj_req->bvec_count,
2576 sizeof(*obj_req->bvec_pos.bvecs),
2578 if (!obj_req->bvec_pos.bvecs)
2583 * Fill in each object request's private bio_vec array, splitting and
2584 * rearranging the provided bio_vecs in stripe unit chunks as needed.
2586 fctx->iter = *fctx->pos;
2587 for (i = 0; i < num_img_extents; i++) {
2588 ret = ceph_iterate_extents(&rbd_dev->layout,
2589 img_extents[i].fe_off,
2590 img_extents[i].fe_len,
2591 &img_req->object_extents,
2592 fctx->copy_fn, &fctx->iter);
2597 return __rbd_img_fill_request(img_req);
2600 static int rbd_img_fill_nodata(struct rbd_img_request *img_req,
2603 struct ceph_file_extent ex = { off, len };
2604 union rbd_img_fill_iter dummy = {};
2605 struct rbd_img_fill_ctx fctx = {
2606 .pos_type = OBJ_REQUEST_NODATA,
2610 return rbd_img_fill_request(img_req, &ex, 1, &fctx);
2613 static void set_bio_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2615 struct rbd_obj_request *obj_req =
2616 container_of(ex, struct rbd_obj_request, ex);
2617 struct ceph_bio_iter *it = arg;
2619 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2620 obj_req->bio_pos = *it;
2621 ceph_bio_iter_advance(it, bytes);
2624 static void count_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2626 struct rbd_obj_request *obj_req =
2627 container_of(ex, struct rbd_obj_request, ex);
2628 struct ceph_bio_iter *it = arg;
2630 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2631 ceph_bio_iter_advance_step(it, bytes, ({
2632 obj_req->bvec_count++;
2637 static void copy_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2639 struct rbd_obj_request *obj_req =
2640 container_of(ex, struct rbd_obj_request, ex);
2641 struct ceph_bio_iter *it = arg;
2643 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2644 ceph_bio_iter_advance_step(it, bytes, ({
2645 obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2646 obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2650 static int __rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2651 struct ceph_file_extent *img_extents,
2652 u32 num_img_extents,
2653 struct ceph_bio_iter *bio_pos)
2655 struct rbd_img_fill_ctx fctx = {
2656 .pos_type = OBJ_REQUEST_BIO,
2657 .pos = (union rbd_img_fill_iter *)bio_pos,
2658 .set_pos_fn = set_bio_pos,
2659 .count_fn = count_bio_bvecs,
2660 .copy_fn = copy_bio_bvecs,
2663 return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2667 static int rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2668 u64 off, u64 len, struct bio *bio)
2670 struct ceph_file_extent ex = { off, len };
2671 struct ceph_bio_iter it = { .bio = bio, .iter = bio->bi_iter };
2673 return __rbd_img_fill_from_bio(img_req, &ex, 1, &it);
2676 static void set_bvec_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2678 struct rbd_obj_request *obj_req =
2679 container_of(ex, struct rbd_obj_request, ex);
2680 struct ceph_bvec_iter *it = arg;
2682 obj_req->bvec_pos = *it;
2683 ceph_bvec_iter_shorten(&obj_req->bvec_pos, bytes);
2684 ceph_bvec_iter_advance(it, bytes);
2687 static void count_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2689 struct rbd_obj_request *obj_req =
2690 container_of(ex, struct rbd_obj_request, ex);
2691 struct ceph_bvec_iter *it = arg;
2693 ceph_bvec_iter_advance_step(it, bytes, ({
2694 obj_req->bvec_count++;
2698 static void copy_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2700 struct rbd_obj_request *obj_req =
2701 container_of(ex, struct rbd_obj_request, ex);
2702 struct ceph_bvec_iter *it = arg;
2704 ceph_bvec_iter_advance_step(it, bytes, ({
2705 obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2706 obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2710 static int __rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2711 struct ceph_file_extent *img_extents,
2712 u32 num_img_extents,
2713 struct ceph_bvec_iter *bvec_pos)
2715 struct rbd_img_fill_ctx fctx = {
2716 .pos_type = OBJ_REQUEST_BVECS,
2717 .pos = (union rbd_img_fill_iter *)bvec_pos,
2718 .set_pos_fn = set_bvec_pos,
2719 .count_fn = count_bvecs,
2720 .copy_fn = copy_bvecs,
2723 return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2727 static int rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2728 struct ceph_file_extent *img_extents,
2729 u32 num_img_extents,
2730 struct bio_vec *bvecs)
2732 struct ceph_bvec_iter it = {
2734 .iter = { .bi_size = ceph_file_extents_bytes(img_extents,
2738 return __rbd_img_fill_from_bvecs(img_req, img_extents, num_img_extents,
2742 static void rbd_img_handle_request_work(struct work_struct *work)
2744 struct rbd_img_request *img_req =
2745 container_of(work, struct rbd_img_request, work);
2747 rbd_img_handle_request(img_req, img_req->work_result);
2750 static void rbd_img_schedule(struct rbd_img_request *img_req, int result)
2752 INIT_WORK(&img_req->work, rbd_img_handle_request_work);
2753 img_req->work_result = result;
2754 queue_work(rbd_wq, &img_req->work);
2757 static bool rbd_obj_may_exist(struct rbd_obj_request *obj_req)
2759 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2761 if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno)) {
2762 obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2766 dout("%s %p objno %llu assuming dne\n", __func__, obj_req,
2767 obj_req->ex.oe_objno);
2771 static int rbd_obj_read_object(struct rbd_obj_request *obj_req)
2773 struct ceph_osd_request *osd_req;
2776 osd_req = __rbd_obj_add_osd_request(obj_req, NULL, 1);
2777 if (IS_ERR(osd_req))
2778 return PTR_ERR(osd_req);
2780 osd_req_op_extent_init(osd_req, 0, CEPH_OSD_OP_READ,
2781 obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2782 rbd_osd_setup_data(osd_req, 0);
2783 rbd_osd_format_read(osd_req);
2785 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2789 rbd_osd_submit(osd_req);
2793 static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
2795 struct rbd_img_request *img_req = obj_req->img_request;
2796 struct rbd_device *parent = img_req->rbd_dev->parent;
2797 struct rbd_img_request *child_img_req;
2800 child_img_req = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
2804 rbd_img_request_init(child_img_req, parent, OBJ_OP_READ);
2805 __set_bit(IMG_REQ_CHILD, &child_img_req->flags);
2806 child_img_req->obj_request = obj_req;
2808 down_read(&parent->header_rwsem);
2809 rbd_img_capture_header(child_img_req);
2810 up_read(&parent->header_rwsem);
2812 dout("%s child_img_req %p for obj_req %p\n", __func__, child_img_req,
2815 if (!rbd_img_is_write(img_req)) {
2816 switch (img_req->data_type) {
2817 case OBJ_REQUEST_BIO:
2818 ret = __rbd_img_fill_from_bio(child_img_req,
2819 obj_req->img_extents,
2820 obj_req->num_img_extents,
2823 case OBJ_REQUEST_BVECS:
2824 case OBJ_REQUEST_OWN_BVECS:
2825 ret = __rbd_img_fill_from_bvecs(child_img_req,
2826 obj_req->img_extents,
2827 obj_req->num_img_extents,
2828 &obj_req->bvec_pos);
2834 ret = rbd_img_fill_from_bvecs(child_img_req,
2835 obj_req->img_extents,
2836 obj_req->num_img_extents,
2837 obj_req->copyup_bvecs);
2840 rbd_img_request_destroy(child_img_req);
2844 /* avoid parent chain recursion */
2845 rbd_img_schedule(child_img_req, 0);
2849 static bool rbd_obj_advance_read(struct rbd_obj_request *obj_req, int *result)
2851 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2855 switch (obj_req->read_state) {
2856 case RBD_OBJ_READ_START:
2857 rbd_assert(!*result);
2859 if (!rbd_obj_may_exist(obj_req)) {
2861 obj_req->read_state = RBD_OBJ_READ_OBJECT;
2865 ret = rbd_obj_read_object(obj_req);
2870 obj_req->read_state = RBD_OBJ_READ_OBJECT;
2872 case RBD_OBJ_READ_OBJECT:
2873 if (*result == -ENOENT && rbd_dev->parent_overlap) {
2874 /* reverse map this object extent onto the parent */
2875 ret = rbd_obj_calc_img_extents(obj_req, false);
2880 if (obj_req->num_img_extents) {
2881 ret = rbd_obj_read_from_parent(obj_req);
2886 obj_req->read_state = RBD_OBJ_READ_PARENT;
2892 * -ENOENT means a hole in the image -- zero-fill the entire
2893 * length of the request. A short read also implies zero-fill
2894 * to the end of the request.
2896 if (*result == -ENOENT) {
2897 rbd_obj_zero_range(obj_req, 0, obj_req->ex.oe_len);
2899 } else if (*result >= 0) {
2900 if (*result < obj_req->ex.oe_len)
2901 rbd_obj_zero_range(obj_req, *result,
2902 obj_req->ex.oe_len - *result);
2904 rbd_assert(*result == obj_req->ex.oe_len);
2908 case RBD_OBJ_READ_PARENT:
2910 * The parent image is read only up to the overlap -- zero-fill
2911 * from the overlap to the end of the request.
2914 u32 obj_overlap = rbd_obj_img_extents_bytes(obj_req);
2916 if (obj_overlap < obj_req->ex.oe_len)
2917 rbd_obj_zero_range(obj_req, obj_overlap,
2918 obj_req->ex.oe_len - obj_overlap);
2926 static bool rbd_obj_write_is_noop(struct rbd_obj_request *obj_req)
2928 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2930 if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno))
2931 obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2933 if (!(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST) &&
2934 (obj_req->flags & RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT)) {
2935 dout("%s %p noop for nonexistent\n", __func__, obj_req);
2944 * 0 - object map update sent
2945 * 1 - object map update isn't needed
2948 static int rbd_obj_write_pre_object_map(struct rbd_obj_request *obj_req)
2950 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2953 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
2956 if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
2957 new_state = OBJECT_PENDING;
2959 new_state = OBJECT_EXISTS;
2961 return rbd_object_map_update(obj_req, CEPH_NOSNAP, new_state, NULL);
2964 static int rbd_obj_write_object(struct rbd_obj_request *obj_req)
2966 struct ceph_osd_request *osd_req;
2967 int num_ops = count_write_ops(obj_req);
2971 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
2972 num_ops++; /* stat */
2974 osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
2975 if (IS_ERR(osd_req))
2976 return PTR_ERR(osd_req);
2978 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
2979 ret = rbd_osd_setup_stat(osd_req, which++);
2984 rbd_osd_setup_write_ops(osd_req, which);
2985 rbd_osd_format_write(osd_req);
2987 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2991 rbd_osd_submit(osd_req);
2996 * copyup_bvecs pages are never highmem pages
2998 static bool is_zero_bvecs(struct bio_vec *bvecs, u32 bytes)
3000 struct ceph_bvec_iter it = {
3002 .iter = { .bi_size = bytes },
3005 ceph_bvec_iter_advance_step(&it, bytes, ({
3006 if (memchr_inv(bvec_virt(&bv), 0, bv.bv_len))
3012 #define MODS_ONLY U32_MAX
3014 static int rbd_obj_copyup_empty_snapc(struct rbd_obj_request *obj_req,
3017 struct ceph_osd_request *osd_req;
3020 dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3021 rbd_assert(bytes > 0 && bytes != MODS_ONLY);
3023 osd_req = __rbd_obj_add_osd_request(obj_req, &rbd_empty_snapc, 1);
3024 if (IS_ERR(osd_req))
3025 return PTR_ERR(osd_req);
3027 ret = rbd_osd_setup_copyup(osd_req, 0, bytes);
3031 rbd_osd_format_write(osd_req);
3033 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3037 rbd_osd_submit(osd_req);
3041 static int rbd_obj_copyup_current_snapc(struct rbd_obj_request *obj_req,
3044 struct ceph_osd_request *osd_req;
3045 int num_ops = count_write_ops(obj_req);
3049 dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3051 if (bytes != MODS_ONLY)
3052 num_ops++; /* copyup */
3054 osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
3055 if (IS_ERR(osd_req))
3056 return PTR_ERR(osd_req);
3058 if (bytes != MODS_ONLY) {
3059 ret = rbd_osd_setup_copyup(osd_req, which++, bytes);
3064 rbd_osd_setup_write_ops(osd_req, which);
3065 rbd_osd_format_write(osd_req);
3067 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3071 rbd_osd_submit(osd_req);
3075 static int setup_copyup_bvecs(struct rbd_obj_request *obj_req, u64 obj_overlap)
3079 rbd_assert(!obj_req->copyup_bvecs);
3080 obj_req->copyup_bvec_count = calc_pages_for(0, obj_overlap);
3081 obj_req->copyup_bvecs = kcalloc(obj_req->copyup_bvec_count,
3082 sizeof(*obj_req->copyup_bvecs),
3084 if (!obj_req->copyup_bvecs)
3087 for (i = 0; i < obj_req->copyup_bvec_count; i++) {
3088 unsigned int len = min(obj_overlap, (u64)PAGE_SIZE);
3089 struct page *page = alloc_page(GFP_NOIO);
3094 bvec_set_page(&obj_req->copyup_bvecs[i], page, len, 0);
3098 rbd_assert(!obj_overlap);
3103 * The target object doesn't exist. Read the data for the entire
3104 * target object up to the overlap point (if any) from the parent,
3105 * so we can use it for a copyup.
3107 static int rbd_obj_copyup_read_parent(struct rbd_obj_request *obj_req)
3109 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3112 rbd_assert(obj_req->num_img_extents);
3113 prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
3114 rbd_dev->parent_overlap);
3115 if (!obj_req->num_img_extents) {
3117 * The overlap has become 0 (most likely because the
3118 * image has been flattened). Re-submit the original write
3119 * request -- pass MODS_ONLY since the copyup isn't needed
3122 return rbd_obj_copyup_current_snapc(obj_req, MODS_ONLY);
3125 ret = setup_copyup_bvecs(obj_req, rbd_obj_img_extents_bytes(obj_req));
3129 return rbd_obj_read_from_parent(obj_req);
3132 static void rbd_obj_copyup_object_maps(struct rbd_obj_request *obj_req)
3134 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3135 struct ceph_snap_context *snapc = obj_req->img_request->snapc;
3140 rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3142 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3145 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3148 for (i = 0; i < snapc->num_snaps; i++) {
3149 if ((rbd_dev->header.features & RBD_FEATURE_FAST_DIFF) &&
3150 i + 1 < snapc->num_snaps)
3151 new_state = OBJECT_EXISTS_CLEAN;
3153 new_state = OBJECT_EXISTS;
3155 ret = rbd_object_map_update(obj_req, snapc->snaps[i],
3158 obj_req->pending.result = ret;
3163 obj_req->pending.num_pending++;
3167 static void rbd_obj_copyup_write_object(struct rbd_obj_request *obj_req)
3169 u32 bytes = rbd_obj_img_extents_bytes(obj_req);
3172 rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3175 * Only send non-zero copyup data to save some I/O and network
3176 * bandwidth -- zero copyup data is equivalent to the object not
3179 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3182 if (obj_req->img_request->snapc->num_snaps && bytes > 0) {
3184 * Send a copyup request with an empty snapshot context to
3185 * deep-copyup the object through all existing snapshots.
3186 * A second request with the current snapshot context will be
3187 * sent for the actual modification.
3189 ret = rbd_obj_copyup_empty_snapc(obj_req, bytes);
3191 obj_req->pending.result = ret;
3195 obj_req->pending.num_pending++;
3199 ret = rbd_obj_copyup_current_snapc(obj_req, bytes);
3201 obj_req->pending.result = ret;
3205 obj_req->pending.num_pending++;
3208 static bool rbd_obj_advance_copyup(struct rbd_obj_request *obj_req, int *result)
3210 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3214 switch (obj_req->copyup_state) {
3215 case RBD_OBJ_COPYUP_START:
3216 rbd_assert(!*result);
3218 ret = rbd_obj_copyup_read_parent(obj_req);
3223 if (obj_req->num_img_extents)
3224 obj_req->copyup_state = RBD_OBJ_COPYUP_READ_PARENT;
3226 obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3228 case RBD_OBJ_COPYUP_READ_PARENT:
3232 if (is_zero_bvecs(obj_req->copyup_bvecs,
3233 rbd_obj_img_extents_bytes(obj_req))) {
3234 dout("%s %p detected zeros\n", __func__, obj_req);
3235 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ZEROS;
3238 rbd_obj_copyup_object_maps(obj_req);
3239 if (!obj_req->pending.num_pending) {
3240 *result = obj_req->pending.result;
3241 obj_req->copyup_state = RBD_OBJ_COPYUP_OBJECT_MAPS;
3244 obj_req->copyup_state = __RBD_OBJ_COPYUP_OBJECT_MAPS;
3246 case __RBD_OBJ_COPYUP_OBJECT_MAPS:
3247 if (!pending_result_dec(&obj_req->pending, result))
3250 case RBD_OBJ_COPYUP_OBJECT_MAPS:
3252 rbd_warn(rbd_dev, "snap object map update failed: %d",
3257 rbd_obj_copyup_write_object(obj_req);
3258 if (!obj_req->pending.num_pending) {
3259 *result = obj_req->pending.result;
3260 obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3263 obj_req->copyup_state = __RBD_OBJ_COPYUP_WRITE_OBJECT;
3265 case __RBD_OBJ_COPYUP_WRITE_OBJECT:
3266 if (!pending_result_dec(&obj_req->pending, result))
3269 case RBD_OBJ_COPYUP_WRITE_OBJECT:
3278 * 0 - object map update sent
3279 * 1 - object map update isn't needed
3282 static int rbd_obj_write_post_object_map(struct rbd_obj_request *obj_req)
3284 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3285 u8 current_state = OBJECT_PENDING;
3287 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3290 if (!(obj_req->flags & RBD_OBJ_FLAG_DELETION))
3293 return rbd_object_map_update(obj_req, CEPH_NOSNAP, OBJECT_NONEXISTENT,
3297 static bool rbd_obj_advance_write(struct rbd_obj_request *obj_req, int *result)
3299 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3303 switch (obj_req->write_state) {
3304 case RBD_OBJ_WRITE_START:
3305 rbd_assert(!*result);
3307 rbd_obj_set_copyup_enabled(obj_req);
3308 if (rbd_obj_write_is_noop(obj_req))
3311 ret = rbd_obj_write_pre_object_map(obj_req);
3316 obj_req->write_state = RBD_OBJ_WRITE_PRE_OBJECT_MAP;
3320 case RBD_OBJ_WRITE_PRE_OBJECT_MAP:
3322 rbd_warn(rbd_dev, "pre object map update failed: %d",
3326 ret = rbd_obj_write_object(obj_req);
3331 obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
3333 case RBD_OBJ_WRITE_OBJECT:
3334 if (*result == -ENOENT) {
3335 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
3337 obj_req->copyup_state = RBD_OBJ_COPYUP_START;
3338 obj_req->write_state = __RBD_OBJ_WRITE_COPYUP;
3342 * On a non-existent object:
3343 * delete - -ENOENT, truncate/zero - 0
3345 if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
3351 obj_req->write_state = RBD_OBJ_WRITE_COPYUP;
3353 case __RBD_OBJ_WRITE_COPYUP:
3354 if (!rbd_obj_advance_copyup(obj_req, result))
3357 case RBD_OBJ_WRITE_COPYUP:
3359 rbd_warn(rbd_dev, "copyup failed: %d", *result);
3362 ret = rbd_obj_write_post_object_map(obj_req);
3367 obj_req->write_state = RBD_OBJ_WRITE_POST_OBJECT_MAP;
3371 case RBD_OBJ_WRITE_POST_OBJECT_MAP:
3373 rbd_warn(rbd_dev, "post object map update failed: %d",
3382 * Return true if @obj_req is completed.
3384 static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
3387 struct rbd_img_request *img_req = obj_req->img_request;
3388 struct rbd_device *rbd_dev = img_req->rbd_dev;
3391 mutex_lock(&obj_req->state_mutex);
3392 if (!rbd_img_is_write(img_req))
3393 done = rbd_obj_advance_read(obj_req, result);
3395 done = rbd_obj_advance_write(obj_req, result);
3396 mutex_unlock(&obj_req->state_mutex);
3398 if (done && *result) {
3399 rbd_assert(*result < 0);
3400 rbd_warn(rbd_dev, "%s at objno %llu %llu~%llu result %d",
3401 obj_op_name(img_req->op_type), obj_req->ex.oe_objno,
3402 obj_req->ex.oe_off, obj_req->ex.oe_len, *result);
3408 * This is open-coded in rbd_img_handle_request() to avoid parent chain
3411 static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result)
3413 if (__rbd_obj_handle_request(obj_req, &result))
3414 rbd_img_handle_request(obj_req->img_request, result);
3417 static bool need_exclusive_lock(struct rbd_img_request *img_req)
3419 struct rbd_device *rbd_dev = img_req->rbd_dev;
3421 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK))
3424 if (rbd_is_ro(rbd_dev))
3427 rbd_assert(!test_bit(IMG_REQ_CHILD, &img_req->flags));
3428 if (rbd_dev->opts->lock_on_read ||
3429 (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3432 return rbd_img_is_write(img_req);
3435 static bool rbd_lock_add_request(struct rbd_img_request *img_req)
3437 struct rbd_device *rbd_dev = img_req->rbd_dev;
3440 lockdep_assert_held(&rbd_dev->lock_rwsem);
3441 locked = rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED;
3442 spin_lock(&rbd_dev->lock_lists_lock);
3443 rbd_assert(list_empty(&img_req->lock_item));
3445 list_add_tail(&img_req->lock_item, &rbd_dev->acquiring_list);
3447 list_add_tail(&img_req->lock_item, &rbd_dev->running_list);
3448 spin_unlock(&rbd_dev->lock_lists_lock);
3452 static void rbd_lock_del_request(struct rbd_img_request *img_req)
3454 struct rbd_device *rbd_dev = img_req->rbd_dev;
3457 lockdep_assert_held(&rbd_dev->lock_rwsem);
3458 spin_lock(&rbd_dev->lock_lists_lock);
3459 rbd_assert(!list_empty(&img_req->lock_item));
3460 list_del_init(&img_req->lock_item);
3461 need_wakeup = (rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING &&
3462 list_empty(&rbd_dev->running_list));
3463 spin_unlock(&rbd_dev->lock_lists_lock);
3465 complete(&rbd_dev->releasing_wait);
3468 static int rbd_img_exclusive_lock(struct rbd_img_request *img_req)
3470 struct rbd_device *rbd_dev = img_req->rbd_dev;
3472 if (!need_exclusive_lock(img_req))
3475 if (rbd_lock_add_request(img_req))
3478 if (rbd_dev->opts->exclusive) {
3479 WARN_ON(1); /* lock got released? */
3484 * Note the use of mod_delayed_work() in rbd_acquire_lock()
3485 * and cancel_delayed_work() in wake_lock_waiters().
3487 dout("%s rbd_dev %p queueing lock_dwork\n", __func__, rbd_dev);
3488 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
3492 static void rbd_img_object_requests(struct rbd_img_request *img_req)
3494 struct rbd_device *rbd_dev = img_req->rbd_dev;
3495 struct rbd_obj_request *obj_req;
3497 rbd_assert(!img_req->pending.result && !img_req->pending.num_pending);
3498 rbd_assert(!need_exclusive_lock(img_req) ||
3499 __rbd_is_lock_owner(rbd_dev));
3501 if (rbd_img_is_write(img_req)) {
3502 rbd_assert(!img_req->snapc);
3503 down_read(&rbd_dev->header_rwsem);
3504 img_req->snapc = ceph_get_snap_context(rbd_dev->header.snapc);
3505 up_read(&rbd_dev->header_rwsem);
3508 for_each_obj_request(img_req, obj_req) {
3511 if (__rbd_obj_handle_request(obj_req, &result)) {
3513 img_req->pending.result = result;
3517 img_req->pending.num_pending++;
3522 static bool rbd_img_advance(struct rbd_img_request *img_req, int *result)
3527 switch (img_req->state) {
3529 rbd_assert(!*result);
3531 ret = rbd_img_exclusive_lock(img_req);
3536 img_req->state = RBD_IMG_EXCLUSIVE_LOCK;
3540 case RBD_IMG_EXCLUSIVE_LOCK:
3544 rbd_img_object_requests(img_req);
3545 if (!img_req->pending.num_pending) {
3546 *result = img_req->pending.result;
3547 img_req->state = RBD_IMG_OBJECT_REQUESTS;
3550 img_req->state = __RBD_IMG_OBJECT_REQUESTS;
3552 case __RBD_IMG_OBJECT_REQUESTS:
3553 if (!pending_result_dec(&img_req->pending, result))
3556 case RBD_IMG_OBJECT_REQUESTS:
3564 * Return true if @img_req is completed.
3566 static bool __rbd_img_handle_request(struct rbd_img_request *img_req,
3569 struct rbd_device *rbd_dev = img_req->rbd_dev;
3572 if (need_exclusive_lock(img_req)) {
3573 down_read(&rbd_dev->lock_rwsem);
3574 mutex_lock(&img_req->state_mutex);
3575 done = rbd_img_advance(img_req, result);
3577 rbd_lock_del_request(img_req);
3578 mutex_unlock(&img_req->state_mutex);
3579 up_read(&rbd_dev->lock_rwsem);
3581 mutex_lock(&img_req->state_mutex);
3582 done = rbd_img_advance(img_req, result);
3583 mutex_unlock(&img_req->state_mutex);
3586 if (done && *result) {
3587 rbd_assert(*result < 0);
3588 rbd_warn(rbd_dev, "%s%s result %d",
3589 test_bit(IMG_REQ_CHILD, &img_req->flags) ? "child " : "",
3590 obj_op_name(img_req->op_type), *result);
3595 static void rbd_img_handle_request(struct rbd_img_request *img_req, int result)
3598 if (!__rbd_img_handle_request(img_req, &result))
3601 if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
3602 struct rbd_obj_request *obj_req = img_req->obj_request;
3604 rbd_img_request_destroy(img_req);
3605 if (__rbd_obj_handle_request(obj_req, &result)) {
3606 img_req = obj_req->img_request;
3610 struct request *rq = blk_mq_rq_from_pdu(img_req);
3612 rbd_img_request_destroy(img_req);
3613 blk_mq_end_request(rq, errno_to_blk_status(result));
3617 static const struct rbd_client_id rbd_empty_cid;
3619 static bool rbd_cid_equal(const struct rbd_client_id *lhs,
3620 const struct rbd_client_id *rhs)
3622 return lhs->gid == rhs->gid && lhs->handle == rhs->handle;
3625 static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev)
3627 struct rbd_client_id cid;
3629 mutex_lock(&rbd_dev->watch_mutex);
3630 cid.gid = ceph_client_gid(rbd_dev->rbd_client->client);
3631 cid.handle = rbd_dev->watch_cookie;
3632 mutex_unlock(&rbd_dev->watch_mutex);
3637 * lock_rwsem must be held for write
3639 static void rbd_set_owner_cid(struct rbd_device *rbd_dev,
3640 const struct rbd_client_id *cid)
3642 dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev,
3643 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle,
3644 cid->gid, cid->handle);
3645 rbd_dev->owner_cid = *cid; /* struct */
3648 static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf)
3650 mutex_lock(&rbd_dev->watch_mutex);
3651 sprintf(buf, "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie);
3652 mutex_unlock(&rbd_dev->watch_mutex);
3655 static void __rbd_lock(struct rbd_device *rbd_dev, const char *cookie)
3657 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3659 rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED;
3660 strcpy(rbd_dev->lock_cookie, cookie);
3661 rbd_set_owner_cid(rbd_dev, &cid);
3662 queue_work(rbd_dev->task_wq, &rbd_dev->acquired_lock_work);
3666 * lock_rwsem must be held for write
3668 static int rbd_lock(struct rbd_device *rbd_dev)
3670 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3674 WARN_ON(__rbd_is_lock_owner(rbd_dev) ||
3675 rbd_dev->lock_cookie[0] != '\0');
3677 format_lock_cookie(rbd_dev, cookie);
3678 ret = ceph_cls_lock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3679 RBD_LOCK_NAME, CEPH_CLS_LOCK_EXCLUSIVE, cookie,
3680 RBD_LOCK_TAG, "", 0);
3681 if (ret && ret != -EEXIST)
3684 __rbd_lock(rbd_dev, cookie);
3689 * lock_rwsem must be held for write
3691 static void rbd_unlock(struct rbd_device *rbd_dev)
3693 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3696 WARN_ON(!__rbd_is_lock_owner(rbd_dev) ||
3697 rbd_dev->lock_cookie[0] == '\0');
3699 ret = ceph_cls_unlock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3700 RBD_LOCK_NAME, rbd_dev->lock_cookie);
3701 if (ret && ret != -ENOENT)
3702 rbd_warn(rbd_dev, "failed to unlock header: %d", ret);
3704 /* treat errors as the image is unlocked */
3705 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
3706 rbd_dev->lock_cookie[0] = '\0';
3707 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3708 queue_work(rbd_dev->task_wq, &rbd_dev->released_lock_work);
3711 static int __rbd_notify_op_lock(struct rbd_device *rbd_dev,
3712 enum rbd_notify_op notify_op,
3713 struct page ***preply_pages,
3716 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3717 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3718 char buf[4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN];
3719 int buf_size = sizeof(buf);
3722 dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op);
3724 /* encode *LockPayload NotifyMessage (op + ClientId) */
3725 ceph_start_encoding(&p, 2, 1, buf_size - CEPH_ENCODING_START_BLK_LEN);
3726 ceph_encode_32(&p, notify_op);
3727 ceph_encode_64(&p, cid.gid);
3728 ceph_encode_64(&p, cid.handle);
3730 return ceph_osdc_notify(osdc, &rbd_dev->header_oid,
3731 &rbd_dev->header_oloc, buf, buf_size,
3732 RBD_NOTIFY_TIMEOUT, preply_pages, preply_len);
3735 static void rbd_notify_op_lock(struct rbd_device *rbd_dev,
3736 enum rbd_notify_op notify_op)
3738 __rbd_notify_op_lock(rbd_dev, notify_op, NULL, NULL);
3741 static void rbd_notify_acquired_lock(struct work_struct *work)
3743 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3744 acquired_lock_work);
3746 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_ACQUIRED_LOCK);
3749 static void rbd_notify_released_lock(struct work_struct *work)
3751 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3752 released_lock_work);
3754 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_RELEASED_LOCK);
3757 static int rbd_request_lock(struct rbd_device *rbd_dev)
3759 struct page **reply_pages;
3761 bool lock_owner_responded = false;
3764 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3766 ret = __rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_REQUEST_LOCK,
3767 &reply_pages, &reply_len);
3768 if (ret && ret != -ETIMEDOUT) {
3769 rbd_warn(rbd_dev, "failed to request lock: %d", ret);
3773 if (reply_len > 0 && reply_len <= PAGE_SIZE) {
3774 void *p = page_address(reply_pages[0]);
3775 void *const end = p + reply_len;
3778 ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */
3783 ceph_decode_need(&p, end, 8 + 8, e_inval);
3784 p += 8 + 8; /* skip gid and cookie */
3786 ceph_decode_32_safe(&p, end, len, e_inval);
3790 if (lock_owner_responded) {
3792 "duplicate lock owners detected");
3797 lock_owner_responded = true;
3798 ret = ceph_start_decoding(&p, end, 1, "ResponseMessage",
3802 "failed to decode ResponseMessage: %d",
3807 ret = ceph_decode_32(&p);
3811 if (!lock_owner_responded) {
3812 rbd_warn(rbd_dev, "no lock owners detected");
3817 ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3826 * Either image request state machine(s) or rbd_add_acquire_lock()
3829 static void wake_lock_waiters(struct rbd_device *rbd_dev, int result)
3831 struct rbd_img_request *img_req;
3833 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
3834 lockdep_assert_held_write(&rbd_dev->lock_rwsem);
3836 cancel_delayed_work(&rbd_dev->lock_dwork);
3837 if (!completion_done(&rbd_dev->acquire_wait)) {
3838 rbd_assert(list_empty(&rbd_dev->acquiring_list) &&
3839 list_empty(&rbd_dev->running_list));
3840 rbd_dev->acquire_err = result;
3841 complete_all(&rbd_dev->acquire_wait);
3845 list_for_each_entry(img_req, &rbd_dev->acquiring_list, lock_item) {
3846 mutex_lock(&img_req->state_mutex);
3847 rbd_assert(img_req->state == RBD_IMG_EXCLUSIVE_LOCK);
3848 rbd_img_schedule(img_req, result);
3849 mutex_unlock(&img_req->state_mutex);
3852 list_splice_tail_init(&rbd_dev->acquiring_list, &rbd_dev->running_list);
3855 static bool locker_equal(const struct ceph_locker *lhs,
3856 const struct ceph_locker *rhs)
3858 return lhs->id.name.type == rhs->id.name.type &&
3859 lhs->id.name.num == rhs->id.name.num &&
3860 !strcmp(lhs->id.cookie, rhs->id.cookie) &&
3861 ceph_addr_equal_no_type(&lhs->info.addr, &rhs->info.addr);
3864 static void free_locker(struct ceph_locker *locker)
3867 ceph_free_lockers(locker, 1);
3870 static struct ceph_locker *get_lock_owner_info(struct rbd_device *rbd_dev)
3872 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3873 struct ceph_locker *lockers;
3880 ret = ceph_cls_lock_info(osdc, &rbd_dev->header_oid,
3881 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3882 &lock_type, &lock_tag, &lockers, &num_lockers);
3884 rbd_warn(rbd_dev, "failed to get header lockers: %d", ret);
3885 return ERR_PTR(ret);
3888 if (num_lockers == 0) {
3889 dout("%s rbd_dev %p no lockers detected\n", __func__, rbd_dev);
3894 if (strcmp(lock_tag, RBD_LOCK_TAG)) {
3895 rbd_warn(rbd_dev, "locked by external mechanism, tag %s",
3900 if (lock_type != CEPH_CLS_LOCK_EXCLUSIVE) {
3901 rbd_warn(rbd_dev, "incompatible lock type detected");
3905 WARN_ON(num_lockers != 1);
3906 ret = sscanf(lockers[0].id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu",
3909 rbd_warn(rbd_dev, "locked by external mechanism, cookie %s",
3910 lockers[0].id.cookie);
3913 if (ceph_addr_is_blank(&lockers[0].info.addr)) {
3914 rbd_warn(rbd_dev, "locker has a blank address");
3918 dout("%s rbd_dev %p got locker %s%llu@%pISpc/%u handle %llu\n",
3919 __func__, rbd_dev, ENTITY_NAME(lockers[0].id.name),
3920 &lockers[0].info.addr.in_addr,
3921 le32_to_cpu(lockers[0].info.addr.nonce), handle);
3929 ceph_free_lockers(lockers, num_lockers);
3930 return ERR_PTR(-EBUSY);
3933 static int find_watcher(struct rbd_device *rbd_dev,
3934 const struct ceph_locker *locker)
3936 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3937 struct ceph_watch_item *watchers;
3943 ret = ceph_osdc_list_watchers(osdc, &rbd_dev->header_oid,
3944 &rbd_dev->header_oloc, &watchers,
3947 rbd_warn(rbd_dev, "failed to get watchers: %d", ret);
3951 sscanf(locker->id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu", &cookie);
3952 for (i = 0; i < num_watchers; i++) {
3954 * Ignore addr->type while comparing. This mimics
3955 * entity_addr_t::get_legacy_str() + strcmp().
3957 if (ceph_addr_equal_no_type(&watchers[i].addr,
3958 &locker->info.addr) &&
3959 watchers[i].cookie == cookie) {
3960 struct rbd_client_id cid = {
3961 .gid = le64_to_cpu(watchers[i].name.num),
3965 dout("%s rbd_dev %p found cid %llu-%llu\n", __func__,
3966 rbd_dev, cid.gid, cid.handle);
3967 rbd_set_owner_cid(rbd_dev, &cid);
3973 dout("%s rbd_dev %p no watchers\n", __func__, rbd_dev);
3981 * lock_rwsem must be held for write
3983 static int rbd_try_lock(struct rbd_device *rbd_dev)
3985 struct ceph_client *client = rbd_dev->rbd_client->client;
3986 struct ceph_locker *locker, *refreshed_locker;
3990 locker = refreshed_locker = NULL;
3992 ret = rbd_lock(rbd_dev);
3995 if (ret != -EBUSY) {
3996 rbd_warn(rbd_dev, "failed to lock header: %d", ret);
4000 /* determine if the current lock holder is still alive */
4001 locker = get_lock_owner_info(rbd_dev);
4002 if (IS_ERR(locker)) {
4003 ret = PTR_ERR(locker);
4010 ret = find_watcher(rbd_dev, locker);
4012 goto out; /* request lock or error */
4014 refreshed_locker = get_lock_owner_info(rbd_dev);
4015 if (IS_ERR(refreshed_locker)) {
4016 ret = PTR_ERR(refreshed_locker);
4017 refreshed_locker = NULL;
4020 if (!refreshed_locker ||
4021 !locker_equal(locker, refreshed_locker))
4024 rbd_warn(rbd_dev, "breaking header lock owned by %s%llu",
4025 ENTITY_NAME(locker->id.name));
4027 ret = ceph_monc_blocklist_add(&client->monc,
4028 &locker->info.addr);
4030 rbd_warn(rbd_dev, "failed to blocklist %s%llu: %d",
4031 ENTITY_NAME(locker->id.name), ret);
4035 ret = ceph_cls_break_lock(&client->osdc, &rbd_dev->header_oid,
4036 &rbd_dev->header_oloc, RBD_LOCK_NAME,
4037 locker->id.cookie, &locker->id.name);
4038 if (ret && ret != -ENOENT) {
4039 rbd_warn(rbd_dev, "failed to break header lock: %d",
4045 free_locker(refreshed_locker);
4046 free_locker(locker);
4050 free_locker(refreshed_locker);
4051 free_locker(locker);
4055 static int rbd_post_acquire_action(struct rbd_device *rbd_dev)
4059 ret = rbd_dev_refresh(rbd_dev);
4063 if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) {
4064 ret = rbd_object_map_open(rbd_dev);
4075 * 1 - caller should call rbd_request_lock()
4078 static int rbd_try_acquire_lock(struct rbd_device *rbd_dev)
4082 down_read(&rbd_dev->lock_rwsem);
4083 dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
4084 rbd_dev->lock_state);
4085 if (__rbd_is_lock_owner(rbd_dev)) {
4086 up_read(&rbd_dev->lock_rwsem);
4090 up_read(&rbd_dev->lock_rwsem);
4091 down_write(&rbd_dev->lock_rwsem);
4092 dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
4093 rbd_dev->lock_state);
4094 if (__rbd_is_lock_owner(rbd_dev)) {
4095 up_write(&rbd_dev->lock_rwsem);
4099 ret = rbd_try_lock(rbd_dev);
4101 rbd_warn(rbd_dev, "failed to acquire lock: %d", ret);
4105 up_write(&rbd_dev->lock_rwsem);
4109 rbd_assert(rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED);
4110 rbd_assert(list_empty(&rbd_dev->running_list));
4112 ret = rbd_post_acquire_action(rbd_dev);
4114 rbd_warn(rbd_dev, "post-acquire action failed: %d", ret);
4116 * Can't stay in RBD_LOCK_STATE_LOCKED because
4117 * rbd_lock_add_request() would let the request through,
4118 * assuming that e.g. object map is locked and loaded.
4120 rbd_unlock(rbd_dev);
4124 wake_lock_waiters(rbd_dev, ret);
4125 up_write(&rbd_dev->lock_rwsem);
4129 static void rbd_acquire_lock(struct work_struct *work)
4131 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4132 struct rbd_device, lock_dwork);
4135 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4137 ret = rbd_try_acquire_lock(rbd_dev);
4139 dout("%s rbd_dev %p ret %d - done\n", __func__, rbd_dev, ret);
4143 ret = rbd_request_lock(rbd_dev);
4144 if (ret == -ETIMEDOUT) {
4145 goto again; /* treat this as a dead client */
4146 } else if (ret == -EROFS) {
4147 rbd_warn(rbd_dev, "peer will not release lock");
4148 down_write(&rbd_dev->lock_rwsem);
4149 wake_lock_waiters(rbd_dev, ret);
4150 up_write(&rbd_dev->lock_rwsem);
4151 } else if (ret < 0) {
4152 rbd_warn(rbd_dev, "error requesting lock: %d", ret);
4153 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4157 * lock owner acked, but resend if we don't see them
4160 dout("%s rbd_dev %p requeuing lock_dwork\n", __func__,
4162 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4163 msecs_to_jiffies(2 * RBD_NOTIFY_TIMEOUT * MSEC_PER_SEC));
4167 static bool rbd_quiesce_lock(struct rbd_device *rbd_dev)
4169 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4170 lockdep_assert_held_write(&rbd_dev->lock_rwsem);
4172 if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED)
4176 * Ensure that all in-flight IO is flushed.
4178 rbd_dev->lock_state = RBD_LOCK_STATE_RELEASING;
4179 rbd_assert(!completion_done(&rbd_dev->releasing_wait));
4180 if (list_empty(&rbd_dev->running_list))
4183 up_write(&rbd_dev->lock_rwsem);
4184 wait_for_completion(&rbd_dev->releasing_wait);
4186 down_write(&rbd_dev->lock_rwsem);
4187 if (rbd_dev->lock_state != RBD_LOCK_STATE_RELEASING)
4190 rbd_assert(list_empty(&rbd_dev->running_list));
4194 static void rbd_pre_release_action(struct rbd_device *rbd_dev)
4196 if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)
4197 rbd_object_map_close(rbd_dev);
4200 static void __rbd_release_lock(struct rbd_device *rbd_dev)
4202 rbd_assert(list_empty(&rbd_dev->running_list));
4204 rbd_pre_release_action(rbd_dev);
4205 rbd_unlock(rbd_dev);
4209 * lock_rwsem must be held for write
4211 static void rbd_release_lock(struct rbd_device *rbd_dev)
4213 if (!rbd_quiesce_lock(rbd_dev))
4216 __rbd_release_lock(rbd_dev);
4219 * Give others a chance to grab the lock - we would re-acquire
4220 * almost immediately if we got new IO while draining the running
4221 * list otherwise. We need to ack our own notifications, so this
4222 * lock_dwork will be requeued from rbd_handle_released_lock() by
4223 * way of maybe_kick_acquire().
4225 cancel_delayed_work(&rbd_dev->lock_dwork);
4228 static void rbd_release_lock_work(struct work_struct *work)
4230 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
4233 down_write(&rbd_dev->lock_rwsem);
4234 rbd_release_lock(rbd_dev);
4235 up_write(&rbd_dev->lock_rwsem);
4238 static void maybe_kick_acquire(struct rbd_device *rbd_dev)
4242 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4243 if (__rbd_is_lock_owner(rbd_dev))
4246 spin_lock(&rbd_dev->lock_lists_lock);
4247 have_requests = !list_empty(&rbd_dev->acquiring_list);
4248 spin_unlock(&rbd_dev->lock_lists_lock);
4249 if (have_requests || delayed_work_pending(&rbd_dev->lock_dwork)) {
4250 dout("%s rbd_dev %p kicking lock_dwork\n", __func__, rbd_dev);
4251 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4255 static void rbd_handle_acquired_lock(struct rbd_device *rbd_dev, u8 struct_v,
4258 struct rbd_client_id cid = { 0 };
4260 if (struct_v >= 2) {
4261 cid.gid = ceph_decode_64(p);
4262 cid.handle = ceph_decode_64(p);
4265 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4267 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4268 down_write(&rbd_dev->lock_rwsem);
4269 if (rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4270 dout("%s rbd_dev %p cid %llu-%llu == owner_cid\n",
4271 __func__, rbd_dev, cid.gid, cid.handle);
4273 rbd_set_owner_cid(rbd_dev, &cid);
4275 downgrade_write(&rbd_dev->lock_rwsem);
4277 down_read(&rbd_dev->lock_rwsem);
4280 maybe_kick_acquire(rbd_dev);
4281 up_read(&rbd_dev->lock_rwsem);
4284 static void rbd_handle_released_lock(struct rbd_device *rbd_dev, u8 struct_v,
4287 struct rbd_client_id cid = { 0 };
4289 if (struct_v >= 2) {
4290 cid.gid = ceph_decode_64(p);
4291 cid.handle = ceph_decode_64(p);
4294 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4296 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4297 down_write(&rbd_dev->lock_rwsem);
4298 if (!rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4299 dout("%s rbd_dev %p cid %llu-%llu != owner_cid %llu-%llu\n",
4300 __func__, rbd_dev, cid.gid, cid.handle,
4301 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle);
4303 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4305 downgrade_write(&rbd_dev->lock_rwsem);
4307 down_read(&rbd_dev->lock_rwsem);
4310 maybe_kick_acquire(rbd_dev);
4311 up_read(&rbd_dev->lock_rwsem);
4315 * Returns result for ResponseMessage to be encoded (<= 0), or 1 if no
4316 * ResponseMessage is needed.
4318 static int rbd_handle_request_lock(struct rbd_device *rbd_dev, u8 struct_v,
4321 struct rbd_client_id my_cid = rbd_get_cid(rbd_dev);
4322 struct rbd_client_id cid = { 0 };
4325 if (struct_v >= 2) {
4326 cid.gid = ceph_decode_64(p);
4327 cid.handle = ceph_decode_64(p);
4330 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4332 if (rbd_cid_equal(&cid, &my_cid))
4335 down_read(&rbd_dev->lock_rwsem);
4336 if (__rbd_is_lock_owner(rbd_dev)) {
4337 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED &&
4338 rbd_cid_equal(&rbd_dev->owner_cid, &rbd_empty_cid))
4342 * encode ResponseMessage(0) so the peer can detect
4347 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED) {
4348 if (!rbd_dev->opts->exclusive) {
4349 dout("%s rbd_dev %p queueing unlock_work\n",
4351 queue_work(rbd_dev->task_wq,
4352 &rbd_dev->unlock_work);
4354 /* refuse to release the lock */
4361 up_read(&rbd_dev->lock_rwsem);
4365 static void __rbd_acknowledge_notify(struct rbd_device *rbd_dev,
4366 u64 notify_id, u64 cookie, s32 *result)
4368 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4369 char buf[4 + CEPH_ENCODING_START_BLK_LEN];
4370 int buf_size = sizeof(buf);
4376 /* encode ResponseMessage */
4377 ceph_start_encoding(&p, 1, 1,
4378 buf_size - CEPH_ENCODING_START_BLK_LEN);
4379 ceph_encode_32(&p, *result);
4384 ret = ceph_osdc_notify_ack(osdc, &rbd_dev->header_oid,
4385 &rbd_dev->header_oloc, notify_id, cookie,
4388 rbd_warn(rbd_dev, "acknowledge_notify failed: %d", ret);
4391 static void rbd_acknowledge_notify(struct rbd_device *rbd_dev, u64 notify_id,
4394 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4395 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, NULL);
4398 static void rbd_acknowledge_notify_result(struct rbd_device *rbd_dev,
4399 u64 notify_id, u64 cookie, s32 result)
4401 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
4402 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, &result);
4405 static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie,
4406 u64 notifier_id, void *data, size_t data_len)
4408 struct rbd_device *rbd_dev = arg;
4410 void *const end = p + data_len;
4416 dout("%s rbd_dev %p cookie %llu notify_id %llu data_len %zu\n",
4417 __func__, rbd_dev, cookie, notify_id, data_len);
4419 ret = ceph_start_decoding(&p, end, 1, "NotifyMessage",
4422 rbd_warn(rbd_dev, "failed to decode NotifyMessage: %d",
4427 notify_op = ceph_decode_32(&p);
4429 /* legacy notification for header updates */
4430 notify_op = RBD_NOTIFY_OP_HEADER_UPDATE;
4434 dout("%s rbd_dev %p notify_op %u\n", __func__, rbd_dev, notify_op);
4435 switch (notify_op) {
4436 case RBD_NOTIFY_OP_ACQUIRED_LOCK:
4437 rbd_handle_acquired_lock(rbd_dev, struct_v, &p);
4438 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4440 case RBD_NOTIFY_OP_RELEASED_LOCK:
4441 rbd_handle_released_lock(rbd_dev, struct_v, &p);
4442 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4444 case RBD_NOTIFY_OP_REQUEST_LOCK:
4445 ret = rbd_handle_request_lock(rbd_dev, struct_v, &p);
4447 rbd_acknowledge_notify_result(rbd_dev, notify_id,
4450 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4452 case RBD_NOTIFY_OP_HEADER_UPDATE:
4453 ret = rbd_dev_refresh(rbd_dev);
4455 rbd_warn(rbd_dev, "refresh failed: %d", ret);
4457 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4460 if (rbd_is_lock_owner(rbd_dev))
4461 rbd_acknowledge_notify_result(rbd_dev, notify_id,
4462 cookie, -EOPNOTSUPP);
4464 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4469 static void __rbd_unregister_watch(struct rbd_device *rbd_dev);
4471 static void rbd_watch_errcb(void *arg, u64 cookie, int err)
4473 struct rbd_device *rbd_dev = arg;
4475 rbd_warn(rbd_dev, "encountered watch error: %d", err);
4477 down_write(&rbd_dev->lock_rwsem);
4478 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4479 up_write(&rbd_dev->lock_rwsem);
4481 mutex_lock(&rbd_dev->watch_mutex);
4482 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED) {
4483 __rbd_unregister_watch(rbd_dev);
4484 rbd_dev->watch_state = RBD_WATCH_STATE_ERROR;
4486 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->watch_dwork, 0);
4488 mutex_unlock(&rbd_dev->watch_mutex);
4492 * watch_mutex must be locked
4494 static int __rbd_register_watch(struct rbd_device *rbd_dev)
4496 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4497 struct ceph_osd_linger_request *handle;
4499 rbd_assert(!rbd_dev->watch_handle);
4500 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4502 handle = ceph_osdc_watch(osdc, &rbd_dev->header_oid,
4503 &rbd_dev->header_oloc, rbd_watch_cb,
4504 rbd_watch_errcb, rbd_dev);
4506 return PTR_ERR(handle);
4508 rbd_dev->watch_handle = handle;
4513 * watch_mutex must be locked
4515 static void __rbd_unregister_watch(struct rbd_device *rbd_dev)
4517 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4520 rbd_assert(rbd_dev->watch_handle);
4521 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4523 ret = ceph_osdc_unwatch(osdc, rbd_dev->watch_handle);
4525 rbd_warn(rbd_dev, "failed to unwatch: %d", ret);
4527 rbd_dev->watch_handle = NULL;
4530 static int rbd_register_watch(struct rbd_device *rbd_dev)
4534 mutex_lock(&rbd_dev->watch_mutex);
4535 rbd_assert(rbd_dev->watch_state == RBD_WATCH_STATE_UNREGISTERED);
4536 ret = __rbd_register_watch(rbd_dev);
4540 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4541 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4544 mutex_unlock(&rbd_dev->watch_mutex);
4548 static void cancel_tasks_sync(struct rbd_device *rbd_dev)
4550 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4552 cancel_work_sync(&rbd_dev->acquired_lock_work);
4553 cancel_work_sync(&rbd_dev->released_lock_work);
4554 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
4555 cancel_work_sync(&rbd_dev->unlock_work);
4559 * header_rwsem must not be held to avoid a deadlock with
4560 * rbd_dev_refresh() when flushing notifies.
4562 static void rbd_unregister_watch(struct rbd_device *rbd_dev)
4564 cancel_tasks_sync(rbd_dev);
4566 mutex_lock(&rbd_dev->watch_mutex);
4567 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED)
4568 __rbd_unregister_watch(rbd_dev);
4569 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
4570 mutex_unlock(&rbd_dev->watch_mutex);
4572 cancel_delayed_work_sync(&rbd_dev->watch_dwork);
4573 ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
4577 * lock_rwsem must be held for write
4579 static void rbd_reacquire_lock(struct rbd_device *rbd_dev)
4581 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4585 if (!rbd_quiesce_lock(rbd_dev))
4588 format_lock_cookie(rbd_dev, cookie);
4589 ret = ceph_cls_set_cookie(osdc, &rbd_dev->header_oid,
4590 &rbd_dev->header_oloc, RBD_LOCK_NAME,
4591 CEPH_CLS_LOCK_EXCLUSIVE, rbd_dev->lock_cookie,
4592 RBD_LOCK_TAG, cookie);
4594 if (ret != -EOPNOTSUPP)
4595 rbd_warn(rbd_dev, "failed to update lock cookie: %d",
4599 * Lock cookie cannot be updated on older OSDs, so do
4600 * a manual release and queue an acquire.
4602 __rbd_release_lock(rbd_dev);
4603 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4605 __rbd_lock(rbd_dev, cookie);
4606 wake_lock_waiters(rbd_dev, 0);
4610 static void rbd_reregister_watch(struct work_struct *work)
4612 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4613 struct rbd_device, watch_dwork);
4616 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4618 mutex_lock(&rbd_dev->watch_mutex);
4619 if (rbd_dev->watch_state != RBD_WATCH_STATE_ERROR) {
4620 mutex_unlock(&rbd_dev->watch_mutex);
4624 ret = __rbd_register_watch(rbd_dev);
4626 rbd_warn(rbd_dev, "failed to reregister watch: %d", ret);
4627 if (ret != -EBLOCKLISTED && ret != -ENOENT) {
4628 queue_delayed_work(rbd_dev->task_wq,
4629 &rbd_dev->watch_dwork,
4631 mutex_unlock(&rbd_dev->watch_mutex);
4635 mutex_unlock(&rbd_dev->watch_mutex);
4636 down_write(&rbd_dev->lock_rwsem);
4637 wake_lock_waiters(rbd_dev, ret);
4638 up_write(&rbd_dev->lock_rwsem);
4642 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4643 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4644 mutex_unlock(&rbd_dev->watch_mutex);
4646 down_write(&rbd_dev->lock_rwsem);
4647 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED)
4648 rbd_reacquire_lock(rbd_dev);
4649 up_write(&rbd_dev->lock_rwsem);
4651 ret = rbd_dev_refresh(rbd_dev);
4653 rbd_warn(rbd_dev, "reregistration refresh failed: %d", ret);
4657 * Synchronous osd object method call. Returns the number of bytes
4658 * returned in the outbound buffer, or a negative error code.
4660 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
4661 struct ceph_object_id *oid,
4662 struct ceph_object_locator *oloc,
4663 const char *method_name,
4664 const void *outbound,
4665 size_t outbound_size,
4667 size_t inbound_size)
4669 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4670 struct page *req_page = NULL;
4671 struct page *reply_page;
4675 * Method calls are ultimately read operations. The result
4676 * should placed into the inbound buffer provided. They
4677 * also supply outbound data--parameters for the object
4678 * method. Currently if this is present it will be a
4682 if (outbound_size > PAGE_SIZE)
4685 req_page = alloc_page(GFP_KERNEL);
4689 memcpy(page_address(req_page), outbound, outbound_size);
4692 reply_page = alloc_page(GFP_KERNEL);
4695 __free_page(req_page);
4699 ret = ceph_osdc_call(osdc, oid, oloc, RBD_DRV_NAME, method_name,
4700 CEPH_OSD_FLAG_READ, req_page, outbound_size,
4701 &reply_page, &inbound_size);
4703 memcpy(inbound, page_address(reply_page), inbound_size);
4708 __free_page(req_page);
4709 __free_page(reply_page);
4713 static void rbd_queue_workfn(struct work_struct *work)
4715 struct rbd_img_request *img_request =
4716 container_of(work, struct rbd_img_request, work);
4717 struct rbd_device *rbd_dev = img_request->rbd_dev;
4718 enum obj_operation_type op_type = img_request->op_type;
4719 struct request *rq = blk_mq_rq_from_pdu(img_request);
4720 u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
4721 u64 length = blk_rq_bytes(rq);
4725 /* Ignore/skip any zero-length requests */
4727 dout("%s: zero-length request\n", __func__);
4729 goto err_img_request;
4732 blk_mq_start_request(rq);
4734 down_read(&rbd_dev->header_rwsem);
4735 mapping_size = rbd_dev->mapping.size;
4736 rbd_img_capture_header(img_request);
4737 up_read(&rbd_dev->header_rwsem);
4739 if (offset + length > mapping_size) {
4740 rbd_warn(rbd_dev, "beyond EOD (%llu~%llu > %llu)", offset,
4741 length, mapping_size);
4743 goto err_img_request;
4746 dout("%s rbd_dev %p img_req %p %s %llu~%llu\n", __func__, rbd_dev,
4747 img_request, obj_op_name(op_type), offset, length);
4749 if (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_ZEROOUT)
4750 result = rbd_img_fill_nodata(img_request, offset, length);
4752 result = rbd_img_fill_from_bio(img_request, offset, length,
4755 goto err_img_request;
4757 rbd_img_handle_request(img_request, 0);
4761 rbd_img_request_destroy(img_request);
4763 rbd_warn(rbd_dev, "%s %llx at %llx result %d",
4764 obj_op_name(op_type), length, offset, result);
4765 blk_mq_end_request(rq, errno_to_blk_status(result));
4768 static blk_status_t rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
4769 const struct blk_mq_queue_data *bd)
4771 struct rbd_device *rbd_dev = hctx->queue->queuedata;
4772 struct rbd_img_request *img_req = blk_mq_rq_to_pdu(bd->rq);
4773 enum obj_operation_type op_type;
4775 switch (req_op(bd->rq)) {
4776 case REQ_OP_DISCARD:
4777 op_type = OBJ_OP_DISCARD;
4779 case REQ_OP_WRITE_ZEROES:
4780 op_type = OBJ_OP_ZEROOUT;
4783 op_type = OBJ_OP_WRITE;
4786 op_type = OBJ_OP_READ;
4789 rbd_warn(rbd_dev, "unknown req_op %d", req_op(bd->rq));
4790 return BLK_STS_IOERR;
4793 rbd_img_request_init(img_req, rbd_dev, op_type);
4795 if (rbd_img_is_write(img_req)) {
4796 if (rbd_is_ro(rbd_dev)) {
4797 rbd_warn(rbd_dev, "%s on read-only mapping",
4798 obj_op_name(img_req->op_type));
4799 return BLK_STS_IOERR;
4801 rbd_assert(!rbd_is_snap(rbd_dev));
4804 INIT_WORK(&img_req->work, rbd_queue_workfn);
4805 queue_work(rbd_wq, &img_req->work);
4809 static void rbd_free_disk(struct rbd_device *rbd_dev)
4811 put_disk(rbd_dev->disk);
4812 blk_mq_free_tag_set(&rbd_dev->tag_set);
4813 rbd_dev->disk = NULL;
4816 static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
4817 struct ceph_object_id *oid,
4818 struct ceph_object_locator *oloc,
4819 void *buf, int buf_len)
4822 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4823 struct ceph_osd_request *req;
4824 struct page **pages;
4825 int num_pages = calc_pages_for(0, buf_len);
4828 req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_KERNEL);
4832 ceph_oid_copy(&req->r_base_oid, oid);
4833 ceph_oloc_copy(&req->r_base_oloc, oloc);
4834 req->r_flags = CEPH_OSD_FLAG_READ;
4836 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
4837 if (IS_ERR(pages)) {
4838 ret = PTR_ERR(pages);
4842 osd_req_op_extent_init(req, 0, CEPH_OSD_OP_READ, 0, buf_len, 0, 0);
4843 osd_req_op_extent_osd_data_pages(req, 0, pages, buf_len, 0, false,
4846 ret = ceph_osdc_alloc_messages(req, GFP_KERNEL);
4850 ceph_osdc_start_request(osdc, req);
4851 ret = ceph_osdc_wait_request(osdc, req);
4853 ceph_copy_from_page_vector(pages, buf, 0, ret);
4856 ceph_osdc_put_request(req);
4861 * Read the complete header for the given rbd device. On successful
4862 * return, the rbd_dev->header field will contain up-to-date
4863 * information about the image.
4865 static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev,
4866 struct rbd_image_header *header,
4869 struct rbd_image_header_ondisk *ondisk = NULL;
4876 * The complete header will include an array of its 64-bit
4877 * snapshot ids, followed by the names of those snapshots as
4878 * a contiguous block of NUL-terminated strings. Note that
4879 * the number of snapshots could change by the time we read
4880 * it in, in which case we re-read it.
4887 size = sizeof (*ondisk);
4888 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
4890 ondisk = kmalloc(size, GFP_KERNEL);
4894 ret = rbd_obj_read_sync(rbd_dev, &rbd_dev->header_oid,
4895 &rbd_dev->header_oloc, ondisk, size);
4898 if ((size_t)ret < size) {
4900 rbd_warn(rbd_dev, "short header read (want %zd got %d)",
4904 if (!rbd_dev_ondisk_valid(ondisk)) {
4906 rbd_warn(rbd_dev, "invalid header");
4910 names_size = le64_to_cpu(ondisk->snap_names_len);
4911 want_count = snap_count;
4912 snap_count = le32_to_cpu(ondisk->snap_count);
4913 } while (snap_count != want_count);
4915 ret = rbd_header_from_disk(header, ondisk, first_time);
4922 static void rbd_dev_update_size(struct rbd_device *rbd_dev)
4927 * If EXISTS is not set, rbd_dev->disk may be NULL, so don't
4928 * try to update its size. If REMOVING is set, updating size
4929 * is just useless work since the device can't be opened.
4931 if (test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags) &&
4932 !test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) {
4933 size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
4934 dout("setting size to %llu sectors", (unsigned long long)size);
4935 set_capacity_and_notify(rbd_dev->disk, size);
4939 static const struct blk_mq_ops rbd_mq_ops = {
4940 .queue_rq = rbd_queue_rq,
4943 static int rbd_init_disk(struct rbd_device *rbd_dev)
4945 struct gendisk *disk;
4946 struct request_queue *q;
4947 unsigned int objset_bytes =
4948 rbd_dev->layout.object_size * rbd_dev->layout.stripe_count;
4951 memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
4952 rbd_dev->tag_set.ops = &rbd_mq_ops;
4953 rbd_dev->tag_set.queue_depth = rbd_dev->opts->queue_depth;
4954 rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
4955 rbd_dev->tag_set.flags = BLK_MQ_F_SHOULD_MERGE;
4956 rbd_dev->tag_set.nr_hw_queues = num_present_cpus();
4957 rbd_dev->tag_set.cmd_size = sizeof(struct rbd_img_request);
4959 err = blk_mq_alloc_tag_set(&rbd_dev->tag_set);
4963 disk = blk_mq_alloc_disk(&rbd_dev->tag_set, rbd_dev);
4965 err = PTR_ERR(disk);
4970 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
4972 disk->major = rbd_dev->major;
4973 disk->first_minor = rbd_dev->minor;
4975 disk->minors = (1 << RBD_SINGLE_MAJOR_PART_SHIFT);
4977 disk->minors = RBD_MINORS_PER_MAJOR;
4978 disk->fops = &rbd_bd_ops;
4979 disk->private_data = rbd_dev;
4981 blk_queue_flag_set(QUEUE_FLAG_NONROT, q);
4982 /* QUEUE_FLAG_ADD_RANDOM is off by default for blk-mq */
4984 blk_queue_max_hw_sectors(q, objset_bytes >> SECTOR_SHIFT);
4985 q->limits.max_sectors = queue_max_hw_sectors(q);
4986 blk_queue_max_segments(q, USHRT_MAX);
4987 blk_queue_max_segment_size(q, UINT_MAX);
4988 blk_queue_io_min(q, rbd_dev->opts->alloc_size);
4989 blk_queue_io_opt(q, rbd_dev->opts->alloc_size);
4991 if (rbd_dev->opts->trim) {
4992 q->limits.discard_granularity = rbd_dev->opts->alloc_size;
4993 blk_queue_max_discard_sectors(q, objset_bytes >> SECTOR_SHIFT);
4994 blk_queue_max_write_zeroes_sectors(q, objset_bytes >> SECTOR_SHIFT);
4997 if (!ceph_test_opt(rbd_dev->rbd_client->client, NOCRC))
4998 blk_queue_flag_set(QUEUE_FLAG_STABLE_WRITES, q);
5000 rbd_dev->disk = disk;
5004 blk_mq_free_tag_set(&rbd_dev->tag_set);
5012 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
5014 return container_of(dev, struct rbd_device, dev);
5017 static ssize_t rbd_size_show(struct device *dev,
5018 struct device_attribute *attr, char *buf)
5020 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5022 return sprintf(buf, "%llu\n",
5023 (unsigned long long)rbd_dev->mapping.size);
5026 static ssize_t rbd_features_show(struct device *dev,
5027 struct device_attribute *attr, char *buf)
5029 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5031 return sprintf(buf, "0x%016llx\n", rbd_dev->header.features);
5034 static ssize_t rbd_major_show(struct device *dev,
5035 struct device_attribute *attr, char *buf)
5037 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5040 return sprintf(buf, "%d\n", rbd_dev->major);
5042 return sprintf(buf, "(none)\n");
5045 static ssize_t rbd_minor_show(struct device *dev,
5046 struct device_attribute *attr, char *buf)
5048 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5050 return sprintf(buf, "%d\n", rbd_dev->minor);
5053 static ssize_t rbd_client_addr_show(struct device *dev,
5054 struct device_attribute *attr, char *buf)
5056 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5057 struct ceph_entity_addr *client_addr =
5058 ceph_client_addr(rbd_dev->rbd_client->client);
5060 return sprintf(buf, "%pISpc/%u\n", &client_addr->in_addr,
5061 le32_to_cpu(client_addr->nonce));
5064 static ssize_t rbd_client_id_show(struct device *dev,
5065 struct device_attribute *attr, char *buf)
5067 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5069 return sprintf(buf, "client%lld\n",
5070 ceph_client_gid(rbd_dev->rbd_client->client));
5073 static ssize_t rbd_cluster_fsid_show(struct device *dev,
5074 struct device_attribute *attr, char *buf)
5076 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5078 return sprintf(buf, "%pU\n", &rbd_dev->rbd_client->client->fsid);
5081 static ssize_t rbd_config_info_show(struct device *dev,
5082 struct device_attribute *attr, char *buf)
5084 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5086 if (!capable(CAP_SYS_ADMIN))
5089 return sprintf(buf, "%s\n", rbd_dev->config_info);
5092 static ssize_t rbd_pool_show(struct device *dev,
5093 struct device_attribute *attr, char *buf)
5095 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5097 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
5100 static ssize_t rbd_pool_id_show(struct device *dev,
5101 struct device_attribute *attr, char *buf)
5103 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5105 return sprintf(buf, "%llu\n",
5106 (unsigned long long) rbd_dev->spec->pool_id);
5109 static ssize_t rbd_pool_ns_show(struct device *dev,
5110 struct device_attribute *attr, char *buf)
5112 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5114 return sprintf(buf, "%s\n", rbd_dev->spec->pool_ns ?: "");
5117 static ssize_t rbd_name_show(struct device *dev,
5118 struct device_attribute *attr, char *buf)
5120 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5122 if (rbd_dev->spec->image_name)
5123 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
5125 return sprintf(buf, "(unknown)\n");
5128 static ssize_t rbd_image_id_show(struct device *dev,
5129 struct device_attribute *attr, char *buf)
5131 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5133 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
5137 * Shows the name of the currently-mapped snapshot (or
5138 * RBD_SNAP_HEAD_NAME for the base image).
5140 static ssize_t rbd_snap_show(struct device *dev,
5141 struct device_attribute *attr,
5144 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5146 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
5149 static ssize_t rbd_snap_id_show(struct device *dev,
5150 struct device_attribute *attr, char *buf)
5152 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5154 return sprintf(buf, "%llu\n", rbd_dev->spec->snap_id);
5158 * For a v2 image, shows the chain of parent images, separated by empty
5159 * lines. For v1 images or if there is no parent, shows "(no parent
5162 static ssize_t rbd_parent_show(struct device *dev,
5163 struct device_attribute *attr,
5166 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5169 if (!rbd_dev->parent)
5170 return sprintf(buf, "(no parent image)\n");
5172 for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) {
5173 struct rbd_spec *spec = rbd_dev->parent_spec;
5175 count += sprintf(&buf[count], "%s"
5176 "pool_id %llu\npool_name %s\n"
5178 "image_id %s\nimage_name %s\n"
5179 "snap_id %llu\nsnap_name %s\n"
5181 !count ? "" : "\n", /* first? */
5182 spec->pool_id, spec->pool_name,
5183 spec->pool_ns ?: "",
5184 spec->image_id, spec->image_name ?: "(unknown)",
5185 spec->snap_id, spec->snap_name,
5186 rbd_dev->parent_overlap);
5192 static ssize_t rbd_image_refresh(struct device *dev,
5193 struct device_attribute *attr,
5197 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5200 if (!capable(CAP_SYS_ADMIN))
5203 ret = rbd_dev_refresh(rbd_dev);
5210 static DEVICE_ATTR(size, 0444, rbd_size_show, NULL);
5211 static DEVICE_ATTR(features, 0444, rbd_features_show, NULL);
5212 static DEVICE_ATTR(major, 0444, rbd_major_show, NULL);
5213 static DEVICE_ATTR(minor, 0444, rbd_minor_show, NULL);
5214 static DEVICE_ATTR(client_addr, 0444, rbd_client_addr_show, NULL);
5215 static DEVICE_ATTR(client_id, 0444, rbd_client_id_show, NULL);
5216 static DEVICE_ATTR(cluster_fsid, 0444, rbd_cluster_fsid_show, NULL);
5217 static DEVICE_ATTR(config_info, 0400, rbd_config_info_show, NULL);
5218 static DEVICE_ATTR(pool, 0444, rbd_pool_show, NULL);
5219 static DEVICE_ATTR(pool_id, 0444, rbd_pool_id_show, NULL);
5220 static DEVICE_ATTR(pool_ns, 0444, rbd_pool_ns_show, NULL);
5221 static DEVICE_ATTR(name, 0444, rbd_name_show, NULL);
5222 static DEVICE_ATTR(image_id, 0444, rbd_image_id_show, NULL);
5223 static DEVICE_ATTR(refresh, 0200, NULL, rbd_image_refresh);
5224 static DEVICE_ATTR(current_snap, 0444, rbd_snap_show, NULL);
5225 static DEVICE_ATTR(snap_id, 0444, rbd_snap_id_show, NULL);
5226 static DEVICE_ATTR(parent, 0444, rbd_parent_show, NULL);
5228 static struct attribute *rbd_attrs[] = {
5229 &dev_attr_size.attr,
5230 &dev_attr_features.attr,
5231 &dev_attr_major.attr,
5232 &dev_attr_minor.attr,
5233 &dev_attr_client_addr.attr,
5234 &dev_attr_client_id.attr,
5235 &dev_attr_cluster_fsid.attr,
5236 &dev_attr_config_info.attr,
5237 &dev_attr_pool.attr,
5238 &dev_attr_pool_id.attr,
5239 &dev_attr_pool_ns.attr,
5240 &dev_attr_name.attr,
5241 &dev_attr_image_id.attr,
5242 &dev_attr_current_snap.attr,
5243 &dev_attr_snap_id.attr,
5244 &dev_attr_parent.attr,
5245 &dev_attr_refresh.attr,
5249 static struct attribute_group rbd_attr_group = {
5253 static const struct attribute_group *rbd_attr_groups[] = {
5258 static void rbd_dev_release(struct device *dev);
5260 static const struct device_type rbd_device_type = {
5262 .groups = rbd_attr_groups,
5263 .release = rbd_dev_release,
5266 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
5268 kref_get(&spec->kref);
5273 static void rbd_spec_free(struct kref *kref);
5274 static void rbd_spec_put(struct rbd_spec *spec)
5277 kref_put(&spec->kref, rbd_spec_free);
5280 static struct rbd_spec *rbd_spec_alloc(void)
5282 struct rbd_spec *spec;
5284 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
5288 spec->pool_id = CEPH_NOPOOL;
5289 spec->snap_id = CEPH_NOSNAP;
5290 kref_init(&spec->kref);
5295 static void rbd_spec_free(struct kref *kref)
5297 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
5299 kfree(spec->pool_name);
5300 kfree(spec->pool_ns);
5301 kfree(spec->image_id);
5302 kfree(spec->image_name);
5303 kfree(spec->snap_name);
5307 static void rbd_dev_free(struct rbd_device *rbd_dev)
5309 WARN_ON(rbd_dev->watch_state != RBD_WATCH_STATE_UNREGISTERED);
5310 WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_UNLOCKED);
5312 ceph_oid_destroy(&rbd_dev->header_oid);
5313 ceph_oloc_destroy(&rbd_dev->header_oloc);
5314 kfree(rbd_dev->config_info);
5316 rbd_put_client(rbd_dev->rbd_client);
5317 rbd_spec_put(rbd_dev->spec);
5318 kfree(rbd_dev->opts);
5322 static void rbd_dev_release(struct device *dev)
5324 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5325 bool need_put = !!rbd_dev->opts;
5328 destroy_workqueue(rbd_dev->task_wq);
5329 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5332 rbd_dev_free(rbd_dev);
5335 * This is racy, but way better than putting module outside of
5336 * the release callback. The race window is pretty small, so
5337 * doing something similar to dm (dm-builtin.c) is overkill.
5340 module_put(THIS_MODULE);
5343 static struct rbd_device *__rbd_dev_create(struct rbd_spec *spec)
5345 struct rbd_device *rbd_dev;
5347 rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
5351 spin_lock_init(&rbd_dev->lock);
5352 INIT_LIST_HEAD(&rbd_dev->node);
5353 init_rwsem(&rbd_dev->header_rwsem);
5355 rbd_dev->header.data_pool_id = CEPH_NOPOOL;
5356 ceph_oid_init(&rbd_dev->header_oid);
5357 rbd_dev->header_oloc.pool = spec->pool_id;
5358 if (spec->pool_ns) {
5359 WARN_ON(!*spec->pool_ns);
5360 rbd_dev->header_oloc.pool_ns =
5361 ceph_find_or_create_string(spec->pool_ns,
5362 strlen(spec->pool_ns));
5365 mutex_init(&rbd_dev->watch_mutex);
5366 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
5367 INIT_DELAYED_WORK(&rbd_dev->watch_dwork, rbd_reregister_watch);
5369 init_rwsem(&rbd_dev->lock_rwsem);
5370 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
5371 INIT_WORK(&rbd_dev->acquired_lock_work, rbd_notify_acquired_lock);
5372 INIT_WORK(&rbd_dev->released_lock_work, rbd_notify_released_lock);
5373 INIT_DELAYED_WORK(&rbd_dev->lock_dwork, rbd_acquire_lock);
5374 INIT_WORK(&rbd_dev->unlock_work, rbd_release_lock_work);
5375 spin_lock_init(&rbd_dev->lock_lists_lock);
5376 INIT_LIST_HEAD(&rbd_dev->acquiring_list);
5377 INIT_LIST_HEAD(&rbd_dev->running_list);
5378 init_completion(&rbd_dev->acquire_wait);
5379 init_completion(&rbd_dev->releasing_wait);
5381 spin_lock_init(&rbd_dev->object_map_lock);
5383 rbd_dev->dev.bus = &rbd_bus_type;
5384 rbd_dev->dev.type = &rbd_device_type;
5385 rbd_dev->dev.parent = &rbd_root_dev;
5386 device_initialize(&rbd_dev->dev);
5392 * Create a mapping rbd_dev.
5394 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
5395 struct rbd_spec *spec,
5396 struct rbd_options *opts)
5398 struct rbd_device *rbd_dev;
5400 rbd_dev = __rbd_dev_create(spec);
5404 /* get an id and fill in device name */
5405 rbd_dev->dev_id = ida_simple_get(&rbd_dev_id_ida, 0,
5406 minor_to_rbd_dev_id(1 << MINORBITS),
5408 if (rbd_dev->dev_id < 0)
5411 sprintf(rbd_dev->name, RBD_DRV_NAME "%d", rbd_dev->dev_id);
5412 rbd_dev->task_wq = alloc_ordered_workqueue("%s-tasks", WQ_MEM_RECLAIM,
5414 if (!rbd_dev->task_wq)
5417 /* we have a ref from do_rbd_add() */
5418 __module_get(THIS_MODULE);
5420 rbd_dev->rbd_client = rbdc;
5421 rbd_dev->spec = spec;
5422 rbd_dev->opts = opts;
5424 dout("%s rbd_dev %p dev_id %d\n", __func__, rbd_dev, rbd_dev->dev_id);
5428 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5430 rbd_dev_free(rbd_dev);
5434 static void rbd_dev_destroy(struct rbd_device *rbd_dev)
5437 put_device(&rbd_dev->dev);
5441 * Get the size and object order for an image snapshot, or if
5442 * snap_id is CEPH_NOSNAP, gets this information for the base
5445 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
5446 u8 *order, u64 *snap_size)
5448 __le64 snapid = cpu_to_le64(snap_id);
5453 } __attribute__ ((packed)) size_buf = { 0 };
5455 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5456 &rbd_dev->header_oloc, "get_size",
5457 &snapid, sizeof(snapid),
5458 &size_buf, sizeof(size_buf));
5459 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5462 if (ret < sizeof (size_buf))
5466 *order = size_buf.order;
5467 dout(" order %u", (unsigned int)*order);
5469 *snap_size = le64_to_cpu(size_buf.size);
5471 dout(" snap_id 0x%016llx snap_size = %llu\n",
5472 (unsigned long long)snap_id,
5473 (unsigned long long)*snap_size);
5478 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev,
5479 char **pobject_prefix)
5483 char *object_prefix;
5487 /* Response will be an encoded string, which includes a length */
5488 size = sizeof(__le32) + RBD_OBJ_PREFIX_LEN_MAX;
5489 reply_buf = kzalloc(size, GFP_KERNEL);
5493 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5494 &rbd_dev->header_oloc, "get_object_prefix",
5495 NULL, 0, reply_buf, size);
5496 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5501 object_prefix = ceph_extract_encoded_string(&p, p + ret, NULL,
5503 if (IS_ERR(object_prefix)) {
5504 ret = PTR_ERR(object_prefix);
5509 *pobject_prefix = object_prefix;
5510 dout(" object_prefix = %s\n", object_prefix);
5517 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
5518 bool read_only, u64 *snap_features)
5527 } __attribute__ ((packed)) features_buf = { 0 };
5531 features_in.snap_id = cpu_to_le64(snap_id);
5532 features_in.read_only = read_only;
5534 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5535 &rbd_dev->header_oloc, "get_features",
5536 &features_in, sizeof(features_in),
5537 &features_buf, sizeof(features_buf));
5538 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5541 if (ret < sizeof (features_buf))
5544 unsup = le64_to_cpu(features_buf.incompat) & ~RBD_FEATURES_SUPPORTED;
5546 rbd_warn(rbd_dev, "image uses unsupported features: 0x%llx",
5551 *snap_features = le64_to_cpu(features_buf.features);
5553 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
5554 (unsigned long long)snap_id,
5555 (unsigned long long)*snap_features,
5556 (unsigned long long)le64_to_cpu(features_buf.incompat));
5562 * These are generic image flags, but since they are used only for
5563 * object map, store them in rbd_dev->object_map_flags.
5565 * For the same reason, this function is called only on object map
5566 * (re)load and not on header refresh.
5568 static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev)
5570 __le64 snapid = cpu_to_le64(rbd_dev->spec->snap_id);
5574 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5575 &rbd_dev->header_oloc, "get_flags",
5576 &snapid, sizeof(snapid),
5577 &flags, sizeof(flags));
5580 if (ret < sizeof(flags))
5583 rbd_dev->object_map_flags = le64_to_cpu(flags);
5587 struct parent_image_info {
5589 const char *pool_ns;
5590 const char *image_id;
5597 static void rbd_parent_info_cleanup(struct parent_image_info *pii)
5599 kfree(pii->pool_ns);
5600 kfree(pii->image_id);
5602 memset(pii, 0, sizeof(*pii));
5606 * The caller is responsible for @pii.
5608 static int decode_parent_image_spec(void **p, void *end,
5609 struct parent_image_info *pii)
5615 ret = ceph_start_decoding(p, end, 1, "ParentImageSpec",
5616 &struct_v, &struct_len);
5620 ceph_decode_64_safe(p, end, pii->pool_id, e_inval);
5621 pii->pool_ns = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5622 if (IS_ERR(pii->pool_ns)) {
5623 ret = PTR_ERR(pii->pool_ns);
5624 pii->pool_ns = NULL;
5627 pii->image_id = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5628 if (IS_ERR(pii->image_id)) {
5629 ret = PTR_ERR(pii->image_id);
5630 pii->image_id = NULL;
5633 ceph_decode_64_safe(p, end, pii->snap_id, e_inval);
5640 static int __get_parent_info(struct rbd_device *rbd_dev,
5641 struct page *req_page,
5642 struct page *reply_page,
5643 struct parent_image_info *pii)
5645 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5646 size_t reply_len = PAGE_SIZE;
5650 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5651 "rbd", "parent_get", CEPH_OSD_FLAG_READ,
5652 req_page, sizeof(u64), &reply_page, &reply_len);
5654 return ret == -EOPNOTSUPP ? 1 : ret;
5656 p = page_address(reply_page);
5657 end = p + reply_len;
5658 ret = decode_parent_image_spec(&p, end, pii);
5662 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5663 "rbd", "parent_overlap_get", CEPH_OSD_FLAG_READ,
5664 req_page, sizeof(u64), &reply_page, &reply_len);
5668 p = page_address(reply_page);
5669 end = p + reply_len;
5670 ceph_decode_8_safe(&p, end, pii->has_overlap, e_inval);
5671 if (pii->has_overlap)
5672 ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5674 dout("%s pool_id %llu pool_ns %s image_id %s snap_id %llu has_overlap %d overlap %llu\n",
5675 __func__, pii->pool_id, pii->pool_ns, pii->image_id, pii->snap_id,
5676 pii->has_overlap, pii->overlap);
5684 * The caller is responsible for @pii.
5686 static int __get_parent_info_legacy(struct rbd_device *rbd_dev,
5687 struct page *req_page,
5688 struct page *reply_page,
5689 struct parent_image_info *pii)
5691 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5692 size_t reply_len = PAGE_SIZE;
5696 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5697 "rbd", "get_parent", CEPH_OSD_FLAG_READ,
5698 req_page, sizeof(u64), &reply_page, &reply_len);
5702 p = page_address(reply_page);
5703 end = p + reply_len;
5704 ceph_decode_64_safe(&p, end, pii->pool_id, e_inval);
5705 pii->image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
5706 if (IS_ERR(pii->image_id)) {
5707 ret = PTR_ERR(pii->image_id);
5708 pii->image_id = NULL;
5711 ceph_decode_64_safe(&p, end, pii->snap_id, e_inval);
5712 pii->has_overlap = true;
5713 ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5715 dout("%s pool_id %llu pool_ns %s image_id %s snap_id %llu has_overlap %d overlap %llu\n",
5716 __func__, pii->pool_id, pii->pool_ns, pii->image_id, pii->snap_id,
5717 pii->has_overlap, pii->overlap);
5724 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev,
5725 struct parent_image_info *pii)
5727 struct page *req_page, *reply_page;
5731 req_page = alloc_page(GFP_KERNEL);
5735 reply_page = alloc_page(GFP_KERNEL);
5737 __free_page(req_page);
5741 p = page_address(req_page);
5742 ceph_encode_64(&p, rbd_dev->spec->snap_id);
5743 ret = __get_parent_info(rbd_dev, req_page, reply_page, pii);
5745 ret = __get_parent_info_legacy(rbd_dev, req_page, reply_page,
5748 __free_page(req_page);
5749 __free_page(reply_page);
5753 static int rbd_dev_setup_parent(struct rbd_device *rbd_dev)
5755 struct rbd_spec *parent_spec;
5756 struct parent_image_info pii = { 0 };
5759 parent_spec = rbd_spec_alloc();
5763 ret = rbd_dev_v2_parent_info(rbd_dev, &pii);
5767 if (pii.pool_id == CEPH_NOPOOL || !pii.has_overlap)
5768 goto out; /* No parent? No problem. */
5770 /* The ceph file layout needs to fit pool id in 32 bits */
5773 if (pii.pool_id > (u64)U32_MAX) {
5774 rbd_warn(NULL, "parent pool id too large (%llu > %u)",
5775 (unsigned long long)pii.pool_id, U32_MAX);
5780 * The parent won't change except when the clone is flattened,
5781 * so we only need to record the parent image spec once.
5783 parent_spec->pool_id = pii.pool_id;
5784 if (pii.pool_ns && *pii.pool_ns) {
5785 parent_spec->pool_ns = pii.pool_ns;
5788 parent_spec->image_id = pii.image_id;
5789 pii.image_id = NULL;
5790 parent_spec->snap_id = pii.snap_id;
5792 rbd_assert(!rbd_dev->parent_spec);
5793 rbd_dev->parent_spec = parent_spec;
5794 parent_spec = NULL; /* rbd_dev now owns this */
5797 * Record the parent overlap. If it's zero, issue a warning as
5798 * we will proceed as if there is no parent.
5801 rbd_warn(rbd_dev, "clone is standalone (overlap 0)");
5802 rbd_dev->parent_overlap = pii.overlap;
5807 rbd_parent_info_cleanup(&pii);
5808 rbd_spec_put(parent_spec);
5812 static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev,
5813 u64 *stripe_unit, u64 *stripe_count)
5817 __le64 stripe_count;
5818 } __attribute__ ((packed)) striping_info_buf = { 0 };
5819 size_t size = sizeof (striping_info_buf);
5822 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5823 &rbd_dev->header_oloc, "get_stripe_unit_count",
5824 NULL, 0, &striping_info_buf, size);
5825 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5831 *stripe_unit = le64_to_cpu(striping_info_buf.stripe_unit);
5832 *stripe_count = le64_to_cpu(striping_info_buf.stripe_count);
5833 dout(" stripe_unit = %llu stripe_count = %llu\n", *stripe_unit,
5839 static int rbd_dev_v2_data_pool(struct rbd_device *rbd_dev, s64 *data_pool_id)
5841 __le64 data_pool_buf;
5844 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5845 &rbd_dev->header_oloc, "get_data_pool",
5846 NULL, 0, &data_pool_buf,
5847 sizeof(data_pool_buf));
5848 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5851 if (ret < sizeof(data_pool_buf))
5854 *data_pool_id = le64_to_cpu(data_pool_buf);
5855 dout(" data_pool_id = %lld\n", *data_pool_id);
5856 WARN_ON(*data_pool_id == CEPH_NOPOOL);
5861 static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
5863 CEPH_DEFINE_OID_ONSTACK(oid);
5864 size_t image_id_size;
5869 void *reply_buf = NULL;
5871 char *image_name = NULL;
5874 rbd_assert(!rbd_dev->spec->image_name);
5876 len = strlen(rbd_dev->spec->image_id);
5877 image_id_size = sizeof (__le32) + len;
5878 image_id = kmalloc(image_id_size, GFP_KERNEL);
5883 end = image_id + image_id_size;
5884 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len);
5886 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
5887 reply_buf = kmalloc(size, GFP_KERNEL);
5891 ceph_oid_printf(&oid, "%s", RBD_DIRECTORY);
5892 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
5893 "dir_get_name", image_id, image_id_size,
5898 end = reply_buf + ret;
5900 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
5901 if (IS_ERR(image_name))
5904 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
5912 static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5914 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5915 const char *snap_name;
5918 /* Skip over names until we find the one we are looking for */
5920 snap_name = rbd_dev->header.snap_names;
5921 while (which < snapc->num_snaps) {
5922 if (!strcmp(name, snap_name))
5923 return snapc->snaps[which];
5924 snap_name += strlen(snap_name) + 1;
5930 static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5932 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5937 for (which = 0; !found && which < snapc->num_snaps; which++) {
5938 const char *snap_name;
5940 snap_id = snapc->snaps[which];
5941 snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
5942 if (IS_ERR(snap_name)) {
5943 /* ignore no-longer existing snapshots */
5944 if (PTR_ERR(snap_name) == -ENOENT)
5949 found = !strcmp(name, snap_name);
5952 return found ? snap_id : CEPH_NOSNAP;
5956 * Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if
5957 * no snapshot by that name is found, or if an error occurs.
5959 static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5961 if (rbd_dev->image_format == 1)
5962 return rbd_v1_snap_id_by_name(rbd_dev, name);
5964 return rbd_v2_snap_id_by_name(rbd_dev, name);
5968 * An image being mapped will have everything but the snap id.
5970 static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev)
5972 struct rbd_spec *spec = rbd_dev->spec;
5974 rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name);
5975 rbd_assert(spec->image_id && spec->image_name);
5976 rbd_assert(spec->snap_name);
5978 if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) {
5981 snap_id = rbd_snap_id_by_name(rbd_dev, spec->snap_name);
5982 if (snap_id == CEPH_NOSNAP)
5985 spec->snap_id = snap_id;
5987 spec->snap_id = CEPH_NOSNAP;
5994 * A parent image will have all ids but none of the names.
5996 * All names in an rbd spec are dynamically allocated. It's OK if we
5997 * can't figure out the name for an image id.
5999 static int rbd_spec_fill_names(struct rbd_device *rbd_dev)
6001 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
6002 struct rbd_spec *spec = rbd_dev->spec;
6003 const char *pool_name;
6004 const char *image_name;
6005 const char *snap_name;
6008 rbd_assert(spec->pool_id != CEPH_NOPOOL);
6009 rbd_assert(spec->image_id);
6010 rbd_assert(spec->snap_id != CEPH_NOSNAP);
6012 /* Get the pool name; we have to make our own copy of this */
6014 pool_name = ceph_pg_pool_name_by_id(osdc->osdmap, spec->pool_id);
6016 rbd_warn(rbd_dev, "no pool with id %llu", spec->pool_id);
6019 pool_name = kstrdup(pool_name, GFP_KERNEL);
6023 /* Fetch the image name; tolerate failure here */
6025 image_name = rbd_dev_image_name(rbd_dev);
6027 rbd_warn(rbd_dev, "unable to get image name");
6029 /* Fetch the snapshot name */
6031 snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
6032 if (IS_ERR(snap_name)) {
6033 ret = PTR_ERR(snap_name);
6037 spec->pool_name = pool_name;
6038 spec->image_name = image_name;
6039 spec->snap_name = snap_name;
6049 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev,
6050 struct ceph_snap_context **psnapc)
6059 struct ceph_snap_context *snapc;
6063 * We'll need room for the seq value (maximum snapshot id),
6064 * snapshot count, and array of that many snapshot ids.
6065 * For now we have a fixed upper limit on the number we're
6066 * prepared to receive.
6068 size = sizeof (__le64) + sizeof (__le32) +
6069 RBD_MAX_SNAP_COUNT * sizeof (__le64);
6070 reply_buf = kzalloc(size, GFP_KERNEL);
6074 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6075 &rbd_dev->header_oloc, "get_snapcontext",
6076 NULL, 0, reply_buf, size);
6077 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6082 end = reply_buf + ret;
6084 ceph_decode_64_safe(&p, end, seq, out);
6085 ceph_decode_32_safe(&p, end, snap_count, out);
6088 * Make sure the reported number of snapshot ids wouldn't go
6089 * beyond the end of our buffer. But before checking that,
6090 * make sure the computed size of the snapshot context we
6091 * allocate is representable in a size_t.
6093 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
6098 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
6102 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
6108 for (i = 0; i < snap_count; i++)
6109 snapc->snaps[i] = ceph_decode_64(&p);
6112 dout(" snap context seq = %llu, snap_count = %u\n",
6113 (unsigned long long)seq, (unsigned int)snap_count);
6120 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
6131 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
6132 reply_buf = kmalloc(size, GFP_KERNEL);
6134 return ERR_PTR(-ENOMEM);
6136 snapid = cpu_to_le64(snap_id);
6137 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6138 &rbd_dev->header_oloc, "get_snapshot_name",
6139 &snapid, sizeof(snapid), reply_buf, size);
6140 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6142 snap_name = ERR_PTR(ret);
6147 end = reply_buf + ret;
6148 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
6149 if (IS_ERR(snap_name))
6152 dout(" snap_id 0x%016llx snap_name = %s\n",
6153 (unsigned long long)snap_id, snap_name);
6160 static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev,
6161 struct rbd_image_header *header,
6166 ret = _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
6167 first_time ? &header->obj_order : NULL,
6168 &header->image_size);
6173 ret = rbd_dev_v2_header_onetime(rbd_dev, header);
6178 ret = rbd_dev_v2_snap_context(rbd_dev, &header->snapc);
6185 static int rbd_dev_header_info(struct rbd_device *rbd_dev,
6186 struct rbd_image_header *header,
6189 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6190 rbd_assert(!header->object_prefix && !header->snapc);
6192 if (rbd_dev->image_format == 1)
6193 return rbd_dev_v1_header_info(rbd_dev, header, first_time);
6195 return rbd_dev_v2_header_info(rbd_dev, header, first_time);
6199 * Skips over white space at *buf, and updates *buf to point to the
6200 * first found non-space character (if any). Returns the length of
6201 * the token (string of non-white space characters) found. Note
6202 * that *buf must be terminated with '\0'.
6204 static inline size_t next_token(const char **buf)
6207 * These are the characters that produce nonzero for
6208 * isspace() in the "C" and "POSIX" locales.
6210 static const char spaces[] = " \f\n\r\t\v";
6212 *buf += strspn(*buf, spaces); /* Find start of token */
6214 return strcspn(*buf, spaces); /* Return token length */
6218 * Finds the next token in *buf, dynamically allocates a buffer big
6219 * enough to hold a copy of it, and copies the token into the new
6220 * buffer. The copy is guaranteed to be terminated with '\0'. Note
6221 * that a duplicate buffer is created even for a zero-length token.
6223 * Returns a pointer to the newly-allocated duplicate, or a null
6224 * pointer if memory for the duplicate was not available. If
6225 * the lenp argument is a non-null pointer, the length of the token
6226 * (not including the '\0') is returned in *lenp.
6228 * If successful, the *buf pointer will be updated to point beyond
6229 * the end of the found token.
6231 * Note: uses GFP_KERNEL for allocation.
6233 static inline char *dup_token(const char **buf, size_t *lenp)
6238 len = next_token(buf);
6239 dup = kmemdup(*buf, len + 1, GFP_KERNEL);
6242 *(dup + len) = '\0';
6251 static int rbd_parse_param(struct fs_parameter *param,
6252 struct rbd_parse_opts_ctx *pctx)
6254 struct rbd_options *opt = pctx->opts;
6255 struct fs_parse_result result;
6256 struct p_log log = {.prefix = "rbd"};
6259 ret = ceph_parse_param(param, pctx->copts, NULL);
6260 if (ret != -ENOPARAM)
6263 token = __fs_parse(&log, rbd_parameters, param, &result);
6264 dout("%s fs_parse '%s' token %d\n", __func__, param->key, token);
6266 if (token == -ENOPARAM)
6267 return inval_plog(&log, "Unknown parameter '%s'",
6273 case Opt_queue_depth:
6274 if (result.uint_32 < 1)
6276 opt->queue_depth = result.uint_32;
6278 case Opt_alloc_size:
6279 if (result.uint_32 < SECTOR_SIZE)
6281 if (!is_power_of_2(result.uint_32))
6282 return inval_plog(&log, "alloc_size must be a power of 2");
6283 opt->alloc_size = result.uint_32;
6285 case Opt_lock_timeout:
6286 /* 0 is "wait forever" (i.e. infinite timeout) */
6287 if (result.uint_32 > INT_MAX / 1000)
6289 opt->lock_timeout = msecs_to_jiffies(result.uint_32 * 1000);
6292 kfree(pctx->spec->pool_ns);
6293 pctx->spec->pool_ns = param->string;
6294 param->string = NULL;
6296 case Opt_compression_hint:
6297 switch (result.uint_32) {
6298 case Opt_compression_hint_none:
6299 opt->alloc_hint_flags &=
6300 ~(CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE |
6301 CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE);
6303 case Opt_compression_hint_compressible:
6304 opt->alloc_hint_flags |=
6305 CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6306 opt->alloc_hint_flags &=
6307 ~CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6309 case Opt_compression_hint_incompressible:
6310 opt->alloc_hint_flags |=
6311 CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6312 opt->alloc_hint_flags &=
6313 ~CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6320 opt->read_only = true;
6322 case Opt_read_write:
6323 opt->read_only = false;
6325 case Opt_lock_on_read:
6326 opt->lock_on_read = true;
6329 opt->exclusive = true;
6341 return inval_plog(&log, "%s out of range", param->key);
6345 * This duplicates most of generic_parse_monolithic(), untying it from
6346 * fs_context and skipping standard superblock and security options.
6348 static int rbd_parse_options(char *options, struct rbd_parse_opts_ctx *pctx)
6353 dout("%s '%s'\n", __func__, options);
6354 while ((key = strsep(&options, ",")) != NULL) {
6356 struct fs_parameter param = {
6358 .type = fs_value_is_flag,
6360 char *value = strchr(key, '=');
6367 v_len = strlen(value);
6368 param.string = kmemdup_nul(value, v_len,
6372 param.type = fs_value_is_string;
6376 ret = rbd_parse_param(¶m, pctx);
6377 kfree(param.string);
6387 * Parse the options provided for an "rbd add" (i.e., rbd image
6388 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
6389 * and the data written is passed here via a NUL-terminated buffer.
6390 * Returns 0 if successful or an error code otherwise.
6392 * The information extracted from these options is recorded in
6393 * the other parameters which return dynamically-allocated
6396 * The address of a pointer that will refer to a ceph options
6397 * structure. Caller must release the returned pointer using
6398 * ceph_destroy_options() when it is no longer needed.
6400 * Address of an rbd options pointer. Fully initialized by
6401 * this function; caller must release with kfree().
6403 * Address of an rbd image specification pointer. Fully
6404 * initialized by this function based on parsed options.
6405 * Caller must release with rbd_spec_put().
6407 * The options passed take this form:
6408 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
6411 * A comma-separated list of one or more monitor addresses.
6412 * A monitor address is an ip address, optionally followed
6413 * by a port number (separated by a colon).
6414 * I.e.: ip1[:port1][,ip2[:port2]...]
6416 * A comma-separated list of ceph and/or rbd options.
6418 * The name of the rados pool containing the rbd image.
6420 * The name of the image in that pool to map.
6422 * An optional snapshot id. If provided, the mapping will
6423 * present data from the image at the time that snapshot was
6424 * created. The image head is used if no snapshot id is
6425 * provided. Snapshot mappings are always read-only.
6427 static int rbd_add_parse_args(const char *buf,
6428 struct ceph_options **ceph_opts,
6429 struct rbd_options **opts,
6430 struct rbd_spec **rbd_spec)
6434 const char *mon_addrs;
6436 size_t mon_addrs_size;
6437 struct rbd_parse_opts_ctx pctx = { 0 };
6440 /* The first four tokens are required */
6442 len = next_token(&buf);
6444 rbd_warn(NULL, "no monitor address(es) provided");
6448 mon_addrs_size = len;
6452 options = dup_token(&buf, NULL);
6456 rbd_warn(NULL, "no options provided");
6460 pctx.spec = rbd_spec_alloc();
6464 pctx.spec->pool_name = dup_token(&buf, NULL);
6465 if (!pctx.spec->pool_name)
6467 if (!*pctx.spec->pool_name) {
6468 rbd_warn(NULL, "no pool name provided");
6472 pctx.spec->image_name = dup_token(&buf, NULL);
6473 if (!pctx.spec->image_name)
6475 if (!*pctx.spec->image_name) {
6476 rbd_warn(NULL, "no image name provided");
6481 * Snapshot name is optional; default is to use "-"
6482 * (indicating the head/no snapshot).
6484 len = next_token(&buf);
6486 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
6487 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
6488 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
6489 ret = -ENAMETOOLONG;
6492 snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
6495 *(snap_name + len) = '\0';
6496 pctx.spec->snap_name = snap_name;
6498 pctx.copts = ceph_alloc_options();
6502 /* Initialize all rbd options to the defaults */
6504 pctx.opts = kzalloc(sizeof(*pctx.opts), GFP_KERNEL);
6508 pctx.opts->read_only = RBD_READ_ONLY_DEFAULT;
6509 pctx.opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
6510 pctx.opts->alloc_size = RBD_ALLOC_SIZE_DEFAULT;
6511 pctx.opts->lock_timeout = RBD_LOCK_TIMEOUT_DEFAULT;
6512 pctx.opts->lock_on_read = RBD_LOCK_ON_READ_DEFAULT;
6513 pctx.opts->exclusive = RBD_EXCLUSIVE_DEFAULT;
6514 pctx.opts->trim = RBD_TRIM_DEFAULT;
6516 ret = ceph_parse_mon_ips(mon_addrs, mon_addrs_size, pctx.copts, NULL,
6521 ret = rbd_parse_options(options, &pctx);
6525 *ceph_opts = pctx.copts;
6527 *rbd_spec = pctx.spec;
6535 ceph_destroy_options(pctx.copts);
6536 rbd_spec_put(pctx.spec);
6541 static void rbd_dev_image_unlock(struct rbd_device *rbd_dev)
6543 down_write(&rbd_dev->lock_rwsem);
6544 if (__rbd_is_lock_owner(rbd_dev))
6545 __rbd_release_lock(rbd_dev);
6546 up_write(&rbd_dev->lock_rwsem);
6550 * If the wait is interrupted, an error is returned even if the lock
6551 * was successfully acquired. rbd_dev_image_unlock() will release it
6554 static int rbd_add_acquire_lock(struct rbd_device *rbd_dev)
6558 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK)) {
6559 if (!rbd_dev->opts->exclusive && !rbd_dev->opts->lock_on_read)
6562 rbd_warn(rbd_dev, "exclusive-lock feature is not enabled");
6566 if (rbd_is_ro(rbd_dev))
6569 rbd_assert(!rbd_is_lock_owner(rbd_dev));
6570 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
6571 ret = wait_for_completion_killable_timeout(&rbd_dev->acquire_wait,
6572 ceph_timeout_jiffies(rbd_dev->opts->lock_timeout));
6574 ret = rbd_dev->acquire_err;
6576 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
6580 rbd_warn(rbd_dev, "failed to acquire lock: %ld", ret);
6586 * The lock may have been released by now, unless automatic lock
6587 * transitions are disabled.
6589 rbd_assert(!rbd_dev->opts->exclusive || rbd_is_lock_owner(rbd_dev));
6594 * An rbd format 2 image has a unique identifier, distinct from the
6595 * name given to it by the user. Internally, that identifier is
6596 * what's used to specify the names of objects related to the image.
6598 * A special "rbd id" object is used to map an rbd image name to its
6599 * id. If that object doesn't exist, then there is no v2 rbd image
6600 * with the supplied name.
6602 * This function will record the given rbd_dev's image_id field if
6603 * it can be determined, and in that case will return 0. If any
6604 * errors occur a negative errno will be returned and the rbd_dev's
6605 * image_id field will be unchanged (and should be NULL).
6607 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
6611 CEPH_DEFINE_OID_ONSTACK(oid);
6616 * When probing a parent image, the image id is already
6617 * known (and the image name likely is not). There's no
6618 * need to fetch the image id again in this case. We
6619 * do still need to set the image format though.
6621 if (rbd_dev->spec->image_id) {
6622 rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
6628 * First, see if the format 2 image id file exists, and if
6629 * so, get the image's persistent id from it.
6631 ret = ceph_oid_aprintf(&oid, GFP_KERNEL, "%s%s", RBD_ID_PREFIX,
6632 rbd_dev->spec->image_name);
6636 dout("rbd id object name is %s\n", oid.name);
6638 /* Response will be an encoded string, which includes a length */
6639 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
6640 response = kzalloc(size, GFP_NOIO);
6646 /* If it doesn't exist we'll assume it's a format 1 image */
6648 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
6651 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6652 if (ret == -ENOENT) {
6653 image_id = kstrdup("", GFP_KERNEL);
6654 ret = image_id ? 0 : -ENOMEM;
6656 rbd_dev->image_format = 1;
6657 } else if (ret >= 0) {
6660 image_id = ceph_extract_encoded_string(&p, p + ret,
6662 ret = PTR_ERR_OR_ZERO(image_id);
6664 rbd_dev->image_format = 2;
6668 rbd_dev->spec->image_id = image_id;
6669 dout("image_id is %s\n", image_id);
6673 ceph_oid_destroy(&oid);
6678 * Undo whatever state changes are made by v1 or v2 header info
6681 static void rbd_dev_unprobe(struct rbd_device *rbd_dev)
6683 rbd_dev_parent_put(rbd_dev);
6684 rbd_object_map_free(rbd_dev);
6685 rbd_dev_mapping_clear(rbd_dev);
6687 /* Free dynamic fields from the header, then zero it out */
6689 rbd_image_header_cleanup(&rbd_dev->header);
6692 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev,
6693 struct rbd_image_header *header)
6697 ret = rbd_dev_v2_object_prefix(rbd_dev, &header->object_prefix);
6702 * Get the and check features for the image. Currently the
6703 * features are assumed to never change.
6705 ret = _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
6706 rbd_is_ro(rbd_dev), &header->features);
6710 /* If the image supports fancy striping, get its parameters */
6712 if (header->features & RBD_FEATURE_STRIPINGV2) {
6713 ret = rbd_dev_v2_striping_info(rbd_dev, &header->stripe_unit,
6714 &header->stripe_count);
6719 if (header->features & RBD_FEATURE_DATA_POOL) {
6720 ret = rbd_dev_v2_data_pool(rbd_dev, &header->data_pool_id);
6729 * @depth is rbd_dev_image_probe() -> rbd_dev_probe_parent() ->
6730 * rbd_dev_image_probe() recursion depth, which means it's also the
6731 * length of the already discovered part of the parent chain.
6733 static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
6735 struct rbd_device *parent = NULL;
6738 if (!rbd_dev->parent_spec)
6741 if (++depth > RBD_MAX_PARENT_CHAIN_LEN) {
6742 pr_info("parent chain is too long (%d)\n", depth);
6747 parent = __rbd_dev_create(rbd_dev->parent_spec);
6754 * Images related by parent/child relationships always share
6755 * rbd_client and spec/parent_spec, so bump their refcounts.
6757 parent->rbd_client = __rbd_get_client(rbd_dev->rbd_client);
6758 parent->spec = rbd_spec_get(rbd_dev->parent_spec);
6760 __set_bit(RBD_DEV_FLAG_READONLY, &parent->flags);
6762 ret = rbd_dev_image_probe(parent, depth);
6766 rbd_dev->parent = parent;
6767 atomic_set(&rbd_dev->parent_ref, 1);
6771 rbd_dev_unparent(rbd_dev);
6772 rbd_dev_destroy(parent);
6776 static void rbd_dev_device_release(struct rbd_device *rbd_dev)
6778 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6779 rbd_free_disk(rbd_dev);
6781 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6785 * rbd_dev->header_rwsem must be locked for write and will be unlocked
6788 static int rbd_dev_device_setup(struct rbd_device *rbd_dev)
6792 /* Record our major and minor device numbers. */
6794 if (!single_major) {
6795 ret = register_blkdev(0, rbd_dev->name);
6797 goto err_out_unlock;
6799 rbd_dev->major = ret;
6802 rbd_dev->major = rbd_major;
6803 rbd_dev->minor = rbd_dev_id_to_minor(rbd_dev->dev_id);
6806 /* Set up the blkdev mapping. */
6808 ret = rbd_init_disk(rbd_dev);
6810 goto err_out_blkdev;
6812 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
6813 set_disk_ro(rbd_dev->disk, rbd_is_ro(rbd_dev));
6815 ret = dev_set_name(&rbd_dev->dev, "%d", rbd_dev->dev_id);
6819 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6820 up_write(&rbd_dev->header_rwsem);
6824 rbd_free_disk(rbd_dev);
6827 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6829 up_write(&rbd_dev->header_rwsem);
6833 static int rbd_dev_header_name(struct rbd_device *rbd_dev)
6835 struct rbd_spec *spec = rbd_dev->spec;
6838 /* Record the header object name for this rbd image. */
6840 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6841 if (rbd_dev->image_format == 1)
6842 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6843 spec->image_name, RBD_SUFFIX);
6845 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6846 RBD_HEADER_PREFIX, spec->image_id);
6851 static void rbd_print_dne(struct rbd_device *rbd_dev, bool is_snap)
6854 pr_info("image %s/%s%s%s does not exist\n",
6855 rbd_dev->spec->pool_name,
6856 rbd_dev->spec->pool_ns ?: "",
6857 rbd_dev->spec->pool_ns ? "/" : "",
6858 rbd_dev->spec->image_name);
6860 pr_info("snap %s/%s%s%s@%s does not exist\n",
6861 rbd_dev->spec->pool_name,
6862 rbd_dev->spec->pool_ns ?: "",
6863 rbd_dev->spec->pool_ns ? "/" : "",
6864 rbd_dev->spec->image_name,
6865 rbd_dev->spec->snap_name);
6869 static void rbd_dev_image_release(struct rbd_device *rbd_dev)
6871 if (!rbd_is_ro(rbd_dev))
6872 rbd_unregister_watch(rbd_dev);
6874 rbd_dev_unprobe(rbd_dev);
6875 rbd_dev->image_format = 0;
6876 kfree(rbd_dev->spec->image_id);
6877 rbd_dev->spec->image_id = NULL;
6881 * Probe for the existence of the header object for the given rbd
6882 * device. If this image is the one being mapped (i.e., not a
6883 * parent), initiate a watch on its header object before using that
6884 * object to get detailed information about the rbd image.
6886 * On success, returns with header_rwsem held for write if called
6889 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
6891 bool need_watch = !rbd_is_ro(rbd_dev);
6895 * Get the id from the image id object. Unless there's an
6896 * error, rbd_dev->spec->image_id will be filled in with
6897 * a dynamically-allocated string, and rbd_dev->image_format
6898 * will be set to either 1 or 2.
6900 ret = rbd_dev_image_id(rbd_dev);
6904 ret = rbd_dev_header_name(rbd_dev);
6906 goto err_out_format;
6909 ret = rbd_register_watch(rbd_dev);
6912 rbd_print_dne(rbd_dev, false);
6913 goto err_out_format;
6918 down_write(&rbd_dev->header_rwsem);
6920 ret = rbd_dev_header_info(rbd_dev, &rbd_dev->header, true);
6922 if (ret == -ENOENT && !need_watch)
6923 rbd_print_dne(rbd_dev, false);
6927 rbd_init_layout(rbd_dev);
6930 * If this image is the one being mapped, we have pool name and
6931 * id, image name and id, and snap name - need to fill snap id.
6932 * Otherwise this is a parent image, identified by pool, image
6933 * and snap ids - need to fill in names for those ids.
6936 ret = rbd_spec_fill_snap_id(rbd_dev);
6938 ret = rbd_spec_fill_names(rbd_dev);
6941 rbd_print_dne(rbd_dev, true);
6945 ret = rbd_dev_mapping_set(rbd_dev);
6949 if (rbd_is_snap(rbd_dev) &&
6950 (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)) {
6951 ret = rbd_object_map_load(rbd_dev);
6956 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
6957 ret = rbd_dev_setup_parent(rbd_dev);
6962 ret = rbd_dev_probe_parent(rbd_dev, depth);
6966 dout("discovered format %u image, header name is %s\n",
6967 rbd_dev->image_format, rbd_dev->header_oid.name);
6972 up_write(&rbd_dev->header_rwsem);
6974 rbd_unregister_watch(rbd_dev);
6975 rbd_dev_unprobe(rbd_dev);
6977 rbd_dev->image_format = 0;
6978 kfree(rbd_dev->spec->image_id);
6979 rbd_dev->spec->image_id = NULL;
6983 static void rbd_dev_update_header(struct rbd_device *rbd_dev,
6984 struct rbd_image_header *header)
6986 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6987 rbd_assert(rbd_dev->header.object_prefix); /* !first_time */
6989 if (rbd_dev->header.image_size != header->image_size) {
6990 rbd_dev->header.image_size = header->image_size;
6992 if (!rbd_is_snap(rbd_dev)) {
6993 rbd_dev->mapping.size = header->image_size;
6994 rbd_dev_update_size(rbd_dev);
6998 ceph_put_snap_context(rbd_dev->header.snapc);
6999 rbd_dev->header.snapc = header->snapc;
7000 header->snapc = NULL;
7002 if (rbd_dev->image_format == 1) {
7003 kfree(rbd_dev->header.snap_names);
7004 rbd_dev->header.snap_names = header->snap_names;
7005 header->snap_names = NULL;
7007 kfree(rbd_dev->header.snap_sizes);
7008 rbd_dev->header.snap_sizes = header->snap_sizes;
7009 header->snap_sizes = NULL;
7013 static void rbd_dev_update_parent(struct rbd_device *rbd_dev,
7014 struct parent_image_info *pii)
7016 if (pii->pool_id == CEPH_NOPOOL || !pii->has_overlap) {
7018 * Either the parent never existed, or we have
7019 * record of it but the image got flattened so it no
7020 * longer has a parent. When the parent of a
7021 * layered image disappears we immediately set the
7022 * overlap to 0. The effect of this is that all new
7023 * requests will be treated as if the image had no
7026 * If !pii.has_overlap, the parent image spec is not
7027 * applicable. It's there to avoid duplication in each
7030 if (rbd_dev->parent_overlap) {
7031 rbd_dev->parent_overlap = 0;
7032 rbd_dev_parent_put(rbd_dev);
7033 pr_info("%s: clone has been flattened\n",
7034 rbd_dev->disk->disk_name);
7037 rbd_assert(rbd_dev->parent_spec);
7040 * Update the parent overlap. If it became zero, issue
7041 * a warning as we will proceed as if there is no parent.
7043 if (!pii->overlap && rbd_dev->parent_overlap)
7045 "clone has become standalone (overlap 0)");
7046 rbd_dev->parent_overlap = pii->overlap;
7050 static int rbd_dev_refresh(struct rbd_device *rbd_dev)
7052 struct rbd_image_header header = { 0 };
7053 struct parent_image_info pii = { 0 };
7056 dout("%s rbd_dev %p\n", __func__, rbd_dev);
7058 ret = rbd_dev_header_info(rbd_dev, &header, false);
7063 * If there is a parent, see if it has disappeared due to the
7064 * mapped image getting flattened.
7066 if (rbd_dev->parent) {
7067 ret = rbd_dev_v2_parent_info(rbd_dev, &pii);
7072 down_write(&rbd_dev->header_rwsem);
7073 rbd_dev_update_header(rbd_dev, &header);
7074 if (rbd_dev->parent)
7075 rbd_dev_update_parent(rbd_dev, &pii);
7076 up_write(&rbd_dev->header_rwsem);
7079 rbd_parent_info_cleanup(&pii);
7080 rbd_image_header_cleanup(&header);
7084 static ssize_t do_rbd_add(const char *buf, size_t count)
7086 struct rbd_device *rbd_dev = NULL;
7087 struct ceph_options *ceph_opts = NULL;
7088 struct rbd_options *rbd_opts = NULL;
7089 struct rbd_spec *spec = NULL;
7090 struct rbd_client *rbdc;
7093 if (!capable(CAP_SYS_ADMIN))
7096 if (!try_module_get(THIS_MODULE))
7099 /* parse add command */
7100 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
7104 rbdc = rbd_get_client(ceph_opts);
7111 rc = ceph_pg_poolid_by_name(rbdc->client->osdc.osdmap, spec->pool_name);
7114 pr_info("pool %s does not exist\n", spec->pool_name);
7115 goto err_out_client;
7117 spec->pool_id = (u64)rc;
7119 rbd_dev = rbd_dev_create(rbdc, spec, rbd_opts);
7122 goto err_out_client;
7124 rbdc = NULL; /* rbd_dev now owns this */
7125 spec = NULL; /* rbd_dev now owns this */
7126 rbd_opts = NULL; /* rbd_dev now owns this */
7128 /* if we are mapping a snapshot it will be a read-only mapping */
7129 if (rbd_dev->opts->read_only ||
7130 strcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME))
7131 __set_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
7133 rbd_dev->config_info = kstrdup(buf, GFP_KERNEL);
7134 if (!rbd_dev->config_info) {
7136 goto err_out_rbd_dev;
7139 rc = rbd_dev_image_probe(rbd_dev, 0);
7141 goto err_out_rbd_dev;
7143 if (rbd_dev->opts->alloc_size > rbd_dev->layout.object_size) {
7144 rbd_warn(rbd_dev, "alloc_size adjusted to %u",
7145 rbd_dev->layout.object_size);
7146 rbd_dev->opts->alloc_size = rbd_dev->layout.object_size;
7149 rc = rbd_dev_device_setup(rbd_dev);
7151 goto err_out_image_probe;
7153 rc = rbd_add_acquire_lock(rbd_dev);
7155 goto err_out_image_lock;
7157 /* Everything's ready. Announce the disk to the world. */
7159 rc = device_add(&rbd_dev->dev);
7161 goto err_out_image_lock;
7163 rc = device_add_disk(&rbd_dev->dev, rbd_dev->disk, NULL);
7165 goto err_out_cleanup_disk;
7167 spin_lock(&rbd_dev_list_lock);
7168 list_add_tail(&rbd_dev->node, &rbd_dev_list);
7169 spin_unlock(&rbd_dev_list_lock);
7171 pr_info("%s: capacity %llu features 0x%llx\n", rbd_dev->disk->disk_name,
7172 (unsigned long long)get_capacity(rbd_dev->disk) << SECTOR_SHIFT,
7173 rbd_dev->header.features);
7176 module_put(THIS_MODULE);
7179 err_out_cleanup_disk:
7180 rbd_free_disk(rbd_dev);
7182 rbd_dev_image_unlock(rbd_dev);
7183 rbd_dev_device_release(rbd_dev);
7184 err_out_image_probe:
7185 rbd_dev_image_release(rbd_dev);
7187 rbd_dev_destroy(rbd_dev);
7189 rbd_put_client(rbdc);
7196 static ssize_t add_store(const struct bus_type *bus, const char *buf, size_t count)
7201 return do_rbd_add(buf, count);
7204 static ssize_t add_single_major_store(const struct bus_type *bus, const char *buf,
7207 return do_rbd_add(buf, count);
7210 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev)
7212 while (rbd_dev->parent) {
7213 struct rbd_device *first = rbd_dev;
7214 struct rbd_device *second = first->parent;
7215 struct rbd_device *third;
7218 * Follow to the parent with no grandparent and
7221 while (second && (third = second->parent)) {
7226 rbd_dev_image_release(second);
7227 rbd_dev_destroy(second);
7228 first->parent = NULL;
7229 first->parent_overlap = 0;
7231 rbd_assert(first->parent_spec);
7232 rbd_spec_put(first->parent_spec);
7233 first->parent_spec = NULL;
7237 static ssize_t do_rbd_remove(const char *buf, size_t count)
7239 struct rbd_device *rbd_dev = NULL;
7245 if (!capable(CAP_SYS_ADMIN))
7250 sscanf(buf, "%d %5s", &dev_id, opt_buf);
7252 pr_err("dev_id out of range\n");
7255 if (opt_buf[0] != '\0') {
7256 if (!strcmp(opt_buf, "force")) {
7259 pr_err("bad remove option at '%s'\n", opt_buf);
7265 spin_lock(&rbd_dev_list_lock);
7266 list_for_each_entry(rbd_dev, &rbd_dev_list, node) {
7267 if (rbd_dev->dev_id == dev_id) {
7273 spin_lock_irq(&rbd_dev->lock);
7274 if (rbd_dev->open_count && !force)
7276 else if (test_and_set_bit(RBD_DEV_FLAG_REMOVING,
7279 spin_unlock_irq(&rbd_dev->lock);
7281 spin_unlock(&rbd_dev_list_lock);
7287 * Prevent new IO from being queued and wait for existing
7288 * IO to complete/fail.
7290 blk_mq_freeze_queue(rbd_dev->disk->queue);
7291 blk_mark_disk_dead(rbd_dev->disk);
7294 del_gendisk(rbd_dev->disk);
7295 spin_lock(&rbd_dev_list_lock);
7296 list_del_init(&rbd_dev->node);
7297 spin_unlock(&rbd_dev_list_lock);
7298 device_del(&rbd_dev->dev);
7300 rbd_dev_image_unlock(rbd_dev);
7301 rbd_dev_device_release(rbd_dev);
7302 rbd_dev_image_release(rbd_dev);
7303 rbd_dev_destroy(rbd_dev);
7307 static ssize_t remove_store(const struct bus_type *bus, const char *buf, size_t count)
7312 return do_rbd_remove(buf, count);
7315 static ssize_t remove_single_major_store(const struct bus_type *bus, const char *buf,
7318 return do_rbd_remove(buf, count);
7322 * create control files in sysfs
7325 static int __init rbd_sysfs_init(void)
7329 ret = device_register(&rbd_root_dev);
7331 put_device(&rbd_root_dev);
7335 ret = bus_register(&rbd_bus_type);
7337 device_unregister(&rbd_root_dev);
7342 static void __exit rbd_sysfs_cleanup(void)
7344 bus_unregister(&rbd_bus_type);
7345 device_unregister(&rbd_root_dev);
7348 static int __init rbd_slab_init(void)
7350 rbd_assert(!rbd_img_request_cache);
7351 rbd_img_request_cache = KMEM_CACHE(rbd_img_request, 0);
7352 if (!rbd_img_request_cache)
7355 rbd_assert(!rbd_obj_request_cache);
7356 rbd_obj_request_cache = KMEM_CACHE(rbd_obj_request, 0);
7357 if (!rbd_obj_request_cache)
7363 kmem_cache_destroy(rbd_img_request_cache);
7364 rbd_img_request_cache = NULL;
7368 static void rbd_slab_exit(void)
7370 rbd_assert(rbd_obj_request_cache);
7371 kmem_cache_destroy(rbd_obj_request_cache);
7372 rbd_obj_request_cache = NULL;
7374 rbd_assert(rbd_img_request_cache);
7375 kmem_cache_destroy(rbd_img_request_cache);
7376 rbd_img_request_cache = NULL;
7379 static int __init rbd_init(void)
7383 if (!libceph_compatible(NULL)) {
7384 rbd_warn(NULL, "libceph incompatibility (quitting)");
7388 rc = rbd_slab_init();
7393 * The number of active work items is limited by the number of
7394 * rbd devices * queue depth, so leave @max_active at default.
7396 rbd_wq = alloc_workqueue(RBD_DRV_NAME, WQ_MEM_RECLAIM, 0);
7403 rbd_major = register_blkdev(0, RBD_DRV_NAME);
7404 if (rbd_major < 0) {
7410 rc = rbd_sysfs_init();
7412 goto err_out_blkdev;
7415 pr_info("loaded (major %d)\n", rbd_major);
7417 pr_info("loaded\n");
7423 unregister_blkdev(rbd_major, RBD_DRV_NAME);
7425 destroy_workqueue(rbd_wq);
7431 static void __exit rbd_exit(void)
7433 ida_destroy(&rbd_dev_id_ida);
7434 rbd_sysfs_cleanup();
7436 unregister_blkdev(rbd_major, RBD_DRV_NAME);
7437 destroy_workqueue(rbd_wq);
7441 module_init(rbd_init);
7442 module_exit(rbd_exit);
7444 MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
7445 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
7446 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
7447 /* following authorship retained from original osdblk.c */
7448 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
7450 MODULE_DESCRIPTION("RADOS Block Device (RBD) driver");
7451 MODULE_LICENSE("GPL");