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