[PATCH v1 0/6] svcrdma Receive batch-posting, take 2

[PATCH v1 0/6] svcrdma Receive batch-posting, take 2

[Chuck Lever]

Hi-

I found the problem with the original (now reverted) commit to
batch-post Receives. I've taken some time to split the change
into smaller pieces and update a few documenting comments.


---

Chuck Lever (6):
      svcrdma: RPCDBG_FACILITY is no longer used
      svcrdma: Provide an explanatory comment in CMA event handler
      svcrdma: Remove stale comment for svc_rdma_wc_receive()
      svcrdma: Add a batch Receive posting mechanism
      svcrdma: Use svc_rdma_refresh_recvs() in wc_receive
      svcrdma: Maintain a Receive water mark


 include/linux/sunrpc/svc_rdma.h          |  2 +
 net/sunrpc/xprtrdma/svc_rdma_recvfrom.c  | 91 +++++++++++++-----------
 net/sunrpc/xprtrdma/svc_rdma_transport.c |  8 ++-
 3 files changed, 60 insertions(+), 41 deletions(-)

--
Chuck Lever

[PATCH v1 1/6] svcrdma: RPCDBG_FACILITY is no longer used

[Chuck Lever]

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 net/sunrpc/xprtrdma/svc_rdma_recvfrom.c |    2 --
 net/sunrpc/xprtrdma/svc_rdma_sendto.c   |    2 --
 2 files changed, 4 deletions(-)

diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index 7d34290e2ff8..215d2adadbdd 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -107,8 +107,6 @@
 #include "xprt_rdma.h"
 #include <trace/events/rpcrdma.h>
 
-#define RPCDBG_FACILITY	RPCDBG_SVCXPRT
-
 static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc);
 
 static inline struct svc_rdma_recv_ctxt *
diff --git a/net/sunrpc/xprtrdma/svc_rdma_sendto.c b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
index 52c759a8543e..e6fab5dd20d0 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_sendto.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
@@ -111,8 +111,6 @@
 #include "xprt_rdma.h"
 #include <trace/events/rpcrdma.h>
 
-#define RPCDBG_FACILITY	RPCDBG_SVCXPRT
-
 static void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc);
 
 static inline struct svc_rdma_send_ctxt *

[PATCH v1 2/6] svcrdma: Provide an explanatory comment in CMA event handler

[Chuck Lever]

Clean up: explain why svc_xprt_enqueue() is invoked in the event
handler even though no xpt_flags bits are toggled here.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 net/sunrpc/xprtrdma/svc_rdma_transport.c |    3 +++
 1 file changed, 3 insertions(+)

diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index c895f80df659..046a07da5cf9 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -279,6 +279,9 @@ static int svc_rdma_cma_handler(struct rdma_cm_id *cma_id,
 	switch (event->event) {
 	case RDMA_CM_EVENT_ESTABLISHED:
 		clear_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags);
+
+		/* Handle any requests that were received while
+		 * CONN_PENDING was set. */
 		svc_xprt_enqueue(xprt);
 		break;
 	case RDMA_CM_EVENT_DISCONNECTED:

[PATCH v1 3/6] svcrdma: Remove stale comment for svc_rdma_wc_receive()

[Chuck Lever]

xprt pinning was removed in commit 365e9992b90f ("svcrdma: Remove
transport reference counting"), but this comment was not updated
to reflect that change.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 net/sunrpc/xprtrdma/svc_rdma_recvfrom.c |    2 --
 1 file changed, 2 deletions(-)

diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index 215d2adadbdd..04148a656b2a 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -322,8 +322,6 @@ bool svc_rdma_post_recvs(struct svcxprt_rdma *rdma)
  * @cq: Completion Queue context
  * @wc: Work Completion object
  *
- * NB: The svc_xprt/svcxprt_rdma is pinned whenever it's possible that
- * the Receive completion handler could be running.
  */
 static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
 {

[PATCH v1 4/6] svcrdma: Add a batch Receive posting mechanism

[Chuck Lever]

Introduce a server-side mechanism similar to commit e340c2d6ef2a
("xprtrdma: Reduce the doorbell rate (Receive)") to post Receive
WRs in batch. It's first consumer is svc_rdma_post_recvs().

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 net/sunrpc/xprtrdma/svc_rdma_recvfrom.c |   56 +++++++++++++++++++++++--------
 1 file changed, 42 insertions(+), 14 deletions(-)

diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index 04148a656b2a..0c6aa8693f20 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -264,6 +264,47 @@ void svc_rdma_release_rqst(struct svc_rqst *rqstp)
 		svc_rdma_recv_ctxt_put(rdma, ctxt);
 }
 
+static bool svc_rdma_refresh_recvs(struct svcxprt_rdma *rdma,
+				   unsigned int wanted, bool temp)
+{
+	const struct ib_recv_wr *bad_wr = NULL;
+	struct svc_rdma_recv_ctxt *ctxt;
+	struct ib_recv_wr *recv_chain;
+	int ret;
+
+	if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags))
+		return false;
+
+	recv_chain = NULL;
+	while (wanted--) {
+		ctxt = svc_rdma_recv_ctxt_get(rdma);
+		if (!ctxt)
+			break;
+
+		trace_svcrdma_post_recv(ctxt);
+		ctxt->rc_temp = temp;
+		ctxt->rc_recv_wr.next = recv_chain;
+		recv_chain = &ctxt->rc_recv_wr;
+	}
+	if (!recv_chain)
+		return false;
+
+	ret = ib_post_recv(rdma->sc_qp, recv_chain, &bad_wr);
+	if (ret)
+		goto err_free;
+	return true;
+
+err_free:
+	trace_svcrdma_rq_post_err(rdma, ret);
+	while (bad_wr) {
+		ctxt = container_of(bad_wr, struct svc_rdma_recv_ctxt,
+				    rc_recv_wr);
+		bad_wr = bad_wr->next;
+		svc_rdma_recv_ctxt_put(rdma, ctxt);
+	}
+	return false;
+}
+
 static int __svc_rdma_post_recv(struct svcxprt_rdma *rdma,
 				struct svc_rdma_recv_ctxt *ctxt)
 {
@@ -301,20 +342,7 @@ static int svc_rdma_post_recv(struct svcxprt_rdma *rdma)
  */
 bool svc_rdma_post_recvs(struct svcxprt_rdma *rdma)
 {
-	struct svc_rdma_recv_ctxt *ctxt;
-	unsigned int i;
-	int ret;
-
-	for (i = 0; i < rdma->sc_max_requests; i++) {
-		ctxt = svc_rdma_recv_ctxt_get(rdma);
-		if (!ctxt)
-			return false;
-		ctxt->rc_temp = true;
-		ret = __svc_rdma_post_recv(rdma, ctxt);
-		if (ret)
-			return false;
-	}
-	return true;
+	return svc_rdma_refresh_recvs(rdma, rdma->sc_max_requests, true);
 }
 
 /**

Re: [PATCH v1 4/6] svcrdma: Add a batch Receive posting mechanism

[J. Bruce Fields]

On Fri, Mar 19, 2021 at 10:31:35AM -0400, Chuck Lever wrote:
> Introduce a server-side mechanism similar to commit e340c2d6ef2a
> ("xprtrdma: Reduce the doorbell rate (Receive)") to post Receive
> WRs in batch. It's first consumer is svc_rdma_post_recvs().

s/It's/Its'/.

Patches look OK to me.--b.

> 
> Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
> ---
>  net/sunrpc/xprtrdma/svc_rdma_recvfrom.c |   56 +++++++++++++++++++++++--------
>  1 file changed, 42 insertions(+), 14 deletions(-)
> 
> diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
> index 04148a656b2a..0c6aa8693f20 100644
> --- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
> +++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
> @@ -264,6 +264,47 @@ void svc_rdma_release_rqst(struct svc_rqst *rqstp)
>  		svc_rdma_recv_ctxt_put(rdma, ctxt);
>  }
>  
> +static bool svc_rdma_refresh_recvs(struct svcxprt_rdma *rdma,
> +				   unsigned int wanted, bool temp)
> +{
> +	const struct ib_recv_wr *bad_wr = NULL;
> +	struct svc_rdma_recv_ctxt *ctxt;
> +	struct ib_recv_wr *recv_chain;
> +	int ret;
> +
> +	if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags))
> +		return false;
> +
> +	recv_chain = NULL;
> +	while (wanted--) {
> +		ctxt = svc_rdma_recv_ctxt_get(rdma);
> +		if (!ctxt)
> +			break;
> +
> +		trace_svcrdma_post_recv(ctxt);
> +		ctxt->rc_temp = temp;
> +		ctxt->rc_recv_wr.next = recv_chain;
> +		recv_chain = &ctxt->rc_recv_wr;
> +	}
> +	if (!recv_chain)
> +		return false;
> +
> +	ret = ib_post_recv(rdma->sc_qp, recv_chain, &bad_wr);
> +	if (ret)
> +		goto err_free;
> +	return true;
> +
> +err_free:
> +	trace_svcrdma_rq_post_err(rdma, ret);
> +	while (bad_wr) {
> +		ctxt = container_of(bad_wr, struct svc_rdma_recv_ctxt,
> +				    rc_recv_wr);
> +		bad_wr = bad_wr->next;
> +		svc_rdma_recv_ctxt_put(rdma, ctxt);
> +	}
> +	return false;
> +}
> +
>  static int __svc_rdma_post_recv(struct svcxprt_rdma *rdma,
>  				struct svc_rdma_recv_ctxt *ctxt)
>  {
> @@ -301,20 +342,7 @@ static int svc_rdma_post_recv(struct svcxprt_rdma *rdma)
>   */
>  bool svc_rdma_post_recvs(struct svcxprt_rdma *rdma)
>  {
> -	struct svc_rdma_recv_ctxt *ctxt;
> -	unsigned int i;
> -	int ret;
> -
> -	for (i = 0; i < rdma->sc_max_requests; i++) {
> -		ctxt = svc_rdma_recv_ctxt_get(rdma);
> -		if (!ctxt)
> -			return false;
> -		ctxt->rc_temp = true;
> -		ret = __svc_rdma_post_recv(rdma, ctxt);
> -		if (ret)
> -			return false;
> -	}
> -	return true;
> +	return svc_rdma_refresh_recvs(rdma, rdma->sc_max_requests, true);
>  }
>  
>  /**
> 

Re: [PATCH v1 4/6] svcrdma: Add a batch Receive posting mechanism

[Chuck Lever III]

> On Mar 22, 2021, at 1:16 PM, J. Bruce Fields <bfields@fieldses.org> wrote:
> 
> On Fri, Mar 19, 2021 at 10:31:35AM -0400, Chuck Lever wrote:
>> Introduce a server-side mechanism similar to commit e340c2d6ef2a
>> ("xprtrdma: Reduce the doorbell rate (Receive)") to post Receive
>> WRs in batch. It's first consumer is svc_rdma_post_recvs().
> 
> s/It's/Its'/.

D'oh!


> Patches look OK to me.--b.

Thanks for the review!


>> Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
>> ---
>> net/sunrpc/xprtrdma/svc_rdma_recvfrom.c |   56 +++++++++++++++++++++++--------
>> 1 file changed, 42 insertions(+), 14 deletions(-)
>> 
>> diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
>> index 04148a656b2a..0c6aa8693f20 100644
>> --- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
>> +++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
>> @@ -264,6 +264,47 @@ void svc_rdma_release_rqst(struct svc_rqst *rqstp)
>> 		svc_rdma_recv_ctxt_put(rdma, ctxt);
>> }
>> 
>> +static bool svc_rdma_refresh_recvs(struct svcxprt_rdma *rdma,
>> +				   unsigned int wanted, bool temp)
>> +{
>> +	const struct ib_recv_wr *bad_wr = NULL;
>> +	struct svc_rdma_recv_ctxt *ctxt;
>> +	struct ib_recv_wr *recv_chain;
>> +	int ret;
>> +
>> +	if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags))
>> +		return false;
>> +
>> +	recv_chain = NULL;
>> +	while (wanted--) {
>> +		ctxt = svc_rdma_recv_ctxt_get(rdma);
>> +		if (!ctxt)
>> +			break;
>> +
>> +		trace_svcrdma_post_recv(ctxt);
>> +		ctxt->rc_temp = temp;
>> +		ctxt->rc_recv_wr.next = recv_chain;
>> +		recv_chain = &ctxt->rc_recv_wr;
>> +	}
>> +	if (!recv_chain)
>> +		return false;
>> +
>> +	ret = ib_post_recv(rdma->sc_qp, recv_chain, &bad_wr);
>> +	if (ret)
>> +		goto err_free;
>> +	return true;
>> +
>> +err_free:
>> +	trace_svcrdma_rq_post_err(rdma, ret);
>> +	while (bad_wr) {
>> +		ctxt = container_of(bad_wr, struct svc_rdma_recv_ctxt,
>> +				    rc_recv_wr);
>> +		bad_wr = bad_wr->next;
>> +		svc_rdma_recv_ctxt_put(rdma, ctxt);
>> +	}
>> +	return false;
>> +}
>> +
>> static int __svc_rdma_post_recv(struct svcxprt_rdma *rdma,
>> 				struct svc_rdma_recv_ctxt *ctxt)
>> {
>> @@ -301,20 +342,7 @@ static int svc_rdma_post_recv(struct svcxprt_rdma *rdma)
>>  */
>> bool svc_rdma_post_recvs(struct svcxprt_rdma *rdma)
>> {
>> -	struct svc_rdma_recv_ctxt *ctxt;
>> -	unsigned int i;
>> -	int ret;
>> -
>> -	for (i = 0; i < rdma->sc_max_requests; i++) {
>> -		ctxt = svc_rdma_recv_ctxt_get(rdma);
>> -		if (!ctxt)
>> -			return false;
>> -		ctxt->rc_temp = true;
>> -		ret = __svc_rdma_post_recv(rdma, ctxt);
>> -		if (ret)
>> -			return false;
>> -	}
>> -	return true;
>> +	return svc_rdma_refresh_recvs(rdma, rdma->sc_max_requests, true);
>> }
>> 
>> /**
>> 

--
Chuck Lever

Re: [PATCH v1 4/6] svcrdma: Add a batch Receive posting mechanism

[Daniel Forrest]

On Mon, Mar 22, 2021 at 05:17:35PM +0000, Chuck Lever III wrote:
> 
> 
> > On Mar 22, 2021, at 1:16 PM, J. Bruce Fields <bfields@fieldses.org> wrote:
> > 
> > On Fri, Mar 19, 2021 at 10:31:35AM -0400, Chuck Lever wrote:
> >> Introduce a server-side mechanism similar to commit e340c2d6ef2a
> >> ("xprtrdma: Reduce the doorbell rate (Receive)") to post Receive
> >> WRs in batch. It's first consumer is svc_rdma_post_recvs().
> > 
> > s/It's/Its'/.
> 
> D'oh!
> 

Except there should be no apostrophe at all, just plain "Its".

> 
> > Patches look OK to me.--b.
> 
> Thanks for the review!
> 
> 
> >> Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
> >> ---
> >> net/sunrpc/xprtrdma/svc_rdma_recvfrom.c |   56 +++++++++++++++++++++++--------
> >> 1 file changed, 42 insertions(+), 14 deletions(-)
> >> 
> >> diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
> >> index 04148a656b2a..0c6aa8693f20 100644
> >> --- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
> >> +++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
> >> @@ -264,6 +264,47 @@ void svc_rdma_release_rqst(struct svc_rqst *rqstp)
> >> 		svc_rdma_recv_ctxt_put(rdma, ctxt);
> >> }
> >> 
> >> +static bool svc_rdma_refresh_recvs(struct svcxprt_rdma *rdma,
> >> +				   unsigned int wanted, bool temp)
> >> +{
> >> +	const struct ib_recv_wr *bad_wr = NULL;
> >> +	struct svc_rdma_recv_ctxt *ctxt;
> >> +	struct ib_recv_wr *recv_chain;
> >> +	int ret;
> >> +
> >> +	if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags))
> >> +		return false;
> >> +
> >> +	recv_chain = NULL;
> >> +	while (wanted--) {
> >> +		ctxt = svc_rdma_recv_ctxt_get(rdma);
> >> +		if (!ctxt)
> >> +			break;
> >> +
> >> +		trace_svcrdma_post_recv(ctxt);
> >> +		ctxt->rc_temp = temp;
> >> +		ctxt->rc_recv_wr.next = recv_chain;
> >> +		recv_chain = &ctxt->rc_recv_wr;
> >> +	}
> >> +	if (!recv_chain)
> >> +		return false;
> >> +
> >> +	ret = ib_post_recv(rdma->sc_qp, recv_chain, &bad_wr);
> >> +	if (ret)
> >> +		goto err_free;
> >> +	return true;
> >> +
> >> +err_free:
> >> +	trace_svcrdma_rq_post_err(rdma, ret);
> >> +	while (bad_wr) {
> >> +		ctxt = container_of(bad_wr, struct svc_rdma_recv_ctxt,
> >> +				    rc_recv_wr);
> >> +		bad_wr = bad_wr->next;
> >> +		svc_rdma_recv_ctxt_put(rdma, ctxt);
> >> +	}
> >> +	return false;
> >> +}
> >> +
> >> static int __svc_rdma_post_recv(struct svcxprt_rdma *rdma,
> >> 				struct svc_rdma_recv_ctxt *ctxt)
> >> {
> >> @@ -301,20 +342,7 @@ static int svc_rdma_post_recv(struct svcxprt_rdma *rdma)
> >>  */
> >> bool svc_rdma_post_recvs(struct svcxprt_rdma *rdma)
> >> {
> >> -	struct svc_rdma_recv_ctxt *ctxt;
> >> -	unsigned int i;
> >> -	int ret;
> >> -
> >> -	for (i = 0; i < rdma->sc_max_requests; i++) {
> >> -		ctxt = svc_rdma_recv_ctxt_get(rdma);
> >> -		if (!ctxt)
> >> -			return false;
> >> -		ctxt->rc_temp = true;
> >> -		ret = __svc_rdma_post_recv(rdma, ctxt);
> >> -		if (ret)
> >> -			return false;
> >> -	}
> >> -	return true;
> >> +	return svc_rdma_refresh_recvs(rdma, rdma->sc_max_requests, true);
> >> }
> >> 
> >> /**
> >> 
> 
> --
> Chuck Lever

-- 
Dan

[PATCH v1 5/6] svcrdma: Use svc_rdma_refresh_recvs() in wc_receive

[Chuck Lever]

Replace svc_rdma_post_recv() with the new batch receive mechanism.
For the moment it is posting just a single Receive WR at a time,
so no change in behavior is expected.

Since svc_rdma_wc_receive() was the last call site for
svc_rdma_post_recv(), it is removed.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 net/sunrpc/xprtrdma/svc_rdma_recvfrom.c |   43 ++++++++-----------------------
 1 file changed, 11 insertions(+), 32 deletions(-)

diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index 0c6aa8693f20..1e7381ff948b 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -305,35 +305,6 @@ static bool svc_rdma_refresh_recvs(struct svcxprt_rdma *rdma,
 	return false;
 }
 
-static int __svc_rdma_post_recv(struct svcxprt_rdma *rdma,
-				struct svc_rdma_recv_ctxt *ctxt)
-{
-	int ret;
-
-	trace_svcrdma_post_recv(ctxt);
-	ret = ib_post_recv(rdma->sc_qp, &ctxt->rc_recv_wr, NULL);
-	if (ret)
-		goto err_post;
-	return 0;
-
-err_post:
-	trace_svcrdma_rq_post_err(rdma, ret);
-	svc_rdma_recv_ctxt_put(rdma, ctxt);
-	return ret;
-}
-
-static int svc_rdma_post_recv(struct svcxprt_rdma *rdma)
-{
-	struct svc_rdma_recv_ctxt *ctxt;
-
-	if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags))
-		return 0;
-	ctxt = svc_rdma_recv_ctxt_get(rdma);
-	if (!ctxt)
-		return -ENOMEM;
-	return __svc_rdma_post_recv(rdma, ctxt);
-}
-
 /**
  * svc_rdma_post_recvs - Post initial set of Recv WRs
  * @rdma: fresh svcxprt_rdma
@@ -364,8 +335,17 @@ static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
 	if (wc->status != IB_WC_SUCCESS)
 		goto flushed;
 
-	if (svc_rdma_post_recv(rdma))
-		goto post_err;
+	/* If receive posting fails, the connection is about to be
+	 * lost anyway. The server will not be able to send a reply
+	 * for this RPC, and the client will retransmit this RPC
+	 * anyway when it reconnects.
+	 *
+	 * Therefore we drop the Receive, even if status was SUCCESS
+	 * to reduce the likelihood of replayed requests once the
+	 * client reconnects.
+	 */
+	if (!svc_rdma_refresh_recvs(rdma, 1, false))
+		goto flushed;
 
 	/* All wc fields are now known to be valid */
 	ctxt->rc_byte_len = wc->byte_len;
@@ -380,7 +360,6 @@ static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
 	return;
 
 flushed:
-post_err:
 	svc_rdma_recv_ctxt_put(rdma, ctxt);
 	set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags);
 	svc_xprt_enqueue(&rdma->sc_xprt);

