~hc/RK356X_SDK_RELEASE.git

..	..	@@ -1,3 +1,4 @@
	1	+// SPDX-License-Identifier: GPL-2.0-only
1	2	/*
2	3	* linux/net/sunrpc/xprt.c
3	4	*
..	..	@@ -49,6 +50,7 @@
49	50	#include <linux/sunrpc/metrics.h>
50	51	#include <linux/sunrpc/bc_xprt.h>
51	52	#include <linux/rcupdate.h>
	53	+#include <linux/sched/mm.h>
52	54
53	55	#include <trace/events/sunrpc.h>
54	56
..	..	@@ -67,13 +69,20 @@
67	69	*/
68	70	static void xprt_init(struct rpc_xprt xprt, struct net net);
69	71	static __be32 xprt_alloc_xid(struct rpc_xprt *xprt);
70		-static void xprt_connect_status(struct rpc_task *task);
71		-static int __xprt_get_cong(struct rpc_xprt , struct rpc_task );
72		-static void __xprt_put_cong(struct rpc_xprt , struct rpc_rqst );
73	72	static void xprt_destroy(struct rpc_xprt *xprt);
	73	+static void xprt_request_init(struct rpc_task *task);
74	74
75	75	static DEFINE_SPINLOCK(xprt_list_lock);
76	76	static LIST_HEAD(xprt_list);
	77	+
	78	+static unsigned long xprt_request_timeout(const struct rpc_rqst *req)
	79	+{
	80	+ unsigned long timeout = jiffies + req->rq_timeout;
	81	+
	82	+ if (time_before(timeout, req->rq_majortimeo))
	83	+ return timeout;
	84	+ return req->rq_majortimeo;
	85	+}
77	86
78	87	/**
79	88	* xprt_register_transport - register a transport implementation
..	..	@@ -204,6 +213,17 @@
204	213	}
205	214	EXPORT_SYMBOL_GPL(xprt_load_transport);
206	215
	216	+static void xprt_clear_locked(struct rpc_xprt *xprt)
	217	+{
	218	+ xprt->snd_task = NULL;
	219	+ if (!test_bit(XPRT_CLOSE_WAIT, &xprt->state)) {
	220	+ smp_mb__before_atomic();
	221	+ clear_bit(XPRT_LOCKED, &xprt->state);
	222	+ smp_mb__after_atomic();
	223	+ } else
	224	+ queue_work(xprtiod_workqueue, &xprt->task_cleanup);
	225	+}
	226	+
207	227	/**
208	228	* xprt_reserve_xprt - serialize write access to transports
209	229	* @task: task that is requesting access to the transport
..	..	@@ -216,44 +236,56 @@
216	236	int xprt_reserve_xprt(struct rpc_xprt xprt, struct rpc_task task)
217	237	{
218	238	struct rpc_rqst *req = task->tk_rqstp;
219		- int priority;
220	239
221	240	if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
222	241	if (task == xprt->snd_task)
223		- return 1;
	242	+ goto out_locked;
224	243	goto out_sleep;
225	244	}
	245	+ if (test_bit(XPRT_WRITE_SPACE, &xprt->state))
	246	+ goto out_unlock;
226	247	xprt->snd_task = task;
227		- if (req != NULL)
228		- req->rq_ntrans++;
229	248
	249	+out_locked:
	250	+ trace_xprt_reserve_xprt(xprt, task);
230	251	return 1;
231	252
	253	+out_unlock:
	254	+ xprt_clear_locked(xprt);
232	255	out_sleep:
233		- dprintk("RPC: %5u failed to lock transport %p\n",
234		- task->tk_pid, xprt);
235		- task->tk_timeout = 0;
236	256	task->tk_status = -EAGAIN;
237		- if (req == NULL)
238		- priority = RPC_PRIORITY_LOW;
239		- else if (!req->rq_ntrans)
240		- priority = RPC_PRIORITY_NORMAL;
	257	+ if (RPC_IS_SOFT(task))
	258	+ rpc_sleep_on_timeout(&xprt->sending, task, NULL,
	259	+ xprt_request_timeout(req));
241	260	else
242		- priority = RPC_PRIORITY_HIGH;
243		- rpc_sleep_on_priority(&xprt->sending, task, NULL, priority);
	261	+ rpc_sleep_on(&xprt->sending, task, NULL);
244	262	return 0;
245	263	}
246	264	EXPORT_SYMBOL_GPL(xprt_reserve_xprt);
247	265
248		-static void xprt_clear_locked(struct rpc_xprt *xprt)
	266	+static bool
	267	+xprt_need_congestion_window_wait(struct rpc_xprt *xprt)
249	268	{
250		- xprt->snd_task = NULL;
251		- if (!test_bit(XPRT_CLOSE_WAIT, &xprt->state)) {
252		- smp_mb__before_atomic();
253		- clear_bit(XPRT_LOCKED, &xprt->state);
254		- smp_mb__after_atomic();
255		- } else
256		- queue_work(xprtiod_workqueue, &xprt->task_cleanup);
	269	+ return test_bit(XPRT_CWND_WAIT, &xprt->state);
	270	+}
	271	+
	272	+static void
	273	+xprt_set_congestion_window_wait(struct rpc_xprt *xprt)
	274	+{
	275	+ if (!list_empty(&xprt->xmit_queue)) {
	276	+ /* Peek at head of queue to see if it can make progress */
	277	+ if (list_first_entry(&xprt->xmit_queue, struct rpc_rqst,
	278	+ rq_xmit)->rq_cong)
	279	+ return;
	280	+ }
	281	+ set_bit(XPRT_CWND_WAIT, &xprt->state);
	282	+}
	283	+
	284	+static void
	285	+xprt_test_and_clear_congestion_window_wait(struct rpc_xprt *xprt)
	286	+{
	287	+ if (!RPCXPRT_CONGESTED(xprt))
	288	+ clear_bit(XPRT_CWND_WAIT, &xprt->state);
257	289	}
258	290
259	291	/*
..	..	@@ -263,41 +295,40 @@
263	295	* Same as xprt_reserve_xprt, but Van Jacobson congestion control is
264	296	* integrated into the decision of whether a request is allowed to be
265	297	* woken up and given access to the transport.
	298	+ * Note that the lock is only granted if we know there are free slots.
266	299	*/
267	300	int xprt_reserve_xprt_cong(struct rpc_xprt xprt, struct rpc_task task)
268	301	{
269	302	struct rpc_rqst *req = task->tk_rqstp;
270		- int priority;
271	303
272	304	if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
273	305	if (task == xprt->snd_task)
274		- return 1;
	306	+ goto out_locked;
275	307	goto out_sleep;
276	308	}
277	309	if (req == NULL) {
278	310	xprt->snd_task = task;
279		- return 1;
	311	+ goto out_locked;
280	312	}
281		- if (__xprt_get_cong(xprt, task)) {
	313	+ if (test_bit(XPRT_WRITE_SPACE, &xprt->state))
	314	+ goto out_unlock;
	315	+ if (!xprt_need_congestion_window_wait(xprt)) {
282	316	xprt->snd_task = task;
283		- req->rq_ntrans++;
284		- return 1;
	317	+ goto out_locked;
285	318	}
	319	+out_unlock:
286	320	xprt_clear_locked(xprt);
287	321	out_sleep:
288		- if (req)
289		- __xprt_put_cong(xprt, req);
290		- dprintk("RPC: %5u failed to lock transport %p\n", task->tk_pid, xprt);
291		- task->tk_timeout = 0;
292	322	task->tk_status = -EAGAIN;
293		- if (req == NULL)
294		- priority = RPC_PRIORITY_LOW;
295		- else if (!req->rq_ntrans)
296		- priority = RPC_PRIORITY_NORMAL;
	323	+ if (RPC_IS_SOFT(task))
	324	+ rpc_sleep_on_timeout(&xprt->sending, task, NULL,
	325	+ xprt_request_timeout(req));
297	326	else
298		- priority = RPC_PRIORITY_HIGH;
299		- rpc_sleep_on_priority(&xprt->sending, task, NULL, priority);
	327	+ rpc_sleep_on(&xprt->sending, task, NULL);
300	328	return 0;
	329	+out_locked:
	330	+ trace_xprt_reserve_cong(xprt, task);
	331	+ return 1;
301	332	}
302	333	EXPORT_SYMBOL_GPL(xprt_reserve_xprt_cong);
303	334
..	..	@@ -305,21 +336,19 @@
305	336	{
306	337	int retval;
307	338
308		- spin_lock_bh(&xprt->transport_lock);
	339	+ if (test_bit(XPRT_LOCKED, &xprt->state) && xprt->snd_task == task)
	340	+ return 1;
	341	+ spin_lock(&xprt->transport_lock);
309	342	retval = xprt->ops->reserve_xprt(xprt, task);
310		- spin_unlock_bh(&xprt->transport_lock);
	343	+ spin_unlock(&xprt->transport_lock);
311	344	return retval;
312	345	}
313	346
314	347	static bool __xprt_lock_write_func(struct rpc_task task, void data)
315	348	{
316	349	struct rpc_xprt *xprt = data;
317		- struct rpc_rqst *req;
318	350
319		- req = task->tk_rqstp;
320	351	xprt->snd_task = task;
321		- if (req)
322		- req->rq_ntrans++;
323	352	return true;
324	353	}
325	354
..	..	@@ -327,51 +356,28 @@
327	356	{
328	357	if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
329	358	return;
330		-
	359	+ if (test_bit(XPRT_WRITE_SPACE, &xprt->state))
	360	+ goto out_unlock;
331	361	if (rpc_wake_up_first_on_wq(xprtiod_workqueue, &xprt->sending,
332	362	__xprt_lock_write_func, xprt))
333	363	return;
	364	+out_unlock:
334	365	xprt_clear_locked(xprt);
335		-}
336		-
337		-static bool __xprt_lock_write_cong_func(struct rpc_task task, void data)
338		-{
339		- struct rpc_xprt *xprt = data;
340		- struct rpc_rqst *req;
341		-
342		- req = task->tk_rqstp;
343		- if (req == NULL) {
344		- xprt->snd_task = task;
345		- return true;
346		- }
347		- if (__xprt_get_cong(xprt, task)) {
348		- xprt->snd_task = task;
349		- req->rq_ntrans++;
350		- return true;
351		- }
352		- return false;
353	366	}
354	367
355	368	static void __xprt_lock_write_next_cong(struct rpc_xprt *xprt)
356	369	{
357	370	if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
358	371	return;
359		- if (RPCXPRT_CONGESTED(xprt))
	372	+ if (test_bit(XPRT_WRITE_SPACE, &xprt->state))
	373	+ goto out_unlock;
	374	+ if (xprt_need_congestion_window_wait(xprt))
360	375	goto out_unlock;
361	376	if (rpc_wake_up_first_on_wq(xprtiod_workqueue, &xprt->sending,
362		- __xprt_lock_write_cong_func, xprt))
	377	+ __xprt_lock_write_func, xprt))
