]> git.proxmox.com Git - mirror_ubuntu-bionic-kernel.git/blame - drivers/block/rbd.c
Merge tag 'platform-drivers-x86-v4.15-3' of git://git.infradead.org/linux-platform...
[mirror_ubuntu-bionic-kernel.git] / drivers / block / rbd.c
CommitLineData
e2a58ee5 1
602adf40
YS
2/*
3 rbd.c -- Export ceph rados objects as a Linux block device
4
5
6 based on drivers/block/osdblk.c:
7
8 Copyright 2009 Red Hat, Inc.
9
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.
13
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.
18
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.
22
23
24
dfc5606d 25 For usage instructions, please refer to:
602adf40 26
dfc5606d 27 Documentation/ABI/testing/sysfs-bus-rbd
602adf40
YS
28
29 */
30
31#include <linux/ceph/libceph.h>
32#include <linux/ceph/osd_client.h>
33#include <linux/ceph/mon_client.h>
ed95b21a 34#include <linux/ceph/cls_lock_client.h>
602adf40 35#include <linux/ceph/decode.h>
59c2be1e 36#include <linux/parser.h>
30d1cff8 37#include <linux/bsearch.h>
602adf40
YS
38
39#include <linux/kernel.h>
40#include <linux/device.h>
41#include <linux/module.h>
7ad18afa 42#include <linux/blk-mq.h>
602adf40
YS
43#include <linux/fs.h>
44#include <linux/blkdev.h>
1c2a9dfe 45#include <linux/slab.h>
f8a22fc2 46#include <linux/idr.h>
bc1ecc65 47#include <linux/workqueue.h>
602adf40
YS
48
49#include "rbd_types.h"
50
aafb230e
AE
51#define RBD_DEBUG /* Activate rbd_assert() calls */
52
593a9e7b
AE
53/*
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.
58 */
59#define SECTOR_SHIFT 9
60#define SECTOR_SIZE (1ULL << SECTOR_SHIFT)
61
a2acd00e
AE
62/*
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.
67 */
68static int atomic_inc_return_safe(atomic_t *v)
69{
70 unsigned int counter;
71
72 counter = (unsigned int)__atomic_add_unless(v, 1, 0);
73 if (counter <= (unsigned int)INT_MAX)
74 return (int)counter;
75
76 atomic_dec(v);
77
78 return -EINVAL;
79}
80
81/* Decrement the counter. Return the resulting value, or -EINVAL */
82static int atomic_dec_return_safe(atomic_t *v)
83{
84 int counter;
85
86 counter = atomic_dec_return(v);
87 if (counter >= 0)
88 return counter;
89
90 atomic_inc(v);
91
92 return -EINVAL;
93}
94
f0f8cef5 95#define RBD_DRV_NAME "rbd"
602adf40 96
7e513d43
ID
97#define RBD_MINORS_PER_MAJOR 256
98#define RBD_SINGLE_MAJOR_PART_SHIFT 4
602adf40 99
6d69bb53
ID
100#define RBD_MAX_PARENT_CHAIN_LEN 16
101
d4b125e9
AE
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))
105
35d489f9 106#define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */
602adf40
YS
107
108#define RBD_SNAP_HEAD_NAME "-"
109
9682fc6d
AE
110#define BAD_SNAP_INDEX U32_MAX /* invalid index into snap array */
111
9e15b77d
AE
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)
1e130199 114#define RBD_IMAGE_ID_LEN_MAX 64
9e15b77d 115
1e130199 116#define RBD_OBJ_PREFIX_LEN_MAX 64
589d30e0 117
ed95b21a 118#define RBD_NOTIFY_TIMEOUT 5 /* seconds */
99d16943
ID
119#define RBD_RETRY_DELAY msecs_to_jiffies(1000)
120
d889140c
AE
121/* Feature bits */
122
8767b293
ID
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)
127
ed95b21a
ID
128#define RBD_FEATURES_ALL (RBD_FEATURE_LAYERING | \
129 RBD_FEATURE_STRIPINGV2 | \
7e97332e
ID
130 RBD_FEATURE_EXCLUSIVE_LOCK | \
131 RBD_FEATURE_DATA_POOL)
d889140c
AE
132
133/* Features supported by this (client software) implementation. */
134
770eba6e 135#define RBD_FEATURES_SUPPORTED (RBD_FEATURES_ALL)
d889140c 136
81a89793
AE
137/*
138 * An RBD device name will be "rbd#", where the "rbd" comes from
139 * RBD_DRV_NAME above, and # is a unique integer identifier.
81a89793 140 */
602adf40
YS
141#define DEV_NAME_LEN 32
142
143/*
144 * block device image metadata (in-memory version)
145 */
146struct rbd_image_header {
f35a4dee 147 /* These six fields never change for a given rbd image */
849b4260 148 char *object_prefix;
602adf40 149 __u8 obj_order;
f35a4dee
AE
150 u64 stripe_unit;
151 u64 stripe_count;
7e97332e 152 s64 data_pool_id;
f35a4dee 153 u64 features; /* Might be changeable someday? */
602adf40 154
f84344f3
AE
155 /* The remaining fields need to be updated occasionally */
156 u64 image_size;
157 struct ceph_snap_context *snapc;
f35a4dee
AE
158 char *snap_names; /* format 1 only */
159 u64 *snap_sizes; /* format 1 only */
59c2be1e
YS
160};
161
0d7dbfce
AE
162/*
163 * An rbd image specification.
164 *
165 * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
c66c6e0c
AE
166 * identify an image. Each rbd_dev structure includes a pointer to
167 * an rbd_spec structure that encapsulates this identity.
168 *
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.
173 *
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).
179 *
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.
183 *
184 * Note that code herein does not assume the image name is known (it
185 * could be a null pointer).
0d7dbfce
AE
186 */
187struct rbd_spec {
188 u64 pool_id;
ecb4dc22 189 const char *pool_name;
0d7dbfce 190
ecb4dc22
AE
191 const char *image_id;
192 const char *image_name;
0d7dbfce
AE
193
194 u64 snap_id;
ecb4dc22 195 const char *snap_name;
0d7dbfce
AE
196
197 struct kref kref;
198};
199
602adf40 200/*
f0f8cef5 201 * an instance of the client. multiple devices may share an rbd client.
602adf40
YS
202 */
203struct rbd_client {
204 struct ceph_client *client;
205 struct kref kref;
206 struct list_head node;
207};
208
bf0d5f50
AE
209struct rbd_img_request;
210typedef void (*rbd_img_callback_t)(struct rbd_img_request *);
211
212#define BAD_WHICH U32_MAX /* Good which or bad which, which? */
213
214struct rbd_obj_request;
215typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);
216
9969ebc5
AE
217enum obj_request_type {
218 OBJ_REQUEST_NODATA, OBJ_REQUEST_BIO, OBJ_REQUEST_PAGES
219};
bf0d5f50 220
6d2940c8
GZ
221enum obj_operation_type {
222 OBJ_OP_WRITE,
223 OBJ_OP_READ,
90e98c52 224 OBJ_OP_DISCARD,
6d2940c8
GZ
225};
226
926f9b3f
AE
227enum obj_req_flags {
228 OBJ_REQ_DONE, /* completion flag: not done = 0, done = 1 */
6365d33a 229 OBJ_REQ_IMG_DATA, /* object usage: standalone = 0, image = 1 */
5679c59f
AE
230 OBJ_REQ_KNOWN, /* EXISTS flag valid: no = 0, yes = 1 */
231 OBJ_REQ_EXISTS, /* target exists: no = 0, yes = 1 */
926f9b3f
AE
232};
233
bf0d5f50 234struct rbd_obj_request {
a90bb0c1 235 u64 object_no;
bf0d5f50
AE
236 u64 offset; /* object start byte */
237 u64 length; /* bytes from offset */
926f9b3f 238 unsigned long flags;
bf0d5f50 239
c5b5ef6c
AE
240 /*
241 * An object request associated with an image will have its
242 * img_data flag set; a standalone object request will not.
243 *
244 * A standalone object request will have which == BAD_WHICH
245 * and a null obj_request pointer.
246 *
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.
250 *
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).
255 */
256 union {
257 struct rbd_obj_request *obj_request; /* STAT op */
258 struct {
259 struct rbd_img_request *img_request;
260 u64 img_offset;
261 /* links for img_request->obj_requests list */
262 struct list_head links;
263 };
264 };
bf0d5f50
AE
265 u32 which; /* posn image request list */
266
267 enum obj_request_type type;
788e2df3
AE
268 union {
269 struct bio *bio_list;
270 struct {
271 struct page **pages;
272 u32 page_count;
273 };
274 };
0eefd470 275 struct page **copyup_pages;
ebda6408 276 u32 copyup_page_count;
bf0d5f50
AE
277
278 struct ceph_osd_request *osd_req;
279
280 u64 xferred; /* bytes transferred */
1b83bef2 281 int result;
bf0d5f50
AE
282
283 rbd_obj_callback_t callback;
788e2df3 284 struct completion completion;
bf0d5f50
AE
285
286 struct kref kref;
287};
288
0c425248 289enum img_req_flags {
9849e986
AE
290 IMG_REQ_WRITE, /* I/O direction: read = 0, write = 1 */
291 IMG_REQ_CHILD, /* initiator: block = 0, child image = 1 */
d0b2e944 292 IMG_REQ_LAYERED, /* ENOENT handling: normal = 0, layered = 1 */
90e98c52 293 IMG_REQ_DISCARD, /* discard: normal = 0, discard request = 1 */
0c425248
AE
294};
295
bf0d5f50 296struct rbd_img_request {
bf0d5f50
AE
297 struct rbd_device *rbd_dev;
298 u64 offset; /* starting image byte offset */
299 u64 length; /* byte count from offset */
0c425248 300 unsigned long flags;
bf0d5f50 301 union {
9849e986 302 u64 snap_id; /* for reads */
bf0d5f50 303 struct ceph_snap_context *snapc; /* for writes */
9849e986
AE
304 };
305 union {
306 struct request *rq; /* block request */
307 struct rbd_obj_request *obj_request; /* obj req initiator */
bf0d5f50 308 };
3d7efd18 309 struct page **copyup_pages;
ebda6408 310 u32 copyup_page_count;
bf0d5f50
AE
311 spinlock_t completion_lock;/* protects next_completion */
312 u32 next_completion;
313 rbd_img_callback_t callback;
55f27e09 314 u64 xferred;/* aggregate bytes transferred */
a5a337d4 315 int result; /* first nonzero obj_request result */
bf0d5f50
AE
316
317 u32 obj_request_count;
318 struct list_head obj_requests; /* rbd_obj_request structs */
319
320 struct kref kref;
321};
322
323#define for_each_obj_request(ireq, oreq) \
ef06f4d3 324 list_for_each_entry(oreq, &(ireq)->obj_requests, links)
bf0d5f50 325#define for_each_obj_request_from(ireq, oreq) \
ef06f4d3 326 list_for_each_entry_from(oreq, &(ireq)->obj_requests, links)
bf0d5f50 327#define for_each_obj_request_safe(ireq, oreq, n) \
ef06f4d3 328 list_for_each_entry_safe_reverse(oreq, n, &(ireq)->obj_requests, links)
bf0d5f50 329
99d16943
ID
330enum rbd_watch_state {
331 RBD_WATCH_STATE_UNREGISTERED,
332 RBD_WATCH_STATE_REGISTERED,
333 RBD_WATCH_STATE_ERROR,
334};
335
ed95b21a
ID
336enum rbd_lock_state {
337 RBD_LOCK_STATE_UNLOCKED,
338 RBD_LOCK_STATE_LOCKED,
339 RBD_LOCK_STATE_RELEASING,
340};
341
342/* WatchNotify::ClientId */
343struct rbd_client_id {
344 u64 gid;
345 u64 handle;
346};
347
f84344f3 348struct rbd_mapping {
99c1f08f 349 u64 size;
34b13184 350 u64 features;
f84344f3
AE
351};
352
602adf40
YS
353/*
354 * a single device
355 */
356struct rbd_device {
de71a297 357 int dev_id; /* blkdev unique id */
602adf40
YS
358
359 int major; /* blkdev assigned major */
dd82fff1 360 int minor;
602adf40 361 struct gendisk *disk; /* blkdev's gendisk and rq */
602adf40 362
a30b71b9 363 u32 image_format; /* Either 1 or 2 */
602adf40
YS
364 struct rbd_client *rbd_client;
365
366 char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
367
b82d167b 368 spinlock_t lock; /* queue, flags, open_count */
602adf40
YS
369
370 struct rbd_image_header header;
b82d167b 371 unsigned long flags; /* possibly lock protected */
0d7dbfce 372 struct rbd_spec *spec;
d147543d 373 struct rbd_options *opts;
0d6d1e9c 374 char *config_info; /* add{,_single_major} string */
602adf40 375
c41d13a3 376 struct ceph_object_id header_oid;
922dab61 377 struct ceph_object_locator header_oloc;
971f839a 378
1643dfa4 379 struct ceph_file_layout layout; /* used for all rbd requests */
0903e875 380
99d16943
ID
381 struct mutex watch_mutex;
382 enum rbd_watch_state watch_state;
922dab61 383 struct ceph_osd_linger_request *watch_handle;
99d16943
ID
384 u64 watch_cookie;
385 struct delayed_work watch_dwork;
59c2be1e 386
ed95b21a
ID
387 struct rw_semaphore lock_rwsem;
388 enum rbd_lock_state lock_state;
cbbfb0ff 389 char lock_cookie[32];
ed95b21a
ID
390 struct rbd_client_id owner_cid;
391 struct work_struct acquired_lock_work;
392 struct work_struct released_lock_work;
393 struct delayed_work lock_dwork;
394 struct work_struct unlock_work;
395 wait_queue_head_t lock_waitq;
396
1643dfa4 397 struct workqueue_struct *task_wq;
59c2be1e 398
86b00e0d
AE
399 struct rbd_spec *parent_spec;
400 u64 parent_overlap;
a2acd00e 401 atomic_t parent_ref;
2f82ee54 402 struct rbd_device *parent;
86b00e0d 403
7ad18afa
CH
404 /* Block layer tags. */
405 struct blk_mq_tag_set tag_set;
406
c666601a
JD
407 /* protects updating the header */
408 struct rw_semaphore header_rwsem;
f84344f3
AE
409
410 struct rbd_mapping mapping;
602adf40
YS
411
412 struct list_head node;
dfc5606d 413
dfc5606d
YS
414 /* sysfs related */
415 struct device dev;
b82d167b 416 unsigned long open_count; /* protected by lock */
dfc5606d
YS
417};
418
b82d167b 419/*
87c0fded
ID
420 * Flag bits for rbd_dev->flags:
421 * - REMOVING (which is coupled with rbd_dev->open_count) is protected
422 * by rbd_dev->lock
423 * - BLACKLISTED is protected by rbd_dev->lock_rwsem
b82d167b 424 */
6d292906
AE
425enum rbd_dev_flags {
426 RBD_DEV_FLAG_EXISTS, /* mapped snapshot has not been deleted */
b82d167b 427 RBD_DEV_FLAG_REMOVING, /* this mapping is being removed */
87c0fded 428 RBD_DEV_FLAG_BLACKLISTED, /* our ceph_client is blacklisted */
6d292906
AE
429};
430
cfbf6377 431static DEFINE_MUTEX(client_mutex); /* Serialize client creation */
e124a82f 432
602adf40 433static LIST_HEAD(rbd_dev_list); /* devices */
e124a82f
AE
434static DEFINE_SPINLOCK(rbd_dev_list_lock);
435
432b8587
AE
436static LIST_HEAD(rbd_client_list); /* clients */
437static DEFINE_SPINLOCK(rbd_client_list_lock);
602adf40 438
78c2a44a
AE
439/* Slab caches for frequently-allocated structures */
440
1c2a9dfe 441static struct kmem_cache *rbd_img_request_cache;
868311b1 442static struct kmem_cache *rbd_obj_request_cache;
1c2a9dfe 443
f856dc36
N
444static struct bio_set *rbd_bio_clone;
445
9b60e70b 446static int rbd_major;
f8a22fc2
ID
447static DEFINE_IDA(rbd_dev_id_ida);
448
f5ee37bd
ID
449static struct workqueue_struct *rbd_wq;
450
9b60e70b 451/*
3cfa3b16 452 * single-major requires >= 0.75 version of userspace rbd utility.
9b60e70b 453 */
3cfa3b16 454static bool single_major = true;
9b60e70b 455module_param(single_major, bool, S_IRUGO);
3cfa3b16 456MODULE_PARM_DESC(single_major, "Use a single major number for all rbd devices (default: true)");
9b60e70b 457
3d7efd18
AE
458static int rbd_img_request_submit(struct rbd_img_request *img_request);
459
f0f8cef5
AE
460static ssize_t rbd_add(struct bus_type *bus, const char *buf,
461 size_t count);
462static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
463 size_t count);
9b60e70b
ID
464static ssize_t rbd_add_single_major(struct bus_type *bus, const char *buf,
465 size_t count);
466static ssize_t rbd_remove_single_major(struct bus_type *bus, const char *buf,
467 size_t count);
6d69bb53 468static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth);
a2acd00e 469static void rbd_spec_put(struct rbd_spec *spec);
f0f8cef5 470
9b60e70b
ID
471static int rbd_dev_id_to_minor(int dev_id)
472{
7e513d43 473 return dev_id << RBD_SINGLE_MAJOR_PART_SHIFT;
9b60e70b
ID
474}
475
476static int minor_to_rbd_dev_id(int minor)
477{
7e513d43 478 return minor >> RBD_SINGLE_MAJOR_PART_SHIFT;
9b60e70b
ID
479}
480
ed95b21a
ID
481static bool __rbd_is_lock_owner(struct rbd_device *rbd_dev)
482{
483 return rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED ||
484 rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING;
485}
486
487static bool rbd_is_lock_owner(struct rbd_device *rbd_dev)
488{
489 bool is_lock_owner;
490
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;
495}
496
8767b293
ID
497static ssize_t rbd_supported_features_show(struct bus_type *bus, char *buf)
498{
499 return sprintf(buf, "0x%llx\n", RBD_FEATURES_SUPPORTED);
500}
501
b15a21dd
GKH
502static BUS_ATTR(add, S_IWUSR, NULL, rbd_add);
503static BUS_ATTR(remove, S_IWUSR, NULL, rbd_remove);
9b60e70b
ID
504static BUS_ATTR(add_single_major, S_IWUSR, NULL, rbd_add_single_major);
505static BUS_ATTR(remove_single_major, S_IWUSR, NULL, rbd_remove_single_major);
8767b293 506static BUS_ATTR(supported_features, S_IRUGO, rbd_supported_features_show, NULL);
b15a21dd
GKH
507
508static struct attribute *rbd_bus_attrs[] = {
509 &bus_attr_add.attr,
510 &bus_attr_remove.attr,
9b60e70b
ID
511 &bus_attr_add_single_major.attr,
512 &bus_attr_remove_single_major.attr,
8767b293 513 &bus_attr_supported_features.attr,
b15a21dd 514 NULL,
f0f8cef5 515};
92c76dc0
ID
516
517static umode_t rbd_bus_is_visible(struct kobject *kobj,
518 struct attribute *attr, int index)
519{
9b60e70b
ID
520 if (!single_major &&
521 (attr == &bus_attr_add_single_major.attr ||
522 attr == &bus_attr_remove_single_major.attr))
523 return 0;
524
92c76dc0
ID
525 return attr->mode;
526}
527
528static const struct attribute_group rbd_bus_group = {
529 .attrs = rbd_bus_attrs,
530 .is_visible = rbd_bus_is_visible,
531};
532__ATTRIBUTE_GROUPS(rbd_bus);
f0f8cef5
AE
533
534static struct bus_type rbd_bus_type = {
535 .name = "rbd",
b15a21dd 536 .bus_groups = rbd_bus_groups,
f0f8cef5
AE
537};
538
539static void rbd_root_dev_release(struct device *dev)
540{
541}
542
543static struct device rbd_root_dev = {
544 .init_name = "rbd",
545 .release = rbd_root_dev_release,
546};
547
06ecc6cb
AE
548static __printf(2, 3)
549void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
550{
551 struct va_format vaf;
552 va_list args;
553
554 va_start(args, fmt);
555 vaf.fmt = fmt;
556 vaf.va = &args;
557
558 if (!rbd_dev)
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);
569 else /* punt */
570 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
571 RBD_DRV_NAME, rbd_dev, &vaf);
572 va_end(args);
573}
574
aafb230e
AE
575#ifdef RBD_DEBUG
576#define rbd_assert(expr) \
577 if (unlikely(!(expr))) { \
578 printk(KERN_ERR "\nAssertion failure in %s() " \
579 "at line %d:\n\n" \
580 "\trbd_assert(%s);\n\n", \
581 __func__, __LINE__, #expr); \
582 BUG(); \
583 }
584#else /* !RBD_DEBUG */
585# define rbd_assert(expr) ((void) 0)
586#endif /* !RBD_DEBUG */
dfc5606d 587
2761713d 588static void rbd_osd_copyup_callback(struct rbd_obj_request *obj_request);
b454e36d 589static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request);
05a46afd
AE
590static void rbd_img_parent_read(struct rbd_obj_request *obj_request);
591static void rbd_dev_remove_parent(struct rbd_device *rbd_dev);
8b3e1a56 592
cc4a38bd 593static int rbd_dev_refresh(struct rbd_device *rbd_dev);
2df3fac7 594static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev);
a720ae09 595static int rbd_dev_header_info(struct rbd_device *rbd_dev);
e8f59b59 596static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev);
54cac61f
AE
597static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
598 u64 snap_id);
2ad3d716
AE
599static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
600 u8 *order, u64 *snap_size);
601static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
602 u64 *snap_features);
59c2be1e 603
602adf40
YS
604static int rbd_open(struct block_device *bdev, fmode_t mode)
605{
f0f8cef5 606 struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
b82d167b 607 bool removing = false;
602adf40 608
a14ea269 609 spin_lock_irq(&rbd_dev->lock);
b82d167b
AE
610 if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
611 removing = true;
612 else
613 rbd_dev->open_count++;
a14ea269 614 spin_unlock_irq(&rbd_dev->lock);
b82d167b
AE
615 if (removing)
616 return -ENOENT;
617
c3e946ce 618 (void) get_device(&rbd_dev->dev);
340c7a2b 619
602adf40
YS
620 return 0;
621}
622
db2a144b 623static void rbd_release(struct gendisk *disk, fmode_t mode)
dfc5606d
YS
624{
625 struct rbd_device *rbd_dev = disk->private_data;
b82d167b
AE
626 unsigned long open_count_before;
627
a14ea269 628 spin_lock_irq(&rbd_dev->lock);
b82d167b 629 open_count_before = rbd_dev->open_count--;
a14ea269 630 spin_unlock_irq(&rbd_dev->lock);
b82d167b 631 rbd_assert(open_count_before > 0);
dfc5606d 632
c3e946ce 633 put_device(&rbd_dev->dev);
dfc5606d
YS
634}
635
131fd9f6
GZ
636static int rbd_ioctl_set_ro(struct rbd_device *rbd_dev, unsigned long arg)
637{
1de797bb 638 int ro;
131fd9f6 639
1de797bb 640 if (get_user(ro, (int __user *)arg))
131fd9f6
GZ
641 return -EFAULT;
642
1de797bb 643 /* Snapshots can't be marked read-write */
131fd9f6
GZ
644 if (rbd_dev->spec->snap_id != CEPH_NOSNAP && !ro)
645 return -EROFS;
646
1de797bb
ID
647 /* Let blkdev_roset() handle it */
648 return -ENOTTY;
131fd9f6
GZ
649}
650
651static int rbd_ioctl(struct block_device *bdev, fmode_t mode,
652 unsigned int cmd, unsigned long arg)
653{
654 struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
1de797bb 655 int ret;
131fd9f6 656
131fd9f6
GZ
657 switch (cmd) {
658 case BLKROSET:
659 ret = rbd_ioctl_set_ro(rbd_dev, arg);
660 break;
661 default:
662 ret = -ENOTTY;
663 }
664
131fd9f6
GZ
665 return ret;
666}
667
668#ifdef CONFIG_COMPAT
669static int rbd_compat_ioctl(struct block_device *bdev, fmode_t mode,
670 unsigned int cmd, unsigned long arg)
671{
672 return rbd_ioctl(bdev, mode, cmd, arg);
673}
674#endif /* CONFIG_COMPAT */
675
602adf40
YS
676static const struct block_device_operations rbd_bd_ops = {
677 .owner = THIS_MODULE,
678 .open = rbd_open,
dfc5606d 679 .release = rbd_release,
131fd9f6
GZ
680 .ioctl = rbd_ioctl,
681#ifdef CONFIG_COMPAT
682 .compat_ioctl = rbd_compat_ioctl,
683#endif
602adf40
YS
684};
685
686/*
7262cfca 687 * Initialize an rbd client instance. Success or not, this function
cfbf6377 688 * consumes ceph_opts. Caller holds client_mutex.
602adf40 689 */
f8c38929 690static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
602adf40
YS
691{
692 struct rbd_client *rbdc;
693 int ret = -ENOMEM;
694
37206ee5 695 dout("%s:\n", __func__);
602adf40
YS
696 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
697 if (!rbdc)
698 goto out_opt;
699
700 kref_init(&rbdc->kref);
701 INIT_LIST_HEAD(&rbdc->node);
702
74da4a0f 703 rbdc->client = ceph_create_client(ceph_opts, rbdc);
602adf40 704 if (IS_ERR(rbdc->client))
08f75463 705 goto out_rbdc;
43ae4701 706 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
602adf40
YS
707
708 ret = ceph_open_session(rbdc->client);
709 if (ret < 0)
08f75463 710 goto out_client;
602adf40 711
432b8587 712 spin_lock(&rbd_client_list_lock);
602adf40 713 list_add_tail(&rbdc->node, &rbd_client_list);
432b8587 714 spin_unlock(&rbd_client_list_lock);
602adf40 715
37206ee5 716 dout("%s: rbdc %p\n", __func__, rbdc);
bc534d86 717
602adf40 718 return rbdc;
08f75463 719out_client:
602adf40 720 ceph_destroy_client(rbdc->client);
08f75463 721out_rbdc:
602adf40
YS
722 kfree(rbdc);
723out_opt:
43ae4701
AE
724 if (ceph_opts)
725 ceph_destroy_options(ceph_opts);
37206ee5
AE
726 dout("%s: error %d\n", __func__, ret);
727
28f259b7 728 return ERR_PTR(ret);
602adf40
YS
729}
730
2f82ee54
AE
731static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
732{
733 kref_get(&rbdc->kref);
734
735 return rbdc;
736}
737
602adf40 738/*
1f7ba331
AE
739 * Find a ceph client with specific addr and configuration. If
740 * found, bump its reference count.
602adf40 741 */
1f7ba331 742static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
602adf40
YS
743{
744 struct rbd_client *client_node;
1f7ba331 745 bool found = false;
602adf40 746
43ae4701 747 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
602adf40
YS
748 return NULL;
749
1f7ba331
AE
750 spin_lock(&rbd_client_list_lock);
751 list_for_each_entry(client_node, &rbd_client_list, node) {
752 if (!ceph_compare_options(ceph_opts, client_node->client)) {
2f82ee54
AE
753 __rbd_get_client(client_node);
754
1f7ba331
AE
755 found = true;
756 break;
757 }
758 }
759 spin_unlock(&rbd_client_list_lock);
760
761 return found ? client_node : NULL;
602adf40
YS
762}
763
59c2be1e 764/*
210c104c 765 * (Per device) rbd map options
59c2be1e
YS
766 */
767enum {
b5584180 768 Opt_queue_depth,
59c2be1e
YS
769 Opt_last_int,
770 /* int args above */
771 Opt_last_string,
772 /* string args above */
cc0538b6
AE
773 Opt_read_only,
774 Opt_read_write,
80de1912 775 Opt_lock_on_read,
e010dd0a 776 Opt_exclusive,
210c104c 777 Opt_err
59c2be1e
YS
778};
779
43ae4701 780static match_table_t rbd_opts_tokens = {
b5584180 781 {Opt_queue_depth, "queue_depth=%d"},
59c2be1e
YS
782 /* int args above */
783 /* string args above */
be466c1c 784 {Opt_read_only, "read_only"},
cc0538b6
AE
785 {Opt_read_only, "ro"}, /* Alternate spelling */
786 {Opt_read_write, "read_write"},
787 {Opt_read_write, "rw"}, /* Alternate spelling */
80de1912 788 {Opt_lock_on_read, "lock_on_read"},
e010dd0a 789 {Opt_exclusive, "exclusive"},
210c104c 790 {Opt_err, NULL}
59c2be1e
YS
791};
792
98571b5a 793struct rbd_options {
b5584180 794 int queue_depth;
98571b5a 795 bool read_only;
80de1912 796 bool lock_on_read;
e010dd0a 797 bool exclusive;
98571b5a
AE
798};
799
b5584180 800#define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_MAX_RQ
98571b5a 801#define RBD_READ_ONLY_DEFAULT false
80de1912 802#define RBD_LOCK_ON_READ_DEFAULT false
e010dd0a 803#define RBD_EXCLUSIVE_DEFAULT false
98571b5a 804
59c2be1e
YS
805static int parse_rbd_opts_token(char *c, void *private)
806{
43ae4701 807 struct rbd_options *rbd_opts = private;
59c2be1e
YS
808 substring_t argstr[MAX_OPT_ARGS];
809 int token, intval, ret;
810
43ae4701 811 token = match_token(c, rbd_opts_tokens, argstr);
59c2be1e
YS
812 if (token < Opt_last_int) {
813 ret = match_int(&argstr[0], &intval);
814 if (ret < 0) {
210c104c 815 pr_err("bad mount option arg (not int) at '%s'\n", c);
59c2be1e
YS
816 return ret;
817 }
818 dout("got int token %d val %d\n", token, intval);
819 } else if (token > Opt_last_int && token < Opt_last_string) {
210c104c 820 dout("got string token %d val %s\n", token, argstr[0].from);
59c2be1e
YS
821 } else {
822 dout("got token %d\n", token);
823 }
824
825 switch (token) {
b5584180
ID
826 case Opt_queue_depth:
827 if (intval < 1) {
828 pr_err("queue_depth out of range\n");
829 return -EINVAL;
830 }
831 rbd_opts->queue_depth = intval;
832 break;
cc0538b6
AE
833 case Opt_read_only:
834 rbd_opts->read_only = true;
835 break;
836 case Opt_read_write:
837 rbd_opts->read_only = false;
838 break;
80de1912
ID
839 case Opt_lock_on_read:
840 rbd_opts->lock_on_read = true;
841 break;
e010dd0a
ID
842 case Opt_exclusive:
843 rbd_opts->exclusive = true;
844 break;
59c2be1e 845 default:
210c104c
ID
846 /* libceph prints "bad option" msg */
847 return -EINVAL;
59c2be1e 848 }
210c104c 849
59c2be1e
YS
850 return 0;
851}
852
6d2940c8
GZ
853static char* obj_op_name(enum obj_operation_type op_type)
854{
855 switch (op_type) {
856 case OBJ_OP_READ:
857 return "read";
858 case OBJ_OP_WRITE:
859 return "write";
90e98c52
GZ
860 case OBJ_OP_DISCARD:
861 return "discard";
6d2940c8
GZ
862 default:
863 return "???";
864 }
865}
866
602adf40
YS
867/*
868 * Get a ceph client with specific addr and configuration, if one does
7262cfca
AE
869 * not exist create it. Either way, ceph_opts is consumed by this
870 * function.
602adf40 871 */
9d3997fd 872static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
602adf40 873{
f8c38929 874 struct rbd_client *rbdc;
59c2be1e 875
cfbf6377 876 mutex_lock_nested(&client_mutex, SINGLE_DEPTH_NESTING);
1f7ba331 877 rbdc = rbd_client_find(ceph_opts);
9d3997fd 878 if (rbdc) /* using an existing client */
43ae4701 879 ceph_destroy_options(ceph_opts);
9d3997fd 880 else
f8c38929 881 rbdc = rbd_client_create(ceph_opts);
cfbf6377 882 mutex_unlock(&client_mutex);
602adf40 883
9d3997fd 884 return rbdc;
602adf40
YS
885}
886
887/*
888 * Destroy ceph client
d23a4b3f 889 *
432b8587 890 * Caller must hold rbd_client_list_lock.
602adf40
YS
891 */
892static void rbd_client_release(struct kref *kref)
893{
894 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
895
37206ee5 896 dout("%s: rbdc %p\n", __func__, rbdc);
cd9d9f5d 897 spin_lock(&rbd_client_list_lock);
602adf40 898 list_del(&rbdc->node);
cd9d9f5d 899 spin_unlock(&rbd_client_list_lock);
602adf40
YS
900
901 ceph_destroy_client(rbdc->client);
902 kfree(rbdc);
903}
904
905/*
906 * Drop reference to ceph client node. If it's not referenced anymore, release
907 * it.
908 */
9d3997fd 909static void rbd_put_client(struct rbd_client *rbdc)
602adf40 910{
c53d5893
AE
911 if (rbdc)
912 kref_put(&rbdc->kref, rbd_client_release);
602adf40
YS
913}
914
a30b71b9
AE
915static bool rbd_image_format_valid(u32 image_format)
916{
917 return image_format == 1 || image_format == 2;
918}
919
8e94af8e
AE
920static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
921{
103a150f
AE
922 size_t size;
923 u32 snap_count;
924
925 /* The header has to start with the magic rbd header text */
926 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
927 return false;
928
db2388b6
AE
929 /* The bio layer requires at least sector-sized I/O */
930
931 if (ondisk->options.order < SECTOR_SHIFT)
932 return false;
933
934 /* If we use u64 in a few spots we may be able to loosen this */
935
936 if (ondisk->options.order > 8 * sizeof (int) - 1)
937 return false;
938
103a150f
AE
939 /*
940 * The size of a snapshot header has to fit in a size_t, and
941 * that limits the number of snapshots.
942 */
943 snap_count = le32_to_cpu(ondisk->snap_count);
944 size = SIZE_MAX - sizeof (struct ceph_snap_context);
945 if (snap_count > size / sizeof (__le64))
946 return false;
947
948 /*
949 * Not only that, but the size of the entire the snapshot
950 * header must also be representable in a size_t.
951 */
952 size -= snap_count * sizeof (__le64);
953 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
954 return false;
955
956 return true;
8e94af8e
AE
957}
958
5bc3fb17
ID
959/*
960 * returns the size of an object in the image
961 */
962static u32 rbd_obj_bytes(struct rbd_image_header *header)
963{
964 return 1U << header->obj_order;
965}
966
263423f8
ID
967static void rbd_init_layout(struct rbd_device *rbd_dev)
968{
969 if (rbd_dev->header.stripe_unit == 0 ||
970 rbd_dev->header.stripe_count == 0) {
971 rbd_dev->header.stripe_unit = rbd_obj_bytes(&rbd_dev->header);
972 rbd_dev->header.stripe_count = 1;
973 }
974
975 rbd_dev->layout.stripe_unit = rbd_dev->header.stripe_unit;
976 rbd_dev->layout.stripe_count = rbd_dev->header.stripe_count;
977 rbd_dev->layout.object_size = rbd_obj_bytes(&rbd_dev->header);
7e97332e
ID
978 rbd_dev->layout.pool_id = rbd_dev->header.data_pool_id == CEPH_NOPOOL ?
979 rbd_dev->spec->pool_id : rbd_dev->header.data_pool_id;
263423f8
ID
980 RCU_INIT_POINTER(rbd_dev->layout.pool_ns, NULL);
981}
982
602adf40 983/*
bb23e37a
AE
984 * Fill an rbd image header with information from the given format 1
985 * on-disk header.
602adf40 986 */
662518b1 987static int rbd_header_from_disk(struct rbd_device *rbd_dev,
4156d998 988 struct rbd_image_header_ondisk *ondisk)
602adf40 989{
662518b1 990 struct rbd_image_header *header = &rbd_dev->header;
bb23e37a
AE
991 bool first_time = header->object_prefix == NULL;
992 struct ceph_snap_context *snapc;
993 char *object_prefix = NULL;
994 char *snap_names = NULL;
995 u64 *snap_sizes = NULL;
ccece235 996 u32 snap_count;
bb23e37a 997 int ret = -ENOMEM;
621901d6 998 u32 i;
602adf40 999
bb23e37a 1000 /* Allocate this now to avoid having to handle failure below */
6a52325f 1001
bb23e37a 1002 if (first_time) {
848d796c
ID
1003 object_prefix = kstrndup(ondisk->object_prefix,
1004 sizeof(ondisk->object_prefix),
1005 GFP_KERNEL);
bb23e37a
AE
1006 if (!object_prefix)
1007 return -ENOMEM;
bb23e37a 1008 }
00f1f36f 1009
bb23e37a 1010 /* Allocate the snapshot context and fill it in */
00f1f36f 1011
bb23e37a
AE
1012 snap_count = le32_to_cpu(ondisk->snap_count);
1013 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
1014 if (!snapc)
1015 goto out_err;
1016 snapc->seq = le64_to_cpu(ondisk->snap_seq);
602adf40 1017 if (snap_count) {
bb23e37a 1018 struct rbd_image_snap_ondisk *snaps;
f785cc1d
AE
1019 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
1020
bb23e37a 1021 /* We'll keep a copy of the snapshot names... */
621901d6 1022
bb23e37a
AE
1023 if (snap_names_len > (u64)SIZE_MAX)
1024 goto out_2big;
1025 snap_names = kmalloc(snap_names_len, GFP_KERNEL);
1026 if (!snap_names)
6a52325f
AE
1027 goto out_err;
1028
bb23e37a 1029 /* ...as well as the array of their sizes. */
88a25a5f
ME
1030 snap_sizes = kmalloc_array(snap_count,
1031 sizeof(*header->snap_sizes),
1032 GFP_KERNEL);
bb23e37a 1033 if (!snap_sizes)
6a52325f 1034 goto out_err;
bb23e37a 1035
f785cc1d 1036 /*
bb23e37a
AE
1037 * Copy the names, and fill in each snapshot's id
1038 * and size.
1039 *
99a41ebc 1040 * Note that rbd_dev_v1_header_info() guarantees the
bb23e37a 1041 * ondisk buffer we're working with has
f785cc1d
AE
1042 * snap_names_len bytes beyond the end of the
1043 * snapshot id array, this memcpy() is safe.
1044 */
bb23e37a
AE
1045 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
1046 snaps = ondisk->snaps;
1047 for (i = 0; i < snap_count; i++) {
1048 snapc->snaps[i] = le64_to_cpu(snaps[i].id);
1049 snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
1050 }
602adf40 1051 }
6a52325f 1052
bb23e37a 1053 /* We won't fail any more, fill in the header */
621901d6 1054
bb23e37a
AE
1055 if (first_time) {
1056 header->object_prefix = object_prefix;
1057 header->obj_order = ondisk->options.order;
263423f8 1058 rbd_init_layout(rbd_dev);
602adf40 1059 } else {
662518b1
AE
1060 ceph_put_snap_context(header->snapc);
1061 kfree(header->snap_names);
1062 kfree(header->snap_sizes);
602adf40 1063 }
849b4260 1064
bb23e37a 1065 /* The remaining fields always get updated (when we refresh) */
621901d6 1066
f84344f3 1067 header->image_size = le64_to_cpu(ondisk->image_size);
bb23e37a
AE
1068 header->snapc = snapc;
1069 header->snap_names = snap_names;
1070 header->snap_sizes = snap_sizes;
468521c1 1071
602adf40 1072 return 0;
bb23e37a
AE
1073out_2big:
1074 ret = -EIO;
6a52325f 1075out_err:
bb23e37a
AE
1076 kfree(snap_sizes);
1077 kfree(snap_names);
1078 ceph_put_snap_context(snapc);
1079 kfree(object_prefix);
ccece235 1080
bb23e37a 1081 return ret;
602adf40
YS
1082}
1083
9682fc6d
AE
1084static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1085{
1086 const char *snap_name;
1087
1088 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1089
1090 /* Skip over names until we find the one we are looking for */
1091
1092 snap_name = rbd_dev->header.snap_names;
1093 while (which--)
1094 snap_name += strlen(snap_name) + 1;
1095
1096 return kstrdup(snap_name, GFP_KERNEL);
1097}
1098
30d1cff8
AE
1099/*
1100 * Snapshot id comparison function for use with qsort()/bsearch().
1101 * Note that result is for snapshots in *descending* order.
1102 */
1103static int snapid_compare_reverse(const void *s1, const void *s2)
1104{
1105 u64 snap_id1 = *(u64 *)s1;
1106 u64 snap_id2 = *(u64 *)s2;
1107
1108 if (snap_id1 < snap_id2)
1109 return 1;
1110 return snap_id1 == snap_id2 ? 0 : -1;
1111}
1112
1113/*
1114 * Search a snapshot context to see if the given snapshot id is
1115 * present.
1116 *
1117 * Returns the position of the snapshot id in the array if it's found,
1118 * or BAD_SNAP_INDEX otherwise.
1119 *
1120 * Note: The snapshot array is in kept sorted (by the osd) in
1121 * reverse order, highest snapshot id first.
1122 */
9682fc6d
AE
1123static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1124{
1125 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
30d1cff8 1126 u64 *found;
9682fc6d 1127
30d1cff8
AE
1128 found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1129 sizeof (snap_id), snapid_compare_reverse);
9682fc6d 1130
30d1cff8 1131 return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
9682fc6d
AE
1132}
1133
2ad3d716
AE
1134static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1135 u64 snap_id)
9e15b77d 1136{
54cac61f 1137 u32 which;
da6a6b63 1138 const char *snap_name;
9e15b77d 1139
54cac61f
AE
1140 which = rbd_dev_snap_index(rbd_dev, snap_id);
1141 if (which == BAD_SNAP_INDEX)
da6a6b63 1142 return ERR_PTR(-ENOENT);
54cac61f 1143
da6a6b63
JD
1144 snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1145 return snap_name ? snap_name : ERR_PTR(-ENOMEM);
54cac61f
AE
1146}
1147
1148static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1149{
9e15b77d
AE
1150 if (snap_id == CEPH_NOSNAP)
1151 return RBD_SNAP_HEAD_NAME;
1152
54cac61f
AE
1153 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1154 if (rbd_dev->image_format == 1)
1155 return rbd_dev_v1_snap_name(rbd_dev, snap_id);
9e15b77d 1156
54cac61f 1157 return rbd_dev_v2_snap_name(rbd_dev, snap_id);
9e15b77d
AE
1158}
1159
2ad3d716
AE
1160static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1161 u64 *snap_size)
602adf40 1162{
2ad3d716
AE
1163 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1164 if (snap_id == CEPH_NOSNAP) {
1165 *snap_size = rbd_dev->header.image_size;
1166 } else if (rbd_dev->image_format == 1) {
1167 u32 which;
602adf40 1168
2ad3d716
AE
1169 which = rbd_dev_snap_index(rbd_dev, snap_id);
1170 if (which == BAD_SNAP_INDEX)
1171 return -ENOENT;
e86924a8 1172
2ad3d716
AE
1173 *snap_size = rbd_dev->header.snap_sizes[which];
1174 } else {
1175 u64 size = 0;
1176 int ret;
1177
1178 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1179 if (ret)
1180 return ret;
1181
1182 *snap_size = size;
1183 }
1184 return 0;
602adf40
YS
1185}
1186
2ad3d716
AE
1187static int rbd_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
1188 u64 *snap_features)
602adf40 1189{
2ad3d716
AE
1190 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1191 if (snap_id == CEPH_NOSNAP) {
1192 *snap_features = rbd_dev->header.features;
1193 } else if (rbd_dev->image_format == 1) {
1194 *snap_features = 0; /* No features for format 1 */
602adf40 1195 } else {
2ad3d716
AE
1196 u64 features = 0;
1197 int ret;
8b0241f8 1198
2ad3d716
AE
1199 ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, &features);
1200 if (ret)
1201 return ret;
1202
1203 *snap_features = features;
1204 }
1205 return 0;
1206}
1207
1208static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1209{
8f4b7d98 1210 u64 snap_id = rbd_dev->spec->snap_id;
2ad3d716
AE
1211 u64 size = 0;
1212 u64 features = 0;
1213 int ret;
1214
2ad3d716
AE
1215 ret = rbd_snap_size(rbd_dev, snap_id, &size);
1216 if (ret)
1217 return ret;
1218 ret = rbd_snap_features(rbd_dev, snap_id, &features);
1219 if (ret)
1220 return ret;
1221
1222 rbd_dev->mapping.size = size;
1223 rbd_dev->mapping.features = features;
1224
8b0241f8 1225 return 0;
602adf40
YS
1226}
1227
d1cf5788
AE
1228static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1229{
1230 rbd_dev->mapping.size = 0;
1231 rbd_dev->mapping.features = 0;
200a6a8b
AE
1232}
1233
65ccfe21
AE
1234static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset)
1235{
5bc3fb17 1236 u64 segment_size = rbd_obj_bytes(&rbd_dev->header);
602adf40 1237
65ccfe21
AE
1238 return offset & (segment_size - 1);
1239}
1240
1241static u64 rbd_segment_length(struct rbd_device *rbd_dev,
1242 u64 offset, u64 length)
1243{
5bc3fb17 1244 u64 segment_size = rbd_obj_bytes(&rbd_dev->header);
65ccfe21
AE
1245
1246 offset &= segment_size - 1;
1247
aafb230e 1248 rbd_assert(length <= U64_MAX - offset);
65ccfe21
AE
1249 if (offset + length > segment_size)
1250 length = segment_size - offset;
1251
1252 return length;
602adf40
YS
1253}
1254
1255/*
1256 * bio helpers
1257 */
1258
1259static void bio_chain_put(struct bio *chain)
1260{
1261 struct bio *tmp;
1262
1263 while (chain) {
1264 tmp = chain;
1265 chain = chain->bi_next;
1266 bio_put(tmp);
1267 }
1268}
1269
1270/*
1271 * zeros a bio chain, starting at specific offset
1272 */
1273static void zero_bio_chain(struct bio *chain, int start_ofs)
1274{
7988613b
KO
1275 struct bio_vec bv;
1276 struct bvec_iter iter;
602adf40
YS
1277 unsigned long flags;
1278 void *buf;
602adf40
YS
1279 int pos = 0;
1280
1281 while (chain) {
7988613b
KO
1282 bio_for_each_segment(bv, chain, iter) {
1283 if (pos + bv.bv_len > start_ofs) {
602adf40 1284 int remainder = max(start_ofs - pos, 0);
7988613b 1285 buf = bvec_kmap_irq(&bv, &flags);
602adf40 1286 memset(buf + remainder, 0,
7988613b
KO
1287 bv.bv_len - remainder);
1288 flush_dcache_page(bv.bv_page);
85b5aaa6 1289 bvec_kunmap_irq(buf, &flags);
602adf40 1290 }
7988613b 1291 pos += bv.bv_len;
602adf40
YS
1292 }
1293
1294 chain = chain->bi_next;
1295 }
1296}
1297
b9434c5b
AE
1298/*
1299 * similar to zero_bio_chain(), zeros data defined by a page array,
1300 * starting at the given byte offset from the start of the array and
1301 * continuing up to the given end offset. The pages array is
1302 * assumed to be big enough to hold all bytes up to the end.
1303 */
1304static void zero_pages(struct page **pages, u64 offset, u64 end)
1305{
1306 struct page **page = &pages[offset >> PAGE_SHIFT];
1307
1308 rbd_assert(end > offset);
1309 rbd_assert(end - offset <= (u64)SIZE_MAX);
1310 while (offset < end) {
1311 size_t page_offset;
1312 size_t length;
1313 unsigned long flags;
1314 void *kaddr;
1315
491205a8
GU
1316 page_offset = offset & ~PAGE_MASK;
1317 length = min_t(size_t, PAGE_SIZE - page_offset, end - offset);
b9434c5b
AE
1318 local_irq_save(flags);
1319 kaddr = kmap_atomic(*page);
1320 memset(kaddr + page_offset, 0, length);
e2156054 1321 flush_dcache_page(*page);
b9434c5b
AE
1322 kunmap_atomic(kaddr);
1323 local_irq_restore(flags);
1324
1325 offset += length;
1326 page++;
1327 }
1328}
1329
602adf40 1330/*
f7760dad
AE
1331 * Clone a portion of a bio, starting at the given byte offset
1332 * and continuing for the number of bytes indicated.
602adf40 1333 */
f7760dad
AE
1334static struct bio *bio_clone_range(struct bio *bio_src,
1335 unsigned int offset,
1336 unsigned int len,
1337 gfp_t gfpmask)
602adf40 1338{
f7760dad
AE
1339 struct bio *bio;
1340
f856dc36 1341 bio = bio_clone_fast(bio_src, gfpmask, rbd_bio_clone);
f7760dad
AE
1342 if (!bio)
1343 return NULL; /* ENOMEM */
602adf40 1344
5341a627 1345 bio_advance(bio, offset);
4f024f37 1346 bio->bi_iter.bi_size = len;
f7760dad
AE
1347
1348 return bio;
1349}
1350
1351/*
1352 * Clone a portion of a bio chain, starting at the given byte offset
1353 * into the first bio in the source chain and continuing for the
1354 * number of bytes indicated. The result is another bio chain of
1355 * exactly the given length, or a null pointer on error.
1356 *
1357 * The bio_src and offset parameters are both in-out. On entry they
1358 * refer to the first source bio and the offset into that bio where
1359 * the start of data to be cloned is located.
1360 *
1361 * On return, bio_src is updated to refer to the bio in the source
1362 * chain that contains first un-cloned byte, and *offset will
1363 * contain the offset of that byte within that bio.
1364 */
1365static struct bio *bio_chain_clone_range(struct bio **bio_src,
1366 unsigned int *offset,
1367 unsigned int len,
1368 gfp_t gfpmask)
1369{
1370 struct bio *bi = *bio_src;
1371 unsigned int off = *offset;
1372 struct bio *chain = NULL;
1373 struct bio **end;
1374
1375 /* Build up a chain of clone bios up to the limit */
1376
4f024f37 1377 if (!bi || off >= bi->bi_iter.bi_size || !len)
f7760dad 1378 return NULL; /* Nothing to clone */
602adf40 1379
f7760dad
AE
1380 end = &chain;
1381 while (len) {
1382 unsigned int bi_size;
1383 struct bio *bio;
1384
f5400b7a
AE
1385 if (!bi) {
1386 rbd_warn(NULL, "bio_chain exhausted with %u left", len);
f7760dad 1387 goto out_err; /* EINVAL; ran out of bio's */
f5400b7a 1388 }
4f024f37 1389 bi_size = min_t(unsigned int, bi->bi_iter.bi_size - off, len);
f7760dad
AE
1390 bio = bio_clone_range(bi, off, bi_size, gfpmask);
1391 if (!bio)
1392 goto out_err; /* ENOMEM */
1393
1394 *end = bio;
1395 end = &bio->bi_next;
602adf40 1396
f7760dad 1397 off += bi_size;
4f024f37 1398 if (off == bi->bi_iter.bi_size) {
f7760dad
AE
1399 bi = bi->bi_next;
1400 off = 0;
1401 }
1402 len -= bi_size;
1403 }
1404 *bio_src = bi;
1405 *offset = off;
1406
1407 return chain;
1408out_err:
1409 bio_chain_put(chain);
602adf40 1410
602adf40
YS
1411 return NULL;
1412}
1413
926f9b3f
AE
1414/*
1415 * The default/initial value for all object request flags is 0. For
1416 * each flag, once its value is set to 1 it is never reset to 0
1417 * again.
1418 */
57acbaa7 1419static void obj_request_img_data_set(struct rbd_obj_request *obj_request)
926f9b3f 1420{
57acbaa7 1421 if (test_and_set_bit(OBJ_REQ_IMG_DATA, &obj_request->flags)) {
926f9b3f
AE
1422 struct rbd_device *rbd_dev;
1423
57acbaa7 1424 rbd_dev = obj_request->img_request->rbd_dev;
9584d508 1425 rbd_warn(rbd_dev, "obj_request %p already marked img_data",
926f9b3f
AE
1426 obj_request);
1427 }
1428}
1429
57acbaa7 1430static bool obj_request_img_data_test(struct rbd_obj_request *obj_request)
926f9b3f
AE
1431{
1432 smp_mb();
57acbaa7 1433 return test_bit(OBJ_REQ_IMG_DATA, &obj_request->flags) != 0;
926f9b3f
AE
1434}
1435
57acbaa7 1436static void obj_request_done_set(struct rbd_obj_request *obj_request)
6365d33a 1437{
57acbaa7
AE
1438 if (test_and_set_bit(OBJ_REQ_DONE, &obj_request->flags)) {
1439 struct rbd_device *rbd_dev = NULL;
6365d33a 1440
57acbaa7
AE
1441 if (obj_request_img_data_test(obj_request))
1442 rbd_dev = obj_request->img_request->rbd_dev;
9584d508 1443 rbd_warn(rbd_dev, "obj_request %p already marked done",
6365d33a
AE
1444 obj_request);
1445 }
1446}
1447
57acbaa7 1448static bool obj_request_done_test(struct rbd_obj_request *obj_request)
6365d33a
AE
1449{
1450 smp_mb();
57acbaa7 1451 return test_bit(OBJ_REQ_DONE, &obj_request->flags) != 0;
6365d33a
AE
1452}
1453
5679c59f
AE
1454/*
1455 * This sets the KNOWN flag after (possibly) setting the EXISTS
1456 * flag. The latter is set based on the "exists" value provided.
1457 *
1458 * Note that for our purposes once an object exists it never goes
1459 * away again. It's possible that the response from two existence
1460 * checks are separated by the creation of the target object, and
1461 * the first ("doesn't exist") response arrives *after* the second
1462 * ("does exist"). In that case we ignore the second one.
1463 */
1464static void obj_request_existence_set(struct rbd_obj_request *obj_request,
1465 bool exists)
1466{
1467 if (exists)
1468 set_bit(OBJ_REQ_EXISTS, &obj_request->flags);
1469 set_bit(OBJ_REQ_KNOWN, &obj_request->flags);
1470 smp_mb();
1471}
1472
1473static bool obj_request_known_test(struct rbd_obj_request *obj_request)
1474{
1475 smp_mb();
1476 return test_bit(OBJ_REQ_KNOWN, &obj_request->flags) != 0;
1477}
1478
1479static bool obj_request_exists_test(struct rbd_obj_request *obj_request)
1480{
1481 smp_mb();
1482 return test_bit(OBJ_REQ_EXISTS, &obj_request->flags) != 0;
1483}
1484
9638556a
ID
1485static bool obj_request_overlaps_parent(struct rbd_obj_request *obj_request)
1486{
1487 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1488
1489 return obj_request->img_offset <
1490 round_up(rbd_dev->parent_overlap, rbd_obj_bytes(&rbd_dev->header));
1491}
1492
bf0d5f50
AE
1493static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
1494{
37206ee5 1495 dout("%s: obj %p (was %d)\n", __func__, obj_request,
2c935bc5 1496 kref_read(&obj_request->kref));
bf0d5f50
AE
1497 kref_get(&obj_request->kref);
1498}
1499
1500static void rbd_obj_request_destroy(struct kref *kref);
1501static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1502{
1503 rbd_assert(obj_request != NULL);
37206ee5 1504 dout("%s: obj %p (was %d)\n", __func__, obj_request,
2c935bc5 1505 kref_read(&obj_request->kref));
bf0d5f50
AE
1506 kref_put(&obj_request->kref, rbd_obj_request_destroy);
1507}
1508
0f2d5be7
AE
1509static void rbd_img_request_get(struct rbd_img_request *img_request)
1510{
1511 dout("%s: img %p (was %d)\n", __func__, img_request,
2c935bc5 1512 kref_read(&img_request->kref));
0f2d5be7
AE
1513 kref_get(&img_request->kref);
1514}
1515
e93f3152
AE
1516static bool img_request_child_test(struct rbd_img_request *img_request);
1517static void rbd_parent_request_destroy(struct kref *kref);
bf0d5f50
AE
1518static void rbd_img_request_destroy(struct kref *kref);
1519static void rbd_img_request_put(struct rbd_img_request *img_request)
1520{
1521 rbd_assert(img_request != NULL);
37206ee5 1522 dout("%s: img %p (was %d)\n", __func__, img_request,
2c935bc5 1523 kref_read(&img_request->kref));
e93f3152
AE
1524 if (img_request_child_test(img_request))
1525 kref_put(&img_request->kref, rbd_parent_request_destroy);
1526 else
1527 kref_put(&img_request->kref, rbd_img_request_destroy);
bf0d5f50
AE
1528}
1529
1530static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1531 struct rbd_obj_request *obj_request)
1532{
25dcf954
AE
1533 rbd_assert(obj_request->img_request == NULL);
1534
b155e86c 1535 /* Image request now owns object's original reference */
bf0d5f50 1536 obj_request->img_request = img_request;
25dcf954 1537 obj_request->which = img_request->obj_request_count;
6365d33a
AE
1538 rbd_assert(!obj_request_img_data_test(obj_request));
1539 obj_request_img_data_set(obj_request);
bf0d5f50 1540 rbd_assert(obj_request->which != BAD_WHICH);
25dcf954
AE
1541 img_request->obj_request_count++;
1542 list_add_tail(&obj_request->links, &img_request->obj_requests);
37206ee5
AE
1543 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1544 obj_request->which);
bf0d5f50
AE
1545}
1546
1547static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1548 struct rbd_obj_request *obj_request)
1549{
1550 rbd_assert(obj_request->which != BAD_WHICH);
25dcf954 1551
37206ee5
AE
1552 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1553 obj_request->which);
bf0d5f50 1554 list_del(&obj_request->links);
25dcf954
AE
1555 rbd_assert(img_request->obj_request_count > 0);
1556 img_request->obj_request_count--;
1557 rbd_assert(obj_request->which == img_request->obj_request_count);
1558 obj_request->which = BAD_WHICH;
6365d33a 1559 rbd_assert(obj_request_img_data_test(obj_request));
bf0d5f50 1560 rbd_assert(obj_request->img_request == img_request);
bf0d5f50 1561 obj_request->img_request = NULL;
25dcf954 1562 obj_request->callback = NULL;
bf0d5f50
AE
1563 rbd_obj_request_put(obj_request);
1564}
1565
1566static bool obj_request_type_valid(enum obj_request_type type)
1567{
1568 switch (type) {
9969ebc5 1569 case OBJ_REQUEST_NODATA:
bf0d5f50 1570 case OBJ_REQUEST_BIO:
788e2df3 1571 case OBJ_REQUEST_PAGES:
bf0d5f50
AE
1572 return true;
1573 default:
1574 return false;
1575 }
1576}
1577
4a17dadc
ID
1578static void rbd_img_obj_callback(struct rbd_obj_request *obj_request);
1579
980917fc 1580static void rbd_obj_request_submit(struct rbd_obj_request *obj_request)
bf0d5f50 1581{
980917fc
ID
1582 struct ceph_osd_request *osd_req = obj_request->osd_req;
1583
a90bb0c1
ID
1584 dout("%s %p object_no %016llx %llu~%llu osd_req %p\n", __func__,
1585 obj_request, obj_request->object_no, obj_request->offset,
67e2b652 1586 obj_request->length, osd_req);
4a17dadc
ID
1587 if (obj_request_img_data_test(obj_request)) {
1588 WARN_ON(obj_request->callback != rbd_img_obj_callback);
1589 rbd_img_request_get(obj_request->img_request);
1590 }
980917fc 1591 ceph_osdc_start_request(osd_req->r_osdc, osd_req, false);
bf0d5f50
AE
1592}
1593
1594static void rbd_img_request_complete(struct rbd_img_request *img_request)
1595{
55f27e09 1596
37206ee5 1597 dout("%s: img %p\n", __func__, img_request);
55f27e09
AE
1598
1599 /*
1600 * If no error occurred, compute the aggregate transfer
1601 * count for the image request. We could instead use
1602 * atomic64_cmpxchg() to update it as each object request
1603 * completes; not clear which way is better off hand.
1604 */
1605 if (!img_request->result) {
1606 struct rbd_obj_request *obj_request;
1607 u64 xferred = 0;
1608
1609 for_each_obj_request(img_request, obj_request)
1610 xferred += obj_request->xferred;
1611 img_request->xferred = xferred;
1612 }
1613
bf0d5f50
AE
1614 if (img_request->callback)
1615 img_request->callback(img_request);
1616 else
1617 rbd_img_request_put(img_request);
1618}
1619
0c425248
AE
1620/*
1621 * The default/initial value for all image request flags is 0. Each
1622 * is conditionally set to 1 at image request initialization time
1623 * and currently never change thereafter.
1624 */
1625static void img_request_write_set(struct rbd_img_request *img_request)
1626{
1627 set_bit(IMG_REQ_WRITE, &img_request->flags);
1628 smp_mb();
1629}
1630
1631static bool img_request_write_test(struct rbd_img_request *img_request)
1632{
1633 smp_mb();
1634 return test_bit(IMG_REQ_WRITE, &img_request->flags) != 0;
1635}
1636
90e98c52
GZ
1637/*
1638 * Set the discard flag when the img_request is an discard request
1639 */
1640static void img_request_discard_set(struct rbd_img_request *img_request)
1641{
1642 set_bit(IMG_REQ_DISCARD, &img_request->flags);
1643 smp_mb();
1644}
1645
1646static bool img_request_discard_test(struct rbd_img_request *img_request)
1647{
1648 smp_mb();
1649 return test_bit(IMG_REQ_DISCARD, &img_request->flags) != 0;
1650}
1651
9849e986
AE
1652static void img_request_child_set(struct rbd_img_request *img_request)
1653{
1654 set_bit(IMG_REQ_CHILD, &img_request->flags);
1655 smp_mb();
1656}
1657
e93f3152
AE
1658static void img_request_child_clear(struct rbd_img_request *img_request)
1659{
1660 clear_bit(IMG_REQ_CHILD, &img_request->flags);
1661 smp_mb();
1662}
1663
9849e986
AE
1664static bool img_request_child_test(struct rbd_img_request *img_request)
1665{
1666 smp_mb();
1667 return test_bit(IMG_REQ_CHILD, &img_request->flags) != 0;
1668}
1669
d0b2e944
AE
1670static void img_request_layered_set(struct rbd_img_request *img_request)
1671{
1672 set_bit(IMG_REQ_LAYERED, &img_request->flags);
1673 smp_mb();
1674}
1675
a2acd00e
AE
1676static void img_request_layered_clear(struct rbd_img_request *img_request)
1677{
1678 clear_bit(IMG_REQ_LAYERED, &img_request->flags);
1679 smp_mb();
1680}
1681
d0b2e944
AE
1682static bool img_request_layered_test(struct rbd_img_request *img_request)
1683{
1684 smp_mb();
1685 return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1686}
1687
3b434a2a
JD
1688static enum obj_operation_type
1689rbd_img_request_op_type(struct rbd_img_request *img_request)
1690{
1691 if (img_request_write_test(img_request))
1692 return OBJ_OP_WRITE;
1693 else if (img_request_discard_test(img_request))
1694 return OBJ_OP_DISCARD;
1695 else
1696 return OBJ_OP_READ;
1697}
1698
6e2a4505
AE
1699static void
1700rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
1701{
b9434c5b
AE
1702 u64 xferred = obj_request->xferred;
1703 u64 length = obj_request->length;
1704
6e2a4505
AE
1705 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1706 obj_request, obj_request->img_request, obj_request->result,
b9434c5b 1707 xferred, length);
6e2a4505 1708 /*
17c1cc1d
JD
1709 * ENOENT means a hole in the image. We zero-fill the entire
1710 * length of the request. A short read also implies zero-fill
1711 * to the end of the request. An error requires the whole
1712 * length of the request to be reported finished with an error
1713 * to the block layer. In each case we update the xferred
1714 * count to indicate the whole request was satisfied.
6e2a4505 1715 */
b9434c5b 1716 rbd_assert(obj_request->type != OBJ_REQUEST_NODATA);
6e2a4505 1717 if (obj_request->result == -ENOENT) {
b9434c5b
AE
1718 if (obj_request->type == OBJ_REQUEST_BIO)
1719 zero_bio_chain(obj_request->bio_list, 0);
1720 else
1721 zero_pages(obj_request->pages, 0, length);
6e2a4505 1722 obj_request->result = 0;
b9434c5b
AE
1723 } else if (xferred < length && !obj_request->result) {
1724 if (obj_request->type == OBJ_REQUEST_BIO)
1725 zero_bio_chain(obj_request->bio_list, xferred);
1726 else
1727 zero_pages(obj_request->pages, xferred, length);
6e2a4505 1728 }
17c1cc1d 1729 obj_request->xferred = length;
6e2a4505
AE
1730 obj_request_done_set(obj_request);
1731}
1732
bf0d5f50
AE
1733static void rbd_obj_request_complete(struct rbd_obj_request *obj_request)
1734{
37206ee5
AE
1735 dout("%s: obj %p cb %p\n", __func__, obj_request,
1736 obj_request->callback);
bf0d5f50
AE
1737 if (obj_request->callback)
1738 obj_request->callback(obj_request);
788e2df3
AE
1739 else
1740 complete_all(&obj_request->completion);
bf0d5f50
AE
1741}
1742
0dcc685e
ID
1743static void rbd_obj_request_error(struct rbd_obj_request *obj_request, int err)
1744{
1745 obj_request->result = err;
1746 obj_request->xferred = 0;
1747 /*
1748 * kludge - mirror rbd_obj_request_submit() to match a put in
1749 * rbd_img_obj_callback()
1750 */
1751 if (obj_request_img_data_test(obj_request)) {
1752 WARN_ON(obj_request->callback != rbd_img_obj_callback);
1753 rbd_img_request_get(obj_request->img_request);
1754 }
1755 obj_request_done_set(obj_request);
1756 rbd_obj_request_complete(obj_request);
1757}
1758
c47f9371 1759static void rbd_osd_read_callback(struct rbd_obj_request *obj_request)
bf0d5f50 1760{
57acbaa7 1761 struct rbd_img_request *img_request = NULL;
a9e8ba2c 1762 struct rbd_device *rbd_dev = NULL;
57acbaa7
AE
1763 bool layered = false;
1764
1765 if (obj_request_img_data_test(obj_request)) {
1766 img_request = obj_request->img_request;
1767 layered = img_request && img_request_layered_test(img_request);
a9e8ba2c 1768 rbd_dev = img_request->rbd_dev;
57acbaa7 1769 }
8b3e1a56
AE
1770
1771 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1772 obj_request, img_request, obj_request->result,
1773 obj_request->xferred, obj_request->length);
a9e8ba2c
AE
1774 if (layered && obj_request->result == -ENOENT &&
1775 obj_request->img_offset < rbd_dev->parent_overlap)
8b3e1a56
AE
1776 rbd_img_parent_read(obj_request);
1777 else if (img_request)
6e2a4505
AE
1778 rbd_img_obj_request_read_callback(obj_request);
1779 else
1780 obj_request_done_set(obj_request);
bf0d5f50
AE
1781}
1782
c47f9371 1783static void rbd_osd_write_callback(struct rbd_obj_request *obj_request)
bf0d5f50 1784{
1b83bef2
SW
1785 dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1786 obj_request->result, obj_request->length);
1787 /*
8b3e1a56
AE
1788 * There is no such thing as a successful short write. Set
1789 * it to our originally-requested length.
1b83bef2
SW
1790 */
1791 obj_request->xferred = obj_request->length;
07741308 1792 obj_request_done_set(obj_request);
bf0d5f50
AE
1793}
1794
90e98c52
GZ
1795static void rbd_osd_discard_callback(struct rbd_obj_request *obj_request)
1796{
1797 dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1798 obj_request->result, obj_request->length);
1799 /*
1800 * There is no such thing as a successful short discard. Set
1801 * it to our originally-requested length.
1802 */
1803 obj_request->xferred = obj_request->length;
d0265de7
JD
1804 /* discarding a non-existent object is not a problem */
1805 if (obj_request->result == -ENOENT)
1806 obj_request->result = 0;
90e98c52
GZ
1807 obj_request_done_set(obj_request);
1808}
1809
fbfab539
AE
1810/*
1811 * For a simple stat call there's nothing to do. We'll do more if
1812 * this is part of a write sequence for a layered image.
1813 */
c47f9371 1814static void rbd_osd_stat_callback(struct rbd_obj_request *obj_request)
fbfab539 1815{
37206ee5 1816 dout("%s: obj %p\n", __func__, obj_request);
fbfab539
AE
1817 obj_request_done_set(obj_request);
1818}
1819
2761713d
ID
1820static void rbd_osd_call_callback(struct rbd_obj_request *obj_request)
1821{
1822 dout("%s: obj %p\n", __func__, obj_request);
1823
1824 if (obj_request_img_data_test(obj_request))
1825 rbd_osd_copyup_callback(obj_request);
1826 else
1827 obj_request_done_set(obj_request);
1828}
1829
85e084fe 1830static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
bf0d5f50
AE
1831{
1832 struct rbd_obj_request *obj_request = osd_req->r_priv;
bf0d5f50
AE
1833 u16 opcode;
1834
85e084fe 1835 dout("%s: osd_req %p\n", __func__, osd_req);
bf0d5f50 1836 rbd_assert(osd_req == obj_request->osd_req);
57acbaa7
AE
1837 if (obj_request_img_data_test(obj_request)) {
1838 rbd_assert(obj_request->img_request);
1839 rbd_assert(obj_request->which != BAD_WHICH);
1840 } else {
1841 rbd_assert(obj_request->which == BAD_WHICH);
1842 }
bf0d5f50 1843
1b83bef2
SW
1844 if (osd_req->r_result < 0)
1845 obj_request->result = osd_req->r_result;
bf0d5f50 1846
c47f9371
AE
1847 /*
1848 * We support a 64-bit length, but ultimately it has to be
7ad18afa
CH
1849 * passed to the block layer, which just supports a 32-bit
1850 * length field.
c47f9371 1851 */
7665d85b 1852 obj_request->xferred = osd_req->r_ops[0].outdata_len;
8b3e1a56 1853 rbd_assert(obj_request->xferred < (u64)UINT_MAX);
0ccd5926 1854
79528734 1855 opcode = osd_req->r_ops[0].op;
bf0d5f50
AE
1856 switch (opcode) {
1857 case CEPH_OSD_OP_READ:
c47f9371 1858 rbd_osd_read_callback(obj_request);
bf0d5f50 1859 break;
0ccd5926 1860 case CEPH_OSD_OP_SETALLOCHINT:
e30b7577
ID
1861 rbd_assert(osd_req->r_ops[1].op == CEPH_OSD_OP_WRITE ||
1862 osd_req->r_ops[1].op == CEPH_OSD_OP_WRITEFULL);
0ccd5926 1863 /* fall through */
bf0d5f50 1864 case CEPH_OSD_OP_WRITE:
e30b7577 1865 case CEPH_OSD_OP_WRITEFULL:
c47f9371 1866 rbd_osd_write_callback(obj_request);
bf0d5f50 1867 break;
fbfab539 1868 case CEPH_OSD_OP_STAT:
c47f9371 1869 rbd_osd_stat_callback(obj_request);
fbfab539 1870 break;
90e98c52
GZ
1871 case CEPH_OSD_OP_DELETE:
1872 case CEPH_OSD_OP_TRUNCATE:
1873 case CEPH_OSD_OP_ZERO:
1874 rbd_osd_discard_callback(obj_request);
1875 break;
36be9a76 1876 case CEPH_OSD_OP_CALL:
2761713d
ID
1877 rbd_osd_call_callback(obj_request);
1878 break;
bf0d5f50 1879 default:
a90bb0c1
ID
1880 rbd_warn(NULL, "unexpected OSD op: object_no %016llx opcode %d",
1881 obj_request->object_no, opcode);
bf0d5f50
AE
1882 break;
1883 }
1884
07741308 1885 if (obj_request_done_test(obj_request))
bf0d5f50
AE
1886 rbd_obj_request_complete(obj_request);
1887}
1888
9d4df01f 1889static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request)
430c28c3 1890{
8c042b0d 1891 struct ceph_osd_request *osd_req = obj_request->osd_req;
430c28c3 1892
7c84883a
ID
1893 rbd_assert(obj_request_img_data_test(obj_request));
1894 osd_req->r_snapid = obj_request->img_request->snap_id;
9d4df01f
AE
1895}
1896
1897static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request)
1898{
9d4df01f 1899 struct ceph_osd_request *osd_req = obj_request->osd_req;
9d4df01f 1900
1134e091 1901 ktime_get_real_ts(&osd_req->r_mtime);
bb873b53 1902 osd_req->r_data_offset = obj_request->offset;
430c28c3
AE
1903}
1904
bc81207e
ID
1905static struct ceph_osd_request *
1906__rbd_osd_req_create(struct rbd_device *rbd_dev,
1907 struct ceph_snap_context *snapc,
1908 int num_ops, unsigned int flags,
1909 struct rbd_obj_request *obj_request)
1910{
1911 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1912 struct ceph_osd_request *req;
a90bb0c1
ID
1913 const char *name_format = rbd_dev->image_format == 1 ?
1914 RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
bc81207e
ID
1915
1916 req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, GFP_NOIO);
1917 if (!req)
1918 return NULL;
1919
1920 req->r_flags = flags;
1921 req->r_callback = rbd_osd_req_callback;
1922 req->r_priv = obj_request;
1923
1924 req->r_base_oloc.pool = rbd_dev->layout.pool_id;
a90bb0c1
ID
1925 if (ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
1926 rbd_dev->header.object_prefix, obj_request->object_no))
bc81207e
ID
1927 goto err_req;
1928
1929 if (ceph_osdc_alloc_messages(req, GFP_NOIO))
1930 goto err_req;
1931
1932 return req;
1933
1934err_req:
1935 ceph_osdc_put_request(req);
1936 return NULL;
1937}
1938
0ccd5926
ID
1939/*
1940 * Create an osd request. A read request has one osd op (read).
1941 * A write request has either one (watch) or two (hint+write) osd ops.
1942 * (All rbd data writes are prefixed with an allocation hint op, but
1943 * technically osd watch is a write request, hence this distinction.)
1944 */
bf0d5f50
AE
1945static struct ceph_osd_request *rbd_osd_req_create(
1946 struct rbd_device *rbd_dev,
6d2940c8 1947 enum obj_operation_type op_type,
deb236b3 1948 unsigned int num_ops,
430c28c3 1949 struct rbd_obj_request *obj_request)
bf0d5f50 1950{
bf0d5f50 1951 struct ceph_snap_context *snapc = NULL;
bf0d5f50 1952
90e98c52
GZ
1953 if (obj_request_img_data_test(obj_request) &&
1954 (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_WRITE)) {
6365d33a 1955 struct rbd_img_request *img_request = obj_request->img_request;
90e98c52
GZ
1956 if (op_type == OBJ_OP_WRITE) {
1957 rbd_assert(img_request_write_test(img_request));
1958 } else {
1959 rbd_assert(img_request_discard_test(img_request));
1960 }
6d2940c8 1961 snapc = img_request->snapc;
bf0d5f50
AE
1962 }
1963
6d2940c8 1964 rbd_assert(num_ops == 1 || ((op_type == OBJ_OP_WRITE) && num_ops == 2));
deb236b3 1965
bc81207e
ID
1966 return __rbd_osd_req_create(rbd_dev, snapc, num_ops,
1967 (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD) ?
54ea0046 1968 CEPH_OSD_FLAG_WRITE : CEPH_OSD_FLAG_READ, obj_request);
bf0d5f50
AE
1969}
1970
0eefd470 1971/*
d3246fb0
JD
1972 * Create a copyup osd request based on the information in the object
1973 * request supplied. A copyup request has two or three osd ops, a
1974 * copyup method call, potentially a hint op, and a write or truncate
1975 * or zero op.
0eefd470
AE
1976 */
1977static struct ceph_osd_request *
1978rbd_osd_req_create_copyup(struct rbd_obj_request *obj_request)
1979{
1980 struct rbd_img_request *img_request;
d3246fb0 1981 int num_osd_ops = 3;
0eefd470
AE
1982
1983 rbd_assert(obj_request_img_data_test(obj_request));
1984 img_request = obj_request->img_request;
1985 rbd_assert(img_request);
d3246fb0
JD
1986 rbd_assert(img_request_write_test(img_request) ||
1987 img_request_discard_test(img_request));
0eefd470 1988
d3246fb0
JD
1989 if (img_request_discard_test(img_request))
1990 num_osd_ops = 2;
1991
bc81207e
ID
1992 return __rbd_osd_req_create(img_request->rbd_dev,
1993 img_request->snapc, num_osd_ops,
54ea0046 1994 CEPH_OSD_FLAG_WRITE, obj_request);
0eefd470
AE
1995}
1996
bf0d5f50
AE
1997static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req)
1998{
1999 ceph_osdc_put_request(osd_req);
2000}
2001
6c696d85
ID
2002static struct rbd_obj_request *
2003rbd_obj_request_create(enum obj_request_type type)
bf0d5f50
AE
2004{
2005 struct rbd_obj_request *obj_request;
bf0d5f50
AE
2006
2007 rbd_assert(obj_request_type_valid(type));
2008
5a60e876 2009 obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_NOIO);
6c696d85 2010 if (!obj_request)
f907ad55 2011 return NULL;
f907ad55 2012
bf0d5f50
AE
2013 obj_request->which = BAD_WHICH;
2014 obj_request->type = type;
2015 INIT_LIST_HEAD(&obj_request->links);
788e2df3 2016 init_completion(&obj_request->completion);
bf0d5f50
AE
2017 kref_init(&obj_request->kref);
2018
67e2b652 2019 dout("%s %p\n", __func__, obj_request);
bf0d5f50
AE
2020 return obj_request;
2021}
2022
2023static void rbd_obj_request_destroy(struct kref *kref)
2024{
2025 struct rbd_obj_request *obj_request;
2026
2027 obj_request = container_of(kref, struct rbd_obj_request, kref);
2028
37206ee5
AE
2029 dout("%s: obj %p\n", __func__, obj_request);
2030
bf0d5f50
AE
2031 rbd_assert(obj_request->img_request == NULL);
2032 rbd_assert(obj_request->which == BAD_WHICH);
2033
2034 if (obj_request->osd_req)
2035 rbd_osd_req_destroy(obj_request->osd_req);
2036
2037 rbd_assert(obj_request_type_valid(obj_request->type));
2038 switch (obj_request->type) {
9969ebc5
AE
2039 case OBJ_REQUEST_NODATA:
2040 break; /* Nothing to do */
bf0d5f50
AE
2041 case OBJ_REQUEST_BIO:
2042 if (obj_request->bio_list)
2043 bio_chain_put(obj_request->bio_list);
2044 break;
788e2df3 2045 case OBJ_REQUEST_PAGES:
04dc923c
ID
2046 /* img_data requests don't own their page array */
2047 if (obj_request->pages &&
2048 !obj_request_img_data_test(obj_request))
788e2df3
AE
2049 ceph_release_page_vector(obj_request->pages,
2050 obj_request->page_count);
2051 break;
bf0d5f50
AE
2052 }
2053
868311b1 2054 kmem_cache_free(rbd_obj_request_cache, obj_request);
bf0d5f50
AE
2055}
2056
fb65d228
AE
2057/* It's OK to call this for a device with no parent */
2058
2059static void rbd_spec_put(struct rbd_spec *spec);
2060static void rbd_dev_unparent(struct rbd_device *rbd_dev)
2061{
2062 rbd_dev_remove_parent(rbd_dev);
2063 rbd_spec_put(rbd_dev->parent_spec);
2064 rbd_dev->parent_spec = NULL;
2065 rbd_dev->parent_overlap = 0;
2066}
2067
a2acd00e
AE
2068/*
2069 * Parent image reference counting is used to determine when an
2070 * image's parent fields can be safely torn down--after there are no
2071 * more in-flight requests to the parent image. When the last
2072 * reference is dropped, cleaning them up is safe.
2073 */
2074static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
2075{
2076 int counter;
2077
2078 if (!rbd_dev->parent_spec)
2079 return;
2080
2081 counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
2082 if (counter > 0)
2083 return;
2084
2085 /* Last reference; clean up parent data structures */
2086
2087 if (!counter)
2088 rbd_dev_unparent(rbd_dev);
2089 else
9584d508 2090 rbd_warn(rbd_dev, "parent reference underflow");
a2acd00e
AE
2091}
2092
2093/*
2094 * If an image has a non-zero parent overlap, get a reference to its
2095 * parent.
2096 *
2097 * Returns true if the rbd device has a parent with a non-zero
2098 * overlap and a reference for it was successfully taken, or
2099 * false otherwise.
2100 */
2101static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
2102{
ae43e9d0 2103 int counter = 0;
a2acd00e
AE
2104
2105 if (!rbd_dev->parent_spec)
2106 return false;
2107
ae43e9d0
ID
2108 down_read(&rbd_dev->header_rwsem);
2109 if (rbd_dev->parent_overlap)
2110 counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
2111 up_read(&rbd_dev->header_rwsem);
a2acd00e
AE
2112
2113 if (counter < 0)
9584d508 2114 rbd_warn(rbd_dev, "parent reference overflow");
a2acd00e 2115
ae43e9d0 2116 return counter > 0;
a2acd00e
AE
2117}
2118
bf0d5f50
AE
2119/*
2120 * Caller is responsible for filling in the list of object requests
2121 * that comprises the image request, and the Linux request pointer
2122 * (if there is one).
2123 */
cc344fa1
AE
2124static struct rbd_img_request *rbd_img_request_create(
2125 struct rbd_device *rbd_dev,
bf0d5f50 2126 u64 offset, u64 length,
6d2940c8 2127 enum obj_operation_type op_type,
4e752f0a 2128 struct ceph_snap_context *snapc)
bf0d5f50
AE
2129{
2130 struct rbd_img_request *img_request;
bf0d5f50 2131
7a716aac 2132 img_request = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
bf0d5f50
AE
2133 if (!img_request)
2134 return NULL;
2135
bf0d5f50
AE
2136 img_request->rq = NULL;
2137 img_request->rbd_dev = rbd_dev;
2138 img_request->offset = offset;
2139 img_request->length = length;
0c425248 2140 img_request->flags = 0;
90e98c52
GZ
2141 if (op_type == OBJ_OP_DISCARD) {
2142 img_request_discard_set(img_request);
2143 img_request->snapc = snapc;
2144 } else if (op_type == OBJ_OP_WRITE) {
0c425248 2145 img_request_write_set(img_request);
4e752f0a 2146 img_request->snapc = snapc;
0c425248 2147 } else {
bf0d5f50 2148 img_request->snap_id = rbd_dev->spec->snap_id;
0c425248 2149 }
a2acd00e 2150 if (rbd_dev_parent_get(rbd_dev))
d0b2e944 2151 img_request_layered_set(img_request);
bf0d5f50
AE
2152 spin_lock_init(&img_request->completion_lock);
2153 img_request->next_completion = 0;
2154 img_request->callback = NULL;
a5a337d4 2155 img_request->result = 0;
bf0d5f50
AE
2156 img_request->obj_request_count = 0;
2157 INIT_LIST_HEAD(&img_request->obj_requests);
2158 kref_init(&img_request->kref);
2159
37206ee5 2160 dout("%s: rbd_dev %p %s %llu/%llu -> img %p\n", __func__, rbd_dev,
6d2940c8 2161 obj_op_name(op_type), offset, length, img_request);
37206ee5 2162
bf0d5f50
AE
2163 return img_request;
2164}
2165
2166static void rbd_img_request_destroy(struct kref *kref)
2167{
2168 struct rbd_img_request *img_request;
2169 struct rbd_obj_request *obj_request;
2170 struct rbd_obj_request *next_obj_request;
2171
2172 img_request = container_of(kref, struct rbd_img_request, kref);
2173
37206ee5
AE
2174 dout("%s: img %p\n", __func__, img_request);
2175
bf0d5f50
AE
2176 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
2177 rbd_img_obj_request_del(img_request, obj_request);
25dcf954 2178 rbd_assert(img_request->obj_request_count == 0);
bf0d5f50 2179
a2acd00e
AE
2180 if (img_request_layered_test(img_request)) {
2181 img_request_layered_clear(img_request);
2182 rbd_dev_parent_put(img_request->rbd_dev);
2183 }
2184
bef95455
JD
2185 if (img_request_write_test(img_request) ||
2186 img_request_discard_test(img_request))
812164f8 2187 ceph_put_snap_context(img_request->snapc);
bf0d5f50 2188
1c2a9dfe 2189 kmem_cache_free(rbd_img_request_cache, img_request);
bf0d5f50
AE
2190}
2191
e93f3152
AE
2192static struct rbd_img_request *rbd_parent_request_create(
2193 struct rbd_obj_request *obj_request,
2194 u64 img_offset, u64 length)
2195{
2196 struct rbd_img_request *parent_request;
2197 struct rbd_device *rbd_dev;
2198
2199 rbd_assert(obj_request->img_request);
2200 rbd_dev = obj_request->img_request->rbd_dev;
2201
4e752f0a 2202 parent_request = rbd_img_request_create(rbd_dev->parent, img_offset,
6d2940c8 2203 length, OBJ_OP_READ, NULL);
e93f3152
AE
2204 if (!parent_request)
2205 return NULL;
2206
2207 img_request_child_set(parent_request);
2208 rbd_obj_request_get(obj_request);
2209 parent_request->obj_request = obj_request;
2210
2211 return parent_request;
2212}
2213
2214static void rbd_parent_request_destroy(struct kref *kref)
2215{
2216 struct rbd_img_request *parent_request;
2217 struct rbd_obj_request *orig_request;
2218
2219 parent_request = container_of(kref, struct rbd_img_request, kref);
2220 orig_request = parent_request->obj_request;
2221
2222 parent_request->obj_request = NULL;
2223 rbd_obj_request_put(orig_request);
2224 img_request_child_clear(parent_request);
2225
2226 rbd_img_request_destroy(kref);
2227}
2228
1217857f
AE
2229static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request)
2230{
6365d33a 2231 struct rbd_img_request *img_request;
1217857f
AE
2232 unsigned int xferred;
2233 int result;
8b3e1a56 2234 bool more;
1217857f 2235
6365d33a
AE
2236 rbd_assert(obj_request_img_data_test(obj_request));
2237 img_request = obj_request->img_request;
2238
1217857f
AE
2239 rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
2240 xferred = (unsigned int)obj_request->xferred;
2241 result = obj_request->result;
2242 if (result) {
2243 struct rbd_device *rbd_dev = img_request->rbd_dev;
6d2940c8
GZ
2244 enum obj_operation_type op_type;
2245
90e98c52
GZ
2246 if (img_request_discard_test(img_request))
2247 op_type = OBJ_OP_DISCARD;
2248 else if (img_request_write_test(img_request))
2249 op_type = OBJ_OP_WRITE;
2250 else
2251 op_type = OBJ_OP_READ;
1217857f 2252
9584d508 2253 rbd_warn(rbd_dev, "%s %llx at %llx (%llx)",
6d2940c8
GZ
2254 obj_op_name(op_type), obj_request->length,
2255 obj_request->img_offset, obj_request->offset);
9584d508 2256 rbd_warn(rbd_dev, " result %d xferred %x",
1217857f
AE
2257 result, xferred);
2258 if (!img_request->result)
2259 img_request->result = result;
082a75da
ID
2260 /*
2261 * Need to end I/O on the entire obj_request worth of
2262 * bytes in case of error.
2263 */
2264 xferred = obj_request->length;
1217857f
AE
2265 }
2266
8b3e1a56
AE
2267 if (img_request_child_test(img_request)) {
2268 rbd_assert(img_request->obj_request != NULL);
2269 more = obj_request->which < img_request->obj_request_count - 1;
2270 } else {
2a842aca
CH
2271 blk_status_t status = errno_to_blk_status(result);
2272
8b3e1a56 2273 rbd_assert(img_request->rq != NULL);
7ad18afa 2274
2a842aca 2275 more = blk_update_request(img_request->rq, status, xferred);
7ad18afa 2276 if (!more)
2a842aca 2277 __blk_mq_end_request(img_request->rq, status);
8b3e1a56
AE
2278 }
2279
2280 return more;
1217857f
AE
2281}
2282
2169238d
AE
2283static void rbd_img_obj_callback(struct rbd_obj_request *obj_request)
2284{
2285 struct rbd_img_request *img_request;
2286 u32 which = obj_request->which;
2287 bool more = true;
2288
6365d33a 2289 rbd_assert(obj_request_img_data_test(obj_request));
2169238d
AE
2290 img_request = obj_request->img_request;
2291
2292 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
2293 rbd_assert(img_request != NULL);
2169238d
AE
2294 rbd_assert(img_request->obj_request_count > 0);
2295 rbd_assert(which != BAD_WHICH);
2296 rbd_assert(which < img_request->obj_request_count);
2169238d
AE
2297
2298 spin_lock_irq(&img_request->completion_lock);
2299 if (which != img_request->next_completion)
2300 goto out;
2301
2302 for_each_obj_request_from(img_request, obj_request) {
2169238d
AE
2303 rbd_assert(more);
2304 rbd_assert(which < img_request->obj_request_count);
2305
2306 if (!obj_request_done_test(obj_request))
2307 break;
1217857f 2308 more = rbd_img_obj_end_request(obj_request);
2169238d
AE
2309 which++;
2310 }
2311
2312 rbd_assert(more ^ (which == img_request->obj_request_count));
2313 img_request->next_completion = which;
2314out:
2315 spin_unlock_irq(&img_request->completion_lock);
0f2d5be7 2316 rbd_img_request_put(img_request);
2169238d
AE
2317
2318 if (!more)
2319 rbd_img_request_complete(img_request);
2320}
2321
3b434a2a
JD
2322/*
2323 * Add individual osd ops to the given ceph_osd_request and prepare
2324 * them for submission. num_ops is the current number of
2325 * osd operations already to the object request.
2326 */
2327static void rbd_img_obj_request_fill(struct rbd_obj_request *obj_request,
2328 struct ceph_osd_request *osd_request,
2329 enum obj_operation_type op_type,
2330 unsigned int num_ops)
2331{
2332 struct rbd_img_request *img_request = obj_request->img_request;
2333 struct rbd_device *rbd_dev = img_request->rbd_dev;
2334 u64 object_size = rbd_obj_bytes(&rbd_dev->header);
2335 u64 offset = obj_request->offset;
2336 u64 length = obj_request->length;
2337 u64 img_end;
2338 u16 opcode;
2339
2340 if (op_type == OBJ_OP_DISCARD) {
d3246fb0
JD
2341 if (!offset && length == object_size &&
2342 (!img_request_layered_test(img_request) ||
2343 !obj_request_overlaps_parent(obj_request))) {
3b434a2a
JD
2344 opcode = CEPH_OSD_OP_DELETE;
2345 } else if ((offset + length == object_size)) {
2346 opcode = CEPH_OSD_OP_TRUNCATE;
2347 } else {
2348 down_read(&rbd_dev->header_rwsem);
2349 img_end = rbd_dev->header.image_size;
2350 up_read(&rbd_dev->header_rwsem);
2351
2352 if (obj_request->img_offset + length == img_end)
2353 opcode = CEPH_OSD_OP_TRUNCATE;
2354 else
2355 opcode = CEPH_OSD_OP_ZERO;
2356 }
2357 } else if (op_type == OBJ_OP_WRITE) {
e30b7577
ID
2358 if (!offset && length == object_size)
2359 opcode = CEPH_OSD_OP_WRITEFULL;
2360 else
2361 opcode = CEPH_OSD_OP_WRITE;
3b434a2a
JD
2362 osd_req_op_alloc_hint_init(osd_request, num_ops,
2363 object_size, object_size);
2364 num_ops++;
2365 } else {
2366 opcode = CEPH_OSD_OP_READ;
2367 }
2368
7e868b6e 2369 if (opcode == CEPH_OSD_OP_DELETE)
144cba14 2370 osd_req_op_init(osd_request, num_ops, opcode, 0);
7e868b6e
ID
2371 else
2372 osd_req_op_extent_init(osd_request, num_ops, opcode,
2373 offset, length, 0, 0);
2374
3b434a2a
JD
2375 if (obj_request->type == OBJ_REQUEST_BIO)
2376 osd_req_op_extent_osd_data_bio(osd_request, num_ops,
2377 obj_request->bio_list, length);
2378 else if (obj_request->type == OBJ_REQUEST_PAGES)
2379 osd_req_op_extent_osd_data_pages(osd_request, num_ops,
2380 obj_request->pages, length,
2381 offset & ~PAGE_MASK, false, false);
2382
2383 /* Discards are also writes */
2384 if (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD)
2385 rbd_osd_req_format_write(obj_request);
2386 else
2387 rbd_osd_req_format_read(obj_request);
2388}
2389
f1a4739f
AE
2390/*
2391 * Split up an image request into one or more object requests, each
2392 * to a different object. The "type" parameter indicates whether
2393 * "data_desc" is the pointer to the head of a list of bio
2394 * structures, or the base of a page array. In either case this
2395 * function assumes data_desc describes memory sufficient to hold
2396 * all data described by the image request.
2397 */
2398static int rbd_img_request_fill(struct rbd_img_request *img_request,
2399 enum obj_request_type type,
2400 void *data_desc)
bf0d5f50
AE
2401{
2402 struct rbd_device *rbd_dev = img_request->rbd_dev;
2403 struct rbd_obj_request *obj_request = NULL;
2404 struct rbd_obj_request *next_obj_request;
a158073c 2405 struct bio *bio_list = NULL;
f1a4739f 2406 unsigned int bio_offset = 0;
a158073c 2407 struct page **pages = NULL;
6d2940c8 2408 enum obj_operation_type op_type;
7da22d29 2409 u64 img_offset;
bf0d5f50 2410 u64 resid;
bf0d5f50 2411
f1a4739f
AE
2412 dout("%s: img %p type %d data_desc %p\n", __func__, img_request,
2413 (int)type, data_desc);
37206ee5 2414
7da22d29 2415 img_offset = img_request->offset;
bf0d5f50 2416 resid = img_request->length;
4dda41d3 2417 rbd_assert(resid > 0);
3b434a2a 2418 op_type = rbd_img_request_op_type(img_request);
f1a4739f
AE
2419
2420 if (type == OBJ_REQUEST_BIO) {
2421 bio_list = data_desc;
4f024f37
KO
2422 rbd_assert(img_offset ==
2423 bio_list->bi_iter.bi_sector << SECTOR_SHIFT);
90e98c52 2424 } else if (type == OBJ_REQUEST_PAGES) {
f1a4739f
AE
2425 pages = data_desc;
2426 }
2427
bf0d5f50 2428 while (resid) {
2fa12320 2429 struct ceph_osd_request *osd_req;
a90bb0c1 2430 u64 object_no = img_offset >> rbd_dev->header.obj_order;
67e2b652
ID
2431 u64 offset = rbd_segment_offset(rbd_dev, img_offset);
2432 u64 length = rbd_segment_length(rbd_dev, img_offset, resid);
bf0d5f50 2433
6c696d85 2434 obj_request = rbd_obj_request_create(type);
bf0d5f50
AE
2435 if (!obj_request)
2436 goto out_unwind;
62054da6 2437
a90bb0c1 2438 obj_request->object_no = object_no;
67e2b652
ID
2439 obj_request->offset = offset;
2440 obj_request->length = length;
2441
03507db6
JD
2442 /*
2443 * set obj_request->img_request before creating the
2444 * osd_request so that it gets the right snapc
2445 */
2446 rbd_img_obj_request_add(img_request, obj_request);
bf0d5f50 2447
f1a4739f
AE
2448 if (type == OBJ_REQUEST_BIO) {
2449 unsigned int clone_size;
2450
2451 rbd_assert(length <= (u64)UINT_MAX);
2452 clone_size = (unsigned int)length;
2453 obj_request->bio_list =
2454 bio_chain_clone_range(&bio_list,
2455 &bio_offset,
2456 clone_size,
2224d879 2457 GFP_NOIO);
f1a4739f 2458 if (!obj_request->bio_list)
62054da6 2459 goto out_unwind;
90e98c52 2460 } else if (type == OBJ_REQUEST_PAGES) {
f1a4739f
AE
2461 unsigned int page_count;
2462
2463 obj_request->pages = pages;
2464 page_count = (u32)calc_pages_for(offset, length);
2465 obj_request->page_count = page_count;
2466 if ((offset + length) & ~PAGE_MASK)
2467 page_count--; /* more on last page */
2468 pages += page_count;
2469 }
bf0d5f50 2470
6d2940c8
GZ
2471 osd_req = rbd_osd_req_create(rbd_dev, op_type,
2472 (op_type == OBJ_OP_WRITE) ? 2 : 1,
2473 obj_request);
2fa12320 2474 if (!osd_req)
62054da6 2475 goto out_unwind;
3b434a2a 2476
2fa12320 2477 obj_request->osd_req = osd_req;
2169238d 2478 obj_request->callback = rbd_img_obj_callback;
3b434a2a 2479 obj_request->img_offset = img_offset;
9d4df01f 2480
3b434a2a 2481 rbd_img_obj_request_fill(obj_request, osd_req, op_type, 0);
430c28c3 2482
7da22d29 2483 img_offset += length;
bf0d5f50
AE
2484 resid -= length;
2485 }
2486
2487 return 0;
2488
bf0d5f50
AE
2489out_unwind:
2490 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
42dd037c 2491 rbd_img_obj_request_del(img_request, obj_request);
bf0d5f50
AE
2492
2493 return -ENOMEM;
2494}
2495
0eefd470 2496static void
2761713d 2497rbd_osd_copyup_callback(struct rbd_obj_request *obj_request)
0eefd470
AE
2498{
2499 struct rbd_img_request *img_request;
2500 struct rbd_device *rbd_dev;
ebda6408 2501 struct page **pages;
0eefd470
AE
2502 u32 page_count;
2503
2761713d
ID
2504 dout("%s: obj %p\n", __func__, obj_request);
2505
d3246fb0
JD
2506 rbd_assert(obj_request->type == OBJ_REQUEST_BIO ||
2507 obj_request->type == OBJ_REQUEST_NODATA);
0eefd470
AE
2508 rbd_assert(obj_request_img_data_test(obj_request));
2509 img_request = obj_request->img_request;
2510 rbd_assert(img_request);
2511
2512 rbd_dev = img_request->rbd_dev;
2513 rbd_assert(rbd_dev);
0eefd470 2514
ebda6408
AE
2515 pages = obj_request->copyup_pages;
2516 rbd_assert(pages != NULL);
0eefd470 2517 obj_request->copyup_pages = NULL;
ebda6408
AE
2518 page_count = obj_request->copyup_page_count;
2519 rbd_assert(page_count);
2520 obj_request->copyup_page_count = 0;
2521 ceph_release_page_vector(pages, page_count);
0eefd470
AE
2522
2523 /*
2524 * We want the transfer count to reflect the size of the
2525 * original write request. There is no such thing as a
2526 * successful short write, so if the request was successful
2527 * we can just set it to the originally-requested length.
2528 */
2529 if (!obj_request->result)
2530 obj_request->xferred = obj_request->length;
2531
2761713d 2532 obj_request_done_set(obj_request);
0eefd470
AE
2533}
2534
3d7efd18
AE
2535static void
2536rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request)
2537{
2538 struct rbd_obj_request *orig_request;
0eefd470 2539 struct ceph_osd_request *osd_req;
0eefd470 2540 struct rbd_device *rbd_dev;
3d7efd18 2541 struct page **pages;
d3246fb0 2542 enum obj_operation_type op_type;
ebda6408 2543 u32 page_count;
bbea1c1a 2544 int img_result;
ebda6408 2545 u64 parent_length;
3d7efd18
AE
2546
2547 rbd_assert(img_request_child_test(img_request));
2548
2549 /* First get what we need from the image request */
2550
2551 pages = img_request->copyup_pages;
2552 rbd_assert(pages != NULL);
2553 img_request->copyup_pages = NULL;
ebda6408
AE
2554 page_count = img_request->copyup_page_count;
2555 rbd_assert(page_count);
2556 img_request->copyup_page_count = 0;
3d7efd18
AE
2557
2558 orig_request = img_request->obj_request;
2559 rbd_assert(orig_request != NULL);
b91f09f1 2560 rbd_assert(obj_request_type_valid(orig_request->type));
bbea1c1a 2561 img_result = img_request->result;
ebda6408 2562 parent_length = img_request->length;
fa355112 2563 rbd_assert(img_result || parent_length == img_request->xferred);
91c6febb 2564 rbd_img_request_put(img_request);
3d7efd18 2565
91c6febb
AE
2566 rbd_assert(orig_request->img_request);
2567 rbd_dev = orig_request->img_request->rbd_dev;
0eefd470 2568 rbd_assert(rbd_dev);
0eefd470 2569
bbea1c1a
AE
2570 /*
2571 * If the overlap has become 0 (most likely because the
2572 * image has been flattened) we need to free the pages
2573 * and re-submit the original write request.
2574 */
2575 if (!rbd_dev->parent_overlap) {
bbea1c1a 2576 ceph_release_page_vector(pages, page_count);
980917fc
ID
2577 rbd_obj_request_submit(orig_request);
2578 return;
bbea1c1a 2579 }
0eefd470 2580
bbea1c1a 2581 if (img_result)
0eefd470 2582 goto out_err;
0eefd470 2583
8785b1d4
AE
2584 /*
2585 * The original osd request is of no use to use any more.
0ccd5926 2586 * We need a new one that can hold the three ops in a copyup
8785b1d4
AE
2587 * request. Allocate the new copyup osd request for the
2588 * original request, and release the old one.
2589 */
bbea1c1a 2590 img_result = -ENOMEM;
0eefd470
AE
2591 osd_req = rbd_osd_req_create_copyup(orig_request);
2592 if (!osd_req)
2593 goto out_err;
8785b1d4 2594 rbd_osd_req_destroy(orig_request->osd_req);
0eefd470
AE
2595 orig_request->osd_req = osd_req;
2596 orig_request->copyup_pages = pages;
ebda6408 2597 orig_request->copyup_page_count = page_count;
3d7efd18 2598
0eefd470 2599 /* Initialize the copyup op */
3d7efd18 2600
0eefd470 2601 osd_req_op_cls_init(osd_req, 0, CEPH_OSD_OP_CALL, "rbd", "copyup");
ebda6408 2602 osd_req_op_cls_request_data_pages(osd_req, 0, pages, parent_length, 0,
0eefd470 2603 false, false);
3d7efd18 2604
d3246fb0 2605 /* Add the other op(s) */
0eefd470 2606
d3246fb0
JD
2607 op_type = rbd_img_request_op_type(orig_request->img_request);
2608 rbd_img_obj_request_fill(orig_request, osd_req, op_type, 1);
0eefd470
AE
2609
2610 /* All set, send it off. */
2611
980917fc
ID
2612 rbd_obj_request_submit(orig_request);
2613 return;
0eefd470 2614
0eefd470 2615out_err:
fa355112 2616 ceph_release_page_vector(pages, page_count);
0dcc685e 2617 rbd_obj_request_error(orig_request, img_result);
3d7efd18
AE
2618}
2619
2620/*
2621 * Read from the parent image the range of data that covers the
2622 * entire target of the given object request. This is used for
2623 * satisfying a layered image write request when the target of an
2624 * object request from the image request does not exist.
2625 *
2626 * A page array big enough to hold the returned data is allocated
2627 * and supplied to rbd_img_request_fill() as the "data descriptor."
2628 * When the read completes, this page array will be transferred to
2629 * the original object request for the copyup operation.
2630 *
c2e82414
ID
2631 * If an error occurs, it is recorded as the result of the original
2632 * object request in rbd_img_obj_exists_callback().
3d7efd18
AE
2633 */
2634static int rbd_img_obj_parent_read_full(struct rbd_obj_request *obj_request)
2635{
058aa991 2636 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
3d7efd18 2637 struct rbd_img_request *parent_request = NULL;
3d7efd18
AE
2638 u64 img_offset;
2639 u64 length;
2640 struct page **pages = NULL;
2641 u32 page_count;
2642 int result;
2643
3d7efd18
AE
2644 rbd_assert(rbd_dev->parent != NULL);
2645
2646 /*
2647 * Determine the byte range covered by the object in the
2648 * child image to which the original request was to be sent.
2649 */
2650 img_offset = obj_request->img_offset - obj_request->offset;
5bc3fb17 2651 length = rbd_obj_bytes(&rbd_dev->header);
3d7efd18 2652
a9e8ba2c
AE
2653 /*
2654 * There is no defined parent data beyond the parent
2655 * overlap, so limit what we read at that boundary if
2656 * necessary.
2657 */
2658 if (img_offset + length > rbd_dev->parent_overlap) {
2659 rbd_assert(img_offset < rbd_dev->parent_overlap);
2660 length = rbd_dev->parent_overlap - img_offset;
2661 }
2662
3d7efd18
AE
2663 /*
2664 * Allocate a page array big enough to receive the data read
2665 * from the parent.
2666 */
2667 page_count = (u32)calc_pages_for(0, length);
1e37f2f8 2668 pages = ceph_alloc_page_vector(page_count, GFP_NOIO);
3d7efd18
AE
2669 if (IS_ERR(pages)) {
2670 result = PTR_ERR(pages);
2671 pages = NULL;
2672 goto out_err;
2673 }
2674
2675 result = -ENOMEM;
e93f3152
AE
2676 parent_request = rbd_parent_request_create(obj_request,
2677 img_offset, length);
3d7efd18
AE
2678 if (!parent_request)
2679 goto out_err;
3d7efd18
AE
2680
2681 result = rbd_img_request_fill(parent_request, OBJ_REQUEST_PAGES, pages);
2682 if (result)
2683 goto out_err;
058aa991 2684
3d7efd18 2685 parent_request->copyup_pages = pages;
ebda6408 2686 parent_request->copyup_page_count = page_count;
3d7efd18 2687 parent_request->callback = rbd_img_obj_parent_read_full_callback;
058aa991 2688
3d7efd18
AE
2689 result = rbd_img_request_submit(parent_request);
2690 if (!result)
2691 return 0;
2692
2693 parent_request->copyup_pages = NULL;
ebda6408 2694 parent_request->copyup_page_count = 0;
3d7efd18
AE
2695 parent_request->obj_request = NULL;
2696 rbd_obj_request_put(obj_request);
2697out_err:
2698 if (pages)
2699 ceph_release_page_vector(pages, page_count);
2700 if (parent_request)
2701 rbd_img_request_put(parent_request);
3d7efd18
AE
2702 return result;
2703}
2704
c5b5ef6c
AE
2705static void rbd_img_obj_exists_callback(struct rbd_obj_request *obj_request)
2706{
c5b5ef6c 2707 struct rbd_obj_request *orig_request;
638f5abe 2708 struct rbd_device *rbd_dev;
c5b5ef6c
AE
2709 int result;
2710
2711 rbd_assert(!obj_request_img_data_test(obj_request));
2712
2713 /*
2714 * All we need from the object request is the original
2715 * request and the result of the STAT op. Grab those, then
2716 * we're done with the request.
2717 */
2718 orig_request = obj_request->obj_request;
2719 obj_request->obj_request = NULL;
912c317d 2720 rbd_obj_request_put(orig_request);
c5b5ef6c
AE
2721 rbd_assert(orig_request);
2722 rbd_assert(orig_request->img_request);
2723
2724 result = obj_request->result;
2725 obj_request->result = 0;
2726
2727 dout("%s: obj %p for obj %p result %d %llu/%llu\n", __func__,
2728 obj_request, orig_request, result,
2729 obj_request->xferred, obj_request->length);
2730 rbd_obj_request_put(obj_request);
2731
638f5abe
AE
2732 /*
2733 * If the overlap has become 0 (most likely because the
980917fc
ID
2734 * image has been flattened) we need to re-submit the
2735 * original request.
638f5abe
AE
2736 */
2737 rbd_dev = orig_request->img_request->rbd_dev;
2738 if (!rbd_dev->parent_overlap) {
980917fc
ID
2739 rbd_obj_request_submit(orig_request);
2740 return;
638f5abe 2741 }
c5b5ef6c
AE
2742
2743 /*
2744 * Our only purpose here is to determine whether the object
2745 * exists, and we don't want to treat the non-existence as
2746 * an error. If something else comes back, transfer the
2747 * error to the original request and complete it now.
2748 */
2749 if (!result) {
2750 obj_request_existence_set(orig_request, true);
2751 } else if (result == -ENOENT) {
2752 obj_request_existence_set(orig_request, false);
c2e82414
ID
2753 } else {
2754 goto fail_orig_request;
c5b5ef6c
AE
2755 }
2756
2757 /*
2758 * Resubmit the original request now that we have recorded
2759 * whether the target object exists.
2760 */
c2e82414
ID
2761 result = rbd_img_obj_request_submit(orig_request);
2762 if (result)
2763 goto fail_orig_request;
2764
2765 return;
2766
2767fail_orig_request:
0dcc685e 2768 rbd_obj_request_error(orig_request, result);
c5b5ef6c
AE
2769}
2770
2771static int rbd_img_obj_exists_submit(struct rbd_obj_request *obj_request)
2772{
058aa991 2773 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
c5b5ef6c 2774 struct rbd_obj_request *stat_request;
710214e3 2775 struct page **pages;
c5b5ef6c
AE
2776 u32 page_count;
2777 size_t size;
2778 int ret;
2779
6c696d85 2780 stat_request = rbd_obj_request_create(OBJ_REQUEST_PAGES);
710214e3
ID
2781 if (!stat_request)
2782 return -ENOMEM;
2783
a90bb0c1
ID
2784 stat_request->object_no = obj_request->object_no;
2785
710214e3
ID
2786 stat_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1,
2787 stat_request);
2788 if (!stat_request->osd_req) {
2789 ret = -ENOMEM;
2790 goto fail_stat_request;
2791 }
2792
c5b5ef6c
AE
2793 /*
2794 * The response data for a STAT call consists of:
2795 * le64 length;
2796 * struct {
2797 * le32 tv_sec;
2798 * le32 tv_nsec;
2799 * } mtime;
2800 */
2801 size = sizeof (__le64) + sizeof (__le32) + sizeof (__le32);
2802 page_count = (u32)calc_pages_for(0, size);
1e37f2f8 2803 pages = ceph_alloc_page_vector(page_count, GFP_NOIO);
710214e3
ID
2804 if (IS_ERR(pages)) {
2805 ret = PTR_ERR(pages);
2806 goto fail_stat_request;
2807 }
c5b5ef6c 2808
710214e3
ID
2809 osd_req_op_init(stat_request->osd_req, 0, CEPH_OSD_OP_STAT, 0);
2810 osd_req_op_raw_data_in_pages(stat_request->osd_req, 0, pages, size, 0,
2811 false, false);
c5b5ef6c
AE
2812
2813 rbd_obj_request_get(obj_request);
2814 stat_request->obj_request = obj_request;
2815 stat_request->pages = pages;
2816 stat_request->page_count = page_count;
c5b5ef6c
AE
2817 stat_request->callback = rbd_img_obj_exists_callback;
2818
980917fc
ID
2819 rbd_obj_request_submit(stat_request);
2820 return 0;
c5b5ef6c 2821
710214e3
ID
2822fail_stat_request:
2823 rbd_obj_request_put(stat_request);
c5b5ef6c
AE
2824 return ret;
2825}
2826
70d045f6 2827static bool img_obj_request_simple(struct rbd_obj_request *obj_request)
b454e36d 2828{
058aa991
ID
2829 struct rbd_img_request *img_request = obj_request->img_request;
2830 struct rbd_device *rbd_dev = img_request->rbd_dev;
b454e36d 2831
70d045f6 2832 /* Reads */
1c220881
JD
2833 if (!img_request_write_test(img_request) &&
2834 !img_request_discard_test(img_request))
70d045f6
ID
2835 return true;
2836
2837 /* Non-layered writes */
2838 if (!img_request_layered_test(img_request))
2839 return true;
2840
b454e36d 2841 /*
70d045f6
ID
2842 * Layered writes outside of the parent overlap range don't
2843 * share any data with the parent.
b454e36d 2844 */
70d045f6
ID
2845 if (!obj_request_overlaps_parent(obj_request))
2846 return true;
b454e36d 2847
c622d226
GZ
2848 /*
2849 * Entire-object layered writes - we will overwrite whatever
2850 * parent data there is anyway.
2851 */
2852 if (!obj_request->offset &&
2853 obj_request->length == rbd_obj_bytes(&rbd_dev->header))
2854 return true;
2855
70d045f6
ID
2856 /*
2857 * If the object is known to already exist, its parent data has
2858 * already been copied.
2859 */
2860 if (obj_request_known_test(obj_request) &&
2861 obj_request_exists_test(obj_request))
2862 return true;
2863
2864 return false;
2865}
2866
2867static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request)
2868{
058aa991
ID
2869 rbd_assert(obj_request_img_data_test(obj_request));
2870 rbd_assert(obj_request_type_valid(obj_request->type));
2871 rbd_assert(obj_request->img_request);
b454e36d 2872
70d045f6 2873 if (img_obj_request_simple(obj_request)) {
980917fc
ID
2874 rbd_obj_request_submit(obj_request);
2875 return 0;
b454e36d
AE
2876 }
2877
2878 /*
3d7efd18
AE
2879 * It's a layered write. The target object might exist but
2880 * we may not know that yet. If we know it doesn't exist,
2881 * start by reading the data for the full target object from
2882 * the parent so we can use it for a copyup to the target.
b454e36d 2883 */
70d045f6 2884 if (obj_request_known_test(obj_request))
3d7efd18
AE
2885 return rbd_img_obj_parent_read_full(obj_request);
2886
2887 /* We don't know whether the target exists. Go find out. */
b454e36d
AE
2888
2889 return rbd_img_obj_exists_submit(obj_request);
2890}
2891
bf0d5f50
AE
2892static int rbd_img_request_submit(struct rbd_img_request *img_request)
2893{
bf0d5f50 2894 struct rbd_obj_request *obj_request;
46faeed4 2895 struct rbd_obj_request *next_obj_request;
663ae2cc 2896 int ret = 0;
bf0d5f50 2897
37206ee5 2898 dout("%s: img %p\n", __func__, img_request);
bf0d5f50 2899
663ae2cc
ID
2900 rbd_img_request_get(img_request);
2901 for_each_obj_request_safe(img_request, obj_request, next_obj_request) {
b454e36d 2902 ret = rbd_img_obj_request_submit(obj_request);
bf0d5f50 2903 if (ret)
663ae2cc 2904 goto out_put_ireq;
bf0d5f50
AE
2905 }
2906
663ae2cc
ID
2907out_put_ireq:
2908 rbd_img_request_put(img_request);
2909 return ret;
bf0d5f50 2910}
8b3e1a56
AE
2911
2912static void rbd_img_parent_read_callback(struct rbd_img_request *img_request)
2913{
2914 struct rbd_obj_request *obj_request;
a9e8ba2c
AE
2915 struct rbd_device *rbd_dev;
2916 u64 obj_end;
02c74fba
AE
2917 u64 img_xferred;
2918 int img_result;
8b3e1a56
AE
2919
2920 rbd_assert(img_request_child_test(img_request));
2921
02c74fba
AE
2922 /* First get what we need from the image request and release it */
2923
8b3e1a56 2924 obj_request = img_request->obj_request;
02c74fba
AE
2925 img_xferred = img_request->xferred;
2926 img_result = img_request->result;
2927 rbd_img_request_put(img_request);
2928
2929 /*
2930 * If the overlap has become 0 (most likely because the
2931 * image has been flattened) we need to re-submit the
2932 * original request.
2933 */
a9e8ba2c
AE
2934 rbd_assert(obj_request);
2935 rbd_assert(obj_request->img_request);
02c74fba
AE
2936 rbd_dev = obj_request->img_request->rbd_dev;
2937 if (!rbd_dev->parent_overlap) {
980917fc
ID
2938 rbd_obj_request_submit(obj_request);
2939 return;
02c74fba 2940 }
a9e8ba2c 2941
02c74fba 2942 obj_request->result = img_result;
a9e8ba2c
AE
2943 if (obj_request->result)
2944 goto out;
2945
2946 /*
2947 * We need to zero anything beyond the parent overlap
2948 * boundary. Since rbd_img_obj_request_read_callback()
2949 * will zero anything beyond the end of a short read, an
2950 * easy way to do this is to pretend the data from the
2951 * parent came up short--ending at the overlap boundary.
2952 */
2953 rbd_assert(obj_request->img_offset < U64_MAX - obj_request->length);
2954 obj_end = obj_request->img_offset + obj_request->length;
a9e8ba2c
AE
2955 if (obj_end > rbd_dev->parent_overlap) {
2956 u64 xferred = 0;
2957
2958 if (obj_request->img_offset < rbd_dev->parent_overlap)
2959 xferred = rbd_dev->parent_overlap -
2960 obj_request->img_offset;
8b3e1a56 2961
02c74fba 2962 obj_request->xferred = min(img_xferred, xferred);
a9e8ba2c 2963 } else {
02c74fba 2964 obj_request->xferred = img_xferred;
a9e8ba2c
AE
2965 }
2966out:
8b3e1a56
AE
2967 rbd_img_obj_request_read_callback(obj_request);
2968 rbd_obj_request_complete(obj_request);
2969}
2970
2971static void rbd_img_parent_read(struct rbd_obj_request *obj_request)
2972{
8b3e1a56
AE
2973 struct rbd_img_request *img_request;
2974 int result;
2975
2976 rbd_assert(obj_request_img_data_test(obj_request));
2977 rbd_assert(obj_request->img_request != NULL);
2978 rbd_assert(obj_request->result == (s32) -ENOENT);
5b2ab72d 2979 rbd_assert(obj_request_type_valid(obj_request->type));
8b3e1a56 2980
8b3e1a56 2981 /* rbd_read_finish(obj_request, obj_request->length); */
e93f3152 2982 img_request = rbd_parent_request_create(obj_request,
8b3e1a56 2983 obj_request->img_offset,
e93f3152 2984 obj_request->length);
8b3e1a56
AE
2985 result = -ENOMEM;
2986 if (!img_request)
2987 goto out_err;
2988
5b2ab72d
AE
2989 if (obj_request->type == OBJ_REQUEST_BIO)
2990 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
2991 obj_request->bio_list);
2992 else
2993 result = rbd_img_request_fill(img_request, OBJ_REQUEST_PAGES,
2994 obj_request->pages);
8b3e1a56
AE
2995 if (result)
2996 goto out_err;
2997
2998 img_request->callback = rbd_img_parent_read_callback;
2999 result = rbd_img_request_submit(img_request);
3000 if (result)
3001 goto out_err;
3002
3003 return;
3004out_err:
3005 if (img_request)
3006 rbd_img_request_put(img_request);
3007 obj_request->result = result;
3008 obj_request->xferred = 0;
3009 obj_request_done_set(obj_request);
3010}
bf0d5f50 3011
ed95b21a 3012static const struct rbd_client_id rbd_empty_cid;
b8d70035 3013
ed95b21a
ID
3014static bool rbd_cid_equal(const struct rbd_client_id *lhs,
3015 const struct rbd_client_id *rhs)
3016{
3017 return lhs->gid == rhs->gid && lhs->handle == rhs->handle;
3018}
3019
3020static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev)
3021{
3022 struct rbd_client_id cid;
3023
3024 mutex_lock(&rbd_dev->watch_mutex);
3025 cid.gid = ceph_client_gid(rbd_dev->rbd_client->client);
3026 cid.handle = rbd_dev->watch_cookie;
3027 mutex_unlock(&rbd_dev->watch_mutex);
3028 return cid;
3029}
3030
3031/*
3032 * lock_rwsem must be held for write
3033 */
3034static void rbd_set_owner_cid(struct rbd_device *rbd_dev,
3035 const struct rbd_client_id *cid)
3036{
3037 dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev,
3038 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle,
3039 cid->gid, cid->handle);
3040 rbd_dev->owner_cid = *cid; /* struct */
3041}
3042
3043static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf)
3044{
3045 mutex_lock(&rbd_dev->watch_mutex);
3046 sprintf(buf, "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie);
3047 mutex_unlock(&rbd_dev->watch_mutex);
3048}
3049
3050/*
3051 * lock_rwsem must be held for write
3052 */
3053static int rbd_lock(struct rbd_device *rbd_dev)
b8d70035 3054{
922dab61 3055 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
ed95b21a
ID
3056 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3057 char cookie[32];
e627db08 3058 int ret;
b8d70035 3059
cbbfb0ff
ID
3060 WARN_ON(__rbd_is_lock_owner(rbd_dev) ||
3061 rbd_dev->lock_cookie[0] != '\0');
52bb1f9b 3062
ed95b21a
ID
3063 format_lock_cookie(rbd_dev, cookie);
3064 ret = ceph_cls_lock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3065 RBD_LOCK_NAME, CEPH_CLS_LOCK_EXCLUSIVE, cookie,
3066 RBD_LOCK_TAG, "", 0);
e627db08 3067 if (ret)
ed95b21a 3068 return ret;
b8d70035 3069
ed95b21a 3070 rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED;
cbbfb0ff 3071 strcpy(rbd_dev->lock_cookie, cookie);
ed95b21a
ID
3072 rbd_set_owner_cid(rbd_dev, &cid);
3073 queue_work(rbd_dev->task_wq, &rbd_dev->acquired_lock_work);
3074 return 0;
b8d70035
AE
3075}
3076
ed95b21a
ID
3077/*
3078 * lock_rwsem must be held for write
3079 */
bbead745 3080static void rbd_unlock(struct rbd_device *rbd_dev)
bb040aa0 3081{
922dab61 3082 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
bb040aa0
ID
3083 int ret;
3084
cbbfb0ff
ID
3085 WARN_ON(!__rbd_is_lock_owner(rbd_dev) ||
3086 rbd_dev->lock_cookie[0] == '\0');
bb040aa0 3087
ed95b21a 3088 ret = ceph_cls_unlock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
cbbfb0ff 3089 RBD_LOCK_NAME, rbd_dev->lock_cookie);
bbead745
ID
3090 if (ret && ret != -ENOENT)
3091 rbd_warn(rbd_dev, "failed to unlock: %d", ret);
bb040aa0 3092
bbead745
ID
3093 /* treat errors as the image is unlocked */
3094 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
cbbfb0ff 3095 rbd_dev->lock_cookie[0] = '\0';
ed95b21a
ID
3096 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3097 queue_work(rbd_dev->task_wq, &rbd_dev->released_lock_work);
bb040aa0
ID
3098}
3099
ed95b21a
ID
3100static int __rbd_notify_op_lock(struct rbd_device *rbd_dev,
3101 enum rbd_notify_op notify_op,
3102 struct page ***preply_pages,
3103 size_t *preply_len)
9969ebc5
AE
3104{
3105 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
ed95b21a
ID
3106 struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3107 int buf_size = 4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN;
3108 char buf[buf_size];
3109 void *p = buf;
9969ebc5 3110
ed95b21a 3111 dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op);
9969ebc5 3112
ed95b21a
ID
3113 /* encode *LockPayload NotifyMessage (op + ClientId) */
3114 ceph_start_encoding(&p, 2, 1, buf_size - CEPH_ENCODING_START_BLK_LEN);
3115 ceph_encode_32(&p, notify_op);
3116 ceph_encode_64(&p, cid.gid);
3117 ceph_encode_64(&p, cid.handle);
8eb87565 3118
ed95b21a
ID
3119 return ceph_osdc_notify(osdc, &rbd_dev->header_oid,
3120 &rbd_dev->header_oloc, buf, buf_size,
3121 RBD_NOTIFY_TIMEOUT, preply_pages, preply_len);
b30a01f2
ID
3122}
3123
ed95b21a
ID
3124static void rbd_notify_op_lock(struct rbd_device *rbd_dev,
3125 enum rbd_notify_op notify_op)
b30a01f2 3126{
ed95b21a
ID
3127 struct page **reply_pages;
3128 size_t reply_len;
b30a01f2 3129
ed95b21a
ID
3130 __rbd_notify_op_lock(rbd_dev, notify_op, &reply_pages, &reply_len);
3131 ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3132}
b30a01f2 3133
ed95b21a
ID
3134static void rbd_notify_acquired_lock(struct work_struct *work)
3135{
3136 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3137 acquired_lock_work);
76756a51 3138
ed95b21a 3139 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_ACQUIRED_LOCK);
c525f036
ID
3140}
3141
ed95b21a 3142static void rbd_notify_released_lock(struct work_struct *work)
c525f036 3143{
ed95b21a
ID
3144 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3145 released_lock_work);
811c6688 3146
ed95b21a 3147 rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_RELEASED_LOCK);
fca27065
ID
3148}
3149
ed95b21a 3150static int rbd_request_lock(struct rbd_device *rbd_dev)
36be9a76 3151{
ed95b21a
ID
3152 struct page **reply_pages;
3153 size_t reply_len;
3154 bool lock_owner_responded = false;
36be9a76
AE
3155 int ret;
3156
ed95b21a 3157 dout("%s rbd_dev %p\n", __func__, rbd_dev);
36be9a76 3158
ed95b21a
ID
3159 ret = __rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_REQUEST_LOCK,
3160 &reply_pages, &reply_len);
3161 if (ret && ret != -ETIMEDOUT) {
3162 rbd_warn(rbd_dev, "failed to request lock: %d", ret);
36be9a76 3163 goto out;
ed95b21a 3164 }
36be9a76 3165
ed95b21a
ID
3166 if (reply_len > 0 && reply_len <= PAGE_SIZE) {
3167 void *p = page_address(reply_pages[0]);
3168 void *const end = p + reply_len;
3169 u32 n;
36be9a76 3170
ed95b21a
ID
3171 ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */
3172 while (n--) {
3173 u8 struct_v;
3174 u32 len;
36be9a76 3175
ed95b21a
ID
3176 ceph_decode_need(&p, end, 8 + 8, e_inval);
3177 p += 8 + 8; /* skip gid and cookie */
04017e29 3178
ed95b21a
ID
3179 ceph_decode_32_safe(&p, end, len, e_inval);
3180 if (!len)
3181 continue;
3182
3183 if (lock_owner_responded) {
3184 rbd_warn(rbd_dev,
3185 "duplicate lock owners detected");
3186 ret = -EIO;
3187 goto out;
3188 }
3189
3190 lock_owner_responded = true;
3191 ret = ceph_start_decoding(&p, end, 1, "ResponseMessage",
3192 &struct_v, &len);
3193 if (ret) {
3194 rbd_warn(rbd_dev,
3195 "failed to decode ResponseMessage: %d",
3196 ret);
3197 goto e_inval;
3198 }
3199
3200 ret = ceph_decode_32(&p);
3201 }
3202 }
3203
3204 if (!lock_owner_responded) {
3205 rbd_warn(rbd_dev, "no lock owners detected");
3206 ret = -ETIMEDOUT;
3207 }
3208
3209out:
3210 ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3211 return ret;
3212
3213e_inval:
3214 ret = -EINVAL;
3215 goto out;
3216}
3217
3218static void wake_requests(struct rbd_device *rbd_dev, bool wake_all)
3219{
3220 dout("%s rbd_dev %p wake_all %d\n", __func__, rbd_dev, wake_all);
3221
3222 cancel_delayed_work(&rbd_dev->lock_dwork);
3223 if (wake_all)
3224 wake_up_all(&rbd_dev->lock_waitq);
3225 else
3226 wake_up(&rbd_dev->lock_waitq);
3227}
3228
3229static int get_lock_owner_info(struct rbd_device *rbd_dev,
3230 struct ceph_locker **lockers, u32 *num_lockers)
3231{
3232 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3233 u8 lock_type;
3234 char *lock_tag;
3235 int ret;
3236
3237 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3238
3239 ret = ceph_cls_lock_info(osdc, &rbd_dev->header_oid,
3240 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3241 &lock_type, &lock_tag, lockers, num_lockers);
3242 if (ret)
3243 return ret;
3244
3245 if (*num_lockers == 0) {
3246 dout("%s rbd_dev %p no lockers detected\n", __func__, rbd_dev);
3247 goto out;
3248 }
3249
3250 if (strcmp(lock_tag, RBD_LOCK_TAG)) {
3251 rbd_warn(rbd_dev, "locked by external mechanism, tag %s",
3252 lock_tag);
3253 ret = -EBUSY;
3254 goto out;
3255 }
3256
3257 if (lock_type == CEPH_CLS_LOCK_SHARED) {
3258 rbd_warn(rbd_dev, "shared lock type detected");
3259 ret = -EBUSY;
3260 goto out;
3261 }
3262
3263 if (strncmp((*lockers)[0].id.cookie, RBD_LOCK_COOKIE_PREFIX,
3264 strlen(RBD_LOCK_COOKIE_PREFIX))) {
3265 rbd_warn(rbd_dev, "locked by external mechanism, cookie %s",
3266 (*lockers)[0].id.cookie);
3267 ret = -EBUSY;
3268 goto out;
3269 }
3270
3271out:
3272 kfree(lock_tag);
3273 return ret;
3274}
3275
3276static int find_watcher(struct rbd_device *rbd_dev,
3277 const struct ceph_locker *locker)
3278{
3279 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3280 struct ceph_watch_item *watchers;
3281 u32 num_watchers;
3282 u64 cookie;
3283 int i;
3284 int ret;
3285
3286 ret = ceph_osdc_list_watchers(osdc, &rbd_dev->header_oid,
3287 &rbd_dev->header_oloc, &watchers,
3288 &num_watchers);
3289 if (ret)
3290 return ret;
3291
3292 sscanf(locker->id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu", &cookie);
3293 for (i = 0; i < num_watchers; i++) {
3294 if (!memcmp(&watchers[i].addr, &locker->info.addr,
3295 sizeof(locker->info.addr)) &&
3296 watchers[i].cookie == cookie) {
3297 struct rbd_client_id cid = {
3298 .gid = le64_to_cpu(watchers[i].name.num),
3299 .handle = cookie,
3300 };
3301
3302 dout("%s rbd_dev %p found cid %llu-%llu\n", __func__,
3303 rbd_dev, cid.gid, cid.handle);
3304 rbd_set_owner_cid(rbd_dev, &cid);
3305 ret = 1;
3306 goto out;
3307 }
3308 }
3309
3310 dout("%s rbd_dev %p no watchers\n", __func__, rbd_dev);
3311 ret = 0;
3312out:
3313 kfree(watchers);
3314 return ret;
3315}
3316
3317/*
3318 * lock_rwsem must be held for write
3319 */
3320static int rbd_try_lock(struct rbd_device *rbd_dev)
3321{
3322 struct ceph_client *client = rbd_dev->rbd_client->client;
3323 struct ceph_locker *lockers;
3324 u32 num_lockers;
3325 int ret;
3326
3327 for (;;) {
3328 ret = rbd_lock(rbd_dev);
3329 if (ret != -EBUSY)
3330 return ret;
3331
3332 /* determine if the current lock holder is still alive */
3333 ret = get_lock_owner_info(rbd_dev, &lockers, &num_lockers);
3334 if (ret)
3335 return ret;
3336
3337 if (num_lockers == 0)
3338 goto again;
3339
3340 ret = find_watcher(rbd_dev, lockers);
3341 if (ret) {
3342 if (ret > 0)
3343 ret = 0; /* have to request lock */
3344 goto out;
3345 }
3346
3347 rbd_warn(rbd_dev, "%s%llu seems dead, breaking lock",
3348 ENTITY_NAME(lockers[0].id.name));
3349
3350 ret = ceph_monc_blacklist_add(&client->monc,
3351 &lockers[0].info.addr);
3352 if (ret) {
3353 rbd_warn(rbd_dev, "blacklist of %s%llu failed: %d",
3354 ENTITY_NAME(lockers[0].id.name), ret);
3355 goto out;
3356 }
3357
3358 ret = ceph_cls_break_lock(&client->osdc, &rbd_dev->header_oid,
3359 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3360 lockers[0].id.cookie,
3361 &lockers[0].id.name);
3362 if (ret && ret != -ENOENT)
3363 goto out;
3364
3365again:
3366 ceph_free_lockers(lockers, num_lockers);
3367 }
3368
3369out:
3370 ceph_free_lockers(lockers, num_lockers);
3371 return ret;
3372}
3373
3374/*
3375 * ret is set only if lock_state is RBD_LOCK_STATE_UNLOCKED
3376 */
3377static enum rbd_lock_state rbd_try_acquire_lock(struct rbd_device *rbd_dev,
3378 int *pret)
3379{
3380 enum rbd_lock_state lock_state;
3381
3382 down_read(&rbd_dev->lock_rwsem);
3383 dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
3384 rbd_dev->lock_state);
3385 if (__rbd_is_lock_owner(rbd_dev)) {
3386 lock_state = rbd_dev->lock_state;
3387 up_read(&rbd_dev->lock_rwsem);
3388 return lock_state;
3389 }
3390
3391 up_read(&rbd_dev->lock_rwsem);
3392 down_write(&rbd_dev->lock_rwsem);
3393 dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
3394 rbd_dev->lock_state);
3395 if (!__rbd_is_lock_owner(rbd_dev)) {
3396 *pret = rbd_try_lock(rbd_dev);
3397 if (*pret)
3398 rbd_warn(rbd_dev, "failed to acquire lock: %d", *pret);
3399 }
3400
3401 lock_state = rbd_dev->lock_state;
3402 up_write(&rbd_dev->lock_rwsem);
3403 return lock_state;
3404}
3405
3406static void rbd_acquire_lock(struct work_struct *work)
3407{
3408 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
3409 struct rbd_device, lock_dwork);
3410 enum rbd_lock_state lock_state;
37f13252 3411 int ret = 0;
ed95b21a
ID
3412
3413 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3414again:
3415 lock_state = rbd_try_acquire_lock(rbd_dev, &ret);
3416 if (lock_state != RBD_LOCK_STATE_UNLOCKED || ret == -EBLACKLISTED) {
3417 if (lock_state == RBD_LOCK_STATE_LOCKED)
3418 wake_requests(rbd_dev, true);
3419 dout("%s rbd_dev %p lock_state %d ret %d - done\n", __func__,
3420 rbd_dev, lock_state, ret);
3421 return;
3422 }
3423
3424 ret = rbd_request_lock(rbd_dev);
3425 if (ret == -ETIMEDOUT) {
3426 goto again; /* treat this as a dead client */
e010dd0a
ID
3427 } else if (ret == -EROFS) {
3428 rbd_warn(rbd_dev, "peer will not release lock");
3429 /*
3430 * If this is rbd_add_acquire_lock(), we want to fail
3431 * immediately -- reuse BLACKLISTED flag. Otherwise we
3432 * want to block.
3433 */
3434 if (!(rbd_dev->disk->flags & GENHD_FL_UP)) {
3435 set_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags);
3436 /* wake "rbd map --exclusive" process */
3437 wake_requests(rbd_dev, false);
3438 }
ed95b21a
ID
3439 } else if (ret < 0) {
3440 rbd_warn(rbd_dev, "error requesting lock: %d", ret);
3441 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
3442 RBD_RETRY_DELAY);
3443 } else {
3444 /*
3445 * lock owner acked, but resend if we don't see them
3446 * release the lock
3447 */
3448 dout("%s rbd_dev %p requeueing lock_dwork\n", __func__,
3449 rbd_dev);
3450 mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
3451 msecs_to_jiffies(2 * RBD_NOTIFY_TIMEOUT * MSEC_PER_SEC));
3452 }
3453}
3454
3455/*
3456 * lock_rwsem must be held for write
3457 */
3458static bool rbd_release_lock(struct rbd_device *rbd_dev)
3459{
3460 dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
3461 rbd_dev->lock_state);
3462 if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED)
3463 return false;
3464
3465 rbd_dev->lock_state = RBD_LOCK_STATE_RELEASING;
3466 downgrade_write(&rbd_dev->lock_rwsem);
52bb1f9b 3467 /*
ed95b21a 3468 * Ensure that all in-flight IO is flushed.
52bb1f9b 3469 *
ed95b21a
ID
3470 * FIXME: ceph_osdc_sync() flushes the entire OSD client, which
3471 * may be shared with other devices.
52bb1f9b 3472 */
ed95b21a
ID
3473 ceph_osdc_sync(&rbd_dev->rbd_client->client->osdc);
3474 up_read(&rbd_dev->lock_rwsem);
3475
3476 down_write(&rbd_dev->lock_rwsem);
3477 dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
3478 rbd_dev->lock_state);
3479 if (rbd_dev->lock_state != RBD_LOCK_STATE_RELEASING)
3480 return false;
3481
bbead745
ID
3482 rbd_unlock(rbd_dev);
3483 /*
3484 * Give others a chance to grab the lock - we would re-acquire
3485 * almost immediately if we got new IO during ceph_osdc_sync()
3486 * otherwise. We need to ack our own notifications, so this
3487 * lock_dwork will be requeued from rbd_wait_state_locked()
3488 * after wake_requests() in rbd_handle_released_lock().
3489 */
3490 cancel_delayed_work(&rbd_dev->lock_dwork);
ed95b21a
ID
3491 return true;
3492}
3493
3494static void rbd_release_lock_work(struct work_struct *work)
3495{
3496 struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3497 unlock_work);
3498
3499 down_write(&rbd_dev->lock_rwsem);
3500 rbd_release_lock(rbd_dev);
3501 up_write(&rbd_dev->lock_rwsem);
3502}
3503
3504static void rbd_handle_acquired_lock(struct rbd_device *rbd_dev, u8 struct_v,
3505 void **p)
3506{
3507 struct rbd_client_id cid = { 0 };
3508
3509 if (struct_v >= 2) {
3510 cid.gid = ceph_decode_64(p);
3511 cid.handle = ceph_decode_64(p);
3512 }
3513
3514 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
3515 cid.handle);
3516 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
3517 down_write(&rbd_dev->lock_rwsem);
3518 if (rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
3519 /*
3520 * we already know that the remote client is
3521 * the owner
3522 */
3523 up_write(&rbd_dev->lock_rwsem);
3524 return;
3525 }
3526
3527 rbd_set_owner_cid(rbd_dev, &cid);
3528 downgrade_write(&rbd_dev->lock_rwsem);
3529 } else {
3530 down_read(&rbd_dev->lock_rwsem);
3531 }
3532
3533 if (!__rbd_is_lock_owner(rbd_dev))
3534 wake_requests(rbd_dev, false);
3535 up_read(&rbd_dev->lock_rwsem);
3536}
3537
3538static void rbd_handle_released_lock(struct rbd_device *rbd_dev, u8 struct_v,
3539 void **p)
3540{
3541 struct rbd_client_id cid = { 0 };
3542
3543 if (struct_v >= 2) {
3544 cid.gid = ceph_decode_64(p);
3545 cid.handle = ceph_decode_64(p);
3546 }
3547
3548 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
3549 cid.handle);
3550 if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
3551 down_write(&rbd_dev->lock_rwsem);
3552 if (!rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
3553 dout("%s rbd_dev %p unexpected owner, cid %llu-%llu != owner_cid %llu-%llu\n",
3554 __func__, rbd_dev, cid.gid, cid.handle,
3555 rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle);
3556 up_write(&rbd_dev->lock_rwsem);
3557 return;
3558 }
3559
3560 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3561 downgrade_write(&rbd_dev->lock_rwsem);
3562 } else {
3563 down_read(&rbd_dev->lock_rwsem);
3564 }
3565
3566 if (!__rbd_is_lock_owner(rbd_dev))
3567 wake_requests(rbd_dev, false);
3568 up_read(&rbd_dev->lock_rwsem);
3569}
3570
3b77faa0
ID
3571/*
3572 * Returns result for ResponseMessage to be encoded (<= 0), or 1 if no
3573 * ResponseMessage is needed.
3574 */
3575static int rbd_handle_request_lock(struct rbd_device *rbd_dev, u8 struct_v,
3576 void **p)
ed95b21a
ID
3577{
3578 struct rbd_client_id my_cid = rbd_get_cid(rbd_dev);
3579 struct rbd_client_id cid = { 0 };
3b77faa0 3580 int result = 1;
ed95b21a
ID
3581
3582 if (struct_v >= 2) {
3583 cid.gid = ceph_decode_64(p);
3584 cid.handle = ceph_decode_64(p);
3585 }
3586
3587 dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
3588 cid.handle);
3589 if (rbd_cid_equal(&cid, &my_cid))
3b77faa0 3590 return result;
ed95b21a
ID
3591
3592 down_read(&rbd_dev->lock_rwsem);
3b77faa0
ID
3593 if (__rbd_is_lock_owner(rbd_dev)) {
3594 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED &&
3595 rbd_cid_equal(&rbd_dev->owner_cid, &rbd_empty_cid))
3596 goto out_unlock;
3597
3598 /*
3599 * encode ResponseMessage(0) so the peer can detect
3600 * a missing owner
3601 */
3602 result = 0;
3603
3604 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED) {
e010dd0a
ID
3605 if (!rbd_dev->opts->exclusive) {
3606 dout("%s rbd_dev %p queueing unlock_work\n",
3607 __func__, rbd_dev);
3608 queue_work(rbd_dev->task_wq,
3609 &rbd_dev->unlock_work);
3610 } else {
3611 /* refuse to release the lock */
3612 result = -EROFS;
3613 }
ed95b21a
ID
3614 }
3615 }
3b77faa0
ID
3616
3617out_unlock:
ed95b21a 3618 up_read(&rbd_dev->lock_rwsem);
3b77faa0 3619 return result;
ed95b21a
ID
3620}
3621
3622static void __rbd_acknowledge_notify(struct rbd_device *rbd_dev,
3623 u64 notify_id, u64 cookie, s32 *result)
3624{
3625 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3626 int buf_size = 4 + CEPH_ENCODING_START_BLK_LEN;
3627 char buf[buf_size];
3628 int ret;
3629
3630 if (result) {
3631 void *p = buf;
3632
3633 /* encode ResponseMessage */
3634 ceph_start_encoding(&p, 1, 1,
3635 buf_size - CEPH_ENCODING_START_BLK_LEN);
3636 ceph_encode_32(&p, *result);
3637 } else {
3638 buf_size = 0;
3639 }
b8d70035 3640
922dab61
ID
3641 ret = ceph_osdc_notify_ack(osdc, &rbd_dev->header_oid,
3642 &rbd_dev->header_oloc, notify_id, cookie,
ed95b21a 3643 buf, buf_size);
52bb1f9b 3644 if (ret)
ed95b21a
ID
3645 rbd_warn(rbd_dev, "acknowledge_notify failed: %d", ret);
3646}
3647
3648static void rbd_acknowledge_notify(struct rbd_device *rbd_dev, u64 notify_id,
3649 u64 cookie)
3650{
3651 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3652 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, NULL);
3653}
3654
3655static void rbd_acknowledge_notify_result(struct rbd_device *rbd_dev,
3656 u64 notify_id, u64 cookie, s32 result)
3657{
3658 dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
3659 __rbd_acknowledge_notify(rbd_dev, notify_id, cookie, &result);
3660}
3661
3662static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie,
3663 u64 notifier_id, void *data, size_t data_len)
3664{
3665 struct rbd_device *rbd_dev = arg;
3666 void *p = data;
3667 void *const end = p + data_len;
d4c2269b 3668 u8 struct_v = 0;
ed95b21a
ID
3669 u32 len;
3670 u32 notify_op;
3671 int ret;
3672
3673 dout("%s rbd_dev %p cookie %llu notify_id %llu data_len %zu\n",
3674 __func__, rbd_dev, cookie, notify_id, data_len);
3675 if (data_len) {
3676 ret = ceph_start_decoding(&p, end, 1, "NotifyMessage",
3677 &struct_v, &len);
3678 if (ret) {
3679 rbd_warn(rbd_dev, "failed to decode NotifyMessage: %d",
3680 ret);
3681 return;
3682 }
3683
3684 notify_op = ceph_decode_32(&p);
3685 } else {
3686 /* legacy notification for header updates */
3687 notify_op = RBD_NOTIFY_OP_HEADER_UPDATE;
3688 len = 0;
3689 }
3690
3691 dout("%s rbd_dev %p notify_op %u\n", __func__, rbd_dev, notify_op);
3692 switch (notify_op) {
3693 case RBD_NOTIFY_OP_ACQUIRED_LOCK:
3694 rbd_handle_acquired_lock(rbd_dev, struct_v, &p);
3695 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
3696 break;
3697 case RBD_NOTIFY_OP_RELEASED_LOCK:
3698 rbd_handle_released_lock(rbd_dev, struct_v, &p);
3699 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
3700 break;
3701 case RBD_NOTIFY_OP_REQUEST_LOCK:
3b77faa0
ID
3702 ret = rbd_handle_request_lock(rbd_dev, struct_v, &p);
3703 if (ret <= 0)
ed95b21a 3704 rbd_acknowledge_notify_result(rbd_dev, notify_id,
3b77faa0 3705 cookie, ret);
ed95b21a
ID
3706 else
3707 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
3708 break;
3709 case RBD_NOTIFY_OP_HEADER_UPDATE:
3710 ret = rbd_dev_refresh(rbd_dev);
3711 if (ret)
3712 rbd_warn(rbd_dev, "refresh failed: %d", ret);
3713
3714 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
3715 break;
3716 default:
3717 if (rbd_is_lock_owner(rbd_dev))
3718 rbd_acknowledge_notify_result(rbd_dev, notify_id,
3719 cookie, -EOPNOTSUPP);
3720 else
3721 rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
3722 break;
3723 }
b8d70035
AE
3724}
3725
99d16943
ID
3726static void __rbd_unregister_watch(struct rbd_device *rbd_dev);
3727
922dab61 3728static void rbd_watch_errcb(void *arg, u64 cookie, int err)
bb040aa0 3729{
922dab61 3730 struct rbd_device *rbd_dev = arg;
bb040aa0 3731
922dab61 3732 rbd_warn(rbd_dev, "encountered watch error: %d", err);
bb040aa0 3733
ed95b21a
ID
3734 down_write(&rbd_dev->lock_rwsem);
3735 rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3736 up_write(&rbd_dev->lock_rwsem);
3737
99d16943
ID
3738 mutex_lock(&rbd_dev->watch_mutex);
3739 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED) {
3740 __rbd_unregister_watch(rbd_dev);
3741 rbd_dev->watch_state = RBD_WATCH_STATE_ERROR;
bb040aa0 3742
99d16943 3743 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->watch_dwork, 0);
bb040aa0 3744 }
99d16943 3745 mutex_unlock(&rbd_dev->watch_mutex);
bb040aa0
ID
3746}
3747
9969ebc5 3748/*
99d16943 3749 * watch_mutex must be locked
9969ebc5 3750 */
99d16943 3751static int __rbd_register_watch(struct rbd_device *rbd_dev)
9969ebc5
AE
3752{
3753 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
922dab61 3754 struct ceph_osd_linger_request *handle;
9969ebc5 3755
922dab61 3756 rbd_assert(!rbd_dev->watch_handle);
99d16943 3757 dout("%s rbd_dev %p\n", __func__, rbd_dev);
9969ebc5 3758
922dab61
ID
3759 handle = ceph_osdc_watch(osdc, &rbd_dev->header_oid,
3760 &rbd_dev->header_oloc, rbd_watch_cb,
3761 rbd_watch_errcb, rbd_dev);
3762 if (IS_ERR(handle))
3763 return PTR_ERR(handle);
8eb87565 3764
922dab61 3765 rbd_dev->watch_handle = handle;
b30a01f2 3766 return 0;
b30a01f2
ID
3767}
3768
99d16943
ID
3769/*
3770 * watch_mutex must be locked
3771 */
3772static void __rbd_unregister_watch(struct rbd_device *rbd_dev)
b30a01f2 3773{
922dab61
ID
3774 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3775 int ret;
b30a01f2 3776
99d16943
ID
3777 rbd_assert(rbd_dev->watch_handle);
3778 dout("%s rbd_dev %p\n", __func__, rbd_dev);
b30a01f2 3779
922dab61
ID
3780 ret = ceph_osdc_unwatch(osdc, rbd_dev->watch_handle);
3781 if (ret)
3782 rbd_warn(rbd_dev, "failed to unwatch: %d", ret);
76756a51 3783
922dab61 3784 rbd_dev->watch_handle = NULL;
c525f036
ID
3785}
3786
99d16943
ID
3787static int rbd_register_watch(struct rbd_device *rbd_dev)
3788{
3789 int ret;
3790
3791 mutex_lock(&rbd_dev->watch_mutex);
3792 rbd_assert(rbd_dev->watch_state == RBD_WATCH_STATE_UNREGISTERED);
3793 ret = __rbd_register_watch(rbd_dev);
3794 if (ret)
3795 goto out;
3796
3797 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
3798 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
3799
3800out:
3801 mutex_unlock(&rbd_dev->watch_mutex);
3802 return ret;
3803}
3804
3805static void cancel_tasks_sync(struct rbd_device *rbd_dev)
c525f036 3806{
99d16943
ID
3807 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3808
3809 cancel_delayed_work_sync(&rbd_dev->watch_dwork);
ed95b21a
ID
3810 cancel_work_sync(&rbd_dev->acquired_lock_work);
3811 cancel_work_sync(&rbd_dev->released_lock_work);
3812 cancel_delayed_work_sync(&rbd_dev->lock_dwork);
3813 cancel_work_sync(&rbd_dev->unlock_work);
99d16943
ID
3814}
3815
3816static void rbd_unregister_watch(struct rbd_device *rbd_dev)
3817{
ed95b21a 3818 WARN_ON(waitqueue_active(&rbd_dev->lock_waitq));
99d16943
ID
3819 cancel_tasks_sync(rbd_dev);
3820
3821 mutex_lock(&rbd_dev->watch_mutex);
3822 if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED)
3823 __rbd_unregister_watch(rbd_dev);
3824 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
3825 mutex_unlock(&rbd_dev->watch_mutex);
811c6688 3826
811c6688 3827 ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
fca27065
ID
3828}
3829
14bb211d
ID
3830/*
3831 * lock_rwsem must be held for write
3832 */
3833static void rbd_reacquire_lock(struct rbd_device *rbd_dev)
3834{
3835 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3836 char cookie[32];
3837 int ret;
3838
3839 WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED);
3840
3841 format_lock_cookie(rbd_dev, cookie);
3842 ret = ceph_cls_set_cookie(osdc, &rbd_dev->header_oid,
3843 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3844 CEPH_CLS_LOCK_EXCLUSIVE, rbd_dev->lock_cookie,
3845 RBD_LOCK_TAG, cookie);
3846 if (ret) {
3847 if (ret != -EOPNOTSUPP)
3848 rbd_warn(rbd_dev, "failed to update lock cookie: %d",
3849 ret);
3850
3851 /*
3852 * Lock cookie cannot be updated on older OSDs, so do
3853 * a manual release and queue an acquire.
3854 */
3855 if (rbd_release_lock(rbd_dev))
3856 queue_delayed_work(rbd_dev->task_wq,
3857 &rbd_dev->lock_dwork, 0);
3858 } else {
3859 strcpy(rbd_dev->lock_cookie, cookie);
3860 }
3861}
3862
99d16943
ID
3863static void rbd_reregister_watch(struct work_struct *work)
3864{
3865 struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
3866 struct rbd_device, watch_dwork);
3867 int ret;
3868
3869 dout("%s rbd_dev %p\n", __func__, rbd_dev);
3870
3871 mutex_lock(&rbd_dev->watch_mutex);
87c0fded
ID
3872 if (rbd_dev->watch_state != RBD_WATCH_STATE_ERROR) {
3873 mutex_unlock(&rbd_dev->watch_mutex);
14bb211d 3874 return;
87c0fded 3875 }
99d16943
ID
3876
3877 ret = __rbd_register_watch(rbd_dev);
3878 if (ret) {
3879 rbd_warn(rbd_dev, "failed to reregister watch: %d", ret);
4d73644b 3880 if (ret == -EBLACKLISTED || ret == -ENOENT) {
87c0fded 3881 set_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags);
14bb211d 3882 wake_requests(rbd_dev, true);
87c0fded 3883 } else {
99d16943
ID
3884 queue_delayed_work(rbd_dev->task_wq,
3885 &rbd_dev->watch_dwork,
3886 RBD_RETRY_DELAY);
87c0fded
ID
3887 }
3888 mutex_unlock(&rbd_dev->watch_mutex);
14bb211d 3889 return;
99d16943
ID
3890 }
3891
3892 rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
3893 rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
3894 mutex_unlock(&rbd_dev->watch_mutex);
3895
14bb211d
ID
3896 down_write(&rbd_dev->lock_rwsem);
3897 if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED)
3898 rbd_reacquire_lock(rbd_dev);
3899 up_write(&rbd_dev->lock_rwsem);
3900
99d16943
ID
3901 ret = rbd_dev_refresh(rbd_dev);
3902 if (ret)
3903 rbd_warn(rbd_dev, "reregisteration refresh failed: %d", ret);
99d16943
ID
3904}
3905
36be9a76 3906/*
f40eb349
AE
3907 * Synchronous osd object method call. Returns the number of bytes
3908 * returned in the outbound buffer, or a negative error code.
36be9a76
AE
3909 */
3910static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
ecd4a68a
ID
3911 struct ceph_object_id *oid,
3912 struct ceph_object_locator *oloc,
36be9a76 3913 const char *method_name,
4157976b 3914 const void *outbound,
36be9a76 3915 size_t outbound_size,
4157976b 3916 void *inbound,
e2a58ee5 3917 size_t inbound_size)
36be9a76 3918{
ecd4a68a
ID
3919 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3920 struct page *req_page = NULL;
3921 struct page *reply_page;
36be9a76
AE
3922 int ret;
3923
3924 /*
6010a451
AE
3925 * Method calls are ultimately read operations. The result
3926 * should placed into the inbound buffer provided. They
3927 * also supply outbound data--parameters for the object
3928 * method. Currently if this is present it will be a
3929 * snapshot id.
36be9a76 3930 */
ecd4a68a
ID
3931 if (outbound) {
3932 if (outbound_size > PAGE_SIZE)
3933 return -E2BIG;
36be9a76 3934
ecd4a68a
ID
3935 req_page = alloc_page(GFP_KERNEL);
3936 if (!req_page)
3937 return -ENOMEM;
04017e29 3938
ecd4a68a 3939 memcpy(page_address(req_page), outbound, outbound_size);
04017e29 3940 }
36be9a76 3941
ecd4a68a
ID
3942 reply_page = alloc_page(GFP_KERNEL);
3943 if (!reply_page) {
3944 if (req_page)
3945 __free_page(req_page);
3946 return -ENOMEM;
3947 }
57385b51 3948
ecd4a68a
ID
3949 ret = ceph_osdc_call(osdc, oid, oloc, RBD_DRV_NAME, method_name,
3950 CEPH_OSD_FLAG_READ, req_page, outbound_size,
3951 reply_page, &inbound_size);
3952 if (!ret) {
3953 memcpy(inbound, page_address(reply_page), inbound_size);
3954 ret = inbound_size;
3955 }
36be9a76 3956
ecd4a68a
ID
3957 if (req_page)
3958 __free_page(req_page);
3959 __free_page(reply_page);
36be9a76
AE
3960 return ret;
3961}
3962
ed95b21a
ID
3963/*
3964 * lock_rwsem must be held for read
3965 */
3966static void rbd_wait_state_locked(struct rbd_device *rbd_dev)
3967{
3968 DEFINE_WAIT(wait);
3969
3970 do {
3971 /*
3972 * Note the use of mod_delayed_work() in rbd_acquire_lock()
3973 * and cancel_delayed_work() in wake_requests().
3974 */
3975 dout("%s rbd_dev %p queueing lock_dwork\n", __func__, rbd_dev);
3976 queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
3977 prepare_to_wait_exclusive(&rbd_dev->lock_waitq, &wait,
3978 TASK_UNINTERRUPTIBLE);
3979 up_read(&rbd_dev->lock_rwsem);
3980 schedule();
3981 down_read(&rbd_dev->lock_rwsem);
87c0fded
ID
3982 } while (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED &&
3983 !test_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags));
3984
ed95b21a
ID
3985 finish_wait(&rbd_dev->lock_waitq, &wait);
3986}
3987
7ad18afa 3988static void rbd_queue_workfn(struct work_struct *work)
bf0d5f50 3989{
7ad18afa
CH
3990 struct request *rq = blk_mq_rq_from_pdu(work);
3991 struct rbd_device *rbd_dev = rq->q->queuedata;
bc1ecc65 3992 struct rbd_img_request *img_request;
4e752f0a 3993 struct ceph_snap_context *snapc = NULL;
bc1ecc65
ID
3994 u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
3995 u64 length = blk_rq_bytes(rq);
6d2940c8 3996 enum obj_operation_type op_type;
4e752f0a 3997 u64 mapping_size;
80de1912 3998 bool must_be_locked;
bf0d5f50
AE
3999 int result;
4000
aebf526b
CH
4001 switch (req_op(rq)) {
4002 case REQ_OP_DISCARD:
6ac56951 4003 case REQ_OP_WRITE_ZEROES:
90e98c52 4004 op_type = OBJ_OP_DISCARD;
aebf526b
CH
4005 break;
4006 case REQ_OP_WRITE:
6d2940c8 4007 op_type = OBJ_OP_WRITE;
aebf526b
CH
4008 break;
4009 case REQ_OP_READ:
6d2940c8 4010 op_type = OBJ_OP_READ;
aebf526b
CH
4011 break;
4012 default:
4013 dout("%s: non-fs request type %d\n", __func__, req_op(rq));
4014 result = -EIO;
4015 goto err;
4016 }
6d2940c8 4017
bc1ecc65 4018 /* Ignore/skip any zero-length requests */
bf0d5f50 4019
bc1ecc65
ID
4020 if (!length) {
4021 dout("%s: zero-length request\n", __func__);
4022 result = 0;
4023 goto err_rq;
4024 }
bf0d5f50 4025
9568c93e
ID
4026 rbd_assert(op_type == OBJ_OP_READ ||
4027 rbd_dev->spec->snap_id == CEPH_NOSNAP);
4dda41d3 4028
bc1ecc65
ID
4029 /*
4030 * Quit early if the mapped snapshot no longer exists. It's
4031 * still possible the snapshot will have disappeared by the
4032 * time our request arrives at the osd, but there's no sense in
4033 * sending it if we already know.
4034 */
4035 if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags)) {
4036 dout("request for non-existent snapshot");
4037 rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
4038 result = -ENXIO;
4039 goto err_rq;
4040 }
4dda41d3 4041
bc1ecc65
ID
4042 if (offset && length > U64_MAX - offset + 1) {
4043 rbd_warn(rbd_dev, "bad request range (%llu~%llu)", offset,
4044 length);
4045 result = -EINVAL;
4046 goto err_rq; /* Shouldn't happen */
4047 }
4dda41d3 4048
7ad18afa
CH
4049 blk_mq_start_request(rq);
4050
4e752f0a
JD
4051 down_read(&rbd_dev->header_rwsem);
4052 mapping_size = rbd_dev->mapping.size;
6d2940c8 4053 if (op_type != OBJ_OP_READ) {
4e752f0a
JD
4054 snapc = rbd_dev->header.snapc;
4055 ceph_get_snap_context(snapc);
4056 }
4057 up_read(&rbd_dev->header_rwsem);
4058
4059 if (offset + length > mapping_size) {
bc1ecc65 4060 rbd_warn(rbd_dev, "beyond EOD (%llu~%llu > %llu)", offset,
4e752f0a 4061 length, mapping_size);
bc1ecc65
ID
4062 result = -EIO;
4063 goto err_rq;
4064 }
bf0d5f50 4065
f9bebd58
ID
4066 must_be_locked =
4067 (rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK) &&
4068 (op_type != OBJ_OP_READ || rbd_dev->opts->lock_on_read);
ed95b21a
ID
4069 if (must_be_locked) {
4070 down_read(&rbd_dev->lock_rwsem);
87c0fded 4071 if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED &&
e010dd0a
ID
4072 !test_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags)) {
4073 if (rbd_dev->opts->exclusive) {
4074 rbd_warn(rbd_dev, "exclusive lock required");
4075 result = -EROFS;
4076 goto err_unlock;
4077 }
ed95b21a 4078 rbd_wait_state_locked(rbd_dev);
e010dd0a 4079 }
87c0fded
ID
4080 if (test_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags)) {
4081 result = -EBLACKLISTED;
4082 goto err_unlock;
4083 }
ed95b21a
ID
4084 }
4085
6d2940c8 4086 img_request = rbd_img_request_create(rbd_dev, offset, length, op_type,
4e752f0a 4087 snapc);
bc1ecc65
ID
4088 if (!img_request) {
4089 result = -ENOMEM;
ed95b21a 4090 goto err_unlock;
bc1ecc65
ID
4091 }
4092 img_request->rq = rq;
70b16db8 4093 snapc = NULL; /* img_request consumes a ref */
bf0d5f50 4094
90e98c52
GZ
4095 if (op_type == OBJ_OP_DISCARD)
4096 result = rbd_img_request_fill(img_request, OBJ_REQUEST_NODATA,
4097 NULL);
4098 else
4099 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO,
4100 rq->bio);
bc1ecc65
ID
4101 if (result)
4102 goto err_img_request;
bf0d5f50 4103
bc1ecc65
ID
4104 result = rbd_img_request_submit(img_request);
4105 if (result)
4106 goto err_img_request;
bf0d5f50 4107
ed95b21a
ID
4108 if (must_be_locked)
4109 up_read(&rbd_dev->lock_rwsem);
bc1ecc65 4110 return;
bf0d5f50 4111
bc1ecc65
ID
4112err_img_request:
4113 rbd_img_request_put(img_request);
ed95b21a
ID
4114err_unlock:
4115 if (must_be_locked)
4116 up_read(&rbd_dev->lock_rwsem);
bc1ecc65
ID
4117err_rq:
4118 if (result)
4119 rbd_warn(rbd_dev, "%s %llx at %llx result %d",
6d2940c8 4120 obj_op_name(op_type), length, offset, result);
e96a650a 4121 ceph_put_snap_context(snapc);
7ad18afa 4122err:
2a842aca 4123 blk_mq_end_request(rq, errno_to_blk_status(result));
bc1ecc65 4124}
bf0d5f50 4125
fc17b653 4126static blk_status_t rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
7ad18afa 4127 const struct blk_mq_queue_data *bd)
bc1ecc65 4128{
7ad18afa
CH
4129 struct request *rq = bd->rq;
4130 struct work_struct *work = blk_mq_rq_to_pdu(rq);
bf0d5f50 4131
7ad18afa 4132 queue_work(rbd_wq, work);
fc17b653 4133 return BLK_STS_OK;
bf0d5f50
AE
4134}
4135
602adf40
YS
4136static void rbd_free_disk(struct rbd_device *rbd_dev)
4137{
5769ed0c
ID
4138 blk_cleanup_queue(rbd_dev->disk->queue);
4139 blk_mq_free_tag_set(&rbd_dev->tag_set);
4140 put_disk(rbd_dev->disk);
a0cab924 4141 rbd_dev->disk = NULL;
602adf40
YS
4142}
4143
788e2df3 4144static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
fe5478e0
ID
4145 struct ceph_object_id *oid,
4146 struct ceph_object_locator *oloc,
4147 void *buf, int buf_len)
788e2df3
AE
4148
4149{
fe5478e0
ID
4150 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4151 struct ceph_osd_request *req;
4152 struct page **pages;
4153 int num_pages = calc_pages_for(0, buf_len);
788e2df3
AE
4154 int ret;
4155
fe5478e0
ID
4156 req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_KERNEL);
4157 if (!req)
4158 return -ENOMEM;
788e2df3 4159
fe5478e0
ID
4160 ceph_oid_copy(&req->r_base_oid, oid);
4161 ceph_oloc_copy(&req->r_base_oloc, oloc);
4162 req->r_flags = CEPH_OSD_FLAG_READ;
430c28c3 4163
fe5478e0 4164 ret = ceph_osdc_alloc_messages(req, GFP_KERNEL);
788e2df3 4165 if (ret)
fe5478e0 4166 goto out_req;
788e2df3 4167
fe5478e0
ID
4168 pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
4169 if (IS_ERR(pages)) {
4170 ret = PTR_ERR(pages);
4171 goto out_req;
4172 }
1ceae7ef 4173
fe5478e0
ID
4174 osd_req_op_extent_init(req, 0, CEPH_OSD_OP_READ, 0, buf_len, 0, 0);
4175 osd_req_op_extent_osd_data_pages(req, 0, pages, buf_len, 0, false,
4176 true);
4177
4178 ceph_osdc_start_request(osdc, req, false);
4179 ret = ceph_osdc_wait_request(osdc, req);
4180 if (ret >= 0)
4181 ceph_copy_from_page_vector(pages, buf, 0, ret);
788e2df3 4182
fe5478e0
ID
4183out_req:
4184 ceph_osdc_put_request(req);
788e2df3
AE
4185 return ret;
4186}
4187
602adf40 4188/*
662518b1
AE
4189 * Read the complete header for the given rbd device. On successful
4190 * return, the rbd_dev->header field will contain up-to-date
4191 * information about the image.
602adf40 4192 */
99a41ebc 4193static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev)
602adf40 4194{
4156d998 4195 struct rbd_image_header_ondisk *ondisk = NULL;
50f7c4c9 4196 u32 snap_count = 0;
4156d998
AE
4197 u64 names_size = 0;
4198 u32 want_count;
4199 int ret;
602adf40 4200
00f1f36f 4201 /*
4156d998
AE
4202 * The complete header will include an array of its 64-bit
4203 * snapshot ids, followed by the names of those snapshots as
4204 * a contiguous block of NUL-terminated strings. Note that
4205 * the number of snapshots could change by the time we read
4206 * it in, in which case we re-read it.
00f1f36f 4207 */
4156d998
AE
4208 do {
4209 size_t size;
4210
4211 kfree(ondisk);
4212
4213 size = sizeof (*ondisk);
4214 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
4215 size += names_size;
4216 ondisk = kmalloc(size, GFP_KERNEL);
4217 if (!ondisk)
662518b1 4218 return -ENOMEM;
4156d998 4219
fe5478e0
ID
4220 ret = rbd_obj_read_sync(rbd_dev, &rbd_dev->header_oid,
4221 &rbd_dev->header_oloc, ondisk, size);
4156d998 4222 if (ret < 0)
662518b1 4223 goto out;
c0cd10db 4224 if ((size_t)ret < size) {
4156d998 4225 ret = -ENXIO;
06ecc6cb
AE
4226 rbd_warn(rbd_dev, "short header read (want %zd got %d)",
4227 size, ret);
662518b1 4228 goto out;
4156d998
AE
4229 }
4230 if (!rbd_dev_ondisk_valid(ondisk)) {
4231 ret = -ENXIO;
06ecc6cb 4232 rbd_warn(rbd_dev, "invalid header");
662518b1 4233 goto out;
81e759fb 4234 }
602adf40 4235
4156d998
AE
4236 names_size = le64_to_cpu(ondisk->snap_names_len);
4237 want_count = snap_count;
4238 snap_count = le32_to_cpu(ondisk->snap_count);
4239 } while (snap_count != want_count);
00f1f36f 4240
662518b1
AE
4241 ret = rbd_header_from_disk(rbd_dev, ondisk);
4242out:
4156d998
AE
4243 kfree(ondisk);
4244
4245 return ret;
602adf40
YS
4246}
4247
15228ede
AE
4248/*
4249 * Clear the rbd device's EXISTS flag if the snapshot it's mapped to
4250 * has disappeared from the (just updated) snapshot context.
4251 */
4252static void rbd_exists_validate(struct rbd_device *rbd_dev)
4253{
4254 u64 snap_id;
4255
4256 if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags))
4257 return;
4258
4259 snap_id = rbd_dev->spec->snap_id;
4260 if (snap_id == CEPH_NOSNAP)
4261 return;
4262
4263 if (rbd_dev_snap_index(rbd_dev, snap_id) == BAD_SNAP_INDEX)
4264 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
4265}
4266
9875201e
JD
4267static void rbd_dev_update_size(struct rbd_device *rbd_dev)
4268{
4269 sector_t size;
9875201e
JD
4270
4271 /*
811c6688
ID
4272 * If EXISTS is not set, rbd_dev->disk may be NULL, so don't
4273 * try to update its size. If REMOVING is set, updating size
4274 * is just useless work since the device can't be opened.
9875201e 4275 */
811c6688
ID
4276 if (test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags) &&
4277 !test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) {
9875201e
JD
4278 size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
4279 dout("setting size to %llu sectors", (unsigned long long)size);
4280 set_capacity(rbd_dev->disk, size);
4281 revalidate_disk(rbd_dev->disk);
4282 }
4283}
4284
cc4a38bd 4285static int rbd_dev_refresh(struct rbd_device *rbd_dev)
1fe5e993 4286{
e627db08 4287 u64 mapping_size;
1fe5e993
AE
4288 int ret;
4289
cfbf6377 4290 down_write(&rbd_dev->header_rwsem);
3b5cf2a2 4291 mapping_size = rbd_dev->mapping.size;
a720ae09
ID
4292
4293 ret = rbd_dev_header_info(rbd_dev);
52bb1f9b 4294 if (ret)
73e39e4d 4295 goto out;
15228ede 4296
e8f59b59
ID
4297 /*
4298 * If there is a parent, see if it has disappeared due to the
4299 * mapped image getting flattened.
4300 */
4301 if (rbd_dev->parent) {
4302 ret = rbd_dev_v2_parent_info(rbd_dev);
4303 if (ret)
73e39e4d 4304 goto out;
e8f59b59
ID
4305 }
4306
5ff1108c 4307 if (rbd_dev->spec->snap_id == CEPH_NOSNAP) {
73e39e4d 4308 rbd_dev->mapping.size = rbd_dev->header.image_size;
5ff1108c
ID
4309 } else {
4310 /* validate mapped snapshot's EXISTS flag */
4311 rbd_exists_validate(rbd_dev);
4312 }
15228ede 4313
73e39e4d 4314out:
cfbf6377 4315 up_write(&rbd_dev->header_rwsem);
73e39e4d 4316 if (!ret && mapping_size != rbd_dev->mapping.size)
9875201e 4317 rbd_dev_update_size(rbd_dev);
1fe5e993 4318
73e39e4d 4319 return ret;
1fe5e993
AE
4320}
4321
d6296d39
CH
4322static int rbd_init_request(struct blk_mq_tag_set *set, struct request *rq,
4323 unsigned int hctx_idx, unsigned int numa_node)
7ad18afa
CH
4324{
4325 struct work_struct *work = blk_mq_rq_to_pdu(rq);
4326
4327 INIT_WORK(work, rbd_queue_workfn);
4328 return 0;
4329}
4330
f363b089 4331static const struct blk_mq_ops rbd_mq_ops = {
7ad18afa 4332 .queue_rq = rbd_queue_rq,
7ad18afa
CH
4333 .init_request = rbd_init_request,
4334};
4335
602adf40
YS
4336static int rbd_init_disk(struct rbd_device *rbd_dev)
4337{
4338 struct gendisk *disk;
4339 struct request_queue *q;
593a9e7b 4340 u64 segment_size;
7ad18afa 4341 int err;
602adf40 4342
602adf40 4343 /* create gendisk info */
7e513d43
ID
4344 disk = alloc_disk(single_major ?
4345 (1 << RBD_SINGLE_MAJOR_PART_SHIFT) :
4346 RBD_MINORS_PER_MAJOR);
602adf40 4347 if (!disk)
1fcdb8aa 4348 return -ENOMEM;
602adf40 4349
f0f8cef5 4350 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
de71a297 4351 rbd_dev->dev_id);
602adf40 4352 disk->major = rbd_dev->major;
dd82fff1 4353 disk->first_minor = rbd_dev->minor;
7e513d43
ID
4354 if (single_major)
4355 disk->flags |= GENHD_FL_EXT_DEVT;
602adf40
YS
4356 disk->fops = &rbd_bd_ops;
4357 disk->private_data = rbd_dev;
4358
7ad18afa
CH
4359 memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
4360 rbd_dev->tag_set.ops = &rbd_mq_ops;
b5584180 4361 rbd_dev->tag_set.queue_depth = rbd_dev->opts->queue_depth;
7ad18afa 4362 rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
b5584180 4363 rbd_dev->tag_set.flags = BLK_MQ_F_SHOULD_MERGE | BLK_MQ_F_SG_MERGE;
7ad18afa
CH
4364 rbd_dev->tag_set.nr_hw_queues = 1;
4365 rbd_dev->tag_set.cmd_size = sizeof(struct work_struct);
4366
4367 err = blk_mq_alloc_tag_set(&rbd_dev->tag_set);
4368 if (err)
602adf40 4369 goto out_disk;
029bcbd8 4370
7ad18afa
CH
4371 q = blk_mq_init_queue(&rbd_dev->tag_set);
4372 if (IS_ERR(q)) {
4373 err = PTR_ERR(q);
4374 goto out_tag_set;
4375 }
4376
d8a2c89c
ID
4377 queue_flag_set_unlocked(QUEUE_FLAG_NONROT, q);
4378 /* QUEUE_FLAG_ADD_RANDOM is off by default for blk-mq */
593a9e7b 4379
029bcbd8 4380 /* set io sizes to object size */
593a9e7b
AE
4381 segment_size = rbd_obj_bytes(&rbd_dev->header);
4382 blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE);
0d9fde4f 4383 q->limits.max_sectors = queue_max_hw_sectors(q);
d3834fef 4384 blk_queue_max_segments(q, segment_size / SECTOR_SIZE);
593a9e7b
AE
4385 blk_queue_max_segment_size(q, segment_size);
4386 blk_queue_io_min(q, segment_size);
4387 blk_queue_io_opt(q, segment_size);
029bcbd8 4388
90e98c52
GZ
4389 /* enable the discard support */
4390 queue_flag_set_unlocked(QUEUE_FLAG_DISCARD, q);
4391 q->limits.discard_granularity = segment_size;
2bb4cd5c 4392 blk_queue_max_discard_sectors(q, segment_size / SECTOR_SIZE);
6ac56951 4393 blk_queue_max_write_zeroes_sectors(q, segment_size / SECTOR_SIZE);
90e98c52 4394
bae818ee 4395 if (!ceph_test_opt(rbd_dev->rbd_client->client, NOCRC))
dc3b17cc 4396 q->backing_dev_info->capabilities |= BDI_CAP_STABLE_WRITES;
bae818ee 4397
5769ed0c
ID
4398 /*
4399 * disk_release() expects a queue ref from add_disk() and will
4400 * put it. Hold an extra ref until add_disk() is called.
4401 */
4402 WARN_ON(!blk_get_queue(q));
602adf40 4403 disk->queue = q;
602adf40
YS
4404 q->queuedata = rbd_dev;
4405
4406 rbd_dev->disk = disk;
602adf40 4407
602adf40 4408 return 0;
7ad18afa
CH
4409out_tag_set:
4410 blk_mq_free_tag_set(&rbd_dev->tag_set);
602adf40
YS
4411out_disk:
4412 put_disk(disk);
7ad18afa 4413 return err;
602adf40
YS
4414}
4415
dfc5606d
YS
4416/*
4417 sysfs
4418*/
4419
593a9e7b
AE
4420static struct rbd_device *dev_to_rbd_dev(struct device *dev)
4421{
4422 return container_of(dev, struct rbd_device, dev);
4423}
4424
dfc5606d
YS
4425static ssize_t rbd_size_show(struct device *dev,
4426 struct device_attribute *attr, char *buf)
4427{
593a9e7b 4428 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
a51aa0c0 4429
fc71d833
AE
4430 return sprintf(buf, "%llu\n",
4431 (unsigned long long)rbd_dev->mapping.size);
dfc5606d
YS
4432}
4433
34b13184
AE
4434/*
4435 * Note this shows the features for whatever's mapped, which is not
4436 * necessarily the base image.
4437 */
4438static ssize_t rbd_features_show(struct device *dev,
4439 struct device_attribute *attr, char *buf)
4440{
4441 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4442
4443 return sprintf(buf, "0x%016llx\n",
fc71d833 4444 (unsigned long long)rbd_dev->mapping.features);
34b13184
AE
4445}
4446
dfc5606d
YS
4447static ssize_t rbd_major_show(struct device *dev,
4448 struct device_attribute *attr, char *buf)
4449{
593a9e7b 4450 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
602adf40 4451
fc71d833
AE
4452 if (rbd_dev->major)
4453 return sprintf(buf, "%d\n", rbd_dev->major);
4454
4455 return sprintf(buf, "(none)\n");
dd82fff1
ID
4456}
4457
4458static ssize_t rbd_minor_show(struct device *dev,
4459 struct device_attribute *attr, char *buf)
4460{
4461 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
fc71d833 4462
dd82fff1 4463 return sprintf(buf, "%d\n", rbd_dev->minor);
dfc5606d
YS
4464}
4465
005a07bf
ID
4466static ssize_t rbd_client_addr_show(struct device *dev,
4467 struct device_attribute *attr, char *buf)
4468{
4469 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4470 struct ceph_entity_addr *client_addr =
4471 ceph_client_addr(rbd_dev->rbd_client->client);
4472
4473 return sprintf(buf, "%pISpc/%u\n", &client_addr->in_addr,
4474 le32_to_cpu(client_addr->nonce));
4475}
4476
dfc5606d
YS
4477static ssize_t rbd_client_id_show(struct device *dev,
4478 struct device_attribute *attr, char *buf)
602adf40 4479{
593a9e7b 4480 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
dfc5606d 4481
1dbb4399 4482 return sprintf(buf, "client%lld\n",
033268a5 4483 ceph_client_gid(rbd_dev->rbd_client->client));
602adf40
YS
4484}
4485
267fb90b
MC
4486static ssize_t rbd_cluster_fsid_show(struct device *dev,
4487 struct device_attribute *attr, char *buf)
4488{
4489 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4490
4491 return sprintf(buf, "%pU\n", &rbd_dev->rbd_client->client->fsid);
4492}
4493
0d6d1e9c
MC
4494static ssize_t rbd_config_info_show(struct device *dev,
4495 struct device_attribute *attr, char *buf)
4496{
4497 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4498
4499 return sprintf(buf, "%s\n", rbd_dev->config_info);
602adf40
YS
4500}
4501
dfc5606d
YS
4502static ssize_t rbd_pool_show(struct device *dev,
4503 struct device_attribute *attr, char *buf)
602adf40 4504{
593a9e7b 4505 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
dfc5606d 4506
0d7dbfce 4507 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
dfc5606d
YS
4508}
4509
9bb2f334
AE
4510static ssize_t rbd_pool_id_show(struct device *dev,
4511 struct device_attribute *attr, char *buf)
4512{
4513 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4514
0d7dbfce 4515 return sprintf(buf, "%llu\n",
fc71d833 4516 (unsigned long long) rbd_dev->spec->pool_id);
9bb2f334
AE
4517}
4518
dfc5606d
YS
4519static ssize_t rbd_name_show(struct device *dev,
4520 struct device_attribute *attr, char *buf)
4521{
593a9e7b 4522 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
dfc5606d 4523
a92ffdf8
AE
4524 if (rbd_dev->spec->image_name)
4525 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
4526
4527 return sprintf(buf, "(unknown)\n");
dfc5606d
YS
4528}
4529
589d30e0
AE
4530static ssize_t rbd_image_id_show(struct device *dev,
4531 struct device_attribute *attr, char *buf)
4532{
4533 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4534
0d7dbfce 4535 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
589d30e0
AE
4536}
4537
34b13184
AE
4538/*
4539 * Shows the name of the currently-mapped snapshot (or
4540 * RBD_SNAP_HEAD_NAME for the base image).
4541 */
dfc5606d
YS
4542static ssize_t rbd_snap_show(struct device *dev,
4543 struct device_attribute *attr,
4544 char *buf)
4545{
593a9e7b 4546 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
dfc5606d 4547
0d7dbfce 4548 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
dfc5606d
YS
4549}
4550
92a58671
MC
4551static ssize_t rbd_snap_id_show(struct device *dev,
4552 struct device_attribute *attr, char *buf)
4553{
4554 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4555
4556 return sprintf(buf, "%llu\n", rbd_dev->spec->snap_id);
4557}
4558
86b00e0d 4559/*
ff96128f
ID
4560 * For a v2 image, shows the chain of parent images, separated by empty
4561 * lines. For v1 images or if there is no parent, shows "(no parent
4562 * image)".
86b00e0d
AE
4563 */
4564static ssize_t rbd_parent_show(struct device *dev,
ff96128f
ID
4565 struct device_attribute *attr,
4566 char *buf)
86b00e0d
AE
4567{
4568 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
ff96128f 4569 ssize_t count = 0;
86b00e0d 4570
ff96128f 4571 if (!rbd_dev->parent)
86b00e0d
AE
4572 return sprintf(buf, "(no parent image)\n");
4573
ff96128f
ID
4574 for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) {
4575 struct rbd_spec *spec = rbd_dev->parent_spec;
4576
4577 count += sprintf(&buf[count], "%s"
4578 "pool_id %llu\npool_name %s\n"
4579 "image_id %s\nimage_name %s\n"
4580 "snap_id %llu\nsnap_name %s\n"
4581 "overlap %llu\n",
4582 !count ? "" : "\n", /* first? */
4583 spec->pool_id, spec->pool_name,
4584 spec->image_id, spec->image_name ?: "(unknown)",
4585 spec->snap_id, spec->snap_name,
4586 rbd_dev->parent_overlap);
4587 }
4588
4589 return count;
86b00e0d
AE
4590}
4591
dfc5606d
YS
4592static ssize_t rbd_image_refresh(struct device *dev,
4593 struct device_attribute *attr,
4594 const char *buf,
4595 size_t size)
4596{
593a9e7b 4597 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
b813623a 4598 int ret;
602adf40 4599
cc4a38bd 4600 ret = rbd_dev_refresh(rbd_dev);
e627db08 4601 if (ret)
52bb1f9b 4602 return ret;
b813623a 4603
52bb1f9b 4604 return size;
dfc5606d 4605}
602adf40 4606
dfc5606d 4607static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
34b13184 4608static DEVICE_ATTR(features, S_IRUGO, rbd_features_show, NULL);
dfc5606d 4609static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
dd82fff1 4610static DEVICE_ATTR(minor, S_IRUGO, rbd_minor_show, NULL);
005a07bf 4611static DEVICE_ATTR(client_addr, S_IRUGO, rbd_client_addr_show, NULL);
dfc5606d 4612static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
267fb90b 4613static DEVICE_ATTR(cluster_fsid, S_IRUGO, rbd_cluster_fsid_show, NULL);
0d6d1e9c 4614static DEVICE_ATTR(config_info, S_IRUSR, rbd_config_info_show, NULL);
dfc5606d 4615static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
9bb2f334 4616static DEVICE_ATTR(pool_id, S_IRUGO, rbd_pool_id_show, NULL);
dfc5606d 4617static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
589d30e0 4618static DEVICE_ATTR(image_id, S_IRUGO, rbd_image_id_show, NULL);
dfc5606d
YS
4619static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
4620static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
92a58671 4621static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
86b00e0d 4622static DEVICE_ATTR(parent, S_IRUGO, rbd_parent_show, NULL);
dfc5606d
YS
4623
4624static struct attribute *rbd_attrs[] = {
4625 &dev_attr_size.attr,
34b13184 4626 &dev_attr_features.attr,
dfc5606d 4627 &dev_attr_major.attr,
dd82fff1 4628 &dev_attr_minor.attr,
005a07bf 4629 &dev_attr_client_addr.attr,
dfc5606d 4630 &dev_attr_client_id.attr,
267fb90b 4631 &dev_attr_cluster_fsid.attr,
0d6d1e9c 4632 &dev_attr_config_info.attr,
dfc5606d 4633 &dev_attr_pool.attr,
9bb2f334 4634 &dev_attr_pool_id.attr,
dfc5606d 4635 &dev_attr_name.attr,
589d30e0 4636 &dev_attr_image_id.attr,
dfc5606d 4637 &dev_attr_current_snap.attr,
92a58671 4638 &dev_attr_snap_id.attr,
86b00e0d 4639 &dev_attr_parent.attr,
dfc5606d 4640 &dev_attr_refresh.attr,
dfc5606d
YS
4641 NULL
4642};
4643
4644static struct attribute_group rbd_attr_group = {
4645 .attrs = rbd_attrs,
4646};
4647
4648static const struct attribute_group *rbd_attr_groups[] = {
4649 &rbd_attr_group,
4650 NULL
4651};
4652
6cac4695 4653static void rbd_dev_release(struct device *dev);
dfc5606d 4654
b9942bc9 4655static const struct device_type rbd_device_type = {
dfc5606d
YS
4656 .name = "rbd",
4657 .groups = rbd_attr_groups,
6cac4695 4658 .release = rbd_dev_release,
dfc5606d
YS
4659};
4660
8b8fb99c
AE
4661static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
4662{
4663 kref_get(&spec->kref);
4664
4665 return spec;
4666}
4667
4668static void rbd_spec_free(struct kref *kref);
4669static void rbd_spec_put(struct rbd_spec *spec)
4670{
4671 if (spec)
4672 kref_put(&spec->kref, rbd_spec_free);
4673}
4674
4675static struct rbd_spec *rbd_spec_alloc(void)
4676{
4677 struct rbd_spec *spec;
4678
4679 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
4680 if (!spec)
4681 return NULL;
04077599
ID
4682
4683 spec->pool_id = CEPH_NOPOOL;
4684 spec->snap_id = CEPH_NOSNAP;
8b8fb99c
AE
4685 kref_init(&spec->kref);
4686
8b8fb99c
AE
4687 return spec;
4688}
4689
4690static void rbd_spec_free(struct kref *kref)
4691{
4692 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
4693
4694 kfree(spec->pool_name);
4695 kfree(spec->image_id);
4696 kfree(spec->image_name);
4697 kfree(spec->snap_name);
4698 kfree(spec);
4699}
4700
1643dfa4 4701static void rbd_dev_free(struct rbd_device *rbd_dev)
dd5ac32d 4702{
99d16943 4703 WARN_ON(rbd_dev->watch_state != RBD_WATCH_STATE_UNREGISTERED);
ed95b21a 4704 WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_UNLOCKED);
dd5ac32d 4705
c41d13a3 4706 ceph_oid_destroy(&rbd_dev->header_oid);
6b6dddbe 4707 ceph_oloc_destroy(&rbd_dev->header_oloc);
0d6d1e9c 4708 kfree(rbd_dev->config_info);
c41d13a3 4709
dd5ac32d
ID
4710 rbd_put_client(rbd_dev->rbd_client);
4711 rbd_spec_put(rbd_dev->spec);
4712 kfree(rbd_dev->opts);
4713 kfree(rbd_dev);
1643dfa4
ID
4714}
4715
4716static void rbd_dev_release(struct device *dev)
4717{
4718 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4719 bool need_put = !!rbd_dev->opts;
4720
4721 if (need_put) {
4722 destroy_workqueue(rbd_dev->task_wq);
4723 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
4724 }
4725
4726 rbd_dev_free(rbd_dev);
dd5ac32d
ID
4727
4728 /*
4729 * This is racy, but way better than putting module outside of
4730 * the release callback. The race window is pretty small, so
4731 * doing something similar to dm (dm-builtin.c) is overkill.
4732 */
4733 if (need_put)
4734 module_put(THIS_MODULE);
4735}
4736
1643dfa4
ID
4737static struct rbd_device *__rbd_dev_create(struct rbd_client *rbdc,
4738 struct rbd_spec *spec)
c53d5893
AE
4739{
4740 struct rbd_device *rbd_dev;
4741
1643dfa4 4742 rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
c53d5893
AE
4743 if (!rbd_dev)
4744 return NULL;
4745
4746 spin_lock_init(&rbd_dev->lock);
4747 INIT_LIST_HEAD(&rbd_dev->node);
c53d5893
AE
4748 init_rwsem(&rbd_dev->header_rwsem);
4749
7e97332e 4750 rbd_dev->header.data_pool_id = CEPH_NOPOOL;
c41d13a3 4751 ceph_oid_init(&rbd_dev->header_oid);
431a02cd 4752 rbd_dev->header_oloc.pool = spec->pool_id;
c41d13a3 4753
99d16943
ID
4754 mutex_init(&rbd_dev->watch_mutex);
4755 rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
4756 INIT_DELAYED_WORK(&rbd_dev->watch_dwork, rbd_reregister_watch);
4757
ed95b21a
ID
4758 init_rwsem(&rbd_dev->lock_rwsem);
4759 rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
4760 INIT_WORK(&rbd_dev->acquired_lock_work, rbd_notify_acquired_lock);
4761 INIT_WORK(&rbd_dev->released_lock_work, rbd_notify_released_lock);
4762 INIT_DELAYED_WORK(&rbd_dev->lock_dwork, rbd_acquire_lock);
4763 INIT_WORK(&rbd_dev->unlock_work, rbd_release_lock_work);
4764 init_waitqueue_head(&rbd_dev->lock_waitq);
4765
dd5ac32d
ID
4766 rbd_dev->dev.bus = &rbd_bus_type;
4767 rbd_dev->dev.type = &rbd_device_type;
4768 rbd_dev->dev.parent = &rbd_root_dev;
dd5ac32d
ID
4769 device_initialize(&rbd_dev->dev);
4770
c53d5893 4771 rbd_dev->rbd_client = rbdc;
d147543d 4772 rbd_dev->spec = spec;
0903e875 4773
1643dfa4
ID
4774 return rbd_dev;
4775}
4776
4777/*
4778 * Create a mapping rbd_dev.
4779 */
4780static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
4781 struct rbd_spec *spec,
4782 struct rbd_options *opts)
4783{
4784 struct rbd_device *rbd_dev;
4785
4786 rbd_dev = __rbd_dev_create(rbdc, spec);
4787 if (!rbd_dev)
4788 return NULL;
4789
4790 rbd_dev->opts = opts;
4791
4792 /* get an id and fill in device name */
4793 rbd_dev->dev_id = ida_simple_get(&rbd_dev_id_ida, 0,
4794 minor_to_rbd_dev_id(1 << MINORBITS),
4795 GFP_KERNEL);
4796 if (rbd_dev->dev_id < 0)
4797 goto fail_rbd_dev;
4798
4799 sprintf(rbd_dev->name, RBD_DRV_NAME "%d", rbd_dev->dev_id);
4800 rbd_dev->task_wq = alloc_ordered_workqueue("%s-tasks", WQ_MEM_RECLAIM,
4801 rbd_dev->name);
4802 if (!rbd_dev->task_wq)
4803 goto fail_dev_id;
dd5ac32d 4804
1643dfa4
ID
4805 /* we have a ref from do_rbd_add() */
4806 __module_get(THIS_MODULE);
dd5ac32d 4807
1643dfa4 4808 dout("%s rbd_dev %p dev_id %d\n", __func__, rbd_dev, rbd_dev->dev_id);
c53d5893 4809 return rbd_dev;
1643dfa4
ID
4810
4811fail_dev_id:
4812 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
4813fail_rbd_dev:
4814 rbd_dev_free(rbd_dev);
4815 return NULL;
c53d5893
AE
4816}
4817
4818static void rbd_dev_destroy(struct rbd_device *rbd_dev)
4819{
dd5ac32d
ID
4820 if (rbd_dev)
4821 put_device(&rbd_dev->dev);
c53d5893
AE
4822}
4823
9d475de5
AE
4824/*
4825 * Get the size and object order for an image snapshot, or if
4826 * snap_id is CEPH_NOSNAP, gets this information for the base
4827 * image.
4828 */
4829static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
4830 u8 *order, u64 *snap_size)
4831{
4832 __le64 snapid = cpu_to_le64(snap_id);
4833 int ret;
4834 struct {
4835 u8 order;
4836 __le64 size;
4837 } __attribute__ ((packed)) size_buf = { 0 };
4838
ecd4a68a
ID
4839 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
4840 &rbd_dev->header_oloc, "get_size",
4841 &snapid, sizeof(snapid),
4842 &size_buf, sizeof(size_buf));
36be9a76 4843 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
9d475de5
AE
4844 if (ret < 0)
4845 return ret;
57385b51
AE
4846 if (ret < sizeof (size_buf))
4847 return -ERANGE;
9d475de5 4848
c3545579 4849 if (order) {
c86f86e9 4850 *order = size_buf.order;
c3545579
JD
4851 dout(" order %u", (unsigned int)*order);
4852 }
9d475de5
AE
4853 *snap_size = le64_to_cpu(size_buf.size);
4854
c3545579
JD
4855 dout(" snap_id 0x%016llx snap_size = %llu\n",
4856 (unsigned long long)snap_id,
57385b51 4857 (unsigned long long)*snap_size);
9d475de5
AE
4858
4859 return 0;
4860}
4861
4862static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
4863{
4864 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
4865 &rbd_dev->header.obj_order,
4866 &rbd_dev->header.image_size);
4867}
4868
1e130199
AE
4869static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
4870{
4871 void *reply_buf;
4872 int ret;
4873 void *p;
4874
4875 reply_buf = kzalloc(RBD_OBJ_PREFIX_LEN_MAX, GFP_KERNEL);
4876 if (!reply_buf)
4877 return -ENOMEM;
4878
ecd4a68a
ID
4879 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
4880 &rbd_dev->header_oloc, "get_object_prefix",
4881 NULL, 0, reply_buf, RBD_OBJ_PREFIX_LEN_MAX);
36be9a76 4882 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
1e130199
AE
4883 if (ret < 0)
4884 goto out;
4885
4886 p = reply_buf;
4887 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
57385b51
AE
4888 p + ret, NULL, GFP_NOIO);
4889 ret = 0;
1e130199
AE
4890
4891 if (IS_ERR(rbd_dev->header.object_prefix)) {
4892 ret = PTR_ERR(rbd_dev->header.object_prefix);
4893 rbd_dev->header.object_prefix = NULL;
4894 } else {
4895 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix);
4896 }
1e130199
AE
4897out:
4898 kfree(reply_buf);
4899
4900 return ret;
4901}
4902
b1b5402a
AE
4903static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
4904 u64 *snap_features)
4905{
4906 __le64 snapid = cpu_to_le64(snap_id);
4907 struct {
4908 __le64 features;
4909 __le64 incompat;
4157976b 4910 } __attribute__ ((packed)) features_buf = { 0 };
d3767f0f 4911 u64 unsup;
b1b5402a
AE
4912 int ret;
4913
ecd4a68a
ID
4914 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
4915 &rbd_dev->header_oloc, "get_features",
4916 &snapid, sizeof(snapid),
4917 &features_buf, sizeof(features_buf));
36be9a76 4918 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
b1b5402a
AE
4919 if (ret < 0)
4920 return ret;
57385b51
AE
4921 if (ret < sizeof (features_buf))
4922 return -ERANGE;
d889140c 4923
d3767f0f
ID
4924 unsup = le64_to_cpu(features_buf.incompat) & ~RBD_FEATURES_SUPPORTED;
4925 if (unsup) {
4926 rbd_warn(rbd_dev, "image uses unsupported features: 0x%llx",
4927 unsup);
b8f5c6ed 4928 return -ENXIO;
d3767f0f 4929 }
d889140c 4930
b1b5402a
AE
4931 *snap_features = le64_to_cpu(features_buf.features);
4932
4933 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
57385b51
AE
4934 (unsigned long long)snap_id,
4935 (unsigned long long)*snap_features,
4936 (unsigned long long)le64_to_cpu(features_buf.incompat));
b1b5402a
AE
4937
4938 return 0;
4939}
4940
4941static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
4942{
4943 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
4944 &rbd_dev->header.features);
4945}
4946
86b00e0d
AE
4947static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
4948{
4949 struct rbd_spec *parent_spec;
4950 size_t size;
4951 void *reply_buf = NULL;
4952 __le64 snapid;
4953 void *p;
4954 void *end;
642a2537 4955 u64 pool_id;
86b00e0d 4956 char *image_id;
3b5cf2a2 4957 u64 snap_id;
86b00e0d 4958 u64 overlap;
86b00e0d
AE
4959 int ret;
4960
4961 parent_spec = rbd_spec_alloc();
4962 if (!parent_spec)
4963 return -ENOMEM;
4964
4965 size = sizeof (__le64) + /* pool_id */
4966 sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX + /* image_id */
4967 sizeof (__le64) + /* snap_id */
4968 sizeof (__le64); /* overlap */
4969 reply_buf = kmalloc(size, GFP_KERNEL);
4970 if (!reply_buf) {
4971 ret = -ENOMEM;
4972 goto out_err;
4973 }
4974
4d9b67cd 4975 snapid = cpu_to_le64(rbd_dev->spec->snap_id);
ecd4a68a
ID
4976 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
4977 &rbd_dev->header_oloc, "get_parent",
4978 &snapid, sizeof(snapid), reply_buf, size);
36be9a76 4979 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
86b00e0d
AE
4980 if (ret < 0)
4981 goto out_err;
4982
86b00e0d 4983 p = reply_buf;
57385b51
AE
4984 end = reply_buf + ret;
4985 ret = -ERANGE;
642a2537 4986 ceph_decode_64_safe(&p, end, pool_id, out_err);
392a9dad
AE
4987 if (pool_id == CEPH_NOPOOL) {
4988 /*
4989 * Either the parent never existed, or we have
4990 * record of it but the image got flattened so it no
4991 * longer has a parent. When the parent of a
4992 * layered image disappears we immediately set the
4993 * overlap to 0. The effect of this is that all new
4994 * requests will be treated as if the image had no
4995 * parent.
4996 */
4997 if (rbd_dev->parent_overlap) {
4998 rbd_dev->parent_overlap = 0;
392a9dad
AE
4999 rbd_dev_parent_put(rbd_dev);
5000 pr_info("%s: clone image has been flattened\n",
5001 rbd_dev->disk->disk_name);
5002 }
5003
86b00e0d 5004 goto out; /* No parent? No problem. */
392a9dad 5005 }
86b00e0d 5006
0903e875
AE
5007 /* The ceph file layout needs to fit pool id in 32 bits */
5008
5009 ret = -EIO;
642a2537 5010 if (pool_id > (u64)U32_MAX) {
9584d508 5011 rbd_warn(NULL, "parent pool id too large (%llu > %u)",
642a2537 5012 (unsigned long long)pool_id, U32_MAX);
57385b51 5013 goto out_err;
c0cd10db 5014 }
0903e875 5015
979ed480 5016 image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
86b00e0d
AE
5017 if (IS_ERR(image_id)) {
5018 ret = PTR_ERR(image_id);
5019 goto out_err;
5020 }
3b5cf2a2 5021 ceph_decode_64_safe(&p, end, snap_id, out_err);
86b00e0d
AE
5022 ceph_decode_64_safe(&p, end, overlap, out_err);
5023
3b5cf2a2
AE
5024 /*
5025 * The parent won't change (except when the clone is
5026 * flattened, already handled that). So we only need to
5027 * record the parent spec we have not already done so.
5028 */
5029 if (!rbd_dev->parent_spec) {
5030 parent_spec->pool_id = pool_id;
5031 parent_spec->image_id = image_id;
5032 parent_spec->snap_id = snap_id;
70cf49cf
AE
5033 rbd_dev->parent_spec = parent_spec;
5034 parent_spec = NULL; /* rbd_dev now owns this */
fbba11b3
ID
5035 } else {
5036 kfree(image_id);
3b5cf2a2
AE
5037 }
5038
5039 /*
cf32bd9c
ID
5040 * We always update the parent overlap. If it's zero we issue
5041 * a warning, as we will proceed as if there was no parent.
3b5cf2a2 5042 */
3b5cf2a2 5043 if (!overlap) {
3b5cf2a2 5044 if (parent_spec) {
cf32bd9c
ID
5045 /* refresh, careful to warn just once */
5046 if (rbd_dev->parent_overlap)
5047 rbd_warn(rbd_dev,
5048 "clone now standalone (overlap became 0)");
3b5cf2a2 5049 } else {
cf32bd9c
ID
5050 /* initial probe */
5051 rbd_warn(rbd_dev, "clone is standalone (overlap 0)");
3b5cf2a2 5052 }
70cf49cf 5053 }
cf32bd9c
ID
5054 rbd_dev->parent_overlap = overlap;
5055
86b00e0d
AE
5056out:
5057 ret = 0;
5058out_err:
5059 kfree(reply_buf);
5060 rbd_spec_put(parent_spec);
5061
5062 return ret;
5063}
5064
cc070d59
AE
5065static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev)
5066{
5067 struct {
5068 __le64 stripe_unit;
5069 __le64 stripe_count;
5070 } __attribute__ ((packed)) striping_info_buf = { 0 };
5071 size_t size = sizeof (striping_info_buf);
5072 void *p;
5073 u64 obj_size;
5074 u64 stripe_unit;
5075 u64 stripe_count;
5076 int ret;
5077
ecd4a68a
ID
5078 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5079 &rbd_dev->header_oloc, "get_stripe_unit_count",
5080 NULL, 0, &striping_info_buf, size);
cc070d59
AE
5081 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5082 if (ret < 0)
5083 return ret;
5084 if (ret < size)
5085 return -ERANGE;
5086
5087 /*
5088 * We don't actually support the "fancy striping" feature
5089 * (STRIPINGV2) yet, but if the striping sizes are the
5090 * defaults the behavior is the same as before. So find
5091 * out, and only fail if the image has non-default values.
5092 */
5093 ret = -EINVAL;
5bc3fb17 5094 obj_size = rbd_obj_bytes(&rbd_dev->header);
cc070d59
AE
5095 p = &striping_info_buf;
5096 stripe_unit = ceph_decode_64(&p);
5097 if (stripe_unit != obj_size) {
5098 rbd_warn(rbd_dev, "unsupported stripe unit "
5099 "(got %llu want %llu)",
5100 stripe_unit, obj_size);
5101 return -EINVAL;
5102 }
5103 stripe_count = ceph_decode_64(&p);
5104 if (stripe_count != 1) {
5105 rbd_warn(rbd_dev, "unsupported stripe count "
5106 "(got %llu want 1)", stripe_count);
5107 return -EINVAL;
5108 }
500d0c0f
AE
5109 rbd_dev->header.stripe_unit = stripe_unit;
5110 rbd_dev->header.stripe_count = stripe_count;
cc070d59
AE
5111
5112 return 0;
5113}
5114
7e97332e
ID
5115static int rbd_dev_v2_data_pool(struct rbd_device *rbd_dev)
5116{
5117 __le64 data_pool_id;
5118 int ret;
5119
5120 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5121 &rbd_dev->header_oloc, "get_data_pool",
5122 NULL, 0, &data_pool_id, sizeof(data_pool_id));
5123 if (ret < 0)
5124 return ret;
5125 if (ret < sizeof(data_pool_id))
5126 return -EBADMSG;
5127
5128 rbd_dev->header.data_pool_id = le64_to_cpu(data_pool_id);
5129 WARN_ON(rbd_dev->header.data_pool_id == CEPH_NOPOOL);
5130 return 0;
5131}
5132
9e15b77d
AE
5133static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
5134{
ecd4a68a 5135 CEPH_DEFINE_OID_ONSTACK(oid);
9e15b77d
AE
5136 size_t image_id_size;
5137 char *image_id;
5138 void *p;
5139 void *end;
5140 size_t size;
5141 void *reply_buf = NULL;
5142 size_t len = 0;
5143 char *image_name = NULL;
5144 int ret;
5145
5146 rbd_assert(!rbd_dev->spec->image_name);
5147
69e7a02f
AE
5148 len = strlen(rbd_dev->spec->image_id);
5149 image_id_size = sizeof (__le32) + len;
9e15b77d
AE
5150 image_id = kmalloc(image_id_size, GFP_KERNEL);
5151 if (!image_id)
5152 return NULL;
5153
5154 p = image_id;
4157976b 5155 end = image_id + image_id_size;
57385b51 5156 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len);
9e15b77d
AE
5157
5158 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
5159 reply_buf = kmalloc(size, GFP_KERNEL);
5160 if (!reply_buf)
5161 goto out;
5162
ecd4a68a
ID
5163 ceph_oid_printf(&oid, "%s", RBD_DIRECTORY);
5164 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
5165 "dir_get_name", image_id, image_id_size,
5166 reply_buf, size);
9e15b77d
AE
5167 if (ret < 0)
5168 goto out;
5169 p = reply_buf;
f40eb349
AE
5170 end = reply_buf + ret;
5171
9e15b77d
AE
5172 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
5173 if (IS_ERR(image_name))
5174 image_name = NULL;
5175 else
5176 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
5177out:
5178 kfree(reply_buf);
5179 kfree(image_id);
5180
5181 return image_name;
5182}
5183
2ad3d716
AE
5184static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5185{
5186 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5187 const char *snap_name;
5188 u32 which = 0;
5189
5190 /* Skip over names until we find the one we are looking for */
5191
5192 snap_name = rbd_dev->header.snap_names;
5193 while (which < snapc->num_snaps) {
5194 if (!strcmp(name, snap_name))
5195 return snapc->snaps[which];
5196 snap_name += strlen(snap_name) + 1;
5197 which++;
5198 }
5199 return CEPH_NOSNAP;
5200}
5201
5202static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5203{
5204 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5205 u32 which;
5206 bool found = false;
5207 u64 snap_id;
5208
5209 for (which = 0; !found && which < snapc->num_snaps; which++) {
5210 const char *snap_name;
5211
5212 snap_id = snapc->snaps[which];
5213 snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
efadc98a
JD
5214 if (IS_ERR(snap_name)) {
5215 /* ignore no-longer existing snapshots */
5216 if (PTR_ERR(snap_name) == -ENOENT)
5217 continue;
5218 else
5219 break;
5220 }
2ad3d716
AE
5221 found = !strcmp(name, snap_name);
5222 kfree(snap_name);
5223 }
5224 return found ? snap_id : CEPH_NOSNAP;
5225}
5226
5227/*
5228 * Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if
5229 * no snapshot by that name is found, or if an error occurs.
5230 */
5231static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5232{
5233 if (rbd_dev->image_format == 1)
5234 return rbd_v1_snap_id_by_name(rbd_dev, name);
5235
5236 return rbd_v2_snap_id_by_name(rbd_dev, name);
5237}
5238
9e15b77d 5239/*
04077599
ID
5240 * An image being mapped will have everything but the snap id.
5241 */
5242static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev)
5243{
5244 struct rbd_spec *spec = rbd_dev->spec;
5245
5246 rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name);
5247 rbd_assert(spec->image_id && spec->image_name);
5248 rbd_assert(spec->snap_name);
5249
5250 if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) {
5251 u64 snap_id;
5252
5253 snap_id = rbd_snap_id_by_name(rbd_dev, spec->snap_name);
5254 if (snap_id == CEPH_NOSNAP)
5255 return -ENOENT;
5256
5257 spec->snap_id = snap_id;
5258 } else {
5259 spec->snap_id = CEPH_NOSNAP;
5260 }
5261
5262 return 0;
5263}
5264
5265/*
5266 * A parent image will have all ids but none of the names.
e1d4213f 5267 *
04077599
ID
5268 * All names in an rbd spec are dynamically allocated. It's OK if we
5269 * can't figure out the name for an image id.
9e15b77d 5270 */
04077599 5271static int rbd_spec_fill_names(struct rbd_device *rbd_dev)
9e15b77d 5272{
2e9f7f1c
AE
5273 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5274 struct rbd_spec *spec = rbd_dev->spec;
5275 const char *pool_name;
5276 const char *image_name;
5277 const char *snap_name;
9e15b77d
AE
5278 int ret;
5279
04077599
ID
5280 rbd_assert(spec->pool_id != CEPH_NOPOOL);
5281 rbd_assert(spec->image_id);
5282 rbd_assert(spec->snap_id != CEPH_NOSNAP);
9e15b77d 5283
2e9f7f1c 5284 /* Get the pool name; we have to make our own copy of this */
9e15b77d 5285
2e9f7f1c
AE
5286 pool_name = ceph_pg_pool_name_by_id(osdc->osdmap, spec->pool_id);
5287 if (!pool_name) {
5288 rbd_warn(rbd_dev, "no pool with id %llu", spec->pool_id);
935dc89f
AE
5289 return -EIO;
5290 }
2e9f7f1c
AE
5291 pool_name = kstrdup(pool_name, GFP_KERNEL);
5292 if (!pool_name)
9e15b77d
AE
5293 return -ENOMEM;
5294
5295 /* Fetch the image name; tolerate failure here */
5296
2e9f7f1c
AE
5297 image_name = rbd_dev_image_name(rbd_dev);
5298 if (!image_name)
06ecc6cb 5299 rbd_warn(rbd_dev, "unable to get image name");
9e15b77d 5300
04077599 5301 /* Fetch the snapshot name */
9e15b77d 5302
2e9f7f1c 5303 snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
da6a6b63
JD
5304 if (IS_ERR(snap_name)) {
5305 ret = PTR_ERR(snap_name);
9e15b77d 5306 goto out_err;
2e9f7f1c
AE
5307 }
5308
5309 spec->pool_name = pool_name;
5310 spec->image_name = image_name;
5311 spec->snap_name = snap_name;
9e15b77d
AE
5312
5313 return 0;
04077599 5314
9e15b77d 5315out_err:
2e9f7f1c
AE
5316 kfree(image_name);
5317 kfree(pool_name);
9e15b77d
AE
5318 return ret;
5319}
5320
cc4a38bd 5321static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev)
35d489f9
AE
5322{
5323 size_t size;
5324 int ret;
5325 void *reply_buf;
5326 void *p;
5327 void *end;
5328 u64 seq;
5329 u32 snap_count;
5330 struct ceph_snap_context *snapc;
5331 u32 i;
5332
5333 /*
5334 * We'll need room for the seq value (maximum snapshot id),
5335 * snapshot count, and array of that many snapshot ids.
5336 * For now we have a fixed upper limit on the number we're
5337 * prepared to receive.
5338 */
5339 size = sizeof (__le64) + sizeof (__le32) +
5340 RBD_MAX_SNAP_COUNT * sizeof (__le64);
5341 reply_buf = kzalloc(size, GFP_KERNEL);
5342 if (!reply_buf)
5343 return -ENOMEM;
5344
ecd4a68a
ID
5345 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5346 &rbd_dev->header_oloc, "get_snapcontext",
5347 NULL, 0, reply_buf, size);
36be9a76 5348 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
35d489f9
AE
5349 if (ret < 0)
5350 goto out;
5351
35d489f9 5352 p = reply_buf;
57385b51
AE
5353 end = reply_buf + ret;
5354 ret = -ERANGE;
35d489f9
AE
5355 ceph_decode_64_safe(&p, end, seq, out);
5356 ceph_decode_32_safe(&p, end, snap_count, out);
5357
5358 /*
5359 * Make sure the reported number of snapshot ids wouldn't go
5360 * beyond the end of our buffer. But before checking that,
5361 * make sure the computed size of the snapshot context we
5362 * allocate is representable in a size_t.
5363 */
5364 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
5365 / sizeof (u64)) {
5366 ret = -EINVAL;
5367 goto out;
5368 }
5369 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
5370 goto out;
468521c1 5371 ret = 0;
35d489f9 5372
812164f8 5373 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
35d489f9
AE
5374 if (!snapc) {
5375 ret = -ENOMEM;
5376 goto out;
5377 }
35d489f9 5378 snapc->seq = seq;
35d489f9
AE
5379 for (i = 0; i < snap_count; i++)
5380 snapc->snaps[i] = ceph_decode_64(&p);
5381
49ece554 5382 ceph_put_snap_context(rbd_dev->header.snapc);
35d489f9
AE
5383 rbd_dev->header.snapc = snapc;
5384
5385 dout(" snap context seq = %llu, snap_count = %u\n",
57385b51 5386 (unsigned long long)seq, (unsigned int)snap_count);
35d489f9
AE
5387out:
5388 kfree(reply_buf);
5389
57385b51 5390 return ret;
35d489f9
AE
5391}
5392
54cac61f
AE
5393static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
5394 u64 snap_id)
b8b1e2db
AE
5395{
5396 size_t size;
5397 void *reply_buf;
54cac61f 5398 __le64 snapid;
b8b1e2db
AE
5399 int ret;
5400 void *p;
5401 void *end;
b8b1e2db
AE
5402 char *snap_name;
5403
5404 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
5405 reply_buf = kmalloc(size, GFP_KERNEL);
5406 if (!reply_buf)
5407 return ERR_PTR(-ENOMEM);
5408
54cac61f 5409 snapid = cpu_to_le64(snap_id);
ecd4a68a
ID
5410 ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5411 &rbd_dev->header_oloc, "get_snapshot_name",
5412 &snapid, sizeof(snapid), reply_buf, size);
36be9a76 5413 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
f40eb349
AE
5414 if (ret < 0) {
5415 snap_name = ERR_PTR(ret);
b8b1e2db 5416 goto out;
f40eb349 5417 }
b8b1e2db
AE
5418
5419 p = reply_buf;
f40eb349 5420 end = reply_buf + ret;
e5c35534 5421 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
f40eb349 5422 if (IS_ERR(snap_name))
b8b1e2db 5423 goto out;
b8b1e2db 5424
f40eb349 5425 dout(" snap_id 0x%016llx snap_name = %s\n",
54cac61f 5426 (unsigned long long)snap_id, snap_name);
b8b1e2db
AE
5427out:
5428 kfree(reply_buf);
5429
f40eb349 5430 return snap_name;
b8b1e2db
AE
5431}
5432
2df3fac7 5433static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev)
117973fb 5434{
2df3fac7 5435 bool first_time = rbd_dev->header.object_prefix == NULL;
117973fb 5436 int ret;
117973fb 5437
1617e40c
JD
5438 ret = rbd_dev_v2_image_size(rbd_dev);
5439 if (ret)
cfbf6377 5440 return ret;
1617e40c 5441
2df3fac7
AE
5442 if (first_time) {
5443 ret = rbd_dev_v2_header_onetime(rbd_dev);
5444 if (ret)
cfbf6377 5445 return ret;
2df3fac7
AE
5446 }
5447
cc4a38bd 5448 ret = rbd_dev_v2_snap_context(rbd_dev);
d194cd1d
ID
5449 if (ret && first_time) {
5450 kfree(rbd_dev->header.object_prefix);
5451 rbd_dev->header.object_prefix = NULL;
5452 }
117973fb
AE
5453
5454 return ret;
5455}
5456
a720ae09
ID
5457static int rbd_dev_header_info(struct rbd_device *rbd_dev)
5458{
5459 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
5460
5461 if (rbd_dev->image_format == 1)
5462 return rbd_dev_v1_header_info(rbd_dev);
5463
5464 return rbd_dev_v2_header_info(rbd_dev);
5465}
5466
e28fff26
AE
5467/*
5468 * Skips over white space at *buf, and updates *buf to point to the
5469 * first found non-space character (if any). Returns the length of
593a9e7b
AE
5470 * the token (string of non-white space characters) found. Note
5471 * that *buf must be terminated with '\0'.
e28fff26
AE
5472 */
5473static inline size_t next_token(const char **buf)
5474{
5475 /*
5476 * These are the characters that produce nonzero for
5477 * isspace() in the "C" and "POSIX" locales.
5478 */
5479 const char *spaces = " \f\n\r\t\v";
5480
5481 *buf += strspn(*buf, spaces); /* Find start of token */
5482
5483 return strcspn(*buf, spaces); /* Return token length */
5484}
5485
ea3352f4
AE
5486/*
5487 * Finds the next token in *buf, dynamically allocates a buffer big
5488 * enough to hold a copy of it, and copies the token into the new
5489 * buffer. The copy is guaranteed to be terminated with '\0'. Note
5490 * that a duplicate buffer is created even for a zero-length token.
5491 *
5492 * Returns a pointer to the newly-allocated duplicate, or a null
5493 * pointer if memory for the duplicate was not available. If
5494 * the lenp argument is a non-null pointer, the length of the token
5495 * (not including the '\0') is returned in *lenp.
5496 *
5497 * If successful, the *buf pointer will be updated to point beyond
5498 * the end of the found token.
5499 *
5500 * Note: uses GFP_KERNEL for allocation.
5501 */
5502static inline char *dup_token(const char **buf, size_t *lenp)
5503{
5504 char *dup;
5505 size_t len;
5506
5507 len = next_token(buf);
4caf35f9 5508 dup = kmemdup(*buf, len + 1, GFP_KERNEL);
ea3352f4
AE
5509 if (!dup)
5510 return NULL;
ea3352f4
AE
5511 *(dup + len) = '\0';
5512 *buf += len;
5513
5514 if (lenp)
5515 *lenp = len;
5516
5517 return dup;
5518}
5519
a725f65e 5520/*
859c31df
AE
5521 * Parse the options provided for an "rbd add" (i.e., rbd image
5522 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
5523 * and the data written is passed here via a NUL-terminated buffer.
5524 * Returns 0 if successful or an error code otherwise.
d22f76e7 5525 *
859c31df
AE
5526 * The information extracted from these options is recorded in
5527 * the other parameters which return dynamically-allocated
5528 * structures:
5529 * ceph_opts
5530 * The address of a pointer that will refer to a ceph options
5531 * structure. Caller must release the returned pointer using
5532 * ceph_destroy_options() when it is no longer needed.
5533 * rbd_opts
5534 * Address of an rbd options pointer. Fully initialized by
5535 * this function; caller must release with kfree().
5536 * spec
5537 * Address of an rbd image specification pointer. Fully
5538 * initialized by this function based on parsed options.
5539 * Caller must release with rbd_spec_put().
5540 *
5541 * The options passed take this form:
5542 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
5543 * where:
5544 * <mon_addrs>
5545 * A comma-separated list of one or more monitor addresses.
5546 * A monitor address is an ip address, optionally followed
5547 * by a port number (separated by a colon).
5548 * I.e.: ip1[:port1][,ip2[:port2]...]
5549 * <options>
5550 * A comma-separated list of ceph and/or rbd options.
5551 * <pool_name>
5552 * The name of the rados pool containing the rbd image.
5553 * <image_name>
5554 * The name of the image in that pool to map.
5555 * <snap_id>
5556 * An optional snapshot id. If provided, the mapping will
5557 * present data from the image at the time that snapshot was
5558 * created. The image head is used if no snapshot id is
5559 * provided. Snapshot mappings are always read-only.
a725f65e 5560 */
859c31df 5561static int rbd_add_parse_args(const char *buf,
dc79b113 5562 struct ceph_options **ceph_opts,
859c31df
AE
5563 struct rbd_options **opts,
5564 struct rbd_spec **rbd_spec)
e28fff26 5565{
d22f76e7 5566 size_t len;
859c31df 5567 char *options;
0ddebc0c 5568 const char *mon_addrs;
ecb4dc22 5569 char *snap_name;
0ddebc0c 5570 size_t mon_addrs_size;
859c31df 5571 struct rbd_spec *spec = NULL;
4e9afeba 5572 struct rbd_options *rbd_opts = NULL;
859c31df 5573 struct ceph_options *copts;
dc79b113 5574 int ret;
e28fff26
AE
5575
5576 /* The first four tokens are required */
5577
7ef3214a 5578 len = next_token(&buf);
4fb5d671
AE
5579 if (!len) {
5580 rbd_warn(NULL, "no monitor address(es) provided");
5581 return -EINVAL;
5582 }
0ddebc0c 5583 mon_addrs = buf;
f28e565a 5584 mon_addrs_size = len + 1;
7ef3214a 5585 buf += len;
a725f65e 5586
dc79b113 5587 ret = -EINVAL;
f28e565a
AE
5588 options = dup_token(&buf, NULL);
5589 if (!options)
dc79b113 5590 return -ENOMEM;
4fb5d671
AE
5591 if (!*options) {
5592 rbd_warn(NULL, "no options provided");
5593 goto out_err;
5594 }
e28fff26 5595
859c31df
AE
5596 spec = rbd_spec_alloc();
5597 if (!spec)
f28e565a 5598 goto out_mem;
859c31df
AE
5599
5600 spec->pool_name = dup_token(&buf, NULL);
5601 if (!spec->pool_name)
5602 goto out_mem;
4fb5d671
AE
5603 if (!*spec->pool_name) {
5604 rbd_warn(NULL, "no pool name provided");
5605 goto out_err;
5606 }
e28fff26 5607
69e7a02f 5608 spec->image_name = dup_token(&buf, NULL);
859c31df 5609 if (!spec->image_name)
f28e565a 5610 goto out_mem;
4fb5d671
AE
5611 if (!*spec->image_name) {
5612 rbd_warn(NULL, "no image name provided");
5613 goto out_err;
5614 }
d4b125e9 5615
f28e565a
AE
5616 /*
5617 * Snapshot name is optional; default is to use "-"
5618 * (indicating the head/no snapshot).
5619 */
3feeb894 5620 len = next_token(&buf);
820a5f3e 5621 if (!len) {
3feeb894
AE
5622 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
5623 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
f28e565a 5624 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
dc79b113 5625 ret = -ENAMETOOLONG;
f28e565a 5626 goto out_err;
849b4260 5627 }
ecb4dc22
AE
5628 snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
5629 if (!snap_name)
f28e565a 5630 goto out_mem;
ecb4dc22
AE
5631 *(snap_name + len) = '\0';
5632 spec->snap_name = snap_name;
e5c35534 5633
0ddebc0c 5634 /* Initialize all rbd options to the defaults */
e28fff26 5635
4e9afeba
AE
5636 rbd_opts = kzalloc(sizeof (*rbd_opts), GFP_KERNEL);
5637 if (!rbd_opts)
5638 goto out_mem;
5639
5640 rbd_opts->read_only = RBD_READ_ONLY_DEFAULT;
b5584180 5641 rbd_opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
80de1912 5642 rbd_opts->lock_on_read = RBD_LOCK_ON_READ_DEFAULT;
e010dd0a 5643 rbd_opts->exclusive = RBD_EXCLUSIVE_DEFAULT;
d22f76e7 5644
859c31df 5645 copts = ceph_parse_options(options, mon_addrs,
0ddebc0c 5646 mon_addrs + mon_addrs_size - 1,
4e9afeba 5647 parse_rbd_opts_token, rbd_opts);
859c31df
AE
5648 if (IS_ERR(copts)) {
5649 ret = PTR_ERR(copts);
dc79b113
AE
5650 goto out_err;
5651 }
859c31df
AE
5652 kfree(options);
5653
5654 *ceph_opts = copts;
4e9afeba 5655 *opts = rbd_opts;
859c31df 5656 *rbd_spec = spec;
0ddebc0c 5657
dc79b113 5658 return 0;
f28e565a 5659out_mem:
dc79b113 5660 ret = -ENOMEM;
d22f76e7 5661out_err:
859c31df
AE
5662 kfree(rbd_opts);
5663 rbd_spec_put(spec);
f28e565a 5664 kfree(options);
d22f76e7 5665
dc79b113 5666 return ret;
a725f65e
AE
5667}
5668
30ba1f02
ID
5669/*
5670 * Return pool id (>= 0) or a negative error code.
5671 */
5672static int rbd_add_get_pool_id(struct rbd_client *rbdc, const char *pool_name)
5673{
a319bf56 5674 struct ceph_options *opts = rbdc->client->options;
30ba1f02 5675 u64 newest_epoch;
30ba1f02
ID
5676 int tries = 0;
5677 int ret;
5678
5679again:
5680 ret = ceph_pg_poolid_by_name(rbdc->client->osdc.osdmap, pool_name);
5681 if (ret == -ENOENT && tries++ < 1) {
d0b19705
ID
5682 ret = ceph_monc_get_version(&rbdc->client->monc, "osdmap",
5683 &newest_epoch);
30ba1f02
ID
5684 if (ret < 0)
5685 return ret;
5686
5687 if (rbdc->client->osdc.osdmap->epoch < newest_epoch) {
7cca78c9 5688 ceph_osdc_maybe_request_map(&rbdc->client->osdc);
30ba1f02 5689 (void) ceph_monc_wait_osdmap(&rbdc->client->monc,
a319bf56
ID
5690 newest_epoch,
5691 opts->mount_timeout);
30ba1f02
ID
5692 goto again;
5693 } else {
5694 /* the osdmap we have is new enough */
5695 return -ENOENT;
5696 }
5697 }
5698
5699 return ret;
5700}
5701
e010dd0a
ID
5702static void rbd_dev_image_unlock(struct rbd_device *rbd_dev)
5703{
5704 down_write(&rbd_dev->lock_rwsem);
5705 if (__rbd_is_lock_owner(rbd_dev))
5706 rbd_unlock(rbd_dev);
5707 up_write(&rbd_dev->lock_rwsem);
5708}
5709
5710static int rbd_add_acquire_lock(struct rbd_device *rbd_dev)
5711{
5712 if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK)) {
5713 rbd_warn(rbd_dev, "exclusive-lock feature is not enabled");
5714 return -EINVAL;
5715 }
5716
5717 /* FIXME: "rbd map --exclusive" should be in interruptible */
5718 down_read(&rbd_dev->lock_rwsem);
5719 rbd_wait_state_locked(rbd_dev);
5720 up_read(&rbd_dev->lock_rwsem);
5721 if (test_bit(RBD_DEV_FLAG_BLACKLISTED, &rbd_dev->flags)) {
5722 rbd_warn(rbd_dev, "failed to acquire exclusive lock");
5723 return -EROFS;
5724 }
5725
5726 return 0;
5727}
5728
589d30e0
AE
5729/*
5730 * An rbd format 2 image has a unique identifier, distinct from the
5731 * name given to it by the user. Internally, that identifier is
5732 * what's used to specify the names of objects related to the image.
5733 *
5734 * A special "rbd id" object is used to map an rbd image name to its
5735 * id. If that object doesn't exist, then there is no v2 rbd image
5736 * with the supplied name.
5737 *
5738 * This function will record the given rbd_dev's image_id field if
5739 * it can be determined, and in that case will return 0. If any
5740 * errors occur a negative errno will be returned and the rbd_dev's
5741 * image_id field will be unchanged (and should be NULL).
5742 */
5743static int rbd_dev_image_id(struct rbd_device *rbd_dev)
5744{
5745 int ret;
5746 size_t size;
ecd4a68a 5747 CEPH_DEFINE_OID_ONSTACK(oid);
589d30e0 5748 void *response;
c0fba368 5749 char *image_id;
2f82ee54 5750
2c0d0a10
AE
5751 /*
5752 * When probing a parent image, the image id is already
5753 * known (and the image name likely is not). There's no
c0fba368
AE
5754 * need to fetch the image id again in this case. We
5755 * do still need to set the image format though.
2c0d0a10 5756 */
c0fba368
AE
5757 if (rbd_dev->spec->image_id) {
5758 rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
5759
2c0d0a10 5760 return 0;
c0fba368 5761 }
2c0d0a10 5762
589d30e0
AE
5763 /*
5764 * First, see if the format 2 image id file exists, and if
5765 * so, get the image's persistent id from it.
5766 */
ecd4a68a
ID
5767 ret = ceph_oid_aprintf(&oid, GFP_KERNEL, "%s%s", RBD_ID_PREFIX,
5768 rbd_dev->spec->image_name);
5769 if (ret)
5770 return ret;
5771
5772 dout("rbd id object name is %s\n", oid.name);
589d30e0
AE
5773
5774 /* Response will be an encoded string, which includes a length */
5775
5776 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
5777 response = kzalloc(size, GFP_NOIO);
5778 if (!response) {
5779 ret = -ENOMEM;
5780 goto out;
5781 }
5782
c0fba368
AE
5783 /* If it doesn't exist we'll assume it's a format 1 image */
5784
ecd4a68a
ID
5785 ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
5786 "get_id", NULL, 0,
5787 response, RBD_IMAGE_ID_LEN_MAX);
36be9a76 5788 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
c0fba368
AE
5789 if (ret == -ENOENT) {
5790 image_id = kstrdup("", GFP_KERNEL);
5791 ret = image_id ? 0 : -ENOMEM;
5792 if (!ret)
5793 rbd_dev->image_format = 1;
7dd440c9 5794 } else if (ret >= 0) {
c0fba368
AE
5795 void *p = response;
5796
5797 image_id = ceph_extract_encoded_string(&p, p + ret,
979ed480 5798 NULL, GFP_NOIO);
461f758a 5799 ret = PTR_ERR_OR_ZERO(image_id);
c0fba368
AE
5800 if (!ret)
5801 rbd_dev->image_format = 2;
c0fba368
AE
5802 }
5803
5804 if (!ret) {
5805 rbd_dev->spec->image_id = image_id;
5806 dout("image_id is %s\n", image_id);
589d30e0
AE
5807 }
5808out:
5809 kfree(response);
ecd4a68a 5810 ceph_oid_destroy(&oid);
589d30e0
AE
5811 return ret;
5812}
5813
3abef3b3
AE
5814/*
5815 * Undo whatever state changes are made by v1 or v2 header info
5816 * call.
5817 */
6fd48b3b
AE
5818static void rbd_dev_unprobe(struct rbd_device *rbd_dev)
5819{
5820 struct rbd_image_header *header;
5821
e69b8d41 5822 rbd_dev_parent_put(rbd_dev);
6fd48b3b
AE
5823
5824 /* Free dynamic fields from the header, then zero it out */
5825
5826 header = &rbd_dev->header;
812164f8 5827 ceph_put_snap_context(header->snapc);
6fd48b3b
AE
5828 kfree(header->snap_sizes);
5829 kfree(header->snap_names);
5830 kfree(header->object_prefix);
5831 memset(header, 0, sizeof (*header));
5832}
5833
2df3fac7 5834static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev)
a30b71b9
AE
5835{
5836 int ret;
a30b71b9 5837
1e130199 5838 ret = rbd_dev_v2_object_prefix(rbd_dev);
57385b51 5839 if (ret)
b1b5402a
AE
5840 goto out_err;
5841
2df3fac7
AE
5842 /*
5843 * Get the and check features for the image. Currently the
5844 * features are assumed to never change.
5845 */
b1b5402a 5846 ret = rbd_dev_v2_features(rbd_dev);
57385b51 5847 if (ret)
9d475de5 5848 goto out_err;
35d489f9 5849
cc070d59
AE
5850 /* If the image supports fancy striping, get its parameters */
5851
5852 if (rbd_dev->header.features & RBD_FEATURE_STRIPINGV2) {
5853 ret = rbd_dev_v2_striping_info(rbd_dev);
5854 if (ret < 0)
5855 goto out_err;
5856 }
a30b71b9 5857
7e97332e
ID
5858 if (rbd_dev->header.features & RBD_FEATURE_DATA_POOL) {
5859 ret = rbd_dev_v2_data_pool(rbd_dev);
5860 if (ret)
5861 goto out_err;
5862 }
5863
263423f8 5864 rbd_init_layout(rbd_dev);
35152979 5865 return 0;
263423f8 5866
9d475de5 5867out_err:
642a2537 5868 rbd_dev->header.features = 0;
1e130199
AE
5869 kfree(rbd_dev->header.object_prefix);
5870 rbd_dev->header.object_prefix = NULL;
9d475de5 5871 return ret;
a30b71b9
AE
5872}
5873
6d69bb53
ID
5874/*
5875 * @depth is rbd_dev_image_probe() -> rbd_dev_probe_parent() ->
5876 * rbd_dev_image_probe() recursion depth, which means it's also the
5877 * length of the already discovered part of the parent chain.
5878 */
5879static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
83a06263 5880{
2f82ee54 5881 struct rbd_device *parent = NULL;
124afba2
AE
5882 int ret;
5883
5884 if (!rbd_dev->parent_spec)
5885 return 0;
124afba2 5886
6d69bb53
ID
5887 if (++depth > RBD_MAX_PARENT_CHAIN_LEN) {
5888 pr_info("parent chain is too long (%d)\n", depth);
5889 ret = -EINVAL;
5890 goto out_err;
5891 }
5892
1643dfa4 5893 parent = __rbd_dev_create(rbd_dev->rbd_client, rbd_dev->parent_spec);
1f2c6651
ID
5894 if (!parent) {
5895 ret = -ENOMEM;
124afba2 5896 goto out_err;
1f2c6651
ID
5897 }
5898
5899 /*
5900 * Images related by parent/child relationships always share
5901 * rbd_client and spec/parent_spec, so bump their refcounts.
5902 */
5903 __rbd_get_client(rbd_dev->rbd_client);
5904 rbd_spec_get(rbd_dev->parent_spec);
124afba2 5905
6d69bb53 5906 ret = rbd_dev_image_probe(parent, depth);
124afba2
AE
5907 if (ret < 0)
5908 goto out_err;
1f2c6651 5909
124afba2 5910 rbd_dev->parent = parent;
a2acd00e 5911 atomic_set(&rbd_dev->parent_ref, 1);
124afba2 5912 return 0;
1f2c6651 5913
124afba2 5914out_err:
1f2c6651 5915 rbd_dev_unparent(rbd_dev);
1761b229 5916 rbd_dev_destroy(parent);
124afba2
AE
5917 return ret;
5918}
5919
5769ed0c
ID
5920static void rbd_dev_device_release(struct rbd_device *rbd_dev)
5921{
5922 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
5923 rbd_dev_mapping_clear(rbd_dev);
5924 rbd_free_disk(rbd_dev);
5925 if (!single_major)
5926 unregister_blkdev(rbd_dev->major, rbd_dev->name);
5927}
5928
811c6688
ID
5929/*
5930 * rbd_dev->header_rwsem must be locked for write and will be unlocked
5931 * upon return.
5932 */
200a6a8b 5933static int rbd_dev_device_setup(struct rbd_device *rbd_dev)
124afba2 5934{
83a06263 5935 int ret;
d1cf5788 5936
9b60e70b 5937 /* Record our major and minor device numbers. */
83a06263 5938
9b60e70b
ID
5939 if (!single_major) {
5940 ret = register_blkdev(0, rbd_dev->name);
5941 if (ret < 0)
1643dfa4 5942 goto err_out_unlock;
9b60e70b
ID
5943
5944 rbd_dev->major = ret;
5945 rbd_dev->minor = 0;
5946 } else {
5947 rbd_dev->major = rbd_major;
5948 rbd_dev->minor = rbd_dev_id_to_minor(rbd_dev->dev_id);
5949 }
83a06263
AE
5950
5951 /* Set up the blkdev mapping. */
5952
5953 ret = rbd_init_disk(rbd_dev);
5954 if (ret)
5955 goto err_out_blkdev;
5956
f35a4dee 5957 ret = rbd_dev_mapping_set(rbd_dev);
83a06263
AE
5958 if (ret)
5959 goto err_out_disk;
bc1ecc65 5960
f35a4dee 5961 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
9568c93e 5962 set_disk_ro(rbd_dev->disk, rbd_dev->opts->read_only);
f35a4dee 5963
5769ed0c 5964 ret = dev_set_name(&rbd_dev->dev, "%d", rbd_dev->dev_id);
f35a4dee 5965 if (ret)
f5ee37bd 5966 goto err_out_mapping;
83a06263 5967
129b79d4 5968 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
811c6688 5969 up_write(&rbd_dev->header_rwsem);
5769ed0c 5970 return 0;
2f82ee54 5971
f35a4dee
AE
5972err_out_mapping:
5973 rbd_dev_mapping_clear(rbd_dev);
83a06263
AE
5974err_out_disk:
5975 rbd_free_disk(rbd_dev);
5976err_out_blkdev:
9b60e70b
ID
5977 if (!single_major)
5978 unregister_blkdev(rbd_dev->major, rbd_dev->name);
811c6688
ID
5979err_out_unlock:
5980 up_write(&rbd_dev->header_rwsem);
83a06263
AE
5981 return ret;
5982}
5983
332bb12d
AE
5984static int rbd_dev_header_name(struct rbd_device *rbd_dev)
5985{
5986 struct rbd_spec *spec = rbd_dev->spec;
c41d13a3 5987 int ret;
332bb12d
AE
5988
5989 /* Record the header object name for this rbd image. */
5990
5991 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
332bb12d 5992 if (rbd_dev->image_format == 1)
c41d13a3
ID
5993 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
5994 spec->image_name, RBD_SUFFIX);
332bb12d 5995 else
c41d13a3
ID
5996 ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
5997 RBD_HEADER_PREFIX, spec->image_id);
332bb12d 5998
c41d13a3 5999 return ret;
332bb12d
AE
6000}
6001
200a6a8b
AE
6002static void rbd_dev_image_release(struct rbd_device *rbd_dev)
6003{
6fd48b3b 6004 rbd_dev_unprobe(rbd_dev);
fd22aef8
ID
6005 if (rbd_dev->opts)
6006 rbd_unregister_watch(rbd_dev);
6fd48b3b
AE
6007 rbd_dev->image_format = 0;
6008 kfree(rbd_dev->spec->image_id);
6009 rbd_dev->spec->image_id = NULL;
200a6a8b
AE
6010}
6011
a30b71b9
AE
6012/*
6013 * Probe for the existence of the header object for the given rbd
1f3ef788
AE
6014 * device. If this image is the one being mapped (i.e., not a
6015 * parent), initiate a watch on its header object before using that
6016 * object to get detailed information about the rbd image.
a30b71b9 6017 */
6d69bb53 6018static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
a30b71b9
AE
6019{
6020 int ret;
6021
6022 /*
3abef3b3
AE
6023 * Get the id from the image id object. Unless there's an
6024 * error, rbd_dev->spec->image_id will be filled in with
6025 * a dynamically-allocated string, and rbd_dev->image_format
6026 * will be set to either 1 or 2.
a30b71b9
AE
6027 */
6028 ret = rbd_dev_image_id(rbd_dev);
6029 if (ret)
c0fba368 6030 return ret;
c0fba368 6031
332bb12d
AE
6032 ret = rbd_dev_header_name(rbd_dev);
6033 if (ret)
6034 goto err_out_format;
6035
6d69bb53 6036 if (!depth) {
99d16943 6037 ret = rbd_register_watch(rbd_dev);
1fe48023
ID
6038 if (ret) {
6039 if (ret == -ENOENT)
6040 pr_info("image %s/%s does not exist\n",
6041 rbd_dev->spec->pool_name,
6042 rbd_dev->spec->image_name);
c41d13a3 6043 goto err_out_format;
1fe48023 6044 }
1f3ef788 6045 }
b644de2b 6046
a720ae09 6047 ret = rbd_dev_header_info(rbd_dev);
5655c4d9 6048 if (ret)
b644de2b 6049 goto err_out_watch;
83a06263 6050
04077599
ID
6051 /*
6052 * If this image is the one being mapped, we have pool name and
6053 * id, image name and id, and snap name - need to fill snap id.
6054 * Otherwise this is a parent image, identified by pool, image
6055 * and snap ids - need to fill in names for those ids.
6056 */
6d69bb53 6057 if (!depth)
04077599
ID
6058 ret = rbd_spec_fill_snap_id(rbd_dev);
6059 else
6060 ret = rbd_spec_fill_names(rbd_dev);
1fe48023
ID
6061 if (ret) {
6062 if (ret == -ENOENT)
6063 pr_info("snap %s/%s@%s does not exist\n",
6064 rbd_dev->spec->pool_name,
6065 rbd_dev->spec->image_name,
6066 rbd_dev->spec->snap_name);
33dca39f 6067 goto err_out_probe;
1fe48023 6068 }
9bb81c9b 6069
e8f59b59
ID
6070 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
6071 ret = rbd_dev_v2_parent_info(rbd_dev);
6072 if (ret)
6073 goto err_out_probe;
6074
6075 /*
6076 * Need to warn users if this image is the one being
6077 * mapped and has a parent.
6078 */
6d69bb53 6079 if (!depth && rbd_dev->parent_spec)
e8f59b59
ID
6080 rbd_warn(rbd_dev,
6081 "WARNING: kernel layering is EXPERIMENTAL!");
6082 }
6083
6d69bb53 6084 ret = rbd_dev_probe_parent(rbd_dev, depth);
30d60ba2
AE
6085 if (ret)
6086 goto err_out_probe;
6087
6088 dout("discovered format %u image, header name is %s\n",
c41d13a3 6089 rbd_dev->image_format, rbd_dev->header_oid.name);
30d60ba2 6090 return 0;
e8f59b59 6091
6fd48b3b
AE
6092err_out_probe:
6093 rbd_dev_unprobe(rbd_dev);
b644de2b 6094err_out_watch:
6d69bb53 6095 if (!depth)
99d16943 6096 rbd_unregister_watch(rbd_dev);
332bb12d
AE
6097err_out_format:
6098 rbd_dev->image_format = 0;
5655c4d9
AE
6099 kfree(rbd_dev->spec->image_id);
6100 rbd_dev->spec->image_id = NULL;
a30b71b9
AE
6101 return ret;
6102}
6103
9b60e70b
ID
6104static ssize_t do_rbd_add(struct bus_type *bus,
6105 const char *buf,
6106 size_t count)
602adf40 6107{
cb8627c7 6108 struct rbd_device *rbd_dev = NULL;
dc79b113 6109 struct ceph_options *ceph_opts = NULL;
4e9afeba 6110 struct rbd_options *rbd_opts = NULL;
859c31df 6111 struct rbd_spec *spec = NULL;
9d3997fd 6112 struct rbd_client *rbdc;
b51c83c2 6113 int rc;
602adf40
YS
6114
6115 if (!try_module_get(THIS_MODULE))
6116 return -ENODEV;
6117
602adf40 6118 /* parse add command */
859c31df 6119 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
dc79b113 6120 if (rc < 0)
dd5ac32d 6121 goto out;
78cea76e 6122
9d3997fd
AE
6123 rbdc = rbd_get_client(ceph_opts);
6124 if (IS_ERR(rbdc)) {
6125 rc = PTR_ERR(rbdc);
0ddebc0c 6126 goto err_out_args;
9d3997fd 6127 }
602adf40 6128
602adf40 6129 /* pick the pool */
30ba1f02 6130 rc = rbd_add_get_pool_id(rbdc, spec->pool_name);
1fe48023
ID
6131 if (rc < 0) {
6132 if (rc == -ENOENT)
6133 pr_info("pool %s does not exist\n", spec->pool_name);
602adf40 6134 goto err_out_client;
1fe48023 6135 }
c0cd10db 6136 spec->pool_id = (u64)rc;
859c31df 6137
d147543d 6138 rbd_dev = rbd_dev_create(rbdc, spec, rbd_opts);
b51c83c2
ID
6139 if (!rbd_dev) {
6140 rc = -ENOMEM;
bd4ba655 6141 goto err_out_client;
b51c83c2 6142 }
c53d5893
AE
6143 rbdc = NULL; /* rbd_dev now owns this */
6144 spec = NULL; /* rbd_dev now owns this */
d147543d 6145 rbd_opts = NULL; /* rbd_dev now owns this */
602adf40 6146
0d6d1e9c
MC
6147 rbd_dev->config_info = kstrdup(buf, GFP_KERNEL);
6148 if (!rbd_dev->config_info) {
6149 rc = -ENOMEM;
6150 goto err_out_rbd_dev;
6151 }
6152
811c6688 6153 down_write(&rbd_dev->header_rwsem);
6d69bb53 6154 rc = rbd_dev_image_probe(rbd_dev, 0);
0d6d1e9c
MC
6155 if (rc < 0) {
6156 up_write(&rbd_dev->header_rwsem);
c53d5893 6157 goto err_out_rbd_dev;
0d6d1e9c 6158 }
05fd6f6f 6159
7ce4eef7 6160 /* If we are mapping a snapshot it must be marked read-only */
7ce4eef7 6161 if (rbd_dev->spec->snap_id != CEPH_NOSNAP)
9568c93e 6162 rbd_dev->opts->read_only = true;
7ce4eef7 6163
b536f69a 6164 rc = rbd_dev_device_setup(rbd_dev);
fd22aef8 6165 if (rc)
8b679ec5 6166 goto err_out_image_probe;
3abef3b3 6167
e010dd0a
ID
6168 if (rbd_dev->opts->exclusive) {
6169 rc = rbd_add_acquire_lock(rbd_dev);
6170 if (rc)
6171 goto err_out_device_setup;
3abef3b3
AE
6172 }
6173
5769ed0c
ID
6174 /* Everything's ready. Announce the disk to the world. */
6175
6176 rc = device_add(&rbd_dev->dev);
6177 if (rc)
e010dd0a 6178 goto err_out_image_lock;
5769ed0c
ID
6179
6180 add_disk(rbd_dev->disk);
6181 /* see rbd_init_disk() */
6182 blk_put_queue(rbd_dev->disk->queue);
6183
6184 spin_lock(&rbd_dev_list_lock);
6185 list_add_tail(&rbd_dev->node, &rbd_dev_list);
6186 spin_unlock(&rbd_dev_list_lock);
6187
6188 pr_info("%s: capacity %llu features 0x%llx\n", rbd_dev->disk->disk_name,
6189 (unsigned long long)get_capacity(rbd_dev->disk) << SECTOR_SHIFT,
6190 rbd_dev->header.features);
dd5ac32d
ID
6191 rc = count;
6192out:
6193 module_put(THIS_MODULE);
6194 return rc;
b536f69a 6195
e010dd0a
ID
6196err_out_image_lock:
6197 rbd_dev_image_unlock(rbd_dev);
5769ed0c
ID
6198err_out_device_setup:
6199 rbd_dev_device_release(rbd_dev);
8b679ec5
ID
6200err_out_image_probe:
6201 rbd_dev_image_release(rbd_dev);
c53d5893
AE
6202err_out_rbd_dev:
6203 rbd_dev_destroy(rbd_dev);
bd4ba655 6204err_out_client:
9d3997fd 6205 rbd_put_client(rbdc);
0ddebc0c 6206err_out_args:
859c31df 6207 rbd_spec_put(spec);
d147543d 6208 kfree(rbd_opts);
dd5ac32d 6209 goto out;
602adf40
YS
6210}
6211
9b60e70b
ID
6212static ssize_t rbd_add(struct bus_type *bus,
6213 const char *buf,
6214 size_t count)
6215{
6216 if (single_major)
6217 return -EINVAL;
6218
6219 return do_rbd_add(bus, buf, count);
6220}
6221
6222static ssize_t rbd_add_single_major(struct bus_type *bus,
6223 const char *buf,
6224 size_t count)
6225{
6226 return do_rbd_add(bus, buf, count);
6227}
6228
05a46afd
AE
6229static void rbd_dev_remove_parent(struct rbd_device *rbd_dev)
6230{
ad945fc1 6231 while (rbd_dev->parent) {
05a46afd
AE
6232 struct rbd_device *first = rbd_dev;
6233 struct rbd_device *second = first->parent;
6234 struct rbd_device *third;
6235
6236 /*
6237 * Follow to the parent with no grandparent and
6238 * remove it.
6239 */
6240 while (second && (third = second->parent)) {
6241 first = second;
6242 second = third;
6243 }
ad945fc1 6244 rbd_assert(second);
8ad42cd0 6245 rbd_dev_image_release(second);
8b679ec5 6246 rbd_dev_destroy(second);
ad945fc1
AE
6247 first->parent = NULL;
6248 first->parent_overlap = 0;
6249
6250 rbd_assert(first->parent_spec);
05a46afd
AE
6251 rbd_spec_put(first->parent_spec);
6252 first->parent_spec = NULL;
05a46afd
AE
6253 }
6254}
6255
9b60e70b
ID
6256static ssize_t do_rbd_remove(struct bus_type *bus,
6257 const char *buf,
6258 size_t count)
602adf40
YS
6259{
6260 struct rbd_device *rbd_dev = NULL;
751cc0e3
AE
6261 struct list_head *tmp;
6262 int dev_id;
0276dca6 6263 char opt_buf[6];
82a442d2 6264 bool already = false;
0276dca6 6265 bool force = false;
0d8189e1 6266 int ret;
602adf40 6267
0276dca6
MC
6268 dev_id = -1;
6269 opt_buf[0] = '\0';
6270 sscanf(buf, "%d %5s", &dev_id, opt_buf);
6271 if (dev_id < 0) {
6272 pr_err("dev_id out of range\n");
602adf40 6273 return -EINVAL;
0276dca6
MC
6274 }
6275 if (opt_buf[0] != '\0') {
6276 if (!strcmp(opt_buf, "force")) {
6277 force = true;
6278 } else {
6279 pr_err("bad remove option at '%s'\n", opt_buf);
6280 return -EINVAL;
6281 }
6282 }
602adf40 6283
751cc0e3
AE
6284 ret = -ENOENT;
6285 spin_lock(&rbd_dev_list_lock);
6286 list_for_each(tmp, &rbd_dev_list) {
6287 rbd_dev = list_entry(tmp, struct rbd_device, node);
6288 if (rbd_dev->dev_id == dev_id) {
6289 ret = 0;
6290 break;
6291 }
42382b70 6292 }
751cc0e3
AE
6293 if (!ret) {
6294 spin_lock_irq(&rbd_dev->lock);
0276dca6 6295 if (rbd_dev->open_count && !force)
751cc0e3
AE
6296 ret = -EBUSY;
6297 else
82a442d2
AE
6298 already = test_and_set_bit(RBD_DEV_FLAG_REMOVING,
6299 &rbd_dev->flags);
751cc0e3
AE
6300 spin_unlock_irq(&rbd_dev->lock);
6301 }
6302 spin_unlock(&rbd_dev_list_lock);
82a442d2 6303 if (ret < 0 || already)
1ba0f1e7 6304 return ret;
751cc0e3 6305
0276dca6
MC
6306 if (force) {
6307 /*
6308 * Prevent new IO from being queued and wait for existing
6309 * IO to complete/fail.
6310 */
6311 blk_mq_freeze_queue(rbd_dev->disk->queue);
6312 blk_set_queue_dying(rbd_dev->disk->queue);
6313 }
6314
5769ed0c
ID
6315 del_gendisk(rbd_dev->disk);
6316 spin_lock(&rbd_dev_list_lock);
6317 list_del_init(&rbd_dev->node);
6318 spin_unlock(&rbd_dev_list_lock);
6319 device_del(&rbd_dev->dev);
fca27065 6320
e010dd0a 6321 rbd_dev_image_unlock(rbd_dev);
dd5ac32d 6322 rbd_dev_device_release(rbd_dev);
8ad42cd0 6323 rbd_dev_image_release(rbd_dev);
8b679ec5 6324 rbd_dev_destroy(rbd_dev);
1ba0f1e7 6325 return count;
602adf40
YS
6326}
6327
9b60e70b
ID
6328static ssize_t rbd_remove(struct bus_type *bus,
6329 const char *buf,
6330 size_t count)
6331{
6332 if (single_major)
6333 return -EINVAL;
6334
6335 return do_rbd_remove(bus, buf, count);
6336}
6337
6338static ssize_t rbd_remove_single_major(struct bus_type *bus,
6339 const char *buf,
6340 size_t count)
6341{
6342 return do_rbd_remove(bus, buf, count);
6343}
6344
602adf40
YS
6345/*
6346 * create control files in sysfs
dfc5606d 6347 * /sys/bus/rbd/...
602adf40
YS
6348 */
6349static int rbd_sysfs_init(void)
6350{
dfc5606d 6351 int ret;
602adf40 6352
fed4c143 6353 ret = device_register(&rbd_root_dev);
21079786 6354 if (ret < 0)
dfc5606d 6355 return ret;
602adf40 6356
fed4c143
AE
6357 ret = bus_register(&rbd_bus_type);
6358 if (ret < 0)
6359 device_unregister(&rbd_root_dev);
602adf40 6360
602adf40
YS
6361 return ret;
6362}
6363
6364static void rbd_sysfs_cleanup(void)
6365{
dfc5606d 6366 bus_unregister(&rbd_bus_type);
fed4c143 6367 device_unregister(&rbd_root_dev);
602adf40
YS
6368}
6369
1c2a9dfe
AE
6370static int rbd_slab_init(void)
6371{
6372 rbd_assert(!rbd_img_request_cache);
03d94406 6373 rbd_img_request_cache = KMEM_CACHE(rbd_img_request, 0);
868311b1
AE
6374 if (!rbd_img_request_cache)
6375 return -ENOMEM;
6376
6377 rbd_assert(!rbd_obj_request_cache);
03d94406 6378 rbd_obj_request_cache = KMEM_CACHE(rbd_obj_request, 0);
78c2a44a
AE
6379 if (!rbd_obj_request_cache)
6380 goto out_err;
6381
f856dc36
N
6382 rbd_assert(!rbd_bio_clone);
6383 rbd_bio_clone = bioset_create(BIO_POOL_SIZE, 0, 0);
6384 if (!rbd_bio_clone)
6385 goto out_err_clone;
6386
6c696d85 6387 return 0;
1c2a9dfe 6388
f856dc36
N
6389out_err_clone:
6390 kmem_cache_destroy(rbd_obj_request_cache);
6391 rbd_obj_request_cache = NULL;
6c696d85 6392out_err:
868311b1
AE
6393 kmem_cache_destroy(rbd_img_request_cache);
6394 rbd_img_request_cache = NULL;
1c2a9dfe
AE
6395 return -ENOMEM;
6396}
6397
6398static void rbd_slab_exit(void)
6399{
868311b1
AE
6400 rbd_assert(rbd_obj_request_cache);
6401 kmem_cache_destroy(rbd_obj_request_cache);
6402 rbd_obj_request_cache = NULL;
6403
1c2a9dfe
AE
6404 rbd_assert(rbd_img_request_cache);
6405 kmem_cache_destroy(rbd_img_request_cache);
6406 rbd_img_request_cache = NULL;
f856dc36
N
6407
6408 rbd_assert(rbd_bio_clone);
6409 bioset_free(rbd_bio_clone);
6410 rbd_bio_clone = NULL;
1c2a9dfe
AE
6411}
6412
cc344fa1 6413static int __init rbd_init(void)
602adf40
YS
6414{
6415 int rc;
6416
1e32d34c
AE
6417 if (!libceph_compatible(NULL)) {
6418 rbd_warn(NULL, "libceph incompatibility (quitting)");
1e32d34c
AE
6419 return -EINVAL;
6420 }
e1b4d96d 6421
1c2a9dfe 6422 rc = rbd_slab_init();
602adf40
YS
6423 if (rc)
6424 return rc;
e1b4d96d 6425
f5ee37bd
ID
6426 /*
6427 * The number of active work items is limited by the number of
f77303bd 6428 * rbd devices * queue depth, so leave @max_active at default.
f5ee37bd
ID
6429 */
6430 rbd_wq = alloc_workqueue(RBD_DRV_NAME, WQ_MEM_RECLAIM, 0);
6431 if (!rbd_wq) {
6432 rc = -ENOMEM;
6433 goto err_out_slab;
6434 }
6435
9b60e70b
ID
6436 if (single_major) {
6437 rbd_major = register_blkdev(0, RBD_DRV_NAME);
6438 if (rbd_major < 0) {
6439 rc = rbd_major;
f5ee37bd 6440 goto err_out_wq;
9b60e70b
ID
6441 }
6442 }
6443
1c2a9dfe
AE
6444 rc = rbd_sysfs_init();
6445 if (rc)
9b60e70b
ID
6446 goto err_out_blkdev;
6447
6448 if (single_major)
6449 pr_info("loaded (major %d)\n", rbd_major);
6450 else
6451 pr_info("loaded\n");
1c2a9dfe 6452
e1b4d96d
ID
6453 return 0;
6454
9b60e70b
ID
6455err_out_blkdev:
6456 if (single_major)
6457 unregister_blkdev(rbd_major, RBD_DRV_NAME);
f5ee37bd
ID
6458err_out_wq:
6459 destroy_workqueue(rbd_wq);
e1b4d96d
ID
6460err_out_slab:
6461 rbd_slab_exit();
1c2a9dfe 6462 return rc;
602adf40
YS
6463}
6464
cc344fa1 6465static void __exit rbd_exit(void)
602adf40 6466{
ffe312cf 6467 ida_destroy(&rbd_dev_id_ida);
602adf40 6468 rbd_sysfs_cleanup();
9b60e70b
ID
6469 if (single_major)
6470 unregister_blkdev(rbd_major, RBD_DRV_NAME);
f5ee37bd 6471 destroy_workqueue(rbd_wq);
1c2a9dfe 6472 rbd_slab_exit();
602adf40
YS
6473}
6474
6475module_init(rbd_init);
6476module_exit(rbd_exit);
6477
d552c619 6478MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
602adf40
YS
6479MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
6480MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
602adf40
YS
6481/* following authorship retained from original osdblk.c */
6482MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
6483
90da258b 6484MODULE_DESCRIPTION("RADOS Block Device (RBD) driver");
602adf40 6485MODULE_LICENSE("GPL");