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