[PATCH v1 6/6] svcrdma: Maintain a Receive water mark

[Chuck Lever]

Post more Receives when the number of pending Receives drops below
a water mark. The batch mechanism is disabled if the underlying
device cannot support a reasonably-sized Receive Queue.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
 include/linux/sunrpc/svc_rdma.h          |    2 ++
 net/sunrpc/xprtrdma/svc_rdma_recvfrom.c  |   10 ++++++++--
 net/sunrpc/xprtrdma/svc_rdma_transport.c |    5 ++++-
 3 files changed, 14 insertions(+), 3 deletions(-)

diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 1e76ed688044..722fc7c48725 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -94,6 +94,8 @@ struct svcxprt_rdma {
 	spinlock_t	     sc_rw_ctxt_lock;
 	struct list_head     sc_rw_ctxts;
 
+	u32		     sc_pending_recvs;
+	u32		     sc_recv_batch;
 	struct list_head     sc_rq_dto_q;
 	spinlock_t	     sc_rq_dto_lock;
 	struct ib_qp         *sc_qp;
diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index 1e7381ff948b..2571188ef7f2 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -285,6 +285,7 @@ static bool svc_rdma_refresh_recvs(struct svcxprt_rdma *rdma,
 		ctxt->rc_temp = temp;
 		ctxt->rc_recv_wr.next = recv_chain;
 		recv_chain = &ctxt->rc_recv_wr;
+		rdma->sc_pending_recvs++;
 	}
 	if (!recv_chain)
 		return false;
@@ -302,6 +303,8 @@ static bool svc_rdma_refresh_recvs(struct svcxprt_rdma *rdma,
 		bad_wr = bad_wr->next;
 		svc_rdma_recv_ctxt_put(rdma, ctxt);
 	}
+	/* Since we're destroying the xprt, no need to reset
+	 * sc_pending_recvs. */
 	return false;
 }
 
