projects / deliverable / linux.git / blob
? search:


03fdfce45493fbade5b3651bb84dbfaa7bccfbd3
[deliverable/linux.git] / net / sunrpc / xprtrdma / svc_rdma_transport.c
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 dto_tasklet_func(unsigned long data);
67 static void svc_rdma_detach(struct svc_xprt *xprt);
68 static void svc_rdma_free(struct svc_xprt *xprt);
69 static int svc_rdma_has_wspace(struct svc_xprt *xprt);
70 static int svc_rdma_secure_port(struct svc_rqst *);
71 static void rq_cq_reap(struct svcxprt_rdma *xprt);
72 static void sq_cq_reap(struct svcxprt_rdma *xprt);
73
74 static DECLARE_TASKLET(dto_tasklet, dto_tasklet_func, 0UL);
75 static DEFINE_SPINLOCK(dto_lock);
76 static LIST_HEAD(dto_xprt_q);
77
78 static struct svc_xprt_ops svc_rdma_ops = {
79 .xpo_create = svc_rdma_create,
80 .xpo_recvfrom = svc_rdma_recvfrom,
81 .xpo_sendto = svc_rdma_sendto,
82 .xpo_release_rqst = svc_rdma_release_rqst,
83 .xpo_detach = svc_rdma_detach,
84 .xpo_free = svc_rdma_free,
85 .xpo_prep_reply_hdr = svc_rdma_prep_reply_hdr,
86 .xpo_has_wspace = svc_rdma_has_wspace,
87 .xpo_accept = svc_rdma_accept,
88 .xpo_secure_port = svc_rdma_secure_port,
89 };
90
91 struct svc_xprt_class svc_rdma_class = {
92 .xcl_name = "rdma",
93 .xcl_owner = THIS_MODULE,
94 .xcl_ops = &svc_rdma_ops,
95 .xcl_max_payload = RPCSVC_MAXPAYLOAD_RDMA,
96 .xcl_ident = XPRT_TRANSPORT_RDMA,
97 };
98
99 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
100 static struct svc_xprt *svc_rdma_bc_create(struct svc_serv *, struct net *,
101 struct sockaddr *, int, int);
102 static void svc_rdma_bc_detach(struct svc_xprt *);
103 static void svc_rdma_bc_free(struct svc_xprt *);
104
105 static struct svc_xprt_ops svc_rdma_bc_ops = {
106 .xpo_create = svc_rdma_bc_create,
107 .xpo_detach = svc_rdma_bc_detach,
108 .xpo_free = svc_rdma_bc_free,
109 .xpo_prep_reply_hdr = svc_rdma_prep_reply_hdr,
110 .xpo_secure_port = svc_rdma_secure_port,
111 };
112
113 struct svc_xprt_class svc_rdma_bc_class = {
114 .xcl_name = "rdma-bc",
115 .xcl_owner = THIS_MODULE,
116 .xcl_ops = &svc_rdma_bc_ops,
117 .xcl_max_payload = (1024 - RPCRDMA_HDRLEN_MIN)
118 };
119
120 static struct svc_xprt *svc_rdma_bc_create(struct svc_serv *serv,
121 struct net *net,
122 struct sockaddr *sa, int salen,
123 int flags)
124 {
125 struct svcxprt_rdma *cma_xprt;
126 struct svc_xprt *xprt;
127
128 cma_xprt = rdma_create_xprt(serv, 0);
129 if (!cma_xprt)
130 return ERR_PTR(-ENOMEM);
131 xprt = &cma_xprt->sc_xprt;
132
133 svc_xprt_init(net, &svc_rdma_bc_class, xprt, serv);
134 serv->sv_bc_xprt = xprt;
135
136 dprintk("svcrdma: %s(%p)\n", __func__, xprt);
137 return xprt;
138 }
139
140 static void svc_rdma_bc_detach(struct svc_xprt *xprt)
141 {
142 dprintk("svcrdma: %s(%p)\n", __func__, xprt);
143 }
144
145 static void svc_rdma_bc_free(struct svc_xprt *xprt)
146 {
147 struct svcxprt_rdma *rdma =
148 container_of(xprt, struct svcxprt_rdma, sc_xprt);
149
150 dprintk("svcrdma: %s(%p)\n", __func__, xprt);
151 if (xprt)
152 kfree(rdma);
153 }
154 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
155
156 static struct svc_rdma_op_ctxt *alloc_ctxt(struct svcxprt_rdma *xprt,
157 gfp_t flags)
158 {
159 struct svc_rdma_op_ctxt *ctxt;
160
161 ctxt = kmalloc(sizeof(*ctxt), flags);
162 if (ctxt) {
163 ctxt->xprt = xprt;
164 INIT_LIST_HEAD(&ctxt->free);
165 INIT_LIST_HEAD(&ctxt->dto_q);
166 }
167 return ctxt;
168 }
169
170 static bool svc_rdma_prealloc_ctxts(struct svcxprt_rdma *xprt)
171 {
172 unsigned int i;
173
174 /* Each RPC/RDMA credit can consume a number of send
175 * and receive WQEs. One ctxt is allocated for each.
176 */
177 i = xprt->sc_sq_depth + xprt->sc_rq_depth;
178
179 while (i--) {
180 struct svc_rdma_op_ctxt *ctxt;
181
182 ctxt = alloc_ctxt(xprt, GFP_KERNEL);
183 if (!ctxt) {
184 dprintk("svcrdma: No memory for RDMA ctxt\n");
185 return false;
186 }
187 list_add(&ctxt->free, &xprt->sc_ctxts);
188 }
189 return true;
190 }
191
192 struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *xprt)
193 {
194 struct svc_rdma_op_ctxt *ctxt = NULL;
195
196 spin_lock_bh(&xprt->sc_ctxt_lock);
197 xprt->sc_ctxt_used++;
198 if (list_empty(&xprt->sc_ctxts))
199 goto out_empty;
200
201 ctxt = list_first_entry(&xprt->sc_ctxts,
202 struct svc_rdma_op_ctxt, free);
203 list_del_init(&ctxt->free);
204 spin_unlock_bh(&xprt->sc_ctxt_lock);
205
206 out:
207 ctxt->count = 0;
208 ctxt->frmr = NULL;
209 return ctxt;
210
211 out_empty:
212 /* Either pre-allocation missed the mark, or send
213 * queue accounting is broken.
214 */
215 spin_unlock_bh(&xprt->sc_ctxt_lock);
216
217 ctxt = alloc_ctxt(xprt, GFP_NOIO);
218 if (ctxt)
219 goto out;
220
221 spin_lock_bh(&xprt->sc_ctxt_lock);
222 xprt->sc_ctxt_used--;
223 spin_unlock_bh(&xprt->sc_ctxt_lock);
224 WARN_ONCE(1, "svcrdma: empty RDMA ctxt list?\n");
225 return NULL;
226 }
227
228 void svc_rdma_unmap_dma(struct svc_rdma_op_ctxt *ctxt)
229 {
230 struct svcxprt_rdma *xprt = ctxt->xprt;
231 int i;
232 for (i = 0; i < ctxt->count && ctxt->sge[i].length; i++) {
233 /*
234 * Unmap the DMA addr in the SGE if the lkey matches
235 * the local_dma_lkey, otherwise, ignore it since it is
236 * an FRMR lkey and will be unmapped later when the
237 * last WR that uses it completes.
238 */
239 if (ctxt->sge[i].lkey == xprt->sc_pd->local_dma_lkey) {
240 atomic_dec(&xprt->sc_dma_used);
241 ib_dma_unmap_page(xprt->sc_cm_id->device,
242 ctxt->sge[i].addr,
243 ctxt->sge[i].length,
244 ctxt->direction);
245 }
246 }
247 }
248
249 void svc_rdma_put_context(struct svc_rdma_op_ctxt *ctxt, int free_pages)
250 {
251 struct svcxprt_rdma *xprt = ctxt->xprt;
252 int i;
253
254 if (free_pages)
255 for (i = 0; i < ctxt->count; i++)
256 put_page(ctxt->pages[i]);
257
258 spin_lock_bh(&xprt->sc_ctxt_lock);
259 xprt->sc_ctxt_used--;
260 list_add(&ctxt->free, &xprt->sc_ctxts);
261 spin_unlock_bh(&xprt->sc_ctxt_lock);
262 }
263
264 static void svc_rdma_destroy_ctxts(struct svcxprt_rdma *xprt)
265 {
266 while (!list_empty(&xprt->sc_ctxts)) {
267 struct svc_rdma_op_ctxt *ctxt;
268
269 ctxt = list_first_entry(&xprt->sc_ctxts,
270 struct svc_rdma_op_ctxt, free);
271 list_del(&ctxt->free);
272 kfree(ctxt);
273 }
274 }
275
276 static struct svc_rdma_req_map *alloc_req_map(gfp_t flags)
277 {
278 struct svc_rdma_req_map *map;
279
280 map = kmalloc(sizeof(*map), flags);
281 if (map)
282 INIT_LIST_HEAD(&map->free);
283 return map;
284 }
285
286 static bool svc_rdma_prealloc_maps(struct svcxprt_rdma *xprt)
287 {
288 unsigned int i;
289
290 /* One for each receive buffer on this connection. */
291 i = xprt->sc_max_requests;
292
293 while (i--) {
294 struct svc_rdma_req_map *map;
295
296 map = alloc_req_map(GFP_KERNEL);
297 if (!map) {
298 dprintk("svcrdma: No memory for request map\n");
299 return false;
300 }
301 list_add(&map->free, &xprt->sc_maps);
302 }
303 return true;
304 }
305
306 struct svc_rdma_req_map *svc_rdma_get_req_map(struct svcxprt_rdma *xprt)
307 {
308 struct svc_rdma_req_map *map = NULL;
309
310 spin_lock(&xprt->sc_map_lock);
311 if (list_empty(&xprt->sc_maps))
312 goto out_empty;
313
314 map = list_first_entry(&xprt->sc_maps,
315 struct svc_rdma_req_map, free);
316 list_del_init(&map->free);
317 spin_unlock(&xprt->sc_map_lock);
318
319 out:
320 map->count = 0;
321 return map;
322
323 out_empty:
324 spin_unlock(&xprt->sc_map_lock);
325
326 /* Pre-allocation amount was incorrect */
327 map = alloc_req_map(GFP_NOIO);
328 if (map)
329 goto out;
330
331 WARN_ONCE(1, "svcrdma: empty request map list?\n");
332 return NULL;
333 }
334
335 void svc_rdma_put_req_map(struct svcxprt_rdma *xprt,
336 struct svc_rdma_req_map *map)
337 {
338 spin_lock(&xprt->sc_map_lock);
339 list_add(&map->free, &xprt->sc_maps);
340 spin_unlock(&xprt->sc_map_lock);
341 }
342
343 static void svc_rdma_destroy_maps(struct svcxprt_rdma *xprt)
344 {
345 while (!list_empty(&xprt->sc_maps)) {
346 struct svc_rdma_req_map *map;
347
348 map = list_first_entry(&xprt->sc_maps,
349 struct svc_rdma_req_map, free);
350 list_del(&map->free);
351 kfree(map);
352 }
353 }
354
355 /* ib_cq event handler */
356 static void cq_event_handler(struct ib_event *event, void *context)
357 {
358 struct svc_xprt *xprt = context;
359 dprintk("svcrdma: received CQ event %s (%d), context=%p\n",
360 ib_event_msg(event->event), event->event, context);
361 set_bit(XPT_CLOSE, &xprt->xpt_flags);
362 }
363
364 /* QP event handler */
365 static void qp_event_handler(struct ib_event *event, void *context)
366 {
367 struct svc_xprt *xprt = context;
368
369 switch (event->event) {
370 /* These are considered benign events */
371 case IB_EVENT_PATH_MIG:
372 case IB_EVENT_COMM_EST:
373 case IB_EVENT_SQ_DRAINED:
374 case IB_EVENT_QP_LAST_WQE_REACHED:
375 dprintk("svcrdma: QP event %s (%d) received for QP=%p\n",
376 ib_event_msg(event->event), event->event,
377 event->element.qp);
378 break;
379 /* These are considered fatal events */
380 case IB_EVENT_PATH_MIG_ERR:
381 case IB_EVENT_QP_FATAL:
382 case IB_EVENT_QP_REQ_ERR:
383 case IB_EVENT_QP_ACCESS_ERR:
384 case IB_EVENT_DEVICE_FATAL:
385 default:
386 dprintk("svcrdma: QP ERROR event %s (%d) received for QP=%p, "
387 "closing transport\n",
388 ib_event_msg(event->event), event->event,
389 event->element.qp);
390 set_bit(XPT_CLOSE, &xprt->xpt_flags);
391 break;
392 }
393 }
394
395 /*
396 * Data Transfer Operation Tasklet
397 *
398 * Walks a list of transports with I/O pending, removing entries as
399 * they are added to the server's I/O pending list. Two bits indicate
400 * if SQ, RQ, or both have I/O pending. The dto_lock is an irqsave
401 * spinlock that serializes access to the transport list with the RQ
402 * and SQ interrupt handlers.
403 */
404 static void dto_tasklet_func(unsigned long data)
405 {
406 struct svcxprt_rdma *xprt;
407 unsigned long flags;
408
409 spin_lock_irqsave(&dto_lock, flags);
410 while (!list_empty(&dto_xprt_q)) {
411 xprt = list_entry(dto_xprt_q.next,
412 struct svcxprt_rdma, sc_dto_q);
413 list_del_init(&xprt->sc_dto_q);
414 spin_unlock_irqrestore(&dto_lock, flags);
415
416 rq_cq_reap(xprt);
417 sq_cq_reap(xprt);
418
419 svc_xprt_put(&xprt->sc_xprt);
420 spin_lock_irqsave(&dto_lock, flags);
421 }
422 spin_unlock_irqrestore(&dto_lock, flags);
423 }
424
425 /*
426 * Receive Queue Completion Handler
427 *
428 * Since an RQ completion handler is called on interrupt context, we
429 * need to defer the handling of the I/O to a tasklet
430 */
431 static void rq_comp_handler(struct ib_cq *cq, void *cq_context)
432 {
433 struct svcxprt_rdma *xprt = cq_context;
434 unsigned long flags;
435
436 /* Guard against unconditional flush call for destroyed QP */
437 if (atomic_read(&xprt->sc_xprt.xpt_ref.refcount)==0)
438 return;
439
440 /*
441 * Set the bit regardless of whether or not it's on the list
442 * because it may be on the list already due to an SQ
443 * completion.
444 */
445 set_bit(RDMAXPRT_RQ_PENDING, &xprt->sc_flags);
446
447 /*
448 * If this transport is not already on the DTO transport queue,
449 * add it
450 */
451 spin_lock_irqsave(&dto_lock, flags);
452 if (list_empty(&xprt->sc_dto_q)) {
453 svc_xprt_get(&xprt->sc_xprt);
454 list_add_tail(&xprt->sc_dto_q, &dto_xprt_q);
455 }
456 spin_unlock_irqrestore(&dto_lock, flags);
457
458 /* Tasklet does all the work to avoid irqsave locks. */
459 tasklet_schedule(&dto_tasklet);
460 }
461
462 /*
463 * rq_cq_reap - Process the RQ CQ.
464 *
465 * Take all completing WC off the CQE and enqueue the associated DTO
466 * context on the dto_q for the transport.
467 *
468 * Note that caller must hold a transport reference.
469 */
470 static void rq_cq_reap(struct svcxprt_rdma *xprt)
471 {
472 int ret;
473 struct ib_wc wc;
474 struct svc_rdma_op_ctxt *ctxt = NULL;
475
476 if (!test_and_clear_bit(RDMAXPRT_RQ_PENDING, &xprt->sc_flags))
477 return;
478
479 ib_req_notify_cq(xprt->sc_rq_cq, IB_CQ_NEXT_COMP);
480 atomic_inc(&rdma_stat_rq_poll);
481
482 while ((ret = ib_poll_cq(xprt->sc_rq_cq, 1, &wc)) > 0) {
483 ctxt = (struct svc_rdma_op_ctxt *)(unsigned long)wc.wr_id;
484 ctxt->wc_status = wc.status;
485 ctxt->byte_len = wc.byte_len;
486 svc_rdma_unmap_dma(ctxt);
487 if (wc.status != IB_WC_SUCCESS) {
488 /* Close the transport */
489 dprintk("svcrdma: transport closing putting ctxt %p\n", ctxt);
490 set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
491 svc_rdma_put_context(ctxt, 1);
492 svc_xprt_put(&xprt->sc_xprt);
493 continue;
494 }
495 spin_lock_bh(&xprt->sc_rq_dto_lock);
496 list_add_tail(&ctxt->dto_q, &xprt->sc_rq_dto_q);
497 spin_unlock_bh(&xprt->sc_rq_dto_lock);
498 svc_xprt_put(&xprt->sc_xprt);
499 }
500
501 if (ctxt)
502 atomic_inc(&rdma_stat_rq_prod);
503
504 set_bit(XPT_DATA, &xprt->sc_xprt.xpt_flags);
505 /*
506 * If data arrived before established event,
507 * don't enqueue. This defers RPC I/O until the
508 * RDMA connection is complete.
509 */
510 if (!test_bit(RDMAXPRT_CONN_PENDING, &xprt->sc_flags))
511 svc_xprt_enqueue(&xprt->sc_xprt);
512 }
513
514 /*
515 * Process a completion context
516 */
517 static void process_context(struct svcxprt_rdma *xprt,
518 struct svc_rdma_op_ctxt *ctxt)
519 {
520 struct svc_rdma_op_ctxt *read_hdr;
521 int free_pages = 0;
522
523 svc_rdma_unmap_dma(ctxt);
524
525 switch (ctxt->wr_op) {
526 case IB_WR_SEND:
527 free_pages = 1;
528 break;
529
530 case IB_WR_RDMA_WRITE:
531 break;
532
533 case IB_WR_RDMA_READ:
534 case IB_WR_RDMA_READ_WITH_INV:
535 svc_rdma_put_frmr(xprt, ctxt->frmr);
536
537 if (!test_bit(RDMACTXT_F_LAST_CTXT, &ctxt->flags))
538 break;
539
540 read_hdr = ctxt->read_hdr;
541 svc_rdma_put_context(ctxt, 0);
542
543 spin_lock_bh(&xprt->sc_rq_dto_lock);
544 set_bit(XPT_DATA, &xprt->sc_xprt.xpt_flags);
545 list_add_tail(&read_hdr->dto_q,
546 &xprt->sc_read_complete_q);
547 spin_unlock_bh(&xprt->sc_rq_dto_lock);
548 svc_xprt_enqueue(&xprt->sc_xprt);
549 return;
550
551 default:
552 dprintk("svcrdma: unexpected completion opcode=%d\n",
553 ctxt->wr_op);
554 break;
555 }
556
557 svc_rdma_put_context(ctxt, free_pages);
558 }
559
560 /*
561 * Send Queue Completion Handler - potentially called on interrupt context.
562 *
563 * Note that caller must hold a transport reference.
564 */
565 static void sq_cq_reap(struct svcxprt_rdma *xprt)
566 {
567 struct svc_rdma_op_ctxt *ctxt = NULL;
568 struct ib_wc wc_a[6];
569 struct ib_wc *wc;
570 struct ib_cq *cq = xprt->sc_sq_cq;
571 int ret;
572
573 memset(wc_a, 0, sizeof(wc_a));
574
575 if (!test_and_clear_bit(RDMAXPRT_SQ_PENDING, &xprt->sc_flags))
576 return;
577
578 ib_req_notify_cq(xprt->sc_sq_cq, IB_CQ_NEXT_COMP);
579 atomic_inc(&rdma_stat_sq_poll);
580 while ((ret = ib_poll_cq(cq, ARRAY_SIZE(wc_a), wc_a)) > 0) {
581 int i;
582
583 for (i = 0; i < ret; i++) {
584 wc = &wc_a[i];
585 if (wc->status != IB_WC_SUCCESS) {
586 dprintk("svcrdma: sq wc err status %s (%d)\n",
587 ib_wc_status_msg(wc->status),
588 wc->status);
589
590 /* Close the transport */
591 set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
592 }
593
594 /* Decrement used SQ WR count */
595 atomic_dec(&xprt->sc_sq_count);
596 wake_up(&xprt->sc_send_wait);
597
598 ctxt = (struct svc_rdma_op_ctxt *)
599 (unsigned long)wc->wr_id;
600 if (ctxt)
601 process_context(xprt, ctxt);
602
603 svc_xprt_put(&xprt->sc_xprt);
604 }
605 }
606
607 if (ctxt)
608 atomic_inc(&rdma_stat_sq_prod);
609 }
610
611 static void sq_comp_handler(struct ib_cq *cq, void *cq_context)
612 {
613 struct svcxprt_rdma *xprt = cq_context;
614 unsigned long flags;
615
616 /* Guard against unconditional flush call for destroyed QP */
617 if (atomic_read(&xprt->sc_xprt.xpt_ref.refcount)==0)
618 return;
619
620 /*
621 * Set the bit regardless of whether or not it's on the list
622 * because it may be on the list already due to an RQ
623 * completion.
624 */
625 set_bit(RDMAXPRT_SQ_PENDING, &xprt->sc_flags);
626
627 /*
628 * If this transport is not already on the DTO transport queue,
629 * add it
630 */
631 spin_lock_irqsave(&dto_lock, flags);
632 if (list_empty(&xprt->sc_dto_q)) {
633 svc_xprt_get(&xprt->sc_xprt);
634 list_add_tail(&xprt->sc_dto_q, &dto_xprt_q);
635 }
636 spin_unlock_irqrestore(&dto_lock, flags);
637
638 /* Tasklet does all the work to avoid irqsave locks. */
639 tasklet_schedule(&dto_tasklet);
640 }
641
642 static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
643 int listener)
644 {
645 struct svcxprt_rdma *cma_xprt = kzalloc(sizeof *cma_xprt, GFP_KERNEL);
646
647 if (!cma_xprt)
648 return NULL;
649 svc_xprt_init(&init_net, &svc_rdma_class, &cma_xprt->sc_xprt, serv);
650 INIT_LIST_HEAD(&cma_xprt->sc_accept_q);
651 INIT_LIST_HEAD(&cma_xprt->sc_dto_q);
652 INIT_LIST_HEAD(&cma_xprt->sc_rq_dto_q);
653 INIT_LIST_HEAD(&cma_xprt->sc_read_complete_q);
654 INIT_LIST_HEAD(&cma_xprt->sc_frmr_q);
655 INIT_LIST_HEAD(&cma_xprt->sc_ctxts);
656 INIT_LIST_HEAD(&cma_xprt->sc_maps);
657 init_waitqueue_head(&cma_xprt->sc_send_wait);
658
659 spin_lock_init(&cma_xprt->sc_lock);
660 spin_lock_init(&cma_xprt->sc_rq_dto_lock);
661 spin_lock_init(&cma_xprt->sc_frmr_q_lock);
662 spin_lock_init(&cma_xprt->sc_ctxt_lock);
663 spin_lock_init(&cma_xprt->sc_map_lock);
664
665 if (listener)
666 set_bit(XPT_LISTENER, &cma_xprt->sc_xprt.xpt_flags);
667
668 return cma_xprt;
669 }
670
671 int svc_rdma_post_recv(struct svcxprt_rdma *xprt, gfp_t flags)
672 {
673 struct ib_recv_wr recv_wr, *bad_recv_wr;
674 struct svc_rdma_op_ctxt *ctxt;
675 struct page *page;
676 dma_addr_t pa;
677 int sge_no;
678 int buflen;
679 int ret;
680
681 ctxt = svc_rdma_get_context(xprt);
682 buflen = 0;
683 ctxt->direction = DMA_FROM_DEVICE;
684 for (sge_no = 0; buflen < xprt->sc_max_req_size; sge_no++) {
685 if (sge_no >= xprt->sc_max_sge) {
686 pr_err("svcrdma: Too many sges (%d)\n", sge_no);
687 goto err_put_ctxt;
688 }
689 page = alloc_page(flags);
690 if (!page)
691 goto err_put_ctxt;
692 ctxt->pages[sge_no] = page;
693 pa = ib_dma_map_page(xprt->sc_cm_id->device,
694 page, 0, PAGE_SIZE,
695 DMA_FROM_DEVICE);
696 if (ib_dma_mapping_error(xprt->sc_cm_id->device, pa))
697 goto err_put_ctxt;
698 atomic_inc(&xprt->sc_dma_used);
699 ctxt->sge[sge_no].addr = pa;
700 ctxt->sge[sge_no].length = PAGE_SIZE;
701 ctxt->sge[sge_no].lkey = xprt->sc_pd->local_dma_lkey;
702 ctxt->count = sge_no + 1;
703 buflen += PAGE_SIZE;
704 }
705 recv_wr.next = NULL;
706 recv_wr.sg_list = &ctxt->sge[0];
707 recv_wr.num_sge = ctxt->count;
708 recv_wr.wr_id = (u64)(unsigned long)ctxt;
709
710 svc_xprt_get(&xprt->sc_xprt);
711 ret = ib_post_recv(xprt->sc_qp, &recv_wr, &bad_recv_wr);
712 if (ret) {
713 svc_rdma_unmap_dma(ctxt);
714 svc_rdma_put_context(ctxt, 1);
715 svc_xprt_put(&xprt->sc_xprt);
716 }
717 return ret;
718
719 err_put_ctxt:
720 svc_rdma_unmap_dma(ctxt);
721 svc_rdma_put_context(ctxt, 1);
722 return -ENOMEM;
723 }
724
725 int svc_rdma_repost_recv(struct svcxprt_rdma *xprt, gfp_t flags)
726 {
727 int ret = 0;
728
729 ret = svc_rdma_post_recv(xprt, flags);
730 if (ret) {
731 pr_err("svcrdma: could not post a receive buffer, err=%d.\n",
732 ret);
733 pr_err("svcrdma: closing transport %p.\n", xprt);
734 set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
735 ret = -ENOTCONN;
736 }
737 return ret;
738 }
739
740 /*
741 * This function handles the CONNECT_REQUEST event on a listening
742 * endpoint. It is passed the cma_id for the _new_ connection. The context in
743 * this cma_id is inherited from the listening cma_id and is the svc_xprt
744 * structure for the listening endpoint.
745 *
746 * This function creates a new xprt for the new connection and enqueues it on
747 * the accept queue for the listent xprt. When the listen thread is kicked, it
748 * will call the recvfrom method on the listen xprt which will accept the new
749 * connection.
750 */
751 static void handle_connect_req(struct rdma_cm_id *new_cma_id, size_t client_ird)
752 {
753 struct svcxprt_rdma *listen_xprt = new_cma_id->context;
754 struct svcxprt_rdma *newxprt;
755 struct sockaddr *sa;
756
757 /* Create a new transport */
758 newxprt = rdma_create_xprt(listen_xprt->sc_xprt.xpt_server, 0);
759 if (!newxprt) {
760 dprintk("svcrdma: failed to create new transport\n");
761 return;
762 }
763 newxprt->sc_cm_id = new_cma_id;
764 new_cma_id->context = newxprt;
765 dprintk("svcrdma: Creating newxprt=%p, cm_id=%p, listenxprt=%p\n",
766 newxprt, newxprt->sc_cm_id, listen_xprt);
767
768 /* Save client advertised inbound read limit for use later in accept. */
769 newxprt->sc_ord = client_ird;
770
771 /* Set the local and remote addresses in the transport */
772 sa = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.dst_addr;
773 svc_xprt_set_remote(&newxprt->sc_xprt, sa, svc_addr_len(sa));
774 sa = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.src_addr;
775 svc_xprt_set_local(&newxprt->sc_xprt, sa, svc_addr_len(sa));
776
777 /*
778 * Enqueue the new transport on the accept queue of the listening
779 * transport
780 */
781 spin_lock_bh(&listen_xprt->sc_lock);
782 list_add_tail(&newxprt->sc_accept_q, &listen_xprt->sc_accept_q);
783 spin_unlock_bh(&listen_xprt->sc_lock);
784
785 set_bit(XPT_CONN, &listen_xprt->sc_xprt.xpt_flags);
786 svc_xprt_enqueue(&listen_xprt->sc_xprt);
787 }
788
789 /*
790 * Handles events generated on the listening endpoint. These events will be
791 * either be incoming connect requests or adapter removal events.
792 */
793 static int rdma_listen_handler(struct rdma_cm_id *cma_id,
794 struct rdma_cm_event *event)
795 {
796 struct svcxprt_rdma *xprt = cma_id->context;
797 int ret = 0;
798
799 switch (event->event) {
800 case RDMA_CM_EVENT_CONNECT_REQUEST:
801 dprintk("svcrdma: Connect request on cma_id=%p, xprt = %p, "
802 "event = %s (%d)\n", cma_id, cma_id->context,
803 rdma_event_msg(event->event), event->event);
804 handle_connect_req(cma_id,
805 event->param.conn.initiator_depth);
806 break;
807
808 case RDMA_CM_EVENT_ESTABLISHED:
809 /* Accept complete */
810 dprintk("svcrdma: Connection completed on LISTEN xprt=%p, "
811 "cm_id=%p\n", xprt, cma_id);
812 break;
813
814 case RDMA_CM_EVENT_DEVICE_REMOVAL:
815 dprintk("svcrdma: Device removal xprt=%p, cm_id=%p\n",
816 xprt, cma_id);
817 if (xprt)
818 set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
819 break;
820
821 default:
822 dprintk("svcrdma: Unexpected event on listening endpoint %p, "
823 "event = %s (%d)\n", cma_id,
824 rdma_event_msg(event->event), event->event);
825 break;
826 }
827
828 return ret;
829 }
830
831 static int rdma_cma_handler(struct rdma_cm_id *cma_id,
832 struct rdma_cm_event *event)
833 {
834 struct svc_xprt *xprt = cma_id->context;
835 struct svcxprt_rdma *rdma =
836 container_of(xprt, struct svcxprt_rdma, sc_xprt);
837 switch (event->event) {
838 case RDMA_CM_EVENT_ESTABLISHED:
839 /* Accept complete */
840 svc_xprt_get(xprt);
841 dprintk("svcrdma: Connection completed on DTO xprt=%p, "
842 "cm_id=%p\n", xprt, cma_id);
843 clear_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags);
844 svc_xprt_enqueue(xprt);
845 break;
846 case RDMA_CM_EVENT_DISCONNECTED:
847 dprintk("svcrdma: Disconnect on DTO xprt=%p, cm_id=%p\n",
848 xprt, cma_id);
849 if (xprt) {
850 set_bit(XPT_CLOSE, &xprt->xpt_flags);
851 svc_xprt_enqueue(xprt);
852 svc_xprt_put(xprt);
853 }
854 break;
855 case RDMA_CM_EVENT_DEVICE_REMOVAL:
856 dprintk("svcrdma: Device removal cma_id=%p, xprt = %p, "
857 "event = %s (%d)\n", cma_id, xprt,
858 rdma_event_msg(event->event), event->event);
859 if (xprt) {
860 set_bit(XPT_CLOSE, &xprt->xpt_flags);
861 svc_xprt_enqueue(xprt);
862 svc_xprt_put(xprt);
863 }
864 break;
865 default:
866 dprintk("svcrdma: Unexpected event on DTO endpoint %p, "
867 "event = %s (%d)\n", cma_id,
868 rdma_event_msg(event->event), event->event);
869 break;
870 }
871 return 0;
872 }
873
874 /*
875 * Create a listening RDMA service endpoint.
876 */
877 static struct svc_xprt *svc_rdma_create(struct svc_serv *serv,
878 struct net *net,
879 struct sockaddr *sa, int salen,
880 int flags)
881 {
882 struct rdma_cm_id *listen_id;
883 struct svcxprt_rdma *cma_xprt;
884 int ret;
885
886 dprintk("svcrdma: Creating RDMA socket\n");
887 if (sa->sa_family != AF_INET) {
888 dprintk("svcrdma: Address family %d is not supported.\n", sa->sa_family);
889 return ERR_PTR(-EAFNOSUPPORT);
890 }
891 cma_xprt = rdma_create_xprt(serv, 1);
892 if (!cma_xprt)
893 return ERR_PTR(-ENOMEM);
894
895 listen_id = rdma_create_id(&init_net, rdma_listen_handler, cma_xprt,
896 RDMA_PS_TCP, IB_QPT_RC);
897 if (IS_ERR(listen_id)) {
898 ret = PTR_ERR(listen_id);
899 dprintk("svcrdma: rdma_create_id failed = %d\n", ret);
900 goto err0;
901 }
902
903 ret = rdma_bind_addr(listen_id, sa);
904 if (ret) {
905 dprintk("svcrdma: rdma_bind_addr failed = %d\n", ret);
906 goto err1;
907 }
908 cma_xprt->sc_cm_id = listen_id;
909
910 ret = rdma_listen(listen_id, RPCRDMA_LISTEN_BACKLOG);
911 if (ret) {
912 dprintk("svcrdma: rdma_listen failed = %d\n", ret);
913 goto err1;
914 }
915
916 /*
917 * We need to use the address from the cm_id in case the
918 * caller specified 0 for the port number.
919 */
920 sa = (struct sockaddr *)&cma_xprt->sc_cm_id->route.addr.src_addr;
921 svc_xprt_set_local(&cma_xprt->sc_xprt, sa, salen);
922
923 return &cma_xprt->sc_xprt;
924
925 err1:
926 rdma_destroy_id(listen_id);
927 err0:
928 kfree(cma_xprt);
929 return ERR_PTR(ret);
930 }
931
932 static struct svc_rdma_fastreg_mr *rdma_alloc_frmr(struct svcxprt_rdma *xprt)
933 {
934 struct ib_mr *mr;
935 struct scatterlist *sg;
936 struct svc_rdma_fastreg_mr *frmr;
937 u32 num_sg;
938
939 frmr = kmalloc(sizeof(*frmr), GFP_KERNEL);
940 if (!frmr)
941 goto err;
942
943 num_sg = min_t(u32, RPCSVC_MAXPAGES, xprt->sc_frmr_pg_list_len);
944 mr = ib_alloc_mr(xprt->sc_pd, IB_MR_TYPE_MEM_REG, num_sg);
945 if (IS_ERR(mr))
946 goto err_free_frmr;
947
948 sg = kcalloc(RPCSVC_MAXPAGES, sizeof(*sg), GFP_KERNEL);
949 if (!sg)
950 goto err_free_mr;
951
952 sg_init_table(sg, RPCSVC_MAXPAGES);
953
954 frmr->mr = mr;
955 frmr->sg = sg;
956 INIT_LIST_HEAD(&frmr->frmr_list);
957 return frmr;
958
959 err_free_mr:
960 ib_dereg_mr(mr);
961 err_free_frmr:
962 kfree(frmr);
963 err:
964 return ERR_PTR(-ENOMEM);
965 }
966
967 static void rdma_dealloc_frmr_q(struct svcxprt_rdma *xprt)
968 {
969 struct svc_rdma_fastreg_mr *frmr;
970
971 while (!list_empty(&xprt->sc_frmr_q)) {
972 frmr = list_entry(xprt->sc_frmr_q.next,
973 struct svc_rdma_fastreg_mr, frmr_list);
974 list_del_init(&frmr->frmr_list);
975 kfree(frmr->sg);
976 ib_dereg_mr(frmr->mr);
977 kfree(frmr);
978 }
979 }
980
981 struct svc_rdma_fastreg_mr *svc_rdma_get_frmr(struct svcxprt_rdma *rdma)
982 {
983 struct svc_rdma_fastreg_mr *frmr = NULL;
984
985 spin_lock_bh(&rdma->sc_frmr_q_lock);
986 if (!list_empty(&rdma->sc_frmr_q)) {
987 frmr = list_entry(rdma->sc_frmr_q.next,
988 struct svc_rdma_fastreg_mr, frmr_list);
989 list_del_init(&frmr->frmr_list);
990 frmr->sg_nents = 0;
991 }
992 spin_unlock_bh(&rdma->sc_frmr_q_lock);
993 if (frmr)
994 return frmr;
995
996 return rdma_alloc_frmr(rdma);
997 }
998
999 void svc_rdma_put_frmr(struct svcxprt_rdma *rdma,
1000 struct svc_rdma_fastreg_mr *frmr)
1001 {
1002 if (frmr) {
1003 ib_dma_unmap_sg(rdma->sc_cm_id->device,
1004 frmr->sg, frmr->sg_nents, frmr->direction);
1005 atomic_dec(&rdma->sc_dma_used);
1006 spin_lock_bh(&rdma->sc_frmr_q_lock);
1007 WARN_ON_ONCE(!list_empty(&frmr->frmr_list));
1008 list_add(&frmr->frmr_list, &rdma->sc_frmr_q);
1009 spin_unlock_bh(&rdma->sc_frmr_q_lock);
1010 }
1011 }
1012
1013 /*
1014 * This is the xpo_recvfrom function for listening endpoints. Its
1015 * purpose is to accept incoming connections. The CMA callback handler
1016 * has already created a new transport and attached it to the new CMA
1017 * ID.
1018 *
1019 * There is a queue of pending connections hung on the listening
1020 * transport. This queue contains the new svc_xprt structure. This
1021 * function takes svc_xprt structures off the accept_q and completes
1022 * the connection.
1023 */
1024 static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
1025 {
1026 struct svcxprt_rdma *listen_rdma;
1027 struct svcxprt_rdma *newxprt = NULL;
1028 struct rdma_conn_param conn_param;
1029 struct ib_cq_init_attr cq_attr = {};
1030 struct ib_qp_init_attr qp_attr;
1031 struct ib_device *dev;
1032 unsigned int i;
1033 int ret = 0;
1034
1035 listen_rdma = container_of(xprt, struct svcxprt_rdma, sc_xprt);
1036 clear_bit(XPT_CONN, &xprt->xpt_flags);
1037 /* Get the next entry off the accept list */
1038 spin_lock_bh(&listen_rdma->sc_lock);
1039 if (!list_empty(&listen_rdma->sc_accept_q)) {
1040 newxprt = list_entry(listen_rdma->sc_accept_q.next,
1041 struct svcxprt_rdma, sc_accept_q);
1042 list_del_init(&newxprt->sc_accept_q);
1043 }
1044 if (!list_empty(&listen_rdma->sc_accept_q))
1045 set_bit(XPT_CONN, &listen_rdma->sc_xprt.xpt_flags);
1046 spin_unlock_bh(&listen_rdma->sc_lock);
1047 if (!newxprt)
1048 return NULL;
1049
1050 dprintk("svcrdma: newxprt from accept queue = %p, cm_id=%p\n",
1051 newxprt, newxprt->sc_cm_id);
1052
1053 dev = newxprt->sc_cm_id->device;
1054
1055 /* Qualify the transport resource defaults with the
1056 * capabilities of this particular device */
1057 newxprt->sc_max_sge = min((size_t)dev->attrs.max_sge,
1058 (size_t)RPCSVC_MAXPAGES);
1059 newxprt->sc_max_sge_rd = min_t(size_t, dev->attrs.max_sge_rd,
1060 RPCSVC_MAXPAGES);
1061 newxprt->sc_max_req_size = svcrdma_max_req_size;
1062 newxprt->sc_max_requests = min_t(u32, dev->attrs.max_qp_wr,
1063 svcrdma_max_requests);
1064 newxprt->sc_max_bc_requests = min_t(u32, dev->attrs.max_qp_wr,
1065 svcrdma_max_bc_requests);
1066 newxprt->sc_rq_depth = newxprt->sc_max_requests +
1067 newxprt->sc_max_bc_requests;
1068 newxprt->sc_sq_depth = RPCRDMA_SQ_DEPTH_MULT * newxprt->sc_rq_depth;
1069
1070 if (!svc_rdma_prealloc_ctxts(newxprt))
1071 goto errout;
1072 if (!svc_rdma_prealloc_maps(newxprt))
1073 goto errout;
1074
1075 /*
1076 * Limit ORD based on client limit, local device limit, and
1077 * configured svcrdma limit.
1078 */
1079 newxprt->sc_ord = min_t(size_t, dev->attrs.max_qp_rd_atom, newxprt->sc_ord);
1080 newxprt->sc_ord = min_t(size_t, svcrdma_ord, newxprt->sc_ord);
1081
1082 newxprt->sc_pd = ib_alloc_pd(dev);
1083 if (IS_ERR(newxprt->sc_pd)) {
1084 dprintk("svcrdma: error creating PD for connect request\n");
1085 goto errout;
1086 }
1087 cq_attr.cqe = newxprt->sc_sq_depth;
1088 newxprt->sc_sq_cq = ib_create_cq(dev,
1089 sq_comp_handler,
1090 cq_event_handler,
1091 newxprt,
1092 &cq_attr);
1093 if (IS_ERR(newxprt->sc_sq_cq)) {
1094 dprintk("svcrdma: error creating SQ CQ for connect request\n");
1095 goto errout;
1096 }
1097 cq_attr.cqe = newxprt->sc_rq_depth;
1098 newxprt->sc_rq_cq = ib_create_cq(dev,
1099 rq_comp_handler,
1100 cq_event_handler,
1101 newxprt,
1102 &cq_attr);
1103 if (IS_ERR(newxprt->sc_rq_cq)) {
1104 dprintk("svcrdma: error creating RQ CQ for connect request\n");
1105 goto errout;
1106 }
1107
1108 memset(&qp_attr, 0, sizeof qp_attr);
1109 qp_attr.event_handler = qp_event_handler;
1110 qp_attr.qp_context = &newxprt->sc_xprt;
1111 qp_attr.cap.max_send_wr = newxprt->sc_sq_depth;
1112 qp_attr.cap.max_recv_wr = newxprt->sc_rq_depth;
1113 qp_attr.cap.max_send_sge = newxprt->sc_max_sge;
1114 qp_attr.cap.max_recv_sge = newxprt->sc_max_sge;
1115 qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
1116 qp_attr.qp_type = IB_QPT_RC;
1117 qp_attr.send_cq = newxprt->sc_sq_cq;
1118 qp_attr.recv_cq = newxprt->sc_rq_cq;
1119 dprintk("svcrdma: newxprt->sc_cm_id=%p, newxprt->sc_pd=%p\n"
1120 " cm_id->device=%p, sc_pd->device=%p\n"
1121 " cap.max_send_wr = %d\n"
1122 " cap.max_recv_wr = %d\n"
1123 " cap.max_send_sge = %d\n"
1124 " cap.max_recv_sge = %d\n",
1125 newxprt->sc_cm_id, newxprt->sc_pd,
1126 dev, newxprt->sc_pd->device,
1127 qp_attr.cap.max_send_wr,
1128 qp_attr.cap.max_recv_wr,
1129 qp_attr.cap.max_send_sge,
1130 qp_attr.cap.max_recv_sge);
1131
1132 ret = rdma_create_qp(newxprt->sc_cm_id, newxprt->sc_pd, &qp_attr);
1133 if (ret) {
1134 dprintk("svcrdma: failed to create QP, ret=%d\n", ret);
1135 goto errout;
1136 }
1137 newxprt->sc_qp = newxprt->sc_cm_id->qp;
1138
1139 /*
1140 * Use the most secure set of MR resources based on the
1141 * transport type and available memory management features in
1142 * the device. Here's the table implemented below:
1143 *
1144 * Fast Global DMA Remote WR
1145 * Reg LKEY MR Access
1146 * Sup'd Sup'd Needed Needed
1147 *
1148 * IWARP N N Y Y
1149 * N Y Y Y
1150 * Y N Y N
1151 * Y Y N -
1152 *
1153 * IB N N Y N
1154 * N Y N -
1155 * Y N Y N
1156 * Y Y N -
1157 *
1158 * NB: iWARP requires remote write access for the data sink
1159 * of an RDMA_READ. IB does not.
1160 */
1161 newxprt->sc_reader = rdma_read_chunk_lcl;
1162 if (dev->attrs.device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS) {
1163 newxprt->sc_frmr_pg_list_len =
1164 dev->attrs.max_fast_reg_page_list_len;
1165 newxprt->sc_dev_caps |= SVCRDMA_DEVCAP_FAST_REG;
1166 newxprt->sc_reader = rdma_read_chunk_frmr;
1167 }
1168
1169 /*
1170 * Determine if a DMA MR is required and if so, what privs are required
1171 */
1172 if (!rdma_protocol_iwarp(dev, newxprt->sc_cm_id->port_num) &&
1173 !rdma_ib_or_roce(dev, newxprt->sc_cm_id->port_num))
1174 goto errout;
1175
1176 if (rdma_protocol_iwarp(dev, newxprt->sc_cm_id->port_num))
1177 newxprt->sc_dev_caps |= SVCRDMA_DEVCAP_READ_W_INV;
1178
1179 /* Post receive buffers */
1180 for (i = 0; i < newxprt->sc_rq_depth; i++) {
1181 ret = svc_rdma_post_recv(newxprt, GFP_KERNEL);
1182 if (ret) {
1183 dprintk("svcrdma: failure posting receive buffers\n");
1184 goto errout;
1185 }
1186 }
1187
1188 /* Swap out the handler */
1189 newxprt->sc_cm_id->event_handler = rdma_cma_handler;
1190
1191 /*
1192 * Arm the CQs for the SQ and RQ before accepting so we can't
1193 * miss the first message
1194 */
1195 ib_req_notify_cq(newxprt->sc_sq_cq, IB_CQ_NEXT_COMP);
1196 ib_req_notify_cq(newxprt->sc_rq_cq, IB_CQ_NEXT_COMP);
1197
1198 /* Accept Connection */
1199 set_bit(RDMAXPRT_CONN_PENDING, &newxprt->sc_flags);
1200 memset(&conn_param, 0, sizeof conn_param);
1201 conn_param.responder_resources = 0;
1202 conn_param.initiator_depth = newxprt->sc_ord;
1203 ret = rdma_accept(newxprt->sc_cm_id, &conn_param);
1204 if (ret) {
1205 dprintk("svcrdma: failed to accept new connection, ret=%d\n",
1206 ret);
1207 goto errout;
1208 }
1209
1210 dprintk("svcrdma: new connection %p accepted with the following "
1211 "attributes:\n"
1212 " local_ip : %pI4\n"
1213 " local_port : %d\n"
1214 " remote_ip : %pI4\n"
1215 " remote_port : %d\n"
1216 " max_sge : %d\n"
1217 " max_sge_rd : %d\n"
1218 " sq_depth : %d\n"
1219 " max_requests : %d\n"
1220 " ord : %d\n",
1221 newxprt,
1222 &((struct sockaddr_in *)&newxprt->sc_cm_id->
1223 route.addr.src_addr)->sin_addr.s_addr,
1224 ntohs(((struct sockaddr_in *)&newxprt->sc_cm_id->
1225 route.addr.src_addr)->sin_port),
1226 &((struct sockaddr_in *)&newxprt->sc_cm_id->
1227 route.addr.dst_addr)->sin_addr.s_addr,
1228 ntohs(((struct sockaddr_in *)&newxprt->sc_cm_id->
1229 route.addr.dst_addr)->sin_port),
1230 newxprt->sc_max_sge,
1231 newxprt->sc_max_sge_rd,
1232 newxprt->sc_sq_depth,
1233 newxprt->sc_max_requests,
1234 newxprt->sc_ord);
1235
1236 return &newxprt->sc_xprt;
1237
1238 errout:
1239 dprintk("svcrdma: failure accepting new connection rc=%d.\n", ret);
1240 /* Take a reference in case the DTO handler runs */
1241 svc_xprt_get(&newxprt->sc_xprt);
1242 if (newxprt->sc_qp && !IS_ERR(newxprt->sc_qp))
1243 ib_destroy_qp(newxprt->sc_qp);
1244 rdma_destroy_id(newxprt->sc_cm_id);
1245 /* This call to put will destroy the transport */
1246 svc_xprt_put(&newxprt->sc_xprt);
1247 return NULL;
1248 }
1249
1250 static void svc_rdma_release_rqst(struct svc_rqst *rqstp)
1251 {
1252 }
1253
1254 /*
1255 * When connected, an svc_xprt has at least two references:
1256 *
1257 * - A reference held by the cm_id between the ESTABLISHED and
1258 * DISCONNECTED events. If the remote peer disconnected first, this
1259 * reference could be gone.
1260 *
1261 * - A reference held by the svc_recv code that called this function
1262 * as part of close processing.
1263 *
1264 * At a minimum one references should still be held.
1265 */
1266 static void svc_rdma_detach(struct svc_xprt *xprt)
1267 {
1268 struct svcxprt_rdma *rdma =
1269 container_of(xprt, struct svcxprt_rdma, sc_xprt);
1270 dprintk("svc: svc_rdma_detach(%p)\n", xprt);
1271
1272 /* Disconnect and flush posted WQE */
1273 rdma_disconnect(rdma->sc_cm_id);
1274 }
1275
1276 static void __svc_rdma_free(struct work_struct *work)
1277 {
1278 struct svcxprt_rdma *rdma =
1279 container_of(work, struct svcxprt_rdma, sc_work);
1280 struct svc_xprt *xprt = &rdma->sc_xprt;
1281
1282 dprintk("svcrdma: %s(%p)\n", __func__, rdma);
1283
1284 /* We should only be called from kref_put */
1285 if (atomic_read(&xprt->xpt_ref.refcount) != 0)
1286 pr_err("svcrdma: sc_xprt still in use? (%d)\n",
1287 atomic_read(&xprt->xpt_ref.refcount));
1288
1289 /*
1290 * Destroy queued, but not processed read completions. Note
1291 * that this cleanup has to be done before destroying the
1292 * cm_id because the device ptr is needed to unmap the dma in
1293 * svc_rdma_put_context.
1294 */
1295 while (!list_empty(&rdma->sc_read_complete_q)) {
1296 struct svc_rdma_op_ctxt *ctxt;
1297 ctxt = list_entry(rdma->sc_read_complete_q.next,
1298 struct svc_rdma_op_ctxt,
1299 dto_q);
1300 list_del_init(&ctxt->dto_q);
1301 svc_rdma_put_context(ctxt, 1);
1302 }
1303
1304 /* Destroy queued, but not processed recv completions */
1305 while (!list_empty(&rdma->sc_rq_dto_q)) {
1306 struct svc_rdma_op_ctxt *ctxt;
1307 ctxt = list_entry(rdma->sc_rq_dto_q.next,
1308 struct svc_rdma_op_ctxt,
1309 dto_q);
1310 list_del_init(&ctxt->dto_q);
1311 svc_rdma_put_context(ctxt, 1);
1312 }
1313
1314 /* Warn if we leaked a resource or under-referenced */
1315 if (rdma->sc_ctxt_used != 0)
1316 pr_err("svcrdma: ctxt still in use? (%d)\n",
1317 rdma->sc_ctxt_used);
1318 if (atomic_read(&rdma->sc_dma_used) != 0)
1319 pr_err("svcrdma: dma still in use? (%d)\n",
1320 atomic_read(&rdma->sc_dma_used));
1321
1322 /* Final put of backchannel client transport */
1323 if (xprt->xpt_bc_xprt) {
1324 xprt_put(xprt->xpt_bc_xprt);
1325 xprt->xpt_bc_xprt = NULL;
1326 }
1327
1328 rdma_dealloc_frmr_q(rdma);
1329 svc_rdma_destroy_ctxts(rdma);
1330 svc_rdma_destroy_maps(rdma);
1331
1332 /* Destroy the QP if present (not a listener) */
1333 if (rdma->sc_qp && !IS_ERR(rdma->sc_qp))
1334 ib_destroy_qp(rdma->sc_qp);
1335
1336 if (rdma->sc_sq_cq && !IS_ERR(rdma->sc_sq_cq))
1337 ib_destroy_cq(rdma->sc_sq_cq);
1338
1339 if (rdma->sc_rq_cq && !IS_ERR(rdma->sc_rq_cq))
1340 ib_destroy_cq(rdma->sc_rq_cq);
1341
1342 if (rdma->sc_pd && !IS_ERR(rdma->sc_pd))
1343 ib_dealloc_pd(rdma->sc_pd);
1344
1345 /* Destroy the CM ID */
1346 rdma_destroy_id(rdma->sc_cm_id);
1347
1348 kfree(rdma);
1349 }
1350
1351 static void svc_rdma_free(struct svc_xprt *xprt)
1352 {
1353 struct svcxprt_rdma *rdma =
1354 container_of(xprt, struct svcxprt_rdma, sc_xprt);
1355 INIT_WORK(&rdma->sc_work, __svc_rdma_free);
1356 queue_work(svc_rdma_wq, &rdma->sc_work);
1357 }
1358
1359 static int svc_rdma_has_wspace(struct svc_xprt *xprt)
1360 {
1361 struct svcxprt_rdma *rdma =
1362 container_of(xprt, struct svcxprt_rdma, sc_xprt);
1363
1364 /*
1365 * If there are already waiters on the SQ,
1366 * return false.
1367 */
1368 if (waitqueue_active(&rdma->sc_send_wait))
1369 return 0;
1370
1371 /* Otherwise return true. */
1372 return 1;
1373 }
1374
1375 static int svc_rdma_secure_port(struct svc_rqst *rqstp)
1376 {
1377 return 1;
1378 }
1379
1380 int svc_rdma_send(struct svcxprt_rdma *xprt, struct ib_send_wr *wr)
1381 {
1382 struct ib_send_wr *bad_wr, *n_wr;
1383 int wr_count;
1384 int i;
1385 int ret;
1386
1387 if (test_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags))
1388 return -ENOTCONN;
1389
1390 wr_count = 1;
1391 for (n_wr = wr->next; n_wr; n_wr = n_wr->next)
1392 wr_count++;
1393
1394 /* If the SQ is full, wait until an SQ entry is available */
1395 while (1) {
1396 spin_lock_bh(&xprt->sc_lock);
1397 if (xprt->sc_sq_depth < atomic_read(&xprt->sc_sq_count) + wr_count) {
1398 spin_unlock_bh(&xprt->sc_lock);
1399 atomic_inc(&rdma_stat_sq_starve);
1400
1401 /* See if we can opportunistically reap SQ WR to make room */
1402 sq_cq_reap(xprt);
1403
1404 /* Wait until SQ WR available if SQ still full */
1405 wait_event(xprt->sc_send_wait,
1406 atomic_read(&xprt->sc_sq_count) <
1407 xprt->sc_sq_depth);
1408 if (test_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags))
1409 return -ENOTCONN;
1410 continue;
1411 }
1412 /* Take a transport ref for each WR posted */
1413 for (i = 0; i < wr_count; i++)
1414 svc_xprt_get(&xprt->sc_xprt);
1415
1416 /* Bump used SQ WR count and post */
1417 atomic_add(wr_count, &xprt->sc_sq_count);
1418 ret = ib_post_send(xprt->sc_qp, wr, &bad_wr);
1419 if (ret) {
1420 set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
1421 atomic_sub(wr_count, &xprt->sc_sq_count);
1422 for (i = 0; i < wr_count; i ++)
1423 svc_xprt_put(&xprt->sc_xprt);
1424 dprintk("svcrdma: failed to post SQ WR rc=%d, "
1425 "sc_sq_count=%d, sc_sq_depth=%d\n",
1426 ret, atomic_read(&xprt->sc_sq_count),
1427 xprt->sc_sq_depth);
1428 }
1429 spin_unlock_bh(&xprt->sc_lock);
1430 if (ret)
1431 wake_up(&xprt->sc_send_wait);
1432 break;
1433 }
1434 return ret;
1435 }
1436
1437 void svc_rdma_send_error(struct svcxprt_rdma *xprt, struct rpcrdma_msg *rmsgp,
1438 enum rpcrdma_errcode err)
1439 {
1440 struct ib_send_wr err_wr;
1441 struct page *p;
1442 struct svc_rdma_op_ctxt *ctxt;
1443 __be32 *va;
1444 int length;
1445 int ret;
1446
1447 p = alloc_page(GFP_KERNEL);
1448 if (!p)
1449 return;
1450 va = page_address(p);
1451
1452 /* XDR encode error */
1453 length = svc_rdma_xdr_encode_error(xprt, rmsgp, err, va);
1454
1455 ctxt = svc_rdma_get_context(xprt);
1456 ctxt->direction = DMA_FROM_DEVICE;
1457 ctxt->count = 1;
1458 ctxt->pages[0] = p;
1459
1460 /* Prepare SGE for local address */
1461 ctxt->sge[0].addr = ib_dma_map_page(xprt->sc_cm_id->device,
1462 p, 0, length, DMA_FROM_DEVICE);
1463 if (ib_dma_mapping_error(xprt->sc_cm_id->device, ctxt->sge[0].addr)) {
1464 put_page(p);
1465 svc_rdma_put_context(ctxt, 1);
1466 return;
1467 }
1468 atomic_inc(&xprt->sc_dma_used);
1469 ctxt->sge[0].lkey = xprt->sc_pd->local_dma_lkey;
1470 ctxt->sge[0].length = length;
1471
1472 /* Prepare SEND WR */
1473 memset(&err_wr, 0, sizeof err_wr);
1474 ctxt->wr_op = IB_WR_SEND;
1475 err_wr.wr_id = (unsigned long)ctxt;
1476 err_wr.sg_list = ctxt->sge;
1477 err_wr.num_sge = 1;
1478 err_wr.opcode = IB_WR_SEND;
1479 err_wr.send_flags = IB_SEND_SIGNALED;
1480
1481 /* Post It */
1482 ret = svc_rdma_send(xprt, &err_wr);
1483 if (ret) {
1484 dprintk("svcrdma: Error %d posting send for protocol error\n",
1485 ret);
1486 svc_rdma_unmap_dma(ctxt);
1487 svc_rdma_put_context(ctxt, 1);
1488 }
1489 }
Linux kernel - Deliverables
This page took 0.086888 seconds and 4 git commands to generate.