363	378	return;
364	379	out_unlock:
365	380	xprt_clear_locked(xprt);
366		-}
367		-
368		-static void xprt_task_clear_bytes_sent(struct rpc_task *task)
369		-{
370		- if (task != NULL) {
371		- struct rpc_rqst *req = task->tk_rqstp;
372		- if (req != NULL)
373		- req->rq_bytes_sent = 0;
374		- }
375	381	}
376	382
377	383	/**
..	..	@@ -384,10 +390,10 @@
384	390	void xprt_release_xprt(struct rpc_xprt xprt, struct rpc_task task)
385	391	{
386	392	if (xprt->snd_task == task) {
387		- xprt_task_clear_bytes_sent(task);
388	393	xprt_clear_locked(xprt);
389	394	__xprt_lock_write_next(xprt);
390	395	}
	396	+ trace_xprt_release_xprt(xprt, task);
391	397	}
392	398	EXPORT_SYMBOL_GPL(xprt_release_xprt);
393	399
..	..	@@ -402,18 +408,20 @@
402	408	void xprt_release_xprt_cong(struct rpc_xprt xprt, struct rpc_task task)
403	409	{
404	410	if (xprt->snd_task == task) {
405		- xprt_task_clear_bytes_sent(task);
406	411	xprt_clear_locked(xprt);
407	412	__xprt_lock_write_next_cong(xprt);
408	413	}
	414	+ trace_xprt_release_cong(xprt, task);
409	415	}
410	416	EXPORT_SYMBOL_GPL(xprt_release_xprt_cong);
411	417
412	418	static inline void xprt_release_write(struct rpc_xprt xprt, struct rpc_task task)
413	419	{
414		- spin_lock_bh(&xprt->transport_lock);
	420	+ if (xprt->snd_task != task)
	421	+ return;
	422	+ spin_lock(&xprt->transport_lock);
415	423	xprt->ops->release_xprt(xprt, task);
416		- spin_unlock_bh(&xprt->transport_lock);
	424	+ spin_unlock(&xprt->transport_lock);
417	425	}
418	426
419	427	/*
..	..	@@ -421,16 +429,15 @@
421	429	* overflowed. Put the task to sleep if this is the case.
422	430	*/
423	431	static int
424		-__xprt_get_cong(struct rpc_xprt xprt, struct rpc_task task)
	432	+__xprt_get_cong(struct rpc_xprt xprt, struct rpc_rqst req)
425	433	{
426		- struct rpc_rqst *req = task->tk_rqstp;
427		-
428	434	if (req->rq_cong)
429	435	return 1;
430		- dprintk("RPC: %5u xprt_cwnd_limited cong = %lu cwnd = %lu\n",
431		- task->tk_pid, xprt->cong, xprt->cwnd);
432		- if (RPCXPRT_CONGESTED(xprt))
	436	+ trace_xprt_get_cong(xprt, req->rq_task);
	437	+ if (RPCXPRT_CONGESTED(xprt)) {
	438	+ xprt_set_congestion_window_wait(xprt);
433	439	return 0;
	440	+ }
434	441	req->rq_cong = 1;
435	442	xprt->cong += RPC_CWNDSCALE;
436	443	return 1;
..	..	@@ -447,8 +454,31 @@
447	454	return;
448	455	req->rq_cong = 0;
449	456	xprt->cong -= RPC_CWNDSCALE;
	457	+ xprt_test_and_clear_congestion_window_wait(xprt);
	458	+ trace_xprt_put_cong(xprt, req->rq_task);
450	459	__xprt_lock_write_next_cong(xprt);
451	460	}
	461	+
	462	+/**
	463	+ * xprt_request_get_cong - Request congestion control credits
	464	+ * @xprt: pointer to transport
	465	+ * @req: pointer to RPC request
	466	+ *
	467	+ * Useful for transports that require congestion control.
	468	+ */
	469	+bool
	470	+xprt_request_get_cong(struct rpc_xprt xprt, struct rpc_rqst req)
	471	+{
	472	+ bool ret = false;
	473	+
	474	+ if (req->rq_cong)
	475	+ return true;
	476	+ spin_lock(&xprt->transport_lock);
	477	+ ret = __xprt_get_cong(xprt, req) != 0;
	478	+ spin_unlock(&xprt->transport_lock);
	479	+ return ret;
	480	+}
	481	+EXPORT_SYMBOL_GPL(xprt_request_get_cong);
452	482
453	483	/**
454	484	* xprt_release_rqst_cong - housekeeping when request is complete
..	..	@@ -463,6 +493,26 @@
463	493	__xprt_put_cong(req->rq_xprt, req);
464	494	}
465	495	EXPORT_SYMBOL_GPL(xprt_release_rqst_cong);
	496	+
	497	+static void xprt_clear_congestion_window_wait_locked(struct rpc_xprt *xprt)
	498	+{
	499	+ if (test_and_clear_bit(XPRT_CWND_WAIT, &xprt->state))
	500	+ __xprt_lock_write_next_cong(xprt);
	501	+}
	502	+
	503	+/*
	504	+ * Clear the congestion window wait flag and wake up the next
	505	+ * entry on xprt->sending
	506	+ */
	507	+static void
	508	+xprt_clear_congestion_window_wait(struct rpc_xprt *xprt)
	509	+{
	510	+ if (test_and_clear_bit(XPRT_CWND_WAIT, &xprt->state)) {
	511	+ spin_lock(&xprt->transport_lock);
	512	+ __xprt_lock_write_next_cong(xprt);
	513	+ spin_unlock(&xprt->transport_lock);
	514	+ }
	515	+}
466	516
467	517	/**
468	518	* xprt_adjust_cwnd - adjust transport congestion window
..	..	@@ -521,22 +571,29 @@
521	571
522	572	/**
523	573	* xprt_wait_for_buffer_space - wait for transport output buffer to clear
524		- * @task: task to be put to sleep
525		- * @action: function pointer to be executed after wait
	574	+ * @xprt: transport
526	575	*
527	576	* Note that we only set the timer for the case of RPC_IS_SOFT(), since
528	577	* we don't in general want to force a socket disconnection due to
529	578	* an incomplete RPC call transmission.
530	579	*/
531		-void xprt_wait_for_buffer_space(struct rpc_task *task, rpc_action action)
	580	+void xprt_wait_for_buffer_space(struct rpc_xprt *xprt)
532	581	{
533		- struct rpc_rqst *req = task->tk_rqstp;
534		- struct rpc_xprt *xprt = req->rq_xprt;
535		-
536		- task->tk_timeout = RPC_IS_SOFT(task) ? req->rq_timeout : 0;
537		- rpc_sleep_on(&xprt->pending, task, action);
	582	+ set_bit(XPRT_WRITE_SPACE, &xprt->state);
538	583	}
539	584	EXPORT_SYMBOL_GPL(xprt_wait_for_buffer_space);
	585	+
	586	+static bool
	587	+xprt_clear_write_space_locked(struct rpc_xprt *xprt)
	588	+{
	589	+ if (test_and_clear_bit(XPRT_WRITE_SPACE, &xprt->state)) {
	590	+ __xprt_lock_write_next(xprt);
	591	+ dprintk("RPC: write space: waking waiting task on "
	592	+ "xprt %p\n", xprt);
	593	+ return true;
	594	+ }
	595	+ return false;
	596	+}
540	597
541	598	/**
542	599	* xprt_write_space - wake the task waiting for transport output buffer space
..	..	@@ -544,66 +601,63 @@
544	601	*
545	602	* Can be called in a soft IRQ context, so xprt_write_space never sleeps.
546	603	*/
547		-void xprt_write_space(struct rpc_xprt *xprt)
	604	+bool xprt_write_space(struct rpc_xprt *xprt)
548	605	{
549		- spin_lock_bh(&xprt->transport_lock);
550		- if (xprt->snd_task) {
551		- dprintk("RPC: write space: waking waiting task on "
552		- "xprt %p\n", xprt);
553		- rpc_wake_up_queued_task_on_wq(xprtiod_workqueue,
554		- &xprt->pending, xprt->snd_task);
555		- }
556		- spin_unlock_bh(&xprt->transport_lock);
	606	+ bool ret;
	607	+
	608	+ if (!test_bit(XPRT_WRITE_SPACE, &xprt->state))
	609	+ return false;
	610	+ spin_lock(&xprt->transport_lock);
	611	+ ret = xprt_clear_write_space_locked(xprt);
	612	+ spin_unlock(&xprt->transport_lock);
	613	+ return ret;
557	614	}
558	615	EXPORT_SYMBOL_GPL(xprt_write_space);
559	616
560		-/**
561		- * xprt_set_retrans_timeout_def - set a request's retransmit timeout
562		- * @task: task whose timeout is to be set
563		- *
564		- * Set a request's retransmit timeout based on the transport's
565		- * default timeout parameters. Used by transports that don't adjust
566		- * the retransmit timeout based on round-trip time estimation.
567		- */
568		-void xprt_set_retrans_timeout_def(struct rpc_task *task)
	617	+static unsigned long xprt_abs_ktime_to_jiffies(ktime_t abstime)
569	618	{
570		- task->tk_timeout = task->tk_rqstp->rq_timeout;
	619	+ s64 delta = ktime_to_ns(ktime_get() - abstime);
	620	+ return likely(delta >= 0) ?
	621	+ jiffies - nsecs_to_jiffies(delta) :
	622	+ jiffies + nsecs_to_jiffies(-delta);
571	623	}
572		-EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_def);
573	624
574		-/**
575		- * xprt_set_retrans_timeout_rtt - set a request's retransmit timeout
576		- * @task: task whose timeout is to be set
577		- *
578		- * Set a request's retransmit timeout using the RTT estimator.
579		- */
580		-void xprt_set_retrans_timeout_rtt(struct rpc_task *task)
	625	+static unsigned long xprt_calc_majortimeo(struct rpc_rqst *req)
581	626	{
582		- int timer = task->tk_msg.rpc_proc->p_timer;
583		- struct rpc_clnt *clnt = task->tk_client;
584		- struct rpc_rtt *rtt = clnt->cl_rtt;
585		- struct rpc_rqst *req = task->tk_rqstp;
586		- unsigned long max_timeout = clnt->cl_timeout->to_maxval;
	627	+ const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout;
	628	+ unsigned long majortimeo = req->rq_timeout;
587	629
588		- task->tk_timeout = rpc_calc_rto(rtt, timer);
589		- task->tk_timeout <<= rpc_ntimeo(rtt, timer) + req->rq_retries;
590		- if (task->tk_timeout > max_timeout \|\| task->tk_timeout == 0)
591		- task->tk_timeout = max_timeout;
	630	+ if (to->to_exponential)
	631	+ majortimeo <<= to->to_retries;
	632	+ else
	633	+ majortimeo += to->to_increment * to->to_retries;
	634	+ if (majortimeo > to->to_maxval \|\| majortimeo == 0)
	635	+ majortimeo = to->to_maxval;
	636	+ return majortimeo;
592	637	}
593		-EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_rtt);
594	638
595	639	static void xprt_reset_majortimeo(struct rpc_rqst *req)
596	640	{
597		- const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout;
	641	+ req->rq_majortimeo += xprt_calc_majortimeo(req);
	642	+}
