~hc/RK356X_SDK_RELEASE.git

..	..	@@ -9,6 +9,8 @@
9	9	#include <linux/debugfs.h>
10	10	#include <linux/seq_file.h>
11	11	#include <linux/ratelimit.h>
	12	+#include <linux/bits.h>
	13	+#include <linux/ktime.h>
12	14
13	15	#include "super.h"
14	16	#include "mds_client.h"
..	..	@@ -19,6 +21,8 @@
19	21	#include <linux/ceph/pagelist.h>
20	22	#include <linux/ceph/auth.h>
21	23	#include <linux/ceph/debugfs.h>
	24	+
	25	+#define RECONNECT_MAX_SIZE (INT_MAX - PAGE_SIZE)
22	26
23	27	/*
24	28	* A cluster of MDS (metadata server) daemons is responsible for
..	..	@@ -46,13 +50,17 @@
46	50	*/
47	51
48	52	struct ceph_reconnect_state {
49		- int nr_caps;
	53	+ struct ceph_mds_session *session;
	54	+ int nr_caps, nr_realms;
50	55	struct ceph_pagelist *pagelist;
51	56	unsigned msg_version;
	57	+ bool allow_multi;
52	58	};
53	59
54	60	static void __wake_requests(struct ceph_mds_client *mdsc,
55	61	struct list_head *head);
	62	+static void ceph_cap_release_work(struct work_struct *work);
	63	+static void ceph_cap_reclaim_work(struct work_struct *work);
56	64
57	65	static const struct ceph_connection_operations mds_con_ops;
58	66
..	..	@@ -61,6 +69,29 @@
61	69	* mds reply parsing
62	70	*/
63	71
	72	+static int parse_reply_info_quota(void *p, void end,
	73	+ struct ceph_mds_reply_info_in *info)
	74	+{
	75	+ u8 struct_v, struct_compat;
	76	+ u32 struct_len;
	77	+
	78	+ ceph_decode_8_safe(p, end, struct_v, bad);
	79	+ ceph_decode_8_safe(p, end, struct_compat, bad);
	80	+ /* struct_v is expected to be >= 1. we only
	81	+ * understand encoding with struct_compat == 1. */
	82	+ if (!struct_v \|\| struct_compat != 1)
	83	+ goto bad;
	84	+ ceph_decode_32_safe(p, end, struct_len, bad);
	85	+ ceph_decode_need(p, end, struct_len, bad);
	86	+ end = *p + struct_len;
	87	+ ceph_decode_64_safe(p, end, info->max_bytes, bad);
	88	+ ceph_decode_64_safe(p, end, info->max_files, bad);
	89	+ *p = end;
	90	+ return 0;
	91	+bad:
	92	+ return -EIO;
	93	+}
	94	+
64	95	/*
65	96	* parse individual inode info
66	97	*/
..	..	@@ -68,8 +99,24 @@
68	99	struct ceph_mds_reply_info_in *info,
69	100	u64 features)
70	101	{
71		- int err = -EIO;
	102	+ int err = 0;
	103	+ u8 struct_v = 0;
72	104
	105	+ if (features == (u64)-1) {
	106	+ u32 struct_len;
	107	+ u8 struct_compat;
	108	+ ceph_decode_8_safe(p, end, struct_v, bad);
	109	+ ceph_decode_8_safe(p, end, struct_compat, bad);
	110	+ /* struct_v is expected to be >= 1. we only understand
	111	+ * encoding with struct_compat == 1. */
	112	+ if (!struct_v \|\| struct_compat != 1)
	113	+ goto bad;
	114	+ ceph_decode_32_safe(p, end, struct_len, bad);
	115	+ ceph_decode_need(p, end, struct_len, bad);
	116	+ end = *p + struct_len;
	117	+ }
	118	+
	119	+ ceph_decode_need(p, end, sizeof(struct ceph_mds_reply_inode), bad);
73	120	info->in = *p;
74	121	*p += sizeof(struct ceph_mds_reply_inode) +
75	122	sizeof(info->in->fragtree.splits)
..	..	@@ -80,60 +127,158 @@
80	127	info->symlink = *p;
81	128	*p += info->symlink_len;
82	129
83		- if (features & CEPH_FEATURE_DIRLAYOUTHASH)
84		- ceph_decode_copy_safe(p, end, &info->dir_layout,
85		- sizeof(info->dir_layout), bad);
86		- else
87		- memset(&info->dir_layout, 0, sizeof(info->dir_layout));
88		-
	130	+ ceph_decode_copy_safe(p, end, &info->dir_layout,
	131	+ sizeof(info->dir_layout), bad);
89	132	ceph_decode_32_safe(p, end, info->xattr_len, bad);
90	133	ceph_decode_need(p, end, info->xattr_len, bad);
91	134	info->xattr_data = *p;
92	135	*p += info->xattr_len;
93	136
94		- if (features & CEPH_FEATURE_MDS_INLINE_DATA) {
	137	+ if (features == (u64)-1) {
	138	+ /* inline data */
95	139	ceph_decode_64_safe(p, end, info->inline_version, bad);
96	140	ceph_decode_32_safe(p, end, info->inline_len, bad);
97	141	ceph_decode_need(p, end, info->inline_len, bad);
98	142	info->inline_data = *p;
99	143	*p += info->inline_len;
100		- } else
101		- info->inline_version = CEPH_INLINE_NONE;
102		-
103		- if (features & CEPH_FEATURE_MDS_QUOTA) {
104		- u8 struct_v, struct_compat;
105		- u32 struct_len;
106		-
107		- /*
108		- * both struct_v and struct_compat are expected to be >= 1
109		- */
110		- ceph_decode_8_safe(p, end, struct_v, bad);
111		- ceph_decode_8_safe(p, end, struct_compat, bad);
112		- if (!struct_v \|\| !struct_compat)
113		- goto bad;
114		- ceph_decode_32_safe(p, end, struct_len, bad);
115		- ceph_decode_need(p, end, struct_len, bad);
116		- ceph_decode_64_safe(p, end, info->max_bytes, bad);
117		- ceph_decode_64_safe(p, end, info->max_files, bad);
118		- } else {
119		- info->max_bytes = 0;
120		- info->max_files = 0;
121		- }
122		-
123		- info->pool_ns_len = 0;
124		- info->pool_ns_data = NULL;
125		- if (features & CEPH_FEATURE_FS_FILE_LAYOUT_V2) {
	144	+ /* quota */
	145	+ err = parse_reply_info_quota(p, end, info);
	146	+ if (err < 0)
	147	+ goto out_bad;
	148	+ /* pool namespace */
126	149	ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
127	150	if (info->pool_ns_len > 0) {
128	151	ceph_decode_need(p, end, info->pool_ns_len, bad);
129	152	info->pool_ns_data = *p;
130	153	*p += info->pool_ns_len;
131	154	}
132		- }
133	155
	156	+ /* btime */
	157	+ ceph_decode_need(p, end, sizeof(info->btime), bad);
	158	+ ceph_decode_copy(p, &info->btime, sizeof(info->btime));
	159	+
	160	+ /* change attribute */
	161	+ ceph_decode_64_safe(p, end, info->change_attr, bad);
	162	+
	163	+ /* dir pin */
	164	+ if (struct_v >= 2) {
	165	+ ceph_decode_32_safe(p, end, info->dir_pin, bad);
	166	+ } else {
	167	+ info->dir_pin = -ENODATA;
	168	+ }
	169	+
	170	+ /* snapshot birth time, remains zero for v<=2 */
	171	+ if (struct_v >= 3) {
	172	+ ceph_decode_need(p, end, sizeof(info->snap_btime), bad);
	173	+ ceph_decode_copy(p, &info->snap_btime,
	174	+ sizeof(info->snap_btime));
	175	+ } else {
	176	+ memset(&info->snap_btime, 0, sizeof(info->snap_btime));
	177	+ }
	178	+
	179	+ *p = end;
	180	+ } else {
	181	+ if (features & CEPH_FEATURE_MDS_INLINE_DATA) {
	182	+ ceph_decode_64_safe(p, end, info->inline_version, bad);
	183	+ ceph_decode_32_safe(p, end, info->inline_len, bad);
	184	+ ceph_decode_need(p, end, info->inline_len, bad);
	185	+ info->inline_data = *p;
	186	+ *p += info->inline_len;
	187	+ } else
	188	+ info->inline_version = CEPH_INLINE_NONE;
	189	+
	190	+ if (features & CEPH_FEATURE_MDS_QUOTA) {
	191	+ err = parse_reply_info_quota(p, end, info);
	192	+ if (err < 0)
	193	+ goto out_bad;
	194	+ } else {
	195	+ info->max_bytes = 0;
	196	+ info->max_files = 0;
	197	+ }
	198	+
	199	+ info->pool_ns_len = 0;
	200	+ info->pool_ns_data = NULL;
	201	+ if (features & CEPH_FEATURE_FS_FILE_LAYOUT_V2) {
	202	+ ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
	203	+ if (info->pool_ns_len > 0) {
	204	+ ceph_decode_need(p, end, info->pool_ns_len, bad);
	205	+ info->pool_ns_data = *p;
	206	+ *p += info->pool_ns_len;
	207	+ }
	208	+ }
	209	+
	210	+ if (features & CEPH_FEATURE_FS_BTIME) {
	211	+ ceph_decode_need(p, end, sizeof(info->btime), bad);
	212	+ ceph_decode_copy(p, &info->btime, sizeof(info->btime));
	213	+ ceph_decode_64_safe(p, end, info->change_attr, bad);
	214	+ }
	215	+
	216	+ info->dir_pin = -ENODATA;
	217	+ /* info->snap_btime remains zero */
	218	+ }
134	219	return 0;
135	220	bad:
	221	+ err = -EIO;
	222	+out_bad:
136	223	return err;
	224	+}
	225	+
	226	+static int parse_reply_info_dir(void *p, void end,
	227	+ struct ceph_mds_reply_dirfrag **dirfrag,
	228	+ u64 features)
	229	+{
	230	+ if (features == (u64)-1) {
	231	+ u8 struct_v, struct_compat;
	232	+ u32 struct_len;
	233	+ ceph_decode_8_safe(p, end, struct_v, bad);
	234	+ ceph_decode_8_safe(p, end, struct_compat, bad);
	235	+ /* struct_v is expected to be >= 1. we only understand
	236	+ * encoding whose struct_compat == 1. */
	237	+ if (!struct_v \|\| struct_compat != 1)
	238	+ goto bad;
	239	+ ceph_decode_32_safe(p, end, struct_len, bad);
	240	+ ceph_decode_need(p, end, struct_len, bad);
	241	+ end = *p + struct_len;
	242	+ }
	243	+
	244	+ ceph_decode_need(p, end, sizeof(**dirfrag), bad);
	245	+ dirfrag = p;
	246	+ p += sizeof(dirfrag) + sizeof(u32) le32_to_cpu((*dirfrag)->ndist);
	247	+ if (unlikely(*p > end))
	248	+ goto bad;
	249	+ if (features == (u64)-1)
	250	+ *p = end;
	251	+ return 0;
	252	+bad:
	253	+ return -EIO;
	254	+}
	255	+
	256	+static int parse_reply_info_lease(void *p, void end,
	257	+ struct ceph_mds_reply_lease **lease,
	258	+ u64 features)
	259	+{
	260	+ if (features == (u64)-1) {
	261	+ u8 struct_v, struct_compat;
	262	+ u32 struct_len;
	263	+ ceph_decode_8_safe(p, end, struct_v, bad);
	264	+ ceph_decode_8_safe(p, end, struct_compat, bad);
	265	+ /* struct_v is expected to be >= 1. we only understand
	266	+ * encoding whose struct_compat == 1. */
	267	+ if (!struct_v \|\| struct_compat != 1)
	268	+ goto bad;
	269	+ ceph_decode_32_safe(p, end, struct_len, bad);
	270	+ ceph_decode_need(p, end, struct_len, bad);
	271	+ end = *p + struct_len;
	272	+ }
	273	+
	274	+ ceph_decode_need(p, end, sizeof(**lease), bad);
	275	+ lease = p;
	276	+ p += sizeof(*lease);
	277	+ if (features == (u64)-1)
	278	+ *p = end;
	279	+ return 0;
	280	+bad:
	281	+ return -EIO;
137	282	}
138	283
139	284	/*
..	..	@@ -151,20 +296,18 @@
151	296	if (err < 0)
152	297	goto out_bad;
153	298
154		- if (unlikely(p + sizeof(info->dirfrag) > end))
155		- goto bad;
156		- info->dirfrag = *p;
157		- p += sizeof(info->dirfrag) +
158		- sizeof(u32)*le32_to_cpu(info->dirfrag->ndist);
159		- if (unlikely(*p > end))
160		- goto bad;
	299	+ err = parse_reply_info_dir(p, end, &info->dirfrag, features);
	300	+ if (err < 0)
	301	+ goto out_bad;
161	302
162	303	ceph_decode_32_safe(p, end, info->dname_len, bad);
163	304	ceph_decode_need(p, end, info->dname_len, bad);
164	305	info->dname = *p;
165	306	*p += info->dname_len;
166		- info->dlease = *p;
167		- p += sizeof(info->dlease);
	307	+
	308	+ err = parse_reply_info_lease(p, end, &info->dlease, features);
	309	+ if (err < 0)
	310	+ goto out_bad;
168	311	}
169	312
170	313	if (info->head->is_target) {
..	..	@@ -187,20 +330,16 @@
187	330	/*
188	331	* parse readdir results
189	332	*/
190		-static int parse_reply_info_dir(void *p, void end,
	333	+static int parse_reply_info_readdir(void *p, void end,
191	334	struct ceph_mds_reply_info_parsed *info,
192	335	u64 features)
193	336	{
194	337	u32 num, i = 0;
195	338	int err;
196	339
197		- info->dir_dir = *p;
198		- if (p + sizeof(info->dir_dir) > end)
199		- goto bad;
200		- p += sizeof(info->dir_dir) +
201		- sizeof(u32)*le32_to_cpu(info->dir_dir->ndist);
202		- if (*p > end)
203		- goto bad;
	340	+ err = parse_reply_info_dir(p, end, &info->dir_dir, features);
	341	+ if (err < 0)
	342	+ goto out_bad;
204	343
205	344	ceph_decode_need(p, end, sizeof(num) + 2, bad);
206	345	num = ceph_decode_32(p);
..	..	@@ -226,15 +365,16 @@
226	365	while (num) {
227	366	struct ceph_mds_reply_dir_entry *rde = info->dir_entries + i;
228	367	/* dentry */
229		- ceph_decode_need(p, end, sizeof(u32)*2, bad);
230		- rde->name_len = ceph_decode_32(p);
	368	+ ceph_decode_32_safe(p, end, rde->name_len, bad);
231	369	ceph_decode_need(p, end, rde->name_len, bad);
232	370	rde->name = *p;
233	371	*p += rde->name_len;
234	372	dout("parsed dir dname '%.*s'\n", rde->name_len, rde->name);
235		- rde->lease = *p;
236		- *p += sizeof(struct ceph_mds_reply_lease);
237	373
	374	+ /* dentry lease */
	375	+ err = parse_reply_info_lease(p, end, &rde->lease, features);
	376	+ if (err)
	377	+ goto out_bad;
238	378	/* inode */
239	379	err = parse_reply_info_in(p, end, &rde->inode, features);
240	380	if (err < 0)
..	..	@@ -246,8 +386,8 @@
246	386	}
247	387
248	388	done:
249		- if (*p != end)
250		- goto bad;
	389	+ /* Skip over any unrecognized fields */
	390	+ *p = end;
251	391	return 0;
252	392
253	393	bad:
..	..	@@ -268,36 +408,145 @@
268	408	goto bad;
269	409
270	410	info->filelock_reply = *p;
271		- p += sizeof(info->filelock_reply);
272	411
273		- if (unlikely(*p != end))
274		- goto bad;
	412	+ /* Skip over any unrecognized fields */
	413	+ *p = end;
275	414	return 0;
276		-
277	415	bad:
278	416	return -EIO;
279	417	}
	418	+
	419	+
	420	+#if BITS_PER_LONG == 64
	421	+
	422	+#define DELEGATED_INO_AVAILABLE xa_mk_value(1)
	423	+
	424	+static int ceph_parse_deleg_inos(void *p, void end,
	425	+ struct ceph_mds_session *s)
	426	+{
	427	+ u32 sets;
	428	+
	429	+ ceph_decode_32_safe(p, end, sets, bad);
	430	+ dout("got %u sets of delegated inodes\n", sets);
	431	+ while (sets--) {
	432	+ u64 start, len, ino;
	433	+
	434	+ ceph_decode_64_safe(p, end, start, bad);
	435	+ ceph_decode_64_safe(p, end, len, bad);
	436	+
	437	+ /* Don't accept a delegation of system inodes */
	438	+ if (start < CEPH_INO_SYSTEM_BASE) {
	439	+ pr_warn_ratelimited("ceph: ignoring reserved inode range delegation (start=0x%llx len=0x%llx)\n",
	440	+ start, len);
	441	+ continue;
	442	+ }
	443	+ while (len--) {
	444	+ int err = xa_insert(&s->s_delegated_inos, ino = start++,
	445	+ DELEGATED_INO_AVAILABLE,
	446	+ GFP_KERNEL);
	447	+ if (!err) {
	448	+ dout("added delegated inode 0x%llx\n",
	449	+ start - 1);
	450	+ } else if (err == -EBUSY) {
	451	+ pr_warn("ceph: MDS delegated inode 0x%llx more than once.\n",
	452	+ start - 1);
	453	+ } else {
	454	+ return err;
	455	+ }
	456	+ }
	457	+ }
	458	+ return 0;
	459	+bad:
	460	+ return -EIO;
	461	+}
	462	+
	463	+u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
	464	+{
	465	+ unsigned long ino;
	466	+ void *val;
	467	+
	468	+ xa_for_each(&s->s_delegated_inos, ino, val) {
	469	+ val = xa_erase(&s->s_delegated_inos, ino);
	470	+ if (val == DELEGATED_INO_AVAILABLE)
	471	+ return ino;
	472	+ }
	473	+ return 0;
	474	+}
	475	+
	476	+int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
	477	+{
	478	+ return xa_insert(&s->s_delegated_inos, ino, DELEGATED_INO_AVAILABLE,
	479	+ GFP_KERNEL);
	480	+}
	481	+#else /* BITS_PER_LONG == 64 */
	482	+/*
	483	+ * FIXME: xarrays can't handle 64-bit indexes on a 32-bit arch. For now, just
	484	+ * ignore delegated_inos on 32 bit arch. Maybe eventually add xarrays for top
	485	+ * and bottom words?
	486	+ */
	487	+static int ceph_parse_deleg_inos(void *p, void end,
	488	+ struct ceph_mds_session *s)
	489	+{
	490	+ u32 sets;
	491	+
	492	+ ceph_decode_32_safe(p, end, sets, bad);
	493	+ if (sets)
	494	+ ceph_decode_skip_n(p, end, sets * 2 * sizeof(__le64), bad);
	495	+ return 0;
	496	+bad:
	497	+ return -EIO;
	498	+}
	499	+
	500	+u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
	501	+{
	502	+ return 0;
	503	+}
	504	+
	505	+int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
	506	+{
	507	+ return 0;
	508	+}
	509	+#endif /* BITS_PER_LONG == 64 */
280	510
281	511	/*
282	512	* parse create results
283	513	*/
284	514	static int parse_reply_info_create(void *p, void end,
285	515	struct ceph_mds_reply_info_parsed *info,
286		- u64 features)
	516	+ u64 features, struct ceph_mds_session *s)
