lael-labs/linux

include/linux/sunrpc/rpc_rdma.h+1−0

@@ -1,3 +1,4 @@
		1	/* SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause */
1	/*	2	/*
2	* Copyright (c) 2015-2017 Oracle. All rights reserved.	3	* Copyright (c) 2015-2017 Oracle. All rights reserved.
3	* Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.	4	* Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.

include/linux/sunrpc/xprt.h+5−1

 	void (*rq_release_snd_buf)(struct rpc_rqst *); /* release rq_enc_pages */
 	struct list_head	rq_list;
-	void			*rq_xprtdata;	/* Per-xprt private data */
 	void			*rq_buffer;	/* Call XDR encode buffer */
 	size_t			rq_callsize;
 	void			*rq_rbuffer;	/* Reply XDR decode buffer */
 	int		(*reserve_xprt)(struct rpc_xprt *xprt, struct rpc_task *task);
 	void		(*release_xprt)(struct rpc_xprt *xprt, struct rpc_task *task);
 	void		(*alloc_slot)(struct rpc_xprt *xprt, struct rpc_task *task);
+	void		(*free_slot)(struct rpc_xprt *xprt,
+				     struct rpc_rqst *req);
 	void		(*rpcbind)(struct rpc_task *task);
 	void		(*set_port)(struct rpc_xprt *xprt, unsigned short port);
 	void		(*connect)(struct rpc_xprt *xprt, struct rpc_task *task);
 struct rpc_xprt		*xprt_create_transport(struct xprt_create *args);
 void			xprt_connect(struct rpc_task *task);
 void			xprt_reserve(struct rpc_task *task);
+void			xprt_request_init(struct rpc_task *task);
 void			xprt_retry_reserve(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);
 void			xprt_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task);
+void			xprt_free_slot(struct rpc_xprt *xprt,
+				       struct rpc_rqst *req);
 void			xprt_lock_and_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task);
 bool			xprt_prepare_transmit(struct rpc_task *task);
 void			xprt_transmit(struct rpc_task *task);

include/linux/sunrpc/xprtrdma.h+1−0

@@ -1,3 +1,4 @@
		1	/* SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause */
1	/*	2	/*
2	* Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.	3	* Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
3	*	4	*

include/trace/events/rpcrdma.h+53−23

 TRACE_EVENT(xprtrdma_post_recv,
 	TP_PROTO(
-		const struct rpcrdma_rep *rep,
+		const struct ib_cqe *cqe
+	),
+	TP_ARGS(cqe),
+	TP_STRUCT__entry(
+		__field(const void *, cqe)
+	),
+	TP_fast_assign(
+		__entry->cqe = cqe;
+	),
+	TP_printk("cqe=%p",
+		__entry->cqe
+	)
+);
+TRACE_EVENT(xprtrdma_post_recvs,
+	TP_PROTO(
+		const struct rpcrdma_xprt *r_xprt,
+		unsigned int count,
 		int status
 	),
-	TP_ARGS(rep, status),
+	TP_ARGS(r_xprt, count, status),
 	TP_STRUCT__entry(
-		__field(const void *, rep)
+		__field(const void *, r_xprt)
+		__field(unsigned int, count)
 		__field(int, status)
+		__field(int, posted)
+		__string(addr, rpcrdma_addrstr(r_xprt))
+		__string(port, rpcrdma_portstr(r_xprt))
 	),
 	TP_fast_assign(
-		__entry->rep = rep;
+		__entry->r_xprt = r_xprt;
+		__entry->count = count;
 		__entry->status = status;
+		__entry->posted = r_xprt->rx_buf.rb_posted_receives;
+		__assign_str(addr, rpcrdma_addrstr(r_xprt));
+		__assign_str(port, rpcrdma_portstr(r_xprt));
 	),
-	TP_printk("rep=%p status=%d",
+	TP_printk("peer=[%s]:%s r_xprt=%p: %u new recvs, %d active (rc %d)",
-		__entry->rep, __entry->status
+		__get_str(addr), __get_str(port), __entry->r_xprt,
+		__entry->count, __entry->posted, __entry->status
 	)
 );
 TRACE_EVENT(xprtrdma_wc_receive,
 	TP_PROTO(
-		const struct rpcrdma_rep *rep,
 		const struct ib_wc *wc
 	),
-	TP_ARGS(rep, wc),
+	TP_ARGS(wc),
 	TP_STRUCT__entry(
-		__field(const void *, rep)
+		__field(const void *, cqe)
-		__field(unsigned int, byte_len)
+		__field(u32, byte_len)
 		__field(unsigned int, status)
-		__field(unsigned int, vendor_err)
+		__field(u32, vendor_err)
 	),
 	TP_fast_assign(
-		__entry->rep = rep;
+		__entry->cqe = wc->wr_cqe;
-		__entry->byte_len = wc->byte_len;
 		__entry->status = wc->status;
-		__entry->vendor_err = __entry->status ? wc->vendor_err : 0;
+		if (wc->status) {
+			__entry->byte_len = 0;
+			__entry->vendor_err = wc->vendor_err;
+		} else {
+			__entry->byte_len = wc->byte_len;
+			__entry->vendor_err = 0;
+		}
 	),
-	TP_printk("rep=%p, %u bytes: %s (%u/0x%x)",
+	TP_printk("cqe=%p %u bytes: %s (%u/0x%x)",
-		__entry->rep, __entry->byte_len,
+		__entry->cqe, __entry->byte_len,
 		rdma_show_wc_status(__entry->status),
 		__entry->status, __entry->vendor_err
 	)
 DEFINE_FRWR_DONE_EVENT(xprtrdma_wc_li_wake);
 DEFINE_MR_EVENT(xprtrdma_localinv);
+DEFINE_MR_EVENT(xprtrdma_dma_map);
 DEFINE_MR_EVENT(xprtrdma_dma_unmap);
 DEFINE_MR_EVENT(xprtrdma_remoteinv);
 DEFINE_MR_EVENT(xprtrdma_recover_mr);
 		__field(unsigned int, task_id)
 		__field(unsigned int, client_id)
 		__field(const void *, req)
-		__field(const void *, rep)
 		__field(size_t, callsize)
 		__field(size_t, rcvsize)
 	),
 		__entry->task_id = task->tk_pid;
 		__entry->client_id = task->tk_client->cl_clid;
 		__entry->req = req;
-		__entry->rep = req ? req->rl_reply : NULL;
 		__entry->callsize = task->tk_rqstp->rq_callsize;
 		__entry->rcvsize = task->tk_rqstp->rq_rcvsize;
 	),
-	TP_printk("task:%u@%u req=%p rep=%p (%zu, %zu)",
+	TP_printk("task:%u@%u req=%p (%zu, %zu)",
 		__entry->task_id, __entry->client_id,
-		__entry->req, __entry->rep,
+		__entry->req, __entry->callsize, __entry->rcvsize
-		__entry->callsize, __entry->rcvsize
 	)
 );
 	)
 );
-DEFINE_RXPRT_EVENT(xprtrdma_noreps);
 /**
  ** Callback events
  **/

net/sunrpc/clnt.c+1−0

@@ -1546,6 +1546,7 @@ call_reserveresult(struct rpc_task *task) · 1545 unmodified lines
1546	task->tk_status = 0;	1546	task->tk_status = 0;
1547	if (status >= 0) {	1547	if (status >= 0) {
1548	if (task->tk_rqstp) {	1548	if (task->tk_rqstp) {
		1549	xprt_request_init(task);
1549	task->tk_action = call_refresh;	1550	task->tk_action = call_refresh;
1550	return;	1551	return;
1551	}	1552	}

net/sunrpc/xprt.c+10−7

  * Local functions
  */
 static void	 xprt_init(struct rpc_xprt *xprt, struct net *net);
-static void	xprt_request_init(struct rpc_task *, struct rpc_xprt *);
+static __be32	xprt_alloc_xid(struct rpc_xprt *xprt);
 static void	xprt_connect_status(struct rpc_task *task);
 static int      __xprt_get_cong(struct rpc_xprt *, struct rpc_task *);
 static void     __xprt_put_cong(struct rpc_xprt *, struct rpc_rqst *);
 		task->tk_status = -EAGAIN;
 		goto out_unlock;
 	}
+	if (!bc_prealloc(req) && !req->rq_xmit_bytes_sent)
+		req->rq_xid = xprt_alloc_xid(xprt);
 	ret = true;
 out_unlock:
 	spin_unlock_bh(&xprt->transport_lock);
 out_init_req:
 	xprt->stat.max_slots = max_t(unsigned int, xprt->stat.max_slots,
 				     xprt->num_reqs);
+	spin_unlock(&xprt->reserve_lock);
 	task->tk_status = 0;
 	task->tk_rqstp = req;
-	xprt_request_init(task, xprt);
-	spin_unlock(&xprt->reserve_lock);
 }
 EXPORT_SYMBOL_GPL(xprt_alloc_slot);
 }
 EXPORT_SYMBOL_GPL(xprt_lock_and_alloc_slot);
