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 static int rbd_dev_header_info(struct rbd_device *rbd_dev);
637 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev);
638 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
640 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
641 u8 *order, u64 *snap_size);
642 static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev);
644 static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result);
645 static void rbd_img_handle_request(struct rbd_img_request *img_req, int result);
648 * Return true if nothing else is pending.
650 static bool pending_result_dec(struct pending_result *pending, int *result)
652 rbd_assert(pending->num_pending > 0);
654 if (*result && !pending->result)
655 pending->result = *result;
656 if (--pending->num_pending)
659 *result = pending->result;
663 static int rbd_open(struct gendisk *disk, blk_mode_t mode)
665 struct rbd_device *rbd_dev = disk->private_data;
666 bool removing = false;
668 spin_lock_irq(&rbd_dev->lock);
669 if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
672 rbd_dev->open_count++;
673 spin_unlock_irq(&rbd_dev->lock);
677 (void) get_device(&rbd_dev->dev);
682 static void rbd_release(struct gendisk *disk)
684 struct rbd_device *rbd_dev = disk->private_data;
685 unsigned long open_count_before;
687 spin_lock_irq(&rbd_dev->lock);
688 open_count_before = rbd_dev->open_count--;
689 spin_unlock_irq(&rbd_dev->lock);
690 rbd_assert(open_count_before > 0);
692 put_device(&rbd_dev->dev);
695 static const struct block_device_operations rbd_bd_ops = {
696 .owner = THIS_MODULE,
698 .release = rbd_release,
702 * Initialize an rbd client instance. Success or not, this function
703 * consumes ceph_opts. Caller holds client_mutex.
705 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
707 struct rbd_client *rbdc;
710 dout("%s:\n", __func__);
711 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
715 kref_init(&rbdc->kref);
716 INIT_LIST_HEAD(&rbdc->node);
718 rbdc->client = ceph_create_client(ceph_opts, rbdc);
719 if (IS_ERR(rbdc->client))
721 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
723 ret = ceph_open_session(rbdc->client);
727 spin_lock(&rbd_client_list_lock);
728 list_add_tail(&rbdc->node, &rbd_client_list);
729 spin_unlock(&rbd_client_list_lock);
731 dout("%s: rbdc %p\n", __func__, rbdc);
735 ceph_destroy_client(rbdc->client);
740 ceph_destroy_options(ceph_opts);
741 dout("%s: error %d\n", __func__, ret);
746 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
748 kref_get(&rbdc->kref);
754 * Find a ceph client with specific addr and configuration. If
755 * found, bump its reference count.
757 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
759 struct rbd_client *rbdc = NULL, *iter;
761 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
764 spin_lock(&rbd_client_list_lock);
765 list_for_each_entry(iter, &rbd_client_list, node) {
766 if (!ceph_compare_options(ceph_opts, iter->client)) {
767 __rbd_get_client(iter);
773 spin_unlock(&rbd_client_list_lock);
779 * (Per device) rbd map options
787 Opt_compression_hint,
788 /* string args above */
797 Opt_compression_hint_none,
798 Opt_compression_hint_compressible,
799 Opt_compression_hint_incompressible,
802 static const struct constant_table rbd_param_compression_hint[] = {
803 {"none", Opt_compression_hint_none},
804 {"compressible", Opt_compression_hint_compressible},
805 {"incompressible", Opt_compression_hint_incompressible},
809 static const struct fs_parameter_spec rbd_parameters[] = {
810 fsparam_u32 ("alloc_size", Opt_alloc_size),
811 fsparam_enum ("compression_hint", Opt_compression_hint,
812 rbd_param_compression_hint),
813 fsparam_flag ("exclusive", Opt_exclusive),
814 fsparam_flag ("lock_on_read", Opt_lock_on_read),
815 fsparam_u32 ("lock_timeout", Opt_lock_timeout),
816 fsparam_flag ("notrim", Opt_notrim),
817 fsparam_string ("_pool_ns", Opt_pool_ns),
818 fsparam_u32 ("queue_depth", Opt_queue_depth),
819 fsparam_flag ("read_only", Opt_read_only),
820 fsparam_flag ("read_write", Opt_read_write),
821 fsparam_flag ("ro", Opt_read_only),
822 fsparam_flag ("rw", Opt_read_write),
829 unsigned long lock_timeout;
835 u32 alloc_hint_flags; /* CEPH_OSD_OP_ALLOC_HINT_FLAG_* */
838 #define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_DEFAULT_RQ
839 #define RBD_ALLOC_SIZE_DEFAULT (64 * 1024)
840 #define RBD_LOCK_TIMEOUT_DEFAULT 0 /* no timeout */
841 #define RBD_READ_ONLY_DEFAULT false
842 #define RBD_LOCK_ON_READ_DEFAULT false
843 #define RBD_EXCLUSIVE_DEFAULT false
844 #define RBD_TRIM_DEFAULT true
846 struct rbd_parse_opts_ctx {
847 struct rbd_spec *spec;
848 struct ceph_options *copts;
849 struct rbd_options *opts;
852 static char* obj_op_name(enum obj_operation_type op_type)
869 * Destroy ceph client
871 * Caller must hold rbd_client_list_lock.
873 static void rbd_client_release(struct kref *kref)
875 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
877 dout("%s: rbdc %p\n", __func__, rbdc);
878 spin_lock(&rbd_client_list_lock);
879 list_del(&rbdc->node);
880 spin_unlock(&rbd_client_list_lock);
882 ceph_destroy_client(rbdc->client);
887 * Drop reference to ceph client node. If it's not referenced anymore, release
890 static void rbd_put_client(struct rbd_client *rbdc)
893 kref_put(&rbdc->kref, rbd_client_release);
897 * Get a ceph client with specific addr and configuration, if one does
898 * not exist create it. Either way, ceph_opts is consumed by this
901 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
903 struct rbd_client *rbdc;
906 mutex_lock(&client_mutex);
907 rbdc = rbd_client_find(ceph_opts);
909 ceph_destroy_options(ceph_opts);
912 * Using an existing client. Make sure ->pg_pools is up to
913 * date before we look up the pool id in do_rbd_add().
915 ret = ceph_wait_for_latest_osdmap(rbdc->client,
916 rbdc->client->options->mount_timeout);
918 rbd_warn(NULL, "failed to get latest osdmap: %d", ret);
919 rbd_put_client(rbdc);
923 rbdc = rbd_client_create(ceph_opts);
925 mutex_unlock(&client_mutex);
930 static bool rbd_image_format_valid(u32 image_format)
932 return image_format == 1 || image_format == 2;
935 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
940 /* The header has to start with the magic rbd header text */
941 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
944 /* The bio layer requires at least sector-sized I/O */
946 if (ondisk->options.order < SECTOR_SHIFT)
949 /* If we use u64 in a few spots we may be able to loosen this */
951 if (ondisk->options.order > 8 * sizeof (int) - 1)
955 * The size of a snapshot header has to fit in a size_t, and
956 * that limits the number of snapshots.
958 snap_count = le32_to_cpu(ondisk->snap_count);
959 size = SIZE_MAX - sizeof (struct ceph_snap_context);
960 if (snap_count > size / sizeof (__le64))
964 * Not only that, but the size of the entire the snapshot
965 * header must also be representable in a size_t.
967 size -= snap_count * sizeof (__le64);
968 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
975 * returns the size of an object in the image
977 static u32 rbd_obj_bytes(struct rbd_image_header *header)
979 return 1U << header->obj_order;
982 static void rbd_init_layout(struct rbd_device *rbd_dev)
984 if (rbd_dev->header.stripe_unit == 0 ||
985 rbd_dev->header.stripe_count == 0) {
986 rbd_dev->header.stripe_unit = rbd_obj_bytes(&rbd_dev->header);
987 rbd_dev->header.stripe_count = 1;
990 rbd_dev->layout.stripe_unit = rbd_dev->header.stripe_unit;
991 rbd_dev->layout.stripe_count = rbd_dev->header.stripe_count;
992 rbd_dev->layout.object_size = rbd_obj_bytes(&rbd_dev->header);
993 rbd_dev->layout.pool_id = rbd_dev->header.data_pool_id == CEPH_NOPOOL ?
994 rbd_dev->spec->pool_id : rbd_dev->header.data_pool_id;
995 RCU_INIT_POINTER(rbd_dev->layout.pool_ns, NULL);
999 * Fill an rbd image header with information from the given format 1
1002 static int rbd_header_from_disk(struct rbd_device *rbd_dev,
1003 struct rbd_image_header_ondisk *ondisk)
1005 struct rbd_image_header *header = &rbd_dev->header;
1006 bool first_time = header->object_prefix == NULL;
1007 struct ceph_snap_context *snapc;
1008 char *object_prefix = NULL;
1009 char *snap_names = NULL;
1010 u64 *snap_sizes = NULL;
1015 /* Allocate this now to avoid having to handle failure below */
1018 object_prefix = kstrndup(ondisk->object_prefix,
1019 sizeof(ondisk->object_prefix),
1025 /* Allocate the snapshot context and fill it in */
1027 snap_count = le32_to_cpu(ondisk->snap_count);
1028 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
1031 snapc->seq = le64_to_cpu(ondisk->snap_seq);
1033 struct rbd_image_snap_ondisk *snaps;
1034 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
1036 /* We'll keep a copy of the snapshot names... */
1038 if (snap_names_len > (u64)SIZE_MAX)
1040 snap_names = kmalloc(snap_names_len, GFP_KERNEL);
1044 /* ...as well as the array of their sizes. */
1045 snap_sizes = kmalloc_array(snap_count,
1046 sizeof(*header->snap_sizes),
1052 * Copy the names, and fill in each snapshot's id
1055 * Note that rbd_dev_v1_header_info() guarantees the
1056 * ondisk buffer we're working with has
1057 * snap_names_len bytes beyond the end of the
1058 * snapshot id array, this memcpy() is safe.
1060 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
1061 snaps = ondisk->snaps;
1062 for (i = 0; i < snap_count; i++) {
1063 snapc->snaps[i] = le64_to_cpu(snaps[i].id);
1064 snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
1068 /* We won't fail any more, fill in the header */
1071 header->object_prefix = object_prefix;
1072 header->obj_order = ondisk->options.order;
1073 rbd_init_layout(rbd_dev);
1075 ceph_put_snap_context(header->snapc);
1076 kfree(header->snap_names);
1077 kfree(header->snap_sizes);
1080 /* The remaining fields always get updated (when we refresh) */
1082 header->image_size = le64_to_cpu(ondisk->image_size);
1083 header->snapc = snapc;
1084 header->snap_names = snap_names;
1085 header->snap_sizes = snap_sizes;
1093 ceph_put_snap_context(snapc);
1094 kfree(object_prefix);
1099 static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1101 const char *snap_name;
1103 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1105 /* Skip over names until we find the one we are looking for */
1107 snap_name = rbd_dev->header.snap_names;
1109 snap_name += strlen(snap_name) + 1;
1111 return kstrdup(snap_name, GFP_KERNEL);
1115 * Snapshot id comparison function for use with qsort()/bsearch().
1116 * Note that result is for snapshots in *descending* order.
1118 static int snapid_compare_reverse(const void *s1, const void *s2)
1120 u64 snap_id1 = *(u64 *)s1;
1121 u64 snap_id2 = *(u64 *)s2;
1123 if (snap_id1 < snap_id2)
1125 return snap_id1 == snap_id2 ? 0 : -1;
1129 * Search a snapshot context to see if the given snapshot id is
1132 * Returns the position of the snapshot id in the array if it's found,
1133 * or BAD_SNAP_INDEX otherwise.
1135 * Note: The snapshot array is in kept sorted (by the osd) in
1136 * reverse order, highest snapshot id first.
1138 static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1140 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1143 found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1144 sizeof (snap_id), snapid_compare_reverse);
1146 return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1149 static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1153 const char *snap_name;
1155 which = rbd_dev_snap_index(rbd_dev, snap_id);
1156 if (which == BAD_SNAP_INDEX)
1157 return ERR_PTR(-ENOENT);
1159 snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1160 return snap_name ? snap_name : ERR_PTR(-ENOMEM);
1163 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1165 if (snap_id == CEPH_NOSNAP)
1166 return RBD_SNAP_HEAD_NAME;
1168 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1169 if (rbd_dev->image_format == 1)
1170 return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1172 return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1175 static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1178 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1179 if (snap_id == CEPH_NOSNAP) {
1180 *snap_size = rbd_dev->header.image_size;
1181 } else if (rbd_dev->image_format == 1) {
1184 which = rbd_dev_snap_index(rbd_dev, snap_id);
1185 if (which == BAD_SNAP_INDEX)
1188 *snap_size = rbd_dev->header.snap_sizes[which];
1193 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1202 static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1204 u64 snap_id = rbd_dev->spec->snap_id;
1208 ret = rbd_snap_size(rbd_dev, snap_id, &size);
1212 rbd_dev->mapping.size = size;
1216 static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1218 rbd_dev->mapping.size = 0;
1221 static void zero_bios(struct ceph_bio_iter *bio_pos, u32 off, u32 bytes)
1223 struct ceph_bio_iter it = *bio_pos;
1225 ceph_bio_iter_advance(&it, off);
1226 ceph_bio_iter_advance_step(&it, bytes, ({
1231 static void zero_bvecs(struct ceph_bvec_iter *bvec_pos, u32 off, u32 bytes)
1233 struct ceph_bvec_iter it = *bvec_pos;
1235 ceph_bvec_iter_advance(&it, off);
1236 ceph_bvec_iter_advance_step(&it, bytes, ({
1242 * Zero a range in @obj_req data buffer defined by a bio (list) or
1243 * (private) bio_vec array.
1245 * @off is relative to the start of the data buffer.
1247 static void rbd_obj_zero_range(struct rbd_obj_request *obj_req, u32 off,
1250 dout("%s %p data buf %u~%u\n", __func__, obj_req, off, bytes);
1252 switch (obj_req->img_request->data_type) {
1253 case OBJ_REQUEST_BIO:
1254 zero_bios(&obj_req->bio_pos, off, bytes);
1256 case OBJ_REQUEST_BVECS:
1257 case OBJ_REQUEST_OWN_BVECS:
1258 zero_bvecs(&obj_req->bvec_pos, off, bytes);
1265 static void rbd_obj_request_destroy(struct kref *kref);
1266 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1268 rbd_assert(obj_request != NULL);
1269 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1270 kref_read(&obj_request->kref));
1271 kref_put(&obj_request->kref, rbd_obj_request_destroy);
1274 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1275 struct rbd_obj_request *obj_request)
1277 rbd_assert(obj_request->img_request == NULL);
1279 /* Image request now owns object's original reference */
1280 obj_request->img_request = img_request;
1281 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1284 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1285 struct rbd_obj_request *obj_request)
1287 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1288 list_del(&obj_request->ex.oe_item);
1289 rbd_assert(obj_request->img_request == img_request);
1290 rbd_obj_request_put(obj_request);
1293 static void rbd_osd_submit(struct ceph_osd_request *osd_req)
1295 struct rbd_obj_request *obj_req = osd_req->r_priv;
1297 dout("%s osd_req %p for obj_req %p objno %llu %llu~%llu\n",
1298 __func__, osd_req, obj_req, obj_req->ex.oe_objno,
1299 obj_req->ex.oe_off, obj_req->ex.oe_len);
1300 ceph_osdc_start_request(osd_req->r_osdc, osd_req);
1304 * The default/initial value for all image request flags is 0. Each
1305 * is conditionally set to 1 at image request initialization time
1306 * and currently never change thereafter.
1308 static void img_request_layered_set(struct rbd_img_request *img_request)
1310 set_bit(IMG_REQ_LAYERED, &img_request->flags);
1313 static bool img_request_layered_test(struct rbd_img_request *img_request)
1315 return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1318 static bool rbd_obj_is_entire(struct rbd_obj_request *obj_req)
1320 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1322 return !obj_req->ex.oe_off &&
1323 obj_req->ex.oe_len == rbd_dev->layout.object_size;
1326 static bool rbd_obj_is_tail(struct rbd_obj_request *obj_req)
1328 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1330 return obj_req->ex.oe_off + obj_req->ex.oe_len ==
1331 rbd_dev->layout.object_size;
1335 * Must be called after rbd_obj_calc_img_extents().
1337 static void rbd_obj_set_copyup_enabled(struct rbd_obj_request *obj_req)
1339 rbd_assert(obj_req->img_request->snapc);
1341 if (obj_req->img_request->op_type == OBJ_OP_DISCARD) {
1342 dout("%s %p objno %llu discard\n", __func__, obj_req,
1343 obj_req->ex.oe_objno);
1347 if (!obj_req->num_img_extents) {
1348 dout("%s %p objno %llu not overlapping\n", __func__, obj_req,
1349 obj_req->ex.oe_objno);
1353 if (rbd_obj_is_entire(obj_req) &&
1354 !obj_req->img_request->snapc->num_snaps) {
1355 dout("%s %p objno %llu entire\n", __func__, obj_req,
1356 obj_req->ex.oe_objno);
1360 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
1363 static u64 rbd_obj_img_extents_bytes(struct rbd_obj_request *obj_req)
1365 return ceph_file_extents_bytes(obj_req->img_extents,
1366 obj_req->num_img_extents);
1369 static bool rbd_img_is_write(struct rbd_img_request *img_req)
1371 switch (img_req->op_type) {
1375 case OBJ_OP_DISCARD:
1376 case OBJ_OP_ZEROOUT:
1383 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
1385 struct rbd_obj_request *obj_req = osd_req->r_priv;
1388 dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
1389 osd_req->r_result, obj_req);
1392 * Writes aren't allowed to return a data payload. In some
1393 * guarded write cases (e.g. stat + zero on an empty object)
1394 * a stat response makes it through, but we don't care.
1396 if (osd_req->r_result > 0 && rbd_img_is_write(obj_req->img_request))
1399 result = osd_req->r_result;
1401 rbd_obj_handle_request(obj_req, result);
1404 static void rbd_osd_format_read(struct ceph_osd_request *osd_req)
1406 struct rbd_obj_request *obj_request = osd_req->r_priv;
1407 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1408 struct ceph_options *opt = rbd_dev->rbd_client->client->options;
1410 osd_req->r_flags = CEPH_OSD_FLAG_READ | opt->read_from_replica;
1411 osd_req->r_snapid = obj_request->img_request->snap_id;
1414 static void rbd_osd_format_write(struct ceph_osd_request *osd_req)
1416 struct rbd_obj_request *obj_request = osd_req->r_priv;
1418 osd_req->r_flags = CEPH_OSD_FLAG_WRITE;
1419 ktime_get_real_ts64(&osd_req->r_mtime);
1420 osd_req->r_data_offset = obj_request->ex.oe_off;
1423 static struct ceph_osd_request *
1424 __rbd_obj_add_osd_request(struct rbd_obj_request *obj_req,
1425 struct ceph_snap_context *snapc, int num_ops)
1427 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1428 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1429 struct ceph_osd_request *req;
1430 const char *name_format = rbd_dev->image_format == 1 ?
1431 RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
1434 req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, GFP_NOIO);
1436 return ERR_PTR(-ENOMEM);
1438 list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
1439 req->r_callback = rbd_osd_req_callback;
1440 req->r_priv = obj_req;
1443 * Data objects may be stored in a separate pool, but always in
1444 * the same namespace in that pool as the header in its pool.
1446 ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
1447 req->r_base_oloc.pool = rbd_dev->layout.pool_id;
1449 ret = ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
1450 rbd_dev->header.object_prefix,
1451 obj_req->ex.oe_objno);
1453 return ERR_PTR(ret);
1458 static struct ceph_osd_request *
1459 rbd_obj_add_osd_request(struct rbd_obj_request *obj_req, int num_ops)
1461 rbd_assert(obj_req->img_request->snapc);
1462 return __rbd_obj_add_osd_request(obj_req, obj_req->img_request->snapc,
1466 static struct rbd_obj_request *rbd_obj_request_create(void)
1468 struct rbd_obj_request *obj_request;
1470 obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_NOIO);
1474 ceph_object_extent_init(&obj_request->ex);
1475 INIT_LIST_HEAD(&obj_request->osd_reqs);
1476 mutex_init(&obj_request->state_mutex);
1477 kref_init(&obj_request->kref);
1479 dout("%s %p\n", __func__, obj_request);
1483 static void rbd_obj_request_destroy(struct kref *kref)
1485 struct rbd_obj_request *obj_request;
1486 struct ceph_osd_request *osd_req;
1489 obj_request = container_of(kref, struct rbd_obj_request, kref);
1491 dout("%s: obj %p\n", __func__, obj_request);
1493 while (!list_empty(&obj_request->osd_reqs)) {
1494 osd_req = list_first_entry(&obj_request->osd_reqs,
1495 struct ceph_osd_request, r_private_item);
1496 list_del_init(&osd_req->r_private_item);
1497 ceph_osdc_put_request(osd_req);
1500 switch (obj_request->img_request->data_type) {
1501 case OBJ_REQUEST_NODATA:
1502 case OBJ_REQUEST_BIO:
1503 case OBJ_REQUEST_BVECS:
1504 break; /* Nothing to do */
1505 case OBJ_REQUEST_OWN_BVECS:
1506 kfree(obj_request->bvec_pos.bvecs);
1512 kfree(obj_request->img_extents);
1513 if (obj_request->copyup_bvecs) {
1514 for (i = 0; i < obj_request->copyup_bvec_count; i++) {
1515 if (obj_request->copyup_bvecs[i].bv_page)
1516 __free_page(obj_request->copyup_bvecs[i].bv_page);
1518 kfree(obj_request->copyup_bvecs);
1521 kmem_cache_free(rbd_obj_request_cache, obj_request);
1524 /* It's OK to call this for a device with no parent */
1526 static void rbd_spec_put(struct rbd_spec *spec);
1527 static void rbd_dev_unparent(struct rbd_device *rbd_dev)
1529 rbd_dev_remove_parent(rbd_dev);
1530 rbd_spec_put(rbd_dev->parent_spec);
1531 rbd_dev->parent_spec = NULL;
1532 rbd_dev->parent_overlap = 0;
1536 * Parent image reference counting is used to determine when an
1537 * image's parent fields can be safely torn down--after there are no
1538 * more in-flight requests to the parent image. When the last
1539 * reference is dropped, cleaning them up is safe.
1541 static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
1545 if (!rbd_dev->parent_spec)
1548 counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
1552 /* Last reference; clean up parent data structures */
1555 rbd_dev_unparent(rbd_dev);
1557 rbd_warn(rbd_dev, "parent reference underflow");
1561 * If an image has a non-zero parent overlap, get a reference to its
1564 * Returns true if the rbd device has a parent with a non-zero
1565 * overlap and a reference for it was successfully taken, or
1568 static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
1572 if (!rbd_dev->parent_spec)
1575 if (rbd_dev->parent_overlap)
1576 counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
1579 rbd_warn(rbd_dev, "parent reference overflow");
1584 static void rbd_img_request_init(struct rbd_img_request *img_request,
1585 struct rbd_device *rbd_dev,
1586 enum obj_operation_type op_type)
1588 memset(img_request, 0, sizeof(*img_request));
1590 img_request->rbd_dev = rbd_dev;
1591 img_request->op_type = op_type;
1593 INIT_LIST_HEAD(&img_request->lock_item);
1594 INIT_LIST_HEAD(&img_request->object_extents);
1595 mutex_init(&img_request->state_mutex);
1599 * Only snap_id is captured here, for reads. For writes, snapshot
1600 * context is captured in rbd_img_object_requests() after exclusive
1601 * lock is ensured to be held.
1603 static void rbd_img_capture_header(struct rbd_img_request *img_req)
1605 struct rbd_device *rbd_dev = img_req->rbd_dev;
1607 lockdep_assert_held(&rbd_dev->header_rwsem);
1609 if (!rbd_img_is_write(img_req))
1610 img_req->snap_id = rbd_dev->spec->snap_id;
1612 if (rbd_dev_parent_get(rbd_dev))
1613 img_request_layered_set(img_req);
1616 static void rbd_img_request_destroy(struct rbd_img_request *img_request)
1618 struct rbd_obj_request *obj_request;
1619 struct rbd_obj_request *next_obj_request;
1621 dout("%s: img %p\n", __func__, img_request);
1623 WARN_ON(!list_empty(&img_request->lock_item));
1624 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1625 rbd_img_obj_request_del(img_request, obj_request);
1627 if (img_request_layered_test(img_request))
1628 rbd_dev_parent_put(img_request->rbd_dev);
1630 if (rbd_img_is_write(img_request))
1631 ceph_put_snap_context(img_request->snapc);
1633 if (test_bit(IMG_REQ_CHILD, &img_request->flags))
1634 kmem_cache_free(rbd_img_request_cache, img_request);
1637 #define BITS_PER_OBJ 2
1638 #define OBJS_PER_BYTE (BITS_PER_BYTE / BITS_PER_OBJ)
1639 #define OBJ_MASK ((1 << BITS_PER_OBJ) - 1)
1641 static void __rbd_object_map_index(struct rbd_device *rbd_dev, u64 objno,
1642 u64 *index, u8 *shift)
1646 rbd_assert(objno < rbd_dev->object_map_size);
1647 *index = div_u64_rem(objno, OBJS_PER_BYTE, &off);
1648 *shift = (OBJS_PER_BYTE - off - 1) * BITS_PER_OBJ;
1651 static u8 __rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1656 lockdep_assert_held(&rbd_dev->object_map_lock);
1657 __rbd_object_map_index(rbd_dev, objno, &index, &shift);
1658 return (rbd_dev->object_map[index] >> shift) & OBJ_MASK;
1661 static void __rbd_object_map_set(struct rbd_device *rbd_dev, u64 objno, u8 val)
1667 lockdep_assert_held(&rbd_dev->object_map_lock);
1668 rbd_assert(!(val & ~OBJ_MASK));
1670 __rbd_object_map_index(rbd_dev, objno, &index, &shift);
1671 p = &rbd_dev->object_map[index];
1672 *p = (*p & ~(OBJ_MASK << shift)) | (val << shift);
1675 static u8 rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1679 spin_lock(&rbd_dev->object_map_lock);
1680 state = __rbd_object_map_get(rbd_dev, objno);
1681 spin_unlock(&rbd_dev->object_map_lock);
1685 static bool use_object_map(struct rbd_device *rbd_dev)
1688 * An image mapped read-only can't use the object map -- it isn't
1689 * loaded because the header lock isn't acquired. Someone else can
1690 * write to the image and update the object map behind our back.
1692 * A snapshot can't be written to, so using the object map is always
1695 if (!rbd_is_snap(rbd_dev) && rbd_is_ro(rbd_dev))
1698 return ((rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) &&
1699 !(rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID));
1702 static bool rbd_object_map_may_exist(struct rbd_device *rbd_dev, u64 objno)
1706 /* fall back to default logic if object map is disabled or invalid */
1707 if (!use_object_map(rbd_dev))
1710 state = rbd_object_map_get(rbd_dev, objno);
1711 return state != OBJECT_NONEXISTENT;
1714 static void rbd_object_map_name(struct rbd_device *rbd_dev, u64 snap_id,
1715 struct ceph_object_id *oid)
1717 if (snap_id == CEPH_NOSNAP)
1718 ceph_oid_printf(oid, "%s%s", RBD_OBJECT_MAP_PREFIX,
1719 rbd_dev->spec->image_id);
1721 ceph_oid_printf(oid, "%s%s.%016llx", RBD_OBJECT_MAP_PREFIX,
1722 rbd_dev->spec->image_id, snap_id);
1725 static int rbd_object_map_lock(struct rbd_device *rbd_dev)
1727 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1728 CEPH_DEFINE_OID_ONSTACK(oid);
1731 struct ceph_locker *lockers;
1733 bool broke_lock = false;
1736 rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1739 ret = ceph_cls_lock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1740 CEPH_CLS_LOCK_EXCLUSIVE, "", "", "", 0);
1741 if (ret != -EBUSY || broke_lock) {
1743 ret = 0; /* already locked by myself */
1745 rbd_warn(rbd_dev, "failed to lock object map: %d", ret);
1749 ret = ceph_cls_lock_info(osdc, &oid, &rbd_dev->header_oloc,
1750 RBD_LOCK_NAME, &lock_type, &lock_tag,
1751 &lockers, &num_lockers);
1756 rbd_warn(rbd_dev, "failed to get object map lockers: %d", ret);
1761 if (num_lockers == 0)
1764 rbd_warn(rbd_dev, "breaking object map lock owned by %s%llu",
1765 ENTITY_NAME(lockers[0].id.name));
1767 ret = ceph_cls_break_lock(osdc, &oid, &rbd_dev->header_oloc,
1768 RBD_LOCK_NAME, lockers[0].id.cookie,
1769 &lockers[0].id.name);
1770 ceph_free_lockers(lockers, num_lockers);
1775 rbd_warn(rbd_dev, "failed to break object map lock: %d", ret);
1783 static void rbd_object_map_unlock(struct rbd_device *rbd_dev)
1785 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1786 CEPH_DEFINE_OID_ONSTACK(oid);
1789 rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1791 ret = ceph_cls_unlock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1793 if (ret && ret != -ENOENT)
1794 rbd_warn(rbd_dev, "failed to unlock object map: %d", ret);
1797 static int decode_object_map_header(void **p, void *end, u64 *object_map_size)
1805 ceph_decode_32_safe(p, end, header_len, e_inval);
1806 header_end = *p + header_len;
1808 ret = ceph_start_decoding(p, end, 1, "BitVector header", &struct_v,
1813 ceph_decode_64_safe(p, end, *object_map_size, e_inval);
1822 static int __rbd_object_map_load(struct rbd_device *rbd_dev)
1824 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1825 CEPH_DEFINE_OID_ONSTACK(oid);
1826 struct page **pages;
1830 u64 object_map_bytes;
1831 u64 object_map_size;
1835 rbd_assert(!rbd_dev->object_map && !rbd_dev->object_map_size);
1837 num_objects = ceph_get_num_objects(&rbd_dev->layout,
1838 rbd_dev->mapping.size);
1839 object_map_bytes = DIV_ROUND_UP_ULL(num_objects * BITS_PER_OBJ,
1841 num_pages = calc_pages_for(0, object_map_bytes) + 1;
1842 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1844 return PTR_ERR(pages);
1846 reply_len = num_pages * PAGE_SIZE;
1847 rbd_object_map_name(rbd_dev, rbd_dev->spec->snap_id, &oid);
1848 ret = ceph_osdc_call(osdc, &oid, &rbd_dev->header_oloc,
1849 "rbd", "object_map_load", CEPH_OSD_FLAG_READ,
1850 NULL, 0, pages, &reply_len);
1854 p = page_address(pages[0]);
1855 end = p + min(reply_len, (size_t)PAGE_SIZE);
1856 ret = decode_object_map_header(&p, end, &object_map_size);
1860 if (object_map_size != num_objects) {
1861 rbd_warn(rbd_dev, "object map size mismatch: %llu vs %llu",
1862 object_map_size, num_objects);
1867 if (offset_in_page(p) + object_map_bytes > reply_len) {
1872 rbd_dev->object_map = kvmalloc(object_map_bytes, GFP_KERNEL);
1873 if (!rbd_dev->object_map) {
1878 rbd_dev->object_map_size = object_map_size;
1879 ceph_copy_from_page_vector(pages, rbd_dev->object_map,
1880 offset_in_page(p), object_map_bytes);
1883 ceph_release_page_vector(pages, num_pages);
1887 static void rbd_object_map_free(struct rbd_device *rbd_dev)
1889 kvfree(rbd_dev->object_map);
1890 rbd_dev->object_map = NULL;
1891 rbd_dev->object_map_size = 0;
1894 static int rbd_object_map_load(struct rbd_device *rbd_dev)
1898 ret = __rbd_object_map_load(rbd_dev);
1902 ret = rbd_dev_v2_get_flags(rbd_dev);
1904 rbd_object_map_free(rbd_dev);
1908 if (rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID)
1909 rbd_warn(rbd_dev, "object map is invalid");
1914 static int rbd_object_map_open(struct rbd_device *rbd_dev)
1918 ret = rbd_object_map_lock(rbd_dev);
1922 ret = rbd_object_map_load(rbd_dev);
1924 rbd_object_map_unlock(rbd_dev);
1931 static void rbd_object_map_close(struct rbd_device *rbd_dev)
1933 rbd_object_map_free(rbd_dev);
1934 rbd_object_map_unlock(rbd_dev);
1938 * This function needs snap_id (or more precisely just something to
1939 * distinguish between HEAD and snapshot object maps), new_state and
1940 * current_state that were passed to rbd_object_map_update().
1942 * To avoid allocating and stashing a context we piggyback on the OSD
1943 * request. A HEAD update has two ops (assert_locked). For new_state
1944 * and current_state we decode our own object_map_update op, encoded in
1945 * rbd_cls_object_map_update().
1947 static int rbd_object_map_update_finish(struct rbd_obj_request *obj_req,
1948 struct ceph_osd_request *osd_req)
1950 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1951 struct ceph_osd_data *osd_data;
1953 u8 state, new_state, current_state;
1954 bool has_current_state;
1957 if (osd_req->r_result)
1958 return osd_req->r_result;
1961 * Nothing to do for a snapshot object map.
1963 if (osd_req->r_num_ops == 1)
1967 * Update in-memory HEAD object map.
1969 rbd_assert(osd_req->r_num_ops == 2);
1970 osd_data = osd_req_op_data(osd_req, 1, cls, request_data);
1971 rbd_assert(osd_data->type == CEPH_OSD_DATA_TYPE_PAGES);
1973 p = page_address(osd_data->pages[0]);
1974 objno = ceph_decode_64(&p);
1975 rbd_assert(objno == obj_req->ex.oe_objno);
1976 rbd_assert(ceph_decode_64(&p) == objno + 1);
1977 new_state = ceph_decode_8(&p);
1978 has_current_state = ceph_decode_8(&p);
1979 if (has_current_state)
1980 current_state = ceph_decode_8(&p);
1982 spin_lock(&rbd_dev->object_map_lock);
1983 state = __rbd_object_map_get(rbd_dev, objno);
1984 if (!has_current_state || current_state == state ||
1985 (current_state == OBJECT_EXISTS && state == OBJECT_EXISTS_CLEAN))
1986 __rbd_object_map_set(rbd_dev, objno, new_state);
1987 spin_unlock(&rbd_dev->object_map_lock);
1992 static void rbd_object_map_callback(struct ceph_osd_request *osd_req)
1994 struct rbd_obj_request *obj_req = osd_req->r_priv;
1997 dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
1998 osd_req->r_result, obj_req);
2000 result = rbd_object_map_update_finish(obj_req, osd_req);
2001 rbd_obj_handle_request(obj_req, result);
2004 static bool update_needed(struct rbd_device *rbd_dev, u64 objno, u8 new_state)
2006 u8 state = rbd_object_map_get(rbd_dev, objno);
2008 if (state == new_state ||
2009 (new_state == OBJECT_PENDING && state == OBJECT_NONEXISTENT) ||
2010 (new_state == OBJECT_NONEXISTENT && state != OBJECT_PENDING))
2016 static int rbd_cls_object_map_update(struct ceph_osd_request *req,
2017 int which, u64 objno, u8 new_state,
2018 const u8 *current_state)
2020 struct page **pages;
2024 ret = osd_req_op_cls_init(req, which, "rbd", "object_map_update");
2028 pages = ceph_alloc_page_vector(1, GFP_NOIO);
2030 return PTR_ERR(pages);
2032 p = start = page_address(pages[0]);
2033 ceph_encode_64(&p, objno);
2034 ceph_encode_64(&p, objno + 1);
2035 ceph_encode_8(&p, new_state);
2036 if (current_state) {
2037 ceph_encode_8(&p, 1);
2038 ceph_encode_8(&p, *current_state);
2040 ceph_encode_8(&p, 0);
2043 osd_req_op_cls_request_data_pages(req, which, pages, p - start, 0,
2050 * 0 - object map update sent
2051 * 1 - object map update isn't needed
2054 static int rbd_object_map_update(struct rbd_obj_request *obj_req, u64 snap_id,
2055 u8 new_state, const u8 *current_state)
2057 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2058 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2059 struct ceph_osd_request *req;
2064 if (snap_id == CEPH_NOSNAP) {
2065 if (!update_needed(rbd_dev, obj_req->ex.oe_objno, new_state))
2068 num_ops++; /* assert_locked */
2071 req = ceph_osdc_alloc_request(osdc, NULL, num_ops, false, GFP_NOIO);
2075 list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
2076 req->r_callback = rbd_object_map_callback;
2077 req->r_priv = obj_req;
2079 rbd_object_map_name(rbd_dev, snap_id, &req->r_base_oid);
2080 ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
2081 req->r_flags = CEPH_OSD_FLAG_WRITE;
2082 ktime_get_real_ts64(&req->r_mtime);
2084 if (snap_id == CEPH_NOSNAP) {
2086 * Protect against possible race conditions during lock
2087 * ownership transitions.
2089 ret = ceph_cls_assert_locked(req, which++, RBD_LOCK_NAME,
2090 CEPH_CLS_LOCK_EXCLUSIVE, "", "");
2095 ret = rbd_cls_object_map_update(req, which, obj_req->ex.oe_objno,
2096 new_state, current_state);
2100 ret = ceph_osdc_alloc_messages(req, GFP_NOIO);
2104 ceph_osdc_start_request(osdc, req);
2108 static void prune_extents(struct ceph_file_extent *img_extents,
2109 u32 *num_img_extents, u64 overlap)
2111 u32 cnt = *num_img_extents;
2113 /* drop extents completely beyond the overlap */
2114 while (cnt && img_extents[cnt - 1].fe_off >= overlap)
2118 struct ceph_file_extent *ex = &img_extents[cnt - 1];
2120 /* trim final overlapping extent */
2121 if (ex->fe_off + ex->fe_len > overlap)
2122 ex->fe_len = overlap - ex->fe_off;
2125 *num_img_extents = cnt;
2129 * Determine the byte range(s) covered by either just the object extent
2130 * or the entire object in the parent image.
2132 static int rbd_obj_calc_img_extents(struct rbd_obj_request *obj_req,
2135 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2138 if (!rbd_dev->parent_overlap)
2141 ret = ceph_extent_to_file(&rbd_dev->layout, obj_req->ex.oe_objno,
2142 entire ? 0 : obj_req->ex.oe_off,
2143 entire ? rbd_dev->layout.object_size :
2145 &obj_req->img_extents,
2146 &obj_req->num_img_extents);
2150 prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
2151 rbd_dev->parent_overlap);
2155 static void rbd_osd_setup_data(struct ceph_osd_request *osd_req, int which)
2157 struct rbd_obj_request *obj_req = osd_req->r_priv;
2159 switch (obj_req->img_request->data_type) {
2160 case OBJ_REQUEST_BIO:
2161 osd_req_op_extent_osd_data_bio(osd_req, which,
2163 obj_req->ex.oe_len);
2165 case OBJ_REQUEST_BVECS:
2166 case OBJ_REQUEST_OWN_BVECS:
2167 rbd_assert(obj_req->bvec_pos.iter.bi_size ==
2168 obj_req->ex.oe_len);
2169 rbd_assert(obj_req->bvec_idx == obj_req->bvec_count);
2170 osd_req_op_extent_osd_data_bvec_pos(osd_req, which,
2171 &obj_req->bvec_pos);
2178 static int rbd_osd_setup_stat(struct ceph_osd_request *osd_req, int which)
2180 struct page **pages;
2183 * The response data for a STAT call consists of:
2190 pages = ceph_alloc_page_vector(1, GFP_NOIO);
2192 return PTR_ERR(pages);
2194 osd_req_op_init(osd_req, which, CEPH_OSD_OP_STAT, 0);
2195 osd_req_op_raw_data_in_pages(osd_req, which, pages,
2196 8 + sizeof(struct ceph_timespec),
2201 static int rbd_osd_setup_copyup(struct ceph_osd_request *osd_req, int which,
2204 struct rbd_obj_request *obj_req = osd_req->r_priv;
2207 ret = osd_req_op_cls_init(osd_req, which, "rbd", "copyup");
2211 osd_req_op_cls_request_data_bvecs(osd_req, which, obj_req->copyup_bvecs,
2212 obj_req->copyup_bvec_count, bytes);
2216 static int rbd_obj_init_read(struct rbd_obj_request *obj_req)
2218 obj_req->read_state = RBD_OBJ_READ_START;
2222 static void __rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2225 struct rbd_obj_request *obj_req = osd_req->r_priv;
2226 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2229 if (!use_object_map(rbd_dev) ||
2230 !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST)) {
2231 osd_req_op_alloc_hint_init(osd_req, which++,
2232 rbd_dev->layout.object_size,
2233 rbd_dev->layout.object_size,
2234 rbd_dev->opts->alloc_hint_flags);
2237 if (rbd_obj_is_entire(obj_req))
2238 opcode = CEPH_OSD_OP_WRITEFULL;
2240 opcode = CEPH_OSD_OP_WRITE;
2242 osd_req_op_extent_init(osd_req, which, opcode,
2243 obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2244 rbd_osd_setup_data(osd_req, which);
2247 static int rbd_obj_init_write(struct rbd_obj_request *obj_req)
2251 /* reverse map the entire object onto the parent */
2252 ret = rbd_obj_calc_img_extents(obj_req, true);
2256 obj_req->write_state = RBD_OBJ_WRITE_START;
2260 static u16 truncate_or_zero_opcode(struct rbd_obj_request *obj_req)
2262 return rbd_obj_is_tail(obj_req) ? CEPH_OSD_OP_TRUNCATE :
2266 static void __rbd_osd_setup_discard_ops(struct ceph_osd_request *osd_req,
2269 struct rbd_obj_request *obj_req = osd_req->r_priv;
2271 if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
2272 rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2273 osd_req_op_init(osd_req, which, CEPH_OSD_OP_DELETE, 0);
2275 osd_req_op_extent_init(osd_req, which,
2276 truncate_or_zero_opcode(obj_req),
2277 obj_req->ex.oe_off, obj_req->ex.oe_len,
2282 static int rbd_obj_init_discard(struct rbd_obj_request *obj_req)
2284 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2289 * Align the range to alloc_size boundary and punt on discards
2290 * that are too small to free up any space.
2292 * alloc_size == object_size && is_tail() is a special case for
2293 * filestore with filestore_punch_hole = false, needed to allow
2294 * truncate (in addition to delete).
2296 if (rbd_dev->opts->alloc_size != rbd_dev->layout.object_size ||
2297 !rbd_obj_is_tail(obj_req)) {
2298 off = round_up(obj_req->ex.oe_off, rbd_dev->opts->alloc_size);
2299 next_off = round_down(obj_req->ex.oe_off + obj_req->ex.oe_len,
2300 rbd_dev->opts->alloc_size);
2301 if (off >= next_off)
2304 dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
2305 obj_req, obj_req->ex.oe_off, obj_req->ex.oe_len,
2306 off, next_off - off);
2307 obj_req->ex.oe_off = off;
2308 obj_req->ex.oe_len = next_off - off;
2311 /* reverse map the entire object onto the parent */
2312 ret = rbd_obj_calc_img_extents(obj_req, true);
2316 obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2317 if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents)
2318 obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2320 obj_req->write_state = RBD_OBJ_WRITE_START;
2324 static void __rbd_osd_setup_zeroout_ops(struct ceph_osd_request *osd_req,
2327 struct rbd_obj_request *obj_req = osd_req->r_priv;
2330 if (rbd_obj_is_entire(obj_req)) {
2331 if (obj_req->num_img_extents) {
2332 if (!(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2333 osd_req_op_init(osd_req, which++,
2334 CEPH_OSD_OP_CREATE, 0);
2335 opcode = CEPH_OSD_OP_TRUNCATE;
2337 rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2338 osd_req_op_init(osd_req, which++,
2339 CEPH_OSD_OP_DELETE, 0);
2343 opcode = truncate_or_zero_opcode(obj_req);
2347 osd_req_op_extent_init(osd_req, which, opcode,
2348 obj_req->ex.oe_off, obj_req->ex.oe_len,
2352 static int rbd_obj_init_zeroout(struct rbd_obj_request *obj_req)
2356 /* reverse map the entire object onto the parent */
2357 ret = rbd_obj_calc_img_extents(obj_req, true);
2361 if (!obj_req->num_img_extents) {
2362 obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2363 if (rbd_obj_is_entire(obj_req))
2364 obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2367 obj_req->write_state = RBD_OBJ_WRITE_START;
2371 static int count_write_ops(struct rbd_obj_request *obj_req)
2373 struct rbd_img_request *img_req = obj_req->img_request;
2375 switch (img_req->op_type) {
2377 if (!use_object_map(img_req->rbd_dev) ||
2378 !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST))
2379 return 2; /* setallochint + write/writefull */
2381 return 1; /* write/writefull */
2382 case OBJ_OP_DISCARD:
2383 return 1; /* delete/truncate/zero */
2384 case OBJ_OP_ZEROOUT:
2385 if (rbd_obj_is_entire(obj_req) && obj_req->num_img_extents &&
2386 !(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2387 return 2; /* create + truncate */
2389 return 1; /* delete/truncate/zero */
2395 static void rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2398 struct rbd_obj_request *obj_req = osd_req->r_priv;
2400 switch (obj_req->img_request->op_type) {
2402 __rbd_osd_setup_write_ops(osd_req, which);
2404 case OBJ_OP_DISCARD:
2405 __rbd_osd_setup_discard_ops(osd_req, which);
2407 case OBJ_OP_ZEROOUT:
2408 __rbd_osd_setup_zeroout_ops(osd_req, which);
2416 * Prune the list of object requests (adjust offset and/or length, drop
2417 * redundant requests). Prepare object request state machines and image
2418 * request state machine for execution.
2420 static int __rbd_img_fill_request(struct rbd_img_request *img_req)
2422 struct rbd_obj_request *obj_req, *next_obj_req;
2425 for_each_obj_request_safe(img_req, obj_req, next_obj_req) {
2426 switch (img_req->op_type) {
2428 ret = rbd_obj_init_read(obj_req);
2431 ret = rbd_obj_init_write(obj_req);
2433 case OBJ_OP_DISCARD:
2434 ret = rbd_obj_init_discard(obj_req);
2436 case OBJ_OP_ZEROOUT:
2437 ret = rbd_obj_init_zeroout(obj_req);
2445 rbd_img_obj_request_del(img_req, obj_req);
2450 img_req->state = RBD_IMG_START;
2454 union rbd_img_fill_iter {
2455 struct ceph_bio_iter bio_iter;
2456 struct ceph_bvec_iter bvec_iter;
2459 struct rbd_img_fill_ctx {
2460 enum obj_request_type pos_type;
2461 union rbd_img_fill_iter *pos;
2462 union rbd_img_fill_iter iter;
2463 ceph_object_extent_fn_t set_pos_fn;
2464 ceph_object_extent_fn_t count_fn;
2465 ceph_object_extent_fn_t copy_fn;
2468 static struct ceph_object_extent *alloc_object_extent(void *arg)
2470 struct rbd_img_request *img_req = arg;
2471 struct rbd_obj_request *obj_req;
2473 obj_req = rbd_obj_request_create();
2477 rbd_img_obj_request_add(img_req, obj_req);
2478 return &obj_req->ex;
2482 * While su != os && sc == 1 is technically not fancy (it's the same
2483 * layout as su == os && sc == 1), we can't use the nocopy path for it
2484 * because ->set_pos_fn() should be called only once per object.
2485 * ceph_file_to_extents() invokes action_fn once per stripe unit, so
2486 * treat su != os && sc == 1 as fancy.
2488 static bool rbd_layout_is_fancy(struct ceph_file_layout *l)
2490 return l->stripe_unit != l->object_size;
2493 static int rbd_img_fill_request_nocopy(struct rbd_img_request *img_req,
2494 struct ceph_file_extent *img_extents,
2495 u32 num_img_extents,
2496 struct rbd_img_fill_ctx *fctx)
2501 img_req->data_type = fctx->pos_type;
2504 * Create object requests and set each object request's starting
2505 * position in the provided bio (list) or bio_vec array.
2507 fctx->iter = *fctx->pos;
2508 for (i = 0; i < num_img_extents; i++) {
2509 ret = ceph_file_to_extents(&img_req->rbd_dev->layout,
2510 img_extents[i].fe_off,
2511 img_extents[i].fe_len,
2512 &img_req->object_extents,
2513 alloc_object_extent, img_req,
2514 fctx->set_pos_fn, &fctx->iter);
2519 return __rbd_img_fill_request(img_req);
2523 * Map a list of image extents to a list of object extents, create the
2524 * corresponding object requests (normally each to a different object,
2525 * but not always) and add them to @img_req. For each object request,
2526 * set up its data descriptor to point to the corresponding chunk(s) of
2527 * @fctx->pos data buffer.
2529 * Because ceph_file_to_extents() will merge adjacent object extents
2530 * together, each object request's data descriptor may point to multiple
2531 * different chunks of @fctx->pos data buffer.
2533 * @fctx->pos data buffer is assumed to be large enough.
2535 static int rbd_img_fill_request(struct rbd_img_request *img_req,
2536 struct ceph_file_extent *img_extents,
2537 u32 num_img_extents,
2538 struct rbd_img_fill_ctx *fctx)
2540 struct rbd_device *rbd_dev = img_req->rbd_dev;
2541 struct rbd_obj_request *obj_req;
2545 if (fctx->pos_type == OBJ_REQUEST_NODATA ||
2546 !rbd_layout_is_fancy(&rbd_dev->layout))
2547 return rbd_img_fill_request_nocopy(img_req, img_extents,
2548 num_img_extents, fctx);
2550 img_req->data_type = OBJ_REQUEST_OWN_BVECS;
2553 * Create object requests and determine ->bvec_count for each object
2554 * request. Note that ->bvec_count sum over all object requests may
2555 * be greater than the number of bio_vecs in the provided bio (list)
2556 * or bio_vec array because when mapped, those bio_vecs can straddle
2557 * stripe unit boundaries.
2559 fctx->iter = *fctx->pos;
2560 for (i = 0; i < num_img_extents; i++) {
2561 ret = ceph_file_to_extents(&rbd_dev->layout,
2562 img_extents[i].fe_off,
2563 img_extents[i].fe_len,
2564 &img_req->object_extents,
2565 alloc_object_extent, img_req,
2566 fctx->count_fn, &fctx->iter);
2571 for_each_obj_request(img_req, obj_req) {
2572 obj_req->bvec_pos.bvecs = kmalloc_array(obj_req->bvec_count,
2573 sizeof(*obj_req->bvec_pos.bvecs),
2575 if (!obj_req->bvec_pos.bvecs)
2580 * Fill in each object request's private bio_vec array, splitting and
2581 * rearranging the provided bio_vecs in stripe unit chunks as needed.
2583 fctx->iter = *fctx->pos;
2584 for (i = 0; i < num_img_extents; i++) {
2585 ret = ceph_iterate_extents(&rbd_dev->layout,
2586 img_extents[i].fe_off,
2587 img_extents[i].fe_len,
2588 &img_req->object_extents,
2589 fctx->copy_fn, &fctx->iter);
2594 return __rbd_img_fill_request(img_req);
2597 static int rbd_img_fill_nodata(struct rbd_img_request *img_req,
2600 struct ceph_file_extent ex = { off, len };
2601 union rbd_img_fill_iter dummy = {};
2602 struct rbd_img_fill_ctx fctx = {
2603 .pos_type = OBJ_REQUEST_NODATA,
2607 return rbd_img_fill_request(img_req, &ex, 1, &fctx);
2610 static void set_bio_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2612 struct rbd_obj_request *obj_req =
2613 container_of(ex, struct rbd_obj_request, ex);
2614 struct ceph_bio_iter *it = arg;
2616 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2617 obj_req->bio_pos = *it;
2618 ceph_bio_iter_advance(it, bytes);
2621 static void count_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2623 struct rbd_obj_request *obj_req =
2624 container_of(ex, struct rbd_obj_request, ex);
2625 struct ceph_bio_iter *it = arg;
2627 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2628 ceph_bio_iter_advance_step(it, bytes, ({
2629 obj_req->bvec_count++;
2634 static void copy_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2636 struct rbd_obj_request *obj_req =
2637 container_of(ex, struct rbd_obj_request, ex);
2638 struct ceph_bio_iter *it = arg;
2640 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2641 ceph_bio_iter_advance_step(it, bytes, ({
2642 obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2643 obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2647 static int __rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2648 struct ceph_file_extent *img_extents,
2649 u32 num_img_extents,
2650 struct ceph_bio_iter *bio_pos)
2652 struct rbd_img_fill_ctx fctx = {
2653 .pos_type = OBJ_REQUEST_BIO,
2654 .pos = (union rbd_img_fill_iter *)bio_pos,
2655 .set_pos_fn = set_bio_pos,
2656 .count_fn = count_bio_bvecs,
2657 .copy_fn = copy_bio_bvecs,
2660 return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2664 static int rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2665 u64 off, u64 len, struct bio *bio)
2667 struct ceph_file_extent ex = { off, len };
2668 struct ceph_bio_iter it = { .bio = bio, .iter = bio->bi_iter };
2670 return __rbd_img_fill_from_bio(img_req, &ex, 1, &it);
2673 static void set_bvec_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2675 struct rbd_obj_request *obj_req =
2676 container_of(ex, struct rbd_obj_request, ex);
2677 struct ceph_bvec_iter *it = arg;
2679 obj_req->bvec_pos = *it;
2680 ceph_bvec_iter_shorten(&obj_req->bvec_pos, bytes);
2681 ceph_bvec_iter_advance(it, bytes);
2684 static void count_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2686 struct rbd_obj_request *obj_req =
2687 container_of(ex, struct rbd_obj_request, ex);
2688 struct ceph_bvec_iter *it = arg;
2690 ceph_bvec_iter_advance_step(it, bytes, ({
2691 obj_req->bvec_count++;
2695 static void copy_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2697 struct rbd_obj_request *obj_req =
2698 container_of(ex, struct rbd_obj_request, ex);
2699 struct ceph_bvec_iter *it = arg;
2701 ceph_bvec_iter_advance_step(it, bytes, ({
2702 obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2703 obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2707 static int __rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2708 struct ceph_file_extent *img_extents,
2709 u32 num_img_extents,
2710 struct ceph_bvec_iter *bvec_pos)
2712 struct rbd_img_fill_ctx fctx = {
2713 .pos_type = OBJ_REQUEST_BVECS,
2714 .pos = (union rbd_img_fill_iter *)bvec_pos,
2715 .set_pos_fn = set_bvec_pos,
2716 .count_fn = count_bvecs,
2717 .copy_fn = copy_bvecs,
2720 return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2724 static int rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2725 struct ceph_file_extent *img_extents,
2726 u32 num_img_extents,
2727 struct bio_vec *bvecs)
2729 struct ceph_bvec_iter it = {
2731 .iter = { .bi_size = ceph_file_extents_bytes(img_extents,
2735 return __rbd_img_fill_from_bvecs(img_req, img_extents, num_img_extents,
2739 static void rbd_img_handle_request_work(struct work_struct *work)
2741 struct rbd_img_request *img_req =
2742 container_of(work, struct rbd_img_request, work);
2744 rbd_img_handle_request(img_req, img_req->work_result);
2747 static void rbd_img_schedule(struct rbd_img_request *img_req, int result)
2749 INIT_WORK(&img_req->work, rbd_img_handle_request_work);
2750 img_req->work_result = result;
2751 queue_work(rbd_wq, &img_req->work);
2754 static bool rbd_obj_may_exist(struct rbd_obj_request *obj_req)
2756 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2758 if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno)) {
2759 obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2763 dout("%s %p objno %llu assuming dne\n", __func__, obj_req,
2764 obj_req->ex.oe_objno);
2768 static int rbd_obj_read_object(struct rbd_obj_request *obj_req)
2770 struct ceph_osd_request *osd_req;
2773 osd_req = __rbd_obj_add_osd_request(obj_req, NULL, 1);
2774 if (IS_ERR(osd_req))
2775 return PTR_ERR(osd_req);
2777 osd_req_op_extent_init(osd_req, 0, CEPH_OSD_OP_READ,
2778 obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2779 rbd_osd_setup_data(osd_req, 0);
2780 rbd_osd_format_read(osd_req);
2782 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2786 rbd_osd_submit(osd_req);
2790 static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
2792 struct rbd_img_request *img_req = obj_req->img_request;
2793 struct rbd_device *parent = img_req->rbd_dev->parent;
2794 struct rbd_img_request *child_img_req;
2797 child_img_req = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
2801 rbd_img_request_init(child_img_req, parent, OBJ_OP_READ);
2802 __set_bit(IMG_REQ_CHILD, &child_img_req->flags);
2803 child_img_req->obj_request = obj_req;
2805 down_read(&parent->header_rwsem);
2806 rbd_img_capture_header(child_img_req);
2807 up_read(&parent->header_rwsem);
2809 dout("%s child_img_req %p for obj_req %p\n", __func__, child_img_req,
2812 if (!rbd_img_is_write(img_req)) {
2813 switch (img_req->data_type) {
2814 case OBJ_REQUEST_BIO:
2815 ret = __rbd_img_fill_from_bio(child_img_req,
2816 obj_req->img_extents,
2817 obj_req->num_img_extents,
2820 case OBJ_REQUEST_BVECS:
2821 case OBJ_REQUEST_OWN_BVECS:
2822 ret = __rbd_img_fill_from_bvecs(child_img_req,
2823 obj_req->img_extents,
2824 obj_req->num_img_extents,
2825 &obj_req->bvec_pos);
2831 ret = rbd_img_fill_from_bvecs(child_img_req,
2832 obj_req->img_extents,
2833 obj_req->num_img_extents,
2834 obj_req->copyup_bvecs);
2837 rbd_img_request_destroy(child_img_req);
2841 /* avoid parent chain recursion */
2842 rbd_img_schedule(child_img_req, 0);
2846 static bool rbd_obj_advance_read(struct rbd_obj_request *obj_req, int *result)
2848 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2852 switch (obj_req->read_state) {
2853 case RBD_OBJ_READ_START:
2854 rbd_assert(!*result);
2856 if (!rbd_obj_may_exist(obj_req)) {
2858 obj_req->read_state = RBD_OBJ_READ_OBJECT;
2862 ret = rbd_obj_read_object(obj_req);
2867 obj_req->read_state = RBD_OBJ_READ_OBJECT;
2869 case RBD_OBJ_READ_OBJECT:
2870 if (*result == -ENOENT && rbd_dev->parent_overlap) {
2871 /* reverse map this object extent onto the parent */
2872 ret = rbd_obj_calc_img_extents(obj_req, false);
2877 if (obj_req->num_img_extents) {
2878 ret = rbd_obj_read_from_parent(obj_req);
2883 obj_req->read_state = RBD_OBJ_READ_PARENT;
2889 * -ENOENT means a hole in the image -- zero-fill the entire
2890 * length of the request. A short read also implies zero-fill
2891 * to the end of the request.
2893 if (*result == -ENOENT) {
2894 rbd_obj_zero_range(obj_req, 0, obj_req->ex.oe_len);
2896 } else if (*result >= 0) {
2897 if (*result < obj_req->ex.oe_len)
2898 rbd_obj_zero_range(obj_req, *result,
2899 obj_req->ex.oe_len - *result);
2901 rbd_assert(*result == obj_req->ex.oe_len);
2905 case RBD_OBJ_READ_PARENT:
2907 * The parent image is read only up to the overlap -- zero-fill
2908 * from the overlap to the end of the request.
2911 u32 obj_overlap = rbd_obj_img_extents_bytes(obj_req);
2913 if (obj_overlap < obj_req->ex.oe_len)
2914 rbd_obj_zero_range(obj_req, obj_overlap,
2915 obj_req->ex.oe_len - obj_overlap);
2923 static bool rbd_obj_write_is_noop(struct rbd_obj_request *obj_req)
2925 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2927 if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno))
2928 obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2930 if (!(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST) &&
2931 (obj_req->flags & RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT)) {
2932 dout("%s %p noop for nonexistent\n", __func__, obj_req);
2941 * 0 - object map update sent
2942 * 1 - object map update isn't needed
2945 static int rbd_obj_write_pre_object_map(struct rbd_obj_request *obj_req)
2947 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2950 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
2953 if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
2954 new_state = OBJECT_PENDING;
2956 new_state = OBJECT_EXISTS;
2958 return rbd_object_map_update(obj_req, CEPH_NOSNAP, new_state, NULL);
2961 static int rbd_obj_write_object(struct rbd_obj_request *obj_req)
2963 struct ceph_osd_request *osd_req;
2964 int num_ops = count_write_ops(obj_req);
2968 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
2969 num_ops++; /* stat */
2971 osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
2972 if (IS_ERR(osd_req))
2973 return PTR_ERR(osd_req);
2975 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
2976 ret = rbd_osd_setup_stat(osd_req, which++);
2981 rbd_osd_setup_write_ops(osd_req, which);
2982 rbd_osd_format_write(osd_req);
2984 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2988 rbd_osd_submit(osd_req);
2993 * copyup_bvecs pages are never highmem pages
2995 static bool is_zero_bvecs(struct bio_vec *bvecs, u32 bytes)
2997 struct ceph_bvec_iter it = {
2999 .iter = { .bi_size = bytes },
3002 ceph_bvec_iter_advance_step(&it, bytes, ({
3003 if (memchr_inv(bvec_virt(&bv), 0, bv.bv_len))
3009 #define MODS_ONLY U32_MAX
3011 static int rbd_obj_copyup_empty_snapc(struct rbd_obj_request *obj_req,
3014 struct ceph_osd_request *osd_req;
3017 dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3018 rbd_assert(bytes > 0 && bytes != MODS_ONLY);
3020 osd_req = __rbd_obj_add_osd_request(obj_req, &rbd_empty_snapc, 1);
3021 if (IS_ERR(osd_req))
3022 return PTR_ERR(osd_req);
3024 ret = rbd_osd_setup_copyup(osd_req, 0, bytes);
3028 rbd_osd_format_write(osd_req);
3030 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3034 rbd_osd_submit(osd_req);
3038 static int rbd_obj_copyup_current_snapc(struct rbd_obj_request *obj_req,
3041 struct ceph_osd_request *osd_req;
3042 int num_ops = count_write_ops(obj_req);
3046 dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3048 if (bytes != MODS_ONLY)
3049 num_ops++; /* copyup */
3051 osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
3052 if (IS_ERR(osd_req))
3053 return PTR_ERR(osd_req);
3055 if (bytes != MODS_ONLY) {
3056 ret = rbd_osd_setup_copyup(osd_req, which++, bytes);
3061 rbd_osd_setup_write_ops(osd_req, which);
3062 rbd_osd_format_write(osd_req);
3064 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3068 rbd_osd_submit(osd_req);
3072 static int setup_copyup_bvecs(struct rbd_obj_request *obj_req, u64 obj_overlap)
3076 rbd_assert(!obj_req->copyup_bvecs);
3077 obj_req->copyup_bvec_count = calc_pages_for(0, obj_overlap);
3078 obj_req->copyup_bvecs = kcalloc(obj_req->copyup_bvec_count,
3079 sizeof(*obj_req->copyup_bvecs),
3081 if (!obj_req->copyup_bvecs)
3084 for (i = 0; i < obj_req->copyup_bvec_count; i++) {
3085 unsigned int len = min(obj_overlap, (u64)PAGE_SIZE);
3086 struct page *page = alloc_page(GFP_NOIO);
3091 bvec_set_page(&obj_req->copyup_bvecs[i], page, len, 0);
3095 rbd_assert(!obj_overlap);
3100 * The target object doesn't exist. Read the data for the entire
3101 * target object up to the overlap point (if any) from the parent,
3102 * so we can use it for a copyup.
3104 static int rbd_obj_copyup_read_parent(struct rbd_obj_request *obj_req)
3106 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3109 rbd_assert(obj_req->num_img_extents);
3110 prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
3111 rbd_dev->parent_overlap);
3112 if (!obj_req->num_img_extents) {
3114 * The overlap has become 0 (most likely because the
3115 * image has been flattened). Re-submit the original write
3116 * request -- pass MODS_ONLY since the copyup isn't needed
3119 return rbd_obj_copyup_current_snapc(obj_req, MODS_ONLY);
3122 ret = setup_copyup_bvecs(obj_req, rbd_obj_img_extents_bytes(obj_req));
3126 return rbd_obj_read_from_parent(obj_req);
3129 static void rbd_obj_copyup_object_maps(struct rbd_obj_request *obj_req)
3131 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3132 struct ceph_snap_context *snapc = obj_req->img_request->snapc;
3137 rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3139 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3142 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3145 for (i = 0; i < snapc->num_snaps; i++) {
3146 if ((rbd_dev->header.features & RBD_FEATURE_FAST_DIFF) &&
3147 i + 1 < snapc->num_snaps)
3148 new_state = OBJECT_EXISTS_CLEAN;
3150 new_state = OBJECT_EXISTS;
3152 ret = rbd_object_map_update(obj_req, snapc->snaps[i],
3155 obj_req->pending.result = ret;
3160 obj_req->pending.num_pending++;
3164 static void rbd_obj_copyup_write_object(struct rbd_obj_request *obj_req)
3166 u32 bytes = rbd_obj_img_extents_bytes(obj_req);
3169 rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3172 * Only send non-zero copyup data to save some I/O and network
3173 * bandwidth -- zero copyup data is equivalent to the object not
3176 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3179 if (obj_req->img_request->snapc->num_snaps && bytes > 0) {
3181 * Send a copyup request with an empty snapshot context to
3182 * deep-copyup the object through all existing snapshots.
3183 * A second request with the current snapshot context will be
3184 * sent for the actual modification.
3186 ret = rbd_obj_copyup_empty_snapc(obj_req, bytes);
3188 obj_req->pending.result = ret;
3192 obj_req->pending.num_pending++;
3196 ret = rbd_obj_copyup_current_snapc(obj_req, bytes);
3198 obj_req->pending.result = ret;
3202 obj_req->pending.num_pending++;
3205 static bool rbd_obj_advance_copyup(struct rbd_obj_request *obj_req, int *result)
3207 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3211 switch (obj_req->copyup_state) {
3212 case RBD_OBJ_COPYUP_START:
3213 rbd_assert(!*result);
3215 ret = rbd_obj_copyup_read_parent(obj_req);
3220 if (obj_req->num_img_extents)
3221 obj_req->copyup_state = RBD_OBJ_COPYUP_READ_PARENT;
3223 obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3225 case RBD_OBJ_COPYUP_READ_PARENT:
3229 if (is_zero_bvecs(obj_req->copyup_bvecs,
3230 rbd_obj_img_extents_bytes(obj_req))) {
3231 dout("%s %p detected zeros\n", __func__, obj_req);
3232 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ZEROS;
3235 rbd_obj_copyup_object_maps(obj_req);
3236 if (!obj_req->pending.num_pending) {
3237 *result = obj_req->pending.result;
3238 obj_req->copyup_state = RBD_OBJ_COPYUP_OBJECT_MAPS;
3241 obj_req->copyup_state = __RBD_OBJ_COPYUP_OBJECT_MAPS;
3243 case __RBD_OBJ_COPYUP_OBJECT_MAPS:
3244 if (!pending_result_dec(&obj_req->pending, result))
3247 case RBD_OBJ_COPYUP_OBJECT_MAPS:
3249 rbd_warn(rbd_dev, "snap object map update failed: %d",
3254 rbd_obj_copyup_write_object(obj_req);
3255 if (!obj_req->pending.num_pending) {
3256 *result = obj_req->pending.result;
3257 obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3260 obj_req->copyup_state = __RBD_OBJ_COPYUP_WRITE_OBJECT;
3262 case __RBD_OBJ_COPYUP_WRITE_OBJECT:
3263 if (!pending_result_dec(&obj_req->pending, result))
3266 case RBD_OBJ_COPYUP_WRITE_OBJECT:
3275 * 0 - object map update sent
3276 * 1 - object map update isn't needed
3279 static int rbd_obj_write_post_object_map(struct rbd_obj_request *obj_req)
3281 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3282 u8 current_state = OBJECT_PENDING;
3284 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3287 if (!(obj_req->flags & RBD_OBJ_FLAG_DELETION))
3290 return rbd_object_map_update(obj_req, CEPH_NOSNAP, OBJECT_NONEXISTENT,
3294 static bool rbd_obj_advance_write(struct rbd_obj_request *obj_req, int *result)
3296 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3300 switch (obj_req->write_state) {
3301 case RBD_OBJ_WRITE_START:
3302 rbd_assert(!*result);
3304 rbd_obj_set_copyup_enabled(obj_req);
3305 if (rbd_obj_write_is_noop(obj_req))
3308 ret = rbd_obj_write_pre_object_map(obj_req);
3313 obj_req->write_state = RBD_OBJ_WRITE_PRE_OBJECT_MAP;
3317 case RBD_OBJ_WRITE_PRE_OBJECT_MAP:
3319 rbd_warn(rbd_dev, "pre object map update failed: %d",
3323 ret = rbd_obj_write_object(obj_req);
3328 obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
3330 case RBD_OBJ_WRITE_OBJECT:
3331 if (*result == -ENOENT) {
3332 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
3334 obj_req->copyup_state = RBD_OBJ_COPYUP_START;
3335 obj_req->write_state = __RBD_OBJ_WRITE_COPYUP;
3339 * On a non-existent object:
3340 * delete - -ENOENT, truncate/zero - 0
3342 if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
3348 obj_req->write_state = RBD_OBJ_WRITE_COPYUP;
3350 case __RBD_OBJ_WRITE_COPYUP:
3351 if (!rbd_obj_advance_copyup(obj_req, result))
3354 case RBD_OBJ_WRITE_COPYUP:
3356 rbd_warn(rbd_dev, "copyup failed: %d", *result);
3359 ret = rbd_obj_write_post_object_map(obj_req);
3364 obj_req->write_state = RBD_OBJ_WRITE_POST_OBJECT_MAP;
3368 case RBD_OBJ_WRITE_POST_OBJECT_MAP:
3370 rbd_warn(rbd_dev, "post object map update failed: %d",
3379 * Return true if @obj_req is completed.
3381 static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
3384 struct rbd_img_request *img_req = obj_req->img_request;
3385 struct rbd_device *rbd_dev = img_req->rbd_dev;
3388 mutex_lock(&obj_req->state_mutex);
3389 if (!rbd_img_is_write(img_req))
3390 done = rbd_obj_advance_read(obj_req, result);
3392 done = rbd_obj_advance_write(obj_req, result);
3393 mutex_unlock(&obj_req->state_mutex);
3395 if (done && *result) {
3396 rbd_assert(*result < 0);
3397 rbd_warn(rbd_dev, "%s at objno %llu %llu~%llu result %d",
3398 obj_op_name(img_req->op_type), obj_req->ex.oe_objno,
3399 obj_req->ex.oe_off, obj_req->ex.oe_len, *result);
3405 * This is open-coded in rbd_img_handle_request() to avoid parent chain
3408 static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result)
3410 if (__rbd_obj_handle_request(obj_req, &result))
3411 rbd_img_handle_request(obj_req->img_request, result);
3414 static bool need_exclusive_lock(struct rbd_img_request *img_req)
3416 struct rbd_device *rbd_dev = img_req->rbd_dev;
3418 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK))
3421 if (rbd_is_ro(rbd_dev))
3424 rbd_assert(!test_bit(IMG_REQ_CHILD, &img_req->flags));
3425 if (rbd_dev->opts->lock_on_read ||
3426 (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3429 return rbd_img_is_write(img_req);
3432 static bool rbd_lock_add_request(struct rbd_img_request *img_req)
3434 struct rbd_device *rbd_dev = img_req->rbd_dev;
3437 lockdep_assert_held(&rbd_dev->lock_rwsem);
3438 locked = rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED;
3439 spin_lock(&rbd_dev->lock_lists_lock);
3440 rbd_assert(list_empty(&img_req->lock_item));
3442 list_add_tail(&img_req->lock_item, &rbd_dev->acquiring_list);
3444 list_add_tail(&img_req->lock_item, &rbd_dev->running_list);
3445 spin_unlock(&rbd_dev->lock_lists_lock);
3449 static void rbd_lock_del_request(struct rbd_img_request *img_req)
3451 struct rbd_device *rbd_dev = img_req->rbd_dev;
3454 lockdep_assert_held(&rbd_dev->lock_rwsem);
3455 spin_lock(&rbd_dev->lock_lists_lock);
3456 rbd_assert(!list_empty(&img_req->lock_item));
3457 list_del_init(&img_req->lock_item);
3458 need_wakeup = (rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING &&
3459 list_empty(&rbd_dev->running_list));
3460 spin_unlock(&rbd_dev->lock_lists_lock);
3462 complete(&rbd_dev->releasing_wait);
3465 static int rbd_img_exclusive_lock(struct rbd_img_request *img_req)
3467 struct rbd_device *rbd_dev = img_req->rbd_dev;
3469 if (!need_exclusive_lock(img_req))
3472 if (rbd_lock_add_request(img_req))
3475 if (rbd_dev->opts->exclusive) {
3476 WARN_ON(1); /* lock got released? */
3481 * Note the use of mod_delayed_work() in rbd_acquire_lock()
3482 * and cancel_delayed_work() in wake_lock_waiters().
3484 dout("%s rbd_dev %p queueing lock_dwork\n", __func__, rbd_dev);
3485 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
3489 static void rbd_img_object_requests(struct rbd_img_request *img_req)
3491 struct rbd_device *rbd_dev = img_req->rbd_dev;
3492 struct rbd_obj_request *obj_req;
3494 rbd_assert(!img_req->pending.result && !img_req->pending.num_pending);
3495 rbd_assert(!need_exclusive_lock(img_req) ||
3496 __rbd_is_lock_owner(rbd_dev));
3498 if (rbd_img_is_write(img_req)) {
3499 rbd_assert(!img_req->snapc);
3500 down_read(&rbd_dev->header_rwsem);
3501 img_req->snapc = ceph_get_snap_context(rbd_dev->header.snapc);
3502 up_read(&rbd_dev->header_rwsem);
3505 for_each_obj_request(img_req, obj_req) {
3508 if (__rbd_obj_handle_request(obj_req, &result)) {
3510 img_req->pending.result = result;
3514 img_req->pending.num_pending++;
3519 static bool rbd_img_advance(struct rbd_img_request *img_req, int *result)
3524 switch (img_req->state) {
3526 rbd_assert(!*result);
3528 ret = rbd_img_exclusive_lock(img_req);
3533 img_req->state = RBD_IMG_EXCLUSIVE_LOCK;
3537 case RBD_IMG_EXCLUSIVE_LOCK:
3541 rbd_img_object_requests(img_req);
3542 if (!img_req->pending.num_pending) {
3543 *result = img_req->pending.result;
3544 img_req->state = RBD_IMG_OBJECT_REQUESTS;
3547 img_req->state = __RBD_IMG_OBJECT_REQUESTS;
3549 case __RBD_IMG_OBJECT_REQUESTS:
3550 if (!pending_result_dec(&img_req->pending, result))
3553 case RBD_IMG_OBJECT_REQUESTS:
3561 * Return true if @img_req is completed.
3563 static bool __rbd_img_handle_request(struct rbd_img_request *img_req,
3566 struct rbd_device *rbd_dev = img_req->rbd_dev;
3569 if (need_exclusive_lock(img_req)) {
3570 down_read(&rbd_dev->lock_rwsem);
3571 mutex_lock(&img_req->state_mutex);
3572 done = rbd_img_advance(img_req, result);
3574 rbd_lock_del_request(img_req);
3575 mutex_unlock(&img_req->state_mutex);
3576 up_read(&rbd_dev->lock_rwsem);
3578 mutex_lock(&img_req->state_mutex);
3579 done = rbd_img_advance(img_req, result);
3580 mutex_unlock(&img_req->state_mutex);
3583 if (done && *result) {
3584 rbd_assert(*result < 0);
3585 rbd_warn(rbd_dev, "%s%s result %d",
3586 test_bit(IMG_REQ_CHILD, &img_req->flags) ? "child " : "",
3587 obj_op_name(img_req->op_type), *result);
3592 static void rbd_img_handle_request(struct rbd_img_request *img_req, int result)
3595 if (!__rbd_img_handle_request(img_req, &result))
3598 if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
3599 struct rbd_obj_request *obj_req = img_req->obj_request;
3601 rbd_img_request_destroy(img_req);
3602 if (__rbd_obj_handle_request(obj_req, &result)) {
3603 img_req = obj_req->img_request;
3607 struct request *rq = blk_mq_rq_from_pdu(img_req);
3609 rbd_img_request_destroy(img_req);
3610 blk_mq_end_request(rq, errno_to_blk_status(result));
3614 static const struct rbd_client_id rbd_empty_cid;
3616 static bool rbd_cid_equal(const struct rbd_client_id *lhs,
3617 const struct rbd_client_id *rhs)
3619 return lhs->gid == rhs->gid && lhs->handle == rhs->handle;
3622 static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev)
3624 struct rbd_client_id cid;
3626 mutex_lock(&rbd_dev->watch_mutex);
3627 cid.gid = ceph_client_gid(rbd_dev->rbd_client->client);
3628 cid.handle = rbd_dev->watch_cookie;
3629 mutex_unlock(&rbd_dev->watch_mutex);
3634 * lock_rwsem must be held for write
3636 static void rbd_set_owner_cid(struct rbd_device *rbd_dev,
3637 const struct rbd_client_id *cid)
3639 dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev,
3640 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle,
3641 cid->gid, cid->handle);
3642 rbd_dev->owner_cid = *cid; /* struct */
3645 static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf)
3647 mutex_lock(&rbd_dev->watch_mutex);
3648 sprintf(buf, "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie);
3649 mutex_unlock(&rbd_dev->watch_mutex);
3652 static void __rbd_lock(struct rbd_device *rbd_dev, const char *cookie)
3654 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3656 rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED;
3657 strcpy(rbd_dev->lock_cookie, cookie);
3658 rbd_set_owner_cid(rbd_dev, &cid);
3659 queue_work(rbd_dev->task_wq, &rbd_dev->acquired_lock_work);
3663 * lock_rwsem must be held for write
3665 static int rbd_lock(struct rbd_device *rbd_dev)
3667 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3671 WARN_ON(__rbd_is_lock_owner(rbd_dev) ||
3672 rbd_dev->lock_cookie[0] != '\0');
3674 format_lock_cookie(rbd_dev, cookie);
3675 ret = ceph_cls_lock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3676 RBD_LOCK_NAME, CEPH_CLS_LOCK_EXCLUSIVE, cookie,
3677 RBD_LOCK_TAG, "", 0);
3681 __rbd_lock(rbd_dev, cookie);
3686 * lock_rwsem must be held for write
3688 static void rbd_unlock(struct rbd_device *rbd_dev)
3690 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3693 WARN_ON(!__rbd_is_lock_owner(rbd_dev) ||
3694 rbd_dev->lock_cookie[0] == '\0');
3696 ret = ceph_cls_unlock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3697 RBD_LOCK_NAME, rbd_dev->lock_cookie);
3698 if (ret && ret != -ENOENT)
3699 rbd_warn(rbd_dev, "failed to unlock header: %d", ret);
3701 /* treat errors as the image is unlocked */
3702 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
3703 rbd_dev->lock_cookie[0] = '\0';
3704 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3705 queue_work(rbd_dev->task_wq, &rbd_dev->released_lock_work);
3708 static int __rbd_notify_op_lock(struct rbd_device *rbd_dev,
3709 enum rbd_notify_op notify_op,
3710 struct page ***preply_pages,
3713 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3714 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3715 char buf[4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN];
3716 int buf_size = sizeof(buf);
3719 dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op);
3721 /* encode *LockPayload NotifyMessage (op + ClientId) */
3722 ceph_start_encoding(&p, 2, 1, buf_size - CEPH_ENCODING_START_BLK_LEN);
3723 ceph_encode_32(&p, notify_op);
3724 ceph_encode_64(&p, cid.gid);
3725 ceph_encode_64(&p, cid.handle);
3727 return ceph_osdc_notify(osdc, &rbd_dev->header_oid,
3728 &rbd_dev->header_oloc, buf, buf_size,
3729 RBD_NOTIFY_TIMEOUT, preply_pages, preply_len);
3732 static void rbd_notify_op_lock(struct rbd_device *rbd_dev,
3733 enum rbd_notify_op notify_op)
3735 __rbd_notify_op_lock(rbd_dev, notify_op, NULL, NULL);
3738 static void rbd_notify_acquired_lock(struct work_struct *work)
3740 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3741 acquired_lock_work);
3743 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_ACQUIRED_LOCK);
3746 static void rbd_notify_released_lock(struct work_struct *work)
3748 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3749 released_lock_work);
3751 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_RELEASED_LOCK);
3754 static int rbd_request_lock(struct rbd_device *rbd_dev)
3756 struct page **reply_pages;
3758 bool lock_owner_responded = false;
3761 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3763 ret = __rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_REQUEST_LOCK,
3764 &reply_pages, &reply_len);
3765 if (ret && ret != -ETIMEDOUT) {
3766 rbd_warn(rbd_dev, "failed to request lock: %d", ret);
3770 if (reply_len > 0 && reply_len <= PAGE_SIZE) {
3771 void *p = page_address(reply_pages[0]);
3772 void *const end = p + reply_len;
3775 ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */
3780 ceph_decode_need(&p, end, 8 + 8, e_inval);
3781 p += 8 + 8; /* skip gid and cookie */
3783 ceph_decode_32_safe(&p, end, len, e_inval);
3787 if (lock_owner_responded) {
3789 "duplicate lock owners detected");
3794 lock_owner_responded = true;
3795 ret = ceph_start_decoding(&p, end, 1, "ResponseMessage",
3799 "failed to decode ResponseMessage: %d",
3804 ret = ceph_decode_32(&p);
3808 if (!lock_owner_responded) {
3809 rbd_warn(rbd_dev, "no lock owners detected");
3814 ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3823 * Either image request state machine(s) or rbd_add_acquire_lock()
3826 static void wake_lock_waiters(struct rbd_device *rbd_dev, int result)
3828 struct rbd_img_request *img_req;
3830 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
3831 lockdep_assert_held_write(&rbd_dev->lock_rwsem);
3833 cancel_delayed_work(&rbd_dev->lock_dwork);
3834 if (!completion_done(&rbd_dev->acquire_wait)) {
3835 rbd_assert(list_empty(&rbd_dev->acquiring_list) &&
3836 list_empty(&rbd_dev->running_list));
3837 rbd_dev->acquire_err = result;
3838 complete_all(&rbd_dev->acquire_wait);
3842 list_for_each_entry(img_req, &rbd_dev->acquiring_list, lock_item) {
3843 mutex_lock(&img_req->state_mutex);
3844 rbd_assert(img_req->state == RBD_IMG_EXCLUSIVE_LOCK);
3845 rbd_img_schedule(img_req, result);
3846 mutex_unlock(&img_req->state_mutex);
3849 list_splice_tail_init(&rbd_dev->acquiring_list, &rbd_dev->running_list);
3852 static int get_lock_owner_info(struct rbd_device *rbd_dev,
3853 struct ceph_locker **lockers, u32 *num_lockers)
3855 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3860 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3862 ret = ceph_cls_lock_info(osdc, &rbd_dev->header_oid,
3863 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3864 &lock_type, &lock_tag, lockers, num_lockers);
3868 if (*num_lockers == 0) {
3869 dout("%s rbd_dev %p no lockers detected\n", __func__, rbd_dev);
3873 if (strcmp(lock_tag, RBD_LOCK_TAG)) {
3874 rbd_warn(rbd_dev, "locked by external mechanism, tag %s",
3880 if (lock_type == CEPH_CLS_LOCK_SHARED) {
3881 rbd_warn(rbd_dev, "shared lock type detected");
3886 if (strncmp((*lockers)[0].id.cookie, RBD_LOCK_COOKIE_PREFIX,
3887 strlen(RBD_LOCK_COOKIE_PREFIX))) {
3888 rbd_warn(rbd_dev, "locked by external mechanism, cookie %s",
3889 (*lockers)[0].id.cookie);
3899 static int find_watcher(struct rbd_device *rbd_dev,
3900 const struct ceph_locker *locker)
3902 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3903 struct ceph_watch_item *watchers;
3909 ret = ceph_osdc_list_watchers(osdc, &rbd_dev->header_oid,
3910 &rbd_dev->header_oloc, &watchers,
3915 sscanf(locker->id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu", &cookie);
3916 for (i = 0; i < num_watchers; i++) {
3918 * Ignore addr->type while comparing. This mimics
3919 * entity_addr_t::get_legacy_str() + strcmp().
3921 if (ceph_addr_equal_no_type(&watchers[i].addr,
3922 &locker->info.addr) &&
3923 watchers[i].cookie == cookie) {
3924 struct rbd_client_id cid = {
3925 .gid = le64_to_cpu(watchers[i].name.num),
3929 dout("%s rbd_dev %p found cid %llu-%llu\n", __func__,
3930 rbd_dev, cid.gid, cid.handle);
3931 rbd_set_owner_cid(rbd_dev, &cid);
3937 dout("%s rbd_dev %p no watchers\n", __func__, rbd_dev);
3945 * lock_rwsem must be held for write
3947 static int rbd_try_lock(struct rbd_device *rbd_dev)
3949 struct ceph_client *client = rbd_dev->rbd_client->client;
3950 struct ceph_locker *lockers;
3955 ret = rbd_lock(rbd_dev);
3959 /* determine if the current lock holder is still alive */
3960 ret = get_lock_owner_info(rbd_dev, &lockers, &num_lockers);
3964 if (num_lockers == 0)
3967 ret = find_watcher(rbd_dev, lockers);
3969 goto out; /* request lock or error */
3971 rbd_warn(rbd_dev, "breaking header lock owned by %s%llu",
3972 ENTITY_NAME(lockers[0].id.name));
3974 ret = ceph_monc_blocklist_add(&client->monc,
3975 &lockers[0].info.addr);
3977 rbd_warn(rbd_dev, "blocklist of %s%llu failed: %d",
3978 ENTITY_NAME(lockers[0].id.name), ret);
3982 ret = ceph_cls_break_lock(&client->osdc, &rbd_dev->header_oid,
3983 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3984 lockers[0].id.cookie,
3985 &lockers[0].id.name);
3986 if (ret && ret != -ENOENT)
3990 ceph_free_lockers(lockers, num_lockers);
3994 ceph_free_lockers(lockers, num_lockers);
3998 static int rbd_post_acquire_action(struct rbd_device *rbd_dev)
4002 ret = rbd_dev_refresh(rbd_dev);
4006 if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) {
4007 ret = rbd_object_map_open(rbd_dev);
4018 * 1 - caller should call rbd_request_lock()
4021 static int rbd_try_acquire_lock(struct rbd_device *rbd_dev)
4025 down_read(&rbd_dev->lock_rwsem);
4026 dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
4027 rbd_dev->lock_state);
4028 if (__rbd_is_lock_owner(rbd_dev)) {
4029 up_read(&rbd_dev->lock_rwsem);
4033 up_read(&rbd_dev->lock_rwsem);
4034 down_write(&rbd_dev->lock_rwsem);
4035 dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
4036 rbd_dev->lock_state);
4037 if (__rbd_is_lock_owner(rbd_dev)) {
4038 up_write(&rbd_dev->lock_rwsem);
4042 ret = rbd_try_lock(rbd_dev);
4044 rbd_warn(rbd_dev, "failed to lock header: %d", ret);
4045 if (ret == -EBLOCKLISTED)
4048 ret = 1; /* request lock anyway */
4051 up_write(&rbd_dev->lock_rwsem);
4055 rbd_assert(rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED);
4056 rbd_assert(list_empty(&rbd_dev->running_list));
4058 ret = rbd_post_acquire_action(rbd_dev);
4060 rbd_warn(rbd_dev, "post-acquire action failed: %d", ret);
4062 * Can't stay in RBD_LOCK_STATE_LOCKED because
4063 * rbd_lock_add_request() would let the request through,
4064 * assuming that e.g. object map is locked and loaded.
4066 rbd_unlock(rbd_dev);
4070 wake_lock_waiters(rbd_dev, ret);
4071 up_write(&rbd_dev->lock_rwsem);
4075 static void rbd_acquire_lock(struct work_struct *work)
4077 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4078 struct rbd_device, lock_dwork);
4081 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4083 ret = rbd_try_acquire_lock(rbd_dev);
4085 dout("%s rbd_dev %p ret %d - done\n", __func__, rbd_dev, ret);
4089 ret = rbd_request_lock(rbd_dev);
4090 if (ret == -ETIMEDOUT) {
4091 goto again; /* treat this as a dead client */
4092 } else if (ret == -EROFS) {
4093 rbd_warn(rbd_dev, "peer will not release lock");
4094 down_write(&rbd_dev->lock_rwsem);
4095 wake_lock_waiters(rbd_dev, ret);
4096 up_write(&rbd_dev->lock_rwsem);
4097 } else if (ret < 0) {
4098 rbd_warn(rbd_dev, "error requesting lock: %d", ret);
4099 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4103 * lock owner acked, but resend if we don't see them
4106 dout("%s rbd_dev %p requeuing lock_dwork\n", __func__,
4108 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4109 msecs_to_jiffies(2 * RBD_NOTIFY_TIMEOUT * MSEC_PER_SEC));
4113 static bool rbd_quiesce_lock(struct rbd_device *rbd_dev)
4115 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4116 lockdep_assert_held_write(&rbd_dev->lock_rwsem);
4118 if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED)
4122 * Ensure that all in-flight IO is flushed.
4124 rbd_dev->lock_state = RBD_LOCK_STATE_RELEASING;
4125 rbd_assert(!completion_done(&rbd_dev->releasing_wait));
4126 if (list_empty(&rbd_dev->running_list))
4129 up_write(&rbd_dev->lock_rwsem);
4130 wait_for_completion(&rbd_dev->releasing_wait);
4132 down_write(&rbd_dev->lock_rwsem);
4133 if (rbd_dev->lock_state != RBD_LOCK_STATE_RELEASING)
4136 rbd_assert(list_empty(&rbd_dev->running_list));
4140 static void rbd_pre_release_action(struct rbd_device *rbd_dev)
4142 if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)
4143 rbd_object_map_close(rbd_dev);
4146 static void __rbd_release_lock(struct rbd_device *rbd_dev)
4148 rbd_assert(list_empty(&rbd_dev->running_list));
4150 rbd_pre_release_action(rbd_dev);
4151 rbd_unlock(rbd_dev);
4155 * lock_rwsem must be held for write
4157 static void rbd_release_lock(struct rbd_device *rbd_dev)
4159 if (!rbd_quiesce_lock(rbd_dev))
4162 __rbd_release_lock(rbd_dev);
4165 * Give others a chance to grab the lock - we would re-acquire
4166 * almost immediately if we got new IO while draining the running
4167 * list otherwise. We need to ack our own notifications, so this
4168 * lock_dwork will be requeued from rbd_handle_released_lock() by
4169 * way of maybe_kick_acquire().
4171 cancel_delayed_work(&rbd_dev->lock_dwork);
4174 static void rbd_release_lock_work(struct work_struct *work)
4176 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
4179 down_write(&rbd_dev->lock_rwsem);
4180 rbd_release_lock(rbd_dev);
4181 up_write(&rbd_dev->lock_rwsem);
4184 static void maybe_kick_acquire(struct rbd_device *rbd_dev)
4188 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4189 if (__rbd_is_lock_owner(rbd_dev))
4192 spin_lock(&rbd_dev->lock_lists_lock);
4193 have_requests = !list_empty(&rbd_dev->acquiring_list);
4194 spin_unlock(&rbd_dev->lock_lists_lock);
4195 if (have_requests || delayed_work_pending(&rbd_dev->lock_dwork)) {
4196 dout("%s rbd_dev %p kicking lock_dwork\n", __func__, rbd_dev);
4197 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4201 static void rbd_handle_acquired_lock(struct rbd_device *rbd_dev, u8 struct_v,
4204 struct rbd_client_id cid = { 0 };
4206 if (struct_v >= 2) {
4207 cid.gid = ceph_decode_64(p);
4208 cid.handle = ceph_decode_64(p);
4211 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4213 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4214 down_write(&rbd_dev->lock_rwsem);
4215 if (rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4216 dout("%s rbd_dev %p cid %llu-%llu == owner_cid\n",
4217 __func__, rbd_dev, cid.gid, cid.handle);
4219 rbd_set_owner_cid(rbd_dev, &cid);
4221 downgrade_write(&rbd_dev->lock_rwsem);
4223 down_read(&rbd_dev->lock_rwsem);
4226 maybe_kick_acquire(rbd_dev);
4227 up_read(&rbd_dev->lock_rwsem);
4230 static void rbd_handle_released_lock(struct rbd_device *rbd_dev, u8 struct_v,
4233 struct rbd_client_id cid = { 0 };
4235 if (struct_v >= 2) {
4236 cid.gid = ceph_decode_64(p);
4237 cid.handle = ceph_decode_64(p);
4240 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4242 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4243 down_write(&rbd_dev->lock_rwsem);
4244 if (!rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4245 dout("%s rbd_dev %p cid %llu-%llu != owner_cid %llu-%llu\n",
4246 __func__, rbd_dev, cid.gid, cid.handle,
4247 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle);
4249 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4251 downgrade_write(&rbd_dev->lock_rwsem);
4253 down_read(&rbd_dev->lock_rwsem);
4256 maybe_kick_acquire(rbd_dev);
4257 up_read(&rbd_dev->lock_rwsem);
4261 * Returns result for ResponseMessage to be encoded (<= 0), or 1 if no
4262 * ResponseMessage is needed.
4264 static int rbd_handle_request_lock(struct rbd_device *rbd_dev, u8 struct_v,
4267 struct rbd_client_id my_cid = rbd_get_cid(rbd_dev);
4268 struct rbd_client_id cid = { 0 };
4271 if (struct_v >= 2) {
4272 cid.gid = ceph_decode_64(p);
4273 cid.handle = ceph_decode_64(p);
4276 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4278 if (rbd_cid_equal(&cid, &my_cid))
4281 down_read(&rbd_dev->lock_rwsem);
4282 if (__rbd_is_lock_owner(rbd_dev)) {
4283 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED &&
4284 rbd_cid_equal(&rbd_dev->owner_cid, &rbd_empty_cid))
4288 * encode ResponseMessage(0) so the peer can detect
4293 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED) {
4294 if (!rbd_dev->opts->exclusive) {
4295 dout("%s rbd_dev %p queueing unlock_work\n",
4297 queue_work(rbd_dev->task_wq,
4298 &rbd_dev->unlock_work);
4300 /* refuse to release the lock */
4307 up_read(&rbd_dev->lock_rwsem);
4311 static void __rbd_acknowledge_notify(struct rbd_device *rbd_dev,
4312 u64 notify_id, u64 cookie, s32 *result)
4314 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4315 char buf[4 + CEPH_ENCODING_START_BLK_LEN];
4316 int buf_size = sizeof(buf);
4322 /* encode ResponseMessage */
4323 ceph_start_encoding(&p, 1, 1,
4324 buf_size - CEPH_ENCODING_START_BLK_LEN);
4325 ceph_encode_32(&p, *result);
4330 ret = ceph_osdc_notify_ack(osdc, &rbd_dev->header_oid,
4331 &rbd_dev->header_oloc, notify_id, cookie,
4334 rbd_warn(rbd_dev, "acknowledge_notify failed: %d", ret);
4337 static void rbd_acknowledge_notify(struct rbd_device *rbd_dev, u64 notify_id,
4340 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4341 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, NULL);
4344 static void rbd_acknowledge_notify_result(struct rbd_device *rbd_dev,
4345 u64 notify_id, u64 cookie, s32 result)
4347 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
4348 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, &result);
4351 static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie,
4352 u64 notifier_id, void *data, size_t data_len)
4354 struct rbd_device *rbd_dev = arg;
4356 void *const end = p + data_len;
4362 dout("%s rbd_dev %p cookie %llu notify_id %llu data_len %zu\n",
4363 __func__, rbd_dev, cookie, notify_id, data_len);
4365 ret = ceph_start_decoding(&p, end, 1, "NotifyMessage",
4368 rbd_warn(rbd_dev, "failed to decode NotifyMessage: %d",
4373 notify_op = ceph_decode_32(&p);
4375 /* legacy notification for header updates */
4376 notify_op = RBD_NOTIFY_OP_HEADER_UPDATE;
4380 dout("%s rbd_dev %p notify_op %u\n", __func__, rbd_dev, notify_op);
4381 switch (notify_op) {
4382 case RBD_NOTIFY_OP_ACQUIRED_LOCK:
4383 rbd_handle_acquired_lock(rbd_dev, struct_v, &p);
4384 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4386 case RBD_NOTIFY_OP_RELEASED_LOCK:
4387 rbd_handle_released_lock(rbd_dev, struct_v, &p);
4388 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4390 case RBD_NOTIFY_OP_REQUEST_LOCK:
4391 ret = rbd_handle_request_lock(rbd_dev, struct_v, &p);
4393 rbd_acknowledge_notify_result(rbd_dev, notify_id,
4396 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4398 case RBD_NOTIFY_OP_HEADER_UPDATE:
4399 ret = rbd_dev_refresh(rbd_dev);
4401 rbd_warn(rbd_dev, "refresh failed: %d", ret);
4403 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4406 if (rbd_is_lock_owner(rbd_dev))
4407 rbd_acknowledge_notify_result(rbd_dev, notify_id,
4408 cookie, -EOPNOTSUPP);
4410 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4415 static void __rbd_unregister_watch(struct rbd_device *rbd_dev);
4417 static void rbd_watch_errcb(void *arg, u64 cookie, int err)
4419 struct rbd_device *rbd_dev = arg;
4421 rbd_warn(rbd_dev, "encountered watch error: %d", err);
4423 down_write(&rbd_dev->lock_rwsem);
4424 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4425 up_write(&rbd_dev->lock_rwsem);
4427 mutex_lock(&rbd_dev->watch_mutex);
4428 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED) {
4429 __rbd_unregister_watch(rbd_dev);
4430 rbd_dev->watch_state = RBD_WATCH_STATE_ERROR;
4432 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->watch_dwork, 0);
4434 mutex_unlock(&rbd_dev->watch_mutex);
4438 * watch_mutex must be locked
4440 static int __rbd_register_watch(struct rbd_device *rbd_dev)
4442 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4443 struct ceph_osd_linger_request *handle;
4445 rbd_assert(!rbd_dev->watch_handle);
4446 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4448 handle = ceph_osdc_watch(osdc, &rbd_dev->header_oid,
4449 &rbd_dev->header_oloc, rbd_watch_cb,
4450 rbd_watch_errcb, rbd_dev);
4452 return PTR_ERR(handle);
4454 rbd_dev->watch_handle = handle;
4459 * watch_mutex must be locked
4461 static void __rbd_unregister_watch(struct rbd_device *rbd_dev)
4463 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4466 rbd_assert(rbd_dev->watch_handle);
4467 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4469 ret = ceph_osdc_unwatch(osdc, rbd_dev->watch_handle);
4471 rbd_warn(rbd_dev, "failed to unwatch: %d", ret);
4473 rbd_dev->watch_handle = NULL;
4476 static int rbd_register_watch(struct rbd_device *rbd_dev)
4480 mutex_lock(&rbd_dev->watch_mutex);
4481 rbd_assert(rbd_dev->watch_state == RBD_WATCH_STATE_UNREGISTERED);
4482 ret = __rbd_register_watch(rbd_dev);
4486 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4487 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4490 mutex_unlock(&rbd_dev->watch_mutex);
4494 static void cancel_tasks_sync(struct rbd_device *rbd_dev)
4496 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4498 cancel_work_sync(&rbd_dev->acquired_lock_work);
4499 cancel_work_sync(&rbd_dev->released_lock_work);
4500 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
4501 cancel_work_sync(&rbd_dev->unlock_work);
4505 * header_rwsem must not be held to avoid a deadlock with
4506 * rbd_dev_refresh() when flushing notifies.
4508 static void rbd_unregister_watch(struct rbd_device *rbd_dev)
4510 cancel_tasks_sync(rbd_dev);
4512 mutex_lock(&rbd_dev->watch_mutex);
4513 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED)
4514 __rbd_unregister_watch(rbd_dev);
4515 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
4516 mutex_unlock(&rbd_dev->watch_mutex);
4518 cancel_delayed_work_sync(&rbd_dev->watch_dwork);
4519 ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
4523 * lock_rwsem must be held for write
4525 static void rbd_reacquire_lock(struct rbd_device *rbd_dev)
4527 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4531 if (!rbd_quiesce_lock(rbd_dev))
4534 format_lock_cookie(rbd_dev, cookie);
4535 ret = ceph_cls_set_cookie(osdc, &rbd_dev->header_oid,
4536 &rbd_dev->header_oloc, RBD_LOCK_NAME,
4537 CEPH_CLS_LOCK_EXCLUSIVE, rbd_dev->lock_cookie,
4538 RBD_LOCK_TAG, cookie);
4540 if (ret != -EOPNOTSUPP)
4541 rbd_warn(rbd_dev, "failed to update lock cookie: %d",
4545 * Lock cookie cannot be updated on older OSDs, so do
4546 * a manual release and queue an acquire.
4548 __rbd_release_lock(rbd_dev);
4549 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4551 __rbd_lock(rbd_dev, cookie);
4552 wake_lock_waiters(rbd_dev, 0);
4556 static void rbd_reregister_watch(struct work_struct *work)
4558 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4559 struct rbd_device, watch_dwork);
4562 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4564 mutex_lock(&rbd_dev->watch_mutex);
4565 if (rbd_dev->watch_state != RBD_WATCH_STATE_ERROR) {
4566 mutex_unlock(&rbd_dev->watch_mutex);
4570 ret = __rbd_register_watch(rbd_dev);
4572 rbd_warn(rbd_dev, "failed to reregister watch: %d", ret);
4573 if (ret != -EBLOCKLISTED && ret != -ENOENT) {
4574 queue_delayed_work(rbd_dev->task_wq,
4575 &rbd_dev->watch_dwork,
4577 mutex_unlock(&rbd_dev->watch_mutex);
4581 mutex_unlock(&rbd_dev->watch_mutex);
4582 down_write(&rbd_dev->lock_rwsem);
4583 wake_lock_waiters(rbd_dev, ret);
4584 up_write(&rbd_dev->lock_rwsem);
4588 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4589 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4590 mutex_unlock(&rbd_dev->watch_mutex);
4592 down_write(&rbd_dev->lock_rwsem);
4593 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED)
4594 rbd_reacquire_lock(rbd_dev);
4595 up_write(&rbd_dev->lock_rwsem);
4597 ret = rbd_dev_refresh(rbd_dev);
4599 rbd_warn(rbd_dev, "reregistration refresh failed: %d", ret);
4603 * Synchronous osd object method call. Returns the number of bytes
4604 * returned in the outbound buffer, or a negative error code.
4606 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
4607 struct ceph_object_id *oid,
4608 struct ceph_object_locator *oloc,
4609 const char *method_name,
4610 const void *outbound,
4611 size_t outbound_size,
4613 size_t inbound_size)
4615 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4616 struct page *req_page = NULL;
4617 struct page *reply_page;
4621 * Method calls are ultimately read operations. The result
4622 * should placed into the inbound buffer provided. They
4623 * also supply outbound data--parameters for the object
4624 * method. Currently if this is present it will be a
4628 if (outbound_size > PAGE_SIZE)
4631 req_page = alloc_page(GFP_KERNEL);
4635 memcpy(page_address(req_page), outbound, outbound_size);
4638 reply_page = alloc_page(GFP_KERNEL);
4641 __free_page(req_page);
4645 ret = ceph_osdc_call(osdc, oid, oloc, RBD_DRV_NAME, method_name,
4646 CEPH_OSD_FLAG_READ, req_page, outbound_size,
4647 &reply_page, &inbound_size);
4649 memcpy(inbound, page_address(reply_page), inbound_size);
4654 __free_page(req_page);
4655 __free_page(reply_page);
4659 static void rbd_queue_workfn(struct work_struct *work)
4661 struct rbd_img_request *img_request =
4662 container_of(work, struct rbd_img_request, work);
4663 struct rbd_device *rbd_dev = img_request->rbd_dev;
4664 enum obj_operation_type op_type = img_request->op_type;
4665 struct request *rq = blk_mq_rq_from_pdu(img_request);
4666 u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
4667 u64 length = blk_rq_bytes(rq);
4671 /* Ignore/skip any zero-length requests */
4673 dout("%s: zero-length request\n", __func__);
4675 goto err_img_request;
4678 blk_mq_start_request(rq);
4680 down_read(&rbd_dev->header_rwsem);
4681 mapping_size = rbd_dev->mapping.size;
4682 rbd_img_capture_header(img_request);
4683 up_read(&rbd_dev->header_rwsem);
4685 if (offset + length > mapping_size) {
4686 rbd_warn(rbd_dev, "beyond EOD (%llu~%llu > %llu)", offset,
4687 length, mapping_size);
4689 goto err_img_request;
4692 dout("%s rbd_dev %p img_req %p %s %llu~%llu\n", __func__, rbd_dev,
4693 img_request, obj_op_name(op_type), offset, length);
4695 if (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_ZEROOUT)
4696 result = rbd_img_fill_nodata(img_request, offset, length);
4698 result = rbd_img_fill_from_bio(img_request, offset, length,
4701 goto err_img_request;
4703 rbd_img_handle_request(img_request, 0);
4707 rbd_img_request_destroy(img_request);
4709 rbd_warn(rbd_dev, "%s %llx at %llx result %d",
4710 obj_op_name(op_type), length, offset, result);
4711 blk_mq_end_request(rq, errno_to_blk_status(result));
4714 static blk_status_t rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
4715 const struct blk_mq_queue_data *bd)
4717 struct rbd_device *rbd_dev = hctx->queue->queuedata;
4718 struct rbd_img_request *img_req = blk_mq_rq_to_pdu(bd->rq);
4719 enum obj_operation_type op_type;
4721 switch (req_op(bd->rq)) {
4722 case REQ_OP_DISCARD:
4723 op_type = OBJ_OP_DISCARD;
4725 case REQ_OP_WRITE_ZEROES:
4726 op_type = OBJ_OP_ZEROOUT;
4729 op_type = OBJ_OP_WRITE;
4732 op_type = OBJ_OP_READ;
4735 rbd_warn(rbd_dev, "unknown req_op %d", req_op(bd->rq));
4736 return BLK_STS_IOERR;
4739 rbd_img_request_init(img_req, rbd_dev, op_type);
4741 if (rbd_img_is_write(img_req)) {
4742 if (rbd_is_ro(rbd_dev)) {
4743 rbd_warn(rbd_dev, "%s on read-only mapping",
4744 obj_op_name(img_req->op_type));
4745 return BLK_STS_IOERR;
4747 rbd_assert(!rbd_is_snap(rbd_dev));
4750 INIT_WORK(&img_req->work, rbd_queue_workfn);
4751 queue_work(rbd_wq, &img_req->work);
4755 static void rbd_free_disk(struct rbd_device *rbd_dev)
4757 put_disk(rbd_dev->disk);
4758 blk_mq_free_tag_set(&rbd_dev->tag_set);
4759 rbd_dev->disk = NULL;
4762 static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
4763 struct ceph_object_id *oid,
4764 struct ceph_object_locator *oloc,
4765 void *buf, int buf_len)
4768 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4769 struct ceph_osd_request *req;
4770 struct page **pages;
4771 int num_pages = calc_pages_for(0, buf_len);
4774 req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_KERNEL);
4778 ceph_oid_copy(&req->r_base_oid, oid);
4779 ceph_oloc_copy(&req->r_base_oloc, oloc);
4780 req->r_flags = CEPH_OSD_FLAG_READ;
4782 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
4783 if (IS_ERR(pages)) {
4784 ret = PTR_ERR(pages);
4788 osd_req_op_extent_init(req, 0, CEPH_OSD_OP_READ, 0, buf_len, 0, 0);
4789 osd_req_op_extent_osd_data_pages(req, 0, pages, buf_len, 0, false,
4792 ret = ceph_osdc_alloc_messages(req, GFP_KERNEL);
4796 ceph_osdc_start_request(osdc, req);
4797 ret = ceph_osdc_wait_request(osdc, req);
4799 ceph_copy_from_page_vector(pages, buf, 0, ret);
4802 ceph_osdc_put_request(req);
4807 * Read the complete header for the given rbd device. On successful
4808 * return, the rbd_dev->header field will contain up-to-date
4809 * information about the image.
4811 static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev)
4813 struct rbd_image_header_ondisk *ondisk = NULL;
4820 * The complete header will include an array of its 64-bit
4821 * snapshot ids, followed by the names of those snapshots as
4822 * a contiguous block of NUL-terminated strings. Note that
4823 * the number of snapshots could change by the time we read
4824 * it in, in which case we re-read it.
4831 size = sizeof (*ondisk);
4832 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
4834 ondisk = kmalloc(size, GFP_KERNEL);
4838 ret = rbd_obj_read_sync(rbd_dev, &rbd_dev->header_oid,
4839 &rbd_dev->header_oloc, ondisk, size);
4842 if ((size_t)ret < size) {
4844 rbd_warn(rbd_dev, "short header read (want %zd got %d)",
4848 if (!rbd_dev_ondisk_valid(ondisk)) {
4850 rbd_warn(rbd_dev, "invalid header");
4854 names_size = le64_to_cpu(ondisk->snap_names_len);
4855 want_count = snap_count;
4856 snap_count = le32_to_cpu(ondisk->snap_count);
4857 } while (snap_count != want_count);
4859 ret = rbd_header_from_disk(rbd_dev, ondisk);
4866 static void rbd_dev_update_size(struct rbd_device *rbd_dev)
4871 * If EXISTS is not set, rbd_dev->disk may be NULL, so don't
4872 * try to update its size. If REMOVING is set, updating size
4873 * is just useless work since the device can't be opened.
4875 if (test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags) &&
4876 !test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) {
4877 size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
4878 dout("setting size to %llu sectors", (unsigned long long)size);
4879 set_capacity_and_notify(rbd_dev->disk, size);
4883 static int rbd_dev_refresh(struct rbd_device *rbd_dev)
4888 down_write(&rbd_dev->header_rwsem);
4889 mapping_size = rbd_dev->mapping.size;
4891 ret = rbd_dev_header_info(rbd_dev);
4896 * If there is a parent, see if it has disappeared due to the
4897 * mapped image getting flattened.
4899 if (rbd_dev->parent) {
4900 ret = rbd_dev_v2_parent_info(rbd_dev);
4905 rbd_assert(!rbd_is_snap(rbd_dev));
4906 rbd_dev->mapping.size = rbd_dev->header.image_size;
4909 up_write(&rbd_dev->header_rwsem);
4910 if (!ret && mapping_size != rbd_dev->mapping.size)
4911 rbd_dev_update_size(rbd_dev);
4916 static const struct blk_mq_ops rbd_mq_ops = {
4917 .queue_rq = rbd_queue_rq,
4920 static int rbd_init_disk(struct rbd_device *rbd_dev)
4922 struct gendisk *disk;
4923 struct request_queue *q;
4924 unsigned int objset_bytes =
4925 rbd_dev->layout.object_size * rbd_dev->layout.stripe_count;
4928 memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
4929 rbd_dev->tag_set.ops = &rbd_mq_ops;
4930 rbd_dev->tag_set.queue_depth = rbd_dev->opts->queue_depth;
4931 rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
4932 rbd_dev->tag_set.flags = BLK_MQ_F_SHOULD_MERGE;
4933 rbd_dev->tag_set.nr_hw_queues = num_present_cpus();
4934 rbd_dev->tag_set.cmd_size = sizeof(struct rbd_img_request);
4936 err = blk_mq_alloc_tag_set(&rbd_dev->tag_set);
4940 disk = blk_mq_alloc_disk(&rbd_dev->tag_set, rbd_dev);
4942 err = PTR_ERR(disk);
4947 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
4949 disk->major = rbd_dev->major;
4950 disk->first_minor = rbd_dev->minor;
4952 disk->minors = (1 << RBD_SINGLE_MAJOR_PART_SHIFT);
4954 disk->minors = RBD_MINORS_PER_MAJOR;
4955 disk->fops = &rbd_bd_ops;
4956 disk->private_data = rbd_dev;
4958 blk_queue_flag_set(QUEUE_FLAG_NONROT, q);
4959 /* QUEUE_FLAG_ADD_RANDOM is off by default for blk-mq */
4961 blk_queue_max_hw_sectors(q, objset_bytes >> SECTOR_SHIFT);
4962 q->limits.max_sectors = queue_max_hw_sectors(q);
4963 blk_queue_max_segments(q, USHRT_MAX);
4964 blk_queue_max_segment_size(q, UINT_MAX);
4965 blk_queue_io_min(q, rbd_dev->opts->alloc_size);
4966 blk_queue_io_opt(q, rbd_dev->opts->alloc_size);
4968 if (rbd_dev->opts->trim) {
4969 q->limits.discard_granularity = rbd_dev->opts->alloc_size;
4970 blk_queue_max_discard_sectors(q, objset_bytes >> SECTOR_SHIFT);
4971 blk_queue_max_write_zeroes_sectors(q, objset_bytes >> SECTOR_SHIFT);
4974 if (!ceph_test_opt(rbd_dev->rbd_client->client, NOCRC))
4975 blk_queue_flag_set(QUEUE_FLAG_STABLE_WRITES, q);
4977 rbd_dev->disk = disk;
4981 blk_mq_free_tag_set(&rbd_dev->tag_set);
4989 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
4991 return container_of(dev, struct rbd_device, dev);
4994 static ssize_t rbd_size_show(struct device *dev,
4995 struct device_attribute *attr, char *buf)
4997 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4999 return sprintf(buf, "%llu\n",
5000 (unsigned long long)rbd_dev->mapping.size);
5003 static ssize_t rbd_features_show(struct device *dev,
5004 struct device_attribute *attr, char *buf)
5006 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5008 return sprintf(buf, "0x%016llx\n", rbd_dev->header.features);
5011 static ssize_t rbd_major_show(struct device *dev,
5012 struct device_attribute *attr, char *buf)
5014 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5017 return sprintf(buf, "%d\n", rbd_dev->major);
5019 return sprintf(buf, "(none)\n");
5022 static ssize_t rbd_minor_show(struct device *dev,
5023 struct device_attribute *attr, char *buf)
5025 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5027 return sprintf(buf, "%d\n", rbd_dev->minor);
5030 static ssize_t rbd_client_addr_show(struct device *dev,
5031 struct device_attribute *attr, char *buf)
5033 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5034 struct ceph_entity_addr *client_addr =
5035 ceph_client_addr(rbd_dev->rbd_client->client);
5037 return sprintf(buf, "%pISpc/%u\n", &client_addr->in_addr,
5038 le32_to_cpu(client_addr->nonce));
5041 static ssize_t rbd_client_id_show(struct device *dev,
5042 struct device_attribute *attr, char *buf)
5044 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5046 return sprintf(buf, "client%lld\n",
5047 ceph_client_gid(rbd_dev->rbd_client->client));
5050 static ssize_t rbd_cluster_fsid_show(struct device *dev,
5051 struct device_attribute *attr, char *buf)
5053 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5055 return sprintf(buf, "%pU\n", &rbd_dev->rbd_client->client->fsid);
5058 static ssize_t rbd_config_info_show(struct device *dev,
5059 struct device_attribute *attr, char *buf)
5061 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5063 if (!capable(CAP_SYS_ADMIN))
5066 return sprintf(buf, "%s\n", rbd_dev->config_info);
5069 static ssize_t rbd_pool_show(struct device *dev,
5070 struct device_attribute *attr, char *buf)
5072 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5074 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
5077 static ssize_t rbd_pool_id_show(struct device *dev,
5078 struct device_attribute *attr, char *buf)
5080 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5082 return sprintf(buf, "%llu\n",
5083 (unsigned long long) rbd_dev->spec->pool_id);
5086 static ssize_t rbd_pool_ns_show(struct device *dev,
5087 struct device_attribute *attr, char *buf)
5089 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5091 return sprintf(buf, "%s\n", rbd_dev->spec->pool_ns ?: "");
5094 static ssize_t rbd_name_show(struct device *dev,
5095 struct device_attribute *attr, char *buf)
5097 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5099 if (rbd_dev->spec->image_name)
5100 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
5102 return sprintf(buf, "(unknown)\n");
5105 static ssize_t rbd_image_id_show(struct device *dev,
5106 struct device_attribute *attr, char *buf)
5108 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5110 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
5114 * Shows the name of the currently-mapped snapshot (or
5115 * RBD_SNAP_HEAD_NAME for the base image).
5117 static ssize_t rbd_snap_show(struct device *dev,
5118 struct device_attribute *attr,
5121 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5123 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
5126 static ssize_t rbd_snap_id_show(struct device *dev,
5127 struct device_attribute *attr, char *buf)
5129 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5131 return sprintf(buf, "%llu\n", rbd_dev->spec->snap_id);
5135 * For a v2 image, shows the chain of parent images, separated by empty
5136 * lines. For v1 images or if there is no parent, shows "(no parent
5139 static ssize_t rbd_parent_show(struct device *dev,
5140 struct device_attribute *attr,
5143 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5146 if (!rbd_dev->parent)
5147 return sprintf(buf, "(no parent image)\n");
5149 for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) {
5150 struct rbd_spec *spec = rbd_dev->parent_spec;
5152 count += sprintf(&buf[count], "%s"
5153 "pool_id %llu\npool_name %s\n"
5155 "image_id %s\nimage_name %s\n"
5156 "snap_id %llu\nsnap_name %s\n"
5158 !count ? "" : "\n", /* first? */
5159 spec->pool_id, spec->pool_name,
5160 spec->pool_ns ?: "",
5161 spec->image_id, spec->image_name ?: "(unknown)",
5162 spec->snap_id, spec->snap_name,
5163 rbd_dev->parent_overlap);
5169 static ssize_t rbd_image_refresh(struct device *dev,
5170 struct device_attribute *attr,
5174 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5177 if (!capable(CAP_SYS_ADMIN))
5180 ret = rbd_dev_refresh(rbd_dev);
5187 static DEVICE_ATTR(size, 0444, rbd_size_show, NULL);
5188 static DEVICE_ATTR(features, 0444, rbd_features_show, NULL);
5189 static DEVICE_ATTR(major, 0444, rbd_major_show, NULL);
5190 static DEVICE_ATTR(minor, 0444, rbd_minor_show, NULL);
5191 static DEVICE_ATTR(client_addr, 0444, rbd_client_addr_show, NULL);
5192 static DEVICE_ATTR(client_id, 0444, rbd_client_id_show, NULL);
5193 static DEVICE_ATTR(cluster_fsid, 0444, rbd_cluster_fsid_show, NULL);
5194 static DEVICE_ATTR(config_info, 0400, rbd_config_info_show, NULL);
5195 static DEVICE_ATTR(pool, 0444, rbd_pool_show, NULL);
5196 static DEVICE_ATTR(pool_id, 0444, rbd_pool_id_show, NULL);
5197 static DEVICE_ATTR(pool_ns, 0444, rbd_pool_ns_show, NULL);
5198 static DEVICE_ATTR(name, 0444, rbd_name_show, NULL);
5199 static DEVICE_ATTR(image_id, 0444, rbd_image_id_show, NULL);
5200 static DEVICE_ATTR(refresh, 0200, NULL, rbd_image_refresh);
5201 static DEVICE_ATTR(current_snap, 0444, rbd_snap_show, NULL);
5202 static DEVICE_ATTR(snap_id, 0444, rbd_snap_id_show, NULL);
5203 static DEVICE_ATTR(parent, 0444, rbd_parent_show, NULL);
5205 static struct attribute *rbd_attrs[] = {
5206 &dev_attr_size.attr,
5207 &dev_attr_features.attr,
5208 &dev_attr_major.attr,
5209 &dev_attr_minor.attr,
5210 &dev_attr_client_addr.attr,
5211 &dev_attr_client_id.attr,
5212 &dev_attr_cluster_fsid.attr,
5213 &dev_attr_config_info.attr,
5214 &dev_attr_pool.attr,
5215 &dev_attr_pool_id.attr,
5216 &dev_attr_pool_ns.attr,
5217 &dev_attr_name.attr,
5218 &dev_attr_image_id.attr,
5219 &dev_attr_current_snap.attr,
5220 &dev_attr_snap_id.attr,
5221 &dev_attr_parent.attr,
5222 &dev_attr_refresh.attr,
5226 static struct attribute_group rbd_attr_group = {
5230 static const struct attribute_group *rbd_attr_groups[] = {
5235 static void rbd_dev_release(struct device *dev);
5237 static const struct device_type rbd_device_type = {
5239 .groups = rbd_attr_groups,
5240 .release = rbd_dev_release,
5243 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
5245 kref_get(&spec->kref);
5250 static void rbd_spec_free(struct kref *kref);
5251 static void rbd_spec_put(struct rbd_spec *spec)
5254 kref_put(&spec->kref, rbd_spec_free);
5257 static struct rbd_spec *rbd_spec_alloc(void)
5259 struct rbd_spec *spec;
5261 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
5265 spec->pool_id = CEPH_NOPOOL;
5266 spec->snap_id = CEPH_NOSNAP;
5267 kref_init(&spec->kref);
5272 static void rbd_spec_free(struct kref *kref)
5274 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
5276 kfree(spec->pool_name);
5277 kfree(spec->pool_ns);
5278 kfree(spec->image_id);
5279 kfree(spec->image_name);
5280 kfree(spec->snap_name);
5284 static void rbd_dev_free(struct rbd_device *rbd_dev)
5286 WARN_ON(rbd_dev->watch_state != RBD_WATCH_STATE_UNREGISTERED);
5287 WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_UNLOCKED);
5289 ceph_oid_destroy(&rbd_dev->header_oid);
5290 ceph_oloc_destroy(&rbd_dev->header_oloc);
5291 kfree(rbd_dev->config_info);
5293 rbd_put_client(rbd_dev->rbd_client);
5294 rbd_spec_put(rbd_dev->spec);
5295 kfree(rbd_dev->opts);
5299 static void rbd_dev_release(struct device *dev)
5301 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5302 bool need_put = !!rbd_dev->opts;
5305 destroy_workqueue(rbd_dev->task_wq);
5306 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5309 rbd_dev_free(rbd_dev);
5312 * This is racy, but way better than putting module outside of
5313 * the release callback. The race window is pretty small, so
5314 * doing something similar to dm (dm-builtin.c) is overkill.
5317 module_put(THIS_MODULE);
5320 static struct rbd_device *__rbd_dev_create(struct rbd_spec *spec)
5322 struct rbd_device *rbd_dev;
5324 rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
5328 spin_lock_init(&rbd_dev->lock);
5329 INIT_LIST_HEAD(&rbd_dev->node);
5330 init_rwsem(&rbd_dev->header_rwsem);
5332 rbd_dev->header.data_pool_id = CEPH_NOPOOL;
5333 ceph_oid_init(&rbd_dev->header_oid);
5334 rbd_dev->header_oloc.pool = spec->pool_id;
5335 if (spec->pool_ns) {
5336 WARN_ON(!*spec->pool_ns);
5337 rbd_dev->header_oloc.pool_ns =
5338 ceph_find_or_create_string(spec->pool_ns,
5339 strlen(spec->pool_ns));
5342 mutex_init(&rbd_dev->watch_mutex);
5343 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
5344 INIT_DELAYED_WORK(&rbd_dev->watch_dwork, rbd_reregister_watch);
5346 init_rwsem(&rbd_dev->lock_rwsem);
5347 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
5348 INIT_WORK(&rbd_dev->acquired_lock_work, rbd_notify_acquired_lock);
5349 INIT_WORK(&rbd_dev->released_lock_work, rbd_notify_released_lock);
5350 INIT_DELAYED_WORK(&rbd_dev->lock_dwork, rbd_acquire_lock);
5351 INIT_WORK(&rbd_dev->unlock_work, rbd_release_lock_work);
5352 spin_lock_init(&rbd_dev->lock_lists_lock);
5353 INIT_LIST_HEAD(&rbd_dev->acquiring_list);
5354 INIT_LIST_HEAD(&rbd_dev->running_list);
5355 init_completion(&rbd_dev->acquire_wait);
5356 init_completion(&rbd_dev->releasing_wait);
5358 spin_lock_init(&rbd_dev->object_map_lock);
5360 rbd_dev->dev.bus = &rbd_bus_type;
5361 rbd_dev->dev.type = &rbd_device_type;
5362 rbd_dev->dev.parent = &rbd_root_dev;
5363 device_initialize(&rbd_dev->dev);
5369 * Create a mapping rbd_dev.
5371 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
5372 struct rbd_spec *spec,
5373 struct rbd_options *opts)
5375 struct rbd_device *rbd_dev;
5377 rbd_dev = __rbd_dev_create(spec);
5381 /* get an id and fill in device name */
5382 rbd_dev->dev_id = ida_simple_get(&rbd_dev_id_ida, 0,
5383 minor_to_rbd_dev_id(1 << MINORBITS),
5385 if (rbd_dev->dev_id < 0)
5388 sprintf(rbd_dev->name, RBD_DRV_NAME "%d", rbd_dev->dev_id);
5389 rbd_dev->task_wq = alloc_ordered_workqueue("%s-tasks", WQ_MEM_RECLAIM,
5391 if (!rbd_dev->task_wq)
5394 /* we have a ref from do_rbd_add() */
5395 __module_get(THIS_MODULE);
5397 rbd_dev->rbd_client = rbdc;
5398 rbd_dev->spec = spec;
5399 rbd_dev->opts = opts;
5401 dout("%s rbd_dev %p dev_id %d\n", __func__, rbd_dev, rbd_dev->dev_id);
5405 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5407 rbd_dev_free(rbd_dev);
5411 static void rbd_dev_destroy(struct rbd_device *rbd_dev)
5414 put_device(&rbd_dev->dev);
5418 * Get the size and object order for an image snapshot, or if
5419 * snap_id is CEPH_NOSNAP, gets this information for the base
5422 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
5423 u8 *order, u64 *snap_size)
5425 __le64 snapid = cpu_to_le64(snap_id);
5430 } __attribute__ ((packed)) size_buf = { 0 };
5432 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5433 &rbd_dev->header_oloc, "get_size",
5434 &snapid, sizeof(snapid),
5435 &size_buf, sizeof(size_buf));
5436 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5439 if (ret < sizeof (size_buf))
5443 *order = size_buf.order;
5444 dout(" order %u", (unsigned int)*order);
5446 *snap_size = le64_to_cpu(size_buf.size);
5448 dout(" snap_id 0x%016llx snap_size = %llu\n",
5449 (unsigned long long)snap_id,
5450 (unsigned long long)*snap_size);
5455 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
5457 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
5458 &rbd_dev->header.obj_order,
5459 &rbd_dev->header.image_size);
5462 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
5469 /* Response will be an encoded string, which includes a length */
5470 size = sizeof(__le32) + RBD_OBJ_PREFIX_LEN_MAX;
5471 reply_buf = kzalloc(size, GFP_KERNEL);
5475 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5476 &rbd_dev->header_oloc, "get_object_prefix",
5477 NULL, 0, reply_buf, size);
5478 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5483 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
5484 p + ret, NULL, GFP_NOIO);
5487 if (IS_ERR(rbd_dev->header.object_prefix)) {
5488 ret = PTR_ERR(rbd_dev->header.object_prefix);
5489 rbd_dev->header.object_prefix = NULL;
5491 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix);
5499 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
5500 bool read_only, u64 *snap_features)
5509 } __attribute__ ((packed)) features_buf = { 0 };
5513 features_in.snap_id = cpu_to_le64(snap_id);
5514 features_in.read_only = read_only;
5516 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5517 &rbd_dev->header_oloc, "get_features",
5518 &features_in, sizeof(features_in),
5519 &features_buf, sizeof(features_buf));
5520 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5523 if (ret < sizeof (features_buf))
5526 unsup = le64_to_cpu(features_buf.incompat) & ~RBD_FEATURES_SUPPORTED;
5528 rbd_warn(rbd_dev, "image uses unsupported features: 0x%llx",
5533 *snap_features = le64_to_cpu(features_buf.features);
5535 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
5536 (unsigned long long)snap_id,
5537 (unsigned long long)*snap_features,
5538 (unsigned long long)le64_to_cpu(features_buf.incompat));
5543 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
5545 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
5547 &rbd_dev->header.features);
5551 * These are generic image flags, but since they are used only for
5552 * object map, store them in rbd_dev->object_map_flags.
5554 * For the same reason, this function is called only on object map
5555 * (re)load and not on header refresh.
5557 static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev)
5559 __le64 snapid = cpu_to_le64(rbd_dev->spec->snap_id);
5563 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5564 &rbd_dev->header_oloc, "get_flags",
5565 &snapid, sizeof(snapid),
5566 &flags, sizeof(flags));
5569 if (ret < sizeof(flags))
5572 rbd_dev->object_map_flags = le64_to_cpu(flags);
5576 struct parent_image_info {
5578 const char *pool_ns;
5579 const char *image_id;
5587 * The caller is responsible for @pii.
5589 static int decode_parent_image_spec(void **p, void *end,
5590 struct parent_image_info *pii)
5596 ret = ceph_start_decoding(p, end, 1, "ParentImageSpec",
5597 &struct_v, &struct_len);
5601 ceph_decode_64_safe(p, end, pii->pool_id, e_inval);
5602 pii->pool_ns = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5603 if (IS_ERR(pii->pool_ns)) {
5604 ret = PTR_ERR(pii->pool_ns);
5605 pii->pool_ns = NULL;
5608 pii->image_id = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5609 if (IS_ERR(pii->image_id)) {
5610 ret = PTR_ERR(pii->image_id);
5611 pii->image_id = NULL;
5614 ceph_decode_64_safe(p, end, pii->snap_id, e_inval);
5621 static int __get_parent_info(struct rbd_device *rbd_dev,
5622 struct page *req_page,
5623 struct page *reply_page,
5624 struct parent_image_info *pii)
5626 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5627 size_t reply_len = PAGE_SIZE;
5631 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5632 "rbd", "parent_get", CEPH_OSD_FLAG_READ,
5633 req_page, sizeof(u64), &reply_page, &reply_len);
5635 return ret == -EOPNOTSUPP ? 1 : ret;
5637 p = page_address(reply_page);
5638 end = p + reply_len;
5639 ret = decode_parent_image_spec(&p, end, pii);
5643 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5644 "rbd", "parent_overlap_get", CEPH_OSD_FLAG_READ,
5645 req_page, sizeof(u64), &reply_page, &reply_len);
5649 p = page_address(reply_page);
5650 end = p + reply_len;
5651 ceph_decode_8_safe(&p, end, pii->has_overlap, e_inval);
5652 if (pii->has_overlap)
5653 ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5662 * The caller is responsible for @pii.
5664 static int __get_parent_info_legacy(struct rbd_device *rbd_dev,
5665 struct page *req_page,
5666 struct page *reply_page,
5667 struct parent_image_info *pii)
5669 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5670 size_t reply_len = PAGE_SIZE;
5674 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5675 "rbd", "get_parent", CEPH_OSD_FLAG_READ,
5676 req_page, sizeof(u64), &reply_page, &reply_len);
5680 p = page_address(reply_page);
5681 end = p + reply_len;
5682 ceph_decode_64_safe(&p, end, pii->pool_id, e_inval);
5683 pii->image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
5684 if (IS_ERR(pii->image_id)) {
5685 ret = PTR_ERR(pii->image_id);
5686 pii->image_id = NULL;
5689 ceph_decode_64_safe(&p, end, pii->snap_id, e_inval);
5690 pii->has_overlap = true;
5691 ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5699 static int get_parent_info(struct rbd_device *rbd_dev,
5700 struct parent_image_info *pii)
5702 struct page *req_page, *reply_page;
5706 req_page = alloc_page(GFP_KERNEL);
5710 reply_page = alloc_page(GFP_KERNEL);
5712 __free_page(req_page);
5716 p = page_address(req_page);
5717 ceph_encode_64(&p, rbd_dev->spec->snap_id);
5718 ret = __get_parent_info(rbd_dev, req_page, reply_page, pii);
5720 ret = __get_parent_info_legacy(rbd_dev, req_page, reply_page,
5723 __free_page(req_page);
5724 __free_page(reply_page);
5728 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
5730 struct rbd_spec *parent_spec;
5731 struct parent_image_info pii = { 0 };
5734 parent_spec = rbd_spec_alloc();
5738 ret = get_parent_info(rbd_dev, &pii);
5742 dout("%s pool_id %llu pool_ns %s image_id %s snap_id %llu has_overlap %d overlap %llu\n",
5743 __func__, pii.pool_id, pii.pool_ns, pii.image_id, pii.snap_id,
5744 pii.has_overlap, pii.overlap);
5746 if (pii.pool_id == CEPH_NOPOOL || !pii.has_overlap) {
5748 * Either the parent never existed, or we have
5749 * record of it but the image got flattened so it no
5750 * longer has a parent. When the parent of a
5751 * layered image disappears we immediately set the
5752 * overlap to 0. The effect of this is that all new
5753 * requests will be treated as if the image had no
5756 * If !pii.has_overlap, the parent image spec is not
5757 * applicable. It's there to avoid duplication in each
5760 if (rbd_dev->parent_overlap) {
5761 rbd_dev->parent_overlap = 0;
5762 rbd_dev_parent_put(rbd_dev);
5763 pr_info("%s: clone image has been flattened\n",
5764 rbd_dev->disk->disk_name);
5767 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
5781 * flattened, already handled that). So we only need to
5782 * record the parent spec we have not already done so.
5784 if (!rbd_dev->parent_spec) {
5785 parent_spec->pool_id = pii.pool_id;
5786 if (pii.pool_ns && *pii.pool_ns) {
5787 parent_spec->pool_ns = pii.pool_ns;
5790 parent_spec->image_id = pii.image_id;
5791 pii.image_id = NULL;
5792 parent_spec->snap_id = pii.snap_id;
5794 rbd_dev->parent_spec = parent_spec;
5795 parent_spec = NULL; /* rbd_dev now owns this */
5799 * We always update the parent overlap. If it's zero we issue
5800 * a warning, as we will proceed as if there was no parent.
5804 /* refresh, careful to warn just once */
5805 if (rbd_dev->parent_overlap)
5807 "clone now standalone (overlap became 0)");
5810 rbd_warn(rbd_dev, "clone is standalone (overlap 0)");
5813 rbd_dev->parent_overlap = pii.overlap;
5819 kfree(pii.image_id);
5820 rbd_spec_put(parent_spec);
5824 static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev)
5828 __le64 stripe_count;
5829 } __attribute__ ((packed)) striping_info_buf = { 0 };
5830 size_t size = sizeof (striping_info_buf);
5834 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5835 &rbd_dev->header_oloc, "get_stripe_unit_count",
5836 NULL, 0, &striping_info_buf, size);
5837 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5843 p = &striping_info_buf;
5844 rbd_dev->header.stripe_unit = ceph_decode_64(&p);
5845 rbd_dev->header.stripe_count = ceph_decode_64(&p);
5849 static int rbd_dev_v2_data_pool(struct rbd_device *rbd_dev)
5851 __le64 data_pool_id;
5854 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5855 &rbd_dev->header_oloc, "get_data_pool",
5856 NULL, 0, &data_pool_id, sizeof(data_pool_id));
5859 if (ret < sizeof(data_pool_id))
5862 rbd_dev->header.data_pool_id = le64_to_cpu(data_pool_id);
5863 WARN_ON(rbd_dev->header.data_pool_id == CEPH_NOPOOL);
5867 static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
5869 CEPH_DEFINE_OID_ONSTACK(oid);
5870 size_t image_id_size;
5875 void *reply_buf = NULL;
5877 char *image_name = NULL;
5880 rbd_assert(!rbd_dev->spec->image_name);
5882 len = strlen(rbd_dev->spec->image_id);
5883 image_id_size = sizeof (__le32) + len;
5884 image_id = kmalloc(image_id_size, GFP_KERNEL);
5889 end = image_id + image_id_size;
5890 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len);
5892 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
5893 reply_buf = kmalloc(size, GFP_KERNEL);
5897 ceph_oid_printf(&oid, "%s", RBD_DIRECTORY);
5898 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
5899 "dir_get_name", image_id, image_id_size,
5904 end = reply_buf + ret;
5906 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
5907 if (IS_ERR(image_name))
5910 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
5918 static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5920 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5921 const char *snap_name;
5924 /* Skip over names until we find the one we are looking for */
5926 snap_name = rbd_dev->header.snap_names;
5927 while (which < snapc->num_snaps) {
5928 if (!strcmp(name, snap_name))
5929 return snapc->snaps[which];
5930 snap_name += strlen(snap_name) + 1;
5936 static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5938 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5943 for (which = 0; !found && which < snapc->num_snaps; which++) {
5944 const char *snap_name;
5946 snap_id = snapc->snaps[which];
5947 snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
5948 if (IS_ERR(snap_name)) {
5949 /* ignore no-longer existing snapshots */
5950 if (PTR_ERR(snap_name) == -ENOENT)
5955 found = !strcmp(name, snap_name);
5958 return found ? snap_id : CEPH_NOSNAP;
5962 * Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if
5963 * no snapshot by that name is found, or if an error occurs.
5965 static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5967 if (rbd_dev->image_format == 1)
5968 return rbd_v1_snap_id_by_name(rbd_dev, name);
5970 return rbd_v2_snap_id_by_name(rbd_dev, name);
5974 * An image being mapped will have everything but the snap id.
5976 static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev)
5978 struct rbd_spec *spec = rbd_dev->spec;
5980 rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name);
5981 rbd_assert(spec->image_id && spec->image_name);
5982 rbd_assert(spec->snap_name);
5984 if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) {
5987 snap_id = rbd_snap_id_by_name(rbd_dev, spec->snap_name);
5988 if (snap_id == CEPH_NOSNAP)
5991 spec->snap_id = snap_id;
5993 spec->snap_id = CEPH_NOSNAP;
6000 * A parent image will have all ids but none of the names.
6002 * All names in an rbd spec are dynamically allocated. It's OK if we
6003 * can't figure out the name for an image id.
6005 static int rbd_spec_fill_names(struct rbd_device *rbd_dev)
6007 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
6008 struct rbd_spec *spec = rbd_dev->spec;
6009 const char *pool_name;
6010 const char *image_name;
6011 const char *snap_name;
6014 rbd_assert(spec->pool_id != CEPH_NOPOOL);
6015 rbd_assert(spec->image_id);
6016 rbd_assert(spec->snap_id != CEPH_NOSNAP);
6018 /* Get the pool name; we have to make our own copy of this */
6020 pool_name = ceph_pg_pool_name_by_id(osdc->osdmap, spec->pool_id);
6022 rbd_warn(rbd_dev, "no pool with id %llu", spec->pool_id);
6025 pool_name = kstrdup(pool_name, GFP_KERNEL);
6029 /* Fetch the image name; tolerate failure here */
6031 image_name = rbd_dev_image_name(rbd_dev);
6033 rbd_warn(rbd_dev, "unable to get image name");
6035 /* Fetch the snapshot name */
6037 snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
6038 if (IS_ERR(snap_name)) {
6039 ret = PTR_ERR(snap_name);
6043 spec->pool_name = pool_name;
6044 spec->image_name = image_name;
6045 spec->snap_name = snap_name;
6055 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev)
6064 struct ceph_snap_context *snapc;
6068 * We'll need room for the seq value (maximum snapshot id),
6069 * snapshot count, and array of that many snapshot ids.
6070 * For now we have a fixed upper limit on the number we're
6071 * prepared to receive.
6073 size = sizeof (__le64) + sizeof (__le32) +
6074 RBD_MAX_SNAP_COUNT * sizeof (__le64);
6075 reply_buf = kzalloc(size, GFP_KERNEL);
6079 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6080 &rbd_dev->header_oloc, "get_snapcontext",
6081 NULL, 0, reply_buf, size);
6082 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6087 end = reply_buf + ret;
6089 ceph_decode_64_safe(&p, end, seq, out);
6090 ceph_decode_32_safe(&p, end, snap_count, out);
6093 * Make sure the reported number of snapshot ids wouldn't go
6094 * beyond the end of our buffer. But before checking that,
6095 * make sure the computed size of the snapshot context we
6096 * allocate is representable in a size_t.
6098 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
6103 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
6107 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
6113 for (i = 0; i < snap_count; i++)
6114 snapc->snaps[i] = ceph_decode_64(&p);
6116 ceph_put_snap_context(rbd_dev->header.snapc);
6117 rbd_dev->header.snapc = snapc;
6119 dout(" snap context seq = %llu, snap_count = %u\n",
6120 (unsigned long long)seq, (unsigned int)snap_count);
6127 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
6138 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
6139 reply_buf = kmalloc(size, GFP_KERNEL);
6141 return ERR_PTR(-ENOMEM);
6143 snapid = cpu_to_le64(snap_id);
6144 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6145 &rbd_dev->header_oloc, "get_snapshot_name",
6146 &snapid, sizeof(snapid), reply_buf, size);
6147 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6149 snap_name = ERR_PTR(ret);
6154 end = reply_buf + ret;
6155 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
6156 if (IS_ERR(snap_name))
6159 dout(" snap_id 0x%016llx snap_name = %s\n",
6160 (unsigned long long)snap_id, snap_name);
6167 static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev)
6169 bool first_time = rbd_dev->header.object_prefix == NULL;
6172 ret = rbd_dev_v2_image_size(rbd_dev);
6177 ret = rbd_dev_v2_header_onetime(rbd_dev);
6182 ret = rbd_dev_v2_snap_context(rbd_dev);
6183 if (ret && first_time) {
6184 kfree(rbd_dev->header.object_prefix);
6185 rbd_dev->header.object_prefix = NULL;
6191 static int rbd_dev_header_info(struct rbd_device *rbd_dev)
6193 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6195 if (rbd_dev->image_format == 1)
6196 return rbd_dev_v1_header_info(rbd_dev);
6198 return rbd_dev_v2_header_info(rbd_dev);
6202 * Skips over white space at *buf, and updates *buf to point to the
6203 * first found non-space character (if any). Returns the length of
6204 * the token (string of non-white space characters) found. Note
6205 * that *buf must be terminated with '\0'.
6207 static inline size_t next_token(const char **buf)
6210 * These are the characters that produce nonzero for
6211 * isspace() in the "C" and "POSIX" locales.
6213 static const char spaces[] = " \f\n\r\t\v";
6215 *buf += strspn(*buf, spaces); /* Find start of token */
6217 return strcspn(*buf, spaces); /* Return token length */
6221 * Finds the next token in *buf, dynamically allocates a buffer big
6222 * enough to hold a copy of it, and copies the token into the new
6223 * buffer. The copy is guaranteed to be terminated with '\0'. Note
6224 * that a duplicate buffer is created even for a zero-length token.
6226 * Returns a pointer to the newly-allocated duplicate, or a null
6227 * pointer if memory for the duplicate was not available. If
6228 * the lenp argument is a non-null pointer, the length of the token
6229 * (not including the '\0') is returned in *lenp.
6231 * If successful, the *buf pointer will be updated to point beyond
6232 * the end of the found token.
6234 * Note: uses GFP_KERNEL for allocation.
6236 static inline char *dup_token(const char **buf, size_t *lenp)
6241 len = next_token(buf);
6242 dup = kmemdup(*buf, len + 1, GFP_KERNEL);
6245 *(dup + len) = '\0';
6254 static int rbd_parse_param(struct fs_parameter *param,
6255 struct rbd_parse_opts_ctx *pctx)
6257 struct rbd_options *opt = pctx->opts;
6258 struct fs_parse_result result;
6259 struct p_log log = {.prefix = "rbd"};
6262 ret = ceph_parse_param(param, pctx->copts, NULL);
6263 if (ret != -ENOPARAM)
6266 token = __fs_parse(&log, rbd_parameters, param, &result);
6267 dout("%s fs_parse '%s' token %d\n", __func__, param->key, token);
6269 if (token == -ENOPARAM)
6270 return inval_plog(&log, "Unknown parameter '%s'",
6276 case Opt_queue_depth:
6277 if (result.uint_32 < 1)
6279 opt->queue_depth = result.uint_32;
6281 case Opt_alloc_size:
6282 if (result.uint_32 < SECTOR_SIZE)
6284 if (!is_power_of_2(result.uint_32))
6285 return inval_plog(&log, "alloc_size must be a power of 2");
6286 opt->alloc_size = result.uint_32;
6288 case Opt_lock_timeout:
6289 /* 0 is "wait forever" (i.e. infinite timeout) */
6290 if (result.uint_32 > INT_MAX / 1000)
6292 opt->lock_timeout = msecs_to_jiffies(result.uint_32 * 1000);
6295 kfree(pctx->spec->pool_ns);
6296 pctx->spec->pool_ns = param->string;
6297 param->string = NULL;
6299 case Opt_compression_hint:
6300 switch (result.uint_32) {
6301 case Opt_compression_hint_none:
6302 opt->alloc_hint_flags &=
6303 ~(CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE |
6304 CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE);
6306 case Opt_compression_hint_compressible:
6307 opt->alloc_hint_flags |=
6308 CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6309 opt->alloc_hint_flags &=
6310 ~CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6312 case Opt_compression_hint_incompressible:
6313 opt->alloc_hint_flags |=
6314 CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6315 opt->alloc_hint_flags &=
6316 ~CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6323 opt->read_only = true;
6325 case Opt_read_write:
6326 opt->read_only = false;
6328 case Opt_lock_on_read:
6329 opt->lock_on_read = true;
6332 opt->exclusive = true;
6344 return inval_plog(&log, "%s out of range", param->key);
6348 * This duplicates most of generic_parse_monolithic(), untying it from
6349 * fs_context and skipping standard superblock and security options.
6351 static int rbd_parse_options(char *options, struct rbd_parse_opts_ctx *pctx)
6356 dout("%s '%s'\n", __func__, options);
6357 while ((key = strsep(&options, ",")) != NULL) {
6359 struct fs_parameter param = {
6361 .type = fs_value_is_flag,
6363 char *value = strchr(key, '=');
6370 v_len = strlen(value);
6371 param.string = kmemdup_nul(value, v_len,
6375 param.type = fs_value_is_string;
6379 ret = rbd_parse_param(¶m, pctx);
6380 kfree(param.string);
6390 * Parse the options provided for an "rbd add" (i.e., rbd image
6391 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
6392 * and the data written is passed here via a NUL-terminated buffer.
6393 * Returns 0 if successful or an error code otherwise.
6395 * The information extracted from these options is recorded in
6396 * the other parameters which return dynamically-allocated
6399 * The address of a pointer that will refer to a ceph options
6400 * structure. Caller must release the returned pointer using
6401 * ceph_destroy_options() when it is no longer needed.
6403 * Address of an rbd options pointer. Fully initialized by
6404 * this function; caller must release with kfree().
6406 * Address of an rbd image specification pointer. Fully
6407 * initialized by this function based on parsed options.
6408 * Caller must release with rbd_spec_put().
6410 * The options passed take this form:
6411 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
6414 * A comma-separated list of one or more monitor addresses.
6415 * A monitor address is an ip address, optionally followed
6416 * by a port number (separated by a colon).
6417 * I.e.: ip1[:port1][,ip2[:port2]...]
6419 * A comma-separated list of ceph and/or rbd options.
6421 * The name of the rados pool containing the rbd image.
6423 * The name of the image in that pool to map.
6425 * An optional snapshot id. If provided, the mapping will
6426 * present data from the image at the time that snapshot was
6427 * created. The image head is used if no snapshot id is
6428 * provided. Snapshot mappings are always read-only.
6430 static int rbd_add_parse_args(const char *buf,
6431 struct ceph_options **ceph_opts,
6432 struct rbd_options **opts,
6433 struct rbd_spec **rbd_spec)
6437 const char *mon_addrs;
6439 size_t mon_addrs_size;
6440 struct rbd_parse_opts_ctx pctx = { 0 };
6443 /* The first four tokens are required */
6445 len = next_token(&buf);
6447 rbd_warn(NULL, "no monitor address(es) provided");
6451 mon_addrs_size = len;
6455 options = dup_token(&buf, NULL);
6459 rbd_warn(NULL, "no options provided");
6463 pctx.spec = rbd_spec_alloc();
6467 pctx.spec->pool_name = dup_token(&buf, NULL);
6468 if (!pctx.spec->pool_name)
6470 if (!*pctx.spec->pool_name) {
6471 rbd_warn(NULL, "no pool name provided");
6475 pctx.spec->image_name = dup_token(&buf, NULL);
6476 if (!pctx.spec->image_name)
6478 if (!*pctx.spec->image_name) {
6479 rbd_warn(NULL, "no image name provided");
6484 * Snapshot name is optional; default is to use "-"
6485 * (indicating the head/no snapshot).
6487 len = next_token(&buf);
6489 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
6490 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
6491 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
6492 ret = -ENAMETOOLONG;
6495 snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
6498 *(snap_name + len) = '\0';
6499 pctx.spec->snap_name = snap_name;
6501 pctx.copts = ceph_alloc_options();
6505 /* Initialize all rbd options to the defaults */
6507 pctx.opts = kzalloc(sizeof(*pctx.opts), GFP_KERNEL);
6511 pctx.opts->read_only = RBD_READ_ONLY_DEFAULT;
6512 pctx.opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
6513 pctx.opts->alloc_size = RBD_ALLOC_SIZE_DEFAULT;
6514 pctx.opts->lock_timeout = RBD_LOCK_TIMEOUT_DEFAULT;
6515 pctx.opts->lock_on_read = RBD_LOCK_ON_READ_DEFAULT;
6516 pctx.opts->exclusive = RBD_EXCLUSIVE_DEFAULT;
6517 pctx.opts->trim = RBD_TRIM_DEFAULT;
6519 ret = ceph_parse_mon_ips(mon_addrs, mon_addrs_size, pctx.copts, NULL,
6524 ret = rbd_parse_options(options, &pctx);
6528 *ceph_opts = pctx.copts;
6530 *rbd_spec = pctx.spec;
6538 ceph_destroy_options(pctx.copts);
6539 rbd_spec_put(pctx.spec);
6544 static void rbd_dev_image_unlock(struct rbd_device *rbd_dev)
6546 down_write(&rbd_dev->lock_rwsem);
6547 if (__rbd_is_lock_owner(rbd_dev))
6548 __rbd_release_lock(rbd_dev);
6549 up_write(&rbd_dev->lock_rwsem);
6553 * If the wait is interrupted, an error is returned even if the lock
6554 * was successfully acquired. rbd_dev_image_unlock() will release it
6557 static int rbd_add_acquire_lock(struct rbd_device *rbd_dev)
6561 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK)) {
6562 if (!rbd_dev->opts->exclusive && !rbd_dev->opts->lock_on_read)
6565 rbd_warn(rbd_dev, "exclusive-lock feature is not enabled");
6569 if (rbd_is_ro(rbd_dev))
6572 rbd_assert(!rbd_is_lock_owner(rbd_dev));
6573 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
6574 ret = wait_for_completion_killable_timeout(&rbd_dev->acquire_wait,
6575 ceph_timeout_jiffies(rbd_dev->opts->lock_timeout));
6577 ret = rbd_dev->acquire_err;
6579 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
6585 rbd_warn(rbd_dev, "failed to acquire exclusive lock: %ld", ret);
6590 * The lock may have been released by now, unless automatic lock
6591 * transitions are disabled.
6593 rbd_assert(!rbd_dev->opts->exclusive || rbd_is_lock_owner(rbd_dev));
6598 * An rbd format 2 image has a unique identifier, distinct from the
6599 * name given to it by the user. Internally, that identifier is
6600 * what's used to specify the names of objects related to the image.
6602 * A special "rbd id" object is used to map an rbd image name to its
6603 * id. If that object doesn't exist, then there is no v2 rbd image
6604 * with the supplied name.
6606 * This function will record the given rbd_dev's image_id field if
6607 * it can be determined, and in that case will return 0. If any
6608 * errors occur a negative errno will be returned and the rbd_dev's
6609 * image_id field will be unchanged (and should be NULL).
6611 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
6615 CEPH_DEFINE_OID_ONSTACK(oid);
6620 * When probing a parent image, the image id is already
6621 * known (and the image name likely is not). There's no
6622 * need to fetch the image id again in this case. We
6623 * do still need to set the image format though.
6625 if (rbd_dev->spec->image_id) {
6626 rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
6632 * First, see if the format 2 image id file exists, and if
6633 * so, get the image's persistent id from it.
6635 ret = ceph_oid_aprintf(&oid, GFP_KERNEL, "%s%s", RBD_ID_PREFIX,
6636 rbd_dev->spec->image_name);
6640 dout("rbd id object name is %s\n", oid.name);
6642 /* Response will be an encoded string, which includes a length */
6643 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
6644 response = kzalloc(size, GFP_NOIO);
6650 /* If it doesn't exist we'll assume it's a format 1 image */
6652 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
6655 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6656 if (ret == -ENOENT) {
6657 image_id = kstrdup("", GFP_KERNEL);
6658 ret = image_id ? 0 : -ENOMEM;
6660 rbd_dev->image_format = 1;
6661 } else if (ret >= 0) {
6664 image_id = ceph_extract_encoded_string(&p, p + ret,
6666 ret = PTR_ERR_OR_ZERO(image_id);
6668 rbd_dev->image_format = 2;
6672 rbd_dev->spec->image_id = image_id;
6673 dout("image_id is %s\n", image_id);
6677 ceph_oid_destroy(&oid);
6682 * Undo whatever state changes are made by v1 or v2 header info
6685 static void rbd_dev_unprobe(struct rbd_device *rbd_dev)
6687 struct rbd_image_header *header;
6689 rbd_dev_parent_put(rbd_dev);
6690 rbd_object_map_free(rbd_dev);
6691 rbd_dev_mapping_clear(rbd_dev);
6693 /* Free dynamic fields from the header, then zero it out */
6695 header = &rbd_dev->header;
6696 ceph_put_snap_context(header->snapc);
6697 kfree(header->snap_sizes);
6698 kfree(header->snap_names);
6699 kfree(header->object_prefix);
6700 memset(header, 0, sizeof (*header));
6703 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev)
6707 ret = rbd_dev_v2_object_prefix(rbd_dev);
6712 * Get the and check features for the image. Currently the
6713 * features are assumed to never change.
6715 ret = rbd_dev_v2_features(rbd_dev);
6719 /* If the image supports fancy striping, get its parameters */
6721 if (rbd_dev->header.features & RBD_FEATURE_STRIPINGV2) {
6722 ret = rbd_dev_v2_striping_info(rbd_dev);
6727 if (rbd_dev->header.features & RBD_FEATURE_DATA_POOL) {
6728 ret = rbd_dev_v2_data_pool(rbd_dev);
6733 rbd_init_layout(rbd_dev);
6737 rbd_dev->header.features = 0;
6738 kfree(rbd_dev->header.object_prefix);
6739 rbd_dev->header.object_prefix = NULL;
6744 * @depth is rbd_dev_image_probe() -> rbd_dev_probe_parent() ->
6745 * rbd_dev_image_probe() recursion depth, which means it's also the
6746 * length of the already discovered part of the parent chain.
6748 static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
6750 struct rbd_device *parent = NULL;
6753 if (!rbd_dev->parent_spec)
6756 if (++depth > RBD_MAX_PARENT_CHAIN_LEN) {
6757 pr_info("parent chain is too long (%d)\n", depth);
6762 parent = __rbd_dev_create(rbd_dev->parent_spec);
6769 * Images related by parent/child relationships always share
6770 * rbd_client and spec/parent_spec, so bump their refcounts.
6772 parent->rbd_client = __rbd_get_client(rbd_dev->rbd_client);
6773 parent->spec = rbd_spec_get(rbd_dev->parent_spec);
6775 __set_bit(RBD_DEV_FLAG_READONLY, &parent->flags);
6777 ret = rbd_dev_image_probe(parent, depth);
6781 rbd_dev->parent = parent;
6782 atomic_set(&rbd_dev->parent_ref, 1);
6786 rbd_dev_unparent(rbd_dev);
6787 rbd_dev_destroy(parent);
6791 static void rbd_dev_device_release(struct rbd_device *rbd_dev)
6793 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6794 rbd_free_disk(rbd_dev);
6796 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6800 * rbd_dev->header_rwsem must be locked for write and will be unlocked
6803 static int rbd_dev_device_setup(struct rbd_device *rbd_dev)
6807 /* Record our major and minor device numbers. */
6809 if (!single_major) {
6810 ret = register_blkdev(0, rbd_dev->name);
6812 goto err_out_unlock;
6814 rbd_dev->major = ret;
6817 rbd_dev->major = rbd_major;
6818 rbd_dev->minor = rbd_dev_id_to_minor(rbd_dev->dev_id);
6821 /* Set up the blkdev mapping. */
6823 ret = rbd_init_disk(rbd_dev);
6825 goto err_out_blkdev;
6827 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
6828 set_disk_ro(rbd_dev->disk, rbd_is_ro(rbd_dev));
6830 ret = dev_set_name(&rbd_dev->dev, "%d", rbd_dev->dev_id);
6834 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6835 up_write(&rbd_dev->header_rwsem);
6839 rbd_free_disk(rbd_dev);
6842 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6844 up_write(&rbd_dev->header_rwsem);
6848 static int rbd_dev_header_name(struct rbd_device *rbd_dev)
6850 struct rbd_spec *spec = rbd_dev->spec;
6853 /* Record the header object name for this rbd image. */
6855 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6856 if (rbd_dev->image_format == 1)
6857 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6858 spec->image_name, RBD_SUFFIX);
6860 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6861 RBD_HEADER_PREFIX, spec->image_id);
6866 static void rbd_print_dne(struct rbd_device *rbd_dev, bool is_snap)
6869 pr_info("image %s/%s%s%s does not exist\n",
6870 rbd_dev->spec->pool_name,
6871 rbd_dev->spec->pool_ns ?: "",
6872 rbd_dev->spec->pool_ns ? "/" : "",
6873 rbd_dev->spec->image_name);
6875 pr_info("snap %s/%s%s%s@%s does not exist\n",
6876 rbd_dev->spec->pool_name,
6877 rbd_dev->spec->pool_ns ?: "",
6878 rbd_dev->spec->pool_ns ? "/" : "",
6879 rbd_dev->spec->image_name,
6880 rbd_dev->spec->snap_name);
6884 static void rbd_dev_image_release(struct rbd_device *rbd_dev)
6886 if (!rbd_is_ro(rbd_dev))
6887 rbd_unregister_watch(rbd_dev);
6889 rbd_dev_unprobe(rbd_dev);
6890 rbd_dev->image_format = 0;
6891 kfree(rbd_dev->spec->image_id);
6892 rbd_dev->spec->image_id = NULL;
6896 * Probe for the existence of the header object for the given rbd
6897 * device. If this image is the one being mapped (i.e., not a
6898 * parent), initiate a watch on its header object before using that
6899 * object to get detailed information about the rbd image.
6901 * On success, returns with header_rwsem held for write if called
6904 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
6906 bool need_watch = !rbd_is_ro(rbd_dev);
6910 * Get the id from the image id object. Unless there's an
6911 * error, rbd_dev->spec->image_id will be filled in with
6912 * a dynamically-allocated string, and rbd_dev->image_format
6913 * will be set to either 1 or 2.
6915 ret = rbd_dev_image_id(rbd_dev);
6919 ret = rbd_dev_header_name(rbd_dev);
6921 goto err_out_format;
6924 ret = rbd_register_watch(rbd_dev);
6927 rbd_print_dne(rbd_dev, false);
6928 goto err_out_format;
6933 down_write(&rbd_dev->header_rwsem);
6935 ret = rbd_dev_header_info(rbd_dev);
6937 if (ret == -ENOENT && !need_watch)
6938 rbd_print_dne(rbd_dev, false);
6943 * If this image is the one being mapped, we have pool name and
6944 * id, image name and id, and snap name - need to fill snap id.
6945 * Otherwise this is a parent image, identified by pool, image
6946 * and snap ids - need to fill in names for those ids.
6949 ret = rbd_spec_fill_snap_id(rbd_dev);
6951 ret = rbd_spec_fill_names(rbd_dev);
6954 rbd_print_dne(rbd_dev, true);
6958 ret = rbd_dev_mapping_set(rbd_dev);
6962 if (rbd_is_snap(rbd_dev) &&
6963 (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)) {
6964 ret = rbd_object_map_load(rbd_dev);
6969 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
6970 ret = rbd_dev_v2_parent_info(rbd_dev);
6975 ret = rbd_dev_probe_parent(rbd_dev, depth);
6979 dout("discovered format %u image, header name is %s\n",
6980 rbd_dev->image_format, rbd_dev->header_oid.name);
6985 up_write(&rbd_dev->header_rwsem);
6987 rbd_unregister_watch(rbd_dev);
6988 rbd_dev_unprobe(rbd_dev);
6990 rbd_dev->image_format = 0;
6991 kfree(rbd_dev->spec->image_id);
6992 rbd_dev->spec->image_id = NULL;
6996 static ssize_t do_rbd_add(const char *buf, size_t count)
6998 struct rbd_device *rbd_dev = NULL;
6999 struct ceph_options *ceph_opts = NULL;
7000 struct rbd_options *rbd_opts = NULL;
7001 struct rbd_spec *spec = NULL;
7002 struct rbd_client *rbdc;
7005 if (!capable(CAP_SYS_ADMIN))
7008 if (!try_module_get(THIS_MODULE))
7011 /* parse add command */
7012 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
7016 rbdc = rbd_get_client(ceph_opts);
7023 rc = ceph_pg_poolid_by_name(rbdc->client->osdc.osdmap, spec->pool_name);
7026 pr_info("pool %s does not exist\n", spec->pool_name);
7027 goto err_out_client;
7029 spec->pool_id = (u64)rc;
7031 rbd_dev = rbd_dev_create(rbdc, spec, rbd_opts);
7034 goto err_out_client;
7036 rbdc = NULL; /* rbd_dev now owns this */
7037 spec = NULL; /* rbd_dev now owns this */
7038 rbd_opts = NULL; /* rbd_dev now owns this */
7040 /* if we are mapping a snapshot it will be a read-only mapping */
7041 if (rbd_dev->opts->read_only ||
7042 strcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME))
7043 __set_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
7045 rbd_dev->config_info = kstrdup(buf, GFP_KERNEL);
7046 if (!rbd_dev->config_info) {
7048 goto err_out_rbd_dev;
7051 rc = rbd_dev_image_probe(rbd_dev, 0);
7053 goto err_out_rbd_dev;
7055 if (rbd_dev->opts->alloc_size > rbd_dev->layout.object_size) {
7056 rbd_warn(rbd_dev, "alloc_size adjusted to %u",
7057 rbd_dev->layout.object_size);
7058 rbd_dev->opts->alloc_size = rbd_dev->layout.object_size;
7061 rc = rbd_dev_device_setup(rbd_dev);
7063 goto err_out_image_probe;
7065 rc = rbd_add_acquire_lock(rbd_dev);
7067 goto err_out_image_lock;
7069 /* Everything's ready. Announce the disk to the world. */
7071 rc = device_add(&rbd_dev->dev);
7073 goto err_out_image_lock;
7075 rc = device_add_disk(&rbd_dev->dev, rbd_dev->disk, NULL);
7077 goto err_out_cleanup_disk;
7079 spin_lock(&rbd_dev_list_lock);
7080 list_add_tail(&rbd_dev->node, &rbd_dev_list);
7081 spin_unlock(&rbd_dev_list_lock);
7083 pr_info("%s: capacity %llu features 0x%llx\n", rbd_dev->disk->disk_name,
7084 (unsigned long long)get_capacity(rbd_dev->disk) << SECTOR_SHIFT,
7085 rbd_dev->header.features);
7088 module_put(THIS_MODULE);
7091 err_out_cleanup_disk:
7092 rbd_free_disk(rbd_dev);
7094 rbd_dev_image_unlock(rbd_dev);
7095 rbd_dev_device_release(rbd_dev);
7096 err_out_image_probe:
7097 rbd_dev_image_release(rbd_dev);
7099 rbd_dev_destroy(rbd_dev);
7101 rbd_put_client(rbdc);
7108 static ssize_t add_store(const struct bus_type *bus, const char *buf, size_t count)
7113 return do_rbd_add(buf, count);
7116 static ssize_t add_single_major_store(const struct bus_type *bus, const char *buf,
7119 return do_rbd_add(buf, count);
7122 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev)
7124 while (rbd_dev->parent) {
7125 struct rbd_device *first = rbd_dev;
7126 struct rbd_device *second = first->parent;
7127 struct rbd_device *third;
7130 * Follow to the parent with no grandparent and
7133 while (second && (third = second->parent)) {
7138 rbd_dev_image_release(second);
7139 rbd_dev_destroy(second);
7140 first->parent = NULL;
7141 first->parent_overlap = 0;
7143 rbd_assert(first->parent_spec);
7144 rbd_spec_put(first->parent_spec);
7145 first->parent_spec = NULL;
7149 static ssize_t do_rbd_remove(const char *buf, size_t count)
7151 struct rbd_device *rbd_dev = NULL;
7152 struct list_head *tmp;
7158 if (!capable(CAP_SYS_ADMIN))
7163 sscanf(buf, "%d %5s", &dev_id, opt_buf);
7165 pr_err("dev_id out of range\n");
7168 if (opt_buf[0] != '\0') {
7169 if (!strcmp(opt_buf, "force")) {
7172 pr_err("bad remove option at '%s'\n", opt_buf);
7178 spin_lock(&rbd_dev_list_lock);
7179 list_for_each(tmp, &rbd_dev_list) {
7180 rbd_dev = list_entry(tmp, struct rbd_device, node);
7181 if (rbd_dev->dev_id == dev_id) {
7187 spin_lock_irq(&rbd_dev->lock);
7188 if (rbd_dev->open_count && !force)
7190 else if (test_and_set_bit(RBD_DEV_FLAG_REMOVING,
7193 spin_unlock_irq(&rbd_dev->lock);
7195 spin_unlock(&rbd_dev_list_lock);
7201 * Prevent new IO from being queued and wait for existing
7202 * IO to complete/fail.
7204 blk_mq_freeze_queue(rbd_dev->disk->queue);
7205 blk_mark_disk_dead(rbd_dev->disk);
7208 del_gendisk(rbd_dev->disk);
7209 spin_lock(&rbd_dev_list_lock);
7210 list_del_init(&rbd_dev->node);
7211 spin_unlock(&rbd_dev_list_lock);
7212 device_del(&rbd_dev->dev);
7214 rbd_dev_image_unlock(rbd_dev);
7215 rbd_dev_device_release(rbd_dev);
7216 rbd_dev_image_release(rbd_dev);
7217 rbd_dev_destroy(rbd_dev);
7221 static ssize_t remove_store(const struct bus_type *bus, const char *buf, size_t count)
7226 return do_rbd_remove(buf, count);
7229 static ssize_t remove_single_major_store(const struct bus_type *bus, const char *buf,
7232 return do_rbd_remove(buf, count);
7236 * create control files in sysfs
7239 static int __init rbd_sysfs_init(void)
7243 ret = device_register(&rbd_root_dev);
7245 put_device(&rbd_root_dev);
7249 ret = bus_register(&rbd_bus_type);
7251 device_unregister(&rbd_root_dev);
7256 static void __exit rbd_sysfs_cleanup(void)
7258 bus_unregister(&rbd_bus_type);
7259 device_unregister(&rbd_root_dev);
7262 static int __init rbd_slab_init(void)
7264 rbd_assert(!rbd_img_request_cache);
7265 rbd_img_request_cache = KMEM_CACHE(rbd_img_request, 0);
7266 if (!rbd_img_request_cache)
7269 rbd_assert(!rbd_obj_request_cache);
7270 rbd_obj_request_cache = KMEM_CACHE(rbd_obj_request, 0);
7271 if (!rbd_obj_request_cache)
7277 kmem_cache_destroy(rbd_img_request_cache);
7278 rbd_img_request_cache = NULL;
7282 static void rbd_slab_exit(void)
7284 rbd_assert(rbd_obj_request_cache);
7285 kmem_cache_destroy(rbd_obj_request_cache);
7286 rbd_obj_request_cache = NULL;
7288 rbd_assert(rbd_img_request_cache);
7289 kmem_cache_destroy(rbd_img_request_cache);
7290 rbd_img_request_cache = NULL;
7293 static int __init rbd_init(void)
7297 if (!libceph_compatible(NULL)) {
7298 rbd_warn(NULL, "libceph incompatibility (quitting)");
7302 rc = rbd_slab_init();
7307 * The number of active work items is limited by the number of
7308 * rbd devices * queue depth, so leave @max_active at default.
7310 rbd_wq = alloc_workqueue(RBD_DRV_NAME, WQ_MEM_RECLAIM, 0);
7317 rbd_major = register_blkdev(0, RBD_DRV_NAME);
7318 if (rbd_major < 0) {
7324 rc = rbd_sysfs_init();
7326 goto err_out_blkdev;
7329 pr_info("loaded (major %d)\n", rbd_major);
7331 pr_info("loaded\n");
7337 unregister_blkdev(rbd_major, RBD_DRV_NAME);
7339 destroy_workqueue(rbd_wq);
7345 static void __exit rbd_exit(void)
7347 ida_destroy(&rbd_dev_id_ida);
7348 rbd_sysfs_cleanup();
7350 unregister_blkdev(rbd_major, RBD_DRV_NAME);
7351 destroy_workqueue(rbd_wq);
7355 module_init(rbd_init);
7356 module_exit(rbd_exit);
7358 MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
7359 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
7360 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
7361 /* following authorship retained from original osdblk.c */
7362 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
7364 MODULE_DESCRIPTION("RADOS Block Device (RBD) driver");
7365 MODULE_LICENSE("GPL");