287	517	{
288		- if (features & CEPH_FEATURE_REPLY_CREATE_INODE) {
	518	+ int ret;
	519	+
	520	+ if (features == (u64)-1 \|\|
	521	+ (features & CEPH_FEATURE_REPLY_CREATE_INODE)) {
289	522	if (*p == end) {
	523	+ /* Malformed reply? */
290	524	info->has_create_ino = false;
291		- } else {
	525	+ } else if (test_bit(CEPHFS_FEATURE_DELEG_INO, &s->s_features)) {
	526	+ u8 struct_v, struct_compat;
	527	+ u32 len;
	528	+
292	529	info->has_create_ino = true;
293		- info->ino = ceph_decode_64(p);
	530	+ ceph_decode_8_safe(p, end, struct_v, bad);
	531	+ ceph_decode_8_safe(p, end, struct_compat, bad);
	532	+ ceph_decode_32_safe(p, end, len, bad);
	533	+ ceph_decode_64_safe(p, end, info->ino, bad);
	534	+ ret = ceph_parse_deleg_inos(p, end, s);
	535	+ if (ret)
	536	+ return ret;
	537	+ } else {
	538	+ /* legacy */
	539	+ ceph_decode_64_safe(p, end, info->ino, bad);
	540	+ info->has_create_ino = true;
294	541	}
	542	+ } else {
	543	+ if (*p != end)
	544	+ goto bad;
295	545	}
296	546
297		- if (unlikely(*p != end))
298		- goto bad;
	547	+ /* Skip over any unrecognized fields */
	548	+ *p = end;
299	549	return 0;
300		-
301	550	bad:
302	551	return -EIO;
303	552	}
..	..	@@ -307,16 +556,16 @@
307	556	*/
308	557	static int parse_reply_info_extra(void *p, void end,
309	558	struct ceph_mds_reply_info_parsed *info,
310		- u64 features)
	559	+ u64 features, struct ceph_mds_session *s)
311	560	{
312	561	u32 op = le32_to_cpu(info->head->op);
313	562
314	563	if (op == CEPH_MDS_OP_GETFILELOCK)
315	564	return parse_reply_info_filelock(p, end, info, features);
316	565	else if (op == CEPH_MDS_OP_READDIR \|\| op == CEPH_MDS_OP_LSSNAP)
317		- return parse_reply_info_dir(p, end, info, features);
	566	+ return parse_reply_info_readdir(p, end, info, features);
318	567	else if (op == CEPH_MDS_OP_CREATE)
319		- return parse_reply_info_create(p, end, info, features);
	568	+ return parse_reply_info_create(p, end, info, features, s);
320	569	else
321	570	return -EIO;
322	571	}
..	..	@@ -324,7 +573,7 @@
324	573	/*
325	574	* parse entire mds reply
326	575	*/
327		-static int parse_reply_info(struct ceph_msg *msg,
	576	+static int parse_reply_info(struct ceph_mds_session s, struct ceph_msg msg,
328	577	struct ceph_mds_reply_info_parsed *info,
329	578	u64 features)
330	579	{
..	..	@@ -349,7 +598,7 @@
349	598	ceph_decode_32_safe(&p, end, len, bad);
350	599	if (len > 0) {
351	600	ceph_decode_need(&p, end, len, bad);
352		- err = parse_reply_info_extra(&p, p+len, info, features);
	601	+ err = parse_reply_info_extra(&p, p+len, info, features, s);
353	602	if (err < 0)
354	603	goto out_bad;
355	604	}
..	..	@@ -390,6 +639,7 @@
390	639	case CEPH_MDS_SESSION_OPEN: return "open";
391	640	case CEPH_MDS_SESSION_HUNG: return "hung";
392	641	case CEPH_MDS_SESSION_CLOSING: return "closing";
	642	+ case CEPH_MDS_SESSION_CLOSED: return "closed";
393	643	case CEPH_MDS_SESSION_RESTARTING: return "restarting";
394	644	case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting";
395	645	case CEPH_MDS_SESSION_REJECTED: return "rejected";
..	..	@@ -397,7 +647,7 @@
397	647	}
398	648	}
399	649
400		-static struct ceph_mds_session get_session(struct ceph_mds_session s)
	650	+struct ceph_mds_session ceph_get_mds_session(struct ceph_mds_session s)
401	651	{
402	652	if (refcount_inc_not_zero(&s->s_ref)) {
403	653	dout("mdsc get_session %p %d -> %d\n", s,
..	..	@@ -411,11 +661,16 @@
411	661
412	662	void ceph_put_mds_session(struct ceph_mds_session *s)
413	663	{
	664	+ if (IS_ERR_OR_NULL(s))
	665	+ return;
	666	+
414	667	dout("mdsc put_session %p %d -> %d\n", s,
415	668	refcount_read(&s->s_ref), refcount_read(&s->s_ref)-1);
416	669	if (refcount_dec_and_test(&s->s_ref)) {
417	670	if (s->s_auth.authorizer)
418	671	ceph_auth_destroy_authorizer(s->s_auth.authorizer);
	672	+ WARN_ON(mutex_is_locked(&s->s_mutex));
	673	+ xa_destroy(&s->s_delegated_inos);
419	674	kfree(s);
420	675	}
421	676	}
..	..	@@ -426,15 +681,9 @@
426	681	struct ceph_mds_session __ceph_lookup_mds_session(struct ceph_mds_client mdsc,
427	682	int mds)
428	683	{
429		- struct ceph_mds_session *session;
430		-
431	684	if (mds >= mdsc->max_sessions \|\| !mdsc->sessions[mds])
432	685	return NULL;
433		- session = mdsc->sessions[mds];
434		- dout("lookup_mds_session %p %d\n", session,
435		- refcount_read(&session->s_ref));
436		- get_session(session);
437		- return session;
	686	+ return ceph_get_mds_session(mdsc->sessions[mds]);
438	687	}
439	688
440	689	static bool __have_session(struct ceph_mds_client *mdsc, int mds)
..	..	@@ -463,7 +712,7 @@
463	712	{
464	713	struct ceph_mds_session *s;
465	714
466		- if (mds >= mdsc->mdsmap->m_num_mds)
	715	+ if (mds >= mdsc->mdsmap->possible_max_rank)
467	716	return ERR_PTR(-EINVAL);
468	717
469	718	s = kzalloc(sizeof(*s), GFP_NOFS);
..	..	@@ -498,7 +747,7 @@
498	747	ceph_con_init(&s->s_con, s, &mds_con_ops, &mdsc->fsc->client->msgr);
499	748
500	749	spin_lock_init(&s->s_gen_ttl_lock);
501		- s->s_cap_gen = 0;
	750	+ s->s_cap_gen = 1;
502	751	s->s_cap_ttl = jiffies - 1;
503	752
504	753	spin_lock_init(&s->s_cap_lock);
..	..	@@ -506,14 +755,17 @@
506	755	s->s_renew_seq = 0;
507	756	INIT_LIST_HEAD(&s->s_caps);
508	757	s->s_nr_caps = 0;
509		- s->s_trim_caps = 0;
510	758	refcount_set(&s->s_ref, 1);
511	759	INIT_LIST_HEAD(&s->s_waiting);
512	760	INIT_LIST_HEAD(&s->s_unsafe);
	761	+ xa_init(&s->s_delegated_inos);
513	762	s->s_num_cap_releases = 0;
514	763	s->s_cap_reconnect = 0;
515	764	s->s_cap_iterator = NULL;
516	765	INIT_LIST_HEAD(&s->s_cap_releases);
	766	+ INIT_WORK(&s->s_cap_release_work, ceph_cap_release_work);
	767	+
	768	+ INIT_LIST_HEAD(&s->s_cap_dirty);
517	769	INIT_LIST_HEAD(&s->s_cap_flushing);
518	770
519	771	mdsc->sessions[mds] = s;
..	..	@@ -557,11 +809,39 @@
557	809	}
558	810	}
559	811
	812	+void ceph_mdsc_iterate_sessions(struct ceph_mds_client *mdsc,
	813	+ void (cb)(struct ceph_mds_session ),
	814	+ bool check_state)
	815	+{
	816	+ int mds;
	817	+
	818	+ mutex_lock(&mdsc->mutex);
	819	+ for (mds = 0; mds < mdsc->max_sessions; ++mds) {
	820	+ struct ceph_mds_session *s;
	821	+
	822	+ s = __ceph_lookup_mds_session(mdsc, mds);
	823	+ if (!s)
	824	+ continue;
	825	+
	826	+ if (check_state && !check_session_state(s)) {
	827	+ ceph_put_mds_session(s);
	828	+ continue;
	829	+ }
	830	+
	831	+ mutex_unlock(&mdsc->mutex);
	832	+ cb(s);
	833	+ ceph_put_mds_session(s);
	834	+ mutex_lock(&mdsc->mutex);
	835	+ }
	836	+ mutex_unlock(&mdsc->mutex);
	837	+}
	838	+
560	839	void ceph_mdsc_release_request(struct kref *kref)
561	840	{
562	841	struct ceph_mds_request *req = container_of(kref,
563	842	struct ceph_mds_request,
564	843	r_kref);
	844	+ ceph_mdsc_release_dir_caps_no_check(req);
565	845	destroy_reply_info(&req->r_reply_info);
566	846	if (req->r_request)
567	847	ceph_msg_put(req->r_request);
..	..	@@ -569,11 +849,14 @@
569	849	ceph_msg_put(req->r_reply);
570	850	if (req->r_inode) {
571	851	ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
572		- iput(req->r_inode);
	852	+ /* avoid calling iput_final() in mds dispatch threads */
	853	+ ceph_async_iput(req->r_inode);
573	854	}
574		- if (req->r_parent)
	855	+ if (req->r_parent) {
575	856	ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
576		- iput(req->r_target_inode);
	857	+ ceph_async_iput(req->r_parent);
	858	+ }
	859	+ ceph_async_iput(req->r_target_inode);
577	860	if (req->r_dentry)
578	861	dput(req->r_dentry);
579	862	if (req->r_old_dentry)
..	..	@@ -587,7 +870,7 @@
587	870	*/
588	871	ceph_put_cap_refs(ceph_inode(req->r_old_dentry_dir),
589	872	CEPH_CAP_PIN);
590		- iput(req->r_old_dentry_dir);
	873	+ ceph_async_iput(req->r_old_dentry_dir);
591	874	}
592	875	kfree(req->r_path1);
593	876	kfree(req->r_path2);
..	..	@@ -595,7 +878,8 @@
595	878	ceph_pagelist_release(req->r_pagelist);
596	879	put_request_session(req);
597	880	ceph_unreserve_caps(req->r_mdsc, &req->r_caps_reservation);
598		- kfree(req);
	881	+ WARN_ON_ONCE(!list_empty(&req->r_wait));
	882	+ kmem_cache_free(ceph_mds_request_cachep, req);
599	883	}
600	884
601	885	DEFINE_RB_FUNCS(request, struct ceph_mds_request, r_tid, r_node)
..	..	@@ -652,8 +936,13 @@
652	936	mdsc->oldest_tid = req->r_tid;
653	937
654	938	if (dir) {
	939	+ struct ceph_inode_info *ci = ceph_inode(dir);
	940	+
655	941	ihold(dir);
656	942	req->r_unsafe_dir = dir;
	943	+ spin_lock(&ci->i_unsafe_lock);
	944	+ list_add_tail(&req->r_unsafe_dir_item, &ci->i_unsafe_dirops);
	945	+ spin_unlock(&ci->i_unsafe_lock);
657	946	}
658	947	}
659	948
..	..	@@ -681,8 +970,7 @@
681	970
682	971	erase_request(&mdsc->request_tree, req);
683	972
684		- if (req->r_unsafe_dir &&
685		- test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
	973	+ if (req->r_unsafe_dir) {
686	974	struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
687	975	spin_lock(&ci->i_unsafe_lock);
688	976	list_del_init(&req->r_unsafe_dir_item);
..	..	@@ -697,7 +985,8 @@
697	985	}
698	986
699	987	if (req->r_unsafe_dir) {
700		- iput(req->r_unsafe_dir);
	988	+ /* avoid calling iput_final() in mds dispatch threads */
	989	+ ceph_async_iput(req->r_unsafe_dir);
701	990	req->r_unsafe_dir = NULL;
702	991	}
703	992
..	..	@@ -737,7 +1026,8 @@
737	1026	* Called under mdsc->mutex.
738	1027	*/
739	1028	static int __choose_mds(struct ceph_mds_client *mdsc,
740		- struct ceph_mds_request *req)
	1029	+ struct ceph_mds_request *req,
	1030	+ bool *random)
741	1031	{
742	1032	struct inode *inode;
743	1033	struct ceph_inode_info *ci;
..	..	@@ -747,6 +1037,9 @@
747	1037	u32 hash = req->r_direct_hash;
748	1038	bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
749	1039
	1040	+ if (random)
	1041	+ *random = false;
	1042	+
750	1043	/*
751	1044	* is there a specific mds we should try? ignore hint if we have
752	1045	* no session and the mds is not up (active or recovering).
..	..	@@ -754,7 +1047,7 @@
754	1047	if (req->r_resend_mds >= 0 &&
755	1048	(__have_session(mdsc, req->r_resend_mds) \|\|
756	1049	ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) {
757		- dout("choose_mds using resend_mds mds%d\n",
	1050	+ dout("%s using resend_mds mds%d\n", __func__,
758	1051	req->r_resend_mds);
759	1052	return req->r_resend_mds;
760	1053	}
..	..	@@ -772,7 +1065,7 @@
772	1065	rcu_read_lock();
773	1066	inode = get_nonsnap_parent(req->r_dentry);
774	1067	rcu_read_unlock();
775		- dout("__choose_mds using snapdir's parent %p\n", inode);
	1068	+ dout("%s using snapdir's parent %p\n", __func__, inode);
776	1069	}
777	1070	} else if (req->r_dentry) {
778	1071	/* ignore race with rename; old or new d_parent is okay */
..	..	@@ -780,7 +1073,7 @@
780	1073	struct inode *dir;
781	1074
782	1075	rcu_read_lock();
783		- parent = req->r_dentry->d_parent;
	1076	+ parent = READ_ONCE(req->r_dentry->d_parent);
784	1077	dir = req->r_parent ? : d_inode_rcu(parent);
785	1078
786	1079	if (!dir \|\| dir->i_sb != mdsc->fsc->sb) {
..	..	@@ -792,7 +1085,7 @@
792	1085	/* direct snapped/virtual snapdir requests
793	1086	* based on parent dir inode */
794	1087	inode = get_nonsnap_parent(parent);
795		- dout("__choose_mds using nonsnap parent %p\n", inode);
	1088	+ dout("%s using nonsnap parent %p\n", __func__, inode);
796	1089	} else {
797	1090	/* dentry target */
798	1091	inode = d_inode(req->r_dentry);
..	..	@@ -808,8 +1101,8 @@
808	1101	rcu_read_unlock();
809	1102	}
810	1103
811		- dout("__choose_mds %p is_hash=%d (%d) mode %d\n", inode, (int)is_hash,
812		- (int)hash, mode);
	1104	+ dout("%s %p is_hash=%d (0x%x) mode %d\n", __func__, inode, (int)is_hash,
	1105	+ hash, mode);
813	1106	if (!inode)
814	1107	goto random;
815	1108	ci = ceph_inode(inode);
..	..	@@ -827,30 +1120,32 @@
827	1120	get_random_bytes(&r, 1);
828	1121	r %= frag.ndist;
829	1122	mds = frag.dist[r];
830		- dout("choose_mds %p %llx.%llx "
831		- "frag %u mds%d (%d/%d)\n",
832		- inode, ceph_vinop(inode),
833		- frag.frag, mds,
834		- (int)r, frag.ndist);
	1123	+ dout("%s %p %llx.%llx frag %u mds%d (%d/%d)\n",
	1124	+ __func__, inode, ceph_vinop(inode),
	1125	+ frag.frag, mds, (int)r, frag.ndist);
835	1126	if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
836		- CEPH_MDS_STATE_ACTIVE)
	1127	+ CEPH_MDS_STATE_ACTIVE &&
	1128	+ !ceph_mdsmap_is_laggy(mdsc->mdsmap, mds))
837	1129	goto out;
838	1130	}
839	1131
840	1132	/* since this file/dir wasn't known to be
841	1133	* replicated, then we want to look for the
842	1134	* authoritative mds. */
843		- mode = USE_AUTH_MDS;
844	1135	if (frag.mds >= 0) {
845	1136	/* choose auth mds */
846	1137	mds = frag.mds;
847		- dout("choose_mds %p %llx.%llx "
848		- "frag %u mds%d (auth)\n",
849		- inode, ceph_vinop(inode), frag.frag, mds);
	1138	+ dout("%s %p %llx.%llx frag %u mds%d (auth)\n",
	1139	+ __func__, inode, ceph_vinop(inode),
	1140	+ frag.frag, mds);
850	1141	if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
851		- CEPH_MDS_STATE_ACTIVE)
852		- goto out;
	1142	+ CEPH_MDS_STATE_ACTIVE) {
	1143	+ if (!ceph_mdsmap_is_laggy(mdsc->mdsmap,
	1144	+ mds))
	1145	+ goto out;
	1146	+ }
853	1147	}
	1148	+ mode = USE_AUTH_MDS;
854	1149	}
855	1150	}
856	1151
..	..	@@ -862,21 +1157,26 @@
862	1157	cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node);
863	1158	if (!cap) {
864	1159	spin_unlock(&ci->i_ceph_lock);
865		- iput(inode);
	1160	+ ceph_async_iput(inode);
866	1161	goto random;
867	1162	}
868	1163	mds = cap->session->s_mds;
869		- dout("choose_mds %p %llx.%llx mds%d (%scap %p)\n",
	1164	+ dout("%s %p %llx.%llx mds%d (%scap %p)\n", __func__,
870	1165	inode, ceph_vinop(inode), mds,
871	1166	cap == ci->i_auth_cap ? "auth " : "", cap);
872	1167	spin_unlock(&ci->i_ceph_lock);
873	1168	out:
874		- iput(inode);
	1169	+ /* avoid calling iput_final() while holding mdsc->mutex or
	1170	+ * in mds dispatch threads */
	1171	+ ceph_async_iput(inode);
875	1172	return mds;
876	1173
877	1174	random:
	1175	+ if (random)
	1176	+ *random = true;
	1177	+
878	1178	mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap);
879		- dout("choose_mds chose random mds%d\n", mds);
	1179	+ dout("%s chose random mds%d\n", __func__, mds);
880	1180	return mds;
881	1181	}
882	1182
..	..	@@ -884,7 +1184,7 @@
884	1184	/*
885	1185	* session messages
886	1186	*/
887		-static struct ceph_msg *create_session_msg(u32 op, u64 seq)
	1187	+struct ceph_msg *ceph_create_session_msg(u32 op, u64 seq)
888	1188	{
889	1189	struct ceph_msg *msg;
890	1190	struct ceph_mds_session_head *h;
..	..	@@ -892,7 +1192,8 @@
892	1192	msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS,
893	1193	false);
894	1194	if (!msg) {
895		- pr_err("create_session_msg ENOMEM creating msg\n");
	1195	+ pr_err("ENOMEM creating session %s msg\n",
	1196	+ ceph_session_op_name(op));
896	1197	return NULL;
897	1198	}
898	1199	h = msg->front.iov_base;
..	..	@@ -902,25 +1203,77 @@
902	1203	return msg;
903	1204	}
904	1205
905		-static void encode_supported_features(void *p, void end)
	1206	+static const unsigned char feature_bits[] = CEPHFS_FEATURES_CLIENT_SUPPORTED;
	1207	+#define FEATURE_BYTES(c) (DIV_ROUND_UP((size_t)feature_bits[c - 1] + 1, 64) * 8)
	1208	+static int encode_supported_features(void *p, void end)
