nfs41 client patches for review

index f78720b..133ad76 100644
@@ -130,6 +130,10 @@ extern __be32 * nfs_decode_dirent(__be32 *, struct nfs_entry *, int);
 extern struct rpc_procinfo nfs3_procedures[];
 extern __be32 *nfs3_decode_dirent(__be32 *, struct nfs_entry *, int);
 
+/* nfs4proc.c */
+extern void nfs4_restart_rpc(struct rpc_task *task,
+			     const struct nfs_server *server);
+
 /* nfs4xdr.c */
 #ifdef CONFIG_NFS_V4
 extern __be32 *nfs4_decode_dirent(__be32 *p, struct nfs_entry *entry, int plus);

index f6107b0..917d630 100644
@@ -593,6 +593,18 @@ static void nfs4_sequence_done(const struct nfs_server *server,
 #endif /* CONFIG_NFS_V4_1 */
 }
 
+void nfs4_restart_rpc(struct rpc_task *task, const struct nfs_server *server)
+{
+#ifdef CONFIG_NFS_V4_1
+	if (server->nfs_client->cl_minorversion &&
+	    nfs41_test_session_reset(server->nfs_client->cl_session)) {
+		rpc_restart_call_prepare(task);
+		return;
+	}
+#endif /* CONFIG_NFS_V4_1 */
+	rpc_restart_call(task);
+}
+
 static void update_changeattr(struct inode *dir, struct nfs4_change_info *cinfo)
 {
 	struct nfs_inode *nfsi = NFS_I(dir);

@@ -1638,7 +1650,7 @@ static void nfs4_close_done(struct rpc_task *task, void *data)
 			break;
 		default:
 			if (nfs4_async_handle_error(task, server) == -EAGAIN) {
-				rpc_restart_call(task);
+				nfs4_restart_rpc(task, server);
 				return;
 			}
 	}

@@ -2871,7 +2883,7 @@ static int nfs4_read_done(struct rpc_task *task, struct nfs_read_data *data)
 	/* nfs4_sequence_free_slot called in the read rpc_call_done */
 	nfs4_sequence_done(server, &data->res.seq_res, task->tk_status);
 	if (nfs4_async_handle_error(task, server) == -EAGAIN) {
-		rpc_restart_call(task);
+		nfs4_restart_rpc(task, server);
 		dprintk("<-- %s status= %d\n", __func__, -EAGAIN);
 		return -EAGAIN;
 	}

@@ -2899,7 +2911,7 @@ static int nfs4_write_done(struct rpc_task *task, struct nfs_write_data *data)
 	/* slot is freed in nfs_writeback_done */
 	nfs4_sequence_done(server, &data->res.seq_res, task->tk_status);
 	if (nfs4_async_handle_error(task, server) == -EAGAIN) {
-		rpc_restart_call(task);
+		nfs4_restart_rpc(task, server);
 		return -EAGAIN;
 	}
 	if (task->tk_status >= 0) {

@@ -2927,7 +2939,7 @@ static int nfs4_commit_done(struct rpc_task *task, struct nfs_write_data *data)
 
 	nfs4_sequence_done(server, &data->res.seq_res, task->tk_status);
 	if (nfs4_async_handle_error(task, server) == -EAGAIN) {
-		rpc_restart_call(task);
+		nfs4_restart_rpc(task, server);
 		return -EAGAIN;
 	}
 	nfs4_sequence_free_slot(server, &data->res.seq_res);

@@ -3701,7 +3713,7 @@ static void nfs4_locku_done(struct rpc_task *task, void *data)
 			break;
 		default:
 			if (nfs4_async_handle_error(task, server) == -EAGAIN)
-				rpc_restart_call(task);
+				nfs4_restart_rpc(task, server);
 	}
 	nfs4_sequence_free_slot(server, &calldata->res.seq_res);
 }

index 3c77b26..5b8db19 100644
@@ -367,7 +367,7 @@ static void nfs_readpage_retry(struct rpc_task *task, struct nfs_read_data *data
 	argp->offset += resp->count;
 	argp->pgbase += resp->count;
 	argp->count -= resp->count;
-	rpc_restart_call(task);
+	nfs4_restart_rpc(task, NFS_SERVER(data->inode));
 	return;
 out:
 	nfs4_sequence_free_slot(NFS_SERVER(data->inode), &data->res.seq_res);

index 3df3794..2ac5dd5 100644
@@ -83,7 +83,7 @@ static void nfs_async_unlink_done(struct rpc_task *task, void *calldata)
 	struct inode *dir = data->dir;
 
 	if (!NFS_PROTO(dir)->unlink_done(task, dir))
-		rpc_restart_call(task);
+		nfs4_restart_rpc(task, NFS_SERVER(dir));
 }
 
 /**

index 7d3fdc8..86573f7 100644
@@ -1179,7 +1179,7 @@ int nfs_writeback_done(struct rpc_task *task, struct nfs_write_data *data)
 				 */
 				argp->stable = NFS_FILE_SYNC;
 			}
-			rpc_restart_call(task);
+			nfs4_restart_rpc(task, NFS_SERVER(data->inode));
 			return -EAGAIN;
 		}
 		if (time_before(complain, jiffies)) {

index f78720b..133ad76 100644
@@ -130,6 +130,10 @@ extern __be32 * nfs_decode_dirent(__be32 *, struct nfs_entry *, int);
 extern struct rpc_procinfo nfs3_procedures[];
 extern __be32 *nfs3_decode_dirent(__be32 *, struct nfs_entry *, int);
 
+/* nfs4proc.c */
+extern void nfs4_restart_rpc(struct rpc_task *task,
+                            const struct nfs_server *server);
+
 /* nfs4xdr.c */
 #ifdef CONFIG_NFS_V4
 extern __be32 *nfs4_decode_dirent(__be32 *p, struct nfs_entry *entry, int plus);

index f6107b0..917d630 100644
@@ -593,6 +593,18 @@ static void nfs4_sequence_done(const struct nfs_server *server,
 #endif /* CONFIG_NFS_V4_1 */
 }
 
+void nfs4_restart_rpc(struct rpc_task *task, const struct nfs_server *server)
+{
+#ifdef CONFIG_NFS_V4_1
+       if (server->nfs_client->cl_minorversion &&
+           nfs41_test_session_reset(server->nfs_client->cl_session)) {
+               rpc_restart_call_prepare(task);
+               return;
+       }
+#endif /* CONFIG_NFS_V4_1 */
+       rpc_restart_call(task);
+}
+
 static void update_changeattr(struct inode *dir, struct nfs4_change_info *cinfo)
 {
        struct nfs_inode *nfsi = NFS_I(dir);

@@ -1638,7 +1650,7 @@ static void nfs4_close_done(struct rpc_task *task, void *data)
                        break;
                default:
                        if (nfs4_async_handle_error(task, server) == -EAGAIN) {
-                               rpc_restart_call(task);
+                               nfs4_restart_rpc(task, server);
                                return;
                        }
        }

@@ -2871,7 +2883,7 @@ static int nfs4_read_done(struct rpc_task *task, struct nfs_read_data *data)
        /* nfs4_sequence_free_slot called in the read rpc_call_done */
        nfs4_sequence_done(server, &data->res.seq_res, task->tk_status);
        if (nfs4_async_handle_error(task, server) == -EAGAIN) {
-               rpc_restart_call(task);
+               nfs4_restart_rpc(task, server);
                dprintk("<-- %s status= %d\n", __func__, -EAGAIN);
                return -EAGAIN;
        }

@@ -2899,7 +2911,7 @@ static int nfs4_write_done(struct rpc_task *task, struct nfs_write_data *data)
        /* slot is freed in nfs_writeback_done */
        nfs4_sequence_done(server, &data->res.seq_res, task->tk_status);
        if (nfs4_async_handle_error(task, server) == -EAGAIN) {
-               rpc_restart_call(task);
+               nfs4_restart_rpc(task, server);
                return -EAGAIN;
        }
        if (task->tk_status >= 0) {

@@ -2927,7 +2939,7 @@ static int nfs4_commit_done(struct rpc_task *task, struct nfs_write_data *data)
 
        nfs4_sequence_done(server, &data->res.seq_res, task->tk_status);
        if (nfs4_async_handle_error(task, server) == -EAGAIN) {
-               rpc_restart_call(task);
+               nfs4_restart_rpc(task, server);
                return -EAGAIN;
        }
        nfs4_sequence_free_slot(server, &data->res.seq_res);

@@ -3701,7 +3713,7 @@ static void nfs4_locku_done(struct rpc_task *task, void *data)
                        break;
                default:
                        if (nfs4_async_handle_error(task, server) == -EAGAIN)
-                               rpc_restart_call(task);
+                               nfs4_restart_rpc(task, server);
        }
        nfs4_sequence_free_slot(server, &calldata->res.seq_res);
 }

index 3c77b26..5b8db19 100644
@@ -367,7 +367,7 @@ static void nfs_readpage_retry(struct rpc_task *task, struct nfs_read_data *data
        argp->offset += resp->count;
        argp->pgbase += resp->count;
        argp->count -= resp->count;
-       rpc_restart_call(task);
+       nfs4_restart_rpc(task, NFS_SERVER(data->inode));
        return;
 out:
        nfs4_sequence_free_slot(NFS_SERVER(data->inode), &data->res.seq_res);

index 3df3794..2ac5dd5 100644
@@ -83,7 +83,7 @@ static void nfs_async_unlink_done(struct rpc_task *task, void *calldata)
        struct inode *dir = data->dir;
 
        if (!NFS_PROTO(dir)->unlink_done(task, dir))
-               rpc_restart_call(task);
+               nfs4_restart_rpc(task, NFS_SERVER(dir));
 }
 
 /**

index 7d3fdc8..86573f7 100644
@@ -1179,7 +1179,7 @@ int nfs_writeback_done(struct rpc_task *task, struct nfs_write_data *data)
                                 */
                                argp->stable = NFS_FILE_SYNC;
                        }