@@ -328,6 +331,8 @@ static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
 	struct ib_cqe *cqe = wc->wr_cqe;
 	struct svc_rdma_recv_ctxt *ctxt;
 
+	rdma->sc_pending_recvs--;
+
 	/* WARNING: Only wc->wr_cqe and wc->status are reliable */
 	ctxt = container_of(cqe, struct svc_rdma_recv_ctxt, rc_cqe);
 
@@ -344,8 +349,9 @@ static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
 	 * to reduce the likelihood of replayed requests once the
 	 * client reconnects.
 	 */
-	if (!svc_rdma_refresh_recvs(rdma, 1, false))
-		goto flushed;
+	if (rdma->sc_pending_recvs < rdma->sc_max_requests)
+		if (!svc_rdma_refresh_recvs(rdma, rdma->sc_recv_batch, false))
+			goto flushed;
 
 	/* All wc fields are now known to be valid */
 	ctxt->rc_byte_len = wc->byte_len;
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index 046a07da5cf9..e629eacfedfc 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -407,11 +407,14 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
 	newxprt->sc_max_req_size = svcrdma_max_req_size;
 	newxprt->sc_max_requests = svcrdma_max_requests;
 	newxprt->sc_max_bc_requests = svcrdma_max_bc_requests;
-	rq_depth = newxprt->sc_max_requests + newxprt->sc_max_bc_requests;
+	newxprt->sc_recv_batch = RPCRDMA_MAX_RECV_BATCH;
+	rq_depth = newxprt->sc_max_requests + newxprt->sc_max_bc_requests +
+		   newxprt->sc_recv_batch;
 	if (rq_depth > dev->attrs.max_qp_wr) {
 		pr_warn("svcrdma: reducing receive depth to %d\n",
 			dev->attrs.max_qp_wr);
 		rq_depth = dev->attrs.max_qp_wr;
+		newxprt->sc_recv_batch = 1;
 		newxprt->sc_max_requests = rq_depth - 2;
 		newxprt->sc_max_bc_requests = 2;
 	}