906	1209	{
907		- static const unsigned char bits[] = CEPHFS_FEATURES_CLIENT_SUPPORTED;
908		- static const size_t count = ARRAY_SIZE(bits);
	1210	+ static const size_t count = ARRAY_SIZE(feature_bits);
909	1211
910	1212	if (count > 0) {
911	1213	size_t i;
912		- size_t size = ((size_t)bits[count - 1] + 64) / 64 * 8;
	1214	+ size_t size = FEATURE_BYTES(count);
	1215	+ unsigned long bit;
913	1216
914		- BUG_ON(*p + 4 + size > end);
	1217	+ if (WARN_ON_ONCE(*p + 4 + size > end))
	1218	+ return -ERANGE;
	1219	+
	1220	+ ceph_encode_32(p, size);
	1221	+ memset(*p, 0, size);
	1222	+ for (i = 0; i < count; i++) {
	1223	+ bit = feature_bits[i];
	1224	+ ((unsigned char )(p))[bit / 8] \|= BIT(bit % 8);
	1225	+ }
	1226	+ *p += size;
	1227	+ } else {
	1228	+ if (WARN_ON_ONCE(*p + 4 > end))
	1229	+ return -ERANGE;
	1230	+
	1231	+ ceph_encode_32(p, 0);
	1232	+ }
	1233	+
	1234	+ return 0;
	1235	+}
	1236	+
	1237	+static const unsigned char metric_bits[] = CEPHFS_METRIC_SPEC_CLIENT_SUPPORTED;
	1238	+#define METRIC_BYTES(cnt) (DIV_ROUND_UP((size_t)metric_bits[cnt - 1] + 1, 64) * 8)
	1239	+static int encode_metric_spec(void *p, void end)
	1240	+{
	1241	+ static const size_t count = ARRAY_SIZE(metric_bits);
	1242	+
	1243	+ /* header */
	1244	+ if (WARN_ON_ONCE(*p + 2 > end))
	1245	+ return -ERANGE;
	1246	+
	1247	+ ceph_encode_8(p, 1); /* version */
	1248	+ ceph_encode_8(p, 1); /* compat */
	1249	+
	1250	+ if (count > 0) {
	1251	+ size_t i;
	1252	+ size_t size = METRIC_BYTES(count);
	1253	+
	1254	+ if (WARN_ON_ONCE(*p + 4 + 4 + size > end))
	1255	+ return -ERANGE;
	1256	+
	1257	+ /* metric spec info length */
	1258	+ ceph_encode_32(p, 4 + size);
	1259	+
	1260	+ /* metric spec */
915	1261	ceph_encode_32(p, size);
916	1262	memset(*p, 0, size);
917	1263	for (i = 0; i < count; i++)
918		- ((unsigned char)(p))[i / 8] \|= 1 << (bits[i] % 8);
	1264	+ ((unsigned char )(p))[i / 8] \|= BIT(metric_bits[i] % 8);
919	1265	*p += size;
920	1266	} else {
921		- BUG_ON(*p + 4 > end);
	1267	+ if (WARN_ON_ONCE(*p + 4 + 4 > end))
	1268	+ return -ERANGE;
	1269	+
	1270	+ /* metric spec info length */
	1271	+ ceph_encode_32(p, 4);
	1272	+ /* metric spec */
922	1273	ceph_encode_32(p, 0);
923	1274	}
	1275	+
	1276	+ return 0;
924	1277	}
925	1278
926	1279	/*
..	..	@@ -936,7 +1289,9 @@
936	1289	int metadata_key_count = 0;
937	1290	struct ceph_options *opt = mdsc->fsc->client->options;
938	1291	struct ceph_mount_options *fsopt = mdsc->fsc->mount_options;
	1292	+ size_t size, count;
939	1293	void p, end;
	1294	+ int ret;
940	1295
941	1296	const char* metadata[][2] = {
942	1297	{"hostname", mdsc->nodename},
..	..	@@ -953,15 +1308,27 @@
953	1308	strlen(metadata[i][1]);
954	1309	metadata_key_count++;
955	1310	}
	1311	+
956	1312	/* supported feature */
957		- extra_bytes += 4 + 8;
	1313	+ size = 0;
	1314	+ count = ARRAY_SIZE(feature_bits);
	1315	+ if (count > 0)
	1316	+ size = FEATURE_BYTES(count);
	1317	+ extra_bytes += 4 + size;
	1318	+
	1319	+ /* metric spec */
	1320	+ size = 0;
	1321	+ count = ARRAY_SIZE(metric_bits);
	1322	+ if (count > 0)
	1323	+ size = METRIC_BYTES(count);
	1324	+ extra_bytes += 2 + 4 + 4 + size;
958	1325
959	1326	/* Allocate the message */
960	1327	msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h) + extra_bytes,
961	1328	GFP_NOFS, false);
962	1329	if (!msg) {
963		- pr_err("create_session_msg ENOMEM creating msg\n");
964		- return NULL;
	1330	+ pr_err("ENOMEM creating session open msg\n");
	1331	+ return ERR_PTR(-ENOMEM);
965	1332	}
966	1333	p = msg->front.iov_base;
967	1334	end = p + msg->front.iov_len;
..	..	@@ -974,9 +1341,9 @@
974	1341	* Serialize client metadata into waiting buffer space, using
975	1342	* the format that userspace expects for map<string, string>
976	1343	*
977		- * ClientSession messages with metadata are v2
	1344	+ * ClientSession messages with metadata are v4
978	1345	*/
979		- msg->hdr.version = cpu_to_le16(3);
	1346	+ msg->hdr.version = cpu_to_le16(4);
980	1347	msg->hdr.compat_version = cpu_to_le16(1);
981	1348
982	1349	/* The write pointer, following the session_head structure */
..	..	@@ -998,7 +1365,20 @@
998	1365	p += val_len;
999	1366	}
1000	1367
1001		- encode_supported_features(&p, end);
	1368	+ ret = encode_supported_features(&p, end);
	1369	+ if (ret) {
	1370	+ pr_err("encode_supported_features failed!\n");
	1371	+ ceph_msg_put(msg);
	1372	+ return ERR_PTR(ret);
	1373	+ }
	1374	+
	1375	+ ret = encode_metric_spec(&p, end);
	1376	+ if (ret) {
	1377	+ pr_err("encode_metric_spec failed!\n");
	1378	+ ceph_msg_put(msg);
	1379	+ return ERR_PTR(ret);
	1380	+ }
	1381	+
1002	1382	msg->front.iov_len = p - msg->front.iov_base;
1003	1383	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
1004	1384
..	..	@@ -1026,8 +1406,8 @@
1026	1406
1027	1407	/* send connect message */
1028	1408	msg = create_session_open_msg(mdsc, session->s_seq);
1029		- if (!msg)
1030		- return -ENOMEM;
	1409	+ if (IS_ERR(msg))
	1410	+ return PTR_ERR(msg);
1031	1411	ceph_con_send(&session->s_con, msg);
1032	1412	return 0;
1033	1413	}
..	..	@@ -1041,6 +1421,7 @@
1041	1421	__open_export_target_session(struct ceph_mds_client *mdsc, int target)
1042	1422	{
1043	1423	struct ceph_mds_session *session;
	1424	+ int ret;
1044	1425
1045	1426	session = __ceph_lookup_mds_session(mdsc, target);
1046	1427	if (!session) {
..	..	@@ -1049,8 +1430,11 @@
1049	1430	return session;
1050	1431	}
1051	1432	if (session->s_state == CEPH_MDS_SESSION_NEW \|\|
1052		- session->s_state == CEPH_MDS_SESSION_CLOSING)
1053		- __open_session(mdsc, session);
	1433	+ session->s_state == CEPH_MDS_SESSION_CLOSING) {
	1434	+ ret = __open_session(mdsc, session);
	1435	+ if (ret)
	1436	+ return ERR_PTR(ret);
	1437	+ }
1054	1438
1055	1439	return session;
1056	1440	}
..	..	@@ -1076,7 +1460,7 @@
1076	1460	struct ceph_mds_session *ts;
1077	1461	int i, mds = session->s_mds;
1078	1462
1079		- if (mds >= mdsc->mdsmap->m_num_mds)
	1463	+ if (mds >= mdsc->mdsmap->possible_max_rank)
1080	1464	return;
1081	1465
1082	1466	mi = &mdsc->mdsmap->m_info[mds];
..	..	@@ -1085,8 +1469,7 @@
1085	1469
1086	1470	for (i = 0; i < mi->num_export_targets; i++) {
1087	1471	ts = __open_export_target_session(mdsc, mi->export_targets[i]);
1088		- if (!IS_ERR(ts))
1089		- ceph_put_mds_session(ts);
	1472	+ ceph_put_mds_session(ts);
1090	1473	}
1091	1474	}
1092	1475
..	..	@@ -1137,6 +1520,10 @@
1137	1520	struct ceph_mds_request, r_unsafe_item);
1138	1521	pr_warn_ratelimited(" dropping unsafe request %llu\n",
1139	1522	req->r_tid);
	1523	+ if (req->r_target_inode)
	1524	+ mapping_set_error(req->r_target_inode->i_mapping, -EIO);
	1525	+ if (req->r_unsafe_dir)
	1526	+ mapping_set_error(req->r_unsafe_dir->i_mapping, -EIO);
1140	1527	__unregister_request(mdsc, req);
1141	1528	}
1142	1529	/* zero r_attempts, so kick_requests() will re-send requests */
..	..	@@ -1157,9 +1544,9 @@
1157	1544	*
1158	1545	* Caller must hold session s_mutex.
1159	1546	*/
1160		-static int iterate_session_caps(struct ceph_mds_session *session,
1161		- int (cb)(struct inode , struct ceph_cap *,
1162		- void ), void arg)
	1547	+int ceph_iterate_session_caps(struct ceph_mds_session *session,
	1548	+ int (cb)(struct inode , struct ceph_cap *,
	1549	+ void ), void arg)
1163	1550	{
1164	1551	struct list_head *p;
1165	1552	struct ceph_cap *cap;
..	..	@@ -1181,7 +1568,9 @@
1181	1568	spin_unlock(&session->s_cap_lock);
1182	1569
1183	1570	if (last_inode) {
1184		- iput(last_inode);
	1571	+ /* avoid calling iput_final() while holding
	1572	+ * s_mutex or in mds dispatch threads */
	1573	+ ceph_async_iput(last_inode);
1185	1574	last_inode = NULL;
1186	1575	}
1187	1576	if (old_cap) {
..	..	@@ -1201,13 +1590,11 @@
1201	1590	cap->session = NULL;
1202	1591	list_del_init(&cap->session_caps);
1203	1592	session->s_nr_caps--;
1204		- if (cap->queue_release) {
1205		- list_add_tail(&cap->session_caps,
1206		- &session->s_cap_releases);
1207		- session->s_num_cap_releases++;
1208		- } else {
	1593	+ atomic64_dec(&session->s_mdsc->metric.total_caps);
	1594	+ if (cap->queue_release)
	1595	+ __ceph_queue_cap_release(session, cap);
	1596	+ else
1209	1597	old_cap = cap; /* put_cap it w/o locks held */
1210		- }
1211	1598	}
1212	1599	if (ret < 0)
1213	1600	goto out;
..	..	@@ -1217,21 +1604,46 @@
1217	1604	session->s_cap_iterator = NULL;
1218	1605	spin_unlock(&session->s_cap_lock);
1219	1606
1220		- iput(last_inode);
	1607	+ ceph_async_iput(last_inode);
1221	1608	if (old_cap)
1222	1609	ceph_put_cap(session->s_mdsc, old_cap);
1223	1610
1224	1611	return ret;
1225	1612	}
1226	1613
	1614	+static int remove_capsnaps(struct ceph_mds_client mdsc, struct inode inode)
	1615	+{
	1616	+ struct ceph_inode_info *ci = ceph_inode(inode);
	1617	+ struct ceph_cap_snap *capsnap;
	1618	+ int capsnap_release = 0;
	1619	+
	1620	+ lockdep_assert_held(&ci->i_ceph_lock);
	1621	+
	1622	+ dout("removing capsnaps, ci is %p, inode is %p\n", ci, inode);
	1623	+
	1624	+ while (!list_empty(&ci->i_cap_snaps)) {
	1625	+ capsnap = list_first_entry(&ci->i_cap_snaps,
	1626	+ struct ceph_cap_snap, ci_item);
	1627	+ __ceph_remove_capsnap(inode, capsnap, NULL, NULL);
	1628	+ ceph_put_snap_context(capsnap->context);
	1629	+ ceph_put_cap_snap(capsnap);
	1630	+ capsnap_release++;
	1631	+ }
	1632	+ wake_up_all(&ci->i_cap_wq);
	1633	+ wake_up_all(&mdsc->cap_flushing_wq);
	1634	+ return capsnap_release;
	1635	+}
	1636	+
1227	1637	static int remove_session_caps_cb(struct inode inode, struct ceph_cap cap,
1228	1638	void *arg)
1229	1639	{
1230	1640	struct ceph_fs_client fsc = (struct ceph_fs_client )arg;
	1641	+ struct ceph_mds_client *mdsc = fsc->mdsc;
1231	1642	struct ceph_inode_info *ci = ceph_inode(inode);
1232	1643	LIST_HEAD(to_remove);
1233		- bool drop = false;
	1644	+ bool dirty_dropped = false;
1234	1645	bool invalidate = false;
	1646	+ int capsnap_release = 0;
1235	1647
1236	1648	dout("removing cap %p, ci is %p, inode is %p\n",
1237	1649	cap, ci, &ci->vfs_inode);
..	..	@@ -1239,13 +1651,13 @@
1239	1651	__ceph_remove_cap(cap, false);
1240	1652	if (!ci->i_auth_cap) {
1241	1653	struct ceph_cap_flush *cf;
1242		- struct ceph_mds_client *mdsc = fsc->mdsc;
1243	1654
1244		- ci->i_ceph_flags \|= CEPH_I_CAP_DROPPED;
1245		-
1246		- if (ci->i_wrbuffer_ref > 0 &&
1247		- READ_ONCE(fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)
1248		- invalidate = true;
	1655	+ if (READ_ONCE(fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
	1656	+ if (inode->i_data.nrpages > 0)
	1657	+ invalidate = true;
	1658	+ if (ci->i_wrbuffer_ref > 0)
	1659	+ mapping_set_error(&inode->i_data, -EIO);
	1660	+ }
1249	1661
1250	1662	while (!list_empty(&ci->i_cap_flush_list)) {
1251	1663	cf = list_first_entry(&ci->i_cap_flush_list,
..	..	@@ -1256,7 +1668,7 @@
1256	1668	spin_lock(&mdsc->cap_dirty_lock);
1257	1669
1258	1670	list_for_each_entry(cf, &to_remove, i_list)
1259		- list_del(&cf->g_list);
	1671	+ list_del_init(&cf->g_list);
1260	1672
1261	1673	if (!list_empty(&ci->i_dirty_item)) {
1262	1674	pr_warn_ratelimited(
..	..	@@ -1265,7 +1677,7 @@
1265	1677	inode, ceph_ino(inode));
1266	1678	ci->i_dirty_caps = 0;
1267	1679	list_del_init(&ci->i_dirty_item);
1268		- drop = true;
	1680	+ dirty_dropped = true;
1269	1681	}
1270	1682	if (!list_empty(&ci->i_flushing_item)) {
1271	1683	pr_warn_ratelimited(
..	..	@@ -1275,9 +1687,21 @@
1275	1687	ci->i_flushing_caps = 0;
1276	1688	list_del_init(&ci->i_flushing_item);
1277	1689	mdsc->num_cap_flushing--;
1278		- drop = true;
	1690	+ dirty_dropped = true;
1279	1691	}
1280	1692	spin_unlock(&mdsc->cap_dirty_lock);
	1693	+
	1694	+ if (dirty_dropped) {
	1695	+ mapping_set_error(inode->i_mapping, -EIO);
	1696	+
	1697	+ if (ci->i_wrbuffer_ref_head == 0 &&
	1698	+ ci->i_wr_ref == 0 &&
	1699	+ ci->i_dirty_caps == 0 &&
	1700	+ ci->i_flushing_caps == 0) {
	1701	+ ceph_put_snap_context(ci->i_head_snapc);
	1702	+ ci->i_head_snapc = NULL;
	1703	+ }
	1704	+ }
1281	1705
1282	1706	if (atomic_read(&ci->i_filelock_ref) > 0) {
1283	1707	/* make further file lock syscall return -EIO */
..	..	@@ -1291,28 +1715,25 @@
1291	1715	ci->i_prealloc_cap_flush = NULL;
1292	1716	}
1293	1717
1294		- if (drop &&
1295		- ci->i_wrbuffer_ref_head == 0 &&
1296		- ci->i_wr_ref == 0 &&
1297		- ci->i_dirty_caps == 0 &&
1298		- ci->i_flushing_caps == 0) {
1299		- ceph_put_snap_context(ci->i_head_snapc);
1300		- ci->i_head_snapc = NULL;
1301		- }
	1718	+ if (!list_empty(&ci->i_cap_snaps))
	1719	+ capsnap_release = remove_capsnaps(mdsc, inode);
1302	1720	}
1303	1721	spin_unlock(&ci->i_ceph_lock);
1304	1722	while (!list_empty(&to_remove)) {
1305	1723	struct ceph_cap_flush *cf;
1306	1724	cf = list_first_entry(&to_remove,
1307	1725	struct ceph_cap_flush, i_list);
1308		- list_del(&cf->i_list);
1309		- ceph_free_cap_flush(cf);
	1726	+ list_del_init(&cf->i_list);
	1727	+ if (!cf->is_capsnap)
	1728	+ ceph_free_cap_flush(cf);
1310	1729	}
1311	1730
1312	1731	wake_up_all(&ci->i_cap_wq);
1313	1732	if (invalidate)
1314	1733	ceph_queue_invalidate(inode);
1315		- if (drop)
	1734	+ if (dirty_dropped)
	1735	+ iput(inode);
	1736	+ while (capsnap_release--)
1316	1737	iput(inode);
1317	1738	return 0;
1318	1739	}
..	..	@@ -1327,7 +1748,7 @@
1327	1748	LIST_HEAD(dispose);
1328	1749
1329	1750	dout("remove_session_caps on %p\n", session);
1330		- iterate_session_caps(session, remove_session_caps_cb, fsc);
	1751	+ ceph_iterate_session_caps(session, remove_session_caps_cb, fsc);
1331	1752
1332	1753	wake_up_all(&fsc->mdsc->cap_flushing_wq);
1333	1754
..	..	@@ -1353,7 +1774,8 @@
1353	1774	spin_unlock(&session->s_cap_lock);
1354	1775
1355	1776	inode = ceph_find_inode(sb, vino);
1356		- iput(inode);
	1777	+ /* avoid calling iput_final() while holding s_mutex */
	1778	+ ceph_async_iput(inode);
1357	1779
1358	1780	spin_lock(&session->s_cap_lock);
1359	1781	}
..	..	@@ -1368,6 +1790,12 @@
1368	1790	dispose_cap_releases(session->s_mdsc, &dispose);
1369	1791	}
1370	1792
	1793	+enum {
	1794	+ RECONNECT,
	1795	+ RENEWCAPS,
	1796	+ FORCE_RO,
	1797	+};
	1798	+
1371	1799	/*
1372	1800	* wake up any threads waiting on this session's caps. if the cap is
1373	1801	* old (didn't get renewed on the client reconnect), remove it now.
..	..	@@ -1378,23 +1806,31 @@
1378	1806	void *arg)
1379	1807	{
1380	1808	struct ceph_inode_info *ci = ceph_inode(inode);
	1809	+ unsigned long ev = (unsigned long)arg;
1381	1810
1382		- if (arg) {
	1811	+ if (ev == RECONNECT) {
1383	1812	spin_lock(&ci->i_ceph_lock);
1384	1813	ci->i_wanted_max_size = 0;
1385	1814	ci->i_requested_max_size = 0;
1386	1815	spin_unlock(&ci->i_ceph_lock);
	1816	+ } else if (ev == RENEWCAPS) {
	1817	+ if (cap->cap_gen < cap->session->s_cap_gen) {
	1818	+ /* mds did not re-issue stale cap */
	1819	+ spin_lock(&ci->i_ceph_lock);
	1820	+ cap->issued = cap->implemented = CEPH_CAP_PIN;
	1821	+ spin_unlock(&ci->i_ceph_lock);
	1822	+ }
	1823	+ } else if (ev == FORCE_RO) {
1387	1824	}
1388	1825	wake_up_all(&ci->i_cap_wq);
1389	1826	return 0;
1390	1827	}
1391	1828
1392		-static void wake_up_session_caps(struct ceph_mds_session *session,
1393		- int reconnect)
	1829	+static void wake_up_session_caps(struct ceph_mds_session *session, int ev)