-static void xprt_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
+void xprt_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
 {
 	spin_lock(&xprt->reserve_lock);
 	if (!xprt_dynamic_free_slot(xprt, req)) {
 	xprt_wake_up_backlog(xprt);
 	spin_unlock(&xprt->reserve_lock);
 }
+EXPORT_SYMBOL_GPL(xprt_free_slot);
 static void xprt_free_all_slots(struct rpc_xprt *xprt)
 {
 	xprt->xid = prandom_u32();
 }
-static void xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
+void xprt_request_init(struct rpc_task *task)
 {
+	struct rpc_xprt *xprt = task->tk_xprt;
 	struct rpc_rqst	*req = task->tk_rqstp;
 	INIT_LIST_HEAD(&req->rq_list);
 	req->rq_task	= task;
 	req->rq_xprt    = xprt;
 	req->rq_buffer  = NULL;
-	req->rq_xid     = xprt_alloc_xid(xprt);
 	req->rq_connect_cookie = xprt->connect_cookie - 1;
 	req->rq_bytes_sent = 0;
 	req->rq_snd_buf.len = 0;
 	dprintk("RPC: %5u release request %p\n", task->tk_pid, req);
 	if (likely(!bc_prealloc(req)))
-		xprt_free_slot(xprt, req);
+		xprt->ops->free_slot(xprt, req);
 	else
 		xprt_free_bc_request(req);
 }

net/sunrpc/xprtrdma/backchannel.c+39−66

@@ -29,29 +29,41 @@ static void rpcrdma_bc_free_rqst(struct rpcrdma_xprt *r_xprt, · 28 unmodified lines
29	spin_unlock(&buf->rb_reqslock);	29	spin_unlock(&buf->rb_reqslock);
30		30
31	rpcrdma_destroy_req(req);	31	rpcrdma_destroy_req(req);
32
33	kfree(rqst);
34	}	32	}
35		33
36	static int rpcrdma_bc_setup_rqst(struct rpcrdma_xprt *r_xprt,	34	static int rpcrdma_bc_setup_reqs(struct rpcrdma_xprt *r_xprt,
37	struct rpc_rqst *rqst)	35	unsigned int count)
38	{	36	{
39	struct rpcrdma_regbuf *rb;	37	struct rpc_xprt *xprt = &r_xprt->rx_xprt;
40	struct rpcrdma_req *req;	38	struct rpc_rqst *rqst;
41	size_t size;	39	unsigned int i;
		40
		41	for (i = 0; i < (count << 1); i++) {
		42	struct rpcrdma_regbuf *rb;
		43	struct rpcrdma_req *req;
		44	size_t size;
		45
		46	req = rpcrdma_create_req(r_xprt);
		47	if (IS_ERR(req))
		48	return PTR_ERR(req);
		49	rqst = &req->rl_slot;
		50
		51	rqst->rq_xprt = xprt;
		52	INIT_LIST_HEAD(&rqst->rq_list);
		53	INIT_LIST_HEAD(&rqst->rq_bc_list);
		54	__set_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state);
		55	spin_lock_bh(&xprt->bc_pa_lock);
		56	list_add(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);
		57	spin_unlock_bh(&xprt->bc_pa_lock);
42		58
43	req = rpcrdma_create_req(r_xprt);	59	size = r_xprt->rx_data.inline_rsize;
44	if (IS_ERR(req))	60	rb = rpcrdma_alloc_regbuf(size, DMA_TO_DEVICE, GFP_KERNEL);
45	return PTR_ERR(req);	61	if (IS_ERR(rb))
46		62	goto out_fail;
47	size = r_xprt->rx_data.inline_rsize;	63	req->rl_sendbuf = rb;
48	rb = rpcrdma_alloc_regbuf(size, DMA_TO_DEVICE, GFP_KERNEL);	64	xdr_buf_init(&rqst->rq_snd_buf, rb->rg_base,
49	if (IS_ERR(rb))	65	min_t(size_t, size, PAGE_SIZE));
50	goto out_fail;	66	}
51	req->rl_sendbuf = rb;
52	xdr_buf_init(&rqst->rq_snd_buf, rb->rg_base,
53	min_t(size_t, size, PAGE_SIZE));
54	rpcrdma_set_xprtdata(rqst, req);
55	return 0;	67	return 0;
56		68
57	out_fail:	69	out_fail:
@@ -59,23 +71,6 @@ out_fail: · 1 unmodified line
59	return -ENOMEM;	71	return -ENOMEM;
60	}	72	}
61		73
62	/* Allocate and add receive buffers to the rpcrdma_buffer's
63	* existing list of rep's. These are released when the
64	* transport is destroyed.
65	*/
66	static int rpcrdma_bc_setup_reps(struct rpcrdma_xprt *r_xprt,
67	unsigned int count)
68	{
69	int rc = 0;
70
71	while (count--) {
72	rc = rpcrdma_create_rep(r_xprt);
73	if (rc)
74	break;
75	}
76	return rc;
77	}
78
79	/**	74	/**
80	* xprt_rdma_bc_setup - Pre-allocate resources for handling backchannel requests	75	* xprt_rdma_bc_setup - Pre-allocate resources for handling backchannel requests
81	* @xprt: transport associated with these backchannel resources	76	* @xprt: transport associated with these backchannel resources
@@ -86,9 +81,6 @@ static int rpcrdma_bc_setup_reps(struct rpcrdma_xprt *r_xprt, · 4 unmodified lines
86	int xprt_rdma_bc_setup(struct rpc_xprt *xprt, unsigned int reqs)	81	int xprt_rdma_bc_setup(struct rpc_xprt *xprt, unsigned int reqs)
87	{	82	{
88	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);	83	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
89	struct rpcrdma_buffer *buffer = &r_xprt->rx_buf;
90	struct rpc_rqst *rqst;
91	unsigned int i;
92	int rc;	84	int rc;
93		85
94	/* The backchannel reply path returns each rpc_rqst to the	86	/* The backchannel reply path returns each rpc_rqst to the
@@ -103,35 +95,11 @@ int xprt_rdma_bc_setup(struct rpc_xprt *xprt, unsigned int reqs) · 8 unmodified lines
103	if (reqs > RPCRDMA_BACKWARD_WRS >> 1)	95	if (reqs > RPCRDMA_BACKWARD_WRS >> 1)
104	goto out_err;	96	goto out_err;
105		97
106	for (i = 0; i < (reqs << 1); i++) {	98	rc = rpcrdma_bc_setup_reqs(r_xprt, reqs);
107	rqst = kzalloc(sizeof(*rqst), GFP_KERNEL);
108	if (!rqst)
109	goto out_free;
110
111	dprintk("RPC: %s: new rqst %p\n", __func__, rqst);
112
113	rqst->rq_xprt = &r_xprt->rx_xprt;
114	INIT_LIST_HEAD(&rqst->rq_list);
115	INIT_LIST_HEAD(&rqst->rq_bc_list);
116	__set_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state);
117
118	if (rpcrdma_bc_setup_rqst(r_xprt, rqst))
119	goto out_free;
120
121	spin_lock_bh(&xprt->bc_pa_lock);
122	list_add(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);
123	spin_unlock_bh(&xprt->bc_pa_lock);
124	}
125
126	rc = rpcrdma_bc_setup_reps(r_xprt, reqs);
127	if (rc)	99	if (rc)
128	goto out_free;	100	goto out_free;
129		101
130	rc = rpcrdma_ep_post_extra_recv(r_xprt, reqs);	102	r_xprt->rx_buf.rb_bc_srv_max_requests = reqs;
131	if (rc)
132	goto out_free;
133
134	buffer->rb_bc_srv_max_requests = reqs;
135	request_module("svcrdma");	103	request_module("svcrdma");
136	trace_xprtrdma_cb_setup(r_xprt, reqs);	104	trace_xprtrdma_cb_setup(r_xprt, reqs);
137	return 0;	105	return 0;
@@ -235,6 +203,7 @@ int xprt_rdma_bc_send_reply(struct rpc_rqst *rqst) · 97 unmodified lines
235	if (rc < 0)	203	if (rc < 0)
236	goto failed_marshal;	204	goto failed_marshal;
237		205
		206	rpcrdma_post_recvs(r_xprt, true);
238	if (rpcrdma_ep_post(&r_xprt->rx_ia, &r_xprt->rx_ep, req))	207	if (rpcrdma_ep_post(&r_xprt->rx_ia, &r_xprt->rx_ep, req))
239	goto drop_connection;	208	goto drop_connection;
240	return 0;	209	return 0;
@@ -275,10 +244,14 @@ void xprt_rdma_bc_destroy(struct rpc_xprt *xprt, unsigned int reqs) · 34 unmodified lines
275	*/	244	*/
276	void xprt_rdma_bc_free_rqst(struct rpc_rqst *rqst)	245	void xprt_rdma_bc_free_rqst(struct rpc_rqst *rqst)
277	{	246	{
		247	struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
278	struct rpc_xprt *xprt = rqst->rq_xprt;	248	struct rpc_xprt *xprt = rqst->rq_xprt;
279		249
280	dprintk("RPC: %s: freeing rqst %p (req %p)\n",	250	dprintk("RPC: %s: freeing rqst %p (req %p)\n",
281	__func__, rqst, rpcr_to_rdmar(rqst));	251	__func__, rqst, req);
		252
		253	rpcrdma_recv_buffer_put(req->rl_reply);
		254	req->rl_reply = NULL;
282		255
283	spin_lock_bh(&xprt->bc_pa_lock);	256	spin_lock_bh(&xprt->bc_pa_lock);
284	list_add_tail(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);	257	list_add_tail(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);

net/sunrpc/xprtrdma/fmr_ops.c+23−0

@@ -156,10 +156,32 @@ out_release: · 155 unmodified lines
156	fmr_op_release_mr(mr);	156	fmr_op_release_mr(mr);
157	}	157	}
158		158
		159	/* On success, sets:
		160	* ep->rep_attr.cap.max_send_wr
		161	* ep->rep_attr.cap.max_recv_wr
		162	* cdata->max_requests
		163	* ia->ri_max_segs
		164	*/
