Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit d8f532d2 authored by Chuck Lever's avatar Chuck Lever Committed by Anna Schumaker
Browse files

xprtrdma: Invoke rpcrdma_reply_handler directly from RECV completion 



I noticed that the soft IRQ thread looked pretty busy under heavy
I/O workloads. perf suggested one area that was expensive was the
queue_work() call in rpcrdma_wc_receive. That gave me some ideas.

Instead of scheduling a separate worker to process RPC Replies,
promote the Receive completion handler to IB_POLL_WORKQUEUE, and
invoke rpcrdma_reply_handler directly.

Note that the poll workqueue is single-threaded. In order to keep
memory invalidation from serializing all RPC Replies, handle any
necessary invalidation tasks in a separate multi-threaded workqueue.

This provides a two-tier scheme, similar to OS I/O interrupt
handlers: A fast interrupt handler that schedules the slow handler
and re-enables the interrupt, and a slower handler that is invoked
for any needed heavy lifting.

Benefits include:
- One less context switch for RPCs that don't register memory
- Receive completion handling is moved out of soft IRQ context to
  make room for other users of soft IRQ
- The same CPU core now DMA syncs and XDR decodes the Receive buffer

Signed-off-by: default avatarChuck Lever <chuck.lever@oracle.com>
Signed-off-by: default avatarAnna Schumaker <Anna.Schumaker@Netapp.com>
parent e1352c96

net/sunrpc/xprtrdma/rpc_rdma.c

+28 −18
Original line number Diff line number Diff line
@@ -1265,16 +1265,36 @@ void rpcrdma_complete_rqst(struct rpcrdma_rep *rep) 
	goto out;

}



/* Reply handling runs in the poll worker thread. Anything that

 * might wait is deferred to a separate workqueue.

 */

void rpcrdma_deferred_completion(struct work_struct *work)

{

	struct rpcrdma_rep *rep =

			container_of(work, struct rpcrdma_rep, rr_work);

	struct rpcrdma_req *req = rpcr_to_rdmar(rep->rr_rqst);

	struct rpcrdma_xprt *r_xprt = rep->rr_rxprt;



	/* Invalidate and unmap the data payloads before waking

	 * the waiting application. This guarantees the memory

	 * regions are properly fenced from the server before the

	 * application accesses the data. It also ensures proper

	 * send flow control: waking the next RPC waits until this

	 * RPC has relinquished all its Send Queue entries.

	 */

	rpcrdma_mark_remote_invalidation(&req->rl_registered, rep);

	r_xprt->rx_ia.ri_ops->ro_unmap_sync(r_xprt, &req->rl_registered);



	rpcrdma_complete_rqst(rep);

}



/* Process received RPC/RDMA messages.

 *

 * Errors must result in the RPC task either being awakened, or

 * allowed to timeout, to discover the errors at that time.

 */

void

rpcrdma_reply_handler(struct work_struct *work)

void rpcrdma_reply_handler(struct rpcrdma_rep *rep)

{

	struct rpcrdma_rep *rep =

			container_of(work, struct rpcrdma_rep, rr_work);

	struct rpcrdma_xprt *r_xprt = rep->rr_rxprt;

	struct rpc_xprt *xprt = &r_xprt->rx_xprt;

	struct rpcrdma_req *req;
@@ -1320,20 +1340,10 @@ rpcrdma_reply_handler(struct work_struct *work) 
	dprintk("RPC:       %s: reply %p completes request %p (xid 0x%08x)\n",

		__func__, rep, req, be32_to_cpu(rep->rr_xid));



	/* Invalidate and unmap the data payloads before waking the

	 * waiting application. This guarantees the memory regions

	 * are properly fenced from the server before the application

	 * accesses the data. It also ensures proper send flow control:

	 * waking the next RPC waits until this RPC has relinquished

	 * all its Send Queue entries.

	 */

	if (!list_empty(&req->rl_registered)) {

		rpcrdma_mark_remote_invalidation(&req->rl_registered, rep);

		r_xprt->rx_ia.ri_ops->ro_unmap_sync(r_xprt,

						    &req->rl_registered);

	}



	if (list_empty(&req->rl_registered))

		rpcrdma_complete_rqst(rep);

	else

		queue_work(rpcrdma_receive_wq, &rep->rr_work);

	return;



out_badstatus:

net/sunrpc/xprtrdma/verbs.c

+4 −4
Original line number Diff line number Diff line
@@ -73,7 +73,7 @@ static void rpcrdma_create_mrs(struct rpcrdma_xprt *r_xprt); 
static void rpcrdma_destroy_mrs(struct rpcrdma_buffer *buf);

static void rpcrdma_dma_unmap_regbuf(struct rpcrdma_regbuf *rb);



static struct workqueue_struct *rpcrdma_receive_wq __read_mostly;

struct workqueue_struct *rpcrdma_receive_wq __read_mostly;



int

rpcrdma_alloc_wq(void)
@@ -185,7 +185,7 @@ rpcrdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc) 
		rpcrdma_update_granted_credits(rep);



out_schedule:

	queue_work(rpcrdma_receive_wq, &rep->rr_work);

	rpcrdma_reply_handler(rep);

	return;



out_fail:
@@ -583,7 +583,7 @@ rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia, 


	recvcq = ib_alloc_cq(ia->ri_device, NULL,

			     ep->rep_attr.cap.max_recv_wr + 1,

			     0, IB_POLL_SOFTIRQ);

			     0, IB_POLL_WORKQUEUE);

	if (IS_ERR(recvcq)) {

		rc = PTR_ERR(recvcq);

		dprintk("RPC:       %s: failed to create recv CQ: %i\n",
@@ -974,7 +974,7 @@ rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt) 


	rep->rr_cqe.done = rpcrdma_wc_receive;

	rep->rr_rxprt = r_xprt;

	INIT_WORK(&rep->rr_work, rpcrdma_reply_handler);

	INIT_WORK(&rep->rr_work, rpcrdma_deferred_completion);

	rep->rr_recv_wr.next = NULL;

	rep->rr_recv_wr.wr_cqe = &rep->rr_cqe;

	rep->rr_recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov;

net/sunrpc/xprtrdma/xprt_rdma.h

+4 −1
Original line number Diff line number Diff line
@@ -533,6 +533,8 @@ void rpcrdma_ia_close(struct rpcrdma_ia *); 
bool frwr_is_supported(struct rpcrdma_ia *);

bool fmr_is_supported(struct rpcrdma_ia *);



extern struct workqueue_struct *rpcrdma_receive_wq;



/*

 * Endpoint calls - xprtrdma/verbs.c

 */
@@ -617,7 +619,8 @@ void rpcrdma_unmap_sges(struct rpcrdma_ia *, struct rpcrdma_req *); 
int rpcrdma_marshal_req(struct rpcrdma_xprt *r_xprt, struct rpc_rqst *rqst);

void rpcrdma_set_max_header_sizes(struct rpcrdma_xprt *);

void rpcrdma_complete_rqst(struct rpcrdma_rep *rep);

void rpcrdma_reply_handler(struct work_struct *work);

void rpcrdma_reply_handler(struct rpcrdma_rep *rep);

void rpcrdma_deferred_completion(struct work_struct *work);



static inline void rpcrdma_set_xdrlen(struct xdr_buf *xdr, size_t len)

{