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/decode.h>
36 #include <linux/parser.h>
37 #include <linux/bsearch.h>
39 #include <linux/kernel.h>
40 #include <linux/device.h>
41 #include <linux/module.h>
42 #include <linux/blk-mq.h>
44 #include <linux/blkdev.h>
45 #include <linux/slab.h>
46 #include <linux/idr.h>
47 #include <linux/workqueue.h>
49 #include "rbd_types.h"
51 #define RBD_DEBUG /* Activate rbd_assert() calls */
54 * The basic unit of block I/O is a sector. It is interpreted in a
55 * number of contexts in Linux (blk, bio, genhd), but the default is
56 * universally 512 bytes. These symbols are just slightly more
57 * meaningful than the bare numbers they represent.
59 #define SECTOR_SHIFT 9
60 #define SECTOR_SIZE (1ULL << SECTOR_SHIFT)
63 * Increment the given counter and return its updated value.
64 * If the counter is already 0 it will not be incremented.
65 * If the counter is already at its maximum value returns
66 * -EINVAL without updating it.
68 static int atomic_inc_return_safe(atomic_t *v)
72 counter = (unsigned int)__atomic_add_unless(v, 1, 0);
73 if (counter <= (unsigned int)INT_MAX)
81 /* Decrement the counter. Return the resulting value, or -EINVAL */
82 static int atomic_dec_return_safe(atomic_t *v)
86 counter = atomic_dec_return(v);
95 #define RBD_DRV_NAME "rbd"
97 #define RBD_MINORS_PER_MAJOR 256
98 #define RBD_SINGLE_MAJOR_PART_SHIFT 4
100 #define RBD_MAX_PARENT_CHAIN_LEN 16
102 #define RBD_SNAP_DEV_NAME_PREFIX "snap_"
103 #define RBD_MAX_SNAP_NAME_LEN \
104 (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
106 #define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */
108 #define RBD_SNAP_HEAD_NAME "-"
110 #define BAD_SNAP_INDEX U32_MAX /* invalid index into snap array */
112 /* This allows a single page to hold an image name sent by OSD */
113 #define RBD_IMAGE_NAME_LEN_MAX (PAGE_SIZE - sizeof (__le32) - 1)
114 #define RBD_IMAGE_ID_LEN_MAX 64
116 #define RBD_OBJ_PREFIX_LEN_MAX 64
118 #define RBD_NOTIFY_TIMEOUT 5 /* seconds */
119 #define RBD_RETRY_DELAY msecs_to_jiffies(1000)
123 #define RBD_FEATURE_LAYERING (1ULL<<0)
124 #define RBD_FEATURE_STRIPINGV2 (1ULL<<1)
125 #define RBD_FEATURE_EXCLUSIVE_LOCK (1ULL<<2)
126 #define RBD_FEATURE_DATA_POOL (1ULL<<7)
128 #define RBD_FEATURES_ALL (RBD_FEATURE_LAYERING | \
129 RBD_FEATURE_STRIPINGV2 | \
130 RBD_FEATURE_EXCLUSIVE_LOCK | \
131 RBD_FEATURE_DATA_POOL)
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;
191 const char *image_id;
192 const char *image_name;
195 const char *snap_name;
201 * an instance of the client. multiple devices may share an rbd client.
204 struct ceph_client *client;
206 struct list_head node;
209 struct rbd_img_request;
210 typedef void (*rbd_img_callback_t)(struct rbd_img_request *);
212 #define BAD_WHICH U32_MAX /* Good which or bad which, which? */
214 struct rbd_obj_request;
215 typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);
217 enum obj_request_type {
218 OBJ_REQUEST_NODATA, OBJ_REQUEST_BIO, OBJ_REQUEST_PAGES
221 enum obj_operation_type {
228 OBJ_REQ_DONE, /* completion flag: not done = 0, done = 1 */
229 OBJ_REQ_IMG_DATA, /* object usage: standalone = 0, image = 1 */
230 OBJ_REQ_KNOWN, /* EXISTS flag valid: no = 0, yes = 1 */
231 OBJ_REQ_EXISTS, /* target exists: no = 0, yes = 1 */
234 struct rbd_obj_request {
236 u64 offset; /* object start byte */
237 u64 length; /* bytes from offset */
241 * An object request associated with an image will have its
242 * img_data flag set; a standalone object request will not.
244 * A standalone object request will have which == BAD_WHICH
245 * and a null obj_request pointer.
247 * An object request initiated in support of a layered image
248 * object (to check for its existence before a write) will
249 * have which == BAD_WHICH and a non-null obj_request pointer.
251 * Finally, an object request for rbd image data will have
252 * which != BAD_WHICH, and will have a non-null img_request
253 * pointer. The value of which will be in the range
254 * 0..(img_request->obj_request_count-1).
257 struct rbd_obj_request *obj_request; /* STAT op */
259 struct rbd_img_request *img_request;
261 /* links for img_request->obj_requests list */
262 struct list_head links;
265 u32 which; /* posn image request list */
267 enum obj_request_type type;
269 struct bio *bio_list;
275 struct page **copyup_pages;
276 u32 copyup_page_count;
278 struct ceph_osd_request *osd_req;
280 u64 xferred; /* bytes transferred */
283 rbd_obj_callback_t callback;
284 struct completion completion;
290 IMG_REQ_WRITE, /* I/O direction: read = 0, write = 1 */
291 IMG_REQ_CHILD, /* initiator: block = 0, child image = 1 */
292 IMG_REQ_LAYERED, /* ENOENT handling: normal = 0, layered = 1 */
293 IMG_REQ_DISCARD, /* discard: normal = 0, discard request = 1 */
296 struct rbd_img_request {
297 struct rbd_device *rbd_dev;
298 u64 offset; /* starting image byte offset */
299 u64 length; /* byte count from offset */
302 u64 snap_id; /* for reads */
303 struct ceph_snap_context *snapc; /* for writes */
306 struct request *rq; /* block request */
307 struct rbd_obj_request *obj_request; /* obj req initiator */
309 struct page **copyup_pages;
310 u32 copyup_page_count;
311 spinlock_t completion_lock;/* protects next_completion */
313 rbd_img_callback_t callback;
314 u64 xferred;/* aggregate bytes transferred */
315 int result; /* first nonzero obj_request result */
317 u32 obj_request_count;
318 struct list_head obj_requests; /* rbd_obj_request structs */
323 #define for_each_obj_request(ireq, oreq) \
324 list_for_each_entry(oreq, &(ireq)->obj_requests, links)
325 #define for_each_obj_request_from(ireq, oreq) \
326 list_for_each_entry_from(oreq, &(ireq)->obj_requests, links)
327 #define for_each_obj_request_safe(ireq, oreq, n) \
328 list_for_each_entry_safe_reverse(oreq, n, &(ireq)->obj_requests, links)
330 enum rbd_watch_state {
331 RBD_WATCH_STATE_UNREGISTERED,
332 RBD_WATCH_STATE_REGISTERED,
333 RBD_WATCH_STATE_ERROR,
336 enum rbd_lock_state {
337 RBD_LOCK_STATE_UNLOCKED,
338 RBD_LOCK_STATE_LOCKED,
339 RBD_LOCK_STATE_RELEASING,
342 /* WatchNotify::ClientId */
343 struct rbd_client_id {
358 int dev_id; /* blkdev unique id */
360 int major; /* blkdev assigned major */
362 struct gendisk *disk; /* blkdev's gendisk and rq */
364 u32 image_format; /* Either 1 or 2 */
365 struct rbd_client *rbd_client;
367 char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
369 spinlock_t lock; /* queue, flags, open_count */
371 struct rbd_image_header header;
372 unsigned long flags; /* possibly lock protected */
373 struct rbd_spec *spec;
374 struct rbd_options *opts;
375 char *config_info; /* add{,_single_major} string */
377 struct ceph_object_id header_oid;
378 struct ceph_object_locator header_oloc;
380 struct ceph_file_layout layout; /* used for all rbd requests */
382 struct mutex watch_mutex;
383 enum rbd_watch_state watch_state;
384 struct ceph_osd_linger_request *watch_handle;
386 struct delayed_work watch_dwork;
388 struct rw_semaphore lock_rwsem;
389 enum rbd_lock_state lock_state;
390 char lock_cookie[32];
391 struct rbd_client_id owner_cid;
392 struct work_struct acquired_lock_work;
393 struct work_struct released_lock_work;
394 struct delayed_work lock_dwork;
395 struct work_struct unlock_work;
396 wait_queue_head_t lock_waitq;
398 struct workqueue_struct *task_wq;
400 struct rbd_spec *parent_spec;
403 struct rbd_device *parent;
405 /* Block layer tags. */
406 struct blk_mq_tag_set tag_set;
408 /* protects updating the header */
409 struct rw_semaphore header_rwsem;
411 struct rbd_mapping mapping;
413 struct list_head node;
417 unsigned long open_count; /* protected by lock */
421 * Flag bits for rbd_dev->flags:
422 * - REMOVING (which is coupled with rbd_dev->open_count) is protected
424 * - BLACKLISTED is protected by rbd_dev->lock_rwsem
427 RBD_DEV_FLAG_EXISTS, /* mapped snapshot has not been deleted */
428 RBD_DEV_FLAG_REMOVING, /* this mapping is being removed */
429 RBD_DEV_FLAG_BLACKLISTED, /* our ceph_client is blacklisted */
432 static DEFINE_MUTEX(client_mutex); /* Serialize client creation */
434 static LIST_HEAD(rbd_dev_list); /* devices */
435 static DEFINE_SPINLOCK(rbd_dev_list_lock);
437 static LIST_HEAD(rbd_client_list); /* clients */
438 static DEFINE_SPINLOCK(rbd_client_list_lock);
440 /* Slab caches for frequently-allocated structures */
442 static struct kmem_cache *rbd_img_request_cache;
443 static struct kmem_cache *rbd_obj_request_cache;
445 static int rbd_major;
446 static DEFINE_IDA(rbd_dev_id_ida);
448 static struct workqueue_struct *rbd_wq;
451 * Default to false for now, as single-major requires >= 0.75 version of
452 * userspace rbd utility.
454 static bool single_major = false;
455 module_param(single_major, bool, S_IRUGO);
456 MODULE_PARM_DESC(single_major, "Use a single major number for all rbd devices (default: false)");
458 static int rbd_img_request_submit(struct rbd_img_request *img_request);
460 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
462 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
464 static ssize_t rbd_add_single_major(struct bus_type *bus, const char *buf,
466 static ssize_t rbd_remove_single_major(struct bus_type *bus, const char *buf,
468 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth);
469 static void rbd_spec_put(struct rbd_spec *spec);
471 static int rbd_dev_id_to_minor(int dev_id)
473 return dev_id << RBD_SINGLE_MAJOR_PART_SHIFT;
476 static int minor_to_rbd_dev_id(int minor)
478 return minor >> RBD_SINGLE_MAJOR_PART_SHIFT;
481 static bool __rbd_is_lock_owner(struct rbd_device *rbd_dev)
483 return rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED ||
484 rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING;
487 static bool rbd_is_lock_owner(struct rbd_device *rbd_dev)
491 down_read(&rbd_dev->lock_rwsem);
492 is_lock_owner = __rbd_is_lock_owner(rbd_dev);
493 up_read(&rbd_dev->lock_rwsem);
494 return is_lock_owner;
497 static ssize_t rbd_supported_features_show(struct bus_type *bus, char *buf)
499 return sprintf(buf, "0x%llx\n", RBD_FEATURES_SUPPORTED);
502 static BUS_ATTR(add, S_IWUSR, NULL, rbd_add);
503 static BUS_ATTR(remove, S_IWUSR, NULL, rbd_remove);
504 static BUS_ATTR(add_single_major, S_IWUSR, NULL, rbd_add_single_major);
505 static BUS_ATTR(remove_single_major, S_IWUSR, NULL, rbd_remove_single_major);
506 static BUS_ATTR(supported_features, S_IRUGO, rbd_supported_features_show, NULL);
508 static struct attribute *rbd_bus_attrs[] = {
510 &bus_attr_remove.attr,
511 &bus_attr_add_single_major.attr,
512 &bus_attr_remove_single_major.attr,
513 &bus_attr_supported_features.attr,
517 static umode_t rbd_bus_is_visible(struct kobject *kobj,
518 struct attribute *attr, int index)
521 (attr == &bus_attr_add_single_major.attr ||
522 attr == &bus_attr_remove_single_major.attr))
528 static const struct attribute_group rbd_bus_group = {
529 .attrs = rbd_bus_attrs,
530 .is_visible = rbd_bus_is_visible,
532 __ATTRIBUTE_GROUPS(rbd_bus);
534 static struct bus_type rbd_bus_type = {
536 .bus_groups = rbd_bus_groups,
539 static void rbd_root_dev_release(struct device *dev)
543 static struct device rbd_root_dev = {
545 .release = rbd_root_dev_release,
548 static __printf(2, 3)
549 void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
551 struct va_format vaf;
559 printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
560 else if (rbd_dev->disk)
561 printk(KERN_WARNING "%s: %s: %pV\n",
562 RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
563 else if (rbd_dev->spec && rbd_dev->spec->image_name)
564 printk(KERN_WARNING "%s: image %s: %pV\n",
565 RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
566 else if (rbd_dev->spec && rbd_dev->spec->image_id)
567 printk(KERN_WARNING "%s: id %s: %pV\n",
568 RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
570 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
571 RBD_DRV_NAME, rbd_dev, &vaf);
576 #define rbd_assert(expr) \
577 if (unlikely(!(expr))) { \
578 printk(KERN_ERR "\nAssertion failure in %s() " \
580 "\trbd_assert(%s);\n\n", \
581 __func__, __LINE__, #expr); \
584 #else /* !RBD_DEBUG */
585 # define rbd_assert(expr) ((void) 0)
586 #endif /* !RBD_DEBUG */
588 static void rbd_osd_copyup_callback(struct rbd_obj_request *obj_request);
589 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request);
590 static void rbd_img_parent_read(struct rbd_obj_request *obj_request);
591 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev);
593 static int rbd_dev_refresh(struct rbd_device *rbd_dev);
594 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev);
595 static int rbd_dev_header_info(struct rbd_device *rbd_dev);
596 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev);
597 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
599 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
600 u8 *order, u64 *snap_size);
601 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
604 static int rbd_open(struct block_device *bdev, fmode_t mode)
606 struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
607 bool removing = false;
609 if ((mode & FMODE_WRITE) && rbd_dev->mapping.read_only)
612 spin_lock_irq(&rbd_dev->lock);
613 if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
616 rbd_dev->open_count++;
617 spin_unlock_irq(&rbd_dev->lock);
621 (void) get_device(&rbd_dev->dev);
626 static void rbd_release(struct gendisk *disk, fmode_t mode)
628 struct rbd_device *rbd_dev = disk->private_data;
629 unsigned long open_count_before;
631 spin_lock_irq(&rbd_dev->lock);
632 open_count_before = rbd_dev->open_count--;
633 spin_unlock_irq(&rbd_dev->lock);
634 rbd_assert(open_count_before > 0);
636 put_device(&rbd_dev->dev);
639 static int rbd_ioctl_set_ro(struct rbd_device *rbd_dev, unsigned long arg)
644 bool ro_changed = false;
646 /* get_user() may sleep, so call it before taking rbd_dev->lock */
647 if (get_user(val, (int __user *)(arg)))
650 ro = val ? true : false;
651 /* Snapshot doesn't allow to write*/
652 if (rbd_dev->spec->snap_id != CEPH_NOSNAP && !ro)
655 spin_lock_irq(&rbd_dev->lock);
656 /* prevent others open this device */
657 if (rbd_dev->open_count > 1) {
662 if (rbd_dev->mapping.read_only != ro) {
663 rbd_dev->mapping.read_only = ro;
668 spin_unlock_irq(&rbd_dev->lock);
669 /* set_disk_ro() may sleep, so call it after releasing rbd_dev->lock */
670 if (ret == 0 && ro_changed)
671 set_disk_ro(rbd_dev->disk, ro ? 1 : 0);
676 static int rbd_ioctl(struct block_device *bdev, fmode_t mode,
677 unsigned int cmd, unsigned long arg)
679 struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
684 ret = rbd_ioctl_set_ro(rbd_dev, arg);
694 static int rbd_compat_ioctl(struct block_device *bdev, fmode_t mode,
695 unsigned int cmd, unsigned long arg)
697 return rbd_ioctl(bdev, mode, cmd, arg);
699 #endif /* CONFIG_COMPAT */
701 static const struct block_device_operations rbd_bd_ops = {
702 .owner = THIS_MODULE,
704 .release = rbd_release,
707 .compat_ioctl = rbd_compat_ioctl,
712 * Initialize an rbd client instance. Success or not, this function
713 * consumes ceph_opts. Caller holds client_mutex.
715 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
717 struct rbd_client *rbdc;
720 dout("%s:\n", __func__);
721 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
725 kref_init(&rbdc->kref);
726 INIT_LIST_HEAD(&rbdc->node);
728 rbdc->client = ceph_create_client(ceph_opts, rbdc);
729 if (IS_ERR(rbdc->client))
731 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
733 ret = ceph_open_session(rbdc->client);
737 spin_lock(&rbd_client_list_lock);
738 list_add_tail(&rbdc->node, &rbd_client_list);
739 spin_unlock(&rbd_client_list_lock);
741 dout("%s: rbdc %p\n", __func__, rbdc);
745 ceph_destroy_client(rbdc->client);
750 ceph_destroy_options(ceph_opts);
751 dout("%s: error %d\n", __func__, ret);
756 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
758 kref_get(&rbdc->kref);
764 * Find a ceph client with specific addr and configuration. If
765 * found, bump its reference count.
767 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
769 struct rbd_client *client_node;
772 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
775 spin_lock(&rbd_client_list_lock);
776 list_for_each_entry(client_node, &rbd_client_list, node) {
777 if (!ceph_compare_options(ceph_opts, client_node->client)) {
778 __rbd_get_client(client_node);
784 spin_unlock(&rbd_client_list_lock);
786 return found ? client_node : NULL;
790 * (Per device) rbd map options
797 /* string args above */
805 static match_table_t rbd_opts_tokens = {
806 {Opt_queue_depth, "queue_depth=%d"},
808 /* string args above */
809 {Opt_read_only, "read_only"},
810 {Opt_read_only, "ro"}, /* Alternate spelling */
811 {Opt_read_write, "read_write"},
812 {Opt_read_write, "rw"}, /* Alternate spelling */
813 {Opt_lock_on_read, "lock_on_read"},
814 {Opt_exclusive, "exclusive"},
825 #define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_MAX_RQ
826 #define RBD_READ_ONLY_DEFAULT false
827 #define RBD_LOCK_ON_READ_DEFAULT false
828 #define RBD_EXCLUSIVE_DEFAULT false
830 static int parse_rbd_opts_token(char *c, void *private)
832 struct rbd_options *rbd_opts = private;
833 substring_t argstr[MAX_OPT_ARGS];
834 int token, intval, ret;
836 token = match_token(c, rbd_opts_tokens, argstr);
837 if (token < Opt_last_int) {
838 ret = match_int(&argstr[0], &intval);
840 pr_err("bad mount option arg (not int) at '%s'\n", c);
843 dout("got int token %d val %d\n", token, intval);
844 } else if (token > Opt_last_int && token < Opt_last_string) {
845 dout("got string token %d val %s\n", token, argstr[0].from);
847 dout("got token %d\n", token);
851 case Opt_queue_depth:
853 pr_err("queue_depth out of range\n");
856 rbd_opts->queue_depth = intval;
859 rbd_opts->read_only = true;
862 rbd_opts->read_only = false;
864 case Opt_lock_on_read:
865 rbd_opts->lock_on_read = true;
868 rbd_opts->exclusive = true;
871 /* libceph prints "bad option" msg */
878 static char* obj_op_name(enum obj_operation_type op_type)
893 * Get a ceph client with specific addr and configuration, if one does
894 * not exist create it. Either way, ceph_opts is consumed by this
897 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
899 struct rbd_client *rbdc;
901 mutex_lock_nested(&client_mutex, SINGLE_DEPTH_NESTING);
902 rbdc = rbd_client_find(ceph_opts);
903 if (rbdc) /* using an existing client */
904 ceph_destroy_options(ceph_opts);
906 rbdc = rbd_client_create(ceph_opts);
907 mutex_unlock(&client_mutex);
913 * Destroy ceph client
915 * Caller must hold rbd_client_list_lock.
917 static void rbd_client_release(struct kref *kref)
919 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
921 dout("%s: rbdc %p\n", __func__, rbdc);
922 spin_lock(&rbd_client_list_lock);
923 list_del(&rbdc->node);
924 spin_unlock(&rbd_client_list_lock);
926 ceph_destroy_client(rbdc->client);
931 * Drop reference to ceph client node. If it's not referenced anymore, release
934 static void rbd_put_client(struct rbd_client *rbdc)
937 kref_put(&rbdc->kref, rbd_client_release);
940 static bool rbd_image_format_valid(u32 image_format)
942 return image_format == 1 || image_format == 2;
945 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
950 /* The header has to start with the magic rbd header text */
951 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
954 /* The bio layer requires at least sector-sized I/O */
956 if (ondisk->options.order < SECTOR_SHIFT)
959 /* If we use u64 in a few spots we may be able to loosen this */
961 if (ondisk->options.order > 8 * sizeof (int) - 1)
965 * The size of a snapshot header has to fit in a size_t, and
966 * that limits the number of snapshots.
968 snap_count = le32_to_cpu(ondisk->snap_count);
969 size = SIZE_MAX - sizeof (struct ceph_snap_context);
970 if (snap_count > size / sizeof (__le64))
974 * Not only that, but the size of the entire the snapshot
975 * header must also be representable in a size_t.
977 size -= snap_count * sizeof (__le64);
978 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
985 * returns the size of an object in the image
987 static u32 rbd_obj_bytes(struct rbd_image_header *header)
989 return 1U << header->obj_order;
992 static void rbd_init_layout(struct rbd_device *rbd_dev)
994 if (rbd_dev->header.stripe_unit == 0 ||
995 rbd_dev->header.stripe_count == 0) {
996 rbd_dev->header.stripe_unit = rbd_obj_bytes(&rbd_dev->header);
997 rbd_dev->header.stripe_count = 1;
1000 rbd_dev->layout.stripe_unit = rbd_dev->header.stripe_unit;
1001 rbd_dev->layout.stripe_count = rbd_dev->header.stripe_count;
1002 rbd_dev->layout.object_size = rbd_obj_bytes(&rbd_dev->header);
1003 rbd_dev->layout.pool_id = rbd_dev->header.data_pool_id == CEPH_NOPOOL ?
1004 rbd_dev->spec->pool_id : rbd_dev->header.data_pool_id;
1005 RCU_INIT_POINTER(rbd_dev->layout.pool_ns, NULL);
1009 * Fill an rbd image header with information from the given format 1
1012 static int rbd_header_from_disk(struct rbd_device *rbd_dev,
1013 struct rbd_image_header_ondisk *ondisk)
1015 struct rbd_image_header *header = &rbd_dev->header;
1016 bool first_time = header->object_prefix == NULL;
1017 struct ceph_snap_context *snapc;
1018 char *object_prefix = NULL;
1019 char *snap_names = NULL;
1020 u64 *snap_sizes = NULL;
1025 /* Allocate this now to avoid having to handle failure below */
1028 object_prefix = kstrndup(ondisk->object_prefix,
1029 sizeof(ondisk->object_prefix),
1035 /* Allocate the snapshot context and fill it in */
1037 snap_count = le32_to_cpu(ondisk->snap_count);
1038 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
1041 snapc->seq = le64_to_cpu(ondisk->snap_seq);
1043 struct rbd_image_snap_ondisk *snaps;
1044 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
1046 /* We'll keep a copy of the snapshot names... */
1048 if (snap_names_len > (u64)SIZE_MAX)
1050 snap_names = kmalloc(snap_names_len, GFP_KERNEL);
1054 /* ...as well as the array of their sizes. */
1055 snap_sizes = kmalloc_array(snap_count,
1056 sizeof(*header->snap_sizes),
1062 * Copy the names, and fill in each snapshot's id
1065 * Note that rbd_dev_v1_header_info() guarantees the
1066 * ondisk buffer we're working with has
1067 * snap_names_len bytes beyond the end of the
1068 * snapshot id array, this memcpy() is safe.
1070 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
1071 snaps = ondisk->snaps;
1072 for (i = 0; i < snap_count; i++) {
1073 snapc->snaps[i] = le64_to_cpu(snaps[i].id);
1074 snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
1078 /* We won't fail any more, fill in the header */
1081 header->object_prefix = object_prefix;
1082 header->obj_order = ondisk->options.order;
1083 rbd_init_layout(rbd_dev);
1085 ceph_put_snap_context(header->snapc);
1086 kfree(header->snap_names);
1087 kfree(header->snap_sizes);
1090 /* The remaining fields always get updated (when we refresh) */
1092 header->image_size = le64_to_cpu(ondisk->image_size);
1093 header->snapc = snapc;
1094 header->snap_names = snap_names;
1095 header->snap_sizes = snap_sizes;
1103 ceph_put_snap_context(snapc);
1104 kfree(object_prefix);
1109 static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1111 const char *snap_name;
1113 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1115 /* Skip over names until we find the one we are looking for */
1117 snap_name = rbd_dev->header.snap_names;
1119 snap_name += strlen(snap_name) + 1;
1121 return kstrdup(snap_name, GFP_KERNEL);
1125 * Snapshot id comparison function for use with qsort()/bsearch().
1126 * Note that result is for snapshots in *descending* order.
1128 static int snapid_compare_reverse(const void *s1, const void *s2)
1130 u64 snap_id1 = *(u64 *)s1;
1131 u64 snap_id2 = *(u64 *)s2;
1133 if (snap_id1 < snap_id2)
1135 return snap_id1 == snap_id2 ? 0 : -1;
1139 * Search a snapshot context to see if the given snapshot id is
1142 * Returns the position of the snapshot id in the array if it's found,
1143 * or BAD_SNAP_INDEX otherwise.
1145 * Note: The snapshot array is in kept sorted (by the osd) in
1146 * reverse order, highest snapshot id first.
1148 static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1150 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1153 found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1154 sizeof (snap_id), snapid_compare_reverse);
1156 return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1159 static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1163 const char *snap_name;
1165 which = rbd_dev_snap_index(rbd_dev, snap_id);
1166 if (which == BAD_SNAP_INDEX)
1167 return ERR_PTR(-ENOENT);
1169 snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1170 return snap_name ? snap_name : ERR_PTR(-ENOMEM);
1173 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1175 if (snap_id == CEPH_NOSNAP)
1176 return RBD_SNAP_HEAD_NAME;
1178 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1179 if (rbd_dev->image_format == 1)
1180 return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1182 return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1185 static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1188 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1189 if (snap_id == CEPH_NOSNAP) {
1190 *snap_size = rbd_dev->header.image_size;
1191 } else if (rbd_dev->image_format == 1) {
1194 which = rbd_dev_snap_index(rbd_dev, snap_id);
1195 if (which == BAD_SNAP_INDEX)
1198 *snap_size = rbd_dev->header.snap_sizes[which];
1203 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1212 static int rbd_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
1215 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1216 if (snap_id == CEPH_NOSNAP) {
1217 *snap_features = rbd_dev->header.features;
1218 } else if (rbd_dev->image_format == 1) {
1219 *snap_features = 0; /* No features for format 1 */
1224 ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, &features);
1228 *snap_features = features;
1233 static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1235 u64 snap_id = rbd_dev->spec->snap_id;
1240 ret = rbd_snap_size(rbd_dev, snap_id, &size);
1243 ret = rbd_snap_features(rbd_dev, snap_id, &features);
1247 rbd_dev->mapping.size = size;
1248 rbd_dev->mapping.features = features;
1253 static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1255 rbd_dev->mapping.size = 0;
1256 rbd_dev->mapping.features = 0;
1259 static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset)
1261 u64 segment_size = rbd_obj_bytes(&rbd_dev->header);
1263 return offset & (segment_size - 1);
1266 static u64 rbd_segment_length(struct rbd_device *rbd_dev,
1267 u64 offset, u64 length)
1269 u64 segment_size = rbd_obj_bytes(&rbd_dev->header);
1271 offset &= segment_size - 1;
1273 rbd_assert(length <= U64_MAX - offset);
1274 if (offset + length > segment_size)
1275 length = segment_size - offset;
1284 static void bio_chain_put(struct bio *chain)
1290 chain = chain->bi_next;
1296 * zeros a bio chain, starting at specific offset
1298 static void zero_bio_chain(struct bio *chain, int start_ofs)
1301 struct bvec_iter iter;
1302 unsigned long flags;
1307 bio_for_each_segment(bv, chain, iter) {
1308 if (pos + bv.bv_len > start_ofs) {
1309 int remainder = max(start_ofs - pos, 0);
1310 buf = bvec_kmap_irq(&bv, &flags);
1311 memset(buf + remainder, 0,
1312 bv.bv_len - remainder);
1313 flush_dcache_page(bv.bv_page);
1314 bvec_kunmap_irq(buf, &flags);
1319 chain = chain->bi_next;
1324 * similar to zero_bio_chain(), zeros data defined by a page array,
1325 * starting at the given byte offset from the start of the array and
1326 * continuing up to the given end offset. The pages array is
1327 * assumed to be big enough to hold all bytes up to the end.
1329 static void zero_pages(struct page **pages, u64 offset, u64 end)
1331 struct page **page = &pages[offset >> PAGE_SHIFT];
1333 rbd_assert(end > offset);
1334 rbd_assert(end - offset <= (u64)SIZE_MAX);
1335 while (offset < end) {
1338 unsigned long flags;
1341 page_offset = offset & ~PAGE_MASK;
1342 length = min_t(size_t, PAGE_SIZE - page_offset, end - offset);
1343 local_irq_save(flags);
1344 kaddr = kmap_atomic(*page);
1345 memset(kaddr + page_offset, 0, length);
1346 flush_dcache_page(*page);
1347 kunmap_atomic(kaddr);
1348 local_irq_restore(flags);
1356 * Clone a portion of a bio, starting at the given byte offset
1357 * and continuing for the number of bytes indicated.
1359 static struct bio *bio_clone_range(struct bio *bio_src,
1360 unsigned int offset,
1366 bio = bio_clone(bio_src, gfpmask);
1368 return NULL; /* ENOMEM */
1370 bio_advance(bio, offset);
1371 bio->bi_iter.bi_size = len;
1377 * Clone a portion of a bio chain, starting at the given byte offset
1378 * into the first bio in the source chain and continuing for the
1379 * number of bytes indicated. The result is another bio chain of
1380 * exactly the given length, or a null pointer on error.
1382 * The bio_src and offset parameters are both in-out. On entry they
1383 * refer to the first source bio and the offset into that bio where
1384 * the start of data to be cloned is located.
1386 * On return, bio_src is updated to refer to the bio in the source
1387 * chain that contains first un-cloned byte, and *offset will
1388 * contain the offset of that byte within that bio.
1390 static struct bio *bio_chain_clone_range(struct bio **bio_src,
1391 unsigned int *offset,
1395 struct bio *bi = *bio_src;
1396 unsigned int off = *offset;
1397 struct bio *chain = NULL;
1400 /* Build up a chain of clone bios up to the limit */
1402 if (!bi || off >= bi->bi_iter.bi_size || !len)
1403 return NULL; /* Nothing to clone */
1407 unsigned int bi_size;
1411 rbd_warn(NULL, "bio_chain exhausted with %u left", len);
1412 goto out_err; /* EINVAL; ran out of bio's */
1414 bi_size = min_t(unsigned int, bi->bi_iter.bi_size - off, len);
1415 bio = bio_clone_range(bi, off, bi_size, gfpmask);
1417 goto out_err; /* ENOMEM */
1420 end = &bio->bi_next;
1423 if (off == bi->bi_iter.bi_size) {
1434 bio_chain_put(chain);
1440 * The default/initial value for all object request flags is 0. For
1441 * each flag, once its value is set to 1 it is never reset to 0
1444 static void obj_request_img_data_set(struct rbd_obj_request *obj_request)
1446 if (test_and_set_bit(OBJ_REQ_IMG_DATA, &obj_request->flags)) {
1447 struct rbd_device *rbd_dev;
1449 rbd_dev = obj_request->img_request->rbd_dev;
1450 rbd_warn(rbd_dev, "obj_request %p already marked img_data",
1455 static bool obj_request_img_data_test(struct rbd_obj_request *obj_request)
1458 return test_bit(OBJ_REQ_IMG_DATA, &obj_request->flags) != 0;
1461 static void obj_request_done_set(struct rbd_obj_request *obj_request)
1463 if (test_and_set_bit(OBJ_REQ_DONE, &obj_request->flags)) {
1464 struct rbd_device *rbd_dev = NULL;
1466 if (obj_request_img_data_test(obj_request))
1467 rbd_dev = obj_request->img_request->rbd_dev;
1468 rbd_warn(rbd_dev, "obj_request %p already marked done",
1473 static bool obj_request_done_test(struct rbd_obj_request *obj_request)
1476 return test_bit(OBJ_REQ_DONE, &obj_request->flags) != 0;
1480 * This sets the KNOWN flag after (possibly) setting the EXISTS
1481 * flag. The latter is set based on the "exists" value provided.
1483 * Note that for our purposes once an object exists it never goes
1484 * away again. It's possible that the response from two existence
1485 * checks are separated by the creation of the target object, and
1486 * the first ("doesn't exist") response arrives *after* the second
1487 * ("does exist"). In that case we ignore the second one.
1489 static void obj_request_existence_set(struct rbd_obj_request *obj_request,
1493 set_bit(OBJ_REQ_EXISTS, &obj_request->flags);
1494 set_bit(OBJ_REQ_KNOWN, &obj_request->flags);
1498 static bool obj_request_known_test(struct rbd_obj_request *obj_request)
1501 return test_bit(OBJ_REQ_KNOWN, &obj_request->flags) != 0;
1504 static bool obj_request_exists_test(struct rbd_obj_request *obj_request)
1507 return test_bit(OBJ_REQ_EXISTS, &obj_request->flags) != 0;
1510 static bool obj_request_overlaps_parent(struct rbd_obj_request *obj_request)
1512 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1514 return obj_request->img_offset <
1515 round_up(rbd_dev->parent_overlap, rbd_obj_bytes(&rbd_dev->header));
1518 static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
1520 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1521 kref_read(&obj_request->kref));
1522 kref_get(&obj_request->kref);
1525 static void rbd_obj_request_destroy(struct kref *kref);
1526 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1528 rbd_assert(obj_request != NULL);
1529 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1530 kref_read(&obj_request->kref));
1531 kref_put(&obj_request->kref, rbd_obj_request_destroy);
1534 static void rbd_img_request_get(struct rbd_img_request *img_request)
1536 dout("%s: img %p (was %d)\n", __func__, img_request,
1537 kref_read(&img_request->kref));
1538 kref_get(&img_request->kref);
1541 static bool img_request_child_test(struct rbd_img_request *img_request);
1542 static void rbd_parent_request_destroy(struct kref *kref);
1543 static void rbd_img_request_destroy(struct kref *kref);
1544 static void rbd_img_request_put(struct rbd_img_request *img_request)
1546 rbd_assert(img_request != NULL);
1547 dout("%s: img %p (was %d)\n", __func__, img_request,
1548 kref_read(&img_request->kref));
1549 if (img_request_child_test(img_request))
1550 kref_put(&img_request->kref, rbd_parent_request_destroy);
1552 kref_put(&img_request->kref, rbd_img_request_destroy);
1555 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1556 struct rbd_obj_request *obj_request)
1558 rbd_assert(obj_request->img_request == NULL);
1560 /* Image request now owns object's original reference */
1561 obj_request->img_request = img_request;
1562 obj_request->which = img_request->obj_request_count;
1563 rbd_assert(!obj_request_img_data_test(obj_request));
1564 obj_request_img_data_set(obj_request);
1565 rbd_assert(obj_request->which != BAD_WHICH);
1566 img_request->obj_request_count++;
1567 list_add_tail(&obj_request->links, &img_request->obj_requests);
1568 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1569 obj_request->which);
1572 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1573 struct rbd_obj_request *obj_request)
1575 rbd_assert(obj_request->which != BAD_WHICH);
1577 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1578 obj_request->which);
1579 list_del(&obj_request->links);
1580 rbd_assert(img_request->obj_request_count > 0);
1581 img_request->obj_request_count--;
1582 rbd_assert(obj_request->which == img_request->obj_request_count);
1583 obj_request->which = BAD_WHICH;
1584 rbd_assert(obj_request_img_data_test(obj_request));
1585 rbd_assert(obj_request->img_request == img_request);
1586 obj_request->img_request = NULL;
1587 obj_request->callback = NULL;
1588 rbd_obj_request_put(obj_request);
1591 static bool obj_request_type_valid(enum obj_request_type type)
1594 case OBJ_REQUEST_NODATA:
1595 case OBJ_REQUEST_BIO:
1596 case OBJ_REQUEST_PAGES:
1603 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request);
1605 static void rbd_obj_request_submit(struct rbd_obj_request *obj_request)
1607 struct ceph_osd_request *osd_req = obj_request->osd_req;
1609 dout("%s %p object_no %016llx %llu~%llu osd_req %p\n", __func__,
1610 obj_request, obj_request->object_no, obj_request->offset,
1611 obj_request->length, osd_req);
1612 if (obj_request_img_data_test(obj_request)) {
1613 WARN_ON(obj_request->callback != rbd_img_obj_callback);
1614 rbd_img_request_get(obj_request->img_request);
1616 ceph_osdc_start_request(osd_req->r_osdc, osd_req, false);
1619 static void rbd_img_request_complete(struct rbd_img_request *img_request)
1622 dout("%s: img %p\n", __func__, img_request);
1625 * If no error occurred, compute the aggregate transfer
1626 * count for the image request. We could instead use
1627 * atomic64_cmpxchg() to update it as each object request
1628 * completes; not clear which way is better off hand.
1630 if (!img_request->result) {
1631 struct rbd_obj_request *obj_request;
1634 for_each_obj_request(img_request, obj_request)
1635 xferred += obj_request->xferred;
1636 img_request->xferred = xferred;
1639 if (img_request->callback)
1640 img_request->callback(img_request);
1642 rbd_img_request_put(img_request);
1646 * The default/initial value for all image request flags is 0. Each
1647 * is conditionally set to 1 at image request initialization time
1648 * and currently never change thereafter.
1650 static void img_request_write_set(struct rbd_img_request *img_request)
1652 set_bit(IMG_REQ_WRITE, &img_request->flags);
1656 static bool img_request_write_test(struct rbd_img_request *img_request)
1659 return test_bit(IMG_REQ_WRITE, &img_request->flags) != 0;
1663 * Set the discard flag when the img_request is an discard request
1665 static void img_request_discard_set(struct rbd_img_request *img_request)
1667 set_bit(IMG_REQ_DISCARD, &img_request->flags);
1671 static bool img_request_discard_test(struct rbd_img_request *img_request)
1674 return test_bit(IMG_REQ_DISCARD, &img_request->flags) != 0;
1677 static void img_request_child_set(struct rbd_img_request *img_request)
1679 set_bit(IMG_REQ_CHILD, &img_request->flags);
1683 static void img_request_child_clear(struct rbd_img_request *img_request)
1685 clear_bit(IMG_REQ_CHILD, &img_request->flags);
1689 static bool img_request_child_test(struct rbd_img_request *img_request)
1692 return test_bit(IMG_REQ_CHILD, &img_request->flags) != 0;
1695 static void img_request_layered_set(struct rbd_img_request *img_request)
1697 set_bit(IMG_REQ_LAYERED, &img_request->flags);
1701 static void img_request_layered_clear(struct rbd_img_request *img_request)
1703 clear_bit(IMG_REQ_LAYERED, &img_request->flags);
1707 static bool img_request_layered_test(struct rbd_img_request *img_request)
1710 return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1713 static enum obj_operation_type
1714 rbd_img_request_op_type(struct rbd_img_request *img_request)
1716 if (img_request_write_test(img_request))
1717 return OBJ_OP_WRITE;
1718 else if (img_request_discard_test(img_request))
1719 return OBJ_OP_DISCARD;
1725 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
1727 u64 xferred = obj_request->xferred;
1728 u64 length = obj_request->length;
1730 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1731 obj_request, obj_request->img_request, obj_request->result,
1734 * ENOENT means a hole in the image. We zero-fill the entire
1735 * length of the request. A short read also implies zero-fill
1736 * to the end of the request. An error requires the whole
1737 * length of the request to be reported finished with an error
1738 * to the block layer. In each case we update the xferred
1739 * count to indicate the whole request was satisfied.
1741 rbd_assert(obj_request->type != OBJ_REQUEST_NODATA);
1742 if (obj_request->result == -ENOENT) {
1743 if (obj_request->type == OBJ_REQUEST_BIO)
1744 zero_bio_chain(obj_request->bio_list, 0);
1746 zero_pages(obj_request->pages, 0, length);
1747 obj_request->result = 0;
1748 } else if (xferred < length && !obj_request->result) {
1749 if (obj_request->type == OBJ_REQUEST_BIO)
1750 zero_bio_chain(obj_request->bio_list, xferred);
1752 zero_pages(obj_request->pages, xferred, length);
1754 obj_request->xferred = length;
1755 obj_request_done_set(obj_request);
1758 static void rbd_obj_request_complete(struct rbd_obj_request *obj_request)
1760 dout("%s: obj %p cb %p\n", __func__, obj_request,
1761 obj_request->callback);
1762 if (obj_request->callback)
1763 obj_request->callback(obj_request);
1765 complete_all(&obj_request->completion);
1768 static void rbd_obj_request_error(struct rbd_obj_request *obj_request, int err)
1770 obj_request->result = err;
1771 obj_request->xferred = 0;
1773 * kludge - mirror rbd_obj_request_submit() to match a put in
1774 * rbd_img_obj_callback()
1776 if (obj_request_img_data_test(obj_request)) {
1777 WARN_ON(obj_request->callback != rbd_img_obj_callback);
1778 rbd_img_request_get(obj_request->img_request);
1780 obj_request_done_set(obj_request);
1781 rbd_obj_request_complete(obj_request);
1784 static void rbd_osd_read_callback(struct rbd_obj_request *obj_request)
1786 struct rbd_img_request *img_request = NULL;
1787 struct rbd_device *rbd_dev = NULL;
1788 bool layered = false;
1790 if (obj_request_img_data_test(obj_request)) {
1791 img_request = obj_request->img_request;
1792 layered = img_request && img_request_layered_test(img_request);
1793 rbd_dev = img_request->rbd_dev;
1796 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1797 obj_request, img_request, obj_request->result,
1798 obj_request->xferred, obj_request->length);
1799 if (layered && obj_request->result == -ENOENT &&
1800 obj_request->img_offset < rbd_dev->parent_overlap)
1801 rbd_img_parent_read(obj_request);
1802 else if (img_request)
1803 rbd_img_obj_request_read_callback(obj_request);
1805 obj_request_done_set(obj_request);
1808 static void rbd_osd_write_callback(struct rbd_obj_request *obj_request)
1810 dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1811 obj_request->result, obj_request->length);
1813 * There is no such thing as a successful short write. Set
1814 * it to our originally-requested length.
1816 obj_request->xferred = obj_request->length;
1817 obj_request_done_set(obj_request);
1820 static void rbd_osd_discard_callback(struct rbd_obj_request *obj_request)
1822 dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1823 obj_request->result, obj_request->length);
1825 * There is no such thing as a successful short discard. Set
1826 * it to our originally-requested length.
1828 obj_request->xferred = obj_request->length;
1829 /* discarding a non-existent object is not a problem */
1830 if (obj_request->result == -ENOENT)
1831 obj_request->result = 0;
1832 obj_request_done_set(obj_request);
1836 * For a simple stat call there's nothing to do. We'll do more if
1837 * this is part of a write sequence for a layered image.
1839 static void rbd_osd_stat_callback(struct rbd_obj_request *obj_request)
1841 dout("%s: obj %p\n", __func__, obj_request);
1842 obj_request_done_set(obj_request);
1845 static void rbd_osd_call_callback(struct rbd_obj_request *obj_request)
1847 dout("%s: obj %p\n", __func__, obj_request);
1849 if (obj_request_img_data_test(obj_request))
1850 rbd_osd_copyup_callback(obj_request);
1852 obj_request_done_set(obj_request);
1855 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
1857 struct rbd_obj_request *obj_request = osd_req->r_priv;
1860 dout("%s: osd_req %p\n", __func__, osd_req);
1861 rbd_assert(osd_req == obj_request->osd_req);
1862 if (obj_request_img_data_test(obj_request)) {
1863 rbd_assert(obj_request->img_request);
1864 rbd_assert(obj_request->which != BAD_WHICH);
1866 rbd_assert(obj_request->which == BAD_WHICH);
1869 if (osd_req->r_result < 0)
1870 obj_request->result = osd_req->r_result;
1873 * We support a 64-bit length, but ultimately it has to be
1874 * passed to the block layer, which just supports a 32-bit
1877 obj_request->xferred = osd_req->r_ops[0].outdata_len;
1878 rbd_assert(obj_request->xferred < (u64)UINT_MAX);
1880 opcode = osd_req->r_ops[0].op;
1882 case CEPH_OSD_OP_READ:
1883 rbd_osd_read_callback(obj_request);
1885 case CEPH_OSD_OP_SETALLOCHINT:
1886 rbd_assert(osd_req->r_ops[1].op == CEPH_OSD_OP_WRITE ||
1887 osd_req->r_ops[1].op == CEPH_OSD_OP_WRITEFULL);
1889 case CEPH_OSD_OP_WRITE:
1890 case CEPH_OSD_OP_WRITEFULL:
1891 rbd_osd_write_callback(obj_request);
1893 case CEPH_OSD_OP_STAT:
1894 rbd_osd_stat_callback(obj_request);
1896 case CEPH_OSD_OP_DELETE:
1897 case CEPH_OSD_OP_TRUNCATE:
1898 case CEPH_OSD_OP_ZERO:
1899 rbd_osd_discard_callback(obj_request);
1901 case CEPH_OSD_OP_CALL:
1902 rbd_osd_call_callback(obj_request);
1905 rbd_warn(NULL, "unexpected OSD op: object_no %016llx opcode %d",
1906 obj_request->object_no, opcode);
1910 if (obj_request_done_test(obj_request))
1911 rbd_obj_request_complete(obj_request);
1914 static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request)
1916 struct ceph_osd_request *osd_req = obj_request->osd_req;
1918 rbd_assert(obj_request_img_data_test(obj_request));
1919 osd_req->r_snapid = obj_request->img_request->snap_id;
1922 static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request)
1924 struct ceph_osd_request *osd_req = obj_request->osd_req;
1926 ktime_get_real_ts(&osd_req->r_mtime);
1927 osd_req->r_data_offset = obj_request->offset;
1930 static struct ceph_osd_request *
1931 __rbd_osd_req_create(struct rbd_device *rbd_dev,
1932 struct ceph_snap_context *snapc,
1933 int num_ops, unsigned int flags,
1934 struct rbd_obj_request *obj_request)
1936 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1937 struct ceph_osd_request *req;
1938 const char *name_format = rbd_dev->image_format == 1 ?
1939 RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
1941 req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, GFP_NOIO);
1945 req->r_flags = flags;
1946 req->r_callback = rbd_osd_req_callback;
1947 req->r_priv = obj_request;
1949 req->r_base_oloc.pool = rbd_dev->layout.pool_id;
1950 if (ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
1951 rbd_dev->header.object_prefix, obj_request->object_no))
1954 if (ceph_osdc_alloc_messages(req, GFP_NOIO))
1960 ceph_osdc_put_request(req);
1965 * Create an osd request. A read request has one osd op (read).
1966 * A write request has either one (watch) or two (hint+write) osd ops.
1967 * (All rbd data writes are prefixed with an allocation hint op, but
1968 * technically osd watch is a write request, hence this distinction.)
1970 static struct ceph_osd_request *rbd_osd_req_create(
1971 struct rbd_device *rbd_dev,
1972 enum obj_operation_type op_type,
1973 unsigned int num_ops,
1974 struct rbd_obj_request *obj_request)
1976 struct ceph_snap_context *snapc = NULL;
1978 if (obj_request_img_data_test(obj_request) &&
1979 (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_WRITE)) {
1980 struct rbd_img_request *img_request = obj_request->img_request;
1981 if (op_type == OBJ_OP_WRITE) {
1982 rbd_assert(img_request_write_test(img_request));
1984 rbd_assert(img_request_discard_test(img_request));
1986 snapc = img_request->snapc;
1989 rbd_assert(num_ops == 1 || ((op_type == OBJ_OP_WRITE) && num_ops == 2));
1991 return __rbd_osd_req_create(rbd_dev, snapc, num_ops,
1992 (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD) ?
1993 CEPH_OSD_FLAG_WRITE : CEPH_OSD_FLAG_READ, obj_request);
1997 * Create a copyup osd request based on the information in the object
1998 * request supplied. A copyup request has two or three osd ops, a
1999 * copyup method call, potentially a hint op, and a write or truncate
2002 static struct ceph_osd_request *
2003 rbd_osd_req_create_copyup(struct rbd_obj_request *obj_request)
2005 struct rbd_img_request *img_request;
2006 int num_osd_ops = 3;
2008 rbd_assert(obj_request_img_data_test(obj_request));
2009 img_request = obj_request->img_request;
2010 rbd_assert(img_request);
2011 rbd_assert(img_request_write_test(img_request) ||
2012 img_request_discard_test(img_request));
2014 if (img_request_discard_test(img_request))
2017 return __rbd_osd_req_create(img_request->rbd_dev,
2018 img_request->snapc, num_osd_ops,
2019 CEPH_OSD_FLAG_WRITE, obj_request);
2022 static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req)
2024 ceph_osdc_put_request(osd_req);
2027 static struct rbd_obj_request *
2028 rbd_obj_request_create(enum obj_request_type type)
2030 struct rbd_obj_request *obj_request;
2032 rbd_assert(obj_request_type_valid(type));
2034 obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_NOIO);
2038 obj_request->which = BAD_WHICH;
2039 obj_request->type = type;
2040 INIT_LIST_HEAD(&obj_request->links);
2041 init_completion(&obj_request->completion);
2042 kref_init(&obj_request->kref);
2044 dout("%s %p\n", __func__, obj_request);
2048 static void rbd_obj_request_destroy(struct kref *kref)
2050 struct rbd_obj_request *obj_request;
2052 obj_request = container_of(kref, struct rbd_obj_request, kref);
2054 dout("%s: obj %p\n", __func__, obj_request);
2056 rbd_assert(obj_request->img_request == NULL);
2057 rbd_assert(obj_request->which == BAD_WHICH);
2059 if (obj_request->osd_req)
2060 rbd_osd_req_destroy(obj_request->osd_req);
2062 rbd_assert(obj_request_type_valid(obj_request->type));
2063 switch (obj_request->type) {
2064 case OBJ_REQUEST_NODATA:
2065 break; /* Nothing to do */
2066 case OBJ_REQUEST_BIO:
2067 if (obj_request->bio_list)
2068 bio_chain_put(obj_request->bio_list);
2070 case OBJ_REQUEST_PAGES:
2071 /* img_data requests don't own their page array */
2072 if (obj_request->pages &&
2073 !obj_request_img_data_test(obj_request))
2074 ceph_release_page_vector(obj_request->pages,
2075 obj_request->page_count);
2079 kmem_cache_free(rbd_obj_request_cache, obj_request);
2082 /* It's OK to call this for a device with no parent */
2084 static void rbd_spec_put(struct rbd_spec *spec);
2085 static void rbd_dev_unparent(struct rbd_device *rbd_dev)
2087 rbd_dev_remove_parent(rbd_dev);
2088 rbd_spec_put(rbd_dev->parent_spec);
2089 rbd_dev->parent_spec = NULL;
2090 rbd_dev->parent_overlap = 0;
2094 * Parent image reference counting is used to determine when an
2095 * image's parent fields can be safely torn down--after there are no
2096 * more in-flight requests to the parent image. When the last
2097 * reference is dropped, cleaning them up is safe.
2099 static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
2103 if (!rbd_dev->parent_spec)
2106 counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
2110 /* Last reference; clean up parent data structures */
2113 rbd_dev_unparent(rbd_dev);
2115 rbd_warn(rbd_dev, "parent reference underflow");
2119 * If an image has a non-zero parent overlap, get a reference to its
2122 * Returns true if the rbd device has a parent with a non-zero
2123 * overlap and a reference for it was successfully taken, or
2126 static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
2130 if (!rbd_dev->parent_spec)
2133 down_read(&rbd_dev->header_rwsem);
2134 if (rbd_dev->parent_overlap)
2135 counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
2136 up_read(&rbd_dev->header_rwsem);
2139 rbd_warn(rbd_dev, "parent reference overflow");
2145 * Caller is responsible for filling in the list of object requests
2146 * that comprises the image request, and the Linux request pointer
2147 * (if there is one).
2149 static struct rbd_img_request *rbd_img_request_create(
2150 struct rbd_device *rbd_dev,
2151 u64 offset, u64 length,
2152 enum obj_operation_type op_type,
2153 struct ceph_snap_context *snapc)
2155 struct rbd_img_request *img_request;
2157 img_request = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
2161 img_request->rq = NULL;
2162 img_request->rbd_dev = rbd_dev;
2163 img_request->offset = offset;
2164 img_request->length = length;
2165 img_request->flags = 0;
2166 if (op_type == OBJ_OP_DISCARD) {
2167 img_request_discard_set(img_request);
2168 img_request->snapc = snapc;
2169 } else if (op_type == OBJ_OP_WRITE) {
2170 img_request_write_set(img_request);
2171 img_request->snapc = snapc;
2173 img_request->snap_id = rbd_dev->spec->snap_id;
2175 if (rbd_dev_parent_get(rbd_dev))
2176 img_request_layered_set(img_request);
2177 spin_lock_init(&img_request->completion_lock);
2178 img_request->next_completion = 0;
2179 img_request->callback = NULL;
2180 img_request->result = 0;
2181 img_request->obj_request_count = 0;
2182 INIT_LIST_HEAD(&img_request->obj_requests);
2183 kref_init(&img_request->kref);
2185 dout("%s: rbd_dev %p %s %llu/%llu -> img %p\n", __func__, rbd_dev,
2186 obj_op_name(op_type), offset, length, img_request);
2191 static void rbd_img_request_destroy(struct kref *kref)
2193 struct rbd_img_request *img_request;
2194 struct rbd_obj_request *obj_request;
2195 struct rbd_obj_request *next_obj_request;
2197 img_request = container_of(kref, struct rbd_img_request, kref);
2199 dout("%s: img %p\n", __func__, img_request);
2201 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
2202 rbd_img_obj_request_del(img_request, obj_request);
2203 rbd_assert(img_request->obj_request_count == 0);
2205 if (img_request_layered_test(img_request)) {
2206 img_request_layered_clear(img_request);
2207 rbd_dev_parent_put(img_request->rbd_dev);
2210 if (img_request_write_test(img_request) ||
2211 img_request_discard_test(img_request))
2212 ceph_put_snap_context(img_request->snapc);
2214 kmem_cache_free(rbd_img_request_cache, img_request);
2217 static struct rbd_img_request *rbd_parent_request_create(
2218 struct rbd_obj_request *obj_request,
2219 u64 img_offset, u64 length)
2221 struct rbd_img_request *parent_request;
2222 struct rbd_device *rbd_dev;
2224 rbd_assert(obj_request->img_request);
2225 rbd_dev = obj_request->img_request->rbd_dev;
2227 parent_request = rbd_img_request_create(rbd_dev->parent, img_offset,
2228 length, OBJ_OP_READ, NULL);
2229 if (!parent_request)
2232 img_request_child_set(parent_request);
2233 rbd_obj_request_get(obj_request);
2234 parent_request->obj_request = obj_request;
2236 return parent_request;
2239 static void rbd_parent_request_destroy(struct kref *kref)
2241 struct rbd_img_request *parent_request;
2242 struct rbd_obj_request *orig_request;
2244 parent_request = container_of(kref, struct rbd_img_request, kref);
2245 orig_request = parent_request->obj_request;
2247 parent_request->obj_request = NULL;
2248 rbd_obj_request_put(orig_request);
2249 img_request_child_clear(parent_request);
2251 rbd_img_request_destroy(kref);
2254 static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request)
2256 struct rbd_img_request *img_request;
2257 unsigned int xferred;
2261 rbd_assert(obj_request_img_data_test(obj_request));
2262 img_request = obj_request->img_request;
2264 rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
2265 xferred = (unsigned int)obj_request->xferred;
2266 result = obj_request->result;
2268 struct rbd_device *rbd_dev = img_request->rbd_dev;
2269 enum obj_operation_type op_type;
2271 if (img_request_discard_test(img_request))
2272 op_type = OBJ_OP_DISCARD;
2273 else if (img_request_write_test(img_request))
2274 op_type = OBJ_OP_WRITE;
2276 op_type = OBJ_OP_READ;
2278 rbd_warn(rbd_dev, "%s %llx at %llx (%llx)",
2279 obj_op_name(op_type), obj_request->length,
2280 obj_request->img_offset, obj_request->offset);
2281 rbd_warn(rbd_dev, " result %d xferred %x",
2283 if (!img_request->result)
2284 img_request->result = result;
2286 * Need to end I/O on the entire obj_request worth of
2287 * bytes in case of error.
2289 xferred = obj_request->length;
2292 if (img_request_child_test(img_request)) {
2293 rbd_assert(img_request->obj_request != NULL);
2294 more = obj_request->which < img_request->obj_request_count - 1;
2296 blk_status_t status = errno_to_blk_status(result);
2298 rbd_assert(img_request->rq != NULL);
2300 more = blk_update_request(img_request->rq, status, xferred);
2302 __blk_mq_end_request(img_request->rq, status);
2308 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request)
2310 struct rbd_img_request *img_request;
2311 u32 which = obj_request->which;
2314 rbd_assert(obj_request_img_data_test(obj_request));
2315 img_request = obj_request->img_request;
2317 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
2318 rbd_assert(img_request != NULL);
2319 rbd_assert(img_request->obj_request_count > 0);
2320 rbd_assert(which != BAD_WHICH);
2321 rbd_assert(which < img_request->obj_request_count);
2323 spin_lock_irq(&img_request->completion_lock);
2324 if (which != img_request->next_completion)
2327 for_each_obj_request_from(img_request, obj_request) {
2329 rbd_assert(which < img_request->obj_request_count);
2331 if (!obj_request_done_test(obj_request))
2333 more = rbd_img_obj_end_request(obj_request);
2337 rbd_assert(more ^ (which == img_request->obj_request_count));
2338 img_request->next_completion = which;
2340 spin_unlock_irq(&img_request->completion_lock);
2341 rbd_img_request_put(img_request);
2344 rbd_img_request_complete(img_request);
2348 * Add individual osd ops to the given ceph_osd_request and prepare
2349 * them for submission. num_ops is the current number of
2350 * osd operations already to the object request.
2352 static void rbd_img_obj_request_fill(struct rbd_obj_request *obj_request,
2353 struct ceph_osd_request *osd_request,
2354 enum obj_operation_type op_type,
2355 unsigned int num_ops)
2357 struct rbd_img_request *img_request = obj_request->img_request;
2358 struct rbd_device *rbd_dev = img_request->rbd_dev;
2359 u64 object_size = rbd_obj_bytes(&rbd_dev->header);
2360 u64 offset = obj_request->offset;
2361 u64 length = obj_request->length;
2365 if (op_type == OBJ_OP_DISCARD) {
2366 if (!offset && length == object_size &&
2367 (!img_request_layered_test(img_request) ||
2368 !obj_request_overlaps_parent(obj_request))) {
2369 opcode = CEPH_OSD_OP_DELETE;
2370 } else if ((offset + length == object_size)) {
2371 opcode = CEPH_OSD_OP_TRUNCATE;
2373 down_read(&rbd_dev->header_rwsem);
2374 img_end = rbd_dev->header.image_size;
2375 up_read(&rbd_dev->header_rwsem);
2377 if (obj_request->img_offset + length == img_end)
2378 opcode = CEPH_OSD_OP_TRUNCATE;
2380 opcode = CEPH_OSD_OP_ZERO;
2382 } else if (op_type == OBJ_OP_WRITE) {
2383 if (!offset && length == object_size)
2384 opcode = CEPH_OSD_OP_WRITEFULL;
2386 opcode = CEPH_OSD_OP_WRITE;
2387 osd_req_op_alloc_hint_init(osd_request, num_ops,
2388 object_size, object_size);
2391 opcode = CEPH_OSD_OP_READ;
2394 if (opcode == CEPH_OSD_OP_DELETE)
2395 osd_req_op_init(osd_request, num_ops, opcode, 0);
2397 osd_req_op_extent_init(osd_request, num_ops, opcode,
2398 offset, length, 0, 0);
2400 if (obj_request->type == OBJ_REQUEST_BIO)
2401 osd_req_op_extent_osd_data_bio(osd_request, num_ops,
2402 obj_request->bio_list, length);
2403 else if (obj_request->type == OBJ_REQUEST_PAGES)
2404 osd_req_op_extent_osd_data_pages(osd_request, num_ops,
2405 obj_request->pages, length,
2406 offset & ~PAGE_MASK, false, false);
2408 /* Discards are also writes */
2409 if (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD)
2410 rbd_osd_req_format_write(obj_request);
2412 rbd_osd_req_format_read(obj_request);
2416 * Split up an image request into one or more object requests, each
2417 * to a different object. The "type" parameter indicates whether
2418 * "data_desc" is the pointer to the head of a list of bio
2419 * structures, or the base of a page array. In either case this
2420 * function assumes data_desc describes memory sufficient to hold
2421 * all data described by the image request.
2423 static int rbd_img_request_fill(struct rbd_img_request *img_request,
2424 enum obj_request_type type,
2427 struct rbd_device *rbd_dev = img_request->rbd_dev;
2428 struct rbd_obj_request *obj_request = NULL;
2429 struct rbd_obj_request *next_obj_request;
2430 struct bio *bio_list = NULL;
2431 unsigned int bio_offset = 0;
2432 struct page **pages = NULL;
2433 enum obj_operation_type op_type;
2437 dout("%s: img %p type %d data_desc %p\n", __func__, img_request,
2438 (int)type, data_desc);
2440 img_offset = img_request->offset;
2441 resid = img_request->length;
2442 rbd_assert(resid > 0);
2443 op_type = rbd_img_request_op_type(img_request);
2445 if (type == OBJ_REQUEST_BIO) {
2446 bio_list = data_desc;
2447 rbd_assert(img_offset ==
2448 bio_list->bi_iter.bi_sector << SECTOR_SHIFT);
2449 } else if (type == OBJ_REQUEST_PAGES) {
2454 struct ceph_osd_request *osd_req;
2455 u64 object_no = img_offset >> rbd_dev->header.obj_order;
2456 u64 offset = rbd_segment_offset(rbd_dev, img_offset);
2457 u64 length = rbd_segment_length(rbd_dev, img_offset, resid);
2459 obj_request = rbd_obj_request_create(type);
2463 obj_request->object_no = object_no;
2464 obj_request->offset = offset;
2465 obj_request->length = length;
2468 * set obj_request->img_request before creating the
2469 * osd_request so that it gets the right snapc
2471 rbd_img_obj_request_add(img_request, obj_request);
2473 if (type == OBJ_REQUEST_BIO) {
2474 unsigned int clone_size;
2476 rbd_assert(length <= (u64)UINT_MAX);
2477 clone_size = (unsigned int)length;
2478 obj_request->bio_list =
2479 bio_chain_clone_range(&bio_list,
2483 if (!obj_request->bio_list)
2485 } else if (type == OBJ_REQUEST_PAGES) {
2486 unsigned int page_count;
2488 obj_request->pages = pages;
2489 page_count = (u32)calc_pages_for(offset, length);
2490 obj_request->page_count = page_count;
2491 if ((offset + length) & ~PAGE_MASK)
2492 page_count--; /* more on last page */
2493 pages += page_count;
2496 osd_req = rbd_osd_req_create(rbd_dev, op_type,
2497 (op_type == OBJ_OP_WRITE) ? 2 : 1,
2502 obj_request->osd_req = osd_req;
2503 obj_request->callback = rbd_img_obj_callback;
2504 obj_request->img_offset = img_offset;
2506 rbd_img_obj_request_fill(obj_request, osd_req, op_type, 0);
2508 img_offset += length;
2515 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
2516 rbd_img_obj_request_del(img_request, obj_request);
2522 rbd_osd_copyup_callback(struct rbd_obj_request *obj_request)
2524 struct rbd_img_request *img_request;
2525 struct rbd_device *rbd_dev;
2526 struct page **pages;
2529 dout("%s: obj %p\n", __func__, obj_request);
2531 rbd_assert(obj_request->type == OBJ_REQUEST_BIO ||
2532 obj_request->type == OBJ_REQUEST_NODATA);
2533 rbd_assert(obj_request_img_data_test(obj_request));
2534 img_request = obj_request->img_request;
2535 rbd_assert(img_request);
2537 rbd_dev = img_request->rbd_dev;
2538 rbd_assert(rbd_dev);
2540 pages = obj_request->copyup_pages;
2541 rbd_assert(pages != NULL);
2542 obj_request->copyup_pages = NULL;
2543 page_count = obj_request->copyup_page_count;
2544 rbd_assert(page_count);
2545 obj_request->copyup_page_count = 0;
2546 ceph_release_page_vector(pages, page_count);
2549 * We want the transfer count to reflect the size of the
2550 * original write request. There is no such thing as a
2551 * successful short write, so if the request was successful
2552 * we can just set it to the originally-requested length.
2554 if (!obj_request->result)
2555 obj_request->xferred = obj_request->length;
2557 obj_request_done_set(obj_request);
2561 rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request)
2563 struct rbd_obj_request *orig_request;
2564 struct ceph_osd_request *osd_req;
2565 struct rbd_device *rbd_dev;
2566 struct page **pages;
2567 enum obj_operation_type op_type;
2572 rbd_assert(img_request_child_test(img_request));
2574 /* First get what we need from the image request */
2576 pages = img_request->copyup_pages;
2577 rbd_assert(pages != NULL);
2578 img_request->copyup_pages = NULL;
2579 page_count = img_request->copyup_page_count;
2580 rbd_assert(page_count);
2581 img_request->copyup_page_count = 0;
2583 orig_request = img_request->obj_request;
2584 rbd_assert(orig_request != NULL);
2585 rbd_assert(obj_request_type_valid(orig_request->type));
2586 img_result = img_request->result;
2587 parent_length = img_request->length;
2588 rbd_assert(img_result || parent_length == img_request->xferred);
2589 rbd_img_request_put(img_request);
2591 rbd_assert(orig_request->img_request);
2592 rbd_dev = orig_request->img_request->rbd_dev;
2593 rbd_assert(rbd_dev);
2596 * If the overlap has become 0 (most likely because the
2597 * image has been flattened) we need to free the pages
2598 * and re-submit the original write request.
2600 if (!rbd_dev->parent_overlap) {
2601 ceph_release_page_vector(pages, page_count);
2602 rbd_obj_request_submit(orig_request);
2610 * The original osd request is of no use to use any more.
2611 * We need a new one that can hold the three ops in a copyup
2612 * request. Allocate the new copyup osd request for the
2613 * original request, and release the old one.
2615 img_result = -ENOMEM;
2616 osd_req = rbd_osd_req_create_copyup(orig_request);
2619 rbd_osd_req_destroy(orig_request->osd_req);
2620 orig_request->osd_req = osd_req;
2621 orig_request->copyup_pages = pages;
2622 orig_request->copyup_page_count = page_count;
2624 /* Initialize the copyup op */
2626 osd_req_op_cls_init(osd_req, 0, CEPH_OSD_OP_CALL, "rbd", "copyup");
2627 osd_req_op_cls_request_data_pages(osd_req, 0, pages, parent_length, 0,
2630 /* Add the other op(s) */
2632 op_type = rbd_img_request_op_type(orig_request->img_request);
2633 rbd_img_obj_request_fill(orig_request, osd_req, op_type, 1);
2635 /* All set, send it off. */
2637 rbd_obj_request_submit(orig_request);
2641 ceph_release_page_vector(pages, page_count);
2642 rbd_obj_request_error(orig_request, img_result);
2646 * Read from the parent image the range of data that covers the
2647 * entire target of the given object request. This is used for
2648 * satisfying a layered image write request when the target of an
2649 * object request from the image request does not exist.
2651 * A page array big enough to hold the returned data is allocated
2652 * and supplied to rbd_img_request_fill() as the "data descriptor."
2653 * When the read completes, this page array will be transferred to
2654 * the original object request for the copyup operation.
2656 * If an error occurs, it is recorded as the result of the original
2657 * object request in rbd_img_obj_exists_callback().
2659 static int rbd_img_obj_parent_read_full(struct rbd_obj_request *obj_request)
2661 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
2662 struct rbd_img_request *parent_request = NULL;
2665 struct page **pages = NULL;
2669 rbd_assert(rbd_dev->parent != NULL);
2672 * Determine the byte range covered by the object in the
2673 * child image to which the original request was to be sent.
2675 img_offset = obj_request->img_offset - obj_request->offset;
2676 length = rbd_obj_bytes(&rbd_dev->header);
2679 * There is no defined parent data beyond the parent
2680 * overlap, so limit what we read at that boundary if
2683 if (img_offset + length > rbd_dev->parent_overlap) {
2684 rbd_assert(img_offset < rbd_dev->parent_overlap);
2685 length = rbd_dev->parent_overlap - img_offset;
2689 * Allocate a page array big enough to receive the data read
2692 page_count = (u32)calc_pages_for(0, length);
2693 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2694 if (IS_ERR(pages)) {
2695 result = PTR_ERR(pages);
2701 parent_request = rbd_parent_request_create(obj_request,
2702 img_offset, length);
2703 if (!parent_request)
2706 result = rbd_img_request_fill(parent_request, OBJ_REQUEST_PAGES, pages);
2710 parent_request->copyup_pages = pages;
2711 parent_request->copyup_page_count = page_count;
2712 parent_request->callback = rbd_img_obj_parent_read_full_callback;
2714 result = rbd_img_request_submit(parent_request);
2718 parent_request->copyup_pages = NULL;
2719 parent_request->copyup_page_count = 0;
2720 parent_request->obj_request = NULL;
2721 rbd_obj_request_put(obj_request);
2724 ceph_release_page_vector(pages, page_count);
2726 rbd_img_request_put(parent_request);
2730 static void rbd_img_obj_exists_callback(struct rbd_obj_request *obj_request)
2732 struct rbd_obj_request *orig_request;
2733 struct rbd_device *rbd_dev;
2736 rbd_assert(!obj_request_img_data_test(obj_request));
2739 * All we need from the object request is the original
2740 * request and the result of the STAT op. Grab those, then
2741 * we're done with the request.
2743 orig_request = obj_request->obj_request;
2744 obj_request->obj_request = NULL;
2745 rbd_obj_request_put(orig_request);
2746 rbd_assert(orig_request);
2747 rbd_assert(orig_request->img_request);
2749 result = obj_request->result;
2750 obj_request->result = 0;
2752 dout("%s: obj %p for obj %p result %d %llu/%llu\n", __func__,
2753 obj_request, orig_request, result,
2754 obj_request->xferred, obj_request->length);
2755 rbd_obj_request_put(obj_request);
2758 * If the overlap has become 0 (most likely because the
2759 * image has been flattened) we need to re-submit the
2762 rbd_dev = orig_request->img_request->rbd_dev;
2763 if (!rbd_dev->parent_overlap) {
2764 rbd_obj_request_submit(orig_request);
2769 * Our only purpose here is to determine whether the object
2770 * exists, and we don't want to treat the non-existence as
2771 * an error. If something else comes back, transfer the
2772 * error to the original request and complete it now.
2775 obj_request_existence_set(orig_request, true);
2776 } else if (result == -ENOENT) {
2777 obj_request_existence_set(orig_request, false);
2779 goto fail_orig_request;
2783 * Resubmit the original request now that we have recorded
2784 * whether the target object exists.
2786 result = rbd_img_obj_request_submit(orig_request);
2788 goto fail_orig_request;
2793 rbd_obj_request_error(orig_request, result);
2796 static int rbd_img_obj_exists_submit(struct rbd_obj_request *obj_request)
2798 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
2799 struct rbd_obj_request *stat_request;
2800 struct page **pages;
2805 stat_request = rbd_obj_request_create(OBJ_REQUEST_PAGES);
2809 stat_request->object_no = obj_request->object_no;
2811 stat_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1,
2813 if (!stat_request->osd_req) {
2815 goto fail_stat_request;
2819 * The response data for a STAT call consists of:
2826 size = sizeof (__le64) + sizeof (__le32) + sizeof (__le32);
2827 page_count = (u32)calc_pages_for(0, size);
2828 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2829 if (IS_ERR(pages)) {
2830 ret = PTR_ERR(pages);
2831 goto fail_stat_request;
2834 osd_req_op_init(stat_request->osd_req, 0, CEPH_OSD_OP_STAT, 0);
2835 osd_req_op_raw_data_in_pages(stat_request->osd_req, 0, pages, size, 0,
2838 rbd_obj_request_get(obj_request);
2839 stat_request->obj_request = obj_request;
2840 stat_request->pages = pages;
2841 stat_request->page_count = page_count;
2842 stat_request->callback = rbd_img_obj_exists_callback;
2844 rbd_obj_request_submit(stat_request);
2848 rbd_obj_request_put(stat_request);
2852 static bool img_obj_request_simple(struct rbd_obj_request *obj_request)
2854 struct rbd_img_request *img_request = obj_request->img_request;
2855 struct rbd_device *rbd_dev = img_request->rbd_dev;
2858 if (!img_request_write_test(img_request) &&
2859 !img_request_discard_test(img_request))
2862 /* Non-layered writes */
2863 if (!img_request_layered_test(img_request))
2867 * Layered writes outside of the parent overlap range don't
2868 * share any data with the parent.
2870 if (!obj_request_overlaps_parent(obj_request))
2874 * Entire-object layered writes - we will overwrite whatever
2875 * parent data there is anyway.
2877 if (!obj_request->offset &&
2878 obj_request->length == rbd_obj_bytes(&rbd_dev->header))
2882 * If the object is known to already exist, its parent data has
2883 * already been copied.
2885 if (obj_request_known_test(obj_request) &&
2886 obj_request_exists_test(obj_request))
2892 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request)
2894 rbd_assert(obj_request_img_data_test(obj_request));
2895 rbd_assert(obj_request_type_valid(obj_request->type));
2896 rbd_assert(obj_request->img_request);
2898 if (img_obj_request_simple(obj_request)) {
2899 rbd_obj_request_submit(obj_request);
2904 * It's a layered write. The target object might exist but
2905 * we may not know that yet. If we know it doesn't exist,
2906 * start by reading the data for the full target object from
2907 * the parent so we can use it for a copyup to the target.
2909 if (obj_request_known_test(obj_request))
2910 return rbd_img_obj_parent_read_full(obj_request);
2912 /* We don't know whether the target exists. Go find out. */
2914 return rbd_img_obj_exists_submit(obj_request);
2917 static int rbd_img_request_submit(struct rbd_img_request *img_request)
2919 struct rbd_obj_request *obj_request;
2920 struct rbd_obj_request *next_obj_request;
2923 dout("%s: img %p\n", __func__, img_request);
2925 rbd_img_request_get(img_request);
2926 for_each_obj_request_safe(img_request, obj_request, next_obj_request) {
2927 ret = rbd_img_obj_request_submit(obj_request);
2933 rbd_img_request_put(img_request);
2937 static void rbd_img_parent_read_callback(struct rbd_img_request *img_request)
2939 struct rbd_obj_request *obj_request;
2940 struct rbd_device *rbd_dev;
2945 rbd_assert(img_request_child_test(img_request));
2947 /* First get what we need from the image request and release it */
2949 obj_request = img_request->obj_request;
2950 img_xferred = img_request->xferred;
2951 img_result = img_request->result;
2952 rbd_img_request_put(img_request);
2955 * If the overlap has become 0 (most likely because the
2956 * image has been flattened) we need to re-submit the
2959 rbd_assert(obj_request);
2960 rbd_assert(obj_request->img_request);
2961 rbd_dev = obj_request->img_request->rbd_dev;
2962 if (!rbd_dev->parent_overlap) {
2963 rbd_obj_request_submit(obj_request);
2967 obj_request->result = img_result;
2968 if (obj_request->result)
2972 * We need to zero anything beyond the parent overlap
2973 * boundary. Since rbd_img_obj_request_read_callback()
2974 * will zero anything beyond the end of a short read, an
2975 * easy way to do this is to pretend the data from the
2976 * parent came up short--ending at the overlap boundary.
2978 rbd_assert(obj_request->img_offset < U64_MAX - obj_request->length);
2979 obj_end = obj_request->img_offset + obj_request->length;
2980 if (obj_end > rbd_dev->parent_overlap) {
2983 if (obj_request->img_offset < rbd_dev->parent_overlap)
2984 xferred = rbd_dev->parent_overlap -
2985 obj_request->img_offset;
2987 obj_request->xferred = min(img_xferred, xferred);
2989 obj_request->xferred = img_xferred;
2992 rbd_img_obj_request_read_callback(obj_request);
2993 rbd_obj_request_complete(obj_request);
2996 static void rbd_img_parent_read(struct rbd_obj_request *obj_request)
2998 struct rbd_img_request *img_request;
3001 rbd_assert(obj_request_img_data_test(obj_request));
3002 rbd_assert(obj_request->img_request != NULL);
3003 rbd_assert(obj_request->result == (s32) -ENOENT);
3004 rbd_assert(obj_request_type_valid(obj_request->type));
3006 /* rbd_read_finish(obj_request, obj_request->length); */
3007 img_request = rbd_parent_request_create(obj_request,
3008 obj_request->img_offset,
3009 obj_request->length);
3014 if (obj_request->type == OBJ_REQUEST_BIO)
3015 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
3016 obj_request->bio_list);
3018 result = rbd_img_request_fill(img_request, OBJ_REQUEST_PAGES,
3019 obj_request->pages);
3023 img_request->callback = rbd_img_parent_read_callback;
3024 result = rbd_img_request_submit(img_request);
3031 rbd_img_request_put(img_request);
3032 obj_request->result = result;
3033 obj_request->xferred = 0;
3034 obj_request_done_set(obj_request);
3037 static const struct rbd_client_id rbd_empty_cid;
3039 static bool rbd_cid_equal(const struct rbd_client_id *lhs,
3040 const struct rbd_client_id *rhs)
3042 return lhs->gid == rhs->gid && lhs->handle == rhs->handle;
3045 static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev)
3047 struct rbd_client_id cid;
3049 mutex_lock(&rbd_dev->watch_mutex);
3050 cid.gid = ceph_client_gid(rbd_dev->rbd_client->client);
3051 cid.handle = rbd_dev->watch_cookie;
3052 mutex_unlock(&rbd_dev->watch_mutex);
3057 * lock_rwsem must be held for write
3059 static void rbd_set_owner_cid(struct rbd_device *rbd_dev,
3060 const struct rbd_client_id *cid)
3062 dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev,
3063 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle,
3064 cid->gid, cid->handle);
3065 rbd_dev->owner_cid = *cid; /* struct */
3068 static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf)
3070 mutex_lock(&rbd_dev->watch_mutex);
3071 sprintf(buf, "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie);
3072 mutex_unlock(&rbd_dev->watch_mutex);
3076 * lock_rwsem must be held for write
3078 static int rbd_lock(struct rbd_device *rbd_dev)
3080 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3081 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3085 WARN_ON(__rbd_is_lock_owner(rbd_dev) ||
3086 rbd_dev->lock_cookie[0] != '\0');
3088 format_lock_cookie(rbd_dev, cookie);
3089 ret = ceph_cls_lock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3090 RBD_LOCK_NAME, CEPH_CLS_LOCK_EXCLUSIVE, cookie,
3091 RBD_LOCK_TAG, "", 0);
3095 rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED;
3096 strcpy(rbd_dev->lock_cookie, cookie);
3097 rbd_set_owner_cid(rbd_dev, &cid);
3098 queue_work(rbd_dev->task_wq, &rbd_dev->acquired_lock_work);
3103 * lock_rwsem must be held for write
3105 static void rbd_unlock(struct rbd_device *rbd_dev)
3107 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3110 WARN_ON(!__rbd_is_lock_owner(rbd_dev) ||
3111 rbd_dev->lock_cookie[0] == '\0');
3113 ret = ceph_cls_unlock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3114 RBD_LOCK_NAME, rbd_dev->lock_cookie);
3115 if (ret && ret != -ENOENT)
3116 rbd_warn(rbd_dev, "failed to unlock: %d", ret);
3118 /* treat errors as the image is unlocked */
3119 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
3120 rbd_dev->lock_cookie[0] = '\0';
3121 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3122 queue_work(rbd_dev->task_wq, &rbd_dev->released_lock_work);
3125 static int __rbd_notify_op_lock(struct rbd_device *rbd_dev,
3126 enum rbd_notify_op notify_op,
3127 struct page ***preply_pages,
3130 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3131 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3132 int buf_size = 4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN;
3136 dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op);
3138 /* encode *LockPayload NotifyMessage (op + ClientId) */
3139 ceph_start_encoding(&p, 2, 1, buf_size - CEPH_ENCODING_START_BLK_LEN);
3140 ceph_encode_32(&p, notify_op);
3141 ceph_encode_64(&p, cid.gid);
3142 ceph_encode_64(&p, cid.handle);
3144 return ceph_osdc_notify(osdc, &rbd_dev->header_oid,
3145 &rbd_dev->header_oloc, buf, buf_size,
3146 RBD_NOTIFY_TIMEOUT, preply_pages, preply_len);
3149 static void rbd_notify_op_lock(struct rbd_device *rbd_dev,
3150 enum rbd_notify_op notify_op)
3152 struct page **reply_pages;
3155 __rbd_notify_op_lock(rbd_dev, notify_op, &reply_pages, &reply_len);
3156 ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3159 static void rbd_notify_acquired_lock(struct work_struct *work)
3161 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3162 acquired_lock_work);
3164 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_ACQUIRED_LOCK);
3167 static void rbd_notify_released_lock(struct work_struct *work)
3169 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3170 released_lock_work);
3172 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_RELEASED_LOCK);
3175 static int rbd_request_lock(struct rbd_device *rbd_dev)
3177 struct page **reply_pages;
3179 bool lock_owner_responded = false;
3182 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3184 ret = __rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_REQUEST_LOCK,
3185 &reply_pages, &reply_len);
3186 if (ret && ret != -ETIMEDOUT) {
3187 rbd_warn(rbd_dev, "failed to request lock: %d", ret);
3191 if (reply_len > 0 && reply_len <= PAGE_SIZE) {
3192 void *p = page_address(reply_pages[0]);
3193 void *const end = p + reply_len;
3196 ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */
3201 ceph_decode_need(&p, end, 8 + 8, e_inval);
3202 p += 8 + 8; /* skip gid and cookie */
3204 ceph_decode_32_safe(&p, end, len, e_inval);
3208 if (lock_owner_responded) {
3210 "duplicate lock owners detected");
3215 lock_owner_responded = true;
3216 ret = ceph_start_decoding(&p, end, 1, "ResponseMessage",
3220 "failed to decode ResponseMessage: %d",
3225 ret = ceph_decode_32(&p);
3229 if (!lock_owner_responded) {
3230 rbd_warn(rbd_dev, "no lock owners detected");
3235 ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3243 static void wake_requests(struct rbd_device *rbd_dev, bool wake_all)
3245 dout("%s rbd_dev %p wake_all %d\n", __func__, rbd_dev, wake_all);
3247 cancel_delayed_work(&rbd_dev->lock_dwork);
3249 wake_up_all(&rbd_dev->lock_waitq);
3251 wake_up(&rbd_dev->lock_waitq);
3254 static int get_lock_owner_info(struct rbd_device *rbd_dev,
3255 struct ceph_locker **lockers, u32 *num_lockers)
3257 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3262 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3264 ret = ceph_cls_lock_info(osdc, &rbd_dev->header_oid,
3265 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3266 &lock_type, &lock_tag, lockers, num_lockers);
3270 if (*num_lockers == 0) {
3271 dout("%s rbd_dev %p no lockers detected\n", __func__, rbd_dev);
3275 if (strcmp(lock_tag, RBD_LOCK_TAG)) {
3276 rbd_warn(rbd_dev, "locked by external mechanism, tag %s",
3282 if (lock_type == CEPH_CLS_LOCK_SHARED) {
3283 rbd_warn(rbd_dev, "shared lock type detected");
3288 if (strncmp((*lockers)[0].id.cookie, RBD_LOCK_COOKIE_PREFIX,
3289 strlen(RBD_LOCK_COOKIE_PREFIX))) {
3290 rbd_warn(rbd_dev, "locked by external mechanism, cookie %s",
3291 (*lockers)[0].id.cookie);
3301 static int find_watcher(struct rbd_device *rbd_dev,
3302 const struct ceph_locker *locker)
3304 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3305 struct ceph_watch_item *watchers;
3311 ret = ceph_osdc_list_watchers(osdc, &rbd_dev->header_oid,
3312 &rbd_dev->header_oloc, &watchers,
3317 sscanf(locker->id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu", &cookie);
3318 for (i = 0; i < num_watchers; i++) {
3319 if (!memcmp(&watchers[i].addr, &locker->info.addr,
3320 sizeof(locker->info.addr)) &&
3321 watchers[i].cookie == cookie) {
3322 struct rbd_client_id cid = {
3323 .gid = le64_to_cpu(watchers[i].name.num),
3327 dout("%s rbd_dev %p found cid %llu-%llu\n", __func__,
3328 rbd_dev, cid.gid, cid.handle);
3329 rbd_set_owner_cid(rbd_dev, &cid);
3335 dout("%s rbd_dev %p no watchers\n", __func__, rbd_dev);
3343 * lock_rwsem must be held for write
3345 static int rbd_try_lock(struct rbd_device *rbd_dev)
3347 struct ceph_client *client = rbd_dev->rbd_client->client;
3348 struct ceph_locker *lockers;
3353 ret = rbd_lock(rbd_dev);
3357 /* determine if the current lock holder is still alive */
3358 ret = get_lock_owner_info(rbd_dev, &lockers, &num_lockers);
3362 if (num_lockers == 0)
3365 ret = find_watcher(rbd_dev, lockers);
3368 ret = 0; /* have to request lock */
3372 rbd_warn(rbd_dev, "%s%llu seems dead, breaking lock",
3373 ENTITY_NAME(lockers[0].id.name));
3375 ret = ceph_monc_blacklist_add(&client->monc,
3376 &lockers[0].info.addr);
3378 rbd_warn(rbd_dev, "blacklist of %s%llu failed: %d",
3379 ENTITY_NAME(lockers[0].id.name), ret);
3383 ret = ceph_cls_break_lock(&client->osdc, &rbd_dev->header_oid,
3384 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3385 lockers[0].id.cookie,
3386 &lockers[0].id.name);
3387 if (ret && ret != -ENOENT)
3391 ceph_free_lockers(lockers, num_lockers);
3395 ceph_free_lockers(lockers, num_lockers);
3400 * ret is set only if lock_state is RBD_LOCK_STATE_UNLOCKED
3402 static enum rbd_lock_state rbd_try_acquire_lock(struct rbd_device *rbd_dev,
3405 enum rbd_lock_state lock_state;
3407 down_read(&rbd_dev->lock_rwsem);
3408 dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
3409 rbd_dev->lock_state);
3410 if (__rbd_is_lock_owner(rbd_dev)) {
3411 lock_state = rbd_dev->lock_state;
3412 up_read(&rbd_dev->lock_rwsem);
3416 up_read(&rbd_dev->lock_rwsem);
3417 down_write(&rbd_dev->lock_rwsem);
3418 dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
3419 rbd_dev->lock_state);
3420 if (!__rbd_is_lock_owner(rbd_dev)) {
3421 *pret = rbd_try_lock(rbd_dev);
3423 rbd_warn(rbd_dev, "failed to acquire lock: %d", *pret);
3426 lock_state = rbd_dev->lock_state;
3427 up_write(&rbd_dev->lock_rwsem);
3431 static void rbd_acquire_lock(struct work_struct *work)
3433 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
3434 struct rbd_device, lock_dwork);
3435 enum rbd_lock_state lock_state;
3438 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3440 lock_state = rbd_try_acquire_lock(rbd_dev, &ret);
3441 if (lock_state != RBD_LOCK_STATE_UNLOCKED || ret == -EBLACKLISTED) {
3442 if (lock_state == RBD_LOCK_STATE_LOCKED)
3443 wake_requests(rbd_dev, true);
3444 dout("%s rbd_dev %p lock_state %d ret %d - done\n", __func__,
3445 rbd_dev, lock_state, ret);
3449 ret = rbd_request_lock(rbd_dev);
3450 if (ret == -ETIMEDOUT) {
3451 goto again; /* treat this as a dead client */
3452 } else if (ret == -EROFS) {
3453 rbd_warn(rbd_dev, "peer will not release lock");
3455 * If this is rbd_add_acquire_lock(), we want to fail
3456 * immediately -- reuse BLACKLISTED flag. Otherwise we
3459 if (!(rbd_dev->disk->flags & GENHD_FL_UP)) {
3460 set_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags);
3461 /* wake "rbd map --exclusive" process */
3462 wake_requests(rbd_dev, false);
3464 } else if (ret < 0) {
3465 rbd_warn(rbd_dev, "error requesting lock: %d", ret);
3466 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
3470 * lock owner acked, but resend if we don't see them
3473 dout("%s rbd_dev %p requeueing lock_dwork\n", __func__,
3475 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
3476 msecs_to_jiffies(2 * RBD_NOTIFY_TIMEOUT * MSEC_PER_SEC));
3481 * lock_rwsem must be held for write
3483 static bool rbd_release_lock(struct rbd_device *rbd_dev)
3485 dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
3486 rbd_dev->lock_state);
3487 if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED)
3490 rbd_dev->lock_state = RBD_LOCK_STATE_RELEASING;
3491 downgrade_write(&rbd_dev->lock_rwsem);
3493 * Ensure that all in-flight IO is flushed.
3495 * FIXME: ceph_osdc_sync() flushes the entire OSD client, which
3496 * may be shared with other devices.
3498 ceph_osdc_sync(&rbd_dev->rbd_client->client->osdc);
3499 up_read(&rbd_dev->lock_rwsem);
3501 down_write(&rbd_dev->lock_rwsem);
3502 dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
3503 rbd_dev->lock_state);
3504 if (rbd_dev->lock_state != RBD_LOCK_STATE_RELEASING)
3507 rbd_unlock(rbd_dev);
3509 * Give others a chance to grab the lock - we would re-acquire
3510 * almost immediately if we got new IO during ceph_osdc_sync()
3511 * otherwise. We need to ack our own notifications, so this
3512 * lock_dwork will be requeued from rbd_wait_state_locked()
3513 * after wake_requests() in rbd_handle_released_lock().
3515 cancel_delayed_work(&rbd_dev->lock_dwork);
3519 static void rbd_release_lock_work(struct work_struct *work)
3521 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3524 down_write(&rbd_dev->lock_rwsem);
3525 rbd_release_lock(rbd_dev);
3526 up_write(&rbd_dev->lock_rwsem);
3529 static void rbd_handle_acquired_lock(struct rbd_device *rbd_dev, u8 struct_v,
3532 struct rbd_client_id cid = { 0 };
3534 if (struct_v >= 2) {
3535 cid.gid = ceph_decode_64(p);
3536 cid.handle = ceph_decode_64(p);
3539 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
3541 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
3542 down_write(&rbd_dev->lock_rwsem);
3543 if (rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
3545 * we already know that the remote client is
3548 up_write(&rbd_dev->lock_rwsem);
3552 rbd_set_owner_cid(rbd_dev, &cid);
3553 downgrade_write(&rbd_dev->lock_rwsem);
3555 down_read(&rbd_dev->lock_rwsem);
3558 if (!__rbd_is_lock_owner(rbd_dev))
3559 wake_requests(rbd_dev, false);
3560 up_read(&rbd_dev->lock_rwsem);
3563 static void rbd_handle_released_lock(struct rbd_device *rbd_dev, u8 struct_v,
3566 struct rbd_client_id cid = { 0 };
3568 if (struct_v >= 2) {
3569 cid.gid = ceph_decode_64(p);
3570 cid.handle = ceph_decode_64(p);
3573 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
3575 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
3576 down_write(&rbd_dev->lock_rwsem);
3577 if (!rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
3578 dout("%s rbd_dev %p unexpected owner, cid %llu-%llu != owner_cid %llu-%llu\n",
3579 __func__, rbd_dev, cid.gid, cid.handle,
3580 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle);
3581 up_write(&rbd_dev->lock_rwsem);
3585 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3586 downgrade_write(&rbd_dev->lock_rwsem);
3588 down_read(&rbd_dev->lock_rwsem);
3591 if (!__rbd_is_lock_owner(rbd_dev))
3592 wake_requests(rbd_dev, false);
3593 up_read(&rbd_dev->lock_rwsem);
3597 * Returns result for ResponseMessage to be encoded (<= 0), or 1 if no
3598 * ResponseMessage is needed.
3600 static int rbd_handle_request_lock(struct rbd_device *rbd_dev, u8 struct_v,
3603 struct rbd_client_id my_cid = rbd_get_cid(rbd_dev);
3604 struct rbd_client_id cid = { 0 };
3607 if (struct_v >= 2) {
3608 cid.gid = ceph_decode_64(p);
3609 cid.handle = ceph_decode_64(p);
3612 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
3614 if (rbd_cid_equal(&cid, &my_cid))
3617 down_read(&rbd_dev->lock_rwsem);
3618 if (__rbd_is_lock_owner(rbd_dev)) {
3619 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED &&
3620 rbd_cid_equal(&rbd_dev->owner_cid, &rbd_empty_cid))
3624 * encode ResponseMessage(0) so the peer can detect
3629 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED) {
3630 if (!rbd_dev->opts->exclusive) {
3631 dout("%s rbd_dev %p queueing unlock_work\n",
3633 queue_work(rbd_dev->task_wq,
3634 &rbd_dev->unlock_work);
3636 /* refuse to release the lock */
3643 up_read(&rbd_dev->lock_rwsem);
3647 static void __rbd_acknowledge_notify(struct rbd_device *rbd_dev,
3648 u64 notify_id, u64 cookie, s32 *result)
3650 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3651 int buf_size = 4 + CEPH_ENCODING_START_BLK_LEN;
3658 /* encode ResponseMessage */
3659 ceph_start_encoding(&p, 1, 1,
3660 buf_size - CEPH_ENCODING_START_BLK_LEN);
3661 ceph_encode_32(&p, *result);
3666 ret = ceph_osdc_notify_ack(osdc, &rbd_dev->header_oid,
3667 &rbd_dev->header_oloc, notify_id, cookie,
3670 rbd_warn(rbd_dev, "acknowledge_notify failed: %d", ret);
3673 static void rbd_acknowledge_notify(struct rbd_device *rbd_dev, u64 notify_id,
3676 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3677 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, NULL);
3680 static void rbd_acknowledge_notify_result(struct rbd_device *rbd_dev,
3681 u64 notify_id, u64 cookie, s32 result)
3683 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
3684 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, &result);
3687 static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie,
3688 u64 notifier_id, void *data, size_t data_len)
3690 struct rbd_device *rbd_dev = arg;
3692 void *const end = p + data_len;
3698 dout("%s rbd_dev %p cookie %llu notify_id %llu data_len %zu\n",
3699 __func__, rbd_dev, cookie, notify_id, data_len);
3701 ret = ceph_start_decoding(&p, end, 1, "NotifyMessage",
3704 rbd_warn(rbd_dev, "failed to decode NotifyMessage: %d",
3709 notify_op = ceph_decode_32(&p);
3711 /* legacy notification for header updates */
3712 notify_op = RBD_NOTIFY_OP_HEADER_UPDATE;
3716 dout("%s rbd_dev %p notify_op %u\n", __func__, rbd_dev, notify_op);
3717 switch (notify_op) {
3718 case RBD_NOTIFY_OP_ACQUIRED_LOCK:
3719 rbd_handle_acquired_lock(rbd_dev, struct_v, &p);
3720 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
3722 case RBD_NOTIFY_OP_RELEASED_LOCK:
3723 rbd_handle_released_lock(rbd_dev, struct_v, &p);
3724 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
3726 case RBD_NOTIFY_OP_REQUEST_LOCK:
3727 ret = rbd_handle_request_lock(rbd_dev, struct_v, &p);
3729 rbd_acknowledge_notify_result(rbd_dev, notify_id,
3732 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
3734 case RBD_NOTIFY_OP_HEADER_UPDATE:
3735 ret = rbd_dev_refresh(rbd_dev);
3737 rbd_warn(rbd_dev, "refresh failed: %d", ret);
3739 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
3742 if (rbd_is_lock_owner(rbd_dev))
3743 rbd_acknowledge_notify_result(rbd_dev, notify_id,
3744 cookie, -EOPNOTSUPP);
3746 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
3751 static void __rbd_unregister_watch(struct rbd_device *rbd_dev);
3753 static void rbd_watch_errcb(void *arg, u64 cookie, int err)
3755 struct rbd_device *rbd_dev = arg;
3757 rbd_warn(rbd_dev, "encountered watch error: %d", err);
3759 down_write(&rbd_dev->lock_rwsem);
3760 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3761 up_write(&rbd_dev->lock_rwsem);
3763 mutex_lock(&rbd_dev->watch_mutex);
3764 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED) {
3765 __rbd_unregister_watch(rbd_dev);
3766 rbd_dev->watch_state = RBD_WATCH_STATE_ERROR;
3768 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->watch_dwork, 0);
3770 mutex_unlock(&rbd_dev->watch_mutex);
3774 * watch_mutex must be locked
3776 static int __rbd_register_watch(struct rbd_device *rbd_dev)
3778 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3779 struct ceph_osd_linger_request *handle;
3781 rbd_assert(!rbd_dev->watch_handle);
3782 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3784 handle = ceph_osdc_watch(osdc, &rbd_dev->header_oid,
3785 &rbd_dev->header_oloc, rbd_watch_cb,
3786 rbd_watch_errcb, rbd_dev);
3788 return PTR_ERR(handle);
3790 rbd_dev->watch_handle = handle;
3795 * watch_mutex must be locked
3797 static void __rbd_unregister_watch(struct rbd_device *rbd_dev)
3799 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3802 rbd_assert(rbd_dev->watch_handle);
3803 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3805 ret = ceph_osdc_unwatch(osdc, rbd_dev->watch_handle);
3807 rbd_warn(rbd_dev, "failed to unwatch: %d", ret);
3809 rbd_dev->watch_handle = NULL;
3812 static int rbd_register_watch(struct rbd_device *rbd_dev)
3816 mutex_lock(&rbd_dev->watch_mutex);
3817 rbd_assert(rbd_dev->watch_state == RBD_WATCH_STATE_UNREGISTERED);
3818 ret = __rbd_register_watch(rbd_dev);
3822 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
3823 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
3826 mutex_unlock(&rbd_dev->watch_mutex);
3830 static void cancel_tasks_sync(struct rbd_device *rbd_dev)
3832 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3834 cancel_delayed_work_sync(&rbd_dev->watch_dwork);
3835 cancel_work_sync(&rbd_dev->acquired_lock_work);
3836 cancel_work_sync(&rbd_dev->released_lock_work);
3837 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
3838 cancel_work_sync(&rbd_dev->unlock_work);
3841 static void rbd_unregister_watch(struct rbd_device *rbd_dev)
3843 WARN_ON(waitqueue_active(&rbd_dev->lock_waitq));
3844 cancel_tasks_sync(rbd_dev);
3846 mutex_lock(&rbd_dev->watch_mutex);
3847 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED)
3848 __rbd_unregister_watch(rbd_dev);
3849 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
3850 mutex_unlock(&rbd_dev->watch_mutex);
3852 ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
3856 * lock_rwsem must be held for write
3858 static void rbd_reacquire_lock(struct rbd_device *rbd_dev)
3860 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3864 WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED);
3866 format_lock_cookie(rbd_dev, cookie);
3867 ret = ceph_cls_set_cookie(osdc, &rbd_dev->header_oid,
3868 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3869 CEPH_CLS_LOCK_EXCLUSIVE, rbd_dev->lock_cookie,
3870 RBD_LOCK_TAG, cookie);
3872 if (ret != -EOPNOTSUPP)
3873 rbd_warn(rbd_dev, "failed to update lock cookie: %d",
3877 * Lock cookie cannot be updated on older OSDs, so do
3878 * a manual release and queue an acquire.
3880 if (rbd_release_lock(rbd_dev))
3881 queue_delayed_work(rbd_dev->task_wq,
3882 &rbd_dev->lock_dwork, 0);
3884 strcpy(rbd_dev->lock_cookie, cookie);
3888 static void rbd_reregister_watch(struct work_struct *work)
3890 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
3891 struct rbd_device, watch_dwork);
3894 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3896 mutex_lock(&rbd_dev->watch_mutex);
3897 if (rbd_dev->watch_state != RBD_WATCH_STATE_ERROR) {
3898 mutex_unlock(&rbd_dev->watch_mutex);
3902 ret = __rbd_register_watch(rbd_dev);
3904 rbd_warn(rbd_dev, "failed to reregister watch: %d", ret);
3905 if (ret == -EBLACKLISTED || ret == -ENOENT) {
3906 set_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags);
3907 wake_requests(rbd_dev, true);
3909 queue_delayed_work(rbd_dev->task_wq,
3910 &rbd_dev->watch_dwork,
3913 mutex_unlock(&rbd_dev->watch_mutex);
3917 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
3918 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
3919 mutex_unlock(&rbd_dev->watch_mutex);
3921 down_write(&rbd_dev->lock_rwsem);
3922 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED)
3923 rbd_reacquire_lock(rbd_dev);
3924 up_write(&rbd_dev->lock_rwsem);
3926 ret = rbd_dev_refresh(rbd_dev);
3928 rbd_warn(rbd_dev, "reregisteration refresh failed: %d", ret);
3932 * Synchronous osd object method call. Returns the number of bytes
3933 * returned in the outbound buffer, or a negative error code.
3935 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
3936 struct ceph_object_id *oid,
3937 struct ceph_object_locator *oloc,
3938 const char *method_name,
3939 const void *outbound,
3940 size_t outbound_size,
3942 size_t inbound_size)
3944 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3945 struct page *req_page = NULL;
3946 struct page *reply_page;
3950 * Method calls are ultimately read operations. The result
3951 * should placed into the inbound buffer provided. They
3952 * also supply outbound data--parameters for the object
3953 * method. Currently if this is present it will be a
3957 if (outbound_size > PAGE_SIZE)
3960 req_page = alloc_page(GFP_KERNEL);
3964 memcpy(page_address(req_page), outbound, outbound_size);
3967 reply_page = alloc_page(GFP_KERNEL);
3970 __free_page(req_page);
3974 ret = ceph_osdc_call(osdc, oid, oloc, RBD_DRV_NAME, method_name,
3975 CEPH_OSD_FLAG_READ, req_page, outbound_size,
3976 reply_page, &inbound_size);
3978 memcpy(inbound, page_address(reply_page), inbound_size);
3983 __free_page(req_page);
3984 __free_page(reply_page);
3989 * lock_rwsem must be held for read
3991 static void rbd_wait_state_locked(struct rbd_device *rbd_dev)
3997 * Note the use of mod_delayed_work() in rbd_acquire_lock()
3998 * and cancel_delayed_work() in wake_requests().
4000 dout("%s rbd_dev %p queueing lock_dwork\n", __func__, rbd_dev);
4001 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4002 prepare_to_wait_exclusive(&rbd_dev->lock_waitq, &wait,
4003 TASK_UNINTERRUPTIBLE);
4004 up_read(&rbd_dev->lock_rwsem);
4006 down_read(&rbd_dev->lock_rwsem);
4007 } while (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED &&
4008 !test_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags));
4010 finish_wait(&rbd_dev->lock_waitq, &wait);
4013 static void rbd_queue_workfn(struct work_struct *work)
4015 struct request *rq = blk_mq_rq_from_pdu(work);
4016 struct rbd_device *rbd_dev = rq->q->queuedata;
4017 struct rbd_img_request *img_request;
4018 struct ceph_snap_context *snapc = NULL;
4019 u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
4020 u64 length = blk_rq_bytes(rq);
4021 enum obj_operation_type op_type;
4023 bool must_be_locked;
4026 switch (req_op(rq)) {
4027 case REQ_OP_DISCARD:
4028 op_type = OBJ_OP_DISCARD;
4031 op_type = OBJ_OP_WRITE;
4034 op_type = OBJ_OP_READ;
4037 dout("%s: non-fs request type %d\n", __func__, req_op(rq));
4042 /* Ignore/skip any zero-length requests */
4045 dout("%s: zero-length request\n", __func__);
4050 /* Only reads are allowed to a read-only device */
4052 if (op_type != OBJ_OP_READ) {
4053 if (rbd_dev->mapping.read_only) {
4057 rbd_assert(rbd_dev->spec->snap_id == CEPH_NOSNAP);
4061 * Quit early if the mapped snapshot no longer exists. It's
4062 * still possible the snapshot will have disappeared by the
4063 * time our request arrives at the osd, but there's no sense in
4064 * sending it if we already know.
4066 if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags)) {
4067 dout("request for non-existent snapshot");
4068 rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
4073 if (offset && length > U64_MAX - offset + 1) {
4074 rbd_warn(rbd_dev, "bad request range (%llu~%llu)", offset,
4077 goto err_rq; /* Shouldn't happen */
4080 blk_mq_start_request(rq);
4082 down_read(&rbd_dev->header_rwsem);
4083 mapping_size = rbd_dev->mapping.size;
4084 if (op_type != OBJ_OP_READ) {
4085 snapc = rbd_dev->header.snapc;
4086 ceph_get_snap_context(snapc);
4088 up_read(&rbd_dev->header_rwsem);
4090 if (offset + length > mapping_size) {
4091 rbd_warn(rbd_dev, "beyond EOD (%llu~%llu > %llu)", offset,
4092 length, mapping_size);
4098 (rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK) &&
4099 (op_type != OBJ_OP_READ || rbd_dev->opts->lock_on_read);
4100 if (must_be_locked) {
4101 down_read(&rbd_dev->lock_rwsem);
4102 if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED &&
4103 !test_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags)) {
4104 if (rbd_dev->opts->exclusive) {
4105 rbd_warn(rbd_dev, "exclusive lock required");
4109 rbd_wait_state_locked(rbd_dev);
4111 if (test_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags)) {
4112 result = -EBLACKLISTED;
4117 img_request = rbd_img_request_create(rbd_dev, offset, length, op_type,
4123 img_request->rq = rq;
4124 snapc = NULL; /* img_request consumes a ref */
4126 if (op_type == OBJ_OP_DISCARD)
4127 result = rbd_img_request_fill(img_request, OBJ_REQUEST_NODATA,
4130 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
4133 goto err_img_request;
4135 result = rbd_img_request_submit(img_request);
4137 goto err_img_request;
4140 up_read(&rbd_dev->lock_rwsem);
4144 rbd_img_request_put(img_request);
4147 up_read(&rbd_dev->lock_rwsem);
4150 rbd_warn(rbd_dev, "%s %llx at %llx result %d",
4151 obj_op_name(op_type), length, offset, result);
4152 ceph_put_snap_context(snapc);
4154 blk_mq_end_request(rq, errno_to_blk_status(result));
4157 static int rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
4158 const struct blk_mq_queue_data *bd)
4160 struct request *rq = bd->rq;
4161 struct work_struct *work = blk_mq_rq_to_pdu(rq);
4163 queue_work(rbd_wq, work);
4164 return BLK_MQ_RQ_QUEUE_OK;
4167 static void rbd_free_disk(struct rbd_device *rbd_dev)
4169 blk_cleanup_queue(rbd_dev->disk->queue);
4170 blk_mq_free_tag_set(&rbd_dev->tag_set);
4171 put_disk(rbd_dev->disk);
4172 rbd_dev->disk = NULL;
4175 static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
4176 struct ceph_object_id *oid,
4177 struct ceph_object_locator *oloc,
4178 void *buf, int buf_len)
4181 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4182 struct ceph_osd_request *req;
4183 struct page **pages;
4184 int num_pages = calc_pages_for(0, buf_len);
4187 req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_KERNEL);
4191 ceph_oid_copy(&req->r_base_oid, oid);
4192 ceph_oloc_copy(&req->r_base_oloc, oloc);
4193 req->r_flags = CEPH_OSD_FLAG_READ;
4195 ret = ceph_osdc_alloc_messages(req, GFP_KERNEL);
4199 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
4200 if (IS_ERR(pages)) {
4201 ret = PTR_ERR(pages);
4205 osd_req_op_extent_init(req, 0, CEPH_OSD_OP_READ, 0, buf_len, 0, 0);
4206 osd_req_op_extent_osd_data_pages(req, 0, pages, buf_len, 0, false,
4209 ceph_osdc_start_request(osdc, req, false);
4210 ret = ceph_osdc_wait_request(osdc, req);
4212 ceph_copy_from_page_vector(pages, buf, 0, ret);
4215 ceph_osdc_put_request(req);
4220 * Read the complete header for the given rbd device. On successful
4221 * return, the rbd_dev->header field will contain up-to-date
4222 * information about the image.
4224 static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev)
4226 struct rbd_image_header_ondisk *ondisk = NULL;
4233 * The complete header will include an array of its 64-bit
4234 * snapshot ids, followed by the names of those snapshots as
4235 * a contiguous block of NUL-terminated strings. Note that
4236 * the number of snapshots could change by the time we read
4237 * it in, in which case we re-read it.
4244 size = sizeof (*ondisk);
4245 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
4247 ondisk = kmalloc(size, GFP_KERNEL);
4251 ret = rbd_obj_read_sync(rbd_dev, &rbd_dev->header_oid,
4252 &rbd_dev->header_oloc, ondisk, size);
4255 if ((size_t)ret < size) {
4257 rbd_warn(rbd_dev, "short header read (want %zd got %d)",
4261 if (!rbd_dev_ondisk_valid(ondisk)) {
4263 rbd_warn(rbd_dev, "invalid header");
4267 names_size = le64_to_cpu(ondisk->snap_names_len);
4268 want_count = snap_count;
4269 snap_count = le32_to_cpu(ondisk->snap_count);
4270 } while (snap_count != want_count);
4272 ret = rbd_header_from_disk(rbd_dev, ondisk);
4280 * Clear the rbd device's EXISTS flag if the snapshot it's mapped to
4281 * has disappeared from the (just updated) snapshot context.
4283 static void rbd_exists_validate(struct rbd_device *rbd_dev)
4287 if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags))
4290 snap_id = rbd_dev->spec->snap_id;
4291 if (snap_id == CEPH_NOSNAP)
4294 if (rbd_dev_snap_index(rbd_dev, snap_id) == BAD_SNAP_INDEX)
4295 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
4298 static void rbd_dev_update_size(struct rbd_device *rbd_dev)
4303 * If EXISTS is not set, rbd_dev->disk may be NULL, so don't
4304 * try to update its size. If REMOVING is set, updating size
4305 * is just useless work since the device can't be opened.
4307 if (test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags) &&
4308 !test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) {
4309 size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
4310 dout("setting size to %llu sectors", (unsigned long long)size);
4311 set_capacity(rbd_dev->disk, size);
4312 revalidate_disk(rbd_dev->disk);
4316 static int rbd_dev_refresh(struct rbd_device *rbd_dev)
4321 down_write(&rbd_dev->header_rwsem);
4322 mapping_size = rbd_dev->mapping.size;
4324 ret = rbd_dev_header_info(rbd_dev);
4329 * If there is a parent, see if it has disappeared due to the
4330 * mapped image getting flattened.
4332 if (rbd_dev->parent) {
4333 ret = rbd_dev_v2_parent_info(rbd_dev);
4338 if (rbd_dev->spec->snap_id == CEPH_NOSNAP) {
4339 rbd_dev->mapping.size = rbd_dev->header.image_size;
4341 /* validate mapped snapshot's EXISTS flag */
4342 rbd_exists_validate(rbd_dev);
4346 up_write(&rbd_dev->header_rwsem);
4347 if (!ret && mapping_size != rbd_dev->mapping.size)
4348 rbd_dev_update_size(rbd_dev);
4353 static int rbd_init_request(struct blk_mq_tag_set *set, struct request *rq,
4354 unsigned int hctx_idx, unsigned int numa_node)
4356 struct work_struct *work = blk_mq_rq_to_pdu(rq);
4358 INIT_WORK(work, rbd_queue_workfn);
4362 static const struct blk_mq_ops rbd_mq_ops = {
4363 .queue_rq = rbd_queue_rq,
4364 .init_request = rbd_init_request,
4367 static int rbd_init_disk(struct rbd_device *rbd_dev)
4369 struct gendisk *disk;
4370 struct request_queue *q;
4374 /* create gendisk info */
4375 disk = alloc_disk(single_major ?
4376 (1 << RBD_SINGLE_MAJOR_PART_SHIFT) :
4377 RBD_MINORS_PER_MAJOR);
4381 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
4383 disk->major = rbd_dev->major;
4384 disk->first_minor = rbd_dev->minor;
4386 disk->flags |= GENHD_FL_EXT_DEVT;
4387 disk->fops = &rbd_bd_ops;
4388 disk->private_data = rbd_dev;
4390 memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
4391 rbd_dev->tag_set.ops = &rbd_mq_ops;
4392 rbd_dev->tag_set.queue_depth = rbd_dev->opts->queue_depth;
4393 rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
4394 rbd_dev->tag_set.flags = BLK_MQ_F_SHOULD_MERGE | BLK_MQ_F_SG_MERGE;
4395 rbd_dev->tag_set.nr_hw_queues = 1;
4396 rbd_dev->tag_set.cmd_size = sizeof(struct work_struct);
4398 err = blk_mq_alloc_tag_set(&rbd_dev->tag_set);
4402 q = blk_mq_init_queue(&rbd_dev->tag_set);
4408 queue_flag_set_unlocked(QUEUE_FLAG_NONROT, q);
4409 /* QUEUE_FLAG_ADD_RANDOM is off by default for blk-mq */
4411 /* set io sizes to object size */
4412 segment_size = rbd_obj_bytes(&rbd_dev->header);
4413 blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE);
4414 q->limits.max_sectors = queue_max_hw_sectors(q);
4415 blk_queue_max_segments(q, segment_size / SECTOR_SIZE);
4416 blk_queue_max_segment_size(q, segment_size);
4417 blk_queue_io_min(q, segment_size);
4418 blk_queue_io_opt(q, segment_size);
4420 /* enable the discard support */
4421 queue_flag_set_unlocked(QUEUE_FLAG_DISCARD, q);
4422 q->limits.discard_granularity = segment_size;
4423 q->limits.discard_alignment = segment_size;
4424 blk_queue_max_discard_sectors(q, segment_size / SECTOR_SIZE);
4426 if (!ceph_test_opt(rbd_dev->rbd_client->client, NOCRC))
4427 q->backing_dev_info->capabilities |= BDI_CAP_STABLE_WRITES;
4430 * disk_release() expects a queue ref from add_disk() and will
4431 * put it. Hold an extra ref until add_disk() is called.
4433 WARN_ON(!blk_get_queue(q));
4435 q->queuedata = rbd_dev;
4437 rbd_dev->disk = disk;
4441 blk_mq_free_tag_set(&rbd_dev->tag_set);
4451 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
4453 return container_of(dev, struct rbd_device, dev);
4456 static ssize_t rbd_size_show(struct device *dev,
4457 struct device_attribute *attr, char *buf)
4459 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4461 return sprintf(buf, "%llu\n",
4462 (unsigned long long)rbd_dev->mapping.size);
4466 * Note this shows the features for whatever's mapped, which is not
4467 * necessarily the base image.
4469 static ssize_t rbd_features_show(struct device *dev,
4470 struct device_attribute *attr, char *buf)
4472 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4474 return sprintf(buf, "0x%016llx\n",
4475 (unsigned long long)rbd_dev->mapping.features);
4478 static ssize_t rbd_major_show(struct device *dev,
4479 struct device_attribute *attr, char *buf)
4481 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4484 return sprintf(buf, "%d\n", rbd_dev->major);
4486 return sprintf(buf, "(none)\n");
4489 static ssize_t rbd_minor_show(struct device *dev,
4490 struct device_attribute *attr, char *buf)
4492 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4494 return sprintf(buf, "%d\n", rbd_dev->minor);
4497 static ssize_t rbd_client_addr_show(struct device *dev,
4498 struct device_attribute *attr, char *buf)
4500 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4501 struct ceph_entity_addr *client_addr =
4502 ceph_client_addr(rbd_dev->rbd_client->client);
4504 return sprintf(buf, "%pISpc/%u\n", &client_addr->in_addr,
4505 le32_to_cpu(client_addr->nonce));
4508 static ssize_t rbd_client_id_show(struct device *dev,
4509 struct device_attribute *attr, char *buf)
4511 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4513 return sprintf(buf, "client%lld\n",
4514 ceph_client_gid(rbd_dev->rbd_client->client));
4517 static ssize_t rbd_cluster_fsid_show(struct device *dev,
4518 struct device_attribute *attr, char *buf)
4520 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4522 return sprintf(buf, "%pU\n", &rbd_dev->rbd_client->client->fsid);
4525 static ssize_t rbd_config_info_show(struct device *dev,
4526 struct device_attribute *attr, char *buf)
4528 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4530 return sprintf(buf, "%s\n", rbd_dev->config_info);
4533 static ssize_t rbd_pool_show(struct device *dev,
4534 struct device_attribute *attr, char *buf)
4536 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4538 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
4541 static ssize_t rbd_pool_id_show(struct device *dev,
4542 struct device_attribute *attr, char *buf)
4544 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4546 return sprintf(buf, "%llu\n",
4547 (unsigned long long) rbd_dev->spec->pool_id);
4550 static ssize_t rbd_name_show(struct device *dev,
4551 struct device_attribute *attr, char *buf)
4553 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4555 if (rbd_dev->spec->image_name)
4556 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
4558 return sprintf(buf, "(unknown)\n");
4561 static ssize_t rbd_image_id_show(struct device *dev,
4562 struct device_attribute *attr, char *buf)
4564 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4566 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
4570 * Shows the name of the currently-mapped snapshot (or
4571 * RBD_SNAP_HEAD_NAME for the base image).
4573 static ssize_t rbd_snap_show(struct device *dev,
4574 struct device_attribute *attr,
4577 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4579 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
4582 static ssize_t rbd_snap_id_show(struct device *dev,
4583 struct device_attribute *attr, char *buf)
4585 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4587 return sprintf(buf, "%llu\n", rbd_dev->spec->snap_id);
4591 * For a v2 image, shows the chain of parent images, separated by empty
4592 * lines. For v1 images or if there is no parent, shows "(no parent
4595 static ssize_t rbd_parent_show(struct device *dev,
4596 struct device_attribute *attr,
4599 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4602 if (!rbd_dev->parent)
4603 return sprintf(buf, "(no parent image)\n");
4605 for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) {
4606 struct rbd_spec *spec = rbd_dev->parent_spec;
4608 count += sprintf(&buf[count], "%s"
4609 "pool_id %llu\npool_name %s\n"
4610 "image_id %s\nimage_name %s\n"
4611 "snap_id %llu\nsnap_name %s\n"
4613 !count ? "" : "\n", /* first? */
4614 spec->pool_id, spec->pool_name,
4615 spec->image_id, spec->image_name ?: "(unknown)",
4616 spec->snap_id, spec->snap_name,
4617 rbd_dev->parent_overlap);
4623 static ssize_t rbd_image_refresh(struct device *dev,
4624 struct device_attribute *attr,
4628 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4631 ret = rbd_dev_refresh(rbd_dev);
4638 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
4639 static DEVICE_ATTR(features, S_IRUGO, rbd_features_show, NULL);
4640 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
4641 static DEVICE_ATTR(minor, S_IRUGO, rbd_minor_show, NULL);
4642 static DEVICE_ATTR(client_addr, S_IRUGO, rbd_client_addr_show, NULL);
4643 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
4644 static DEVICE_ATTR(cluster_fsid, S_IRUGO, rbd_cluster_fsid_show, NULL);
4645 static DEVICE_ATTR(config_info, S_IRUSR, rbd_config_info_show, NULL);
4646 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
4647 static DEVICE_ATTR(pool_id, S_IRUGO, rbd_pool_id_show, NULL);
4648 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
4649 static DEVICE_ATTR(image_id, S_IRUGO, rbd_image_id_show, NULL);
4650 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
4651 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
4652 static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
4653 static DEVICE_ATTR(parent, S_IRUGO, rbd_parent_show, NULL);
4655 static struct attribute *rbd_attrs[] = {
4656 &dev_attr_size.attr,
4657 &dev_attr_features.attr,
4658 &dev_attr_major.attr,
4659 &dev_attr_minor.attr,
4660 &dev_attr_client_addr.attr,
4661 &dev_attr_client_id.attr,
4662 &dev_attr_cluster_fsid.attr,
4663 &dev_attr_config_info.attr,
4664 &dev_attr_pool.attr,
4665 &dev_attr_pool_id.attr,
4666 &dev_attr_name.attr,
4667 &dev_attr_image_id.attr,
4668 &dev_attr_current_snap.attr,
4669 &dev_attr_snap_id.attr,
4670 &dev_attr_parent.attr,
4671 &dev_attr_refresh.attr,
4675 static struct attribute_group rbd_attr_group = {
4679 static const struct attribute_group *rbd_attr_groups[] = {
4684 static void rbd_dev_release(struct device *dev);
4686 static const struct device_type rbd_device_type = {
4688 .groups = rbd_attr_groups,
4689 .release = rbd_dev_release,
4692 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
4694 kref_get(&spec->kref);
4699 static void rbd_spec_free(struct kref *kref);
4700 static void rbd_spec_put(struct rbd_spec *spec)
4703 kref_put(&spec->kref, rbd_spec_free);
4706 static struct rbd_spec *rbd_spec_alloc(void)
4708 struct rbd_spec *spec;
4710 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
4714 spec->pool_id = CEPH_NOPOOL;
4715 spec->snap_id = CEPH_NOSNAP;
4716 kref_init(&spec->kref);
4721 static void rbd_spec_free(struct kref *kref)
4723 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
4725 kfree(spec->pool_name);
4726 kfree(spec->image_id);
4727 kfree(spec->image_name);
4728 kfree(spec->snap_name);
4732 static void rbd_dev_free(struct rbd_device *rbd_dev)
4734 WARN_ON(rbd_dev->watch_state != RBD_WATCH_STATE_UNREGISTERED);
4735 WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_UNLOCKED);
4737 ceph_oid_destroy(&rbd_dev->header_oid);
4738 ceph_oloc_destroy(&rbd_dev->header_oloc);
4739 kfree(rbd_dev->config_info);
4741 rbd_put_client(rbd_dev->rbd_client);
4742 rbd_spec_put(rbd_dev->spec);
4743 kfree(rbd_dev->opts);
4747 static void rbd_dev_release(struct device *dev)
4749 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4750 bool need_put = !!rbd_dev->opts;
4753 destroy_workqueue(rbd_dev->task_wq);
4754 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
4757 rbd_dev_free(rbd_dev);
4760 * This is racy, but way better than putting module outside of
4761 * the release callback. The race window is pretty small, so
4762 * doing something similar to dm (dm-builtin.c) is overkill.
4765 module_put(THIS_MODULE);
4768 static struct rbd_device *__rbd_dev_create(struct rbd_client *rbdc,
4769 struct rbd_spec *spec)
4771 struct rbd_device *rbd_dev;
4773 rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
4777 spin_lock_init(&rbd_dev->lock);
4778 INIT_LIST_HEAD(&rbd_dev->node);
4779 init_rwsem(&rbd_dev->header_rwsem);
4781 rbd_dev->header.data_pool_id = CEPH_NOPOOL;
4782 ceph_oid_init(&rbd_dev->header_oid);
4783 rbd_dev->header_oloc.pool = spec->pool_id;
4785 mutex_init(&rbd_dev->watch_mutex);
4786 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
4787 INIT_DELAYED_WORK(&rbd_dev->watch_dwork, rbd_reregister_watch);
4789 init_rwsem(&rbd_dev->lock_rwsem);
4790 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
4791 INIT_WORK(&rbd_dev->acquired_lock_work, rbd_notify_acquired_lock);
4792 INIT_WORK(&rbd_dev->released_lock_work, rbd_notify_released_lock);
4793 INIT_DELAYED_WORK(&rbd_dev->lock_dwork, rbd_acquire_lock);
4794 INIT_WORK(&rbd_dev->unlock_work, rbd_release_lock_work);
4795 init_waitqueue_head(&rbd_dev->lock_waitq);
4797 rbd_dev->dev.bus = &rbd_bus_type;
4798 rbd_dev->dev.type = &rbd_device_type;
4799 rbd_dev->dev.parent = &rbd_root_dev;
4800 device_initialize(&rbd_dev->dev);
4802 rbd_dev->rbd_client = rbdc;
4803 rbd_dev->spec = spec;
4809 * Create a mapping rbd_dev.
4811 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
4812 struct rbd_spec *spec,
4813 struct rbd_options *opts)
4815 struct rbd_device *rbd_dev;
4817 rbd_dev = __rbd_dev_create(rbdc, spec);
4821 rbd_dev->opts = opts;
4823 /* get an id and fill in device name */
4824 rbd_dev->dev_id = ida_simple_get(&rbd_dev_id_ida, 0,
4825 minor_to_rbd_dev_id(1 << MINORBITS),
4827 if (rbd_dev->dev_id < 0)
4830 sprintf(rbd_dev->name, RBD_DRV_NAME "%d", rbd_dev->dev_id);
4831 rbd_dev->task_wq = alloc_ordered_workqueue("%s-tasks", WQ_MEM_RECLAIM,
4833 if (!rbd_dev->task_wq)
4836 /* we have a ref from do_rbd_add() */
4837 __module_get(THIS_MODULE);
4839 dout("%s rbd_dev %p dev_id %d\n", __func__, rbd_dev, rbd_dev->dev_id);
4843 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
4845 rbd_dev_free(rbd_dev);
4849 static void rbd_dev_destroy(struct rbd_device *rbd_dev)
4852 put_device(&rbd_dev->dev);
4856 * Get the size and object order for an image snapshot, or if
4857 * snap_id is CEPH_NOSNAP, gets this information for the base
4860 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
4861 u8 *order, u64 *snap_size)
4863 __le64 snapid = cpu_to_le64(snap_id);
4868 } __attribute__ ((packed)) size_buf = { 0 };
4870 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
4871 &rbd_dev->header_oloc, "get_size",
4872 &snapid, sizeof(snapid),
4873 &size_buf, sizeof(size_buf));
4874 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4877 if (ret < sizeof (size_buf))
4881 *order = size_buf.order;
4882 dout(" order %u", (unsigned int)*order);
4884 *snap_size = le64_to_cpu(size_buf.size);
4886 dout(" snap_id 0x%016llx snap_size = %llu\n",
4887 (unsigned long long)snap_id,
4888 (unsigned long long)*snap_size);
4893 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
4895 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
4896 &rbd_dev->header.obj_order,
4897 &rbd_dev->header.image_size);
4900 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
4906 reply_buf = kzalloc(RBD_OBJ_PREFIX_LEN_MAX, GFP_KERNEL);
4910 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
4911 &rbd_dev->header_oloc, "get_object_prefix",
4912 NULL, 0, reply_buf, RBD_OBJ_PREFIX_LEN_MAX);
4913 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4918 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
4919 p + ret, NULL, GFP_NOIO);
4922 if (IS_ERR(rbd_dev->header.object_prefix)) {
4923 ret = PTR_ERR(rbd_dev->header.object_prefix);
4924 rbd_dev->header.object_prefix = NULL;
4926 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix);
4934 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
4937 __le64 snapid = cpu_to_le64(snap_id);
4941 } __attribute__ ((packed)) features_buf = { 0 };
4945 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
4946 &rbd_dev->header_oloc, "get_features",
4947 &snapid, sizeof(snapid),
4948 &features_buf, sizeof(features_buf));
4949 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
4952 if (ret < sizeof (features_buf))
4955 unsup = le64_to_cpu(features_buf.incompat) & ~RBD_FEATURES_SUPPORTED;
4957 rbd_warn(rbd_dev, "image uses unsupported features: 0x%llx",
4962 *snap_features = le64_to_cpu(features_buf.features);
4964 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
4965 (unsigned long long)snap_id,
4966 (unsigned long long)*snap_features,
4967 (unsigned long long)le64_to_cpu(features_buf.incompat));
4972 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
4974 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
4975 &rbd_dev->header.features);
4978 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
4980 struct rbd_spec *parent_spec;
4982 void *reply_buf = NULL;
4992 parent_spec = rbd_spec_alloc();
4996 size = sizeof (__le64) + /* pool_id */
4997 sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX + /* image_id */
4998 sizeof (__le64) + /* snap_id */
4999 sizeof (__le64); /* overlap */
5000 reply_buf = kmalloc(size, GFP_KERNEL);
5006 snapid = cpu_to_le64(rbd_dev->spec->snap_id);
5007 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5008 &rbd_dev->header_oloc, "get_parent",
5009 &snapid, sizeof(snapid), reply_buf, size);
5010 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5015 end = reply_buf + ret;
5017 ceph_decode_64_safe(&p, end, pool_id, out_err);
5018 if (pool_id == CEPH_NOPOOL) {
5020 * Either the parent never existed, or we have
5021 * record of it but the image got flattened so it no
5022 * longer has a parent. When the parent of a
5023 * layered image disappears we immediately set the
5024 * overlap to 0. The effect of this is that all new
5025 * requests will be treated as if the image had no
5028 if (rbd_dev->parent_overlap) {
5029 rbd_dev->parent_overlap = 0;
5030 rbd_dev_parent_put(rbd_dev);
5031 pr_info("%s: clone image has been flattened\n",
5032 rbd_dev->disk->disk_name);
5035 goto out; /* No parent? No problem. */
5038 /* The ceph file layout needs to fit pool id in 32 bits */
5041 if (pool_id > (u64)U32_MAX) {
5042 rbd_warn(NULL, "parent pool id too large (%llu > %u)",
5043 (unsigned long long)pool_id, U32_MAX);
5047 image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
5048 if (IS_ERR(image_id)) {
5049 ret = PTR_ERR(image_id);
5052 ceph_decode_64_safe(&p, end, snap_id, out_err);
5053 ceph_decode_64_safe(&p, end, overlap, out_err);
5056 * The parent won't change (except when the clone is
5057 * flattened, already handled that). So we only need to
5058 * record the parent spec we have not already done so.
5060 if (!rbd_dev->parent_spec) {
5061 parent_spec->pool_id = pool_id;
5062 parent_spec->image_id = image_id;
5063 parent_spec->snap_id = snap_id;
5064 rbd_dev->parent_spec = parent_spec;
5065 parent_spec = NULL; /* rbd_dev now owns this */
5071 * We always update the parent overlap. If it's zero we issue
5072 * a warning, as we will proceed as if there was no parent.
5076 /* refresh, careful to warn just once */
5077 if (rbd_dev->parent_overlap)
5079 "clone now standalone (overlap became 0)");
5082 rbd_warn(rbd_dev, "clone is standalone (overlap 0)");
5085 rbd_dev->parent_overlap = overlap;
5091 rbd_spec_put(parent_spec);
5096 static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev)
5100 __le64 stripe_count;
5101 } __attribute__ ((packed)) striping_info_buf = { 0 };
5102 size_t size = sizeof (striping_info_buf);
5109 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5110 &rbd_dev->header_oloc, "get_stripe_unit_count",
5111 NULL, 0, &striping_info_buf, size);
5112 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5119 * We don't actually support the "fancy striping" feature
5120 * (STRIPINGV2) yet, but if the striping sizes are the
5121 * defaults the behavior is the same as before. So find
5122 * out, and only fail if the image has non-default values.
5125 obj_size = rbd_obj_bytes(&rbd_dev->header);
5126 p = &striping_info_buf;
5127 stripe_unit = ceph_decode_64(&p);
5128 if (stripe_unit != obj_size) {
5129 rbd_warn(rbd_dev, "unsupported stripe unit "
5130 "(got %llu want %llu)",
5131 stripe_unit, obj_size);
5134 stripe_count = ceph_decode_64(&p);
5135 if (stripe_count != 1) {
5136 rbd_warn(rbd_dev, "unsupported stripe count "
5137 "(got %llu want 1)", stripe_count);
5140 rbd_dev->header.stripe_unit = stripe_unit;
5141 rbd_dev->header.stripe_count = stripe_count;
5146 static int rbd_dev_v2_data_pool(struct rbd_device *rbd_dev)
5148 __le64 data_pool_id;
5151 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5152 &rbd_dev->header_oloc, "get_data_pool",
5153 NULL, 0, &data_pool_id, sizeof(data_pool_id));
5156 if (ret < sizeof(data_pool_id))
5159 rbd_dev->header.data_pool_id = le64_to_cpu(data_pool_id);
5160 WARN_ON(rbd_dev->header.data_pool_id == CEPH_NOPOOL);
5164 static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
5166 CEPH_DEFINE_OID_ONSTACK(oid);
5167 size_t image_id_size;
5172 void *reply_buf = NULL;
5174 char *image_name = NULL;
5177 rbd_assert(!rbd_dev->spec->image_name);
5179 len = strlen(rbd_dev->spec->image_id);
5180 image_id_size = sizeof (__le32) + len;
5181 image_id = kmalloc(image_id_size, GFP_KERNEL);
5186 end = image_id + image_id_size;
5187 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len);
5189 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
5190 reply_buf = kmalloc(size, GFP_KERNEL);
5194 ceph_oid_printf(&oid, "%s", RBD_DIRECTORY);
5195 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
5196 "dir_get_name", image_id, image_id_size,
5201 end = reply_buf + ret;
5203 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
5204 if (IS_ERR(image_name))
5207 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
5215 static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5217 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5218 const char *snap_name;
5221 /* Skip over names until we find the one we are looking for */
5223 snap_name = rbd_dev->header.snap_names;
5224 while (which < snapc->num_snaps) {
5225 if (!strcmp(name, snap_name))
5226 return snapc->snaps[which];
5227 snap_name += strlen(snap_name) + 1;
5233 static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5235 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5240 for (which = 0; !found && which < snapc->num_snaps; which++) {
5241 const char *snap_name;
5243 snap_id = snapc->snaps[which];
5244 snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
5245 if (IS_ERR(snap_name)) {
5246 /* ignore no-longer existing snapshots */
5247 if (PTR_ERR(snap_name) == -ENOENT)
5252 found = !strcmp(name, snap_name);
5255 return found ? snap_id : CEPH_NOSNAP;
5259 * Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if
5260 * no snapshot by that name is found, or if an error occurs.
5262 static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5264 if (rbd_dev->image_format == 1)
5265 return rbd_v1_snap_id_by_name(rbd_dev, name);
5267 return rbd_v2_snap_id_by_name(rbd_dev, name);
5271 * An image being mapped will have everything but the snap id.
5273 static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev)
5275 struct rbd_spec *spec = rbd_dev->spec;
5277 rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name);
5278 rbd_assert(spec->image_id && spec->image_name);
5279 rbd_assert(spec->snap_name);
5281 if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) {
5284 snap_id = rbd_snap_id_by_name(rbd_dev, spec->snap_name);
5285 if (snap_id == CEPH_NOSNAP)
5288 spec->snap_id = snap_id;
5290 spec->snap_id = CEPH_NOSNAP;
5297 * A parent image will have all ids but none of the names.
5299 * All names in an rbd spec are dynamically allocated. It's OK if we
5300 * can't figure out the name for an image id.
5302 static int rbd_spec_fill_names(struct rbd_device *rbd_dev)
5304 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5305 struct rbd_spec *spec = rbd_dev->spec;
5306 const char *pool_name;
5307 const char *image_name;
5308 const char *snap_name;
5311 rbd_assert(spec->pool_id != CEPH_NOPOOL);
5312 rbd_assert(spec->image_id);
5313 rbd_assert(spec->snap_id != CEPH_NOSNAP);
5315 /* Get the pool name; we have to make our own copy of this */
5317 pool_name = ceph_pg_pool_name_by_id(osdc->osdmap, spec->pool_id);
5319 rbd_warn(rbd_dev, "no pool with id %llu", spec->pool_id);
5322 pool_name = kstrdup(pool_name, GFP_KERNEL);
5326 /* Fetch the image name; tolerate failure here */
5328 image_name = rbd_dev_image_name(rbd_dev);
5330 rbd_warn(rbd_dev, "unable to get image name");
5332 /* Fetch the snapshot name */
5334 snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
5335 if (IS_ERR(snap_name)) {
5336 ret = PTR_ERR(snap_name);
5340 spec->pool_name = pool_name;
5341 spec->image_name = image_name;
5342 spec->snap_name = snap_name;
5352 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev)
5361 struct ceph_snap_context *snapc;
5365 * We'll need room for the seq value (maximum snapshot id),
5366 * snapshot count, and array of that many snapshot ids.
5367 * For now we have a fixed upper limit on the number we're
5368 * prepared to receive.
5370 size = sizeof (__le64) + sizeof (__le32) +
5371 RBD_MAX_SNAP_COUNT * sizeof (__le64);
5372 reply_buf = kzalloc(size, GFP_KERNEL);
5376 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5377 &rbd_dev->header_oloc, "get_snapcontext",
5378 NULL, 0, reply_buf, size);
5379 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5384 end = reply_buf + ret;
5386 ceph_decode_64_safe(&p, end, seq, out);
5387 ceph_decode_32_safe(&p, end, snap_count, out);
5390 * Make sure the reported number of snapshot ids wouldn't go
5391 * beyond the end of our buffer. But before checking that,
5392 * make sure the computed size of the snapshot context we
5393 * allocate is representable in a size_t.
5395 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
5400 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
5404 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
5410 for (i = 0; i < snap_count; i++)
5411 snapc->snaps[i] = ceph_decode_64(&p);
5413 ceph_put_snap_context(rbd_dev->header.snapc);
5414 rbd_dev->header.snapc = snapc;
5416 dout(" snap context seq = %llu, snap_count = %u\n",
5417 (unsigned long long)seq, (unsigned int)snap_count);
5424 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
5435 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
5436 reply_buf = kmalloc(size, GFP_KERNEL);
5438 return ERR_PTR(-ENOMEM);
5440 snapid = cpu_to_le64(snap_id);
5441 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5442 &rbd_dev->header_oloc, "get_snapshot_name",
5443 &snapid, sizeof(snapid), reply_buf, size);
5444 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5446 snap_name = ERR_PTR(ret);
5451 end = reply_buf + ret;
5452 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
5453 if (IS_ERR(snap_name))
5456 dout(" snap_id 0x%016llx snap_name = %s\n",
5457 (unsigned long long)snap_id, snap_name);
5464 static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev)
5466 bool first_time = rbd_dev->header.object_prefix == NULL;
5469 ret = rbd_dev_v2_image_size(rbd_dev);
5474 ret = rbd_dev_v2_header_onetime(rbd_dev);
5479 ret = rbd_dev_v2_snap_context(rbd_dev);
5480 if (ret && first_time) {
5481 kfree(rbd_dev->header.object_prefix);
5482 rbd_dev->header.object_prefix = NULL;
5488 static int rbd_dev_header_info(struct rbd_device *rbd_dev)
5490 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
5492 if (rbd_dev->image_format == 1)
5493 return rbd_dev_v1_header_info(rbd_dev);
5495 return rbd_dev_v2_header_info(rbd_dev);
5499 * Skips over white space at *buf, and updates *buf to point to the
5500 * first found non-space character (if any). Returns the length of
5501 * the token (string of non-white space characters) found. Note
5502 * that *buf must be terminated with '\0'.
5504 static inline size_t next_token(const char **buf)
5507 * These are the characters that produce nonzero for
5508 * isspace() in the "C" and "POSIX" locales.
5510 const char *spaces = " \f\n\r\t\v";
5512 *buf += strspn(*buf, spaces); /* Find start of token */
5514 return strcspn(*buf, spaces); /* Return token length */
5518 * Finds the next token in *buf, dynamically allocates a buffer big
5519 * enough to hold a copy of it, and copies the token into the new
5520 * buffer. The copy is guaranteed to be terminated with '\0'. Note
5521 * that a duplicate buffer is created even for a zero-length token.
5523 * Returns a pointer to the newly-allocated duplicate, or a null
5524 * pointer if memory for the duplicate was not available. If
5525 * the lenp argument is a non-null pointer, the length of the token
5526 * (not including the '\0') is returned in *lenp.
5528 * If successful, the *buf pointer will be updated to point beyond
5529 * the end of the found token.
5531 * Note: uses GFP_KERNEL for allocation.
5533 static inline char *dup_token(const char **buf, size_t *lenp)
5538 len = next_token(buf);
5539 dup = kmemdup(*buf, len + 1, GFP_KERNEL);
5542 *(dup + len) = '\0';
5552 * Parse the options provided for an "rbd add" (i.e., rbd image
5553 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
5554 * and the data written is passed here via a NUL-terminated buffer.
5555 * Returns 0 if successful or an error code otherwise.
5557 * The information extracted from these options is recorded in
5558 * the other parameters which return dynamically-allocated
5561 * The address of a pointer that will refer to a ceph options
5562 * structure. Caller must release the returned pointer using
5563 * ceph_destroy_options() when it is no longer needed.
5565 * Address of an rbd options pointer. Fully initialized by
5566 * this function; caller must release with kfree().
5568 * Address of an rbd image specification pointer. Fully
5569 * initialized by this function based on parsed options.
5570 * Caller must release with rbd_spec_put().
5572 * The options passed take this form:
5573 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
5576 * A comma-separated list of one or more monitor addresses.
5577 * A monitor address is an ip address, optionally followed
5578 * by a port number (separated by a colon).
5579 * I.e.: ip1[:port1][,ip2[:port2]...]
5581 * A comma-separated list of ceph and/or rbd options.
5583 * The name of the rados pool containing the rbd image.
5585 * The name of the image in that pool to map.
5587 * An optional snapshot id. If provided, the mapping will
5588 * present data from the image at the time that snapshot was
5589 * created. The image head is used if no snapshot id is
5590 * provided. Snapshot mappings are always read-only.
5592 static int rbd_add_parse_args(const char *buf,
5593 struct ceph_options **ceph_opts,
5594 struct rbd_options **opts,
5595 struct rbd_spec **rbd_spec)
5599 const char *mon_addrs;
5601 size_t mon_addrs_size;
5602 struct rbd_spec *spec = NULL;
5603 struct rbd_options *rbd_opts = NULL;
5604 struct ceph_options *copts;
5607 /* The first four tokens are required */
5609 len = next_token(&buf);
5611 rbd_warn(NULL, "no monitor address(es) provided");
5615 mon_addrs_size = len + 1;
5619 options = dup_token(&buf, NULL);
5623 rbd_warn(NULL, "no options provided");
5627 spec = rbd_spec_alloc();
5631 spec->pool_name = dup_token(&buf, NULL);
5632 if (!spec->pool_name)
5634 if (!*spec->pool_name) {
5635 rbd_warn(NULL, "no pool name provided");
5639 spec->image_name = dup_token(&buf, NULL);
5640 if (!spec->image_name)
5642 if (!*spec->image_name) {
5643 rbd_warn(NULL, "no image name provided");
5648 * Snapshot name is optional; default is to use "-"
5649 * (indicating the head/no snapshot).
5651 len = next_token(&buf);
5653 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
5654 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
5655 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
5656 ret = -ENAMETOOLONG;
5659 snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
5662 *(snap_name + len) = '\0';
5663 spec->snap_name = snap_name;
5665 /* Initialize all rbd options to the defaults */
5667 rbd_opts = kzalloc(sizeof (*rbd_opts), GFP_KERNEL);
5671 rbd_opts->read_only = RBD_READ_ONLY_DEFAULT;
5672 rbd_opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
5673 rbd_opts->lock_on_read = RBD_LOCK_ON_READ_DEFAULT;
5674 rbd_opts->exclusive = RBD_EXCLUSIVE_DEFAULT;
5676 copts = ceph_parse_options(options, mon_addrs,
5677 mon_addrs + mon_addrs_size - 1,
5678 parse_rbd_opts_token, rbd_opts);
5679 if (IS_ERR(copts)) {
5680 ret = PTR_ERR(copts);
5701 * Return pool id (>= 0) or a negative error code.
5703 static int rbd_add_get_pool_id(struct rbd_client *rbdc, const char *pool_name)
5705 struct ceph_options *opts = rbdc->client->options;
5711 ret = ceph_pg_poolid_by_name(rbdc->client->osdc.osdmap, pool_name);
5712 if (ret == -ENOENT && tries++ < 1) {
5713 ret = ceph_monc_get_version(&rbdc->client->monc, "osdmap",
5718 if (rbdc->client->osdc.osdmap->epoch < newest_epoch) {
5719 ceph_osdc_maybe_request_map(&rbdc->client->osdc);
5720 (void) ceph_monc_wait_osdmap(&rbdc->client->monc,
5722 opts->mount_timeout);
5725 /* the osdmap we have is new enough */
5733 static void rbd_dev_image_unlock(struct rbd_device *rbd_dev)
5735 down_write(&rbd_dev->lock_rwsem);
5736 if (__rbd_is_lock_owner(rbd_dev))
5737 rbd_unlock(rbd_dev);
5738 up_write(&rbd_dev->lock_rwsem);
5741 static int rbd_add_acquire_lock(struct rbd_device *rbd_dev)
5743 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK)) {
5744 rbd_warn(rbd_dev, "exclusive-lock feature is not enabled");
5748 /* FIXME: "rbd map --exclusive" should be in interruptible */
5749 down_read(&rbd_dev->lock_rwsem);
5750 rbd_wait_state_locked(rbd_dev);
5751 up_read(&rbd_dev->lock_rwsem);
5752 if (test_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags)) {
5753 rbd_warn(rbd_dev, "failed to acquire exclusive lock");
5761 * An rbd format 2 image has a unique identifier, distinct from the
5762 * name given to it by the user. Internally, that identifier is
5763 * what's used to specify the names of objects related to the image.
5765 * A special "rbd id" object is used to map an rbd image name to its
5766 * id. If that object doesn't exist, then there is no v2 rbd image
5767 * with the supplied name.
5769 * This function will record the given rbd_dev's image_id field if
5770 * it can be determined, and in that case will return 0. If any
5771 * errors occur a negative errno will be returned and the rbd_dev's
5772 * image_id field will be unchanged (and should be NULL).
5774 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
5778 CEPH_DEFINE_OID_ONSTACK(oid);
5783 * When probing a parent image, the image id is already
5784 * known (and the image name likely is not). There's no
5785 * need to fetch the image id again in this case. We
5786 * do still need to set the image format though.
5788 if (rbd_dev->spec->image_id) {
5789 rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
5795 * First, see if the format 2 image id file exists, and if
5796 * so, get the image's persistent id from it.
5798 ret = ceph_oid_aprintf(&oid, GFP_KERNEL, "%s%s", RBD_ID_PREFIX,
5799 rbd_dev->spec->image_name);
5803 dout("rbd id object name is %s\n", oid.name);
5805 /* Response will be an encoded string, which includes a length */
5807 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
5808 response = kzalloc(size, GFP_NOIO);
5814 /* If it doesn't exist we'll assume it's a format 1 image */
5816 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
5818 response, RBD_IMAGE_ID_LEN_MAX);
5819 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5820 if (ret == -ENOENT) {
5821 image_id = kstrdup("", GFP_KERNEL);
5822 ret = image_id ? 0 : -ENOMEM;
5824 rbd_dev->image_format = 1;
5825 } else if (ret >= 0) {
5828 image_id = ceph_extract_encoded_string(&p, p + ret,
5830 ret = PTR_ERR_OR_ZERO(image_id);
5832 rbd_dev->image_format = 2;
5836 rbd_dev->spec->image_id = image_id;
5837 dout("image_id is %s\n", image_id);
5841 ceph_oid_destroy(&oid);
5846 * Undo whatever state changes are made by v1 or v2 header info
5849 static void rbd_dev_unprobe(struct rbd_device *rbd_dev)
5851 struct rbd_image_header *header;
5853 rbd_dev_parent_put(rbd_dev);
5855 /* Free dynamic fields from the header, then zero it out */
5857 header = &rbd_dev->header;
5858 ceph_put_snap_context(header->snapc);
5859 kfree(header->snap_sizes);
5860 kfree(header->snap_names);
5861 kfree(header->object_prefix);
5862 memset(header, 0, sizeof (*header));
5865 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev)
5869 ret = rbd_dev_v2_object_prefix(rbd_dev);
5874 * Get the and check features for the image. Currently the
5875 * features are assumed to never change.
5877 ret = rbd_dev_v2_features(rbd_dev);
5881 /* If the image supports fancy striping, get its parameters */
5883 if (rbd_dev->header.features & RBD_FEATURE_STRIPINGV2) {
5884 ret = rbd_dev_v2_striping_info(rbd_dev);
5889 if (rbd_dev->header.features & RBD_FEATURE_DATA_POOL) {
5890 ret = rbd_dev_v2_data_pool(rbd_dev);
5895 rbd_init_layout(rbd_dev);
5899 rbd_dev->header.features = 0;
5900 kfree(rbd_dev->header.object_prefix);
5901 rbd_dev->header.object_prefix = NULL;
5906 * @depth is rbd_dev_image_probe() -> rbd_dev_probe_parent() ->
5907 * rbd_dev_image_probe() recursion depth, which means it's also the
5908 * length of the already discovered part of the parent chain.
5910 static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
5912 struct rbd_device *parent = NULL;
5915 if (!rbd_dev->parent_spec)
5918 if (++depth > RBD_MAX_PARENT_CHAIN_LEN) {
5919 pr_info("parent chain is too long (%d)\n", depth);
5924 parent = __rbd_dev_create(rbd_dev->rbd_client, rbd_dev->parent_spec);
5931 * Images related by parent/child relationships always share
5932 * rbd_client and spec/parent_spec, so bump their refcounts.
5934 __rbd_get_client(rbd_dev->rbd_client);
5935 rbd_spec_get(rbd_dev->parent_spec);
5937 ret = rbd_dev_image_probe(parent, depth);
5941 rbd_dev->parent = parent;
5942 atomic_set(&rbd_dev->parent_ref, 1);
5946 rbd_dev_unparent(rbd_dev);
5947 rbd_dev_destroy(parent);
5951 static void rbd_dev_device_release(struct rbd_device *rbd_dev)
5953 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
5954 rbd_dev_mapping_clear(rbd_dev);
5955 rbd_free_disk(rbd_dev);
5957 unregister_blkdev(rbd_dev->major, rbd_dev->name);
5961 * rbd_dev->header_rwsem must be locked for write and will be unlocked
5964 static int rbd_dev_device_setup(struct rbd_device *rbd_dev)
5968 /* Record our major and minor device numbers. */
5970 if (!single_major) {
5971 ret = register_blkdev(0, rbd_dev->name);
5973 goto err_out_unlock;
5975 rbd_dev->major = ret;
5978 rbd_dev->major = rbd_major;
5979 rbd_dev->minor = rbd_dev_id_to_minor(rbd_dev->dev_id);
5982 /* Set up the blkdev mapping. */
5984 ret = rbd_init_disk(rbd_dev);
5986 goto err_out_blkdev;
5988 ret = rbd_dev_mapping_set(rbd_dev);
5992 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
5993 set_disk_ro(rbd_dev->disk, rbd_dev->mapping.read_only);
5995 ret = dev_set_name(&rbd_dev->dev, "%d", rbd_dev->dev_id);
5997 goto err_out_mapping;
5999 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6000 up_write(&rbd_dev->header_rwsem);
6004 rbd_dev_mapping_clear(rbd_dev);
6006 rbd_free_disk(rbd_dev);
6009 unregister_blkdev(rbd_dev->major, rbd_dev->name);
6011 up_write(&rbd_dev->header_rwsem);
6015 static int rbd_dev_header_name(struct rbd_device *rbd_dev)
6017 struct rbd_spec *spec = rbd_dev->spec;
6020 /* Record the header object name for this rbd image. */
6022 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6023 if (rbd_dev->image_format == 1)
6024 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6025 spec->image_name, RBD_SUFFIX);
6027 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6028 RBD_HEADER_PREFIX, spec->image_id);
6033 static void rbd_dev_image_release(struct rbd_device *rbd_dev)
6035 rbd_dev_unprobe(rbd_dev);
6037 rbd_unregister_watch(rbd_dev);
6038 rbd_dev->image_format = 0;
6039 kfree(rbd_dev->spec->image_id);
6040 rbd_dev->spec->image_id = NULL;
6044 * Probe for the existence of the header object for the given rbd
6045 * device. If this image is the one being mapped (i.e., not a
6046 * parent), initiate a watch on its header object before using that
6047 * object to get detailed information about the rbd image.
6049 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
6054 * Get the id from the image id object. Unless there's an
6055 * error, rbd_dev->spec->image_id will be filled in with
6056 * a dynamically-allocated string, and rbd_dev->image_format
6057 * will be set to either 1 or 2.
6059 ret = rbd_dev_image_id(rbd_dev);
6063 ret = rbd_dev_header_name(rbd_dev);
6065 goto err_out_format;
6068 ret = rbd_register_watch(rbd_dev);
6071 pr_info("image %s/%s does not exist\n",
6072 rbd_dev->spec->pool_name,
6073 rbd_dev->spec->image_name);
6074 goto err_out_format;
6078 ret = rbd_dev_header_info(rbd_dev);
6083 * If this image is the one being mapped, we have pool name and
6084 * id, image name and id, and snap name - need to fill snap id.
6085 * Otherwise this is a parent image, identified by pool, image
6086 * and snap ids - need to fill in names for those ids.
6089 ret = rbd_spec_fill_snap_id(rbd_dev);
6091 ret = rbd_spec_fill_names(rbd_dev);
6094 pr_info("snap %s/%s@%s does not exist\n",
6095 rbd_dev->spec->pool_name,
6096 rbd_dev->spec->image_name,
6097 rbd_dev->spec->snap_name);
6101 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
6102 ret = rbd_dev_v2_parent_info(rbd_dev);
6107 * Need to warn users if this image is the one being
6108 * mapped and has a parent.
6110 if (!depth && rbd_dev->parent_spec)
6112 "WARNING: kernel layering is EXPERIMENTAL!");
6115 ret = rbd_dev_probe_parent(rbd_dev, depth);
6119 dout("discovered format %u image, header name is %s\n",
6120 rbd_dev->image_format, rbd_dev->header_oid.name);
6124 rbd_dev_unprobe(rbd_dev);
6127 rbd_unregister_watch(rbd_dev);
6129 rbd_dev->image_format = 0;
6130 kfree(rbd_dev->spec->image_id);
6131 rbd_dev->spec->image_id = NULL;
6135 static ssize_t do_rbd_add(struct bus_type *bus,
6139 struct rbd_device *rbd_dev = NULL;
6140 struct ceph_options *ceph_opts = NULL;
6141 struct rbd_options *rbd_opts = NULL;
6142 struct rbd_spec *spec = NULL;
6143 struct rbd_client *rbdc;
6147 if (!try_module_get(THIS_MODULE))
6150 /* parse add command */
6151 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
6155 rbdc = rbd_get_client(ceph_opts);
6162 rc = rbd_add_get_pool_id(rbdc, spec->pool_name);
6165 pr_info("pool %s does not exist\n", spec->pool_name);
6166 goto err_out_client;
6168 spec->pool_id = (u64)rc;
6170 rbd_dev = rbd_dev_create(rbdc, spec, rbd_opts);
6173 goto err_out_client;
6175 rbdc = NULL; /* rbd_dev now owns this */
6176 spec = NULL; /* rbd_dev now owns this */
6177 rbd_opts = NULL; /* rbd_dev now owns this */
6179 rbd_dev->config_info = kstrdup(buf, GFP_KERNEL);
6180 if (!rbd_dev->config_info) {
6182 goto err_out_rbd_dev;
6185 down_write(&rbd_dev->header_rwsem);
6186 rc = rbd_dev_image_probe(rbd_dev, 0);
6188 up_write(&rbd_dev->header_rwsem);
6189 goto err_out_rbd_dev;
6192 /* If we are mapping a snapshot it must be marked read-only */
6194 read_only = rbd_dev->opts->read_only;
6195 if (rbd_dev->spec->snap_id != CEPH_NOSNAP)
6197 rbd_dev->mapping.read_only = read_only;
6199 rc = rbd_dev_device_setup(rbd_dev);
6201 goto err_out_image_probe;
6203 if (rbd_dev->opts->exclusive) {
6204 rc = rbd_add_acquire_lock(rbd_dev);
6206 goto err_out_device_setup;
6209 /* Everything's ready. Announce the disk to the world. */
6211 rc = device_add(&rbd_dev->dev);
6213 goto err_out_image_lock;
6215 add_disk(rbd_dev->disk);
6216 /* see rbd_init_disk() */
6217 blk_put_queue(rbd_dev->disk->queue);
6219 spin_lock(&rbd_dev_list_lock);
6220 list_add_tail(&rbd_dev->node, &rbd_dev_list);
6221 spin_unlock(&rbd_dev_list_lock);
6223 pr_info("%s: capacity %llu features 0x%llx\n", rbd_dev->disk->disk_name,
6224 (unsigned long long)get_capacity(rbd_dev->disk) << SECTOR_SHIFT,
6225 rbd_dev->header.features);
6228 module_put(THIS_MODULE);
6232 rbd_dev_image_unlock(rbd_dev);
6233 err_out_device_setup:
6234 rbd_dev_device_release(rbd_dev);
6235 err_out_image_probe:
6236 rbd_dev_image_release(rbd_dev);
6238 rbd_dev_destroy(rbd_dev);
6240 rbd_put_client(rbdc);
6247 static ssize_t rbd_add(struct bus_type *bus,
6254 return do_rbd_add(bus, buf, count);
6257 static ssize_t rbd_add_single_major(struct bus_type *bus,
6261 return do_rbd_add(bus, buf, count);
6264 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev)
6266 while (rbd_dev->parent) {
6267 struct rbd_device *first = rbd_dev;
6268 struct rbd_device *second = first->parent;
6269 struct rbd_device *third;
6272 * Follow to the parent with no grandparent and
6275 while (second && (third = second->parent)) {
6280 rbd_dev_image_release(second);
6281 rbd_dev_destroy(second);
6282 first->parent = NULL;
6283 first->parent_overlap = 0;
6285 rbd_assert(first->parent_spec);
6286 rbd_spec_put(first->parent_spec);
6287 first->parent_spec = NULL;
6291 static ssize_t do_rbd_remove(struct bus_type *bus,
6295 struct rbd_device *rbd_dev = NULL;
6296 struct list_head *tmp;
6299 bool already = false;
6305 sscanf(buf, "%d %5s", &dev_id, opt_buf);
6307 pr_err("dev_id out of range\n");
6310 if (opt_buf[0] != '\0') {
6311 if (!strcmp(opt_buf, "force")) {
6314 pr_err("bad remove option at '%s'\n", opt_buf);
6320 spin_lock(&rbd_dev_list_lock);
6321 list_for_each(tmp, &rbd_dev_list) {
6322 rbd_dev = list_entry(tmp, struct rbd_device, node);
6323 if (rbd_dev->dev_id == dev_id) {
6329 spin_lock_irq(&rbd_dev->lock);
6330 if (rbd_dev->open_count && !force)
6333 already = test_and_set_bit(RBD_DEV_FLAG_REMOVING,
6335 spin_unlock_irq(&rbd_dev->lock);
6337 spin_unlock(&rbd_dev_list_lock);
6338 if (ret < 0 || already)
6343 * Prevent new IO from being queued and wait for existing
6344 * IO to complete/fail.
6346 blk_mq_freeze_queue(rbd_dev->disk->queue);
6347 blk_set_queue_dying(rbd_dev->disk->queue);
6350 del_gendisk(rbd_dev->disk);
6351 spin_lock(&rbd_dev_list_lock);
6352 list_del_init(&rbd_dev->node);
6353 spin_unlock(&rbd_dev_list_lock);
6354 device_del(&rbd_dev->dev);
6356 rbd_dev_image_unlock(rbd_dev);
6357 rbd_dev_device_release(rbd_dev);
6358 rbd_dev_image_release(rbd_dev);
6359 rbd_dev_destroy(rbd_dev);
6363 static ssize_t rbd_remove(struct bus_type *bus,
6370 return do_rbd_remove(bus, buf, count);
6373 static ssize_t rbd_remove_single_major(struct bus_type *bus,
6377 return do_rbd_remove(bus, buf, count);
6381 * create control files in sysfs
6384 static int rbd_sysfs_init(void)
6388 ret = device_register(&rbd_root_dev);
6392 ret = bus_register(&rbd_bus_type);
6394 device_unregister(&rbd_root_dev);
6399 static void rbd_sysfs_cleanup(void)
6401 bus_unregister(&rbd_bus_type);
6402 device_unregister(&rbd_root_dev);
6405 static int rbd_slab_init(void)
6407 rbd_assert(!rbd_img_request_cache);
6408 rbd_img_request_cache = KMEM_CACHE(rbd_img_request, 0);
6409 if (!rbd_img_request_cache)
6412 rbd_assert(!rbd_obj_request_cache);
6413 rbd_obj_request_cache = KMEM_CACHE(rbd_obj_request, 0);
6414 if (!rbd_obj_request_cache)
6420 kmem_cache_destroy(rbd_img_request_cache);
6421 rbd_img_request_cache = NULL;
6425 static void rbd_slab_exit(void)
6427 rbd_assert(rbd_obj_request_cache);
6428 kmem_cache_destroy(rbd_obj_request_cache);
6429 rbd_obj_request_cache = NULL;
6431 rbd_assert(rbd_img_request_cache);
6432 kmem_cache_destroy(rbd_img_request_cache);
6433 rbd_img_request_cache = NULL;
6436 static int __init rbd_init(void)
6440 if (!libceph_compatible(NULL)) {
6441 rbd_warn(NULL, "libceph incompatibility (quitting)");
6445 rc = rbd_slab_init();
6450 * The number of active work items is limited by the number of
6451 * rbd devices * queue depth, so leave @max_active at default.
6453 rbd_wq = alloc_workqueue(RBD_DRV_NAME, WQ_MEM_RECLAIM, 0);
6460 rbd_major = register_blkdev(0, RBD_DRV_NAME);
6461 if (rbd_major < 0) {
6467 rc = rbd_sysfs_init();
6469 goto err_out_blkdev;
6472 pr_info("loaded (major %d)\n", rbd_major);
6474 pr_info("loaded\n");
6480 unregister_blkdev(rbd_major, RBD_DRV_NAME);
6482 destroy_workqueue(rbd_wq);
6488 static void __exit rbd_exit(void)
6490 ida_destroy(&rbd_dev_id_ida);
6491 rbd_sysfs_cleanup();
6493 unregister_blkdev(rbd_major, RBD_DRV_NAME);
6494 destroy_workqueue(rbd_wq);
6498 module_init(rbd_init);
6499 module_exit(rbd_exit);
6501 MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
6502 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
6503 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
6504 /* following authorship retained from original osdblk.c */
6505 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
6507 MODULE_DESCRIPTION("RADOS Block Device (RBD) driver");
6508 MODULE_LICENSE("GPL");