159	static int	165	static int
160	fmr_op_open(struct rpcrdma_ia ia, struct rpcrdma_ep ep,	166	fmr_op_open(struct rpcrdma_ia ia, struct rpcrdma_ep ep,
161	struct rpcrdma_create_data_internal *cdata)	167	struct rpcrdma_create_data_internal *cdata)
162	{	168	{
		169	int max_qp_wr;
		170
		171	max_qp_wr = ia->ri_device->attrs.max_qp_wr;
		172	max_qp_wr -= RPCRDMA_BACKWARD_WRS;
		173	max_qp_wr -= 1;
		174	if (max_qp_wr < RPCRDMA_MIN_SLOT_TABLE)
		175	return -ENOMEM;
		176	if (cdata->max_requests > max_qp_wr)
		177	cdata->max_requests = max_qp_wr;
		178	ep->rep_attr.cap.max_send_wr = cdata->max_requests;
		179	ep->rep_attr.cap.max_send_wr += RPCRDMA_BACKWARD_WRS;
		180	ep->rep_attr.cap.max_send_wr += 1; /* for ib_drain_sq */
		181	ep->rep_attr.cap.max_recv_wr = cdata->max_requests;
		182	ep->rep_attr.cap.max_recv_wr += RPCRDMA_BACKWARD_WRS;
		183	ep->rep_attr.cap.max_recv_wr += 1; /* for ib_drain_rq */
		184
163	ia->ri_max_segs = max_t(unsigned int, 1, RPCRDMA_MAX_DATA_SEGS /	185	ia->ri_max_segs = max_t(unsigned int, 1, RPCRDMA_MAX_DATA_SEGS /
164	RPCRDMA_MAX_FMR_SGES);	186	RPCRDMA_MAX_FMR_SGES);
165	return 0;	187	return 0;
@@ -219,6 +241,7 @@ fmr_op_map(struct rpcrdma_xprt r_xprt, struct rpcrdma_mr_seg seg, · 53 unmodified lines
219	mr->mr_sg, i, mr->mr_dir);	241	mr->mr_sg, i, mr->mr_dir);
220	if (!mr->mr_nents)	242	if (!mr->mr_nents)
221	goto out_dmamap_err;	243	goto out_dmamap_err;
		244	trace_xprtrdma_dma_map(mr);
222		245
223	for (i = 0, dma_pages = mr->fmr.fm_physaddrs; i < mr->mr_nents; i++)	246	for (i = 0, dma_pages = mr->fmr.fm_physaddrs; i < mr->mr_nents; i++)
224	dma_pages[i] = sg_dma_address(&mr->mr_sg[i]);	247	dma_pages[i] = sg_dma_address(&mr->mr_sg[i]);

net/sunrpc/xprtrdma/frwr_ops.c+27−4

 	frwr_op_release_mr(mr);
 }
+/* On success, sets:
+ *	ep->rep_attr.cap.max_send_wr
+ *	ep->rep_attr.cap.max_recv_wr
+ *	cdata->max_requests
+ *	ia->ri_max_segs
+ *
+ * And these FRWR-related fields:
+ *	ia->ri_max_frwr_depth
+ *	ia->ri_mrtype
+ */
 static int
 frwr_op_open(struct rpcrdma_ia *ia, struct rpcrdma_ep *ep,
 	     struct rpcrdma_create_data_internal *cdata)
 {
 	struct ib_device_attr *attrs = &ia->ri_device->attrs;
-	int depth, delta;
+	int max_qp_wr, depth, delta;
 	ia->ri_mrtype = IB_MR_TYPE_MEM_REG;
 	if (attrs->device_cap_flags & IB_DEVICE_SG_GAPS_REG)
 		} while (delta > 0);
 	}
-	ep->rep_attr.cap.max_send_wr *= depth;
+	max_qp_wr = ia->ri_device->attrs.max_qp_wr;
-	if (ep->rep_attr.cap.max_send_wr > attrs->max_qp_wr) {
+	max_qp_wr -= RPCRDMA_BACKWARD_WRS;
-		cdata->max_requests = attrs->max_qp_wr / depth;
+	max_qp_wr -= 1;
+	if (max_qp_wr < RPCRDMA_MIN_SLOT_TABLE)
+		return -ENOMEM;
+	if (cdata->max_requests > max_qp_wr)
+		cdata->max_requests = max_qp_wr;
+	ep->rep_attr.cap.max_send_wr = cdata->max_requests * depth;
+	if (ep->rep_attr.cap.max_send_wr > max_qp_wr) {
+		cdata->max_requests = max_qp_wr / depth;
 		if (!cdata->max_requests)
 			return -EINVAL;
 		ep->rep_attr.cap.max_send_wr = cdata->max_requests *
 					       depth;
 	}
+	ep->rep_attr.cap.max_send_wr += RPCRDMA_BACKWARD_WRS;
+	ep->rep_attr.cap.max_send_wr += 1; /* for ib_drain_sq */
+	ep->rep_attr.cap.max_recv_wr = cdata->max_requests;
+	ep->rep_attr.cap.max_recv_wr += RPCRDMA_BACKWARD_WRS;
+	ep->rep_attr.cap.max_recv_wr += 1; /* for ib_drain_rq */
 	ia->ri_max_segs = max_t(unsigned int, 1, RPCRDMA_MAX_DATA_SEGS /
 				ia->ri_max_frwr_depth);
 	mr->mr_nents = ib_dma_map_sg(ia->ri_device, mr->mr_sg, i, mr->mr_dir);
 	if (!mr->mr_nents)
 		goto out_dmamap_err;
+	trace_xprtrdma_dma_map(mr);
 	ibmr = frwr->fr_mr;
 	n = ib_map_mr_sg(ibmr, mr->mr_sg, mr->mr_nents, NULL, PAGE_SIZE);

net/sunrpc/xprtrdma/module.c+1−0

@@ -1,3 +1,4 @@
		1	// SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
1	/*	2	/*
2	* Copyright (c) 2015, 2017 Oracle. All rights reserved.	3	* Copyright (c) 2015, 2017 Oracle. All rights reserved.
3	*/	4	*/

net/sunrpc/xprtrdma/rpc_rdma.c+22−44

+// SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
 /*
  * Copyright (c) 2014-2017 Oracle.  All rights reserved.
  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
 # define RPCDBG_FACILITY	RPCDBG_TRANS
 #endif
-static const char transfertypes[][12] = {
-	"inline",	/* no chunks */
-	"read list",	/* some argument via rdma read */
-	"*read list",	/* entire request via rdma read */
-	"write list",	/* some result via rdma write */
-	"reply chunk"	/* entire reply via rdma write */
-};
 /* Returns size of largest RPC-over-RDMA header in a Call message
  *
  * The largest Call header contains a full-size Read list and a
 			 */
 			*ppages = alloc_page(GFP_ATOMIC);
 			if (!*ppages)
-				return -EAGAIN;
+				return -ENOBUFS;
 		}
 		seg->mr_page = *ppages;
 		seg->mr_offset = (char *)page_base;
 		seg = r_xprt->rx_ia.ri_ops->ro_map(r_xprt, seg, nsegs,
 						   false, &mr);
 		if (IS_ERR(seg))