598	643
599		- req->rq_majortimeo = req->rq_timeout;
600		- if (to->to_exponential)
601		- req->rq_majortimeo <<= to->to_retries;
	644	+static void xprt_reset_minortimeo(struct rpc_rqst *req)
	645	+{
	646	+ req->rq_minortimeo += req->rq_timeout;
	647	+}
	648	+
	649	+static void xprt_init_majortimeo(struct rpc_task task, struct rpc_rqst req)
	650	+{
	651	+ unsigned long time_init;
	652	+ struct rpc_xprt *xprt = req->rq_xprt;
	653	+
	654	+ if (likely(xprt && xprt_connected(xprt)))
	655	+ time_init = jiffies;
602	656	else
603		- req->rq_majortimeo += to->to_increment * to->to_retries;
604		- if (req->rq_majortimeo > to->to_maxval \|\| req->rq_majortimeo == 0)
605		- req->rq_majortimeo = to->to_maxval;
606		- req->rq_majortimeo += jiffies;
	657	+ time_init = xprt_abs_ktime_to_jiffies(task->tk_start);
	658	+ req->rq_timeout = task->tk_client->cl_timeout->to_initval;
	659	+ req->rq_majortimeo = time_init + xprt_calc_majortimeo(req);
	660	+ req->rq_minortimeo = time_init + req->rq_timeout;
607	661	}
608	662
609	663	/**
..	..	@@ -618,6 +672,8 @@
618	672	int status = 0;
619	673
620	674	if (time_before(jiffies, req->rq_majortimeo)) {
	675	+ if (time_before(jiffies, req->rq_minortimeo))
	676	+ return status;
621	677	if (to->to_exponential)
622	678	req->rq_timeout <<= 1;
623	679	else
..	..	@@ -630,11 +686,12 @@
630	686	req->rq_retries = 0;
631	687	xprt_reset_majortimeo(req);
632	688	/* Reset the RTT counters == "slow start" */
633		- spin_lock_bh(&xprt->transport_lock);
	689	+ spin_lock(&xprt->transport_lock);
634	690	rpc_init_rtt(req->rq_task->tk_client->cl_rtt, to->to_initval);
635		- spin_unlock_bh(&xprt->transport_lock);
	691	+ spin_unlock(&xprt->transport_lock);
636	692	status = -ETIMEDOUT;
637	693	}
	694	+ xprt_reset_minortimeo(req);
