1 /*
2 * Copyright (c) 2014 Open Grid Computing, Inc. All rights reserved.
3 * Copyright (c) 2005-2007 Network Appliance, Inc. All rights reserved.
4 *
5 * This software is available to you under a choice of one of two
6 * licenses. You may choose to be licensed under the terms of the GNU
7 * General Public License (GPL) Version 2, available from the file
8 * COPYING in the main directory of this source tree, or the BSD-type
9 * license below:
10 *
11 * Redistribution and use in source and binary forms, with or without
12 * modification, are permitted provided that the following conditions
13 * are met:
14 *
15 * Redistributions of source code must retain the above copyright
16 * notice, this list of conditions and the following disclaimer.
17 *
18 * Redistributions in binary form must reproduce the above
19 * copyright notice, this list of conditions and the following
20 * disclaimer in the documentation and/or other materials provided
21 * with the distribution.
22 *
23 * Neither the name of the Network Appliance, Inc. nor the names of
24 * its contributors may be used to endorse or promote products
25 * derived from this software without specific prior written
26 * permission.
27 *
28 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
29 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
30 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
31 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
32 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
33 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
34 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
35 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
36 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
37 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
38 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
39 *
40 * Author: Tom Tucker <tom@opengridcomputing.com>
41 */
42
43 #include <linux/sunrpc/svc_xprt.h>
44 #include <linux/sunrpc/debug.h>
45 #include <linux/sunrpc/rpc_rdma.h>
46 #include <linux/interrupt.h>
47 #include <linux/sched.h>
48 #include <linux/slab.h>
49 #include <linux/spinlock.h>
50 #include <linux/workqueue.h>
51 #include <rdma/ib_verbs.h>
52 #include <rdma/rdma_cm.h>
53 #include <linux/sunrpc/svc_rdma.h>
54 #include <linux/export.h>
55 #include "xprt_rdma.h"
56
57 #define RPCDBG_FACILITY RPCDBG_SVCXPRT
58
59 static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *, int);
60 static struct svc_xprt *svc_rdma_create(struct svc_serv *serv,
61 struct net *net,
62 struct sockaddr *sa, int salen,
63 int flags);
64 static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt);
65 static void svc_rdma_release_rqst(struct svc_rqst *);
66 static void svc_rdma_detach(struct svc_xprt *xprt);
67 static void svc_rdma_free(struct svc_xprt *xprt);
68 static int svc_rdma_has_wspace(struct svc_xprt *xprt);
69 static int svc_rdma_secure_port(struct svc_rqst *);
70 static void svc_rdma_kill_temp_xprt(struct svc_xprt *);
71
72 static struct svc_xprt_ops svc_rdma_ops = {
73 .xpo_create = svc_rdma_create,
74 .xpo_recvfrom = svc_rdma_recvfrom,
75 .xpo_sendto = svc_rdma_sendto,
76 .xpo_release_rqst = svc_rdma_release_rqst,
77 .xpo_detach = svc_rdma_detach,
78 .xpo_free = svc_rdma_free,
79 .xpo_prep_reply_hdr = svc_rdma_prep_reply_hdr,
80 .xpo_has_wspace = svc_rdma_has_wspace,
81 .xpo_accept = svc_rdma_accept,
82 .xpo_secure_port = svc_rdma_secure_port,
83 .xpo_kill_temp_xprt = svc_rdma_kill_temp_xprt,
84 };
85
86 struct svc_xprt_class svc_rdma_class = {
87 .xcl_name = "rdma",
88 .xcl_owner = THIS_MODULE,
89 .xcl_ops = &svc_rdma_ops,
90 .xcl_max_payload = RPCSVC_MAXPAYLOAD_RDMA,
91 .xcl_ident = XPRT_TRANSPORT_RDMA,
92 };
93
94 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
95 static struct svc_xprt *svc_rdma_bc_create(struct svc_serv *, struct net *,
96 struct sockaddr *, int, int);
97 static void svc_rdma_bc_detach(struct svc_xprt *);
98 static void svc_rdma_bc_free(struct svc_xprt *);
99
100 static struct svc_xprt_ops svc_rdma_bc_ops = {
101 .xpo_create = svc_rdma_bc_create,
102 .xpo_detach = svc_rdma_bc_detach,
103 .xpo_free = svc_rdma_bc_free,
104 .xpo_prep_reply_hdr = svc_rdma_prep_reply_hdr,
105 .xpo_secure_port = svc_rdma_secure_port,
106 };
107
108 struct svc_xprt_class svc_rdma_bc_class = {
109 .xcl_name = "rdma-bc",
110 .xcl_owner = THIS_MODULE,
111 .xcl_ops = &svc_rdma_bc_ops,
112 .xcl_max_payload = (1024 - RPCRDMA_HDRLEN_MIN)
113 };
114
svc_rdma_bc_create(struct svc_serv * serv,struct net * net,struct sockaddr * sa,int salen,int flags)115 static struct svc_xprt *svc_rdma_bc_create(struct svc_serv *serv,
116 struct net *net,
117 struct sockaddr *sa, int salen,
118 int flags)
119 {
120 struct svcxprt_rdma *cma_xprt;
121 struct svc_xprt *xprt;
122
123 cma_xprt = rdma_create_xprt(serv, 0);
124 if (!cma_xprt)
125 return ERR_PTR(-ENOMEM);
126 xprt = &cma_xprt->sc_xprt;
127
128 svc_xprt_init(net, &svc_rdma_bc_class, xprt, serv);
129 serv->sv_bc_xprt = xprt;
130
131 dprintk("svcrdma: %s(%p)\n", __func__, xprt);
132 return xprt;
133 }
134
svc_rdma_bc_detach(struct svc_xprt * xprt)135 static void svc_rdma_bc_detach(struct svc_xprt *xprt)
136 {
137 dprintk("svcrdma: %s(%p)\n", __func__, xprt);
138 }
139
svc_rdma_bc_free(struct svc_xprt * xprt)140 static void svc_rdma_bc_free(struct svc_xprt *xprt)
141 {
142 struct svcxprt_rdma *rdma =
143 container_of(xprt, struct svcxprt_rdma, sc_xprt);
144
145 dprintk("svcrdma: %s(%p)\n", __func__, xprt);
146 if (xprt)
147 kfree(rdma);
148 }
149 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
150
alloc_ctxt(struct svcxprt_rdma * xprt,gfp_t flags)151 static struct svc_rdma_op_ctxt *alloc_ctxt(struct svcxprt_rdma *xprt,
152 gfp_t flags)
153 {
154 struct svc_rdma_op_ctxt *ctxt;
155
156 ctxt = kmalloc(sizeof(*ctxt), flags);
157 if (ctxt) {
158 ctxt->xprt = xprt;
159 INIT_LIST_HEAD(&ctxt->free);
160 INIT_LIST_HEAD(&ctxt->dto_q);
161 }
162 return ctxt;
163 }
164
svc_rdma_prealloc_ctxts(struct svcxprt_rdma * xprt)165 static bool svc_rdma_prealloc_ctxts(struct svcxprt_rdma *xprt)
166 {
167 unsigned int i;
168
169 /* Each RPC/RDMA credit can consume a number of send
170 * and receive WQEs. One ctxt is allocated for each.
171 */
172 i = xprt->sc_sq_depth + xprt->sc_rq_depth;
173
174 while (i--) {
175 struct svc_rdma_op_ctxt *ctxt;
176
177 ctxt = alloc_ctxt(xprt, GFP_KERNEL);
178 if (!ctxt) {
179 dprintk("svcrdma: No memory for RDMA ctxt\n");
180 return false;
181 }
182 list_add(&ctxt->free, &xprt->sc_ctxts);
183 }
184 return true;
185 }
186
svc_rdma_get_context(struct svcxprt_rdma * xprt)187 struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *xprt)
188 {
189 struct svc_rdma_op_ctxt *ctxt = NULL;
190
191 spin_lock_bh(&xprt->sc_ctxt_lock);
192 xprt->sc_ctxt_used++;
193 if (list_empty(&xprt->sc_ctxts))
194 goto out_empty;
195
196 ctxt = list_first_entry(&xprt->sc_ctxts,
197 struct svc_rdma_op_ctxt, free);
198 list_del_init(&ctxt->free);
199 spin_unlock_bh(&xprt->sc_ctxt_lock);
200
201 out:
202 ctxt->count = 0;
203 ctxt->mapped_sges = 0;
204 ctxt->frmr = NULL;
205 return ctxt;
206
207 out_empty:
208 /* Either pre-allocation missed the mark, or send
209 * queue accounting is broken.
210 */
211 spin_unlock_bh(&xprt->sc_ctxt_lock);
212
213 ctxt = alloc_ctxt(xprt, GFP_NOIO);
214 if (ctxt)
215 goto out;
216
217 spin_lock_bh(&xprt->sc_ctxt_lock);
218 xprt->sc_ctxt_used--;
219 spin_unlock_bh(&xprt->sc_ctxt_lock);
220 WARN_ONCE(1, "svcrdma: empty RDMA ctxt list?\n");
221 return NULL;
222 }
223
svc_rdma_unmap_dma(struct svc_rdma_op_ctxt * ctxt)224 void svc_rdma_unmap_dma(struct svc_rdma_op_ctxt *ctxt)
225 {
226 struct svcxprt_rdma *xprt = ctxt->xprt;
227 struct ib_device *device = xprt->sc_cm_id->device;
228 u32 lkey = xprt->sc_pd->local_dma_lkey;
229 unsigned int i, count;
230
231 for (count = 0, i = 0; i < ctxt->mapped_sges; i++) {
232 /*
233 * Unmap the DMA addr in the SGE if the lkey matches
234 * the local_dma_lkey, otherwise, ignore it since it is
235 * an FRMR lkey and will be unmapped later when the
236 * last WR that uses it completes.
237 */
238 if (ctxt->sge[i].lkey == lkey) {
239 count++;
240 ib_dma_unmap_page(device,
241 ctxt->sge[i].addr,
242 ctxt->sge[i].length,
243 ctxt->direction);
244 }
245 }
246 ctxt->mapped_sges = 0;
247 atomic_sub(count, &xprt->sc_dma_used);
248 }
249
svc_rdma_put_context(struct svc_rdma_op_ctxt * ctxt,int free_pages)250 void svc_rdma_put_context(struct svc_rdma_op_ctxt *ctxt, int free_pages)
251 {
252 struct svcxprt_rdma *xprt = ctxt->xprt;
253 int i;
254
255 if (free_pages)
256 for (i = 0; i < ctxt->count; i++)
257 put_page(ctxt->pages[i]);
258
259 spin_lock_bh(&xprt->sc_ctxt_lock);
260 xprt->sc_ctxt_used--;
261 list_add(&ctxt->free, &xprt->sc_ctxts);
262 spin_unlock_bh(&xprt->sc_ctxt_lock);
263 }
264
svc_rdma_destroy_ctxts(struct svcxprt_rdma * xprt)265 static void svc_rdma_destroy_ctxts(struct svcxprt_rdma *xprt)
266 {
267 while (!list_empty(&xprt->sc_ctxts)) {
268 struct svc_rdma_op_ctxt *ctxt;
269
270 ctxt = list_first_entry(&xprt->sc_ctxts,
271 struct svc_rdma_op_ctxt, free);
272 list_del(&ctxt->free);
273 kfree(ctxt);
274 }
275 }
276
alloc_req_map(gfp_t flags)277 static struct svc_rdma_req_map *alloc_req_map(gfp_t flags)
278 {
279 struct svc_rdma_req_map *map;
280
281 map = kmalloc(sizeof(*map), flags);
282 if (map)
283 INIT_LIST_HEAD(&map->free);
284 return map;
285 }
286
svc_rdma_prealloc_maps(struct svcxprt_rdma * xprt)287 static bool svc_rdma_prealloc_maps(struct svcxprt_rdma *xprt)
288 {
289 unsigned int i;
290
291 /* One for each receive buffer on this connection. */
292 i = xprt->sc_max_requests;
293
294 while (i--) {
295 struct svc_rdma_req_map *map;
296
297 map = alloc_req_map(GFP_KERNEL);
298 if (!map) {
299 dprintk("svcrdma: No memory for request map\n");
300 return false;
301 }
302 list_add(&map->free, &xprt->sc_maps);
303 }
304 return true;
305 }
306
svc_rdma_get_req_map(struct svcxprt_rdma * xprt)307 struct svc_rdma_req_map *svc_rdma_get_req_map(struct svcxprt_rdma *xprt)
308 {
309 struct svc_rdma_req_map *map = NULL;
310
311 spin_lock(&xprt->sc_map_lock);
312 if (list_empty(&xprt->sc_maps))
313 goto out_empty;
314
315 map = list_first_entry(&xprt->sc_maps,
316 struct svc_rdma_req_map, free);
317 list_del_init(&map->free);
318 spin_unlock(&xprt->sc_map_lock);
319
320 out:
321 map->count = 0;
322 return map;
323
324 out_empty:
325 spin_unlock(&xprt->sc_map_lock);
326
327 /* Pre-allocation amount was incorrect */
328 map = alloc_req_map(GFP_NOIO);
329 if (map)
330 goto out;
331
332 WARN_ONCE(1, "svcrdma: empty request map list?\n");
333 return NULL;
334 }
335
svc_rdma_put_req_map(struct svcxprt_rdma * xprt,struct svc_rdma_req_map * map)336 void svc_rdma_put_req_map(struct svcxprt_rdma *xprt,
337 struct svc_rdma_req_map *map)
338 {
339 spin_lock(&xprt->sc_map_lock);
340 list_add(&map->free, &xprt->sc_maps);
341 spin_unlock(&xprt->sc_map_lock);
342 }
343
svc_rdma_destroy_maps(struct svcxprt_rdma * xprt)344 static void svc_rdma_destroy_maps(struct svcxprt_rdma *xprt)
345 {
346 while (!list_empty(&xprt->sc_maps)) {
347 struct svc_rdma_req_map *map;
348
349 map = list_first_entry(&xprt->sc_maps,
350 struct svc_rdma_req_map, free);
351 list_del(&map->free);
352 kfree(map);
353 }
354 }
355
356 /* QP event handler */
qp_event_handler(struct ib_event * event,void * context)357 static void qp_event_handler(struct ib_event *event, void *context)
358 {
359 struct svc_xprt *xprt = context;
360
361 switch (event->event) {
362 /* These are considered benign events */
363 case IB_EVENT_PATH_MIG:
364 case IB_EVENT_COMM_EST:
365 case IB_EVENT_SQ_DRAINED:
366 case IB_EVENT_QP_LAST_WQE_REACHED:
367 dprintk("svcrdma: QP event %s (%d) received for QP=%p\n",
368 ib_event_msg(event->event), event->event,
369 event->element.qp);
370 break;
371 /* These are considered fatal events */
372 case IB_EVENT_PATH_MIG_ERR:
373 case IB_EVENT_QP_FATAL:
374 case IB_EVENT_QP_REQ_ERR:
375 case IB_EVENT_QP_ACCESS_ERR:
376 case IB_EVENT_DEVICE_FATAL:
377 default:
378 dprintk("svcrdma: QP ERROR event %s (%d) received for QP=%p, "
379 "closing transport\n",
380 ib_event_msg(event->event), event->event,
381 event->element.qp);
382 set_bit(XPT_CLOSE, &xprt->xpt_flags);
383 break;
384 }
385 }
386
387 /**
388 * svc_rdma_wc_receive - Invoked by RDMA provider for each polled Receive WC
389 * @cq: completion queue
390 * @wc: completed WR
391 *
392 */
svc_rdma_wc_receive(struct ib_cq * cq,struct ib_wc * wc)393 static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
394 {
395 struct svcxprt_rdma *xprt = cq->cq_context;
396 struct ib_cqe *cqe = wc->wr_cqe;
397 struct svc_rdma_op_ctxt *ctxt;
398
399 /* WARNING: Only wc->wr_cqe and wc->status are reliable */
400 ctxt = container_of(cqe, struct svc_rdma_op_ctxt, cqe);
401 ctxt->wc_status = wc->status;
402 svc_rdma_unmap_dma(ctxt);
403
404 if (wc->status != IB_WC_SUCCESS)
405 goto flushed;
406
407 /* All wc fields are now known to be valid */
408 ctxt->byte_len = wc->byte_len;
409 spin_lock(&xprt->sc_rq_dto_lock);
410 list_add_tail(&ctxt->dto_q, &xprt->sc_rq_dto_q);
411 spin_unlock(&xprt->sc_rq_dto_lock);
412
413 set_bit(XPT_DATA, &xprt->sc_xprt.xpt_flags);
414 if (test_bit(RDMAXPRT_CONN_PENDING, &xprt->sc_flags))
415 goto out;
416 svc_xprt_enqueue(&xprt->sc_xprt);
417 goto out;
418
419 flushed:
420 if (wc->status != IB_WC_WR_FLUSH_ERR)
421 pr_warn("svcrdma: receive: %s (%u/0x%x)\n",
422 ib_wc_status_msg(wc->status),
423 wc->status, wc->vendor_err);
424 set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
425 svc_rdma_put_context(ctxt, 1);
426
427 out:
428 svc_xprt_put(&xprt->sc_xprt);
429 }
430
svc_rdma_send_wc_common(struct svcxprt_rdma * xprt,struct ib_wc * wc,const char * opname)431 static void svc_rdma_send_wc_common(struct svcxprt_rdma *xprt,
432 struct ib_wc *wc,
433 const char *opname)
434 {
435 if (wc->status != IB_WC_SUCCESS)
436 goto err;
437
438 out:
439 atomic_dec(&xprt->sc_sq_count);
440 wake_up(&xprt->sc_send_wait);
441 return;
442
443 err:
444 set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
445 if (wc->status != IB_WC_WR_FLUSH_ERR)
446 pr_err("svcrdma: %s: %s (%u/0x%x)\n",
447 opname, ib_wc_status_msg(wc->status),
448 wc->status, wc->vendor_err);
449 goto out;
450 }
451
svc_rdma_send_wc_common_put(struct ib_cq * cq,struct ib_wc * wc,const char * opname)452 static void svc_rdma_send_wc_common_put(struct ib_cq *cq, struct ib_wc *wc,
453 const char *opname)
454 {
455 struct svcxprt_rdma *xprt = cq->cq_context;
456
457 svc_rdma_send_wc_common(xprt, wc, opname);
458 svc_xprt_put(&xprt->sc_xprt);
459 }
460
461 /**
462 * svc_rdma_wc_send - Invoked by RDMA provider for each polled Send WC
463 * @cq: completion queue
464 * @wc: completed WR
465 *
466 */
svc_rdma_wc_send(struct ib_cq * cq,struct ib_wc * wc)467 void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc)
468 {
469 struct ib_cqe *cqe = wc->wr_cqe;
470 struct svc_rdma_op_ctxt *ctxt;
471
472 svc_rdma_send_wc_common_put(cq, wc, "send");
473
474 ctxt = container_of(cqe, struct svc_rdma_op_ctxt, cqe);
475 svc_rdma_unmap_dma(ctxt);
476 svc_rdma_put_context(ctxt, 1);
477 }
478
479 /**
480 * svc_rdma_wc_write - Invoked by RDMA provider for each polled Write WC
481 * @cq: completion queue
482 * @wc: completed WR
483 *
484 */
svc_rdma_wc_write(struct ib_cq * cq,struct ib_wc * wc)485 void svc_rdma_wc_write(struct ib_cq *cq, struct ib_wc *wc)
486 {
487 struct ib_cqe *cqe = wc->wr_cqe;
488 struct svc_rdma_op_ctxt *ctxt;
489
490 svc_rdma_send_wc_common_put(cq, wc, "write");
491
492 ctxt = container_of(cqe, struct svc_rdma_op_ctxt, cqe);
493 svc_rdma_unmap_dma(ctxt);
494 svc_rdma_put_context(ctxt, 0);
495 }
496
497 /**
498 * svc_rdma_wc_reg - Invoked by RDMA provider for each polled FASTREG WC
499 * @cq: completion queue
500 * @wc: completed WR
501 *
502 */
svc_rdma_wc_reg(struct ib_cq * cq,struct ib_wc * wc)503 void svc_rdma_wc_reg(struct ib_cq *cq, struct ib_wc *wc)
504 {
505 svc_rdma_send_wc_common_put(cq, wc, "fastreg");
506 }
507
508 /**
509 * svc_rdma_wc_read - Invoked by RDMA provider for each polled Read WC
510 * @cq: completion queue
511 * @wc: completed WR
512 *
513 */
svc_rdma_wc_read(struct ib_cq * cq,struct ib_wc * wc)514 void svc_rdma_wc_read(struct ib_cq *cq, struct ib_wc *wc)
515 {
516 struct svcxprt_rdma *xprt = cq->cq_context;
517 struct ib_cqe *cqe = wc->wr_cqe;
518 struct svc_rdma_op_ctxt *ctxt;
519
520 svc_rdma_send_wc_common(xprt, wc, "read");
521
522 ctxt = container_of(cqe, struct svc_rdma_op_ctxt, cqe);
523 svc_rdma_unmap_dma(ctxt);
524 svc_rdma_put_frmr(xprt, ctxt->frmr);
525
526 if (test_bit(RDMACTXT_F_LAST_CTXT, &ctxt->flags)) {
527 struct svc_rdma_op_ctxt *read_hdr;
528
529 read_hdr = ctxt->read_hdr;
530 spin_lock(&xprt->sc_rq_dto_lock);
531 list_add_tail(&read_hdr->dto_q,
532 &xprt->sc_read_complete_q);
533 spin_unlock(&xprt->sc_rq_dto_lock);
534
535 set_bit(XPT_DATA, &xprt->sc_xprt.xpt_flags);
536 svc_xprt_enqueue(&xprt->sc_xprt);
537 }
538
539 svc_rdma_put_context(ctxt, 0);
540 svc_xprt_put(&xprt->sc_xprt);
541 }
542
543 /**
544 * svc_rdma_wc_inv - Invoked by RDMA provider for each polled LOCAL_INV WC
545 * @cq: completion queue
546 * @wc: completed WR
547 *
548 */
svc_rdma_wc_inv(struct ib_cq * cq,struct ib_wc * wc)549 void svc_rdma_wc_inv(struct ib_cq *cq, struct ib_wc *wc)
550 {
551 svc_rdma_send_wc_common_put(cq, wc, "localInv");
552 }
553
rdma_create_xprt(struct svc_serv * serv,int listener)554 static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
555 int listener)
556 {
557 struct svcxprt_rdma *cma_xprt = kzalloc(sizeof *cma_xprt, GFP_KERNEL);
558
559 if (!cma_xprt)
560 return NULL;
561 svc_xprt_init(&init_net, &svc_rdma_class, &cma_xprt->sc_xprt, serv);
562 INIT_LIST_HEAD(&cma_xprt->sc_accept_q);
563 INIT_LIST_HEAD(&cma_xprt->sc_dto_q);
564 INIT_LIST_HEAD(&cma_xprt->sc_rq_dto_q);
565 INIT_LIST_HEAD(&cma_xprt->sc_read_complete_q);
566 INIT_LIST_HEAD(&cma_xprt->sc_frmr_q);
567 INIT_LIST_HEAD(&cma_xprt->sc_ctxts);
568 INIT_LIST_HEAD(&cma_xprt->sc_maps);
569 init_waitqueue_head(&cma_xprt->sc_send_wait);
570
571 spin_lock_init(&cma_xprt->sc_lock);
572 spin_lock_init(&cma_xprt->sc_rq_dto_lock);
573 spin_lock_init(&cma_xprt->sc_frmr_q_lock);
574 spin_lock_init(&cma_xprt->sc_ctxt_lock);
575 spin_lock_init(&cma_xprt->sc_map_lock);
576
577 if (listener)
578 set_bit(XPT_LISTENER, &cma_xprt->sc_xprt.xpt_flags);
579
580 return cma_xprt;
581 }
582
svc_rdma_post_recv(struct svcxprt_rdma * xprt,gfp_t flags)583 int svc_rdma_post_recv(struct svcxprt_rdma *xprt, gfp_t flags)
584 {
585 struct ib_recv_wr recv_wr, *bad_recv_wr;
586 struct svc_rdma_op_ctxt *ctxt;
587 struct page *page;
588 dma_addr_t pa;
589 int sge_no;
590 int buflen;
591 int ret;
592
593 ctxt = svc_rdma_get_context(xprt);
594 buflen = 0;
595 ctxt->direction = DMA_FROM_DEVICE;
596 ctxt->cqe.done = svc_rdma_wc_receive;
597 for (sge_no = 0; buflen < xprt->sc_max_req_size; sge_no++) {
598 if (sge_no >= xprt->sc_max_sge) {
599 pr_err("svcrdma: Too many sges (%d)\n", sge_no);
600 goto err_put_ctxt;
601 }
602 page = alloc_page(flags);
603 if (!page)
604 goto err_put_ctxt;
605 ctxt->pages[sge_no] = page;
606 pa = ib_dma_map_page(xprt->sc_cm_id->device,
607 page, 0, PAGE_SIZE,
608 DMA_FROM_DEVICE);
609 if (ib_dma_mapping_error(xprt->sc_cm_id->device, pa))
610 goto err_put_ctxt;
611 svc_rdma_count_mappings(xprt, ctxt);
612 ctxt->sge[sge_no].addr = pa;
613 ctxt->sge[sge_no].length = PAGE_SIZE;
614 ctxt->sge[sge_no].lkey = xprt->sc_pd->local_dma_lkey;
615 ctxt->count = sge_no + 1;
616 buflen += PAGE_SIZE;
617 }
618 recv_wr.next = NULL;
619 recv_wr.sg_list = &ctxt->sge[0];
620 recv_wr.num_sge = ctxt->count;
621 recv_wr.wr_cqe = &ctxt->cqe;
622
623 svc_xprt_get(&xprt->sc_xprt);
624 ret = ib_post_recv(xprt->sc_qp, &recv_wr, &bad_recv_wr);
625 if (ret) {
626 svc_rdma_unmap_dma(ctxt);
627 svc_rdma_put_context(ctxt, 1);
628 svc_xprt_put(&xprt->sc_xprt);
629 }
630 return ret;
631
632 err_put_ctxt:
633 svc_rdma_unmap_dma(ctxt);
634 svc_rdma_put_context(ctxt, 1);
635 return -ENOMEM;
636 }
637
svc_rdma_repost_recv(struct svcxprt_rdma * xprt,gfp_t flags)638 int svc_rdma_repost_recv(struct svcxprt_rdma *xprt, gfp_t flags)
639 {
640 int ret = 0;
641
642 ret = svc_rdma_post_recv(xprt, flags);
643 if (ret) {
644 pr_err("svcrdma: could not post a receive buffer, err=%d.\n",
645 ret);
646 pr_err("svcrdma: closing transport %p.\n", xprt);
647 set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
648 ret = -ENOTCONN;
649 }
650 return ret;
651 }
652
653 static void
svc_rdma_parse_connect_private(struct svcxprt_rdma * newxprt,struct rdma_conn_param * param)654 svc_rdma_parse_connect_private(struct svcxprt_rdma *newxprt,
655 struct rdma_conn_param *param)
656 {
657 const struct rpcrdma_connect_private *pmsg = param->private_data;
658
659 if (pmsg &&
660 pmsg->cp_magic == rpcrdma_cmp_magic &&
661 pmsg->cp_version == RPCRDMA_CMP_VERSION) {
662 newxprt->sc_snd_w_inv = pmsg->cp_flags &
663 RPCRDMA_CMP_F_SND_W_INV_OK;
664
665 dprintk("svcrdma: client send_size %u, recv_size %u "
666 "remote inv %ssupported\n",
667 rpcrdma_decode_buffer_size(pmsg->cp_send_size),
668 rpcrdma_decode_buffer_size(pmsg->cp_recv_size),
669 newxprt->sc_snd_w_inv ? "" : "un");
670 }
671 }
672
673 /*
674 * This function handles the CONNECT_REQUEST event on a listening
675 * endpoint. It is passed the cma_id for the _new_ connection. The context in
676 * this cma_id is inherited from the listening cma_id and is the svc_xprt
677 * structure for the listening endpoint.
678 *
679 * This function creates a new xprt for the new connection and enqueues it on
680 * the accept queue for the listent xprt. When the listen thread is kicked, it
681 * will call the recvfrom method on the listen xprt which will accept the new
682 * connection.
683 */
handle_connect_req(struct rdma_cm_id * new_cma_id,struct rdma_conn_param * param)684 static void handle_connect_req(struct rdma_cm_id *new_cma_id,
685 struct rdma_conn_param *param)
686 {
687 struct svcxprt_rdma *listen_xprt = new_cma_id->context;
688 struct svcxprt_rdma *newxprt;
689 struct sockaddr *sa;
690
691 /* Create a new transport */
692 newxprt = rdma_create_xprt(listen_xprt->sc_xprt.xpt_server, 0);
693 if (!newxprt) {
694 dprintk("svcrdma: failed to create new transport\n");
695 return;
696 }
697 newxprt->sc_cm_id = new_cma_id;
698 new_cma_id->context = newxprt;
699 dprintk("svcrdma: Creating newxprt=%p, cm_id=%p, listenxprt=%p\n",
700 newxprt, newxprt->sc_cm_id, listen_xprt);
701 svc_rdma_parse_connect_private(newxprt, param);
702
703 /* Save client advertised inbound read limit for use later in accept. */
704 newxprt->sc_ord = param->initiator_depth;
705
706 /* Set the local and remote addresses in the transport */
707 sa = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.dst_addr;
708 svc_xprt_set_remote(&newxprt->sc_xprt, sa, svc_addr_len(sa));
709 sa = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.src_addr;
710 svc_xprt_set_local(&newxprt->sc_xprt, sa, svc_addr_len(sa));
711
712 /*
713 * Enqueue the new transport on the accept queue of the listening
714 * transport
715 */
716 spin_lock_bh(&listen_xprt->sc_lock);
717 list_add_tail(&newxprt->sc_accept_q, &listen_xprt->sc_accept_q);
718 spin_unlock_bh(&listen_xprt->sc_lock);
719
720 set_bit(XPT_CONN, &listen_xprt->sc_xprt.xpt_flags);
721 svc_xprt_enqueue(&listen_xprt->sc_xprt);
722 }
723
724 /*
725 * Handles events generated on the listening endpoint. These events will be
726 * either be incoming connect requests or adapter removal events.
727 */
rdma_listen_handler(struct rdma_cm_id * cma_id,struct rdma_cm_event * event)728 static int rdma_listen_handler(struct rdma_cm_id *cma_id,
729 struct rdma_cm_event *event)
730 {
731 struct svcxprt_rdma *xprt = cma_id->context;
732 int ret = 0;
733
734 switch (event->event) {
735 case RDMA_CM_EVENT_CONNECT_REQUEST:
736 dprintk("svcrdma: Connect request on cma_id=%p, xprt = %p, "
737 "event = %s (%d)\n", cma_id, cma_id->context,
738 rdma_event_msg(event->event), event->event);
739 handle_connect_req(cma_id, &event->param.conn);
740 break;
741
742 case RDMA_CM_EVENT_ESTABLISHED:
743 /* Accept complete */
744 dprintk("svcrdma: Connection completed on LISTEN xprt=%p, "
745 "cm_id=%p\n", xprt, cma_id);
746 break;
747
748 case RDMA_CM_EVENT_DEVICE_REMOVAL:
749 dprintk("svcrdma: Device removal xprt=%p, cm_id=%p\n",
750 xprt, cma_id);
751 if (xprt)
752 set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
753 break;
754
755 default:
756 dprintk("svcrdma: Unexpected event on listening endpoint %p, "
757 "event = %s (%d)\n", cma_id,
758 rdma_event_msg(event->event), event->event);
759 break;
760 }
761
762 return ret;
763 }
764
rdma_cma_handler(struct rdma_cm_id * cma_id,struct rdma_cm_event * event)765 static int rdma_cma_handler(struct rdma_cm_id *cma_id,
766 struct rdma_cm_event *event)
767 {
768 struct svc_xprt *xprt = cma_id->context;
769 struct svcxprt_rdma *rdma =
770 container_of(xprt, struct svcxprt_rdma, sc_xprt);
771 switch (event->event) {
772 case RDMA_CM_EVENT_ESTABLISHED:
773 /* Accept complete */
774 svc_xprt_get(xprt);
775 dprintk("svcrdma: Connection completed on DTO xprt=%p, "
776 "cm_id=%p\n", xprt, cma_id);
777 clear_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags);
778 svc_xprt_enqueue(xprt);
779 break;
780 case RDMA_CM_EVENT_DISCONNECTED:
781 dprintk("svcrdma: Disconnect on DTO xprt=%p, cm_id=%p\n",
782 xprt, cma_id);
783 if (xprt) {
784 set_bit(XPT_CLOSE, &xprt->xpt_flags);
785 svc_xprt_enqueue(xprt);
786 svc_xprt_put(xprt);
787 }
788 break;
789 case RDMA_CM_EVENT_DEVICE_REMOVAL:
790 dprintk("svcrdma: Device removal cma_id=%p, xprt = %p, "
791 "event = %s (%d)\n", cma_id, xprt,
792 rdma_event_msg(event->event), event->event);
793 if (xprt) {
794 set_bit(XPT_CLOSE, &xprt->xpt_flags);
795 svc_xprt_enqueue(xprt);
796 svc_xprt_put(xprt);
797 }
798 break;
799 default:
800 dprintk("svcrdma: Unexpected event on DTO endpoint %p, "
801 "event = %s (%d)\n", cma_id,
802 rdma_event_msg(event->event), event->event);
803 break;
804 }
805 return 0;
806 }
807
808 /*
809 * Create a listening RDMA service endpoint.
810 */
svc_rdma_create(struct svc_serv * serv,struct net * net,struct sockaddr * sa,int salen,int flags)811 static struct svc_xprt *svc_rdma_create(struct svc_serv *serv,
812 struct net *net,
813 struct sockaddr *sa, int salen,
814 int flags)
815 {
816 struct rdma_cm_id *listen_id;
817 struct svcxprt_rdma *cma_xprt;
818 int ret;
819
820 dprintk("svcrdma: Creating RDMA socket\n");
821 if ((sa->sa_family != AF_INET) && (sa->sa_family != AF_INET6)) {
822 dprintk("svcrdma: Address family %d is not supported.\n", sa->sa_family);
823 return ERR_PTR(-EAFNOSUPPORT);
824 }
825 cma_xprt = rdma_create_xprt(serv, 1);
826 if (!cma_xprt)
827 return ERR_PTR(-ENOMEM);
828
829 listen_id = rdma_create_id(&init_net, rdma_listen_handler, cma_xprt,
830 RDMA_PS_TCP, IB_QPT_RC);
831 if (IS_ERR(listen_id)) {
832 ret = PTR_ERR(listen_id);
833 dprintk("svcrdma: rdma_create_id failed = %d\n", ret);
834 goto err0;
835 }
836
837 /* Allow both IPv4 and IPv6 sockets to bind a single port
838 * at the same time.
839 */
840 #if IS_ENABLED(CONFIG_IPV6)
841 ret = rdma_set_afonly(listen_id, 1);
842 if (ret) {
843 dprintk("svcrdma: rdma_set_afonly failed = %d\n", ret);
844 goto err1;
845 }
846 #endif
847 ret = rdma_bind_addr(listen_id, sa);
848 if (ret) {
849 dprintk("svcrdma: rdma_bind_addr failed = %d\n", ret);
850 goto err1;
851 }
852 cma_xprt->sc_cm_id = listen_id;
853
854 ret = rdma_listen(listen_id, RPCRDMA_LISTEN_BACKLOG);
855 if (ret) {
856 dprintk("svcrdma: rdma_listen failed = %d\n", ret);
857 goto err1;
858 }
859
860 /*
861 * We need to use the address from the cm_id in case the
862 * caller specified 0 for the port number.
863 */
864 sa = (struct sockaddr *)&cma_xprt->sc_cm_id->route.addr.src_addr;
865 svc_xprt_set_local(&cma_xprt->sc_xprt, sa, salen);
866
867 return &cma_xprt->sc_xprt;
868
869 err1:
870 rdma_destroy_id(listen_id);
871 err0:
872 kfree(cma_xprt);
873 return ERR_PTR(ret);
874 }
875
rdma_alloc_frmr(struct svcxprt_rdma * xprt)876 static struct svc_rdma_fastreg_mr *rdma_alloc_frmr(struct svcxprt_rdma *xprt)
877 {
878 struct ib_mr *mr;
879 struct scatterlist *sg;
880 struct svc_rdma_fastreg_mr *frmr;
881 u32 num_sg;
882
883 frmr = kmalloc(sizeof(*frmr), GFP_KERNEL);
884 if (!frmr)
885 goto err;
886
887 num_sg = min_t(u32, RPCSVC_MAXPAGES, xprt->sc_frmr_pg_list_len);
888 mr = ib_alloc_mr(xprt->sc_pd, IB_MR_TYPE_MEM_REG, num_sg);
889 if (IS_ERR(mr))
890 goto err_free_frmr;
891
892 sg = kcalloc(RPCSVC_MAXPAGES, sizeof(*sg), GFP_KERNEL);
893 if (!sg)
894 goto err_free_mr;
895
896 sg_init_table(sg, RPCSVC_MAXPAGES);
897
898 frmr->mr = mr;
899 frmr->sg = sg;
900 INIT_LIST_HEAD(&frmr->frmr_list);
901 return frmr;
902
903 err_free_mr:
904 ib_dereg_mr(mr);
905 err_free_frmr:
906 kfree(frmr);
907 err:
908 return ERR_PTR(-ENOMEM);
909 }
910
rdma_dealloc_frmr_q(struct svcxprt_rdma * xprt)911 static void rdma_dealloc_frmr_q(struct svcxprt_rdma *xprt)
912 {
913 struct svc_rdma_fastreg_mr *frmr;
914
915 while (!list_empty(&xprt->sc_frmr_q)) {
916 frmr = list_entry(xprt->sc_frmr_q.next,
917 struct svc_rdma_fastreg_mr, frmr_list);
918 list_del_init(&frmr->frmr_list);
919 kfree(frmr->sg);
920 ib_dereg_mr(frmr->mr);
921 kfree(frmr);
922 }
923 }
924
svc_rdma_get_frmr(struct svcxprt_rdma * rdma)925 struct svc_rdma_fastreg_mr *svc_rdma_get_frmr(struct svcxprt_rdma *rdma)
926 {
927 struct svc_rdma_fastreg_mr *frmr = NULL;
928
929 spin_lock_bh(&rdma->sc_frmr_q_lock);
930 if (!list_empty(&rdma->sc_frmr_q)) {
931 frmr = list_entry(rdma->sc_frmr_q.next,
932 struct svc_rdma_fastreg_mr, frmr_list);
933 list_del_init(&frmr->frmr_list);
934 frmr->sg_nents = 0;
935 }
936 spin_unlock_bh(&rdma->sc_frmr_q_lock);
937 if (frmr)
938 return frmr;
939
940 return rdma_alloc_frmr(rdma);
941 }
942
svc_rdma_put_frmr(struct svcxprt_rdma * rdma,struct svc_rdma_fastreg_mr * frmr)943 void svc_rdma_put_frmr(struct svcxprt_rdma *rdma,
944 struct svc_rdma_fastreg_mr *frmr)
945 {
946 if (frmr) {
947 ib_dma_unmap_sg(rdma->sc_cm_id->device,
948 frmr->sg, frmr->sg_nents, frmr->direction);
949 atomic_dec(&rdma->sc_dma_used);
950 spin_lock_bh(&rdma->sc_frmr_q_lock);
951 WARN_ON_ONCE(!list_empty(&frmr->frmr_list));
952 list_add(&frmr->frmr_list, &rdma->sc_frmr_q);
953 spin_unlock_bh(&rdma->sc_frmr_q_lock);
954 }
955 }
956
957 /*
958 * This is the xpo_recvfrom function for listening endpoints. Its
959 * purpose is to accept incoming connections. The CMA callback handler
960 * has already created a new transport and attached it to the new CMA
961 * ID.
962 *
963 * There is a queue of pending connections hung on the listening
964 * transport. This queue contains the new svc_xprt structure. This
965 * function takes svc_xprt structures off the accept_q and completes
966 * the connection.
967 */
svc_rdma_accept(struct svc_xprt * xprt)968 static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
969 {
970 struct svcxprt_rdma *listen_rdma;
971 struct svcxprt_rdma *newxprt = NULL;
972 struct rdma_conn_param conn_param;
973 struct rpcrdma_connect_private pmsg;
974 struct ib_qp_init_attr qp_attr;
975 struct ib_device *dev;
976 unsigned int i;
977 int ret = 0;
978
979 listen_rdma = container_of(xprt, struct svcxprt_rdma, sc_xprt);
980 clear_bit(XPT_CONN, &xprt->xpt_flags);
981 /* Get the next entry off the accept list */
982 spin_lock_bh(&listen_rdma->sc_lock);
983 if (!list_empty(&listen_rdma->sc_accept_q)) {
984 newxprt = list_entry(listen_rdma->sc_accept_q.next,
985 struct svcxprt_rdma, sc_accept_q);
986 list_del_init(&newxprt->sc_accept_q);
987 }
988 if (!list_empty(&listen_rdma->sc_accept_q))
989 set_bit(XPT_CONN, &listen_rdma->sc_xprt.xpt_flags);
990 spin_unlock_bh(&listen_rdma->sc_lock);
991 if (!newxprt)
992 return NULL;
993
994 dprintk("svcrdma: newxprt from accept queue = %p, cm_id=%p\n",
995 newxprt, newxprt->sc_cm_id);
996
997 dev = newxprt->sc_cm_id->device;
998
999 /* Qualify the transport resource defaults with the
1000 * capabilities of this particular device */
1001 newxprt->sc_max_sge = min((size_t)dev->attrs.max_sge,
1002 (size_t)RPCSVC_MAXPAGES);
1003 newxprt->sc_max_sge_rd = min_t(size_t, dev->attrs.max_sge_rd,
1004 RPCSVC_MAXPAGES);
1005 newxprt->sc_max_req_size = svcrdma_max_req_size;
1006 newxprt->sc_max_requests = min_t(u32, dev->attrs.max_qp_wr,
1007 svcrdma_max_requests);
1008 newxprt->sc_max_bc_requests = min_t(u32, dev->attrs.max_qp_wr,
1009 svcrdma_max_bc_requests);
1010 newxprt->sc_rq_depth = newxprt->sc_max_requests +
1011 newxprt->sc_max_bc_requests;
1012 newxprt->sc_sq_depth = RPCRDMA_SQ_DEPTH_MULT * newxprt->sc_rq_depth;
1013
1014 if (!svc_rdma_prealloc_ctxts(newxprt))
1015 goto errout;
1016 if (!svc_rdma_prealloc_maps(newxprt))
1017 goto errout;
1018
1019 /*
1020 * Limit ORD based on client limit, local device limit, and
1021 * configured svcrdma limit.
1022 */
1023 newxprt->sc_ord = min_t(size_t, dev->attrs.max_qp_rd_atom, newxprt->sc_ord);
1024 newxprt->sc_ord = min_t(size_t, svcrdma_ord, newxprt->sc_ord);
1025
1026 newxprt->sc_pd = ib_alloc_pd(dev, 0);
1027 if (IS_ERR(newxprt->sc_pd)) {
1028 dprintk("svcrdma: error creating PD for connect request\n");
1029 goto errout;
1030 }
1031 newxprt->sc_sq_cq = ib_alloc_cq(dev, newxprt, newxprt->sc_sq_depth,
1032 0, IB_POLL_SOFTIRQ);
1033 if (IS_ERR(newxprt->sc_sq_cq)) {
1034 dprintk("svcrdma: error creating SQ CQ for connect request\n");
1035 goto errout;
1036 }
1037 newxprt->sc_rq_cq = ib_alloc_cq(dev, newxprt, newxprt->sc_rq_depth,
1038 0, IB_POLL_SOFTIRQ);
1039 if (IS_ERR(newxprt->sc_rq_cq)) {
1040 dprintk("svcrdma: error creating RQ CQ for connect request\n");
1041 goto errout;
1042 }
1043
1044 memset(&qp_attr, 0, sizeof qp_attr);
1045 qp_attr.event_handler = qp_event_handler;
1046 qp_attr.qp_context = &newxprt->sc_xprt;
1047 qp_attr.cap.max_send_wr = newxprt->sc_sq_depth;
1048 qp_attr.cap.max_recv_wr = newxprt->sc_rq_depth;
1049 qp_attr.cap.max_send_sge = newxprt->sc_max_sge;
1050 qp_attr.cap.max_recv_sge = newxprt->sc_max_sge;
1051 qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
1052 qp_attr.qp_type = IB_QPT_RC;
1053 qp_attr.send_cq = newxprt->sc_sq_cq;
1054 qp_attr.recv_cq = newxprt->sc_rq_cq;
1055 dprintk("svcrdma: newxprt->sc_cm_id=%p, newxprt->sc_pd=%p\n"
1056 " cm_id->device=%p, sc_pd->device=%p\n"
1057 " cap.max_send_wr = %d\n"
1058 " cap.max_recv_wr = %d\n"
1059 " cap.max_send_sge = %d\n"
1060 " cap.max_recv_sge = %d\n",
1061 newxprt->sc_cm_id, newxprt->sc_pd,
1062 dev, newxprt->sc_pd->device,
1063 qp_attr.cap.max_send_wr,
1064 qp_attr.cap.max_recv_wr,
1065 qp_attr.cap.max_send_sge,
1066 qp_attr.cap.max_recv_sge);
1067
1068 ret = rdma_create_qp(newxprt->sc_cm_id, newxprt->sc_pd, &qp_attr);
1069 if (ret) {
1070 dprintk("svcrdma: failed to create QP, ret=%d\n", ret);
1071 goto errout;
1072 }
1073 newxprt->sc_qp = newxprt->sc_cm_id->qp;
1074
1075 /*
1076 * Use the most secure set of MR resources based on the
1077 * transport type and available memory management features in
1078 * the device. Here's the table implemented below:
1079 *
1080 * Fast Global DMA Remote WR
1081 * Reg LKEY MR Access
1082 * Sup'd Sup'd Needed Needed
1083 *
1084 * IWARP N N Y Y
1085 * N Y Y Y
1086 * Y N Y N
1087 * Y Y N -
1088 *
1089 * IB N N Y N
1090 * N Y N -
1091 * Y N Y N
1092 * Y Y N -
1093 *
1094 * NB: iWARP requires remote write access for the data sink
1095 * of an RDMA_READ. IB does not.
1096 */
1097 newxprt->sc_reader = rdma_read_chunk_lcl;
1098 if (dev->attrs.device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS) {
1099 newxprt->sc_frmr_pg_list_len =
1100 dev->attrs.max_fast_reg_page_list_len;
1101 newxprt->sc_dev_caps |= SVCRDMA_DEVCAP_FAST_REG;
1102 newxprt->sc_reader = rdma_read_chunk_frmr;
1103 } else
1104 newxprt->sc_snd_w_inv = false;
1105
1106 /*
1107 * Determine if a DMA MR is required and if so, what privs are required
1108 */
1109 if (!rdma_protocol_iwarp(dev, newxprt->sc_cm_id->port_num) &&
1110 !rdma_ib_or_roce(dev, newxprt->sc_cm_id->port_num))
1111 goto errout;
1112
1113 if (rdma_protocol_iwarp(dev, newxprt->sc_cm_id->port_num))
1114 newxprt->sc_dev_caps |= SVCRDMA_DEVCAP_READ_W_INV;
1115
1116 /* Post receive buffers */
1117 for (i = 0; i < newxprt->sc_max_requests; i++) {
1118 ret = svc_rdma_post_recv(newxprt, GFP_KERNEL);
1119 if (ret) {
1120 dprintk("svcrdma: failure posting receive buffers\n");
1121 goto errout;
1122 }
1123 }
1124
1125 /* Swap out the handler */
1126 newxprt->sc_cm_id->event_handler = rdma_cma_handler;
1127
1128 /* Construct RDMA-CM private message */
1129 pmsg.cp_magic = rpcrdma_cmp_magic;
1130 pmsg.cp_version = RPCRDMA_CMP_VERSION;
1131 pmsg.cp_flags = 0;
1132 pmsg.cp_send_size = pmsg.cp_recv_size =
1133 rpcrdma_encode_buffer_size(newxprt->sc_max_req_size);
1134
1135 /* Accept Connection */
1136 set_bit(RDMAXPRT_CONN_PENDING, &newxprt->sc_flags);
1137 memset(&conn_param, 0, sizeof conn_param);
1138 conn_param.responder_resources = 0;
1139 conn_param.initiator_depth = newxprt->sc_ord;
1140 conn_param.private_data = &pmsg;
1141 conn_param.private_data_len = sizeof(pmsg);
1142 ret = rdma_accept(newxprt->sc_cm_id, &conn_param);
1143 if (ret) {
1144 dprintk("svcrdma: failed to accept new connection, ret=%d\n",
1145 ret);
1146 goto errout;
1147 }
1148
1149 dprintk("svcrdma: new connection %p accepted with the following "
1150 "attributes:\n"
1151 " local_ip : %pI4\n"
1152 " local_port : %d\n"
1153 " remote_ip : %pI4\n"
1154 " remote_port : %d\n"
1155 " max_sge : %d\n"
1156 " max_sge_rd : %d\n"
1157 " sq_depth : %d\n"
1158 " max_requests : %d\n"
1159 " ord : %d\n",
1160 newxprt,
1161 &((struct sockaddr_in *)&newxprt->sc_cm_id->
1162 route.addr.src_addr)->sin_addr.s_addr,
1163 ntohs(((struct sockaddr_in *)&newxprt->sc_cm_id->
1164 route.addr.src_addr)->sin_port),
1165 &((struct sockaddr_in *)&newxprt->sc_cm_id->
1166 route.addr.dst_addr)->sin_addr.s_addr,
1167 ntohs(((struct sockaddr_in *)&newxprt->sc_cm_id->
1168 route.addr.dst_addr)->sin_port),
1169 newxprt->sc_max_sge,
1170 newxprt->sc_max_sge_rd,
1171 newxprt->sc_sq_depth,
1172 newxprt->sc_max_requests,
1173 newxprt->sc_ord);
1174
1175 return &newxprt->sc_xprt;
1176
1177 errout:
1178 dprintk("svcrdma: failure accepting new connection rc=%d.\n", ret);
1179 /* Take a reference in case the DTO handler runs */
1180 svc_xprt_get(&newxprt->sc_xprt);
1181 if (newxprt->sc_qp && !IS_ERR(newxprt->sc_qp))
1182 ib_destroy_qp(newxprt->sc_qp);
1183 rdma_destroy_id(newxprt->sc_cm_id);
1184 /* This call to put will destroy the transport */
1185 svc_xprt_put(&newxprt->sc_xprt);
1186 return NULL;
1187 }
1188
svc_rdma_release_rqst(struct svc_rqst * rqstp)1189 static void svc_rdma_release_rqst(struct svc_rqst *rqstp)
1190 {
1191 }
1192
1193 /*
1194 * When connected, an svc_xprt has at least two references:
1195 *
1196 * - A reference held by the cm_id between the ESTABLISHED and
1197 * DISCONNECTED events. If the remote peer disconnected first, this
1198 * reference could be gone.
1199 *
1200 * - A reference held by the svc_recv code that called this function
1201 * as part of close processing.
1202 *
1203 * At a minimum one references should still be held.
1204 */
svc_rdma_detach(struct svc_xprt * xprt)1205 static void svc_rdma_detach(struct svc_xprt *xprt)
1206 {
1207 struct svcxprt_rdma *rdma =
1208 container_of(xprt, struct svcxprt_rdma, sc_xprt);
1209 dprintk("svc: svc_rdma_detach(%p)\n", xprt);
1210
1211 /* Disconnect and flush posted WQE */
1212 rdma_disconnect(rdma->sc_cm_id);
1213 }
1214
__svc_rdma_free(struct work_struct * work)1215 static void __svc_rdma_free(struct work_struct *work)
1216 {
1217 struct svcxprt_rdma *rdma =
1218 container_of(work, struct svcxprt_rdma, sc_work);
1219 struct svc_xprt *xprt = &rdma->sc_xprt;
1220
1221 dprintk("svcrdma: %s(%p)\n", __func__, rdma);
1222
1223 if (rdma->sc_qp && !IS_ERR(rdma->sc_qp))
1224 ib_drain_qp(rdma->sc_qp);
1225
1226 /* We should only be called from kref_put */
1227 if (atomic_read(&xprt->xpt_ref.refcount) != 0)
1228 pr_err("svcrdma: sc_xprt still in use? (%d)\n",
1229 atomic_read(&xprt->xpt_ref.refcount));
1230
1231 /*
1232 * Destroy queued, but not processed read completions. Note
1233 * that this cleanup has to be done before destroying the
1234 * cm_id because the device ptr is needed to unmap the dma in
1235 * svc_rdma_put_context.
1236 */
1237 while (!list_empty(&rdma->sc_read_complete_q)) {
1238 struct svc_rdma_op_ctxt *ctxt;
1239 ctxt = list_entry(rdma->sc_read_complete_q.next,
1240 struct svc_rdma_op_ctxt,
1241 dto_q);
1242 list_del_init(&ctxt->dto_q);
1243 svc_rdma_put_context(ctxt, 1);
1244 }
1245
1246 /* Destroy queued, but not processed recv completions */
1247 while (!list_empty(&rdma->sc_rq_dto_q)) {
1248 struct svc_rdma_op_ctxt *ctxt;
1249 ctxt = list_entry(rdma->sc_rq_dto_q.next,
1250 struct svc_rdma_op_ctxt,
1251 dto_q);
1252 list_del_init(&ctxt->dto_q);
1253 svc_rdma_put_context(ctxt, 1);
1254 }
1255
1256 /* Warn if we leaked a resource or under-referenced */
1257 if (rdma->sc_ctxt_used != 0)
1258 pr_err("svcrdma: ctxt still in use? (%d)\n",
1259 rdma->sc_ctxt_used);
1260 if (atomic_read(&rdma->sc_dma_used) != 0)
1261 pr_err("svcrdma: dma still in use? (%d)\n",
1262 atomic_read(&rdma->sc_dma_used));
1263
1264 /* Final put of backchannel client transport */
1265 if (xprt->xpt_bc_xprt) {
1266 xprt_put(xprt->xpt_bc_xprt);
1267 xprt->xpt_bc_xprt = NULL;
1268 }
1269
1270 rdma_dealloc_frmr_q(rdma);
1271 svc_rdma_destroy_ctxts(rdma);
1272 svc_rdma_destroy_maps(rdma);
1273
1274 /* Destroy the QP if present (not a listener) */
1275 if (rdma->sc_qp && !IS_ERR(rdma->sc_qp))
1276 ib_destroy_qp(rdma->sc_qp);
1277
1278 if (rdma->sc_sq_cq && !IS_ERR(rdma->sc_sq_cq))
1279 ib_free_cq(rdma->sc_sq_cq);
1280
1281 if (rdma->sc_rq_cq && !IS_ERR(rdma->sc_rq_cq))
1282 ib_free_cq(rdma->sc_rq_cq);
1283
1284 if (rdma->sc_pd && !IS_ERR(rdma->sc_pd))
1285 ib_dealloc_pd(rdma->sc_pd);
1286
1287 /* Destroy the CM ID */
1288 rdma_destroy_id(rdma->sc_cm_id);
1289
1290 kfree(rdma);
1291 }
1292
svc_rdma_free(struct svc_xprt * xprt)1293 static void svc_rdma_free(struct svc_xprt *xprt)
1294 {
1295 struct svcxprt_rdma *rdma =
1296 container_of(xprt, struct svcxprt_rdma, sc_xprt);
1297 INIT_WORK(&rdma->sc_work, __svc_rdma_free);
1298 queue_work(svc_rdma_wq, &rdma->sc_work);
1299 }
1300
svc_rdma_has_wspace(struct svc_xprt * xprt)1301 static int svc_rdma_has_wspace(struct svc_xprt *xprt)
1302 {
1303 struct svcxprt_rdma *rdma =
1304 container_of(xprt, struct svcxprt_rdma, sc_xprt);
1305
1306 /*
1307 * If there are already waiters on the SQ,
1308 * return false.
1309 */
1310 if (waitqueue_active(&rdma->sc_send_wait))
1311 return 0;
1312
1313 /* Otherwise return true. */
1314 return 1;
1315 }
1316
svc_rdma_secure_port(struct svc_rqst * rqstp)1317 static int svc_rdma_secure_port(struct svc_rqst *rqstp)
1318 {
1319 return 1;
1320 }
1321
svc_rdma_kill_temp_xprt(struct svc_xprt * xprt)1322 static void svc_rdma_kill_temp_xprt(struct svc_xprt *xprt)
1323 {
1324 }
1325
svc_rdma_send(struct svcxprt_rdma * xprt,struct ib_send_wr * wr)1326 int svc_rdma_send(struct svcxprt_rdma *xprt, struct ib_send_wr *wr)
1327 {
1328 struct ib_send_wr *bad_wr, *n_wr;
1329 int wr_count;
1330 int i;
1331 int ret;
1332
1333 if (test_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags))
1334 return -ENOTCONN;
1335
1336 wr_count = 1;
1337 for (n_wr = wr->next; n_wr; n_wr = n_wr->next)
1338 wr_count++;
1339
1340 /* If the SQ is full, wait until an SQ entry is available */
1341 while (1) {
1342 spin_lock_bh(&xprt->sc_lock);
1343 if (xprt->sc_sq_depth < atomic_read(&xprt->sc_sq_count) + wr_count) {
1344 spin_unlock_bh(&xprt->sc_lock);
1345 atomic_inc(&rdma_stat_sq_starve);
1346
1347 /* Wait until SQ WR available if SQ still full */
1348 wait_event(xprt->sc_send_wait,
1349 atomic_read(&xprt->sc_sq_count) <
1350 xprt->sc_sq_depth);
1351 if (test_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags))
1352 return -ENOTCONN;
1353 continue;
1354 }
1355 /* Take a transport ref for each WR posted */
1356 for (i = 0; i < wr_count; i++)
1357 svc_xprt_get(&xprt->sc_xprt);
1358
1359 /* Bump used SQ WR count and post */
1360 atomic_add(wr_count, &xprt->sc_sq_count);
1361 ret = ib_post_send(xprt->sc_qp, wr, &bad_wr);
1362 if (ret) {
1363 set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
1364 atomic_sub(wr_count, &xprt->sc_sq_count);
1365 for (i = 0; i < wr_count; i ++)
1366 svc_xprt_put(&xprt->sc_xprt);
1367 dprintk("svcrdma: failed to post SQ WR rc=%d, "
1368 "sc_sq_count=%d, sc_sq_depth=%d\n",
1369 ret, atomic_read(&xprt->sc_sq_count),
1370 xprt->sc_sq_depth);
1371 }
1372 spin_unlock_bh(&xprt->sc_lock);
1373 if (ret)
1374 wake_up(&xprt->sc_send_wait);
1375 break;
1376 }
1377 return ret;
1378 }
1379