-			goto out_maperr;
+			return PTR_ERR(seg);
 		rpcrdma_mr_push(mr, &req->rl_registered);
 		if (encode_read_segment(xdr, mr, pos) < 0)
 	} while (nsegs);
 	return 0;
-out_maperr:
-	if (PTR_ERR(seg) == -EAGAIN)
-		xprt_wait_for_buffer_space(rqst->rq_task, NULL);
-	return PTR_ERR(seg);
 }
 /* Register and XDR encode the Write list. Supports encoding a list
 		seg = r_xprt->rx_ia.ri_ops->ro_map(r_xprt, seg, nsegs,
 						   true, &mr);
 		if (IS_ERR(seg))
-			goto out_maperr;
+			return PTR_ERR(seg);
 		rpcrdma_mr_push(mr, &req->rl_registered);
 		if (encode_rdma_segment(xdr, mr) < 0)
 	*segcount = cpu_to_be32(nchunks);
 	return 0;
-out_maperr:
-	if (PTR_ERR(seg) == -EAGAIN)
-		xprt_wait_for_buffer_space(rqst->rq_task, NULL);
-	return PTR_ERR(seg);
 }
 /* Register and XDR encode the Reply chunk. Supports encoding an array
 		seg = r_xprt->rx_ia.ri_ops->ro_map(r_xprt, seg, nsegs,
 						   true, &mr);
 		if (IS_ERR(seg))
-			goto out_maperr;
+			return PTR_ERR(seg);
 		rpcrdma_mr_push(mr, &req->rl_registered);
 		if (encode_rdma_segment(xdr, mr) < 0)
 	*segcount = cpu_to_be32(nchunks);
 	return 0;
-out_maperr:
-	if (PTR_ERR(seg) == -EAGAIN)
-		xprt_wait_for_buffer_space(rqst->rq_task, NULL);
-	return PTR_ERR(seg);
 }
 /**
 {
 	req->rl_sendctx = rpcrdma_sendctx_get_locked(&r_xprt->rx_buf);
 	if (!req->rl_sendctx)
-		return -ENOBUFS;
+		return -EAGAIN;
 	req->rl_sendctx->sc_wr.num_sge = 0;
 	req->rl_sendctx->sc_unmap_count = 0;
 	req->rl_sendctx->sc_req = req;
 	return 0;
 out_err:
-	r_xprt->rx_stats.failed_marshal_count++;
+	switch (ret) {
+	case -EAGAIN:
+		xprt_wait_for_buffer_space(rqst->rq_task, NULL);
+		break;
+	case -ENOBUFS:
+		break;
+	default:
+		r_xprt->rx_stats.failed_marshal_count++;
+	}
 	return ret;
 }
 out_short:
 	pr_warn("RPC/RDMA short backward direction call\n");
-	if (rpcrdma_ep_post_recv(&r_xprt->rx_ia, rep))
-		xprt_disconnect_done(&r_xprt->rx_xprt);
 	return true;
 }
 #else	/* CONFIG_SUNRPC_BACKCHANNEL */
 	u32 credits;
 	__be32 *p;
+	--buf->rb_posted_receives;
 	if (rep->rr_hdrbuf.head[0].iov_len == 0)
 		goto out_badstatus;
+	/* Fixed transport header fields */
 	xdr_init_decode(&rep->rr_stream, &rep->rr_hdrbuf,
 			rep->rr_hdrbuf.head[0].iov_base);
-	/* Fixed transport header fields */
 	p = xdr_inline_decode(&rep->rr_stream, 4 * sizeof(*p));
 	if (unlikely(!p))
 		goto out_shortreply;
 	trace_xprtrdma_reply(rqst->rq_task, rep, req, credits);
+	rpcrdma_post_recvs(r_xprt, false);
 	queue_work(rpcrdma_receive_wq, &rep->rr_work);
 	return;
-out_badstatus:
-	rpcrdma_recv_buffer_put(rep);
-	if (r_xprt->rx_ep.rep_connected == 1) {
-		r_xprt->rx_ep.rep_connected = -EIO;
-		rpcrdma_conn_func(&r_xprt->rx_ep);
-	}
-	return;
 out_badversion:
 	trace_xprtrdma_reply_vers(rep);
 	goto repost;
  * receive buffer before returning.
  */
 repost:
-	r_xprt->rx_stats.bad_reply_count++;
+	rpcrdma_post_recvs(r_xprt, false);
-	if (rpcrdma_ep_post_recv(&r_xprt->rx_ia, rep))
+out_badstatus:
-		rpcrdma_recv_buffer_put(rep);
+	rpcrdma_recv_buffer_put(rep);
 }

net/sunrpc/xprtrdma/svc_rdma_backchannel.c+1−0

@@ -273,6 +273,7 @@ static const struct rpc_xprt_ops xprt_rdma_bc_procs = { · 272 unmodified lines
273	.reserve_xprt = xprt_reserve_xprt_cong,	273	.reserve_xprt = xprt_reserve_xprt_cong,
274	.release_xprt = xprt_release_xprt_cong,	274	.release_xprt = xprt_release_xprt_cong,
275	.alloc_slot = xprt_alloc_slot,	275	.alloc_slot = xprt_alloc_slot,
		276	.free_slot = xprt_free_slot,
276	.release_request = xprt_release_rqst_cong,	277	.release_request = xprt_release_rqst_cong,
277	.buf_alloc = xprt_rdma_bc_allocate,	278	.buf_alloc = xprt_rdma_bc_allocate,
278	.buf_free = xprt_rdma_bc_free,	279	.buf_free = xprt_rdma_bc_free,

net/sunrpc/xprtrdma/transport.c+47−17

+// SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
 /*
  * Copyright (c) 2014-2017 Oracle.  All rights reserved.
  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
 		return ERR_PTR(-EBADF);
 	}
-	xprt = xprt_alloc(args->net, sizeof(struct rpcrdma_xprt),
+	xprt = xprt_alloc(args->net, sizeof(struct rpcrdma_xprt), 0, 0);
-			xprt_rdma_slot_table_entries,
-			xprt_rdma_slot_table_entries);
 	if (xprt == NULL) {
 		dprintk("RPC:       %s: couldn't allocate rpcrdma_xprt\n",
 			__func__);
 		xprt_set_bound(xprt);
 	xprt_rdma_format_addresses(xprt, sap);
-	cdata.max_requests = xprt->max_reqs;
+	cdata.max_requests = xprt_rdma_slot_table_entries;
 	cdata.rsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA write max */
 	cdata.wsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA read max */
 	}
 }
+/**
+ * xprt_rdma_alloc_slot - allocate an rpc_rqst
+ * @xprt: controlling RPC transport
+ * @task: RPC task requesting a fresh rpc_rqst
+ *
+ * tk_status values:
+ *	%0 if task->tk_rqstp points to a fresh rpc_rqst
+ *	%-EAGAIN if no rpc_rqst is available; queued on backlog
+ */
+static void
+xprt_rdma_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task)
+{
+	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
+	struct rpcrdma_req *req;
+	req = rpcrdma_buffer_get(&r_xprt->rx_buf);
+	if (!req)
+		goto out_sleep;
+	task->tk_rqstp = &req->rl_slot;
+	task->tk_status = 0;
+	return;
+out_sleep:
+	rpc_sleep_on(&xprt->backlog, task, NULL);
+	task->tk_status = -EAGAIN;
+}
+/**
+ * xprt_rdma_free_slot - release an rpc_rqst
+ * @xprt: controlling RPC transport
+ * @rqst: rpc_rqst to release
+ *
+ */
+static void
+xprt_rdma_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *rqst)
+{
+	memset(rqst, 0, sizeof(*rqst));
+	rpcrdma_buffer_put(rpcr_to_rdmar(rqst));
+	rpc_wake_up_next(&xprt->backlog);
+}
 static bool
 rpcrdma_get_sendbuf(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req,
 		    size_t size, gfp_t flags)
 {
 	struct rpc_rqst *rqst = task->tk_rqstp;
 	struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt);
-	struct rpcrdma_req *req;
+	struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
 	gfp_t flags;
-	req = rpcrdma_buffer_get(&r_xprt->rx_buf);
-	if (req == NULL)
-		goto out_get;
 	flags = RPCRDMA_DEF_GFP;
 	if (RPC_IS_SWAPPER(task))
 		flags = __GFP_MEMALLOC | GFP_NOWAIT | __GFP_NOWARN;
 	if (!rpcrdma_get_recvbuf(r_xprt, req, rqst->rq_rcvsize, flags))
 		goto out_fail;
-	rpcrdma_set_xprtdata(rqst, req);
 	rqst->rq_buffer = req->rl_sendbuf->rg_base;
 	rqst->rq_rbuffer = req->rl_recvbuf->rg_base;
 	trace_xprtrdma_allocate(task, req);
 	return 0;
 out_fail:
-	rpcrdma_buffer_put(req);
-out_get:
 	trace_xprtrdma_allocate(task, NULL);
 	return -ENOMEM;
 }
 	if (test_bit(RPCRDMA_REQ_F_PENDING, &req->rl_flags))
 		rpcrdma_release_rqst(r_xprt, req);
 	trace_xprtrdma_rpc_done(task, req);
-	rpcrdma_buffer_put(req);
 }
 /**
 	if (rc < 0)
 		goto failed_marshal;
-	if (req->rl_reply == NULL) 		/* e.g. reconnection */
-		rpcrdma_recv_buffer_get(req);
 	/* Must suppress retransmit to maintain credits */
 	if (rqst->rq_connect_cookie == xprt->connect_cookie)
 		goto drop_connection;
 static const struct rpc_xprt_ops xprt_rdma_procs = {
 	.reserve_xprt		= xprt_reserve_xprt_cong,
 	.release_xprt		= xprt_release_xprt_cong, /* sunrpc/xprt.c */
-	.alloc_slot		= xprt_alloc_slot,
+	.alloc_slot		= xprt_rdma_alloc_slot,
+	.free_slot		= xprt_rdma_free_slot,
 	.release_request	= xprt_release_rqst_cong,       /* ditto */
 	.set_retrans_timeout	= xprt_set_retrans_timeout_def, /* ditto */
 	.timer			= xprt_rdma_timer,

net/sunrpc/xprtrdma/verbs.c+112−179

@@ -1,3 +1,4 @@
		1	// SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
1	/*	2	/*
2	* Copyright (c) 2014-2017 Oracle. All rights reserved.	3	* Copyright (c) 2014-2017 Oracle. All rights reserved.
3	* Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.	4	* Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
@@ -71,8 +72,10 @@ · 67 unmodified lines
71	/*	72	/*
72	* internal functions	73	* internal functions
73	*/	74	*/
		75	static void rpcrdma_sendctx_put_locked(struct rpcrdma_sendctx *sc);
74	static void rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt);	76	static void rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt);
75	static void rpcrdma_mrs_destroy(struct rpcrdma_buffer *buf);	77	static void rpcrdma_mrs_destroy(struct rpcrdma_buffer *buf);
		78	static int rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt, bool temp);
