1 files changed, 677 insertions, 440 deletions

diff --git a/fs/nfsd/nfs4state.c b/fs/nfsd/nfs4state.c
index cf0d2ffb3c84..e98f3c2e9492 100644
--- a/fs/nfsd/nfs4state.c
+++ b/fs/nfsd/nfs4state.c
@@ -33,7 +33,7 @@
 */
 
 #include <linux/file.h>
-#include <linux/smp_lock.h>
+#include <linux/fs.h>
 #include <linux/slab.h>
 #include <linux/namei.h>
 #include <linux/swap.h>
@@ -148,7 +148,7 @@ 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)
-#define FILE_HASH_MASK                  (FILE_HASH_SIZE - 1)
+
 /* hash table for (open)nfs4_stateid */
 #define STATEID_HASH_BITS              10
 #define STATEID_HASH_SIZE              (1 << STATEID_HASH_BITS)
@@ -207,7 +207,6 @@ alloc_init_deleg(struct nfs4_client *clp, struct nfs4_stateid *stp, struct svc_f
 {
         struct nfs4_delegation *dp;
         struct nfs4_file *fp = stp->st_file;
-        struct nfs4_cb_conn *cb = &stp->st_stateowner->so_client->cl_cb_conn;
 
         dprintk("NFSD alloc_init_deleg\n");
         /*
@@ -231,10 +230,7 @@ alloc_init_deleg(struct nfs4_client *clp, struct nfs4_stateid *stp, struct svc_f
         dp->dl_client = clp;
         get_nfs4_file(fp);
         dp->dl_file = fp;
-        nfs4_file_get_access(fp, O_RDONLY);
-        dp->dl_flock = NULL;
         dp->dl_type = type;
-        dp->dl_ident = cb->cb_ident;
         dp->dl_stateid.si_boot = boot_time;
         dp->dl_stateid.si_stateownerid = current_delegid++;
         dp->dl_stateid.si_fileid = 0;
@@ -242,8 +238,6 @@ alloc_init_deleg(struct nfs4_client *clp, struct nfs4_stateid *stp, struct svc_f
         fh_copy_shallow(&dp->dl_fh, &current_fh->fh_handle);
         dp->dl_time = 0;
         atomic_set(&dp->dl_count, 1);
-        list_add(&dp->dl_perfile, &fp->fi_delegations);
-        list_add(&dp->dl_perclnt, &clp->cl_delegations);
         INIT_WORK(&dp->dl_recall.cb_work, nfsd4_do_callback_rpc);
         return dp;
 }
@@ -259,32 +253,26 @@ nfs4_put_delegation(struct nfs4_delegation *dp)
         }
 }
 
-/* Remove the associated file_lock first, then remove the delegation.
- * lease_modify() is called to remove the FS_LEASE file_lock from
- * the i_flock list, eventually calling nfsd's lock_manager
- * fl_release_callback.
- */
-static void
-nfs4_close_delegation(struct nfs4_delegation *dp)
+static void nfs4_put_deleg_lease(struct nfs4_file *fp)
 {
-        struct file *filp = find_readable_file(dp->dl_file);
-
-        dprintk("NFSD: close_delegation dp %p\n",dp);
-        if (dp->dl_flock)
-                vfs_setlease(filp, F_UNLCK, &dp->dl_flock);
-        nfs4_file_put_access(dp->dl_file, O_RDONLY);
+        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;
+        }
 }
 
 /* Called under the state lock. */
 static void
 unhash_delegation(struct nfs4_delegation *dp)
 {
-        list_del_init(&dp->dl_perfile);
         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_close_delegation(dp);
+        nfs4_put_deleg_lease(dp->dl_file);
         nfs4_put_delegation(dp);
 }
 
@@ -329,64 +317,6 @@ static struct list_head	unconf_id_hashtbl[CLIENT_HASH_SIZE];
 static struct list_head client_lru;
 static struct list_head close_lru;
 
-static void unhash_generic_stateid(struct nfs4_stateid *stp)
-{
-        list_del(&stp->st_hash);
-        list_del(&stp->st_perfile);
-        list_del(&stp->st_perstateowner);
-}
-
-static void free_generic_stateid(struct nfs4_stateid *stp)
-{
-        put_nfs4_file(stp->st_file);
-        kmem_cache_free(stateid_slab, stp);
-}
-
-static void release_lock_stateid(struct nfs4_stateid *stp)
-{
-        struct file *file;
-
-        unhash_generic_stateid(stp);
-        file = find_any_file(stp->st_file);
-        if (file)
-                locks_remove_posix(file, (fl_owner_t)stp->st_stateowner);
-        free_generic_stateid(stp);
-}
-
-static void unhash_lockowner(struct nfs4_stateowner *sop)
-{
-        struct nfs4_stateid *stp;
-
-        list_del(&sop->so_idhash);
-        list_del(&sop->so_strhash);
-        list_del(&sop->so_perstateid);
-        while (!list_empty(&sop->so_stateids)) {
-                stp = list_first_entry(&sop->so_stateids,
-                                struct nfs4_stateid, st_perstateowner);
-                release_lock_stateid(stp);
-        }
-}
-
-static void release_lockowner(struct nfs4_stateowner *sop)
-{
-        unhash_lockowner(sop);
-        nfs4_put_stateowner(sop);
-}
-
-static void
-release_stateid_lockowners(struct nfs4_stateid *open_stp)
-{
-        struct nfs4_stateowner *lock_sop;
-
-        while (!list_empty(&open_stp->st_lockowners)) {
-                lock_sop = list_entry(open_stp->st_lockowners.next,
-                                struct nfs4_stateowner, so_perstateid);
-                /* list_del(&open_stp->st_lockowners);  */
-                BUG_ON(lock_sop->so_is_open_owner);
-                release_lockowner(lock_sop);
-        }
-}
-
 /*
  * 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
@@ -459,13 +389,74 @@ static int nfs4_access_bmap_to_omode(struct nfs4_stateid *stp)
         return nfs4_access_to_omode(access);
 }
 
-static void release_open_stateid(struct nfs4_stateid *stp)
+static void unhash_generic_stateid(struct nfs4_stateid *stp)
+{
+        list_del(&stp->st_hash);
+        list_del(&stp->st_perfile);
+        list_del(&stp->st_perstateowner);
+}
+
+static void free_generic_stateid(struct nfs4_stateid *stp)
 {
-        int oflag = nfs4_access_bmap_to_omode(stp);
+        int oflag;
+
+        if (stp->st_access_bmap) {
+                oflag = nfs4_access_bmap_to_omode(stp);
+                nfs4_file_put_access(stp->st_file, oflag);
+        }
+        put_nfs4_file(stp->st_file);
+        kmem_cache_free(stateid_slab, stp);
+}
+
+static void release_lock_stateid(struct nfs4_stateid *stp)
+{
+        struct file *file;
 
         unhash_generic_stateid(stp);
+        file = find_any_file(stp->st_file);
+        if (file)
+                locks_remove_posix(file, (fl_owner_t)stp->st_stateowner);
+        free_generic_stateid(stp);
+}
+
+static void unhash_lockowner(struct nfs4_stateowner *sop)
+{
+        struct nfs4_stateid *stp;
+
+        list_del(&sop->so_idhash);
+        list_del(&sop->so_strhash);
+        list_del(&sop->so_perstateid);
+        while (!list_empty(&sop->so_stateids)) {
+                stp = list_first_entry(&sop->so_stateids,
+                                struct nfs4_stateid, st_perstateowner);
+                release_lock_stateid(stp);
+        }
+}
+
+static void release_lockowner(struct nfs4_stateowner *sop)
+{
+        unhash_lockowner(sop);
+        nfs4_put_stateowner(sop);
+}
+
+static void
+release_stateid_lockowners(struct nfs4_stateid *open_stp)
+{
+        struct nfs4_stateowner *lock_sop;
+
+        while (!list_empty(&open_stp->st_lockowners)) {
+                lock_sop = list_entry(open_stp->st_lockowners.next,
+                                struct nfs4_stateowner, so_perstateid);
+                /* list_del(&open_stp->st_lockowners);  */
+                BUG_ON(lock_sop->so_is_open_owner);
+                release_lockowner(lock_sop);
+        }
+}
+
+static void release_open_stateid(struct nfs4_stateid *stp)
+{
+        unhash_generic_stateid(stp);
         release_stateid_lockowners(stp);
-        nfs4_file_put_access(stp->st_file, oflag);
         free_generic_stateid(stp);
 }
 
