SUNRPC: Ensure that we grab the XPRT_LOCK before calling xprt_alloc_slot

This throttles the allocation of new slots when the socket is busy reconnecting and/or is out of buffer space. Signed-off-by: Trond Myklebust <Trond.Myklebust@netapp.com>

SUNRPC: Ensure that we grab the XPRT_LOCK before calling xprt_alloc_slot
This throttles the allocation of new slots when the socket is busy reconnecting and/or is out of buffer space. Signed-off-by: Trond Myklebust <Trond.Myklebust@netapp.com>
Trond Myklebust
1 parent f85ef69ce0
Showing 4 changed files with 49 additions and 30 deletions Side-by-side Diff
include/linux/sunrpc/xprt.h
net/sunrpc/xprt.c
net/sunrpc/xprtrdma/transport.c
net/sunrpc/xprtsock.c
@@ -111,7 +111,7 @@
  
 struct rpc_xprt_ops {
 	void		(*set_buffer_size)(struct rpc_xprt *xprt, size_t sndsize, size_t rcvsize);
-	int		(*reserve_xprt)(struct rpc_task *task);
+	int		(*reserve_xprt)(struct rpc_xprt *xprt, struct rpc_task *task);
 	void		(*release_xprt)(struct rpc_xprt *xprt, struct rpc_task *task);
 	void		(*rpcbind)(struct rpc_task *task);
 	void		(*set_port)(struct rpc_xprt *xprt, unsigned short port);
@@ -271,8 +271,8 @@
 struct rpc_xprt		*xprt_create_transport(struct xprt_create *args);
 void			xprt_connect(struct rpc_task *task);
 void			xprt_reserve(struct rpc_task *task);
-int			xprt_reserve_xprt(struct rpc_task *task);
-int			xprt_reserve_xprt_cong(struct rpc_task *task);
+int			xprt_reserve_xprt(struct rpc_xprt *xprt, struct rpc_task *task);
+int			xprt_reserve_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task);
 int			xprt_prepare_transmit(struct rpc_task *task);
 void			xprt_transmit(struct rpc_task *task);
 void			xprt_end_transmit(struct rpc_task *task);
@@ -191,10 +191,9 @@
  * transport connects from colliding with writes.  No congestion control
  * is provided.
  */
-int xprt_reserve_xprt(struct rpc_task *task)
+int xprt_reserve_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
 {
 	struct rpc_rqst *req = task->tk_rqstp;
-	struct rpc_xprt	*xprt = req->rq_xprt;
  
 	if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
 		if (task == xprt->snd_task)
@@ -202,8 +201,10 @@
 		goto out_sleep;
 	}
 	xprt->snd_task = task;
-	req->rq_bytes_sent = 0;
-	req->rq_ntrans++;
+	if (req != NULL) {
+		req->rq_bytes_sent = 0;
+		req->rq_ntrans++;
+	}
  
 	return 1;
  
@@ -212,7 +213,7 @@
 			task->tk_pid, xprt);
 	task->tk_timeout = 0;
 	task->tk_status = -EAGAIN;
-	if (req->rq_ntrans)
+	if (req != NULL && req->rq_ntrans)
 		rpc_sleep_on(&xprt->resend, task, NULL);
 	else
 		rpc_sleep_on(&xprt->sending, task, NULL);
  
@@ -239,9 +240,8 @@
  * integrated into the decision of whether a request is allowed to be
  * woken up and given access to the transport.
  */
-int xprt_reserve_xprt_cong(struct rpc_task *task)
+int xprt_reserve_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task)
 {
-	struct rpc_xprt	*xprt = task->tk_xprt;
 	struct rpc_rqst *req = task->tk_rqstp;
  
 	if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
  
@@ -249,12 +249,14 @@
 			return 1;
 		goto out_sleep;
 	}