76	static void rpcrdma_dma_unmap_regbuf(struct rpcrdma_regbuf *rb);	79	static void rpcrdma_dma_unmap_regbuf(struct rpcrdma_regbuf *rb);
77		80
78	struct workqueue_struct *rpcrdma_receive_wq __read_mostly;	81	struct workqueue_struct *rpcrdma_receive_wq __read_mostly;
@@ -159,7 +162,7 @@ rpcrdma_wc_receive(struct ib_cq cq, struct ib_wc wc) · 80 unmodified lines
159	rr_cqe);	162	rr_cqe);
160		163
161	/* WARNING: Only wr_id and status are reliable at this point */	164	/* WARNING: Only wr_id and status are reliable at this point */
162	trace_xprtrdma_wc_receive(rep, wc);	165	trace_xprtrdma_wc_receive(wc);
163	if (wc->status != IB_WC_SUCCESS)	166	if (wc->status != IB_WC_SUCCESS)
164	goto out_fail;	167	goto out_fail;
165		168
@@ -231,7 +234,7 @@ rpcrdma_conn_upcall(struct rdma_cm_id id, struct rdma_cm_event event) · 65 unmodified lines
231	complete(&ia->ri_done);	234	complete(&ia->ri_done);
232	break;	235	break;
233	case RDMA_CM_EVENT_ADDR_ERROR:	236	case RDMA_CM_EVENT_ADDR_ERROR:
234	ia->ri_async_rc = -EHOSTUNREACH;	237	ia->ri_async_rc = -EPROTO;
235	complete(&ia->ri_done);	238	complete(&ia->ri_done);
236	break;	239	break;
237	case RDMA_CM_EVENT_ROUTE_ERROR:	240	case RDMA_CM_EVENT_ROUTE_ERROR:
@@ -262,7 +265,7 @@ rpcrdma_conn_upcall(struct rdma_cm_id id, struct rdma_cm_event event) · 24 unmodified lines
262	connstate = -ENOTCONN;	265	connstate = -ENOTCONN;
263	goto connected;	266	goto connected;
264	case RDMA_CM_EVENT_UNREACHABLE:	267	case RDMA_CM_EVENT_UNREACHABLE:
265	connstate = -ENETDOWN;	268	connstate = -ENETUNREACH;
266	goto connected;	269	goto connected;
267	case RDMA_CM_EVENT_REJECTED:	270	case RDMA_CM_EVENT_REJECTED:
268	dprintk("rpcrdma: connection to %s:%s rejected: %s\n",	271	dprintk("rpcrdma: connection to %s:%s rejected: %s\n",
@@ -305,8 +308,8 @@ rpcrdma_create_id(struct rpcrdma_xprt xprt, struct rpcrdma_ia ia) · 36 unmodified lines
305	init_completion(&ia->ri_done);	308	init_completion(&ia->ri_done);
306	init_completion(&ia->ri_remove_done);	309	init_completion(&ia->ri_remove_done);
307		310
308	id = rdma_create_id(&init_net, rpcrdma_conn_upcall, xprt, RDMA_PS_TCP,	311	id = rdma_create_id(xprt->rx_xprt.xprt_net, rpcrdma_conn_upcall,
309	IB_QPT_RC);	312	xprt, RDMA_PS_TCP, IB_QPT_RC);
310	if (IS_ERR(id)) {	313	if (IS_ERR(id)) {
311	rc = PTR_ERR(id);	314	rc = PTR_ERR(id);
312	dprintk("RPC: %s: rdma_create_id() failed %i\n",	315	dprintk("RPC: %s: rdma_create_id() failed %i\n",
@@ -500,8 +503,8 @@ rpcrdma_ep_create(struct rpcrdma_ep ep, struct rpcrdma_ia ia, · 187 unmodified lines
500	struct rpcrdma_create_data_internal *cdata)	503	struct rpcrdma_create_data_internal *cdata)
501	{	504	{
502	struct rpcrdma_connect_private *pmsg = &ep->rep_cm_private;	505	struct rpcrdma_connect_private *pmsg = &ep->rep_cm_private;
503	unsigned int max_qp_wr, max_sge;
504	struct ib_cq sendcq, recvcq;	506	struct ib_cq sendcq, recvcq;
		507	unsigned int max_sge;
505	int rc;	508	int rc;
506		509
507	max_sge = min_t(unsigned int, ia->ri_device->attrs.max_sge,	510	max_sge = min_t(unsigned int, ia->ri_device->attrs.max_sge,
@@ -512,29 +515,13 @@ rpcrdma_ep_create(struct rpcrdma_ep ep, struct rpcrdma_ia ia, · 4 unmodified lines
512	}	515	}
513	ia->ri_max_send_sges = max_sge;	516	ia->ri_max_send_sges = max_sge;
514		517
515	if (ia->ri_device->attrs.max_qp_wr <= RPCRDMA_BACKWARD_WRS) {	518	rc = ia->ri_ops->ro_open(ia, ep, cdata);
516	dprintk("RPC: %s: insufficient wqe's available\n",	519	if (rc)
517	__func__);	520	return rc;
518	return -ENOMEM;
519	}
520	max_qp_wr = ia->ri_device->attrs.max_qp_wr - RPCRDMA_BACKWARD_WRS - 1;
521
522	/* check provider's send/recv wr limits */
523	if (cdata->max_requests > max_qp_wr)
524	cdata->max_requests = max_qp_wr;
525		521
526	ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall;	522	ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall;
527	ep->rep_attr.qp_context = ep;	523	ep->rep_attr.qp_context = ep;
528	ep->rep_attr.srq = NULL;	524	ep->rep_attr.srq = NULL;
529	ep->rep_attr.cap.max_send_wr = cdata->max_requests;
530	ep->rep_attr.cap.max_send_wr += RPCRDMA_BACKWARD_WRS;
531	ep->rep_attr.cap.max_send_wr += 1; /* drain cqe */
532	rc = ia->ri_ops->ro_open(ia, ep, cdata);
533	if (rc)
534	return rc;
535	ep->rep_attr.cap.max_recv_wr = cdata->max_requests;
536	ep->rep_attr.cap.max_recv_wr += RPCRDMA_BACKWARD_WRS;
537	ep->rep_attr.cap.max_recv_wr += 1; /* drain cqe */
538	ep->rep_attr.cap.max_send_sge = max_sge;	525	ep->rep_attr.cap.max_send_sge = max_sge;
539	ep->rep_attr.cap.max_recv_sge = 1;	526	ep->rep_attr.cap.max_recv_sge = 1;
540	ep->rep_attr.cap.max_inline_data = 0;	527	ep->rep_attr.cap.max_inline_data = 0;
@@ -741,7 +728,6 @@ rpcrdma_ep_connect(struct rpcrdma_ep ep, struct rpcrdma_ia ia) · 200 unmodified lines
741	{	728	{
742	struct rpcrdma_xprt *r_xprt = container_of(ia, struct rpcrdma_xprt,	729	struct rpcrdma_xprt *r_xprt = container_of(ia, struct rpcrdma_xprt,
743	rx_ia);	730	rx_ia);
744	unsigned int extras;
745	int rc;	731	int rc;
746		732
747	retry:	733	retry:
@@ -785,9 +771,8 @@ retry: · 37 unmodified lines
785	}	771	}
786		772
787	dprintk("RPC: %s: connected\n", __func__);	773	dprintk("RPC: %s: connected\n", __func__);
788	extras = r_xprt->rx_buf.rb_bc_srv_max_requests;	774
789	if (extras)	775	rpcrdma_post_recvs(r_xprt, true);
790	rpcrdma_ep_post_extra_recv(r_xprt, extras);
791		776
792	out:	777	out:
793	if (rc)	778	if (rc)
@@ -893,6 +878,7 @@ static int rpcrdma_sendctxs_create(struct rpcrdma_xprt *r_xprt) · 99 unmodified lines
893	sc->sc_xprt = r_xprt;	878	sc->sc_xprt = r_xprt;
894	buf->rb_sc_ctxs[i] = sc;	879	buf->rb_sc_ctxs[i] = sc;
895	}	880	}
		881	buf->rb_flags = 0;
896		882
897	return 0;	883	return 0;
898		884
@@ -950,7 +936,7 @@ out_emptyq: · 51 unmodified lines
950	* completions recently. This is a sign the Send Queue is	936	* completions recently. This is a sign the Send Queue is
951	* backing up. Cause the caller to pause and try again.	937	* backing up. Cause the caller to pause and try again.
952	*/	938	*/
953	dprintk("RPC: %s: empty sendctx queue\n", __func__);	939	set_bit(RPCRDMA_BUF_F_EMPTY_SCQ, &buf->rb_flags);
954	r_xprt = container_of(buf, struct rpcrdma_xprt, rx_buf);	940	r_xprt = container_of(buf, struct rpcrdma_xprt, rx_buf);
955	r_xprt->rx_stats.empty_sendctx_q++;	941	r_xprt->rx_stats.empty_sendctx_q++;
956	return NULL;	942	return NULL;
@@ -965,7 +951,8 @@ out_emptyq: · 8 unmodified lines
965	*	951	*
966	* The caller serializes calls to this function (per rpcrdma_buffer).	952	* The caller serializes calls to this function (per rpcrdma_buffer).
967	*/	953	*/
968	void rpcrdma_sendctx_put_locked(struct rpcrdma_sendctx *sc)	954	static void
		955	rpcrdma_sendctx_put_locked(struct rpcrdma_sendctx *sc)
969	{	956	{
970	struct rpcrdma_buffer *buf = &sc->sc_xprt->rx_buf;	957	struct rpcrdma_buffer *buf = &sc->sc_xprt->rx_buf;
971	unsigned long next_tail;	958	unsigned long next_tail;
@@ -984,6 +971,11 @@ void rpcrdma_sendctx_put_locked(struct rpcrdma_sendctx *sc) · 12 unmodified lines
984		971
985	/* Paired with READ_ONCE */	972	/* Paired with READ_ONCE */
986	smp_store_release(&buf->rb_sc_tail, next_tail);	973	smp_store_release(&buf->rb_sc_tail, next_tail);
		974
		975	if (test_and_clear_bit(RPCRDMA_BUF_F_EMPTY_SCQ, &buf->rb_flags)) {
		976	smp_mb__after_atomic();
		977	xprt_write_space(&sc->sc_xprt->rx_xprt);
		978	}
987	}	979	}
988		980
989	static void	981	static void
@@ -1097,14 +1089,8 @@ rpcrdma_create_req(struct rpcrdma_xprt *r_xprt) · 107 unmodified lines
1097	return req;	1089	return req;
1098	}	1090	}
1099		1091
1100	/**	1092	static int
1101	* rpcrdma_create_rep - Allocate an rpcrdma_rep object	1093	rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt, bool temp)
1102	* @r_xprt: controlling transport
1103	*
1104	* Returns 0 on success or a negative errno on failure.
1105	*/
1106	int
1107	rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt)
1108	{	1094	{
1109	struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;	1095	struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
1110	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;	1096	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
@@ -1132,6 +1118,7 @@ rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt) · 21 unmodified lines
1132	rep->rr_recv_wr.wr_cqe = &rep->rr_cqe;	1118	rep->rr_recv_wr.wr_cqe = &rep->rr_cqe;
1133	rep->rr_recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov;	1119	rep->rr_recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov;
1134	rep->rr_recv_wr.num_sge = 1;	1120	rep->rr_recv_wr.num_sge = 1;
		1121	rep->rr_temp = temp;
1135		1122
1136	spin_lock(&buf->rb_lock);	1123	spin_lock(&buf->rb_lock);
1137	list_add(&rep->rr_list, &buf->rb_recv_bufs);	1124	list_add(&rep->rr_list, &buf->rb_recv_bufs);
@@ -1183,12 +1170,8 @@ rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt) · 45 unmodified lines
1183	list_add(&req->rl_list, &buf->rb_send_bufs);	1170	list_add(&req->rl_list, &buf->rb_send_bufs);
1184	}	1171	}
1185		1172
		1173	buf->rb_posted_receives = 0;
1186	INIT_LIST_HEAD(&buf->rb_recv_bufs);	1174	INIT_LIST_HEAD(&buf->rb_recv_bufs);
1187	for (i = 0; i <= buf->rb_max_requests; i++) {
1188	rc = rpcrdma_create_rep(r_xprt);
1189	if (rc)
1190	goto out;
1191	}
1192		1175
1193	rc = rpcrdma_sendctxs_create(r_xprt);	1176	rc = rpcrdma_sendctxs_create(r_xprt);
1194	if (rc)	1177	if (rc)
@@ -1200,28 +1183,6 @@ out: · 5 unmodified lines
1200	return rc;	1183	return rc;
1201	}	1184	}
1202		1185
1203	static struct rpcrdma_req *
1204	rpcrdma_buffer_get_req_locked(struct rpcrdma_buffer *buf)
1205	{
1206	struct rpcrdma_req *req;
1207
1208	req = list_first_entry(&buf->rb_send_bufs,
1209	struct rpcrdma_req, rl_list);
1210	list_del_init(&req->rl_list);
1211	return req;
1212	}
1213
1214	static struct rpcrdma_rep *
1215	rpcrdma_buffer_get_rep_locked(struct rpcrdma_buffer *buf)
1216	{
1217	struct rpcrdma_rep *rep;
1218
1219	rep = list_first_entry(&buf->rb_recv_bufs,
1220	struct rpcrdma_rep, rr_list);
1221	list_del(&rep->rr_list);
1222	return rep;
1223	}
1224
1225	static void	1186	static void
1226	rpcrdma_destroy_rep(struct rpcrdma_rep *rep)	1187	rpcrdma_destroy_rep(struct rpcrdma_rep *rep)
1227	{	1188	{
@@ -1280,10 +1241,11 @@ rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf) · 52 unmodified lines
1280	while (!list_empty(&buf->rb_recv_bufs)) {	1241	while (!list_empty(&buf->rb_recv_bufs)) {
1281	struct rpcrdma_rep *rep;	1242	struct rpcrdma_rep *rep;
1282		1243
1283	rep = rpcrdma_buffer_get_rep_locked(buf);	1244	rep = list_first_entry(&buf->rb_recv_bufs,
		1245	struct rpcrdma_rep, rr_list);
		1246	list_del(&rep->rr_list);
1284	rpcrdma_destroy_rep(rep);	1247	rpcrdma_destroy_rep(rep);
1285	}	1248	}
1286	buf->rb_send_count = 0;
1287		1249
1288	spin_lock(&buf->rb_reqslock);	1250	spin_lock(&buf->rb_reqslock);
1289	while (!list_empty(&buf->rb_allreqs)) {	1251	while (!list_empty(&buf->rb_allreqs)) {
@@ -1298,7 +1260,6 @@ rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf) · 8 unmodified lines
1298	spin_lock(&buf->rb_reqslock);	1260	spin_lock(&buf->rb_reqslock);
1299	}	1261	}
1300	spin_unlock(&buf->rb_reqslock);	1262	spin_unlock(&buf->rb_reqslock);
1301	buf->rb_recv_count = 0;
1302		1263
1303	rpcrdma_mrs_destroy(buf);	1264	rpcrdma_mrs_destroy(buf);
1304	}	1265	}
@@ -1371,27 +1332,11 @@ rpcrdma_mr_unmap_and_put(struct rpcrdma_mr *mr) · 66 unmodified lines
1371	__rpcrdma_mr_put(&r_xprt->rx_buf, mr);	1332	__rpcrdma_mr_put(&r_xprt->rx_buf, mr);
1372	}	1333	}
1373		1334
1374	static struct rpcrdma_rep *	1335	/**
1375	rpcrdma_buffer_get_rep(struct rpcrdma_buffer *buffers)	1336	* rpcrdma_buffer_get - Get a request buffer
1376	{	1337	* @buffers: Buffer pool from which to obtain a buffer
1377	/* If an RPC previously completed without a reply (say, a
1378	* credential problem or a soft timeout occurs) then hold off
1379	* on supplying more Receive buffers until the number of new
1380	* pending RPCs catches up to the number of posted Receives.
1381	*/
1382	if (unlikely(buffers->rb_send_count < buffers->rb_recv_count))
1383	return NULL;
1384
1385	if (unlikely(list_empty(&buffers->rb_recv_bufs)))
1386	return NULL;
1387	buffers->rb_recv_count++;
1388	return rpcrdma_buffer_get_rep_locked(buffers);
1389	}
1390
1391	/*
1392	* Get a set of request/reply buffers.
1393	*	1338	*
1394	* Reply buffer (if available) is attached to send buffer upon return.	1339	* Returns a fresh rpcrdma_req, or NULL if none are available.
1395	*/	1340	*/
1396	struct rpcrdma_req *	1341	struct rpcrdma_req *
1397	rpcrdma_buffer_get(struct rpcrdma_buffer *buffers)	1342	rpcrdma_buffer_get(struct rpcrdma_buffer *buffers)
@@ -1399,23 +1344,18 @@ rpcrdma_buffer_get(struct rpcrdma_buffer *buffers) · 1 unmodified line
1399	struct rpcrdma_req *req;	1344	struct rpcrdma_req *req;
1400		1345
1401	spin_lock(&buffers->rb_lock);	1346	spin_lock(&buffers->rb_lock);
1402	if (list_empty(&buffers->rb_send_bufs))	1347	req = list_first_entry_or_null(&buffers->rb_send_bufs,
1403	goto out_reqbuf;	1348	struct rpcrdma_req, rl_list);
1404	buffers->rb_send_count++;	1349	if (req)
1405	req = rpcrdma_buffer_get_req_locked(buffers);	1350	list_del_init(&req->rl_list);
1406	req->rl_reply = rpcrdma_buffer_get_rep(buffers);
1407	spin_unlock(&buffers->rb_lock);	1351	spin_unlock(&buffers->rb_lock);
1408
1409	return req;	1352	return req;
1410
1411	out_reqbuf:
1412	spin_unlock(&buffers->rb_lock);
1413	return NULL;
1414	}	1353	}
1415		1354
1416	/*	1355	/**
1417	* Put request/reply buffers back into pool.	1356	* rpcrdma_buffer_put - Put request/reply buffers back into pool
1418	* Pre-decrement counter/array index.	1357	* @req: object to return
		1358	*
1419	*/	1359	*/
1420	void	1360	void
1421	rpcrdma_buffer_put(struct rpcrdma_req *req)	1361	rpcrdma_buffer_put(struct rpcrdma_req *req)
@@ -1426,27 +1366,16 @@ rpcrdma_buffer_put(struct rpcrdma_req *req) · 4 unmodified lines
1426	req->rl_reply = NULL;	1366	req->rl_reply = NULL;
1427		1367
1428	spin_lock(&buffers->rb_lock);	1368	spin_lock(&buffers->rb_lock);
1429	buffers->rb_send_count--;	1369	list_add(&req->rl_list, &buffers->rb_send_bufs);
1430	list_add_tail(&req->rl_list, &buffers->rb_send_bufs);
1431	if (rep) {	1370	if (rep) {
1432	buffers->rb_recv_count--;	1371	if (!rep->rr_temp) {
1433	list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs);	1372	list_add(&rep->rr_list, &buffers->rb_recv_bufs);
		1373	rep = NULL;
		1374	}
1434	}	1375	}
1435	spin_unlock(&buffers->rb_lock);	1376	spin_unlock(&buffers->rb_lock);
1436	}	1377	if (rep)
1437		1378	rpcrdma_destroy_rep(rep);
1438	/*
1439	* Recover reply buffers from pool.
1440	* This happens when recovering from disconnect.
1441	*/
1442	void
1443	rpcrdma_recv_buffer_get(struct rpcrdma_req *req)
1444	{
1445	struct rpcrdma_buffer *buffers = req->rl_buffer;
1446
1447	spin_lock(&buffers->rb_lock);
1448	req->rl_reply = rpcrdma_buffer_get_rep(buffers);
1449	spin_unlock(&buffers->rb_lock);
1450	}	1379	}
1451		1380
1452	/*	1381	/*
@@ -1458,10 +1387,13 @@ rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep) · 5 unmodified lines
1458	{	1387	{
1459	struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf;	1388	struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf;
1460		1389
1461	spin_lock(&buffers->rb_lock);	1390	if (!rep->rr_temp) {
1462	buffers->rb_recv_count--;	1391	spin_lock(&buffers->rb_lock);
1463	list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs);	1392	list_add(&rep->rr_list, &buffers->rb_recv_bufs);
1464	spin_unlock(&buffers->rb_lock);	1393	spin_unlock(&buffers->rb_lock);
		1394	} else {
		1395	rpcrdma_destroy_rep(rep);
		1396	}
1465	}	1397	}
1466		1398
1467	/**	1399	/**
@@ -1557,13 +1489,6 @@ rpcrdma_ep_post(struct rpcrdma_ia *ia, · 89 unmodified lines
1557	struct ib_send_wr *send_wr = &req->rl_sendctx->sc_wr;	1489	struct ib_send_wr *send_wr = &req->rl_sendctx->sc_wr;
1558	int rc;	1490	int rc;
1559		1491
1560	if (req->rl_reply) {
1561	rc = rpcrdma_ep_post_recv(ia, req->rl_reply);
1562	if (rc)
1563	return rc;
1564	req->rl_reply = NULL;
1565	}
1566
1567	if (!ep->rep_send_count \|\|	1492	if (!ep->rep_send_count \|\|
1568	test_bit(RPCRDMA_REQ_F_TX_RESOURCES, &req->rl_flags)) {	1493	test_bit(RPCRDMA_REQ_F_TX_RESOURCES, &req->rl_flags)) {
1569	send_wr->send_flags \|= IB_SEND_SIGNALED;	1494	send_wr->send_flags \|= IB_SEND_SIGNALED;
@@ -1580,61 +1505,69 @@ rpcrdma_ep_post(struct rpcrdma_ia *ia, · 10 unmodified lines
1580	return 0;	1505	return 0;
1581	}	1506	}
1582		1507
1583	int
1584	rpcrdma_ep_post_recv(struct rpcrdma_ia *ia,
1585	struct rpcrdma_rep *rep)
1586	{
1587	struct ib_recv_wr *recv_wr_fail;
1588	int rc;
1589
1590	if (!rpcrdma_dma_map_regbuf(ia, rep->rr_rdmabuf))
1591	goto out_map;
1592	rc = ib_post_recv(ia->ri_id->qp, &rep->rr_recv_wr, &recv_wr_fail);
1593	trace_xprtrdma_post_recv(rep, rc);
1594	if (rc)
1595	return -ENOTCONN;
1596	return 0;
1597
1598	out_map:
1599	pr_err("rpcrdma: failed to DMA map the Receive buffer\n");
1600	return -EIO;
1601	}
1602
1603	/**	1508	/**
1604	* rpcrdma_ep_post_extra_recv - Post buffers for incoming backchannel requests	1509	* rpcrdma_post_recvs - Maybe post some Receive buffers
1605	* @r_xprt: transport associated with these backchannel resources	1510	* @r_xprt: controlling transport
1606	* @count: minimum number of incoming requests expected	1511	* @temp: when true, allocate temp rpcrdma_rep objects
1607	*	1512	*
1608	* Returns zero if all requested buffers were posted, or a negative errno.
1609	*/	1513	*/
1610	int	1514	void
1611	rpcrdma_ep_post_extra_recv(struct rpcrdma_xprt *r_xprt, unsigned int count)	1515	rpcrdma_post_recvs(struct rpcrdma_xprt *r_xprt, bool temp)
1612	{	1516	{
1613	struct rpcrdma_buffer *buffers = &r_xprt->rx_buf;	1517	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1614	struct rpcrdma_ia *ia = &r_xprt->rx_ia;	1518	struct ib_recv_wr wr, bad_wr;
1615	struct rpcrdma_rep *rep;	1519	int needed, count, rc;
1616	int rc;
1617		1520
1618	while (count--) {	1521	needed = buf->rb_credits + (buf->rb_bc_srv_max_requests << 1);
1619	spin_lock(&buffers->rb_lock);	1522	if (buf->rb_posted_receives > needed)
1620	if (list_empty(&buffers->rb_recv_bufs))	1523	return;
1621	goto out_reqbuf;	1524	needed -= buf->rb_posted_receives;
1622	rep = rpcrdma_buffer_get_rep_locked(buffers);
1623	spin_unlock(&buffers->rb_lock);
1624		1525
1625	rc = rpcrdma_ep_post_recv(ia, rep);	1526	count = 0;
1626	if (rc)	1527	wr = NULL;
1627	goto out_rc;	1528	while (needed) {
1628	}	1529	struct rpcrdma_regbuf *rb;
		1530	struct rpcrdma_rep *rep;
1629		1531
1630	return 0;	1532	spin_lock(&buf->rb_lock);
		1533	rep = list_first_entry_or_null(&buf->rb_recv_bufs,
		1534	struct rpcrdma_rep, rr_list);
		1535	if (likely(rep))
		1536	list_del(&rep->rr_list);
		1537	spin_unlock(&buf->rb_lock);
		1538	if (!rep) {
		1539	if (rpcrdma_create_rep(r_xprt, temp))
		1540	break;
		1541	continue;
		1542	}
1631		1543
1632	out_reqbuf:	1544	rb = rep->rr_rdmabuf;
1633	spin_unlock(&buffers->rb_lock);	1545	if (!rpcrdma_regbuf_is_mapped(rb)) {
1634	trace_xprtrdma_noreps(r_xprt);	1546	if (!__rpcrdma_dma_map_regbuf(&r_xprt->rx_ia, rb)) {
1635	return -ENOMEM;	1547	rpcrdma_recv_buffer_put(rep);
		1548	break;
		1549	}
		1550	}
1636		1551
1637	out_rc:	1552	trace_xprtrdma_post_recv(rep->rr_recv_wr.wr_cqe);
1638	rpcrdma_recv_buffer_put(rep);	1553	rep->rr_recv_wr.next = wr;
1639	return rc;	1554	wr = &rep->rr_recv_wr;
		1555	++count;
		1556	--needed;
		1557	}
		1558	if (!count)
		1559	return;
		1560
		1561	rc = ib_post_recv(r_xprt->rx_ia.ri_id->qp, wr, &bad_wr);
		1562	if (rc) {
		1563	for (wr = bad_wr; wr; wr = wr->next) {
		1564	struct rpcrdma_rep *rep;
		1565
		1566	rep = container_of(wr, struct rpcrdma_rep, rr_recv_wr);
		1567	rpcrdma_recv_buffer_put(rep);
		1568	--count;
		1569	}
		1570	}
		1571	buf->rb_posted_receives += count;
		1572	trace_xprtrdma_post_recvs(r_xprt, count, rc);
1640	}	1573	}

