[PATCH] Fix brown paper bag race in RPC receive code

Changeset 1.403.142.29 introduces a pretty nasty race into the RPC
code. Once we've decoded the RPC reply, it needs to be protected
against being overwritten by any resends.

The following patch achieves this by ensuring that the request is
removed from the list xprt->recv in xprt_complete_rqst(). This again
ensures that xprt_lookup_rqst() will fail to find that request until
we put it back on the list.

net/sunrpc/clnt.c

@@ -605,17 +605,17 @@ call_status(struct rpc_task *task)
 {
 	struct rpc_clnt	*clnt = task->tk_client;
 	struct rpc_xprt *xprt = clnt->cl_xprt;
-	struct rpc_rqst	*req;
-	int		status = task->tk_status;
+	struct rpc_rqst	*req = task->tk_rqstp;
+	int		status;
+
+	if (req->rq_received != 0)
+		task->tk_status = req->rq_received;
 
 	dprintk("RPC: %4d call_status (status %d)\n", 
 				task->tk_pid, task->tk_status);
 
-	req = task->tk_rqstp;
-	if (req->rq_received != 0)
-		status = req->rq_received;
+	status = task->tk_status;
 	if (status >= 0) {
-		req->rq_received = 0;
 		task->tk_action = call_decode;
 		return;
 	}

net/sunrpc/xprt.c

@@ -137,17 +137,14 @@ xprt_from_sock(struct sock *sk)
  * Also prevents TCP socket reconnections from colliding with writes.
  */
 static int
-xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
+__xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
 {
-	int retval;
-	spin_lock_bh(&xprt->sock_lock);
 	if (!xprt->snd_task) {
 		if (xprt->nocong || __xprt_get_cong(xprt, task))
 			xprt->snd_task = task;
 	}
 	if (xprt->snd_task != task) {
-		dprintk("RPC: %4d TCP write queue full (task %d)\n",
-			task->tk_pid, xprt->snd_task->tk_pid);
+		dprintk("RPC: %4d TCP write queue full\n", task->tk_pid);
 		task->tk_timeout = 0;
 		task->tk_status = -EAGAIN;
 		if (task->tk_rqstp->rq_nresend)
@@ -155,11 +152,21 @@ xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
 		else
 			rpc_sleep_on(&xprt->sending, task, NULL, NULL);
 	}
-	retval = xprt->snd_task == task;
+	return xprt->snd_task == task;
+}
+
+static inline int
+xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
+{
+	int retval;
+
+	spin_lock_bh(&xprt->sock_lock);
+	retval = __xprt_lock_write(xprt, task);
 	spin_unlock_bh(&xprt->sock_lock);
 	return retval;
 }
 
+
 static void
 __xprt_lock_write_next(struct rpc_xprt *xprt)
 {
@@ -564,8 +571,8 @@ xprt_complete_rqst(struct rpc_xprt *xprt, struct rpc_rqst *req, int copied)
 #endif
 
 	dprintk("RPC: %4d has input (%d bytes)\n", task->tk_pid, copied);
-	task->tk_status = copied;
 	req->rq_received = copied;
+	list_del_init(&req->rq_list);
 
 	/* ... and wake up the process. */
 	rpc_wake_up_task(task);
@@ -1057,8 +1064,16 @@ xprt_transmit(struct rpc_task *task)
 		*marker = htonl(0x80000000|(req->rq_slen-sizeof(*marker)));
 	}
 
-	if (!xprt_lock_write(xprt, task))
+	spin_lock_bh(&xprt->sock_lock);
+	if (!__xprt_lock_write(xprt, task)) {
+		spin_unlock_bh(&xprt->sock_lock);
 		return;
+	}
+	if (list_empty(&req->rq_list)) {
+		list_add_tail(&req->rq_list, &xprt->recv);
+		req->rq_received = 0;
+	}
+	spin_unlock_bh(&xprt->sock_lock);
 
 	do_xprt_transmit(task);
 }
@@ -1242,9 +1257,6 @@ xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
 	if (!xid)
 		xid++;
 	INIT_LIST_HEAD(&req->rq_list);
-	spin_lock_bh(&xprt->sock_lock);
-	list_add_tail(&req->rq_list, &xprt->recv);
-	spin_unlock_bh(&xprt->sock_lock);
 }
 
 /*