Page MenuHomeFreeBSD
Files F171122973

D59278.id185964.diff
No OneTemporary
Actions

D59278.id185964.diff
View Options

diff --git a/clnt_rdma.c.rdma b/clnt_rdma.c
--- a/clnt_rdma.c.rdma
+++ b/clnt_rdma.c
@@ -0,0 +1,1614 @@
+/* $NetBSD: clnt_vc.c,v 1.4 2000/07/14 08:40:42 fvdl Exp $ */
+
+/*-
+ * SPDX-License-Identifier: BSD-3-Clause
+ *
+ * Copyright (c) 2009, Sun Microsystems, Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ * - Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ * - Neither the name of Sun Microsystems, Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived
+ * from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include <sys/cdefs.h>
+/*
+ * clnt_tcp.c, Implements a TCP/IP based, client side RPC.
+ *
+ * Copyright (C) 1984, Sun Microsystems, Inc.
+ *
+ * TCP based RPC supports 'batched calls'.
+ * A sequence of calls may be batched-up in a send buffer. The rpc call
+ * return immediately to the client even though the call was not necessarily
+ * sent. The batching occurs if the results' xdr routine is NULL (0) AND
+ * the rpc timeout value is zero (see clnt.h, rpc).
+ *
+ * Clients should NOT casually batch calls that in fact return results; that is,
+ * the server side should be aware that a call is batched and not produce any
+ * return message. Batched calls that produce many result messages can
+ * deadlock (netlock) the client and the server....
+ *
+ * Now go hang yourself.
+ */
+
+/*
+ * The code in this file handles RPC over RDMA as described by
+ * RFC-8166 and RFC-8267 and is loosely based on code that the above
+ * comment discusses.
+ */
+
+#include <sys/param.h>
+#include <sys/systm.h>
+#include <sys/kernel.h>
+#include <sys/kthread.h>
+#include <sys/ktls.h>
+#include <sys/lock.h>
+#include <sys/malloc.h>
+#include <sys/mbuf.h>
+#include <sys/module.h>
+#include <sys/mutex.h>
+#include <sys/pcpu.h>
+#include <sys/proc.h>
+#include <sys/protosw.h>
+#include <sys/socket.h>
+#include <sys/socketvar.h>
+#include <sys/sx.h>
+#include <sys/syslog.h>
+#include <sys/time.h>
+#include <sys/uio.h>
+
+#include <vm/vm.h>
+#include <vm/vm_page.h>
+
+#include <net/vnet.h>
+
+#include <netinet/tcp.h>
+
+#include <rpc/rpc.h>
+#include <rpc/rpc_com.h>
+#include <rpc/krpc.h>
+#include <rpc/clntrdma.h>
+
+/*
+ * This structure stores the internals for an RDMA connection endpoint.
+ */
+struct crdma_data {
+ int crdma_threads; /* # of threads in clnt_rdma_call */
+ bool_t crdma_closing; /* TRUE if we are closing */
+ bool_t crdma_closed; /* TRUE if we are closed */
+ struct rpcrdma_xprt crdma_xp; /* RDMA endpoint */
+ bool_t crdma_closeit; /* close it on destroy */
+ struct timeval crdma_wait; /* wait interval in milliseconds */
+ struct sockaddr_storage crdma_addr; /* remote addr */
+ struct rpc_err crdma_error;
+ uint32_t crdma_xid;
+ uint32_t crdma_prog;
+ uint32_t crdma_vers;
+ char crdma_mcallc[MCALL_MSG_SIZE]; /* call msg */
+ size_t crdma_mpos; /* pos after marshal */
+ const char *crdma_waitchan;
+ int crdma_waitflag;
+ struct mbuf *crdma_record; /* current reply record */
+ size_t crdma_record_resid; /* how much left of reply to read */
+ bool_t crdma_record_eor; /* true if reading last fragment */
+ struct ct_request_list crdma_pending;
+ SVCXPRT *crdma_backchannelxprt; /* xprt for backchannel */
+ struct mbuf *crdma_raw; /* Raw mbufs recv'd */
+ uint32_t crdma_small_reply; /* Max small reply M_PROTO8 */
+ uint32_t crdma_max_io; /* Max size of an I/O M_PROTO9 */
+};
+#define crdma_lock crdma_xp.mtx
+
+MALLOC_DECLARE(M_RPCRDMA);
+
+SYSCTL_DECL(_kern_rpc);
+SYSCTL_NODE(_kern_rpc, OID_AUTO, rdma, CTLFLAG_RW | CTLFLAG_MPSAFE, 0,
+ "RPC over RDMA");
+int rpcrdma_debuglevel = 0;
+SYSCTL_INT(_kern_rpc_rdma, OID_AUTO, debuglevel, CTLFLAG_RW,
+ &rpcrdma_debuglevel, 0, "Debug level for RPC over RDMA client");
+
+struct cmessage {
+ struct cmsghdr cmsg;
+ struct cmsgcred cmcred;
+};
+
+static enum clnt_stat clnt_rdma_call(CLIENT *, struct rpc_callextra *,
+ rpcproc_t, struct mbuf *, struct mbuf **, struct timeval);
+static void clnt_rdma_geterr(CLIENT *, struct rpc_err *);
+static bool_t clnt_rdma_freeres(CLIENT *, xdrproc_t, void *);
+static void clnt_rdma_abort(CLIENT *);
+static bool_t clnt_rdma_control(CLIENT *, u_int, void *);
+static void clnt_rdma_close(CLIENT *);
+static void clnt_rdma_destroy(CLIENT *);
+static bool_t time_not_ok(struct timeval *);
+static void clnt_rdma_upcall(struct rpcrdma_xprt *xp, struct mbuf *);
+static bool_t clnt_rdma_bcksend(SVCXPRT *, struct mbuf *);
+
+static const struct clnt_ops clnt_rdma_ops = {
+ .cl_call = clnt_rdma_call,
+ .cl_abort = clnt_rdma_abort,
+ .cl_geterr = clnt_rdma_geterr,
+ .cl_freeres = clnt_rdma_freeres,
+ .cl_close = clnt_rdma_close,
+ .cl_destroy = clnt_rdma_destroy,
+ .cl_control = clnt_rdma_control
+};
+
+
+/*
+ * Create a client handle for an RDMA connection.
+ */
+CLIENT *
+clnt_rdma_create(
+ struct sockaddr *raddr, /* servers address */
+ const rpcprog_t prog, /* program number */
+ const rpcvers_t vers, /* version number */
+ int intrflag, /* interruptible */
+ uint32_t small_reply, /* Max. medium reply */
+ uint32_t max_io, /* size of largest RPC I/O */
+ uint32_t cbslots, /* Max # of callbacks */
+ struct rpc_err *err) /* Return error and stat */
+{
+ CLIENT *cl; /* client handle */
+ struct crdma_data *crdma = NULL; /* client handle */
+ struct timeval now;
+ int error;
+ static uint32_t disrupt;
+
+ KASSERT(raddr->sa_family != AF_LOCAL,
+ ("%s: kernel RPC over unix(4) not supported", __func__));
+
+ if (disrupt == 0)
+ disrupt = (uint32_t)(long)raddr;
+
+ cl = (CLIENT *)mem_alloc(sizeof (*cl));
+ crdma = (struct crdma_data *)mem_alloc(sizeof (*crdma));
+
+ mtx_init(&crdma->crdma_lock, "crdma->crdma_lock", NULL, MTX_DEF);
+ crdma->crdma_threads = 0;
+ crdma->crdma_closing = FALSE;
+ crdma->crdma_closed = FALSE;
+ crdma->crdma_closeit = FALSE;
+
+ /*
+ * Set up private data struct
+ */
+ crdma->crdma_xp.credits = 1;
+ xprt_rdma_init(&crdma->crdma_xp, clnt_rdma_upcall);
+ crdma->crdma_wait.tv_sec = -1;
+ crdma->crdma_wait.tv_usec = -1;
+ memcpy(&crdma->crdma_addr, raddr, raddr->sa_len);
+ crdma->crdma_small_reply = round_page(small_reply);
+ crdma->crdma_max_io = round_page(max_io + 2048);
+
+ /* Attempt to connect.. */
+ error = xprt_rdma_connect(vnet0, raddr, &crdma->crdma_xp,
+ RPCRDMA_MAX_SMALL_MSG, cbslots);
+ RPCRDMA_DEBUG(1, "clnt_rdma_create: xprt_rdma_connect=%d\n", error);
+ if (error != 0) {
+ err->re_errno = error;
+ if (error == ENETUNREACH)
+ err->re_status = RPC_SYSTEMERROR;
+ else if (error == EPROTO)
+ err->re_status = RPC_UNKNOWNADDR;
+ else
+ err->re_status = RPC_UNKNOWNPROTO;
+ RPCRDMA_DEBUG(1, "clnt_rdma_create: Fail stat=%d\n",
+ err->re_status);
+ goto fail;
+ }
+
+ /*
+ * Initialize the xid.
+ */
+ getmicrotime(&now);
+ crdma->crdma_xid = ((uint32_t)++disrupt) ^ __RPC_GETXID(&now);
+
+ crdma->crdma_prog = prog;
+ crdma->crdma_vers = vers;
+
+ crdma->crdma_waitchan = "rdmarecv";
+ crdma->crdma_waitflag = 0;
+
+ cl->cl_refs = 1;
+ cl->cl_ops = &clnt_rdma_ops;
+ cl->cl_private = crdma;
+ cl->cl_auth = authnone_create();
+
+ crdma->crdma_raw = NULL;
+ crdma->crdma_record = NULL;
+ crdma->crdma_record_resid = 0;
+ TAILQ_INIT(&crdma->crdma_pending);
+ RPCRDMA_DEBUG(1, "clnt_rdma_create: client=%p\n", cl);
+ return (cl);
+fail:
+ RPCRDMA_DEBUG(1, "clnt_rdma_create: failed err=%d\n", error);
+ free(crdma->crdma_xp.ep, M_RPCRDMA);
+ mtx_destroy(&crdma->crdma_lock);
+ mem_free(crdma, sizeof (struct crdma_data));
+ mem_free(cl, sizeof (CLIENT));
+
+ return ((CLIENT *)NULL);
+}
+
+#define RPCRDMA_HEADER_LEN(r, w, c) \
+ (4 * BYTES_PER_XDR_UNIT + \
+ ((r) > 0 ? (r) * 6 * BYTES_PER_XDR_UNIT + BYTES_PER_XDR_UNIT : \
+ BYTES_PER_XDR_UNIT) + \
+ ((w) > 0 ? (w) * 4 * BYTES_PER_XDR_UNIT + 3 * BYTES_PER_XDR_UNIT : \
+ BYTES_PER_XDR_UNIT) + \
+ ((c) > 0 ? (c) * 4 * BYTES_PER_XDR_UNIT + 2 * BYTES_PER_XDR_UNIT : \
+ BYTES_PER_XDR_UNIT))
+
+#define READLIST 0
+#define WRITELIST 1
+#define WRITECHUNK 2
+
+/*
+ * Create an M_EXTPG mbuf list of numpg pages.
+ */
+static struct mbuf *
+_rpc_create_mextpg(int numpg)
+{
+ struct mbuf *m, *mhead, **mprev;
+ int i, j;
+
+ mprev = &mhead;
+ mhead = NULL;
+ for (i = numpg; i > 0; i -= j) {
+ j = MIN(i, MBUF_PEXT_MAX_PGS);
+ m = mb_alloc_ext_plus_pages(j * PAGE_SIZE, M_WAITOK);
+ if (m == NULL) {
+ if (mhead != NULL)
+ m_freem(mhead);
+ return (NULL);
+ }
+ m->m_epg_last_len = PAGE_SIZE;
+ m->m_len = j * PAGE_SIZE;
+ *mprev = m;
+ mprev = &m->m_next;
+ }
+ return (mhead);
+}
+
+/*
+ * Trim the tail of an M_EXTPG mbuf chain created by _rpc_create_mextpg().
+ */
+static void
+_rpc_trim_mextpg(struct mbuf *mextpg, uint32_t newlen)
+{
+ struct mbuf *m, *m2;
+ vm_page_t pg;
+ uint32_t i, j;
+
+ i = 0;
+ for (m = mextpg; m != NULL; m = m->m_next) {
+ /* At last mbuf. */
+ if (i + m->m_len > newlen)
+ break;
+ i += m->m_len;
+ }
+ if (m == NULL)
+ return;
+
+ /* Throw away any additional mbufs. */
+ m2 = m->m_next;
+ m->m_next = NULL;
+ if (m2 != NULL)
+ m_freem(m2);
+
+ /* Now trim the last mbuf. */
+ newlen -= i;
+ i = howmany(newlen, PAGE_SIZE);
+ for (j = i; j < m->m_epg_npgs; j++) {
+ pg = PHYS_TO_VM_PAGE(m->m_epg_pa[j]);
+ vm_page_unwire_noq(pg);
+ vm_page_free(pg);
+ }
+ m->m_epg_npgs = i;
+ m->m_len = newlen;
+ m->m_epg_last_len = newlen % PAGE_SIZE;
+ if (m->m_epg_last_len == 0)
+ m->m_epg_last_len = PAGE_SIZE;
+}
+
+/*
+ * Some M_PROTOnn flags are used in "args" for RDMA chunks.
+ * M_PROTO11 - Indicates that the RPC gets reduced.
+ * (Read/Write/Symlink for NFS.)
+ * M_PROTO10 - On an mbuf in the list (marked by a M_PROTO11) that has the
+ * reduction information, such as the pages. This mbuf is removed
+ * from the list before being sent.
+ * M_PROTO9 - This RPC needs a large rdma_reply write chunk.
+ * M_PROTO8 - This RPC needs a small sized rdma_reply write chunk.
+ * (Readdir for NFS.)
+ * M_PROTO7 - This RPC needs a single page rdma_reply write chunk.
+ * (Readlink for NFS.)
+ * M_PROTO6 - Set in reply to indicate reduction was done.
+ * Except for M_PROTO10, these flags are always set in the first mbuf in "args".
+ *
+ * At this time, only one read or write chunk is provided for an RPC,
+ * which implies only one reduction and no rdma_reply write chunk for
+ * RPCs that have a reduction.
+ */
+static enum clnt_stat
+clnt_rdma_call(
+ CLIENT *cl, /* client handle */
+ struct rpc_callextra *ext, /* call metadata */
+ rpcproc_t proc, /* procedure number */
+ struct mbuf *args, /* pointer to args */
+ struct mbuf **resultsp, /* pointer to results */
+ struct timeval utimeout)
+{
+ struct crdma_data *crdma = (struct crdma_data *) cl->cl_private;
+ AUTH *auth;
+ struct rpc_err *errp;
+ enum clnt_stat stat;
+ XDR xdrs;
+ struct rpc_msg call_msg;
+ struct rpc_msg reply_msg;
+ uint32_t chunks[3];
+ bool_t bval, falseval, ok;
+ int nrefreshes = 2; /* number of times to refresh cred */
+ struct timeval timeout;
+ uint32_t xid, rdma_xid, rdma_vers, rdma_credit, rdma_body, rpos;
+ uint32_t tlen, tlen2, length, handle;
+ uint64_t offset;
+ struct mbuf *m1, *m2, **mreduce_prev, *mreq, *results, *mr, *mreduce;
+ struct ct_request *cr;
+ struct rpcrdma_chunk *reduce_chp, *reply_chp, *request_chp;
+ int error, i, ind, reduce_ind;
+
+ cr = malloc(sizeof(struct ct_request), M_RPC, M_WAITOK);
+
+ ind = -1;
+ reduce_chp = NULL;
+ reply_chp = NULL;
+ request_chp = NULL;
+ rpos = 0;
+ reduce_ind = 0;
+ mreq = NULL;
+ mr = NULL;
+ mreduce = NULL;
+ chunks[READLIST] = chunks[WRITELIST] = chunks[WRITECHUNK] = 0;
+ mtx_lock(&crdma->crdma_lock);
+
+ if (crdma->crdma_closing || crdma->crdma_closed ||
+ xprt_rdma_disconnected(&crdma->crdma_xp) != 0) {
+ mtx_unlock(&crdma->crdma_lock);
+ free(cr, M_RPC);
+ RPCRDMA_DEBUG(2, "clnt_rdma_call: First closing, return "
+ "RPC_CANTSEND\n");
+ return (RPC_CANTSEND);
+ }
+ crdma->crdma_threads++;
+
+ if (ext) {
+ auth = ext->rc_auth;
+ errp = &ext->rc_err;
+ } else {
+ auth = cl->cl_auth;
+ errp = &crdma->crdma_error;
+ }
+
+ cr->cr_mrep = NULL;
+ cr->cr_error = 0;
+
+ if (crdma->crdma_wait.tv_usec == -1) {
+ timeout = utimeout; /* use supplied timeout */
+ } else {
+ timeout = crdma->crdma_wait; /* use default timeout */
+ }
+
+ /*
+ * After 15sec of looping, allow it to return RPC_CANTSEND, which will
+ * cause the clnt_reconnect layer to create a new RDMA connection.
+ */
+call_again:
+ mtx_assert(&crdma->crdma_lock, MA_OWNED);
+ if (crdma->crdma_closing || crdma->crdma_closed) {
+closing:
+ crdma->crdma_threads--;
+ wakeup(crdma);
+ mtx_unlock(&crdma->crdma_lock);
+ if (ind >= 0)
+ xprt_rdma_release_send(&crdma->crdma_xp, ind,
+ reduce_chp, reply_chp, request_chp);
+ free(cr, M_RPC);
+ RPCRDMA_DEBUG(2, "clnt_rdma_call: Second closing, return "
+ "RPC_CANTSEND\n");
+ return (RPC_CANTSEND);
+ }
+
+ crdma->crdma_xid++;
+ xid = crdma->crdma_xid;
+
+ if (ind < 0)
+ ind = xprt_rdma_acquire_buf(&crdma->crdma_xp, 0,
+ crdma->crdma_xp.maxrpc);
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: buffer xid=0x%x ind=%d maxrpc=%d\n",
+ xid, ind, crdma->crdma_xp.maxrpc);
+ if (ind < 0)
+ goto closing;
+ mtx_unlock(&crdma->crdma_lock);
+
+ if ((args->m_flags & M_PROTO9) != 0 && reply_chp == NULL) {
+ /* Create a large rdma_reply chunk, as required. */
+ tlen = howmany(crdma->crdma_max_io, PAGE_SIZE);
+ m1 = _rpc_create_mextpg(tlen);
+ reply_chp = xprt_rdma_create_chunk(&crdma->crdma_xp, tlen,
+ NULL, m1, true, ind);
+ if (reply_chp != NULL)
+ chunks[WRITECHUNK] = reply_chp->num_segment;
+ else
+ printf("clnt_rdma_call: Couldn't create large reply "
+ "chunk\n");
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: Large (M_PROTO9) reply chunk"
+ " reply_chp=%p num_pg=%d\n", reply_chp, tlen);
+ } else if ((args->m_flags & M_PROTO8) != 0 && reply_chp == NULL) {
+ /* Create a small (Readdir) sized rdma_reply chunk. */
+ tlen = howmany(crdma->crdma_small_reply, PAGE_SIZE);
+ m1 = _rpc_create_mextpg(tlen);
+ reply_chp = xprt_rdma_create_chunk(&crdma->crdma_xp, tlen,
+ NULL, m1, true, ind);
+ if (reply_chp != NULL)
+ chunks[WRITECHUNK] = reply_chp->num_segment;
+ else
+ printf("clnt_rdma_call: Couldn't create small reply "
+ "chunk\n");
+
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: Small (M_PROTO8) reply chunk"
+ " reply_chp=%p num_pg=%d\n", reply_chp, tlen);
+ } else if ((args->m_flags & M_PROTO7) != 0 && reply_chp == NULL) {
+ /* Create a single page sized rdma_reply chunk. (Readlink) */
+ m1 = _rpc_create_mextpg(1);
+ reply_chp = xprt_rdma_create_chunk(&crdma->crdma_xp, 1,
+ NULL, m1, true, ind);
+ if (reply_chp != NULL)
+ chunks[WRITECHUNK] = reply_chp->num_segment;
+ else
+ printf("clnt_rdma_call: Couldn't create 1 page reply "
+ "chunk\n");
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: Single page (M_PROTO7) reply"
+ " chunk reply_chp=%p num_pg=%d\n", reply_chp, tlen);
+ }
+ if ((args->m_flags & M_PROTO11) != 0 && reduce_chp == NULL) {
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: M_PROTO11 to be reduced\n");
+ /*
+ * This RPC needs to be reduced. The M_PROTO10 mbuf provides
+ * the information to create the chunk for reduction.
+ */
+ mreduce_prev = &args;
+ for (mreduce = args; mreduce != NULL &&
+ (mreduce->m_flags & M_PROTO10) == 0;
+ mreduce = mreduce->m_next) {
+ rpos += mreduce->m_len;
+ mreduce_prev = &mreduce->m_next;
+ }
+ if (mreduce != NULL) {
+ struct rpcrdma_reduce_pg *rb;
+
+ rb = mtod(mreduce, struct rpcrdma_reduce_pg *);
+ reduce_chp = xprt_rdma_create_chunk(&crdma->crdma_xp,
+ 0, rb, NULL, (rb->into_mem != 0) ? true : false,
+ ind);
+ if (reduce_chp != NULL) {
+ if (rb->into_mem != 0)
+ reduce_ind = WRITELIST;
+ else
+ reduce_ind = READLIST;
+ chunks[reduce_ind] = reduce_chp->num_segment;
+ tlen = rb->len;
+ rpos = rb->pos;
+ }
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: reduction pos=%d "
+ "len=%d into_mem=%d reduce_chp=%p\n", rpos, tlen,
+ rb->into_mem, reduce_chp);
+ } else
+ printf("clnt_rdma_call: Bogus M_PROTO10\n");
+ }
+
+ rdma_vers = 1;
+ rdma_credit = crdma->crdma_xp.maxrpc;
+ rdma_body = RDMA_MSG;
+
+ /*
+ * Create the RPC header now in m2, so we can size it and adjust the
+ * rdma_readlist positions.
+ */
+ m2 = m_get(M_WAITOK, MT_DATA);
+ m2->m_len = 0;
+ xdrmbuf_create(&xdrs, m2, XDR_ENCODE);
+ /*
+ * Put the RPC header in the XDR stream.
+ */
+ call_msg.rm_xid = xid;
+ call_msg.rm_direction = CALL;
+ call_msg.rm_call.cb_rpcvers = RPC_MSG_VERSION;
+ call_msg.rm_call.cb_prog = crdma->crdma_prog;
+ call_msg.rm_call.cb_vers = crdma->crdma_vers;
+ if (! xdr_callhdr(&xdrs, &call_msg))
+ goto cannotencode;
+ m1 = m_copym(args, 0, M_COPYALL, M_WAITOK);
+ if (m1 == NULL)
+ goto cannotencode;
+ m1->m_flags &= ~(M_PROTO7 | M_PROTO8 | M_PROTO9 | M_PROTO11);
+ if (mreduce != NULL) {
+ /* Get rid of the M_PROTO10 mbuf in the copy. */
+ mreduce_prev = &args;
+ for (mreduce = args; mreduce != NULL &&
+ (mreduce->m_flags & M_PROTO10) == 0;
+ mreduce = mreduce->m_next) {
+ rpos += mreduce->m_len;
+ mreduce_prev = &mreduce->m_next;
+ }
+ if (mreduce != NULL) {
+ *mreduce_prev = m2->m_next;
+ m2->m_next = NULL;
+ } else
+ printf("clnt_rdma_call: Cannot find M_PROTO10 mbuf\n");
+ }
+
+ if ((! XDR_PUTINT32(&xdrs, &proc)) ||
+ (! AUTH_MARSHALL(auth, xid, &xdrs, m1))) {
+cannotencode:
+ if (m2)
+ m_freem(m2);
+ if (mreq)
+ m_freem(mreq);
+ errp->re_status = stat = RPC_CANTENCODEARGS;
+ RPCRDMA_DEBUG(1, "xprt_rdma_cal: cannot decode args\n");
+ mtx_lock(&crdma->crdma_lock);
+ goto out;
+ }
+ tlen2 = m_length(m2, NULL);
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: RPC header length=%d\n", tlen2);
+ tlen2 += RPCRDMA_HEADER_LEN(chunks[READLIST], chunks[WRITELIST],
+ chunks[WRITECHUNK]);
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: RPC header plus RDMA header "
+ "length=%d\n", tlen2);
+
+ /*
+ * If the RPC request message is too large
+ * create a large enough buffer and copy the mbuf
+ * list into it.
+ */
+ if (rdma_body == RDMA_MSG && tlen2 > RPCRDMA_MAX_INLINE) {
+ struct rpcrdma_reduce_pg *rb;
+
+ KASSERT(request_chp == NULL, ("clnt_rdma_call: request_chp not "
+ "NULL"));
+ tlen2 -= RPCRDMA_HEADER_LEN(chunks[READLIST], chunks[WRITELIST],
+ chunks[WRITECHUNK]);
+ if (reduce_chp != NULL) {
+ printf("clnt_rdma_call: Request msg too big with chunk"
+ " %d\n", tlen2);
+ goto cannotencode;
+ }
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: Large request len=%d\n",
+ tlen2);
+ mr = rpc_reduce_pg(tlen2, 0, false);
+ rb = mtod(mr, struct rpcrdma_reduce_pg *);
+ request_chp = xprt_rdma_create_chunk(&crdma->crdma_xp,
+ 0, rb, NULL, false, ind);
+ if (request_chp != NULL) {
+ rpc_copy_mbuf_to_rb(m2, rb);
+ m_freem(m2);
+ m2 = NULL;
+ chunks[READLIST] = request_chp->num_segment;
+ rdma_body = RDMA_NOMSG;
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: Got request_chp, "
+ "num_segment=%d\n", request_chp->num_segment);
+ } else
+ goto cannotencode;
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: Creating large RPC request "
+ "chunk length=%d request_chp=%p\n", tlen2, request_chp);
+ }
+
+ mreq = m_gethdr(M_WAITOK, MT_DATA);
+ mreq->m_len = 0;
+ xdrmbuf_create(&xdrs, mreq, XDR_ENCODE);
+
+ errp->re_status = stat = RPC_SUCCESS;
+
+ /*
+ * Put the RDMA header in the XDR stream.
+ */
+ falseval = FALSE;
+ bval = (chunks[READLIST] > 0) ? TRUE : FALSE;
+ if ((! XDR_PUTINT32(&xdrs, &xid)) ||
+ (! XDR_PUTINT32(&xdrs, &rdma_vers)) ||
+ (! XDR_PUTINT32(&xdrs, &rdma_credit)) ||
+ (! XDR_PUTINT32(&xdrs, &rdma_body)) ||
+ (! XDR_PUTINT32(&xdrs, &bval)))
+ goto cannotencode;
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: xid=0x%x vers=%d credits=%d "
+ "rdma_body=%d readlist=%d\n", xid, rdma_vers, rdma_credit,
+ rdma_body, bval);
+ if (bval) {
+ struct rpcrdma_chunk *tchp;
+
+ KASSERT(rpos > 0 || rdma_body == RDMA_NOMSG,
+ ("clnt_rdma_call: rpos zero for RDMA_MSG"));
+ rpos += tlen2;
+ tchp = reduce_chp;
+ if (rdma_body == RDMA_NOMSG) {
+ rpos = 0; /* 0 for a long call message. */
+ tchp = request_chp;
+ }
+ bval = TRUE;
+ for (i = 0; i < chunks[READLIST]; i++) {
+ if (i == chunks[READLIST] - 1)
+ bval = FALSE;
+ if ((! XDR_PUTINT32(&xdrs, &rpos)) ||
+ (! XDR_PUTINT32(&xdrs, &tchp->handle[i])) ||
+ (! XDR_PUTINT32(&xdrs, &tchp->length[i])) ||
+ (! xdr_uint64_t(&xdrs, &tchp->offset[i])) ||
+ (! XDR_PUTINT32(&xdrs, &bval)))
+ goto cannotencode;
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: readlist chunk "
+ "handle=0x%x length=%d offset=%ju numseg=%d\n",
+ tchp->handle[i], tchp->length[i],
+ (uintmax_t)tchp->offset[i], chunks[READLIST]);
+ }
+ RPCRDMA_DEBUG(2, "clnt_rdma_call: Readlist rdma_body=%d "
+ "numseg=%d\n", rdma_body, chunks[READLIST]);
+ }
+ bval = (chunks[WRITELIST] > 0) ? TRUE : FALSE;
+ if (bval) {
+ if ((! XDR_PUTINT32(&xdrs, &bval)) ||
+ (! XDR_PUTINT32(&xdrs, &chunks[WRITELIST])))
+ goto cannotencode;
+ for (i = 0; i < chunks[WRITELIST]; i++) {
+ if ((! XDR_PUTINT32(&xdrs, &reduce_chp->handle[i])) ||
+ (! XDR_PUTINT32(&xdrs, &reduce_chp->length[i])) ||
+ (! xdr_uint64_t(&xdrs, &reduce_chp->offset[i])))
+ goto cannotencode;
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: writelist chunk "
+ "handle=0x%x length=%d offset=%ju numseg=%d\n",
+ reduce_chp->handle[i], reduce_chp->length[i],
+ (uintmax_t)reduce_chp->offset[i],
+ chunks[WRITELIST]);
+ }
+ if (! XDR_PUTINT32(&xdrs, &falseval))
+ goto cannotencode;
+ RPCRDMA_DEBUG(2, "clnt_rdma_call: Writelist numseg=%d\n",
+ chunks[WRITELIST]);
+ } else {
+ if (! XDR_PUTINT32(&xdrs, &bval))
+ goto cannotencode;
+ }
+ bval = (chunks[WRITECHUNK] > 0) ? TRUE : FALSE;
+ if (bval) {
+ if ((! XDR_PUTINT32(&xdrs, &bval)) ||
+ (! XDR_PUTINT32(&xdrs, &chunks[WRITECHUNK])))
+ goto cannotencode;
+ for (i = 0; i < chunks[WRITECHUNK]; i++) {
+ if ((! XDR_PUTINT32(&xdrs, &reply_chp->handle[i])) ||
+ (! XDR_PUTINT32(&xdrs, &reply_chp->length[i])) ||
+ (! xdr_uint64_t(&xdrs, &reply_chp->offset[i])))
+ goto cannotencode;
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: replylist chunk "
+ "handle=0x%x length=%d offset=%ju numseg=%d\n",
+ reply_chp->handle[i], reply_chp->length[i],
+ (uintmax_t)reply_chp->offset[i],
+ chunks[WRITECHUNK]);
+ }
+ RPCRDMA_DEBUG(2, "clnt_rdma_call: Replylist numseg=%d\n",
+ chunks[WRITECHUNK]);
+ } else {
+ if (! XDR_PUTINT32(&xdrs, &bval))
+ goto cannotencode;
+ }
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: RDMA header len=%d rdma_body=%d\n",
+ m_length(mreq, NULL), rdma_body);
+ if (rdma_body == RDMA_MSG) {
+ if (! XDR_PUTMBUF(&xdrs, m2))
+ goto cannotencode;
+ m2 = NULL;
+ }
+ mreq->m_pkthdr.len = m_length(mreq, NULL);
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: RDMA header plus RPC message "
+ "len=%d\n", mreq->m_pkthdr.len);
+
+ cr->cr_xid = xid;
+ mtx_lock(&crdma->crdma_lock);
+ /*
+ * Check to see if the other end has already started to close down
+ * the connection. The upcall will have set crdma_error.re_status
+ * to RPC_CANTRECV if this is the case.
+ * If the other end starts to close down the connection after this
+ * point, it will be detected later when cr_error is checked,
+ * since the request is in the crdma_pending queue.
+ */
+ if (crdma->crdma_error.re_status == RPC_CANTRECV) {
+ if (errp != &crdma->crdma_error) {
+ errp->re_errno = crdma->crdma_error.re_errno;
+ errp->re_status = RPC_CANTRECV;
+ }
+ stat = RPC_CANTRECV;
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: return RPC_CANTRECV\n");
+ goto out;
+ }
+
+ TAILQ_INSERT_TAIL(&crdma->crdma_pending, cr, cr_link);
+ mtx_unlock(&crdma->crdma_lock);
+
+ /*
+ * Send the message on the RDMA payload stream.
+ */
+ error = xprt_rdma_send(&crdma->crdma_xp, mreq, ind);
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: xprt_rdma_send index=%d error=%d\n",
+ ind, error);
+ mreq = NULL;
+ reply_msg.acpted_rply.ar_verf.oa_flavor = AUTH_NULL;
+ reply_msg.acpted_rply.ar_verf.oa_base = cr->cr_verf;
+ reply_msg.acpted_rply.ar_verf.oa_length = 0;
+ reply_msg.acpted_rply.ar_results.where = NULL;
+ reply_msg.acpted_rply.ar_results.proc = (xdrproc_t)xdr_void;
+
+ mtx_lock(&crdma->crdma_lock);
+ if (error) {
+ TAILQ_REMOVE(&crdma->crdma_pending, cr, cr_link);
+ errp->re_errno = error;
+ errp->re_status = stat = RPC_CANTSEND;
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: Third RPC_CANTSEND\n");
+ goto out;
+ }
+
+ /*
+ * Check to see if we got an upcall while waiting for the
+ * lock. In both these cases, the request has been removed
+ * from crdma->crdma_pending.
+ */
+ if (cr->cr_error) {
+ TAILQ_REMOVE(&crdma->crdma_pending, cr, cr_link);
+ errp->re_errno = cr->cr_error;
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: First RPC_CANTRECV\n");
+ errp->re_status = stat = RPC_CANTRECV;
+ goto out;
+ }
+ if (cr->cr_mrep) {
+ TAILQ_REMOVE(&crdma->crdma_pending, cr, cr_link);
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: Got reply to RPC\n");
+ goto got_reply;
+ }
+
+ /*
+ * Hack to provide rpc-based message passing
+ */
+ if (timeout.tv_sec == 0 && timeout.tv_usec == 0) {
+ TAILQ_REMOVE(&crdma->crdma_pending, cr, cr_link);
+ errp->re_status = stat = RPC_TIMEDOUT;
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: Weird zero timeout\n");
+ goto out;
+ }
+
+ error = msleep(cr, &crdma->crdma_lock, crdma->crdma_waitflag,
+ crdma->crdma_waitchan, tvtohz(&timeout));
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: msleep=%d\n", error);
+
+ TAILQ_REMOVE(&crdma->crdma_pending, cr, cr_link);
+
+ if (error) {
+ /*
+ * The sleep returned an error so our request is still
+ * on the list. Turn the error code into an
+ * appropriate client status.
+ */
+ errp->re_errno = error;
+ switch (error) {
+ case EINTR:
+ stat = RPC_INTR;
+ break;
+ case EWOULDBLOCK:
+ stat = RPC_TIMEDOUT;
+ break;
+ default:
+ stat = RPC_CANTRECV;
+ }
+ errp->re_status = stat;
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: msleep error return "
+ "stat=%d\n", stat);
+ goto out;
+ } else {
+ /*
+ * We were woken up by the upcall. If the
+ * upcall had a receive error, report that,
+ * otherwise we have a reply.
+ */
+ if (cr->cr_error) {
+ errp->re_errno = cr->cr_error;
+ errp->re_status = stat = RPC_CANTRECV;
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: Upcall recv=%d, "
+ "second RPC_CANTRECV\n", errp->re_errno);
+ goto out;
+ }
+ }
+
+got_reply:
+ /*
+ * Now decode and validate the response. We need to drop the
+ * lock since xdr_replymsg may end up sleeping in malloc.
+ */
+ mtx_unlock(&crdma->crdma_lock);
+
+ if (ext && ext->rc_feedback)
+ ext->rc_feedback(FEEDBACK_OK, proc, ext->rc_feedback_arg);
+
+ xdrmbuf_create(&xdrs, cr->cr_mrep, XDR_DECODE);
+ cr->cr_mrep = NULL;
+
+ /* Decode the RDMA header. */
+ if ((! XDR_GETINT32(&xdrs, &rdma_xid)) ||
+ (! XDR_GETINT32(&xdrs, &rdma_vers)) ||
+ (! XDR_GETINT32(&xdrs, &rdma_credit)) ||
+ (! XDR_GETINT32(&xdrs, &rdma_body)))
+ goto cannotdecode;
+ RPCRDMA_DEBUG(2, "clnt_rdma_call: Got reply RDMA header xid=0x%x "
+ "vers=%d rdma_credit=%d rdma_body=%d\n", rdma_xid, rdma_vers,
+ rdma_credit, rdma_body);
+ if (xid != rdma_xid || rdma_vers != 1) {
+ printf("clnt_rdma_call: xid=%u rdma_xid=%u rdma_vers=%d\n",
+ xid, rdma_xid, rdma_vers);
+ goto cannotdecode;
+ }
+ crdma->crdma_xp.credits = rdma_credit;
+ if (rdma_body == RDMA_MSG || rdma_body == RDMA_NOMSG) {
+ /* Check the 3 chunks in the reply. */
+ for (i = 0; i < 3; i++) {
+ tlen = 0;
+ if (! XDR_GETINT32(&xdrs, &bval))
+ goto cannotdecode;
+ if (bval) {
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: Got chunk "
+ "in reply, type=%d\n", i);
+ if (chunks[i] == 0) {
+ printf("clnt_rdma_call: chunks%d not "
+ "set\n", i);
+ goto cannotdecode;
+ }
+ /*
+ * Set chunks[i] to 0 to indicate it was used.
+ */
+ chunks[i] = 0;
+
+ rpos = 0;
+ /*
+ * For i > 0, rpos is the count of segments
+ * in the xdr_write_chunk.
+ */
+ if (i > 0) {
+ if (! XDR_GETINT32(&xdrs, &rpos)) {
+ printf("clnt_rdma_call: No "
+ "chunk cnt\n");
+ goto cannotdecode;
+ }
+ if (rpos < 1 ||
+ rpos > RPCRDMA_MAX_SEGMENTS) {
+ printf("clnt_rdma_call: Bad "
+ "write chunk cnt\n");
+ goto cannotdecode;
+ }
+ }
+ while (bval && ((rpos == 0 && i == 0) ||
+ (rpos > 0 && i > 0))) {
+ if (i == 0 &&
+ (! XDR_GETINT32(&xdrs, &rpos))) {
+ printf("clnt_rdma_call: No "
+ "chunk pos\n");
+ goto cannotdecode;
+ }
+ if ((! XDR_GETINT32(&xdrs, &handle)) ||
+ (! XDR_GETINT32(&xdrs, &length)) ||
+ (! xdr_int64_t(&xdrs, &offset))) {
+ printf("clnt_rdma_call: chunk "
+ "bogus\n");
+ goto cannotdecode;
+ }
+ tlen += length;
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: Got "
+ "segment handle=0x%x length=%d "
+ "offset=%ju\n", handle, length,
+ (uintmax_t)offset);
+ if (i > 0)
+ rpos--;
+ if (i == 0 || (i == 1 && rpos == 0)) {
+ if (!XDR_GETINT32(&xdrs, &bval))
+ goto cannotdecode;
+ }
+ }
+ if (reduce_chp == NULL && reply_chp == NULL) {
+ printf("clnt_rdma_call: No chunk\n");
+ goto cannotdecode;
+ }
+ }
+ }
+ if (rdma_body == RDMA_NOMSG) {
+ struct mbuf *mextpg;
+
+ if (reply_chp == NULL || tlen == 0) {
+ printf("clnt_rdma_call: Bad reply chunk chp=%p"
+ " len=%d\n", reply_chp, tlen);
+ goto cannotdecode;
+ }
+ RPCRDMA_DEBUG(2, "clnt_rdma_call: Got RDMA_NOMSG for "
+ "large reply, length=%d\n", tlen);
+ mextpg = reply_chp->mextpg;
+ xprt_rdma_unmap_chunk(&crdma->crdma_xp, reply_chp);
+ reply_chp = NULL;
+ _rpc_trim_mextpg(mextpg, tlen);
+
+ /*
+ * Until nfsm_dissect() and friends know how to handle
+ * an M_EXTPG mbuf list, we need a cluster copy.
+ */
+ error = mb_unmapped_to_ext(mextpg, &m1);
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: mb_unmapped_to_ext"
+ " error=%d\n", error);
+ if (error != 0)
+ goto cannotdecode;
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: length=%d mbuf "
+ "len=%d\n", tlen, m_length(m1, NULL));
+
+ XDR_DESTROY(&xdrs);
+ xdrmbuf_create(&xdrs, m1, XDR_DECODE);
+ }
+ } else if (rdma_body == RDMA_ERROR) {
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: Got RDMA_ERROR reply\n");
+ if ((! XDR_GETINT32(&xdrs, &rpos)))
+ goto cannotdecode;
+ if (rpos == RDMA_ERR_VERS) {
+ if ((! XDR_GETINT32(&xdrs, &rpos)) ||
+ (! XDR_GETINT32(&xdrs, &rpos)))
+ goto cannotdecode;
+ printf("clnt_rdma_call: Wrong RDMA verion\n");
+ goto cannotdecode;
+ } else if (rpos == RDMA_ERR_CHUNK) {
+ printf("clnt_rdma_call: ERR_CHUNK\n");
+ clnt_rdma_upcall(&crdma->crdma_xp, NULL);
+ goto cannotdecode;
+ }
+ }
+
+ ok = xdr_replymsg(&xdrs, &reply_msg);
+ if (ok) {
+ if ((reply_msg.rm_reply.rp_stat == MSG_ACCEPTED) &&
+ (reply_msg.acpted_rply.ar_stat == SUCCESS))
+ errp->re_status = stat = RPC_SUCCESS;
+ else
+ stat = _seterr_reply(&reply_msg, errp);
+
+ if (stat == RPC_SUCCESS) {
+ results = xdrmbuf_getall(&xdrs);
+ if (!AUTH_VALIDATE(auth, xid,
+ &reply_msg.acpted_rply.ar_verf,
+ &results)) {
+ errp->re_status = stat = RPC_AUTHERROR;
+ errp->re_why = AUTH_INVALIDRESP;
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: Auth err, "
+ "invalid response\n");
+ } else {
+ KASSERT(results,
+ ("auth validated but no result"));
+ if (reduce_chp != NULL &&
+ chunks[reduce_ind] == 0)
+ results->m_flags |= M_PROTO6;
+ *resultsp = results;
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: Got RPC reply"
+ " (NFS) : reduce_chp=%p reduced=%d\n",
+ reduce_chp, chunks[reduce_ind]);
+ }
+ } /* end successful completion */
+ /*
+ * If unsuccessful AND error is an authentication error
+ * then refresh credentials and try again, else break
+ */
+ else if (stat == RPC_AUTHERROR)
+ /* maybe our credentials need to be refreshed ... */
+ if (nrefreshes > 0 &&
+ AUTH_REFRESH(auth, &reply_msg)) {
+ nrefreshes--;
+ XDR_DESTROY(&xdrs);
+ if (request_chp != NULL) {
+ KASSERT(mr != NULL, ("clnt_rdma_call: "
+ "request_chp, but no mr"));
+ xprt_rdma_unmap_chunk(&crdma->crdma_xp,
+ request_chp);
+ request_chp = NULL;
+ rpc_free_rdma_reduction(mr);
+ }
+ RPCRDMA_DEBUG(4, "clnt_rdma_call: Auth retry "
+ "xid=0x%x\n", xid);
+ mtx_lock(&crdma->crdma_lock);
+ goto call_again;
+ }
+ /* end of unsuccessful completion */
+ } /* end of valid reply message */
+ else {
+cannotdecode:
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: RPC_CANTDECODERES\n");
+ errp->re_status = stat = RPC_CANTDECODERES;
+ }
+ /* Done with the Send/Receive buffer, so release it. */
+ xprt_rdma_release_send(&crdma->crdma_xp, ind, reduce_chp, reply_chp,
+ request_chp);
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: xprt_rdma_release send done\n");
+ ind = -1;
+ XDR_DESTROY(&xdrs);
+ mtx_lock(&crdma->crdma_lock);
+out:
+ mtx_assert(&crdma->crdma_lock, MA_OWNED);
+
+ KASSERT(stat != RPC_SUCCESS || *resultsp,
+ ("RPC_SUCCESS without reply"));
+
+ if (mreq)
+ m_freem(mreq);
+ if (cr->cr_mrep)
+ m_freem(cr->cr_mrep);
+
+ crdma->crdma_threads--;
+ if (crdma->crdma_closing)
+ wakeup(crdma);
+
+ mtx_unlock(&crdma->crdma_lock);
+
+ if (mr)
+ rpc_free_rdma_reduction(mr);
+ if (ind >= 0)
+ xprt_rdma_release_send(&crdma->crdma_xp, ind, reduce_chp,
+ reply_chp, request_chp);
+
+ if (auth && stat != RPC_SUCCESS)
+ AUTH_VALIDATE(auth, xid, NULL, NULL);
+
+ free(cr, M_RPC);
+
+ RPCRDMA_DEBUG(1, "clnt_rdma_call: Returning stat=%d\n", stat);
+ return (stat);
+}
+
+static void
+clnt_rdma_geterr(CLIENT *cl, struct rpc_err *errp)
+{
+ struct crdma_data *crdma = (struct crdma_data *) cl->cl_private;
+
+ *errp = crdma->crdma_error;
+}
+
+static bool_t
+clnt_rdma_freeres(CLIENT *cl, xdrproc_t xdr_res, void *res_ptr)
+{
+ XDR xdrs;
+ bool_t dummy;
+
+ xdrs.x_op = XDR_FREE;
+ dummy = (*xdr_res)(&xdrs, res_ptr);
+
+ return (dummy);
+}
+
+/*ARGSUSED*/
+static void
+clnt_rdma_abort(CLIENT *cl)
+{
+}
+
+static bool_t
+clnt_rdma_control(CLIENT *cl, u_int request, void *info)
+{
+ struct crdma_data *crdma = (struct crdma_data *)cl->cl_private;
+ void *infop = info;
+ uint32_t max_io;
+ SVCXPRT *xprt;
+
+ mtx_lock(&crdma->crdma_lock);
+
+ switch (request) {
+ case CLSET_FD_CLOSE:
+ crdma->crdma_closeit = TRUE;
+ mtx_unlock(&crdma->crdma_lock);
+ return (TRUE);
+ case CLSET_FD_NCLOSE:
+ crdma->crdma_closeit = FALSE;
+ mtx_unlock(&crdma->crdma_lock);
+ return (TRUE);
+ default:
+ break;
+ }
+
+ /* for other requests which use info */
+ if (info == NULL) {
+ mtx_unlock(&crdma->crdma_lock);
+ return (FALSE);
+ }
+ switch (request) {
+ case CLSET_TIMEOUT:
+ if (time_not_ok((struct timeval *)info)) {
+ mtx_unlock(&crdma->crdma_lock);
+ return (FALSE);
+ }
+ crdma->crdma_wait = *(struct timeval *)infop;
+ break;
+ case CLGET_TIMEOUT:
+ *(struct timeval *)infop = crdma->crdma_wait;
+ break;
+ case CLGET_SERVER_ADDR:
+ memcpy(info, &crdma->crdma_addr, crdma->crdma_addr.ss_len);
+ break;
+ case CLGET_SVC_ADDR:
+ /*
+ * Slightly different semantics to userland - we use
+ * sockaddr instead of netbuf.
+ */
+ memcpy(info, &crdma->crdma_addr, crdma->crdma_addr.ss_len);
+ break;
+ case CLSET_SVC_ADDR: /* set to new address */
+ mtx_unlock(&crdma->crdma_lock);
+ return (FALSE);
+ case CLGET_XID:
+ *(uint32_t *)info = crdma->crdma_xid;
+ break;
+ case CLSET_XID:
+ /* This will set the xid of the NEXT call */
+ /* decrement by 1 as clnt_rdma_call() increments once */
+ crdma->crdma_xid = *(uint32_t *)info - 1;
+ break;
+ case CLGET_VERS:
+ /*
+ * This RELIES on the information that, in the call body,
+ * the version number field is the fifth field from the
+ * beginning of the RPC header. MUST be changed if the
+ * call_struct is changed
+ */
+ *(uint32_t *)info =
+ ntohl(*(uint32_t *)(void *)(crdma->crdma_mcallc +
+ 4 * BYTES_PER_XDR_UNIT));
+ break;
+ case CLSET_VERS:
+ *(uint32_t *)(void *)(crdma->crdma_mcallc +
+ 4 * BYTES_PER_XDR_UNIT) =
+ htonl(*(uint32_t *)info);
+ break;
+ case CLGET_PROG:
+ /*
+ * This RELIES on the information that, in the call body,
+ * the program number field is the fourth field from the
+ * beginning of the RPC header. MUST be changed if the
+ * call_struct is changed
+ */
+ *(uint32_t *)info =
+ ntohl(*(uint32_t *)(void *)(crdma->crdma_mcallc +
+ 3 * BYTES_PER_XDR_UNIT));
+ break;
+ case CLSET_PROG:
+ *(uint32_t *)(void *)(crdma->crdma_mcallc +
+ 3 * BYTES_PER_XDR_UNIT) =
+ htonl(*(uint32_t *)info);
+ break;
+ case CLSET_WAITCHAN:
+ crdma->crdma_waitchan = (const char *)info;
+ break;
+ case CLGET_WAITCHAN:
+ *(const char **) info = crdma->crdma_waitchan;
+ break;
+ case CLSET_INTERRUPTIBLE:
+ if (*(int *) info)
+ crdma->crdma_waitflag = PCATCH;
+ else
+ crdma->crdma_waitflag = 0;
+ break;
+ case CLGET_INTERRUPTIBLE:
+ if (crdma->crdma_waitflag)
+ *(int *) info = TRUE;
+ else
+ *(int *) info = FALSE;
+ break;
+ case CLSET_RDMAMAX_IO:
+ max_io = *(uint32_t *)info;
+ crdma->crdma_max_io = round_page(max_io + 2048);
+ break;
+ case CLGET_RDMAMAX_IO:
+ *(uint32_t *)info = crdma->crdma_xp.maxio;
+ break;
+ case CLSET_BACKCHANNEL:
+ xprt = (SVCXPRT *)info;
+ if (crdma->crdma_backchannelxprt == NULL) {
+ struct cf_conn *cd;
+
+ SVC_ACQUIRE(xprt);
+ xprt->xp_p2 = crdma;
+ cd = (struct cf_conn *)xprt->xp_p1;
+ cd->rdma = TRUE;
+ crdma->crdma_backchannelxprt = xprt;
+ }
+ break;
+ default:
+ mtx_unlock(&crdma->crdma_lock);
+ return (FALSE);
+ }
+
+ mtx_unlock(&crdma->crdma_lock);
+ return (TRUE);
+}
+
+static void
+clnt_rdma_close(CLIENT *cl)
+{
+ struct crdma_data *crdma = (struct crdma_data *) cl->cl_private;
+
+ mtx_lock(&crdma->crdma_lock);
+
+ if (crdma->crdma_closed) {
+ mtx_unlock(&crdma->crdma_lock);
+ return;
+ }
+
+ if (crdma->crdma_closing) {
+ while (crdma->crdma_closing)
+ msleep(crdma, &crdma->crdma_lock, 0, "rdmaclose", 0);
+ KASSERT(crdma->crdma_closed, ("client should be closed"));
+ mtx_unlock(&crdma->crdma_lock);
+ return;
+ }
+
+
+ crdma->crdma_closing = FALSE;
+ crdma->crdma_closed = TRUE;
+ mtx_unlock(&crdma->crdma_lock);
+ RPCRDMA_DEBUG(1, "clnt_rdma_close: Close of connection at "
+ "disconnect\n");
+ xprt_rdma_disconnect(&crdma->crdma_xp);
+ wakeup(crdma);
+}
+
+static void
+clnt_rdma_destroy(CLIENT *cl)
+{
+ struct crdma_data *crdma = (struct crdma_data *) cl->cl_private;
+ SVCXPRT *xprt;
+
+ clnt_rdma_close(cl);
+
+ mtx_lock(&crdma->crdma_lock);
+ xprt = crdma->crdma_backchannelxprt;
+ crdma->crdma_backchannelxprt = NULL;
+ if (xprt != NULL) {
+ mtx_unlock(&crdma->crdma_lock); /* To avoid a LOR. */
+ sx_xlock(&xprt->xp_lock);
+ mtx_lock(&crdma->crdma_lock);
+ xprt->xp_p2 = NULL;
+ sx_xunlock(&xprt->xp_lock);
+ SVC_RELEASE(xprt);
+ }
+ mtx_unlock(&crdma->crdma_lock);
+
+ RPCRDMA_DEBUG(1, "clnt_rdma_destroy: after clnt_rdma_close() call\n");
+ mtx_destroy(&crdma->crdma_lock);
+
+ if (crdma->crdma_xp.ep != NULL)
+ free(crdma->crdma_xp.ep, M_RPCRDMA);
+ m_freem(crdma->crdma_record);
+ m_freem(crdma->crdma_raw);
+ mem_free(crdma, sizeof(struct crdma_data));
+ if (cl->cl_netid && cl->cl_netid[0])
+ mem_free(cl->cl_netid, strlen(cl->cl_netid) +1);
+ if (cl->cl_tp && cl->cl_tp[0])
+ mem_free(cl->cl_tp, strlen(cl->cl_tp) +1);
+ mem_free(cl, sizeof(CLIENT));
+}
+
+/*
+ * Make sure that the time is not garbage. -1 value is disallowed.
+ * Note this is different from time_not_ok in clnt_dg.c
+ */
+static bool_t
+time_not_ok(struct timeval *t)
+{
+ return (t->tv_sec <= -1 || t->tv_sec > 100000000 ||
+ t->tv_usec <= -1 || t->tv_usec > 1000000);
+}
+
+/* First 9 words of an RDMA message with no chunks. */
+struct first_rdma {
+ uint32_t xid;
+ uint32_t vers;
+ uint32_t credits;
+ uint32_t body;
+ uint32_t readlist;
+ uint32_t writelist;
+ uint32_t reply;
+ uint32_t rpcxid;
+ uint32_t call;
+};
+
+static void
+clnt_rdma_upcall(struct rpcrdma_xprt *xp, struct mbuf *m)
+{
+ struct first_rdma first_rdma;
+ struct ct_request *cr;
+ int error, foundreq;
+ u_int rawlen;
+ struct crdma_data *crdma;
+ struct mbuf *m2;
+ SVCXPRT *xprt;
+ struct cf_conn *cd;
+
+ RPCRDMA_DEBUG(1, "clnt_rdma_upcall: Recv upcall m=%p\n", m);
+ crdma = (struct crdma_data *)__containerof((void *)xp,
+ struct crdma_data, crdma_xp);
+
+ /* Put the RPC message onto crdma_raw. */
+ error = 0;
+
+ mtx_lock(&crdma->crdma_lock);
+ if (m != NULL) {
+ if (crdma->crdma_raw != NULL)
+ m_last(crdma->crdma_raw)->m_next = m;
+ else
+ crdma->crdma_raw = m;
+ rawlen = m_length(crdma->crdma_raw, NULL);
+ } else {
+ /*
+ * We must have got EOF trying
+ * to read from the stream.
+ */
+ error = ECONNRESET;
+ RPCRDMA_DEBUG(1, "clnt_rdma_upcall: Connection reset\n");
+ }
+
+ /* Now, process as much of crdma_raw as possible. */
+ for (; error == 0;) {
+ /*
+ * Move a record, if possible.
+ */
+ if (rawlen == 0)
+ break;
+ if (crdma->crdma_record != NULL)
+ m_last(crdma->crdma_record)->m_next =
+ crdma->crdma_raw;
+ else
+ crdma->crdma_record = crdma->crdma_raw;
+ RPCRDMA_DEBUG(2, "clnt_rdma_call: Received a record\n");
+ crdma->crdma_raw = NULL;
+ rawlen = 0;
+
+ /*
+ * There are always more than 9 words in an
+ * RDMA message. For callbacks, I assume
+ * that there are no chunks, so the RDMA
+ * header is 7 words + the first two words
+ * of the RPC header.
+ */
+ if (crdma->crdma_record->m_len < sizeof(first_rdma) &&
+ m_length(crdma->crdma_record, NULL) < sizeof(first_rdma)) {
+ /*
+ * What to do now?
+ * The data in the RDMA stream is
+ * corrupted such that there is no
+ * valid RPC message to parse.
+ * I think it best to close this
+ * connection and allow
+ * clnt_reconnect_call() to try
+ * and establish a new one.
+ */
+ printf("clnt_rdma_upcall: "
+ "connection data corrupted\n");
+ error = ECONNRESET;
+ goto wakeup_all;
+ }
+ m_copydata(crdma->crdma_record, 0, sizeof(first_rdma),
+ (char *)&first_rdma);
+ first_rdma.xid = ntohl(first_rdma.xid);
+ first_rdma.call = ntohl(first_rdma.call);
+ /*
+ * RDMA_MSG and FALSE are both 0, so ntohl()
+ * is not needed for the comparisons.
+ * Check message direction.
+ */
+ if (first_rdma.body == RDMA_MSG &&
+ first_rdma.readlist == FALSE &&
+ first_rdma.writelist == FALSE &&
+ first_rdma.reply == FALSE &&
+ first_rdma.call == CALL) {
+ /*
+ * This is a backchannel request.
+ * Strip off the RDMA header.
+ */
+ m_adj(crdma->crdma_record, 7 * BYTES_PER_XDR_UNIT);
+ xprt = crdma->crdma_backchannelxprt;
+ if (xprt == NULL) {
+ /* Just throw it away. */
+ m_freem(crdma->crdma_record);
+ crdma->crdma_record = NULL;
+ } else {
+ cd = (struct cf_conn *)xprt->xp_p1;
+ m2 = cd->mreq;
+ /*
+ * The requests are chained
+ * in the m_nextpkt list.
+ */
+ while (m2 != NULL && m2->m_nextpkt != NULL)
+ /* Find end of list. */
+ m2 = m2->m_nextpkt;
+ if (m2 != NULL)
+ m2->m_nextpkt = crdma->crdma_record;
+ else
+ cd->mreq = crdma->crdma_record;
+ crdma->crdma_record->m_nextpkt = NULL;
+ crdma->crdma_record = NULL;
+ xprt_active(xprt);
+ }
+ } else {
+ /* This is a RPC reply. */
+ RPCRDMA_DEBUG(1, "clnt_rdma_upcall: Got an RPC "
+ "reply\n");
+ foundreq = 0;
+ TAILQ_FOREACH(cr, &crdma->crdma_pending, cr_link) {
+ if (cr->cr_xid == first_rdma.xid) {
+ /*
+ * This one
+ * matches. We leave
+ * the reply mbuf in
+ * cr->cr_mrep. Set
+ * the XID to zero so
+ * that we will ignore
+ * any duplicated
+ * replies.
+ */
+ cr->cr_xid = 0;
+ cr->cr_mrep = crdma->crdma_record;
+ cr->cr_error = 0;
+ foundreq = 1;
+ wakeup(cr);
+ RPCRDMA_DEBUG(1, "clnt_rdma_upcall: "
+ "Found reply for record\n");
+ break;
+ }
+ }
+ }
+
+ if (!foundreq) {
+ RPCRDMA_DEBUG(1, "clnt_rdma_upcall: No reply for "
+ "record\n");
+ m_freem(crdma->crdma_record);
+ }
+ crdma->crdma_record = NULL;
+ }
+
+ if (error != 0) {
+ wakeup_all:
+ /*
+ * This endpoint is broken, so mark that it cannot
+ * receive and fail all RPCs waiting for a reply
+ * on it, so that they will be retried on a new
+ * RDMA connection created by clnt_reconnect_connect().
+ */
+ RPCRDMA_DEBUG(1, "clnt_rdma_upcall: Wakeup all RPC_CANTRECV\n");
+ crdma->crdma_error.re_status = RPC_CANTRECV;
+ crdma->crdma_error.re_errno = error;
+ TAILQ_FOREACH(cr, &crdma->crdma_pending, cr_link) {
+ cr->cr_error = error;
+ wakeup(cr);
+ }
+ }
+ mtx_unlock(&crdma->crdma_lock);
+ RPCRDMA_DEBUG(1, "End of clnt_rdma_upcall\n");
+}
+
+/*
+ * Do a send of a backchannel reply.
+ */
+static bool_t
+clnt_rdma_bcksend(SVCXPRT *xprt, struct mbuf *m)
+{
+ struct crdma_data *crdma = (struct crdma_data *)xprt->xp_p2;
+ struct mbuf *mrep;
+ XDR xdrs;
+ uint32_t xid, rdma_vers, rdma_credit, rdma_body;
+ int error, ind;
+ bool_t bval;
+
+ /*
+ * The XID is in the first uint32_t of the reply.
+ */
+ if (m->m_len < sizeof(xid) && m_length(m, NULL) < sizeof(xid)) {
+ m_freem(m);
+ return (FALSE);
+ }
+ m_copydata(m, 0, sizeof(xid), (char *)&xid);
+ xid = ntohl(xid);
+
+ mrep = m_gethdr(M_WAITOK, MT_DATA);
+ mrep->m_len = 0;
+ xdrmbuf_create(&xdrs, mrep, XDR_ENCODE);
+
+ rdma_vers = 1;
+ rdma_credit = crdma->crdma_xp.maxbck;
+ rdma_body = RDMA_MSG;
+ bval = FALSE;
+ if ((! XDR_PUTINT32(&xdrs, &xid)) ||
+ (! XDR_PUTINT32(&xdrs, &rdma_vers)) ||
+ (! XDR_PUTINT32(&xdrs, &rdma_credit)) ||
+ (! XDR_PUTINT32(&xdrs, &rdma_body)) ||
+ (! XDR_PUTINT32(&xdrs, &bval)) ||
+ (! XDR_PUTINT32(&xdrs, &bval)) ||
+ (! XDR_PUTINT32(&xdrs, &bval))) {
+ m_freem(m);
+ return (FALSE);
+ }
+
+ if (! XDR_PUTMBUF(&xdrs, m))
+ return (FALSE);
+
+ ind = xprt_rdma_acquire_buf(&crdma->crdma_xp, crdma->crdma_xp.maxrpc,
+ crdma->crdma_xp.maxbck);
+ if (ind < 0) {
+ m_freem(mrep);
+ return (FALSE);
+ }
+ error = xprt_rdma_send(&crdma->crdma_xp, mrep, ind);
+ xprt_rdma_release_send(&crdma->crdma_xp, ind, NULL, NULL, NULL);
+ if (error != 0)
+ return (FALSE);
+ return (TRUE);
+}
+
+/*
+ * Kernel module glue
+ */
+static int
+nfsclrdma_modevent(module_t mod, int type, void *data)
+{
+ int error = 0;
+
+ switch (type) {
+ case MOD_LOAD:
+ rdma_check_route = xprt_rdma_check_route;
+ clnt_rdma_create_call = clnt_rdma_create;
+ clnt_rdma_bcksend_call = clnt_rdma_bcksend;
+ break;
+ case MOD_UNLOAD:
+ /*
+ * Cannot be unloaded.
+ */
+ /* FALLTHROUGH */
+ default:
+ error = EOPNOTSUPP;
+ }
+ return (error);
+}
+static moduledata_t nfsclrdma_mod = {
+ "nfsclrdma",
+ nfsclrdma_modevent,
+ NULL,
+};
+DECLARE_MODULE(nfsclrdma, nfsclrdma_mod, SI_SUB_VFS, SI_ORDER_FIRST);
+
+/* So that loader and kldload(2) can find us, wherever we are.. */
+MODULE_VERSION(nfsclrdma, 1);
+MODULE_DEPEND(nfsclrdma, krpc, 1, 1, 1);
+MODULE_DEPEND(nfsclrdma, nfscommon, 1, 1, 1);
+MODULE_DEPEND(nfsclrdma, nfscl, 1, 1, 1);
+MODULE_DEPEND(nfsclrdma, ibcore, 1, 1, 1);
diff --git a/rdma/xprt_rdma.h.rdma b/rdma/xprt_rdma.h
--- a/rdma/xprt_rdma.h.rdma
+++ b/rdma/xprt_rdma.h
@@ -0,0 +1,87 @@
+/* SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause */
+/*
+ * Copyright (c) 2014-2017 Oracle. All rights reserved.
+ * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
+ *
+ * This software is available to you under a choice of one of two
+ * licenses. You may choose to be licensed under the terms of the GNU
+ * General Public License (GPL) Version 2, available from the file
+ * COPYING in the main directory of this source tree, or the BSD-type
+ * license below:
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials provided
+ * with the distribution.
+ *
+ * Neither the name of the Network Appliance, Inc. nor the names of
+ * its contributors may be used to endorse or promote products
+ * derived from this software without specific prior written
+ * permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+/*
+ * This file is loosely based on net/sunrpc/xprtrdma/xprt_rdma.h
+ * in Linux.
+ */
+
+#ifndef _RDMA_XPRT_RDMA_H
+#define _RDMA_XPRT_RDMA_H
+
+#ifdef _KERNEL
+#define RDMA_RESOLVE_TIMEOUT (5000) /* 5 seconds */
+#define RDMA_SEND_TIMEOUT (30000) /* 30 seconds */
+#define RDMA_REG_TIMEOUT (500) /* 0.5 seconds */
+#define RDMA_KEYINV_TIMEOUT (5000) /* 5.0 seconds */
+#define RDMA_DISCONNECT_TIMEOUT (5000) /* 5 seconds */
+#define RDMA_CONNECT_RETRY_MAX (2) /* retries if no listener backlog */
+
+#define RPCRDMA_BIND_TO (60U * HZ)
+#define RPCRDMA_INIT_REEST_TO (5U * HZ)
+#define RPCRDMA_MAX_REEST_TO (30U * HZ)
+#define RPCRDMA_IDLE_DISC_TO (5U * 60 * HZ)
+
+/* send/receive buffer stuff. */
+
+struct rpcrdma_buf {
+ uint32_t pgcnt;
+ uint8_t into_mem;
+ struct iovec iov[];
+};
+
+struct rpcrdma_regwr {
+ struct ib_reg_wr wr;
+ struct ib_cqe cqe;
+ int async_rc;
+ struct completion done;
+};
+
+struct _rpcrdma_chunk_priv {
+ struct rpcrdma_chunk ch;
+ uint32_t rbpages;
+ struct rpcrdma_regwr regwr[RPCRDMA_MAX_SEGMENTS];
+ struct scatterlist sge[RPCRDMA_MAX_SEGMENTS][RPCRDMA_MAX_SGE];
+};
+
+#endif /* _KERNEL */
+
+#endif /* _RDMA_XPRT_RDMA_H */
diff --git a/xprt_verbs.c.rdma b/xprt_verbs.c
--- a/xprt_verbs.c.rdma
+++ b/xprt_verbs.c
@@ -0,0 +1,1397 @@
+// SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
+/*
+ * Copyright (c) 2014-2017 Oracle. All rights reserved.
+ * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
+ *
+ * This software is available to you under a choice of one of two
+ * licenses. You may choose to be licensed under the terms of the GNU
+ * General Public License (GPL) Version 2, available from the file
+ * COPYING in the main directory of this source tree, or the BSD-type
+ * license below:
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials provided
+ * with the distribution.
+ *
+ * Neither the name of the Network Appliance, Inc. nor the names of
+ * its contributors may be used to endorse or promote products
+ * derived from this software without specific prior written
+ * permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+/*
+ * This file is loosely based on net/sunrpc/xprtrdma/verbs.c.
+ * in Linux.
+ * It is significantly different, due to FreeBSD's rdma_XXX()
+ * functions having different arguments, among other things.
+ */
+/*
+ * verbs.c
+ *
+ * Encapsulates the major functions managing:
+ * o adapters
+ * o endpoints
+ * o connections
+ * o buffer memory
+ */
+
+#include <linux/errno.h>
+#include <linux/kernel.h>
+#include <sys/mbuf.h>
+#include <sys/socket.h>
+#include <rpc/clntrdma.h>
+#include <rdma/rdma_cm.h>
+#include <rdma/xprt_rdma.h>
+
+#include <net/if.h>
+#include <net/if_var.h>
+#include <net/if_private.h>
+#include <net/route.h>
+#include <net/route/nhop.h>
+#include <netinet/tcp.h>
+#include <netinet/in.h>
+#include <netinet/in_fib.h>
+#include <netinet/in_pcb.h>
+#include <netinet6/in6_fib.h>
+#include <netinet6/scope6_var.h>
+
+#include "opt_inet.h"
+#include "opt_inet6.h"
+
+MALLOC_DEFINE(M_RPCRDMA, "Krpc RDMA", "Krpc RDMA");
+
+struct rpcrdma_send {
+ char *buf;
+ struct ib_cqe cqe;
+ struct ib_send_wr wr;
+ struct ib_sge sge;
+ struct rpcrdma_ep *ep;
+ int async_rc;
+ struct completion done;
+};
+
+struct rpcrdma_recv {
+ char *buf;
+ struct ib_cqe cqe;
+ struct ib_recv_wr wr;
+ struct ib_sge sge;
+ struct rpcrdma_ep *ep;
+ int ind;
+ int async_rc;
+ struct completion done;
+};
+
+/* Structure for a connection endpoint. */
+struct rpcrdma_ep {
+ struct rpcrdma_xprt *re_xp;
+ int re_async_rc;
+ struct completion re_done;
+ struct ib_qp_init_attr re_attr;
+ xprt_rdma_upcall re_upcall;
+#define re_startzero re_id
+ struct rdma_cm_id *re_id;
+ struct ib_pd *re_pd;
+ struct ib_cq *re_recv_cq;
+ struct ib_cq *re_send_cq;
+ uint64_t re_busy;
+ struct rpcrdma_send re_send[RPCRDMA_IO_NUMBUFS];
+ struct rpcrdma_recv re_recv[RPCRDMA_IO_NUMBUFS];
+ struct ib_mr *re_mr[RPCRDMA_IO_NUMBUFS][RPCRDMA_MAX_SEGMENTS];
+ struct ib_mr *re_reply[RPCRDMA_IO_NUMBUFS][RPCRDMA_MAX_SEGMENTS];
+ size_t re_buflen;
+ unsigned int re_send_count;
+ int re_receive_count;
+ int re_connect_status;
+ struct rdma_conn_param re_remote_cma;
+#define re_endzero re_mrtype
+ enum ib_mr_type re_mrtype;
+};
+
+typedef void (*rpcrdma_done)(struct ib_cq *cq, struct ib_wc *wc);
+
+static void
+rpcrdma_free_ep(struct rpcrdma_ep *ep)
+{
+ int i, j;
+
+ ep->re_xp = NULL;
+ for (i = 0; i < RPCRDMA_IO_NUMBUFS; i++) {
+ if (ep->re_id && ep->re_recv[i].sge.addr != 0)
+ ib_dma_unmap_single(ep->re_id->device,
+ ep->re_recv[i].sge.addr,
+ ep->re_buflen, DMA_FROM_DEVICE);
+ free(ep->re_recv[i].buf, M_RPCRDMA);
+ if (ep->re_id && ep->re_send[i].sge.addr != 0)
+ ib_dma_unmap_single(ep->re_id->device,
+ ep->re_send[i].sge.addr,
+ ep->re_buflen, DMA_TO_DEVICE);
+ free(ep->re_send[i].buf, M_RPCRDMA);
+ for (j = 0; j < RPCRDMA_MAX_SEGMENTS; j++) {
+ if (ep->re_id && ep->re_mr[i][j] != NULL)
+ ib_dereg_mr(ep->re_mr[i][j]);
+ if (ep->re_id && ep->re_reply[i][j] != NULL)
+ ib_dereg_mr(ep->re_reply[i][j]);
+ }
+ }
+#ifdef notnow
+ if (ep->re_pd)
+ ib_dealloc_pd(ep->re_pd);
+ if (ep->re_recv_cq)
+ ib_destroy_cq(ep->re_recv_cq);
+ if (ep->re_send_cq)
+ ib_destroy_cq(ep->re_send_cq);
+#endif
+ if (ep->re_connect_status == 0)
+ ep->re_connect_status = -ECONNABORTED;
+}
+
+static int
+rpcrdma_cm_event_handler(struct rdma_cm_id *id, struct rdma_cm_event *event)
+{
+ struct rpcrdma_ep *ep = id->context;
+ int rc;
+
+ might_sleep();
+
+ RPCRDMA_DEBUG(3, "rpcrdma_cm_event_handler: event=%d\n", event->event);
+ switch (event->event) {
+ case RDMA_CM_EVENT_ADDR_RESOLVED:
+ case RDMA_CM_EVENT_ROUTE_RESOLVED:
+ case RDMA_CM_EVENT_ESTABLISHED:
+ ep->re_async_rc = 0;
+ complete(&ep->re_done);
+ return (0);
+ case RDMA_CM_EVENT_ADDR_ERROR:
+ ep->re_async_rc = -EPROTO;
+ complete(&ep->re_done);
+ return (-1);
+ case RDMA_CM_EVENT_ROUTE_ERROR:
+ case RDMA_CM_EVENT_REJECTED:
+ case RDMA_CM_EVENT_UNREACHABLE:
+ /*
+ * This error matters. When ENETUNREACH is returned by
+ * xprt_rdma_connect(), clnt_rdma_create() sets
+ * RPC_SYSTEMERROR in rc_err->re_status. This results in
+ * clnt_reconnect_connect() returning that status and that
+ * causes reconnection retries in clnt_reconnect_call().
+ */
+ ep->re_async_rc = -ENETUNREACH;
+ complete(&ep->re_done);
+ return (-1);
+ case RDMA_CM_EVENT_DISCONNECTED:
+ ep->re_connect_status = -ECONNABORTED;
+ if (ep->re_xp != NULL && ep->re_upcall != NULL)
+ ep->re_upcall(ep->re_xp, NULL);
+ rc = rdma_disconnect(id);
+ RPCRDMA_DEBUG(4, "rpcrdma_cm_event_handler: "
+ "rdma_disconnect=%d\n", rc);
+ return (0);
+ case RDMA_CM_EVENT_TIMEWAIT_EXIT:
+ ep->re_id = NULL; /* id is going away, since we return -1. */
+ ep->re_connect_status = -ECONNABORTED;
+ ep->re_async_rc = 0;
+ complete(&ep->re_done);
+ return (-1);
+ default:
+ printf("rpcrdma_cm_event_handler: Unknown event=%d\n",
+ event->event);
+ ep->re_async_rc = -EIO;
+ complete(&ep->re_done);
+ return (-1);
+ }
+
+ return (0);
+}
+
+void
+xprt_rdma_init(struct rpcrdma_xprt *xp, xprt_rdma_upcall rdma_upcall)
+{
+ struct rpcrdma_ep *ep;
+
+ KASSERT(xp->ep == NULL, ("xprt_rdma_init: ep not NULL"));
+ ep = malloc(sizeof(*ep), M_RPCRDMA, M_WAITOK | M_ZERO);
+
+ /* And set the upcall. */
+ ep->re_upcall = rdma_upcall;
+
+ xp->ep = ep;
+ ep->re_xp = xp;
+}
+
+static struct rpcrdma_ep *
+rpcrdma_acquire_ep(struct rpcrdma_xprt *xp)
+{
+ struct rpcrdma_ep *ep;
+
+ mtx_lock(&xp->mtx);
+ ep = xp->ep;
+ if (ep == NULL || ep->re_connect_status < 0) {
+ mtx_unlock(&xp->mtx);
+ RPCRDMA_DEBUG(4, "rpcrdma_acquire_ep: failed ep=%p\n", ep);
+ return (NULL);
+ }
+ mtx_unlock(&xp->mtx);
+ return (ep);
+}
+
+/*
+ * Acquire a buffer, waiting until one is available.
+ * *busyp points to a uint64_t bitmap, so it can handle up to 64 buffers.
+ */
+int
+xprt_rdma_acquire_buf(struct rpcrdma_xprt *xp, int start, int end)
+{
+ struct rpcrdma_ep *ep;
+ uint32_t credits;
+ int i;
+
+ mtx_assert(&xp->mtx, MA_OWNED);
+ for (;;) {
+ ep = (struct rpcrdma_ep *)xp->ep;
+ if (ep == NULL || ep->re_connect_status != 0)
+ return (-1);
+ credits = UINT32_MAX;
+ if (start == 0)
+ credits = xp->credits;
+ for (i = start; i < credits && i < end; i++) {
+ if (((1ull << i) & ep->re_busy) == 0) {
+ ep->re_busy |= (1ull << i);
+ return (i);
+ }
+ }
+ RPCRDMA_DEBUG(4, "xprt_rdma_acquire_buf: sleeping credits=%u "
+ "busy=0x%ju start=%d end=%d\n", credits,
+ (uintmax_t)ep->re_busy, start, end);
+ msleep(&ep->re_busy, &xp->mtx, PVFS, "rpcrdma", hz);
+ }
+}
+
+/*
+ * Release a buffer.
+ */
+static void
+rpcrdma_release_buf(struct rpcrdma_xprt *xp, uint64_t *busyp, u_int index)
+{
+
+ mtx_lock(&xp->mtx);
+ *busyp &= ~(1ull << index);
+ wakeup(busyp);
+ mtx_unlock(&xp->mtx);
+}
+
+static void
+rpcrdma_send_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+ struct rpcrdma_send *rs;
+ struct rpcrdma_regwr *regwr;
+
+ RPCRDMA_DEBUG(4, "rpcrdma_send_done: op=%d stat=%d\n", wc->opcode,
+ wc->status);
+ if (wc->status != IB_WC_SUCCESS)
+ WARN_ONCE(wc->status, "rpcrdma_send_done: failed opcode=%u"
+ " status=%u\n", wc->opcode, wc->status);
+ switch (wc->opcode) {
+ case IB_WC_SEND:
+ rs = container_of(wc->wr_cqe, struct rpcrdma_send, cqe);
+ rs->async_rc = 0;
+ if (wc->status != IB_WC_SUCCESS)
+ rs->async_rc = EPIPE;
+ complete(&rs->done);
+ break;
+ case IB_WC_REG_MR:
+ case IB_WC_LOCAL_INV:
+ regwr = container_of(wc->wr_cqe, struct rpcrdma_regwr, cqe);
+ regwr->async_rc = 0;
+ if (wc->status != IB_WC_SUCCESS)
+ regwr->async_rc = EPIPE;
+ complete(&regwr->done);
+ break;
+ default:
+ printf("rpcrdma_send_done: Unknown opcode=%u status=%u\n",
+ wc->opcode, wc->status);
+ }
+}
+
+static int
+rpcrdma_post_recv(struct rpcrdma_ep *ep, int ind)
+{
+ const struct ib_recv_wr *bad_rwr;
+ int ret;
+
+ init_completion(&ep->re_recv[ind].done);
+ ep->re_recv[ind].async_rc = -ETIMEDOUT;
+ ret = ib_post_recv(ep->re_id->qp, &ep->re_recv[ind].wr, &bad_rwr);
+ RPCRDMA_DEBUG(1, "rpcrdma_post_recv: ind=%d ib_post_recv=%d\n", ind,
+ ret);
+ return (ret);
+}
+
+static void
+rpcrdma_recv_done(struct ib_cq *cq, struct ib_wc *wc)
+{
+ struct uio uio;
+ struct iovec iov;
+ struct rpcrdma_recv *rs;
+ struct rpcrdma_ep *ep;
+ struct mbuf *m;
+ int rc;
+
+ if (cq == NULL || wc == NULL) {
+ printf("rpcrdma_recv_done: cq=%p wc=%p\n", cq, wc);
+ return;
+ }
+ RPCRDMA_DEBUG(4, "rpcrdma_recv_done: code=%d stat=%d len=%d\n",
+ wc->opcode, wc->status, wc->byte_len);
+ rs = container_of(wc->wr_cqe, struct rpcrdma_recv, cqe);
+ if (rs == NULL) {
+ printf("rpcrdma_recv_done: rs NULL\n");
+ return;
+ }
+ ep = rs->ep;
+ if (ep == NULL) {
+ printf("rpcrdma_recv_done: ep NULL\n");
+ return;
+ }
+
+ if (wc->opcode != IB_WC_RECV || wc->status != IB_WC_SUCCESS ||
+ wc->byte_len == 0) {
+ if (wc->status != IB_WC_WR_FLUSH_ERR)
+ printf("rpcrdma_recv_done: opcode=%d status=%d "
+ "len=%d\n", wc->opcode, wc->status, wc->byte_len);
+ return;
+ }
+
+ if (wc->byte_len > ep->re_buflen) {
+ printf("rpcrdma_recv_done: byte_len=%d too large\n",
+ wc->byte_len);
+ return;
+ }
+
+ /* fill in an mbuf chain with the data. */
+ iov.iov_base = rs->buf;
+ iov.iov_len = wc->byte_len;
+ uio.uio_iov = &iov;
+ uio.uio_iovcnt = 1;
+ uio.uio_resid = wc->byte_len;
+ uio.uio_segflg = UIO_SYSSPACE;
+ uio.uio_rw = UIO_WRITE;
+ uio.uio_offset = 0;
+ uio.uio_td = NULL;
+ m = m_uiotombuf(&uio, M_WAITOK, 0, 0, M_PKTHDR);
+ if (m == NULL) {
+ printf("rpcrdma_recv_done: m_uiotombuf failed\n");
+ return;
+ }
+
+ /* And re-post the recv buffer. */
+ rc = rpcrdma_post_recv(ep, rs->ind);
+ if (rc) {
+ RPCRDMA_DEBUG(1, "rpcrdma_recv_done: rpcrdma_post_recv ind=%d "
+ "failed=%d\n", rs->ind, rc);
+ return;
+ }
+ /* Re-enable the receive buffer. */
+ RPCRDMA_DEBUG(4, "rpcrdma_recv_done: calling upcall m=%p\n", m);
+ if (ep->re_xp != NULL)
+ ep->re_upcall(ep->re_xp, m);
+}
+
+/*
+ * Try and figure out the srcaddr.
+ * Return 0 if a link local INET6 address.
+ * Return 1 if the address is resolved.
+ * Return -1 for failure.
+ */
+static int
+_rpcrdma_get_srcaddr(struct sockaddr *saddr, struct sockaddr_storage *srcaddr)
+{
+ struct nhop_object *nh = NULL;
+ struct epoch_tracker et;
+#ifdef INET
+ struct sockaddr_in *sin;
+#endif
+#ifdef INET6
+ struct sockaddr_in6 *sin6;
+ struct in6_addr kdst;
+ uint32_t scopeid;
+#endif
+
+ CURVNET_SET_QUIET(TD_TO_VNET(curthread));
+ switch (saddr->sa_family) {
+#ifdef INET
+ case AF_INET:
+ sin = (struct sockaddr_in *)saddr;
+ NET_EPOCH_ENTER(et);
+ nh = fib4_lookup(curthread->td_proc->p_fibnum, sin->sin_addr, 0,
+ NHR_NONE, 0);
+ if (nh == NULL || nh->nh_ifa == NULL ||
+ nh->nh_ifa->ifa_addr == NULL) {
+ NET_EPOCH_EXIT(et);
+ CURVNET_RESTORE();
+ return (-1);
+ }
+ memcpy(srcaddr, nh->nh_ifa->ifa_addr, sizeof(*sin));
+ NET_EPOCH_EXIT(et);
+ CURVNET_RESTORE();
+ RPCRDMA_DEBUG(3, "_rpcrdma_get_srcaddr: NIC to be used %s\n",
+ nh->nh_ifp->if_xname);
+ return (0);
+#endif
+#ifdef INET6
+ case AF_INET6:
+ sin6 = (struct sockaddr_in6 *)saddr;
+ if (IN6_IS_ADDR_LINKLOCAL(&sin6->sin6_addr)) {
+ CURVNET_RESTORE();
+ return (0);
+ }
+ in6_splitscope(&sin6->sin6_addr, &kdst, &scopeid);
+ NET_EPOCH_ENTER(et);
+ nh = fib6_lookup(curthread->td_proc->p_fibnum, &kdst, scopeid,
+ NHR_NONE, 0);
+ if (nh == NULL || nh->nh_ifa == NULL ||
+ nh->nh_ifa->ifa_addr == NULL) {
+ NET_EPOCH_EXIT(et);
+ CURVNET_RESTORE();
+ return (-1);
+ }
+ memcpy(srcaddr, nh->nh_ifa->ifa_addr, sizeof(*sin));
+ NET_EPOCH_EXIT(et);
+ CURVNET_RESTORE();
+ RPCRDMA_DEBUG(3, "_rpcrdma_get_srcaddr: NIC to be used %s\n",
+ nh->nh_ifp->if_xname);
+ return (0);
+#endif
+ default:
+ CURVNET_RESTORE();
+ RPCRDMA_DEBUG(3, "_rpcrdma_get_srcaddr: No NIC found\n");
+ return (-1);
+ }
+ CURVNET_RESTORE();
+ return (-1);
+}
+
+static struct rdma_cm_id *
+rpcrdma_create_id(struct vnet *net, struct sockaddr *saddr,
+ struct rpcrdma_ep *ep, struct rpcrdma_xprt *xp, uint32_t cbslots)
+{
+ struct rdma_cm_id *id;
+ const struct ib_device_attr *attr;
+ struct sockaddr_storage ss;
+ struct sockaddr *srcaddr;
+ int numbufs, numsge, rc;
+
+ /* Try and figure out what the srcaddr is via routing. */
+ rc = _rpcrdma_get_srcaddr(saddr, &ss);
+ switch (rc) {
+ case 0:
+ srcaddr = NULL;
+ break;
+ case 1:
+ srcaddr = (struct sockaddr *)&ss;
+ break;
+ case -1:
+ RPCRDMA_DEBUG(3, "rpcrdma_create_id: No NIC addr found\n");
+ return (ERR_PTR(-EADDRNOTAVAIL));
+ }
+
+ id = rdma_create_id(net, rpcrdma_cm_event_handler, ep,
+ RDMA_PS_TCP, IB_QPT_RC);
+ RPCRDMA_DEBUG(4, "rpcrdma_create_id: rdma_create_id=%p\n", id);
+ if (IS_ERR(id))
+ return (id);
+
+ init_completion(&ep->re_done);
+ ep->re_async_rc = -ETIMEDOUT;
+ rc = rdma_resolve_addr(id, srcaddr, saddr, RDMA_RESOLVE_TIMEOUT);
+ if (rc) {
+ rdma_destroy_id(id);
+ goto out;
+ }
+ rc = wait_for_completion(&ep->re_done);
+ if (rc < 0)
+ goto out;
+ rc = ep->re_async_rc;
+ RPCRDMA_DEBUG(4, "rpcrdma_create_id: rdma_resolve_addr=%d\n", rc);
+ if (rc)
+ goto out;
+
+ init_completion(&ep->re_done);
+ ep->re_async_rc = -ETIMEDOUT;
+ rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT);
+ if (rc) {
+ rdma_destroy_id(id);
+ goto out;
+ }
+ rc = wait_for_completion(&ep->re_done);
+ if (rc < 0)
+ goto out;
+ rc = ep->re_async_rc;
+ RPCRDMA_DEBUG(4, "rpcrdma_create_id: rdma_resolve_route=%d\n", rc);
+ if (rc)
+ goto out;
+
+ if (id->device == NULL) {
+ /* I don't think this should ever happen? */
+ printf("rpcrdma_create_id: No valid device\n");
+ rc = -EPROTONOSUPPORT;
+ rdma_destroy_id(id);
+ goto out;
+ }
+
+ attr = &id->device->attrs;
+ /* The device attrs should now be set, so check them. */
+ if ((attr->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS) == 0) {
+ printf("rpcrdma_create_id: No memory extensions\n");
+ rc = -EPROTONOSUPPORT;
+ rdma_destroy_id(id);
+ goto out;
+ }
+
+ ep->re_mrtype = IB_MR_TYPE_MEM_REG;
+ if ((attr->device_cap_flags & IB_DEVICE_SG_GAPS_REG) != 0)
+ ep->re_mrtype = IB_MR_TYPE_SG_GAPS;
+
+ /*
+ * I am still pretty sketchy w.r.t. ofed/rdma, but I think..
+ * - MIN(max_sg, max_sg_rd) is the max # of pages per segment (MR).
+ * - I assume it also sets the limit on the # of send/recv buffers per
+ * QP. Some drivers may allow larger values for the max_send_sge
+ * and max_recv_sge but assuming only one MR seems to work ok.
+ * - RFC-8267 sets a limit of 16 segments (MRs) per chunk and 1 chunk
+ * per RPC.
+ * For a worse case of all RPCs (Read/Write/Readdir) needing a chunk and
+ * NFS I/O sizes (nm_rsize, nm_wsize) are always a power of 2, so
+ * round MIN(max_sg, max_sg_rd) down to a power of 2.
+ * - Add 1 for the rest of the RPC message other than the data.
+ * - Add 1 for the minimum send/recv buffers needed for backchannel msg.
+ * maxrpc - Set to the maximum # of concurrent RPCs.
+ * maxbck - Set to the maximum # of concurrent back channel RPCs.
+ * maxio - Set to the maximum I/O size allowed.
+ * maxsge - Set to the maximum sge size.
+ * For now, it appears that the device attributes max_sge, max_sge_rd
+ * and max_qp_rd_atom set the limits.
+ */
+ numbufs = RPCRDMA_IO_NUMBUFS;
+
+ numsge = MIN(attr->max_sge, attr->max_sge_rd);
+ numbufs = MIN(numbufs, numsge);
+ numbufs = 1 << (fls(numbufs) - 1); /* Round down to a power of 2. */
+ if (numbufs + 2 > numsge || numbufs + 1 > RPCRDMA_MAX_SGE)
+ numbufs /= 2;
+ xp->maxsge = MIN(numbufs + 1, RPCRDMA_MAX_SGE);
+ xp->maxrpc = numbufs;
+ xp->maxbck = MIN(numsge - numbufs, cbslots);
+ /* Since all RPCs might require a Read chunk... */
+ if (xp->maxrpc > attr->max_qp_rd_atom) {
+ xp->maxrpc = attr->max_qp_rd_atom;
+ xp->maxbck = numsge - xp->maxrpc;
+ xp->maxbck = MIN(xp->maxbck, cbslots);
+ }
+ xp->maxio = numbufs * RPCRDMA_MAX_SEGMENTS * PAGE_SIZE;
+ RPCRDMA_DEBUG(1, "rpcrdma_create_id: maxrpc=%u maxbck=%u maxio=%u "
+ "id=%p\n", xp->maxrpc, xp->maxbck, xp->maxio, id);
+ return (id);
+
+out:
+ RPCRDMA_DEBUG(1, "rpcrdma_create_id: Failed %d\n", rc);
+ return (ERR_PTR(rc));
+}
+
+static int
+rpcrdma_ep_create(struct vnet *net, struct sockaddr *saddr,
+ struct rpcrdma_ep *ep, struct rpcrdma_xprt *xp, size_t buflen,
+ uint32_t cbslots)
+{
+ struct ib_device *device;
+ struct rdma_cm_id *id;
+ struct ib_pd *pd;
+ struct ib_cq *recv_cq, *send_cq;
+ int i, j, rc;
+
+ pd = NULL;
+ memset(&ep->re_startzero, 0,
+ __rangeof(struct rpcrdma_ep, re_startzero, re_endzero));
+ id = rpcrdma_create_id(net, saddr, ep, xp, cbslots);
+ if (IS_ERR(id))
+ return (PTR_ERR(id));
+ device = id->device;
+ reinit_completion(&ep->re_done);
+
+ pd = ib_alloc_pd(device, 0);
+ if (IS_ERR(pd)) {
+ rc = PTR_ERR(pd);
+ pd = NULL;
+ goto out_destroy;
+ }
+
+ recv_cq = ib_alloc_cq(device, xp, 1024, 0, IB_POLL_WORKQUEUE);
+ if (IS_ERR(recv_cq)) {
+ rc = PTR_ERR(recv_cq);
+ ib_dealloc_pd(pd);
+ pd = NULL;
+ recv_cq = NULL;
+ goto out_destroy;
+ }
+
+ send_cq = ib_alloc_cq(device, xp, 1024, 0, IB_POLL_WORKQUEUE);
+ if (IS_ERR(send_cq)) {
+ rc = PTR_ERR(send_cq);
+ ib_dealloc_pd(pd);
+ ib_destroy_cq(recv_cq);
+ pd = NULL;
+ recv_cq = NULL;
+ send_cq = NULL;
+ goto out_destroy;
+ }
+
+ /* The cap fields of re_attr should be filled in, the rest zeroed. */
+ memset(&ep->re_attr, 0, sizeof(ep->re_attr));
+ ep->re_attr.qp_type = IB_QPT_RC;
+ ep->re_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
+ ep->re_attr.qp_context = ep;
+ ep->re_attr.recv_cq = recv_cq;
+ ep->re_attr.send_cq = send_cq;
+
+ /*
+ * For max_send_sge and max_recv_sge, set them to maxsge.
+ * The plus 2 is for the rest of a large RPC message and the send/recv
+ * buffer used along with a chunk.
+ */
+ ep->re_attr.cap.max_send_wr = xp->maxrpc + xp->maxbck + xp->maxrpc *
+ RPCRDMA_MAX_SEGMENTS;
+ ep->re_attr.cap.max_recv_wr = xp->maxrpc + xp->maxbck + xp->maxrpc *
+ RPCRDMA_MAX_SEGMENTS;
+ ep->re_attr.cap.max_recv_sge = xp->maxsge;
+ ep->re_attr.cap.max_send_sge = xp->maxsge;
+ rc = rdma_create_qp(id, pd, &ep->re_attr);
+ if (rc) {
+ RPCRDMA_DEBUG(1, "rpcrdma_ep_create: rdma_create_qp "
+ "failed=%d\n", rc);
+ ib_dealloc_pd(pd);
+ ib_destroy_cq(recv_cq);
+ ib_destroy_cq(send_cq);
+ recv_cq = NULL;
+ send_cq = NULL;
+ pd = NULL;
+ goto out_destroy;
+ }
+
+ ep->re_buflen = buflen;
+ RPCRDMA_DEBUG(4, "rpcrdma_ep_create: Initialize %u send/recv buffers "
+ "len=%ju\n", xp->maxrpc + xp->maxbck, (uintmax_t)buflen);
+ for (i = 0; i < xp->maxrpc + xp->maxbck; i++) {
+ ep->re_send[i].buf = malloc(buflen, M_RPCRDMA, M_WAITOK);
+ ep->re_send[i].sge.addr = ib_dma_map_single(device,
+ ep->re_send[i].buf, buflen, DMA_TO_DEVICE);
+ ep->re_send[i].sge.length = buflen;
+ ep->re_send[i].sge.lkey = pd->local_dma_lkey;
+ rc = ib_dma_mapping_error(device, ep->re_send[i].sge.addr);
+ if (rc) {
+ ep->re_send[i].sge.addr = 0;
+ RPCRDMA_DEBUG(4, "rpcrdma_ep_create: dma mapping "
+ "failed=%d\n", rc);
+ goto out_destroy_qp;
+ }
+ ep->re_send[i].wr.next = NULL;
+ ep->re_send[i].wr.num_sge = 1;
+ ep->re_send[i].wr.wr_cqe = &ep->re_send[i].cqe;
+ ep->re_send[i].wr.sg_list = &ep->re_send[i].sge;
+ ep->re_send[i].cqe.done = rpcrdma_send_done;
+ ep->re_send[i].wr.send_flags = IB_SEND_SIGNALED;
+ ep->re_send[i].wr.opcode = IB_WR_SEND;
+
+ ep->re_recv[i].buf = malloc(buflen, M_RPCRDMA, M_WAITOK);
+ ep->re_recv[i].sge.addr = ib_dma_map_single(device,
+ ep->re_recv[i].buf, buflen, DMA_FROM_DEVICE);
+ ep->re_recv[i].sge.length = buflen;
+ ep->re_recv[i].sge.lkey = pd->local_dma_lkey;
+ rc = ib_dma_mapping_error(device, ep->re_recv[i].sge.addr);
+ if (rc) {
+ ep->re_recv[i].sge.addr = 0;
+ RPCRDMA_DEBUG(4, "rpcrdma_ep_create: dma mapping "
+ "failed=%d\n", rc);
+ goto out_destroy_qp;
+ }
+ ep->re_recv[i].ep = ep;
+ ep->re_recv[i].wr.next = NULL;
+ ep->re_recv[i].wr.num_sge = 1;
+ ep->re_recv[i].wr.sg_list = &ep->re_recv[i].sge;
+ ep->re_recv[i].wr.wr_cqe = &ep->re_recv[i].cqe;
+ ep->re_recv[i].cqe.done = rpcrdma_recv_done;
+ ep->re_recv[i].ind = i;
+
+ for (j = 0; j < RPCRDMA_MAX_SEGMENTS; j++) {
+ ep->re_mr[i][j] = ib_alloc_mr(pd, IB_MR_TYPE_MEM_REG,
+ xp->maxsge);
+ if (IS_ERR(ep->re_mr[i][j])) {
+ RPCRDMA_DEBUG(4, "rpcrdma_ep_create: alloc mr "
+ "failed=%p\n", ep->re_mr[i][j]);
+ ep->re_mr[i][j] = NULL;
+ goto out_destroy_qp;
+ }
+ ep->re_reply[i][j] = ib_alloc_mr(pd, IB_MR_TYPE_MEM_REG,
+ xp->maxsge);
+ if (IS_ERR(ep->re_reply[i][j])) {
+ RPCRDMA_DEBUG(4, "rpcrdma_ep_create: alloc mr "
+ "failed=%p\n", ep->re_mr[i][j]);
+ ep->re_reply[i][j] = NULL;
+ goto out_destroy_qp;
+ }
+ }
+ }
+
+ ep->re_pd = pd;
+ ep->re_id = id;
+ ep->re_recv_cq = recv_cq;
+ ep->re_send_cq = send_cq;
+ RPCRDMA_DEBUG(1, "rpcrdma_ep_create succeeded\n");
+ return (0);
+
+out_destroy_qp:
+ rdma_destroy_qp(id);
+out_destroy:
+ for (i = 0; i < xp->maxrpc + xp->maxbck; i++) {
+ if (ep->re_recv[i].sge.addr != 0)
+ ib_dma_unmap_single(device, ep->re_recv[i].sge.addr,
+ ep->re_buflen, DMA_FROM_DEVICE);
+ free(ep->re_recv[i].buf, M_RPCRDMA);
+ if (ep->re_send[i].sge.addr != 0)
+ ib_dma_unmap_single(device, ep->re_send[i].sge.addr,
+ ep->re_buflen, DMA_TO_DEVICE);
+ free(ep->re_send[i].buf, M_RPCRDMA);
+ for (j = 0; j < RPCRDMA_MAX_SEGMENTS; j++) {
+ if (ep->re_mr[i][j] != NULL)
+ ib_dereg_mr(ep->re_mr[i][j]);
+ if (ep->re_reply[i][j] != NULL)
+ ib_dereg_mr(ep->re_reply[i][j]);
+ }
+ }
+ rdma_destroy_id(id);
+ memset(&ep->re_startzero, 0,
+ __rangeof(struct rpcrdma_ep, re_startzero, re_endzero));
+ RPCRDMA_DEBUG(1, "rpcrdma_ep_create: failed=%d\n", rc);
+ return (rc);
+}
+
+int
+xprt_rdma_check_route(struct vnet *net, struct sockaddr *saddr,
+ uint32_t cbslots)
+{
+ struct rpcrdma_xprt xpr;
+ struct rdma_cm_id *id;
+
+ memset(&xpr, 0, sizeof(xpr));
+ xprt_rdma_init(&xpr, NULL);
+ mtx_init(&xpr.mtx, "xprt_check_mtx", NULL, MTX_DEF);
+ id = rpcrdma_create_id(net, saddr, xpr.ep, &xpr, cbslots);
+ if (IS_ERR(id)) {
+ mtx_destroy(&xpr.mtx);
+ return (EPROTONOSUPPORT);
+ }
+ rdma_disconnect(id);
+ rdma_destroy_id(id);
+ free(xpr.ep, M_RPCRDMA);
+ mtx_destroy(&xpr.mtx);
+ return (0);
+}
+
+int
+xprt_rdma_connect(struct vnet *net, struct sockaddr *saddr,
+ struct rpcrdma_xprt *xp, size_t buflen, uint32_t cbslots)
+{
+ struct rpcrdma_ep *ep;
+ struct ib_device *device;
+ int i, rc;
+
+ ep = rpcrdma_acquire_ep(xp);
+ if (ep == NULL)
+ return (ECONNREFUSED);
+ rc = rpcrdma_ep_create(net, saddr, ep, xp, buflen, cbslots);
+ if (rc)
+ goto out;
+
+ /* Initialize cma parameters */
+ /* Client offers RDMA Read but does not initiate */
+ ep->re_remote_cma.initiator_depth = 0;
+ device = ep->re_id->device;
+ /*
+ * responder_resources is a uint8_t, but Intel reports 256 for
+ * max_qp_rd_atom, which is why it is the minimum of max_qp_rd_atom
+ * and UINT8_MAX.
+ */
+ ep->re_remote_cma.responder_resources =
+ min_t(int, U8_MAX, device->attrs.max_qp_rd_atom);
+
+ /*
+ * Try an infinite retry count. The clnt_reconnect layer should
+ * handle the case where the connection is hung.
+ */
+ ep->re_remote_cma.retry_count = 7;
+
+ /*
+ * RPC-over-RDMA handles its own flow control. In addition,
+ * make all RNR NAKs visible so we know that RPC-over-RDMA
+ * flow control is working correctly (no NAKs should be seen).
+ */
+ ep->re_remote_cma.flow_control = 0;
+ ep->re_remote_cma.rnr_retry_count = 0;
+
+ init_completion(&ep->re_done);
+ ep->re_async_rc = -ETIMEDOUT;
+ rc = rdma_connect(ep->re_id, &ep->re_remote_cma);
+ if (rc)
+ goto out_destroy;
+ rc = wait_for_completion(&ep->re_done);
+ if (rc < 0)
+ goto out_destroy;
+ rc = ep->re_async_rc;
+ RPCRDMA_DEBUG(1, "xprt_rdma_connect: rdma_connect=%d\n", rc);
+ if (rc)
+ goto out_destroy;
+
+ for (i = 0; i < xp->maxrpc + xp->maxbck; i++) {
+ /* And post the receive buffer now. */
+ rc = rpcrdma_post_recv(ep, i);
+ if (rc) {
+ RPCRDMA_DEBUG(4, "rpcrdma_ep_create: ind=%d "
+ "rpcrdma_post_recv=%d\n", i, rc);
+ goto out_destroy;
+ }
+ }
+ return (rc);
+out_destroy:
+#ifdef notnow
+ ib_dealloc_pd(ep->re_pd);
+ ib_destroy_cq(ep->re_recv_cq);
+ ib_destroy_cq(ep->re_send_cq);
+ rdma_destroy_qp(ep->re_id);
+ rdma_destroy_id(ep->re_id);
+#endif
+out:
+ ep->re_id = NULL;
+ ep->re_pd = NULL;
+ if (rc < 0)
+ rc = -rc;
+ RPCRDMA_DEBUG(1, "xprt_rdma_connect: failed=%d\n", rc);
+ return (rc);
+}
+
+void
+xprt_rdma_disconnect(struct rpcrdma_xprt *xp)
+{
+ int rc;
+ struct rpcrdma_ep *ep;
+ struct rdma_cm_id *id;
+
+ mtx_lock(&xp->mtx);
+ ep = (struct rpcrdma_ep *)xp->ep;
+ if (ep == NULL) {
+ mtx_unlock(&xp->mtx);
+ RPCRDMA_DEBUG(1, "xprt_rdma_disconnect: Null ep\n");
+ return;
+ }
+ if (ep->re_id != NULL) {
+ id = ep->re_id;
+ ep->re_id = NULL;
+ mtx_unlock(&xp->mtx);
+ if (ep->re_connect_status != 0)
+ ep->re_connect_status = -ECONNRESET;
+ rpcrdma_free_ep(ep);
+
+ init_completion(&ep->re_done);
+ ep->re_async_rc = -ETIMEDOUT;
+ rc = rdma_disconnect(id);
+ if (rc) {
+ printf("xprt_rdma_disconnect: rdma_disconnect failed "
+ "rc=%d\n", rc);
+ return;
+ }
+ rc = wait_for_completion(&ep->re_done);
+ if (rc < 0) {
+ printf("xprt_rdma_disconnect: rdma_disconnect wait "
+ "failed rc=%d\n", rc);
+ return;
+ }
+ rc = ep->re_async_rc;
+ if (rc) {
+ printf("xprt_rdma_disconnect: rdma_disconnect "
+ "completion failed rc=%d\n", rc);
+ return;
+ }
+ RPCRDMA_DEBUG(1, "xprt_rdma_disconnect: Disconnected ep=%p\n",
+ ep);
+ } else {
+ mtx_unlock(&xp->mtx);
+ RPCRDMA_DEBUG(1, "xprt_rdma_disconnect: Null id\n");
+ }
+}
+
+static void
+rpcrdma_free_mr(struct rpcrdma_ep *ep, struct rpcrdma_regwr *regwr)
+{
+ const struct ib_send_wr *bad_wr;
+ int ret;
+
+ memset(&regwr->wr.wr, 0, sizeof(regwr->wr.wr));
+ regwr->wr.wr.opcode = IB_WR_LOCAL_INV;
+ regwr->wr.wr.ex.invalidate_rkey = regwr->wr.key;
+ RPCRDMA_DEBUG(4, "rpcrdma_free_mr: Invalidating key=0x%x\n",
+ regwr->wr.key);
+ regwr->wr.wr.send_flags = IB_SEND_SIGNALED;
+ regwr->wr.wr.wr_cqe = &regwr->cqe;
+ regwr->cqe.done = rpcrdma_send_done;
+ init_completion(&regwr->done);
+ regwr->async_rc = -ETIMEDOUT;
+ ret = ib_post_send(ep->re_id->qp, &regwr->wr.wr, &bad_wr);
+ if (ret != 0) {
+ printf("rpcrdma_free_mr: Failed to post send=%d\n", ret);
+ return;
+ }
+ ret = wait_for_completion(&regwr->done);
+ if (ret < 0 || regwr->async_rc != 0)
+ printf("rpcrdma_free_mr: Failed to complete ret=%d rc=%d\n",
+ ret, regwr->async_rc);
+}
+
+/* Unmap a chunk. */
+static void
+rpcrdma_unmap(struct rpcrdma_ep *ep, struct _rpcrdma_chunk_priv *chp)
+{
+ enum dma_data_direction dma_dir;
+ vm_page_t pg;
+ int i, j;
+ uint32_t plen;
+
+ dma_dir = (chp->ch.into_mem) ? DMA_FROM_DEVICE : DMA_TO_DEVICE;
+ plen = PAGE_SIZE;
+ RPCRDMA_DEBUG(4, "rpcrdma_unmap: Unmapping nseg=%d\n",
+ chp->ch.num_segment);
+ for (i = 0; i < chp->ch.num_segment; i++) {
+ rpcrdma_free_mr(ep, &chp->regwr[i]);
+ for (j = 0; j < chp->ch.sge_cnt[i]; j++) {
+ if (j + 1 == chp->ch.sge_cnt[i] &&
+ i + 1 == chp->ch.num_segment)
+ plen = chp->ch.last_len;
+ ib_dma_unmap_page(ep->re_id->device,
+ chp->sge[i][j].page_link, plen, dma_dir);
+ }
+ }
+ if (chp->rbpages == 0 && chp->ch.mextpg == NULL) {
+ RPCRDMA_DEBUG(4, "rpcrdma_unmap: Freeing anonymous pages\n");
+ for (i = 0; i < chp->ch.num_segment; i++) {
+ for (j = 0; j < chp->ch.sge_cnt[i]; j++) {
+ KASSERT((chp->sge[i][j].page_link &
+ SG_PAGE_LINK_MASK) == 0,
+ ("rpcrdma_unmap: page_link has mask"));
+ pg = (vm_page_t)chp->sge[i][j].page_link;
+ vm_page_unwire_noq(pg);
+ vm_page_free(pg);
+ }
+ }
+ }
+}
+
+/* Unmap a single chunk. */
+void
+xprt_rdma_unmap_chunk(struct rpcrdma_xprt *xp, struct rpcrdma_chunk *extern_chp)
+{
+ struct rpcrdma_ep *ep;
+ struct _rpcrdma_chunk_priv *chp;
+
+ ep = rpcrdma_acquire_ep(xp);
+
+ chp = (struct _rpcrdma_chunk_priv *)extern_chp;
+ if (ep != NULL)
+ rpcrdma_unmap(ep, chp);
+ kfree(chp);
+}
+
+void
+xprt_rdma_release_send(struct rpcrdma_xprt *xp, int ind,
+ struct rpcrdma_chunk *extern_chp, struct rpcrdma_chunk *extern_reply_chp,
+ struct rpcrdma_chunk *extern_request_chp)
+{
+ struct rpcrdma_ep *ep;
+ struct _rpcrdma_chunk_priv *chp;
+
+ ep = rpcrdma_acquire_ep(xp);
+ KASSERT(ind >= 0 && ind <= xp->maxrpc + xp->maxbck,
+ ("xprt_rdma_release_send: ind out of range"));
+ RPCRDMA_DEBUG(4, "xprt_rdma_release_send: ind=%d rwlist_chp=%p "
+ "reply_chp=%p request_chp=%p\n", ind, extern_chp, extern_reply_chp,
+ extern_request_chp);
+ if (extern_chp != NULL) {
+ chp = (struct _rpcrdma_chunk_priv *)extern_chp;
+ if (ep != NULL)
+ rpcrdma_unmap(ep, chp);
+ kfree(chp);
+ }
+ if (extern_reply_chp != NULL) {
+ chp = (struct _rpcrdma_chunk_priv *)extern_reply_chp;
+ if (ep != NULL)
+ rpcrdma_unmap(ep, chp);
+ kfree(chp);
+ }
+ if (extern_request_chp != NULL) {
+ chp = (struct _rpcrdma_chunk_priv *)extern_request_chp;
+ if (ep != NULL)
+ rpcrdma_unmap(ep, chp);
+ kfree(chp);
+ }
+ if (ep != NULL)
+ rpcrdma_release_buf(ep->re_xp, &ep->re_busy, ind);
+}
+
+int
+xprt_rdma_send(struct rpcrdma_xprt *xp, struct mbuf *mreq, int ind)
+{
+ struct rpcrdma_ep *ep;
+ const struct ib_send_wr *bad_wr;
+ int ret;
+
+ ep = rpcrdma_acquire_ep(xp);
+ RPCRDMA_DEBUG(1, "xprt_rdma_send: ep=%p\n", ep);
+ if (ep == NULL)
+ return (EPIPE);
+
+ RPCRDMA_DEBUG(1, "xprt_rdma_send: Send data len=%d ind=%d\n",
+ mreq->m_pkthdr.len, ind);
+ m_copydata(mreq, 0, mreq->m_pkthdr.len, ep->re_send[ind].buf);
+
+ /* Fill in the ib_send_wr. */
+ ep->re_send[ind].sge.length = mreq->m_pkthdr.len;
+ m_freem(mreq);
+
+#ifdef notnow
+ /* And post the receive buffer. */
+ ret = rpcrdma_post_recv(ep, ind);
+ if (ret)
+ return (EPIPE);
+#endif
+
+ init_completion(&ep->re_send[ind].done);
+ ep->re_send[ind].async_rc = -ETIMEDOUT;
+ ret = ib_post_send(ep->re_id->qp, &ep->re_send[ind].wr, &bad_wr);
+ RPCRDMA_DEBUG(1, "xprt_rdma_send: Post a send buffer\n");
+ if (ret) {
+ WARN_ONCE(ret, "failed to drain send queue: %d\n", ret);
+ return (EPIPE);
+ }
+ ret = wait_for_completion(&ep->re_send[ind].done);
+ if (ret < 0 || ep->re_send[ind].async_rc != 0) {
+ /* Should I release the buffer? */
+ WARN_ONCE(ret, "xprt_rdma_send: Send failed ret=%d async=%d\n",
+ ret, ep->re_send[ind].async_rc);
+ return (EPIPE);
+ }
+
+ return (0);
+}
+
+/*
+ * FRWR map a chunk.
+ */
+static int
+rpcrdma_map_chunk(struct rpcrdma_ep *ep, struct _rpcrdma_chunk_priv *chp,
+ bool map_seg)
+{
+ const struct ib_send_wr *badwr;
+ struct rpcrdma_regwr *regwr;
+ struct ib_mr *mr;
+ enum dma_data_direction dma_dir;
+ int i, j, k, l, ret;
+ uint32_t plen;
+ uint8_t key;
+
+ RPCRDMA_DEBUG(3, "rpcrdma_map_chunk: map_seg=%d\n", map_seg);
+ dma_dir = (chp->ch.into_mem) ? DMA_FROM_DEVICE : DMA_TO_DEVICE;
+
+ /* An ib_reg_wr has a ib_send_wr in it. */
+ plen = PAGE_SIZE;
+ for (i = 0; i < chp->ch.num_segment; i++) {
+ regwr = &chp->regwr[i];
+ if (chp->ch.mextpg != NULL)
+ mr = ep->re_reply[chp->ch.ind][i];
+ else
+ mr = ep->re_mr[chp->ch.ind][i];
+ RPCRDMA_DEBUG(4, "rpcrdma_map_chunk: ind=%d seg=%d mextpg=%p\n",
+ chp->ch.ind, i, chp->ch.mextpg);
+ memset(&regwr->wr.wr, 0, sizeof(regwr->wr.wr));
+
+ /* Map the sge(s) into bus dma space. */
+ if (map_seg) {
+ for (j = 0; j < chp->ch.sge_cnt[i]; j++) {
+ if (j + 1 == chp->ch.sge_cnt[i] &&
+ i + 1 == chp->ch.num_segment)
+ plen = chp->ch.last_len;
+ for (k = 0; k < 1000; k++) {
+ sg_dma_address(&chp->sge[i][j]) =
+ ib_dma_map_page(ep->re_id->device,
+ (vm_page_t)chp->sge[i][j].page_link,
+ 0, plen, dma_dir);
+ if (ib_dma_mapping_error(
+ ep->re_id->device,
+ sg_dma_address(&chp->sge[i][j]))==0)
+ break;
+ printf("rpcrdma_map_chunk: ib dma "
+ "mapping failed, will be "
+ "retried\n");
+ }
+ if (k == 1000)
+ goto reg_err;
+ }
+ RPCRDMA_DEBUG(4, "rpcrdma_map_chunk: mapped seg%d "
+ "sgecnt=%d\n", i, chp->ch.sge_cnt[i]);
+ ret = ib_map_mr_sg_zbva(mr, &chp->sge[i][0],
+ chp->ch.sge_cnt[i], NULL, PAGE_SIZE);
+ RPCRDMA_DEBUG(4, "rpcrdma_map_chunk: ib_map_mr_sg_zbva"
+ " ret=%d\n", ret);
+ if (ret != chp->ch.sge_cnt[i]) {
+ RPCRDMA_DEBUG(4, "rpcrdma_map_chunk: "
+ "ib_map_mr_sg_zbva failed sgecnt=%d "
+ "ret=%d\n", chp->ch.sge_cnt[i], ret);
+ goto reg_err;
+ }
+ }
+
+ /* Set up keys, etc. */
+ key = mr->rkey;
+ ib_update_fast_reg_key(mr, ++key);
+ chp->ch.handle[i] = mr->rkey;
+ chp->ch.offset[i] = mr->iova;
+ RPCRDMA_DEBUG(4, "rpcrdma_map_chunk: frwr handle=0x%x "
+ "offset=0x%jx length=%d\n", chp->ch.handle[i],
+ (uintmax_t)chp->ch.offset[i], chp->ch.length[i]);
+
+ /* Register this mr. */
+ regwr->wr.wr.next = NULL;
+ regwr->wr.wr.wr_cqe = &regwr->cqe;
+ regwr->cqe.done = rpcrdma_send_done;
+ regwr->wr.wr.opcode = IB_WR_REG_MR;
+ regwr->wr.mr = mr;
+ regwr->wr.key = mr->rkey;
+ if (chp->ch.into_mem)
+ regwr->wr.access = IB_ACCESS_REMOTE_WRITE |
+ IB_ACCESS_LOCAL_WRITE;
+ else
+ regwr->wr.access = IB_ACCESS_REMOTE_READ;
+ regwr->wr.wr.send_flags = IB_SEND_SIGNALED;
+ init_completion(&regwr->done);
+ regwr->async_rc = -ETIMEDOUT;
+ ret = ib_post_send(ep->re_id->qp, &regwr->wr.wr, &badwr);
+ RPCRDMA_DEBUG(4, "rpcrdma_map_chunk: ib_post_send=%d\n", ret);
+ if (ret != 0)
+ goto reg_err;
+ }
+
+ /* Now, wait for all the registration(s) to complete. */
+ for (j = 0; j < chp->ch.num_segment; j++) {
+ regwr = &chp->regwr[j];
+ ret = wait_for_completion(&regwr->done);
+ if (ret < 0 || regwr->async_rc != 0) {
+ RPCRDMA_DEBUG(4, "rpcrdma_map_chunk: frwr failed "
+ "ret=%d async_rc=%d\n", ret, regwr->async_rc);
+ goto reg_err;
+ }
+ }
+ RPCRDMA_DEBUG(1, "rpcrdma_map_chunk: successful\n");
+ return (0);
+reg_err:
+ if (map_seg) {
+ /* 0 to i - 1 were mapped. */
+ for (k = 0; k < i; k++) {
+ for (l = 0; l < j; l++) {
+ if (l + 1 == j && k + 1 == i)
+ ib_dma_unmap_page(ep->re_id->device,
+ sg_dma_address(&chp->sge[k][l]),
+ plen, dma_dir);
+ else
+ ib_dma_unmap_page(ep->re_id->device,
+ sg_dma_address(&chp->sge[k][l]),
+ PAGE_SIZE, dma_dir);
+ }
+ }
+ }
+ RPCRDMA_DEBUG(1, "rpcrdma_map_chunk: failed\n");
+ return (-1);
+}
+
+/*
+ * Create a chunk for RDMA, dma map it and associate it with an MR.
+ * The chunk can be created from pages from any of these sources:
+ * mextpg - If not NULL, it is a list of M_EXTPG mbufs with pages in them.
+ * rb - If not NULL, this structure has a list of pages in it.
+ * num_pg - If both of the above are NULL, allocate numpg anonymous pages.
+ */
+struct rpcrdma_chunk *
+xprt_rdma_create_chunk(struct rpcrdma_xprt *xp, uint32_t num_pg,
+ struct rpcrdma_reduce_pg *rb, struct mbuf *mextpg, bool into_mem, int ind)
+{
+ struct _rpcrdma_chunk_priv *chp;
+ struct rpcrdma_ep *ep;
+ struct mbuf *m2;
+ vm_page_t pg;
+ uint32_t plen;
+ int i, j, k;
+
+ ep = rpcrdma_acquire_ep(xp);
+ if (ep == NULL)
+ return (NULL);
+
+ RPCRDMA_DEBUG(1, "xprt_rdma_create_chunk: num_pg=%d rb=%p mextpg=%p "
+ "into_mem=%d ind=%d\n", num_pg, rb, mextpg, into_mem, ind);
+ chp = kzalloc(sizeof(*chp), GFP_KERNEL);
+ chp->ch.ind = ind;
+
+ /* Fill in the scatter list(s). */
+ if (rb != NULL) {
+ chp->ch.last_len = rb->len % PAGE_SIZE;
+ if (chp->ch.last_len == 0)
+ chp->ch.last_len = PAGE_SIZE;
+ chp->ch.first_off = 0;
+ chp->ch.into_mem = rb->into_mem;
+ /* Fill in the scatter list(s) from the page vector. */
+ for (i = 0, j = 0, k = 0; k < rb->npg; k++) {
+ if (k + 1 == rb->npg)
+ plen = chp->ch.last_len;
+ else
+ plen = PAGE_SIZE;
+ sg_set_page(&chp->sge[i][j], rb->pg[k], plen, 0);
+ chp->ch.length[i] += plen;
+ if (++j == xp->maxsge) {
+ chp->ch.sge_cnt[i] = j;
+ j = 0;
+ i++;
+ }
+ }
+ if (j != 0) {
+ chp->ch.sge_cnt[i] = j;
+ i++;
+ }
+ chp->ch.num_segment = i;
+ chp->rbpages = rb->npg;
+ RPCRDMA_DEBUG(3, "xprt_rdma_create_chunk: num_seg=%d "
+ "rb->npg=%d\n", i, rb->npg);
+ } else if (mextpg != NULL) {
+ for (i = 0, j = 0, m2 = mextpg; m2 != NULL; m2 = m2->m_next) {
+ KASSERT((m2->m_flags & M_EXTPG) != 0,
+ ("clnt_rmda_create_chunk: non-m_extpg mbuf"));
+ if (mextpg == m2)
+ chp->ch.first_off = m2->m_epg_1st_off;
+ plen = PAGE_SIZE;
+ for (k = 0; k < m2->m_epg_npgs; k++) {
+ if (m2->m_next == NULL &&
+ k == m2->m_epg_npgs - 1) {
+ chp->ch.last_len = plen =
+ m2->m_epg_last_len;
+ }
+ if (k == 0)
+ chp->ch.length[i] += (plen -
+ chp->ch.first_off);
+ else
+ chp->ch.length[i] += plen;
+ pg = PHYS_TO_VM_PAGE(m2->m_epg_pa[k]);
+ sg_set_page(&chp->sge[i][j], pg, plen, 0);
+ if (++j == xp->maxsge) {
+ chp->ch.sge_cnt[i] = j;
+ j = 0;
+ i++;
+ }
+ plen = PAGE_SIZE;
+ }
+ }
+ if (j != 0) {
+ chp->ch.sge_cnt[i] = j;
+ i++;
+ }
+ chp->ch.num_segment = i;
+ chp->ch.into_mem = (into_mem) ? TRUE : FALSE;
+ chp->ch.mextpg = mextpg;
+ RPCRDMA_DEBUG(3, "xprt_rdma_create_chunk: num_seg=%d "
+ "mextpg=%p into_mem=%d\n", i, mextpg, into_mem);
+ } else {
+ chp->ch.last_len = PAGE_SIZE;
+ chp->ch.first_off = 0;
+ chp->ch.into_mem = (into_mem) ? TRUE : FALSE;
+ /* Fill in the scatter list(s) from the page vector. */
+ for (i = 0, j = 0; i * xp->maxsge + j < num_pg; ) {
+ pg = vm_page_alloc_noobj(VM_ALLOC_WAITOK |
+ VM_ALLOC_NODUMP | VM_ALLOC_WIRED);
+ sg_set_page(&chp->sge[i][j], pg, PAGE_SIZE, 0);
+ chp->ch.length[i] += PAGE_SIZE;
+ if (++j == xp->maxsge) {
+ chp->ch.sge_cnt[i] = j;
+ j = 0;
+ i++;
+ }
+ }
+ if (j != 0) {
+ chp->ch.sge_cnt[i] = j;
+ i++;
+ }
+ chp->ch.num_segment = i;
+ RPCRDMA_DEBUG(3, "xprt_rdma_create_chunk: num_seg=%d "
+ "num_pg=%d\n", i, num_pg);
+ }
+
+ i = rpcrdma_map_chunk(ep, chp, true);
+ RPCRDMA_DEBUG(3, "xprt_rdma_create_chunk: rpcrdma_map_chunk=%d\n", i);
+ if (i != 0)
+ goto errout;
+ /*
+ * Return with a reference count on ep, which will be released
+ * by xprt_rdma_release_send().
+ */
+ RPCRDMA_DEBUG(1, "xprt_rdma_create_chunk: succeeded\n");
+ return (&chp->ch);
+errout:
+ RPCRDMA_DEBUG(1, "xprt_rdma_create_chunk: failed\n");
+ rpcrdma_release_buf(ep->re_xp, &ep->re_busy, chp->ch.ind);
+ if (mextpg != NULL)
+ m_freem(mextpg);
+ if (rb == NULL && mextpg == NULL) {
+ for (i = 0, k = 0; i < chp->ch.num_segment && k < num_pg; i++) {
+ for (j = 0; j < chp->ch.sge_cnt[i] && k < num_pg; j++,
+ k++) {
+ pg = (vm_page_t)chp->sge[i][j].page_link;
+ vm_page_unwire_noq(pg);
+ vm_page_free(pg);
+ }
+ }
+ }
+ kfree(chp);
+ return (NULL);
+}
+
+/* Check the re_connect_status. */
+int
+xprt_rdma_disconnected(struct rpcrdma_xprt *xp)
+{
+ struct rpcrdma_ep *ep;
+
+ ep = (struct rpcrdma_ep *)xp->ep;
+ if (ep == NULL || ep->re_connect_status != 0)
+ return (ECONNABORTED);
+ return (0);
+}
+

File Metadata

Mime Type
text/plain
Expires
Wed, Sep 9, 8:37 PM (6 h, 9 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
38462506
Default Alt Text
D59278.id185964.diff (90 KB)

Event Timeline