• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1 // SPDX-License-Identifier: GPL-2.0
2 #include <linux/ceph/ceph_debug.h>
3 
4 #include <linux/fs.h>
5 #include <linux/wait.h>
6 #include <linux/slab.h>
7 #include <linux/gfp.h>
8 #include <linux/sched.h>
9 #include <linux/debugfs.h>
10 #include <linux/seq_file.h>
11 #include <linux/ratelimit.h>
12 #include <linux/bits.h>
13 #include <linux/ktime.h>
14 #include <linux/bitmap.h>
15 
16 #include "super.h"
17 #include "mds_client.h"
18 
19 #include <linux/ceph/ceph_features.h>
20 #include <linux/ceph/messenger.h>
21 #include <linux/ceph/decode.h>
22 #include <linux/ceph/pagelist.h>
23 #include <linux/ceph/auth.h>
24 #include <linux/ceph/debugfs.h>
25 
26 #define RECONNECT_MAX_SIZE (INT_MAX - PAGE_SIZE)
27 
28 /*
29  * A cluster of MDS (metadata server) daemons is responsible for
30  * managing the file system namespace (the directory hierarchy and
31  * inodes) and for coordinating shared access to storage.  Metadata is
32  * partitioning hierarchically across a number of servers, and that
33  * partition varies over time as the cluster adjusts the distribution
34  * in order to balance load.
35  *
36  * The MDS client is primarily responsible to managing synchronous
37  * metadata requests for operations like open, unlink, and so forth.
38  * If there is a MDS failure, we find out about it when we (possibly
39  * request and) receive a new MDS map, and can resubmit affected
40  * requests.
41  *
42  * For the most part, though, we take advantage of a lossless
43  * communications channel to the MDS, and do not need to worry about
44  * timing out or resubmitting requests.
45  *
46  * We maintain a stateful "session" with each MDS we interact with.
47  * Within each session, we sent periodic heartbeat messages to ensure
48  * any capabilities or leases we have been issues remain valid.  If
49  * the session times out and goes stale, our leases and capabilities
50  * are no longer valid.
51  */
52 
53 struct ceph_reconnect_state {
54 	struct ceph_mds_session *session;
55 	int nr_caps, nr_realms;
56 	struct ceph_pagelist *pagelist;
57 	unsigned msg_version;
58 	bool allow_multi;
59 };
60 
61 static void __wake_requests(struct ceph_mds_client *mdsc,
62 			    struct list_head *head);
63 static void ceph_cap_release_work(struct work_struct *work);
64 static void ceph_cap_reclaim_work(struct work_struct *work);
65 
66 static const struct ceph_connection_operations mds_con_ops;
67 
68 
69 /*
70  * mds reply parsing
71  */
72 
parse_reply_info_quota(void ** p,void * end,struct ceph_mds_reply_info_in * info)73 static int parse_reply_info_quota(void **p, void *end,
74 				  struct ceph_mds_reply_info_in *info)
75 {
76 	u8 struct_v, struct_compat;
77 	u32 struct_len;
78 
79 	ceph_decode_8_safe(p, end, struct_v, bad);
80 	ceph_decode_8_safe(p, end, struct_compat, bad);
81 	/* struct_v is expected to be >= 1. we only
82 	 * understand encoding with struct_compat == 1. */
83 	if (!struct_v || struct_compat != 1)
84 		goto bad;
85 	ceph_decode_32_safe(p, end, struct_len, bad);
86 	ceph_decode_need(p, end, struct_len, bad);
87 	end = *p + struct_len;
88 	ceph_decode_64_safe(p, end, info->max_bytes, bad);
89 	ceph_decode_64_safe(p, end, info->max_files, bad);
90 	*p = end;
91 	return 0;
92 bad:
93 	return -EIO;
94 }
95 
96 /*
97  * parse individual inode info
98  */
parse_reply_info_in(void ** p,void * end,struct ceph_mds_reply_info_in * info,u64 features)99 static int parse_reply_info_in(void **p, void *end,
100 			       struct ceph_mds_reply_info_in *info,
101 			       u64 features)
102 {
103 	int err = 0;
104 	u8 struct_v = 0;
105 
106 	if (features == (u64)-1) {
107 		u32 struct_len;
108 		u8 struct_compat;
109 		ceph_decode_8_safe(p, end, struct_v, bad);
110 		ceph_decode_8_safe(p, end, struct_compat, bad);
111 		/* struct_v is expected to be >= 1. we only understand
112 		 * encoding with struct_compat == 1. */
113 		if (!struct_v || struct_compat != 1)
114 			goto bad;
115 		ceph_decode_32_safe(p, end, struct_len, bad);
116 		ceph_decode_need(p, end, struct_len, bad);
117 		end = *p + struct_len;
118 	}
119 
120 	ceph_decode_need(p, end, sizeof(struct ceph_mds_reply_inode), bad);
121 	info->in = *p;
122 	*p += sizeof(struct ceph_mds_reply_inode) +
123 		sizeof(*info->in->fragtree.splits) *
124 		le32_to_cpu(info->in->fragtree.nsplits);
125 
126 	ceph_decode_32_safe(p, end, info->symlink_len, bad);
127 	ceph_decode_need(p, end, info->symlink_len, bad);
128 	info->symlink = *p;
129 	*p += info->symlink_len;
130 
131 	ceph_decode_copy_safe(p, end, &info->dir_layout,
132 			      sizeof(info->dir_layout), bad);
133 	ceph_decode_32_safe(p, end, info->xattr_len, bad);
134 	ceph_decode_need(p, end, info->xattr_len, bad);
135 	info->xattr_data = *p;
136 	*p += info->xattr_len;
137 
138 	if (features == (u64)-1) {
139 		/* inline data */
140 		ceph_decode_64_safe(p, end, info->inline_version, bad);
141 		ceph_decode_32_safe(p, end, info->inline_len, bad);
142 		ceph_decode_need(p, end, info->inline_len, bad);
143 		info->inline_data = *p;
144 		*p += info->inline_len;
145 		/* quota */
146 		err = parse_reply_info_quota(p, end, info);
147 		if (err < 0)
148 			goto out_bad;
149 		/* pool namespace */
150 		ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
151 		if (info->pool_ns_len > 0) {
152 			ceph_decode_need(p, end, info->pool_ns_len, bad);
153 			info->pool_ns_data = *p;
154 			*p += info->pool_ns_len;
155 		}
156 
157 		/* btime */
158 		ceph_decode_need(p, end, sizeof(info->btime), bad);
159 		ceph_decode_copy(p, &info->btime, sizeof(info->btime));
160 
161 		/* change attribute */
162 		ceph_decode_64_safe(p, end, info->change_attr, bad);
163 
164 		/* dir pin */
165 		if (struct_v >= 2) {
166 			ceph_decode_32_safe(p, end, info->dir_pin, bad);
167 		} else {
168 			info->dir_pin = -ENODATA;
169 		}
170 
171 		/* snapshot birth time, remains zero for v<=2 */
172 		if (struct_v >= 3) {
173 			ceph_decode_need(p, end, sizeof(info->snap_btime), bad);
174 			ceph_decode_copy(p, &info->snap_btime,
175 					 sizeof(info->snap_btime));
176 		} else {
177 			memset(&info->snap_btime, 0, sizeof(info->snap_btime));
178 		}
179 
180 		/* snapshot count, remains zero for v<=3 */
181 		if (struct_v >= 4) {
182 			ceph_decode_64_safe(p, end, info->rsnaps, bad);
183 		} else {
184 			info->rsnaps = 0;
185 		}
186 
187 		*p = end;
188 	} else {
189 		if (features & CEPH_FEATURE_MDS_INLINE_DATA) {
190 			ceph_decode_64_safe(p, end, info->inline_version, bad);
191 			ceph_decode_32_safe(p, end, info->inline_len, bad);
192 			ceph_decode_need(p, end, info->inline_len, bad);
193 			info->inline_data = *p;
194 			*p += info->inline_len;
195 		} else
196 			info->inline_version = CEPH_INLINE_NONE;
197 
198 		if (features & CEPH_FEATURE_MDS_QUOTA) {
199 			err = parse_reply_info_quota(p, end, info);
200 			if (err < 0)
201 				goto out_bad;
202 		} else {
203 			info->max_bytes = 0;
204 			info->max_files = 0;
205 		}
206 
207 		info->pool_ns_len = 0;
208 		info->pool_ns_data = NULL;
209 		if (features & CEPH_FEATURE_FS_FILE_LAYOUT_V2) {
210 			ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
211 			if (info->pool_ns_len > 0) {
212 				ceph_decode_need(p, end, info->pool_ns_len, bad);
213 				info->pool_ns_data = *p;
214 				*p += info->pool_ns_len;
215 			}
216 		}
217 
218 		if (features & CEPH_FEATURE_FS_BTIME) {
219 			ceph_decode_need(p, end, sizeof(info->btime), bad);
220 			ceph_decode_copy(p, &info->btime, sizeof(info->btime));
221 			ceph_decode_64_safe(p, end, info->change_attr, bad);
222 		}
223 
224 		info->dir_pin = -ENODATA;
225 		/* info->snap_btime and info->rsnaps remain zero */
226 	}
227 	return 0;
228 bad:
229 	err = -EIO;
230 out_bad:
231 	return err;
232 }
233 
parse_reply_info_dir(void ** p,void * end,struct ceph_mds_reply_dirfrag ** dirfrag,u64 features)234 static int parse_reply_info_dir(void **p, void *end,
235 				struct ceph_mds_reply_dirfrag **dirfrag,
236 				u64 features)
237 {
238 	if (features == (u64)-1) {
239 		u8 struct_v, struct_compat;
240 		u32 struct_len;
241 		ceph_decode_8_safe(p, end, struct_v, bad);
242 		ceph_decode_8_safe(p, end, struct_compat, bad);
243 		/* struct_v is expected to be >= 1. we only understand
244 		 * encoding whose struct_compat == 1. */
245 		if (!struct_v || struct_compat != 1)
246 			goto bad;
247 		ceph_decode_32_safe(p, end, struct_len, bad);
248 		ceph_decode_need(p, end, struct_len, bad);
249 		end = *p + struct_len;
250 	}
251 
252 	ceph_decode_need(p, end, sizeof(**dirfrag), bad);
253 	*dirfrag = *p;
254 	*p += sizeof(**dirfrag) + sizeof(u32) * le32_to_cpu((*dirfrag)->ndist);
255 	if (unlikely(*p > end))
256 		goto bad;
257 	if (features == (u64)-1)
258 		*p = end;
259 	return 0;
260 bad:
261 	return -EIO;
262 }
263 
parse_reply_info_lease(void ** p,void * end,struct ceph_mds_reply_lease ** lease,u64 features)264 static int parse_reply_info_lease(void **p, void *end,
265 				  struct ceph_mds_reply_lease **lease,
266 				  u64 features)
267 {
268 	if (features == (u64)-1) {
269 		u8 struct_v, struct_compat;
270 		u32 struct_len;
271 		ceph_decode_8_safe(p, end, struct_v, bad);
272 		ceph_decode_8_safe(p, end, struct_compat, bad);
273 		/* struct_v is expected to be >= 1. we only understand
274 		 * encoding whose struct_compat == 1. */
275 		if (!struct_v || struct_compat != 1)
276 			goto bad;
277 		ceph_decode_32_safe(p, end, struct_len, bad);
278 		ceph_decode_need(p, end, struct_len, bad);
279 		end = *p + struct_len;
280 	}
281 
282 	ceph_decode_need(p, end, sizeof(**lease), bad);
283 	*lease = *p;
284 	*p += sizeof(**lease);
285 	if (features == (u64)-1)
286 		*p = end;
287 	return 0;
288 bad:
289 	return -EIO;
290 }
291 
292 /*
293  * parse a normal reply, which may contain a (dir+)dentry and/or a
294  * target inode.
295  */
parse_reply_info_trace(void ** p,void * end,struct ceph_mds_reply_info_parsed * info,u64 features)296 static int parse_reply_info_trace(void **p, void *end,
297 				  struct ceph_mds_reply_info_parsed *info,
298 				  u64 features)
299 {
300 	int err;
301 
302 	if (info->head->is_dentry) {
303 		err = parse_reply_info_in(p, end, &info->diri, features);
304 		if (err < 0)
305 			goto out_bad;
306 
307 		err = parse_reply_info_dir(p, end, &info->dirfrag, features);
308 		if (err < 0)
309 			goto out_bad;
310 
311 		ceph_decode_32_safe(p, end, info->dname_len, bad);
312 		ceph_decode_need(p, end, info->dname_len, bad);
313 		info->dname = *p;
314 		*p += info->dname_len;
315 
316 		err = parse_reply_info_lease(p, end, &info->dlease, features);
317 		if (err < 0)
318 			goto out_bad;
319 	}
320 
321 	if (info->head->is_target) {
322 		err = parse_reply_info_in(p, end, &info->targeti, features);
323 		if (err < 0)
324 			goto out_bad;
325 	}
326 
327 	if (unlikely(*p != end))
328 		goto bad;
329 	return 0;
330 
331 bad:
332 	err = -EIO;
333 out_bad:
334 	pr_err("problem parsing mds trace %d\n", err);
335 	return err;
336 }
337 
338 /*
339  * parse readdir results
340  */
parse_reply_info_readdir(void ** p,void * end,struct ceph_mds_reply_info_parsed * info,u64 features)341 static int parse_reply_info_readdir(void **p, void *end,
342 				struct ceph_mds_reply_info_parsed *info,
343 				u64 features)
344 {
345 	u32 num, i = 0;
346 	int err;
347 
348 	err = parse_reply_info_dir(p, end, &info->dir_dir, features);
349 	if (err < 0)
350 		goto out_bad;
351 
352 	ceph_decode_need(p, end, sizeof(num) + 2, bad);
353 	num = ceph_decode_32(p);
354 	{
355 		u16 flags = ceph_decode_16(p);
356 		info->dir_end = !!(flags & CEPH_READDIR_FRAG_END);
357 		info->dir_complete = !!(flags & CEPH_READDIR_FRAG_COMPLETE);
358 		info->hash_order = !!(flags & CEPH_READDIR_HASH_ORDER);
359 		info->offset_hash = !!(flags & CEPH_READDIR_OFFSET_HASH);
360 	}
361 	if (num == 0)
362 		goto done;
363 
364 	BUG_ON(!info->dir_entries);
365 	if ((unsigned long)(info->dir_entries + num) >
366 	    (unsigned long)info->dir_entries + info->dir_buf_size) {
367 		pr_err("dir contents are larger than expected\n");
368 		WARN_ON(1);
369 		goto bad;
370 	}
371 
372 	info->dir_nr = num;
373 	while (num) {
374 		struct ceph_mds_reply_dir_entry *rde = info->dir_entries + i;
375 		/* dentry */
376 		ceph_decode_32_safe(p, end, rde->name_len, bad);
377 		ceph_decode_need(p, end, rde->name_len, bad);
378 		rde->name = *p;
379 		*p += rde->name_len;
380 		dout("parsed dir dname '%.*s'\n", rde->name_len, rde->name);
381 
382 		/* dentry lease */
383 		err = parse_reply_info_lease(p, end, &rde->lease, features);
384 		if (err)
385 			goto out_bad;
386 		/* inode */
387 		err = parse_reply_info_in(p, end, &rde->inode, features);
388 		if (err < 0)
389 			goto out_bad;
390 		/* ceph_readdir_prepopulate() will update it */
391 		rde->offset = 0;
392 		i++;
393 		num--;
394 	}
395 
396 done:
397 	/* Skip over any unrecognized fields */
398 	*p = end;
399 	return 0;
400 
401 bad:
402 	err = -EIO;
403 out_bad:
404 	pr_err("problem parsing dir contents %d\n", err);
405 	return err;
406 }
407 
408 /*
409  * parse fcntl F_GETLK results
410  */
parse_reply_info_filelock(void ** p,void * end,struct ceph_mds_reply_info_parsed * info,u64 features)411 static int parse_reply_info_filelock(void **p, void *end,
412 				     struct ceph_mds_reply_info_parsed *info,
413 				     u64 features)
414 {
415 	if (*p + sizeof(*info->filelock_reply) > end)
416 		goto bad;
417 
418 	info->filelock_reply = *p;
419 
420 	/* Skip over any unrecognized fields */
421 	*p = end;
422 	return 0;
423 bad:
424 	return -EIO;
425 }
426 
427 
428 #if BITS_PER_LONG == 64
429 
430 #define DELEGATED_INO_AVAILABLE		xa_mk_value(1)
431 
ceph_parse_deleg_inos(void ** p,void * end,struct ceph_mds_session * s)432 static int ceph_parse_deleg_inos(void **p, void *end,
433 				 struct ceph_mds_session *s)
434 {
435 	u32 sets;
436 
437 	ceph_decode_32_safe(p, end, sets, bad);
438 	dout("got %u sets of delegated inodes\n", sets);
439 	while (sets--) {
440 		u64 start, len;
441 
442 		ceph_decode_64_safe(p, end, start, bad);
443 		ceph_decode_64_safe(p, end, len, bad);
444 
445 		/* Don't accept a delegation of system inodes */
446 		if (start < CEPH_INO_SYSTEM_BASE) {
447 			pr_warn_ratelimited("ceph: ignoring reserved inode range delegation (start=0x%llx len=0x%llx)\n",
448 					start, len);
449 			continue;
450 		}
451 		while (len--) {
452 			int err = xa_insert(&s->s_delegated_inos, start++,
453 					    DELEGATED_INO_AVAILABLE,
454 					    GFP_KERNEL);
455 			if (!err) {
456 				dout("added delegated inode 0x%llx\n",
457 				     start - 1);
458 			} else if (err == -EBUSY) {
459 				pr_warn("MDS delegated inode 0x%llx more than once.\n",
460 					start - 1);
461 			} else {
462 				return err;
463 			}
464 		}
465 	}
466 	return 0;
467 bad:
468 	return -EIO;
469 }
470 
ceph_get_deleg_ino(struct ceph_mds_session * s)471 u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
472 {
473 	unsigned long ino;
474 	void *val;
475 
476 	xa_for_each(&s->s_delegated_inos, ino, val) {
477 		val = xa_erase(&s->s_delegated_inos, ino);
478 		if (val == DELEGATED_INO_AVAILABLE)
479 			return ino;
480 	}
481 	return 0;
482 }
483 
ceph_restore_deleg_ino(struct ceph_mds_session * s,u64 ino)484 int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
485 {
486 	return xa_insert(&s->s_delegated_inos, ino, DELEGATED_INO_AVAILABLE,
487 			 GFP_KERNEL);
488 }
489 #else /* BITS_PER_LONG == 64 */
490 /*
491  * FIXME: xarrays can't handle 64-bit indexes on a 32-bit arch. For now, just
492  * ignore delegated_inos on 32 bit arch. Maybe eventually add xarrays for top
493  * and bottom words?
494  */
ceph_parse_deleg_inos(void ** p,void * end,struct ceph_mds_session * s)495 static int ceph_parse_deleg_inos(void **p, void *end,
496 				 struct ceph_mds_session *s)
497 {
498 	u32 sets;
499 
500 	ceph_decode_32_safe(p, end, sets, bad);
501 	if (sets)
502 		ceph_decode_skip_n(p, end, sets * 2 * sizeof(__le64), bad);
503 	return 0;
504 bad:
505 	return -EIO;
506 }
507 
ceph_get_deleg_ino(struct ceph_mds_session * s)508 u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
509 {
510 	return 0;
511 }
512 
ceph_restore_deleg_ino(struct ceph_mds_session * s,u64 ino)513 int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
514 {
515 	return 0;
516 }
517 #endif /* BITS_PER_LONG == 64 */
518 
519 /*
520  * parse create results
521  */
parse_reply_info_create(void ** p,void * end,struct ceph_mds_reply_info_parsed * info,u64 features,struct ceph_mds_session * s)522 static int parse_reply_info_create(void **p, void *end,
523 				  struct ceph_mds_reply_info_parsed *info,
524 				  u64 features, struct ceph_mds_session *s)
525 {
526 	int ret;
527 
528 	if (features == (u64)-1 ||
529 	    (features & CEPH_FEATURE_REPLY_CREATE_INODE)) {
530 		if (*p == end) {
531 			/* Malformed reply? */
532 			info->has_create_ino = false;
533 		} else if (test_bit(CEPHFS_FEATURE_DELEG_INO, &s->s_features)) {
534 			info->has_create_ino = true;
535 			/* struct_v, struct_compat, and len */
536 			ceph_decode_skip_n(p, end, 2 + sizeof(u32), bad);
537 			ceph_decode_64_safe(p, end, info->ino, bad);
538 			ret = ceph_parse_deleg_inos(p, end, s);
539 			if (ret)
540 				return ret;
541 		} else {
542 			/* legacy */
543 			ceph_decode_64_safe(p, end, info->ino, bad);
544 			info->has_create_ino = true;
545 		}
546 	} else {
547 		if (*p != end)
548 			goto bad;
549 	}
550 
551 	/* Skip over any unrecognized fields */
552 	*p = end;
553 	return 0;
554 bad:
555 	return -EIO;
556 }
557 
parse_reply_info_getvxattr(void ** p,void * end,struct ceph_mds_reply_info_parsed * info,u64 features)558 static int parse_reply_info_getvxattr(void **p, void *end,
559 				      struct ceph_mds_reply_info_parsed *info,
560 				      u64 features)
561 {
562 	u32 value_len;
563 
564 	ceph_decode_skip_8(p, end, bad); /* skip current version: 1 */
565 	ceph_decode_skip_8(p, end, bad); /* skip first version: 1 */
566 	ceph_decode_skip_32(p, end, bad); /* skip payload length */
567 
568 	ceph_decode_32_safe(p, end, value_len, bad);
569 
570 	if (value_len == end - *p) {
571 	  info->xattr_info.xattr_value = *p;
572 	  info->xattr_info.xattr_value_len = value_len;
573 	  *p = end;
574 	  return value_len;
575 	}
576 bad:
577 	return -EIO;
578 }
579 
580 /*
581  * parse extra results
582  */
parse_reply_info_extra(void ** p,void * end,struct ceph_mds_reply_info_parsed * info,u64 features,struct ceph_mds_session * s)583 static int parse_reply_info_extra(void **p, void *end,
584 				  struct ceph_mds_reply_info_parsed *info,
585 				  u64 features, struct ceph_mds_session *s)
586 {
587 	u32 op = le32_to_cpu(info->head->op);
588 
589 	if (op == CEPH_MDS_OP_GETFILELOCK)
590 		return parse_reply_info_filelock(p, end, info, features);
591 	else if (op == CEPH_MDS_OP_READDIR || op == CEPH_MDS_OP_LSSNAP)
592 		return parse_reply_info_readdir(p, end, info, features);
593 	else if (op == CEPH_MDS_OP_CREATE)
594 		return parse_reply_info_create(p, end, info, features, s);
595 	else if (op == CEPH_MDS_OP_GETVXATTR)
596 		return parse_reply_info_getvxattr(p, end, info, features);
597 	else
598 		return -EIO;
599 }
600 
601 /*
602  * parse entire mds reply
603  */
parse_reply_info(struct ceph_mds_session * s,struct ceph_msg * msg,struct ceph_mds_reply_info_parsed * info,u64 features)604 static int parse_reply_info(struct ceph_mds_session *s, struct ceph_msg *msg,
605 			    struct ceph_mds_reply_info_parsed *info,
606 			    u64 features)
607 {
608 	void *p, *end;
609 	u32 len;
610 	int err;
611 
612 	info->head = msg->front.iov_base;
613 	p = msg->front.iov_base + sizeof(struct ceph_mds_reply_head);
614 	end = p + msg->front.iov_len - sizeof(struct ceph_mds_reply_head);
615 
616 	/* trace */
617 	ceph_decode_32_safe(&p, end, len, bad);
618 	if (len > 0) {
619 		ceph_decode_need(&p, end, len, bad);
620 		err = parse_reply_info_trace(&p, p+len, info, features);
621 		if (err < 0)
622 			goto out_bad;
623 	}
624 
625 	/* extra */
626 	ceph_decode_32_safe(&p, end, len, bad);
627 	if (len > 0) {
628 		ceph_decode_need(&p, end, len, bad);
629 		err = parse_reply_info_extra(&p, p+len, info, features, s);
630 		if (err < 0)
631 			goto out_bad;
632 	}
633 
634 	/* snap blob */
635 	ceph_decode_32_safe(&p, end, len, bad);
636 	info->snapblob_len = len;
637 	info->snapblob = p;
638 	p += len;
639 
640 	if (p != end)
641 		goto bad;
642 	return 0;
643 
644 bad:
645 	err = -EIO;
646 out_bad:
647 	pr_err("mds parse_reply err %d\n", err);
648 	ceph_msg_dump(msg);
649 	return err;
650 }
651 
destroy_reply_info(struct ceph_mds_reply_info_parsed * info)652 static void destroy_reply_info(struct ceph_mds_reply_info_parsed *info)
653 {
654 	if (!info->dir_entries)
655 		return;
656 	free_pages((unsigned long)info->dir_entries, get_order(info->dir_buf_size));
657 }
658 
659 /*
660  * In async unlink case the kclient won't wait for the first reply
661  * from MDS and just drop all the links and unhash the dentry and then
662  * succeeds immediately.
663  *
664  * For any new create/link/rename,etc requests followed by using the
665  * same file names we must wait for the first reply of the inflight
666  * unlink request, or the MDS possibly will fail these following
667  * requests with -EEXIST if the inflight async unlink request was
668  * delayed for some reasons.
669  *
670  * And the worst case is that for the none async openc request it will
671  * successfully open the file if the CDentry hasn't been unlinked yet,
672  * but later the previous delayed async unlink request will remove the
673  * CDenty. That means the just created file is possiblly deleted later
674  * by accident.
675  *
676  * We need to wait for the inflight async unlink requests to finish
677  * when creating new files/directories by using the same file names.
678  */
ceph_wait_on_conflict_unlink(struct dentry * dentry)679 int ceph_wait_on_conflict_unlink(struct dentry *dentry)
680 {
681 	struct ceph_fs_client *fsc = ceph_sb_to_client(dentry->d_sb);
682 	struct dentry *pdentry = dentry->d_parent;
683 	struct dentry *udentry, *found = NULL;
684 	struct ceph_dentry_info *di;
685 	struct qstr dname;
686 	u32 hash = dentry->d_name.hash;
687 	int err;
688 
689 	dname.name = dentry->d_name.name;
690 	dname.len = dentry->d_name.len;
691 
692 	rcu_read_lock();
693 	hash_for_each_possible_rcu(fsc->async_unlink_conflict, di,
694 				   hnode, hash) {
695 		udentry = di->dentry;
696 
697 		spin_lock(&udentry->d_lock);
698 		if (udentry->d_name.hash != hash)
699 			goto next;
700 		if (unlikely(udentry->d_parent != pdentry))
701 			goto next;
702 		if (!hash_hashed(&di->hnode))
703 			goto next;
704 
705 		if (!test_bit(CEPH_DENTRY_ASYNC_UNLINK_BIT, &di->flags))
706 			pr_warn("%s dentry %p:%pd async unlink bit is not set\n",
707 				__func__, dentry, dentry);
708 
709 		if (!d_same_name(udentry, pdentry, &dname))
710 			goto next;
711 
712 		found = dget_dlock(udentry);
713 		spin_unlock(&udentry->d_lock);
714 		break;
715 next:
716 		spin_unlock(&udentry->d_lock);
717 	}
718 	rcu_read_unlock();
719 
720 	if (likely(!found))
721 		return 0;
722 
723 	dout("%s dentry %p:%pd conflict with old %p:%pd\n", __func__,
724 	     dentry, dentry, found, found);
725 
726 	err = wait_on_bit(&di->flags, CEPH_DENTRY_ASYNC_UNLINK_BIT,
727 			  TASK_KILLABLE);
728 	dput(found);
729 	return err;
730 }
731 
732 
733 /*
734  * sessions
735  */
ceph_session_state_name(int s)736 const char *ceph_session_state_name(int s)
737 {
738 	switch (s) {
739 	case CEPH_MDS_SESSION_NEW: return "new";
740 	case CEPH_MDS_SESSION_OPENING: return "opening";
741 	case CEPH_MDS_SESSION_OPEN: return "open";
742 	case CEPH_MDS_SESSION_HUNG: return "hung";
743 	case CEPH_MDS_SESSION_CLOSING: return "closing";
744 	case CEPH_MDS_SESSION_CLOSED: return "closed";
745 	case CEPH_MDS_SESSION_RESTARTING: return "restarting";
746 	case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting";
747 	case CEPH_MDS_SESSION_REJECTED: return "rejected";
748 	default: return "???";
749 	}
750 }
751 
ceph_get_mds_session(struct ceph_mds_session * s)752 struct ceph_mds_session *ceph_get_mds_session(struct ceph_mds_session *s)
753 {
754 	if (refcount_inc_not_zero(&s->s_ref))
755 		return s;
756 	return NULL;
757 }
758 
ceph_put_mds_session(struct ceph_mds_session * s)759 void ceph_put_mds_session(struct ceph_mds_session *s)
760 {
761 	if (IS_ERR_OR_NULL(s))
762 		return;
763 
764 	if (refcount_dec_and_test(&s->s_ref)) {
765 		if (s->s_auth.authorizer)
766 			ceph_auth_destroy_authorizer(s->s_auth.authorizer);
767 		WARN_ON(mutex_is_locked(&s->s_mutex));
768 		xa_destroy(&s->s_delegated_inos);
769 		kfree(s);
770 	}
771 }
772 
773 /*
774  * called under mdsc->mutex
775  */
__ceph_lookup_mds_session(struct ceph_mds_client * mdsc,int mds)776 struct ceph_mds_session *__ceph_lookup_mds_session(struct ceph_mds_client *mdsc,
777 						   int mds)
778 {
779 	if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])
780 		return NULL;
781 	return ceph_get_mds_session(mdsc->sessions[mds]);
782 }
783 
__have_session(struct ceph_mds_client * mdsc,int mds)784 static bool __have_session(struct ceph_mds_client *mdsc, int mds)
785 {
786 	if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])
787 		return false;
788 	else
789 		return true;
790 }
791 
__verify_registered_session(struct ceph_mds_client * mdsc,struct ceph_mds_session * s)792 static int __verify_registered_session(struct ceph_mds_client *mdsc,
793 				       struct ceph_mds_session *s)
794 {
795 	if (s->s_mds >= mdsc->max_sessions ||
796 	    mdsc->sessions[s->s_mds] != s)
797 		return -ENOENT;
798 	return 0;
799 }
800 
801 /*
802  * create+register a new session for given mds.
803  * called under mdsc->mutex.
804  */
register_session(struct ceph_mds_client * mdsc,int mds)805 static struct ceph_mds_session *register_session(struct ceph_mds_client *mdsc,
806 						 int mds)
807 {
808 	struct ceph_mds_session *s;
809 
810 	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO)
811 		return ERR_PTR(-EIO);
812 
813 	if (mds >= mdsc->mdsmap->possible_max_rank)
814 		return ERR_PTR(-EINVAL);
815 
816 	s = kzalloc(sizeof(*s), GFP_NOFS);
817 	if (!s)
818 		return ERR_PTR(-ENOMEM);
819 
820 	if (mds >= mdsc->max_sessions) {
821 		int newmax = 1 << get_count_order(mds + 1);
822 		struct ceph_mds_session **sa;
823 
824 		dout("%s: realloc to %d\n", __func__, newmax);
825 		sa = kcalloc(newmax, sizeof(void *), GFP_NOFS);
826 		if (!sa)
827 			goto fail_realloc;
828 		if (mdsc->sessions) {
829 			memcpy(sa, mdsc->sessions,
830 			       mdsc->max_sessions * sizeof(void *));
831 			kfree(mdsc->sessions);
832 		}
833 		mdsc->sessions = sa;
834 		mdsc->max_sessions = newmax;
835 	}
836 
837 	dout("%s: mds%d\n", __func__, mds);
838 	s->s_mdsc = mdsc;
839 	s->s_mds = mds;
840 	s->s_state = CEPH_MDS_SESSION_NEW;
841 	mutex_init(&s->s_mutex);
842 
843 	ceph_con_init(&s->s_con, s, &mds_con_ops, &mdsc->fsc->client->msgr);
844 
845 	atomic_set(&s->s_cap_gen, 1);
846 	s->s_cap_ttl = jiffies - 1;
847 
848 	spin_lock_init(&s->s_cap_lock);
849 	INIT_LIST_HEAD(&s->s_caps);
850 	refcount_set(&s->s_ref, 1);
851 	INIT_LIST_HEAD(&s->s_waiting);
852 	INIT_LIST_HEAD(&s->s_unsafe);
853 	xa_init(&s->s_delegated_inos);
854 	INIT_LIST_HEAD(&s->s_cap_releases);
855 	INIT_WORK(&s->s_cap_release_work, ceph_cap_release_work);
856 
857 	INIT_LIST_HEAD(&s->s_cap_dirty);
858 	INIT_LIST_HEAD(&s->s_cap_flushing);
859 
860 	mdsc->sessions[mds] = s;
861 	atomic_inc(&mdsc->num_sessions);
862 	refcount_inc(&s->s_ref);  /* one ref to sessions[], one to caller */
863 
864 	ceph_con_open(&s->s_con, CEPH_ENTITY_TYPE_MDS, mds,
865 		      ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
866 
867 	return s;
868 
869 fail_realloc:
870 	kfree(s);
871 	return ERR_PTR(-ENOMEM);
872 }
873 
874 /*
875  * called under mdsc->mutex
876  */
__unregister_session(struct ceph_mds_client * mdsc,struct ceph_mds_session * s)877 static void __unregister_session(struct ceph_mds_client *mdsc,
878 			       struct ceph_mds_session *s)
879 {
880 	dout("__unregister_session mds%d %p\n", s->s_mds, s);
881 	BUG_ON(mdsc->sessions[s->s_mds] != s);
882 	mdsc->sessions[s->s_mds] = NULL;
883 	ceph_con_close(&s->s_con);
884 	ceph_put_mds_session(s);
885 	atomic_dec(&mdsc->num_sessions);
886 }
887 
888 /*
889  * drop session refs in request.
890  *
891  * should be last request ref, or hold mdsc->mutex
892  */
put_request_session(struct ceph_mds_request * req)893 static void put_request_session(struct ceph_mds_request *req)
894 {
895 	if (req->r_session) {
896 		ceph_put_mds_session(req->r_session);
897 		req->r_session = NULL;
898 	}
899 }
900 
ceph_mdsc_iterate_sessions(struct ceph_mds_client * mdsc,void (* cb)(struct ceph_mds_session *),bool check_state)901 void ceph_mdsc_iterate_sessions(struct ceph_mds_client *mdsc,
902 				void (*cb)(struct ceph_mds_session *),
903 				bool check_state)
904 {
905 	int mds;
906 
907 	mutex_lock(&mdsc->mutex);
908 	for (mds = 0; mds < mdsc->max_sessions; ++mds) {
909 		struct ceph_mds_session *s;
910 
911 		s = __ceph_lookup_mds_session(mdsc, mds);
912 		if (!s)
913 			continue;
914 
915 		if (check_state && !check_session_state(s)) {
916 			ceph_put_mds_session(s);
917 			continue;
918 		}
919 
920 		mutex_unlock(&mdsc->mutex);
921 		cb(s);
922 		ceph_put_mds_session(s);
923 		mutex_lock(&mdsc->mutex);
924 	}
925 	mutex_unlock(&mdsc->mutex);
926 }
927 
ceph_mdsc_release_request(struct kref * kref)928 void ceph_mdsc_release_request(struct kref *kref)
929 {
930 	struct ceph_mds_request *req = container_of(kref,
931 						    struct ceph_mds_request,
932 						    r_kref);
933 	ceph_mdsc_release_dir_caps_no_check(req);
934 	destroy_reply_info(&req->r_reply_info);
935 	if (req->r_request)
936 		ceph_msg_put(req->r_request);
937 	if (req->r_reply)
938 		ceph_msg_put(req->r_reply);
939 	if (req->r_inode) {
940 		ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
941 		iput(req->r_inode);
942 	}
943 	if (req->r_parent) {
944 		ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
945 		iput(req->r_parent);
946 	}
947 	iput(req->r_target_inode);
948 	if (req->r_dentry)
949 		dput(req->r_dentry);
950 	if (req->r_old_dentry)
951 		dput(req->r_old_dentry);
952 	if (req->r_old_dentry_dir) {
953 		/*
954 		 * track (and drop pins for) r_old_dentry_dir
955 		 * separately, since r_old_dentry's d_parent may have
956 		 * changed between the dir mutex being dropped and
957 		 * this request being freed.
958 		 */
959 		ceph_put_cap_refs(ceph_inode(req->r_old_dentry_dir),
960 				  CEPH_CAP_PIN);
961 		iput(req->r_old_dentry_dir);
962 	}
963 	kfree(req->r_path1);
964 	kfree(req->r_path2);
965 	put_cred(req->r_cred);
966 	if (req->r_pagelist)
967 		ceph_pagelist_release(req->r_pagelist);
968 	put_request_session(req);
969 	ceph_unreserve_caps(req->r_mdsc, &req->r_caps_reservation);
970 	WARN_ON_ONCE(!list_empty(&req->r_wait));
971 	kmem_cache_free(ceph_mds_request_cachep, req);
972 }
973 
DEFINE_RB_FUNCS(request,struct ceph_mds_request,r_tid,r_node)974 DEFINE_RB_FUNCS(request, struct ceph_mds_request, r_tid, r_node)
975 
976 /*
977  * lookup session, bump ref if found.
978  *
979  * called under mdsc->mutex.
980  */
981 static struct ceph_mds_request *
982 lookup_get_request(struct ceph_mds_client *mdsc, u64 tid)
983 {
984 	struct ceph_mds_request *req;
985 
986 	req = lookup_request(&mdsc->request_tree, tid);
987 	if (req)
988 		ceph_mdsc_get_request(req);
989 
990 	return req;
991 }
992 
993 /*
994  * Register an in-flight request, and assign a tid.  Link to directory
995  * are modifying (if any).
996  *
997  * Called under mdsc->mutex.
998  */
__register_request(struct ceph_mds_client * mdsc,struct ceph_mds_request * req,struct inode * dir)999 static void __register_request(struct ceph_mds_client *mdsc,
1000 			       struct ceph_mds_request *req,
1001 			       struct inode *dir)
1002 {
1003 	int ret = 0;
1004 
1005 	req->r_tid = ++mdsc->last_tid;
1006 	if (req->r_num_caps) {
1007 		ret = ceph_reserve_caps(mdsc, &req->r_caps_reservation,
1008 					req->r_num_caps);
1009 		if (ret < 0) {
1010 			pr_err("__register_request %p "
1011 			       "failed to reserve caps: %d\n", req, ret);
1012 			/* set req->r_err to fail early from __do_request */
1013 			req->r_err = ret;
1014 			return;
1015 		}
1016 	}
1017 	dout("__register_request %p tid %lld\n", req, req->r_tid);
1018 	ceph_mdsc_get_request(req);
1019 	insert_request(&mdsc->request_tree, req);
1020 
1021 	req->r_cred = get_current_cred();
1022 
1023 	if (mdsc->oldest_tid == 0 && req->r_op != CEPH_MDS_OP_SETFILELOCK)
1024 		mdsc->oldest_tid = req->r_tid;
1025 
1026 	if (dir) {
1027 		struct ceph_inode_info *ci = ceph_inode(dir);
1028 
1029 		ihold(dir);
1030 		req->r_unsafe_dir = dir;
1031 		spin_lock(&ci->i_unsafe_lock);
1032 		list_add_tail(&req->r_unsafe_dir_item, &ci->i_unsafe_dirops);
1033 		spin_unlock(&ci->i_unsafe_lock);
1034 	}
1035 }
1036 
__unregister_request(struct ceph_mds_client * mdsc,struct ceph_mds_request * req)1037 static void __unregister_request(struct ceph_mds_client *mdsc,
1038 				 struct ceph_mds_request *req)
1039 {
1040 	dout("__unregister_request %p tid %lld\n", req, req->r_tid);
1041 
1042 	/* Never leave an unregistered request on an unsafe list! */
1043 	list_del_init(&req->r_unsafe_item);
1044 
1045 	if (req->r_tid == mdsc->oldest_tid) {
1046 		struct rb_node *p = rb_next(&req->r_node);
1047 		mdsc->oldest_tid = 0;
1048 		while (p) {
1049 			struct ceph_mds_request *next_req =
1050 				rb_entry(p, struct ceph_mds_request, r_node);
1051 			if (next_req->r_op != CEPH_MDS_OP_SETFILELOCK) {
1052 				mdsc->oldest_tid = next_req->r_tid;
1053 				break;
1054 			}
1055 			p = rb_next(p);
1056 		}
1057 	}
1058 
1059 	erase_request(&mdsc->request_tree, req);
1060 
1061 	if (req->r_unsafe_dir) {
1062 		struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
1063 		spin_lock(&ci->i_unsafe_lock);
1064 		list_del_init(&req->r_unsafe_dir_item);
1065 		spin_unlock(&ci->i_unsafe_lock);
1066 	}
1067 	if (req->r_target_inode &&
1068 	    test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
1069 		struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
1070 		spin_lock(&ci->i_unsafe_lock);
1071 		list_del_init(&req->r_unsafe_target_item);
1072 		spin_unlock(&ci->i_unsafe_lock);
1073 	}
1074 
1075 	if (req->r_unsafe_dir) {
1076 		iput(req->r_unsafe_dir);
1077 		req->r_unsafe_dir = NULL;
1078 	}
1079 
1080 	complete_all(&req->r_safe_completion);
1081 
1082 	ceph_mdsc_put_request(req);
1083 }
1084 
1085 /*
1086  * Walk back up the dentry tree until we hit a dentry representing a
1087  * non-snapshot inode. We do this using the rcu_read_lock (which must be held
1088  * when calling this) to ensure that the objects won't disappear while we're
1089  * working with them. Once we hit a candidate dentry, we attempt to take a
1090  * reference to it, and return that as the result.
1091  */
get_nonsnap_parent(struct dentry * dentry)1092 static struct inode *get_nonsnap_parent(struct dentry *dentry)
1093 {
1094 	struct inode *inode = NULL;
1095 
1096 	while (dentry && !IS_ROOT(dentry)) {
1097 		inode = d_inode_rcu(dentry);
1098 		if (!inode || ceph_snap(inode) == CEPH_NOSNAP)
1099 			break;
1100 		dentry = dentry->d_parent;
1101 	}
1102 	if (inode)
1103 		inode = igrab(inode);
1104 	return inode;
1105 }
1106 
1107 /*
1108  * Choose mds to send request to next.  If there is a hint set in the
1109  * request (e.g., due to a prior forward hint from the mds), use that.
1110  * Otherwise, consult frag tree and/or caps to identify the
1111  * appropriate mds.  If all else fails, choose randomly.
1112  *
1113  * Called under mdsc->mutex.
1114  */
__choose_mds(struct ceph_mds_client * mdsc,struct ceph_mds_request * req,bool * random)1115 static int __choose_mds(struct ceph_mds_client *mdsc,
1116 			struct ceph_mds_request *req,
1117 			bool *random)
1118 {
1119 	struct inode *inode;
1120 	struct ceph_inode_info *ci;
1121 	struct ceph_cap *cap;
1122 	int mode = req->r_direct_mode;
1123 	int mds = -1;
1124 	u32 hash = req->r_direct_hash;
1125 	bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
1126 
1127 	if (random)
1128 		*random = false;
1129 
1130 	/*
1131 	 * is there a specific mds we should try?  ignore hint if we have
1132 	 * no session and the mds is not up (active or recovering).
1133 	 */
1134 	if (req->r_resend_mds >= 0 &&
1135 	    (__have_session(mdsc, req->r_resend_mds) ||
1136 	     ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) {
1137 		dout("%s using resend_mds mds%d\n", __func__,
1138 		     req->r_resend_mds);
1139 		return req->r_resend_mds;
1140 	}
1141 
1142 	if (mode == USE_RANDOM_MDS)
1143 		goto random;
1144 
1145 	inode = NULL;
1146 	if (req->r_inode) {
1147 		if (ceph_snap(req->r_inode) != CEPH_SNAPDIR) {
1148 			inode = req->r_inode;
1149 			ihold(inode);
1150 		} else {
1151 			/* req->r_dentry is non-null for LSSNAP request */
1152 			rcu_read_lock();
1153 			inode = get_nonsnap_parent(req->r_dentry);
1154 			rcu_read_unlock();
1155 			dout("%s using snapdir's parent %p\n", __func__, inode);
1156 		}
1157 	} else if (req->r_dentry) {
1158 		/* ignore race with rename; old or new d_parent is okay */
1159 		struct dentry *parent;
1160 		struct inode *dir;
1161 
1162 		rcu_read_lock();
1163 		parent = READ_ONCE(req->r_dentry->d_parent);
1164 		dir = req->r_parent ? : d_inode_rcu(parent);
1165 
1166 		if (!dir || dir->i_sb != mdsc->fsc->sb) {
1167 			/*  not this fs or parent went negative */
1168 			inode = d_inode(req->r_dentry);
1169 			if (inode)
1170 				ihold(inode);
1171 		} else if (ceph_snap(dir) != CEPH_NOSNAP) {
1172 			/* direct snapped/virtual snapdir requests
1173 			 * based on parent dir inode */
1174 			inode = get_nonsnap_parent(parent);
1175 			dout("%s using nonsnap parent %p\n", __func__, inode);
1176 		} else {
1177 			/* dentry target */
1178 			inode = d_inode(req->r_dentry);
1179 			if (!inode || mode == USE_AUTH_MDS) {
1180 				/* dir + name */
1181 				inode = igrab(dir);
1182 				hash = ceph_dentry_hash(dir, req->r_dentry);
1183 				is_hash = true;
1184 			} else {
1185 				ihold(inode);
1186 			}
1187 		}
1188 		rcu_read_unlock();
1189 	}
1190 
1191 	dout("%s %p is_hash=%d (0x%x) mode %d\n", __func__, inode, (int)is_hash,
1192 	     hash, mode);
1193 	if (!inode)
1194 		goto random;
1195 	ci = ceph_inode(inode);
1196 
1197 	if (is_hash && S_ISDIR(inode->i_mode)) {
1198 		struct ceph_inode_frag frag;
1199 		int found;
1200 
1201 		ceph_choose_frag(ci, hash, &frag, &found);
1202 		if (found) {
1203 			if (mode == USE_ANY_MDS && frag.ndist > 0) {
1204 				u8 r;
1205 
1206 				/* choose a random replica */
1207 				get_random_bytes(&r, 1);
1208 				r %= frag.ndist;
1209 				mds = frag.dist[r];
1210 				dout("%s %p %llx.%llx frag %u mds%d (%d/%d)\n",
1211 				     __func__, inode, ceph_vinop(inode),
1212 				     frag.frag, mds, (int)r, frag.ndist);
1213 				if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
1214 				    CEPH_MDS_STATE_ACTIVE &&
1215 				    !ceph_mdsmap_is_laggy(mdsc->mdsmap, mds))
1216 					goto out;
1217 			}
1218 
1219 			/* since this file/dir wasn't known to be
1220 			 * replicated, then we want to look for the
1221 			 * authoritative mds. */
1222 			if (frag.mds >= 0) {
1223 				/* choose auth mds */
1224 				mds = frag.mds;
1225 				dout("%s %p %llx.%llx frag %u mds%d (auth)\n",
1226 				     __func__, inode, ceph_vinop(inode),
1227 				     frag.frag, mds);
1228 				if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
1229 				    CEPH_MDS_STATE_ACTIVE) {
1230 					if (!ceph_mdsmap_is_laggy(mdsc->mdsmap,
1231 								  mds))
1232 						goto out;
1233 				}
1234 			}
1235 			mode = USE_AUTH_MDS;
1236 		}
1237 	}
1238 
1239 	spin_lock(&ci->i_ceph_lock);
1240 	cap = NULL;
1241 	if (mode == USE_AUTH_MDS)
1242 		cap = ci->i_auth_cap;
1243 	if (!cap && !RB_EMPTY_ROOT(&ci->i_caps))
1244 		cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node);
1245 	if (!cap) {
1246 		spin_unlock(&ci->i_ceph_lock);
1247 		iput(inode);
1248 		goto random;
1249 	}
1250 	mds = cap->session->s_mds;
1251 	dout("%s %p %llx.%llx mds%d (%scap %p)\n", __func__,
1252 	     inode, ceph_vinop(inode), mds,
1253 	     cap == ci->i_auth_cap ? "auth " : "", cap);
1254 	spin_unlock(&ci->i_ceph_lock);
1255 out:
1256 	iput(inode);
1257 	return mds;
1258 
1259 random:
1260 	if (random)
1261 		*random = true;
1262 
1263 	mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap);
1264 	dout("%s chose random mds%d\n", __func__, mds);
1265 	return mds;
1266 }
1267 
1268 
1269 /*
1270  * session messages
1271  */
ceph_create_session_msg(u32 op,u64 seq)1272 struct ceph_msg *ceph_create_session_msg(u32 op, u64 seq)
1273 {
1274 	struct ceph_msg *msg;
1275 	struct ceph_mds_session_head *h;
1276 
1277 	msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS,
1278 			   false);
1279 	if (!msg) {
1280 		pr_err("ENOMEM creating session %s msg\n",
1281 		       ceph_session_op_name(op));
1282 		return NULL;
1283 	}
1284 	h = msg->front.iov_base;
1285 	h->op = cpu_to_le32(op);
1286 	h->seq = cpu_to_le64(seq);
1287 
1288 	return msg;
1289 }
1290 
1291 static const unsigned char feature_bits[] = CEPHFS_FEATURES_CLIENT_SUPPORTED;
1292 #define FEATURE_BYTES(c) (DIV_ROUND_UP((size_t)feature_bits[c - 1] + 1, 64) * 8)
encode_supported_features(void ** p,void * end)1293 static int encode_supported_features(void **p, void *end)
1294 {
1295 	static const size_t count = ARRAY_SIZE(feature_bits);
1296 
1297 	if (count > 0) {
1298 		size_t i;
1299 		size_t size = FEATURE_BYTES(count);
1300 		unsigned long bit;
1301 
1302 		if (WARN_ON_ONCE(*p + 4 + size > end))
1303 			return -ERANGE;
1304 
1305 		ceph_encode_32(p, size);
1306 		memset(*p, 0, size);
1307 		for (i = 0; i < count; i++) {
1308 			bit = feature_bits[i];
1309 			((unsigned char *)(*p))[bit / 8] |= BIT(bit % 8);
1310 		}
1311 		*p += size;
1312 	} else {
1313 		if (WARN_ON_ONCE(*p + 4 > end))
1314 			return -ERANGE;
1315 
1316 		ceph_encode_32(p, 0);
1317 	}
1318 
1319 	return 0;
1320 }
1321 
1322 static const unsigned char metric_bits[] = CEPHFS_METRIC_SPEC_CLIENT_SUPPORTED;
1323 #define METRIC_BYTES(cnt) (DIV_ROUND_UP((size_t)metric_bits[cnt - 1] + 1, 64) * 8)
encode_metric_spec(void ** p,void * end)1324 static int encode_metric_spec(void **p, void *end)
1325 {
1326 	static const size_t count = ARRAY_SIZE(metric_bits);
1327 
1328 	/* header */
1329 	if (WARN_ON_ONCE(*p + 2 > end))
1330 		return -ERANGE;
1331 
1332 	ceph_encode_8(p, 1); /* version */
1333 	ceph_encode_8(p, 1); /* compat */
1334 
1335 	if (count > 0) {
1336 		size_t i;
1337 		size_t size = METRIC_BYTES(count);
1338 
1339 		if (WARN_ON_ONCE(*p + 4 + 4 + size > end))
1340 			return -ERANGE;
1341 
1342 		/* metric spec info length */
1343 		ceph_encode_32(p, 4 + size);
1344 
1345 		/* metric spec */
1346 		ceph_encode_32(p, size);
1347 		memset(*p, 0, size);
1348 		for (i = 0; i < count; i++)
1349 			((unsigned char *)(*p))[i / 8] |= BIT(metric_bits[i] % 8);
1350 		*p += size;
1351 	} else {
1352 		if (WARN_ON_ONCE(*p + 4 + 4 > end))
1353 			return -ERANGE;
1354 
1355 		/* metric spec info length */
1356 		ceph_encode_32(p, 4);
1357 		/* metric spec */
1358 		ceph_encode_32(p, 0);
1359 	}
1360 
1361 	return 0;
1362 }
1363 
1364 /*
1365  * session message, specialization for CEPH_SESSION_REQUEST_OPEN
1366  * to include additional client metadata fields.
1367  */
create_session_open_msg(struct ceph_mds_client * mdsc,u64 seq)1368 static struct ceph_msg *create_session_open_msg(struct ceph_mds_client *mdsc, u64 seq)
1369 {
1370 	struct ceph_msg *msg;
1371 	struct ceph_mds_session_head *h;
1372 	int i;
1373 	int extra_bytes = 0;
1374 	int metadata_key_count = 0;
1375 	struct ceph_options *opt = mdsc->fsc->client->options;
1376 	struct ceph_mount_options *fsopt = mdsc->fsc->mount_options;
1377 	size_t size, count;
1378 	void *p, *end;
1379 	int ret;
1380 
1381 	const char* metadata[][2] = {
1382 		{"hostname", mdsc->nodename},
1383 		{"kernel_version", init_utsname()->release},
1384 		{"entity_id", opt->name ? : ""},
1385 		{"root", fsopt->server_path ? : "/"},
1386 		{NULL, NULL}
1387 	};
1388 
1389 	/* Calculate serialized length of metadata */
1390 	extra_bytes = 4;  /* map length */
1391 	for (i = 0; metadata[i][0]; ++i) {
1392 		extra_bytes += 8 + strlen(metadata[i][0]) +
1393 			strlen(metadata[i][1]);
1394 		metadata_key_count++;
1395 	}
1396 
1397 	/* supported feature */
1398 	size = 0;
1399 	count = ARRAY_SIZE(feature_bits);
1400 	if (count > 0)
1401 		size = FEATURE_BYTES(count);
1402 	extra_bytes += 4 + size;
1403 
1404 	/* metric spec */
1405 	size = 0;
1406 	count = ARRAY_SIZE(metric_bits);
1407 	if (count > 0)
1408 		size = METRIC_BYTES(count);
1409 	extra_bytes += 2 + 4 + 4 + size;
1410 
1411 	/* Allocate the message */
1412 	msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h) + extra_bytes,
1413 			   GFP_NOFS, false);
1414 	if (!msg) {
1415 		pr_err("ENOMEM creating session open msg\n");
1416 		return ERR_PTR(-ENOMEM);
1417 	}
1418 	p = msg->front.iov_base;
1419 	end = p + msg->front.iov_len;
1420 
1421 	h = p;
1422 	h->op = cpu_to_le32(CEPH_SESSION_REQUEST_OPEN);
1423 	h->seq = cpu_to_le64(seq);
1424 
1425 	/*
1426 	 * Serialize client metadata into waiting buffer space, using
1427 	 * the format that userspace expects for map<string, string>
1428 	 *
1429 	 * ClientSession messages with metadata are v4
1430 	 */
1431 	msg->hdr.version = cpu_to_le16(4);
1432 	msg->hdr.compat_version = cpu_to_le16(1);
1433 
1434 	/* The write pointer, following the session_head structure */
1435 	p += sizeof(*h);
1436 
1437 	/* Number of entries in the map */
1438 	ceph_encode_32(&p, metadata_key_count);
1439 
1440 	/* Two length-prefixed strings for each entry in the map */
1441 	for (i = 0; metadata[i][0]; ++i) {
1442 		size_t const key_len = strlen(metadata[i][0]);
1443 		size_t const val_len = strlen(metadata[i][1]);
1444 
1445 		ceph_encode_32(&p, key_len);
1446 		memcpy(p, metadata[i][0], key_len);
1447 		p += key_len;
1448 		ceph_encode_32(&p, val_len);
1449 		memcpy(p, metadata[i][1], val_len);
1450 		p += val_len;
1451 	}
1452 
1453 	ret = encode_supported_features(&p, end);
1454 	if (ret) {
1455 		pr_err("encode_supported_features failed!\n");
1456 		ceph_msg_put(msg);
1457 		return ERR_PTR(ret);
1458 	}
1459 
1460 	ret = encode_metric_spec(&p, end);
1461 	if (ret) {
1462 		pr_err("encode_metric_spec failed!\n");
1463 		ceph_msg_put(msg);
1464 		return ERR_PTR(ret);
1465 	}
1466 
1467 	msg->front.iov_len = p - msg->front.iov_base;
1468 	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
1469 
1470 	return msg;
1471 }
1472 
1473 /*
1474  * send session open request.
1475  *
1476  * called under mdsc->mutex
1477  */
__open_session(struct ceph_mds_client * mdsc,struct ceph_mds_session * session)1478 static int __open_session(struct ceph_mds_client *mdsc,
1479 			  struct ceph_mds_session *session)
1480 {
1481 	struct ceph_msg *msg;
1482 	int mstate;
1483 	int mds = session->s_mds;
1484 
1485 	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO)
1486 		return -EIO;
1487 
1488 	/* wait for mds to go active? */
1489 	mstate = ceph_mdsmap_get_state(mdsc->mdsmap, mds);
1490 	dout("open_session to mds%d (%s)\n", mds,
1491 	     ceph_mds_state_name(mstate));
1492 	session->s_state = CEPH_MDS_SESSION_OPENING;
1493 	session->s_renew_requested = jiffies;
1494 
1495 	/* send connect message */
1496 	msg = create_session_open_msg(mdsc, session->s_seq);
1497 	if (IS_ERR(msg))
1498 		return PTR_ERR(msg);
1499 	ceph_con_send(&session->s_con, msg);
1500 	return 0;
1501 }
1502 
1503 /*
1504  * open sessions for any export targets for the given mds
1505  *
1506  * called under mdsc->mutex
1507  */
1508 static struct ceph_mds_session *
__open_export_target_session(struct ceph_mds_client * mdsc,int target)1509 __open_export_target_session(struct ceph_mds_client *mdsc, int target)
1510 {
1511 	struct ceph_mds_session *session;
1512 	int ret;
1513 
1514 	session = __ceph_lookup_mds_session(mdsc, target);
1515 	if (!session) {
1516 		session = register_session(mdsc, target);
1517 		if (IS_ERR(session))
1518 			return session;
1519 	}
1520 	if (session->s_state == CEPH_MDS_SESSION_NEW ||
1521 	    session->s_state == CEPH_MDS_SESSION_CLOSING) {
1522 		ret = __open_session(mdsc, session);
1523 		if (ret)
1524 			return ERR_PTR(ret);
1525 	}
1526 
1527 	return session;
1528 }
1529 
1530 struct ceph_mds_session *
ceph_mdsc_open_export_target_session(struct ceph_mds_client * mdsc,int target)1531 ceph_mdsc_open_export_target_session(struct ceph_mds_client *mdsc, int target)
1532 {
1533 	struct ceph_mds_session *session;
1534 
1535 	dout("open_export_target_session to mds%d\n", target);
1536 
1537 	mutex_lock(&mdsc->mutex);
1538 	session = __open_export_target_session(mdsc, target);
1539 	mutex_unlock(&mdsc->mutex);
1540 
1541 	return session;
1542 }
1543 
__open_export_target_sessions(struct ceph_mds_client * mdsc,struct ceph_mds_session * session)1544 static void __open_export_target_sessions(struct ceph_mds_client *mdsc,
1545 					  struct ceph_mds_session *session)
1546 {
1547 	struct ceph_mds_info *mi;
1548 	struct ceph_mds_session *ts;
1549 	int i, mds = session->s_mds;
1550 
1551 	if (mds >= mdsc->mdsmap->possible_max_rank)
1552 		return;
1553 
1554 	mi = &mdsc->mdsmap->m_info[mds];
1555 	dout("open_export_target_sessions for mds%d (%d targets)\n",
1556 	     session->s_mds, mi->num_export_targets);
1557 
1558 	for (i = 0; i < mi->num_export_targets; i++) {
1559 		ts = __open_export_target_session(mdsc, mi->export_targets[i]);
1560 		ceph_put_mds_session(ts);
1561 	}
1562 }
1563 
ceph_mdsc_open_export_target_sessions(struct ceph_mds_client * mdsc,struct ceph_mds_session * session)1564 void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,
1565 					   struct ceph_mds_session *session)
1566 {
1567 	mutex_lock(&mdsc->mutex);
1568 	__open_export_target_sessions(mdsc, session);
1569 	mutex_unlock(&mdsc->mutex);
1570 }
1571 
1572 /*
1573  * session caps
1574  */
1575 
detach_cap_releases(struct ceph_mds_session * session,struct list_head * target)1576 static void detach_cap_releases(struct ceph_mds_session *session,
1577 				struct list_head *target)
1578 {
1579 	lockdep_assert_held(&session->s_cap_lock);
1580 
1581 	list_splice_init(&session->s_cap_releases, target);
1582 	session->s_num_cap_releases = 0;
1583 	dout("dispose_cap_releases mds%d\n", session->s_mds);
1584 }
1585 
dispose_cap_releases(struct ceph_mds_client * mdsc,struct list_head * dispose)1586 static void dispose_cap_releases(struct ceph_mds_client *mdsc,
1587 				 struct list_head *dispose)
1588 {
1589 	while (!list_empty(dispose)) {
1590 		struct ceph_cap *cap;
1591 		/* zero out the in-progress message */
1592 		cap = list_first_entry(dispose, struct ceph_cap, session_caps);
1593 		list_del(&cap->session_caps);
1594 		ceph_put_cap(mdsc, cap);
1595 	}
1596 }
1597 
cleanup_session_requests(struct ceph_mds_client * mdsc,struct ceph_mds_session * session)1598 static void cleanup_session_requests(struct ceph_mds_client *mdsc,
1599 				     struct ceph_mds_session *session)
1600 {
1601 	struct ceph_mds_request *req;
1602 	struct rb_node *p;
1603 
1604 	dout("cleanup_session_requests mds%d\n", session->s_mds);
1605 	mutex_lock(&mdsc->mutex);
1606 	while (!list_empty(&session->s_unsafe)) {
1607 		req = list_first_entry(&session->s_unsafe,
1608 				       struct ceph_mds_request, r_unsafe_item);
1609 		pr_warn_ratelimited(" dropping unsafe request %llu\n",
1610 				    req->r_tid);
1611 		if (req->r_target_inode)
1612 			mapping_set_error(req->r_target_inode->i_mapping, -EIO);
1613 		if (req->r_unsafe_dir)
1614 			mapping_set_error(req->r_unsafe_dir->i_mapping, -EIO);
1615 		__unregister_request(mdsc, req);
1616 	}
1617 	/* zero r_attempts, so kick_requests() will re-send requests */
1618 	p = rb_first(&mdsc->request_tree);
1619 	while (p) {
1620 		req = rb_entry(p, struct ceph_mds_request, r_node);
1621 		p = rb_next(p);
1622 		if (req->r_session &&
1623 		    req->r_session->s_mds == session->s_mds)
1624 			req->r_attempts = 0;
1625 	}
1626 	mutex_unlock(&mdsc->mutex);
1627 }
1628 
1629 /*
1630  * Helper to safely iterate over all caps associated with a session, with
1631  * special care taken to handle a racing __ceph_remove_cap().
1632  *
1633  * Caller must hold session s_mutex.
1634  */
ceph_iterate_session_caps(struct ceph_mds_session * session,int (* cb)(struct inode *,int mds,void *),void * arg)1635 int ceph_iterate_session_caps(struct ceph_mds_session *session,
1636 			      int (*cb)(struct inode *, int mds, void *),
1637 			      void *arg)
1638 {
1639 	struct list_head *p;
1640 	struct ceph_cap *cap;
1641 	struct inode *inode, *last_inode = NULL;
1642 	struct ceph_cap *old_cap = NULL;
1643 	int ret;
1644 
1645 	dout("iterate_session_caps %p mds%d\n", session, session->s_mds);
1646 	spin_lock(&session->s_cap_lock);
1647 	p = session->s_caps.next;
1648 	while (p != &session->s_caps) {
1649 		int mds;
1650 
1651 		cap = list_entry(p, struct ceph_cap, session_caps);
1652 		inode = igrab(&cap->ci->netfs.inode);
1653 		if (!inode) {
1654 			p = p->next;
1655 			continue;
1656 		}
1657 		session->s_cap_iterator = cap;
1658 		mds = cap->mds;
1659 		spin_unlock(&session->s_cap_lock);
1660 
1661 		if (last_inode) {
1662 			iput(last_inode);
1663 			last_inode = NULL;
1664 		}
1665 		if (old_cap) {
1666 			ceph_put_cap(session->s_mdsc, old_cap);
1667 			old_cap = NULL;
1668 		}
1669 
1670 		ret = cb(inode, mds, arg);
1671 		last_inode = inode;
1672 
1673 		spin_lock(&session->s_cap_lock);
1674 		p = p->next;
1675 		if (!cap->ci) {
1676 			dout("iterate_session_caps  finishing cap %p removal\n",
1677 			     cap);
1678 			BUG_ON(cap->session != session);
1679 			cap->session = NULL;
1680 			list_del_init(&cap->session_caps);
1681 			session->s_nr_caps--;
1682 			atomic64_dec(&session->s_mdsc->metric.total_caps);
1683 			if (cap->queue_release)
1684 				__ceph_queue_cap_release(session, cap);
1685 			else
1686 				old_cap = cap;  /* put_cap it w/o locks held */
1687 		}
1688 		if (ret < 0)
1689 			goto out;
1690 	}
1691 	ret = 0;
1692 out:
1693 	session->s_cap_iterator = NULL;
1694 	spin_unlock(&session->s_cap_lock);
1695 
1696 	iput(last_inode);
1697 	if (old_cap)
1698 		ceph_put_cap(session->s_mdsc, old_cap);
1699 
1700 	return ret;
1701 }
1702 
remove_session_caps_cb(struct inode * inode,int mds,void * arg)1703 static int remove_session_caps_cb(struct inode *inode, int mds, void *arg)
1704 {
1705 	struct ceph_inode_info *ci = ceph_inode(inode);
1706 	bool invalidate = false;
1707 	struct ceph_cap *cap;
1708 	int iputs = 0;
1709 
1710 	spin_lock(&ci->i_ceph_lock);
1711 	cap = __get_cap_for_mds(ci, mds);
1712 	if (cap) {
1713 		dout(" removing cap %p, ci is %p, inode is %p\n",
1714 		     cap, ci, &ci->netfs.inode);
1715 
1716 		iputs = ceph_purge_inode_cap(inode, cap, &invalidate);
1717 	}
1718 	spin_unlock(&ci->i_ceph_lock);
1719 
1720 	if (cap)
1721 		wake_up_all(&ci->i_cap_wq);
1722 	if (invalidate)
1723 		ceph_queue_invalidate(inode);
1724 	while (iputs--)
1725 		iput(inode);
1726 	return 0;
1727 }
1728 
1729 /*
1730  * caller must hold session s_mutex
1731  */
remove_session_caps(struct ceph_mds_session * session)1732 static void remove_session_caps(struct ceph_mds_session *session)
1733 {
1734 	struct ceph_fs_client *fsc = session->s_mdsc->fsc;
1735 	struct super_block *sb = fsc->sb;
1736 	LIST_HEAD(dispose);
1737 
1738 	dout("remove_session_caps on %p\n", session);
1739 	ceph_iterate_session_caps(session, remove_session_caps_cb, fsc);
1740 
1741 	wake_up_all(&fsc->mdsc->cap_flushing_wq);
1742 
1743 	spin_lock(&session->s_cap_lock);
1744 	if (session->s_nr_caps > 0) {
1745 		struct inode *inode;
1746 		struct ceph_cap *cap, *prev = NULL;
1747 		struct ceph_vino vino;
1748 		/*
1749 		 * iterate_session_caps() skips inodes that are being
1750 		 * deleted, we need to wait until deletions are complete.
1751 		 * __wait_on_freeing_inode() is designed for the job,
1752 		 * but it is not exported, so use lookup inode function
1753 		 * to access it.
1754 		 */
1755 		while (!list_empty(&session->s_caps)) {
1756 			cap = list_entry(session->s_caps.next,
1757 					 struct ceph_cap, session_caps);
1758 			if (cap == prev)
1759 				break;
1760 			prev = cap;
1761 			vino = cap->ci->i_vino;
1762 			spin_unlock(&session->s_cap_lock);
1763 
1764 			inode = ceph_find_inode(sb, vino);
1765 			iput(inode);
1766 
1767 			spin_lock(&session->s_cap_lock);
1768 		}
1769 	}
1770 
1771 	// drop cap expires and unlock s_cap_lock
1772 	detach_cap_releases(session, &dispose);
1773 
1774 	BUG_ON(session->s_nr_caps > 0);
1775 	BUG_ON(!list_empty(&session->s_cap_flushing));
1776 	spin_unlock(&session->s_cap_lock);
1777 	dispose_cap_releases(session->s_mdsc, &dispose);
1778 }
1779 
1780 enum {
1781 	RECONNECT,
1782 	RENEWCAPS,
1783 	FORCE_RO,
1784 };
1785 
1786 /*
1787  * wake up any threads waiting on this session's caps.  if the cap is
1788  * old (didn't get renewed on the client reconnect), remove it now.
1789  *
1790  * caller must hold s_mutex.
1791  */
wake_up_session_cb(struct inode * inode,int mds,void * arg)1792 static int wake_up_session_cb(struct inode *inode, int mds, void *arg)
1793 {
1794 	struct ceph_inode_info *ci = ceph_inode(inode);
1795 	unsigned long ev = (unsigned long)arg;
1796 
1797 	if (ev == RECONNECT) {
1798 		spin_lock(&ci->i_ceph_lock);
1799 		ci->i_wanted_max_size = 0;
1800 		ci->i_requested_max_size = 0;
1801 		spin_unlock(&ci->i_ceph_lock);
1802 	} else if (ev == RENEWCAPS) {
1803 		struct ceph_cap *cap;
1804 
1805 		spin_lock(&ci->i_ceph_lock);
1806 		cap = __get_cap_for_mds(ci, mds);
1807 		/* mds did not re-issue stale cap */
1808 		if (cap && cap->cap_gen < atomic_read(&cap->session->s_cap_gen))
1809 			cap->issued = cap->implemented = CEPH_CAP_PIN;
1810 		spin_unlock(&ci->i_ceph_lock);
1811 	} else if (ev == FORCE_RO) {
1812 	}
1813 	wake_up_all(&ci->i_cap_wq);
1814 	return 0;
1815 }
1816 
wake_up_session_caps(struct ceph_mds_session * session,int ev)1817 static void wake_up_session_caps(struct ceph_mds_session *session, int ev)
1818 {
1819 	dout("wake_up_session_caps %p mds%d\n", session, session->s_mds);
1820 	ceph_iterate_session_caps(session, wake_up_session_cb,
1821 				  (void *)(unsigned long)ev);
1822 }
1823 
1824 /*
1825  * Send periodic message to MDS renewing all currently held caps.  The
1826  * ack will reset the expiration for all caps from this session.
1827  *
1828  * caller holds s_mutex
1829  */
send_renew_caps(struct ceph_mds_client * mdsc,struct ceph_mds_session * session)1830 static int send_renew_caps(struct ceph_mds_client *mdsc,
1831 			   struct ceph_mds_session *session)
1832 {
1833 	struct ceph_msg *msg;
1834 	int state;
1835 
1836 	if (time_after_eq(jiffies, session->s_cap_ttl) &&
1837 	    time_after_eq(session->s_cap_ttl, session->s_renew_requested))
1838 		pr_info("mds%d caps stale\n", session->s_mds);
1839 	session->s_renew_requested = jiffies;
1840 
1841 	/* do not try to renew caps until a recovering mds has reconnected
1842 	 * with its clients. */
1843 	state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds);
1844 	if (state < CEPH_MDS_STATE_RECONNECT) {
1845 		dout("send_renew_caps ignoring mds%d (%s)\n",
1846 		     session->s_mds, ceph_mds_state_name(state));
1847 		return 0;
1848 	}
1849 
1850 	dout("send_renew_caps to mds%d (%s)\n", session->s_mds,
1851 		ceph_mds_state_name(state));
1852 	msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS,
1853 				      ++session->s_renew_seq);
1854 	if (!msg)
1855 		return -ENOMEM;
1856 	ceph_con_send(&session->s_con, msg);
1857 	return 0;
1858 }
1859 
send_flushmsg_ack(struct ceph_mds_client * mdsc,struct ceph_mds_session * session,u64 seq)1860 static int send_flushmsg_ack(struct ceph_mds_client *mdsc,
1861 			     struct ceph_mds_session *session, u64 seq)
1862 {
1863 	struct ceph_msg *msg;
1864 
1865 	dout("send_flushmsg_ack to mds%d (%s)s seq %lld\n",
1866 	     session->s_mds, ceph_session_state_name(session->s_state), seq);
1867 	msg = ceph_create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq);
1868 	if (!msg)
1869 		return -ENOMEM;
1870 	ceph_con_send(&session->s_con, msg);
1871 	return 0;
1872 }
1873 
1874 
1875 /*
1876  * Note new cap ttl, and any transition from stale -> not stale (fresh?).
1877  *
1878  * Called under session->s_mutex
1879  */
renewed_caps(struct ceph_mds_client * mdsc,struct ceph_mds_session * session,int is_renew)1880 static void renewed_caps(struct ceph_mds_client *mdsc,
1881 			 struct ceph_mds_session *session, int is_renew)
1882 {
1883 	int was_stale;
1884 	int wake = 0;
1885 
1886 	spin_lock(&session->s_cap_lock);
1887 	was_stale = is_renew && time_after_eq(jiffies, session->s_cap_ttl);
1888 
1889 	session->s_cap_ttl = session->s_renew_requested +
1890 		mdsc->mdsmap->m_session_timeout*HZ;
1891 
1892 	if (was_stale) {
1893 		if (time_before(jiffies, session->s_cap_ttl)) {
1894 			pr_info("mds%d caps renewed\n", session->s_mds);
1895 			wake = 1;
1896 		} else {
1897 			pr_info("mds%d caps still stale\n", session->s_mds);
1898 		}
1899 	}
1900 	dout("renewed_caps mds%d ttl now %lu, was %s, now %s\n",
1901 	     session->s_mds, session->s_cap_ttl, was_stale ? "stale" : "fresh",
1902 	     time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh");
1903 	spin_unlock(&session->s_cap_lock);
1904 
1905 	if (wake)
1906 		wake_up_session_caps(session, RENEWCAPS);
1907 }
1908 
1909 /*
1910  * send a session close request
1911  */
request_close_session(struct ceph_mds_session * session)1912 static int request_close_session(struct ceph_mds_session *session)
1913 {
1914 	struct ceph_msg *msg;
1915 
1916 	dout("request_close_session mds%d state %s seq %lld\n",
1917 	     session->s_mds, ceph_session_state_name(session->s_state),
1918 	     session->s_seq);
1919 	msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_CLOSE,
1920 				      session->s_seq);
1921 	if (!msg)
1922 		return -ENOMEM;
1923 	ceph_con_send(&session->s_con, msg);
1924 	return 1;
1925 }
1926 
1927 /*
1928  * Called with s_mutex held.
1929  */
__close_session(struct ceph_mds_client * mdsc,struct ceph_mds_session * session)1930 static int __close_session(struct ceph_mds_client *mdsc,
1931 			 struct ceph_mds_session *session)
1932 {
1933 	if (session->s_state >= CEPH_MDS_SESSION_CLOSING)
1934 		return 0;
1935 	session->s_state = CEPH_MDS_SESSION_CLOSING;
1936 	return request_close_session(session);
1937 }
1938 
drop_negative_children(struct dentry * dentry)1939 static bool drop_negative_children(struct dentry *dentry)
1940 {
1941 	struct dentry *child;
1942 	bool all_negative = true;
1943 
1944 	if (!d_is_dir(dentry))
1945 		goto out;
1946 
1947 	spin_lock(&dentry->d_lock);
1948 	list_for_each_entry(child, &dentry->d_subdirs, d_child) {
1949 		if (d_really_is_positive(child)) {
1950 			all_negative = false;
1951 			break;
1952 		}
1953 	}
1954 	spin_unlock(&dentry->d_lock);
1955 
1956 	if (all_negative)
1957 		shrink_dcache_parent(dentry);
1958 out:
1959 	return all_negative;
1960 }
1961 
1962 /*
1963  * Trim old(er) caps.
1964  *
1965  * Because we can't cache an inode without one or more caps, we do
1966  * this indirectly: if a cap is unused, we prune its aliases, at which
1967  * point the inode will hopefully get dropped to.
1968  *
1969  * Yes, this is a bit sloppy.  Our only real goal here is to respond to
1970  * memory pressure from the MDS, though, so it needn't be perfect.
1971  */
trim_caps_cb(struct inode * inode,int mds,void * arg)1972 static int trim_caps_cb(struct inode *inode, int mds, void *arg)
1973 {
1974 	int *remaining = arg;
1975 	struct ceph_inode_info *ci = ceph_inode(inode);
1976 	int used, wanted, oissued, mine;
1977 	struct ceph_cap *cap;
1978 
1979 	if (*remaining <= 0)
1980 		return -1;
1981 
1982 	spin_lock(&ci->i_ceph_lock);
1983 	cap = __get_cap_for_mds(ci, mds);
1984 	if (!cap) {
1985 		spin_unlock(&ci->i_ceph_lock);
1986 		return 0;
1987 	}
1988 	mine = cap->issued | cap->implemented;
1989 	used = __ceph_caps_used(ci);
1990 	wanted = __ceph_caps_file_wanted(ci);
1991 	oissued = __ceph_caps_issued_other(ci, cap);
1992 
1993 	dout("trim_caps_cb %p cap %p mine %s oissued %s used %s wanted %s\n",
1994 	     inode, cap, ceph_cap_string(mine), ceph_cap_string(oissued),
1995 	     ceph_cap_string(used), ceph_cap_string(wanted));
1996 	if (cap == ci->i_auth_cap) {
1997 		if (ci->i_dirty_caps || ci->i_flushing_caps ||
1998 		    !list_empty(&ci->i_cap_snaps))
1999 			goto out;
2000 		if ((used | wanted) & CEPH_CAP_ANY_WR)
2001 			goto out;
2002 		/* Note: it's possible that i_filelock_ref becomes non-zero
2003 		 * after dropping auth caps. It doesn't hurt because reply
2004 		 * of lock mds request will re-add auth caps. */
2005 		if (atomic_read(&ci->i_filelock_ref) > 0)
2006 			goto out;
2007 	}
2008 	/* The inode has cached pages, but it's no longer used.
2009 	 * we can safely drop it */
2010 	if (S_ISREG(inode->i_mode) &&
2011 	    wanted == 0 && used == CEPH_CAP_FILE_CACHE &&
2012 	    !(oissued & CEPH_CAP_FILE_CACHE)) {
2013 	  used = 0;
2014 	  oissued = 0;
2015 	}
2016 	if ((used | wanted) & ~oissued & mine)
2017 		goto out;   /* we need these caps */
2018 
2019 	if (oissued) {
2020 		/* we aren't the only cap.. just remove us */
2021 		ceph_remove_cap(cap, true);
2022 		(*remaining)--;
2023 	} else {
2024 		struct dentry *dentry;
2025 		/* try dropping referring dentries */
2026 		spin_unlock(&ci->i_ceph_lock);
2027 		dentry = d_find_any_alias(inode);
2028 		if (dentry && drop_negative_children(dentry)) {
2029 			int count;
2030 			dput(dentry);
2031 			d_prune_aliases(inode);
2032 			count = atomic_read(&inode->i_count);
2033 			if (count == 1)
2034 				(*remaining)--;
2035 			dout("trim_caps_cb %p cap %p pruned, count now %d\n",
2036 			     inode, cap, count);
2037 		} else {
2038 			dput(dentry);
2039 		}
2040 		return 0;
2041 	}
2042 
2043 out:
2044 	spin_unlock(&ci->i_ceph_lock);
2045 	return 0;
2046 }
2047 
2048 /*
2049  * Trim session cap count down to some max number.
2050  */
ceph_trim_caps(struct ceph_mds_client * mdsc,struct ceph_mds_session * session,int max_caps)2051 int ceph_trim_caps(struct ceph_mds_client *mdsc,
2052 		   struct ceph_mds_session *session,
2053 		   int max_caps)
2054 {
2055 	int trim_caps = session->s_nr_caps - max_caps;
2056 
2057 	dout("trim_caps mds%d start: %d / %d, trim %d\n",
2058 	     session->s_mds, session->s_nr_caps, max_caps, trim_caps);
2059 	if (trim_caps > 0) {
2060 		int remaining = trim_caps;
2061 
2062 		ceph_iterate_session_caps(session, trim_caps_cb, &remaining);
2063 		dout("trim_caps mds%d done: %d / %d, trimmed %d\n",
2064 		     session->s_mds, session->s_nr_caps, max_caps,
2065 			trim_caps - remaining);
2066 	}
2067 
2068 	ceph_flush_cap_releases(mdsc, session);
2069 	return 0;
2070 }
2071 
check_caps_flush(struct ceph_mds_client * mdsc,u64 want_flush_tid)2072 static int check_caps_flush(struct ceph_mds_client *mdsc,
2073 			    u64 want_flush_tid)
2074 {
2075 	int ret = 1;
2076 
2077 	spin_lock(&mdsc->cap_dirty_lock);
2078 	if (!list_empty(&mdsc->cap_flush_list)) {
2079 		struct ceph_cap_flush *cf =
2080 			list_first_entry(&mdsc->cap_flush_list,
2081 					 struct ceph_cap_flush, g_list);
2082 		if (cf->tid <= want_flush_tid) {
2083 			dout("check_caps_flush still flushing tid "
2084 			     "%llu <= %llu\n", cf->tid, want_flush_tid);
2085 			ret = 0;
2086 		}
2087 	}
2088 	spin_unlock(&mdsc->cap_dirty_lock);
2089 	return ret;
2090 }
2091 
2092 /*
2093  * flush all dirty inode data to disk.
2094  *
2095  * returns true if we've flushed through want_flush_tid
2096  */
wait_caps_flush(struct ceph_mds_client * mdsc,u64 want_flush_tid)2097 static void wait_caps_flush(struct ceph_mds_client *mdsc,
2098 			    u64 want_flush_tid)
2099 {
2100 	dout("check_caps_flush want %llu\n", want_flush_tid);
2101 
2102 	wait_event(mdsc->cap_flushing_wq,
2103 		   check_caps_flush(mdsc, want_flush_tid));
2104 
2105 	dout("check_caps_flush ok, flushed thru %llu\n", want_flush_tid);
2106 }
2107 
2108 /*
2109  * called under s_mutex
2110  */
ceph_send_cap_releases(struct ceph_mds_client * mdsc,struct ceph_mds_session * session)2111 static void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
2112 				   struct ceph_mds_session *session)
2113 {
2114 	struct ceph_msg *msg = NULL;
2115 	struct ceph_mds_cap_release *head;
2116 	struct ceph_mds_cap_item *item;
2117 	struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc;
2118 	struct ceph_cap *cap;
2119 	LIST_HEAD(tmp_list);
2120 	int num_cap_releases;
2121 	__le32	barrier, *cap_barrier;
2122 
2123 	down_read(&osdc->lock);
2124 	barrier = cpu_to_le32(osdc->epoch_barrier);
2125 	up_read(&osdc->lock);
2126 
2127 	spin_lock(&session->s_cap_lock);
2128 again:
2129 	list_splice_init(&session->s_cap_releases, &tmp_list);
2130 	num_cap_releases = session->s_num_cap_releases;
2131 	session->s_num_cap_releases = 0;
2132 	spin_unlock(&session->s_cap_lock);
2133 
2134 	while (!list_empty(&tmp_list)) {
2135 		if (!msg) {
2136 			msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE,
2137 					PAGE_SIZE, GFP_NOFS, false);
2138 			if (!msg)
2139 				goto out_err;
2140 			head = msg->front.iov_base;
2141 			head->num = cpu_to_le32(0);
2142 			msg->front.iov_len = sizeof(*head);
2143 
2144 			msg->hdr.version = cpu_to_le16(2);
2145 			msg->hdr.compat_version = cpu_to_le16(1);
2146 		}
2147 
2148 		cap = list_first_entry(&tmp_list, struct ceph_cap,
2149 					session_caps);
2150 		list_del(&cap->session_caps);
2151 		num_cap_releases--;
2152 
2153 		head = msg->front.iov_base;
2154 		put_unaligned_le32(get_unaligned_le32(&head->num) + 1,
2155 				   &head->num);
2156 		item = msg->front.iov_base + msg->front.iov_len;
2157 		item->ino = cpu_to_le64(cap->cap_ino);
2158 		item->cap_id = cpu_to_le64(cap->cap_id);
2159 		item->migrate_seq = cpu_to_le32(cap->mseq);
2160 		item->seq = cpu_to_le32(cap->issue_seq);
2161 		msg->front.iov_len += sizeof(*item);
2162 
2163 		ceph_put_cap(mdsc, cap);
2164 
2165 		if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) {
2166 			// Append cap_barrier field
2167 			cap_barrier = msg->front.iov_base + msg->front.iov_len;
2168 			*cap_barrier = barrier;
2169 			msg->front.iov_len += sizeof(*cap_barrier);
2170 
2171 			msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2172 			dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
2173 			ceph_con_send(&session->s_con, msg);
2174 			msg = NULL;
2175 		}
2176 	}
2177 
2178 	BUG_ON(num_cap_releases != 0);
2179 
2180 	spin_lock(&session->s_cap_lock);
2181 	if (!list_empty(&session->s_cap_releases))
2182 		goto again;
2183 	spin_unlock(&session->s_cap_lock);
2184 
2185 	if (msg) {
2186 		// Append cap_barrier field
2187 		cap_barrier = msg->front.iov_base + msg->front.iov_len;
2188 		*cap_barrier = barrier;
2189 		msg->front.iov_len += sizeof(*cap_barrier);
2190 
2191 		msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2192 		dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
2193 		ceph_con_send(&session->s_con, msg);
2194 	}
2195 	return;
2196 out_err:
2197 	pr_err("send_cap_releases mds%d, failed to allocate message\n",
2198 		session->s_mds);
2199 	spin_lock(&session->s_cap_lock);
2200 	list_splice(&tmp_list, &session->s_cap_releases);
2201 	session->s_num_cap_releases += num_cap_releases;
2202 	spin_unlock(&session->s_cap_lock);
2203 }
2204 
ceph_cap_release_work(struct work_struct * work)2205 static void ceph_cap_release_work(struct work_struct *work)
2206 {
2207 	struct ceph_mds_session *session =
2208 		container_of(work, struct ceph_mds_session, s_cap_release_work);
2209 
2210 	mutex_lock(&session->s_mutex);
2211 	if (session->s_state == CEPH_MDS_SESSION_OPEN ||
2212 	    session->s_state == CEPH_MDS_SESSION_HUNG)
2213 		ceph_send_cap_releases(session->s_mdsc, session);
2214 	mutex_unlock(&session->s_mutex);
2215 	ceph_put_mds_session(session);
2216 }
2217 
ceph_flush_cap_releases(struct ceph_mds_client * mdsc,struct ceph_mds_session * session)2218 void ceph_flush_cap_releases(struct ceph_mds_client *mdsc,
2219 		             struct ceph_mds_session *session)
2220 {
2221 	if (mdsc->stopping)
2222 		return;
2223 
2224 	ceph_get_mds_session(session);
2225 	if (queue_work(mdsc->fsc->cap_wq,
2226 		       &session->s_cap_release_work)) {
2227 		dout("cap release work queued\n");
2228 	} else {
2229 		ceph_put_mds_session(session);
2230 		dout("failed to queue cap release work\n");
2231 	}
2232 }
2233 
2234 /*
2235  * caller holds session->s_cap_lock
2236  */
__ceph_queue_cap_release(struct ceph_mds_session * session,struct ceph_cap * cap)2237 void __ceph_queue_cap_release(struct ceph_mds_session *session,
2238 			      struct ceph_cap *cap)
2239 {
2240 	list_add_tail(&cap->session_caps, &session->s_cap_releases);
2241 	session->s_num_cap_releases++;
2242 
2243 	if (!(session->s_num_cap_releases % CEPH_CAPS_PER_RELEASE))
2244 		ceph_flush_cap_releases(session->s_mdsc, session);
2245 }
2246 
ceph_cap_reclaim_work(struct work_struct * work)2247 static void ceph_cap_reclaim_work(struct work_struct *work)
2248 {
2249 	struct ceph_mds_client *mdsc =
2250 		container_of(work, struct ceph_mds_client, cap_reclaim_work);
2251 	int ret = ceph_trim_dentries(mdsc);
2252 	if (ret == -EAGAIN)
2253 		ceph_queue_cap_reclaim_work(mdsc);
2254 }
2255 
ceph_queue_cap_reclaim_work(struct ceph_mds_client * mdsc)2256 void ceph_queue_cap_reclaim_work(struct ceph_mds_client *mdsc)
2257 {
2258 	if (mdsc->stopping)
2259 		return;
2260 
2261         if (queue_work(mdsc->fsc->cap_wq, &mdsc->cap_reclaim_work)) {
2262                 dout("caps reclaim work queued\n");
2263         } else {
2264                 dout("failed to queue caps release work\n");
2265         }
2266 }
2267 
ceph_reclaim_caps_nr(struct ceph_mds_client * mdsc,int nr)2268 void ceph_reclaim_caps_nr(struct ceph_mds_client *mdsc, int nr)
2269 {
2270 	int val;
2271 	if (!nr)
2272 		return;
2273 	val = atomic_add_return(nr, &mdsc->cap_reclaim_pending);
2274 	if ((val % CEPH_CAPS_PER_RELEASE) < nr) {
2275 		atomic_set(&mdsc->cap_reclaim_pending, 0);
2276 		ceph_queue_cap_reclaim_work(mdsc);
2277 	}
2278 }
2279 
2280 /*
2281  * requests
2282  */
2283 
ceph_alloc_readdir_reply_buffer(struct ceph_mds_request * req,struct inode * dir)2284 int ceph_alloc_readdir_reply_buffer(struct ceph_mds_request *req,
2285 				    struct inode *dir)
2286 {
2287 	struct ceph_inode_info *ci = ceph_inode(dir);
2288 	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
2289 	struct ceph_mount_options *opt = req->r_mdsc->fsc->mount_options;
2290 	size_t size = sizeof(struct ceph_mds_reply_dir_entry);
2291 	unsigned int num_entries;
2292 	int order;
2293 
2294 	spin_lock(&ci->i_ceph_lock);
2295 	num_entries = ci->i_files + ci->i_subdirs;
2296 	spin_unlock(&ci->i_ceph_lock);
2297 	num_entries = max(num_entries, 1U);
2298 	num_entries = min(num_entries, opt->max_readdir);
2299 
2300 	order = get_order(size * num_entries);
2301 	while (order >= 0) {
2302 		rinfo->dir_entries = (void*)__get_free_pages(GFP_KERNEL |
2303 							     __GFP_NOWARN |
2304 							     __GFP_ZERO,
2305 							     order);
2306 		if (rinfo->dir_entries)
2307 			break;
2308 		order--;
2309 	}
2310 	if (!rinfo->dir_entries)
2311 		return -ENOMEM;
2312 
2313 	num_entries = (PAGE_SIZE << order) / size;
2314 	num_entries = min(num_entries, opt->max_readdir);
2315 
2316 	rinfo->dir_buf_size = PAGE_SIZE << order;
2317 	req->r_num_caps = num_entries + 1;
2318 	req->r_args.readdir.max_entries = cpu_to_le32(num_entries);
2319 	req->r_args.readdir.max_bytes = cpu_to_le32(opt->max_readdir_bytes);
2320 	return 0;
2321 }
2322 
2323 /*
2324  * Create an mds request.
2325  */
2326 struct ceph_mds_request *
ceph_mdsc_create_request(struct ceph_mds_client * mdsc,int op,int mode)2327 ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
2328 {
2329 	struct ceph_mds_request *req;
2330 
2331 	req = kmem_cache_zalloc(ceph_mds_request_cachep, GFP_NOFS);
2332 	if (!req)
2333 		return ERR_PTR(-ENOMEM);
2334 
2335 	mutex_init(&req->r_fill_mutex);
2336 	req->r_mdsc = mdsc;
2337 	req->r_started = jiffies;
2338 	req->r_start_latency = ktime_get();
2339 	req->r_resend_mds = -1;
2340 	INIT_LIST_HEAD(&req->r_unsafe_dir_item);
2341 	INIT_LIST_HEAD(&req->r_unsafe_target_item);
2342 	req->r_fmode = -1;
2343 	req->r_feature_needed = -1;
2344 	kref_init(&req->r_kref);
2345 	RB_CLEAR_NODE(&req->r_node);
2346 	INIT_LIST_HEAD(&req->r_wait);
2347 	init_completion(&req->r_completion);
2348 	init_completion(&req->r_safe_completion);
2349 	INIT_LIST_HEAD(&req->r_unsafe_item);
2350 
2351 	ktime_get_coarse_real_ts64(&req->r_stamp);
2352 
2353 	req->r_op = op;
2354 	req->r_direct_mode = mode;
2355 	return req;
2356 }
2357 
2358 /*
2359  * return oldest (lowest) request, tid in request tree, 0 if none.
2360  *
2361  * called under mdsc->mutex.
2362  */
__get_oldest_req(struct ceph_mds_client * mdsc)2363 static struct ceph_mds_request *__get_oldest_req(struct ceph_mds_client *mdsc)
2364 {
2365 	if (RB_EMPTY_ROOT(&mdsc->request_tree))
2366 		return NULL;
2367 	return rb_entry(rb_first(&mdsc->request_tree),
2368 			struct ceph_mds_request, r_node);
2369 }
2370 
__get_oldest_tid(struct ceph_mds_client * mdsc)2371 static inline  u64 __get_oldest_tid(struct ceph_mds_client *mdsc)
2372 {
2373 	return mdsc->oldest_tid;
2374 }
2375 
2376 /*
2377  * Build a dentry's path.  Allocate on heap; caller must kfree.  Based
2378  * on build_path_from_dentry in fs/cifs/dir.c.
2379  *
2380  * If @stop_on_nosnap, generate path relative to the first non-snapped
2381  * inode.
2382  *
2383  * Encode hidden .snap dirs as a double /, i.e.
2384  *   foo/.snap/bar -> foo//bar
2385  */
ceph_mdsc_build_path(struct dentry * dentry,int * plen,u64 * pbase,int stop_on_nosnap)2386 char *ceph_mdsc_build_path(struct dentry *dentry, int *plen, u64 *pbase,
2387 			   int stop_on_nosnap)
2388 {
2389 	struct dentry *temp;
2390 	char *path;
2391 	int pos;
2392 	unsigned seq;
2393 	u64 base;
2394 
2395 	if (!dentry)
2396 		return ERR_PTR(-EINVAL);
2397 
2398 	path = __getname();
2399 	if (!path)
2400 		return ERR_PTR(-ENOMEM);
2401 retry:
2402 	pos = PATH_MAX - 1;
2403 	path[pos] = '\0';
2404 
2405 	seq = read_seqbegin(&rename_lock);
2406 	rcu_read_lock();
2407 	temp = dentry;
2408 	for (;;) {
2409 		struct inode *inode;
2410 
2411 		spin_lock(&temp->d_lock);
2412 		inode = d_inode(temp);
2413 		if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {
2414 			dout("build_path path+%d: %p SNAPDIR\n",
2415 			     pos, temp);
2416 		} else if (stop_on_nosnap && inode && dentry != temp &&
2417 			   ceph_snap(inode) == CEPH_NOSNAP) {
2418 			spin_unlock(&temp->d_lock);
2419 			pos++; /* get rid of any prepended '/' */
2420 			break;
2421 		} else {
2422 			pos -= temp->d_name.len;
2423 			if (pos < 0) {
2424 				spin_unlock(&temp->d_lock);
2425 				break;
2426 			}
2427 			memcpy(path + pos, temp->d_name.name, temp->d_name.len);
2428 		}
2429 		spin_unlock(&temp->d_lock);
2430 		temp = READ_ONCE(temp->d_parent);
2431 
2432 		/* Are we at the root? */
2433 		if (IS_ROOT(temp))
2434 			break;
2435 
2436 		/* Are we out of buffer? */
2437 		if (--pos < 0)
2438 			break;
2439 
2440 		path[pos] = '/';
2441 	}
2442 	base = ceph_ino(d_inode(temp));
2443 	rcu_read_unlock();
2444 
2445 	if (read_seqretry(&rename_lock, seq))
2446 		goto retry;
2447 
2448 	if (pos < 0) {
2449 		/*
2450 		 * A rename didn't occur, but somehow we didn't end up where
2451 		 * we thought we would. Throw a warning and try again.
2452 		 */
2453 		pr_warn("build_path did not end path lookup where "
2454 			"expected, pos is %d\n", pos);
2455 		goto retry;
2456 	}
2457 
2458 	*pbase = base;
2459 	*plen = PATH_MAX - 1 - pos;
2460 	dout("build_path on %p %d built %llx '%.*s'\n",
2461 	     dentry, d_count(dentry), base, *plen, path + pos);
2462 	return path + pos;
2463 }
2464 
build_dentry_path(struct dentry * dentry,struct inode * dir,const char ** ppath,int * ppathlen,u64 * pino,bool * pfreepath,bool parent_locked)2465 static int build_dentry_path(struct dentry *dentry, struct inode *dir,
2466 			     const char **ppath, int *ppathlen, u64 *pino,
2467 			     bool *pfreepath, bool parent_locked)
2468 {
2469 	char *path;
2470 
2471 	rcu_read_lock();
2472 	if (!dir)
2473 		dir = d_inode_rcu(dentry->d_parent);
2474 	if (dir && parent_locked && ceph_snap(dir) == CEPH_NOSNAP) {
2475 		*pino = ceph_ino(dir);
2476 		rcu_read_unlock();
2477 		*ppath = dentry->d_name.name;
2478 		*ppathlen = dentry->d_name.len;
2479 		return 0;
2480 	}
2481 	rcu_read_unlock();
2482 	path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
2483 	if (IS_ERR(path))
2484 		return PTR_ERR(path);
2485 	*ppath = path;
2486 	*pfreepath = true;
2487 	return 0;
2488 }
2489 
build_inode_path(struct inode * inode,const char ** ppath,int * ppathlen,u64 * pino,bool * pfreepath)2490 static int build_inode_path(struct inode *inode,
2491 			    const char **ppath, int *ppathlen, u64 *pino,
2492 			    bool *pfreepath)
2493 {
2494 	struct dentry *dentry;
2495 	char *path;
2496 
2497 	if (ceph_snap(inode) == CEPH_NOSNAP) {
2498 		*pino = ceph_ino(inode);
2499 		*ppathlen = 0;
2500 		return 0;
2501 	}
2502 	dentry = d_find_alias(inode);
2503 	path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
2504 	dput(dentry);
2505 	if (IS_ERR(path))
2506 		return PTR_ERR(path);
2507 	*ppath = path;
2508 	*pfreepath = true;
2509 	return 0;
2510 }
2511 
2512 /*
2513  * request arguments may be specified via an inode *, a dentry *, or
2514  * an explicit ino+path.
2515  */
set_request_path_attr(struct inode * rinode,struct dentry * rdentry,struct inode * rdiri,const char * rpath,u64 rino,const char ** ppath,int * pathlen,u64 * ino,bool * freepath,bool parent_locked)2516 static int set_request_path_attr(struct inode *rinode, struct dentry *rdentry,
2517 				  struct inode *rdiri, const char *rpath,
2518 				  u64 rino, const char **ppath, int *pathlen,
2519 				  u64 *ino, bool *freepath, bool parent_locked)
2520 {
2521 	int r = 0;
2522 
2523 	if (rinode) {
2524 		r = build_inode_path(rinode, ppath, pathlen, ino, freepath);
2525 		dout(" inode %p %llx.%llx\n", rinode, ceph_ino(rinode),
2526 		     ceph_snap(rinode));
2527 	} else if (rdentry) {
2528 		r = build_dentry_path(rdentry, rdiri, ppath, pathlen, ino,
2529 					freepath, parent_locked);
2530 		dout(" dentry %p %llx/%.*s\n", rdentry, *ino, *pathlen,
2531 		     *ppath);
2532 	} else if (rpath || rino) {
2533 		*ino = rino;
2534 		*ppath = rpath;
2535 		*pathlen = rpath ? strlen(rpath) : 0;
2536 		dout(" path %.*s\n", *pathlen, rpath);
2537 	}
2538 
2539 	return r;
2540 }
2541 
encode_timestamp_and_gids(void ** p,const struct ceph_mds_request * req)2542 static void encode_timestamp_and_gids(void **p,
2543 				      const struct ceph_mds_request *req)
2544 {
2545 	struct ceph_timespec ts;
2546 	int i;
2547 
2548 	ceph_encode_timespec64(&ts, &req->r_stamp);
2549 	ceph_encode_copy(p, &ts, sizeof(ts));
2550 
2551 	/* gid_list */
2552 	ceph_encode_32(p, req->r_cred->group_info->ngroups);
2553 	for (i = 0; i < req->r_cred->group_info->ngroups; i++)
2554 		ceph_encode_64(p, from_kgid(&init_user_ns,
2555 					    req->r_cred->group_info->gid[i]));
2556 }
2557 
2558 /*
2559  * called under mdsc->mutex
2560  */
create_request_message(struct ceph_mds_session * session,struct ceph_mds_request * req,bool drop_cap_releases)2561 static struct ceph_msg *create_request_message(struct ceph_mds_session *session,
2562 					       struct ceph_mds_request *req,
2563 					       bool drop_cap_releases)
2564 {
2565 	int mds = session->s_mds;
2566 	struct ceph_mds_client *mdsc = session->s_mdsc;
2567 	struct ceph_msg *msg;
2568 	struct ceph_mds_request_head_old *head;
2569 	const char *path1 = NULL;
2570 	const char *path2 = NULL;
2571 	u64 ino1 = 0, ino2 = 0;
2572 	int pathlen1 = 0, pathlen2 = 0;
2573 	bool freepath1 = false, freepath2 = false;
2574 	int len;
2575 	u16 releases;
2576 	void *p, *end;
2577 	int ret;
2578 	bool legacy = !(session->s_con.peer_features & CEPH_FEATURE_FS_BTIME);
2579 
2580 	ret = set_request_path_attr(req->r_inode, req->r_dentry,
2581 			      req->r_parent, req->r_path1, req->r_ino1.ino,
2582 			      &path1, &pathlen1, &ino1, &freepath1,
2583 			      test_bit(CEPH_MDS_R_PARENT_LOCKED,
2584 					&req->r_req_flags));
2585 	if (ret < 0) {
2586 		msg = ERR_PTR(ret);
2587 		goto out;
2588 	}
2589 
2590 	/* If r_old_dentry is set, then assume that its parent is locked */
2591 	ret = set_request_path_attr(NULL, req->r_old_dentry,
2592 			      req->r_old_dentry_dir,
2593 			      req->r_path2, req->r_ino2.ino,
2594 			      &path2, &pathlen2, &ino2, &freepath2, true);
2595 	if (ret < 0) {
2596 		msg = ERR_PTR(ret);
2597 		goto out_free1;
2598 	}
2599 
2600 	len = legacy ? sizeof(*head) : sizeof(struct ceph_mds_request_head);
2601 	len += pathlen1 + pathlen2 + 2*(1 + sizeof(u32) + sizeof(u64)) +
2602 		sizeof(struct ceph_timespec);
2603 	len += sizeof(u32) + (sizeof(u64) * req->r_cred->group_info->ngroups);
2604 
2605 	/* calculate (max) length for cap releases */
2606 	len += sizeof(struct ceph_mds_request_release) *
2607 		(!!req->r_inode_drop + !!req->r_dentry_drop +
2608 		 !!req->r_old_inode_drop + !!req->r_old_dentry_drop);
2609 
2610 	if (req->r_dentry_drop)
2611 		len += pathlen1;
2612 	if (req->r_old_dentry_drop)
2613 		len += pathlen2;
2614 
2615 	msg = ceph_msg_new2(CEPH_MSG_CLIENT_REQUEST, len, 1, GFP_NOFS, false);
2616 	if (!msg) {
2617 		msg = ERR_PTR(-ENOMEM);
2618 		goto out_free2;
2619 	}
2620 
2621 	msg->hdr.tid = cpu_to_le64(req->r_tid);
2622 
2623 	/*
2624 	 * The old ceph_mds_request_head didn't contain a version field, and
2625 	 * one was added when we moved the message version from 3->4.
2626 	 */
2627 	if (legacy) {
2628 		msg->hdr.version = cpu_to_le16(3);
2629 		head = msg->front.iov_base;
2630 		p = msg->front.iov_base + sizeof(*head);
2631 	} else {
2632 		struct ceph_mds_request_head *new_head = msg->front.iov_base;
2633 
2634 		msg->hdr.version = cpu_to_le16(4);
2635 		new_head->version = cpu_to_le16(CEPH_MDS_REQUEST_HEAD_VERSION);
2636 		head = (struct ceph_mds_request_head_old *)&new_head->oldest_client_tid;
2637 		p = msg->front.iov_base + sizeof(*new_head);
2638 	}
2639 
2640 	end = msg->front.iov_base + msg->front.iov_len;
2641 
2642 	head->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);
2643 	head->op = cpu_to_le32(req->r_op);
2644 	head->caller_uid = cpu_to_le32(from_kuid(&init_user_ns,
2645 						 req->r_cred->fsuid));
2646 	head->caller_gid = cpu_to_le32(from_kgid(&init_user_ns,
2647 						 req->r_cred->fsgid));
2648 	head->ino = cpu_to_le64(req->r_deleg_ino);
2649 	head->args = req->r_args;
2650 
2651 	ceph_encode_filepath(&p, end, ino1, path1);
2652 	ceph_encode_filepath(&p, end, ino2, path2);
2653 
2654 	/* make note of release offset, in case we need to replay */
2655 	req->r_request_release_offset = p - msg->front.iov_base;
2656 
2657 	/* cap releases */
2658 	releases = 0;
2659 	if (req->r_inode_drop)
2660 		releases += ceph_encode_inode_release(&p,
2661 		      req->r_inode ? req->r_inode : d_inode(req->r_dentry),
2662 		      mds, req->r_inode_drop, req->r_inode_unless,
2663 		      req->r_op == CEPH_MDS_OP_READDIR);
2664 	if (req->r_dentry_drop)
2665 		releases += ceph_encode_dentry_release(&p, req->r_dentry,
2666 				req->r_parent, mds, req->r_dentry_drop,
2667 				req->r_dentry_unless);
2668 	if (req->r_old_dentry_drop)
2669 		releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
2670 				req->r_old_dentry_dir, mds,
2671 				req->r_old_dentry_drop,
2672 				req->r_old_dentry_unless);
2673 	if (req->r_old_inode_drop)
2674 		releases += ceph_encode_inode_release(&p,
2675 		      d_inode(req->r_old_dentry),
2676 		      mds, req->r_old_inode_drop, req->r_old_inode_unless, 0);
2677 
2678 	if (drop_cap_releases) {
2679 		releases = 0;
2680 		p = msg->front.iov_base + req->r_request_release_offset;
2681 	}
2682 
2683 	head->num_releases = cpu_to_le16(releases);
2684 
2685 	encode_timestamp_and_gids(&p, req);
2686 
2687 	if (WARN_ON_ONCE(p > end)) {
2688 		ceph_msg_put(msg);
2689 		msg = ERR_PTR(-ERANGE);
2690 		goto out_free2;
2691 	}
2692 
2693 	msg->front.iov_len = p - msg->front.iov_base;
2694 	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2695 
2696 	if (req->r_pagelist) {
2697 		struct ceph_pagelist *pagelist = req->r_pagelist;
2698 		ceph_msg_data_add_pagelist(msg, pagelist);
2699 		msg->hdr.data_len = cpu_to_le32(pagelist->length);
2700 	} else {
2701 		msg->hdr.data_len = 0;
2702 	}
2703 
2704 	msg->hdr.data_off = cpu_to_le16(0);
2705 
2706 out_free2:
2707 	if (freepath2)
2708 		ceph_mdsc_free_path((char *)path2, pathlen2);
2709 out_free1:
2710 	if (freepath1)
2711 		ceph_mdsc_free_path((char *)path1, pathlen1);
2712 out:
2713 	return msg;
2714 }
2715 
2716 /*
2717  * called under mdsc->mutex if error, under no mutex if
2718  * success.
2719  */
complete_request(struct ceph_mds_client * mdsc,struct ceph_mds_request * req)2720 static void complete_request(struct ceph_mds_client *mdsc,
2721 			     struct ceph_mds_request *req)
2722 {
2723 	req->r_end_latency = ktime_get();
2724 
2725 	if (req->r_callback)
2726 		req->r_callback(mdsc, req);
2727 	complete_all(&req->r_completion);
2728 }
2729 
2730 static struct ceph_mds_request_head_old *
find_old_request_head(void * p,u64 features)2731 find_old_request_head(void *p, u64 features)
2732 {
2733 	bool legacy = !(features & CEPH_FEATURE_FS_BTIME);
2734 	struct ceph_mds_request_head *new_head;
2735 
2736 	if (legacy)
2737 		return (struct ceph_mds_request_head_old *)p;
2738 	new_head = (struct ceph_mds_request_head *)p;
2739 	return (struct ceph_mds_request_head_old *)&new_head->oldest_client_tid;
2740 }
2741 
2742 /*
2743  * called under mdsc->mutex
2744  */
__prepare_send_request(struct ceph_mds_session * session,struct ceph_mds_request * req,bool drop_cap_releases)2745 static int __prepare_send_request(struct ceph_mds_session *session,
2746 				  struct ceph_mds_request *req,
2747 				  bool drop_cap_releases)
2748 {
2749 	int mds = session->s_mds;
2750 	struct ceph_mds_client *mdsc = session->s_mdsc;
2751 	struct ceph_mds_request_head_old *rhead;
2752 	struct ceph_msg *msg;
2753 	int flags = 0, max_retry;
2754 
2755 	/*
2756 	 * The type of 'r_attempts' in kernel 'ceph_mds_request'
2757 	 * is 'int', while in 'ceph_mds_request_head' the type of
2758 	 * 'num_retry' is '__u8'. So in case the request retries
2759 	 *  exceeding 256 times, the MDS will receive a incorrect
2760 	 *  retry seq.
2761 	 *
2762 	 * In this case it's ususally a bug in MDS and continue
2763 	 * retrying the request makes no sense.
2764 	 *
2765 	 * In future this could be fixed in ceph code, so avoid
2766 	 * using the hardcode here.
2767 	 */
2768 	max_retry = sizeof_field(struct ceph_mds_request_head, num_retry);
2769 	max_retry = 1 << (max_retry * BITS_PER_BYTE);
2770 	if (req->r_attempts >= max_retry) {
2771 		pr_warn_ratelimited("%s request tid %llu seq overflow\n",
2772 				    __func__, req->r_tid);
2773 		return -EMULTIHOP;
2774 	}
2775 
2776 	req->r_attempts++;
2777 	if (req->r_inode) {
2778 		struct ceph_cap *cap =
2779 			ceph_get_cap_for_mds(ceph_inode(req->r_inode), mds);
2780 
2781 		if (cap)
2782 			req->r_sent_on_mseq = cap->mseq;
2783 		else
2784 			req->r_sent_on_mseq = -1;
2785 	}
2786 	dout("%s %p tid %lld %s (attempt %d)\n", __func__, req,
2787 	     req->r_tid, ceph_mds_op_name(req->r_op), req->r_attempts);
2788 
2789 	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
2790 		void *p;
2791 
2792 		/*
2793 		 * Replay.  Do not regenerate message (and rebuild
2794 		 * paths, etc.); just use the original message.
2795 		 * Rebuilding paths will break for renames because
2796 		 * d_move mangles the src name.
2797 		 */
2798 		msg = req->r_request;
2799 		rhead = find_old_request_head(msg->front.iov_base,
2800 					      session->s_con.peer_features);
2801 
2802 		flags = le32_to_cpu(rhead->flags);
2803 		flags |= CEPH_MDS_FLAG_REPLAY;
2804 		rhead->flags = cpu_to_le32(flags);
2805 
2806 		if (req->r_target_inode)
2807 			rhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode));
2808 
2809 		rhead->num_retry = req->r_attempts - 1;
2810 
2811 		/* remove cap/dentry releases from message */
2812 		rhead->num_releases = 0;
2813 
2814 		p = msg->front.iov_base + req->r_request_release_offset;
2815 		encode_timestamp_and_gids(&p, req);
2816 
2817 		msg->front.iov_len = p - msg->front.iov_base;
2818 		msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2819 		return 0;
2820 	}
2821 
2822 	if (req->r_request) {
2823 		ceph_msg_put(req->r_request);
2824 		req->r_request = NULL;
2825 	}
2826 	msg = create_request_message(session, req, drop_cap_releases);
2827 	if (IS_ERR(msg)) {
2828 		req->r_err = PTR_ERR(msg);
2829 		return PTR_ERR(msg);
2830 	}
2831 	req->r_request = msg;
2832 
2833 	rhead = find_old_request_head(msg->front.iov_base,
2834 				      session->s_con.peer_features);
2835 	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
2836 	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
2837 		flags |= CEPH_MDS_FLAG_REPLAY;
2838 	if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags))
2839 		flags |= CEPH_MDS_FLAG_ASYNC;
2840 	if (req->r_parent)
2841 		flags |= CEPH_MDS_FLAG_WANT_DENTRY;
2842 	rhead->flags = cpu_to_le32(flags);
2843 	rhead->num_fwd = req->r_num_fwd;
2844 	rhead->num_retry = req->r_attempts - 1;
2845 
2846 	dout(" r_parent = %p\n", req->r_parent);
2847 	return 0;
2848 }
2849 
2850 /*
2851  * called under mdsc->mutex
2852  */
__send_request(struct ceph_mds_session * session,struct ceph_mds_request * req,bool drop_cap_releases)2853 static int __send_request(struct ceph_mds_session *session,
2854 			  struct ceph_mds_request *req,
2855 			  bool drop_cap_releases)
2856 {
2857 	int err;
2858 
2859 	err = __prepare_send_request(session, req, drop_cap_releases);
2860 	if (!err) {
2861 		ceph_msg_get(req->r_request);
2862 		ceph_con_send(&session->s_con, req->r_request);
2863 	}
2864 
2865 	return err;
2866 }
2867 
2868 /*
2869  * send request, or put it on the appropriate wait list.
2870  */
__do_request(struct ceph_mds_client * mdsc,struct ceph_mds_request * req)2871 static void __do_request(struct ceph_mds_client *mdsc,
2872 			struct ceph_mds_request *req)
2873 {
2874 	struct ceph_mds_session *session = NULL;
2875 	int mds = -1;
2876 	int err = 0;
2877 	bool random;
2878 
2879 	if (req->r_err || test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
2880 		if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
2881 			__unregister_request(mdsc, req);
2882 		return;
2883 	}
2884 
2885 	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO) {
2886 		dout("do_request metadata corrupted\n");
2887 		err = -EIO;
2888 		goto finish;
2889 	}
2890 	if (req->r_timeout &&
2891 	    time_after_eq(jiffies, req->r_started + req->r_timeout)) {
2892 		dout("do_request timed out\n");
2893 		err = -ETIMEDOUT;
2894 		goto finish;
2895 	}
2896 	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
2897 		dout("do_request forced umount\n");
2898 		err = -EIO;
2899 		goto finish;
2900 	}
2901 	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_MOUNTING) {
2902 		if (mdsc->mdsmap_err) {
2903 			err = mdsc->mdsmap_err;
2904 			dout("do_request mdsmap err %d\n", err);
2905 			goto finish;
2906 		}
2907 		if (mdsc->mdsmap->m_epoch == 0) {
2908 			dout("do_request no mdsmap, waiting for map\n");
2909 			list_add(&req->r_wait, &mdsc->waiting_for_map);
2910 			return;
2911 		}
2912 		if (!(mdsc->fsc->mount_options->flags &
2913 		      CEPH_MOUNT_OPT_MOUNTWAIT) &&
2914 		    !ceph_mdsmap_is_cluster_available(mdsc->mdsmap)) {
2915 			err = -EHOSTUNREACH;
2916 			goto finish;
2917 		}
2918 	}
2919 
2920 	put_request_session(req);
2921 
2922 	mds = __choose_mds(mdsc, req, &random);
2923 	if (mds < 0 ||
2924 	    ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {
2925 		if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {
2926 			err = -EJUKEBOX;
2927 			goto finish;
2928 		}
2929 		dout("do_request no mds or not active, waiting for map\n");
2930 		list_add(&req->r_wait, &mdsc->waiting_for_map);
2931 		return;
2932 	}
2933 
2934 	/* get, open session */
2935 	session = __ceph_lookup_mds_session(mdsc, mds);
2936 	if (!session) {
2937 		session = register_session(mdsc, mds);
2938 		if (IS_ERR(session)) {
2939 			err = PTR_ERR(session);
2940 			goto finish;
2941 		}
2942 	}
2943 	req->r_session = ceph_get_mds_session(session);
2944 
2945 	dout("do_request mds%d session %p state %s\n", mds, session,
2946 	     ceph_session_state_name(session->s_state));
2947 
2948 	/*
2949 	 * The old ceph will crash the MDSs when see unknown OPs
2950 	 */
2951 	if (req->r_feature_needed > 0 &&
2952 	    !test_bit(req->r_feature_needed, &session->s_features)) {
2953 		err = -EOPNOTSUPP;
2954 		goto out_session;
2955 	}
2956 
2957 	if (session->s_state != CEPH_MDS_SESSION_OPEN &&
2958 	    session->s_state != CEPH_MDS_SESSION_HUNG) {
2959 		/*
2960 		 * We cannot queue async requests since the caps and delegated
2961 		 * inodes are bound to the session. Just return -EJUKEBOX and
2962 		 * let the caller retry a sync request in that case.
2963 		 */
2964 		if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {
2965 			err = -EJUKEBOX;
2966 			goto out_session;
2967 		}
2968 
2969 		/*
2970 		 * If the session has been REJECTED, then return a hard error,
2971 		 * unless it's a CLEANRECOVER mount, in which case we'll queue
2972 		 * it to the mdsc queue.
2973 		 */
2974 		if (session->s_state == CEPH_MDS_SESSION_REJECTED) {
2975 			if (ceph_test_mount_opt(mdsc->fsc, CLEANRECOVER))
2976 				list_add(&req->r_wait, &mdsc->waiting_for_map);
2977 			else
2978 				err = -EACCES;
2979 			goto out_session;
2980 		}
2981 
2982 		if (session->s_state == CEPH_MDS_SESSION_NEW ||
2983 		    session->s_state == CEPH_MDS_SESSION_CLOSING) {
2984 			err = __open_session(mdsc, session);
2985 			if (err)
2986 				goto out_session;
2987 			/* retry the same mds later */
2988 			if (random)
2989 				req->r_resend_mds = mds;
2990 		}
2991 		list_add(&req->r_wait, &session->s_waiting);
2992 		goto out_session;
2993 	}
2994 
2995 	/* send request */
2996 	req->r_resend_mds = -1;   /* forget any previous mds hint */
2997 
2998 	if (req->r_request_started == 0)   /* note request start time */
2999 		req->r_request_started = jiffies;
3000 
3001 	/*
3002 	 * For async create we will choose the auth MDS of frag in parent
3003 	 * directory to send the request and ususally this works fine, but
3004 	 * if the migrated the dirtory to another MDS before it could handle
3005 	 * it the request will be forwarded.
3006 	 *
3007 	 * And then the auth cap will be changed.
3008 	 */
3009 	if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags) && req->r_num_fwd) {
3010 		struct ceph_dentry_info *di = ceph_dentry(req->r_dentry);
3011 		struct ceph_inode_info *ci;
3012 		struct ceph_cap *cap;
3013 
3014 		/*
3015 		 * The request maybe handled very fast and the new inode
3016 		 * hasn't been linked to the dentry yet. We need to wait
3017 		 * for the ceph_finish_async_create(), which shouldn't be
3018 		 * stuck too long or fail in thoery, to finish when forwarding
3019 		 * the request.
3020 		 */
3021 		if (!d_inode(req->r_dentry)) {
3022 			err = wait_on_bit(&di->flags, CEPH_DENTRY_ASYNC_CREATE_BIT,
3023 					  TASK_KILLABLE);
3024 			if (err) {
3025 				mutex_lock(&req->r_fill_mutex);
3026 				set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
3027 				mutex_unlock(&req->r_fill_mutex);
3028 				goto out_session;
3029 			}
3030 		}
3031 
3032 		ci = ceph_inode(d_inode(req->r_dentry));
3033 
3034 		spin_lock(&ci->i_ceph_lock);
3035 		cap = ci->i_auth_cap;
3036 		if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE && mds != cap->mds) {
3037 			dout("do_request session changed for auth cap %d -> %d\n",
3038 			     cap->session->s_mds, session->s_mds);
3039 
3040 			/* Remove the auth cap from old session */
3041 			spin_lock(&cap->session->s_cap_lock);
3042 			cap->session->s_nr_caps--;
3043 			list_del_init(&cap->session_caps);
3044 			spin_unlock(&cap->session->s_cap_lock);
3045 
3046 			/* Add the auth cap to the new session */
3047 			cap->mds = mds;
3048 			cap->session = session;
3049 			spin_lock(&session->s_cap_lock);
3050 			session->s_nr_caps++;
3051 			list_add_tail(&cap->session_caps, &session->s_caps);
3052 			spin_unlock(&session->s_cap_lock);
3053 
3054 			change_auth_cap_ses(ci, session);
3055 		}
3056 		spin_unlock(&ci->i_ceph_lock);
3057 	}
3058 
3059 	err = __send_request(session, req, false);
3060 
3061 out_session:
3062 	ceph_put_mds_session(session);
3063 finish:
3064 	if (err) {
3065 		dout("__do_request early error %d\n", err);
3066 		req->r_err = err;
3067 		complete_request(mdsc, req);
3068 		__unregister_request(mdsc, req);
3069 	}
3070 	return;
3071 }
3072 
3073 /*
3074  * called under mdsc->mutex
3075  */
__wake_requests(struct ceph_mds_client * mdsc,struct list_head * head)3076 static void __wake_requests(struct ceph_mds_client *mdsc,
3077 			    struct list_head *head)
3078 {
3079 	struct ceph_mds_request *req;
3080 	LIST_HEAD(tmp_list);
3081 
3082 	list_splice_init(head, &tmp_list);
3083 
3084 	while (!list_empty(&tmp_list)) {
3085 		req = list_entry(tmp_list.next,
3086 				 struct ceph_mds_request, r_wait);
3087 		list_del_init(&req->r_wait);
3088 		dout(" wake request %p tid %llu\n", req, req->r_tid);
3089 		__do_request(mdsc, req);
3090 	}
3091 }
3092 
3093 /*
3094  * Wake up threads with requests pending for @mds, so that they can
3095  * resubmit their requests to a possibly different mds.
3096  */
kick_requests(struct ceph_mds_client * mdsc,int mds)3097 static void kick_requests(struct ceph_mds_client *mdsc, int mds)
3098 {
3099 	struct ceph_mds_request *req;
3100 	struct rb_node *p = rb_first(&mdsc->request_tree);
3101 
3102 	dout("kick_requests mds%d\n", mds);
3103 	while (p) {
3104 		req = rb_entry(p, struct ceph_mds_request, r_node);
3105 		p = rb_next(p);
3106 		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
3107 			continue;
3108 		if (req->r_attempts > 0)
3109 			continue; /* only new requests */
3110 		if (req->r_session &&
3111 		    req->r_session->s_mds == mds) {
3112 			dout(" kicking tid %llu\n", req->r_tid);
3113 			list_del_init(&req->r_wait);
3114 			__do_request(mdsc, req);
3115 		}
3116 	}
3117 }
3118 
ceph_mdsc_submit_request(struct ceph_mds_client * mdsc,struct inode * dir,struct ceph_mds_request * req)3119 int ceph_mdsc_submit_request(struct ceph_mds_client *mdsc, struct inode *dir,
3120 			      struct ceph_mds_request *req)
3121 {
3122 	int err = 0;
3123 
3124 	/* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
3125 	if (req->r_inode)
3126 		ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
3127 	if (req->r_parent) {
3128 		struct ceph_inode_info *ci = ceph_inode(req->r_parent);
3129 		int fmode = (req->r_op & CEPH_MDS_OP_WRITE) ?
3130 			    CEPH_FILE_MODE_WR : CEPH_FILE_MODE_RD;
3131 		spin_lock(&ci->i_ceph_lock);
3132 		ceph_take_cap_refs(ci, CEPH_CAP_PIN, false);
3133 		__ceph_touch_fmode(ci, mdsc, fmode);
3134 		spin_unlock(&ci->i_ceph_lock);
3135 	}
3136 	if (req->r_old_dentry_dir)
3137 		ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
3138 				  CEPH_CAP_PIN);
3139 
3140 	if (req->r_inode) {
3141 		err = ceph_wait_on_async_create(req->r_inode);
3142 		if (err) {
3143 			dout("%s: wait for async create returned: %d\n",
3144 			     __func__, err);
3145 			return err;
3146 		}
3147 	}
3148 
3149 	if (!err && req->r_old_inode) {
3150 		err = ceph_wait_on_async_create(req->r_old_inode);
3151 		if (err) {
3152 			dout("%s: wait for async create returned: %d\n",
3153 			     __func__, err);
3154 			return err;
3155 		}
3156 	}
3157 
3158 	dout("submit_request on %p for inode %p\n", req, dir);
3159 	mutex_lock(&mdsc->mutex);
3160 	__register_request(mdsc, req, dir);
3161 	__do_request(mdsc, req);
3162 	err = req->r_err;
3163 	mutex_unlock(&mdsc->mutex);
3164 	return err;
3165 }
3166 
ceph_mdsc_wait_request(struct ceph_mds_client * mdsc,struct ceph_mds_request * req,ceph_mds_request_wait_callback_t wait_func)3167 int ceph_mdsc_wait_request(struct ceph_mds_client *mdsc,
3168 			   struct ceph_mds_request *req,
3169 			   ceph_mds_request_wait_callback_t wait_func)
3170 {
3171 	int err;
3172 
3173 	/* wait */
3174 	dout("do_request waiting\n");
3175 	if (wait_func) {
3176 		err = wait_func(mdsc, req);
3177 	} else {
3178 		long timeleft = wait_for_completion_killable_timeout(
3179 					&req->r_completion,
3180 					ceph_timeout_jiffies(req->r_timeout));
3181 		if (timeleft > 0)
3182 			err = 0;
3183 		else if (!timeleft)
3184 			err = -ETIMEDOUT;  /* timed out */
3185 		else
3186 			err = timeleft;  /* killed */
3187 	}
3188 	dout("do_request waited, got %d\n", err);
3189 	mutex_lock(&mdsc->mutex);
3190 
3191 	/* only abort if we didn't race with a real reply */
3192 	if (test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
3193 		err = le32_to_cpu(req->r_reply_info.head->result);
3194 	} else if (err < 0) {
3195 		dout("aborted request %lld with %d\n", req->r_tid, err);
3196 
3197 		/*
3198 		 * ensure we aren't running concurrently with
3199 		 * ceph_fill_trace or ceph_readdir_prepopulate, which
3200 		 * rely on locks (dir mutex) held by our caller.
3201 		 */
3202 		mutex_lock(&req->r_fill_mutex);
3203 		req->r_err = err;
3204 		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
3205 		mutex_unlock(&req->r_fill_mutex);
3206 
3207 		if (req->r_parent &&
3208 		    (req->r_op & CEPH_MDS_OP_WRITE))
3209 			ceph_invalidate_dir_request(req);
3210 	} else {
3211 		err = req->r_err;
3212 	}
3213 
3214 	mutex_unlock(&mdsc->mutex);
3215 	return err;
3216 }
3217 
3218 /*
3219  * Synchrously perform an mds request.  Take care of all of the
3220  * session setup, forwarding, retry details.
3221  */
ceph_mdsc_do_request(struct ceph_mds_client * mdsc,struct inode * dir,struct ceph_mds_request * req)3222 int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
3223 			 struct inode *dir,
3224 			 struct ceph_mds_request *req)
3225 {
3226 	int err;
3227 
3228 	dout("do_request on %p\n", req);
3229 
3230 	/* issue */
3231 	err = ceph_mdsc_submit_request(mdsc, dir, req);
3232 	if (!err)
3233 		err = ceph_mdsc_wait_request(mdsc, req, NULL);
3234 	dout("do_request %p done, result %d\n", req, err);
3235 	return err;
3236 }
3237 
3238 /*
3239  * Invalidate dir's completeness, dentry lease state on an aborted MDS
3240  * namespace request.
3241  */
ceph_invalidate_dir_request(struct ceph_mds_request * req)3242 void ceph_invalidate_dir_request(struct ceph_mds_request *req)
3243 {
3244 	struct inode *dir = req->r_parent;
3245 	struct inode *old_dir = req->r_old_dentry_dir;
3246 
3247 	dout("invalidate_dir_request %p %p (complete, lease(s))\n", dir, old_dir);
3248 
3249 	ceph_dir_clear_complete(dir);
3250 	if (old_dir)
3251 		ceph_dir_clear_complete(old_dir);
3252 	if (req->r_dentry)
3253 		ceph_invalidate_dentry_lease(req->r_dentry);
3254 	if (req->r_old_dentry)
3255 		ceph_invalidate_dentry_lease(req->r_old_dentry);
3256 }
3257 
3258 /*
3259  * Handle mds reply.
3260  *
3261  * We take the session mutex and parse and process the reply immediately.
3262  * This preserves the logical ordering of replies, capabilities, etc., sent
3263  * by the MDS as they are applied to our local cache.
3264  */
handle_reply(struct ceph_mds_session * session,struct ceph_msg * msg)3265 static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
3266 {
3267 	struct ceph_mds_client *mdsc = session->s_mdsc;
3268 	struct ceph_mds_request *req;
3269 	struct ceph_mds_reply_head *head = msg->front.iov_base;
3270 	struct ceph_mds_reply_info_parsed *rinfo;  /* parsed reply info */
3271 	struct ceph_snap_realm *realm;
3272 	u64 tid;
3273 	int err, result;
3274 	int mds = session->s_mds;
3275 	bool close_sessions = false;
3276 
3277 	if (msg->front.iov_len < sizeof(*head)) {
3278 		pr_err("mdsc_handle_reply got corrupt (short) reply\n");
3279 		ceph_msg_dump(msg);
3280 		return;
3281 	}
3282 
3283 	/* get request, session */
3284 	tid = le64_to_cpu(msg->hdr.tid);
3285 	mutex_lock(&mdsc->mutex);
3286 	req = lookup_get_request(mdsc, tid);
3287 	if (!req) {
3288 		dout("handle_reply on unknown tid %llu\n", tid);
3289 		mutex_unlock(&mdsc->mutex);
3290 		return;
3291 	}
3292 	dout("handle_reply %p\n", req);
3293 
3294 	/* correct session? */
3295 	if (req->r_session != session) {
3296 		pr_err("mdsc_handle_reply got %llu on session mds%d"
3297 		       " not mds%d\n", tid, session->s_mds,
3298 		       req->r_session ? req->r_session->s_mds : -1);
3299 		mutex_unlock(&mdsc->mutex);
3300 		goto out;
3301 	}
3302 
3303 	/* dup? */
3304 	if ((test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags) && !head->safe) ||
3305 	    (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags) && head->safe)) {
3306 		pr_warn("got a dup %s reply on %llu from mds%d\n",
3307 			   head->safe ? "safe" : "unsafe", tid, mds);
3308 		mutex_unlock(&mdsc->mutex);
3309 		goto out;
3310 	}
3311 	if (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags)) {
3312 		pr_warn("got unsafe after safe on %llu from mds%d\n",
3313 			   tid, mds);
3314 		mutex_unlock(&mdsc->mutex);
3315 		goto out;
3316 	}
3317 
3318 	result = le32_to_cpu(head->result);
3319 
3320 	if (head->safe) {
3321 		set_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags);
3322 		__unregister_request(mdsc, req);
3323 
3324 		/* last request during umount? */
3325 		if (mdsc->stopping && !__get_oldest_req(mdsc))
3326 			complete_all(&mdsc->safe_umount_waiters);
3327 
3328 		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
3329 			/*
3330 			 * We already handled the unsafe response, now do the
3331 			 * cleanup.  No need to examine the response; the MDS
3332 			 * doesn't include any result info in the safe
3333 			 * response.  And even if it did, there is nothing
3334 			 * useful we could do with a revised return value.
3335 			 */
3336 			dout("got safe reply %llu, mds%d\n", tid, mds);
3337 
3338 			mutex_unlock(&mdsc->mutex);
3339 			goto out;
3340 		}
3341 	} else {
3342 		set_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags);
3343 		list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
3344 	}
3345 
3346 	dout("handle_reply tid %lld result %d\n", tid, result);
3347 	rinfo = &req->r_reply_info;
3348 	if (test_bit(CEPHFS_FEATURE_REPLY_ENCODING, &session->s_features))
3349 		err = parse_reply_info(session, msg, rinfo, (u64)-1);
3350 	else
3351 		err = parse_reply_info(session, msg, rinfo, session->s_con.peer_features);
3352 	mutex_unlock(&mdsc->mutex);
3353 
3354 	/* Must find target inode outside of mutexes to avoid deadlocks */
3355 	if ((err >= 0) && rinfo->head->is_target) {
3356 		struct inode *in;
3357 		struct ceph_vino tvino = {
3358 			.ino  = le64_to_cpu(rinfo->targeti.in->ino),
3359 			.snap = le64_to_cpu(rinfo->targeti.in->snapid)
3360 		};
3361 
3362 		in = ceph_get_inode(mdsc->fsc->sb, tvino);
3363 		if (IS_ERR(in)) {
3364 			err = PTR_ERR(in);
3365 			mutex_lock(&session->s_mutex);
3366 			goto out_err;
3367 		}
3368 		req->r_target_inode = in;
3369 	}
3370 
3371 	mutex_lock(&session->s_mutex);
3372 	if (err < 0) {
3373 		pr_err("mdsc_handle_reply got corrupt reply mds%d(tid:%lld)\n", mds, tid);
3374 		ceph_msg_dump(msg);
3375 		goto out_err;
3376 	}
3377 
3378 	/* snap trace */
3379 	realm = NULL;
3380 	if (rinfo->snapblob_len) {
3381 		down_write(&mdsc->snap_rwsem);
3382 		err = ceph_update_snap_trace(mdsc, rinfo->snapblob,
3383 				rinfo->snapblob + rinfo->snapblob_len,
3384 				le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP,
3385 				&realm);
3386 		if (err) {
3387 			up_write(&mdsc->snap_rwsem);
3388 			close_sessions = true;
3389 			if (err == -EIO)
3390 				ceph_msg_dump(msg);
3391 			goto out_err;
3392 		}
3393 		downgrade_write(&mdsc->snap_rwsem);
3394 	} else {
3395 		down_read(&mdsc->snap_rwsem);
3396 	}
3397 
3398 	/* insert trace into our cache */
3399 	mutex_lock(&req->r_fill_mutex);
3400 	current->journal_info = req;
3401 	err = ceph_fill_trace(mdsc->fsc->sb, req);
3402 	if (err == 0) {
3403 		if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR ||
3404 				    req->r_op == CEPH_MDS_OP_LSSNAP))
3405 			ceph_readdir_prepopulate(req, req->r_session);
3406 	}
3407 	current->journal_info = NULL;
3408 	mutex_unlock(&req->r_fill_mutex);
3409 
3410 	up_read(&mdsc->snap_rwsem);
3411 	if (realm)
3412 		ceph_put_snap_realm(mdsc, realm);
3413 
3414 	if (err == 0) {
3415 		if (req->r_target_inode &&
3416 		    test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
3417 			struct ceph_inode_info *ci =
3418 				ceph_inode(req->r_target_inode);
3419 			spin_lock(&ci->i_unsafe_lock);
3420 			list_add_tail(&req->r_unsafe_target_item,
3421 				      &ci->i_unsafe_iops);
3422 			spin_unlock(&ci->i_unsafe_lock);
3423 		}
3424 
3425 		ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
3426 	}
3427 out_err:
3428 	mutex_lock(&mdsc->mutex);
3429 	if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
3430 		if (err) {
3431 			req->r_err = err;
3432 		} else {
3433 			req->r_reply =  ceph_msg_get(msg);
3434 			set_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags);
3435 		}
3436 	} else {
3437 		dout("reply arrived after request %lld was aborted\n", tid);
3438 	}
3439 	mutex_unlock(&mdsc->mutex);
3440 
3441 	mutex_unlock(&session->s_mutex);
3442 
3443 	/* kick calling process */
3444 	complete_request(mdsc, req);
3445 
3446 	ceph_update_metadata_metrics(&mdsc->metric, req->r_start_latency,
3447 				     req->r_end_latency, err);
3448 out:
3449 	ceph_mdsc_put_request(req);
3450 
3451 	/* Defer closing the sessions after s_mutex lock being released */
3452 	if (close_sessions)
3453 		ceph_mdsc_close_sessions(mdsc);
3454 	return;
3455 }
3456 
3457 
3458 
3459 /*
3460  * handle mds notification that our request has been forwarded.
3461  */
handle_forward(struct ceph_mds_client * mdsc,struct ceph_mds_session * session,struct ceph_msg * msg)3462 static void handle_forward(struct ceph_mds_client *mdsc,
3463 			   struct ceph_mds_session *session,
3464 			   struct ceph_msg *msg)
3465 {
3466 	struct ceph_mds_request *req;
3467 	u64 tid = le64_to_cpu(msg->hdr.tid);
3468 	u32 next_mds;
3469 	u32 fwd_seq;
3470 	int err = -EINVAL;
3471 	void *p = msg->front.iov_base;
3472 	void *end = p + msg->front.iov_len;
3473 	bool aborted = false;
3474 
3475 	ceph_decode_need(&p, end, 2*sizeof(u32), bad);
3476 	next_mds = ceph_decode_32(&p);
3477 	fwd_seq = ceph_decode_32(&p);
3478 
3479 	mutex_lock(&mdsc->mutex);
3480 	req = lookup_get_request(mdsc, tid);
3481 	if (!req) {
3482 		mutex_unlock(&mdsc->mutex);
3483 		dout("forward tid %llu to mds%d - req dne\n", tid, next_mds);
3484 		return;  /* dup reply? */
3485 	}
3486 
3487 	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
3488 		dout("forward tid %llu aborted, unregistering\n", tid);
3489 		__unregister_request(mdsc, req);
3490 	} else if (fwd_seq <= req->r_num_fwd) {
3491 		/*
3492 		 * The type of 'num_fwd' in ceph 'MClientRequestForward'
3493 		 * is 'int32_t', while in 'ceph_mds_request_head' the
3494 		 * type is '__u8'. So in case the request bounces between
3495 		 * MDSes exceeding 256 times, the client will get stuck.
3496 		 *
3497 		 * In this case it's ususally a bug in MDS and continue
3498 		 * bouncing the request makes no sense.
3499 		 *
3500 		 * In future this could be fixed in ceph code, so avoid
3501 		 * using the hardcode here.
3502 		 */
3503 		int max = sizeof_field(struct ceph_mds_request_head, num_fwd);
3504 		max = 1 << (max * BITS_PER_BYTE);
3505 		if (req->r_num_fwd >= max) {
3506 			mutex_lock(&req->r_fill_mutex);
3507 			req->r_err = -EMULTIHOP;
3508 			set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
3509 			mutex_unlock(&req->r_fill_mutex);
3510 			aborted = true;
3511 			pr_warn_ratelimited("forward tid %llu seq overflow\n",
3512 					    tid);
3513 		} else {
3514 			dout("forward tid %llu to mds%d - old seq %d <= %d\n",
3515 			     tid, next_mds, req->r_num_fwd, fwd_seq);
3516 		}
3517 	} else {
3518 		/* resend. forward race not possible; mds would drop */
3519 		dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds);
3520 		BUG_ON(req->r_err);
3521 		BUG_ON(test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags));
3522 		req->r_attempts = 0;
3523 		req->r_num_fwd = fwd_seq;
3524 		req->r_resend_mds = next_mds;
3525 		put_request_session(req);
3526 		__do_request(mdsc, req);
3527 	}
3528 	mutex_unlock(&mdsc->mutex);
3529 
3530 	/* kick calling process */
3531 	if (aborted)
3532 		complete_request(mdsc, req);
3533 	ceph_mdsc_put_request(req);
3534 	return;
3535 
3536 bad:
3537 	pr_err("mdsc_handle_forward decode error err=%d\n", err);
3538 	ceph_msg_dump(msg);
3539 }
3540 
__decode_session_metadata(void ** p,void * end,bool * blocklisted)3541 static int __decode_session_metadata(void **p, void *end,
3542 				     bool *blocklisted)
3543 {
3544 	/* map<string,string> */
3545 	u32 n;
3546 	bool err_str;
3547 	ceph_decode_32_safe(p, end, n, bad);
3548 	while (n-- > 0) {
3549 		u32 len;
3550 		ceph_decode_32_safe(p, end, len, bad);
3551 		ceph_decode_need(p, end, len, bad);
3552 		err_str = !strncmp(*p, "error_string", len);
3553 		*p += len;
3554 		ceph_decode_32_safe(p, end, len, bad);
3555 		ceph_decode_need(p, end, len, bad);
3556 		/*
3557 		 * Match "blocklisted (blacklisted)" from newer MDSes,
3558 		 * or "blacklisted" from older MDSes.
3559 		 */
3560 		if (err_str && strnstr(*p, "blacklisted", len))
3561 			*blocklisted = true;
3562 		*p += len;
3563 	}
3564 	return 0;
3565 bad:
3566 	return -1;
3567 }
3568 
3569 /*
3570  * handle a mds session control message
3571  */
handle_session(struct ceph_mds_session * session,struct ceph_msg * msg)3572 static void handle_session(struct ceph_mds_session *session,
3573 			   struct ceph_msg *msg)
3574 {
3575 	struct ceph_mds_client *mdsc = session->s_mdsc;
3576 	int mds = session->s_mds;
3577 	int msg_version = le16_to_cpu(msg->hdr.version);
3578 	void *p = msg->front.iov_base;
3579 	void *end = p + msg->front.iov_len;
3580 	struct ceph_mds_session_head *h;
3581 	u32 op;
3582 	u64 seq, features = 0;
3583 	int wake = 0;
3584 	bool blocklisted = false;
3585 
3586 	/* decode */
3587 	ceph_decode_need(&p, end, sizeof(*h), bad);
3588 	h = p;
3589 	p += sizeof(*h);
3590 
3591 	op = le32_to_cpu(h->op);
3592 	seq = le64_to_cpu(h->seq);
3593 
3594 	if (msg_version >= 3) {
3595 		u32 len;
3596 		/* version >= 2 and < 5, decode metadata, skip otherwise
3597 		 * as it's handled via flags.
3598 		 */
3599 		if (msg_version >= 5)
3600 			ceph_decode_skip_map(&p, end, string, string, bad);
3601 		else if (__decode_session_metadata(&p, end, &blocklisted) < 0)
3602 			goto bad;
3603 
3604 		/* version >= 3, feature bits */
3605 		ceph_decode_32_safe(&p, end, len, bad);
3606 		if (len) {
3607 			ceph_decode_64_safe(&p, end, features, bad);
3608 			p += len - sizeof(features);
3609 		}
3610 	}
3611 
3612 	if (msg_version >= 5) {
3613 		u32 flags, len;
3614 
3615 		/* version >= 4 */
3616 		ceph_decode_skip_16(&p, end, bad); /* struct_v, struct_cv */
3617 		ceph_decode_32_safe(&p, end, len, bad); /* len */
3618 		ceph_decode_skip_n(&p, end, len, bad); /* metric_spec */
3619 
3620 		/* version >= 5, flags   */
3621 		ceph_decode_32_safe(&p, end, flags, bad);
3622 		if (flags & CEPH_SESSION_BLOCKLISTED) {
3623 			pr_warn("mds%d session blocklisted\n", session->s_mds);
3624 			blocklisted = true;
3625 		}
3626 	}
3627 
3628 	mutex_lock(&mdsc->mutex);
3629 	if (op == CEPH_SESSION_CLOSE) {
3630 		ceph_get_mds_session(session);
3631 		__unregister_session(mdsc, session);
3632 	}
3633 	/* FIXME: this ttl calculation is generous */
3634 	session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;
3635 	mutex_unlock(&mdsc->mutex);
3636 
3637 	mutex_lock(&session->s_mutex);
3638 
3639 	dout("handle_session mds%d %s %p state %s seq %llu\n",
3640 	     mds, ceph_session_op_name(op), session,
3641 	     ceph_session_state_name(session->s_state), seq);
3642 
3643 	if (session->s_state == CEPH_MDS_SESSION_HUNG) {
3644 		session->s_state = CEPH_MDS_SESSION_OPEN;
3645 		pr_info("mds%d came back\n", session->s_mds);
3646 	}
3647 
3648 	switch (op) {
3649 	case CEPH_SESSION_OPEN:
3650 		if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
3651 			pr_info("mds%d reconnect success\n", session->s_mds);
3652 
3653 		session->s_features = features;
3654 		if (session->s_state == CEPH_MDS_SESSION_OPEN) {
3655 			pr_notice("mds%d is already opened\n", session->s_mds);
3656 		} else {
3657 			session->s_state = CEPH_MDS_SESSION_OPEN;
3658 			renewed_caps(mdsc, session, 0);
3659 			if (test_bit(CEPHFS_FEATURE_METRIC_COLLECT,
3660 				     &session->s_features))
3661 				metric_schedule_delayed(&mdsc->metric);
3662 		}
3663 
3664 		/*
3665 		 * The connection maybe broken and the session in client
3666 		 * side has been reinitialized, need to update the seq
3667 		 * anyway.
3668 		 */
3669 		if (!session->s_seq && seq)
3670 			session->s_seq = seq;
3671 
3672 		wake = 1;
3673 		if (mdsc->stopping)
3674 			__close_session(mdsc, session);
3675 		break;
3676 
3677 	case CEPH_SESSION_RENEWCAPS:
3678 		if (session->s_renew_seq == seq)
3679 			renewed_caps(mdsc, session, 1);
3680 		break;
3681 
3682 	case CEPH_SESSION_CLOSE:
3683 		if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
3684 			pr_info("mds%d reconnect denied\n", session->s_mds);
3685 		session->s_state = CEPH_MDS_SESSION_CLOSED;
3686 		cleanup_session_requests(mdsc, session);
3687 		remove_session_caps(session);
3688 		wake = 2; /* for good measure */
3689 		wake_up_all(&mdsc->session_close_wq);
3690 		break;
3691 
3692 	case CEPH_SESSION_STALE:
3693 		pr_info("mds%d caps went stale, renewing\n",
3694 			session->s_mds);
3695 		atomic_inc(&session->s_cap_gen);
3696 		session->s_cap_ttl = jiffies - 1;
3697 		send_renew_caps(mdsc, session);
3698 		break;
3699 
3700 	case CEPH_SESSION_RECALL_STATE:
3701 		ceph_trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
3702 		break;
3703 
3704 	case CEPH_SESSION_FLUSHMSG:
3705 		/* flush cap releases */
3706 		spin_lock(&session->s_cap_lock);
3707 		if (session->s_num_cap_releases)
3708 			ceph_flush_cap_releases(mdsc, session);
3709 		spin_unlock(&session->s_cap_lock);
3710 
3711 		send_flushmsg_ack(mdsc, session, seq);
3712 		break;
3713 
3714 	case CEPH_SESSION_FORCE_RO:
3715 		dout("force_session_readonly %p\n", session);
3716 		spin_lock(&session->s_cap_lock);
3717 		session->s_readonly = true;
3718 		spin_unlock(&session->s_cap_lock);
3719 		wake_up_session_caps(session, FORCE_RO);
3720 		break;
3721 
3722 	case CEPH_SESSION_REJECT:
3723 		WARN_ON(session->s_state != CEPH_MDS_SESSION_OPENING);
3724 		pr_info("mds%d rejected session\n", session->s_mds);
3725 		session->s_state = CEPH_MDS_SESSION_REJECTED;
3726 		cleanup_session_requests(mdsc, session);
3727 		remove_session_caps(session);
3728 		if (blocklisted)
3729 			mdsc->fsc->blocklisted = true;
3730 		wake = 2; /* for good measure */
3731 		break;
3732 
3733 	default:
3734 		pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds);
3735 		WARN_ON(1);
3736 	}
3737 
3738 	mutex_unlock(&session->s_mutex);
3739 	if (wake) {
3740 		mutex_lock(&mdsc->mutex);
3741 		__wake_requests(mdsc, &session->s_waiting);
3742 		if (wake == 2)
3743 			kick_requests(mdsc, mds);
3744 		mutex_unlock(&mdsc->mutex);
3745 	}
3746 	if (op == CEPH_SESSION_CLOSE)
3747 		ceph_put_mds_session(session);
3748 	return;
3749 
3750 bad:
3751 	pr_err("mdsc_handle_session corrupt message mds%d len %d\n", mds,
3752 	       (int)msg->front.iov_len);
3753 	ceph_msg_dump(msg);
3754 	return;
3755 }
3756 
ceph_mdsc_release_dir_caps(struct ceph_mds_request * req)3757 void ceph_mdsc_release_dir_caps(struct ceph_mds_request *req)
3758 {
3759 	int dcaps;
3760 
3761 	dcaps = xchg(&req->r_dir_caps, 0);
3762 	if (dcaps) {
3763 		dout("releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));
3764 		ceph_put_cap_refs(ceph_inode(req->r_parent), dcaps);
3765 	}
3766 }
3767 
ceph_mdsc_release_dir_caps_no_check(struct ceph_mds_request * req)3768 void ceph_mdsc_release_dir_caps_no_check(struct ceph_mds_request *req)
3769 {
3770 	int dcaps;
3771 
3772 	dcaps = xchg(&req->r_dir_caps, 0);
3773 	if (dcaps) {
3774 		dout("releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));
3775 		ceph_put_cap_refs_no_check_caps(ceph_inode(req->r_parent),
3776 						dcaps);
3777 	}
3778 }
3779 
3780 /*
3781  * called under session->mutex.
3782  */
replay_unsafe_requests(struct ceph_mds_client * mdsc,struct ceph_mds_session * session)3783 static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
3784 				   struct ceph_mds_session *session)
3785 {
3786 	struct ceph_mds_request *req, *nreq;
3787 	struct rb_node *p;
3788 
3789 	dout("replay_unsafe_requests mds%d\n", session->s_mds);
3790 
3791 	mutex_lock(&mdsc->mutex);
3792 	list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item)
3793 		__send_request(session, req, true);
3794 
3795 	/*
3796 	 * also re-send old requests when MDS enters reconnect stage. So that MDS
3797 	 * can process completed request in clientreplay stage.
3798 	 */
3799 	p = rb_first(&mdsc->request_tree);
3800 	while (p) {
3801 		req = rb_entry(p, struct ceph_mds_request, r_node);
3802 		p = rb_next(p);
3803 		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
3804 			continue;
3805 		if (req->r_attempts == 0)
3806 			continue; /* only old requests */
3807 		if (!req->r_session)
3808 			continue;
3809 		if (req->r_session->s_mds != session->s_mds)
3810 			continue;
3811 
3812 		ceph_mdsc_release_dir_caps_no_check(req);
3813 
3814 		__send_request(session, req, true);
3815 	}
3816 	mutex_unlock(&mdsc->mutex);
3817 }
3818 
send_reconnect_partial(struct ceph_reconnect_state * recon_state)3819 static int send_reconnect_partial(struct ceph_reconnect_state *recon_state)
3820 {
3821 	struct ceph_msg *reply;
3822 	struct ceph_pagelist *_pagelist;
3823 	struct page *page;
3824 	__le32 *addr;
3825 	int err = -ENOMEM;
3826 
3827 	if (!recon_state->allow_multi)
3828 		return -ENOSPC;
3829 
3830 	/* can't handle message that contains both caps and realm */
3831 	BUG_ON(!recon_state->nr_caps == !recon_state->nr_realms);
3832 
3833 	/* pre-allocate new pagelist */
3834 	_pagelist = ceph_pagelist_alloc(GFP_NOFS);
3835 	if (!_pagelist)
3836 		return -ENOMEM;
3837 
3838 	reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);
3839 	if (!reply)
3840 		goto fail_msg;
3841 
3842 	/* placeholder for nr_caps */
3843 	err = ceph_pagelist_encode_32(_pagelist, 0);
3844 	if (err < 0)
3845 		goto fail;
3846 
3847 	if (recon_state->nr_caps) {
3848 		/* currently encoding caps */
3849 		err = ceph_pagelist_encode_32(recon_state->pagelist, 0);
3850 		if (err)
3851 			goto fail;
3852 	} else {
3853 		/* placeholder for nr_realms (currently encoding relams) */
3854 		err = ceph_pagelist_encode_32(_pagelist, 0);
3855 		if (err < 0)
3856 			goto fail;
3857 	}
3858 
3859 	err = ceph_pagelist_encode_8(recon_state->pagelist, 1);
3860 	if (err)
3861 		goto fail;
3862 
3863 	page = list_first_entry(&recon_state->pagelist->head, struct page, lru);
3864 	addr = kmap_atomic(page);
3865 	if (recon_state->nr_caps) {
3866 		/* currently encoding caps */
3867 		*addr = cpu_to_le32(recon_state->nr_caps);
3868 	} else {
3869 		/* currently encoding relams */
3870 		*(addr + 1) = cpu_to_le32(recon_state->nr_realms);
3871 	}
3872 	kunmap_atomic(addr);
3873 
3874 	reply->hdr.version = cpu_to_le16(5);
3875 	reply->hdr.compat_version = cpu_to_le16(4);
3876 
3877 	reply->hdr.data_len = cpu_to_le32(recon_state->pagelist->length);
3878 	ceph_msg_data_add_pagelist(reply, recon_state->pagelist);
3879 
3880 	ceph_con_send(&recon_state->session->s_con, reply);
3881 	ceph_pagelist_release(recon_state->pagelist);
3882 
3883 	recon_state->pagelist = _pagelist;
3884 	recon_state->nr_caps = 0;
3885 	recon_state->nr_realms = 0;
3886 	recon_state->msg_version = 5;
3887 	return 0;
3888 fail:
3889 	ceph_msg_put(reply);
3890 fail_msg:
3891 	ceph_pagelist_release(_pagelist);
3892 	return err;
3893 }
3894 
d_find_primary(struct inode * inode)3895 static struct dentry* d_find_primary(struct inode *inode)
3896 {
3897 	struct dentry *alias, *dn = NULL;
3898 
3899 	if (hlist_empty(&inode->i_dentry))
3900 		return NULL;
3901 
3902 	spin_lock(&inode->i_lock);
3903 	if (hlist_empty(&inode->i_dentry))
3904 		goto out_unlock;
3905 
3906 	if (S_ISDIR(inode->i_mode)) {
3907 		alias = hlist_entry(inode->i_dentry.first, struct dentry, d_u.d_alias);
3908 		if (!IS_ROOT(alias))
3909 			dn = dget(alias);
3910 		goto out_unlock;
3911 	}
3912 
3913 	hlist_for_each_entry(alias, &inode->i_dentry, d_u.d_alias) {
3914 		spin_lock(&alias->d_lock);
3915 		if (!d_unhashed(alias) &&
3916 		    (ceph_dentry(alias)->flags & CEPH_DENTRY_PRIMARY_LINK)) {
3917 			dn = dget_dlock(alias);
3918 		}
3919 		spin_unlock(&alias->d_lock);
3920 		if (dn)
3921 			break;
3922 	}
3923 out_unlock:
3924 	spin_unlock(&inode->i_lock);
3925 	return dn;
3926 }
3927 
3928 /*
3929  * Encode information about a cap for a reconnect with the MDS.
3930  */
reconnect_caps_cb(struct inode * inode,int mds,void * arg)3931 static int reconnect_caps_cb(struct inode *inode, int mds, void *arg)
3932 {
3933 	union {
3934 		struct ceph_mds_cap_reconnect v2;
3935 		struct ceph_mds_cap_reconnect_v1 v1;
3936 	} rec;
3937 	struct ceph_inode_info *ci = ceph_inode(inode);
3938 	struct ceph_reconnect_state *recon_state = arg;
3939 	struct ceph_pagelist *pagelist = recon_state->pagelist;
3940 	struct dentry *dentry;
3941 	struct ceph_cap *cap;
3942 	char *path;
3943 	int pathlen = 0, err;
3944 	u64 pathbase;
3945 	u64 snap_follows;
3946 
3947 	dentry = d_find_primary(inode);
3948 	if (dentry) {
3949 		/* set pathbase to parent dir when msg_version >= 2 */
3950 		path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase,
3951 					    recon_state->msg_version >= 2);
3952 		dput(dentry);
3953 		if (IS_ERR(path)) {
3954 			err = PTR_ERR(path);
3955 			goto out_err;
3956 		}
3957 	} else {
3958 		path = NULL;
3959 		pathbase = 0;
3960 	}
3961 
3962 	spin_lock(&ci->i_ceph_lock);
3963 	cap = __get_cap_for_mds(ci, mds);
3964 	if (!cap) {
3965 		spin_unlock(&ci->i_ceph_lock);
3966 		err = 0;
3967 		goto out_err;
3968 	}
3969 	dout(" adding %p ino %llx.%llx cap %p %lld %s\n",
3970 	     inode, ceph_vinop(inode), cap, cap->cap_id,
3971 	     ceph_cap_string(cap->issued));
3972 
3973 	cap->seq = 0;        /* reset cap seq */
3974 	cap->issue_seq = 0;  /* and issue_seq */
3975 	cap->mseq = 0;       /* and migrate_seq */
3976 	cap->cap_gen = atomic_read(&cap->session->s_cap_gen);
3977 
3978 	/* These are lost when the session goes away */
3979 	if (S_ISDIR(inode->i_mode)) {
3980 		if (cap->issued & CEPH_CAP_DIR_CREATE) {
3981 			ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns));
3982 			memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));
3983 		}
3984 		cap->issued &= ~CEPH_CAP_ANY_DIR_OPS;
3985 	}
3986 
3987 	if (recon_state->msg_version >= 2) {
3988 		rec.v2.cap_id = cpu_to_le64(cap->cap_id);
3989 		rec.v2.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
3990 		rec.v2.issued = cpu_to_le32(cap->issued);
3991 		rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
3992 		rec.v2.pathbase = cpu_to_le64(pathbase);
3993 		rec.v2.flock_len = (__force __le32)
3994 			((ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK) ? 0 : 1);
3995 	} else {
3996 		rec.v1.cap_id = cpu_to_le64(cap->cap_id);
3997 		rec.v1.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
3998 		rec.v1.issued = cpu_to_le32(cap->issued);
3999 		rec.v1.size = cpu_to_le64(i_size_read(inode));
4000 		ceph_encode_timespec64(&rec.v1.mtime, &inode->i_mtime);
4001 		ceph_encode_timespec64(&rec.v1.atime, &inode->i_atime);
4002 		rec.v1.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
4003 		rec.v1.pathbase = cpu_to_le64(pathbase);
4004 	}
4005 
4006 	if (list_empty(&ci->i_cap_snaps)) {
4007 		snap_follows = ci->i_head_snapc ? ci->i_head_snapc->seq : 0;
4008 	} else {
4009 		struct ceph_cap_snap *capsnap =
4010 			list_first_entry(&ci->i_cap_snaps,
4011 					 struct ceph_cap_snap, ci_item);
4012 		snap_follows = capsnap->follows;
4013 	}
4014 	spin_unlock(&ci->i_ceph_lock);
4015 
4016 	if (recon_state->msg_version >= 2) {
4017 		int num_fcntl_locks, num_flock_locks;
4018 		struct ceph_filelock *flocks = NULL;
4019 		size_t struct_len, total_len = sizeof(u64);
4020 		u8 struct_v = 0;
4021 
4022 encode_again:
4023 		if (rec.v2.flock_len) {
4024 			ceph_count_locks(inode, &num_fcntl_locks, &num_flock_locks);
4025 		} else {
4026 			num_fcntl_locks = 0;
4027 			num_flock_locks = 0;
4028 		}
4029 		if (num_fcntl_locks + num_flock_locks > 0) {
4030 			flocks = kmalloc_array(num_fcntl_locks + num_flock_locks,
4031 					       sizeof(struct ceph_filelock),
4032 					       GFP_NOFS);
4033 			if (!flocks) {
4034 				err = -ENOMEM;
4035 				goto out_err;
4036 			}
4037 			err = ceph_encode_locks_to_buffer(inode, flocks,
4038 							  num_fcntl_locks,
4039 							  num_flock_locks);
4040 			if (err) {
4041 				kfree(flocks);
4042 				flocks = NULL;
4043 				if (err == -ENOSPC)
4044 					goto encode_again;
4045 				goto out_err;
4046 			}
4047 		} else {
4048 			kfree(flocks);
4049 			flocks = NULL;
4050 		}
4051 
4052 		if (recon_state->msg_version >= 3) {
4053 			/* version, compat_version and struct_len */
4054 			total_len += 2 * sizeof(u8) + sizeof(u32);
4055 			struct_v = 2;
4056 		}
4057 		/*
4058 		 * number of encoded locks is stable, so copy to pagelist
4059 		 */
4060 		struct_len = 2 * sizeof(u32) +
4061 			    (num_fcntl_locks + num_flock_locks) *
4062 			    sizeof(struct ceph_filelock);
4063 		rec.v2.flock_len = cpu_to_le32(struct_len);
4064 
4065 		struct_len += sizeof(u32) + pathlen + sizeof(rec.v2);
4066 
4067 		if (struct_v >= 2)
4068 			struct_len += sizeof(u64); /* snap_follows */
4069 
4070 		total_len += struct_len;
4071 
4072 		if (pagelist->length + total_len > RECONNECT_MAX_SIZE) {
4073 			err = send_reconnect_partial(recon_state);
4074 			if (err)
4075 				goto out_freeflocks;
4076 			pagelist = recon_state->pagelist;
4077 		}
4078 
4079 		err = ceph_pagelist_reserve(pagelist, total_len);
4080 		if (err)
4081 			goto out_freeflocks;
4082 
4083 		ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
4084 		if (recon_state->msg_version >= 3) {
4085 			ceph_pagelist_encode_8(pagelist, struct_v);
4086 			ceph_pagelist_encode_8(pagelist, 1);
4087 			ceph_pagelist_encode_32(pagelist, struct_len);
4088 		}
4089 		ceph_pagelist_encode_string(pagelist, path, pathlen);
4090 		ceph_pagelist_append(pagelist, &rec, sizeof(rec.v2));
4091 		ceph_locks_to_pagelist(flocks, pagelist,
4092 				       num_fcntl_locks, num_flock_locks);
4093 		if (struct_v >= 2)
4094 			ceph_pagelist_encode_64(pagelist, snap_follows);
4095 out_freeflocks:
4096 		kfree(flocks);
4097 	} else {
4098 		err = ceph_pagelist_reserve(pagelist,
4099 					    sizeof(u64) + sizeof(u32) +
4100 					    pathlen + sizeof(rec.v1));
4101 		if (err)
4102 			goto out_err;
4103 
4104 		ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
4105 		ceph_pagelist_encode_string(pagelist, path, pathlen);
4106 		ceph_pagelist_append(pagelist, &rec, sizeof(rec.v1));
4107 	}
4108 
4109 out_err:
4110 	ceph_mdsc_free_path(path, pathlen);
4111 	if (!err)
4112 		recon_state->nr_caps++;
4113 	return err;
4114 }
4115 
encode_snap_realms(struct ceph_mds_client * mdsc,struct ceph_reconnect_state * recon_state)4116 static int encode_snap_realms(struct ceph_mds_client *mdsc,
4117 			      struct ceph_reconnect_state *recon_state)
4118 {
4119 	struct rb_node *p;
4120 	struct ceph_pagelist *pagelist = recon_state->pagelist;
4121 	int err = 0;
4122 
4123 	if (recon_state->msg_version >= 4) {
4124 		err = ceph_pagelist_encode_32(pagelist, mdsc->num_snap_realms);
4125 		if (err < 0)
4126 			goto fail;
4127 	}
4128 
4129 	/*
4130 	 * snaprealms.  we provide mds with the ino, seq (version), and
4131 	 * parent for all of our realms.  If the mds has any newer info,
4132 	 * it will tell us.
4133 	 */
4134 	for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
4135 		struct ceph_snap_realm *realm =
4136 		       rb_entry(p, struct ceph_snap_realm, node);
4137 		struct ceph_mds_snaprealm_reconnect sr_rec;
4138 
4139 		if (recon_state->msg_version >= 4) {
4140 			size_t need = sizeof(u8) * 2 + sizeof(u32) +
4141 				      sizeof(sr_rec);
4142 
4143 			if (pagelist->length + need > RECONNECT_MAX_SIZE) {
4144 				err = send_reconnect_partial(recon_state);
4145 				if (err)
4146 					goto fail;
4147 				pagelist = recon_state->pagelist;
4148 			}
4149 
4150 			err = ceph_pagelist_reserve(pagelist, need);
4151 			if (err)
4152 				goto fail;
4153 
4154 			ceph_pagelist_encode_8(pagelist, 1);
4155 			ceph_pagelist_encode_8(pagelist, 1);
4156 			ceph_pagelist_encode_32(pagelist, sizeof(sr_rec));
4157 		}
4158 
4159 		dout(" adding snap realm %llx seq %lld parent %llx\n",
4160 		     realm->ino, realm->seq, realm->parent_ino);
4161 		sr_rec.ino = cpu_to_le64(realm->ino);
4162 		sr_rec.seq = cpu_to_le64(realm->seq);
4163 		sr_rec.parent = cpu_to_le64(realm->parent_ino);
4164 
4165 		err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
4166 		if (err)
4167 			goto fail;
4168 
4169 		recon_state->nr_realms++;
4170 	}
4171 fail:
4172 	return err;
4173 }
4174 
4175 
4176 /*
4177  * If an MDS fails and recovers, clients need to reconnect in order to
4178  * reestablish shared state.  This includes all caps issued through
4179  * this session _and_ the snap_realm hierarchy.  Because it's not
4180  * clear which snap realms the mds cares about, we send everything we
4181  * know about.. that ensures we'll then get any new info the
4182  * recovering MDS might have.
4183  *
4184  * This is a relatively heavyweight operation, but it's rare.
4185  */
send_mds_reconnect(struct ceph_mds_client * mdsc,struct ceph_mds_session * session)4186 static void send_mds_reconnect(struct ceph_mds_client *mdsc,
4187 			       struct ceph_mds_session *session)
4188 {
4189 	struct ceph_msg *reply;
4190 	int mds = session->s_mds;
4191 	int err = -ENOMEM;
4192 	struct ceph_reconnect_state recon_state = {
4193 		.session = session,
4194 	};
4195 	LIST_HEAD(dispose);
4196 
4197 	pr_info("mds%d reconnect start\n", mds);
4198 
4199 	recon_state.pagelist = ceph_pagelist_alloc(GFP_NOFS);
4200 	if (!recon_state.pagelist)
4201 		goto fail_nopagelist;
4202 
4203 	reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);
4204 	if (!reply)
4205 		goto fail_nomsg;
4206 
4207 	xa_destroy(&session->s_delegated_inos);
4208 
4209 	mutex_lock(&session->s_mutex);
4210 	session->s_state = CEPH_MDS_SESSION_RECONNECTING;
4211 	session->s_seq = 0;
4212 
4213 	dout("session %p state %s\n", session,
4214 	     ceph_session_state_name(session->s_state));
4215 
4216 	atomic_inc(&session->s_cap_gen);
4217 
4218 	spin_lock(&session->s_cap_lock);
4219 	/* don't know if session is readonly */
4220 	session->s_readonly = 0;
4221 	/*
4222 	 * notify __ceph_remove_cap() that we are composing cap reconnect.
4223 	 * If a cap get released before being added to the cap reconnect,
4224 	 * __ceph_remove_cap() should skip queuing cap release.
4225 	 */
4226 	session->s_cap_reconnect = 1;
4227 	/* drop old cap expires; we're about to reestablish that state */
4228 	detach_cap_releases(session, &dispose);
4229 	spin_unlock(&session->s_cap_lock);
4230 	dispose_cap_releases(mdsc, &dispose);
4231 
4232 	/* trim unused caps to reduce MDS's cache rejoin time */
4233 	if (mdsc->fsc->sb->s_root)
4234 		shrink_dcache_parent(mdsc->fsc->sb->s_root);
4235 
4236 	ceph_con_close(&session->s_con);
4237 	ceph_con_open(&session->s_con,
4238 		      CEPH_ENTITY_TYPE_MDS, mds,
4239 		      ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
4240 
4241 	/* replay unsafe requests */
4242 	replay_unsafe_requests(mdsc, session);
4243 
4244 	ceph_early_kick_flushing_caps(mdsc, session);
4245 
4246 	down_read(&mdsc->snap_rwsem);
4247 
4248 	/* placeholder for nr_caps */
4249 	err = ceph_pagelist_encode_32(recon_state.pagelist, 0);
4250 	if (err)
4251 		goto fail;
4252 
4253 	if (test_bit(CEPHFS_FEATURE_MULTI_RECONNECT, &session->s_features)) {
4254 		recon_state.msg_version = 3;
4255 		recon_state.allow_multi = true;
4256 	} else if (session->s_con.peer_features & CEPH_FEATURE_MDSENC) {
4257 		recon_state.msg_version = 3;
4258 	} else {
4259 		recon_state.msg_version = 2;
4260 	}
4261 	/* trsaverse this session's caps */
4262 	err = ceph_iterate_session_caps(session, reconnect_caps_cb, &recon_state);
4263 
4264 	spin_lock(&session->s_cap_lock);
4265 	session->s_cap_reconnect = 0;
4266 	spin_unlock(&session->s_cap_lock);
4267 
4268 	if (err < 0)
4269 		goto fail;
4270 
4271 	/* check if all realms can be encoded into current message */
4272 	if (mdsc->num_snap_realms) {
4273 		size_t total_len =
4274 			recon_state.pagelist->length +
4275 			mdsc->num_snap_realms *
4276 			sizeof(struct ceph_mds_snaprealm_reconnect);
4277 		if (recon_state.msg_version >= 4) {
4278 			/* number of realms */
4279 			total_len += sizeof(u32);
4280 			/* version, compat_version and struct_len */
4281 			total_len += mdsc->num_snap_realms *
4282 				     (2 * sizeof(u8) + sizeof(u32));
4283 		}
4284 		if (total_len > RECONNECT_MAX_SIZE) {
4285 			if (!recon_state.allow_multi) {
4286 				err = -ENOSPC;
4287 				goto fail;
4288 			}
4289 			if (recon_state.nr_caps) {
4290 				err = send_reconnect_partial(&recon_state);
4291 				if (err)
4292 					goto fail;
4293 			}
4294 			recon_state.msg_version = 5;
4295 		}
4296 	}
4297 
4298 	err = encode_snap_realms(mdsc, &recon_state);
4299 	if (err < 0)
4300 		goto fail;
4301 
4302 	if (recon_state.msg_version >= 5) {
4303 		err = ceph_pagelist_encode_8(recon_state.pagelist, 0);
4304 		if (err < 0)
4305 			goto fail;
4306 	}
4307 
4308 	if (recon_state.nr_caps || recon_state.nr_realms) {
4309 		struct page *page =
4310 			list_first_entry(&recon_state.pagelist->head,
4311 					struct page, lru);
4312 		__le32 *addr = kmap_atomic(page);
4313 		if (recon_state.nr_caps) {
4314 			WARN_ON(recon_state.nr_realms != mdsc->num_snap_realms);
4315 			*addr = cpu_to_le32(recon_state.nr_caps);
4316 		} else if (recon_state.msg_version >= 4) {
4317 			*(addr + 1) = cpu_to_le32(recon_state.nr_realms);
4318 		}
4319 		kunmap_atomic(addr);
4320 	}
4321 
4322 	reply->hdr.version = cpu_to_le16(recon_state.msg_version);
4323 	if (recon_state.msg_version >= 4)
4324 		reply->hdr.compat_version = cpu_to_le16(4);
4325 
4326 	reply->hdr.data_len = cpu_to_le32(recon_state.pagelist->length);
4327 	ceph_msg_data_add_pagelist(reply, recon_state.pagelist);
4328 
4329 	ceph_con_send(&session->s_con, reply);
4330 
4331 	mutex_unlock(&session->s_mutex);
4332 
4333 	mutex_lock(&mdsc->mutex);
4334 	__wake_requests(mdsc, &session->s_waiting);
4335 	mutex_unlock(&mdsc->mutex);
4336 
4337 	up_read(&mdsc->snap_rwsem);
4338 	ceph_pagelist_release(recon_state.pagelist);
4339 	return;
4340 
4341 fail:
4342 	ceph_msg_put(reply);
4343 	up_read(&mdsc->snap_rwsem);
4344 	mutex_unlock(&session->s_mutex);
4345 fail_nomsg:
4346 	ceph_pagelist_release(recon_state.pagelist);
4347 fail_nopagelist:
4348 	pr_err("error %d preparing reconnect for mds%d\n", err, mds);
4349 	return;
4350 }
4351 
4352 
4353 /*
4354  * compare old and new mdsmaps, kicking requests
4355  * and closing out old connections as necessary
4356  *
4357  * called under mdsc->mutex.
4358  */
check_new_map(struct ceph_mds_client * mdsc,struct ceph_mdsmap * newmap,struct ceph_mdsmap * oldmap)4359 static void check_new_map(struct ceph_mds_client *mdsc,
4360 			  struct ceph_mdsmap *newmap,
4361 			  struct ceph_mdsmap *oldmap)
4362 {
4363 	int i, j, err;
4364 	int oldstate, newstate;
4365 	struct ceph_mds_session *s;
4366 	unsigned long targets[DIV_ROUND_UP(CEPH_MAX_MDS, sizeof(unsigned long))] = {0};
4367 
4368 	dout("check_new_map new %u old %u\n",
4369 	     newmap->m_epoch, oldmap->m_epoch);
4370 
4371 	if (newmap->m_info) {
4372 		for (i = 0; i < newmap->possible_max_rank; i++) {
4373 			for (j = 0; j < newmap->m_info[i].num_export_targets; j++)
4374 				set_bit(newmap->m_info[i].export_targets[j], targets);
4375 		}
4376 	}
4377 
4378 	for (i = 0; i < oldmap->possible_max_rank && i < mdsc->max_sessions; i++) {
4379 		if (!mdsc->sessions[i])
4380 			continue;
4381 		s = mdsc->sessions[i];
4382 		oldstate = ceph_mdsmap_get_state(oldmap, i);
4383 		newstate = ceph_mdsmap_get_state(newmap, i);
4384 
4385 		dout("check_new_map mds%d state %s%s -> %s%s (session %s)\n",
4386 		     i, ceph_mds_state_name(oldstate),
4387 		     ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "",
4388 		     ceph_mds_state_name(newstate),
4389 		     ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "",
4390 		     ceph_session_state_name(s->s_state));
4391 
4392 		if (i >= newmap->possible_max_rank) {
4393 			/* force close session for stopped mds */
4394 			ceph_get_mds_session(s);
4395 			__unregister_session(mdsc, s);
4396 			__wake_requests(mdsc, &s->s_waiting);
4397 			mutex_unlock(&mdsc->mutex);
4398 
4399 			mutex_lock(&s->s_mutex);
4400 			cleanup_session_requests(mdsc, s);
4401 			remove_session_caps(s);
4402 			mutex_unlock(&s->s_mutex);
4403 
4404 			ceph_put_mds_session(s);
4405 
4406 			mutex_lock(&mdsc->mutex);
4407 			kick_requests(mdsc, i);
4408 			continue;
4409 		}
4410 
4411 		if (memcmp(ceph_mdsmap_get_addr(oldmap, i),
4412 			   ceph_mdsmap_get_addr(newmap, i),
4413 			   sizeof(struct ceph_entity_addr))) {
4414 			/* just close it */
4415 			mutex_unlock(&mdsc->mutex);
4416 			mutex_lock(&s->s_mutex);
4417 			mutex_lock(&mdsc->mutex);
4418 			ceph_con_close(&s->s_con);
4419 			mutex_unlock(&s->s_mutex);
4420 			s->s_state = CEPH_MDS_SESSION_RESTARTING;
4421 		} else if (oldstate == newstate) {
4422 			continue;  /* nothing new with this mds */
4423 		}
4424 
4425 		/*
4426 		 * send reconnect?
4427 		 */
4428 		if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&
4429 		    newstate >= CEPH_MDS_STATE_RECONNECT) {
4430 			mutex_unlock(&mdsc->mutex);
4431 			clear_bit(i, targets);
4432 			send_mds_reconnect(mdsc, s);
4433 			mutex_lock(&mdsc->mutex);
4434 		}
4435 
4436 		/*
4437 		 * kick request on any mds that has gone active.
4438 		 */
4439 		if (oldstate < CEPH_MDS_STATE_ACTIVE &&
4440 		    newstate >= CEPH_MDS_STATE_ACTIVE) {
4441 			if (oldstate != CEPH_MDS_STATE_CREATING &&
4442 			    oldstate != CEPH_MDS_STATE_STARTING)
4443 				pr_info("mds%d recovery completed\n", s->s_mds);
4444 			kick_requests(mdsc, i);
4445 			mutex_unlock(&mdsc->mutex);
4446 			mutex_lock(&s->s_mutex);
4447 			mutex_lock(&mdsc->mutex);
4448 			ceph_kick_flushing_caps(mdsc, s);
4449 			mutex_unlock(&s->s_mutex);
4450 			wake_up_session_caps(s, RECONNECT);
4451 		}
4452 	}
4453 
4454 	/*
4455 	 * Only open and reconnect sessions that don't exist yet.
4456 	 */
4457 	for (i = 0; i < newmap->possible_max_rank; i++) {
4458 		/*
4459 		 * In case the import MDS is crashed just after
4460 		 * the EImportStart journal is flushed, so when
4461 		 * a standby MDS takes over it and is replaying
4462 		 * the EImportStart journal the new MDS daemon
4463 		 * will wait the client to reconnect it, but the
4464 		 * client may never register/open the session yet.
4465 		 *
4466 		 * Will try to reconnect that MDS daemon if the
4467 		 * rank number is in the export targets array and
4468 		 * is the up:reconnect state.
4469 		 */
4470 		newstate = ceph_mdsmap_get_state(newmap, i);
4471 		if (!test_bit(i, targets) || newstate != CEPH_MDS_STATE_RECONNECT)
4472 			continue;
4473 
4474 		/*
4475 		 * The session maybe registered and opened by some
4476 		 * requests which were choosing random MDSes during
4477 		 * the mdsc->mutex's unlock/lock gap below in rare
4478 		 * case. But the related MDS daemon will just queue
4479 		 * that requests and be still waiting for the client's
4480 		 * reconnection request in up:reconnect state.
4481 		 */
4482 		s = __ceph_lookup_mds_session(mdsc, i);
4483 		if (likely(!s)) {
4484 			s = __open_export_target_session(mdsc, i);
4485 			if (IS_ERR(s)) {
4486 				err = PTR_ERR(s);
4487 				pr_err("failed to open export target session, err %d\n",
4488 				       err);
4489 				continue;
4490 			}
4491 		}
4492 		dout("send reconnect to export target mds.%d\n", i);
4493 		mutex_unlock(&mdsc->mutex);
4494 		send_mds_reconnect(mdsc, s);
4495 		ceph_put_mds_session(s);
4496 		mutex_lock(&mdsc->mutex);
4497 	}
4498 
4499 	for (i = 0; i < newmap->possible_max_rank && i < mdsc->max_sessions; i++) {
4500 		s = mdsc->sessions[i];
4501 		if (!s)
4502 			continue;
4503 		if (!ceph_mdsmap_is_laggy(newmap, i))
4504 			continue;
4505 		if (s->s_state == CEPH_MDS_SESSION_OPEN ||
4506 		    s->s_state == CEPH_MDS_SESSION_HUNG ||
4507 		    s->s_state == CEPH_MDS_SESSION_CLOSING) {
4508 			dout(" connecting to export targets of laggy mds%d\n",
4509 			     i);
4510 			__open_export_target_sessions(mdsc, s);
4511 		}
4512 	}
4513 }
4514 
4515 
4516 
4517 /*
4518  * leases
4519  */
4520 
4521 /*
4522  * caller must hold session s_mutex, dentry->d_lock
4523  */
__ceph_mdsc_drop_dentry_lease(struct dentry * dentry)4524 void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry)
4525 {
4526 	struct ceph_dentry_info *di = ceph_dentry(dentry);
4527 
4528 	ceph_put_mds_session(di->lease_session);
4529 	di->lease_session = NULL;
4530 }
4531 
handle_lease(struct ceph_mds_client * mdsc,struct ceph_mds_session * session,struct ceph_msg * msg)4532 static void handle_lease(struct ceph_mds_client *mdsc,
4533 			 struct ceph_mds_session *session,
4534 			 struct ceph_msg *msg)
4535 {
4536 	struct super_block *sb = mdsc->fsc->sb;
4537 	struct inode *inode;
4538 	struct dentry *parent, *dentry;
4539 	struct ceph_dentry_info *di;
4540 	int mds = session->s_mds;
4541 	struct ceph_mds_lease *h = msg->front.iov_base;
4542 	u32 seq;
4543 	struct ceph_vino vino;
4544 	struct qstr dname;
4545 	int release = 0;
4546 
4547 	dout("handle_lease from mds%d\n", mds);
4548 
4549 	if (!ceph_inc_mds_stopping_blocker(mdsc, session))
4550 		return;
4551 
4552 	/* decode */
4553 	if (msg->front.iov_len < sizeof(*h) + sizeof(u32))
4554 		goto bad;
4555 	vino.ino = le64_to_cpu(h->ino);
4556 	vino.snap = CEPH_NOSNAP;
4557 	seq = le32_to_cpu(h->seq);
4558 	dname.len = get_unaligned_le32(h + 1);
4559 	if (msg->front.iov_len < sizeof(*h) + sizeof(u32) + dname.len)
4560 		goto bad;
4561 	dname.name = (void *)(h + 1) + sizeof(u32);
4562 
4563 	/* lookup inode */
4564 	inode = ceph_find_inode(sb, vino);
4565 	dout("handle_lease %s, ino %llx %p %.*s\n",
4566 	     ceph_lease_op_name(h->action), vino.ino, inode,
4567 	     dname.len, dname.name);
4568 
4569 	mutex_lock(&session->s_mutex);
4570 	if (!inode) {
4571 		dout("handle_lease no inode %llx\n", vino.ino);
4572 		goto release;
4573 	}
4574 
4575 	/* dentry */
4576 	parent = d_find_alias(inode);
4577 	if (!parent) {
4578 		dout("no parent dentry on inode %p\n", inode);
4579 		WARN_ON(1);
4580 		goto release;  /* hrm... */
4581 	}
4582 	dname.hash = full_name_hash(parent, dname.name, dname.len);
4583 	dentry = d_lookup(parent, &dname);
4584 	dput(parent);
4585 	if (!dentry)
4586 		goto release;
4587 
4588 	spin_lock(&dentry->d_lock);
4589 	di = ceph_dentry(dentry);
4590 	switch (h->action) {
4591 	case CEPH_MDS_LEASE_REVOKE:
4592 		if (di->lease_session == session) {
4593 			if (ceph_seq_cmp(di->lease_seq, seq) > 0)
4594 				h->seq = cpu_to_le32(di->lease_seq);
4595 			__ceph_mdsc_drop_dentry_lease(dentry);
4596 		}
4597 		release = 1;
4598 		break;
4599 
4600 	case CEPH_MDS_LEASE_RENEW:
4601 		if (di->lease_session == session &&
4602 		    di->lease_gen == atomic_read(&session->s_cap_gen) &&
4603 		    di->lease_renew_from &&
4604 		    di->lease_renew_after == 0) {
4605 			unsigned long duration =
4606 				msecs_to_jiffies(le32_to_cpu(h->duration_ms));
4607 
4608 			di->lease_seq = seq;
4609 			di->time = di->lease_renew_from + duration;
4610 			di->lease_renew_after = di->lease_renew_from +
4611 				(duration >> 1);
4612 			di->lease_renew_from = 0;
4613 		}
4614 		break;
4615 	}
4616 	spin_unlock(&dentry->d_lock);
4617 	dput(dentry);
4618 
4619 	if (!release)
4620 		goto out;
4621 
4622 release:
4623 	/* let's just reuse the same message */
4624 	h->action = CEPH_MDS_LEASE_REVOKE_ACK;
4625 	ceph_msg_get(msg);
4626 	ceph_con_send(&session->s_con, msg);
4627 
4628 out:
4629 	mutex_unlock(&session->s_mutex);
4630 	iput(inode);
4631 
4632 	ceph_dec_mds_stopping_blocker(mdsc);
4633 	return;
4634 
4635 bad:
4636 	ceph_dec_mds_stopping_blocker(mdsc);
4637 
4638 	pr_err("corrupt lease message\n");
4639 	ceph_msg_dump(msg);
4640 }
4641 
ceph_mdsc_lease_send_msg(struct ceph_mds_session * session,struct dentry * dentry,char action,u32 seq)4642 void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,
4643 			      struct dentry *dentry, char action,
4644 			      u32 seq)
4645 {
4646 	struct ceph_msg *msg;
4647 	struct ceph_mds_lease *lease;
4648 	struct inode *dir;
4649 	int len = sizeof(*lease) + sizeof(u32) + NAME_MAX;
4650 
4651 	dout("lease_send_msg identry %p %s to mds%d\n",
4652 	     dentry, ceph_lease_op_name(action), session->s_mds);
4653 
4654 	msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS, false);
4655 	if (!msg)
4656 		return;
4657 	lease = msg->front.iov_base;
4658 	lease->action = action;
4659 	lease->seq = cpu_to_le32(seq);
4660 
4661 	spin_lock(&dentry->d_lock);
4662 	dir = d_inode(dentry->d_parent);
4663 	lease->ino = cpu_to_le64(ceph_ino(dir));
4664 	lease->first = lease->last = cpu_to_le64(ceph_snap(dir));
4665 
4666 	put_unaligned_le32(dentry->d_name.len, lease + 1);
4667 	memcpy((void *)(lease + 1) + 4,
4668 	       dentry->d_name.name, dentry->d_name.len);
4669 	spin_unlock(&dentry->d_lock);
4670 
4671 	ceph_con_send(&session->s_con, msg);
4672 }
4673 
4674 /*
4675  * lock unlock the session, to wait ongoing session activities
4676  */
lock_unlock_session(struct ceph_mds_session * s)4677 static void lock_unlock_session(struct ceph_mds_session *s)
4678 {
4679 	mutex_lock(&s->s_mutex);
4680 	mutex_unlock(&s->s_mutex);
4681 }
4682 
maybe_recover_session(struct ceph_mds_client * mdsc)4683 static void maybe_recover_session(struct ceph_mds_client *mdsc)
4684 {
4685 	struct ceph_fs_client *fsc = mdsc->fsc;
4686 
4687 	if (!ceph_test_mount_opt(fsc, CLEANRECOVER))
4688 		return;
4689 
4690 	if (READ_ONCE(fsc->mount_state) != CEPH_MOUNT_MOUNTED)
4691 		return;
4692 
4693 	if (!READ_ONCE(fsc->blocklisted))
4694 		return;
4695 
4696 	pr_info("auto reconnect after blocklisted\n");
4697 	ceph_force_reconnect(fsc->sb);
4698 }
4699 
check_session_state(struct ceph_mds_session * s)4700 bool check_session_state(struct ceph_mds_session *s)
4701 {
4702 	switch (s->s_state) {
4703 	case CEPH_MDS_SESSION_OPEN:
4704 		if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
4705 			s->s_state = CEPH_MDS_SESSION_HUNG;
4706 			pr_info("mds%d hung\n", s->s_mds);
4707 		}
4708 		break;
4709 	case CEPH_MDS_SESSION_CLOSING:
4710 	case CEPH_MDS_SESSION_NEW:
4711 	case CEPH_MDS_SESSION_RESTARTING:
4712 	case CEPH_MDS_SESSION_CLOSED:
4713 	case CEPH_MDS_SESSION_REJECTED:
4714 		return false;
4715 	}
4716 
4717 	return true;
4718 }
4719 
4720 /*
4721  * If the sequence is incremented while we're waiting on a REQUEST_CLOSE reply,
4722  * then we need to retransmit that request.
4723  */
inc_session_sequence(struct ceph_mds_session * s)4724 void inc_session_sequence(struct ceph_mds_session *s)
4725 {
4726 	lockdep_assert_held(&s->s_mutex);
4727 
4728 	s->s_seq++;
4729 
4730 	if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
4731 		int ret;
4732 
4733 		dout("resending session close request for mds%d\n", s->s_mds);
4734 		ret = request_close_session(s);
4735 		if (ret < 0)
4736 			pr_err("unable to close session to mds%d: %d\n",
4737 			       s->s_mds, ret);
4738 	}
4739 }
4740 
4741 /*
4742  * delayed work -- periodically trim expired leases, renew caps with mds.  If
4743  * the @delay parameter is set to 0 or if it's more than 5 secs, the default
4744  * workqueue delay value of 5 secs will be used.
4745  */
schedule_delayed(struct ceph_mds_client * mdsc,unsigned long delay)4746 static void schedule_delayed(struct ceph_mds_client *mdsc, unsigned long delay)
4747 {
4748 	unsigned long max_delay = HZ * 5;
4749 
4750 	/* 5 secs default delay */
4751 	if (!delay || (delay > max_delay))
4752 		delay = max_delay;
4753 	schedule_delayed_work(&mdsc->delayed_work,
4754 			      round_jiffies_relative(delay));
4755 }
4756 
delayed_work(struct work_struct * work)4757 static void delayed_work(struct work_struct *work)
4758 {
4759 	struct ceph_mds_client *mdsc =
4760 		container_of(work, struct ceph_mds_client, delayed_work.work);
4761 	unsigned long delay;
4762 	int renew_interval;
4763 	int renew_caps;
4764 	int i;
4765 
4766 	dout("mdsc delayed_work\n");
4767 
4768 	if (mdsc->stopping >= CEPH_MDSC_STOPPING_FLUSHED)
4769 		return;
4770 
4771 	mutex_lock(&mdsc->mutex);
4772 	renew_interval = mdsc->mdsmap->m_session_timeout >> 2;
4773 	renew_caps = time_after_eq(jiffies, HZ*renew_interval +
4774 				   mdsc->last_renew_caps);
4775 	if (renew_caps)
4776 		mdsc->last_renew_caps = jiffies;
4777 
4778 	for (i = 0; i < mdsc->max_sessions; i++) {
4779 		struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
4780 		if (!s)
4781 			continue;
4782 
4783 		if (!check_session_state(s)) {
4784 			ceph_put_mds_session(s);
4785 			continue;
4786 		}
4787 		mutex_unlock(&mdsc->mutex);
4788 
4789 		mutex_lock(&s->s_mutex);
4790 		if (renew_caps)
4791 			send_renew_caps(mdsc, s);
4792 		else
4793 			ceph_con_keepalive(&s->s_con);
4794 		if (s->s_state == CEPH_MDS_SESSION_OPEN ||
4795 		    s->s_state == CEPH_MDS_SESSION_HUNG)
4796 			ceph_send_cap_releases(mdsc, s);
4797 		mutex_unlock(&s->s_mutex);
4798 		ceph_put_mds_session(s);
4799 
4800 		mutex_lock(&mdsc->mutex);
4801 	}
4802 	mutex_unlock(&mdsc->mutex);
4803 
4804 	delay = ceph_check_delayed_caps(mdsc);
4805 
4806 	ceph_queue_cap_reclaim_work(mdsc);
4807 
4808 	ceph_trim_snapid_map(mdsc);
4809 
4810 	maybe_recover_session(mdsc);
4811 
4812 	schedule_delayed(mdsc, delay);
4813 }
4814 
ceph_mdsc_init(struct ceph_fs_client * fsc)4815 int ceph_mdsc_init(struct ceph_fs_client *fsc)
4816 
4817 {
4818 	struct ceph_mds_client *mdsc;
4819 	int err;
4820 
4821 	mdsc = kzalloc(sizeof(struct ceph_mds_client), GFP_NOFS);
4822 	if (!mdsc)
4823 		return -ENOMEM;
4824 	mdsc->fsc = fsc;
4825 	mutex_init(&mdsc->mutex);
4826 	mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS);
4827 	if (!mdsc->mdsmap) {
4828 		err = -ENOMEM;
4829 		goto err_mdsc;
4830 	}
4831 
4832 	init_completion(&mdsc->safe_umount_waiters);
4833 	spin_lock_init(&mdsc->stopping_lock);
4834 	atomic_set(&mdsc->stopping_blockers, 0);
4835 	init_completion(&mdsc->stopping_waiter);
4836 	init_waitqueue_head(&mdsc->session_close_wq);
4837 	INIT_LIST_HEAD(&mdsc->waiting_for_map);
4838 	mdsc->quotarealms_inodes = RB_ROOT;
4839 	mutex_init(&mdsc->quotarealms_inodes_mutex);
4840 	init_rwsem(&mdsc->snap_rwsem);
4841 	mdsc->snap_realms = RB_ROOT;
4842 	INIT_LIST_HEAD(&mdsc->snap_empty);
4843 	spin_lock_init(&mdsc->snap_empty_lock);
4844 	mdsc->request_tree = RB_ROOT;
4845 	INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);
4846 	mdsc->last_renew_caps = jiffies;
4847 	INIT_LIST_HEAD(&mdsc->cap_delay_list);
4848 	INIT_LIST_HEAD(&mdsc->cap_wait_list);
4849 	spin_lock_init(&mdsc->cap_delay_lock);
4850 	INIT_LIST_HEAD(&mdsc->snap_flush_list);
4851 	spin_lock_init(&mdsc->snap_flush_lock);
4852 	mdsc->last_cap_flush_tid = 1;
4853 	INIT_LIST_HEAD(&mdsc->cap_flush_list);
4854 	INIT_LIST_HEAD(&mdsc->cap_dirty_migrating);
4855 	spin_lock_init(&mdsc->cap_dirty_lock);
4856 	init_waitqueue_head(&mdsc->cap_flushing_wq);
4857 	INIT_WORK(&mdsc->cap_reclaim_work, ceph_cap_reclaim_work);
4858 	err = ceph_metric_init(&mdsc->metric);
4859 	if (err)
4860 		goto err_mdsmap;
4861 
4862 	spin_lock_init(&mdsc->dentry_list_lock);
4863 	INIT_LIST_HEAD(&mdsc->dentry_leases);
4864 	INIT_LIST_HEAD(&mdsc->dentry_dir_leases);
4865 
4866 	ceph_caps_init(mdsc);
4867 	ceph_adjust_caps_max_min(mdsc, fsc->mount_options);
4868 
4869 	spin_lock_init(&mdsc->snapid_map_lock);
4870 	mdsc->snapid_map_tree = RB_ROOT;
4871 	INIT_LIST_HEAD(&mdsc->snapid_map_lru);
4872 
4873 	init_rwsem(&mdsc->pool_perm_rwsem);
4874 	mdsc->pool_perm_tree = RB_ROOT;
4875 
4876 	strscpy(mdsc->nodename, utsname()->nodename,
4877 		sizeof(mdsc->nodename));
4878 
4879 	fsc->mdsc = mdsc;
4880 	return 0;
4881 
4882 err_mdsmap:
4883 	kfree(mdsc->mdsmap);
4884 err_mdsc:
4885 	kfree(mdsc);
4886 	return err;
4887 }
4888 
4889 /*
4890  * Wait for safe replies on open mds requests.  If we time out, drop
4891  * all requests from the tree to avoid dangling dentry refs.
4892  */
wait_requests(struct ceph_mds_client * mdsc)4893 static void wait_requests(struct ceph_mds_client *mdsc)
4894 {
4895 	struct ceph_options *opts = mdsc->fsc->client->options;
4896 	struct ceph_mds_request *req;
4897 
4898 	mutex_lock(&mdsc->mutex);
4899 	if (__get_oldest_req(mdsc)) {
4900 		mutex_unlock(&mdsc->mutex);
4901 
4902 		dout("wait_requests waiting for requests\n");
4903 		wait_for_completion_timeout(&mdsc->safe_umount_waiters,
4904 				    ceph_timeout_jiffies(opts->mount_timeout));
4905 
4906 		/* tear down remaining requests */
4907 		mutex_lock(&mdsc->mutex);
4908 		while ((req = __get_oldest_req(mdsc))) {
4909 			dout("wait_requests timed out on tid %llu\n",
4910 			     req->r_tid);
4911 			list_del_init(&req->r_wait);
4912 			__unregister_request(mdsc, req);
4913 		}
4914 	}
4915 	mutex_unlock(&mdsc->mutex);
4916 	dout("wait_requests done\n");
4917 }
4918 
send_flush_mdlog(struct ceph_mds_session * s)4919 void send_flush_mdlog(struct ceph_mds_session *s)
4920 {
4921 	struct ceph_msg *msg;
4922 
4923 	/*
4924 	 * Pre-luminous MDS crashes when it sees an unknown session request
4925 	 */
4926 	if (!CEPH_HAVE_FEATURE(s->s_con.peer_features, SERVER_LUMINOUS))
4927 		return;
4928 
4929 	mutex_lock(&s->s_mutex);
4930 	dout("request mdlog flush to mds%d (%s)s seq %lld\n", s->s_mds,
4931 	     ceph_session_state_name(s->s_state), s->s_seq);
4932 	msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_FLUSH_MDLOG,
4933 				      s->s_seq);
4934 	if (!msg) {
4935 		pr_err("failed to request mdlog flush to mds%d (%s) seq %lld\n",
4936 		       s->s_mds, ceph_session_state_name(s->s_state), s->s_seq);
4937 	} else {
4938 		ceph_con_send(&s->s_con, msg);
4939 	}
4940 	mutex_unlock(&s->s_mutex);
4941 }
4942 
4943 /*
4944  * called before mount is ro, and before dentries are torn down.
4945  * (hmm, does this still race with new lookups?)
4946  */
ceph_mdsc_pre_umount(struct ceph_mds_client * mdsc)4947 void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc)
4948 {
4949 	dout("pre_umount\n");
4950 	mdsc->stopping = CEPH_MDSC_STOPPING_BEGIN;
4951 
4952 	ceph_mdsc_iterate_sessions(mdsc, send_flush_mdlog, true);
4953 	ceph_mdsc_iterate_sessions(mdsc, lock_unlock_session, false);
4954 	ceph_flush_dirty_caps(mdsc);
4955 	wait_requests(mdsc);
4956 
4957 	/*
4958 	 * wait for reply handlers to drop their request refs and
4959 	 * their inode/dcache refs
4960 	 */
4961 	ceph_msgr_flush();
4962 
4963 	ceph_cleanup_quotarealms_inodes(mdsc);
4964 }
4965 
4966 /*
4967  * flush the mdlog and wait for all write mds requests to flush.
4968  */
flush_mdlog_and_wait_mdsc_unsafe_requests(struct ceph_mds_client * mdsc,u64 want_tid)4969 static void flush_mdlog_and_wait_mdsc_unsafe_requests(struct ceph_mds_client *mdsc,
4970 						 u64 want_tid)
4971 {
4972 	struct ceph_mds_request *req = NULL, *nextreq;
4973 	struct ceph_mds_session *last_session = NULL;
4974 	struct rb_node *n;
4975 
4976 	mutex_lock(&mdsc->mutex);
4977 	dout("%s want %lld\n", __func__, want_tid);
4978 restart:
4979 	req = __get_oldest_req(mdsc);
4980 	while (req && req->r_tid <= want_tid) {
4981 		/* find next request */
4982 		n = rb_next(&req->r_node);
4983 		if (n)
4984 			nextreq = rb_entry(n, struct ceph_mds_request, r_node);
4985 		else
4986 			nextreq = NULL;
4987 		if (req->r_op != CEPH_MDS_OP_SETFILELOCK &&
4988 		    (req->r_op & CEPH_MDS_OP_WRITE)) {
4989 			struct ceph_mds_session *s = req->r_session;
4990 
4991 			if (!s) {
4992 				req = nextreq;
4993 				continue;
4994 			}
4995 
4996 			/* write op */
4997 			ceph_mdsc_get_request(req);
4998 			if (nextreq)
4999 				ceph_mdsc_get_request(nextreq);
5000 			s = ceph_get_mds_session(s);
5001 			mutex_unlock(&mdsc->mutex);
5002 
5003 			/* send flush mdlog request to MDS */
5004 			if (last_session != s) {
5005 				send_flush_mdlog(s);
5006 				ceph_put_mds_session(last_session);
5007 				last_session = s;
5008 			} else {
5009 				ceph_put_mds_session(s);
5010 			}
5011 			dout("%s wait on %llu (want %llu)\n", __func__,
5012 			     req->r_tid, want_tid);
5013 			wait_for_completion(&req->r_safe_completion);
5014 
5015 			mutex_lock(&mdsc->mutex);
5016 			ceph_mdsc_put_request(req);
5017 			if (!nextreq)
5018 				break;  /* next dne before, so we're done! */
5019 			if (RB_EMPTY_NODE(&nextreq->r_node)) {
5020 				/* next request was removed from tree */
5021 				ceph_mdsc_put_request(nextreq);
5022 				goto restart;
5023 			}
5024 			ceph_mdsc_put_request(nextreq);  /* won't go away */
5025 		}
5026 		req = nextreq;
5027 	}
5028 	mutex_unlock(&mdsc->mutex);
5029 	ceph_put_mds_session(last_session);
5030 	dout("%s done\n", __func__);
5031 }
5032 
ceph_mdsc_sync(struct ceph_mds_client * mdsc)5033 void ceph_mdsc_sync(struct ceph_mds_client *mdsc)
5034 {
5035 	u64 want_tid, want_flush;
5036 
5037 	if (READ_ONCE(mdsc->fsc->mount_state) >= CEPH_MOUNT_SHUTDOWN)
5038 		return;
5039 
5040 	dout("sync\n");
5041 	mutex_lock(&mdsc->mutex);
5042 	want_tid = mdsc->last_tid;
5043 	mutex_unlock(&mdsc->mutex);
5044 
5045 	ceph_flush_dirty_caps(mdsc);
5046 	spin_lock(&mdsc->cap_dirty_lock);
5047 	want_flush = mdsc->last_cap_flush_tid;
5048 	if (!list_empty(&mdsc->cap_flush_list)) {
5049 		struct ceph_cap_flush *cf =
5050 			list_last_entry(&mdsc->cap_flush_list,
5051 					struct ceph_cap_flush, g_list);
5052 		cf->wake = true;
5053 	}
5054 	spin_unlock(&mdsc->cap_dirty_lock);
5055 
5056 	dout("sync want tid %lld flush_seq %lld\n",
5057 	     want_tid, want_flush);
5058 
5059 	flush_mdlog_and_wait_mdsc_unsafe_requests(mdsc, want_tid);
5060 	wait_caps_flush(mdsc, want_flush);
5061 }
5062 
5063 /*
5064  * true if all sessions are closed, or we force unmount
5065  */
done_closing_sessions(struct ceph_mds_client * mdsc,int skipped)5066 static bool done_closing_sessions(struct ceph_mds_client *mdsc, int skipped)
5067 {
5068 	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)
5069 		return true;
5070 	return atomic_read(&mdsc->num_sessions) <= skipped;
5071 }
5072 
5073 /*
5074  * called after sb is ro or when metadata corrupted.
5075  */
ceph_mdsc_close_sessions(struct ceph_mds_client * mdsc)5076 void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc)
5077 {
5078 	struct ceph_options *opts = mdsc->fsc->client->options;
5079 	struct ceph_mds_session *session;
5080 	int i;
5081 	int skipped = 0;
5082 
5083 	dout("close_sessions\n");
5084 
5085 	/* close sessions */
5086 	mutex_lock(&mdsc->mutex);
5087 	for (i = 0; i < mdsc->max_sessions; i++) {
5088 		session = __ceph_lookup_mds_session(mdsc, i);
5089 		if (!session)
5090 			continue;
5091 		mutex_unlock(&mdsc->mutex);
5092 		mutex_lock(&session->s_mutex);
5093 		if (__close_session(mdsc, session) <= 0)
5094 			skipped++;
5095 		mutex_unlock(&session->s_mutex);
5096 		ceph_put_mds_session(session);
5097 		mutex_lock(&mdsc->mutex);
5098 	}
5099 	mutex_unlock(&mdsc->mutex);
5100 
5101 	dout("waiting for sessions to close\n");
5102 	wait_event_timeout(mdsc->session_close_wq,
5103 			   done_closing_sessions(mdsc, skipped),
5104 			   ceph_timeout_jiffies(opts->mount_timeout));
5105 
5106 	/* tear down remaining sessions */
5107 	mutex_lock(&mdsc->mutex);
5108 	for (i = 0; i < mdsc->max_sessions; i++) {
5109 		if (mdsc->sessions[i]) {
5110 			session = ceph_get_mds_session(mdsc->sessions[i]);
5111 			__unregister_session(mdsc, session);
5112 			mutex_unlock(&mdsc->mutex);
5113 			mutex_lock(&session->s_mutex);
5114 			remove_session_caps(session);
5115 			mutex_unlock(&session->s_mutex);
5116 			ceph_put_mds_session(session);
5117 			mutex_lock(&mdsc->mutex);
5118 		}
5119 	}
5120 	WARN_ON(!list_empty(&mdsc->cap_delay_list));
5121 	mutex_unlock(&mdsc->mutex);
5122 
5123 	ceph_cleanup_snapid_map(mdsc);
5124 	ceph_cleanup_global_and_empty_realms(mdsc);
5125 
5126 	cancel_work_sync(&mdsc->cap_reclaim_work);
5127 	cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
5128 
5129 	dout("stopped\n");
5130 }
5131 
ceph_mdsc_force_umount(struct ceph_mds_client * mdsc)5132 void ceph_mdsc_force_umount(struct ceph_mds_client *mdsc)
5133 {
5134 	struct ceph_mds_session *session;
5135 	int mds;
5136 
5137 	dout("force umount\n");
5138 
5139 	mutex_lock(&mdsc->mutex);
5140 	for (mds = 0; mds < mdsc->max_sessions; mds++) {
5141 		session = __ceph_lookup_mds_session(mdsc, mds);
5142 		if (!session)
5143 			continue;
5144 
5145 		if (session->s_state == CEPH_MDS_SESSION_REJECTED)
5146 			__unregister_session(mdsc, session);
5147 		__wake_requests(mdsc, &session->s_waiting);
5148 		mutex_unlock(&mdsc->mutex);
5149 
5150 		mutex_lock(&session->s_mutex);
5151 		__close_session(mdsc, session);
5152 		if (session->s_state == CEPH_MDS_SESSION_CLOSING) {
5153 			cleanup_session_requests(mdsc, session);
5154 			remove_session_caps(session);
5155 		}
5156 		mutex_unlock(&session->s_mutex);
5157 		ceph_put_mds_session(session);
5158 
5159 		mutex_lock(&mdsc->mutex);
5160 		kick_requests(mdsc, mds);
5161 	}
5162 	__wake_requests(mdsc, &mdsc->waiting_for_map);
5163 	mutex_unlock(&mdsc->mutex);
5164 }
5165 
ceph_mdsc_stop(struct ceph_mds_client * mdsc)5166 static void ceph_mdsc_stop(struct ceph_mds_client *mdsc)
5167 {
5168 	dout("stop\n");
5169 	/*
5170 	 * Make sure the delayed work stopped before releasing
5171 	 * the resources.
5172 	 *
5173 	 * Because the cancel_delayed_work_sync() will only
5174 	 * guarantee that the work finishes executing. But the
5175 	 * delayed work will re-arm itself again after that.
5176 	 */
5177 	flush_delayed_work(&mdsc->delayed_work);
5178 
5179 	if (mdsc->mdsmap)
5180 		ceph_mdsmap_destroy(mdsc->mdsmap);
5181 	kfree(mdsc->sessions);
5182 	ceph_caps_finalize(mdsc);
5183 	ceph_pool_perm_destroy(mdsc);
5184 }
5185 
ceph_mdsc_destroy(struct ceph_fs_client * fsc)5186 void ceph_mdsc_destroy(struct ceph_fs_client *fsc)
5187 {
5188 	struct ceph_mds_client *mdsc = fsc->mdsc;
5189 	dout("mdsc_destroy %p\n", mdsc);
5190 
5191 	if (!mdsc)
5192 		return;
5193 
5194 	/* flush out any connection work with references to us */
5195 	ceph_msgr_flush();
5196 
5197 	ceph_mdsc_stop(mdsc);
5198 
5199 	ceph_metric_destroy(&mdsc->metric);
5200 
5201 	fsc->mdsc = NULL;
5202 	kfree(mdsc);
5203 	dout("mdsc_destroy %p done\n", mdsc);
5204 }
5205 
ceph_mdsc_handle_fsmap(struct ceph_mds_client * mdsc,struct ceph_msg * msg)5206 void ceph_mdsc_handle_fsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
5207 {
5208 	struct ceph_fs_client *fsc = mdsc->fsc;
5209 	const char *mds_namespace = fsc->mount_options->mds_namespace;
5210 	void *p = msg->front.iov_base;
5211 	void *end = p + msg->front.iov_len;
5212 	u32 epoch;
5213 	u32 num_fs;
5214 	u32 mount_fscid = (u32)-1;
5215 	int err = -EINVAL;
5216 
5217 	ceph_decode_need(&p, end, sizeof(u32), bad);
5218 	epoch = ceph_decode_32(&p);
5219 
5220 	dout("handle_fsmap epoch %u\n", epoch);
5221 
5222 	/* struct_v, struct_cv, map_len, epoch, legacy_client_fscid */
5223 	ceph_decode_skip_n(&p, end, 2 + sizeof(u32) * 3, bad);
5224 
5225 	ceph_decode_32_safe(&p, end, num_fs, bad);
5226 	while (num_fs-- > 0) {
5227 		void *info_p, *info_end;
5228 		u32 info_len;
5229 		u32 fscid, namelen;
5230 
5231 		ceph_decode_need(&p, end, 2 + sizeof(u32), bad);
5232 		p += 2;		// info_v, info_cv
5233 		info_len = ceph_decode_32(&p);
5234 		ceph_decode_need(&p, end, info_len, bad);
5235 		info_p = p;
5236 		info_end = p + info_len;
5237 		p = info_end;
5238 
5239 		ceph_decode_need(&info_p, info_end, sizeof(u32) * 2, bad);
5240 		fscid = ceph_decode_32(&info_p);
5241 		namelen = ceph_decode_32(&info_p);
5242 		ceph_decode_need(&info_p, info_end, namelen, bad);
5243 
5244 		if (mds_namespace &&
5245 		    strlen(mds_namespace) == namelen &&
5246 		    !strncmp(mds_namespace, (char *)info_p, namelen)) {
5247 			mount_fscid = fscid;
5248 			break;
5249 		}
5250 	}
5251 
5252 	ceph_monc_got_map(&fsc->client->monc, CEPH_SUB_FSMAP, epoch);
5253 	if (mount_fscid != (u32)-1) {
5254 		fsc->client->monc.fs_cluster_id = mount_fscid;
5255 		ceph_monc_want_map(&fsc->client->monc, CEPH_SUB_MDSMAP,
5256 				   0, true);
5257 		ceph_monc_renew_subs(&fsc->client->monc);
5258 	} else {
5259 		err = -ENOENT;
5260 		goto err_out;
5261 	}
5262 	return;
5263 
5264 bad:
5265 	pr_err("error decoding fsmap %d. Shutting down mount.\n", err);
5266 	ceph_umount_begin(mdsc->fsc->sb);
5267 	ceph_msg_dump(msg);
5268 err_out:
5269 	mutex_lock(&mdsc->mutex);
5270 	mdsc->mdsmap_err = err;
5271 	__wake_requests(mdsc, &mdsc->waiting_for_map);
5272 	mutex_unlock(&mdsc->mutex);
5273 }
5274 
5275 /*
5276  * handle mds map update.
5277  */
ceph_mdsc_handle_mdsmap(struct ceph_mds_client * mdsc,struct ceph_msg * msg)5278 void ceph_mdsc_handle_mdsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
5279 {
5280 	u32 epoch;
5281 	u32 maplen;
5282 	void *p = msg->front.iov_base;
5283 	void *end = p + msg->front.iov_len;
5284 	struct ceph_mdsmap *newmap, *oldmap;
5285 	struct ceph_fsid fsid;
5286 	int err = -EINVAL;
5287 
5288 	ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad);
5289 	ceph_decode_copy(&p, &fsid, sizeof(fsid));
5290 	if (ceph_check_fsid(mdsc->fsc->client, &fsid) < 0)
5291 		return;
5292 	epoch = ceph_decode_32(&p);
5293 	maplen = ceph_decode_32(&p);
5294 	dout("handle_map epoch %u len %d\n", epoch, (int)maplen);
5295 
5296 	/* do we need it? */
5297 	mutex_lock(&mdsc->mutex);
5298 	if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {
5299 		dout("handle_map epoch %u <= our %u\n",
5300 		     epoch, mdsc->mdsmap->m_epoch);
5301 		mutex_unlock(&mdsc->mutex);
5302 		return;
5303 	}
5304 
5305 	newmap = ceph_mdsmap_decode(&p, end, ceph_msgr2(mdsc->fsc->client));
5306 	if (IS_ERR(newmap)) {
5307 		err = PTR_ERR(newmap);
5308 		goto bad_unlock;
5309 	}
5310 
5311 	/* swap into place */
5312 	if (mdsc->mdsmap) {
5313 		oldmap = mdsc->mdsmap;
5314 		mdsc->mdsmap = newmap;
5315 		check_new_map(mdsc, newmap, oldmap);
5316 		ceph_mdsmap_destroy(oldmap);
5317 	} else {
5318 		mdsc->mdsmap = newmap;  /* first mds map */
5319 	}
5320 	mdsc->fsc->max_file_size = min((loff_t)mdsc->mdsmap->m_max_file_size,
5321 					MAX_LFS_FILESIZE);
5322 
5323 	__wake_requests(mdsc, &mdsc->waiting_for_map);
5324 	ceph_monc_got_map(&mdsc->fsc->client->monc, CEPH_SUB_MDSMAP,
5325 			  mdsc->mdsmap->m_epoch);
5326 
5327 	mutex_unlock(&mdsc->mutex);
5328 	schedule_delayed(mdsc, 0);
5329 	return;
5330 
5331 bad_unlock:
5332 	mutex_unlock(&mdsc->mutex);
5333 bad:
5334 	pr_err("error decoding mdsmap %d. Shutting down mount.\n", err);
5335 	ceph_umount_begin(mdsc->fsc->sb);
5336 	ceph_msg_dump(msg);
5337 	return;
5338 }
5339 
mds_get_con(struct ceph_connection * con)5340 static struct ceph_connection *mds_get_con(struct ceph_connection *con)
5341 {
5342 	struct ceph_mds_session *s = con->private;
5343 
5344 	if (ceph_get_mds_session(s))
5345 		return con;
5346 	return NULL;
5347 }
5348 
mds_put_con(struct ceph_connection * con)5349 static void mds_put_con(struct ceph_connection *con)
5350 {
5351 	struct ceph_mds_session *s = con->private;
5352 
5353 	ceph_put_mds_session(s);
5354 }
5355 
5356 /*
5357  * if the client is unresponsive for long enough, the mds will kill
5358  * the session entirely.
5359  */
mds_peer_reset(struct ceph_connection * con)5360 static void mds_peer_reset(struct ceph_connection *con)
5361 {
5362 	struct ceph_mds_session *s = con->private;
5363 	struct ceph_mds_client *mdsc = s->s_mdsc;
5364 
5365 	pr_warn("mds%d closed our session\n", s->s_mds);
5366 	if (READ_ONCE(mdsc->fsc->mount_state) != CEPH_MOUNT_FENCE_IO)
5367 		send_mds_reconnect(mdsc, s);
5368 }
5369 
mds_dispatch(struct ceph_connection * con,struct ceph_msg * msg)5370 static void mds_dispatch(struct ceph_connection *con, struct ceph_msg *msg)
5371 {
5372 	struct ceph_mds_session *s = con->private;
5373 	struct ceph_mds_client *mdsc = s->s_mdsc;
5374 	int type = le16_to_cpu(msg->hdr.type);
5375 
5376 	mutex_lock(&mdsc->mutex);
5377 	if (__verify_registered_session(mdsc, s) < 0) {
5378 		mutex_unlock(&mdsc->mutex);
5379 		goto out;
5380 	}
5381 	mutex_unlock(&mdsc->mutex);
5382 
5383 	switch (type) {
5384 	case CEPH_MSG_MDS_MAP:
5385 		ceph_mdsc_handle_mdsmap(mdsc, msg);
5386 		break;
5387 	case CEPH_MSG_FS_MAP_USER:
5388 		ceph_mdsc_handle_fsmap(mdsc, msg);
5389 		break;
5390 	case CEPH_MSG_CLIENT_SESSION:
5391 		handle_session(s, msg);
5392 		break;
5393 	case CEPH_MSG_CLIENT_REPLY:
5394 		handle_reply(s, msg);
5395 		break;
5396 	case CEPH_MSG_CLIENT_REQUEST_FORWARD:
5397 		handle_forward(mdsc, s, msg);
5398 		break;
5399 	case CEPH_MSG_CLIENT_CAPS:
5400 		ceph_handle_caps(s, msg);
5401 		break;
5402 	case CEPH_MSG_CLIENT_SNAP:
5403 		ceph_handle_snap(mdsc, s, msg);
5404 		break;
5405 	case CEPH_MSG_CLIENT_LEASE:
5406 		handle_lease(mdsc, s, msg);
5407 		break;
5408 	case CEPH_MSG_CLIENT_QUOTA:
5409 		ceph_handle_quota(mdsc, s, msg);
5410 		break;
5411 
5412 	default:
5413 		pr_err("received unknown message type %d %s\n", type,
5414 		       ceph_msg_type_name(type));
5415 	}
5416 out:
5417 	ceph_msg_put(msg);
5418 }
5419 
5420 /*
5421  * authentication
5422  */
5423 
5424 /*
5425  * Note: returned pointer is the address of a structure that's
5426  * managed separately.  Caller must *not* attempt to free it.
5427  */
5428 static struct ceph_auth_handshake *
mds_get_authorizer(struct ceph_connection * con,int * proto,int force_new)5429 mds_get_authorizer(struct ceph_connection *con, int *proto, int force_new)
5430 {
5431 	struct ceph_mds_session *s = con->private;
5432 	struct ceph_mds_client *mdsc = s->s_mdsc;
5433 	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
5434 	struct ceph_auth_handshake *auth = &s->s_auth;
5435 	int ret;
5436 
5437 	ret = __ceph_auth_get_authorizer(ac, auth, CEPH_ENTITY_TYPE_MDS,
5438 					 force_new, proto, NULL, NULL);
5439 	if (ret)
5440 		return ERR_PTR(ret);
5441 
5442 	return auth;
5443 }
5444 
mds_add_authorizer_challenge(struct ceph_connection * con,void * challenge_buf,int challenge_buf_len)5445 static int mds_add_authorizer_challenge(struct ceph_connection *con,
5446 				    void *challenge_buf, int challenge_buf_len)
5447 {
5448 	struct ceph_mds_session *s = con->private;
5449 	struct ceph_mds_client *mdsc = s->s_mdsc;
5450 	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
5451 
5452 	return ceph_auth_add_authorizer_challenge(ac, s->s_auth.authorizer,
5453 					    challenge_buf, challenge_buf_len);
5454 }
5455 
mds_verify_authorizer_reply(struct ceph_connection * con)5456 static int mds_verify_authorizer_reply(struct ceph_connection *con)
5457 {
5458 	struct ceph_mds_session *s = con->private;
5459 	struct ceph_mds_client *mdsc = s->s_mdsc;
5460 	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
5461 	struct ceph_auth_handshake *auth = &s->s_auth;
5462 
5463 	return ceph_auth_verify_authorizer_reply(ac, auth->authorizer,
5464 		auth->authorizer_reply_buf, auth->authorizer_reply_buf_len,
5465 		NULL, NULL, NULL, NULL);
5466 }
5467 
mds_invalidate_authorizer(struct ceph_connection * con)5468 static int mds_invalidate_authorizer(struct ceph_connection *con)
5469 {
5470 	struct ceph_mds_session *s = con->private;
5471 	struct ceph_mds_client *mdsc = s->s_mdsc;
5472 	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
5473 
5474 	ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS);
5475 
5476 	return ceph_monc_validate_auth(&mdsc->fsc->client->monc);
5477 }
5478 
mds_get_auth_request(struct ceph_connection * con,void * buf,int * buf_len,void ** authorizer,int * authorizer_len)5479 static int mds_get_auth_request(struct ceph_connection *con,
5480 				void *buf, int *buf_len,
5481 				void **authorizer, int *authorizer_len)
5482 {
5483 	struct ceph_mds_session *s = con->private;
5484 	struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;
5485 	struct ceph_auth_handshake *auth = &s->s_auth;
5486 	int ret;
5487 
5488 	ret = ceph_auth_get_authorizer(ac, auth, CEPH_ENTITY_TYPE_MDS,
5489 				       buf, buf_len);
5490 	if (ret)
5491 		return ret;
5492 
5493 	*authorizer = auth->authorizer_buf;
5494 	*authorizer_len = auth->authorizer_buf_len;
5495 	return 0;
5496 }
5497 
mds_handle_auth_reply_more(struct ceph_connection * con,void * reply,int reply_len,void * buf,int * buf_len,void ** authorizer,int * authorizer_len)5498 static int mds_handle_auth_reply_more(struct ceph_connection *con,
5499 				      void *reply, int reply_len,
5500 				      void *buf, int *buf_len,
5501 				      void **authorizer, int *authorizer_len)
5502 {
5503 	struct ceph_mds_session *s = con->private;
5504 	struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;
5505 	struct ceph_auth_handshake *auth = &s->s_auth;
5506 	int ret;
5507 
5508 	ret = ceph_auth_handle_svc_reply_more(ac, auth, reply, reply_len,
5509 					      buf, buf_len);
5510 	if (ret)
5511 		return ret;
5512 
5513 	*authorizer = auth->authorizer_buf;
5514 	*authorizer_len = auth->authorizer_buf_len;
5515 	return 0;
5516 }
5517 
mds_handle_auth_done(struct ceph_connection * con,u64 global_id,void * reply,int reply_len,u8 * session_key,int * session_key_len,u8 * con_secret,int * con_secret_len)5518 static int mds_handle_auth_done(struct ceph_connection *con,
5519 				u64 global_id, void *reply, int reply_len,
5520 				u8 *session_key, int *session_key_len,
5521 				u8 *con_secret, int *con_secret_len)
5522 {
5523 	struct ceph_mds_session *s = con->private;
5524 	struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;
5525 	struct ceph_auth_handshake *auth = &s->s_auth;
5526 
5527 	return ceph_auth_handle_svc_reply_done(ac, auth, reply, reply_len,
5528 					       session_key, session_key_len,
5529 					       con_secret, con_secret_len);
5530 }
5531 
mds_handle_auth_bad_method(struct ceph_connection * con,int used_proto,int result,const int * allowed_protos,int proto_cnt,const int * allowed_modes,int mode_cnt)5532 static int mds_handle_auth_bad_method(struct ceph_connection *con,
5533 				      int used_proto, int result,
5534 				      const int *allowed_protos, int proto_cnt,
5535 				      const int *allowed_modes, int mode_cnt)
5536 {
5537 	struct ceph_mds_session *s = con->private;
5538 	struct ceph_mon_client *monc = &s->s_mdsc->fsc->client->monc;
5539 	int ret;
5540 
5541 	if (ceph_auth_handle_bad_authorizer(monc->auth, CEPH_ENTITY_TYPE_MDS,
5542 					    used_proto, result,
5543 					    allowed_protos, proto_cnt,
5544 					    allowed_modes, mode_cnt)) {
5545 		ret = ceph_monc_validate_auth(monc);
5546 		if (ret)
5547 			return ret;
5548 	}
5549 
5550 	return -EACCES;
5551 }
5552 
mds_alloc_msg(struct ceph_connection * con,struct ceph_msg_header * hdr,int * skip)5553 static struct ceph_msg *mds_alloc_msg(struct ceph_connection *con,
5554 				struct ceph_msg_header *hdr, int *skip)
5555 {
5556 	struct ceph_msg *msg;
5557 	int type = (int) le16_to_cpu(hdr->type);
5558 	int front_len = (int) le32_to_cpu(hdr->front_len);
5559 
5560 	if (con->in_msg)
5561 		return con->in_msg;
5562 
5563 	*skip = 0;
5564 	msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
5565 	if (!msg) {
5566 		pr_err("unable to allocate msg type %d len %d\n",
5567 		       type, front_len);
5568 		return NULL;
5569 	}
5570 
5571 	return msg;
5572 }
5573 
mds_sign_message(struct ceph_msg * msg)5574 static int mds_sign_message(struct ceph_msg *msg)
5575 {
5576        struct ceph_mds_session *s = msg->con->private;
5577        struct ceph_auth_handshake *auth = &s->s_auth;
5578 
5579        return ceph_auth_sign_message(auth, msg);
5580 }
5581 
mds_check_message_signature(struct ceph_msg * msg)5582 static int mds_check_message_signature(struct ceph_msg *msg)
5583 {
5584        struct ceph_mds_session *s = msg->con->private;
5585        struct ceph_auth_handshake *auth = &s->s_auth;
5586 
5587        return ceph_auth_check_message_signature(auth, msg);
5588 }
5589 
5590 static const struct ceph_connection_operations mds_con_ops = {
5591 	.get = mds_get_con,
5592 	.put = mds_put_con,
5593 	.alloc_msg = mds_alloc_msg,
5594 	.dispatch = mds_dispatch,
5595 	.peer_reset = mds_peer_reset,
5596 	.get_authorizer = mds_get_authorizer,
5597 	.add_authorizer_challenge = mds_add_authorizer_challenge,
5598 	.verify_authorizer_reply = mds_verify_authorizer_reply,
5599 	.invalidate_authorizer = mds_invalidate_authorizer,
5600 	.sign_message = mds_sign_message,
5601 	.check_message_signature = mds_check_message_signature,
5602 	.get_auth_request = mds_get_auth_request,
5603 	.handle_auth_reply_more = mds_handle_auth_reply_more,
5604 	.handle_auth_done = mds_handle_auth_done,
5605 	.handle_auth_bad_method = mds_handle_auth_bad_method,
5606 };
5607 
5608 /* eof */
5609