// SPDX-License-Identifier: GPL-2.0

  #include <linux/ceph/ceph_debug.h>

  #include <linux/fs.h>

  #include <linux/wait.h>

  #include <linux/slab.h>

  #include <linux/gfp.h>

  #include <linux/sched.h>

  #include <linux/debugfs.h>
  #include <linux/seq_file.h>

  #include <linux/ratelimit.h>

  #include <linux/bits.h>

  #include <linux/ktime.h>

  #include "super.h"

  #include "mds_client.h"

  #include <linux/ceph/ceph_features.h>

  #include <linux/ceph/messenger.h>
  #include <linux/ceph/decode.h>
  #include <linux/ceph/pagelist.h>
  #include <linux/ceph/auth.h>
  #include <linux/ceph/debugfs.h>

  #define RECONNECT_MAX_SIZE (INT_MAX - PAGE_SIZE)

  /*
   * A cluster of MDS (metadata server) daemons is responsible for
   * managing the file system namespace (the directory hierarchy and
   * inodes) and for coordinating shared access to storage.  Metadata is
   * partitioning hierarchically across a number of servers, and that
   * partition varies over time as the cluster adjusts the distribution
   * in order to balance load.
   *
   * The MDS client is primarily responsible to managing synchronous
   * metadata requests for operations like open, unlink, and so forth.
   * If there is a MDS failure, we find out about it when we (possibly
   * request and) receive a new MDS map, and can resubmit affected
   * requests.
   *
   * For the most part, though, we take advantage of a lossless
   * communications channel to the MDS, and do not need to worry about
   * timing out or resubmitting requests.
   *
   * We maintain a stateful "session" with each MDS we interact with.
   * Within each session, we sent periodic heartbeat messages to ensure
   * any capabilities or leases we have been issues remain valid.  If
   * the session times out and goes stale, our leases and capabilities
   * are no longer valid.
   */

  struct ceph_reconnect_state {

  	struct ceph_mds_session *session;
  	int nr_caps, nr_realms;

  	struct ceph_pagelist *pagelist;

  	unsigned msg_version;

  	bool allow_multi;

  };

  static void __wake_requests(struct ceph_mds_client *mdsc,
  			    struct list_head *head);

  static void ceph_cap_release_work(struct work_struct *work);

  static void ceph_cap_reclaim_work(struct work_struct *work);

  static const struct ceph_connection_operations mds_con_ops;

  
  
  /*
   * mds reply parsing
   */

  static int parse_reply_info_quota(void **p, void *end,
  				  struct ceph_mds_reply_info_in *info)
  {
  	u8 struct_v, struct_compat;
  	u32 struct_len;
  
  	ceph_decode_8_safe(p, end, struct_v, bad);
  	ceph_decode_8_safe(p, end, struct_compat, bad);
  	/* struct_v is expected to be >= 1. we only
  	 * understand encoding with struct_compat == 1. */
  	if (!struct_v || struct_compat != 1)
  		goto bad;
  	ceph_decode_32_safe(p, end, struct_len, bad);
  	ceph_decode_need(p, end, struct_len, bad);
  	end = *p + struct_len;
  	ceph_decode_64_safe(p, end, info->max_bytes, bad);
  	ceph_decode_64_safe(p, end, info->max_files, bad);
  	*p = end;
  	return 0;
  bad:
  	return -EIO;
  }

  /*
   * parse individual inode info
   */
  static int parse_reply_info_in(void **p, void *end,

  			       struct ceph_mds_reply_info_in *info,

  			       u64 features)

  {

  	int err = 0;
  	u8 struct_v = 0;

  	if (features == (u64)-1) {
  		u32 struct_len;
  		u8 struct_compat;
  		ceph_decode_8_safe(p, end, struct_v, bad);
  		ceph_decode_8_safe(p, end, struct_compat, bad);
  		/* struct_v is expected to be >= 1. we only understand
  		 * encoding with struct_compat == 1. */
  		if (!struct_v || struct_compat != 1)
  			goto bad;
  		ceph_decode_32_safe(p, end, struct_len, bad);
  		ceph_decode_need(p, end, struct_len, bad);
  		end = *p + struct_len;
  	}
  
  	ceph_decode_need(p, end, sizeof(struct ceph_mds_reply_inode), bad);

  	info->in = *p;
  	*p += sizeof(struct ceph_mds_reply_inode) +
  		sizeof(*info->in->fragtree.splits) *
  		le32_to_cpu(info->in->fragtree.nsplits);
  
  	ceph_decode_32_safe(p, end, info->symlink_len, bad);
  	ceph_decode_need(p, end, info->symlink_len, bad);
  	info->symlink = *p;
  	*p += info->symlink_len;

  	ceph_decode_copy_safe(p, end, &info->dir_layout,
  			      sizeof(info->dir_layout), bad);

  	ceph_decode_32_safe(p, end, info->xattr_len, bad);
  	ceph_decode_need(p, end, info->xattr_len, bad);
  	info->xattr_data = *p;
  	*p += info->xattr_len;

  	if (features == (u64)-1) {
  		/* inline data */

  		ceph_decode_64_safe(p, end, info->inline_version, bad);
  		ceph_decode_32_safe(p, end, info->inline_len, bad);
  		ceph_decode_need(p, end, info->inline_len, bad);
  		info->inline_data = *p;
  		*p += info->inline_len;

  		/* quota */
  		err = parse_reply_info_quota(p, end, info);
  		if (err < 0)
  			goto out_bad;
  		/* pool namespace */
  		ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
  		if (info->pool_ns_len > 0) {
  			ceph_decode_need(p, end, info->pool_ns_len, bad);
  			info->pool_ns_data = *p;
  			*p += info->pool_ns_len;
  		}

  
  		/* btime */
  		ceph_decode_need(p, end, sizeof(info->btime), bad);
  		ceph_decode_copy(p, &info->btime, sizeof(info->btime));
  
  		/* change attribute */

  		ceph_decode_64_safe(p, end, info->change_attr, bad);

  		/* dir pin */
  		if (struct_v >= 2) {
  			ceph_decode_32_safe(p, end, info->dir_pin, bad);
  		} else {
  			info->dir_pin = -ENODATA;
  		}

  		/* snapshot birth time, remains zero for v<=2 */
  		if (struct_v >= 3) {
  			ceph_decode_need(p, end, sizeof(info->snap_btime), bad);
  			ceph_decode_copy(p, &info->snap_btime,
  					 sizeof(info->snap_btime));
  		} else {
  			memset(&info->snap_btime, 0, sizeof(info->snap_btime));
  		}

  		*p = end;
  	} else {
  		if (features & CEPH_FEATURE_MDS_INLINE_DATA) {
  			ceph_decode_64_safe(p, end, info->inline_version, bad);
  			ceph_decode_32_safe(p, end, info->inline_len, bad);
  			ceph_decode_need(p, end, info->inline_len, bad);
  			info->inline_data = *p;
  			*p += info->inline_len;
  		} else
  			info->inline_version = CEPH_INLINE_NONE;
  
  		if (features & CEPH_FEATURE_MDS_QUOTA) {
  			err = parse_reply_info_quota(p, end, info);
  			if (err < 0)
  				goto out_bad;
  		} else {
  			info->max_bytes = 0;
  			info->max_files = 0;
  		}
  
  		info->pool_ns_len = 0;
  		info->pool_ns_data = NULL;
  		if (features & CEPH_FEATURE_FS_FILE_LAYOUT_V2) {
  			ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
  			if (info->pool_ns_len > 0) {
  				ceph_decode_need(p, end, info->pool_ns_len, bad);
  				info->pool_ns_data = *p;
  				*p += info->pool_ns_len;
  			}
  		}

  		if (features & CEPH_FEATURE_FS_BTIME) {
  			ceph_decode_need(p, end, sizeof(info->btime), bad);
  			ceph_decode_copy(p, &info->btime, sizeof(info->btime));

  			ceph_decode_64_safe(p, end, info->change_attr, bad);

  		}

  		info->dir_pin = -ENODATA;

  		/* info->snap_btime remains zero */

  	}
  	return 0;
  bad:
  	err = -EIO;
  out_bad:
  	return err;
  }
  
  static int parse_reply_info_dir(void **p, void *end,
  				struct ceph_mds_reply_dirfrag **dirfrag,
  				u64 features)
  {
  	if (features == (u64)-1) {

  		u8 struct_v, struct_compat;
  		u32 struct_len;

  		ceph_decode_8_safe(p, end, struct_v, bad);
  		ceph_decode_8_safe(p, end, struct_compat, bad);

  		/* struct_v is expected to be >= 1. we only understand
  		 * encoding whose struct_compat == 1. */
  		if (!struct_v || struct_compat != 1)

  			goto bad;
  		ceph_decode_32_safe(p, end, struct_len, bad);
  		ceph_decode_need(p, end, struct_len, bad);

  		end = *p + struct_len;

  	}

  	ceph_decode_need(p, end, sizeof(**dirfrag), bad);
  	*dirfrag = *p;
  	*p += sizeof(**dirfrag) + sizeof(u32) * le32_to_cpu((*dirfrag)->ndist);
  	if (unlikely(*p > end))
  		goto bad;
  	if (features == (u64)-1)
  		*p = end;
  	return 0;
  bad:
  	return -EIO;
  }
  
  static int parse_reply_info_lease(void **p, void *end,
  				  struct ceph_mds_reply_lease **lease,
  				  u64 features)
  {
  	if (features == (u64)-1) {
  		u8 struct_v, struct_compat;
  		u32 struct_len;
  		ceph_decode_8_safe(p, end, struct_v, bad);
  		ceph_decode_8_safe(p, end, struct_compat, bad);
  		/* struct_v is expected to be >= 1. we only understand
  		 * encoding whose struct_compat == 1. */
  		if (!struct_v || struct_compat != 1)
  			goto bad;
  		ceph_decode_32_safe(p, end, struct_len, bad);
  		ceph_decode_need(p, end, struct_len, bad);
  		end = *p + struct_len;

  	}

  	ceph_decode_need(p, end, sizeof(**lease), bad);
  	*lease = *p;
  	*p += sizeof(**lease);
  	if (features == (u64)-1)
  		*p = end;

  	return 0;
  bad:

  	return -EIO;

  }
  
  /*
   * parse a normal reply, which may contain a (dir+)dentry and/or a
   * target inode.
   */
  static int parse_reply_info_trace(void **p, void *end,

  				  struct ceph_mds_reply_info_parsed *info,

  				  u64 features)

  {
  	int err;
  
  	if (info->head->is_dentry) {

  		err = parse_reply_info_in(p, end, &info->diri, features);

  		if (err < 0)
  			goto out_bad;

  		err = parse_reply_info_dir(p, end, &info->dirfrag, features);
  		if (err < 0)
  			goto out_bad;

  
  		ceph_decode_32_safe(p, end, info->dname_len, bad);
  		ceph_decode_need(p, end, info->dname_len, bad);
  		info->dname = *p;
  		*p += info->dname_len;

  
  		err = parse_reply_info_lease(p, end, &info->dlease, features);
  		if (err < 0)
  			goto out_bad;

  	}
  
  	if (info->head->is_target) {

  		err = parse_reply_info_in(p, end, &info->targeti, features);

  		if (err < 0)
  			goto out_bad;
  	}
  
  	if (unlikely(*p != end))
  		goto bad;
  	return 0;
  
  bad:
  	err = -EIO;
  out_bad:
  	pr_err("problem parsing mds trace %d
  ", err);
  	return err;
  }
  
  /*
   * parse readdir results
   */

  static int parse_reply_info_readdir(void **p, void *end,

  				struct ceph_mds_reply_info_parsed *info,

  				u64 features)

  {
  	u32 num, i = 0;
  	int err;

  	err = parse_reply_info_dir(p, end, &info->dir_dir, features);
  	if (err < 0)
  		goto out_bad;

  
  	ceph_decode_need(p, end, sizeof(num) + 2, bad);

  	num = ceph_decode_32(p);

  	{
  		u16 flags = ceph_decode_16(p);
  		info->dir_end = !!(flags & CEPH_READDIR_FRAG_END);
  		info->dir_complete = !!(flags & CEPH_READDIR_FRAG_COMPLETE);

  		info->hash_order = !!(flags & CEPH_READDIR_HASH_ORDER);

  		info->offset_hash = !!(flags & CEPH_READDIR_OFFSET_HASH);

  	}

  	if (num == 0)
  		goto done;

  	BUG_ON(!info->dir_entries);
  	if ((unsigned long)(info->dir_entries + num) >
  	    (unsigned long)info->dir_entries + info->dir_buf_size) {

  		pr_err("dir contents are larger than expected
  ");
  		WARN_ON(1);
  		goto bad;
  	}

  	info->dir_nr = num;

  	while (num) {

  		struct ceph_mds_reply_dir_entry *rde = info->dir_entries + i;

  		/* dentry */

  		ceph_decode_32_safe(p, end, rde->name_len, bad);

  		ceph_decode_need(p, end, rde->name_len, bad);
  		rde->name = *p;
  		*p += rde->name_len;
  		dout("parsed dir dname '%.*s'
  ", rde->name_len, rde->name);

  		/* dentry lease */
  		err = parse_reply_info_lease(p, end, &rde->lease, features);
  		if (err)
  			goto out_bad;

  		/* inode */

  		err = parse_reply_info_in(p, end, &rde->inode, features);

  		if (err < 0)
  			goto out_bad;

  		/* ceph_readdir_prepopulate() will update it */
  		rde->offset = 0;

  		i++;
  		num--;
  	}
  
  done:

  	/* Skip over any unrecognized fields */
  	*p = end;

  	return 0;
  
  bad:
  	err = -EIO;
  out_bad:
  	pr_err("problem parsing dir contents %d
  ", err);
  	return err;
  }
  
  /*

   * parse fcntl F_GETLK results
   */
  static int parse_reply_info_filelock(void **p, void *end,

  				     struct ceph_mds_reply_info_parsed *info,

  				     u64 features)

  {
  	if (*p + sizeof(*info->filelock_reply) > end)
  		goto bad;
  
  	info->filelock_reply = *p;

  	/* Skip over any unrecognized fields */
  	*p = end;

  	return 0;

  bad:
  	return -EIO;
  }

  
  #if BITS_PER_LONG == 64
  
  #define DELEGATED_INO_AVAILABLE		xa_mk_value(1)
  
  static int ceph_parse_deleg_inos(void **p, void *end,
  				 struct ceph_mds_session *s)
  {
  	u32 sets;
  
  	ceph_decode_32_safe(p, end, sets, bad);
  	dout("got %u sets of delegated inodes
  ", sets);
  	while (sets--) {
  		u64 start, len, ino;
  
  		ceph_decode_64_safe(p, end, start, bad);
  		ceph_decode_64_safe(p, end, len, bad);
  		while (len--) {
  			int err = xa_insert(&s->s_delegated_inos, ino = start++,
  					    DELEGATED_INO_AVAILABLE,
  					    GFP_KERNEL);
  			if (!err) {
  				dout("added delegated inode 0x%llx
  ",
  				     start - 1);
  			} else if (err == -EBUSY) {
  				pr_warn("ceph: MDS delegated inode 0x%llx more than once.
  ",
  					start - 1);
  			} else {
  				return err;
  			}
  		}
  	}
  	return 0;
  bad:
  	return -EIO;
  }
  
  u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
  {
  	unsigned long ino;
  	void *val;
  
  	xa_for_each(&s->s_delegated_inos, ino, val) {
  		val = xa_erase(&s->s_delegated_inos, ino);
  		if (val == DELEGATED_INO_AVAILABLE)
  			return ino;
  	}
  	return 0;
  }
  
  int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
  {
  	return xa_insert(&s->s_delegated_inos, ino, DELEGATED_INO_AVAILABLE,
  			 GFP_KERNEL);
  }
  #else /* BITS_PER_LONG == 64 */
  /*
   * FIXME: xarrays can't handle 64-bit indexes on a 32-bit arch. For now, just
   * ignore delegated_inos on 32 bit arch. Maybe eventually add xarrays for top
   * and bottom words?
   */
  static int ceph_parse_deleg_inos(void **p, void *end,
  				 struct ceph_mds_session *s)
  {
  	u32 sets;
  
  	ceph_decode_32_safe(p, end, sets, bad);
  	if (sets)
  		ceph_decode_skip_n(p, end, sets * 2 * sizeof(__le64), bad);
  	return 0;
  bad:
  	return -EIO;
  }
  
  u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
  {
  	return 0;
  }
  
  int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
  {
  	return 0;
  }
  #endif /* BITS_PER_LONG == 64 */

  /*

   * parse create results
   */
  static int parse_reply_info_create(void **p, void *end,
  				  struct ceph_mds_reply_info_parsed *info,

  				  u64 features, struct ceph_mds_session *s)

  {

  	int ret;

  	if (features == (u64)-1 ||
  	    (features & CEPH_FEATURE_REPLY_CREATE_INODE)) {

  		if (*p == end) {

  			/* Malformed reply? */

  			info->has_create_ino = false;

  		} else if (test_bit(CEPHFS_FEATURE_DELEG_INO, &s->s_features)) {
  			u8 struct_v, struct_compat;
  			u32 len;

  			info->has_create_ino = true;

  			ceph_decode_8_safe(p, end, struct_v, bad);
  			ceph_decode_8_safe(p, end, struct_compat, bad);
  			ceph_decode_32_safe(p, end, len, bad);
  			ceph_decode_64_safe(p, end, info->ino, bad);
  			ret = ceph_parse_deleg_inos(p, end, s);
  			if (ret)
  				return ret;
  		} else {
  			/* legacy */

  			ceph_decode_64_safe(p, end, info->ino, bad);

  			info->has_create_ino = true;

  		}

  	} else {
  		if (*p != end)
  			goto bad;

  	}

  	/* Skip over any unrecognized fields */
  	*p = end;

  	return 0;

  bad:
  	return -EIO;
  }
  
  /*

   * parse extra results
   */
  static int parse_reply_info_extra(void **p, void *end,

  				  struct ceph_mds_reply_info_parsed *info,

  				  u64 features, struct ceph_mds_session *s)

  {

  	u32 op = le32_to_cpu(info->head->op);
  
  	if (op == CEPH_MDS_OP_GETFILELOCK)

  		return parse_reply_info_filelock(p, end, info, features);

  	else if (op == CEPH_MDS_OP_READDIR || op == CEPH_MDS_OP_LSSNAP)

  		return parse_reply_info_readdir(p, end, info, features);

  	else if (op == CEPH_MDS_OP_CREATE)

  		return parse_reply_info_create(p, end, info, features, s);

  	else
  		return -EIO;

  }
  
  /*

   * parse entire mds reply
   */

  static int parse_reply_info(struct ceph_mds_session *s, struct ceph_msg *msg,

  			    struct ceph_mds_reply_info_parsed *info,

  			    u64 features)

  {
  	void *p, *end;
  	u32 len;
  	int err;
  
  	info->head = msg->front.iov_base;
  	p = msg->front.iov_base + sizeof(struct ceph_mds_reply_head);
  	end = p + msg->front.iov_len - sizeof(struct ceph_mds_reply_head);
  
  	/* trace */
  	ceph_decode_32_safe(&p, end, len, bad);
  	if (len > 0) {

  		ceph_decode_need(&p, end, len, bad);

  		err = parse_reply_info_trace(&p, p+len, info, features);

  		if (err < 0)
  			goto out_bad;
  	}

  	/* extra */

  	ceph_decode_32_safe(&p, end, len, bad);
  	if (len > 0) {

  		ceph_decode_need(&p, end, len, bad);

  		err = parse_reply_info_extra(&p, p+len, info, features, s);

  		if (err < 0)
  			goto out_bad;
  	}
  
  	/* snap blob */
  	ceph_decode_32_safe(&p, end, len, bad);
  	info->snapblob_len = len;
  	info->snapblob = p;
  	p += len;
  
  	if (p != end)
  		goto bad;
  	return 0;
  
  bad:
  	err = -EIO;
  out_bad:
  	pr_err("mds parse_reply err %d
  ", err);
  	return err;
  }
  
  static void destroy_reply_info(struct ceph_mds_reply_info_parsed *info)
  {

  	if (!info->dir_entries)

  		return;

  	free_pages((unsigned long)info->dir_entries, get_order(info->dir_buf_size));

  }
  
  
  /*
   * sessions
   */

  const char *ceph_session_state_name(int s)

  {
  	switch (s) {
  	case CEPH_MDS_SESSION_NEW: return "new";
  	case CEPH_MDS_SESSION_OPENING: return "opening";
  	case CEPH_MDS_SESSION_OPEN: return "open";
  	case CEPH_MDS_SESSION_HUNG: return "hung";
  	case CEPH_MDS_SESSION_CLOSING: return "closing";

  	case CEPH_MDS_SESSION_CLOSED: return "closed";

  	case CEPH_MDS_SESSION_RESTARTING: return "restarting";

  	case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting";

  	case CEPH_MDS_SESSION_REJECTED: return "rejected";

  	default: return "???";
  	}
  }

  struct ceph_mds_session *ceph_get_mds_session(struct ceph_mds_session *s)

  {

  	if (refcount_inc_not_zero(&s->s_ref)) {

  		dout("mdsc get_session %p %d -> %d
  ", s,

  		     refcount_read(&s->s_ref)-1, refcount_read(&s->s_ref));

  		return s;
  	} else {

  		dout("mdsc get_session %p 0 -- FAIL
  ", s);

  		return NULL;
  	}
  }
  
  void ceph_put_mds_session(struct ceph_mds_session *s)
  {
  	dout("mdsc put_session %p %d -> %d
  ", s,

  	     refcount_read(&s->s_ref), refcount_read(&s->s_ref)-1);
  	if (refcount_dec_and_test(&s->s_ref)) {

  		if (s->s_auth.authorizer)

  			ceph_auth_destroy_authorizer(s->s_auth.authorizer);

  		WARN_ON(mutex_is_locked(&s->s_mutex));

  		xa_destroy(&s->s_delegated_inos);

  		kfree(s);

  	}

  }
  
  /*
   * called under mdsc->mutex
   */
  struct ceph_mds_session *__ceph_lookup_mds_session(struct ceph_mds_client *mdsc,
  						   int mds)
  {

  	if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])

  		return NULL;

  	return ceph_get_mds_session(mdsc->sessions[mds]);

  }
  
  static bool __have_session(struct ceph_mds_client *mdsc, int mds)
  {

  	if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])

  		return false;

  	else
  		return true;

  }

  static int __verify_registered_session(struct ceph_mds_client *mdsc,
  				       struct ceph_mds_session *s)
  {
  	if (s->s_mds >= mdsc->max_sessions ||
  	    mdsc->sessions[s->s_mds] != s)
  		return -ENOENT;
  	return 0;
  }

  /*
   * create+register a new session for given mds.
   * called under mdsc->mutex.
   */
  static struct ceph_mds_session *register_session(struct ceph_mds_client *mdsc,
  						 int mds)
  {
  	struct ceph_mds_session *s;

  	if (mds >= mdsc->mdsmap->possible_max_rank)

  		return ERR_PTR(-EINVAL);

  	s = kzalloc(sizeof(*s), GFP_NOFS);

  	if (!s)
  		return ERR_PTR(-ENOMEM);

  
  	if (mds >= mdsc->max_sessions) {
  		int newmax = 1 << get_count_order(mds + 1);
  		struct ceph_mds_session **sa;
  
  		dout("%s: realloc to %d
  ", __func__, newmax);
  		sa = kcalloc(newmax, sizeof(void *), GFP_NOFS);
  		if (!sa)
  			goto fail_realloc;
  		if (mdsc->sessions) {
  			memcpy(sa, mdsc->sessions,
  			       mdsc->max_sessions * sizeof(void *));
  			kfree(mdsc->sessions);
  		}
  		mdsc->sessions = sa;
  		mdsc->max_sessions = newmax;
  	}
  
  	dout("%s: mds%d
  ", __func__, mds);

  	s->s_mdsc = mdsc;
  	s->s_mds = mds;
  	s->s_state = CEPH_MDS_SESSION_NEW;
  	s->s_ttl = 0;
  	s->s_seq = 0;
  	mutex_init(&s->s_mutex);

  	ceph_con_init(&s->s_con, s, &mds_con_ops, &mdsc->fsc->client->msgr);

  	spin_lock_init(&s->s_gen_ttl_lock);

  	s->s_cap_gen = 1;

  	s->s_cap_ttl = jiffies - 1;

  
  	spin_lock_init(&s->s_cap_lock);

  	s->s_renew_requested = 0;
  	s->s_renew_seq = 0;
  	INIT_LIST_HEAD(&s->s_caps);
  	s->s_nr_caps = 0;

  	refcount_set(&s->s_ref, 1);

  	INIT_LIST_HEAD(&s->s_waiting);
  	INIT_LIST_HEAD(&s->s_unsafe);

  	xa_init(&s->s_delegated_inos);

  	s->s_num_cap_releases = 0;

  	s->s_cap_reconnect = 0;

  	s->s_cap_iterator = NULL;

  	INIT_LIST_HEAD(&s->s_cap_releases);

  	INIT_WORK(&s->s_cap_release_work, ceph_cap_release_work);

  	INIT_LIST_HEAD(&s->s_cap_dirty);

  	INIT_LIST_HEAD(&s->s_cap_flushing);

  	mdsc->sessions[mds] = s;

  	atomic_inc(&mdsc->num_sessions);

  	refcount_inc(&s->s_ref);  /* one ref to sessions[], one to caller */

  	ceph_con_open(&s->s_con, CEPH_ENTITY_TYPE_MDS, mds,
  		      ceph_mdsmap_get_addr(mdsc->mdsmap, mds));

  	return s;

  
  fail_realloc:
  	kfree(s);
  	return ERR_PTR(-ENOMEM);

  }
  
  /*
   * called under mdsc->mutex
   */

  static void __unregister_session(struct ceph_mds_client *mdsc,

  			       struct ceph_mds_session *s)

  {

  	dout("__unregister_session mds%d %p
  ", s->s_mds, s);
  	BUG_ON(mdsc->sessions[s->s_mds] != s);

  	mdsc->sessions[s->s_mds] = NULL;
  	ceph_con_close(&s->s_con);
  	ceph_put_mds_session(s);

  	atomic_dec(&mdsc->num_sessions);

  }
  
  /*
   * drop session refs in request.
   *
   * should be last request ref, or hold mdsc->mutex
   */
  static void put_request_session(struct ceph_mds_request *req)
  {
  	if (req->r_session) {
  		ceph_put_mds_session(req->r_session);
  		req->r_session = NULL;
  	}
  }

  void ceph_mdsc_release_request(struct kref *kref)

  {

  	struct ceph_mds_request *req = container_of(kref,
  						    struct ceph_mds_request,
  						    r_kref);

  	ceph_mdsc_release_dir_caps_no_check(req);

  	destroy_reply_info(&req->r_reply_info);

  	if (req->r_request)
  		ceph_msg_put(req->r_request);

  	if (req->r_reply)

  		ceph_msg_put(req->r_reply);

  	if (req->r_inode) {

  		ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);

  		/* avoid calling iput_final() in mds dispatch threads */
  		ceph_async_iput(req->r_inode);

  	}

  	if (req->r_parent) {

  		ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);

  		ceph_async_iput(req->r_parent);
  	}

  	ceph_async_iput(req->r_target_inode);

  	if (req->r_dentry)
  		dput(req->r_dentry);

  	if (req->r_old_dentry)
  		dput(req->r_old_dentry);
  	if (req->r_old_dentry_dir) {

  		/*
  		 * track (and drop pins for) r_old_dentry_dir
  		 * separately, since r_old_dentry's d_parent may have
  		 * changed between the dir mutex being dropped and
  		 * this request being freed.
  		 */
  		ceph_put_cap_refs(ceph_inode(req->r_old_dentry_dir),
  				  CEPH_CAP_PIN);

  		ceph_async_iput(req->r_old_dentry_dir);

  	}

  	kfree(req->r_path1);
  	kfree(req->r_path2);

  	if (req->r_pagelist)
  		ceph_pagelist_release(req->r_pagelist);

  	put_request_session(req);

  	ceph_unreserve_caps(req->r_mdsc, &req->r_caps_reservation);

  	WARN_ON_ONCE(!list_empty(&req->r_wait));

  	kmem_cache_free(ceph_mds_request_cachep, req);

  }

  DEFINE_RB_FUNCS(request, struct ceph_mds_request, r_tid, r_node)

  /*
   * lookup session, bump ref if found.
   *
   * called under mdsc->mutex.
   */

  static struct ceph_mds_request *
  lookup_get_request(struct ceph_mds_client *mdsc, u64 tid)

  {
  	struct ceph_mds_request *req;

  	req = lookup_request(&mdsc->request_tree, tid);
  	if (req)
  		ceph_mdsc_get_request(req);

  	return req;

  }
  
  /*
   * Register an in-flight request, and assign a tid.  Link to directory
   * are modifying (if any).
   *
   * Called under mdsc->mutex.
   */
  static void __register_request(struct ceph_mds_client *mdsc,
  			       struct ceph_mds_request *req,
  			       struct inode *dir)
  {

  	int ret = 0;

  	req->r_tid = ++mdsc->last_tid;

  	if (req->r_num_caps) {
  		ret = ceph_reserve_caps(mdsc, &req->r_caps_reservation,
  					req->r_num_caps);
  		if (ret < 0) {
  			pr_err("__register_request %p "
  			       "failed to reserve caps: %d
  ", req, ret);
  			/* set req->r_err to fail early from __do_request */
  			req->r_err = ret;
  			return;
  		}
  	}

  	dout("__register_request %p tid %lld
  ", req, req->r_tid);
  	ceph_mdsc_get_request(req);

  	insert_request(&mdsc->request_tree, req);

  	req->r_uid = current_fsuid();
  	req->r_gid = current_fsgid();

  	if (mdsc->oldest_tid == 0 && req->r_op != CEPH_MDS_OP_SETFILELOCK)
  		mdsc->oldest_tid = req->r_tid;

  	if (dir) {

  		struct ceph_inode_info *ci = ceph_inode(dir);

  		ihold(dir);

  		req->r_unsafe_dir = dir;

  		spin_lock(&ci->i_unsafe_lock);
  		list_add_tail(&req->r_unsafe_dir_item, &ci->i_unsafe_dirops);
  		spin_unlock(&ci->i_unsafe_lock);

  	}
  }
  
  static void __unregister_request(struct ceph_mds_client *mdsc,
  				 struct ceph_mds_request *req)
  {
  	dout("__unregister_request %p tid %lld
  ", req, req->r_tid);

  	/* Never leave an unregistered request on an unsafe list! */
  	list_del_init(&req->r_unsafe_item);

  	if (req->r_tid == mdsc->oldest_tid) {
  		struct rb_node *p = rb_next(&req->r_node);
  		mdsc->oldest_tid = 0;
  		while (p) {
  			struct ceph_mds_request *next_req =
  				rb_entry(p, struct ceph_mds_request, r_node);
  			if (next_req->r_op != CEPH_MDS_OP_SETFILELOCK) {
  				mdsc->oldest_tid = next_req->r_tid;
  				break;
  			}
  			p = rb_next(p);
  		}
  	}

  	erase_request(&mdsc->request_tree, req);

  	if (req->r_unsafe_dir) {

  		struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);

  		spin_lock(&ci->i_unsafe_lock);
  		list_del_init(&req->r_unsafe_dir_item);
  		spin_unlock(&ci->i_unsafe_lock);

  	}

  	if (req->r_target_inode &&
  	    test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {

  		struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
  		spin_lock(&ci->i_unsafe_lock);
  		list_del_init(&req->r_unsafe_target_item);
  		spin_unlock(&ci->i_unsafe_lock);
  	}

  	if (req->r_unsafe_dir) {

  		/* avoid calling iput_final() in mds dispatch threads */
  		ceph_async_iput(req->r_unsafe_dir);

  		req->r_unsafe_dir = NULL;

  	}

  	complete_all(&req->r_safe_completion);

  	ceph_mdsc_put_request(req);

  }
  
  /*

   * Walk back up the dentry tree until we hit a dentry representing a
   * non-snapshot inode. We do this using the rcu_read_lock (which must be held
   * when calling this) to ensure that the objects won't disappear while we're
   * working with them. Once we hit a candidate dentry, we attempt to take a
   * reference to it, and return that as the result.
   */

  static struct inode *get_nonsnap_parent(struct dentry *dentry)
  {
  	struct inode *inode = NULL;

  
  	while (dentry && !IS_ROOT(dentry)) {
  		inode = d_inode_rcu(dentry);
  		if (!inode || ceph_snap(inode) == CEPH_NOSNAP)
  			break;
  		dentry = dentry->d_parent;
  	}
  	if (inode)
  		inode = igrab(inode);
  	return inode;
  }
  
  /*

   * Choose mds to send request to next.  If there is a hint set in the
   * request (e.g., due to a prior forward hint from the mds), use that.
   * Otherwise, consult frag tree and/or caps to identify the
   * appropriate mds.  If all else fails, choose randomly.
   *
   * Called under mdsc->mutex.
   */
  static int __choose_mds(struct ceph_mds_client *mdsc,

  			struct ceph_mds_request *req,
  			bool *random)

  {
  	struct inode *inode;
  	struct ceph_inode_info *ci;
  	struct ceph_cap *cap;
  	int mode = req->r_direct_mode;
  	int mds = -1;
  	u32 hash = req->r_direct_hash;

  	bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);

  	if (random)
  		*random = false;

  	/*
  	 * is there a specific mds we should try?  ignore hint if we have
  	 * no session and the mds is not up (active or recovering).
  	 */
  	if (req->r_resend_mds >= 0 &&
  	    (__have_session(mdsc, req->r_resend_mds) ||
  	     ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) {

  		dout("%s using resend_mds mds%d
  ", __func__,

  		     req->r_resend_mds);
  		return req->r_resend_mds;
  	}
  
  	if (mode == USE_RANDOM_MDS)
  		goto random;
  
  	inode = NULL;
  	if (req->r_inode) {

  		if (ceph_snap(req->r_inode) != CEPH_SNAPDIR) {
  			inode = req->r_inode;
  			ihold(inode);
  		} else {

  			/* req->r_dentry is non-null for LSSNAP request */
  			rcu_read_lock();
  			inode = get_nonsnap_parent(req->r_dentry);
  			rcu_read_unlock();

  			dout("%s using snapdir's parent %p
  ", __func__, inode);

  		}

  	} else if (req->r_dentry) {

  		/* ignore race with rename; old or new d_parent is okay */

  		struct dentry *parent;
  		struct inode *dir;
  
  		rcu_read_lock();

  		parent = READ_ONCE(req->r_dentry->d_parent);

  		dir = req->r_parent ? : d_inode_rcu(parent);

  		if (!dir || dir->i_sb != mdsc->fsc->sb) {
  			/*  not this fs or parent went negative */

  			inode = d_inode(req->r_dentry);

  			if (inode)
  				ihold(inode);

  		} else if (ceph_snap(dir) != CEPH_NOSNAP) {
  			/* direct snapped/virtual snapdir requests
  			 * based on parent dir inode */

  			inode = get_nonsnap_parent(parent);

  			dout("%s using nonsnap parent %p
  ", __func__, inode);

  		} else {

  			/* dentry target */

  			inode = d_inode(req->r_dentry);

  			if (!inode || mode == USE_AUTH_MDS) {
  				/* dir + name */

  				inode = igrab(dir);

  				hash = ceph_dentry_hash(dir, req->r_dentry);
  				is_hash = true;

  			} else {
  				ihold(inode);

  			}

  		}

  		rcu_read_unlock();

  	}

  	dout("%s %p is_hash=%d (0x%x) mode %d
  ", __func__, inode, (int)is_hash,
  	     hash, mode);

  	if (!inode)
  		goto random;
  	ci = ceph_inode(inode);
  
  	if (is_hash && S_ISDIR(inode->i_mode)) {
  		struct ceph_inode_frag frag;
  		int found;
  
  		ceph_choose_frag(ci, hash, &frag, &found);
  		if (found) {
  			if (mode == USE_ANY_MDS && frag.ndist > 0) {
  				u8 r;
  
  				/* choose a random replica */
  				get_random_bytes(&r, 1);
  				r %= frag.ndist;
  				mds = frag.dist[r];

  				dout("%s %p %llx.%llx frag %u mds%d (%d/%d)
  ",
  				     __func__, inode, ceph_vinop(inode),
  				     frag.frag, mds, (int)r, frag.ndist);

  				if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=

  				    CEPH_MDS_STATE_ACTIVE &&
  				    !ceph_mdsmap_is_laggy(mdsc->mdsmap, mds))

  					goto out;

  			}
  
  			/* since this file/dir wasn't known to be
  			 * replicated, then we want to look for the
  			 * authoritative mds. */

  			if (frag.mds >= 0) {
  				/* choose auth mds */
  				mds = frag.mds;

  				dout("%s %p %llx.%llx frag %u mds%d (auth)
  ",
  				     __func__, inode, ceph_vinop(inode),
  				     frag.frag, mds);

  				if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=

  				    CEPH_MDS_STATE_ACTIVE) {

  					if (!ceph_mdsmap_is_laggy(mdsc->mdsmap,

  								  mds))
  						goto out;
  				}

  			}

  			mode = USE_AUTH_MDS;

  		}
  	}

  	spin_lock(&ci->i_ceph_lock);

  	cap = NULL;
  	if (mode == USE_AUTH_MDS)
  		cap = ci->i_auth_cap;
  	if (!cap && !RB_EMPTY_ROOT(&ci->i_caps))
  		cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node);
  	if (!cap) {

  		spin_unlock(&ci->i_ceph_lock);

  		ceph_async_iput(inode);

  		goto random;
  	}
  	mds = cap->session->s_mds;

  	dout("%s %p %llx.%llx mds%d (%scap %p)
  ", __func__,

  	     inode, ceph_vinop(inode), mds,
  	     cap == ci->i_auth_cap ? "auth " : "", cap);

  	spin_unlock(&ci->i_ceph_lock);

  out:

  	/* avoid calling iput_final() while holding mdsc->mutex or
  	 * in mds dispatch threads */
  	ceph_async_iput(inode);

  	return mds;
  
  random:

  	if (random)
  		*random = true;

  	mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap);

  	dout("%s chose random mds%d
  ", __func__, mds);

  	return mds;
  }
  
  
  /*
   * session messages
   */
  static struct ceph_msg *create_session_msg(u32 op, u64 seq)
  {
  	struct ceph_msg *msg;
  	struct ceph_mds_session_head *h;

  	msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS,
  			   false);

  	if (!msg) {

  		pr_err("create_session_msg ENOMEM creating msg
  ");

  		return NULL;

  	}
  	h = msg->front.iov_base;
  	h->op = cpu_to_le32(op);
  	h->seq = cpu_to_le64(seq);

  
  	return msg;
  }

  static const unsigned char feature_bits[] = CEPHFS_FEATURES_CLIENT_SUPPORTED;
  #define FEATURE_BYTES(c) (DIV_ROUND_UP((size_t)feature_bits[c - 1] + 1, 64) * 8)

  static int encode_supported_features(void **p, void *end)

  {

  	static const size_t count = ARRAY_SIZE(feature_bits);

  
  	if (count > 0) {
  		size_t i;

  		size_t size = FEATURE_BYTES(count);

  		if (WARN_ON_ONCE(*p + 4 + size > end))
  			return -ERANGE;

  		ceph_encode_32(p, size);
  		memset(*p, 0, size);
  		for (i = 0; i < count; i++)

  			((unsigned char*)(*p))[i / 8] |= BIT(feature_bits[i] % 8);

  		*p += size;
  	} else {

  		if (WARN_ON_ONCE(*p + 4 > end))
  			return -ERANGE;

  		ceph_encode_32(p, 0);
  	}

  
  	return 0;

  }

  static const unsigned char metric_bits[] = CEPHFS_METRIC_SPEC_CLIENT_SUPPORTED;
  #define METRIC_BYTES(cnt) (DIV_ROUND_UP((size_t)metric_bits[cnt - 1] + 1, 64) * 8)
  static int encode_metric_spec(void **p, void *end)
  {
  	static const size_t count = ARRAY_SIZE(metric_bits);
  
  	/* header */
  	if (WARN_ON_ONCE(*p + 2 > end))
  		return -ERANGE;
  
  	ceph_encode_8(p, 1); /* version */
  	ceph_encode_8(p, 1); /* compat */
  
  	if (count > 0) {
  		size_t i;
  		size_t size = METRIC_BYTES(count);
  
  		if (WARN_ON_ONCE(*p + 4 + 4 + size > end))
  			return -ERANGE;
  
  		/* metric spec info length */
  		ceph_encode_32(p, 4 + size);
  
  		/* metric spec */
  		ceph_encode_32(p, size);
  		memset(*p, 0, size);
  		for (i = 0; i < count; i++)
  			((unsigned char *)(*p))[i / 8] |= BIT(metric_bits[i] % 8);
  		*p += size;
  	} else {
  		if (WARN_ON_ONCE(*p + 4 + 4 > end))
  			return -ERANGE;
  
  		/* metric spec info length */
  		ceph_encode_32(p, 4);
  		/* metric spec */
  		ceph_encode_32(p, 0);
  	}
  
  	return 0;
  }

  /*
   * session message, specialization for CEPH_SESSION_REQUEST_OPEN
   * to include additional client metadata fields.
   */
  static struct ceph_msg *create_session_open_msg(struct ceph_mds_client *mdsc, u64 seq)
  {
  	struct ceph_msg *msg;
  	struct ceph_mds_session_head *h;
  	int i = -1;

  	int extra_bytes = 0;

  	int metadata_key_count = 0;
  	struct ceph_options *opt = mdsc->fsc->client->options;

  	struct ceph_mount_options *fsopt = mdsc->fsc->mount_options;

  	size_t size, count;

  	void *p, *end;

  	int ret;

  	const char* metadata[][2] = {

  		{"hostname", mdsc->nodename},
  		{"kernel_version", init_utsname()->release},

  		{"entity_id", opt->name ? : ""},
  		{"root", fsopt->server_path ? : "/"},

  		{NULL, NULL}
  	};
  
  	/* Calculate serialized length of metadata */

  	extra_bytes = 4;  /* map length */

  	for (i = 0; metadata[i][0]; ++i) {

  		extra_bytes += 8 + strlen(metadata[i][0]) +

  			strlen(metadata[i][1]);
  		metadata_key_count++;
  	}

  	/* supported feature */

  	size = 0;
  	count = ARRAY_SIZE(feature_bits);
  	if (count > 0)
  		size = FEATURE_BYTES(count);
  	extra_bytes += 4 + size;

  	/* metric spec */
  	size = 0;
  	count = ARRAY_SIZE(metric_bits);
  	if (count > 0)
  		size = METRIC_BYTES(count);
  	extra_bytes += 2 + 4 + 4 + size;

  	/* Allocate the message */

  	msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h) + extra_bytes,

  			   GFP_NOFS, false);
  	if (!msg) {
  		pr_err("create_session_msg ENOMEM creating msg
  ");

  		return ERR_PTR(-ENOMEM);

  	}

  	p = msg->front.iov_base;
  	end = p + msg->front.iov_len;
  
  	h = p;

  	h->op = cpu_to_le32(CEPH_SESSION_REQUEST_OPEN);
  	h->seq = cpu_to_le64(seq);
  
  	/*
  	 * Serialize client metadata into waiting buffer space, using
  	 * the format that userspace expects for map<string, string>

  	 *

  	 * ClientSession messages with metadata are v4

  	 */

  	msg->hdr.version = cpu_to_le16(4);

  	msg->hdr.compat_version = cpu_to_le16(1);

  
  	/* The write pointer, following the session_head structure */

  	p += sizeof(*h);

  
  	/* Number of entries in the map */
  	ceph_encode_32(&p, metadata_key_count);
  
  	/* Two length-prefixed strings for each entry in the map */

  	for (i = 0; metadata[i][0]; ++i) {

  		size_t const key_len = strlen(metadata[i][0]);
  		size_t const val_len = strlen(metadata[i][1]);
  
  		ceph_encode_32(&p, key_len);
  		memcpy(p, metadata[i][0], key_len);
  		p += key_len;
  		ceph_encode_32(&p, val_len);
  		memcpy(p, metadata[i][1], val_len);
  		p += val_len;
  	}

  	ret = encode_supported_features(&p, end);
  	if (ret) {
  		pr_err("encode_supported_features failed!
  ");
  		ceph_msg_put(msg);
  		return ERR_PTR(ret);
  	}

  	ret = encode_metric_spec(&p, end);
  	if (ret) {
  		pr_err("encode_metric_spec failed!
  ");
  		ceph_msg_put(msg);
  		return ERR_PTR(ret);
  	}

  	msg->front.iov_len = p - msg->front.iov_base;
  	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);

  	return msg;
  }
  
  /*
   * send session open request.
   *
   * called under mdsc->mutex
   */
  static int __open_session(struct ceph_mds_client *mdsc,
  			  struct ceph_mds_session *session)
  {
  	struct ceph_msg *msg;
  	int mstate;
  	int mds = session->s_mds;

  
  	/* wait for mds to go active? */
  	mstate = ceph_mdsmap_get_state(mdsc->mdsmap, mds);
  	dout("open_session to mds%d (%s)
  ", mds,
  	     ceph_mds_state_name(mstate));
  	session->s_state = CEPH_MDS_SESSION_OPENING;
  	session->s_renew_requested = jiffies;
  
  	/* send connect message */

  	msg = create_session_open_msg(mdsc, session->s_seq);

  	if (IS_ERR(msg))
  		return PTR_ERR(msg);

  	ceph_con_send(&session->s_con, msg);

  	return 0;
  }
  
  /*

   * open sessions for any export targets for the given mds
   *
   * called under mdsc->mutex
   */

  static struct ceph_mds_session *
  __open_export_target_session(struct ceph_mds_client *mdsc, int target)
  {
  	struct ceph_mds_session *session;

  	int ret;

  
  	session = __ceph_lookup_mds_session(mdsc, target);
  	if (!session) {
  		session = register_session(mdsc, target);
  		if (IS_ERR(session))
  			return session;
  	}
  	if (session->s_state == CEPH_MDS_SESSION_NEW ||

  	    session->s_state == CEPH_MDS_SESSION_CLOSING) {
  		ret = __open_session(mdsc, session);
  		if (ret)
  			return ERR_PTR(ret);
  	}

  
  	return session;
  }
  
  struct ceph_mds_session *
  ceph_mdsc_open_export_target_session(struct ceph_mds_client *mdsc, int target)
  {
  	struct ceph_mds_session *session;
  
  	dout("open_export_target_session to mds%d
  ", target);
  
  	mutex_lock(&mdsc->mutex);
  	session = __open_export_target_session(mdsc, target);
  	mutex_unlock(&mdsc->mutex);
  
  	return session;
  }

  static void __open_export_target_sessions(struct ceph_mds_client *mdsc,
  					  struct ceph_mds_session *session)
  {
  	struct ceph_mds_info *mi;
  	struct ceph_mds_session *ts;
  	int i, mds = session->s_mds;

  	if (mds >= mdsc->mdsmap->possible_max_rank)

  		return;

  	mi = &mdsc->mdsmap->m_info[mds];
  	dout("open_export_target_sessions for mds%d (%d targets)
  ",
  	     session->s_mds, mi->num_export_targets);
  
  	for (i = 0; i < mi->num_export_targets; i++) {

  		ts = __open_export_target_session(mdsc, mi->export_targets[i]);
  		if (!IS_ERR(ts))
  			ceph_put_mds_session(ts);

  	}
  }

  void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,
  					   struct ceph_mds_session *session)
  {
  	mutex_lock(&mdsc->mutex);
  	__open_export_target_sessions(mdsc, session);
  	mutex_unlock(&mdsc->mutex);
  }

  /*

   * session caps
   */

  static void detach_cap_releases(struct ceph_mds_session *session,
  				struct list_head *target)

  {

  	lockdep_assert_held(&session->s_cap_lock);
  
  	list_splice_init(&session->s_cap_releases, target);

  	session->s_num_cap_releases = 0;

  	dout("dispose_cap_releases mds%d
  ", session->s_mds);
  }

  static void dispose_cap_releases(struct ceph_mds_client *mdsc,
  				 struct list_head *dispose)
  {
  	while (!list_empty(dispose)) {

  		struct ceph_cap *cap;
  		/* zero out the in-progress message */

  		cap = list_first_entry(dispose, struct ceph_cap, session_caps);

  		list_del(&cap->session_caps);
  		ceph_put_cap(mdsc, cap);

  	}

  }

  static void cleanup_session_requests(struct ceph_mds_client *mdsc,
  				     struct ceph_mds_session *session)
  {
  	struct ceph_mds_request *req;
  	struct rb_node *p;

  	struct ceph_inode_info *ci;

  
  	dout("cleanup_session_requests mds%d
  ", session->s_mds);
  	mutex_lock(&mdsc->mutex);
  	while (!list_empty(&session->s_unsafe)) {
  		req = list_first_entry(&session->s_unsafe,
  				       struct ceph_mds_request, r_unsafe_item);

  		pr_warn_ratelimited(" dropping unsafe request %llu
  ",
  				    req->r_tid);

  		if (req->r_target_inode) {
  			/* dropping unsafe change of inode's attributes */
  			ci = ceph_inode(req->r_target_inode);
  			errseq_set(&ci->i_meta_err, -EIO);
  		}
  		if (req->r_unsafe_dir) {
  			/* dropping unsafe directory operation */
  			ci = ceph_inode(req->r_unsafe_dir);
  			errseq_set(&ci->i_meta_err, -EIO);
  		}

  		__unregister_request(mdsc, req);
  	}
  	/* zero r_attempts, so kick_requests() will re-send requests */
  	p = rb_first(&mdsc->request_tree);
  	while (p) {
  		req = rb_entry(p, struct ceph_mds_request, r_node);
  		p = rb_next(p);
  		if (req->r_session &&
  		    req->r_session->s_mds == session->s_mds)
  			req->r_attempts = 0;
  	}
  	mutex_unlock(&mdsc->mutex);
  }

  /*

   * Helper to safely iterate over all caps associated with a session, with
   * special care taken to handle a racing __ceph_remove_cap().

   *

   * Caller must hold session s_mutex.

   */

  int ceph_iterate_session_caps(struct ceph_mds_session *session,
  			      int (*cb)(struct inode *, struct ceph_cap *,
  					void *), void *arg)

  {

  	struct list_head *p;
  	struct ceph_cap *cap;
  	struct inode *inode, *last_inode = NULL;
  	struct ceph_cap *old_cap = NULL;

  	int ret;
  
  	dout("iterate_session_caps %p mds%d
  ", session, session->s_mds);
  	spin_lock(&session->s_cap_lock);

  	p = session->s_caps.next;
  	while (p != &session->s_caps) {
  		cap = list_entry(p, struct ceph_cap, session_caps);

  		inode = igrab(&cap->ci->vfs_inode);

  		if (!inode) {
  			p = p->next;

  			continue;

  		}
  		session->s_cap_iterator = cap;

  		spin_unlock(&session->s_cap_lock);

  
  		if (last_inode) {

  			/* avoid calling iput_final() while holding
  			 * s_mutex or in mds dispatch threads */
  			ceph_async_iput(last_inode);

  			last_inode = NULL;
  		}
  		if (old_cap) {

  			ceph_put_cap(session->s_mdsc, old_cap);

  			old_cap = NULL;
  		}

  		ret = cb(inode, cap, arg);

  		last_inode = inode;

  		spin_lock(&session->s_cap_lock);

  		p = p->next;

  		if (!cap->ci) {

  			dout("iterate_session_caps  finishing cap %p removal
  ",
  			     cap);
  			BUG_ON(cap->session != session);

  			cap->session = NULL;

  			list_del_init(&cap->session_caps);
  			session->s_nr_caps--;

  			atomic64_dec(&session->s_mdsc->metric.total_caps);

  			if (cap->queue_release)
  				__ceph_queue_cap_release(session, cap);
  			else

  				old_cap = cap;  /* put_cap it w/o locks held */

  		}

  		if (ret < 0)
  			goto out;

  	}

  	ret = 0;
  out:

  	session->s_cap_iterator = NULL;

  	spin_unlock(&session->s_cap_lock);

  	ceph_async_iput(last_inode);

  	if (old_cap)

  		ceph_put_cap(session->s_mdsc, old_cap);

  	return ret;

  }
  
  static int remove_session_caps_cb(struct inode *inode, struct ceph_cap *cap,

  				  void *arg)

  {

  	struct ceph_fs_client *fsc = (struct ceph_fs_client *)arg;

  	struct ceph_inode_info *ci = ceph_inode(inode);

  	LIST_HEAD(to_remove);

  	bool dirty_dropped = false;

  	bool invalidate = false;

  	dout("removing cap %p, ci is %p, inode is %p
  ",
  	     cap, ci, &ci->vfs_inode);

  	spin_lock(&ci->i_ceph_lock);

  	__ceph_remove_cap(cap, false);

  	if (!ci->i_auth_cap) {

  		struct ceph_cap_flush *cf;

  		struct ceph_mds_client *mdsc = fsc->mdsc;

  		if (READ_ONCE(fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
  			if (inode->i_data.nrpages > 0)
  				invalidate = true;
  			if (ci->i_wrbuffer_ref > 0)
  				mapping_set_error(&inode->i_data, -EIO);
  		}

  		while (!list_empty(&ci->i_cap_flush_list)) {
  			cf = list_first_entry(&ci->i_cap_flush_list,
  					      struct ceph_cap_flush, i_list);

  			list_move(&cf->i_list, &to_remove);

  		}

  		spin_lock(&mdsc->cap_dirty_lock);

  		list_for_each_entry(cf, &to_remove, i_list)
  			list_del(&cf->g_list);

  		if (!list_empty(&ci->i_dirty_item)) {

  			pr_warn_ratelimited(
  				" dropping dirty %s state for %p %lld
  ",

  				ceph_cap_string(ci->i_dirty_caps),
  				inode, ceph_ino(inode));
  			ci->i_dirty_caps = 0;
  			list_del_init(&ci->i_dirty_item);

  			dirty_dropped = true;

  		}
  		if (!list_empty(&ci->i_flushing_item)) {

  			pr_warn_ratelimited(
  				" dropping dirty+flushing %s state for %p %lld
  ",

  				ceph_cap_string(ci->i_flushing_caps),
  				inode, ceph_ino(inode));
  			ci->i_flushing_caps = 0;
  			list_del_init(&ci->i_flushing_item);
  			mdsc->num_cap_flushing--;

  			dirty_dropped = true;

  		}

  		spin_unlock(&mdsc->cap_dirty_lock);

  		if (dirty_dropped) {
  			errseq_set(&ci->i_meta_err, -EIO);
  
  			if (ci->i_wrbuffer_ref_head == 0 &&
  			    ci->i_wr_ref == 0 &&
  			    ci->i_dirty_caps == 0 &&
  			    ci->i_flushing_caps == 0) {
  				ceph_put_snap_context(ci->i_head_snapc);
  				ci->i_head_snapc = NULL;
  			}
  		}

  		if (atomic_read(&ci->i_filelock_ref) > 0) {
  			/* make further file lock syscall return -EIO */
  			ci->i_ceph_flags |= CEPH_I_ERROR_FILELOCK;
  			pr_warn_ratelimited(" dropping file locks for %p %lld
  ",
  					    inode, ceph_ino(inode));
  		}

  		if (!ci->i_dirty_caps && ci->i_prealloc_cap_flush) {

  			list_add(&ci->i_prealloc_cap_flush->i_list, &to_remove);

  			ci->i_prealloc_cap_flush = NULL;
  		}

  	}

  	spin_unlock(&ci->i_ceph_lock);

  	while (!list_empty(&to_remove)) {
  		struct ceph_cap_flush *cf;
  		cf = list_first_entry(&to_remove,

  				      struct ceph_cap_flush, i_list);
  		list_del(&cf->i_list);

  		ceph_free_cap_flush(cf);

  	}

  
  	wake_up_all(&ci->i_cap_wq);

  	if (invalidate)
  		ceph_queue_invalidate(inode);

  	if (dirty_dropped)

  		iput(inode);

  	return 0;
  }
  
  /*
   * caller must hold session s_mutex
   */
  static void remove_session_caps(struct ceph_mds_session *session)
  {

  	struct ceph_fs_client *fsc = session->s_mdsc->fsc;
  	struct super_block *sb = fsc->sb;

  	LIST_HEAD(dispose);

  	dout("remove_session_caps on %p
  ", session);

  	ceph_iterate_session_caps(session, remove_session_caps_cb, fsc);

  	wake_up_all(&fsc->mdsc->cap_flushing_wq);

  	spin_lock(&session->s_cap_lock);
  	if (session->s_nr_caps > 0) {

  		struct inode *inode;
  		struct ceph_cap *cap, *prev = NULL;
  		struct ceph_vino vino;
  		/*
  		 * iterate_session_caps() skips inodes that are being
  		 * deleted, we need to wait until deletions are complete.
  		 * __wait_on_freeing_inode() is designed for the job,
  		 * but it is not exported, so use lookup inode function
  		 * to access it.
  		 */
  		while (!list_empty(&session->s_caps)) {
  			cap = list_entry(session->s_caps.next,
  					 struct ceph_cap, session_caps);
  			if (cap == prev)
  				break;
  			prev = cap;
  			vino = cap->ci->i_vino;
  			spin_unlock(&session->s_cap_lock);

  			inode = ceph_find_inode(sb, vino);

  			 /* avoid calling iput_final() while holding s_mutex */
  			ceph_async_iput(inode);

  
  			spin_lock(&session->s_cap_lock);
  		}
  	}

  
  	// drop cap expires and unlock s_cap_lock

  	detach_cap_releases(session, &dispose);

  	BUG_ON(session->s_nr_caps > 0);

  	BUG_ON(!list_empty(&session->s_cap_flushing));

  	spin_unlock(&session->s_cap_lock);
  	dispose_cap_releases(session->s_mdsc, &dispose);

  }

  enum {
  	RECONNECT,
  	RENEWCAPS,
  	FORCE_RO,
  };

  /*
   * wake up any threads waiting on this session's caps.  if the cap is
   * old (didn't get renewed on the client reconnect), remove it now.
   *
   * caller must hold s_mutex.
   */
  static int wake_up_session_cb(struct inode *inode, struct ceph_cap *cap,
  			      void *arg)
  {

  	struct ceph_inode_info *ci = ceph_inode(inode);

  	unsigned long ev = (unsigned long)arg;

  	if (ev == RECONNECT) {

  		spin_lock(&ci->i_ceph_lock);

  		ci->i_wanted_max_size = 0;
  		ci->i_requested_max_size = 0;

  		spin_unlock(&ci->i_ceph_lock);

  	} else if (ev == RENEWCAPS) {
  		if (cap->cap_gen < cap->session->s_cap_gen) {
  			/* mds did not re-issue stale cap */
  			spin_lock(&ci->i_ceph_lock);
  			cap->issued = cap->implemented = CEPH_CAP_PIN;

  			spin_unlock(&ci->i_ceph_lock);
  		}
  	} else if (ev == FORCE_RO) {

  	}

  	wake_up_all(&ci->i_cap_wq);

  	return 0;
  }

  static void wake_up_session_caps(struct ceph_mds_session *session, int ev)

  {
  	dout("wake_up_session_caps %p mds%d
  ", session, session->s_mds);

  	ceph_iterate_session_caps(session, wake_up_session_cb,
  				  (void *)(unsigned long)ev);

  }
  
  /*
   * Send periodic message to MDS renewing all currently held caps.  The
   * ack will reset the expiration for all caps from this session.
   *
   * caller holds s_mutex
   */
  static int send_renew_caps(struct ceph_mds_client *mdsc,
  			   struct ceph_mds_session *session)
  {
  	struct ceph_msg *msg;
  	int state;
  
  	if (time_after_eq(jiffies, session->s_cap_ttl) &&
  	    time_after_eq(session->s_cap_ttl, session->s_renew_requested))
  		pr_info("mds%d caps stale
  ", session->s_mds);

  	session->s_renew_requested = jiffies;

  
  	/* do not try to renew caps until a recovering mds has reconnected
  	 * with its clients. */
  	state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds);
  	if (state < CEPH_MDS_STATE_RECONNECT) {
  		dout("send_renew_caps ignoring mds%d (%s)
  ",
  		     session->s_mds, ceph_mds_state_name(state));
  		return 0;
  	}
  
  	dout("send_renew_caps to mds%d (%s)
  ", session->s_mds,
  		ceph_mds_state_name(state));

  	msg = create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS,
  				 ++session->s_renew_seq);

  	if (!msg)
  		return -ENOMEM;

  	ceph_con_send(&session->s_con, msg);
  	return 0;
  }

  static int send_flushmsg_ack(struct ceph_mds_client *mdsc,
  			     struct ceph_mds_session *session, u64 seq)
  {
  	struct ceph_msg *msg;
  
  	dout("send_flushmsg_ack to mds%d (%s)s seq %lld
  ",

  	     session->s_mds, ceph_session_state_name(session->s_state), seq);

  	msg = create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq);
  	if (!msg)
  		return -ENOMEM;
  	ceph_con_send(&session->s_con, msg);
  	return 0;
  }

  /*
   * Note new cap ttl, and any transition from stale -> not stale (fresh?).

   *
   * Called under session->s_mutex

   */
  static void renewed_caps(struct ceph_mds_client *mdsc,
  			 struct ceph_mds_session *session, int is_renew)
  {
  	int was_stale;
  	int wake = 0;
  
  	spin_lock(&session->s_cap_lock);

  	was_stale = is_renew && time_after_eq(jiffies, session->s_cap_ttl);

  
  	session->s_cap_ttl = session->s_renew_requested +
  		mdsc->mdsmap->m_session_timeout*HZ;
  
  	if (was_stale) {
  		if (time_before(jiffies, session->s_cap_ttl)) {
  			pr_info("mds%d caps renewed
  ", session->s_mds);
  			wake = 1;
  		} else {
  			pr_info("mds%d caps still stale
  ", session->s_mds);
  		}
  	}
  	dout("renewed_caps mds%d ttl now %lu, was %s, now %s
  ",
  	     session->s_mds, session->s_cap_ttl, was_stale ? "stale" : "fresh",
  	     time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh");
  	spin_unlock(&session->s_cap_lock);
  
  	if (wake)

  		wake_up_session_caps(session, RENEWCAPS);

  }
  
  /*
   * send a session close request
   */

  static int request_close_session(struct ceph_mds_session *session)

  {
  	struct ceph_msg *msg;

  
  	dout("request_close_session mds%d state %s seq %lld
  ",

  	     session->s_mds, ceph_session_state_name(session->s_state),

  	     session->s_seq);
  	msg = create_session_msg(CEPH_SESSION_REQUEST_CLOSE, session->s_seq);

  	if (!msg)
  		return -ENOMEM;
  	ceph_con_send(&session->s_con, msg);

  	return 1;

  }
  
  /*
   * Called with s_mutex held.
   */
  static int __close_session(struct ceph_mds_client *mdsc,
  			 struct ceph_mds_session *session)
  {
  	if (session->s_state >= CEPH_MDS_SESSION_CLOSING)
  		return 0;
  	session->s_state = CEPH_MDS_SESSION_CLOSING;

  	return request_close_session(session);

  }

  static bool drop_negative_children(struct dentry *dentry)
  {
  	struct dentry *child;
  	bool all_negative = true;
  
  	if (!d_is_dir(dentry))
  		goto out;
  
  	spin_lock(&dentry->d_lock);
  	list_for_each_entry(child, &dentry->d_subdirs, d_child) {
  		if (d_really_is_positive(child)) {
  			all_negative = false;
  			break;
  		}
  	}
  	spin_unlock(&dentry->d_lock);
  
  	if (all_negative)
  		shrink_dcache_parent(dentry);
  out:
  	return all_negative;
  }

  /*
   * Trim old(er) caps.
   *
   * Because we can't cache an inode without one or more caps, we do
   * this indirectly: if a cap is unused, we prune its aliases, at which
   * point the inode will hopefully get dropped to.
   *
   * Yes, this is a bit sloppy.  Our only real goal here is to respond to
   * memory pressure from the MDS, though, so it needn't be perfect.
   */
  static int trim_caps_cb(struct inode *inode, struct ceph_cap *cap, void *arg)
  {

  	int *remaining = arg;

  	struct ceph_inode_info *ci = ceph_inode(inode);

  	int used, wanted, oissued, mine;

  	if (*remaining <= 0)

  		return -1;

  	spin_lock(&ci->i_ceph_lock);

  	mine = cap->issued | cap->implemented;
  	used = __ceph_caps_used(ci);

  	wanted = __ceph_caps_file_wanted(ci);

  	oissued = __ceph_caps_issued_other(ci, cap);

  	dout("trim_caps_cb %p cap %p mine %s oissued %s used %s wanted %s
  ",

  	     inode, cap, ceph_cap_string(mine), ceph_cap_string(oissued),

  	     ceph_cap_string(used), ceph_cap_string(wanted));
  	if (cap == ci->i_auth_cap) {

  		if (ci->i_dirty_caps || ci->i_flushing_caps ||
  		    !list_empty(&ci->i_cap_snaps))

  			goto out;
  		if ((used | wanted) & CEPH_CAP_ANY_WR)
  			goto out;