1394	1830	{
1395	1831	dout("wake_up_session_caps %p mds%d\n", session, session->s_mds);
1396		- iterate_session_caps(session, wake_up_session_cb,
1397		- (void *)(unsigned long)reconnect);
	1832	+ ceph_iterate_session_caps(session, wake_up_session_cb,
	1833	+ (void *)(unsigned long)ev);
1398	1834	}
1399	1835
1400	1836	/*
..	..	@@ -1425,8 +1861,8 @@
1425	1861
1426	1862	dout("send_renew_caps to mds%d (%s)\n", session->s_mds,
1427	1863	ceph_mds_state_name(state));
1428		- msg = create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS,
1429		- ++session->s_renew_seq);
	1864	+ msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS,
	1865	+ ++session->s_renew_seq);
1430	1866	if (!msg)
1431	1867	return -ENOMEM;
1432	1868	ceph_con_send(&session->s_con, msg);
..	..	@@ -1440,7 +1876,7 @@
1440	1876
1441	1877	dout("send_flushmsg_ack to mds%d (%s)s seq %lld\n",
1442	1878	session->s_mds, ceph_session_state_name(session->s_state), seq);
1443		- msg = create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq);
	1879	+ msg = ceph_create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq);
1444	1880	if (!msg)
1445	1881	return -ENOMEM;
1446	1882	ceph_con_send(&session->s_con, msg);
..	..	@@ -1479,21 +1915,21 @@
1479	1915	spin_unlock(&session->s_cap_lock);
1480	1916
1481	1917	if (wake)
1482		- wake_up_session_caps(session, 0);
	1918	+ wake_up_session_caps(session, RENEWCAPS);
1483	1919	}
1484	1920
1485	1921	/*
1486	1922	* send a session close request
1487	1923	*/
1488		-static int request_close_session(struct ceph_mds_client *mdsc,
1489		- struct ceph_mds_session *session)
	1924	+static int request_close_session(struct ceph_mds_session *session)
1490	1925	{
1491	1926	struct ceph_msg *msg;
1492	1927
1493	1928	dout("request_close_session mds%d state %s seq %lld\n",
1494	1929	session->s_mds, ceph_session_state_name(session->s_state),
1495	1930	session->s_seq);
1496		- msg = create_session_msg(CEPH_SESSION_REQUEST_CLOSE, session->s_seq);
	1931	+ msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_CLOSE,
	1932	+ session->s_seq);
1497	1933	if (!msg)
1498	1934	return -ENOMEM;
1499	1935	ceph_con_send(&session->s_con, msg);
..	..	@@ -1509,7 +1945,7 @@
1509	1945	if (session->s_state >= CEPH_MDS_SESSION_CLOSING)
1510	1946	return 0;
1511	1947	session->s_state = CEPH_MDS_SESSION_CLOSING;
1512		- return request_close_session(mdsc, session);
	1948	+ return request_close_session(session);
1513	1949	}
1514	1950
1515	1951	static bool drop_negative_children(struct dentry *dentry)
..	..	@@ -1547,11 +1983,11 @@
1547	1983	*/
1548	1984	static int trim_caps_cb(struct inode inode, struct ceph_cap cap, void *arg)
1549	1985	{
1550		- struct ceph_mds_session *session = arg;
	1986	+ int *remaining = arg;
1551	1987	struct ceph_inode_info *ci = ceph_inode(inode);
1552	1988	int used, wanted, oissued, mine;
1553	1989
1554		- if (session->s_trim_caps <= 0)
	1990	+ if (*remaining <= 0)
1555	1991	return -1;
1556	1992
1557	1993	spin_lock(&ci->i_ceph_lock);
..	..	@@ -1577,7 +2013,8 @@
1577	2013	}
1578	2014	/* The inode has cached pages, but it's no longer used.
1579	2015	* we can safely drop it */
1580		- if (wanted == 0 && used == CEPH_CAP_FILE_CACHE &&
	2016	+ if (S_ISREG(inode->i_mode) &&
	2017	+ wanted == 0 && used == CEPH_CAP_FILE_CACHE &&
1581	2018	!(oissued & CEPH_CAP_FILE_CACHE)) {
1582	2019	used = 0;
1583	2020	oissued = 0;
..	..	@@ -1588,7 +2025,7 @@
1588	2025	if (oissued) {
1589	2026	/* we aren't the only cap.. just remove us */
1590	2027	__ceph_remove_cap(cap, true);
1591		- session->s_trim_caps--;
	2028	+ (*remaining)--;
1592	2029	} else {
1593	2030	struct dentry *dentry;
1594	2031	/* try dropping referring dentries */
..	..	@@ -1600,7 +2037,7 @@
1600	2037	d_prune_aliases(inode);
1601	2038	count = atomic_read(&inode->i_count);
1602	2039	if (count == 1)
1603		- session->s_trim_caps--;
	2040	+ (*remaining)--;
1604	2041	dout("trim_caps_cb %p cap %p pruned, count now %d\n",
1605	2042	inode, cap, count);
1606	2043	} else {
..	..	@@ -1626,15 +2063,15 @@
1626	2063	dout("trim_caps mds%d start: %d / %d, trim %d\n",
1627	2064	session->s_mds, session->s_nr_caps, max_caps, trim_caps);
1628	2065	if (trim_caps > 0) {
1629		- session->s_trim_caps = trim_caps;
1630		- iterate_session_caps(session, trim_caps_cb, session);
	2066	+ int remaining = trim_caps;
	2067	+
	2068	+ ceph_iterate_session_caps(session, trim_caps_cb, &remaining);
1631	2069	dout("trim_caps mds%d done: %d / %d, trimmed %d\n",
1632	2070	session->s_mds, session->s_nr_caps, max_caps,
1633		- trim_caps - session->s_trim_caps);
1634		- session->s_trim_caps = 0;
	2071	+ trim_caps - remaining);
1635	2072	}
1636	2073
1637		- ceph_send_cap_releases(mdsc, session);
	2074	+ ceph_flush_cap_releases(mdsc, session);
1638	2075	return 0;
1639	2076	}
1640	2077
..	..	@@ -1677,8 +2114,8 @@
1677	2114	/*
1678	2115	* called under s_mutex
1679	2116	*/
1680		-void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
1681		- struct ceph_mds_session *session)
	2117	+static void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
	2118	+ struct ceph_mds_session *session)
1682	2119	{
1683	2120	struct ceph_msg *msg = NULL;
1684	2121	struct ceph_mds_cap_release *head;
..	..	@@ -1720,7 +2157,8 @@
1720	2157	num_cap_releases--;
1721	2158
1722	2159	head = msg->front.iov_base;
1723		- le32_add_cpu(&head->num, 1);
	2160	+ put_unaligned_le32(get_unaligned_le32(&head->num) + 1,
	2161	+ &head->num);
1724	2162	item = msg->front.iov_base + msg->front.iov_len;
1725	2163	item->ino = cpu_to_le64(cap->cap_ino);
1726	2164	item->cap_id = cpu_to_le64(cap->cap_id);
..	..	@@ -1770,6 +2208,81 @@
1770	2208	spin_unlock(&session->s_cap_lock);
1771	2209	}
1772	2210
	2211	+static void ceph_cap_release_work(struct work_struct *work)
	2212	+{
	2213	+ struct ceph_mds_session *session =
	2214	+ container_of(work, struct ceph_mds_session, s_cap_release_work);
	2215	+
	2216	+ mutex_lock(&session->s_mutex);
	2217	+ if (session->s_state == CEPH_MDS_SESSION_OPEN \|\|
	2218	+ session->s_state == CEPH_MDS_SESSION_HUNG)
	2219	+ ceph_send_cap_releases(session->s_mdsc, session);
	2220	+ mutex_unlock(&session->s_mutex);
	2221	+ ceph_put_mds_session(session);
	2222	+}
	2223	+
	2224	+void ceph_flush_cap_releases(struct ceph_mds_client *mdsc,
	2225	+ struct ceph_mds_session *session)
	2226	+{
	2227	+ if (mdsc->stopping)
	2228	+ return;
	2229	+
	2230	+ ceph_get_mds_session(session);
	2231	+ if (queue_work(mdsc->fsc->cap_wq,
	2232	+ &session->s_cap_release_work)) {
	2233	+ dout("cap release work queued\n");
	2234	+ } else {
	2235	+ ceph_put_mds_session(session);
	2236	+ dout("failed to queue cap release work\n");
	2237	+ }
	2238	+}
	2239	+
	2240	+/*
	2241	+ * caller holds session->s_cap_lock
	2242	+ */
	2243	+void __ceph_queue_cap_release(struct ceph_mds_session *session,
	2244	+ struct ceph_cap *cap)
	2245	+{
	2246	+ list_add_tail(&cap->session_caps, &session->s_cap_releases);
	2247	+ session->s_num_cap_releases++;
	2248	+
	2249	+ if (!(session->s_num_cap_releases % CEPH_CAPS_PER_RELEASE))
	2250	+ ceph_flush_cap_releases(session->s_mdsc, session);
	2251	+}
	2252	+
	2253	+static void ceph_cap_reclaim_work(struct work_struct *work)
	2254	+{
	2255	+ struct ceph_mds_client *mdsc =
	2256	+ container_of(work, struct ceph_mds_client, cap_reclaim_work);
	2257	+ int ret = ceph_trim_dentries(mdsc);
	2258	+ if (ret == -EAGAIN)
	2259	+ ceph_queue_cap_reclaim_work(mdsc);
	2260	+}
	2261	+
	2262	+void ceph_queue_cap_reclaim_work(struct ceph_mds_client *mdsc)
	2263	+{
	2264	+ if (mdsc->stopping)
	2265	+ return;
	2266	+
	2267	+ if (queue_work(mdsc->fsc->cap_wq, &mdsc->cap_reclaim_work)) {
	2268	+ dout("caps reclaim work queued\n");
	2269	+ } else {
	2270	+ dout("failed to queue caps release work\n");
	2271	+ }
	2272	+}
	2273	+
	2274	+void ceph_reclaim_caps_nr(struct ceph_mds_client *mdsc, int nr)
	2275	+{
	2276	+ int val;
	2277	+ if (!nr)
	2278	+ return;
	2279	+ val = atomic_add_return(nr, &mdsc->cap_reclaim_pending);
	2280	+ if ((val % CEPH_CAPS_PER_RELEASE) < nr) {
	2281	+ atomic_set(&mdsc->cap_reclaim_pending, 0);
	2282	+ ceph_queue_cap_reclaim_work(mdsc);
	2283	+ }
	2284	+}
	2285	+
1773	2286	/*
1774	2287	* requests
1775	2288	*/
..	..	@@ -1781,12 +2294,13 @@
1781	2294	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
1782	2295	struct ceph_mount_options *opt = req->r_mdsc->fsc->mount_options;
1783	2296	size_t size = sizeof(struct ceph_mds_reply_dir_entry);
1784		- int order, num_entries;
	2297	+ unsigned int num_entries;
	2298	+ int order;
1785	2299
1786	2300	spin_lock(&ci->i_ceph_lock);
1787	2301	num_entries = ci->i_files + ci->i_subdirs;
1788	2302	spin_unlock(&ci->i_ceph_lock);
1789		- num_entries = max(num_entries, 1);
	2303	+ num_entries = max(num_entries, 1U);