@@ -535,171 +526,279 @@ gen_sessionid(struct nfsd4_session *ses)
  */
 #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]);
+}
+
 /*
- * Give the client the number of ca_maxresponsesize_cached slots it
- * requests, of size bounded by NFSD_SLOT_CACHE_SIZE,
- * NFSD_MAX_MEM_PER_SESSION, and nfsd_drc_max_mem. Do not allow more
- * than NFSD_MAX_SLOTS_PER_SESSION.
- *
- * If we run out of reserved DRC memory we should (up to a point)
+ * 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
  * rooom for new connections. For now we just fail the create session.
  */
-static int set_forechannel_drc_size(struct nfsd4_channel_attrs *fchan)
+static int nfsd4_get_drc_mem(int slotsize, u32 num)
 {
-        int mem, size = fchan->maxresp_cached;
+        int avail;
 
-        if (fchan->maxreqs < 1)
-                return nfserr_inval;
+        num = min_t(u32, num, NFSD_MAX_SLOTS_PER_SESSION);
 
-        if (size < NFSD_MIN_HDR_SEQ_SZ)
-                size = NFSD_MIN_HDR_SEQ_SZ;
-        size -= NFSD_MIN_HDR_SEQ_SZ;
-        if (size > NFSD_SLOT_CACHE_SIZE)
-                size = NFSD_SLOT_CACHE_SIZE;
-
-        /* bound the maxreqs by NFSD_MAX_MEM_PER_SESSION */
-        mem = fchan->maxreqs * size;
-        if (mem > NFSD_MAX_MEM_PER_SESSION) {
-                fchan->maxreqs = NFSD_MAX_MEM_PER_SESSION / size;
-                if (fchan->maxreqs > NFSD_MAX_SLOTS_PER_SESSION)
-                        fchan->maxreqs = NFSD_MAX_SLOTS_PER_SESSION;
-                mem = fchan->maxreqs * size;
-        }
+        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);
-        /* bound the total session drc memory ussage */
-        if (mem + nfsd_drc_mem_used > nfsd_drc_max_mem) {
-                fchan->maxreqs = (nfsd_drc_max_mem - nfsd_drc_mem_used) / size;
-                mem = fchan->maxreqs * size;
-        }
-        nfsd_drc_mem_used += mem;
+        nfsd_drc_mem_used -= slotsize * num;
         spin_unlock(&nfsd_drc_lock);
+}
 
-        if (fchan->maxreqs == 0)
-                return nfserr_jukebox;
+static struct nfsd4_session *alloc_session(int slotsize, int numslots)
+{
+        struct nfsd4_session *new;
+        int mem, i;
 
-        fchan->maxresp_cached = size + NFSD_MIN_HDR_SEQ_SZ;
-        return 0;
+        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;
 }
 
-/*
- * fchan holds the client values on input, and the server values on output
- * sv_max_mesg is the maximum payload plus one page for overhead.
- */
-static int init_forechannel_attrs(struct svc_rqst *rqstp,
-                                  struct nfsd4_channel_attrs *session_fchan,
-                                  struct nfsd4_channel_attrs *fchan)
+static void init_forechannel_attrs(struct nfsd4_channel_attrs *new, struct nfsd4_channel_attrs *req, int numslots, int slotsize)
 {
-        int status = 0;
-        __u32   maxcount = nfsd_serv->sv_max_mesg;
+        u32 maxrpc = nfsd_serv->sv_max_mesg;
 
-        /* headerpadsz set to zero in encode routine */
+        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);
+}
 
-        /* Use the client's max request and max response size if possible */
-        if (fchan->maxreq_sz > maxcount)
-                fchan->maxreq_sz = maxcount;
-        session_fchan->maxreq_sz = fchan->maxreq_sz;
+static void free_conn(struct nfsd4_conn *c)
+{
+        svc_xprt_put(c->cn_xprt);
+        kfree(c);
+}
 
-        if (fchan->maxresp_sz > maxcount)
-                fchan->maxresp_sz = maxcount;
-        session_fchan->maxresp_sz = fchan->maxresp_sz;
+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;
 
-        /* Use the client's maxops if possible */
-        if (fchan->maxops > NFSD_MAX_OPS_PER_COMPOUND)
-                fchan->maxops = NFSD_MAX_OPS_PER_COMPOUND;
-        session_fchan->maxops = fchan->maxops;
+        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);
+}
 
-        /* FIXME: Error means no more DRC pages so the server should
-         * recover pages from existing sessions. For now fail session
-         * creation.
-         */
-        status = set_forechannel_drc_size(fchan);
+static struct nfsd4_conn *alloc_conn(struct svc_rqst *rqstp, u32 flags)
+{
+        struct nfsd4_conn *conn;
 
-        session_fchan->maxresp_cached = fchan->maxresp_cached;
-        session_fchan->maxreqs = fchan->maxreqs;
+        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;
+}
 
-        dprintk("%s status %d\n", __func__, status);
-        return status;
+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
-free_session_slots(struct nfsd4_session *ses)
+static void nfsd4_hash_conn(struct nfsd4_conn *conn, struct nfsd4_session *ses)
 {
-        int i;
+        struct nfs4_client *clp = ses->se_client;
 
-        for (i = 0; i < ses->se_fchannel.maxreqs; i++)
-                kfree(ses->se_slots[i]);
+        spin_lock(&clp->cl_lock);
+        __nfsd4_hash_conn(conn, ses);
+        spin_unlock(&clp->cl_lock);
 }
 
-/*
- * 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)
+static int nfsd4_register_conn(struct nfsd4_conn *conn)
 {
-        return ca->maxresp_cached - NFSD_MIN_HDR_SEQ_SZ;
+        conn->cn_xpt_user.callback = nfsd4_conn_lost;
+        return register_xpt_user(conn->cn_xprt, &conn->cn_xpt_user);
 }
 
-static int
-alloc_init_session(struct svc_rqst *rqstp, struct nfs4_client *clp,
-                   struct nfsd4_create_session *cses)
+static __be32 nfsd4_new_conn(struct svc_rqst *rqstp, struct nfsd4_session *ses, u32 dir)
 {
-        struct nfsd4_session *new, tmp;
-        struct nfsd4_slot *sp;
-        int idx, slotsize, cachesize, i;
-        int status;
+        struct nfsd4_conn *conn;
+        int ret;
 
-        memset(&tmp, 0, sizeof(tmp));
+        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;
+}
 
-        /* FIXME: For now, we just accept the client back channel attributes. */
-        tmp.se_bchannel = cses->back_channel;
-        status = init_forechannel_attrs(rqstp, &tmp.se_fchannel,
-                                        &cses->fore_channel);
-        if (status)
-                goto out;
+static __be32 nfsd4_new_conn_from_crses(struct svc_rqst *rqstp, struct nfsd4_session *ses)
+{
+        u32 dir = NFS4_CDFC4_FORE;
 
-        BUILD_BUG_ON(NFSD_MAX_SLOTS_PER_SESSION * sizeof(struct nfsd4_slot)
-                     + sizeof(struct nfsd4_session) > PAGE_SIZE);
+        if (ses->se_flags & SESSION4_BACK_CHAN)
+                dir |= NFS4_CDFC4_BACK;
 
-        status = nfserr_jukebox;
-        /* allocate struct nfsd4_session and slot table pointers in one piece */
-        slotsize = tmp.se_fchannel.maxreqs * sizeof(struct nfsd4_slot *);
-        new = kzalloc(sizeof(*new) + slotsize, GFP_KERNEL);
-        if (!new)
-                goto out;
+        return nfsd4_new_conn(rqstp, ses, dir);
+}
 
