2 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
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
10 * Redistribution and use in source and binary forms, with or without
11 * modification, are permitted provided that the following conditions
14 * Redistributions of source code must retain the above copyright
15 * notice, this list of conditions and the following disclaimer.
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.
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
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.
43 * Encapsulates the major functions managing:
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>
56 #include "xprt_rdma.h"
62 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
63 # define RPCDBG_FACILITY RPCDBG_TRANS
71 * handle replies in tasklet context, using a single, global list
72 * rdma tasklet function -- just turn around and call the func
73 * for all replies on the list
76 static DEFINE_SPINLOCK(rpcrdma_tk_lock_g);
77 static LIST_HEAD(rpcrdma_tasklets_g);
80 rpcrdma_run_tasklet(unsigned long data)
82 struct rpcrdma_rep *rep;
86 spin_lock_irqsave(&rpcrdma_tk_lock_g, flags);
87 while (!list_empty(&rpcrdma_tasklets_g)) {
88 rep = list_entry(rpcrdma_tasklets_g.next,
89 struct rpcrdma_rep, rr_list);
90 list_del(&rep->rr_list);
91 spin_unlock_irqrestore(&rpcrdma_tk_lock_g, flags);
93 rpcrdma_reply_handler(rep);
95 spin_lock_irqsave(&rpcrdma_tk_lock_g, flags);
97 spin_unlock_irqrestore(&rpcrdma_tk_lock_g, flags);
100 static DECLARE_TASKLET(rpcrdma_tasklet_g, rpcrdma_run_tasklet, 0UL);
103 rpcrdma_schedule_tasklet(struct list_head *sched_list)
107 spin_lock_irqsave(&rpcrdma_tk_lock_g, flags);
108 list_splice_tail(sched_list, &rpcrdma_tasklets_g);
109 spin_unlock_irqrestore(&rpcrdma_tk_lock_g, flags);
110 tasklet_schedule(&rpcrdma_tasklet_g);
114 rpcrdma_qp_async_error_upcall(struct ib_event *event, void *context)
116 struct rpcrdma_ep *ep = context;
118 pr_err("RPC: %s: %s on device %s ep %p\n",
119 __func__, ib_event_msg(event->event),
120 event->device->name, context);
121 if (ep->rep_connected == 1) {
122 ep->rep_connected = -EIO;
123 rpcrdma_conn_func(ep);
124 wake_up_all(&ep->rep_connect_wait);
129 rpcrdma_cq_async_error_upcall(struct ib_event *event, void *context)
131 struct rpcrdma_ep *ep = context;
133 pr_err("RPC: %s: %s on device %s ep %p\n",
134 __func__, ib_event_msg(event->event),
135 event->device->name, context);
136 if (ep->rep_connected == 1) {
137 ep->rep_connected = -EIO;
138 rpcrdma_conn_func(ep);
139 wake_up_all(&ep->rep_connect_wait);
144 rpcrdma_sendcq_process_wc(struct ib_wc *wc)
146 /* WARNING: Only wr_id and status are reliable at this point */
147 if (wc->wr_id == RPCRDMA_IGNORE_COMPLETION) {
148 if (wc->status != IB_WC_SUCCESS &&
149 wc->status != IB_WC_WR_FLUSH_ERR)
150 pr_err("RPC: %s: SEND: %s\n",
151 __func__, ib_wc_status_msg(wc->status));
153 struct rpcrdma_mw *r;
155 r = (struct rpcrdma_mw *)(unsigned long)wc->wr_id;
156 r->mw_sendcompletion(wc);
161 rpcrdma_sendcq_poll(struct ib_cq *cq, struct rpcrdma_ep *ep)
164 int budget, count, rc;
166 budget = RPCRDMA_WC_BUDGET / RPCRDMA_POLLSIZE;
168 wcs = ep->rep_send_wcs;
170 rc = ib_poll_cq(cq, RPCRDMA_POLLSIZE, wcs);
176 rpcrdma_sendcq_process_wc(wcs++);
177 } while (rc == RPCRDMA_POLLSIZE && --budget);
182 * Handle send, fast_reg_mr, and local_inv completions.
184 * Send events are typically suppressed and thus do not result
185 * in an upcall. Occasionally one is signaled, however. This
186 * prevents the provider's completion queue from wrapping and
187 * losing a completion.
190 rpcrdma_sendcq_upcall(struct ib_cq *cq, void *cq_context)
192 struct rpcrdma_ep *ep = (struct rpcrdma_ep *)cq_context;
195 rc = rpcrdma_sendcq_poll(cq, ep);
197 dprintk("RPC: %s: ib_poll_cq failed: %i\n",
202 rc = ib_req_notify_cq(cq,
203 IB_CQ_NEXT_COMP | IB_CQ_REPORT_MISSED_EVENTS);
207 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n",
212 rpcrdma_sendcq_poll(cq, ep);
216 rpcrdma_recvcq_process_wc(struct ib_wc *wc, struct list_head *sched_list)
218 struct rpcrdma_rep *rep =
219 (struct rpcrdma_rep *)(unsigned long)wc->wr_id;
221 /* WARNING: Only wr_id and status are reliable at this point */
222 if (wc->status != IB_WC_SUCCESS)
225 /* status == SUCCESS means all fields in wc are trustworthy */
226 if (wc->opcode != IB_WC_RECV)
229 dprintk("RPC: %s: rep %p opcode 'recv', length %u: success\n",
230 __func__, rep, wc->byte_len);
232 rep->rr_len = wc->byte_len;
233 ib_dma_sync_single_for_cpu(rep->rr_device,
234 rdmab_addr(rep->rr_rdmabuf),
235 rep->rr_len, DMA_FROM_DEVICE);
236 prefetch(rdmab_to_msg(rep->rr_rdmabuf));
239 list_add_tail(&rep->rr_list, sched_list);
242 if (wc->status != IB_WC_WR_FLUSH_ERR)
243 pr_err("RPC: %s: rep %p: %s\n",
244 __func__, rep, ib_wc_status_msg(wc->status));
250 rpcrdma_recvcq_poll(struct ib_cq *cq, struct rpcrdma_ep *ep)
252 struct list_head sched_list;
254 int budget, count, rc;
256 INIT_LIST_HEAD(&sched_list);
257 budget = RPCRDMA_WC_BUDGET / RPCRDMA_POLLSIZE;
259 wcs = ep->rep_recv_wcs;
261 rc = ib_poll_cq(cq, RPCRDMA_POLLSIZE, wcs);
267 rpcrdma_recvcq_process_wc(wcs++, &sched_list);
268 } while (rc == RPCRDMA_POLLSIZE && --budget);
272 rpcrdma_schedule_tasklet(&sched_list);
277 * Handle receive completions.
279 * It is reentrant but processes single events in order to maintain
280 * ordering of receives to keep server credits.
282 * It is the responsibility of the scheduled tasklet to return
283 * recv buffers to the pool. NOTE: this affects synchronization of
284 * connection shutdown. That is, the structures required for
285 * the completion of the reply handler must remain intact until
286 * all memory has been reclaimed.
289 rpcrdma_recvcq_upcall(struct ib_cq *cq, void *cq_context)
291 struct rpcrdma_ep *ep = (struct rpcrdma_ep *)cq_context;
294 rc = rpcrdma_recvcq_poll(cq, ep);
296 dprintk("RPC: %s: ib_poll_cq failed: %i\n",
301 rc = ib_req_notify_cq(cq,
302 IB_CQ_NEXT_COMP | IB_CQ_REPORT_MISSED_EVENTS);
306 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n",
311 rpcrdma_recvcq_poll(cq, ep);
315 rpcrdma_flush_cqs(struct rpcrdma_ep *ep)
318 LIST_HEAD(sched_list);
320 while (ib_poll_cq(ep->rep_attr.recv_cq, 1, &wc) > 0)
321 rpcrdma_recvcq_process_wc(&wc, &sched_list);
322 if (!list_empty(&sched_list))
323 rpcrdma_schedule_tasklet(&sched_list);
324 while (ib_poll_cq(ep->rep_attr.send_cq, 1, &wc) > 0)
325 rpcrdma_sendcq_process_wc(&wc);
329 rpcrdma_conn_upcall(struct rdma_cm_id *id, struct rdma_cm_event *event)
331 struct rpcrdma_xprt *xprt = id->context;
332 struct rpcrdma_ia *ia = &xprt->rx_ia;
333 struct rpcrdma_ep *ep = &xprt->rx_ep;
334 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
335 struct sockaddr *sap = (struct sockaddr *)&ep->rep_remote_addr;
337 struct ib_qp_attr *attr = &ia->ri_qp_attr;
338 struct ib_qp_init_attr *iattr = &ia->ri_qp_init_attr;
341 switch (event->event) {
342 case RDMA_CM_EVENT_ADDR_RESOLVED:
343 case RDMA_CM_EVENT_ROUTE_RESOLVED:
345 complete(&ia->ri_done);
347 case RDMA_CM_EVENT_ADDR_ERROR:
348 ia->ri_async_rc = -EHOSTUNREACH;
349 dprintk("RPC: %s: CM address resolution error, ep 0x%p\n",
351 complete(&ia->ri_done);
353 case RDMA_CM_EVENT_ROUTE_ERROR:
354 ia->ri_async_rc = -ENETUNREACH;
355 dprintk("RPC: %s: CM route resolution error, ep 0x%p\n",
357 complete(&ia->ri_done);
359 case RDMA_CM_EVENT_ESTABLISHED:
361 ib_query_qp(ia->ri_id->qp, attr,
362 IB_QP_MAX_QP_RD_ATOMIC | IB_QP_MAX_DEST_RD_ATOMIC,
364 dprintk("RPC: %s: %d responder resources"
366 __func__, attr->max_dest_rd_atomic,
367 attr->max_rd_atomic);
369 case RDMA_CM_EVENT_CONNECT_ERROR:
370 connstate = -ENOTCONN;
372 case RDMA_CM_EVENT_UNREACHABLE:
373 connstate = -ENETDOWN;
375 case RDMA_CM_EVENT_REJECTED:
376 connstate = -ECONNREFUSED;
378 case RDMA_CM_EVENT_DISCONNECTED:
379 connstate = -ECONNABORTED;
381 case RDMA_CM_EVENT_DEVICE_REMOVAL:
384 dprintk("RPC: %s: %sconnected\n",
385 __func__, connstate > 0 ? "" : "dis");
386 ep->rep_connected = connstate;
387 rpcrdma_conn_func(ep);
388 wake_up_all(&ep->rep_connect_wait);
391 dprintk("RPC: %s: %pIS:%u (ep 0x%p): %s\n",
392 __func__, sap, rpc_get_port(sap), ep,
393 rdma_event_msg(event->event));
397 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
398 if (connstate == 1) {
399 int ird = attr->max_dest_rd_atomic;
400 int tird = ep->rep_remote_cma.responder_resources;
402 pr_info("rpcrdma: connection to %pIS:%u on %s, memreg '%s', %d credits, %d responders%s\n",
403 sap, rpc_get_port(sap),
405 ia->ri_ops->ro_displayname,
406 xprt->rx_buf.rb_max_requests,
407 ird, ird < 4 && ird < tird / 2 ? " (low!)" : "");
408 } else if (connstate < 0) {
409 pr_info("rpcrdma: connection to %pIS:%u closed (%d)\n",
410 sap, rpc_get_port(sap), connstate);
417 static struct rdma_cm_id *
418 rpcrdma_create_id(struct rpcrdma_xprt *xprt,
419 struct rpcrdma_ia *ia, struct sockaddr *addr)
421 struct rdma_cm_id *id;
424 init_completion(&ia->ri_done);
426 id = rdma_create_id(rpcrdma_conn_upcall, xprt, RDMA_PS_TCP, IB_QPT_RC);
429 dprintk("RPC: %s: rdma_create_id() failed %i\n",
434 ia->ri_async_rc = -ETIMEDOUT;
435 rc = rdma_resolve_addr(id, NULL, addr, RDMA_RESOLVE_TIMEOUT);
437 dprintk("RPC: %s: rdma_resolve_addr() failed %i\n",
441 wait_for_completion_interruptible_timeout(&ia->ri_done,
442 msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1);
443 rc = ia->ri_async_rc;
447 ia->ri_async_rc = -ETIMEDOUT;
448 rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT);
450 dprintk("RPC: %s: rdma_resolve_route() failed %i\n",
454 wait_for_completion_interruptible_timeout(&ia->ri_done,
455 msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1);
456 rc = ia->ri_async_rc;
468 * Drain any cq, prior to teardown.
471 rpcrdma_clean_cq(struct ib_cq *cq)
476 while (1 == ib_poll_cq(cq, 1, &wc))
480 dprintk("RPC: %s: flushed %d events (last 0x%x)\n",
481 __func__, count, wc.opcode);
485 * Exported functions.
489 * Open and initialize an Interface Adapter.
490 * o initializes fields of struct rpcrdma_ia, including
491 * interface and provider attributes and protection zone.
494 rpcrdma_ia_open(struct rpcrdma_xprt *xprt, struct sockaddr *addr, int memreg)
497 struct rpcrdma_ia *ia = &xprt->rx_ia;
498 struct ib_device_attr *devattr = &ia->ri_devattr;
500 ia->ri_id = rpcrdma_create_id(xprt, ia, addr);
501 if (IS_ERR(ia->ri_id)) {
502 rc = PTR_ERR(ia->ri_id);
505 ia->ri_device = ia->ri_id->device;
507 ia->ri_pd = ib_alloc_pd(ia->ri_device);
508 if (IS_ERR(ia->ri_pd)) {
509 rc = PTR_ERR(ia->ri_pd);
510 dprintk("RPC: %s: ib_alloc_pd() failed %i\n",
515 rc = ib_query_device(ia->ri_device, devattr);
517 dprintk("RPC: %s: ib_query_device failed %d\n",
522 if (devattr->device_cap_flags & IB_DEVICE_LOCAL_DMA_LKEY) {
523 ia->ri_have_dma_lkey = 1;
524 ia->ri_dma_lkey = ia->ri_device->local_dma_lkey;
527 if (memreg == RPCRDMA_FRMR) {
528 /* Requires both frmr reg and local dma lkey */
529 if (((devattr->device_cap_flags &
530 (IB_DEVICE_MEM_MGT_EXTENSIONS|IB_DEVICE_LOCAL_DMA_LKEY)) !=
531 (IB_DEVICE_MEM_MGT_EXTENSIONS|IB_DEVICE_LOCAL_DMA_LKEY)) ||
532 (devattr->max_fast_reg_page_list_len == 0)) {
533 dprintk("RPC: %s: FRMR registration "
534 "not supported by HCA\n", __func__);
535 memreg = RPCRDMA_MTHCAFMR;
538 if (memreg == RPCRDMA_MTHCAFMR) {
539 if (!ia->ri_device->alloc_fmr) {
540 dprintk("RPC: %s: MTHCAFMR registration "
541 "not supported by HCA\n", __func__);
542 memreg = RPCRDMA_ALLPHYSICAL;
547 * Optionally obtain an underlying physical identity mapping in
548 * order to do a memory window-based bind. This base registration
549 * is protected from remote access - that is enabled only by binding
550 * for the specific bytes targeted during each RPC operation, and
551 * revoked after the corresponding completion similar to a storage
556 ia->ri_ops = &rpcrdma_frwr_memreg_ops;
558 case RPCRDMA_ALLPHYSICAL:
559 ia->ri_ops = &rpcrdma_physical_memreg_ops;
560 mem_priv = IB_ACCESS_LOCAL_WRITE |
561 IB_ACCESS_REMOTE_WRITE |
562 IB_ACCESS_REMOTE_READ;
564 case RPCRDMA_MTHCAFMR:
565 ia->ri_ops = &rpcrdma_fmr_memreg_ops;
566 if (ia->ri_have_dma_lkey)
568 mem_priv = IB_ACCESS_LOCAL_WRITE;
570 ia->ri_bind_mem = ib_get_dma_mr(ia->ri_pd, mem_priv);
571 if (IS_ERR(ia->ri_bind_mem)) {
572 printk(KERN_ALERT "%s: ib_get_dma_mr for "
573 "phys register failed with %lX\n",
574 __func__, PTR_ERR(ia->ri_bind_mem));
580 printk(KERN_ERR "RPC: Unsupported memory "
581 "registration mode: %d\n", memreg);
585 dprintk("RPC: %s: memory registration strategy is '%s'\n",
586 __func__, ia->ri_ops->ro_displayname);
588 rwlock_init(&ia->ri_qplock);
592 ib_dealloc_pd(ia->ri_pd);
595 rdma_destroy_id(ia->ri_id);
602 * Clean up/close an IA.
603 * o if event handles and PD have been initialized, free them.
607 rpcrdma_ia_close(struct rpcrdma_ia *ia)
611 dprintk("RPC: %s: entering\n", __func__);
612 if (ia->ri_bind_mem != NULL) {
613 rc = ib_dereg_mr(ia->ri_bind_mem);
614 dprintk("RPC: %s: ib_dereg_mr returned %i\n",
618 if (ia->ri_id != NULL && !IS_ERR(ia->ri_id)) {
620 rdma_destroy_qp(ia->ri_id);
621 rdma_destroy_id(ia->ri_id);
625 /* If the pd is still busy, xprtrdma missed freeing a resource */
626 if (ia->ri_pd && !IS_ERR(ia->ri_pd))
627 WARN_ON(ib_dealloc_pd(ia->ri_pd));
631 * Create unconnected endpoint.
634 rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia,
635 struct rpcrdma_create_data_internal *cdata)
637 struct ib_device_attr *devattr = &ia->ri_devattr;
638 struct ib_cq *sendcq, *recvcq;
639 struct ib_cq_init_attr cq_attr = {};
642 /* check provider's send/recv wr limits */
643 if (cdata->max_requests > devattr->max_qp_wr)
644 cdata->max_requests = devattr->max_qp_wr;
646 ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall;
647 ep->rep_attr.qp_context = ep;
648 ep->rep_attr.srq = NULL;
649 ep->rep_attr.cap.max_send_wr = cdata->max_requests;
650 rc = ia->ri_ops->ro_open(ia, ep, cdata);
653 ep->rep_attr.cap.max_recv_wr = cdata->max_requests;
654 ep->rep_attr.cap.max_send_sge = (cdata->padding ? 4 : 2);
655 ep->rep_attr.cap.max_recv_sge = 1;
656 ep->rep_attr.cap.max_inline_data = 0;
657 ep->rep_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
658 ep->rep_attr.qp_type = IB_QPT_RC;
659 ep->rep_attr.port_num = ~0;
661 if (cdata->padding) {
662 ep->rep_padbuf = rpcrdma_alloc_regbuf(ia, cdata->padding,
664 if (IS_ERR(ep->rep_padbuf))
665 return PTR_ERR(ep->rep_padbuf);
667 ep->rep_padbuf = NULL;
669 dprintk("RPC: %s: requested max: dtos: send %d recv %d; "
670 "iovs: send %d recv %d\n",
672 ep->rep_attr.cap.max_send_wr,
673 ep->rep_attr.cap.max_recv_wr,
674 ep->rep_attr.cap.max_send_sge,
675 ep->rep_attr.cap.max_recv_sge);
677 /* set trigger for requesting send completion */
678 ep->rep_cqinit = ep->rep_attr.cap.max_send_wr/2 - 1;
679 if (ep->rep_cqinit > RPCRDMA_MAX_UNSIGNALED_SENDS)
680 ep->rep_cqinit = RPCRDMA_MAX_UNSIGNALED_SENDS;
681 else if (ep->rep_cqinit <= 2)
684 init_waitqueue_head(&ep->rep_connect_wait);
685 INIT_DELAYED_WORK(&ep->rep_connect_worker, rpcrdma_connect_worker);
687 cq_attr.cqe = ep->rep_attr.cap.max_send_wr + 1;
688 sendcq = ib_create_cq(ia->ri_device, rpcrdma_sendcq_upcall,
689 rpcrdma_cq_async_error_upcall, ep, &cq_attr);
690 if (IS_ERR(sendcq)) {
691 rc = PTR_ERR(sendcq);
692 dprintk("RPC: %s: failed to create send CQ: %i\n",
697 rc = ib_req_notify_cq(sendcq, IB_CQ_NEXT_COMP);
699 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n",
704 cq_attr.cqe = ep->rep_attr.cap.max_recv_wr + 1;
705 recvcq = ib_create_cq(ia->ri_device, rpcrdma_recvcq_upcall,
706 rpcrdma_cq_async_error_upcall, ep, &cq_attr);
707 if (IS_ERR(recvcq)) {
708 rc = PTR_ERR(recvcq);
709 dprintk("RPC: %s: failed to create recv CQ: %i\n",
714 rc = ib_req_notify_cq(recvcq, IB_CQ_NEXT_COMP);
716 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n",
718 ib_destroy_cq(recvcq);
722 ep->rep_attr.send_cq = sendcq;
723 ep->rep_attr.recv_cq = recvcq;
725 /* Initialize cma parameters */
727 /* RPC/RDMA does not use private data */
728 ep->rep_remote_cma.private_data = NULL;
729 ep->rep_remote_cma.private_data_len = 0;
731 /* Client offers RDMA Read but does not initiate */
732 ep->rep_remote_cma.initiator_depth = 0;
733 if (devattr->max_qp_rd_atom > 32) /* arbitrary but <= 255 */
734 ep->rep_remote_cma.responder_resources = 32;
736 ep->rep_remote_cma.responder_resources =
737 devattr->max_qp_rd_atom;
739 ep->rep_remote_cma.retry_count = 7;
740 ep->rep_remote_cma.flow_control = 0;
741 ep->rep_remote_cma.rnr_retry_count = 0;
746 err = ib_destroy_cq(sendcq);
748 dprintk("RPC: %s: ib_destroy_cq returned %i\n",
751 rpcrdma_free_regbuf(ia, ep->rep_padbuf);
758 * Disconnect and destroy endpoint. After this, the only
759 * valid operations on the ep are to free it (if dynamically
760 * allocated) or re-create it.
763 rpcrdma_ep_destroy(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
767 dprintk("RPC: %s: entering, connected is %d\n",
768 __func__, ep->rep_connected);
770 cancel_delayed_work_sync(&ep->rep_connect_worker);
773 rpcrdma_ep_disconnect(ep, ia);
774 rdma_destroy_qp(ia->ri_id);
775 ia->ri_id->qp = NULL;
778 rpcrdma_free_regbuf(ia, ep->rep_padbuf);
780 rpcrdma_clean_cq(ep->rep_attr.recv_cq);
781 rc = ib_destroy_cq(ep->rep_attr.recv_cq);
783 dprintk("RPC: %s: ib_destroy_cq returned %i\n",
786 rpcrdma_clean_cq(ep->rep_attr.send_cq);
787 rc = ib_destroy_cq(ep->rep_attr.send_cq);
789 dprintk("RPC: %s: ib_destroy_cq returned %i\n",
794 * Connect unconnected endpoint.
797 rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
799 struct rdma_cm_id *id, *old;
803 if (ep->rep_connected != 0) {
804 struct rpcrdma_xprt *xprt;
806 dprintk("RPC: %s: reconnecting...\n", __func__);
808 rpcrdma_ep_disconnect(ep, ia);
809 rpcrdma_flush_cqs(ep);
811 xprt = container_of(ia, struct rpcrdma_xprt, rx_ia);
812 id = rpcrdma_create_id(xprt, ia,
813 (struct sockaddr *)&xprt->rx_data.addr);
818 /* TEMP TEMP TEMP - fail if new device:
819 * Deregister/remarshal *all* requests!
820 * Close and recreate adapter, pd, etc!
821 * Re-determine all attributes still sane!
822 * More stuff I haven't thought of!
825 if (ia->ri_device != id->device) {
826 printk("RPC: %s: can't reconnect on "
827 "different device!\n", __func__);
833 rc = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr);
835 dprintk("RPC: %s: rdma_create_qp failed %i\n",
842 write_lock(&ia->ri_qplock);
845 write_unlock(&ia->ri_qplock);
847 rdma_destroy_qp(old);
848 rdma_destroy_id(old);
850 dprintk("RPC: %s: connecting...\n", __func__);
851 rc = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr);
853 dprintk("RPC: %s: rdma_create_qp failed %i\n",
855 /* do not update ep->rep_connected */
860 ep->rep_connected = 0;
862 rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma);
864 dprintk("RPC: %s: rdma_connect() failed with %i\n",
869 wait_event_interruptible(ep->rep_connect_wait, ep->rep_connected != 0);
872 * Check state. A non-peer reject indicates no listener
873 * (ECONNREFUSED), which may be a transient state. All
874 * others indicate a transport condition which has already
875 * undergone a best-effort.
877 if (ep->rep_connected == -ECONNREFUSED &&
878 ++retry_count <= RDMA_CONNECT_RETRY_MAX) {
879 dprintk("RPC: %s: non-peer_reject, retry\n", __func__);
882 if (ep->rep_connected <= 0) {
883 /* Sometimes, the only way to reliably connect to remote
884 * CMs is to use same nonzero values for ORD and IRD. */
885 if (retry_count++ <= RDMA_CONNECT_RETRY_MAX + 1 &&
886 (ep->rep_remote_cma.responder_resources == 0 ||
887 ep->rep_remote_cma.initiator_depth !=
888 ep->rep_remote_cma.responder_resources)) {
889 if (ep->rep_remote_cma.responder_resources == 0)
890 ep->rep_remote_cma.responder_resources = 1;
891 ep->rep_remote_cma.initiator_depth =
892 ep->rep_remote_cma.responder_resources;
895 rc = ep->rep_connected;
897 dprintk("RPC: %s: connected\n", __func__);
902 ep->rep_connected = rc;
907 * rpcrdma_ep_disconnect
909 * This is separate from destroy to facilitate the ability
910 * to reconnect without recreating the endpoint.
912 * This call is not reentrant, and must not be made in parallel
913 * on the same endpoint.
916 rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
920 rpcrdma_flush_cqs(ep);
921 rc = rdma_disconnect(ia->ri_id);
923 /* returns without wait if not connected */
924 wait_event_interruptible(ep->rep_connect_wait,
925 ep->rep_connected != 1);
926 dprintk("RPC: %s: after wait, %sconnected\n", __func__,
927 (ep->rep_connected == 1) ? "still " : "dis");
929 dprintk("RPC: %s: rdma_disconnect %i\n", __func__, rc);
930 ep->rep_connected = rc;
934 static struct rpcrdma_req *
935 rpcrdma_create_req(struct rpcrdma_xprt *r_xprt)
937 struct rpcrdma_req *req;
939 req = kzalloc(sizeof(*req), GFP_KERNEL);
941 return ERR_PTR(-ENOMEM);
943 req->rl_buffer = &r_xprt->rx_buf;
947 static struct rpcrdma_rep *
948 rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt)
950 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
951 struct rpcrdma_ia *ia = &r_xprt->rx_ia;
952 struct rpcrdma_rep *rep;
956 rep = kzalloc(sizeof(*rep), GFP_KERNEL);
960 rep->rr_rdmabuf = rpcrdma_alloc_regbuf(ia, cdata->inline_rsize,
962 if (IS_ERR(rep->rr_rdmabuf)) {
963 rc = PTR_ERR(rep->rr_rdmabuf);
967 rep->rr_device = ia->ri_device;
968 rep->rr_rxprt = r_xprt;
978 rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
980 struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
981 struct rpcrdma_ia *ia = &r_xprt->rx_ia;
982 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
987 buf->rb_max_requests = cdata->max_requests;
988 spin_lock_init(&buf->rb_lock);
991 * 1. arrays for send and recv pointers
992 * 2. arrays of struct rpcrdma_req to fill in pointers
993 * 3. array of struct rpcrdma_rep for replies
994 * Send/recv buffers in req/rep need to be registered
996 len = buf->rb_max_requests *
997 (sizeof(struct rpcrdma_req *) + sizeof(struct rpcrdma_rep *));
999 p = kzalloc(len, GFP_KERNEL);
1001 dprintk("RPC: %s: req_t/rep_t/pad kzalloc(%zd) failed\n",
1006 buf->rb_pool = p; /* for freeing it later */
1008 buf->rb_send_bufs = (struct rpcrdma_req **) p;
1009 p = (char *) &buf->rb_send_bufs[buf->rb_max_requests];
1010 buf->rb_recv_bufs = (struct rpcrdma_rep **) p;
1011 p = (char *) &buf->rb_recv_bufs[buf->rb_max_requests];
1013 rc = ia->ri_ops->ro_init(r_xprt);
1017 for (i = 0; i < buf->rb_max_requests; i++) {
1018 struct rpcrdma_req *req;
1019 struct rpcrdma_rep *rep;
1021 req = rpcrdma_create_req(r_xprt);
1023 dprintk("RPC: %s: request buffer %d alloc"
1024 " failed\n", __func__, i);
1028 buf->rb_send_bufs[i] = req;
1030 rep = rpcrdma_create_rep(r_xprt);
1032 dprintk("RPC: %s: reply buffer %d alloc failed\n",
1037 buf->rb_recv_bufs[i] = rep;
1042 rpcrdma_buffer_destroy(buf);
1047 rpcrdma_destroy_rep(struct rpcrdma_ia *ia, struct rpcrdma_rep *rep)
1052 rpcrdma_free_regbuf(ia, rep->rr_rdmabuf);
1057 rpcrdma_destroy_req(struct rpcrdma_ia *ia, struct rpcrdma_req *req)
1062 rpcrdma_free_regbuf(ia, req->rl_sendbuf);
1063 rpcrdma_free_regbuf(ia, req->rl_rdmabuf);
1068 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf)
1070 struct rpcrdma_ia *ia = rdmab_to_ia(buf);
1073 /* clean up in reverse order from create
1074 * 1. recv mr memory (mr free, then kfree)
1075 * 2. send mr memory (mr free, then kfree)
1078 dprintk("RPC: %s: entering\n", __func__);
1080 for (i = 0; i < buf->rb_max_requests; i++) {
1081 if (buf->rb_recv_bufs)
1082 rpcrdma_destroy_rep(ia, buf->rb_recv_bufs[i]);
1083 if (buf->rb_send_bufs)
1084 rpcrdma_destroy_req(ia, buf->rb_send_bufs[i]);
1087 ia->ri_ops->ro_destroy(buf);
1089 kfree(buf->rb_pool);
1093 rpcrdma_get_mw(struct rpcrdma_xprt *r_xprt)
1095 struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1096 struct rpcrdma_mw *mw = NULL;
1098 spin_lock(&buf->rb_mwlock);
1099 if (!list_empty(&buf->rb_mws)) {
1100 mw = list_first_entry(&buf->rb_mws,
1101 struct rpcrdma_mw, mw_list);
1102 list_del_init(&mw->mw_list);
1104 spin_unlock(&buf->rb_mwlock);
1107 pr_err("RPC: %s: no MWs available\n", __func__);
1112 rpcrdma_put_mw(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mw *mw)
1114 struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1116 spin_lock(&buf->rb_mwlock);
1117 list_add_tail(&mw->mw_list, &buf->rb_mws);
1118 spin_unlock(&buf->rb_mwlock);
1122 rpcrdma_buffer_put_sendbuf(struct rpcrdma_req *req, struct rpcrdma_buffer *buf)
1124 buf->rb_send_bufs[--buf->rb_send_index] = req;
1126 if (req->rl_reply) {
1127 buf->rb_recv_bufs[--buf->rb_recv_index] = req->rl_reply;
1128 req->rl_reply = NULL;
1133 * Get a set of request/reply buffers.
1135 * Reply buffer (if needed) is attached to send buffer upon return.
1137 * rb_send_index and rb_recv_index MUST always be pointing to the
1138 * *next* available buffer (non-NULL). They are incremented after
1139 * removing buffers, and decremented *before* returning them.
1141 struct rpcrdma_req *
1142 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers)
1144 struct rpcrdma_req *req;
1145 unsigned long flags;
1147 spin_lock_irqsave(&buffers->rb_lock, flags);
1149 if (buffers->rb_send_index == buffers->rb_max_requests) {
1150 spin_unlock_irqrestore(&buffers->rb_lock, flags);
1151 dprintk("RPC: %s: out of request buffers\n", __func__);
1152 return ((struct rpcrdma_req *)NULL);
1155 req = buffers->rb_send_bufs[buffers->rb_send_index];
1156 if (buffers->rb_send_index < buffers->rb_recv_index) {
1157 dprintk("RPC: %s: %d extra receives outstanding (ok)\n",
1159 buffers->rb_recv_index - buffers->rb_send_index);
1160 req->rl_reply = NULL;
1162 req->rl_reply = buffers->rb_recv_bufs[buffers->rb_recv_index];
1163 buffers->rb_recv_bufs[buffers->rb_recv_index++] = NULL;
1165 buffers->rb_send_bufs[buffers->rb_send_index++] = NULL;
1167 spin_unlock_irqrestore(&buffers->rb_lock, flags);
1172 * Put request/reply buffers back into pool.
1173 * Pre-decrement counter/array index.
1176 rpcrdma_buffer_put(struct rpcrdma_req *req)
1178 struct rpcrdma_buffer *buffers = req->rl_buffer;
1179 unsigned long flags;
1181 spin_lock_irqsave(&buffers->rb_lock, flags);
1182 rpcrdma_buffer_put_sendbuf(req, buffers);
1183 spin_unlock_irqrestore(&buffers->rb_lock, flags);
1187 * Recover reply buffers from pool.
1188 * This happens when recovering from error conditions.
1189 * Post-increment counter/array index.
1192 rpcrdma_recv_buffer_get(struct rpcrdma_req *req)
1194 struct rpcrdma_buffer *buffers = req->rl_buffer;
1195 unsigned long flags;
1197 spin_lock_irqsave(&buffers->rb_lock, flags);
1198 if (buffers->rb_recv_index < buffers->rb_max_requests) {
1199 req->rl_reply = buffers->rb_recv_bufs[buffers->rb_recv_index];
1200 buffers->rb_recv_bufs[buffers->rb_recv_index++] = NULL;
1202 spin_unlock_irqrestore(&buffers->rb_lock, flags);
1206 * Put reply buffers back into pool when not attached to
1207 * request. This happens in error conditions.
1210 rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep)
1212 struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf;
1213 unsigned long flags;
1215 spin_lock_irqsave(&buffers->rb_lock, flags);
1216 buffers->rb_recv_bufs[--buffers->rb_recv_index] = rep;
1217 spin_unlock_irqrestore(&buffers->rb_lock, flags);
1221 * Wrappers for internal-use kmalloc memory registration, used by buffer code.
1225 rpcrdma_mapping_error(struct rpcrdma_mr_seg *seg)
1227 dprintk("RPC: map_one: offset %p iova %llx len %zu\n",
1229 (unsigned long long)seg->mr_dma, seg->mr_dmalen);
1233 rpcrdma_register_internal(struct rpcrdma_ia *ia, void *va, int len,
1234 struct ib_mr **mrp, struct ib_sge *iov)
1236 struct ib_phys_buf ipb;
1241 * All memory passed here was kmalloc'ed, therefore phys-contiguous.
1243 iov->addr = ib_dma_map_single(ia->ri_device,
1244 va, len, DMA_BIDIRECTIONAL);
1245 if (ib_dma_mapping_error(ia->ri_device, iov->addr))
1250 if (ia->ri_have_dma_lkey) {
1252 iov->lkey = ia->ri_dma_lkey;
1254 } else if (ia->ri_bind_mem != NULL) {
1256 iov->lkey = ia->ri_bind_mem->lkey;
1260 ipb.addr = iov->addr;
1261 ipb.size = iov->length;
1262 mr = ib_reg_phys_mr(ia->ri_pd, &ipb, 1,
1263 IB_ACCESS_LOCAL_WRITE, &iov->addr);
1265 dprintk("RPC: %s: phys convert: 0x%llx "
1266 "registered 0x%llx length %d\n",
1267 __func__, (unsigned long long)ipb.addr,
1268 (unsigned long long)iov->addr, len);
1273 dprintk("RPC: %s: failed with %i\n", __func__, rc);
1276 iov->lkey = mr->lkey;
1284 rpcrdma_deregister_internal(struct rpcrdma_ia *ia,
1285 struct ib_mr *mr, struct ib_sge *iov)
1289 ib_dma_unmap_single(ia->ri_device,
1290 iov->addr, iov->length, DMA_BIDIRECTIONAL);
1295 rc = ib_dereg_mr(mr);
1297 dprintk("RPC: %s: ib_dereg_mr failed %i\n", __func__, rc);
1302 * rpcrdma_alloc_regbuf - kmalloc and register memory for SEND/RECV buffers
1303 * @ia: controlling rpcrdma_ia
1304 * @size: size of buffer to be allocated, in bytes
1307 * Returns pointer to private header of an area of internally
1308 * registered memory, or an ERR_PTR. The registered buffer follows
1309 * the end of the private header.
1311 * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for
1312 * receiving the payload of RDMA RECV operations. regbufs are not
1313 * used for RDMA READ/WRITE operations, thus are registered only for
1316 struct rpcrdma_regbuf *
1317 rpcrdma_alloc_regbuf(struct rpcrdma_ia *ia, size_t size, gfp_t flags)
1319 struct rpcrdma_regbuf *rb;
1323 rb = kmalloc(sizeof(*rb) + size, flags);
1328 rb->rg_owner = NULL;
1329 rc = rpcrdma_register_internal(ia, rb->rg_base, size,
1330 &rb->rg_mr, &rb->rg_iov);
1343 * rpcrdma_free_regbuf - deregister and free registered buffer
1344 * @ia: controlling rpcrdma_ia
1345 * @rb: regbuf to be deregistered and freed
1348 rpcrdma_free_regbuf(struct rpcrdma_ia *ia, struct rpcrdma_regbuf *rb)
1351 rpcrdma_deregister_internal(ia, rb->rg_mr, &rb->rg_iov);
1357 * Prepost any receive buffer, then post send.
1359 * Receive buffer is donated to hardware, reclaimed upon recv completion.
1362 rpcrdma_ep_post(struct rpcrdma_ia *ia,
1363 struct rpcrdma_ep *ep,
1364 struct rpcrdma_req *req)
1366 struct ib_send_wr send_wr, *send_wr_fail;
1367 struct rpcrdma_rep *rep = req->rl_reply;
1371 rc = rpcrdma_ep_post_recv(ia, ep, rep);
1374 req->rl_reply = NULL;
1377 send_wr.next = NULL;
1378 send_wr.wr_id = RPCRDMA_IGNORE_COMPLETION;
1379 send_wr.sg_list = req->rl_send_iov;
1380 send_wr.num_sge = req->rl_niovs;
1381 send_wr.opcode = IB_WR_SEND;
1382 if (send_wr.num_sge == 4) /* no need to sync any pad (constant) */
1383 ib_dma_sync_single_for_device(ia->ri_device,
1384 req->rl_send_iov[3].addr,
1385 req->rl_send_iov[3].length,
1387 ib_dma_sync_single_for_device(ia->ri_device,
1388 req->rl_send_iov[1].addr,
1389 req->rl_send_iov[1].length,
1391 ib_dma_sync_single_for_device(ia->ri_device,
1392 req->rl_send_iov[0].addr,
1393 req->rl_send_iov[0].length,
1396 if (DECR_CQCOUNT(ep) > 0)
1397 send_wr.send_flags = 0;
1398 else { /* Provider must take a send completion every now and then */
1400 send_wr.send_flags = IB_SEND_SIGNALED;
1403 rc = ib_post_send(ia->ri_id->qp, &send_wr, &send_wr_fail);
1405 dprintk("RPC: %s: ib_post_send returned %i\n", __func__,
1412 * (Re)post a receive buffer.
1415 rpcrdma_ep_post_recv(struct rpcrdma_ia *ia,
1416 struct rpcrdma_ep *ep,
1417 struct rpcrdma_rep *rep)
1419 struct ib_recv_wr recv_wr, *recv_wr_fail;
1422 recv_wr.next = NULL;
1423 recv_wr.wr_id = (u64) (unsigned long) rep;
1424 recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov;
1425 recv_wr.num_sge = 1;
1427 ib_dma_sync_single_for_cpu(ia->ri_device,
1428 rdmab_addr(rep->rr_rdmabuf),
1429 rdmab_length(rep->rr_rdmabuf),
1432 rc = ib_post_recv(ia->ri_id->qp, &recv_wr, &recv_wr_fail);
1435 dprintk("RPC: %s: ib_post_recv returned %i\n", __func__,
1440 /* How many chunk list items fit within our inline buffers?
1443 rpcrdma_max_segments(struct rpcrdma_xprt *r_xprt)
1445 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
1446 int bytes, segments;
1448 bytes = min_t(unsigned int, cdata->inline_wsize, cdata->inline_rsize);
1449 bytes -= RPCRDMA_HDRLEN_MIN;
1450 if (bytes < sizeof(struct rpcrdma_segment) * 2) {
1451 pr_warn("RPC: %s: inline threshold too small\n",
1456 segments = 1 << (fls(bytes / sizeof(struct rpcrdma_segment)) - 1);
1457 dprintk("RPC: %s: max chunk list size = %d segments\n",
1458 __func__, segments);