net/sunrpc/xprtrdma/xprt_rdma.h+12−14

+/* SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause */
 /*
  * Copyright (c) 2014-2017 Oracle.  All rights reserved.
  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
 	__be32			rr_proc;
 	int			rr_wc_flags;
 	u32			rr_inv_rkey;
+	bool			rr_temp;
 	struct rpcrdma_regbuf	*rr_rdmabuf;
 	struct rpcrdma_xprt	*rr_rxprt;
 	struct work_struct	rr_work;
 struct rpcrdma_buffer;
 struct rpcrdma_req {
 	struct list_head	rl_list;
+	struct rpc_rqst		rl_slot;
 	struct rpcrdma_buffer	*rl_buffer;
 	struct rpcrdma_rep	*rl_reply;
 	struct xdr_stream	rl_stream;
 	RPCRDMA_REQ_F_TX_RESOURCES,
 };
-static inline void
-rpcrdma_set_xprtdata(struct rpc_rqst *rqst, struct rpcrdma_req *req)
-{
-	rqst->rq_xprtdata = req;
-}
 static inline struct rpcrdma_req *
 rpcr_to_rdmar(const struct rpc_rqst *rqst)
 {
-	return rqst->rq_xprtdata;
+	return container_of(rqst, struct rpcrdma_req, rl_slot);
 }
 static inline void
 	struct rpcrdma_sendctx	**rb_sc_ctxs;
 	spinlock_t		rb_lock;	/* protect buf lists */