638	695
639	696	if (req->rq_timeout == 0) {
640	697	printk(KERN_WARNING "xprt_adjust_timeout: rq_timeout = 0!\n");
..	..	@@ -647,11 +704,14 @@
647	704	{
648	705	struct rpc_xprt *xprt =
649	706	container_of(work, struct rpc_xprt, task_cleanup);
	707	+ unsigned int pflags = memalloc_nofs_save();
650	708
	709	+ trace_xprt_disconnect_auto(xprt);
651	710	clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
652	711	xprt->ops->close(xprt);
653	712	xprt_release_write(xprt, NULL);
654	713	wake_up_bit(&xprt->state, XPRT_LOCKED);
	714	+ memalloc_nofs_restore(pflags);
655	715	}
656	716
657	717	/**
..	..	@@ -661,13 +721,30 @@
661	721	*/
662	722	void xprt_disconnect_done(struct rpc_xprt *xprt)
663	723	{
664		- dprintk("RPC: disconnected transport %p\n", xprt);
665		- spin_lock_bh(&xprt->transport_lock);
	724	+ trace_xprt_disconnect_done(xprt);
	725	+ spin_lock(&xprt->transport_lock);
666	726	xprt_clear_connected(xprt);
667		- xprt_wake_pending_tasks(xprt, -EAGAIN);
668		- spin_unlock_bh(&xprt->transport_lock);
	727	+ xprt_clear_write_space_locked(xprt);
	728	+ xprt_clear_congestion_window_wait_locked(xprt);
	729	+ xprt_wake_pending_tasks(xprt, -ENOTCONN);
	730	+ spin_unlock(&xprt->transport_lock);
669	731	}
670	732	EXPORT_SYMBOL_GPL(xprt_disconnect_done);
	733	+
	734	+/**
	735	+ * xprt_schedule_autoclose_locked - Try to schedule an autoclose RPC call
	736	+ * @xprt: transport to disconnect
	737	+ */
	738	+static void xprt_schedule_autoclose_locked(struct rpc_xprt *xprt)
	739	+{
	740	+ if (test_and_set_bit(XPRT_CLOSE_WAIT, &xprt->state))
	741	+ return;
	742	+ if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
	743	+ queue_work(xprtiod_workqueue, &xprt->task_cleanup);
	744	+ else if (xprt->snd_task && !test_bit(XPRT_SND_IS_COOKIE, &xprt->state))
	745	+ rpc_wake_up_queued_task_set_status(&xprt->pending,
	746	+ xprt->snd_task, -ENOTCONN);
	747	+}
671	748
672	749	/**
673	750	* xprt_force_disconnect - force a transport to disconnect
..	..	@@ -676,16 +753,30 @@
676	753	*/
677	754	void xprt_force_disconnect(struct rpc_xprt *xprt)
678	755	{
	756	+ trace_xprt_disconnect_force(xprt);
	757	+
679	758	/* Don't race with the test_bit() in xprt_clear_locked() */
680		- spin_lock_bh(&xprt->transport_lock);
681		- set_bit(XPRT_CLOSE_WAIT, &xprt->state);
682		- /* Try to schedule an autoclose RPC call */
683		- if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
684		- queue_work(xprtiod_workqueue, &xprt->task_cleanup);
685		- xprt_wake_pending_tasks(xprt, -EAGAIN);
686		- spin_unlock_bh(&xprt->transport_lock);
	759	+ spin_lock(&xprt->transport_lock);
	760	+ xprt_schedule_autoclose_locked(xprt);
	761	+ spin_unlock(&xprt->transport_lock);
687	762	}
688	763	EXPORT_SYMBOL_GPL(xprt_force_disconnect);
	764	+
	765	+static unsigned int
	766	+xprt_connect_cookie(struct rpc_xprt *xprt)
	767	+{
	768	+ return READ_ONCE(xprt->connect_cookie);
	769	+}
	770	+
	771	+static bool
	772	+xprt_request_retransmit_after_disconnect(struct rpc_task *task)
	773	+{
	774	+ struct rpc_rqst *req = task->tk_rqstp;
	775	+ struct rpc_xprt *xprt = req->rq_xprt;
	776	+
	777	+ return req->rq_connect_cookie != xprt_connect_cookie(xprt) \|\|
	778	+ !xprt_connected(xprt);
	779	+}
689	780
690	781	/**
691	782	* xprt_conditional_disconnect - force a transport to disconnect
..	..	@@ -701,18 +792,14 @@
701	792	void xprt_conditional_disconnect(struct rpc_xprt *xprt, unsigned int cookie)
702	793	{
703	794	/* Don't race with the test_bit() in xprt_clear_locked() */
704		- spin_lock_bh(&xprt->transport_lock);
	795	+ spin_lock(&xprt->transport_lock);
705	796	if (cookie != xprt->connect_cookie)
706	797	goto out;
707	798	if (test_bit(XPRT_CLOSING, &xprt->state))
708	799	goto out;
709		- set_bit(XPRT_CLOSE_WAIT, &xprt->state);
710		- /* Try to schedule an autoclose RPC call */
711		- if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
712		- queue_work(xprtiod_workqueue, &xprt->task_cleanup);
713		- xprt_wake_pending_tasks(xprt, -EAGAIN);
	800	+ xprt_schedule_autoclose_locked(xprt);
714	801	out:
715		- spin_unlock_bh(&xprt->transport_lock);
	802	+ spin_unlock(&xprt->transport_lock);
716	803	}
717	804
718	805	static bool
..	..	@@ -725,7 +812,8 @@
725	812	xprt_schedule_autodisconnect(struct rpc_xprt *xprt)
726	813	__must_hold(&xprt->transport_lock)
727	814	{
728		- if (list_empty(&xprt->recv) && xprt_has_timer(xprt))
	815	+ xprt->last_used = jiffies;
	816	+ if (RB_EMPTY_ROOT(&xprt->recv_queue) && xprt_has_timer(xprt))
729	817	mod_timer(&xprt->timer, xprt->last_used + xprt->idle_timeout);
730	818	}
731	819
..	..	@@ -734,18 +822,13 @@
734	822	{
735	823	struct rpc_xprt *xprt = from_timer(xprt, t, timer);
736	824
737		- spin_lock(&xprt->transport_lock);
738		- if (!list_empty(&xprt->recv))
739		- goto out_abort;
	825	+ if (!RB_EMPTY_ROOT(&xprt->recv_queue))
	826	+ return;
740	827	/* Reset xprt->last_used to avoid connect/autodisconnect cycling */
741	828	xprt->last_used = jiffies;
742	829	if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
743		- goto out_abort;
744		- spin_unlock(&xprt->transport_lock);
	830	+ return;
745	831	queue_work(xprtiod_workqueue, &xprt->task_cleanup);
746		- return;
747		-out_abort:
748		- spin_unlock(&xprt->transport_lock);
749	832	}
750	833
751	834	bool xprt_lock_connect(struct rpc_xprt *xprt,
..	..	@@ -754,33 +837,36 @@
754	837	{
755	838	bool ret = false;
756	839
757		- spin_lock_bh(&xprt->transport_lock);
	840	+ spin_lock(&xprt->transport_lock);
758	841	if (!test_bit(XPRT_LOCKED, &xprt->state))
759	842	goto out;
760	843	if (xprt->snd_task != task)
761	844	goto out;
762		- xprt_task_clear_bytes_sent(task);
	845	+ set_bit(XPRT_SND_IS_COOKIE, &xprt->state);
763	846	xprt->snd_task = cookie;
764	847	ret = true;
765	848	out:
766		- spin_unlock_bh(&xprt->transport_lock);
	849	+ spin_unlock(&xprt->transport_lock);
767	850	return ret;
768	851	}
	852	+EXPORT_SYMBOL_GPL(xprt_lock_connect);
769	853
770	854	void xprt_unlock_connect(struct rpc_xprt xprt, void cookie)
771	855	{
772		- spin_lock_bh(&xprt->transport_lock);
	856	+ spin_lock(&xprt->transport_lock);
773	857	if (xprt->snd_task != cookie)
774	858	goto out;
775	859	if (!test_bit(XPRT_LOCKED, &xprt->state))
776	860	goto out;
777	861	xprt->snd_task =NULL;
	862	+ clear_bit(XPRT_SND_IS_COOKIE, &xprt->state);
778	863	xprt->ops->release_xprt(xprt, NULL);
779	864	xprt_schedule_autodisconnect(xprt);
780	865	out:
781		- spin_unlock_bh(&xprt->transport_lock);
	866	+ spin_unlock(&xprt->transport_lock);
782	867	wake_up_bit(&xprt->state, XPRT_LOCKED);
783	868	}
	869	+EXPORT_SYMBOL_GPL(xprt_unlock_connect);
784	870
785	871	/**
786	872	* xprt_connect - schedule a transport connect operation
..	..	@@ -791,8 +877,7 @@
791	877	{
792	878	struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
793	879
794		- dprintk("RPC: %5u xprt_connect xprt %p %s connected\n", task->tk_pid,
795		- xprt, (xprt_connected(xprt) ? "is" : "is not"));
	880	+ trace_xprt_connect(xprt);
796	881
797	882	if (!xprt_bound(xprt)) {
798	883	task->tk_status = -EAGAIN;
..	..	@@ -801,14 +886,10 @@
801	886	if (!xprt_lock_write(xprt, task))
802	887	return;
803	888
804		- if (test_and_clear_bit(XPRT_CLOSE_WAIT, &xprt->state))
805		- xprt->ops->close(xprt);
806		-
807		- if (!xprt_connected(xprt)) {
808		- task->tk_rqstp->rq_bytes_sent = 0;
809		- task->tk_timeout = task->tk_rqstp->rq_timeout;
	889	+ if (!xprt_connected(xprt) && !test_bit(XPRT_CLOSE_WAIT, &xprt->state)) {
810	890	task->tk_rqstp->rq_connect_cookie = xprt->connect_cookie;
811		- rpc_sleep_on(&xprt->pending, task, xprt_connect_status);
	891	+ rpc_sleep_on_timeout(&xprt->pending, task, NULL,
	892	+ xprt_request_timeout(task->tk_rqstp));
812	893
813	894	if (test_bit(XPRT_CLOSING, &xprt->state))
814	895	return;
..	..	@@ -827,32 +908,105 @@
827	908	xprt_release_write(xprt, task);
828	909	}
829	910
830		-static void xprt_connect_status(struct rpc_task *task)
	911	+/**
	912	+ * xprt_reconnect_delay - compute the wait before scheduling a connect
	913	+ * @xprt: transport instance
	914	+ *
	915	+ */
	916	+unsigned long xprt_reconnect_delay(const struct rpc_xprt *xprt)
831	917	{
832		- switch (task->tk_status) {
833		- case 0:
834		- dprintk("RPC: %5u xprt_connect_status: connection established\n",
835		- task->tk_pid);
836		- break;
837		- case -ECONNREFUSED:
838		- case -ECONNRESET:
839		- case -ECONNABORTED:
840		- case -ENETUNREACH:
841		- case -EHOSTUNREACH:
842		- case -EPIPE:
843		- case -EAGAIN:
844		- dprintk("RPC: %5u xprt_connect_status: retrying\n", task->tk_pid);
845		- break;
846		- case -ETIMEDOUT:
847		- dprintk("RPC: %5u xprt_connect_status: connect attempt timed "
848		- "out\n", task->tk_pid);
849		- break;
850		- default:
851		- dprintk("RPC: %5u xprt_connect_status: error %d connecting to "
852		- "server %s\n", task->tk_pid, -task->tk_status,
853		- task->tk_rqstp->rq_xprt->servername);
854		- task->tk_status = -EIO;
	918	+ unsigned long start, now = jiffies;
	919	+
	920	+ start = xprt->stat.connect_start + xprt->reestablish_timeout;
	921	+ if (time_after(start, now))
	922	+ return start - now;
	923	+ return 0;
	924	+}
	925	+EXPORT_SYMBOL_GPL(xprt_reconnect_delay);
	926	+
	927	+/**
	928	+ * xprt_reconnect_backoff - compute the new re-establish timeout
	929	+ * @xprt: transport instance
	930	+ * @init_to: initial reestablish timeout
	931	+ *
	932	+ */
	933	+void xprt_reconnect_backoff(struct rpc_xprt *xprt, unsigned long init_to)
	934	+{
	935	+ xprt->reestablish_timeout <<= 1;
	936	+ if (xprt->reestablish_timeout > xprt->max_reconnect_timeout)
	937	+ xprt->reestablish_timeout = xprt->max_reconnect_timeout;
	938	+ if (xprt->reestablish_timeout < init_to)
	939	+ xprt->reestablish_timeout = init_to;
	940	+}
	941	+EXPORT_SYMBOL_GPL(xprt_reconnect_backoff);
	942	+
	943	+enum xprt_xid_rb_cmp {
	944	+ XID_RB_EQUAL,
	945	+ XID_RB_LEFT,
	946	+ XID_RB_RIGHT,
	947	+};
	948	+static enum xprt_xid_rb_cmp
	949	+xprt_xid_cmp(__be32 xid1, __be32 xid2)
	950	+{
	951	+ if (xid1 == xid2)
	952	+ return XID_RB_EQUAL;
	953	+ if ((__force u32)xid1 < (__force u32)xid2)
	954	+ return XID_RB_LEFT;
	955	+ return XID_RB_RIGHT;
	956	+}
	957	+
	958	+static struct rpc_rqst *
	959	+xprt_request_rb_find(struct rpc_xprt *xprt, __be32 xid)
	960	+{
	961	+ struct rb_node *n = xprt->recv_queue.rb_node;
	962	+ struct rpc_rqst *req;
	963	+
	964	+ while (n != NULL) {
	965	+ req = rb_entry(n, struct rpc_rqst, rq_recv);
	966	+ switch (xprt_xid_cmp(xid, req->rq_xid)) {
	967	+ case XID_RB_LEFT:
	968	+ n = n->rb_left;
	969	+ break;
	970	+ case XID_RB_RIGHT:
	971	+ n = n->rb_right;
	972	+ break;
	973	+ case XID_RB_EQUAL:
	974	+ return req;
	975	+ }
855	976	}
	977	+ return NULL;
	978	+}
	979	+
	980	+static void
	981	+xprt_request_rb_insert(struct rpc_xprt xprt, struct rpc_rqst new)
	982	+{
	983	+ struct rb_node **p = &xprt->recv_queue.rb_node;
	984	+ struct rb_node *n = NULL;
	985	+ struct rpc_rqst *req;
	986	+
	987	+ while (*p != NULL) {
	988	+ n = *p;
	989	+ req = rb_entry(n, struct rpc_rqst, rq_recv);
	990	+ switch(xprt_xid_cmp(new->rq_xid, req->rq_xid)) {
	991	+ case XID_RB_LEFT:
	992	+ p = &n->rb_left;
	993	+ break;
	994	+ case XID_RB_RIGHT:
	995	+ p = &n->rb_right;
	996	+ break;
	997	+ case XID_RB_EQUAL:
	998	+ WARN_ON_ONCE(new != req);
	999	+ return;
	1000	+ }
	1001	+ }
	1002	+ rb_link_node(&new->rq_recv, n, p);
	1003	+ rb_insert_color(&new->rq_recv, &xprt->recv_queue);
	1004	+}
	1005	+
	1006	+static void
	1007	+xprt_request_rb_remove(struct rpc_xprt xprt, struct rpc_rqst req)
	1008	+{
	1009	+ rb_erase(&req->rq_recv, &xprt->recv_queue);
856	1010	}
857	1011
858	1012	/**
..	..	@@ -860,18 +1014,18 @@
860	1014	* @xprt: transport on which the original request was transmitted
861	1015	* @xid: RPC XID of incoming reply
862	1016	*
863		- * Caller holds xprt->recv_lock.
	1017	+ * Caller holds xprt->queue_lock.
864	1018	*/
865	1019	struct rpc_rqst xprt_lookup_rqst(struct rpc_xprt xprt, __be32 xid)
866	1020	{
867	1021	struct rpc_rqst *entry;
868	1022
869		- list_for_each_entry(entry, &xprt->recv, rq_list)
870		- if (entry->rq_xid == xid) {
871		- trace_xprt_lookup_rqst(xprt, xid, 0);
872		- entry->rq_rtt = ktime_sub(ktime_get(), entry->rq_xtime);
873		- return entry;
874		- }
	1023	+ entry = xprt_request_rb_find(xprt, xid);
	1024	+ if (entry != NULL) {
	1025	+ trace_xprt_lookup_rqst(xprt, xid, 0);
	1026	+ entry->rq_rtt = ktime_sub(ktime_get(), entry->rq_xtime);
	1027	+ return entry;
	1028	+ }
875	1029
876	1030	dprintk("RPC: xprt_lookup_rqst did not find xid %08x\n",
877	1031	ntohl(xid));
..	..	@@ -881,16 +1035,22 @@
881	1035	}
882	1036	EXPORT_SYMBOL_GPL(xprt_lookup_rqst);
883	1037
	1038	+static bool
	1039	+xprt_is_pinned_rqst(struct rpc_rqst *req)
	1040	+{
	1041	+ return atomic_read(&req->rq_pin) != 0;
	1042	+}
	1043	+
884	1044	/**
885	1045	* xprt_pin_rqst - Pin a request on the transport receive list
886	1046	* @req: Request to pin
887	1047	*
888	1048	* Caller must ensure this is atomic with the call to xprt_lookup_rqst()
889		- * so should be holding the xprt transport lock.
	1049	+ * so should be holding xprt->queue_lock.
890	1050	*/
891	1051	void xprt_pin_rqst(struct rpc_rqst *req)
892	1052	{
893		- set_bit(RPC_TASK_MSG_RECV, &req->rq_task->tk_runstate);
	1053	+ atomic_inc(&req->rq_pin);
894	1054	}
895	1055	EXPORT_SYMBOL_GPL(xprt_pin_rqst);
896	1056
..	..	@@ -898,38 +1058,88 @@
898	1058	* xprt_unpin_rqst - Unpin a request on the transport receive list
899	1059	* @req: Request to pin
900	1060	*
901		- * Caller should be holding the xprt transport lock.
	1061	+ * Caller should be holding xprt->queue_lock.
902	1062	*/
903	1063	void xprt_unpin_rqst(struct rpc_rqst *req)
904	1064	{
905		- struct rpc_task *task = req->rq_task;
906		-
907		- clear_bit(RPC_TASK_MSG_RECV, &task->tk_runstate);
908		- if (test_bit(RPC_TASK_MSG_RECV_WAIT, &task->tk_runstate))
909		- wake_up_bit(&task->tk_runstate, RPC_TASK_MSG_RECV);
	1065	+ if (!test_bit(RPC_TASK_MSG_PIN_WAIT, &req->rq_task->tk_runstate)) {
	1066	+ atomic_dec(&req->rq_pin);
	1067	+ return;
	1068	+ }
	1069	+ if (atomic_dec_and_test(&req->rq_pin))
	1070	+ wake_up_var(&req->rq_pin);
910	1071	}
911	1072	EXPORT_SYMBOL_GPL(xprt_unpin_rqst);
912	1073
913	1074	static void xprt_wait_on_pinned_rqst(struct rpc_rqst *req)
914		-__must_hold(&req->rq_xprt->recv_lock)
915	1075	{
916		- struct rpc_task *task = req->rq_task;
	1076	+ wait_var_event(&req->rq_pin, !xprt_is_pinned_rqst(req));
	1077	+}
917	1078
918		- if (task && test_bit(RPC_TASK_MSG_RECV, &task->tk_runstate)) {
919		- spin_unlock(&req->rq_xprt->recv_lock);
920		- set_bit(RPC_TASK_MSG_RECV_WAIT, &task->tk_runstate);
921		- wait_on_bit(&task->tk_runstate, RPC_TASK_MSG_RECV,
922		- TASK_UNINTERRUPTIBLE);
923		- clear_bit(RPC_TASK_MSG_RECV_WAIT, &task->tk_runstate);
924		- spin_lock(&req->rq_xprt->recv_lock);
925		- }
	1079	+static bool
	1080	+xprt_request_data_received(struct rpc_task *task)
	1081	+{
	1082	+ return !test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate) &&
	1083	+ READ_ONCE(task->tk_rqstp->rq_reply_bytes_recvd) != 0;
	1084	+}
	1085	+
	1086	+static bool
	1087	+xprt_request_need_enqueue_receive(struct rpc_task task, struct rpc_rqst req)
	1088	+{
	1089	+ return !test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate) &&
	1090	+ READ_ONCE(task->tk_rqstp->rq_reply_bytes_recvd) == 0;
	1091	+}
	1092	+
	1093	+/**
	1094	+ * xprt_request_enqueue_receive - Add an request to the receive queue
	1095	+ * @task: RPC task
	1096	+ *
	1097	+ */
	1098	+void
	1099	+xprt_request_enqueue_receive(struct rpc_task *task)
	1100	+{
	1101	+ struct rpc_rqst *req = task->tk_rqstp;
	1102	+ struct rpc_xprt *xprt = req->rq_xprt;
	1103	+
	1104	+ if (!xprt_request_need_enqueue_receive(task, req))
	1105	+ return;
	1106	+
	1107	+ xprt_request_prepare(task->tk_rqstp);
	1108	+ spin_lock(&xprt->queue_lock);
	1109	+
	1110	+ /* Update the softirq receive buffer */
	1111	+ memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
	1112	+ sizeof(req->rq_private_buf));
	1113	+
	1114	+ /* Add request to the receive list */
	1115	+ xprt_request_rb_insert(xprt, req);
	1116	+ set_bit(RPC_TASK_NEED_RECV, &task->tk_runstate);
	1117	+ spin_unlock(&xprt->queue_lock);
	1118	+
	1119	+ /* Turn off autodisconnect */
	1120	+ del_singleshot_timer_sync(&xprt->timer);
	1121	+}
	1122	+
	1123	+/**
	1124	+ * xprt_request_dequeue_receive_locked - Remove a request from the receive queue
	1125	+ * @task: RPC task
	1126	+ *
	1127	+ * Caller must hold xprt->queue_lock.
	1128	+ */
	1129	+static void
	1130	+xprt_request_dequeue_receive_locked(struct rpc_task *task)
	1131	+{
	1132	+ struct rpc_rqst *req = task->tk_rqstp;
	1133	+
	1134	+ if (test_and_clear_bit(RPC_TASK_NEED_RECV, &task->tk_runstate))
	1135	+ xprt_request_rb_remove(req->rq_xprt, req);
926	1136	}
927	1137
928	1138	/**
929	1139	* xprt_update_rtt - Update RPC RTT statistics
930	1140	* @task: RPC request that recently completed
931	1141	*
932		- * Caller holds xprt->recv_lock.
	1142	+ * Caller holds xprt->queue_lock.
933	1143	*/
934	1144	void xprt_update_rtt(struct rpc_task *task)
935	1145	{
..	..	@@ -951,25 +1161,21 @@
951	1161	* @task: RPC request that recently completed
952	1162	* @copied: actual number of bytes received from the transport
953	1163	*
954		- * Caller holds xprt->recv_lock.
	1164	+ * Caller holds xprt->queue_lock.
955	1165	*/
956	1166	void xprt_complete_rqst(struct rpc_task *task, int copied)
957	1167	{
958	1168	struct rpc_rqst *req = task->tk_rqstp;
959	1169	struct rpc_xprt *xprt = req->rq_xprt;
960	1170
961		- dprintk("RPC: %5u xid %08x complete (%d bytes received)\n",
962		- task->tk_pid, ntohl(req->rq_xid), copied);
963		- trace_xprt_complete_rqst(xprt, req->rq_xid, copied);
964		-
965	1171	xprt->stat.recvs++;
966	1172
967		- list_del_init(&req->rq_list);
968	1173	req->rq_private_buf.len = copied;
969	1174	/* Ensure all writes are done before we update */
970	1175	/* req->rq_reply_bytes_recvd */
971	1176	smp_wmb();
972	1177	req->rq_reply_bytes_recvd = copied;
	1178	+ xprt_request_dequeue_receive_locked(task);
973	1179	rpc_wake_up_queued_task(&xprt->pending, task);
974	1180	}
975	1181	EXPORT_SYMBOL_GPL(xprt_complete_rqst);
..	..	@@ -991,6 +1197,234 @@
991	1197	}
992	1198
993	1199	/**
	1200	+ * xprt_wait_for_reply_request_def - wait for reply
	1201	+ * @task: pointer to rpc_task
	1202	+ *
	1203	+ * Set a request's retransmit timeout based on the transport's
	1204	+ * default timeout parameters. Used by transports that don't adjust
	1205	+ * the retransmit timeout based on round-trip time estimation,
	1206	+ * and put the task to sleep on the pending queue.
	1207	+ */
	1208	+void xprt_wait_for_reply_request_def(struct rpc_task *task)
	1209	+{
	1210	+ struct rpc_rqst *req = task->tk_rqstp;
	1211	+
	1212	+ rpc_sleep_on_timeout(&req->rq_xprt->pending, task, xprt_timer,
	1213	+ xprt_request_timeout(req));
	1214	+}
	1215	+EXPORT_SYMBOL_GPL(xprt_wait_for_reply_request_def);
	1216	+
	1217	+/**
	1218	+ * xprt_wait_for_reply_request_rtt - wait for reply using RTT estimator
	1219	+ * @task: pointer to rpc_task
	1220	+ *
	1221	+ * Set a request's retransmit timeout using the RTT estimator,
	1222	+ * and put the task to sleep on the pending queue.
	1223	+ */
	1224	+void xprt_wait_for_reply_request_rtt(struct rpc_task *task)
	1225	+{
	1226	+ int timer = task->tk_msg.rpc_proc->p_timer;
	1227	+ struct rpc_clnt *clnt = task->tk_client;
	1228	+ struct rpc_rtt *rtt = clnt->cl_rtt;
	1229	+ struct rpc_rqst *req = task->tk_rqstp;
	1230	+ unsigned long max_timeout = clnt->cl_timeout->to_maxval;
	1231	+ unsigned long timeout;
	1232	+
	1233	+ timeout = rpc_calc_rto(rtt, timer);
	1234	+ timeout <<= rpc_ntimeo(rtt, timer) + req->rq_retries;
	1235	+ if (timeout > max_timeout \|\| timeout == 0)
	1236	+ timeout = max_timeout;
	1237	+ rpc_sleep_on_timeout(&req->rq_xprt->pending, task, xprt_timer,
	1238	+ jiffies + timeout);
	1239	+}
	1240	+EXPORT_SYMBOL_GPL(xprt_wait_for_reply_request_rtt);
	1241	+
	1242	+/**
	1243	+ * xprt_request_wait_receive - wait for the reply to an RPC request
	1244	+ * @task: RPC task about to send a request
	1245	+ *
	1246	+ */
	1247	+void xprt_request_wait_receive(struct rpc_task *task)
	1248	+{
	1249	+ struct rpc_rqst *req = task->tk_rqstp;
	1250	+ struct rpc_xprt *xprt = req->rq_xprt;
	1251	+
	1252	+ if (!test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate))
	1253	+ return;
	1254	+ /*
	1255	+ * Sleep on the pending queue if we're expecting a reply.
	1256	+ * The spinlock ensures atomicity between the test of
	1257	+ * req->rq_reply_bytes_recvd, and the call to rpc_sleep_on().
	1258	+ */
	1259	+ spin_lock(&xprt->queue_lock);
	1260	+ if (test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate)) {
	1261	+ xprt->ops->wait_for_reply_request(task);
	1262	+ /*
	1263	+ * Send an extra queue wakeup call if the
	1264	+ * connection was dropped in case the call to
	1265	+ * rpc_sleep_on() raced.
	1266	+ */
	1267	+ if (xprt_request_retransmit_after_disconnect(task))
	1268	+ rpc_wake_up_queued_task_set_status(&xprt->pending,
	1269	+ task, -ENOTCONN);
	1270	+ }
	1271	+ spin_unlock(&xprt->queue_lock);
	1272	+}
	1273	+
	1274	+static bool
	1275	+xprt_request_need_enqueue_transmit(struct rpc_task task, struct rpc_rqst req)
	1276	+{
	1277	+ return !test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate);
	1278	+}
	1279	+
	1280	+/**
	1281	+ * xprt_request_enqueue_transmit - queue a task for transmission
	1282	+ * @task: pointer to rpc_task
	1283	+ *
	1284	+ * Add a task to the transmission queue.
	1285	+ */
	1286	+void
	1287	+xprt_request_enqueue_transmit(struct rpc_task *task)
	1288	+{
	1289	+ struct rpc_rqst pos, req = task->tk_rqstp;
	1290	+ struct rpc_xprt *xprt = req->rq_xprt;
	1291	+
	1292	+ if (xprt_request_need_enqueue_transmit(task, req)) {
	1293	+ req->rq_bytes_sent = 0;
	1294	+ spin_lock(&xprt->queue_lock);
	1295	+ /*
	1296	+ * Requests that carry congestion control credits are added
	1297	+ * to the head of the list to avoid starvation issues.
	1298	+ */
	1299	+ if (req->rq_cong) {
	1300	+ xprt_clear_congestion_window_wait(xprt);
	1301	+ list_for_each_entry(pos, &xprt->xmit_queue, rq_xmit) {
	1302	+ if (pos->rq_cong)
	1303	+ continue;
	1304	+ /* Note: req is added _before_ pos */
	1305	+ list_add_tail(&req->rq_xmit, &pos->rq_xmit);
	1306	+ INIT_LIST_HEAD(&req->rq_xmit2);
	1307	+ goto out;
	1308	+ }
	1309	+ } else if (!req->rq_seqno) {
	1310	+ list_for_each_entry(pos, &xprt->xmit_queue, rq_xmit) {
	1311	+ if (pos->rq_task->tk_owner != task->tk_owner)
	1312	+ continue;
	1313	+ list_add_tail(&req->rq_xmit2, &pos->rq_xmit2);
	1314	+ INIT_LIST_HEAD(&req->rq_xmit);
	1315	+ goto out;
	1316	+ }
	1317	+ }
	1318	+ list_add_tail(&req->rq_xmit, &xprt->xmit_queue);
	1319	+ INIT_LIST_HEAD(&req->rq_xmit2);
	1320	+out:
	1321	+ set_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate);
	1322	+ spin_unlock(&xprt->queue_lock);
	1323	+ }
	1324	+}
	1325	+
	1326	+/**
	1327	+ * xprt_request_dequeue_transmit_locked - remove a task from the transmission queue
	1328	+ * @task: pointer to rpc_task
	1329	+ *
	1330	+ * Remove a task from the transmission queue
	1331	+ * Caller must hold xprt->queue_lock
	1332	+ */
	1333	+static void
	1334	+xprt_request_dequeue_transmit_locked(struct rpc_task *task)
	1335	+{
	1336	+ struct rpc_rqst *req = task->tk_rqstp;
	1337	+
	1338	+ if (!test_and_clear_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate))
	1339	+ return;
	1340	+ if (!list_empty(&req->rq_xmit)) {
	1341	+ list_del(&req->rq_xmit);
	1342	+ if (!list_empty(&req->rq_xmit2)) {
	1343	+ struct rpc_rqst *next = list_first_entry(&req->rq_xmit2,
	1344	+ struct rpc_rqst, rq_xmit2);
	1345	+ list_del(&req->rq_xmit2);
	1346	+ list_add_tail(&next->rq_xmit, &next->rq_xprt->xmit_queue);
	1347	+ }
	1348	+ } else
	1349	+ list_del(&req->rq_xmit2);
	1350	+}
	1351	+
	1352	+/**
	1353	+ * xprt_request_dequeue_transmit - remove a task from the transmission queue
	1354	+ * @task: pointer to rpc_task
	1355	+ *
	1356	+ * Remove a task from the transmission queue
	1357	+ */
	1358	+static void
	1359	+xprt_request_dequeue_transmit(struct rpc_task *task)
	1360	+{
	1361	+ struct rpc_rqst *req = task->tk_rqstp;
	1362	+ struct rpc_xprt *xprt = req->rq_xprt;
	1363	+
	1364	+ spin_lock(&xprt->queue_lock);
	1365	+ xprt_request_dequeue_transmit_locked(task);
	1366	+ spin_unlock(&xprt->queue_lock);
	1367	+}
	1368	+
	1369	+/**
	1370	+ * xprt_request_dequeue_xprt - remove a task from the transmit+receive queue
	1371	+ * @task: pointer to rpc_task
	1372	+ *
	1373	+ * Remove a task from the transmit and receive queues, and ensure that
	1374	+ * it is not pinned by the receive work item.
	1375	+ */
	1376	+void
	1377	+xprt_request_dequeue_xprt(struct rpc_task *task)
	1378	+{
	1379	+ struct rpc_rqst *req = task->tk_rqstp;
	1380	+ struct rpc_xprt *xprt = req->rq_xprt;
	1381	+
	1382	+ if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate) \|\|
	1383	+ test_bit(RPC_TASK_NEED_RECV, &task->tk_runstate) \|\|
	1384	+ xprt_is_pinned_rqst(req)) {
	1385	+ spin_lock(&xprt->queue_lock);
	1386	+ xprt_request_dequeue_transmit_locked(task);
	1387	+ xprt_request_dequeue_receive_locked(task);
	1388	+ while (xprt_is_pinned_rqst(req)) {
	1389	+ set_bit(RPC_TASK_MSG_PIN_WAIT, &task->tk_runstate);
	1390	+ spin_unlock(&xprt->queue_lock);
	1391	+ xprt_wait_on_pinned_rqst(req);
	1392	+ spin_lock(&xprt->queue_lock);
	1393	+ clear_bit(RPC_TASK_MSG_PIN_WAIT, &task->tk_runstate);
	1394	+ }
	1395	+ spin_unlock(&xprt->queue_lock);
	1396	+ }
	1397	+}
	1398	+
	1399	+/**
	1400	+ * xprt_request_prepare - prepare an encoded request for transport
	1401	+ * @req: pointer to rpc_rqst
	1402	+ *
	1403	+ * Calls into the transport layer to do whatever is needed to prepare
	1404	+ * the request for transmission or receive.
	1405	+ */
	1406	+void
	1407	+xprt_request_prepare(struct rpc_rqst *req)
	1408	+{
	1409	+ struct rpc_xprt *xprt = req->rq_xprt;
	1410	+
	1411	+ if (xprt->ops->prepare_request)
	1412	+ xprt->ops->prepare_request(req);
	1413	+}
	1414	+
	1415	+/**
	1416	+ * xprt_request_need_retransmit - Test if a task needs retransmission
	1417	+ * @task: pointer to rpc_task
	1418	+ *
	1419	+ * Test for whether a connection breakage requires the task to retransmit
	1420	+ */
	1421	+bool
	1422	+xprt_request_need_retransmit(struct rpc_task *task)
	1423	+{
	1424	+ return xprt_request_retransmit_after_disconnect(task);
	1425	+}
	1426	+
	1427	+/**
994	1428	* xprt_prepare_transmit - reserve the transport before sending a request
995	1429	* @task: RPC task about to send a request
996	1430	*
..	..	@@ -999,128 +1433,177 @@
999	1433	{
1000	1434	struct rpc_rqst *req = task->tk_rqstp;
1001	1435	struct rpc_xprt *xprt = req->rq_xprt;
1002		- bool ret = false;
1003	1436
1004		- dprintk("RPC: %5u xprt_prepare_transmit\n", task->tk_pid);
	1437	+ if (!xprt_lock_write(xprt, task)) {
	1438	+ /* Race breaker: someone may have transmitted us */
	1439	+ if (!test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate))
	1440	+ rpc_wake_up_queued_task_set_status(&xprt->sending,
	1441	+ task, 0);
	1442	+ return false;