1790	2304	num_entries = min(num_entries, opt->max_readdir);
1791	2305
1792	2306	order = get_order(size * num_entries);
..	..	@@ -1817,15 +2331,16 @@
1817	2331	struct ceph_mds_request *
1818	2332	ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
1819	2333	{
1820		- struct ceph_mds_request req = kzalloc(sizeof(req), GFP_NOFS);
1821		- struct timespec64 ts;
	2334	+ struct ceph_mds_request *req;
1822	2335
	2336	+ req = kmem_cache_zalloc(ceph_mds_request_cachep, GFP_NOFS);
1823	2337	if (!req)
1824	2338	return ERR_PTR(-ENOMEM);
1825	2339
1826	2340	mutex_init(&req->r_fill_mutex);
1827	2341	req->r_mdsc = mdsc;
1828	2342	req->r_started = jiffies;
	2343	+ req->r_start_latency = ktime_get();
1829	2344	req->r_resend_mds = -1;
1830	2345	INIT_LIST_HEAD(&req->r_unsafe_dir_item);
1831	2346	INIT_LIST_HEAD(&req->r_unsafe_target_item);
..	..	@@ -1837,8 +2352,7 @@
1837	2352	init_completion(&req->r_safe_completion);
1838	2353	INIT_LIST_HEAD(&req->r_unsafe_item);
1839	2354
1840		- ktime_get_coarse_real_ts64(&ts);
1841		- req->r_stamp = timespec64_trunc(ts, mdsc->fsc->sb->s_time_gran);
	2355	+ ktime_get_coarse_real_ts64(&req->r_stamp);
1842	2356
1843	2357	req->r_op = op;
1844	2358	req->r_direct_mode = mode;
..	..	@@ -1873,43 +2387,29 @@
1873	2387	* Encode hidden .snap dirs as a double /, i.e.
1874	2388	* foo/.snap/bar -> foo//bar
1875	2389	*/
1876		-char ceph_mdsc_build_path(struct dentry dentry, int plen, u64 base,
	2390	+char ceph_mdsc_build_path(struct dentry dentry, int plen, u64 pbase,
1877	2391	int stop_on_nosnap)
1878	2392	{
1879	2393	struct dentry *temp;
1880	2394	char *path;
1881		- int len, pos;
	2395	+ int pos;
1882	2396	unsigned seq;
	2397	+ u64 base;
1883	2398
1884	2399	if (!dentry)
1885	2400	return ERR_PTR(-EINVAL);
1886	2401
1887		-retry:
1888		- len = 0;
1889		- seq = read_seqbegin(&rename_lock);
1890		- rcu_read_lock();
1891		- for (temp = dentry; !IS_ROOT(temp);) {
1892		- struct inode *inode = d_inode(temp);
1893		- if (inode && ceph_snap(inode) == CEPH_SNAPDIR)
1894		- len++; /* slash only */
1895		- else if (stop_on_nosnap && inode &&
1896		- ceph_snap(inode) == CEPH_NOSNAP)
1897		- break;
1898		- else
1899		- len += 1 + temp->d_name.len;
1900		- temp = temp->d_parent;
1901		- }
1902		- rcu_read_unlock();
1903		- if (len)
1904		- len--; /* no leading '/' */
1905		-
1906		- path = kmalloc(len+1, GFP_NOFS);
	2402	+ path = __getname();
1907	2403	if (!path)
1908	2404	return ERR_PTR(-ENOMEM);
1909		- pos = len;
1910		- path[pos] = 0; /* trailing null */
	2405	+retry:
	2406	+ pos = PATH_MAX - 1;
	2407	+ path[pos] = '\0';
	2408	+
	2409	+ seq = read_seqbegin(&rename_lock);
1911	2410	rcu_read_lock();
1912		- for (temp = dentry; !IS_ROOT(temp) && pos != 0; ) {
	2411	+ temp = dentry;
	2412	+ for (;;) {
1913	2413	struct inode *inode;
1914	2414
1915	2415	spin_lock(&temp->d_lock);
..	..	@@ -1917,9 +2417,10 @@
1917	2417	if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {
1918	2418	dout("build_path path+%d: %p SNAPDIR\n",
1919	2419	pos, temp);
1920		- } else if (stop_on_nosnap && inode &&
	2420	+ } else if (stop_on_nosnap && inode && dentry != temp &&
1921	2421	ceph_snap(inode) == CEPH_NOSNAP) {
1922	2422	spin_unlock(&temp->d_lock);
	2423	+ pos++; /* get rid of any prepended '/' */
1923	2424	break;
1924	2425	} else {
1925	2426	pos -= temp->d_name.len;
..	..	@@ -1927,83 +2428,58 @@
1927	2428	spin_unlock(&temp->d_lock);
1928	2429	break;
1929	2430	}
1930		- strncpy(path + pos, temp->d_name.name,
1931		- temp->d_name.len);
	2431	+ memcpy(path + pos, temp->d_name.name, temp->d_name.len);
1932	2432	}
1933	2433	spin_unlock(&temp->d_lock);
1934		- if (pos)
1935		- path[--pos] = '/';
1936		- temp = temp->d_parent;
	2434	+ temp = READ_ONCE(temp->d_parent);
	2435	+
	2436	+ /* Are we at the root? */
	2437	+ if (IS_ROOT(temp))
	2438	+ break;
	2439	+
	2440	+ /* Are we out of buffer? */
	2441	+ if (--pos < 0)
	2442	+ break;
	2443	+
	2444	+ path[pos] = '/';
1937	2445	}
	2446	+ base = ceph_ino(d_inode(temp));
1938	2447	rcu_read_unlock();
1939		- if (pos != 0 \|\| read_seqretry(&rename_lock, seq)) {
1940		- pr_err("build_path did not end path lookup where "
1941		- "expected, namelen is %d, pos is %d\n", len, pos);
1942		- /* presumably this is only possible if racing with a
1943		- rename of one of the parent directories (we can not
1944		- lock the dentries above us to prevent this, but
1945		- retrying should be harmless) */
1946		- kfree(path);
	2448	+
	2449	+ if (read_seqretry(&rename_lock, seq))
	2450	+ goto retry;
	2451	+
	2452	+ if (pos < 0) {
	2453	+ /*
	2454	+ * A rename didn't occur, but somehow we didn't end up where
	2455	+ * we thought we would. Throw a warning and try again.
	2456	+ */
	2457	+ pr_warn("build_path did not end path lookup where "
	2458	+ "expected, pos is %d\n", pos);
1947	2459	goto retry;
1948	2460	}
1949	2461
1950		- *base = ceph_ino(d_inode(temp));
1951		- *plen = len;
	2462	+ *pbase = base;
	2463	+ *plen = PATH_MAX - 1 - pos;
1952	2464	dout("build_path on %p %d built %llx '%.*s'\n",
1953		- dentry, d_count(dentry), *base, len, path);
1954		- return path;
1955		-}
1956		-
1957		-/* Duplicate the dentry->d_name.name safely */
1958		-static int clone_dentry_name(struct dentry dentry, const char *ppath,
1959		- int *ppathlen)
1960		-{
1961		- u32 len;
1962		- char *name;
1963		-
1964		-retry:
1965		- len = READ_ONCE(dentry->d_name.len);
1966		- name = kmalloc(len + 1, GFP_NOFS);
1967		- if (!name)
1968		- return -ENOMEM;
1969		-
1970		- spin_lock(&dentry->d_lock);
1971		- if (dentry->d_name.len != len) {
1972		- spin_unlock(&dentry->d_lock);
1973		- kfree(name);
1974		- goto retry;
1975		- }
1976		- memcpy(name, dentry->d_name.name, len);
1977		- spin_unlock(&dentry->d_lock);
1978		-
1979		- name[len] = '\0';
1980		- *ppath = name;
1981		- *ppathlen = len;
1982		- return 0;
	2465	+ dentry, d_count(dentry), base, *plen, path + pos);
	2466	+ return path + pos;
1983	2467	}
1984	2468
1985	2469	static int build_dentry_path(struct dentry dentry, struct inode dir,
1986	2470	const char *ppath, int ppathlen, u64 *pino,
1987	2471	bool *pfreepath, bool parent_locked)
1988	2472	{
1989		- int ret;
1990	2473	char *path;
1991	2474
1992	2475	rcu_read_lock();
1993	2476	if (!dir)
1994	2477	dir = d_inode_rcu(dentry->d_parent);
1995		- if (dir && ceph_snap(dir) == CEPH_NOSNAP) {
	2478	+ if (dir && parent_locked && ceph_snap(dir) == CEPH_NOSNAP) {
1996	2479	*pino = ceph_ino(dir);
1997	2480	rcu_read_unlock();
1998		- if (parent_locked) {
1999		- *ppath = dentry->d_name.name;
2000		- *ppathlen = dentry->d_name.len;
2001		- } else {
2002		- ret = clone_dentry_name(dentry, ppath, ppathlen);
2003		- if (ret)
2004		- return ret;
2005		- *pfreepath = true;
2006		- }
	2481	+ *ppath = dentry->d_name.name;
	2482	+ *ppathlen = dentry->d_name.len;
2007	2483	return 0;
2008	2484	}
2009	2485	rcu_read_unlock();
..	..	@@ -2115,11 +2591,11 @@
2115	2591	(!!req->r_inode_drop + !!req->r_dentry_drop +
2116	2592	!!req->r_old_inode_drop + !!req->r_old_dentry_drop);
2117	2593	if (req->r_dentry_drop)
2118		- len += req->r_dentry->d_name.len;
	2594	+ len += pathlen1;
2119	2595	if (req->r_old_dentry_drop)
2120		- len += req->r_old_dentry->d_name.len;
	2596	+ len += pathlen2;
2121	2597
2122		- msg = ceph_msg_new(CEPH_MSG_CLIENT_REQUEST, len, GFP_NOFS, false);
	2598	+ msg = ceph_msg_new2(CEPH_MSG_CLIENT_REQUEST, len, 1, GFP_NOFS, false);
2123	2599	if (!msg) {
2124	2600	msg = ERR_PTR(-ENOMEM);
2125	2601	goto out_free2;
..	..	@@ -2136,6 +2612,7 @@
2136	2612	head->op = cpu_to_le32(req->r_op);
2137	2613	head->caller_uid = cpu_to_le32(from_kuid(&init_user_ns, req->r_uid));
2138	2614	head->caller_gid = cpu_to_le32(from_kgid(&init_user_ns, req->r_gid));
	2615	+ head->ino = cpu_to_le64(req->r_deleg_ino);
2139	2616	head->args = req->r_args;
2140	2617
2141	2618	ceph_encode_filepath(&p, end, ino1, path1);
..	..	@@ -2149,7 +2626,8 @@
2149	2626	if (req->r_inode_drop)
2150	2627	releases += ceph_encode_inode_release(&p,
2151	2628	req->r_inode ? req->r_inode : d_inode(req->r_dentry),
2152		- mds, req->r_inode_drop, req->r_inode_unless, 0);
	2629	+ mds, req->r_inode_drop, req->r_inode_unless,
	2630	+ req->r_op == CEPH_MDS_OP_READDIR);
2153	2631	if (req->r_dentry_drop)
2154	2632	releases += ceph_encode_dentry_release(&p, req->r_dentry,
2155	2633	req->r_parent, mds, req->r_dentry_drop,
..	..	@@ -2178,13 +2656,17 @@
2178	2656	ceph_encode_copy(&p, &ts, sizeof(ts));
2179	2657	}
2180	2658
2181		- BUG_ON(p > end);
	2659	+ if (WARN_ON_ONCE(p > end)) {
	2660	+ ceph_msg_put(msg);
	2661	+ msg = ERR_PTR(-ERANGE);
	2662	+ goto out_free2;
	2663	+ }
	2664	+
2182	2665	msg->front.iov_len = p - msg->front.iov_base;
2183	2666	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2184	2667
2185	2668	if (req->r_pagelist) {
2186	2669	struct ceph_pagelist *pagelist = req->r_pagelist;
2187		- refcount_inc(&pagelist->refcnt);
2188	2670	ceph_msg_data_add_pagelist(msg, pagelist);
2189	2671	msg->hdr.data_len = cpu_to_le32(pagelist->length);
2190	2672	} else {
..	..	@@ -2195,10 +2677,10 @@
2195	2677
2196	2678	out_free2:
2197	2679	if (freepath2)
2198		- kfree((char *)path2);
	2680	+ ceph_mdsc_free_path((char *)path2, pathlen2);
2199	2681	out_free1:
2200	2682	if (freepath1)
2201		- kfree((char *)path1);
	2683	+ ceph_mdsc_free_path((char *)path1, pathlen1);
2202	2684	out:
2203	2685	return msg;
2204	2686	}
..	..	@@ -2210,10 +2692,11 @@
2210	2692	static void complete_request(struct ceph_mds_client *mdsc,
2211	2693	struct ceph_mds_request *req)
2212	2694	{
	2695	+ req->r_end_latency = ktime_get();
	2696	+
2213	2697	if (req->r_callback)
2214	2698	req->r_callback(mdsc, req);
2215		- else
2216		- complete_all(&req->r_completion);
	2699	+ complete_all(&req->r_completion);
2217	2700	}
2218	2701
2219	2702	/*
..	..	@@ -2291,15 +2774,36 @@
2291	2774	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
2292	2775	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
2293	2776	flags \|= CEPH_MDS_FLAG_REPLAY;
	2777	+ if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags))
	2778	+ flags \|= CEPH_MDS_FLAG_ASYNC;
2294	2779	if (req->r_parent)
2295	2780	flags \|= CEPH_MDS_FLAG_WANT_DENTRY;
2296	2781	rhead->flags = cpu_to_le32(flags);
2297	2782	rhead->num_fwd = req->r_num_fwd;
2298	2783	rhead->num_retry = req->r_attempts - 1;
2299		- rhead->ino = 0;
2300	2784
2301	2785	dout(" r_parent = %p\n", req->r_parent);
2302	2786	return 0;
	2787	+}
	2788	+
	2789	+/*
	2790	+ * called under mdsc->mutex
	2791	+ */
	2792	+static int __send_request(struct ceph_mds_client *mdsc,
	2793	+ struct ceph_mds_session *session,
	2794	+ struct ceph_mds_request *req,
	2795	+ bool drop_cap_releases)
	2796	+{
	2797	+ int err;
	2798	+
	2799	+ err = __prepare_send_request(mdsc, req, session->s_mds,
	2800	+ drop_cap_releases);
	2801	+ if (!err) {
	2802	+ ceph_msg_get(req->r_request);
	2803	+ ceph_con_send(&session->s_con, req->r_request);
	2804	+ }
	2805	+
	2806	+ return err;
2303	2807	}
2304	2808
2305	2809	/*
..	..	@@ -2311,6 +2815,7 @@
2311	2815	struct ceph_mds_session *session = NULL;
2312	2816	int mds = -1;
2313	2817	int err = 0;
	2818	+ bool random;
2314	2819
2315	2820	if (req->r_err \|\| test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
2316	2821	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
..	..	@@ -2321,7 +2826,7 @@
2321	2826	if (req->r_timeout &&
2322	2827	time_after_eq(jiffies, req->r_started + req->r_timeout)) {
2323	2828	dout("do_request timed out\n");
2324		- err = -EIO;
	2829	+ err = -ETIMEDOUT;
2325	2830	goto finish;
2326	2831	}
2327	2832	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
..	..	@@ -2350,9 +2855,13 @@
2350	2855
2351	2856	put_request_session(req);
2352	2857
2353		- mds = __choose_mds(mdsc, req);
	2858	+ mds = __choose_mds(mdsc, req, &random);
2354	2859	if (mds < 0 \|\|
2355	2860	ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {
	2861	+ if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {
	2862	+ err = -EJUKEBOX;
	2863	+ goto finish;
	2864	+ }
2356	2865	dout("do_request no mds or not active, waiting for map\n");
2357	2866	list_add(&req->r_wait, &mdsc->waiting_for_map);
2358	2867	return;
..	..	@@ -2367,7 +2876,7 @@
2367	2876	goto finish;
2368	2877	}
2369	2878	}
2370		- req->r_session = get_session(session);
	2879	+ req->r_session = ceph_get_mds_session(session);
2371	2880
2372	2881	dout("do_request mds%d session %p state %s\n", mds, session,
2373	2882	ceph_session_state_name(session->s_state));
..	..	@@ -2377,9 +2886,24 @@
2377	2886	err = -EACCES;
2378	2887	goto out_session;
2379	2888	}
	2889	+ /*
	2890	+ * We cannot queue async requests since the caps and delegated
	2891	+ * inodes are bound to the session. Just return -EJUKEBOX and
	2892	+ * let the caller retry a sync request in that case.
	2893	+ */
	2894	+ if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {
	2895	+ err = -EJUKEBOX;
	2896	+ goto out_session;
	2897	+ }
2380	2898	if (session->s_state == CEPH_MDS_SESSION_NEW \|\|
2381		- session->s_state == CEPH_MDS_SESSION_CLOSING)
2382		- __open_session(mdsc, session);
	2899	+ session->s_state == CEPH_MDS_SESSION_CLOSING) {
	2900	+ err = __open_session(mdsc, session);
	2901	+ if (err)
	2902	+ goto out_session;
	2903	+ /* retry the same mds later */
	2904	+ if (random)
	2905	+ req->r_resend_mds = mds;
	2906	+ }
2383	2907	list_add(&req->r_wait, &session->s_waiting);
2384	2908	goto out_session;
2385	2909	}
..	..	@@ -2390,11 +2914,7 @@
2390	2914	if (req->r_request_started == 0) /* note request start time */
2391	2915	req->r_request_started = jiffies;
2392	2916
2393		- err = __prepare_send_request(mdsc, req, mds, false);
2394		- if (!err) {
2395		- ceph_msg_get(req->r_request);
2396		- ceph_con_send(&session->s_con, req->r_request);
2397		- }
	2917	+ err = __send_request(mdsc, session, req, false);
2398	2918
2399	2919	out_session:
2400	2920	ceph_put_mds_session(session);
..	..	@@ -2454,49 +2974,61 @@
2454	2974	}
2455	2975	}
2456	2976
2457		-void ceph_mdsc_submit_request(struct ceph_mds_client *mdsc,
	2977	+int ceph_mdsc_submit_request(struct ceph_mds_client mdsc, struct inode dir,
2458	2978	struct ceph_mds_request *req)
2459	2979	{
2460		- dout("submit_request on %p\n", req);
2461		- mutex_lock(&mdsc->mutex);
2462		- __register_request(mdsc, req, NULL);
2463		- __do_request(mdsc, req);
2464		- mutex_unlock(&mdsc->mutex);
2465		-}
2466		-
2467		-/*
2468		- * Synchrously perform an mds request. Take care of all of the
2469		- * session setup, forwarding, retry details.
2470		- */
2471		-int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
2472		- struct inode *dir,
2473		- struct ceph_mds_request *req)
2474		-{
2475		- int err;
2476		-
2477		- dout("do_request on %p\n", req);
	2980	+ int err = 0;
2478	2981
2479	2982	/* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
2480	2983	if (req->r_inode)
2481	2984	ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
2482		- if (req->r_parent)
2483		- ceph_get_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
	2985	+ if (req->r_parent) {
	2986	+ struct ceph_inode_info *ci = ceph_inode(req->r_parent);
	2987	+ int fmode = (req->r_op & CEPH_MDS_OP_WRITE) ?
	2988	+ CEPH_FILE_MODE_WR : CEPH_FILE_MODE_RD;
	2989	+ spin_lock(&ci->i_ceph_lock);
	2990	+ ceph_take_cap_refs(ci, CEPH_CAP_PIN, false);
	2991	+ __ceph_touch_fmode(ci, mdsc, fmode);
	2992	+ spin_unlock(&ci->i_ceph_lock);
	2993	+ ihold(req->r_parent);
	2994	+ }
2484	2995	if (req->r_old_dentry_dir)
2485	2996	ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
2486	2997	CEPH_CAP_PIN);
2487	2998
2488		- /* issue */
	2999	+ if (req->r_inode) {
	3000	+ err = ceph_wait_on_async_create(req->r_inode);
	3001	+ if (err) {
	3002	+ dout("%s: wait for async create returned: %d\n",
	3003	+ __func__, err);
	3004	+ return err;
	3005	+ }
	3006	+ }
	3007	+
	3008	+ if (!err && req->r_old_inode) {
	3009	+ err = ceph_wait_on_async_create(req->r_old_inode);
	3010	+ if (err) {
	3011	+ dout("%s: wait for async create returned: %d\n",
	3012	+ __func__, err);
	3013	+ return err;
	3014	+ }
	3015	+ }
	3016	+
	3017	+ dout("submit_request on %p for inode %p\n", req, dir);
2489	3018	mutex_lock(&mdsc->mutex);
2490	3019	__register_request(mdsc, req, dir);
2491	3020	__do_request(mdsc, req);
	3021	+ err = req->r_err;
	3022	+ mutex_unlock(&mdsc->mutex);
	3023	+ return err;
	3024	+}
2492	3025
2493		- if (req->r_err) {
2494		- err = req->r_err;
2495		- goto out;
2496		- }
	3026	+static int ceph_mdsc_wait_request(struct ceph_mds_client *mdsc,
	3027	+ struct ceph_mds_request *req)
	3028	+{
	3029	+ int err;
2497	3030
2498	3031	/* wait */
2499		- mutex_unlock(&mdsc->mutex);
2500	3032	dout("do_request waiting\n");
2501	3033	if (!req->r_timeout && req->r_wait_for_completion) {
2502	3034	err = req->r_wait_for_completion(mdsc, req);
..	..	@@ -2507,7 +3039,7 @@
2507	3039	if (timeleft > 0)
2508	3040	err = 0;
2509	3041	else if (!timeleft)
2510		- err = -EIO; /* timed out */
	3042	+ err = -ETIMEDOUT; /* timed out */
2511	3043	else
2512	3044	err = timeleft; /* killed */
2513	3045	}
..	..	@@ -2537,8 +3069,26 @@
2537	3069	err = req->r_err;
2538	3070	}
2539	3071
2540		-out:
2541	3072	mutex_unlock(&mdsc->mutex);
	3073	+ return err;
	3074	+}
	3075	+
	3076	+/*
	3077	+ * Synchrously perform an mds request. Take care of all of the
	3078	+ * session setup, forwarding, retry details.
	3079	+ */
	3080	+int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
	3081	+ struct inode *dir,
	3082	+ struct ceph_mds_request *req)
	3083	+{
	3084	+ int err;
	3085	+
	3086	+ dout("do_request on %p\n", req);
	3087	+
	3088	+ /* issue */
	3089	+ err = ceph_mdsc_submit_request(mdsc, dir, req);
	3090	+ if (!err)
	3091	+ err = ceph_mdsc_wait_request(mdsc, req);
2542	3092	dout("do_request %p done, result %d\n", req, err);
2543	3093	return err;
2544	3094	}
..	..	@@ -2641,7 +3191,7 @@
2641	3191	mutex_unlock(&mdsc->mutex);
2642	3192	goto out;
2643	3193	} else {
2644		- int mds = __choose_mds(mdsc, req);
	3194	+ int mds = __choose_mds(mdsc, req, NULL);
2645	3195	if (mds >= 0 && mds != req->r_session->s_mds) {
2646	3196	dout("but auth changed, so resending\n");
2647	3197	__do_request(mdsc, req);
..	..	@@ -2657,6 +3207,10 @@
2657	3207	set_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags);
2658	3208	__unregister_request(mdsc, req);
2659	3209
	3210	+ /* last request during umount? */
	3211	+ if (mdsc->stopping && !__get_oldest_req(mdsc))
	3212	+ complete_all(&mdsc->safe_umount_waiters);
	3213	+
2660	3214	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
2661	3215	/*
2662	3216	* We already handled the unsafe response, now do the
..	..	@@ -2667,28 +3221,20 @@
2667	3221	*/
2668	3222	dout("got safe reply %llu, mds%d\n", tid, mds);
2669	3223
2670		- /* last unsafe request during umount? */
2671		- if (mdsc->stopping && !__get_oldest_req(mdsc))
2672		- complete_all(&mdsc->safe_umount_waiters);
2673	3224	mutex_unlock(&mdsc->mutex);
2674	3225	goto out;
2675	3226	}
2676	3227	} else {
2677	3228	set_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags);
2678	3229	list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
2679		- if (req->r_unsafe_dir) {
2680		- struct ceph_inode_info *ci =
2681		- ceph_inode(req->r_unsafe_dir);
2682		- spin_lock(&ci->i_unsafe_lock);
2683		- list_add_tail(&req->r_unsafe_dir_item,
2684		- &ci->i_unsafe_dirops);
2685		- spin_unlock(&ci->i_unsafe_lock);
2686		- }
2687	3230	}
2688	3231
2689	3232	dout("handle_reply tid %lld result %d\n", tid, result);
2690	3233	rinfo = &req->r_reply_info;
2691		- err = parse_reply_info(msg, rinfo, session->s_con.peer_features);
	3234	+ if (test_bit(CEPHFS_FEATURE_REPLY_ENCODING, &session->s_features))
	3235	+ err = parse_reply_info(session, msg, rinfo, (u64)-1);
	3236	+ else
	3237	+ err = parse_reply_info(session, msg, rinfo, session->s_con.peer_features);
