1 // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
2 /*
3  * Copyright (c) 2015-2018 Oracle. All rights reserved.
4  * Copyright (c) 2014 Open Grid Computing, Inc. All rights reserved.
5  * Copyright (c) 2005-2007 Network Appliance, Inc. All rights reserved.
6  *
7  * This software is available to you under a choice of one of two
8  * licenses.  You may choose to be licensed under the terms of the GNU
9  * General Public License (GPL) Version 2, available from the file
10  * COPYING in the main directory of this source tree, or the BSD-type
11  * license below:
12  *
13  * Redistribution and use in source and binary forms, with or without
14  * modification, are permitted provided that the following conditions
15  * are met:
16  *
17  *      Redistributions of source code must retain the above copyright
18  *      notice, this list of conditions and the following disclaimer.
19  *
20  *      Redistributions in binary form must reproduce the above
21  *      copyright notice, this list of conditions and the following
22  *      disclaimer in the documentation and/or other materials provided
23  *      with the distribution.
24  *
25  *      Neither the name of the Network Appliance, Inc. nor the names of
26  *      its contributors may be used to endorse or promote products
27  *      derived from this software without specific prior written
28  *      permission.
29  *
30  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
31  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
32  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
33  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
34  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
35  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
36  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
37  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
38  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
39  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
40  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
41  *
42  * Author: Tom Tucker <tom@opengridcomputing.com>
43  */
44 
45 #include <linux/interrupt.h>
46 #include <linux/sched.h>
47 #include <linux/slab.h>
48 #include <linux/spinlock.h>
49 #include <linux/workqueue.h>
50 #include <linux/export.h>
51 
52 #include <rdma/ib_verbs.h>
53 #include <rdma/rdma_cm.h>
54 #include <rdma/rw.h>
55 
56 #include <linux/sunrpc/addr.h>
57 #include <linux/sunrpc/debug.h>
58 #include <linux/sunrpc/rpc_rdma.h>
59 #include <linux/sunrpc/svc_xprt.h>
60 #include <linux/sunrpc/svc_rdma.h>
61 
62 #include "xprt_rdma.h"
63 #include <trace/events/rpcrdma.h>
64 
65 #define RPCDBG_FACILITY	RPCDBG_SVCXPRT
66 
67 static struct svcxprt_rdma *svc_rdma_create_xprt(struct svc_serv *serv,
68 						 struct net *net);
69 static struct svc_xprt *svc_rdma_create(struct svc_serv *serv,
70 					struct net *net,
71 					struct sockaddr *sa, int salen,
72 					int flags);
73 static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt);
74 static void svc_rdma_release_rqst(struct svc_rqst *);
75 static void svc_rdma_detach(struct svc_xprt *xprt);
76 static void svc_rdma_free(struct svc_xprt *xprt);
77 static int svc_rdma_has_wspace(struct svc_xprt *xprt);
78 static void svc_rdma_secure_port(struct svc_rqst *);
79 static void svc_rdma_kill_temp_xprt(struct svc_xprt *);
80 
81 static const struct svc_xprt_ops svc_rdma_ops = {
82 	.xpo_create = svc_rdma_create,
83 	.xpo_recvfrom = svc_rdma_recvfrom,
84 	.xpo_sendto = svc_rdma_sendto,
85 	.xpo_release_rqst = svc_rdma_release_rqst,
86 	.xpo_detach = svc_rdma_detach,
87 	.xpo_free = svc_rdma_free,
88 	.xpo_prep_reply_hdr = svc_rdma_prep_reply_hdr,
89 	.xpo_has_wspace = svc_rdma_has_wspace,
90 	.xpo_accept = svc_rdma_accept,
91 	.xpo_secure_port = svc_rdma_secure_port,
92 	.xpo_kill_temp_xprt = svc_rdma_kill_temp_xprt,
93 };
94 
95 struct svc_xprt_class svc_rdma_class = {
96 	.xcl_name = "rdma",
97 	.xcl_owner = THIS_MODULE,
98 	.xcl_ops = &svc_rdma_ops,
99 	.xcl_max_payload = RPCSVC_MAXPAYLOAD_RDMA,
100 	.xcl_ident = XPRT_TRANSPORT_RDMA,
101 };
102 
103 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
104 static struct svc_xprt *svc_rdma_bc_create(struct svc_serv *, struct net *,
105 					   struct sockaddr *, int, int);
106 static void svc_rdma_bc_detach(struct svc_xprt *);
107 static void svc_rdma_bc_free(struct svc_xprt *);
108 
109 static const struct svc_xprt_ops svc_rdma_bc_ops = {
110 	.xpo_create = svc_rdma_bc_create,
111 	.xpo_detach = svc_rdma_bc_detach,
112 	.xpo_free = svc_rdma_bc_free,
113 	.xpo_prep_reply_hdr = svc_rdma_prep_reply_hdr,
114 	.xpo_secure_port = svc_rdma_secure_port,
115 };
116 
117 struct svc_xprt_class svc_rdma_bc_class = {
118 	.xcl_name = "rdma-bc",
119 	.xcl_owner = THIS_MODULE,
120 	.xcl_ops = &svc_rdma_bc_ops,
121 	.xcl_max_payload = (1024 - RPCRDMA_HDRLEN_MIN)
122 };
123 
124 static struct svc_xprt *svc_rdma_bc_create(struct svc_serv *serv,
125 					   struct net *net,
126 					   struct sockaddr *sa, int salen,
127 					   int flags)
128 {
129 	struct svcxprt_rdma *cma_xprt;
130 	struct svc_xprt *xprt;
131 
132 	cma_xprt = svc_rdma_create_xprt(serv, net);
133 	if (!cma_xprt)
134 		return ERR_PTR(-ENOMEM);
135 	xprt = &cma_xprt->sc_xprt;
136 
137 	svc_xprt_init(net, &svc_rdma_bc_class, xprt, serv);
138 	set_bit(XPT_CONG_CTRL, &xprt->xpt_flags);
139 	serv->sv_bc_xprt = xprt;
140 
141 	dprintk("svcrdma: %s(%p)\n", __func__, xprt);
142 	return xprt;
143 }
144 
145 static void svc_rdma_bc_detach(struct svc_xprt *xprt)
146 {
147 	dprintk("svcrdma: %s(%p)\n", __func__, xprt);
148 }
149 
150 static void svc_rdma_bc_free(struct svc_xprt *xprt)
151 {
152 	struct svcxprt_rdma *rdma =
153 		container_of(xprt, struct svcxprt_rdma, sc_xprt);
154 
155 	dprintk("svcrdma: %s(%p)\n", __func__, xprt);
156 	if (xprt)
157 		kfree(rdma);
158 }
159 #endif	/* CONFIG_SUNRPC_BACKCHANNEL */
160 
161 /* QP event handler */
162 static void qp_event_handler(struct ib_event *event, void *context)
163 {
164 	struct svc_xprt *xprt = context;
165 
166 	trace_svcrdma_qp_error(event, (struct sockaddr *)&xprt->xpt_remote);
167 	switch (event->event) {
168 	/* These are considered benign events */
169 	case IB_EVENT_PATH_MIG:
170 	case IB_EVENT_COMM_EST:
171 	case IB_EVENT_SQ_DRAINED:
172 	case IB_EVENT_QP_LAST_WQE_REACHED:
173 		break;
174 
175 	/* These are considered fatal events */
176 	case IB_EVENT_PATH_MIG_ERR:
177 	case IB_EVENT_QP_FATAL:
178 	case IB_EVENT_QP_REQ_ERR:
179 	case IB_EVENT_QP_ACCESS_ERR:
180 	case IB_EVENT_DEVICE_FATAL:
181 	default:
182 		set_bit(XPT_CLOSE, &xprt->xpt_flags);
183 		svc_xprt_enqueue(xprt);
184 		break;
185 	}
186 }
187 
188 static struct svcxprt_rdma *svc_rdma_create_xprt(struct svc_serv *serv,
189 						 struct net *net)
190 {
191 	struct svcxprt_rdma *cma_xprt = kzalloc(sizeof *cma_xprt, GFP_KERNEL);
192 
193 	if (!cma_xprt) {
194 		dprintk("svcrdma: failed to create new transport\n");
195 		return NULL;
196 	}
197 	svc_xprt_init(net, &svc_rdma_class, &cma_xprt->sc_xprt, serv);
198 	INIT_LIST_HEAD(&cma_xprt->sc_accept_q);
199 	INIT_LIST_HEAD(&cma_xprt->sc_rq_dto_q);
200 	INIT_LIST_HEAD(&cma_xprt->sc_read_complete_q);
201 	INIT_LIST_HEAD(&cma_xprt->sc_send_ctxts);
202 	INIT_LIST_HEAD(&cma_xprt->sc_recv_ctxts);
203 	INIT_LIST_HEAD(&cma_xprt->sc_rw_ctxts);
204 	init_waitqueue_head(&cma_xprt->sc_send_wait);
205 
206 	spin_lock_init(&cma_xprt->sc_lock);
207 	spin_lock_init(&cma_xprt->sc_rq_dto_lock);
208 	spin_lock_init(&cma_xprt->sc_send_lock);
209 	spin_lock_init(&cma_xprt->sc_recv_lock);
210 	spin_lock_init(&cma_xprt->sc_rw_ctxt_lock);
211 
212 	/*
213 	 * Note that this implies that the underlying transport support
214 	 * has some form of congestion control (see RFC 7530 section 3.1
215 	 * paragraph 2). For now, we assume that all supported RDMA
216 	 * transports are suitable here.
217 	 */
218 	set_bit(XPT_CONG_CTRL, &cma_xprt->sc_xprt.xpt_flags);
219 
220 	return cma_xprt;
221 }
222 
223 static void
224 svc_rdma_parse_connect_private(struct svcxprt_rdma *newxprt,
225 			       struct rdma_conn_param *param)
226 {
227 	const struct rpcrdma_connect_private *pmsg = param->private_data;
228 
229 	if (pmsg &&
230 	    pmsg->cp_magic == rpcrdma_cmp_magic &&
231 	    pmsg->cp_version == RPCRDMA_CMP_VERSION) {
232 		newxprt->sc_snd_w_inv = pmsg->cp_flags &
233 					RPCRDMA_CMP_F_SND_W_INV_OK;
234 
235 		dprintk("svcrdma: client send_size %u, recv_size %u "
236 			"remote inv %ssupported\n",
237 			rpcrdma_decode_buffer_size(pmsg->cp_send_size),
238 			rpcrdma_decode_buffer_size(pmsg->cp_recv_size),
239 			newxprt->sc_snd_w_inv ? "" : "un");
240 	}
241 }
242 
243 /*
244  * This function handles the CONNECT_REQUEST event on a listening
245  * endpoint. It is passed the cma_id for the _new_ connection. The context in
246  * this cma_id is inherited from the listening cma_id and is the svc_xprt
247  * structure for the listening endpoint.
248  *
249  * This function creates a new xprt for the new connection and enqueues it on
250  * the accept queue for the listent xprt. When the listen thread is kicked, it
251  * will call the recvfrom method on the listen xprt which will accept the new
252  * connection.
253  */
254 static void handle_connect_req(struct rdma_cm_id *new_cma_id,
255 			       struct rdma_conn_param *param)
256 {
257 	struct svcxprt_rdma *listen_xprt = new_cma_id->context;
258 	struct svcxprt_rdma *newxprt;
259 	struct sockaddr *sa;
260 
261 	/* Create a new transport */
262 	newxprt = svc_rdma_create_xprt(listen_xprt->sc_xprt.xpt_server,
263 				       listen_xprt->sc_xprt.xpt_net);
264 	if (!newxprt)
265 		return;
266 	newxprt->sc_cm_id = new_cma_id;
267 	new_cma_id->context = newxprt;
268 	svc_rdma_parse_connect_private(newxprt, param);
269 
270 	/* Save client advertised inbound read limit for use later in accept. */
271 	newxprt->sc_ord = param->initiator_depth;
272 
273 	/* Set the local and remote addresses in the transport */
274 	sa = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.dst_addr;
275 	svc_xprt_set_remote(&newxprt->sc_xprt, sa, svc_addr_len(sa));
276 	sa = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.src_addr;
277 	svc_xprt_set_local(&newxprt->sc_xprt, sa, svc_addr_len(sa));
278 
279 	/*
280 	 * Enqueue the new transport on the accept queue of the listening
281 	 * transport
282 	 */
283 	spin_lock_bh(&listen_xprt->sc_lock);
284 	list_add_tail(&newxprt->sc_accept_q, &listen_xprt->sc_accept_q);
285 	spin_unlock_bh(&listen_xprt->sc_lock);
286 
287 	set_bit(XPT_CONN, &listen_xprt->sc_xprt.xpt_flags);
288 	svc_xprt_enqueue(&listen_xprt->sc_xprt);
289 }
290 
291 /*
292  * Handles events generated on the listening endpoint. These events will be
293  * either be incoming connect requests or adapter removal  events.
294  */
295 static int rdma_listen_handler(struct rdma_cm_id *cma_id,
296 			       struct rdma_cm_event *event)
297 {
298 	struct sockaddr *sap = (struct sockaddr *)&cma_id->route.addr.src_addr;
299 	int ret = 0;
300 
301 	trace_svcrdma_cm_event(event, sap);
302 
303 	switch (event->event) {
304 	case RDMA_CM_EVENT_CONNECT_REQUEST:
305 		dprintk("svcrdma: Connect request on cma_id=%p, xprt = %p, "
306 			"event = %s (%d)\n", cma_id, cma_id->context,
307 			rdma_event_msg(event->event), event->event);
308 		handle_connect_req(cma_id, &event->param.conn);
309 		break;
310 	default:
311 		/* NB: No device removal upcall for INADDR_ANY listeners */
312 		dprintk("svcrdma: Unexpected event on listening endpoint %p, "
313 			"event = %s (%d)\n", cma_id,
314 			rdma_event_msg(event->event), event->event);
315 		break;
316 	}
317 
318 	return ret;
319 }
320 
321 static int rdma_cma_handler(struct rdma_cm_id *cma_id,
322 			    struct rdma_cm_event *event)
323 {
324 	struct sockaddr *sap = (struct sockaddr *)&cma_id->route.addr.dst_addr;
325 	struct svcxprt_rdma *rdma = cma_id->context;
326 	struct svc_xprt *xprt = &rdma->sc_xprt;
327 
328 	trace_svcrdma_cm_event(event, sap);
329 
330 	switch (event->event) {
331 	case RDMA_CM_EVENT_ESTABLISHED:
332 		/* Accept complete */
333 		svc_xprt_get(xprt);
334 		dprintk("svcrdma: Connection completed on DTO xprt=%p, "
335 			"cm_id=%p\n", xprt, cma_id);
336 		clear_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags);
337 		svc_xprt_enqueue(xprt);
338 		break;
339 	case RDMA_CM_EVENT_DISCONNECTED:
340 		dprintk("svcrdma: Disconnect on DTO xprt=%p, cm_id=%p\n",
341 			xprt, cma_id);
342 		set_bit(XPT_CLOSE, &xprt->xpt_flags);
343 		svc_xprt_enqueue(xprt);
344 		svc_xprt_put(xprt);
345 		break;
346 	case RDMA_CM_EVENT_DEVICE_REMOVAL:
347 		dprintk("svcrdma: Device removal cma_id=%p, xprt = %p, "
348 			"event = %s (%d)\n", cma_id, xprt,
349 			rdma_event_msg(event->event), event->event);
350 		set_bit(XPT_CLOSE, &xprt->xpt_flags);
351 		svc_xprt_enqueue(xprt);
352 		svc_xprt_put(xprt);
353 		break;
354 	default:
355 		dprintk("svcrdma: Unexpected event on DTO endpoint %p, "
356 			"event = %s (%d)\n", cma_id,
357 			rdma_event_msg(event->event), event->event);
358 		break;
359 	}
360 	return 0;
361 }
362 
363 /*
364  * Create a listening RDMA service endpoint.
365  */
366 static struct svc_xprt *svc_rdma_create(struct svc_serv *serv,
367 					struct net *net,
368 					struct sockaddr *sa, int salen,
369 					int flags)
370 {
371 	struct rdma_cm_id *listen_id;
372 	struct svcxprt_rdma *cma_xprt;
373 	int ret;
374 
375 	dprintk("svcrdma: Creating RDMA listener\n");
376 	if ((sa->sa_family != AF_INET) && (sa->sa_family != AF_INET6)) {
377 		dprintk("svcrdma: Address family %d is not supported.\n", sa->sa_family);
378 		return ERR_PTR(-EAFNOSUPPORT);
379 	}
380 	cma_xprt = svc_rdma_create_xprt(serv, net);
381 	if (!cma_xprt)
382 		return ERR_PTR(-ENOMEM);
383 	set_bit(XPT_LISTENER, &cma_xprt->sc_xprt.xpt_flags);
384 	strcpy(cma_xprt->sc_xprt.xpt_remotebuf, "listener");
385 
386 	listen_id = rdma_create_id(net, rdma_listen_handler, cma_xprt,
387 				   RDMA_PS_TCP, IB_QPT_RC);
388 	if (IS_ERR(listen_id)) {
389 		ret = PTR_ERR(listen_id);
390 		dprintk("svcrdma: rdma_create_id failed = %d\n", ret);
391 		goto err0;
392 	}
393 
394 	/* Allow both IPv4 and IPv6 sockets to bind a single port
395 	 * at the same time.
396 	 */
397 #if IS_ENABLED(CONFIG_IPV6)
398 	ret = rdma_set_afonly(listen_id, 1);
399 	if (ret) {
400 		dprintk("svcrdma: rdma_set_afonly failed = %d\n", ret);
401 		goto err1;
402 	}
403 #endif
404 	ret = rdma_bind_addr(listen_id, sa);
405 	if (ret) {
406 		dprintk("svcrdma: rdma_bind_addr failed = %d\n", ret);
407 		goto err1;
408 	}
409 	cma_xprt->sc_cm_id = listen_id;
410 
411 	ret = rdma_listen(listen_id, RPCRDMA_LISTEN_BACKLOG);
412 	if (ret) {
413 		dprintk("svcrdma: rdma_listen failed = %d\n", ret);
414 		goto err1;
415 	}
416 
417 	/*
418 	 * We need to use the address from the cm_id in case the
419 	 * caller specified 0 for the port number.
420 	 */
421 	sa = (struct sockaddr *)&cma_xprt->sc_cm_id->route.addr.src_addr;
422 	svc_xprt_set_local(&cma_xprt->sc_xprt, sa, salen);
423 
424 	return &cma_xprt->sc_xprt;
425 
426  err1:
427 	rdma_destroy_id(listen_id);
428  err0:
429 	kfree(cma_xprt);
430 	return ERR_PTR(ret);
431 }
432 
433 /*
434  * This is the xpo_recvfrom function for listening endpoints. Its
435  * purpose is to accept incoming connections. The CMA callback handler
436  * has already created a new transport and attached it to the new CMA
437  * ID.
438  *
439  * There is a queue of pending connections hung on the listening
440  * transport. This queue contains the new svc_xprt structure. This
441  * function takes svc_xprt structures off the accept_q and completes
442  * the connection.
443  */
444 static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
445 {
446 	struct svcxprt_rdma *listen_rdma;
447 	struct svcxprt_rdma *newxprt = NULL;
448 	struct rdma_conn_param conn_param;
449 	struct rpcrdma_connect_private pmsg;
450 	struct ib_qp_init_attr qp_attr;
451 	unsigned int ctxts, rq_depth;
452 	struct ib_device *dev;
453 	struct sockaddr *sap;
454 	int ret = 0;
455 
456 	listen_rdma = container_of(xprt, struct svcxprt_rdma, sc_xprt);
457 	clear_bit(XPT_CONN, &xprt->xpt_flags);
458 	/* Get the next entry off the accept list */
459 	spin_lock_bh(&listen_rdma->sc_lock);
460 	if (!list_empty(&listen_rdma->sc_accept_q)) {
461 		newxprt = list_entry(listen_rdma->sc_accept_q.next,
462 				     struct svcxprt_rdma, sc_accept_q);
463 		list_del_init(&newxprt->sc_accept_q);
464 	}
465 	if (!list_empty(&listen_rdma->sc_accept_q))
466 		set_bit(XPT_CONN, &listen_rdma->sc_xprt.xpt_flags);
467 	spin_unlock_bh(&listen_rdma->sc_lock);
468 	if (!newxprt)
469 		return NULL;
470 
471 	dprintk("svcrdma: newxprt from accept queue = %p, cm_id=%p\n",
472 		newxprt, newxprt->sc_cm_id);
473 
474 	dev = newxprt->sc_cm_id->device;
475 	newxprt->sc_port_num = newxprt->sc_cm_id->port_num;
476 
477 	/* Qualify the transport resource defaults with the
478 	 * capabilities of this particular device */
479 	newxprt->sc_max_send_sges = dev->attrs.max_sge;
480 	/* transport hdr, head iovec, one page list entry, tail iovec */
481 	if (newxprt->sc_max_send_sges < 4) {
482 		pr_err("svcrdma: too few Send SGEs available (%d)\n",
483 		       newxprt->sc_max_send_sges);
484 		goto errout;
485 	}
486 	newxprt->sc_max_req_size = svcrdma_max_req_size;
487 	newxprt->sc_max_requests = svcrdma_max_requests;
488 	newxprt->sc_max_bc_requests = svcrdma_max_bc_requests;
489 	rq_depth = newxprt->sc_max_requests + newxprt->sc_max_bc_requests;
490 	if (rq_depth > dev->attrs.max_qp_wr) {
491 		pr_warn("svcrdma: reducing receive depth to %d\n",
492 			dev->attrs.max_qp_wr);
493 		rq_depth = dev->attrs.max_qp_wr;
494 		newxprt->sc_max_requests = rq_depth - 2;
495 		newxprt->sc_max_bc_requests = 2;
496 	}
497 	newxprt->sc_fc_credits = cpu_to_be32(newxprt->sc_max_requests);
498 	ctxts = rdma_rw_mr_factor(dev, newxprt->sc_port_num, RPCSVC_MAXPAGES);
499 	ctxts *= newxprt->sc_max_requests;
500 	newxprt->sc_sq_depth = rq_depth + ctxts;
501 	if (newxprt->sc_sq_depth > dev->attrs.max_qp_wr) {
502 		pr_warn("svcrdma: reducing send depth to %d\n",
503 			dev->attrs.max_qp_wr);
504 		newxprt->sc_sq_depth = dev->attrs.max_qp_wr;
505 	}
506 	atomic_set(&newxprt->sc_sq_avail, newxprt->sc_sq_depth);
507 
508 	newxprt->sc_pd = ib_alloc_pd(dev, 0);
509 	if (IS_ERR(newxprt->sc_pd)) {
510 		dprintk("svcrdma: error creating PD for connect request\n");
511 		goto errout;
512 	}
513 	newxprt->sc_sq_cq = ib_alloc_cq(dev, newxprt, newxprt->sc_sq_depth,
514 					0, IB_POLL_WORKQUEUE);
515 	if (IS_ERR(newxprt->sc_sq_cq)) {
516 		dprintk("svcrdma: error creating SQ CQ for connect request\n");
517 		goto errout;
518 	}
519 	newxprt->sc_rq_cq = ib_alloc_cq(dev, newxprt, rq_depth,
520 					0, IB_POLL_WORKQUEUE);
521 	if (IS_ERR(newxprt->sc_rq_cq)) {
522 		dprintk("svcrdma: error creating RQ CQ for connect request\n");
523 		goto errout;
524 	}
525 
526 	memset(&qp_attr, 0, sizeof qp_attr);
527 	qp_attr.event_handler = qp_event_handler;
528 	qp_attr.qp_context = &newxprt->sc_xprt;
529 	qp_attr.port_num = newxprt->sc_port_num;
530 	qp_attr.cap.max_rdma_ctxs = ctxts;
531 	qp_attr.cap.max_send_wr = newxprt->sc_sq_depth - ctxts;
532 	qp_attr.cap.max_recv_wr = rq_depth;
533 	qp_attr.cap.max_send_sge = newxprt->sc_max_send_sges;
534 	qp_attr.cap.max_recv_sge = 1;
535 	qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
536 	qp_attr.qp_type = IB_QPT_RC;
537 	qp_attr.send_cq = newxprt->sc_sq_cq;
538 	qp_attr.recv_cq = newxprt->sc_rq_cq;
539 	dprintk("svcrdma: newxprt->sc_cm_id=%p, newxprt->sc_pd=%p\n",
540 		newxprt->sc_cm_id, newxprt->sc_pd);
541 	dprintk("    cap.max_send_wr = %d, cap.max_recv_wr = %d\n",
542 		qp_attr.cap.max_send_wr, qp_attr.cap.max_recv_wr);
543 	dprintk("    cap.max_send_sge = %d, cap.max_recv_sge = %d\n",
544 		qp_attr.cap.max_send_sge, qp_attr.cap.max_recv_sge);
545 
546 	ret = rdma_create_qp(newxprt->sc_cm_id, newxprt->sc_pd, &qp_attr);
547 	if (ret) {
548 		dprintk("svcrdma: failed to create QP, ret=%d\n", ret);
549 		goto errout;
550 	}
551 	newxprt->sc_qp = newxprt->sc_cm_id->qp;
552 
553 	if (!(dev->attrs.device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
554 		newxprt->sc_snd_w_inv = false;
555 	if (!rdma_protocol_iwarp(dev, newxprt->sc_port_num) &&
556 	    !rdma_ib_or_roce(dev, newxprt->sc_port_num))
557 		goto errout;
558 
559 	if (!svc_rdma_post_recvs(newxprt))
560 		goto errout;
561 
562 	/* Swap out the handler */
563 	newxprt->sc_cm_id->event_handler = rdma_cma_handler;
564 
565 	/* Construct RDMA-CM private message */
566 	pmsg.cp_magic = rpcrdma_cmp_magic;
567 	pmsg.cp_version = RPCRDMA_CMP_VERSION;
568 	pmsg.cp_flags = 0;
569 	pmsg.cp_send_size = pmsg.cp_recv_size =
570 		rpcrdma_encode_buffer_size(newxprt->sc_max_req_size);
571 
572 	/* Accept Connection */
573 	set_bit(RDMAXPRT_CONN_PENDING, &newxprt->sc_flags);
574 	memset(&conn_param, 0, sizeof conn_param);
575 	conn_param.responder_resources = 0;
576 	conn_param.initiator_depth = min_t(int, newxprt->sc_ord,
577 					   dev->attrs.max_qp_init_rd_atom);
578 	if (!conn_param.initiator_depth) {
579 		dprintk("svcrdma: invalid ORD setting\n");
580 		ret = -EINVAL;
581 		goto errout;
582 	}
583 	conn_param.private_data = &pmsg;
584 	conn_param.private_data_len = sizeof(pmsg);
585 	ret = rdma_accept(newxprt->sc_cm_id, &conn_param);
586 	if (ret)
587 		goto errout;
588 
589 	dprintk("svcrdma: new connection %p accepted:\n", newxprt);
590 	sap = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.src_addr;
591 	dprintk("    local address   : %pIS:%u\n", sap, rpc_get_port(sap));
592 	sap = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.dst_addr;
593 	dprintk("    remote address  : %pIS:%u\n", sap, rpc_get_port(sap));
594 	dprintk("    max_sge         : %d\n", newxprt->sc_max_send_sges);
595 	dprintk("    sq_depth        : %d\n", newxprt->sc_sq_depth);
596 	dprintk("    rdma_rw_ctxs    : %d\n", ctxts);
597 	dprintk("    max_requests    : %d\n", newxprt->sc_max_requests);
598 	dprintk("    ord             : %d\n", conn_param.initiator_depth);
599 
600 	trace_svcrdma_xprt_accept(&newxprt->sc_xprt);
601 	return &newxprt->sc_xprt;
602 
603  errout:
604 	dprintk("svcrdma: failure accepting new connection rc=%d.\n", ret);
605 	trace_svcrdma_xprt_fail(&newxprt->sc_xprt);
606 	/* Take a reference in case the DTO handler runs */
607 	svc_xprt_get(&newxprt->sc_xprt);
608 	if (newxprt->sc_qp && !IS_ERR(newxprt->sc_qp))
609 		ib_destroy_qp(newxprt->sc_qp);
610 	rdma_destroy_id(newxprt->sc_cm_id);
611 	/* This call to put will destroy the transport */
612 	svc_xprt_put(&newxprt->sc_xprt);
613 	return NULL;
614 }
615 
616 static void svc_rdma_release_rqst(struct svc_rqst *rqstp)
617 {
618 }
619 
620 /*
621  * When connected, an svc_xprt has at least two references:
622  *
623  * - A reference held by the cm_id between the ESTABLISHED and
624  *   DISCONNECTED events. If the remote peer disconnected first, this
625  *   reference could be gone.
626  *
627  * - A reference held by the svc_recv code that called this function
628  *   as part of close processing.
629  *
630  * At a minimum one references should still be held.
631  */
632 static void svc_rdma_detach(struct svc_xprt *xprt)
633 {
634 	struct svcxprt_rdma *rdma =
635 		container_of(xprt, struct svcxprt_rdma, sc_xprt);
636 
637 	/* Disconnect and flush posted WQE */
638 	rdma_disconnect(rdma->sc_cm_id);
639 }
640 
641 static void __svc_rdma_free(struct work_struct *work)
642 {
643 	struct svcxprt_rdma *rdma =
644 		container_of(work, struct svcxprt_rdma, sc_work);
645 	struct svc_xprt *xprt = &rdma->sc_xprt;
646 
647 	trace_svcrdma_xprt_free(xprt);
648 
649 	if (rdma->sc_qp && !IS_ERR(rdma->sc_qp))
650 		ib_drain_qp(rdma->sc_qp);
651 
652 	/* We should only be called from kref_put */
653 	if (kref_read(&xprt->xpt_ref) != 0)
654 		pr_err("svcrdma: sc_xprt still in use? (%d)\n",
655 		       kref_read(&xprt->xpt_ref));
656 
657 	svc_rdma_flush_recv_queues(rdma);
658 
659 	/* Final put of backchannel client transport */
660 	if (xprt->xpt_bc_xprt) {
661 		xprt_put(xprt->xpt_bc_xprt);
662 		xprt->xpt_bc_xprt = NULL;
663 	}
664 
665 	svc_rdma_destroy_rw_ctxts(rdma);
666 	svc_rdma_send_ctxts_destroy(rdma);
667 	svc_rdma_recv_ctxts_destroy(rdma);
668 
669 	/* Destroy the QP if present (not a listener) */
670 	if (rdma->sc_qp && !IS_ERR(rdma->sc_qp))
671 		ib_destroy_qp(rdma->sc_qp);
672 
673 	if (rdma->sc_sq_cq && !IS_ERR(rdma->sc_sq_cq))
674 		ib_free_cq(rdma->sc_sq_cq);
675 
676 	if (rdma->sc_rq_cq && !IS_ERR(rdma->sc_rq_cq))
677 		ib_free_cq(rdma->sc_rq_cq);
678 
679 	if (rdma->sc_pd && !IS_ERR(rdma->sc_pd))
680 		ib_dealloc_pd(rdma->sc_pd);
681 
682 	/* Destroy the CM ID */
683 	rdma_destroy_id(rdma->sc_cm_id);
684 
685 	kfree(rdma);
686 }
687 
688 static void svc_rdma_free(struct svc_xprt *xprt)
689 {
690 	struct svcxprt_rdma *rdma =
691 		container_of(xprt, struct svcxprt_rdma, sc_xprt);
692 	INIT_WORK(&rdma->sc_work, __svc_rdma_free);
693 	queue_work(svc_rdma_wq, &rdma->sc_work);
694 }
695 
696 static int svc_rdma_has_wspace(struct svc_xprt *xprt)
697 {
698 	struct svcxprt_rdma *rdma =
699 		container_of(xprt, struct svcxprt_rdma, sc_xprt);
700 
701 	/*
702 	 * If there are already waiters on the SQ,
703 	 * return false.
704 	 */
705 	if (waitqueue_active(&rdma->sc_send_wait))
706 		return 0;
707 
708 	/* Otherwise return true. */
709 	return 1;
710 }
711 
712 static void svc_rdma_secure_port(struct svc_rqst *rqstp)
713 {
714 	set_bit(RQ_SECURE, &rqstp->rq_flags);
715 }
716 
717 static void svc_rdma_kill_temp_xprt(struct svc_xprt *xprt)
718 {
719 }
720