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(struct bus_type *bus, const char *buf, size_t count);
495 static ssize_t remove_store(struct bus_type *bus, const char *buf,
497 static ssize_t add_single_major_store(struct bus_type *bus, const char *buf,
499 static ssize_t remove_single_major_store(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(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 block_device *bdev, fmode_t mode)
665 struct rbd_device *rbd_dev = bdev->bd_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, fmode_t mode)
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 int rbd_set_read_only(struct block_device *bdev, bool ro)
697 struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
700 * Both images mapped read-only and snapshots can't be marked
704 if (rbd_is_ro(rbd_dev))
707 rbd_assert(!rbd_is_snap(rbd_dev));
713 static const struct block_device_operations rbd_bd_ops = {
714 .owner = THIS_MODULE,
716 .release = rbd_release,
717 .set_read_only = rbd_set_read_only,
721 * Initialize an rbd client instance. Success or not, this function
722 * consumes ceph_opts. Caller holds client_mutex.
724 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
726 struct rbd_client *rbdc;
729 dout("%s:\n", __func__);
730 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
734 kref_init(&rbdc->kref);
735 INIT_LIST_HEAD(&rbdc->node);
737 rbdc->client = ceph_create_client(ceph_opts, rbdc);
738 if (IS_ERR(rbdc->client))
740 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
742 ret = ceph_open_session(rbdc->client);
746 spin_lock(&rbd_client_list_lock);
747 list_add_tail(&rbdc->node, &rbd_client_list);
748 spin_unlock(&rbd_client_list_lock);
750 dout("%s: rbdc %p\n", __func__, rbdc);
754 ceph_destroy_client(rbdc->client);
759 ceph_destroy_options(ceph_opts);
760 dout("%s: error %d\n", __func__, ret);
765 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
767 kref_get(&rbdc->kref);
773 * Find a ceph client with specific addr and configuration. If
774 * found, bump its reference count.
776 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
778 struct rbd_client *client_node;
781 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
784 spin_lock(&rbd_client_list_lock);
785 list_for_each_entry(client_node, &rbd_client_list, node) {
786 if (!ceph_compare_options(ceph_opts, client_node->client)) {
787 __rbd_get_client(client_node);
793 spin_unlock(&rbd_client_list_lock);
795 return found ? client_node : NULL;
799 * (Per device) rbd map options
807 Opt_compression_hint,
808 /* string args above */
817 Opt_compression_hint_none,
818 Opt_compression_hint_compressible,
819 Opt_compression_hint_incompressible,
822 static const struct constant_table rbd_param_compression_hint[] = {
823 {"none", Opt_compression_hint_none},
824 {"compressible", Opt_compression_hint_compressible},
825 {"incompressible", Opt_compression_hint_incompressible},
829 static const struct fs_parameter_spec rbd_parameters[] = {
830 fsparam_u32 ("alloc_size", Opt_alloc_size),
831 fsparam_enum ("compression_hint", Opt_compression_hint,
832 rbd_param_compression_hint),
833 fsparam_flag ("exclusive", Opt_exclusive),
834 fsparam_flag ("lock_on_read", Opt_lock_on_read),
835 fsparam_u32 ("lock_timeout", Opt_lock_timeout),
836 fsparam_flag ("notrim", Opt_notrim),
837 fsparam_string ("_pool_ns", Opt_pool_ns),
838 fsparam_u32 ("queue_depth", Opt_queue_depth),
839 fsparam_flag ("read_only", Opt_read_only),
840 fsparam_flag ("read_write", Opt_read_write),
841 fsparam_flag ("ro", Opt_read_only),
842 fsparam_flag ("rw", Opt_read_write),
849 unsigned long lock_timeout;
855 u32 alloc_hint_flags; /* CEPH_OSD_OP_ALLOC_HINT_FLAG_* */
858 #define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_MAX_RQ
859 #define RBD_ALLOC_SIZE_DEFAULT (64 * 1024)
860 #define RBD_LOCK_TIMEOUT_DEFAULT 0 /* no timeout */
861 #define RBD_READ_ONLY_DEFAULT false
862 #define RBD_LOCK_ON_READ_DEFAULT false
863 #define RBD_EXCLUSIVE_DEFAULT false
864 #define RBD_TRIM_DEFAULT true
866 struct rbd_parse_opts_ctx {
867 struct rbd_spec *spec;
868 struct ceph_options *copts;
869 struct rbd_options *opts;
872 static char* obj_op_name(enum obj_operation_type op_type)
889 * Destroy ceph client
891 * Caller must hold rbd_client_list_lock.
893 static void rbd_client_release(struct kref *kref)
895 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
897 dout("%s: rbdc %p\n", __func__, rbdc);
898 spin_lock(&rbd_client_list_lock);
899 list_del(&rbdc->node);
900 spin_unlock(&rbd_client_list_lock);
902 ceph_destroy_client(rbdc->client);
907 * Drop reference to ceph client node. If it's not referenced anymore, release
910 static void rbd_put_client(struct rbd_client *rbdc)
913 kref_put(&rbdc->kref, rbd_client_release);
917 * Get a ceph client with specific addr and configuration, if one does
918 * not exist create it. Either way, ceph_opts is consumed by this
921 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
923 struct rbd_client *rbdc;
926 mutex_lock(&client_mutex);
927 rbdc = rbd_client_find(ceph_opts);
929 ceph_destroy_options(ceph_opts);
932 * Using an existing client. Make sure ->pg_pools is up to
933 * date before we look up the pool id in do_rbd_add().
935 ret = ceph_wait_for_latest_osdmap(rbdc->client,
936 rbdc->client->options->mount_timeout);
938 rbd_warn(NULL, "failed to get latest osdmap: %d", ret);
939 rbd_put_client(rbdc);
943 rbdc = rbd_client_create(ceph_opts);
945 mutex_unlock(&client_mutex);
950 static bool rbd_image_format_valid(u32 image_format)
952 return image_format == 1 || image_format == 2;
955 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
960 /* The header has to start with the magic rbd header text */
961 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
964 /* The bio layer requires at least sector-sized I/O */
966 if (ondisk->options.order < SECTOR_SHIFT)
969 /* If we use u64 in a few spots we may be able to loosen this */
971 if (ondisk->options.order > 8 * sizeof (int) - 1)
975 * The size of a snapshot header has to fit in a size_t, and
976 * that limits the number of snapshots.
978 snap_count = le32_to_cpu(ondisk->snap_count);
979 size = SIZE_MAX - sizeof (struct ceph_snap_context);
980 if (snap_count > size / sizeof (__le64))
984 * Not only that, but the size of the entire the snapshot
985 * header must also be representable in a size_t.
987 size -= snap_count * sizeof (__le64);
988 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
995 * returns the size of an object in the image
997 static u32 rbd_obj_bytes(struct rbd_image_header *header)
999 return 1U << header->obj_order;
1002 static void rbd_init_layout(struct rbd_device *rbd_dev)
1004 if (rbd_dev->header.stripe_unit == 0 ||
1005 rbd_dev->header.stripe_count == 0) {
1006 rbd_dev->header.stripe_unit = rbd_obj_bytes(&rbd_dev->header);
1007 rbd_dev->header.stripe_count = 1;
1010 rbd_dev->layout.stripe_unit = rbd_dev->header.stripe_unit;
1011 rbd_dev->layout.stripe_count = rbd_dev->header.stripe_count;
1012 rbd_dev->layout.object_size = rbd_obj_bytes(&rbd_dev->header);
1013 rbd_dev->layout.pool_id = rbd_dev->header.data_pool_id == CEPH_NOPOOL ?
1014 rbd_dev->spec->pool_id : rbd_dev->header.data_pool_id;
1015 RCU_INIT_POINTER(rbd_dev->layout.pool_ns, NULL);
1019 * Fill an rbd image header with information from the given format 1
1022 static int rbd_header_from_disk(struct rbd_device *rbd_dev,
1023 struct rbd_image_header_ondisk *ondisk)
1025 struct rbd_image_header *header = &rbd_dev->header;
1026 bool first_time = header->object_prefix == NULL;
1027 struct ceph_snap_context *snapc;
1028 char *object_prefix = NULL;
1029 char *snap_names = NULL;
1030 u64 *snap_sizes = NULL;
1035 /* Allocate this now to avoid having to handle failure below */
1038 object_prefix = kstrndup(ondisk->object_prefix,
1039 sizeof(ondisk->object_prefix),
1045 /* Allocate the snapshot context and fill it in */
1047 snap_count = le32_to_cpu(ondisk->snap_count);
1048 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
1051 snapc->seq = le64_to_cpu(ondisk->snap_seq);
1053 struct rbd_image_snap_ondisk *snaps;
1054 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
1056 /* We'll keep a copy of the snapshot names... */
1058 if (snap_names_len > (u64)SIZE_MAX)
1060 snap_names = kmalloc(snap_names_len, GFP_KERNEL);
1064 /* ...as well as the array of their sizes. */
1065 snap_sizes = kmalloc_array(snap_count,
1066 sizeof(*header->snap_sizes),
1072 * Copy the names, and fill in each snapshot's id
1075 * Note that rbd_dev_v1_header_info() guarantees the
1076 * ondisk buffer we're working with has
1077 * snap_names_len bytes beyond the end of the
1078 * snapshot id array, this memcpy() is safe.
1080 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
1081 snaps = ondisk->snaps;
1082 for (i = 0; i < snap_count; i++) {
1083 snapc->snaps[i] = le64_to_cpu(snaps[i].id);
1084 snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
1088 /* We won't fail any more, fill in the header */
1091 header->object_prefix = object_prefix;
1092 header->obj_order = ondisk->options.order;
1093 rbd_init_layout(rbd_dev);
1095 ceph_put_snap_context(header->snapc);
1096 kfree(header->snap_names);
1097 kfree(header->snap_sizes);
1100 /* The remaining fields always get updated (when we refresh) */
1102 header->image_size = le64_to_cpu(ondisk->image_size);
1103 header->snapc = snapc;
1104 header->snap_names = snap_names;
1105 header->snap_sizes = snap_sizes;
1113 ceph_put_snap_context(snapc);
1114 kfree(object_prefix);
1119 static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1121 const char *snap_name;
1123 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1125 /* Skip over names until we find the one we are looking for */
1127 snap_name = rbd_dev->header.snap_names;
1129 snap_name += strlen(snap_name) + 1;
1131 return kstrdup(snap_name, GFP_KERNEL);
1135 * Snapshot id comparison function for use with qsort()/bsearch().
1136 * Note that result is for snapshots in *descending* order.
1138 static int snapid_compare_reverse(const void *s1, const void *s2)
1140 u64 snap_id1 = *(u64 *)s1;
1141 u64 snap_id2 = *(u64 *)s2;
1143 if (snap_id1 < snap_id2)
1145 return snap_id1 == snap_id2 ? 0 : -1;
1149 * Search a snapshot context to see if the given snapshot id is
1152 * Returns the position of the snapshot id in the array if it's found,
1153 * or BAD_SNAP_INDEX otherwise.
1155 * Note: The snapshot array is in kept sorted (by the osd) in
1156 * reverse order, highest snapshot id first.
1158 static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1160 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1163 found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1164 sizeof (snap_id), snapid_compare_reverse);
1166 return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1169 static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1173 const char *snap_name;
1175 which = rbd_dev_snap_index(rbd_dev, snap_id);
1176 if (which == BAD_SNAP_INDEX)
1177 return ERR_PTR(-ENOENT);
1179 snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1180 return snap_name ? snap_name : ERR_PTR(-ENOMEM);
1183 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1185 if (snap_id == CEPH_NOSNAP)
1186 return RBD_SNAP_HEAD_NAME;
1188 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1189 if (rbd_dev->image_format == 1)
1190 return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1192 return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1195 static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1198 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1199 if (snap_id == CEPH_NOSNAP) {
1200 *snap_size = rbd_dev->header.image_size;
1201 } else if (rbd_dev->image_format == 1) {
1204 which = rbd_dev_snap_index(rbd_dev, snap_id);
1205 if (which == BAD_SNAP_INDEX)
1208 *snap_size = rbd_dev->header.snap_sizes[which];
1213 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1222 static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1224 u64 snap_id = rbd_dev->spec->snap_id;
1228 ret = rbd_snap_size(rbd_dev, snap_id, &size);
1232 rbd_dev->mapping.size = size;
1236 static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1238 rbd_dev->mapping.size = 0;
1241 static void zero_bvec(struct bio_vec *bv)
1244 unsigned long flags;
1246 buf = bvec_kmap_irq(bv, &flags);
1247 memset(buf, 0, bv->bv_len);
1248 flush_dcache_page(bv->bv_page);
1249 bvec_kunmap_irq(buf, &flags);
1252 static void zero_bios(struct ceph_bio_iter *bio_pos, u32 off, u32 bytes)
1254 struct ceph_bio_iter it = *bio_pos;
1256 ceph_bio_iter_advance(&it, off);
1257 ceph_bio_iter_advance_step(&it, bytes, ({
1262 static void zero_bvecs(struct ceph_bvec_iter *bvec_pos, u32 off, u32 bytes)
1264 struct ceph_bvec_iter it = *bvec_pos;
1266 ceph_bvec_iter_advance(&it, off);
1267 ceph_bvec_iter_advance_step(&it, bytes, ({
1273 * Zero a range in @obj_req data buffer defined by a bio (list) or
1274 * (private) bio_vec array.
1276 * @off is relative to the start of the data buffer.
1278 static void rbd_obj_zero_range(struct rbd_obj_request *obj_req, u32 off,
1281 dout("%s %p data buf %u~%u\n", __func__, obj_req, off, bytes);
1283 switch (obj_req->img_request->data_type) {
1284 case OBJ_REQUEST_BIO:
1285 zero_bios(&obj_req->bio_pos, off, bytes);
1287 case OBJ_REQUEST_BVECS:
1288 case OBJ_REQUEST_OWN_BVECS:
1289 zero_bvecs(&obj_req->bvec_pos, off, bytes);
1296 static void rbd_obj_request_destroy(struct kref *kref);
1297 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1299 rbd_assert(obj_request != NULL);
1300 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1301 kref_read(&obj_request->kref));
1302 kref_put(&obj_request->kref, rbd_obj_request_destroy);
1305 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1306 struct rbd_obj_request *obj_request)
1308 rbd_assert(obj_request->img_request == NULL);
1310 /* Image request now owns object's original reference */
1311 obj_request->img_request = img_request;
1312 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1315 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1316 struct rbd_obj_request *obj_request)
1318 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1319 list_del(&obj_request->ex.oe_item);
1320 rbd_assert(obj_request->img_request == img_request);
1321 rbd_obj_request_put(obj_request);
1324 static void rbd_osd_submit(struct ceph_osd_request *osd_req)
1326 struct rbd_obj_request *obj_req = osd_req->r_priv;
1328 dout("%s osd_req %p for obj_req %p objno %llu %llu~%llu\n",
1329 __func__, osd_req, obj_req, obj_req->ex.oe_objno,
1330 obj_req->ex.oe_off, obj_req->ex.oe_len);
1331 ceph_osdc_start_request(osd_req->r_osdc, osd_req, false);
1335 * The default/initial value for all image request flags is 0. Each
1336 * is conditionally set to 1 at image request initialization time
1337 * and currently never change thereafter.
1339 static void img_request_layered_set(struct rbd_img_request *img_request)
1341 set_bit(IMG_REQ_LAYERED, &img_request->flags);
1344 static bool img_request_layered_test(struct rbd_img_request *img_request)
1346 return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1349 static bool rbd_obj_is_entire(struct rbd_obj_request *obj_req)
1351 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1353 return !obj_req->ex.oe_off &&
1354 obj_req->ex.oe_len == rbd_dev->layout.object_size;
1357 static bool rbd_obj_is_tail(struct rbd_obj_request *obj_req)
1359 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1361 return obj_req->ex.oe_off + obj_req->ex.oe_len ==
1362 rbd_dev->layout.object_size;
1366 * Must be called after rbd_obj_calc_img_extents().
1368 static bool rbd_obj_copyup_enabled(struct rbd_obj_request *obj_req)
1370 if (!obj_req->num_img_extents ||
1371 (rbd_obj_is_entire(obj_req) &&
1372 !obj_req->img_request->snapc->num_snaps))
1378 static u64 rbd_obj_img_extents_bytes(struct rbd_obj_request *obj_req)
1380 return ceph_file_extents_bytes(obj_req->img_extents,
1381 obj_req->num_img_extents);
1384 static bool rbd_img_is_write(struct rbd_img_request *img_req)
1386 switch (img_req->op_type) {
1390 case OBJ_OP_DISCARD:
1391 case OBJ_OP_ZEROOUT:
1398 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
1400 struct rbd_obj_request *obj_req = osd_req->r_priv;
1403 dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
1404 osd_req->r_result, obj_req);
1407 * Writes aren't allowed to return a data payload. In some
1408 * guarded write cases (e.g. stat + zero on an empty object)
1409 * a stat response makes it through, but we don't care.
1411 if (osd_req->r_result > 0 && rbd_img_is_write(obj_req->img_request))
1414 result = osd_req->r_result;
1416 rbd_obj_handle_request(obj_req, result);
1419 static void rbd_osd_format_read(struct ceph_osd_request *osd_req)
1421 struct rbd_obj_request *obj_request = osd_req->r_priv;
1422 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1423 struct ceph_options *opt = rbd_dev->rbd_client->client->options;
1425 osd_req->r_flags = CEPH_OSD_FLAG_READ | opt->read_from_replica;
1426 osd_req->r_snapid = obj_request->img_request->snap_id;
1429 static void rbd_osd_format_write(struct ceph_osd_request *osd_req)
1431 struct rbd_obj_request *obj_request = osd_req->r_priv;
1433 osd_req->r_flags = CEPH_OSD_FLAG_WRITE;
1434 ktime_get_real_ts64(&osd_req->r_mtime);
1435 osd_req->r_data_offset = obj_request->ex.oe_off;
1438 static struct ceph_osd_request *
1439 __rbd_obj_add_osd_request(struct rbd_obj_request *obj_req,
1440 struct ceph_snap_context *snapc, int num_ops)
1442 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1443 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1444 struct ceph_osd_request *req;
1445 const char *name_format = rbd_dev->image_format == 1 ?
1446 RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
1449 req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, GFP_NOIO);
1451 return ERR_PTR(-ENOMEM);
1453 list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
1454 req->r_callback = rbd_osd_req_callback;
1455 req->r_priv = obj_req;
1458 * Data objects may be stored in a separate pool, but always in
1459 * the same namespace in that pool as the header in its pool.
1461 ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
1462 req->r_base_oloc.pool = rbd_dev->layout.pool_id;
1464 ret = ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
1465 rbd_dev->header.object_prefix,
1466 obj_req->ex.oe_objno);
1468 return ERR_PTR(ret);
1473 static struct ceph_osd_request *
1474 rbd_obj_add_osd_request(struct rbd_obj_request *obj_req, int num_ops)
1476 return __rbd_obj_add_osd_request(obj_req, obj_req->img_request->snapc,
1480 static struct rbd_obj_request *rbd_obj_request_create(void)
1482 struct rbd_obj_request *obj_request;
1484 obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_NOIO);
1488 ceph_object_extent_init(&obj_request->ex);
1489 INIT_LIST_HEAD(&obj_request->osd_reqs);
1490 mutex_init(&obj_request->state_mutex);
1491 kref_init(&obj_request->kref);
1493 dout("%s %p\n", __func__, obj_request);
1497 static void rbd_obj_request_destroy(struct kref *kref)
1499 struct rbd_obj_request *obj_request;
1500 struct ceph_osd_request *osd_req;
1503 obj_request = container_of(kref, struct rbd_obj_request, kref);
1505 dout("%s: obj %p\n", __func__, obj_request);
1507 while (!list_empty(&obj_request->osd_reqs)) {
1508 osd_req = list_first_entry(&obj_request->osd_reqs,
1509 struct ceph_osd_request, r_private_item);
1510 list_del_init(&osd_req->r_private_item);
1511 ceph_osdc_put_request(osd_req);
1514 switch (obj_request->img_request->data_type) {
1515 case OBJ_REQUEST_NODATA:
1516 case OBJ_REQUEST_BIO:
1517 case OBJ_REQUEST_BVECS:
1518 break; /* Nothing to do */
1519 case OBJ_REQUEST_OWN_BVECS:
1520 kfree(obj_request->bvec_pos.bvecs);
1526 kfree(obj_request->img_extents);
1527 if (obj_request->copyup_bvecs) {
1528 for (i = 0; i < obj_request->copyup_bvec_count; i++) {
1529 if (obj_request->copyup_bvecs[i].bv_page)
1530 __free_page(obj_request->copyup_bvecs[i].bv_page);
1532 kfree(obj_request->copyup_bvecs);
1535 kmem_cache_free(rbd_obj_request_cache, obj_request);
1538 /* It's OK to call this for a device with no parent */
1540 static void rbd_spec_put(struct rbd_spec *spec);
1541 static void rbd_dev_unparent(struct rbd_device *rbd_dev)
1543 rbd_dev_remove_parent(rbd_dev);
1544 rbd_spec_put(rbd_dev->parent_spec);
1545 rbd_dev->parent_spec = NULL;
1546 rbd_dev->parent_overlap = 0;
1550 * Parent image reference counting is used to determine when an
1551 * image's parent fields can be safely torn down--after there are no
1552 * more in-flight requests to the parent image. When the last
1553 * reference is dropped, cleaning them up is safe.
1555 static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
1559 if (!rbd_dev->parent_spec)
1562 counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
1566 /* Last reference; clean up parent data structures */
1569 rbd_dev_unparent(rbd_dev);
1571 rbd_warn(rbd_dev, "parent reference underflow");
1575 * If an image has a non-zero parent overlap, get a reference to its
1578 * Returns true if the rbd device has a parent with a non-zero
1579 * overlap and a reference for it was successfully taken, or
1582 static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
1586 if (!rbd_dev->parent_spec)
1589 if (rbd_dev->parent_overlap)
1590 counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
1593 rbd_warn(rbd_dev, "parent reference overflow");
1598 static void rbd_img_request_init(struct rbd_img_request *img_request,
1599 struct rbd_device *rbd_dev,
1600 enum obj_operation_type op_type)
1602 memset(img_request, 0, sizeof(*img_request));
1604 img_request->rbd_dev = rbd_dev;
1605 img_request->op_type = op_type;
1607 INIT_LIST_HEAD(&img_request->lock_item);
1608 INIT_LIST_HEAD(&img_request->object_extents);
1609 mutex_init(&img_request->state_mutex);
1612 static void rbd_img_capture_header(struct rbd_img_request *img_req)
1614 struct rbd_device *rbd_dev = img_req->rbd_dev;
1616 lockdep_assert_held(&rbd_dev->header_rwsem);
1618 if (rbd_img_is_write(img_req))
1619 img_req->snapc = ceph_get_snap_context(rbd_dev->header.snapc);
1621 img_req->snap_id = rbd_dev->spec->snap_id;
1623 if (rbd_dev_parent_get(rbd_dev))
1624 img_request_layered_set(img_req);
1627 static void rbd_img_request_destroy(struct rbd_img_request *img_request)
1629 struct rbd_obj_request *obj_request;
1630 struct rbd_obj_request *next_obj_request;
1632 dout("%s: img %p\n", __func__, img_request);
1634 WARN_ON(!list_empty(&img_request->lock_item));
1635 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1636 rbd_img_obj_request_del(img_request, obj_request);
1638 if (img_request_layered_test(img_request))
1639 rbd_dev_parent_put(img_request->rbd_dev);
1641 if (rbd_img_is_write(img_request))
1642 ceph_put_snap_context(img_request->snapc);
1644 if (test_bit(IMG_REQ_CHILD, &img_request->flags))
1645 kmem_cache_free(rbd_img_request_cache, img_request);
1648 #define BITS_PER_OBJ 2
1649 #define OBJS_PER_BYTE (BITS_PER_BYTE / BITS_PER_OBJ)
1650 #define OBJ_MASK ((1 << BITS_PER_OBJ) - 1)
1652 static void __rbd_object_map_index(struct rbd_device *rbd_dev, u64 objno,
1653 u64 *index, u8 *shift)
1657 rbd_assert(objno < rbd_dev->object_map_size);
1658 *index = div_u64_rem(objno, OBJS_PER_BYTE, &off);
1659 *shift = (OBJS_PER_BYTE - off - 1) * BITS_PER_OBJ;
1662 static u8 __rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1667 lockdep_assert_held(&rbd_dev->object_map_lock);
1668 __rbd_object_map_index(rbd_dev, objno, &index, &shift);
1669 return (rbd_dev->object_map[index] >> shift) & OBJ_MASK;
1672 static void __rbd_object_map_set(struct rbd_device *rbd_dev, u64 objno, u8 val)
1678 lockdep_assert_held(&rbd_dev->object_map_lock);
1679 rbd_assert(!(val & ~OBJ_MASK));
1681 __rbd_object_map_index(rbd_dev, objno, &index, &shift);
1682 p = &rbd_dev->object_map[index];
1683 *p = (*p & ~(OBJ_MASK << shift)) | (val << shift);
1686 static u8 rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1690 spin_lock(&rbd_dev->object_map_lock);
1691 state = __rbd_object_map_get(rbd_dev, objno);
1692 spin_unlock(&rbd_dev->object_map_lock);
1696 static bool use_object_map(struct rbd_device *rbd_dev)
1699 * An image mapped read-only can't use the object map -- it isn't
1700 * loaded because the header lock isn't acquired. Someone else can
1701 * write to the image and update the object map behind our back.
1703 * A snapshot can't be written to, so using the object map is always
1706 if (!rbd_is_snap(rbd_dev) && rbd_is_ro(rbd_dev))
1709 return ((rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) &&
1710 !(rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID));
1713 static bool rbd_object_map_may_exist(struct rbd_device *rbd_dev, u64 objno)
1717 /* fall back to default logic if object map is disabled or invalid */
1718 if (!use_object_map(rbd_dev))
1721 state = rbd_object_map_get(rbd_dev, objno);
1722 return state != OBJECT_NONEXISTENT;
1725 static void rbd_object_map_name(struct rbd_device *rbd_dev, u64 snap_id,
1726 struct ceph_object_id *oid)
1728 if (snap_id == CEPH_NOSNAP)
1729 ceph_oid_printf(oid, "%s%s", RBD_OBJECT_MAP_PREFIX,
1730 rbd_dev->spec->image_id);
1732 ceph_oid_printf(oid, "%s%s.%016llx", RBD_OBJECT_MAP_PREFIX,
1733 rbd_dev->spec->image_id, snap_id);
1736 static int rbd_object_map_lock(struct rbd_device *rbd_dev)
1738 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1739 CEPH_DEFINE_OID_ONSTACK(oid);
1742 struct ceph_locker *lockers;
1744 bool broke_lock = false;
1747 rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1750 ret = ceph_cls_lock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1751 CEPH_CLS_LOCK_EXCLUSIVE, "", "", "", 0);
1752 if (ret != -EBUSY || broke_lock) {
1754 ret = 0; /* already locked by myself */
1756 rbd_warn(rbd_dev, "failed to lock object map: %d", ret);
1760 ret = ceph_cls_lock_info(osdc, &oid, &rbd_dev->header_oloc,
1761 RBD_LOCK_NAME, &lock_type, &lock_tag,
1762 &lockers, &num_lockers);
1767 rbd_warn(rbd_dev, "failed to get object map lockers: %d", ret);
1772 if (num_lockers == 0)
1775 rbd_warn(rbd_dev, "breaking object map lock owned by %s%llu",
1776 ENTITY_NAME(lockers[0].id.name));
1778 ret = ceph_cls_break_lock(osdc, &oid, &rbd_dev->header_oloc,
1779 RBD_LOCK_NAME, lockers[0].id.cookie,
1780 &lockers[0].id.name);
1781 ceph_free_lockers(lockers, num_lockers);
1786 rbd_warn(rbd_dev, "failed to break object map lock: %d", ret);
1794 static void rbd_object_map_unlock(struct rbd_device *rbd_dev)
1796 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1797 CEPH_DEFINE_OID_ONSTACK(oid);
1800 rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1802 ret = ceph_cls_unlock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1804 if (ret && ret != -ENOENT)
1805 rbd_warn(rbd_dev, "failed to unlock object map: %d", ret);
1808 static int decode_object_map_header(void **p, void *end, u64 *object_map_size)
1816 ceph_decode_32_safe(p, end, header_len, e_inval);
1817 header_end = *p + header_len;
1819 ret = ceph_start_decoding(p, end, 1, "BitVector header", &struct_v,
1824 ceph_decode_64_safe(p, end, *object_map_size, e_inval);
1833 static int __rbd_object_map_load(struct rbd_device *rbd_dev)
1835 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1836 CEPH_DEFINE_OID_ONSTACK(oid);
1837 struct page **pages;
1841 u64 object_map_bytes;
1842 u64 object_map_size;
1846 rbd_assert(!rbd_dev->object_map && !rbd_dev->object_map_size);
1848 num_objects = ceph_get_num_objects(&rbd_dev->layout,
1849 rbd_dev->mapping.size);
1850 object_map_bytes = DIV_ROUND_UP_ULL(num_objects * BITS_PER_OBJ,
1852 num_pages = calc_pages_for(0, object_map_bytes) + 1;
1853 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1855 return PTR_ERR(pages);
1857 reply_len = num_pages * PAGE_SIZE;
1858 rbd_object_map_name(rbd_dev, rbd_dev->spec->snap_id, &oid);
1859 ret = ceph_osdc_call(osdc, &oid, &rbd_dev->header_oloc,
1860 "rbd", "object_map_load", CEPH_OSD_FLAG_READ,
1861 NULL, 0, pages, &reply_len);
1865 p = page_address(pages[0]);
1866 end = p + min(reply_len, (size_t)PAGE_SIZE);
1867 ret = decode_object_map_header(&p, end, &object_map_size);
1871 if (object_map_size != num_objects) {
1872 rbd_warn(rbd_dev, "object map size mismatch: %llu vs %llu",
1873 object_map_size, num_objects);
1878 if (offset_in_page(p) + object_map_bytes > reply_len) {
1883 rbd_dev->object_map = kvmalloc(object_map_bytes, GFP_KERNEL);
1884 if (!rbd_dev->object_map) {
1889 rbd_dev->object_map_size = object_map_size;
1890 ceph_copy_from_page_vector(pages, rbd_dev->object_map,
1891 offset_in_page(p), object_map_bytes);
1894 ceph_release_page_vector(pages, num_pages);
1898 static void rbd_object_map_free(struct rbd_device *rbd_dev)
1900 kvfree(rbd_dev->object_map);
1901 rbd_dev->object_map = NULL;
1902 rbd_dev->object_map_size = 0;
1905 static int rbd_object_map_load(struct rbd_device *rbd_dev)
1909 ret = __rbd_object_map_load(rbd_dev);
1913 ret = rbd_dev_v2_get_flags(rbd_dev);
1915 rbd_object_map_free(rbd_dev);
1919 if (rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID)
1920 rbd_warn(rbd_dev, "object map is invalid");
1925 static int rbd_object_map_open(struct rbd_device *rbd_dev)
1929 ret = rbd_object_map_lock(rbd_dev);
1933 ret = rbd_object_map_load(rbd_dev);
1935 rbd_object_map_unlock(rbd_dev);
1942 static void rbd_object_map_close(struct rbd_device *rbd_dev)
1944 rbd_object_map_free(rbd_dev);
1945 rbd_object_map_unlock(rbd_dev);
1949 * This function needs snap_id (or more precisely just something to
1950 * distinguish between HEAD and snapshot object maps), new_state and
1951 * current_state that were passed to rbd_object_map_update().
1953 * To avoid allocating and stashing a context we piggyback on the OSD
1954 * request. A HEAD update has two ops (assert_locked). For new_state
1955 * and current_state we decode our own object_map_update op, encoded in
1956 * rbd_cls_object_map_update().
1958 static int rbd_object_map_update_finish(struct rbd_obj_request *obj_req,
1959 struct ceph_osd_request *osd_req)
1961 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1962 struct ceph_osd_data *osd_data;
1964 u8 state, new_state, current_state;
1965 bool has_current_state;
1968 if (osd_req->r_result)
1969 return osd_req->r_result;
1972 * Nothing to do for a snapshot object map.
1974 if (osd_req->r_num_ops == 1)
1978 * Update in-memory HEAD object map.
1980 rbd_assert(osd_req->r_num_ops == 2);
1981 osd_data = osd_req_op_data(osd_req, 1, cls, request_data);
1982 rbd_assert(osd_data->type == CEPH_OSD_DATA_TYPE_PAGES);
1984 p = page_address(osd_data->pages[0]);
1985 objno = ceph_decode_64(&p);
1986 rbd_assert(objno == obj_req->ex.oe_objno);
1987 rbd_assert(ceph_decode_64(&p) == objno + 1);
1988 new_state = ceph_decode_8(&p);
1989 has_current_state = ceph_decode_8(&p);
1990 if (has_current_state)
1991 current_state = ceph_decode_8(&p);
1993 spin_lock(&rbd_dev->object_map_lock);
1994 state = __rbd_object_map_get(rbd_dev, objno);
1995 if (!has_current_state || current_state == state ||
1996 (current_state == OBJECT_EXISTS && state == OBJECT_EXISTS_CLEAN))
1997 __rbd_object_map_set(rbd_dev, objno, new_state);
1998 spin_unlock(&rbd_dev->object_map_lock);
2003 static void rbd_object_map_callback(struct ceph_osd_request *osd_req)
2005 struct rbd_obj_request *obj_req = osd_req->r_priv;
2008 dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
2009 osd_req->r_result, obj_req);
2011 result = rbd_object_map_update_finish(obj_req, osd_req);
2012 rbd_obj_handle_request(obj_req, result);
2015 static bool update_needed(struct rbd_device *rbd_dev, u64 objno, u8 new_state)
2017 u8 state = rbd_object_map_get(rbd_dev, objno);
2019 if (state == new_state ||
2020 (new_state == OBJECT_PENDING && state == OBJECT_NONEXISTENT) ||
2021 (new_state == OBJECT_NONEXISTENT && state != OBJECT_PENDING))
2027 static int rbd_cls_object_map_update(struct ceph_osd_request *req,
2028 int which, u64 objno, u8 new_state,
2029 const u8 *current_state)
2031 struct page **pages;
2035 ret = osd_req_op_cls_init(req, which, "rbd", "object_map_update");
2039 pages = ceph_alloc_page_vector(1, GFP_NOIO);
2041 return PTR_ERR(pages);
2043 p = start = page_address(pages[0]);
2044 ceph_encode_64(&p, objno);
2045 ceph_encode_64(&p, objno + 1);
2046 ceph_encode_8(&p, new_state);
2047 if (current_state) {
2048 ceph_encode_8(&p, 1);
2049 ceph_encode_8(&p, *current_state);
2051 ceph_encode_8(&p, 0);
2054 osd_req_op_cls_request_data_pages(req, which, pages, p - start, 0,
2061 * 0 - object map update sent
2062 * 1 - object map update isn't needed
2065 static int rbd_object_map_update(struct rbd_obj_request *obj_req, u64 snap_id,
2066 u8 new_state, const u8 *current_state)
2068 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2069 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2070 struct ceph_osd_request *req;
2075 if (snap_id == CEPH_NOSNAP) {
2076 if (!update_needed(rbd_dev, obj_req->ex.oe_objno, new_state))
2079 num_ops++; /* assert_locked */
2082 req = ceph_osdc_alloc_request(osdc, NULL, num_ops, false, GFP_NOIO);
2086 list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
2087 req->r_callback = rbd_object_map_callback;
2088 req->r_priv = obj_req;
2090 rbd_object_map_name(rbd_dev, snap_id, &req->r_base_oid);
2091 ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
2092 req->r_flags = CEPH_OSD_FLAG_WRITE;
2093 ktime_get_real_ts64(&req->r_mtime);
2095 if (snap_id == CEPH_NOSNAP) {
2097 * Protect against possible race conditions during lock
2098 * ownership transitions.
2100 ret = ceph_cls_assert_locked(req, which++, RBD_LOCK_NAME,
2101 CEPH_CLS_LOCK_EXCLUSIVE, "", "");
2106 ret = rbd_cls_object_map_update(req, which, obj_req->ex.oe_objno,
2107 new_state, current_state);
2111 ret = ceph_osdc_alloc_messages(req, GFP_NOIO);
2115 ceph_osdc_start_request(osdc, req, false);
2119 static void prune_extents(struct ceph_file_extent *img_extents,
2120 u32 *num_img_extents, u64 overlap)
2122 u32 cnt = *num_img_extents;
2124 /* drop extents completely beyond the overlap */
2125 while (cnt && img_extents[cnt - 1].fe_off >= overlap)
2129 struct ceph_file_extent *ex = &img_extents[cnt - 1];
2131 /* trim final overlapping extent */
2132 if (ex->fe_off + ex->fe_len > overlap)
2133 ex->fe_len = overlap - ex->fe_off;
2136 *num_img_extents = cnt;
2140 * Determine the byte range(s) covered by either just the object extent
2141 * or the entire object in the parent image.
2143 static int rbd_obj_calc_img_extents(struct rbd_obj_request *obj_req,
2146 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2149 if (!rbd_dev->parent_overlap)
2152 ret = ceph_extent_to_file(&rbd_dev->layout, obj_req->ex.oe_objno,
2153 entire ? 0 : obj_req->ex.oe_off,
2154 entire ? rbd_dev->layout.object_size :
2156 &obj_req->img_extents,
2157 &obj_req->num_img_extents);
2161 prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
2162 rbd_dev->parent_overlap);
2166 static void rbd_osd_setup_data(struct ceph_osd_request *osd_req, int which)
2168 struct rbd_obj_request *obj_req = osd_req->r_priv;
2170 switch (obj_req->img_request->data_type) {
2171 case OBJ_REQUEST_BIO:
2172 osd_req_op_extent_osd_data_bio(osd_req, which,
2174 obj_req->ex.oe_len);
2176 case OBJ_REQUEST_BVECS:
2177 case OBJ_REQUEST_OWN_BVECS:
2178 rbd_assert(obj_req->bvec_pos.iter.bi_size ==
2179 obj_req->ex.oe_len);
2180 rbd_assert(obj_req->bvec_idx == obj_req->bvec_count);
2181 osd_req_op_extent_osd_data_bvec_pos(osd_req, which,
2182 &obj_req->bvec_pos);
2189 static int rbd_osd_setup_stat(struct ceph_osd_request *osd_req, int which)
2191 struct page **pages;
2194 * The response data for a STAT call consists of:
2201 pages = ceph_alloc_page_vector(1, GFP_NOIO);
2203 return PTR_ERR(pages);
2205 osd_req_op_init(osd_req, which, CEPH_OSD_OP_STAT, 0);
2206 osd_req_op_raw_data_in_pages(osd_req, which, pages,
2207 8 + sizeof(struct ceph_timespec),
2212 static int rbd_osd_setup_copyup(struct ceph_osd_request *osd_req, int which,
2215 struct rbd_obj_request *obj_req = osd_req->r_priv;
2218 ret = osd_req_op_cls_init(osd_req, which, "rbd", "copyup");
2222 osd_req_op_cls_request_data_bvecs(osd_req, which, obj_req->copyup_bvecs,
2223 obj_req->copyup_bvec_count, bytes);
2227 static int rbd_obj_init_read(struct rbd_obj_request *obj_req)
2229 obj_req->read_state = RBD_OBJ_READ_START;
2233 static void __rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2236 struct rbd_obj_request *obj_req = osd_req->r_priv;
2237 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2240 if (!use_object_map(rbd_dev) ||
2241 !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST)) {
2242 osd_req_op_alloc_hint_init(osd_req, which++,
2243 rbd_dev->layout.object_size,
2244 rbd_dev->layout.object_size,
2245 rbd_dev->opts->alloc_hint_flags);
2248 if (rbd_obj_is_entire(obj_req))
2249 opcode = CEPH_OSD_OP_WRITEFULL;
2251 opcode = CEPH_OSD_OP_WRITE;
2253 osd_req_op_extent_init(osd_req, which, opcode,
2254 obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2255 rbd_osd_setup_data(osd_req, which);
2258 static int rbd_obj_init_write(struct rbd_obj_request *obj_req)
2262 /* reverse map the entire object onto the parent */
2263 ret = rbd_obj_calc_img_extents(obj_req, true);
2267 if (rbd_obj_copyup_enabled(obj_req))
2268 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
2270 obj_req->write_state = RBD_OBJ_WRITE_START;
2274 static u16 truncate_or_zero_opcode(struct rbd_obj_request *obj_req)
2276 return rbd_obj_is_tail(obj_req) ? CEPH_OSD_OP_TRUNCATE :
2280 static void __rbd_osd_setup_discard_ops(struct ceph_osd_request *osd_req,
2283 struct rbd_obj_request *obj_req = osd_req->r_priv;
2285 if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
2286 rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2287 osd_req_op_init(osd_req, which, CEPH_OSD_OP_DELETE, 0);
2289 osd_req_op_extent_init(osd_req, which,
2290 truncate_or_zero_opcode(obj_req),
2291 obj_req->ex.oe_off, obj_req->ex.oe_len,
2296 static int rbd_obj_init_discard(struct rbd_obj_request *obj_req)
2298 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2303 * Align the range to alloc_size boundary and punt on discards
2304 * that are too small to free up any space.
2306 * alloc_size == object_size && is_tail() is a special case for
2307 * filestore with filestore_punch_hole = false, needed to allow
2308 * truncate (in addition to delete).
2310 if (rbd_dev->opts->alloc_size != rbd_dev->layout.object_size ||
2311 !rbd_obj_is_tail(obj_req)) {
2312 off = round_up(obj_req->ex.oe_off, rbd_dev->opts->alloc_size);
2313 next_off = round_down(obj_req->ex.oe_off + obj_req->ex.oe_len,
2314 rbd_dev->opts->alloc_size);
2315 if (off >= next_off)
2318 dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
2319 obj_req, obj_req->ex.oe_off, obj_req->ex.oe_len,
2320 off, next_off - off);
2321 obj_req->ex.oe_off = off;
2322 obj_req->ex.oe_len = next_off - off;
2325 /* reverse map the entire object onto the parent */
2326 ret = rbd_obj_calc_img_extents(obj_req, true);
2330 obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2331 if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents)
2332 obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2334 obj_req->write_state = RBD_OBJ_WRITE_START;
2338 static void __rbd_osd_setup_zeroout_ops(struct ceph_osd_request *osd_req,
2341 struct rbd_obj_request *obj_req = osd_req->r_priv;
2344 if (rbd_obj_is_entire(obj_req)) {
2345 if (obj_req->num_img_extents) {
2346 if (!(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2347 osd_req_op_init(osd_req, which++,
2348 CEPH_OSD_OP_CREATE, 0);
2349 opcode = CEPH_OSD_OP_TRUNCATE;
2351 rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2352 osd_req_op_init(osd_req, which++,
2353 CEPH_OSD_OP_DELETE, 0);
2357 opcode = truncate_or_zero_opcode(obj_req);
2361 osd_req_op_extent_init(osd_req, which, opcode,
2362 obj_req->ex.oe_off, obj_req->ex.oe_len,
2366 static int rbd_obj_init_zeroout(struct rbd_obj_request *obj_req)
2370 /* reverse map the entire object onto the parent */
2371 ret = rbd_obj_calc_img_extents(obj_req, true);
2375 if (rbd_obj_copyup_enabled(obj_req))
2376 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
2377 if (!obj_req->num_img_extents) {
2378 obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2379 if (rbd_obj_is_entire(obj_req))
2380 obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2383 obj_req->write_state = RBD_OBJ_WRITE_START;
2387 static int count_write_ops(struct rbd_obj_request *obj_req)
2389 struct rbd_img_request *img_req = obj_req->img_request;
2391 switch (img_req->op_type) {
2393 if (!use_object_map(img_req->rbd_dev) ||
2394 !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST))
2395 return 2; /* setallochint + write/writefull */
2397 return 1; /* write/writefull */
2398 case OBJ_OP_DISCARD:
2399 return 1; /* delete/truncate/zero */
2400 case OBJ_OP_ZEROOUT:
2401 if (rbd_obj_is_entire(obj_req) && obj_req->num_img_extents &&
2402 !(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2403 return 2; /* create + truncate */
2405 return 1; /* delete/truncate/zero */
2411 static void rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2414 struct rbd_obj_request *obj_req = osd_req->r_priv;
2416 switch (obj_req->img_request->op_type) {
2418 __rbd_osd_setup_write_ops(osd_req, which);
2420 case OBJ_OP_DISCARD:
2421 __rbd_osd_setup_discard_ops(osd_req, which);
2423 case OBJ_OP_ZEROOUT:
2424 __rbd_osd_setup_zeroout_ops(osd_req, which);
2432 * Prune the list of object requests (adjust offset and/or length, drop
2433 * redundant requests). Prepare object request state machines and image
2434 * request state machine for execution.
2436 static int __rbd_img_fill_request(struct rbd_img_request *img_req)
2438 struct rbd_obj_request *obj_req, *next_obj_req;
2441 for_each_obj_request_safe(img_req, obj_req, next_obj_req) {
2442 switch (img_req->op_type) {
2444 ret = rbd_obj_init_read(obj_req);
2447 ret = rbd_obj_init_write(obj_req);
2449 case OBJ_OP_DISCARD:
2450 ret = rbd_obj_init_discard(obj_req);
2452 case OBJ_OP_ZEROOUT:
2453 ret = rbd_obj_init_zeroout(obj_req);
2461 rbd_img_obj_request_del(img_req, obj_req);
2466 img_req->state = RBD_IMG_START;
2470 union rbd_img_fill_iter {
2471 struct ceph_bio_iter bio_iter;
2472 struct ceph_bvec_iter bvec_iter;
2475 struct rbd_img_fill_ctx {
2476 enum obj_request_type pos_type;
2477 union rbd_img_fill_iter *pos;
2478 union rbd_img_fill_iter iter;
2479 ceph_object_extent_fn_t set_pos_fn;
2480 ceph_object_extent_fn_t count_fn;
2481 ceph_object_extent_fn_t copy_fn;
2484 static struct ceph_object_extent *alloc_object_extent(void *arg)
2486 struct rbd_img_request *img_req = arg;
2487 struct rbd_obj_request *obj_req;
2489 obj_req = rbd_obj_request_create();
2493 rbd_img_obj_request_add(img_req, obj_req);
2494 return &obj_req->ex;
2498 * While su != os && sc == 1 is technically not fancy (it's the same
2499 * layout as su == os && sc == 1), we can't use the nocopy path for it
2500 * because ->set_pos_fn() should be called only once per object.
2501 * ceph_file_to_extents() invokes action_fn once per stripe unit, so
2502 * treat su != os && sc == 1 as fancy.
2504 static bool rbd_layout_is_fancy(struct ceph_file_layout *l)
2506 return l->stripe_unit != l->object_size;
2509 static int rbd_img_fill_request_nocopy(struct rbd_img_request *img_req,
2510 struct ceph_file_extent *img_extents,
2511 u32 num_img_extents,
2512 struct rbd_img_fill_ctx *fctx)
2517 img_req->data_type = fctx->pos_type;
2520 * Create object requests and set each object request's starting
2521 * position in the provided bio (list) or bio_vec array.
2523 fctx->iter = *fctx->pos;
2524 for (i = 0; i < num_img_extents; i++) {
2525 ret = ceph_file_to_extents(&img_req->rbd_dev->layout,
2526 img_extents[i].fe_off,
2527 img_extents[i].fe_len,
2528 &img_req->object_extents,
2529 alloc_object_extent, img_req,
2530 fctx->set_pos_fn, &fctx->iter);
2535 return __rbd_img_fill_request(img_req);
2539 * Map a list of image extents to a list of object extents, create the
2540 * corresponding object requests (normally each to a different object,
2541 * but not always) and add them to @img_req. For each object request,
2542 * set up its data descriptor to point to the corresponding chunk(s) of
2543 * @fctx->pos data buffer.
2545 * Because ceph_file_to_extents() will merge adjacent object extents
2546 * together, each object request's data descriptor may point to multiple
2547 * different chunks of @fctx->pos data buffer.
2549 * @fctx->pos data buffer is assumed to be large enough.
2551 static int rbd_img_fill_request(struct rbd_img_request *img_req,
2552 struct ceph_file_extent *img_extents,
2553 u32 num_img_extents,
2554 struct rbd_img_fill_ctx *fctx)
2556 struct rbd_device *rbd_dev = img_req->rbd_dev;
2557 struct rbd_obj_request *obj_req;
2561 if (fctx->pos_type == OBJ_REQUEST_NODATA ||
2562 !rbd_layout_is_fancy(&rbd_dev->layout))
2563 return rbd_img_fill_request_nocopy(img_req, img_extents,
2564 num_img_extents, fctx);
2566 img_req->data_type = OBJ_REQUEST_OWN_BVECS;
2569 * Create object requests and determine ->bvec_count for each object
2570 * request. Note that ->bvec_count sum over all object requests may
2571 * be greater than the number of bio_vecs in the provided bio (list)
2572 * or bio_vec array because when mapped, those bio_vecs can straddle
2573 * stripe unit boundaries.
2575 fctx->iter = *fctx->pos;
2576 for (i = 0; i < num_img_extents; i++) {
2577 ret = ceph_file_to_extents(&rbd_dev->layout,
2578 img_extents[i].fe_off,
2579 img_extents[i].fe_len,
2580 &img_req->object_extents,
2581 alloc_object_extent, img_req,
2582 fctx->count_fn, &fctx->iter);
2587 for_each_obj_request(img_req, obj_req) {
2588 obj_req->bvec_pos.bvecs = kmalloc_array(obj_req->bvec_count,
2589 sizeof(*obj_req->bvec_pos.bvecs),
2591 if (!obj_req->bvec_pos.bvecs)
2596 * Fill in each object request's private bio_vec array, splitting and
2597 * rearranging the provided bio_vecs in stripe unit chunks as needed.
2599 fctx->iter = *fctx->pos;
2600 for (i = 0; i < num_img_extents; i++) {
2601 ret = ceph_iterate_extents(&rbd_dev->layout,
2602 img_extents[i].fe_off,
2603 img_extents[i].fe_len,
2604 &img_req->object_extents,
2605 fctx->copy_fn, &fctx->iter);
2610 return __rbd_img_fill_request(img_req);
2613 static int rbd_img_fill_nodata(struct rbd_img_request *img_req,
2616 struct ceph_file_extent ex = { off, len };
2617 union rbd_img_fill_iter dummy = {};
2618 struct rbd_img_fill_ctx fctx = {
2619 .pos_type = OBJ_REQUEST_NODATA,
2623 return rbd_img_fill_request(img_req, &ex, 1, &fctx);
2626 static void set_bio_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2628 struct rbd_obj_request *obj_req =
2629 container_of(ex, struct rbd_obj_request, ex);
2630 struct ceph_bio_iter *it = arg;
2632 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2633 obj_req->bio_pos = *it;
2634 ceph_bio_iter_advance(it, bytes);
2637 static void count_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2639 struct rbd_obj_request *obj_req =
2640 container_of(ex, struct rbd_obj_request, ex);
2641 struct ceph_bio_iter *it = arg;
2643 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2644 ceph_bio_iter_advance_step(it, bytes, ({
2645 obj_req->bvec_count++;
2650 static void copy_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2652 struct rbd_obj_request *obj_req =
2653 container_of(ex, struct rbd_obj_request, ex);
2654 struct ceph_bio_iter *it = arg;
2656 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2657 ceph_bio_iter_advance_step(it, bytes, ({
2658 obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2659 obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2663 static int __rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2664 struct ceph_file_extent *img_extents,
2665 u32 num_img_extents,
2666 struct ceph_bio_iter *bio_pos)
2668 struct rbd_img_fill_ctx fctx = {
2669 .pos_type = OBJ_REQUEST_BIO,
2670 .pos = (union rbd_img_fill_iter *)bio_pos,
2671 .set_pos_fn = set_bio_pos,
2672 .count_fn = count_bio_bvecs,
2673 .copy_fn = copy_bio_bvecs,
2676 return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2680 static int rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2681 u64 off, u64 len, struct bio *bio)
2683 struct ceph_file_extent ex = { off, len };
2684 struct ceph_bio_iter it = { .bio = bio, .iter = bio->bi_iter };
2686 return __rbd_img_fill_from_bio(img_req, &ex, 1, &it);
2689 static void set_bvec_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2691 struct rbd_obj_request *obj_req =
2692 container_of(ex, struct rbd_obj_request, ex);
2693 struct ceph_bvec_iter *it = arg;
2695 obj_req->bvec_pos = *it;
2696 ceph_bvec_iter_shorten(&obj_req->bvec_pos, bytes);
2697 ceph_bvec_iter_advance(it, bytes);
2700 static void count_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2702 struct rbd_obj_request *obj_req =
2703 container_of(ex, struct rbd_obj_request, ex);
2704 struct ceph_bvec_iter *it = arg;
2706 ceph_bvec_iter_advance_step(it, bytes, ({
2707 obj_req->bvec_count++;
2711 static void copy_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2713 struct rbd_obj_request *obj_req =
2714 container_of(ex, struct rbd_obj_request, ex);
2715 struct ceph_bvec_iter *it = arg;
2717 ceph_bvec_iter_advance_step(it, bytes, ({
2718 obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2719 obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2723 static int __rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2724 struct ceph_file_extent *img_extents,
2725 u32 num_img_extents,
2726 struct ceph_bvec_iter *bvec_pos)
2728 struct rbd_img_fill_ctx fctx = {
2729 .pos_type = OBJ_REQUEST_BVECS,
2730 .pos = (union rbd_img_fill_iter *)bvec_pos,
2731 .set_pos_fn = set_bvec_pos,
2732 .count_fn = count_bvecs,
2733 .copy_fn = copy_bvecs,
2736 return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2740 static int rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2741 struct ceph_file_extent *img_extents,
2742 u32 num_img_extents,
2743 struct bio_vec *bvecs)
2745 struct ceph_bvec_iter it = {
2747 .iter = { .bi_size = ceph_file_extents_bytes(img_extents,
2751 return __rbd_img_fill_from_bvecs(img_req, img_extents, num_img_extents,
2755 static void rbd_img_handle_request_work(struct work_struct *work)
2757 struct rbd_img_request *img_req =
2758 container_of(work, struct rbd_img_request, work);
2760 rbd_img_handle_request(img_req, img_req->work_result);
2763 static void rbd_img_schedule(struct rbd_img_request *img_req, int result)
2765 INIT_WORK(&img_req->work, rbd_img_handle_request_work);
2766 img_req->work_result = result;
2767 queue_work(rbd_wq, &img_req->work);
2770 static bool rbd_obj_may_exist(struct rbd_obj_request *obj_req)
2772 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2774 if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno)) {
2775 obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2779 dout("%s %p objno %llu assuming dne\n", __func__, obj_req,
2780 obj_req->ex.oe_objno);
2784 static int rbd_obj_read_object(struct rbd_obj_request *obj_req)
2786 struct ceph_osd_request *osd_req;
2789 osd_req = __rbd_obj_add_osd_request(obj_req, NULL, 1);
2790 if (IS_ERR(osd_req))
2791 return PTR_ERR(osd_req);
2793 osd_req_op_extent_init(osd_req, 0, CEPH_OSD_OP_READ,
2794 obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2795 rbd_osd_setup_data(osd_req, 0);
2796 rbd_osd_format_read(osd_req);
2798 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2802 rbd_osd_submit(osd_req);
2806 static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
2808 struct rbd_img_request *img_req = obj_req->img_request;
2809 struct rbd_device *parent = img_req->rbd_dev->parent;
2810 struct rbd_img_request *child_img_req;
2813 child_img_req = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
2817 rbd_img_request_init(child_img_req, parent, OBJ_OP_READ);
2818 __set_bit(IMG_REQ_CHILD, &child_img_req->flags);
2819 child_img_req->obj_request = obj_req;
2821 down_read(&parent->header_rwsem);
2822 rbd_img_capture_header(child_img_req);
2823 up_read(&parent->header_rwsem);
2825 dout("%s child_img_req %p for obj_req %p\n", __func__, child_img_req,
2828 if (!rbd_img_is_write(img_req)) {
2829 switch (img_req->data_type) {
2830 case OBJ_REQUEST_BIO:
2831 ret = __rbd_img_fill_from_bio(child_img_req,
2832 obj_req->img_extents,
2833 obj_req->num_img_extents,
2836 case OBJ_REQUEST_BVECS:
2837 case OBJ_REQUEST_OWN_BVECS:
2838 ret = __rbd_img_fill_from_bvecs(child_img_req,
2839 obj_req->img_extents,
2840 obj_req->num_img_extents,
2841 &obj_req->bvec_pos);
2847 ret = rbd_img_fill_from_bvecs(child_img_req,
2848 obj_req->img_extents,
2849 obj_req->num_img_extents,
2850 obj_req->copyup_bvecs);
2853 rbd_img_request_destroy(child_img_req);
2857 /* avoid parent chain recursion */
2858 rbd_img_schedule(child_img_req, 0);
2862 static bool rbd_obj_advance_read(struct rbd_obj_request *obj_req, int *result)
2864 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2868 switch (obj_req->read_state) {
2869 case RBD_OBJ_READ_START:
2870 rbd_assert(!*result);
2872 if (!rbd_obj_may_exist(obj_req)) {
2874 obj_req->read_state = RBD_OBJ_READ_OBJECT;
2878 ret = rbd_obj_read_object(obj_req);
2883 obj_req->read_state = RBD_OBJ_READ_OBJECT;
2885 case RBD_OBJ_READ_OBJECT:
2886 if (*result == -ENOENT && rbd_dev->parent_overlap) {
2887 /* reverse map this object extent onto the parent */
2888 ret = rbd_obj_calc_img_extents(obj_req, false);
2893 if (obj_req->num_img_extents) {
2894 ret = rbd_obj_read_from_parent(obj_req);
2899 obj_req->read_state = RBD_OBJ_READ_PARENT;
2905 * -ENOENT means a hole in the image -- zero-fill the entire
2906 * length of the request. A short read also implies zero-fill
2907 * to the end of the request.
2909 if (*result == -ENOENT) {
2910 rbd_obj_zero_range(obj_req, 0, obj_req->ex.oe_len);
2912 } else if (*result >= 0) {
2913 if (*result < obj_req->ex.oe_len)
2914 rbd_obj_zero_range(obj_req, *result,
2915 obj_req->ex.oe_len - *result);
2917 rbd_assert(*result == obj_req->ex.oe_len);
2921 case RBD_OBJ_READ_PARENT:
2923 * The parent image is read only up to the overlap -- zero-fill
2924 * from the overlap to the end of the request.
2927 u32 obj_overlap = rbd_obj_img_extents_bytes(obj_req);
2929 if (obj_overlap < obj_req->ex.oe_len)
2930 rbd_obj_zero_range(obj_req, obj_overlap,
2931 obj_req->ex.oe_len - obj_overlap);
2939 static bool rbd_obj_write_is_noop(struct rbd_obj_request *obj_req)
2941 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2943 if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno))
2944 obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2946 if (!(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST) &&
2947 (obj_req->flags & RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT)) {
2948 dout("%s %p noop for nonexistent\n", __func__, obj_req);
2957 * 0 - object map update sent
2958 * 1 - object map update isn't needed
2961 static int rbd_obj_write_pre_object_map(struct rbd_obj_request *obj_req)
2963 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2966 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
2969 if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
2970 new_state = OBJECT_PENDING;
2972 new_state = OBJECT_EXISTS;
2974 return rbd_object_map_update(obj_req, CEPH_NOSNAP, new_state, NULL);
2977 static int rbd_obj_write_object(struct rbd_obj_request *obj_req)
2979 struct ceph_osd_request *osd_req;
2980 int num_ops = count_write_ops(obj_req);
2984 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
2985 num_ops++; /* stat */
2987 osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
2988 if (IS_ERR(osd_req))
2989 return PTR_ERR(osd_req);
2991 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
2992 ret = rbd_osd_setup_stat(osd_req, which++);
2997 rbd_osd_setup_write_ops(osd_req, which);
2998 rbd_osd_format_write(osd_req);
3000 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3004 rbd_osd_submit(osd_req);
3009 * copyup_bvecs pages are never highmem pages
3011 static bool is_zero_bvecs(struct bio_vec *bvecs, u32 bytes)
3013 struct ceph_bvec_iter it = {
3015 .iter = { .bi_size = bytes },
3018 ceph_bvec_iter_advance_step(&it, bytes, ({
3019 if (memchr_inv(page_address(bv.bv_page) + bv.bv_offset, 0,
3026 #define MODS_ONLY U32_MAX
3028 static int rbd_obj_copyup_empty_snapc(struct rbd_obj_request *obj_req,
3031 struct ceph_osd_request *osd_req;
3034 dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3035 rbd_assert(bytes > 0 && bytes != MODS_ONLY);
3037 osd_req = __rbd_obj_add_osd_request(obj_req, &rbd_empty_snapc, 1);
3038 if (IS_ERR(osd_req))
3039 return PTR_ERR(osd_req);
3041 ret = rbd_osd_setup_copyup(osd_req, 0, bytes);
3045 rbd_osd_format_write(osd_req);
3047 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3051 rbd_osd_submit(osd_req);
3055 static int rbd_obj_copyup_current_snapc(struct rbd_obj_request *obj_req,
3058 struct ceph_osd_request *osd_req;
3059 int num_ops = count_write_ops(obj_req);
3063 dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3065 if (bytes != MODS_ONLY)
3066 num_ops++; /* copyup */
3068 osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
3069 if (IS_ERR(osd_req))
3070 return PTR_ERR(osd_req);
3072 if (bytes != MODS_ONLY) {
3073 ret = rbd_osd_setup_copyup(osd_req, which++, bytes);
3078 rbd_osd_setup_write_ops(osd_req, which);
3079 rbd_osd_format_write(osd_req);
3081 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3085 rbd_osd_submit(osd_req);
3089 static int setup_copyup_bvecs(struct rbd_obj_request *obj_req, u64 obj_overlap)
3093 rbd_assert(!obj_req->copyup_bvecs);
3094 obj_req->copyup_bvec_count = calc_pages_for(0, obj_overlap);
3095 obj_req->copyup_bvecs = kcalloc(obj_req->copyup_bvec_count,
3096 sizeof(*obj_req->copyup_bvecs),
3098 if (!obj_req->copyup_bvecs)
3101 for (i = 0; i < obj_req->copyup_bvec_count; i++) {
3102 unsigned int len = min(obj_overlap, (u64)PAGE_SIZE);
3104 obj_req->copyup_bvecs[i].bv_page = alloc_page(GFP_NOIO);
3105 if (!obj_req->copyup_bvecs[i].bv_page)
3108 obj_req->copyup_bvecs[i].bv_offset = 0;
3109 obj_req->copyup_bvecs[i].bv_len = len;
3113 rbd_assert(!obj_overlap);
3118 * The target object doesn't exist. Read the data for the entire
3119 * target object up to the overlap point (if any) from the parent,
3120 * so we can use it for a copyup.
3122 static int rbd_obj_copyup_read_parent(struct rbd_obj_request *obj_req)
3124 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3127 rbd_assert(obj_req->num_img_extents);
3128 prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
3129 rbd_dev->parent_overlap);
3130 if (!obj_req->num_img_extents) {
3132 * The overlap has become 0 (most likely because the
3133 * image has been flattened). Re-submit the original write
3134 * request -- pass MODS_ONLY since the copyup isn't needed
3137 return rbd_obj_copyup_current_snapc(obj_req, MODS_ONLY);
3140 ret = setup_copyup_bvecs(obj_req, rbd_obj_img_extents_bytes(obj_req));
3144 return rbd_obj_read_from_parent(obj_req);
3147 static void rbd_obj_copyup_object_maps(struct rbd_obj_request *obj_req)
3149 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3150 struct ceph_snap_context *snapc = obj_req->img_request->snapc;
3155 rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3157 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3160 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3163 for (i = 0; i < snapc->num_snaps; i++) {
3164 if ((rbd_dev->header.features & RBD_FEATURE_FAST_DIFF) &&
3165 i + 1 < snapc->num_snaps)
3166 new_state = OBJECT_EXISTS_CLEAN;
3168 new_state = OBJECT_EXISTS;
3170 ret = rbd_object_map_update(obj_req, snapc->snaps[i],
3173 obj_req->pending.result = ret;
3178 obj_req->pending.num_pending++;
3182 static void rbd_obj_copyup_write_object(struct rbd_obj_request *obj_req)
3184 u32 bytes = rbd_obj_img_extents_bytes(obj_req);
3187 rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3190 * Only send non-zero copyup data to save some I/O and network
3191 * bandwidth -- zero copyup data is equivalent to the object not
3194 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3197 if (obj_req->img_request->snapc->num_snaps && bytes > 0) {
3199 * Send a copyup request with an empty snapshot context to
3200 * deep-copyup the object through all existing snapshots.
3201 * A second request with the current snapshot context will be
3202 * sent for the actual modification.
3204 ret = rbd_obj_copyup_empty_snapc(obj_req, bytes);
3206 obj_req->pending.result = ret;
3210 obj_req->pending.num_pending++;
3214 ret = rbd_obj_copyup_current_snapc(obj_req, bytes);
3216 obj_req->pending.result = ret;
3220 obj_req->pending.num_pending++;
3223 static bool rbd_obj_advance_copyup(struct rbd_obj_request *obj_req, int *result)
3225 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3229 switch (obj_req->copyup_state) {
3230 case RBD_OBJ_COPYUP_START:
3231 rbd_assert(!*result);
3233 ret = rbd_obj_copyup_read_parent(obj_req);
3238 if (obj_req->num_img_extents)
3239 obj_req->copyup_state = RBD_OBJ_COPYUP_READ_PARENT;
3241 obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3243 case RBD_OBJ_COPYUP_READ_PARENT:
3247 if (is_zero_bvecs(obj_req->copyup_bvecs,
3248 rbd_obj_img_extents_bytes(obj_req))) {
3249 dout("%s %p detected zeros\n", __func__, obj_req);
3250 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ZEROS;
3253 rbd_obj_copyup_object_maps(obj_req);
3254 if (!obj_req->pending.num_pending) {
3255 *result = obj_req->pending.result;
3256 obj_req->copyup_state = RBD_OBJ_COPYUP_OBJECT_MAPS;
3259 obj_req->copyup_state = __RBD_OBJ_COPYUP_OBJECT_MAPS;
3261 case __RBD_OBJ_COPYUP_OBJECT_MAPS:
3262 if (!pending_result_dec(&obj_req->pending, result))
3265 case RBD_OBJ_COPYUP_OBJECT_MAPS:
3267 rbd_warn(rbd_dev, "snap object map update failed: %d",
3272 rbd_obj_copyup_write_object(obj_req);
3273 if (!obj_req->pending.num_pending) {
3274 *result = obj_req->pending.result;
3275 obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3278 obj_req->copyup_state = __RBD_OBJ_COPYUP_WRITE_OBJECT;
3280 case __RBD_OBJ_COPYUP_WRITE_OBJECT:
3281 if (!pending_result_dec(&obj_req->pending, result))
3284 case RBD_OBJ_COPYUP_WRITE_OBJECT:
3293 * 0 - object map update sent
3294 * 1 - object map update isn't needed
3297 static int rbd_obj_write_post_object_map(struct rbd_obj_request *obj_req)
3299 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3300 u8 current_state = OBJECT_PENDING;
3302 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3305 if (!(obj_req->flags & RBD_OBJ_FLAG_DELETION))
3308 return rbd_object_map_update(obj_req, CEPH_NOSNAP, OBJECT_NONEXISTENT,
3312 static bool rbd_obj_advance_write(struct rbd_obj_request *obj_req, int *result)
3314 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3318 switch (obj_req->write_state) {
3319 case RBD_OBJ_WRITE_START:
3320 rbd_assert(!*result);
3322 if (rbd_obj_write_is_noop(obj_req))
3325 ret = rbd_obj_write_pre_object_map(obj_req);
3330 obj_req->write_state = RBD_OBJ_WRITE_PRE_OBJECT_MAP;
3334 case RBD_OBJ_WRITE_PRE_OBJECT_MAP:
3336 rbd_warn(rbd_dev, "pre object map update failed: %d",
3340 ret = rbd_obj_write_object(obj_req);
3345 obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
3347 case RBD_OBJ_WRITE_OBJECT:
3348 if (*result == -ENOENT) {
3349 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
3351 obj_req->copyup_state = RBD_OBJ_COPYUP_START;
3352 obj_req->write_state = __RBD_OBJ_WRITE_COPYUP;
3356 * On a non-existent object:
3357 * delete - -ENOENT, truncate/zero - 0
3359 if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
3365 obj_req->write_state = RBD_OBJ_WRITE_COPYUP;
3367 case __RBD_OBJ_WRITE_COPYUP:
3368 if (!rbd_obj_advance_copyup(obj_req, result))
3371 case RBD_OBJ_WRITE_COPYUP:
3373 rbd_warn(rbd_dev, "copyup failed: %d", *result);
3376 ret = rbd_obj_write_post_object_map(obj_req);
3381 obj_req->write_state = RBD_OBJ_WRITE_POST_OBJECT_MAP;
3385 case RBD_OBJ_WRITE_POST_OBJECT_MAP:
3387 rbd_warn(rbd_dev, "post object map update failed: %d",
3396 * Return true if @obj_req is completed.
3398 static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
3401 struct rbd_img_request *img_req = obj_req->img_request;
3402 struct rbd_device *rbd_dev = img_req->rbd_dev;
3405 mutex_lock(&obj_req->state_mutex);
3406 if (!rbd_img_is_write(img_req))
3407 done = rbd_obj_advance_read(obj_req, result);
3409 done = rbd_obj_advance_write(obj_req, result);
3410 mutex_unlock(&obj_req->state_mutex);
3412 if (done && *result) {
3413 rbd_assert(*result < 0);
3414 rbd_warn(rbd_dev, "%s at objno %llu %llu~%llu result %d",
3415 obj_op_name(img_req->op_type), obj_req->ex.oe_objno,
3416 obj_req->ex.oe_off, obj_req->ex.oe_len, *result);
3422 * This is open-coded in rbd_img_handle_request() to avoid parent chain
3425 static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result)
3427 if (__rbd_obj_handle_request(obj_req, &result))
3428 rbd_img_handle_request(obj_req->img_request, result);
3431 static bool need_exclusive_lock(struct rbd_img_request *img_req)
3433 struct rbd_device *rbd_dev = img_req->rbd_dev;
3435 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK))
3438 if (rbd_is_ro(rbd_dev))
3441 rbd_assert(!test_bit(IMG_REQ_CHILD, &img_req->flags));
3442 if (rbd_dev->opts->lock_on_read ||
3443 (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3446 return rbd_img_is_write(img_req);
3449 static bool rbd_lock_add_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 locked = rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED;
3456 spin_lock(&rbd_dev->lock_lists_lock);
3457 rbd_assert(list_empty(&img_req->lock_item));
3459 list_add_tail(&img_req->lock_item, &rbd_dev->acquiring_list);
3461 list_add_tail(&img_req->lock_item, &rbd_dev->running_list);
3462 spin_unlock(&rbd_dev->lock_lists_lock);
3466 static void rbd_lock_del_request(struct rbd_img_request *img_req)
3468 struct rbd_device *rbd_dev = img_req->rbd_dev;
3471 lockdep_assert_held(&rbd_dev->lock_rwsem);
3472 spin_lock(&rbd_dev->lock_lists_lock);
3473 rbd_assert(!list_empty(&img_req->lock_item));
3474 list_del_init(&img_req->lock_item);
3475 need_wakeup = (rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING &&
3476 list_empty(&rbd_dev->running_list));
3477 spin_unlock(&rbd_dev->lock_lists_lock);
3479 complete(&rbd_dev->releasing_wait);
3482 static int rbd_img_exclusive_lock(struct rbd_img_request *img_req)
3484 struct rbd_device *rbd_dev = img_req->rbd_dev;
3486 if (!need_exclusive_lock(img_req))
3489 if (rbd_lock_add_request(img_req))
3492 if (rbd_dev->opts->exclusive) {
3493 WARN_ON(1); /* lock got released? */
3498 * Note the use of mod_delayed_work() in rbd_acquire_lock()
3499 * and cancel_delayed_work() in wake_lock_waiters().
3501 dout("%s rbd_dev %p queueing lock_dwork\n", __func__, rbd_dev);
3502 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
3506 static void rbd_img_object_requests(struct rbd_img_request *img_req)
3508 struct rbd_obj_request *obj_req;
3510 rbd_assert(!img_req->pending.result && !img_req->pending.num_pending);
3512 for_each_obj_request(img_req, obj_req) {
3515 if (__rbd_obj_handle_request(obj_req, &result)) {
3517 img_req->pending.result = result;
3521 img_req->pending.num_pending++;
3526 static bool rbd_img_advance(struct rbd_img_request *img_req, int *result)
3528 struct rbd_device *rbd_dev = img_req->rbd_dev;
3532 switch (img_req->state) {
3534 rbd_assert(!*result);
3536 ret = rbd_img_exclusive_lock(img_req);
3541 img_req->state = RBD_IMG_EXCLUSIVE_LOCK;
3545 case RBD_IMG_EXCLUSIVE_LOCK:
3549 rbd_assert(!need_exclusive_lock(img_req) ||
3550 __rbd_is_lock_owner(rbd_dev));
3552 rbd_img_object_requests(img_req);
3553 if (!img_req->pending.num_pending) {
3554 *result = img_req->pending.result;
3555 img_req->state = RBD_IMG_OBJECT_REQUESTS;
3558 img_req->state = __RBD_IMG_OBJECT_REQUESTS;
3560 case __RBD_IMG_OBJECT_REQUESTS:
3561 if (!pending_result_dec(&img_req->pending, result))
3564 case RBD_IMG_OBJECT_REQUESTS:
3572 * Return true if @img_req is completed.
3574 static bool __rbd_img_handle_request(struct rbd_img_request *img_req,
3577 struct rbd_device *rbd_dev = img_req->rbd_dev;
3580 if (need_exclusive_lock(img_req)) {
3581 down_read(&rbd_dev->lock_rwsem);
3582 mutex_lock(&img_req->state_mutex);
3583 done = rbd_img_advance(img_req, result);
3585 rbd_lock_del_request(img_req);
3586 mutex_unlock(&img_req->state_mutex);
3587 up_read(&rbd_dev->lock_rwsem);
3589 mutex_lock(&img_req->state_mutex);
3590 done = rbd_img_advance(img_req, result);
3591 mutex_unlock(&img_req->state_mutex);
3594 if (done && *result) {
3595 rbd_assert(*result < 0);
3596 rbd_warn(rbd_dev, "%s%s result %d",
3597 test_bit(IMG_REQ_CHILD, &img_req->flags) ? "child " : "",
3598 obj_op_name(img_req->op_type), *result);
3603 static void rbd_img_handle_request(struct rbd_img_request *img_req, int result)
3606 if (!__rbd_img_handle_request(img_req, &result))
3609 if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
3610 struct rbd_obj_request *obj_req = img_req->obj_request;
3612 rbd_img_request_destroy(img_req);
3613 if (__rbd_obj_handle_request(obj_req, &result)) {
3614 img_req = obj_req->img_request;
3618 struct request *rq = blk_mq_rq_from_pdu(img_req);
3620 rbd_img_request_destroy(img_req);
3621 blk_mq_end_request(rq, errno_to_blk_status(result));
3625 static const struct rbd_client_id rbd_empty_cid;
3627 static bool rbd_cid_equal(const struct rbd_client_id *lhs,
3628 const struct rbd_client_id *rhs)
3630 return lhs->gid == rhs->gid && lhs->handle == rhs->handle;
3633 static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev)
3635 struct rbd_client_id cid;
3637 mutex_lock(&rbd_dev->watch_mutex);
3638 cid.gid = ceph_client_gid(rbd_dev->rbd_client->client);
3639 cid.handle = rbd_dev->watch_cookie;
3640 mutex_unlock(&rbd_dev->watch_mutex);
3645 * lock_rwsem must be held for write
3647 static void rbd_set_owner_cid(struct rbd_device *rbd_dev,
3648 const struct rbd_client_id *cid)
3650 dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev,
3651 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle,
3652 cid->gid, cid->handle);
3653 rbd_dev->owner_cid = *cid; /* struct */
3656 static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf)
3658 mutex_lock(&rbd_dev->watch_mutex);
3659 sprintf(buf, "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie);
3660 mutex_unlock(&rbd_dev->watch_mutex);
3663 static void __rbd_lock(struct rbd_device *rbd_dev, const char *cookie)
3665 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3667 rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED;
3668 strcpy(rbd_dev->lock_cookie, cookie);
3669 rbd_set_owner_cid(rbd_dev, &cid);
3670 queue_work(rbd_dev->task_wq, &rbd_dev->acquired_lock_work);
3674 * lock_rwsem must be held for write
3676 static int rbd_lock(struct rbd_device *rbd_dev)
3678 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3682 WARN_ON(__rbd_is_lock_owner(rbd_dev) ||
3683 rbd_dev->lock_cookie[0] != '\0');
3685 format_lock_cookie(rbd_dev, cookie);
3686 ret = ceph_cls_lock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3687 RBD_LOCK_NAME, CEPH_CLS_LOCK_EXCLUSIVE, cookie,
3688 RBD_LOCK_TAG, "", 0);
3692 __rbd_lock(rbd_dev, cookie);
3697 * lock_rwsem must be held for write
3699 static void rbd_unlock(struct rbd_device *rbd_dev)
3701 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3704 WARN_ON(!__rbd_is_lock_owner(rbd_dev) ||
3705 rbd_dev->lock_cookie[0] == '\0');
3707 ret = ceph_cls_unlock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3708 RBD_LOCK_NAME, rbd_dev->lock_cookie);
3709 if (ret && ret != -ENOENT)
3710 rbd_warn(rbd_dev, "failed to unlock header: %d", ret);
3712 /* treat errors as the image is unlocked */
3713 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
3714 rbd_dev->lock_cookie[0] = '\0';
3715 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3716 queue_work(rbd_dev->task_wq, &rbd_dev->released_lock_work);
3719 static int __rbd_notify_op_lock(struct rbd_device *rbd_dev,
3720 enum rbd_notify_op notify_op,
3721 struct page ***preply_pages,
3724 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3725 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3726 char buf[4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN];
3727 int buf_size = sizeof(buf);
3730 dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op);
3732 /* encode *LockPayload NotifyMessage (op + ClientId) */
3733 ceph_start_encoding(&p, 2, 1, buf_size - CEPH_ENCODING_START_BLK_LEN);
3734 ceph_encode_32(&p, notify_op);
3735 ceph_encode_64(&p, cid.gid);
3736 ceph_encode_64(&p, cid.handle);
3738 return ceph_osdc_notify(osdc, &rbd_dev->header_oid,
3739 &rbd_dev->header_oloc, buf, buf_size,
3740 RBD_NOTIFY_TIMEOUT, preply_pages, preply_len);
3743 static void rbd_notify_op_lock(struct rbd_device *rbd_dev,
3744 enum rbd_notify_op notify_op)
3746 __rbd_notify_op_lock(rbd_dev, notify_op, NULL, NULL);
3749 static void rbd_notify_acquired_lock(struct work_struct *work)
3751 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3752 acquired_lock_work);
3754 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_ACQUIRED_LOCK);
3757 static void rbd_notify_released_lock(struct work_struct *work)
3759 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3760 released_lock_work);
3762 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_RELEASED_LOCK);
3765 static int rbd_request_lock(struct rbd_device *rbd_dev)
3767 struct page **reply_pages;
3769 bool lock_owner_responded = false;
3772 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3774 ret = __rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_REQUEST_LOCK,
3775 &reply_pages, &reply_len);
3776 if (ret && ret != -ETIMEDOUT) {
3777 rbd_warn(rbd_dev, "failed to request lock: %d", ret);
3781 if (reply_len > 0 && reply_len <= PAGE_SIZE) {
3782 void *p = page_address(reply_pages[0]);
3783 void *const end = p + reply_len;
3786 ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */
3791 ceph_decode_need(&p, end, 8 + 8, e_inval);
3792 p += 8 + 8; /* skip gid and cookie */
3794 ceph_decode_32_safe(&p, end, len, e_inval);
3798 if (lock_owner_responded) {
3800 "duplicate lock owners detected");
3805 lock_owner_responded = true;
3806 ret = ceph_start_decoding(&p, end, 1, "ResponseMessage",
3810 "failed to decode ResponseMessage: %d",
3815 ret = ceph_decode_32(&p);
3819 if (!lock_owner_responded) {
3820 rbd_warn(rbd_dev, "no lock owners detected");
3825 ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3834 * Either image request state machine(s) or rbd_add_acquire_lock()
3837 static void wake_lock_waiters(struct rbd_device *rbd_dev, int result)
3839 struct rbd_img_request *img_req;
3841 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
3842 lockdep_assert_held_write(&rbd_dev->lock_rwsem);
3844 cancel_delayed_work(&rbd_dev->lock_dwork);
3845 if (!completion_done(&rbd_dev->acquire_wait)) {
3846 rbd_assert(list_empty(&rbd_dev->acquiring_list) &&
3847 list_empty(&rbd_dev->running_list));
3848 rbd_dev->acquire_err = result;
3849 complete_all(&rbd_dev->acquire_wait);
3853 list_for_each_entry(img_req, &rbd_dev->acquiring_list, lock_item) {
3854 mutex_lock(&img_req->state_mutex);
3855 rbd_assert(img_req->state == RBD_IMG_EXCLUSIVE_LOCK);
3856 rbd_img_schedule(img_req, result);
3857 mutex_unlock(&img_req->state_mutex);
3860 list_splice_tail_init(&rbd_dev->acquiring_list, &rbd_dev->running_list);
3863 static int get_lock_owner_info(struct rbd_device *rbd_dev,
3864 struct ceph_locker **lockers, u32 *num_lockers)
3866 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3871 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3873 ret = ceph_cls_lock_info(osdc, &rbd_dev->header_oid,
3874 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3875 &lock_type, &lock_tag, lockers, num_lockers);
3879 if (*num_lockers == 0) {
3880 dout("%s rbd_dev %p no lockers detected\n", __func__, rbd_dev);
3884 if (strcmp(lock_tag, RBD_LOCK_TAG)) {
3885 rbd_warn(rbd_dev, "locked by external mechanism, tag %s",
3891 if (lock_type == CEPH_CLS_LOCK_SHARED) {
3892 rbd_warn(rbd_dev, "shared lock type detected");
3897 if (strncmp((*lockers)[0].id.cookie, RBD_LOCK_COOKIE_PREFIX,
3898 strlen(RBD_LOCK_COOKIE_PREFIX))) {
3899 rbd_warn(rbd_dev, "locked by external mechanism, cookie %s",
3900 (*lockers)[0].id.cookie);
3910 static int find_watcher(struct rbd_device *rbd_dev,
3911 const struct ceph_locker *locker)
3913 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3914 struct ceph_watch_item *watchers;
3920 ret = ceph_osdc_list_watchers(osdc, &rbd_dev->header_oid,
3921 &rbd_dev->header_oloc, &watchers,
3926 sscanf(locker->id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu", &cookie);
3927 for (i = 0; i < num_watchers; i++) {
3928 if (!memcmp(&watchers[i].addr, &locker->info.addr,
3929 sizeof(locker->info.addr)) &&
3930 watchers[i].cookie == cookie) {
3931 struct rbd_client_id cid = {
3932 .gid = le64_to_cpu(watchers[i].name.num),
3936 dout("%s rbd_dev %p found cid %llu-%llu\n", __func__,
3937 rbd_dev, cid.gid, cid.handle);
3938 rbd_set_owner_cid(rbd_dev, &cid);
3944 dout("%s rbd_dev %p no watchers\n", __func__, rbd_dev);
3952 * lock_rwsem must be held for write
3954 static int rbd_try_lock(struct rbd_device *rbd_dev)
3956 struct ceph_client *client = rbd_dev->rbd_client->client;
3957 struct ceph_locker *lockers;
3962 ret = rbd_lock(rbd_dev);
3966 /* determine if the current lock holder is still alive */
3967 ret = get_lock_owner_info(rbd_dev, &lockers, &num_lockers);
3971 if (num_lockers == 0)
3974 ret = find_watcher(rbd_dev, lockers);
3976 goto out; /* request lock or error */
3978 rbd_warn(rbd_dev, "breaking header lock owned by %s%llu",
3979 ENTITY_NAME(lockers[0].id.name));
3981 ret = ceph_monc_blocklist_add(&client->monc,
3982 &lockers[0].info.addr);
3984 rbd_warn(rbd_dev, "blocklist of %s%llu failed: %d",
3985 ENTITY_NAME(lockers[0].id.name), ret);
3989 ret = ceph_cls_break_lock(&client->osdc, &rbd_dev->header_oid,
3990 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3991 lockers[0].id.cookie,
3992 &lockers[0].id.name);
3993 if (ret && ret != -ENOENT)
3997 ceph_free_lockers(lockers, num_lockers);
4001 ceph_free_lockers(lockers, num_lockers);
4005 static int rbd_post_acquire_action(struct rbd_device *rbd_dev)
4009 if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) {
4010 ret = rbd_object_map_open(rbd_dev);
4021 * 1 - caller should call rbd_request_lock()
4024 static int rbd_try_acquire_lock(struct rbd_device *rbd_dev)
4028 down_read(&rbd_dev->lock_rwsem);
4029 dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
4030 rbd_dev->lock_state);
4031 if (__rbd_is_lock_owner(rbd_dev)) {
4032 up_read(&rbd_dev->lock_rwsem);
4036 up_read(&rbd_dev->lock_rwsem);
4037 down_write(&rbd_dev->lock_rwsem);
4038 dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
4039 rbd_dev->lock_state);
4040 if (__rbd_is_lock_owner(rbd_dev)) {
4041 up_write(&rbd_dev->lock_rwsem);
4045 ret = rbd_try_lock(rbd_dev);
4047 rbd_warn(rbd_dev, "failed to lock header: %d", ret);
4048 if (ret == -EBLOCKLISTED)
4051 ret = 1; /* request lock anyway */
4054 up_write(&rbd_dev->lock_rwsem);
4058 rbd_assert(rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED);
4059 rbd_assert(list_empty(&rbd_dev->running_list));
4061 ret = rbd_post_acquire_action(rbd_dev);
4063 rbd_warn(rbd_dev, "post-acquire action failed: %d", ret);
4065 * Can't stay in RBD_LOCK_STATE_LOCKED because
4066 * rbd_lock_add_request() would let the request through,
4067 * assuming that e.g. object map is locked and loaded.
4069 rbd_unlock(rbd_dev);
4073 wake_lock_waiters(rbd_dev, ret);
4074 up_write(&rbd_dev->lock_rwsem);
4078 static void rbd_acquire_lock(struct work_struct *work)
4080 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4081 struct rbd_device, lock_dwork);
4084 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4086 ret = rbd_try_acquire_lock(rbd_dev);
4088 dout("%s rbd_dev %p ret %d - done\n", __func__, rbd_dev, ret);
4092 ret = rbd_request_lock(rbd_dev);
4093 if (ret == -ETIMEDOUT) {
4094 goto again; /* treat this as a dead client */
4095 } else if (ret == -EROFS) {
4096 rbd_warn(rbd_dev, "peer will not release lock");
4097 down_write(&rbd_dev->lock_rwsem);
4098 wake_lock_waiters(rbd_dev, ret);
4099 up_write(&rbd_dev->lock_rwsem);
4100 } else if (ret < 0) {
4101 rbd_warn(rbd_dev, "error requesting lock: %d", ret);
4102 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4106 * lock owner acked, but resend if we don't see them
4109 dout("%s rbd_dev %p requeuing lock_dwork\n", __func__,
4111 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4112 msecs_to_jiffies(2 * RBD_NOTIFY_TIMEOUT * MSEC_PER_SEC));
4116 static bool rbd_quiesce_lock(struct rbd_device *rbd_dev)
4120 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4121 lockdep_assert_held_write(&rbd_dev->lock_rwsem);
4123 if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED)
4127 * Ensure that all in-flight IO is flushed.
4129 rbd_dev->lock_state = RBD_LOCK_STATE_RELEASING;
4130 rbd_assert(!completion_done(&rbd_dev->releasing_wait));
4131 need_wait = !list_empty(&rbd_dev->running_list);
4132 downgrade_write(&rbd_dev->lock_rwsem);
4134 wait_for_completion(&rbd_dev->releasing_wait);
4135 up_read(&rbd_dev->lock_rwsem);
4137 down_write(&rbd_dev->lock_rwsem);
4138 if (rbd_dev->lock_state != RBD_LOCK_STATE_RELEASING)
4141 rbd_assert(list_empty(&rbd_dev->running_list));
4145 static void rbd_pre_release_action(struct rbd_device *rbd_dev)
4147 if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)
4148 rbd_object_map_close(rbd_dev);
4151 static void __rbd_release_lock(struct rbd_device *rbd_dev)
4153 rbd_assert(list_empty(&rbd_dev->running_list));
4155 rbd_pre_release_action(rbd_dev);
4156 rbd_unlock(rbd_dev);
4160 * lock_rwsem must be held for write
4162 static void rbd_release_lock(struct rbd_device *rbd_dev)
4164 if (!rbd_quiesce_lock(rbd_dev))
4167 __rbd_release_lock(rbd_dev);
4170 * Give others a chance to grab the lock - we would re-acquire
4171 * almost immediately if we got new IO while draining the running
4172 * list otherwise. We need to ack our own notifications, so this
4173 * lock_dwork will be requeued from rbd_handle_released_lock() by
4174 * way of maybe_kick_acquire().
4176 cancel_delayed_work(&rbd_dev->lock_dwork);
4179 static void rbd_release_lock_work(struct work_struct *work)
4181 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
4184 down_write(&rbd_dev->lock_rwsem);
4185 rbd_release_lock(rbd_dev);
4186 up_write(&rbd_dev->lock_rwsem);
4189 static void maybe_kick_acquire(struct rbd_device *rbd_dev)
4193 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4194 if (__rbd_is_lock_owner(rbd_dev))
4197 spin_lock(&rbd_dev->lock_lists_lock);
4198 have_requests = !list_empty(&rbd_dev->acquiring_list);
4199 spin_unlock(&rbd_dev->lock_lists_lock);
4200 if (have_requests || delayed_work_pending(&rbd_dev->lock_dwork)) {
4201 dout("%s rbd_dev %p kicking lock_dwork\n", __func__, rbd_dev);
4202 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4206 static void rbd_handle_acquired_lock(struct rbd_device *rbd_dev, u8 struct_v,
4209 struct rbd_client_id cid = { 0 };
4211 if (struct_v >= 2) {
4212 cid.gid = ceph_decode_64(p);
4213 cid.handle = ceph_decode_64(p);
4216 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4218 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4219 down_write(&rbd_dev->lock_rwsem);
4220 if (rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4222 * we already know that the remote client is
4225 up_write(&rbd_dev->lock_rwsem);
4229 rbd_set_owner_cid(rbd_dev, &cid);
4230 downgrade_write(&rbd_dev->lock_rwsem);
4232 down_read(&rbd_dev->lock_rwsem);
4235 maybe_kick_acquire(rbd_dev);
4236 up_read(&rbd_dev->lock_rwsem);
4239 static void rbd_handle_released_lock(struct rbd_device *rbd_dev, u8 struct_v,
4242 struct rbd_client_id cid = { 0 };
4244 if (struct_v >= 2) {
4245 cid.gid = ceph_decode_64(p);
4246 cid.handle = ceph_decode_64(p);
4249 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4251 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4252 down_write(&rbd_dev->lock_rwsem);
4253 if (!rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4254 dout("%s rbd_dev %p unexpected owner, cid %llu-%llu != owner_cid %llu-%llu\n",
4255 __func__, rbd_dev, cid.gid, cid.handle,
4256 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle);
4257 up_write(&rbd_dev->lock_rwsem);
4261 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4262 downgrade_write(&rbd_dev->lock_rwsem);
4264 down_read(&rbd_dev->lock_rwsem);
4267 maybe_kick_acquire(rbd_dev);
4268 up_read(&rbd_dev->lock_rwsem);
4272 * Returns result for ResponseMessage to be encoded (<= 0), or 1 if no
4273 * ResponseMessage is needed.
4275 static int rbd_handle_request_lock(struct rbd_device *rbd_dev, u8 struct_v,
4278 struct rbd_client_id my_cid = rbd_get_cid(rbd_dev);
4279 struct rbd_client_id cid = { 0 };
4282 if (struct_v >= 2) {
4283 cid.gid = ceph_decode_64(p);
4284 cid.handle = ceph_decode_64(p);
4287 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4289 if (rbd_cid_equal(&cid, &my_cid))
4292 down_read(&rbd_dev->lock_rwsem);
4293 if (__rbd_is_lock_owner(rbd_dev)) {
4294 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED &&
4295 rbd_cid_equal(&rbd_dev->owner_cid, &rbd_empty_cid))
4299 * encode ResponseMessage(0) so the peer can detect
4304 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED) {
4305 if (!rbd_dev->opts->exclusive) {
4306 dout("%s rbd_dev %p queueing unlock_work\n",
4308 queue_work(rbd_dev->task_wq,
4309 &rbd_dev->unlock_work);
4311 /* refuse to release the lock */
4318 up_read(&rbd_dev->lock_rwsem);
4322 static void __rbd_acknowledge_notify(struct rbd_device *rbd_dev,
4323 u64 notify_id, u64 cookie, s32 *result)
4325 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4326 char buf[4 + CEPH_ENCODING_START_BLK_LEN];
4327 int buf_size = sizeof(buf);
4333 /* encode ResponseMessage */
4334 ceph_start_encoding(&p, 1, 1,
4335 buf_size - CEPH_ENCODING_START_BLK_LEN);
4336 ceph_encode_32(&p, *result);
4341 ret = ceph_osdc_notify_ack(osdc, &rbd_dev->header_oid,
4342 &rbd_dev->header_oloc, notify_id, cookie,
4345 rbd_warn(rbd_dev, "acknowledge_notify failed: %d", ret);
4348 static void rbd_acknowledge_notify(struct rbd_device *rbd_dev, u64 notify_id,
4351 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4352 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, NULL);
4355 static void rbd_acknowledge_notify_result(struct rbd_device *rbd_dev,
4356 u64 notify_id, u64 cookie, s32 result)
4358 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
4359 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, &result);
4362 static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie,
4363 u64 notifier_id, void *data, size_t data_len)
4365 struct rbd_device *rbd_dev = arg;
4367 void *const end = p + data_len;
4373 dout("%s rbd_dev %p cookie %llu notify_id %llu data_len %zu\n",
4374 __func__, rbd_dev, cookie, notify_id, data_len);
4376 ret = ceph_start_decoding(&p, end, 1, "NotifyMessage",
4379 rbd_warn(rbd_dev, "failed to decode NotifyMessage: %d",
4384 notify_op = ceph_decode_32(&p);
4386 /* legacy notification for header updates */
4387 notify_op = RBD_NOTIFY_OP_HEADER_UPDATE;
4391 dout("%s rbd_dev %p notify_op %u\n", __func__, rbd_dev, notify_op);
4392 switch (notify_op) {
4393 case RBD_NOTIFY_OP_ACQUIRED_LOCK:
4394 rbd_handle_acquired_lock(rbd_dev, struct_v, &p);
4395 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4397 case RBD_NOTIFY_OP_RELEASED_LOCK:
4398 rbd_handle_released_lock(rbd_dev, struct_v, &p);
4399 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4401 case RBD_NOTIFY_OP_REQUEST_LOCK:
4402 ret = rbd_handle_request_lock(rbd_dev, struct_v, &p);
4404 rbd_acknowledge_notify_result(rbd_dev, notify_id,
4407 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4409 case RBD_NOTIFY_OP_HEADER_UPDATE:
4410 ret = rbd_dev_refresh(rbd_dev);
4412 rbd_warn(rbd_dev, "refresh failed: %d", ret);
4414 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4417 if (rbd_is_lock_owner(rbd_dev))
4418 rbd_acknowledge_notify_result(rbd_dev, notify_id,
4419 cookie, -EOPNOTSUPP);
4421 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4426 static void __rbd_unregister_watch(struct rbd_device *rbd_dev);
4428 static void rbd_watch_errcb(void *arg, u64 cookie, int err)
4430 struct rbd_device *rbd_dev = arg;
4432 rbd_warn(rbd_dev, "encountered watch error: %d", err);
4434 down_write(&rbd_dev->lock_rwsem);
4435 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4436 up_write(&rbd_dev->lock_rwsem);
4438 mutex_lock(&rbd_dev->watch_mutex);
4439 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED) {
4440 __rbd_unregister_watch(rbd_dev);
4441 rbd_dev->watch_state = RBD_WATCH_STATE_ERROR;
4443 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->watch_dwork, 0);
4445 mutex_unlock(&rbd_dev->watch_mutex);
4449 * watch_mutex must be locked
4451 static int __rbd_register_watch(struct rbd_device *rbd_dev)
4453 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4454 struct ceph_osd_linger_request *handle;
4456 rbd_assert(!rbd_dev->watch_handle);
4457 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4459 handle = ceph_osdc_watch(osdc, &rbd_dev->header_oid,
4460 &rbd_dev->header_oloc, rbd_watch_cb,
4461 rbd_watch_errcb, rbd_dev);
4463 return PTR_ERR(handle);
4465 rbd_dev->watch_handle = handle;
4470 * watch_mutex must be locked
4472 static void __rbd_unregister_watch(struct rbd_device *rbd_dev)
4474 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4477 rbd_assert(rbd_dev->watch_handle);
4478 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4480 ret = ceph_osdc_unwatch(osdc, rbd_dev->watch_handle);
4482 rbd_warn(rbd_dev, "failed to unwatch: %d", ret);
4484 rbd_dev->watch_handle = NULL;
4487 static int rbd_register_watch(struct rbd_device *rbd_dev)
4491 mutex_lock(&rbd_dev->watch_mutex);
4492 rbd_assert(rbd_dev->watch_state == RBD_WATCH_STATE_UNREGISTERED);
4493 ret = __rbd_register_watch(rbd_dev);
4497 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4498 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4501 mutex_unlock(&rbd_dev->watch_mutex);
4505 static void cancel_tasks_sync(struct rbd_device *rbd_dev)
4507 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4509 cancel_work_sync(&rbd_dev->acquired_lock_work);
4510 cancel_work_sync(&rbd_dev->released_lock_work);
4511 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
4512 cancel_work_sync(&rbd_dev->unlock_work);
4516 * header_rwsem must not be held to avoid a deadlock with
4517 * rbd_dev_refresh() when flushing notifies.
4519 static void rbd_unregister_watch(struct rbd_device *rbd_dev)
4521 cancel_tasks_sync(rbd_dev);
4523 mutex_lock(&rbd_dev->watch_mutex);
4524 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED)
4525 __rbd_unregister_watch(rbd_dev);
4526 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
4527 mutex_unlock(&rbd_dev->watch_mutex);
4529 cancel_delayed_work_sync(&rbd_dev->watch_dwork);
4530 ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
4534 * lock_rwsem must be held for write
4536 static void rbd_reacquire_lock(struct rbd_device *rbd_dev)
4538 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4542 if (!rbd_quiesce_lock(rbd_dev))
4545 format_lock_cookie(rbd_dev, cookie);
4546 ret = ceph_cls_set_cookie(osdc, &rbd_dev->header_oid,
4547 &rbd_dev->header_oloc, RBD_LOCK_NAME,
4548 CEPH_CLS_LOCK_EXCLUSIVE, rbd_dev->lock_cookie,
4549 RBD_LOCK_TAG, cookie);
4551 if (ret != -EOPNOTSUPP)
4552 rbd_warn(rbd_dev, "failed to update lock cookie: %d",
4556 * Lock cookie cannot be updated on older OSDs, so do
4557 * a manual release and queue an acquire.
4559 __rbd_release_lock(rbd_dev);
4560 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4562 __rbd_lock(rbd_dev, cookie);
4563 wake_lock_waiters(rbd_dev, 0);
4567 static void rbd_reregister_watch(struct work_struct *work)
4569 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4570 struct rbd_device, watch_dwork);
4573 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4575 mutex_lock(&rbd_dev->watch_mutex);
4576 if (rbd_dev->watch_state != RBD_WATCH_STATE_ERROR) {
4577 mutex_unlock(&rbd_dev->watch_mutex);
4581 ret = __rbd_register_watch(rbd_dev);
4583 rbd_warn(rbd_dev, "failed to reregister watch: %d", ret);
4584 if (ret != -EBLOCKLISTED && ret != -ENOENT) {
4585 queue_delayed_work(rbd_dev->task_wq,
4586 &rbd_dev->watch_dwork,
4588 mutex_unlock(&rbd_dev->watch_mutex);
4592 mutex_unlock(&rbd_dev->watch_mutex);
4593 down_write(&rbd_dev->lock_rwsem);
4594 wake_lock_waiters(rbd_dev, ret);
4595 up_write(&rbd_dev->lock_rwsem);
4599 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4600 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4601 mutex_unlock(&rbd_dev->watch_mutex);
4603 down_write(&rbd_dev->lock_rwsem);
4604 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED)
4605 rbd_reacquire_lock(rbd_dev);
4606 up_write(&rbd_dev->lock_rwsem);
4608 ret = rbd_dev_refresh(rbd_dev);
4610 rbd_warn(rbd_dev, "reregistration refresh failed: %d", ret);
4614 * Synchronous osd object method call. Returns the number of bytes
4615 * returned in the outbound buffer, or a negative error code.
4617 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
4618 struct ceph_object_id *oid,
4619 struct ceph_object_locator *oloc,
4620 const char *method_name,
4621 const void *outbound,
4622 size_t outbound_size,
4624 size_t inbound_size)
4626 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4627 struct page *req_page = NULL;
4628 struct page *reply_page;
4632 * Method calls are ultimately read operations. The result
4633 * should placed into the inbound buffer provided. They
4634 * also supply outbound data--parameters for the object
4635 * method. Currently if this is present it will be a
4639 if (outbound_size > PAGE_SIZE)
4642 req_page = alloc_page(GFP_KERNEL);
4646 memcpy(page_address(req_page), outbound, outbound_size);
4649 reply_page = alloc_page(GFP_KERNEL);
4652 __free_page(req_page);
4656 ret = ceph_osdc_call(osdc, oid, oloc, RBD_DRV_NAME, method_name,
4657 CEPH_OSD_FLAG_READ, req_page, outbound_size,
4658 &reply_page, &inbound_size);
4660 memcpy(inbound, page_address(reply_page), inbound_size);
4665 __free_page(req_page);
4666 __free_page(reply_page);
4670 static void rbd_queue_workfn(struct work_struct *work)
4672 struct rbd_img_request *img_request =
4673 container_of(work, struct rbd_img_request, work);
4674 struct rbd_device *rbd_dev = img_request->rbd_dev;
4675 enum obj_operation_type op_type = img_request->op_type;
4676 struct request *rq = blk_mq_rq_from_pdu(img_request);
4677 u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
4678 u64 length = blk_rq_bytes(rq);
4682 /* Ignore/skip any zero-length requests */
4684 dout("%s: zero-length request\n", __func__);
4686 goto err_img_request;
4689 blk_mq_start_request(rq);
4691 down_read(&rbd_dev->header_rwsem);
4692 mapping_size = rbd_dev->mapping.size;
4693 rbd_img_capture_header(img_request);
4694 up_read(&rbd_dev->header_rwsem);
4696 if (offset + length > mapping_size) {
4697 rbd_warn(rbd_dev, "beyond EOD (%llu~%llu > %llu)", offset,
4698 length, mapping_size);
4700 goto err_img_request;
4703 dout("%s rbd_dev %p img_req %p %s %llu~%llu\n", __func__, rbd_dev,
4704 img_request, obj_op_name(op_type), offset, length);
4706 if (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_ZEROOUT)
4707 result = rbd_img_fill_nodata(img_request, offset, length);
4709 result = rbd_img_fill_from_bio(img_request, offset, length,
4712 goto err_img_request;
4714 rbd_img_handle_request(img_request, 0);
4718 rbd_img_request_destroy(img_request);
4720 rbd_warn(rbd_dev, "%s %llx at %llx result %d",
4721 obj_op_name(op_type), length, offset, result);
4722 blk_mq_end_request(rq, errno_to_blk_status(result));
4725 static blk_status_t rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
4726 const struct blk_mq_queue_data *bd)
4728 struct rbd_device *rbd_dev = hctx->queue->queuedata;
4729 struct rbd_img_request *img_req = blk_mq_rq_to_pdu(bd->rq);
4730 enum obj_operation_type op_type;
4732 switch (req_op(bd->rq)) {
4733 case REQ_OP_DISCARD:
4734 op_type = OBJ_OP_DISCARD;
4736 case REQ_OP_WRITE_ZEROES:
4737 op_type = OBJ_OP_ZEROOUT;
4740 op_type = OBJ_OP_WRITE;
4743 op_type = OBJ_OP_READ;
4746 rbd_warn(rbd_dev, "unknown req_op %d", req_op(bd->rq));
4747 return BLK_STS_IOERR;
4750 rbd_img_request_init(img_req, rbd_dev, op_type);
4752 if (rbd_img_is_write(img_req)) {
4753 if (rbd_is_ro(rbd_dev)) {
4754 rbd_warn(rbd_dev, "%s on read-only mapping",
4755 obj_op_name(img_req->op_type));
4756 return BLK_STS_IOERR;
4758 rbd_assert(!rbd_is_snap(rbd_dev));
4761 INIT_WORK(&img_req->work, rbd_queue_workfn);
4762 queue_work(rbd_wq, &img_req->work);
4766 static void rbd_free_disk(struct rbd_device *rbd_dev)
4768 blk_cleanup_queue(rbd_dev->disk->queue);
4769 blk_mq_free_tag_set(&rbd_dev->tag_set);
4770 put_disk(rbd_dev->disk);
4771 rbd_dev->disk = NULL;
4774 static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
4775 struct ceph_object_id *oid,
4776 struct ceph_object_locator *oloc,
4777 void *buf, int buf_len)
4780 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4781 struct ceph_osd_request *req;
4782 struct page **pages;
4783 int num_pages = calc_pages_for(0, buf_len);
4786 req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_KERNEL);
4790 ceph_oid_copy(&req->r_base_oid, oid);
4791 ceph_oloc_copy(&req->r_base_oloc, oloc);
4792 req->r_flags = CEPH_OSD_FLAG_READ;
4794 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
4795 if (IS_ERR(pages)) {
4796 ret = PTR_ERR(pages);
4800 osd_req_op_extent_init(req, 0, CEPH_OSD_OP_READ, 0, buf_len, 0, 0);
4801 osd_req_op_extent_osd_data_pages(req, 0, pages, buf_len, 0, false,
4804 ret = ceph_osdc_alloc_messages(req, GFP_KERNEL);
4808 ceph_osdc_start_request(osdc, req, false);
4809 ret = ceph_osdc_wait_request(osdc, req);
4811 ceph_copy_from_page_vector(pages, buf, 0, ret);
4814 ceph_osdc_put_request(req);
4819 * Read the complete header for the given rbd device. On successful
4820 * return, the rbd_dev->header field will contain up-to-date
4821 * information about the image.
4823 static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev)
4825 struct rbd_image_header_ondisk *ondisk = NULL;
4832 * The complete header will include an array of its 64-bit
4833 * snapshot ids, followed by the names of those snapshots as
4834 * a contiguous block of NUL-terminated strings. Note that
4835 * the number of snapshots could change by the time we read
4836 * it in, in which case we re-read it.
4843 size = sizeof (*ondisk);
4844 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
4846 ondisk = kmalloc(size, GFP_KERNEL);
4850 ret = rbd_obj_read_sync(rbd_dev, &rbd_dev->header_oid,
4851 &rbd_dev->header_oloc, ondisk, size);
4854 if ((size_t)ret < size) {
4856 rbd_warn(rbd_dev, "short header read (want %zd got %d)",
4860 if (!rbd_dev_ondisk_valid(ondisk)) {
4862 rbd_warn(rbd_dev, "invalid header");
4866 names_size = le64_to_cpu(ondisk->snap_names_len);
4867 want_count = snap_count;
4868 snap_count = le32_to_cpu(ondisk->snap_count);
4869 } while (snap_count != want_count);
4871 ret = rbd_header_from_disk(rbd_dev, ondisk);
4878 static void rbd_dev_update_size(struct rbd_device *rbd_dev)
4883 * If EXISTS is not set, rbd_dev->disk may be NULL, so don't
4884 * try to update its size. If REMOVING is set, updating size
4885 * is just useless work since the device can't be opened.
4887 if (test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags) &&
4888 !test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) {
4889 size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
4890 dout("setting size to %llu sectors", (unsigned long long)size);
4891 set_capacity_and_notify(rbd_dev->disk, size);
4895 static int rbd_dev_refresh(struct rbd_device *rbd_dev)
4900 down_write(&rbd_dev->header_rwsem);
4901 mapping_size = rbd_dev->mapping.size;
4903 ret = rbd_dev_header_info(rbd_dev);
4908 * If there is a parent, see if it has disappeared due to the
4909 * mapped image getting flattened.
4911 if (rbd_dev->parent) {
4912 ret = rbd_dev_v2_parent_info(rbd_dev);
4917 rbd_assert(!rbd_is_snap(rbd_dev));
4918 rbd_dev->mapping.size = rbd_dev->header.image_size;
4921 up_write(&rbd_dev->header_rwsem);
4922 if (!ret && mapping_size != rbd_dev->mapping.size)
4923 rbd_dev_update_size(rbd_dev);
4928 static const struct blk_mq_ops rbd_mq_ops = {
4929 .queue_rq = rbd_queue_rq,
4932 static int rbd_init_disk(struct rbd_device *rbd_dev)
4934 struct gendisk *disk;
4935 struct request_queue *q;
4936 unsigned int objset_bytes =
4937 rbd_dev->layout.object_size * rbd_dev->layout.stripe_count;
4940 /* create gendisk info */
4941 disk = alloc_disk(single_major ?
4942 (1 << RBD_SINGLE_MAJOR_PART_SHIFT) :
4943 RBD_MINORS_PER_MAJOR);
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->flags |= GENHD_FL_EXT_DEVT;
4953 disk->fops = &rbd_bd_ops;
4954 disk->private_data = rbd_dev;
4956 memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
4957 rbd_dev->tag_set.ops = &rbd_mq_ops;
4958 rbd_dev->tag_set.queue_depth = rbd_dev->opts->queue_depth;
4959 rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
4960 rbd_dev->tag_set.flags = BLK_MQ_F_SHOULD_MERGE;
4961 rbd_dev->tag_set.nr_hw_queues = num_present_cpus();
4962 rbd_dev->tag_set.cmd_size = sizeof(struct rbd_img_request);
4964 err = blk_mq_alloc_tag_set(&rbd_dev->tag_set);
4968 q = blk_mq_init_queue(&rbd_dev->tag_set);
4974 blk_queue_flag_set(QUEUE_FLAG_NONROT, q);
4975 /* QUEUE_FLAG_ADD_RANDOM is off by default for blk-mq */
4977 blk_queue_max_hw_sectors(q, objset_bytes >> SECTOR_SHIFT);
4978 q->limits.max_sectors = queue_max_hw_sectors(q);
4979 blk_queue_max_segments(q, USHRT_MAX);
4980 blk_queue_max_segment_size(q, UINT_MAX);
4981 blk_queue_io_min(q, rbd_dev->opts->alloc_size);
4982 blk_queue_io_opt(q, rbd_dev->opts->alloc_size);
4984 if (rbd_dev->opts->trim) {
4985 blk_queue_flag_set(QUEUE_FLAG_DISCARD, q);
4986 q->limits.discard_granularity = rbd_dev->opts->alloc_size;
4987 blk_queue_max_discard_sectors(q, objset_bytes >> SECTOR_SHIFT);
4988 blk_queue_max_write_zeroes_sectors(q, objset_bytes >> SECTOR_SHIFT);
4991 if (!ceph_test_opt(rbd_dev->rbd_client->client, NOCRC))
4992 blk_queue_flag_set(QUEUE_FLAG_STABLE_WRITES, q);
4995 * disk_release() expects a queue ref from add_disk() and will
4996 * put it. Hold an extra ref until add_disk() is called.
4998 WARN_ON(!blk_get_queue(q));
5000 q->queuedata = rbd_dev;
5002 rbd_dev->disk = disk;
5006 blk_mq_free_tag_set(&rbd_dev->tag_set);
5016 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
5018 return container_of(dev, struct rbd_device, dev);
5021 static ssize_t rbd_size_show(struct device *dev,
5022 struct device_attribute *attr, char *buf)
5024 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5026 return sprintf(buf, "%llu\n",
5027 (unsigned long long)rbd_dev->mapping.size);
5030 static ssize_t rbd_features_show(struct device *dev,
5031 struct device_attribute *attr, char *buf)
5033 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5035 return sprintf(buf, "0x%016llx\n", rbd_dev->header.features);
5038 static ssize_t rbd_major_show(struct device *dev,
5039 struct device_attribute *attr, char *buf)
5041 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5044 return sprintf(buf, "%d\n", rbd_dev->major);
5046 return sprintf(buf, "(none)\n");
5049 static ssize_t rbd_minor_show(struct device *dev,
5050 struct device_attribute *attr, char *buf)
5052 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5054 return sprintf(buf, "%d\n", rbd_dev->minor);
5057 static ssize_t rbd_client_addr_show(struct device *dev,
5058 struct device_attribute *attr, char *buf)
5060 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5061 struct ceph_entity_addr *client_addr =
5062 ceph_client_addr(rbd_dev->rbd_client->client);
5064 return sprintf(buf, "%pISpc/%u\n", &client_addr->in_addr,
5065 le32_to_cpu(client_addr->nonce));
5068 static ssize_t rbd_client_id_show(struct device *dev,
5069 struct device_attribute *attr, char *buf)
5071 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5073 return sprintf(buf, "client%lld\n",
5074 ceph_client_gid(rbd_dev->rbd_client->client));
5077 static ssize_t rbd_cluster_fsid_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, "%pU\n", &rbd_dev->rbd_client->client->fsid);
5085 static ssize_t rbd_config_info_show(struct device *dev,
5086 struct device_attribute *attr, char *buf)
5088 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5090 if (!capable(CAP_SYS_ADMIN))
5093 return sprintf(buf, "%s\n", rbd_dev->config_info);
5096 static ssize_t rbd_pool_show(struct device *dev,
5097 struct device_attribute *attr, char *buf)
5099 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5101 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
5104 static ssize_t rbd_pool_id_show(struct device *dev,
5105 struct device_attribute *attr, char *buf)
5107 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5109 return sprintf(buf, "%llu\n",
5110 (unsigned long long) rbd_dev->spec->pool_id);
5113 static ssize_t rbd_pool_ns_show(struct device *dev,
5114 struct device_attribute *attr, char *buf)
5116 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5118 return sprintf(buf, "%s\n", rbd_dev->spec->pool_ns ?: "");
5121 static ssize_t rbd_name_show(struct device *dev,
5122 struct device_attribute *attr, char *buf)
5124 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5126 if (rbd_dev->spec->image_name)
5127 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
5129 return sprintf(buf, "(unknown)\n");
5132 static ssize_t rbd_image_id_show(struct device *dev,
5133 struct device_attribute *attr, char *buf)
5135 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5137 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
5141 * Shows the name of the currently-mapped snapshot (or
5142 * RBD_SNAP_HEAD_NAME for the base image).
5144 static ssize_t rbd_snap_show(struct device *dev,
5145 struct device_attribute *attr,
5148 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5150 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
5153 static ssize_t rbd_snap_id_show(struct device *dev,
5154 struct device_attribute *attr, char *buf)
5156 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5158 return sprintf(buf, "%llu\n", rbd_dev->spec->snap_id);
5162 * For a v2 image, shows the chain of parent images, separated by empty
5163 * lines. For v1 images or if there is no parent, shows "(no parent
5166 static ssize_t rbd_parent_show(struct device *dev,
5167 struct device_attribute *attr,
5170 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5173 if (!rbd_dev->parent)
5174 return sprintf(buf, "(no parent image)\n");
5176 for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) {
5177 struct rbd_spec *spec = rbd_dev->parent_spec;
5179 count += sprintf(&buf[count], "%s"
5180 "pool_id %llu\npool_name %s\n"
5182 "image_id %s\nimage_name %s\n"
5183 "snap_id %llu\nsnap_name %s\n"
5185 !count ? "" : "\n", /* first? */
5186 spec->pool_id, spec->pool_name,
5187 spec->pool_ns ?: "",
5188 spec->image_id, spec->image_name ?: "(unknown)",
5189 spec->snap_id, spec->snap_name,
5190 rbd_dev->parent_overlap);
5196 static ssize_t rbd_image_refresh(struct device *dev,
5197 struct device_attribute *attr,
5201 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5204 if (!capable(CAP_SYS_ADMIN))
5207 ret = rbd_dev_refresh(rbd_dev);
5214 static DEVICE_ATTR(size, 0444, rbd_size_show, NULL);
5215 static DEVICE_ATTR(features, 0444, rbd_features_show, NULL);
5216 static DEVICE_ATTR(major, 0444, rbd_major_show, NULL);
5217 static DEVICE_ATTR(minor, 0444, rbd_minor_show, NULL);
5218 static DEVICE_ATTR(client_addr, 0444, rbd_client_addr_show, NULL);
5219 static DEVICE_ATTR(client_id, 0444, rbd_client_id_show, NULL);
5220 static DEVICE_ATTR(cluster_fsid, 0444, rbd_cluster_fsid_show, NULL);
5221 static DEVICE_ATTR(config_info, 0400, rbd_config_info_show, NULL);
5222 static DEVICE_ATTR(pool, 0444, rbd_pool_show, NULL);
5223 static DEVICE_ATTR(pool_id, 0444, rbd_pool_id_show, NULL);
5224 static DEVICE_ATTR(pool_ns, 0444, rbd_pool_ns_show, NULL);
5225 static DEVICE_ATTR(name, 0444, rbd_name_show, NULL);
5226 static DEVICE_ATTR(image_id, 0444, rbd_image_id_show, NULL);
5227 static DEVICE_ATTR(refresh, 0200, NULL, rbd_image_refresh);
5228 static DEVICE_ATTR(current_snap, 0444, rbd_snap_show, NULL);
5229 static DEVICE_ATTR(snap_id, 0444, rbd_snap_id_show, NULL);
5230 static DEVICE_ATTR(parent, 0444, rbd_parent_show, NULL);
5232 static struct attribute *rbd_attrs[] = {
5233 &dev_attr_size.attr,
5234 &dev_attr_features.attr,
5235 &dev_attr_major.attr,
5236 &dev_attr_minor.attr,
5237 &dev_attr_client_addr.attr,
5238 &dev_attr_client_id.attr,
5239 &dev_attr_cluster_fsid.attr,
5240 &dev_attr_config_info.attr,
5241 &dev_attr_pool.attr,
5242 &dev_attr_pool_id.attr,
5243 &dev_attr_pool_ns.attr,
5244 &dev_attr_name.attr,
5245 &dev_attr_image_id.attr,
5246 &dev_attr_current_snap.attr,
5247 &dev_attr_snap_id.attr,
5248 &dev_attr_parent.attr,
5249 &dev_attr_refresh.attr,
5253 static struct attribute_group rbd_attr_group = {
5257 static const struct attribute_group *rbd_attr_groups[] = {
5262 static void rbd_dev_release(struct device *dev);
5264 static const struct device_type rbd_device_type = {
5266 .groups = rbd_attr_groups,
5267 .release = rbd_dev_release,
5270 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
5272 kref_get(&spec->kref);
5277 static void rbd_spec_free(struct kref *kref);
5278 static void rbd_spec_put(struct rbd_spec *spec)
5281 kref_put(&spec->kref, rbd_spec_free);
5284 static struct rbd_spec *rbd_spec_alloc(void)
5286 struct rbd_spec *spec;
5288 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
5292 spec->pool_id = CEPH_NOPOOL;
5293 spec->snap_id = CEPH_NOSNAP;
5294 kref_init(&spec->kref);
5299 static void rbd_spec_free(struct kref *kref)
5301 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
5303 kfree(spec->pool_name);
5304 kfree(spec->pool_ns);
5305 kfree(spec->image_id);
5306 kfree(spec->image_name);
5307 kfree(spec->snap_name);
5311 static void rbd_dev_free(struct rbd_device *rbd_dev)
5313 WARN_ON(rbd_dev->watch_state != RBD_WATCH_STATE_UNREGISTERED);
5314 WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_UNLOCKED);
5316 ceph_oid_destroy(&rbd_dev->header_oid);
5317 ceph_oloc_destroy(&rbd_dev->header_oloc);
5318 kfree(rbd_dev->config_info);
5320 rbd_put_client(rbd_dev->rbd_client);
5321 rbd_spec_put(rbd_dev->spec);
5322 kfree(rbd_dev->opts);
5326 static void rbd_dev_release(struct device *dev)
5328 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5329 bool need_put = !!rbd_dev->opts;
5332 destroy_workqueue(rbd_dev->task_wq);
5333 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5336 rbd_dev_free(rbd_dev);
5339 * This is racy, but way better than putting module outside of
5340 * the release callback. The race window is pretty small, so
5341 * doing something similar to dm (dm-builtin.c) is overkill.
5344 module_put(THIS_MODULE);
5347 static struct rbd_device *__rbd_dev_create(struct rbd_client *rbdc,
5348 struct rbd_spec *spec)
5350 struct rbd_device *rbd_dev;
5352 rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
5356 spin_lock_init(&rbd_dev->lock);
5357 INIT_LIST_HEAD(&rbd_dev->node);
5358 init_rwsem(&rbd_dev->header_rwsem);
5360 rbd_dev->header.data_pool_id = CEPH_NOPOOL;
5361 ceph_oid_init(&rbd_dev->header_oid);
5362 rbd_dev->header_oloc.pool = spec->pool_id;
5363 if (spec->pool_ns) {
5364 WARN_ON(!*spec->pool_ns);
5365 rbd_dev->header_oloc.pool_ns =
5366 ceph_find_or_create_string(spec->pool_ns,
5367 strlen(spec->pool_ns));
5370 mutex_init(&rbd_dev->watch_mutex);
5371 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
5372 INIT_DELAYED_WORK(&rbd_dev->watch_dwork, rbd_reregister_watch);
5374 init_rwsem(&rbd_dev->lock_rwsem);
5375 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
5376 INIT_WORK(&rbd_dev->acquired_lock_work, rbd_notify_acquired_lock);
5377 INIT_WORK(&rbd_dev->released_lock_work, rbd_notify_released_lock);
5378 INIT_DELAYED_WORK(&rbd_dev->lock_dwork, rbd_acquire_lock);
5379 INIT_WORK(&rbd_dev->unlock_work, rbd_release_lock_work);
5380 spin_lock_init(&rbd_dev->lock_lists_lock);
5381 INIT_LIST_HEAD(&rbd_dev->acquiring_list);
5382 INIT_LIST_HEAD(&rbd_dev->running_list);
5383 init_completion(&rbd_dev->acquire_wait);
5384 init_completion(&rbd_dev->releasing_wait);
5386 spin_lock_init(&rbd_dev->object_map_lock);
5388 rbd_dev->dev.bus = &rbd_bus_type;
5389 rbd_dev->dev.type = &rbd_device_type;
5390 rbd_dev->dev.parent = &rbd_root_dev;
5391 device_initialize(&rbd_dev->dev);
5393 rbd_dev->rbd_client = rbdc;
5394 rbd_dev->spec = spec;
5400 * Create a mapping rbd_dev.
5402 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
5403 struct rbd_spec *spec,
5404 struct rbd_options *opts)
5406 struct rbd_device *rbd_dev;
5408 rbd_dev = __rbd_dev_create(rbdc, spec);
5412 rbd_dev->opts = opts;
5414 /* get an id and fill in device name */
5415 rbd_dev->dev_id = ida_simple_get(&rbd_dev_id_ida, 0,
5416 minor_to_rbd_dev_id(1 << MINORBITS),
5418 if (rbd_dev->dev_id < 0)
5421 sprintf(rbd_dev->name, RBD_DRV_NAME "%d", rbd_dev->dev_id);
5422 rbd_dev->task_wq = alloc_ordered_workqueue("%s-tasks", WQ_MEM_RECLAIM,
5424 if (!rbd_dev->task_wq)
5427 /* we have a ref from do_rbd_add() */
5428 __module_get(THIS_MODULE);
5430 dout("%s rbd_dev %p dev_id %d\n", __func__, rbd_dev, rbd_dev->dev_id);
5434 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5436 rbd_dev_free(rbd_dev);
5440 static void rbd_dev_destroy(struct rbd_device *rbd_dev)
5443 put_device(&rbd_dev->dev);
5447 * Get the size and object order for an image snapshot, or if
5448 * snap_id is CEPH_NOSNAP, gets this information for the base
5451 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
5452 u8 *order, u64 *snap_size)
5454 __le64 snapid = cpu_to_le64(snap_id);
5459 } __attribute__ ((packed)) size_buf = { 0 };
5461 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5462 &rbd_dev->header_oloc, "get_size",
5463 &snapid, sizeof(snapid),
5464 &size_buf, sizeof(size_buf));
5465 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5468 if (ret < sizeof (size_buf))
5472 *order = size_buf.order;
5473 dout(" order %u", (unsigned int)*order);
5475 *snap_size = le64_to_cpu(size_buf.size);
5477 dout(" snap_id 0x%016llx snap_size = %llu\n",
5478 (unsigned long long)snap_id,
5479 (unsigned long long)*snap_size);
5484 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
5486 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
5487 &rbd_dev->header.obj_order,
5488 &rbd_dev->header.image_size);
5491 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
5498 /* Response will be an encoded string, which includes a length */
5499 size = sizeof(__le32) + RBD_OBJ_PREFIX_LEN_MAX;
5500 reply_buf = kzalloc(size, GFP_KERNEL);
5504 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5505 &rbd_dev->header_oloc, "get_object_prefix",
5506 NULL, 0, reply_buf, size);
5507 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5512 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
5513 p + ret, NULL, GFP_NOIO);
5516 if (IS_ERR(rbd_dev->header.object_prefix)) {
5517 ret = PTR_ERR(rbd_dev->header.object_prefix);
5518 rbd_dev->header.object_prefix = NULL;
5520 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix);
5528 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
5529 bool read_only, u64 *snap_features)
5538 } __attribute__ ((packed)) features_buf = { 0 };
5542 features_in.snap_id = cpu_to_le64(snap_id);
5543 features_in.read_only = read_only;
5545 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5546 &rbd_dev->header_oloc, "get_features",
5547 &features_in, sizeof(features_in),
5548 &features_buf, sizeof(features_buf));
5549 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5552 if (ret < sizeof (features_buf))
5555 unsup = le64_to_cpu(features_buf.incompat) & ~RBD_FEATURES_SUPPORTED;
5557 rbd_warn(rbd_dev, "image uses unsupported features: 0x%llx",
5562 *snap_features = le64_to_cpu(features_buf.features);
5564 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
5565 (unsigned long long)snap_id,
5566 (unsigned long long)*snap_features,
5567 (unsigned long long)le64_to_cpu(features_buf.incompat));
5572 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
5574 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
5576 &rbd_dev->header.features);
5580 * These are generic image flags, but since they are used only for
5581 * object map, store them in rbd_dev->object_map_flags.
5583 * For the same reason, this function is called only on object map
5584 * (re)load and not on header refresh.
5586 static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev)
5588 __le64 snapid = cpu_to_le64(rbd_dev->spec->snap_id);
5592 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5593 &rbd_dev->header_oloc, "get_flags",
5594 &snapid, sizeof(snapid),
5595 &flags, sizeof(flags));
5598 if (ret < sizeof(flags))
5601 rbd_dev->object_map_flags = le64_to_cpu(flags);
5605 struct parent_image_info {
5607 const char *pool_ns;
5608 const char *image_id;
5616 * The caller is responsible for @pii.
5618 static int decode_parent_image_spec(void **p, void *end,
5619 struct parent_image_info *pii)
5625 ret = ceph_start_decoding(p, end, 1, "ParentImageSpec",
5626 &struct_v, &struct_len);
5630 ceph_decode_64_safe(p, end, pii->pool_id, e_inval);
5631 pii->pool_ns = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5632 if (IS_ERR(pii->pool_ns)) {
5633 ret = PTR_ERR(pii->pool_ns);
5634 pii->pool_ns = NULL;
5637 pii->image_id = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5638 if (IS_ERR(pii->image_id)) {
5639 ret = PTR_ERR(pii->image_id);
5640 pii->image_id = NULL;
5643 ceph_decode_64_safe(p, end, pii->snap_id, e_inval);
5650 static int __get_parent_info(struct rbd_device *rbd_dev,
5651 struct page *req_page,
5652 struct page *reply_page,
5653 struct parent_image_info *pii)
5655 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5656 size_t reply_len = PAGE_SIZE;
5660 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5661 "rbd", "parent_get", CEPH_OSD_FLAG_READ,
5662 req_page, sizeof(u64), &reply_page, &reply_len);
5664 return ret == -EOPNOTSUPP ? 1 : ret;
5666 p = page_address(reply_page);
5667 end = p + reply_len;
5668 ret = decode_parent_image_spec(&p, end, pii);
5672 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5673 "rbd", "parent_overlap_get", CEPH_OSD_FLAG_READ,
5674 req_page, sizeof(u64), &reply_page, &reply_len);
5678 p = page_address(reply_page);
5679 end = p + reply_len;
5680 ceph_decode_8_safe(&p, end, pii->has_overlap, e_inval);
5681 if (pii->has_overlap)
5682 ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5691 * The caller is responsible for @pii.
5693 static int __get_parent_info_legacy(struct rbd_device *rbd_dev,
5694 struct page *req_page,
5695 struct page *reply_page,
5696 struct parent_image_info *pii)
5698 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5699 size_t reply_len = PAGE_SIZE;
5703 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5704 "rbd", "get_parent", CEPH_OSD_FLAG_READ,
5705 req_page, sizeof(u64), &reply_page, &reply_len);
5709 p = page_address(reply_page);
5710 end = p + reply_len;
5711 ceph_decode_64_safe(&p, end, pii->pool_id, e_inval);
5712 pii->image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
5713 if (IS_ERR(pii->image_id)) {
5714 ret = PTR_ERR(pii->image_id);
5715 pii->image_id = NULL;
5718 ceph_decode_64_safe(&p, end, pii->snap_id, e_inval);
5719 pii->has_overlap = true;
5720 ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5728 static int get_parent_info(struct rbd_device *rbd_dev,
5729 struct parent_image_info *pii)
5731 struct page *req_page, *reply_page;
5735 req_page = alloc_page(GFP_KERNEL);
5739 reply_page = alloc_page(GFP_KERNEL);
5741 __free_page(req_page);
5745 p = page_address(req_page);
5746 ceph_encode_64(&p, rbd_dev->spec->snap_id);
5747 ret = __get_parent_info(rbd_dev, req_page, reply_page, pii);
5749 ret = __get_parent_info_legacy(rbd_dev, req_page, reply_page,
5752 __free_page(req_page);
5753 __free_page(reply_page);
5757 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
5759 struct rbd_spec *parent_spec;
5760 struct parent_image_info pii = { 0 };
5763 parent_spec = rbd_spec_alloc();
5767 ret = get_parent_info(rbd_dev, &pii);
5771 dout("%s pool_id %llu pool_ns %s image_id %s snap_id %llu has_overlap %d overlap %llu\n",
5772 __func__, pii.pool_id, pii.pool_ns, pii.image_id, pii.snap_id,
5773 pii.has_overlap, pii.overlap);
5775 if (pii.pool_id == CEPH_NOPOOL || !pii.has_overlap) {
5777 * Either the parent never existed, or we have
5778 * record of it but the image got flattened so it no
5779 * longer has a parent. When the parent of a
5780 * layered image disappears we immediately set the
5781 * overlap to 0. The effect of this is that all new
5782 * requests will be treated as if the image had no
5785 * If !pii.has_overlap, the parent image spec is not
5786 * applicable. It's there to avoid duplication in each
5789 if (rbd_dev->parent_overlap) {
5790 rbd_dev->parent_overlap = 0;
5791 rbd_dev_parent_put(rbd_dev);
5792 pr_info("%s: clone image has been flattened\n",
5793 rbd_dev->disk->disk_name);
5796 goto out; /* No parent? No problem. */
5799 /* The ceph file layout needs to fit pool id in 32 bits */
5802 if (pii.pool_id > (u64)U32_MAX) {
5803 rbd_warn(NULL, "parent pool id too large (%llu > %u)",
5804 (unsigned long long)pii.pool_id, U32_MAX);
5809 * The parent won't change (except when the clone is
5810 * flattened, already handled that). So we only need to
5811 * record the parent spec we have not already done so.
5813 if (!rbd_dev->parent_spec) {
5814 parent_spec->pool_id = pii.pool_id;
5815 if (pii.pool_ns && *pii.pool_ns) {
5816 parent_spec->pool_ns = pii.pool_ns;
5819 parent_spec->image_id = pii.image_id;
5820 pii.image_id = NULL;
5821 parent_spec->snap_id = pii.snap_id;
5823 rbd_dev->parent_spec = parent_spec;
5824 parent_spec = NULL; /* rbd_dev now owns this */
5828 * We always update the parent overlap. If it's zero we issue
5829 * a warning, as we will proceed as if there was no parent.
5833 /* refresh, careful to warn just once */
5834 if (rbd_dev->parent_overlap)
5836 "clone now standalone (overlap became 0)");
5839 rbd_warn(rbd_dev, "clone is standalone (overlap 0)");
5842 rbd_dev->parent_overlap = pii.overlap;
5848 kfree(pii.image_id);
5849 rbd_spec_put(parent_spec);
5853 static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev)
5857 __le64 stripe_count;
5858 } __attribute__ ((packed)) striping_info_buf = { 0 };
5859 size_t size = sizeof (striping_info_buf);
5863 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5864 &rbd_dev->header_oloc, "get_stripe_unit_count",
5865 NULL, 0, &striping_info_buf, size);
5866 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5872 p = &striping_info_buf;
5873 rbd_dev->header.stripe_unit = ceph_decode_64(&p);
5874 rbd_dev->header.stripe_count = ceph_decode_64(&p);
5878 static int rbd_dev_v2_data_pool(struct rbd_device *rbd_dev)
5880 __le64 data_pool_id;
5883 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5884 &rbd_dev->header_oloc, "get_data_pool",
5885 NULL, 0, &data_pool_id, sizeof(data_pool_id));
5888 if (ret < sizeof(data_pool_id))
5891 rbd_dev->header.data_pool_id = le64_to_cpu(data_pool_id);
5892 WARN_ON(rbd_dev->header.data_pool_id == CEPH_NOPOOL);
5896 static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
5898 CEPH_DEFINE_OID_ONSTACK(oid);
5899 size_t image_id_size;
5904 void *reply_buf = NULL;
5906 char *image_name = NULL;
5909 rbd_assert(!rbd_dev->spec->image_name);
5911 len = strlen(rbd_dev->spec->image_id);
5912 image_id_size = sizeof (__le32) + len;
5913 image_id = kmalloc(image_id_size, GFP_KERNEL);
5918 end = image_id + image_id_size;
5919 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len);
5921 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
5922 reply_buf = kmalloc(size, GFP_KERNEL);
5926 ceph_oid_printf(&oid, "%s", RBD_DIRECTORY);
5927 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
5928 "dir_get_name", image_id, image_id_size,
5933 end = reply_buf + ret;
5935 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
5936 if (IS_ERR(image_name))
5939 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
5947 static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5949 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5950 const char *snap_name;
5953 /* Skip over names until we find the one we are looking for */
5955 snap_name = rbd_dev->header.snap_names;
5956 while (which < snapc->num_snaps) {
5957 if (!strcmp(name, snap_name))
5958 return snapc->snaps[which];
5959 snap_name += strlen(snap_name) + 1;
5965 static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5967 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5972 for (which = 0; !found && which < snapc->num_snaps; which++) {
5973 const char *snap_name;
5975 snap_id = snapc->snaps[which];
5976 snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
5977 if (IS_ERR(snap_name)) {
5978 /* ignore no-longer existing snapshots */
5979 if (PTR_ERR(snap_name) == -ENOENT)
5984 found = !strcmp(name, snap_name);
5987 return found ? snap_id : CEPH_NOSNAP;
5991 * Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if
5992 * no snapshot by that name is found, or if an error occurs.
5994 static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5996 if (rbd_dev->image_format == 1)
5997 return rbd_v1_snap_id_by_name(rbd_dev, name);
5999 return rbd_v2_snap_id_by_name(rbd_dev, name);
6003 * An image being mapped will have everything but the snap id.
6005 static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev)
6007 struct rbd_spec *spec = rbd_dev->spec;
6009 rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name);
6010 rbd_assert(spec->image_id && spec->image_name);
6011 rbd_assert(spec->snap_name);
6013 if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) {
6016 snap_id = rbd_snap_id_by_name(rbd_dev, spec->snap_name);
6017 if (snap_id == CEPH_NOSNAP)
6020 spec->snap_id = snap_id;
6022 spec->snap_id = CEPH_NOSNAP;
6029 * A parent image will have all ids but none of the names.
6031 * All names in an rbd spec are dynamically allocated. It's OK if we
6032 * can't figure out the name for an image id.
6034 static int rbd_spec_fill_names(struct rbd_device *rbd_dev)
6036 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
6037 struct rbd_spec *spec = rbd_dev->spec;
6038 const char *pool_name;
6039 const char *image_name;
6040 const char *snap_name;
6043 rbd_assert(spec->pool_id != CEPH_NOPOOL);
6044 rbd_assert(spec->image_id);
6045 rbd_assert(spec->snap_id != CEPH_NOSNAP);
6047 /* Get the pool name; we have to make our own copy of this */
6049 pool_name = ceph_pg_pool_name_by_id(osdc->osdmap, spec->pool_id);
6051 rbd_warn(rbd_dev, "no pool with id %llu", spec->pool_id);
6054 pool_name = kstrdup(pool_name, GFP_KERNEL);
6058 /* Fetch the image name; tolerate failure here */
6060 image_name = rbd_dev_image_name(rbd_dev);
6062 rbd_warn(rbd_dev, "unable to get image name");
6064 /* Fetch the snapshot name */
6066 snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
6067 if (IS_ERR(snap_name)) {
6068 ret = PTR_ERR(snap_name);
6072 spec->pool_name = pool_name;
6073 spec->image_name = image_name;
6074 spec->snap_name = snap_name;
6084 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev)
6093 struct ceph_snap_context *snapc;
6097 * We'll need room for the seq value (maximum snapshot id),
6098 * snapshot count, and array of that many snapshot ids.
6099 * For now we have a fixed upper limit on the number we're
6100 * prepared to receive.
6102 size = sizeof (__le64) + sizeof (__le32) +
6103 RBD_MAX_SNAP_COUNT * sizeof (__le64);
6104 reply_buf = kzalloc(size, GFP_KERNEL);
6108 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6109 &rbd_dev->header_oloc, "get_snapcontext",
6110 NULL, 0, reply_buf, size);
6111 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6116 end = reply_buf + ret;
6118 ceph_decode_64_safe(&p, end, seq, out);
6119 ceph_decode_32_safe(&p, end, snap_count, out);
6122 * Make sure the reported number of snapshot ids wouldn't go
6123 * beyond the end of our buffer. But before checking that,
6124 * make sure the computed size of the snapshot context we
6125 * allocate is representable in a size_t.
6127 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
6132 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
6136 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
6142 for (i = 0; i < snap_count; i++)
6143 snapc->snaps[i] = ceph_decode_64(&p);
6145 ceph_put_snap_context(rbd_dev->header.snapc);
6146 rbd_dev->header.snapc = snapc;
6148 dout(" snap context seq = %llu, snap_count = %u\n",
6149 (unsigned long long)seq, (unsigned int)snap_count);
6156 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
6167 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
6168 reply_buf = kmalloc(size, GFP_KERNEL);
6170 return ERR_PTR(-ENOMEM);
6172 snapid = cpu_to_le64(snap_id);
6173 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6174 &rbd_dev->header_oloc, "get_snapshot_name",
6175 &snapid, sizeof(snapid), reply_buf, size);
6176 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6178 snap_name = ERR_PTR(ret);
6183 end = reply_buf + ret;
6184 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
6185 if (IS_ERR(snap_name))
6188 dout(" snap_id 0x%016llx snap_name = %s\n",
6189 (unsigned long long)snap_id, snap_name);
6196 static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev)
6198 bool first_time = rbd_dev->header.object_prefix == NULL;
6201 ret = rbd_dev_v2_image_size(rbd_dev);
6206 ret = rbd_dev_v2_header_onetime(rbd_dev);
6211 ret = rbd_dev_v2_snap_context(rbd_dev);
6212 if (ret && first_time) {
6213 kfree(rbd_dev->header.object_prefix);
6214 rbd_dev->header.object_prefix = NULL;
6220 static int rbd_dev_header_info(struct rbd_device *rbd_dev)
6222 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6224 if (rbd_dev->image_format == 1)
6225 return rbd_dev_v1_header_info(rbd_dev);
6227 return rbd_dev_v2_header_info(rbd_dev);
6231 * Skips over white space at *buf, and updates *buf to point to the
6232 * first found non-space character (if any). Returns the length of
6233 * the token (string of non-white space characters) found. Note
6234 * that *buf must be terminated with '\0'.
6236 static inline size_t next_token(const char **buf)
6239 * These are the characters that produce nonzero for
6240 * isspace() in the "C" and "POSIX" locales.
6242 const char *spaces = " \f\n\r\t\v";
6244 *buf += strspn(*buf, spaces); /* Find start of token */
6246 return strcspn(*buf, spaces); /* Return token length */
6250 * Finds the next token in *buf, dynamically allocates a buffer big
6251 * enough to hold a copy of it, and copies the token into the new
6252 * buffer. The copy is guaranteed to be terminated with '\0'. Note
6253 * that a duplicate buffer is created even for a zero-length token.
6255 * Returns a pointer to the newly-allocated duplicate, or a null
6256 * pointer if memory for the duplicate was not available. If
6257 * the lenp argument is a non-null pointer, the length of the token
6258 * (not including the '\0') is returned in *lenp.
6260 * If successful, the *buf pointer will be updated to point beyond
6261 * the end of the found token.
6263 * Note: uses GFP_KERNEL for allocation.
6265 static inline char *dup_token(const char **buf, size_t *lenp)
6270 len = next_token(buf);
6271 dup = kmemdup(*buf, len + 1, GFP_KERNEL);
6274 *(dup + len) = '\0';
6283 static int rbd_parse_param(struct fs_parameter *param,
6284 struct rbd_parse_opts_ctx *pctx)
6286 struct rbd_options *opt = pctx->opts;
6287 struct fs_parse_result result;
6288 struct p_log log = {.prefix = "rbd"};
6291 ret = ceph_parse_param(param, pctx->copts, NULL);
6292 if (ret != -ENOPARAM)
6295 token = __fs_parse(&log, rbd_parameters, param, &result);
6296 dout("%s fs_parse '%s' token %d\n", __func__, param->key, token);
6298 if (token == -ENOPARAM)
6299 return inval_plog(&log, "Unknown parameter '%s'",
6305 case Opt_queue_depth:
6306 if (result.uint_32 < 1)
6308 opt->queue_depth = result.uint_32;
6310 case Opt_alloc_size:
6311 if (result.uint_32 < SECTOR_SIZE)
6313 if (!is_power_of_2(result.uint_32))
6314 return inval_plog(&log, "alloc_size must be a power of 2");
6315 opt->alloc_size = result.uint_32;
6317 case Opt_lock_timeout:
6318 /* 0 is "wait forever" (i.e. infinite timeout) */
6319 if (result.uint_32 > INT_MAX / 1000)
6321 opt->lock_timeout = msecs_to_jiffies(result.uint_32 * 1000);
6324 kfree(pctx->spec->pool_ns);
6325 pctx->spec->pool_ns = param->string;
6326 param->string = NULL;
6328 case Opt_compression_hint:
6329 switch (result.uint_32) {
6330 case Opt_compression_hint_none:
6331 opt->alloc_hint_flags &=
6332 ~(CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE |
6333 CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE);
6335 case Opt_compression_hint_compressible:
6336 opt->alloc_hint_flags |=
6337 CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6338 opt->alloc_hint_flags &=
6339 ~CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6341 case Opt_compression_hint_incompressible:
6342 opt->alloc_hint_flags |=
6343 CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6344 opt->alloc_hint_flags &=
6345 ~CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6352 opt->read_only = true;
6354 case Opt_read_write:
6355 opt->read_only = false;
6357 case Opt_lock_on_read:
6358 opt->lock_on_read = true;
6361 opt->exclusive = true;
6373 return inval_plog(&log, "%s out of range", param->key);
6377 * This duplicates most of generic_parse_monolithic(), untying it from
6378 * fs_context and skipping standard superblock and security options.
6380 static int rbd_parse_options(char *options, struct rbd_parse_opts_ctx *pctx)
6385 dout("%s '%s'\n", __func__, options);
6386 while ((key = strsep(&options, ",")) != NULL) {
6388 struct fs_parameter param = {
6390 .type = fs_value_is_flag,
6392 char *value = strchr(key, '=');
6399 v_len = strlen(value);
6400 param.string = kmemdup_nul(value, v_len,
6404 param.type = fs_value_is_string;
6408 ret = rbd_parse_param(¶m, pctx);
6409 kfree(param.string);
6419 * Parse the options provided for an "rbd add" (i.e., rbd image
6420 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
6421 * and the data written is passed here via a NUL-terminated buffer.
6422 * Returns 0 if successful or an error code otherwise.
6424 * The information extracted from these options is recorded in
6425 * the other parameters which return dynamically-allocated
6428 * The address of a pointer that will refer to a ceph options
6429 * structure. Caller must release the returned pointer using
6430 * ceph_destroy_options() when it is no longer needed.
6432 * Address of an rbd options pointer. Fully initialized by
6433 * this function; caller must release with kfree().
6435 * Address of an rbd image specification pointer. Fully
6436 * initialized by this function based on parsed options.
6437 * Caller must release with rbd_spec_put().
6439 * The options passed take this form:
6440 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
6443 * A comma-separated list of one or more monitor addresses.
6444 * A monitor address is an ip address, optionally followed
6445 * by a port number (separated by a colon).
6446 * I.e.: ip1[:port1][,ip2[:port2]...]
6448 * A comma-separated list of ceph and/or rbd options.
6450 * The name of the rados pool containing the rbd image.
6452 * The name of the image in that pool to map.
6454 * An optional snapshot id. If provided, the mapping will
6455 * present data from the image at the time that snapshot was
6456 * created. The image head is used if no snapshot id is
6457 * provided. Snapshot mappings are always read-only.
6459 static int rbd_add_parse_args(const char *buf,
6460 struct ceph_options **ceph_opts,
6461 struct rbd_options **opts,
6462 struct rbd_spec **rbd_spec)
6466 const char *mon_addrs;
6468 size_t mon_addrs_size;
6469 struct rbd_parse_opts_ctx pctx = { 0 };
6472 /* The first four tokens are required */
6474 len = next_token(&buf);
6476 rbd_warn(NULL, "no monitor address(es) provided");
6480 mon_addrs_size = len;
6484 options = dup_token(&buf, NULL);
6488 rbd_warn(NULL, "no options provided");
6492 pctx.spec = rbd_spec_alloc();
6496 pctx.spec->pool_name = dup_token(&buf, NULL);
6497 if (!pctx.spec->pool_name)
6499 if (!*pctx.spec->pool_name) {
6500 rbd_warn(NULL, "no pool name provided");
6504 pctx.spec->image_name = dup_token(&buf, NULL);
6505 if (!pctx.spec->image_name)
6507 if (!*pctx.spec->image_name) {
6508 rbd_warn(NULL, "no image name provided");
6513 * Snapshot name is optional; default is to use "-"
6514 * (indicating the head/no snapshot).
6516 len = next_token(&buf);
6518 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
6519 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
6520 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
6521 ret = -ENAMETOOLONG;
6524 snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
6527 *(snap_name + len) = '\0';
6528 pctx.spec->snap_name = snap_name;
6530 pctx.copts = ceph_alloc_options();
6534 /* Initialize all rbd options to the defaults */
6536 pctx.opts = kzalloc(sizeof(*pctx.opts), GFP_KERNEL);
6540 pctx.opts->read_only = RBD_READ_ONLY_DEFAULT;
6541 pctx.opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
6542 pctx.opts->alloc_size = RBD_ALLOC_SIZE_DEFAULT;
6543 pctx.opts->lock_timeout = RBD_LOCK_TIMEOUT_DEFAULT;
6544 pctx.opts->lock_on_read = RBD_LOCK_ON_READ_DEFAULT;
6545 pctx.opts->exclusive = RBD_EXCLUSIVE_DEFAULT;
6546 pctx.opts->trim = RBD_TRIM_DEFAULT;
6548 ret = ceph_parse_mon_ips(mon_addrs, mon_addrs_size, pctx.copts, NULL);
6552 ret = rbd_parse_options(options, &pctx);
6556 *ceph_opts = pctx.copts;
6558 *rbd_spec = pctx.spec;
6566 ceph_destroy_options(pctx.copts);
6567 rbd_spec_put(pctx.spec);
6572 static void rbd_dev_image_unlock(struct rbd_device *rbd_dev)
6574 down_write(&rbd_dev->lock_rwsem);
6575 if (__rbd_is_lock_owner(rbd_dev))
6576 __rbd_release_lock(rbd_dev);
6577 up_write(&rbd_dev->lock_rwsem);
6581 * If the wait is interrupted, an error is returned even if the lock
6582 * was successfully acquired. rbd_dev_image_unlock() will release it
6585 static int rbd_add_acquire_lock(struct rbd_device *rbd_dev)
6589 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK)) {
6590 if (!rbd_dev->opts->exclusive && !rbd_dev->opts->lock_on_read)
6593 rbd_warn(rbd_dev, "exclusive-lock feature is not enabled");
6597 if (rbd_is_ro(rbd_dev))
6600 rbd_assert(!rbd_is_lock_owner(rbd_dev));
6601 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
6602 ret = wait_for_completion_killable_timeout(&rbd_dev->acquire_wait,
6603 ceph_timeout_jiffies(rbd_dev->opts->lock_timeout));
6605 ret = rbd_dev->acquire_err;
6607 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
6613 rbd_warn(rbd_dev, "failed to acquire exclusive lock: %ld", ret);
6618 * The lock may have been released by now, unless automatic lock
6619 * transitions are disabled.
6621 rbd_assert(!rbd_dev->opts->exclusive || rbd_is_lock_owner(rbd_dev));
6626 * An rbd format 2 image has a unique identifier, distinct from the
6627 * name given to it by the user. Internally, that identifier is
6628 * what's used to specify the names of objects related to the image.
6630 * A special "rbd id" object is used to map an rbd image name to its
6631 * id. If that object doesn't exist, then there is no v2 rbd image
6632 * with the supplied name.
6634 * This function will record the given rbd_dev's image_id field if
6635 * it can be determined, and in that case will return 0. If any
6636 * errors occur a negative errno will be returned and the rbd_dev's
6637 * image_id field will be unchanged (and should be NULL).
6639 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
6643 CEPH_DEFINE_OID_ONSTACK(oid);
6648 * When probing a parent image, the image id is already
6649 * known (and the image name likely is not). There's no
6650 * need to fetch the image id again in this case. We
6651 * do still need to set the image format though.
6653 if (rbd_dev->spec->image_id) {
6654 rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
6660 * First, see if the format 2 image id file exists, and if
6661 * so, get the image's persistent id from it.
6663 ret = ceph_oid_aprintf(&oid, GFP_KERNEL, "%s%s", RBD_ID_PREFIX,
6664 rbd_dev->spec->image_name);
6668 dout("rbd id object name is %s\n", oid.name);
6670 /* Response will be an encoded string, which includes a length */
6671 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
6672 response = kzalloc(size, GFP_NOIO);
6678 /* If it doesn't exist we'll assume it's a format 1 image */
6680 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
6683 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6684 if (ret == -ENOENT) {
6685 image_id = kstrdup("", GFP_KERNEL);
6686 ret = image_id ? 0 : -ENOMEM;
6688 rbd_dev->image_format = 1;
6689 } else if (ret >= 0) {
6692 image_id = ceph_extract_encoded_string(&p, p + ret,
6694 ret = PTR_ERR_OR_ZERO(image_id);
6696 rbd_dev->image_format = 2;
6700 rbd_dev->spec->image_id = image_id;
6701 dout("image_id is %s\n", image_id);
6705 ceph_oid_destroy(&oid);
6710 * Undo whatever state changes are made by v1 or v2 header info
6713 static void rbd_dev_unprobe(struct rbd_device *rbd_dev)
6715 struct rbd_image_header *header;
6717 rbd_dev_parent_put(rbd_dev);
6718 rbd_object_map_free(rbd_dev);
6719 rbd_dev_mapping_clear(rbd_dev);
6721 /* Free dynamic fields from the header, then zero it out */
6723 header = &rbd_dev->header;
6724 ceph_put_snap_context(header->snapc);
6725 kfree(header->snap_sizes);
6726 kfree(header->snap_names);
6727 kfree(header->object_prefix);
6728 memset(header, 0, sizeof (*header));
6731 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev)
6735 ret = rbd_dev_v2_object_prefix(rbd_dev);
6740 * Get the and check features for the image. Currently the
6741 * features are assumed to never change.
6743 ret = rbd_dev_v2_features(rbd_dev);
6747 /* If the image supports fancy striping, get its parameters */
6749 if (rbd_dev->header.features & RBD_FEATURE_STRIPINGV2) {
6750 ret = rbd_dev_v2_striping_info(rbd_dev);
6755 if (rbd_dev->header.features & RBD_FEATURE_DATA_POOL) {
6756 ret = rbd_dev_v2_data_pool(rbd_dev);
6761 rbd_init_layout(rbd_dev);
6765 rbd_dev->header.features = 0;
6766 kfree(rbd_dev->header.object_prefix);
6767 rbd_dev->header.object_prefix = NULL;
6772 * @depth is rbd_dev_image_probe() -> rbd_dev_probe_parent() ->
6773 * rbd_dev_image_probe() recursion depth, which means it's also the
6774 * length of the already discovered part of the parent chain.
6776 static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
6778 struct rbd_device *parent = NULL;
6781 if (!rbd_dev->parent_spec)
6784 if (++depth > RBD_MAX_PARENT_CHAIN_LEN) {
6785 pr_info("parent chain is too long (%d)\n", depth);
6790 parent = __rbd_dev_create(rbd_dev->rbd_client, rbd_dev->parent_spec);
6797 * Images related by parent/child relationships always share
6798 * rbd_client and spec/parent_spec, so bump their refcounts.
6800 __rbd_get_client(rbd_dev->rbd_client);
6801 rbd_spec_get(rbd_dev->parent_spec);
6803 __set_bit(RBD_DEV_FLAG_READONLY, &parent->flags);
6805 ret = rbd_dev_image_probe(parent, depth);
6809 rbd_dev->parent = parent;
6810 atomic_set(&rbd_dev->parent_ref, 1);
6814 rbd_dev_unparent(rbd_dev);
6815 rbd_dev_destroy(parent);
6819 static void rbd_dev_device_release(struct rbd_device *rbd_dev)
6821 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6822 rbd_free_disk(rbd_dev);
6824 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6828 * rbd_dev->header_rwsem must be locked for write and will be unlocked
6831 static int rbd_dev_device_setup(struct rbd_device *rbd_dev)
6835 /* Record our major and minor device numbers. */
6837 if (!single_major) {
6838 ret = register_blkdev(0, rbd_dev->name);
6840 goto err_out_unlock;
6842 rbd_dev->major = ret;
6845 rbd_dev->major = rbd_major;
6846 rbd_dev->minor = rbd_dev_id_to_minor(rbd_dev->dev_id);
6849 /* Set up the blkdev mapping. */
6851 ret = rbd_init_disk(rbd_dev);
6853 goto err_out_blkdev;
6855 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
6856 set_disk_ro(rbd_dev->disk, rbd_is_ro(rbd_dev));
6858 ret = dev_set_name(&rbd_dev->dev, "%d", rbd_dev->dev_id);
6862 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6863 up_write(&rbd_dev->header_rwsem);
6867 rbd_free_disk(rbd_dev);
6870 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6872 up_write(&rbd_dev->header_rwsem);
6876 static int rbd_dev_header_name(struct rbd_device *rbd_dev)
6878 struct rbd_spec *spec = rbd_dev->spec;
6881 /* Record the header object name for this rbd image. */
6883 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6884 if (rbd_dev->image_format == 1)
6885 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6886 spec->image_name, RBD_SUFFIX);
6888 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6889 RBD_HEADER_PREFIX, spec->image_id);
6894 static void rbd_print_dne(struct rbd_device *rbd_dev, bool is_snap)
6897 pr_info("image %s/%s%s%s does not exist\n",
6898 rbd_dev->spec->pool_name,
6899 rbd_dev->spec->pool_ns ?: "",
6900 rbd_dev->spec->pool_ns ? "/" : "",
6901 rbd_dev->spec->image_name);
6903 pr_info("snap %s/%s%s%s@%s does not exist\n",
6904 rbd_dev->spec->pool_name,
6905 rbd_dev->spec->pool_ns ?: "",
6906 rbd_dev->spec->pool_ns ? "/" : "",
6907 rbd_dev->spec->image_name,
6908 rbd_dev->spec->snap_name);
6912 static void rbd_dev_image_release(struct rbd_device *rbd_dev)
6914 if (!rbd_is_ro(rbd_dev))
6915 rbd_unregister_watch(rbd_dev);
6917 rbd_dev_unprobe(rbd_dev);
6918 rbd_dev->image_format = 0;
6919 kfree(rbd_dev->spec->image_id);
6920 rbd_dev->spec->image_id = NULL;
6924 * Probe for the existence of the header object for the given rbd
6925 * device. If this image is the one being mapped (i.e., not a
6926 * parent), initiate a watch on its header object before using that
6927 * object to get detailed information about the rbd image.
6929 * On success, returns with header_rwsem held for write if called
6932 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
6934 bool need_watch = !rbd_is_ro(rbd_dev);
6938 * Get the id from the image id object. Unless there's an
6939 * error, rbd_dev->spec->image_id will be filled in with
6940 * a dynamically-allocated string, and rbd_dev->image_format
6941 * will be set to either 1 or 2.
6943 ret = rbd_dev_image_id(rbd_dev);
6947 ret = rbd_dev_header_name(rbd_dev);
6949 goto err_out_format;
6952 ret = rbd_register_watch(rbd_dev);
6955 rbd_print_dne(rbd_dev, false);
6956 goto err_out_format;
6961 down_write(&rbd_dev->header_rwsem);
6963 ret = rbd_dev_header_info(rbd_dev);
6965 if (ret == -ENOENT && !need_watch)
6966 rbd_print_dne(rbd_dev, false);
6971 * If this image is the one being mapped, we have pool name and
6972 * id, image name and id, and snap name - need to fill snap id.
6973 * Otherwise this is a parent image, identified by pool, image
6974 * and snap ids - need to fill in names for those ids.
6977 ret = rbd_spec_fill_snap_id(rbd_dev);
6979 ret = rbd_spec_fill_names(rbd_dev);
6982 rbd_print_dne(rbd_dev, true);
6986 ret = rbd_dev_mapping_set(rbd_dev);
6990 if (rbd_is_snap(rbd_dev) &&
6991 (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)) {
6992 ret = rbd_object_map_load(rbd_dev);
6997 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
6998 ret = rbd_dev_v2_parent_info(rbd_dev);
7003 ret = rbd_dev_probe_parent(rbd_dev, depth);
7007 dout("discovered format %u image, header name is %s\n",
7008 rbd_dev->image_format, rbd_dev->header_oid.name);
7013 up_write(&rbd_dev->header_rwsem);
7015 rbd_unregister_watch(rbd_dev);
7016 rbd_dev_unprobe(rbd_dev);
7018 rbd_dev->image_format = 0;
7019 kfree(rbd_dev->spec->image_id);
7020 rbd_dev->spec->image_id = NULL;
7024 static ssize_t do_rbd_add(struct bus_type *bus,
7028 struct rbd_device *rbd_dev = NULL;
7029 struct ceph_options *ceph_opts = NULL;
7030 struct rbd_options *rbd_opts = NULL;
7031 struct rbd_spec *spec = NULL;
7032 struct rbd_client *rbdc;
7035 if (!capable(CAP_SYS_ADMIN))
7038 if (!try_module_get(THIS_MODULE))
7041 /* parse add command */
7042 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
7046 rbdc = rbd_get_client(ceph_opts);
7053 rc = ceph_pg_poolid_by_name(rbdc->client->osdc.osdmap, spec->pool_name);
7056 pr_info("pool %s does not exist\n", spec->pool_name);
7057 goto err_out_client;
7059 spec->pool_id = (u64)rc;
7061 rbd_dev = rbd_dev_create(rbdc, spec, rbd_opts);
7064 goto err_out_client;
7066 rbdc = NULL; /* rbd_dev now owns this */
7067 spec = NULL; /* rbd_dev now owns this */
7068 rbd_opts = NULL; /* rbd_dev now owns this */
7070 /* if we are mapping a snapshot it will be a read-only mapping */
7071 if (rbd_dev->opts->read_only ||
7072 strcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME))
7073 __set_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
7075 rbd_dev->config_info = kstrdup(buf, GFP_KERNEL);
7076 if (!rbd_dev->config_info) {
7078 goto err_out_rbd_dev;
7081 rc = rbd_dev_image_probe(rbd_dev, 0);
7083 goto err_out_rbd_dev;
7085 if (rbd_dev->opts->alloc_size > rbd_dev->layout.object_size) {
7086 rbd_warn(rbd_dev, "alloc_size adjusted to %u",
7087 rbd_dev->layout.object_size);
7088 rbd_dev->opts->alloc_size = rbd_dev->layout.object_size;
7091 rc = rbd_dev_device_setup(rbd_dev);
7093 goto err_out_image_probe;
7095 rc = rbd_add_acquire_lock(rbd_dev);
7097 goto err_out_image_lock;
7099 /* Everything's ready. Announce the disk to the world. */
7101 rc = device_add(&rbd_dev->dev);
7103 goto err_out_image_lock;
7105 device_add_disk(&rbd_dev->dev, rbd_dev->disk, NULL);
7106 /* see rbd_init_disk() */
7107 blk_put_queue(rbd_dev->disk->queue);
7109 spin_lock(&rbd_dev_list_lock);
7110 list_add_tail(&rbd_dev->node, &rbd_dev_list);
7111 spin_unlock(&rbd_dev_list_lock);
7113 pr_info("%s: capacity %llu features 0x%llx\n", rbd_dev->disk->disk_name,
7114 (unsigned long long)get_capacity(rbd_dev->disk) << SECTOR_SHIFT,
7115 rbd_dev->header.features);
7118 module_put(THIS_MODULE);
7122 rbd_dev_image_unlock(rbd_dev);
7123 rbd_dev_device_release(rbd_dev);
7124 err_out_image_probe:
7125 rbd_dev_image_release(rbd_dev);
7127 rbd_dev_destroy(rbd_dev);
7129 rbd_put_client(rbdc);
7136 static ssize_t add_store(struct bus_type *bus, const char *buf, size_t count)
7141 return do_rbd_add(bus, buf, count);
7144 static ssize_t add_single_major_store(struct bus_type *bus, const char *buf,
7147 return do_rbd_add(bus, buf, count);
7150 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev)
7152 while (rbd_dev->parent) {
7153 struct rbd_device *first = rbd_dev;
7154 struct rbd_device *second = first->parent;
7155 struct rbd_device *third;
7158 * Follow to the parent with no grandparent and
7161 while (second && (third = second->parent)) {
7166 rbd_dev_image_release(second);
7167 rbd_dev_destroy(second);
7168 first->parent = NULL;
7169 first->parent_overlap = 0;
7171 rbd_assert(first->parent_spec);
7172 rbd_spec_put(first->parent_spec);
7173 first->parent_spec = NULL;
7177 static ssize_t do_rbd_remove(struct bus_type *bus,
7181 struct rbd_device *rbd_dev = NULL;
7182 struct list_head *tmp;
7188 if (!capable(CAP_SYS_ADMIN))
7193 sscanf(buf, "%d %5s", &dev_id, opt_buf);
7195 pr_err("dev_id out of range\n");
7198 if (opt_buf[0] != '\0') {
7199 if (!strcmp(opt_buf, "force")) {
7202 pr_err("bad remove option at '%s'\n", opt_buf);
7208 spin_lock(&rbd_dev_list_lock);
7209 list_for_each(tmp, &rbd_dev_list) {
7210 rbd_dev = list_entry(tmp, struct rbd_device, node);
7211 if (rbd_dev->dev_id == dev_id) {
7217 spin_lock_irq(&rbd_dev->lock);
7218 if (rbd_dev->open_count && !force)
7220 else if (test_and_set_bit(RBD_DEV_FLAG_REMOVING,
7223 spin_unlock_irq(&rbd_dev->lock);
7225 spin_unlock(&rbd_dev_list_lock);
7231 * Prevent new IO from being queued and wait for existing
7232 * IO to complete/fail.
7234 blk_mq_freeze_queue(rbd_dev->disk->queue);
7235 blk_set_queue_dying(rbd_dev->disk->queue);
7238 del_gendisk(rbd_dev->disk);
7239 spin_lock(&rbd_dev_list_lock);
7240 list_del_init(&rbd_dev->node);
7241 spin_unlock(&rbd_dev_list_lock);
7242 device_del(&rbd_dev->dev);
7244 rbd_dev_image_unlock(rbd_dev);
7245 rbd_dev_device_release(rbd_dev);
7246 rbd_dev_image_release(rbd_dev);
7247 rbd_dev_destroy(rbd_dev);
7251 static ssize_t remove_store(struct bus_type *bus, const char *buf, size_t count)
7256 return do_rbd_remove(bus, buf, count);
7259 static ssize_t remove_single_major_store(struct bus_type *bus, const char *buf,
7262 return do_rbd_remove(bus, buf, count);
7266 * create control files in sysfs
7269 static int __init rbd_sysfs_init(void)
7273 ret = device_register(&rbd_root_dev);
7277 ret = bus_register(&rbd_bus_type);
7279 device_unregister(&rbd_root_dev);
7284 static void __exit rbd_sysfs_cleanup(void)
7286 bus_unregister(&rbd_bus_type);
7287 device_unregister(&rbd_root_dev);
7290 static int __init rbd_slab_init(void)
7292 rbd_assert(!rbd_img_request_cache);
7293 rbd_img_request_cache = KMEM_CACHE(rbd_img_request, 0);
7294 if (!rbd_img_request_cache)
7297 rbd_assert(!rbd_obj_request_cache);
7298 rbd_obj_request_cache = KMEM_CACHE(rbd_obj_request, 0);
7299 if (!rbd_obj_request_cache)
7305 kmem_cache_destroy(rbd_img_request_cache);
7306 rbd_img_request_cache = NULL;
7310 static void rbd_slab_exit(void)
7312 rbd_assert(rbd_obj_request_cache);
7313 kmem_cache_destroy(rbd_obj_request_cache);
7314 rbd_obj_request_cache = NULL;
7316 rbd_assert(rbd_img_request_cache);
7317 kmem_cache_destroy(rbd_img_request_cache);
7318 rbd_img_request_cache = NULL;
7321 static int __init rbd_init(void)
7325 if (!libceph_compatible(NULL)) {
7326 rbd_warn(NULL, "libceph incompatibility (quitting)");
7330 rc = rbd_slab_init();
7335 * The number of active work items is limited by the number of
7336 * rbd devices * queue depth, so leave @max_active at default.
7338 rbd_wq = alloc_workqueue(RBD_DRV_NAME, WQ_MEM_RECLAIM, 0);
7345 rbd_major = register_blkdev(0, RBD_DRV_NAME);
7346 if (rbd_major < 0) {
7352 rc = rbd_sysfs_init();
7354 goto err_out_blkdev;
7357 pr_info("loaded (major %d)\n", rbd_major);
7359 pr_info("loaded\n");
7365 unregister_blkdev(rbd_major, RBD_DRV_NAME);
7367 destroy_workqueue(rbd_wq);
7373 static void __exit rbd_exit(void)
7375 ida_destroy(&rbd_dev_id_ida);
7376 rbd_sysfs_cleanup();
7378 unregister_blkdev(rbd_major, RBD_DRV_NAME);
7379 destroy_workqueue(rbd_wq);
7383 module_init(rbd_init);
7384 module_exit(rbd_exit);
7386 MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
7387 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
7388 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
7389 /* following authorship retained from original osdblk.c */
7390 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
7392 MODULE_DESCRIPTION("RADOS Block Device (RBD) driver");
7393 MODULE_LICENSE("GPL");