+	if (req == NULL) {
+		xprt->snd_task = task;
+		return 1;
+	}
 	if (__xprt_get_cong(xprt, task)) {
 		xprt->snd_task = task;
-		if (req) {
-			req->rq_bytes_sent = 0;
-			req->rq_ntrans++;
-		}
+		req->rq_bytes_sent = 0;
+		req->rq_ntrans++;
 		return 1;
 	}
 	xprt_clear_locked(xprt);
@@ -262,7 +264,7 @@
 	dprintk("RPC: %5u failed to lock transport %p\n", task->tk_pid, xprt);
 	task->tk_timeout = 0;
 	task->tk_status = -EAGAIN;
-	if (req && req->rq_ntrans)
+	if (req != NULL && req->rq_ntrans)
 		rpc_sleep_on(&xprt->resend, task, NULL);
 	else
 		rpc_sleep_on(&xprt->sending, task, NULL);
@@ -275,7 +277,7 @@
 	int retval;
  
 	spin_lock_bh(&xprt->transport_lock);
-	retval = xprt->ops->reserve_xprt(task);
+	retval = xprt->ops->reserve_xprt(xprt, task);
 	spin_unlock_bh(&xprt->transport_lock);
 	return retval;
 }
@@ -291,7 +293,7 @@
 	task = rpc_wake_up_next(&xprt->resend);
 	if (!task) {
 		task = rpc_wake_up_next(&xprt->sending);
-		if (!task)
+		if (task == NULL)
 			goto out_unlock;
 	}
  
@@ -310,6 +312,7 @@
 static void __xprt_lock_write_next_cong(struct rpc_xprt *xprt)
 {
 	struct rpc_task *task;
+	struct rpc_rqst *req;
  
 	if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
 		return;
  
  
  
@@ -318,16 +321,19 @@
 	task = rpc_wake_up_next(&xprt->resend);
 	if (!task) {
 		task = rpc_wake_up_next(&xprt->sending);
-		if (!task)
+		if (task == NULL)
 			goto out_unlock;
 	}
+
+	req = task->tk_rqstp;
+	if (req == NULL) {
+		xprt->snd_task = task;
+		return;
+	}
 	if (__xprt_get_cong(xprt, task)) {
-		struct rpc_rqst *req = task->tk_rqstp;
 		xprt->snd_task = task;
-		if (req) {
-			req->rq_bytes_sent = 0;
-			req->rq_ntrans++;
-		}
+		req->rq_bytes_sent = 0;
+		req->rq_ntrans++;
 		return;
 	}
 out_unlock:
@@ -852,7 +858,7 @@
 		err = req->rq_reply_bytes_recvd;
 		goto out_unlock;
 	}
-	if (!xprt->ops->reserve_xprt(task))
+	if (!xprt->ops->reserve_xprt(xprt, task))
 		err = -EAGAIN;
 out_unlock:
 	spin_unlock_bh(&xprt->transport_lock);
@@ -933,8 +939,6 @@
 	struct rpc_xprt	*xprt = task->tk_xprt;
  
 	task->tk_status = 0;
-	if (task->tk_rqstp)
-		return;
 	if (!list_empty(&xprt->free)) {
 		struct rpc_rqst	*req = list_entry(xprt->free.next, struct rpc_rqst, rq_list);
 		list_del_init(&req->rq_list);
@@ -944,7 +948,6 @@
 	}
 	dprintk("RPC:       waiting for request slot\n");
 	task->tk_status = -EAGAIN;
-	task->tk_timeout = 0;
 	rpc_sleep_on(&xprt->backlog, task, NULL);
 }
  
  