1005	1443
1006		- spin_lock_bh(&xprt->transport_lock);
1007		- if (!req->rq_bytes_sent) {
1008		- if (req->rq_reply_bytes_recvd) {
1009		- task->tk_status = req->rq_reply_bytes_recvd;
1010		- goto out_unlock;
1011		- }
1012		- if ((task->tk_flags & RPC_TASK_NO_RETRANS_TIMEOUT)
1013		- && xprt_connected(xprt)
1014		- && req->rq_connect_cookie == xprt->connect_cookie) {
1015		- xprt->ops->set_retrans_timeout(task);
1016		- rpc_sleep_on(&xprt->pending, task, xprt_timer);
1017		- goto out_unlock;
1018		- }
1019	1444	}
1020		- if (!xprt->ops->reserve_xprt(xprt, task)) {
1021		- task->tk_status = -EAGAIN;
1022		- goto out_unlock;
1023		- }
1024		- ret = true;
1025		-out_unlock:
1026		- spin_unlock_bh(&xprt->transport_lock);
1027		- return ret;
	1445	+ return true;
1028	1446	}
1029	1447
1030	1448	void xprt_end_transmit(struct rpc_task *task)
1031	1449	{
1032		- xprt_release_write(task->tk_rqstp->rq_xprt, task);
	1450	+ struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
	1451	+
	1452	+ xprt_inject_disconnect(xprt);
	1453	+ xprt_release_write(xprt, task);
1033	1454	}
1034	1455
1035	1456	/**
1036		- * xprt_transmit - send an RPC request on a transport
1037		- * @task: controlling RPC task
	1457	+ * xprt_request_transmit - send an RPC request on a transport
	1458	+ * @req: pointer to request to transmit
	1459	+ * @snd_task: RPC task that owns the transport lock
1038	1460	*
1039		- * We have to copy the iovec because sendmsg fiddles with its contents.
	1461	+ * This performs the transmission of a single request.
	1462	+ * Note that if the request is not the same as snd_task, then it
	1463	+ * does need to be pinned.
	1464	+ * Returns '0' on success.
1040	1465	*/
1041		-void xprt_transmit(struct rpc_task *task)
	1466	+static int
	1467	+xprt_request_transmit(struct rpc_rqst req, struct rpc_task snd_task)
