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