@@ -1001,10 +1004,25 @@
 {
 	struct rpc_xprt	*xprt = task->tk_xprt;
  
+	task->tk_status = 0;
+	if (task->tk_rqstp != NULL)
+		return;
+
+	/* Note: grabbing the xprt_lock_write() here is not strictly needed,
+	 * but ensures that we throttle new slot allocation if the transport
+	 * is congested (e.g. if reconnecting or if we're out of socket
+	 * write buffer space).
+	 */
+	task->tk_timeout = 0;
+	task->tk_status = -EAGAIN;
+	if (!xprt_lock_write(xprt, task))
+		return;
+
 	task->tk_status = -EIO;
 	spin_lock(&xprt->reserve_lock);
 	xprt_alloc_slot(task);
 	spin_unlock(&xprt->reserve_lock);
+	xprt_release_write(xprt, task);
 }
  
 static inline __be32 xprt_alloc_xid(struct rpc_xprt *xprt)
@@ -452,9 +452,8 @@
 }
  
 static int
-xprt_rdma_reserve_xprt(struct rpc_task *task)
+xprt_rdma_reserve_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
 {
-	struct rpc_xprt *xprt = task->tk_xprt;
 	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
 	int credits = atomic_read(&r_xprt->rx_buf.rb_credits);
  
@@ -466,7 +465,7 @@
 		BUG_ON(r_xprt->rx_buf.rb_cwndscale <= 0);
 	}
 	xprt->cwnd = credits * r_xprt->rx_buf.rb_cwndscale;
-	return xprt_reserve_xprt_cong(task);
+	return xprt_reserve_xprt_cong(xprt, task);
 }
  
 /*
@@ -755,6 +755,8 @@
 	if (task == NULL)
 		goto out_release;
 	req = task->tk_rqstp;
+	if (req == NULL)
+		goto out_release;
 	if (req->rq_bytes_sent == 0)
 		goto out_release;
 	if (req->rq_bytes_sent == req->rq_snd_buf.len)
...	...	@@ -111,7 +111,7 @@
111	111
112	112	struct rpc_xprt_ops {
113	113	void (set_buffer_size)(struct rpc_xprt xprt, size_t sndsize, size_t rcvsize);
114		- int (reserve_xprt)(struct rpc_task task);
	114	+ int (reserve_xprt)(struct rpc_xprt xprt, struct rpc_task *task);
115	115	void (release_xprt)(struct rpc_xprt xprt, struct rpc_task *task);
116	116	void (rpcbind)(struct rpc_task task);
117	117	void (set_port)(struct rpc_xprt xprt, unsigned short port);
...	...	@@ -271,8 +271,8 @@
271	271	struct rpc_xprt xprt_create_transport(struct xprt_create args);
272	272	void xprt_connect(struct rpc_task *task);
273	273	void xprt_reserve(struct rpc_task *task);
274		-int xprt_reserve_xprt(struct rpc_task *task);
275		-int xprt_reserve_xprt_cong(struct rpc_task *task);
	274	+int xprt_reserve_xprt(struct rpc_xprt xprt, struct rpc_task task);
	275	+int xprt_reserve_xprt_cong(struct rpc_xprt xprt, struct rpc_task task);
276	276	int xprt_prepare_transmit(struct rpc_task *task);
277	277	void xprt_transmit(struct rpc_task *task);
278	278	void xprt_end_transmit(struct rpc_task *task);
...	...	@@ -191,10 +191,9 @@
191	191	* transport connects from colliding with writes. No congestion control
192	192	* is provided.
193	193	*/
194		-int xprt_reserve_xprt(struct rpc_task *task)
	194	+int xprt_reserve_xprt(struct rpc_xprt xprt, struct rpc_task task)
195	195	{
196	196	struct rpc_rqst *req = task->tk_rqstp;
197		- struct rpc_xprt *xprt = req->rq_xprt;
198	197
199	198	if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
200	199	if (task == xprt->snd_task)
...	...	@@ -202,8 +201,10 @@
202	201	goto out_sleep;
203	202	}
204	203	xprt->snd_task = task;
205		- req->rq_bytes_sent = 0;
206		- req->rq_ntrans++;
	204	+ if (req != NULL) {
	205	+ req->rq_bytes_sent = 0;
	206	+ req->rq_ntrans++;
	207	+ }