1042	1468	{
1043		- struct rpc_rqst *req = task->tk_rqstp;
1044		- struct rpc_xprt *xprt = req->rq_xprt;
	1469	+ struct rpc_xprt *xprt = req->rq_xprt;
	1470	+ struct rpc_task *task = req->rq_task;
1045	1471	unsigned int connect_cookie;
	1472	+ int is_retrans = RPC_WAS_SENT(task);
1046	1473	int status;
1047	1474
1048		- dprintk("RPC: %5u xprt_transmit(%u)\n", task->tk_pid, req->rq_slen);
1049		-
1050		- if (!req->rq_reply_bytes_recvd) {
1051		- if (list_empty(&req->rq_list) && rpc_reply_expected(task)) {
1052		- /*
1053		- * Add to the list only if we're expecting a reply
1054		- */
1055		- /* Update the softirq receive buffer */
1056		- memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
1057		- sizeof(req->rq_private_buf));
1058		- /* Add request to the receive list */
1059		- spin_lock(&xprt->recv_lock);
1060		- list_add_tail(&req->rq_list, &xprt->recv);
1061		- spin_unlock(&xprt->recv_lock);
1062		- xprt_reset_majortimeo(req);
1063		- /* Turn off autodisconnect */
1064		- del_singleshot_timer_sync(&xprt->timer);
	1475	+ if (!req->rq_bytes_sent) {
	1476	+ if (xprt_request_data_received(task)) {
	1477	+ status = 0;
	1478	+ goto out_dequeue;
1065	1479	}
1066		- } else if (!req->rq_bytes_sent)
1067		- return;
1068		-
1069		- connect_cookie = xprt->connect_cookie;
1070		- status = xprt->ops->send_request(task);
1071		- trace_xprt_transmit(xprt, req->rq_xid, status);
1072		- if (status != 0) {
1073		- task->tk_status = status;
1074		- return;
	1480	+ /* Verify that our message lies in the RPCSEC_GSS window */
	1481	+ if (rpcauth_xmit_need_reencode(task)) {
	1482	+ status = -EBADMSG;
	1483	+ goto out_dequeue;
	1484	+ }
	1485	+ if (RPC_SIGNALLED(task)) {
	1486	+ status = -ERESTARTSYS;
	1487	+ goto out_dequeue;
	1488	+ }
1075	1489	}
	1490	+
	1491	+ /*
	1492	+ * Update req->rq_ntrans before transmitting to avoid races with
	1493	+ * xprt_update_rtt(), which needs to know that it is recording a
	1494	+ * reply to the first transmission.
	1495	+ */
	1496	+ req->rq_ntrans++;
	1497	+
	1498	+ trace_rpc_xdr_sendto(task, &req->rq_snd_buf);
	1499	+ connect_cookie = xprt->connect_cookie;
	1500	+ status = xprt->ops->send_request(req);
	1501	+ if (status != 0) {
	1502	+ req->rq_ntrans--;
	1503	+ trace_xprt_transmit(req, status);
	1504	+ return status;
	1505	+ }
	1506	+
	1507	+ if (is_retrans)
	1508	+ task->tk_client->cl_stats->rpcretrans++;
	1509	+
1076	1510	xprt_inject_disconnect(xprt);
1077	1511
1078		- dprintk("RPC: %5u xmit complete\n", task->tk_pid);
1079	1512	task->tk_flags \|= RPC_TASK_SENT;
1080		- spin_lock_bh(&xprt->transport_lock);
1081		-
1082		- xprt->ops->set_retrans_timeout(task);
	1513	+ spin_lock(&xprt->transport_lock);
1083	1514
1084	1515	xprt->stat.sends++;
1085	1516	xprt->stat.req_u += xprt->stat.sends - xprt->stat.recvs;
1086	1517	xprt->stat.bklog_u += xprt->backlog.qlen;
1087	1518	xprt->stat.sending_u += xprt->sending.qlen;
1088	1519	xprt->stat.pending_u += xprt->pending.qlen;
1089		- spin_unlock_bh(&xprt->transport_lock);
	1520	+ spin_unlock(&xprt->transport_lock);
1090	1521
1091	1522	req->rq_connect_cookie = connect_cookie;
1092		- if (rpc_reply_expected(task) && !READ_ONCE(req->rq_reply_bytes_recvd)) {
1093		- /*
1094		- * Sleep on the pending queue if we're expecting a reply.
1095		- * The spinlock ensures atomicity between the test of
1096		- * req->rq_reply_bytes_recvd, and the call to rpc_sleep_on().
1097		- */
1098		- spin_lock(&xprt->recv_lock);
1099		- if (!req->rq_reply_bytes_recvd) {
1100		- rpc_sleep_on(&xprt->pending, task, xprt_timer);
1101		- /*
1102		- * Send an extra queue wakeup call if the
1103		- * connection was dropped in case the call to
1104		- * rpc_sleep_on() raced.
1105		- */
1106		- if (!xprt_connected(xprt))
1107		- xprt_wake_pending_tasks(xprt, -ENOTCONN);
1108		- }
1109		- spin_unlock(&xprt->recv_lock);
1110		- }
	1523	+out_dequeue:
	1524	+ trace_xprt_transmit(req, status);
	1525	+ xprt_request_dequeue_transmit(task);
	1526	+ rpc_wake_up_queued_task_set_status(&xprt->sending, task, status);
	1527	+ return status;
1111	1528	}
1112	1529
1113		-static void xprt_add_backlog(struct rpc_xprt xprt, struct rpc_task task)
	1530	+/**
	1531	+ * xprt_transmit - send an RPC request on a transport
	1532	+ * @task: controlling RPC task
	1533	+ *
	1534	+ * Attempts to drain the transmit queue. On exit, either the transport
	1535	+ * signalled an error that needs to be handled before transmission can
	1536	+ * resume, or @task finished transmitting, and detected that it already
	1537	+ * received a reply.
	1538	+ */
	1539	+void
	1540	+xprt_transmit(struct rpc_task *task)
	1541	+{
	1542	+ struct rpc_rqst next, req = task->tk_rqstp;
	1543	+ struct rpc_xprt *xprt = req->rq_xprt;
	1544	+ int status;
	1545	+
	1546	+ spin_lock(&xprt->queue_lock);
	1547	+ for (;;) {
	1548	+ next = list_first_entry_or_null(&xprt->xmit_queue,
	1549	+ struct rpc_rqst, rq_xmit);
	1550	+ if (!next)
	1551	+ break;
	1552	+ xprt_pin_rqst(next);
	1553	+ spin_unlock(&xprt->queue_lock);
	1554	+ status = xprt_request_transmit(next, task);
	1555	+ if (status == -EBADMSG && next != req)
	1556	+ status = 0;
	1557	+ spin_lock(&xprt->queue_lock);
	1558	+ xprt_unpin_rqst(next);
	1559	+ if (status < 0) {
	1560	+ if (test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate))
	1561	+ task->tk_status = status;
	1562	+ break;
	1563	+ }
	1564	+ /* Was @task transmitted, and has it received a reply? */
	1565	+ if (xprt_request_data_received(task) &&
	1566	+ !test_bit(RPC_TASK_NEED_XMIT, &task->tk_runstate))
	1567	+ break;
	1568	+ cond_resched_lock(&xprt->queue_lock);
	1569	+ }
	1570	+ spin_unlock(&xprt->queue_lock);
	1571	+}
	1572	+
	1573	+static void xprt_complete_request_init(struct rpc_task *task)
	1574	+{
	1575	+ if (task->tk_rqstp)
	1576	+ xprt_request_init(task);
	1577	+}
	1578	+
	1579	+void xprt_add_backlog(struct rpc_xprt xprt, struct rpc_task task)