2692	3238	mutex_unlock(&mdsc->mutex);
2693	3239
2694	3240	mutex_lock(&session->s_mutex);
..	..	@@ -2719,7 +3265,6 @@
2719	3265	if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR \|\|
2720	3266	req->r_op == CEPH_MDS_OP_LSSNAP))
2721	3267	ceph_readdir_prepopulate(req, req->r_session);
2722		- ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
2723	3268	}
2724	3269	current->journal_info = NULL;
2725	3270	mutex_unlock(&req->r_fill_mutex);
..	..	@@ -2728,12 +3273,18 @@
2728	3273	if (realm)
2729	3274	ceph_put_snap_realm(mdsc, realm);
2730	3275
2731		- if (err == 0 && req->r_target_inode &&
2732		- test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
2733		- struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
2734		- spin_lock(&ci->i_unsafe_lock);
2735		- list_add_tail(&req->r_unsafe_target_item, &ci->i_unsafe_iops);
2736		- spin_unlock(&ci->i_unsafe_lock);
	3276	+ if (err == 0) {
	3277	+ if (req->r_target_inode &&
	3278	+ test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
	3279	+ struct ceph_inode_info *ci =
	3280	+ ceph_inode(req->r_target_inode);
	3281	+ spin_lock(&ci->i_unsafe_lock);
	3282	+ list_add_tail(&req->r_unsafe_target_item,
	3283	+ &ci->i_unsafe_iops);
	3284	+ spin_unlock(&ci->i_unsafe_lock);
	3285	+ }
	3286	+
	3287	+ ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
2737	3288	}
2738	3289	out_err:
2739	3290	mutex_lock(&mdsc->mutex);
..	..	@@ -2753,6 +3304,9 @@
2753	3304
2754	3305	/* kick calling process */
2755	3306	complete_request(mdsc, req);
	3307	+
	3308	+ ceph_update_metadata_latency(&mdsc->metric, req->r_start_latency,
	3309	+ req->r_end_latency, err);
2756	3310	out:
2757	3311	ceph_mdsc_put_request(req);
2758	3312	return;
..	..	@@ -2812,6 +3366,34 @@
2812	3366	pr_err("mdsc_handle_forward decode error err=%d\n", err);
2813	3367	}
2814	3368
	3369	+static int __decode_session_metadata(void *p, void end,
	3370	+ bool *blocklisted)
	3371	+{
	3372	+ /* map<string,string> */
	3373	+ u32 n;
	3374	+ bool err_str;
	3375	+ ceph_decode_32_safe(p, end, n, bad);
	3376	+ while (n-- > 0) {
	3377	+ u32 len;
	3378	+ ceph_decode_32_safe(p, end, len, bad);
	3379	+ ceph_decode_need(p, end, len, bad);
	3380	+ err_str = !strncmp(*p, "error_string", len);
	3381	+ *p += len;
	3382	+ ceph_decode_32_safe(p, end, len, bad);
	3383	+ ceph_decode_need(p, end, len, bad);
	3384	+ /*
	3385	+ * Match "blocklisted (blacklisted)" from newer MDSes,
	3386	+ * or "blacklisted" from older MDSes.
	3387	+ */
	3388	+ if (err_str && strnstr(*p, "blacklisted", len))
	3389	+ *blocklisted = true;
	3390	+ *p += len;
	3391	+ }
	3392	+ return 0;
	3393	+bad:
	3394	+ return -1;
	3395	+}
	3396	+
2815	3397	/*
2816	3398	* handle a mds session control message
2817	3399	*/
..	..	@@ -2819,21 +3401,40 @@
2819	3401	struct ceph_msg *msg)
2820	3402	{
2821	3403	struct ceph_mds_client *mdsc = session->s_mdsc;
2822		- u32 op;
2823		- u64 seq;
2824	3404	int mds = session->s_mds;
2825		- struct ceph_mds_session_head *h = msg->front.iov_base;
	3405	+ int msg_version = le16_to_cpu(msg->hdr.version);
	3406	+ void *p = msg->front.iov_base;
	3407	+ void *end = p + msg->front.iov_len;
	3408	+ struct ceph_mds_session_head *h;
	3409	+ u32 op;
	3410	+ u64 seq, features = 0;
2826	3411	int wake = 0;
	3412	+ bool blocklisted = false;
2827	3413
2828	3414	/* decode */
2829		- if (msg->front.iov_len < sizeof(*h))
2830		- goto bad;
	3415	+ ceph_decode_need(&p, end, sizeof(*h), bad);
	3416	+ h = p;
	3417	+ p += sizeof(*h);
	3418	+
2831	3419	op = le32_to_cpu(h->op);
2832	3420	seq = le64_to_cpu(h->seq);
2833	3421
	3422	+ if (msg_version >= 3) {
	3423	+ u32 len;
	3424	+ /* version >= 2, metadata */
	3425	+ if (__decode_session_metadata(&p, end, &blocklisted) < 0)
	3426	+ goto bad;
	3427	+ /* version >= 3, feature bits */
	3428	+ ceph_decode_32_safe(&p, end, len, bad);
	3429	+ if (len) {
	3430	+ ceph_decode_64_safe(&p, end, features, bad);
	3431	+ p += len - sizeof(features);
	3432	+ }
	3433	+ }
	3434	+
2834	3435	mutex_lock(&mdsc->mutex);
2835	3436	if (op == CEPH_SESSION_CLOSE) {
2836		- get_session(session);
	3437	+ ceph_get_mds_session(session);
2837	3438	__unregister_session(mdsc, session);
2838	3439	}
2839	3440	/* FIXME: this ttl calculation is generous */
..	..	@@ -2856,7 +3457,10 @@
2856	3457	if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
2857	3458	pr_info("mds%d reconnect success\n", session->s_mds);
2858	3459	session->s_state = CEPH_MDS_SESSION_OPEN;
	3460	+ session->s_features = features;
2859	3461	renewed_caps(mdsc, session, 0);
	3462	+ if (test_bit(CEPHFS_FEATURE_METRIC_COLLECT, &session->s_features))
	3463	+ metric_schedule_delayed(&mdsc->metric);
2860	3464	wake = 1;
2861	3465	if (mdsc->stopping)
2862	3466	__close_session(mdsc, session);
..	..	@@ -2870,6 +3474,7 @@
2870	3474	case CEPH_SESSION_CLOSE:
2871	3475	if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
2872	3476	pr_info("mds%d reconnect denied\n", session->s_mds);
	3477	+ session->s_state = CEPH_MDS_SESSION_CLOSED;
2873	3478	cleanup_session_requests(mdsc, session);
2874	3479	remove_session_caps(session);
2875	3480	wake = 2; /* for good measure */
..	..	@@ -2899,7 +3504,7 @@
2899	3504	spin_lock(&session->s_cap_lock);
2900	3505	session->s_readonly = true;
2901	3506	spin_unlock(&session->s_cap_lock);
2902		- wake_up_session_caps(session, 0);
	3507	+ wake_up_session_caps(session, FORCE_RO);
2903	3508	break;
2904	3509
2905	3510	case CEPH_SESSION_REJECT:
..	..	@@ -2908,6 +3513,8 @@
2908	3513	session->s_state = CEPH_MDS_SESSION_REJECTED;
2909	3514	cleanup_session_requests(mdsc, session);
2910	3515	remove_session_caps(session);
	3516	+ if (blocklisted)
	3517	+ mdsc->fsc->blocklisted = true;
2911	3518	wake = 2; /* for good measure */
2912	3519	break;
2913	3520
..	..	@@ -2935,6 +3542,28 @@
2935	3542	return;
2936	3543	}
2937	3544
	3545	+void ceph_mdsc_release_dir_caps(struct ceph_mds_request *req)
	3546	+{
	3547	+ int dcaps;
	3548	+
	3549	+ dcaps = xchg(&req->r_dir_caps, 0);
	3550	+ if (dcaps) {
	3551	+ dout("releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));
	3552	+ ceph_put_cap_refs(ceph_inode(req->r_parent), dcaps);
	3553	+ }
	3554	+}
	3555	+
	3556	+void ceph_mdsc_release_dir_caps_no_check(struct ceph_mds_request *req)
	3557	+{
	3558	+ int dcaps;
	3559	+
	3560	+ dcaps = xchg(&req->r_dir_caps, 0);
	3561	+ if (dcaps) {
	3562	+ dout("releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));
	3563	+ ceph_put_cap_refs_no_check_caps(ceph_inode(req->r_parent),
	3564	+ dcaps);
	3565	+ }
	3566	+}
2938	3567
2939	3568	/*
2940	3569	* called under session->mutex.
..	..	@@ -2944,18 +3573,12 @@
2944	3573	{
2945	3574	struct ceph_mds_request req, nreq;
2946	3575	struct rb_node *p;
2947		- int err;
2948	3576
2949	3577	dout("replay_unsafe_requests mds%d\n", session->s_mds);
2950	3578
2951	3579	mutex_lock(&mdsc->mutex);
2952		- list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item) {
2953		- err = __prepare_send_request(mdsc, req, session->s_mds, true);
2954		- if (!err) {
2955		- ceph_msg_get(req->r_request);
2956		- ceph_con_send(&session->s_con, req->r_request);
2957		- }
2958		- }
	3580	+ list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item)
	3581	+ __send_request(mdsc, session, req, true);
2959	3582
2960	3583	/*
2961	3584	* also re-send old requests when MDS enters reconnect stage. So that MDS
..	..	@@ -2969,23 +3592,131 @@
2969	3592	continue;
2970	3593	if (req->r_attempts == 0)
2971	3594	continue; /* only old requests */
2972		- if (req->r_session &&
2973		- req->r_session->s_mds == session->s_mds) {
2974		- err = __prepare_send_request(mdsc, req,
2975		- session->s_mds, true);
2976		- if (!err) {
2977		- ceph_msg_get(req->r_request);
2978		- ceph_con_send(&session->s_con, req->r_request);
2979		- }
2980		- }
	3595	+ if (!req->r_session)
	3596	+ continue;
	3597	+ if (req->r_session->s_mds != session->s_mds)
	3598	+ continue;
	3599	+
	3600	+ ceph_mdsc_release_dir_caps_no_check(req);
	3601	+
	3602	+ __send_request(mdsc, session, req, true);
2981	3603	}
2982	3604	mutex_unlock(&mdsc->mutex);
	3605	+}
	3606	+
	3607	+static int send_reconnect_partial(struct ceph_reconnect_state *recon_state)
	3608	+{
	3609	+ struct ceph_msg *reply;
	3610	+ struct ceph_pagelist *_pagelist;
	3611	+ struct page *page;
	3612	+ __le32 *addr;
	3613	+ int err = -ENOMEM;
	3614	+
	3615	+ if (!recon_state->allow_multi)
	3616	+ return -ENOSPC;
	3617	+
	3618	+ /* can't handle message that contains both caps and realm */
	3619	+ BUG_ON(!recon_state->nr_caps == !recon_state->nr_realms);
	3620	+
	3621	+ /* pre-allocate new pagelist */
	3622	+ _pagelist = ceph_pagelist_alloc(GFP_NOFS);
	3623	+ if (!_pagelist)
	3624	+ return -ENOMEM;
	3625	+
	3626	+ reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);
	3627	+ if (!reply)
	3628	+ goto fail_msg;
	3629	+
	3630	+ /* placeholder for nr_caps */
	3631	+ err = ceph_pagelist_encode_32(_pagelist, 0);
	3632	+ if (err < 0)
	3633	+ goto fail;
	3634	+
	3635	+ if (recon_state->nr_caps) {
	3636	+ /* currently encoding caps */
	3637	+ err = ceph_pagelist_encode_32(recon_state->pagelist, 0);
	3638	+ if (err)
	3639	+ goto fail;
	3640	+ } else {
	3641	+ /* placeholder for nr_realms (currently encoding relams) */
	3642	+ err = ceph_pagelist_encode_32(_pagelist, 0);
	3643	+ if (err < 0)
	3644	+ goto fail;
	3645	+ }
	3646	+
	3647	+ err = ceph_pagelist_encode_8(recon_state->pagelist, 1);
	3648	+ if (err)
	3649	+ goto fail;
	3650	+
	3651	+ page = list_first_entry(&recon_state->pagelist->head, struct page, lru);
	3652	+ addr = kmap_atomic(page);
	3653	+ if (recon_state->nr_caps) {
	3654	+ /* currently encoding caps */
	3655	+ *addr = cpu_to_le32(recon_state->nr_caps);
	3656	+ } else {
	3657	+ /* currently encoding relams */
	3658	+ *(addr + 1) = cpu_to_le32(recon_state->nr_realms);
	3659	+ }
	3660	+ kunmap_atomic(addr);
	3661	+
	3662	+ reply->hdr.version = cpu_to_le16(5);
	3663	+ reply->hdr.compat_version = cpu_to_le16(4);
	3664	+
	3665	+ reply->hdr.data_len = cpu_to_le32(recon_state->pagelist->length);
	3666	+ ceph_msg_data_add_pagelist(reply, recon_state->pagelist);
	3667	+
	3668	+ ceph_con_send(&recon_state->session->s_con, reply);
	3669	+ ceph_pagelist_release(recon_state->pagelist);
	3670	+
	3671	+ recon_state->pagelist = _pagelist;
	3672	+ recon_state->nr_caps = 0;
	3673	+ recon_state->nr_realms = 0;
	3674	+ recon_state->msg_version = 5;
	3675	+ return 0;
	3676	+fail:
	3677	+ ceph_msg_put(reply);
	3678	+fail_msg:
	3679	+ ceph_pagelist_release(_pagelist);
	3680	+ return err;
	3681	+}
	3682	+
	3683	+static struct dentry* d_find_primary(struct inode *inode)
	3684	+{
	3685	+ struct dentry alias, dn = NULL;
	3686	+
	3687	+ if (hlist_empty(&inode->i_dentry))
	3688	+ return NULL;
	3689	+
	3690	+ spin_lock(&inode->i_lock);
	3691	+ if (hlist_empty(&inode->i_dentry))
	3692	+ goto out_unlock;
	3693	+
	3694	+ if (S_ISDIR(inode->i_mode)) {
	3695	+ alias = hlist_entry(inode->i_dentry.first, struct dentry, d_u.d_alias);
	3696	+ if (!IS_ROOT(alias))
	3697	+ dn = dget(alias);
	3698	+ goto out_unlock;
	3699	+ }
	3700	+
	3701	+ hlist_for_each_entry(alias, &inode->i_dentry, d_u.d_alias) {
	3702	+ spin_lock(&alias->d_lock);
	3703	+ if (!d_unhashed(alias) &&
	3704	+ (ceph_dentry(alias)->flags & CEPH_DENTRY_PRIMARY_LINK)) {
	3705	+ dn = dget_dlock(alias);
	3706	+ }
	3707	+ spin_unlock(&alias->d_lock);
	3708	+ if (dn)
	3709	+ break;
	3710	+ }
	3711	+out_unlock:
	3712	+ spin_unlock(&inode->i_lock);
	3713	+ return dn;
2983	3714	}
2984	3715
2985	3716	/*
2986	3717	* Encode information about a cap for a reconnect with the MDS.
2987	3718	*/
2988		-static int encode_caps_cb(struct inode inode, struct ceph_cap cap,
	3719	+static int reconnect_caps_cb(struct inode inode, struct ceph_cap cap,
2989	3720	void *arg)
2990	3721	{
2991	3722	union {
..	..	@@ -2995,29 +3726,28 @@
2995	3726	struct ceph_inode_info *ci = cap->ci;
2996	3727	struct ceph_reconnect_state *recon_state = arg;
2997	3728	struct ceph_pagelist *pagelist = recon_state->pagelist;
	3729	+ struct dentry *dentry;
2998	3730	char *path;
2999		- int pathlen, err;
	3731	+ int pathlen = 0, err;
3000	3732	u64 pathbase;
3001	3733	u64 snap_follows;
3002		- struct dentry *dentry;
3003	3734
3004	3735	dout(" adding %p ino %llx.%llx cap %p %lld %s\n",
3005	3736	inode, ceph_vinop(inode), cap, cap->cap_id,
3006	3737	ceph_cap_string(cap->issued));
3007		- err = ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
3008		- if (err)
3009		- return err;
3010	3738
3011		- dentry = d_find_alias(inode);
	3739	+ dentry = d_find_primary(inode);
3012	3740	if (dentry) {
3013		- path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase, 0);
	3741	+ /* set pathbase to parent dir when msg_version >= 2 */
	3742	+ path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase,
	3743	+ recon_state->msg_version >= 2);
	3744	+ dput(dentry);
3014	3745	if (IS_ERR(path)) {
3015	3746	err = PTR_ERR(path);
3016		- goto out_dput;
	3747	+ goto out_err;
3017	3748	}
3018	3749	} else {
3019	3750	path = NULL;
3020		- pathlen = 0;
3021	3751	pathbase = 0;
3022	3752	}
3023	3753
..	..	@@ -3026,6 +3756,15 @@
3026	3756	cap->issue_seq = 0; /* and issue_seq */
3027	3757	cap->mseq = 0; /* and migrate_seq */
3028	3758	cap->cap_gen = cap->session->s_cap_gen;
	3759	+
	3760	+ /* These are lost when the session goes away */
	3761	+ if (S_ISDIR(inode->i_mode)) {
	3762	+ if (cap->issued & CEPH_CAP_DIR_CREATE) {
	3763	+ ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns));
	3764	+ memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));
	3765	+ }
	3766	+ cap->issued &= ~CEPH_CAP_ANY_DIR_OPS;
	3767	+ }
