brintos / linux-shallow public Read only

1// SPDX-License-Identifier: GPL-2.02#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 for32 * managing the file system namespace (the directory hierarchy and33 * inodes) and for coordinating shared access to storage.  Metadata is34 * partitioning hierarchically across a number of servers, and that35 * partition varies over time as the cluster adjusts the distribution36 * in order to balance load.37 *38 * The MDS client is primarily responsible to managing synchronous39 * metadata requests for operations like open, unlink, and so forth.40 * If there is a MDS failure, we find out about it when we (possibly41 * request and) receive a new MDS map, and can resubmit affected42 * requests.43 *44 * For the most part, though, we take advantage of a lossless45 * communications channel to the MDS, and do not need to worry about46 * 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 ensure50 * any capabilities or leases we have been issues remain valid.  If51 * the session times out and goes stale, our leases and capabilities52 * are no longer valid.53 */54 55struct 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 63static void __wake_requests(struct ceph_mds_client *mdsc,64			    struct list_head *head);65static void ceph_cap_release_work(struct work_struct *work);66static void ceph_cap_reclaim_work(struct work_struct *work);67 68static const struct ceph_connection_operations mds_con_ops;69 70 71/*72 * mds reply parsing73 */74 75static 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 only84	 * 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;94bad:95	return -EIO;96}97 98/*99 * parse individual inode info100 */101static 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 understand114		 * 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		} else244			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;276bad:277	err = -EIO;278out_bad:279	return err;280}281 282static 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 understand292		 * 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;308bad:309	return -EIO;310}311 312static 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 understand327		 * 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;356bad:357	return -EIO;358}359 360/*361 * parse a normal reply, which may contain a (dir+)dentry and/or a362 * target inode.363 */364static 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 400bad:401	err = -EIO;402out_bad:403	pr_err("problem parsing mds trace %d\n", err);404	return err;405}406 407/*408 * parse readdir results409 */410static 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 them473		 * 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 as482		 * when the human readable name length is in range of483		 * (CEPH_NOHASH_NAME_MAX, CEPH_NOHASH_NAME_MAX + SHA256_DIGEST_SIZE),484		 * then the copy in ceph_fname_to_usr will corrupt the485		 * data if there has no encryption key.486		 *487		 * Just set the no_copy flag and then if there has no488		 * encryption key the oname.name will be assigned to489		 * _name always.490		 */491		fname.no_copy = true;492		if (altname_len == 0) {493			/*494			 * Set tname to _name, and this will be used495			 * to do the base64_decode in-place. It's496			 * safe because the decoded string should497			 * always be shorter, which is 3/4 of origin498			 * string.499			 */500			tname.name = _name;501 502			/*503			 * Set oname to _name too, and this will be504			 * 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-place511			 * 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 536done:537	/* Skip over any unrecognized fields */538	*p = end;539	return 0;540 541bad:542	err = -EIO;543out_bad:544	pr_err_client(cl, "problem parsing dir contents %d\n", err);545	return err;546}547 548/*549 * parse fcntl F_GETLK results550 */551static 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;563bad:564	return -EIO;565}566 567 568#if BITS_PER_LONG == 64569 570#define DELEGATED_INO_AVAILABLE		xa_mk_value(1)571 572static 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;609bad:610	return -EIO;611}612 613u64 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 626int 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, just634 * ignore delegated_inos on 32 bit arch. Maybe eventually add xarrays for top635 * and bottom words?636 */637static 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;646bad:647	return -EIO;648}649 650u64 ceph_get_deleg_ino(struct ceph_mds_session *s)651{652	return 0;653}654 655int 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 results663 */664static 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;696bad:697	return -EIO;698}699 700static 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	}718bad:719	return -EIO;720}721 722/*723 * parse extra results724 */725static 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	else741		return -EIO;742}743 744/*745 * parse entire mds reply746 */747static 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 788bad:789	err = -EIO;790out_bad:791	pr_err_client(cl, "mds parse_reply err %d\n", err);792	ceph_msg_dump(msg);793	return err;794}795 796static 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 reply818 * from MDS and just drop all the links and unhash the dentry and then819 * succeeds immediately.820 *821 * For any new create/link/rename,etc requests followed by using the822 * same file names we must wait for the first reply of the inflight823 * unlink request, or the MDS possibly will fail these following824 * requests with -EEXIST if the inflight async unlink request was825 * delayed for some reasons.826 *827 * And the worst case is that for the none async openc request it will828 * successfully open the file if the CDentry hasn't been unlinked yet,829 * but later the previous delayed async unlink request will remove the830 * CDenty. That means the just created file is possiblly deleted later831 * by accident.832 *833 * We need to wait for the inflight async unlink requests to finish834 * when creating new files/directories by using the same file names.835 */836int 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;873next: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 * sessions893 */894const 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 910struct 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 917void 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->mutex933 */934struct 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 942static bool __have_session(struct ceph_mds_client *mdsc, int mds)943{944	if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])945		return false;946	else947		return true;948}949 950static 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 */963static 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 1028fail_realloc:1029	kfree(s);1030	return ERR_PTR(-ENOMEM);1031}1032 1033/*1034 * called under mdsc->mutex1035 */1036static 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->mutex1051 */1052static 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 1060void 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 1087void 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_dir1115		 * separately, since r_old_dentry's d_parent may have1116		 * changed between the dir mutex being dropped and1117		 * 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 1138DEFINE_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 */1145static struct ceph_mds_request *1146lookup_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 directory1159 * are modifying (if any).1160 *1161 * Called under mdsc->mutex.1162 */1163static 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 1204static 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 a1254 * non-snapshot inode. We do this using the rcu_read_lock (which must be held1255 * when calling this) to ensure that the objects won't disappear while we're1256 * working with them. Once we hit a candidate dentry, we attempt to take a1257 * reference to it, and return that as the result.1258 */1259static 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 the1276 * request (e.g., due to a prior forward hint from the mds), use that.1277 * Otherwise, consult frag tree and/or caps to identify the1278 * appropriate mds.  If all else fails, choose randomly.1279 *1280 * Called under mdsc->mutex.1281 */1282static 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 have1300	 * 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 requests1341			 * 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 be1390			 * replicated, then we want to look for the1391			 * 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);1424out:1425	iput(inode);1426	return mds;1427 1428random: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 messages1440 */1441struct 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 1460static 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)1462static 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 1491static 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)1493static 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_OPEN1535 * to include additional client metadata fields.1536 */1537static struct ceph_msg *1538create_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, using1601	 * the format that userspace expects for map<string, string>1602	 *1603	 * ClientSession messages with metadata are v71604	 */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->mutex1660 */1661static 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 mds1689 *1690 * called under mdsc->mutex1691 */1692static struct ceph_mds_session *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 1714struct ceph_mds_session *1715ceph_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 1729static 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 1750void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,1751					   struct ceph_mds_session *session)1752{1753	mutex_lock(&mdsc->mutex);1754	__open_export_target_sessions(mdsc, session);1755	mutex_unlock(&mdsc->mutex);1756}1757 1758/*1759 * session caps1760 */1761 1762static void detach_cap_releases(struct ceph_mds_session *session,1763				struct list_head *target)1764{1765	struct ceph_client *cl = session->s_mdsc->fsc->client;1766 1767	lockdep_assert_held(&session->s_cap_lock);1768 1769	list_splice_init(&session->s_cap_releases, target);1770	session->s_num_cap_releases = 0;1771	doutc(cl, "mds%d\n", session->s_mds);1772}1773 1774static void dispose_cap_releases(struct ceph_mds_client *mdsc,1775				 struct list_head *dispose)1776{1777	while (!list_empty(dispose)) {1778		struct ceph_cap *cap;1779		/* zero out the in-progress message */1780		cap = list_first_entry(dispose, struct ceph_cap, session_caps);1781		list_del(&cap->session_caps);1782		ceph_put_cap(mdsc, cap);1783	}1784}1785 1786static void cleanup_session_requests(struct ceph_mds_client *mdsc,1787				     struct ceph_mds_session *session)1788{1789	struct ceph_client *cl = mdsc->fsc->client;1790	struct ceph_mds_request *req;1791	struct rb_node *p;1792 1793	doutc(cl, "mds%d\n", session->s_mds);1794	mutex_lock(&mdsc->mutex);1795	while (!list_empty(&session->s_unsafe)) {1796		req = list_first_entry(&session->s_unsafe,1797				       struct ceph_mds_request, r_unsafe_item);1798		pr_warn_ratelimited_client(cl, " dropping unsafe request %llu\n",1799					   req->r_tid);1800		if (req->r_target_inode)1801			mapping_set_error(req->r_target_inode->i_mapping, -EIO);1802		if (req->r_unsafe_dir)1803			mapping_set_error(req->r_unsafe_dir->i_mapping, -EIO);1804		__unregister_request(mdsc, req);1805	}1806	/* zero r_attempts, so kick_requests() will re-send requests */1807	p = rb_first(&mdsc->request_tree);1808	while (p) {1809		req = rb_entry(p, struct ceph_mds_request, r_node);1810		p = rb_next(p);1811		if (req->r_session &&1812		    req->r_session->s_mds == session->s_mds)1813			req->r_attempts = 0;1814	}1815	mutex_unlock(&mdsc->mutex);1816}1817 1818/*1819 * Helper to safely iterate over all caps associated with a session, with1820 * special care taken to handle a racing __ceph_remove_cap().1821 *1822 * Caller must hold session s_mutex.1823 */1824int ceph_iterate_session_caps(struct ceph_mds_session *session,1825			      int (*cb)(struct inode *, int mds, void *),1826			      void *arg)1827{1828	struct ceph_client *cl = session->s_mdsc->fsc->client;1829	struct list_head *p;1830	struct ceph_cap *cap;1831	struct inode *inode, *last_inode = NULL;1832	struct ceph_cap *old_cap = NULL;1833	int ret;1834 1835	doutc(cl, "%p mds%d\n", session, session->s_mds);1836	spin_lock(&session->s_cap_lock);1837	p = session->s_caps.next;1838	while (p != &session->s_caps) {1839		int mds;1840 1841		cap = list_entry(p, struct ceph_cap, session_caps);1842		inode = igrab(&cap->ci->netfs.inode);1843		if (!inode) {1844			p = p->next;1845			continue;1846		}1847		session->s_cap_iterator = cap;1848		mds = cap->mds;1849		spin_unlock(&session->s_cap_lock);1850 1851		if (last_inode) {1852			iput(last_inode);1853			last_inode = NULL;1854		}1855		if (old_cap) {1856			ceph_put_cap(session->s_mdsc, old_cap);1857			old_cap = NULL;1858		}1859 1860		ret = cb(inode, mds, arg);1861		last_inode = inode;1862 1863		spin_lock(&session->s_cap_lock);1864		p = p->next;1865		if (!cap->ci) {1866			doutc(cl, "finishing cap %p removal\n", cap);1867			BUG_ON(cap->session != session);1868			cap->session = NULL;1869			list_del_init(&cap->session_caps);1870			session->s_nr_caps--;1871			atomic64_dec(&session->s_mdsc->metric.total_caps);1872			if (cap->queue_release)1873				__ceph_queue_cap_release(session, cap);1874			else1875				old_cap = cap;  /* put_cap it w/o locks held */1876		}1877		if (ret < 0)1878			goto out;1879	}1880	ret = 0;1881out:1882	session->s_cap_iterator = NULL;1883	spin_unlock(&session->s_cap_lock);1884 1885	iput(last_inode);1886	if (old_cap)1887		ceph_put_cap(session->s_mdsc, old_cap);1888 1889	return ret;1890}1891 1892static int remove_session_caps_cb(struct inode *inode, int mds, void *arg)1893{1894	struct ceph_inode_info *ci = ceph_inode(inode);1895	struct ceph_client *cl = ceph_inode_to_client(inode);1896	bool invalidate = false;1897	struct ceph_cap *cap;1898	int iputs = 0;1899 1900	spin_lock(&ci->i_ceph_lock);1901	cap = __get_cap_for_mds(ci, mds);1902	if (cap) {1903		doutc(cl, " removing cap %p, ci is %p, inode is %p\n",1904		      cap, ci, &ci->netfs.inode);1905 1906		iputs = ceph_purge_inode_cap(inode, cap, &invalidate);1907	}1908	spin_unlock(&ci->i_ceph_lock);1909 1910	if (cap)1911		wake_up_all(&ci->i_cap_wq);1912	if (invalidate)1913		ceph_queue_invalidate(inode);1914	while (iputs--)1915		iput(inode);1916	return 0;1917}1918 1919/*1920 * caller must hold session s_mutex1921 */1922static void remove_session_caps(struct ceph_mds_session *session)1923{1924	struct ceph_fs_client *fsc = session->s_mdsc->fsc;1925	struct super_block *sb = fsc->sb;1926	LIST_HEAD(dispose);1927 1928	doutc(fsc->client, "on %p\n", session);1929	ceph_iterate_session_caps(session, remove_session_caps_cb, fsc);1930 1931	wake_up_all(&fsc->mdsc->cap_flushing_wq);1932 1933	spin_lock(&session->s_cap_lock);1934	if (session->s_nr_caps > 0) {1935		struct inode *inode;1936		struct ceph_cap *cap, *prev = NULL;1937		struct ceph_vino vino;1938		/*1939		 * iterate_session_caps() skips inodes that are being1940		 * deleted, we need to wait until deletions are complete.1941		 * __wait_on_freeing_inode() is designed for the job,1942		 * but it is not exported, so use lookup inode function1943		 * to access it.1944		 */1945		while (!list_empty(&session->s_caps)) {1946			cap = list_entry(session->s_caps.next,1947					 struct ceph_cap, session_caps);1948			if (cap == prev)1949				break;1950			prev = cap;1951			vino = cap->ci->i_vino;1952			spin_unlock(&session->s_cap_lock);1953 1954			inode = ceph_find_inode(sb, vino);1955			iput(inode);1956 1957			spin_lock(&session->s_cap_lock);1958		}1959	}1960 1961	// drop cap expires and unlock s_cap_lock1962	detach_cap_releases(session, &dispose);1963 1964	BUG_ON(session->s_nr_caps > 0);1965	BUG_ON(!list_empty(&session->s_cap_flushing));1966	spin_unlock(&session->s_cap_lock);1967	dispose_cap_releases(session->s_mdsc, &dispose);1968}1969 1970enum {1971	RECONNECT,1972	RENEWCAPS,1973	FORCE_RO,1974};1975 1976/*1977 * wake up any threads waiting on this session's caps.  if the cap is1978 * old (didn't get renewed on the client reconnect), remove it now.1979 *1980 * caller must hold s_mutex.1981 */1982static int wake_up_session_cb(struct inode *inode, int mds, void *arg)1983{1984	struct ceph_inode_info *ci = ceph_inode(inode);1985	unsigned long ev = (unsigned long)arg;1986 1987	if (ev == RECONNECT) {1988		spin_lock(&ci->i_ceph_lock);1989		ci->i_wanted_max_size = 0;1990		ci->i_requested_max_size = 0;1991		spin_unlock(&ci->i_ceph_lock);1992	} else if (ev == RENEWCAPS) {1993		struct ceph_cap *cap;1994 1995		spin_lock(&ci->i_ceph_lock);1996		cap = __get_cap_for_mds(ci, mds);1997		/* mds did not re-issue stale cap */1998		if (cap && cap->cap_gen < atomic_read(&cap->session->s_cap_gen))1999			cap->issued = cap->implemented = CEPH_CAP_PIN;2000		spin_unlock(&ci->i_ceph_lock);2001	} else if (ev == FORCE_RO) {2002	}2003	wake_up_all(&ci->i_cap_wq);2004	return 0;2005}2006 2007static void wake_up_session_caps(struct ceph_mds_session *session, int ev)2008{2009	struct ceph_client *cl = session->s_mdsc->fsc->client;2010 2011	doutc(cl, "session %p mds%d\n", session, session->s_mds);2012	ceph_iterate_session_caps(session, wake_up_session_cb,2013				  (void *)(unsigned long)ev);2014}2015 2016/*2017 * Send periodic message to MDS renewing all currently held caps.  The2018 * ack will reset the expiration for all caps from this session.2019 *2020 * caller holds s_mutex2021 */2022static int send_renew_caps(struct ceph_mds_client *mdsc,2023			   struct ceph_mds_session *session)2024{2025	struct ceph_client *cl = mdsc->fsc->client;2026	struct ceph_msg *msg;2027	int state;2028 2029	if (time_after_eq(jiffies, session->s_cap_ttl) &&2030	    time_after_eq(session->s_cap_ttl, session->s_renew_requested))2031		pr_info_client(cl, "mds%d caps stale\n", session->s_mds);2032	session->s_renew_requested = jiffies;2033 2034	/* do not try to renew caps until a recovering mds has reconnected2035	 * with its clients. */2036	state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds);2037	if (state < CEPH_MDS_STATE_RECONNECT) {2038		doutc(cl, "ignoring mds%d (%s)\n", session->s_mds,2039		      ceph_mds_state_name(state));2040		return 0;2041	}2042 2043	doutc(cl, "to mds%d (%s)\n", session->s_mds,2044	      ceph_mds_state_name(state));2045	msg = create_session_full_msg(mdsc, CEPH_SESSION_REQUEST_RENEWCAPS,2046				      ++session->s_renew_seq);2047	if (IS_ERR(msg))2048		return PTR_ERR(msg);2049	ceph_con_send(&session->s_con, msg);2050	return 0;2051}2052 2053static int send_flushmsg_ack(struct ceph_mds_client *mdsc,2054			     struct ceph_mds_session *session, u64 seq)2055{2056	struct ceph_client *cl = mdsc->fsc->client;2057	struct ceph_msg *msg;2058 2059	doutc(cl, "to mds%d (%s)s seq %lld\n", session->s_mds,2060	      ceph_session_state_name(session->s_state), seq);2061	msg = ceph_create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq);2062	if (!msg)2063		return -ENOMEM;2064	ceph_con_send(&session->s_con, msg);2065	return 0;2066}2067 2068 2069/*2070 * Note new cap ttl, and any transition from stale -> not stale (fresh?).2071 *2072 * Called under session->s_mutex2073 */2074static void renewed_caps(struct ceph_mds_client *mdsc,2075			 struct ceph_mds_session *session, int is_renew)2076{2077	struct ceph_client *cl = mdsc->fsc->client;2078	int was_stale;2079	int wake = 0;2080 2081	spin_lock(&session->s_cap_lock);2082	was_stale = is_renew && time_after_eq(jiffies, session->s_cap_ttl);2083 2084	session->s_cap_ttl = session->s_renew_requested +2085		mdsc->mdsmap->m_session_timeout*HZ;2086 2087	if (was_stale) {2088		if (time_before(jiffies, session->s_cap_ttl)) {2089			pr_info_client(cl, "mds%d caps renewed\n",2090				       session->s_mds);2091			wake = 1;2092		} else {2093			pr_info_client(cl, "mds%d caps still stale\n",2094				       session->s_mds);2095		}2096	}2097	doutc(cl, "mds%d ttl now %lu, was %s, now %s\n", session->s_mds,2098	      session->s_cap_ttl, was_stale ? "stale" : "fresh",2099	      time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh");2100	spin_unlock(&session->s_cap_lock);2101 2102	if (wake)2103		wake_up_session_caps(session, RENEWCAPS);2104}2105 2106/*2107 * send a session close request2108 */2109static int request_close_session(struct ceph_mds_session *session)2110{2111	struct ceph_client *cl = session->s_mdsc->fsc->client;2112	struct ceph_msg *msg;2113 2114	doutc(cl, "mds%d state %s seq %lld\n", session->s_mds,2115	      ceph_session_state_name(session->s_state), session->s_seq);2116	msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_CLOSE,2117				      session->s_seq);2118	if (!msg)2119		return -ENOMEM;2120	ceph_con_send(&session->s_con, msg);2121	return 1;2122}2123 2124/*2125 * Called with s_mutex held.2126 */2127static int __close_session(struct ceph_mds_client *mdsc,2128			 struct ceph_mds_session *session)2129{2130	if (session->s_state >= CEPH_MDS_SESSION_CLOSING)2131		return 0;2132	session->s_state = CEPH_MDS_SESSION_CLOSING;2133	return request_close_session(session);2134}2135 2136static bool drop_negative_children(struct dentry *dentry)2137{2138	struct dentry *child;2139	bool all_negative = true;2140 2141	if (!d_is_dir(dentry))2142		goto out;2143 2144	spin_lock(&dentry->d_lock);2145	hlist_for_each_entry(child, &dentry->d_children, d_sib) {2146		if (d_really_is_positive(child)) {2147			all_negative = false;2148			break;2149		}2150	}2151	spin_unlock(&dentry->d_lock);2152 2153	if (all_negative)2154		shrink_dcache_parent(dentry);2155out:2156	return all_negative;2157}2158 2159/*2160 * Trim old(er) caps.2161 *2162 * Because we can't cache an inode without one or more caps, we do2163 * this indirectly: if a cap is unused, we prune its aliases, at which2164 * point the inode will hopefully get dropped to.2165 *2166 * Yes, this is a bit sloppy.  Our only real goal here is to respond to2167 * memory pressure from the MDS, though, so it needn't be perfect.2168 */2169static int trim_caps_cb(struct inode *inode, int mds, void *arg)2170{2171	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);2172	struct ceph_client *cl = mdsc->fsc->client;2173	int *remaining = arg;2174	struct ceph_inode_info *ci = ceph_inode(inode);2175	int used, wanted, oissued, mine;2176	struct ceph_cap *cap;2177 2178	if (*remaining <= 0)2179		return -1;2180 2181	spin_lock(&ci->i_ceph_lock);2182	cap = __get_cap_for_mds(ci, mds);2183	if (!cap) {2184		spin_unlock(&ci->i_ceph_lock);2185		return 0;2186	}2187	mine = cap->issued | cap->implemented;2188	used = __ceph_caps_used(ci);2189	wanted = __ceph_caps_file_wanted(ci);2190	oissued = __ceph_caps_issued_other(ci, cap);2191 2192	doutc(cl, "%p %llx.%llx cap %p mine %s oissued %s used %s wanted %s\n",2193	      inode, ceph_vinop(inode), cap, ceph_cap_string(mine),2194	      ceph_cap_string(oissued), ceph_cap_string(used),2195	      ceph_cap_string(wanted));2196	if (cap == ci->i_auth_cap) {2197		if (ci->i_dirty_caps || ci->i_flushing_caps ||2198		    !list_empty(&ci->i_cap_snaps))2199			goto out;2200		if ((used | wanted) & CEPH_CAP_ANY_WR)2201			goto out;2202		/* Note: it's possible that i_filelock_ref becomes non-zero2203		 * after dropping auth caps. It doesn't hurt because reply2204		 * of lock mds request will re-add auth caps. */2205		if (atomic_read(&ci->i_filelock_ref) > 0)2206			goto out;2207	}2208	/* The inode has cached pages, but it's no longer used.2209	 * we can safely drop it */2210	if (S_ISREG(inode->i_mode) &&2211	    wanted == 0 && used == CEPH_CAP_FILE_CACHE &&2212	    !(oissued & CEPH_CAP_FILE_CACHE)) {2213	  used = 0;2214	  oissued = 0;2215	}2216	if ((used | wanted) & ~oissued & mine)2217		goto out;   /* we need these caps */2218 2219	if (oissued) {2220		/* we aren't the only cap.. just remove us */2221		ceph_remove_cap(mdsc, cap, true);2222		(*remaining)--;2223	} else {2224		struct dentry *dentry;2225		/* try dropping referring dentries */2226		spin_unlock(&ci->i_ceph_lock);2227		dentry = d_find_any_alias(inode);2228		if (dentry && drop_negative_children(dentry)) {2229			int count;2230			dput(dentry);2231			d_prune_aliases(inode);2232			count = atomic_read(&inode->i_count);2233			if (count == 1)2234				(*remaining)--;2235			doutc(cl, "%p %llx.%llx cap %p pruned, count now %d\n",2236			      inode, ceph_vinop(inode), cap, count);2237		} else {2238			dput(dentry);2239		}2240		return 0;2241	}2242 2243out:2244	spin_unlock(&ci->i_ceph_lock);2245	return 0;2246}2247 2248/*2249 * Trim session cap count down to some max number.2250 */2251int ceph_trim_caps(struct ceph_mds_client *mdsc,2252		   struct ceph_mds_session *session,2253		   int max_caps)2254{2255	struct ceph_client *cl = mdsc->fsc->client;2256	int trim_caps = session->s_nr_caps - max_caps;2257 2258	doutc(cl, "mds%d start: %d / %d, trim %d\n", session->s_mds,2259	      session->s_nr_caps, max_caps, trim_caps);2260	if (trim_caps > 0) {2261		int remaining = trim_caps;2262 2263		ceph_iterate_session_caps(session, trim_caps_cb, &remaining);2264		doutc(cl, "mds%d done: %d / %d, trimmed %d\n",2265		      session->s_mds, session->s_nr_caps, max_caps,2266		      trim_caps - remaining);2267	}2268 2269	ceph_flush_session_cap_releases(mdsc, session);2270	return 0;2271}2272 2273static int check_caps_flush(struct ceph_mds_client *mdsc,2274			    u64 want_flush_tid)2275{2276	struct ceph_client *cl = mdsc->fsc->client;2277	int ret = 1;2278 2279	spin_lock(&mdsc->cap_dirty_lock);2280	if (!list_empty(&mdsc->cap_flush_list)) {2281		struct ceph_cap_flush *cf =2282			list_first_entry(&mdsc->cap_flush_list,2283					 struct ceph_cap_flush, g_list);2284		if (cf->tid <= want_flush_tid) {2285			doutc(cl, "still flushing tid %llu <= %llu\n",2286			      cf->tid, want_flush_tid);2287			ret = 0;2288		}2289	}2290	spin_unlock(&mdsc->cap_dirty_lock);2291	return ret;2292}2293 2294/*2295 * flush all dirty inode data to disk.2296 *2297 * returns true if we've flushed through want_flush_tid2298 */2299static void wait_caps_flush(struct ceph_mds_client *mdsc,2300			    u64 want_flush_tid)2301{2302	struct ceph_client *cl = mdsc->fsc->client;2303 2304	doutc(cl, "want %llu\n", want_flush_tid);2305 2306	wait_event(mdsc->cap_flushing_wq,2307		   check_caps_flush(mdsc, want_flush_tid));2308 2309	doutc(cl, "ok, flushed thru %llu\n", want_flush_tid);2310}2311 2312/*2313 * called under s_mutex2314 */2315static void ceph_send_cap_releases(struct ceph_mds_client *mdsc,2316				   struct ceph_mds_session *session)2317{2318	struct ceph_client *cl = mdsc->fsc->client;2319	struct ceph_msg *msg = NULL;2320	struct ceph_mds_cap_release *head;2321	struct ceph_mds_cap_item *item;2322	struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc;2323	struct ceph_cap *cap;2324	LIST_HEAD(tmp_list);2325	int num_cap_releases;2326	__le32	barrier, *cap_barrier;2327 2328	down_read(&osdc->lock);2329	barrier = cpu_to_le32(osdc->epoch_barrier);2330	up_read(&osdc->lock);2331 2332	spin_lock(&session->s_cap_lock);2333again:2334	list_splice_init(&session->s_cap_releases, &tmp_list);2335	num_cap_releases = session->s_num_cap_releases;2336	session->s_num_cap_releases = 0;2337	spin_unlock(&session->s_cap_lock);2338 2339	while (!list_empty(&tmp_list)) {2340		if (!msg) {2341			msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE,2342					PAGE_SIZE, GFP_NOFS, false);2343			if (!msg)2344				goto out_err;2345			head = msg->front.iov_base;2346			head->num = cpu_to_le32(0);2347			msg->front.iov_len = sizeof(*head);2348 2349			msg->hdr.version = cpu_to_le16(2);2350			msg->hdr.compat_version = cpu_to_le16(1);2351		}2352 2353		cap = list_first_entry(&tmp_list, struct ceph_cap,2354					session_caps);2355		list_del(&cap->session_caps);2356		num_cap_releases--;2357 2358		head = msg->front.iov_base;2359		put_unaligned_le32(get_unaligned_le32(&head->num) + 1,2360				   &head->num);2361		item = msg->front.iov_base + msg->front.iov_len;2362		item->ino = cpu_to_le64(cap->cap_ino);2363		item->cap_id = cpu_to_le64(cap->cap_id);2364		item->migrate_seq = cpu_to_le32(cap->mseq);2365		item->seq = cpu_to_le32(cap->issue_seq);2366		msg->front.iov_len += sizeof(*item);2367 2368		ceph_put_cap(mdsc, cap);2369 2370		if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) {2371			// Append cap_barrier field2372			cap_barrier = msg->front.iov_base + msg->front.iov_len;2373			*cap_barrier = barrier;2374			msg->front.iov_len += sizeof(*cap_barrier);2375 2376			msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);2377			doutc(cl, "mds%d %p\n", session->s_mds, msg);2378			ceph_con_send(&session->s_con, msg);2379			msg = NULL;2380		}2381	}2382 2383	BUG_ON(num_cap_releases != 0);2384 2385	spin_lock(&session->s_cap_lock);2386	if (!list_empty(&session->s_cap_releases))2387		goto again;2388	spin_unlock(&session->s_cap_lock);2389 2390	if (msg) {2391		// Append cap_barrier field2392		cap_barrier = msg->front.iov_base + msg->front.iov_len;2393		*cap_barrier = barrier;2394		msg->front.iov_len += sizeof(*cap_barrier);2395 2396		msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);2397		doutc(cl, "mds%d %p\n", session->s_mds, msg);2398		ceph_con_send(&session->s_con, msg);2399	}2400	return;2401out_err:2402	pr_err_client(cl, "mds%d, failed to allocate message\n",2403		      session->s_mds);2404	spin_lock(&session->s_cap_lock);2405	list_splice(&tmp_list, &session->s_cap_releases);2406	session->s_num_cap_releases += num_cap_releases;2407	spin_unlock(&session->s_cap_lock);2408}2409 2410static void ceph_cap_release_work(struct work_struct *work)2411{2412	struct ceph_mds_session *session =2413		container_of(work, struct ceph_mds_session, s_cap_release_work);2414 2415	mutex_lock(&session->s_mutex);2416	if (session->s_state == CEPH_MDS_SESSION_OPEN ||2417	    session->s_state == CEPH_MDS_SESSION_HUNG)2418		ceph_send_cap_releases(session->s_mdsc, session);2419	mutex_unlock(&session->s_mutex);2420	ceph_put_mds_session(session);2421}2422 2423void ceph_flush_session_cap_releases(struct ceph_mds_client *mdsc,2424		             struct ceph_mds_session *session)2425{2426	struct ceph_client *cl = mdsc->fsc->client;2427	if (mdsc->stopping)2428		return;2429 2430	ceph_get_mds_session(session);2431	if (queue_work(mdsc->fsc->cap_wq,2432		       &session->s_cap_release_work)) {2433		doutc(cl, "cap release work queued\n");2434	} else {2435		ceph_put_mds_session(session);2436		doutc(cl, "failed to queue cap release work\n");2437	}2438}2439 2440/*2441 * caller holds session->s_cap_lock2442 */2443void __ceph_queue_cap_release(struct ceph_mds_session *session,2444			      struct ceph_cap *cap)2445{2446	list_add_tail(&cap->session_caps, &session->s_cap_releases);2447	session->s_num_cap_releases++;2448 2449	if (!(session->s_num_cap_releases % CEPH_CAPS_PER_RELEASE))2450		ceph_flush_session_cap_releases(session->s_mdsc, session);2451}2452 2453static void ceph_cap_reclaim_work(struct work_struct *work)2454{2455	struct ceph_mds_client *mdsc =2456		container_of(work, struct ceph_mds_client, cap_reclaim_work);2457	int ret = ceph_trim_dentries(mdsc);2458	if (ret == -EAGAIN)2459		ceph_queue_cap_reclaim_work(mdsc);2460}2461 2462void ceph_queue_cap_reclaim_work(struct ceph_mds_client *mdsc)2463{2464	struct ceph_client *cl = mdsc->fsc->client;2465	if (mdsc->stopping)2466		return;2467 2468        if (queue_work(mdsc->fsc->cap_wq, &mdsc->cap_reclaim_work)) {2469                doutc(cl, "caps reclaim work queued\n");2470        } else {2471                doutc(cl, "failed to queue caps release work\n");2472        }2473}2474 2475void ceph_reclaim_caps_nr(struct ceph_mds_client *mdsc, int nr)2476{2477	int val;2478	if (!nr)2479		return;2480	val = atomic_add_return(nr, &mdsc->cap_reclaim_pending);2481	if ((val % CEPH_CAPS_PER_RELEASE) < nr) {2482		atomic_set(&mdsc->cap_reclaim_pending, 0);2483		ceph_queue_cap_reclaim_work(mdsc);2484	}2485}2486 2487void ceph_queue_cap_unlink_work(struct ceph_mds_client *mdsc)2488{2489	struct ceph_client *cl = mdsc->fsc->client;2490	if (mdsc->stopping)2491		return;2492 2493        if (queue_work(mdsc->fsc->cap_wq, &mdsc->cap_unlink_work)) {2494                doutc(cl, "caps unlink work queued\n");2495        } else {2496                doutc(cl, "failed to queue caps unlink work\n");2497        }2498}2499 2500static void ceph_cap_unlink_work(struct work_struct *work)2501{2502	struct ceph_mds_client *mdsc =2503		container_of(work, struct ceph_mds_client, cap_unlink_work);2504	struct ceph_client *cl = mdsc->fsc->client;2505 2506	doutc(cl, "begin\n");2507	spin_lock(&mdsc->cap_delay_lock);2508	while (!list_empty(&mdsc->cap_unlink_delay_list)) {2509		struct ceph_inode_info *ci;2510		struct inode *inode;2511 2512		ci = list_first_entry(&mdsc->cap_unlink_delay_list,2513				      struct ceph_inode_info,2514				      i_cap_delay_list);2515		list_del_init(&ci->i_cap_delay_list);2516 2517		inode = igrab(&ci->netfs.inode);2518		if (inode) {2519			spin_unlock(&mdsc->cap_delay_lock);2520			doutc(cl, "on %p %llx.%llx\n", inode,2521			      ceph_vinop(inode));2522			ceph_check_caps(ci, CHECK_CAPS_FLUSH);2523			iput(inode);2524			spin_lock(&mdsc->cap_delay_lock);2525		}2526	}2527	spin_unlock(&mdsc->cap_delay_lock);2528	doutc(cl, "done\n");2529}2530 2531/*2532 * requests2533 */2534 2535int ceph_alloc_readdir_reply_buffer(struct ceph_mds_request *req,2536				    struct inode *dir)2537{2538	struct ceph_inode_info *ci = ceph_inode(dir);2539	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;2540	struct ceph_mount_options *opt = req->r_mdsc->fsc->mount_options;2541	size_t size = sizeof(struct ceph_mds_reply_dir_entry);2542	unsigned int num_entries;2543	int order;2544 2545	spin_lock(&ci->i_ceph_lock);2546	num_entries = ci->i_files + ci->i_subdirs;2547	spin_unlock(&ci->i_ceph_lock);2548	num_entries = max(num_entries, 1U);2549	num_entries = min(num_entries, opt->max_readdir);2550 2551	order = get_order(size * num_entries);2552	while (order >= 0) {2553		rinfo->dir_entries = (void*)__get_free_pages(GFP_KERNEL |2554							     __GFP_NOWARN |2555							     __GFP_ZERO,2556							     order);2557		if (rinfo->dir_entries)2558			break;2559		order--;2560	}2561	if (!rinfo->dir_entries)2562		return -ENOMEM;2563 2564	num_entries = (PAGE_SIZE << order) / size;2565	num_entries = min(num_entries, opt->max_readdir);2566 2567	rinfo->dir_buf_size = PAGE_SIZE << order;2568	req->r_num_caps = num_entries + 1;2569	req->r_args.readdir.max_entries = cpu_to_le32(num_entries);2570	req->r_args.readdir.max_bytes = cpu_to_le32(opt->max_readdir_bytes);2571	return 0;2572}2573 2574/*2575 * Create an mds request.2576 */2577struct ceph_mds_request *2578ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)2579{2580	struct ceph_mds_request *req;2581 2582	req = kmem_cache_zalloc(ceph_mds_request_cachep, GFP_NOFS);2583	if (!req)2584		return ERR_PTR(-ENOMEM);2585 2586	mutex_init(&req->r_fill_mutex);2587	req->r_mdsc = mdsc;2588	req->r_started = jiffies;2589	req->r_start_latency = ktime_get();2590	req->r_resend_mds = -1;2591	INIT_LIST_HEAD(&req->r_unsafe_dir_item);2592	INIT_LIST_HEAD(&req->r_unsafe_target_item);2593	req->r_fmode = -1;2594	req->r_feature_needed = -1;2595	kref_init(&req->r_kref);2596	RB_CLEAR_NODE(&req->r_node);2597	INIT_LIST_HEAD(&req->r_wait);2598	init_completion(&req->r_completion);2599	init_completion(&req->r_safe_completion);2600	INIT_LIST_HEAD(&req->r_unsafe_item);2601 2602	ktime_get_coarse_real_ts64(&req->r_stamp);2603 2604	req->r_op = op;2605	req->r_direct_mode = mode;2606	return req;2607}2608 2609/*2610 * return oldest (lowest) request, tid in request tree, 0 if none.2611 *2612 * called under mdsc->mutex.2613 */2614static struct ceph_mds_request *__get_oldest_req(struct ceph_mds_client *mdsc)2615{2616	if (RB_EMPTY_ROOT(&mdsc->request_tree))2617		return NULL;2618	return rb_entry(rb_first(&mdsc->request_tree),2619			struct ceph_mds_request, r_node);2620}2621 2622static inline  u64 __get_oldest_tid(struct ceph_mds_client *mdsc)2623{2624	return mdsc->oldest_tid;2625}2626 2627#if IS_ENABLED(CONFIG_FS_ENCRYPTION)2628static u8 *get_fscrypt_altname(const struct ceph_mds_request *req, u32 *plen)2629{2630	struct inode *dir = req->r_parent;2631	struct dentry *dentry = req->r_dentry;2632	u8 *cryptbuf = NULL;2633	u32 len = 0;2634	int ret = 0;2635 2636	/* only encode if we have parent and dentry */2637	if (!dir || !dentry)2638		goto success;2639 2640	/* No-op unless this is encrypted */2641	if (!IS_ENCRYPTED(dir))2642		goto success;2643 2644	ret = ceph_fscrypt_prepare_readdir(dir);2645	if (ret < 0)2646		return ERR_PTR(ret);2647 2648	/* No key? Just ignore it. */2649	if (!fscrypt_has_encryption_key(dir))2650		goto success;2651 2652	if (!fscrypt_fname_encrypted_size(dir, dentry->d_name.len, NAME_MAX,2653					  &len)) {2654		WARN_ON_ONCE(1);2655		return ERR_PTR(-ENAMETOOLONG);2656	}2657 2658	/* No need to append altname if name is short enough */2659	if (len <= CEPH_NOHASH_NAME_MAX) {2660		len = 0;2661		goto success;2662	}2663 2664	cryptbuf = kmalloc(len, GFP_KERNEL);2665	if (!cryptbuf)2666		return ERR_PTR(-ENOMEM);2667 2668	ret = fscrypt_fname_encrypt(dir, &dentry->d_name, cryptbuf, len);2669	if (ret) {2670		kfree(cryptbuf);2671		return ERR_PTR(ret);2672	}2673success:2674	*plen = len;2675	return cryptbuf;2676}2677#else2678static u8 *get_fscrypt_altname(const struct ceph_mds_request *req, u32 *plen)2679{2680	*plen = 0;2681	return NULL;2682}2683#endif2684 2685/**2686 * ceph_mdsc_build_path - build a path string to a given dentry2687 * @mdsc: mds client2688 * @dentry: dentry to which path should be built2689 * @plen: returned length of string2690 * @pbase: returned base inode number2691 * @for_wire: is this path going to be sent to the MDS?2692 *2693 * Build a string that represents the path to the dentry. This is mostly called2694 * for two different purposes:2695 *2696 * 1) we need to build a path string to send to the MDS (for_wire == true)2697 * 2) we need a path string for local presentation (e.g. debugfs)2698 *    (for_wire == false)2699 *2700 * The path is built in reverse, starting with the dentry. Walk back up toward2701 * the root, building the path until the first non-snapped inode is reached2702 * (for_wire) or the root inode is reached (!for_wire).2703 *2704 * Encode hidden .snap dirs as a double /, i.e.2705 *   foo/.snap/bar -> foo//bar2706 */2707char *ceph_mdsc_build_path(struct ceph_mds_client *mdsc, struct dentry *dentry,2708			   int *plen, u64 *pbase, int for_wire)2709{2710	struct ceph_client *cl = mdsc->fsc->client;2711	struct dentry *cur;2712	struct inode *inode;2713	char *path;2714	int pos;2715	unsigned seq;2716	u64 base;2717 2718	if (!dentry)2719		return ERR_PTR(-EINVAL);2720 2721	path = __getname();2722	if (!path)2723		return ERR_PTR(-ENOMEM);2724retry:2725	pos = PATH_MAX - 1;2726	path[pos] = '\0';2727 2728	seq = read_seqbegin(&rename_lock);2729	cur = dget(dentry);2730	for (;;) {2731		struct dentry *parent;2732 2733		spin_lock(&cur->d_lock);2734		inode = d_inode(cur);2735		if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {2736			doutc(cl, "path+%d: %p SNAPDIR\n", pos, cur);2737			spin_unlock(&cur->d_lock);2738			parent = dget_parent(cur);2739		} else if (for_wire && inode && dentry != cur &&2740			   ceph_snap(inode) == CEPH_NOSNAP) {2741			spin_unlock(&cur->d_lock);2742			pos++; /* get rid of any prepended '/' */2743			break;2744		} else if (!for_wire || !IS_ENCRYPTED(d_inode(cur->d_parent))) {2745			pos -= cur->d_name.len;2746			if (pos < 0) {2747				spin_unlock(&cur->d_lock);2748				break;2749			}2750			memcpy(path + pos, cur->d_name.name, cur->d_name.len);2751			spin_unlock(&cur->d_lock);2752			parent = dget_parent(cur);2753		} else {2754			int len, ret;2755			char buf[NAME_MAX];2756 2757			/*2758			 * Proactively copy name into buf, in case we need to2759			 * present it as-is.2760			 */2761			memcpy(buf, cur->d_name.name, cur->d_name.len);2762			len = cur->d_name.len;2763			spin_unlock(&cur->d_lock);2764			parent = dget_parent(cur);2765 2766			ret = ceph_fscrypt_prepare_readdir(d_inode(parent));2767			if (ret < 0) {2768				dput(parent);2769				dput(cur);2770				return ERR_PTR(ret);2771			}2772 2773			if (fscrypt_has_encryption_key(d_inode(parent))) {2774				len = ceph_encode_encrypted_fname(d_inode(parent),2775								  cur, buf);2776				if (len < 0) {2777					dput(parent);2778					dput(cur);2779					return ERR_PTR(len);2780				}2781			}2782			pos -= len;2783			if (pos < 0) {2784				dput(parent);2785				break;2786			}2787			memcpy(path + pos, buf, len);2788		}2789		dput(cur);2790		cur = parent;2791 2792		/* Are we at the root? */2793		if (IS_ROOT(cur))2794			break;2795 2796		/* Are we out of buffer? */2797		if (--pos < 0)2798			break;2799 2800		path[pos] = '/';2801	}2802	inode = d_inode(cur);2803	base = inode ? ceph_ino(inode) : 0;2804	dput(cur);2805 2806	if (read_seqretry(&rename_lock, seq))2807		goto retry;2808 2809	if (pos < 0) {2810		/*2811		 * A rename didn't occur, but somehow we didn't end up where2812		 * we thought we would. Throw a warning and try again.2813		 */2814		pr_warn_client(cl, "did not end path lookup where expected (pos = %d)\n",2815			       pos);2816		goto retry;2817	}2818 2819	*pbase = base;2820	*plen = PATH_MAX - 1 - pos;2821	doutc(cl, "on %p %d built %llx '%.*s'\n", dentry, d_count(dentry),2822	      base, *plen, path + pos);2823	return path + pos;2824}2825 2826static int build_dentry_path(struct ceph_mds_client *mdsc, struct dentry *dentry,2827			     struct inode *dir, const char **ppath, int *ppathlen,2828			     u64 *pino, bool *pfreepath, bool parent_locked)2829{2830	char *path;2831 2832	rcu_read_lock();2833	if (!dir)2834		dir = d_inode_rcu(dentry->d_parent);2835	if (dir && parent_locked && ceph_snap(dir) == CEPH_NOSNAP &&2836	    !IS_ENCRYPTED(dir)) {2837		*pino = ceph_ino(dir);2838		rcu_read_unlock();2839		*ppath = dentry->d_name.name;2840		*ppathlen = dentry->d_name.len;2841		return 0;2842	}2843	rcu_read_unlock();2844	path = ceph_mdsc_build_path(mdsc, dentry, ppathlen, pino, 1);2845	if (IS_ERR(path))2846		return PTR_ERR(path);2847	*ppath = path;2848	*pfreepath = true;2849	return 0;2850}2851 2852static int build_inode_path(struct inode *inode,2853			    const char **ppath, int *ppathlen, u64 *pino,2854			    bool *pfreepath)2855{2856	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);2857	struct dentry *dentry;2858	char *path;2859 2860	if (ceph_snap(inode) == CEPH_NOSNAP) {2861		*pino = ceph_ino(inode);2862		*ppathlen = 0;2863		return 0;2864	}2865	dentry = d_find_alias(inode);2866	path = ceph_mdsc_build_path(mdsc, dentry, ppathlen, pino, 1);2867	dput(dentry);2868	if (IS_ERR(path))2869		return PTR_ERR(path);2870	*ppath = path;2871	*pfreepath = true;2872	return 0;2873}2874 2875/*2876 * request arguments may be specified via an inode *, a dentry *, or2877 * an explicit ino+path.2878 */2879static int set_request_path_attr(struct ceph_mds_client *mdsc, struct inode *rinode,2880				 struct dentry *rdentry, struct inode *rdiri,2881				 const char *rpath, u64 rino, const char **ppath,2882				 int *pathlen, u64 *ino, bool *freepath,2883				 bool parent_locked)2884{2885	struct ceph_client *cl = mdsc->fsc->client;2886	int r = 0;2887 2888	if (rinode) {2889		r = build_inode_path(rinode, ppath, pathlen, ino, freepath);2890		doutc(cl, " inode %p %llx.%llx\n", rinode, ceph_ino(rinode),2891		      ceph_snap(rinode));2892	} else if (rdentry) {2893		r = build_dentry_path(mdsc, rdentry, rdiri, ppath, pathlen, ino,2894					freepath, parent_locked);2895		doutc(cl, " dentry %p %llx/%.*s\n", rdentry, *ino, *pathlen, *ppath);2896	} else if (rpath || rino) {2897		*ino = rino;2898		*ppath = rpath;2899		*pathlen = rpath ? strlen(rpath) : 0;2900		doutc(cl, " path %.*s\n", *pathlen, rpath);2901	}2902 2903	return r;2904}2905 2906static void encode_mclientrequest_tail(void **p,2907				       const struct ceph_mds_request *req)2908{2909	struct ceph_timespec ts;2910	int i;2911 2912	ceph_encode_timespec64(&ts, &req->r_stamp);2913	ceph_encode_copy(p, &ts, sizeof(ts));2914 2915	/* v4: gid_list */2916	ceph_encode_32(p, req->r_cred->group_info->ngroups);2917	for (i = 0; i < req->r_cred->group_info->ngroups; i++)2918		ceph_encode_64(p, from_kgid(&init_user_ns,2919					    req->r_cred->group_info->gid[i]));2920 2921	/* v5: altname */2922	ceph_encode_32(p, req->r_altname_len);2923	ceph_encode_copy(p, req->r_altname, req->r_altname_len);2924 2925	/* v6: fscrypt_auth and fscrypt_file */2926	if (req->r_fscrypt_auth) {2927		u32 authlen = ceph_fscrypt_auth_len(req->r_fscrypt_auth);2928 2929		ceph_encode_32(p, authlen);2930		ceph_encode_copy(p, req->r_fscrypt_auth, authlen);2931	} else {2932		ceph_encode_32(p, 0);2933	}2934	if (test_bit(CEPH_MDS_R_FSCRYPT_FILE, &req->r_req_flags)) {2935		ceph_encode_32(p, sizeof(__le64));2936		ceph_encode_64(p, req->r_fscrypt_file);2937	} else {2938		ceph_encode_32(p, 0);2939	}2940}2941 2942static inline u16 mds_supported_head_version(struct ceph_mds_session *session)2943{2944	if (!test_bit(CEPHFS_FEATURE_32BITS_RETRY_FWD, &session->s_features))2945		return 1;2946 2947	if (!test_bit(CEPHFS_FEATURE_HAS_OWNER_UIDGID, &session->s_features))2948		return 2;2949 2950	return CEPH_MDS_REQUEST_HEAD_VERSION;2951}2952 2953static struct ceph_mds_request_head_legacy *2954find_legacy_request_head(void *p, u64 features)2955{2956	bool legacy = !(features & CEPH_FEATURE_FS_BTIME);2957	struct ceph_mds_request_head_old *ohead;2958 2959	if (legacy)2960		return (struct ceph_mds_request_head_legacy *)p;2961	ohead = (struct ceph_mds_request_head_old *)p;2962	return (struct ceph_mds_request_head_legacy *)&ohead->oldest_client_tid;2963}2964 2965/*2966 * called under mdsc->mutex2967 */2968static struct ceph_msg *create_request_message(struct ceph_mds_session *session,2969					       struct ceph_mds_request *req,2970					       bool drop_cap_releases)2971{2972	int mds = session->s_mds;2973	struct ceph_mds_client *mdsc = session->s_mdsc;2974	struct ceph_client *cl = mdsc->fsc->client;2975	struct ceph_msg *msg;2976	struct ceph_mds_request_head_legacy *lhead;2977	const char *path1 = NULL;2978	const char *path2 = NULL;2979	u64 ino1 = 0, ino2 = 0;2980	int pathlen1 = 0, pathlen2 = 0;2981	bool freepath1 = false, freepath2 = false;2982	struct dentry *old_dentry = NULL;2983	int len;2984	u16 releases;2985	void *p, *end;2986	int ret;2987	bool legacy = !(session->s_con.peer_features & CEPH_FEATURE_FS_BTIME);2988	u16 request_head_version = mds_supported_head_version(session);2989	kuid_t caller_fsuid = req->r_cred->fsuid;2990	kgid_t caller_fsgid = req->r_cred->fsgid;2991 2992	ret = set_request_path_attr(mdsc, req->r_inode, req->r_dentry,2993			      req->r_parent, req->r_path1, req->r_ino1.ino,2994			      &path1, &pathlen1, &ino1, &freepath1,2995			      test_bit(CEPH_MDS_R_PARENT_LOCKED,2996					&req->r_req_flags));2997	if (ret < 0) {2998		msg = ERR_PTR(ret);2999		goto out;3000	}3001 3002	/* If r_old_dentry is set, then assume that its parent is locked */3003	if (req->r_old_dentry &&3004	    !(req->r_old_dentry->d_flags & DCACHE_DISCONNECTED))3005		old_dentry = req->r_old_dentry;3006	ret = set_request_path_attr(mdsc, NULL, old_dentry,3007			      req->r_old_dentry_dir,3008			      req->r_path2, req->r_ino2.ino,3009			      &path2, &pathlen2, &ino2, &freepath2, true);3010	if (ret < 0) {3011		msg = ERR_PTR(ret);3012		goto out_free1;3013	}3014 3015	req->r_altname = get_fscrypt_altname(req, &req->r_altname_len);3016	if (IS_ERR(req->r_altname)) {3017		msg = ERR_CAST(req->r_altname);3018		req->r_altname = NULL;3019		goto out_free2;3020	}3021 3022	/*3023	 * For old cephs without supporting the 32bit retry/fwd feature3024	 * it will copy the raw memories directly when decoding the3025	 * requests. While new cephs will decode the head depending the3026	 * version member, so we need to make sure it will be compatible3027	 * with them both.3028	 */3029	if (legacy)3030		len = sizeof(struct ceph_mds_request_head_legacy);3031	else if (request_head_version == 1)3032		len = sizeof(struct ceph_mds_request_head_old);3033	else if (request_head_version == 2)3034		len = offsetofend(struct ceph_mds_request_head, ext_num_fwd);3035	else3036		len = sizeof(struct ceph_mds_request_head);3037 3038	/* filepaths */3039	len += 2 * (1 + sizeof(u32) + sizeof(u64));3040	len += pathlen1 + pathlen2;3041 3042	/* cap releases */3043	len += sizeof(struct ceph_mds_request_release) *3044		(!!req->r_inode_drop + !!req->r_dentry_drop +3045		 !!req->r_old_inode_drop + !!req->r_old_dentry_drop);3046 3047	if (req->r_dentry_drop)3048		len += pathlen1;3049	if (req->r_old_dentry_drop)3050		len += pathlen2;3051 3052	/* MClientRequest tail */3053 3054	/* req->r_stamp */3055	len += sizeof(struct ceph_timespec);3056 3057	/* gid list */3058	len += sizeof(u32) + (sizeof(u64) * req->r_cred->group_info->ngroups);3059 3060	/* alternate name */3061	len += sizeof(u32) + req->r_altname_len;3062 3063	/* fscrypt_auth */3064	len += sizeof(u32); // fscrypt_auth3065	if (req->r_fscrypt_auth)3066		len += ceph_fscrypt_auth_len(req->r_fscrypt_auth);3067 3068	/* fscrypt_file */3069	len += sizeof(u32);3070	if (test_bit(CEPH_MDS_R_FSCRYPT_FILE, &req->r_req_flags))3071		len += sizeof(__le64);3072 3073	msg = ceph_msg_new2(CEPH_MSG_CLIENT_REQUEST, len, 1, GFP_NOFS, false);3074	if (!msg) {3075		msg = ERR_PTR(-ENOMEM);3076		goto out_free2;3077	}3078 3079	msg->hdr.tid = cpu_to_le64(req->r_tid);3080 3081	lhead = find_legacy_request_head(msg->front.iov_base,3082					 session->s_con.peer_features);3083 3084	if ((req->r_mnt_idmap != &nop_mnt_idmap) &&3085	    !test_bit(CEPHFS_FEATURE_HAS_OWNER_UIDGID, &session->s_features)) {3086		WARN_ON_ONCE(!IS_CEPH_MDS_OP_NEWINODE(req->r_op));3087 3088		if (enable_unsafe_idmap) {3089			pr_warn_once_client(cl,3090				"idmapped mount is used and CEPHFS_FEATURE_HAS_OWNER_UIDGID"3091				" is not supported by MDS. UID/GID-based restrictions may"3092				" not work properly.\n");3093 3094			caller_fsuid = from_vfsuid(req->r_mnt_idmap, &init_user_ns,3095						   VFSUIDT_INIT(req->r_cred->fsuid));3096			caller_fsgid = from_vfsgid(req->r_mnt_idmap, &init_user_ns,3097						   VFSGIDT_INIT(req->r_cred->fsgid));3098		} else {3099			pr_err_ratelimited_client(cl,3100				"idmapped mount is used and CEPHFS_FEATURE_HAS_OWNER_UIDGID"3101				" is not supported by MDS. Fail request with -EIO.\n");3102 3103			ret = -EIO;3104			goto out_err;3105		}3106	}3107 3108	/*3109	 * The ceph_mds_request_head_legacy didn't contain a version field, and3110	 * one was added when we moved the message version from 3->4.3111	 */3112	if (legacy) {3113		msg->hdr.version = cpu_to_le16(3);3114		p = msg->front.iov_base + sizeof(*lhead);3115	} else if (request_head_version == 1) {3116		struct ceph_mds_request_head_old *ohead = msg->front.iov_base;3117 3118		msg->hdr.version = cpu_to_le16(4);3119		ohead->version = cpu_to_le16(1);3120		p = msg->front.iov_base + sizeof(*ohead);3121	} else if (request_head_version == 2) {3122		struct ceph_mds_request_head *nhead = msg->front.iov_base;3123 3124		msg->hdr.version = cpu_to_le16(6);3125		nhead->version = cpu_to_le16(2);3126 3127		p = msg->front.iov_base + offsetofend(struct ceph_mds_request_head, ext_num_fwd);3128	} else {3129		struct ceph_mds_request_head *nhead = msg->front.iov_base;3130		kuid_t owner_fsuid;3131		kgid_t owner_fsgid;3132 3133		msg->hdr.version = cpu_to_le16(6);3134		nhead->version = cpu_to_le16(CEPH_MDS_REQUEST_HEAD_VERSION);3135		nhead->struct_len = cpu_to_le32(sizeof(struct ceph_mds_request_head));3136 3137		if (IS_CEPH_MDS_OP_NEWINODE(req->r_op)) {3138			owner_fsuid = from_vfsuid(req->r_mnt_idmap, &init_user_ns,3139						VFSUIDT_INIT(req->r_cred->fsuid));3140			owner_fsgid = from_vfsgid(req->r_mnt_idmap, &init_user_ns,3141						VFSGIDT_INIT(req->r_cred->fsgid));3142			nhead->owner_uid = cpu_to_le32(from_kuid(&init_user_ns, owner_fsuid));3143			nhead->owner_gid = cpu_to_le32(from_kgid(&init_user_ns, owner_fsgid));3144		} else {3145			nhead->owner_uid = cpu_to_le32(-1);3146			nhead->owner_gid = cpu_to_le32(-1);3147		}3148 3149		p = msg->front.iov_base + sizeof(*nhead);3150	}3151 3152	end = msg->front.iov_base + msg->front.iov_len;3153 3154	lhead->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);3155	lhead->op = cpu_to_le32(req->r_op);3156	lhead->caller_uid = cpu_to_le32(from_kuid(&init_user_ns,3157						  caller_fsuid));3158	lhead->caller_gid = cpu_to_le32(from_kgid(&init_user_ns,3159						  caller_fsgid));3160	lhead->ino = cpu_to_le64(req->r_deleg_ino);3161	lhead->args = req->r_args;3162 3163	ceph_encode_filepath(&p, end, ino1, path1);3164	ceph_encode_filepath(&p, end, ino2, path2);3165 3166	/* make note of release offset, in case we need to replay */3167	req->r_request_release_offset = p - msg->front.iov_base;3168 3169	/* cap releases */3170	releases = 0;3171	if (req->r_inode_drop)3172		releases += ceph_encode_inode_release(&p,3173		      req->r_inode ? req->r_inode : d_inode(req->r_dentry),3174		      mds, req->r_inode_drop, req->r_inode_unless,3175		      req->r_op == CEPH_MDS_OP_READDIR);3176	if (req->r_dentry_drop) {3177		ret = ceph_encode_dentry_release(&p, req->r_dentry,3178				req->r_parent, mds, req->r_dentry_drop,3179				req->r_dentry_unless);3180		if (ret < 0)3181			goto out_err;3182		releases += ret;3183	}3184	if (req->r_old_dentry_drop) {3185		ret = ceph_encode_dentry_release(&p, req->r_old_dentry,3186				req->r_old_dentry_dir, mds,3187				req->r_old_dentry_drop,3188				req->r_old_dentry_unless);3189		if (ret < 0)3190			goto out_err;3191		releases += ret;3192	}3193	if (req->r_old_inode_drop)3194		releases += ceph_encode_inode_release(&p,3195		      d_inode(req->r_old_dentry),3196		      mds, req->r_old_inode_drop, req->r_old_inode_unless, 0);3197 3198	if (drop_cap_releases) {3199		releases = 0;3200		p = msg->front.iov_base + req->r_request_release_offset;3201	}3202 3203	lhead->num_releases = cpu_to_le16(releases);3204 3205	encode_mclientrequest_tail(&p, req);3206 3207	if (WARN_ON_ONCE(p > end)) {3208		ceph_msg_put(msg);3209		msg = ERR_PTR(-ERANGE);3210		goto out_free2;3211	}3212 3213	msg->front.iov_len = p - msg->front.iov_base;3214	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);3215 3216	if (req->r_pagelist) {3217		struct ceph_pagelist *pagelist = req->r_pagelist;3218		ceph_msg_data_add_pagelist(msg, pagelist);3219		msg->hdr.data_len = cpu_to_le32(pagelist->length);3220	} else {3221		msg->hdr.data_len = 0;3222	}3223 3224	msg->hdr.data_off = cpu_to_le16(0);3225 3226out_free2:3227	if (freepath2)3228		ceph_mdsc_free_path((char *)path2, pathlen2);3229out_free1:3230	if (freepath1)3231		ceph_mdsc_free_path((char *)path1, pathlen1);3232out:3233	return msg;3234out_err:3235	ceph_msg_put(msg);3236	msg = ERR_PTR(ret);3237	goto out_free2;3238}3239 3240/*3241 * called under mdsc->mutex if error, under no mutex if3242 * success.3243 */3244static void complete_request(struct ceph_mds_client *mdsc,3245			     struct ceph_mds_request *req)3246{3247	req->r_end_latency = ktime_get();3248 3249	if (req->r_callback)3250		req->r_callback(mdsc, req);3251	complete_all(&req->r_completion);3252}3253 3254/*3255 * called under mdsc->mutex3256 */3257static int __prepare_send_request(struct ceph_mds_session *session,3258				  struct ceph_mds_request *req,3259				  bool drop_cap_releases)3260{3261	int mds = session->s_mds;3262	struct ceph_mds_client *mdsc = session->s_mdsc;3263	struct ceph_client *cl = mdsc->fsc->client;3264	struct ceph_mds_request_head_legacy *lhead;3265	struct ceph_mds_request_head *nhead;3266	struct ceph_msg *msg;3267	int flags = 0, old_max_retry;3268	bool old_version = !test_bit(CEPHFS_FEATURE_32BITS_RETRY_FWD,3269				     &session->s_features);3270 3271	/*3272	 * Avoid inifinite retrying after overflow. The client will3273	 * increase the retry count and if the MDS is old version,3274	 * so we limit to retry at most 256 times.3275	 */3276	if (req->r_attempts) {3277	       old_max_retry = sizeof_field(struct ceph_mds_request_head_old,3278					    num_retry);3279	       old_max_retry = 1 << (old_max_retry * BITS_PER_BYTE);3280	       if ((old_version && req->r_attempts >= old_max_retry) ||3281		   ((uint32_t)req->r_attempts >= U32_MAX)) {3282			pr_warn_ratelimited_client(cl, "request tid %llu seq overflow\n",3283						   req->r_tid);3284			return -EMULTIHOP;3285	       }3286	}3287 3288	req->r_attempts++;3289	if (req->r_inode) {3290		struct ceph_cap *cap =3291			ceph_get_cap_for_mds(ceph_inode(req->r_inode), mds);3292 3293		if (cap)3294			req->r_sent_on_mseq = cap->mseq;3295		else3296			req->r_sent_on_mseq = -1;3297	}3298	doutc(cl, "%p tid %lld %s (attempt %d)\n", req, req->r_tid,3299	      ceph_mds_op_name(req->r_op), req->r_attempts);3300 3301	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {3302		void *p;3303 3304		/*3305		 * Replay.  Do not regenerate message (and rebuild3306		 * paths, etc.); just use the original message.3307		 * Rebuilding paths will break for renames because3308		 * d_move mangles the src name.3309		 */3310		msg = req->r_request;3311		lhead = find_legacy_request_head(msg->front.iov_base,3312						 session->s_con.peer_features);3313 3314		flags = le32_to_cpu(lhead->flags);3315		flags |= CEPH_MDS_FLAG_REPLAY;3316		lhead->flags = cpu_to_le32(flags);3317 3318		if (req->r_target_inode)3319			lhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode));3320 3321		lhead->num_retry = req->r_attempts - 1;3322		if (!old_version) {3323			nhead = (struct ceph_mds_request_head*)msg->front.iov_base;3324			nhead->ext_num_retry = cpu_to_le32(req->r_attempts - 1);3325		}3326 3327		/* remove cap/dentry releases from message */3328		lhead->num_releases = 0;3329 3330		p = msg->front.iov_base + req->r_request_release_offset;3331		encode_mclientrequest_tail(&p, req);3332 3333		msg->front.iov_len = p - msg->front.iov_base;3334		msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);3335		return 0;3336	}3337 3338	if (req->r_request) {3339		ceph_msg_put(req->r_request);3340		req->r_request = NULL;3341	}3342	msg = create_request_message(session, req, drop_cap_releases);3343	if (IS_ERR(msg)) {3344		req->r_err = PTR_ERR(msg);3345		return PTR_ERR(msg);3346	}3347	req->r_request = msg;3348 3349	lhead = find_legacy_request_head(msg->front.iov_base,3350					 session->s_con.peer_features);3351	lhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));3352	if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))3353		flags |= CEPH_MDS_FLAG_REPLAY;3354	if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags))3355		flags |= CEPH_MDS_FLAG_ASYNC;3356	if (req->r_parent)3357		flags |= CEPH_MDS_FLAG_WANT_DENTRY;3358	lhead->flags = cpu_to_le32(flags);3359	lhead->num_fwd = req->r_num_fwd;3360	lhead->num_retry = req->r_attempts - 1;3361	if (!old_version) {3362		nhead = (struct ceph_mds_request_head*)msg->front.iov_base;3363		nhead->ext_num_fwd = cpu_to_le32(req->r_num_fwd);3364		nhead->ext_num_retry = cpu_to_le32(req->r_attempts - 1);3365	}3366 3367	doutc(cl, " r_parent = %p\n", req->r_parent);3368	return 0;3369}3370 3371/*3372 * called under mdsc->mutex3373 */3374static int __send_request(struct ceph_mds_session *session,3375			  struct ceph_mds_request *req,3376			  bool drop_cap_releases)3377{3378	int err;3379 3380	err = __prepare_send_request(session, req, drop_cap_releases);3381	if (!err) {3382		ceph_msg_get(req->r_request);3383		ceph_con_send(&session->s_con, req->r_request);3384	}3385 3386	return err;3387}3388 3389/*3390 * send request, or put it on the appropriate wait list.3391 */3392static void __do_request(struct ceph_mds_client *mdsc,3393			struct ceph_mds_request *req)3394{3395	struct ceph_client *cl = mdsc->fsc->client;3396	struct ceph_mds_session *session = NULL;3397	int mds = -1;3398	int err = 0;3399	bool random;3400 3401	if (req->r_err || test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {3402		if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))3403			__unregister_request(mdsc, req);3404		return;3405	}3406 3407	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_FENCE_IO) {3408		doutc(cl, "metadata corrupted\n");3409		err = -EIO;3410		goto finish;3411	}3412	if (req->r_timeout &&3413	    time_after_eq(jiffies, req->r_started + req->r_timeout)) {3414		doutc(cl, "timed out\n");3415		err = -ETIMEDOUT;3416		goto finish;3417	}3418	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {3419		doutc(cl, "forced umount\n");3420		err = -EIO;3421		goto finish;3422	}3423	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_MOUNTING) {3424		if (mdsc->mdsmap_err) {3425			err = mdsc->mdsmap_err;3426			doutc(cl, "mdsmap err %d\n", err);3427			goto finish;3428		}3429		if (mdsc->mdsmap->m_epoch == 0) {3430			doutc(cl, "no mdsmap, waiting for map\n");3431			list_add(&req->r_wait, &mdsc->waiting_for_map);3432			return;3433		}3434		if (!(mdsc->fsc->mount_options->flags &3435		      CEPH_MOUNT_OPT_MOUNTWAIT) &&3436		    !ceph_mdsmap_is_cluster_available(mdsc->mdsmap)) {3437			err = -EHOSTUNREACH;3438			goto finish;3439		}3440	}3441 3442	put_request_session(req);3443 3444	mds = __choose_mds(mdsc, req, &random);3445	if (mds < 0 ||3446	    ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {3447		if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {3448			err = -EJUKEBOX;3449			goto finish;3450		}3451		doutc(cl, "no mds or not active, waiting for map\n");3452		list_add(&req->r_wait, &mdsc->waiting_for_map);3453		return;3454	}3455 3456	/* get, open session */3457	session = __ceph_lookup_mds_session(mdsc, mds);3458	if (!session) {3459		session = register_session(mdsc, mds);3460		if (IS_ERR(session)) {3461			err = PTR_ERR(session);3462			goto finish;3463		}3464	}3465	req->r_session = ceph_get_mds_session(session);3466 3467	doutc(cl, "mds%d session %p state %s\n", mds, session,3468	      ceph_session_state_name(session->s_state));3469 3470	/*3471	 * The old ceph will crash the MDSs when see unknown OPs3472	 */3473	if (req->r_feature_needed > 0 &&3474	    !test_bit(req->r_feature_needed, &session->s_features)) {3475		err = -EOPNOTSUPP;3476		goto out_session;3477	}3478 3479	if (session->s_state != CEPH_MDS_SESSION_OPEN &&3480	    session->s_state != CEPH_MDS_SESSION_HUNG) {3481		/*3482		 * We cannot queue async requests since the caps and delegated3483		 * inodes are bound to the session. Just return -EJUKEBOX and3484		 * let the caller retry a sync request in that case.3485		 */3486		if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {3487			err = -EJUKEBOX;3488			goto out_session;3489		}3490 3491		/*3492		 * If the session has been REJECTED, then return a hard error,3493		 * unless it's a CLEANRECOVER mount, in which case we'll queue3494		 * it to the mdsc queue.3495		 */3496		if (session->s_state == CEPH_MDS_SESSION_REJECTED) {3497			if (ceph_test_mount_opt(mdsc->fsc, CLEANRECOVER))3498				list_add(&req->r_wait, &mdsc->waiting_for_map);3499			else3500				err = -EACCES;3501			goto out_session;3502		}3503 3504		if (session->s_state == CEPH_MDS_SESSION_NEW ||3505		    session->s_state == CEPH_MDS_SESSION_CLOSING) {3506			err = __open_session(mdsc, session);3507			if (err)3508				goto out_session;3509			/* retry the same mds later */3510			if (random)3511				req->r_resend_mds = mds;3512		}3513		list_add(&req->r_wait, &session->s_waiting);3514		goto out_session;3515	}3516 3517	/* send request */3518	req->r_resend_mds = -1;   /* forget any previous mds hint */3519 3520	if (req->r_request_started == 0)   /* note request start time */3521		req->r_request_started = jiffies;3522 3523	/*3524	 * For async create we will choose the auth MDS of frag in parent3525	 * directory to send the request and ususally this works fine, but3526	 * if the migrated the dirtory to another MDS before it could handle3527	 * it the request will be forwarded.3528	 *3529	 * And then the auth cap will be changed.3530	 */3531	if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags) && req->r_num_fwd) {3532		struct ceph_dentry_info *di = ceph_dentry(req->r_dentry);3533		struct ceph_inode_info *ci;3534		struct ceph_cap *cap;3535 3536		/*3537		 * The request maybe handled very fast and the new inode3538		 * hasn't been linked to the dentry yet. We need to wait3539		 * for the ceph_finish_async_create(), which shouldn't be3540		 * stuck too long or fail in thoery, to finish when forwarding3541		 * the request.3542		 */3543		if (!d_inode(req->r_dentry)) {3544			err = wait_on_bit(&di->flags, CEPH_DENTRY_ASYNC_CREATE_BIT,3545					  TASK_KILLABLE);3546			if (err) {3547				mutex_lock(&req->r_fill_mutex);3548				set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);3549				mutex_unlock(&req->r_fill_mutex);3550				goto out_session;3551			}3552		}3553 3554		ci = ceph_inode(d_inode(req->r_dentry));3555 3556		spin_lock(&ci->i_ceph_lock);3557		cap = ci->i_auth_cap;3558		if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE && mds != cap->mds) {3559			doutc(cl, "session changed for auth cap %d -> %d\n",3560			      cap->session->s_mds, session->s_mds);3561 3562			/* Remove the auth cap from old session */3563			spin_lock(&cap->session->s_cap_lock);3564			cap->session->s_nr_caps--;3565			list_del_init(&cap->session_caps);3566			spin_unlock(&cap->session->s_cap_lock);3567 3568			/* Add the auth cap to the new session */3569			cap->mds = mds;3570			cap->session = session;3571			spin_lock(&session->s_cap_lock);3572			session->s_nr_caps++;3573			list_add_tail(&cap->session_caps, &session->s_caps);3574			spin_unlock(&session->s_cap_lock);3575 3576			change_auth_cap_ses(ci, session);3577		}3578		spin_unlock(&ci->i_ceph_lock);3579	}3580 3581	err = __send_request(session, req, false);3582 3583out_session:3584	ceph_put_mds_session(session);3585finish:3586	if (err) {3587		doutc(cl, "early error %d\n", err);3588		req->r_err = err;3589		complete_request(mdsc, req);3590		__unregister_request(mdsc, req);3591	}3592	return;3593}3594 3595/*3596 * called under mdsc->mutex3597 */3598static void __wake_requests(struct ceph_mds_client *mdsc,3599			    struct list_head *head)3600{3601	struct ceph_client *cl = mdsc->fsc->client;3602	struct ceph_mds_request *req;3603	LIST_HEAD(tmp_list);3604 3605	list_splice_init(head, &tmp_list);3606 3607	while (!list_empty(&tmp_list)) {3608		req = list_entry(tmp_list.next,3609				 struct ceph_mds_request, r_wait);3610		list_del_init(&req->r_wait);3611		doutc(cl, " wake request %p tid %llu\n", req,3612		      req->r_tid);3613		__do_request(mdsc, req);3614	}3615}3616 3617/*3618 * Wake up threads with requests pending for @mds, so that they can3619 * resubmit their requests to a possibly different mds.3620 */3621static void kick_requests(struct ceph_mds_client *mdsc, int mds)3622{3623	struct ceph_client *cl = mdsc->fsc->client;3624	struct ceph_mds_request *req;3625	struct rb_node *p = rb_first(&mdsc->request_tree);3626 3627	doutc(cl, "kick_requests mds%d\n", mds);3628	while (p) {3629		req = rb_entry(p, struct ceph_mds_request, r_node);3630		p = rb_next(p);3631		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))3632			continue;3633		if (req->r_attempts > 0)3634			continue; /* only new requests */3635		if (req->r_session &&3636		    req->r_session->s_mds == mds) {3637			doutc(cl, " kicking tid %llu\n", req->r_tid);3638			list_del_init(&req->r_wait);3639			__do_request(mdsc, req);3640		}3641	}3642}3643 3644int ceph_mdsc_submit_request(struct ceph_mds_client *mdsc, struct inode *dir,3645			      struct ceph_mds_request *req)3646{3647	struct ceph_client *cl = mdsc->fsc->client;3648	int err = 0;3649 3650	/* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */3651	if (req->r_inode)3652		ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);3653	if (req->r_parent) {3654		struct ceph_inode_info *ci = ceph_inode(req->r_parent);3655		int fmode = (req->r_op & CEPH_MDS_OP_WRITE) ?3656			    CEPH_FILE_MODE_WR : CEPH_FILE_MODE_RD;3657		spin_lock(&ci->i_ceph_lock);3658		ceph_take_cap_refs(ci, CEPH_CAP_PIN, false);3659		__ceph_touch_fmode(ci, mdsc, fmode);3660		spin_unlock(&ci->i_ceph_lock);3661	}3662	if (req->r_old_dentry_dir)3663		ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),3664				  CEPH_CAP_PIN);3665 3666	if (req->r_inode) {3667		err = ceph_wait_on_async_create(req->r_inode);3668		if (err) {3669			doutc(cl, "wait for async create returned: %d\n", err);3670			return err;3671		}3672	}3673 3674	if (!err && req->r_old_inode) {3675		err = ceph_wait_on_async_create(req->r_old_inode);3676		if (err) {3677			doutc(cl, "wait for async create returned: %d\n", err);3678			return err;3679		}3680	}3681 3682	doutc(cl, "submit_request on %p for inode %p\n", req, dir);3683	mutex_lock(&mdsc->mutex);3684	__register_request(mdsc, req, dir);3685	__do_request(mdsc, req);3686	err = req->r_err;3687	mutex_unlock(&mdsc->mutex);3688	return err;3689}3690 3691int ceph_mdsc_wait_request(struct ceph_mds_client *mdsc,3692			   struct ceph_mds_request *req,3693			   ceph_mds_request_wait_callback_t wait_func)3694{3695	struct ceph_client *cl = mdsc->fsc->client;3696	int err;3697 3698	/* wait */3699	doutc(cl, "do_request waiting\n");3700	if (wait_func) {3701		err = wait_func(mdsc, req);3702	} else {3703		long timeleft = wait_for_completion_killable_timeout(3704					&req->r_completion,3705					ceph_timeout_jiffies(req->r_timeout));3706		if (timeleft > 0)3707			err = 0;3708		else if (!timeleft)3709			err = -ETIMEDOUT;  /* timed out */3710		else3711			err = timeleft;  /* killed */3712	}3713	doutc(cl, "do_request waited, got %d\n", err);3714	mutex_lock(&mdsc->mutex);3715 3716	/* only abort if we didn't race with a real reply */3717	if (test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {3718		err = le32_to_cpu(req->r_reply_info.head->result);3719	} else if (err < 0) {3720		doutc(cl, "aborted request %lld with %d\n", req->r_tid, err);3721 3722		/*3723		 * ensure we aren't running concurrently with3724		 * ceph_fill_trace or ceph_readdir_prepopulate, which3725		 * rely on locks (dir mutex) held by our caller.3726		 */3727		mutex_lock(&req->r_fill_mutex);3728		req->r_err = err;3729		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);3730		mutex_unlock(&req->r_fill_mutex);3731 3732		if (req->r_parent &&3733		    (req->r_op & CEPH_MDS_OP_WRITE))3734			ceph_invalidate_dir_request(req);3735	} else {3736		err = req->r_err;3737	}3738 3739	mutex_unlock(&mdsc->mutex);3740	return err;3741}3742 3743/*3744 * Synchrously perform an mds request.  Take care of all of the3745 * session setup, forwarding, retry details.3746 */3747int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,3748			 struct inode *dir,3749			 struct ceph_mds_request *req)3750{3751	struct ceph_client *cl = mdsc->fsc->client;3752	int err;3753 3754	doutc(cl, "do_request on %p\n", req);3755 3756	/* issue */3757	err = ceph_mdsc_submit_request(mdsc, dir, req);3758	if (!err)3759		err = ceph_mdsc_wait_request(mdsc, req, NULL);3760	doutc(cl, "do_request %p done, result %d\n", req, err);3761	return err;3762}3763 3764/*3765 * Invalidate dir's completeness, dentry lease state on an aborted MDS3766 * namespace request.3767 */3768void ceph_invalidate_dir_request(struct ceph_mds_request *req)3769{3770	struct inode *dir = req->r_parent;3771	struct inode *old_dir = req->r_old_dentry_dir;3772	struct ceph_client *cl = req->r_mdsc->fsc->client;3773 3774	doutc(cl, "invalidate_dir_request %p %p (complete, lease(s))\n",3775	      dir, old_dir);3776 3777	ceph_dir_clear_complete(dir);3778	if (old_dir)3779		ceph_dir_clear_complete(old_dir);3780	if (req->r_dentry)3781		ceph_invalidate_dentry_lease(req->r_dentry);3782	if (req->r_old_dentry)3783		ceph_invalidate_dentry_lease(req->r_old_dentry);3784}3785 3786/*3787 * Handle mds reply.3788 *3789 * We take the session mutex and parse and process the reply immediately.3790 * This preserves the logical ordering of replies, capabilities, etc., sent3791 * by the MDS as they are applied to our local cache.3792 */3793static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)3794{3795	struct ceph_mds_client *mdsc = session->s_mdsc;3796	struct ceph_client *cl = mdsc->fsc->client;3797	struct ceph_mds_request *req;3798	struct ceph_mds_reply_head *head = msg->front.iov_base;3799	struct ceph_mds_reply_info_parsed *rinfo;  /* parsed reply info */3800	struct ceph_snap_realm *realm;3801	u64 tid;3802	int err, result;3803	int mds = session->s_mds;3804	bool close_sessions = false;3805 3806	if (msg->front.iov_len < sizeof(*head)) {3807		pr_err_client(cl, "got corrupt (short) reply\n");3808		ceph_msg_dump(msg);3809		return;3810	}3811 3812	/* get request, session */3813	tid = le64_to_cpu(msg->hdr.tid);3814	mutex_lock(&mdsc->mutex);3815	req = lookup_get_request(mdsc, tid);3816	if (!req) {3817		doutc(cl, "on unknown tid %llu\n", tid);3818		mutex_unlock(&mdsc->mutex);3819		return;3820	}3821	doutc(cl, "handle_reply %p\n", req);3822 3823	/* correct session? */3824	if (req->r_session != session) {3825		pr_err_client(cl, "got %llu on session mds%d not mds%d\n",3826			      tid, session->s_mds,3827			      req->r_session ? req->r_session->s_mds : -1);3828		mutex_unlock(&mdsc->mutex);3829		goto out;3830	}3831 3832	/* dup? */3833	if ((test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags) && !head->safe) ||3834	    (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags) && head->safe)) {3835		pr_warn_client(cl, "got a dup %s reply on %llu from mds%d\n",3836			       head->safe ? "safe" : "unsafe", tid, mds);3837		mutex_unlock(&mdsc->mutex);3838		goto out;3839	}3840	if (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags)) {3841		pr_warn_client(cl, "got unsafe after safe on %llu from mds%d\n",3842			       tid, mds);3843		mutex_unlock(&mdsc->mutex);3844		goto out;3845	}3846 3847	result = le32_to_cpu(head->result);3848 3849	if (head->safe) {3850		set_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags);3851		__unregister_request(mdsc, req);3852 3853		/* last request during umount? */3854		if (mdsc->stopping && !__get_oldest_req(mdsc))3855			complete_all(&mdsc->safe_umount_waiters);3856 3857		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {3858			/*3859			 * We already handled the unsafe response, now do the3860			 * cleanup.  No need to examine the response; the MDS3861			 * doesn't include any result info in the safe3862			 * response.  And even if it did, there is nothing3863			 * useful we could do with a revised return value.3864			 */3865			doutc(cl, "got safe reply %llu, mds%d\n", tid, mds);3866 3867			mutex_unlock(&mdsc->mutex);3868			goto out;3869		}3870	} else {3871		set_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags);3872		list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);3873	}3874 3875	doutc(cl, "tid %lld result %d\n", tid, result);3876	if (test_bit(CEPHFS_FEATURE_REPLY_ENCODING, &session->s_features))3877		err = parse_reply_info(session, msg, req, (u64)-1);3878	else3879		err = parse_reply_info(session, msg, req,3880				       session->s_con.peer_features);3881	mutex_unlock(&mdsc->mutex);3882 3883	/* Must find target inode outside of mutexes to avoid deadlocks */3884	rinfo = &req->r_reply_info;3885	if ((err >= 0) && rinfo->head->is_target) {3886		struct inode *in = xchg(&req->r_new_inode, NULL);3887		struct ceph_vino tvino = {3888			.ino  = le64_to_cpu(rinfo->targeti.in->ino),3889			.snap = le64_to_cpu(rinfo->targeti.in->snapid)3890		};3891 3892		/*3893		 * If we ended up opening an existing inode, discard3894		 * r_new_inode3895		 */3896		if (req->r_op == CEPH_MDS_OP_CREATE &&3897		    !req->r_reply_info.has_create_ino) {3898			/* This should never happen on an async create */3899			WARN_ON_ONCE(req->r_deleg_ino);3900			iput(in);3901			in = NULL;3902		}3903 3904		in = ceph_get_inode(mdsc->fsc->sb, tvino, in);3905		if (IS_ERR(in)) {3906			err = PTR_ERR(in);3907			mutex_lock(&session->s_mutex);3908			goto out_err;3909		}3910		req->r_target_inode = in;3911	}3912 3913	mutex_lock(&session->s_mutex);3914	if (err < 0) {3915		pr_err_client(cl, "got corrupt reply mds%d(tid:%lld)\n",3916			      mds, tid);3917		ceph_msg_dump(msg);3918		goto out_err;3919	}3920 3921	/* snap trace */3922	realm = NULL;3923	if (rinfo->snapblob_len) {3924		down_write(&mdsc->snap_rwsem);3925		err = ceph_update_snap_trace(mdsc, rinfo->snapblob,3926				rinfo->snapblob + rinfo->snapblob_len,3927				le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP,3928				&realm);3929		if (err) {3930			up_write(&mdsc->snap_rwsem);3931			close_sessions = true;3932			if (err == -EIO)3933				ceph_msg_dump(msg);3934			goto out_err;3935		}3936		downgrade_write(&mdsc->snap_rwsem);3937	} else {3938		down_read(&mdsc->snap_rwsem);3939	}3940 3941	/* insert trace into our cache */3942	mutex_lock(&req->r_fill_mutex);3943	current->journal_info = req;3944	err = ceph_fill_trace(mdsc->fsc->sb, req);3945	if (err == 0) {3946		if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR ||3947				    req->r_op == CEPH_MDS_OP_LSSNAP))3948			err = ceph_readdir_prepopulate(req, req->r_session);3949	}3950	current->journal_info = NULL;3951	mutex_unlock(&req->r_fill_mutex);3952 3953	up_read(&mdsc->snap_rwsem);3954	if (realm)3955		ceph_put_snap_realm(mdsc, realm);3956 3957	if (err == 0) {3958		if (req->r_target_inode &&3959		    test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {3960			struct ceph_inode_info *ci =3961				ceph_inode(req->r_target_inode);3962			spin_lock(&ci->i_unsafe_lock);3963			list_add_tail(&req->r_unsafe_target_item,3964				      &ci->i_unsafe_iops);3965			spin_unlock(&ci->i_unsafe_lock);3966		}3967 3968		ceph_unreserve_caps(mdsc, &req->r_caps_reservation);3969	}3970out_err:3971	mutex_lock(&mdsc->mutex);3972	if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {3973		if (err) {3974			req->r_err = err;3975		} else {3976			req->r_reply =  ceph_msg_get(msg);3977			set_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags);3978		}3979	} else {3980		doutc(cl, "reply arrived after request %lld was aborted\n", tid);3981	}3982	mutex_unlock(&mdsc->mutex);3983 3984	mutex_unlock(&session->s_mutex);3985 3986	/* kick calling process */3987	complete_request(mdsc, req);3988 3989	ceph_update_metadata_metrics(&mdsc->metric, req->r_start_latency,3990				     req->r_end_latency, err);3991out:3992	ceph_mdsc_put_request(req);3993 3994	/* Defer closing the sessions after s_mutex lock being released */3995	if (close_sessions)3996		ceph_mdsc_close_sessions(mdsc);3997	return;3998}3999 4000 4001 4002/*4003 * handle mds notification that our request has been forwarded.4004 */4005static void handle_forward(struct ceph_mds_client *mdsc,4006			   struct ceph_mds_session *session,4007			   struct ceph_msg *msg)4008{4009	struct ceph_client *cl = mdsc->fsc->client;4010	struct ceph_mds_request *req;4011	u64 tid = le64_to_cpu(msg->hdr.tid);4012	u32 next_mds;4013	u32 fwd_seq;4014	int err = -EINVAL;4015	void *p = msg->front.iov_base;4016	void *end = p + msg->front.iov_len;4017	bool aborted = false;4018 4019	ceph_decode_need(&p, end, 2*sizeof(u32), bad);4020	next_mds = ceph_decode_32(&p);4021	fwd_seq = ceph_decode_32(&p);4022 4023	mutex_lock(&mdsc->mutex);4024	req = lookup_get_request(mdsc, tid);4025	if (!req) {4026		mutex_unlock(&mdsc->mutex);4027		doutc(cl, "forward tid %llu to mds%d - req dne\n", tid, next_mds);4028		return;  /* dup reply? */4029	}4030 4031	if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {4032		doutc(cl, "forward tid %llu aborted, unregistering\n", tid);4033		__unregister_request(mdsc, req);4034	} else if (fwd_seq <= req->r_num_fwd || (uint32_t)fwd_seq >= U32_MAX) {4035		/*4036		 * Avoid inifinite retrying after overflow.4037		 *4038		 * The MDS will increase the fwd count and in client side4039		 * if the num_fwd is less than the one saved in request4040		 * that means the MDS is an old version and overflowed of4041		 * 8 bits.4042		 */4043		mutex_lock(&req->r_fill_mutex);4044		req->r_err = -EMULTIHOP;4045		set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);4046		mutex_unlock(&req->r_fill_mutex);4047		aborted = true;4048		pr_warn_ratelimited_client(cl, "forward tid %llu seq overflow\n",4049					   tid);4050	} else {4051		/* resend. forward race not possible; mds would drop */4052		doutc(cl, "forward tid %llu to mds%d (we resend)\n", tid, next_mds);4053		BUG_ON(req->r_err);4054		BUG_ON(test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags));4055		req->r_attempts = 0;4056		req->r_num_fwd = fwd_seq;4057		req->r_resend_mds = next_mds;4058		put_request_session(req);4059		__do_request(mdsc, req);4060	}4061	mutex_unlock(&mdsc->mutex);4062 4063	/* kick calling process */4064	if (aborted)4065		complete_request(mdsc, req);4066	ceph_mdsc_put_request(req);4067	return;4068 4069bad:4070	pr_err_client(cl, "decode error err=%d\n", err);4071	ceph_msg_dump(msg);4072}4073 4074static int __decode_session_metadata(void **p, void *end,4075				     bool *blocklisted)4076{4077	/* map<string,string> */4078	u32 n;4079	bool err_str;4080	ceph_decode_32_safe(p, end, n, bad);4081	while (n-- > 0) {4082		u32 len;4083		ceph_decode_32_safe(p, end, len, bad);4084		ceph_decode_need(p, end, len, bad);4085		err_str = !strncmp(*p, "error_string", len);4086		*p += len;4087		ceph_decode_32_safe(p, end, len, bad);4088		ceph_decode_need(p, end, len, bad);4089		/*4090		 * Match "blocklisted (blacklisted)" from newer MDSes,4091		 * or "blacklisted" from older MDSes.4092		 */4093		if (err_str && strnstr(*p, "blacklisted", len))4094			*blocklisted = true;4095		*p += len;4096	}4097	return 0;4098bad:4099	return -1;4100}4101 4102/*4103 * handle a mds session control message4104 */4105static void handle_session(struct ceph_mds_session *session,4106			   struct ceph_msg *msg)4107{4108	struct ceph_mds_client *mdsc = session->s_mdsc;4109	struct ceph_client *cl = mdsc->fsc->client;4110	int mds = session->s_mds;4111	int msg_version = le16_to_cpu(msg->hdr.version);4112	void *p = msg->front.iov_base;4113	void *end = p + msg->front.iov_len;4114	struct ceph_mds_session_head *h;4115	struct ceph_mds_cap_auth *cap_auths = NULL;4116	u32 op, cap_auths_num = 0;4117	u64 seq, features = 0;4118	int wake = 0;4119	bool blocklisted = false;4120	u32 i;4121 4122 4123	/* decode */4124	ceph_decode_need(&p, end, sizeof(*h), bad);4125	h = p;4126	p += sizeof(*h);4127 4128	op = le32_to_cpu(h->op);4129	seq = le64_to_cpu(h->seq);4130 4131	if (msg_version >= 3) {4132		u32 len;4133		/* version >= 2 and < 5, decode metadata, skip otherwise4134		 * as it's handled via flags.4135		 */4136		if (msg_version >= 5)4137			ceph_decode_skip_map(&p, end, string, string, bad);4138		else if (__decode_session_metadata(&p, end, &blocklisted) < 0)4139			goto bad;4140 4141		/* version >= 3, feature bits */4142		ceph_decode_32_safe(&p, end, len, bad);4143		if (len) {4144			ceph_decode_64_safe(&p, end, features, bad);4145			p += len - sizeof(features);4146		}4147	}4148 4149	if (msg_version >= 5) {4150		u32 flags, len;4151 4152		/* version >= 4 */4153		ceph_decode_skip_16(&p, end, bad); /* struct_v, struct_cv */4154		ceph_decode_32_safe(&p, end, len, bad); /* len */4155		ceph_decode_skip_n(&p, end, len, bad); /* metric_spec */4156 4157		/* version >= 5, flags   */4158		ceph_decode_32_safe(&p, end, flags, bad);4159		if (flags & CEPH_SESSION_BLOCKLISTED) {4160			pr_warn_client(cl, "mds%d session blocklisted\n",4161				       session->s_mds);4162			blocklisted = true;4163		}4164	}4165 4166	if (msg_version >= 6) {4167		ceph_decode_32_safe(&p, end, cap_auths_num, bad);4168		doutc(cl, "cap_auths_num %d\n", cap_auths_num);4169 4170		if (cap_auths_num && op != CEPH_SESSION_OPEN) {4171			WARN_ON_ONCE(op != CEPH_SESSION_OPEN);4172			goto skip_cap_auths;4173		}4174 4175		cap_auths = kcalloc(cap_auths_num,4176				    sizeof(struct ceph_mds_cap_auth),4177				    GFP_KERNEL);4178		if (!cap_auths) {4179			pr_err_client(cl, "No memory for cap_auths\n");4180			return;4181		}4182 4183		for (i = 0; i < cap_auths_num; i++) {4184			u32 _len, j;4185 4186			/* struct_v, struct_compat, and struct_len in MDSCapAuth */4187			ceph_decode_skip_n(&p, end, 2 + sizeof(u32), bad);4188 4189			/* struct_v, struct_compat, and struct_len in MDSCapMatch */4190			ceph_decode_skip_n(&p, end, 2 + sizeof(u32), bad);4191			ceph_decode_64_safe(&p, end, cap_auths[i].match.uid, bad);4192			ceph_decode_32_safe(&p, end, _len, bad);4193			if (_len) {4194				cap_auths[i].match.gids = kcalloc(_len, sizeof(u32),4195								  GFP_KERNEL);4196				if (!cap_auths[i].match.gids) {4197					pr_err_client(cl, "No memory for gids\n");4198					goto fail;4199				}4200 4201				cap_auths[i].match.num_gids = _len;4202				for (j = 0; j < _len; j++)4203					ceph_decode_32_safe(&p, end,4204							    cap_auths[i].match.gids[j],4205							    bad);4206			}4207 4208			ceph_decode_32_safe(&p, end, _len, bad);4209			if (_len) {4210				cap_auths[i].match.path = kcalloc(_len + 1, sizeof(char),4211								  GFP_KERNEL);4212				if (!cap_auths[i].match.path) {4213					pr_err_client(cl, "No memory for path\n");4214					goto fail;4215				}4216				ceph_decode_copy(&p, cap_auths[i].match.path, _len);4217 4218				/* Remove the tailing '/' */4219				while (_len && cap_auths[i].match.path[_len - 1] == '/') {4220					cap_auths[i].match.path[_len - 1] = '\0';4221					_len -= 1;4222				}4223			}4224 4225			ceph_decode_32_safe(&p, end, _len, bad);4226			if (_len) {4227				cap_auths[i].match.fs_name = kcalloc(_len + 1, sizeof(char),4228								     GFP_KERNEL);4229				if (!cap_auths[i].match.fs_name) {4230					pr_err_client(cl, "No memory for fs_name\n");4231					goto fail;4232				}4233				ceph_decode_copy(&p, cap_auths[i].match.fs_name, _len);4234			}4235 4236			ceph_decode_8_safe(&p, end, cap_auths[i].match.root_squash, bad);4237			ceph_decode_8_safe(&p, end, cap_auths[i].readable, bad);4238			ceph_decode_8_safe(&p, end, cap_auths[i].writeable, bad);4239			doutc(cl, "uid %lld, num_gids %u, path %s, fs_name %s, root_squash %d, readable %d, writeable %d\n",4240			      cap_auths[i].match.uid, cap_auths[i].match.num_gids,4241			      cap_auths[i].match.path, cap_auths[i].match.fs_name,4242			      cap_auths[i].match.root_squash,4243			      cap_auths[i].readable, cap_auths[i].writeable);4244		}4245	}4246 4247skip_cap_auths:4248	mutex_lock(&mdsc->mutex);4249	if (op == CEPH_SESSION_OPEN) {4250		if (mdsc->s_cap_auths) {4251			for (i = 0; i < mdsc->s_cap_auths_num; i++) {4252				kfree(mdsc->s_cap_auths[i].match.gids);4253				kfree(mdsc->s_cap_auths[i].match.path);4254				kfree(mdsc->s_cap_auths[i].match.fs_name);4255			}4256			kfree(mdsc->s_cap_auths);4257		}4258		mdsc->s_cap_auths_num = cap_auths_num;4259		mdsc->s_cap_auths = cap_auths;4260	}4261	if (op == CEPH_SESSION_CLOSE) {4262		ceph_get_mds_session(session);4263		__unregister_session(mdsc, session);4264	}4265	/* FIXME: this ttl calculation is generous */4266	session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;4267	mutex_unlock(&mdsc->mutex);4268 4269	mutex_lock(&session->s_mutex);4270 4271	doutc(cl, "mds%d %s %p state %s seq %llu\n", mds,4272	      ceph_session_op_name(op), session,4273	      ceph_session_state_name(session->s_state), seq);4274 4275	if (session->s_state == CEPH_MDS_SESSION_HUNG) {4276		session->s_state = CEPH_MDS_SESSION_OPEN;4277		pr_info_client(cl, "mds%d came back\n", session->s_mds);4278	}4279 4280	switch (op) {4281	case CEPH_SESSION_OPEN:4282		if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)4283			pr_info_client(cl, "mds%d reconnect success\n",4284				       session->s_mds);4285 4286		session->s_features = features;4287		if (session->s_state == CEPH_MDS_SESSION_OPEN) {4288			pr_notice_client(cl, "mds%d is already opened\n",4289					 session->s_mds);4290		} else {4291			session->s_state = CEPH_MDS_SESSION_OPEN;4292			renewed_caps(mdsc, session, 0);4293			if (test_bit(CEPHFS_FEATURE_METRIC_COLLECT,4294				     &session->s_features))4295				metric_schedule_delayed(&mdsc->metric);4296		}4297 4298		/*4299		 * The connection maybe broken and the session in client4300		 * side has been reinitialized, need to update the seq4301		 * anyway.4302		 */4303		if (!session->s_seq && seq)4304			session->s_seq = seq;4305 4306		wake = 1;4307		if (mdsc->stopping)4308			__close_session(mdsc, session);4309		break;4310 4311	case CEPH_SESSION_RENEWCAPS:4312		if (session->s_renew_seq == seq)4313			renewed_caps(mdsc, session, 1);4314		break;4315 4316	case CEPH_SESSION_CLOSE:4317		if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)4318			pr_info_client(cl, "mds%d reconnect denied\n",4319				       session->s_mds);4320		session->s_state = CEPH_MDS_SESSION_CLOSED;4321		cleanup_session_requests(mdsc, session);4322		remove_session_caps(session);4323		wake = 2; /* for good measure */4324		wake_up_all(&mdsc->session_close_wq);4325		break;4326 4327	case CEPH_SESSION_STALE:4328		pr_info_client(cl, "mds%d caps went stale, renewing\n",4329			       session->s_mds);4330		atomic_inc(&session->s_cap_gen);4331		session->s_cap_ttl = jiffies - 1;4332		send_renew_caps(mdsc, session);4333		break;4334 4335	case CEPH_SESSION_RECALL_STATE:4336		ceph_trim_caps(mdsc, session, le32_to_cpu(h->max_caps));4337		break;4338 4339	case CEPH_SESSION_FLUSHMSG:4340		/* flush cap releases */4341		spin_lock(&session->s_cap_lock);4342		if (session->s_num_cap_releases)4343			ceph_flush_session_cap_releases(mdsc, session);4344		spin_unlock(&session->s_cap_lock);4345 4346		send_flushmsg_ack(mdsc, session, seq);4347		break;4348 4349	case CEPH_SESSION_FORCE_RO:4350		doutc(cl, "force_session_readonly %p\n", session);4351		spin_lock(&session->s_cap_lock);4352		session->s_readonly = true;4353		spin_unlock(&session->s_cap_lock);4354		wake_up_session_caps(session, FORCE_RO);4355		break;4356 4357	case CEPH_SESSION_REJECT:4358		WARN_ON(session->s_state != CEPH_MDS_SESSION_OPENING);4359		pr_info_client(cl, "mds%d rejected session\n",4360			       session->s_mds);4361		session->s_state = CEPH_MDS_SESSION_REJECTED;4362		cleanup_session_requests(mdsc, session);4363		remove_session_caps(session);4364		if (blocklisted)4365			mdsc->fsc->blocklisted = true;4366		wake = 2; /* for good measure */4367		break;4368 4369	default:4370		pr_err_client(cl, "bad op %d mds%d\n", op, mds);4371		WARN_ON(1);4372	}4373 4374	mutex_unlock(&session->s_mutex);4375	if (wake) {4376		mutex_lock(&mdsc->mutex);4377		__wake_requests(mdsc, &session->s_waiting);4378		if (wake == 2)4379			kick_requests(mdsc, mds);4380		mutex_unlock(&mdsc->mutex);4381	}4382	if (op == CEPH_SESSION_CLOSE)4383		ceph_put_mds_session(session);4384	return;4385 4386bad:4387	pr_err_client(cl, "corrupt message mds%d len %d\n", mds,4388		      (int)msg->front.iov_len);4389	ceph_msg_dump(msg);4390fail:4391	for (i = 0; i < cap_auths_num; i++) {4392		kfree(cap_auths[i].match.gids);4393		kfree(cap_auths[i].match.path);4394		kfree(cap_auths[i].match.fs_name);4395	}4396	kfree(cap_auths);4397	return;4398}4399 4400void ceph_mdsc_release_dir_caps(struct ceph_mds_request *req)4401{4402	struct ceph_client *cl = req->r_mdsc->fsc->client;4403	int dcaps;4404 4405	dcaps = xchg(&req->r_dir_caps, 0);4406	if (dcaps) {4407		doutc(cl, "releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));4408		ceph_put_cap_refs(ceph_inode(req->r_parent), dcaps);4409	}4410}4411 4412void ceph_mdsc_release_dir_caps_async(struct ceph_mds_request *req)4413{4414	struct ceph_client *cl = req->r_mdsc->fsc->client;4415	int dcaps;4416 4417	dcaps = xchg(&req->r_dir_caps, 0);4418	if (dcaps) {4419		doutc(cl, "releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));4420		ceph_put_cap_refs_async(ceph_inode(req->r_parent), dcaps);4421	}4422}4423 4424/*4425 * called under session->mutex.4426 */4427static void replay_unsafe_requests(struct ceph_mds_client *mdsc,4428				   struct ceph_mds_session *session)4429{4430	struct ceph_mds_request *req, *nreq;4431	struct rb_node *p;4432 4433	doutc(mdsc->fsc->client, "mds%d\n", session->s_mds);4434 4435	mutex_lock(&mdsc->mutex);4436	list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item)4437		__send_request(session, req, true);4438 4439	/*4440	 * also re-send old requests when MDS enters reconnect stage. So that MDS4441	 * can process completed request in clientreplay stage.4442	 */4443	p = rb_first(&mdsc->request_tree);4444	while (p) {4445		req = rb_entry(p, struct ceph_mds_request, r_node);4446		p = rb_next(p);4447		if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))4448			continue;4449		if (req->r_attempts == 0)4450			continue; /* only old requests */4451		if (!req->r_session)4452			continue;4453		if (req->r_session->s_mds != session->s_mds)4454			continue;4455 4456		ceph_mdsc_release_dir_caps_async(req);4457 4458		__send_request(session, req, true);4459	}4460	mutex_unlock(&mdsc->mutex);4461}4462 4463static int send_reconnect_partial(struct ceph_reconnect_state *recon_state)4464{4465	struct ceph_msg *reply;4466	struct ceph_pagelist *_pagelist;4467	struct page *page;4468	__le32 *addr;4469	int err = -ENOMEM;4470 4471	if (!recon_state->allow_multi)4472		return -ENOSPC;4473 4474	/* can't handle message that contains both caps and realm */4475	BUG_ON(!recon_state->nr_caps == !recon_state->nr_realms);4476 4477	/* pre-allocate new pagelist */4478	_pagelist = ceph_pagelist_alloc(GFP_NOFS);4479	if (!_pagelist)4480		return -ENOMEM;4481 4482	reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);4483	if (!reply)4484		goto fail_msg;4485 4486	/* placeholder for nr_caps */4487	err = ceph_pagelist_encode_32(_pagelist, 0);4488	if (err < 0)4489		goto fail;4490 4491	if (recon_state->nr_caps) {4492		/* currently encoding caps */4493		err = ceph_pagelist_encode_32(recon_state->pagelist, 0);4494		if (err)4495			goto fail;4496	} else {4497		/* placeholder for nr_realms (currently encoding relams) */4498		err = ceph_pagelist_encode_32(_pagelist, 0);4499		if (err < 0)4500			goto fail;4501	}4502 4503	err = ceph_pagelist_encode_8(recon_state->pagelist, 1);4504	if (err)4505		goto fail;4506 4507	page = list_first_entry(&recon_state->pagelist->head, struct page, lru);4508	addr = kmap_atomic(page);4509	if (recon_state->nr_caps) {4510		/* currently encoding caps */4511		*addr = cpu_to_le32(recon_state->nr_caps);4512	} else {4513		/* currently encoding relams */4514		*(addr + 1) = cpu_to_le32(recon_state->nr_realms);4515	}4516	kunmap_atomic(addr);4517 4518	reply->hdr.version = cpu_to_le16(5);4519	reply->hdr.compat_version = cpu_to_le16(4);4520 4521	reply->hdr.data_len = cpu_to_le32(recon_state->pagelist->length);4522	ceph_msg_data_add_pagelist(reply, recon_state->pagelist);4523 4524	ceph_con_send(&recon_state->session->s_con, reply);4525	ceph_pagelist_release(recon_state->pagelist);4526 4527	recon_state->pagelist = _pagelist;4528	recon_state->nr_caps = 0;4529	recon_state->nr_realms = 0;4530	recon_state->msg_version = 5;4531	return 0;4532fail:4533	ceph_msg_put(reply);4534fail_msg:4535	ceph_pagelist_release(_pagelist);4536	return err;4537}4538 4539static struct dentry* d_find_primary(struct inode *inode)4540{4541	struct dentry *alias, *dn = NULL;4542 4543	if (hlist_empty(&inode->i_dentry))4544		return NULL;4545 4546	spin_lock(&inode->i_lock);4547	if (hlist_empty(&inode->i_dentry))4548		goto out_unlock;4549 4550	if (S_ISDIR(inode->i_mode)) {4551		alias = hlist_entry(inode->i_dentry.first, struct dentry, d_u.d_alias);4552		if (!IS_ROOT(alias))4553			dn = dget(alias);4554		goto out_unlock;4555	}4556 4557	hlist_for_each_entry(alias, &inode->i_dentry, d_u.d_alias) {4558		spin_lock(&alias->d_lock);4559		if (!d_unhashed(alias) &&4560		    (ceph_dentry(alias)->flags & CEPH_DENTRY_PRIMARY_LINK)) {4561			dn = dget_dlock(alias);4562		}4563		spin_unlock(&alias->d_lock);4564		if (dn)4565			break;4566	}4567out_unlock:4568	spin_unlock(&inode->i_lock);4569	return dn;4570}4571 4572/*4573 * Encode information about a cap for a reconnect with the MDS.4574 */4575static int reconnect_caps_cb(struct inode *inode, int mds, void *arg)4576{4577	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);4578	struct ceph_client *cl = ceph_inode_to_client(inode);4579	union {4580		struct ceph_mds_cap_reconnect v2;4581		struct ceph_mds_cap_reconnect_v1 v1;4582	} rec;4583	struct ceph_inode_info *ci = ceph_inode(inode);4584	struct ceph_reconnect_state *recon_state = arg;4585	struct ceph_pagelist *pagelist = recon_state->pagelist;4586	struct dentry *dentry;4587	struct ceph_cap *cap;4588	char *path;4589	int pathlen = 0, err;4590	u64 pathbase;4591	u64 snap_follows;4592 4593	dentry = d_find_primary(inode);4594	if (dentry) {4595		/* set pathbase to parent dir when msg_version >= 2 */4596		path = ceph_mdsc_build_path(mdsc, dentry, &pathlen, &pathbase,4597					    recon_state->msg_version >= 2);4598		dput(dentry);4599		if (IS_ERR(path)) {4600			err = PTR_ERR(path);4601			goto out_err;4602		}4603	} else {4604		path = NULL;4605		pathbase = 0;4606	}4607 4608	spin_lock(&ci->i_ceph_lock);4609	cap = __get_cap_for_mds(ci, mds);4610	if (!cap) {4611		spin_unlock(&ci->i_ceph_lock);4612		err = 0;4613		goto out_err;4614	}4615	doutc(cl, " adding %p ino %llx.%llx cap %p %lld %s\n", inode,4616	      ceph_vinop(inode), cap, cap->cap_id,4617	      ceph_cap_string(cap->issued));4618 4619	cap->seq = 0;        /* reset cap seq */4620	cap->issue_seq = 0;  /* and issue_seq */4621	cap->mseq = 0;       /* and migrate_seq */4622	cap->cap_gen = atomic_read(&cap->session->s_cap_gen);4623 4624	/* These are lost when the session goes away */4625	if (S_ISDIR(inode->i_mode)) {4626		if (cap->issued & CEPH_CAP_DIR_CREATE) {4627			ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns));4628			memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));4629		}4630		cap->issued &= ~CEPH_CAP_ANY_DIR_OPS;4631	}4632 4633	if (recon_state->msg_version >= 2) {4634		rec.v2.cap_id = cpu_to_le64(cap->cap_id);4635		rec.v2.wanted = cpu_to_le32(__ceph_caps_wanted(ci));4636		rec.v2.issued = cpu_to_le32(cap->issued);4637		rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);4638		rec.v2.pathbase = cpu_to_le64(pathbase);4639		rec.v2.flock_len = (__force __le32)4640			((ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK) ? 0 : 1);4641	} else {4642		struct timespec64 ts;4643 4644		rec.v1.cap_id = cpu_to_le64(cap->cap_id);4645		rec.v1.wanted = cpu_to_le32(__ceph_caps_wanted(ci));4646		rec.v1.issued = cpu_to_le32(cap->issued);4647		rec.v1.size = cpu_to_le64(i_size_read(inode));4648		ts = inode_get_mtime(inode);4649		ceph_encode_timespec64(&rec.v1.mtime, &ts);4650		ts = inode_get_atime(inode);4651		ceph_encode_timespec64(&rec.v1.atime, &ts);4652		rec.v1.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);4653		rec.v1.pathbase = cpu_to_le64(pathbase);4654	}4655 4656	if (list_empty(&ci->i_cap_snaps)) {4657		snap_follows = ci->i_head_snapc ? ci->i_head_snapc->seq : 0;4658	} else {4659		struct ceph_cap_snap *capsnap =4660			list_first_entry(&ci->i_cap_snaps,4661					 struct ceph_cap_snap, ci_item);4662		snap_follows = capsnap->follows;4663	}4664	spin_unlock(&ci->i_ceph_lock);4665 4666	if (recon_state->msg_version >= 2) {4667		int num_fcntl_locks, num_flock_locks;4668		struct ceph_filelock *flocks = NULL;4669		size_t struct_len, total_len = sizeof(u64);4670		u8 struct_v = 0;4671 4672encode_again:4673		if (rec.v2.flock_len) {4674			ceph_count_locks(inode, &num_fcntl_locks, &num_flock_locks);4675		} else {4676			num_fcntl_locks = 0;4677			num_flock_locks = 0;4678		}4679		if (num_fcntl_locks + num_flock_locks > 0) {4680			flocks = kmalloc_array(num_fcntl_locks + num_flock_locks,4681					       sizeof(struct ceph_filelock),4682					       GFP_NOFS);4683			if (!flocks) {4684				err = -ENOMEM;4685				goto out_err;4686			}4687			err = ceph_encode_locks_to_buffer(inode, flocks,4688							  num_fcntl_locks,4689							  num_flock_locks);4690			if (err) {4691				kfree(flocks);4692				flocks = NULL;4693				if (err == -ENOSPC)4694					goto encode_again;4695				goto out_err;4696			}4697		} else {4698			kfree(flocks);4699			flocks = NULL;4700		}4701 4702		if (recon_state->msg_version >= 3) {4703			/* version, compat_version and struct_len */4704			total_len += 2 * sizeof(u8) + sizeof(u32);4705			struct_v = 2;4706		}4707		/*4708		 * number of encoded locks is stable, so copy to pagelist4709		 */4710		struct_len = 2 * sizeof(u32) +4711			    (num_fcntl_locks + num_flock_locks) *4712			    sizeof(struct ceph_filelock);4713		rec.v2.flock_len = cpu_to_le32(struct_len);4714 4715		struct_len += sizeof(u32) + pathlen + sizeof(rec.v2);4716 4717		if (struct_v >= 2)4718			struct_len += sizeof(u64); /* snap_follows */4719 4720		total_len += struct_len;4721 4722		if (pagelist->length + total_len > RECONNECT_MAX_SIZE) {4723			err = send_reconnect_partial(recon_state);4724			if (err)4725				goto out_freeflocks;4726			pagelist = recon_state->pagelist;4727		}4728 4729		err = ceph_pagelist_reserve(pagelist, total_len);4730		if (err)4731			goto out_freeflocks;4732 4733		ceph_pagelist_encode_64(pagelist, ceph_ino(inode));4734		if (recon_state->msg_version >= 3) {4735			ceph_pagelist_encode_8(pagelist, struct_v);4736			ceph_pagelist_encode_8(pagelist, 1);4737			ceph_pagelist_encode_32(pagelist, struct_len);4738		}4739		ceph_pagelist_encode_string(pagelist, path, pathlen);4740		ceph_pagelist_append(pagelist, &rec, sizeof(rec.v2));4741		ceph_locks_to_pagelist(flocks, pagelist,4742				       num_fcntl_locks, num_flock_locks);4743		if (struct_v >= 2)4744			ceph_pagelist_encode_64(pagelist, snap_follows);4745out_freeflocks:4746		kfree(flocks);4747	} else {4748		err = ceph_pagelist_reserve(pagelist,4749					    sizeof(u64) + sizeof(u32) +4750					    pathlen + sizeof(rec.v1));4751		if (err)4752			goto out_err;4753 4754		ceph_pagelist_encode_64(pagelist, ceph_ino(inode));4755		ceph_pagelist_encode_string(pagelist, path, pathlen);4756		ceph_pagelist_append(pagelist, &rec, sizeof(rec.v1));4757	}4758 4759out_err:4760	ceph_mdsc_free_path(path, pathlen);4761	if (!err)4762		recon_state->nr_caps++;4763	return err;4764}4765 4766static int encode_snap_realms(struct ceph_mds_client *mdsc,4767			      struct ceph_reconnect_state *recon_state)4768{4769	struct rb_node *p;4770	struct ceph_pagelist *pagelist = recon_state->pagelist;4771	struct ceph_client *cl = mdsc->fsc->client;4772	int err = 0;4773 4774	if (recon_state->msg_version >= 4) {4775		err = ceph_pagelist_encode_32(pagelist, mdsc->num_snap_realms);4776		if (err < 0)4777			goto fail;4778	}4779 4780	/*4781	 * snaprealms.  we provide mds with the ino, seq (version), and4782	 * parent for all of our realms.  If the mds has any newer info,4783	 * it will tell us.4784	 */4785	for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {4786		struct ceph_snap_realm *realm =4787		       rb_entry(p, struct ceph_snap_realm, node);4788		struct ceph_mds_snaprealm_reconnect sr_rec;4789 4790		if (recon_state->msg_version >= 4) {4791			size_t need = sizeof(u8) * 2 + sizeof(u32) +4792				      sizeof(sr_rec);4793 4794			if (pagelist->length + need > RECONNECT_MAX_SIZE) {4795				err = send_reconnect_partial(recon_state);4796				if (err)4797					goto fail;4798				pagelist = recon_state->pagelist;4799			}4800 4801			err = ceph_pagelist_reserve(pagelist, need);4802			if (err)4803				goto fail;4804 4805			ceph_pagelist_encode_8(pagelist, 1);4806			ceph_pagelist_encode_8(pagelist, 1);4807			ceph_pagelist_encode_32(pagelist, sizeof(sr_rec));4808		}4809 4810		doutc(cl, " adding snap realm %llx seq %lld parent %llx\n",4811		      realm->ino, realm->seq, realm->parent_ino);4812		sr_rec.ino = cpu_to_le64(realm->ino);4813		sr_rec.seq = cpu_to_le64(realm->seq);4814		sr_rec.parent = cpu_to_le64(realm->parent_ino);4815 4816		err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));4817		if (err)4818			goto fail;4819 4820		recon_state->nr_realms++;4821	}4822fail:4823	return err;4824}4825 4826 4827/*4828 * If an MDS fails and recovers, clients need to reconnect in order to4829 * reestablish shared state.  This includes all caps issued through4830 * this session _and_ the snap_realm hierarchy.  Because it's not4831 * clear which snap realms the mds cares about, we send everything we4832 * know about.. that ensures we'll then get any new info the4833 * recovering MDS might have.4834 *4835 * This is a relatively heavyweight operation, but it's rare.4836 */4837static void send_mds_reconnect(struct ceph_mds_client *mdsc,4838			       struct ceph_mds_session *session)4839{4840	struct ceph_client *cl = mdsc->fsc->client;4841	struct ceph_msg *reply;4842	int mds = session->s_mds;4843	int err = -ENOMEM;4844	struct ceph_reconnect_state recon_state = {4845		.session = session,4846	};4847	LIST_HEAD(dispose);4848 4849	pr_info_client(cl, "mds%d reconnect start\n", mds);4850 4851	recon_state.pagelist = ceph_pagelist_alloc(GFP_NOFS);4852	if (!recon_state.pagelist)4853		goto fail_nopagelist;4854 4855	reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);4856	if (!reply)4857		goto fail_nomsg;4858 4859	xa_destroy(&session->s_delegated_inos);4860 4861	mutex_lock(&session->s_mutex);4862	session->s_state = CEPH_MDS_SESSION_RECONNECTING;4863	session->s_seq = 0;4864 4865	doutc(cl, "session %p state %s\n", session,4866	      ceph_session_state_name(session->s_state));4867 4868	atomic_inc(&session->s_cap_gen);4869 4870	spin_lock(&session->s_cap_lock);4871	/* don't know if session is readonly */4872	session->s_readonly = 0;4873	/*4874	 * notify __ceph_remove_cap() that we are composing cap reconnect.4875	 * If a cap get released before being added to the cap reconnect,4876	 * __ceph_remove_cap() should skip queuing cap release.4877	 */4878	session->s_cap_reconnect = 1;4879	/* drop old cap expires; we're about to reestablish that state */4880	detach_cap_releases(session, &dispose);4881	spin_unlock(&session->s_cap_lock);4882	dispose_cap_releases(mdsc, &dispose);4883 4884	/* trim unused caps to reduce MDS's cache rejoin time */4885	if (mdsc->fsc->sb->s_root)4886		shrink_dcache_parent(mdsc->fsc->sb->s_root);4887 4888	ceph_con_close(&session->s_con);4889	ceph_con_open(&session->s_con,4890		      CEPH_ENTITY_TYPE_MDS, mds,4891		      ceph_mdsmap_get_addr(mdsc->mdsmap, mds));4892 4893	/* replay unsafe requests */4894	replay_unsafe_requests(mdsc, session);4895 4896	ceph_early_kick_flushing_caps(mdsc, session);4897 4898	down_read(&mdsc->snap_rwsem);4899 4900	/* placeholder for nr_caps */4901	err = ceph_pagelist_encode_32(recon_state.pagelist, 0);4902	if (err)4903		goto fail;4904 4905	if (test_bit(CEPHFS_FEATURE_MULTI_RECONNECT, &session->s_features)) {4906		recon_state.msg_version = 3;4907		recon_state.allow_multi = true;4908	} else if (session->s_con.peer_features & CEPH_FEATURE_MDSENC) {4909		recon_state.msg_version = 3;4910	} else {4911		recon_state.msg_version = 2;4912	}4913	/* traverse this session's caps */4914	err = ceph_iterate_session_caps(session, reconnect_caps_cb, &recon_state);4915 4916	spin_lock(&session->s_cap_lock);4917	session->s_cap_reconnect = 0;4918	spin_unlock(&session->s_cap_lock);4919 4920	if (err < 0)4921		goto fail;4922 4923	/* check if all realms can be encoded into current message */4924	if (mdsc->num_snap_realms) {4925		size_t total_len =4926			recon_state.pagelist->length +4927			mdsc->num_snap_realms *4928			sizeof(struct ceph_mds_snaprealm_reconnect);4929		if (recon_state.msg_version >= 4) {4930			/* number of realms */4931			total_len += sizeof(u32);4932			/* version, compat_version and struct_len */4933			total_len += mdsc->num_snap_realms *4934				     (2 * sizeof(u8) + sizeof(u32));4935		}4936		if (total_len > RECONNECT_MAX_SIZE) {4937			if (!recon_state.allow_multi) {4938				err = -ENOSPC;4939				goto fail;4940			}4941			if (recon_state.nr_caps) {4942				err = send_reconnect_partial(&recon_state);4943				if (err)4944					goto fail;4945			}4946			recon_state.msg_version = 5;4947		}4948	}4949 4950	err = encode_snap_realms(mdsc, &recon_state);4951	if (err < 0)4952		goto fail;4953 4954	if (recon_state.msg_version >= 5) {4955		err = ceph_pagelist_encode_8(recon_state.pagelist, 0);4956		if (err < 0)4957			goto fail;4958	}4959 4960	if (recon_state.nr_caps || recon_state.nr_realms) {4961		struct page *page =4962			list_first_entry(&recon_state.pagelist->head,4963					struct page, lru);4964		__le32 *addr = kmap_atomic(page);4965		if (recon_state.nr_caps) {4966			WARN_ON(recon_state.nr_realms != mdsc->num_snap_realms);4967			*addr = cpu_to_le32(recon_state.nr_caps);4968		} else if (recon_state.msg_version >= 4) {4969			*(addr + 1) = cpu_to_le32(recon_state.nr_realms);4970		}4971		kunmap_atomic(addr);4972	}4973 4974	reply->hdr.version = cpu_to_le16(recon_state.msg_version);4975	if (recon_state.msg_version >= 4)4976		reply->hdr.compat_version = cpu_to_le16(4);4977 4978	reply->hdr.data_len = cpu_to_le32(recon_state.pagelist->length);4979	ceph_msg_data_add_pagelist(reply, recon_state.pagelist);4980 4981	ceph_con_send(&session->s_con, reply);4982 4983	mutex_unlock(&session->s_mutex);4984 4985	mutex_lock(&mdsc->mutex);4986	__wake_requests(mdsc, &session->s_waiting);4987	mutex_unlock(&mdsc->mutex);4988 4989	up_read(&mdsc->snap_rwsem);4990	ceph_pagelist_release(recon_state.pagelist);4991	return;4992 4993fail:4994	ceph_msg_put(reply);4995	up_read(&mdsc->snap_rwsem);4996	mutex_unlock(&session->s_mutex);4997fail_nomsg:4998	ceph_pagelist_release(recon_state.pagelist);4999fail_nopagelist:5000	pr_err_client(cl, "error %d preparing reconnect for mds%d\n",5001		      err, mds);5002	return;5003}5004 5005 5006/*5007 * compare old and new mdsmaps, kicking requests5008 * and closing out old connections as necessary5009 *5010 * called under mdsc->mutex.5011 */5012static void check_new_map(struct ceph_mds_client *mdsc,5013			  struct ceph_mdsmap *newmap,5014			  struct ceph_mdsmap *oldmap)5015{5016	int i, j, err;5017	int oldstate, newstate;5018	struct ceph_mds_session *s;5019	unsigned long targets[DIV_ROUND_UP(CEPH_MAX_MDS, sizeof(unsigned long))] = {0};5020	struct ceph_client *cl = mdsc->fsc->client;5021 5022	doutc(cl, "new %u old %u\n", newmap->m_epoch, oldmap->m_epoch);5023 5024	if (newmap->m_info) {5025		for (i = 0; i < newmap->possible_max_rank; i++) {5026			for (j = 0; j < newmap->m_info[i].num_export_targets; j++)5027				set_bit(newmap->m_info[i].export_targets[j], targets);5028		}5029	}5030 5031	for (i = 0; i < oldmap->possible_max_rank && i < mdsc->max_sessions; i++) {5032		if (!mdsc->sessions[i])5033			continue;5034		s = mdsc->sessions[i];5035		oldstate = ceph_mdsmap_get_state(oldmap, i);5036		newstate = ceph_mdsmap_get_state(newmap, i);5037 5038		doutc(cl, "mds%d state %s%s -> %s%s (session %s)\n",5039		      i, ceph_mds_state_name(oldstate),5040		      ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "",5041		      ceph_mds_state_name(newstate),5042		      ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "",5043		      ceph_session_state_name(s->s_state));5044 5045		if (i >= newmap->possible_max_rank) {5046			/* force close session for stopped mds */5047			ceph_get_mds_session(s);5048			__unregister_session(mdsc, s);5049			__wake_requests(mdsc, &s->s_waiting);5050			mutex_unlock(&mdsc->mutex);5051 5052			mutex_lock(&s->s_mutex);5053			cleanup_session_requests(mdsc, s);5054			remove_session_caps(s);5055			mutex_unlock(&s->s_mutex);5056 5057			ceph_put_mds_session(s);5058 5059			mutex_lock(&mdsc->mutex);5060			kick_requests(mdsc, i);5061			continue;5062		}5063 5064		if (memcmp(ceph_mdsmap_get_addr(oldmap, i),5065			   ceph_mdsmap_get_addr(newmap, i),5066			   sizeof(struct ceph_entity_addr))) {5067			/* just close it */5068			mutex_unlock(&mdsc->mutex);5069			mutex_lock(&s->s_mutex);5070			mutex_lock(&mdsc->mutex);5071			ceph_con_close(&s->s_con);5072			mutex_unlock(&s->s_mutex);5073			s->s_state = CEPH_MDS_SESSION_RESTARTING;5074		} else if (oldstate == newstate) {5075			continue;  /* nothing new with this mds */5076		}5077 5078		/*5079		 * send reconnect?5080		 */5081		if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&5082		    newstate >= CEPH_MDS_STATE_RECONNECT) {5083			mutex_unlock(&mdsc->mutex);5084			clear_bit(i, targets);5085			send_mds_reconnect(mdsc, s);5086			mutex_lock(&mdsc->mutex);5087		}5088 5089		/*5090		 * kick request on any mds that has gone active.5091		 */5092		if (oldstate < CEPH_MDS_STATE_ACTIVE &&5093		    newstate >= CEPH_MDS_STATE_ACTIVE) {5094			if (oldstate != CEPH_MDS_STATE_CREATING &&5095			    oldstate != CEPH_MDS_STATE_STARTING)5096				pr_info_client(cl, "mds%d recovery completed\n",5097					       s->s_mds);5098			kick_requests(mdsc, i);5099			mutex_unlock(&mdsc->mutex);5100			mutex_lock(&s->s_mutex);5101			mutex_lock(&mdsc->mutex);5102			ceph_kick_flushing_caps(mdsc, s);5103			mutex_unlock(&s->s_mutex);5104			wake_up_session_caps(s, RECONNECT);5105		}5106	}5107 5108	/*5109	 * Only open and reconnect sessions that don't exist yet.5110	 */5111	for (i = 0; i < newmap->possible_max_rank; i++) {5112		/*5113		 * In case the import MDS is crashed just after5114		 * the EImportStart journal is flushed, so when5115		 * a standby MDS takes over it and is replaying5116		 * the EImportStart journal the new MDS daemon5117		 * will wait the client to reconnect it, but the5118		 * client may never register/open the session yet.5119		 *5120		 * Will try to reconnect that MDS daemon if the5121		 * rank number is in the export targets array and5122		 * is the up:reconnect state.5123		 */5124		newstate = ceph_mdsmap_get_state(newmap, i);5125		if (!test_bit(i, targets) || newstate != CEPH_MDS_STATE_RECONNECT)5126			continue;5127 5128		/*5129		 * The session maybe registered and opened by some5130		 * requests which were choosing random MDSes during5131		 * the mdsc->mutex's unlock/lock gap below in rare5132		 * case. But the related MDS daemon will just queue5133		 * that requests and be still waiting for the client's5134		 * reconnection request in up:reconnect state.5135		 */5136		s = __ceph_lookup_mds_session(mdsc, i);5137		if (likely(!s)) {5138			s = __open_export_target_session(mdsc, i);5139			if (IS_ERR(s)) {5140				err = PTR_ERR(s);5141				pr_err_client(cl,5142					      "failed to open export target session, err %d\n",5143					      err);5144				continue;5145			}5146		}5147		doutc(cl, "send reconnect to export target mds.%d\n", i);5148		mutex_unlock(&mdsc->mutex);5149		send_mds_reconnect(mdsc, s);5150		ceph_put_mds_session(s);5151		mutex_lock(&mdsc->mutex);5152	}5153 5154	for (i = 0; i < newmap->possible_max_rank && i < mdsc->max_sessions; i++) {5155		s = mdsc->sessions[i];5156		if (!s)5157			continue;5158		if (!ceph_mdsmap_is_laggy(newmap, i))5159			continue;5160		if (s->s_state == CEPH_MDS_SESSION_OPEN ||5161		    s->s_state == CEPH_MDS_SESSION_HUNG ||5162		    s->s_state == CEPH_MDS_SESSION_CLOSING) {5163			doutc(cl, " connecting to export targets of laggy mds%d\n", i);5164			__open_export_target_sessions(mdsc, s);5165		}5166	}5167}5168 5169 5170 5171/*5172 * leases5173 */5174 5175/*5176 * caller must hold session s_mutex, dentry->d_lock5177 */5178void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry)5179{5180	struct ceph_dentry_info *di = ceph_dentry(dentry);5181 5182	ceph_put_mds_session(di->lease_session);5183	di->lease_session = NULL;5184}5185 5186static void handle_lease(struct ceph_mds_client *mdsc,5187			 struct ceph_mds_session *session,5188			 struct ceph_msg *msg)5189{5190	struct ceph_client *cl = mdsc->fsc->client;5191	struct super_block *sb = mdsc->fsc->sb;5192	struct inode *inode;5193	struct dentry *parent, *dentry;5194	struct ceph_dentry_info *di;5195	int mds = session->s_mds;5196	struct ceph_mds_lease *h = msg->front.iov_base;5197	u32 seq;5198	struct ceph_vino vino;5199	struct qstr dname;5200	int release = 0;5201 5202	doutc(cl, "from mds%d\n", mds);5203 5204	if (!ceph_inc_mds_stopping_blocker(mdsc, session))5205		return;5206 5207	/* decode */5208	if (msg->front.iov_len < sizeof(*h) + sizeof(u32))5209		goto bad;5210	vino.ino = le64_to_cpu(h->ino);5211	vino.snap = CEPH_NOSNAP;5212	seq = le32_to_cpu(h->seq);5213	dname.len = get_unaligned_le32(h + 1);5214	if (msg->front.iov_len < sizeof(*h) + sizeof(u32) + dname.len)5215		goto bad;5216	dname.name = (void *)(h + 1) + sizeof(u32);5217 5218	/* lookup inode */5219	inode = ceph_find_inode(sb, vino);5220	doutc(cl, "%s, ino %llx %p %.*s\n", ceph_lease_op_name(h->action),5221	      vino.ino, inode, dname.len, dname.name);5222 5223	mutex_lock(&session->s_mutex);5224	if (!inode) {5225		doutc(cl, "no inode %llx\n", vino.ino);5226		goto release;5227	}5228 5229	/* dentry */5230	parent = d_find_alias(inode);5231	if (!parent) {5232		doutc(cl, "no parent dentry on inode %p\n", inode);5233		WARN_ON(1);5234		goto release;  /* hrm... */5235	}5236	dname.hash = full_name_hash(parent, dname.name, dname.len);5237	dentry = d_lookup(parent, &dname);5238	dput(parent);5239	if (!dentry)5240		goto release;5241 5242	spin_lock(&dentry->d_lock);5243	di = ceph_dentry(dentry);5244	switch (h->action) {5245	case CEPH_MDS_LEASE_REVOKE:5246		if (di->lease_session == session) {5247			if (ceph_seq_cmp(di->lease_seq, seq) > 0)5248				h->seq = cpu_to_le32(di->lease_seq);5249			__ceph_mdsc_drop_dentry_lease(dentry);5250		}5251		release = 1;5252		break;5253 5254	case CEPH_MDS_LEASE_RENEW:5255		if (di->lease_session == session &&5256		    di->lease_gen == atomic_read(&session->s_cap_gen) &&5257		    di->lease_renew_from &&5258		    di->lease_renew_after == 0) {5259			unsigned long duration =5260				msecs_to_jiffies(le32_to_cpu(h->duration_ms));5261 5262			di->lease_seq = seq;5263			di->time = di->lease_renew_from + duration;5264			di->lease_renew_after = di->lease_renew_from +5265				(duration >> 1);5266			di->lease_renew_from = 0;5267		}5268		break;5269	}5270	spin_unlock(&dentry->d_lock);5271	dput(dentry);5272 5273	if (!release)5274		goto out;5275 5276release:5277	/* let's just reuse the same message */5278	h->action = CEPH_MDS_LEASE_REVOKE_ACK;5279	ceph_msg_get(msg);5280	ceph_con_send(&session->s_con, msg);5281 5282out:5283	mutex_unlock(&session->s_mutex);5284	iput(inode);5285 5286	ceph_dec_mds_stopping_blocker(mdsc);5287	return;5288 5289bad:5290	ceph_dec_mds_stopping_blocker(mdsc);5291 5292	pr_err_client(cl, "corrupt lease message\n");5293	ceph_msg_dump(msg);5294}5295 5296void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,5297			      struct dentry *dentry, char action,5298			      u32 seq)5299{5300	struct ceph_client *cl = session->s_mdsc->fsc->client;5301	struct ceph_msg *msg;5302	struct ceph_mds_lease *lease;5303	struct inode *dir;5304	int len = sizeof(*lease) + sizeof(u32) + NAME_MAX;5305 5306	doutc(cl, "identry %p %s to mds%d\n", dentry, ceph_lease_op_name(action),5307	      session->s_mds);5308 5309	msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS, false);5310	if (!msg)5311		return;5312	lease = msg->front.iov_base;5313	lease->action = action;5314	lease->seq = cpu_to_le32(seq);5315 5316	spin_lock(&dentry->d_lock);5317	dir = d_inode(dentry->d_parent);5318	lease->ino = cpu_to_le64(ceph_ino(dir));5319	lease->first = lease->last = cpu_to_le64(ceph_snap(dir));5320 5321	put_unaligned_le32(dentry->d_name.len, lease + 1);5322	memcpy((void *)(lease + 1) + 4,5323	       dentry->d_name.name, dentry->d_name.len);5324	spin_unlock(&dentry->d_lock);5325 5326	ceph_con_send(&session->s_con, msg);5327}5328 5329/*5330 * lock unlock the session, to wait ongoing session activities5331 */5332static void lock_unlock_session(struct ceph_mds_session *s)5333{5334	mutex_lock(&s->s_mutex);5335	mutex_unlock(&s->s_mutex);5336}5337 5338static void maybe_recover_session(struct ceph_mds_client *mdsc)5339{5340	struct ceph_client *cl = mdsc->fsc->client;5341	struct ceph_fs_client *fsc = mdsc->fsc;5342 5343	if (!ceph_test_mount_opt(fsc, CLEANRECOVER))5344		return;5345 5346	if (READ_ONCE(fsc->mount_state) != CEPH_MOUNT_MOUNTED)5347		return;5348 5349	if (!READ_ONCE(fsc->blocklisted))5350		return;5351 5352	pr_info_client(cl, "auto reconnect after blocklisted\n");5353	ceph_force_reconnect(fsc->sb);5354}5355 5356bool check_session_state(struct ceph_mds_session *s)5357{5358	struct ceph_client *cl = s->s_mdsc->fsc->client;5359 5360	switch (s->s_state) {5361	case CEPH_MDS_SESSION_OPEN:5362		if (s->s_ttl && time_after(jiffies, s->s_ttl)) {5363			s->s_state = CEPH_MDS_SESSION_HUNG;5364			pr_info_client(cl, "mds%d hung\n", s->s_mds);5365		}5366		break;5367	case CEPH_MDS_SESSION_CLOSING:5368	case CEPH_MDS_SESSION_NEW:5369	case CEPH_MDS_SESSION_RESTARTING:5370	case CEPH_MDS_SESSION_CLOSED:5371	case CEPH_MDS_SESSION_REJECTED:5372		return false;5373	}5374 5375	return true;5376}5377 5378/*5379 * If the sequence is incremented while we're waiting on a REQUEST_CLOSE reply,5380 * then we need to retransmit that request.5381 */5382void inc_session_sequence(struct ceph_mds_session *s)5383{5384	struct ceph_client *cl = s->s_mdsc->fsc->client;5385 5386	lockdep_assert_held(&s->s_mutex);5387 5388	s->s_seq++;5389 5390	if (s->s_state == CEPH_MDS_SESSION_CLOSING) {5391		int ret;5392 5393		doutc(cl, "resending session close request for mds%d\n", s->s_mds);5394		ret = request_close_session(s);5395		if (ret < 0)5396			pr_err_client(cl, "unable to close session to mds%d: %d\n",5397				      s->s_mds, ret);5398	}5399}5400 5401/*5402 * delayed work -- periodically trim expired leases, renew caps with mds.  If5403 * the @delay parameter is set to 0 or if it's more than 5 secs, the default5404 * workqueue delay value of 5 secs will be used.5405 */5406static void schedule_delayed(struct ceph_mds_client *mdsc, unsigned long delay)5407{5408	unsigned long max_delay = HZ * 5;5409 5410	/* 5 secs default delay */5411	if (!delay || (delay > max_delay))5412		delay = max_delay;5413	schedule_delayed_work(&mdsc->delayed_work,5414			      round_jiffies_relative(delay));5415}5416 5417static void delayed_work(struct work_struct *work)5418{5419	struct ceph_mds_client *mdsc =5420		container_of(work, struct ceph_mds_client, delayed_work.work);5421	unsigned long delay;5422	int renew_interval;5423	int renew_caps;5424	int i;5425 5426	doutc(mdsc->fsc->client, "mdsc delayed_work\n");5427 5428	if (mdsc->stopping >= CEPH_MDSC_STOPPING_FLUSHED)5429		return;5430 5431	mutex_lock(&mdsc->mutex);5432	renew_interval = mdsc->mdsmap->m_session_timeout >> 2;5433	renew_caps = time_after_eq(jiffies, HZ*renew_interval +5434				   mdsc->last_renew_caps);5435	if (renew_caps)5436		mdsc->last_renew_caps = jiffies;5437 5438	for (i = 0; i < mdsc->max_sessions; i++) {5439		struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);5440		if (!s)5441			continue;5442 5443		if (!check_session_state(s)) {5444			ceph_put_mds_session(s);5445			continue;5446		}5447		mutex_unlock(&mdsc->mutex);5448 5449		ceph_flush_session_cap_releases(mdsc, s);5450 5451		mutex_lock(&s->s_mutex);5452		if (renew_caps)5453			send_renew_caps(mdsc, s);5454		else5455			ceph_con_keepalive(&s->s_con);5456		if (s->s_state == CEPH_MDS_SESSION_OPEN ||5457		    s->s_state == CEPH_MDS_SESSION_HUNG)5458			ceph_send_cap_releases(mdsc, s);5459		mutex_unlock(&s->s_mutex);5460		ceph_put_mds_session(s);5461 5462		mutex_lock(&mdsc->mutex);5463	}5464	mutex_unlock(&mdsc->mutex);5465 5466	delay = ceph_check_delayed_caps(mdsc);5467 5468	ceph_queue_cap_reclaim_work(mdsc);5469 5470	ceph_trim_snapid_map(mdsc);5471 5472	maybe_recover_session(mdsc);5473 5474	schedule_delayed(mdsc, delay);5475}5476 5477int ceph_mdsc_init(struct ceph_fs_client *fsc)5478 5479{5480	struct ceph_mds_client *mdsc;5481	int err;5482 5483	mdsc = kzalloc(sizeof(struct ceph_mds_client), GFP_NOFS);5484	if (!mdsc)5485		return -ENOMEM;5486	mdsc->fsc = fsc;5487	mutex_init(&mdsc->mutex);5488	mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS);5489	if (!mdsc->mdsmap) {5490		err = -ENOMEM;5491		goto err_mdsc;5492	}5493 5494	init_completion(&mdsc->safe_umount_waiters);5495	spin_lock_init(&mdsc->stopping_lock);5496	atomic_set(&mdsc->stopping_blockers, 0);5497	init_completion(&mdsc->stopping_waiter);5498	init_waitqueue_head(&mdsc->session_close_wq);5499	INIT_LIST_HEAD(&mdsc->waiting_for_map);5500	mdsc->quotarealms_inodes = RB_ROOT;5501	mutex_init(&mdsc->quotarealms_inodes_mutex);5502	init_rwsem(&mdsc->snap_rwsem);5503	mdsc->snap_realms = RB_ROOT;5504	INIT_LIST_HEAD(&mdsc->snap_empty);5505	spin_lock_init(&mdsc->snap_empty_lock);5506	mdsc->request_tree = RB_ROOT;5507	INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);5508	mdsc->last_renew_caps = jiffies;5509	INIT_LIST_HEAD(&mdsc->cap_delay_list);5510#ifdef CONFIG_DEBUG_FS5511	INIT_LIST_HEAD(&mdsc->cap_wait_list);5512#endif5513	spin_lock_init(&mdsc->cap_delay_lock);5514	INIT_LIST_HEAD(&mdsc->cap_unlink_delay_list);5515	INIT_LIST_HEAD(&mdsc->snap_flush_list);5516	spin_lock_init(&mdsc->snap_flush_lock);5517	mdsc->last_cap_flush_tid = 1;5518	INIT_LIST_HEAD(&mdsc->cap_flush_list);5519	INIT_LIST_HEAD(&mdsc->cap_dirty_migrating);5520	spin_lock_init(&mdsc->cap_dirty_lock);5521	init_waitqueue_head(&mdsc->cap_flushing_wq);5522	INIT_WORK(&mdsc->cap_reclaim_work, ceph_cap_reclaim_work);5523	INIT_WORK(&mdsc->cap_unlink_work, ceph_cap_unlink_work);5524	err = ceph_metric_init(&mdsc->metric);5525	if (err)5526		goto err_mdsmap;5527 5528	spin_lock_init(&mdsc->dentry_list_lock);5529	INIT_LIST_HEAD(&mdsc->dentry_leases);5530	INIT_LIST_HEAD(&mdsc->dentry_dir_leases);5531 5532	ceph_caps_init(mdsc);5533	ceph_adjust_caps_max_min(mdsc, fsc->mount_options);5534 5535	spin_lock_init(&mdsc->snapid_map_lock);5536	mdsc->snapid_map_tree = RB_ROOT;5537	INIT_LIST_HEAD(&mdsc->snapid_map_lru);5538 5539	init_rwsem(&mdsc->pool_perm_rwsem);5540	mdsc->pool_perm_tree = RB_ROOT;5541 5542	strscpy(mdsc->nodename, utsname()->nodename,5543		sizeof(mdsc->nodename));5544 5545	fsc->mdsc = mdsc;5546	return 0;5547 5548err_mdsmap:5549	kfree(mdsc->mdsmap);5550err_mdsc:5551	kfree(mdsc);5552	return err;5553}5554 5555/*5556 * Wait for safe replies on open mds requests.  If we time out, drop5557 * all requests from the tree to avoid dangling dentry refs.5558 */5559static void wait_requests(struct ceph_mds_client *mdsc)5560{5561	struct ceph_client *cl = mdsc->fsc->client;5562	struct ceph_options *opts = mdsc->fsc->client->options;5563	struct ceph_mds_request *req;5564 5565	mutex_lock(&mdsc->mutex);5566	if (__get_oldest_req(mdsc)) {5567		mutex_unlock(&mdsc->mutex);5568 5569		doutc(cl, "waiting for requests\n");5570		wait_for_completion_timeout(&mdsc->safe_umount_waiters,5571				    ceph_timeout_jiffies(opts->mount_timeout));5572 5573		/* tear down remaining requests */5574		mutex_lock(&mdsc->mutex);5575		while ((req = __get_oldest_req(mdsc))) {5576			doutc(cl, "timed out on tid %llu\n", req->r_tid);5577			list_del_init(&req->r_wait);5578			__unregister_request(mdsc, req);5579		}5580	}5581	mutex_unlock(&mdsc->mutex);5582	doutc(cl, "done\n");5583}5584 5585void send_flush_mdlog(struct ceph_mds_session *s)5586{5587	struct ceph_client *cl = s->s_mdsc->fsc->client;5588	struct ceph_msg *msg;5589 5590	/*5591	 * Pre-luminous MDS crashes when it sees an unknown session request5592	 */5593	if (!CEPH_HAVE_FEATURE(s->s_con.peer_features, SERVER_LUMINOUS))5594		return;5595 5596	mutex_lock(&s->s_mutex);5597	doutc(cl, "request mdlog flush to mds%d (%s)s seq %lld\n",5598	      s->s_mds, ceph_session_state_name(s->s_state), s->s_seq);5599	msg = ceph_create_session_msg(CEPH_SESSION_REQUEST_FLUSH_MDLOG,5600				      s->s_seq);5601	if (!msg) {5602		pr_err_client(cl, "failed to request mdlog flush to mds%d (%s) seq %lld\n",5603			      s->s_mds, ceph_session_state_name(s->s_state), s->s_seq);5604	} else {5605		ceph_con_send(&s->s_con, msg);5606	}5607	mutex_unlock(&s->s_mutex);5608}5609 5610static int ceph_mds_auth_match(struct ceph_mds_client *mdsc,5611			       struct ceph_mds_cap_auth *auth,5612			       char *tpath)5613{5614	const struct cred *cred = get_current_cred();5615	u32 caller_uid = from_kuid(&init_user_ns, cred->fsuid);5616	u32 caller_gid = from_kgid(&init_user_ns, cred->fsgid);5617	struct ceph_client *cl = mdsc->fsc->client;5618	const char *spath = mdsc->fsc->mount_options->server_path;5619	bool gid_matched = false;5620	u32 gid, tlen, len;5621	int i, j;5622 5623	doutc(cl, "match.uid %lld\n", auth->match.uid);5624	if (auth->match.uid != MDS_AUTH_UID_ANY) {5625		if (auth->match.uid != caller_uid)5626			return 0;5627		if (auth->match.num_gids) {5628			for (i = 0; i < auth->match.num_gids; i++) {5629				if (caller_gid == auth->match.gids[i])5630					gid_matched = true;5631			}5632			if (!gid_matched && cred->group_info->ngroups) {5633				for (i = 0; i < cred->group_info->ngroups; i++) {5634					gid = from_kgid(&init_user_ns,5635							cred->group_info->gid[i]);5636					for (j = 0; j < auth->match.num_gids; j++) {5637						if (gid == auth->match.gids[j]) {5638							gid_matched = true;5639							break;5640						}5641					}5642					if (gid_matched)5643						break;5644				}5645			}5646			if (!gid_matched)5647				return 0;5648		}5649	}5650 5651	/* path match */5652	if (auth->match.path) {5653		if (!tpath)5654			return 0;5655 5656		tlen = strlen(tpath);5657		len = strlen(auth->match.path);5658		if (len) {5659			char *_tpath = tpath;5660			bool free_tpath = false;5661			int m, n;5662 5663			doutc(cl, "server path %s, tpath %s, match.path %s\n",5664			      spath, tpath, auth->match.path);5665			if (spath && (m = strlen(spath)) != 1) {5666				/* mount path + '/' + tpath + an extra space */5667				n = m + 1 + tlen + 1;5668				_tpath = kmalloc(n, GFP_NOFS);5669				if (!_tpath)5670					return -ENOMEM;5671				/* remove the leading '/' */5672				snprintf(_tpath, n, "%s/%s", spath + 1, tpath);5673				free_tpath = true;5674				tlen = strlen(_tpath);5675			}5676 5677			/*5678			 * Please note the tailing '/' for match.path has already5679			 * been removed when parsing.5680			 *5681			 * Remove the tailing '/' for the target path.5682			 */5683			while (tlen && _tpath[tlen - 1] == '/') {5684				_tpath[tlen - 1] = '\0';5685				tlen -= 1;5686			}5687			doutc(cl, "_tpath %s\n", _tpath);5688 5689			/*5690			 * In case first == _tpath && tlen == len:5691			 *  match.path=/foo  --> /foo _path=/foo     --> match5692			 *  match.path=/foo/ --> /foo _path=/foo     --> match5693			 *5694			 * In case first == _tmatch.path && tlen > len:5695			 *  match.path=/foo/ --> /foo _path=/foo/    --> match5696			 *  match.path=/foo  --> /foo _path=/foo/    --> match5697			 *  match.path=/foo/ --> /foo _path=/foo/d   --> match5698			 *  match.path=/foo  --> /foo _path=/food    --> mismatch5699			 *5700			 * All the other cases                       --> mismatch5701			 */5702			char *first = strstr(_tpath, auth->match.path);5703			if (first != _tpath) {5704				if (free_tpath)5705					kfree(_tpath);5706				return 0;5707			}5708 5709			if (tlen > len && _tpath[len] != '/') {5710				if (free_tpath)5711					kfree(_tpath);5712				return 0;5713			}5714		}5715	}5716 5717	doutc(cl, "matched\n");5718	return 1;5719}5720 5721int ceph_mds_check_access(struct ceph_mds_client *mdsc, char *tpath, int mask)5722{5723	const struct cred *cred = get_current_cred();5724	u32 caller_uid = from_kuid(&init_user_ns, cred->fsuid);5725	u32 caller_gid = from_kgid(&init_user_ns, cred->fsgid);5726	struct ceph_mds_cap_auth *rw_perms_s = NULL;5727	struct ceph_client *cl = mdsc->fsc->client;5728	bool root_squash_perms = true;5729	int i, err;5730 5731	doutc(cl, "tpath '%s', mask %d, caller_uid %d, caller_gid %d\n",5732	      tpath, mask, caller_uid, caller_gid);5733 5734	for (i = 0; i < mdsc->s_cap_auths_num; i++) {5735		struct ceph_mds_cap_auth *s = &mdsc->s_cap_auths[i];5736 5737		err = ceph_mds_auth_match(mdsc, s, tpath);5738		if (err < 0) {5739			return err;5740		} else if (err > 0) {5741			/* always follow the last auth caps' permision */5742			root_squash_perms = true;5743			rw_perms_s = NULL;5744			if ((mask & MAY_WRITE) && s->writeable &&5745			    s->match.root_squash && (!caller_uid || !caller_gid))5746				root_squash_perms = false;5747 5748			if (((mask & MAY_WRITE) && !s->writeable) ||5749			    ((mask & MAY_READ) && !s->readable))5750				rw_perms_s = s;5751		}5752	}5753 5754	doutc(cl, "root_squash_perms %d, rw_perms_s %p\n", root_squash_perms,5755	      rw_perms_s);5756	if (root_squash_perms && rw_perms_s == NULL) {5757		doutc(cl, "access allowed\n");5758		return 0;5759	}5760 5761	if (!root_squash_perms) {5762		doutc(cl, "root_squash is enabled and user(%d %d) isn't allowed to write",5763		      caller_uid, caller_gid);5764	}5765	if (rw_perms_s) {5766		doutc(cl, "mds auth caps readable/writeable %d/%d while request r/w %d/%d",5767		      rw_perms_s->readable, rw_perms_s->writeable,5768		      !!(mask & MAY_READ), !!(mask & MAY_WRITE));5769	}5770	doutc(cl, "access denied\n");5771	return -EACCES;5772}5773 5774/*5775 * called before mount is ro, and before dentries are torn down.5776 * (hmm, does this still race with new lookups?)5777 */5778void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc)5779{5780	doutc(mdsc->fsc->client, "begin\n");5781	mdsc->stopping = CEPH_MDSC_STOPPING_BEGIN;5782 5783	ceph_mdsc_iterate_sessions(mdsc, send_flush_mdlog, true);5784	ceph_mdsc_iterate_sessions(mdsc, lock_unlock_session, false);5785	ceph_flush_dirty_caps(mdsc);5786	wait_requests(mdsc);5787 5788	/*5789	 * wait for reply handlers to drop their request refs and5790	 * their inode/dcache refs5791	 */5792	ceph_msgr_flush();5793 5794	ceph_cleanup_quotarealms_inodes(mdsc);5795	doutc(mdsc->fsc->client, "done\n");5796}5797 5798/*5799 * flush the mdlog and wait for all write mds requests to flush.5800 */5801static void flush_mdlog_and_wait_mdsc_unsafe_requests(struct ceph_mds_client *mdsc,5802						 u64 want_tid)5803{5804	struct ceph_client *cl = mdsc->fsc->client;5805	struct ceph_mds_request *req = NULL, *nextreq;5806	struct ceph_mds_session *last_session = NULL;5807	struct rb_node *n;5808 5809	mutex_lock(&mdsc->mutex);5810	doutc(cl, "want %lld\n", want_tid);5811restart:5812	req = __get_oldest_req(mdsc);5813	while (req && req->r_tid <= want_tid) {5814		/* find next request */5815		n = rb_next(&req->r_node);5816		if (n)5817			nextreq = rb_entry(n, struct ceph_mds_request, r_node);5818		else5819			nextreq = NULL;5820		if (req->r_op != CEPH_MDS_OP_SETFILELOCK &&5821		    (req->r_op & CEPH_MDS_OP_WRITE)) {5822			struct ceph_mds_session *s = req->r_session;5823 5824			if (!s) {5825				req = nextreq;5826				continue;5827			}5828 5829			/* write op */5830			ceph_mdsc_get_request(req);5831			if (nextreq)5832				ceph_mdsc_get_request(nextreq);5833			s = ceph_get_mds_session(s);5834			mutex_unlock(&mdsc->mutex);5835 5836			/* send flush mdlog request to MDS */5837			if (last_session != s) {5838				send_flush_mdlog(s);5839				ceph_put_mds_session(last_session);5840				last_session = s;5841			} else {5842				ceph_put_mds_session(s);5843			}5844			doutc(cl, "wait on %llu (want %llu)\n",5845			      req->r_tid, want_tid);5846			wait_for_completion(&req->r_safe_completion);5847 5848			mutex_lock(&mdsc->mutex);5849			ceph_mdsc_put_request(req);5850			if (!nextreq)5851				break;  /* next dne before, so we're done! */5852			if (RB_EMPTY_NODE(&nextreq->r_node)) {5853				/* next request was removed from tree */5854				ceph_mdsc_put_request(nextreq);5855				goto restart;5856			}5857			ceph_mdsc_put_request(nextreq);  /* won't go away */5858		}5859		req = nextreq;5860	}5861	mutex_unlock(&mdsc->mutex);5862	ceph_put_mds_session(last_session);5863	doutc(cl, "done\n");5864}5865 5866void ceph_mdsc_sync(struct ceph_mds_client *mdsc)5867{5868	struct ceph_client *cl = mdsc->fsc->client;5869	u64 want_tid, want_flush;5870 5871	if (READ_ONCE(mdsc->fsc->mount_state) >= CEPH_MOUNT_SHUTDOWN)5872		return;5873 5874	doutc(cl, "sync\n");5875	mutex_lock(&mdsc->mutex);5876	want_tid = mdsc->last_tid;5877	mutex_unlock(&mdsc->mutex);5878 5879	ceph_flush_dirty_caps(mdsc);5880	ceph_flush_cap_releases(mdsc);5881	spin_lock(&mdsc->cap_dirty_lock);5882	want_flush = mdsc->last_cap_flush_tid;5883	if (!list_empty(&mdsc->cap_flush_list)) {5884		struct ceph_cap_flush *cf =5885			list_last_entry(&mdsc->cap_flush_list,5886					struct ceph_cap_flush, g_list);5887		cf->wake = true;5888	}5889	spin_unlock(&mdsc->cap_dirty_lock);5890 5891	doutc(cl, "sync want tid %lld flush_seq %lld\n", want_tid, want_flush);5892 5893	flush_mdlog_and_wait_mdsc_unsafe_requests(mdsc, want_tid);5894	wait_caps_flush(mdsc, want_flush);5895}5896 5897/*5898 * true if all sessions are closed, or we force unmount5899 */5900static bool done_closing_sessions(struct ceph_mds_client *mdsc, int skipped)5901{5902	if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)5903		return true;5904	return atomic_read(&mdsc->num_sessions) <= skipped;5905}5906 5907/*5908 * called after sb is ro or when metadata corrupted.5909 */5910void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc)5911{5912	struct ceph_options *opts = mdsc->fsc->client->options;5913	struct ceph_client *cl = mdsc->fsc->client;5914	struct ceph_mds_session *session;5915	int i;5916	int skipped = 0;5917 5918	doutc(cl, "begin\n");5919 5920	/* close sessions */5921	mutex_lock(&mdsc->mutex);5922	for (i = 0; i < mdsc->max_sessions; i++) {5923		session = __ceph_lookup_mds_session(mdsc, i);5924		if (!session)5925			continue;5926		mutex_unlock(&mdsc->mutex);5927		mutex_lock(&session->s_mutex);5928		if (__close_session(mdsc, session) <= 0)5929			skipped++;5930		mutex_unlock(&session->s_mutex);5931		ceph_put_mds_session(session);5932		mutex_lock(&mdsc->mutex);5933	}5934	mutex_unlock(&mdsc->mutex);5935 5936	doutc(cl, "waiting for sessions to close\n");5937	wait_event_timeout(mdsc->session_close_wq,5938			   done_closing_sessions(mdsc, skipped),5939			   ceph_timeout_jiffies(opts->mount_timeout));5940 5941	/* tear down remaining sessions */5942	mutex_lock(&mdsc->mutex);5943	for (i = 0; i < mdsc->max_sessions; i++) {5944		if (mdsc->sessions[i]) {5945			session = ceph_get_mds_session(mdsc->sessions[i]);5946			__unregister_session(mdsc, session);5947			mutex_unlock(&mdsc->mutex);5948			mutex_lock(&session->s_mutex);5949			remove_session_caps(session);5950			mutex_unlock(&session->s_mutex);5951			ceph_put_mds_session(session);5952			mutex_lock(&mdsc->mutex);5953		}5954	}5955	WARN_ON(!list_empty(&mdsc->cap_delay_list));5956	mutex_unlock(&mdsc->mutex);5957 5958	ceph_cleanup_snapid_map(mdsc);5959	ceph_cleanup_global_and_empty_realms(mdsc);5960 5961	cancel_work_sync(&mdsc->cap_reclaim_work);5962	cancel_work_sync(&mdsc->cap_unlink_work);5963	cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */5964 5965	doutc(cl, "done\n");5966}5967 5968void ceph_mdsc_force_umount(struct ceph_mds_client *mdsc)5969{5970	struct ceph_mds_session *session;5971	int mds;5972 5973	doutc(mdsc->fsc->client, "force umount\n");5974 5975	mutex_lock(&mdsc->mutex);5976	for (mds = 0; mds < mdsc->max_sessions; mds++) {5977		session = __ceph_lookup_mds_session(mdsc, mds);5978		if (!session)5979			continue;5980 5981		if (session->s_state == CEPH_MDS_SESSION_REJECTED)5982			__unregister_session(mdsc, session);5983		__wake_requests(mdsc, &session->s_waiting);5984		mutex_unlock(&mdsc->mutex);5985 5986		mutex_lock(&session->s_mutex);5987		__close_session(mdsc, session);5988		if (session->s_state == CEPH_MDS_SESSION_CLOSING) {5989			cleanup_session_requests(mdsc, session);5990			remove_session_caps(session);5991		}5992		mutex_unlock(&session->s_mutex);5993		ceph_put_mds_session(session);5994 5995		mutex_lock(&mdsc->mutex);5996		kick_requests(mdsc, mds);5997	}5998	__wake_requests(mdsc, &mdsc->waiting_for_map);5999	mutex_unlock(&mdsc->mutex);6000}6001 6002static void ceph_mdsc_stop(struct ceph_mds_client *mdsc)6003{6004	doutc(mdsc->fsc->client, "stop\n");6005	/*6006	 * Make sure the delayed work stopped before releasing6007	 * the resources.6008	 *6009	 * Because the cancel_delayed_work_sync() will only6010	 * guarantee that the work finishes executing. But the6011	 * delayed work will re-arm itself again after that.6012	 */6013	flush_delayed_work(&mdsc->delayed_work);6014 6015	if (mdsc->mdsmap)6016		ceph_mdsmap_destroy(mdsc->mdsmap);6017	kfree(mdsc->sessions);6018	ceph_caps_finalize(mdsc);6019 6020	if (mdsc->s_cap_auths) {6021		int i;6022 6023		for (i = 0; i < mdsc->s_cap_auths_num; i++) {6024			kfree(mdsc->s_cap_auths[i].match.gids);6025			kfree(mdsc->s_cap_auths[i].match.path);6026			kfree(mdsc->s_cap_auths[i].match.fs_name);6027		}6028		kfree(mdsc->s_cap_auths);6029	}6030 6031	ceph_pool_perm_destroy(mdsc);6032}6033 6034void ceph_mdsc_destroy(struct ceph_fs_client *fsc)6035{6036	struct ceph_mds_client *mdsc = fsc->mdsc;6037	doutc(fsc->client, "%p\n", mdsc);6038 6039	if (!mdsc)6040		return;6041 6042	/* flush out any connection work with references to us */6043	ceph_msgr_flush();6044 6045	ceph_mdsc_stop(mdsc);6046 6047	ceph_metric_destroy(&mdsc->metric);6048 6049	fsc->mdsc = NULL;6050	kfree(mdsc);6051	doutc(fsc->client, "%p done\n", mdsc);6052}6053 6054void ceph_mdsc_handle_fsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg)6055{6056	struct ceph_fs_client *fsc = mdsc->fsc;6057	struct ceph_client *cl = fsc->client;6058	const char *mds_namespace = fsc->mount_options->mds_namespace;6059	void *p = msg->front.iov_base;6060	void *end = p + msg->front.iov_len;6061	u32 epoch;6062	u32 num_fs;6063	u32 mount_fscid = (u32)-1;6064	int err = -EINVAL;6065 6066	ceph_decode_need(&p, end, sizeof(u32), bad);6067	epoch = ceph_decode_32(&p);6068 6069	doutc(cl, "epoch %u\n", epoch);6070 6071	/* struct_v, struct_cv, map_len, epoch, legacy_client_fscid */6072	ceph_decode_skip_n(&p, end, 2 + sizeof(u32) * 3, bad);6073 6074	ceph_decode_32_safe(&p, end, num_fs, bad);6075	while (num_fs-- > 0) {6076		void *info_p, *info_end;6077		u32 info_len;6078		u32 fscid, namelen;6079 6080		ceph_decode_need(&p, end, 2 + sizeof(u32), bad);6081		p += 2;		// info_v, info_cv6082		info_len = ceph_decode_32(&p);6083		ceph_decode_need(&p, end, info_len, bad);6084		info_p = p;6085		info_end = p + info_len;6086		p = info_end;6087 6088		ceph_decode_need(&info_p, info_end, sizeof(u32) * 2, bad);6089		fscid = ceph_decode_32(&info_p);6090		namelen = ceph_decode_32(&info_p);6091		ceph_decode_need(&info_p, info_end, namelen, bad);6092 6093		if (mds_namespace &&6094		    strlen(mds_namespace) == namelen &&6095		    !strncmp(mds_namespace, (char *)info_p, namelen)) {6096			mount_fscid = fscid;6097			break;6098		}6099	}6100 6101	ceph_monc_got_map(&fsc->client->monc, CEPH_SUB_FSMAP, epoch);6102	if (mount_fscid != (u32)-1) {6103		fsc->client->monc.fs_cluster_id = mount_fscid;6104		ceph_monc_want_map(&fsc->client->monc, CEPH_SUB_MDSMAP,6105				   0, true);6106		ceph_monc_renew_subs(&fsc->client->monc);6107	} else {6108		err = -ENOENT;6109		goto err_out;6110	}6111	return;6112 6113bad:6114	pr_err_client(cl, "error decoding fsmap %d. Shutting down mount.\n",6115		      err);6116	ceph_umount_begin(mdsc->fsc->sb);6117	ceph_msg_dump(msg);6118err_out:6119	mutex_lock(&mdsc->mutex);6120	mdsc->mdsmap_err = err;6121	__wake_requests(mdsc, &mdsc->waiting_for_map);6122	mutex_unlock(&mdsc->mutex);6123}6124 6125/*6126 * handle mds map update.6127 */6128void ceph_mdsc_handle_mdsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg)6129{6130	struct ceph_client *cl = mdsc->fsc->client;6131	u32 epoch;6132	u32 maplen;6133	void *p = msg->front.iov_base;6134	void *end = p + msg->front.iov_len;6135	struct ceph_mdsmap *newmap, *oldmap;6136	struct ceph_fsid fsid;6137	int err = -EINVAL;6138 6139	ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad);6140	ceph_decode_copy(&p, &fsid, sizeof(fsid));6141	if (ceph_check_fsid(mdsc->fsc->client, &fsid) < 0)6142		return;6143	epoch = ceph_decode_32(&p);6144	maplen = ceph_decode_32(&p);6145	doutc(cl, "epoch %u len %d\n", epoch, (int)maplen);6146 6147	/* do we need it? */6148	mutex_lock(&mdsc->mutex);6149	if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {6150		doutc(cl, "epoch %u <= our %u\n", epoch, mdsc->mdsmap->m_epoch);6151		mutex_unlock(&mdsc->mutex);6152		return;6153	}6154 6155	newmap = ceph_mdsmap_decode(mdsc, &p, end, ceph_msgr2(mdsc->fsc->client));6156	if (IS_ERR(newmap)) {6157		err = PTR_ERR(newmap);6158		goto bad_unlock;6159	}6160 6161	/* swap into place */6162	if (mdsc->mdsmap) {6163		oldmap = mdsc->mdsmap;6164		mdsc->mdsmap = newmap;6165		check_new_map(mdsc, newmap, oldmap);6166		ceph_mdsmap_destroy(oldmap);6167	} else {6168		mdsc->mdsmap = newmap;  /* first mds map */6169	}6170	mdsc->fsc->max_file_size = min((loff_t)mdsc->mdsmap->m_max_file_size,6171					MAX_LFS_FILESIZE);6172 6173	__wake_requests(mdsc, &mdsc->waiting_for_map);6174	ceph_monc_got_map(&mdsc->fsc->client->monc, CEPH_SUB_MDSMAP,6175			  mdsc->mdsmap->m_epoch);6176 6177	mutex_unlock(&mdsc->mutex);6178	schedule_delayed(mdsc, 0);6179	return;6180 6181bad_unlock:6182	mutex_unlock(&mdsc->mutex);6183bad:6184	pr_err_client(cl, "error decoding mdsmap %d. Shutting down mount.\n",6185		      err);6186	ceph_umount_begin(mdsc->fsc->sb);6187	ceph_msg_dump(msg);6188	return;6189}6190 6191static struct ceph_connection *mds_get_con(struct ceph_connection *con)6192{6193	struct ceph_mds_session *s = con->private;6194 6195	if (ceph_get_mds_session(s))6196		return con;6197	return NULL;6198}6199 6200static void mds_put_con(struct ceph_connection *con)6201{6202	struct ceph_mds_session *s = con->private;6203 6204	ceph_put_mds_session(s);6205}6206 6207/*6208 * if the client is unresponsive for long enough, the mds will kill6209 * the session entirely.6210 */6211static void mds_peer_reset(struct ceph_connection *con)6212{6213	struct ceph_mds_session *s = con->private;6214	struct ceph_mds_client *mdsc = s->s_mdsc;6215 6216	pr_warn_client(mdsc->fsc->client, "mds%d closed our session\n",6217		       s->s_mds);6218	if (READ_ONCE(mdsc->fsc->mount_state) != CEPH_MOUNT_FENCE_IO &&6219	    ceph_mdsmap_get_state(mdsc->mdsmap, s->s_mds) >= CEPH_MDS_STATE_RECONNECT)6220		send_mds_reconnect(mdsc, s);6221}6222 6223static void mds_dispatch(struct ceph_connection *con, struct ceph_msg *msg)6224{6225	struct ceph_mds_session *s = con->private;6226	struct ceph_mds_client *mdsc = s->s_mdsc;6227	struct ceph_client *cl = mdsc->fsc->client;6228	int type = le16_to_cpu(msg->hdr.type);6229 6230	mutex_lock(&mdsc->mutex);6231	if (__verify_registered_session(mdsc, s) < 0) {6232		mutex_unlock(&mdsc->mutex);6233		goto out;6234	}6235	mutex_unlock(&mdsc->mutex);6236 6237	switch (type) {6238	case CEPH_MSG_MDS_MAP:6239		ceph_mdsc_handle_mdsmap(mdsc, msg);6240		break;6241	case CEPH_MSG_FS_MAP_USER:6242		ceph_mdsc_handle_fsmap(mdsc, msg);6243		break;6244	case CEPH_MSG_CLIENT_SESSION:6245		handle_session(s, msg);6246		break;6247	case CEPH_MSG_CLIENT_REPLY:6248		handle_reply(s, msg);6249		break;6250	case CEPH_MSG_CLIENT_REQUEST_FORWARD:6251		handle_forward(mdsc, s, msg);6252		break;6253	case CEPH_MSG_CLIENT_CAPS:6254		ceph_handle_caps(s, msg);6255		break;6256	case CEPH_MSG_CLIENT_SNAP:6257		ceph_handle_snap(mdsc, s, msg);6258		break;6259	case CEPH_MSG_CLIENT_LEASE:6260		handle_lease(mdsc, s, msg);6261		break;6262	case CEPH_MSG_CLIENT_QUOTA:6263		ceph_handle_quota(mdsc, s, msg);6264		break;6265 6266	default:6267		pr_err_client(cl, "received unknown message type %d %s\n",6268			      type, ceph_msg_type_name(type));6269	}6270out:6271	ceph_msg_put(msg);6272}6273 6274/*6275 * authentication6276 */6277 6278/*6279 * Note: returned pointer is the address of a structure that's6280 * managed separately.  Caller must *not* attempt to free it.6281 */6282static struct ceph_auth_handshake *6283mds_get_authorizer(struct ceph_connection *con, int *proto, int force_new)6284{6285	struct ceph_mds_session *s = con->private;6286	struct ceph_mds_client *mdsc = s->s_mdsc;6287	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;6288	struct ceph_auth_handshake *auth = &s->s_auth;6289	int ret;6290 6291	ret = __ceph_auth_get_authorizer(ac, auth, CEPH_ENTITY_TYPE_MDS,6292					 force_new, proto, NULL, NULL);6293	if (ret)6294		return ERR_PTR(ret);6295 6296	return auth;6297}6298 6299static int mds_add_authorizer_challenge(struct ceph_connection *con,6300				    void *challenge_buf, int challenge_buf_len)6301{6302	struct ceph_mds_session *s = con->private;6303	struct ceph_mds_client *mdsc = s->s_mdsc;6304	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;6305 6306	return ceph_auth_add_authorizer_challenge(ac, s->s_auth.authorizer,6307					    challenge_buf, challenge_buf_len);6308}6309 6310static int mds_verify_authorizer_reply(struct ceph_connection *con)6311{6312	struct ceph_mds_session *s = con->private;6313	struct ceph_mds_client *mdsc = s->s_mdsc;6314	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;6315	struct ceph_auth_handshake *auth = &s->s_auth;6316 6317	return ceph_auth_verify_authorizer_reply(ac, auth->authorizer,6318		auth->authorizer_reply_buf, auth->authorizer_reply_buf_len,6319		NULL, NULL, NULL, NULL);6320}6321 6322static int mds_invalidate_authorizer(struct ceph_connection *con)6323{6324	struct ceph_mds_session *s = con->private;6325	struct ceph_mds_client *mdsc = s->s_mdsc;6326	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;6327 6328	ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS);6329 6330	return ceph_monc_validate_auth(&mdsc->fsc->client->monc);6331}6332 6333static int mds_get_auth_request(struct ceph_connection *con,6334				void *buf, int *buf_len,6335				void **authorizer, int *authorizer_len)6336{6337	struct ceph_mds_session *s = con->private;6338	struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;6339	struct ceph_auth_handshake *auth = &s->s_auth;6340	int ret;6341 6342	ret = ceph_auth_get_authorizer(ac, auth, CEPH_ENTITY_TYPE_MDS,6343				       buf, buf_len);6344	if (ret)6345		return ret;6346 6347	*authorizer = auth->authorizer_buf;6348	*authorizer_len = auth->authorizer_buf_len;6349	return 0;6350}6351 6352static int mds_handle_auth_reply_more(struct ceph_connection *con,6353				      void *reply, int reply_len,6354				      void *buf, int *buf_len,6355				      void **authorizer, int *authorizer_len)6356{6357	struct ceph_mds_session *s = con->private;6358	struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;6359	struct ceph_auth_handshake *auth = &s->s_auth;6360	int ret;6361 6362	ret = ceph_auth_handle_svc_reply_more(ac, auth, reply, reply_len,6363					      buf, buf_len);6364	if (ret)6365		return ret;6366 6367	*authorizer = auth->authorizer_buf;6368	*authorizer_len = auth->authorizer_buf_len;6369	return 0;6370}6371 6372static int mds_handle_auth_done(struct ceph_connection *con,6373				u64 global_id, void *reply, int reply_len,6374				u8 *session_key, int *session_key_len,6375				u8 *con_secret, int *con_secret_len)6376{6377	struct ceph_mds_session *s = con->private;6378	struct ceph_auth_client *ac = s->s_mdsc->fsc->client->monc.auth;6379	struct ceph_auth_handshake *auth = &s->s_auth;6380 6381	return ceph_auth_handle_svc_reply_done(ac, auth, reply, reply_len,6382					       session_key, session_key_len,6383					       con_secret, con_secret_len);6384}6385 6386static int mds_handle_auth_bad_method(struct ceph_connection *con,6387				      int used_proto, int result,6388				      const int *allowed_protos, int proto_cnt,6389				      const int *allowed_modes, int mode_cnt)6390{6391	struct ceph_mds_session *s = con->private;6392	struct ceph_mon_client *monc = &s->s_mdsc->fsc->client->monc;6393	int ret;6394 6395	if (ceph_auth_handle_bad_authorizer(monc->auth, CEPH_ENTITY_TYPE_MDS,6396					    used_proto, result,6397					    allowed_protos, proto_cnt,6398					    allowed_modes, mode_cnt)) {6399		ret = ceph_monc_validate_auth(monc);6400		if (ret)6401			return ret;6402	}6403 6404	return -EACCES;6405}6406 6407static struct ceph_msg *mds_alloc_msg(struct ceph_connection *con,6408				struct ceph_msg_header *hdr, int *skip)6409{6410	struct ceph_msg *msg;6411	int type = (int) le16_to_cpu(hdr->type);6412	int front_len = (int) le32_to_cpu(hdr->front_len);6413 6414	if (con->in_msg)6415		return con->in_msg;6416 6417	*skip = 0;6418	msg = ceph_msg_new(type, front_len, GFP_NOFS, false);6419	if (!msg) {6420		pr_err("unable to allocate msg type %d len %d\n",6421		       type, front_len);6422		return NULL;6423	}6424 6425	return msg;6426}6427 6428static int mds_sign_message(struct ceph_msg *msg)6429{6430       struct ceph_mds_session *s = msg->con->private;6431       struct ceph_auth_handshake *auth = &s->s_auth;6432 6433       return ceph_auth_sign_message(auth, msg);6434}6435 6436static int mds_check_message_signature(struct ceph_msg *msg)6437{6438       struct ceph_mds_session *s = msg->con->private;6439       struct ceph_auth_handshake *auth = &s->s_auth;6440 6441       return ceph_auth_check_message_signature(auth, msg);6442}6443 6444static const struct ceph_connection_operations mds_con_ops = {6445	.get = mds_get_con,6446	.put = mds_put_con,6447	.alloc_msg = mds_alloc_msg,6448	.dispatch = mds_dispatch,6449	.peer_reset = mds_peer_reset,6450	.get_authorizer = mds_get_authorizer,6451	.add_authorizer_challenge = mds_add_authorizer_challenge,6452	.verify_authorizer_reply = mds_verify_authorizer_reply,6453	.invalidate_authorizer = mds_invalidate_authorizer,6454	.sign_message = mds_sign_message,6455	.check_message_signature = mds_check_message_signature,6456	.get_auth_request = mds_get_auth_request,6457	.handle_auth_reply_more = mds_handle_auth_reply_more,6458	.handle_auth_done = mds_handle_auth_done,6459	.handle_auth_bad_method = mds_handle_auth_bad_method,6460};6461 6462/* eof */6463