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