3 rbd.c -- Export ceph rados objects as a Linux block device
6 based on drivers/block/osdblk.c:
8 Copyright 2009 Red Hat, Inc.
10 This program is free software; you can redistribute it and/or modify
11 it under the terms of the GNU General Public License as published by
12 the Free Software Foundation.
14 This program is distributed in the hope that it will be useful,
15 but WITHOUT ANY WARRANTY; without even the implied warranty of
16 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 GNU General Public License for more details.
19 You should have received a copy of the GNU General Public License
20 along with this program; see the file COPYING. If not, write to
21 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
25 For usage instructions, please refer to:
27 Documentation/ABI/testing/sysfs-bus-rbd
31 #include <linux/ceph/libceph.h>
32 #include <linux/ceph/osd_client.h>
33 #include <linux/ceph/mon_client.h>
34 #include <linux/ceph/cls_lock_client.h>
35 #include <linux/ceph/striper.h>
36 #include <linux/ceph/decode.h>
37 #include <linux/fs_parser.h>
38 #include <linux/bsearch.h>
40 #include <linux/kernel.h>
41 #include <linux/device.h>
42 #include <linux/module.h>
43 #include <linux/blk-mq.h>
45 #include <linux/blkdev.h>
46 #include <linux/slab.h>
47 #include <linux/idr.h>
48 #include <linux/workqueue.h>
50 #include "rbd_types.h"
52 #define RBD_DEBUG /* Activate rbd_assert() calls */
55 * Increment the given counter and return its updated value.
56 * If the counter is already 0 it will not be incremented.
57 * If the counter is already at its maximum value returns
58 * -EINVAL without updating it.
60 static int atomic_inc_return_safe(atomic_t *v)
64 counter = (unsigned int)atomic_fetch_add_unless(v, 1, 0);
65 if (counter <= (unsigned int)INT_MAX)
73 /* Decrement the counter. Return the resulting value, or -EINVAL */
74 static int atomic_dec_return_safe(atomic_t *v)
78 counter = atomic_dec_return(v);
87 #define RBD_DRV_NAME "rbd"
89 #define RBD_MINORS_PER_MAJOR 256
90 #define RBD_SINGLE_MAJOR_PART_SHIFT 4
92 #define RBD_MAX_PARENT_CHAIN_LEN 16
94 #define RBD_SNAP_DEV_NAME_PREFIX "snap_"
95 #define RBD_MAX_SNAP_NAME_LEN \
96 (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
98 #define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */
100 #define RBD_SNAP_HEAD_NAME "-"
102 #define BAD_SNAP_INDEX U32_MAX /* invalid index into snap array */
104 /* This allows a single page to hold an image name sent by OSD */
105 #define RBD_IMAGE_NAME_LEN_MAX (PAGE_SIZE - sizeof (__le32) - 1)
106 #define RBD_IMAGE_ID_LEN_MAX 64
108 #define RBD_OBJ_PREFIX_LEN_MAX 64
110 #define RBD_NOTIFY_TIMEOUT 5 /* seconds */
111 #define RBD_RETRY_DELAY msecs_to_jiffies(1000)
115 #define RBD_FEATURE_LAYERING (1ULL<<0)
116 #define RBD_FEATURE_STRIPINGV2 (1ULL<<1)
117 #define RBD_FEATURE_EXCLUSIVE_LOCK (1ULL<<2)
118 #define RBD_FEATURE_OBJECT_MAP (1ULL<<3)
119 #define RBD_FEATURE_FAST_DIFF (1ULL<<4)
120 #define RBD_FEATURE_DEEP_FLATTEN (1ULL<<5)
121 #define RBD_FEATURE_DATA_POOL (1ULL<<7)
122 #define RBD_FEATURE_OPERATIONS (1ULL<<8)
124 #define RBD_FEATURES_ALL (RBD_FEATURE_LAYERING | \
125 RBD_FEATURE_STRIPINGV2 | \
126 RBD_FEATURE_EXCLUSIVE_LOCK | \
127 RBD_FEATURE_OBJECT_MAP | \
128 RBD_FEATURE_FAST_DIFF | \
129 RBD_FEATURE_DEEP_FLATTEN | \
130 RBD_FEATURE_DATA_POOL | \
131 RBD_FEATURE_OPERATIONS)
133 /* Features supported by this (client software) implementation. */
135 #define RBD_FEATURES_SUPPORTED (RBD_FEATURES_ALL)
138 * An RBD device name will be "rbd#", where the "rbd" comes from
139 * RBD_DRV_NAME above, and # is a unique integer identifier.
141 #define DEV_NAME_LEN 32
144 * block device image metadata (in-memory version)
146 struct rbd_image_header {
147 /* These six fields never change for a given rbd image */
153 u64 features; /* Might be changeable someday? */
155 /* The remaining fields need to be updated occasionally */
157 struct ceph_snap_context *snapc;
158 char *snap_names; /* format 1 only */
159 u64 *snap_sizes; /* format 1 only */
163 * An rbd image specification.
165 * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
166 * identify an image. Each rbd_dev structure includes a pointer to
167 * an rbd_spec structure that encapsulates this identity.
169 * Each of the id's in an rbd_spec has an associated name. For a
170 * user-mapped image, the names are supplied and the id's associated
171 * with them are looked up. For a layered image, a parent image is
172 * defined by the tuple, and the names are looked up.
174 * An rbd_dev structure contains a parent_spec pointer which is
175 * non-null if the image it represents is a child in a layered
176 * image. This pointer will refer to the rbd_spec structure used
177 * by the parent rbd_dev for its own identity (i.e., the structure
178 * is shared between the parent and child).
180 * Since these structures are populated once, during the discovery
181 * phase of image construction, they are effectively immutable so
182 * we make no effort to synchronize access to them.
184 * Note that code herein does not assume the image name is known (it
185 * could be a null pointer).
189 const char *pool_name;
190 const char *pool_ns; /* NULL if default, never "" */
192 const char *image_id;
193 const char *image_name;
196 const char *snap_name;
202 * an instance of the client. multiple devices may share an rbd client.
205 struct ceph_client *client;
207 struct list_head node;
210 struct pending_result {
211 int result; /* first nonzero result */
215 struct rbd_img_request;
217 enum obj_request_type {
218 OBJ_REQUEST_NODATA = 1,
219 OBJ_REQUEST_BIO, /* pointer into provided bio (list) */
220 OBJ_REQUEST_BVECS, /* pointer into provided bio_vec array */
221 OBJ_REQUEST_OWN_BVECS, /* private bio_vec array, doesn't own pages */
224 enum obj_operation_type {
231 #define RBD_OBJ_FLAG_DELETION (1U << 0)
232 #define RBD_OBJ_FLAG_COPYUP_ENABLED (1U << 1)
233 #define RBD_OBJ_FLAG_COPYUP_ZEROS (1U << 2)
234 #define RBD_OBJ_FLAG_MAY_EXIST (1U << 3)
235 #define RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT (1U << 4)
237 enum rbd_obj_read_state {
238 RBD_OBJ_READ_START = 1,
244 * Writes go through the following state machine to deal with
247 * . . . . . RBD_OBJ_WRITE_GUARD. . . . . . . . . . . . . .
250 * . RBD_OBJ_WRITE_READ_FROM_PARENT. . . .
252 * . v v (deep-copyup .
253 * (image . RBD_OBJ_WRITE_COPYUP_EMPTY_SNAPC . not needed) .
256 * . . . .RBD_OBJ_WRITE_COPYUP_OPS. . . . . (copyup .
259 * done . . . . . . . . . . . . . . . . . .
264 * Writes start in RBD_OBJ_WRITE_GUARD or _FLAT, depending on whether
265 * assert_exists guard is needed or not (in some cases it's not needed
266 * even if there is a parent).
268 enum rbd_obj_write_state {
269 RBD_OBJ_WRITE_START = 1,
270 RBD_OBJ_WRITE_PRE_OBJECT_MAP,
271 RBD_OBJ_WRITE_OBJECT,
272 __RBD_OBJ_WRITE_COPYUP,
273 RBD_OBJ_WRITE_COPYUP,
274 RBD_OBJ_WRITE_POST_OBJECT_MAP,
277 enum rbd_obj_copyup_state {
278 RBD_OBJ_COPYUP_START = 1,
279 RBD_OBJ_COPYUP_READ_PARENT,
280 __RBD_OBJ_COPYUP_OBJECT_MAPS,
281 RBD_OBJ_COPYUP_OBJECT_MAPS,
282 __RBD_OBJ_COPYUP_WRITE_OBJECT,
283 RBD_OBJ_COPYUP_WRITE_OBJECT,
286 struct rbd_obj_request {
287 struct ceph_object_extent ex;
288 unsigned int flags; /* RBD_OBJ_FLAG_* */
290 enum rbd_obj_read_state read_state; /* for reads */
291 enum rbd_obj_write_state write_state; /* for writes */
294 struct rbd_img_request *img_request;
295 struct ceph_file_extent *img_extents;
299 struct ceph_bio_iter bio_pos;
301 struct ceph_bvec_iter bvec_pos;
307 enum rbd_obj_copyup_state copyup_state;
308 struct bio_vec *copyup_bvecs;
309 u32 copyup_bvec_count;
311 struct list_head osd_reqs; /* w/ r_private_item */
313 struct mutex state_mutex;
314 struct pending_result pending;
319 IMG_REQ_CHILD, /* initiator: block = 0, child image = 1 */
320 IMG_REQ_LAYERED, /* ENOENT handling: normal = 0, layered = 1 */
325 RBD_IMG_EXCLUSIVE_LOCK,
326 __RBD_IMG_OBJECT_REQUESTS,
327 RBD_IMG_OBJECT_REQUESTS,
330 struct rbd_img_request {
331 struct rbd_device *rbd_dev;
332 enum obj_operation_type op_type;
333 enum obj_request_type data_type;
335 enum rbd_img_state state;
337 u64 snap_id; /* for reads */
338 struct ceph_snap_context *snapc; /* for writes */
340 struct rbd_obj_request *obj_request; /* obj req initiator */
342 struct list_head lock_item;
343 struct list_head object_extents; /* obj_req.ex structs */
345 struct mutex state_mutex;
346 struct pending_result pending;
347 struct work_struct work;
351 #define for_each_obj_request(ireq, oreq) \
352 list_for_each_entry(oreq, &(ireq)->object_extents, ex.oe_item)
353 #define for_each_obj_request_safe(ireq, oreq, n) \
354 list_for_each_entry_safe(oreq, n, &(ireq)->object_extents, ex.oe_item)
356 enum rbd_watch_state {
357 RBD_WATCH_STATE_UNREGISTERED,
358 RBD_WATCH_STATE_REGISTERED,
359 RBD_WATCH_STATE_ERROR,
362 enum rbd_lock_state {
363 RBD_LOCK_STATE_UNLOCKED,
364 RBD_LOCK_STATE_LOCKED,
365 RBD_LOCK_STATE_RELEASING,
368 /* WatchNotify::ClientId */
369 struct rbd_client_id {
382 int dev_id; /* blkdev unique id */
384 int major; /* blkdev assigned major */
386 struct gendisk *disk; /* blkdev's gendisk and rq */
388 u32 image_format; /* Either 1 or 2 */
389 struct rbd_client *rbd_client;
391 char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
393 spinlock_t lock; /* queue, flags, open_count */
395 struct rbd_image_header header;
396 unsigned long flags; /* possibly lock protected */
397 struct rbd_spec *spec;
398 struct rbd_options *opts;
399 char *config_info; /* add{,_single_major} string */
401 struct ceph_object_id header_oid;
402 struct ceph_object_locator header_oloc;
404 struct ceph_file_layout layout; /* used for all rbd requests */
406 struct mutex watch_mutex;
407 enum rbd_watch_state watch_state;
408 struct ceph_osd_linger_request *watch_handle;
410 struct delayed_work watch_dwork;
412 struct rw_semaphore lock_rwsem;
413 enum rbd_lock_state lock_state;
414 char lock_cookie[32];
415 struct rbd_client_id owner_cid;
416 struct work_struct acquired_lock_work;
417 struct work_struct released_lock_work;
418 struct delayed_work lock_dwork;
419 struct work_struct unlock_work;
420 spinlock_t lock_lists_lock;
421 struct list_head acquiring_list;
422 struct list_head running_list;
423 struct completion acquire_wait;
425 struct completion releasing_wait;
427 spinlock_t object_map_lock;
429 u64 object_map_size; /* in objects */
430 u64 object_map_flags;
432 struct workqueue_struct *task_wq;
434 struct rbd_spec *parent_spec;
437 struct rbd_device *parent;
439 /* Block layer tags. */
440 struct blk_mq_tag_set tag_set;
442 /* protects updating the header */
443 struct rw_semaphore header_rwsem;
445 struct rbd_mapping mapping;
447 struct list_head node;
451 unsigned long open_count; /* protected by lock */
455 * Flag bits for rbd_dev->flags:
456 * - REMOVING (which is coupled with rbd_dev->open_count) is protected
460 RBD_DEV_FLAG_EXISTS, /* rbd_dev_device_setup() ran */
461 RBD_DEV_FLAG_REMOVING, /* this mapping is being removed */
462 RBD_DEV_FLAG_READONLY, /* -o ro or snapshot */
465 static DEFINE_MUTEX(client_mutex); /* Serialize client creation */
467 static LIST_HEAD(rbd_dev_list); /* devices */
468 static DEFINE_SPINLOCK(rbd_dev_list_lock);
470 static LIST_HEAD(rbd_client_list); /* clients */
471 static DEFINE_SPINLOCK(rbd_client_list_lock);
473 /* Slab caches for frequently-allocated structures */
475 static struct kmem_cache *rbd_img_request_cache;
476 static struct kmem_cache *rbd_obj_request_cache;
478 static int rbd_major;
479 static DEFINE_IDA(rbd_dev_id_ida);
481 static struct workqueue_struct *rbd_wq;
483 static struct ceph_snap_context rbd_empty_snapc = {
484 .nref = REFCOUNT_INIT(1),
488 * single-major requires >= 0.75 version of userspace rbd utility.
490 static bool single_major = true;
491 module_param(single_major, bool, 0444);
492 MODULE_PARM_DESC(single_major, "Use a single major number for all rbd devices (default: true)");
494 static ssize_t add_store(const struct bus_type *bus, const char *buf, size_t count);
495 static ssize_t remove_store(const struct bus_type *bus, const char *buf,
497 static ssize_t add_single_major_store(const struct bus_type *bus, const char *buf,
499 static ssize_t remove_single_major_store(const struct bus_type *bus, const char *buf,
501 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth);
503 static int rbd_dev_id_to_minor(int dev_id)
505 return dev_id << RBD_SINGLE_MAJOR_PART_SHIFT;
508 static int minor_to_rbd_dev_id(int minor)
510 return minor >> RBD_SINGLE_MAJOR_PART_SHIFT;
513 static bool rbd_is_ro(struct rbd_device *rbd_dev)
515 return test_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
518 static bool rbd_is_snap(struct rbd_device *rbd_dev)
520 return rbd_dev->spec->snap_id != CEPH_NOSNAP;
523 static bool __rbd_is_lock_owner(struct rbd_device *rbd_dev)
525 lockdep_assert_held(&rbd_dev->lock_rwsem);
527 return rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED ||
528 rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING;
531 static bool rbd_is_lock_owner(struct rbd_device *rbd_dev)
535 down_read(&rbd_dev->lock_rwsem);
536 is_lock_owner = __rbd_is_lock_owner(rbd_dev);
537 up_read(&rbd_dev->lock_rwsem);
538 return is_lock_owner;
541 static ssize_t supported_features_show(const struct bus_type *bus, char *buf)
543 return sprintf(buf, "0x%llx\n", RBD_FEATURES_SUPPORTED);
546 static BUS_ATTR_WO(add);
547 static BUS_ATTR_WO(remove);
548 static BUS_ATTR_WO(add_single_major);
549 static BUS_ATTR_WO(remove_single_major);
550 static BUS_ATTR_RO(supported_features);
552 static struct attribute *rbd_bus_attrs[] = {
554 &bus_attr_remove.attr,
555 &bus_attr_add_single_major.attr,
556 &bus_attr_remove_single_major.attr,
557 &bus_attr_supported_features.attr,
561 static umode_t rbd_bus_is_visible(struct kobject *kobj,
562 struct attribute *attr, int index)
565 (attr == &bus_attr_add_single_major.attr ||
566 attr == &bus_attr_remove_single_major.attr))
572 static const struct attribute_group rbd_bus_group = {
573 .attrs = rbd_bus_attrs,
574 .is_visible = rbd_bus_is_visible,
576 __ATTRIBUTE_GROUPS(rbd_bus);
578 static struct bus_type rbd_bus_type = {
580 .bus_groups = rbd_bus_groups,
583 static void rbd_root_dev_release(struct device *dev)
587 static struct device rbd_root_dev = {
589 .release = rbd_root_dev_release,
592 static __printf(2, 3)
593 void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
595 struct va_format vaf;
603 printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
604 else if (rbd_dev->disk)
605 printk(KERN_WARNING "%s: %s: %pV\n",
606 RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
607 else if (rbd_dev->spec && rbd_dev->spec->image_name)
608 printk(KERN_WARNING "%s: image %s: %pV\n",
609 RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
610 else if (rbd_dev->spec && rbd_dev->spec->image_id)
611 printk(KERN_WARNING "%s: id %s: %pV\n",
612 RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
614 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
615 RBD_DRV_NAME, rbd_dev, &vaf);
620 #define rbd_assert(expr) \
621 if (unlikely(!(expr))) { \
622 printk(KERN_ERR "\nAssertion failure in %s() " \
624 "\trbd_assert(%s);\n\n", \
625 __func__, __LINE__, #expr); \
628 #else /* !RBD_DEBUG */
629 # define rbd_assert(expr) ((void) 0)
630 #endif /* !RBD_DEBUG */
632 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev);
634 static int rbd_dev_refresh(struct rbd_device *rbd_dev);
635 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev);
636 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
638 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
639 u8 *order, u64 *snap_size);
640 static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev);
642 static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result);
643 static void rbd_img_handle_request(struct rbd_img_request *img_req, int result);
646 * Return true if nothing else is pending.
648 static bool pending_result_dec(struct pending_result *pending, int *result)
650 rbd_assert(pending->num_pending > 0);
652 if (*result && !pending->result)
653 pending->result = *result;
654 if (--pending->num_pending)
657 *result = pending->result;
661 static int rbd_open(struct gendisk *disk, blk_mode_t mode)
663 struct rbd_device *rbd_dev = disk->private_data;
664 bool removing = false;
666 spin_lock_irq(&rbd_dev->lock);
667 if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
670 rbd_dev->open_count++;
671 spin_unlock_irq(&rbd_dev->lock);
675 (void) get_device(&rbd_dev->dev);
680 static void rbd_release(struct gendisk *disk)
682 struct rbd_device *rbd_dev = disk->private_data;
683 unsigned long open_count_before;
685 spin_lock_irq(&rbd_dev->lock);
686 open_count_before = rbd_dev->open_count--;
687 spin_unlock_irq(&rbd_dev->lock);
688 rbd_assert(open_count_before > 0);
690 put_device(&rbd_dev->dev);
693 static const struct block_device_operations rbd_bd_ops = {
694 .owner = THIS_MODULE,
696 .release = rbd_release,
700 * Initialize an rbd client instance. Success or not, this function
701 * consumes ceph_opts. Caller holds client_mutex.
703 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
705 struct rbd_client *rbdc;
708 dout("%s:\n", __func__);
709 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
713 kref_init(&rbdc->kref);
714 INIT_LIST_HEAD(&rbdc->node);
716 rbdc->client = ceph_create_client(ceph_opts, rbdc);
717 if (IS_ERR(rbdc->client))
719 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
721 ret = ceph_open_session(rbdc->client);
725 spin_lock(&rbd_client_list_lock);
726 list_add_tail(&rbdc->node, &rbd_client_list);
727 spin_unlock(&rbd_client_list_lock);
729 dout("%s: rbdc %p\n", __func__, rbdc);
733 ceph_destroy_client(rbdc->client);
738 ceph_destroy_options(ceph_opts);
739 dout("%s: error %d\n", __func__, ret);
744 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
746 kref_get(&rbdc->kref);
752 * Find a ceph client with specific addr and configuration. If
753 * found, bump its reference count.
755 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
757 struct rbd_client *rbdc = NULL, *iter;
759 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
762 spin_lock(&rbd_client_list_lock);
763 list_for_each_entry(iter, &rbd_client_list, node) {
764 if (!ceph_compare_options(ceph_opts, iter->client)) {
765 __rbd_get_client(iter);
771 spin_unlock(&rbd_client_list_lock);
777 * (Per device) rbd map options
785 Opt_compression_hint,
786 /* string args above */
795 Opt_compression_hint_none,
796 Opt_compression_hint_compressible,
797 Opt_compression_hint_incompressible,
800 static const struct constant_table rbd_param_compression_hint[] = {
801 {"none", Opt_compression_hint_none},
802 {"compressible", Opt_compression_hint_compressible},
803 {"incompressible", Opt_compression_hint_incompressible},
807 static const struct fs_parameter_spec rbd_parameters[] = {
808 fsparam_u32 ("alloc_size", Opt_alloc_size),
809 fsparam_enum ("compression_hint", Opt_compression_hint,
810 rbd_param_compression_hint),
811 fsparam_flag ("exclusive", Opt_exclusive),
812 fsparam_flag ("lock_on_read", Opt_lock_on_read),
813 fsparam_u32 ("lock_timeout", Opt_lock_timeout),
814 fsparam_flag ("notrim", Opt_notrim),
815 fsparam_string ("_pool_ns", Opt_pool_ns),
816 fsparam_u32 ("queue_depth", Opt_queue_depth),
817 fsparam_flag ("read_only", Opt_read_only),
818 fsparam_flag ("read_write", Opt_read_write),
819 fsparam_flag ("ro", Opt_read_only),
820 fsparam_flag ("rw", Opt_read_write),
827 unsigned long lock_timeout;
833 u32 alloc_hint_flags; /* CEPH_OSD_OP_ALLOC_HINT_FLAG_* */
836 #define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_DEFAULT_RQ
837 #define RBD_ALLOC_SIZE_DEFAULT (64 * 1024)
838 #define RBD_LOCK_TIMEOUT_DEFAULT 0 /* no timeout */
839 #define RBD_READ_ONLY_DEFAULT false
840 #define RBD_LOCK_ON_READ_DEFAULT false
841 #define RBD_EXCLUSIVE_DEFAULT false
842 #define RBD_TRIM_DEFAULT true
844 struct rbd_parse_opts_ctx {
845 struct rbd_spec *spec;
846 struct ceph_options *copts;
847 struct rbd_options *opts;
850 static char* obj_op_name(enum obj_operation_type op_type)
867 * Destroy ceph client
869 * Caller must hold rbd_client_list_lock.
871 static void rbd_client_release(struct kref *kref)
873 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
875 dout("%s: rbdc %p\n", __func__, rbdc);
876 spin_lock(&rbd_client_list_lock);
877 list_del(&rbdc->node);
878 spin_unlock(&rbd_client_list_lock);
880 ceph_destroy_client(rbdc->client);
885 * Drop reference to ceph client node. If it's not referenced anymore, release
888 static void rbd_put_client(struct rbd_client *rbdc)
891 kref_put(&rbdc->kref, rbd_client_release);
895 * Get a ceph client with specific addr and configuration, if one does
896 * not exist create it. Either way, ceph_opts is consumed by this
899 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
901 struct rbd_client *rbdc;
904 mutex_lock(&client_mutex);
905 rbdc = rbd_client_find(ceph_opts);
907 ceph_destroy_options(ceph_opts);
910 * Using an existing client. Make sure ->pg_pools is up to
911 * date before we look up the pool id in do_rbd_add().
913 ret = ceph_wait_for_latest_osdmap(rbdc->client,
914 rbdc->client->options->mount_timeout);
916 rbd_warn(NULL, "failed to get latest osdmap: %d", ret);
917 rbd_put_client(rbdc);
921 rbdc = rbd_client_create(ceph_opts);
923 mutex_unlock(&client_mutex);
928 static bool rbd_image_format_valid(u32 image_format)
930 return image_format == 1 || image_format == 2;
933 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
938 /* The header has to start with the magic rbd header text */
939 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
942 /* The bio layer requires at least sector-sized I/O */
944 if (ondisk->options.order < SECTOR_SHIFT)
947 /* If we use u64 in a few spots we may be able to loosen this */
949 if (ondisk->options.order > 8 * sizeof (int) - 1)
953 * The size of a snapshot header has to fit in a size_t, and
954 * that limits the number of snapshots.
956 snap_count = le32_to_cpu(ondisk->snap_count);
957 size = SIZE_MAX - sizeof (struct ceph_snap_context);
958 if (snap_count > size / sizeof (__le64))
962 * Not only that, but the size of the entire the snapshot
963 * header must also be representable in a size_t.
965 size -= snap_count * sizeof (__le64);
966 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
973 * returns the size of an object in the image
975 static u32 rbd_obj_bytes(struct rbd_image_header *header)
977 return 1U << header->obj_order;
980 static void rbd_init_layout(struct rbd_device *rbd_dev)
982 if (rbd_dev->header.stripe_unit == 0 ||
983 rbd_dev->header.stripe_count == 0) {
984 rbd_dev->header.stripe_unit = rbd_obj_bytes(&rbd_dev->header);
985 rbd_dev->header.stripe_count = 1;
988 rbd_dev->layout.stripe_unit = rbd_dev->header.stripe_unit;
989 rbd_dev->layout.stripe_count = rbd_dev->header.stripe_count;
990 rbd_dev->layout.object_size = rbd_obj_bytes(&rbd_dev->header);
991 rbd_dev->layout.pool_id = rbd_dev->header.data_pool_id == CEPH_NOPOOL ?
992 rbd_dev->spec->pool_id : rbd_dev->header.data_pool_id;
993 RCU_INIT_POINTER(rbd_dev->layout.pool_ns, NULL);
997 * Fill an rbd image header with information from the given format 1
1000 static int rbd_header_from_disk(struct rbd_device *rbd_dev,
1001 struct rbd_image_header_ondisk *ondisk)
1003 struct rbd_image_header *header = &rbd_dev->header;
1004 bool first_time = header->object_prefix == NULL;
1005 struct ceph_snap_context *snapc;
1006 char *object_prefix = NULL;
1007 char *snap_names = NULL;
1008 u64 *snap_sizes = NULL;
1013 /* Allocate this now to avoid having to handle failure below */
1016 object_prefix = kstrndup(ondisk->object_prefix,
1017 sizeof(ondisk->object_prefix),
1023 /* Allocate the snapshot context and fill it in */
1025 snap_count = le32_to_cpu(ondisk->snap_count);
1026 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
1029 snapc->seq = le64_to_cpu(ondisk->snap_seq);
1031 struct rbd_image_snap_ondisk *snaps;
1032 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
1034 /* We'll keep a copy of the snapshot names... */
1036 if (snap_names_len > (u64)SIZE_MAX)
1038 snap_names = kmalloc(snap_names_len, GFP_KERNEL);
1042 /* ...as well as the array of their sizes. */
1043 snap_sizes = kmalloc_array(snap_count,
1044 sizeof(*header->snap_sizes),
1050 * Copy the names, and fill in each snapshot's id
1053 * Note that rbd_dev_v1_header_info() guarantees the
1054 * ondisk buffer we're working with has
1055 * snap_names_len bytes beyond the end of the
1056 * snapshot id array, this memcpy() is safe.
1058 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
1059 snaps = ondisk->snaps;
1060 for (i = 0; i < snap_count; i++) {
1061 snapc->snaps[i] = le64_to_cpu(snaps[i].id);
1062 snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
1066 /* We won't fail any more, fill in the header */
1069 header->object_prefix = object_prefix;
1070 header->obj_order = ondisk->options.order;
1071 rbd_init_layout(rbd_dev);
1073 ceph_put_snap_context(header->snapc);
1074 kfree(header->snap_names);
1075 kfree(header->snap_sizes);
1078 /* The remaining fields always get updated (when we refresh) */
1080 header->image_size = le64_to_cpu(ondisk->image_size);
1081 header->snapc = snapc;
1082 header->snap_names = snap_names;
1083 header->snap_sizes = snap_sizes;
1091 ceph_put_snap_context(snapc);
1092 kfree(object_prefix);
1097 static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1099 const char *snap_name;
1101 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1103 /* Skip over names until we find the one we are looking for */
1105 snap_name = rbd_dev->header.snap_names;
1107 snap_name += strlen(snap_name) + 1;
1109 return kstrdup(snap_name, GFP_KERNEL);
1113 * Snapshot id comparison function for use with qsort()/bsearch().
1114 * Note that result is for snapshots in *descending* order.
1116 static int snapid_compare_reverse(const void *s1, const void *s2)
1118 u64 snap_id1 = *(u64 *)s1;
1119 u64 snap_id2 = *(u64 *)s2;
1121 if (snap_id1 < snap_id2)
1123 return snap_id1 == snap_id2 ? 0 : -1;
1127 * Search a snapshot context to see if the given snapshot id is
1130 * Returns the position of the snapshot id in the array if it's found,
1131 * or BAD_SNAP_INDEX otherwise.
1133 * Note: The snapshot array is in kept sorted (by the osd) in
1134 * reverse order, highest snapshot id first.
1136 static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1138 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1141 found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1142 sizeof (snap_id), snapid_compare_reverse);
1144 return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1147 static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1151 const char *snap_name;
1153 which = rbd_dev_snap_index(rbd_dev, snap_id);
1154 if (which == BAD_SNAP_INDEX)
1155 return ERR_PTR(-ENOENT);
1157 snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1158 return snap_name ? snap_name : ERR_PTR(-ENOMEM);
1161 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1163 if (snap_id == CEPH_NOSNAP)
1164 return RBD_SNAP_HEAD_NAME;
1166 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1167 if (rbd_dev->image_format == 1)
1168 return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1170 return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1173 static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1176 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1177 if (snap_id == CEPH_NOSNAP) {
1178 *snap_size = rbd_dev->header.image_size;
1179 } else if (rbd_dev->image_format == 1) {
1182 which = rbd_dev_snap_index(rbd_dev, snap_id);
1183 if (which == BAD_SNAP_INDEX)
1186 *snap_size = rbd_dev->header.snap_sizes[which];
1191 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1200 static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1202 u64 snap_id = rbd_dev->spec->snap_id;
1206 ret = rbd_snap_size(rbd_dev, snap_id, &size);
1210 rbd_dev->mapping.size = size;
1214 static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1216 rbd_dev->mapping.size = 0;
1219 static void zero_bios(struct ceph_bio_iter *bio_pos, u32 off, u32 bytes)
1221 struct ceph_bio_iter it = *bio_pos;
1223 ceph_bio_iter_advance(&it, off);
1224 ceph_bio_iter_advance_step(&it, bytes, ({
1229 static void zero_bvecs(struct ceph_bvec_iter *bvec_pos, u32 off, u32 bytes)
1231 struct ceph_bvec_iter it = *bvec_pos;
1233 ceph_bvec_iter_advance(&it, off);
1234 ceph_bvec_iter_advance_step(&it, bytes, ({
1240 * Zero a range in @obj_req data buffer defined by a bio (list) or
1241 * (private) bio_vec array.
1243 * @off is relative to the start of the data buffer.
1245 static void rbd_obj_zero_range(struct rbd_obj_request *obj_req, u32 off,
1248 dout("%s %p data buf %u~%u\n", __func__, obj_req, off, bytes);
1250 switch (obj_req->img_request->data_type) {
1251 case OBJ_REQUEST_BIO:
1252 zero_bios(&obj_req->bio_pos, off, bytes);
1254 case OBJ_REQUEST_BVECS:
1255 case OBJ_REQUEST_OWN_BVECS:
1256 zero_bvecs(&obj_req->bvec_pos, off, bytes);
1263 static void rbd_obj_request_destroy(struct kref *kref);
1264 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1266 rbd_assert(obj_request != NULL);
1267 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1268 kref_read(&obj_request->kref));
1269 kref_put(&obj_request->kref, rbd_obj_request_destroy);
1272 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1273 struct rbd_obj_request *obj_request)
1275 rbd_assert(obj_request->img_request == NULL);
1277 /* Image request now owns object's original reference */
1278 obj_request->img_request = img_request;
1279 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1282 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1283 struct rbd_obj_request *obj_request)
1285 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1286 list_del(&obj_request->ex.oe_item);
1287 rbd_assert(obj_request->img_request == img_request);
1288 rbd_obj_request_put(obj_request);
1291 static void rbd_osd_submit(struct ceph_osd_request *osd_req)
1293 struct rbd_obj_request *obj_req = osd_req->r_priv;
1295 dout("%s osd_req %p for obj_req %p objno %llu %llu~%llu\n",
1296 __func__, osd_req, obj_req, obj_req->ex.oe_objno,
1297 obj_req->ex.oe_off, obj_req->ex.oe_len);
1298 ceph_osdc_start_request(osd_req->r_osdc, osd_req);
1302 * The default/initial value for all image request flags is 0. Each
1303 * is conditionally set to 1 at image request initialization time
1304 * and currently never change thereafter.
1306 static void img_request_layered_set(struct rbd_img_request *img_request)
1308 set_bit(IMG_REQ_LAYERED, &img_request->flags);
1311 static bool img_request_layered_test(struct rbd_img_request *img_request)
1313 return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1316 static bool rbd_obj_is_entire(struct rbd_obj_request *obj_req)
1318 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1320 return !obj_req->ex.oe_off &&
1321 obj_req->ex.oe_len == rbd_dev->layout.object_size;
1324 static bool rbd_obj_is_tail(struct rbd_obj_request *obj_req)
1326 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1328 return obj_req->ex.oe_off + obj_req->ex.oe_len ==
1329 rbd_dev->layout.object_size;
1333 * Must be called after rbd_obj_calc_img_extents().
1335 static void rbd_obj_set_copyup_enabled(struct rbd_obj_request *obj_req)
1337 rbd_assert(obj_req->img_request->snapc);
1339 if (obj_req->img_request->op_type == OBJ_OP_DISCARD) {
1340 dout("%s %p objno %llu discard\n", __func__, obj_req,
1341 obj_req->ex.oe_objno);
1345 if (!obj_req->num_img_extents) {
1346 dout("%s %p objno %llu not overlapping\n", __func__, obj_req,
1347 obj_req->ex.oe_objno);
1351 if (rbd_obj_is_entire(obj_req) &&
1352 !obj_req->img_request->snapc->num_snaps) {
1353 dout("%s %p objno %llu entire\n", __func__, obj_req,
1354 obj_req->ex.oe_objno);
1358 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
1361 static u64 rbd_obj_img_extents_bytes(struct rbd_obj_request *obj_req)
1363 return ceph_file_extents_bytes(obj_req->img_extents,
1364 obj_req->num_img_extents);
1367 static bool rbd_img_is_write(struct rbd_img_request *img_req)
1369 switch (img_req->op_type) {
1373 case OBJ_OP_DISCARD:
1374 case OBJ_OP_ZEROOUT:
1381 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
1383 struct rbd_obj_request *obj_req = osd_req->r_priv;
1386 dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
1387 osd_req->r_result, obj_req);
1390 * Writes aren't allowed to return a data payload. In some
1391 * guarded write cases (e.g. stat + zero on an empty object)
1392 * a stat response makes it through, but we don't care.
1394 if (osd_req->r_result > 0 && rbd_img_is_write(obj_req->img_request))
1397 result = osd_req->r_result;
1399 rbd_obj_handle_request(obj_req, result);
1402 static void rbd_osd_format_read(struct ceph_osd_request *osd_req)
1404 struct rbd_obj_request *obj_request = osd_req->r_priv;
1405 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1406 struct ceph_options *opt = rbd_dev->rbd_client->client->options;
1408 osd_req->r_flags = CEPH_OSD_FLAG_READ | opt->read_from_replica;
1409 osd_req->r_snapid = obj_request->img_request->snap_id;
1412 static void rbd_osd_format_write(struct ceph_osd_request *osd_req)
1414 struct rbd_obj_request *obj_request = osd_req->r_priv;
1416 osd_req->r_flags = CEPH_OSD_FLAG_WRITE;
1417 ktime_get_real_ts64(&osd_req->r_mtime);
1418 osd_req->r_data_offset = obj_request->ex.oe_off;
1421 static struct ceph_osd_request *
1422 __rbd_obj_add_osd_request(struct rbd_obj_request *obj_req,
1423 struct ceph_snap_context *snapc, int num_ops)
1425 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1426 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1427 struct ceph_osd_request *req;
1428 const char *name_format = rbd_dev->image_format == 1 ?
1429 RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
1432 req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, GFP_NOIO);
1434 return ERR_PTR(-ENOMEM);
1436 list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
1437 req->r_callback = rbd_osd_req_callback;
1438 req->r_priv = obj_req;
1441 * Data objects may be stored in a separate pool, but always in
1442 * the same namespace in that pool as the header in its pool.
1444 ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
1445 req->r_base_oloc.pool = rbd_dev->layout.pool_id;
1447 ret = ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
1448 rbd_dev->header.object_prefix,
1449 obj_req->ex.oe_objno);
1451 return ERR_PTR(ret);
1456 static struct ceph_osd_request *
1457 rbd_obj_add_osd_request(struct rbd_obj_request *obj_req, int num_ops)
1459 rbd_assert(obj_req->img_request->snapc);
1460 return __rbd_obj_add_osd_request(obj_req, obj_req->img_request->snapc,
1464 static struct rbd_obj_request *rbd_obj_request_create(void)
1466 struct rbd_obj_request *obj_request;
1468 obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_NOIO);
1472 ceph_object_extent_init(&obj_request->ex);
1473 INIT_LIST_HEAD(&obj_request->osd_reqs);
1474 mutex_init(&obj_request->state_mutex);
1475 kref_init(&obj_request->kref);
1477 dout("%s %p\n", __func__, obj_request);
1481 static void rbd_obj_request_destroy(struct kref *kref)
1483 struct rbd_obj_request *obj_request;
1484 struct ceph_osd_request *osd_req;
1487 obj_request = container_of(kref, struct rbd_obj_request, kref);
1489 dout("%s: obj %p\n", __func__, obj_request);
1491 while (!list_empty(&obj_request->osd_reqs)) {
1492 osd_req = list_first_entry(&obj_request->osd_reqs,
1493 struct ceph_osd_request, r_private_item);
1494 list_del_init(&osd_req->r_private_item);
1495 ceph_osdc_put_request(osd_req);
1498 switch (obj_request->img_request->data_type) {
1499 case OBJ_REQUEST_NODATA:
1500 case OBJ_REQUEST_BIO:
1501 case OBJ_REQUEST_BVECS:
1502 break; /* Nothing to do */
1503 case OBJ_REQUEST_OWN_BVECS:
1504 kfree(obj_request->bvec_pos.bvecs);
1510 kfree(obj_request->img_extents);
1511 if (obj_request->copyup_bvecs) {
1512 for (i = 0; i < obj_request->copyup_bvec_count; i++) {
1513 if (obj_request->copyup_bvecs[i].bv_page)
1514 __free_page(obj_request->copyup_bvecs[i].bv_page);
1516 kfree(obj_request->copyup_bvecs);
1519 kmem_cache_free(rbd_obj_request_cache, obj_request);
1522 /* It's OK to call this for a device with no parent */
1524 static void rbd_spec_put(struct rbd_spec *spec);
1525 static void rbd_dev_unparent(struct rbd_device *rbd_dev)
1527 rbd_dev_remove_parent(rbd_dev);
1528 rbd_spec_put(rbd_dev->parent_spec);
1529 rbd_dev->parent_spec = NULL;
1530 rbd_dev->parent_overlap = 0;
1534 * Parent image reference counting is used to determine when an
1535 * image's parent fields can be safely torn down--after there are no
1536 * more in-flight requests to the parent image. When the last
1537 * reference is dropped, cleaning them up is safe.
1539 static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
1543 if (!rbd_dev->parent_spec)
1546 counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
1550 /* Last reference; clean up parent data structures */
1553 rbd_dev_unparent(rbd_dev);
1555 rbd_warn(rbd_dev, "parent reference underflow");
1559 * If an image has a non-zero parent overlap, get a reference to its
1562 * Returns true if the rbd device has a parent with a non-zero
1563 * overlap and a reference for it was successfully taken, or
1566 static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
1570 if (!rbd_dev->parent_spec)
1573 if (rbd_dev->parent_overlap)
1574 counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
1577 rbd_warn(rbd_dev, "parent reference overflow");
1582 static void rbd_img_request_init(struct rbd_img_request *img_request,
1583 struct rbd_device *rbd_dev,
1584 enum obj_operation_type op_type)
1586 memset(img_request, 0, sizeof(*img_request));
1588 img_request->rbd_dev = rbd_dev;
1589 img_request->op_type = op_type;
1591 INIT_LIST_HEAD(&img_request->lock_item);
1592 INIT_LIST_HEAD(&img_request->object_extents);
1593 mutex_init(&img_request->state_mutex);
1597 * Only snap_id is captured here, for reads. For writes, snapshot
1598 * context is captured in rbd_img_object_requests() after exclusive
1599 * lock is ensured to be held.
1601 static void rbd_img_capture_header(struct rbd_img_request *img_req)
1603 struct rbd_device *rbd_dev = img_req->rbd_dev;
1605 lockdep_assert_held(&rbd_dev->header_rwsem);
1607 if (!rbd_img_is_write(img_req))
1608 img_req->snap_id = rbd_dev->spec->snap_id;
1610 if (rbd_dev_parent_get(rbd_dev))
1611 img_request_layered_set(img_req);
1614 static void rbd_img_request_destroy(struct rbd_img_request *img_request)
1616 struct rbd_obj_request *obj_request;
1617 struct rbd_obj_request *next_obj_request;
1619 dout("%s: img %p\n", __func__, img_request);
1621 WARN_ON(!list_empty(&img_request->lock_item));
1622 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1623 rbd_img_obj_request_del(img_request, obj_request);
1625 if (img_request_layered_test(img_request))
1626 rbd_dev_parent_put(img_request->rbd_dev);
1628 if (rbd_img_is_write(img_request))
1629 ceph_put_snap_context(img_request->snapc);
1631 if (test_bit(IMG_REQ_CHILD, &img_request->flags))
1632 kmem_cache_free(rbd_img_request_cache, img_request);
1635 #define BITS_PER_OBJ 2
1636 #define OBJS_PER_BYTE (BITS_PER_BYTE / BITS_PER_OBJ)
1637 #define OBJ_MASK ((1 << BITS_PER_OBJ) - 1)
1639 static void __rbd_object_map_index(struct rbd_device *rbd_dev, u64 objno,
1640 u64 *index, u8 *shift)
1644 rbd_assert(objno < rbd_dev->object_map_size);
1645 *index = div_u64_rem(objno, OBJS_PER_BYTE, &off);
1646 *shift = (OBJS_PER_BYTE - off - 1) * BITS_PER_OBJ;
1649 static u8 __rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1654 lockdep_assert_held(&rbd_dev->object_map_lock);
1655 __rbd_object_map_index(rbd_dev, objno, &index, &shift);
1656 return (rbd_dev->object_map[index] >> shift) & OBJ_MASK;
1659 static void __rbd_object_map_set(struct rbd_device *rbd_dev, u64 objno, u8 val)
1665 lockdep_assert_held(&rbd_dev->object_map_lock);
1666 rbd_assert(!(val & ~OBJ_MASK));
1668 __rbd_object_map_index(rbd_dev, objno, &index, &shift);
1669 p = &rbd_dev->object_map[index];
1670 *p = (*p & ~(OBJ_MASK << shift)) | (val << shift);
1673 static u8 rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1677 spin_lock(&rbd_dev->object_map_lock);
1678 state = __rbd_object_map_get(rbd_dev, objno);
1679 spin_unlock(&rbd_dev->object_map_lock);
1683 static bool use_object_map(struct rbd_device *rbd_dev)
1686 * An image mapped read-only can't use the object map -- it isn't
1687 * loaded because the header lock isn't acquired. Someone else can
1688 * write to the image and update the object map behind our back.
1690 * A snapshot can't be written to, so using the object map is always
1693 if (!rbd_is_snap(rbd_dev) && rbd_is_ro(rbd_dev))
1696 return ((rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) &&
1697 !(rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID));
1700 static bool rbd_object_map_may_exist(struct rbd_device *rbd_dev, u64 objno)
1704 /* fall back to default logic if object map is disabled or invalid */
1705 if (!use_object_map(rbd_dev))
1708 state = rbd_object_map_get(rbd_dev, objno);
1709 return state != OBJECT_NONEXISTENT;
1712 static void rbd_object_map_name(struct rbd_device *rbd_dev, u64 snap_id,
1713 struct ceph_object_id *oid)
1715 if (snap_id == CEPH_NOSNAP)
1716 ceph_oid_printf(oid, "%s%s", RBD_OBJECT_MAP_PREFIX,
1717 rbd_dev->spec->image_id);
1719 ceph_oid_printf(oid, "%s%s.%016llx", RBD_OBJECT_MAP_PREFIX,
1720 rbd_dev->spec->image_id, snap_id);
1723 static int rbd_object_map_lock(struct rbd_device *rbd_dev)
1725 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1726 CEPH_DEFINE_OID_ONSTACK(oid);
1729 struct ceph_locker *lockers;
1731 bool broke_lock = false;
1734 rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1737 ret = ceph_cls_lock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1738 CEPH_CLS_LOCK_EXCLUSIVE, "", "", "", 0);
1739 if (ret != -EBUSY || broke_lock) {
1741 ret = 0; /* already locked by myself */
1743 rbd_warn(rbd_dev, "failed to lock object map: %d", ret);
1747 ret = ceph_cls_lock_info(osdc, &oid, &rbd_dev->header_oloc,
1748 RBD_LOCK_NAME, &lock_type, &lock_tag,
1749 &lockers, &num_lockers);
1754 rbd_warn(rbd_dev, "failed to get object map lockers: %d", ret);
1759 if (num_lockers == 0)
1762 rbd_warn(rbd_dev, "breaking object map lock owned by %s%llu",
1763 ENTITY_NAME(lockers[0].id.name));
1765 ret = ceph_cls_break_lock(osdc, &oid, &rbd_dev->header_oloc,
1766 RBD_LOCK_NAME, lockers[0].id.cookie,
1767 &lockers[0].id.name);
1768 ceph_free_lockers(lockers, num_lockers);
1773 rbd_warn(rbd_dev, "failed to break object map lock: %d", ret);
1781 static void rbd_object_map_unlock(struct rbd_device *rbd_dev)
1783 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1784 CEPH_DEFINE_OID_ONSTACK(oid);
1787 rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1789 ret = ceph_cls_unlock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1791 if (ret && ret != -ENOENT)
1792 rbd_warn(rbd_dev, "failed to unlock object map: %d", ret);
1795 static int decode_object_map_header(void **p, void *end, u64 *object_map_size)
1803 ceph_decode_32_safe(p, end, header_len, e_inval);
1804 header_end = *p + header_len;
1806 ret = ceph_start_decoding(p, end, 1, "BitVector header", &struct_v,
1811 ceph_decode_64_safe(p, end, *object_map_size, e_inval);
1820 static int __rbd_object_map_load(struct rbd_device *rbd_dev)
1822 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1823 CEPH_DEFINE_OID_ONSTACK(oid);
1824 struct page **pages;
1828 u64 object_map_bytes;
1829 u64 object_map_size;
1833 rbd_assert(!rbd_dev->object_map && !rbd_dev->object_map_size);
1835 num_objects = ceph_get_num_objects(&rbd_dev->layout,
1836 rbd_dev->mapping.size);
1837 object_map_bytes = DIV_ROUND_UP_ULL(num_objects * BITS_PER_OBJ,
1839 num_pages = calc_pages_for(0, object_map_bytes) + 1;
1840 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1842 return PTR_ERR(pages);
1844 reply_len = num_pages * PAGE_SIZE;
1845 rbd_object_map_name(rbd_dev, rbd_dev->spec->snap_id, &oid);
1846 ret = ceph_osdc_call(osdc, &oid, &rbd_dev->header_oloc,
1847 "rbd", "object_map_load", CEPH_OSD_FLAG_READ,
1848 NULL, 0, pages, &reply_len);
1852 p = page_address(pages[0]);
1853 end = p + min(reply_len, (size_t)PAGE_SIZE);
1854 ret = decode_object_map_header(&p, end, &object_map_size);
1858 if (object_map_size != num_objects) {
1859 rbd_warn(rbd_dev, "object map size mismatch: %llu vs %llu",
1860 object_map_size, num_objects);
1865 if (offset_in_page(p) + object_map_bytes > reply_len) {
1870 rbd_dev->object_map = kvmalloc(object_map_bytes, GFP_KERNEL);
1871 if (!rbd_dev->object_map) {
1876 rbd_dev->object_map_size = object_map_size;
1877 ceph_copy_from_page_vector(pages, rbd_dev->object_map,
1878 offset_in_page(p), object_map_bytes);
1881 ceph_release_page_vector(pages, num_pages);
1885 static void rbd_object_map_free(struct rbd_device *rbd_dev)
1887 kvfree(rbd_dev->object_map);
1888 rbd_dev->object_map = NULL;
1889 rbd_dev->object_map_size = 0;
1892 static int rbd_object_map_load(struct rbd_device *rbd_dev)
1896 ret = __rbd_object_map_load(rbd_dev);
1900 ret = rbd_dev_v2_get_flags(rbd_dev);
1902 rbd_object_map_free(rbd_dev);
1906 if (rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID)
1907 rbd_warn(rbd_dev, "object map is invalid");
1912 static int rbd_object_map_open(struct rbd_device *rbd_dev)
1916 ret = rbd_object_map_lock(rbd_dev);
1920 ret = rbd_object_map_load(rbd_dev);
1922 rbd_object_map_unlock(rbd_dev);
1929 static void rbd_object_map_close(struct rbd_device *rbd_dev)
1931 rbd_object_map_free(rbd_dev);
1932 rbd_object_map_unlock(rbd_dev);
1936 * This function needs snap_id (or more precisely just something to
1937 * distinguish between HEAD and snapshot object maps), new_state and
1938 * current_state that were passed to rbd_object_map_update().
1940 * To avoid allocating and stashing a context we piggyback on the OSD
1941 * request. A HEAD update has two ops (assert_locked). For new_state
1942 * and current_state we decode our own object_map_update op, encoded in
1943 * rbd_cls_object_map_update().
1945 static int rbd_object_map_update_finish(struct rbd_obj_request *obj_req,
1946 struct ceph_osd_request *osd_req)
1948 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1949 struct ceph_osd_data *osd_data;
1951 u8 state, new_state, current_state;
1952 bool has_current_state;
1955 if (osd_req->r_result)
1956 return osd_req->r_result;
1959 * Nothing to do for a snapshot object map.
1961 if (osd_req->r_num_ops == 1)
1965 * Update in-memory HEAD object map.
1967 rbd_assert(osd_req->r_num_ops == 2);
1968 osd_data = osd_req_op_data(osd_req, 1, cls, request_data);
1969 rbd_assert(osd_data->type == CEPH_OSD_DATA_TYPE_PAGES);
1971 p = page_address(osd_data->pages[0]);
1972 objno = ceph_decode_64(&p);
1973 rbd_assert(objno == obj_req->ex.oe_objno);
1974 rbd_assert(ceph_decode_64(&p) == objno + 1);
1975 new_state = ceph_decode_8(&p);
1976 has_current_state = ceph_decode_8(&p);
1977 if (has_current_state)
1978 current_state = ceph_decode_8(&p);
1980 spin_lock(&rbd_dev->object_map_lock);
1981 state = __rbd_object_map_get(rbd_dev, objno);
1982 if (!has_current_state || current_state == state ||
1983 (current_state == OBJECT_EXISTS && state == OBJECT_EXISTS_CLEAN))
1984 __rbd_object_map_set(rbd_dev, objno, new_state);
1985 spin_unlock(&rbd_dev->object_map_lock);
1990 static void rbd_object_map_callback(struct ceph_osd_request *osd_req)
1992 struct rbd_obj_request *obj_req = osd_req->r_priv;
1995 dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
1996 osd_req->r_result, obj_req);
1998 result = rbd_object_map_update_finish(obj_req, osd_req);
1999 rbd_obj_handle_request(obj_req, result);
2002 static bool update_needed(struct rbd_device *rbd_dev, u64 objno, u8 new_state)
2004 u8 state = rbd_object_map_get(rbd_dev, objno);
2006 if (state == new_state ||
2007 (new_state == OBJECT_PENDING && state == OBJECT_NONEXISTENT) ||
2008 (new_state == OBJECT_NONEXISTENT && state != OBJECT_PENDING))
2014 static int rbd_cls_object_map_update(struct ceph_osd_request *req,
2015 int which, u64 objno, u8 new_state,
2016 const u8 *current_state)
2018 struct page **pages;
2022 ret = osd_req_op_cls_init(req, which, "rbd", "object_map_update");
2026 pages = ceph_alloc_page_vector(1, GFP_NOIO);
2028 return PTR_ERR(pages);
2030 p = start = page_address(pages[0]);
2031 ceph_encode_64(&p, objno);
2032 ceph_encode_64(&p, objno + 1);
2033 ceph_encode_8(&p, new_state);
2034 if (current_state) {
2035 ceph_encode_8(&p, 1);
2036 ceph_encode_8(&p, *current_state);
2038 ceph_encode_8(&p, 0);
2041 osd_req_op_cls_request_data_pages(req, which, pages, p - start, 0,
2048 * 0 - object map update sent
2049 * 1 - object map update isn't needed
2052 static int rbd_object_map_update(struct rbd_obj_request *obj_req, u64 snap_id,
2053 u8 new_state, const u8 *current_state)
2055 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2056 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2057 struct ceph_osd_request *req;
2062 if (snap_id == CEPH_NOSNAP) {
2063 if (!update_needed(rbd_dev, obj_req->ex.oe_objno, new_state))
2066 num_ops++; /* assert_locked */
2069 req = ceph_osdc_alloc_request(osdc, NULL, num_ops, false, GFP_NOIO);
2073 list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
2074 req->r_callback = rbd_object_map_callback;
2075 req->r_priv = obj_req;
2077 rbd_object_map_name(rbd_dev, snap_id, &req->r_base_oid);
2078 ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
2079 req->r_flags = CEPH_OSD_FLAG_WRITE;
2080 ktime_get_real_ts64(&req->r_mtime);
2082 if (snap_id == CEPH_NOSNAP) {
2084 * Protect against possible race conditions during lock
2085 * ownership transitions.
2087 ret = ceph_cls_assert_locked(req, which++, RBD_LOCK_NAME,
2088 CEPH_CLS_LOCK_EXCLUSIVE, "", "");
2093 ret = rbd_cls_object_map_update(req, which, obj_req->ex.oe_objno,
2094 new_state, current_state);
2098 ret = ceph_osdc_alloc_messages(req, GFP_NOIO);
2102 ceph_osdc_start_request(osdc, req);
2106 static void prune_extents(struct ceph_file_extent *img_extents,
2107 u32 *num_img_extents, u64 overlap)
2109 u32 cnt = *num_img_extents;
2111 /* drop extents completely beyond the overlap */
2112 while (cnt && img_extents[cnt - 1].fe_off >= overlap)
2116 struct ceph_file_extent *ex = &img_extents[cnt - 1];
2118 /* trim final overlapping extent */
2119 if (ex->fe_off + ex->fe_len > overlap)
2120 ex->fe_len = overlap - ex->fe_off;
2123 *num_img_extents = cnt;
2127 * Determine the byte range(s) covered by either just the object extent
2128 * or the entire object in the parent image.
2130 static int rbd_obj_calc_img_extents(struct rbd_obj_request *obj_req,
2133 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2136 if (!rbd_dev->parent_overlap)
2139 ret = ceph_extent_to_file(&rbd_dev->layout, obj_req->ex.oe_objno,
2140 entire ? 0 : obj_req->ex.oe_off,
2141 entire ? rbd_dev->layout.object_size :
2143 &obj_req->img_extents,
2144 &obj_req->num_img_extents);
2148 prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
2149 rbd_dev->parent_overlap);
2153 static void rbd_osd_setup_data(struct ceph_osd_request *osd_req, int which)
2155 struct rbd_obj_request *obj_req = osd_req->r_priv;
2157 switch (obj_req->img_request->data_type) {
2158 case OBJ_REQUEST_BIO:
2159 osd_req_op_extent_osd_data_bio(osd_req, which,
2161 obj_req->ex.oe_len);
2163 case OBJ_REQUEST_BVECS:
2164 case OBJ_REQUEST_OWN_BVECS:
2165 rbd_assert(obj_req->bvec_pos.iter.bi_size ==
2166 obj_req->ex.oe_len);
2167 rbd_assert(obj_req->bvec_idx == obj_req->bvec_count);
2168 osd_req_op_extent_osd_data_bvec_pos(osd_req, which,
2169 &obj_req->bvec_pos);
2176 static int rbd_osd_setup_stat(struct ceph_osd_request *osd_req, int which)
2178 struct page **pages;
2181 * The response data for a STAT call consists of:
2188 pages = ceph_alloc_page_vector(1, GFP_NOIO);
2190 return PTR_ERR(pages);
2192 osd_req_op_init(osd_req, which, CEPH_OSD_OP_STAT, 0);
2193 osd_req_op_raw_data_in_pages(osd_req, which, pages,
2194 8 + sizeof(struct ceph_timespec),
2199 static int rbd_osd_setup_copyup(struct ceph_osd_request *osd_req, int which,
2202 struct rbd_obj_request *obj_req = osd_req->r_priv;
2205 ret = osd_req_op_cls_init(osd_req, which, "rbd", "copyup");
2209 osd_req_op_cls_request_data_bvecs(osd_req, which, obj_req->copyup_bvecs,
2210 obj_req->copyup_bvec_count, bytes);
2214 static int rbd_obj_init_read(struct rbd_obj_request *obj_req)
2216 obj_req->read_state = RBD_OBJ_READ_START;
2220 static void __rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2223 struct rbd_obj_request *obj_req = osd_req->r_priv;
2224 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2227 if (!use_object_map(rbd_dev) ||
2228 !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST)) {
2229 osd_req_op_alloc_hint_init(osd_req, which++,
2230 rbd_dev->layout.object_size,
2231 rbd_dev->layout.object_size,
2232 rbd_dev->opts->alloc_hint_flags);
2235 if (rbd_obj_is_entire(obj_req))
2236 opcode = CEPH_OSD_OP_WRITEFULL;
2238 opcode = CEPH_OSD_OP_WRITE;
2240 osd_req_op_extent_init(osd_req, which, opcode,
2241 obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2242 rbd_osd_setup_data(osd_req, which);
2245 static int rbd_obj_init_write(struct rbd_obj_request *obj_req)
2249 /* reverse map the entire object onto the parent */
2250 ret = rbd_obj_calc_img_extents(obj_req, true);
2254 obj_req->write_state = RBD_OBJ_WRITE_START;
2258 static u16 truncate_or_zero_opcode(struct rbd_obj_request *obj_req)
2260 return rbd_obj_is_tail(obj_req) ? CEPH_OSD_OP_TRUNCATE :
2264 static void __rbd_osd_setup_discard_ops(struct ceph_osd_request *osd_req,
2267 struct rbd_obj_request *obj_req = osd_req->r_priv;
2269 if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
2270 rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2271 osd_req_op_init(osd_req, which, CEPH_OSD_OP_DELETE, 0);
2273 osd_req_op_extent_init(osd_req, which,
2274 truncate_or_zero_opcode(obj_req),
2275 obj_req->ex.oe_off, obj_req->ex.oe_len,
2280 static int rbd_obj_init_discard(struct rbd_obj_request *obj_req)
2282 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2287 * Align the range to alloc_size boundary and punt on discards
2288 * that are too small to free up any space.
2290 * alloc_size == object_size && is_tail() is a special case for
2291 * filestore with filestore_punch_hole = false, needed to allow
2292 * truncate (in addition to delete).
2294 if (rbd_dev->opts->alloc_size != rbd_dev->layout.object_size ||
2295 !rbd_obj_is_tail(obj_req)) {
2296 off = round_up(obj_req->ex.oe_off, rbd_dev->opts->alloc_size);
2297 next_off = round_down(obj_req->ex.oe_off + obj_req->ex.oe_len,
2298 rbd_dev->opts->alloc_size);
2299 if (off >= next_off)
2302 dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
2303 obj_req, obj_req->ex.oe_off, obj_req->ex.oe_len,
2304 off, next_off - off);
2305 obj_req->ex.oe_off = off;
2306 obj_req->ex.oe_len = next_off - off;
2309 /* reverse map the entire object onto the parent */
2310 ret = rbd_obj_calc_img_extents(obj_req, true);
2314 obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2315 if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents)
2316 obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2318 obj_req->write_state = RBD_OBJ_WRITE_START;
2322 static void __rbd_osd_setup_zeroout_ops(struct ceph_osd_request *osd_req,
2325 struct rbd_obj_request *obj_req = osd_req->r_priv;
2328 if (rbd_obj_is_entire(obj_req)) {
2329 if (obj_req->num_img_extents) {
2330 if (!(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2331 osd_req_op_init(osd_req, which++,
2332 CEPH_OSD_OP_CREATE, 0);
2333 opcode = CEPH_OSD_OP_TRUNCATE;
2335 rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2336 osd_req_op_init(osd_req, which++,
2337 CEPH_OSD_OP_DELETE, 0);
2341 opcode = truncate_or_zero_opcode(obj_req);
2345 osd_req_op_extent_init(osd_req, which, opcode,
2346 obj_req->ex.oe_off, obj_req->ex.oe_len,
2350 static int rbd_obj_init_zeroout(struct rbd_obj_request *obj_req)
2354 /* reverse map the entire object onto the parent */
2355 ret = rbd_obj_calc_img_extents(obj_req, true);
2359 if (!obj_req->num_img_extents) {
2360 obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2361 if (rbd_obj_is_entire(obj_req))
2362 obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2365 obj_req->write_state = RBD_OBJ_WRITE_START;
2369 static int count_write_ops(struct rbd_obj_request *obj_req)
2371 struct rbd_img_request *img_req = obj_req->img_request;
2373 switch (img_req->op_type) {
2375 if (!use_object_map(img_req->rbd_dev) ||
2376 !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST))
2377 return 2; /* setallochint + write/writefull */
2379 return 1; /* write/writefull */
2380 case OBJ_OP_DISCARD:
2381 return 1; /* delete/truncate/zero */
2382 case OBJ_OP_ZEROOUT:
2383 if (rbd_obj_is_entire(obj_req) && obj_req->num_img_extents &&
2384 !(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2385 return 2; /* create + truncate */
2387 return 1; /* delete/truncate/zero */
2393 static void rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2396 struct rbd_obj_request *obj_req = osd_req->r_priv;
2398 switch (obj_req->img_request->op_type) {
2400 __rbd_osd_setup_write_ops(osd_req, which);
2402 case OBJ_OP_DISCARD:
2403 __rbd_osd_setup_discard_ops(osd_req, which);
2405 case OBJ_OP_ZEROOUT:
2406 __rbd_osd_setup_zeroout_ops(osd_req, which);
2414 * Prune the list of object requests (adjust offset and/or length, drop
2415 * redundant requests). Prepare object request state machines and image
2416 * request state machine for execution.
2418 static int __rbd_img_fill_request(struct rbd_img_request *img_req)
2420 struct rbd_obj_request *obj_req, *next_obj_req;
2423 for_each_obj_request_safe(img_req, obj_req, next_obj_req) {
2424 switch (img_req->op_type) {
2426 ret = rbd_obj_init_read(obj_req);
2429 ret = rbd_obj_init_write(obj_req);
2431 case OBJ_OP_DISCARD:
2432 ret = rbd_obj_init_discard(obj_req);
2434 case OBJ_OP_ZEROOUT:
2435 ret = rbd_obj_init_zeroout(obj_req);
2443 rbd_img_obj_request_del(img_req, obj_req);
2448 img_req->state = RBD_IMG_START;
2452 union rbd_img_fill_iter {
2453 struct ceph_bio_iter bio_iter;
2454 struct ceph_bvec_iter bvec_iter;
2457 struct rbd_img_fill_ctx {
2458 enum obj_request_type pos_type;
2459 union rbd_img_fill_iter *pos;
2460 union rbd_img_fill_iter iter;
2461 ceph_object_extent_fn_t set_pos_fn;
2462 ceph_object_extent_fn_t count_fn;
2463 ceph_object_extent_fn_t copy_fn;
2466 static struct ceph_object_extent *alloc_object_extent(void *arg)
2468 struct rbd_img_request *img_req = arg;
2469 struct rbd_obj_request *obj_req;
2471 obj_req = rbd_obj_request_create();
2475 rbd_img_obj_request_add(img_req, obj_req);
2476 return &obj_req->ex;
2480 * While su != os && sc == 1 is technically not fancy (it's the same
2481 * layout as su == os && sc == 1), we can't use the nocopy path for it
2482 * because ->set_pos_fn() should be called only once per object.
2483 * ceph_file_to_extents() invokes action_fn once per stripe unit, so
2484 * treat su != os && sc == 1 as fancy.
2486 static bool rbd_layout_is_fancy(struct ceph_file_layout *l)
2488 return l->stripe_unit != l->object_size;
2491 static int rbd_img_fill_request_nocopy(struct rbd_img_request *img_req,
2492 struct ceph_file_extent *img_extents,
2493 u32 num_img_extents,
2494 struct rbd_img_fill_ctx *fctx)
2499 img_req->data_type = fctx->pos_type;
2502 * Create object requests and set each object request's starting
2503 * position in the provided bio (list) or bio_vec array.
2505 fctx->iter = *fctx->pos;
2506 for (i = 0; i < num_img_extents; i++) {
2507 ret = ceph_file_to_extents(&img_req->rbd_dev->layout,
2508 img_extents[i].fe_off,
2509 img_extents[i].fe_len,
2510 &img_req->object_extents,
2511 alloc_object_extent, img_req,
2512 fctx->set_pos_fn, &fctx->iter);
2517 return __rbd_img_fill_request(img_req);
2521 * Map a list of image extents to a list of object extents, create the
2522 * corresponding object requests (normally each to a different object,
2523 * but not always) and add them to @img_req. For each object request,
2524 * set up its data descriptor to point to the corresponding chunk(s) of
2525 * @fctx->pos data buffer.
2527 * Because ceph_file_to_extents() will merge adjacent object extents
2528 * together, each object request's data descriptor may point to multiple
2529 * different chunks of @fctx->pos data buffer.
2531 * @fctx->pos data buffer is assumed to be large enough.
2533 static int rbd_img_fill_request(struct rbd_img_request *img_req,
2534 struct ceph_file_extent *img_extents,
2535 u32 num_img_extents,
2536 struct rbd_img_fill_ctx *fctx)
2538 struct rbd_device *rbd_dev = img_req->rbd_dev;
2539 struct rbd_obj_request *obj_req;
2543 if (fctx->pos_type == OBJ_REQUEST_NODATA ||
2544 !rbd_layout_is_fancy(&rbd_dev->layout))
2545 return rbd_img_fill_request_nocopy(img_req, img_extents,
2546 num_img_extents, fctx);
2548 img_req->data_type = OBJ_REQUEST_OWN_BVECS;
2551 * Create object requests and determine ->bvec_count for each object
2552 * request. Note that ->bvec_count sum over all object requests may
2553 * be greater than the number of bio_vecs in the provided bio (list)
2554 * or bio_vec array because when mapped, those bio_vecs can straddle
2555 * stripe unit boundaries.
2557 fctx->iter = *fctx->pos;
2558 for (i = 0; i < num_img_extents; i++) {
2559 ret = ceph_file_to_extents(&rbd_dev->layout,
2560 img_extents[i].fe_off,
2561 img_extents[i].fe_len,
2562 &img_req->object_extents,
2563 alloc_object_extent, img_req,
2564 fctx->count_fn, &fctx->iter);
2569 for_each_obj_request(img_req, obj_req) {
2570 obj_req->bvec_pos.bvecs = kmalloc_array(obj_req->bvec_count,
2571 sizeof(*obj_req->bvec_pos.bvecs),
2573 if (!obj_req->bvec_pos.bvecs)
2578 * Fill in each object request's private bio_vec array, splitting and
2579 * rearranging the provided bio_vecs in stripe unit chunks as needed.
2581 fctx->iter = *fctx->pos;
2582 for (i = 0; i < num_img_extents; i++) {
2583 ret = ceph_iterate_extents(&rbd_dev->layout,
2584 img_extents[i].fe_off,
2585 img_extents[i].fe_len,
2586 &img_req->object_extents,
2587 fctx->copy_fn, &fctx->iter);
2592 return __rbd_img_fill_request(img_req);
2595 static int rbd_img_fill_nodata(struct rbd_img_request *img_req,
2598 struct ceph_file_extent ex = { off, len };
2599 union rbd_img_fill_iter dummy = {};
2600 struct rbd_img_fill_ctx fctx = {
2601 .pos_type = OBJ_REQUEST_NODATA,
2605 return rbd_img_fill_request(img_req, &ex, 1, &fctx);
2608 static void set_bio_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2610 struct rbd_obj_request *obj_req =
2611 container_of(ex, struct rbd_obj_request, ex);
2612 struct ceph_bio_iter *it = arg;
2614 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2615 obj_req->bio_pos = *it;
2616 ceph_bio_iter_advance(it, bytes);
2619 static void count_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2621 struct rbd_obj_request *obj_req =
2622 container_of(ex, struct rbd_obj_request, ex);
2623 struct ceph_bio_iter *it = arg;
2625 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2626 ceph_bio_iter_advance_step(it, bytes, ({
2627 obj_req->bvec_count++;
2632 static void copy_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2634 struct rbd_obj_request *obj_req =
2635 container_of(ex, struct rbd_obj_request, ex);
2636 struct ceph_bio_iter *it = arg;
2638 dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2639 ceph_bio_iter_advance_step(it, bytes, ({
2640 obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2641 obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2645 static int __rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2646 struct ceph_file_extent *img_extents,
2647 u32 num_img_extents,
2648 struct ceph_bio_iter *bio_pos)
2650 struct rbd_img_fill_ctx fctx = {
2651 .pos_type = OBJ_REQUEST_BIO,
2652 .pos = (union rbd_img_fill_iter *)bio_pos,
2653 .set_pos_fn = set_bio_pos,
2654 .count_fn = count_bio_bvecs,
2655 .copy_fn = copy_bio_bvecs,
2658 return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2662 static int rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2663 u64 off, u64 len, struct bio *bio)
2665 struct ceph_file_extent ex = { off, len };
2666 struct ceph_bio_iter it = { .bio = bio, .iter = bio->bi_iter };
2668 return __rbd_img_fill_from_bio(img_req, &ex, 1, &it);
2671 static void set_bvec_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2673 struct rbd_obj_request *obj_req =
2674 container_of(ex, struct rbd_obj_request, ex);
2675 struct ceph_bvec_iter *it = arg;
2677 obj_req->bvec_pos = *it;
2678 ceph_bvec_iter_shorten(&obj_req->bvec_pos, bytes);
2679 ceph_bvec_iter_advance(it, bytes);
2682 static void count_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2684 struct rbd_obj_request *obj_req =
2685 container_of(ex, struct rbd_obj_request, ex);
2686 struct ceph_bvec_iter *it = arg;
2688 ceph_bvec_iter_advance_step(it, bytes, ({
2689 obj_req->bvec_count++;
2693 static void copy_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2695 struct rbd_obj_request *obj_req =
2696 container_of(ex, struct rbd_obj_request, ex);
2697 struct ceph_bvec_iter *it = arg;
2699 ceph_bvec_iter_advance_step(it, bytes, ({
2700 obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2701 obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2705 static int __rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2706 struct ceph_file_extent *img_extents,
2707 u32 num_img_extents,
2708 struct ceph_bvec_iter *bvec_pos)
2710 struct rbd_img_fill_ctx fctx = {
2711 .pos_type = OBJ_REQUEST_BVECS,
2712 .pos = (union rbd_img_fill_iter *)bvec_pos,
2713 .set_pos_fn = set_bvec_pos,
2714 .count_fn = count_bvecs,
2715 .copy_fn = copy_bvecs,
2718 return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2722 static int rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2723 struct ceph_file_extent *img_extents,
2724 u32 num_img_extents,
2725 struct bio_vec *bvecs)
2727 struct ceph_bvec_iter it = {
2729 .iter = { .bi_size = ceph_file_extents_bytes(img_extents,
2733 return __rbd_img_fill_from_bvecs(img_req, img_extents, num_img_extents,
2737 static void rbd_img_handle_request_work(struct work_struct *work)
2739 struct rbd_img_request *img_req =
2740 container_of(work, struct rbd_img_request, work);
2742 rbd_img_handle_request(img_req, img_req->work_result);
2745 static void rbd_img_schedule(struct rbd_img_request *img_req, int result)
2747 INIT_WORK(&img_req->work, rbd_img_handle_request_work);
2748 img_req->work_result = result;
2749 queue_work(rbd_wq, &img_req->work);
2752 static bool rbd_obj_may_exist(struct rbd_obj_request *obj_req)
2754 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2756 if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno)) {
2757 obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2761 dout("%s %p objno %llu assuming dne\n", __func__, obj_req,
2762 obj_req->ex.oe_objno);
2766 static int rbd_obj_read_object(struct rbd_obj_request *obj_req)
2768 struct ceph_osd_request *osd_req;
2771 osd_req = __rbd_obj_add_osd_request(obj_req, NULL, 1);
2772 if (IS_ERR(osd_req))
2773 return PTR_ERR(osd_req);
2775 osd_req_op_extent_init(osd_req, 0, CEPH_OSD_OP_READ,
2776 obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2777 rbd_osd_setup_data(osd_req, 0);
2778 rbd_osd_format_read(osd_req);
2780 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2784 rbd_osd_submit(osd_req);
2788 static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
2790 struct rbd_img_request *img_req = obj_req->img_request;
2791 struct rbd_device *parent = img_req->rbd_dev->parent;
2792 struct rbd_img_request *child_img_req;
2795 child_img_req = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
2799 rbd_img_request_init(child_img_req, parent, OBJ_OP_READ);
2800 __set_bit(IMG_REQ_CHILD, &child_img_req->flags);
2801 child_img_req->obj_request = obj_req;
2803 down_read(&parent->header_rwsem);
2804 rbd_img_capture_header(child_img_req);
2805 up_read(&parent->header_rwsem);
2807 dout("%s child_img_req %p for obj_req %p\n", __func__, child_img_req,
2810 if (!rbd_img_is_write(img_req)) {
2811 switch (img_req->data_type) {
2812 case OBJ_REQUEST_BIO:
2813 ret = __rbd_img_fill_from_bio(child_img_req,
2814 obj_req->img_extents,
2815 obj_req->num_img_extents,
2818 case OBJ_REQUEST_BVECS:
2819 case OBJ_REQUEST_OWN_BVECS:
2820 ret = __rbd_img_fill_from_bvecs(child_img_req,
2821 obj_req->img_extents,
2822 obj_req->num_img_extents,
2823 &obj_req->bvec_pos);
2829 ret = rbd_img_fill_from_bvecs(child_img_req,
2830 obj_req->img_extents,
2831 obj_req->num_img_extents,
2832 obj_req->copyup_bvecs);
2835 rbd_img_request_destroy(child_img_req);
2839 /* avoid parent chain recursion */
2840 rbd_img_schedule(child_img_req, 0);
2844 static bool rbd_obj_advance_read(struct rbd_obj_request *obj_req, int *result)
2846 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2850 switch (obj_req->read_state) {
2851 case RBD_OBJ_READ_START:
2852 rbd_assert(!*result);
2854 if (!rbd_obj_may_exist(obj_req)) {
2856 obj_req->read_state = RBD_OBJ_READ_OBJECT;
2860 ret = rbd_obj_read_object(obj_req);
2865 obj_req->read_state = RBD_OBJ_READ_OBJECT;
2867 case RBD_OBJ_READ_OBJECT:
2868 if (*result == -ENOENT && rbd_dev->parent_overlap) {
2869 /* reverse map this object extent onto the parent */
2870 ret = rbd_obj_calc_img_extents(obj_req, false);
2875 if (obj_req->num_img_extents) {
2876 ret = rbd_obj_read_from_parent(obj_req);
2881 obj_req->read_state = RBD_OBJ_READ_PARENT;
2887 * -ENOENT means a hole in the image -- zero-fill the entire
2888 * length of the request. A short read also implies zero-fill
2889 * to the end of the request.
2891 if (*result == -ENOENT) {
2892 rbd_obj_zero_range(obj_req, 0, obj_req->ex.oe_len);
2894 } else if (*result >= 0) {
2895 if (*result < obj_req->ex.oe_len)
2896 rbd_obj_zero_range(obj_req, *result,
2897 obj_req->ex.oe_len - *result);
2899 rbd_assert(*result == obj_req->ex.oe_len);
2903 case RBD_OBJ_READ_PARENT:
2905 * The parent image is read only up to the overlap -- zero-fill
2906 * from the overlap to the end of the request.
2909 u32 obj_overlap = rbd_obj_img_extents_bytes(obj_req);
2911 if (obj_overlap < obj_req->ex.oe_len)
2912 rbd_obj_zero_range(obj_req, obj_overlap,
2913 obj_req->ex.oe_len - obj_overlap);
2921 static bool rbd_obj_write_is_noop(struct rbd_obj_request *obj_req)
2923 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2925 if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno))
2926 obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2928 if (!(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST) &&
2929 (obj_req->flags & RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT)) {
2930 dout("%s %p noop for nonexistent\n", __func__, obj_req);
2939 * 0 - object map update sent
2940 * 1 - object map update isn't needed
2943 static int rbd_obj_write_pre_object_map(struct rbd_obj_request *obj_req)
2945 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2948 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
2951 if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
2952 new_state = OBJECT_PENDING;
2954 new_state = OBJECT_EXISTS;
2956 return rbd_object_map_update(obj_req, CEPH_NOSNAP, new_state, NULL);
2959 static int rbd_obj_write_object(struct rbd_obj_request *obj_req)
2961 struct ceph_osd_request *osd_req;
2962 int num_ops = count_write_ops(obj_req);
2966 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
2967 num_ops++; /* stat */
2969 osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
2970 if (IS_ERR(osd_req))
2971 return PTR_ERR(osd_req);
2973 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
2974 ret = rbd_osd_setup_stat(osd_req, which++);
2979 rbd_osd_setup_write_ops(osd_req, which);
2980 rbd_osd_format_write(osd_req);
2982 ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2986 rbd_osd_submit(osd_req);
2991 * copyup_bvecs pages are never highmem pages
2993 static bool is_zero_bvecs(struct bio_vec *bvecs, u32 bytes)
2995 struct ceph_bvec_iter it = {
2997 .iter = { .bi_size = bytes },
3000 ceph_bvec_iter_advance_step(&it, bytes, ({
3001 if (memchr_inv(bvec_virt(&bv), 0, bv.bv_len))
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);
3084 struct page *page = alloc_page(GFP_NOIO);
3089 bvec_set_page(&obj_req->copyup_bvecs[i], page, len, 0);
3093 rbd_assert(!obj_overlap);
3098 * The target object doesn't exist. Read the data for the entire
3099 * target object up to the overlap point (if any) from the parent,
3100 * so we can use it for a copyup.
3102 static int rbd_obj_copyup_read_parent(struct rbd_obj_request *obj_req)
3104 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3107 rbd_assert(obj_req->num_img_extents);
3108 prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
3109 rbd_dev->parent_overlap);
3110 if (!obj_req->num_img_extents) {
3112 * The overlap has become 0 (most likely because the
3113 * image has been flattened). Re-submit the original write
3114 * request -- pass MODS_ONLY since the copyup isn't needed
3117 return rbd_obj_copyup_current_snapc(obj_req, MODS_ONLY);
3120 ret = setup_copyup_bvecs(obj_req, rbd_obj_img_extents_bytes(obj_req));
3124 return rbd_obj_read_from_parent(obj_req);
3127 static void rbd_obj_copyup_object_maps(struct rbd_obj_request *obj_req)
3129 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3130 struct ceph_snap_context *snapc = obj_req->img_request->snapc;
3135 rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3137 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3140 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3143 for (i = 0; i < snapc->num_snaps; i++) {
3144 if ((rbd_dev->header.features & RBD_FEATURE_FAST_DIFF) &&
3145 i + 1 < snapc->num_snaps)
3146 new_state = OBJECT_EXISTS_CLEAN;
3148 new_state = OBJECT_EXISTS;
3150 ret = rbd_object_map_update(obj_req, snapc->snaps[i],
3153 obj_req->pending.result = ret;
3158 obj_req->pending.num_pending++;
3162 static void rbd_obj_copyup_write_object(struct rbd_obj_request *obj_req)
3164 u32 bytes = rbd_obj_img_extents_bytes(obj_req);
3167 rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3170 * Only send non-zero copyup data to save some I/O and network
3171 * bandwidth -- zero copyup data is equivalent to the object not
3174 if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3177 if (obj_req->img_request->snapc->num_snaps && bytes > 0) {
3179 * Send a copyup request with an empty snapshot context to
3180 * deep-copyup the object through all existing snapshots.
3181 * A second request with the current snapshot context will be
3182 * sent for the actual modification.
3184 ret = rbd_obj_copyup_empty_snapc(obj_req, bytes);
3186 obj_req->pending.result = ret;
3190 obj_req->pending.num_pending++;
3194 ret = rbd_obj_copyup_current_snapc(obj_req, bytes);
3196 obj_req->pending.result = ret;
3200 obj_req->pending.num_pending++;
3203 static bool rbd_obj_advance_copyup(struct rbd_obj_request *obj_req, int *result)
3205 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3209 switch (obj_req->copyup_state) {
3210 case RBD_OBJ_COPYUP_START:
3211 rbd_assert(!*result);
3213 ret = rbd_obj_copyup_read_parent(obj_req);
3218 if (obj_req->num_img_extents)
3219 obj_req->copyup_state = RBD_OBJ_COPYUP_READ_PARENT;
3221 obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3223 case RBD_OBJ_COPYUP_READ_PARENT:
3227 if (is_zero_bvecs(obj_req->copyup_bvecs,
3228 rbd_obj_img_extents_bytes(obj_req))) {
3229 dout("%s %p detected zeros\n", __func__, obj_req);
3230 obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ZEROS;
3233 rbd_obj_copyup_object_maps(obj_req);
3234 if (!obj_req->pending.num_pending) {
3235 *result = obj_req->pending.result;
3236 obj_req->copyup_state = RBD_OBJ_COPYUP_OBJECT_MAPS;
3239 obj_req->copyup_state = __RBD_OBJ_COPYUP_OBJECT_MAPS;
3241 case __RBD_OBJ_COPYUP_OBJECT_MAPS:
3242 if (!pending_result_dec(&obj_req->pending, result))
3245 case RBD_OBJ_COPYUP_OBJECT_MAPS:
3247 rbd_warn(rbd_dev, "snap object map update failed: %d",
3252 rbd_obj_copyup_write_object(obj_req);
3253 if (!obj_req->pending.num_pending) {
3254 *result = obj_req->pending.result;
3255 obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3258 obj_req->copyup_state = __RBD_OBJ_COPYUP_WRITE_OBJECT;
3260 case __RBD_OBJ_COPYUP_WRITE_OBJECT:
3261 if (!pending_result_dec(&obj_req->pending, result))
3264 case RBD_OBJ_COPYUP_WRITE_OBJECT:
3273 * 0 - object map update sent
3274 * 1 - object map update isn't needed
3277 static int rbd_obj_write_post_object_map(struct rbd_obj_request *obj_req)
3279 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3280 u8 current_state = OBJECT_PENDING;
3282 if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3285 if (!(obj_req->flags & RBD_OBJ_FLAG_DELETION))
3288 return rbd_object_map_update(obj_req, CEPH_NOSNAP, OBJECT_NONEXISTENT,
3292 static bool rbd_obj_advance_write(struct rbd_obj_request *obj_req, int *result)
3294 struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3298 switch (obj_req->write_state) {
3299 case RBD_OBJ_WRITE_START:
3300 rbd_assert(!*result);
3302 rbd_obj_set_copyup_enabled(obj_req);
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_device *rbd_dev = img_req->rbd_dev;
3490 struct rbd_obj_request *obj_req;
3492 rbd_assert(!img_req->pending.result && !img_req->pending.num_pending);
3493 rbd_assert(!need_exclusive_lock(img_req) ||
3494 __rbd_is_lock_owner(rbd_dev));
3496 if (rbd_img_is_write(img_req)) {
3497 rbd_assert(!img_req->snapc);
3498 down_read(&rbd_dev->header_rwsem);
3499 img_req->snapc = ceph_get_snap_context(rbd_dev->header.snapc);
3500 up_read(&rbd_dev->header_rwsem);
3503 for_each_obj_request(img_req, obj_req) {
3506 if (__rbd_obj_handle_request(obj_req, &result)) {
3508 img_req->pending.result = result;
3512 img_req->pending.num_pending++;
3517 static bool rbd_img_advance(struct rbd_img_request *img_req, int *result)
3522 switch (img_req->state) {
3524 rbd_assert(!*result);
3526 ret = rbd_img_exclusive_lock(img_req);
3531 img_req->state = RBD_IMG_EXCLUSIVE_LOCK;
3535 case RBD_IMG_EXCLUSIVE_LOCK:
3539 rbd_img_object_requests(img_req);
3540 if (!img_req->pending.num_pending) {
3541 *result = img_req->pending.result;
3542 img_req->state = RBD_IMG_OBJECT_REQUESTS;
3545 img_req->state = __RBD_IMG_OBJECT_REQUESTS;
3547 case __RBD_IMG_OBJECT_REQUESTS:
3548 if (!pending_result_dec(&img_req->pending, result))
3551 case RBD_IMG_OBJECT_REQUESTS:
3559 * Return true if @img_req is completed.
3561 static bool __rbd_img_handle_request(struct rbd_img_request *img_req,
3564 struct rbd_device *rbd_dev = img_req->rbd_dev;
3567 if (need_exclusive_lock(img_req)) {
3568 down_read(&rbd_dev->lock_rwsem);
3569 mutex_lock(&img_req->state_mutex);
3570 done = rbd_img_advance(img_req, result);
3572 rbd_lock_del_request(img_req);
3573 mutex_unlock(&img_req->state_mutex);
3574 up_read(&rbd_dev->lock_rwsem);
3576 mutex_lock(&img_req->state_mutex);
3577 done = rbd_img_advance(img_req, result);
3578 mutex_unlock(&img_req->state_mutex);
3581 if (done && *result) {
3582 rbd_assert(*result < 0);
3583 rbd_warn(rbd_dev, "%s%s result %d",
3584 test_bit(IMG_REQ_CHILD, &img_req->flags) ? "child " : "",
3585 obj_op_name(img_req->op_type), *result);
3590 static void rbd_img_handle_request(struct rbd_img_request *img_req, int result)
3593 if (!__rbd_img_handle_request(img_req, &result))
3596 if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
3597 struct rbd_obj_request *obj_req = img_req->obj_request;
3599 rbd_img_request_destroy(img_req);
3600 if (__rbd_obj_handle_request(obj_req, &result)) {
3601 img_req = obj_req->img_request;
3605 struct request *rq = blk_mq_rq_from_pdu(img_req);
3607 rbd_img_request_destroy(img_req);
3608 blk_mq_end_request(rq, errno_to_blk_status(result));
3612 static const struct rbd_client_id rbd_empty_cid;
3614 static bool rbd_cid_equal(const struct rbd_client_id *lhs,
3615 const struct rbd_client_id *rhs)
3617 return lhs->gid == rhs->gid && lhs->handle == rhs->handle;
3620 static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev)
3622 struct rbd_client_id cid;
3624 mutex_lock(&rbd_dev->watch_mutex);
3625 cid.gid = ceph_client_gid(rbd_dev->rbd_client->client);
3626 cid.handle = rbd_dev->watch_cookie;
3627 mutex_unlock(&rbd_dev->watch_mutex);
3632 * lock_rwsem must be held for write
3634 static void rbd_set_owner_cid(struct rbd_device *rbd_dev,
3635 const struct rbd_client_id *cid)
3637 dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev,
3638 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle,
3639 cid->gid, cid->handle);
3640 rbd_dev->owner_cid = *cid; /* struct */
3643 static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf)
3645 mutex_lock(&rbd_dev->watch_mutex);
3646 sprintf(buf, "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie);
3647 mutex_unlock(&rbd_dev->watch_mutex);
3650 static void __rbd_lock(struct rbd_device *rbd_dev, const char *cookie)
3652 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3654 rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED;
3655 strcpy(rbd_dev->lock_cookie, cookie);
3656 rbd_set_owner_cid(rbd_dev, &cid);
3657 queue_work(rbd_dev->task_wq, &rbd_dev->acquired_lock_work);
3661 * lock_rwsem must be held for write
3663 static int rbd_lock(struct rbd_device *rbd_dev)
3665 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3669 WARN_ON(__rbd_is_lock_owner(rbd_dev) ||
3670 rbd_dev->lock_cookie[0] != '\0');
3672 format_lock_cookie(rbd_dev, cookie);
3673 ret = ceph_cls_lock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3674 RBD_LOCK_NAME, CEPH_CLS_LOCK_EXCLUSIVE, cookie,
3675 RBD_LOCK_TAG, "", 0);
3676 if (ret && ret != -EEXIST)
3679 __rbd_lock(rbd_dev, cookie);
3684 * lock_rwsem must be held for write
3686 static void rbd_unlock(struct rbd_device *rbd_dev)
3688 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3691 WARN_ON(!__rbd_is_lock_owner(rbd_dev) ||
3692 rbd_dev->lock_cookie[0] == '\0');
3694 ret = ceph_cls_unlock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3695 RBD_LOCK_NAME, rbd_dev->lock_cookie);
3696 if (ret && ret != -ENOENT)
3697 rbd_warn(rbd_dev, "failed to unlock header: %d", ret);
3699 /* treat errors as the image is unlocked */
3700 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
3701 rbd_dev->lock_cookie[0] = '\0';
3702 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3703 queue_work(rbd_dev->task_wq, &rbd_dev->released_lock_work);
3706 static int __rbd_notify_op_lock(struct rbd_device *rbd_dev,
3707 enum rbd_notify_op notify_op,
3708 struct page ***preply_pages,
3711 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3712 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3713 char buf[4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN];
3714 int buf_size = sizeof(buf);
3717 dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op);
3719 /* encode *LockPayload NotifyMessage (op + ClientId) */
3720 ceph_start_encoding(&p, 2, 1, buf_size - CEPH_ENCODING_START_BLK_LEN);
3721 ceph_encode_32(&p, notify_op);
3722 ceph_encode_64(&p, cid.gid);
3723 ceph_encode_64(&p, cid.handle);
3725 return ceph_osdc_notify(osdc, &rbd_dev->header_oid,
3726 &rbd_dev->header_oloc, buf, buf_size,
3727 RBD_NOTIFY_TIMEOUT, preply_pages, preply_len);
3730 static void rbd_notify_op_lock(struct rbd_device *rbd_dev,
3731 enum rbd_notify_op notify_op)
3733 __rbd_notify_op_lock(rbd_dev, notify_op, NULL, NULL);
3736 static void rbd_notify_acquired_lock(struct work_struct *work)
3738 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3739 acquired_lock_work);
3741 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_ACQUIRED_LOCK);
3744 static void rbd_notify_released_lock(struct work_struct *work)
3746 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3747 released_lock_work);
3749 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_RELEASED_LOCK);
3752 static int rbd_request_lock(struct rbd_device *rbd_dev)
3754 struct page **reply_pages;
3756 bool lock_owner_responded = false;
3759 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3761 ret = __rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_REQUEST_LOCK,
3762 &reply_pages, &reply_len);
3763 if (ret && ret != -ETIMEDOUT) {
3764 rbd_warn(rbd_dev, "failed to request lock: %d", ret);
3768 if (reply_len > 0 && reply_len <= PAGE_SIZE) {
3769 void *p = page_address(reply_pages[0]);
3770 void *const end = p + reply_len;
3773 ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */
3778 ceph_decode_need(&p, end, 8 + 8, e_inval);
3779 p += 8 + 8; /* skip gid and cookie */
3781 ceph_decode_32_safe(&p, end, len, e_inval);
3785 if (lock_owner_responded) {
3787 "duplicate lock owners detected");
3792 lock_owner_responded = true;
3793 ret = ceph_start_decoding(&p, end, 1, "ResponseMessage",
3797 "failed to decode ResponseMessage: %d",
3802 ret = ceph_decode_32(&p);
3806 if (!lock_owner_responded) {
3807 rbd_warn(rbd_dev, "no lock owners detected");
3812 ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3821 * Either image request state machine(s) or rbd_add_acquire_lock()
3824 static void wake_lock_waiters(struct rbd_device *rbd_dev, int result)
3826 struct rbd_img_request *img_req;
3828 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
3829 lockdep_assert_held_write(&rbd_dev->lock_rwsem);
3831 cancel_delayed_work(&rbd_dev->lock_dwork);
3832 if (!completion_done(&rbd_dev->acquire_wait)) {
3833 rbd_assert(list_empty(&rbd_dev->acquiring_list) &&
3834 list_empty(&rbd_dev->running_list));
3835 rbd_dev->acquire_err = result;
3836 complete_all(&rbd_dev->acquire_wait);
3840 list_for_each_entry(img_req, &rbd_dev->acquiring_list, lock_item) {
3841 mutex_lock(&img_req->state_mutex);
3842 rbd_assert(img_req->state == RBD_IMG_EXCLUSIVE_LOCK);
3843 rbd_img_schedule(img_req, result);
3844 mutex_unlock(&img_req->state_mutex);
3847 list_splice_tail_init(&rbd_dev->acquiring_list, &rbd_dev->running_list);
3850 static bool locker_equal(const struct ceph_locker *lhs,
3851 const struct ceph_locker *rhs)
3853 return lhs->id.name.type == rhs->id.name.type &&
3854 lhs->id.name.num == rhs->id.name.num &&
3855 !strcmp(lhs->id.cookie, rhs->id.cookie) &&
3856 ceph_addr_equal_no_type(&lhs->info.addr, &rhs->info.addr);
3859 static void free_locker(struct ceph_locker *locker)
3862 ceph_free_lockers(locker, 1);
3865 static struct ceph_locker *get_lock_owner_info(struct rbd_device *rbd_dev)
3867 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3868 struct ceph_locker *lockers;
3875 ret = ceph_cls_lock_info(osdc, &rbd_dev->header_oid,
3876 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3877 &lock_type, &lock_tag, &lockers, &num_lockers);
3879 rbd_warn(rbd_dev, "failed to get header lockers: %d", ret);
3880 return ERR_PTR(ret);
3883 if (num_lockers == 0) {
3884 dout("%s rbd_dev %p no lockers detected\n", __func__, rbd_dev);
3889 if (strcmp(lock_tag, RBD_LOCK_TAG)) {
3890 rbd_warn(rbd_dev, "locked by external mechanism, tag %s",
3895 if (lock_type != CEPH_CLS_LOCK_EXCLUSIVE) {
3896 rbd_warn(rbd_dev, "incompatible lock type detected");
3900 WARN_ON(num_lockers != 1);
3901 ret = sscanf(lockers[0].id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu",
3904 rbd_warn(rbd_dev, "locked by external mechanism, cookie %s",
3905 lockers[0].id.cookie);
3908 if (ceph_addr_is_blank(&lockers[0].info.addr)) {
3909 rbd_warn(rbd_dev, "locker has a blank address");
3913 dout("%s rbd_dev %p got locker %s%llu@%pISpc/%u handle %llu\n",
3914 __func__, rbd_dev, ENTITY_NAME(lockers[0].id.name),
3915 &lockers[0].info.addr.in_addr,
3916 le32_to_cpu(lockers[0].info.addr.nonce), handle);
3924 ceph_free_lockers(lockers, num_lockers);
3925 return ERR_PTR(-EBUSY);
3928 static int find_watcher(struct rbd_device *rbd_dev,
3929 const struct ceph_locker *locker)
3931 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3932 struct ceph_watch_item *watchers;
3938 ret = ceph_osdc_list_watchers(osdc, &rbd_dev->header_oid,
3939 &rbd_dev->header_oloc, &watchers,
3942 rbd_warn(rbd_dev, "failed to get watchers: %d", ret);
3946 sscanf(locker->id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu", &cookie);
3947 for (i = 0; i < num_watchers; i++) {
3949 * Ignore addr->type while comparing. This mimics
3950 * entity_addr_t::get_legacy_str() + strcmp().
3952 if (ceph_addr_equal_no_type(&watchers[i].addr,
3953 &locker->info.addr) &&
3954 watchers[i].cookie == cookie) {
3955 struct rbd_client_id cid = {
3956 .gid = le64_to_cpu(watchers[i].name.num),
3960 dout("%s rbd_dev %p found cid %llu-%llu\n", __func__,
3961 rbd_dev, cid.gid, cid.handle);
3962 rbd_set_owner_cid(rbd_dev, &cid);
3968 dout("%s rbd_dev %p no watchers\n", __func__, rbd_dev);
3976 * lock_rwsem must be held for write
3978 static int rbd_try_lock(struct rbd_device *rbd_dev)
3980 struct ceph_client *client = rbd_dev->rbd_client->client;
3981 struct ceph_locker *locker, *refreshed_locker;
3985 locker = refreshed_locker = NULL;
3987 ret = rbd_lock(rbd_dev);
3990 if (ret != -EBUSY) {
3991 rbd_warn(rbd_dev, "failed to lock header: %d", ret);
3995 /* determine if the current lock holder is still alive */
3996 locker = get_lock_owner_info(rbd_dev);
3997 if (IS_ERR(locker)) {
3998 ret = PTR_ERR(locker);
4005 ret = find_watcher(rbd_dev, locker);
4007 goto out; /* request lock or error */
4009 refreshed_locker = get_lock_owner_info(rbd_dev);
4010 if (IS_ERR(refreshed_locker)) {
4011 ret = PTR_ERR(refreshed_locker);
4012 refreshed_locker = NULL;
4015 if (!refreshed_locker ||
4016 !locker_equal(locker, refreshed_locker))
4019 rbd_warn(rbd_dev, "breaking header lock owned by %s%llu",
4020 ENTITY_NAME(locker->id.name));
4022 ret = ceph_monc_blocklist_add(&client->monc,
4023 &locker->info.addr);
4025 rbd_warn(rbd_dev, "failed to blocklist %s%llu: %d",
4026 ENTITY_NAME(locker->id.name), ret);
4030 ret = ceph_cls_break_lock(&client->osdc, &rbd_dev->header_oid,
4031 &rbd_dev->header_oloc, RBD_LOCK_NAME,
4032 locker->id.cookie, &locker->id.name);
4033 if (ret && ret != -ENOENT) {
4034 rbd_warn(rbd_dev, "failed to break header lock: %d",
4040 free_locker(refreshed_locker);
4041 free_locker(locker);
4045 free_locker(refreshed_locker);
4046 free_locker(locker);
4050 static int rbd_post_acquire_action(struct rbd_device *rbd_dev)
4054 ret = rbd_dev_refresh(rbd_dev);
4058 if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) {
4059 ret = rbd_object_map_open(rbd_dev);
4070 * 1 - caller should call rbd_request_lock()
4073 static int rbd_try_acquire_lock(struct rbd_device *rbd_dev)
4077 down_read(&rbd_dev->lock_rwsem);
4078 dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
4079 rbd_dev->lock_state);
4080 if (__rbd_is_lock_owner(rbd_dev)) {
4081 up_read(&rbd_dev->lock_rwsem);
4085 up_read(&rbd_dev->lock_rwsem);
4086 down_write(&rbd_dev->lock_rwsem);
4087 dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
4088 rbd_dev->lock_state);
4089 if (__rbd_is_lock_owner(rbd_dev)) {
4090 up_write(&rbd_dev->lock_rwsem);
4094 ret = rbd_try_lock(rbd_dev);
4096 rbd_warn(rbd_dev, "failed to acquire lock: %d", ret);
4100 up_write(&rbd_dev->lock_rwsem);
4104 rbd_assert(rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED);
4105 rbd_assert(list_empty(&rbd_dev->running_list));
4107 ret = rbd_post_acquire_action(rbd_dev);
4109 rbd_warn(rbd_dev, "post-acquire action failed: %d", ret);
4111 * Can't stay in RBD_LOCK_STATE_LOCKED because
4112 * rbd_lock_add_request() would let the request through,
4113 * assuming that e.g. object map is locked and loaded.
4115 rbd_unlock(rbd_dev);
4119 wake_lock_waiters(rbd_dev, ret);
4120 up_write(&rbd_dev->lock_rwsem);
4124 static void rbd_acquire_lock(struct work_struct *work)
4126 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4127 struct rbd_device, lock_dwork);
4130 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4132 ret = rbd_try_acquire_lock(rbd_dev);
4134 dout("%s rbd_dev %p ret %d - done\n", __func__, rbd_dev, ret);
4138 ret = rbd_request_lock(rbd_dev);
4139 if (ret == -ETIMEDOUT) {
4140 goto again; /* treat this as a dead client */
4141 } else if (ret == -EROFS) {
4142 rbd_warn(rbd_dev, "peer will not release lock");
4143 down_write(&rbd_dev->lock_rwsem);
4144 wake_lock_waiters(rbd_dev, ret);
4145 up_write(&rbd_dev->lock_rwsem);
4146 } else if (ret < 0) {
4147 rbd_warn(rbd_dev, "error requesting lock: %d", ret);
4148 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4152 * lock owner acked, but resend if we don't see them
4155 dout("%s rbd_dev %p requeuing lock_dwork\n", __func__,
4157 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4158 msecs_to_jiffies(2 * RBD_NOTIFY_TIMEOUT * MSEC_PER_SEC));
4162 static bool rbd_quiesce_lock(struct rbd_device *rbd_dev)
4164 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4165 lockdep_assert_held_write(&rbd_dev->lock_rwsem);
4167 if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED)
4171 * Ensure that all in-flight IO is flushed.
4173 rbd_dev->lock_state = RBD_LOCK_STATE_RELEASING;
4174 rbd_assert(!completion_done(&rbd_dev->releasing_wait));
4175 if (list_empty(&rbd_dev->running_list))
4178 up_write(&rbd_dev->lock_rwsem);
4179 wait_for_completion(&rbd_dev->releasing_wait);
4181 down_write(&rbd_dev->lock_rwsem);
4182 if (rbd_dev->lock_state != RBD_LOCK_STATE_RELEASING)
4185 rbd_assert(list_empty(&rbd_dev->running_list));
4189 static void rbd_pre_release_action(struct rbd_device *rbd_dev)
4191 if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)
4192 rbd_object_map_close(rbd_dev);
4195 static void __rbd_release_lock(struct rbd_device *rbd_dev)
4197 rbd_assert(list_empty(&rbd_dev->running_list));
4199 rbd_pre_release_action(rbd_dev);
4200 rbd_unlock(rbd_dev);
4204 * lock_rwsem must be held for write
4206 static void rbd_release_lock(struct rbd_device *rbd_dev)
4208 if (!rbd_quiesce_lock(rbd_dev))
4211 __rbd_release_lock(rbd_dev);
4214 * Give others a chance to grab the lock - we would re-acquire
4215 * almost immediately if we got new IO while draining the running
4216 * list otherwise. We need to ack our own notifications, so this
4217 * lock_dwork will be requeued from rbd_handle_released_lock() by
4218 * way of maybe_kick_acquire().
4220 cancel_delayed_work(&rbd_dev->lock_dwork);
4223 static void rbd_release_lock_work(struct work_struct *work)
4225 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
4228 down_write(&rbd_dev->lock_rwsem);
4229 rbd_release_lock(rbd_dev);
4230 up_write(&rbd_dev->lock_rwsem);
4233 static void maybe_kick_acquire(struct rbd_device *rbd_dev)
4237 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4238 if (__rbd_is_lock_owner(rbd_dev))
4241 spin_lock(&rbd_dev->lock_lists_lock);
4242 have_requests = !list_empty(&rbd_dev->acquiring_list);
4243 spin_unlock(&rbd_dev->lock_lists_lock);
4244 if (have_requests || delayed_work_pending(&rbd_dev->lock_dwork)) {
4245 dout("%s rbd_dev %p kicking lock_dwork\n", __func__, rbd_dev);
4246 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4250 static void rbd_handle_acquired_lock(struct rbd_device *rbd_dev, u8 struct_v,
4253 struct rbd_client_id cid = { 0 };
4255 if (struct_v >= 2) {
4256 cid.gid = ceph_decode_64(p);
4257 cid.handle = ceph_decode_64(p);
4260 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4262 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4263 down_write(&rbd_dev->lock_rwsem);
4264 if (rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4265 dout("%s rbd_dev %p cid %llu-%llu == owner_cid\n",
4266 __func__, rbd_dev, cid.gid, cid.handle);
4268 rbd_set_owner_cid(rbd_dev, &cid);
4270 downgrade_write(&rbd_dev->lock_rwsem);
4272 down_read(&rbd_dev->lock_rwsem);
4275 maybe_kick_acquire(rbd_dev);
4276 up_read(&rbd_dev->lock_rwsem);
4279 static void rbd_handle_released_lock(struct rbd_device *rbd_dev, u8 struct_v,
4282 struct rbd_client_id cid = { 0 };
4284 if (struct_v >= 2) {
4285 cid.gid = ceph_decode_64(p);
4286 cid.handle = ceph_decode_64(p);
4289 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4291 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4292 down_write(&rbd_dev->lock_rwsem);
4293 if (!rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4294 dout("%s rbd_dev %p cid %llu-%llu != owner_cid %llu-%llu\n",
4295 __func__, rbd_dev, cid.gid, cid.handle,
4296 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle);
4298 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4300 downgrade_write(&rbd_dev->lock_rwsem);
4302 down_read(&rbd_dev->lock_rwsem);
4305 maybe_kick_acquire(rbd_dev);
4306 up_read(&rbd_dev->lock_rwsem);
4310 * Returns result for ResponseMessage to be encoded (<= 0), or 1 if no
4311 * ResponseMessage is needed.
4313 static int rbd_handle_request_lock(struct rbd_device *rbd_dev, u8 struct_v,
4316 struct rbd_client_id my_cid = rbd_get_cid(rbd_dev);
4317 struct rbd_client_id cid = { 0 };
4320 if (struct_v >= 2) {
4321 cid.gid = ceph_decode_64(p);
4322 cid.handle = ceph_decode_64(p);
4325 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4327 if (rbd_cid_equal(&cid, &my_cid))
4330 down_read(&rbd_dev->lock_rwsem);
4331 if (__rbd_is_lock_owner(rbd_dev)) {
4332 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED &&
4333 rbd_cid_equal(&rbd_dev->owner_cid, &rbd_empty_cid))
4337 * encode ResponseMessage(0) so the peer can detect
4342 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED) {
4343 if (!rbd_dev->opts->exclusive) {
4344 dout("%s rbd_dev %p queueing unlock_work\n",
4346 queue_work(rbd_dev->task_wq,
4347 &rbd_dev->unlock_work);
4349 /* refuse to release the lock */
4356 up_read(&rbd_dev->lock_rwsem);
4360 static void __rbd_acknowledge_notify(struct rbd_device *rbd_dev,
4361 u64 notify_id, u64 cookie, s32 *result)
4363 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4364 char buf[4 + CEPH_ENCODING_START_BLK_LEN];
4365 int buf_size = sizeof(buf);
4371 /* encode ResponseMessage */
4372 ceph_start_encoding(&p, 1, 1,
4373 buf_size - CEPH_ENCODING_START_BLK_LEN);
4374 ceph_encode_32(&p, *result);
4379 ret = ceph_osdc_notify_ack(osdc, &rbd_dev->header_oid,
4380 &rbd_dev->header_oloc, notify_id, cookie,
4383 rbd_warn(rbd_dev, "acknowledge_notify failed: %d", ret);
4386 static void rbd_acknowledge_notify(struct rbd_device *rbd_dev, u64 notify_id,
4389 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4390 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, NULL);
4393 static void rbd_acknowledge_notify_result(struct rbd_device *rbd_dev,
4394 u64 notify_id, u64 cookie, s32 result)
4396 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
4397 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, &result);
4400 static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie,
4401 u64 notifier_id, void *data, size_t data_len)
4403 struct rbd_device *rbd_dev = arg;
4405 void *const end = p + data_len;
4411 dout("%s rbd_dev %p cookie %llu notify_id %llu data_len %zu\n",
4412 __func__, rbd_dev, cookie, notify_id, data_len);
4414 ret = ceph_start_decoding(&p, end, 1, "NotifyMessage",
4417 rbd_warn(rbd_dev, "failed to decode NotifyMessage: %d",
4422 notify_op = ceph_decode_32(&p);
4424 /* legacy notification for header updates */
4425 notify_op = RBD_NOTIFY_OP_HEADER_UPDATE;
4429 dout("%s rbd_dev %p notify_op %u\n", __func__, rbd_dev, notify_op);
4430 switch (notify_op) {
4431 case RBD_NOTIFY_OP_ACQUIRED_LOCK:
4432 rbd_handle_acquired_lock(rbd_dev, struct_v, &p);
4433 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4435 case RBD_NOTIFY_OP_RELEASED_LOCK:
4436 rbd_handle_released_lock(rbd_dev, struct_v, &p);
4437 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4439 case RBD_NOTIFY_OP_REQUEST_LOCK:
4440 ret = rbd_handle_request_lock(rbd_dev, struct_v, &p);
4442 rbd_acknowledge_notify_result(rbd_dev, notify_id,
4445 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4447 case RBD_NOTIFY_OP_HEADER_UPDATE:
4448 ret = rbd_dev_refresh(rbd_dev);
4450 rbd_warn(rbd_dev, "refresh failed: %d", ret);
4452 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4455 if (rbd_is_lock_owner(rbd_dev))
4456 rbd_acknowledge_notify_result(rbd_dev, notify_id,
4457 cookie, -EOPNOTSUPP);
4459 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4464 static void __rbd_unregister_watch(struct rbd_device *rbd_dev);
4466 static void rbd_watch_errcb(void *arg, u64 cookie, int err)
4468 struct rbd_device *rbd_dev = arg;
4470 rbd_warn(rbd_dev, "encountered watch error: %d", err);
4472 down_write(&rbd_dev->lock_rwsem);
4473 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4474 up_write(&rbd_dev->lock_rwsem);
4476 mutex_lock(&rbd_dev->watch_mutex);
4477 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED) {
4478 __rbd_unregister_watch(rbd_dev);
4479 rbd_dev->watch_state = RBD_WATCH_STATE_ERROR;
4481 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->watch_dwork, 0);
4483 mutex_unlock(&rbd_dev->watch_mutex);
4487 * watch_mutex must be locked
4489 static int __rbd_register_watch(struct rbd_device *rbd_dev)
4491 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4492 struct ceph_osd_linger_request *handle;
4494 rbd_assert(!rbd_dev->watch_handle);
4495 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4497 handle = ceph_osdc_watch(osdc, &rbd_dev->header_oid,
4498 &rbd_dev->header_oloc, rbd_watch_cb,
4499 rbd_watch_errcb, rbd_dev);
4501 return PTR_ERR(handle);
4503 rbd_dev->watch_handle = handle;
4508 * watch_mutex must be locked
4510 static void __rbd_unregister_watch(struct rbd_device *rbd_dev)
4512 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4515 rbd_assert(rbd_dev->watch_handle);
4516 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4518 ret = ceph_osdc_unwatch(osdc, rbd_dev->watch_handle);
4520 rbd_warn(rbd_dev, "failed to unwatch: %d", ret);
4522 rbd_dev->watch_handle = NULL;
4525 static int rbd_register_watch(struct rbd_device *rbd_dev)
4529 mutex_lock(&rbd_dev->watch_mutex);
4530 rbd_assert(rbd_dev->watch_state == RBD_WATCH_STATE_UNREGISTERED);
4531 ret = __rbd_register_watch(rbd_dev);
4535 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4536 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4539 mutex_unlock(&rbd_dev->watch_mutex);
4543 static void cancel_tasks_sync(struct rbd_device *rbd_dev)
4545 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4547 cancel_work_sync(&rbd_dev->acquired_lock_work);
4548 cancel_work_sync(&rbd_dev->released_lock_work);
4549 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
4550 cancel_work_sync(&rbd_dev->unlock_work);
4554 * header_rwsem must not be held to avoid a deadlock with
4555 * rbd_dev_refresh() when flushing notifies.
4557 static void rbd_unregister_watch(struct rbd_device *rbd_dev)
4559 cancel_tasks_sync(rbd_dev);
4561 mutex_lock(&rbd_dev->watch_mutex);
4562 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED)
4563 __rbd_unregister_watch(rbd_dev);
4564 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
4565 mutex_unlock(&rbd_dev->watch_mutex);
4567 cancel_delayed_work_sync(&rbd_dev->watch_dwork);
4568 ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
4572 * lock_rwsem must be held for write
4574 static void rbd_reacquire_lock(struct rbd_device *rbd_dev)
4576 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4580 if (!rbd_quiesce_lock(rbd_dev))
4583 format_lock_cookie(rbd_dev, cookie);
4584 ret = ceph_cls_set_cookie(osdc, &rbd_dev->header_oid,
4585 &rbd_dev->header_oloc, RBD_LOCK_NAME,
4586 CEPH_CLS_LOCK_EXCLUSIVE, rbd_dev->lock_cookie,
4587 RBD_LOCK_TAG, cookie);
4589 if (ret != -EOPNOTSUPP)
4590 rbd_warn(rbd_dev, "failed to update lock cookie: %d",
4594 * Lock cookie cannot be updated on older OSDs, so do
4595 * a manual release and queue an acquire.
4597 __rbd_release_lock(rbd_dev);
4598 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4600 __rbd_lock(rbd_dev, cookie);
4601 wake_lock_waiters(rbd_dev, 0);
4605 static void rbd_reregister_watch(struct work_struct *work)
4607 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4608 struct rbd_device, watch_dwork);
4611 dout("%s rbd_dev %p\n", __func__, rbd_dev);
4613 mutex_lock(&rbd_dev->watch_mutex);
4614 if (rbd_dev->watch_state != RBD_WATCH_STATE_ERROR) {
4615 mutex_unlock(&rbd_dev->watch_mutex);
4619 ret = __rbd_register_watch(rbd_dev);
4621 rbd_warn(rbd_dev, "failed to reregister watch: %d", ret);
4622 if (ret != -EBLOCKLISTED && ret != -ENOENT) {
4623 queue_delayed_work(rbd_dev->task_wq,
4624 &rbd_dev->watch_dwork,
4626 mutex_unlock(&rbd_dev->watch_mutex);
4630 mutex_unlock(&rbd_dev->watch_mutex);
4631 down_write(&rbd_dev->lock_rwsem);
4632 wake_lock_waiters(rbd_dev, ret);
4633 up_write(&rbd_dev->lock_rwsem);
4637 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4638 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4639 mutex_unlock(&rbd_dev->watch_mutex);
4641 down_write(&rbd_dev->lock_rwsem);
4642 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED)
4643 rbd_reacquire_lock(rbd_dev);
4644 up_write(&rbd_dev->lock_rwsem);
4646 ret = rbd_dev_refresh(rbd_dev);
4648 rbd_warn(rbd_dev, "reregistration refresh failed: %d", ret);
4652 * Synchronous osd object method call. Returns the number of bytes
4653 * returned in the outbound buffer, or a negative error code.
4655 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
4656 struct ceph_object_id *oid,
4657 struct ceph_object_locator *oloc,
4658 const char *method_name,
4659 const void *outbound,
4660 size_t outbound_size,
4662 size_t inbound_size)
4664 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4665 struct page *req_page = NULL;
4666 struct page *reply_page;
4670 * Method calls are ultimately read operations. The result
4671 * should placed into the inbound buffer provided. They
4672 * also supply outbound data--parameters for the object
4673 * method. Currently if this is present it will be a
4677 if (outbound_size > PAGE_SIZE)
4680 req_page = alloc_page(GFP_KERNEL);
4684 memcpy(page_address(req_page), outbound, outbound_size);
4687 reply_page = alloc_page(GFP_KERNEL);
4690 __free_page(req_page);
4694 ret = ceph_osdc_call(osdc, oid, oloc, RBD_DRV_NAME, method_name,
4695 CEPH_OSD_FLAG_READ, req_page, outbound_size,
4696 &reply_page, &inbound_size);
4698 memcpy(inbound, page_address(reply_page), inbound_size);
4703 __free_page(req_page);
4704 __free_page(reply_page);
4708 static void rbd_queue_workfn(struct work_struct *work)
4710 struct rbd_img_request *img_request =
4711 container_of(work, struct rbd_img_request, work);
4712 struct rbd_device *rbd_dev = img_request->rbd_dev;
4713 enum obj_operation_type op_type = img_request->op_type;
4714 struct request *rq = blk_mq_rq_from_pdu(img_request);
4715 u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
4716 u64 length = blk_rq_bytes(rq);
4720 /* Ignore/skip any zero-length requests */
4722 dout("%s: zero-length request\n", __func__);
4724 goto err_img_request;
4727 blk_mq_start_request(rq);
4729 down_read(&rbd_dev->header_rwsem);
4730 mapping_size = rbd_dev->mapping.size;
4731 rbd_img_capture_header(img_request);
4732 up_read(&rbd_dev->header_rwsem);
4734 if (offset + length > mapping_size) {
4735 rbd_warn(rbd_dev, "beyond EOD (%llu~%llu > %llu)", offset,
4736 length, mapping_size);
4738 goto err_img_request;
4741 dout("%s rbd_dev %p img_req %p %s %llu~%llu\n", __func__, rbd_dev,
4742 img_request, obj_op_name(op_type), offset, length);
4744 if (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_ZEROOUT)
4745 result = rbd_img_fill_nodata(img_request, offset, length);
4747 result = rbd_img_fill_from_bio(img_request, offset, length,
4750 goto err_img_request;
4752 rbd_img_handle_request(img_request, 0);
4756 rbd_img_request_destroy(img_request);
4758 rbd_warn(rbd_dev, "%s %llx at %llx result %d",
4759 obj_op_name(op_type), length, offset, result);
4760 blk_mq_end_request(rq, errno_to_blk_status(result));
4763 static blk_status_t rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
4764 const struct blk_mq_queue_data *bd)
4766 struct rbd_device *rbd_dev = hctx->queue->queuedata;
4767 struct rbd_img_request *img_req = blk_mq_rq_to_pdu(bd->rq);
4768 enum obj_operation_type op_type;
4770 switch (req_op(bd->rq)) {
4771 case REQ_OP_DISCARD:
4772 op_type = OBJ_OP_DISCARD;
4774 case REQ_OP_WRITE_ZEROES:
4775 op_type = OBJ_OP_ZEROOUT;
4778 op_type = OBJ_OP_WRITE;
4781 op_type = OBJ_OP_READ;
4784 rbd_warn(rbd_dev, "unknown req_op %d", req_op(bd->rq));
4785 return BLK_STS_IOERR;
4788 rbd_img_request_init(img_req, rbd_dev, op_type);
4790 if (rbd_img_is_write(img_req)) {
4791 if (rbd_is_ro(rbd_dev)) {
4792 rbd_warn(rbd_dev, "%s on read-only mapping",
4793 obj_op_name(img_req->op_type));
4794 return BLK_STS_IOERR;
4796 rbd_assert(!rbd_is_snap(rbd_dev));
4799 INIT_WORK(&img_req->work, rbd_queue_workfn);
4800 queue_work(rbd_wq, &img_req->work);
4804 static void rbd_free_disk(struct rbd_device *rbd_dev)
4806 put_disk(rbd_dev->disk);
4807 blk_mq_free_tag_set(&rbd_dev->tag_set);
4808 rbd_dev->disk = NULL;
4811 static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
4812 struct ceph_object_id *oid,
4813 struct ceph_object_locator *oloc,
4814 void *buf, int buf_len)
4817 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4818 struct ceph_osd_request *req;
4819 struct page **pages;
4820 int num_pages = calc_pages_for(0, buf_len);
4823 req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_KERNEL);
4827 ceph_oid_copy(&req->r_base_oid, oid);
4828 ceph_oloc_copy(&req->r_base_oloc, oloc);
4829 req->r_flags = CEPH_OSD_FLAG_READ;
4831 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
4832 if (IS_ERR(pages)) {
4833 ret = PTR_ERR(pages);
4837 osd_req_op_extent_init(req, 0, CEPH_OSD_OP_READ, 0, buf_len, 0, 0);
4838 osd_req_op_extent_osd_data_pages(req, 0, pages, buf_len, 0, false,
4841 ret = ceph_osdc_alloc_messages(req, GFP_KERNEL);
4845 ceph_osdc_start_request(osdc, req);
4846 ret = ceph_osdc_wait_request(osdc, req);
4848 ceph_copy_from_page_vector(pages, buf, 0, ret);
4851 ceph_osdc_put_request(req);
4856 * Read the complete header for the given rbd device. On successful
4857 * return, the rbd_dev->header field will contain up-to-date
4858 * information about the image.
4860 static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev)
4862 struct rbd_image_header_ondisk *ondisk = NULL;
4869 * The complete header will include an array of its 64-bit
4870 * snapshot ids, followed by the names of those snapshots as
4871 * a contiguous block of NUL-terminated strings. Note that
4872 * the number of snapshots could change by the time we read
4873 * it in, in which case we re-read it.
4880 size = sizeof (*ondisk);
4881 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
4883 ondisk = kmalloc(size, GFP_KERNEL);
4887 ret = rbd_obj_read_sync(rbd_dev, &rbd_dev->header_oid,
4888 &rbd_dev->header_oloc, ondisk, size);
4891 if ((size_t)ret < size) {
4893 rbd_warn(rbd_dev, "short header read (want %zd got %d)",
4897 if (!rbd_dev_ondisk_valid(ondisk)) {
4899 rbd_warn(rbd_dev, "invalid header");
4903 names_size = le64_to_cpu(ondisk->snap_names_len);
4904 want_count = snap_count;
4905 snap_count = le32_to_cpu(ondisk->snap_count);
4906 } while (snap_count != want_count);
4908 ret = rbd_header_from_disk(rbd_dev, ondisk);
4915 static void rbd_dev_update_size(struct rbd_device *rbd_dev)
4920 * If EXISTS is not set, rbd_dev->disk may be NULL, so don't
4921 * try to update its size. If REMOVING is set, updating size
4922 * is just useless work since the device can't be opened.
4924 if (test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags) &&
4925 !test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) {
4926 size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
4927 dout("setting size to %llu sectors", (unsigned long long)size);
4928 set_capacity_and_notify(rbd_dev->disk, size);
4932 static const struct blk_mq_ops rbd_mq_ops = {
4933 .queue_rq = rbd_queue_rq,
4936 static int rbd_init_disk(struct rbd_device *rbd_dev)
4938 struct gendisk *disk;
4939 struct request_queue *q;
4940 unsigned int objset_bytes =
4941 rbd_dev->layout.object_size * rbd_dev->layout.stripe_count;
4944 memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
4945 rbd_dev->tag_set.ops = &rbd_mq_ops;
4946 rbd_dev->tag_set.queue_depth = rbd_dev->opts->queue_depth;
4947 rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
4948 rbd_dev->tag_set.flags = BLK_MQ_F_SHOULD_MERGE;
4949 rbd_dev->tag_set.nr_hw_queues = num_present_cpus();
4950 rbd_dev->tag_set.cmd_size = sizeof(struct rbd_img_request);
4952 err = blk_mq_alloc_tag_set(&rbd_dev->tag_set);
4956 disk = blk_mq_alloc_disk(&rbd_dev->tag_set, rbd_dev);
4958 err = PTR_ERR(disk);
4963 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
4965 disk->major = rbd_dev->major;
4966 disk->first_minor = rbd_dev->minor;
4968 disk->minors = (1 << RBD_SINGLE_MAJOR_PART_SHIFT);
4970 disk->minors = RBD_MINORS_PER_MAJOR;
4971 disk->fops = &rbd_bd_ops;
4972 disk->private_data = rbd_dev;
4974 blk_queue_flag_set(QUEUE_FLAG_NONROT, q);
4975 /* QUEUE_FLAG_ADD_RANDOM is off by default for blk-mq */
4977 blk_queue_max_hw_sectors(q, objset_bytes >> SECTOR_SHIFT);
4978 q->limits.max_sectors = queue_max_hw_sectors(q);
4979 blk_queue_max_segments(q, USHRT_MAX);
4980 blk_queue_max_segment_size(q, UINT_MAX);
4981 blk_queue_io_min(q, rbd_dev->opts->alloc_size);
4982 blk_queue_io_opt(q, rbd_dev->opts->alloc_size);
4984 if (rbd_dev->opts->trim) {
4985 q->limits.discard_granularity = rbd_dev->opts->alloc_size;
4986 blk_queue_max_discard_sectors(q, objset_bytes >> SECTOR_SHIFT);
4987 blk_queue_max_write_zeroes_sectors(q, objset_bytes >> SECTOR_SHIFT);
4990 if (!ceph_test_opt(rbd_dev->rbd_client->client, NOCRC))
4991 blk_queue_flag_set(QUEUE_FLAG_STABLE_WRITES, q);
4993 rbd_dev->disk = disk;
4997 blk_mq_free_tag_set(&rbd_dev->tag_set);
5005 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
5007 return container_of(dev, struct rbd_device, dev);
5010 static ssize_t rbd_size_show(struct device *dev,
5011 struct device_attribute *attr, char *buf)
5013 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5015 return sprintf(buf, "%llu\n",
5016 (unsigned long long)rbd_dev->mapping.size);
5019 static ssize_t rbd_features_show(struct device *dev,
5020 struct device_attribute *attr, char *buf)
5022 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5024 return sprintf(buf, "0x%016llx\n", rbd_dev->header.features);
5027 static ssize_t rbd_major_show(struct device *dev,
5028 struct device_attribute *attr, char *buf)
5030 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5033 return sprintf(buf, "%d\n", rbd_dev->major);
5035 return sprintf(buf, "(none)\n");
5038 static ssize_t rbd_minor_show(struct device *dev,
5039 struct device_attribute *attr, char *buf)
5041 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5043 return sprintf(buf, "%d\n", rbd_dev->minor);
5046 static ssize_t rbd_client_addr_show(struct device *dev,
5047 struct device_attribute *attr, char *buf)
5049 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5050 struct ceph_entity_addr *client_addr =
5051 ceph_client_addr(rbd_dev->rbd_client->client);
5053 return sprintf(buf, "%pISpc/%u\n", &client_addr->in_addr,
5054 le32_to_cpu(client_addr->nonce));
5057 static ssize_t rbd_client_id_show(struct device *dev,
5058 struct device_attribute *attr, char *buf)
5060 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5062 return sprintf(buf, "client%lld\n",
5063 ceph_client_gid(rbd_dev->rbd_client->client));
5066 static ssize_t rbd_cluster_fsid_show(struct device *dev,
5067 struct device_attribute *attr, char *buf)
5069 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5071 return sprintf(buf, "%pU\n", &rbd_dev->rbd_client->client->fsid);
5074 static ssize_t rbd_config_info_show(struct device *dev,
5075 struct device_attribute *attr, char *buf)
5077 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5079 if (!capable(CAP_SYS_ADMIN))
5082 return sprintf(buf, "%s\n", rbd_dev->config_info);
5085 static ssize_t rbd_pool_show(struct device *dev,
5086 struct device_attribute *attr, char *buf)
5088 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5090 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
5093 static ssize_t rbd_pool_id_show(struct device *dev,
5094 struct device_attribute *attr, char *buf)
5096 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5098 return sprintf(buf, "%llu\n",
5099 (unsigned long long) rbd_dev->spec->pool_id);
5102 static ssize_t rbd_pool_ns_show(struct device *dev,
5103 struct device_attribute *attr, char *buf)
5105 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5107 return sprintf(buf, "%s\n", rbd_dev->spec->pool_ns ?: "");
5110 static ssize_t rbd_name_show(struct device *dev,
5111 struct device_attribute *attr, char *buf)
5113 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5115 if (rbd_dev->spec->image_name)
5116 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
5118 return sprintf(buf, "(unknown)\n");
5121 static ssize_t rbd_image_id_show(struct device *dev,
5122 struct device_attribute *attr, char *buf)
5124 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5126 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
5130 * Shows the name of the currently-mapped snapshot (or
5131 * RBD_SNAP_HEAD_NAME for the base image).
5133 static ssize_t rbd_snap_show(struct device *dev,
5134 struct device_attribute *attr,
5137 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5139 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
5142 static ssize_t rbd_snap_id_show(struct device *dev,
5143 struct device_attribute *attr, char *buf)
5145 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5147 return sprintf(buf, "%llu\n", rbd_dev->spec->snap_id);
5151 * For a v2 image, shows the chain of parent images, separated by empty
5152 * lines. For v1 images or if there is no parent, shows "(no parent
5155 static ssize_t rbd_parent_show(struct device *dev,
5156 struct device_attribute *attr,
5159 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5162 if (!rbd_dev->parent)
5163 return sprintf(buf, "(no parent image)\n");
5165 for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) {
5166 struct rbd_spec *spec = rbd_dev->parent_spec;
5168 count += sprintf(&buf[count], "%s"
5169 "pool_id %llu\npool_name %s\n"
5171 "image_id %s\nimage_name %s\n"
5172 "snap_id %llu\nsnap_name %s\n"
5174 !count ? "" : "\n", /* first? */
5175 spec->pool_id, spec->pool_name,
5176 spec->pool_ns ?: "",
5177 spec->image_id, spec->image_name ?: "(unknown)",
5178 spec->snap_id, spec->snap_name,
5179 rbd_dev->parent_overlap);
5185 static ssize_t rbd_image_refresh(struct device *dev,
5186 struct device_attribute *attr,
5190 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5193 if (!capable(CAP_SYS_ADMIN))
5196 ret = rbd_dev_refresh(rbd_dev);
5203 static DEVICE_ATTR(size, 0444, rbd_size_show, NULL);
5204 static DEVICE_ATTR(features, 0444, rbd_features_show, NULL);
5205 static DEVICE_ATTR(major, 0444, rbd_major_show, NULL);
5206 static DEVICE_ATTR(minor, 0444, rbd_minor_show, NULL);
5207 static DEVICE_ATTR(client_addr, 0444, rbd_client_addr_show, NULL);
5208 static DEVICE_ATTR(client_id, 0444, rbd_client_id_show, NULL);
5209 static DEVICE_ATTR(cluster_fsid, 0444, rbd_cluster_fsid_show, NULL);
5210 static DEVICE_ATTR(config_info, 0400, rbd_config_info_show, NULL);
5211 static DEVICE_ATTR(pool, 0444, rbd_pool_show, NULL);
5212 static DEVICE_ATTR(pool_id, 0444, rbd_pool_id_show, NULL);
5213 static DEVICE_ATTR(pool_ns, 0444, rbd_pool_ns_show, NULL);
5214 static DEVICE_ATTR(name, 0444, rbd_name_show, NULL);
5215 static DEVICE_ATTR(image_id, 0444, rbd_image_id_show, NULL);
5216 static DEVICE_ATTR(refresh, 0200, NULL, rbd_image_refresh);
5217 static DEVICE_ATTR(current_snap, 0444, rbd_snap_show, NULL);
5218 static DEVICE_ATTR(snap_id, 0444, rbd_snap_id_show, NULL);
5219 static DEVICE_ATTR(parent, 0444, rbd_parent_show, NULL);
5221 static struct attribute *rbd_attrs[] = {
5222 &dev_attr_size.attr,
5223 &dev_attr_features.attr,
5224 &dev_attr_major.attr,
5225 &dev_attr_minor.attr,
5226 &dev_attr_client_addr.attr,
5227 &dev_attr_client_id.attr,
5228 &dev_attr_cluster_fsid.attr,
5229 &dev_attr_config_info.attr,
5230 &dev_attr_pool.attr,
5231 &dev_attr_pool_id.attr,
5232 &dev_attr_pool_ns.attr,
5233 &dev_attr_name.attr,
5234 &dev_attr_image_id.attr,
5235 &dev_attr_current_snap.attr,
5236 &dev_attr_snap_id.attr,
5237 &dev_attr_parent.attr,
5238 &dev_attr_refresh.attr,
5242 static struct attribute_group rbd_attr_group = {
5246 static const struct attribute_group *rbd_attr_groups[] = {
5251 static void rbd_dev_release(struct device *dev);
5253 static const struct device_type rbd_device_type = {
5255 .groups = rbd_attr_groups,
5256 .release = rbd_dev_release,
5259 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
5261 kref_get(&spec->kref);
5266 static void rbd_spec_free(struct kref *kref);
5267 static void rbd_spec_put(struct rbd_spec *spec)
5270 kref_put(&spec->kref, rbd_spec_free);
5273 static struct rbd_spec *rbd_spec_alloc(void)
5275 struct rbd_spec *spec;
5277 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
5281 spec->pool_id = CEPH_NOPOOL;
5282 spec->snap_id = CEPH_NOSNAP;
5283 kref_init(&spec->kref);
5288 static void rbd_spec_free(struct kref *kref)
5290 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
5292 kfree(spec->pool_name);
5293 kfree(spec->pool_ns);
5294 kfree(spec->image_id);
5295 kfree(spec->image_name);
5296 kfree(spec->snap_name);
5300 static void rbd_dev_free(struct rbd_device *rbd_dev)
5302 WARN_ON(rbd_dev->watch_state != RBD_WATCH_STATE_UNREGISTERED);
5303 WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_UNLOCKED);
5305 ceph_oid_destroy(&rbd_dev->header_oid);
5306 ceph_oloc_destroy(&rbd_dev->header_oloc);
5307 kfree(rbd_dev->config_info);
5309 rbd_put_client(rbd_dev->rbd_client);
5310 rbd_spec_put(rbd_dev->spec);
5311 kfree(rbd_dev->opts);
5315 static void rbd_dev_release(struct device *dev)
5317 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5318 bool need_put = !!rbd_dev->opts;
5321 destroy_workqueue(rbd_dev->task_wq);
5322 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5325 rbd_dev_free(rbd_dev);
5328 * This is racy, but way better than putting module outside of
5329 * the release callback. The race window is pretty small, so
5330 * doing something similar to dm (dm-builtin.c) is overkill.
5333 module_put(THIS_MODULE);
5336 static struct rbd_device *__rbd_dev_create(struct rbd_spec *spec)
5338 struct rbd_device *rbd_dev;
5340 rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
5344 spin_lock_init(&rbd_dev->lock);
5345 INIT_LIST_HEAD(&rbd_dev->node);
5346 init_rwsem(&rbd_dev->header_rwsem);
5348 rbd_dev->header.data_pool_id = CEPH_NOPOOL;
5349 ceph_oid_init(&rbd_dev->header_oid);
5350 rbd_dev->header_oloc.pool = spec->pool_id;
5351 if (spec->pool_ns) {
5352 WARN_ON(!*spec->pool_ns);
5353 rbd_dev->header_oloc.pool_ns =
5354 ceph_find_or_create_string(spec->pool_ns,
5355 strlen(spec->pool_ns));
5358 mutex_init(&rbd_dev->watch_mutex);
5359 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
5360 INIT_DELAYED_WORK(&rbd_dev->watch_dwork, rbd_reregister_watch);
5362 init_rwsem(&rbd_dev->lock_rwsem);
5363 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
5364 INIT_WORK(&rbd_dev->acquired_lock_work, rbd_notify_acquired_lock);
5365 INIT_WORK(&rbd_dev->released_lock_work, rbd_notify_released_lock);
5366 INIT_DELAYED_WORK(&rbd_dev->lock_dwork, rbd_acquire_lock);
5367 INIT_WORK(&rbd_dev->unlock_work, rbd_release_lock_work);
5368 spin_lock_init(&rbd_dev->lock_lists_lock);
5369 INIT_LIST_HEAD(&rbd_dev->acquiring_list);
5370 INIT_LIST_HEAD(&rbd_dev->running_list);
5371 init_completion(&rbd_dev->acquire_wait);
5372 init_completion(&rbd_dev->releasing_wait);
5374 spin_lock_init(&rbd_dev->object_map_lock);
5376 rbd_dev->dev.bus = &rbd_bus_type;
5377 rbd_dev->dev.type = &rbd_device_type;
5378 rbd_dev->dev.parent = &rbd_root_dev;
5379 device_initialize(&rbd_dev->dev);
5385 * Create a mapping rbd_dev.
5387 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
5388 struct rbd_spec *spec,
5389 struct rbd_options *opts)
5391 struct rbd_device *rbd_dev;
5393 rbd_dev = __rbd_dev_create(spec);
5397 /* get an id and fill in device name */
5398 rbd_dev->dev_id = ida_simple_get(&rbd_dev_id_ida, 0,
5399 minor_to_rbd_dev_id(1 << MINORBITS),
5401 if (rbd_dev->dev_id < 0)
5404 sprintf(rbd_dev->name, RBD_DRV_NAME "%d", rbd_dev->dev_id);
5405 rbd_dev->task_wq = alloc_ordered_workqueue("%s-tasks", WQ_MEM_RECLAIM,
5407 if (!rbd_dev->task_wq)
5410 /* we have a ref from do_rbd_add() */
5411 __module_get(THIS_MODULE);
5413 rbd_dev->rbd_client = rbdc;
5414 rbd_dev->spec = spec;
5415 rbd_dev->opts = opts;
5417 dout("%s rbd_dev %p dev_id %d\n", __func__, rbd_dev, rbd_dev->dev_id);
5421 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5423 rbd_dev_free(rbd_dev);
5427 static void rbd_dev_destroy(struct rbd_device *rbd_dev)
5430 put_device(&rbd_dev->dev);
5434 * Get the size and object order for an image snapshot, or if
5435 * snap_id is CEPH_NOSNAP, gets this information for the base
5438 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
5439 u8 *order, u64 *snap_size)
5441 __le64 snapid = cpu_to_le64(snap_id);
5446 } __attribute__ ((packed)) size_buf = { 0 };
5448 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5449 &rbd_dev->header_oloc, "get_size",
5450 &snapid, sizeof(snapid),
5451 &size_buf, sizeof(size_buf));
5452 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5455 if (ret < sizeof (size_buf))
5459 *order = size_buf.order;
5460 dout(" order %u", (unsigned int)*order);
5462 *snap_size = le64_to_cpu(size_buf.size);
5464 dout(" snap_id 0x%016llx snap_size = %llu\n",
5465 (unsigned long long)snap_id,
5466 (unsigned long long)*snap_size);
5471 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
5473 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
5474 &rbd_dev->header.obj_order,
5475 &rbd_dev->header.image_size);
5478 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
5485 /* Response will be an encoded string, which includes a length */
5486 size = sizeof(__le32) + RBD_OBJ_PREFIX_LEN_MAX;
5487 reply_buf = kzalloc(size, GFP_KERNEL);
5491 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5492 &rbd_dev->header_oloc, "get_object_prefix",
5493 NULL, 0, reply_buf, size);
5494 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5499 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
5500 p + ret, NULL, GFP_NOIO);
5503 if (IS_ERR(rbd_dev->header.object_prefix)) {
5504 ret = PTR_ERR(rbd_dev->header.object_prefix);
5505 rbd_dev->header.object_prefix = NULL;
5507 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix);
5515 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
5516 bool read_only, u64 *snap_features)
5525 } __attribute__ ((packed)) features_buf = { 0 };
5529 features_in.snap_id = cpu_to_le64(snap_id);
5530 features_in.read_only = read_only;
5532 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5533 &rbd_dev->header_oloc, "get_features",
5534 &features_in, sizeof(features_in),
5535 &features_buf, sizeof(features_buf));
5536 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5539 if (ret < sizeof (features_buf))
5542 unsup = le64_to_cpu(features_buf.incompat) & ~RBD_FEATURES_SUPPORTED;
5544 rbd_warn(rbd_dev, "image uses unsupported features: 0x%llx",
5549 *snap_features = le64_to_cpu(features_buf.features);
5551 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
5552 (unsigned long long)snap_id,
5553 (unsigned long long)*snap_features,
5554 (unsigned long long)le64_to_cpu(features_buf.incompat));
5559 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
5561 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
5563 &rbd_dev->header.features);
5567 * These are generic image flags, but since they are used only for
5568 * object map, store them in rbd_dev->object_map_flags.
5570 * For the same reason, this function is called only on object map
5571 * (re)load and not on header refresh.
5573 static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev)
5575 __le64 snapid = cpu_to_le64(rbd_dev->spec->snap_id);
5579 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5580 &rbd_dev->header_oloc, "get_flags",
5581 &snapid, sizeof(snapid),
5582 &flags, sizeof(flags));
5585 if (ret < sizeof(flags))
5588 rbd_dev->object_map_flags = le64_to_cpu(flags);
5592 struct parent_image_info {
5594 const char *pool_ns;
5595 const char *image_id;
5603 * The caller is responsible for @pii.
5605 static int decode_parent_image_spec(void **p, void *end,
5606 struct parent_image_info *pii)
5612 ret = ceph_start_decoding(p, end, 1, "ParentImageSpec",
5613 &struct_v, &struct_len);
5617 ceph_decode_64_safe(p, end, pii->pool_id, e_inval);
5618 pii->pool_ns = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5619 if (IS_ERR(pii->pool_ns)) {
5620 ret = PTR_ERR(pii->pool_ns);
5621 pii->pool_ns = NULL;
5624 pii->image_id = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5625 if (IS_ERR(pii->image_id)) {
5626 ret = PTR_ERR(pii->image_id);
5627 pii->image_id = NULL;
5630 ceph_decode_64_safe(p, end, pii->snap_id, e_inval);
5637 static int __get_parent_info(struct rbd_device *rbd_dev,
5638 struct page *req_page,
5639 struct page *reply_page,
5640 struct parent_image_info *pii)
5642 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5643 size_t reply_len = PAGE_SIZE;
5647 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5648 "rbd", "parent_get", CEPH_OSD_FLAG_READ,
5649 req_page, sizeof(u64), &reply_page, &reply_len);
5651 return ret == -EOPNOTSUPP ? 1 : ret;
5653 p = page_address(reply_page);
5654 end = p + reply_len;
5655 ret = decode_parent_image_spec(&p, end, pii);
5659 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5660 "rbd", "parent_overlap_get", CEPH_OSD_FLAG_READ,
5661 req_page, sizeof(u64), &reply_page, &reply_len);
5665 p = page_address(reply_page);
5666 end = p + reply_len;
5667 ceph_decode_8_safe(&p, end, pii->has_overlap, e_inval);
5668 if (pii->has_overlap)
5669 ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5678 * The caller is responsible for @pii.
5680 static int __get_parent_info_legacy(struct rbd_device *rbd_dev,
5681 struct page *req_page,
5682 struct page *reply_page,
5683 struct parent_image_info *pii)
5685 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5686 size_t reply_len = PAGE_SIZE;
5690 ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5691 "rbd", "get_parent", CEPH_OSD_FLAG_READ,
5692 req_page, sizeof(u64), &reply_page, &reply_len);
5696 p = page_address(reply_page);
5697 end = p + reply_len;
5698 ceph_decode_64_safe(&p, end, pii->pool_id, e_inval);
5699 pii->image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
5700 if (IS_ERR(pii->image_id)) {
5701 ret = PTR_ERR(pii->image_id);
5702 pii->image_id = NULL;
5705 ceph_decode_64_safe(&p, end, pii->snap_id, e_inval);
5706 pii->has_overlap = true;
5707 ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5715 static int get_parent_info(struct rbd_device *rbd_dev,
5716 struct parent_image_info *pii)
5718 struct page *req_page, *reply_page;
5722 req_page = alloc_page(GFP_KERNEL);
5726 reply_page = alloc_page(GFP_KERNEL);
5728 __free_page(req_page);
5732 p = page_address(req_page);
5733 ceph_encode_64(&p, rbd_dev->spec->snap_id);
5734 ret = __get_parent_info(rbd_dev, req_page, reply_page, pii);
5736 ret = __get_parent_info_legacy(rbd_dev, req_page, reply_page,
5739 __free_page(req_page);
5740 __free_page(reply_page);
5744 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
5746 struct rbd_spec *parent_spec;
5747 struct parent_image_info pii = { 0 };
5750 parent_spec = rbd_spec_alloc();
5754 ret = get_parent_info(rbd_dev, &pii);
5758 dout("%s pool_id %llu pool_ns %s image_id %s snap_id %llu has_overlap %d overlap %llu\n",
5759 __func__, pii.pool_id, pii.pool_ns, pii.image_id, pii.snap_id,
5760 pii.has_overlap, pii.overlap);
5762 if (pii.pool_id == CEPH_NOPOOL || !pii.has_overlap) {
5764 * Either the parent never existed, or we have
5765 * record of it but the image got flattened so it no
5766 * longer has a parent. When the parent of a
5767 * layered image disappears we immediately set the
5768 * overlap to 0. The effect of this is that all new
5769 * requests will be treated as if the image had no
5772 * If !pii.has_overlap, the parent image spec is not
5773 * applicable. It's there to avoid duplication in each
5776 if (rbd_dev->parent_overlap) {
5777 rbd_dev->parent_overlap = 0;
5778 rbd_dev_parent_put(rbd_dev);
5779 pr_info("%s: clone image has been flattened\n",
5780 rbd_dev->disk->disk_name);
5783 goto out; /* No parent? No problem. */
5786 /* The ceph file layout needs to fit pool id in 32 bits */
5789 if (pii.pool_id > (u64)U32_MAX) {
5790 rbd_warn(NULL, "parent pool id too large (%llu > %u)",
5791 (unsigned long long)pii.pool_id, U32_MAX);
5796 * The parent won't change (except when the clone is
5797 * flattened, already handled that). So we only need to
5798 * record the parent spec we have not already done so.
5800 if (!rbd_dev->parent_spec) {
5801 parent_spec->pool_id = pii.pool_id;
5802 if (pii.pool_ns && *pii.pool_ns) {
5803 parent_spec->pool_ns = pii.pool_ns;
5806 parent_spec->image_id = pii.image_id;
5807 pii.image_id = NULL;
5808 parent_spec->snap_id = pii.snap_id;
5810 rbd_dev->parent_spec = parent_spec;
5811 parent_spec = NULL; /* rbd_dev now owns this */
5815 * We always update the parent overlap. If it's zero we issue
5816 * a warning, as we will proceed as if there was no parent.
5820 /* refresh, careful to warn just once */
5821 if (rbd_dev->parent_overlap)
5823 "clone now standalone (overlap became 0)");
5826 rbd_warn(rbd_dev, "clone is standalone (overlap 0)");
5829 rbd_dev->parent_overlap = pii.overlap;
5835 kfree(pii.image_id);
5836 rbd_spec_put(parent_spec);
5840 static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev)
5844 __le64 stripe_count;
5845 } __attribute__ ((packed)) striping_info_buf = { 0 };
5846 size_t size = sizeof (striping_info_buf);
5850 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5851 &rbd_dev->header_oloc, "get_stripe_unit_count",
5852 NULL, 0, &striping_info_buf, size);
5853 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5859 p = &striping_info_buf;
5860 rbd_dev->header.stripe_unit = ceph_decode_64(&p);
5861 rbd_dev->header.stripe_count = ceph_decode_64(&p);
5865 static int rbd_dev_v2_data_pool(struct rbd_device *rbd_dev)
5867 __le64 data_pool_id;
5870 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5871 &rbd_dev->header_oloc, "get_data_pool",
5872 NULL, 0, &data_pool_id, sizeof(data_pool_id));
5875 if (ret < sizeof(data_pool_id))
5878 rbd_dev->header.data_pool_id = le64_to_cpu(data_pool_id);
5879 WARN_ON(rbd_dev->header.data_pool_id == CEPH_NOPOOL);
5883 static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
5885 CEPH_DEFINE_OID_ONSTACK(oid);
5886 size_t image_id_size;
5891 void *reply_buf = NULL;
5893 char *image_name = NULL;
5896 rbd_assert(!rbd_dev->spec->image_name);
5898 len = strlen(rbd_dev->spec->image_id);
5899 image_id_size = sizeof (__le32) + len;
5900 image_id = kmalloc(image_id_size, GFP_KERNEL);
5905 end = image_id + image_id_size;
5906 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len);
5908 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
5909 reply_buf = kmalloc(size, GFP_KERNEL);
5913 ceph_oid_printf(&oid, "%s", RBD_DIRECTORY);
5914 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
5915 "dir_get_name", image_id, image_id_size,
5920 end = reply_buf + ret;
5922 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
5923 if (IS_ERR(image_name))
5926 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
5934 static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5936 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5937 const char *snap_name;
5940 /* Skip over names until we find the one we are looking for */
5942 snap_name = rbd_dev->header.snap_names;
5943 while (which < snapc->num_snaps) {
5944 if (!strcmp(name, snap_name))
5945 return snapc->snaps[which];
5946 snap_name += strlen(snap_name) + 1;
5952 static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5954 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5959 for (which = 0; !found && which < snapc->num_snaps; which++) {
5960 const char *snap_name;
5962 snap_id = snapc->snaps[which];
5963 snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
5964 if (IS_ERR(snap_name)) {
5965 /* ignore no-longer existing snapshots */
5966 if (PTR_ERR(snap_name) == -ENOENT)
5971 found = !strcmp(name, snap_name);
5974 return found ? snap_id : CEPH_NOSNAP;
5978 * Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if
5979 * no snapshot by that name is found, or if an error occurs.
5981 static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5983 if (rbd_dev->image_format == 1)
5984 return rbd_v1_snap_id_by_name(rbd_dev, name);
5986 return rbd_v2_snap_id_by_name(rbd_dev, name);
5990 * An image being mapped will have everything but the snap id.
5992 static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev)
5994 struct rbd_spec *spec = rbd_dev->spec;
5996 rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name);
5997 rbd_assert(spec->image_id && spec->image_name);
5998 rbd_assert(spec->snap_name);
6000 if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) {
6003 snap_id = rbd_snap_id_by_name(rbd_dev, spec->snap_name);
6004 if (snap_id == CEPH_NOSNAP)
6007 spec->snap_id = snap_id;
6009 spec->snap_id = CEPH_NOSNAP;
6016 * A parent image will have all ids but none of the names.
6018 * All names in an rbd spec are dynamically allocated. It's OK if we
6019 * can't figure out the name for an image id.
6021 static int rbd_spec_fill_names(struct rbd_device *rbd_dev)
6023 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
6024 struct rbd_spec *spec = rbd_dev->spec;
6025 const char *pool_name;
6026 const char *image_name;
6027 const char *snap_name;
6030 rbd_assert(spec->pool_id != CEPH_NOPOOL);
6031 rbd_assert(spec->image_id);
6032 rbd_assert(spec->snap_id != CEPH_NOSNAP);
6034 /* Get the pool name; we have to make our own copy of this */
6036 pool_name = ceph_pg_pool_name_by_id(osdc->osdmap, spec->pool_id);
6038 rbd_warn(rbd_dev, "no pool with id %llu", spec->pool_id);
6041 pool_name = kstrdup(pool_name, GFP_KERNEL);
6045 /* Fetch the image name; tolerate failure here */
6047 image_name = rbd_dev_image_name(rbd_dev);
6049 rbd_warn(rbd_dev, "unable to get image name");
6051 /* Fetch the snapshot name */
6053 snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
6054 if (IS_ERR(snap_name)) {
6055 ret = PTR_ERR(snap_name);
6059 spec->pool_name = pool_name;
6060 spec->image_name = image_name;
6061 spec->snap_name = snap_name;
6071 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev)
6080 struct ceph_snap_context *snapc;
6084 * We'll need room for the seq value (maximum snapshot id),
6085 * snapshot count, and array of that many snapshot ids.
6086 * For now we have a fixed upper limit on the number we're
6087 * prepared to receive.
6089 size = sizeof (__le64) + sizeof (__le32) +
6090 RBD_MAX_SNAP_COUNT * sizeof (__le64);
6091 reply_buf = kzalloc(size, GFP_KERNEL);
6095 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6096 &rbd_dev->header_oloc, "get_snapcontext",
6097 NULL, 0, reply_buf, size);
6098 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6103 end = reply_buf + ret;
6105 ceph_decode_64_safe(&p, end, seq, out);
6106 ceph_decode_32_safe(&p, end, snap_count, out);
6109 * Make sure the reported number of snapshot ids wouldn't go
6110 * beyond the end of our buffer. But before checking that,
6111 * make sure the computed size of the snapshot context we
6112 * allocate is representable in a size_t.
6114 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
6119 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
6123 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
6129 for (i = 0; i < snap_count; i++)
6130 snapc->snaps[i] = ceph_decode_64(&p);
6132 ceph_put_snap_context(rbd_dev->header.snapc);
6133 rbd_dev->header.snapc = snapc;
6135 dout(" snap context seq = %llu, snap_count = %u\n",
6136 (unsigned long long)seq, (unsigned int)snap_count);
6143 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
6154 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
6155 reply_buf = kmalloc(size, GFP_KERNEL);
6157 return ERR_PTR(-ENOMEM);
6159 snapid = cpu_to_le64(snap_id);
6160 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6161 &rbd_dev->header_oloc, "get_snapshot_name",
6162 &snapid, sizeof(snapid), reply_buf, size);
6163 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6165 snap_name = ERR_PTR(ret);
6170 end = reply_buf + ret;
6171 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
6172 if (IS_ERR(snap_name))
6175 dout(" snap_id 0x%016llx snap_name = %s\n",
6176 (unsigned long long)snap_id, snap_name);
6183 static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev)
6185 bool first_time = rbd_dev->header.object_prefix == NULL;
6188 ret = rbd_dev_v2_image_size(rbd_dev);
6193 ret = rbd_dev_v2_header_onetime(rbd_dev);
6198 ret = rbd_dev_v2_snap_context(rbd_dev);
6199 if (ret && first_time) {
6200 kfree(rbd_dev->header.object_prefix);
6201 rbd_dev->header.object_prefix = NULL;
6207 static int rbd_dev_header_info(struct rbd_device *rbd_dev)
6209 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6211 if (rbd_dev->image_format == 1)
6212 return rbd_dev_v1_header_info(rbd_dev);
6214 return rbd_dev_v2_header_info(rbd_dev);
6218 * Skips over white space at *buf, and updates *buf to point to the
6219 * first found non-space character (if any). Returns the length of
6220 * the token (string of non-white space characters) found. Note
6221 * that *buf must be terminated with '\0'.
6223 static inline size_t next_token(const char **buf)
6226 * These are the characters that produce nonzero for
6227 * isspace() in the "C" and "POSIX" locales.
6229 static const char spaces[] = " \f\n\r\t\v";
6231 *buf += strspn(*buf, spaces); /* Find start of token */
6233 return strcspn(*buf, spaces); /* Return token length */
6237 * Finds the next token in *buf, dynamically allocates a buffer big
6238 * enough to hold a copy of it, and copies the token into the new
6239 * buffer. The copy is guaranteed to be terminated with '\0'. Note
6240 * that a duplicate buffer is created even for a zero-length token.
6242 * Returns a pointer to the newly-allocated duplicate, or a null
6243 * pointer if memory for the duplicate was not available. If
6244 * the lenp argument is a non-null pointer, the length of the token
6245 * (not including the '\0') is returned in *lenp.
6247 * If successful, the *buf pointer will be updated to point beyond
6248 * the end of the found token.
6250 * Note: uses GFP_KERNEL for allocation.
6252 static inline char *dup_token(const char **buf, size_t *lenp)
6257 len = next_token(buf);
6258 dup = kmemdup(*buf, len + 1, GFP_KERNEL);
6261 *(dup + len) = '\0';
6270 static int rbd_parse_param(struct fs_parameter *param,
6271 struct rbd_parse_opts_ctx *pctx)
6273 struct rbd_options *opt = pctx->opts;
6274 struct fs_parse_result result;
6275 struct p_log log = {.prefix = "rbd"};
6278 ret = ceph_parse_param(param, pctx->copts, NULL);
6279 if (ret != -ENOPARAM)
6282 token = __fs_parse(&log, rbd_parameters, param, &result);
6283 dout("%s fs_parse '%s' token %d\n", __func__, param->key, token);
6285 if (token == -ENOPARAM)
6286 return inval_plog(&log, "Unknown parameter '%s'",
6292 case Opt_queue_depth:
6293 if (result.uint_32 < 1)
6295 opt->queue_depth = result.uint_32;
6297 case Opt_alloc_size:
6298 if (result.uint_32 < SECTOR_SIZE)
6300 if (!is_power_of_2(result.uint_32))
6301 return inval_plog(&log, "alloc_size must be a power of 2");
6302 opt->alloc_size = result.uint_32;
6304 case Opt_lock_timeout:
6305 /* 0 is "wait forever" (i.e. infinite timeout) */
6306 if (result.uint_32 > INT_MAX / 1000)
6308 opt->lock_timeout = msecs_to_jiffies(result.uint_32 * 1000);
6311 kfree(pctx->spec->pool_ns);
6312 pctx->spec->pool_ns = param->string;
6313 param->string = NULL;
6315 case Opt_compression_hint:
6316 switch (result.uint_32) {
6317 case Opt_compression_hint_none:
6318 opt->alloc_hint_flags &=
6319 ~(CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE |
6320 CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE);
6322 case Opt_compression_hint_compressible:
6323 opt->alloc_hint_flags |=
6324 CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6325 opt->alloc_hint_flags &=
6326 ~CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6328 case Opt_compression_hint_incompressible:
6329 opt->alloc_hint_flags |=
6330 CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6331 opt->alloc_hint_flags &=
6332 ~CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6339 opt->read_only = true;
6341 case Opt_read_write:
6342 opt->read_only = false;
6344 case Opt_lock_on_read:
6345 opt->lock_on_read = true;
6348 opt->exclusive = true;
6360 return inval_plog(&log, "%s out of range", param->key);
6364 * This duplicates most of generic_parse_monolithic(), untying it from
6365 * fs_context and skipping standard superblock and security options.
6367 static int rbd_parse_options(char *options, struct rbd_parse_opts_ctx *pctx)
6372 dout("%s '%s'\n", __func__, options);
6373 while ((key = strsep(&options, ",")) != NULL) {
6375 struct fs_parameter param = {
6377 .type = fs_value_is_flag,
6379 char *value = strchr(key, '=');
6386 v_len = strlen(value);
6387 param.string = kmemdup_nul(value, v_len,
6391 param.type = fs_value_is_string;
6395 ret = rbd_parse_param(¶m, pctx);
6396 kfree(param.string);
6406 * Parse the options provided for an "rbd add" (i.e., rbd image
6407 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
6408 * and the data written is passed here via a NUL-terminated buffer.
6409 * Returns 0 if successful or an error code otherwise.
6411 * The information extracted from these options is recorded in
6412 * the other parameters which return dynamically-allocated
6415 * The address of a pointer that will refer to a ceph options
6416 * structure. Caller must release the returned pointer using
6417 * ceph_destroy_options() when it is no longer needed.
6419 * Address of an rbd options pointer. Fully initialized by
6420 * this function; caller must release with kfree().
6422 * Address of an rbd image specification pointer. Fully
6423 * initialized by this function based on parsed options.
6424 * Caller must release with rbd_spec_put().
6426 * The options passed take this form:
6427 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
6430 * A comma-separated list of one or more monitor addresses.
6431 * A monitor address is an ip address, optionally followed
6432 * by a port number (separated by a colon).
6433 * I.e.: ip1[:port1][,ip2[:port2]...]
6435 * A comma-separated list of ceph and/or rbd options.
6437 * The name of the rados pool containing the rbd image.
6439 * The name of the image in that pool to map.
6441 * An optional snapshot id. If provided, the mapping will
6442 * present data from the image at the time that snapshot was
6443 * created. The image head is used if no snapshot id is
6444 * provided. Snapshot mappings are always read-only.
6446 static int rbd_add_parse_args(const char *buf,
6447 struct ceph_options **ceph_opts,
6448 struct rbd_options **opts,
6449 struct rbd_spec **rbd_spec)
6453 const char *mon_addrs;
6455 size_t mon_addrs_size;
6456 struct rbd_parse_opts_ctx pctx = { 0 };
6459 /* The first four tokens are required */
6461 len = next_token(&buf);
6463 rbd_warn(NULL, "no monitor address(es) provided");
6467 mon_addrs_size = len;
6471 options = dup_token(&buf, NULL);
6475 rbd_warn(NULL, "no options provided");
6479 pctx.spec = rbd_spec_alloc();
6483 pctx.spec->pool_name = dup_token(&buf, NULL);
6484 if (!pctx.spec->pool_name)
6486 if (!*pctx.spec->pool_name) {
6487 rbd_warn(NULL, "no pool name provided");
6491 pctx.spec->image_name = dup_token(&buf, NULL);
6492 if (!pctx.spec->image_name)
6494 if (!*pctx.spec->image_name) {
6495 rbd_warn(NULL, "no image name provided");
6500 * Snapshot name is optional; default is to use "-"
6501 * (indicating the head/no snapshot).
6503 len = next_token(&buf);
6505 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
6506 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
6507 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
6508 ret = -ENAMETOOLONG;
6511 snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
6514 *(snap_name + len) = '\0';
6515 pctx.spec->snap_name = snap_name;
6517 pctx.copts = ceph_alloc_options();
6521 /* Initialize all rbd options to the defaults */
6523 pctx.opts = kzalloc(sizeof(*pctx.opts), GFP_KERNEL);
6527 pctx.opts->read_only = RBD_READ_ONLY_DEFAULT;
6528 pctx.opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
6529 pctx.opts->alloc_size = RBD_ALLOC_SIZE_DEFAULT;
6530 pctx.opts->lock_timeout = RBD_LOCK_TIMEOUT_DEFAULT;
6531 pctx.opts->lock_on_read = RBD_LOCK_ON_READ_DEFAULT;
6532 pctx.opts->exclusive = RBD_EXCLUSIVE_DEFAULT;
6533 pctx.opts->trim = RBD_TRIM_DEFAULT;
6535 ret = ceph_parse_mon_ips(mon_addrs, mon_addrs_size, pctx.copts, NULL,
6540 ret = rbd_parse_options(options, &pctx);
6544 *ceph_opts = pctx.copts;
6546 *rbd_spec = pctx.spec;
6554 ceph_destroy_options(pctx.copts);
6555 rbd_spec_put(pctx.spec);
6560 static void rbd_dev_image_unlock(struct rbd_device *rbd_dev)
6562 down_write(&rbd_dev->lock_rwsem);
6563 if (__rbd_is_lock_owner(rbd_dev))
6564 __rbd_release_lock(rbd_dev);
6565 up_write(&rbd_dev->lock_rwsem);
6569 * If the wait is interrupted, an error is returned even if the lock
6570 * was successfully acquired. rbd_dev_image_unlock() will release it
6573 static int rbd_add_acquire_lock(struct rbd_device *rbd_dev)
6577 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK)) {
6578 if (!rbd_dev->opts->exclusive && !rbd_dev->opts->lock_on_read)
6581 rbd_warn(rbd_dev, "exclusive-lock feature is not enabled");
6585 if (rbd_is_ro(rbd_dev))
6588 rbd_assert(!rbd_is_lock_owner(rbd_dev));
6589 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
6590 ret = wait_for_completion_killable_timeout(&rbd_dev->acquire_wait,
6591 ceph_timeout_jiffies(rbd_dev->opts->lock_timeout));
6593 ret = rbd_dev->acquire_err;
6595 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
6599 rbd_warn(rbd_dev, "failed to acquire lock: %ld", ret);
6605 * The lock may have been released by now, unless automatic lock
6606 * transitions are disabled.
6608 rbd_assert(!rbd_dev->opts->exclusive || rbd_is_lock_owner(rbd_dev));
6613 * An rbd format 2 image has a unique identifier, distinct from the
6614 * name given to it by the user. Internally, that identifier is
6615 * what's used to specify the names of objects related to the image.
6617 * A special "rbd id" object is used to map an rbd image name to its
6618 * id. If that object doesn't exist, then there is no v2 rbd image
6619 * with the supplied name.
6621 * This function will record the given rbd_dev's image_id field if
6622 * it can be determined, and in that case will return 0. If any
6623 * errors occur a negative errno will be returned and the rbd_dev's
6624 * image_id field will be unchanged (and should be NULL).
6626 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
6630 CEPH_DEFINE_OID_ONSTACK(oid);
6635 * When probing a parent image, the image id is already
6636 * known (and the image name likely is not). There's no
6637 * need to fetch the image id again in this case. We
6638 * do still need to set the image format though.
6640 if (rbd_dev->spec->image_id) {
6641 rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
6647 * First, see if the format 2 image id file exists, and if
6648 * so, get the image's persistent id from it.
6650 ret = ceph_oid_aprintf(&oid, GFP_KERNEL, "%s%s", RBD_ID_PREFIX,
6651 rbd_dev->spec->image_name);
6655 dout("rbd id object name is %s\n", oid.name);
6657 /* Response will be an encoded string, which includes a length */
6658 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
6659 response = kzalloc(size, GFP_NOIO);
6665 /* If it doesn't exist we'll assume it's a format 1 image */
6667 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
6670 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6671 if (ret == -ENOENT) {
6672 image_id = kstrdup("", GFP_KERNEL);
6673 ret = image_id ? 0 : -ENOMEM;
6675 rbd_dev->image_format = 1;
6676 } else if (ret >= 0) {
6679 image_id = ceph_extract_encoded_string(&p, p + ret,
6681 ret = PTR_ERR_OR_ZERO(image_id);
6683 rbd_dev->image_format = 2;
6687 rbd_dev->spec->image_id = image_id;
6688 dout("image_id is %s\n", image_id);
6692 ceph_oid_destroy(&oid);
6697 * Undo whatever state changes are made by v1 or v2 header info
6700 static void rbd_dev_unprobe(struct rbd_device *rbd_dev)
6702 struct rbd_image_header *header;
6704 rbd_dev_parent_put(rbd_dev);
6705 rbd_object_map_free(rbd_dev);
6706 rbd_dev_mapping_clear(rbd_dev);
6708 /* Free dynamic fields from the header, then zero it out */
6710 header = &rbd_dev->header;
6711 ceph_put_snap_context(header->snapc);
6712 kfree(header->snap_sizes);
6713 kfree(header->snap_names);
6714 kfree(header->object_prefix);
6715 memset(header, 0, sizeof (*header));
6718 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev)
6722 ret = rbd_dev_v2_object_prefix(rbd_dev);
6727 * Get the and check features for the image. Currently the
6728 * features are assumed to never change.
6730 ret = rbd_dev_v2_features(rbd_dev);
6734 /* If the image supports fancy striping, get its parameters */
6736 if (rbd_dev->header.features & RBD_FEATURE_STRIPINGV2) {
6737 ret = rbd_dev_v2_striping_info(rbd_dev);
6742 if (rbd_dev->header.features & RBD_FEATURE_DATA_POOL) {
6743 ret = rbd_dev_v2_data_pool(rbd_dev);
6748 rbd_init_layout(rbd_dev);
6752 rbd_dev->header.features = 0;
6753 kfree(rbd_dev->header.object_prefix);
6754 rbd_dev->header.object_prefix = NULL;
6759 * @depth is rbd_dev_image_probe() -> rbd_dev_probe_parent() ->
6760 * rbd_dev_image_probe() recursion depth, which means it's also the
6761 * length of the already discovered part of the parent chain.
6763 static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
6765 struct rbd_device *parent = NULL;
6768 if (!rbd_dev->parent_spec)
6771 if (++depth > RBD_MAX_PARENT_CHAIN_LEN) {
6772 pr_info("parent chain is too long (%d)\n", depth);
6777 parent = __rbd_dev_create(rbd_dev->parent_spec);
6784 * Images related by parent/child relationships always share
6785 * rbd_client and spec/parent_spec, so bump their refcounts.
6787 parent->rbd_client = __rbd_get_client(rbd_dev->rbd_client);
6788 parent->spec = rbd_spec_get(rbd_dev->parent_spec);
6790 __set_bit(RBD_DEV_FLAG_READONLY, &parent->flags);
6792 ret = rbd_dev_image_probe(parent, depth);
6796 rbd_dev->parent = parent;
6797 atomic_set(&rbd_dev->parent_ref, 1);
6801 rbd_dev_unparent(rbd_dev);
6802 rbd_dev_destroy(parent);
6806 static void rbd_dev_device_release(struct rbd_device *rbd_dev)
6808 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6809 rbd_free_disk(rbd_dev);
6811 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6815 * rbd_dev->header_rwsem must be locked for write and will be unlocked
6818 static int rbd_dev_device_setup(struct rbd_device *rbd_dev)
6822 /* Record our major and minor device numbers. */
6824 if (!single_major) {
6825 ret = register_blkdev(0, rbd_dev->name);
6827 goto err_out_unlock;
6829 rbd_dev->major = ret;
6832 rbd_dev->major = rbd_major;
6833 rbd_dev->minor = rbd_dev_id_to_minor(rbd_dev->dev_id);
6836 /* Set up the blkdev mapping. */
6838 ret = rbd_init_disk(rbd_dev);
6840 goto err_out_blkdev;
6842 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
6843 set_disk_ro(rbd_dev->disk, rbd_is_ro(rbd_dev));
6845 ret = dev_set_name(&rbd_dev->dev, "%d", rbd_dev->dev_id);
6849 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6850 up_write(&rbd_dev->header_rwsem);
6854 rbd_free_disk(rbd_dev);
6857 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6859 up_write(&rbd_dev->header_rwsem);
6863 static int rbd_dev_header_name(struct rbd_device *rbd_dev)
6865 struct rbd_spec *spec = rbd_dev->spec;
6868 /* Record the header object name for this rbd image. */
6870 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6871 if (rbd_dev->image_format == 1)
6872 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6873 spec->image_name, RBD_SUFFIX);
6875 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6876 RBD_HEADER_PREFIX, spec->image_id);
6881 static void rbd_print_dne(struct rbd_device *rbd_dev, bool is_snap)
6884 pr_info("image %s/%s%s%s does not exist\n",
6885 rbd_dev->spec->pool_name,
6886 rbd_dev->spec->pool_ns ?: "",
6887 rbd_dev->spec->pool_ns ? "/" : "",
6888 rbd_dev->spec->image_name);
6890 pr_info("snap %s/%s%s%s@%s does not exist\n",
6891 rbd_dev->spec->pool_name,
6892 rbd_dev->spec->pool_ns ?: "",
6893 rbd_dev->spec->pool_ns ? "/" : "",
6894 rbd_dev->spec->image_name,
6895 rbd_dev->spec->snap_name);
6899 static void rbd_dev_image_release(struct rbd_device *rbd_dev)
6901 if (!rbd_is_ro(rbd_dev))
6902 rbd_unregister_watch(rbd_dev);
6904 rbd_dev_unprobe(rbd_dev);
6905 rbd_dev->image_format = 0;
6906 kfree(rbd_dev->spec->image_id);
6907 rbd_dev->spec->image_id = NULL;
6911 * Probe for the existence of the header object for the given rbd
6912 * device. If this image is the one being mapped (i.e., not a
6913 * parent), initiate a watch on its header object before using that
6914 * object to get detailed information about the rbd image.
6916 * On success, returns with header_rwsem held for write if called
6919 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
6921 bool need_watch = !rbd_is_ro(rbd_dev);
6925 * Get the id from the image id object. Unless there's an
6926 * error, rbd_dev->spec->image_id will be filled in with
6927 * a dynamically-allocated string, and rbd_dev->image_format
6928 * will be set to either 1 or 2.
6930 ret = rbd_dev_image_id(rbd_dev);
6934 ret = rbd_dev_header_name(rbd_dev);
6936 goto err_out_format;
6939 ret = rbd_register_watch(rbd_dev);
6942 rbd_print_dne(rbd_dev, false);
6943 goto err_out_format;
6948 down_write(&rbd_dev->header_rwsem);
6950 ret = rbd_dev_header_info(rbd_dev);
6952 if (ret == -ENOENT && !need_watch)
6953 rbd_print_dne(rbd_dev, false);
6958 * If this image is the one being mapped, we have pool name and
6959 * id, image name and id, and snap name - need to fill snap id.
6960 * Otherwise this is a parent image, identified by pool, image
6961 * and snap ids - need to fill in names for those ids.
6964 ret = rbd_spec_fill_snap_id(rbd_dev);
6966 ret = rbd_spec_fill_names(rbd_dev);
6969 rbd_print_dne(rbd_dev, true);
6973 ret = rbd_dev_mapping_set(rbd_dev);
6977 if (rbd_is_snap(rbd_dev) &&
6978 (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)) {
6979 ret = rbd_object_map_load(rbd_dev);
6984 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
6985 ret = rbd_dev_v2_parent_info(rbd_dev);
6990 ret = rbd_dev_probe_parent(rbd_dev, depth);
6994 dout("discovered format %u image, header name is %s\n",
6995 rbd_dev->image_format, rbd_dev->header_oid.name);
7000 up_write(&rbd_dev->header_rwsem);
7002 rbd_unregister_watch(rbd_dev);
7003 rbd_dev_unprobe(rbd_dev);
7005 rbd_dev->image_format = 0;
7006 kfree(rbd_dev->spec->image_id);
7007 rbd_dev->spec->image_id = NULL;
7011 static int rbd_dev_refresh(struct rbd_device *rbd_dev)
7016 down_write(&rbd_dev->header_rwsem);
7017 mapping_size = rbd_dev->mapping.size;
7019 ret = rbd_dev_header_info(rbd_dev);
7024 * If there is a parent, see if it has disappeared due to the
7025 * mapped image getting flattened.
7027 if (rbd_dev->parent) {
7028 ret = rbd_dev_v2_parent_info(rbd_dev);
7033 rbd_assert(!rbd_is_snap(rbd_dev));
7034 rbd_dev->mapping.size = rbd_dev->header.image_size;
7037 up_write(&rbd_dev->header_rwsem);
7038 if (!ret && mapping_size != rbd_dev->mapping.size)
7039 rbd_dev_update_size(rbd_dev);
7044 static ssize_t do_rbd_add(const char *buf, size_t count)
7046 struct rbd_device *rbd_dev = NULL;
7047 struct ceph_options *ceph_opts = NULL;
7048 struct rbd_options *rbd_opts = NULL;
7049 struct rbd_spec *spec = NULL;
7050 struct rbd_client *rbdc;
7053 if (!capable(CAP_SYS_ADMIN))
7056 if (!try_module_get(THIS_MODULE))
7059 /* parse add command */
7060 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
7064 rbdc = rbd_get_client(ceph_opts);
7071 rc = ceph_pg_poolid_by_name(rbdc->client->osdc.osdmap, spec->pool_name);
7074 pr_info("pool %s does not exist\n", spec->pool_name);
7075 goto err_out_client;
7077 spec->pool_id = (u64)rc;
7079 rbd_dev = rbd_dev_create(rbdc, spec, rbd_opts);
7082 goto err_out_client;
7084 rbdc = NULL; /* rbd_dev now owns this */
7085 spec = NULL; /* rbd_dev now owns this */
7086 rbd_opts = NULL; /* rbd_dev now owns this */
7088 /* if we are mapping a snapshot it will be a read-only mapping */
7089 if (rbd_dev->opts->read_only ||
7090 strcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME))
7091 __set_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
7093 rbd_dev->config_info = kstrdup(buf, GFP_KERNEL);
7094 if (!rbd_dev->config_info) {
7096 goto err_out_rbd_dev;
7099 rc = rbd_dev_image_probe(rbd_dev, 0);
7101 goto err_out_rbd_dev;
7103 if (rbd_dev->opts->alloc_size > rbd_dev->layout.object_size) {
7104 rbd_warn(rbd_dev, "alloc_size adjusted to %u",
7105 rbd_dev->layout.object_size);
7106 rbd_dev->opts->alloc_size = rbd_dev->layout.object_size;
7109 rc = rbd_dev_device_setup(rbd_dev);
7111 goto err_out_image_probe;
7113 rc = rbd_add_acquire_lock(rbd_dev);
7115 goto err_out_image_lock;
7117 /* Everything's ready. Announce the disk to the world. */
7119 rc = device_add(&rbd_dev->dev);
7121 goto err_out_image_lock;
7123 rc = device_add_disk(&rbd_dev->dev, rbd_dev->disk, NULL);
7125 goto err_out_cleanup_disk;
7127 spin_lock(&rbd_dev_list_lock);
7128 list_add_tail(&rbd_dev->node, &rbd_dev_list);
7129 spin_unlock(&rbd_dev_list_lock);
7131 pr_info("%s: capacity %llu features 0x%llx\n", rbd_dev->disk->disk_name,
7132 (unsigned long long)get_capacity(rbd_dev->disk) << SECTOR_SHIFT,
7133 rbd_dev->header.features);
7136 module_put(THIS_MODULE);
7139 err_out_cleanup_disk:
7140 rbd_free_disk(rbd_dev);
7142 rbd_dev_image_unlock(rbd_dev);
7143 rbd_dev_device_release(rbd_dev);
7144 err_out_image_probe:
7145 rbd_dev_image_release(rbd_dev);
7147 rbd_dev_destroy(rbd_dev);
7149 rbd_put_client(rbdc);
7156 static ssize_t add_store(const struct bus_type *bus, const char *buf, size_t count)
7161 return do_rbd_add(buf, count);
7164 static ssize_t add_single_major_store(const struct bus_type *bus, const char *buf,
7167 return do_rbd_add(buf, count);
7170 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev)
7172 while (rbd_dev->parent) {
7173 struct rbd_device *first = rbd_dev;
7174 struct rbd_device *second = first->parent;
7175 struct rbd_device *third;
7178 * Follow to the parent with no grandparent and
7181 while (second && (third = second->parent)) {
7186 rbd_dev_image_release(second);
7187 rbd_dev_destroy(second);
7188 first->parent = NULL;
7189 first->parent_overlap = 0;
7191 rbd_assert(first->parent_spec);
7192 rbd_spec_put(first->parent_spec);
7193 first->parent_spec = NULL;
7197 static ssize_t do_rbd_remove(const char *buf, size_t count)
7199 struct rbd_device *rbd_dev = NULL;
7205 if (!capable(CAP_SYS_ADMIN))
7210 sscanf(buf, "%d %5s", &dev_id, opt_buf);
7212 pr_err("dev_id out of range\n");
7215 if (opt_buf[0] != '\0') {
7216 if (!strcmp(opt_buf, "force")) {
7219 pr_err("bad remove option at '%s'\n", opt_buf);
7225 spin_lock(&rbd_dev_list_lock);
7226 list_for_each_entry(rbd_dev, &rbd_dev_list, node) {
7227 if (rbd_dev->dev_id == dev_id) {
7233 spin_lock_irq(&rbd_dev->lock);
7234 if (rbd_dev->open_count && !force)
7236 else if (test_and_set_bit(RBD_DEV_FLAG_REMOVING,
7239 spin_unlock_irq(&rbd_dev->lock);
7241 spin_unlock(&rbd_dev_list_lock);
7247 * Prevent new IO from being queued and wait for existing
7248 * IO to complete/fail.
7250 blk_mq_freeze_queue(rbd_dev->disk->queue);
7251 blk_mark_disk_dead(rbd_dev->disk);
7254 del_gendisk(rbd_dev->disk);
7255 spin_lock(&rbd_dev_list_lock);
7256 list_del_init(&rbd_dev->node);
7257 spin_unlock(&rbd_dev_list_lock);
7258 device_del(&rbd_dev->dev);
7260 rbd_dev_image_unlock(rbd_dev);
7261 rbd_dev_device_release(rbd_dev);
7262 rbd_dev_image_release(rbd_dev);
7263 rbd_dev_destroy(rbd_dev);
7267 static ssize_t remove_store(const struct bus_type *bus, const char *buf, size_t count)
7272 return do_rbd_remove(buf, count);
7275 static ssize_t remove_single_major_store(const struct bus_type *bus, const char *buf,
7278 return do_rbd_remove(buf, count);
7282 * create control files in sysfs
7285 static int __init rbd_sysfs_init(void)
7289 ret = device_register(&rbd_root_dev);
7291 put_device(&rbd_root_dev);
7295 ret = bus_register(&rbd_bus_type);
7297 device_unregister(&rbd_root_dev);
7302 static void __exit rbd_sysfs_cleanup(void)
7304 bus_unregister(&rbd_bus_type);
7305 device_unregister(&rbd_root_dev);
7308 static int __init rbd_slab_init(void)
7310 rbd_assert(!rbd_img_request_cache);
7311 rbd_img_request_cache = KMEM_CACHE(rbd_img_request, 0);
7312 if (!rbd_img_request_cache)
7315 rbd_assert(!rbd_obj_request_cache);
7316 rbd_obj_request_cache = KMEM_CACHE(rbd_obj_request, 0);
7317 if (!rbd_obj_request_cache)
7323 kmem_cache_destroy(rbd_img_request_cache);
7324 rbd_img_request_cache = NULL;
7328 static void rbd_slab_exit(void)
7330 rbd_assert(rbd_obj_request_cache);
7331 kmem_cache_destroy(rbd_obj_request_cache);
7332 rbd_obj_request_cache = NULL;
7334 rbd_assert(rbd_img_request_cache);
7335 kmem_cache_destroy(rbd_img_request_cache);
7336 rbd_img_request_cache = NULL;
7339 static int __init rbd_init(void)
7343 if (!libceph_compatible(NULL)) {
7344 rbd_warn(NULL, "libceph incompatibility (quitting)");
7348 rc = rbd_slab_init();
7353 * The number of active work items is limited by the number of
7354 * rbd devices * queue depth, so leave @max_active at default.
7356 rbd_wq = alloc_workqueue(RBD_DRV_NAME, WQ_MEM_RECLAIM, 0);
7363 rbd_major = register_blkdev(0, RBD_DRV_NAME);
7364 if (rbd_major < 0) {
7370 rc = rbd_sysfs_init();
7372 goto err_out_blkdev;
7375 pr_info("loaded (major %d)\n", rbd_major);
7377 pr_info("loaded\n");
7383 unregister_blkdev(rbd_major, RBD_DRV_NAME);
7385 destroy_workqueue(rbd_wq);
7391 static void __exit rbd_exit(void)
7393 ida_destroy(&rbd_dev_id_ida);
7394 rbd_sysfs_cleanup();
7396 unregister_blkdev(rbd_major, RBD_DRV_NAME);
7397 destroy_workqueue(rbd_wq);
7401 module_init(rbd_init);
7402 module_exit(rbd_exit);
7404 MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
7405 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
7406 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
7407 /* following authorship retained from original osdblk.c */
7408 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
7410 MODULE_DESCRIPTION("RADOS Block Device (RBD) driver");
7411 MODULE_LICENSE("GPL");