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/fs_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_fetch_add_unless(v, a: 1, u: 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_OBJECT_MAP (1ULL<<3)
119	#define RBD_FEATURE_FAST_DIFF (1ULL<<4)
120	#define RBD_FEATURE_DEEP_FLATTEN (1ULL<<5)
121	#define RBD_FEATURE_DATA_POOL (1ULL<<7)
122	#define RBD_FEATURE_OPERATIONS (1ULL<<8)
123
124	#define RBD_FEATURES_ALL (RBD_FEATURE_LAYERING | \
125	RBD_FEATURE_STRIPINGV2 | \
126	RBD_FEATURE_EXCLUSIVE_LOCK | \
127	RBD_FEATURE_OBJECT_MAP | \
128	RBD_FEATURE_FAST_DIFF | \
129	RBD_FEATURE_DEEP_FLATTEN | \
130	RBD_FEATURE_DATA_POOL | \
131	RBD_FEATURE_OPERATIONS)
132
133	/* Features supported by this (client software) implementation. */
134
135	#define RBD_FEATURES_SUPPORTED (RBD_FEATURES_ALL)
136
137	/*
138	* An RBD device name will be "rbd#", where the "rbd" comes from
139	* RBD_DRV_NAME above, and # is a unique integer identifier.
140	*/
141	#define DEV_NAME_LEN 32
142
143	/*
144	* block device image metadata (in-memory version)
145	*/
146	struct rbd_image_header {
147	/* These six fields never change for a given rbd image */
148	char *object_prefix;
149	__u8 obj_order;
150	u64 stripe_unit;
151	u64 stripe_count;
152	s64 data_pool_id;
153	u64 features; /* Might be changeable someday? */
154
155	/* The remaining fields need to be updated occasionally */
156	u64 image_size;
157	struct ceph_snap_context *snapc;
158	char *snap_names; /* format 1 only */
159	u64 *snap_sizes; /* format 1 only */
160	};
161
162	/*
163	* An rbd image specification.
164	*
165	* The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
166	* identify an image. Each rbd_dev structure includes a pointer to
167	* an rbd_spec structure that encapsulates this identity.
168	*
169	* Each of the id's in an rbd_spec has an associated name. For a
170	* user-mapped image, the names are supplied and the id's associated
171	* with them are looked up. For a layered image, a parent image is
172	* defined by the tuple, and the names are looked up.
173	*
174	* An rbd_dev structure contains a parent_spec pointer which is
175	* non-null if the image it represents is a child in a layered
176	* image. This pointer will refer to the rbd_spec structure used
177	* by the parent rbd_dev for its own identity (i.e., the structure
178	* is shared between the parent and child).
179	*
180	* Since these structures are populated once, during the discovery
181	* phase of image construction, they are effectively immutable so
182	* we make no effort to synchronize access to them.
183	*
184	* Note that code herein does not assume the image name is known (it
185	* could be a null pointer).
186	*/
187	struct rbd_spec {
188	u64 pool_id;
189	const char *pool_name;
190	const char *pool_ns; /* NULL if default, never "" */
191
192	const char *image_id;
193	const char *image_name;
194
195	u64 snap_id;
196	const char *snap_name;
197
198	struct kref kref;
199	};
200
201	/*
202	* an instance of the client. multiple devices may share an rbd client.
203	*/
204	struct rbd_client {
205	struct ceph_client *client;
206	struct kref kref;
207	struct list_head node;
208	};
209
210	struct pending_result {
211	int result; /* first nonzero result */
212	int num_pending;
213	};
214
215	struct rbd_img_request;
216
217	enum obj_request_type {
218	OBJ_REQUEST_NODATA = 1,
219	OBJ_REQUEST_BIO, /* pointer into provided bio (list) */
220	OBJ_REQUEST_BVECS, /* pointer into provided bio_vec array */
221	OBJ_REQUEST_OWN_BVECS, /* private bio_vec array, doesn't own pages */
222	};
223
224	enum obj_operation_type {
225	OBJ_OP_READ = 1,
226	OBJ_OP_WRITE,
227	OBJ_OP_DISCARD,
228	OBJ_OP_ZEROOUT,
229	};
230
231	#define RBD_OBJ_FLAG_DELETION (1U << 0)
232	#define RBD_OBJ_FLAG_COPYUP_ENABLED (1U << 1)
233	#define RBD_OBJ_FLAG_COPYUP_ZEROS (1U << 2)
234	#define RBD_OBJ_FLAG_MAY_EXIST (1U << 3)
235	#define RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT (1U << 4)
236
237	enum rbd_obj_read_state {
238	RBD_OBJ_READ_START = 1,
239	RBD_OBJ_READ_OBJECT,
240	RBD_OBJ_READ_PARENT,
241	};
242
243	/*
244	* Writes go through the following state machine to deal with
245	* layering:
246	*
247	* . . . . . RBD_OBJ_WRITE_GUARD. . . . . . . . . . . . . .
248	* . | .
249	* . v .
250	* . RBD_OBJ_WRITE_READ_FROM_PARENT. . . .
251	* . | . .
252	* . v v (deep-copyup .
253	* (image . RBD_OBJ_WRITE_COPYUP_EMPTY_SNAPC . not needed) .
254	* flattened) v | . .
255	* . v . .
256	* . . . .RBD_OBJ_WRITE_COPYUP_OPS. . . . . (copyup .
257	* | not needed) v
258	* v .
259	* done . . . . . . . . . . . . . . . . . .
260	* ^
261	* |
262	* RBD_OBJ_WRITE_FLAT
263	*
264	* Writes start in RBD_OBJ_WRITE_GUARD or _FLAT, depending on whether
265	* assert_exists guard is needed or not (in some cases it's not needed
266	* even if there is a parent).
267	*/
268	enum rbd_obj_write_state {
269	RBD_OBJ_WRITE_START = 1,
270	RBD_OBJ_WRITE_PRE_OBJECT_MAP,
271	RBD_OBJ_WRITE_OBJECT,
272	__RBD_OBJ_WRITE_COPYUP,
273	RBD_OBJ_WRITE_COPYUP,
274	RBD_OBJ_WRITE_POST_OBJECT_MAP,
275	};
276
277	enum rbd_obj_copyup_state {
278	RBD_OBJ_COPYUP_START = 1,
279	RBD_OBJ_COPYUP_READ_PARENT,
280	__RBD_OBJ_COPYUP_OBJECT_MAPS,
281	RBD_OBJ_COPYUP_OBJECT_MAPS,
282	__RBD_OBJ_COPYUP_WRITE_OBJECT,
283	RBD_OBJ_COPYUP_WRITE_OBJECT,
284	};
285
286	struct rbd_obj_request {
287	struct ceph_object_extent ex;
288	unsigned int flags; /* RBD_OBJ_FLAG_* */
289	union {
290	enum rbd_obj_read_state read_state; /* for reads */
291	enum rbd_obj_write_state write_state; /* for writes */
292	};
293
294	struct rbd_img_request *img_request;
295	struct ceph_file_extent *img_extents;
296	u32 num_img_extents;
297
298	union {
299	struct ceph_bio_iter bio_pos;
300	struct {
301	struct ceph_bvec_iter bvec_pos;
302	u32 bvec_count;
303	u32 bvec_idx;
304	};
305	};
306
307	enum rbd_obj_copyup_state copyup_state;
308	struct bio_vec *copyup_bvecs;
309	u32 copyup_bvec_count;
310
311	struct list_head osd_reqs; /* w/ r_private_item */
312
313	struct mutex state_mutex;
314	struct pending_result pending;
315	struct kref kref;
316	};
317
318	enum img_req_flags {
319	IMG_REQ_CHILD, /* initiator: block = 0, child image = 1 */
320	IMG_REQ_LAYERED, /* ENOENT handling: normal = 0, layered = 1 */
321	};
322
323	enum rbd_img_state {
324	RBD_IMG_START = 1,
325	RBD_IMG_EXCLUSIVE_LOCK,
326	__RBD_IMG_OBJECT_REQUESTS,
327	RBD_IMG_OBJECT_REQUESTS,
328	};
329
330	struct rbd_img_request {
331	struct rbd_device *rbd_dev;
332	enum obj_operation_type op_type;
333	enum obj_request_type data_type;
334	unsigned long flags;
335	enum rbd_img_state state;
336	union {
337	u64 snap_id; /* for reads */
338	struct ceph_snap_context *snapc; /* for writes */
339	};
340	struct rbd_obj_request *obj_request; /* obj req initiator */
341
342	struct list_head lock_item;
343	struct list_head object_extents; /* obj_req.ex structs */
344
345	struct mutex state_mutex;
346	struct pending_result pending;
347	struct work_struct work;
348	int work_result;
349	};
350
351	#define for_each_obj_request(ireq, oreq) \
352	list_for_each_entry(oreq, &(ireq)->object_extents, ex.oe_item)
353	#define for_each_obj_request_safe(ireq, oreq, n) \
354	list_for_each_entry_safe(oreq, n, &(ireq)->object_extents, ex.oe_item)
355
356	enum rbd_watch_state {
357	RBD_WATCH_STATE_UNREGISTERED,
358	RBD_WATCH_STATE_REGISTERED,
359	RBD_WATCH_STATE_ERROR,
360	};
361
362	enum rbd_lock_state {
363	RBD_LOCK_STATE_UNLOCKED,
364	RBD_LOCK_STATE_LOCKED,
365	RBD_LOCK_STATE_RELEASING,
366	};
367
368	/* WatchNotify::ClientId */
369	struct rbd_client_id {
370	u64 gid;
371	u64 handle;
372	};
373
374	struct rbd_mapping {
375	u64 size;
376	};
377
378	/*
379	* a single device
380	*/
381	struct rbd_device {
382	int dev_id; /* blkdev unique id */
383
384	int major; /* blkdev assigned major */
385	int minor;
386	struct gendisk *disk; /* blkdev's gendisk and rq */
387
388	u32 image_format; /* Either 1 or 2 */
389	struct rbd_client *rbd_client;
390
391	char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
392
393	spinlock_t lock; /* queue, flags, open_count */
394
395	struct rbd_image_header header;
396	unsigned long flags; /* possibly lock protected */
397	struct rbd_spec *spec;
398	struct rbd_options *opts;
399	char *config_info; /* add{,_single_major} string */
400
401	struct ceph_object_id header_oid;
402	struct ceph_object_locator header_oloc;
403
404	struct ceph_file_layout layout; /* used for all rbd requests */
405
406	struct mutex watch_mutex;
407	enum rbd_watch_state watch_state;
408	struct ceph_osd_linger_request *watch_handle;
409	u64 watch_cookie;
410	struct delayed_work watch_dwork;
411
412	struct rw_semaphore lock_rwsem;
413	enum rbd_lock_state lock_state;
414	char lock_cookie[32];
415	struct rbd_client_id owner_cid;
416	struct work_struct acquired_lock_work;
417	struct work_struct released_lock_work;
418	struct delayed_work lock_dwork;
419	struct work_struct unlock_work;
420	spinlock_t lock_lists_lock;
421	struct list_head acquiring_list;
422	struct list_head running_list;
423	struct completion acquire_wait;
424	int acquire_err;
425	struct completion releasing_wait;
426
427	spinlock_t object_map_lock;
428	u8 *object_map;
429	u64 object_map_size; /* in objects */
430	u64 object_map_flags;
431
432	struct workqueue_struct *task_wq;
433
434	struct rbd_spec *parent_spec;
435	u64 parent_overlap;
436	atomic_t parent_ref;
437	struct rbd_device *parent;
438
439	/* Block layer tags. */
440	struct blk_mq_tag_set tag_set;
441
442	/* protects updating the header */
443	struct rw_semaphore header_rwsem;
444
445	struct rbd_mapping mapping;
446
447	struct list_head node;
448
449	/* sysfs related */
450	struct device dev;
451	unsigned long open_count; /* protected by lock */
452	};
453
454	/*
455	* Flag bits for rbd_dev->flags:
456	* - REMOVING (which is coupled with rbd_dev->open_count) is protected
457	* by rbd_dev->lock
458	*/
459	enum rbd_dev_flags {
460	RBD_DEV_FLAG_EXISTS, /* rbd_dev_device_setup() ran */
461	RBD_DEV_FLAG_REMOVING, /* this mapping is being removed */
462	RBD_DEV_FLAG_READONLY, /* -o ro or snapshot */
463	};
464
465	static DEFINE_MUTEX(client_mutex); /* Serialize client creation */
466
467	static LIST_HEAD(rbd_dev_list); /* devices */
468	static DEFINE_SPINLOCK(rbd_dev_list_lock);
469
470	static LIST_HEAD(rbd_client_list); /* clients */
471	static DEFINE_SPINLOCK(rbd_client_list_lock);
472
473	/* Slab caches for frequently-allocated structures */
474
475	static struct kmem_cache *rbd_img_request_cache;
476	static struct kmem_cache *rbd_obj_request_cache;
477
478	static int rbd_major;
479	static DEFINE_IDA(rbd_dev_id_ida);
480
481	static struct workqueue_struct *rbd_wq;
482
483	static struct ceph_snap_context rbd_empty_snapc = {
484	.nref = REFCOUNT_INIT(1),
485	};
486
487	/*
488	* single-major requires >= 0.75 version of userspace rbd utility.
489	*/
490	static bool single_major = true;
491	module_param(single_major, bool, 0444);
492	MODULE_PARM_DESC(single_major, "Use a single major number for all rbd devices (default: true)");
493
494	static ssize_t add_store(const struct bus_type *bus, const char *buf, size_t count);
495	static ssize_t remove_store(const struct bus_type *bus, const char *buf,
496	size_t count);
497	static ssize_t add_single_major_store(const struct bus_type *bus, const char *buf,
498	size_t count);
499	static ssize_t remove_single_major_store(const struct bus_type *bus, const char *buf,
500	size_t count);
501	static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth);
502
503	static int rbd_dev_id_to_minor(int dev_id)
504	{
505	return dev_id << RBD_SINGLE_MAJOR_PART_SHIFT;
506	}
507
508	static int minor_to_rbd_dev_id(int minor)
509	{
510	return minor >> RBD_SINGLE_MAJOR_PART_SHIFT;
511	}
512
513	static bool rbd_is_ro(struct rbd_device *rbd_dev)
514	{
515	return test_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
516	}
517
518	static bool rbd_is_snap(struct rbd_device *rbd_dev)
519	{
520	return rbd_dev->spec->snap_id != CEPH_NOSNAP;
521	}
522
523	static bool __rbd_is_lock_owner(struct rbd_device *rbd_dev)
524	{
525	lockdep_assert_held(&rbd_dev->lock_rwsem);
526
527	return rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED ||
528	rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING;
529	}
530
531	static bool rbd_is_lock_owner(struct rbd_device *rbd_dev)
532	{
533	bool is_lock_owner;
534
535	down_read(sem: &rbd_dev->lock_rwsem);
536	is_lock_owner = __rbd_is_lock_owner(rbd_dev);
537	up_read(sem: &rbd_dev->lock_rwsem);
538	return is_lock_owner;
539	}
540
541	static ssize_t supported_features_show(const struct bus_type *bus, char *buf)
542	{
543	return sprintf(buf, fmt: "0x%llx\n", RBD_FEATURES_SUPPORTED);
544	}
545
546	static BUS_ATTR_WO(add);
547	static BUS_ATTR_WO(remove);
548	static BUS_ATTR_WO(add_single_major);
549	static BUS_ATTR_WO(remove_single_major);
550	static BUS_ATTR_RO(supported_features);
551
552	static struct attribute *rbd_bus_attrs[] = {
553	&bus_attr_add.attr,
554	&bus_attr_remove.attr,
555	&bus_attr_add_single_major.attr,
556	&bus_attr_remove_single_major.attr,
557	&bus_attr_supported_features.attr,
558	NULL,
559	};
560
561	static umode_t rbd_bus_is_visible(struct kobject *kobj,
562	struct attribute *attr, int index)
563	{
564	if (!single_major &&
565	(attr == &bus_attr_add_single_major.attr ||
566	attr == &bus_attr_remove_single_major.attr))
567	return 0;
568
569	return attr->mode;
570	}
571
572	static const struct attribute_group rbd_bus_group = {
573	.attrs = rbd_bus_attrs,
574	.is_visible = rbd_bus_is_visible,
575	};
576	__ATTRIBUTE_GROUPS(rbd_bus);
577
578	static struct bus_type rbd_bus_type = {
579	.name = "rbd",
580	.bus_groups = rbd_bus_groups,
581	};
582
583	static void rbd_root_dev_release(struct device *dev)
584	{
585	}
586
587	static struct device rbd_root_dev = {
588	.init_name = "rbd",
589	.release = rbd_root_dev_release,
590	};
591
592	static __printf(2, 3)
593	void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
594	{
595	struct va_format vaf;
596	va_list args;
597
598	va_start(args, fmt);
599	vaf.fmt = fmt;
600	vaf.va = &args;
601
602	if (!rbd_dev)
603	printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
604	else if (rbd_dev->disk)
605	printk(KERN_WARNING "%s: %s: %pV\n",
606	RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
607	else if (rbd_dev->spec && rbd_dev->spec->image_name)
608	printk(KERN_WARNING "%s: image %s: %pV\n",
609	RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
610	else if (rbd_dev->spec && rbd_dev->spec->image_id)
611	printk(KERN_WARNING "%s: id %s: %pV\n",
612	RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
613	else /* punt */
614	printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
615	RBD_DRV_NAME, rbd_dev, &vaf);
616	va_end(args);
617	}
618
619	#ifdef RBD_DEBUG
620	#define rbd_assert(expr) \
621	if (unlikely(!(expr))) { \
622	printk(KERN_ERR "\nAssertion failure in %s() " \
623	"at line %d:\n\n" \
624	"\trbd_assert(%s);\n\n", \
625	__func__, __LINE__, #expr); \
626	BUG(); \
627	}
628	#else /* !RBD_DEBUG */
629	# define rbd_assert(expr) ((void) 0)
630	#endif /* !RBD_DEBUG */
631
632	static void rbd_dev_remove_parent(struct rbd_device *rbd_dev);
633
634	static int rbd_dev_refresh(struct rbd_device *rbd_dev);
635	static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev,
636	struct rbd_image_header *header);
637	static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
638	u64 snap_id);
639	static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
640	u8 *order, u64 *snap_size);
641	static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev);
642
643	static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result);
644	static void rbd_img_handle_request(struct rbd_img_request *img_req, int result);
645
646	/*
647	* Return true if nothing else is pending.
648	*/
649	static bool pending_result_dec(struct pending_result *pending, int *result)
650	{
651	rbd_assert(pending->num_pending > 0);
652
653	if (*result && !pending->result)
654	pending->result = *result;
655	if (--pending->num_pending)
656	return false;
657
658	*result = pending->result;
659	return true;
660	}
661
662	static int rbd_open(struct gendisk *disk, blk_mode_t mode)
663	{
664	struct rbd_device *rbd_dev = disk->private_data;
665	bool removing = false;
666
667	spin_lock_irq(lock: &rbd_dev->lock);
668	if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
669	removing = true;
670	else
671	rbd_dev->open_count++;
672	spin_unlock_irq(lock: &rbd_dev->lock);
673	if (removing)
674	return -ENOENT;
675
676	(void) get_device(dev: &rbd_dev->dev);
677
678	return 0;
679	}
680
681	static void rbd_release(struct gendisk *disk)
682	{
683	struct rbd_device *rbd_dev = disk->private_data;
684	unsigned long open_count_before;
685
686	spin_lock_irq(lock: &rbd_dev->lock);
687	open_count_before = rbd_dev->open_count--;
688	spin_unlock_irq(lock: &rbd_dev->lock);
689	rbd_assert(open_count_before > 0);
690
691	put_device(dev: &rbd_dev->dev);
692	}
693
694	static const struct block_device_operations rbd_bd_ops = {
695	.owner = THIS_MODULE,
696	.open = rbd_open,
697	.release = rbd_release,
698	};
699
700	/*
701	* Initialize an rbd client instance. Success or not, this function
702	* consumes ceph_opts. Caller holds client_mutex.
703	*/
704	static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
705	{
706	struct rbd_client *rbdc;
707	int ret = -ENOMEM;
708
709	dout("%s:\n", __func__);
710	rbdc = kmalloc(size: sizeof(struct rbd_client), GFP_KERNEL);
711	if (!rbdc)
712	goto out_opt;
713
714	kref_init(kref: &rbdc->kref);
715	INIT_LIST_HEAD(list: &rbdc->node);
716
717	rbdc->client = ceph_create_client(opt: ceph_opts, private: rbdc);
718	if (IS_ERR(ptr: rbdc->client))
719	goto out_rbdc;
720	ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
721
722	ret = ceph_open_session(client: rbdc->client);
723	if (ret < 0)
724	goto out_client;
725
726	spin_lock(lock: &rbd_client_list_lock);
727	list_add_tail(new: &rbdc->node, head: &rbd_client_list);
728	spin_unlock(lock: &rbd_client_list_lock);
729
730	dout("%s: rbdc %p\n", __func__, rbdc);
731
732	return rbdc;
733	out_client:
734	ceph_destroy_client(client: rbdc->client);
735	out_rbdc:
736	kfree(objp: rbdc);
737	out_opt:
738	if (ceph_opts)
739	ceph_destroy_options(opt: ceph_opts);
740	dout("%s: error %d\n", __func__, ret);
741
742	return ERR_PTR(error: ret);
743	}
744
745	static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
746	{
747	kref_get(kref: &rbdc->kref);
748
749	return rbdc;
750	}
751
752	/*
753	* Find a ceph client with specific addr and configuration. If
754	* found, bump its reference count.
755	*/
756	static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
757	{
758	struct rbd_client *rbdc = NULL, *iter;
759
760	if (ceph_opts->flags & CEPH_OPT_NOSHARE)
761	return NULL;
762
763	spin_lock(lock: &rbd_client_list_lock);
764	list_for_each_entry(iter, &rbd_client_list, node) {
765	if (!ceph_compare_options(new_opt: ceph_opts, client: iter->client)) {
766	__rbd_get_client(rbdc: iter);
767
768	rbdc = iter;
769	break;
770	}
771	}
772	spin_unlock(lock: &rbd_client_list_lock);
773
774	return rbdc;
775	}
776
777	/*
778	* (Per device) rbd map options
779	*/
780	enum {
781	Opt_queue_depth,
782	Opt_alloc_size,
783	Opt_lock_timeout,
784	/* int args above */
785	Opt_pool_ns,
786	Opt_compression_hint,
787	/* string args above */
788	Opt_read_only,
789	Opt_read_write,
790	Opt_lock_on_read,
791	Opt_exclusive,
792	Opt_notrim,
793	};
794
795	enum {
796	Opt_compression_hint_none,
797	Opt_compression_hint_compressible,
798	Opt_compression_hint_incompressible,
799	};
800
801	static const struct constant_table rbd_param_compression_hint[] = {
802	{"none", Opt_compression_hint_none},
803	{"compressible", Opt_compression_hint_compressible},
804	{"incompressible", Opt_compression_hint_incompressible},
805	{}
806	};
807
808	static const struct fs_parameter_spec rbd_parameters[] = {
809	fsparam_u32 ("alloc_size", Opt_alloc_size),
810	fsparam_enum ("compression_hint", Opt_compression_hint,
811	rbd_param_compression_hint),
812	fsparam_flag ("exclusive", Opt_exclusive),
813	fsparam_flag ("lock_on_read", Opt_lock_on_read),
814	fsparam_u32 ("lock_timeout", Opt_lock_timeout),
815	fsparam_flag ("notrim", Opt_notrim),
816	fsparam_string ("_pool_ns", Opt_pool_ns),
817	fsparam_u32 ("queue_depth", Opt_queue_depth),
818	fsparam_flag ("read_only", Opt_read_only),
819	fsparam_flag ("read_write", Opt_read_write),
820	fsparam_flag ("ro", Opt_read_only),
821	fsparam_flag ("rw", Opt_read_write),
822	{}
823	};
824
825	struct rbd_options {
826	int queue_depth;
827	int alloc_size;
828	unsigned long lock_timeout;
829	bool read_only;
830	bool lock_on_read;
831	bool exclusive;
832	bool trim;
833
834	u32 alloc_hint_flags; /* CEPH_OSD_OP_ALLOC_HINT_FLAG_* */
835	};
836
837	#define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_DEFAULT_RQ
838	#define RBD_ALLOC_SIZE_DEFAULT (64 * 1024)
839	#define RBD_LOCK_TIMEOUT_DEFAULT 0 /* no timeout */
840	#define RBD_READ_ONLY_DEFAULT false
841	#define RBD_LOCK_ON_READ_DEFAULT false
842	#define RBD_EXCLUSIVE_DEFAULT false
843	#define RBD_TRIM_DEFAULT true
844
845	struct rbd_parse_opts_ctx {
846	struct rbd_spec *spec;
847	struct ceph_options *copts;
848	struct rbd_options *opts;
849	};
850
851	static char* obj_op_name(enum obj_operation_type op_type)
852	{
853	switch (op_type) {
854	case OBJ_OP_READ:
855	return "read";
856	case OBJ_OP_WRITE:
857	return "write";
858	case OBJ_OP_DISCARD:
859	return "discard";
860	case OBJ_OP_ZEROOUT:
861	return "zeroout";
862	default:
863	return "???";
864	}
865	}
866
867	/*
868	* Destroy ceph client
869	*
870	* Caller must hold rbd_client_list_lock.
871	*/
872	static void rbd_client_release(struct kref *kref)
873	{
874	struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
875
876	dout("%s: rbdc %p\n", __func__, rbdc);
877	spin_lock(lock: &rbd_client_list_lock);
878	list_del(entry: &rbdc->node);
879	spin_unlock(lock: &rbd_client_list_lock);
880
881	ceph_destroy_client(client: rbdc->client);
882	kfree(objp: rbdc);
883	}
884
885	/*
886	* Drop reference to ceph client node. If it's not referenced anymore, release
887	* it.
888	*/
889	static void rbd_put_client(struct rbd_client *rbdc)
890	{
891	if (rbdc)
892	kref_put(kref: &rbdc->kref, release: rbd_client_release);
893	}
894
895	/*
896	* Get a ceph client with specific addr and configuration, if one does
897	* not exist create it. Either way, ceph_opts is consumed by this
898	* function.
899	*/
900	static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
901	{
902	struct rbd_client *rbdc;
903	int ret;
904
905	mutex_lock(&client_mutex);
906	rbdc = rbd_client_find(ceph_opts);
907	if (rbdc) {
908	ceph_destroy_options(opt: ceph_opts);
909
910	/*
911	* Using an existing client. Make sure ->pg_pools is up to
912	* date before we look up the pool id in do_rbd_add().
913	*/
914	ret = ceph_wait_for_latest_osdmap(client: rbdc->client,
915	timeout: rbdc->client->options->mount_timeout);
916	if (ret) {
917	rbd_warn(NULL, fmt: "failed to get latest osdmap: %d", ret);
918	rbd_put_client(rbdc);
919	rbdc = ERR_PTR(error: ret);
920	}
921	} else {
922	rbdc = rbd_client_create(ceph_opts);
923	}
924	mutex_unlock(lock: &client_mutex);
925
926	return rbdc;
927	}
928
929	static bool rbd_image_format_valid(u32 image_format)
930	{
931	return image_format == 1 || image_format == 2;
932	}
933
934	static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
935	{
936	size_t size;
937	u32 snap_count;
938
939	/* The header has to start with the magic rbd header text */
940	if (memcmp(p: &ondisk->text, RBD_HEADER_TEXT, size: sizeof (RBD_HEADER_TEXT)))
941	return false;
942
943	/* The bio layer requires at least sector-sized I/O */
944
945	if (ondisk->options.order < SECTOR_SHIFT)
946	return false;
947
948	/* If we use u64 in a few spots we may be able to loosen this */
949
950	if (ondisk->options.order > 8 * sizeof (int) - 1)
951	return false;
952
953	/*
954	* The size of a snapshot header has to fit in a size_t, and
955	* that limits the number of snapshots.
956	*/
957	snap_count = le32_to_cpu(ondisk->snap_count);
958	size = SIZE_MAX - sizeof (struct ceph_snap_context);
959	if (snap_count > size / sizeof (__le64))
960	return false;
961
962	/*
963	* Not only that, but the size of the entire the snapshot
964	* header must also be representable in a size_t.
965	*/
966	size -= snap_count * sizeof (__le64);
967	if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
968	return false;
969
970	return true;
971	}
972
973	/*
974	* returns the size of an object in the image
975	*/
976	static u32 rbd_obj_bytes(struct rbd_image_header *header)
977	{
978	return 1U << header->obj_order;
979	}
980
981	static void rbd_init_layout(struct rbd_device *rbd_dev)
982	{
983	if (rbd_dev->header.stripe_unit == 0 ||
984	rbd_dev->header.stripe_count == 0) {
985	rbd_dev->header.stripe_unit = rbd_obj_bytes(header: &rbd_dev->header);
986	rbd_dev->header.stripe_count = 1;
987	}
988
989	rbd_dev->layout.stripe_unit = rbd_dev->header.stripe_unit;
990	rbd_dev->layout.stripe_count = rbd_dev->header.stripe_count;
991	rbd_dev->layout.object_size = rbd_obj_bytes(header: &rbd_dev->header);
992	rbd_dev->layout.pool_id = rbd_dev->header.data_pool_id == CEPH_NOPOOL ?
993	rbd_dev->spec->pool_id : rbd_dev->header.data_pool_id;
994	RCU_INIT_POINTER(rbd_dev->layout.pool_ns, NULL);
995	}
996
997	static void rbd_image_header_cleanup(struct rbd_image_header *header)
998	{
999	kfree(objp: header->object_prefix);
1000	ceph_put_snap_context(sc: header->snapc);
1001	kfree(objp: header->snap_sizes);
1002	kfree(objp: header->snap_names);
1003
1004	memset(header, 0, sizeof(*header));
1005	}
1006
1007	/*
1008	* Fill an rbd image header with information from the given format 1
1009	* on-disk header.
1010	*/
1011	static int rbd_header_from_disk(struct rbd_image_header *header,
1012	struct rbd_image_header_ondisk *ondisk,
1013	bool first_time)
1014	{
1015	struct ceph_snap_context *snapc;
1016	char *object_prefix = NULL;
1017	char *snap_names = NULL;
1018	u64 *snap_sizes = NULL;
1019	u32 snap_count;
1020	int ret = -ENOMEM;
1021	u32 i;
1022
1023	/* Allocate this now to avoid having to handle failure below */
1024
1025	if (first_time) {
1026	object_prefix = kstrndup(s: ondisk->object_prefix,
1027	len: sizeof(ondisk->object_prefix),
1028	GFP_KERNEL);
1029	if (!object_prefix)
1030	return -ENOMEM;
1031	}
1032
1033	/* Allocate the snapshot context and fill it in */
1034
1035	snap_count = le32_to_cpu(ondisk->snap_count);
1036	snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
1037	if (!snapc)
1038	goto out_err;
1039	snapc->seq = le64_to_cpu(ondisk->snap_seq);
1040	if (snap_count) {
1041	struct rbd_image_snap_ondisk *snaps;
1042	u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
1043
1044	/* We'll keep a copy of the snapshot names... */
1045
1046	if (snap_names_len > (u64)SIZE_MAX)
1047	goto out_2big;
1048	snap_names = kmalloc(size: snap_names_len, GFP_KERNEL);
1049	if (!snap_names)
1050	goto out_err;
1051
1052	/* ...as well as the array of their sizes. */
1053	snap_sizes = kmalloc_array(n: snap_count,
1054	size: sizeof(*header->snap_sizes),
1055	GFP_KERNEL);
1056	if (!snap_sizes)
1057	goto out_err;
1058
1059	/*
1060	* Copy the names, and fill in each snapshot's id
1061	* and size.
1062	*
1063	* Note that rbd_dev_v1_header_info() guarantees the
1064	* ondisk buffer we're working with has
1065	* snap_names_len bytes beyond the end of the
1066	* snapshot id array, this memcpy() is safe.
1067	*/
1068	memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
1069	snaps = ondisk->snaps;
1070	for (i = 0; i < snap_count; i++) {
1071	snapc->snaps[i] = le64_to_cpu(snaps[i].id);
1072	snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
1073	}
1074	}
1075
1076	/* We won't fail any more, fill in the header */
1077
1078	if (first_time) {
1079	header->object_prefix = object_prefix;
1080	header->obj_order = ondisk->options.order;
1081	}
1082
1083	/* The remaining fields always get updated (when we refresh) */
1084
1085	header->image_size = le64_to_cpu(ondisk->image_size);
1086	header->snapc = snapc;
1087	header->snap_names = snap_names;
1088	header->snap_sizes = snap_sizes;
1089
1090	return 0;
1091	out_2big:
1092	ret = -EIO;
1093	out_err:
1094	kfree(objp: snap_sizes);
1095	kfree(objp: snap_names);
1096	ceph_put_snap_context(sc: snapc);
1097	kfree(objp: object_prefix);
1098
1099	return ret;
1100	}
1101
1102	static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1103	{
1104	const char *snap_name;
1105
1106	rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1107
1108	/* Skip over names until we find the one we are looking for */
1109
1110	snap_name = rbd_dev->header.snap_names;
1111	while (which--)
1112	snap_name += strlen(snap_name) + 1;
1113
1114	return kstrdup(s: snap_name, GFP_KERNEL);
1115	}
1116
1117	/*
1118	* Snapshot id comparison function for use with qsort()/bsearch().
1119	* Note that result is for snapshots in *descending* order.
1120	*/
1121	static int snapid_compare_reverse(const void *s1, const void *s2)
1122	{
1123	u64 snap_id1 = *(u64 *)s1;
1124	u64 snap_id2 = *(u64 *)s2;
1125
1126	if (snap_id1 < snap_id2)
1127	return 1;
1128	return snap_id1 == snap_id2 ? 0 : -1;
1129	}
1130
1131	/*
1132	* Search a snapshot context to see if the given snapshot id is
1133	* present.
1134	*
1135	* Returns the position of the snapshot id in the array if it's found,
1136	* or BAD_SNAP_INDEX otherwise.
1137	*
1138	* Note: The snapshot array is in kept sorted (by the osd) in
1139	* reverse order, highest snapshot id first.
1140	*/
1141	static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1142	{
1143	struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1144	u64 *found;
1145
1146	found = bsearch(key: &snap_id, base: &snapc->snaps, num: snapc->num_snaps,
1147	size: sizeof (snap_id), cmp: snapid_compare_reverse);
1148
1149	return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1150	}
1151
1152	static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1153	u64 snap_id)
1154	{
1155	u32 which;
1156	const char *snap_name;
1157
1158	which = rbd_dev_snap_index(rbd_dev, snap_id);
1159	if (which == BAD_SNAP_INDEX)
1160	return ERR_PTR(error: -ENOENT);
1161
1162	snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1163	return snap_name ? snap_name : ERR_PTR(error: -ENOMEM);
1164	}
1165
1166	static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1167	{
1168	if (snap_id == CEPH_NOSNAP)
1169	return RBD_SNAP_HEAD_NAME;
1170
1171	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1172	if (rbd_dev->image_format == 1)
1173	return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1174
1175	return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1176	}
1177
1178	static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1179	u64 *snap_size)
1180	{
1181	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1182	if (snap_id == CEPH_NOSNAP) {
1183	*snap_size = rbd_dev->header.image_size;
1184	} else if (rbd_dev->image_format == 1) {
1185	u32 which;
1186
1187	which = rbd_dev_snap_index(rbd_dev, snap_id);
1188	if (which == BAD_SNAP_INDEX)
1189	return -ENOENT;
1190
1191	*snap_size = rbd_dev->header.snap_sizes[which];
1192	} else {
1193	u64 size = 0;
1194	int ret;
1195
1196	ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, snap_size: &size);
1197	if (ret)
1198	return ret;
1199
1200	*snap_size = size;
1201	}
1202	return 0;
1203	}
1204
1205	static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1206	{
1207	u64 snap_id = rbd_dev->spec->snap_id;
1208	u64 size = 0;
1209	int ret;
1210
1211	ret = rbd_snap_size(rbd_dev, snap_id, snap_size: &size);
1212	if (ret)
1213	return ret;
1214
1215	rbd_dev->mapping.size = size;
1216	return 0;
1217	}
1218
1219	static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1220	{
1221	rbd_dev->mapping.size = 0;
1222	}
1223
1224	static void zero_bios(struct ceph_bio_iter *bio_pos, u32 off, u32 bytes)
1225	{
1226	struct ceph_bio_iter it = *bio_pos;
1227
1228	ceph_bio_iter_advance(&it, off);
1229	ceph_bio_iter_advance_step(&it, bytes, ({
1230	memzero_bvec(&bv);
1231	}));
1232	}
1233
1234	static void zero_bvecs(struct ceph_bvec_iter *bvec_pos, u32 off, u32 bytes)
1235	{
1236	struct ceph_bvec_iter it = *bvec_pos;
1237
1238	ceph_bvec_iter_advance(&it, off);
1239	ceph_bvec_iter_advance_step(&it, bytes, ({
1240	memzero_bvec(&bv);
1241	}));
1242	}
1243
1244	/*
1245	* Zero a range in @obj_req data buffer defined by a bio (list) or
1246	* (private) bio_vec array.
1247	*
1248	* @off is relative to the start of the data buffer.
1249	*/
1250	static void rbd_obj_zero_range(struct rbd_obj_request *obj_req, u32 off,
1251	u32 bytes)
1252	{
1253	dout("%s %p data buf %u~%u\n", __func__, obj_req, off, bytes);
1254
1255	switch (obj_req->img_request->data_type) {
1256	case OBJ_REQUEST_BIO:
1257	zero_bios(bio_pos: &obj_req->bio_pos, off, bytes);
1258	break;
1259	case OBJ_REQUEST_BVECS:
1260	case OBJ_REQUEST_OWN_BVECS:
1261	zero_bvecs(bvec_pos: &obj_req->bvec_pos, off, bytes);
1262	break;
1263	default:
1264	BUG();
1265	}
1266	}
1267
1268	static void rbd_obj_request_destroy(struct kref *kref);
1269	static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1270	{
1271	rbd_assert(obj_request != NULL);
1272	dout("%s: obj %p (was %d)\n", __func__, obj_request,
1273	kref_read(&obj_request->kref));
1274	kref_put(kref: &obj_request->kref, release: rbd_obj_request_destroy);
1275	}
1276
1277	static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1278	struct rbd_obj_request *obj_request)
1279	{
1280	rbd_assert(obj_request->img_request == NULL);
1281
1282	/* Image request now owns object's original reference */
1283	obj_request->img_request = img_request;
1284	dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1285	}
1286
1287	static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1288	struct rbd_obj_request *obj_request)
1289	{
1290	dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1291	list_del(entry: &obj_request->ex.oe_item);
1292	rbd_assert(obj_request->img_request == img_request);
1293	rbd_obj_request_put(obj_request);
1294	}
1295
1296	static void rbd_osd_submit(struct ceph_osd_request *osd_req)
1297	{
1298	struct rbd_obj_request *obj_req = osd_req->r_priv;
1299
1300	dout("%s osd_req %p for obj_req %p objno %llu %llu~%llu\n",
1301	__func__, osd_req, obj_req, obj_req->ex.oe_objno,
1302	obj_req->ex.oe_off, obj_req->ex.oe_len);
1303	ceph_osdc_start_request(osdc: osd_req->r_osdc, req: osd_req);
1304	}
1305
1306	/*
1307	* The default/initial value for all image request flags is 0. Each
1308	* is conditionally set to 1 at image request initialization time
1309	* and currently never change thereafter.
1310	*/
1311	static void img_request_layered_set(struct rbd_img_request *img_request)
1312	{
1313	set_bit(nr: IMG_REQ_LAYERED, addr: &img_request->flags);
1314	}
1315
1316	static bool img_request_layered_test(struct rbd_img_request *img_request)
1317	{
1318	return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1319	}
1320
1321	static bool rbd_obj_is_entire(struct rbd_obj_request *obj_req)
1322	{
1323	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1324
1325	return !obj_req->ex.oe_off &&
1326	obj_req->ex.oe_len == rbd_dev->layout.object_size;
1327	}
1328
1329	static bool rbd_obj_is_tail(struct rbd_obj_request *obj_req)
1330	{
1331	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1332
1333	return obj_req->ex.oe_off + obj_req->ex.oe_len ==
1334	rbd_dev->layout.object_size;
1335	}
1336
1337	/*
1338	* Must be called after rbd_obj_calc_img_extents().
1339	*/
1340	static void rbd_obj_set_copyup_enabled(struct rbd_obj_request *obj_req)
1341	{
1342	rbd_assert(obj_req->img_request->snapc);
1343
1344	if (obj_req->img_request->op_type == OBJ_OP_DISCARD) {
1345	dout("%s %p objno %llu discard\n", __func__, obj_req,
1346	obj_req->ex.oe_objno);
1347	return;
1348	}
1349
1350	if (!obj_req->num_img_extents) {
1351	dout("%s %p objno %llu not overlapping\n", __func__, obj_req,
1352	obj_req->ex.oe_objno);
1353	return;
1354	}
1355
1356	if (rbd_obj_is_entire(obj_req) &&
1357	!obj_req->img_request->snapc->num_snaps) {
1358	dout("%s %p objno %llu entire\n", __func__, obj_req,
1359	obj_req->ex.oe_objno);
1360	return;
1361	}
1362
1363	obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
1364	}
1365
1366	static u64 rbd_obj_img_extents_bytes(struct rbd_obj_request *obj_req)
1367	{
1368	return ceph_file_extents_bytes(file_extents: obj_req->img_extents,
1369	num_file_extents: obj_req->num_img_extents);
1370	}
1371
1372	static bool rbd_img_is_write(struct rbd_img_request *img_req)
1373	{
1374	switch (img_req->op_type) {
1375	case OBJ_OP_READ:
1376	return false;
1377	case OBJ_OP_WRITE:
1378	case OBJ_OP_DISCARD:
1379	case OBJ_OP_ZEROOUT:
1380	return true;
1381	default:
1382	BUG();
1383	}
1384	}
1385
1386	static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
1387	{
1388	struct rbd_obj_request *obj_req = osd_req->r_priv;
1389	int result;
1390
1391	dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
1392	osd_req->r_result, obj_req);
1393
1394	/*
1395	* Writes aren't allowed to return a data payload. In some
1396	* guarded write cases (e.g. stat + zero on an empty object)
1397	* a stat response makes it through, but we don't care.
1398	*/
1399	if (osd_req->r_result > 0 && rbd_img_is_write(img_req: obj_req->img_request))
1400	result = 0;
1401	else
1402	result = osd_req->r_result;
1403
1404	rbd_obj_handle_request(obj_req, result);
1405	}
1406
1407	static void rbd_osd_format_read(struct ceph_osd_request *osd_req)
1408	{
1409	struct rbd_obj_request *obj_request = osd_req->r_priv;
1410	struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1411	struct ceph_options *opt = rbd_dev->rbd_client->client->options;
1412
1413	osd_req->r_flags = CEPH_OSD_FLAG_READ | opt->read_from_replica;
1414	osd_req->r_snapid = obj_request->img_request->snap_id;
1415	}
1416
1417	static void rbd_osd_format_write(struct ceph_osd_request *osd_req)
1418	{
1419	struct rbd_obj_request *obj_request = osd_req->r_priv;
1420
1421	osd_req->r_flags = CEPH_OSD_FLAG_WRITE;
1422	ktime_get_real_ts64(tv: &osd_req->r_mtime);
1423	osd_req->r_data_offset = obj_request->ex.oe_off;
1424	}
1425
1426	static struct ceph_osd_request *
1427	__rbd_obj_add_osd_request(struct rbd_obj_request *obj_req,
1428	struct ceph_snap_context *snapc, int num_ops)
1429	{
1430	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1431	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1432	struct ceph_osd_request *req;
1433	const char *name_format = rbd_dev->image_format == 1 ?
1434	RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
1435	int ret;
1436
1437	req = ceph_osdc_alloc_request(osdc, snapc, num_ops, use_mempool: false, GFP_NOIO);
1438	if (!req)
1439	return ERR_PTR(error: -ENOMEM);
1440
1441	list_add_tail(new: &req->r_private_item, head: &obj_req->osd_reqs);
1442	req->r_callback = rbd_osd_req_callback;
1443	req->r_priv = obj_req;
1444
1445	/*
1446	* Data objects may be stored in a separate pool, but always in
1447	* the same namespace in that pool as the header in its pool.
1448	*/
1449	ceph_oloc_copy(dest: &req->r_base_oloc, src: &rbd_dev->header_oloc);
1450	req->r_base_oloc.pool = rbd_dev->layout.pool_id;
1451
1452	ret = ceph_oid_aprintf(oid: &req->r_base_oid, GFP_NOIO, fmt: name_format,
1453	rbd_dev->header.object_prefix,
1454	obj_req->ex.oe_objno);
1455	if (ret)
1456	return ERR_PTR(error: ret);
1457
1458	return req;
1459	}
1460
1461	static struct ceph_osd_request *
1462	rbd_obj_add_osd_request(struct rbd_obj_request *obj_req, int num_ops)
1463	{
1464	rbd_assert(obj_req->img_request->snapc);
1465	return __rbd_obj_add_osd_request(obj_req, snapc: obj_req->img_request->snapc,
1466	num_ops);
1467	}
1468
1469	static struct rbd_obj_request *rbd_obj_request_create(void)
1470	{
1471	struct rbd_obj_request *obj_request;
1472
1473	obj_request = kmem_cache_zalloc(k: rbd_obj_request_cache, GFP_NOIO);
1474	if (!obj_request)
1475	return NULL;
1476
1477	ceph_object_extent_init(ex: &obj_request->ex);
1478	INIT_LIST_HEAD(list: &obj_request->osd_reqs);
1479	mutex_init(&obj_request->state_mutex);
1480	kref_init(kref: &obj_request->kref);
1481
1482	dout("%s %p\n", __func__, obj_request);
1483	return obj_request;
1484	}
1485
1486	static void rbd_obj_request_destroy(struct kref *kref)
1487	{
1488	struct rbd_obj_request *obj_request;
1489	struct ceph_osd_request *osd_req;
1490	u32 i;
1491
1492	obj_request = container_of(kref, struct rbd_obj_request, kref);
1493
1494	dout("%s: obj %p\n", __func__, obj_request);
1495
1496	while (!list_empty(head: &obj_request->osd_reqs)) {
1497	osd_req = list_first_entry(&obj_request->osd_reqs,
1498	struct ceph_osd_request, r_private_item);
1499	list_del_init(entry: &osd_req->r_private_item);
1500	ceph_osdc_put_request(req: osd_req);
1501	}
1502
1503	switch (obj_request->img_request->data_type) {
1504	case OBJ_REQUEST_NODATA:
1505	case OBJ_REQUEST_BIO:
1506	case OBJ_REQUEST_BVECS:
1507	break; /* Nothing to do */
1508	case OBJ_REQUEST_OWN_BVECS:
1509	kfree(objp: obj_request->bvec_pos.bvecs);
1510	break;
1511	default:
1512	BUG();
1513	}
1514
1515	kfree(objp: obj_request->img_extents);
1516	if (obj_request->copyup_bvecs) {
1517	for (i = 0; i < obj_request->copyup_bvec_count; i++) {
1518	if (obj_request->copyup_bvecs[i].bv_page)
1519	__free_page(obj_request->copyup_bvecs[i].bv_page);
1520	}
1521	kfree(objp: obj_request->copyup_bvecs);
1522	}
1523
1524	kmem_cache_free(s: rbd_obj_request_cache, objp: obj_request);
1525	}
1526
1527	/* It's OK to call this for a device with no parent */
1528
1529	static void rbd_spec_put(struct rbd_spec *spec);
1530	static void rbd_dev_unparent(struct rbd_device *rbd_dev)
1531	{
1532	rbd_dev_remove_parent(rbd_dev);
1533	rbd_spec_put(spec: rbd_dev->parent_spec);
1534	rbd_dev->parent_spec = NULL;
1535	rbd_dev->parent_overlap = 0;
1536	}
1537
1538	/*
1539	* Parent image reference counting is used to determine when an
1540	* image's parent fields can be safely torn down--after there are no
1541	* more in-flight requests to the parent image. When the last
1542	* reference is dropped, cleaning them up is safe.
1543	*/
1544	static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
1545	{
1546	int counter;
1547
1548	if (!rbd_dev->parent_spec)
1549	return;
1550
1551	counter = atomic_dec_return_safe(v: &rbd_dev->parent_ref);
1552	if (counter > 0)
1553	return;
1554
1555	/* Last reference; clean up parent data structures */
1556
1557	if (!counter)
1558	rbd_dev_unparent(rbd_dev);
1559	else
1560	rbd_warn(rbd_dev, fmt: "parent reference underflow");
1561	}
1562
1563	/*
1564	* If an image has a non-zero parent overlap, get a reference to its
1565	* parent.
1566	*
1567	* Returns true if the rbd device has a parent with a non-zero
1568	* overlap and a reference for it was successfully taken, or
1569	* false otherwise.
1570	*/
1571	static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
1572	{
1573	int counter = 0;
1574
1575	if (!rbd_dev->parent_spec)
1576	return false;
1577
1578	if (rbd_dev->parent_overlap)
1579	counter = atomic_inc_return_safe(v: &rbd_dev->parent_ref);
1580
1581	if (counter < 0)
1582	rbd_warn(rbd_dev, fmt: "parent reference overflow");
1583
1584	return counter > 0;
1585	}
1586
1587	static void rbd_img_request_init(struct rbd_img_request *img_request,
1588	struct rbd_device *rbd_dev,
1589	enum obj_operation_type op_type)
1590	{
1591	memset(img_request, 0, sizeof(*img_request));
1592
1593	img_request->rbd_dev = rbd_dev;
1594	img_request->op_type = op_type;
1595
1596	INIT_LIST_HEAD(list: &img_request->lock_item);
1597	INIT_LIST_HEAD(list: &img_request->object_extents);
1598	mutex_init(&img_request->state_mutex);
1599	}
1600
1601	/*
1602	* Only snap_id is captured here, for reads. For writes, snapshot
1603	* context is captured in rbd_img_object_requests() after exclusive
1604	* lock is ensured to be held.
1605	*/
1606	static void rbd_img_capture_header(struct rbd_img_request *img_req)
1607	{
1608	struct rbd_device *rbd_dev = img_req->rbd_dev;
1609
1610	lockdep_assert_held(&rbd_dev->header_rwsem);
1611
1612	if (!rbd_img_is_write(img_req))
1613	img_req->snap_id = rbd_dev->spec->snap_id;
1614
1615	if (rbd_dev_parent_get(rbd_dev))
1616	img_request_layered_set(img_request: img_req);
1617	}
1618
1619	static void rbd_img_request_destroy(struct rbd_img_request *img_request)
1620	{
1621	struct rbd_obj_request *obj_request;
1622	struct rbd_obj_request *next_obj_request;
1623
1624	dout("%s: img %p\n", __func__, img_request);
1625
1626	WARN_ON(!list_empty(&img_request->lock_item));
1627	for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1628	rbd_img_obj_request_del(img_request, obj_request);
1629
1630	if (img_request_layered_test(img_request))
1631	rbd_dev_parent_put(rbd_dev: img_request->rbd_dev);
1632
1633	if (rbd_img_is_write(img_req: img_request))
1634	ceph_put_snap_context(sc: img_request->snapc);
1635
1636	if (test_bit(IMG_REQ_CHILD, &img_request->flags))
1637	kmem_cache_free(s: rbd_img_request_cache, objp: img_request);
1638	}
1639
1640	#define BITS_PER_OBJ 2
1641	#define OBJS_PER_BYTE (BITS_PER_BYTE / BITS_PER_OBJ)
1642	#define OBJ_MASK ((1 << BITS_PER_OBJ) - 1)
1643
1644	static void __rbd_object_map_index(struct rbd_device *rbd_dev, u64 objno,
1645	u64 *index, u8 *shift)
1646	{
1647	u32 off;
1648
1649	rbd_assert(objno < rbd_dev->object_map_size);
1650	*index = div_u64_rem(dividend: objno, OBJS_PER_BYTE, remainder: &off);
1651	*shift = (OBJS_PER_BYTE - off - 1) * BITS_PER_OBJ;
1652	}
1653
1654	static u8 __rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1655	{
1656	u64 index;
1657	u8 shift;
1658
1659	lockdep_assert_held(&rbd_dev->object_map_lock);
1660	__rbd_object_map_index(rbd_dev, objno, index: &index, shift: &shift);
1661	return (rbd_dev->object_map[index] >> shift) & OBJ_MASK;
1662	}
1663
1664	static void __rbd_object_map_set(struct rbd_device *rbd_dev, u64 objno, u8 val)
1665	{
1666	u64 index;
1667	u8 shift;
1668	u8 *p;
1669
1670	lockdep_assert_held(&rbd_dev->object_map_lock);
1671	rbd_assert(!(val & ~OBJ_MASK));
1672
1673	__rbd_object_map_index(rbd_dev, objno, index: &index, shift: &shift);
1674	p = &rbd_dev->object_map[index];
1675	*p = (*p & ~(OBJ_MASK << shift)) | (val << shift);
1676	}
1677
1678	static u8 rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1679	{
1680	u8 state;
1681
1682	spin_lock(lock: &rbd_dev->object_map_lock);
1683	state = __rbd_object_map_get(rbd_dev, objno);
1684	spin_unlock(lock: &rbd_dev->object_map_lock);
1685	return state;
1686	}
1687
1688	static bool use_object_map(struct rbd_device *rbd_dev)
1689	{
1690	/*
1691	* An image mapped read-only can't use the object map -- it isn't
1692	* loaded because the header lock isn't acquired. Someone else can
1693	* write to the image and update the object map behind our back.
1694	*
1695	* A snapshot can't be written to, so using the object map is always
1696	* safe.
1697	*/
1698	if (!rbd_is_snap(rbd_dev) && rbd_is_ro(rbd_dev))
1699	return false;
1700
1701	return ((rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) &&
1702	!(rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID));
1703	}
1704
1705	static bool rbd_object_map_may_exist(struct rbd_device *rbd_dev, u64 objno)
1706	{
1707	u8 state;
1708
1709	/* fall back to default logic if object map is disabled or invalid */
1710	if (!use_object_map(rbd_dev))
1711	return true;
1712
1713	state = rbd_object_map_get(rbd_dev, objno);
1714	return state != OBJECT_NONEXISTENT;
1715	}
1716
1717	static void rbd_object_map_name(struct rbd_device *rbd_dev, u64 snap_id,
1718	struct ceph_object_id *oid)
1719	{
1720	if (snap_id == CEPH_NOSNAP)
1721	ceph_oid_printf(oid, fmt: "%s%s", RBD_OBJECT_MAP_PREFIX,
1722	rbd_dev->spec->image_id);
1723	else
1724	ceph_oid_printf(oid, fmt: "%s%s.%016llx", RBD_OBJECT_MAP_PREFIX,
1725	rbd_dev->spec->image_id, snap_id);
1726	}
1727
1728	static int rbd_object_map_lock(struct rbd_device *rbd_dev)
1729	{
1730	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1731	CEPH_DEFINE_OID_ONSTACK(oid);
1732	u8 lock_type;
1733	char *lock_tag;
1734	struct ceph_locker *lockers;
1735	u32 num_lockers;
1736	bool broke_lock = false;
1737	int ret;
1738
1739	rbd_object_map_name(rbd_dev, CEPH_NOSNAP, oid: &oid);
1740
1741	again:
1742	ret = ceph_cls_lock(osdc, oid: &oid, oloc: &rbd_dev->header_oloc, RBD_LOCK_NAME,
1743	type: CEPH_CLS_LOCK_EXCLUSIVE, cookie: "", tag: "", desc: "", flags: 0);
1744	if (ret != -EBUSY || broke_lock) {
1745	if (ret == -EEXIST)
1746	ret = 0; /* already locked by myself */
1747	if (ret)
1748	rbd_warn(rbd_dev, fmt: "failed to lock object map: %d", ret);
1749	return ret;
1750	}
1751
1752	ret = ceph_cls_lock_info(osdc, oid: &oid, oloc: &rbd_dev->header_oloc,
1753	RBD_LOCK_NAME, type: &lock_type, tag: &lock_tag,
1754	lockers: &lockers, num_lockers: &num_lockers);
1755	if (ret) {
1756	if (ret == -ENOENT)
1757	goto again;
1758
1759	rbd_warn(rbd_dev, fmt: "failed to get object map lockers: %d", ret);
1760	return ret;
1761	}
1762
1763	kfree(objp: lock_tag);
1764	if (num_lockers == 0)
1765	goto again;
1766
1767	rbd_warn(rbd_dev, fmt: "breaking object map lock owned by %s%llu",
1768	ENTITY_NAME(lockers[0].id.name));
1769
1770	ret = ceph_cls_break_lock(osdc, oid: &oid, oloc: &rbd_dev->header_oloc,
1771	RBD_LOCK_NAME, cookie: lockers[0].id.cookie,
1772	locker: &lockers[0].id.name);
1773	ceph_free_lockers(lockers, num_lockers);
1774	if (ret) {
1775	if (ret == -ENOENT)
1776	goto again;
1777
1778	rbd_warn(rbd_dev, fmt: "failed to break object map lock: %d", ret);
1779	return ret;
1780	}
1781
1782	broke_lock = true;
1783	goto again;
1784	}
1785
1786	static void rbd_object_map_unlock(struct rbd_device *rbd_dev)
1787	{
1788	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1789	CEPH_DEFINE_OID_ONSTACK(oid);
1790	int ret;
1791
1792	rbd_object_map_name(rbd_dev, CEPH_NOSNAP, oid: &oid);
1793
1794	ret = ceph_cls_unlock(osdc, oid: &oid, oloc: &rbd_dev->header_oloc, RBD_LOCK_NAME,
1795	cookie: "");
1796	if (ret && ret != -ENOENT)
1797	rbd_warn(rbd_dev, fmt: "failed to unlock object map: %d", ret);
1798	}
1799
1800	static int decode_object_map_header(void **p, void *end, u64 *object_map_size)
1801	{
1802	u8 struct_v;
1803	u32 struct_len;
1804	u32 header_len;
1805	void *header_end;
1806	int ret;
1807
1808	ceph_decode_32_safe(p, end, header_len, e_inval);
1809	header_end = *p + header_len;
1810
1811	ret = ceph_start_decoding(p, end, v: 1, name: "BitVector header", struct_v: &struct_v,
1812	struct_len: &struct_len);
1813	if (ret)
1814	return ret;
1815
1816	ceph_decode_64_safe(p, end, *object_map_size, e_inval);
1817
1818	*p = header_end;
1819	return 0;
1820
1821	e_inval:
1822	return -EINVAL;
1823	}
1824
1825	static int __rbd_object_map_load(struct rbd_device *rbd_dev)
1826	{
1827	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1828	CEPH_DEFINE_OID_ONSTACK(oid);
1829	struct page **pages;
1830	void *p, *end;
1831	size_t reply_len;
1832	u64 num_objects;
1833	u64 object_map_bytes;
1834	u64 object_map_size;
1835	int num_pages;
1836	int ret;
1837
1838	rbd_assert(!rbd_dev->object_map && !rbd_dev->object_map_size);
1839
1840	num_objects = ceph_get_num_objects(l: &rbd_dev->layout,
1841	size: rbd_dev->mapping.size);
1842	object_map_bytes = DIV_ROUND_UP_ULL(num_objects * BITS_PER_OBJ,
1843	BITS_PER_BYTE);
1844	num_pages = calc_pages_for(off: 0, len: object_map_bytes) + 1;
1845	pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1846	if (IS_ERR(ptr: pages))
1847	return PTR_ERR(ptr: pages);
1848
1849	reply_len = num_pages * PAGE_SIZE;
1850	rbd_object_map_name(rbd_dev, snap_id: rbd_dev->spec->snap_id, oid: &oid);
1851	ret = ceph_osdc_call(osdc, oid: &oid, oloc: &rbd_dev->header_oloc,
1852	class: "rbd", method: "object_map_load", flags: CEPH_OSD_FLAG_READ,
1853	NULL, req_len: 0, resp_pages: pages, resp_len: &reply_len);
1854	if (ret)
1855	goto out;
1856
1857	p = page_address(pages[0]);
1858	end = p + min(reply_len, (size_t)PAGE_SIZE);
1859	ret = decode_object_map_header(p: &p, end, object_map_size: &object_map_size);
1860	if (ret)
1861	goto out;
1862
1863	if (object_map_size != num_objects) {
1864	rbd_warn(rbd_dev, fmt: "object map size mismatch: %llu vs %llu",
1865	object_map_size, num_objects);
1866	ret = -EINVAL;
1867	goto out;
1868	}
1869
1870	if (offset_in_page(p) + object_map_bytes > reply_len) {
1871	ret = -EINVAL;
1872	goto out;
1873	}
1874
1875	rbd_dev->object_map = kvmalloc(size: object_map_bytes, GFP_KERNEL);
1876	if (!rbd_dev->object_map) {
1877	ret = -ENOMEM;
1878	goto out;
1879	}
1880
1881	rbd_dev->object_map_size = object_map_size;
1882	ceph_copy_from_page_vector(pages, data: rbd_dev->object_map,
1883	offset_in_page(p), len: object_map_bytes);
1884
1885	out:
1886	ceph_release_page_vector(pages, num_pages);
1887	return ret;
1888	}
1889
1890	static void rbd_object_map_free(struct rbd_device *rbd_dev)
1891	{
1892	kvfree(addr: rbd_dev->object_map);
1893	rbd_dev->object_map = NULL;
1894	rbd_dev->object_map_size = 0;
1895	}
1896
1897	static int rbd_object_map_load(struct rbd_device *rbd_dev)
1898	{
1899	int ret;
1900
1901	ret = __rbd_object_map_load(rbd_dev);
1902	if (ret)
1903	return ret;
1904
1905	ret = rbd_dev_v2_get_flags(rbd_dev);
1906	if (ret) {
1907	rbd_object_map_free(rbd_dev);
1908	return ret;
1909	}
1910
1911	if (rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID)
1912	rbd_warn(rbd_dev, fmt: "object map is invalid");
1913
1914	return 0;
1915	}
1916
1917	static int rbd_object_map_open(struct rbd_device *rbd_dev)
1918	{
1919	int ret;
1920
1921	ret = rbd_object_map_lock(rbd_dev);
1922	if (ret)
1923	return ret;
1924
1925	ret = rbd_object_map_load(rbd_dev);
1926	if (ret) {
1927	rbd_object_map_unlock(rbd_dev);
1928	return ret;
1929	}
1930
1931	return 0;
1932	}
1933
1934	static void rbd_object_map_close(struct rbd_device *rbd_dev)
1935	{
1936	rbd_object_map_free(rbd_dev);
1937	rbd_object_map_unlock(rbd_dev);
1938	}
1939
1940	/*
1941	* This function needs snap_id (or more precisely just something to
1942	* distinguish between HEAD and snapshot object maps), new_state and
1943	* current_state that were passed to rbd_object_map_update().
1944	*
1945	* To avoid allocating and stashing a context we piggyback on the OSD
1946	* request. A HEAD update has two ops (assert_locked). For new_state
1947	* and current_state we decode our own object_map_update op, encoded in
1948	* rbd_cls_object_map_update().
1949	*/
1950	static int rbd_object_map_update_finish(struct rbd_obj_request *obj_req,
1951	struct ceph_osd_request *osd_req)
1952	{
1953	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1954	struct ceph_osd_data *osd_data;
1955	u64 objno;
1956	u8 state, new_state, current_state;
1957	bool has_current_state;
1958	void *p;
1959
1960	if (osd_req->r_result)
1961	return osd_req->r_result;
1962
1963	/*
1964	* Nothing to do for a snapshot object map.
1965	*/
1966	if (osd_req->r_num_ops == 1)
1967	return 0;
1968
1969	/*
1970	* Update in-memory HEAD object map.
1971	*/
1972	rbd_assert(osd_req->r_num_ops == 2);
1973	osd_data = osd_req_op_data(osd_req, 1, cls, request_data);
1974	rbd_assert(osd_data->type == CEPH_OSD_DATA_TYPE_PAGES);
1975
1976	p = page_address(osd_data->pages[0]);
1977	objno = ceph_decode_64(p: &p);
1978	rbd_assert(objno == obj_req->ex.oe_objno);
1979	rbd_assert(ceph_decode_64(&p) == objno + 1);
1980	new_state = ceph_decode_8(p: &p);
1981	has_current_state = ceph_decode_8(p: &p);
1982	if (has_current_state)
1983	current_state = ceph_decode_8(p: &p);
1984
1985	spin_lock(lock: &rbd_dev->object_map_lock);
1986	state = __rbd_object_map_get(rbd_dev, objno);
1987	if (!has_current_state || current_state == state ||
1988	(current_state == OBJECT_EXISTS && state == OBJECT_EXISTS_CLEAN))
1989	__rbd_object_map_set(rbd_dev, objno, val: new_state);
1990	spin_unlock(lock: &rbd_dev->object_map_lock);
1991
1992	return 0;
1993	}
1994
1995	static void rbd_object_map_callback(struct ceph_osd_request *osd_req)
1996	{
1997	struct rbd_obj_request *obj_req = osd_req->r_priv;
1998	int result;
1999
2000	dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
2001	osd_req->r_result, obj_req);
2002
2003	result = rbd_object_map_update_finish(obj_req, osd_req);
2004	rbd_obj_handle_request(obj_req, result);
2005	}
2006
2007	static bool update_needed(struct rbd_device *rbd_dev, u64 objno, u8 new_state)
2008	{
2009	u8 state = rbd_object_map_get(rbd_dev, objno);
2010
2011	if (state == new_state ||
2012	(new_state == OBJECT_PENDING && state == OBJECT_NONEXISTENT) ||
2013	(new_state == OBJECT_NONEXISTENT && state != OBJECT_PENDING))
2014	return false;
2015
2016	return true;
2017	}
2018
2019	static int rbd_cls_object_map_update(struct ceph_osd_request *req,
2020	int which, u64 objno, u8 new_state,
2021	const u8 *current_state)
2022	{
2023	struct page **pages;
2024	void *p, *start;
2025	int ret;
2026
2027	ret = osd_req_op_cls_init(osd_req: req, which, class: "rbd", method: "object_map_update");
2028	if (ret)
2029	return ret;
2030
2031	pages = ceph_alloc_page_vector(num_pages: 1, GFP_NOIO);
2032	if (IS_ERR(ptr: pages))
2033	return PTR_ERR(ptr: pages);
2034
2035	p = start = page_address(pages[0]);
2036	ceph_encode_64(p: &p, v: objno);
2037	ceph_encode_64(p: &p, v: objno + 1);
2038	ceph_encode_8(p: &p, v: new_state);
2039	if (current_state) {
2040	ceph_encode_8(p: &p, v: 1);
2041	ceph_encode_8(p: &p, v: *current_state);
2042	} else {
2043	ceph_encode_8(p: &p, v: 0);
2044	}
2045
2046	osd_req_op_cls_request_data_pages(req, which, pages, length: p - start, alignment: 0,
2047	pages_from_pool: false, own_pages: true);
2048	return 0;
2049	}
2050
2051	/*
2052	* Return:
2053	* 0 - object map update sent
2054	* 1 - object map update isn't needed
2055	* <0 - error
2056	*/
2057	static int rbd_object_map_update(struct rbd_obj_request *obj_req, u64 snap_id,
2058	u8 new_state, const u8 *current_state)
2059	{
2060	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2061	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2062	struct ceph_osd_request *req;
2063	int num_ops = 1;
2064	int which = 0;
2065	int ret;
2066
2067	if (snap_id == CEPH_NOSNAP) {
2068	if (!update_needed(rbd_dev, objno: obj_req->ex.oe_objno, new_state))
2069	return 1;
2070
2071	num_ops++; /* assert_locked */
2072	}
2073
2074	req = ceph_osdc_alloc_request(osdc, NULL, num_ops, use_mempool: false, GFP_NOIO);
2075	if (!req)
2076	return -ENOMEM;
2077
2078	list_add_tail(new: &req->r_private_item, head: &obj_req->osd_reqs);
2079	req->r_callback = rbd_object_map_callback;
2080	req->r_priv = obj_req;
2081
2082	rbd_object_map_name(rbd_dev, snap_id, oid: &req->r_base_oid);
2083	ceph_oloc_copy(dest: &req->r_base_oloc, src: &rbd_dev->header_oloc);
2084	req->r_flags = CEPH_OSD_FLAG_WRITE;
2085	ktime_get_real_ts64(tv: &req->r_mtime);
2086
2087	if (snap_id == CEPH_NOSNAP) {
2088	/*
2089	* Protect against possible race conditions during lock
2090	* ownership transitions.
2091	*/
2092	ret = ceph_cls_assert_locked(req, which: which++, RBD_LOCK_NAME,
2093	type: CEPH_CLS_LOCK_EXCLUSIVE, cookie: "", tag: "");
2094	if (ret)
2095	return ret;
2096	}
2097
2098	ret = rbd_cls_object_map_update(req, which, objno: obj_req->ex.oe_objno,
2099	new_state, current_state);
2100	if (ret)
2101	return ret;
2102
2103	ret = ceph_osdc_alloc_messages(req, GFP_NOIO);
2104	if (ret)
2105	return ret;
2106
2107	ceph_osdc_start_request(osdc, req);
2108	return 0;
2109	}
2110
2111	static void prune_extents(struct ceph_file_extent *img_extents,
2112	u32 *num_img_extents, u64 overlap)
2113	{
2114	u32 cnt = *num_img_extents;
2115
2116	/* drop extents completely beyond the overlap */
2117	while (cnt && img_extents[cnt - 1].fe_off >= overlap)
2118	cnt--;
2119
2120	if (cnt) {
2121	struct ceph_file_extent *ex = &img_extents[cnt - 1];
2122
2123	/* trim final overlapping extent */
2124	if (ex->fe_off + ex->fe_len > overlap)
2125	ex->fe_len = overlap - ex->fe_off;
2126	}
2127
2128	*num_img_extents = cnt;
2129	}
2130
2131	/*
2132	* Determine the byte range(s) covered by either just the object extent
2133	* or the entire object in the parent image.
2134	*/
2135	static int rbd_obj_calc_img_extents(struct rbd_obj_request *obj_req,
2136	bool entire)
2137	{
2138	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2139	int ret;
2140
2141	if (!rbd_dev->parent_overlap)
2142	return 0;
2143
2144	ret = ceph_extent_to_file(l: &rbd_dev->layout, objno: obj_req->ex.oe_objno,
2145	objoff: entire ? 0 : obj_req->ex.oe_off,
2146	objlen: entire ? rbd_dev->layout.object_size :
2147	obj_req->ex.oe_len,
2148	file_extents: &obj_req->img_extents,
2149	num_file_extents: &obj_req->num_img_extents);
2150	if (ret)
2151	return ret;
2152
2153	prune_extents(img_extents: obj_req->img_extents, num_img_extents: &obj_req->num_img_extents,
2154	overlap: rbd_dev->parent_overlap);
2155	return 0;
2156	}
2157
2158	static void rbd_osd_setup_data(struct ceph_osd_request *osd_req, int which)
2159	{
2160	struct rbd_obj_request *obj_req = osd_req->r_priv;
2161
2162	switch (obj_req->img_request->data_type) {
2163	case OBJ_REQUEST_BIO:
2164	osd_req_op_extent_osd_data_bio(osd_req, which,
2165	bio_pos: &obj_req->bio_pos,
2166	bio_length: obj_req->ex.oe_len);
2167	break;
2168	case OBJ_REQUEST_BVECS:
2169	case OBJ_REQUEST_OWN_BVECS:
2170	rbd_assert(obj_req->bvec_pos.iter.bi_size ==
2171	obj_req->ex.oe_len);
2172	rbd_assert(obj_req->bvec_idx == obj_req->bvec_count);
2173	osd_req_op_extent_osd_data_bvec_pos(osd_req, which,
2174	bvec_pos: &obj_req->bvec_pos);
2175	break;
2176	default:
2177	BUG();
2178	}
2179	}
2180
2181	static int rbd_osd_setup_stat(struct ceph_osd_request *osd_req, int which)
2182	{
2183	struct page **pages;
2184
2185	/*
2186	* The response data for a STAT call consists of:
2187	* le64 length;
2188	* struct {
2189	* le32 tv_sec;
2190	* le32 tv_nsec;
2191	* } mtime;
2192	*/
2193	pages = ceph_alloc_page_vector(num_pages: 1, GFP_NOIO);
2194	if (IS_ERR(ptr: pages))
2195	return PTR_ERR(ptr: pages);
2196
2197	osd_req_op_init(osd_req, which, opcode: CEPH_OSD_OP_STAT, flags: 0);
2198	osd_req_op_raw_data_in_pages(osd_req, which, pages,
2199	length: 8 + sizeof(struct ceph_timespec),
2200	alignment: 0, pages_from_pool: false, own_pages: true);
2201	return 0;
2202	}
2203
2204	static int rbd_osd_setup_copyup(struct ceph_osd_request *osd_req, int which,
2205	u32 bytes)
2206	{
2207	struct rbd_obj_request *obj_req = osd_req->r_priv;
2208	int ret;
2209
2210	ret = osd_req_op_cls_init(osd_req, which, class: "rbd", method: "copyup");
2211	if (ret)
2212	return ret;
2213
2214	osd_req_op_cls_request_data_bvecs(osd_req, which, bvecs: obj_req->copyup_bvecs,
2215	num_bvecs: obj_req->copyup_bvec_count, bytes);
2216	return 0;
2217	}
2218
2219	static int rbd_obj_init_read(struct rbd_obj_request *obj_req)
2220	{
2221	obj_req->read_state = RBD_OBJ_READ_START;
2222	return 0;
2223	}
2224
2225	static void __rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2226	int which)
2227	{
2228	struct rbd_obj_request *obj_req = osd_req->r_priv;
2229	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2230	u16 opcode;
2231
2232	if (!use_object_map(rbd_dev) ||
2233	!(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST)) {
2234	osd_req_op_alloc_hint_init(osd_req, which: which++,
2235	expected_object_size: rbd_dev->layout.object_size,
2236	expected_write_size: rbd_dev->layout.object_size,
2237	flags: rbd_dev->opts->alloc_hint_flags);
2238	}
2239
2240	if (rbd_obj_is_entire(obj_req))
2241	opcode = CEPH_OSD_OP_WRITEFULL;
2242	else
2243	opcode = CEPH_OSD_OP_WRITE;
2244
2245	osd_req_op_extent_init(osd_req, which, opcode,
2246	offset: obj_req->ex.oe_off, length: obj_req->ex.oe_len, truncate_size: 0, truncate_seq: 0);
2247	rbd_osd_setup_data(osd_req, which);
2248	}
2249
2250	static int rbd_obj_init_write(struct rbd_obj_request *obj_req)
2251	{
2252	int ret;
2253
2254	/* reverse map the entire object onto the parent */
2255	ret = rbd_obj_calc_img_extents(obj_req, entire: true);
2256	if (ret)
2257	return ret;
2258
2259	obj_req->write_state = RBD_OBJ_WRITE_START;
2260	return 0;
2261	}
2262
2263	static u16 truncate_or_zero_opcode(struct rbd_obj_request *obj_req)
2264	{
2265	return rbd_obj_is_tail(obj_req) ? CEPH_OSD_OP_TRUNCATE :
2266	CEPH_OSD_OP_ZERO;
2267	}
2268
2269	static void __rbd_osd_setup_discard_ops(struct ceph_osd_request *osd_req,
2270	int which)
2271	{
2272	struct rbd_obj_request *obj_req = osd_req->r_priv;
2273
2274	if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
2275	rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2276	osd_req_op_init(osd_req, which, opcode: CEPH_OSD_OP_DELETE, flags: 0);
2277	} else {
2278	osd_req_op_extent_init(osd_req, which,
2279	opcode: truncate_or_zero_opcode(obj_req),
2280	offset: obj_req->ex.oe_off, length: obj_req->ex.oe_len,
2281	truncate_size: 0, truncate_seq: 0);
2282	}
2283	}
2284
2285	static int rbd_obj_init_discard(struct rbd_obj_request *obj_req)
2286	{
2287	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2288	u64 off, next_off;
2289	int ret;
2290
2291	/*
2292	* Align the range to alloc_size boundary and punt on discards
2293	* that are too small to free up any space.
2294	*
2295	* alloc_size == object_size && is_tail() is a special case for
2296	* filestore with filestore_punch_hole = false, needed to allow
2297	* truncate (in addition to delete).
2298	*/
2299	if (rbd_dev->opts->alloc_size != rbd_dev->layout.object_size ||
2300	!rbd_obj_is_tail(obj_req)) {
2301	off = round_up(obj_req->ex.oe_off, rbd_dev->opts->alloc_size);
2302	next_off = round_down(obj_req->ex.oe_off + obj_req->ex.oe_len,
2303	rbd_dev->opts->alloc_size);
2304	if (off >= next_off)
2305	return 1;
2306
2307	dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
2308	obj_req, obj_req->ex.oe_off, obj_req->ex.oe_len,
2309	off, next_off - off);
2310	obj_req->ex.oe_off = off;
2311	obj_req->ex.oe_len = next_off - off;
2312	}
2313
2314	/* reverse map the entire object onto the parent */
2315	ret = rbd_obj_calc_img_extents(obj_req, entire: true);
2316	if (ret)
2317	return ret;
2318
2319	obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2320	if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents)
2321	obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2322
2323	obj_req->write_state = RBD_OBJ_WRITE_START;
2324	return 0;
2325	}
2326
2327	static void __rbd_osd_setup_zeroout_ops(struct ceph_osd_request *osd_req,
2328	int which)
2329	{
2330	struct rbd_obj_request *obj_req = osd_req->r_priv;
2331	u16 opcode;
2332
2333	if (rbd_obj_is_entire(obj_req)) {
2334	if (obj_req->num_img_extents) {
2335	if (!(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2336	osd_req_op_init(osd_req, which: which++,
2337	opcode: CEPH_OSD_OP_CREATE, flags: 0);
2338	opcode = CEPH_OSD_OP_TRUNCATE;
2339	} else {
2340	rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2341	osd_req_op_init(osd_req, which: which++,
2342	opcode: CEPH_OSD_OP_DELETE, flags: 0);
2343	opcode = 0;
2344	}
2345	} else {
2346	opcode = truncate_or_zero_opcode(obj_req);
2347	}
2348
2349	if (opcode)
2350	osd_req_op_extent_init(osd_req, which, opcode,
2351	offset: obj_req->ex.oe_off, length: obj_req->ex.oe_len,
2352	truncate_size: 0, truncate_seq: 0);
2353	}
2354
2355	static int rbd_obj_init_zeroout(struct rbd_obj_request *obj_req)
2356	{
2357	int ret;
2358
2359	/* reverse map the entire object onto the parent */
2360	ret = rbd_obj_calc_img_extents(obj_req, entire: true);
2361	if (ret)
2362	return ret;
2363
2364	if (!obj_req->num_img_extents) {
2365	obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2366	if (rbd_obj_is_entire(obj_req))
2367	obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2368	}
2369
2370	obj_req->write_state = RBD_OBJ_WRITE_START;
2371	return 0;
2372	}
2373
2374	static int count_write_ops(struct rbd_obj_request *obj_req)
2375	{
2376	struct rbd_img_request *img_req = obj_req->img_request;
2377
2378	switch (img_req->op_type) {
2379	case OBJ_OP_WRITE:
2380	if (!use_object_map(rbd_dev: img_req->rbd_dev) ||
2381	!(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST))
2382	return 2; /* setallochint + write/writefull */
2383
2384	return 1; /* write/writefull */
2385	case OBJ_OP_DISCARD:
2386	return 1; /* delete/truncate/zero */
2387	case OBJ_OP_ZEROOUT:
2388	if (rbd_obj_is_entire(obj_req) && obj_req->num_img_extents &&
2389	!(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2390	return 2; /* create + truncate */
2391
2392	return 1; /* delete/truncate/zero */
2393	default:
2394	BUG();
2395	}
2396	}
2397
2398	static void rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2399	int which)
2400	{
2401	struct rbd_obj_request *obj_req = osd_req->r_priv;
2402
2403	switch (obj_req->img_request->op_type) {
2404	case OBJ_OP_WRITE:
2405	__rbd_osd_setup_write_ops(osd_req, which);
2406	break;
2407	case OBJ_OP_DISCARD:
2408	__rbd_osd_setup_discard_ops(osd_req, which);
2409	break;
2410	case OBJ_OP_ZEROOUT:
2411	__rbd_osd_setup_zeroout_ops(osd_req, which);
2412	break;
2413	default:
2414	BUG();
2415	}
2416	}
2417
2418	/*
2419	* Prune the list of object requests (adjust offset and/or length, drop
2420	* redundant requests). Prepare object request state machines and image
2421	* request state machine for execution.
2422	*/
2423	static int __rbd_img_fill_request(struct rbd_img_request *img_req)
2424	{
2425	struct rbd_obj_request *obj_req, *next_obj_req;
2426	int ret;
2427
2428	for_each_obj_request_safe(img_req, obj_req, next_obj_req) {
2429	switch (img_req->op_type) {
2430	case OBJ_OP_READ:
2431	ret = rbd_obj_init_read(obj_req);
2432	break;
2433	case OBJ_OP_WRITE:
2434	ret = rbd_obj_init_write(obj_req);
2435	break;
2436	case OBJ_OP_DISCARD:
2437	ret = rbd_obj_init_discard(obj_req);
2438	break;
2439	case OBJ_OP_ZEROOUT:
2440	ret = rbd_obj_init_zeroout(obj_req);
2441	break;
2442	default:
2443	BUG();
2444	}
2445	if (ret < 0)
2446	return ret;
2447	if (ret > 0) {
2448	rbd_img_obj_request_del(img_request: img_req, obj_request: obj_req);
2449	continue;
2450	}
2451	}
2452
2453	img_req->state = RBD_IMG_START;
2454	return 0;
2455	}
2456
2457	union rbd_img_fill_iter {
2458	struct ceph_bio_iter bio_iter;
2459	struct ceph_bvec_iter bvec_iter;
2460	};
2461
2462	struct rbd_img_fill_ctx {
2463	enum obj_request_type pos_type;
2464	union rbd_img_fill_iter *pos;
2465	union rbd_img_fill_iter iter;
2466	ceph_object_extent_fn_t set_pos_fn;
2467	ceph_object_extent_fn_t count_fn;
2468	ceph_object_extent_fn_t copy_fn;
2469	};
2470
2471	static struct ceph_object_extent *alloc_object_extent(void *arg)
2472	{
2473	struct rbd_img_request *img_req = arg;
2474	struct rbd_obj_request *obj_req;
2475
2476	obj_req = rbd_obj_request_create();
2477	if (!obj_req)
2478	return NULL;
2479
2480	rbd_img_obj_request_add(img_request: img_req, obj_request: obj_req);
2481	return &obj_req->ex;
2482	}
2483
2484	/*
2485	* While su != os && sc == 1 is technically not fancy (it's the same
2486	* layout as su == os && sc == 1), we can't use the nocopy path for it
2487	* because ->set_pos_fn() should be called only once per object.
2488	* ceph_file_to_extents() invokes action_fn once per stripe unit, so
2489	* treat su != os && sc == 1 as fancy.
2490	*/
2491	static bool rbd_layout_is_fancy(struct ceph_file_layout *l)
2492	{
2493	return l->stripe_unit != l->object_size;
2494	}
2495
2496	static int rbd_img_fill_request_nocopy(struct rbd_img_request *img_req,
2497	struct ceph_file_extent *img_extents,
2498	u32 num_img_extents,
2499	struct rbd_img_fill_ctx *fctx)
2500	{
2501	u32 i;
2502	int ret;
2503
2504	img_req->data_type = fctx->pos_type;
2505
2506	/*
2507	* Create object requests and set each object request's starting
2508	* position in the provided bio (list) or bio_vec array.
2509	*/
2510	fctx->iter = *fctx->pos;
2511	for (i = 0; i < num_img_extents; i++) {
2512	ret = ceph_file_to_extents(l: &img_req->rbd_dev->layout,
2513	off: img_extents[i].fe_off,
2514	len: img_extents[i].fe_len,
2515	object_extents: &img_req->object_extents,
2516	alloc_fn: alloc_object_extent, alloc_arg: img_req,
2517	action_fn: fctx->set_pos_fn, action_arg: &fctx->iter);
2518	if (ret)
2519	return ret;
2520	}
2521
2522	return __rbd_img_fill_request(img_req);
2523	}
2524
2525	/*
2526	* Map a list of image extents to a list of object extents, create the
2527	* corresponding object requests (normally each to a different object,
2528	* but not always) and add them to @img_req. For each object request,
2529	* set up its data descriptor to point to the corresponding chunk(s) of
2530	* @fctx->pos data buffer.
2531	*
2532	* Because ceph_file_to_extents() will merge adjacent object extents
2533	* together, each object request's data descriptor may point to multiple
2534	* different chunks of @fctx->pos data buffer.
2535	*
2536	* @fctx->pos data buffer is assumed to be large enough.
2537	*/
2538	static int rbd_img_fill_request(struct rbd_img_request *img_req,
2539	struct ceph_file_extent *img_extents,
2540	u32 num_img_extents,
2541	struct rbd_img_fill_ctx *fctx)
2542	{
2543	struct rbd_device *rbd_dev = img_req->rbd_dev;
2544	struct rbd_obj_request *obj_req;
2545	u32 i;
2546	int ret;
2547
2548	if (fctx->pos_type == OBJ_REQUEST_NODATA ||
2549	!rbd_layout_is_fancy(l: &rbd_dev->layout))
2550	return rbd_img_fill_request_nocopy(img_req, img_extents,
2551	num_img_extents, fctx);
2552
2553	img_req->data_type = OBJ_REQUEST_OWN_BVECS;
2554
2555	/*
2556	* Create object requests and determine ->bvec_count for each object
2557	* request. Note that ->bvec_count sum over all object requests may
2558	* be greater than the number of bio_vecs in the provided bio (list)
2559	* or bio_vec array because when mapped, those bio_vecs can straddle
2560	* stripe unit boundaries.
2561	*/
2562	fctx->iter = *fctx->pos;
2563	for (i = 0; i < num_img_extents; i++) {
2564	ret = ceph_file_to_extents(l: &rbd_dev->layout,
2565	off: img_extents[i].fe_off,
2566	len: img_extents[i].fe_len,
2567	object_extents: &img_req->object_extents,
2568	alloc_fn: alloc_object_extent, alloc_arg: img_req,
2569	action_fn: fctx->count_fn, action_arg: &fctx->iter);
2570	if (ret)
2571	return ret;
2572	}
2573
2574	for_each_obj_request(img_req, obj_req) {
2575	obj_req->bvec_pos.bvecs = kmalloc_array(n: obj_req->bvec_count,
2576	size: sizeof(*obj_req->bvec_pos.bvecs),
2577	GFP_NOIO);
2578	if (!obj_req->bvec_pos.bvecs)
2579	return -ENOMEM;
2580	}
2581
2582	/*
2583	* Fill in each object request's private bio_vec array, splitting and
2584	* rearranging the provided bio_vecs in stripe unit chunks as needed.
2585	*/
2586	fctx->iter = *fctx->pos;
2587	for (i = 0; i < num_img_extents; i++) {
2588	ret = ceph_iterate_extents(l: &rbd_dev->layout,
2589	off: img_extents[i].fe_off,
2590	len: img_extents[i].fe_len,
2591	object_extents: &img_req->object_extents,
2592	action_fn: fctx->copy_fn, action_arg: &fctx->iter);
2593	if (ret)
2594	return ret;
2595	}
2596
2597	return __rbd_img_fill_request(img_req);
2598	}
2599
2600	static int rbd_img_fill_nodata(struct rbd_img_request *img_req,
2601	u64 off, u64 len)
2602	{
2603	struct ceph_file_extent ex = { off, len };
2604	union rbd_img_fill_iter dummy = {};
2605	struct rbd_img_fill_ctx fctx = {
2606	.pos_type = OBJ_REQUEST_NODATA,
2607	.pos = &dummy,
2608	};
2609
2610	return rbd_img_fill_request(img_req, img_extents: &ex, num_img_extents: 1, fctx: &fctx);
2611	}
2612
2613	static void set_bio_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2614	{
2615	struct rbd_obj_request *obj_req =
2616	container_of(ex, struct rbd_obj_request, ex);
2617	struct ceph_bio_iter *it = arg;
2618
2619	dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2620	obj_req->bio_pos = *it;
2621	ceph_bio_iter_advance(it, bytes);
2622	}
2623
2624	static void count_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2625	{
2626	struct rbd_obj_request *obj_req =
2627	container_of(ex, struct rbd_obj_request, ex);
2628	struct ceph_bio_iter *it = arg;
2629
2630	dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2631	ceph_bio_iter_advance_step(it, bytes, ({
2632	obj_req->bvec_count++;
2633	}));
2634
2635	}
2636
2637	static void copy_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2638	{
2639	struct rbd_obj_request *obj_req =
2640	container_of(ex, struct rbd_obj_request, ex);
2641	struct ceph_bio_iter *it = arg;
2642
2643	dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2644	ceph_bio_iter_advance_step(it, bytes, ({
2645	obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2646	obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2647	}));
2648	}
2649
2650	static int __rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2651	struct ceph_file_extent *img_extents,
2652	u32 num_img_extents,
2653	struct ceph_bio_iter *bio_pos)
2654	{
2655	struct rbd_img_fill_ctx fctx = {
2656	.pos_type = OBJ_REQUEST_BIO,
2657	.pos = (union rbd_img_fill_iter *)bio_pos,
2658	.set_pos_fn = set_bio_pos,
2659	.count_fn = count_bio_bvecs,
2660	.copy_fn = copy_bio_bvecs,
2661	};
2662
2663	return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2664	fctx: &fctx);
2665	}
2666
2667	static int rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2668	u64 off, u64 len, struct bio *bio)
2669	{
2670	struct ceph_file_extent ex = { off, len };
2671	struct ceph_bio_iter it = { .bio = bio, .iter = bio->bi_iter };
2672
2673	return __rbd_img_fill_from_bio(img_req, img_extents: &ex, num_img_extents: 1, bio_pos: &it);
2674	}
2675
2676	static void set_bvec_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2677	{
2678	struct rbd_obj_request *obj_req =
2679	container_of(ex, struct rbd_obj_request, ex);
2680	struct ceph_bvec_iter *it = arg;
2681
2682	obj_req->bvec_pos = *it;
2683	ceph_bvec_iter_shorten(&obj_req->bvec_pos, bytes);
2684	ceph_bvec_iter_advance(it, bytes);
2685	}
2686
2687	static void count_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2688	{
2689	struct rbd_obj_request *obj_req =
2690	container_of(ex, struct rbd_obj_request, ex);
2691	struct ceph_bvec_iter *it = arg;
2692
2693	ceph_bvec_iter_advance_step(it, bytes, ({
2694	obj_req->bvec_count++;
2695	}));
2696	}
2697
2698	static void copy_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2699	{
2700	struct rbd_obj_request *obj_req =
2701	container_of(ex, struct rbd_obj_request, ex);
2702	struct ceph_bvec_iter *it = arg;
2703
2704	ceph_bvec_iter_advance_step(it, bytes, ({
2705	obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2706	obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2707	}));
2708	}
2709
2710	static int __rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2711	struct ceph_file_extent *img_extents,
2712	u32 num_img_extents,
2713	struct ceph_bvec_iter *bvec_pos)
2714	{
2715	struct rbd_img_fill_ctx fctx = {
2716	.pos_type = OBJ_REQUEST_BVECS,
2717	.pos = (union rbd_img_fill_iter *)bvec_pos,
2718	.set_pos_fn = set_bvec_pos,
2719	.count_fn = count_bvecs,
2720	.copy_fn = copy_bvecs,
2721	};
2722
2723	return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2724	fctx: &fctx);
2725	}
2726
2727	static int rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2728	struct ceph_file_extent *img_extents,
2729	u32 num_img_extents,
2730	struct bio_vec *bvecs)
2731	{
2732	struct ceph_bvec_iter it = {
2733	.bvecs = bvecs,
2734	.iter = { .bi_size = ceph_file_extents_bytes(file_extents: img_extents,
2735	num_file_extents: num_img_extents) },
2736	};
2737
2738	return __rbd_img_fill_from_bvecs(img_req, img_extents, num_img_extents,
2739	bvec_pos: &it);
2740	}
2741
2742	static void rbd_img_handle_request_work(struct work_struct *work)
2743	{
2744	struct rbd_img_request *img_req =
2745	container_of(work, struct rbd_img_request, work);
2746
2747	rbd_img_handle_request(img_req, result: img_req->work_result);
2748	}
2749
2750	static void rbd_img_schedule(struct rbd_img_request *img_req, int result)
2751	{
2752	INIT_WORK(&img_req->work, rbd_img_handle_request_work);
2753	img_req->work_result = result;
2754	queue_work(wq: rbd_wq, work: &img_req->work);
2755	}
2756
2757	static bool rbd_obj_may_exist(struct rbd_obj_request *obj_req)
2758	{
2759	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2760
2761	if (rbd_object_map_may_exist(rbd_dev, objno: obj_req->ex.oe_objno)) {
2762	obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2763	return true;
2764	}
2765
2766	dout("%s %p objno %llu assuming dne\n", __func__, obj_req,
2767	obj_req->ex.oe_objno);
2768	return false;
2769	}
2770
2771	static int rbd_obj_read_object(struct rbd_obj_request *obj_req)
2772	{
2773	struct ceph_osd_request *osd_req;
2774	int ret;
2775
2776	osd_req = __rbd_obj_add_osd_request(obj_req, NULL, num_ops: 1);
2777	if (IS_ERR(ptr: osd_req))
2778	return PTR_ERR(ptr: osd_req);
2779
2780	osd_req_op_extent_init(osd_req, which: 0, opcode: CEPH_OSD_OP_READ,
2781	offset: obj_req->ex.oe_off, length: obj_req->ex.oe_len, truncate_size: 0, truncate_seq: 0);
2782	rbd_osd_setup_data(osd_req, which: 0);
2783	rbd_osd_format_read(osd_req);
2784
2785	ret = ceph_osdc_alloc_messages(req: osd_req, GFP_NOIO);
2786	if (ret)
2787	return ret;
2788
2789	rbd_osd_submit(osd_req);
2790	return 0;
2791	}
2792
2793	static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
2794	{
2795	struct rbd_img_request *img_req = obj_req->img_request;
2796	struct rbd_device *parent = img_req->rbd_dev->parent;
2797	struct rbd_img_request *child_img_req;
2798	int ret;
2799
2800	child_img_req = kmem_cache_alloc(cachep: rbd_img_request_cache, GFP_NOIO);
2801	if (!child_img_req)
2802	return -ENOMEM;
2803
2804	rbd_img_request_init(img_request: child_img_req, rbd_dev: parent, op_type: OBJ_OP_READ);
2805	__set_bit(IMG_REQ_CHILD, &child_img_req->flags);
2806	child_img_req->obj_request = obj_req;
2807
2808	down_read(sem: &parent->header_rwsem);
2809	rbd_img_capture_header(img_req: child_img_req);
2810	up_read(sem: &parent->header_rwsem);
2811
2812	dout("%s child_img_req %p for obj_req %p\n", __func__, child_img_req,
2813	obj_req);
2814
2815	if (!rbd_img_is_write(img_req)) {
2816	switch (img_req->data_type) {
2817	case OBJ_REQUEST_BIO:
2818	ret = __rbd_img_fill_from_bio(img_req: child_img_req,
2819	img_extents: obj_req->img_extents,
2820	num_img_extents: obj_req->num_img_extents,
2821	bio_pos: &obj_req->bio_pos);
2822	break;
2823	case OBJ_REQUEST_BVECS:
2824	case OBJ_REQUEST_OWN_BVECS:
2825	ret = __rbd_img_fill_from_bvecs(img_req: child_img_req,
2826	img_extents: obj_req->img_extents,
2827	num_img_extents: obj_req->num_img_extents,
2828	bvec_pos: &obj_req->bvec_pos);
2829	break;
2830	default:
2831	BUG();
2832	}
2833	} else {
2834	ret = rbd_img_fill_from_bvecs(img_req: child_img_req,
2835	img_extents: obj_req->img_extents,
2836	num_img_extents: obj_req->num_img_extents,
2837	bvecs: obj_req->copyup_bvecs);
2838	}
2839	if (ret) {
2840	rbd_img_request_destroy(img_request: child_img_req);
2841	return ret;
2842	}
2843
2844	/* avoid parent chain recursion */
2845	rbd_img_schedule(img_req: child_img_req, result: 0);
2846	return 0;
2847	}
2848
2849	static bool rbd_obj_advance_read(struct rbd_obj_request *obj_req, int *result)
2850	{
2851	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2852	int ret;
2853
2854	again:
2855	switch (obj_req->read_state) {
2856	case RBD_OBJ_READ_START:
2857	rbd_assert(!*result);
2858
2859	if (!rbd_obj_may_exist(obj_req)) {
2860	*result = -ENOENT;
2861	obj_req->read_state = RBD_OBJ_READ_OBJECT;
2862	goto again;
2863	}
2864
2865	ret = rbd_obj_read_object(obj_req);
2866	if (ret) {
2867	*result = ret;
2868	return true;
2869	}
2870	obj_req->read_state = RBD_OBJ_READ_OBJECT;
2871	return false;
2872	case RBD_OBJ_READ_OBJECT:
2873	if (*result == -ENOENT && rbd_dev->parent_overlap) {
2874	/* reverse map this object extent onto the parent */
2875	ret = rbd_obj_calc_img_extents(obj_req, entire: false);
2876	if (ret) {
2877	*result = ret;
2878	return true;
2879	}
2880	if (obj_req->num_img_extents) {
2881	ret = rbd_obj_read_from_parent(obj_req);
2882	if (ret) {
2883	*result = ret;
2884	return true;
2885	}
2886	obj_req->read_state = RBD_OBJ_READ_PARENT;
2887	return false;
2888	}
2889	}
2890
2891	/*
2892	* -ENOENT means a hole in the image -- zero-fill the entire
2893	* length of the request. A short read also implies zero-fill
2894	* to the end of the request.
2895	*/
2896	if (*result == -ENOENT) {
2897	rbd_obj_zero_range(obj_req, off: 0, bytes: obj_req->ex.oe_len);
2898	*result = 0;
2899	} else if (*result >= 0) {
2900	if (*result < obj_req->ex.oe_len)
2901	rbd_obj_zero_range(obj_req, off: *result,
2902	bytes: obj_req->ex.oe_len - *result);
2903	else
2904	rbd_assert(*result == obj_req->ex.oe_len);
2905	*result = 0;
2906	}
2907	return true;
2908	case RBD_OBJ_READ_PARENT:
2909	/*
2910	* The parent image is read only up to the overlap -- zero-fill
2911	* from the overlap to the end of the request.
2912	*/
2913	if (!*result) {
2914	u32 obj_overlap = rbd_obj_img_extents_bytes(obj_req);
2915
2916	if (obj_overlap < obj_req->ex.oe_len)
2917	rbd_obj_zero_range(obj_req, off: obj_overlap,
2918	bytes: obj_req->ex.oe_len - obj_overlap);
2919	}
2920	return true;
2921	default:
2922	BUG();
2923	}
2924	}
2925
2926	static bool rbd_obj_write_is_noop(struct rbd_obj_request *obj_req)
2927	{
2928	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2929
2930	if (rbd_object_map_may_exist(rbd_dev, objno: obj_req->ex.oe_objno))
2931	obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2932
2933	if (!(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST) &&
2934	(obj_req->flags & RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT)) {
2935	dout("%s %p noop for nonexistent\n", __func__, obj_req);
2936	return true;
2937	}
2938
2939	return false;
2940	}
2941
2942	/*
2943	* Return:
2944	* 0 - object map update sent
2945	* 1 - object map update isn't needed
2946	* <0 - error
2947	*/
2948	static int rbd_obj_write_pre_object_map(struct rbd_obj_request *obj_req)
2949	{
2950	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2951	u8 new_state;
2952
2953	if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
2954	return 1;
2955
2956	if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
2957	new_state = OBJECT_PENDING;
2958	else
2959	new_state = OBJECT_EXISTS;
2960
2961	return rbd_object_map_update(obj_req, CEPH_NOSNAP, new_state, NULL);
2962	}
2963
2964	static int rbd_obj_write_object(struct rbd_obj_request *obj_req)
2965	{
2966	struct ceph_osd_request *osd_req;
2967	int num_ops = count_write_ops(obj_req);
2968	int which = 0;
2969	int ret;
2970
2971	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
2972	num_ops++; /* stat */
2973
2974	osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
2975	if (IS_ERR(ptr: osd_req))
2976	return PTR_ERR(ptr: osd_req);
2977
2978	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
2979	ret = rbd_osd_setup_stat(osd_req, which: which++);
2980	if (ret)
2981	return ret;
2982	}
2983
2984	rbd_osd_setup_write_ops(osd_req, which);
2985	rbd_osd_format_write(osd_req);
2986
2987	ret = ceph_osdc_alloc_messages(req: osd_req, GFP_NOIO);
2988	if (ret)
2989	return ret;
2990
2991	rbd_osd_submit(osd_req);
2992	return 0;
2993	}
2994
2995	/*
2996	* copyup_bvecs pages are never highmem pages
2997	*/
2998	static bool is_zero_bvecs(struct bio_vec *bvecs, u32 bytes)
2999	{
3000	struct ceph_bvec_iter it = {
3001	.bvecs = bvecs,
3002	.iter = { .bi_size = bytes },
3003	};
3004
3005	ceph_bvec_iter_advance_step(&it, bytes, ({
3006	if (memchr_inv(bvec_virt(&bv), 0, bv.bv_len))
3007	return false;
3008	}));
3009	return true;
3010	}
3011
3012	#define MODS_ONLY U32_MAX
3013
3014	static int rbd_obj_copyup_empty_snapc(struct rbd_obj_request *obj_req,
3015	u32 bytes)
3016	{
3017	struct ceph_osd_request *osd_req;
3018	int ret;
3019
3020	dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3021	rbd_assert(bytes > 0 && bytes != MODS_ONLY);
3022
3023	osd_req = __rbd_obj_add_osd_request(obj_req, snapc: &rbd_empty_snapc, num_ops: 1);
3024	if (IS_ERR(ptr: osd_req))
3025	return PTR_ERR(ptr: osd_req);
3026
3027	ret = rbd_osd_setup_copyup(osd_req, which: 0, bytes);
3028	if (ret)
3029	return ret;
3030
3031	rbd_osd_format_write(osd_req);
3032
3033	ret = ceph_osdc_alloc_messages(req: osd_req, GFP_NOIO);
3034	if (ret)
3035	return ret;
3036
3037	rbd_osd_submit(osd_req);
3038	return 0;
3039	}
3040
3041	static int rbd_obj_copyup_current_snapc(struct rbd_obj_request *obj_req,
3042	u32 bytes)
3043	{
3044	struct ceph_osd_request *osd_req;
3045	int num_ops = count_write_ops(obj_req);
3046	int which = 0;
3047	int ret;
3048
3049	dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3050
3051	if (bytes != MODS_ONLY)
3052	num_ops++; /* copyup */
3053
3054	osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
3055	if (IS_ERR(ptr: osd_req))
3056	return PTR_ERR(ptr: osd_req);
3057
3058	if (bytes != MODS_ONLY) {
3059	ret = rbd_osd_setup_copyup(osd_req, which: which++, bytes);
3060	if (ret)
3061	return ret;
3062	}
3063
3064	rbd_osd_setup_write_ops(osd_req, which);
3065	rbd_osd_format_write(osd_req);
3066
3067	ret = ceph_osdc_alloc_messages(req: osd_req, GFP_NOIO);
3068	if (ret)
3069	return ret;
3070
3071	rbd_osd_submit(osd_req);
3072	return 0;
3073	}
3074
3075	static int setup_copyup_bvecs(struct rbd_obj_request *obj_req, u64 obj_overlap)
3076	{
3077	u32 i;
3078
3079	rbd_assert(!obj_req->copyup_bvecs);
3080	obj_req->copyup_bvec_count = calc_pages_for(off: 0, len: obj_overlap);
3081	obj_req->copyup_bvecs = kcalloc(n: obj_req->copyup_bvec_count,
3082	size: sizeof(*obj_req->copyup_bvecs),
3083	GFP_NOIO);
3084	if (!obj_req->copyup_bvecs)
3085	return -ENOMEM;
3086
3087	for (i = 0; i < obj_req->copyup_bvec_count; i++) {
3088	unsigned int len = min(obj_overlap, (u64)PAGE_SIZE);
3089	struct page *page = alloc_page(GFP_NOIO);
3090
3091	if (!page)
3092	return -ENOMEM;
3093
3094	bvec_set_page(bv: &obj_req->copyup_bvecs[i], page, len, offset: 0);
3095	obj_overlap -= len;
3096	}
3097
3098	rbd_assert(!obj_overlap);
3099	return 0;
3100	}
3101
3102	/*
3103	* The target object doesn't exist. Read the data for the entire
3104	* target object up to the overlap point (if any) from the parent,
3105	* so we can use it for a copyup.
3106	*/
3107	static int rbd_obj_copyup_read_parent(struct rbd_obj_request *obj_req)
3108	{
3109	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3110	int ret;
3111
3112	rbd_assert(obj_req->num_img_extents);
3113	prune_extents(img_extents: obj_req->img_extents, num_img_extents: &obj_req->num_img_extents,
3114	overlap: rbd_dev->parent_overlap);
3115	if (!obj_req->num_img_extents) {
3116	/*
3117	* The overlap has become 0 (most likely because the
3118	* image has been flattened). Re-submit the original write
3119	* request -- pass MODS_ONLY since the copyup isn't needed
3120	* anymore.
3121	*/
3122	return rbd_obj_copyup_current_snapc(obj_req, MODS_ONLY);
3123	}
3124
3125	ret = setup_copyup_bvecs(obj_req, obj_overlap: rbd_obj_img_extents_bytes(obj_req));
3126	if (ret)
3127	return ret;
3128
3129	return rbd_obj_read_from_parent(obj_req);
3130	}
3131
3132	static void rbd_obj_copyup_object_maps(struct rbd_obj_request *obj_req)
3133	{
3134	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3135	struct ceph_snap_context *snapc = obj_req->img_request->snapc;
3136	u8 new_state;
3137	u32 i;
3138	int ret;
3139
3140	rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3141
3142	if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3143	return;
3144
3145	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3146	return;
3147
3148	for (i = 0; i < snapc->num_snaps; i++) {
3149	if ((rbd_dev->header.features & RBD_FEATURE_FAST_DIFF) &&
3150	i + 1 < snapc->num_snaps)
3151	new_state = OBJECT_EXISTS_CLEAN;
3152	else
3153	new_state = OBJECT_EXISTS;
3154
3155	ret = rbd_object_map_update(obj_req, snap_id: snapc->snaps[i],
3156	new_state, NULL);
3157	if (ret < 0) {
3158	obj_req->pending.result = ret;
3159	return;
3160	}
3161
3162	rbd_assert(!ret);
3163	obj_req->pending.num_pending++;
3164	}
3165	}
3166
3167	static void rbd_obj_copyup_write_object(struct rbd_obj_request *obj_req)
3168	{
3169	u32 bytes = rbd_obj_img_extents_bytes(obj_req);
3170	int ret;
3171
3172	rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3173
3174	/*
3175	* Only send non-zero copyup data to save some I/O and network
3176	* bandwidth -- zero copyup data is equivalent to the object not
3177	* existing.
3178	*/
3179	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3180	bytes = 0;
3181
3182	if (obj_req->img_request->snapc->num_snaps && bytes > 0) {
3183	/*
3184	* Send a copyup request with an empty snapshot context to
3185	* deep-copyup the object through all existing snapshots.
3186	* A second request with the current snapshot context will be
3187	* sent for the actual modification.
3188	*/
3189	ret = rbd_obj_copyup_empty_snapc(obj_req, bytes);
3190	if (ret) {
3191	obj_req->pending.result = ret;
3192	return;
3193	}
3194
3195	obj_req->pending.num_pending++;
3196	bytes = MODS_ONLY;
3197	}
3198
3199	ret = rbd_obj_copyup_current_snapc(obj_req, bytes);
3200	if (ret) {
3201	obj_req->pending.result = ret;
3202	return;
3203	}
3204
3205	obj_req->pending.num_pending++;
3206	}
3207
3208	static bool rbd_obj_advance_copyup(struct rbd_obj_request *obj_req, int *result)
3209	{
3210	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3211	int ret;
3212
3213	again:
3214	switch (obj_req->copyup_state) {
3215	case RBD_OBJ_COPYUP_START:
3216	rbd_assert(!*result);
3217
3218	ret = rbd_obj_copyup_read_parent(obj_req);
3219	if (ret) {
3220	*result = ret;
3221	return true;
3222	}
3223	if (obj_req->num_img_extents)
3224	obj_req->copyup_state = RBD_OBJ_COPYUP_READ_PARENT;
3225	else
3226	obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3227	return false;
3228	case RBD_OBJ_COPYUP_READ_PARENT:
3229	if (*result)
3230	return true;
3231
3232	if (is_zero_bvecs(bvecs: obj_req->copyup_bvecs,
3233	bytes: rbd_obj_img_extents_bytes(obj_req))) {
3234	dout("%s %p detected zeros\n", __func__, obj_req);
3235	obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ZEROS;
3236	}
3237
3238	rbd_obj_copyup_object_maps(obj_req);
3239	if (!obj_req->pending.num_pending) {
3240	*result = obj_req->pending.result;
3241	obj_req->copyup_state = RBD_OBJ_COPYUP_OBJECT_MAPS;
3242	goto again;
3243	}
3244	obj_req->copyup_state = __RBD_OBJ_COPYUP_OBJECT_MAPS;
3245	return false;
3246	case __RBD_OBJ_COPYUP_OBJECT_MAPS:
3247	if (!pending_result_dec(pending: &obj_req->pending, result))
3248	return false;
3249	fallthrough;
3250	case RBD_OBJ_COPYUP_OBJECT_MAPS:
3251	if (*result) {
3252	rbd_warn(rbd_dev, fmt: "snap object map update failed: %d",
3253	*result);
3254	return true;
3255	}
3256
3257	rbd_obj_copyup_write_object(obj_req);
3258	if (!obj_req->pending.num_pending) {
3259	*result = obj_req->pending.result;
3260	obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3261	goto again;
3262	}
3263	obj_req->copyup_state = __RBD_OBJ_COPYUP_WRITE_OBJECT;
3264	return false;
3265	case __RBD_OBJ_COPYUP_WRITE_OBJECT:
3266	if (!pending_result_dec(pending: &obj_req->pending, result))
3267	return false;
3268	fallthrough;
3269	case RBD_OBJ_COPYUP_WRITE_OBJECT:
3270	return true;
3271	default:
3272	BUG();
3273	}
3274	}
3275
3276	/*
3277	* Return:
3278	* 0 - object map update sent
3279	* 1 - object map update isn't needed
3280	* <0 - error
3281	*/
3282	static int rbd_obj_write_post_object_map(struct rbd_obj_request *obj_req)
3283	{
3284	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3285	u8 current_state = OBJECT_PENDING;
3286
3287	if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3288	return 1;
3289
3290	if (!(obj_req->flags & RBD_OBJ_FLAG_DELETION))
3291	return 1;
3292
3293	return rbd_object_map_update(obj_req, CEPH_NOSNAP, OBJECT_NONEXISTENT,
3294	current_state: &current_state);
3295	}
3296
3297	static bool rbd_obj_advance_write(struct rbd_obj_request *obj_req, int *result)
3298	{
3299	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3300	int ret;
3301
3302	again:
3303	switch (obj_req->write_state) {
3304	case RBD_OBJ_WRITE_START:
3305	rbd_assert(!*result);
3306
3307	rbd_obj_set_copyup_enabled(obj_req);
3308	if (rbd_obj_write_is_noop(obj_req))
3309	return true;
3310
3311	ret = rbd_obj_write_pre_object_map(obj_req);
3312	if (ret < 0) {
3313	*result = ret;
3314	return true;
3315	}
3316	obj_req->write_state = RBD_OBJ_WRITE_PRE_OBJECT_MAP;
3317	if (ret > 0)
3318	goto again;
3319	return false;
3320	case RBD_OBJ_WRITE_PRE_OBJECT_MAP:
3321	if (*result) {
3322	rbd_warn(rbd_dev, fmt: "pre object map update failed: %d",
3323	*result);
3324	return true;
3325	}
3326	ret = rbd_obj_write_object(obj_req);
3327	if (ret) {
3328	*result = ret;
3329	return true;
3330	}
3331	obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
3332	return false;
3333	case RBD_OBJ_WRITE_OBJECT:
3334	if (*result == -ENOENT) {
3335	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
3336	*result = 0;
3337	obj_req->copyup_state = RBD_OBJ_COPYUP_START;
3338	obj_req->write_state = __RBD_OBJ_WRITE_COPYUP;
3339	goto again;
3340	}
3341	/*
3342	* On a non-existent object:
3343	* delete - -ENOENT, truncate/zero - 0
3344	*/
3345	if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
3346	*result = 0;
3347	}
3348	if (*result)
3349	return true;
3350
3351	obj_req->write_state = RBD_OBJ_WRITE_COPYUP;
3352	goto again;
3353	case __RBD_OBJ_WRITE_COPYUP:
3354	if (!rbd_obj_advance_copyup(obj_req, result))
3355	return false;
3356	fallthrough;
3357	case RBD_OBJ_WRITE_COPYUP:
3358	if (*result) {
3359	rbd_warn(rbd_dev, fmt: "copyup failed: %d", *result);
3360	return true;
3361	}
3362	ret = rbd_obj_write_post_object_map(obj_req);
3363	if (ret < 0) {
3364	*result = ret;
3365	return true;
3366	}
3367	obj_req->write_state = RBD_OBJ_WRITE_POST_OBJECT_MAP;
3368	if (ret > 0)
3369	goto again;
3370	return false;
3371	case RBD_OBJ_WRITE_POST_OBJECT_MAP:
3372	if (*result)
3373	rbd_warn(rbd_dev, fmt: "post object map update failed: %d",
3374	*result);
3375	return true;
3376	default:
3377	BUG();
3378	}
3379	}
3380
3381	/*
3382	* Return true if @obj_req is completed.
3383	*/
3384	static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
3385	int *result)
3386	{
3387	struct rbd_img_request *img_req = obj_req->img_request;
3388	struct rbd_device *rbd_dev = img_req->rbd_dev;
3389	bool done;
3390
3391	mutex_lock(&obj_req->state_mutex);
3392	if (!rbd_img_is_write(img_req))
3393	done = rbd_obj_advance_read(obj_req, result);
3394	else
3395	done = rbd_obj_advance_write(obj_req, result);
3396	mutex_unlock(lock: &obj_req->state_mutex);
3397
3398	if (done && *result) {
3399	rbd_assert(*result < 0);
3400	rbd_warn(rbd_dev, fmt: "%s at objno %llu %llu~%llu result %d",
3401	obj_op_name(op_type: img_req->op_type), obj_req->ex.oe_objno,
3402	obj_req->ex.oe_off, obj_req->ex.oe_len, *result);
3403	}
3404	return done;
3405	}
3406
3407	/*
3408	* This is open-coded in rbd_img_handle_request() to avoid parent chain
3409	* recursion.
3410	*/
3411	static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result)
3412	{
3413	if (__rbd_obj_handle_request(obj_req, result: &result))
3414	rbd_img_handle_request(img_req: obj_req->img_request, result);
3415	}
3416
3417	static bool need_exclusive_lock(struct rbd_img_request *img_req)
3418	{
3419	struct rbd_device *rbd_dev = img_req->rbd_dev;
3420
3421	if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK))
3422	return false;
3423
3424	if (rbd_is_ro(rbd_dev))
3425	return false;
3426
3427	rbd_assert(!test_bit(IMG_REQ_CHILD, &img_req->flags));
3428	if (rbd_dev->opts->lock_on_read ||
3429	(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3430	return true;
3431
3432	return rbd_img_is_write(img_req);
3433	}
3434
3435	static bool rbd_lock_add_request(struct rbd_img_request *img_req)
3436	{
3437	struct rbd_device *rbd_dev = img_req->rbd_dev;
3438	bool locked;
3439
3440	lockdep_assert_held(&rbd_dev->lock_rwsem);
3441	locked = rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED;
3442	spin_lock(lock: &rbd_dev->lock_lists_lock);
3443	rbd_assert(list_empty(&img_req->lock_item));
3444	if (!locked)
3445	list_add_tail(new: &img_req->lock_item, head: &rbd_dev->acquiring_list);
3446	else
3447	list_add_tail(new: &img_req->lock_item, head: &rbd_dev->running_list);
3448	spin_unlock(lock: &rbd_dev->lock_lists_lock);
3449	return locked;
3450	}
3451
3452	static void rbd_lock_del_request(struct rbd_img_request *img_req)
3453	{
3454	struct rbd_device *rbd_dev = img_req->rbd_dev;
3455	bool need_wakeup;
3456
3457	lockdep_assert_held(&rbd_dev->lock_rwsem);
3458	spin_lock(lock: &rbd_dev->lock_lists_lock);
3459	rbd_assert(!list_empty(&img_req->lock_item));
3460	list_del_init(entry: &img_req->lock_item);
3461	need_wakeup = (rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING &&
3462	list_empty(head: &rbd_dev->running_list));
3463	spin_unlock(lock: &rbd_dev->lock_lists_lock);
3464	if (need_wakeup)
3465	complete(&rbd_dev->releasing_wait);
3466	}
3467
3468	static int rbd_img_exclusive_lock(struct rbd_img_request *img_req)
3469	{
3470	struct rbd_device *rbd_dev = img_req->rbd_dev;
3471
3472	if (!need_exclusive_lock(img_req))
3473	return 1;
3474
3475	if (rbd_lock_add_request(img_req))
3476	return 1;
3477
3478	if (rbd_dev->opts->exclusive) {
3479	WARN_ON(1); /* lock got released? */
3480	return -EROFS;
3481	}
3482
3483	/*
3484	* Note the use of mod_delayed_work() in rbd_acquire_lock()
3485	* and cancel_delayed_work() in wake_lock_waiters().
3486	*/
3487	dout("%s rbd_dev %p queueing lock_dwork\n", __func__, rbd_dev);
3488	queue_delayed_work(wq: rbd_dev->task_wq, dwork: &rbd_dev->lock_dwork, delay: 0);
3489	return 0;
3490	}
3491
3492	static void rbd_img_object_requests(struct rbd_img_request *img_req)
3493	{
3494	struct rbd_device *rbd_dev = img_req->rbd_dev;
3495	struct rbd_obj_request *obj_req;
3496
3497	rbd_assert(!img_req->pending.result && !img_req->pending.num_pending);
3498	rbd_assert(!need_exclusive_lock(img_req) ||
3499	__rbd_is_lock_owner(rbd_dev));
3500
3501	if (rbd_img_is_write(img_req)) {
3502	rbd_assert(!img_req->snapc);
3503	down_read(sem: &rbd_dev->header_rwsem);
3504	img_req->snapc = ceph_get_snap_context(sc: rbd_dev->header.snapc);
3505	up_read(sem: &rbd_dev->header_rwsem);
3506	}
3507
3508	for_each_obj_request(img_req, obj_req) {
3509	int result = 0;
3510
3511	if (__rbd_obj_handle_request(obj_req, result: &result)) {
3512	if (result) {
3513	img_req->pending.result = result;
3514	return;
3515	}
3516	} else {
3517	img_req->pending.num_pending++;
3518	}
3519	}
3520	}
3521
3522	static bool rbd_img_advance(struct rbd_img_request *img_req, int *result)
3523	{
3524	int ret;
3525
3526	again:
3527	switch (img_req->state) {
3528	case RBD_IMG_START:
3529	rbd_assert(!*result);
3530
3531	ret = rbd_img_exclusive_lock(img_req);
3532	if (ret < 0) {
3533	*result = ret;
3534	return true;
3535	}
3536	img_req->state = RBD_IMG_EXCLUSIVE_LOCK;
3537	if (ret > 0)
3538	goto again;
3539	return false;
3540	case RBD_IMG_EXCLUSIVE_LOCK:
3541	if (*result)
3542	return true;
3543
3544	rbd_img_object_requests(img_req);
3545	if (!img_req->pending.num_pending) {
3546	*result = img_req->pending.result;
3547	img_req->state = RBD_IMG_OBJECT_REQUESTS;
3548	goto again;
3549	}
3550	img_req->state = __RBD_IMG_OBJECT_REQUESTS;
3551	return false;
3552	case __RBD_IMG_OBJECT_REQUESTS:
3553	if (!pending_result_dec(pending: &img_req->pending, result))
3554	return false;
3555	fallthrough;
3556	case RBD_IMG_OBJECT_REQUESTS:
3557	return true;
3558	default:
3559	BUG();
3560	}
3561	}
3562
3563	/*
3564	* Return true if @img_req is completed.
3565	*/
3566	static bool __rbd_img_handle_request(struct rbd_img_request *img_req,
3567	int *result)
3568	{
3569	struct rbd_device *rbd_dev = img_req->rbd_dev;
3570	bool done;
3571
3572	if (need_exclusive_lock(img_req)) {
3573	down_read(sem: &rbd_dev->lock_rwsem);
3574	mutex_lock(&img_req->state_mutex);
3575	done = rbd_img_advance(img_req, result);
3576	if (done)
3577	rbd_lock_del_request(img_req);
3578	mutex_unlock(lock: &img_req->state_mutex);
3579	up_read(sem: &rbd_dev->lock_rwsem);
3580	} else {
3581	mutex_lock(&img_req->state_mutex);
3582	done = rbd_img_advance(img_req, result);
3583	mutex_unlock(lock: &img_req->state_mutex);
3584	}
3585
3586	if (done && *result) {
3587	rbd_assert(*result < 0);
3588	rbd_warn(rbd_dev, fmt: "%s%s result %d",
3589	test_bit(IMG_REQ_CHILD, &img_req->flags) ? "child " : "",
3590	obj_op_name(op_type: img_req->op_type), *result);
3591	}
3592	return done;
3593	}
3594
3595	static void rbd_img_handle_request(struct rbd_img_request *img_req, int result)
3596	{
3597	again:
3598	if (!__rbd_img_handle_request(img_req, result: &result))
3599	return;
3600
3601	if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
3602	struct rbd_obj_request *obj_req = img_req->obj_request;
3603
3604	rbd_img_request_destroy(img_request: img_req);
3605	if (__rbd_obj_handle_request(obj_req, result: &result)) {
3606	img_req = obj_req->img_request;
3607	goto again;
3608	}
3609	} else {
3610	struct request *rq = blk_mq_rq_from_pdu(pdu: img_req);
3611
3612	rbd_img_request_destroy(img_request: img_req);
3613	blk_mq_end_request(rq, error: errno_to_blk_status(errno: result));
3614	}
3615	}
3616
3617	static const struct rbd_client_id rbd_empty_cid;
3618
3619	static bool rbd_cid_equal(const struct rbd_client_id *lhs,
3620	const struct rbd_client_id *rhs)
3621	{
3622	return lhs->gid == rhs->gid && lhs->handle == rhs->handle;
3623	}
3624
3625	static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev)
3626	{
3627	struct rbd_client_id cid;
3628
3629	mutex_lock(&rbd_dev->watch_mutex);
3630	cid.gid = ceph_client_gid(client: rbd_dev->rbd_client->client);
3631	cid.handle = rbd_dev->watch_cookie;
3632	mutex_unlock(lock: &rbd_dev->watch_mutex);
3633	return cid;
3634	}
3635
3636	/*
3637	* lock_rwsem must be held for write
3638	*/
3639	static void rbd_set_owner_cid(struct rbd_device *rbd_dev,
3640	const struct rbd_client_id *cid)
3641	{
3642	dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev,
3643	rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle,
3644	cid->gid, cid->handle);
3645	rbd_dev->owner_cid = *cid; /* struct */
3646	}
3647
3648	static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf)
3649	{
3650	mutex_lock(&rbd_dev->watch_mutex);
3651	sprintf(buf, fmt: "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie);
3652	mutex_unlock(lock: &rbd_dev->watch_mutex);
3653	}
3654
3655	static void __rbd_lock(struct rbd_device *rbd_dev, const char *cookie)
3656	{
3657	struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3658
3659	rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED;
3660	strcpy(p: rbd_dev->lock_cookie, q: cookie);
3661	rbd_set_owner_cid(rbd_dev, cid: &cid);
3662	queue_work(wq: rbd_dev->task_wq, work: &rbd_dev->acquired_lock_work);
3663	}
3664
3665	/*
3666	* lock_rwsem must be held for write
3667	*/
3668	static int rbd_lock(struct rbd_device *rbd_dev)
3669	{
3670	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3671	char cookie[32];
3672	int ret;
3673
3674	WARN_ON(__rbd_is_lock_owner(rbd_dev) ||
3675	rbd_dev->lock_cookie[0] != '\0');
3676
3677	format_lock_cookie(rbd_dev, buf: cookie);
3678	ret = ceph_cls_lock(osdc, oid: &rbd_dev->header_oid, oloc: &rbd_dev->header_oloc,
3679	RBD_LOCK_NAME, type: CEPH_CLS_LOCK_EXCLUSIVE, cookie,
3680	RBD_LOCK_TAG, desc: "", flags: 0);
3681	if (ret && ret != -EEXIST)
3682	return ret;
3683
3684	__rbd_lock(rbd_dev, cookie);
3685	return 0;
3686	}
3687
3688	/*
3689	* lock_rwsem must be held for write
3690	*/
3691	static void rbd_unlock(struct rbd_device *rbd_dev)
3692	{
3693	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3694	int ret;
3695
3696	WARN_ON(!__rbd_is_lock_owner(rbd_dev) ||
3697	rbd_dev->lock_cookie[0] == '\0');
3698
3699	ret = ceph_cls_unlock(osdc, oid: &rbd_dev->header_oid, oloc: &rbd_dev->header_oloc,
3700	RBD_LOCK_NAME, cookie: rbd_dev->lock_cookie);
3701	if (ret && ret != -ENOENT)
3702	rbd_warn(rbd_dev, fmt: "failed to unlock header: %d", ret);
3703
3704	/* treat errors as the image is unlocked */
3705	rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
3706	rbd_dev->lock_cookie[0] = '\0';
3707	rbd_set_owner_cid(rbd_dev, cid: &rbd_empty_cid);
3708	queue_work(wq: rbd_dev->task_wq, work: &rbd_dev->released_lock_work);
3709	}
3710
3711	static int __rbd_notify_op_lock(struct rbd_device *rbd_dev,
3712	enum rbd_notify_op notify_op,
3713	struct page ***preply_pages,
3714	size_t *preply_len)
3715	{
3716	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3717	struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3718	char buf[4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN];
3719	int buf_size = sizeof(buf);
3720	void *p = buf;
3721
3722	dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op);
3723
3724	/* encode *LockPayload NotifyMessage (op + ClientId) */
3725	ceph_start_encoding(p: &p, struct_v: 2, struct_compat: 1, struct_len: buf_size - CEPH_ENCODING_START_BLK_LEN);
3726	ceph_encode_32(p: &p, v: notify_op);
3727	ceph_encode_64(p: &p, v: cid.gid);
3728	ceph_encode_64(p: &p, v: cid.handle);
3729
3730	return ceph_osdc_notify(osdc, oid: &rbd_dev->header_oid,
3731	oloc: &rbd_dev->header_oloc, payload: buf, payload_len: buf_size,
3732	RBD_NOTIFY_TIMEOUT, preply_pages, preply_len);
3733	}
3734
3735	static void rbd_notify_op_lock(struct rbd_device *rbd_dev,
3736	enum rbd_notify_op notify_op)
3737	{
3738	__rbd_notify_op_lock(rbd_dev, notify_op, NULL, NULL);
3739	}
3740
3741	static void rbd_notify_acquired_lock(struct work_struct *work)
3742	{
3743	struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3744	acquired_lock_work);
3745
3746	rbd_notify_op_lock(rbd_dev, notify_op: RBD_NOTIFY_OP_ACQUIRED_LOCK);
3747	}
3748
3749	static void rbd_notify_released_lock(struct work_struct *work)
3750	{
3751	struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3752	released_lock_work);
3753
3754	rbd_notify_op_lock(rbd_dev, notify_op: RBD_NOTIFY_OP_RELEASED_LOCK);
3755	}
3756
3757	static int rbd_request_lock(struct rbd_device *rbd_dev)
3758	{
3759	struct page **reply_pages;
3760	size_t reply_len;
3761	bool lock_owner_responded = false;
3762	int ret;
3763
3764	dout("%s rbd_dev %p\n", __func__, rbd_dev);
3765
3766	ret = __rbd_notify_op_lock(rbd_dev, notify_op: RBD_NOTIFY_OP_REQUEST_LOCK,
3767	preply_pages: &reply_pages, preply_len: &reply_len);
3768	if (ret && ret != -ETIMEDOUT) {
3769	rbd_warn(rbd_dev, fmt: "failed to request lock: %d", ret);
3770	goto out;
3771	}
3772
3773	if (reply_len > 0 && reply_len <= PAGE_SIZE) {
3774	void *p = page_address(reply_pages[0]);
3775	void *const end = p + reply_len;
3776	u32 n;
3777
3778	ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */
3779	while (n--) {
3780	u8 struct_v;
3781	u32 len;
3782
3783	ceph_decode_need(&p, end, 8 + 8, e_inval);
3784	p += 8 + 8; /* skip gid and cookie */
3785
3786	ceph_decode_32_safe(&p, end, len, e_inval);
3787	if (!len)
3788	continue;
3789
3790	if (lock_owner_responded) {
3791	rbd_warn(rbd_dev,
3792	fmt: "duplicate lock owners detected");
3793	ret = -EIO;
3794	goto out;
3795	}
3796
3797	lock_owner_responded = true;
3798	ret = ceph_start_decoding(p: &p, end, v: 1, name: "ResponseMessage",
3799	struct_v: &struct_v, struct_len: &len);
3800	if (ret) {
3801	rbd_warn(rbd_dev,
3802	fmt: "failed to decode ResponseMessage: %d",
3803	ret);
3804	goto e_inval;
3805	}
3806
3807	ret = ceph_decode_32(p: &p);
3808	}
3809	}
3810
3811	if (!lock_owner_responded) {
3812	rbd_warn(rbd_dev, fmt: "no lock owners detected");
3813	ret = -ETIMEDOUT;
3814	}
3815
3816	out:
3817	ceph_release_page_vector(pages: reply_pages, num_pages: calc_pages_for(off: 0, len: reply_len));
3818	return ret;
3819
3820	e_inval:
3821	ret = -EINVAL;
3822	goto out;
3823	}
3824
3825	/*
3826	* Either image request state machine(s) or rbd_add_acquire_lock()
3827	* (i.e. "rbd map").
3828	*/
3829	static void wake_lock_waiters(struct rbd_device *rbd_dev, int result)
3830	{
3831	struct rbd_img_request *img_req;
3832
3833	dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
3834	lockdep_assert_held_write(&rbd_dev->lock_rwsem);
3835
3836	cancel_delayed_work(dwork: &rbd_dev->lock_dwork);
3837	if (!completion_done(x: &rbd_dev->acquire_wait)) {
3838	rbd_assert(list_empty(&rbd_dev->acquiring_list) &&
3839	list_empty(&rbd_dev->running_list));
3840	rbd_dev->acquire_err = result;
3841	complete_all(&rbd_dev->acquire_wait);
3842	return;
3843	}
3844
3845	list_for_each_entry(img_req, &rbd_dev->acquiring_list, lock_item) {
3846	mutex_lock(&img_req->state_mutex);
3847	rbd_assert(img_req->state == RBD_IMG_EXCLUSIVE_LOCK);
3848	rbd_img_schedule(img_req, result);
3849	mutex_unlock(lock: &img_req->state_mutex);
3850	}
3851
3852	list_splice_tail_init(list: &rbd_dev->acquiring_list, head: &rbd_dev->running_list);
3853	}
3854
3855	static bool locker_equal(const struct ceph_locker *lhs,
3856	const struct ceph_locker *rhs)
3857	{
3858	return lhs->id.name.type == rhs->id.name.type &&
3859	lhs->id.name.num == rhs->id.name.num &&
3860	!strcmp(lhs->id.cookie, rhs->id.cookie) &&
3861	ceph_addr_equal_no_type(lhs: &lhs->info.addr, rhs: &rhs->info.addr);
3862	}
3863
3864	static void free_locker(struct ceph_locker *locker)
3865	{
3866	if (locker)
3867	ceph_free_lockers(lockers: locker, num_lockers: 1);
3868	}
3869
3870	static struct ceph_locker *get_lock_owner_info(struct rbd_device *rbd_dev)
3871	{
3872	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3873	struct ceph_locker *lockers;
3874	u32 num_lockers;
3875	u8 lock_type;
3876	char *lock_tag;
3877	u64 handle;
3878	int ret;
3879
3880	ret = ceph_cls_lock_info(osdc, oid: &rbd_dev->header_oid,
3881	oloc: &rbd_dev->header_oloc, RBD_LOCK_NAME,
3882	type: &lock_type, tag: &lock_tag, lockers: &lockers, num_lockers: &num_lockers);
3883	if (ret) {
3884	rbd_warn(rbd_dev, fmt: "failed to get header lockers: %d", ret);
3885	return ERR_PTR(error: ret);
3886	}
3887
3888	if (num_lockers == 0) {
3889	dout("%s rbd_dev %p no lockers detected\n", __func__, rbd_dev);
3890	lockers = NULL;
3891	goto out;
3892	}
3893
3894	if (strcmp(lock_tag, RBD_LOCK_TAG)) {
3895	rbd_warn(rbd_dev, fmt: "locked by external mechanism, tag %s",
3896	lock_tag);
3897	goto err_busy;
3898	}
3899
3900	if (lock_type != CEPH_CLS_LOCK_EXCLUSIVE) {
3901	rbd_warn(rbd_dev, fmt: "incompatible lock type detected");
3902	goto err_busy;
3903	}
3904
3905	WARN_ON(num_lockers != 1);
3906	ret = sscanf(lockers[0].id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu",
3907	&handle);
3908	if (ret != 1) {
3909	rbd_warn(rbd_dev, fmt: "locked by external mechanism, cookie %s",
3910	lockers[0].id.cookie);
3911	goto err_busy;
3912	}
3913	if (ceph_addr_is_blank(addr: &lockers[0].info.addr)) {
3914	rbd_warn(rbd_dev, fmt: "locker has a blank address");
3915	goto err_busy;
3916	}
3917
3918	dout("%s rbd_dev %p got locker %s%llu@%pISpc/%u handle %llu\n",
3919	__func__, rbd_dev, ENTITY_NAME(lockers[0].id.name),
3920	&lockers[0].info.addr.in_addr,
3921	le32_to_cpu(lockers[0].info.addr.nonce), handle);
3922
3923	out:
3924	kfree(objp: lock_tag);
3925	return lockers;
3926
3927	err_busy:
3928	kfree(objp: lock_tag);
3929	ceph_free_lockers(lockers, num_lockers);
3930	return ERR_PTR(error: -EBUSY);
3931	}
3932
3933	static int find_watcher(struct rbd_device *rbd_dev,
3934	const struct ceph_locker *locker)
3935	{
3936	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3937	struct ceph_watch_item *watchers;
3938	u32 num_watchers;
3939	u64 cookie;
3940	int i;
3941	int ret;
3942
3943	ret = ceph_osdc_list_watchers(osdc, oid: &rbd_dev->header_oid,
3944	oloc: &rbd_dev->header_oloc, watchers: &watchers,
3945	num_watchers: &num_watchers);
3946	if (ret) {
3947	rbd_warn(rbd_dev, fmt: "failed to get watchers: %d", ret);
3948	return ret;
3949	}
3950
3951	sscanf(locker->id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu", &cookie);
3952	for (i = 0; i < num_watchers; i++) {
3953	/*
3954	* Ignore addr->type while comparing. This mimics
3955	* entity_addr_t::get_legacy_str() + strcmp().
3956	*/
3957	if (ceph_addr_equal_no_type(lhs: &watchers[i].addr,
3958	rhs: &locker->info.addr) &&
3959	watchers[i].cookie == cookie) {
3960	struct rbd_client_id cid = {
3961	.gid = le64_to_cpu(watchers[i].name.num),
3962	.handle = cookie,
3963	};
3964
3965	dout("%s rbd_dev %p found cid %llu-%llu\n", __func__,
3966	rbd_dev, cid.gid, cid.handle);
3967	rbd_set_owner_cid(rbd_dev, cid: &cid);
3968	ret = 1;
3969	goto out;
3970	}
3971	}
3972
3973	dout("%s rbd_dev %p no watchers\n", __func__, rbd_dev);
3974	ret = 0;
3975	out:
3976	kfree(objp: watchers);
3977	return ret;
3978	}
3979
3980	/*
3981	* lock_rwsem must be held for write
3982	*/
3983	static int rbd_try_lock(struct rbd_device *rbd_dev)
3984	{
3985	struct ceph_client *client = rbd_dev->rbd_client->client;
3986	struct ceph_locker *locker, *refreshed_locker;
3987	int ret;
3988
3989	for (;;) {
3990	locker = refreshed_locker = NULL;
3991
3992	ret = rbd_lock(rbd_dev);
3993	if (!ret)
3994	goto out;
3995	if (ret != -EBUSY) {
3996	rbd_warn(rbd_dev, fmt: "failed to lock header: %d", ret);
3997	goto out;
3998	}
3999
4000	/* determine if the current lock holder is still alive */
4001	locker = get_lock_owner_info(rbd_dev);
4002	if (IS_ERR(ptr: locker)) {
4003	ret = PTR_ERR(ptr: locker);
4004	locker = NULL;
4005	goto out;
4006	}
4007	if (!locker)
4008	goto again;
4009
4010	ret = find_watcher(rbd_dev, locker);
4011	if (ret)
4012	goto out; /* request lock or error */
4013
4014	refreshed_locker = get_lock_owner_info(rbd_dev);
4015	if (IS_ERR(ptr: refreshed_locker)) {
4016	ret = PTR_ERR(ptr: refreshed_locker);
4017	refreshed_locker = NULL;
4018	goto out;
4019	}
4020	if (!refreshed_locker ||
4021	!locker_equal(lhs: locker, rhs: refreshed_locker))
4022	goto again;
4023
4024	rbd_warn(rbd_dev, fmt: "breaking header lock owned by %s%llu",
4025	ENTITY_NAME(locker->id.name));
4026
4027	ret = ceph_monc_blocklist_add(monc: &client->monc,
4028	client_addr: &locker->info.addr);
4029	if (ret) {
4030	rbd_warn(rbd_dev, fmt: "failed to blocklist %s%llu: %d",
4031	ENTITY_NAME(locker->id.name), ret);
4032	goto out;
4033	}
4034
4035	ret = ceph_cls_break_lock(osdc: &client->osdc, oid: &rbd_dev->header_oid,
4036	oloc: &rbd_dev->header_oloc, RBD_LOCK_NAME,
4037	cookie: locker->id.cookie, locker: &locker->id.name);
4038	if (ret && ret != -ENOENT) {
4039	rbd_warn(rbd_dev, fmt: "failed to break header lock: %d",
4040	ret);
4041	goto out;
4042	}
4043
4044	again:
4045	free_locker(locker: refreshed_locker);
4046	free_locker(locker);
4047	}
4048
4049	out:
4050	free_locker(locker: refreshed_locker);
4051	free_locker(locker);
4052	return ret;
4053	}
4054
4055	static int rbd_post_acquire_action(struct rbd_device *rbd_dev)
4056	{
4057	int ret;
4058
4059	ret = rbd_dev_refresh(rbd_dev);
4060	if (ret)
4061	return ret;
4062
4063	if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) {
4064	ret = rbd_object_map_open(rbd_dev);
4065	if (ret)
4066	return ret;
4067	}
4068
4069	return 0;
4070	}
4071
4072	/*
4073	* Return:
4074	* 0 - lock acquired
4075	* 1 - caller should call rbd_request_lock()
4076	* <0 - error
4077	*/
4078	static int rbd_try_acquire_lock(struct rbd_device *rbd_dev)
4079	{
4080	int ret;
4081
4082	down_read(sem: &rbd_dev->lock_rwsem);
4083	dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
4084	rbd_dev->lock_state);
4085	if (__rbd_is_lock_owner(rbd_dev)) {
4086	up_read(sem: &rbd_dev->lock_rwsem);
4087	return 0;
4088	}
4089
4090	up_read(sem: &rbd_dev->lock_rwsem);
4091	down_write(sem: &rbd_dev->lock_rwsem);
4092	dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
4093	rbd_dev->lock_state);
4094	if (__rbd_is_lock_owner(rbd_dev)) {
4095	up_write(sem: &rbd_dev->lock_rwsem);
4096	return 0;
4097	}
4098
4099	ret = rbd_try_lock(rbd_dev);
4100	if (ret < 0) {
4101	rbd_warn(rbd_dev, fmt: "failed to acquire lock: %d", ret);
4102	goto out;
4103	}
4104	if (ret > 0) {
4105	up_write(sem: &rbd_dev->lock_rwsem);
4106	return ret;
4107	}
4108
4109	rbd_assert(rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED);
4110	rbd_assert(list_empty(&rbd_dev->running_list));
4111
4112	ret = rbd_post_acquire_action(rbd_dev);
4113	if (ret) {
4114	rbd_warn(rbd_dev, fmt: "post-acquire action failed: %d", ret);
4115	/*
4116	* Can't stay in RBD_LOCK_STATE_LOCKED because
4117	* rbd_lock_add_request() would let the request through,
4118	* assuming that e.g. object map is locked and loaded.
4119	*/
4120	rbd_unlock(rbd_dev);
4121	}
4122
4123	out:
4124	wake_lock_waiters(rbd_dev, result: ret);
4125	up_write(sem: &rbd_dev->lock_rwsem);
4126	return ret;
4127	}
4128
4129	static void rbd_acquire_lock(struct work_struct *work)
4130	{
4131	struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4132	struct rbd_device, lock_dwork);
4133	int ret;
4134
4135	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4136	again:
4137	ret = rbd_try_acquire_lock(rbd_dev);
4138	if (ret <= 0) {
4139	dout("%s rbd_dev %p ret %d - done\n", __func__, rbd_dev, ret);
4140	return;
4141	}
4142
4143	ret = rbd_request_lock(rbd_dev);
4144	if (ret == -ETIMEDOUT) {
4145	goto again; /* treat this as a dead client */
4146	} else if (ret == -EROFS) {
4147	rbd_warn(rbd_dev, fmt: "peer will not release lock");
4148	down_write(sem: &rbd_dev->lock_rwsem);
4149	wake_lock_waiters(rbd_dev, result: ret);
4150	up_write(sem: &rbd_dev->lock_rwsem);
4151	} else if (ret < 0) {
4152	rbd_warn(rbd_dev, fmt: "error requesting lock: %d", ret);
4153	mod_delayed_work(wq: rbd_dev->task_wq, dwork: &rbd_dev->lock_dwork,
4154	RBD_RETRY_DELAY);
4155	} else {
4156	/*
4157	* lock owner acked, but resend if we don't see them
4158	* release the lock
4159	*/
4160	dout("%s rbd_dev %p requeuing lock_dwork\n", __func__,
4161	rbd_dev);
4162	mod_delayed_work(wq: rbd_dev->task_wq, dwork: &rbd_dev->lock_dwork,
4163	delay: msecs_to_jiffies(m: 2 * RBD_NOTIFY_TIMEOUT * MSEC_PER_SEC));
4164	}
4165	}
4166
4167	static bool rbd_quiesce_lock(struct rbd_device *rbd_dev)
4168	{
4169	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4170	lockdep_assert_held_write(&rbd_dev->lock_rwsem);
4171
4172	if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED)
4173	return false;
4174
4175	/*
4176	* Ensure that all in-flight IO is flushed.
4177	*/
4178	rbd_dev->lock_state = RBD_LOCK_STATE_RELEASING;
4179	rbd_assert(!completion_done(&rbd_dev->releasing_wait));
4180	if (list_empty(head: &rbd_dev->running_list))
4181	return true;
4182
4183	up_write(sem: &rbd_dev->lock_rwsem);
4184	wait_for_completion(&rbd_dev->releasing_wait);
4185
4186	down_write(sem: &rbd_dev->lock_rwsem);
4187	if (rbd_dev->lock_state != RBD_LOCK_STATE_RELEASING)
4188	return false;
4189
4190	rbd_assert(list_empty(&rbd_dev->running_list));
4191	return true;
4192	}
4193
4194	static void rbd_pre_release_action(struct rbd_device *rbd_dev)
4195	{
4196	if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)
4197	rbd_object_map_close(rbd_dev);
4198	}
4199
4200	static void __rbd_release_lock(struct rbd_device *rbd_dev)
4201	{
4202	rbd_assert(list_empty(&rbd_dev->running_list));
4203
4204	rbd_pre_release_action(rbd_dev);
4205	rbd_unlock(rbd_dev);
4206	}
4207
4208	/*
4209	* lock_rwsem must be held for write
4210	*/
4211	static void rbd_release_lock(struct rbd_device *rbd_dev)
4212	{
4213	if (!rbd_quiesce_lock(rbd_dev))
4214	return;
4215
4216	__rbd_release_lock(rbd_dev);
4217
4218	/*
4219	* Give others a chance to grab the lock - we would re-acquire
4220	* almost immediately if we got new IO while draining the running
4221	* list otherwise. We need to ack our own notifications, so this
4222	* lock_dwork will be requeued from rbd_handle_released_lock() by
4223	* way of maybe_kick_acquire().
4224	*/
4225	cancel_delayed_work(dwork: &rbd_dev->lock_dwork);
4226	}
4227
4228	static void rbd_release_lock_work(struct work_struct *work)
4229	{
4230	struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
4231	unlock_work);
4232
4233	down_write(sem: &rbd_dev->lock_rwsem);
4234	rbd_release_lock(rbd_dev);
4235	up_write(sem: &rbd_dev->lock_rwsem);
4236	}
4237
4238	static void maybe_kick_acquire(struct rbd_device *rbd_dev)
4239	{
4240	bool have_requests;
4241
4242	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4243	if (__rbd_is_lock_owner(rbd_dev))
4244	return;
4245
4246	spin_lock(lock: &rbd_dev->lock_lists_lock);
4247	have_requests = !list_empty(head: &rbd_dev->acquiring_list);
4248	spin_unlock(lock: &rbd_dev->lock_lists_lock);
4249	if (have_requests || delayed_work_pending(&rbd_dev->lock_dwork)) {
4250	dout("%s rbd_dev %p kicking lock_dwork\n", __func__, rbd_dev);
4251	mod_delayed_work(wq: rbd_dev->task_wq, dwork: &rbd_dev->lock_dwork, delay: 0);
4252	}
4253	}
4254
4255	static void rbd_handle_acquired_lock(struct rbd_device *rbd_dev, u8 struct_v,
4256	void **p)
4257	{
4258	struct rbd_client_id cid = { 0 };
4259
4260	if (struct_v >= 2) {
4261	cid.gid = ceph_decode_64(p);
4262	cid.handle = ceph_decode_64(p);
4263	}
4264
4265	dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4266	cid.handle);
4267	if (!rbd_cid_equal(lhs: &cid, rhs: &rbd_empty_cid)) {
4268	down_write(sem: &rbd_dev->lock_rwsem);
4269	if (rbd_cid_equal(lhs: &cid, rhs: &rbd_dev->owner_cid)) {
4270	dout("%s rbd_dev %p cid %llu-%llu == owner_cid\n",
4271	__func__, rbd_dev, cid.gid, cid.handle);
4272	} else {
4273	rbd_set_owner_cid(rbd_dev, cid: &cid);
4274	}
4275	downgrade_write(sem: &rbd_dev->lock_rwsem);
4276	} else {
4277	down_read(sem: &rbd_dev->lock_rwsem);
4278	}
4279
4280	maybe_kick_acquire(rbd_dev);
4281	up_read(sem: &rbd_dev->lock_rwsem);
4282	}
4283
4284	static void rbd_handle_released_lock(struct rbd_device *rbd_dev, u8 struct_v,
4285	void **p)
4286	{
4287	struct rbd_client_id cid = { 0 };
4288
4289	if (struct_v >= 2) {
4290	cid.gid = ceph_decode_64(p);
4291	cid.handle = ceph_decode_64(p);
4292	}
4293
4294	dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4295	cid.handle);
4296	if (!rbd_cid_equal(lhs: &cid, rhs: &rbd_empty_cid)) {
4297	down_write(sem: &rbd_dev->lock_rwsem);
4298	if (!rbd_cid_equal(lhs: &cid, rhs: &rbd_dev->owner_cid)) {
4299	dout("%s rbd_dev %p cid %llu-%llu != owner_cid %llu-%llu\n",
4300	__func__, rbd_dev, cid.gid, cid.handle,
4301	rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle);
4302	} else {
4303	rbd_set_owner_cid(rbd_dev, cid: &rbd_empty_cid);
4304	}
4305	downgrade_write(sem: &rbd_dev->lock_rwsem);
4306	} else {
4307	down_read(sem: &rbd_dev->lock_rwsem);
4308	}
4309
4310	maybe_kick_acquire(rbd_dev);
4311	up_read(sem: &rbd_dev->lock_rwsem);
4312	}
4313
4314	/*
4315	* Returns result for ResponseMessage to be encoded (<= 0), or 1 if no
4316	* ResponseMessage is needed.
4317	*/
4318	static int rbd_handle_request_lock(struct rbd_device *rbd_dev, u8 struct_v,
4319	void **p)
4320	{
4321	struct rbd_client_id my_cid = rbd_get_cid(rbd_dev);
4322	struct rbd_client_id cid = { 0 };
4323	int result = 1;
4324
4325	if (struct_v >= 2) {
4326	cid.gid = ceph_decode_64(p);
4327	cid.handle = ceph_decode_64(p);
4328	}
4329
4330	dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4331	cid.handle);
4332	if (rbd_cid_equal(lhs: &cid, rhs: &my_cid))
4333	return result;
4334
4335	down_read(sem: &rbd_dev->lock_rwsem);
4336	if (__rbd_is_lock_owner(rbd_dev)) {
4337	if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED &&
4338	rbd_cid_equal(lhs: &rbd_dev->owner_cid, rhs: &rbd_empty_cid))
4339	goto out_unlock;
4340
4341	/*
4342	* encode ResponseMessage(0) so the peer can detect
4343	* a missing owner
4344	*/
4345	result = 0;
4346
4347	if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED) {
4348	if (!rbd_dev->opts->exclusive) {
4349	dout("%s rbd_dev %p queueing unlock_work\n",
4350	__func__, rbd_dev);
4351	queue_work(wq: rbd_dev->task_wq,
4352	work: &rbd_dev->unlock_work);
4353	} else {
4354	/* refuse to release the lock */
4355	result = -EROFS;
4356	}
4357	}
4358	}
4359
4360	out_unlock:
4361	up_read(sem: &rbd_dev->lock_rwsem);
4362	return result;
4363	}
4364
4365	static void __rbd_acknowledge_notify(struct rbd_device *rbd_dev,
4366	u64 notify_id, u64 cookie, s32 *result)
4367	{
4368	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4369	char buf[4 + CEPH_ENCODING_START_BLK_LEN];
4370	int buf_size = sizeof(buf);
4371	int ret;
4372
4373	if (result) {
4374	void *p = buf;
4375
4376	/* encode ResponseMessage */
4377	ceph_start_encoding(p: &p, struct_v: 1, struct_compat: 1,
4378	struct_len: buf_size - CEPH_ENCODING_START_BLK_LEN);
4379	ceph_encode_32(p: &p, v: *result);
4380	} else {
4381	buf_size = 0;
4382	}
4383
4384	ret = ceph_osdc_notify_ack(osdc, oid: &rbd_dev->header_oid,
4385	oloc: &rbd_dev->header_oloc, notify_id, cookie,
4386	payload: buf, payload_len: buf_size);
4387	if (ret)
4388	rbd_warn(rbd_dev, fmt: "acknowledge_notify failed: %d", ret);
4389	}
4390
4391	static void rbd_acknowledge_notify(struct rbd_device *rbd_dev, u64 notify_id,
4392	u64 cookie)
4393	{
4394	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4395	__rbd_acknowledge_notify(rbd_dev, notify_id, cookie, NULL);
4396	}
4397
4398	static void rbd_acknowledge_notify_result(struct rbd_device *rbd_dev,
4399	u64 notify_id, u64 cookie, s32 result)
4400	{
4401	dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
4402	__rbd_acknowledge_notify(rbd_dev, notify_id, cookie, result: &result);
4403	}
4404
4405	static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie,
4406	u64 notifier_id, void *data, size_t data_len)
4407	{
4408	struct rbd_device *rbd_dev = arg;
4409	void *p = data;
4410	void *const end = p + data_len;
4411	u8 struct_v = 0;
4412	u32 len;
4413	u32 notify_op;
4414	int ret;
4415
4416	dout("%s rbd_dev %p cookie %llu notify_id %llu data_len %zu\n",
4417	__func__, rbd_dev, cookie, notify_id, data_len);
4418	if (data_len) {
4419	ret = ceph_start_decoding(p: &p, end, v: 1, name: "NotifyMessage",
4420	struct_v: &struct_v, struct_len: &len);
4421	if (ret) {
4422	rbd_warn(rbd_dev, fmt: "failed to decode NotifyMessage: %d",
4423	ret);
4424	return;
4425	}
4426
4427	notify_op = ceph_decode_32(p: &p);
4428	} else {
4429	/* legacy notification for header updates */
4430	notify_op = RBD_NOTIFY_OP_HEADER_UPDATE;
4431	len = 0;
4432	}
4433
4434	dout("%s rbd_dev %p notify_op %u\n", __func__, rbd_dev, notify_op);
4435	switch (notify_op) {
4436	case RBD_NOTIFY_OP_ACQUIRED_LOCK:
4437	rbd_handle_acquired_lock(rbd_dev, struct_v, p: &p);
4438	rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4439	break;
4440	case RBD_NOTIFY_OP_RELEASED_LOCK:
4441	rbd_handle_released_lock(rbd_dev, struct_v, p: &p);
4442	rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4443	break;
4444	case RBD_NOTIFY_OP_REQUEST_LOCK:
4445	ret = rbd_handle_request_lock(rbd_dev, struct_v, p: &p);
4446	if (ret <= 0)
4447	rbd_acknowledge_notify_result(rbd_dev, notify_id,
4448	cookie, result: ret);
4449	else
4450	rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4451	break;
4452	case RBD_NOTIFY_OP_HEADER_UPDATE:
4453	ret = rbd_dev_refresh(rbd_dev);
4454	if (ret)
4455	rbd_warn(rbd_dev, fmt: "refresh failed: %d", ret);
4456
4457	rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4458	break;
4459	default:
4460	if (rbd_is_lock_owner(rbd_dev))
4461	rbd_acknowledge_notify_result(rbd_dev, notify_id,
4462	cookie, result: -EOPNOTSUPP);
4463	else
4464	rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4465	break;
4466	}
4467	}
4468
4469	static void __rbd_unregister_watch(struct rbd_device *rbd_dev);
4470
4471	static void rbd_watch_errcb(void *arg, u64 cookie, int err)
4472	{
4473	struct rbd_device *rbd_dev = arg;
4474
4475	rbd_warn(rbd_dev, fmt: "encountered watch error: %d", err);
4476
4477	down_write(sem: &rbd_dev->lock_rwsem);
4478	rbd_set_owner_cid(rbd_dev, cid: &rbd_empty_cid);
4479	up_write(sem: &rbd_dev->lock_rwsem);
4480
4481	mutex_lock(&rbd_dev->watch_mutex);
4482	if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED) {
4483	__rbd_unregister_watch(rbd_dev);
4484	rbd_dev->watch_state = RBD_WATCH_STATE_ERROR;
4485
4486	queue_delayed_work(wq: rbd_dev->task_wq, dwork: &rbd_dev->watch_dwork, delay: 0);
4487	}
4488	mutex_unlock(lock: &rbd_dev->watch_mutex);
4489	}
4490
4491	/*
4492	* watch_mutex must be locked
4493	*/
4494	static int __rbd_register_watch(struct rbd_device *rbd_dev)
4495	{
4496	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4497	struct ceph_osd_linger_request *handle;
4498
4499	rbd_assert(!rbd_dev->watch_handle);
4500	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4501
4502	handle = ceph_osdc_watch(osdc, oid: &rbd_dev->header_oid,
4503	oloc: &rbd_dev->header_oloc, wcb: rbd_watch_cb,
4504	errcb: rbd_watch_errcb, data: rbd_dev);
4505	if (IS_ERR(ptr: handle))
4506	return PTR_ERR(ptr: handle);
4507
4508	rbd_dev->watch_handle = handle;
4509	return 0;
4510	}
4511
4512	/*
4513	* watch_mutex must be locked
4514	*/
4515	static void __rbd_unregister_watch(struct rbd_device *rbd_dev)
4516	{
4517	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4518	int ret;
4519
4520	rbd_assert(rbd_dev->watch_handle);
4521	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4522
4523	ret = ceph_osdc_unwatch(osdc, lreq: rbd_dev->watch_handle);
4524	if (ret)
4525	rbd_warn(rbd_dev, fmt: "failed to unwatch: %d", ret);
4526
4527	rbd_dev->watch_handle = NULL;
4528	}
4529
4530	static int rbd_register_watch(struct rbd_device *rbd_dev)
4531	{
4532	int ret;
4533
4534	mutex_lock(&rbd_dev->watch_mutex);
4535	rbd_assert(rbd_dev->watch_state == RBD_WATCH_STATE_UNREGISTERED);
4536	ret = __rbd_register_watch(rbd_dev);
4537	if (ret)
4538	goto out;
4539
4540	rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4541	rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4542
4543	out:
4544	mutex_unlock(lock: &rbd_dev->watch_mutex);
4545	return ret;
4546	}
4547
4548	static void cancel_tasks_sync(struct rbd_device *rbd_dev)
4549	{
4550	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4551
4552	cancel_work_sync(work: &rbd_dev->acquired_lock_work);
4553	cancel_work_sync(work: &rbd_dev->released_lock_work);
4554	cancel_delayed_work_sync(dwork: &rbd_dev->lock_dwork);
4555	cancel_work_sync(work: &rbd_dev->unlock_work);
4556	}
4557
4558	/*
4559	* header_rwsem must not be held to avoid a deadlock with
4560	* rbd_dev_refresh() when flushing notifies.
4561	*/
4562	static void rbd_unregister_watch(struct rbd_device *rbd_dev)
4563	{
4564	cancel_tasks_sync(rbd_dev);
4565
4566	mutex_lock(&rbd_dev->watch_mutex);
4567	if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED)
4568	__rbd_unregister_watch(rbd_dev);
4569	rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
4570	mutex_unlock(lock: &rbd_dev->watch_mutex);
4571
4572	cancel_delayed_work_sync(dwork: &rbd_dev->watch_dwork);
4573	ceph_osdc_flush_notifies(osdc: &rbd_dev->rbd_client->client->osdc);
4574	}
4575
4576	/*
4577	* lock_rwsem must be held for write
4578	*/
4579	static void rbd_reacquire_lock(struct rbd_device *rbd_dev)
4580	{
4581	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4582	char cookie[32];
4583	int ret;
4584
4585	if (!rbd_quiesce_lock(rbd_dev))
4586	return;
4587
4588	format_lock_cookie(rbd_dev, buf: cookie);
4589	ret = ceph_cls_set_cookie(osdc, oid: &rbd_dev->header_oid,
4590	oloc: &rbd_dev->header_oloc, RBD_LOCK_NAME,
4591	type: CEPH_CLS_LOCK_EXCLUSIVE, old_cookie: rbd_dev->lock_cookie,
4592	RBD_LOCK_TAG, new_cookie: cookie);
4593	if (ret) {
4594	if (ret != -EOPNOTSUPP)
4595	rbd_warn(rbd_dev, fmt: "failed to update lock cookie: %d",
4596	ret);
4597
4598	/*
4599	* Lock cookie cannot be updated on older OSDs, so do
4600	* a manual release and queue an acquire.
4601	*/
4602	__rbd_release_lock(rbd_dev);
4603	queue_delayed_work(wq: rbd_dev->task_wq, dwork: &rbd_dev->lock_dwork, delay: 0);
4604	} else {
4605	__rbd_lock(rbd_dev, cookie);
4606	wake_lock_waiters(rbd_dev, result: 0);
4607	}
4608	}
4609
4610	static void rbd_reregister_watch(struct work_struct *work)
4611	{
4612	struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4613	struct rbd_device, watch_dwork);
4614	int ret;
4615
4616	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4617
4618	mutex_lock(&rbd_dev->watch_mutex);
4619	if (rbd_dev->watch_state != RBD_WATCH_STATE_ERROR) {
4620	mutex_unlock(lock: &rbd_dev->watch_mutex);
4621	return;
4622	}
4623
4624	ret = __rbd_register_watch(rbd_dev);
4625	if (ret) {
4626	rbd_warn(rbd_dev, fmt: "failed to reregister watch: %d", ret);
4627	if (ret != -EBLOCKLISTED && ret != -ENOENT) {
4628	queue_delayed_work(wq: rbd_dev->task_wq,
4629	dwork: &rbd_dev->watch_dwork,
4630	RBD_RETRY_DELAY);
4631	mutex_unlock(lock: &rbd_dev->watch_mutex);
4632	return;
4633	}
4634
4635	mutex_unlock(lock: &rbd_dev->watch_mutex);
4636	down_write(sem: &rbd_dev->lock_rwsem);
4637	wake_lock_waiters(rbd_dev, result: ret);
4638	up_write(sem: &rbd_dev->lock_rwsem);
4639	return;
4640	}
4641
4642	rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4643	rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4644	mutex_unlock(lock: &rbd_dev->watch_mutex);
4645
4646	down_write(sem: &rbd_dev->lock_rwsem);
4647	if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED)
4648	rbd_reacquire_lock(rbd_dev);
4649	up_write(sem: &rbd_dev->lock_rwsem);
4650
4651	ret = rbd_dev_refresh(rbd_dev);
4652	if (ret)
4653	rbd_warn(rbd_dev, fmt: "reregistration refresh failed: %d", ret);
4654	}
4655
4656	/*
4657	* Synchronous osd object method call. Returns the number of bytes
4658	* returned in the outbound buffer, or a negative error code.
4659	*/
4660	static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
4661	struct ceph_object_id *oid,
4662	struct ceph_object_locator *oloc,
4663	const char *method_name,
4664	const void *outbound,
4665	size_t outbound_size,
4666	void *inbound,
4667	size_t inbound_size)
4668	{
4669	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4670	struct page *req_page = NULL;
4671	struct page *reply_page;
4672	int ret;
4673
4674	/*
4675	* Method calls are ultimately read operations. The result
4676	* should placed into the inbound buffer provided. They
4677	* also supply outbound data--parameters for the object
4678	* method. Currently if this is present it will be a
4679	* snapshot id.
4680	*/
4681	if (outbound) {
4682	if (outbound_size > PAGE_SIZE)
4683	return -E2BIG;
4684
4685	req_page = alloc_page(GFP_KERNEL);
4686	if (!req_page)
4687	return -ENOMEM;
4688
4689	memcpy(page_address(req_page), outbound, outbound_size);
4690	}
4691
4692	reply_page = alloc_page(GFP_KERNEL);
4693	if (!reply_page) {
4694	if (req_page)
4695	__free_page(req_page);
4696	return -ENOMEM;
4697	}
4698
4699	ret = ceph_osdc_call(osdc, oid, oloc, RBD_DRV_NAME, method: method_name,
4700	flags: CEPH_OSD_FLAG_READ, req_page, req_len: outbound_size,
4701	resp_pages: &reply_page, resp_len: &inbound_size);
4702	if (!ret) {
4703	memcpy(inbound, page_address(reply_page), inbound_size);
4704	ret = inbound_size;
4705	}
4706
4707	if (req_page)
4708	__free_page(req_page);
4709	__free_page(reply_page);
4710	return ret;
4711	}
4712
4713	static void rbd_queue_workfn(struct work_struct *work)
4714	{
4715	struct rbd_img_request *img_request =
4716	container_of(work, struct rbd_img_request, work);
4717	struct rbd_device *rbd_dev = img_request->rbd_dev;
4718	enum obj_operation_type op_type = img_request->op_type;
4719	struct request *rq = blk_mq_rq_from_pdu(pdu: img_request);
4720	u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
4721	u64 length = blk_rq_bytes(rq);
4722	u64 mapping_size;
4723	int result;
4724
4725	/* Ignore/skip any zero-length requests */
4726	if (!length) {
4727	dout("%s: zero-length request\n", __func__);
4728	result = 0;
4729	goto err_img_request;
4730	}
4731
4732	blk_mq_start_request(rq);
4733
4734	down_read(sem: &rbd_dev->header_rwsem);
4735	mapping_size = rbd_dev->mapping.size;
4736	rbd_img_capture_header(img_req: img_request);
4737	up_read(sem: &rbd_dev->header_rwsem);
4738
4739	if (offset + length > mapping_size) {
4740	rbd_warn(rbd_dev, fmt: "beyond EOD (%llu~%llu > %llu)", offset,
4741	length, mapping_size);
4742	result = -EIO;
4743	goto err_img_request;
4744	}
4745
4746	dout("%s rbd_dev %p img_req %p %s %llu~%llu\n", __func__, rbd_dev,
4747	img_request, obj_op_name(op_type), offset, length);
4748
4749	if (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_ZEROOUT)
4750	result = rbd_img_fill_nodata(img_req: img_request, off: offset, len: length);
4751	else
4752	result = rbd_img_fill_from_bio(img_req: img_request, off: offset, len: length,
4753	bio: rq->bio);
4754	if (result)
4755	goto err_img_request;
4756
4757	rbd_img_handle_request(img_req: img_request, result: 0);
4758	return;
4759
4760	err_img_request:
4761	rbd_img_request_destroy(img_request);
4762	if (result)
4763	rbd_warn(rbd_dev, fmt: "%s %llx at %llx result %d",
4764	obj_op_name(op_type), length, offset, result);
4765	blk_mq_end_request(rq, error: errno_to_blk_status(errno: result));
4766	}
4767
4768	static blk_status_t rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
4769	const struct blk_mq_queue_data *bd)
4770	{
4771	struct rbd_device *rbd_dev = hctx->queue->queuedata;
4772	struct rbd_img_request *img_req = blk_mq_rq_to_pdu(rq: bd->rq);
4773	enum obj_operation_type op_type;
4774
4775	switch (req_op(req: bd->rq)) {
4776	case REQ_OP_DISCARD:
4777	op_type = OBJ_OP_DISCARD;
4778	break;
4779	case REQ_OP_WRITE_ZEROES:
4780	op_type = OBJ_OP_ZEROOUT;
4781	break;
4782	case REQ_OP_WRITE:
4783	op_type = OBJ_OP_WRITE;
4784	break;
4785	case REQ_OP_READ:
4786	op_type = OBJ_OP_READ;
4787	break;
4788	default:
4789	rbd_warn(rbd_dev, fmt: "unknown req_op %d", req_op(req: bd->rq));
4790	return BLK_STS_IOERR;
4791	}
4792
4793	rbd_img_request_init(img_request: img_req, rbd_dev, op_type);
4794
4795	if (rbd_img_is_write(img_req)) {
4796	if (rbd_is_ro(rbd_dev)) {
4797	rbd_warn(rbd_dev, fmt: "%s on read-only mapping",
4798	obj_op_name(op_type: img_req->op_type));
4799	return BLK_STS_IOERR;
4800	}
4801	rbd_assert(!rbd_is_snap(rbd_dev));
4802	}
4803
4804	INIT_WORK(&img_req->work, rbd_queue_workfn);
4805	queue_work(wq: rbd_wq, work: &img_req->work);
4806	return BLK_STS_OK;
4807	}
4808
4809	static void rbd_free_disk(struct rbd_device *rbd_dev)
4810	{
4811	put_disk(disk: rbd_dev->disk);
4812	blk_mq_free_tag_set(set: &rbd_dev->tag_set);
4813	rbd_dev->disk = NULL;
4814	}
4815
4816	static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
4817	struct ceph_object_id *oid,
4818	struct ceph_object_locator *oloc,
4819	void *buf, int buf_len)
4820
4821	{
4822	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4823	struct ceph_osd_request *req;
4824	struct page **pages;
4825	int num_pages = calc_pages_for(off: 0, len: buf_len);
4826	int ret;
4827
4828	req = ceph_osdc_alloc_request(osdc, NULL, num_ops: 1, use_mempool: false, GFP_KERNEL);
4829	if (!req)
4830	return -ENOMEM;
4831
4832	ceph_oid_copy(dest: &req->r_base_oid, src: oid);
4833	ceph_oloc_copy(dest: &req->r_base_oloc, src: oloc);
4834	req->r_flags = CEPH_OSD_FLAG_READ;
4835
4836	pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
4837	if (IS_ERR(ptr: pages)) {
4838	ret = PTR_ERR(ptr: pages);
4839	goto out_req;
4840	}
4841
4842	osd_req_op_extent_init(osd_req: req, which: 0, opcode: CEPH_OSD_OP_READ, offset: 0, length: buf_len, truncate_size: 0, truncate_seq: 0);
4843	osd_req_op_extent_osd_data_pages(req, which: 0, pages, length: buf_len, alignment: 0, pages_from_pool: false,
4844	own_pages: true);
4845
4846	ret = ceph_osdc_alloc_messages(req, GFP_KERNEL);
4847	if (ret)
4848	goto out_req;
4849
4850	ceph_osdc_start_request(osdc, req);
4851	ret = ceph_osdc_wait_request(osdc, req);
4852	if (ret >= 0)
4853	ceph_copy_from_page_vector(pages, data: buf, off: 0, len: ret);
4854
4855	out_req:
4856	ceph_osdc_put_request(req);
4857	return ret;
4858	}
4859
4860	/*
4861	* Read the complete header for the given rbd device. On successful
4862	* return, the rbd_dev->header field will contain up-to-date
4863	* information about the image.
4864	*/
4865	static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev,
4866	struct rbd_image_header *header,
4867	bool first_time)
4868	{
4869	struct rbd_image_header_ondisk *ondisk = NULL;
4870	u32 snap_count = 0;
4871	u64 names_size = 0;
4872	u32 want_count;
4873	int ret;
4874
4875	/*
4876	* The complete header will include an array of its 64-bit
4877	* snapshot ids, followed by the names of those snapshots as
4878	* a contiguous block of NUL-terminated strings. Note that
4879	* the number of snapshots could change by the time we read
4880	* it in, in which case we re-read it.
4881	*/
4882	do {
4883	size_t size;
4884
4885	kfree(objp: ondisk);
4886
4887	size = sizeof (*ondisk);
4888	size += snap_count * sizeof (struct rbd_image_snap_ondisk);
4889	size += names_size;
4890	ondisk = kmalloc(size, GFP_KERNEL);
4891	if (!ondisk)
4892	return -ENOMEM;
4893
4894	ret = rbd_obj_read_sync(rbd_dev, oid: &rbd_dev->header_oid,
4895	oloc: &rbd_dev->header_oloc, buf: ondisk, buf_len: size);
4896	if (ret < 0)
4897	goto out;
4898	if ((size_t)ret < size) {
4899	ret = -ENXIO;
4900	rbd_warn(rbd_dev, fmt: "short header read (want %zd got %d)",
4901	size, ret);
4902	goto out;
4903	}
4904	if (!rbd_dev_ondisk_valid(ondisk)) {
4905	ret = -ENXIO;
4906	rbd_warn(rbd_dev, fmt: "invalid header");
4907	goto out;
4908	}
4909
4910	names_size = le64_to_cpu(ondisk->snap_names_len);
4911	want_count = snap_count;
4912	snap_count = le32_to_cpu(ondisk->snap_count);
4913	} while (snap_count != want_count);
4914
4915	ret = rbd_header_from_disk(header, ondisk, first_time);
4916	out:
4917	kfree(objp: ondisk);
4918
4919	return ret;
4920	}
4921
4922	static void rbd_dev_update_size(struct rbd_device *rbd_dev)
4923	{
4924	sector_t size;
4925
4926	/*
4927	* If EXISTS is not set, rbd_dev->disk may be NULL, so don't
4928	* try to update its size. If REMOVING is set, updating size
4929	* is just useless work since the device can't be opened.
4930	*/
4931	if (test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags) &&
4932	!test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) {
4933	size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
4934	dout("setting size to %llu sectors", (unsigned long long)size);
4935	set_capacity_and_notify(disk: rbd_dev->disk, size);
4936	}
4937	}
4938
4939	static const struct blk_mq_ops rbd_mq_ops = {
4940	.queue_rq = rbd_queue_rq,
4941	};
4942
4943	static int rbd_init_disk(struct rbd_device *rbd_dev)
4944	{
4945	struct gendisk *disk;
4946	struct request_queue *q;
4947	unsigned int objset_bytes =
4948	rbd_dev->layout.object_size * rbd_dev->layout.stripe_count;
4949	int err;
4950
4951	memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
4952	rbd_dev->tag_set.ops = &rbd_mq_ops;
4953	rbd_dev->tag_set.queue_depth = rbd_dev->opts->queue_depth;
4954	rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
4955	rbd_dev->tag_set.flags = BLK_MQ_F_SHOULD_MERGE;
4956	rbd_dev->tag_set.nr_hw_queues = num_present_cpus();
4957	rbd_dev->tag_set.cmd_size = sizeof(struct rbd_img_request);
4958
4959	err = blk_mq_alloc_tag_set(set: &rbd_dev->tag_set);
4960	if (err)
4961	return err;
4962
4963	disk = blk_mq_alloc_disk(&rbd_dev->tag_set, rbd_dev);
4964	if (IS_ERR(ptr: disk)) {
4965	err = PTR_ERR(ptr: disk);
4966	goto out_tag_set;
4967	}
4968	q = disk->queue;
4969
4970	snprintf(buf: disk->disk_name, size: sizeof(disk->disk_name), RBD_DRV_NAME "%d",
4971	rbd_dev->dev_id);
4972	disk->major = rbd_dev->major;
4973	disk->first_minor = rbd_dev->minor;
4974	if (single_major)
4975	disk->minors = (1 << RBD_SINGLE_MAJOR_PART_SHIFT);
4976	else
4977	disk->minors = RBD_MINORS_PER_MAJOR;
4978	disk->fops = &rbd_bd_ops;
4979	disk->private_data = rbd_dev;
4980
4981	blk_queue_flag_set(QUEUE_FLAG_NONROT, q);
4982	/* QUEUE_FLAG_ADD_RANDOM is off by default for blk-mq */
4983
4984	blk_queue_max_hw_sectors(q, objset_bytes >> SECTOR_SHIFT);
4985	q->limits.max_sectors = queue_max_hw_sectors(q);
4986	blk_queue_max_segments(q, USHRT_MAX);
4987	blk_queue_max_segment_size(q, UINT_MAX);
4988	blk_queue_io_min(q, min: rbd_dev->opts->alloc_size);
4989	blk_queue_io_opt(q, opt: rbd_dev->opts->alloc_size);
4990
4991	if (rbd_dev->opts->trim) {
4992	q->limits.discard_granularity = rbd_dev->opts->alloc_size;
4993	blk_queue_max_discard_sectors(q, max_discard_sectors: objset_bytes >> SECTOR_SHIFT);
4994	blk_queue_max_write_zeroes_sectors(q, max_write_same_sectors: objset_bytes >> SECTOR_SHIFT);
4995	}
4996
4997	if (!ceph_test_opt(rbd_dev->rbd_client->client, NOCRC))
4998	blk_queue_flag_set(QUEUE_FLAG_STABLE_WRITES, q);
4999
5000	rbd_dev->disk = disk;
5001
5002	return 0;
5003	out_tag_set:
5004	blk_mq_free_tag_set(set: &rbd_dev->tag_set);
5005	return err;
5006	}
5007
5008	/*
5009	sysfs
5010	*/
5011
5012	static struct rbd_device *dev_to_rbd_dev(struct device *dev)
5013	{
5014	return container_of(dev, struct rbd_device, dev);
5015	}
5016
5017	static ssize_t rbd_size_show(struct device *dev,
5018	struct device_attribute *attr, char *buf)
5019	{
5020	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5021
5022	return sprintf(buf, fmt: "%llu\n",
5023	(unsigned long long)rbd_dev->mapping.size);
5024	}
5025
5026	static ssize_t rbd_features_show(struct device *dev,
5027	struct device_attribute *attr, char *buf)
5028	{
5029	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5030
5031	return sprintf(buf, fmt: "0x%016llx\n", rbd_dev->header.features);
5032	}
5033
5034	static ssize_t rbd_major_show(struct device *dev,
5035	struct device_attribute *attr, char *buf)
5036	{
5037	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5038
5039	if (rbd_dev->major)
5040	return sprintf(buf, fmt: "%d\n", rbd_dev->major);
5041
5042	return sprintf(buf, fmt: "(none)\n");
5043	}
5044
5045	static ssize_t rbd_minor_show(struct device *dev,
5046	struct device_attribute *attr, char *buf)
5047	{
5048	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5049
5050	return sprintf(buf, fmt: "%d\n", rbd_dev->minor);
5051	}
5052
5053	static ssize_t rbd_client_addr_show(struct device *dev,
5054	struct device_attribute *attr, char *buf)
5055	{
5056	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5057	struct ceph_entity_addr *client_addr =
5058	ceph_client_addr(client: rbd_dev->rbd_client->client);
5059
5060	return sprintf(buf, fmt: "%pISpc/%u\n", &client_addr->in_addr,
5061	le32_to_cpu(client_addr->nonce));
5062	}
5063
5064	static ssize_t rbd_client_id_show(struct device *dev,
5065	struct device_attribute *attr, char *buf)
5066	{
5067	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5068
5069	return sprintf(buf, fmt: "client%lld\n",
5070	ceph_client_gid(client: rbd_dev->rbd_client->client));
5071	}
5072
5073	static ssize_t rbd_cluster_fsid_show(struct device *dev,
5074	struct device_attribute *attr, char *buf)
5075	{
5076	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5077
5078	return sprintf(buf, fmt: "%pU\n", &rbd_dev->rbd_client->client->fsid);
5079	}
5080
5081	static ssize_t rbd_config_info_show(struct device *dev,
5082	struct device_attribute *attr, char *buf)
5083	{
5084	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5085
5086	if (!capable(CAP_SYS_ADMIN))
5087	return -EPERM;
5088
5089	return sprintf(buf, fmt: "%s\n", rbd_dev->config_info);
5090	}
5091
5092	static ssize_t rbd_pool_show(struct device *dev,
5093	struct device_attribute *attr, char *buf)
5094	{
5095	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5096
5097	return sprintf(buf, fmt: "%s\n", rbd_dev->spec->pool_name);
5098	}
5099
5100	static ssize_t rbd_pool_id_show(struct device *dev,
5101	struct device_attribute *attr, char *buf)
5102	{
5103	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5104
5105	return sprintf(buf, fmt: "%llu\n",
5106	(unsigned long long) rbd_dev->spec->pool_id);
5107	}
5108
5109	static ssize_t rbd_pool_ns_show(struct device *dev,
5110	struct device_attribute *attr, char *buf)
5111	{
5112	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5113
5114	return sprintf(buf, fmt: "%s\n", rbd_dev->spec->pool_ns ?: "");
5115	}
5116
5117	static ssize_t rbd_name_show(struct device *dev,
5118	struct device_attribute *attr, char *buf)
5119	{
5120	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5121
5122	if (rbd_dev->spec->image_name)
5123	return sprintf(buf, fmt: "%s\n", rbd_dev->spec->image_name);
5124
5125	return sprintf(buf, fmt: "(unknown)\n");
5126	}
5127
5128	static ssize_t rbd_image_id_show(struct device *dev,
5129	struct device_attribute *attr, char *buf)
5130	{
5131	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5132
5133	return sprintf(buf, fmt: "%s\n", rbd_dev->spec->image_id);
5134	}
5135
5136	/*
5137	* Shows the name of the currently-mapped snapshot (or
5138	* RBD_SNAP_HEAD_NAME for the base image).
5139	*/
5140	static ssize_t rbd_snap_show(struct device *dev,
5141	struct device_attribute *attr,
5142	char *buf)
5143	{
5144	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5145
5146	return sprintf(buf, fmt: "%s\n", rbd_dev->spec->snap_name);
5147	}
5148
5149	static ssize_t rbd_snap_id_show(struct device *dev,
5150	struct device_attribute *attr, char *buf)
5151	{
5152	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5153
5154	return sprintf(buf, fmt: "%llu\n", rbd_dev->spec->snap_id);
5155	}
5156
5157	/*
5158	* For a v2 image, shows the chain of parent images, separated by empty
5159	* lines. For v1 images or if there is no parent, shows "(no parent
5160	* image)".
5161	*/
5162	static ssize_t rbd_parent_show(struct device *dev,
5163	struct device_attribute *attr,
5164	char *buf)
5165	{
5166	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5167	ssize_t count = 0;
5168
5169	if (!rbd_dev->parent)
5170	return sprintf(buf, fmt: "(no parent image)\n");
5171
5172	for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) {
5173	struct rbd_spec *spec = rbd_dev->parent_spec;
5174
5175	count += sprintf(buf: &buf[count], fmt: "%s"
5176	"pool_id %llu\npool_name %s\n"
5177	"pool_ns %s\n"
5178	"image_id %s\nimage_name %s\n"
5179	"snap_id %llu\nsnap_name %s\n"
5180	"overlap %llu\n",
5181	!count ? "" : "\n", /* first? */
5182	spec->pool_id, spec->pool_name,
5183	spec->pool_ns ?: "",
5184	spec->image_id, spec->image_name ?: "(unknown)",
5185	spec->snap_id, spec->snap_name,
5186	rbd_dev->parent_overlap);
5187	}
5188
5189	return count;
5190	}
5191
5192	static ssize_t rbd_image_refresh(struct device *dev,
5193	struct device_attribute *attr,
5194	const char *buf,
5195	size_t size)
5196	{
5197	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5198	int ret;
5199
5200	if (!capable(CAP_SYS_ADMIN))
5201	return -EPERM;
5202
5203	ret = rbd_dev_refresh(rbd_dev);
5204	if (ret)
5205	return ret;
5206
5207	return size;
5208	}
5209
5210	static DEVICE_ATTR(size, 0444, rbd_size_show, NULL);
5211	static DEVICE_ATTR(features, 0444, rbd_features_show, NULL);
5212	static DEVICE_ATTR(major, 0444, rbd_major_show, NULL);
5213	static DEVICE_ATTR(minor, 0444, rbd_minor_show, NULL);
5214	static DEVICE_ATTR(client_addr, 0444, rbd_client_addr_show, NULL);
5215	static DEVICE_ATTR(client_id, 0444, rbd_client_id_show, NULL);
5216	static DEVICE_ATTR(cluster_fsid, 0444, rbd_cluster_fsid_show, NULL);
5217	static DEVICE_ATTR(config_info, 0400, rbd_config_info_show, NULL);
5218	static DEVICE_ATTR(pool, 0444, rbd_pool_show, NULL);
5219	static DEVICE_ATTR(pool_id, 0444, rbd_pool_id_show, NULL);
5220	static DEVICE_ATTR(pool_ns, 0444, rbd_pool_ns_show, NULL);
5221	static DEVICE_ATTR(name, 0444, rbd_name_show, NULL);
5222	static DEVICE_ATTR(image_id, 0444, rbd_image_id_show, NULL);
5223	static DEVICE_ATTR(refresh, 0200, NULL, rbd_image_refresh);
5224	static DEVICE_ATTR(current_snap, 0444, rbd_snap_show, NULL);
5225	static DEVICE_ATTR(snap_id, 0444, rbd_snap_id_show, NULL);
5226	static DEVICE_ATTR(parent, 0444, rbd_parent_show, NULL);
5227
5228	static struct attribute *rbd_attrs[] = {
5229	&dev_attr_size.attr,
5230	&dev_attr_features.attr,
5231	&dev_attr_major.attr,
5232	&dev_attr_minor.attr,
5233	&dev_attr_client_addr.attr,
5234	&dev_attr_client_id.attr,
5235	&dev_attr_cluster_fsid.attr,
5236	&dev_attr_config_info.attr,
5237	&dev_attr_pool.attr,
5238	&dev_attr_pool_id.attr,
5239	&dev_attr_pool_ns.attr,
5240	&dev_attr_name.attr,
5241	&dev_attr_image_id.attr,
5242	&dev_attr_current_snap.attr,
5243	&dev_attr_snap_id.attr,
5244	&dev_attr_parent.attr,
5245	&dev_attr_refresh.attr,
5246	NULL
5247	};
5248
5249	static struct attribute_group rbd_attr_group = {
5250	.attrs = rbd_attrs,
5251	};
5252
5253	static const struct attribute_group *rbd_attr_groups[] = {
5254	&rbd_attr_group,
5255	NULL
5256	};
5257
5258	static void rbd_dev_release(struct device *dev);
5259
5260	static const struct device_type rbd_device_type = {
5261	.name = "rbd",
5262	.groups = rbd_attr_groups,
5263	.release = rbd_dev_release,
5264	};
5265
5266	static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
5267	{
5268	kref_get(kref: &spec->kref);
5269
5270	return spec;
5271	}
5272
5273	static void rbd_spec_free(struct kref *kref);
5274	static void rbd_spec_put(struct rbd_spec *spec)
5275	{
5276	if (spec)
5277	kref_put(kref: &spec->kref, release: rbd_spec_free);
5278	}
5279
5280	static struct rbd_spec *rbd_spec_alloc(void)
5281	{
5282	struct rbd_spec *spec;
5283
5284	spec = kzalloc(size: sizeof (*spec), GFP_KERNEL);
5285	if (!spec)
5286	return NULL;
5287
5288	spec->pool_id = CEPH_NOPOOL;
5289	spec->snap_id = CEPH_NOSNAP;
5290	kref_init(kref: &spec->kref);
5291
5292	return spec;
5293	}
5294
5295	static void rbd_spec_free(struct kref *kref)
5296	{
5297	struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
5298
5299	kfree(objp: spec->pool_name);
5300	kfree(objp: spec->pool_ns);
5301	kfree(objp: spec->image_id);
5302	kfree(objp: spec->image_name);
5303	kfree(objp: spec->snap_name);
5304	kfree(objp: spec);
5305	}
5306
5307	static void rbd_dev_free(struct rbd_device *rbd_dev)
5308	{
5309	WARN_ON(rbd_dev->watch_state != RBD_WATCH_STATE_UNREGISTERED);
5310	WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_UNLOCKED);
5311
5312	ceph_oid_destroy(oid: &rbd_dev->header_oid);
5313	ceph_oloc_destroy(oloc: &rbd_dev->header_oloc);
5314	kfree(objp: rbd_dev->config_info);
5315
5316	rbd_put_client(rbdc: rbd_dev->rbd_client);
5317	rbd_spec_put(spec: rbd_dev->spec);
5318	kfree(objp: rbd_dev->opts);
5319	kfree(objp: rbd_dev);
5320	}
5321
5322	static void rbd_dev_release(struct device *dev)
5323	{
5324	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5325	bool need_put = !!rbd_dev->opts;
5326
5327	if (need_put) {
5328	destroy_workqueue(wq: rbd_dev->task_wq);
5329	ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5330	}
5331
5332	rbd_dev_free(rbd_dev);
5333
5334	/*
5335	* This is racy, but way better than putting module outside of
5336	* the release callback. The race window is pretty small, so
5337	* doing something similar to dm (dm-builtin.c) is overkill.
5338	*/
5339	if (need_put)
5340	module_put(THIS_MODULE);
5341	}
5342
5343	static struct rbd_device *__rbd_dev_create(struct rbd_spec *spec)
5344	{
5345	struct rbd_device *rbd_dev;
5346
5347	rbd_dev = kzalloc(size: sizeof(*rbd_dev), GFP_KERNEL);
5348	if (!rbd_dev)
5349	return NULL;
5350
5351	spin_lock_init(&rbd_dev->lock);
5352	INIT_LIST_HEAD(list: &rbd_dev->node);
5353	init_rwsem(&rbd_dev->header_rwsem);
5354
5355	rbd_dev->header.data_pool_id = CEPH_NOPOOL;
5356	ceph_oid_init(oid: &rbd_dev->header_oid);
5357	rbd_dev->header_oloc.pool = spec->pool_id;
5358	if (spec->pool_ns) {
5359	WARN_ON(!*spec->pool_ns);
5360	rbd_dev->header_oloc.pool_ns =
5361	ceph_find_or_create_string(str: spec->pool_ns,
5362	strlen(spec->pool_ns));
5363	}
5364
5365	mutex_init(&rbd_dev->watch_mutex);
5366	rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
5367	INIT_DELAYED_WORK(&rbd_dev->watch_dwork, rbd_reregister_watch);
5368
5369	init_rwsem(&rbd_dev->lock_rwsem);
5370	rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
5371	INIT_WORK(&rbd_dev->acquired_lock_work, rbd_notify_acquired_lock);
5372	INIT_WORK(&rbd_dev->released_lock_work, rbd_notify_released_lock);
5373	INIT_DELAYED_WORK(&rbd_dev->lock_dwork, rbd_acquire_lock);
5374	INIT_WORK(&rbd_dev->unlock_work, rbd_release_lock_work);
5375	spin_lock_init(&rbd_dev->lock_lists_lock);
5376	INIT_LIST_HEAD(list: &rbd_dev->acquiring_list);
5377	INIT_LIST_HEAD(list: &rbd_dev->running_list);
5378	init_completion(x: &rbd_dev->acquire_wait);
5379	init_completion(x: &rbd_dev->releasing_wait);
5380
5381	spin_lock_init(&rbd_dev->object_map_lock);
5382
5383	rbd_dev->dev.bus = &rbd_bus_type;
5384	rbd_dev->dev.type = &rbd_device_type;
5385	rbd_dev->dev.parent = &rbd_root_dev;
5386	device_initialize(dev: &rbd_dev->dev);
5387
5388	return rbd_dev;
5389	}
5390
5391	/*
5392	* Create a mapping rbd_dev.
5393	*/
5394	static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
5395	struct rbd_spec *spec,
5396	struct rbd_options *opts)
5397	{
5398	struct rbd_device *rbd_dev;
5399
5400	rbd_dev = __rbd_dev_create(spec);
5401	if (!rbd_dev)
5402	return NULL;
5403
5404	/* get an id and fill in device name */
5405	rbd_dev->dev_id = ida_simple_get(&rbd_dev_id_ida, 0,
5406	minor_to_rbd_dev_id(1 << MINORBITS),
5407	GFP_KERNEL);
5408	if (rbd_dev->dev_id < 0)
5409	goto fail_rbd_dev;
5410
5411	sprintf(buf: rbd_dev->name, RBD_DRV_NAME "%d", rbd_dev->dev_id);
5412	rbd_dev->task_wq = alloc_ordered_workqueue("%s-tasks", WQ_MEM_RECLAIM,
5413	rbd_dev->name);
5414	if (!rbd_dev->task_wq)
5415	goto fail_dev_id;
5416
5417	/* we have a ref from do_rbd_add() */
5418	__module_get(THIS_MODULE);
5419
5420	rbd_dev->rbd_client = rbdc;
5421	rbd_dev->spec = spec;
5422	rbd_dev->opts = opts;
5423
5424	dout("%s rbd_dev %p dev_id %d\n", __func__, rbd_dev, rbd_dev->dev_id);
5425	return rbd_dev;
5426
5427	fail_dev_id:
5428	ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id);
5429	fail_rbd_dev:
5430	rbd_dev_free(rbd_dev);
5431	return NULL;
5432	}
5433
5434	static void rbd_dev_destroy(struct rbd_device *rbd_dev)
5435	{
5436	if (rbd_dev)
5437	put_device(dev: &rbd_dev->dev);
5438	}
5439
5440	/*
5441	* Get the size and object order for an image snapshot, or if
5442	* snap_id is CEPH_NOSNAP, gets this information for the base
5443	* image.
5444	*/
5445	static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
5446	u8 *order, u64 *snap_size)
5447	{
5448	__le64 snapid = cpu_to_le64(snap_id);
5449	int ret;
5450	struct {
5451	u8 order;
5452	__le64 size;
5453	} __attribute__ ((packed)) size_buf = { 0 };
5454
5455	ret = rbd_obj_method_sync(rbd_dev, oid: &rbd_dev->header_oid,
5456	oloc: &rbd_dev->header_oloc, method_name: "get_size",
5457	outbound: &snapid, outbound_size: sizeof(snapid),
5458	inbound: &size_buf, inbound_size: sizeof(size_buf));
5459	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5460	if (ret < 0)
5461	return ret;
5462	if (ret < sizeof (size_buf))
5463	return -ERANGE;
5464
5465	if (order) {
5466	*order = size_buf.order;
5467	dout(" order %u", (unsigned int)*order);
5468	}
5469	*snap_size = le64_to_cpu(size_buf.size);
5470
5471	dout(" snap_id 0x%016llx snap_size = %llu\n",
5472	(unsigned long long)snap_id,
5473	(unsigned long long)*snap_size);
5474
5475	return 0;
5476	}
5477
5478	static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev,
5479	char **pobject_prefix)
5480	{
5481	size_t size;
5482	void *reply_buf;
5483	char *object_prefix;
5484	int ret;
5485	void *p;
5486
5487	/* Response will be an encoded string, which includes a length */
5488	size = sizeof(__le32) + RBD_OBJ_PREFIX_LEN_MAX;
5489	reply_buf = kzalloc(size, GFP_KERNEL);
5490	if (!reply_buf)
5491	return -ENOMEM;
5492
5493	ret = rbd_obj_method_sync(rbd_dev, oid: &rbd_dev->header_oid,
5494	oloc: &rbd_dev->header_oloc, method_name: "get_object_prefix",
5495	NULL, outbound_size: 0, inbound: reply_buf, inbound_size: size);
5496	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5497	if (ret < 0)
5498	goto out;
5499
5500	p = reply_buf;
5501	object_prefix = ceph_extract_encoded_string(p: &p, end: p + ret, NULL,
5502	GFP_NOIO);
5503	if (IS_ERR(ptr: object_prefix)) {
5504	ret = PTR_ERR(ptr: object_prefix);
5505	goto out;
5506	}
5507	ret = 0;
5508
5509	*pobject_prefix = object_prefix;
5510	dout(" object_prefix = %s\n", object_prefix);
5511	out:
5512	kfree(objp: reply_buf);
5513
5514	return ret;
5515	}
5516
5517	static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
5518	bool read_only, u64 *snap_features)
5519	{
5520	struct {
5521	__le64 snap_id;
5522	u8 read_only;
5523	} features_in;
5524	struct {
5525	__le64 features;
5526	__le64 incompat;
5527	} __attribute__ ((packed)) features_buf = { 0 };
5528	u64 unsup;
5529	int ret;
5530
5531	features_in.snap_id = cpu_to_le64(snap_id);
5532	features_in.read_only = read_only;
5533
5534	ret = rbd_obj_method_sync(rbd_dev, oid: &rbd_dev->header_oid,
5535	oloc: &rbd_dev->header_oloc, method_name: "get_features",
5536	outbound: &features_in, outbound_size: sizeof(features_in),
5537	inbound: &features_buf, inbound_size: sizeof(features_buf));
5538	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5539	if (ret < 0)
5540	return ret;
5541	if (ret < sizeof (features_buf))
5542	return -ERANGE;
5543
5544	unsup = le64_to_cpu(features_buf.incompat) & ~RBD_FEATURES_SUPPORTED;
5545	if (unsup) {
5546	rbd_warn(rbd_dev, fmt: "image uses unsupported features: 0x%llx",
5547	unsup);
5548	return -ENXIO;
5549	}
5550
5551	*snap_features = le64_to_cpu(features_buf.features);
5552
5553	dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
5554	(unsigned long long)snap_id,
5555	(unsigned long long)*snap_features,
5556	(unsigned long long)le64_to_cpu(features_buf.incompat));
5557
5558	return 0;
5559	}
5560
5561	/*
5562	* These are generic image flags, but since they are used only for
5563	* object map, store them in rbd_dev->object_map_flags.
5564	*
5565	* For the same reason, this function is called only on object map
5566	* (re)load and not on header refresh.
5567	*/
5568	static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev)
5569	{
5570	__le64 snapid = cpu_to_le64(rbd_dev->spec->snap_id);
5571	__le64 flags;
5572	int ret;
5573
5574	ret = rbd_obj_method_sync(rbd_dev, oid: &rbd_dev->header_oid,
5575	oloc: &rbd_dev->header_oloc, method_name: "get_flags",
5576	outbound: &snapid, outbound_size: sizeof(snapid),
5577	inbound: &flags, inbound_size: sizeof(flags));
5578	if (ret < 0)
5579	return ret;
5580	if (ret < sizeof(flags))
5581	return -EBADMSG;
5582
5583	rbd_dev->object_map_flags = le64_to_cpu(flags);
5584	return 0;
5585	}
5586
5587	struct parent_image_info {
5588	u64 pool_id;
5589	const char *pool_ns;
5590	const char *image_id;
5591	u64 snap_id;
5592
5593	bool has_overlap;
5594	u64 overlap;
5595	};
5596
5597	static void rbd_parent_info_cleanup(struct parent_image_info *pii)
5598	{
5599	kfree(objp: pii->pool_ns);
5600	kfree(objp: pii->image_id);
5601
5602	memset(pii, 0, sizeof(*pii));
5603	}
5604
5605	/*
5606	* The caller is responsible for @pii.
5607	*/
5608	static int decode_parent_image_spec(void **p, void *end,
5609	struct parent_image_info *pii)
5610	{
5611	u8 struct_v;
5612	u32 struct_len;
5613	int ret;
5614
5615	ret = ceph_start_decoding(p, end, v: 1, name: "ParentImageSpec",
5616	struct_v: &struct_v, struct_len: &struct_len);
5617	if (ret)
5618	return ret;
5619
5620	ceph_decode_64_safe(p, end, pii->pool_id, e_inval);
5621	pii->pool_ns = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5622	if (IS_ERR(ptr: pii->pool_ns)) {
5623	ret = PTR_ERR(ptr: pii->pool_ns);
5624	pii->pool_ns = NULL;
5625	return ret;
5626	}
5627	pii->image_id = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5628	if (IS_ERR(ptr: pii->image_id)) {
5629	ret = PTR_ERR(ptr: pii->image_id);
5630	pii->image_id = NULL;
5631	return ret;
5632	}
5633	ceph_decode_64_safe(p, end, pii->snap_id, e_inval);
5634	return 0;
5635
5636	e_inval:
5637	return -EINVAL;
5638	}
5639
5640	static int __get_parent_info(struct rbd_device *rbd_dev,
5641	struct page *req_page,
5642	struct page *reply_page,
5643	struct parent_image_info *pii)
5644	{
5645	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5646	size_t reply_len = PAGE_SIZE;
5647	void *p, *end;
5648	int ret;
5649
5650	ret = ceph_osdc_call(osdc, oid: &rbd_dev->header_oid, oloc: &rbd_dev->header_oloc,
5651	class: "rbd", method: "parent_get", flags: CEPH_OSD_FLAG_READ,
5652	req_page, req_len: sizeof(u64), resp_pages: &reply_page, resp_len: &reply_len);
5653	if (ret)
5654	return ret == -EOPNOTSUPP ? 1 : ret;
5655
5656	p = page_address(reply_page);
5657	end = p + reply_len;
5658	ret = decode_parent_image_spec(p: &p, end, pii);
5659	if (ret)
5660	return ret;
5661
5662	ret = ceph_osdc_call(osdc, oid: &rbd_dev->header_oid, oloc: &rbd_dev->header_oloc,
5663	class: "rbd", method: "parent_overlap_get", flags: CEPH_OSD_FLAG_READ,
5664	req_page, req_len: sizeof(u64), resp_pages: &reply_page, resp_len: &reply_len);
5665	if (ret)
5666	return ret;
5667
5668	p = page_address(reply_page);
5669	end = p + reply_len;
5670	ceph_decode_8_safe(&p, end, pii->has_overlap, e_inval);
5671	if (pii->has_overlap)
5672	ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5673
5674	dout("%s pool_id %llu pool_ns %s image_id %s snap_id %llu has_overlap %d overlap %llu\n",
5675	__func__, pii->pool_id, pii->pool_ns, pii->image_id, pii->snap_id,
5676	pii->has_overlap, pii->overlap);
5677	return 0;
5678
5679	e_inval:
5680	return -EINVAL;
5681	}
5682
5683	/*
5684	* The caller is responsible for @pii.
5685	*/
5686	static int __get_parent_info_legacy(struct rbd_device *rbd_dev,
5687	struct page *req_page,
5688	struct page *reply_page,
5689	struct parent_image_info *pii)
5690	{
5691	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5692	size_t reply_len = PAGE_SIZE;
5693	void *p, *end;
5694	int ret;
5695
5696	ret = ceph_osdc_call(osdc, oid: &rbd_dev->header_oid, oloc: &rbd_dev->header_oloc,
5697	class: "rbd", method: "get_parent", flags: CEPH_OSD_FLAG_READ,
5698	req_page, req_len: sizeof(u64), resp_pages: &reply_page, resp_len: &reply_len);
5699	if (ret)
5700	return ret;
5701
5702	p = page_address(reply_page);
5703	end = p + reply_len;
5704	ceph_decode_64_safe(&p, end, pii->pool_id, e_inval);
5705	pii->image_id = ceph_extract_encoded_string(p: &p, end, NULL, GFP_KERNEL);
5706	if (IS_ERR(ptr: pii->image_id)) {
5707	ret = PTR_ERR(ptr: pii->image_id);
5708	pii->image_id = NULL;
5709	return ret;
5710	}
5711	ceph_decode_64_safe(&p, end, pii->snap_id, e_inval);
5712	pii->has_overlap = true;
5713	ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5714
5715	dout("%s pool_id %llu pool_ns %s image_id %s snap_id %llu has_overlap %d overlap %llu\n",
5716	__func__, pii->pool_id, pii->pool_ns, pii->image_id, pii->snap_id,
5717	pii->has_overlap, pii->overlap);
5718	return 0;
5719
5720	e_inval:
5721	return -EINVAL;
5722	}
5723
5724	static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev,
5725	struct parent_image_info *pii)
5726	{
5727	struct page *req_page, *reply_page;
5728	void *p;
5729	int ret;
5730
5731	req_page = alloc_page(GFP_KERNEL);
5732	if (!req_page)
5733	return -ENOMEM;
5734
5735	reply_page = alloc_page(GFP_KERNEL);
5736	if (!reply_page) {
5737	__free_page(req_page);
5738	return -ENOMEM;
5739	}
5740
5741	p = page_address(req_page);
5742	ceph_encode_64(p: &p, v: rbd_dev->spec->snap_id);
5743	ret = __get_parent_info(rbd_dev, req_page, reply_page, pii);
5744	if (ret > 0)
5745	ret = __get_parent_info_legacy(rbd_dev, req_page, reply_page,
5746	pii);
5747
5748	__free_page(req_page);
5749	__free_page(reply_page);
5750	return ret;
5751	}
5752
5753	static int rbd_dev_setup_parent(struct rbd_device *rbd_dev)
5754	{
5755	struct rbd_spec *parent_spec;
5756	struct parent_image_info pii = { 0 };
5757	int ret;
5758
5759	parent_spec = rbd_spec_alloc();
5760	if (!parent_spec)
5761	return -ENOMEM;
5762
5763	ret = rbd_dev_v2_parent_info(rbd_dev, pii: &pii);
5764	if (ret)
5765	goto out_err;
5766
5767	if (pii.pool_id == CEPH_NOPOOL || !pii.has_overlap)
5768	goto out; /* No parent? No problem. */
5769
5770	/* The ceph file layout needs to fit pool id in 32 bits */
5771
5772	ret = -EIO;
5773	if (pii.pool_id > (u64)U32_MAX) {
5774	rbd_warn(NULL, fmt: "parent pool id too large (%llu > %u)",
5775	(unsigned long long)pii.pool_id, U32_MAX);
5776	goto out_err;
5777	}
5778
5779	/*
5780	* The parent won't change except when the clone is flattened,
5781	* so we only need to record the parent image spec once.
5782	*/
5783	parent_spec->pool_id = pii.pool_id;
5784	if (pii.pool_ns && *pii.pool_ns) {
5785	parent_spec->pool_ns = pii.pool_ns;
5786	pii.pool_ns = NULL;
5787	}
5788	parent_spec->image_id = pii.image_id;
5789	pii.image_id = NULL;
5790	parent_spec->snap_id = pii.snap_id;
5791
5792	rbd_assert(!rbd_dev->parent_spec);
5793	rbd_dev->parent_spec = parent_spec;
5794	parent_spec = NULL; /* rbd_dev now owns this */
5795
5796	/*
5797	* Record the parent overlap. If it's zero, issue a warning as
5798	* we will proceed as if there is no parent.
5799	*/
5800	if (!pii.overlap)
5801	rbd_warn(rbd_dev, fmt: "clone is standalone (overlap 0)");
5802	rbd_dev->parent_overlap = pii.overlap;
5803
5804	out:
5805	ret = 0;
5806	out_err:
5807	rbd_parent_info_cleanup(pii: &pii);
5808	rbd_spec_put(spec: parent_spec);
5809	return ret;
5810	}
5811
5812	static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev,
5813	u64 *stripe_unit, u64 *stripe_count)
5814	{
5815	struct {
5816	__le64 stripe_unit;
5817	__le64 stripe_count;
5818	} __attribute__ ((packed)) striping_info_buf = { 0 };
5819	size_t size = sizeof (striping_info_buf);
5820	int ret;
5821
5822	ret = rbd_obj_method_sync(rbd_dev, oid: &rbd_dev->header_oid,
5823	oloc: &rbd_dev->header_oloc, method_name: "get_stripe_unit_count",
5824	NULL, outbound_size: 0, inbound: &striping_info_buf, inbound_size: size);
5825	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5826	if (ret < 0)
5827	return ret;
5828	if (ret < size)
5829	return -ERANGE;
5830
5831	*stripe_unit = le64_to_cpu(striping_info_buf.stripe_unit);
5832	*stripe_count = le64_to_cpu(striping_info_buf.stripe_count);
5833	dout(" stripe_unit = %llu stripe_count = %llu\n", *stripe_unit,
5834	*stripe_count);
5835
5836	return 0;
5837	}
5838
5839	static int rbd_dev_v2_data_pool(struct rbd_device *rbd_dev, s64 *data_pool_id)
5840	{
5841	__le64 data_pool_buf;
5842	int ret;
5843
5844	ret = rbd_obj_method_sync(rbd_dev, oid: &rbd_dev->header_oid,
5845	oloc: &rbd_dev->header_oloc, method_name: "get_data_pool",
5846	NULL, outbound_size: 0, inbound: &data_pool_buf,
5847	inbound_size: sizeof(data_pool_buf));
5848	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5849	if (ret < 0)
5850	return ret;
5851	if (ret < sizeof(data_pool_buf))
5852	return -EBADMSG;
5853
5854	*data_pool_id = le64_to_cpu(data_pool_buf);
5855	dout(" data_pool_id = %lld\n", *data_pool_id);
5856	WARN_ON(*data_pool_id == CEPH_NOPOOL);
5857
5858	return 0;
5859	}
5860
5861	static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
5862	{
5863	CEPH_DEFINE_OID_ONSTACK(oid);
5864	size_t image_id_size;
5865	char *image_id;
5866	void *p;
5867	void *end;
5868	size_t size;
5869	void *reply_buf = NULL;
5870	size_t len = 0;
5871	char *image_name = NULL;
5872	int ret;
5873
5874	rbd_assert(!rbd_dev->spec->image_name);
5875
5876	len = strlen(rbd_dev->spec->image_id);
5877	image_id_size = sizeof (__le32) + len;
5878	image_id = kmalloc(size: image_id_size, GFP_KERNEL);
5879	if (!image_id)
5880	return NULL;
5881
5882	p = image_id;
5883	end = image_id + image_id_size;
5884	ceph_encode_string(p: &p, end, s: rbd_dev->spec->image_id, len: (u32)len);
5885
5886	size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
5887	reply_buf = kmalloc(size, GFP_KERNEL);
5888	if (!reply_buf)
5889	goto out;
5890
5891	ceph_oid_printf(oid: &oid, fmt: "%s", RBD_DIRECTORY);
5892	ret = rbd_obj_method_sync(rbd_dev, oid: &oid, oloc: &rbd_dev->header_oloc,
5893	method_name: "dir_get_name", outbound: image_id, outbound_size: image_id_size,
5894	inbound: reply_buf, inbound_size: size);
5895	if (ret < 0)
5896	goto out;
5897	p = reply_buf;
5898	end = reply_buf + ret;
5899
5900	image_name = ceph_extract_encoded_string(p: &p, end, lenp: &len, GFP_KERNEL);
5901	if (IS_ERR(ptr: image_name))
5902	image_name = NULL;
5903	else
5904	dout("%s: name is %s len is %zd\n", __func__, image_name, len);
5905	out:
5906	kfree(objp: reply_buf);
5907	kfree(objp: image_id);
5908
5909	return image_name;
5910	}
5911
5912	static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5913	{
5914	struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5915	const char *snap_name;
5916	u32 which = 0;
5917
5918	/* Skip over names until we find the one we are looking for */
5919
5920	snap_name = rbd_dev->header.snap_names;
5921	while (which < snapc->num_snaps) {
5922	if (!strcmp(name, snap_name))
5923	return snapc->snaps[which];
5924	snap_name += strlen(snap_name) + 1;
5925	which++;
5926	}
5927	return CEPH_NOSNAP;
5928	}
5929
5930	static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5931	{
5932	struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5933	u32 which;
5934	bool found = false;
5935	u64 snap_id;
5936
5937	for (which = 0; !found && which < snapc->num_snaps; which++) {
5938	const char *snap_name;
5939
5940	snap_id = snapc->snaps[which];
5941	snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
5942	if (IS_ERR(ptr: snap_name)) {
5943	/* ignore no-longer existing snapshots */
5944	if (PTR_ERR(ptr: snap_name) == -ENOENT)
5945	continue;
5946	else
5947	break;
5948	}
5949	found = !strcmp(name, snap_name);
5950	kfree(objp: snap_name);
5951	}
5952	return found ? snap_id : CEPH_NOSNAP;
5953	}
5954
5955	/*
5956	* Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if
5957	* no snapshot by that name is found, or if an error occurs.
5958	*/
5959	static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5960	{
5961	if (rbd_dev->image_format == 1)
5962	return rbd_v1_snap_id_by_name(rbd_dev, name);
5963
5964	return rbd_v2_snap_id_by_name(rbd_dev, name);
5965	}
5966
5967	/*
5968	* An image being mapped will have everything but the snap id.
5969	*/
5970	static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev)
5971	{
5972	struct rbd_spec *spec = rbd_dev->spec;
5973
5974	rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name);
5975	rbd_assert(spec->image_id && spec->image_name);
5976	rbd_assert(spec->snap_name);
5977
5978	if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) {
5979	u64 snap_id;
5980
5981	snap_id = rbd_snap_id_by_name(rbd_dev, name: spec->snap_name);
5982	if (snap_id == CEPH_NOSNAP)
5983	return -ENOENT;
5984
5985	spec->snap_id = snap_id;
5986	} else {
5987	spec->snap_id = CEPH_NOSNAP;
5988	}
5989
5990	return 0;
5991	}
5992
5993	/*
5994	* A parent image will have all ids but none of the names.
5995	*
5996	* All names in an rbd spec are dynamically allocated. It's OK if we
5997	* can't figure out the name for an image id.
5998	*/
5999	static int rbd_spec_fill_names(struct rbd_device *rbd_dev)
6000	{
6001	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
6002	struct rbd_spec *spec = rbd_dev->spec;
6003	const char *pool_name;
6004	const char *image_name;
6005	const char *snap_name;
6006	int ret;
6007
6008	rbd_assert(spec->pool_id != CEPH_NOPOOL);
6009	rbd_assert(spec->image_id);
6010	rbd_assert(spec->snap_id != CEPH_NOSNAP);
6011
6012	/* Get the pool name; we have to make our own copy of this */
6013
6014	pool_name = ceph_pg_pool_name_by_id(map: osdc->osdmap, id: spec->pool_id);
6015	if (!pool_name) {
6016	rbd_warn(rbd_dev, fmt: "no pool with id %llu", spec->pool_id);
6017	return -EIO;
6018	}
6019	pool_name = kstrdup(s: pool_name, GFP_KERNEL);
6020	if (!pool_name)
6021	return -ENOMEM;
6022
6023	/* Fetch the image name; tolerate failure here */
6024
6025	image_name = rbd_dev_image_name(rbd_dev);
6026	if (!image_name)
6027	rbd_warn(rbd_dev, fmt: "unable to get image name");
6028
6029	/* Fetch the snapshot name */
6030
6031	snap_name = rbd_snap_name(rbd_dev, snap_id: spec->snap_id);
6032	if (IS_ERR(ptr: snap_name)) {
6033	ret = PTR_ERR(ptr: snap_name);
6034	goto out_err;
6035	}
6036
6037	spec->pool_name = pool_name;
6038	spec->image_name = image_name;
6039	spec->snap_name = snap_name;
6040
6041	return 0;
6042
6043	out_err:
6044	kfree(objp: image_name);
6045	kfree(objp: pool_name);
6046	return ret;
6047	}
6048
6049	static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev,
6050	struct ceph_snap_context **psnapc)
6051	{
6052	size_t size;
6053	int ret;
6054	void *reply_buf;
6055	void *p;
6056	void *end;
6057	u64 seq;
6058	u32 snap_count;
6059	struct ceph_snap_context *snapc;
6060	u32 i;
6061
6062	/*
6063	* We'll need room for the seq value (maximum snapshot id),
6064	* snapshot count, and array of that many snapshot ids.
6065	* For now we have a fixed upper limit on the number we're
6066	* prepared to receive.
6067	*/
6068	size = sizeof (__le64) + sizeof (__le32) +
6069	RBD_MAX_SNAP_COUNT * sizeof (__le64);
6070	reply_buf = kzalloc(size, GFP_KERNEL);
6071	if (!reply_buf)
6072	return -ENOMEM;
6073
6074	ret = rbd_obj_method_sync(rbd_dev, oid: &rbd_dev->header_oid,
6075	oloc: &rbd_dev->header_oloc, method_name: "get_snapcontext",
6076	NULL, outbound_size: 0, inbound: reply_buf, inbound_size: size);
6077	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6078	if (ret < 0)
6079	goto out;
6080
6081	p = reply_buf;
6082	end = reply_buf + ret;
6083	ret = -ERANGE;
6084	ceph_decode_64_safe(&p, end, seq, out);
6085	ceph_decode_32_safe(&p, end, snap_count, out);
6086
6087	/*
6088	* Make sure the reported number of snapshot ids wouldn't go
6089	* beyond the end of our buffer. But before checking that,
6090	* make sure the computed size of the snapshot context we
6091	* allocate is representable in a size_t.
6092	*/
6093	if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
6094	/ sizeof (u64)) {
6095	ret = -EINVAL;
6096	goto out;
6097	}
6098	if (!ceph_has_room(p: &p, end, n: snap_count * sizeof (__le64)))
6099	goto out;
6100	ret = 0;
6101
6102	snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
6103	if (!snapc) {
6104	ret = -ENOMEM;
6105	goto out;
6106	}
6107	snapc->seq = seq;
6108	for (i = 0; i < snap_count; i++)
6109	snapc->snaps[i] = ceph_decode_64(p: &p);
6110
6111	*psnapc = snapc;
6112	dout(" snap context seq = %llu, snap_count = %u\n",
6113	(unsigned long long)seq, (unsigned int)snap_count);
6114	out:
6115	kfree(objp: reply_buf);
6116
6117	return ret;
6118	}
6119
6120	static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
6121	u64 snap_id)
6122	{
6123	size_t size;
6124	void *reply_buf;
6125	__le64 snapid;
6126	int ret;
6127	void *p;
6128	void *end;
6129	char *snap_name;
6130
6131	size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
6132	reply_buf = kmalloc(size, GFP_KERNEL);
6133	if (!reply_buf)
6134	return ERR_PTR(error: -ENOMEM);
6135
6136	snapid = cpu_to_le64(snap_id);
6137	ret = rbd_obj_method_sync(rbd_dev, oid: &rbd_dev->header_oid,
6138	oloc: &rbd_dev->header_oloc, method_name: "get_snapshot_name",
6139	outbound: &snapid, outbound_size: sizeof(snapid), inbound: reply_buf, inbound_size: size);
6140	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6141	if (ret < 0) {
6142	snap_name = ERR_PTR(error: ret);
6143	goto out;
6144	}
6145
6146	p = reply_buf;
6147	end = reply_buf + ret;
6148	snap_name = ceph_extract_encoded_string(p: &p, end, NULL, GFP_KERNEL);
6149	if (IS_ERR(ptr: snap_name))
6150	goto out;
6151
6152	dout(" snap_id 0x%016llx snap_name = %s\n",
6153	(unsigned long long)snap_id, snap_name);
6154	out:
6155	kfree(objp: reply_buf);
6156
6157	return snap_name;
6158	}
6159
6160	static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev,
6161	struct rbd_image_header *header,
6162	bool first_time)
6163	{
6164	int ret;
6165
6166	ret = _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
6167	order: first_time ? &header->obj_order : NULL,
6168	snap_size: &header->image_size);
6169	if (ret)
6170	return ret;
6171
6172	if (first_time) {
6173	ret = rbd_dev_v2_header_onetime(rbd_dev, header);
6174	if (ret)
6175	return ret;
6176	}
6177
6178	ret = rbd_dev_v2_snap_context(rbd_dev, psnapc: &header->snapc);
6179	if (ret)
6180	return ret;
6181
6182	return 0;
6183	}
6184
6185	static int rbd_dev_header_info(struct rbd_device *rbd_dev,
6186	struct rbd_image_header *header,
6187	bool first_time)
6188	{
6189	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6190	rbd_assert(!header->object_prefix && !header->snapc);
6191
6192	if (rbd_dev->image_format == 1)
6193	return rbd_dev_v1_header_info(rbd_dev, header, first_time);
6194
6195	return rbd_dev_v2_header_info(rbd_dev, header, first_time);
6196	}
6197
6198	/*
6199	* Skips over white space at *buf, and updates *buf to point to the
6200	* first found non-space character (if any). Returns the length of
6201	* the token (string of non-white space characters) found. Note
6202	* that *buf must be terminated with '\0'.
6203	*/
6204	static inline size_t next_token(const char **buf)
6205	{
6206	/*
6207	* These are the characters that produce nonzero for
6208	* isspace() in the "C" and "POSIX" locales.
6209	*/
6210	static const char spaces[] = " \f\n\r\t\v";
6211
6212	*buf += strspn(*buf, spaces); /* Find start of token */
6213
6214	return strcspn(*buf, spaces); /* Return token length */
6215	}
6216
6217	/*
6218	* Finds the next token in *buf, dynamically allocates a buffer big
6219	* enough to hold a copy of it, and copies the token into the new
6220	* buffer. The copy is guaranteed to be terminated with '\0'. Note
6221	* that a duplicate buffer is created even for a zero-length token.
6222	*
6223	* Returns a pointer to the newly-allocated duplicate, or a null
6224	* pointer if memory for the duplicate was not available. If
6225	* the lenp argument is a non-null pointer, the length of the token
6226	* (not including the '\0') is returned in *lenp.
6227	*
6228	* If successful, the *buf pointer will be updated to point beyond
6229	* the end of the found token.
6230	*
6231	* Note: uses GFP_KERNEL for allocation.
6232	*/
6233	static inline char *dup_token(const char **buf, size_t *lenp)
6234	{
6235	char *dup;
6236	size_t len;
6237
6238	len = next_token(buf);
6239	dup = kmemdup(p: *buf, size: len + 1, GFP_KERNEL);
6240	if (!dup)
6241	return NULL;
6242	*(dup + len) = '\0';
6243	*buf += len;
6244
6245	if (lenp)
6246	*lenp = len;
6247
6248	return dup;
6249	}
6250
6251	static int rbd_parse_param(struct fs_parameter *param,
6252	struct rbd_parse_opts_ctx *pctx)
6253	{
6254	struct rbd_options *opt = pctx->opts;
6255	struct fs_parse_result result;
6256	struct p_log log = {.prefix = "rbd"};
6257	int token, ret;
6258
6259	ret = ceph_parse_param(param, opt: pctx->copts, NULL);
6260	if (ret != -ENOPARAM)
6261	return ret;
6262
6263	token = __fs_parse(log: &log, desc: rbd_parameters, value: param, result: &result);
6264	dout("%s fs_parse '%s' token %d\n", __func__, param->key, token);
6265	if (token < 0) {
6266	if (token == -ENOPARAM)
6267	return inval_plog(&log, "Unknown parameter '%s'",
6268	param->key);
6269	return token;
6270	}
6271
6272	switch (token) {
6273	case Opt_queue_depth:
6274	if (result.uint_32 < 1)
6275	goto out_of_range;
6276	opt->queue_depth = result.uint_32;
6277	break;
6278	case Opt_alloc_size:
6279	if (result.uint_32 < SECTOR_SIZE)
6280	goto out_of_range;
6281	if (!is_power_of_2(n: result.uint_32))
6282	return inval_plog(&log, "alloc_size must be a power of 2");
6283	opt->alloc_size = result.uint_32;
6284	break;
6285	case Opt_lock_timeout:
6286	/* 0 is "wait forever" (i.e. infinite timeout) */
6287	if (result.uint_32 > INT_MAX / 1000)
6288	goto out_of_range;
6289	opt->lock_timeout = msecs_to_jiffies(m: result.uint_32 * 1000);
6290	break;
6291	case Opt_pool_ns:
6292	kfree(objp: pctx->spec->pool_ns);
6293	pctx->spec->pool_ns = param->string;
6294	param->string = NULL;
6295	break;
6296	case Opt_compression_hint:
6297	switch (result.uint_32) {
6298	case Opt_compression_hint_none:
6299	opt->alloc_hint_flags &=
6300	~(CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE |
6301	CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE);
6302	break;
6303	case Opt_compression_hint_compressible:
6304	opt->alloc_hint_flags |=
6305	CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6306	opt->alloc_hint_flags &=
6307	~CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6308	break;
6309	case Opt_compression_hint_incompressible:
6310	opt->alloc_hint_flags |=
6311	CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6312	opt->alloc_hint_flags &=
6313	~CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6314	break;
6315	default:
6316	BUG();
6317	}
6318	break;
6319	case Opt_read_only:
6320	opt->read_only = true;
6321	break;
6322	case Opt_read_write:
6323	opt->read_only = false;
6324	break;
6325	case Opt_lock_on_read:
6326	opt->lock_on_read = true;
6327	break;
6328	case Opt_exclusive:
6329	opt->exclusive = true;
6330	break;
6331	case Opt_notrim:
6332	opt->trim = false;
6333	break;
6334	default:
6335	BUG();
6336	}
6337
6338	return 0;
6339
6340	out_of_range:
6341	return inval_plog(&log, "%s out of range", param->key);
6342	}
6343
6344	/*
6345	* This duplicates most of generic_parse_monolithic(), untying it from
6346	* fs_context and skipping standard superblock and security options.
6347	*/
6348	static int rbd_parse_options(char *options, struct rbd_parse_opts_ctx *pctx)
6349	{
6350	char *key;
6351	int ret = 0;
6352
6353	dout("%s '%s'\n", __func__, options);
6354	while ((key = strsep(&options, ",")) != NULL) {
6355	if (*key) {
6356	struct fs_parameter param = {
6357	.key = key,
6358	.type = fs_value_is_flag,
6359	};
6360	char *value = strchr(key, '=');
6361	size_t v_len = 0;
6362
6363	if (value) {
6364	if (value == key)
6365	continue;
6366	*value++ = 0;
6367	v_len = strlen(value);
6368	param.string = kmemdup_nul(s: value, len: v_len,
6369	GFP_KERNEL);
6370	if (!param.string)
6371	return -ENOMEM;
6372	param.type = fs_value_is_string;
6373	}
6374	param.size = v_len;
6375
6376	ret = rbd_parse_param(param: &param, pctx);
6377	kfree(objp: param.string);
6378	if (ret)
6379	break;
6380	}
6381	}
6382
6383	return ret;
6384	}
6385
6386	/*
6387	* Parse the options provided for an "rbd add" (i.e., rbd image
6388	* mapping) request. These arrive via a write to /sys/bus/rbd/add,
6389	* and the data written is passed here via a NUL-terminated buffer.
6390	* Returns 0 if successful or an error code otherwise.
6391	*
6392	* The information extracted from these options is recorded in
6393	* the other parameters which return dynamically-allocated
6394	* structures:
6395	* ceph_opts
6396	* The address of a pointer that will refer to a ceph options
6397	* structure. Caller must release the returned pointer using
6398	* ceph_destroy_options() when it is no longer needed.
6399	* rbd_opts
6400	* Address of an rbd options pointer. Fully initialized by
6401	* this function; caller must release with kfree().
6402	* spec
6403	* Address of an rbd image specification pointer. Fully
6404	* initialized by this function based on parsed options.
6405	* Caller must release with rbd_spec_put().
6406	*
6407	* The options passed take this form:
6408	* <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
6409	* where:
6410	* <mon_addrs>
6411	* A comma-separated list of one or more monitor addresses.
6412	* A monitor address is an ip address, optionally followed
6413	* by a port number (separated by a colon).
6414	* I.e.: ip1[:port1][,ip2[:port2]...]
6415	* <options>
6416	* A comma-separated list of ceph and/or rbd options.
6417	* <pool_name>
6418	* The name of the rados pool containing the rbd image.
6419	* <image_name>
6420	* The name of the image in that pool to map.
6421	* <snap_id>
6422	* An optional snapshot id. If provided, the mapping will
6423	* present data from the image at the time that snapshot was
6424	* created. The image head is used if no snapshot id is
6425	* provided. Snapshot mappings are always read-only.
6426	*/
6427	static int rbd_add_parse_args(const char *buf,
6428	struct ceph_options **ceph_opts,
6429	struct rbd_options **opts,
6430	struct rbd_spec **rbd_spec)
6431	{
6432	size_t len;
6433	char *options;
6434	const char *mon_addrs;
6435	char *snap_name;
6436	size_t mon_addrs_size;
6437	struct rbd_parse_opts_ctx pctx = { 0 };
6438	int ret;
6439
6440	/* The first four tokens are required */
6441
6442	len = next_token(buf: &buf);
6443	if (!len) {
6444	rbd_warn(NULL, fmt: "no monitor address(es) provided");
6445	return -EINVAL;
6446	}
6447	mon_addrs = buf;
6448	mon_addrs_size = len;
6449	buf += len;
6450
6451	ret = -EINVAL;
6452	options = dup_token(buf: &buf, NULL);
6453	if (!options)
6454	return -ENOMEM;
6455	if (!*options) {
6456	rbd_warn(NULL, fmt: "no options provided");
6457	goto out_err;
6458	}
6459
6460	pctx.spec = rbd_spec_alloc();
6461	if (!pctx.spec)
6462	goto out_mem;
6463
6464	pctx.spec->pool_name = dup_token(buf: &buf, NULL);
6465	if (!pctx.spec->pool_name)
6466	goto out_mem;
6467	if (!*pctx.spec->pool_name) {
6468	rbd_warn(NULL, fmt: "no pool name provided");
6469	goto out_err;
6470	}
6471
6472	pctx.spec->image_name = dup_token(buf: &buf, NULL);
6473	if (!pctx.spec->image_name)
6474	goto out_mem;
6475	if (!*pctx.spec->image_name) {
6476	rbd_warn(NULL, fmt: "no image name provided");
6477	goto out_err;
6478	}
6479
6480	/*
6481	* Snapshot name is optional; default is to use "-"
6482	* (indicating the head/no snapshot).
6483	*/
6484	len = next_token(buf: &buf);
6485	if (!len) {
6486	buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
6487	len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
6488	} else if (len > RBD_MAX_SNAP_NAME_LEN) {
6489	ret = -ENAMETOOLONG;
6490	goto out_err;
6491	}
6492	snap_name = kmemdup(p: buf, size: len + 1, GFP_KERNEL);
6493	if (!snap_name)
6494	goto out_mem;
6495	*(snap_name + len) = '\0';
6496	pctx.spec->snap_name = snap_name;
6497
6498	pctx.copts = ceph_alloc_options();
6499	if (!pctx.copts)
6500	goto out_mem;
6501
6502	/* Initialize all rbd options to the defaults */
6503
6504	pctx.opts = kzalloc(size: sizeof(*pctx.opts), GFP_KERNEL);
6505	if (!pctx.opts)
6506	goto out_mem;
6507
6508	pctx.opts->read_only = RBD_READ_ONLY_DEFAULT;
6509	pctx.opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
6510	pctx.opts->alloc_size = RBD_ALLOC_SIZE_DEFAULT;
6511	pctx.opts->lock_timeout = RBD_LOCK_TIMEOUT_DEFAULT;
6512	pctx.opts->lock_on_read = RBD_LOCK_ON_READ_DEFAULT;
6513	pctx.opts->exclusive = RBD_EXCLUSIVE_DEFAULT;
6514	pctx.opts->trim = RBD_TRIM_DEFAULT;
6515
6516	ret = ceph_parse_mon_ips(buf: mon_addrs, len: mon_addrs_size, opt: pctx.copts, NULL,
6517	delim: ',');
6518	if (ret)
6519	goto out_err;
6520
6521	ret = rbd_parse_options(options, pctx: &pctx);
6522	if (ret)
6523	goto out_err;
6524
6525	*ceph_opts = pctx.copts;
6526	*opts = pctx.opts;
6527	*rbd_spec = pctx.spec;
6528	kfree(objp: options);
6529	return 0;
6530
6531	out_mem:
6532	ret = -ENOMEM;
6533	out_err:
6534	kfree(objp: pctx.opts);
6535	ceph_destroy_options(opt: pctx.copts);
6536	rbd_spec_put(spec: pctx.spec);
6537	kfree(objp: options);
6538	return ret;
6539	}
6540
6541	static void rbd_dev_image_unlock(struct rbd_device *rbd_dev)
6542	{
6543	down_write(sem: &rbd_dev->lock_rwsem);
6544	if (__rbd_is_lock_owner(rbd_dev))
6545	__rbd_release_lock(rbd_dev);
6546	up_write(sem: &rbd_dev->lock_rwsem);
6547	}
6548
6549	/*
6550	* If the wait is interrupted, an error is returned even if the lock
6551	* was successfully acquired. rbd_dev_image_unlock() will release it
6552	* if needed.
6553	*/
6554	static int rbd_add_acquire_lock(struct rbd_device *rbd_dev)
6555	{
6556	long ret;
6557
6558	if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK)) {
6559	if (!rbd_dev->opts->exclusive && !rbd_dev->opts->lock_on_read)
6560	return 0;
6561
6562	rbd_warn(rbd_dev, fmt: "exclusive-lock feature is not enabled");
6563	return -EINVAL;
6564	}
6565
6566	if (rbd_is_ro(rbd_dev))
6567	return 0;
6568
6569	rbd_assert(!rbd_is_lock_owner(rbd_dev));
6570	queue_delayed_work(wq: rbd_dev->task_wq, dwork: &rbd_dev->lock_dwork, delay: 0);
6571	ret = wait_for_completion_killable_timeout(x: &rbd_dev->acquire_wait,
6572	timeout: ceph_timeout_jiffies(timeout: rbd_dev->opts->lock_timeout));
6573	if (ret > 0) {
6574	ret = rbd_dev->acquire_err;
6575	} else {
6576	cancel_delayed_work_sync(dwork: &rbd_dev->lock_dwork);
6577	if (!ret)
6578	ret = -ETIMEDOUT;
6579
6580	rbd_warn(rbd_dev, fmt: "failed to acquire lock: %ld", ret);
6581	}
6582	if (ret)
6583	return ret;
6584
6585	/*
6586	* The lock may have been released by now, unless automatic lock
6587	* transitions are disabled.
6588	*/
6589	rbd_assert(!rbd_dev->opts->exclusive || rbd_is_lock_owner(rbd_dev));
6590	return 0;
6591	}
6592
6593	/*
6594	* An rbd format 2 image has a unique identifier, distinct from the
6595	* name given to it by the user. Internally, that identifier is
6596	* what's used to specify the names of objects related to the image.
6597	*
6598	* A special "rbd id" object is used to map an rbd image name to its
6599	* id. If that object doesn't exist, then there is no v2 rbd image
6600	* with the supplied name.
6601	*
6602	* This function will record the given rbd_dev's image_id field if
6603	* it can be determined, and in that case will return 0. If any
6604	* errors occur a negative errno will be returned and the rbd_dev's
6605	* image_id field will be unchanged (and should be NULL).
6606	*/
6607	static int rbd_dev_image_id(struct rbd_device *rbd_dev)
6608	{
6609	int ret;
6610	size_t size;
6611	CEPH_DEFINE_OID_ONSTACK(oid);
6612	void *response;
6613	char *image_id;
6614
6615	/*
6616	* When probing a parent image, the image id is already
6617	* known (and the image name likely is not). There's no
6618	* need to fetch the image id again in this case. We
6619	* do still need to set the image format though.
6620	*/
6621	if (rbd_dev->spec->image_id) {
6622	rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
6623
6624	return 0;
6625	}
6626
6627	/*
6628	* First, see if the format 2 image id file exists, and if
6629	* so, get the image's persistent id from it.
6630	*/
6631	ret = ceph_oid_aprintf(oid: &oid, GFP_KERNEL, fmt: "%s%s", RBD_ID_PREFIX,
6632	rbd_dev->spec->image_name);
6633	if (ret)
6634	return ret;
6635
6636	dout("rbd id object name is %s\n", oid.name);
6637
6638	/* Response will be an encoded string, which includes a length */
6639	size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
6640	response = kzalloc(size, GFP_NOIO);
6641	if (!response) {
6642	ret = -ENOMEM;
6643	goto out;
6644	}
6645
6646	/* If it doesn't exist we'll assume it's a format 1 image */
6647
6648	ret = rbd_obj_method_sync(rbd_dev, oid: &oid, oloc: &rbd_dev->header_oloc,
6649	method_name: "get_id", NULL, outbound_size: 0,
6650	inbound: response, inbound_size: size);
6651	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6652	if (ret == -ENOENT) {
6653	image_id = kstrdup(s: "", GFP_KERNEL);
6654	ret = image_id ? 0 : -ENOMEM;
6655	if (!ret)
6656	rbd_dev->image_format = 1;
6657	} else if (ret >= 0) {
6658	void *p = response;
6659
6660	image_id = ceph_extract_encoded_string(p: &p, end: p + ret,
6661	NULL, GFP_NOIO);
6662	ret = PTR_ERR_OR_ZERO(ptr: image_id);
6663	if (!ret)
6664	rbd_dev->image_format = 2;
6665	}
6666
6667	if (!ret) {
6668	rbd_dev->spec->image_id = image_id;
6669	dout("image_id is %s\n", image_id);
6670	}
6671	out:
6672	kfree(objp: response);
6673	ceph_oid_destroy(oid: &oid);
6674	return ret;
6675	}
6676
6677	/*
6678	* Undo whatever state changes are made by v1 or v2 header info
6679	* call.
6680	*/
6681	static void rbd_dev_unprobe(struct rbd_device *rbd_dev)
6682	{
6683	rbd_dev_parent_put(rbd_dev);
6684	rbd_object_map_free(rbd_dev);
6685	rbd_dev_mapping_clear(rbd_dev);
6686
6687	/* Free dynamic fields from the header, then zero it out */
6688
6689	rbd_image_header_cleanup(header: &rbd_dev->header);
6690	}
6691
6692	static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev,
6693	struct rbd_image_header *header)
6694	{
6695	int ret;
6696
6697	ret = rbd_dev_v2_object_prefix(rbd_dev, pobject_prefix: &header->object_prefix);
6698	if (ret)
6699	return ret;
6700
6701	/*
6702	* Get the and check features for the image. Currently the
6703	* features are assumed to never change.
6704	*/
6705	ret = _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
6706	read_only: rbd_is_ro(rbd_dev), snap_features: &header->features);
6707	if (ret)
6708	return ret;
6709
6710	/* If the image supports fancy striping, get its parameters */
6711
6712	if (header->features & RBD_FEATURE_STRIPINGV2) {
6713	ret = rbd_dev_v2_striping_info(rbd_dev, stripe_unit: &header->stripe_unit,
6714	stripe_count: &header->stripe_count);
6715	if (ret)
6716	return ret;
6717	}
6718
6719	if (header->features & RBD_FEATURE_DATA_POOL) {
6720	ret = rbd_dev_v2_data_pool(rbd_dev, data_pool_id: &header->data_pool_id);
6721	if (ret)
6722	return ret;
6723	}
6724
6725	return 0;
6726	}
6727
6728	/*
6729	* @depth is rbd_dev_image_probe() -> rbd_dev_probe_parent() ->
6730	* rbd_dev_image_probe() recursion depth, which means it's also the
6731	* length of the already discovered part of the parent chain.
6732	*/
6733	static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
6734	{
6735	struct rbd_device *parent = NULL;
6736	int ret;
6737
6738	if (!rbd_dev->parent_spec)
6739	return 0;
6740
6741	if (++depth > RBD_MAX_PARENT_CHAIN_LEN) {
6742	pr_info("parent chain is too long (%d)\n", depth);
6743	ret = -EINVAL;
6744	goto out_err;
6745	}
6746
6747	parent = __rbd_dev_create(spec: rbd_dev->parent_spec);
6748	if (!parent) {
6749	ret = -ENOMEM;
6750	goto out_err;
6751	}
6752
6753	/*
6754	* Images related by parent/child relationships always share
6755	* rbd_client and spec/parent_spec, so bump their refcounts.
6756	*/
6757	parent->rbd_client = __rbd_get_client(rbdc: rbd_dev->rbd_client);
6758	parent->spec = rbd_spec_get(spec: rbd_dev->parent_spec);
6759
6760	__set_bit(RBD_DEV_FLAG_READONLY, &parent->flags);
6761
6762	ret = rbd_dev_image_probe(rbd_dev: parent, depth);
6763	if (ret < 0)
6764	goto out_err;
6765
6766	rbd_dev->parent = parent;
6767	atomic_set(v: &rbd_dev->parent_ref, i: 1);
6768	return 0;
6769
6770	out_err:
6771	rbd_dev_unparent(rbd_dev);
6772	rbd_dev_destroy(rbd_dev: parent);
6773	return ret;
6774	}
6775
6776	static void rbd_dev_device_release(struct rbd_device *rbd_dev)
6777	{
6778	clear_bit(nr: RBD_DEV_FLAG_EXISTS, addr: &rbd_dev->flags);
6779	rbd_free_disk(rbd_dev);
6780	if (!single_major)
6781	unregister_blkdev(major: rbd_dev->major, name: rbd_dev->name);
6782	}
6783
6784	/*
6785	* rbd_dev->header_rwsem must be locked for write and will be unlocked
6786	* upon return.
6787	*/
6788	static int rbd_dev_device_setup(struct rbd_device *rbd_dev)
6789	{
6790	int ret;
6791
6792	/* Record our major and minor device numbers. */
6793
6794	if (!single_major) {
6795	ret = register_blkdev(0, rbd_dev->name);
6796	if (ret < 0)
6797	goto err_out_unlock;
6798
6799	rbd_dev->major = ret;
6800	rbd_dev->minor = 0;
6801	} else {
6802	rbd_dev->major = rbd_major;
6803	rbd_dev->minor = rbd_dev_id_to_minor(dev_id: rbd_dev->dev_id);
6804	}
6805
6806	/* Set up the blkdev mapping. */
6807
6808	ret = rbd_init_disk(rbd_dev);
6809	if (ret)
6810	goto err_out_blkdev;
6811
6812	set_capacity(disk: rbd_dev->disk, size: rbd_dev->mapping.size / SECTOR_SIZE);
6813	set_disk_ro(disk: rbd_dev->disk, read_only: rbd_is_ro(rbd_dev));
6814
6815	ret = dev_set_name(dev: &rbd_dev->dev, name: "%d", rbd_dev->dev_id);
6816	if (ret)
6817	goto err_out_disk;
6818
6819	set_bit(nr: RBD_DEV_FLAG_EXISTS, addr: &rbd_dev->flags);
6820	up_write(sem: &rbd_dev->header_rwsem);
6821	return 0;
6822
6823	err_out_disk:
6824	rbd_free_disk(rbd_dev);
6825	err_out_blkdev:
6826	if (!single_major)
6827	unregister_blkdev(major: rbd_dev->major, name: rbd_dev->name);
6828	err_out_unlock:
6829	up_write(sem: &rbd_dev->header_rwsem);
6830	return ret;
6831	}
6832
6833	static int rbd_dev_header_name(struct rbd_device *rbd_dev)
6834	{
6835	struct rbd_spec *spec = rbd_dev->spec;
6836	int ret;
6837
6838	/* Record the header object name for this rbd image. */
6839
6840	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6841	if (rbd_dev->image_format == 1)
6842	ret = ceph_oid_aprintf(oid: &rbd_dev->header_oid, GFP_KERNEL, fmt: "%s%s",
6843	spec->image_name, RBD_SUFFIX);
6844	else
6845	ret = ceph_oid_aprintf(oid: &rbd_dev->header_oid, GFP_KERNEL, fmt: "%s%s",
6846	RBD_HEADER_PREFIX, spec->image_id);
6847
6848	return ret;
6849	}
6850
6851	static void rbd_print_dne(struct rbd_device *rbd_dev, bool is_snap)
6852	{
6853	if (!is_snap) {
6854	pr_info("image %s/%s%s%s does not exist\n",
6855	rbd_dev->spec->pool_name,
6856	rbd_dev->spec->pool_ns ?: "",
6857	rbd_dev->spec->pool_ns ? "/" : "",
6858	rbd_dev->spec->image_name);
6859	} else {
6860	pr_info("snap %s/%s%s%s@%s does not exist\n",
6861	rbd_dev->spec->pool_name,
6862	rbd_dev->spec->pool_ns ?: "",
6863	rbd_dev->spec->pool_ns ? "/" : "",
6864	rbd_dev->spec->image_name,
6865	rbd_dev->spec->snap_name);
6866	}
6867	}
6868
6869	static void rbd_dev_image_release(struct rbd_device *rbd_dev)
6870	{
6871	if (!rbd_is_ro(rbd_dev))
6872	rbd_unregister_watch(rbd_dev);
6873
6874	rbd_dev_unprobe(rbd_dev);
6875	rbd_dev->image_format = 0;
6876	kfree(objp: rbd_dev->spec->image_id);
6877	rbd_dev->spec->image_id = NULL;
6878	}
6879
6880	/*
6881	* Probe for the existence of the header object for the given rbd
6882	* device. If this image is the one being mapped (i.e., not a
6883	* parent), initiate a watch on its header object before using that
6884	* object to get detailed information about the rbd image.
6885	*
6886	* On success, returns with header_rwsem held for write if called
6887	* with @depth == 0.
6888	*/
6889	static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
6890	{
6891	bool need_watch = !rbd_is_ro(rbd_dev);
6892	int ret;
6893
6894	/*
6895	* Get the id from the image id object. Unless there's an
6896	* error, rbd_dev->spec->image_id will be filled in with
6897	* a dynamically-allocated string, and rbd_dev->image_format
6898	* will be set to either 1 or 2.
6899	*/
6900	ret = rbd_dev_image_id(rbd_dev);
6901	if (ret)
6902	return ret;
6903
6904	ret = rbd_dev_header_name(rbd_dev);
6905	if (ret)
6906	goto err_out_format;
6907
6908	if (need_watch) {
6909	ret = rbd_register_watch(rbd_dev);
6910	if (ret) {
6911	if (ret == -ENOENT)
6912	rbd_print_dne(rbd_dev, is_snap: false);
6913	goto err_out_format;
6914	}
6915	}
6916
6917	if (!depth)
6918	down_write(sem: &rbd_dev->header_rwsem);
6919
6920	ret = rbd_dev_header_info(rbd_dev, header: &rbd_dev->header, first_time: true);
6921	if (ret) {
6922	if (ret == -ENOENT && !need_watch)
6923	rbd_print_dne(rbd_dev, is_snap: false);
6924	goto err_out_probe;
6925	}
6926
6927	rbd_init_layout(rbd_dev);
6928
6929	/*
6930	* If this image is the one being mapped, we have pool name and
6931	* id, image name and id, and snap name - need to fill snap id.
6932	* Otherwise this is a parent image, identified by pool, image
6933	* and snap ids - need to fill in names for those ids.
6934	*/
6935	if (!depth)
6936	ret = rbd_spec_fill_snap_id(rbd_dev);
6937	else
6938	ret = rbd_spec_fill_names(rbd_dev);
6939	if (ret) {
6940	if (ret == -ENOENT)
6941	rbd_print_dne(rbd_dev, is_snap: true);
6942	goto err_out_probe;
6943	}
6944
6945	ret = rbd_dev_mapping_set(rbd_dev);
6946	if (ret)
6947	goto err_out_probe;
6948
6949	if (rbd_is_snap(rbd_dev) &&
6950	(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)) {
6951	ret = rbd_object_map_load(rbd_dev);
6952	if (ret)
6953	goto err_out_probe;
6954	}
6955
6956	if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
6957	ret = rbd_dev_setup_parent(rbd_dev);
6958	if (ret)
6959	goto err_out_probe;
6960	}
6961
6962	ret = rbd_dev_probe_parent(rbd_dev, depth);
6963	if (ret)
6964	goto err_out_probe;
6965
6966	dout("discovered format %u image, header name is %s\n",
6967	rbd_dev->image_format, rbd_dev->header_oid.name);
6968	return 0;
6969
6970	err_out_probe:
6971	if (!depth)
6972	up_write(sem: &rbd_dev->header_rwsem);
6973	if (need_watch)
6974	rbd_unregister_watch(rbd_dev);
6975	rbd_dev_unprobe(rbd_dev);
6976	err_out_format:
6977	rbd_dev->image_format = 0;
6978	kfree(objp: rbd_dev->spec->image_id);
6979	rbd_dev->spec->image_id = NULL;
6980	return ret;
6981	}
6982
6983	static void rbd_dev_update_header(struct rbd_device *rbd_dev,
6984	struct rbd_image_header *header)
6985	{
6986	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6987	rbd_assert(rbd_dev->header.object_prefix); /* !first_time */
6988
6989	if (rbd_dev->header.image_size != header->image_size) {
6990	rbd_dev->header.image_size = header->image_size;
6991
6992	if (!rbd_is_snap(rbd_dev)) {
6993	rbd_dev->mapping.size = header->image_size;
6994	rbd_dev_update_size(rbd_dev);
6995	}
6996	}
6997
6998	ceph_put_snap_context(sc: rbd_dev->header.snapc);
6999	rbd_dev->header.snapc = header->snapc;
7000	header->snapc = NULL;
7001
7002	if (rbd_dev->image_format == 1) {
7003	kfree(objp: rbd_dev->header.snap_names);
7004	rbd_dev->header.snap_names = header->snap_names;
7005	header->snap_names = NULL;
7006
7007	kfree(objp: rbd_dev->header.snap_sizes);
7008	rbd_dev->header.snap_sizes = header->snap_sizes;
7009	header->snap_sizes = NULL;
7010	}
7011	}
7012
7013	static void rbd_dev_update_parent(struct rbd_device *rbd_dev,
7014	struct parent_image_info *pii)
7015	{
7016	if (pii->pool_id == CEPH_NOPOOL || !pii->has_overlap) {
7017	/*
7018	* Either the parent never existed, or we have
7019	* record of it but the image got flattened so it no
7020	* longer has a parent. When the parent of a
7021	* layered image disappears we immediately set the
7022	* overlap to 0. The effect of this is that all new
7023	* requests will be treated as if the image had no
7024	* parent.
7025	*
7026	* If !pii.has_overlap, the parent image spec is not
7027	* applicable. It's there to avoid duplication in each
7028	* snapshot record.
7029	*/
7030	if (rbd_dev->parent_overlap) {
7031	rbd_dev->parent_overlap = 0;
7032	rbd_dev_parent_put(rbd_dev);
7033	pr_info("%s: clone has been flattened\n",
7034	rbd_dev->disk->disk_name);
7035	}
7036	} else {
7037	rbd_assert(rbd_dev->parent_spec);
7038
7039	/*
7040	* Update the parent overlap. If it became zero, issue
7041	* a warning as we will proceed as if there is no parent.
7042	*/
7043	if (!pii->overlap && rbd_dev->parent_overlap)
7044	rbd_warn(rbd_dev,
7045	fmt: "clone has become standalone (overlap 0)");
7046	rbd_dev->parent_overlap = pii->overlap;
7047	}
7048	}
7049
7050	static int rbd_dev_refresh(struct rbd_device *rbd_dev)
7051	{
7052	struct rbd_image_header header = { 0 };
7053	struct parent_image_info pii = { 0 };
7054	int ret;
7055
7056	dout("%s rbd_dev %p\n", __func__, rbd_dev);
7057
7058	ret = rbd_dev_header_info(rbd_dev, header: &header, first_time: false);
7059	if (ret)
7060	goto out;
7061
7062	/*
7063	* If there is a parent, see if it has disappeared due to the
7064	* mapped image getting flattened.
7065	*/
7066	if (rbd_dev->parent) {
7067	ret = rbd_dev_v2_parent_info(rbd_dev, pii: &pii);
7068	if (ret)
7069	goto out;
7070	}
7071
7072	down_write(sem: &rbd_dev->header_rwsem);
7073	rbd_dev_update_header(rbd_dev, header: &header);
7074	if (rbd_dev->parent)
7075	rbd_dev_update_parent(rbd_dev, pii: &pii);
7076	up_write(sem: &rbd_dev->header_rwsem);
7077
7078	out:
7079	rbd_parent_info_cleanup(pii: &pii);
7080	rbd_image_header_cleanup(header: &header);
7081	return ret;
7082	}
7083
7084	static ssize_t do_rbd_add(const char *buf, size_t count)
7085	{
7086	struct rbd_device *rbd_dev = NULL;
7087	struct ceph_options *ceph_opts = NULL;
7088	struct rbd_options *rbd_opts = NULL;
7089	struct rbd_spec *spec = NULL;
7090	struct rbd_client *rbdc;
7091	int rc;
7092
7093	if (!capable(CAP_SYS_ADMIN))
7094	return -EPERM;
7095
7096	if (!try_module_get(THIS_MODULE))
7097	return -ENODEV;
7098
7099	/* parse add command */
7100	rc = rbd_add_parse_args(buf, ceph_opts: &ceph_opts, opts: &rbd_opts, rbd_spec: &spec);
7101	if (rc < 0)
7102	goto out;
7103
7104	rbdc = rbd_get_client(ceph_opts);
7105	if (IS_ERR(ptr: rbdc)) {
7106	rc = PTR_ERR(ptr: rbdc);
7107	goto err_out_args;
7108	}
7109
7110	/* pick the pool */
7111	rc = ceph_pg_poolid_by_name(map: rbdc->client->osdc.osdmap, name: spec->pool_name);
7112	if (rc < 0) {
7113	if (rc == -ENOENT)
7114	pr_info("pool %s does not exist\n", spec->pool_name);
7115	goto err_out_client;
7116	}
7117	spec->pool_id = (u64)rc;
7118
7119	rbd_dev = rbd_dev_create(rbdc, spec, opts: rbd_opts);
7120	if (!rbd_dev) {
7121	rc = -ENOMEM;
7122	goto err_out_client;
7123	}
7124	rbdc = NULL; /* rbd_dev now owns this */
7125	spec = NULL; /* rbd_dev now owns this */
7126	rbd_opts = NULL; /* rbd_dev now owns this */
7127
7128	/* if we are mapping a snapshot it will be a read-only mapping */
7129	if (rbd_dev->opts->read_only ||
7130	strcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME))
7131	__set_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
7132
7133	rbd_dev->config_info = kstrdup(s: buf, GFP_KERNEL);
7134	if (!rbd_dev->config_info) {
7135	rc = -ENOMEM;
7136	goto err_out_rbd_dev;
7137	}
7138
7139	rc = rbd_dev_image_probe(rbd_dev, depth: 0);
7140	if (rc < 0)
7141	goto err_out_rbd_dev;
7142
7143	if (rbd_dev->opts->alloc_size > rbd_dev->layout.object_size) {
7144	rbd_warn(rbd_dev, fmt: "alloc_size adjusted to %u",
7145	rbd_dev->layout.object_size);
7146	rbd_dev->opts->alloc_size = rbd_dev->layout.object_size;
7147	}
7148
7149	rc = rbd_dev_device_setup(rbd_dev);
7150	if (rc)
7151	goto err_out_image_probe;
7152
7153	rc = rbd_add_acquire_lock(rbd_dev);
7154	if (rc)
7155	goto err_out_image_lock;
7156
7157	/* Everything's ready. Announce the disk to the world. */
7158
7159	rc = device_add(dev: &rbd_dev->dev);
7160	if (rc)
7161	goto err_out_image_lock;
7162
7163	rc = device_add_disk(parent: &rbd_dev->dev, disk: rbd_dev->disk, NULL);
7164	if (rc)
7165	goto err_out_cleanup_disk;
7166
7167	spin_lock(lock: &rbd_dev_list_lock);
7168	list_add_tail(new: &rbd_dev->node, head: &rbd_dev_list);
7169	spin_unlock(lock: &rbd_dev_list_lock);
7170
7171	pr_info("%s: capacity %llu features 0x%llx\n", rbd_dev->disk->disk_name,
7172	(unsigned long long)get_capacity(rbd_dev->disk) << SECTOR_SHIFT,
7173	rbd_dev->header.features);
7174	rc = count;
7175	out:
7176	module_put(THIS_MODULE);
7177	return rc;
7178
7179	err_out_cleanup_disk:
7180	rbd_free_disk(rbd_dev);
7181	err_out_image_lock:
7182	rbd_dev_image_unlock(rbd_dev);
7183	rbd_dev_device_release(rbd_dev);
7184	err_out_image_probe:
7185	rbd_dev_image_release(rbd_dev);
7186	err_out_rbd_dev:
7187	rbd_dev_destroy(rbd_dev);
7188	err_out_client:
7189	rbd_put_client(rbdc);
7190	err_out_args:
7191	rbd_spec_put(spec);
7192	kfree(objp: rbd_opts);
7193	goto out;
7194	}
7195
7196	static ssize_t add_store(const struct bus_type *bus, const char *buf, size_t count)
7197	{
7198	if (single_major)
7199	return -EINVAL;
7200
7201	return do_rbd_add(buf, count);
7202	}
7203
7204	static ssize_t add_single_major_store(const struct bus_type *bus, const char *buf,
7205	size_t count)
7206	{
7207	return do_rbd_add(buf, count);
7208	}
7209
7210	static void rbd_dev_remove_parent(struct rbd_device *rbd_dev)
7211	{
7212	while (rbd_dev->parent) {
7213	struct rbd_device *first = rbd_dev;
7214	struct rbd_device *second = first->parent;
7215	struct rbd_device *third;
7216
7217	/*
7218	* Follow to the parent with no grandparent and
7219	* remove it.
7220	*/
7221	while (second && (third = second->parent)) {
7222	first = second;
7223	second = third;
7224	}
7225	rbd_assert(second);
7226	rbd_dev_image_release(rbd_dev: second);
7227	rbd_dev_destroy(rbd_dev: second);
7228	first->parent = NULL;
7229	first->parent_overlap = 0;
7230
7231	rbd_assert(first->parent_spec);
7232	rbd_spec_put(spec: first->parent_spec);
7233	first->parent_spec = NULL;
7234	}
7235	}
7236
7237	static ssize_t do_rbd_remove(const char *buf, size_t count)
7238	{
7239	struct rbd_device *rbd_dev = NULL;
7240	int dev_id;
7241	char opt_buf[6];
7242	bool force = false;
7243	int ret;
7244
7245	if (!capable(CAP_SYS_ADMIN))
7246	return -EPERM;
7247
7248	dev_id = -1;
7249	opt_buf[0] = '\0';
7250	sscanf(buf, "%d %5s", &dev_id, opt_buf);
7251	if (dev_id < 0) {
7252	pr_err("dev_id out of range\n");
7253	return -EINVAL;
7254	}
7255	if (opt_buf[0] != '\0') {
7256	if (!strcmp(opt_buf, "force")) {
7257	force = true;
7258	} else {
7259	pr_err("bad remove option at '%s'\n", opt_buf);
7260	return -EINVAL;
7261	}
7262	}
7263
7264	ret = -ENOENT;
7265	spin_lock(lock: &rbd_dev_list_lock);
7266	list_for_each_entry(rbd_dev, &rbd_dev_list, node) {
7267	if (rbd_dev->dev_id == dev_id) {
7268	ret = 0;
7269	break;
7270	}
7271	}
7272	if (!ret) {
7273	spin_lock_irq(lock: &rbd_dev->lock);
7274	if (rbd_dev->open_count && !force)
7275	ret = -EBUSY;
7276	else if (test_and_set_bit(nr: RBD_DEV_FLAG_REMOVING,
7277	addr: &rbd_dev->flags))
7278	ret = -EINPROGRESS;
7279	spin_unlock_irq(lock: &rbd_dev->lock);
7280	}
7281	spin_unlock(lock: &rbd_dev_list_lock);
7282	if (ret)
7283	return ret;
7284
7285	if (force) {
7286	/*
7287	* Prevent new IO from being queued and wait for existing
7288	* IO to complete/fail.
7289	*/
7290	blk_mq_freeze_queue(q: rbd_dev->disk->queue);
7291	blk_mark_disk_dead(disk: rbd_dev->disk);
7292	}
7293
7294	del_gendisk(gp: rbd_dev->disk);
7295	spin_lock(lock: &rbd_dev_list_lock);
7296	list_del_init(entry: &rbd_dev->node);
7297	spin_unlock(lock: &rbd_dev_list_lock);
7298	device_del(dev: &rbd_dev->dev);
7299
7300	rbd_dev_image_unlock(rbd_dev);
7301	rbd_dev_device_release(rbd_dev);
7302	rbd_dev_image_release(rbd_dev);
7303	rbd_dev_destroy(rbd_dev);
7304	return count;
7305	}
7306
7307	static ssize_t remove_store(const struct bus_type *bus, const char *buf, size_t count)
7308	{
7309	if (single_major)
7310	return -EINVAL;
7311
7312	return do_rbd_remove(buf, count);
7313	}
7314
7315	static ssize_t remove_single_major_store(const struct bus_type *bus, const char *buf,
7316	size_t count)
7317	{
7318	return do_rbd_remove(buf, count);
7319	}
7320
7321	/*
7322	* create control files in sysfs
7323	* /sys/bus/rbd/...
7324	*/
7325	static int __init rbd_sysfs_init(void)
7326	{
7327	int ret;
7328
7329	ret = device_register(dev: &rbd_root_dev);
7330	if (ret < 0) {
7331	put_device(dev: &rbd_root_dev);
7332	return ret;
7333	}
7334
7335	ret = bus_register(bus: &rbd_bus_type);
7336	if (ret < 0)
7337	device_unregister(dev: &rbd_root_dev);
7338
7339	return ret;
7340	}
7341
7342	static void __exit rbd_sysfs_cleanup(void)
7343	{
7344	bus_unregister(bus: &rbd_bus_type);
7345	device_unregister(dev: &rbd_root_dev);
7346	}
7347
7348	static int __init rbd_slab_init(void)
7349	{
7350	rbd_assert(!rbd_img_request_cache);
7351	rbd_img_request_cache = KMEM_CACHE(rbd_img_request, 0);
7352	if (!rbd_img_request_cache)
7353	return -ENOMEM;
7354
7355	rbd_assert(!rbd_obj_request_cache);
7356	rbd_obj_request_cache = KMEM_CACHE(rbd_obj_request, 0);
7357	if (!rbd_obj_request_cache)
7358	goto out_err;
7359
7360	return 0;
7361
7362	out_err:
7363	kmem_cache_destroy(s: rbd_img_request_cache);
7364	rbd_img_request_cache = NULL;
7365	return -ENOMEM;
7366	}
7367
7368	static void rbd_slab_exit(void)
7369	{
7370	rbd_assert(rbd_obj_request_cache);
7371	kmem_cache_destroy(s: rbd_obj_request_cache);
7372	rbd_obj_request_cache = NULL;
7373
7374	rbd_assert(rbd_img_request_cache);
7375	kmem_cache_destroy(s: rbd_img_request_cache);
7376	rbd_img_request_cache = NULL;
7377	}
7378
7379	static int __init rbd_init(void)
7380	{
7381	int rc;
7382
7383	if (!libceph_compatible(NULL)) {
7384	rbd_warn(NULL, fmt: "libceph incompatibility (quitting)");
7385	return -EINVAL;
7386	}
7387
7388	rc = rbd_slab_init();
7389	if (rc)
7390	return rc;
7391
7392	/*
7393	* The number of active work items is limited by the number of
7394	* rbd devices * queue depth, so leave @max_active at default.
7395	*/
7396	rbd_wq = alloc_workqueue(RBD_DRV_NAME, flags: WQ_MEM_RECLAIM, max_active: 0);
7397	if (!rbd_wq) {
7398	rc = -ENOMEM;
7399	goto err_out_slab;
7400	}
7401
7402	if (single_major) {
7403	rbd_major = register_blkdev(0, RBD_DRV_NAME);
7404	if (rbd_major < 0) {
7405	rc = rbd_major;
7406	goto err_out_wq;
7407	}
7408	}
7409
7410	rc = rbd_sysfs_init();
7411	if (rc)
7412	goto err_out_blkdev;
7413
7414	if (single_major)
7415	pr_info("loaded (major %d)\n", rbd_major);
7416	else
7417	pr_info("loaded\n");
7418
7419	return 0;
7420
7421	err_out_blkdev:
7422	if (single_major)
7423	unregister_blkdev(major: rbd_major, RBD_DRV_NAME);
7424	err_out_wq:
7425	destroy_workqueue(wq: rbd_wq);
7426	err_out_slab:
7427	rbd_slab_exit();
7428	return rc;
7429	}
7430
7431	static void __exit rbd_exit(void)
7432	{
7433	ida_destroy(ida: &rbd_dev_id_ida);
7434	rbd_sysfs_cleanup();
7435	if (single_major)
7436	unregister_blkdev(major: rbd_major, RBD_DRV_NAME);
7437	destroy_workqueue(wq: rbd_wq);
7438	rbd_slab_exit();
7439	}
7440
7441	module_init(rbd_init);
7442	module_exit(rbd_exit);
7443
7444	MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
7445	MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
7446	MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
7447	/* following authorship retained from original osdblk.c */
7448	MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
7449
7450	MODULE_DESCRIPTION("RADOS Block Device (RBD) driver");
7451	MODULE_LICENSE("GPL");
7452