-        memcpy(new, &tmp, sizeof(*new));
+/* 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;
 
-        /* allocate each struct nfsd4_slot and data cache in one piece */
-        cachesize = slot_bytes(&new->se_fchannel);
-        for (i = 0; i < new->se_fchannel.maxreqs; i++) {
-                sp = kzalloc(sizeof(*sp) + cachesize, GFP_KERNEL);
-                if (!sp)
-                        goto out_free;
-                new->se_slots[i] = sp;
+        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);
-        idx = hash_sessionid(&new->se_sessionid);
-        memcpy(clp->cl_sessionid.data, new->se_sessionid.data,
-               NFS4_MAX_SESSIONID_LEN);
 
+        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 = nfs_ok;
-out:
-        return status;
-out_free:
-        free_session_slots(new);
-        kfree(new);
-        goto out;
+        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 */
@@ -728,22 +827,9 @@ 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);
-}
-
-void
-free_session(struct kref *kref)
-{
-        struct nfsd4_session *ses;
-        int mem;
-
-        ses = container_of(kref, struct nfsd4_session, se_ref);
-        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);
+        spin_unlock(&ses->se_client->cl_lock);
 }
 
 /* must be called under the client_lock */
@@ -812,6 +898,13 @@ static struct nfs4_client *alloc_client(struct xdr_netobj name)
 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);
@@ -838,15 +931,14 @@ release_session_client(struct nfsd4_session *session)
 static inline void
 unhash_client_locked(struct nfs4_client *clp)
 {
+        struct nfsd4_session *ses;
+
         mark_client_expired(clp);
         list_del(&clp->cl_lru);
-        while (!list_empty(&clp->cl_sessions)) {
-                struct nfsd4_session  *ses;
-                ses = list_entry(clp->cl_sessions.next, struct nfsd4_session,
-                                 se_perclnt);
-                unhash_session(ses);
-                nfsd4_put_session(ses);
-        }
+        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
@@ -860,8 +952,6 @@ expire_client(struct nfs4_client *clp)
         spin_lock(&recall_lock);
         while (!list_empty(&clp->cl_delegations)) {
                 dp = list_entry(clp->cl_delegations.next, struct nfs4_delegation, dl_perclnt);
-                dprintk("NFSD: expire client. dp %p, fp %p\n", dp,
-                                dp->dl_flock);
                 list_del_init(&dp->dl_perclnt);
                 list_move(&dp->dl_recall_lru, &reaplist);
         }
@@ -875,7 +965,7 @@ expire_client(struct nfs4_client *clp)
                 sop = list_entry(clp->cl_openowners.next, struct nfs4_stateowner, so_perclient);
                 release_openowner(sop);
         }
-        nfsd4_set_callback_client(clp, NULL);
+        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);
@@ -960,6 +1050,8 @@ static struct nfs4_client *create_client(struct xdr_netobj name, char *recdir,
         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);
@@ -971,13 +1063,15 @@ static struct nfs4_client *create_client(struct xdr_netobj name, char *recdir,
 
         memcpy(clp->cl_recdir, recdir, HEXDIR_LEN);
         atomic_set(&clp->cl_refcount, 0);
-        atomic_set(&clp->cl_cb_set, 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_sessions);
         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");
@@ -986,7 +1080,7 @@ static struct nfs4_client *create_client(struct xdr_netobj name, char *recdir,
         clp->cl_flavor = rqstp->rq_flavor;
         copy_cred(&clp->cl_cred, &rqstp->rq_cred);
         gen_confirm(clp);
-
+        clp->cl_cb_session = NULL;
         return clp;
 }
 
@@ -1051,54 +1145,55 @@ find_unconfirmed_client(clientid_t *clid)
         return NULL;
 }
 
-/*
- * Return 1 iff clp's clientid establishment method matches the use_exchange_id
- * parameter. Matching is based on the fact the at least one of the
- * EXCHGID4_FLAG_USE_{NON_PNFS,PNFS_MDS,PNFS_DS} flags must be set for v4.1
- *
- * FIXME: we need to unify the clientid namespaces for nfsv4.x
- * and correctly deal with client upgrade/downgrade in EXCHANGE_ID
- * and SET_CLIENTID{,_CONFIRM}
- */
-static inline int
-match_clientid_establishment(struct nfs4_client *clp, bool use_exchange_id)
+static bool clp_used_exchangeid(struct nfs4_client *clp)
 {
-        bool has_exchange_flags = (clp->cl_exchange_flags != 0);
-        return use_exchange_id == has_exchange_flags;
-}
+        return clp->cl_exchange_flags != 0;
+} 
 
 static struct nfs4_client *
-find_confirmed_client_by_str(const char *dname, unsigned int hashval,
-                             bool use_exchange_id)
+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) &&
-                    match_clientid_establishment(clp, use_exchange_id))
+                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,
-                               bool use_exchange_id)
+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) &&
-                    match_clientid_establishment(clp, use_exchange_id))
+                if (same_name(clp->cl_recdir, dname))
                         return clp;
         }
         return NULL;
 }
 
+static void rpc_svcaddr2sockaddr(struct sockaddr *sa, unsigned short family, union svc_addr_u *svcaddr)
+{
+        switch (family) {
+        case AF_INET:
+                ((struct sockaddr_in *)sa)->sin_family = AF_INET;
+                ((struct sockaddr_in *)sa)->sin_addr = svcaddr->addr;
+                return;
+        case AF_INET6:
+                ((struct sockaddr_in6 *)sa)->sin6_family = AF_INET6;
+                ((struct sockaddr_in6 *)sa)->sin6_addr = svcaddr->addr6;
+                return;
+        }
+}
+
 static void
