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