207	208
208	209	return 1;
209	210
...	...	@@ -212,7 +213,7 @@
212	213	task->tk_pid, xprt);
213	214	task->tk_timeout = 0;
214	215	task->tk_status = -EAGAIN;
215		- if (req->rq_ntrans)
	216	+ if (req != NULL && req->rq_ntrans)
216	217	rpc_sleep_on(&xprt->resend, task, NULL);
217	218	else
218	219	rpc_sleep_on(&xprt->sending, task, NULL);
219	220
...	...	@@ -239,9 +240,8 @@
239	240	* integrated into the decision of whether a request is allowed to be
240	241	* woken up and given access to the transport.
241	242	*/
242		-int xprt_reserve_xprt_cong(struct rpc_task *task)
	243	+int xprt_reserve_xprt_cong(struct rpc_xprt xprt, struct rpc_task task)
243	244	{
244		- struct rpc_xprt *xprt = task->tk_xprt;
245	245	struct rpc_rqst *req = task->tk_rqstp;
246	246
247	247	if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
248	248
...	...	@@ -249,12 +249,14 @@
249	249	return 1;
250	250	goto out_sleep;
251	251	}
	252	+ if (req == NULL) {
	253	+ xprt->snd_task = task;
	254	+ return 1;
	255	+ }
252	256	if (__xprt_get_cong(xprt, task)) {
253	257	xprt->snd_task = task;
254		- if (req) {
255		- req->rq_bytes_sent = 0;
256		- req->rq_ntrans++;
257		- }
	258	+ req->rq_bytes_sent = 0;
	259	+ req->rq_ntrans++;
258	260	return 1;
259	261	}
260	262	xprt_clear_locked(xprt);
...	...	@@ -262,7 +264,7 @@
262	264	dprintk("RPC: %5u failed to lock transport %p\n", task->tk_pid, xprt);
263	265	task->tk_timeout = 0;
264	266	task->tk_status = -EAGAIN;
265		- if (req && req->rq_ntrans)
	267	+ if (req != NULL && req->rq_ntrans)
266	268	rpc_sleep_on(&xprt->resend, task, NULL);
267	269	else
268	270	rpc_sleep_on(&xprt->sending, task, NULL);
...	...	@@ -275,7 +277,7 @@
275	277	int retval;
276	278
277	279	spin_lock_bh(&xprt->transport_lock);
278		- retval = xprt->ops->reserve_xprt(task);
	280	+ retval = xprt->ops->reserve_xprt(xprt, task);
279	281	spin_unlock_bh(&xprt->transport_lock);
280	282	return retval;
281	283	}
...	...	@@ -291,7 +293,7 @@
291	293	task = rpc_wake_up_next(&xprt->resend);
292	294	if (!task) {
293	295	task = rpc_wake_up_next(&xprt->sending);
294		- if (!task)
	296	+ if (task == NULL)
295	297	goto out_unlock;
296	298	}
297	299
...	...	@@ -310,6 +312,7 @@
310	312	static void __xprt_lock_write_next_cong(struct rpc_xprt *xprt)
311	313	{
312	314	struct rpc_task *task;
	315	+ struct rpc_rqst *req;
313	316
314	317	if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
315	318	return;
316	319
317	320
318	321
...	...	@@ -318,16 +321,19 @@
318	321	task = rpc_wake_up_next(&xprt->resend);
319	322	if (!task) {
320	323	task = rpc_wake_up_next(&xprt->sending);
321		- if (!task)
	324	+ if (task == NULL)
322	325	goto out_unlock;
323	326	}
	327	+
	328	+ req = task->tk_rqstp;
	329	+ if (req == NULL) {
	330	+ xprt->snd_task = task;
	331	+ return;
	332	+ }
324	333	if (__xprt_get_cong(xprt, task)) {
325		- struct rpc_rqst *req = task->tk_rqstp;
326	334	xprt->snd_task = task;
327		- if (req) {
328		- req->rq_bytes_sent = 0;
329		- req->rq_ntrans++;
330		- }
	335	+ req->rq_bytes_sent = 0;
	336	+ req->rq_ntrans++;
331	337	return;
332	338	}
333	339	out_unlock:
...	...	@@ -852,7 +858,7 @@
852	858	err = req->rq_reply_bytes_recvd;
853	859	goto out_unlock;
854	860	}
855		- if (!xprt->ops->reserve_xprt(task))
	861	+ if (!xprt->ops->reserve_xprt(xprt, task))