-gen_callback(struct nfs4_client *clp, struct nfsd4_setclientid *se, u32 scopeid)
+gen_callback(struct nfs4_client *clp, struct nfsd4_setclientid *se, struct svc_rqst *rqstp)
 {
-        struct nfs4_cb_conn *cb = &clp->cl_cb_conn;
+        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 */
@@ -1111,24 +1206,24 @@ gen_callback(struct nfs4_client *clp, struct nfsd4_setclientid *se, u32 scopeid)
         else
                 goto out_err;
 
-        cb->cb_addrlen = rpc_uaddr2sockaddr(se->se_callback_addr_val,
+        conn->cb_addrlen = rpc_uaddr2sockaddr(se->se_callback_addr_val,
                                             se->se_callback_addr_len,
-                                            (struct sockaddr *) &cb->cb_addr,
-                                            sizeof(cb->cb_addr));
+                                            (struct sockaddr *)&conn->cb_addr,
+                                            sizeof(conn->cb_addr));
 
-        if (!cb->cb_addrlen || cb->cb_addr.ss_family != expected_family)
+        if (!conn->cb_addrlen || conn->cb_addr.ss_family != expected_family)
                 goto out_err;
 
-        if (cb->cb_addr.ss_family == AF_INET6)
-                ((struct sockaddr_in6 *) &cb->cb_addr)->sin6_scope_id = scopeid;
+        if (conn->cb_addr.ss_family == AF_INET6)
+                ((struct sockaddr_in6 *)&conn->cb_addr)->sin6_scope_id = scopeid;
 
-        cb->cb_minorversion = 0;
-        cb->cb_prog = se->se_callback_prog;
-        cb->cb_ident = se->se_callback_ident;
+        conn->cb_prog = se->se_callback_prog;
+        conn->cb_ident = se->se_callback_ident;
+        rpc_svcaddr2sockaddr((struct sockaddr *)&conn->cb_saddr, expected_family, &rqstp->rq_daddr);
         return;
 out_err:
-        cb->cb_addr.ss_family = AF_UNSPEC;
-        cb->cb_addrlen = 0;
+        conn->cb_addr.ss_family = AF_UNSPEC;
+        conn->cb_addrlen = 0;
         dprintk(KERN_INFO "NFSD: this client (clientid %08x/%08x) "
                 "will not receive delegations\n",
                 clp->cl_clientid.cl_boot, clp->cl_clientid.cl_id);
@@ -1264,7 +1359,7 @@ nfsd4_exchange_id(struct svc_rqst *rqstp,
         case SP4_NONE:
                 break;
         case SP4_SSV:
-                return nfserr_encr_alg_unsupp;
+                return nfserr_serverfault;
         default:
                 BUG();                          /* checked by xdr code */
         case SP4_MACH_CRED:
@@ -1281,8 +1376,12 @@ nfsd4_exchange_id(struct svc_rqst *rqstp,
         nfs4_lock_state();
         status = nfs_ok;
 
-        conf = find_confirmed_client_by_str(dname, strhashval, true);
+        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) {
@@ -1323,7 +1422,7 @@ nfsd4_exchange_id(struct svc_rqst *rqstp,
                 goto out;
         }
 
-        unconf  = find_unconfirmed_client_by_str(dname, strhashval, true);
+        unconf  = find_unconfirmed_client_by_str(dname, strhashval);
         if (unconf) {
                 /*
                  * Possible retry or client restart.  Per 18.35.4 case 4,
@@ -1415,9 +1514,14 @@ nfsd4_create_session(struct svc_rqst *rqstp,
 {
         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);
@@ -1438,7 +1542,6 @@ nfsd4_create_session(struct svc_rqst *rqstp,
                                 cs_slot->sl_seqid, cr_ses->seqid);
                         goto out;
                 }
-                cs_slot->sl_seqid++;
         } else if (unconf) {
                 if (!same_creds(&unconf->cl_cred, &rqstp->rq_cred) ||
                     !rpc_cmp_addr(sa, (struct sockaddr *) &unconf->cl_addr)) {
@@ -1451,25 +1554,10 @@ nfsd4_create_session(struct svc_rqst *rqstp,
                 if (status) {
                         /* an unconfirmed replay returns misordered */
                         status = nfserr_seq_misordered;
-                        goto out_cache;
+                        goto out;
                 }
 
-                cs_slot->sl_seqid++; /* from 0 to 1 */
-                move_to_confirmed(unconf);
-
-                if (cr_ses->flags & SESSION4_BACK_CHAN) {
-                        unconf->cl_cb_conn.cb_xprt = rqstp->rq_xprt;
-                        svc_xprt_get(rqstp->rq_xprt);
-                        rpc_copy_addr(
-                                (struct sockaddr *)&unconf->cl_cb_conn.cb_addr,
-                                sa);
-                        unconf->cl_cb_conn.cb_addrlen = svc_addr_len(sa);
-                        unconf->cl_cb_conn.cb_minorversion =
-                                cstate->minorversion;
-                        unconf->cl_cb_conn.cb_prog = cr_ses->callback_prog;
-                        unconf->cl_cb_seq_nr = 1;
-                        nfsd4_probe_callback(unconf, &unconf->cl_cb_conn);
-                }
+                confirm_me = true;
                 conf = unconf;
         } else {
                 status = nfserr_stale_clientid;
@@ -1477,22 +1565,32 @@ nfsd4_create_session(struct svc_rqst *rqstp,
         }
 
         /*
+         * XXX: we should probably set this at creation time, and check
+         * for consistent minorversion use throughout:
+         */
+        conf->cl_minorversion = 1;
+        /*
          * We do not support RDMA or persistent sessions
          */
         cr_ses->flags &= ~SESSION4_PERSIST;
         cr_ses->flags &= ~SESSION4_RDMA;
 
-        status = alloc_init_session(rqstp, conf, cr_ses);
-        if (status)
+        status = nfserr_jukebox;
+        new = alloc_init_session(rqstp, conf, cr_ses);
+        if (!new)
                 goto out;
-
-        memcpy(cr_ses->sessionid.data, conf->cl_sessionid.data,
+        status = nfs_ok;
+        memcpy(cr_ses->sessionid.data, new->se_sessionid.data,
                NFS4_MAX_SESSIONID_LEN);
+        memcpy(&cr_ses->fore_channel, &new->se_fchannel,
+                sizeof(struct nfsd4_channel_attrs));
+        cs_slot->sl_seqid++;
         cr_ses->seqid = cs_slot->sl_seqid;
 
-out_cache:
         /* cache solo and embedded create sessions under the state lock */
         nfsd4_cache_create_session(cr_ses, cs_slot, status);
+        if (confirm_me)
+                move_to_confirmed(conf);
 out:
         nfs4_unlock_state();
         dprintk("%s returns %d\n", __func__, ntohl(status));
@@ -1507,6 +1605,46 @@ static bool nfsd4_last_compound_op(struct svc_rqst *rqstp)
         return argp->opcnt == resp->opcnt;
 }
 
+static __be32 nfsd4_map_bcts_dir(u32 *dir)
+{
+        switch (*dir) {
+        case NFS4_CDFC4_FORE:
+        case NFS4_CDFC4_BACK:
+                return nfs_ok;
+        case NFS4_CDFC4_FORE_OR_BOTH:
+        case NFS4_CDFC4_BACK_OR_BOTH:
+                *dir = NFS4_CDFC4_BOTH;
+                return nfs_ok;
+        };
+        return nfserr_inval;
+}
+
+__be32 nfsd4_bind_conn_to_session(struct svc_rqst *rqstp,
+                     struct nfsd4_compound_state *cstate,
+                     struct nfsd4_bind_conn_to_session *bcts)
+{
+        __be32 status;
+
+        if (!nfsd4_last_compound_op(rqstp))
+                return nfserr_not_only_op;
+        spin_lock(&client_lock);
+        cstate->session = find_in_sessionid_hashtbl(&bcts->sessionid);
+        /* Sorta weird: we only need the refcnt'ing because new_conn acquires
+         * client_lock iself: */
+        if (cstate->session) {
+                nfsd4_get_session(cstate->session);
+                atomic_inc(&cstate->session->se_client->cl_refcount);
+        }
+        spin_unlock(&client_lock);
+        if (!cstate->session)
+                return nfserr_badsession;
+
+        status = nfsd4_map_bcts_dir(&bcts->dir);
+        if (!status)
+                nfsd4_new_conn(rqstp, cstate->session, bcts->dir);
+        return status;
+}
+
 static bool nfsd4_compound_in_session(struct nfsd4_session *session, struct nfs4_sessionid *sid)
 {
         if (!session)
@@ -1545,9 +1683,11 @@ nfsd4_destroy_session(struct svc_rqst *r,
         spin_unlock(&client_lock);
 
         nfs4_lock_state();
-        /* wait for callbacks */
-        nfsd4_set_callback_client(ses->se_client, NULL);
+        nfsd4_probe_callback_sync(ses->se_client);
         nfs4_unlock_state();
+
+        nfsd4_del_conns(ses);
+
         nfsd4_put_session(ses);
         status = nfs_ok;
 out:
@@ -1555,6 +1695,47 @@ out:
         return status;
 }
 
+static struct nfsd4_conn *__nfsd4_find_conn(struct svc_xprt *xpt, struct nfsd4_session *s)
+{
+        struct nfsd4_conn *c;
+
+        list_for_each_entry(c, &s->se_conns, cn_persession) {
+                if (c->cn_xprt == xpt) {
+                        return c;
+                }
+        }
+        return NULL;
+}
+
+static void nfsd4_sequence_check_conn(struct nfsd4_conn *new, struct nfsd4_session *ses)
+{
+        struct nfs4_client *clp = ses->se_client;
+        struct nfsd4_conn *c;
+        int ret;
+
+        spin_lock(&clp->cl_lock);
+        c = __nfsd4_find_conn(new->cn_xprt, ses);
+        if (c) {
+                spin_unlock(&clp->cl_lock);
+                free_conn(new);
+                return;
+        }
+        __nfsd4_hash_conn(new, ses);
+        spin_unlock(&clp->cl_lock);
+        ret = nfsd4_register_conn(new);
+        if (ret)
+                /* oops; xprt is already down: */
+                nfsd4_conn_lost(&new->cn_xpt_user);
+        return;
+}
+
+static bool nfsd4_session_too_many_ops(struct svc_rqst *rqstp, struct nfsd4_session *session)
+{
+        struct nfsd4_compoundargs *args = rqstp->rq_argp;
+
+        return args->opcnt > session->se_fchannel.maxops;
+}
+
 __be32
 nfsd4_sequence(struct svc_rqst *rqstp,
                struct nfsd4_compound_state *cstate,
@@ -1563,17 +1744,30 @@ nfsd4_sequence(struct svc_rqst *rqstp,
         struct nfsd4_compoundres *resp = rqstp->rq_resp;
         struct nfsd4_session *session;
         struct nfsd4_slot *slot;
+        struct nfsd4_conn *conn;
         int status;
 
         if (resp->opcnt != 1)
                 return nfserr_sequence_pos;
 
+        /*
+         * Will be either used or freed by nfsd4_sequence_check_conn
+         * below.
+         */
+        conn = alloc_conn(rqstp, NFS4_CDFC4_FORE);
+        if (!conn)
+                return nfserr_jukebox;
+
         spin_lock(&client_lock);
         status = nfserr_badsession;
         session = find_in_sessionid_hashtbl(&seq->sessionid);
         if (!session)
                 goto out;
 
+        status = nfserr_too_many_ops;
+        if (nfsd4_session_too_many_ops(rqstp, session))
+                goto out;
+
         status = nfserr_badslot;
         if (seq->slotid >= session->se_fchannel.maxreqs)
                 goto out;
@@ -1599,6 +1793,9 @@ nfsd4_sequence(struct svc_rqst *rqstp,
         if (status)
                 goto out;
 
+        nfsd4_sequence_check_conn(conn, session);
+        conn = NULL;
+
         /* Success! bump slot seqid */
         slot->sl_inuse = true;
         slot->sl_seqid = seq->seqid;
@@ -1610,9 +1807,14 @@ nfsd4_sequence(struct svc_rqst *rqstp,
 out:
         /* Hold a session reference until done processing the compound. */
         if (cstate->session) {
+                struct nfs4_client *clp = session->se_client;
+
                 nfsd4_get_session(cstate->session);
-                atomic_inc(&session->se_client->cl_refcount);
+                atomic_inc(&clp->cl_refcount);
+                if (clp->cl_cb_state == NFSD4_CB_DOWN)
+                        seq->status_flags |= SEQ4_STATUS_CB_PATH_DOWN;
         }
+        kfree(conn);
         spin_unlock(&client_lock);
         dprintk("%s: return %d\n", __func__, ntohl(status));
         return status;
@@ -1621,6 +1823,8 @@ out:
 __be32
 nfsd4_reclaim_complete(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate, struct nfsd4_reclaim_complete *rc)
 {
+        int status = 0;
+
         if (rc->rca_one_fs) {
                 if (!cstate->current_fh.fh_dentry)
                         return nfserr_nofilehandle;
@@ -1630,9 +1834,14 @@ nfsd4_reclaim_complete(struct svc_rqst *rqstp, struct nfsd4_compound_state *csta
                  */
                  return nfs_ok;
         }
+
         nfs4_lock_state();
-        if (is_client_expired(cstate->session->se_client)) {
-                nfs4_unlock_state();
+        status = nfserr_complete_already;
+        if (cstate->session->se_client->cl_firststate)
+                goto out;
+
+        status = nfserr_stale_clientid;
+        if (is_client_expired(cstate->session->se_client))
                 /*
                  * The following error isn't really legal.
                  * But we only get here if the client just explicitly
@@ -1640,18 +1849,19 @@ nfsd4_reclaim_complete(struct svc_rqst *rqstp, struct nfsd4_compound_state *csta
                  * error it gets back on an operation for the dead
                  * client.
                  */
-                return nfserr_stale_clientid;
-        }
+                goto out;
+
+        status = nfs_ok;
         nfsd4_create_clid_dir(cstate->session->se_client);
+out:
         nfs4_unlock_state();
-        return nfs_ok;
+        return status;
 }
 
 __be32
 nfsd4_setclientid(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
                   struct nfsd4_setclientid *setclid)
 {
-        struct sockaddr         *sa = svc_addr(rqstp);
         struct xdr_netobj       clname = { 
                 .len = setclid->se_namelen,
                 .data = setclid->se_name,
@@ -1677,10 +1887,12 @@ nfsd4_setclientid(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
         strhashval = clientstr_hashval(dname);
 
         nfs4_lock_state();
-        conf = find_confirmed_client_by_str(dname, strhashval, false);
+        conf = find_confirmed_client_by_str(dname, strhashval);
         if (conf) {
                 /* RFC 3530 14.2.33 CASE 0: */
                 status = nfserr_clid_inuse;
+                if (clp_used_exchangeid(conf))
+                        goto out;
                 if (!same_creds(&conf->cl_cred, &rqstp->rq_cred)) {
                         char addr_str[INET6_ADDRSTRLEN];
                         rpc_ntop((struct sockaddr *) &conf->cl_addr, addr_str,
@@ -1695,7 +1907,7 @@ nfsd4_setclientid(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
          * has a description of SETCLIENTID request processing consisting
          * of 5 bullet points, labeled as CASE0 - CASE4 below.
          */
-        unconf = find_unconfirmed_client_by_str(dname, strhashval, false);
+        unconf = find_unconfirmed_client_by_str(dname, strhashval);
         status = nfserr_resource;
         if (!conf) {
                 /*
@@ -1747,7 +1959,12 @@ nfsd4_setclientid(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
                         goto out;
                 gen_clid(new);
         }
-        gen_callback(new, setclid, rpc_get_scope_id(sa));
+        /*
+         * XXX: we should probably set this at creation time, and check
+         * for consistent minorversion use throughout:
+         */
+        new->cl_minorversion = 0;
+        gen_callback(new, setclid, rqstp);
         add_to_unconfirmed(new, strhashval);
         setclid->se_clientid.cl_boot = new->cl_clientid.cl_boot;
         setclid->se_clientid.cl_id = new->cl_clientid.cl_id;
@@ -1806,8 +2023,8 @@ nfsd4_setclientid_confirm(struct svc_rqst *rqstp,
                 if (!same_creds(&conf->cl_cred, &unconf->cl_cred))
                         status = nfserr_clid_inuse;
                 else {
-                        atomic_set(&conf->cl_cb_set, 0);
-                        nfsd4_probe_callback(conf, &unconf->cl_cb_conn);
+                        nfsd4_change_callback(conf, &unconf->cl_cb_conn);
+                        nfsd4_probe_callback(conf);
                         expire_client(unconf);
                         status = nfs_ok;
 
@@ -1834,14 +2051,14 @@ nfsd4_setclientid_confirm(struct svc_rqst *rqstp,
                         unsigned int hash =
                                 clientstr_hashval(unconf->cl_recdir);
                         conf = find_confirmed_client_by_str(unconf->cl_recdir,
-                                                            hash, false);
+                                                            hash);
                         if (conf) {
                                 nfsd4_remove_clid_dir(conf);
                                 expire_client(conf);
                         }
                         move_to_confirmed(unconf);
                         conf = unconf;
-                        nfsd4_probe_callback(conf, &conf->cl_cb_conn);
+                        nfsd4_probe_callback(conf);
                         status = nfs_ok;
                 }
         } else if ((!conf || (conf && !same_verf(&conf->cl_confirm, &confirm)))
@@ -1877,6 +2094,7 @@ alloc_init_file(struct inode *ino)
                 fp->fi_inode = igrab(ino);
                 fp->fi_id = current_fileid++;
                 fp->fi_had_conflict = false;
+                fp->fi_lease = NULL;
                 memset(fp->fi_fds, 0, sizeof(fp->fi_fds));
                 memset(fp->fi_access, 0, sizeof(fp->fi_access));
                 spin_lock(&recall_lock);
@@ -2128,23 +2346,8 @@ nfs4_file_downgrade(struct nfs4_file *fp, unsigned int share_access)
                 nfs4_file_put_access(fp, O_RDONLY);
 }
 
-/*
- * Spawn a thread to perform a recall on the delegation represented
- * by the lease (file_lock)
- *
- * Called from break_lease() with lock_kernel() held.
- * Note: we assume break_lease will only call this *once* for any given
- * lease.
- */
-static
-void nfsd_break_deleg_cb(struct file_lock *fl)
+static void nfsd_break_one_deleg(struct nfs4_delegation *dp)
 {
-        struct nfs4_delegation *dp = (struct nfs4_delegation *)fl->fl_owner;
-
-        dprintk("NFSD nfsd_break_deleg_cb: dp %p fl %p\n",dp,fl);
-        if (!dp)
-                return;
-
         /* We're assuming the state code never drops its reference
          * without first removing the lease.  Since we're in this lease
          * callback (and since the lease code is serialized by the kernel
@@ -2152,75 +2355,37 @@ void nfsd_break_deleg_cb(struct file_lock *fl)
          * it's safe to take a reference: */
         atomic_inc(&dp->dl_count);
 
-        spin_lock(&recall_lock);
         list_add_tail(&dp->dl_recall_lru, &del_recall_lru);
-        spin_unlock(&recall_lock);
 
-        /* only place dl_time is set. protected by lock_kernel*/
+        /* only place dl_time is set. protected by lock_flocks*/
         dp->dl_time = get_seconds();
 
-        /*
-         * We don't want the locks code to timeout the lease for us;
-         * we'll remove it ourself if the delegation isn't returned
-         * in time.
-         */
-        fl->fl_break_time = 0;
-
-        dp->dl_file->fi_had_conflict = true;
         nfsd4_cb_recall(dp);
 }
 
-/*
- * The file_lock is being reapd.
- *
- * Called by locks_free_lock() with lock_kernel() held.
- */
-static
-void nfsd_release_deleg_cb(struct file_lock *fl)
-{
-        struct nfs4_delegation *dp = (struct nfs4_delegation *)fl->fl_owner;
-
-        dprintk("NFSD nfsd_release_deleg_cb: fl %p dp %p dl_count %d\n", fl,dp, atomic_read(&dp->dl_count));
-
-        if (!(fl->fl_flags & FL_LEASE) || !dp)
-                return;
-        dp->dl_flock = NULL;
-}
-
-/*
- * Set the delegation file_lock back pointer.
- *
- * Called from setlease() with lock_kernel() held.
- */
-static
-void nfsd_copy_lock_deleg_cb(struct file_lock *new, struct file_lock *fl)
-{
-        struct nfs4_delegation *dp = (struct nfs4_delegation *)new->fl_owner;
-
-        dprintk("NFSD: nfsd_copy_lock_deleg_cb: new fl %p dp %p\n", new, dp);
-        if (!dp)
-                return;
-        dp->dl_flock = new;
-}
-
-/*
- * Called from setlease() with lock_kernel() held
- */
-static
-int nfsd_same_client_deleg_cb(struct file_lock *onlist, struct file_lock *try)
+/* Called from break_lease() with lock_flocks() held. */
+static void nfsd_break_deleg_cb(struct file_lock *fl)
 {
-        struct nfs4_delegation *onlistd =
-                (struct nfs4_delegation *)onlist->fl_owner;
-        struct nfs4_delegation *tryd =
-                (struct nfs4_delegation *)try->fl_owner;
+        struct nfs4_file *fp = (struct nfs4_file *)fl->fl_owner;
+        struct nfs4_delegation *dp;
 
-        if (onlist->fl_lmops != try->fl_lmops)
-                return 0;
+        BUG_ON(!fp);
+        /* We assume break_lease is only called once per lease: */
+        BUG_ON(fp->fi_had_conflict);
+        /*
+         * We don't want the locks code to timeout the lease for us;
+         * we'll remove it ourself if a delegation isn't returned
+         * in time:
+         */
+        fl->fl_break_time = 0;
 
-        return onlistd->dl_client == tryd->dl_client;
+        spin_lock(&recall_lock);
+        fp->fi_had_conflict = true;
+        list_for_each_entry(dp, &fp->fi_delegations, dl_perfile)
+                nfsd_break_one_deleg(dp);
+        spin_unlock(&recall_lock);
 }
 
-
 static
 int nfsd_change_deleg_cb(struct file_lock **onlist, int arg)
 {
@@ -2232,9 +2397,6 @@ int nfsd_change_deleg_cb(struct file_lock **onlist, int arg)
 
 static const struct lock_manager_operations nfsd_lease_mng_ops = {
         .fl_break = nfsd_break_deleg_cb,
-        .fl_release_private = nfsd_release_deleg_cb,
-        .fl_copy_lock = nfsd_copy_lock_deleg_cb,
-        .fl_mylease = nfsd_same_client_deleg_cb,
         .fl_change = nfsd_change_deleg_cb,
 };
 
@@ -2314,14 +2476,17 @@ find_delegation_file(struct nfs4_file *fp, stateid_t *stid)
 {
         struct nfs4_delegation *dp;
 
-        list_for_each_entry(dp, &fp->fi_delegations, dl_perfile) {
-                if (dp->dl_stateid.si_stateownerid == stid->si_stateownerid)
+        spin_lock(&recall_lock);
+        list_for_each_entry(dp, &fp->fi_delegations, dl_perfile)
+                if (dp->dl_stateid.si_stateownerid == stid->si_stateownerid) {
+                        spin_unlock(&recall_lock);
                         return dp;
-        }
+                }
+        spin_unlock(&recall_lock);
         return NULL;
 }
 
-int share_access_to_flags(u32 share_access)
+static int share_access_to_flags(u32 share_access)
 {
         share_access &= ~NFS4_SHARE_WANT_MASK;
 
@@ -2401,8 +2566,6 @@ static __be32 nfs4_get_vfs_file(struct svc_rqst *rqstp, struct nfs4_file
         if (!fp->fi_fds[oflag]) {
                 status = nfsd_open(rqstp, cur_fh, S_IFREG, access,
                         &fp->fi_fds[oflag]);
-                if (status == nfserr_dropit)
-                        status = nfserr_jukebox;
                 if (status)
                         return status;
         }
@@ -2483,6 +2646,79 @@ nfs4_set_claim_prev(struct nfsd4_open *open)
         open->op_stateowner->so_client->cl_firststate = 1;
 }
 
+/* Should we give out recallable state?: */
+static bool nfsd4_cb_channel_good(struct nfs4_client *clp)
+{
+        if (clp->cl_cb_state == NFSD4_CB_UP)
+                return true;
+        /*
+         * In the sessions case, since we don't have to establish a
+         * separate connection for callbacks, we assume it's OK
+         * until we hear otherwise:
+         */
+        return clp->cl_minorversion && clp->cl_cb_state == NFSD4_CB_UNKNOWN;
+}
+
+static struct file_lock *nfs4_alloc_init_lease(struct nfs4_delegation *dp, int flag)
+{
+        struct file_lock *fl;
+
+        fl = locks_alloc_lock();
+        if (!fl)
+                return NULL;
+        locks_init_lock(fl);
+        fl->fl_lmops = &nfsd_lease_mng_ops;
+        fl->fl_flags = FL_LEASE;
+        fl->fl_type = flag == NFS4_OPEN_DELEGATE_READ? F_RDLCK: F_WRLCK;
+        fl->fl_end = OFFSET_MAX;
+        fl->fl_owner = (fl_owner_t)(dp->dl_file);
+        fl->fl_pid = current->tgid;
+        return fl;
+}
+
+static int nfs4_setlease(struct nfs4_delegation *dp, int flag)
+{
+        struct nfs4_file *fp = dp->dl_file;
+        struct file_lock *fl;
+        int status;
+
+        fl = nfs4_alloc_init_lease(dp, flag);
+        if (!fl)
+                return -ENOMEM;
+        fl->fl_file = find_readable_file(fp);
+        list_add(&dp->dl_perclnt, &dp->dl_client->cl_delegations);
+        status = vfs_setlease(fl->fl_file, fl->fl_type, &fl);
+        if (status) {
+                list_del_init(&dp->dl_perclnt);
+                locks_free_lock(fl);
+                return -ENOMEM;
+        }
+        fp->fi_lease = fl;
+        fp->fi_deleg_file = fl->fl_file;
+        get_file(fp->fi_deleg_file);
+        atomic_set(&fp->fi_delegees, 1);
+        list_add(&dp->dl_perfile, &fp->fi_delegations);
+        return 0;
+}
+
+static int nfs4_set_delegation(struct nfs4_delegation *dp, int flag)
+{
+        struct nfs4_file *fp = dp->dl_file;
+
+        if (!fp->fi_lease)
+                return nfs4_setlease(dp, flag);
+        spin_lock(&recall_lock);
+        if (fp->fi_had_conflict) {
+                spin_unlock(&recall_lock);
+                return -EAGAIN;
+        }
+        atomic_inc(&fp->fi_delegees);
+        list_add(&dp->dl_perfile, &fp->fi_delegations);
+        spin_unlock(&recall_lock);
+        list_add(&dp->dl_perclnt, &dp->dl_client->cl_delegations);
+        return 0;
+}
+
 /*
  * Attempt to hand out a delegation.
  */
@@ -2491,10 +2727,10 @@ nfs4_open_delegation(struct svc_fh *fh, struct nfsd4_open *open, struct nfs4_sta
 {
         struct nfs4_delegation *dp;
         struct nfs4_stateowner *sop = stp->st_stateowner;
-        int cb_up = atomic_read(&sop->so_client->cl_cb_set);
-        struct file_lock fl, *flp = &fl;
+        int cb_up;
         int status, flag = 0;
 
+        cb_up = nfsd4_cb_channel_good(sop->so_client);
         flag = NFS4_OPEN_DELEGATE_NONE;
         open->op_recall = 0;
         switch (open->op_claim_type) {
@@ -2522,29 +2758,11 @@ nfs4_open_delegation(struct svc_fh *fh, struct nfsd4_open *open, struct nfs4_sta
         }
 
         dp = alloc_init_deleg(sop->so_client, stp, fh, flag);
-        if (dp == NULL) {
-                flag = NFS4_OPEN_DELEGATE_NONE;
-                goto out;
-        }
-        locks_init_lock(&fl);
-        fl.fl_lmops = &nfsd_lease_mng_ops;
-        fl.fl_flags = FL_LEASE;
-        fl.fl_type = flag == NFS4_OPEN_DELEGATE_READ? F_RDLCK: F_WRLCK;
-        fl.fl_end = OFFSET_MAX;
-        fl.fl_owner =  (fl_owner_t)dp;
-        fl.fl_file = find_readable_file(stp->st_file);
-        BUG_ON(!fl.fl_file);
-        fl.fl_pid = current->tgid;
-
-        /* vfs_setlease checks to see if delegation should be handed out.
-         * the lock_manager callbacks fl_mylease and fl_change are used
-         */
-        if ((status = vfs_setlease(fl.fl_file, fl.fl_type, &flp))) {
-                dprintk("NFSD: setlease failed [%d], no delegation\n", status);
-                unhash_delegation(dp);
-                flag = NFS4_OPEN_DELEGATE_NONE;
-                goto out;
-        }
+        if (dp == NULL)
+                goto out_no_deleg;
+        status = nfs4_set_delegation(dp, flag);
+        if (status)
+                goto out_free;
 
         memcpy(&open->op_delegate_stateid, &dp->dl_stateid, sizeof(dp->dl_stateid));
 
@@ -2556,6 +2774,12 @@ out:
                         && open->op_delegate_type != NFS4_OPEN_DELEGATE_NONE)
                 dprintk("NFSD: WARNING: refusing delegation reclaim\n");
         open->op_delegate_type = flag;
+        return;
+out_free:
+        nfs4_put_delegation(dp);
+out_no_deleg:
+        flag = NFS4_OPEN_DELEGATE_NONE;
+        goto out;
 }
 
 /*
@@ -2674,7 +2898,7 @@ nfsd4_renew(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
         renew_client(clp);
         status = nfserr_cb_path_down;
         if (!list_empty(&clp->cl_delegations)
-                        && !atomic_read(&clp->cl_cb_set))
+                        && clp->cl_cb_state != NFSD4_CB_UP)
                 goto out;
         status = nfs_ok;
 out:
@@ -2682,7 +2906,7 @@ out:
         return status;
 }
 
-struct lock_manager nfsd4_manager = {
+static struct lock_manager nfsd4_manager = {
 };
 
 static void
@@ -2750,8 +2974,6 @@ nfs4_laundromat(void)
                                 test_val = u;
                         break;
                 }
-                dprintk("NFSD: purging unused delegation dp %p, fp %p\n",
-                                    dp, dp->dl_flock);
                 list_move(&dp->dl_recall_lru, &reaplist);
         }
         spin_unlock(&recall_lock);
@@ -2861,7 +3083,7 @@ check_special_stateids(svc_fh *current_fh, stateid_t *stateid, int flags)
         if (ONE_STATEID(stateid) && (flags & RD_STATE))
                 return nfs_ok;
         else if (locks_in_grace()) {
-                /* Answer in remaining cases depends on existance of
+                /* Answer in remaining cases depends on existence of
                  * conflicting state; so we must wait out the grace period. */
                 return nfserr_grace;
         } else if (flags & WR_STATE)
@@ -2944,7 +3166,11 @@ nfs4_preprocess_stateid_op(struct nfsd4_compound_state *cstate,
         if (STALE_STATEID(stateid)) 
                 goto out;
 
-        status = nfserr_bad_stateid;
+        /*
+         * We assume that any stateid that has the current boot time,
+         * but that we can't find, is expired:
+         */
+        status = nfserr_expired;
         if (is_delegation_stateid(stateid)) {
                 dp = find_delegation_stateid(ino, stateid);
                 if (!dp)
@@ -2957,13 +3183,15 @@ nfs4_preprocess_stateid_op(struct nfsd4_compound_state *cstate,
                 if (status)
                         goto out;
                 renew_client(dp->dl_client);
-                if (filpp)
-                        *filpp = find_readable_file(dp->dl_file);
-                BUG_ON(!*filpp);
+                if (filpp) {
+                        *filpp = dp->dl_file->fi_deleg_file;
+                        BUG_ON(!*filpp);
+                }
         } else { /* open or lock stateid */
                 stp = find_stateid(stateid, flags);
                 if (!stp)
                         goto out;
+                status = nfserr_bad_stateid;
                 if (nfs4_check_fh(current_fh, stp))
                         goto out;
                 if (!stp->st_stateowner->so_confirmed)
@@ -3038,8 +3266,9 @@ nfs4_preprocess_seqid_op(struct nfsd4_compound_state *cstate, u32 seqid,
                  * a replayed close:
                  */
                 sop = search_close_lru(stateid->si_stateownerid, flags);
+                /* It's not stale; let's assume it's expired: */
                 if (sop == NULL)
-                        return nfserr_bad_stateid;
+                        return nfserr_expired;
                 *sopp = sop;
                 goto check_replay;
         }
@@ -3304,6 +3533,7 @@ nfsd4_delegreturn(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
         status = nfserr_bad_stateid;
         if (!is_delegation_stateid(stateid))
                 goto out;
+        status = nfserr_expired;
         dp = find_delegation_stateid(inode, stateid);
         if (!dp)
                 goto out;
@@ -3473,7 +3703,7 @@ find_lockstateowner_str(struct inode *inode, clientid_t *clid,
 /*
  * Alloc a lock owner structure.
  * Called in nfsd4_lock - therefore, OPEN and OPEN_CONFIRM (if needed) has 
- * occured. 
+ * occurred. 
  *
  * strhashval = lock_ownerstr_hashval 
  */
@@ -3534,6 +3764,7 @@ alloc_init_lock_stateid(struct nfs4_stateowner *sop, struct nfs4_file *fp, struc
         stp->st_stateid.si_stateownerid = sop->so_id;
         stp->st_stateid.si_fileid = fp->fi_id;
         stp->st_stateid.si_generation = 0;
+        stp->st_access_bmap = 0;
         stp->st_deny_bmap = open_stp->st_deny_bmap;
         stp->st_openstp = open_stp;
 
@@ -3548,6 +3779,17 @@ check_lock_length(u64 offset, u64 length)
              LOFF_OVERFLOW(offset, length)));
 }
 
+static void get_lock_access(struct nfs4_stateid *lock_stp, u32 access)
+{
+        struct nfs4_file *fp = lock_stp->st_file;
+        int oflag = nfs4_access_to_omode(access);
+
+        if (test_bit(access, &lock_stp->st_access_bmap))
+                return;
+        nfs4_file_get_access(fp, oflag);
+        __set_bit(access, &lock_stp->st_access_bmap);
+}
+
 /*
  *  LOCK operation 
  */
@@ -3564,7 +3806,6 @@ nfsd4_lock(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
         struct file_lock conflock;
         __be32 status = 0;
         unsigned int strhashval;
-        unsigned int cmd;
         int err;
 
         dprintk("NFSD: nfsd4_lock: start=%Ld length=%Ld\n",
@@ -3646,22 +3887,18 @@ nfsd4_lock(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
         switch (lock->lk_type) {
                 case NFS4_READ_LT:
                 case NFS4_READW_LT:
-                        if (find_readable_file(lock_stp->st_file)) {
-                                nfs4_get_vfs_file(rqstp, fp, &cstate->current_fh, NFS4_SHARE_ACCESS_READ);
-                                filp = find_readable_file(lock_stp->st_file);
-                        }
+                        filp = find_readable_file(lock_stp->st_file);
+                        if (filp)
+                                get_lock_access(lock_stp, NFS4_SHARE_ACCESS_READ);
                         file_lock.fl_type = F_RDLCK;
-                        cmd = F_SETLK;
-                break;
+                        break;
                 case NFS4_WRITE_LT:
                 case NFS4_WRITEW_LT:
-                        if (find_writeable_file(lock_stp->st_file)) {
-                                nfs4_get_vfs_file(rqstp, fp, &cstate->current_fh, NFS4_SHARE_ACCESS_WRITE);
-                                filp = find_writeable_file(lock_stp->st_file);
-                        }
+                        filp = find_writeable_file(lock_stp->st_file);
+                        if (filp)
+                                get_lock_access(lock_stp, NFS4_SHARE_ACCESS_WRITE);
                         file_lock.fl_type = F_WRLCK;
-                        cmd = F_SETLK;
-                break;
+                        break;
                 default:
                         status = nfserr_inval;
                 goto out;
@@ -3685,7 +3922,7 @@ nfsd4_lock(struct svc_rqst *rqstp, struct nfsd4_compound_state *cstate,
         * Note: locks.c uses the BKL to protect the inode's lock list.
         */
 
-        err = vfs_lock_file(filp, cmd, &file_lock, &conflock);
+        err = vfs_lock_file(filp, F_SETLK, &file_lock, &conflock);
         switch (-err) {
         case 0: /* success! */
                 update_stateid(&lock_stp->st_stateid);
@@ -3895,7 +4132,7 @@ check_for_locks(struct nfs4_file *filp, struct nfs4_stateowner *lowner)
         struct inode *inode = filp->fi_inode;
         int status = 0;
 
-        lock_kernel();
+        lock_flocks();
         for (flpp = &inode->i_flock; *flpp != NULL; flpp = &(*flpp)->fl_next) {
                 if ((*flpp)->fl_owner == (fl_owner_t)lowner) {
                         status = 1;
@@ -3903,7 +4140,7 @@ check_for_locks(struct nfs4_file *filp, struct nfs4_stateowner *lowner)
                 }
         }
 out:
-        unlock_kernel();
+        unlock_flocks();
         return status;
 }
 
@@ -3980,7 +4217,7 @@ nfs4_has_reclaimed_state(const char *name, bool use_exchange_id)
         unsigned int strhashval = clientstr_hashval(name);
         struct nfs4_client *clp;
 
-        clp = find_confirmed_client_by_str(name, strhashval, use_exchange_id);
+        clp = find_confirmed_client_by_str(name, strhashval);
         return clp ? 1 : 0;
 }
 
@@ -4209,7 +4446,7 @@ __nfs4_state_shutdown(void)
 void
 nfs4_state_shutdown(void)
 {
-        cancel_rearming_delayed_workqueue(laundry_wq, &laundromat_work);
+        cancel_delayed_work_sync(&laundromat_work);
         destroy_workqueue(laundry_wq);
         locks_end_grace(&nfsd4_manager);
         nfs4_lock_state();