1114	1580	{
1115	1581	set_bit(XPRT_CONGESTED, &xprt->state);
1116		- rpc_sleep_on(&xprt->backlog, task, NULL);
	1582	+ rpc_sleep_on(&xprt->backlog, task, xprt_complete_request_init);
	1583	+}
	1584	+EXPORT_SYMBOL_GPL(xprt_add_backlog);
	1585	+
	1586	+static bool __xprt_set_rq(struct rpc_task task, void data)
	1587	+{
	1588	+ struct rpc_rqst *req = data;
	1589	+
	1590	+ if (task->tk_rqstp == NULL) {
	1591	+ memset(req, 0, sizeof(req)); / mark unused */
	1592	+ task->tk_rqstp = req;
	1593	+ return true;
	1594	+ }
	1595	+ return false;
1117	1596	}
1118	1597
1119		-static void xprt_wake_up_backlog(struct rpc_xprt *xprt)
	1598	+bool xprt_wake_up_backlog(struct rpc_xprt xprt, struct rpc_rqst req)
1120	1599	{
1121		- if (rpc_wake_up_next(&xprt->backlog) == NULL)
	1600	+ if (rpc_wake_up_first(&xprt->backlog, __xprt_set_rq, req) == NULL) {
1122	1601	clear_bit(XPRT_CONGESTED, &xprt->state);
	1602	+ return false;
	1603	+ }
	1604	+ return true;
1123	1605	}
	1606	+EXPORT_SYMBOL_GPL(xprt_wake_up_backlog);
1124	1607
1125	1608	static bool xprt_throttle_congested(struct rpc_xprt xprt, struct rpc_task task)
1126	1609	{
..	..	@@ -1130,7 +1613,7 @@
1130	1613	goto out;
1131	1614	spin_lock(&xprt->reserve_lock);
1132	1615	if (test_bit(XPRT_CONGESTED, &xprt->state)) {
1133		- rpc_sleep_on(&xprt->backlog, task, NULL);
	1616	+ xprt_add_backlog(xprt, task);
1134	1617	ret = true;
1135	1618	}
1136	1619	spin_unlock(&xprt->reserve_lock);
..	..	@@ -1141,12 +1624,15 @@
1141	1624	static struct rpc_rqst xprt_dynamic_alloc_slot(struct rpc_xprt xprt)
1142	1625	{
1143	1626	struct rpc_rqst *req = ERR_PTR(-EAGAIN);
	1627	+ gfp_t gfp_mask = GFP_KERNEL;
1144	1628
1145	1629	if (xprt->num_reqs >= xprt->max_reqs)
1146	1630	goto out;
1147	1631	++xprt->num_reqs;
1148	1632	spin_unlock(&xprt->reserve_lock);
1149		- req = kzalloc(sizeof(struct rpc_rqst), GFP_NOFS);
	1633	+ if (current->flags & PF_WQ_WORKER)
	1634	+ gfp_mask \|= __GFP_NORETRY \| __GFP_NOWARN;
	1635	+ req = kzalloc(sizeof(*req), gfp_mask);
1150	1636	spin_lock(&xprt->reserve_lock);
1151	1637	if (req != NULL)
1152	1638	goto out;
..	..	@@ -1188,7 +1674,7 @@
1188	1674	case -EAGAIN:
1189	1675	xprt_add_backlog(xprt, task);
1190	1676	dprintk("RPC: waiting for request slot\n");
1191		- /* fall through */
	1677	+ fallthrough;
1192	1678	default:
1193	1679	task->tk_status = -EAGAIN;
1194	1680	}
..	..	@@ -1204,28 +1690,14 @@
1204	1690	}
1205	1691	EXPORT_SYMBOL_GPL(xprt_alloc_slot);
1206	1692
1207		-void xprt_lock_and_alloc_slot(struct rpc_xprt xprt, struct rpc_task task)
1208		-{
1209		- /* Note: grabbing the xprt_lock_write() ensures that we throttle
1210		- * new slot allocation if the transport is congested (i.e. when
1211		- * reconnecting a stream transport or when out of socket write
1212		- * buffer space).
1213		- */
1214		- if (xprt_lock_write(xprt, task)) {
1215		- xprt_alloc_slot(xprt, task);
1216		- xprt_release_write(xprt, task);
1217		- }
1218		-}
1219		-EXPORT_SYMBOL_GPL(xprt_lock_and_alloc_slot);
1220		-
1221	1693	void xprt_free_slot(struct rpc_xprt xprt, struct rpc_rqst req)
1222	1694	{
1223	1695	spin_lock(&xprt->reserve_lock);
1224		- if (!xprt_dynamic_free_slot(xprt, req)) {
	1696	+ if (!xprt_wake_up_backlog(xprt, req) &&
	1697	+ !xprt_dynamic_free_slot(xprt, req)) {
1225	1698	memset(req, 0, sizeof(req)); / mark unused */
1226	1699	list_add(&req->rq_list, &xprt->free);
1227	1700	}
1228		- xprt_wake_up_backlog(xprt);
1229	1701	spin_unlock(&xprt->reserve_lock);
1230	1702	}
1231	1703	EXPORT_SYMBOL_GPL(xprt_free_slot);
..	..	@@ -1284,6 +1756,12 @@
1284	1756	}
1285	1757	EXPORT_SYMBOL_GPL(xprt_free);
1286	1758
	1759	+static void
	1760	+xprt_init_connect_cookie(struct rpc_rqst req, struct rpc_xprt xprt)
	1761	+{
	1762	+ req->rq_connect_cookie = xprt_connect_cookie(xprt) - 1;
	1763	+}
	1764	+