-                       rpc_restart_call(task);
+                       nfs4_restart_rpc(task, NFS_SERVER(data->inode));
                        return -EAGAIN;
                }
                if (time_before(complain, jiffies)) {

index 917d630..a4f5b15 100644
@@ -3256,6 +3256,25 @@ nfs4_async_handle_error(struct rpc_task *task, const struct nfs_server *server)
 				rpc_wake_up_queued_task(&clp->cl_rpcwaitq, task);
 			task->tk_status = 0;
 			return -EAGAIN;
+#if defined(CONFIG_NFS_V4_1)
+		case -NFS4ERR_BADSESSION:
+		case -NFS4ERR_BADSLOT:
+		case -NFS4ERR_CONN_NOT_BOUND_TO_SESSION:
+		case -NFS4ERR_BACK_CHAN_BUSY:
+		case -NFS4ERR_SEQ_MISORDERED:
+		case -NFS4ERR_SEQUENCE_POS:
+		case -NFS4ERR_REQ_TOO_BIG:
+		case -NFS4ERR_REP_TOO_BIG:
+		case -NFS4ERR_REP_TOO_BIG_TO_CACHE:
+		case -NFS4ERR_RETRY_UNCACHED_REP:
+		case -NFS4ERR_TOO_MANY_OPS:
+		case -NFS4ERR_OP_NOT_IN_SESSION:
+			dprintk("%s ERROR %d, Reset session\n", __func__,
+				task->tk_status);
+			nfs41_set_session_reset(clp->cl_session);
+			task->tk_status = 0;
+			return -EAGAIN;
+#endif /* CONFIG_NFS_V4_1 */
 		case -NFS4ERR_DELAY:
 			nfs_inc_server_stats(server, NFSIOS_DELAY);
 		case -NFS4ERR_GRACE:

@@ -3331,7 +3350,30 @@ static int nfs4_handle_exception(struct nfs_server *server, int errorcode, struc
 			ret = nfs4_wait_clnt_recover(server->client, clp);
 			if (ret == 0)
 				exception->retry = 1;
+#if !defined(CONFIG_NFS_V4_1)
 			break;
+#else /* !defined(CONFIG_NFS_V4_1) */
+			if (!server->nfs_client->cl_minorversion)
+				break;
+			/* FALLTHROUGH */
+		case -NFS4ERR_BADSESSION:
+		case -NFS4ERR_BADSLOT:
+		case -NFS4ERR_CONN_NOT_BOUND_TO_SESSION:
+		case -NFS4ERR_BACK_CHAN_BUSY:
+		case -NFS4ERR_SEQ_MISORDERED:
+		case -NFS4ERR_SEQUENCE_POS:
+		case -NFS4ERR_REQ_TOO_BIG:
+		case -NFS4ERR_REP_TOO_BIG:
+		case -NFS4ERR_REP_TOO_BIG_TO_CACHE:
+		case -NFS4ERR_RETRY_UNCACHED_REP:
+		case -NFS4ERR_TOO_MANY_OPS:
+		case -NFS4ERR_OP_NOT_IN_SESSION:
+			dprintk("%s ERROR: %d Reset session\n", __func__,
+				errorcode);
+			nfs41_set_session_reset(clp->cl_session);
+			exception->retry = 1;
+			/* FALLTHROUGH */
+#endif /* !defined(CONFIG_NFS_V4_1) */
 		case -NFS4ERR_FILE_OPEN:
 		case -NFS4ERR_GRACE:
 		case -NFS4ERR_DELAY:

@@ -4735,8 +4777,37 @@ static int nfs4_proc_sequence(struct nfs_client *clp, struct rpc_cred *cred)
 void nfs41_sequence_call_done(struct rpc_task *task, void *data)
 {
 	struct nfs_server *server = (struct nfs_server *)data;
+	struct nfs_client *clp = server->nfs_client;
 
 	nfs4_sequence_done(server, task->tk_msg.rpc_resp, task->tk_status);
+
+	if (task->tk_status < 0) {
+		switch (task->tk_status) {
+		case -NFS4ERR_STALE_CLIENTID:
+		case -NFS4ERR_EXPIRED:
+		case -NFS4ERR_CB_PATH_DOWN:
+			nfs4_schedule_state_recovery(clp);
+		case -NFS4ERR_BADSLOT:
+		case -NFS4ERR_BADSESSION:
+		case -NFS4ERR_CONN_NOT_BOUND_TO_SESSION:
+		case -NFS4ERR_BACK_CHAN_BUSY:
+		case -NFS4ERR_SEQ_MISORDERED:
+		case -NFS4ERR_SEQUENCE_POS:
+		case -NFS4ERR_REQ_TOO_BIG:
+		case -NFS4ERR_REP_TOO_BIG:
+		case -NFS4ERR_REP_TOO_BIG_TO_CACHE:
+		case -NFS4ERR_RETRY_UNCACHED_REP:
+		case -NFS4ERR_TOO_MANY_OPS:
+		case -NFS4ERR_OP_NOT_IN_SESSION:
+			dprintk("%s ERROR %d Reset session\n", __func__,
+				task->tk_status);
+
+			if (nfs4_async_handle_error(task, server) == -EAGAIN) {
+				nfs4_restart_rpc(task, server);
+				return;
+			}
+		}
+	}
 	nfs4_sequence_free_slot(server, task->tk_msg.rpc_resp);
 	dprintk("%s rpc_cred %p\n", __func__, task->tk_msg.rpc_cred);
 

index a4f5b15..12f3b9f 100644
@@ -491,8 +491,11 @@ int nfs4_setup_sequence(struct nfs_client *clp,
 	if (clp->cl_minorversion == 0 || res->sr_slot)
 		goto out;
 	BUG_ON(clp->cl_minorversion != 1);
-	ret = nfs41_setup_sequence(clp->cl_session, args, res,
-				   cache_reply, task);
+	if (nfs41_test_session_alloc(clp->cl_session))
+		ret = nfs41_recover_session_sync(clp->cl_session);
+	if (!ret)
+		ret = nfs41_setup_sequence(clp->cl_session, args, res,
+					   cache_reply, task);
 	if (ret) {
 		memset(res, 0, sizeof(*res));
 		if (ret != -EAGAIN) {

index 560b10b..c69424d 100644
@@ -207,6 +207,10 @@ static void nfs_free_client(struct nfs_client *clp)
 
 	if (clp->cl_machine_cred != NULL)
 		put_rpccred(clp->cl_machine_cred);
+#ifdef CONFIG_NFS_V4_1
+	if (clp->cl_ex_cred != NULL)
+		put_rpccred(clp->cl_ex_cred);
+#endif /* CONFIG_NFS_V4_1 */
 
 	kfree(clp->cl_hostname);
 	kfree(clp);

index ff367df..904b19e 100644
@@ -167,6 +167,7 @@ struct nfs4_state_recovery_ops {
 
 struct nfs4_state_maintenance_ops {
 	int (*sched_state_renewal)(struct nfs_client *, struct rpc_cred *);
+	struct rpc_cred * (*get_state_renewal_cred)(struct nfs_client *);
 };
 
 extern struct dentry_operations nfs4_dentry_operations;

@@ -228,6 +229,9 @@ extern void nfs4_renew_state(struct work_struct *);
 
 /* nfs4state.c */
 struct rpc_cred *nfs4_get_renew_cred(struct nfs_client *clp);
+#ifdef CONFIG_NFS_V4_1
+struct rpc_cred *nfs41_get_state_renewal_cred(struct nfs_client *clp);
+#endif /* CONFIG_NFS_V4_1 */
 
 extern struct nfs4_state_owner * nfs4_get_state_owner(struct nfs_server *, struct rpc_cred *);
 extern void nfs4_put_state_owner(struct nfs4_state_owner *);

index 12f3b9f..5b1f6d0 100644
@@ -4317,6 +4317,17 @@ static int nfs4_proc_exchange_id(struct nfs_client *clp, struct rpc_cred *cred)
 		else if (++clp->cl_id_uniquifier == 0)
 			break;
 	}
+	if (status == 0) {
+		if (clp->cl_ex_cred) {
+			dprintk("%s put cl_ex_cred %p\n", __func__,
+				clp->cl_ex_cred);
+			put_rpccred(clp->cl_ex_cred);
+		}
+
+		BUG_ON(cred == NULL);
+		clp->cl_ex_cred = get_rpccred(cred);
+		dprintk("%s set cl_ex_cred %p\n", __func__, clp->cl_ex_cred);
+	}
 
 	dprintk("<-- %s status= %d\n", __func__, status);
 	return status;

@@ -4907,11 +4918,13 @@ struct nfs4_state_recovery_ops nfs41_network_partition_recovery_ops = {
 
 struct nfs4_state_maintenance_ops nfs40_state_renewal_ops = {
 	.sched_state_renewal = nfs4_proc_async_renew,
+	.get_state_renewal_cred = nfs4_get_renew_cred,
 };
 
 #if defined(CONFIG_NFS_V4_1)
 struct nfs4_state_maintenance_ops nfs41_state_renewal_ops = {
 	.sched_state_renewal = nfs41_proc_async_sequence,
+	.get_state_renewal_cred = nfs41_get_state_renewal_cred,
 };
 #endif
 

index a1bab0c..1d609ed 100644
@@ -79,7 +79,7 @@ nfs4_renew_state(struct work_struct *work)
 	timeout = (2 * lease) / 3 + (long)last - (long)now;
 	/* Are we close to a lease timeout? */
 	if (time_after(now, last + lease/3)) {
-		cred = nfs4_get_renew_cred(clp);
+		cred = ops->get_state_renewal_cred(clp);
 		if (cred == NULL) {
 			set_bit(NFS4CLNT_LEASE_EXPIRED, &clp->cl_state);
 			spin_unlock(&clp->cl_lock);

index cf6666d..5a91072 100644
@@ -73,6 +73,16 @@ static int nfs4_init_client(struct nfs_client *clp, struct rpc_cred *cred)
 	return status;
 }
 
+#if defined(CONFIG_NFS_V4_1)
+struct rpc_cred *
+nfs41_get_state_renewal_cred(struct nfs_client *clp)
+{
+	if (clp->cl_ex_cred)
+		get_rpccred(clp->cl_ex_cred);
+	return (clp->cl_ex_cred);
+}
+#endif /* CONFIG_NFS_V4_1 */
+
 static struct rpc_cred *nfs4_get_machine_cred(struct nfs_client *clp)
 {
 	struct rpc_cred *cred = NULL;

@@ -943,6 +953,7 @@ static int reclaimer(void *ptr)
 	struct nfs4_state_owner *sp;
 	struct rb_node *pos;
 	struct nfs4_state_recovery_ops *ops;
+	struct nfs4_state_maintenance_ops *m_ops;
 	struct rpc_cred *cred;
 	int status = 0;
 

@@ -957,8 +968,9 @@ static int reclaimer(void *ptr)
 restart_loop:
 	dprintk("%s: starting loop\n", __func__);
 	ops = nfs4_network_partition_recovery_ops[clp->cl_minorversion];
+	m_ops = nfs4_state_renewal_ops[clp->cl_minorversion];
 	/* Are there any open files on this volume? */
-	cred = nfs4_get_renew_cred(clp);
+	cred = m_ops->get_state_renewal_cred(clp);
 	if (cred != NULL) {
 		/* Yes there are: try to renew the old lease */
 		status = ops->renew_lease(clp, cred);

index 9917c09..17720b9 100644
@@ -81,6 +81,7 @@ struct nfs_client {
 	u32			cl_seqid;
 	/* The flags used for obtaining the clientid during EXCHANGE_ID */
 	u32			cl_exchange_flags;
+	struct rpc_cred		*cl_ex_cred;	/* exchange_id credential */
 	struct nfs4_session	*cl_session; 	/* sharred session */
 #endif /* CONFIG_NFS_V4_1 */
 };

index 904b19e..b9771f2 100644
@@ -229,6 +229,7 @@ extern void nfs4_renew_state(struct work_struct *);
 
 /* nfs4state.c */
 struct rpc_cred *nfs4_get_renew_cred(struct nfs_client *clp);
+struct rpc_cred *nfs4_get_setclientid_cred(struct nfs_client *clp);
 #ifdef CONFIG_NFS_V4_1
 struct rpc_cred *nfs41_get_state_renewal_cred(struct nfs_client *clp);
 #endif /* CONFIG_NFS_V4_1 */

index 5b1f6d0..8c28d0c 100644
@@ -4276,6 +4276,7 @@ static int nfs4_proc_exchange_id(struct nfs_client *clp, struct rpc_cred *cred)
 	};
 	int status;
 	int loop = 0;
+	int got_cred = 0;
 	struct rpc_message msg = {
 		.rpc_proc = nfs4_proc(clp, NFSPROC4_CLNT_EXCHANGE_ID),
 		.rpc_argp = &args,

@@ -4286,6 +4287,16 @@ static int nfs4_proc_exchange_id(struct nfs_client *clp, struct rpc_cred *cred)
 
 	dprintk("--> %s\n", __func__);
 	BUG_ON(clp == NULL);
+
+	if (!cred) {
+		/* FIXME: do we need to lock clp and take a ref count on cl_ex_cred? */
+		cred = clp->cl_ex_cred;
+		if (!cred) {
+			cred = nfs4_get_setclientid_cred(clp);
+			got_cred = (cred != NULL);
+		}
+	}
+
 	p = (u32 *)verifier.data;
 	*p++ = htonl((u32)clp->cl_boot_time.tv_sec);
 	*p = htonl((u32)clp->cl_boot_time.tv_nsec);

@@ -4317,17 +4328,18 @@ static int nfs4_proc_exchange_id(struct nfs_client *clp, struct rpc_cred *cred)
 		else if (++clp->cl_id_uniquifier == 0)
 			break;
 	}
-	if (status == 0) {
+	if ((status == 0) && (cred != clp->cl_ex_cred) && (cred != NULL)) {
+		/* FIXME: do we need to lock clp here? */
 		if (clp->cl_ex_cred) {
 			dprintk("%s put cl_ex_cred %p\n", __func__,
 				clp->cl_ex_cred);
 			put_rpccred(clp->cl_ex_cred);
 		}
-
-		BUG_ON(cred == NULL);
 		clp->cl_ex_cred = get_rpccred(cred);
 		dprintk("%s set cl_ex_cred %p\n", __func__, clp->cl_ex_cred);
 	}
+	if (got_cred)
+		put_rpccred(cred);
 
 	dprintk("<-- %s status= %d\n", __func__, status);
 	return status;

index 5a91072..693e5a6 100644
@@ -122,7 +122,7 @@ struct rpc_cred *nfs4_get_renew_cred(struct nfs_client *clp)
 	return cred;
 }
 
-static struct rpc_cred *nfs4_get_setclientid_cred(struct nfs_client *clp)
+struct rpc_cred *nfs4_get_setclientid_cred(struct nfs_client *clp)
 {
 	struct nfs4_state_owner *sp;
 	struct rb_node *pos;

index b9771f2..44282b0 100644
@@ -163,6 +163,7 @@ struct nfs4_state_recovery_ops {
 	int (*recover_open)(struct nfs4_state_owner *, struct nfs4_state *);
 	int (*recover_lock)(struct nfs4_state *, struct file_lock *);
 	int (*renew_lease)(struct nfs_client *, struct rpc_cred *);
+	int (*establish_clid)(struct nfs_client *, struct rpc_cred *);
 };
 
 struct nfs4_state_maintenance_ops {

@@ -186,6 +187,7 @@ extern int nfs4_proc_setclientid_confirm(struct nfs_client *, struct rpc_cred *)
 extern int nfs4_proc_async_renew(struct nfs_client *, struct rpc_cred *);
 extern int nfs4_proc_renew(struct nfs_client *, struct rpc_cred *);
 extern int nfs4_wait_bit_killable(void *);
+extern int nfs4_init_clientid(struct nfs_client *, struct rpc_cred *);
 extern int nfs4_do_close(struct path *path, struct nfs4_state *state, int wait);
 extern struct dentry *nfs4_atomic_open(struct inode *, struct dentry *, struct nameidata *);
 extern int nfs4_open_revalidate(struct inode *, struct dentry *, int, struct nameidata *);

index 8c28d0c..749027d 100644
@@ -4904,6 +4904,7 @@ struct nfs4_state_recovery_ops nfs40_reboot_recovery_ops = {
 	.recover_open	= nfs4_open_reclaim,
 	.recover_lock	= nfs4_lock_reclaim,
 	.renew_lease	= nfs4_proc_renew,
+	.establish_clid = nfs4_init_clientid,
 };
 
 #if defined(CONFIG_NFS_V4_1)

@@ -4911,6 +4912,7 @@ struct nfs4_state_recovery_ops nfs41_reboot_recovery_ops = {
 	.recover_open	= nfs4_open_reclaim,
 	.recover_lock	= nfs4_lock_reclaim,
 	.renew_lease	= nfs4_proc_sequence,
+	.establish_clid = nfs4_proc_exchange_id,
 };
 #endif /* CONFIG_NFS_V4_1 */
 

@@ -4918,6 +4920,7 @@ struct nfs4_state_recovery_ops nfs40_network_partition_recovery_ops = {
 	.recover_open	= nfs4_open_expired,
 	.recover_lock	= nfs4_lock_expired,
 	.renew_lease	= nfs4_proc_renew,
+	.establish_clid = nfs4_init_clientid,
 };
 
 #if defined(CONFIG_NFS_V4_1)

@@ -4925,6 +4928,7 @@ struct nfs4_state_recovery_ops nfs41_network_partition_recovery_ops = {
 	.recover_open	= nfs4_open_expired,
 	.recover_lock	= nfs4_lock_expired,
 	.renew_lease	= nfs4_proc_sequence,
+	.establish_clid = nfs4_proc_exchange_id,
 };
 #endif /* CONFIG_NFS_V4_1 */
 

index 693e5a6..d3c5e92 100644
@@ -62,7 +62,7 @@ const nfs4_stateid zero_stateid;
 
 static LIST_HEAD(nfs4_clientid_list);
 
-static int nfs4_init_client(struct nfs_client *clp, struct rpc_cred *cred)
+int nfs4_init_clientid(struct nfs_client *clp, struct rpc_cred *cred)
 {
 	int status = nfs4_proc_setclientid(clp, NFS4_CALLBACK,
 			nfs_callback_tcpport, cred);

@@ -991,8 +991,9 @@ restart_loop:
 	nfs4_state_mark_reclaim(clp);
 	status = -ENOENT;
 	cred = nfs4_get_setclientid_cred(clp);
+	dprintk("%s: setclientid_cred %p\n", __func__, cred);
 	if (cred != NULL) {
-		status = nfs4_init_client(clp, cred);
+		status = ops->establish_clid(clp, cred);
 		put_rpccred(cred);
 		/* Handle case where the user hasn't set up machine creds */
 		if (status == -EACCES && cred == clp->cl_machine_cred) {

index 749027d..179ea6c 100644
@@ -1956,7 +1956,11 @@ static int _nfs4_lookup_root(struct nfs_server *server, struct nfs_fh *fhandle,
 
 	dprintk("--> %s\n", __func__);
 	nfs_fattr_init(info->fattr);
+	status = nfs4_recover_expired_lease(server);
+	if (status != 0)
+		goto out;
 	status = nfs4_call_sync(server, &msg, &args, &res, 0);
+out:
 	dprintk("<-- %s status= %d\n", __func__, status);
 	return status;
 }

index 179ea6c..699bbe7 100644
@@ -403,6 +403,13 @@ nfs4_find_slot(struct nfs4_slot_table *tbl, struct rpc_task *task,
 			dprintk("%s Slot Table Empty\n", __func__);
 			spin_unlock(&tbl->slot_tbl_lock);
 			ret = nfs41_recover_session_sync(session);
+			if (test_and_clear_bit(NFS4CLNT_LEASE_EXPIRED,
+						&session->clp->cl_state)) {
+
+				printk(KERN_WARNING "%s Clientid Reset\n",
+					__func__);
+				nfs4_schedule_state_recovery(session->clp);
+			}
 			if (ret)
 				return slot;
 			nfs41_wait_session_reset(session);

index d3c5e92..0d6b902 100644
@@ -975,18 +975,24 @@ restart_loop:
 		/* Yes there are: try to renew the old lease */
 		status = ops->renew_lease(clp, cred);
 		put_rpccred(cred);
-		switch (status) {
-		case 0:
-		case -NFS4ERR_CB_PATH_DOWN:
+		if (!status)
 			goto out;
-		case -NFS4ERR_STALE_CLIENTID:
-		case -NFS4ERR_LEASE_MOVED:
-			ops = nfs4_reboot_recovery_ops[clp->cl_minorversion];
-		}
+		if (clp->cl_minorversion == 0)
+			switch (status) {
+			case -NFS4ERR_CB_PATH_DOWN:
+				goto out;
+			case -NFS4ERR_STALE_CLIENTID:
+			case -NFS4ERR_LEASE_MOVED:
+				break;
+			default:
+				goto establish_clid;
+			}
+		ops = nfs4_reboot_recovery_ops[clp->cl_minorversion];
 	} else {
 		/* "reboot" to ensure we clear all state on the server */
 		clp->cl_boot_time = CURRENT_TIME;
 	}
+establish_clid:
 	/* We're going to have to re-establish a clientid */
 	nfs4_state_mark_reclaim(clp);
 	status = -ENOENT;

index 699bbe7..28c3cf8 100644
@@ -1757,6 +1757,10 @@ int nfs4_do_close(struct path *path, struct nfs4_state *state, int wait)
 	calldata->state = state;
 	calldata->arg.fh = NFS_FH(state->inode);
 	calldata->arg.stateid = &state->open_stateid;
+	if (server->minorversion) {
+		BUG_ON(server->minorversion != 1);
+		memset(calldata->arg.stateid->data, 0, 4);    /* clear seqid */
+	}
 	/* Serialization for the sequence id */
 	calldata->arg.seqid = nfs_alloc_seqid(&state->owner->so_seqid);
 	if (calldata->arg.seqid == NULL)

index 0d6b902..6035634 100644
@@ -747,12 +747,16 @@ static void nfs_increment_seqid(int status, struct nfs_seqid *seqid)
 
 void nfs_increment_open_seqid(int status, struct nfs_seqid *seqid)
 {
-	if (status == -NFS4ERR_BAD_SEQID) {
-		struct nfs4_state_owner *sp = container_of(seqid->sequence,
-				struct nfs4_state_owner, so_seqid);
+	struct nfs4_state_owner *sp = container_of(seqid->sequence,
+					struct nfs4_state_owner, so_seqid);
+	struct nfs_server *server = sp->so_server;
+
+	if (status == -NFS4ERR_BAD_SEQID)
 		nfs4_drop_state_owner(sp);
-	}
-	nfs_increment_seqid(status, seqid);
+	if (server->minorversion == 0)
+		nfs_increment_seqid(status, seqid);
+	else
+		BUG_ON(server->minorversion != 1);
 }
 
 /*

index 0a50361..1ce05d7 100644
@@ -222,7 +222,8 @@ struct sock_xprt {
 	 * State of TCP reply receive
 	 */
 	__be32			tcp_fraghdr,
-				tcp_xid;
+				tcp_xid,
+				tcp_calldir;	/* Is this a call or a reply? */
 
 	u32			tcp_offset,
 				tcp_reclen;

@@ -259,6 +260,7 @@ struct sock_xprt {
 #define TCP_RCV_COPY_FRAGHDR	(1UL << 1)
 #define TCP_RCV_COPY_XID	(1UL << 2)
 #define TCP_RCV_COPY_DATA	(1UL << 3)
+#define TCP_RCV_COPY_CALLDIR	(1UL << 4)
 
 static inline struct sockaddr *xs_addr(struct rpc_xprt *xprt)
 {

@@ -919,7 +921,7 @@ static inline void xs_tcp_read_fraghdr(struct rpc_xprt *xprt, struct xdr_skb_rea
 	transport->tcp_offset = 0;
 
 	/* Sanity check of the record length */
-	if (unlikely(transport->tcp_reclen < 4)) {
+	if (unlikely(transport->tcp_reclen < 8)) {
 		dprintk("RPC:       invalid TCP record fragment length\n");
 		xprt_force_disconnect(xprt);
 		return;

@@ -954,13 +956,47 @@ static inline void xs_tcp_read_xid(struct sock_xprt *transport, struct xdr_skb_r
 	if (used != len)
 		return;
 	transport->tcp_flags &= ~TCP_RCV_COPY_XID;
-	transport->tcp_flags |= TCP_RCV_COPY_DATA;
+	transport->tcp_flags |= TCP_RCV_COPY_CALLDIR;
 	transport->tcp_copied = 4;
-	dprintk("RPC:       reading reply for XID %08x\n",
+	dprintk("RPC:       reading %s XID %08x\n",
+			(transport->tcp_calldir == RPC_REPLY) ? "reply for"
+							      : "request with",
 			ntohl(transport->tcp_xid));
 	xs_tcp_check_fraghdr(transport);
 }
 
+static inline void xs_tcp_read_calldir(struct sock_xprt *transport,
+				       struct xdr_skb_reader *desc)
+{
+	size_t len, used;
+	u32 offset;
+	char *p;
+
+	/*
+	 * We want transport->tcp_offset to be 8 at the end of this routine
+	 * (4 bytes for the xid and 4 bytes for the call/reply flag).
+	 * When this function is called for the first time,
+	 * transport->tcp_offset is 4 (after having already read the xid).
+	 */
+	offset = transport->tcp_offset - sizeof(transport->tcp_xid);
+	len = sizeof(transport->tcp_calldir) - offset;
+	dprintk("RPC:       reading CALL/REPLY flag (%Zu bytes)\n", len);
+	p = ((char *) &transport->tcp_calldir) + offset;
+	used = xdr_skb_read_bits(desc, p, len);
+	transport->tcp_offset += used;
+	if (used != len)
+		return;
+	transport->tcp_flags &= ~TCP_RCV_COPY_CALLDIR;
+	transport->tcp_flags |= TCP_RCV_COPY_DATA;
+	transport->tcp_copied += 4;
+	transport->tcp_calldir = ntohl(transport->tcp_calldir);
+	dprintk("RPC:       reading %s CALL/REPLY flag %08x\n",
+			(transport->tcp_calldir == RPC_REPLY) ? "reply for"
+							      : "request with",
+							transport->tcp_calldir);
+	xs_tcp_check_fraghdr(transport);
+}
+
 static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
 {
 	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);

@@ -1077,6 +1113,11 @@ static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, uns
 			xs_tcp_read_xid(transport, &desc);
 			continue;
 		}
+		/* Read in the call/reply flag */
+		if (transport->tcp_flags & TCP_RCV_COPY_CALLDIR) {
+			xs_tcp_read_calldir(transport, &desc);
+			continue;
+		}
 		/* Read in the request data */
 		if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
 			xs_tcp_read_request(xprt, &desc);

index c31ded3..dd0ff2d 100644
@@ -1371,13 +1371,14 @@ rpc_verify_header(struct rpc_task *task)
 	}
 	if ((len -= 3) < 0)
 		goto out_overflow;
-	p += 1;	/* skip XID */
 
-	if ((n = ntohl(*p++)) != RPC_REPLY) {
-		dprintk("RPC: %5u %s: not an RPC reply: %x\n",
-				task->tk_pid, __func__, n);
-		goto out_garbage;
-	}
+	/*
+	 * Skip the XID and call direction.
+	 * The underlying transport has read the XID and RPC call direction
+	 * to determine this is an RPC reply.
+	 */
+	p += 2;
+
 	if ((n = ntohl(*p++)) != RPC_MSG_ACCEPTED) {
 		if (--len < 0)
 			goto out_overflow;

index c2e9cfd..e792706 100644
@@ -67,7 +67,7 @@ module_param_call(callback_tcpport, param_set_port, param_get_int,
  * This is the callback kernel thread.
  */
 static int
-nfs_callback_svc(void *vrqstp)
+nfs4_callback_svc(void *vrqstp)
 {
 	int err, preverr = 0;
 	struct svc_rqst *rqstp = vrqstp;

@@ -104,12 +104,37 @@ nfs_callback_svc(void *vrqstp)
 	return 0;
 }
 
+
+/*
+ * Prepare to bring up the NFSv4 callback service
+ */
+struct svc_rqst *
+nfs4_callback_up(struct svc_serv *serv)
+{
+	int ret;
+
+	ret = svc_create_xprt(serv, "tcp", nfs_callback_set_tcpport,
+			      SVC_SOCK_ANONYMOUS);
+	if (unlikely(ret <= 0)) {
+		if (ret == 0)
+			ret = -EIO;
+		return ERR_PTR(ret);
+	}
+	nfs_callback_tcpport = ret;
+	dprintk("NFS: Callback listener port = %u (af %u)\n",
+			nfs_callback_tcpport, nfs_callback_family);
+	return svc_prepare_thread(serv, &serv->sv_pools[0]);
+}
+
 /*
  * Bring up the callback thread if it is not already up.
  */
-int nfs_callback_up(void)
+int nfs_callback_up(u32 minorversion, void *args)
 {
 	struct svc_serv *serv = NULL;
+	struct svc_rqst *rqstp;
+	int (* callback_svc)(void *vrqstp);
+	char svc_name[12];
 	int ret = 0;
 
 	mutex_lock(&nfs_callback_mutex);

@@ -117,30 +142,31 @@ int nfs_callback_up(void)
 		goto out;
 	serv = svc_create(&nfs4_callback_program, NFS4_CALLBACK_BUFSIZE,
 				nfs_callback_family, NULL);
-	ret = -ENOMEM;
-	if (!serv)
-		goto out_err;
-
-	ret = svc_create_xprt(serv, "tcp", nfs_callback_set_tcpport,
-			      SVC_SOCK_ANONYMOUS);
-	if (ret <= 0)
+	if (!serv) {
+		ret = -ENOMEM;
 		goto out_err;
-	nfs_callback_tcpport = ret;
-	dprintk("NFS: Callback listener port = %u (af %u)\n",
-			nfs_callback_tcpport, nfs_callback_family);
+	}
 
-	nfs_callback_info.rqst = svc_prepare_thread(serv, &serv->sv_pools[0]);
-	if (IS_ERR(nfs_callback_info.rqst)) {
-		ret = PTR_ERR(nfs_callback_info.rqst);
-		nfs_callback_info.rqst = NULL;
+	/* FIXME: either 4.0 or 4.1 callback service can be up at a time
+	 * need to monitor and control them both */
+	if (!minorversion) {
+		rqstp = nfs4_callback_up(serv);
+		callback_svc = nfs4_callback_svc;
+	} else {
+		BUG();	/* for now */
+	}
+	if (IS_ERR(rqstp)) {
+		ret = PTR_ERR(rqstp);
 		goto out_err;
 	}
 
 	svc_sock_update_bufs(serv);
 
-	nfs_callback_info.task = kthread_run(nfs_callback_svc,
+	sprintf(svc_name, "nfsv4.%u-svc", minorversion);
+	nfs_callback_info.rqst = rqstp;
+	nfs_callback_info.task = kthread_run(callback_svc,
 					     nfs_callback_info.rqst,
-					     "nfsv4-svc");
+					     svc_name);
 	if (IS_ERR(nfs_callback_info.task)) {
 		ret = PTR_ERR(nfs_callback_info.task);
 		svc_exit_thread(nfs_callback_info.rqst);

index bb25d21..a2b3ce4 100644
@@ -63,7 +63,7 @@ extern __be32 nfs4_callback_getattr(struct cb_getattrargs *args, struct cb_getat
 extern __be32 nfs4_callback_recall(struct cb_recallargs *args, void *dummy);
 
 #ifdef CONFIG_NFS_V4
-extern int nfs_callback_up(void);
+extern int nfs_callback_up(u32 minorversion, void *args);
 extern void nfs_callback_down(void);
 #else
 #define nfs_callback_up()	(0)

index c69424d..1c0cae9 100644
@@ -129,12 +129,6 @@ static struct nfs_client *nfs_alloc_client(const struct nfs_client_initdata *cl_
 
 	clp->rpc_ops = cl_init->rpc_ops;
 
-	if (cl_init->rpc_ops->version == 4) {
-		if (nfs_callback_up() < 0)
-			goto error_2;
-		__set_bit(NFS_CS_CALLBACK, &clp->cl_res_state);
-	}
-
 	atomic_set(&clp->cl_count, 1);
 	clp->cl_cons_state = NFS_CS_INITING;
 

@@ -144,7 +138,7 @@ static struct nfs_client *nfs_alloc_client(const struct nfs_client_initdata *cl_
 	if (cl_init->hostname) {
 		clp->cl_hostname = kstrdup(cl_init->hostname, GFP_KERNEL);
 		if (!clp->cl_hostname)
-			goto error_3;
+			goto error_cleanup;
 	}
 
 	INIT_LIST_HEAD(&clp->cl_superblocks);

@@ -167,10 +161,7 @@ static struct nfs_client *nfs_alloc_client(const struct nfs_client_initdata *cl_
 
 	return clp;
 
-error_3:
-	if (__test_and_clear_bit(NFS_CS_CALLBACK, &clp->cl_res_state))
-		nfs_callback_down();
-error_2:
+error_cleanup:
 	kfree(clp);
 error_0:
 	return NULL;

@@ -1008,6 +999,16 @@ static int nfs4_init_client(struct nfs_client *clp,
 		goto error;
 	memcpy(clp->cl_ipaddr, ip_addr, sizeof(clp->cl_ipaddr));
 
+	/* Start the callback server */
+	error = nfs_callback_up(clp->cl_minorversion,
+		clp->cl_rpcclient->cl_xprt);
+	if (error < 0) {
+		dprintk("%s: failed to start callback. Error = %d\n",
+			__func__, error);
+		goto error;
+	}
+	__set_bit(NFS_CS_CALLBACK, &clp->cl_res_state);
+
 	error = nfs_idmap_new(clp);
 	if (error < 0) {
 		dprintk("%s: failed to create idmapper. Error = %d\n",

index 3afe7fb..1d3b971 100644
@@ -81,6 +81,16 @@ struct svc_serv {
 	struct module *		sv_module;	/* optional module to count when
 						 * adding threads */
 	svc_thread_fn		sv_function;	/* main function for threads */
+#if defined(CONFIG_NFS_V4_1)
+	struct list_head 	sv_cb_list;	/* queue for callback requests
+						 * that arrive over the same
+						 * connection
+						 */
+	spinlock_t		sv_cb_lock;	/* protects the svc_cb_list */
+	wait_queue_head_t	sv_cb_waitq;	/* sleep here if there are no
+						 * entries in the svc_cb_list
+						 */
+#endif /* CONFIG_NFS_V4_1 */
 };
 
 /*

index 11fc71d..c5d3c7c 100644
@@ -97,6 +97,12 @@ struct rpc_rqst {
 
 	unsigned long		rq_xtime;	/* when transmitted */
 	int			rq_ntrans;
+
+#if defined(CONFIG_NFS_V4_1)
+	struct list_head	rq_bc_list;	/* Callback service list */
+	unsigned long		rq_bc_pa_state;	/* Backchannel prealloc state */
+	struct list_head	rq_bc_pa_list;	/* Backchannel prealloc list */
+#endif /* CONFIG_NFS_V4_1 */
 };
 #define rq_svec			rq_snd_buf.head
 #define rq_slen			rq_snd_buf.len

@@ -174,6 +180,14 @@ struct rpc_xprt {
 	spinlock_t		reserve_lock;	/* lock slot table */
 	u32			xid;		/* Next XID value to use */
 	struct rpc_task *	snd_task;	/* Task blocked in send */
+#if defined(CONFIG_NFS_V4_1)
+	struct svc_serv		*bc_serv;	/* The RPC service which will */
+						/* process the callback */
+	spinlock_t		bc_pa_lock;	/* Protects the preallocated */
+						/* items */
+	struct list_head	bc_pa_list;	/* List of preallocated */
+						/* backchannel rpc_rqst's */
+#endif /* CONFIG_NFS_V4_1 */
 	struct list_head	recv;
 
 	struct {

@@ -192,6 +206,14 @@ struct rpc_xprt {
 	const char		*address_strings[RPC_DISPLAY_MAX];
 };
 
+#if defined(CONFIG_NFS_V4_1)
+/*
+ * Backchannel flags
+ */
+#define	RPC_BC_PA_IN_USE	0x0001		/* Preallocated backchannel */
+						/* buffer in use */
+#endif /* CONFIG_NFS_V4_1 */
+
 struct xprt_create {
 	int			ident;		/* XPRT_TRANSPORT identifier */
 	struct sockaddr *	srcaddr;	/* optional local address */

index 29e401b..848a7af 100644
@@ -1034,6 +1034,11 @@ found:
 
 	INIT_LIST_HEAD(&xprt->free);
 	INIT_LIST_HEAD(&xprt->recv);
+#if defined(CONFIG_NFS_V4_1)
+	spin_lock_init(&xprt->bc_pa_lock);
+	INIT_LIST_HEAD(&xprt->bc_pa_list);
+#endif /* CONFIG_NFS_V4_1 */
+
 	INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
 	setup_timer(&xprt->timer, xprt_init_autodisconnect,
 			(unsigned long)xprt);

index c5d3c7c..035431b 100644
@@ -183,6 +183,7 @@ struct rpc_xprt {
 #if defined(CONFIG_NFS_V4_1)
 	struct svc_serv		*bc_serv;	/* The RPC service which will */
 						/* process the callback */
+	unsigned int		bc_alloc_count;	/* Total number of preallocs */
 	spinlock_t		bc_pa_lock;	/* Protects the preallocated */
 						/* items */
 	struct list_head	bc_pa_list;	/* List of preallocated */

268 lines not on this page · raw message

new file mode 100644
index 0000000..8ad9337
@@ -0,0 +1,40 @@
+/******************************************************************************
+
+(c) 2008 Network Appliance, Inc.  All Rights Reserved.
+
+Network Appliance provides this source code under the GPL v2 License.
+The GPL v2 license is available at
+http://opensource.org/licenses/gpl-license.php.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"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 COPYRIGHT OWNER 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.
+
+******************************************************************************/
+
+/*
+ * Functions to create and manage the backchannel
+ */
+
+#ifndef _LINUX_SUNRPC_BC_XPRT_H
+#define _LINUX_SUNRPC_BC_XPRT_H
+
+#include <linux/sunrpc/svcsock.h>
+#include <linux/sunrpc/xprt.h>
+
+#ifdef CONFIG_NFS_V4_1
+struct rpc_rqst *xprt_alloc_bc_request(struct rpc_xprt *xprt);
+void xprt_free_bc_request(struct rpc_rqst *req);
+int xprt_setup_backchannel(struct rpc_xprt *, unsigned int min_reqs);
+void xprt_destroy_backchannel(struct rpc_xprt *, int max_reqs);
+#endif /* CONFIG_NFS_V4_1 */
+#endif /* _LINUX_SUNRPC_BC_XPRT_H */
+

index 1ce05d7..6ccdec8 100644
@@ -34,6 +34,9 @@
 #include <linux/sunrpc/sched.h>
 #include <linux/sunrpc/xprtsock.h>
 #include <linux/file.h>
+#ifdef CONFIG_NFS_V4_1
+#include <linux/sunrpc/bc_xprt.h>
+#endif
 
 #include <net/sock.h>
 #include <net/checksum.h>

@@ -997,25 +1000,16 @@ static inline void xs_tcp_read_calldir(struct sock_xprt *transport,
 	xs_tcp_check_fraghdr(transport);
 }
 
-static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_reader *desc)
+static inline void xs_tcp_read_common(struct rpc_xprt *xprt,
+				     struct xdr_skb_reader *desc,
+				     struct rpc_rqst *req)
 {
-	struct sock_xprt *transport = container_of(xprt, struct sock_xprt, xprt);
-	struct rpc_rqst *req;
+	struct sock_xprt *transport =
+				container_of(xprt, struct sock_xprt, xprt);
 	struct xdr_buf *rcvbuf;
 	size_t len;
 	ssize_t r;
 
-	/* Find and lock the request corresponding to this xid */
-	spin_lock(&xprt->transport_lock);
-	req = xprt_lookup_rqst(xprt, transport->tcp_xid);
-	if (!req) {
-		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
-		dprintk("RPC:       XID %08x request not found!\n",
-				ntohl(transport->tcp_xid));
-		spin_unlock(&xprt->transport_lock);
-		return;
-	}
-
 	rcvbuf = &req->rq_private_buf;
 	len = desc->count;
 	if (len > transport->tcp_reclen - transport->tcp_offset) {

@@ -1053,7 +1047,7 @@ static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_rea
 				"tcp_offset = %u, tcp_reclen = %u\n",
 				xprt, transport->tcp_copied,
 				transport->tcp_offset, transport->tcp_reclen);
-		goto out;
+		return;
 	}
 
 	dprintk("RPC:       XID %08x read %Zd bytes\n",

@@ -1069,11 +1063,120 @@ static inline void xs_tcp_read_request(struct rpc_xprt *xprt, struct xdr_skb_rea
 			transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
 	}
 
-out:
+	return;
+}
+
+/*
+ * Finds the request corresponding to the RPC xid and invokes the common
+ * tcp read code to read the data.
+ */
+static inline int xs_tcp_read_reply(struct rpc_xprt *xprt,
+				    struct xdr_skb_reader *desc)
+{
+	struct sock_xprt *transport =
+				container_of(xprt, struct sock_xprt, xprt);
+	struct rpc_rqst *req;
+
+	dprintk("RPC:       read reply XID %08x\n", ntohl(transport->tcp_xid));
+
+	/* Find and lock the request corresponding to this xid */
+	spin_lock(&xprt->transport_lock);
+	req = xprt_lookup_rqst(xprt, transport->tcp_xid);
+	if (!req) {
+		dprintk("RPC:       XID %08x request not found!\n",
+				ntohl(transport->tcp_xid));
+		spin_unlock(&xprt->transport_lock);
+		return -1;
+	}
+
+	xs_tcp_read_common(xprt, desc, req);
+
 	if (!(transport->tcp_flags & TCP_RCV_COPY_DATA))
 		xprt_complete_rqst(req->rq_task, transport->tcp_copied);
+
 	spin_unlock(&xprt->transport_lock);
-	xs_tcp_check_fraghdr(transport);
+	return 0;
+}
+
+#if defined(CONFIG_NFS_V4_1)
+/*
+ * Obtains an rpc_rqst previously allocated and invokes the common
+ * tcp read code to read the data.  The result is placed in the callback
+ * queue.
+ * If we're unable to obtain the rpc_rqst we schedule the closing of the
+ * connection and return -1.
+ */
+static inline int xs_tcp_read_callback(struct rpc_xprt *xprt,
+				       struct xdr_skb_reader *desc)
+{
+	struct sock_xprt *transport =
+				container_of(xprt, struct sock_xprt, xprt);
+	struct rpc_rqst *req;
+
+	req = xprt_alloc_bc_request(xprt);
+	if (req == NULL) {
+		/*
+		 * Schedule an autoclose RPC call
+		 */
+		printk(KERN_WARNING "Callback slot table overflowed\n");
+		set_bit(XPRT_CLOSE_WAIT, &xprt->state);
+		if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
+			queue_work(rpciod_workqueue, &xprt->task_cleanup);
+		return -1;
+	}
+
+	req->rq_xid = transport->tcp_xid;
+	dprintk("RPC:       read callback  XID %08x\n", ntohl(req->rq_xid));
+	xs_tcp_read_common(xprt, desc, req);
+
+	if (!(transport->tcp_flags & TCP_RCV_COPY_DATA)) {
+		struct svc_serv *bc_serv = xprt->bc_serv;
+
+		/*
+		 * Add callback request to callback list.  The callback
+		 * service sleeps on the sv_cb_waitq waiting for new
+		 * requests.  Wake it up after adding enqueing the
+		 * request.
+		 */
+		dprintk("RPC:       add callback request to list\n");
+		spin_lock(&bc_serv->sv_cb_lock);
+		list_add(&req->rq_bc_list, &bc_serv->sv_cb_list);
+		spin_unlock(&bc_serv->sv_cb_lock);
+		wake_up(&bc_serv->sv_cb_waitq);
+	}
+
+	return 0;
+}
+#endif /* CONFIG_NFS_V4_1 */
+
+/*
+ * Read data off the transport.  This can be either an RPC_CALL or an
+ * RPC_REPLY.  Relay the processing to helper functions.
+ */
+static void xs_tcp_read_data(struct rpc_xprt *xprt,
+				    struct xdr_skb_reader *desc)
+{
+	struct sock_xprt *transport =
+				container_of(xprt, struct sock_xprt, xprt);
+	int status;
+
+#if defined(CONFIG_NFS_V4_1)
+	status = (transport->tcp_calldir == RPC_REPLY) ?
+		xs_tcp_read_reply(xprt, desc) :
+		xs_tcp_read_callback(xprt, desc);
+#else
+	status = xs_tcp_read_reply(xprt, desc);
+#endif /* CONFIG_NFS_V4_1 */
+
+	if (status == 0)
+		xs_tcp_check_fraghdr(transport);
+	else {
+		/*
+		 * The transport_lock protects the request handling.
+		 * There's no need to hold it to update the tcp_flags.
+		 */
+		transport->tcp_flags &= ~TCP_RCV_COPY_DATA;
+	}
 }
 
 static inline void xs_tcp_read_discard(struct sock_xprt *transport, struct xdr_skb_reader *desc)

@@ -1120,7 +1223,7 @@ static int xs_tcp_data_recv(read_descriptor_t *rd_desc, struct sk_buff *skb, uns
 		}
 		/* Read in the request data */
 		if (transport->tcp_flags & TCP_RCV_COPY_DATA) {
-			xs_tcp_read_request(xprt, &desc);
+			xs_tcp_read_data(xprt, &desc);
 			continue;
 		}
 		/* Skip over any trailing bytes on short reads */

index 1773768..4010977 100644
@@ -210,6 +210,8 @@ struct rpc_wait_queue {
  */
 struct rpc_task *rpc_new_task(const struct rpc_task_setup *);
 struct rpc_task *rpc_run_task(const struct rpc_task_setup *);
+struct rpc_task *rpc_run_bc_task(struct rpc_rqst *req,
+				const struct rpc_call_ops *ops);
 void		rpc_put_task(struct rpc_task *);
 void		rpc_exit_task(struct rpc_task *);
 void		rpc_release_calldata(const struct rpc_call_ops *, void *);

index 035431b..45a92e2 100644
@@ -215,6 +215,18 @@ struct rpc_xprt {
 						/* buffer in use */
 #endif /* CONFIG_NFS_V4_1 */
 
+#if defined(CONFIG_NFS_V4_1)
+static inline int bc_prealloc(struct rpc_rqst *req)
+{
+	return test_bit(RPC_BC_PA_IN_USE, &req->rq_bc_pa_state);
+}
+#else
+static inline int bc_prealloc(struct rpc_rqst *req)
+{
+	return 0;
+}
+#endif /* CONFIG_NFS_V4_1 */
+
 struct xprt_create {
 	int			ident;		/* XPRT_TRANSPORT identifier */
 	struct sockaddr *	srcaddr;	/* optional local address */

73 lines not on this page · raw message

index 50b049c..89dc28e 100644
@@ -141,12 +141,14 @@ EXPORT_SYMBOL_GPL(rpc_free_iostats);
 void rpc_count_iostats(struct rpc_task *task)
 {
 	struct rpc_rqst *req = task->tk_rqstp;
-	struct rpc_iostats *stats = task->tk_client->cl_metrics;
+	struct rpc_iostats *stats;
 	struct rpc_iostats *op_metrics;
 	long rtt, execute, queue;
 
-	if (!stats || !req)
+	if (!task->tk_client || task->tk_client->cl_metrics || !req)
 		return;
+
+	stats = task->tk_client->cl_metrics;
 	op_metrics = &stats[task->tk_msg.rpc_proc->p_statidx];
 
 	op_metrics->om_ops++;

new file mode 100644
index 0000000..b462de4
@@ -0,0 +1,35 @@
+/******************************************************************************
+
+(c) 2008 Network Appliance, Inc.  All Rights Reserved.
+
+Network Appliance provides this source code under the GPL v2 License.
+The GPL v2 license is available at
+http://opensource.org/licenses/gpl-license.php.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"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 COPYRIGHT OWNER 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.
+
+******************************************************************************/
+
+/*
+ * Functions and macros used internally by RPC
+ */
+
+#ifndef _NET_SUNRPC_SUNRPC_H
+#define _NET_SUNRPC_SUNRPC_H
+
+#define rpc_reply_expected(task) \
+	(((task)->tk_msg.rpc_proc != NULL) && \
+	((task)->tk_msg.rpc_proc->p_decode != NULL))
+
+#endif /* _NET_SUNRPC_SUNRPC_H */
+

index 848a7af..cc1e5f2 100644
@@ -12,8 +12,9 @@
  *  -	Next, the caller puts together the RPC message, stuffs it into
  *	the request struct, and calls xprt_transmit().
  *  -	xprt_transmit sends the message and installs the caller on the
- *	transport's wait list. At the same time, it installs a timer that
- *	is run after the packet's timeout has expired.
+ *	transport's wait list. At the same time, if a reply is expected,
+ *	it installs a timer that is run after the packet's timeout has
+ *	expired.
  *  -	When a packet arrives, the data_ready handler walks the list of
  *	pending requests for that transport. If a matching XID is found, the
  *	caller is woken up, and the timer removed.

@@ -46,6 +47,8 @@
 #include <linux/sunrpc/clnt.h>
 #include <linux/sunrpc/metrics.h>
 
+#include "sunrpc.h"
+
 /*
  * Local variables
  */

@@ -852,7 +855,10 @@ void xprt_transmit(struct rpc_task *task)
 	dprintk("RPC: %5u xprt_transmit(%u)\n", task->tk_pid, req->rq_slen);
 
 	if (!req->rq_received) {
-		if (list_empty(&req->rq_list)) {
+		if (list_empty(&req->rq_list) && rpc_reply_expected(task)) {
+			/*
+			 * Add to the list only if we're expecting a reply
+			 */
 			spin_lock_bh(&xprt->transport_lock);
 			/* Update the softirq receive buffer */
 			memcpy(&req->rq_private_buf, &req->rq_rcv_buf,

@@ -883,8 +889,13 @@ void xprt_transmit(struct rpc_task *task)
 		/* Don't race with disconnect */
 		if (!xprt_connected(xprt))
 			task->tk_status = -ENOTCONN;
-		else if (!req->rq_received)
+		else if (!req->rq_received && rpc_reply_expected(task)) {
+			/*
+			 * Sleep on the pending queue since
+			 * we're expecting a reply.
+			 */
 			rpc_sleep_on(&xprt->pending, task, xprt_timer);
+		}
 		spin_unlock_bh(&xprt->transport_lock);
 		return;
 	}

@@ -967,11 +978,15 @@ static void xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
  */
 void xprt_release(struct rpc_task *task)
 {
-	struct rpc_xprt	*xprt = task->tk_xprt;
+	struct rpc_xprt	*xprt;
 	struct rpc_rqst	*req;
+	int prealloc;
 
+	BUG_ON(atomic_read(&task->tk_count) < 0);
 	if (!(req = task->tk_rqstp))
 		return;
+	prealloc = bc_prealloc(req);	/* Preallocated backchannel request? */
+	xprt = req->rq_xprt;
 	rpc_count_iostats(task);
 	spin_lock_bh(&xprt->transport_lock);
 	xprt->ops->release_xprt(xprt, task);

@@ -984,10 +999,19 @@ void xprt_release(struct rpc_task *task)
 		mod_timer(&xprt->timer,
 				xprt->last_used + xprt->idle_timeout);
 	spin_unlock_bh(&xprt->transport_lock);
-	xprt->ops->buf_free(req->rq_buffer);
+	if (!bc_prealloc(req))
+		xprt->ops->buf_free(req->rq_buffer);
 	task->tk_rqstp = NULL;
 	if (req->rq_release_snd_buf)
 		req->rq_release_snd_buf(req);
+
+	/*
+	 * Early exit if this is a backchannel preallocated request.
+	 * There is no need to have it added to the RPC slot list.
+	 */
+	if (prealloc)
+		return;
+
 	memset(req, 0, sizeof(*req));	/* mark unused */
 
 	dprintk("RPC: %5u release request %p\n", task->tk_pid, req);

index 8ad9337..9dbeb77 100644
@@ -29,12 +29,15 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 #include <linux/sunrpc/svcsock.h>
 #include <linux/sunrpc/xprt.h>
+#include <linux/sunrpc/sched.h>
 
 #ifdef CONFIG_NFS_V4_1
 struct rpc_rqst *xprt_alloc_bc_request(struct rpc_xprt *xprt);
 void xprt_free_bc_request(struct rpc_rqst *req);
 int xprt_setup_backchannel(struct rpc_xprt *, unsigned int min_reqs);
 void xprt_destroy_backchannel(struct rpc_xprt *, int max_reqs);
+void bc_release_request(struct rpc_task *);
+int bc_send(struct rpc_rqst *req);
 #endif /* CONFIG_NFS_V4_1 */
 #endif /* _LINUX_SUNRPC_BC_XPRT_H */
 

81 lines not on this page · raw message

index 6ccdec8..8513d50 100644
@@ -2046,6 +2046,9 @@ static struct rpc_xprt_ops xs_tcp_ops = {
 	.buf_free		= rpc_free,
 	.send_request		= xs_tcp_send_request,
 	.set_retrans_timeout	= xprt_set_retrans_timeout_def,
+#if defined(CONFIG_NFS_V4_1)
+	.release_request	= bc_release_request,
+#endif /* CONFIG_NFS_V4_1 */
 	.close			= xs_tcp_shutdown,
 	.destroy		= xs_destroy,
 	.print_stats		= xs_tcp_print_stats,

index f5418ac..fcadd8e 100644
@@ -1204,7 +1204,8 @@ call_status(struct rpc_task *task)
 		break;
 	case -ECONNREFUSED:
 	case -ENOTCONN:
-		rpc_force_rebind(clnt);
+		if (clnt)
+			rpc_force_rebind(clnt);
 		task->tk_action = call_bind;
 		break;
 	case -EAGAIN:

@@ -1215,9 +1216,10 @@ call_status(struct rpc_task *task)
 		rpc_exit(task, status);
 		break;
 	default:
-		if (clnt->cl_chatty)
+		if (!clnt || clnt->cl_chatty)
 			printk("%s: RPC call returned error %d\n",
-			       clnt->cl_protname, -status);
+			       clnt ? clnt->cl_protname : "<unknown protocol>",
+			       -status);
 		rpc_exit(task, status);
 	}
 }

index e792706..2f5acd3 100644
@@ -132,7 +132,7 @@ nfs4_callback_up(struct svc_serv *serv)
 int nfs_callback_up(u32 minorversion, void *args)
 {
 	struct svc_serv *serv = NULL;
-	struct svc_rqst *rqstp;
+	struct svc_rqst *rqst;
 	int (* callback_svc)(void *vrqstp);
 	char svc_name[12];
 	int ret = 0;

@@ -150,20 +150,20 @@ int nfs_callback_up(u32 minorversion, void *args)
 	/* FIXME: either 4.0 or 4.1 callback service can be up at a time
 	 * need to monitor and control them both */
 	if (!minorversion) {
-		rqstp = nfs4_callback_up(serv);
+		rqst = nfs4_callback_up(serv);
 		callback_svc = nfs4_callback_svc;
 	} else {
 		BUG();	/* for now */
 	}
-	if (IS_ERR(rqstp)) {
-		ret = PTR_ERR(rqstp);
+	if (IS_ERR(rqst)) {
+		ret = PTR_ERR(rqst);
 		goto out_err;
 	}
 
 	svc_sock_update_bufs(serv);
 
 	sprintf(svc_name, "nfsv4.%u-svc", minorversion);
-	nfs_callback_info.rqst = rqstp;
+	nfs_callback_info.rqst = rqst;
 	nfs_callback_info.task = kthread_run(callback_svc,
 					     nfs_callback_info.rqst,
 					     svc_name);

131 lines not on this page · raw message

index e4efeeb..906b992 100644
@@ -25,6 +25,7 @@
 #include <linux/sunrpc/stats.h>
 #include <linux/sunrpc/svcsock.h>
 #include <linux/sunrpc/clnt.h>
+#include <linux/sunrpc/bc_xprt.h>
 
 #define RPCDBG_FACILITY	RPCDBG_SVCDSP
 

@@ -1253,6 +1254,54 @@ svc_process(struct svc_rqst *rqstp)
 	return svc_send(rqstp);
 }
 
+#if defined(CONFIG_NFS_V4_1)
+/*
+ * Process a backchannel RPC request that arrived over an existing
+ * outbound connection
+ */
+int
+bc_svc_process(struct svc_serv *serv, struct rpc_rqst *req,
+	       struct svc_rqst *rqstp)
+{
+	struct kvec	*argv = &rqstp->rq_arg.head[0];
+	struct kvec	*resv = &rqstp->rq_res.head[0];
+	int 		error;
+
+	/* Build the svc_rqst used by the common processing routine */
+	rqstp->rq_xid = req->rq_xid;
+	rqstp->rq_prot = req->rq_xprt->prot;
+	rqstp->rq_server = serv;
+
+	rqstp->rq_addrlen = sizeof(req->rq_xprt->addr);
+	memcpy(&rqstp->rq_addr, &req->rq_xprt->addr, rqstp->rq_addrlen);
+	memcpy(&rqstp->rq_arg, &req->rq_private_buf, sizeof(rqstp->rq_arg));
+	memcpy(&rqstp->rq_res, &req->rq_snd_buf, sizeof(rqstp->rq_res));
+
+	/* reset result send buffer "put" position */
+	resv->iov_len = 0;
+
+	if (rqstp->rq_prot != IPPROTO_TCP) {
+		printk(KERN_ERR "No support for Non-TCP transports!\n");
+		BUG();
+	}
+
+	/*
+	 * Skip the next two words because they've already been
+	 * processed in the trasport
+	 */
+	svc_getu32(argv);	/* XID */
+	svc_getnl(argv);	/* CALLDIR */
+
+	error = svc_process_common(rqstp, argv, resv);
+	if (error <= 0)
+		return error;
+
+	memcpy(&req->rq_snd_buf, &rqstp->rq_res, sizeof(req->rq_snd_buf));
+	return bc_send(req);
+}
+EXPORT_SYMBOL(bc_svc_process);
+#endif /* CONFIG_NFS_V4_1 */
+
 /*
  * Return (transport-specific) limit on the rpc payload.
  */

index 2f5acd3..86f4d88 100644
@@ -16,6 +16,9 @@
 #include <linux/mutex.h>
 #include <linux/freezer.h>
 #include <linux/kthread.h>
+#if defined(CONFIG_NFS_V4_1)
+#include <linux/sunrpc/bc_xprt.h>
+#endif
 
 #include <net/inet_sock.h>
 

@@ -126,6 +129,69 @@ nfs4_callback_up(struct svc_serv *serv)
 	return svc_prepare_thread(serv, &serv->sv_pools[0]);
 }
 
+#if defined(CONFIG_NFS_V4_1)
+/*
+ * The callback service for NFSv4.1 callbacks
+ */
+static int
+nfs41_callback_svc(void *vrqstp)
+{
+	struct svc_rqst *rqstp = vrqstp;
+	struct svc_serv *serv = rqstp->rq_server;
+	struct rpc_rqst *req;
+	int error;
+	DEFINE_WAIT(wq);
+
+	set_freezable();
+
+	/*
+	 * FIXME: do we really need to run this under the BKL? If so, please
+	 * add a comment about what it's intended to protect.
+	 */
+	lock_kernel();
+	while (!kthread_should_stop()) {
+		prepare_to_wait(&serv->sv_cb_waitq, &wq, TASK_INTERRUPTIBLE);
+		spin_lock_bh(&serv->sv_cb_lock);
+		if (!list_empty(&serv->sv_cb_list)) {
+			req = list_first_entry(&serv->sv_cb_list,
+					struct rpc_rqst, rq_bc_list);
+			list_del(&req->rq_bc_list);
+			spin_unlock_bh(&serv->sv_cb_lock);
+			dprintk("Invoking bc_svc_process()\n");
+			error = bc_svc_process(serv, req, rqstp);
+			dprintk("bc_svc_process() returned w/ error code= %d\n",
+				error);
+		} else {
+			spin_unlock_bh(&serv->sv_cb_lock);
+			schedule();
+		}
+		finish_wait(&serv->sv_cb_waitq, &wq);
+	}
+	unlock_kernel();
+	nfs_callback_info.task = NULL;
+	svc_exit_thread(rqstp);
+	return 0;
+}
+
+/*
+ * Bring up the NFSv4.1 callback service
+ */
+struct svc_rqst *
+nfs41_callback_up(struct svc_serv *serv, struct rpc_xprt *xprt)
+{
+	/*
+	 * Save the svc_serv in the transport so that it can
+	 * be referenced when the session backchannel is initialized
+	 */
+	xprt->bc_serv = serv;
+
+	INIT_LIST_HEAD(&serv->sv_cb_list);
+	spin_lock_init(&serv->sv_cb_lock);
+	init_waitqueue_head(&serv->sv_cb_waitq);
+	return svc_prepare_thread(serv, &serv->sv_pools[0]);
+}
+#endif /* CONFIG_NFS_V4_1 */
+
 /*
  * Bring up the callback thread if it is not already up.
  */

@@ -136,6 +202,9 @@ int nfs_callback_up(u32 minorversion, void *args)
 	int (* callback_svc)(void *vrqstp);
 	char svc_name[12];
 	int ret = 0;
+#if defined(CONFIG_NFS_V4_1)
+	struct rpc_xprt *xprt = (struct rpc_xprt *)args;
+#endif /* CONFIG_NFS_V4_1 */
 
 	mutex_lock(&nfs_callback_mutex);
 	if (nfs_callback_info.users++ || nfs_callback_info.task != NULL)

@@ -153,7 +222,12 @@ int nfs_callback_up(u32 minorversion, void *args)
 		rqst = nfs4_callback_up(serv);
 		callback_svc = nfs4_callback_svc;
 	} else {
-		BUG();	/* for now */
+#if defined(CONFIG_NFS_V4_1)
+		rqst = nfs41_callback_up(serv, xprt);
+		callback_svc = nfs41_callback_svc;
+#else  /* CONFIG_NFS_V4_1 */
+		BUG();
+#endif /* CONFIG_NFS_V4_1 */
 	}
 	if (IS_ERR(rqst)) {
 		ret = PTR_ERR(rqst);

index a2b3ce4..49ea368 100644
@@ -70,6 +70,14 @@ extern void nfs_callback_down(void);
 #define nfs_callback_down()	do {} while(0)
 #endif
 
+#ifdef CONFIG_NFS_V4_1
+/*
+ * Callbacks are expected to not cause substantial latency,
+ * so we limit their concurrency to 1.
+ */
+#define NFS41_BC_MIN_CALLBACKS 1
+#endif /* CONFIG_NFS_V4_1 */
+
 extern unsigned int nfs_callback_set_tcpport;
 extern unsigned short nfs_callback_tcpport;
 

index 1d3b971..1bb51dc 100644
@@ -403,6 +403,8 @@ struct svc_serv *  svc_create_pooled(struct svc_program *, unsigned int,
 int		   svc_set_num_threads(struct svc_serv *, struct svc_pool *, int);
 void		   svc_destroy(struct svc_serv *);
 int		   svc_process(struct svc_rqst *);
+int		   bc_svc_process(struct svc_serv *, struct rpc_rqst *,
+			struct svc_rqst *);
 int		   svc_register(const struct svc_serv *, const unsigned short,
 				const unsigned short);
 

index 483e103..6bb1ec4 100644
@@ -42,6 +42,8 @@ int		svc_sock_names(char *buf, struct svc_serv *serv, char *toclose);
 int		svc_addsock(struct svc_serv *serv, int fd, char *name_return);
 void		svc_init_xprt_sock(void);
 void		svc_cleanup_xprt_sock(void);
+struct svc_xprt *svc_sock_create(struct svc_serv *serv, int prot);
+void		svc_sock_destroy(struct svc_xprt *);
 
 /*
  * svc_makesock socket characteristics

index 95293f5..c2b6fa4 100644
@@ -1298,3 +1298,42 @@ static void svc_sock_free(struct svc_xprt *xprt)
 		sock_release(svsk->sk_sock);
 	kfree(svsk);
 }
+
+/*
+ * Create a svc_xprt.
+ *
+ * For internal use only (e.g. nfsv4.1 backchannel).
+ * Callers should typically use the xpo_create() method.
+ */
+struct svc_xprt *svc_sock_create(struct svc_serv *serv, int prot)
+{
+	struct svc_sock *svsk;
+	struct svc_xprt *xprt = NULL;
+
+	dprintk("svc: %s\n", __func__);
+	svsk = kzalloc(sizeof(*svsk), GFP_KERNEL);
+	if (!svsk)
+		goto out;
+
+	xprt = &svsk->sk_xprt;
+	if (prot == IPPROTO_TCP)
+		svc_xprt_init(&svc_tcp_class, xprt, serv);
+	else if (prot == IPPROTO_UDP)
+		svc_xprt_init(&svc_udp_class, xprt, serv);
+	else
+		BUG();
+out:
+	dprintk("svc: %s return %p\n", __func__, xprt);
+	return xprt;
+}
+EXPORT_SYMBOL_GPL(svc_sock_create);
+
+/*
+ * Destroy a svc_sock.
+ */
+void svc_sock_destroy(struct svc_xprt *xprt)
+{
+	if (xprt)
+		kfree(container_of(xprt, struct svc_sock, sk_xprt));
+}
+EXPORT_SYMBOL_GPL(svc_sock_destroy);

index 1bb51dc..7653dd3 100644
@@ -90,6 +90,7 @@ struct svc_serv {
 	wait_queue_head_t	sv_cb_waitq;	/* sleep here if there are no
 						 * entries in the svc_cb_list
 						 */
+	struct svc_xprt		*bc_xprt;
 #endif /* CONFIG_NFS_V4_1 */
 };
 

index 906b992..b84ae54 100644
@@ -489,6 +489,10 @@ svc_destroy(struct svc_serv *serv)
 	if (svc_serv_is_pooled(serv))
 		svc_pool_map_put();
 
+#if defined(CONFIG_NFS_V4_1)
+	svc_sock_destroy(serv->bc_xprt);
+#endif /* CONFIG_NFS_V4_1 */
+
 	svc_unregister(serv);
 	kfree(serv->sv_pools);
 	kfree(serv);

index 86f4d88..fc30f87 100644
@@ -179,16 +179,31 @@ nfs41_callback_svc(void *vrqstp)
 struct svc_rqst *
 nfs41_callback_up(struct svc_serv *serv, struct rpc_xprt *xprt)
 {
+	struct svc_xprt *bc_xprt;
+	struct svc_rqst *rqstp = ERR_PTR(-ENOMEM);
+
+	dprintk("--> %s\n", __func__);
+	/* Create a svc_sock for the service */
+	bc_xprt = svc_sock_create(serv, xprt->prot);
+	if (!bc_xprt)
+		goto out;
+
 	/*
 	 * Save the svc_serv in the transport so that it can
 	 * be referenced when the session backchannel is initialized
 	 */
+	serv->bc_xprt = bc_xprt;
 	xprt->bc_serv = serv;
 
 	INIT_LIST_HEAD(&serv->sv_cb_list);
 	spin_lock_init(&serv->sv_cb_lock);
 	init_waitqueue_head(&serv->sv_cb_waitq);
-	return svc_prepare_thread(serv, &serv->sv_pools[0]);
+	rqstp = svc_prepare_thread(serv, &serv->sv_pools[0]);
+	if (IS_ERR(rqstp))
+		svc_sock_destroy(bc_xprt);
+out:
+	dprintk("--> %s return %p\n", __func__, rqstp);
+	return rqstp;
 }
 #endif /* CONFIG_NFS_V4_1 */
 

index b84ae54..1b95de7 100644
@@ -1272,6 +1272,7 @@ bc_svc_process(struct svc_serv *serv, struct rpc_rqst *req,
 	int 		error;
 
 	/* Build the svc_rqst used by the common processing routine */
+	rqstp->rq_xprt = serv->bc_xprt;
 	rqstp->rq_xid = req->rq_xid;
 	rqstp->rq_prot = req->rq_xprt->prot;
 	rqstp->rq_server = serv;

index fc30f87..be8ac50 100644
@@ -30,6 +30,7 @@
 
 struct nfs_callback_data {
 	unsigned int users;
+	struct svc_serv *serv;
 	struct svc_rqst *rqst;
 	struct task_struct *task;
 };

@@ -252,6 +253,7 @@ int nfs_callback_up(u32 minorversion, void *args)
 	svc_sock_update_bufs(serv);
 
 	sprintf(svc_name, "nfsv4.%u-svc", minorversion);
+	nfs_callback_info.serv = serv;
 	nfs_callback_info.rqst = rqst;
 	nfs_callback_info.task = kthread_run(callback_svc,
 					     nfs_callback_info.rqst,

@@ -259,6 +261,7 @@ int nfs_callback_up(u32 minorversion, void *args)
 	if (IS_ERR(nfs_callback_info.task)) {
 		ret = PTR_ERR(nfs_callback_info.task);
 		svc_exit_thread(nfs_callback_info.rqst);
+		nfs_callback_info.serv = NULL;
 		nfs_callback_info.rqst = NULL;
 		nfs_callback_info.task = NULL;
 		goto out_err;

@@ -291,6 +294,7 @@ void nfs_callback_down(void)
 	if (nfs_callback_info.users == 0 && nfs_callback_info.task != NULL) {
 		kthread_stop(nfs_callback_info.task);
 		svc_exit_thread(nfs_callback_info.rqst);
+		nfs_callback_info.serv = NULL;
 		nfs_callback_info.rqst = NULL;
 		nfs_callback_info.task = NULL;
 	}

index be8ac50..036d66a 100644
@@ -223,8 +223,13 @@ int nfs_callback_up(u32 minorversion, void *args)
 #endif /* CONFIG_NFS_V4_1 */
 
 	mutex_lock(&nfs_callback_mutex);
-	if (nfs_callback_info.users++ || nfs_callback_info.task != NULL)
+	if (nfs_callback_info.users++ || nfs_callback_info.task != NULL) {
+#if defined(CONFIG_NFS_V4_1)
+		if (minorversion)
+			xprt->bc_serv = nfs_callback_info.serv;
+#endif /* CONFIG_NFS_V4_1 */
 		goto out;
+	}
 	serv = svc_create(&nfs4_callback_program, NFS4_CALLBACK_BUFSIZE,
 				nfs_callback_family, NULL);
 	if (!serv) {

index 036d66a..ff9ce3e 100644
@@ -35,7 +35,7 @@ struct nfs_callback_data {
 	struct task_struct *task;
 };
 
-static struct nfs_callback_data nfs_callback_info;
+static struct nfs_callback_data nfs_callback_info[NFS4_MAX_MINOR_VERSION + 1];
 static DEFINE_MUTEX(nfs_callback_mutex);
 static struct svc_program nfs4_callback_program;
 

@@ -68,7 +68,7 @@ module_param_call(callback_tcpport, param_set_port, param_get_int,
 		 &nfs_callback_set_tcpport, 0644);
 
 /*
- * This is the callback kernel thread.
+ * This is the NFSv4 callback kernel thread.
  */
 static int
 nfs4_callback_svc(void *vrqstp)

@@ -169,8 +169,6 @@ nfs41_callback_svc(void *vrqstp)
 		finish_wait(&serv->sv_cb_waitq, &wq);
 	}
 	unlock_kernel();
-	nfs_callback_info.task = NULL;
-	svc_exit_thread(rqstp);
 	return 0;
 }
 

@@ -216,6 +214,7 @@ int nfs_callback_up(u32 minorversion, void *args)
 	struct svc_serv *serv = NULL;
 	struct svc_rqst *rqst;
 	int (* callback_svc)(void *vrqstp);
+	struct nfs_callback_data *cb_info = &nfs_callback_info[minorversion];
 	char svc_name[12];
 	int ret = 0;
 #if defined(CONFIG_NFS_V4_1)

@@ -223,10 +222,10 @@ int nfs_callback_up(u32 minorversion, void *args)
 #endif /* CONFIG_NFS_V4_1 */
 
 	mutex_lock(&nfs_callback_mutex);
-	if (nfs_callback_info.users++ || nfs_callback_info.task != NULL) {
+	if (cb_info->users++ || cb_info->task != NULL) {
 #if defined(CONFIG_NFS_V4_1)
 		if (minorversion)
-			xprt->bc_serv = nfs_callback_info.serv;
+			xprt->bc_serv = cb_info->serv;
 #endif /* CONFIG_NFS_V4_1 */
 		goto out;
 	}

@@ -258,17 +257,14 @@ int nfs_callback_up(u32 minorversion, void *args)
 	svc_sock_update_bufs(serv);
 
 	sprintf(svc_name, "nfsv4.%u-svc", minorversion);
-	nfs_callback_info.serv = serv;
-	nfs_callback_info.rqst = rqst;
-	nfs_callback_info.task = kthread_run(callback_svc,
-					     nfs_callback_info.rqst,
-					     svc_name);
-	if (IS_ERR(nfs_callback_info.task)) {
-		ret = PTR_ERR(nfs_callback_info.task);
-		svc_exit_thread(nfs_callback_info.rqst);
-		nfs_callback_info.serv = NULL;
-		nfs_callback_info.rqst = NULL;
-		nfs_callback_info.task = NULL;
+	cb_info->serv = serv;
+	cb_info->rqst = rqst;
+	cb_info->task = kthread_run(callback_svc, cb_info->rqst, svc_name);
+	if (IS_ERR(cb_info->task)) {
+		ret = PTR_ERR(cb_info->task);
+		svc_exit_thread(cb_info->rqst);
+		cb_info->rqst = NULL;
+		cb_info->task = NULL;
 		goto out_err;
 	}
 out:

@@ -285,23 +281,25 @@ out:
 out_err:
 	dprintk("NFS: Couldn't create callback socket or server thread; "
 		"err = %d\n", ret);
-	nfs_callback_info.users--;
+	cb_info->users--;
 	goto out;
 }
 
 /*
  * Kill the callback thread if it's no longer being used.
  */
-void nfs_callback_down(void)
+void nfs_callback_down(int minorversion)
 {
+	struct nfs_callback_data *cb_info = &nfs_callback_info[minorversion];
+
 	mutex_lock(&nfs_callback_mutex);
-	nfs_callback_info.users--;
-	if (nfs_callback_info.users == 0 && nfs_callback_info.task != NULL) {
-		kthread_stop(nfs_callback_info.task);
-		svc_exit_thread(nfs_callback_info.rqst);
-		nfs_callback_info.serv = NULL;
-		nfs_callback_info.rqst = NULL;
-		nfs_callback_info.task = NULL;
+	cb_info->users--;
+	if (cb_info->users == 0 && cb_info->task != NULL) {
+		kthread_stop(cb_info->task);
+		svc_exit_thread(cb_info->rqst);
+		cb_info->serv = NULL;
+		cb_info->rqst = NULL;
+		cb_info->task = NULL;
 	}
 	mutex_unlock(&nfs_callback_mutex);
 }

index 49ea368..4c69b27 100644
@@ -64,11 +64,8 @@ extern __be32 nfs4_callback_recall(struct cb_recallargs *args, void *dummy);
 
 #ifdef CONFIG_NFS_V4
 extern int nfs_callback_up(u32 minorversion, void *args);
-extern void nfs_callback_down(void);
-#else
-#define nfs_callback_up()	(0)
-#define nfs_callback_down()	do {} while(0)
-#endif
+extern void nfs_callback_down(int minorversion);
+#endif /* CONFIG_NFS_V4 */
 
 #ifdef CONFIG_NFS_V4_1
 /*

index 1c0cae9..dc5b6e7 100644
@@ -193,8 +193,10 @@ static void nfs_free_client(struct nfs_client *clp)
 	if (!IS_ERR(clp->cl_rpcclient))
 		rpc_shutdown_client(clp->cl_rpcclient);
 
+#ifdef CONFIG_NFS_V4
 	if (__test_and_clear_bit(NFS_CS_CALLBACK, &clp->cl_res_state))
-		nfs_callback_down();
+		nfs_callback_down(clp->cl_minorversion);
+#endif /* CONFIG_NFS_V4 */
 
 	if (clp->cl_machine_cred != NULL)
 		put_rpccred(clp->cl_machine_cred);

index dc5b6e7..2a24d43 100644
@@ -39,6 +39,7 @@
 #include <linux/nfs_xdr.h>
 #if defined(CONFIG_NFS_V4_1)
 #include <linux/nfs41_session_recovery.h>
+#include <linux/sunrpc/bc_xprt.h>
 #endif /* CONFIG_NFS_V4_1 */
 
 #include <asm/system.h>

@@ -1125,7 +1126,7 @@ error:
 }
 
 /*
- * Allocate and initialize a session if required
+ * Allocate and initialize a session if required, including its backchannel.
  */
 int nfs4_init_session(struct nfs_client *clp, struct rpc_clnt *clnt)
 {

@@ -1145,6 +1146,12 @@ int nfs4_init_session(struct nfs_client *clp, struct rpc_clnt *clnt)
 		else {
 			session->clnt = clnt;
 			session->clp = clp;
+
+			error = xprt_setup_backchannel(clnt->cl_xprt,
+					NFS41_BC_MIN_CALLBACKS);
+			if (error < 0)
+				/* nfs4_put session sets the session to NULL */
+				nfs4_put_session(&session);
 		}
 		clp->cl_session = session;
 	}