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 const struct block_device_operations rbd_bd_ops = {
696 .owner = THIS_MODULE,
698 .release = rbd_release,
702 * Initialize an rbd client instance. Success or not, this function
703 * consumes ceph_opts. Caller holds client_mutex.
705 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
707 struct rbd_client *rbdc;
710 dout("%s:\n", __func__);
711 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
715 kref_init(&rbdc->kref);
716 INIT_LIST_HEAD(&rbdc->node);
718 rbdc->client = ceph_create_client(ceph_opts, rbdc);
719 if (IS_ERR(rbdc->client))
721 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
723 ret = ceph_open_session(rbdc->client);
727 spin_lock(&rbd_client_list_lock);
728 list_add_tail(&rbdc->node, &rbd_client_list);
729 spin_unlock(&rbd_client_list_lock);
731 dout("%s: rbdc %p\n", __func__, rbdc);
735 ceph_destroy_client(rbdc->client);
740 ceph_destroy_options(ceph_opts);
741 dout("%s: error %d\n", __func__, ret);
746 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
748 kref_get(&rbdc->kref);
754 * Find a ceph client with specific addr and configuration. If
755 * found, bump its reference count.
757 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
759 struct rbd_client *client_node;
762 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
765 spin_lock(&rbd_client_list_lock);
766 list_for_each_entry(client_node, &rbd_client_list, node) {
767 if (!ceph_compare_options(ceph_opts, client_node->client)) {
768 __rbd_get_client(client_node);
774 spin_unlock(&rbd_client_list_lock);
776 return found ? client_node : NULL;
780 * (Per device) rbd map options
788 Opt_compression_hint,
789 /* string args above */
798 Opt_compression_hint_none,
799 Opt_compression_hint_compressible,
800 Opt_compression_hint_incompressible,
803 static const struct constant_table rbd_param_compression_hint[] = {
804 {"none", Opt_compression_hint_none},
805 {"compressible", Opt_compression_hint_compressible},
806 {"incompressible", Opt_compression_hint_incompressible},
810 static const struct fs_parameter_spec rbd_parameters[] = {
811 fsparam_u32 ("alloc_size", Opt_alloc_size),
812 fsparam_enum ("compression_hint", Opt_compression_hint,
813 rbd_param_compression_hint),
814 fsparam_flag ("exclusive", Opt_exclusive),
815 fsparam_flag ("lock_on_read", Opt_lock_on_read),
816 fsparam_u32 ("lock_timeout", Opt_lock_timeout),
817 fsparam_flag ("notrim", Opt_notrim),
818 fsparam_string ("_pool_ns", Opt_pool_ns),
819 fsparam_u32 ("queue_depth", Opt_queue_depth),
820 fsparam_flag ("read_only", Opt_read_only),
821 fsparam_flag ("read_write", Opt_read_write),
822 fsparam_flag ("ro", Opt_read_only),
823 fsparam_flag ("rw", Opt_read_write),
830 unsigned long lock_timeout;
836 u32 alloc_hint_flags; /* CEPH_OSD_OP_ALLOC_HINT_FLAG_* */
839 #define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_MAX_RQ
840 #define RBD_ALLOC_SIZE_DEFAULT (64 * 1024)
841 #define RBD_LOCK_TIMEOUT_DEFAULT 0 /* no timeout */
842 #define RBD_READ_ONLY_DEFAULT false
843 #define RBD_LOCK_ON_READ_DEFAULT false
844 #define RBD_EXCLUSIVE_DEFAULT false
845 #define RBD_TRIM_DEFAULT true
847 struct rbd_parse_opts_ctx {
848 struct rbd_spec *spec;
849 struct ceph_options *copts;
850 struct rbd_options *opts;
853 static char* obj_op_name(enum obj_operation_type op_type)
870 * Destroy ceph client
872 * Caller must hold rbd_client_list_lock.
874 static void rbd_client_release(struct kref *kref)
876 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
878 dout("%s: rbdc %p\n", __func__, rbdc);
879 spin_lock(&rbd_client_list_lock);
880 list_del(&rbdc->node);
881 spin_unlock(&rbd_client_list_lock);
883 ceph_destroy_client(rbdc->client);
888 * Drop reference to ceph client node. If it's not referenced anymore, release
891 static void rbd_put_client(struct rbd_client *rbdc)
894 kref_put(&rbdc->kref, rbd_client_release);
898 * Get a ceph client with specific addr and configuration, if one does
899 * not exist create it. Either way, ceph_opts is consumed by this
902 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
904 struct rbd_client *rbdc;
907 mutex_lock(&client_mutex);
908 rbdc = rbd_client_find(ceph_opts);
910 ceph_destroy_options(ceph_opts);
913 * Using an existing client. Make sure ->pg_pools is up to
914 * date before we look up the pool id in do_rbd_add().
916 ret = ceph_wait_for_latest_osdmap(rbdc->client,
917 rbdc->client->options->mount_timeout);
919 rbd_warn(NULL, "failed to get latest osdmap: %d", ret);
920 rbd_put_client(rbdc);
924 rbdc = rbd_client_create(ceph_opts);
926 mutex_unlock(&client_mutex);
931 static bool rbd_image_format_valid(u32 image_format)
933 return image_format == 1 || image_format == 2;
936 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
941 /* The header has to start with the magic rbd header text */
942 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
945 /* The bio layer requires at least sector-sized I/O */
947 if (ondisk->options.order < SECTOR_SHIFT)
950 /* If we use u64 in a few spots we may be able to loosen this */
952 if (ondisk->options.order > 8 * sizeof (int) - 1)
956 * The size of a snapshot header has to fit in a size_t, and
957 * that limits the number of snapshots.
959 snap_count = le32_to_cpu(ondisk->snap_count);
960 size = SIZE_MAX - sizeof (struct ceph_snap_context);
961 if (snap_count > size / sizeof (__le64))
965 * Not only that, but the size of the entire the snapshot
966 * header must also be representable in a size_t.
968 size -= snap_count * sizeof (__le64);
969 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
976 * returns the size of an object in the image
978 static u32 rbd_obj_bytes(struct rbd_image_header *header)
980 return 1U << header->obj_order;
983 static void rbd_init_layout(struct rbd_device *rbd_dev)
985 if (rbd_dev->header.stripe_unit == 0 ||
986 rbd_dev->header.stripe_count == 0) {
987 rbd_dev->header.stripe_unit = rbd_obj_bytes(&rbd_dev->header);
988 rbd_dev->header.stripe_count = 1;
991 rbd_dev->layout.stripe_unit = rbd_dev->header.stripe_unit;
992 rbd_dev->layout.stripe_count = rbd_dev->header.stripe_count;
993 rbd_dev->layout.object_size = rbd_obj_bytes(&rbd_dev->header);
994 rbd_dev->layout.pool_id = rbd_dev->header.data_pool_id == CEPH_NOPOOL ?
995 rbd_dev->spec->pool_id : rbd_dev->header.data_pool_id;
996 RCU_INIT_POINTER(rbd_dev->layout.pool_ns, NULL);
1000 * Fill an rbd image header with information from the given format 1
1003 static int rbd_header_from_disk(struct rbd_device *rbd_dev,
1004 struct rbd_image_header_ondisk *ondisk)
1006 struct rbd_image_header *header = &rbd_dev->header;
1007 bool first_time = header->object_prefix == NULL;
1008 struct ceph_snap_context *snapc;
1009 char *object_prefix = NULL;
1010 char *snap_names = NULL;
1011 u64 *snap_sizes = NULL;
1016 /* Allocate this now to avoid having to handle failure below */
1019 object_prefix = kstrndup(ondisk->object_prefix,
1020 sizeof(ondisk->object_prefix),
1026 /* Allocate the snapshot context and fill it in */
1028 snap_count = le32_to_cpu(ondisk->snap_count);
1029 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
1032 snapc->seq = le64_to_cpu(ondisk->snap_seq);
1034 struct rbd_image_snap_ondisk *snaps;
1035 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
1037 /* We'll keep a copy of the snapshot names... */
1039 if (snap_names_len > (u64)SIZE_MAX)
1041 snap_names = kmalloc(snap_names_len, GFP_KERNEL);
1045 /* ...as well as the array of their sizes. */
1046 snap_sizes = kmalloc_array(snap_count,
1047 sizeof(*header->snap_sizes),
1053 * Copy the names, and fill in each snapshot's id
1056 * Note that rbd_dev_v1_header_info() guarantees the
1057 * ondisk buffer we're working with has
1058 * snap_names_len bytes beyond the end of the
1059 * snapshot id array, this memcpy() is safe.
1061 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
1062 snaps = ondisk->snaps;
1063 for (i = 0; i < snap_count; i++) {
1064 snapc->snaps[i] = le64_to_cpu(snaps[i].id);
1065 snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
1069 /* We won't fail any more, fill in the header */
1072 header->object_prefix = object_prefix;
1073 header->obj_order = ondisk->options.order;
1074 rbd_init_layout(rbd_dev);
1076 ceph_put_snap_context(header->snapc);
1077 kfree(header->snap_names);
1078 kfree(header->snap_sizes);
1081 /* The remaining fields always get updated (when we refresh) */
1083 header->image_size = le64_to_cpu(ondisk->image_size);
1084 header->snapc = snapc;
1085 header->snap_names = snap_names;
1086 header->snap_sizes = snap_sizes;
1094 ceph_put_snap_context(snapc);
1095 kfree(object_prefix);
1100 static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1102 const char *snap_name;
1104 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1106 /* Skip over names until we find the one we are looking for */
1108 snap_name = rbd_dev->header.snap_names;
1110 snap_name += strlen(snap_name) + 1;
1112 return kstrdup(snap_name, GFP_KERNEL);
1116 * Snapshot id comparison function for use with qsort()/bsearch().
1117 * Note that result is for snapshots in *descending* order.
1119 static int snapid_compare_reverse(const void *s1, const void *s2)
1121 u64 snap_id1 = *(u64 *)s1;
1122 u64 snap_id2 = *(u64 *)s2;
1124 if (snap_id1 < snap_id2)
1126 return snap_id1 == snap_id2 ? 0 : -1;
1130 * Search a snapshot context to see if the given snapshot id is
1133 * Returns the position of the snapshot id in the array if it's found,
1134 * or BAD_SNAP_INDEX otherwise.
1136 * Note: The snapshot array is in kept sorted (by the osd) in
1137 * reverse order, highest snapshot id first.
1139 static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1141 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1144 found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1145 sizeof (snap_id), snapid_compare_reverse);
1147 return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1150 static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1154 const char *snap_name;
1156 which = rbd_dev_snap_index(rbd_dev, snap_id);
1157 if (which == BAD_SNAP_INDEX)
1158 return ERR_PTR(-ENOENT);
1160 snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1161 return snap_name ? snap_name : ERR_PTR(-ENOMEM);
1164 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1166 if (snap_id == CEPH_NOSNAP)
1167 return RBD_SNAP_HEAD_NAME;
1169 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1170 if (rbd_dev->image_format == 1)
1171 return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1173 return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1176 static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1179 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1180 if (snap_id == CEPH_NOSNAP) {
1181 *snap_size = rbd_dev->header.image_size;
1182 } else if (rbd_dev->image_format == 1) {
1185 which = rbd_dev_snap_index(rbd_dev, snap_id);
1186 if (which == BAD_SNAP_INDEX)
1189 *snap_size = rbd_dev->header.snap_sizes[which];
1194 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1203 static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1205 u64 snap_id = rbd_dev->spec->snap_id;
1209 ret = rbd_snap_size(rbd_dev, snap_id, &size);
1213 rbd_dev->mapping.size = size;
1217 static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1219 rbd_dev->mapping.size = 0;
1222 static void zero_bvec(struct bio_vec *bv)
1225 unsigned long flags;
1227 buf = bvec_kmap_irq(bv, &flags);
1228 memset(buf, 0, bv->bv_len);
1229 flush_dcache_page(bv->bv_page);
1230 bvec_kunmap_irq(buf, &flags);
1233 static void zero_bios(struct ceph_bio_iter *bio_pos, u32 off, u32 bytes)
1235 struct ceph_bio_iter it = *bio_pos;
1237 ceph_bio_iter_advance(&it, off);
1238 ceph_bio_iter_advance_step(&it, bytes, ({
1243 static void zero_bvecs(struct ceph_bvec_iter *bvec_pos, u32 off, u32 bytes)
1245 struct ceph_bvec_iter it = *bvec_pos;
1247 ceph_bvec_iter_advance(&it, off);
1248 ceph_bvec_iter_advance_step(&it, bytes, ({
1254 * Zero a range in @obj_req data buffer defined by a bio (list) or
1255 * (private) bio_vec array.
1257 * @off is relative to the start of the data buffer.
1259 static void rbd_obj_zero_range(struct rbd_obj_request *obj_req, u32 off,
1262 dout("%s %p data buf %u~%u\n", __func__, obj_req, off, bytes);
1264 switch (obj_req->img_request->data_type) {
1265 case OBJ_REQUEST_BIO:
1266 zero_bios(&obj_req->bio_pos, off, bytes);
1268 case OBJ_REQUEST_BVECS:
1269 case OBJ_REQUEST_OWN_BVECS:
1270 zero_bvecs(&obj_req->bvec_pos, off, bytes);
1277 static void rbd_obj_request_destroy(struct kref *kref);
1278 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1280 rbd_assert(obj_request != NULL);
1281 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1282 kref_read(&obj_request->kref));
1283 kref_put(&obj_request->kref, rbd_obj_request_destroy);
1286 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1287 struct rbd_obj_request *obj_request)
1289 rbd_assert(obj_request->img_request == NULL);
1291 /* Image request now owns object's original reference */
1292 obj_request->img_request = img_request;
1293 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1296 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1297 struct rbd_obj_request *obj_request)
1299 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1300 list_del(&obj_request->ex.oe_item);
1301 rbd_assert(obj_request->img_request == img_request);
1302 rbd_obj_request_put(obj_request);
1305 static void rbd_osd_submit(struct ceph_osd_request *osd_req)
1307 struct rbd_obj_request *obj_req = osd_req->r_priv;
1309 dout("%s osd_req %p for obj_req %p objno %llu %llu~%llu\n",
1310 __func__, osd_req, obj_req, obj_req->ex.oe_objno,
1311 obj_req->ex.oe_off, obj_req->ex.oe_len);
1312 ceph_osdc_start_request(osd_req->r_osdc, osd_req, false);
1316 * The default/initial value for all image request flags is 0. Each
1317 * is conditionally set to 1 at image request initialization time
1318 * and currently never change thereafter.
1320 static void img_request_layered_set(struct rbd_img_request *img_request)
1322 set_bit(IMG_REQ_LAYERED, &img_request->flags);
1325 static bool img_request_layered_test(struct rbd_img_request *img_request)
1327 return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1330 static bool rbd_obj_is_entire(struct rbd_obj_request *obj_req)
1332 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1334 return !obj_req->ex.oe_off &&
1335 obj_req->ex.oe_len == rbd_dev->layout.object_size;
1338 static bool rbd_obj_is_tail(struct rbd_obj_request *obj_req)
1340 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1342 return obj_req->ex.oe_off + obj_req->ex.oe_len ==
1343 rbd_dev->layout.object_size;
1347 * Must be called after rbd_obj_calc_img_extents().
1349 static bool rbd_obj_copyup_enabled(struct rbd_obj_request *obj_req)
1351 if (!obj_req->num_img_extents ||
1352 (rbd_obj_is_entire(obj_req) &&
1353 !obj_req->img_request->snapc->num_snaps))
1359 static u64 rbd_obj_img_extents_bytes(struct rbd_obj_request *obj_req)
1361 return ceph_file_extents_bytes(obj_req->img_extents,
1362 obj_req->num_img_extents);
1365 static bool rbd_img_is_write(struct rbd_img_request *img_req)
1367 switch (img_req->op_type) {
1371 case OBJ_OP_DISCARD:
1372 case OBJ_OP_ZEROOUT:
1379 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
1381 struct rbd_obj_request *obj_req = osd_req->r_priv;
1384 dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
1385 osd_req->r_result, obj_req);
1388 * Writes aren't allowed to return a data payload. In some
1389 * guarded write cases (e.g. stat + zero on an empty object)
1390 * a stat response makes it through, but we don't care.
1392 if (osd_req->r_result > 0 && rbd_img_is_write(obj_req->img_request))
1395 result = osd_req->r_result;
1397 rbd_obj_handle_request(obj_req, result);
1400 static void rbd_osd_format_read(struct ceph_osd_request *osd_req)
1402 struct rbd_obj_request *obj_request = osd_req->r_priv;
1403 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1404 struct ceph_options *opt = rbd_dev->rbd_client->client->options;
1406 osd_req->r_flags = CEPH_OSD_FLAG_READ | opt->read_from_replica;
1407 osd_req->r_snapid = obj_request->img_request->snap_id;
1410 static void rbd_osd_format_write(struct ceph_osd_request *osd_req)
1412 struct rbd_obj_request *obj_request = osd_req->r_priv;
1414 osd_req->r_flags = CEPH_OSD_FLAG_WRITE;
1415 ktime_get_real_ts64(&osd_req->r_mtime);
1416 osd_req->r_data_offset = obj_request->ex.oe_off;
1419 static struct ceph_osd_request *
1420 __rbd_obj_add_osd_request(struct rbd_obj_request *obj_req,
1421 struct ceph_snap_context *snapc, int num_ops)
1423 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1424 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1425 struct ceph_osd_request *req;
1426 const char *name_format = rbd_dev->image_format == 1 ?
1427 RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
1430 req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, GFP_NOIO);
1432 return ERR_PTR(-ENOMEM);
1434 list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
1435 req->r_callback = rbd_osd_req_callback;
1436 req->r_priv = obj_req;
1439 * Data objects may be stored in a separate pool, but always in
1440 * the same namespace in that pool as the header in its pool.
1442 ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
1443 req->r_base_oloc.pool = rbd_dev->layout.pool_id;
1445 ret = ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
1446 rbd_dev->header.object_prefix,
1447 obj_req->ex.oe_objno);
1449 return ERR_PTR(ret);
1454 static struct ceph_osd_request *
1455 rbd_obj_add_osd_request(struct rbd_obj_request *obj_req, int num_ops)
1457 return __rbd_obj_add_osd_request(obj_req, obj_req->img_request->snapc,
1461 static struct rbd_obj_request *rbd_obj_request_create(void)
1463 struct rbd_obj_request *obj_request;
1465 obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_NOIO);
1469 ceph_object_extent_init(&obj_request->ex);
1470 INIT_LIST_HEAD(&obj_request->osd_reqs);
1471 mutex_init(&obj_request->state_mutex);
1472 kref_init(&obj_request->kref);
1474 dout("%s %p\n", __func__, obj_request);
1478 static void rbd_obj_request_destroy(struct kref *kref)
1480 struct rbd_obj_request *obj_request;
1481 struct ceph_osd_request *osd_req;
1484 obj_request = container_of(kref, struct rbd_obj_request, kref);
1486 dout("%s: obj %p\n", __func__, obj_request);
1488 while (!list_empty(&obj_request->osd_reqs)) {
1489 osd_req = list_first_entry(&obj_request->osd_reqs,
1490 struct ceph_osd_request, r_private_item);
1491 list_del_init(&osd_req->r_private_item);
1492 ceph_osdc_put_request(osd_req);
1495 switch (obj_request->img_request->data_type) {
1496 case OBJ_REQUEST_NODATA:
1497 case OBJ_REQUEST_BIO:
1498 case OBJ_REQUEST_BVECS:
1499 break; /* Nothing to do */
1500 case OBJ_REQUEST_OWN_BVECS:
1501 kfree(obj_request->bvec_pos.bvecs);
1507 kfree(obj_request->img_extents);
1508 if (obj_request->copyup_bvecs) {
1509 for (i = 0; i < obj_request->copyup_bvec_count; i++) {
1510 if (obj_request->copyup_bvecs[i].bv_page)
1511 __free_page(obj_request->copyup_bvecs[i].bv_page);
1513 kfree(obj_request->copyup_bvecs);
1516 kmem_cache_free(rbd_obj_request_cache, obj_request);
1519 /* It's OK to call this for a device with no parent */
1521 static void rbd_spec_put(struct rbd_spec *spec);
1522 static void rbd_dev_unparent(struct rbd_device *rbd_dev)
1524 rbd_dev_remove_parent(rbd_dev);
1525 rbd_spec_put(rbd_dev->parent_spec);
1526 rbd_dev->parent_spec = NULL;
1527 rbd_dev->parent_overlap = 0;
1531 * Parent image reference counting is used to determine when an
1532 * image's parent fields can be safely torn down--after there are no
1533 * more in-flight requests to the parent image. When the last
1534 * reference is dropped, cleaning them up is safe.
1536 static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
1540 if (!rbd_dev->parent_spec)
1543 counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
1547 /* Last reference; clean up parent data structures */
1550 rbd_dev_unparent(rbd_dev);
1552 rbd_warn(rbd_dev, "parent reference underflow");
1556 * If an image has a non-zero parent overlap, get a reference to its
1559 * Returns true if the rbd device has a parent with a non-zero
1560 * overlap and a reference for it was successfully taken, or
1563 static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
1567 if (!rbd_dev->parent_spec)
1570 if (rbd_dev->parent_overlap)
1571 counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
1574 rbd_warn(rbd_dev, "parent reference overflow");
1579 static void rbd_img_request_init(struct rbd_img_request *img_request,
1580 struct rbd_device *rbd_dev,
1581 enum obj_operation_type op_type)
1583 memset(img_request, 0, sizeof(*img_request));
1585 img_request->rbd_dev = rbd_dev;
1586 img_request->op_type = op_type;
1588 INIT_LIST_HEAD(&img_request->lock_item);
1589 INIT_LIST_HEAD(&img_request->object_extents);
1590 mutex_init(&img_request->state_mutex);
1593 static void rbd_img_capture_header(struct rbd_img_request *img_req)
1595 struct rbd_device *rbd_dev = img_req->rbd_dev;
1597 lockdep_assert_held(&rbd_dev->header_rwsem);
1599 if (rbd_img_is_write(img_req))
1600 img_req->snapc = ceph_get_snap_context(rbd_dev->header.snapc);
1602 img_req->snap_id = rbd_dev->spec->snap_id;
1604 if (rbd_dev_parent_get(rbd_dev))
1605 img_request_layered_set(img_req);
1608 static void rbd_img_request_destroy(struct rbd_img_request *img_request)
1610 struct rbd_obj_request *obj_request;
1611 struct rbd_obj_request *next_obj_request;
1613 dout("%s: img %p\n", __func__, img_request);
1615 WARN_ON(!list_empty(&img_request->lock_item));
1616 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1617 rbd_img_obj_request_del(img_request, obj_request);
1619 if (img_request_layered_test(img_request))
1620 rbd_dev_parent_put(img_request->rbd_dev);
1622 if (rbd_img_is_write(img_request))
1623 ceph_put_snap_context(img_request->snapc);
1625 if (test_bit(IMG_REQ_CHILD, &img_request->flags))
1626 kmem_cache_free(rbd_img_request_cache, img_request);
1629 #define BITS_PER_OBJ 2
1630 #define OBJS_PER_BYTE (BITS_PER_BYTE / BITS_PER_OBJ)
1631 #define OBJ_MASK ((1 << BITS_PER_OBJ) - 1)
1633 static void __rbd_object_map_index(struct rbd_device *rbd_dev, u64 objno,
1634 u64 *index, u8 *shift)
1638 rbd_assert(objno < rbd_dev->object_map_size);
1639 *index = div_u64_rem(objno, OBJS_PER_BYTE, &off);
1640 *shift = (OBJS_PER_BYTE - off - 1) * BITS_PER_OBJ;
1643 static u8 __rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1648 lockdep_assert_held(&rbd_dev->object_map_lock);
1649 __rbd_object_map_index(rbd_dev, objno, &index, &shift);
1650 return (rbd_dev->object_map[index] >> shift) & OBJ_MASK;
1653 static void __rbd_object_map_set(struct rbd_device *rbd_dev, u64 objno, u8 val)
1659 lockdep_assert_held(&rbd_dev->object_map_lock);
1660 rbd_assert(!(val & ~OBJ_MASK));
1662 __rbd_object_map_index(rbd_dev, objno, &index, &shift);
1663 p = &rbd_dev->object_map[index];
1664 *p = (*p & ~(OBJ_MASK << shift)) | (val << shift);
1667 static u8 rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1671 spin_lock(&rbd_dev->object_map_lock);
1672 state = __rbd_object_map_get(rbd_dev, objno);
1673 spin_unlock(&rbd_dev->object_map_lock);
1677 static bool use_object_map(struct rbd_device *rbd_dev)
1680 * An image mapped read-only can't use the object map -- it isn't
1681 * loaded because the header lock isn't acquired. Someone else can
1682 * write to the image and update the object map behind our back.
1684 * A snapshot can't be written to, so using the object map is always
1687 if (!rbd_is_snap(rbd_dev) && rbd_is_ro(rbd_dev))
1690 return ((rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) &&
1691 !(rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID));
1694 static bool rbd_object_map_may_exist(struct rbd_device *rbd_dev, u64 objno)
1698 /* fall back to default logic if object map is disabled or invalid */
1699 if (!use_object_map(rbd_dev))
1702 state = rbd_object_map_get(rbd_dev, objno);
1703 return state != OBJECT_NONEXISTENT;
1706 static void rbd_object_map_name(struct rbd_device *rbd_dev, u64 snap_id,
1707 struct ceph_object_id *oid)
1709 if (snap_id == CEPH_NOSNAP)
1710 ceph_oid_printf(oid, "%s%s", RBD_OBJECT_MAP_PREFIX,
1711 rbd_dev->spec->image_id);
1713 ceph_oid_printf(oid, "%s%s.%016llx", RBD_OBJECT_MAP_PREFIX,
1714 rbd_dev->spec->image_id, snap_id);
1717 static int rbd_object_map_lock(struct rbd_device *rbd_dev)
1719 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1720 CEPH_DEFINE_OID_ONSTACK(oid);
1723 struct ceph_locker *lockers;
1725 bool broke_lock = false;
1728 rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1731 ret = ceph_cls_lock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1732 CEPH_CLS_LOCK_EXCLUSIVE, "", "", "", 0);
1733 if (ret != -EBUSY || broke_lock) {
1735 ret = 0; /* already locked by myself */
1737 rbd_warn(rbd_dev, "failed to lock object map: %d", ret);
1741 ret = ceph_cls_lock_info(osdc, &oid, &rbd_dev->header_oloc,
1742 RBD_LOCK_NAME, &lock_type, &lock_tag,
1743 &lockers, &num_lockers);
1748 rbd_warn(rbd_dev, "failed to get object map lockers: %d", ret);
1753 if (num_lockers == 0)
1756 rbd_warn(rbd_dev, "breaking object map lock owned by %s%llu",
1757 ENTITY_NAME(lockers[0].id.name));
1759 ret = ceph_cls_break_lock(osdc, &oid, &rbd_dev->header_oloc,
1760 RBD_LOCK_NAME, lockers[0].id.cookie,
1761 &lockers[0].id.name);
1762 ceph_free_lockers(lockers, num_lockers);
1767 rbd_warn(rbd_dev, "failed to break object map lock: %d", ret);
1775 static void rbd_object_map_unlock(struct rbd_device *rbd_dev)
1777 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1778 CEPH_DEFINE_OID_ONSTACK(oid);
1781 rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1783 ret = ceph_cls_unlock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1785 if (ret && ret != -ENOENT)
1786 rbd_warn(rbd_dev, "failed to unlock object map: %d", ret);
1789 static int decode_object_map_header(void **p, void *end, u64 *object_map_size)
1797 ceph_decode_32_safe(p, end, header_len, e_inval);
1798 header_end = *p + header_len;
1800 ret = ceph_start_decoding(p, end, 1, "BitVector header", &struct_v,
1805 ceph_decode_64_safe(p, end, *object_map_size, e_inval);
1814 static int __rbd_object_map_load(struct rbd_device *rbd_dev)
1816 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1817 CEPH_DEFINE_OID_ONSTACK(oid);
1818 struct page **pages;
1822 u64 object_map_bytes;
1823 u64 object_map_size;
1827 rbd_assert(!rbd_dev->object_map && !rbd_dev->object_map_size);
1829 num_objects = ceph_get_num_objects(&rbd_dev->layout,
1830 rbd_dev->mapping.size);
1831 object_map_bytes = DIV_ROUND_UP_ULL(num_objects * BITS_PER_OBJ,
1833 num_pages = calc_pages_for(0, object_map_bytes) + 1;
1834 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1836 return PTR_ERR(pages);
1838 reply_len = num_pages * PAGE_SIZE;
1839 rbd_object_map_name(rbd_dev, rbd_dev->spec->snap_id, &oid);
1840 ret = ceph_osdc_call(osdc, &oid, &rbd_dev->header_oloc,
1841 "rbd", "object_map_load", CEPH_OSD_FLAG_READ,
1842 NULL, 0, pages, &reply_len);
1846 p = page_address(pages[0]);
1847 end = p + min(reply_len, (size_t)PAGE_SIZE);
1848 ret = decode_object_map_header(&p, end, &object_map_size);
1852 if (object_map_size != num_objects) {
1853 rbd_warn(rbd_dev, "object map size mismatch: %llu vs %llu",
1854 object_map_size, num_objects);
1859 if (offset_in_page(p) + object_map_bytes > reply_len) {
1864 rbd_dev->object_map = kvmalloc(object_map_bytes, GFP_KERNEL);
1865 if (!rbd_dev->object_map) {
1870 rbd_dev->object_map_size = object_map_size;
1871 ceph_copy_from_page_vector(pages, rbd_dev->object_map,
1872 offset_in_page(p), object_map_bytes);
1875 ceph_release_page_vector(pages, num_pages);
1879 static void rbd_object_map_free(struct rbd_device *rbd_dev)
1881 kvfree(rbd_dev->object_map);
1882 rbd_dev->object_map = NULL;
1883 rbd_dev->object_map_size = 0;
1886 static int rbd_object_map_load(struct rbd_device *rbd_dev)
1890 ret = __rbd_object_map_load(rbd_dev);
1894 ret = rbd_dev_v2_get_flags(rbd_dev);
1896 rbd_object_map_free(rbd_dev);
1900 if (rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID)
1901 rbd_warn(rbd_dev, "object map is invalid");
1906 static int rbd_object_map_open(struct rbd_device *rbd_dev)
1910 ret = rbd_object_map_lock(rbd_dev);
1914 ret = rbd_object_map_load(rbd_dev);
1916 rbd_object_map_unlock(rbd_dev);
1923 static void rbd_object_map_close(struct rbd_device *rbd_dev)
1925 rbd_object_map_free(rbd_dev);
1926 rbd_object_map_unlock(rbd_dev);
1930 * This function needs snap_id (or more precisely just something to
1931 * distinguish between HEAD and snapshot object maps), new_state and
1932 * current_state that were passed to rbd_object_map_update().
1934 * To avoid allocating and stashing a context we piggyback on the OSD
1935 * request. A HEAD update has two ops (assert_locked). For new_state
1936 * and current_state we decode our own object_map_update op, encoded in
1937 * rbd_cls_object_map_update().
1939 static int rbd_object_map_update_finish(struct rbd_obj_request *obj_req,
1940 struct ceph_osd_request *osd_req)
1942 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1943 struct ceph_osd_data *osd_data;
1945 u8 state, new_state, current_state;
1946 bool has_current_state;
1949 if (osd_req->r_result)
1950 return osd_req->r_result;
1953 * Nothing to do for a snapshot object map.
1955 if (osd_req->r_num_ops == 1)
1959 * Update in-memory HEAD object map.
1961 rbd_assert(osd_req->r_num_ops == 2);
1962 osd_data = osd_req_op_data(osd_req, 1, cls, request_data);
1963 rbd_assert(osd_data->type == CEPH_OSD_DATA_TYPE_PAGES);
1965 p = page_address(osd_data->pages[0]);
1966 objno = ceph_decode_64(&p);
1967 rbd_assert(objno == obj_req->ex.oe_objno);
1968 rbd_assert(ceph_decode_64(&p) == objno + 1);
1969 new_state = ceph_decode_8(&p);
1970 has_current_state = ceph_decode_8(&p);
1971 if (has_current_state)
1972 current_state = ceph_decode_8(&p);
1974 spin_lock(&rbd_dev->object_map_lock);
1975 state = __rbd_object_map_get(rbd_dev, objno);
1976 if (!has_current_state || current_state == state ||
1977 (current_state == OBJECT_EXISTS && state == OBJECT_EXISTS_CLEAN))
1978 __rbd_object_map_set(rbd_dev, objno, new_state);
1979 spin_unlock(&rbd_dev->object_map_lock);
1984 static void rbd_object_map_callback(struct ceph_osd_request *osd_req)
1986 struct rbd_obj_request *obj_req = osd_req->r_priv;
1989 dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
1990 osd_req->r_result, obj_req);
1992 result = rbd_object_map_update_finish(obj_req, osd_req);
1993 rbd_obj_handle_request(obj_req, result);
1996 static bool update_needed(struct rbd_device *rbd_dev, u64 objno, u8 new_state)
1998 u8 state = rbd_object_map_get(rbd_dev, objno);
2000 if (state == new_state ||
2001 (new_state == OBJECT_PENDING && state == OBJECT_NONEXISTENT) ||
2002 (new_state == OBJECT_NONEXISTENT && state != OBJECT_PENDING))
2008 static int rbd_cls_object_map_update(struct ceph_osd_request *req,
2009 int which, u64 objno, u8 new_state,
2010 const u8 *current_state)
2012 struct page **pages;
2016 ret = osd_req_op_cls_init(req, which, "rbd", "object_map_update");
2020 pages = ceph_alloc_page_vector(1, GFP_NOIO);
2022 return PTR_ERR(pages);
2024 p = start = page_address(pages[0]);
2025 ceph_encode_64(&p, objno);
2026 ceph_encode_64(&p, objno + 1);
2027 ceph_encode_8(&p, new_state);
2028 if (current_state) {
2029 ceph_encode_8(&p, 1);
2030 ceph_encode_8(&p, *current_state);
2032 ceph_encode_8(&p, 0);
2035 osd_req_op_cls_request_data_pages(req, which, pages, p - start, 0,
2042 * 0 - object map update sent
2043 * 1 - object map update isn't needed
2046 static int rbd_object_map_update(struct rbd_obj_request *obj_req, u64 snap_id,
2047 u8 new_state, const u8 *current_state)
2049 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2050 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2051 struct ceph_osd_request *req;
2056 if (snap_id == CEPH_NOSNAP) {
2057 if (!update_needed(rbd_dev, obj_req->ex.oe_objno, new_state))
2060 num_ops++; /* assert_locked */
2063 req = ceph_osdc_alloc_request(osdc, NULL, num_ops, false, GFP_NOIO);
2067 list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
2068 req->r_callback = rbd_object_map_callback;
2069 req->r_priv = obj_req;
2071 rbd_object_map_name(rbd_dev, snap_id, &req->r_base_oid);
2072 ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
2073 req->r_flags = CEPH_OSD_FLAG_WRITE;
2074 ktime_get_real_ts64(&req->r_mtime);
2076 if (snap_id == CEPH_NOSNAP) {
2078 * Protect against possible race conditions during lock
2079 * ownership transitions.
2081 ret = ceph_cls_assert_locked(req, which++, RBD_LOCK_NAME,
2082 CEPH_CLS_LOCK_EXCLUSIVE, "", "");
2087 ret = rbd_cls_object_map_update(req, which, obj_req->ex.oe_objno,
2088 new_state, current_state);
2092 ret = ceph_osdc_alloc_messages(req, GFP_NOIO);
2096 ceph_osdc_start_request(osdc, req, false);
2100 static void prune_extents(struct ceph_file_extent *img_extents,
2101 u32 *num_img_extents, u64 overlap)
2103 u32 cnt = *num_img_extents;
2105 /* drop extents completely beyond the overlap */
2106 while (cnt && img_extents[cnt - 1].fe_off >= overlap)
2110 struct ceph_file_extent *ex = &img_extents[cnt - 1];
2112 /* trim final overlapping extent */
2113 if (ex->fe_off + ex->fe_len > overlap)
2114 ex->fe_len = overlap - ex->fe_off;
2117 *num_img_extents = cnt;
2121 * Determine the byte range(s) covered by either just the object extent
2122 * or the entire object in the parent image.
2124 static int rbd_obj_calc_img_extents(struct rbd_obj_request *obj_req,
2127 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2130 if (!rbd_dev->parent_overlap)
2133 ret = ceph_extent_to_file(&rbd_dev->layout, obj_req->ex.oe_objno,
2134 entire ? 0 : obj_req->ex.oe_off,
2135 entire ? rbd_dev->layout.object_size :
2137 &obj_req->img_extents,
2138 &obj_req->num_img_extents);
2142 prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
2143 rbd_dev->parent_overlap);
2147 static void rbd_osd_setup_data(struct ceph_osd_request *osd_req, int which)
2149 struct rbd_obj_request *obj_req = osd_req->r_priv;
2151 switch (obj_req->img_request->data_type) {
2152 case OBJ_REQUEST_BIO:
2153 osd_req_op_extent_osd_data_bio(osd_req, which,
2155 obj_req->ex.oe_len);
2157 case OBJ_REQUEST_BVECS:
2158 case OBJ_REQUEST_OWN_BVECS:
2159 rbd_assert(obj_req->bvec_pos.iter.bi_size ==
2160 obj_req->ex.oe_len);
2161 rbd_assert(obj_req->bvec_idx == obj_req->bvec_count);
2162 osd_req_op_extent_osd_data_bvec_pos(osd_req, which,
2163 &obj_req->bvec_pos);
2170 static int rbd_osd_setup_stat(struct ceph_osd_request *osd_req, int which)
2172 struct page **pages;
2175 * The response data for a STAT call consists of:
2182 pages = ceph_alloc_page_vector(1, GFP_NOIO);
2184 return PTR_ERR(pages);
2186 osd_req_op_init(osd_req, which, CEPH_OSD_OP_STAT, 0);
2187 osd_req_op_raw_data_in_pages(osd_req, which, pages,
2188 8 + sizeof(struct ceph_timespec),
2193 static int rbd_osd_setup_copyup(struct ceph_osd_request *osd_req, int which,
2196 struct rbd_obj_request *obj_req = osd_req->r_priv;
2199 ret = osd_req_op_cls_init(osd_req, which, "rbd", "copyup");
2203 osd_req_op_cls_request_data_bvecs(osd_req, which, obj_req->copyup_bvecs,
2204 obj_req->copyup_bvec_count, bytes);
2208 static int rbd_obj_init_read(struct rbd_obj_request *obj_req)
2210 obj_req->read_state = RBD_OBJ_READ_START;
2214 static void __rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2217 struct rbd_obj_request *obj_req = osd_req->r_priv;
2218 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2221 if (!use_object_map(rbd_dev) ||
2222 !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST)) {
2223 osd_req_op_alloc_hint_init(osd_req, which++,
2224 rbd_dev->layout.object_size,
2225 rbd_dev->layout.object_size,
2226 rbd_dev->opts->alloc_hint_flags);
2229 if (rbd_obj_is_entire(obj_req))
2230 opcode = CEPH_OSD_OP_WRITEFULL;
2232 opcode = CEPH_OSD_OP_WRITE;
2234 osd_req_op_extent_init(osd_req, which, opcode,
2235 obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2236 rbd_osd_setup_data(osd_req, which);
2239 static int rbd_obj_init_write(struct rbd_obj_request *obj_req)
2243 /* reverse map the entire object onto the parent */
2244 ret = rbd_obj_calc_img_extents(obj_req, true);
2248 if (rbd_obj_copyup_enabled(obj_req))
2249 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
2251 obj_req->write_state = RBD_OBJ_WRITE_START;
2255 static u16 truncate_or_zero_opcode(struct rbd_obj_request *obj_req)
2257 return rbd_obj_is_tail(obj_req) ? CEPH_OSD_OP_TRUNCATE :
2261 static void __rbd_osd_setup_discard_ops(struct ceph_osd_request *osd_req,
2264 struct rbd_obj_request *obj_req = osd_req->r_priv;
2266 if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
2267 rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2268 osd_req_op_init(osd_req, which, CEPH_OSD_OP_DELETE, 0);
2270 osd_req_op_extent_init(osd_req, which,
2271 truncate_or_zero_opcode(obj_req),
2272 obj_req->ex.oe_off, obj_req->ex.oe_len,
2277 static int rbd_obj_init_discard(struct rbd_obj_request *obj_req)
2279 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2284 * Align the range to alloc_size boundary and punt on discards
2285 * that are too small to free up any space.
2287 * alloc_size == object_size && is_tail() is a special case for
2288 * filestore with filestore_punch_hole = false, needed to allow
2289 * truncate (in addition to delete).
2291 if (rbd_dev->opts->alloc_size != rbd_dev->layout.object_size ||
2292 !rbd_obj_is_tail(obj_req)) {
2293 off = round_up(obj_req->ex.oe_off, rbd_dev->opts->alloc_size);
2294 next_off = round_down(obj_req->ex.oe_off + obj_req->ex.oe_len,
2295 rbd_dev->opts->alloc_size);
2296 if (off >= next_off)
2299 dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
2300 obj_req, obj_req->ex.oe_off, obj_req->ex.oe_len,
2301 off, next_off - off);
2302 obj_req->ex.oe_off = off;
2303 obj_req->ex.oe_len = next_off - off;
2306 /* reverse map the entire object onto the parent */
2307 ret = rbd_obj_calc_img_extents(obj_req, true);
2311 obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2312 if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents)
2313 obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2315 obj_req->write_state = RBD_OBJ_WRITE_START;
2319 static void __rbd_osd_setup_zeroout_ops(struct ceph_osd_request *osd_req,
2322 struct rbd_obj_request *obj_req = osd_req->r_priv;
2325 if (rbd_obj_is_entire(obj_req)) {
2326 if (obj_req->num_img_extents) {
2327 if (!(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2328 osd_req_op_init(osd_req, which++,
2329 CEPH_OSD_OP_CREATE, 0);
2330 opcode = CEPH_OSD_OP_TRUNCATE;
2332 rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2333 osd_req_op_init(osd_req, which++,
2334 CEPH_OSD_OP_DELETE, 0);
2338 opcode = truncate_or_zero_opcode(obj_req);
2342 osd_req_op_extent_init(osd_req, which, opcode,
2343 obj_req->ex.oe_off, obj_req->ex.oe_len,
2347 static int rbd_obj_init_zeroout(struct rbd_obj_request *obj_req)
2351 /* reverse map the entire object onto the parent */
2352 ret = rbd_obj_calc_img_extents(obj_req, true);
2356 if (rbd_obj_copyup_enabled(obj_req))
2357 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
2358 if (!obj_req->num_img_extents) {
2359 obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2360 if (rbd_obj_is_entire(obj_req))
2361 obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2364 obj_req->write_state = RBD_OBJ_WRITE_START;
2368 static int count_write_ops(struct rbd_obj_request *obj_req)
2370 struct rbd_img_request *img_req = obj_req->img_request;
2372 switch (img_req->op_type) {
2374 if (!use_object_map(img_req->rbd_dev) ||
2375 !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST))
2376 return 2; /* setallochint + write/writefull */
2378 return 1; /* write/writefull */
2379 case OBJ_OP_DISCARD:
2380 return 1; /* delete/truncate/zero */
2381 case OBJ_OP_ZEROOUT:
2382 if (rbd_obj_is_entire(obj_req) && obj_req->num_img_extents &&
2383 !(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2384 return 2; /* create + truncate */
2386 return 1; /* delete/truncate/zero */
2392 static void rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2395 struct rbd_obj_request *obj_req = osd_req->r_priv;
2397 switch (obj_req->img_request->op_type) {
2399 __rbd_osd_setup_write_ops(osd_req, which);
2401 case OBJ_OP_DISCARD:
2402 __rbd_osd_setup_discard_ops(osd_req, which);
2404 case OBJ_OP_ZEROOUT:
2405 __rbd_osd_setup_zeroout_ops(osd_req, which);
2413 * Prune the list of object requests (adjust offset and/or length, drop
2414 * redundant requests). Prepare object request state machines and image
2415 * request state machine for execution.
2417 static int __rbd_img_fill_request(struct rbd_img_request *img_req)
2419 struct rbd_obj_request *obj_req, *next_obj_req;
2422 for_each_obj_request_safe(img_req, obj_req, next_obj_req) {
2423 switch (img_req->op_type) {
2425 ret = rbd_obj_init_read(obj_req);
2428 ret = rbd_obj_init_write(obj_req);
2430 case OBJ_OP_DISCARD:
2431 ret = rbd_obj_init_discard(obj_req);
2433 case OBJ_OP_ZEROOUT:
2434 ret = rbd_obj_init_zeroout(obj_req);
2442 rbd_img_obj_request_del(img_req, obj_req);
2447 img_req->state = RBD_IMG_START;
2451 union rbd_img_fill_iter {
2452 struct ceph_bio_iter bio_iter;
2453 struct ceph_bvec_iter bvec_iter;
2456 struct rbd_img_fill_ctx {
2457 enum obj_request_type pos_type;
2458 union rbd_img_fill_iter *pos;
2459 union rbd_img_fill_iter iter;
2460 ceph_object_extent_fn_t set_pos_fn;
2461 ceph_object_extent_fn_t count_fn;
2462 ceph_object_extent_fn_t copy_fn;
2465 static struct ceph_object_extent *alloc_object_extent(void *arg)
2467 struct rbd_img_request *img_req = arg;
2468 struct rbd_obj_request *obj_req;
2470 obj_req = rbd_obj_request_create();
2474 rbd_img_obj_request_add(img_req, obj_req);
2475 return &obj_req->ex;
2479 * While su != os && sc == 1 is technically not fancy (it's the same
2480 * layout as su == os && sc == 1), we can't use the nocopy path for it
2481 * because ->set_pos_fn() should be called only once per object.
2482 * ceph_file_to_extents() invokes action_fn once per stripe unit, so
2483 * treat su != os && sc == 1 as fancy.
2485 static bool rbd_layout_is_fancy(struct ceph_file_layout *l)
2487 return l->stripe_unit != l->object_size;
2490 static int rbd_img_fill_request_nocopy(struct rbd_img_request *img_req,
2491 struct ceph_file_extent *img_extents,
2492 u32 num_img_extents,
2493 struct rbd_img_fill_ctx *fctx)
2498 img_req->data_type = fctx->pos_type;
2501 * Create object requests and set each object request's starting
2502 * position in the provided bio (list) or bio_vec array.
2504 fctx->iter = *fctx->pos;
2505 for (i = 0; i < num_img_extents; i++) {
2506 ret = ceph_file_to_extents(&img_req->rbd_dev->layout,
2507 img_extents[i].fe_off,
2508 img_extents[i].fe_len,
2509 &img_req->object_extents,
2510 alloc_object_extent, img_req,
2511 fctx->set_pos_fn, &fctx->iter);
2516 return __rbd_img_fill_request(img_req);
2520 * Map a list of image extents to a list of object extents, create the
2521 * corresponding object requests (normally each to a different object,
2522 * but not always) and add them to @img_req. For each object request,
2523 * set up its data descriptor to point to the corresponding chunk(s) of
2524 * @fctx->pos data buffer.
2526 * Because ceph_file_to_extents() will merge adjacent object extents
2527 * together, each object request's data descriptor may point to multiple
2528 * different chunks of @fctx->pos data buffer.
2530 * @fctx->pos data buffer is assumed to be large enough.
2532 static int rbd_img_fill_request(struct rbd_img_request *img_req,
2533 struct ceph_file_extent *img_extents,
2534 u32 num_img_extents,
2535 struct rbd_img_fill_ctx *fctx)
2537 struct rbd_device *rbd_dev = img_req->rbd_dev;
2538 struct rbd_obj_request *obj_req;
2542 if (fctx->pos_type == OBJ_REQUEST_NODATA ||
2543 !rbd_layout_is_fancy(&rbd_dev->layout))
2544 return rbd_img_fill_request_nocopy(img_req, img_extents,
2545 num_img_extents, fctx);
2547 img_req->data_type = OBJ_REQUEST_OWN_BVECS;
2550 * Create object requests and determine ->bvec_count for each object
2551 * request. Note that ->bvec_count sum over all object requests may
2552 * be greater than the number of bio_vecs in the provided bio (list)
2553 * or bio_vec array because when mapped, those bio_vecs can straddle
2554 * stripe unit boundaries.
2556 fctx->iter = *fctx->pos;
2557 for (i = 0; i < num_img_extents; i++) {
2558 ret = ceph_file_to_extents(&rbd_dev->layout,
2559 img_extents[i].fe_off,
2560 img_extents[i].fe_len,
2561 &img_req->object_extents,
2562 alloc_object_extent, img_req,
2563 fctx->count_fn, &fctx->iter);
2568 for_each_obj_request(img_req, obj_req) {
2569 obj_req->bvec_pos.bvecs = kmalloc_array(obj_req->bvec_count,
2570 sizeof(*obj_req->bvec_pos.bvecs),
2572 if (!obj_req->bvec_pos.bvecs)
2577 * Fill in each object request's private bio_vec array, splitting and
2578 * rearranging the provided bio_vecs in stripe unit chunks as needed.
2580 fctx->iter = *fctx->pos;
2581 for (i = 0; i < num_img_extents; i++) {
2582 ret = ceph_iterate_extents(&rbd_dev->layout,
2583 img_extents[i].fe_off,
2584 img_extents[i].fe_len,
2585 &img_req->object_extents,
2586 fctx->copy_fn, &fctx->iter);
2591 return __rbd_img_fill_request(img_req);
2594 static int rbd_img_fill_nodata(struct rbd_img_request *img_req,
2597 struct ceph_file_extent ex = { off, len };
2598 union rbd_img_fill_iter dummy = {};
2599 struct rbd_img_fill_ctx fctx = {
2600 .pos_type = OBJ_REQUEST_NODATA,
2604 return rbd_img_fill_request(img_req, &ex, 1, &fctx);
2607 static void set_bio_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2609 struct rbd_obj_request *obj_req =
2610 container_of(ex, struct rbd_obj_request, ex);
2611 struct ceph_bio_iter *it = arg;
2613 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2614 obj_req->bio_pos = *it;
2615 ceph_bio_iter_advance(it, bytes);
2618 static void count_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2620 struct rbd_obj_request *obj_req =
2621 container_of(ex, struct rbd_obj_request, ex);
2622 struct ceph_bio_iter *it = arg;
2624 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2625 ceph_bio_iter_advance_step(it, bytes, ({
2626 obj_req->bvec_count++;
2631 static void copy_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2633 struct rbd_obj_request *obj_req =
2634 container_of(ex, struct rbd_obj_request, ex);
2635 struct ceph_bio_iter *it = arg;
2637 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2638 ceph_bio_iter_advance_step(it, bytes, ({
2639 obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2640 obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2644 static int __rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2645 struct ceph_file_extent *img_extents,
2646 u32 num_img_extents,
2647 struct ceph_bio_iter *bio_pos)
2649 struct rbd_img_fill_ctx fctx = {
2650 .pos_type = OBJ_REQUEST_BIO,
2651 .pos = (union rbd_img_fill_iter *)bio_pos,
2652 .set_pos_fn = set_bio_pos,
2653 .count_fn = count_bio_bvecs,
2654 .copy_fn = copy_bio_bvecs,
2657 return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2661 static int rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2662 u64 off, u64 len, struct bio *bio)
2664 struct ceph_file_extent ex = { off, len };
2665 struct ceph_bio_iter it = { .bio = bio, .iter = bio->bi_iter };
2667 return __rbd_img_fill_from_bio(img_req, &ex, 1, &it);
2670 static void set_bvec_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2672 struct rbd_obj_request *obj_req =
2673 container_of(ex, struct rbd_obj_request, ex);
2674 struct ceph_bvec_iter *it = arg;
2676 obj_req->bvec_pos = *it;
2677 ceph_bvec_iter_shorten(&obj_req->bvec_pos, bytes);
2678 ceph_bvec_iter_advance(it, bytes);
2681 static void count_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2683 struct rbd_obj_request *obj_req =
2684 container_of(ex, struct rbd_obj_request, ex);
2685 struct ceph_bvec_iter *it = arg;
2687 ceph_bvec_iter_advance_step(it, bytes, ({
2688 obj_req->bvec_count++;
2692 static void copy_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2694 struct rbd_obj_request *obj_req =
2695 container_of(ex, struct rbd_obj_request, ex);
2696 struct ceph_bvec_iter *it = arg;
2698 ceph_bvec_iter_advance_step(it, bytes, ({
2699 obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2700 obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2704 static int __rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2705 struct ceph_file_extent *img_extents,
2706 u32 num_img_extents,
2707 struct ceph_bvec_iter *bvec_pos)
2709 struct rbd_img_fill_ctx fctx = {
2710 .pos_type = OBJ_REQUEST_BVECS,
2711 .pos = (union rbd_img_fill_iter *)bvec_pos,
2712 .set_pos_fn = set_bvec_pos,
2713 .count_fn = count_bvecs,
2714 .copy_fn = copy_bvecs,
2717 return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2721 static int rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2722 struct ceph_file_extent *img_extents,
2723 u32 num_img_extents,
2724 struct bio_vec *bvecs)
2726 struct ceph_bvec_iter it = {
2728 .iter = { .bi_size = ceph_file_extents_bytes(img_extents,
2732 return __rbd_img_fill_from_bvecs(img_req, img_extents, num_img_extents,
2736 static void rbd_img_handle_request_work(struct work_struct *work)
2738 struct rbd_img_request *img_req =
2739 container_of(work, struct rbd_img_request, work);
2741 rbd_img_handle_request(img_req, img_req->work_result);
2744 static void rbd_img_schedule(struct rbd_img_request *img_req, int result)
2746 INIT_WORK(&img_req->work, rbd_img_handle_request_work);
2747 img_req->work_result = result;
2748 queue_work(rbd_wq, &img_req->work);
2751 static bool rbd_obj_may_exist(struct rbd_obj_request *obj_req)
2753 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2755 if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno)) {
2756 obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2760 dout("%s %p objno %llu assuming dne\n", __func__, obj_req,
2761 obj_req->ex.oe_objno);
2765 static int rbd_obj_read_object(struct rbd_obj_request *obj_req)
2767 struct ceph_osd_request *osd_req;
2770 osd_req = __rbd_obj_add_osd_request(obj_req, NULL, 1);
2771 if (IS_ERR(osd_req))
2772 return PTR_ERR(osd_req);
2774 osd_req_op_extent_init(osd_req, 0, CEPH_OSD_OP_READ,
2775 obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2776 rbd_osd_setup_data(osd_req, 0);
2777 rbd_osd_format_read(osd_req);
2779 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2783 rbd_osd_submit(osd_req);
2787 static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
2789 struct rbd_img_request *img_req = obj_req->img_request;
2790 struct rbd_device *parent = img_req->rbd_dev->parent;
2791 struct rbd_img_request *child_img_req;
2794 child_img_req = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
2798 rbd_img_request_init(child_img_req, parent, OBJ_OP_READ);
2799 __set_bit(IMG_REQ_CHILD, &child_img_req->flags);
2800 child_img_req->obj_request = obj_req;
2802 down_read(&parent->header_rwsem);
2803 rbd_img_capture_header(child_img_req);
2804 up_read(&parent->header_rwsem);
2806 dout("%s child_img_req %p for obj_req %p\n", __func__, child_img_req,
2809 if (!rbd_img_is_write(img_req)) {
2810 switch (img_req->data_type) {
2811 case OBJ_REQUEST_BIO:
2812 ret = __rbd_img_fill_from_bio(child_img_req,
2813 obj_req->img_extents,
2814 obj_req->num_img_extents,
2817 case OBJ_REQUEST_BVECS:
2818 case OBJ_REQUEST_OWN_BVECS:
2819 ret = __rbd_img_fill_from_bvecs(child_img_req,
2820 obj_req->img_extents,
2821 obj_req->num_img_extents,
2822 &obj_req->bvec_pos);
2828 ret = rbd_img_fill_from_bvecs(child_img_req,
2829 obj_req->img_extents,
2830 obj_req->num_img_extents,
2831 obj_req->copyup_bvecs);
2834 rbd_img_request_destroy(child_img_req);
2838 /* avoid parent chain recursion */
2839 rbd_img_schedule(child_img_req, 0);
2843 static bool rbd_obj_advance_read(struct rbd_obj_request *obj_req, int *result)
2845 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2849 switch (obj_req->read_state) {
2850 case RBD_OBJ_READ_START:
2851 rbd_assert(!*result);
2853 if (!rbd_obj_may_exist(obj_req)) {
2855 obj_req->read_state = RBD_OBJ_READ_OBJECT;
2859 ret = rbd_obj_read_object(obj_req);
2864 obj_req->read_state = RBD_OBJ_READ_OBJECT;
2866 case RBD_OBJ_READ_OBJECT:
2867 if (*result == -ENOENT && rbd_dev->parent_overlap) {
2868 /* reverse map this object extent onto the parent */
2869 ret = rbd_obj_calc_img_extents(obj_req, false);
2874 if (obj_req->num_img_extents) {
2875 ret = rbd_obj_read_from_parent(obj_req);
2880 obj_req->read_state = RBD_OBJ_READ_PARENT;
2886 * -ENOENT means a hole in the image -- zero-fill the entire
2887 * length of the request. A short read also implies zero-fill
2888 * to the end of the request.
2890 if (*result == -ENOENT) {
2891 rbd_obj_zero_range(obj_req, 0, obj_req->ex.oe_len);
2893 } else if (*result >= 0) {
2894 if (*result < obj_req->ex.oe_len)
2895 rbd_obj_zero_range(obj_req, *result,
2896 obj_req->ex.oe_len - *result);
2898 rbd_assert(*result == obj_req->ex.oe_len);
2902 case RBD_OBJ_READ_PARENT:
2904 * The parent image is read only up to the overlap -- zero-fill
2905 * from the overlap to the end of the request.
2908 u32 obj_overlap = rbd_obj_img_extents_bytes(obj_req);
2910 if (obj_overlap < obj_req->ex.oe_len)
2911 rbd_obj_zero_range(obj_req, obj_overlap,
2912 obj_req->ex.oe_len - obj_overlap);
2920 static bool rbd_obj_write_is_noop(struct rbd_obj_request *obj_req)
2922 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2924 if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno))
2925 obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2927 if (!(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST) &&
2928 (obj_req->flags & RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT)) {
2929 dout("%s %p noop for nonexistent\n", __func__, obj_req);
2938 * 0 - object map update sent
2939 * 1 - object map update isn't needed
2942 static int rbd_obj_write_pre_object_map(struct rbd_obj_request *obj_req)
2944 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2947 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
2950 if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
2951 new_state = OBJECT_PENDING;
2953 new_state = OBJECT_EXISTS;
2955 return rbd_object_map_update(obj_req, CEPH_NOSNAP, new_state, NULL);
2958 static int rbd_obj_write_object(struct rbd_obj_request *obj_req)
2960 struct ceph_osd_request *osd_req;
2961 int num_ops = count_write_ops(obj_req);
2965 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
2966 num_ops++; /* stat */
2968 osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
2969 if (IS_ERR(osd_req))
2970 return PTR_ERR(osd_req);
2972 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
2973 ret = rbd_osd_setup_stat(osd_req, which++);
2978 rbd_osd_setup_write_ops(osd_req, which);
2979 rbd_osd_format_write(osd_req);
2981 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2985 rbd_osd_submit(osd_req);
2990 * copyup_bvecs pages are never highmem pages
2992 static bool is_zero_bvecs(struct bio_vec *bvecs, u32 bytes)
2994 struct ceph_bvec_iter it = {
2996 .iter = { .bi_size = bytes },
2999 ceph_bvec_iter_advance_step(&it, bytes, ({
3000 if (memchr_inv(page_address(bv.bv_page) + bv.bv_offset, 0,
3007 #define MODS_ONLY U32_MAX
3009 static int rbd_obj_copyup_empty_snapc(struct rbd_obj_request *obj_req,
3012 struct ceph_osd_request *osd_req;
3015 dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3016 rbd_assert(bytes > 0 && bytes != MODS_ONLY);
3018 osd_req = __rbd_obj_add_osd_request(obj_req, &rbd_empty_snapc, 1);
3019 if (IS_ERR(osd_req))
3020 return PTR_ERR(osd_req);
3022 ret = rbd_osd_setup_copyup(osd_req, 0, bytes);
3026 rbd_osd_format_write(osd_req);
3028 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3032 rbd_osd_submit(osd_req);
3036 static int rbd_obj_copyup_current_snapc(struct rbd_obj_request *obj_req,
3039 struct ceph_osd_request *osd_req;
3040 int num_ops = count_write_ops(obj_req);
3044 dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3046 if (bytes != MODS_ONLY)
3047 num_ops++; /* copyup */
3049 osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
3050 if (IS_ERR(osd_req))
3051 return PTR_ERR(osd_req);
3053 if (bytes != MODS_ONLY) {
3054 ret = rbd_osd_setup_copyup(osd_req, which++, bytes);
3059 rbd_osd_setup_write_ops(osd_req, which);
3060 rbd_osd_format_write(osd_req);
3062 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3066 rbd_osd_submit(osd_req);
3070 static int setup_copyup_bvecs(struct rbd_obj_request *obj_req, u64 obj_overlap)
3074 rbd_assert(!obj_req->copyup_bvecs);
3075 obj_req->copyup_bvec_count = calc_pages_for(0, obj_overlap);
3076 obj_req->copyup_bvecs = kcalloc(obj_req->copyup_bvec_count,
3077 sizeof(*obj_req->copyup_bvecs),
3079 if (!obj_req->copyup_bvecs)
3082 for (i = 0; i < obj_req->copyup_bvec_count; i++) {
3083 unsigned int len = min(obj_overlap, (u64)PAGE_SIZE);
3085 obj_req->copyup_bvecs[i].bv_page = alloc_page(GFP_NOIO);
3086 if (!obj_req->copyup_bvecs[i].bv_page)
3089 obj_req->copyup_bvecs[i].bv_offset = 0;
3090 obj_req->copyup_bvecs[i].bv_len = len;
3094 rbd_assert(!obj_overlap);
3099 * The target object doesn't exist. Read the data for the entire
3100 * target object up to the overlap point (if any) from the parent,
3101 * so we can use it for a copyup.
3103 static int rbd_obj_copyup_read_parent(struct rbd_obj_request *obj_req)
3105 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3108 rbd_assert(obj_req->num_img_extents);
3109 prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
3110 rbd_dev->parent_overlap);
3111 if (!obj_req->num_img_extents) {
3113 * The overlap has become 0 (most likely because the
3114 * image has been flattened). Re-submit the original write
3115 * request -- pass MODS_ONLY since the copyup isn't needed
3118 return rbd_obj_copyup_current_snapc(obj_req, MODS_ONLY);
3121 ret = setup_copyup_bvecs(obj_req, rbd_obj_img_extents_bytes(obj_req));
3125 return rbd_obj_read_from_parent(obj_req);
3128 static void rbd_obj_copyup_object_maps(struct rbd_obj_request *obj_req)
3130 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3131 struct ceph_snap_context *snapc = obj_req->img_request->snapc;
3136 rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3138 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3141 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3144 for (i = 0; i < snapc->num_snaps; i++) {
3145 if ((rbd_dev->header.features & RBD_FEATURE_FAST_DIFF) &&
3146 i + 1 < snapc->num_snaps)
3147 new_state = OBJECT_EXISTS_CLEAN;
3149 new_state = OBJECT_EXISTS;
3151 ret = rbd_object_map_update(obj_req, snapc->snaps[i],
3154 obj_req->pending.result = ret;
3159 obj_req->pending.num_pending++;
3163 static void rbd_obj_copyup_write_object(struct rbd_obj_request *obj_req)
3165 u32 bytes = rbd_obj_img_extents_bytes(obj_req);
3168 rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3171 * Only send non-zero copyup data to save some I/O and network
3172 * bandwidth -- zero copyup data is equivalent to the object not
3175 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3178 if (obj_req->img_request->snapc->num_snaps && bytes > 0) {
3180 * Send a copyup request with an empty snapshot context to
3181 * deep-copyup the object through all existing snapshots.
3182 * A second request with the current snapshot context will be
3183 * sent for the actual modification.
3185 ret = rbd_obj_copyup_empty_snapc(obj_req, bytes);
3187 obj_req->pending.result = ret;
3191 obj_req->pending.num_pending++;
3195 ret = rbd_obj_copyup_current_snapc(obj_req, bytes);
3197 obj_req->pending.result = ret;
3201 obj_req->pending.num_pending++;
3204 static bool rbd_obj_advance_copyup(struct rbd_obj_request *obj_req, int *result)
3206 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3210 switch (obj_req->copyup_state) {
3211 case RBD_OBJ_COPYUP_START:
3212 rbd_assert(!*result);
3214 ret = rbd_obj_copyup_read_parent(obj_req);
3219 if (obj_req->num_img_extents)
3220 obj_req->copyup_state = RBD_OBJ_COPYUP_READ_PARENT;
3222 obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3224 case RBD_OBJ_COPYUP_READ_PARENT:
3228 if (is_zero_bvecs(obj_req->copyup_bvecs,
3229 rbd_obj_img_extents_bytes(obj_req))) {
3230 dout("%s %p detected zeros\n", __func__, obj_req);
3231 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ZEROS;
3234 rbd_obj_copyup_object_maps(obj_req);
3235 if (!obj_req->pending.num_pending) {
3236 *result = obj_req->pending.result;
3237 obj_req->copyup_state = RBD_OBJ_COPYUP_OBJECT_MAPS;
3240 obj_req->copyup_state = __RBD_OBJ_COPYUP_OBJECT_MAPS;
3242 case __RBD_OBJ_COPYUP_OBJECT_MAPS:
3243 if (!pending_result_dec(&obj_req->pending, result))
3246 case RBD_OBJ_COPYUP_OBJECT_MAPS:
3248 rbd_warn(rbd_dev, "snap object map update failed: %d",
3253 rbd_obj_copyup_write_object(obj_req);
3254 if (!obj_req->pending.num_pending) {
3255 *result = obj_req->pending.result;
3256 obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3259 obj_req->copyup_state = __RBD_OBJ_COPYUP_WRITE_OBJECT;
3261 case __RBD_OBJ_COPYUP_WRITE_OBJECT:
3262 if (!pending_result_dec(&obj_req->pending, result))
3265 case RBD_OBJ_COPYUP_WRITE_OBJECT:
3274 * 0 - object map update sent
3275 * 1 - object map update isn't needed
3278 static int rbd_obj_write_post_object_map(struct rbd_obj_request *obj_req)
3280 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3281 u8 current_state = OBJECT_PENDING;
3283 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3286 if (!(obj_req->flags & RBD_OBJ_FLAG_DELETION))
3289 return rbd_object_map_update(obj_req, CEPH_NOSNAP, OBJECT_NONEXISTENT,
3293 static bool rbd_obj_advance_write(struct rbd_obj_request *obj_req, int *result)
3295 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3299 switch (obj_req->write_state) {
3300 case RBD_OBJ_WRITE_START:
3301 rbd_assert(!*result);
3303 if (rbd_obj_write_is_noop(obj_req))
3306 ret = rbd_obj_write_pre_object_map(obj_req);
3311 obj_req->write_state = RBD_OBJ_WRITE_PRE_OBJECT_MAP;
3315 case RBD_OBJ_WRITE_PRE_OBJECT_MAP:
3317 rbd_warn(rbd_dev, "pre object map update failed: %d",
3321 ret = rbd_obj_write_object(obj_req);
3326 obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
3328 case RBD_OBJ_WRITE_OBJECT:
3329 if (*result == -ENOENT) {
3330 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
3332 obj_req->copyup_state = RBD_OBJ_COPYUP_START;
3333 obj_req->write_state = __RBD_OBJ_WRITE_COPYUP;
3337 * On a non-existent object:
3338 * delete - -ENOENT, truncate/zero - 0
3340 if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
3346 obj_req->write_state = RBD_OBJ_WRITE_COPYUP;
3348 case __RBD_OBJ_WRITE_COPYUP:
3349 if (!rbd_obj_advance_copyup(obj_req, result))
3352 case RBD_OBJ_WRITE_COPYUP:
3354 rbd_warn(rbd_dev, "copyup failed: %d", *result);
3357 ret = rbd_obj_write_post_object_map(obj_req);
3362 obj_req->write_state = RBD_OBJ_WRITE_POST_OBJECT_MAP;
3366 case RBD_OBJ_WRITE_POST_OBJECT_MAP:
3368 rbd_warn(rbd_dev, "post object map update failed: %d",
3377 * Return true if @obj_req is completed.
3379 static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
3382 struct rbd_img_request *img_req = obj_req->img_request;
3383 struct rbd_device *rbd_dev = img_req->rbd_dev;
3386 mutex_lock(&obj_req->state_mutex);
3387 if (!rbd_img_is_write(img_req))
3388 done = rbd_obj_advance_read(obj_req, result);
3390 done = rbd_obj_advance_write(obj_req, result);
3391 mutex_unlock(&obj_req->state_mutex);
3393 if (done && *result) {
3394 rbd_assert(*result < 0);
3395 rbd_warn(rbd_dev, "%s at objno %llu %llu~%llu result %d",
3396 obj_op_name(img_req->op_type), obj_req->ex.oe_objno,
3397 obj_req->ex.oe_off, obj_req->ex.oe_len, *result);
3403 * This is open-coded in rbd_img_handle_request() to avoid parent chain
3406 static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result)
3408 if (__rbd_obj_handle_request(obj_req, &result))
3409 rbd_img_handle_request(obj_req->img_request, result);
3412 static bool need_exclusive_lock(struct rbd_img_request *img_req)
3414 struct rbd_device *rbd_dev = img_req->rbd_dev;
3416 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK))
3419 if (rbd_is_ro(rbd_dev))
3422 rbd_assert(!test_bit(IMG_REQ_CHILD, &img_req->flags));
3423 if (rbd_dev->opts->lock_on_read ||
3424 (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3427 return rbd_img_is_write(img_req);
3430 static bool rbd_lock_add_request(struct rbd_img_request *img_req)
3432 struct rbd_device *rbd_dev = img_req->rbd_dev;
3435 lockdep_assert_held(&rbd_dev->lock_rwsem);
3436 locked = rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED;
3437 spin_lock(&rbd_dev->lock_lists_lock);
3438 rbd_assert(list_empty(&img_req->lock_item));
3440 list_add_tail(&img_req->lock_item, &rbd_dev->acquiring_list);
3442 list_add_tail(&img_req->lock_item, &rbd_dev->running_list);
3443 spin_unlock(&rbd_dev->lock_lists_lock);
3447 static void rbd_lock_del_request(struct rbd_img_request *img_req)
3449 struct rbd_device *rbd_dev = img_req->rbd_dev;
3452 lockdep_assert_held(&rbd_dev->lock_rwsem);
3453 spin_lock(&rbd_dev->lock_lists_lock);
3454 rbd_assert(!list_empty(&img_req->lock_item));
3455 list_del_init(&img_req->lock_item);
3456 need_wakeup = (rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING &&
3457 list_empty(&rbd_dev->running_list));
3458 spin_unlock(&rbd_dev->lock_lists_lock);
3460 complete(&rbd_dev->releasing_wait);
3463 static int rbd_img_exclusive_lock(struct rbd_img_request *img_req)
3465 struct rbd_device *rbd_dev = img_req->rbd_dev;
3467 if (!need_exclusive_lock(img_req))
3470 if (rbd_lock_add_request(img_req))
3473 if (rbd_dev->opts->exclusive) {
3474 WARN_ON(1); /* lock got released? */
3479 * Note the use of mod_delayed_work() in rbd_acquire_lock()
3480 * and cancel_delayed_work() in wake_lock_waiters().
3482 dout("%s rbd_dev %p queueing lock_dwork\n", __func__, rbd_dev);
3483 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
3487 static void rbd_img_object_requests(struct rbd_img_request *img_req)
3489 struct rbd_obj_request *obj_req;
3491 rbd_assert(!img_req->pending.result && !img_req->pending.num_pending);
3493 for_each_obj_request(img_req, obj_req) {
3496 if (__rbd_obj_handle_request(obj_req, &result)) {
3498 img_req->pending.result = result;
3502 img_req->pending.num_pending++;
3507 static bool rbd_img_advance(struct rbd_img_request *img_req, int *result)
3509 struct rbd_device *rbd_dev = img_req->rbd_dev;
3513 switch (img_req->state) {
3515 rbd_assert(!*result);
3517 ret = rbd_img_exclusive_lock(img_req);
3522 img_req->state = RBD_IMG_EXCLUSIVE_LOCK;
3526 case RBD_IMG_EXCLUSIVE_LOCK:
3530 rbd_assert(!need_exclusive_lock(img_req) ||
3531 __rbd_is_lock_owner(rbd_dev));
3533 rbd_img_object_requests(img_req);
3534 if (!img_req->pending.num_pending) {
3535 *result = img_req->pending.result;
3536 img_req->state = RBD_IMG_OBJECT_REQUESTS;
3539 img_req->state = __RBD_IMG_OBJECT_REQUESTS;
3541 case __RBD_IMG_OBJECT_REQUESTS:
3542 if (!pending_result_dec(&img_req->pending, result))
3545 case RBD_IMG_OBJECT_REQUESTS:
3553 * Return true if @img_req is completed.
3555 static bool __rbd_img_handle_request(struct rbd_img_request *img_req,
3558 struct rbd_device *rbd_dev = img_req->rbd_dev;
3561 if (need_exclusive_lock(img_req)) {
3562 down_read(&rbd_dev->lock_rwsem);
3563 mutex_lock(&img_req->state_mutex);
3564 done = rbd_img_advance(img_req, result);
3566 rbd_lock_del_request(img_req);
3567 mutex_unlock(&img_req->state_mutex);
3568 up_read(&rbd_dev->lock_rwsem);
3570 mutex_lock(&img_req->state_mutex);
3571 done = rbd_img_advance(img_req, result);
3572 mutex_unlock(&img_req->state_mutex);
3575 if (done && *result) {
3576 rbd_assert(*result < 0);
3577 rbd_warn(rbd_dev, "%s%s result %d",
3578 test_bit(IMG_REQ_CHILD, &img_req->flags) ? "child " : "",
3579 obj_op_name(img_req->op_type), *result);
3584 static void rbd_img_handle_request(struct rbd_img_request *img_req, int result)
3587 if (!__rbd_img_handle_request(img_req, &result))
3590 if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
3591 struct rbd_obj_request *obj_req = img_req->obj_request;
3593 rbd_img_request_destroy(img_req);
3594 if (__rbd_obj_handle_request(obj_req, &result)) {
3595 img_req = obj_req->img_request;
3599 struct request *rq = blk_mq_rq_from_pdu(img_req);
3601 rbd_img_request_destroy(img_req);
3602 blk_mq_end_request(rq, errno_to_blk_status(result));
3606 static const struct rbd_client_id rbd_empty_cid;
3608 static bool rbd_cid_equal(const struct rbd_client_id *lhs,
3609 const struct rbd_client_id *rhs)
3611 return lhs->gid == rhs->gid && lhs->handle == rhs->handle;
3614 static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev)
3616 struct rbd_client_id cid;
3618 mutex_lock(&rbd_dev->watch_mutex);
3619 cid.gid = ceph_client_gid(rbd_dev->rbd_client->client);
3620 cid.handle = rbd_dev->watch_cookie;
3621 mutex_unlock(&rbd_dev->watch_mutex);
3626 * lock_rwsem must be held for write
3628 static void rbd_set_owner_cid(struct rbd_device *rbd_dev,
3629 const struct rbd_client_id *cid)
3631 dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev,
3632 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle,
3633 cid->gid, cid->handle);
3634 rbd_dev->owner_cid = *cid; /* struct */
3637 static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf)
3639 mutex_lock(&rbd_dev->watch_mutex);
3640 sprintf(buf, "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie);
3641 mutex_unlock(&rbd_dev->watch_mutex);
3644 static void __rbd_lock(struct rbd_device *rbd_dev, const char *cookie)
3646 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3648 rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED;
3649 strcpy(rbd_dev->lock_cookie, cookie);
3650 rbd_set_owner_cid(rbd_dev, &cid);
3651 queue_work(rbd_dev->task_wq, &rbd_dev->acquired_lock_work);
3655 * lock_rwsem must be held for write
3657 static int rbd_lock(struct rbd_device *rbd_dev)
3659 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3663 WARN_ON(__rbd_is_lock_owner(rbd_dev) ||
3664 rbd_dev->lock_cookie[0] != '\0');
3666 format_lock_cookie(rbd_dev, cookie);
3667 ret = ceph_cls_lock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3668 RBD_LOCK_NAME, CEPH_CLS_LOCK_EXCLUSIVE, cookie,
3669 RBD_LOCK_TAG, "", 0);
3673 __rbd_lock(rbd_dev, cookie);
3678 * lock_rwsem must be held for write
3680 static void rbd_unlock(struct rbd_device *rbd_dev)
3682 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3685 WARN_ON(!__rbd_is_lock_owner(rbd_dev) ||
3686 rbd_dev->lock_cookie[0] == '\0');
3688 ret = ceph_cls_unlock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3689 RBD_LOCK_NAME, rbd_dev->lock_cookie);
3690 if (ret && ret != -ENOENT)
3691 rbd_warn(rbd_dev, "failed to unlock header: %d", ret);
3693 /* treat errors as the image is unlocked */
3694 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
3695 rbd_dev->lock_cookie[0] = '\0';
3696 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3697 queue_work(rbd_dev->task_wq, &rbd_dev->released_lock_work);
3700 static int __rbd_notify_op_lock(struct rbd_device *rbd_dev,
3701 enum rbd_notify_op notify_op,
3702 struct page ***preply_pages,
3705 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3706 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3707 char buf[4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN];
3708 int buf_size = sizeof(buf);
3711 dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op);
3713 /* encode *LockPayload NotifyMessage (op + ClientId) */
3714 ceph_start_encoding(&p, 2, 1, buf_size - CEPH_ENCODING_START_BLK_LEN);
3715 ceph_encode_32(&p, notify_op);
3716 ceph_encode_64(&p, cid.gid);
3717 ceph_encode_64(&p, cid.handle);
3719 return ceph_osdc_notify(osdc, &rbd_dev->header_oid,
3720 &rbd_dev->header_oloc, buf, buf_size,
3721 RBD_NOTIFY_TIMEOUT, preply_pages, preply_len);
3724 static void rbd_notify_op_lock(struct rbd_device *rbd_dev,
3725 enum rbd_notify_op notify_op)
3727 __rbd_notify_op_lock(rbd_dev, notify_op, NULL, NULL);
3730 static void rbd_notify_acquired_lock(struct work_struct *work)
3732 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3733 acquired_lock_work);
3735 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_ACQUIRED_LOCK);
3738 static void rbd_notify_released_lock(struct work_struct *work)
3740 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3741 released_lock_work);
3743 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_RELEASED_LOCK);
3746 static int rbd_request_lock(struct rbd_device *rbd_dev)
3748 struct page **reply_pages;
3750 bool lock_owner_responded = false;
3753 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3755 ret = __rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_REQUEST_LOCK,
3756 &reply_pages, &reply_len);
3757 if (ret && ret != -ETIMEDOUT) {
3758 rbd_warn(rbd_dev, "failed to request lock: %d", ret);
3762 if (reply_len > 0 && reply_len <= PAGE_SIZE) {
3763 void *p = page_address(reply_pages[0]);
3764 void *const end = p + reply_len;
3767 ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */
3772 ceph_decode_need(&p, end, 8 + 8, e_inval);
3773 p += 8 + 8; /* skip gid and cookie */
3775 ceph_decode_32_safe(&p, end, len, e_inval);
3779 if (lock_owner_responded) {
3781 "duplicate lock owners detected");
3786 lock_owner_responded = true;
3787 ret = ceph_start_decoding(&p, end, 1, "ResponseMessage",
3791 "failed to decode ResponseMessage: %d",
3796 ret = ceph_decode_32(&p);
3800 if (!lock_owner_responded) {
3801 rbd_warn(rbd_dev, "no lock owners detected");
3806 ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3815 * Either image request state machine(s) or rbd_add_acquire_lock()
3818 static void wake_lock_waiters(struct rbd_device *rbd_dev, int result)
3820 struct rbd_img_request *img_req;
3822 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
3823 lockdep_assert_held_write(&rbd_dev->lock_rwsem);
3825 cancel_delayed_work(&rbd_dev->lock_dwork);
3826 if (!completion_done(&rbd_dev->acquire_wait)) {
3827 rbd_assert(list_empty(&rbd_dev->acquiring_list) &&
3828 list_empty(&rbd_dev->running_list));
3829 rbd_dev->acquire_err = result;
3830 complete_all(&rbd_dev->acquire_wait);
3834 list_for_each_entry(img_req, &rbd_dev->acquiring_list, lock_item) {
3835 mutex_lock(&img_req->state_mutex);
3836 rbd_assert(img_req->state == RBD_IMG_EXCLUSIVE_LOCK);
3837 rbd_img_schedule(img_req, result);
3838 mutex_unlock(&img_req->state_mutex);
3841 list_splice_tail_init(&rbd_dev->acquiring_list, &rbd_dev->running_list);
3844 static int get_lock_owner_info(struct rbd_device *rbd_dev,
3845 struct ceph_locker **lockers, u32 *num_lockers)
3847 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3852 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3854 ret = ceph_cls_lock_info(osdc, &rbd_dev->header_oid,
3855 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3856 &lock_type, &lock_tag, lockers, num_lockers);
3860 if (*num_lockers == 0) {
3861 dout("%s rbd_dev %p no lockers detected\n", __func__, rbd_dev);
3865 if (strcmp(lock_tag, RBD_LOCK_TAG)) {
3866 rbd_warn(rbd_dev, "locked by external mechanism, tag %s",
3872 if (lock_type == CEPH_CLS_LOCK_SHARED) {
3873 rbd_warn(rbd_dev, "shared lock type detected");
3878 if (strncmp((*lockers)[0].id.cookie, RBD_LOCK_COOKIE_PREFIX,
3879 strlen(RBD_LOCK_COOKIE_PREFIX))) {
3880 rbd_warn(rbd_dev, "locked by external mechanism, cookie %s",
3881 (*lockers)[0].id.cookie);
3891 static int find_watcher(struct rbd_device *rbd_dev,
3892 const struct ceph_locker *locker)
3894 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3895 struct ceph_watch_item *watchers;
3901 ret = ceph_osdc_list_watchers(osdc, &rbd_dev->header_oid,
3902 &rbd_dev->header_oloc, &watchers,
3907 sscanf(locker->id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu", &cookie);
3908 for (i = 0; i < num_watchers; i++) {
3910 * Ignore addr->type while comparing. This mimics
3911 * entity_addr_t::get_legacy_str() + strcmp().
3913 if (ceph_addr_equal_no_type(&watchers[i].addr,
3914 &locker->info.addr) &&
3915 watchers[i].cookie == cookie) {
3916 struct rbd_client_id cid = {
3917 .gid = le64_to_cpu(watchers[i].name.num),
3921 dout("%s rbd_dev %p found cid %llu-%llu\n", __func__,
3922 rbd_dev, cid.gid, cid.handle);
3923 rbd_set_owner_cid(rbd_dev, &cid);
3929 dout("%s rbd_dev %p no watchers\n", __func__, rbd_dev);
3937 * lock_rwsem must be held for write
3939 static int rbd_try_lock(struct rbd_device *rbd_dev)
3941 struct ceph_client *client = rbd_dev->rbd_client->client;
3942 struct ceph_locker *lockers;
3947 ret = rbd_lock(rbd_dev);
3951 /* determine if the current lock holder is still alive */
3952 ret = get_lock_owner_info(rbd_dev, &lockers, &num_lockers);
3956 if (num_lockers == 0)
3959 ret = find_watcher(rbd_dev, lockers);
3961 goto out; /* request lock or error */
3963 rbd_warn(rbd_dev, "breaking header lock owned by %s%llu",
3964 ENTITY_NAME(lockers[0].id.name));
3966 ret = ceph_monc_blocklist_add(&client->monc,
3967 &lockers[0].info.addr);
3969 rbd_warn(rbd_dev, "blocklist of %s%llu failed: %d",
3970 ENTITY_NAME(lockers[0].id.name), ret);
3974 ret = ceph_cls_break_lock(&client->osdc, &rbd_dev->header_oid,
3975 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3976 lockers[0].id.cookie,
3977 &lockers[0].id.name);
3978 if (ret && ret != -ENOENT)
3982 ceph_free_lockers(lockers, num_lockers);
3986 ceph_free_lockers(lockers, num_lockers);
3990 static int rbd_post_acquire_action(struct rbd_device *rbd_dev)
3994 if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) {
3995 ret = rbd_object_map_open(rbd_dev);
4006 * 1 - caller should call rbd_request_lock()
4009 static int rbd_try_acquire_lock(struct rbd_device *rbd_dev)
4013 down_read(&rbd_dev->lock_rwsem);
4014 dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
4015 rbd_dev->lock_state);
4016 if (__rbd_is_lock_owner(rbd_dev)) {
4017 up_read(&rbd_dev->lock_rwsem);
4021 up_read(&rbd_dev->lock_rwsem);
4022 down_write(&rbd_dev->lock_rwsem);
4023 dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
4024 rbd_dev->lock_state);
4025 if (__rbd_is_lock_owner(rbd_dev)) {
4026 up_write(&rbd_dev->lock_rwsem);
4030 ret = rbd_try_lock(rbd_dev);
4032 rbd_warn(rbd_dev, "failed to lock header: %d", ret);
4033 if (ret == -EBLOCKLISTED)
4036 ret = 1; /* request lock anyway */
4039 up_write(&rbd_dev->lock_rwsem);
4043 rbd_assert(rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED);
4044 rbd_assert(list_empty(&rbd_dev->running_list));
4046 ret = rbd_post_acquire_action(rbd_dev);
4048 rbd_warn(rbd_dev, "post-acquire action failed: %d", ret);
4050 * Can't stay in RBD_LOCK_STATE_LOCKED because
4051 * rbd_lock_add_request() would let the request through,
4052 * assuming that e.g. object map is locked and loaded.
4054 rbd_unlock(rbd_dev);
4058 wake_lock_waiters(rbd_dev, ret);
4059 up_write(&rbd_dev->lock_rwsem);
4063 static void rbd_acquire_lock(struct work_struct *work)
4065 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4066 struct rbd_device, lock_dwork);
4069 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4071 ret = rbd_try_acquire_lock(rbd_dev);
4073 dout("%s rbd_dev %p ret %d - done\n", __func__, rbd_dev, ret);
4077 ret = rbd_request_lock(rbd_dev);
4078 if (ret == -ETIMEDOUT) {
4079 goto again; /* treat this as a dead client */
4080 } else if (ret == -EROFS) {
4081 rbd_warn(rbd_dev, "peer will not release lock");
4082 down_write(&rbd_dev->lock_rwsem);
4083 wake_lock_waiters(rbd_dev, ret);
4084 up_write(&rbd_dev->lock_rwsem);
4085 } else if (ret < 0) {
4086 rbd_warn(rbd_dev, "error requesting lock: %d", ret);
4087 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4091 * lock owner acked, but resend if we don't see them
4094 dout("%s rbd_dev %p requeuing lock_dwork\n", __func__,
4096 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4097 msecs_to_jiffies(2 * RBD_NOTIFY_TIMEOUT * MSEC_PER_SEC));
4101 static bool rbd_quiesce_lock(struct rbd_device *rbd_dev)
4105 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4106 lockdep_assert_held_write(&rbd_dev->lock_rwsem);
4108 if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED)
4112 * Ensure that all in-flight IO is flushed.
4114 rbd_dev->lock_state = RBD_LOCK_STATE_RELEASING;
4115 rbd_assert(!completion_done(&rbd_dev->releasing_wait));
4116 need_wait = !list_empty(&rbd_dev->running_list);
4117 downgrade_write(&rbd_dev->lock_rwsem);
4119 wait_for_completion(&rbd_dev->releasing_wait);
4120 up_read(&rbd_dev->lock_rwsem);
4122 down_write(&rbd_dev->lock_rwsem);
4123 if (rbd_dev->lock_state != RBD_LOCK_STATE_RELEASING)
4126 rbd_assert(list_empty(&rbd_dev->running_list));
4130 static void rbd_pre_release_action(struct rbd_device *rbd_dev)
4132 if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)
4133 rbd_object_map_close(rbd_dev);
4136 static void __rbd_release_lock(struct rbd_device *rbd_dev)
4138 rbd_assert(list_empty(&rbd_dev->running_list));
4140 rbd_pre_release_action(rbd_dev);
4141 rbd_unlock(rbd_dev);
4145 * lock_rwsem must be held for write
4147 static void rbd_release_lock(struct rbd_device *rbd_dev)
4149 if (!rbd_quiesce_lock(rbd_dev))
4152 __rbd_release_lock(rbd_dev);
4155 * Give others a chance to grab the lock - we would re-acquire
4156 * almost immediately if we got new IO while draining the running
4157 * list otherwise. We need to ack our own notifications, so this
4158 * lock_dwork will be requeued from rbd_handle_released_lock() by
4159 * way of maybe_kick_acquire().
4161 cancel_delayed_work(&rbd_dev->lock_dwork);
4164 static void rbd_release_lock_work(struct work_struct *work)
4166 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
4169 down_write(&rbd_dev->lock_rwsem);
4170 rbd_release_lock(rbd_dev);
4171 up_write(&rbd_dev->lock_rwsem);
4174 static void maybe_kick_acquire(struct rbd_device *rbd_dev)
4178 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4179 if (__rbd_is_lock_owner(rbd_dev))
4182 spin_lock(&rbd_dev->lock_lists_lock);
4183 have_requests = !list_empty(&rbd_dev->acquiring_list);
4184 spin_unlock(&rbd_dev->lock_lists_lock);
4185 if (have_requests || delayed_work_pending(&rbd_dev->lock_dwork)) {
4186 dout("%s rbd_dev %p kicking lock_dwork\n", __func__, rbd_dev);
4187 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4191 static void rbd_handle_acquired_lock(struct rbd_device *rbd_dev, u8 struct_v,
4194 struct rbd_client_id cid = { 0 };
4196 if (struct_v >= 2) {
4197 cid.gid = ceph_decode_64(p);
4198 cid.handle = ceph_decode_64(p);
4201 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4203 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4204 down_write(&rbd_dev->lock_rwsem);
4205 if (rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4206 dout("%s rbd_dev %p cid %llu-%llu == owner_cid\n",
4207 __func__, rbd_dev, cid.gid, cid.handle);
4209 rbd_set_owner_cid(rbd_dev, &cid);
4211 downgrade_write(&rbd_dev->lock_rwsem);
4213 down_read(&rbd_dev->lock_rwsem);
4216 maybe_kick_acquire(rbd_dev);
4217 up_read(&rbd_dev->lock_rwsem);
4220 static void rbd_handle_released_lock(struct rbd_device *rbd_dev, u8 struct_v,
4223 struct rbd_client_id cid = { 0 };
4225 if (struct_v >= 2) {
4226 cid.gid = ceph_decode_64(p);
4227 cid.handle = ceph_decode_64(p);
4230 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4232 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4233 down_write(&rbd_dev->lock_rwsem);
4234 if (!rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4235 dout("%s rbd_dev %p cid %llu-%llu != owner_cid %llu-%llu\n",
4236 __func__, rbd_dev, cid.gid, cid.handle,
4237 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle);
4239 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4241 downgrade_write(&rbd_dev->lock_rwsem);
4243 down_read(&rbd_dev->lock_rwsem);
4246 maybe_kick_acquire(rbd_dev);
4247 up_read(&rbd_dev->lock_rwsem);
4251 * Returns result for ResponseMessage to be encoded (<= 0), or 1 if no
4252 * ResponseMessage is needed.
4254 static int rbd_handle_request_lock(struct rbd_device *rbd_dev, u8 struct_v,
4257 struct rbd_client_id my_cid = rbd_get_cid(rbd_dev);
4258 struct rbd_client_id cid = { 0 };
4261 if (struct_v >= 2) {
4262 cid.gid = ceph_decode_64(p);
4263 cid.handle = ceph_decode_64(p);
4266 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4268 if (rbd_cid_equal(&cid, &my_cid))
4271 down_read(&rbd_dev->lock_rwsem);
4272 if (__rbd_is_lock_owner(rbd_dev)) {
4273 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED &&
4274 rbd_cid_equal(&rbd_dev->owner_cid, &rbd_empty_cid))
4278 * encode ResponseMessage(0) so the peer can detect
4283 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED) {
4284 if (!rbd_dev->opts->exclusive) {
4285 dout("%s rbd_dev %p queueing unlock_work\n",
4287 queue_work(rbd_dev->task_wq,
4288 &rbd_dev->unlock_work);
4290 /* refuse to release the lock */
4297 up_read(&rbd_dev->lock_rwsem);
4301 static void __rbd_acknowledge_notify(struct rbd_device *rbd_dev,
4302 u64 notify_id, u64 cookie, s32 *result)
4304 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4305 char buf[4 + CEPH_ENCODING_START_BLK_LEN];
4306 int buf_size = sizeof(buf);
4312 /* encode ResponseMessage */
4313 ceph_start_encoding(&p, 1, 1,
4314 buf_size - CEPH_ENCODING_START_BLK_LEN);
4315 ceph_encode_32(&p, *result);
4320 ret = ceph_osdc_notify_ack(osdc, &rbd_dev->header_oid,
4321 &rbd_dev->header_oloc, notify_id, cookie,
4324 rbd_warn(rbd_dev, "acknowledge_notify failed: %d", ret);
4327 static void rbd_acknowledge_notify(struct rbd_device *rbd_dev, u64 notify_id,
4330 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4331 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, NULL);
4334 static void rbd_acknowledge_notify_result(struct rbd_device *rbd_dev,
4335 u64 notify_id, u64 cookie, s32 result)
4337 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
4338 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, &result);
4341 static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie,
4342 u64 notifier_id, void *data, size_t data_len)
4344 struct rbd_device *rbd_dev = arg;
4346 void *const end = p + data_len;
4352 dout("%s rbd_dev %p cookie %llu notify_id %llu data_len %zu\n",
4353 __func__, rbd_dev, cookie, notify_id, data_len);
4355 ret = ceph_start_decoding(&p, end, 1, "NotifyMessage",
4358 rbd_warn(rbd_dev, "failed to decode NotifyMessage: %d",
4363 notify_op = ceph_decode_32(&p);
4365 /* legacy notification for header updates */
4366 notify_op = RBD_NOTIFY_OP_HEADER_UPDATE;
4370 dout("%s rbd_dev %p notify_op %u\n", __func__, rbd_dev, notify_op);
4371 switch (notify_op) {
4372 case RBD_NOTIFY_OP_ACQUIRED_LOCK:
4373 rbd_handle_acquired_lock(rbd_dev, struct_v, &p);
4374 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4376 case RBD_NOTIFY_OP_RELEASED_LOCK:
4377 rbd_handle_released_lock(rbd_dev, struct_v, &p);
4378 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4380 case RBD_NOTIFY_OP_REQUEST_LOCK:
4381 ret = rbd_handle_request_lock(rbd_dev, struct_v, &p);
4383 rbd_acknowledge_notify_result(rbd_dev, notify_id,
4386 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4388 case RBD_NOTIFY_OP_HEADER_UPDATE:
4389 ret = rbd_dev_refresh(rbd_dev);
4391 rbd_warn(rbd_dev, "refresh failed: %d", ret);
4393 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4396 if (rbd_is_lock_owner(rbd_dev))
4397 rbd_acknowledge_notify_result(rbd_dev, notify_id,
4398 cookie, -EOPNOTSUPP);
4400 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4405 static void __rbd_unregister_watch(struct rbd_device *rbd_dev);
4407 static void rbd_watch_errcb(void *arg, u64 cookie, int err)
4409 struct rbd_device *rbd_dev = arg;
4411 rbd_warn(rbd_dev, "encountered watch error: %d", err);
4413 down_write(&rbd_dev->lock_rwsem);
4414 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4415 up_write(&rbd_dev->lock_rwsem);
4417 mutex_lock(&rbd_dev->watch_mutex);
4418 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED) {
4419 __rbd_unregister_watch(rbd_dev);
4420 rbd_dev->watch_state = RBD_WATCH_STATE_ERROR;
4422 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->watch_dwork, 0);
4424 mutex_unlock(&rbd_dev->watch_mutex);
4428 * watch_mutex must be locked
4430 static int __rbd_register_watch(struct rbd_device *rbd_dev)
4432 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4433 struct ceph_osd_linger_request *handle;
4435 rbd_assert(!rbd_dev->watch_handle);
4436 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4438 handle = ceph_osdc_watch(osdc, &rbd_dev->header_oid,
4439 &rbd_dev->header_oloc, rbd_watch_cb,
4440 rbd_watch_errcb, rbd_dev);
4442 return PTR_ERR(handle);
4444 rbd_dev->watch_handle = handle;
4449 * watch_mutex must be locked
4451 static void __rbd_unregister_watch(struct rbd_device *rbd_dev)
4453 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4456 rbd_assert(rbd_dev->watch_handle);
4457 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4459 ret = ceph_osdc_unwatch(osdc, rbd_dev->watch_handle);
4461 rbd_warn(rbd_dev, "failed to unwatch: %d", ret);
4463 rbd_dev->watch_handle = NULL;
4466 static int rbd_register_watch(struct rbd_device *rbd_dev)
4470 mutex_lock(&rbd_dev->watch_mutex);
4471 rbd_assert(rbd_dev->watch_state == RBD_WATCH_STATE_UNREGISTERED);
4472 ret = __rbd_register_watch(rbd_dev);
4476 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4477 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4480 mutex_unlock(&rbd_dev->watch_mutex);
4484 static void cancel_tasks_sync(struct rbd_device *rbd_dev)
4486 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4488 cancel_work_sync(&rbd_dev->acquired_lock_work);
4489 cancel_work_sync(&rbd_dev->released_lock_work);
4490 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
4491 cancel_work_sync(&rbd_dev->unlock_work);
4495 * header_rwsem must not be held to avoid a deadlock with
4496 * rbd_dev_refresh() when flushing notifies.
4498 static void rbd_unregister_watch(struct rbd_device *rbd_dev)
4500 cancel_tasks_sync(rbd_dev);
4502 mutex_lock(&rbd_dev->watch_mutex);
4503 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED)
4504 __rbd_unregister_watch(rbd_dev);
4505 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
4506 mutex_unlock(&rbd_dev->watch_mutex);
4508 cancel_delayed_work_sync(&rbd_dev->watch_dwork);
4509 ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
4513 * lock_rwsem must be held for write
4515 static void rbd_reacquire_lock(struct rbd_device *rbd_dev)
4517 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4521 if (!rbd_quiesce_lock(rbd_dev))
4524 format_lock_cookie(rbd_dev, cookie);
4525 ret = ceph_cls_set_cookie(osdc, &rbd_dev->header_oid,
4526 &rbd_dev->header_oloc, RBD_LOCK_NAME,
4527 CEPH_CLS_LOCK_EXCLUSIVE, rbd_dev->lock_cookie,
4528 RBD_LOCK_TAG, cookie);
4530 if (ret != -EOPNOTSUPP)
4531 rbd_warn(rbd_dev, "failed to update lock cookie: %d",
4535 * Lock cookie cannot be updated on older OSDs, so do
4536 * a manual release and queue an acquire.
4538 __rbd_release_lock(rbd_dev);
4539 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4541 __rbd_lock(rbd_dev, cookie);
4542 wake_lock_waiters(rbd_dev, 0);
4546 static void rbd_reregister_watch(struct work_struct *work)
4548 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4549 struct rbd_device, watch_dwork);
4552 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4554 mutex_lock(&rbd_dev->watch_mutex);
4555 if (rbd_dev->watch_state != RBD_WATCH_STATE_ERROR) {
4556 mutex_unlock(&rbd_dev->watch_mutex);
4560 ret = __rbd_register_watch(rbd_dev);
4562 rbd_warn(rbd_dev, "failed to reregister watch: %d", ret);
4563 if (ret != -EBLOCKLISTED && ret != -ENOENT) {
4564 queue_delayed_work(rbd_dev->task_wq,
4565 &rbd_dev->watch_dwork,
4567 mutex_unlock(&rbd_dev->watch_mutex);
4571 mutex_unlock(&rbd_dev->watch_mutex);
4572 down_write(&rbd_dev->lock_rwsem);
4573 wake_lock_waiters(rbd_dev, ret);
4574 up_write(&rbd_dev->lock_rwsem);
4578 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4579 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4580 mutex_unlock(&rbd_dev->watch_mutex);
4582 down_write(&rbd_dev->lock_rwsem);
4583 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED)
4584 rbd_reacquire_lock(rbd_dev);
4585 up_write(&rbd_dev->lock_rwsem);
4587 ret = rbd_dev_refresh(rbd_dev);
4589 rbd_warn(rbd_dev, "reregistration refresh failed: %d", ret);
4593 * Synchronous osd object method call. Returns the number of bytes
4594 * returned in the outbound buffer, or a negative error code.
4596 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
4597 struct ceph_object_id *oid,
4598 struct ceph_object_locator *oloc,
4599 const char *method_name,
4600 const void *outbound,
4601 size_t outbound_size,
4603 size_t inbound_size)
4605 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4606 struct page *req_page = NULL;
4607 struct page *reply_page;
4611 * Method calls are ultimately read operations. The result
4612 * should placed into the inbound buffer provided. They
4613 * also supply outbound data--parameters for the object
4614 * method. Currently if this is present it will be a
4618 if (outbound_size > PAGE_SIZE)
4621 req_page = alloc_page(GFP_KERNEL);
4625 memcpy(page_address(req_page), outbound, outbound_size);
4628 reply_page = alloc_page(GFP_KERNEL);
4631 __free_page(req_page);
4635 ret = ceph_osdc_call(osdc, oid, oloc, RBD_DRV_NAME, method_name,
4636 CEPH_OSD_FLAG_READ, req_page, outbound_size,
4637 &reply_page, &inbound_size);
4639 memcpy(inbound, page_address(reply_page), inbound_size);
4644 __free_page(req_page);
4645 __free_page(reply_page);
4649 static void rbd_queue_workfn(struct work_struct *work)
4651 struct rbd_img_request *img_request =
4652 container_of(work, struct rbd_img_request, work);
4653 struct rbd_device *rbd_dev = img_request->rbd_dev;
4654 enum obj_operation_type op_type = img_request->op_type;
4655 struct request *rq = blk_mq_rq_from_pdu(img_request);
4656 u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
4657 u64 length = blk_rq_bytes(rq);
4661 /* Ignore/skip any zero-length requests */
4663 dout("%s: zero-length request\n", __func__);
4665 goto err_img_request;
4668 blk_mq_start_request(rq);
4670 down_read(&rbd_dev->header_rwsem);
4671 mapping_size = rbd_dev->mapping.size;
4672 rbd_img_capture_header(img_request);
4673 up_read(&rbd_dev->header_rwsem);
4675 if (offset + length > mapping_size) {
4676 rbd_warn(rbd_dev, "beyond EOD (%llu~%llu > %llu)", offset,
4677 length, mapping_size);
4679 goto err_img_request;
4682 dout("%s rbd_dev %p img_req %p %s %llu~%llu\n", __func__, rbd_dev,
4683 img_request, obj_op_name(op_type), offset, length);
4685 if (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_ZEROOUT)
4686 result = rbd_img_fill_nodata(img_request, offset, length);
4688 result = rbd_img_fill_from_bio(img_request, offset, length,
4691 goto err_img_request;
4693 rbd_img_handle_request(img_request, 0);
4697 rbd_img_request_destroy(img_request);
4699 rbd_warn(rbd_dev, "%s %llx at %llx result %d",
4700 obj_op_name(op_type), length, offset, result);
4701 blk_mq_end_request(rq, errno_to_blk_status(result));
4704 static blk_status_t rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
4705 const struct blk_mq_queue_data *bd)
4707 struct rbd_device *rbd_dev = hctx->queue->queuedata;
4708 struct rbd_img_request *img_req = blk_mq_rq_to_pdu(bd->rq);
4709 enum obj_operation_type op_type;
4711 switch (req_op(bd->rq)) {
4712 case REQ_OP_DISCARD:
4713 op_type = OBJ_OP_DISCARD;
4715 case REQ_OP_WRITE_ZEROES:
4716 op_type = OBJ_OP_ZEROOUT;
4719 op_type = OBJ_OP_WRITE;
4722 op_type = OBJ_OP_READ;
4725 rbd_warn(rbd_dev, "unknown req_op %d", req_op(bd->rq));
4726 return BLK_STS_IOERR;
4729 rbd_img_request_init(img_req, rbd_dev, op_type);
4731 if (rbd_img_is_write(img_req)) {
4732 if (rbd_is_ro(rbd_dev)) {
4733 rbd_warn(rbd_dev, "%s on read-only mapping",
4734 obj_op_name(img_req->op_type));
4735 return BLK_STS_IOERR;
4737 rbd_assert(!rbd_is_snap(rbd_dev));
4740 INIT_WORK(&img_req->work, rbd_queue_workfn);
4741 queue_work(rbd_wq, &img_req->work);
4745 static void rbd_free_disk(struct rbd_device *rbd_dev)
4747 blk_cleanup_disk(rbd_dev->disk);
4748 blk_mq_free_tag_set(&rbd_dev->tag_set);
4749 rbd_dev->disk = NULL;
4752 static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
4753 struct ceph_object_id *oid,
4754 struct ceph_object_locator *oloc,
4755 void *buf, int buf_len)
4758 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4759 struct ceph_osd_request *req;
4760 struct page **pages;
4761 int num_pages = calc_pages_for(0, buf_len);
4764 req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_KERNEL);
4768 ceph_oid_copy(&req->r_base_oid, oid);
4769 ceph_oloc_copy(&req->r_base_oloc, oloc);
4770 req->r_flags = CEPH_OSD_FLAG_READ;
4772 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
4773 if (IS_ERR(pages)) {
4774 ret = PTR_ERR(pages);
4778 osd_req_op_extent_init(req, 0, CEPH_OSD_OP_READ, 0, buf_len, 0, 0);
4779 osd_req_op_extent_osd_data_pages(req, 0, pages, buf_len, 0, false,
4782 ret = ceph_osdc_alloc_messages(req, GFP_KERNEL);
4786 ceph_osdc_start_request(osdc, req, false);
4787 ret = ceph_osdc_wait_request(osdc, req);
4789 ceph_copy_from_page_vector(pages, buf, 0, ret);
4792 ceph_osdc_put_request(req);
4797 * Read the complete header for the given rbd device. On successful
4798 * return, the rbd_dev->header field will contain up-to-date
4799 * information about the image.
4801 static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev)
4803 struct rbd_image_header_ondisk *ondisk = NULL;
4810 * The complete header will include an array of its 64-bit
4811 * snapshot ids, followed by the names of those snapshots as
4812 * a contiguous block of NUL-terminated strings. Note that
4813 * the number of snapshots could change by the time we read
4814 * it in, in which case we re-read it.
4821 size = sizeof (*ondisk);
4822 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
4824 ondisk = kmalloc(size, GFP_KERNEL);
4828 ret = rbd_obj_read_sync(rbd_dev, &rbd_dev->header_oid,
4829 &rbd_dev->header_oloc, ondisk, size);
4832 if ((size_t)ret < size) {
4834 rbd_warn(rbd_dev, "short header read (want %zd got %d)",
4838 if (!rbd_dev_ondisk_valid(ondisk)) {
4840 rbd_warn(rbd_dev, "invalid header");
4844 names_size = le64_to_cpu(ondisk->snap_names_len);
4845 want_count = snap_count;
4846 snap_count = le32_to_cpu(ondisk->snap_count);
4847 } while (snap_count != want_count);
4849 ret = rbd_header_from_disk(rbd_dev, ondisk);
4856 static void rbd_dev_update_size(struct rbd_device *rbd_dev)
4861 * If EXISTS is not set, rbd_dev->disk may be NULL, so don't
4862 * try to update its size. If REMOVING is set, updating size
4863 * is just useless work since the device can't be opened.
4865 if (test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags) &&
4866 !test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) {
4867 size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
4868 dout("setting size to %llu sectors", (unsigned long long)size);
4869 set_capacity_and_notify(rbd_dev->disk, size);
4873 static int rbd_dev_refresh(struct rbd_device *rbd_dev)
4878 down_write(&rbd_dev->header_rwsem);
4879 mapping_size = rbd_dev->mapping.size;
4881 ret = rbd_dev_header_info(rbd_dev);
4886 * If there is a parent, see if it has disappeared due to the
4887 * mapped image getting flattened.
4889 if (rbd_dev->parent) {
4890 ret = rbd_dev_v2_parent_info(rbd_dev);
4895 rbd_assert(!rbd_is_snap(rbd_dev));
4896 rbd_dev->mapping.size = rbd_dev->header.image_size;
4899 up_write(&rbd_dev->header_rwsem);
4900 if (!ret && mapping_size != rbd_dev->mapping.size)
4901 rbd_dev_update_size(rbd_dev);
4906 static const struct blk_mq_ops rbd_mq_ops = {
4907 .queue_rq = rbd_queue_rq,
4910 static int rbd_init_disk(struct rbd_device *rbd_dev)
4912 struct gendisk *disk;
4913 struct request_queue *q;
4914 unsigned int objset_bytes =
4915 rbd_dev->layout.object_size * rbd_dev->layout.stripe_count;
4918 memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
4919 rbd_dev->tag_set.ops = &rbd_mq_ops;
4920 rbd_dev->tag_set.queue_depth = rbd_dev->opts->queue_depth;
4921 rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
4922 rbd_dev->tag_set.flags = BLK_MQ_F_SHOULD_MERGE;
4923 rbd_dev->tag_set.nr_hw_queues = num_present_cpus();
4924 rbd_dev->tag_set.cmd_size = sizeof(struct rbd_img_request);
4926 err = blk_mq_alloc_tag_set(&rbd_dev->tag_set);
4930 disk = blk_mq_alloc_disk(&rbd_dev->tag_set, rbd_dev);
4932 err = PTR_ERR(disk);
4937 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
4939 disk->major = rbd_dev->major;
4940 disk->first_minor = rbd_dev->minor;
4942 disk->minors = (1 << RBD_SINGLE_MAJOR_PART_SHIFT);
4943 disk->flags |= GENHD_FL_EXT_DEVT;
4945 disk->minors = RBD_MINORS_PER_MAJOR;
4947 disk->fops = &rbd_bd_ops;
4949 blk_queue_flag_set(QUEUE_FLAG_NONROT, q);
4950 /* QUEUE_FLAG_ADD_RANDOM is off by default for blk-mq */
4952 blk_queue_max_hw_sectors(q, objset_bytes >> SECTOR_SHIFT);
4953 q->limits.max_sectors = queue_max_hw_sectors(q);
4954 blk_queue_max_segments(q, USHRT_MAX);
4955 blk_queue_max_segment_size(q, UINT_MAX);
4956 blk_queue_io_min(q, rbd_dev->opts->alloc_size);
4957 blk_queue_io_opt(q, rbd_dev->opts->alloc_size);
4959 if (rbd_dev->opts->trim) {
4960 blk_queue_flag_set(QUEUE_FLAG_DISCARD, q);
4961 q->limits.discard_granularity = rbd_dev->opts->alloc_size;
4962 blk_queue_max_discard_sectors(q, objset_bytes >> SECTOR_SHIFT);
4963 blk_queue_max_write_zeroes_sectors(q, objset_bytes >> SECTOR_SHIFT);
4966 if (!ceph_test_opt(rbd_dev->rbd_client->client, NOCRC))
4967 blk_queue_flag_set(QUEUE_FLAG_STABLE_WRITES, q);
4969 rbd_dev->disk = disk;
4973 blk_mq_free_tag_set(&rbd_dev->tag_set);
4981 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
4983 return container_of(dev, struct rbd_device, dev);
4986 static ssize_t rbd_size_show(struct device *dev,
4987 struct device_attribute *attr, char *buf)
4989 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4991 return sprintf(buf, "%llu\n",
4992 (unsigned long long)rbd_dev->mapping.size);
4995 static ssize_t rbd_features_show(struct device *dev,
4996 struct device_attribute *attr, char *buf)
4998 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5000 return sprintf(buf, "0x%016llx\n", rbd_dev->header.features);
5003 static ssize_t rbd_major_show(struct device *dev,
5004 struct device_attribute *attr, char *buf)
5006 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5009 return sprintf(buf, "%d\n", rbd_dev->major);
5011 return sprintf(buf, "(none)\n");
5014 static ssize_t rbd_minor_show(struct device *dev,
5015 struct device_attribute *attr, char *buf)
5017 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5019 return sprintf(buf, "%d\n", rbd_dev->minor);
5022 static ssize_t rbd_client_addr_show(struct device *dev,
5023 struct device_attribute *attr, char *buf)
5025 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5026 struct ceph_entity_addr *client_addr =
5027 ceph_client_addr(rbd_dev->rbd_client->client);
5029 return sprintf(buf, "%pISpc/%u\n", &client_addr->in_addr,
5030 le32_to_cpu(client_addr->nonce));
5033 static ssize_t rbd_client_id_show(struct device *dev,
5034 struct device_attribute *attr, char *buf)
5036 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5038 return sprintf(buf, "client%lld\n",
5039 ceph_client_gid(rbd_dev->rbd_client->client));
5042 static ssize_t rbd_cluster_fsid_show(struct device *dev,
5043 struct device_attribute *attr, char *buf)
5045 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5047 return sprintf(buf, "%pU\n", &rbd_dev->rbd_client->client->fsid);
5050 static ssize_t rbd_config_info_show(struct device *dev,
5051 struct device_attribute *attr, char *buf)
5053 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5055 if (!capable(CAP_SYS_ADMIN))
5058 return sprintf(buf, "%s\n", rbd_dev->config_info);
5061 static ssize_t rbd_pool_show(struct device *dev,
5062 struct device_attribute *attr, char *buf)
5064 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5066 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
5069 static ssize_t rbd_pool_id_show(struct device *dev,
5070 struct device_attribute *attr, char *buf)
5072 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5074 return sprintf(buf, "%llu\n",
5075 (unsigned long long) rbd_dev->spec->pool_id);
5078 static ssize_t rbd_pool_ns_show(struct device *dev,
5079 struct device_attribute *attr, char *buf)
5081 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5083 return sprintf(buf, "%s\n", rbd_dev->spec->pool_ns ?: "");
5086 static ssize_t rbd_name_show(struct device *dev,
5087 struct device_attribute *attr, char *buf)
5089 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5091 if (rbd_dev->spec->image_name)
5092 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
5094 return sprintf(buf, "(unknown)\n");
5097 static ssize_t rbd_image_id_show(struct device *dev,
5098 struct device_attribute *attr, char *buf)
5100 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5102 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
5106 * Shows the name of the currently-mapped snapshot (or
5107 * RBD_SNAP_HEAD_NAME for the base image).
5109 static ssize_t rbd_snap_show(struct device *dev,
5110 struct device_attribute *attr,
5113 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5115 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
5118 static ssize_t rbd_snap_id_show(struct device *dev,
5119 struct device_attribute *attr, char *buf)
5121 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5123 return sprintf(buf, "%llu\n", rbd_dev->spec->snap_id);
5127 * For a v2 image, shows the chain of parent images, separated by empty
5128 * lines. For v1 images or if there is no parent, shows "(no parent
5131 static ssize_t rbd_parent_show(struct device *dev,
5132 struct device_attribute *attr,
5135 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5138 if (!rbd_dev->parent)
5139 return sprintf(buf, "(no parent image)\n");
5141 for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) {
5142 struct rbd_spec *spec = rbd_dev->parent_spec;
5144 count += sprintf(&buf[count], "%s"
5145 "pool_id %llu\npool_name %s\n"
5147 "image_id %s\nimage_name %s\n"
5148 "snap_id %llu\nsnap_name %s\n"
5150 !count ? "" : "\n", /* first? */
5151 spec->pool_id, spec->pool_name,
5152 spec->pool_ns ?: "",
5153 spec->image_id, spec->image_name ?: "(unknown)",
5154 spec->snap_id, spec->snap_name,
5155 rbd_dev->parent_overlap);
5161 static ssize_t rbd_image_refresh(struct device *dev,
5162 struct device_attribute *attr,
5166 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5169 if (!capable(CAP_SYS_ADMIN))
5172 ret = rbd_dev_refresh(rbd_dev);
5179 static DEVICE_ATTR(size, 0444, rbd_size_show, NULL);
5180 static DEVICE_ATTR(features, 0444, rbd_features_show, NULL);
5181 static DEVICE_ATTR(major, 0444, rbd_major_show, NULL);
5182 static DEVICE_ATTR(minor, 0444, rbd_minor_show, NULL);
5183 static DEVICE_ATTR(client_addr, 0444, rbd_client_addr_show, NULL);
5184 static DEVICE_ATTR(client_id, 0444, rbd_client_id_show, NULL);
5185 static DEVICE_ATTR(cluster_fsid, 0444, rbd_cluster_fsid_show, NULL);
5186 static DEVICE_ATTR(config_info, 0400, rbd_config_info_show, NULL);
5187 static DEVICE_ATTR(pool, 0444, rbd_pool_show, NULL);
5188 static DEVICE_ATTR(pool_id, 0444, rbd_pool_id_show, NULL);
5189 static DEVICE_ATTR(pool_ns, 0444, rbd_pool_ns_show, NULL);
5190 static DEVICE_ATTR(name, 0444, rbd_name_show, NULL);
5191 static DEVICE_ATTR(image_id, 0444, rbd_image_id_show, NULL);
5192 static DEVICE_ATTR(refresh, 0200, NULL, rbd_image_refresh);
5193 static DEVICE_ATTR(current_snap, 0444, rbd_snap_show, NULL);
5194 static DEVICE_ATTR(snap_id, 0444, rbd_snap_id_show, NULL);
5195 static DEVICE_ATTR(parent, 0444, rbd_parent_show, NULL);
5197 static struct attribute *rbd_attrs[] = {
5198 &dev_attr_size.attr,
5199 &dev_attr_features.attr,
5200 &dev_attr_major.attr,
5201 &dev_attr_minor.attr,
5202 &dev_attr_client_addr.attr,
5203 &dev_attr_client_id.attr,
5204 &dev_attr_cluster_fsid.attr,
5205 &dev_attr_config_info.attr,
5206 &dev_attr_pool.attr,
5207 &dev_attr_pool_id.attr,
5208 &dev_attr_pool_ns.attr,
5209 &dev_attr_name.attr,
5210 &dev_attr_image_id.attr,
5211 &dev_attr_current_snap.attr,
5212 &dev_attr_snap_id.attr,
5213 &dev_attr_parent.attr,
5214 &dev_attr_refresh.attr,
5218 static struct attribute_group rbd_attr_group = {
5222 static const struct attribute_group *rbd_attr_groups[] = {
5227 static void rbd_dev_release(struct device *dev);
5229 static const struct device_type rbd_device_type = {
5231 .groups = rbd_attr_groups,
5232 .release = rbd_dev_release,
5235 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
5237 kref_get(&spec->kref);
5242 static void rbd_spec_free(struct kref *kref);
5243 static void rbd_spec_put(struct rbd_spec *spec)
5246 kref_put(&spec->kref, rbd_spec_free);
5249 static struct rbd_spec *rbd_spec_alloc(void)
5251 struct rbd_spec *spec;
5253 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
5257 spec->pool_id = CEPH_NOPOOL;
5258 spec->snap_id = CEPH_NOSNAP;
5259 kref_init(&spec->kref);
5264 static void rbd_spec_free(struct kref *kref)
5266 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
5268 kfree(spec->pool_name);
5269 kfree(spec->pool_ns);
5270 kfree(spec->image_id);
5271 kfree(spec->image_name);
5272 kfree(spec->snap_name);
5276 static void rbd_dev_free(struct rbd_device *rbd_dev)
5278 WARN_ON(rbd_dev->watch_state != RBD_WATCH_STATE_UNREGISTERED);
5279 WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_UNLOCKED);
5281 ceph_oid_destroy(&rbd_dev->header_oid);
5282 ceph_oloc_destroy(&rbd_dev->header_oloc);
5283 kfree(rbd_dev->config_info);
5285 rbd_put_client(rbd_dev->rbd_client);
5286 rbd_spec_put(rbd_dev->spec);
5287 kfree(rbd_dev->opts);
5291 static void rbd_dev_release(struct device *dev)
5293 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5294 bool need_put = !!rbd_dev->opts;
5297 destroy_workqueue(rbd_dev->task_wq);
5298 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5301 rbd_dev_free(rbd_dev);
5304 * This is racy, but way better than putting module outside of
5305 * the release callback. The race window is pretty small, so
5306 * doing something similar to dm (dm-builtin.c) is overkill.
5309 module_put(THIS_MODULE);
5312 static struct rbd_device *__rbd_dev_create(struct rbd_client *rbdc,
5313 struct rbd_spec *spec)
5315 struct rbd_device *rbd_dev;
5317 rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
5321 spin_lock_init(&rbd_dev->lock);
5322 INIT_LIST_HEAD(&rbd_dev->node);
5323 init_rwsem(&rbd_dev->header_rwsem);
5325 rbd_dev->header.data_pool_id = CEPH_NOPOOL;
5326 ceph_oid_init(&rbd_dev->header_oid);
5327 rbd_dev->header_oloc.pool = spec->pool_id;
5328 if (spec->pool_ns) {
5329 WARN_ON(!*spec->pool_ns);
5330 rbd_dev->header_oloc.pool_ns =
5331 ceph_find_or_create_string(spec->pool_ns,
5332 strlen(spec->pool_ns));
5335 mutex_init(&rbd_dev->watch_mutex);
5336 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
5337 INIT_DELAYED_WORK(&rbd_dev->watch_dwork, rbd_reregister_watch);
5339 init_rwsem(&rbd_dev->lock_rwsem);
5340 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
5341 INIT_WORK(&rbd_dev->acquired_lock_work, rbd_notify_acquired_lock);
5342 INIT_WORK(&rbd_dev->released_lock_work, rbd_notify_released_lock);
5343 INIT_DELAYED_WORK(&rbd_dev->lock_dwork, rbd_acquire_lock);
5344 INIT_WORK(&rbd_dev->unlock_work, rbd_release_lock_work);
5345 spin_lock_init(&rbd_dev->lock_lists_lock);
5346 INIT_LIST_HEAD(&rbd_dev->acquiring_list);
5347 INIT_LIST_HEAD(&rbd_dev->running_list);
5348 init_completion(&rbd_dev->acquire_wait);
5349 init_completion(&rbd_dev->releasing_wait);
5351 spin_lock_init(&rbd_dev->object_map_lock);
5353 rbd_dev->dev.bus = &rbd_bus_type;
5354 rbd_dev->dev.type = &rbd_device_type;
5355 rbd_dev->dev.parent = &rbd_root_dev;
5356 device_initialize(&rbd_dev->dev);
5358 rbd_dev->rbd_client = rbdc;
5359 rbd_dev->spec = spec;
5365 * Create a mapping rbd_dev.
5367 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
5368 struct rbd_spec *spec,
5369 struct rbd_options *opts)
5371 struct rbd_device *rbd_dev;
5373 rbd_dev = __rbd_dev_create(rbdc, spec);
5377 rbd_dev->opts = opts;
5379 /* get an id and fill in device name */
5380 rbd_dev->dev_id = ida_simple_get(&rbd_dev_id_ida, 0,
5381 minor_to_rbd_dev_id(1 << MINORBITS),
5383 if (rbd_dev->dev_id < 0)
5386 sprintf(rbd_dev->name, RBD_DRV_NAME "%d", rbd_dev->dev_id);
5387 rbd_dev->task_wq = alloc_ordered_workqueue("%s-tasks", WQ_MEM_RECLAIM,
5389 if (!rbd_dev->task_wq)
5392 /* we have a ref from do_rbd_add() */
5393 __module_get(THIS_MODULE);
5395 dout("%s rbd_dev %p dev_id %d\n", __func__, rbd_dev, rbd_dev->dev_id);
5399 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5401 rbd_dev_free(rbd_dev);
5405 static void rbd_dev_destroy(struct rbd_device *rbd_dev)
5408 put_device(&rbd_dev->dev);
5412 * Get the size and object order for an image snapshot, or if
5413 * snap_id is CEPH_NOSNAP, gets this information for the base
5416 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
5417 u8 *order, u64 *snap_size)
5419 __le64 snapid = cpu_to_le64(snap_id);
5424 } __attribute__ ((packed)) size_buf = { 0 };
5426 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5427 &rbd_dev->header_oloc, "get_size",
5428 &snapid, sizeof(snapid),
5429 &size_buf, sizeof(size_buf));
5430 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5433 if (ret < sizeof (size_buf))
5437 *order = size_buf.order;
5438 dout(" order %u", (unsigned int)*order);
5440 *snap_size = le64_to_cpu(size_buf.size);
5442 dout(" snap_id 0x%016llx snap_size = %llu\n",
5443 (unsigned long long)snap_id,
5444 (unsigned long long)*snap_size);
5449 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
5451 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
5452 &rbd_dev->header.obj_order,
5453 &rbd_dev->header.image_size);
5456 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
5463 /* Response will be an encoded string, which includes a length */
5464 size = sizeof(__le32) + RBD_OBJ_PREFIX_LEN_MAX;
5465 reply_buf = kzalloc(size, GFP_KERNEL);
5469 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5470 &rbd_dev->header_oloc, "get_object_prefix",
5471 NULL, 0, reply_buf, size);
5472 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5477 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
5478 p + ret, NULL, GFP_NOIO);
5481 if (IS_ERR(rbd_dev->header.object_prefix)) {
5482 ret = PTR_ERR(rbd_dev->header.object_prefix);
5483 rbd_dev->header.object_prefix = NULL;
5485 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix);
5493 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
5494 bool read_only, u64 *snap_features)
5503 } __attribute__ ((packed)) features_buf = { 0 };
5507 features_in.snap_id = cpu_to_le64(snap_id);
5508 features_in.read_only = read_only;
5510 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5511 &rbd_dev->header_oloc, "get_features",
5512 &features_in, sizeof(features_in),
5513 &features_buf, sizeof(features_buf));
5514 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5517 if (ret < sizeof (features_buf))
5520 unsup = le64_to_cpu(features_buf.incompat) & ~RBD_FEATURES_SUPPORTED;
5522 rbd_warn(rbd_dev, "image uses unsupported features: 0x%llx",
5527 *snap_features = le64_to_cpu(features_buf.features);
5529 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
5530 (unsigned long long)snap_id,
5531 (unsigned long long)*snap_features,
5532 (unsigned long long)le64_to_cpu(features_buf.incompat));
5537 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
5539 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
5541 &rbd_dev->header.features);
5545 * These are generic image flags, but since they are used only for
5546 * object map, store them in rbd_dev->object_map_flags.
5548 * For the same reason, this function is called only on object map
5549 * (re)load and not on header refresh.
5551 static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev)
5553 __le64 snapid = cpu_to_le64(rbd_dev->spec->snap_id);
5557 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5558 &rbd_dev->header_oloc, "get_flags",
5559 &snapid, sizeof(snapid),
5560 &flags, sizeof(flags));
5563 if (ret < sizeof(flags))
5566 rbd_dev->object_map_flags = le64_to_cpu(flags);
5570 struct parent_image_info {
5572 const char *pool_ns;
5573 const char *image_id;
5581 * The caller is responsible for @pii.
5583 static int decode_parent_image_spec(void **p, void *end,
5584 struct parent_image_info *pii)
5590 ret = ceph_start_decoding(p, end, 1, "ParentImageSpec",
5591 &struct_v, &struct_len);
5595 ceph_decode_64_safe(p, end, pii->pool_id, e_inval);
5596 pii->pool_ns = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5597 if (IS_ERR(pii->pool_ns)) {
5598 ret = PTR_ERR(pii->pool_ns);
5599 pii->pool_ns = NULL;
5602 pii->image_id = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5603 if (IS_ERR(pii->image_id)) {
5604 ret = PTR_ERR(pii->image_id);
5605 pii->image_id = NULL;
5608 ceph_decode_64_safe(p, end, pii->snap_id, e_inval);
5615 static int __get_parent_info(struct rbd_device *rbd_dev,
5616 struct page *req_page,
5617 struct page *reply_page,
5618 struct parent_image_info *pii)
5620 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5621 size_t reply_len = PAGE_SIZE;
5625 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5626 "rbd", "parent_get", CEPH_OSD_FLAG_READ,
5627 req_page, sizeof(u64), &reply_page, &reply_len);
5629 return ret == -EOPNOTSUPP ? 1 : ret;
5631 p = page_address(reply_page);
5632 end = p + reply_len;
5633 ret = decode_parent_image_spec(&p, end, pii);
5637 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5638 "rbd", "parent_overlap_get", CEPH_OSD_FLAG_READ,
5639 req_page, sizeof(u64), &reply_page, &reply_len);
5643 p = page_address(reply_page);
5644 end = p + reply_len;
5645 ceph_decode_8_safe(&p, end, pii->has_overlap, e_inval);
5646 if (pii->has_overlap)
5647 ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5656 * The caller is responsible for @pii.
5658 static int __get_parent_info_legacy(struct rbd_device *rbd_dev,
5659 struct page *req_page,
5660 struct page *reply_page,
5661 struct parent_image_info *pii)
5663 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5664 size_t reply_len = PAGE_SIZE;
5668 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5669 "rbd", "get_parent", CEPH_OSD_FLAG_READ,
5670 req_page, sizeof(u64), &reply_page, &reply_len);
5674 p = page_address(reply_page);
5675 end = p + reply_len;
5676 ceph_decode_64_safe(&p, end, pii->pool_id, e_inval);
5677 pii->image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
5678 if (IS_ERR(pii->image_id)) {
5679 ret = PTR_ERR(pii->image_id);
5680 pii->image_id = NULL;
5683 ceph_decode_64_safe(&p, end, pii->snap_id, e_inval);
5684 pii->has_overlap = true;
5685 ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5693 static int get_parent_info(struct rbd_device *rbd_dev,
5694 struct parent_image_info *pii)
5696 struct page *req_page, *reply_page;
5700 req_page = alloc_page(GFP_KERNEL);
5704 reply_page = alloc_page(GFP_KERNEL);
5706 __free_page(req_page);
5710 p = page_address(req_page);
5711 ceph_encode_64(&p, rbd_dev->spec->snap_id);
5712 ret = __get_parent_info(rbd_dev, req_page, reply_page, pii);
5714 ret = __get_parent_info_legacy(rbd_dev, req_page, reply_page,
5717 __free_page(req_page);
5718 __free_page(reply_page);
5722 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
5724 struct rbd_spec *parent_spec;
5725 struct parent_image_info pii = { 0 };
5728 parent_spec = rbd_spec_alloc();
5732 ret = get_parent_info(rbd_dev, &pii);
5736 dout("%s pool_id %llu pool_ns %s image_id %s snap_id %llu has_overlap %d overlap %llu\n",
5737 __func__, pii.pool_id, pii.pool_ns, pii.image_id, pii.snap_id,
5738 pii.has_overlap, pii.overlap);
5740 if (pii.pool_id == CEPH_NOPOOL || !pii.has_overlap) {
5742 * Either the parent never existed, or we have
5743 * record of it but the image got flattened so it no
5744 * longer has a parent. When the parent of a
5745 * layered image disappears we immediately set the
5746 * overlap to 0. The effect of this is that all new
5747 * requests will be treated as if the image had no
5750 * If !pii.has_overlap, the parent image spec is not
5751 * applicable. It's there to avoid duplication in each
5754 if (rbd_dev->parent_overlap) {
5755 rbd_dev->parent_overlap = 0;
5756 rbd_dev_parent_put(rbd_dev);
5757 pr_info("%s: clone image has been flattened\n",
5758 rbd_dev->disk->disk_name);
5761 goto out; /* No parent? No problem. */
5764 /* The ceph file layout needs to fit pool id in 32 bits */
5767 if (pii.pool_id > (u64)U32_MAX) {
5768 rbd_warn(NULL, "parent pool id too large (%llu > %u)",
5769 (unsigned long long)pii.pool_id, U32_MAX);
5774 * The parent won't change (except when the clone is
5775 * flattened, already handled that). So we only need to
5776 * record the parent spec we have not already done so.
5778 if (!rbd_dev->parent_spec) {
5779 parent_spec->pool_id = pii.pool_id;
5780 if (pii.pool_ns && *pii.pool_ns) {
5781 parent_spec->pool_ns = pii.pool_ns;
5784 parent_spec->image_id = pii.image_id;
5785 pii.image_id = NULL;
5786 parent_spec->snap_id = pii.snap_id;
5788 rbd_dev->parent_spec = parent_spec;
5789 parent_spec = NULL; /* rbd_dev now owns this */
5793 * We always update the parent overlap. If it's zero we issue
5794 * a warning, as we will proceed as if there was no parent.
5798 /* refresh, careful to warn just once */
5799 if (rbd_dev->parent_overlap)
5801 "clone now standalone (overlap became 0)");
5804 rbd_warn(rbd_dev, "clone is standalone (overlap 0)");
5807 rbd_dev->parent_overlap = pii.overlap;
5813 kfree(pii.image_id);
5814 rbd_spec_put(parent_spec);
5818 static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev)
5822 __le64 stripe_count;
5823 } __attribute__ ((packed)) striping_info_buf = { 0 };
5824 size_t size = sizeof (striping_info_buf);
5828 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5829 &rbd_dev->header_oloc, "get_stripe_unit_count",
5830 NULL, 0, &striping_info_buf, size);
5831 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5837 p = &striping_info_buf;
5838 rbd_dev->header.stripe_unit = ceph_decode_64(&p);
5839 rbd_dev->header.stripe_count = ceph_decode_64(&p);
5843 static int rbd_dev_v2_data_pool(struct rbd_device *rbd_dev)
5845 __le64 data_pool_id;
5848 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5849 &rbd_dev->header_oloc, "get_data_pool",
5850 NULL, 0, &data_pool_id, sizeof(data_pool_id));
5853 if (ret < sizeof(data_pool_id))
5856 rbd_dev->header.data_pool_id = le64_to_cpu(data_pool_id);
5857 WARN_ON(rbd_dev->header.data_pool_id == CEPH_NOPOOL);
5861 static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
5863 CEPH_DEFINE_OID_ONSTACK(oid);
5864 size_t image_id_size;
5869 void *reply_buf = NULL;
5871 char *image_name = NULL;
5874 rbd_assert(!rbd_dev->spec->image_name);
5876 len = strlen(rbd_dev->spec->image_id);
5877 image_id_size = sizeof (__le32) + len;
5878 image_id = kmalloc(image_id_size, GFP_KERNEL);
5883 end = image_id + image_id_size;
5884 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len);
5886 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
5887 reply_buf = kmalloc(size, GFP_KERNEL);
5891 ceph_oid_printf(&oid, "%s", RBD_DIRECTORY);
5892 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
5893 "dir_get_name", image_id, image_id_size,
5898 end = reply_buf + ret;
5900 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
5901 if (IS_ERR(image_name))
5904 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
5912 static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5914 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5915 const char *snap_name;
5918 /* Skip over names until we find the one we are looking for */
5920 snap_name = rbd_dev->header.snap_names;
5921 while (which < snapc->num_snaps) {
5922 if (!strcmp(name, snap_name))
5923 return snapc->snaps[which];
5924 snap_name += strlen(snap_name) + 1;
5930 static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5932 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5937 for (which = 0; !found && which < snapc->num_snaps; which++) {
5938 const char *snap_name;
5940 snap_id = snapc->snaps[which];
5941 snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
5942 if (IS_ERR(snap_name)) {
5943 /* ignore no-longer existing snapshots */
5944 if (PTR_ERR(snap_name) == -ENOENT)
5949 found = !strcmp(name, snap_name);
5952 return found ? snap_id : CEPH_NOSNAP;
5956 * Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if
5957 * no snapshot by that name is found, or if an error occurs.
5959 static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5961 if (rbd_dev->image_format == 1)
5962 return rbd_v1_snap_id_by_name(rbd_dev, name);
5964 return rbd_v2_snap_id_by_name(rbd_dev, name);
5968 * An image being mapped will have everything but the snap id.
5970 static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev)
5972 struct rbd_spec *spec = rbd_dev->spec;
5974 rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name);
5975 rbd_assert(spec->image_id && spec->image_name);
5976 rbd_assert(spec->snap_name);
5978 if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) {
5981 snap_id = rbd_snap_id_by_name(rbd_dev, spec->snap_name);
5982 if (snap_id == CEPH_NOSNAP)
5985 spec->snap_id = snap_id;
5987 spec->snap_id = CEPH_NOSNAP;
5994 * A parent image will have all ids but none of the names.
5996 * All names in an rbd spec are dynamically allocated. It's OK if we
5997 * can't figure out the name for an image id.
5999 static int rbd_spec_fill_names(struct rbd_device *rbd_dev)
6001 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
6002 struct rbd_spec *spec = rbd_dev->spec;
6003 const char *pool_name;
6004 const char *image_name;
6005 const char *snap_name;
6008 rbd_assert(spec->pool_id != CEPH_NOPOOL);
6009 rbd_assert(spec->image_id);
6010 rbd_assert(spec->snap_id != CEPH_NOSNAP);
6012 /* Get the pool name; we have to make our own copy of this */
6014 pool_name = ceph_pg_pool_name_by_id(osdc->osdmap, spec->pool_id);
6016 rbd_warn(rbd_dev, "no pool with id %llu", spec->pool_id);
6019 pool_name = kstrdup(pool_name, GFP_KERNEL);
6023 /* Fetch the image name; tolerate failure here */
6025 image_name = rbd_dev_image_name(rbd_dev);
6027 rbd_warn(rbd_dev, "unable to get image name");
6029 /* Fetch the snapshot name */
6031 snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
6032 if (IS_ERR(snap_name)) {
6033 ret = PTR_ERR(snap_name);
6037 spec->pool_name = pool_name;
6038 spec->image_name = image_name;
6039 spec->snap_name = snap_name;
6049 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev)
6058 struct ceph_snap_context *snapc;
6062 * We'll need room for the seq value (maximum snapshot id),
6063 * snapshot count, and array of that many snapshot ids.
6064 * For now we have a fixed upper limit on the number we're
6065 * prepared to receive.
6067 size = sizeof (__le64) + sizeof (__le32) +
6068 RBD_MAX_SNAP_COUNT * sizeof (__le64);
6069 reply_buf = kzalloc(size, GFP_KERNEL);
6073 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6074 &rbd_dev->header_oloc, "get_snapcontext",
6075 NULL, 0, reply_buf, size);
6076 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6081 end = reply_buf + ret;
6083 ceph_decode_64_safe(&p, end, seq, out);
6084 ceph_decode_32_safe(&p, end, snap_count, out);
6087 * Make sure the reported number of snapshot ids wouldn't go
6088 * beyond the end of our buffer. But before checking that,
6089 * make sure the computed size of the snapshot context we
6090 * allocate is representable in a size_t.
6092 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
6097 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
6101 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
6107 for (i = 0; i < snap_count; i++)
6108 snapc->snaps[i] = ceph_decode_64(&p);
6110 ceph_put_snap_context(rbd_dev->header.snapc);
6111 rbd_dev->header.snapc = snapc;
6113 dout(" snap context seq = %llu, snap_count = %u\n",
6114 (unsigned long long)seq, (unsigned int)snap_count);
6121 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
6132 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
6133 reply_buf = kmalloc(size, GFP_KERNEL);
6135 return ERR_PTR(-ENOMEM);
6137 snapid = cpu_to_le64(snap_id);
6138 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6139 &rbd_dev->header_oloc, "get_snapshot_name",
6140 &snapid, sizeof(snapid), reply_buf, size);
6141 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6143 snap_name = ERR_PTR(ret);
6148 end = reply_buf + ret;
6149 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
6150 if (IS_ERR(snap_name))
6153 dout(" snap_id 0x%016llx snap_name = %s\n",
6154 (unsigned long long)snap_id, snap_name);
6161 static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev)
6163 bool first_time = rbd_dev->header.object_prefix == NULL;
6166 ret = rbd_dev_v2_image_size(rbd_dev);
6171 ret = rbd_dev_v2_header_onetime(rbd_dev);
6176 ret = rbd_dev_v2_snap_context(rbd_dev);
6177 if (ret && first_time) {
6178 kfree(rbd_dev->header.object_prefix);
6179 rbd_dev->header.object_prefix = NULL;
6185 static int rbd_dev_header_info(struct rbd_device *rbd_dev)
6187 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6189 if (rbd_dev->image_format == 1)
6190 return rbd_dev_v1_header_info(rbd_dev);
6192 return rbd_dev_v2_header_info(rbd_dev);
6196 * Skips over white space at *buf, and updates *buf to point to the
6197 * first found non-space character (if any). Returns the length of
6198 * the token (string of non-white space characters) found. Note
6199 * that *buf must be terminated with '\0'.
6201 static inline size_t next_token(const char **buf)
6204 * These are the characters that produce nonzero for
6205 * isspace() in the "C" and "POSIX" locales.
6207 const char *spaces = " \f\n\r\t\v";
6209 *buf += strspn(*buf, spaces); /* Find start of token */
6211 return strcspn(*buf, spaces); /* Return token length */
6215 * Finds the next token in *buf, dynamically allocates a buffer big
6216 * enough to hold a copy of it, and copies the token into the new
6217 * buffer. The copy is guaranteed to be terminated with '\0'. Note
6218 * that a duplicate buffer is created even for a zero-length token.
6220 * Returns a pointer to the newly-allocated duplicate, or a null
6221 * pointer if memory for the duplicate was not available. If
6222 * the lenp argument is a non-null pointer, the length of the token
6223 * (not including the '\0') is returned in *lenp.
6225 * If successful, the *buf pointer will be updated to point beyond
6226 * the end of the found token.
6228 * Note: uses GFP_KERNEL for allocation.
6230 static inline char *dup_token(const char **buf, size_t *lenp)
6235 len = next_token(buf);
6236 dup = kmemdup(*buf, len + 1, GFP_KERNEL);
6239 *(dup + len) = '\0';
6248 static int rbd_parse_param(struct fs_parameter *param,
6249 struct rbd_parse_opts_ctx *pctx)
6251 struct rbd_options *opt = pctx->opts;
6252 struct fs_parse_result result;
6253 struct p_log log = {.prefix = "rbd"};
6256 ret = ceph_parse_param(param, pctx->copts, NULL);
6257 if (ret != -ENOPARAM)
6260 token = __fs_parse(&log, rbd_parameters, param, &result);
6261 dout("%s fs_parse '%s' token %d\n", __func__, param->key, token);
6263 if (token == -ENOPARAM)
6264 return inval_plog(&log, "Unknown parameter '%s'",
6270 case Opt_queue_depth:
6271 if (result.uint_32 < 1)
6273 opt->queue_depth = result.uint_32;
6275 case Opt_alloc_size:
6276 if (result.uint_32 < SECTOR_SIZE)
6278 if (!is_power_of_2(result.uint_32))
6279 return inval_plog(&log, "alloc_size must be a power of 2");
6280 opt->alloc_size = result.uint_32;
6282 case Opt_lock_timeout:
6283 /* 0 is "wait forever" (i.e. infinite timeout) */
6284 if (result.uint_32 > INT_MAX / 1000)
6286 opt->lock_timeout = msecs_to_jiffies(result.uint_32 * 1000);
6289 kfree(pctx->spec->pool_ns);
6290 pctx->spec->pool_ns = param->string;
6291 param->string = NULL;
6293 case Opt_compression_hint:
6294 switch (result.uint_32) {
6295 case Opt_compression_hint_none:
6296 opt->alloc_hint_flags &=
6297 ~(CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE |
6298 CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE);
6300 case Opt_compression_hint_compressible:
6301 opt->alloc_hint_flags |=
6302 CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6303 opt->alloc_hint_flags &=
6304 ~CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6306 case Opt_compression_hint_incompressible:
6307 opt->alloc_hint_flags |=
6308 CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6309 opt->alloc_hint_flags &=
6310 ~CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6317 opt->read_only = true;
6319 case Opt_read_write:
6320 opt->read_only = false;
6322 case Opt_lock_on_read:
6323 opt->lock_on_read = true;
6326 opt->exclusive = true;
6338 return inval_plog(&log, "%s out of range", param->key);
6342 * This duplicates most of generic_parse_monolithic(), untying it from
6343 * fs_context and skipping standard superblock and security options.
6345 static int rbd_parse_options(char *options, struct rbd_parse_opts_ctx *pctx)
6350 dout("%s '%s'\n", __func__, options);
6351 while ((key = strsep(&options, ",")) != NULL) {
6353 struct fs_parameter param = {
6355 .type = fs_value_is_flag,
6357 char *value = strchr(key, '=');
6364 v_len = strlen(value);
6365 param.string = kmemdup_nul(value, v_len,
6369 param.type = fs_value_is_string;
6373 ret = rbd_parse_param(¶m, pctx);
6374 kfree(param.string);
6384 * Parse the options provided for an "rbd add" (i.e., rbd image
6385 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
6386 * and the data written is passed here via a NUL-terminated buffer.
6387 * Returns 0 if successful or an error code otherwise.
6389 * The information extracted from these options is recorded in
6390 * the other parameters which return dynamically-allocated
6393 * The address of a pointer that will refer to a ceph options
6394 * structure. Caller must release the returned pointer using
6395 * ceph_destroy_options() when it is no longer needed.
6397 * Address of an rbd options pointer. Fully initialized by
6398 * this function; caller must release with kfree().
6400 * Address of an rbd image specification pointer. Fully
6401 * initialized by this function based on parsed options.
6402 * Caller must release with rbd_spec_put().
6404 * The options passed take this form:
6405 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
6408 * A comma-separated list of one or more monitor addresses.
6409 * A monitor address is an ip address, optionally followed
6410 * by a port number (separated by a colon).
6411 * I.e.: ip1[:port1][,ip2[:port2]...]
6413 * A comma-separated list of ceph and/or rbd options.
6415 * The name of the rados pool containing the rbd image.
6417 * The name of the image in that pool to map.
6419 * An optional snapshot id. If provided, the mapping will
6420 * present data from the image at the time that snapshot was
6421 * created. The image head is used if no snapshot id is
6422 * provided. Snapshot mappings are always read-only.
6424 static int rbd_add_parse_args(const char *buf,
6425 struct ceph_options **ceph_opts,
6426 struct rbd_options **opts,
6427 struct rbd_spec **rbd_spec)
6431 const char *mon_addrs;
6433 size_t mon_addrs_size;
6434 struct rbd_parse_opts_ctx pctx = { 0 };
6437 /* The first four tokens are required */
6439 len = next_token(&buf);
6441 rbd_warn(NULL, "no monitor address(es) provided");
6445 mon_addrs_size = len;
6449 options = dup_token(&buf, NULL);
6453 rbd_warn(NULL, "no options provided");
6457 pctx.spec = rbd_spec_alloc();
6461 pctx.spec->pool_name = dup_token(&buf, NULL);
6462 if (!pctx.spec->pool_name)
6464 if (!*pctx.spec->pool_name) {
6465 rbd_warn(NULL, "no pool name provided");
6469 pctx.spec->image_name = dup_token(&buf, NULL);
6470 if (!pctx.spec->image_name)
6472 if (!*pctx.spec->image_name) {
6473 rbd_warn(NULL, "no image name provided");
6478 * Snapshot name is optional; default is to use "-"
6479 * (indicating the head/no snapshot).
6481 len = next_token(&buf);
6483 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
6484 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
6485 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
6486 ret = -ENAMETOOLONG;
6489 snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
6492 *(snap_name + len) = '\0';
6493 pctx.spec->snap_name = snap_name;
6495 pctx.copts = ceph_alloc_options();
6499 /* Initialize all rbd options to the defaults */
6501 pctx.opts = kzalloc(sizeof(*pctx.opts), GFP_KERNEL);
6505 pctx.opts->read_only = RBD_READ_ONLY_DEFAULT;
6506 pctx.opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
6507 pctx.opts->alloc_size = RBD_ALLOC_SIZE_DEFAULT;
6508 pctx.opts->lock_timeout = RBD_LOCK_TIMEOUT_DEFAULT;
6509 pctx.opts->lock_on_read = RBD_LOCK_ON_READ_DEFAULT;
6510 pctx.opts->exclusive = RBD_EXCLUSIVE_DEFAULT;
6511 pctx.opts->trim = RBD_TRIM_DEFAULT;
6513 ret = ceph_parse_mon_ips(mon_addrs, mon_addrs_size, pctx.copts, NULL);
6517 ret = rbd_parse_options(options, &pctx);
6521 *ceph_opts = pctx.copts;
6523 *rbd_spec = pctx.spec;
6531 ceph_destroy_options(pctx.copts);
6532 rbd_spec_put(pctx.spec);
6537 static void rbd_dev_image_unlock(struct rbd_device *rbd_dev)
6539 down_write(&rbd_dev->lock_rwsem);
6540 if (__rbd_is_lock_owner(rbd_dev))
6541 __rbd_release_lock(rbd_dev);
6542 up_write(&rbd_dev->lock_rwsem);
6546 * If the wait is interrupted, an error is returned even if the lock
6547 * was successfully acquired. rbd_dev_image_unlock() will release it
6550 static int rbd_add_acquire_lock(struct rbd_device *rbd_dev)
6554 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK)) {
6555 if (!rbd_dev->opts->exclusive && !rbd_dev->opts->lock_on_read)
6558 rbd_warn(rbd_dev, "exclusive-lock feature is not enabled");
6562 if (rbd_is_ro(rbd_dev))
6565 rbd_assert(!rbd_is_lock_owner(rbd_dev));
6566 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
6567 ret = wait_for_completion_killable_timeout(&rbd_dev->acquire_wait,
6568 ceph_timeout_jiffies(rbd_dev->opts->lock_timeout));
6570 ret = rbd_dev->acquire_err;
6572 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
6578 rbd_warn(rbd_dev, "failed to acquire exclusive lock: %ld", ret);
6583 * The lock may have been released by now, unless automatic lock
6584 * transitions are disabled.
6586 rbd_assert(!rbd_dev->opts->exclusive || rbd_is_lock_owner(rbd_dev));
6591 * An rbd format 2 image has a unique identifier, distinct from the
6592 * name given to it by the user. Internally, that identifier is
6593 * what's used to specify the names of objects related to the image.
6595 * A special "rbd id" object is used to map an rbd image name to its
6596 * id. If that object doesn't exist, then there is no v2 rbd image
6597 * with the supplied name.
6599 * This function will record the given rbd_dev's image_id field if
6600 * it can be determined, and in that case will return 0. If any
6601 * errors occur a negative errno will be returned and the rbd_dev's
6602 * image_id field will be unchanged (and should be NULL).
6604 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
6608 CEPH_DEFINE_OID_ONSTACK(oid);
6613 * When probing a parent image, the image id is already
6614 * known (and the image name likely is not). There's no
6615 * need to fetch the image id again in this case. We
6616 * do still need to set the image format though.
6618 if (rbd_dev->spec->image_id) {
6619 rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
6625 * First, see if the format 2 image id file exists, and if
6626 * so, get the image's persistent id from it.
6628 ret = ceph_oid_aprintf(&oid, GFP_KERNEL, "%s%s", RBD_ID_PREFIX,
6629 rbd_dev->spec->image_name);
6633 dout("rbd id object name is %s\n", oid.name);
6635 /* Response will be an encoded string, which includes a length */
6636 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
6637 response = kzalloc(size, GFP_NOIO);
6643 /* If it doesn't exist we'll assume it's a format 1 image */
6645 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
6648 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6649 if (ret == -ENOENT) {
6650 image_id = kstrdup("", GFP_KERNEL);
6651 ret = image_id ? 0 : -ENOMEM;
6653 rbd_dev->image_format = 1;
6654 } else if (ret >= 0) {
6657 image_id = ceph_extract_encoded_string(&p, p + ret,
6659 ret = PTR_ERR_OR_ZERO(image_id);
6661 rbd_dev->image_format = 2;
6665 rbd_dev->spec->image_id = image_id;
6666 dout("image_id is %s\n", image_id);
6670 ceph_oid_destroy(&oid);
6675 * Undo whatever state changes are made by v1 or v2 header info
6678 static void rbd_dev_unprobe(struct rbd_device *rbd_dev)
6680 struct rbd_image_header *header;
6682 rbd_dev_parent_put(rbd_dev);
6683 rbd_object_map_free(rbd_dev);
6684 rbd_dev_mapping_clear(rbd_dev);
6686 /* Free dynamic fields from the header, then zero it out */
6688 header = &rbd_dev->header;
6689 ceph_put_snap_context(header->snapc);
6690 kfree(header->snap_sizes);
6691 kfree(header->snap_names);
6692 kfree(header->object_prefix);
6693 memset(header, 0, sizeof (*header));
6696 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev)
6700 ret = rbd_dev_v2_object_prefix(rbd_dev);
6705 * Get the and check features for the image. Currently the
6706 * features are assumed to never change.
6708 ret = rbd_dev_v2_features(rbd_dev);
6712 /* If the image supports fancy striping, get its parameters */
6714 if (rbd_dev->header.features & RBD_FEATURE_STRIPINGV2) {
6715 ret = rbd_dev_v2_striping_info(rbd_dev);
6720 if (rbd_dev->header.features & RBD_FEATURE_DATA_POOL) {
6721 ret = rbd_dev_v2_data_pool(rbd_dev);
6726 rbd_init_layout(rbd_dev);
6730 rbd_dev->header.features = 0;
6731 kfree(rbd_dev->header.object_prefix);
6732 rbd_dev->header.object_prefix = NULL;
6737 * @depth is rbd_dev_image_probe() -> rbd_dev_probe_parent() ->
6738 * rbd_dev_image_probe() recursion depth, which means it's also the
6739 * length of the already discovered part of the parent chain.
6741 static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
6743 struct rbd_device *parent = NULL;
6746 if (!rbd_dev->parent_spec)
6749 if (++depth > RBD_MAX_PARENT_CHAIN_LEN) {
6750 pr_info("parent chain is too long (%d)\n", depth);
6755 parent = __rbd_dev_create(rbd_dev->rbd_client, rbd_dev->parent_spec);
6762 * Images related by parent/child relationships always share
6763 * rbd_client and spec/parent_spec, so bump their refcounts.
6765 __rbd_get_client(rbd_dev->rbd_client);
6766 rbd_spec_get(rbd_dev->parent_spec);
6768 __set_bit(RBD_DEV_FLAG_READONLY, &parent->flags);
6770 ret = rbd_dev_image_probe(parent, depth);
6774 rbd_dev->parent = parent;
6775 atomic_set(&rbd_dev->parent_ref, 1);
6779 rbd_dev_unparent(rbd_dev);
6780 rbd_dev_destroy(parent);
6784 static void rbd_dev_device_release(struct rbd_device *rbd_dev)
6786 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6787 rbd_free_disk(rbd_dev);
6789 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6793 * rbd_dev->header_rwsem must be locked for write and will be unlocked
6796 static int rbd_dev_device_setup(struct rbd_device *rbd_dev)
6800 /* Record our major and minor device numbers. */
6802 if (!single_major) {
6803 ret = register_blkdev(0, rbd_dev->name);
6805 goto err_out_unlock;
6807 rbd_dev->major = ret;
6810 rbd_dev->major = rbd_major;
6811 rbd_dev->minor = rbd_dev_id_to_minor(rbd_dev->dev_id);
6814 /* Set up the blkdev mapping. */
6816 ret = rbd_init_disk(rbd_dev);
6818 goto err_out_blkdev;
6820 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
6821 set_disk_ro(rbd_dev->disk, rbd_is_ro(rbd_dev));
6823 ret = dev_set_name(&rbd_dev->dev, "%d", rbd_dev->dev_id);
6827 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6828 up_write(&rbd_dev->header_rwsem);
6832 rbd_free_disk(rbd_dev);
6835 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6837 up_write(&rbd_dev->header_rwsem);
6841 static int rbd_dev_header_name(struct rbd_device *rbd_dev)
6843 struct rbd_spec *spec = rbd_dev->spec;
6846 /* Record the header object name for this rbd image. */
6848 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6849 if (rbd_dev->image_format == 1)
6850 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6851 spec->image_name, RBD_SUFFIX);
6853 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6854 RBD_HEADER_PREFIX, spec->image_id);
6859 static void rbd_print_dne(struct rbd_device *rbd_dev, bool is_snap)
6862 pr_info("image %s/%s%s%s does not exist\n",
6863 rbd_dev->spec->pool_name,
6864 rbd_dev->spec->pool_ns ?: "",
6865 rbd_dev->spec->pool_ns ? "/" : "",
6866 rbd_dev->spec->image_name);
6868 pr_info("snap %s/%s%s%s@%s does not exist\n",
6869 rbd_dev->spec->pool_name,
6870 rbd_dev->spec->pool_ns ?: "",
6871 rbd_dev->spec->pool_ns ? "/" : "",
6872 rbd_dev->spec->image_name,
6873 rbd_dev->spec->snap_name);
6877 static void rbd_dev_image_release(struct rbd_device *rbd_dev)
6879 if (!rbd_is_ro(rbd_dev))
6880 rbd_unregister_watch(rbd_dev);
6882 rbd_dev_unprobe(rbd_dev);
6883 rbd_dev->image_format = 0;
6884 kfree(rbd_dev->spec->image_id);
6885 rbd_dev->spec->image_id = NULL;
6889 * Probe for the existence of the header object for the given rbd
6890 * device. If this image is the one being mapped (i.e., not a
6891 * parent), initiate a watch on its header object before using that
6892 * object to get detailed information about the rbd image.
6894 * On success, returns with header_rwsem held for write if called
6897 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
6899 bool need_watch = !rbd_is_ro(rbd_dev);
6903 * Get the id from the image id object. Unless there's an
6904 * error, rbd_dev->spec->image_id will be filled in with
6905 * a dynamically-allocated string, and rbd_dev->image_format
6906 * will be set to either 1 or 2.
6908 ret = rbd_dev_image_id(rbd_dev);
6912 ret = rbd_dev_header_name(rbd_dev);
6914 goto err_out_format;
6917 ret = rbd_register_watch(rbd_dev);
6920 rbd_print_dne(rbd_dev, false);
6921 goto err_out_format;
6926 down_write(&rbd_dev->header_rwsem);
6928 ret = rbd_dev_header_info(rbd_dev);
6930 if (ret == -ENOENT && !need_watch)
6931 rbd_print_dne(rbd_dev, false);
6936 * If this image is the one being mapped, we have pool name and
6937 * id, image name and id, and snap name - need to fill snap id.
6938 * Otherwise this is a parent image, identified by pool, image
6939 * and snap ids - need to fill in names for those ids.
6942 ret = rbd_spec_fill_snap_id(rbd_dev);
6944 ret = rbd_spec_fill_names(rbd_dev);
6947 rbd_print_dne(rbd_dev, true);
6951 ret = rbd_dev_mapping_set(rbd_dev);
6955 if (rbd_is_snap(rbd_dev) &&
6956 (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)) {
6957 ret = rbd_object_map_load(rbd_dev);
6962 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
6963 ret = rbd_dev_v2_parent_info(rbd_dev);
6968 ret = rbd_dev_probe_parent(rbd_dev, depth);
6972 dout("discovered format %u image, header name is %s\n",
6973 rbd_dev->image_format, rbd_dev->header_oid.name);
6978 up_write(&rbd_dev->header_rwsem);
6980 rbd_unregister_watch(rbd_dev);
6981 rbd_dev_unprobe(rbd_dev);
6983 rbd_dev->image_format = 0;
6984 kfree(rbd_dev->spec->image_id);
6985 rbd_dev->spec->image_id = NULL;
6989 static ssize_t do_rbd_add(struct bus_type *bus,
6993 struct rbd_device *rbd_dev = NULL;
6994 struct ceph_options *ceph_opts = NULL;
6995 struct rbd_options *rbd_opts = NULL;
6996 struct rbd_spec *spec = NULL;
6997 struct rbd_client *rbdc;
7000 if (!capable(CAP_SYS_ADMIN))
7003 if (!try_module_get(THIS_MODULE))
7006 /* parse add command */
7007 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
7011 rbdc = rbd_get_client(ceph_opts);
7018 rc = ceph_pg_poolid_by_name(rbdc->client->osdc.osdmap, spec->pool_name);
7021 pr_info("pool %s does not exist\n", spec->pool_name);
7022 goto err_out_client;
7024 spec->pool_id = (u64)rc;
7026 rbd_dev = rbd_dev_create(rbdc, spec, rbd_opts);
7029 goto err_out_client;
7031 rbdc = NULL; /* rbd_dev now owns this */
7032 spec = NULL; /* rbd_dev now owns this */
7033 rbd_opts = NULL; /* rbd_dev now owns this */
7035 /* if we are mapping a snapshot it will be a read-only mapping */
7036 if (rbd_dev->opts->read_only ||
7037 strcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME))
7038 __set_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
7040 rbd_dev->config_info = kstrdup(buf, GFP_KERNEL);
7041 if (!rbd_dev->config_info) {
7043 goto err_out_rbd_dev;
7046 rc = rbd_dev_image_probe(rbd_dev, 0);
7048 goto err_out_rbd_dev;
7050 if (rbd_dev->opts->alloc_size > rbd_dev->layout.object_size) {
7051 rbd_warn(rbd_dev, "alloc_size adjusted to %u",
7052 rbd_dev->layout.object_size);
7053 rbd_dev->opts->alloc_size = rbd_dev->layout.object_size;
7056 rc = rbd_dev_device_setup(rbd_dev);
7058 goto err_out_image_probe;
7060 rc = rbd_add_acquire_lock(rbd_dev);
7062 goto err_out_image_lock;
7064 /* Everything's ready. Announce the disk to the world. */
7066 rc = device_add(&rbd_dev->dev);
7068 goto err_out_image_lock;
7070 device_add_disk(&rbd_dev->dev, rbd_dev->disk, NULL);
7072 spin_lock(&rbd_dev_list_lock);
7073 list_add_tail(&rbd_dev->node, &rbd_dev_list);
7074 spin_unlock(&rbd_dev_list_lock);
7076 pr_info("%s: capacity %llu features 0x%llx\n", rbd_dev->disk->disk_name,
7077 (unsigned long long)get_capacity(rbd_dev->disk) << SECTOR_SHIFT,
7078 rbd_dev->header.features);
7081 module_put(THIS_MODULE);
7085 rbd_dev_image_unlock(rbd_dev);
7086 rbd_dev_device_release(rbd_dev);
7087 err_out_image_probe:
7088 rbd_dev_image_release(rbd_dev);
7090 rbd_dev_destroy(rbd_dev);
7092 rbd_put_client(rbdc);
7099 static ssize_t add_store(struct bus_type *bus, const char *buf, size_t count)
7104 return do_rbd_add(bus, buf, count);
7107 static ssize_t add_single_major_store(struct bus_type *bus, const char *buf,
7110 return do_rbd_add(bus, buf, count);
7113 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev)
7115 while (rbd_dev->parent) {
7116 struct rbd_device *first = rbd_dev;
7117 struct rbd_device *second = first->parent;
7118 struct rbd_device *third;
7121 * Follow to the parent with no grandparent and
7124 while (second && (third = second->parent)) {
7129 rbd_dev_image_release(second);
7130 rbd_dev_destroy(second);
7131 first->parent = NULL;
7132 first->parent_overlap = 0;
7134 rbd_assert(first->parent_spec);
7135 rbd_spec_put(first->parent_spec);
7136 first->parent_spec = NULL;
7140 static ssize_t do_rbd_remove(struct bus_type *bus,
7144 struct rbd_device *rbd_dev = NULL;
7145 struct list_head *tmp;
7151 if (!capable(CAP_SYS_ADMIN))
7156 sscanf(buf, "%d %5s", &dev_id, opt_buf);
7158 pr_err("dev_id out of range\n");
7161 if (opt_buf[0] != '\0') {
7162 if (!strcmp(opt_buf, "force")) {
7165 pr_err("bad remove option at '%s'\n", opt_buf);
7171 spin_lock(&rbd_dev_list_lock);
7172 list_for_each(tmp, &rbd_dev_list) {
7173 rbd_dev = list_entry(tmp, struct rbd_device, node);
7174 if (rbd_dev->dev_id == dev_id) {
7180 spin_lock_irq(&rbd_dev->lock);
7181 if (rbd_dev->open_count && !force)
7183 else if (test_and_set_bit(RBD_DEV_FLAG_REMOVING,
7186 spin_unlock_irq(&rbd_dev->lock);
7188 spin_unlock(&rbd_dev_list_lock);
7194 * Prevent new IO from being queued and wait for existing
7195 * IO to complete/fail.
7197 blk_mq_freeze_queue(rbd_dev->disk->queue);
7198 blk_set_queue_dying(rbd_dev->disk->queue);
7201 del_gendisk(rbd_dev->disk);
7202 spin_lock(&rbd_dev_list_lock);
7203 list_del_init(&rbd_dev->node);
7204 spin_unlock(&rbd_dev_list_lock);
7205 device_del(&rbd_dev->dev);
7207 rbd_dev_image_unlock(rbd_dev);
7208 rbd_dev_device_release(rbd_dev);
7209 rbd_dev_image_release(rbd_dev);
7210 rbd_dev_destroy(rbd_dev);
7214 static ssize_t remove_store(struct bus_type *bus, const char *buf, size_t count)
7219 return do_rbd_remove(bus, buf, count);
7222 static ssize_t remove_single_major_store(struct bus_type *bus, const char *buf,
7225 return do_rbd_remove(bus, buf, count);
7229 * create control files in sysfs
7232 static int __init rbd_sysfs_init(void)
7236 ret = device_register(&rbd_root_dev);
7240 ret = bus_register(&rbd_bus_type);
7242 device_unregister(&rbd_root_dev);
7247 static void __exit rbd_sysfs_cleanup(void)
7249 bus_unregister(&rbd_bus_type);
7250 device_unregister(&rbd_root_dev);
7253 static int __init rbd_slab_init(void)
7255 rbd_assert(!rbd_img_request_cache);
7256 rbd_img_request_cache = KMEM_CACHE(rbd_img_request, 0);
7257 if (!rbd_img_request_cache)
7260 rbd_assert(!rbd_obj_request_cache);
7261 rbd_obj_request_cache = KMEM_CACHE(rbd_obj_request, 0);
7262 if (!rbd_obj_request_cache)
7268 kmem_cache_destroy(rbd_img_request_cache);
7269 rbd_img_request_cache = NULL;
7273 static void rbd_slab_exit(void)
7275 rbd_assert(rbd_obj_request_cache);
7276 kmem_cache_destroy(rbd_obj_request_cache);
7277 rbd_obj_request_cache = NULL;
7279 rbd_assert(rbd_img_request_cache);
7280 kmem_cache_destroy(rbd_img_request_cache);
7281 rbd_img_request_cache = NULL;
7284 static int __init rbd_init(void)
7288 if (!libceph_compatible(NULL)) {
7289 rbd_warn(NULL, "libceph incompatibility (quitting)");
7293 rc = rbd_slab_init();
7298 * The number of active work items is limited by the number of
7299 * rbd devices * queue depth, so leave @max_active at default.
7301 rbd_wq = alloc_workqueue(RBD_DRV_NAME, WQ_MEM_RECLAIM, 0);
7308 rbd_major = register_blkdev(0, RBD_DRV_NAME);
7309 if (rbd_major < 0) {
7315 rc = rbd_sysfs_init();
7317 goto err_out_blkdev;
7320 pr_info("loaded (major %d)\n", rbd_major);
7322 pr_info("loaded\n");
7328 unregister_blkdev(rbd_major, RBD_DRV_NAME);
7330 destroy_workqueue(rbd_wq);
7336 static void __exit rbd_exit(void)
7338 ida_destroy(&rbd_dev_id_ida);
7339 rbd_sysfs_cleanup();
7341 unregister_blkdev(rbd_major, RBD_DRV_NAME);
7342 destroy_workqueue(rbd_wq);
7346 module_init(rbd_init);
7347 module_exit(rbd_exit);
7349 MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
7350 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
7351 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
7352 /* following authorship retained from original osdblk.c */
7353 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
7355 MODULE_DESCRIPTION("RADOS Block Device (RBD) driver");
7356 MODULE_LICENSE("GPL");