856	862	err = -EAGAIN;
857	863	out_unlock:
858	864	spin_unlock_bh(&xprt->transport_lock);
...	...	@@ -933,8 +939,6 @@
933	939	struct rpc_xprt *xprt = task->tk_xprt;
934	940
935	941	task->tk_status = 0;
936		- if (task->tk_rqstp)
937		- return;
938	942	if (!list_empty(&xprt->free)) {
939	943	struct rpc_rqst *req = list_entry(xprt->free.next, struct rpc_rqst, rq_list);
940	944	list_del_init(&req->rq_list);
...	...	@@ -944,7 +948,6 @@
944	948	}
945	949	dprintk("RPC: waiting for request slot\n");
946	950	task->tk_status = -EAGAIN;
947		- task->tk_timeout = 0;
948	951	rpc_sleep_on(&xprt->backlog, task, NULL);
949	952	}
950	953
951	954
...	...	@@ -1001,10 +1004,25 @@
1001	1004	{
1002	1005	struct rpc_xprt *xprt = task->tk_xprt;
1003	1006
	1007	+ task->tk_status = 0;
	1008	+ if (task->tk_rqstp != NULL)
	1009	+ return;
	1010	+
	1011	+ /* Note: grabbing the xprt_lock_write() here is not strictly needed,
	1012	+ * but ensures that we throttle new slot allocation if the transport
	1013	+ * is congested (e.g. if reconnecting or if we're out of socket
	1014	+ * write buffer space).
	1015	+ */
	1016	+ task->tk_timeout = 0;
	1017	+ task->tk_status = -EAGAIN;
	1018	+ if (!xprt_lock_write(xprt, task))
	1019	+ return;
	1020	+
1004	1021	task->tk_status = -EIO;
1005	1022	spin_lock(&xprt->reserve_lock);
1006	1023	xprt_alloc_slot(task);
1007	1024	spin_unlock(&xprt->reserve_lock);
	1025	+ xprt_release_write(xprt, task);
1008	1026	}
1009	1027
1010	1028	static inline __be32 xprt_alloc_xid(struct rpc_xprt *xprt)
...	...	@@ -452,9 +452,8 @@
452	452	}
453	453
454	454	static int
455		-xprt_rdma_reserve_xprt(struct rpc_task *task)
	455	+xprt_rdma_reserve_xprt(struct rpc_xprt xprt, struct rpc_task task)
456	456	{
457		- struct rpc_xprt *xprt = task->tk_xprt;
458	457	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
459	458	int credits = atomic_read(&r_xprt->rx_buf.rb_credits);
460	459
...	...	@@ -466,7 +465,7 @@
466	465	BUG_ON(r_xprt->rx_buf.rb_cwndscale <= 0);
467	466	}
468	467	xprt->cwnd = credits * r_xprt->rx_buf.rb_cwndscale;
469		- return xprt_reserve_xprt_cong(task);
	468	+ return xprt_reserve_xprt_cong(xprt, task);
470	469	}
471	470
472	471	/*
...	...	@@ -755,6 +755,8 @@
755	755	if (task == NULL)
756	756	goto out_release;
757	757	req = task->tk_rqstp;
	758	+ if (req == NULL)
	759	+ goto out_release;
758	760	if (req->rq_bytes_sent == 0)
759	761	goto out_release;
760	762	if (req->rq_bytes_sent == req->rq_snd_buf.len)