1 /*
2 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
3 *
4 * This software is available to you under a choice of one of two
5 * licenses. You may choose to be licensed under the terms of the GNU
6 * General Public License (GPL) Version 2, available from the file
7 * COPYING in the main directory of this source tree, or the BSD-type
8 * license below:
9 *
10 * Redistribution and use in source and binary forms, with or without
11 * modification, are permitted provided that the following conditions
12 * are met:
13 *
14 * Redistributions of source code must retain the above copyright
15 * notice, this list of conditions and the following disclaimer.
16 *
17 * Redistributions in binary form must reproduce the above
18 * copyright notice, this list of conditions and the following
19 * disclaimer in the documentation and/or other materials provided
20 * with the distribution.
21 *
22 * Neither the name of the Network Appliance, Inc. nor the names of
23 * its contributors may be used to endorse or promote products
24 * derived from this software without specific prior written
25 * permission.
26 *
27 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
28 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
29 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
30 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
31 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
32 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
33 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
34 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
35 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
36 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
37 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
38 */
39
40 /*
41 * verbs.c
42 *
43 * Encapsulates the major functions managing:
44 * o adapters
45 * o endpoints
46 * o connections
47 * o buffer memory
48 */
49
50 #include <linux/interrupt.h>
51 #include <linux/slab.h>
52 #include <linux/prefetch.h>
53 #include <linux/sunrpc/addr.h>
54 #include <asm/bitops.h>
55 #include <linux/module.h> /* try_module_get()/module_put() */
56
57 #include "xprt_rdma.h"
58
59 /*
60 * Globals/Macros
61 */
62
63 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
64 # define RPCDBG_FACILITY RPCDBG_TRANS
65 #endif
66
67 /*
68 * internal functions
69 */
70
71 static struct workqueue_struct *rpcrdma_receive_wq;
72
73 int
rpcrdma_alloc_wq(void)74 rpcrdma_alloc_wq(void)
75 {
76 struct workqueue_struct *recv_wq;
77
78 recv_wq = alloc_workqueue("xprtrdma_receive",
79 WQ_MEM_RECLAIM | WQ_UNBOUND | WQ_HIGHPRI,
80 0);
81 if (!recv_wq)
82 return -ENOMEM;
83
84 rpcrdma_receive_wq = recv_wq;
85 return 0;
86 }
87
88 void
rpcrdma_destroy_wq(void)89 rpcrdma_destroy_wq(void)
90 {
91 struct workqueue_struct *wq;
92
93 if (rpcrdma_receive_wq) {
94 wq = rpcrdma_receive_wq;
95 rpcrdma_receive_wq = NULL;
96 destroy_workqueue(wq);
97 }
98 }
99
100 static void
rpcrdma_qp_async_error_upcall(struct ib_event * event,void * context)101 rpcrdma_qp_async_error_upcall(struct ib_event *event, void *context)
102 {
103 struct rpcrdma_ep *ep = context;
104
105 pr_err("RPC: %s: %s on device %s ep %p\n",
106 __func__, ib_event_msg(event->event),
107 event->device->name, context);
108 if (ep->rep_connected == 1) {
109 ep->rep_connected = -EIO;
110 rpcrdma_conn_func(ep);
111 wake_up_all(&ep->rep_connect_wait);
112 }
113 }
114
115 static void
rpcrdma_cq_async_error_upcall(struct ib_event * event,void * context)116 rpcrdma_cq_async_error_upcall(struct ib_event *event, void *context)
117 {
118 struct rpcrdma_ep *ep = context;
119
120 pr_err("RPC: %s: %s on device %s ep %p\n",
121 __func__, ib_event_msg(event->event),
122 event->device->name, context);
123 if (ep->rep_connected == 1) {
124 ep->rep_connected = -EIO;
125 rpcrdma_conn_func(ep);
126 wake_up_all(&ep->rep_connect_wait);
127 }
128 }
129
130 static void
rpcrdma_sendcq_process_wc(struct ib_wc * wc)131 rpcrdma_sendcq_process_wc(struct ib_wc *wc)
132 {
133 /* WARNING: Only wr_id and status are reliable at this point */
134 if (wc->wr_id == RPCRDMA_IGNORE_COMPLETION) {
135 if (wc->status != IB_WC_SUCCESS &&
136 wc->status != IB_WC_WR_FLUSH_ERR)
137 pr_err("RPC: %s: SEND: %s\n",
138 __func__, ib_wc_status_msg(wc->status));
139 } else {
140 struct rpcrdma_mw *r;
141
142 r = (struct rpcrdma_mw *)(unsigned long)wc->wr_id;
143 r->mw_sendcompletion(wc);
144 }
145 }
146
147 /* The common case is a single send completion is waiting. By
148 * passing two WC entries to ib_poll_cq, a return code of 1
149 * means there is exactly one WC waiting and no more. We don't
150 * have to invoke ib_poll_cq again to know that the CQ has been
151 * properly drained.
152 */
153 static void
rpcrdma_sendcq_poll(struct ib_cq * cq)154 rpcrdma_sendcq_poll(struct ib_cq *cq)
155 {
156 struct ib_wc *pos, wcs[2];
157 int count, rc;
158
159 do {
160 pos = wcs;
161
162 rc = ib_poll_cq(cq, ARRAY_SIZE(wcs), pos);
163 if (rc < 0)
164 break;
165
166 count = rc;
167 while (count-- > 0)
168 rpcrdma_sendcq_process_wc(pos++);
169 } while (rc == ARRAY_SIZE(wcs));
170 return;
171 }
172
173 /* Handle provider send completion upcalls.
174 */
175 static void
rpcrdma_sendcq_upcall(struct ib_cq * cq,void * cq_context)176 rpcrdma_sendcq_upcall(struct ib_cq *cq, void *cq_context)
177 {
178 do {
179 rpcrdma_sendcq_poll(cq);
180 } while (ib_req_notify_cq(cq, IB_CQ_NEXT_COMP |
181 IB_CQ_REPORT_MISSED_EVENTS) > 0);
182 }
183
184 static void
rpcrdma_receive_worker(struct work_struct * work)185 rpcrdma_receive_worker(struct work_struct *work)
186 {
187 struct rpcrdma_rep *rep =
188 container_of(work, struct rpcrdma_rep, rr_work);
189
190 rpcrdma_reply_handler(rep);
191 }
192
193 static void
rpcrdma_recvcq_process_wc(struct ib_wc * wc)194 rpcrdma_recvcq_process_wc(struct ib_wc *wc)
195 {
196 struct rpcrdma_rep *rep =
197 (struct rpcrdma_rep *)(unsigned long)wc->wr_id;
198
199 /* WARNING: Only wr_id and status are reliable at this point */
200 if (wc->status != IB_WC_SUCCESS)
201 goto out_fail;
202
203 /* status == SUCCESS means all fields in wc are trustworthy */
204 if (wc->opcode != IB_WC_RECV)
205 return;
206
207 dprintk("RPC: %s: rep %p opcode 'recv', length %u: success\n",
208 __func__, rep, wc->byte_len);
209
210 rep->rr_len = wc->byte_len;
211 ib_dma_sync_single_for_cpu(rep->rr_device,
212 rdmab_addr(rep->rr_rdmabuf),
213 rep->rr_len, DMA_FROM_DEVICE);
214 prefetch(rdmab_to_msg(rep->rr_rdmabuf));
215
216 out_schedule:
217 queue_work(rpcrdma_receive_wq, &rep->rr_work);
218 return;
219
220 out_fail:
221 if (wc->status != IB_WC_WR_FLUSH_ERR)
222 pr_err("RPC: %s: rep %p: %s\n",
223 __func__, rep, ib_wc_status_msg(wc->status));
224 rep->rr_len = RPCRDMA_BAD_LEN;
225 goto out_schedule;
226 }
227
228 /* The wc array is on stack: automatic memory is always CPU-local.
229 *
230 * struct ib_wc is 64 bytes, making the poll array potentially
231 * large. But this is at the bottom of the call chain. Further
232 * substantial work is done in another thread.
233 */
234 static void
rpcrdma_recvcq_poll(struct ib_cq * cq)235 rpcrdma_recvcq_poll(struct ib_cq *cq)
236 {
237 struct ib_wc *pos, wcs[4];
238 int count, rc;
239
240 do {
241 pos = wcs;
242
243 rc = ib_poll_cq(cq, ARRAY_SIZE(wcs), pos);
244 if (rc < 0)
245 break;
246
247 count = rc;
248 while (count-- > 0)
249 rpcrdma_recvcq_process_wc(pos++);
250 } while (rc == ARRAY_SIZE(wcs));
251 }
252
253 /* Handle provider receive completion upcalls.
254 */
255 static void
rpcrdma_recvcq_upcall(struct ib_cq * cq,void * cq_context)256 rpcrdma_recvcq_upcall(struct ib_cq *cq, void *cq_context)
257 {
258 do {
259 rpcrdma_recvcq_poll(cq);
260 } while (ib_req_notify_cq(cq, IB_CQ_NEXT_COMP |
261 IB_CQ_REPORT_MISSED_EVENTS) > 0);
262 }
263
264 static void
rpcrdma_flush_cqs(struct rpcrdma_ep * ep)265 rpcrdma_flush_cqs(struct rpcrdma_ep *ep)
266 {
267 struct ib_wc wc;
268
269 while (ib_poll_cq(ep->rep_attr.recv_cq, 1, &wc) > 0)
270 rpcrdma_recvcq_process_wc(&wc);
271 while (ib_poll_cq(ep->rep_attr.send_cq, 1, &wc) > 0)
272 rpcrdma_sendcq_process_wc(&wc);
273 }
274
275 static int
rpcrdma_conn_upcall(struct rdma_cm_id * id,struct rdma_cm_event * event)276 rpcrdma_conn_upcall(struct rdma_cm_id *id, struct rdma_cm_event *event)
277 {
278 struct rpcrdma_xprt *xprt = id->context;
279 struct rpcrdma_ia *ia = &xprt->rx_ia;
280 struct rpcrdma_ep *ep = &xprt->rx_ep;
281 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
282 struct sockaddr *sap = (struct sockaddr *)&ep->rep_remote_addr;
283 #endif
284 struct ib_qp_attr *attr = &ia->ri_qp_attr;
285 struct ib_qp_init_attr *iattr = &ia->ri_qp_init_attr;
286 int connstate = 0;
287
288 switch (event->event) {
289 case RDMA_CM_EVENT_ADDR_RESOLVED:
290 case RDMA_CM_EVENT_ROUTE_RESOLVED:
291 ia->ri_async_rc = 0;
292 complete(&ia->ri_done);
293 break;
294 case RDMA_CM_EVENT_ADDR_ERROR:
295 ia->ri_async_rc = -EHOSTUNREACH;
296 dprintk("RPC: %s: CM address resolution error, ep 0x%p\n",
297 __func__, ep);
298 complete(&ia->ri_done);
299 break;
300 case RDMA_CM_EVENT_ROUTE_ERROR:
301 ia->ri_async_rc = -ENETUNREACH;
302 dprintk("RPC: %s: CM route resolution error, ep 0x%p\n",
303 __func__, ep);
304 complete(&ia->ri_done);
305 break;
306 case RDMA_CM_EVENT_ESTABLISHED:
307 connstate = 1;
308 ib_query_qp(ia->ri_id->qp, attr,
309 IB_QP_MAX_QP_RD_ATOMIC | IB_QP_MAX_DEST_RD_ATOMIC,
310 iattr);
311 dprintk("RPC: %s: %d responder resources"
312 " (%d initiator)\n",
313 __func__, attr->max_dest_rd_atomic,
314 attr->max_rd_atomic);
315 goto connected;
316 case RDMA_CM_EVENT_CONNECT_ERROR:
317 connstate = -ENOTCONN;
318 goto connected;
319 case RDMA_CM_EVENT_UNREACHABLE:
320 connstate = -ENETDOWN;
321 goto connected;
322 case RDMA_CM_EVENT_REJECTED:
323 connstate = -ECONNREFUSED;
324 goto connected;
325 case RDMA_CM_EVENT_DISCONNECTED:
326 connstate = -ECONNABORTED;
327 goto connected;
328 case RDMA_CM_EVENT_DEVICE_REMOVAL:
329 connstate = -ENODEV;
330 connected:
331 dprintk("RPC: %s: %sconnected\n",
332 __func__, connstate > 0 ? "" : "dis");
333 ep->rep_connected = connstate;
334 rpcrdma_conn_func(ep);
335 wake_up_all(&ep->rep_connect_wait);
336 /*FALLTHROUGH*/
337 default:
338 dprintk("RPC: %s: %pIS:%u (ep 0x%p): %s\n",
339 __func__, sap, rpc_get_port(sap), ep,
340 rdma_event_msg(event->event));
341 break;
342 }
343
344 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
345 if (connstate == 1) {
346 int ird = attr->max_dest_rd_atomic;
347 int tird = ep->rep_remote_cma.responder_resources;
348
349 pr_info("rpcrdma: connection to %pIS:%u on %s, memreg '%s', %d credits, %d responders%s\n",
350 sap, rpc_get_port(sap),
351 ia->ri_device->name,
352 ia->ri_ops->ro_displayname,
353 xprt->rx_buf.rb_max_requests,
354 ird, ird < 4 && ird < tird / 2 ? " (low!)" : "");
355 } else if (connstate < 0) {
356 pr_info("rpcrdma: connection to %pIS:%u closed (%d)\n",
357 sap, rpc_get_port(sap), connstate);
358 }
359 #endif
360
361 return 0;
362 }
363
rpcrdma_destroy_id(struct rdma_cm_id * id)364 static void rpcrdma_destroy_id(struct rdma_cm_id *id)
365 {
366 if (id) {
367 module_put(id->device->owner);
368 rdma_destroy_id(id);
369 }
370 }
371
372 static struct rdma_cm_id *
rpcrdma_create_id(struct rpcrdma_xprt * xprt,struct rpcrdma_ia * ia,struct sockaddr * addr)373 rpcrdma_create_id(struct rpcrdma_xprt *xprt,
374 struct rpcrdma_ia *ia, struct sockaddr *addr)
375 {
376 struct rdma_cm_id *id;
377 int rc;
378
379 init_completion(&ia->ri_done);
380
381 id = rdma_create_id(&init_net, rpcrdma_conn_upcall, xprt, RDMA_PS_TCP,
382 IB_QPT_RC);
383 if (IS_ERR(id)) {
384 rc = PTR_ERR(id);
385 dprintk("RPC: %s: rdma_create_id() failed %i\n",
386 __func__, rc);
387 return id;
388 }
389
390 ia->ri_async_rc = -ETIMEDOUT;
391 rc = rdma_resolve_addr(id, NULL, addr, RDMA_RESOLVE_TIMEOUT);
392 if (rc) {
393 dprintk("RPC: %s: rdma_resolve_addr() failed %i\n",
394 __func__, rc);
395 goto out;
396 }
397 wait_for_completion_interruptible_timeout(&ia->ri_done,
398 msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1);
399
400 /* FIXME:
401 * Until xprtrdma supports DEVICE_REMOVAL, the provider must
402 * be pinned while there are active NFS/RDMA mounts to prevent
403 * hangs and crashes at umount time.
404 */
405 if (!ia->ri_async_rc && !try_module_get(id->device->owner)) {
406 dprintk("RPC: %s: Failed to get device module\n",
407 __func__);
408 ia->ri_async_rc = -ENODEV;
409 }
410 rc = ia->ri_async_rc;
411 if (rc)
412 goto out;
413
414 ia->ri_async_rc = -ETIMEDOUT;
415 rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT);
416 if (rc) {
417 dprintk("RPC: %s: rdma_resolve_route() failed %i\n",
418 __func__, rc);
419 goto put;
420 }
421 wait_for_completion_interruptible_timeout(&ia->ri_done,
422 msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1);
423 rc = ia->ri_async_rc;
424 if (rc)
425 goto put;
426
427 return id;
428 put:
429 module_put(id->device->owner);
430 out:
431 rdma_destroy_id(id);
432 return ERR_PTR(rc);
433 }
434
435 /*
436 * Drain any cq, prior to teardown.
437 */
438 static void
rpcrdma_clean_cq(struct ib_cq * cq)439 rpcrdma_clean_cq(struct ib_cq *cq)
440 {
441 struct ib_wc wc;
442 int count = 0;
443
444 while (1 == ib_poll_cq(cq, 1, &wc))
445 ++count;
446
447 if (count)
448 dprintk("RPC: %s: flushed %d events (last 0x%x)\n",
449 __func__, count, wc.opcode);
450 }
451
452 /*
453 * Exported functions.
454 */
455
456 /*
457 * Open and initialize an Interface Adapter.
458 * o initializes fields of struct rpcrdma_ia, including
459 * interface and provider attributes and protection zone.
460 */
461 int
rpcrdma_ia_open(struct rpcrdma_xprt * xprt,struct sockaddr * addr,int memreg)462 rpcrdma_ia_open(struct rpcrdma_xprt *xprt, struct sockaddr *addr, int memreg)
463 {
464 struct rpcrdma_ia *ia = &xprt->rx_ia;
465 struct ib_device_attr *devattr = &ia->ri_devattr;
466 int rc;
467
468 ia->ri_dma_mr = NULL;
469
470 ia->ri_id = rpcrdma_create_id(xprt, ia, addr);
471 if (IS_ERR(ia->ri_id)) {
472 rc = PTR_ERR(ia->ri_id);
473 goto out1;
474 }
475 ia->ri_device = ia->ri_id->device;
476
477 ia->ri_pd = ib_alloc_pd(ia->ri_device);
478 if (IS_ERR(ia->ri_pd)) {
479 rc = PTR_ERR(ia->ri_pd);
480 dprintk("RPC: %s: ib_alloc_pd() failed %i\n",
481 __func__, rc);
482 goto out2;
483 }
484
485 rc = ib_query_device(ia->ri_device, devattr);
486 if (rc) {
487 dprintk("RPC: %s: ib_query_device failed %d\n",
488 __func__, rc);
489 goto out3;
490 }
491
492 if (memreg == RPCRDMA_FRMR) {
493 if (!(devattr->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS) ||
494 (devattr->max_fast_reg_page_list_len == 0)) {
495 dprintk("RPC: %s: FRMR registration "
496 "not supported by HCA\n", __func__);
497 memreg = RPCRDMA_MTHCAFMR;
498 }
499 }
500 if (memreg == RPCRDMA_MTHCAFMR) {
501 if (!ia->ri_device->alloc_fmr) {
502 dprintk("RPC: %s: MTHCAFMR registration "
503 "not supported by HCA\n", __func__);
504 rc = -EINVAL;
505 goto out3;
506 }
507 }
508
509 switch (memreg) {
510 case RPCRDMA_FRMR:
511 ia->ri_ops = &rpcrdma_frwr_memreg_ops;
512 break;
513 case RPCRDMA_ALLPHYSICAL:
514 ia->ri_ops = &rpcrdma_physical_memreg_ops;
515 break;
516 case RPCRDMA_MTHCAFMR:
517 ia->ri_ops = &rpcrdma_fmr_memreg_ops;
518 break;
519 default:
520 printk(KERN_ERR "RPC: Unsupported memory "
521 "registration mode: %d\n", memreg);
522 rc = -ENOMEM;
523 goto out3;
524 }
525 dprintk("RPC: %s: memory registration strategy is '%s'\n",
526 __func__, ia->ri_ops->ro_displayname);
527
528 rwlock_init(&ia->ri_qplock);
529 return 0;
530
531 out3:
532 ib_dealloc_pd(ia->ri_pd);
533 ia->ri_pd = NULL;
534 out2:
535 rpcrdma_destroy_id(ia->ri_id);
536 ia->ri_id = NULL;
537 out1:
538 return rc;
539 }
540
541 /*
542 * Clean up/close an IA.
543 * o if event handles and PD have been initialized, free them.
544 * o close the IA
545 */
546 void
rpcrdma_ia_close(struct rpcrdma_ia * ia)547 rpcrdma_ia_close(struct rpcrdma_ia *ia)
548 {
549 dprintk("RPC: %s: entering\n", __func__);
550 if (ia->ri_id != NULL && !IS_ERR(ia->ri_id)) {
551 if (ia->ri_id->qp)
552 rdma_destroy_qp(ia->ri_id);
553 rpcrdma_destroy_id(ia->ri_id);
554 ia->ri_id = NULL;
555 }
556
557 /* If the pd is still busy, xprtrdma missed freeing a resource */
558 if (ia->ri_pd && !IS_ERR(ia->ri_pd))
559 ib_dealloc_pd(ia->ri_pd);
560 }
561
562 /*
563 * Create unconnected endpoint.
564 */
565 int
rpcrdma_ep_create(struct rpcrdma_ep * ep,struct rpcrdma_ia * ia,struct rpcrdma_create_data_internal * cdata)566 rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia,
567 struct rpcrdma_create_data_internal *cdata)
568 {
569 struct ib_device_attr *devattr = &ia->ri_devattr;
570 struct ib_cq *sendcq, *recvcq;
571 struct ib_cq_init_attr cq_attr = {};
572 unsigned int max_qp_wr;
573 int rc, err;
574
575 if (devattr->max_sge < RPCRDMA_MAX_IOVS) {
576 dprintk("RPC: %s: insufficient sge's available\n",
577 __func__);
578 return -ENOMEM;
579 }
580
581 if (devattr->max_qp_wr <= RPCRDMA_BACKWARD_WRS) {
582 dprintk("RPC: %s: insufficient wqe's available\n",
583 __func__);
584 return -ENOMEM;
585 }
586 max_qp_wr = devattr->max_qp_wr - RPCRDMA_BACKWARD_WRS;
587
588 /* check provider's send/recv wr limits */
589 if (cdata->max_requests > max_qp_wr)
590 cdata->max_requests = max_qp_wr;
591
592 ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall;
593 ep->rep_attr.qp_context = ep;
594 ep->rep_attr.srq = NULL;
595 ep->rep_attr.cap.max_send_wr = cdata->max_requests;
596 ep->rep_attr.cap.max_send_wr += RPCRDMA_BACKWARD_WRS;
597 rc = ia->ri_ops->ro_open(ia, ep, cdata);
598 if (rc)
599 return rc;
600 ep->rep_attr.cap.max_recv_wr = cdata->max_requests;
601 ep->rep_attr.cap.max_recv_wr += RPCRDMA_BACKWARD_WRS;
602 ep->rep_attr.cap.max_send_sge = RPCRDMA_MAX_IOVS;
603 ep->rep_attr.cap.max_recv_sge = 1;
604 ep->rep_attr.cap.max_inline_data = 0;
605 ep->rep_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
606 ep->rep_attr.qp_type = IB_QPT_RC;
607 ep->rep_attr.port_num = ~0;
608
609 dprintk("RPC: %s: requested max: dtos: send %d recv %d; "
610 "iovs: send %d recv %d\n",
611 __func__,
612 ep->rep_attr.cap.max_send_wr,
613 ep->rep_attr.cap.max_recv_wr,
614 ep->rep_attr.cap.max_send_sge,
615 ep->rep_attr.cap.max_recv_sge);
616
617 /* set trigger for requesting send completion */
618 ep->rep_cqinit = ep->rep_attr.cap.max_send_wr/2 - 1;
619 if (ep->rep_cqinit > RPCRDMA_MAX_UNSIGNALED_SENDS)
620 ep->rep_cqinit = RPCRDMA_MAX_UNSIGNALED_SENDS;
621 else if (ep->rep_cqinit <= 2)
622 ep->rep_cqinit = 0;
623 INIT_CQCOUNT(ep);
624 init_waitqueue_head(&ep->rep_connect_wait);
625 INIT_DELAYED_WORK(&ep->rep_connect_worker, rpcrdma_connect_worker);
626
627 cq_attr.cqe = ep->rep_attr.cap.max_send_wr + 1;
628 sendcq = ib_create_cq(ia->ri_device, rpcrdma_sendcq_upcall,
629 rpcrdma_cq_async_error_upcall, NULL, &cq_attr);
630 if (IS_ERR(sendcq)) {
631 rc = PTR_ERR(sendcq);
632 dprintk("RPC: %s: failed to create send CQ: %i\n",
633 __func__, rc);
634 goto out1;
635 }
636
637 rc = ib_req_notify_cq(sendcq, IB_CQ_NEXT_COMP);
638 if (rc) {
639 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n",
640 __func__, rc);
641 goto out2;
642 }
643
644 cq_attr.cqe = ep->rep_attr.cap.max_recv_wr + 1;
645 recvcq = ib_create_cq(ia->ri_device, rpcrdma_recvcq_upcall,
646 rpcrdma_cq_async_error_upcall, NULL, &cq_attr);
647 if (IS_ERR(recvcq)) {
648 rc = PTR_ERR(recvcq);
649 dprintk("RPC: %s: failed to create recv CQ: %i\n",
650 __func__, rc);
651 goto out2;
652 }
653
654 rc = ib_req_notify_cq(recvcq, IB_CQ_NEXT_COMP);
655 if (rc) {
656 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n",
657 __func__, rc);
658 ib_destroy_cq(recvcq);
659 goto out2;
660 }
661
662 ep->rep_attr.send_cq = sendcq;
663 ep->rep_attr.recv_cq = recvcq;
664
665 /* Initialize cma parameters */
666
667 /* RPC/RDMA does not use private data */
668 ep->rep_remote_cma.private_data = NULL;
669 ep->rep_remote_cma.private_data_len = 0;
670
671 /* Client offers RDMA Read but does not initiate */
672 ep->rep_remote_cma.initiator_depth = 0;
673 if (devattr->max_qp_rd_atom > 32) /* arbitrary but <= 255 */
674 ep->rep_remote_cma.responder_resources = 32;
675 else
676 ep->rep_remote_cma.responder_resources =
677 devattr->max_qp_rd_atom;
678
679 ep->rep_remote_cma.retry_count = 7;
680 ep->rep_remote_cma.flow_control = 0;
681 ep->rep_remote_cma.rnr_retry_count = 0;
682
683 return 0;
684
685 out2:
686 err = ib_destroy_cq(sendcq);
687 if (err)
688 dprintk("RPC: %s: ib_destroy_cq returned %i\n",
689 __func__, err);
690 out1:
691 if (ia->ri_dma_mr)
692 ib_dereg_mr(ia->ri_dma_mr);
693 return rc;
694 }
695
696 /*
697 * rpcrdma_ep_destroy
698 *
699 * Disconnect and destroy endpoint. After this, the only
700 * valid operations on the ep are to free it (if dynamically
701 * allocated) or re-create it.
702 */
703 void
rpcrdma_ep_destroy(struct rpcrdma_ep * ep,struct rpcrdma_ia * ia)704 rpcrdma_ep_destroy(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
705 {
706 int rc;
707
708 dprintk("RPC: %s: entering, connected is %d\n",
709 __func__, ep->rep_connected);
710
711 cancel_delayed_work_sync(&ep->rep_connect_worker);
712
713 if (ia->ri_id->qp)
714 rpcrdma_ep_disconnect(ep, ia);
715
716 rpcrdma_clean_cq(ep->rep_attr.recv_cq);
717 rpcrdma_clean_cq(ep->rep_attr.send_cq);
718
719 if (ia->ri_id->qp) {
720 rdma_destroy_qp(ia->ri_id);
721 ia->ri_id->qp = NULL;
722 }
723
724 rc = ib_destroy_cq(ep->rep_attr.recv_cq);
725 if (rc)
726 dprintk("RPC: %s: ib_destroy_cq returned %i\n",
727 __func__, rc);
728
729 rc = ib_destroy_cq(ep->rep_attr.send_cq);
730 if (rc)
731 dprintk("RPC: %s: ib_destroy_cq returned %i\n",
732 __func__, rc);
733
734 if (ia->ri_dma_mr) {
735 rc = ib_dereg_mr(ia->ri_dma_mr);
736 dprintk("RPC: %s: ib_dereg_mr returned %i\n",
737 __func__, rc);
738 }
739 }
740
741 /*
742 * Connect unconnected endpoint.
743 */
744 int
rpcrdma_ep_connect(struct rpcrdma_ep * ep,struct rpcrdma_ia * ia)745 rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
746 {
747 struct rdma_cm_id *id, *old;
748 int rc = 0;
749 int retry_count = 0;
750
751 if (ep->rep_connected != 0) {
752 struct rpcrdma_xprt *xprt;
753 retry:
754 dprintk("RPC: %s: reconnecting...\n", __func__);
755
756 rpcrdma_ep_disconnect(ep, ia);
757 rpcrdma_flush_cqs(ep);
758
759 xprt = container_of(ia, struct rpcrdma_xprt, rx_ia);
760 id = rpcrdma_create_id(xprt, ia,
761 (struct sockaddr *)&xprt->rx_data.addr);
762 if (IS_ERR(id)) {
763 rc = -EHOSTUNREACH;
764 goto out;
765 }
766 /* TEMP TEMP TEMP - fail if new device:
767 * Deregister/remarshal *all* requests!
768 * Close and recreate adapter, pd, etc!
769 * Re-determine all attributes still sane!
770 * More stuff I haven't thought of!
771 * Rrrgh!
772 */
773 if (ia->ri_device != id->device) {
774 printk("RPC: %s: can't reconnect on "
775 "different device!\n", __func__);
776 rpcrdma_destroy_id(id);
777 rc = -ENETUNREACH;
778 goto out;
779 }
780 /* END TEMP */
781 rc = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr);
782 if (rc) {
783 dprintk("RPC: %s: rdma_create_qp failed %i\n",
784 __func__, rc);
785 rpcrdma_destroy_id(id);
786 rc = -ENETUNREACH;
787 goto out;
788 }
789
790 write_lock(&ia->ri_qplock);
791 old = ia->ri_id;
792 ia->ri_id = id;
793 write_unlock(&ia->ri_qplock);
794
795 rdma_destroy_qp(old);
796 rpcrdma_destroy_id(old);
797 } else {
798 dprintk("RPC: %s: connecting...\n", __func__);
799 rc = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr);
800 if (rc) {
801 dprintk("RPC: %s: rdma_create_qp failed %i\n",
802 __func__, rc);
803 /* do not update ep->rep_connected */
804 return -ENETUNREACH;
805 }
806 }
807
808 ep->rep_connected = 0;
809
810 rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma);
811 if (rc) {
812 dprintk("RPC: %s: rdma_connect() failed with %i\n",
813 __func__, rc);
814 goto out;
815 }
816
817 wait_event_interruptible(ep->rep_connect_wait, ep->rep_connected != 0);
818
819 /*
820 * Check state. A non-peer reject indicates no listener
821 * (ECONNREFUSED), which may be a transient state. All
822 * others indicate a transport condition which has already
823 * undergone a best-effort.
824 */
825 if (ep->rep_connected == -ECONNREFUSED &&
826 ++retry_count <= RDMA_CONNECT_RETRY_MAX) {
827 dprintk("RPC: %s: non-peer_reject, retry\n", __func__);
828 goto retry;
829 }
830 if (ep->rep_connected <= 0) {
831 /* Sometimes, the only way to reliably connect to remote
832 * CMs is to use same nonzero values for ORD and IRD. */
833 if (retry_count++ <= RDMA_CONNECT_RETRY_MAX + 1 &&
834 (ep->rep_remote_cma.responder_resources == 0 ||
835 ep->rep_remote_cma.initiator_depth !=
836 ep->rep_remote_cma.responder_resources)) {
837 if (ep->rep_remote_cma.responder_resources == 0)
838 ep->rep_remote_cma.responder_resources = 1;
839 ep->rep_remote_cma.initiator_depth =
840 ep->rep_remote_cma.responder_resources;
841 goto retry;
842 }
843 rc = ep->rep_connected;
844 } else {
845 struct rpcrdma_xprt *r_xprt;
846 unsigned int extras;
847
848 dprintk("RPC: %s: connected\n", __func__);
849
850 r_xprt = container_of(ia, struct rpcrdma_xprt, rx_ia);
851 extras = r_xprt->rx_buf.rb_bc_srv_max_requests;
852
853 if (extras) {
854 rc = rpcrdma_ep_post_extra_recv(r_xprt, extras);
855 if (rc)
856 pr_warn("%s: rpcrdma_ep_post_extra_recv: %i\n",
857 __func__, rc);
858 rc = 0;
859 }
860 }
861
862 out:
863 if (rc)
864 ep->rep_connected = rc;
865 return rc;
866 }
867
868 /*
869 * rpcrdma_ep_disconnect
870 *
871 * This is separate from destroy to facilitate the ability
872 * to reconnect without recreating the endpoint.
873 *
874 * This call is not reentrant, and must not be made in parallel
875 * on the same endpoint.
876 */
877 void
rpcrdma_ep_disconnect(struct rpcrdma_ep * ep,struct rpcrdma_ia * ia)878 rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
879 {
880 int rc;
881
882 rpcrdma_flush_cqs(ep);
883 rc = rdma_disconnect(ia->ri_id);
884 if (!rc) {
885 /* returns without wait if not connected */
886 wait_event_interruptible(ep->rep_connect_wait,
887 ep->rep_connected != 1);
888 dprintk("RPC: %s: after wait, %sconnected\n", __func__,
889 (ep->rep_connected == 1) ? "still " : "dis");
890 } else {
891 dprintk("RPC: %s: rdma_disconnect %i\n", __func__, rc);
892 ep->rep_connected = rc;
893 }
894 }
895
896 struct rpcrdma_req *
rpcrdma_create_req(struct rpcrdma_xprt * r_xprt)897 rpcrdma_create_req(struct rpcrdma_xprt *r_xprt)
898 {
899 struct rpcrdma_buffer *buffer = &r_xprt->rx_buf;
900 struct rpcrdma_req *req;
901
902 req = kzalloc(sizeof(*req), GFP_KERNEL);
903 if (req == NULL)
904 return ERR_PTR(-ENOMEM);
905
906 INIT_LIST_HEAD(&req->rl_free);
907 spin_lock(&buffer->rb_reqslock);
908 list_add(&req->rl_all, &buffer->rb_allreqs);
909 spin_unlock(&buffer->rb_reqslock);
910 req->rl_buffer = &r_xprt->rx_buf;
911 return req;
912 }
913
914 struct rpcrdma_rep *
rpcrdma_create_rep(struct rpcrdma_xprt * r_xprt)915 rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt)
916 {
917 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
918 struct rpcrdma_ia *ia = &r_xprt->rx_ia;
919 struct rpcrdma_rep *rep;
920 int rc;
921
922 rc = -ENOMEM;
923 rep = kzalloc(sizeof(*rep), GFP_KERNEL);
924 if (rep == NULL)
925 goto out;
926
927 rep->rr_rdmabuf = rpcrdma_alloc_regbuf(ia, cdata->inline_rsize,
928 GFP_KERNEL);
929 if (IS_ERR(rep->rr_rdmabuf)) {
930 rc = PTR_ERR(rep->rr_rdmabuf);
931 goto out_free;
932 }
933
934 rep->rr_device = ia->ri_device;
935 rep->rr_rxprt = r_xprt;
936 INIT_WORK(&rep->rr_work, rpcrdma_receive_worker);
937 return rep;
938
939 out_free:
940 kfree(rep);
941 out:
942 return ERR_PTR(rc);
943 }
944
945 int
rpcrdma_buffer_create(struct rpcrdma_xprt * r_xprt)946 rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
947 {
948 struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
949 struct rpcrdma_ia *ia = &r_xprt->rx_ia;
950 int i, rc;
951
952 buf->rb_max_requests = r_xprt->rx_data.max_requests;
953 buf->rb_bc_srv_max_requests = 0;
954 spin_lock_init(&buf->rb_lock);
955
956 rc = ia->ri_ops->ro_init(r_xprt);
957 if (rc)
958 goto out;
959
960 INIT_LIST_HEAD(&buf->rb_send_bufs);
961 INIT_LIST_HEAD(&buf->rb_allreqs);
962 spin_lock_init(&buf->rb_reqslock);
963 for (i = 0; i < buf->rb_max_requests; i++) {
964 struct rpcrdma_req *req;
965
966 req = rpcrdma_create_req(r_xprt);
967 if (IS_ERR(req)) {
968 dprintk("RPC: %s: request buffer %d alloc"
969 " failed\n", __func__, i);
970 rc = PTR_ERR(req);
971 goto out;
972 }
973 req->rl_backchannel = false;
974 list_add(&req->rl_free, &buf->rb_send_bufs);
975 }
976
977 INIT_LIST_HEAD(&buf->rb_recv_bufs);
978 for (i = 0; i < buf->rb_max_requests + 2; i++) {
979 struct rpcrdma_rep *rep;
980
981 rep = rpcrdma_create_rep(r_xprt);
982 if (IS_ERR(rep)) {
983 dprintk("RPC: %s: reply buffer %d alloc failed\n",
984 __func__, i);
985 rc = PTR_ERR(rep);
986 goto out;
987 }
988 list_add(&rep->rr_list, &buf->rb_recv_bufs);
989 }
990
991 return 0;
992 out:
993 rpcrdma_buffer_destroy(buf);
994 return rc;
995 }
996
997 static struct rpcrdma_req *
rpcrdma_buffer_get_req_locked(struct rpcrdma_buffer * buf)998 rpcrdma_buffer_get_req_locked(struct rpcrdma_buffer *buf)
999 {
1000 struct rpcrdma_req *req;
1001
1002 req = list_first_entry(&buf->rb_send_bufs,
1003 struct rpcrdma_req, rl_free);
1004 list_del(&req->rl_free);
1005 return req;
1006 }
1007
1008 static struct rpcrdma_rep *
rpcrdma_buffer_get_rep_locked(struct rpcrdma_buffer * buf)1009 rpcrdma_buffer_get_rep_locked(struct rpcrdma_buffer *buf)
1010 {
1011 struct rpcrdma_rep *rep;
1012
1013 rep = list_first_entry(&buf->rb_recv_bufs,
1014 struct rpcrdma_rep, rr_list);
1015 list_del(&rep->rr_list);
1016 return rep;
1017 }
1018
1019 static void
rpcrdma_destroy_rep(struct rpcrdma_ia * ia,struct rpcrdma_rep * rep)1020 rpcrdma_destroy_rep(struct rpcrdma_ia *ia, struct rpcrdma_rep *rep)
1021 {
1022 rpcrdma_free_regbuf(ia, rep->rr_rdmabuf);
1023 kfree(rep);
1024 }
1025
1026 void
rpcrdma_destroy_req(struct rpcrdma_ia * ia,struct rpcrdma_req * req)1027 rpcrdma_destroy_req(struct rpcrdma_ia *ia, struct rpcrdma_req *req)
1028 {
1029 rpcrdma_free_regbuf(ia, req->rl_sendbuf);
1030 rpcrdma_free_regbuf(ia, req->rl_rdmabuf);
1031 kfree(req);
1032 }
1033
1034 void
rpcrdma_buffer_destroy(struct rpcrdma_buffer * buf)1035 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf)
1036 {
1037 struct rpcrdma_ia *ia = rdmab_to_ia(buf);
1038
1039 while (!list_empty(&buf->rb_recv_bufs)) {
1040 struct rpcrdma_rep *rep;
1041
1042 rep = rpcrdma_buffer_get_rep_locked(buf);
1043 rpcrdma_destroy_rep(ia, rep);
1044 }
1045
1046 spin_lock(&buf->rb_reqslock);
1047 while (!list_empty(&buf->rb_allreqs)) {
1048 struct rpcrdma_req *req;
1049
1050 req = list_first_entry(&buf->rb_allreqs,
1051 struct rpcrdma_req, rl_all);
1052 list_del(&req->rl_all);
1053
1054 spin_unlock(&buf->rb_reqslock);
1055 rpcrdma_destroy_req(ia, req);
1056 spin_lock(&buf->rb_reqslock);
1057 }
1058 spin_unlock(&buf->rb_reqslock);
1059
1060 ia->ri_ops->ro_destroy(buf);
1061 }
1062
1063 struct rpcrdma_mw *
rpcrdma_get_mw(struct rpcrdma_xprt * r_xprt)1064 rpcrdma_get_mw(struct rpcrdma_xprt *r_xprt)
1065 {
1066 struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1067 struct rpcrdma_mw *mw = NULL;
1068
1069 spin_lock(&buf->rb_mwlock);
1070 if (!list_empty(&buf->rb_mws)) {
1071 mw = list_first_entry(&buf->rb_mws,
1072 struct rpcrdma_mw, mw_list);
1073 list_del_init(&mw->mw_list);
1074 }
1075 spin_unlock(&buf->rb_mwlock);
1076
1077 if (!mw)
1078 pr_err("RPC: %s: no MWs available\n", __func__);
1079 return mw;
1080 }
1081
1082 void
rpcrdma_put_mw(struct rpcrdma_xprt * r_xprt,struct rpcrdma_mw * mw)1083 rpcrdma_put_mw(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mw *mw)
1084 {
1085 struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1086
1087 spin_lock(&buf->rb_mwlock);
1088 list_add_tail(&mw->mw_list, &buf->rb_mws);
1089 spin_unlock(&buf->rb_mwlock);
1090 }
1091
1092 /*
1093 * Get a set of request/reply buffers.
1094 *
1095 * Reply buffer (if available) is attached to send buffer upon return.
1096 */
1097 struct rpcrdma_req *
rpcrdma_buffer_get(struct rpcrdma_buffer * buffers)1098 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers)
1099 {
1100 struct rpcrdma_req *req;
1101
1102 spin_lock(&buffers->rb_lock);
1103 if (list_empty(&buffers->rb_send_bufs))
1104 goto out_reqbuf;
1105 req = rpcrdma_buffer_get_req_locked(buffers);
1106 if (list_empty(&buffers->rb_recv_bufs))
1107 goto out_repbuf;
1108 req->rl_reply = rpcrdma_buffer_get_rep_locked(buffers);
1109 spin_unlock(&buffers->rb_lock);
1110 return req;
1111
1112 out_reqbuf:
1113 spin_unlock(&buffers->rb_lock);
1114 pr_warn("RPC: %s: out of request buffers\n", __func__);
1115 return NULL;
1116 out_repbuf:
1117 spin_unlock(&buffers->rb_lock);
1118 pr_warn("RPC: %s: out of reply buffers\n", __func__);
1119 req->rl_reply = NULL;
1120 return req;
1121 }
1122
1123 /*
1124 * Put request/reply buffers back into pool.
1125 * Pre-decrement counter/array index.
1126 */
1127 void
rpcrdma_buffer_put(struct rpcrdma_req * req)1128 rpcrdma_buffer_put(struct rpcrdma_req *req)
1129 {
1130 struct rpcrdma_buffer *buffers = req->rl_buffer;
1131 struct rpcrdma_rep *rep = req->rl_reply;
1132
1133 req->rl_niovs = 0;
1134 req->rl_reply = NULL;
1135
1136 spin_lock(&buffers->rb_lock);
1137 list_add_tail(&req->rl_free, &buffers->rb_send_bufs);
1138 if (rep)
1139 list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs);
1140 spin_unlock(&buffers->rb_lock);
1141 }
1142
1143 /*
1144 * Recover reply buffers from pool.
1145 * This happens when recovering from disconnect.
1146 */
1147 void
rpcrdma_recv_buffer_get(struct rpcrdma_req * req)1148 rpcrdma_recv_buffer_get(struct rpcrdma_req *req)
1149 {
1150 struct rpcrdma_buffer *buffers = req->rl_buffer;
1151
1152 spin_lock(&buffers->rb_lock);
1153 if (!list_empty(&buffers->rb_recv_bufs))
1154 req->rl_reply = rpcrdma_buffer_get_rep_locked(buffers);
1155 spin_unlock(&buffers->rb_lock);
1156 }
1157
1158 /*
1159 * Put reply buffers back into pool when not attached to
1160 * request. This happens in error conditions.
1161 */
1162 void
rpcrdma_recv_buffer_put(struct rpcrdma_rep * rep)1163 rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep)
1164 {
1165 struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf;
1166
1167 spin_lock(&buffers->rb_lock);
1168 list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs);
1169 spin_unlock(&buffers->rb_lock);
1170 }
1171
1172 /*
1173 * Wrappers for internal-use kmalloc memory registration, used by buffer code.
1174 */
1175
1176 void
rpcrdma_mapping_error(struct rpcrdma_mr_seg * seg)1177 rpcrdma_mapping_error(struct rpcrdma_mr_seg *seg)
1178 {
1179 dprintk("RPC: map_one: offset %p iova %llx len %zu\n",
1180 seg->mr_offset,
1181 (unsigned long long)seg->mr_dma, seg->mr_dmalen);
1182 }
1183
1184 /**
1185 * rpcrdma_alloc_regbuf - kmalloc and register memory for SEND/RECV buffers
1186 * @ia: controlling rpcrdma_ia
1187 * @size: size of buffer to be allocated, in bytes
1188 * @flags: GFP flags
1189 *
1190 * Returns pointer to private header of an area of internally
1191 * registered memory, or an ERR_PTR. The registered buffer follows
1192 * the end of the private header.
1193 *
1194 * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for
1195 * receiving the payload of RDMA RECV operations. regbufs are not
1196 * used for RDMA READ/WRITE operations, thus are registered only for
1197 * LOCAL access.
1198 */
1199 struct rpcrdma_regbuf *
rpcrdma_alloc_regbuf(struct rpcrdma_ia * ia,size_t size,gfp_t flags)1200 rpcrdma_alloc_regbuf(struct rpcrdma_ia *ia, size_t size, gfp_t flags)
1201 {
1202 struct rpcrdma_regbuf *rb;
1203 struct ib_sge *iov;
1204
1205 rb = kmalloc(sizeof(*rb) + size, flags);
1206 if (rb == NULL)
1207 goto out;
1208
1209 iov = &rb->rg_iov;
1210 iov->addr = ib_dma_map_single(ia->ri_device,
1211 (void *)rb->rg_base, size,
1212 DMA_BIDIRECTIONAL);
1213 if (ib_dma_mapping_error(ia->ri_device, iov->addr))
1214 goto out_free;
1215
1216 iov->length = size;
1217 iov->lkey = ia->ri_pd->local_dma_lkey;
1218 rb->rg_size = size;
1219 rb->rg_owner = NULL;
1220 return rb;
1221
1222 out_free:
1223 kfree(rb);
1224 out:
1225 return ERR_PTR(-ENOMEM);
1226 }
1227
1228 /**
1229 * rpcrdma_free_regbuf - deregister and free registered buffer
1230 * @ia: controlling rpcrdma_ia
1231 * @rb: regbuf to be deregistered and freed
1232 */
1233 void
rpcrdma_free_regbuf(struct rpcrdma_ia * ia,struct rpcrdma_regbuf * rb)1234 rpcrdma_free_regbuf(struct rpcrdma_ia *ia, struct rpcrdma_regbuf *rb)
1235 {
1236 struct ib_sge *iov;
1237
1238 if (!rb)
1239 return;
1240
1241 iov = &rb->rg_iov;
1242 ib_dma_unmap_single(ia->ri_device,
1243 iov->addr, iov->length, DMA_BIDIRECTIONAL);
1244 kfree(rb);
1245 }
1246
1247 /*
1248 * Prepost any receive buffer, then post send.
1249 *
1250 * Receive buffer is donated to hardware, reclaimed upon recv completion.
1251 */
1252 int
rpcrdma_ep_post(struct rpcrdma_ia * ia,struct rpcrdma_ep * ep,struct rpcrdma_req * req)1253 rpcrdma_ep_post(struct rpcrdma_ia *ia,
1254 struct rpcrdma_ep *ep,
1255 struct rpcrdma_req *req)
1256 {
1257 struct ib_device *device = ia->ri_device;
1258 struct ib_send_wr send_wr, *send_wr_fail;
1259 struct rpcrdma_rep *rep = req->rl_reply;
1260 struct ib_sge *iov = req->rl_send_iov;
1261 int i, rc;
1262
1263 if (rep) {
1264 rc = rpcrdma_ep_post_recv(ia, ep, rep);
1265 if (rc)
1266 goto out;
1267 req->rl_reply = NULL;
1268 }
1269
1270 send_wr.next = NULL;
1271 send_wr.wr_id = RPCRDMA_IGNORE_COMPLETION;
1272 send_wr.sg_list = iov;
1273 send_wr.num_sge = req->rl_niovs;
1274 send_wr.opcode = IB_WR_SEND;
1275
1276 for (i = 0; i < send_wr.num_sge; i++)
1277 ib_dma_sync_single_for_device(device, iov[i].addr,
1278 iov[i].length, DMA_TO_DEVICE);
1279 dprintk("RPC: %s: posting %d s/g entries\n",
1280 __func__, send_wr.num_sge);
1281
1282 if (DECR_CQCOUNT(ep) > 0)
1283 send_wr.send_flags = 0;
1284 else { /* Provider must take a send completion every now and then */
1285 INIT_CQCOUNT(ep);
1286 send_wr.send_flags = IB_SEND_SIGNALED;
1287 }
1288
1289 rc = ib_post_send(ia->ri_id->qp, &send_wr, &send_wr_fail);
1290 if (rc)
1291 dprintk("RPC: %s: ib_post_send returned %i\n", __func__,
1292 rc);
1293 out:
1294 return rc;
1295 }
1296
1297 /*
1298 * (Re)post a receive buffer.
1299 */
1300 int
rpcrdma_ep_post_recv(struct rpcrdma_ia * ia,struct rpcrdma_ep * ep,struct rpcrdma_rep * rep)1301 rpcrdma_ep_post_recv(struct rpcrdma_ia *ia,
1302 struct rpcrdma_ep *ep,
1303 struct rpcrdma_rep *rep)
1304 {
1305 struct ib_recv_wr recv_wr, *recv_wr_fail;
1306 int rc;
1307
1308 recv_wr.next = NULL;
1309 recv_wr.wr_id = (u64) (unsigned long) rep;
1310 recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov;
1311 recv_wr.num_sge = 1;
1312
1313 ib_dma_sync_single_for_cpu(ia->ri_device,
1314 rdmab_addr(rep->rr_rdmabuf),
1315 rdmab_length(rep->rr_rdmabuf),
1316 DMA_BIDIRECTIONAL);
1317
1318 rc = ib_post_recv(ia->ri_id->qp, &recv_wr, &recv_wr_fail);
1319
1320 if (rc)
1321 dprintk("RPC: %s: ib_post_recv returned %i\n", __func__,
1322 rc);
1323 return rc;
1324 }
1325
1326 /**
1327 * rpcrdma_ep_post_extra_recv - Post buffers for incoming backchannel requests
1328 * @r_xprt: transport associated with these backchannel resources
1329 * @min_reqs: minimum number of incoming requests expected
1330 *
1331 * Returns zero if all requested buffers were posted, or a negative errno.
1332 */
1333 int
rpcrdma_ep_post_extra_recv(struct rpcrdma_xprt * r_xprt,unsigned int count)1334 rpcrdma_ep_post_extra_recv(struct rpcrdma_xprt *r_xprt, unsigned int count)
1335 {
1336 struct rpcrdma_buffer *buffers = &r_xprt->rx_buf;
1337 struct rpcrdma_ia *ia = &r_xprt->rx_ia;
1338 struct rpcrdma_ep *ep = &r_xprt->rx_ep;
1339 struct rpcrdma_rep *rep;
1340 unsigned long flags;
1341 int rc;
1342
1343 while (count--) {
1344 spin_lock_irqsave(&buffers->rb_lock, flags);
1345 if (list_empty(&buffers->rb_recv_bufs))
1346 goto out_reqbuf;
1347 rep = rpcrdma_buffer_get_rep_locked(buffers);
1348 spin_unlock_irqrestore(&buffers->rb_lock, flags);
1349
1350 rc = rpcrdma_ep_post_recv(ia, ep, rep);
1351 if (rc)
1352 goto out_rc;
1353 }
1354
1355 return 0;
1356
1357 out_reqbuf:
1358 spin_unlock_irqrestore(&buffers->rb_lock, flags);
1359 pr_warn("%s: no extra receive buffers\n", __func__);
1360 return -ENOMEM;
1361
1362 out_rc:
1363 rpcrdma_recv_buffer_put(rep);
1364 return rc;
1365 }
1366
1367 /* How many chunk list items fit within our inline buffers?
1368 */
1369 unsigned int
rpcrdma_max_segments(struct rpcrdma_xprt * r_xprt)1370 rpcrdma_max_segments(struct rpcrdma_xprt *r_xprt)
1371 {
1372 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
1373 int bytes, segments;
1374
1375 bytes = min_t(unsigned int, cdata->inline_wsize, cdata->inline_rsize);
1376 bytes -= RPCRDMA_HDRLEN_MIN;
1377 if (bytes < sizeof(struct rpcrdma_segment) * 2) {
1378 pr_warn("RPC: %s: inline threshold too small\n",
1379 __func__);
1380 return 0;
1381 }
1382
1383 segments = 1 << (fls(bytes / sizeof(struct rpcrdma_segment)) - 1);
1384 dprintk("RPC: %s: max chunk list size = %d segments\n",
1385 __func__, segments);
1386 return segments;
1387 }
1388