3029	3768
3030	3769	if (recon_state->msg_version >= 2) {
3031	3770	rec.v2.cap_id = cpu_to_le64(cap->cap_id);
..	..	@@ -3059,7 +3798,7 @@
3059	3798	if (recon_state->msg_version >= 2) {
3060	3799	int num_fcntl_locks, num_flock_locks;
3061	3800	struct ceph_filelock *flocks = NULL;
3062		- size_t struct_len, total_len = 0;
	3801	+ size_t struct_len, total_len = sizeof(u64);
3063	3802	u8 struct_v = 0;
3064	3803
3065	3804	encode_again:
..	..	@@ -3075,7 +3814,7 @@
3075	3814	GFP_NOFS);
3076	3815	if (!flocks) {
3077	3816	err = -ENOMEM;
3078		- goto out_free;
	3817	+ goto out_err;
3079	3818	}
3080	3819	err = ceph_encode_locks_to_buffer(inode, flocks,
3081	3820	num_fcntl_locks,
..	..	@@ -3085,7 +3824,7 @@
3085	3824	flocks = NULL;
3086	3825	if (err == -ENOSPC)
3087	3826	goto encode_again;
3088		- goto out_free;
	3827	+ goto out_err;
3089	3828	}
3090	3829	} else {
3091	3830	kfree(flocks);
..	..	@@ -3094,7 +3833,7 @@
3094	3833
3095	3834	if (recon_state->msg_version >= 3) {
3096	3835	/* version, compat_version and struct_len */
3097		- total_len = 2 * sizeof(u8) + sizeof(u32);
	3836	+ total_len += 2 * sizeof(u8) + sizeof(u32);
3098	3837	struct_v = 2;
3099	3838	}
3100	3839	/*
..	..	@@ -3105,44 +3844,113 @@
3105	3844	sizeof(struct ceph_filelock);
3106	3845	rec.v2.flock_len = cpu_to_le32(struct_len);
3107	3846
3108		- struct_len += sizeof(rec.v2);
3109		- struct_len += sizeof(u32) + pathlen;
	3847	+ struct_len += sizeof(u32) + pathlen + sizeof(rec.v2);
3110	3848
3111	3849	if (struct_v >= 2)
3112	3850	struct_len += sizeof(u64); /* snap_follows */
3113	3851
3114	3852	total_len += struct_len;
3115		- err = ceph_pagelist_reserve(pagelist, total_len);
3116	3853
3117		- if (!err) {
3118		- if (recon_state->msg_version >= 3) {
3119		- ceph_pagelist_encode_8(pagelist, struct_v);
3120		- ceph_pagelist_encode_8(pagelist, 1);
3121		- ceph_pagelist_encode_32(pagelist, struct_len);
3122		- }
3123		- ceph_pagelist_encode_string(pagelist, path, pathlen);
3124		- ceph_pagelist_append(pagelist, &rec, sizeof(rec.v2));
3125		- ceph_locks_to_pagelist(flocks, pagelist,
3126		- num_fcntl_locks,
3127		- num_flock_locks);
3128		- if (struct_v >= 2)
3129		- ceph_pagelist_encode_64(pagelist, snap_follows);
	3854	+ if (pagelist->length + total_len > RECONNECT_MAX_SIZE) {
	3855	+ err = send_reconnect_partial(recon_state);
	3856	+ if (err)
	3857	+ goto out_freeflocks;
	3858	+ pagelist = recon_state->pagelist;
3130	3859	}
	3860	+
	3861	+ err = ceph_pagelist_reserve(pagelist, total_len);
	3862	+ if (err)
	3863	+ goto out_freeflocks;
	3864	+
	3865	+ ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
	3866	+ if (recon_state->msg_version >= 3) {
	3867	+ ceph_pagelist_encode_8(pagelist, struct_v);
	3868	+ ceph_pagelist_encode_8(pagelist, 1);
	3869	+ ceph_pagelist_encode_32(pagelist, struct_len);
	3870	+ }
	3871	+ ceph_pagelist_encode_string(pagelist, path, pathlen);
	3872	+ ceph_pagelist_append(pagelist, &rec, sizeof(rec.v2));
	3873	+ ceph_locks_to_pagelist(flocks, pagelist,
	3874	+ num_fcntl_locks, num_flock_locks);
	3875	+ if (struct_v >= 2)
	3876	+ ceph_pagelist_encode_64(pagelist, snap_follows);
	3877	+out_freeflocks:
3131	3878	kfree(flocks);
3132	3879	} else {
3133		- size_t size = sizeof(u32) + pathlen + sizeof(rec.v1);
3134		- err = ceph_pagelist_reserve(pagelist, size);
3135		- if (!err) {
3136		- ceph_pagelist_encode_string(pagelist, path, pathlen);
3137		- ceph_pagelist_append(pagelist, &rec, sizeof(rec.v1));
3138		- }
	3880	+ err = ceph_pagelist_reserve(pagelist,
	3881	+ sizeof(u64) + sizeof(u32) +
	3882	+ pathlen + sizeof(rec.v1));
	3883	+ if (err)
	3884	+ goto out_err;
	3885	+
	3886	+ ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
	3887	+ ceph_pagelist_encode_string(pagelist, path, pathlen);
	3888	+ ceph_pagelist_append(pagelist, &rec, sizeof(rec.v1));
3139	3889	}
3140	3890
3141		- recon_state->nr_caps++;
3142		-out_free:
3143		- kfree(path);
3144		-out_dput:
3145		- dput(dentry);
	3891	+out_err:
	3892	+ ceph_mdsc_free_path(path, pathlen);
	3893	+ if (!err)
	3894	+ recon_state->nr_caps++;
	3895	+ return err;
	3896	+}
	3897	+
	3898	+static int encode_snap_realms(struct ceph_mds_client *mdsc,
	3899	+ struct ceph_reconnect_state *recon_state)
	3900	+{
	3901	+ struct rb_node *p;
	3902	+ struct ceph_pagelist *pagelist = recon_state->pagelist;
	3903	+ int err = 0;
	3904	+
	3905	+ if (recon_state->msg_version >= 4) {
	3906	+ err = ceph_pagelist_encode_32(pagelist, mdsc->num_snap_realms);
	3907	+ if (err < 0)
	3908	+ goto fail;
	3909	+ }
	3910	+
	3911	+ /*
	3912	+ * snaprealms. we provide mds with the ino, seq (version), and
	3913	+ * parent for all of our realms. If the mds has any newer info,
	3914	+ * it will tell us.
	3915	+ */
	3916	+ for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
	3917	+ struct ceph_snap_realm *realm =
	3918	+ rb_entry(p, struct ceph_snap_realm, node);
	3919	+ struct ceph_mds_snaprealm_reconnect sr_rec;
	3920	+
	3921	+ if (recon_state->msg_version >= 4) {
	3922	+ size_t need = sizeof(u8) * 2 + sizeof(u32) +
	3923	+ sizeof(sr_rec);
	3924	+
	3925	+ if (pagelist->length + need > RECONNECT_MAX_SIZE) {
	3926	+ err = send_reconnect_partial(recon_state);
	3927	+ if (err)
	3928	+ goto fail;
	3929	+ pagelist = recon_state->pagelist;
	3930	+ }
	3931	+
	3932	+ err = ceph_pagelist_reserve(pagelist, need);
	3933	+ if (err)
	3934	+ goto fail;
	3935	+
	3936	+ ceph_pagelist_encode_8(pagelist, 1);
	3937	+ ceph_pagelist_encode_8(pagelist, 1);
	3938	+ ceph_pagelist_encode_32(pagelist, sizeof(sr_rec));
	3939	+ }
	3940	+
	3941	+ dout(" adding snap realm %llx seq %lld parent %llx\n",
	3942	+ realm->ino, realm->seq, realm->parent_ino);
	3943	+ sr_rec.ino = cpu_to_le64(realm->ino);
	3944	+ sr_rec.seq = cpu_to_le64(realm->seq);
	3945	+ sr_rec.parent = cpu_to_le64(realm->parent_ino);
	3946	+
	3947	+ err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
	3948	+ if (err)
	3949	+ goto fail;
	3950	+
	3951	+ recon_state->nr_realms++;
	3952	+ }
	3953	+fail:
3146	3954	return err;
3147	3955	}
3148	3956
..	..	@@ -3156,31 +3964,29 @@
3156	3964	* recovering MDS might have.
3157	3965	*
3158	3966	* This is a relatively heavyweight operation, but it's rare.
3159		- *
3160		- * called with mdsc->mutex held.
3161	3967	*/
3162	3968	static void send_mds_reconnect(struct ceph_mds_client *mdsc,
3163	3969	struct ceph_mds_session *session)
3164	3970	{
3165	3971	struct ceph_msg *reply;
3166		- struct rb_node *p;
3167	3972	int mds = session->s_mds;
3168	3973	int err = -ENOMEM;
3169		- int s_nr_caps;
3170		- struct ceph_pagelist *pagelist;
3171		- struct ceph_reconnect_state recon_state;
	3974	+ struct ceph_reconnect_state recon_state = {
	3975	+ .session = session,
	3976	+ };
3172	3977	LIST_HEAD(dispose);
3173	3978
3174	3979	pr_info("mds%d reconnect start\n", mds);
3175	3980
3176		- pagelist = kmalloc(sizeof(*pagelist), GFP_NOFS);
3177		- if (!pagelist)
	3981	+ recon_state.pagelist = ceph_pagelist_alloc(GFP_NOFS);
	3982	+ if (!recon_state.pagelist)
3178	3983	goto fail_nopagelist;
3179		- ceph_pagelist_init(pagelist);
3180	3984
3181		- reply = ceph_msg_new(CEPH_MSG_CLIENT_RECONNECT, 0, GFP_NOFS, false);
	3985	+ reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);
3182	3986	if (!reply)
3183	3987	goto fail_nomsg;
	3988	+
	3989	+ xa_destroy(&session->s_delegated_inos);
3184	3990
3185	3991	mutex_lock(&session->s_mutex);
3186	3992	session->s_state = CEPH_MDS_SESSION_RECONNECTING;
..	..	@@ -3219,65 +4025,90 @@
3219	4025	/* replay unsafe requests */
3220	4026	replay_unsafe_requests(mdsc, session);
3221	4027
	4028	+ ceph_early_kick_flushing_caps(mdsc, session);
	4029	+
3222	4030	down_read(&mdsc->snap_rwsem);
3223	4031
3224		- /* traverse this session's caps */
3225		- s_nr_caps = session->s_nr_caps;
3226		- err = ceph_pagelist_encode_32(pagelist, s_nr_caps);
	4032	+ /* placeholder for nr_caps */
	4033	+ err = ceph_pagelist_encode_32(recon_state.pagelist, 0);
3227	4034	if (err)
3228	4035	goto fail;
3229	4036
3230		- recon_state.nr_caps = 0;
3231		- recon_state.pagelist = pagelist;
3232		- if (session->s_con.peer_features & CEPH_FEATURE_MDSENC)
	4037	+ if (test_bit(CEPHFS_FEATURE_MULTI_RECONNECT, &session->s_features)) {
3233	4038	recon_state.msg_version = 3;
3234		- else if (session->s_con.peer_features & CEPH_FEATURE_FLOCK)
	4039	+ recon_state.allow_multi = true;
	4040	+ } else if (session->s_con.peer_features & CEPH_FEATURE_MDSENC) {
	4041	+ recon_state.msg_version = 3;
	4042	+ } else {
3235	4043	recon_state.msg_version = 2;
3236		- else
3237		- recon_state.msg_version = 1;
3238		- err = iterate_session_caps(session, encode_caps_cb, &recon_state);
3239		- if (err < 0)
3240		- goto fail;
	4044	+ }
	4045	+ /* trsaverse this session's caps */
	4046	+ err = ceph_iterate_session_caps(session, reconnect_caps_cb, &recon_state);
3241	4047
3242	4048	spin_lock(&session->s_cap_lock);
3243	4049	session->s_cap_reconnect = 0;
3244	4050	spin_unlock(&session->s_cap_lock);
3245	4051
3246		- /*
3247		- * snaprealms. we provide mds with the ino, seq (version), and
3248		- * parent for all of our realms. If the mds has any newer info,
3249		- * it will tell us.
3250		- */
3251		- for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
3252		- struct ceph_snap_realm *realm =
3253		- rb_entry(p, struct ceph_snap_realm, node);
3254		- struct ceph_mds_snaprealm_reconnect sr_rec;
	4052	+ if (err < 0)
	4053	+ goto fail;
3255	4054
3256		- dout(" adding snap realm %llx seq %lld parent %llx\n",
3257		- realm->ino, realm->seq, realm->parent_ino);
3258		- sr_rec.ino = cpu_to_le64(realm->ino);
3259		- sr_rec.seq = cpu_to_le64(realm->seq);
3260		- sr_rec.parent = cpu_to_le64(realm->parent_ino);
3261		- err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
3262		- if (err)
	4055	+ /* check if all realms can be encoded into current message */
	4056	+ if (mdsc->num_snap_realms) {
	4057	+ size_t total_len =
	4058	+ recon_state.pagelist->length +
	4059	+ mdsc->num_snap_realms *
	4060	+ sizeof(struct ceph_mds_snaprealm_reconnect);
	4061	+ if (recon_state.msg_version >= 4) {
	4062	+ /* number of realms */
	4063	+ total_len += sizeof(u32);
	4064	+ /* version, compat_version and struct_len */
	4065	+ total_len += mdsc->num_snap_realms *
	4066	+ (2 * sizeof(u8) + sizeof(u32));
	4067	+ }
	4068	+ if (total_len > RECONNECT_MAX_SIZE) {
	4069	+ if (!recon_state.allow_multi) {
	4070	+ err = -ENOSPC;
	4071	+ goto fail;
	4072	+ }
	4073	+ if (recon_state.nr_caps) {
	4074	+ err = send_reconnect_partial(&recon_state);
	4075	+ if (err)
	4076	+ goto fail;
	4077	+ }
	4078	+ recon_state.msg_version = 5;
	4079	+ }
	4080	+ }
	4081	+
	4082	+ err = encode_snap_realms(mdsc, &recon_state);
	4083	+ if (err < 0)
	4084	+ goto fail;
	4085	+
	4086	+ if (recon_state.msg_version >= 5) {
	4087	+ err = ceph_pagelist_encode_8(recon_state.pagelist, 0);
	4088	+ if (err < 0)
3263	4089	goto fail;
3264	4090	}
3265	4091
3266		- reply->hdr.version = cpu_to_le16(recon_state.msg_version);
3267		-
3268		- /* raced with cap release? */
3269		- if (s_nr_caps != recon_state.nr_caps) {
3270		- struct page *page = list_first_entry(&pagelist->head,
3271		- struct page, lru);
	4092	+ if (recon_state.nr_caps \|\| recon_state.nr_realms) {
	4093	+ struct page *page =
	4094	+ list_first_entry(&recon_state.pagelist->head,
	4095	+ struct page, lru);
3272	4096	__le32 *addr = kmap_atomic(page);
3273		- *addr = cpu_to_le32(recon_state.nr_caps);
	4097	+ if (recon_state.nr_caps) {
	4098	+ WARN_ON(recon_state.nr_realms != mdsc->num_snap_realms);
	4099	+ *addr = cpu_to_le32(recon_state.nr_caps);
	4100	+ } else if (recon_state.msg_version >= 4) {
	4101	+ *(addr + 1) = cpu_to_le32(recon_state.nr_realms);
	4102	+ }
3274	4103	kunmap_atomic(addr);
3275	4104	}
3276	4105
3277		- reply->hdr.data_len = cpu_to_le32(pagelist->length);
3278		- ceph_msg_data_add_pagelist(reply, pagelist);
	4106	+ reply->hdr.version = cpu_to_le16(recon_state.msg_version);
	4107	+ if (recon_state.msg_version >= 4)
	4108	+ reply->hdr.compat_version = cpu_to_le16(4);
3279	4109
3280		- ceph_early_kick_flushing_caps(mdsc, session);
	4110	+ reply->hdr.data_len = cpu_to_le32(recon_state.pagelist->length);
	4111	+ ceph_msg_data_add_pagelist(reply, recon_state.pagelist);
3281	4112
3282	4113	ceph_con_send(&session->s_con, reply);
3283	4114
..	..	@@ -3288,6 +4119,7 @@
3288	4119	mutex_unlock(&mdsc->mutex);
3289	4120
3290	4121	up_read(&mdsc->snap_rwsem);
	4122	+ ceph_pagelist_release(recon_state.pagelist);
3291	4123	return;
3292	4124
3293	4125	fail:
..	..	@@ -3295,7 +4127,7 @@
3295	4127	up_read(&mdsc->snap_rwsem);
3296	4128	mutex_unlock(&session->s_mutex);
3297	4129	fail_nomsg:
3298		- ceph_pagelist_release(pagelist);
	4130	+ ceph_pagelist_release(recon_state.pagelist);
3299	4131	fail_nopagelist:
3300	4132	pr_err("error %d preparing reconnect for mds%d\n", err, mds);
3301	4133	return;
..	..	@@ -3319,7 +4151,7 @@
3319	4151	dout("check_new_map new %u old %u\n",
3320	4152	newmap->m_epoch, oldmap->m_epoch);
3321	4153
3322		- for (i = 0; i < oldmap->m_num_mds && i < mdsc->max_sessions; i++) {
	4154	+ for (i = 0; i < oldmap->possible_max_rank && i < mdsc->max_sessions; i++) {
3323	4155	if (!mdsc->sessions[i])
3324	4156	continue;
3325	4157	s = mdsc->sessions[i];
..	..	@@ -3333,42 +4165,35 @@
3333	4165	ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "",
3334	4166	ceph_session_state_name(s->s_state));
3335	4167
3336		- if (i >= newmap->m_num_mds \|\|
3337		- memcmp(ceph_mdsmap_get_addr(oldmap, i),
	4168	+ if (i >= newmap->possible_max_rank) {
	4169	+ /* force close session for stopped mds */
	4170	+ ceph_get_mds_session(s);
	4171	+ __unregister_session(mdsc, s);
	4172	+ __wake_requests(mdsc, &s->s_waiting);
	4173	+ mutex_unlock(&mdsc->mutex);
	4174	+
	4175	+ mutex_lock(&s->s_mutex);
	4176	+ cleanup_session_requests(mdsc, s);
	4177	+ remove_session_caps(s);
	4178	+ mutex_unlock(&s->s_mutex);
	4179	+
	4180	+ ceph_put_mds_session(s);
	4181	+
	4182	+ mutex_lock(&mdsc->mutex);
	4183	+ kick_requests(mdsc, i);
	4184	+ continue;
	4185	+ }
	4186	+
	4187	+ if (memcmp(ceph_mdsmap_get_addr(oldmap, i),
3338	4188	ceph_mdsmap_get_addr(newmap, i),
3339	4189	sizeof(struct ceph_entity_addr))) {
3340		- if (s->s_state == CEPH_MDS_SESSION_OPENING) {
3341		- /* the session never opened, just close it
3342		- * out now */
3343		- get_session(s);
3344		- __unregister_session(mdsc, s);
3345		- __wake_requests(mdsc, &s->s_waiting);
3346		- ceph_put_mds_session(s);
3347		- } else if (i >= newmap->m_num_mds) {
3348		- /* force close session for stopped mds */
3349		- get_session(s);
3350		- __unregister_session(mdsc, s);
3351		- __wake_requests(mdsc, &s->s_waiting);
3352		- kick_requests(mdsc, i);
3353		- mutex_unlock(&mdsc->mutex);
3354		-
3355		- mutex_lock(&s->s_mutex);
3356		- cleanup_session_requests(mdsc, s);
3357		- remove_session_caps(s);
3358		- mutex_unlock(&s->s_mutex);
3359		-
3360		- ceph_put_mds_session(s);
3361		-
3362		- mutex_lock(&mdsc->mutex);
3363		- } else {
3364		- /* just close it */
3365		- mutex_unlock(&mdsc->mutex);
3366		- mutex_lock(&s->s_mutex);
3367		- mutex_lock(&mdsc->mutex);
3368		- ceph_con_close(&s->s_con);
3369		- mutex_unlock(&s->s_mutex);
3370		- s->s_state = CEPH_MDS_SESSION_RESTARTING;
3371		- }
	4190	+ /* just close it */
	4191	+ mutex_unlock(&mdsc->mutex);
	4192	+ mutex_lock(&s->s_mutex);
	4193	+ mutex_lock(&mdsc->mutex);
	4194	+ ceph_con_close(&s->s_con);
	4195	+ mutex_unlock(&s->s_mutex);
	4196	+ s->s_state = CEPH_MDS_SESSION_RESTARTING;
3372	4197	} else if (oldstate == newstate) {
3373	4198	continue; /* nothing new with this mds */
3374	4199	}
..	..	@@ -3392,12 +4217,16 @@
3392	4217	oldstate != CEPH_MDS_STATE_STARTING)
3393	4218	pr_info("mds%d recovery completed\n", s->s_mds);
3394	4219	kick_requests(mdsc, i);
	4220	+ mutex_unlock(&mdsc->mutex);
	4221	+ mutex_lock(&s->s_mutex);
	4222	+ mutex_lock(&mdsc->mutex);
3395	4223	ceph_kick_flushing_caps(mdsc, s);
3396		- wake_up_session_caps(s, 1);
	4224	+ mutex_unlock(&s->s_mutex);
	4225	+ wake_up_session_caps(s, RECONNECT);
3397	4226	}
3398	4227	}
3399	4228
3400		- for (i = 0; i < newmap->m_num_mds && i < mdsc->max_sessions; i++) {
	4229	+ for (i = 0; i < newmap->possible_max_rank && i < mdsc->max_sessions; i++) {
3401	4230	s = mdsc->sessions[i];
3402	4231	if (!s)
3403	4232	continue;
..	..	@@ -3465,7 +4294,7 @@
3465	4294	dname.len, dname.name);
3466	4295
3467	4296	mutex_lock(&session->s_mutex);
3468		- session->s_seq++;
	4297	+ inc_session_sequence(session);
3469	4298
3470	4299	if (!inode) {
3471	4300	dout("handle_lease no inode %llx\n", vino.ino);
..	..	@@ -3526,8 +4355,9 @@
3526	4355	ceph_con_send(&session->s_con, msg);
3527	4356
3528	4357	out:
3529		- iput(inode);
3530	4358	mutex_unlock(&session->s_mutex);
	4359	+ /* avoid calling iput_final() in mds dispatch threads */
	4360	+ ceph_async_iput(inode);
3531	4361	return;
3532	4362
3533	4363	bad:
..	..	@@ -3536,31 +4366,33 @@
3536	4366	}
3537	4367
3538	4368	void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,
3539		- struct inode *inode,
3540	4369	struct dentry *dentry, char action,
3541	4370	u32 seq)
3542	4371	{
3543	4372	struct ceph_msg *msg;
3544	4373	struct ceph_mds_lease *lease;
3545		- int len = sizeof(*lease) + sizeof(u32);
3546		- int dnamelen = 0;
	4374	+ struct inode *dir;
	4375	+ int len = sizeof(*lease) + sizeof(u32) + NAME_MAX;
3547	4376
3548		- dout("lease_send_msg inode %p dentry %p %s to mds%d\n",
3549		- inode, dentry, ceph_lease_op_name(action), session->s_mds);
3550		- dnamelen = dentry->d_name.len;
3551		- len += dnamelen;
	4377	+ dout("lease_send_msg identry %p %s to mds%d\n",
	4378	+ dentry, ceph_lease_op_name(action), session->s_mds);
3552	4379
3553	4380	msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS, false);
3554	4381	if (!msg)
3555	4382	return;
3556	4383	lease = msg->front.iov_base;
3557	4384	lease->action = action;
3558		- lease->ino = cpu_to_le64(ceph_vino(inode).ino);
3559		- lease->first = lease->last = cpu_to_le64(ceph_vino(inode).snap);
3560	4385	lease->seq = cpu_to_le32(seq);
3561		- put_unaligned_le32(dnamelen, lease + 1);
3562		- memcpy((void *)(lease + 1) + 4, dentry->d_name.name, dnamelen);
3563	4386
	4387	+ spin_lock(&dentry->d_lock);
	4388	+ dir = d_inode(dentry->d_parent);
	4389	+ lease->ino = cpu_to_le64(ceph_ino(dir));
	4390	+ lease->first = lease->last = cpu_to_le64(ceph_snap(dir));
	4391	+
	4392	+ put_unaligned_le32(dentry->d_name.len, lease + 1);
	4393	+ memcpy((void *)(lease + 1) + 4,
	4394	+ dentry->d_name.name, dentry->d_name.len);
	4395	+ spin_unlock(&dentry->d_lock);
3564	4396	/*
3565	4397	* if this is a preemptive lease RELEASE, no need to
3566	4398	* flush request stream, since the actual request will
..	..	@@ -3572,48 +4404,106 @@
3572	4404	}
3573	4405
3574	4406	/*
3575		- * lock unlock sessions, to wait ongoing session activities
	4407	+ * lock unlock the session, to wait ongoing session activities
3576	4408	*/
3577		-static void lock_unlock_sessions(struct ceph_mds_client *mdsc)
	4409	+static void lock_unlock_session(struct ceph_mds_session *s)
3578	4410	{
3579		- int i;
3580		-
3581		- mutex_lock(&mdsc->mutex);
3582		- for (i = 0; i < mdsc->max_sessions; i++) {
3583		- struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
3584		- if (!s)
3585		- continue;
3586		- mutex_unlock(&mdsc->mutex);
3587		- mutex_lock(&s->s_mutex);
3588		- mutex_unlock(&s->s_mutex);
3589		- ceph_put_mds_session(s);
3590		- mutex_lock(&mdsc->mutex);
3591		- }
3592		- mutex_unlock(&mdsc->mutex);
	4411	+ mutex_lock(&s->s_mutex);
	4412	+ mutex_unlock(&s->s_mutex);
3593	4413	}
3594	4414
	4415	+static void maybe_recover_session(struct ceph_mds_client *mdsc)
	4416	+{
	4417	+ struct ceph_fs_client *fsc = mdsc->fsc;
3595	4418
	4419	+ if (!ceph_test_mount_opt(fsc, CLEANRECOVER))
	4420	+ return;
	4421	+
	4422	+ if (READ_ONCE(fsc->mount_state) != CEPH_MOUNT_MOUNTED)
	4423	+ return;
	4424	+
	4425	+ if (!READ_ONCE(fsc->blocklisted))
	4426	+ return;
	4427	+
	4428	+ if (fsc->last_auto_reconnect &&
	4429	+ time_before(jiffies, fsc->last_auto_reconnect + HZ * 60 * 30))
	4430	+ return;
	4431	+
	4432	+ pr_info("auto reconnect after blocklisted\n");
	4433	+ fsc->last_auto_reconnect = jiffies;
	4434	+ ceph_force_reconnect(fsc->sb);
	4435	+}
	4436	+
	4437	+bool check_session_state(struct ceph_mds_session *s)
	4438	+{
	4439	+ switch (s->s_state) {
	4440	+ case CEPH_MDS_SESSION_OPEN:
	4441	+ if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
	4442	+ s->s_state = CEPH_MDS_SESSION_HUNG;
	4443	+ pr_info("mds%d hung\n", s->s_mds);
	4444	+ }
	4445	+ break;
	4446	+ case CEPH_MDS_SESSION_CLOSING:
	4447	+ /* Should never reach this when we're unmounting */
	4448	+ WARN_ON_ONCE(s->s_ttl);
	4449	+ fallthrough;
	4450	+ case CEPH_MDS_SESSION_NEW:
	4451	+ case CEPH_MDS_SESSION_RESTARTING:
	4452	+ case CEPH_MDS_SESSION_CLOSED:
	4453	+ case CEPH_MDS_SESSION_REJECTED:
	4454	+ return false;
	4455	+ }
	4456	+
	4457	+ return true;
	4458	+}
