linux - drivers/block/rbd.c

7445 lines · 186.3 KiB

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