1 // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
2 /*
3  * Copyright (c) 2014-2017 Oracle.  All rights reserved.
4  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
5  *
6  * This software is available to you under a choice of one of two
7  * licenses.  You may choose to be licensed under the terms of the GNU
8  * General Public License (GPL) Version 2, available from the file
9  * COPYING in the main directory of this source tree, or the BSD-type
10  * license below:
11  *
12  * Redistribution and use in source and binary forms, with or without
13  * modification, are permitted provided that the following conditions
14  * are met:
15  *
16  *      Redistributions of source code must retain the above copyright
17  *      notice, this list of conditions and the following disclaimer.
18  *
19  *      Redistributions in binary form must reproduce the above
20  *      copyright notice, this list of conditions and the following
21  *      disclaimer in the documentation and/or other materials provided
22  *      with the distribution.
23  *
24  *      Neither the name of the Network Appliance, Inc. nor the names of
25  *      its contributors may be used to endorse or promote products
26  *      derived from this software without specific prior written
27  *      permission.
28  *
29  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
30  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
31  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
32  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
33  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
34  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
35  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
36  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
37  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
38  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
39  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
40  */
41 
42 /*
43  * verbs.c
44  *
45  * Encapsulates the major functions managing:
46  *  o adapters
47  *  o endpoints
48  *  o connections
49  *  o buffer memory
50  */
51 
52 #include <linux/interrupt.h>
53 #include <linux/slab.h>
54 #include <linux/sunrpc/addr.h>
55 #include <linux/sunrpc/svc_rdma.h>
56 #include <linux/log2.h>
57 
58 #include <asm-generic/barrier.h>
59 #include <asm/bitops.h>
60 
61 #include <rdma/ib_cm.h>
62 
63 #include "xprt_rdma.h"
64 #include <trace/events/rpcrdma.h>
65 
66 static int rpcrdma_sendctxs_create(struct rpcrdma_xprt *r_xprt);
67 static void rpcrdma_sendctxs_destroy(struct rpcrdma_xprt *r_xprt);
68 static void rpcrdma_sendctx_put_locked(struct rpcrdma_xprt *r_xprt,
69 				       struct rpcrdma_sendctx *sc);
70 static int rpcrdma_reqs_setup(struct rpcrdma_xprt *r_xprt);
71 static void rpcrdma_reqs_reset(struct rpcrdma_xprt *r_xprt);
72 static void rpcrdma_rep_destroy(struct rpcrdma_rep *rep);
73 static void rpcrdma_reps_unmap(struct rpcrdma_xprt *r_xprt);
74 static void rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt);
75 static void rpcrdma_mrs_destroy(struct rpcrdma_xprt *r_xprt);
76 static void rpcrdma_ep_get(struct rpcrdma_ep *ep);
77 static int rpcrdma_ep_put(struct rpcrdma_ep *ep);
78 static struct rpcrdma_regbuf *
79 rpcrdma_regbuf_alloc(size_t size, enum dma_data_direction direction);
80 static void rpcrdma_regbuf_dma_unmap(struct rpcrdma_regbuf *rb);
81 static void rpcrdma_regbuf_free(struct rpcrdma_regbuf *rb);
82 
83 /* Wait for outstanding transport work to finish. ib_drain_qp
84  * handles the drains in the wrong order for us, so open code
85  * them here.
86  */
rpcrdma_xprt_drain(struct rpcrdma_xprt * r_xprt)87 static void rpcrdma_xprt_drain(struct rpcrdma_xprt *r_xprt)
88 {
89 	struct rpcrdma_ep *ep = r_xprt->rx_ep;
90 	struct rdma_cm_id *id = ep->re_id;
91 
92 	/* Wait for rpcrdma_post_recvs() to leave its critical
93 	 * section.
94 	 */
95 	if (atomic_inc_return(&ep->re_receiving) > 1)
96 		wait_for_completion(&ep->re_done);
97 
98 	/* Flush Receives, then wait for deferred Reply work
99 	 * to complete.
100 	 */
101 	ib_drain_rq(id->qp);
102 
103 	/* Deferred Reply processing might have scheduled
104 	 * local invalidations.
105 	 */
106 	ib_drain_sq(id->qp);
107 
108 	rpcrdma_ep_put(ep);
109 }
110 
111 /* Ensure xprt_force_disconnect() is invoked exactly once when a
112  * connection is closed or lost. (The important thing is it needs
113  * to be invoked "at least" once).
114  */
rpcrdma_force_disconnect(struct rpcrdma_ep * ep)115 void rpcrdma_force_disconnect(struct rpcrdma_ep *ep)
116 {
117 	if (atomic_add_unless(&ep->re_force_disconnect, 1, 1))
118 		xprt_force_disconnect(ep->re_xprt);
119 }
120 
121 /**
122  * rpcrdma_flush_disconnect - Disconnect on flushed completion
123  * @r_xprt: transport to disconnect
124  * @wc: work completion entry
125  *
126  * Must be called in process context.
127  */
rpcrdma_flush_disconnect(struct rpcrdma_xprt * r_xprt,struct ib_wc * wc)128 void rpcrdma_flush_disconnect(struct rpcrdma_xprt *r_xprt, struct ib_wc *wc)
129 {
130 	if (wc->status != IB_WC_SUCCESS)
131 		rpcrdma_force_disconnect(r_xprt->rx_ep);
132 }
133 
134 /**
135  * rpcrdma_wc_send - Invoked by RDMA provider for each polled Send WC
136  * @cq:	completion queue
137  * @wc:	WCE for a completed Send WR
138  *
139  */
rpcrdma_wc_send(struct ib_cq * cq,struct ib_wc * wc)140 static void rpcrdma_wc_send(struct ib_cq *cq, struct ib_wc *wc)
141 {
142 	struct ib_cqe *cqe = wc->wr_cqe;
143 	struct rpcrdma_sendctx *sc =
144 		container_of(cqe, struct rpcrdma_sendctx, sc_cqe);
145 	struct rpcrdma_xprt *r_xprt = cq->cq_context;
146 
147 	/* WARNING: Only wr_cqe and status are reliable at this point */
148 	trace_xprtrdma_wc_send(wc, &sc->sc_cid);
149 	rpcrdma_sendctx_put_locked(r_xprt, sc);
150 	rpcrdma_flush_disconnect(r_xprt, wc);
151 }
152 
153 /**
154  * rpcrdma_wc_receive - Invoked by RDMA provider for each polled Receive WC
155  * @cq:	completion queue
156  * @wc:	WCE for a completed Receive WR
157  *
158  */
rpcrdma_wc_receive(struct ib_cq * cq,struct ib_wc * wc)159 static void rpcrdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
160 {
161 	struct ib_cqe *cqe = wc->wr_cqe;
162 	struct rpcrdma_rep *rep = container_of(cqe, struct rpcrdma_rep,
163 					       rr_cqe);
164 	struct rpcrdma_xprt *r_xprt = cq->cq_context;
165 
166 	/* WARNING: Only wr_cqe and status are reliable at this point */
167 	trace_xprtrdma_wc_receive(wc, &rep->rr_cid);
168 	--r_xprt->rx_ep->re_receive_count;
169 	if (wc->status != IB_WC_SUCCESS)
170 		goto out_flushed;
171 
172 	/* status == SUCCESS means all fields in wc are trustworthy */
173 	rpcrdma_set_xdrlen(&rep->rr_hdrbuf, wc->byte_len);
174 	rep->rr_wc_flags = wc->wc_flags;
175 	rep->rr_inv_rkey = wc->ex.invalidate_rkey;
176 
177 	ib_dma_sync_single_for_cpu(rdmab_device(rep->rr_rdmabuf),
178 				   rdmab_addr(rep->rr_rdmabuf),
179 				   wc->byte_len, DMA_FROM_DEVICE);
180 
181 	rpcrdma_reply_handler(rep);
182 	return;
183 
184 out_flushed:
185 	rpcrdma_flush_disconnect(r_xprt, wc);
186 	rpcrdma_rep_put(&r_xprt->rx_buf, rep);
187 }
188 
rpcrdma_update_cm_private(struct rpcrdma_ep * ep,struct rdma_conn_param * param)189 static void rpcrdma_update_cm_private(struct rpcrdma_ep *ep,
190 				      struct rdma_conn_param *param)
191 {
192 	const struct rpcrdma_connect_private *pmsg = param->private_data;
193 	unsigned int rsize, wsize;
194 
195 	/* Default settings for RPC-over-RDMA Version One */
196 	rsize = RPCRDMA_V1_DEF_INLINE_SIZE;
197 	wsize = RPCRDMA_V1_DEF_INLINE_SIZE;
198 
199 	if (pmsg &&
200 	    pmsg->cp_magic == rpcrdma_cmp_magic &&
201 	    pmsg->cp_version == RPCRDMA_CMP_VERSION) {
202 		rsize = rpcrdma_decode_buffer_size(pmsg->cp_send_size);
203 		wsize = rpcrdma_decode_buffer_size(pmsg->cp_recv_size);
204 	}
205 
206 	if (rsize < ep->re_inline_recv)
207 		ep->re_inline_recv = rsize;
208 	if (wsize < ep->re_inline_send)
209 		ep->re_inline_send = wsize;
210 
211 	rpcrdma_set_max_header_sizes(ep);
212 }
213 
214 /**
215  * rpcrdma_cm_event_handler - Handle RDMA CM events
216  * @id: rdma_cm_id on which an event has occurred
217  * @event: details of the event
218  *
219  * Called with @id's mutex held. Returns 1 if caller should
220  * destroy @id, otherwise 0.
221  */
222 static int
rpcrdma_cm_event_handler(struct rdma_cm_id * id,struct rdma_cm_event * event)223 rpcrdma_cm_event_handler(struct rdma_cm_id *id, struct rdma_cm_event *event)
224 {
225 	struct sockaddr *sap = (struct sockaddr *)&id->route.addr.dst_addr;
226 	struct rpcrdma_ep *ep = id->context;
227 
228 	might_sleep();
229 
230 	switch (event->event) {
231 	case RDMA_CM_EVENT_ADDR_RESOLVED:
232 	case RDMA_CM_EVENT_ROUTE_RESOLVED:
233 		ep->re_async_rc = 0;
234 		complete(&ep->re_done);
235 		return 0;
236 	case RDMA_CM_EVENT_ADDR_ERROR:
237 		ep->re_async_rc = -EPROTO;
238 		complete(&ep->re_done);
239 		return 0;
240 	case RDMA_CM_EVENT_ROUTE_ERROR:
241 		ep->re_async_rc = -ENETUNREACH;
242 		complete(&ep->re_done);
243 		return 0;
244 	case RDMA_CM_EVENT_DEVICE_REMOVAL:
245 		pr_info("rpcrdma: removing device %s for %pISpc\n",
246 			ep->re_id->device->name, sap);
247 		fallthrough;
248 	case RDMA_CM_EVENT_ADDR_CHANGE:
249 		ep->re_connect_status = -ENODEV;
250 		goto disconnected;
251 	case RDMA_CM_EVENT_ESTABLISHED:
252 		rpcrdma_ep_get(ep);
253 		ep->re_connect_status = 1;
254 		rpcrdma_update_cm_private(ep, &event->param.conn);
255 		trace_xprtrdma_inline_thresh(ep);
256 		wake_up_all(&ep->re_connect_wait);
257 		break;
258 	case RDMA_CM_EVENT_CONNECT_ERROR:
259 		ep->re_connect_status = -ENOTCONN;
260 		goto wake_connect_worker;
261 	case RDMA_CM_EVENT_UNREACHABLE:
262 		ep->re_connect_status = -ENETUNREACH;
263 		goto wake_connect_worker;
264 	case RDMA_CM_EVENT_REJECTED:
265 		ep->re_connect_status = -ECONNREFUSED;
266 		if (event->status == IB_CM_REJ_STALE_CONN)
267 			ep->re_connect_status = -ENOTCONN;
268 wake_connect_worker:
269 		wake_up_all(&ep->re_connect_wait);
270 		return 0;
271 	case RDMA_CM_EVENT_DISCONNECTED:
272 		ep->re_connect_status = -ECONNABORTED;
273 disconnected:
274 		rpcrdma_force_disconnect(ep);
275 		return rpcrdma_ep_put(ep);
276 	default:
277 		break;
278 	}
279 
280 	return 0;
281 }
282 
rpcrdma_create_id(struct rpcrdma_xprt * r_xprt,struct rpcrdma_ep * ep)283 static struct rdma_cm_id *rpcrdma_create_id(struct rpcrdma_xprt *r_xprt,
284 					    struct rpcrdma_ep *ep)
285 {
286 	unsigned long wtimeout = msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1;
287 	struct rpc_xprt *xprt = &r_xprt->rx_xprt;
288 	struct rdma_cm_id *id;
289 	int rc;
290 
291 	init_completion(&ep->re_done);
292 
293 	id = rdma_create_id(xprt->xprt_net, rpcrdma_cm_event_handler, ep,
294 			    RDMA_PS_TCP, IB_QPT_RC);
295 	if (IS_ERR(id))
296 		return id;
297 
298 	ep->re_async_rc = -ETIMEDOUT;
299 	rc = rdma_resolve_addr(id, NULL, (struct sockaddr *)&xprt->addr,
300 			       RDMA_RESOLVE_TIMEOUT);
301 	if (rc)
302 		goto out;
303 	rc = wait_for_completion_interruptible_timeout(&ep->re_done, wtimeout);
304 	if (rc < 0)
305 		goto out;
306 
307 	rc = ep->re_async_rc;
308 	if (rc)
309 		goto out;
310 
311 	ep->re_async_rc = -ETIMEDOUT;
312 	rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT);
313 	if (rc)
314 		goto out;
315 	rc = wait_for_completion_interruptible_timeout(&ep->re_done, wtimeout);
316 	if (rc < 0)
317 		goto out;
318 	rc = ep->re_async_rc;
319 	if (rc)
320 		goto out;
321 
322 	return id;
323 
324 out:
325 	rdma_destroy_id(id);
326 	return ERR_PTR(rc);
327 }
328 
rpcrdma_ep_destroy(struct kref * kref)329 static void rpcrdma_ep_destroy(struct kref *kref)
330 {
331 	struct rpcrdma_ep *ep = container_of(kref, struct rpcrdma_ep, re_kref);
332 
333 	if (ep->re_id->qp) {
334 		rdma_destroy_qp(ep->re_id);
335 		ep->re_id->qp = NULL;
336 	}
337 
338 	if (ep->re_attr.recv_cq)
339 		ib_free_cq(ep->re_attr.recv_cq);
340 	ep->re_attr.recv_cq = NULL;
341 	if (ep->re_attr.send_cq)
342 		ib_free_cq(ep->re_attr.send_cq);
343 	ep->re_attr.send_cq = NULL;
344 
345 	if (ep->re_pd)
346 		ib_dealloc_pd(ep->re_pd);
347 	ep->re_pd = NULL;
348 
349 	kfree(ep);
350 	module_put(THIS_MODULE);
351 }
352 
rpcrdma_ep_get(struct rpcrdma_ep * ep)353 static noinline void rpcrdma_ep_get(struct rpcrdma_ep *ep)
354 {
355 	kref_get(&ep->re_kref);
356 }
357 
358 /* Returns:
359  *     %0 if @ep still has a positive kref count, or
360  *     %1 if @ep was destroyed successfully.
361  */
rpcrdma_ep_put(struct rpcrdma_ep * ep)362 static noinline int rpcrdma_ep_put(struct rpcrdma_ep *ep)
363 {
364 	return kref_put(&ep->re_kref, rpcrdma_ep_destroy);
365 }
366 
rpcrdma_ep_create(struct rpcrdma_xprt * r_xprt)367 static int rpcrdma_ep_create(struct rpcrdma_xprt *r_xprt)
368 {
369 	struct rpcrdma_connect_private *pmsg;
370 	struct ib_device *device;
371 	struct rdma_cm_id *id;
372 	struct rpcrdma_ep *ep;
373 	int rc;
374 
375 	ep = kzalloc(sizeof(*ep), XPRTRDMA_GFP_FLAGS);
376 	if (!ep)
377 		return -ENOTCONN;
378 	ep->re_xprt = &r_xprt->rx_xprt;
379 	kref_init(&ep->re_kref);
380 
381 	id = rpcrdma_create_id(r_xprt, ep);
382 	if (IS_ERR(id)) {
383 		kfree(ep);
384 		return PTR_ERR(id);
385 	}
386 	__module_get(THIS_MODULE);
387 	device = id->device;
388 	ep->re_id = id;
389 	reinit_completion(&ep->re_done);
390 
391 	ep->re_max_requests = r_xprt->rx_xprt.max_reqs;
392 	ep->re_inline_send = xprt_rdma_max_inline_write;
393 	ep->re_inline_recv = xprt_rdma_max_inline_read;
394 	rc = frwr_query_device(ep, device);
395 	if (rc)
396 		goto out_destroy;
397 
398 	r_xprt->rx_buf.rb_max_requests = cpu_to_be32(ep->re_max_requests);
399 
400 	ep->re_attr.srq = NULL;
401 	ep->re_attr.cap.max_inline_data = 0;
402 	ep->re_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
403 	ep->re_attr.qp_type = IB_QPT_RC;
404 	ep->re_attr.port_num = ~0;
405 
406 	ep->re_send_batch = ep->re_max_requests >> 3;
407 	ep->re_send_count = ep->re_send_batch;
408 	init_waitqueue_head(&ep->re_connect_wait);
409 
410 	ep->re_attr.send_cq = ib_alloc_cq_any(device, r_xprt,
411 					      ep->re_attr.cap.max_send_wr,
412 					      IB_POLL_WORKQUEUE);
413 	if (IS_ERR(ep->re_attr.send_cq)) {
414 		rc = PTR_ERR(ep->re_attr.send_cq);
415 		ep->re_attr.send_cq = NULL;
416 		goto out_destroy;
417 	}
418 
419 	ep->re_attr.recv_cq = ib_alloc_cq_any(device, r_xprt,
420 					      ep->re_attr.cap.max_recv_wr,
421 					      IB_POLL_WORKQUEUE);
422 	if (IS_ERR(ep->re_attr.recv_cq)) {
423 		rc = PTR_ERR(ep->re_attr.recv_cq);
424 		ep->re_attr.recv_cq = NULL;
425 		goto out_destroy;
426 	}
427 	ep->re_receive_count = 0;
428 
429 	/* Initialize cma parameters */
430 	memset(&ep->re_remote_cma, 0, sizeof(ep->re_remote_cma));
431 
432 	/* Prepare RDMA-CM private message */
433 	pmsg = &ep->re_cm_private;
434 	pmsg->cp_magic = rpcrdma_cmp_magic;
435 	pmsg->cp_version = RPCRDMA_CMP_VERSION;
436 	pmsg->cp_flags |= RPCRDMA_CMP_F_SND_W_INV_OK;
437 	pmsg->cp_send_size = rpcrdma_encode_buffer_size(ep->re_inline_send);
438 	pmsg->cp_recv_size = rpcrdma_encode_buffer_size(ep->re_inline_recv);
439 	ep->re_remote_cma.private_data = pmsg;
440 	ep->re_remote_cma.private_data_len = sizeof(*pmsg);
441 
442 	/* Client offers RDMA Read but does not initiate */
443 	ep->re_remote_cma.initiator_depth = 0;
444 	ep->re_remote_cma.responder_resources =
445 		min_t(int, U8_MAX, device->attrs.max_qp_rd_atom);
446 
447 	/* Limit transport retries so client can detect server
448 	 * GID changes quickly. RPC layer handles re-establishing
449 	 * transport connection and retransmission.
450 	 */
451 	ep->re_remote_cma.retry_count = 6;
452 
453 	/* RPC-over-RDMA handles its own flow control. In addition,
454 	 * make all RNR NAKs visible so we know that RPC-over-RDMA
455 	 * flow control is working correctly (no NAKs should be seen).
456 	 */
457 	ep->re_remote_cma.flow_control = 0;
458 	ep->re_remote_cma.rnr_retry_count = 0;
459 
460 	ep->re_pd = ib_alloc_pd(device, 0);
461 	if (IS_ERR(ep->re_pd)) {
462 		rc = PTR_ERR(ep->re_pd);
463 		ep->re_pd = NULL;
464 		goto out_destroy;
465 	}
466 
467 	rc = rdma_create_qp(id, ep->re_pd, &ep->re_attr);
468 	if (rc)
469 		goto out_destroy;
470 
471 	r_xprt->rx_ep = ep;
472 	return 0;
473 
474 out_destroy:
475 	rpcrdma_ep_put(ep);
476 	rdma_destroy_id(id);
477 	return rc;
478 }
479 
480 /**
481  * rpcrdma_xprt_connect - Connect an unconnected transport
482  * @r_xprt: controlling transport instance
483  *
484  * Returns 0 on success or a negative errno.
485  */
rpcrdma_xprt_connect(struct rpcrdma_xprt * r_xprt)486 int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt)
487 {
488 	struct rpc_xprt *xprt = &r_xprt->rx_xprt;
489 	struct rpcrdma_ep *ep;
490 	int rc;
491 
492 	rc = rpcrdma_ep_create(r_xprt);
493 	if (rc)
494 		return rc;
495 	ep = r_xprt->rx_ep;
496 
497 	xprt_clear_connected(xprt);
498 	rpcrdma_reset_cwnd(r_xprt);
499 
500 	/* Bump the ep's reference count while there are
501 	 * outstanding Receives.
502 	 */
503 	rpcrdma_ep_get(ep);
504 	rpcrdma_post_recvs(r_xprt, 1, true);
505 
506 	rc = rdma_connect(ep->re_id, &ep->re_remote_cma);
507 	if (rc)
508 		goto out;
509 
510 	if (xprt->reestablish_timeout < RPCRDMA_INIT_REEST_TO)
511 		xprt->reestablish_timeout = RPCRDMA_INIT_REEST_TO;
512 	wait_event_interruptible(ep->re_connect_wait,
513 				 ep->re_connect_status != 0);
514 	if (ep->re_connect_status <= 0) {
515 		rc = ep->re_connect_status;
516 		goto out;
517 	}
518 
519 	rc = rpcrdma_sendctxs_create(r_xprt);
520 	if (rc) {
521 		rc = -ENOTCONN;
522 		goto out;
523 	}
524 
525 	rc = rpcrdma_reqs_setup(r_xprt);
526 	if (rc) {
527 		rc = -ENOTCONN;
528 		goto out;
529 	}
530 	rpcrdma_mrs_create(r_xprt);
531 	frwr_wp_create(r_xprt);
532 
533 out:
534 	trace_xprtrdma_connect(r_xprt, rc);
535 	return rc;
536 }
537 
538 /**
539  * rpcrdma_xprt_disconnect - Disconnect underlying transport
540  * @r_xprt: controlling transport instance
541  *
542  * Caller serializes. Either the transport send lock is held,
543  * or we're being called to destroy the transport.
544  *
545  * On return, @r_xprt is completely divested of all hardware
546  * resources and prepared for the next ->connect operation.
547  */
rpcrdma_xprt_disconnect(struct rpcrdma_xprt * r_xprt)548 void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt)
549 {
550 	struct rpcrdma_ep *ep = r_xprt->rx_ep;
551 	struct rdma_cm_id *id;
552 	int rc;
553 
554 	if (!ep)
555 		return;
556 
557 	id = ep->re_id;
558 	rc = rdma_disconnect(id);
559 	trace_xprtrdma_disconnect(r_xprt, rc);
560 
561 	rpcrdma_xprt_drain(r_xprt);
562 	rpcrdma_reps_unmap(r_xprt);
563 	rpcrdma_reqs_reset(r_xprt);
564 	rpcrdma_mrs_destroy(r_xprt);
565 	rpcrdma_sendctxs_destroy(r_xprt);
566 
567 	if (rpcrdma_ep_put(ep))
568 		rdma_destroy_id(id);
569 
570 	r_xprt->rx_ep = NULL;
571 }
572 
573 /* Fixed-size circular FIFO queue. This implementation is wait-free and
574  * lock-free.
575  *
576  * Consumer is the code path that posts Sends. This path dequeues a
577  * sendctx for use by a Send operation. Multiple consumer threads
578  * are serialized by the RPC transport lock, which allows only one
579  * ->send_request call at a time.
580  *
581  * Producer is the code path that handles Send completions. This path
582  * enqueues a sendctx that has been completed. Multiple producer
583  * threads are serialized by the ib_poll_cq() function.
584  */
585 
586 /* rpcrdma_sendctxs_destroy() assumes caller has already quiesced
587  * queue activity, and rpcrdma_xprt_drain has flushed all remaining
588  * Send requests.
589  */
rpcrdma_sendctxs_destroy(struct rpcrdma_xprt * r_xprt)590 static void rpcrdma_sendctxs_destroy(struct rpcrdma_xprt *r_xprt)
591 {
592 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
593 	unsigned long i;
594 
595 	if (!buf->rb_sc_ctxs)
596 		return;
597 	for (i = 0; i <= buf->rb_sc_last; i++)
598 		kfree(buf->rb_sc_ctxs[i]);
599 	kfree(buf->rb_sc_ctxs);
600 	buf->rb_sc_ctxs = NULL;
601 }
602 
rpcrdma_sendctx_create(struct rpcrdma_ep * ep)603 static struct rpcrdma_sendctx *rpcrdma_sendctx_create(struct rpcrdma_ep *ep)
604 {
605 	struct rpcrdma_sendctx *sc;
606 
607 	sc = kzalloc(struct_size(sc, sc_sges, ep->re_attr.cap.max_send_sge),
608 		     XPRTRDMA_GFP_FLAGS);
609 	if (!sc)
610 		return NULL;
611 
612 	sc->sc_cqe.done = rpcrdma_wc_send;
613 	sc->sc_cid.ci_queue_id = ep->re_attr.send_cq->res.id;
614 	sc->sc_cid.ci_completion_id =
615 		atomic_inc_return(&ep->re_completion_ids);
616 	return sc;
617 }
618 
rpcrdma_sendctxs_create(struct rpcrdma_xprt * r_xprt)619 static int rpcrdma_sendctxs_create(struct rpcrdma_xprt *r_xprt)
620 {
621 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
622 	struct rpcrdma_sendctx *sc;
623 	unsigned long i;
624 
625 	/* Maximum number of concurrent outstanding Send WRs. Capping
626 	 * the circular queue size stops Send Queue overflow by causing
627 	 * the ->send_request call to fail temporarily before too many
628 	 * Sends are posted.
629 	 */
630 	i = r_xprt->rx_ep->re_max_requests + RPCRDMA_MAX_BC_REQUESTS;
631 	buf->rb_sc_ctxs = kcalloc(i, sizeof(sc), XPRTRDMA_GFP_FLAGS);
632 	if (!buf->rb_sc_ctxs)
633 		return -ENOMEM;
634 
635 	buf->rb_sc_last = i - 1;
636 	for (i = 0; i <= buf->rb_sc_last; i++) {
637 		sc = rpcrdma_sendctx_create(r_xprt->rx_ep);
638 		if (!sc)
639 			return -ENOMEM;
640 
641 		buf->rb_sc_ctxs[i] = sc;
642 	}
643 
644 	buf->rb_sc_head = 0;
645 	buf->rb_sc_tail = 0;
646 	return 0;
647 }
648 
649 /* The sendctx queue is not guaranteed to have a size that is a
650  * power of two, thus the helpers in circ_buf.h cannot be used.
651  * The other option is to use modulus (%), which can be expensive.
652  */
rpcrdma_sendctx_next(struct rpcrdma_buffer * buf,unsigned long item)653 static unsigned long rpcrdma_sendctx_next(struct rpcrdma_buffer *buf,
654 					  unsigned long item)
655 {
656 	return likely(item < buf->rb_sc_last) ? item + 1 : 0;
657 }
658 
659 /**
660  * rpcrdma_sendctx_get_locked - Acquire a send context
661  * @r_xprt: controlling transport instance
662  *
663  * Returns pointer to a free send completion context; or NULL if
664  * the queue is empty.
665  *
666  * Usage: Called to acquire an SGE array before preparing a Send WR.
667  *
668  * The caller serializes calls to this function (per transport), and
669  * provides an effective memory barrier that flushes the new value
670  * of rb_sc_head.
671  */
rpcrdma_sendctx_get_locked(struct rpcrdma_xprt * r_xprt)672 struct rpcrdma_sendctx *rpcrdma_sendctx_get_locked(struct rpcrdma_xprt *r_xprt)
673 {
674 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
675 	struct rpcrdma_sendctx *sc;
676 	unsigned long next_head;
677 
678 	next_head = rpcrdma_sendctx_next(buf, buf->rb_sc_head);
679 
680 	if (next_head == READ_ONCE(buf->rb_sc_tail))
681 		goto out_emptyq;
682 
683 	/* ORDER: item must be accessed _before_ head is updated */
684 	sc = buf->rb_sc_ctxs[next_head];
685 
686 	/* Releasing the lock in the caller acts as a memory
687 	 * barrier that flushes rb_sc_head.
688 	 */
689 	buf->rb_sc_head = next_head;
690 
691 	return sc;
692 
693 out_emptyq:
694 	/* The queue is "empty" if there have not been enough Send
695 	 * completions recently. This is a sign the Send Queue is
696 	 * backing up. Cause the caller to pause and try again.
697 	 */
698 	xprt_wait_for_buffer_space(&r_xprt->rx_xprt);
699 	r_xprt->rx_stats.empty_sendctx_q++;
700 	return NULL;
701 }
702 
703 /**
704  * rpcrdma_sendctx_put_locked - Release a send context
705  * @r_xprt: controlling transport instance
706  * @sc: send context to release
707  *
708  * Usage: Called from Send completion to return a sendctxt
709  * to the queue.
710  *
711  * The caller serializes calls to this function (per transport).
712  */
rpcrdma_sendctx_put_locked(struct rpcrdma_xprt * r_xprt,struct rpcrdma_sendctx * sc)713 static void rpcrdma_sendctx_put_locked(struct rpcrdma_xprt *r_xprt,
714 				       struct rpcrdma_sendctx *sc)
715 {
716 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
717 	unsigned long next_tail;
718 
719 	/* Unmap SGEs of previously completed but unsignaled
720 	 * Sends by walking up the queue until @sc is found.
721 	 */
722 	next_tail = buf->rb_sc_tail;
723 	do {
724 		next_tail = rpcrdma_sendctx_next(buf, next_tail);
725 
726 		/* ORDER: item must be accessed _before_ tail is updated */
727 		rpcrdma_sendctx_unmap(buf->rb_sc_ctxs[next_tail]);
728 
729 	} while (buf->rb_sc_ctxs[next_tail] != sc);
730 
731 	/* Paired with READ_ONCE */
732 	smp_store_release(&buf->rb_sc_tail, next_tail);
733 
734 	xprt_write_space(&r_xprt->rx_xprt);
735 }
736 
737 static void
rpcrdma_mrs_create(struct rpcrdma_xprt * r_xprt)738 rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt)
739 {
740 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
741 	struct rpcrdma_ep *ep = r_xprt->rx_ep;
742 	struct ib_device *device = ep->re_id->device;
743 	unsigned int count;
744 
745 	/* Try to allocate enough to perform one full-sized I/O */
746 	for (count = 0; count < ep->re_max_rdma_segs; count++) {
747 		struct rpcrdma_mr *mr;
748 		int rc;
749 
750 		mr = kzalloc_node(sizeof(*mr), XPRTRDMA_GFP_FLAGS,
751 				  ibdev_to_node(device));
752 		if (!mr)
753 			break;
754 
755 		rc = frwr_mr_init(r_xprt, mr);
756 		if (rc) {
757 			kfree(mr);
758 			break;
759 		}
760 
761 		spin_lock(&buf->rb_lock);
762 		rpcrdma_mr_push(mr, &buf->rb_mrs);
763 		list_add(&mr->mr_all, &buf->rb_all_mrs);
764 		spin_unlock(&buf->rb_lock);
765 	}
766 
767 	r_xprt->rx_stats.mrs_allocated += count;
768 	trace_xprtrdma_createmrs(r_xprt, count);
769 }
770 
771 static void
rpcrdma_mr_refresh_worker(struct work_struct * work)772 rpcrdma_mr_refresh_worker(struct work_struct *work)
773 {
774 	struct rpcrdma_buffer *buf = container_of(work, struct rpcrdma_buffer,
775 						  rb_refresh_worker);
776 	struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt,
777 						   rx_buf);
778 
779 	rpcrdma_mrs_create(r_xprt);
780 	xprt_write_space(&r_xprt->rx_xprt);
781 }
782 
783 /**
784  * rpcrdma_mrs_refresh - Wake the MR refresh worker
785  * @r_xprt: controlling transport instance
786  *
787  */
rpcrdma_mrs_refresh(struct rpcrdma_xprt * r_xprt)788 void rpcrdma_mrs_refresh(struct rpcrdma_xprt *r_xprt)
789 {
790 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
791 	struct rpcrdma_ep *ep = r_xprt->rx_ep;
792 
793 	/* If there is no underlying connection, it's no use
794 	 * to wake the refresh worker.
795 	 */
796 	if (ep->re_connect_status != 1)
797 		return;
798 	queue_work(system_highpri_wq, &buf->rb_refresh_worker);
799 }
800 
801 /**
802  * rpcrdma_req_create - Allocate an rpcrdma_req object
803  * @r_xprt: controlling r_xprt
804  * @size: initial size, in bytes, of send and receive buffers
805  *
806  * Returns an allocated and fully initialized rpcrdma_req or NULL.
807  */
rpcrdma_req_create(struct rpcrdma_xprt * r_xprt,size_t size)808 struct rpcrdma_req *rpcrdma_req_create(struct rpcrdma_xprt *r_xprt,
809 				       size_t size)
810 {
811 	struct rpcrdma_buffer *buffer = &r_xprt->rx_buf;
812 	struct rpcrdma_req *req;
813 
814 	req = kzalloc(sizeof(*req), XPRTRDMA_GFP_FLAGS);
815 	if (req == NULL)
816 		goto out1;
817 
818 	req->rl_sendbuf = rpcrdma_regbuf_alloc(size, DMA_TO_DEVICE);
819 	if (!req->rl_sendbuf)
820 		goto out2;
821 
822 	req->rl_recvbuf = rpcrdma_regbuf_alloc(size, DMA_NONE);
823 	if (!req->rl_recvbuf)
824 		goto out3;
825 
826 	INIT_LIST_HEAD(&req->rl_free_mrs);
827 	INIT_LIST_HEAD(&req->rl_registered);
828 	spin_lock(&buffer->rb_lock);
829 	list_add(&req->rl_all, &buffer->rb_allreqs);
830 	spin_unlock(&buffer->rb_lock);
831 	return req;
832 
833 out3:
834 	rpcrdma_regbuf_free(req->rl_sendbuf);
835 out2:
836 	kfree(req);
837 out1:
838 	return NULL;
839 }
840 
841 /**
842  * rpcrdma_req_setup - Per-connection instance setup of an rpcrdma_req object
843  * @r_xprt: controlling transport instance
844  * @req: rpcrdma_req object to set up
845  *
846  * Returns zero on success, and a negative errno on failure.
847  */
rpcrdma_req_setup(struct rpcrdma_xprt * r_xprt,struct rpcrdma_req * req)848 int rpcrdma_req_setup(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req)
849 {
850 	struct rpcrdma_regbuf *rb;
851 	size_t maxhdrsize;
852 
853 	/* Compute maximum header buffer size in bytes */
854 	maxhdrsize = rpcrdma_fixed_maxsz + 3 +
855 		     r_xprt->rx_ep->re_max_rdma_segs * rpcrdma_readchunk_maxsz;
856 	maxhdrsize *= sizeof(__be32);
857 	rb = rpcrdma_regbuf_alloc(__roundup_pow_of_two(maxhdrsize),
858 				  DMA_TO_DEVICE);
859 	if (!rb)
860 		goto out;
861 
862 	if (!__rpcrdma_regbuf_dma_map(r_xprt, rb))
863 		goto out_free;
864 
865 	req->rl_rdmabuf = rb;
866 	xdr_buf_init(&req->rl_hdrbuf, rdmab_data(rb), rdmab_length(rb));
867 	return 0;
868 
869 out_free:
870 	rpcrdma_regbuf_free(rb);
871 out:
872 	return -ENOMEM;
873 }
874 
875 /* ASSUMPTION: the rb_allreqs list is stable for the duration,
876  * and thus can be walked without holding rb_lock. Eg. the
877  * caller is holding the transport send lock to exclude
878  * device removal or disconnection.
879  */
rpcrdma_reqs_setup(struct rpcrdma_xprt * r_xprt)880 static int rpcrdma_reqs_setup(struct rpcrdma_xprt *r_xprt)
881 {
882 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
883 	struct rpcrdma_req *req;
884 	int rc;
885 
886 	list_for_each_entry(req, &buf->rb_allreqs, rl_all) {
887 		rc = rpcrdma_req_setup(r_xprt, req);
888 		if (rc)
889 			return rc;
890 	}
891 	return 0;
892 }
893 
rpcrdma_req_reset(struct rpcrdma_req * req)894 static void rpcrdma_req_reset(struct rpcrdma_req *req)
895 {
896 	/* Credits are valid for only one connection */
897 	req->rl_slot.rq_cong = 0;
898 
899 	rpcrdma_regbuf_free(req->rl_rdmabuf);
900 	req->rl_rdmabuf = NULL;
901 
902 	rpcrdma_regbuf_dma_unmap(req->rl_sendbuf);
903 	rpcrdma_regbuf_dma_unmap(req->rl_recvbuf);
904 
905 	frwr_reset(req);
906 }
907 
908 /* ASSUMPTION: the rb_allreqs list is stable for the duration,
909  * and thus can be walked without holding rb_lock. Eg. the
910  * caller is holding the transport send lock to exclude
911  * device removal or disconnection.
912  */
rpcrdma_reqs_reset(struct rpcrdma_xprt * r_xprt)913 static void rpcrdma_reqs_reset(struct rpcrdma_xprt *r_xprt)
914 {
915 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
916 	struct rpcrdma_req *req;
917 
918 	list_for_each_entry(req, &buf->rb_allreqs, rl_all)
919 		rpcrdma_req_reset(req);
920 }
921 
922 static noinline
rpcrdma_rep_create(struct rpcrdma_xprt * r_xprt,bool temp)923 struct rpcrdma_rep *rpcrdma_rep_create(struct rpcrdma_xprt *r_xprt,
924 				       bool temp)
925 {
926 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
927 	struct rpcrdma_rep *rep;
928 
929 	rep = kzalloc(sizeof(*rep), XPRTRDMA_GFP_FLAGS);
930 	if (rep == NULL)
931 		goto out;
932 
933 	rep->rr_rdmabuf = rpcrdma_regbuf_alloc(r_xprt->rx_ep->re_inline_recv,
934 					       DMA_FROM_DEVICE);
935 	if (!rep->rr_rdmabuf)
936 		goto out_free;
937 
938 	rep->rr_cid.ci_completion_id =
939 		atomic_inc_return(&r_xprt->rx_ep->re_completion_ids);
940 
941 	xdr_buf_init(&rep->rr_hdrbuf, rdmab_data(rep->rr_rdmabuf),
942 		     rdmab_length(rep->rr_rdmabuf));
943 	rep->rr_cqe.done = rpcrdma_wc_receive;
944 	rep->rr_rxprt = r_xprt;
945 	rep->rr_recv_wr.next = NULL;
946 	rep->rr_recv_wr.wr_cqe = &rep->rr_cqe;
947 	rep->rr_recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov;
948 	rep->rr_recv_wr.num_sge = 1;
949 	rep->rr_temp = temp;
950 
951 	spin_lock(&buf->rb_lock);
952 	list_add(&rep->rr_all, &buf->rb_all_reps);
953 	spin_unlock(&buf->rb_lock);
954 	return rep;
955 
956 out_free:
957 	kfree(rep);
958 out:
959 	return NULL;
960 }
961 
rpcrdma_rep_free(struct rpcrdma_rep * rep)962 static void rpcrdma_rep_free(struct rpcrdma_rep *rep)
963 {
964 	rpcrdma_regbuf_free(rep->rr_rdmabuf);
965 	kfree(rep);
966 }
967 
rpcrdma_rep_destroy(struct rpcrdma_rep * rep)968 static void rpcrdma_rep_destroy(struct rpcrdma_rep *rep)
969 {
970 	struct rpcrdma_buffer *buf = &rep->rr_rxprt->rx_buf;
971 
972 	spin_lock(&buf->rb_lock);
973 	list_del(&rep->rr_all);
974 	spin_unlock(&buf->rb_lock);
975 
976 	rpcrdma_rep_free(rep);
977 }
978 
rpcrdma_rep_get_locked(struct rpcrdma_buffer * buf)979 static struct rpcrdma_rep *rpcrdma_rep_get_locked(struct rpcrdma_buffer *buf)
980 {
981 	struct llist_node *node;
982 
983 	/* Calls to llist_del_first are required to be serialized */
984 	node = llist_del_first(&buf->rb_free_reps);
985 	if (!node)
986 		return NULL;
987 	return llist_entry(node, struct rpcrdma_rep, rr_node);
988 }
989 
990 /**
991  * rpcrdma_rep_put - Release rpcrdma_rep back to free list
992  * @buf: buffer pool
993  * @rep: rep to release
994  *
995  */
rpcrdma_rep_put(struct rpcrdma_buffer * buf,struct rpcrdma_rep * rep)996 void rpcrdma_rep_put(struct rpcrdma_buffer *buf, struct rpcrdma_rep *rep)
997 {
998 	llist_add(&rep->rr_node, &buf->rb_free_reps);
999 }
1000 
1001 /* Caller must ensure the QP is quiescent (RQ is drained) before
1002  * invoking this function, to guarantee rb_all_reps is not
1003  * changing.
1004  */
rpcrdma_reps_unmap(struct rpcrdma_xprt * r_xprt)1005 static void rpcrdma_reps_unmap(struct rpcrdma_xprt *r_xprt)
1006 {
1007 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1008 	struct rpcrdma_rep *rep;
1009 
1010 	list_for_each_entry(rep, &buf->rb_all_reps, rr_all) {
1011 		rpcrdma_regbuf_dma_unmap(rep->rr_rdmabuf);
1012 		rep->rr_temp = true;	/* Mark this rep for destruction */
1013 	}
1014 }
1015 
rpcrdma_reps_destroy(struct rpcrdma_buffer * buf)1016 static void rpcrdma_reps_destroy(struct rpcrdma_buffer *buf)
1017 {
1018 	struct rpcrdma_rep *rep;
1019 
1020 	spin_lock(&buf->rb_lock);
1021 	while ((rep = list_first_entry_or_null(&buf->rb_all_reps,
1022 					       struct rpcrdma_rep,
1023 					       rr_all)) != NULL) {
1024 		list_del(&rep->rr_all);
1025 		spin_unlock(&buf->rb_lock);
1026 
1027 		rpcrdma_rep_free(rep);
1028 
1029 		spin_lock(&buf->rb_lock);
1030 	}
1031 	spin_unlock(&buf->rb_lock);
1032 }
1033 
1034 /**
1035  * rpcrdma_buffer_create - Create initial set of req/rep objects
1036  * @r_xprt: transport instance to (re)initialize
1037  *
1038  * Returns zero on success, otherwise a negative errno.
1039  */
rpcrdma_buffer_create(struct rpcrdma_xprt * r_xprt)1040 int rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
1041 {
1042 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1043 	int i, rc;
1044 
1045 	buf->rb_bc_srv_max_requests = 0;
1046 	spin_lock_init(&buf->rb_lock);
1047 	INIT_LIST_HEAD(&buf->rb_mrs);
1048 	INIT_LIST_HEAD(&buf->rb_all_mrs);
1049 	INIT_WORK(&buf->rb_refresh_worker, rpcrdma_mr_refresh_worker);
1050 
1051 	INIT_LIST_HEAD(&buf->rb_send_bufs);
1052 	INIT_LIST_HEAD(&buf->rb_allreqs);
1053 	INIT_LIST_HEAD(&buf->rb_all_reps);
1054 
1055 	rc = -ENOMEM;
1056 	for (i = 0; i < r_xprt->rx_xprt.max_reqs; i++) {
1057 		struct rpcrdma_req *req;
1058 
1059 		req = rpcrdma_req_create(r_xprt,
1060 					 RPCRDMA_V1_DEF_INLINE_SIZE * 2);
1061 		if (!req)
1062 			goto out;
1063 		list_add(&req->rl_list, &buf->rb_send_bufs);
1064 	}
1065 
1066 	init_llist_head(&buf->rb_free_reps);
1067 
1068 	return 0;
1069 out:
1070 	rpcrdma_buffer_destroy(buf);
1071 	return rc;
1072 }
1073 
1074 /**
1075  * rpcrdma_req_destroy - Destroy an rpcrdma_req object
1076  * @req: unused object to be destroyed
1077  *
1078  * Relies on caller holding the transport send lock to protect
1079  * removing req->rl_all from buf->rb_all_reqs safely.
1080  */
rpcrdma_req_destroy(struct rpcrdma_req * req)1081 void rpcrdma_req_destroy(struct rpcrdma_req *req)
1082 {
1083 	struct rpcrdma_mr *mr;
1084 
1085 	list_del(&req->rl_all);
1086 
1087 	while ((mr = rpcrdma_mr_pop(&req->rl_free_mrs))) {
1088 		struct rpcrdma_buffer *buf = &mr->mr_xprt->rx_buf;
1089 
1090 		spin_lock(&buf->rb_lock);
1091 		list_del(&mr->mr_all);
1092 		spin_unlock(&buf->rb_lock);
1093 
1094 		frwr_mr_release(mr);
1095 	}
1096 
1097 	rpcrdma_regbuf_free(req->rl_recvbuf);
1098 	rpcrdma_regbuf_free(req->rl_sendbuf);
1099 	rpcrdma_regbuf_free(req->rl_rdmabuf);
1100 	kfree(req);
1101 }
1102 
1103 /**
1104  * rpcrdma_mrs_destroy - Release all of a transport's MRs
1105  * @r_xprt: controlling transport instance
1106  *
1107  * Relies on caller holding the transport send lock to protect
1108  * removing mr->mr_list from req->rl_free_mrs safely.
1109  */
rpcrdma_mrs_destroy(struct rpcrdma_xprt * r_xprt)1110 static void rpcrdma_mrs_destroy(struct rpcrdma_xprt *r_xprt)
1111 {
1112 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1113 	struct rpcrdma_mr *mr;
1114 
1115 	cancel_work_sync(&buf->rb_refresh_worker);
1116 
1117 	spin_lock(&buf->rb_lock);
1118 	while ((mr = list_first_entry_or_null(&buf->rb_all_mrs,
1119 					      struct rpcrdma_mr,
1120 					      mr_all)) != NULL) {
1121 		list_del(&mr->mr_list);
1122 		list_del(&mr->mr_all);
1123 		spin_unlock(&buf->rb_lock);
1124 
1125 		frwr_mr_release(mr);
1126 
1127 		spin_lock(&buf->rb_lock);
1128 	}
1129 	spin_unlock(&buf->rb_lock);
1130 }
1131 
1132 /**
1133  * rpcrdma_buffer_destroy - Release all hw resources
1134  * @buf: root control block for resources
1135  *
1136  * ORDERING: relies on a prior rpcrdma_xprt_drain :
1137  * - No more Send or Receive completions can occur
1138  * - All MRs, reps, and reqs are returned to their free lists
1139  */
1140 void
rpcrdma_buffer_destroy(struct rpcrdma_buffer * buf)1141 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf)
1142 {
1143 	rpcrdma_reps_destroy(buf);
1144 
1145 	while (!list_empty(&buf->rb_send_bufs)) {
1146 		struct rpcrdma_req *req;
1147 
1148 		req = list_first_entry(&buf->rb_send_bufs,
1149 				       struct rpcrdma_req, rl_list);
1150 		list_del(&req->rl_list);
1151 		rpcrdma_req_destroy(req);
1152 	}
1153 }
1154 
1155 /**
1156  * rpcrdma_mr_get - Allocate an rpcrdma_mr object
1157  * @r_xprt: controlling transport
1158  *
1159  * Returns an initialized rpcrdma_mr or NULL if no free
1160  * rpcrdma_mr objects are available.
1161  */
1162 struct rpcrdma_mr *
rpcrdma_mr_get(struct rpcrdma_xprt * r_xprt)1163 rpcrdma_mr_get(struct rpcrdma_xprt *r_xprt)
1164 {
1165 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1166 	struct rpcrdma_mr *mr;
1167 
1168 	spin_lock(&buf->rb_lock);
1169 	mr = rpcrdma_mr_pop(&buf->rb_mrs);
1170 	spin_unlock(&buf->rb_lock);
1171 	return mr;
1172 }
1173 
1174 /**
1175  * rpcrdma_reply_put - Put reply buffers back into pool
1176  * @buffers: buffer pool
1177  * @req: object to return
1178  *
1179  */
rpcrdma_reply_put(struct rpcrdma_buffer * buffers,struct rpcrdma_req * req)1180 void rpcrdma_reply_put(struct rpcrdma_buffer *buffers, struct rpcrdma_req *req)
1181 {
1182 	if (req->rl_reply) {
1183 		rpcrdma_rep_put(buffers, req->rl_reply);
1184 		req->rl_reply = NULL;
1185 	}
1186 }
1187 
1188 /**
1189  * rpcrdma_buffer_get - Get a request buffer
1190  * @buffers: Buffer pool from which to obtain a buffer
1191  *
1192  * Returns a fresh rpcrdma_req, or NULL if none are available.
1193  */
1194 struct rpcrdma_req *
rpcrdma_buffer_get(struct rpcrdma_buffer * buffers)1195 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers)
1196 {
1197 	struct rpcrdma_req *req;
1198 
1199 	spin_lock(&buffers->rb_lock);
1200 	req = list_first_entry_or_null(&buffers->rb_send_bufs,
1201 				       struct rpcrdma_req, rl_list);
1202 	if (req)
1203 		list_del_init(&req->rl_list);
1204 	spin_unlock(&buffers->rb_lock);
1205 	return req;
1206 }
1207 
1208 /**
1209  * rpcrdma_buffer_put - Put request/reply buffers back into pool
1210  * @buffers: buffer pool
1211  * @req: object to return
1212  *
1213  */
rpcrdma_buffer_put(struct rpcrdma_buffer * buffers,struct rpcrdma_req * req)1214 void rpcrdma_buffer_put(struct rpcrdma_buffer *buffers, struct rpcrdma_req *req)
1215 {
1216 	rpcrdma_reply_put(buffers, req);
1217 
1218 	spin_lock(&buffers->rb_lock);
1219 	list_add(&req->rl_list, &buffers->rb_send_bufs);
1220 	spin_unlock(&buffers->rb_lock);
1221 }
1222 
1223 /* Returns a pointer to a rpcrdma_regbuf object, or NULL.
1224  *
1225  * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for
1226  * receiving the payload of RDMA RECV operations. During Long Calls
1227  * or Replies they may be registered externally via frwr_map.
1228  */
1229 static struct rpcrdma_regbuf *
rpcrdma_regbuf_alloc(size_t size,enum dma_data_direction direction)1230 rpcrdma_regbuf_alloc(size_t size, enum dma_data_direction direction)
1231 {
1232 	struct rpcrdma_regbuf *rb;
1233 
1234 	rb = kmalloc(sizeof(*rb), XPRTRDMA_GFP_FLAGS);
1235 	if (!rb)
1236 		return NULL;
1237 	rb->rg_data = kmalloc(size, XPRTRDMA_GFP_FLAGS);
1238 	if (!rb->rg_data) {
1239 		kfree(rb);
1240 		return NULL;
1241 	}
1242 
1243 	rb->rg_device = NULL;
1244 	rb->rg_direction = direction;
1245 	rb->rg_iov.length = size;
1246 	return rb;
1247 }
1248 
1249 /**
1250  * rpcrdma_regbuf_realloc - re-allocate a SEND/RECV buffer
1251  * @rb: regbuf to reallocate
1252  * @size: size of buffer to be allocated, in bytes
1253  * @flags: GFP flags
1254  *
1255  * Returns true if reallocation was successful. If false is
1256  * returned, @rb is left untouched.
1257  */
rpcrdma_regbuf_realloc(struct rpcrdma_regbuf * rb,size_t size,gfp_t flags)1258 bool rpcrdma_regbuf_realloc(struct rpcrdma_regbuf *rb, size_t size, gfp_t flags)
1259 {
1260 	void *buf;
1261 
1262 	buf = kmalloc(size, flags);
1263 	if (!buf)
1264 		return false;
1265 
1266 	rpcrdma_regbuf_dma_unmap(rb);
1267 	kfree(rb->rg_data);
1268 
1269 	rb->rg_data = buf;
1270 	rb->rg_iov.length = size;
1271 	return true;
1272 }
1273 
1274 /**
1275  * __rpcrdma_regbuf_dma_map - DMA-map a regbuf
1276  * @r_xprt: controlling transport instance
1277  * @rb: regbuf to be mapped
1278  *
1279  * Returns true if the buffer is now DMA mapped to @r_xprt's device
1280  */
__rpcrdma_regbuf_dma_map(struct rpcrdma_xprt * r_xprt,struct rpcrdma_regbuf * rb)1281 bool __rpcrdma_regbuf_dma_map(struct rpcrdma_xprt *r_xprt,
1282 			      struct rpcrdma_regbuf *rb)
1283 {
1284 	struct ib_device *device = r_xprt->rx_ep->re_id->device;
1285 
1286 	if (rb->rg_direction == DMA_NONE)
1287 		return false;
1288 
1289 	rb->rg_iov.addr = ib_dma_map_single(device, rdmab_data(rb),
1290 					    rdmab_length(rb), rb->rg_direction);
1291 	if (ib_dma_mapping_error(device, rdmab_addr(rb))) {
1292 		trace_xprtrdma_dma_maperr(rdmab_addr(rb));
1293 		return false;
1294 	}
1295 
1296 	rb->rg_device = device;
1297 	rb->rg_iov.lkey = r_xprt->rx_ep->re_pd->local_dma_lkey;
1298 	return true;
1299 }
1300 
rpcrdma_regbuf_dma_unmap(struct rpcrdma_regbuf * rb)1301 static void rpcrdma_regbuf_dma_unmap(struct rpcrdma_regbuf *rb)
1302 {
1303 	if (!rb)
1304 		return;
1305 
1306 	if (!rpcrdma_regbuf_is_mapped(rb))
1307 		return;
1308 
1309 	ib_dma_unmap_single(rb->rg_device, rdmab_addr(rb), rdmab_length(rb),
1310 			    rb->rg_direction);
1311 	rb->rg_device = NULL;
1312 }
1313 
rpcrdma_regbuf_free(struct rpcrdma_regbuf * rb)1314 static void rpcrdma_regbuf_free(struct rpcrdma_regbuf *rb)
1315 {
1316 	rpcrdma_regbuf_dma_unmap(rb);
1317 	if (rb)
1318 		kfree(rb->rg_data);
1319 	kfree(rb);
1320 }
1321 
1322 /**
1323  * rpcrdma_post_recvs - Refill the Receive Queue
1324  * @r_xprt: controlling transport instance
1325  * @needed: current credit grant
1326  * @temp: mark Receive buffers to be deleted after one use
1327  *
1328  */
rpcrdma_post_recvs(struct rpcrdma_xprt * r_xprt,int needed,bool temp)1329 void rpcrdma_post_recvs(struct rpcrdma_xprt *r_xprt, int needed, bool temp)
1330 {
1331 	struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1332 	struct rpcrdma_ep *ep = r_xprt->rx_ep;
1333 	struct ib_recv_wr *wr, *bad_wr;
1334 	struct rpcrdma_rep *rep;
1335 	int count, rc;
1336 
1337 	rc = 0;
1338 	count = 0;
1339 
1340 	if (likely(ep->re_receive_count > needed))
1341 		goto out;
1342 	needed -= ep->re_receive_count;
1343 	if (!temp)
1344 		needed += RPCRDMA_MAX_RECV_BATCH;
1345 
1346 	if (atomic_inc_return(&ep->re_receiving) > 1)
1347 		goto out;
1348 
1349 	/* fast path: all needed reps can be found on the free list */
1350 	wr = NULL;
1351 	while (needed) {
1352 		rep = rpcrdma_rep_get_locked(buf);
1353 		if (rep && rep->rr_temp) {
1354 			rpcrdma_rep_destroy(rep);
1355 			continue;
1356 		}
1357 		if (!rep)
1358 			rep = rpcrdma_rep_create(r_xprt, temp);
1359 		if (!rep)
1360 			break;
1361 		if (!rpcrdma_regbuf_dma_map(r_xprt, rep->rr_rdmabuf)) {
1362 			rpcrdma_rep_put(buf, rep);
1363 			break;
1364 		}
1365 
1366 		rep->rr_cid.ci_queue_id = ep->re_attr.recv_cq->res.id;
1367 		trace_xprtrdma_post_recv(rep);
1368 		rep->rr_recv_wr.next = wr;
1369 		wr = &rep->rr_recv_wr;
1370 		--needed;
1371 		++count;
1372 	}
1373 	if (!wr)
1374 		goto out;
1375 
1376 	rc = ib_post_recv(ep->re_id->qp, wr,
1377 			  (const struct ib_recv_wr **)&bad_wr);
1378 	if (rc) {
1379 		trace_xprtrdma_post_recvs_err(r_xprt, rc);
1380 		for (wr = bad_wr; wr;) {
1381 			struct rpcrdma_rep *rep;
1382 
1383 			rep = container_of(wr, struct rpcrdma_rep, rr_recv_wr);
1384 			wr = wr->next;
1385 			rpcrdma_rep_put(buf, rep);
1386 			--count;
1387 		}
1388 	}
1389 	if (atomic_dec_return(&ep->re_receiving) > 0)
1390 		complete(&ep->re_done);
1391 
1392 out:
1393 	trace_xprtrdma_post_recvs(r_xprt, count);
1394 	ep->re_receive_count += count;
1395 	return;
1396 }
1397