-	int			rb_send_count, rb_recv_count;
 	struct list_head	rb_send_bufs;
 	struct list_head	rb_recv_bufs;
+	unsigned long		rb_flags;
 	u32			rb_max_requests;
 	u32			rb_credits;	/* most recent credit grant */
+	int			rb_posted_receives;
 	u32			rb_bc_srv_max_requests;
 	spinlock_t		rb_reqslock;	/* protect rb_allreqs */
 };
 #define rdmab_to_ia(b) (&container_of((b), struct rpcrdma_xprt, rx_buf)->rx_ia)
+/* rb_flags */
+enum {
+	RPCRDMA_BUF_F_EMPTY_SCQ = 0,
+};
 /*
  * Internal structure for transport instance creation. This
  * exists primarily for modularity.
 int rpcrdma_ep_post(struct rpcrdma_ia *, struct rpcrdma_ep *,
 				struct rpcrdma_req *);
-int rpcrdma_ep_post_recv(struct rpcrdma_ia *, struct rpcrdma_rep *);
+void rpcrdma_post_recvs(struct rpcrdma_xprt *r_xprt, bool temp);
 /*
  * Buffer calls - xprtrdma/verbs.c
  */
 struct rpcrdma_req *rpcrdma_create_req(struct rpcrdma_xprt *);
 void rpcrdma_destroy_req(struct rpcrdma_req *);
-int rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt);
 int rpcrdma_buffer_create(struct rpcrdma_xprt *);
 void rpcrdma_buffer_destroy(struct rpcrdma_buffer *);
 struct rpcrdma_sendctx *rpcrdma_sendctx_get_locked(struct rpcrdma_buffer *buf);
-void rpcrdma_sendctx_put_locked(struct rpcrdma_sendctx *sc);
 struct rpcrdma_mr *rpcrdma_mr_get(struct rpcrdma_xprt *r_xprt);
 void rpcrdma_mr_put(struct rpcrdma_mr *mr);
 struct rpcrdma_req *rpcrdma_buffer_get(struct rpcrdma_buffer *);
 void rpcrdma_buffer_put(struct rpcrdma_req *);
-void rpcrdma_recv_buffer_get(struct rpcrdma_req *);
 void rpcrdma_recv_buffer_put(struct rpcrdma_rep *);
 struct rpcrdma_regbuf *rpcrdma_alloc_regbuf(size_t, enum dma_data_direction,
 	return __rpcrdma_dma_map_regbuf(ia, rb);
 }
-int rpcrdma_ep_post_extra_recv(struct rpcrdma_xprt *, unsigned int);
 int rpcrdma_alloc_wq(void);
 void rpcrdma_destroy_wq(void);

net/sunrpc/xprtsock.c+4−0

 	.reserve_xprt		= xprt_reserve_xprt,
 	.release_xprt		= xs_tcp_release_xprt,
 	.alloc_slot		= xprt_alloc_slot,
+	.free_slot		= xprt_free_slot,
 	.rpcbind		= xs_local_rpcbind,
 	.set_port		= xs_local_set_port,
 	.connect		= xs_local_connect,
 	.reserve_xprt		= xprt_reserve_xprt_cong,
 	.release_xprt		= xprt_release_xprt_cong,
 	.alloc_slot		= xprt_alloc_slot,
+	.free_slot		= xprt_free_slot,
 	.rpcbind		= rpcb_getport_async,
 	.set_port		= xs_set_port,
 	.connect		= xs_connect,
 	.reserve_xprt		= xprt_reserve_xprt,
 	.release_xprt		= xs_tcp_release_xprt,
 	.alloc_slot		= xprt_lock_and_alloc_slot,
+	.free_slot		= xprt_free_slot,
 	.rpcbind		= rpcb_getport_async,
 	.set_port		= xs_set_port,
 	.connect		= xs_connect,
 	.reserve_xprt		= xprt_reserve_xprt,
 	.release_xprt		= xprt_release_xprt,
 	.alloc_slot		= xprt_alloc_slot,
+	.free_slot		= xprt_free_slot,
 	.buf_alloc		= bc_malloc,
 	.buf_free		= bc_free,
 	.send_request		= bc_send_request,