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