/*

  *  Copyright (c) 2001 The Regents of the University of Michigan.
  *  All rights reserved.
  *
  *  Kendrick Smith <kmsmith@umich.edu>
  *  Andy Adamson <kandros@umich.edu>
  *
  *  Redistribution and use in source and binary forms, with or without
  *  modification, are permitted provided that the following conditions
  *  are met:
  *
  *  1. Redistributions of source code must retain the above copyright
  *     notice, this list of conditions and the following disclaimer.
  *  2. Redistributions in binary form must reproduce the above copyright
  *     notice, this list of conditions and the following disclaimer in the
  *     documentation and/or other materials provided with the distribution.
  *  3. Neither the name of the University nor the names of its
  *     contributors may be used to endorse or promote products derived
  *     from this software without specific prior written permission.
  *
  *  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESS OR IMPLIED
  *  WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
  *  MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  *  DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
  *  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  *  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  *  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
  *  BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
  *  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
  *  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  *  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  *
  */

  #include <linux/file.h>

  #include <linux/fs.h>

  #include <linux/slab.h>

  #include <linux/namei.h>

  #include <linux/swap.h>

  #include <linux/pagemap.h>

  #include <linux/sunrpc/svcauth_gss.h>

  #include <linux/sunrpc/clnt.h>

  #include "xdr4.h"

  #include "vfs.h"

  
  #define NFSDDBG_FACILITY                NFSDDBG_PROC
  
  /* Globals */

  time_t nfsd4_lease = 90;     /* default lease time */

  time_t nfsd4_grace = 90;

  static time_t boot_time;

  
  #define all_ones {{~0,~0},~0}
  static const stateid_t one_stateid = {
  	.si_generation = ~0,
  	.si_opaque = all_ones,
  };
  static const stateid_t zero_stateid = {
  	/* all fields zero */
  };

  static u64 current_sessionid = 1;

  #define ZERO_STATEID(stateid) (!memcmp((stateid), &zero_stateid, sizeof(stateid_t)))
  #define ONE_STATEID(stateid)  (!memcmp((stateid), &one_stateid, sizeof(stateid_t)))

  /* forward declarations */

  static int check_for_locks(struct nfs4_file *filp, struct nfs4_lockowner *lowner);

  /* Locking: */
  
  /* Currently used for almost all code touching nfsv4 state: */

  static DEFINE_MUTEX(client_mutex);

  /*
   * Currently used for the del_recall_lru and file hash table.  In an
   * effort to decrease the scope of the client_mutex, this spinlock may
   * eventually cover more:
   */
  static DEFINE_SPINLOCK(recall_lock);

  static struct kmem_cache *openowner_slab = NULL;
  static struct kmem_cache *lockowner_slab = NULL;

  static struct kmem_cache *file_slab = NULL;
  static struct kmem_cache *stateid_slab = NULL;
  static struct kmem_cache *deleg_slab = NULL;

  void
  nfs4_lock_state(void)
  {

  	mutex_lock(&client_mutex);

  }
  
  void
  nfs4_unlock_state(void)
  {

  	mutex_unlock(&client_mutex);

  }
  
  static inline u32
  opaque_hashval(const void *ptr, int nbytes)
  {
  	unsigned char *cptr = (unsigned char *) ptr;
  
  	u32 x = 0;
  	while (nbytes--) {
  		x *= 37;
  		x += *cptr++;
  	}
  	return x;
  }

  static struct list_head del_recall_lru;

  static void nfsd4_free_file(struct nfs4_file *f)
  {
  	kmem_cache_free(file_slab, f);
  }

  static inline void
  put_nfs4_file(struct nfs4_file *fi)
  {

  	if (atomic_dec_and_lock(&fi->fi_ref, &recall_lock)) {
  		list_del(&fi->fi_hash);
  		spin_unlock(&recall_lock);
  		iput(fi->fi_inode);

  		nfsd4_free_file(fi);

  	}

  }
  
  static inline void
  get_nfs4_file(struct nfs4_file *fi)
  {

  	atomic_inc(&fi->fi_ref);

  }

  static int num_delegations;

  unsigned int max_delegations;

  
  /*
   * Open owner state (share locks)
   */

  /* hash tables for lock and open owners */
  #define OWNER_HASH_BITS              8
  #define OWNER_HASH_SIZE             (1 << OWNER_HASH_BITS)
  #define OWNER_HASH_MASK             (OWNER_HASH_SIZE - 1)

  static unsigned int ownerstr_hashval(u32 clientid, struct xdr_netobj *ownername)

  {
  	unsigned int ret;
  
  	ret = opaque_hashval(ownername->data, ownername->len);
  	ret += clientid;

  	return ret & OWNER_HASH_MASK;

  }

  static struct list_head	ownerstr_hashtbl[OWNER_HASH_SIZE];

  
  /* hash table for nfs4_file */
  #define FILE_HASH_BITS                   8
  #define FILE_HASH_SIZE                  (1 << FILE_HASH_BITS)

  static unsigned int file_hashval(struct inode *ino)
  {
  	/* XXX: why are we hashing on inode pointer, anyway? */
  	return hash_ptr(ino, FILE_HASH_BITS);
  }

  static struct list_head file_hashtbl[FILE_HASH_SIZE];

  static void __nfs4_file_get_access(struct nfs4_file *fp, int oflag)

  {
  	BUG_ON(!(fp->fi_fds[oflag] || fp->fi_fds[O_RDWR]));
  	atomic_inc(&fp->fi_access[oflag]);
  }

  static void nfs4_file_get_access(struct nfs4_file *fp, int oflag)
  {
  	if (oflag == O_RDWR) {
  		__nfs4_file_get_access(fp, O_RDONLY);
  		__nfs4_file_get_access(fp, O_WRONLY);
  	} else
  		__nfs4_file_get_access(fp, oflag);
  }
  
  static void nfs4_file_put_fd(struct nfs4_file *fp, int oflag)

  {
  	if (fp->fi_fds[oflag]) {
  		fput(fp->fi_fds[oflag]);
  		fp->fi_fds[oflag] = NULL;
  	}
  }

  static void __nfs4_file_put_access(struct nfs4_file *fp, int oflag)

  {
  	if (atomic_dec_and_test(&fp->fi_access[oflag])) {

  		nfs4_file_put_fd(fp, oflag);

  		/*
  		 * It's also safe to get rid of the RDWR open *if*
  		 * we no longer have need of the other kind of access
  		 * or if we already have the other kind of open:
  		 */
  		if (fp->fi_fds[1-oflag]
  			|| atomic_read(&fp->fi_access[1 - oflag]) == 0)
  			nfs4_file_put_fd(fp, O_RDWR);

  	}
  }

  static void nfs4_file_put_access(struct nfs4_file *fp, int oflag)
  {
  	if (oflag == O_RDWR) {
  		__nfs4_file_put_access(fp, O_RDONLY);
  		__nfs4_file_put_access(fp, O_WRONLY);
  	} else
  		__nfs4_file_put_access(fp, oflag);
  }

  static inline int get_new_stid(struct nfs4_stid *stid)

  {

  	static int min_stateid = 0;

  	struct idr *stateids = &stid->sc_client->cl_stateids;

  	int new_stid;
  	int error;

  	error = idr_get_new_above(stateids, stid, min_stateid, &new_stid);

  	/*

  	 * Note: the necessary preallocation was done in
  	 * nfs4_alloc_stateid().  The idr code caps the number of
  	 * preallocations that can exist at a time, but the state lock
  	 * prevents anyone from using ours before we get here:

  	 */
  	BUG_ON(error);
  	/*
  	 * It shouldn't be a problem to reuse an opaque stateid value.
  	 * I don't think it is for 4.1.  But with 4.0 I worry that, for
  	 * example, a stray write retransmission could be accepted by
  	 * the server when it should have been rejected.  Therefore,
  	 * adopt a trick from the sctp code to attempt to maximize the
  	 * amount of time until an id is reused, by ensuring they always
  	 * "increase" (mod INT_MAX):
  	 */

  	min_stateid = new_stid+1;
  	if (min_stateid == INT_MAX)
  		min_stateid = 0;
  	return new_stid;

  }

  static void init_stid(struct nfs4_stid *stid, struct nfs4_client *cl, unsigned char type)

  {
  	stateid_t *s = &stid->sc_stateid;

  	int new_id;

  
  	stid->sc_type = type;
  	stid->sc_client = cl;
  	s->si_opaque.so_clid = cl->cl_clientid;

  	new_id = get_new_stid(stid);

  	s->si_opaque.so_id = (u32)new_id;

  	/* Will be incremented before return to client: */
  	s->si_generation = 0;

  }
  
  static struct nfs4_stid *nfs4_alloc_stid(struct nfs4_client *cl, struct kmem_cache *slab)
  {
  	struct idr *stateids = &cl->cl_stateids;
  
  	if (!idr_pre_get(stateids, GFP_KERNEL))
  		return NULL;
  	/*
  	 * Note: if we fail here (or any time between now and the time
  	 * we actually get the new idr), we won't need to undo the idr
  	 * preallocation, since the idr code caps the number of
  	 * preallocated entries.
  	 */
  	return kmem_cache_alloc(slab, GFP_KERNEL);

  }

  static struct nfs4_ol_stateid * nfs4_alloc_stateid(struct nfs4_client *clp)
  {
  	return openlockstateid(nfs4_alloc_stid(clp, stateid_slab));
  }

  static struct nfs4_delegation *

  alloc_init_deleg(struct nfs4_client *clp, struct nfs4_ol_stateid *stp, struct svc_fh *current_fh, u32 type)

  {
  	struct nfs4_delegation *dp;
  	struct nfs4_file *fp = stp->st_file;

  
  	dprintk("NFSD alloc_init_deleg
  ");

  	/*
  	 * Major work on the lease subsystem (for example, to support
  	 * calbacks on stat) will be required before we can support
  	 * write delegations properly.
  	 */
  	if (type != NFS4_OPEN_DELEGATE_READ)
  		return NULL;

  	if (fp->fi_had_conflict)
  		return NULL;

  	if (num_delegations > max_delegations)

  		return NULL;

  	dp = delegstateid(nfs4_alloc_stid(clp, deleg_slab));

  	if (dp == NULL)

  		return dp;

  	init_stid(&dp->dl_stid, clp, NFS4_DELEG_STID);

  	/*
  	 * delegation seqid's are never incremented.  The 4.1 special

  	 * meaning of seqid 0 isn't meaningful, really, but let's avoid
  	 * 0 anyway just for consistency and use 1:

  	 */
  	dp->dl_stid.sc_stateid.si_generation = 1;

  	num_delegations++;

  	INIT_LIST_HEAD(&dp->dl_perfile);
  	INIT_LIST_HEAD(&dp->dl_perclnt);

  	INIT_LIST_HEAD(&dp->dl_recall_lru);

  	get_nfs4_file(fp);

  	dp->dl_file = fp;

  	dp->dl_type = type;

  	fh_copy_shallow(&dp->dl_fh, &current_fh->fh_handle);

  	dp->dl_time = 0;
  	atomic_set(&dp->dl_count, 1);

  	INIT_WORK(&dp->dl_recall.cb_work, nfsd4_do_callback_rpc);

  	return dp;
  }
  
  void
  nfs4_put_delegation(struct nfs4_delegation *dp)
  {
  	if (atomic_dec_and_test(&dp->dl_count)) {
  		dprintk("NFSD: freeing dp %p
  ",dp);

  		put_nfs4_file(dp->dl_file);

  		kmem_cache_free(deleg_slab, dp);

  		num_delegations--;

  	}
  }

  static void nfs4_put_deleg_lease(struct nfs4_file *fp)

  {

  	if (atomic_dec_and_test(&fp->fi_delegees)) {
  		vfs_setlease(fp->fi_deleg_file, F_UNLCK, &fp->fi_lease);
  		fp->fi_lease = NULL;

  		fput(fp->fi_deleg_file);

  		fp->fi_deleg_file = NULL;
  	}

  }

  static void unhash_stid(struct nfs4_stid *s)
  {

  	struct idr *stateids = &s->sc_client->cl_stateids;
  
  	idr_remove(stateids, s->sc_stateid.si_opaque.so_id);

  }

  /* Called under the state lock. */
  static void
  unhash_delegation(struct nfs4_delegation *dp)
  {

  	unhash_stid(&dp->dl_stid);

  	list_del_init(&dp->dl_perclnt);

  	spin_lock(&recall_lock);

  	list_del_init(&dp->dl_perfile);

  	list_del_init(&dp->dl_recall_lru);
  	spin_unlock(&recall_lock);

  	nfs4_put_deleg_lease(dp->dl_file);

  	nfs4_put_delegation(dp);
  }
  
  /* 
   * SETCLIENTID state 
   */

  /* client_lock protects the client lru list and session hash table */

  static DEFINE_SPINLOCK(client_lock);

  /* Hash tables for nfs4_clientid state */
  #define CLIENT_HASH_BITS                 4
  #define CLIENT_HASH_SIZE                (1 << CLIENT_HASH_BITS)
  #define CLIENT_HASH_MASK                (CLIENT_HASH_SIZE - 1)

  static unsigned int clientid_hashval(u32 id)
  {
  	return id & CLIENT_HASH_MASK;
  }
  
  static unsigned int clientstr_hashval(const char *name)
  {
  	return opaque_hashval(name, 8) & CLIENT_HASH_MASK;
  }

  /*
   * reclaim_str_hashtbl[] holds known client info from previous reset/reboot
   * used in reboot/reset lease grace period processing
   *
   * conf_id_hashtbl[], and conf_str_hashtbl[] hold confirmed
   * setclientid_confirmed info. 
   *
   * unconf_str_hastbl[] and unconf_id_hashtbl[] hold unconfirmed 
   * setclientid info.
   *
   * client_lru holds client queue ordered by nfs4_client.cl_time
   * for lease renewal.
   *
   * close_lru holds (open) stateowner queue ordered by nfs4_stateowner.so_time
   * for last close replay.
   */
  static struct list_head	reclaim_str_hashtbl[CLIENT_HASH_SIZE];
  static int reclaim_str_hashtbl_size = 0;
  static struct list_head	conf_id_hashtbl[CLIENT_HASH_SIZE];
  static struct list_head	conf_str_hashtbl[CLIENT_HASH_SIZE];
  static struct list_head	unconf_str_hashtbl[CLIENT_HASH_SIZE];
  static struct list_head	unconf_id_hashtbl[CLIENT_HASH_SIZE];
  static struct list_head client_lru;
  static struct list_head close_lru;

  /*
   * We store the NONE, READ, WRITE, and BOTH bits separately in the
   * st_{access,deny}_bmap field of the stateid, in order to track not
   * only what share bits are currently in force, but also what
   * combinations of share bits previous opens have used.  This allows us
   * to enforce the recommendation of rfc 3530 14.2.19 that the server
   * return an error if the client attempt to downgrade to a combination
   * of share bits not explicable by closing some of its previous opens.
   *
   * XXX: This enforcement is actually incomplete, since we don't keep
   * track of access/deny bit combinations; so, e.g., we allow:
   *
   *	OPEN allow read, deny write
   *	OPEN allow both, deny none
   *	DOWNGRADE allow read, deny none
   *
   * which we should reject.
   */
  static void
  set_access(unsigned int *access, unsigned long bmap) {
  	int i;
  
  	*access = 0;
  	for (i = 1; i < 4; i++) {
  		if (test_bit(i, &bmap))
  			*access |= i;
  	}
  }
  
  static void
  set_deny(unsigned int *deny, unsigned long bmap) {
  	int i;
  
  	*deny = 0;
  	for (i = 0; i < 4; i++) {
  		if (test_bit(i, &bmap))
  			*deny |= i ;
  	}
  }
  
  static int

  test_share(struct nfs4_ol_stateid *stp, struct nfsd4_open *open) {

  	unsigned int access, deny;
  
  	set_access(&access, stp->st_access_bmap);
  	set_deny(&deny, stp->st_deny_bmap);
  	if ((access & open->op_share_deny) || (deny & open->op_share_access))
  		return 0;
  	return 1;
  }
  
  static int nfs4_access_to_omode(u32 access)
  {

  	switch (access & NFS4_SHARE_ACCESS_BOTH) {

  	case NFS4_SHARE_ACCESS_READ:
  		return O_RDONLY;
  	case NFS4_SHARE_ACCESS_WRITE:
  		return O_WRONLY;
  	case NFS4_SHARE_ACCESS_BOTH:
  		return O_RDWR;
  	}
  	BUG();
  }

  static void unhash_generic_stateid(struct nfs4_ol_stateid *stp)

  {

  	list_del(&stp->st_perfile);
  	list_del(&stp->st_perstateowner);
  }

  static void close_generic_stateid(struct nfs4_ol_stateid *stp)

  {

  	int i;

  	if (stp->st_access_bmap) {

  		for (i = 1; i < 4; i++) {
  			if (test_bit(i, &stp->st_access_bmap))
  				nfs4_file_put_access(stp->st_file,
  						nfs4_access_to_omode(i));

  			__clear_bit(i, &stp->st_access_bmap);

  		}

  	}

  	put_nfs4_file(stp->st_file);

  	stp->st_file = NULL;
  }

  static void free_generic_stateid(struct nfs4_ol_stateid *stp)

  {

  	kmem_cache_free(stateid_slab, stp);
  }

  static void release_lock_stateid(struct nfs4_ol_stateid *stp)

  {
  	struct file *file;
  
  	unhash_generic_stateid(stp);

  	unhash_stid(&stp->st_stid);

  	file = find_any_file(stp->st_file);
  	if (file)

  		locks_remove_posix(file, (fl_owner_t)lockowner(stp->st_stateowner));

  	close_generic_stateid(stp);

  	free_generic_stateid(stp);
  }

  static void unhash_lockowner(struct nfs4_lockowner *lo)

  {

  	struct nfs4_ol_stateid *stp;

  	list_del(&lo->lo_owner.so_strhash);
  	list_del(&lo->lo_perstateid);

  	list_del(&lo->lo_owner_ino_hash);

  	while (!list_empty(&lo->lo_owner.so_stateids)) {
  		stp = list_first_entry(&lo->lo_owner.so_stateids,

  				struct nfs4_ol_stateid, st_perstateowner);

  		release_lock_stateid(stp);
  	}
  }

  static void release_lockowner(struct nfs4_lockowner *lo)

  {

  	unhash_lockowner(lo);
  	nfs4_free_lockowner(lo);

  }
  
  static void

  release_stateid_lockowners(struct nfs4_ol_stateid *open_stp)

  {

  	struct nfs4_lockowner *lo;

  
  	while (!list_empty(&open_stp->st_lockowners)) {

  		lo = list_entry(open_stp->st_lockowners.next,
  				struct nfs4_lockowner, lo_perstateid);
  		release_lockowner(lo);

  	}
  }

  static void unhash_open_stateid(struct nfs4_ol_stateid *stp)

  {
  	unhash_generic_stateid(stp);
  	release_stateid_lockowners(stp);

  	close_generic_stateid(stp);
  }
  
  static void release_open_stateid(struct nfs4_ol_stateid *stp)
  {
  	unhash_open_stateid(stp);

  	unhash_stid(&stp->st_stid);

  	free_generic_stateid(stp);
  }

  static void unhash_openowner(struct nfs4_openowner *oo)

  {

  	struct nfs4_ol_stateid *stp;

  	list_del(&oo->oo_owner.so_strhash);
  	list_del(&oo->oo_perclient);
  	while (!list_empty(&oo->oo_owner.so_stateids)) {
  		stp = list_first_entry(&oo->oo_owner.so_stateids,

  				struct nfs4_ol_stateid, st_perstateowner);

  		release_open_stateid(stp);

  	}
  }

  static void release_last_closed_stateid(struct nfs4_openowner *oo)
  {
  	struct nfs4_ol_stateid *s = oo->oo_last_closed_stid;
  
  	if (s) {

  		unhash_stid(&s->st_stid);

  		free_generic_stateid(s);
  		oo->oo_last_closed_stid = NULL;
  	}
  }

  static void release_openowner(struct nfs4_openowner *oo)

  {

  	unhash_openowner(oo);
  	list_del(&oo->oo_close_lru);

  	release_last_closed_stateid(oo);

  	nfs4_free_openowner(oo);

  }

  #define SESSION_HASH_SIZE	512
  static struct list_head sessionid_hashtbl[SESSION_HASH_SIZE];
  
  static inline int
  hash_sessionid(struct nfs4_sessionid *sessionid)
  {
  	struct nfsd4_sessionid *sid = (struct nfsd4_sessionid *)sessionid;
  
  	return sid->sequence % SESSION_HASH_SIZE;
  }
  
  static inline void
  dump_sessionid(const char *fn, struct nfs4_sessionid *sessionid)
  {
  	u32 *ptr = (u32 *)(&sessionid->data[0]);
  	dprintk("%s: %u:%u:%u:%u
  ", fn, ptr[0], ptr[1], ptr[2], ptr[3]);
  }

  static void
  gen_sessionid(struct nfsd4_session *ses)
  {
  	struct nfs4_client *clp = ses->se_client;
  	struct nfsd4_sessionid *sid;
  
  	sid = (struct nfsd4_sessionid *)ses->se_sessionid.data;
  	sid->clientid = clp->cl_clientid;
  	sid->sequence = current_sessionid++;
  	sid->reserved = 0;
  }
  
  /*

   * The protocol defines ca_maxresponssize_cached to include the size of
   * the rpc header, but all we need to cache is the data starting after
   * the end of the initial SEQUENCE operation--the rest we regenerate
   * each time.  Therefore we can advertise a ca_maxresponssize_cached
   * value that is the number of bytes in our cache plus a few additional
   * bytes.  In order to stay on the safe side, and not promise more than
   * we can cache, those additional bytes must be the minimum possible: 24
   * bytes of rpc header (xid through accept state, with AUTH_NULL
   * verifier), 12 for the compound header (with zero-length tag), and 44
   * for the SEQUENCE op response:
   */
  #define NFSD_MIN_HDR_SEQ_SZ  (24 + 12 + 44)

  static void
  free_session_slots(struct nfsd4_session *ses)
  {
  	int i;
  
  	for (i = 0; i < ses->se_fchannel.maxreqs; i++)
  		kfree(ses->se_slots[i]);
  }

  /*

   * We don't actually need to cache the rpc and session headers, so we
   * can allocate a little less for each slot:
   */
  static inline int slot_bytes(struct nfsd4_channel_attrs *ca)
  {
  	return ca->maxresp_cached - NFSD_MIN_HDR_SEQ_SZ;
  }

  static int nfsd4_sanitize_slot_size(u32 size)

  {

  	size -= NFSD_MIN_HDR_SEQ_SZ; /* We don't cache the rpc header */
  	size = min_t(u32, size, NFSD_SLOT_CACHE_SIZE);

  	return size;
  }

  /*
   * XXX: If we run out of reserved DRC memory we could (up to a point)

   * re-negotiate active sessions and reduce their slot usage to make

   * room for new connections. For now we just fail the create session.

   */

  static int nfsd4_get_drc_mem(int slotsize, u32 num)

  {

  	int avail;

  	num = min_t(u32, num, NFSD_MAX_SLOTS_PER_SESSION);

  	spin_lock(&nfsd_drc_lock);
  	avail = min_t(int, NFSD_MAX_MEM_PER_SESSION,
  			nfsd_drc_max_mem - nfsd_drc_mem_used);
  	num = min_t(int, num, avail / slotsize);
  	nfsd_drc_mem_used += num * slotsize;
  	spin_unlock(&nfsd_drc_lock);

  	return num;
  }

  static void nfsd4_put_drc_mem(int slotsize, int num)
  {

  	spin_lock(&nfsd_drc_lock);

  	nfsd_drc_mem_used -= slotsize * num;

  	spin_unlock(&nfsd_drc_lock);

  }

  static struct nfsd4_session *alloc_session(int slotsize, int numslots)
  {
  	struct nfsd4_session *new;
  	int mem, i;

  	BUILD_BUG_ON(NFSD_MAX_SLOTS_PER_SESSION * sizeof(struct nfsd4_slot *)
  			+ sizeof(struct nfsd4_session) > PAGE_SIZE);
  	mem = numslots * sizeof(struct nfsd4_slot *);

  	new = kzalloc(sizeof(*new) + mem, GFP_KERNEL);
  	if (!new)
  		return NULL;

  	/* allocate each struct nfsd4_slot and data cache in one piece */

  	for (i = 0; i < numslots; i++) {
  		mem = sizeof(struct nfsd4_slot) + slotsize;
  		new->se_slots[i] = kzalloc(mem, GFP_KERNEL);
  		if (!new->se_slots[i])

  			goto out_free;

  	}

  	return new;
  out_free:
  	while (i--)
  		kfree(new->se_slots[i]);
  	kfree(new);
  	return NULL;

  }

  static void init_forechannel_attrs(struct nfsd4_channel_attrs *new, struct nfsd4_channel_attrs *req, int numslots, int slotsize)

  {

  	u32 maxrpc = nfsd_serv->sv_max_mesg;

  	new->maxreqs = numslots;

  	new->maxresp_cached = min_t(u32, req->maxresp_cached,
  					slotsize + NFSD_MIN_HDR_SEQ_SZ);

  	new->maxreq_sz = min_t(u32, req->maxreq_sz, maxrpc);
  	new->maxresp_sz = min_t(u32, req->maxresp_sz, maxrpc);
  	new->maxops = min_t(u32, req->maxops, NFSD_MAX_OPS_PER_COMPOUND);
  }

  static void free_conn(struct nfsd4_conn *c)
  {
  	svc_xprt_put(c->cn_xprt);
  	kfree(c);
  }

  static void nfsd4_conn_lost(struct svc_xpt_user *u)
  {
  	struct nfsd4_conn *c = container_of(u, struct nfsd4_conn, cn_xpt_user);
  	struct nfs4_client *clp = c->cn_session->se_client;

  	spin_lock(&clp->cl_lock);
  	if (!list_empty(&c->cn_persession)) {
  		list_del(&c->cn_persession);
  		free_conn(c);
  	}
  	spin_unlock(&clp->cl_lock);

  	nfsd4_probe_callback(clp);

  }

  static struct nfsd4_conn *alloc_conn(struct svc_rqst *rqstp, u32 flags)

  {

  	struct nfsd4_conn *conn;

  	conn = kmalloc(sizeof(struct nfsd4_conn), GFP_KERNEL);
  	if (!conn)

  		return NULL;

  	svc_xprt_get(rqstp->rq_xprt);
  	conn->cn_xprt = rqstp->rq_xprt;

  	conn->cn_flags = flags;

  	INIT_LIST_HEAD(&conn->cn_xpt_user.list);
  	return conn;
  }

  static void __nfsd4_hash_conn(struct nfsd4_conn *conn, struct nfsd4_session *ses)
  {
  	conn->cn_session = ses;
  	list_add(&conn->cn_persession, &ses->se_conns);

  }

  static void nfsd4_hash_conn(struct nfsd4_conn *conn, struct nfsd4_session *ses)

  {

  	struct nfs4_client *clp = ses->se_client;

  	spin_lock(&clp->cl_lock);

  	__nfsd4_hash_conn(conn, ses);

  	spin_unlock(&clp->cl_lock);

  }

  static int nfsd4_register_conn(struct nfsd4_conn *conn)

  {

  	conn->cn_xpt_user.callback = nfsd4_conn_lost;

  	return register_xpt_user(conn->cn_xprt, &conn->cn_xpt_user);

  }

  static __be32 nfsd4_new_conn(struct svc_rqst *rqstp, struct nfsd4_session *ses, u32 dir)

  {

  	struct nfsd4_conn *conn;

  	int ret;

  	conn = alloc_conn(rqstp, dir);

  	if (!conn)
  		return nfserr_jukebox;
  	nfsd4_hash_conn(conn, ses);

  	ret = nfsd4_register_conn(conn);
  	if (ret)
  		/* oops; xprt is already down: */
  		nfsd4_conn_lost(&conn->cn_xpt_user);

  	return nfs_ok;
  }

  static __be32 nfsd4_new_conn_from_crses(struct svc_rqst *rqstp, struct nfsd4_session *ses)
  {
  	u32 dir = NFS4_CDFC4_FORE;
  
  	if (ses->se_flags & SESSION4_BACK_CHAN)
  		dir |= NFS4_CDFC4_BACK;
  
  	return nfsd4_new_conn(rqstp, ses, dir);
  }
  
  /* must be called under client_lock */

  static void nfsd4_del_conns(struct nfsd4_session *s)

  {

  	struct nfs4_client *clp = s->se_client;
  	struct nfsd4_conn *c;

  	spin_lock(&clp->cl_lock);
  	while (!list_empty(&s->se_conns)) {
  		c = list_first_entry(&s->se_conns, struct nfsd4_conn, cn_persession);
  		list_del_init(&c->cn_persession);
  		spin_unlock(&clp->cl_lock);

  		unregister_xpt_user(c->cn_xprt, &c->cn_xpt_user);
  		free_conn(c);

  		spin_lock(&clp->cl_lock);
  	}
  	spin_unlock(&clp->cl_lock);

  }

  void free_session(struct kref *kref)
  {
  	struct nfsd4_session *ses;
  	int mem;
  
  	ses = container_of(kref, struct nfsd4_session, se_ref);

  	nfsd4_del_conns(ses);

  	spin_lock(&nfsd_drc_lock);
  	mem = ses->se_fchannel.maxreqs * slot_bytes(&ses->se_fchannel);
  	nfsd_drc_mem_used -= mem;
  	spin_unlock(&nfsd_drc_lock);
  	free_session_slots(ses);
  	kfree(ses);
  }

  static struct nfsd4_session *alloc_init_session(struct svc_rqst *rqstp, struct nfs4_client *clp, struct nfsd4_create_session *cses)

  {
  	struct nfsd4_session *new;
  	struct nfsd4_channel_attrs *fchan = &cses->fore_channel;
  	int numslots, slotsize;

  	int status;

  	int idx;
  
  	/*
  	 * Note decreasing slot size below client's request may
  	 * make it difficult for client to function correctly, whereas
  	 * decreasing the number of slots will (just?) affect
  	 * performance.  When short on memory we therefore prefer to
  	 * decrease number of slots instead of their size.
  	 */
  	slotsize = nfsd4_sanitize_slot_size(fchan->maxresp_cached);
  	numslots = nfsd4_get_drc_mem(slotsize, fchan->maxreqs);

  	if (numslots < 1)
  		return NULL;

  
  	new = alloc_session(slotsize, numslots);
  	if (!new) {
  		nfsd4_put_drc_mem(slotsize, fchan->maxreqs);

  		return NULL;

  	}

  	init_forechannel_attrs(&new->se_fchannel, fchan, numslots, slotsize);

  	new->se_client = clp;
  	gen_sessionid(new);

  	INIT_LIST_HEAD(&new->se_conns);

  	new->se_cb_seq_nr = 1;

  	new->se_flags = cses->flags;

  	new->se_cb_prog = cses->callback_prog;

  	kref_init(&new->se_ref);

  	idx = hash_sessionid(&new->se_sessionid);

  	spin_lock(&client_lock);

  	list_add(&new->se_hash, &sessionid_hashtbl[idx]);

  	spin_lock(&clp->cl_lock);

  	list_add(&new->se_perclnt, &clp->cl_sessions);

  	spin_unlock(&clp->cl_lock);

  	spin_unlock(&client_lock);

  	status = nfsd4_new_conn_from_crses(rqstp, new);

  	/* whoops: benny points out, status is ignored! (err, or bogus) */

  	if (status) {
  		free_session(&new->se_ref);

  		return NULL;

  	}

  	if (cses->flags & SESSION4_BACK_CHAN) {

  		struct sockaddr *sa = svc_addr(rqstp);

  		/*
  		 * This is a little silly; with sessions there's no real
  		 * use for the callback address.  Use the peer address
  		 * as a reasonable default for now, but consider fixing
  		 * the rpc client not to require an address in the
  		 * future:
  		 */

  		rpc_copy_addr((struct sockaddr *)&clp->cl_cb_conn.cb_addr, sa);
  		clp->cl_cb_conn.cb_addrlen = svc_addr_len(sa);

  	}

  	nfsd4_probe_callback(clp);

  	return new;

  }

  /* caller must hold client_lock */

  static struct nfsd4_session *
  find_in_sessionid_hashtbl(struct nfs4_sessionid *sessionid)
  {
  	struct nfsd4_session *elem;
  	int idx;
  
  	dump_sessionid(__func__, sessionid);
  	idx = hash_sessionid(sessionid);

  	/* Search in the appropriate list */
  	list_for_each_entry(elem, &sessionid_hashtbl[idx], se_hash) {

  		if (!memcmp(elem->se_sessionid.data, sessionid->data,
  			    NFS4_MAX_SESSIONID_LEN)) {
  			return elem;
  		}
  	}
  
  	dprintk("%s: session not found
  ", __func__);
  	return NULL;
  }

  /* caller must hold client_lock */

  static void

  unhash_session(struct nfsd4_session *ses)

  {
  	list_del(&ses->se_hash);

  	spin_lock(&ses->se_client->cl_lock);

  	list_del(&ses->se_perclnt);

  	spin_unlock(&ses->se_client->cl_lock);

  }

  /* must be called under the client_lock */

  static inline void

  renew_client_locked(struct nfs4_client *clp)

  {

  	if (is_client_expired(clp)) {
  		dprintk("%s: client (clientid %08x/%08x) already expired
  ",
  			__func__,
  			clp->cl_clientid.cl_boot,
  			clp->cl_clientid.cl_id);
  		return;
  	}

  	dprintk("renewing client (clientid %08x/%08x)
  ", 
  			clp->cl_clientid.cl_boot, 
  			clp->cl_clientid.cl_id);
  	list_move_tail(&clp->cl_lru, &client_lru);
  	clp->cl_time = get_seconds();
  }

  static inline void
  renew_client(struct nfs4_client *clp)
  {
  	spin_lock(&client_lock);
  	renew_client_locked(clp);
  	spin_unlock(&client_lock);
  }

  /* SETCLIENTID and SETCLIENTID_CONFIRM Helper functions */
  static int
  STALE_CLIENTID(clientid_t *clid)
  {
  	if (clid->cl_boot == boot_time)
  		return 0;

  	dprintk("NFSD stale clientid (%08x/%08x) boot_time %08lx
  ",
  		clid->cl_boot, clid->cl_id, boot_time);

  	return 1;
  }
  
  /* 
   * XXX Should we use a slab cache ?
   * This type of memory management is somewhat inefficient, but we use it
   * anyway since SETCLIENTID is not a common operation.
   */

  static struct nfs4_client *alloc_client(struct xdr_netobj name)

  {
  	struct nfs4_client *clp;

  	clp = kzalloc(sizeof(struct nfs4_client), GFP_KERNEL);
  	if (clp == NULL)
  		return NULL;

  	clp->cl_name.data = kmemdup(name.data, name.len, GFP_KERNEL);

  	if (clp->cl_name.data == NULL) {
  		kfree(clp);
  		return NULL;

  	}

  	clp->cl_name.len = name.len;

  	return clp;
  }
  
  static inline void
  free_client(struct nfs4_client *clp)
  {

  	while (!list_empty(&clp->cl_sessions)) {
  		struct nfsd4_session *ses;
  		ses = list_entry(clp->cl_sessions.next, struct nfsd4_session,
  				se_perclnt);
  		list_del(&ses->se_perclnt);
  		nfsd4_put_session(ses);
  	}

  	if (clp->cl_cred.cr_group_info)
  		put_group_info(clp->cl_cred.cr_group_info);

  	kfree(clp->cl_principal);

  	kfree(clp->cl_name.data);
  	kfree(clp);
  }

  void
  release_session_client(struct nfsd4_session *session)
  {
  	struct nfs4_client *clp = session->se_client;
  
  	if (!atomic_dec_and_lock(&clp->cl_refcount, &client_lock))
  		return;
  	if (is_client_expired(clp)) {
  		free_client(clp);
  		session->se_client = NULL;
  	} else
  		renew_client_locked(clp);
  	spin_unlock(&client_lock);

  }

  /* must be called under the client_lock */
  static inline void
  unhash_client_locked(struct nfs4_client *clp)
  {

  	struct nfsd4_session *ses;

  	mark_client_expired(clp);

  	list_del(&clp->cl_lru);

  	spin_lock(&clp->cl_lock);

  	list_for_each_entry(ses, &clp->cl_sessions, se_perclnt)
  		list_del_init(&ses->se_hash);

  	spin_unlock(&clp->cl_lock);

  }

  static void
  expire_client(struct nfs4_client *clp)
  {

  	struct nfs4_openowner *oo;

  	struct nfs4_delegation *dp;

  	struct list_head reaplist;

  	INIT_LIST_HEAD(&reaplist);
  	spin_lock(&recall_lock);

  	while (!list_empty(&clp->cl_delegations)) {
  		dp = list_entry(clp->cl_delegations.next, struct nfs4_delegation, dl_perclnt);

  		list_del_init(&dp->dl_perclnt);

  		list_move(&dp->dl_recall_lru, &reaplist);
  	}
  	spin_unlock(&recall_lock);
  	while (!list_empty(&reaplist)) {
  		dp = list_entry(reaplist.next, struct nfs4_delegation, dl_recall_lru);

  		unhash_delegation(dp);
  	}

  	while (!list_empty(&clp->cl_openowners)) {

  		oo = list_entry(clp->cl_openowners.next, struct nfs4_openowner, oo_perclient);
  		release_openowner(oo);

  	}

  	nfsd4_shutdown_callback(clp);

  	if (clp->cl_cb_conn.cb_xprt)
  		svc_xprt_put(clp->cl_cb_conn.cb_xprt);

  	list_del(&clp->cl_idhash);
  	list_del(&clp->cl_strhash);

  	spin_lock(&client_lock);

  	unhash_client_locked(clp);

  	if (atomic_read(&clp->cl_refcount) == 0)
  		free_client(clp);

  	spin_unlock(&client_lock);

  }

  static void copy_verf(struct nfs4_client *target, nfs4_verifier *source)
  {
  	memcpy(target->cl_verifier.data, source->data,
  			sizeof(target->cl_verifier.data));

  }

  static void copy_clid(struct nfs4_client *target, struct nfs4_client *source)
  {

  	target->cl_clientid.cl_boot = source->cl_clientid.cl_boot; 
  	target->cl_clientid.cl_id = source->cl_clientid.cl_id; 
  }

  static void copy_cred(struct svc_cred *target, struct svc_cred *source)
  {

  	target->cr_uid = source->cr_uid;
  	target->cr_gid = source->cr_gid;
  	target->cr_group_info = source->cr_group_info;
  	get_group_info(target->cr_group_info);
  }

  static int same_name(const char *n1, const char *n2)

  {

  	return 0 == memcmp(n1, n2, HEXDIR_LEN);

  }
  
  static int

  same_verf(nfs4_verifier *v1, nfs4_verifier *v2)
  {
  	return 0 == memcmp(v1->data, v2->data, sizeof(v1->data));

  }
  
  static int

  same_clid(clientid_t *cl1, clientid_t *cl2)
  {
  	return (cl1->cl_boot == cl2->cl_boot) && (cl1->cl_id == cl2->cl_id);

  }
  
  /* XXX what about NGROUP */
  static int

  same_creds(struct svc_cred *cr1, struct svc_cred *cr2)
  {
  	return cr1->cr_uid == cr2->cr_uid;

  }

  static void gen_clid(struct nfs4_client *clp)
  {
  	static u32 current_clientid = 1;

  	clp->cl_clientid.cl_boot = boot_time;
  	clp->cl_clientid.cl_id = current_clientid++; 
  }

  static void gen_confirm(struct nfs4_client *clp)
  {
  	static u32 i;
  	u32 *p;

  	p = (u32 *)clp->cl_confirm.data;

  	*p++ = get_seconds();
  	*p++ = i++;

  }

  static struct nfs4_stid *find_stateid(struct nfs4_client *cl, stateid_t *t)

  {

  	return idr_find(&cl->cl_stateids, t->si_opaque.so_id);

  }

  static struct nfs4_stid *find_stateid_by_type(struct nfs4_client *cl, stateid_t *t, char typemask)

  {
  	struct nfs4_stid *s;

  	s = find_stateid(cl, t);

  	if (!s)
  		return NULL;

  	if (typemask & s->sc_type)

  		return s;

  	return NULL;
  }

  static struct nfs4_client *create_client(struct xdr_netobj name, char *recdir,
  		struct svc_rqst *rqstp, nfs4_verifier *verf)
  {
  	struct nfs4_client *clp;
  	struct sockaddr *sa = svc_addr(rqstp);
  	char *princ;
  
  	clp = alloc_client(name);
  	if (clp == NULL)
  		return NULL;

  	INIT_LIST_HEAD(&clp->cl_sessions);

  	princ = svc_gss_principal(rqstp);
  	if (princ) {
  		clp->cl_principal = kstrdup(princ, GFP_KERNEL);
  		if (clp->cl_principal == NULL) {
  			free_client(clp);
  			return NULL;
  		}
  	}

  	idr_init(&clp->cl_stateids);

  	memcpy(clp->cl_recdir, recdir, HEXDIR_LEN);

  	atomic_set(&clp->cl_refcount, 0);

  	clp->cl_cb_state = NFSD4_CB_UNKNOWN;

  	INIT_LIST_HEAD(&clp->cl_idhash);
  	INIT_LIST_HEAD(&clp->cl_strhash);
  	INIT_LIST_HEAD(&clp->cl_openowners);
  	INIT_LIST_HEAD(&clp->cl_delegations);

  	INIT_LIST_HEAD(&clp->cl_lru);

  	INIT_LIST_HEAD(&clp->cl_callbacks);

  	spin_lock_init(&clp->cl_lock);

  	INIT_WORK(&clp->cl_cb_null.cb_work, nfsd4_do_callback_rpc);

  	clp->cl_time = get_seconds();

  	clear_bit(0, &clp->cl_cb_slot_busy);
  	rpc_init_wait_queue(&clp->cl_cb_waitq, "Backchannel slot table");
  	copy_verf(clp, verf);
  	rpc_copy_addr((struct sockaddr *) &clp->cl_addr, sa);
  	clp->cl_flavor = rqstp->rq_flavor;
  	copy_cred(&clp->cl_cred, &rqstp->rq_cred);
  	gen_confirm(clp);

  	clp->cl_cb_session = NULL;

  	return clp;
  }

  static void

  add_to_unconfirmed(struct nfs4_client *clp, unsigned int strhashval)
  {
  	unsigned int idhashval;
  
  	list_add(&clp->cl_strhash, &unconf_str_hashtbl[strhashval]);
  	idhashval = clientid_hashval(clp->cl_clientid.cl_id);
  	list_add(&clp->cl_idhash, &unconf_id_hashtbl[idhashval]);

  	renew_client(clp);

  }

  static void

  move_to_confirmed(struct nfs4_client *clp)
  {
  	unsigned int idhashval = clientid_hashval(clp->cl_clientid.cl_id);
  	unsigned int strhashval;
  
  	dprintk("NFSD: move_to_confirm nfs4_client %p
  ", clp);

  	list_move(&clp->cl_idhash, &conf_id_hashtbl[idhashval]);

  	strhashval = clientstr_hashval(clp->cl_recdir);

  	list_move(&clp->cl_strhash, &conf_str_hashtbl[strhashval]);

  	renew_client(clp);
  }
  
  static struct nfs4_client *
  find_confirmed_client(clientid_t *clid)
  {
  	struct nfs4_client *clp;
  	unsigned int idhashval = clientid_hashval(clid->cl_id);
  
  	list_for_each_entry(clp, &conf_id_hashtbl[idhashval], cl_idhash) {

  		if (same_clid(&clp->cl_clientid, clid)) {
  			renew_client(clp);

  			return clp;

  		}

  	}
  	return NULL;
  }
  
  static struct nfs4_client *
  find_unconfirmed_client(clientid_t *clid)
  {
  	struct nfs4_client *clp;
  	unsigned int idhashval = clientid_hashval(clid->cl_id);
  
  	list_for_each_entry(clp, &unconf_id_hashtbl[idhashval], cl_idhash) {

  		if (same_clid(&clp->cl_clientid, clid))

  			return clp;
  	}
  	return NULL;
  }

  static bool clp_used_exchangeid(struct nfs4_client *clp)

  {

  	return clp->cl_exchange_flags != 0;

  } 

  static struct nfs4_client *

  find_confirmed_client_by_str(const char *dname, unsigned int hashval)

  {
  	struct nfs4_client *clp;
  
  	list_for_each_entry(clp, &conf_str_hashtbl[hashval], cl_strhash) {

  		if (same_name(clp->cl_recdir, dname))

  			return clp;
  	}
  	return NULL;
  }
  
  static struct nfs4_client *

  find_unconfirmed_client_by_str(const char *dname, unsigned int hashval)

  {
  	struct nfs4_client *clp;
  
  	list_for_each_entry(clp, &unconf_str_hashtbl[hashval], cl_strhash) {

  		if (same_name(clp->cl_recdir, dname))

  			return clp;
  	}
  	return NULL;
  }

  static void

  gen_callback(struct nfs4_client *clp, struct nfsd4_setclientid *se, struct svc_rqst *rqstp)

  {

  	struct nfs4_cb_conn *conn = &clp->cl_cb_conn;

  	struct sockaddr	*sa = svc_addr(rqstp);
  	u32 scopeid = rpc_get_scope_id(sa);

  	unsigned short expected_family;
  
  	/* Currently, we only support tcp and tcp6 for the callback channel */
  	if (se->se_callback_netid_len == 3 &&
  	    !memcmp(se->se_callback_netid_val, "tcp", 3))
  		expected_family = AF_INET;
  	else if (se->se_callback_netid_len == 4 &&
  		 !memcmp(se->se_callback_netid_val, "tcp6", 4))
  		expected_family = AF_INET6;
  	else

  		goto out_err;

  	conn->cb_addrlen = rpc_uaddr2sockaddr(se->se_callback_addr_val,

  					    se->se_callback_addr_len,

  					    (struct sockaddr *)&conn->cb_addr,
  					    sizeof(conn->cb_addr));

  	if (!conn->cb_addrlen || conn->cb_addr.ss_family != expected_family)

  		goto out_err;

  	if (conn->cb_addr.ss_family == AF_INET6)
  		((struct sockaddr_in6 *)&conn->cb_addr)->sin6_scope_id = scopeid;

  	conn->cb_prog = se->se_callback_prog;
  	conn->cb_ident = se->se_callback_ident;

  	memcpy(&conn->cb_saddr, &rqstp->rq_daddr, rqstp->rq_daddrlen);

  	return;
  out_err:

  	conn->cb_addr.ss_family = AF_UNSPEC;
  	conn->cb_addrlen = 0;

  	dprintk(KERN_INFO "NFSD: this client (clientid %08x/%08x) "

  		"will not receive delegations
  ",
  		clp->cl_clientid.cl_boot, clp->cl_clientid.cl_id);

  	return;
  }

  /*

   * Cache a reply. nfsd4_check_drc_limit() has bounded the cache size.

   */

  void
  nfsd4_store_cache_entry(struct nfsd4_compoundres *resp)

  {

  	struct nfsd4_slot *slot = resp->cstate.slot;
  	unsigned int base;

  	dprintk("--> %s slot %p
  ", __func__, slot);

  	slot->sl_opcnt = resp->opcnt;
  	slot->sl_status = resp->cstate.status;

  	if (nfsd4_not_cached(resp)) {

  		slot->sl_datalen = 0;

  		return;

  	}

  	slot->sl_datalen = (char *)resp->p - (char *)resp->cstate.datap;
  	base = (char *)resp->cstate.datap -
  					(char *)resp->xbuf->head[0].iov_base;
  	if (read_bytes_from_xdr_buf(resp->xbuf, base, slot->sl_data,
  				    slot->sl_datalen))
  		WARN("%s: sessions DRC could not cache compound
  ", __func__);
  	return;

  }
  
  /*

   * Encode the replay sequence operation from the slot values.
   * If cachethis is FALSE encode the uncached rep error on the next
   * operation which sets resp->p and increments resp->opcnt for
   * nfs4svc_encode_compoundres.

   *

   */

  static __be32
  nfsd4_enc_sequence_replay(struct nfsd4_compoundargs *args,
  			  struct nfsd4_compoundres *resp)

  {

  	struct nfsd4_op *op;
  	struct nfsd4_slot *slot = resp->cstate.slot;

  	dprintk("--> %s resp->opcnt %d cachethis %u 
  ", __func__,

  		resp->opcnt, resp->cstate.slot->sl_cachethis);

  	/* Encode the replayed sequence operation */
  	op = &args->ops[resp->opcnt - 1];
  	nfsd4_encode_operation(resp, op);

  	/* Return nfserr_retry_uncached_rep in next operation. */

  	if (args->opcnt > 1 && slot->sl_cachethis == 0) {

  		op = &args->ops[resp->opcnt++];
  		op->status = nfserr_retry_uncached_rep;
  		nfsd4_encode_operation(resp, op);

  	}

  	return op->status;

  }
  
  /*

   * The sequence operation is not cached because we can use the slot and
   * session values.

   */
  __be32

  nfsd4_replay_cache_entry(struct nfsd4_compoundres *resp,
  			 struct nfsd4_sequence *seq)

  {

  	struct nfsd4_slot *slot = resp->cstate.slot;

  	__be32 status;

  	dprintk("--> %s slot %p
  ", __func__, slot);

  	/* Either returns 0 or nfserr_retry_uncached */
  	status = nfsd4_enc_sequence_replay(resp->rqstp->rq_argp, resp);
  	if (status == nfserr_retry_uncached_rep)
  		return status;

  	/* The sequence operation has been encoded, cstate->datap set. */
  	memcpy(resp->cstate.datap, slot->sl_data, slot->sl_datalen);

  	resp->opcnt = slot->sl_opcnt;
  	resp->p = resp->cstate.datap + XDR_QUADLEN(slot->sl_datalen);
  	status = slot->sl_status;

  
  	return status;
  }

  /*
   * Set the exchange_id flags returned by the server.
   */
  static void
  nfsd4_set_ex_flags(struct nfs4_client *new, struct nfsd4_exchange_id *clid)
  {
  	/* pNFS is not supported */
  	new->cl_exchange_flags |= EXCHGID4_FLAG_USE_NON_PNFS;
  
  	/* Referrals are supported, Migration is not. */
  	new->cl_exchange_flags |= EXCHGID4_FLAG_SUPP_MOVED_REFER;
  
  	/* set the wire flags to return to client. */
  	clid->flags = new->cl_exchange_flags;
  }

  __be32

  nfsd4_exchange_id(struct svc_rqst *rqstp,
  		  struct nfsd4_compound_state *cstate,
  		  struct nfsd4_exchange_id *exid)
  {

  	struct nfs4_client *unconf, *conf, *new;
  	int status;
  	unsigned int		strhashval;
  	char			dname[HEXDIR_LEN];

  	char			addr_str[INET6_ADDRSTRLEN];

  	nfs4_verifier		verf = exid->verifier;

  	struct sockaddr		*sa = svc_addr(rqstp);

  	rpc_ntop(sa, addr_str, sizeof(addr_str));

  	dprintk("%s rqstp=%p exid=%p clname.len=%u clname.data=%p "

  		"ip_addr=%s flags %x, spa_how %d
  ",

  		__func__, rqstp, exid, exid->clname.len, exid->clname.data,

  		addr_str, exid->flags, exid->spa_how);

  	if (exid->flags & ~EXCHGID4_FLAG_MASK_A)

  		return nfserr_inval;
  
  	/* Currently only support SP4_NONE */
  	switch (exid->spa_how) {
  	case SP4_NONE:
  		break;
  	case SP4_SSV:

  		return nfserr_serverfault;

  	default:
  		BUG();				/* checked by xdr code */
  	case SP4_MACH_CRED:
  		return nfserr_serverfault;	/* no excuse :-/ */
  	}
  
  	status = nfs4_make_rec_clidname(dname, &exid->clname);
  
  	if (status)
  		goto error;
  
  	strhashval = clientstr_hashval(dname);
  
  	nfs4_lock_state();
  	status = nfs_ok;

  	conf = find_confirmed_client_by_str(dname, strhashval);

  	if (conf) {

  		if (!clp_used_exchangeid(conf)) {
  			status = nfserr_clid_inuse; /* XXX: ? */
  			goto out;
  		}

  		if (!same_verf(&verf, &conf->cl_verifier)) {
  			/* 18.35.4 case 8 */
  			if (exid->flags & EXCHGID4_FLAG_UPD_CONFIRMED_REC_A) {
  				status = nfserr_not_same;
  				goto out;
  			}
  			/* Client reboot: destroy old state */
  			expire_client(conf);
  			goto out_new;
  		}
  		if (!same_creds(&conf->cl_cred, &rqstp->rq_cred)) {
  			/* 18.35.4 case 9 */
  			if (exid->flags & EXCHGID4_FLAG_UPD_CONFIRMED_REC_A) {
  				status = nfserr_perm;
  				goto out;
  			}
  			expire_client(conf);
  			goto out_new;
  		}

  		/*
  		 * Set bit when the owner id and verifier map to an already
  		 * confirmed client id (18.35.3).
  		 */
  		exid->flags |= EXCHGID4_FLAG_CONFIRMED_R;
  
  		/*
  		 * Falling into 18.35.4 case 2, possible router replay.
  		 * Leave confirmed record intact and return same result.
  		 */
  		copy_verf(conf, &verf);
  		new = conf;
  		goto out_copy;

  	}
  
  	/* 18.35.4 case 7 */
  	if (exid->flags & EXCHGID4_FLAG_UPD_CONFIRMED_REC_A) {
  		status = nfserr_noent;
  		goto out;

  	}

  	unconf  = find_unconfirmed_client_by_str(dname, strhashval);

  	if (unconf) {
  		/*
  		 * Possible retry or client restart.  Per 18.35.4 case 4,
  		 * a new unconfirmed record should be generated regardless
  		 * of whether any properties have changed.
  		 */
  		expire_client(unconf);
  	}
  
  out_new:
  	/* Normal case */

  	new = create_client(exid->clname, dname, rqstp, &verf);

  	if (new == NULL) {

  		status = nfserr_jukebox;

  		goto out;
  	}

  	gen_clid(new);

  	add_to_unconfirmed(new, strhashval);
  out_copy:
  	exid->clientid.cl_boot = new->cl_clientid.cl_boot;
  	exid->clientid.cl_id = new->cl_clientid.cl_id;

  	exid->seqid = 1;

  	nfsd4_set_ex_flags(new, exid);
  
  	dprintk("nfsd4_exchange_id seqid %d flags %x
  ",

  		new->cl_cs_slot.sl_seqid, new->cl_exchange_flags);

  	status = nfs_ok;
  
  out:
  	nfs4_unlock_state();
  error:
  	dprintk("nfsd4_exchange_id returns %d
  ", ntohl(status));
  	return status;

  }

  static int

  check_slot_seqid(u32 seqid, u32 slot_seqid, int slot_inuse)

  {

  	dprintk("%s enter. seqid %d slot_seqid %d
  ", __func__, seqid,
  		slot_seqid);

  
  	/* The slot is in use, and no response has been sent. */

  	if (slot_inuse) {
  		if (seqid == slot_seqid)

  			return nfserr_jukebox;
  		else
  			return nfserr_seq_misordered;
  	}
  	/* Normal */

  	if (likely(seqid == slot_seqid + 1))

  		return nfs_ok;
  	/* Replay */

  	if (seqid == slot_seqid)

  		return nfserr_replay_cache;
  	/* Wraparound */

  	if (seqid == 1 && (slot_seqid + 1) == 0)

  		return nfs_ok;
  	/* Misordered replay or misordered new request */
  	return nfserr_seq_misordered;
  }

  /*
   * Cache the create session result into the create session single DRC
   * slot cache by saving the xdr structure. sl_seqid has been set.
   * Do this for solo or embedded create session operations.
   */
  static void
  nfsd4_cache_create_session(struct nfsd4_create_session *cr_ses,
  			   struct nfsd4_clid_slot *slot, int nfserr)
  {
  	slot->sl_status = nfserr;
  	memcpy(&slot->sl_cr_ses, cr_ses, sizeof(*cr_ses));
  }
  
  static __be32
  nfsd4_replay_create_session(struct nfsd4_create_session *cr_ses,
  			    struct nfsd4_clid_slot *slot)
  {
  	memcpy(cr_ses, &slot->sl_cr_ses, sizeof(*cr_ses));
  	return slot->sl_status;
  }

  #define NFSD_MIN_REQ_HDR_SEQ_SZ	((\
  			2 * 2 + /* credential,verifier: AUTH_NULL, length 0 */ \
  			1 +	/* MIN tag is length with zero, only length */ \
  			3 +	/* version, opcount, opcode */ \
  			XDR_QUADLEN(NFS4_MAX_SESSIONID_LEN) + \
  				/* seqid, slotID, slotID, cache */ \
  			4 ) * sizeof(__be32))
  
  #define NFSD_MIN_RESP_HDR_SEQ_SZ ((\
  			2 +	/* verifier: AUTH_NULL, length 0 */\
  			1 +	/* status */ \
  			1 +	/* MIN tag is length with zero, only length */ \
  			3 +	/* opcount, opcode, opstatus*/ \
  			XDR_QUADLEN(NFS4_MAX_SESSIONID_LEN) + \
  				/* seqid, slotID, slotID, slotID, status */ \
  			5 ) * sizeof(__be32))
  
  static __be32 check_forechannel_attrs(struct nfsd4_channel_attrs fchannel)
  {
  	return fchannel.maxreq_sz < NFSD_MIN_REQ_HDR_SEQ_SZ
  		|| fchannel.maxresp_sz < NFSD_MIN_RESP_HDR_SEQ_SZ;
  }

  __be32
  nfsd4_create_session(struct svc_rqst *rqstp,
  		     struct nfsd4_compound_state *cstate,
  		     struct nfsd4_create_session *cr_ses)
  {

  	struct sockaddr *sa = svc_addr(rqstp);

  	struct nfs4_client *conf, *unconf;

  	struct nfsd4_session *new;

  	struct nfsd4_clid_slot *cs_slot = NULL;

  	bool confirm_me = false;

  	int status = 0;

  	if (cr_ses->flags & ~SESSION4_FLAG_MASK_A)
  		return nfserr_inval;

  	nfs4_lock_state();
  	unconf = find_unconfirmed_client(&cr_ses->clientid);
  	conf = find_confirmed_client(&cr_ses->clientid);
  
  	if (conf) {

  		cs_slot = &conf->cl_cs_slot;
  		status = check_slot_seqid(cr_ses->seqid, cs_slot->sl_seqid, 0);

  		if (status == nfserr_replay_cache) {

  			dprintk("Got a create_session replay! seqid= %d
  ",

  				cs_slot->sl_seqid);

  			/* Return the cached reply status */

  			status = nfsd4_replay_create_session(cr_ses, cs_slot);

  			goto out;

  		} else if (cr_ses->seqid != cs_slot->sl_seqid + 1) {

  			status = nfserr_seq_misordered;
  			dprintk("Sequence misordered!
  ");
  			dprintk("Expected seqid= %d but got seqid= %d
  ",

  				cs_slot->sl_seqid, cr_ses->seqid);

  			goto out;
  		}

  	} else if (unconf) {
  		if (!same_creds(&unconf->cl_cred, &rqstp->rq_cred) ||

  		    !rpc_cmp_addr(sa, (struct sockaddr *) &unconf->cl_addr)) {

  			status = nfserr_clid_inuse;
  			goto out;
  		}

  		cs_slot = &unconf->cl_cs_slot;
  		status = check_slot_seqid(cr_ses->seqid, cs_slot->sl_seqid, 0);

  		if (status) {
  			/* an unconfirmed replay returns misordered */

  			status = nfserr_seq_misordered;

  			goto out;

  		}

  		confirm_me = true;

  		conf = unconf;
  	} else {
  		status = nfserr_stale_clientid;
  		goto out;
  	}

  	/*

  	 * XXX: we should probably set this at creation time, and check
  	 * for consistent minorversion use throughout:
  	 */
  	conf->cl_minorversion = 1;
  	/*