3596	4459
3597	4460	/*
3598		- * delayed work -- periodically trim expired leases, renew caps with mds
	4461	+ * If the sequence is incremented while we're waiting on a REQUEST_CLOSE reply,
	4462	+ * then we need to retransmit that request.
3599	4463	*/
3600		-static void schedule_delayed(struct ceph_mds_client *mdsc)
	4464	+void inc_session_sequence(struct ceph_mds_session *s)
3601	4465	{
3602		- int delay = 5;
3603		- unsigned hz = round_jiffies_relative(HZ * delay);
3604		- schedule_delayed_work(&mdsc->delayed_work, hz);
	4466	+ lockdep_assert_held(&s->s_mutex);
	4467	+
	4468	+ s->s_seq++;
	4469	+
	4470	+ if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
	4471	+ int ret;
	4472	+
	4473	+ dout("resending session close request for mds%d\n", s->s_mds);
	4474	+ ret = request_close_session(s);
	4475	+ if (ret < 0)
	4476	+ pr_err("unable to close session to mds%d: %d\n",
	4477	+ s->s_mds, ret);
	4478	+ }
	4479	+}
	4480	+
	4481	+/*
	4482	+ * delayed work -- periodically trim expired leases, renew caps with mds. If
	4483	+ * the @delay parameter is set to 0 or if it's more than 5 secs, the default
	4484	+ * workqueue delay value of 5 secs will be used.
	4485	+ */
	4486	+static void schedule_delayed(struct ceph_mds_client *mdsc, unsigned long delay)
	4487	+{
	4488	+ unsigned long max_delay = HZ * 5;
	4489	+
	4490	+ /* 5 secs default delay */
	4491	+ if (!delay \|\| (delay > max_delay))
	4492	+ delay = max_delay;
	4493	+ schedule_delayed_work(&mdsc->delayed_work,
	4494	+ round_jiffies_relative(delay));
3605	4495	}
3606	4496
3607	4497	static void delayed_work(struct work_struct *work)
3608	4498	{
3609		- int i;
3610	4499	struct ceph_mds_client *mdsc =
3611	4500	container_of(work, struct ceph_mds_client, delayed_work.work);
	4501	+ unsigned long delay;
3612	4502	int renew_interval;
3613	4503	int renew_caps;
	4504	+ int i;
3614	4505
3615	4506	dout("mdsc delayed_work\n");
3616		- ceph_check_delayed_caps(mdsc);
3617	4507
3618	4508	if (mdsc->stopping)
3619	4509	return;
..	..	@@ -3629,23 +4519,8 @@
3629	4519	struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
3630	4520	if (!s)
3631	4521	continue;
3632		- if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
3633		- dout("resending session close request for mds%d\n",
3634		- s->s_mds);
3635		- request_close_session(mdsc, s);
3636		- ceph_put_mds_session(s);
3637		- continue;
3638		- }
3639		- if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
3640		- if (s->s_state == CEPH_MDS_SESSION_OPEN) {
3641		- s->s_state = CEPH_MDS_SESSION_HUNG;
3642		- pr_info("mds%d hung\n", s->s_mds);
3643		- }
3644		- }
3645		- if (s->s_state == CEPH_MDS_SESSION_NEW \|\|
3646		- s->s_state == CEPH_MDS_SESSION_RESTARTING \|\|
3647		- s->s_state == CEPH_MDS_SESSION_REJECTED) {
3648		- /* this mds is failed or recovering, just wait */
	4522	+
	4523	+ if (!check_session_state(s)) {
3649	4524	ceph_put_mds_session(s);
3650	4525	continue;
3651	4526	}
..	..	@@ -3666,13 +4541,22 @@
3666	4541	}
3667	4542	mutex_unlock(&mdsc->mutex);
3668	4543
3669		- schedule_delayed(mdsc);
	4544	+ delay = ceph_check_delayed_caps(mdsc);
	4545	+
	4546	+ ceph_queue_cap_reclaim_work(mdsc);
	4547	+
	4548	+ ceph_trim_snapid_map(mdsc);
	4549	+
	4550	+ maybe_recover_session(mdsc);
	4551	+
	4552	+ schedule_delayed(mdsc, delay);
3670	4553	}
3671	4554
3672	4555	int ceph_mdsc_init(struct ceph_fs_client *fsc)
3673	4556
3674	4557	{
3675	4558	struct ceph_mds_client *mdsc;
	4559	+ int err;
3676	4560
3677	4561	mdsc = kzalloc(sizeof(struct ceph_mds_client), GFP_NOFS);
3678	4562	if (!mdsc)
..	..	@@ -3681,8 +4565,8 @@
3681	4565	mutex_init(&mdsc->mutex);
3682	4566	mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS);
3683	4567	if (!mdsc->mdsmap) {
3684		- kfree(mdsc);
3685		- return -ENOMEM;
	4568	+ err = -ENOMEM;
	4569	+ goto err_mdsc;
3686	4570	}
3687	4571
3688	4572	init_completion(&mdsc->safe_umount_waiters);
..	..	@@ -3693,10 +4577,13 @@
3693	4577	mdsc->max_sessions = 0;
3694	4578	mdsc->stopping = 0;
3695	4579	atomic64_set(&mdsc->quotarealms_count, 0);
	4580	+ mdsc->quotarealms_inodes = RB_ROOT;
	4581	+ mutex_init(&mdsc->quotarealms_inodes_mutex);
3696	4582	mdsc->last_snap_seq = 0;
3697	4583	init_rwsem(&mdsc->snap_rwsem);
3698	4584	mdsc->snap_realms = RB_ROOT;
3699	4585	INIT_LIST_HEAD(&mdsc->snap_empty);
	4586	+ mdsc->num_snap_realms = 0;
3700	4587	spin_lock_init(&mdsc->snap_empty_lock);
3701	4588	mdsc->last_tid = 0;
3702	4589	mdsc->oldest_tid = 0;
..	..	@@ -3704,21 +4591,32 @@
3704	4591	INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);
3705	4592	mdsc->last_renew_caps = jiffies;
3706	4593	INIT_LIST_HEAD(&mdsc->cap_delay_list);
	4594	+ INIT_LIST_HEAD(&mdsc->cap_wait_list);
3707	4595	spin_lock_init(&mdsc->cap_delay_lock);
3708	4596	INIT_LIST_HEAD(&mdsc->snap_flush_list);
3709	4597	spin_lock_init(&mdsc->snap_flush_lock);
3710	4598	mdsc->last_cap_flush_tid = 1;
3711	4599	INIT_LIST_HEAD(&mdsc->cap_flush_list);
3712		- INIT_LIST_HEAD(&mdsc->cap_dirty);
3713	4600	INIT_LIST_HEAD(&mdsc->cap_dirty_migrating);
3714	4601	mdsc->num_cap_flushing = 0;
3715	4602	spin_lock_init(&mdsc->cap_dirty_lock);
3716	4603	init_waitqueue_head(&mdsc->cap_flushing_wq);
3717		- spin_lock_init(&mdsc->dentry_lru_lock);
3718		- INIT_LIST_HEAD(&mdsc->dentry_lru);
	4604	+ INIT_WORK(&mdsc->cap_reclaim_work, ceph_cap_reclaim_work);
	4605	+ atomic_set(&mdsc->cap_reclaim_pending, 0);
	4606	+ err = ceph_metric_init(&mdsc->metric);
	4607	+ if (err)
	4608	+ goto err_mdsmap;
	4609	+
	4610	+ spin_lock_init(&mdsc->dentry_list_lock);
	4611	+ INIT_LIST_HEAD(&mdsc->dentry_leases);
	4612	+ INIT_LIST_HEAD(&mdsc->dentry_dir_leases);
3719	4613
3720	4614	ceph_caps_init(mdsc);
3721		- ceph_adjust_min_caps(mdsc, fsc->min_caps);
	4615	+ ceph_adjust_caps_max_min(mdsc, fsc->mount_options);
	4616	+
	4617	+ spin_lock_init(&mdsc->snapid_map_lock);
	4618	+ mdsc->snapid_map_tree = RB_ROOT;
	4619	+ INIT_LIST_HEAD(&mdsc->snapid_map_lru);
3722	4620
3723	4621	init_rwsem(&mdsc->pool_perm_rwsem);
3724	4622	mdsc->pool_perm_tree = RB_ROOT;
..	..	@@ -3728,6 +4626,12 @@
3728	4626
3729	4627	fsc->mdsc = mdsc;
3730	4628	return 0;
	4629	+
	4630	+err_mdsmap:
	4631	+ kfree(mdsc->mdsmap);
	4632	+err_mdsc:
	4633	+ kfree(mdsc);
	4634	+ return err;
3731	4635	}
3732	4636
3733	4637	/*
..	..	@@ -3752,11 +4656,36 @@
3752	4656	while ((req = __get_oldest_req(mdsc))) {
3753	4657	dout("wait_requests timed out on tid %llu\n",
3754	4658	req->r_tid);
	4659	+ list_del_init(&req->r_wait);
3755	4660	__unregister_request(mdsc, req);
3756	4661	}
3757	4662	}
3758	4663	mutex_unlock(&mdsc->mutex);
3759	4664	dout("wait_requests done\n");
	4665	+}
	4666	+
	4667	+void send_flush_mdlog(struct ceph_mds_session *s)
	4668	+{
	4669	+ struct ceph_msg *msg;
	4670	+
	4671	+ /*
	4672	+ * Pre-luminous MDS crashes when it sees an unknown session request
	4673	+ */
	4674	+ if (!CEPH_HAVE_FEATURE(s->s_con.peer_features, SERVER_LUMINOUS))
	4675	+ return;
	4676	+
	4677	+ mutex_lock(&s->s_mutex);
	4678	+ dout("request mdlog flush to mds%d (%s)s seq %lld\n", s->s_mds,
	4679	+ ceph_session_state_name(s->s_state), s->s_seq);
	4680	+ msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_FLUSH_MDLOG,
	4681	+ s->s_seq);
	4682	+ if (!msg) {
	4683	+ pr_err("failed to request mdlog flush to mds%d (%s) seq %lld\n",
	4684	+ s->s_mds, ceph_session_state_name(s->s_state), s->s_seq);
	4685	+ } else {
	4686	+ ceph_con_send(&s->s_con, msg);
	4687	+ }
	4688	+ mutex_unlock(&s->s_mutex);
3760	4689	}
3761	4690
3762	4691	/*
..	..	@@ -3768,7 +4697,8 @@
3768	4697	dout("pre_umount\n");
3769	4698	mdsc->stopping = 1;
3770	4699
3771		- lock_unlock_sessions(mdsc);
	4700	+ ceph_mdsc_iterate_sessions(mdsc, send_flush_mdlog, true);
	4701	+ ceph_mdsc_iterate_sessions(mdsc, lock_unlock_session, false);
3772	4702	ceph_flush_dirty_caps(mdsc);
3773	4703	wait_requests(mdsc);
3774	4704
..	..	@@ -3777,6 +4707,8 @@
3777	4707	* their inode/dcache refs
3778	4708	*/
3779	4709	ceph_msgr_flush();
	4710	+
	4711	+ ceph_cleanup_quotarealms_inodes(mdsc);
3780	4712	}
3781	4713
3782	4714	/*
..	..	@@ -3902,7 +4834,7 @@
3902	4834	mutex_lock(&mdsc->mutex);
3903	4835	for (i = 0; i < mdsc->max_sessions; i++) {
3904	4836	if (mdsc->sessions[i]) {
3905		- session = get_session(mdsc->sessions[i]);
	4837	+ session = ceph_get_mds_session(mdsc->sessions[i]);
3906	4838	__unregister_session(mdsc, session);
3907	4839	mutex_unlock(&mdsc->mutex);
3908	4840	mutex_lock(&session->s_mutex);
..	..	@@ -3915,8 +4847,10 @@
3915	4847	WARN_ON(!list_empty(&mdsc->cap_delay_list));
3916	4848	mutex_unlock(&mdsc->mutex);
3917	4849
	4850	+ ceph_cleanup_snapid_map(mdsc);
3918	4851	ceph_cleanup_empty_realms(mdsc);
3919	4852
	4853	+ cancel_work_sync(&mdsc->cap_reclaim_work);
3920	4854	cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
3921	4855
3922	4856	dout("stopped\n");
..	..	@@ -3934,7 +4868,12 @@
3934	4868	session = __ceph_lookup_mds_session(mdsc, mds);
3935	4869	if (!session)
3936	4870	continue;
	4871	+
	4872	+ if (session->s_state == CEPH_MDS_SESSION_REJECTED)
	4873	+ __unregister_session(mdsc, session);
	4874	+ __wake_requests(mdsc, &session->s_waiting);
3937	4875	mutex_unlock(&mdsc->mutex);
	4876	+
3938	4877	mutex_lock(&session->s_mutex);
3939	4878	__close_session(mdsc, session);
3940	4879	if (session->s_state == CEPH_MDS_SESSION_CLOSING) {
..	..	@@ -3943,6 +4882,7 @@
3943	4882	}
3944	4883	mutex_unlock(&session->s_mutex);
3945	4884	ceph_put_mds_session(session);
	4885	+
3946	4886	mutex_lock(&mdsc->mutex);
3947	4887	kick_requests(mdsc, mds);
3948	4888	}
..	..	@@ -3982,6 +4922,8 @@
3982	4922	ceph_msgr_flush();
3983	4923
3984	4924	ceph_mdsc_stop(mdsc);
	4925	+
	4926	+ ceph_metric_destroy(&mdsc->metric);
3985	4927
3986	4928	fsc->mdsc = NULL;
3987	4929	kfree(mdsc);
..	..	@@ -4117,7 +5059,7 @@
4117	5059	mdsc->mdsmap->m_epoch);
4118	5060
4119	5061	mutex_unlock(&mdsc->mutex);
4120		- schedule_delayed(mdsc);
	5062	+ schedule_delayed(mdsc, 0);
4121	5063	return;
4122	5064
4123	5065	bad_unlock:
..	..	@@ -4131,11 +5073,8 @@
4131	5073	{
4132	5074	struct ceph_mds_session *s = con->private;
4133	5075
4134		- if (get_session(s)) {
4135		- dout("mdsc con_get %p ok (%d)\n", s, refcount_read(&s->s_ref));
	5076	+ if (ceph_get_mds_session(s))
4136	5077	return con;
4137		- }
4138		- dout("mdsc con_get %p FAIL\n", s);
4139	5078	return NULL;
4140	5079	}
4141	5080
..	..	@@ -4143,7 +5082,6 @@
4143	5082	{
4144	5083	struct ceph_mds_session *s = con->private;
4145	5084
4146		- dout("mdsc con_put %p (%d)\n", s, refcount_read(&s->s_ref) - 1);
4147	5085	ceph_put_mds_session(s);
4148	5086	}
4149	5087