1287	1765	static __be32
1288	1766	xprt_alloc_xid(struct rpc_xprt *xprt)
1289	1767	{
..	..	@@ -1307,22 +1785,21 @@
1307	1785	struct rpc_xprt *xprt = task->tk_xprt;
1308	1786	struct rpc_rqst *req = task->tk_rqstp;
1309	1787
1310		- INIT_LIST_HEAD(&req->rq_list);
1311		- req->rq_timeout = task->tk_client->cl_timeout->to_initval;
1312	1788	req->rq_task = task;
1313	1789	req->rq_xprt = xprt;
1314	1790	req->rq_buffer = NULL;
1315	1791	req->rq_xid = xprt_alloc_xid(xprt);
1316		- req->rq_connect_cookie = xprt->connect_cookie - 1;
1317		- req->rq_bytes_sent = 0;
	1792	+ xprt_init_connect_cookie(req, xprt);
1318	1793	req->rq_snd_buf.len = 0;
1319	1794	req->rq_snd_buf.buflen = 0;
1320	1795	req->rq_rcv_buf.len = 0;
1321	1796	req->rq_rcv_buf.buflen = 0;
	1797	+ req->rq_snd_buf.bvec = NULL;
	1798	+ req->rq_rcv_buf.bvec = NULL;
1322	1799	req->rq_release_snd_buf = NULL;
1323		- xprt_reset_majortimeo(req);
1324		- dprintk("RPC: %5u reserved req %p xid %08x\n", task->tk_pid,
1325		- req, ntohl(req->rq_xid));
	1800	+ xprt_init_majortimeo(task, req);
	1801	+
	1802	+ trace_xprt_reserve(req);
1326	1803	}
1327	1804
1328	1805	static void
..	..	@@ -1349,7 +1826,6 @@
1349	1826	if (task->tk_rqstp != NULL)
1350	1827	return;
1351	1828
1352		- task->tk_timeout = 0;
1353	1829	task->tk_status = -EAGAIN;
1354	1830	if (!xprt_throttle_congested(xprt, task))
1355	1831	xprt_do_reserve(xprt, task);
..	..	@@ -1372,7 +1848,6 @@
1372	1848	if (task->tk_rqstp != NULL)
1373	1849	return;
1374	1850
1375		- task->tk_timeout = 0;
1376	1851	task->tk_status = -EAGAIN;
1377	1852	xprt_do_reserve(xprt, task);
1378	1853	}
..	..	@@ -1390,45 +1865,52 @@
1390	1865	if (req == NULL) {
1391	1866	if (task->tk_client) {
1392	1867	xprt = task->tk_xprt;
1393		- if (xprt->snd_task == task)
1394		- xprt_release_write(xprt, task);
	1868	+ xprt_release_write(xprt, task);
1395	1869	}
1396	1870	return;
1397	1871	}
1398	1872
1399	1873	xprt = req->rq_xprt;
1400		- if (task->tk_ops->rpc_count_stats != NULL)
1401		- task->tk_ops->rpc_count_stats(task, task->tk_calldata);
1402		- else if (task->tk_client)
1403		- rpc_count_iostats(task, task->tk_client->cl_metrics);
1404		- spin_lock(&xprt->recv_lock);
1405		- if (!list_empty(&req->rq_list)) {
1406		- list_del_init(&req->rq_list);
1407		- xprt_wait_on_pinned_rqst(req);
1408		- }
1409		- spin_unlock(&xprt->recv_lock);
1410		- spin_lock_bh(&xprt->transport_lock);
	1874	+ xprt_request_dequeue_xprt(task);
	1875	+ spin_lock(&xprt->transport_lock);
1411	1876	xprt->ops->release_xprt(xprt, task);
1412	1877	if (xprt->ops->release_request)
1413	1878	xprt->ops->release_request(task);
1414		- xprt->last_used = jiffies;
1415	1879	xprt_schedule_autodisconnect(xprt);
1416		- spin_unlock_bh(&xprt->transport_lock);
	1880	+ spin_unlock(&xprt->transport_lock);
1417	1881	if (req->rq_buffer)
1418	1882	xprt->ops->buf_free(task);
1419		- xprt_inject_disconnect(xprt);
	1883	+ xdr_free_bvec(&req->rq_rcv_buf);
	1884	+ xdr_free_bvec(&req->rq_snd_buf);
1420	1885	if (req->rq_cred != NULL)
1421	1886	put_rpccred(req->rq_cred);
1422		- task->tk_rqstp = NULL;
1423	1887	if (req->rq_release_snd_buf)
1424	1888	req->rq_release_snd_buf(req);
1425	1889
1426		- dprintk("RPC: %5u release request %p\n", task->tk_pid, req);
	1890	+ task->tk_rqstp = NULL;
1427	1891	if (likely(!bc_prealloc(req)))
1428	1892	xprt->ops->free_slot(xprt, req);
1429	1893	else
1430	1894	xprt_free_bc_request(req);
1431	1895	}
	1896	+
	1897	+#ifdef CONFIG_SUNRPC_BACKCHANNEL
	1898	+void
	1899	+xprt_init_bc_request(struct rpc_rqst req, struct rpc_task task)
	1900	+{
	1901	+ struct xdr_buf *xbufp = &req->rq_snd_buf;
	1902	+
	1903	+ task->tk_rqstp = req;
	1904	+ req->rq_task = task;
	1905	+ xprt_init_connect_cookie(req, req->rq_xprt);
	1906	+ /*
	1907	+ * Set up the xdr_buf length.
	1908	+ * This also indicates that the buffer is XDR encoded already.
	1909	+ */
	1910	+ xbufp->len = xbufp->head[0].iov_len + xbufp->page_len +
	1911	+ xbufp->tail[0].iov_len;
	1912	+}
	1913	+#endif
1432	1914
1433	1915	static void xprt_init(struct rpc_xprt xprt, struct net net)
1434	1916	{
..	..	@@ -1436,10 +1918,11 @@
1436	1918
1437	1919	spin_lock_init(&xprt->transport_lock);
1438	1920	spin_lock_init(&xprt->reserve_lock);
1439		- spin_lock_init(&xprt->recv_lock);
	1921	+ spin_lock_init(&xprt->queue_lock);
1440	1922
1441	1923	INIT_LIST_HEAD(&xprt->free);
1442		- INIT_LIST_HEAD(&xprt->recv);
	1924	+ xprt->recv_queue = RB_ROOT;
	1925	+ INIT_LIST_HEAD(&xprt->xmit_queue);
1443	1926	#if defined(CONFIG_SUNRPC_BACKCHANNEL)
1444	1927	spin_lock_init(&xprt->bc_pa_lock);
1445	1928	INIT_LIST_HEAD(&xprt->bc_pa_list);
..	..	@@ -1452,7 +1935,7 @@
1452	1935
1453	1936	rpc_init_wait_queue(&xprt->binding, "xprt_binding");
1454	1937	rpc_init_wait_queue(&xprt->pending, "xprt_pending");
1455		- rpc_init_priority_wait_queue(&xprt->sending, "xprt_sending");
	1938	+ rpc_init_wait_queue(&xprt->sending, "xprt_sending");
1456	1939	rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog");
1457	1940
1458	1941	xprt_init_xid(xprt);
..	..	@@ -1483,11 +1966,8 @@
1483	1966
1484	1967	found:
1485	1968	xprt = t->setup(args);
1486		- if (IS_ERR(xprt)) {
1487		- dprintk("RPC: xprt_create_transport: failed, %ld\n",
1488		- -PTR_ERR(xprt));
	1969	+ if (IS_ERR(xprt))
1489	1970	goto out;
1490		- }
1491	1971	if (args->flags & XPRT_CREATE_NO_IDLE_TIMEOUT)
1492	1972	xprt->idle_timeout = 0;
1493	1973	INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
..	..	@@ -1508,8 +1988,7 @@
1508	1988
1509	1989	rpc_xprt_debugfs_register(xprt);
1510	1990
1511		- dprintk("RPC: created transport %p with %u slots\n", xprt,
1512		- xprt->max_reqs);
	1991	+ trace_xprt_create(xprt);
1513	1992	out:
1514	1993	return xprt;
1515	1994	}
..	..	@@ -1519,12 +1998,19 @@
1519	1998	struct rpc_xprt *xprt =
1520	1999	container_of(work, struct rpc_xprt, task_cleanup);
1521	2000
	2001	+ trace_xprt_destroy(xprt);
	2002	+
1522	2003	rpc_xprt_debugfs_unregister(xprt);
1523	2004	rpc_destroy_wait_queue(&xprt->binding);
1524	2005	rpc_destroy_wait_queue(&xprt->pending);
1525	2006	rpc_destroy_wait_queue(&xprt->sending);
1526	2007	rpc_destroy_wait_queue(&xprt->backlog);
1527	2008	kfree(xprt->servername);
	2009	+ /*
	2010	+ * Destroy any existing back channel
	2011	+ */
	2012	+ xprt_destroy_backchannel(xprt, UINT_MAX);
	2013	+
1528	2014	/*
1529	2015	* Tear down transport state and free the rpc_xprt
1530	2016	*/
..	..	@@ -1538,14 +2024,19 @@
1538	2024	*/
1539	2025	static void xprt_destroy(struct rpc_xprt *xprt)
1540	2026	{
1541		- dprintk("RPC: destroying transport %p\n", xprt);
1542		-
1543	2027	/*
1544	2028	* Exclude transport connect/disconnect handlers and autoclose
1545	2029	*/
1546	2030	wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_UNINTERRUPTIBLE);
1547	2031
	2032	+ /*
	2033	+ * xprt_schedule_autodisconnect() can run after XPRT_LOCKED
	2034	+ * is cleared. We use ->transport_lock to ensure the mod_timer()
	2035	+ * can only run before del_time_sync(), never after.
	2036	+ */
	2037	+ spin_lock(&xprt->transport_lock);
1548	2038	del_timer_sync(&xprt->timer);
	2039	+ spin_unlock(&xprt->transport_lock);
1549	2040
1550	2041	/*
1551	2042	* Destroy sockets etc from the system workqueue so they can