* [PATCH v1 01/22] svcrdma: Another sendto chunk list parsing update
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
@ 2017-01-07 17:15 ` Chuck Lever
2017-01-07 17:15 ` [PATCH v1 02/22] svcrdma: Replace RPCRDMA_SQ_DEPTH_MULT Chuck Lever
` (23 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:15 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Commit 5fdca6531434 ("svcrdma: Renovate sendto chunk list parsing")
missed a spot. svc_rdma_xdr_get_reply_hdr_len() also assumes the
Write list has only Write chunk. There's no harm in making this code
more general.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/rpc_rdma.h | 9 ++++++
include/linux/sunrpc/svc_rdma.h | 2 +
net/sunrpc/xprtrdma/svc_rdma_marshal.c | 49 +++++++++++++++++---------------
net/sunrpc/xprtrdma/svc_rdma_sendto.c | 3 +-
4 files changed, 38 insertions(+), 25 deletions(-)
diff --git a/include/linux/sunrpc/rpc_rdma.h b/include/linux/sunrpc/rpc_rdma.h
index cfda6ad..245fc59 100644
--- a/include/linux/sunrpc/rpc_rdma.h
+++ b/include/linux/sunrpc/rpc_rdma.h
@@ -110,6 +110,15 @@ struct rpcrdma_msg {
};
/*
+ * XDR sizes, in quads
+ */
+enum {
+ rpcrdma_fixed_maxsz = 4,
+ rpcrdma_segment_maxsz = 4,
+ rpcrdma_readchunk_maxsz = 2 + rpcrdma_segment_maxsz,
+};
+
+/*
* Smallest RPC/RDMA header: rm_xid through rm_type, then rm_nochunks
*/
#define RPCRDMA_HDRLEN_MIN (sizeof(__be32) * 7)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 757fb96..551c518 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -218,7 +218,7 @@ extern void svc_rdma_xdr_encode_reply_header(struct svcxprt_rdma *,
struct rpcrdma_msg *,
struct rpcrdma_msg *,
enum rpcrdma_proc);
-extern int svc_rdma_xdr_get_reply_hdr_len(struct rpcrdma_msg *);
+extern unsigned int svc_rdma_xdr_get_reply_hdr_len(__be32 *rdma_resp);
/* svc_rdma_recvfrom.c */
extern int svc_rdma_recvfrom(struct svc_rqst *);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_marshal.c b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
index 0ba9887..4e72034 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_marshal.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
@@ -260,32 +260,35 @@ int svc_rdma_xdr_encode_error(struct svcxprt_rdma *xprt,
return (int)((unsigned long)va - (unsigned long)startp);
}
-int svc_rdma_xdr_get_reply_hdr_len(struct rpcrdma_msg *rmsgp)
+/**
+ * svc_rdma_xdr_get_reply_hdr_length - Get length of Reply transport header
+ * @rdma_resp: buffer containing Reply transport header
+ *
+ * Returns length of transport header, in bytes.
+ */
+unsigned int svc_rdma_xdr_get_reply_hdr_len(__be32 *rdma_resp)
{
- struct rpcrdma_write_array *wr_ary;
+ unsigned int nsegs;
+ __be32 *p;
- /* There is no read-list in a reply */
+ p = rdma_resp;
- /* skip write list */
- wr_ary = (struct rpcrdma_write_array *)
- &rmsgp->rm_body.rm_chunks[1];
- if (wr_ary->wc_discrim)
- wr_ary = (struct rpcrdma_write_array *)
- &wr_ary->wc_array[be32_to_cpu(wr_ary->wc_nchunks)].
- wc_target.rs_length;
- else
- wr_ary = (struct rpcrdma_write_array *)
- &wr_ary->wc_nchunks;
-
- /* skip reply array */
- if (wr_ary->wc_discrim)
- wr_ary = (struct rpcrdma_write_array *)
- &wr_ary->wc_array[be32_to_cpu(wr_ary->wc_nchunks)];
- else
- wr_ary = (struct rpcrdma_write_array *)
- &wr_ary->wc_nchunks;
-
- return (unsigned long) wr_ary - (unsigned long) rmsgp;
+ /* RPC-over-RDMA V1 replies never have a Read list. */
+ p += rpcrdma_fixed_maxsz + 1;
+
+ /* Skip Write list. */
+ while (*p++ != xdr_zero) {
+ nsegs = be32_to_cpup(p++);
+ p += nsegs * rpcrdma_segment_maxsz;
+ }
+
+ /* Skip Reply chunk. */
+ if (*p++ != xdr_zero) {
+ nsegs = be32_to_cpup(p++);
+ p += nsegs * rpcrdma_segment_maxsz;
+ }
+
+ return (unsigned long)p - (unsigned long)rdma_resp;
}
void svc_rdma_xdr_encode_write_list(struct rpcrdma_msg *rmsgp, int chunks)
diff --git a/net/sunrpc/xprtrdma/svc_rdma_sendto.c b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
index ad4d286..ba76f1617 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_sendto.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
@@ -476,7 +476,8 @@ static int send_reply(struct svcxprt_rdma *rdma,
/* Prepare the SGE for the RPCRDMA Header */
ctxt->sge[0].lkey = rdma->sc_pd->local_dma_lkey;
- ctxt->sge[0].length = svc_rdma_xdr_get_reply_hdr_len(rdma_resp);
+ ctxt->sge[0].length =
+ svc_rdma_xdr_get_reply_hdr_len((__be32 *)rdma_resp);
ctxt->sge[0].addr =
ib_dma_map_page(rdma->sc_cm_id->device, page, 0,
ctxt->sge[0].length, DMA_TO_DEVICE);
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 02/22] svcrdma: Replace RPCRDMA_SQ_DEPTH_MULT
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
2017-01-07 17:15 ` [PATCH v1 01/22] svcrdma: Another sendto chunk list parsing update Chuck Lever
@ 2017-01-07 17:15 ` Chuck Lever
2017-01-07 17:15 ` [PATCH v1 03/22] svcrdma: Remove unused sc_dto_q field Chuck Lever
` (22 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:15 UTC (permalink / raw)
To: linux-rdma, linux-nfs
The SQ depth is currently computed using a fixed multipler. For
some configurations this underestimates the needed number of SQEs
and CQEs. Usually that means the server has to pause on occasion
for SQEs to become available before it can send RDMA Reads or RPC
Replies.
There might be some cases where the new estimator generates a SQ
depth that is larger than the local HCA can support. If that is a
frequent problem, then a mechanism can be introduced that
automatically reduces the number of RPC-over-RDMA credits per
connection.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 1 -
net/sunrpc/xprtrdma/svc_rdma.c | 2 -
net/sunrpc/xprtrdma/svc_rdma_transport.c | 44 +++++++++++++++++++++++++++++-
3 files changed, 43 insertions(+), 4 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 551c518..cb3d87a 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -182,7 +182,6 @@ struct svcxprt_rdma {
/* The default ORD value is based on two outstanding full-size writes with a
* page size of 4k, or 32k * 2 ops / 4k = 16 outstanding RDMA_READ. */
#define RPCRDMA_ORD (64/4)
-#define RPCRDMA_SQ_DEPTH_MULT 8
#define RPCRDMA_MAX_REQUESTS 32
#define RPCRDMA_MAX_REQ_SIZE 4096
diff --git a/net/sunrpc/xprtrdma/svc_rdma.c b/net/sunrpc/xprtrdma/svc_rdma.c
index c846ca9..9124441 100644
--- a/net/sunrpc/xprtrdma/svc_rdma.c
+++ b/net/sunrpc/xprtrdma/svc_rdma.c
@@ -247,8 +247,6 @@ int svc_rdma_init(void)
dprintk("SVCRDMA Module Init, register RPC RDMA transport\n");
dprintk("\tsvcrdma_ord : %d\n", svcrdma_ord);
dprintk("\tmax_requests : %u\n", svcrdma_max_requests);
- dprintk("\tsq_depth : %u\n",
- svcrdma_max_requests * RPCRDMA_SQ_DEPTH_MULT);
dprintk("\tmax_bc_requests : %u\n", svcrdma_max_bc_requests);
dprintk("\tmax_inline : %d\n", svcrdma_max_req_size);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index ca2799a..f246197 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -950,6 +950,48 @@ void svc_rdma_put_frmr(struct svcxprt_rdma *rdma,
}
}
+static unsigned int svc_rdma_read_sqes_per_credit(struct svcxprt_rdma *newxprt)
+{
+ struct ib_device_attr *attrs = &newxprt->sc_cm_id->device->attrs;
+
+ if (!(attrs->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
+ return DIV_ROUND_UP(RPCSVC_MAXPAGES, newxprt->sc_max_sge_rd);
+
+ /* FRWR: reg, read, inv */
+ return DIV_ROUND_UP(RPCSVC_MAXPAGES,
+ attrs->max_fast_reg_page_list_len) * 3;
+}
+
+static unsigned int svc_rdma_write_sqes_per_credit(struct svcxprt_rdma *newxprt)
+{
+ return DIV_ROUND_UP(RPCSVC_MAXPAGES, newxprt->sc_max_sge);
+}
+
+static unsigned int svc_rdma_sq_depth(struct svcxprt_rdma *newxprt)
+{
+ unsigned int sqes_per_credit;
+
+ /* Estimate SQEs per credit assuming a full Read chunk payload
+ * and a full Write chunk payload (possible with krb5i/p). Each
+ * credit will consume Read WRs then Write WRs, serially, so
+ * we need just the larger of the two, not the sum.
+ *
+ * This is not an upper bound. Clients can break chunks into
+ * arbitrarily many segments. However, if more SQEs are needed
+ * then are available, the server has Send Queue accounting to
+ * wait until enough SQEs are ready. But we want that waiting
+ * to be very rare.
+ */
+ sqes_per_credit = max_t(unsigned int,
+ svc_rdma_read_sqes_per_credit(newxprt),
+ svc_rdma_write_sqes_per_credit(newxprt));
+
+ /* RDMA Sends per credit */
+ sqes_per_credit += 1;
+
+ return sqes_per_credit * newxprt->sc_rq_depth;
+}
+
/*
* This is the xpo_recvfrom function for listening endpoints. Its
* purpose is to accept incoming connections. The CMA callback handler
@@ -1006,7 +1048,7 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
svcrdma_max_bc_requests);
newxprt->sc_rq_depth = newxprt->sc_max_requests +
newxprt->sc_max_bc_requests;
- newxprt->sc_sq_depth = RPCRDMA_SQ_DEPTH_MULT * newxprt->sc_rq_depth;
+ newxprt->sc_sq_depth = svc_rdma_sq_depth(newxprt);
atomic_set(&newxprt->sc_sq_avail, newxprt->sc_sq_depth);
if (!svc_rdma_prealloc_ctxts(newxprt))
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 03/22] svcrdma: Remove unused sc_dto_q field
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
2017-01-07 17:15 ` [PATCH v1 01/22] svcrdma: Another sendto chunk list parsing update Chuck Lever
2017-01-07 17:15 ` [PATCH v1 02/22] svcrdma: Replace RPCRDMA_SQ_DEPTH_MULT Chuck Lever
@ 2017-01-07 17:15 ` Chuck Lever
2017-01-07 17:15 ` [PATCH v1 04/22] svcrdma: Combine list fields in struct svc_rdma_op_ctxt Chuck Lever
` (21 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:15 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Clean up. Commit be99bb11400c ('svcrdma: Use new CQ API for
RPC-over-RDMA server send CQs') removed code that used the sc_dto_q
field, but neglected to remove sc_dto_q at the same time.
Fixes: be99bb11400c ('svcrdma: Use new CQ API for RPC-over- ...')
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 1 -
net/sunrpc/xprtrdma/svc_rdma_transport.c | 1 -
2 files changed, 2 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index cb3d87a..8dc13d5 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -171,7 +171,6 @@ struct svcxprt_rdma {
wait_queue_head_t sc_send_wait; /* SQ exhaustion waitlist */
unsigned long sc_flags;
- struct list_head sc_dto_q; /* DTO tasklet I/O pending Q */
struct list_head sc_read_complete_q;
struct work_struct sc_work;
};
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index f246197..44bb337 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -557,7 +557,6 @@ static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
return NULL;
svc_xprt_init(&init_net, &svc_rdma_class, &cma_xprt->sc_xprt, serv);
INIT_LIST_HEAD(&cma_xprt->sc_accept_q);
- INIT_LIST_HEAD(&cma_xprt->sc_dto_q);
INIT_LIST_HEAD(&cma_xprt->sc_rq_dto_q);
INIT_LIST_HEAD(&cma_xprt->sc_read_complete_q);
INIT_LIST_HEAD(&cma_xprt->sc_frmr_q);
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 04/22] svcrdma: Combine list fields in struct svc_rdma_op_ctxt
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (2 preceding siblings ...)
2017-01-07 17:15 ` [PATCH v1 03/22] svcrdma: Remove unused sc_dto_q field Chuck Lever
@ 2017-01-07 17:15 ` Chuck Lever
2017-01-07 17:15 ` [PATCH v1 05/22] svcrdma: Poll CQs in "workqueue" mode Chuck Lever
` (20 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:15 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Clean up: The free list and the dto_q list fields are never used at
the same time. Reduce the size of struct svc_rdma_op_ctxt by
combining these fields.
Note that commit be99bb11400c ('svcrdma: Use new CQ API for
RPC-over-RDMA server send CQs') removed code that used the sc_dto_q
field, but neglected to remove sc_dto_q at the same time.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 3 +--
net/sunrpc/xprtrdma/svc_rdma_recvfrom.c | 14 +++++--------
net/sunrpc/xprtrdma/svc_rdma_transport.c | 33 ++++++++++++++----------------
3 files changed, 22 insertions(+), 28 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 8dc13d5..997bc78 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -70,7 +70,7 @@
* completes.
*/
struct svc_rdma_op_ctxt {
- struct list_head free;
+ struct list_head list;
struct svc_rdma_op_ctxt *read_hdr;
struct svc_rdma_fastreg_mr *frmr;
int hdr_count;
@@ -78,7 +78,6 @@ struct svc_rdma_op_ctxt {
struct ib_cqe cqe;
struct ib_cqe reg_cqe;
struct ib_cqe inv_cqe;
- struct list_head dto_q;
u32 byte_len;
u32 position;
struct svcxprt_rdma *xprt;
diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index 57d35fb..3a72360 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -610,18 +610,16 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
spin_lock_bh(&rdma_xprt->sc_rq_dto_lock);
if (!list_empty(&rdma_xprt->sc_read_complete_q)) {
- ctxt = list_entry(rdma_xprt->sc_read_complete_q.next,
- struct svc_rdma_op_ctxt,
- dto_q);
- list_del_init(&ctxt->dto_q);
+ ctxt = list_first_entry(&rdma_xprt->sc_read_complete_q,
+ struct svc_rdma_op_ctxt, list);
+ list_del(&ctxt->list);
spin_unlock_bh(&rdma_xprt->sc_rq_dto_lock);
rdma_read_complete(rqstp, ctxt);
goto complete;
} else if (!list_empty(&rdma_xprt->sc_rq_dto_q)) {
- ctxt = list_entry(rdma_xprt->sc_rq_dto_q.next,
- struct svc_rdma_op_ctxt,
- dto_q);
- list_del_init(&ctxt->dto_q);
+ ctxt = list_first_entry(&rdma_xprt->sc_rq_dto_q,
+ struct svc_rdma_op_ctxt, list);
+ list_del(&ctxt->list);
} else {
atomic_inc(&rdma_stat_rq_starve);
clear_bit(XPT_DATA, &xprt->xpt_flags);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index 44bb337..eca5cec 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -157,8 +157,7 @@ static struct svc_rdma_op_ctxt *alloc_ctxt(struct svcxprt_rdma *xprt,
ctxt = kmalloc(sizeof(*ctxt), flags);
if (ctxt) {
ctxt->xprt = xprt;
- INIT_LIST_HEAD(&ctxt->free);
- INIT_LIST_HEAD(&ctxt->dto_q);
+ INIT_LIST_HEAD(&ctxt->list);
}
return ctxt;
}
@@ -180,7 +179,7 @@ static bool svc_rdma_prealloc_ctxts(struct svcxprt_rdma *xprt)
dprintk("svcrdma: No memory for RDMA ctxt\n");
return false;
}
- list_add(&ctxt->free, &xprt->sc_ctxts);
+ list_add(&ctxt->list, &xprt->sc_ctxts);
}
return true;
}
@@ -195,8 +194,8 @@ struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *xprt)
goto out_empty;
ctxt = list_first_entry(&xprt->sc_ctxts,
- struct svc_rdma_op_ctxt, free);
- list_del_init(&ctxt->free);
+ struct svc_rdma_op_ctxt, list);
+ list_del(&ctxt->list);
spin_unlock_bh(&xprt->sc_ctxt_lock);
out:
@@ -256,7 +255,7 @@ void svc_rdma_put_context(struct svc_rdma_op_ctxt *ctxt, int free_pages)
spin_lock_bh(&xprt->sc_ctxt_lock);
xprt->sc_ctxt_used--;
- list_add(&ctxt->free, &xprt->sc_ctxts);
+ list_add(&ctxt->list, &xprt->sc_ctxts);
spin_unlock_bh(&xprt->sc_ctxt_lock);
}
@@ -266,8 +265,8 @@ static void svc_rdma_destroy_ctxts(struct svcxprt_rdma *xprt)
struct svc_rdma_op_ctxt *ctxt;
ctxt = list_first_entry(&xprt->sc_ctxts,
- struct svc_rdma_op_ctxt, free);
- list_del(&ctxt->free);
+ struct svc_rdma_op_ctxt, list);
+ list_del(&ctxt->list);
kfree(ctxt);
}
}
@@ -404,7 +403,7 @@ static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
/* All wc fields are now known to be valid */
ctxt->byte_len = wc->byte_len;
spin_lock(&xprt->sc_rq_dto_lock);
- list_add_tail(&ctxt->dto_q, &xprt->sc_rq_dto_q);
+ list_add_tail(&ctxt->list, &xprt->sc_rq_dto_q);
spin_unlock(&xprt->sc_rq_dto_lock);
set_bit(XPT_DATA, &xprt->sc_xprt.xpt_flags);
@@ -525,7 +524,7 @@ void svc_rdma_wc_read(struct ib_cq *cq, struct ib_wc *wc)
read_hdr = ctxt->read_hdr;
spin_lock(&xprt->sc_rq_dto_lock);
- list_add_tail(&read_hdr->dto_q,
+ list_add_tail(&read_hdr->list,
&xprt->sc_read_complete_q);
spin_unlock(&xprt->sc_rq_dto_lock);
@@ -1254,20 +1253,18 @@ static void __svc_rdma_free(struct work_struct *work)
*/
while (!list_empty(&rdma->sc_read_complete_q)) {
struct svc_rdma_op_ctxt *ctxt;
- ctxt = list_entry(rdma->sc_read_complete_q.next,
- struct svc_rdma_op_ctxt,
- dto_q);
- list_del_init(&ctxt->dto_q);
+ ctxt = list_first_entry(&rdma->sc_read_complete_q,
+ struct svc_rdma_op_ctxt, list);
+ list_del(&ctxt->list);
svc_rdma_put_context(ctxt, 1);
}
/* Destroy queued, but not processed recv completions */
while (!list_empty(&rdma->sc_rq_dto_q)) {
struct svc_rdma_op_ctxt *ctxt;
- ctxt = list_entry(rdma->sc_rq_dto_q.next,
- struct svc_rdma_op_ctxt,
- dto_q);
- list_del_init(&ctxt->dto_q);
+ ctxt = list_first_entry(&rdma->sc_rq_dto_q,
+ struct svc_rdma_op_ctxt, list);
+ list_del(&ctxt->list);
svc_rdma_put_context(ctxt, 1);
}
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 05/22] svcrdma: Poll CQs in "workqueue" mode
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (3 preceding siblings ...)
2017-01-07 17:15 ` [PATCH v1 04/22] svcrdma: Combine list fields in struct svc_rdma_op_ctxt Chuck Lever
@ 2017-01-07 17:15 ` Chuck Lever
2017-01-07 17:16 ` [PATCH v1 06/22] svcrdma: Clean up RPC-over-RDMA Reply header encoder Chuck Lever
` (19 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:15 UTC (permalink / raw)
To: linux-rdma, linux-nfs
svcrdma calls svc_xprt_put() in its completion handlers, which
currently run in IRQ context.
However, svc_xprt_put() is meant to be invoked in process context,
not in IRQ context. After the last transport reference is gone, it
directly calls a transport release function that expects to run in
process context.
Change the CQ polling modes to IB_POLL_WORKQUEUE so that svcrdma
invokes svc_xprt_put() only in process context. As an added benefit,
bottom half-disabled spin locking can be eliminated from I/O paths.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
net/sunrpc/xprtrdma/svc_rdma_recvfrom.c | 6 +++---
net/sunrpc/xprtrdma/svc_rdma_transport.c | 26 +++++++++++++-------------
2 files changed, 16 insertions(+), 16 deletions(-)
diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index 3a72360..776e1ed 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -608,12 +608,12 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
dprintk("svcrdma: rqstp=%p\n", rqstp);
- spin_lock_bh(&rdma_xprt->sc_rq_dto_lock);
+ spin_lock(&rdma_xprt->sc_rq_dto_lock);
if (!list_empty(&rdma_xprt->sc_read_complete_q)) {
ctxt = list_first_entry(&rdma_xprt->sc_read_complete_q,
struct svc_rdma_op_ctxt, list);
list_del(&ctxt->list);
- spin_unlock_bh(&rdma_xprt->sc_rq_dto_lock);
+ spin_unlock(&rdma_xprt->sc_rq_dto_lock);
rdma_read_complete(rqstp, ctxt);
goto complete;
} else if (!list_empty(&rdma_xprt->sc_rq_dto_q)) {
@@ -625,7 +625,7 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
clear_bit(XPT_DATA, &xprt->xpt_flags);
ctxt = NULL;
}
- spin_unlock_bh(&rdma_xprt->sc_rq_dto_lock);
+ spin_unlock(&rdma_xprt->sc_rq_dto_lock);
if (!ctxt) {
/* This is the EAGAIN path. The svc_recv routine will
* return -EAGAIN, the nfsd thread will go to call into
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index eca5cec..4d34afb 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -188,7 +188,7 @@ struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *xprt)
{
struct svc_rdma_op_ctxt *ctxt = NULL;
- spin_lock_bh(&xprt->sc_ctxt_lock);
+ spin_lock(&xprt->sc_ctxt_lock);
xprt->sc_ctxt_used++;
if (list_empty(&xprt->sc_ctxts))
goto out_empty;
@@ -196,7 +196,7 @@ struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *xprt)
ctxt = list_first_entry(&xprt->sc_ctxts,
struct svc_rdma_op_ctxt, list);
list_del(&ctxt->list);
- spin_unlock_bh(&xprt->sc_ctxt_lock);
+ spin_unlock(&xprt->sc_ctxt_lock);
out:
ctxt->count = 0;
@@ -208,15 +208,15 @@ struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *xprt)
/* Either pre-allocation missed the mark, or send
* queue accounting is broken.
*/
- spin_unlock_bh(&xprt->sc_ctxt_lock);
+ spin_unlock(&xprt->sc_ctxt_lock);
ctxt = alloc_ctxt(xprt, GFP_NOIO);
if (ctxt)
goto out;
- spin_lock_bh(&xprt->sc_ctxt_lock);
+ spin_lock(&xprt->sc_ctxt_lock);
xprt->sc_ctxt_used--;
- spin_unlock_bh(&xprt->sc_ctxt_lock);
+ spin_unlock(&xprt->sc_ctxt_lock);
WARN_ONCE(1, "svcrdma: empty RDMA ctxt list?\n");
return NULL;
}
@@ -253,10 +253,10 @@ void svc_rdma_put_context(struct svc_rdma_op_ctxt *ctxt, int free_pages)
for (i = 0; i < ctxt->count; i++)
put_page(ctxt->pages[i]);
- spin_lock_bh(&xprt->sc_ctxt_lock);
+ spin_lock(&xprt->sc_ctxt_lock);
xprt->sc_ctxt_used--;
list_add(&ctxt->list, &xprt->sc_ctxts);
- spin_unlock_bh(&xprt->sc_ctxt_lock);
+ spin_unlock(&xprt->sc_ctxt_lock);
}
static void svc_rdma_destroy_ctxts(struct svcxprt_rdma *xprt)
@@ -921,14 +921,14 @@ struct svc_rdma_fastreg_mr *svc_rdma_get_frmr(struct svcxprt_rdma *rdma)
{
struct svc_rdma_fastreg_mr *frmr = NULL;
- spin_lock_bh(&rdma->sc_frmr_q_lock);
+ spin_lock(&rdma->sc_frmr_q_lock);
if (!list_empty(&rdma->sc_frmr_q)) {
frmr = list_entry(rdma->sc_frmr_q.next,
struct svc_rdma_fastreg_mr, frmr_list);
list_del_init(&frmr->frmr_list);
frmr->sg_nents = 0;
}
- spin_unlock_bh(&rdma->sc_frmr_q_lock);
+ spin_unlock(&rdma->sc_frmr_q_lock);
if (frmr)
return frmr;
@@ -941,10 +941,10 @@ void svc_rdma_put_frmr(struct svcxprt_rdma *rdma,
if (frmr) {
ib_dma_unmap_sg(rdma->sc_cm_id->device,
frmr->sg, frmr->sg_nents, frmr->direction);
- spin_lock_bh(&rdma->sc_frmr_q_lock);
+ spin_lock(&rdma->sc_frmr_q_lock);
WARN_ON_ONCE(!list_empty(&frmr->frmr_list));
list_add(&frmr->frmr_list, &rdma->sc_frmr_q);
- spin_unlock_bh(&rdma->sc_frmr_q_lock);
+ spin_unlock(&rdma->sc_frmr_q_lock);
}
}
@@ -1067,13 +1067,13 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
goto errout;
}
newxprt->sc_sq_cq = ib_alloc_cq(dev, newxprt, newxprt->sc_sq_depth,
- 0, IB_POLL_SOFTIRQ);
+ 0, IB_POLL_WORKQUEUE);
if (IS_ERR(newxprt->sc_sq_cq)) {
dprintk("svcrdma: error creating SQ CQ for connect request\n");
goto errout;
}
newxprt->sc_rq_cq = ib_alloc_cq(dev, newxprt, newxprt->sc_rq_depth,
- 0, IB_POLL_SOFTIRQ);
+ 0, IB_POLL_WORKQUEUE);
if (IS_ERR(newxprt->sc_rq_cq)) {
dprintk("svcrdma: error creating RQ CQ for connect request\n");
goto errout;
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 06/22] svcrdma: Clean up RPC-over-RDMA Reply header encoder
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (4 preceding siblings ...)
2017-01-07 17:15 ` [PATCH v1 05/22] svcrdma: Poll CQs in "workqueue" mode Chuck Lever
@ 2017-01-07 17:16 ` Chuck Lever
2017-01-07 17:16 ` [PATCH v1 07/22] svcrdma: Clean up RPC-over-RDMA Call header decoder Chuck Lever
` (18 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:16 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Replace C structure-based XDR decoding with pointer arithmetic.
Pointer arithmetic is considered more portable, and is used
throughout the kernel's existing XDR encoders. The gcc optimizer
generates similar assembler code either way.
Byte-swapping before a memory store on x86 typically results in an
instruction pipeline stall. Avoid byte-swapping when encoding a new
header.
svcrdma currently doesn't alter a connection's credit grant value
after the connection has been accepted, so it is effectively a
constant. Cache the byte-swapped value in a separate field.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 11 +++++---
net/sunrpc/xprtrdma/svc_rdma_marshal.c | 39 +++++++++++++++++++-----------
net/sunrpc/xprtrdma/svc_rdma_sendto.c | 10 ++------
net/sunrpc/xprtrdma/svc_rdma_transport.c | 1 +
4 files changed, 35 insertions(+), 26 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 997bc78..61cfee8 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -140,7 +140,8 @@ struct svcxprt_rdma {
atomic_t sc_sq_avail; /* SQEs ready to be consumed */
unsigned int sc_sq_depth; /* Depth of SQ */
unsigned int sc_rq_depth; /* Depth of RQ */
- u32 sc_max_requests; /* Forward credits */
+ __be32 sc_fc_credits; /* Forward credits */
+ u32 sc_max_requests; /* Max requests */
u32 sc_max_bc_requests;/* Backward credits */
int sc_max_req_size; /* Size of each RQ WR buf */
@@ -211,10 +212,10 @@ extern int svc_rdma_xdr_encode_error(struct svcxprt_rdma *,
extern void svc_rdma_xdr_encode_reply_array(struct rpcrdma_write_array *, int);
extern void svc_rdma_xdr_encode_array_chunk(struct rpcrdma_write_array *, int,
__be32, __be64, u32);
-extern void svc_rdma_xdr_encode_reply_header(struct svcxprt_rdma *,
- struct rpcrdma_msg *,
- struct rpcrdma_msg *,
- enum rpcrdma_proc);
+extern void svc_rdma_xdr_encode_reply_header(struct svcxprt_rdma *rdma,
+ __be32 *rdma_argp,
+ __be32 *rdma_resp,
+ __be32 rdma_type);
extern unsigned int svc_rdma_xdr_get_reply_hdr_len(__be32 *rdma_resp);
/* svc_rdma_recvfrom.c */
diff --git a/net/sunrpc/xprtrdma/svc_rdma_marshal.c b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
index 4e72034..5efa019 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_marshal.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
@@ -249,7 +249,7 @@ int svc_rdma_xdr_encode_error(struct svcxprt_rdma *xprt,
*va++ = rmsgp->rm_xid;
*va++ = rmsgp->rm_vers;
- *va++ = cpu_to_be32(xprt->sc_max_requests);
+ *va++ = xprt->sc_fc_credits;
*va++ = rdma_error;
*va++ = cpu_to_be32(err);
if (err == ERR_VERS) {
@@ -330,18 +330,29 @@ void svc_rdma_xdr_encode_array_chunk(struct rpcrdma_write_array *ary,
seg->rs_length = cpu_to_be32(write_len);
}
-void svc_rdma_xdr_encode_reply_header(struct svcxprt_rdma *xprt,
- struct rpcrdma_msg *rdma_argp,
- struct rpcrdma_msg *rdma_resp,
- enum rpcrdma_proc rdma_type)
+/**
+ * svc_rdma_xdr_encode_reply_header - Encode Reply's fixed header fields
+ * @rdma: controlling transport
+ * @rdma_argp: Call's transport header
+ * @rdma_resp: Reply's transport header
+ * @rdma_type: procedure number
+ *
+ */
+void svc_rdma_xdr_encode_reply_header(struct svcxprt_rdma *rdma,
+ __be32 *rdma_argp, __be32 *rdma_resp,
+ __be32 rdma_type)
{
- rdma_resp->rm_xid = rdma_argp->rm_xid;
- rdma_resp->rm_vers = rdma_argp->rm_vers;
- rdma_resp->rm_credit = cpu_to_be32(xprt->sc_max_requests);
- rdma_resp->rm_type = cpu_to_be32(rdma_type);
-
- /* Encode <nul> chunks lists */
- rdma_resp->rm_body.rm_chunks[0] = xdr_zero;
- rdma_resp->rm_body.rm_chunks[1] = xdr_zero;
- rdma_resp->rm_body.rm_chunks[2] = xdr_zero;
+ __be32 *p;
+
+ p = rdma_resp;
+
+ *p++ = *rdma_argp; /* XID */
+ *p++ = *(rdma_argp + 1); /* vers */
+ *p++ = rdma->sc_fc_credits;
+ *p++ = rdma_type;
+
+ /* Start with empty chunks */
+ *p++ = xdr_zero;
+ *p++ = xdr_zero;
+ *p = xdr_zero;
}
diff --git a/net/sunrpc/xprtrdma/svc_rdma_sendto.c b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
index ba76f1617..7150201 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_sendto.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
@@ -560,7 +560,6 @@ int svc_rdma_sendto(struct svc_rqst *rqstp)
struct rpcrdma_msg *rdma_argp;
struct rpcrdma_msg *rdma_resp;
struct rpcrdma_write_array *wr_ary, *rp_ary;
- enum rpcrdma_proc reply_type;
int ret;
int inline_bytes;
struct page *res_page;
@@ -597,12 +596,9 @@ int svc_rdma_sendto(struct svc_rqst *rqstp)
if (!res_page)
goto err0;
rdma_resp = page_address(res_page);
- if (rp_ary)
- reply_type = RDMA_NOMSG;
- else
- reply_type = RDMA_MSG;
- svc_rdma_xdr_encode_reply_header(rdma, rdma_argp,
- rdma_resp, reply_type);
+ svc_rdma_xdr_encode_reply_header(rdma, (__be32 *)rdma_argp,
+ (__be32 *)rdma_resp,
+ rp_ary ? rdma_nomsg : rdma_msg);
/* Send any write-chunk data and build resp write-list */
if (wr_ary) {
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index 4d34afb..3258af7 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -1042,6 +1042,7 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
newxprt->sc_max_req_size = svcrdma_max_req_size;
newxprt->sc_max_requests = min_t(u32, dev->attrs.max_qp_wr,
svcrdma_max_requests);
+ newxprt->sc_fc_credits = cpu_to_be32(newxprt->sc_max_requests);
newxprt->sc_max_bc_requests = min_t(u32, dev->attrs.max_qp_wr,
svcrdma_max_bc_requests);
newxprt->sc_rq_depth = newxprt->sc_max_requests +
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 07/22] svcrdma: Clean up RPC-over-RDMA Call header decoder
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (5 preceding siblings ...)
2017-01-07 17:16 ` [PATCH v1 06/22] svcrdma: Clean up RPC-over-RDMA Reply header encoder Chuck Lever
@ 2017-01-07 17:16 ` Chuck Lever
2017-01-07 17:16 ` [PATCH v1 08/22] svcrdma: Introduce local rdma_rw API helpers Chuck Lever
` (17 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:16 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Replace C structure-based XDR decoding with pointer arithmetic.
Pointer arithmetic is considered more portable.
Rename the "decode" functions. Nothing is decoded here, they
perform only transport header sanity checking. Use existing XDR
naming conventions to help readability.
Straight-line the hot path:
- relocate the dprintk call sites out of line
- remove unnecessary byte-swapping
- reduce count of conditional branches
Deprecate RDMA_MSGP. It's not properly spec'd by RFC5666, and
therefore never used by any V1 client.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
net/sunrpc/xprtrdma/svc_rdma_marshal.c | 232 +++++++++++---------------------
1 file changed, 79 insertions(+), 153 deletions(-)
diff --git a/net/sunrpc/xprtrdma/svc_rdma_marshal.c b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
index 5efa019..7c8d60f 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_marshal.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
@@ -1,4 +1,5 @@
/*
+ * Copyright (c) 2016 Oracle. All rights reserved.
* Copyright (c) 2005-2006 Network Appliance, Inc. All rights reserved.
*
* This software is available to you under a choice of one of two
@@ -47,102 +48,43 @@
#define RPCDBG_FACILITY RPCDBG_SVCXPRT
-/*
- * Decodes a read chunk list. The expected format is as follows:
- * descrim : xdr_one
- * position : __be32 offset into XDR stream
- * handle : __be32 RKEY
- * . . .
- * end-of-list: xdr_zero
- */
-static __be32 *decode_read_list(__be32 *va, __be32 *vaend)
+static __be32 *xdr_check_read_list(__be32 *p, __be32 *end)
{
- struct rpcrdma_read_chunk *ch = (struct rpcrdma_read_chunk *)va;
+ __be32 *next;
- while (ch->rc_discrim != xdr_zero) {
- if (((unsigned long)ch + sizeof(struct rpcrdma_read_chunk)) >
- (unsigned long)vaend) {
- dprintk("svcrdma: vaend=%p, ch=%p\n", vaend, ch);
+ while (*p++ != xdr_zero) {
+ next = p + rpcrdma_readchunk_maxsz - 1;
+ if (next > end)
return NULL;
- }
- ch++;
+ p = next;
}
- return &ch->rc_position;
+ return p;
}
-/*
- * Decodes a write chunk list. The expected format is as follows:
- * descrim : xdr_one
- * nchunks : <count>
- * handle : __be32 RKEY ---+
- * length : __be32 <len of segment> |
- * offset : remove va + <count>
- * . . . |
- * ---+
- */
-static __be32 *decode_write_list(__be32 *va, __be32 *vaend)
+static __be32 *xdr_check_write_list(__be32 *p, __be32 *end)
{
- unsigned long start, end;
- int nchunks;
-
- struct rpcrdma_write_array *ary =
- (struct rpcrdma_write_array *)va;
-
- /* Check for not write-array */
- if (ary->wc_discrim == xdr_zero)
- return &ary->wc_nchunks;
+ __be32 *next;
- if ((unsigned long)ary + sizeof(struct rpcrdma_write_array) >
- (unsigned long)vaend) {
- dprintk("svcrdma: ary=%p, vaend=%p\n", ary, vaend);
- return NULL;
- }
- nchunks = be32_to_cpu(ary->wc_nchunks);
-
- start = (unsigned long)&ary->wc_array[0];
- end = (unsigned long)vaend;
- if (nchunks < 0 ||
- nchunks > (SIZE_MAX - start) / sizeof(struct rpcrdma_write_chunk) ||
- (start + (sizeof(struct rpcrdma_write_chunk) * nchunks)) > end) {
- dprintk("svcrdma: ary=%p, wc_nchunks=%d, vaend=%p\n",
- ary, nchunks, vaend);
- return NULL;
+ while (*p++ != xdr_zero) {
+ next = p + 1 + be32_to_cpup(p) * rpcrdma_segment_maxsz;
+ if (next > end)
+ return NULL;
+ p = next;
}
- /*
- * rs_length is the 2nd 4B field in wc_target and taking its
- * address skips the list terminator
- */
- return &ary->wc_array[nchunks].wc_target.rs_length;
+ return p;
}
-static __be32 *decode_reply_array(__be32 *va, __be32 *vaend)
+static __be32 *xdr_check_reply_chunk(__be32 *p, __be32 *end)
{
- unsigned long start, end;
- int nchunks;
- struct rpcrdma_write_array *ary =
- (struct rpcrdma_write_array *)va;
-
- /* Check for no reply-array */
- if (ary->wc_discrim == xdr_zero)
- return &ary->wc_nchunks;
-
- if ((unsigned long)ary + sizeof(struct rpcrdma_write_array) >
- (unsigned long)vaend) {
- dprintk("svcrdma: ary=%p, vaend=%p\n", ary, vaend);
- return NULL;
- }
- nchunks = be32_to_cpu(ary->wc_nchunks);
-
- start = (unsigned long)&ary->wc_array[0];
- end = (unsigned long)vaend;
- if (nchunks < 0 ||
- nchunks > (SIZE_MAX - start) / sizeof(struct rpcrdma_write_chunk) ||
- (start + (sizeof(struct rpcrdma_write_chunk) * nchunks)) > end) {
- dprintk("svcrdma: ary=%p, wc_nchunks=%d, vaend=%p\n",
- ary, nchunks, vaend);
- return NULL;
+ __be32 *next;
+
+ if (*p++ != xdr_zero) {
+ next = p + 1 + be32_to_cpup(p) * rpcrdma_segment_maxsz;
+ if (next > end)
+ return NULL;
+ p = next;
}
- return (__be32 *)&ary->wc_array[nchunks];
+ return p;
}
/**
@@ -158,87 +100,71 @@ static __be32 *decode_reply_array(__be32 *va, __be32 *vaend)
*/
int svc_rdma_xdr_decode_req(struct xdr_buf *rq_arg)
{
- struct rpcrdma_msg *rmsgp;
- __be32 *va, *vaend;
- unsigned int len;
- u32 hdr_len;
+ __be32 *p, *end, *rdma_argp;
+ unsigned int hdr_len;
/* Verify that there's enough bytes for header + something */
- if (rq_arg->len <= RPCRDMA_HDRLEN_ERR) {
- dprintk("svcrdma: header too short = %d\n",
- rq_arg->len);
- return -EINVAL;
- }
+ if (rq_arg->len <= RPCRDMA_HDRLEN_ERR)
+ goto out_short;
- rmsgp = (struct rpcrdma_msg *)rq_arg->head[0].iov_base;
- if (rmsgp->rm_vers != rpcrdma_version) {
- dprintk("%s: bad version %u\n", __func__,
- be32_to_cpu(rmsgp->rm_vers));
- return -EPROTONOSUPPORT;
- }
+ rdma_argp = rq_arg->head[0].iov_base;
+ if (*(rdma_argp + 1) != rpcrdma_version)
+ goto out_version;
- switch (be32_to_cpu(rmsgp->rm_type)) {
- case RDMA_MSG:
- case RDMA_NOMSG:
+ switch (*(rdma_argp + 3)) {
+ case rdma_msg:
+ case rdma_nomsg:
break;
- case RDMA_DONE:
- /* Just drop it */
- dprintk("svcrdma: dropping RDMA_DONE message\n");
- return 0;
-
- case RDMA_ERROR:
- /* Possible if this is a backchannel reply.
- * XXX: We should cancel this XID, though.
- */
- dprintk("svcrdma: dropping RDMA_ERROR message\n");
- return 0;
-
- case RDMA_MSGP:
- /* Pull in the extra for the padded case, bump our pointer */
- rmsgp->rm_body.rm_padded.rm_align =
- be32_to_cpu(rmsgp->rm_body.rm_padded.rm_align);
- rmsgp->rm_body.rm_padded.rm_thresh =
- be32_to_cpu(rmsgp->rm_body.rm_padded.rm_thresh);
-
- va = &rmsgp->rm_body.rm_padded.rm_pempty[4];
- rq_arg->head[0].iov_base = va;
- len = (u32)((unsigned long)va - (unsigned long)rmsgp);
- rq_arg->head[0].iov_len -= len;
- if (len > rq_arg->len)
- return -EINVAL;
- return len;
- default:
- dprintk("svcrdma: bad rdma procedure (%u)\n",
- be32_to_cpu(rmsgp->rm_type));
- return -EINVAL;
- }
+ case rdma_done:
+ goto out_drop;
- /* The chunk list may contain either a read chunk list or a write
- * chunk list and a reply chunk list.
- */
- va = &rmsgp->rm_body.rm_chunks[0];
- vaend = (__be32 *)((unsigned long)rmsgp + rq_arg->len);
- va = decode_read_list(va, vaend);
- if (!va) {
- dprintk("svcrdma: failed to decode read list\n");
- return -EINVAL;
- }
- va = decode_write_list(va, vaend);
- if (!va) {
- dprintk("svcrdma: failed to decode write list\n");
- return -EINVAL;
- }
- va = decode_reply_array(va, vaend);
- if (!va) {
- dprintk("svcrdma: failed to decode reply chunk\n");
- return -EINVAL;
+ case rdma_error:
+ goto out_drop;
+
+ default:
+ goto out_proc;
}
- rq_arg->head[0].iov_base = va;
- hdr_len = (unsigned long)va - (unsigned long)rmsgp;
+ end = (__be32 *)((unsigned long)rdma_argp + rq_arg->len);
+ p = xdr_check_read_list(rdma_argp + 4, end);
+ if (!p)
+ goto out_inval;
+ p = xdr_check_write_list(p, end);
+ if (!p)
+ goto out_inval;
+ p = xdr_check_reply_chunk(p, end);
+ if (!p)
+ goto out_inval;
+ if (p > end)
+ goto out_inval;
+
+ rq_arg->head[0].iov_base = p;
+ hdr_len = (unsigned long)p - (unsigned long)rdma_argp;
rq_arg->head[0].iov_len -= hdr_len;
return hdr_len;
+
+out_short:
+ dprintk("svcrdma: header too short = %d\n", rq_arg->len);
+ return -EINVAL;
+
+out_version:
+ dprintk("svcrdma: bad xprt version: %u\n",
+ be32_to_cpup(rdma_argp + 1));
+ return -EPROTONOSUPPORT;
+
+out_drop:
+ dprintk("svcrdma: dropping RDMA_DONE/ERROR message\n");
+ return 0;
+
+out_proc:
+ dprintk("svcrdma: bad rdma procedure (%u)\n",
+ be32_to_cpup(rdma_argp + 3));
+ return -EINVAL;
+
+out_inval:
+ dprintk("svcrdma: failed to parse transport header\n");
+ return -EINVAL;
}
int svc_rdma_xdr_encode_error(struct svcxprt_rdma *xprt,
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 08/22] svcrdma: Introduce local rdma_rw API helpers
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (6 preceding siblings ...)
2017-01-07 17:16 ` [PATCH v1 07/22] svcrdma: Clean up RPC-over-RDMA Call header decoder Chuck Lever
@ 2017-01-07 17:16 ` Chuck Lever
2017-01-07 17:16 ` [PATCH v1 09/22] svcrdma: Use rdma_rw API in RPC reply path Chuck Lever
` (16 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:16 UTC (permalink / raw)
To: linux-rdma, linux-nfs
The plan is to replace the bespoke code that constructs and posts
RDMA Read and Write Work Requests with calls to the core rdma_rw
API. This shares code with other RDMA-enabled ULPs that manages the
gory details of buffer registration and posting Work Requests.
Some design notes:
o svc_xprt reference counting is modified, since one rdma_rw_ctx
equals one completion, no matter how many Write WRs are posted.
To accomodate the new reference counting scheme, a new version
of svc_rdma_send() is introduced.
o The structure of RPC-over-RDMA transport headers is flexible,
allowing multiple segments per Reply with arbitrary alignment.
Thus I did not take the further step of chaining Write WRs with
the Send WR containing the RPC Reply message. The Write and Send
WRs continue to be built by separate pieces of code.
o The current code builds the transport header as it is construct-
ing Write WRs. I've replaced that with marshaling of transport
header data items in a separate step. This is because the exact
structure of client-provided segments may not align with the
components of the reply xdr_buf, or pages in the page list. Thus
parts of each client-provided segment may be written at different
points in the send path.
o Since the Write list and Reply chunk marshaling code is being
replaced, I took the opportunity to replace some of the C
structure-based XDR encoding code with more portable code that
uses pointer arithmetic instead.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 23 +
net/sunrpc/xprtrdma/Makefile | 2
net/sunrpc/xprtrdma/svc_rdma_marshal.c | 110 ++++-
net/sunrpc/xprtrdma/svc_rdma_rw.c | 731 ++++++++++++++++++++++++++++++
net/sunrpc/xprtrdma/svc_rdma_sendto.c | 2
net/sunrpc/xprtrdma/svc_rdma_transport.c | 56 ++
6 files changed, 920 insertions(+), 4 deletions(-)
create mode 100644 net/sunrpc/xprtrdma/svc_rdma_rw.c
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 61cfee8..e6eca34 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -144,12 +144,15 @@ struct svcxprt_rdma {
u32 sc_max_requests; /* Max requests */
u32 sc_max_bc_requests;/* Backward credits */
int sc_max_req_size; /* Size of each RQ WR buf */
+ u8 sc_port_num;
struct ib_pd *sc_pd;
spinlock_t sc_ctxt_lock;
struct list_head sc_ctxts;
int sc_ctxt_used;
+ spinlock_t sc_rw_ctxt_lock;
+ struct list_head sc_rw_ctxts;
spinlock_t sc_map_lock;
struct list_head sc_maps;
@@ -208,10 +211,14 @@ extern int svc_rdma_handle_bc_reply(struct rpc_xprt *xprt,
extern int svc_rdma_xdr_encode_error(struct svcxprt_rdma *,
struct rpcrdma_msg *,
enum rpcrdma_errcode, __be32 *);
-extern void svc_rdma_xdr_encode_write_list(struct rpcrdma_msg *, int);
+extern void svc_rdma_old_encode_write_list(struct rpcrdma_msg *, int);
extern void svc_rdma_xdr_encode_reply_array(struct rpcrdma_write_array *, int);
extern void svc_rdma_xdr_encode_array_chunk(struct rpcrdma_write_array *, int,
__be32, __be64, u32);
+extern void svc_rdma_xdr_encode_write_list(__be32 *rdma_resp, __be32 *wr_ch,
+ unsigned int consumed);
+extern void svc_rdma_xdr_encode_reply_chunk(__be32 *rdma_resp, __be32 *rp_ch,
+ unsigned int consumed);
extern void svc_rdma_xdr_encode_reply_header(struct svcxprt_rdma *rdma,
__be32 *rdma_argp,
__be32 *rdma_resp,
@@ -227,6 +234,18 @@ extern int rdma_read_chunk_frmr(struct svcxprt_rdma *, struct svc_rqst *,
struct svc_rdma_op_ctxt *, int *, u32 *,
u32, u32, u64, bool);
+/* svc_rdma_rw.c */
+extern void svc_rdma_destroy_rw_ctxts(struct svcxprt_rdma *rdma);
+extern int svc_rdma_recv_read_list(struct svcxprt_rdma *rdma, __be32 *ch,
+ struct svc_rdma_op_ctxt *head,
+ struct svc_rqst *rqstp);
+extern int svc_rdma_send_write_list(struct svcxprt_rdma *rdma,
+ __be32 *wr_ch, __be32 *rdma_resp,
+ struct xdr_buf *xdr);
+extern int svc_rdma_send_reply_chunk(struct svcxprt_rdma *rdma,
+ __be32 *rp_ch, __be32 *rdma_resp,
+ struct xdr_buf *xdr);
+
/* svc_rdma_sendto.c */
extern int svc_rdma_map_xdr(struct svcxprt_rdma *, struct xdr_buf *,
struct svc_rdma_req_map *, bool);
@@ -241,6 +260,8 @@ extern void svc_rdma_send_error(struct svcxprt_rdma *, struct rpcrdma_msg *,
extern void svc_rdma_wc_read(struct ib_cq *, struct ib_wc *);
extern void svc_rdma_wc_inv(struct ib_cq *, struct ib_wc *);
extern int svc_rdma_send(struct svcxprt_rdma *, struct ib_send_wr *);
+extern int svc_rdma_post_send(struct svcxprt_rdma *rdma,
+ struct ib_send_wr *first_wr, int num_wrs);
extern int svc_rdma_post_recv(struct svcxprt_rdma *, gfp_t);
extern int svc_rdma_repost_recv(struct svcxprt_rdma *, gfp_t);
extern int svc_rdma_create_listen(struct svc_serv *, int, struct sockaddr *);
diff --git a/net/sunrpc/xprtrdma/Makefile b/net/sunrpc/xprtrdma/Makefile
index ef19fa4..c1ae814 100644
--- a/net/sunrpc/xprtrdma/Makefile
+++ b/net/sunrpc/xprtrdma/Makefile
@@ -4,5 +4,5 @@ rpcrdma-y := transport.o rpc_rdma.o verbs.o \
fmr_ops.o frwr_ops.o \
svc_rdma.o svc_rdma_backchannel.o svc_rdma_transport.o \
svc_rdma_marshal.o svc_rdma_sendto.o svc_rdma_recvfrom.o \
- module.o
+ svc_rdma_rw.o module.o
rpcrdma-$(CONFIG_SUNRPC_BACKCHANNEL) += backchannel.o
diff --git a/net/sunrpc/xprtrdma/svc_rdma_marshal.c b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
index 7c8d60f..01cb4e1 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_marshal.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
@@ -217,7 +217,7 @@ unsigned int svc_rdma_xdr_get_reply_hdr_len(__be32 *rdma_resp)
return (unsigned long)p - (unsigned long)rdma_resp;
}
-void svc_rdma_xdr_encode_write_list(struct rpcrdma_msg *rmsgp, int chunks)
+void svc_rdma_old_encode_write_list(struct rpcrdma_msg *rmsgp, int chunks)
{
struct rpcrdma_write_array *ary;
@@ -256,6 +256,114 @@ void svc_rdma_xdr_encode_array_chunk(struct rpcrdma_write_array *ary,
seg->rs_length = cpu_to_be32(write_len);
}
+/* Write chunk is copied from Call header to Reply header. Length
+ * fields are updated to reflect number of bytes consumed in each
+ * segment.
+ *
+ * Returns number of segments in this chunk.
+ */
+static unsigned int xdr_encode_write_chunk( __be32 *dst, __be32 *src,
+ unsigned int remaining)
+{
+ unsigned int i, nsegs;
+ u32 seg_len;
+
+ /* Write list discriminator */
+ *dst++ = *src++;
+
+ /* number of segments in this chunk */
+ nsegs = be32_to_cpup(src);
+ *dst++ = *src++;
+
+ for (i = nsegs; i; i--) {
+ /* segment's RDMA handle */
+ *dst++ = *src++;
+
+ /* bytes returned in this segment */
+ seg_len = be32_to_cpu(*src);
+ if (remaining >= seg_len) {
+ /* entire segment was consumed */
+ *dst = *src;
+ remaining -= seg_len;
+ } else {
+ /* segment only partly filled */
+ *dst = cpu_to_be32(remaining);
+ remaining = 0;
+ }
+ dst++; src++;
+
+ /* segment's RDMA offset */
+ *dst++ = *src++;
+ *dst++ = *src++;
+ }
+
+ return nsegs;
+}
+
+/**
+ * svc_rdma_xdr_encode_write_list - Encode Reply's Write list
+ * @rdma_resp: Reply's transport header
+ * @wr_ch: Write list in Call's transport header
+ * @consumed: total Write chunk bytes consumed in Reply
+ *
+ * The client provided a Write list in the Call message. Fill in
+ * the segments in the first Write chunk in the Reply message with
+ * the number of bytes consumed in each segment. Remaining chunks
+ * are returned unused.
+ *
+ * Assumptions:
+ * - Server can return only one Write chunk.
+ */
+void svc_rdma_xdr_encode_write_list(__be32 *rdma_resp, __be32 *wr_ch,
+ unsigned int consumed)
+{
+ unsigned int nsegs;
+ __be32 *p, *q;
+
+ /* RPC-over-RDMA V1 replies never have a Read list. */
+ p = rdma_resp + rpcrdma_fixed_maxsz + 1;
+
+ q = wr_ch;
+ while (*q != xdr_zero) {
+ nsegs = xdr_encode_write_chunk(p, q, consumed);
+ q += 2 + nsegs * rpcrdma_segment_maxsz;
+ p += 2 + nsegs * rpcrdma_segment_maxsz;
+ }
+
+ /* Terminate Write list */
+ *p++ = xdr_zero;
+
+ /* Reply chunk discriminator; may be replaced later */
+ *p = xdr_zero;
+}
+
+/**
+ * svc_rdma_xdr_encode_reply_chunk - Encode Reply's Reply chunk
+ * @rdma_resp: Reply's transport header
+ * @rp_ch: Reply chunk in Call's transport header
+ * @consumed: total Reply chunk bytes consumed in Reply
+ *
+ * The client provided a Reply chunk in the Call message. Fill in
+ * the segments in the Reply chunk in the Reply message with the
+ * number of bytes consumed in each segment.
+ */
+void svc_rdma_xdr_encode_reply_chunk(__be32 *rdma_resp, __be32 *rp_ch,
+ unsigned int consumed)
+{
+ __be32 *p;
+
+ /* Find the Reply chunk in the Reply's xprt header.
+ * RPC-over-RDMA V1 replies never have a Read list.
+ */
+ p = rdma_resp + rpcrdma_fixed_maxsz + 1;
+
+ /* Skip past Write list */
+ while (*p++ != xdr_zero)
+ p += 1 + be32_to_cpup(p) * rpcrdma_segment_maxsz;
+
+ xdr_encode_write_chunk(p, rp_ch, consumed);
+}
+
/**
* svc_rdma_xdr_encode_reply_header - Encode Reply's fixed header fields
* @rdma: controlling transport
diff --git a/net/sunrpc/xprtrdma/svc_rdma_rw.c b/net/sunrpc/xprtrdma/svc_rdma_rw.c
new file mode 100644
index 0000000..38876f5
--- /dev/null
+++ b/net/sunrpc/xprtrdma/svc_rdma_rw.c
@@ -0,0 +1,731 @@
+/*
+ * Copyright (c) 2016 Oracle. All rights reserved.
+ *
+ * Use the core R/W API to move RPC-over-RDMA Read and Write chunks.
+ */
+
+#include <linux/sunrpc/rpc_rdma.h>
+#include <linux/sunrpc/svc_rdma.h>
+#include <linux/sunrpc/debug.h>
+
+#include <rdma/rw.h>
+
+#define RPCDBG_FACILITY RPCDBG_SVCXPRT
+
+/* Each R/W context contains state for one chain of RDMA Read or
+ * Write Work Requests (one RDMA segment to be read from or written
+ * back to the client).
+ *
+ * Each WR chain handles a single contiguous server-side buffer,
+ * because some registration modes (eg. FRWR) do not support a
+ * discontiguous scatterlist.
+ *
+ * Each WR chain handles only one R_key. Each RPC-over-RDMA segment
+ * from a client may contain a unique R_key, so each WR chain moves
+ * one segment (or less) at a time.
+ *
+ * The scatterlist makes this data structure just over 8KB in size
+ * on 4KB-page platforms. As the size of this structure increases
+ * past one page, it becomes more likely that allocating one of these
+ * will fail. Therefore, these contexts are created on demand, but
+ * cached and reused until the controlling svcxprt_rdma is destroyed.
+ */
+struct svc_rdma_rw_ctxt {
+ struct list_head rw_list;
+ struct ib_cqe rw_cqe;
+ struct svcxprt_rdma *rw_rdma;
+ int rw_nents;
+ int rw_wrcount;
+ enum dma_data_direction rw_dir;
+ struct svc_rdma_op_ctxt *rw_readctxt;
+ struct rdma_rw_ctx rw_ctx;
+ struct scatterlist rw_sg[RPCSVC_MAXPAGES];
+};
+
+static struct svc_rdma_rw_ctxt *
+svc_rdma_get_rw_ctxt(struct svcxprt_rdma *rdma)
+{
+ struct svc_rdma_rw_ctxt *ctxt;
+
+ svc_xprt_get(&rdma->sc_xprt);
+
+ spin_lock(&rdma->sc_rw_ctxt_lock);
+ if (list_empty(&rdma->sc_rw_ctxts))
+ goto out_empty;
+
+ ctxt = list_first_entry(&rdma->sc_rw_ctxts,
+ struct svc_rdma_rw_ctxt, rw_list);
+ list_del_init(&ctxt->rw_list);
+ spin_unlock(&rdma->sc_rw_ctxt_lock);
+
+out:
+ ctxt->rw_dir = DMA_NONE;
+ return ctxt;
+
+out_empty:
+ spin_unlock(&rdma->sc_rw_ctxt_lock);
+
+ ctxt = kmalloc(sizeof(*ctxt), GFP_KERNEL);
+ if (!ctxt) {
+ svc_xprt_put(&rdma->sc_xprt);
+ return NULL;
+ }
+
+ ctxt->rw_rdma = rdma;
+ INIT_LIST_HEAD(&ctxt->rw_list);
+ sg_init_table(ctxt->rw_sg, ARRAY_SIZE(ctxt->rw_sg));
+ goto out;
+}
+
+static void svc_rdma_put_rw_ctxt(struct svc_rdma_rw_ctxt *ctxt)
+{
+ struct svcxprt_rdma *rdma = ctxt->rw_rdma;
+
+ if (ctxt->rw_dir != DMA_NONE)
+ rdma_rw_ctx_destroy(&ctxt->rw_ctx, rdma->sc_qp,
+ rdma->sc_port_num,
+ ctxt->rw_sg, ctxt->rw_nents,
+ ctxt->rw_dir);
+
+ spin_lock(&rdma->sc_rw_ctxt_lock);
+ list_add(&ctxt->rw_list, &rdma->sc_rw_ctxts);
+ spin_unlock(&rdma->sc_rw_ctxt_lock);
+
+ svc_xprt_put(&rdma->sc_xprt);
+}
+
+/**
+ * svc_rdma_destroy_rw_ctxts - Free write contexts
+ * @rdma: xprt about to be destroyed
+ *
+ */
+void svc_rdma_destroy_rw_ctxts(struct svcxprt_rdma *rdma)
+{
+ struct svc_rdma_rw_ctxt *ctxt;
+
+ while (!list_empty(&rdma->sc_rw_ctxts)) {
+ ctxt = list_first_entry(&rdma->sc_rw_ctxts,
+ struct svc_rdma_rw_ctxt, rw_list);
+ list_del(&ctxt->rw_list);
+ kfree(ctxt);
+ }
+}
+
+/**
+ * svc_rdma_wc_write_ctx - Handle completion of an RDMA Write ctx
+ * @cq: controlling Completion Queue
+ * @wc: Work Completion
+ *
+ * Invoked in soft IRQ context.
+ *
+ * Assumptions:
+ * - Write completion is not responsible for freeing pages under
+ * I/O.
+ */
+static void svc_rdma_wc_write_ctx(struct ib_cq *cq, struct ib_wc *wc)
+{
+ struct ib_cqe *cqe = wc->wr_cqe;
+ struct svc_rdma_rw_ctxt *ctxt =
+ container_of(cqe, struct svc_rdma_rw_ctxt, rw_cqe);
+ struct svcxprt_rdma *rdma = ctxt->rw_rdma;
+
+ atomic_add(ctxt->rw_wrcount, &rdma->sc_sq_avail);
+ wake_up(&rdma->sc_send_wait);
+
+ if (wc->status != IB_WC_SUCCESS)
+ goto flush;
+
+out:
+ svc_rdma_put_rw_ctxt(ctxt);
+ return;
+
+flush:
+ set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags);
+ if (wc->status != IB_WC_WR_FLUSH_ERR)
+ pr_err("svcrdma: write ctx: %s (%u/0x%x)\n",
+ ib_wc_status_msg(wc->status),
+ wc->status, wc->vendor_err);
+ goto out;
+}
+
+/**
+ * svc_rdma_wc_read_ctx - Handle completion of an RDMA Read ctx
+ * @cq: controlling Completion Queue
+ * @wc: Work Completion
+ *
+ * Invoked in soft IRQ context.
+ */
+static void svc_rdma_wc_read_ctx(struct ib_cq *cq, struct ib_wc *wc)
+{
+ struct ib_cqe *cqe = wc->wr_cqe;
+ struct svc_rdma_rw_ctxt *ctxt =
+ container_of(cqe, struct svc_rdma_rw_ctxt, rw_cqe);
+ struct svcxprt_rdma *rdma = ctxt->rw_rdma;
+ struct svc_rdma_op_ctxt *head;
+
+ atomic_add(ctxt->rw_wrcount, &rdma->sc_sq_avail);
+ wake_up(&rdma->sc_send_wait);
+
+ if (wc->status != IB_WC_SUCCESS)
+ goto flush;
+
+ head = ctxt->rw_readctxt;
+ if (!head)
+ goto out;
+
+ spin_lock(&rdma->sc_rq_dto_lock);
+ list_add_tail(&head->list, &rdma->sc_read_complete_q);
+ spin_unlock(&rdma->sc_rq_dto_lock);
+ set_bit(XPT_DATA, &rdma->sc_xprt.xpt_flags);
+ svc_xprt_enqueue(&rdma->sc_xprt);
+
+out:
+ svc_rdma_put_rw_ctxt(ctxt);
+ return;
+
+flush:
+ set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags);
+ if (wc->status != IB_WC_WR_FLUSH_ERR)
+ pr_err("svcrdma: read ctx: %s (%u/0x%x)\n",
+ ib_wc_status_msg(wc->status),
+ wc->status, wc->vendor_err);
+ goto out;
+}
+
+static int svc_rdma_send_write_ctx(struct svcxprt_rdma *rdma,
+ struct svc_rdma_rw_ctxt *ctxt,
+ u64 offset, u32 rkey)
+{
+ struct ib_send_wr *first_wr;
+ int ret;
+
+ ret = rdma_rw_ctx_init(&ctxt->rw_ctx,
+ rdma->sc_qp, rdma->sc_port_num,
+ ctxt->rw_sg, ctxt->rw_nents,
+ 0, offset, rkey, DMA_TO_DEVICE);
+ if (ret < 0)
+ return ret;
+
+ ctxt->rw_wrcount = ret;
+ ctxt->rw_dir = DMA_TO_DEVICE;
+ ctxt->rw_cqe.done = svc_rdma_wc_write_ctx;
+ first_wr = rdma_rw_ctx_wrs(&ctxt->rw_ctx,
+ rdma->sc_qp, rdma->sc_port_num,
+ &ctxt->rw_cqe, NULL);
+ atomic_add(ret, &rdma_stat_write);
+ return svc_rdma_post_send(rdma, first_wr, ret);
+}
+
+/* Common information for sending a Write chunk.
+ * - Tracks progress of writing one chunk
+ * - Stores arguments for the SGL constructor function
+ */
+struct svc_rdma_write_info {
+ struct svcxprt_rdma *wi_rdma;
+
+ /* write state of this chunk */
+ unsigned int wi_bytes_consumed;
+ unsigned int wi_seg_off;
+ unsigned int wi_seg_no;
+ unsigned int wi_nsegs;
+ __be32 *wi_segs;
+
+ /* SGL constructor arguments */
+ struct xdr_buf *wi_xdr;
+ unsigned char *wi_base;
+ unsigned int wi_next_off;
+};
+
+static void svc_rdma_init_write_info(struct svcxprt_rdma *rdma, __be32 *chunk,
+ struct svc_rdma_write_info *info)
+{
+ info->wi_rdma = rdma;
+ info->wi_bytes_consumed = 0;
+ info->wi_seg_off = 0;
+ info->wi_seg_no = 0;
+ info->wi_nsegs = be32_to_cpup(chunk + 1);
+ info->wi_segs = chunk + 2;
+}
+
+/* Build and DMA-map an SGL that covers one kvec in an xdr_buf
+ */
+static void svc_rdma_vec_to_sg(struct svc_rdma_write_info *info,
+ unsigned int len,
+ struct svc_rdma_rw_ctxt *ctxt)
+{
+ sg_set_buf(&ctxt->rw_sg[0], info->wi_base, len);
+ info->wi_base += len;
+
+ ctxt->rw_nents = 1;
+}
+
+/* Build and DMA-map an SGL that covers the pagelist of an xdr_buf
+ */
+static void svc_rdma_pagelist_to_sg(struct svc_rdma_write_info *info,
+ unsigned int remaining,
+ struct svc_rdma_rw_ctxt *ctxt)
+{
+ unsigned int sge_no, sge_bytes, page_off, page_no;
+ struct scatterlist *sg = ctxt->rw_sg;
+ struct xdr_buf *xdr = info->wi_xdr;
+
+ page_no = (info->wi_next_off + xdr->page_base) >> PAGE_SHIFT;
+ page_off = (info->wi_next_off + xdr->page_base) & ~PAGE_MASK;
+ info->wi_next_off += remaining;
+
+ sge_no = 0;
+ do {
+ sge_bytes = min_t(unsigned int, remaining,
+ PAGE_SIZE - page_off);
+
+ sg_set_page(&sg[sge_no++], xdr->pages[page_no],
+ sge_bytes, page_off);
+
+ remaining -= sge_bytes;
+ page_no++;
+ page_off = 0;
+ } while (remaining);
+
+ ctxt->rw_nents = sge_no;
+}
+
+/* Post Write WRs to send a portion of an xdr_buf containing
+ * an RPC Reply.
+ *
+ * If successful:
+ * - chunk/segment position are updated
+ * - Returns 0
+ *
+ * Otherwise a negative errno is returned.
+ */
+static int
+svc_rdma_send_writes(struct svc_rdma_write_info *info,
+ void (*constructor)(struct svc_rdma_write_info *info,
+ unsigned int len,
+ struct svc_rdma_rw_ctxt *ctxt),
+ unsigned int total)
+{
+ struct svcxprt_rdma *rdma = info->wi_rdma;
+ unsigned int remaining, seg_no, seg_off;
+ struct svc_rdma_rw_ctxt *ctxt;
+ __be32 *seg;
+ int ret;
+
+ if (!total)
+ return 0;
+
+ remaining = total;
+ seg_no = info->wi_seg_no;
+ seg_off = info->wi_seg_off;
+ seg = info->wi_segs + seg_no * rpcrdma_segment_maxsz;
+ do {
+ unsigned int write_len;
+ u32 rs_length, rs_handle;
+ u64 rs_offset;
+
+ if (seg_no >= info->wi_nsegs)
+ goto out_overflow;
+
+ ctxt = svc_rdma_get_rw_ctxt(rdma);
+ if (!ctxt)
+ goto out_noctx;
+
+ rs_handle = be32_to_cpu(*seg++);
+ rs_length = be32_to_cpu(*seg++);
+ seg = xdr_decode_hyper(seg, &rs_offset);
+
+ write_len = min(remaining, rs_length - seg_off);
+ constructor(info, write_len, ctxt);
+ ret = svc_rdma_send_write_ctx(rdma, ctxt, rs_offset + seg_off,
+ rs_handle);
+ if (ret < 0)
+ goto out_senderr;
+
+ if (write_len == rs_length - seg_off) {
+ seg_no++;
+ seg_off = 0;
+ } else
+ seg_off += write_len;
+ remaining -= write_len;
+ } while (remaining);
+
+ info->wi_bytes_consumed += total;
+ info->wi_seg_no = seg_no;
+ info->wi_seg_off = seg_off;
+ return 0;
+
+out_overflow:
+ pr_err("svcrdma: inadequate space in Write chunk (%u)\n",
+ info->wi_nsegs);
+ return -EIO;
+
+out_noctx:
+ pr_err("svcrdma: no R/W ctxs available\n");
+ return -ENOMEM;
+
+out_senderr:
+ svc_rdma_put_rw_ctxt(ctxt);
+ pr_err("svcrdma: failed to write pagelist (%d)\n", ret);
+ return ret;
+}
+
+/* Send one of an xdr_buf's kvecs by itself. To send a Reply
+ * chunk, the whole RPC Reply is written back to the client.
+ * This function writes either the head or tail of the xdr_buf
+ * containing the Reply.
+ *
+ * If successful:
+ * - chunk/segment position are updated
+ * - Returns 0
+ *
+ * Otherwise a negative errno is returned.
+ */
+static int svc_rdma_send_xdr_kvec(struct svc_rdma_write_info *info,
+ struct kvec *vec)
+{
+ info->wi_base = vec->iov_base;
+
+ return svc_rdma_send_writes(info, svc_rdma_vec_to_sg,
+ vec->iov_len);
+}
+
+/* Send an xdr_buf's page list by itself. A Write chunk is
+ * just the page list. a Reply chunk is the head, page list,
+ * and tail. This function is shared between the two types
+ * of chunk.
+ *
+ * If successful:
+ * - chunk/segment position are updated
+ * - Returns 0
+ *
+ * Otherwise a negative errno is returned.
+ */
+static int svc_rdma_send_xdr_pagelist(struct svc_rdma_write_info *info,
+ struct xdr_buf *xdr)
+{
+ info->wi_xdr = xdr;
+ info->wi_next_off = 0;
+
+ return svc_rdma_send_writes(info, svc_rdma_pagelist_to_sg,
+ xdr->page_len);
+}
+
+/**
+ * svc_rdma_send_write_list - Write all chunks in the Write list
+ * @rdma: controlling RDMA transport
+ * @wr_ch: Write list provided by client
+ * @rdma_resp: buffer containing transport header under construction
+ * @xdr: xdr_buf carrying an RPC Reply
+ *
+ * Returns zero on success, or a negative errno.
+ *
+ * Assumptions:
+ * - Only one Write chunk, and it's the xdr_buf's entire pagelist
+ */
+int svc_rdma_send_write_list(struct svcxprt_rdma *rdma,
+ __be32 *wr_ch, __be32 *rdma_resp,
+ struct xdr_buf *xdr)
+{
+ struct svc_rdma_write_info info;
+ int ret;
+
+ svc_rdma_init_write_info(rdma, wr_ch, &info);
+
+ ret = svc_rdma_send_xdr_pagelist(&info, xdr);
+
+ svc_rdma_xdr_encode_write_list(rdma_resp, wr_ch,
+ info.wi_bytes_consumed);
+ return ret;
+}
+
+/**
+ * svc_rdma_send_reply_chunk - Write all segments in the Reply chunk
+ * @rdma: controlling RDMA transport
+ * @rp_ch: Reply chunk provided by client
+ * @rdma_resp: buffer containing transport header for Reply
+ * @xdr: xdr_buf carrying an RPC Reply
+ *
+ * Returns zero if all consumed segments have been posted successfully,
+ * or a negative errno.
+ *
+ * Assumptions:
+ * - The Reply chunk always carries the whole xdr_buf
+ */
+int svc_rdma_send_reply_chunk(struct svcxprt_rdma *rdma,
+ __be32 *rp_ch, __be32 *rdma_resp,
+ struct xdr_buf *xdr)
+{
+ struct svc_rdma_write_info info;
+ int ret;
+
+ svc_rdma_init_write_info(rdma, rp_ch, &info);
+
+ ret = svc_rdma_send_xdr_kvec(&info, &xdr->head[0]);
+ if (ret < 0)
+ goto out;
+ ret = svc_rdma_send_xdr_pagelist(&info, xdr);
+ if (ret < 0)
+ goto out;
+ ret = svc_rdma_send_xdr_kvec(&info, &xdr->tail[0]);
+
+out:
+ svc_rdma_xdr_encode_reply_chunk(rdma_resp, rp_ch,
+ info.wi_bytes_consumed);
+ return ret;
+}
+
+/* Pull one Read chunk (segment) from the client.
+ *
+ * Returns zero if one or more RDMA Reads have been posted. Otherwise,
+ * returns a negative errno if there is a Read list present but RDMA
+ * Reads could not be posted.
+ *
+ * For incoming Reads, @rqstp provides a page list containing sink pages.
+ * As pages are prepared for I/O, they are transferred to @head. After
+ * all Reads in the list have completed, svc_rdma_recvfrom builds an
+ * xdr_buf from the page list in @head.
+ *
+ * On entry, *page_no and *page_offset point into the rqstp's page list.
+ * On return, *page_no and *page_offset are updated to point to the next
+ * position in the page list.
+ */
+static int svc_rdma_recv_read_segment(struct svcxprt_rdma *rdma,
+ struct svc_rdma_op_ctxt *head,
+ struct svc_rqst *rqstp,
+ unsigned int *page_no,
+ unsigned int *page_offset,
+ u32 rkey, u32 len, u64 offset,
+ bool last)
+{
+ struct svc_rdma_rw_ctxt *ctxt;
+ struct ib_send_wr *first_wr;
+ unsigned int pg_no, seg_no;
+ u32 pg_off;
+ int ret;
+
+ dprintk("svcrdma: reading segment %u@0x%016llx:0x%08x\n",
+ len, offset, rkey);
+
+ ctxt = svc_rdma_get_rw_ctxt(rdma);
+ if (!ctxt)
+ return -ENOMEM;
+
+ pg_off = *page_offset;
+ pg_no = *page_no;
+ ctxt->rw_nents = PAGE_ALIGN(*page_offset + len) >> PAGE_SHIFT;
+ for (seg_no = 0; seg_no < ctxt->rw_nents; seg_no++) {
+ unsigned int seg_len = min_t(unsigned int, len, PAGE_SIZE - pg_off);
+
+ head->arg.pages[pg_no] = rqstp->rq_arg.pages[pg_no];
+ head->arg.page_len += seg_len;
+ head->arg.len += seg_len;
+ if (!pg_off)
+ head->count++;
+
+ sg_set_page(&ctxt->rw_sg[seg_no], rqstp->rq_arg.pages[pg_no],
+ seg_len, pg_off);
+
+ rqstp->rq_respages = &rqstp->rq_arg.pages[pg_no + 1];
+ rqstp->rq_next_page = rqstp->rq_respages + 1;
+
+ pg_off += seg_len;
+ if (pg_off == PAGE_SIZE) {
+ pg_off = 0;
+ pg_no++;
+ }
+ len -= seg_len;
+ }
+
+ ret = rdma_rw_ctx_init(&ctxt->rw_ctx,
+ rdma->sc_qp, rdma->sc_port_num,
+ ctxt->rw_sg, ctxt->rw_nents,
+ 0, offset, rkey, DMA_FROM_DEVICE);
+ if (ret < 0)
+ goto out_err;
+
+ ctxt->rw_wrcount = ret;
+ ctxt->rw_dir = DMA_FROM_DEVICE;
+ ctxt->rw_cqe.done = svc_rdma_wc_read_ctx;
+ ctxt->rw_readctxt = last ? head : NULL;
+ first_wr = rdma_rw_ctx_wrs(&ctxt->rw_ctx,
+ rdma->sc_qp, rdma->sc_port_num,
+ &ctxt->rw_cqe, NULL);
+ atomic_add(ret, &rdma_stat_read);
+ ret = svc_rdma_post_send(rdma, first_wr, ret);
+ if (ret)
+ goto out_err;
+
+ *page_no = pg_no;
+ *page_offset = pg_off;
+ return 0;
+
+out_err:
+ svc_rdma_put_rw_ctxt(ctxt);
+ pr_err("svcrdma: failed to map read segment (%d)\n", ret);
+ return ret;
+}
+
+/* If there was additional inline content, append it to the end of arg.pages.
+ * Tail copy has to be done after the reader function has determined how many
+ * pages were consumed for RDMA Read.
+ */
+static int svc_rdma_copy_tail(struct svc_rqst *rqstp,
+ struct svc_rdma_op_ctxt *head, u32 position,
+ unsigned int page_offset, unsigned int page_no)
+{
+ char *srcp, *destp;
+ u32 byte_count;
+
+ srcp = head->arg.head[0].iov_base + position;
+ byte_count = head->arg.head[0].iov_len - position;
+ if (byte_count > PAGE_SIZE) {
+ dprintk("svcrdma: large tail unsupported\n");
+ return 0;
+ }
+
+ /* Fit as much of the tail on the current page as possible */
+ if (page_offset != PAGE_SIZE) {
+ destp = page_address(rqstp->rq_arg.pages[page_no]);
+ destp += page_offset;
+ while (byte_count--) {
+ *destp++ = *srcp++;
+ page_offset++;
+ if (page_offset == PAGE_SIZE && byte_count)
+ goto more;
+ }
+ goto done;
+ }
+
+more:
+ /* Fit the rest on the next page */
+ page_no++;
+ destp = page_address(rqstp->rq_arg.pages[page_no]);
+ while (byte_count--)
+ *destp++ = *srcp++;
+
+ rqstp->rq_respages = &rqstp->rq_arg.pages[page_no+1];
+ rqstp->rq_next_page = rqstp->rq_respages + 1;
+
+done:
+ byte_count = head->arg.head[0].iov_len - position;
+ head->arg.page_len += byte_count;
+ head->arg.len += byte_count;
+ head->arg.buflen += byte_count;
+ return 1;
+}
+
+static unsigned int svc_rdma_read_chunk_count(__be32 *p)
+{
+ unsigned int nsegs;
+
+ for (nsegs = 0; *p != xdr_zero; p += rpcrdma_readchunk_maxsz)
+ nsegs++;
+ return nsegs;
+}
+
+/**
+ * svc_rdma_recv_read_list - Pull read chunks from the client
+ * @rdma: controlling RDMA transport
+ * @ch: pointer to Read list in the incoming transport header
+ * @head: pages under I/O collect here
+ * @rqstp: set of pages to use as Read sink buffers
+ *
+ * Returns one if RDMA Reads were posted successfully; returns
+ * zero if there was no Read list; or negative errno if there
+ * was a Read list present but Reads could not be posted.
+ *
+ * Assumptions:
+ * - Clients can send multiple Read chunks in a Read list, but
+ * the chunks must all have the same value in their Position
+ * field.
+ */
+int svc_rdma_recv_read_list(struct svcxprt_rdma *rdma, __be32 *ch,
+ struct svc_rdma_op_ctxt *head,
+ struct svc_rqst *rqstp)
+{
+ unsigned int page_no, page_offset;
+ u32 position;
+ __be32 *p;
+ bool last;
+ int ret;
+
+ p = ch;
+
+ /* Sanity check */
+ if (svc_rdma_read_chunk_count(p++) > RPCSVC_MAXPAGES)
+ return -EINVAL;
+
+ /* "head" keeps all the pages that comprise the request.
+ */
+ head->arg.head[0] = rqstp->rq_arg.head[0];
+ head->arg.tail[0] = rqstp->rq_arg.tail[0];
+ head->hdr_count = head->count;
+ head->arg.page_base = 0;
+ head->arg.page_len = 0;
+ head->arg.len = rqstp->rq_arg.len;
+ head->arg.buflen = rqstp->rq_arg.buflen;
+
+ /* RDMA_NOMSG: RDMA Read data should land just after Receive data.
+ */
+ position = be32_to_cpu(*p++);
+ if (position == 0) {
+ head->arg.pages = &head->pages[0];
+ page_offset = head->byte_len;
+ } else {
+ head->arg.pages = &head->pages[head->count];
+ page_offset = 0;
+ }
+
+ /* This server implementation supports only one Read chunk (of one
+ * or more segments) per message. The list walk is terminated once
+ * "position" changes.
+ */
+ page_no = 0;
+ last = false;
+ while (!last) {
+ u32 rs_handle, rs_length;
+ u64 rs_offset;
+
+ rs_handle = be32_to_cpu(*p++),
+ rs_length = be32_to_cpu(*p++);
+ p = xdr_decode_hyper(p, &rs_offset);
+
+ /* Examine next read segment */
+ if (*p == xdr_zero ||
+ ((*p != xdr_zero) && (be32_to_cpu(*(p + 1)) != position)))
+ last = true;
+
+ ret = svc_rdma_recv_read_segment(rdma, head, rqstp,
+ &page_no, &page_offset,
+ rs_handle, rs_length,
+ rs_offset, last);
+ if (ret < 0)
+ goto out;
+
+ p += 2;
+ }
+
+ /* Read list may need XDR round-up (see RFC 5666, s. 3.7) */
+ if (page_offset & 3) {
+ u32 pad = 4 - (page_offset & 3);
+
+ head->arg.tail[0].iov_len += pad;
+ head->arg.len += pad;
+ head->arg.buflen += pad;
+ page_offset += pad;
+ }
+
+ ret = 1;
+ if (position && position < head->arg.head[0].iov_len)
+ ret = svc_rdma_copy_tail(rqstp, head, position,
+ page_offset, page_no);
+ head->arg.head[0].iov_len = position;
+ head->position = position;
+
+ out:
+ /* Detach arg pages. svc_recv will replenish them */
+ for (page_no = 0;
+ &rqstp->rq_pages[page_no] < rqstp->rq_respages; page_no++)
+ rqstp->rq_pages[page_no] = NULL;
+ return ret;
+}
diff --git a/net/sunrpc/xprtrdma/svc_rdma_sendto.c b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
index 7150201..ba43f75 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_sendto.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
@@ -363,7 +363,7 @@ static int send_write_chunks(struct svcxprt_rdma *xprt,
}
}
/* Update the req with the number of chunks actually used */
- svc_rdma_xdr_encode_write_list(rdma_resp, chunk_no);
+ svc_rdma_old_encode_write_list(rdma_resp, chunk_no);
return rqstp->rq_res.page_len;
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index 3258af7..14ecac4 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -560,6 +560,7 @@ static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
INIT_LIST_HEAD(&cma_xprt->sc_read_complete_q);
INIT_LIST_HEAD(&cma_xprt->sc_frmr_q);
INIT_LIST_HEAD(&cma_xprt->sc_ctxts);
+ INIT_LIST_HEAD(&cma_xprt->sc_rw_ctxts);
INIT_LIST_HEAD(&cma_xprt->sc_maps);
init_waitqueue_head(&cma_xprt->sc_send_wait);
@@ -567,6 +568,7 @@ static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
spin_lock_init(&cma_xprt->sc_rq_dto_lock);
spin_lock_init(&cma_xprt->sc_frmr_q_lock);
spin_lock_init(&cma_xprt->sc_ctxt_lock);
+ spin_lock_init(&cma_xprt->sc_rw_ctxt_lock);
spin_lock_init(&cma_xprt->sc_map_lock);
if (listener)
@@ -1032,6 +1034,7 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
newxprt, newxprt->sc_cm_id);
dev = newxprt->sc_cm_id->device;
+ newxprt->sc_port_num = newxprt->sc_cm_id->port_num;
/* Qualify the transport resource defaults with the
* capabilities of this particular device */
@@ -1281,6 +1284,7 @@ static void __svc_rdma_free(struct work_struct *work)
}
rdma_dealloc_frmr_q(rdma);
+ svc_rdma_destroy_rw_ctxts(rdma);
svc_rdma_destroy_ctxts(rdma);
svc_rdma_destroy_maps(rdma);
@@ -1383,3 +1387,55 @@ int svc_rdma_send(struct svcxprt_rdma *xprt, struct ib_send_wr *wr)
}
return ret;
}
+
+/**
+ * svc_rdma_post_send - Post a chain of send WRs
+ * @rdma: RPC-over-RDMA transport control block
+ * @first_wr: chain of WRs to be posted
+ * @num_wrs: number of WRs to be posted
+ *
+ * Returns zero if one or more WRs were posted successfully, or
+ * a negative errno if none could be posted.
+ *
+ * Sleeps if transport's Send Queue is congested. If the connection
+ * closes, posts are still allowed, but they just flush through.
+ *
+ * Assumptions:
+ * - If ib_post_send() succeeds, only one completion is expected,
+ * even if one or more WRs are flushed. This is true when posting
+ * an rdma_rw_ctx or when posting a single signaled WR.
+ */
+int svc_rdma_post_send(struct svcxprt_rdma *rdma,
+ struct ib_send_wr *first_wr,
+ int num_wrs)
+{
+ struct svc_xprt *xprt = &rdma->sc_xprt;
+ struct ib_send_wr *bad_wr;
+ int ret;
+
+ do {
+ if ((atomic_sub_return(num_wrs, &rdma->sc_sq_avail) > 0)) {
+ ret = ib_post_send(rdma->sc_qp, first_wr, &bad_wr);
+ if (ret)
+ break;
+ return 0;
+ } else {
+ atomic_inc(&rdma_stat_sq_starve);
+ atomic_add(num_wrs, &rdma->sc_sq_avail);
+ wait_event(rdma->sc_send_wait,
+ atomic_read(&rdma->sc_sq_avail) > num_wrs);
+ }
+ } while (1);
+
+ pr_err("svcrdma: post_send rc=%d; SQ avail=%d/%u\n",
+ ret, atomic_read(&rdma->sc_sq_avail), rdma->sc_sq_depth);
+ set_bit(XPT_CLOSE, &xprt->xpt_flags);
+
+ /* If even one was posted, there will be a completion. */
+ if (bad_wr != first_wr)
+ return 0;
+
+ atomic_add(num_wrs, &rdma->sc_sq_avail);
+ wake_up(&rdma->sc_send_wait);
+ return -ENOTCONN;
+}
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 09/22] svcrdma: Use rdma_rw API in RPC reply path
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (7 preceding siblings ...)
2017-01-07 17:16 ` [PATCH v1 08/22] svcrdma: Introduce local rdma_rw API helpers Chuck Lever
@ 2017-01-07 17:16 ` Chuck Lever
2017-01-07 17:16 ` [PATCH v1 10/22] svcrdma: Backchannel sendto clean up Chuck Lever
` (15 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:16 UTC (permalink / raw)
To: linux-rdma, linux-nfs
The current svcrdma sendto code path posts one RDMA Write WR at a
time. Each of these Writes typically carries a small number of pages
(for instance, up to 30 pages for mlx4 devices). That means a 1MB
NFS READ reply requires 9 ib_post_send() calls for the Write WRs,
and one for the Send WR carrying the actual RPC Reply message.
Instead, use the new rdma_rw API. This gives two main benefits:
1. All Write WRs for one RDMA segment are posted in a single chain.
Just one doorbell for each RPC's Write chunk data.
2. The Write path can now use FRWR to register the Write buffers.
If the device's maximum page list depth is large, this means a
single Write WR is needed for each RPC's Write chunk data.
But also, the details of Write WR chain construction and memory
registration are taken care of elsewhere. svcrdma can focus on
RPC-over-RDMA protocol details.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 12 -
net/sunrpc/xprtrdma/svc_rdma_marshal.c | 39 --
net/sunrpc/xprtrdma/svc_rdma_sendto.c | 675 +++++++++++++-----------------
net/sunrpc/xprtrdma/svc_rdma_transport.c | 52 +-
4 files changed, 336 insertions(+), 442 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index e6eca34..765f91b 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -85,6 +85,7 @@ struct svc_rdma_op_ctxt {
enum dma_data_direction direction;
int count;
unsigned int mapped_sges;
+ struct ib_send_wr send_wr;
struct ib_sge sge[RPCSVC_MAXPAGES];
struct page *pages[RPCSVC_MAXPAGES];
};
@@ -145,6 +146,8 @@ struct svcxprt_rdma {
u32 sc_max_bc_requests;/* Backward credits */
int sc_max_req_size; /* Size of each RQ WR buf */
u8 sc_port_num;
+ struct ib_device *sc_device;
+ u32 sc_lkey;
struct ib_pd *sc_pd;
@@ -211,10 +214,6 @@ extern int svc_rdma_handle_bc_reply(struct rpc_xprt *xprt,
extern int svc_rdma_xdr_encode_error(struct svcxprt_rdma *,
struct rpcrdma_msg *,
enum rpcrdma_errcode, __be32 *);
-extern void svc_rdma_old_encode_write_list(struct rpcrdma_msg *, int);
-extern void svc_rdma_xdr_encode_reply_array(struct rpcrdma_write_array *, int);
-extern void svc_rdma_xdr_encode_array_chunk(struct rpcrdma_write_array *, int,
- __be32, __be64, u32);
extern void svc_rdma_xdr_encode_write_list(__be32 *rdma_resp, __be32 *wr_ch,
unsigned int consumed);
extern void svc_rdma_xdr_encode_reply_chunk(__be32 *rdma_resp, __be32 *rp_ch,
@@ -249,6 +248,11 @@ extern int svc_rdma_send_reply_chunk(struct svcxprt_rdma *rdma,
/* svc_rdma_sendto.c */
extern int svc_rdma_map_xdr(struct svcxprt_rdma *, struct xdr_buf *,
struct svc_rdma_req_map *, bool);
+extern int svc_rdma_map_reply_hdr(struct svcxprt_rdma *rdma,
+ struct svc_rdma_op_ctxt *ctxt,
+ __be32 *rdma_resp, unsigned int len);
+extern void svc_rdma_build_send_wr(struct svc_rdma_op_ctxt *ctxt,
+ int num_sge);
extern int svc_rdma_sendto(struct svc_rqst *);
extern void svc_rdma_send_error(struct svcxprt_rdma *, struct rpcrdma_msg *,
int);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_marshal.c b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
index 01cb4e1..6ba2fb6 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_marshal.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
@@ -217,45 +217,6 @@ unsigned int svc_rdma_xdr_get_reply_hdr_len(__be32 *rdma_resp)
return (unsigned long)p - (unsigned long)rdma_resp;
}
-void svc_rdma_old_encode_write_list(struct rpcrdma_msg *rmsgp, int chunks)
-{
- struct rpcrdma_write_array *ary;
-
- /* no read-list */
- rmsgp->rm_body.rm_chunks[0] = xdr_zero;
-
- /* write-array discrim */
- ary = (struct rpcrdma_write_array *)
- &rmsgp->rm_body.rm_chunks[1];
- ary->wc_discrim = xdr_one;
- ary->wc_nchunks = cpu_to_be32(chunks);
-
- /* write-list terminator */
- ary->wc_array[chunks].wc_target.rs_handle = xdr_zero;
-
- /* reply-array discriminator */
- ary->wc_array[chunks].wc_target.rs_length = xdr_zero;
-}
-
-void svc_rdma_xdr_encode_reply_array(struct rpcrdma_write_array *ary,
- int chunks)
-{
- ary->wc_discrim = xdr_one;
- ary->wc_nchunks = cpu_to_be32(chunks);
-}
-
-void svc_rdma_xdr_encode_array_chunk(struct rpcrdma_write_array *ary,
- int chunk_no,
- __be32 rs_handle,
- __be64 rs_offset,
- u32 write_len)
-{
- struct rpcrdma_segment *seg = &ary->wc_array[chunk_no].wc_target;
- seg->rs_handle = rs_handle;
- seg->rs_offset = rs_offset;
- seg->rs_length = cpu_to_be32(write_len);
-}
-
/* Write chunk is copied from Call header to Reply header. Length
* fields are updated to reflect number of bytes consumed in each
* segment.
diff --git a/net/sunrpc/xprtrdma/svc_rdma_sendto.c b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
index ba43f75..555daef 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_sendto.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
@@ -1,4 +1,5 @@
/*
+ * Copyright (c) 2016 Oracle. All rights reserved.
* Copyright (c) 2014 Open Grid Computing, Inc. All rights reserved.
* Copyright (c) 2005-2006 Network Appliance, Inc. All rights reserved.
*
@@ -40,6 +41,61 @@
* Author: Tom Tucker <tom@opengridcomputing.com>
*/
+/* Operation
+ *
+ * The main entry point is svc_rdma_sendto. This is called by the
+ * RPC server when an RPC Reply is ready to be transmitted to a client.
+ *
+ * The passed-in svc_rqst contains a struct xdr_buf which holds the
+ * XDR-encoded RPC Reply message. sendto must construct the RPC-over-RDMA
+ * transport header, post all Write WRs needed for this message, then post
+ * a Send WR conveying the transport header and the RPC message itself to
+ * the client.
+ *
+ * svc_rdma_sendto must fully transmit the Reply before returning, as
+ * the svc_rqst will be recycled as soon as sendto returns. Remaining
+ * resources referred to by the svc_rqst are also recycled at that time.
+ *
+ * Page Management
+ *
+ * The I/O that performs Reply transmission is asynchronous, and may
+ * complete well after sendto returns. Thus pages under I/O must be
+ * removed from the svc_rqst before sendto returns.
+ *
+ * The logic here depends on Send Queue and completion ordering. Since
+ * the Send WR is always posted last, it will always complete last. Thus
+ * when it completes, it is guaranteed that all previous Write WRs have
+ * also completed.
+ *
+ * Write WRs are constructed and posted. Each Write segment gets its own
+ * svc_rdma_rw_ctxt, allowing the Write completion handler to find and
+ * DMA-unmap the pages under I/O for that Write segment. The Write
+ * completion handler does not release any pages.
+ *
+ * When the Send WR is constructed, it also gets its own svc_rdma_op_ctxt.
+ * The Reply's pages are transferred into that ctxt, the Send WR is
+ * posted, and sendto returns.
+ *
+ * The svc_rdma_op_ctxt is presented when the Send WR completes. The
+ * Send completion handler finally releases the Reply's pages.
+ *
+ * This mechanism also assumes that completions on the transport's Send
+ * Completion Queue do not run in parallel. Otherwise a Write completion
+ * and Send completion running at the same time could release pages that
+ * are still DMA-mapped.
+ *
+ * Error Handling
+ *
+ * - If the Send WR is posted successfully, it will either complete
+ * successfully, or get flushed; either way, the Send completion handler
+ * releases the Reply's pages.
+ * - If the Send WR cannot be not posted, the forward path releases the
+ * Reply's pages.
+ *
+ * This handles the case, without the use of page reference counting,
+ * where two different Write segments send portions of the same page.
+ */
+
#include <linux/sunrpc/debug.h>
#include <linux/sunrpc/rpc_rdma.h>
#include <linux/spinlock.h>
@@ -123,45 +179,14 @@ int svc_rdma_map_xdr(struct svcxprt_rdma *xprt,
return 0;
}
-static dma_addr_t dma_map_xdr(struct svcxprt_rdma *xprt,
- struct xdr_buf *xdr,
- u32 xdr_off, size_t len, int dir)
-{
- struct page *page;
- dma_addr_t dma_addr;
- if (xdr_off < xdr->head[0].iov_len) {
- /* This offset is in the head */
- xdr_off += (unsigned long)xdr->head[0].iov_base & ~PAGE_MASK;
- page = virt_to_page(xdr->head[0].iov_base);
- } else {
- xdr_off -= xdr->head[0].iov_len;
- if (xdr_off < xdr->page_len) {
- /* This offset is in the page list */
- xdr_off += xdr->page_base;
- page = xdr->pages[xdr_off >> PAGE_SHIFT];
- xdr_off &= ~PAGE_MASK;
- } else {
- /* This offset is in the tail */
- xdr_off -= xdr->page_len;
- xdr_off += (unsigned long)
- xdr->tail[0].iov_base & ~PAGE_MASK;
- page = virt_to_page(xdr->tail[0].iov_base);
- }
- }
- dma_addr = ib_dma_map_page(xprt->sc_cm_id->device, page, xdr_off,
- min_t(size_t, PAGE_SIZE, len), dir);
- return dma_addr;
-}
-
/* Parse the RPC Call's transport header.
*/
-static void svc_rdma_get_write_arrays(struct rpcrdma_msg *rmsgp,
- struct rpcrdma_write_array **write,
- struct rpcrdma_write_array **reply)
+static void svc_rdma_get_write_arrays(__be32 *rdma_argp,
+ __be32 **write, __be32 **reply)
{
__be32 *p;
- p = (__be32 *)&rmsgp->rm_body.rm_chunks[0];
+ p = rdma_argp + rpcrdma_fixed_maxsz;
/* Read list */
while (*p++ != xdr_zero)
@@ -169,7 +194,7 @@ static void svc_rdma_get_write_arrays(struct rpcrdma_msg *rmsgp,
/* Write list */
if (*p != xdr_zero) {
- *write = (struct rpcrdma_write_array *)p;
+ *write = p;
while (*p++ != xdr_zero)
p += 1 + be32_to_cpu(*p) * 4;
} else {
@@ -179,7 +204,7 @@ static void svc_rdma_get_write_arrays(struct rpcrdma_msg *rmsgp,
/* Reply chunk */
if (*p != xdr_zero)
- *reply = (struct rpcrdma_write_array *)p;
+ *reply = p;
else
*reply = NULL;
}
@@ -189,360 +214,268 @@ static void svc_rdma_get_write_arrays(struct rpcrdma_msg *rmsgp,
* Invalidate, and responder chooses one rkey to invalidate.
*
* Find a candidate rkey to invalidate when sending a reply. Picks the
- * first rkey it finds in the chunks lists.
+ * first R_key it finds in the chunk lists.
*
* Returns zero if RPC's chunk lists are empty.
*/
-static u32 svc_rdma_get_inv_rkey(struct rpcrdma_msg *rdma_argp,
- struct rpcrdma_write_array *wr_ary,
- struct rpcrdma_write_array *rp_ary)
+static u32 svc_rdma_get_inv_rkey(__be32 *rdma_argp,
+ __be32 *wr_ch, __be32 *rp_ch)
{
- struct rpcrdma_read_chunk *rd_ary;
- struct rpcrdma_segment *arg_ch;
+ __be32 *p;
- rd_ary = (struct rpcrdma_read_chunk *)&rdma_argp->rm_body.rm_chunks[0];
- if (rd_ary->rc_discrim != xdr_zero)
- return be32_to_cpu(rd_ary->rc_target.rs_handle);
+ p = rdma_argp + rpcrdma_fixed_maxsz;
+ if (*p != xdr_zero) {
+ p += 2;
+ } else if (wr_ch && be32_to_cpup(wr_ch + 1)) {
+ p = wr_ch + 2;
+ } else if (rp_ch && be32_to_cpup(rp_ch + 1)) {
+ p = rp_ch + 2;
+ } else
+ return 0;
+ return be32_to_cpup(p);
+}
- if (wr_ary && be32_to_cpu(wr_ary->wc_nchunks)) {
- arg_ch = &wr_ary->wc_array[0].wc_target;
- return be32_to_cpu(arg_ch->rs_handle);
- }
+/* ib_dma_map_page() is used here because svc_rdma_dma_unmap()
+ * is used during completion to DMA-unmap this memory, and
+ * it uses ib_dma_unmap_page() exclusively.
+ */
+static int svc_rdma_dma_map_buf(struct svcxprt_rdma *rdma,
+ struct svc_rdma_op_ctxt *ctxt,
+ unsigned int sge_no,
+ unsigned char *base,
+ unsigned int len)
+{
+ unsigned long offset = (unsigned long)base & ~PAGE_MASK;
+ struct ib_device *dev = rdma->sc_device;
+ dma_addr_t dma_addr;
- if (rp_ary && be32_to_cpu(rp_ary->wc_nchunks)) {
- arg_ch = &rp_ary->wc_array[0].wc_target;
- return be32_to_cpu(arg_ch->rs_handle);
- }
+ dma_addr = ib_dma_map_page(dev, virt_to_page(base),
+ offset, len, DMA_TO_DEVICE);
+ if (ib_dma_mapping_error(dev, dma_addr))
+ return -EIO;
+ ctxt->sge[sge_no].addr = dma_addr;
+ ctxt->sge[sge_no].length = len;
+ ctxt->sge[sge_no].lkey = rdma->sc_lkey;
+ svc_rdma_count_mappings(rdma, ctxt);
return 0;
}
-/* Assumptions:
- * - The specified write_len can be represented in sc_max_sge * PAGE_SIZE
- */
-static int send_write(struct svcxprt_rdma *xprt, struct svc_rqst *rqstp,
- u32 rmr, u64 to,
- u32 xdr_off, int write_len,
- struct svc_rdma_req_map *vec)
+static int svc_rdma_dma_map_page(struct svcxprt_rdma *rdma,
+ struct svc_rdma_op_ctxt *ctxt,
+ unsigned int sge_no,
+ struct page *page,
+ unsigned int offset,
+ unsigned int len)
{
- struct ib_rdma_wr write_wr;
- struct ib_sge *sge;
- int xdr_sge_no;
- int sge_no;
- int sge_bytes;
- int sge_off;
- int bc;
- struct svc_rdma_op_ctxt *ctxt;
+ struct ib_device *dev = rdma->sc_device;
+ dma_addr_t dma_addr;
- if (vec->count > RPCSVC_MAXPAGES) {
- pr_err("svcrdma: Too many pages (%lu)\n", vec->count);
+ dma_addr = ib_dma_map_page(dev, page, offset, len, DMA_TO_DEVICE);
+ if (ib_dma_mapping_error(dev, dma_addr))
return -EIO;
- }
- dprintk("svcrdma: RDMA_WRITE rmr=%x, to=%llx, xdr_off=%d, "
- "write_len=%d, vec->sge=%p, vec->count=%lu\n",
- rmr, (unsigned long long)to, xdr_off,
- write_len, vec->sge, vec->count);
+ ctxt->sge[sge_no].addr = dma_addr;
+ ctxt->sge[sge_no].length = len;
+ ctxt->sge[sge_no].lkey = rdma->sc_lkey;
+ svc_rdma_count_mappings(rdma, ctxt);
+ return 0;
+}
- ctxt = svc_rdma_get_context(xprt);
+/**
+ * svc_rdma_map_reply_hdr - Map a RPC-over-RDMA transport header for a reply
+ * @rdma: controlling transport
+ * @ctxt: op_ctxt for the Send WR
+ * @rdma_resp: buffer containing transport header
+ * @len: length of transport header
+ *
+ * Returns 0 if successful, or a negative errno if a problem occurs.
+ */
+int svc_rdma_map_reply_hdr(struct svcxprt_rdma *rdma,
+ struct svc_rdma_op_ctxt *ctxt,
+ __be32 *rdma_resp,
+ unsigned int len)
+{
ctxt->direction = DMA_TO_DEVICE;
- sge = ctxt->sge;
-
- /* Find the SGE associated with xdr_off */
- for (bc = xdr_off, xdr_sge_no = 1; bc && xdr_sge_no < vec->count;
- xdr_sge_no++) {
- if (vec->sge[xdr_sge_no].iov_len > bc)
- break;
- bc -= vec->sge[xdr_sge_no].iov_len;
- }
-
- sge_off = bc;
- bc = write_len;
- sge_no = 0;
-
- /* Copy the remaining SGE */
- while (bc != 0) {
- sge_bytes = min_t(size_t,
- bc, vec->sge[xdr_sge_no].iov_len-sge_off);
- sge[sge_no].length = sge_bytes;
- sge[sge_no].addr =
- dma_map_xdr(xprt, &rqstp->rq_res, xdr_off,
- sge_bytes, DMA_TO_DEVICE);
- xdr_off += sge_bytes;
- if (ib_dma_mapping_error(xprt->sc_cm_id->device,
- sge[sge_no].addr))
- goto err;
- svc_rdma_count_mappings(xprt, ctxt);
- sge[sge_no].lkey = xprt->sc_pd->local_dma_lkey;
- ctxt->count++;
- sge_off = 0;
- sge_no++;
- xdr_sge_no++;
- if (xdr_sge_no > vec->count) {
- pr_err("svcrdma: Too many sges (%d)\n", xdr_sge_no);
- goto err;
- }
- bc -= sge_bytes;
- if (sge_no == xprt->sc_max_sge)
- break;
- }
-
- /* Prepare WRITE WR */
- memset(&write_wr, 0, sizeof write_wr);
- ctxt->cqe.done = svc_rdma_wc_write;
- write_wr.wr.wr_cqe = &ctxt->cqe;
- write_wr.wr.sg_list = &sge[0];
- write_wr.wr.num_sge = sge_no;
- write_wr.wr.opcode = IB_WR_RDMA_WRITE;
- write_wr.wr.send_flags = IB_SEND_SIGNALED;
- write_wr.rkey = rmr;
- write_wr.remote_addr = to;
-
- /* Post It */
- atomic_inc(&rdma_stat_write);
- if (svc_rdma_send(xprt, &write_wr.wr))
- goto err;
- return write_len - bc;
- err:
- svc_rdma_unmap_dma(ctxt);
- svc_rdma_put_context(ctxt, 0);
- return -EIO;
+ ctxt->pages[0] = virt_to_page(rdma_resp);
+ ctxt->count = 1;
+ return svc_rdma_dma_map_page(rdma, ctxt, 0, ctxt->pages[0], 0, len);
}
-noinline
-static int send_write_chunks(struct svcxprt_rdma *xprt,
- struct rpcrdma_write_array *wr_ary,
- struct rpcrdma_msg *rdma_resp,
- struct svc_rqst *rqstp,
- struct svc_rdma_req_map *vec)
+/* Load the xdr_buf into the ctxt's sge array, and DMA map each
+ * element as it is added.
+ *
+ * Returns the number of sge elements loaded on success, or
+ * a negative errno on failure.
+ */
+static int svc_rdma_map_reply_msg(struct svcxprt_rdma *rdma,
+ struct svc_rdma_op_ctxt *ctxt,
+ struct xdr_buf *xdr, __be32 *wr_ch)
{
- u32 xfer_len = rqstp->rq_res.page_len;
- int write_len;
- u32 xdr_off;
- int chunk_off;
- int chunk_no;
- int nchunks;
- struct rpcrdma_write_array *res_ary;
+ unsigned int len, sge_no, remaining, page_off;
+ struct page **ppages;
+ unsigned char *base;
+ u32 xdr_pad;
int ret;
- res_ary = (struct rpcrdma_write_array *)
- &rdma_resp->rm_body.rm_chunks[1];
-
- /* Write chunks start at the pagelist */
- nchunks = be32_to_cpu(wr_ary->wc_nchunks);
- for (xdr_off = rqstp->rq_res.head[0].iov_len, chunk_no = 0;
- xfer_len && chunk_no < nchunks;
- chunk_no++) {
- struct rpcrdma_segment *arg_ch;
- u64 rs_offset;
-
- arg_ch = &wr_ary->wc_array[chunk_no].wc_target;
- write_len = min(xfer_len, be32_to_cpu(arg_ch->rs_length));
-
- /* Prepare the response chunk given the length actually
- * written */
- xdr_decode_hyper((__be32 *)&arg_ch->rs_offset, &rs_offset);
- svc_rdma_xdr_encode_array_chunk(res_ary, chunk_no,
- arg_ch->rs_handle,
- arg_ch->rs_offset,
- write_len);
- chunk_off = 0;
- while (write_len) {
- ret = send_write(xprt, rqstp,
- be32_to_cpu(arg_ch->rs_handle),
- rs_offset + chunk_off,
- xdr_off,
- write_len,
- vec);
- if (ret <= 0)
- goto out_err;
- chunk_off += ret;
- xdr_off += ret;
- xfer_len -= ret;
- write_len -= ret;
+ sge_no = 1;
+
+ ret = svc_rdma_dma_map_buf(rdma, ctxt, sge_no++,
+ xdr->head[0].iov_base,
+ xdr->head[0].iov_len);
+ if (ret < 0)
+ return ret;
+
+ /* If a Write chunk is present, the xdr_buf's page list
+ * is not included inline. However the Upper Layer may
+ * have added XDR padding in the tail buffer, and that
+ * should not be included inline.
+ */
+ if (wr_ch) {
+ base = xdr->tail[0].iov_base;
+ len = xdr->tail[0].iov_len;
+ xdr_pad = xdr_padsize(xdr->page_len);
+
+ if (len && xdr_pad) {
+ base += xdr_pad;
+ len -= xdr_pad;
}
+
+ goto tail;
}
- /* Update the req with the number of chunks actually used */
- svc_rdma_old_encode_write_list(rdma_resp, chunk_no);
- return rqstp->rq_res.page_len;
+ ppages = xdr->pages + (xdr->page_base >> PAGE_SHIFT);
+ page_off = xdr->page_base & ~PAGE_MASK;
+ remaining = xdr->page_len;
+ while (remaining) {
+ len = min_t(u32, PAGE_SIZE - page_off, remaining);
+
+ ret = svc_rdma_dma_map_page(rdma, ctxt, sge_no++,
+ *ppages++, page_off, len);
+ if (ret < 0)
+ return 0;
+
+ remaining -= len;
+ page_off = 0;
+ }
+
+ base = xdr->tail[0].iov_base;
+ len = xdr->tail[0].iov_len;
+tail:
+ if (len) {
+ ret = svc_rdma_dma_map_buf(rdma, ctxt, sge_no++, base, len);
+ if (ret < 0)
+ return ret;
+ }
-out_err:
- pr_err("svcrdma: failed to send write chunks, rc=%d\n", ret);
- return -EIO;
+ return sge_no - 1;
}
-noinline
-static int send_reply_chunks(struct svcxprt_rdma *xprt,
- struct rpcrdma_write_array *rp_ary,
- struct rpcrdma_msg *rdma_resp,
- struct svc_rqst *rqstp,
- struct svc_rdma_req_map *vec)
+/* The svc_rqst and all resources it owns are released as soon as
+ * svc_rdma_sendto returns. Transfer pages under I/O to the ctxt
+ * so they are released by the Send completion handler.
+ */
+static void svc_rdma_save_io_pages(struct svc_rqst *rqstp,
+ struct svc_rdma_op_ctxt *ctxt)
{
- u32 xfer_len = rqstp->rq_res.len;
- int write_len;
- u32 xdr_off;
- int chunk_no;
- int chunk_off;
- int nchunks;
- struct rpcrdma_segment *ch;
- struct rpcrdma_write_array *res_ary;
- int ret;
+ int i, pages = rqstp->rq_next_page - rqstp->rq_respages;
- /* XXX: need to fix when reply lists occur with read-list and or
- * write-list */
- res_ary = (struct rpcrdma_write_array *)
- &rdma_resp->rm_body.rm_chunks[2];
-
- /* xdr offset starts at RPC message */
- nchunks = be32_to_cpu(rp_ary->wc_nchunks);
- for (xdr_off = 0, chunk_no = 0;
- xfer_len && chunk_no < nchunks;
- chunk_no++) {
- u64 rs_offset;
- ch = &rp_ary->wc_array[chunk_no].wc_target;
- write_len = min(xfer_len, be32_to_cpu(ch->rs_length));
-
- /* Prepare the reply chunk given the length actually
- * written */
- xdr_decode_hyper((__be32 *)&ch->rs_offset, &rs_offset);
- svc_rdma_xdr_encode_array_chunk(res_ary, chunk_no,
- ch->rs_handle, ch->rs_offset,
- write_len);
- chunk_off = 0;
- while (write_len) {
- ret = send_write(xprt, rqstp,
- be32_to_cpu(ch->rs_handle),
- rs_offset + chunk_off,
- xdr_off,
- write_len,
- vec);
- if (ret <= 0)
- goto out_err;
- chunk_off += ret;
- xdr_off += ret;
- xfer_len -= ret;
- write_len -= ret;
- }
+ ctxt->count += pages;
+ for (i = 0; i < pages; i++) {
+ ctxt->pages[i + 1] = rqstp->rq_respages[i];
+ rqstp->rq_respages[i] = NULL;
}
- /* Update the req with the number of chunks actually used */
- svc_rdma_xdr_encode_reply_array(res_ary, chunk_no);
+ rqstp->rq_next_page = rqstp->rq_respages + 1;
+}
- return rqstp->rq_res.len;
+/**
+ * svc_rdma_build_send_wr - Set up a Send Work Request
+ * @ctxt: op_ctxt for transmitting the Send WR
+ *
+ */
+void svc_rdma_build_send_wr(struct svc_rdma_op_ctxt *ctxt, int num_sge)
+{
+ struct ib_send_wr *send_wr;
-out_err:
- pr_err("svcrdma: failed to send reply chunks, rc=%d\n", ret);
- return -EIO;
+ send_wr = &ctxt->send_wr;
+ send_wr->next = NULL;
+ ctxt->cqe.done = svc_rdma_wc_send;
+ send_wr->wr_cqe = &ctxt->cqe;
+ send_wr->sg_list = ctxt->sge;
+ send_wr->num_sge = num_sge;
+ send_wr->opcode = IB_WR_SEND;
+ send_wr->send_flags = IB_SEND_SIGNALED;
+
+ dprintk("svcrdma: posting Send WR with %u sge(s)\n",
+ send_wr->num_sge);
}
-/* This function prepares the portion of the RPCRDMA message to be
- * sent in the RDMA_SEND. This function is called after data sent via
- * RDMA has already been transmitted. There are three cases:
- * - The RPCRDMA header, RPC header, and payload are all sent in a
- * single RDMA_SEND. This is the "inline" case.
- * - The RPCRDMA header and some portion of the RPC header and data
- * are sent via this RDMA_SEND and another portion of the data is
- * sent via RDMA.
- * - The RPCRDMA header [NOMSG] is sent in this RDMA_SEND and the RPC
- * header and data are all transmitted via RDMA.
- * In all three cases, this function prepares the RPCRDMA header in
- * sge[0], the 'type' parameter indicates the type to place in the
- * RPCRDMA header, and the 'byte_count' field indicates how much of
- * the XDR to include in this RDMA_SEND. NB: The offset of the payload
- * to send is zero in the XDR.
+/* Prepare the portion of the RPC Reply that will be transmitted
+ * via RDMA Send. The RPC-over-RDMA transport header is prepared
+ * in sge[0], and the RPC xdr_buf is prepared in following sges.
+ *
+ * Depending on whether a Write list or Reply chunk is present,
+ * the server may send all, a portion of, or none of the xdr_buf.
+ * In the latter case, only the transport header (sge[0]) is
+ * transmitted.
+ *
+ * RDMA Send is the last step of transmitting an RPC reply. Pages
+ * involved in the earlier RDMA Writes are here transferred out
+ * of the rqstp and into the ctxt's page array. These pages are
+ * DMA unmapped by each Write completion, but the subsequent Send
+ * completion finally releases these pages.
+ *
+ * Returns zero or a negative errno.
+ *
+ * Assumptions:
+ * - The Reply's transport header will never be larger than a page.
*/
-static int send_reply(struct svcxprt_rdma *rdma,
- struct svc_rqst *rqstp,
- struct page *page,
- struct rpcrdma_msg *rdma_resp,
- struct svc_rdma_req_map *vec,
- int byte_count,
- u32 inv_rkey)
+static int svc_rdma_send_reply_msg(struct svcxprt_rdma *rdma,
+ __be32 *rdma_argp, __be32 *rdma_resp,
+ struct svc_rqst *rqstp,
+ __be32 *wr_ch, __be32 *rp_ch)
{
struct svc_rdma_op_ctxt *ctxt;
- struct ib_send_wr send_wr;
- u32 xdr_off;
- int sge_no;
- int sge_bytes;
- int page_no;
- int pages;
- int ret = -EIO;
+ struct ib_send_wr *send_wr;
+ int ret;
+
+ dprintk("svcrdma: sending %s reply: head=%zu, pagelen=%u, tail=%zu\n",
+ (rp_ch ? "RDMA_NOMSG" : "RDMA_MSG"),
+ rqstp->rq_res.head[0].iov_len,
+ rqstp->rq_res.page_len,
+ rqstp->rq_res.tail[0].iov_len);
- /* Prepare the context */
ctxt = svc_rdma_get_context(rdma);
- ctxt->direction = DMA_TO_DEVICE;
- ctxt->pages[0] = page;
- ctxt->count = 1;
- /* Prepare the SGE for the RPCRDMA Header */
- ctxt->sge[0].lkey = rdma->sc_pd->local_dma_lkey;
- ctxt->sge[0].length =
- svc_rdma_xdr_get_reply_hdr_len((__be32 *)rdma_resp);
- ctxt->sge[0].addr =
- ib_dma_map_page(rdma->sc_cm_id->device, page, 0,
- ctxt->sge[0].length, DMA_TO_DEVICE);
- if (ib_dma_mapping_error(rdma->sc_cm_id->device, ctxt->sge[0].addr))
+ ret = svc_rdma_map_reply_hdr(rdma, ctxt, rdma_resp,
+ svc_rdma_xdr_get_reply_hdr_len(rdma_resp));
+ if (ret < 0)
goto err;
- svc_rdma_count_mappings(rdma, ctxt);
-
- ctxt->direction = DMA_TO_DEVICE;
- /* Map the payload indicated by 'byte_count' */
- xdr_off = 0;
- for (sge_no = 1; byte_count && sge_no < vec->count; sge_no++) {
- sge_bytes = min_t(size_t, vec->sge[sge_no].iov_len, byte_count);
- byte_count -= sge_bytes;
- ctxt->sge[sge_no].addr =
- dma_map_xdr(rdma, &rqstp->rq_res, xdr_off,
- sge_bytes, DMA_TO_DEVICE);
- xdr_off += sge_bytes;
- if (ib_dma_mapping_error(rdma->sc_cm_id->device,
- ctxt->sge[sge_no].addr))
+ if (!rp_ch) {
+ ret = svc_rdma_map_reply_msg(rdma, ctxt, &rqstp->rq_res, wr_ch);
+ if (ret < 0)
goto err;
- svc_rdma_count_mappings(rdma, ctxt);
- ctxt->sge[sge_no].lkey = rdma->sc_pd->local_dma_lkey;
- ctxt->sge[sge_no].length = sge_bytes;
- }
- if (byte_count != 0) {
- pr_err("svcrdma: Could not map %d bytes\n", byte_count);
- goto err;
}
- /* Save all respages in the ctxt and remove them from the
- * respages array. They are our pages until the I/O
- * completes.
- */
- pages = rqstp->rq_next_page - rqstp->rq_respages;
- for (page_no = 0; page_no < pages; page_no++) {
- ctxt->pages[page_no+1] = rqstp->rq_respages[page_no];
- ctxt->count++;
- rqstp->rq_respages[page_no] = NULL;
- }
- rqstp->rq_next_page = rqstp->rq_respages + 1;
+ svc_rdma_save_io_pages(rqstp, ctxt);
- if (sge_no > rdma->sc_max_sge) {
- pr_err("svcrdma: Too many sges (%d)\n", sge_no);
- goto err;
+ svc_rdma_build_send_wr(ctxt, 1 + ret);
+ send_wr = &ctxt->send_wr;
+ if (rdma->sc_snd_w_inv) {
+ send_wr->ex.invalidate_rkey =
+ svc_rdma_get_inv_rkey(rdma_argp, wr_ch, rp_ch);
+ if (send_wr->ex.invalidate_rkey)
+ send_wr->opcode = IB_WR_SEND_WITH_INV;
}
- memset(&send_wr, 0, sizeof send_wr);
- ctxt->cqe.done = svc_rdma_wc_send;
- send_wr.wr_cqe = &ctxt->cqe;
- send_wr.sg_list = ctxt->sge;
- send_wr.num_sge = sge_no;
- if (inv_rkey) {
- send_wr.opcode = IB_WR_SEND_WITH_INV;
- send_wr.ex.invalidate_rkey = inv_rkey;
- } else
- send_wr.opcode = IB_WR_SEND;
- send_wr.send_flags = IB_SEND_SIGNALED;
-
- ret = svc_rdma_send(rdma, &send_wr);
+ ret = svc_rdma_send(rdma, send_wr);
if (ret)
goto err;
return 0;
- err:
+err:
+ pr_err("svcrdma: failed to post Send WR (%d)\n", ret);
svc_rdma_unmap_dma(ctxt);
svc_rdma_put_context(ctxt, 1);
return ret;
@@ -552,38 +485,31 @@ void svc_rdma_prep_reply_hdr(struct svc_rqst *rqstp)
{
}
+/**
+ * svc_rdma_sendto - Transmit an RPC reply
+ * @rqstp: processed RPC request, reply XDR already in ::rq_res
+ *
+ * Returns zero if the RPC reply and results have been successfully
+ * posted, or a negative errno is returned and the connection is
+ * terminated.
+ */
int svc_rdma_sendto(struct svc_rqst *rqstp)
{
struct svc_xprt *xprt = rqstp->rq_xprt;
struct svcxprt_rdma *rdma =
container_of(xprt, struct svcxprt_rdma, sc_xprt);
- struct rpcrdma_msg *rdma_argp;
- struct rpcrdma_msg *rdma_resp;
- struct rpcrdma_write_array *wr_ary, *rp_ary;
- int ret;
- int inline_bytes;
+ __be32 *rdma_argp, *rdma_resp, *wr_ch, *rp_ch;
struct page *res_page;
- struct svc_rdma_req_map *vec;
- u32 inv_rkey;
-
- dprintk("svcrdma: sending response for rqstp=%p\n", rqstp);
+ int ret;
- /* Get the RDMA request header. The receive logic always
- * places this at the start of page 0.
+ /* Find the call's chunk lists to decide how to send the reply.
+ * Receive places the Call's xprt header at the start of page 0.
*/
rdma_argp = page_address(rqstp->rq_pages[0]);
- svc_rdma_get_write_arrays(rdma_argp, &wr_ary, &rp_ary);
+ svc_rdma_get_write_arrays(rdma_argp, &wr_ch, &rp_ch);
- inv_rkey = 0;
- if (rdma->sc_snd_w_inv)
- inv_rkey = svc_rdma_get_inv_rkey(rdma_argp, wr_ary, rp_ary);
-
- /* Build an req vec for the XDR */
- vec = svc_rdma_get_req_map(rdma);
- ret = svc_rdma_map_xdr(rdma, &rqstp->rq_res, vec, wr_ary != NULL);
- if (ret)
- goto err0;
- inline_bytes = rqstp->rq_res.len;
+ dprintk("svcrdma: preparing response for XID 0x%08x\n",
+ be32_to_cpup(rdma_argp));
/* Create the RDMA response header. xprt->xpt_mutex,
* acquired in svc_send(), serializes RPC replies. The
@@ -596,44 +522,33 @@ int svc_rdma_sendto(struct svc_rqst *rqstp)
if (!res_page)
goto err0;
rdma_resp = page_address(res_page);
- svc_rdma_xdr_encode_reply_header(rdma, (__be32 *)rdma_argp,
- (__be32 *)rdma_resp,
- rp_ary ? rdma_nomsg : rdma_msg);
-
- /* Send any write-chunk data and build resp write-list */
- if (wr_ary) {
- ret = send_write_chunks(rdma, wr_ary, rdma_resp, rqstp, vec);
+ svc_rdma_xdr_encode_reply_header(rdma, rdma_argp, rdma_resp,
+ rp_ch ? rdma_nomsg : rdma_msg);
+ if (wr_ch) {
+ ret = svc_rdma_send_write_list(rdma, wr_ch, rdma_resp,
+ &rqstp->rq_res);
if (ret < 0)
goto err1;
- inline_bytes -= ret + xdr_padsize(ret);
}
-
- /* Send any reply-list data and update resp reply-list */
- if (rp_ary) {
- ret = send_reply_chunks(rdma, rp_ary, rdma_resp, rqstp, vec);
+ if (rp_ch) {
+ ret = svc_rdma_send_reply_chunk(rdma, rp_ch, rdma_resp,
+ &rqstp->rq_res);
if (ret < 0)
goto err1;
- inline_bytes -= ret;
}
- /* Post a fresh Receive buffer _before_ sending the reply */
ret = svc_rdma_post_recv(rdma, GFP_KERNEL);
if (ret)
goto err1;
-
- ret = send_reply(rdma, rqstp, res_page, rdma_resp, vec,
- inline_bytes, inv_rkey);
+ ret = svc_rdma_send_reply_msg(rdma, rdma_argp, rdma_resp, rqstp,
+ wr_ch, rp_ch);
if (ret < 0)
goto err0;
-
- svc_rdma_put_req_map(rdma, vec);
- dprintk("svcrdma: send_reply returns %d\n", ret);
return ret;
err1:
put_page(res_page);
err0:
- svc_rdma_put_req_map(rdma, vec);
pr_err("svcrdma: Could not send reply, err=%d. Closing transport.\n",
ret);
set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index 14ecac4..a6ddd8f 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -962,29 +962,11 @@ static unsigned int svc_rdma_read_sqes_per_credit(struct svcxprt_rdma *newxprt)
attrs->max_fast_reg_page_list_len) * 3;
}
-static unsigned int svc_rdma_write_sqes_per_credit(struct svcxprt_rdma *newxprt)
-{
- return DIV_ROUND_UP(RPCSVC_MAXPAGES, newxprt->sc_max_sge);
-}
-
static unsigned int svc_rdma_sq_depth(struct svcxprt_rdma *newxprt)
{
unsigned int sqes_per_credit;
- /* Estimate SQEs per credit assuming a full Read chunk payload
- * and a full Write chunk payload (possible with krb5i/p). Each
- * credit will consume Read WRs then Write WRs, serially, so
- * we need just the larger of the two, not the sum.
- *
- * This is not an upper bound. Clients can break chunks into
- * arbitrarily many segments. However, if more SQEs are needed
- * then are available, the server has Send Queue accounting to
- * wait until enough SQEs are ready. But we want that waiting
- * to be very rare.
- */
- sqes_per_credit = max_t(unsigned int,
- svc_rdma_read_sqes_per_credit(newxprt),
- svc_rdma_write_sqes_per_credit(newxprt));
+ sqes_per_credit = svc_rdma_read_sqes_per_credit(newxprt);
/* RDMA Sends per credit */
sqes_per_credit += 1;
@@ -992,6 +974,33 @@ static unsigned int svc_rdma_sq_depth(struct svcxprt_rdma *newxprt)
return sqes_per_credit * newxprt->sc_rq_depth;
}
+/* Estimate ctxs per credit assuming a full Read chunk payload
+ * and a full Write chunk payload (possible with krb5i/p). Each
+ * credit will consume Read WRs then Write WRs serially.
+ *
+ * Unfortunately we have no idea exactly how many ctxs will be
+ * needed.
+ *
+ * The RPC-over-RDMA protocol doesn't provide a way for a client
+ * to inform servers about how many segments it may send per RPC.
+ * Clients can break chunks into arbitrarily many segments. We
+ * need more than one ctx per credit when the client can send
+ * only small segments, or uses multi-segment Reply chunks.
+ *
+ * If more ctxs are needed then are available, the server has
+ * Send Queue accounting to wait until enough ctxs are available.
+ * We want that waiting to be very rare.
+ *
+ * For the moment, allocate one ctx per credit.
+ *
+ * Send Queue resources for these ctxs are provisioned by
+ * rdma_create_qp().
+ */
+static unsigned int svc_rdma_max_rdma_ctxs(struct svcxprt_rdma *newxprt)
+{
+ return newxprt->sc_max_requests;
+}
+
/*
* This is the xpo_recvfrom function for listening endpoints. Its
* purpose is to accept incoming connections. The CMA callback handler
@@ -1034,6 +1043,7 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
newxprt, newxprt->sc_cm_id);
dev = newxprt->sc_cm_id->device;
+ newxprt->sc_device = dev;
newxprt->sc_port_num = newxprt->sc_cm_id->port_num;
/* Qualify the transport resource defaults with the
@@ -1070,6 +1080,8 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
dprintk("svcrdma: error creating PD for connect request\n");
goto errout;
}
+ newxprt->sc_lkey = newxprt->sc_pd->local_dma_lkey;
+
newxprt->sc_sq_cq = ib_alloc_cq(dev, newxprt, newxprt->sc_sq_depth,
0, IB_POLL_WORKQUEUE);
if (IS_ERR(newxprt->sc_sq_cq)) {
@@ -1086,6 +1098,8 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
memset(&qp_attr, 0, sizeof qp_attr);
qp_attr.event_handler = qp_event_handler;
qp_attr.qp_context = &newxprt->sc_xprt;
+ qp_attr.port_num = newxprt->sc_cm_id->port_num;
+ qp_attr.cap.max_rdma_ctxs = svc_rdma_max_rdma_ctxs(newxprt);
qp_attr.cap.max_send_wr = newxprt->sc_sq_depth;
qp_attr.cap.max_recv_wr = newxprt->sc_rq_depth;
qp_attr.cap.max_send_sge = newxprt->sc_max_sge;
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 10/22] svcrdma: Backchannel sendto clean up
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (8 preceding siblings ...)
2017-01-07 17:16 ` [PATCH v1 09/22] svcrdma: Use rdma_rw API in RPC reply path Chuck Lever
@ 2017-01-07 17:16 ` Chuck Lever
2017-01-07 17:16 ` [PATCH v1 11/22] svcrdma: Clean up RDMA_ERROR path Chuck Lever
` (14 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:16 UTC (permalink / raw)
To: linux-rdma, linux-nfs
The backchannel uses part of the sendto path to transmit callback
requests to clients. Refactor svc_rdma_bc_send_request to use the
new APIs.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 2 -
net/sunrpc/xprtrdma/svc_rdma_backchannel.c | 56 +++++++----------------
net/sunrpc/xprtrdma/svc_rdma_sendto.c | 68 ----------------------------
3 files changed, 16 insertions(+), 110 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 765f91b..ea7867a 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -246,8 +246,6 @@ extern int svc_rdma_send_reply_chunk(struct svcxprt_rdma *rdma,
struct xdr_buf *xdr);
/* svc_rdma_sendto.c */
-extern int svc_rdma_map_xdr(struct svcxprt_rdma *, struct xdr_buf *,
- struct svc_rdma_req_map *, bool);
extern int svc_rdma_map_reply_hdr(struct svcxprt_rdma *rdma,
struct svc_rdma_op_ctxt *ctxt,
__be32 *rdma_resp, unsigned int len);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_backchannel.c b/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
index 288e35c..7d49545 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
@@ -89,9 +89,9 @@ int svc_rdma_handle_bc_reply(struct rpc_xprt *xprt, struct rpcrdma_msg *rmsgp,
* Caller holds the connection's mutex and has already marshaled
* the RPC/RDMA request.
*
- * This is similar to svc_rdma_reply, but takes an rpc_rqst
- * instead, does not support chunks, and avoids blocking memory
- * allocation.
+ * This is similar to svc_rdma_send_reply_msg, but takes a struct
+ * rpc_rqst instead, does not support chunks, and avoids blocking
+ * memory allocation.
*
* XXX: There is still an opportunity to block in svc_rdma_send()
* if there are no SQ entries to post the Send. This may occur if
@@ -100,60 +100,36 @@ int svc_rdma_handle_bc_reply(struct rpc_xprt *xprt, struct rpcrdma_msg *rmsgp,
static int svc_rdma_bc_sendto(struct svcxprt_rdma *rdma,
struct rpc_rqst *rqst)
{
- struct xdr_buf *sndbuf = &rqst->rq_snd_buf;
struct svc_rdma_op_ctxt *ctxt;
- struct svc_rdma_req_map *vec;
- struct ib_send_wr send_wr;
int ret;
- vec = svc_rdma_get_req_map(rdma);
- ret = svc_rdma_map_xdr(rdma, sndbuf, vec, false);
- if (ret)
+ ctxt = svc_rdma_get_context(rdma);
+
+ /* rpcrdma_bc_send_request builds the transport header and
+ * the backchannel RPC message in the same buffer. Thus only
+ * one SGE is needed to send both.
+ */
+ ret = svc_rdma_map_reply_hdr(rdma, ctxt, rqst->rq_buffer,
+ rqst->rq_snd_buf.len);
+ if (ret < 0)
goto out_err;
ret = svc_rdma_repost_recv(rdma, GFP_NOIO);
if (ret)
goto out_err;
- ctxt = svc_rdma_get_context(rdma);
- ctxt->pages[0] = virt_to_page(rqst->rq_buffer);
- ctxt->count = 1;
-
- ctxt->direction = DMA_TO_DEVICE;
- ctxt->sge[0].lkey = rdma->sc_pd->local_dma_lkey;
- ctxt->sge[0].length = sndbuf->len;
- ctxt->sge[0].addr =
- ib_dma_map_page(rdma->sc_cm_id->device, ctxt->pages[0], 0,
- sndbuf->len, DMA_TO_DEVICE);
- if (ib_dma_mapping_error(rdma->sc_cm_id->device, ctxt->sge[0].addr)) {
- ret = -EIO;
- goto out_unmap;
- }
- svc_rdma_count_mappings(rdma, ctxt);
-
- memset(&send_wr, 0, sizeof(send_wr));
- ctxt->cqe.done = svc_rdma_wc_send;
- send_wr.wr_cqe = &ctxt->cqe;
- send_wr.sg_list = ctxt->sge;
- send_wr.num_sge = 1;
- send_wr.opcode = IB_WR_SEND;
- send_wr.send_flags = IB_SEND_SIGNALED;
-
- ret = svc_rdma_send(rdma, &send_wr);
+ svc_rdma_build_send_wr(ctxt, 1);
+ ret = svc_rdma_send(rdma, &ctxt->send_wr);
if (ret) {
+ svc_rdma_unmap_dma(ctxt);
+ svc_rdma_put_context(ctxt, 1);
ret = -EIO;
- goto out_unmap;
}
out_err:
- svc_rdma_put_req_map(rdma, vec);
dprintk("svcrdma: %s returns %d\n", __func__, ret);
return ret;
-out_unmap:
- svc_rdma_unmap_dma(ctxt);
- svc_rdma_put_context(ctxt, 1);
- goto out_err;
}
/* Server-side transport endpoint wants a whole page for its send
diff --git a/net/sunrpc/xprtrdma/svc_rdma_sendto.c b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
index 555daef..5e75187 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_sendto.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
@@ -111,74 +111,6 @@ static u32 xdr_padsize(u32 len)
return (len & 3) ? (4 - (len & 3)) : 0;
}
-int svc_rdma_map_xdr(struct svcxprt_rdma *xprt,
- struct xdr_buf *xdr,
- struct svc_rdma_req_map *vec,
- bool write_chunk_present)
-{
- int sge_no;
- u32 sge_bytes;
- u32 page_bytes;
- u32 page_off;
- int page_no;
-
- if (xdr->len !=
- (xdr->head[0].iov_len + xdr->page_len + xdr->tail[0].iov_len)) {
- pr_err("svcrdma: %s: XDR buffer length error\n", __func__);
- return -EIO;
- }
-
- /* Skip the first sge, this is for the RPCRDMA header */
- sge_no = 1;
-
- /* Head SGE */
- vec->sge[sge_no].iov_base = xdr->head[0].iov_base;
- vec->sge[sge_no].iov_len = xdr->head[0].iov_len;
- sge_no++;
-
- /* pages SGE */
- page_no = 0;
- page_bytes = xdr->page_len;
- page_off = xdr->page_base;
- while (page_bytes) {
- vec->sge[sge_no].iov_base =
- page_address(xdr->pages[page_no]) + page_off;
- sge_bytes = min_t(u32, page_bytes, (PAGE_SIZE - page_off));
- page_bytes -= sge_bytes;
- vec->sge[sge_no].iov_len = sge_bytes;
-
- sge_no++;
- page_no++;
- page_off = 0; /* reset for next time through loop */
- }
-
- /* Tail SGE */
- if (xdr->tail[0].iov_len) {
- unsigned char *base = xdr->tail[0].iov_base;
- size_t len = xdr->tail[0].iov_len;
- u32 xdr_pad = xdr_padsize(xdr->page_len);
-
- if (write_chunk_present && xdr_pad) {
- base += xdr_pad;
- len -= xdr_pad;
- }
-
- if (len) {
- vec->sge[sge_no].iov_base = base;
- vec->sge[sge_no].iov_len = len;
- sge_no++;
- }
- }
-
- dprintk("svcrdma: %s: sge_no %d page_no %d "
- "page_base %u page_len %u head_len %zu tail_len %zu\n",
- __func__, sge_no, page_no, xdr->page_base, xdr->page_len,
- xdr->head[0].iov_len, xdr->tail[0].iov_len);
-
- vec->count = sge_no;
- return 0;
-}
-
/* Parse the RPC Call's transport header.
*/
static void svc_rdma_get_write_arrays(__be32 *rdma_argp,
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 11/22] svcrdma: Clean up RDMA_ERROR path
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (9 preceding siblings ...)
2017-01-07 17:16 ` [PATCH v1 10/22] svcrdma: Backchannel sendto clean up Chuck Lever
@ 2017-01-07 17:16 ` Chuck Lever
2017-01-07 17:16 ` [PATCH v1 12/22] svcrdma: Reduce size of sge array in struct svc_rdma_op_ctxt Chuck Lever
` (13 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:16 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Now that svc_rdma_sendto has been renovated, svc_rdma_send_error can
be refactored to reduce code duplication and remove C structure-
based XDR encoding.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/rpc_rdma.h | 3 ++
include/linux/sunrpc/svc_rdma.h | 11 +++---
net/sunrpc/xprtrdma/svc_rdma_marshal.c | 31 ++++++++++--------
net/sunrpc/xprtrdma/svc_rdma_recvfrom.c | 2 +
net/sunrpc/xprtrdma/svc_rdma_sendto.c | 54 ++++++++++---------------------
5 files changed, 45 insertions(+), 56 deletions(-)
diff --git a/include/linux/sunrpc/rpc_rdma.h b/include/linux/sunrpc/rpc_rdma.h
index 245fc59..b7e85b3 100644
--- a/include/linux/sunrpc/rpc_rdma.h
+++ b/include/linux/sunrpc/rpc_rdma.h
@@ -143,6 +143,9 @@ enum rpcrdma_proc {
#define rdma_done cpu_to_be32(RDMA_DONE)
#define rdma_error cpu_to_be32(RDMA_ERROR)
+#define err_vers cpu_to_be32(ERR_VERS)
+#define err_chunk cpu_to_be32(ERR_CHUNK)
+
/*
* Private extension to RPC-over-RDMA Version One.
* Message passed during RDMA-CM connection set-up.
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index ea7867a..65c2425 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -211,9 +211,10 @@ extern int svc_rdma_handle_bc_reply(struct rpc_xprt *xprt,
/* svc_rdma_marshal.c */
extern int svc_rdma_xdr_decode_req(struct xdr_buf *);
-extern int svc_rdma_xdr_encode_error(struct svcxprt_rdma *,
- struct rpcrdma_msg *,
- enum rpcrdma_errcode, __be32 *);
+extern unsigned int svc_rdma_xdr_encode_error(struct svcxprt_rdma *rdma,
+ __be32 *rdma_argp,
+ __be32 *rdma_resp,
+ __be32 errcode);
extern void svc_rdma_xdr_encode_write_list(__be32 *rdma_resp, __be32 *wr_ch,
unsigned int consumed);
extern void svc_rdma_xdr_encode_reply_chunk(__be32 *rdma_resp, __be32 *rp_ch,
@@ -252,8 +253,8 @@ extern int svc_rdma_map_reply_hdr(struct svcxprt_rdma *rdma,
extern void svc_rdma_build_send_wr(struct svc_rdma_op_ctxt *ctxt,
int num_sge);
extern int svc_rdma_sendto(struct svc_rqst *);
-extern void svc_rdma_send_error(struct svcxprt_rdma *, struct rpcrdma_msg *,
- int);
+extern void svc_rdma_send_error(struct svcxprt_rdma *rdma, __be32 *rdma_argp,
+ int status);
/* svc_rdma_transport.c */
extern void svc_rdma_wc_send(struct ib_cq *, struct ib_wc *);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_marshal.c b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
index 6ba2fb6..87cb8fd 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_marshal.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_marshal.c
@@ -167,23 +167,26 @@ int svc_rdma_xdr_decode_req(struct xdr_buf *rq_arg)
return -EINVAL;
}
-int svc_rdma_xdr_encode_error(struct svcxprt_rdma *xprt,
- struct rpcrdma_msg *rmsgp,
- enum rpcrdma_errcode err, __be32 *va)
+unsigned int svc_rdma_xdr_encode_error(struct svcxprt_rdma *rdma,
+ __be32 *rdma_argp, __be32 *rdma_resp,
+ __be32 errcode)
{
- __be32 *startp = va;
-
- *va++ = rmsgp->rm_xid;
- *va++ = rmsgp->rm_vers;
- *va++ = xprt->sc_fc_credits;
- *va++ = rdma_error;
- *va++ = cpu_to_be32(err);
- if (err == ERR_VERS) {
- *va++ = rpcrdma_version;
- *va++ = rpcrdma_version;
+ __be32 *p;
+
+ p = rdma_resp;
+
+ *p++ = *rdma_argp; /* XID */
+ *p++ = *(rdma_argp + 1); /* vers */
+ *p++ = rdma->sc_fc_credits;
+ *p++ = rdma_error;
+
+ *p++ = errcode;
+ if (errcode == err_vers) {
+ *p++ = rpcrdma_version;
+ *p = rpcrdma_version;
}
- return (int)((unsigned long)va - (unsigned long)startp);
+ return (unsigned long)p - (unsigned long)rdma_resp;
}
/**
diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index 776e1ed..f1b5bbf 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -688,7 +688,7 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
return ret;
out_err:
- svc_rdma_send_error(rdma_xprt, rmsgp, ret);
+ svc_rdma_send_error(rdma_xprt, (__be32 *)rmsgp, ret);
svc_rdma_put_context(ctxt, 0);
return 0;
diff --git a/net/sunrpc/xprtrdma/svc_rdma_sendto.c b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
index 5e75187..de54e97 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_sendto.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
@@ -487,60 +487,42 @@ int svc_rdma_sendto(struct svc_rqst *rqstp)
return -ENOTCONN;
}
-void svc_rdma_send_error(struct svcxprt_rdma *xprt, struct rpcrdma_msg *rmsgp,
+void svc_rdma_send_error(struct svcxprt_rdma *xprt, __be32 *rdma_argp,
int status)
{
- struct ib_send_wr err_wr;
- struct page *p;
struct svc_rdma_op_ctxt *ctxt;
- enum rpcrdma_errcode err;
- __be32 *va;
- int length;
+ unsigned int length;
+ struct page *page;
+ __be32 *err_msgp;
+ __be32 errcode;
int ret;
ret = svc_rdma_repost_recv(xprt, GFP_KERNEL);
if (ret)
return;
- p = alloc_page(GFP_KERNEL);
- if (!p)
+ page = alloc_page(GFP_KERNEL);
+ if (!page)
return;
- va = page_address(p);
+ err_msgp = page_address(page);
/* XDR encode an error reply */
- err = ERR_CHUNK;
+ errcode = err_chunk;
if (status == -EPROTONOSUPPORT)
- err = ERR_VERS;
- length = svc_rdma_xdr_encode_error(xprt, rmsgp, err, va);
+ errcode = err_vers;
+ length = svc_rdma_xdr_encode_error(xprt, rdma_argp, err_msgp, errcode);
+ /* Map transport header; no RPC message payload */
ctxt = svc_rdma_get_context(xprt);
- ctxt->direction = DMA_TO_DEVICE;
- ctxt->count = 1;
- ctxt->pages[0] = p;
-
- /* Prepare SGE for local address */
- ctxt->sge[0].lkey = xprt->sc_pd->local_dma_lkey;
- ctxt->sge[0].length = length;
- ctxt->sge[0].addr = ib_dma_map_page(xprt->sc_cm_id->device,
- p, 0, length, DMA_TO_DEVICE);
- if (ib_dma_mapping_error(xprt->sc_cm_id->device, ctxt->sge[0].addr)) {
- dprintk("svcrdma: Error mapping buffer for protocol error\n");
- svc_rdma_put_context(ctxt, 1);
+ ret = svc_rdma_map_reply_hdr(xprt, ctxt, err_msgp, length);
+ if (ret) {
+ dprintk("svcrdma: Error %d mapping send for protocol error\n",
+ ret);
return;
}
- svc_rdma_count_mappings(xprt, ctxt);
- /* Prepare SEND WR */
- memset(&err_wr, 0, sizeof(err_wr));
- ctxt->cqe.done = svc_rdma_wc_send;
- err_wr.wr_cqe = &ctxt->cqe;
- err_wr.sg_list = ctxt->sge;
- err_wr.num_sge = 1;
- err_wr.opcode = IB_WR_SEND;
- err_wr.send_flags = IB_SEND_SIGNALED;
-
- /* Post It */
- ret = svc_rdma_send(xprt, &err_wr);
+ svc_rdma_build_send_wr(ctxt, 1);
+ ret = svc_rdma_send(xprt, &ctxt->send_wr);
if (ret) {
dprintk("svcrdma: Error %d posting send for protocol error\n",
ret);
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 12/22] svcrdma: Reduce size of sge array in struct svc_rdma_op_ctxt
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (10 preceding siblings ...)
2017-01-07 17:16 ` [PATCH v1 11/22] svcrdma: Clean up RDMA_ERROR path Chuck Lever
@ 2017-01-07 17:16 ` Chuck Lever
2017-01-07 17:17 ` [PATCH v1 13/22] svcrdma: Remove old RDMA Write completion handlers Chuck Lever
` (12 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:16 UTC (permalink / raw)
To: linux-rdma, linux-nfs
The sge array in struct svc_rdma_op_ctxt is no longer used for
sending RDMA Write WRs. It need only accomodate the construction of
Send and Receive WRs. The maximum inline size is the largest payload
it needs to handle now.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 9 +++++++--
net/sunrpc/xprtrdma/svc_rdma.c | 6 +++---
2 files changed, 10 insertions(+), 5 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 65c2425..538a220 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -48,6 +48,12 @@
#include <rdma/rdma_cm.h>
#define SVCRDMA_DEBUG
+/* Default and maximum inline threshold sizes */
+enum {
+ RPCRDMA_DEF_REQ_SIZE = 4096,
+ RPCRDMA_MAX_REQ_SIZE = 65536
+};
+
/* RPC/RDMA parameters and stats */
extern unsigned int svcrdma_ord;
extern unsigned int svcrdma_max_requests;
@@ -86,7 +92,7 @@ struct svc_rdma_op_ctxt {
int count;
unsigned int mapped_sges;
struct ib_send_wr send_wr;
- struct ib_sge sge[RPCSVC_MAXPAGES];
+ struct ib_sge sge[1 + RPCRDMA_MAX_REQ_SIZE / PAGE_SIZE];
struct page *pages[RPCSVC_MAXPAGES];
};
@@ -188,7 +194,6 @@ struct svcxprt_rdma {
* page size of 4k, or 32k * 2 ops / 4k = 16 outstanding RDMA_READ. */
#define RPCRDMA_ORD (64/4)
#define RPCRDMA_MAX_REQUESTS 32
-#define RPCRDMA_MAX_REQ_SIZE 4096
/* Typical ULP usage of BC requests is NFSv4.1 backchannel. Our
* current NFSv4.1 implementation supports one backchannel slot.
diff --git a/net/sunrpc/xprtrdma/svc_rdma.c b/net/sunrpc/xprtrdma/svc_rdma.c
index 9124441..1f9f408 100644
--- a/net/sunrpc/xprtrdma/svc_rdma.c
+++ b/net/sunrpc/xprtrdma/svc_rdma.c
@@ -58,9 +58,9 @@
unsigned int svcrdma_max_bc_requests = RPCRDMA_MAX_BC_REQUESTS;
static unsigned int min_max_requests = 4;
static unsigned int max_max_requests = 16384;
-unsigned int svcrdma_max_req_size = RPCRDMA_MAX_REQ_SIZE;
-static unsigned int min_max_inline = 4096;
-static unsigned int max_max_inline = 65536;
+unsigned int svcrdma_max_req_size = RPCRDMA_DEF_REQ_SIZE;
+static unsigned int min_max_inline = RPCRDMA_DEF_REQ_SIZE;
+static unsigned int max_max_inline = RPCRDMA_MAX_REQ_SIZE;
atomic_t rdma_stat_recv;
atomic_t rdma_stat_read;
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 13/22] svcrdma: Remove old RDMA Write completion handlers
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (11 preceding siblings ...)
2017-01-07 17:16 ` [PATCH v1 12/22] svcrdma: Reduce size of sge array in struct svc_rdma_op_ctxt Chuck Lever
@ 2017-01-07 17:17 ` Chuck Lever
2017-01-07 17:17 ` [PATCH v1 14/22] svcrdma: Remove the req_map cache Chuck Lever
` (11 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:17 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Clean up. RDMA Write completions are now handled by the rdma_rw API.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 1 -
net/sunrpc/xprtrdma/svc_rdma_transport.c | 18 ------------------
2 files changed, 19 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 538a220..0e681b1 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -263,7 +263,6 @@ extern void svc_rdma_send_error(struct svcxprt_rdma *rdma, __be32 *rdma_argp,
/* svc_rdma_transport.c */
extern void svc_rdma_wc_send(struct ib_cq *, struct ib_wc *);
-extern void svc_rdma_wc_write(struct ib_cq *, struct ib_wc *);
extern void svc_rdma_wc_reg(struct ib_cq *, struct ib_wc *);
extern void svc_rdma_wc_read(struct ib_cq *, struct ib_wc *);
extern void svc_rdma_wc_inv(struct ib_cq *, struct ib_wc *);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index a6ddd8f..d930f70 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -473,24 +473,6 @@ void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc)
}
/**
- * svc_rdma_wc_write - Invoked by RDMA provider for each polled Write WC
- * @cq: completion queue
- * @wc: completed WR
- *
- */
-void svc_rdma_wc_write(struct ib_cq *cq, struct ib_wc *wc)
-{
- struct ib_cqe *cqe = wc->wr_cqe;
- struct svc_rdma_op_ctxt *ctxt;
-
- svc_rdma_send_wc_common_put(cq, wc, "write");
-
- ctxt = container_of(cqe, struct svc_rdma_op_ctxt, cqe);
- svc_rdma_unmap_dma(ctxt);
- svc_rdma_put_context(ctxt, 0);
-}
-
-/**
* svc_rdma_wc_reg - Invoked by RDMA provider for each polled FASTREG WC
* @cq: completion queue
* @wc: completed WR
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 14/22] svcrdma: Remove the req_map cache
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (12 preceding siblings ...)
2017-01-07 17:17 ` [PATCH v1 13/22] svcrdma: Remove old RDMA Write completion handlers Chuck Lever
@ 2017-01-07 17:17 ` Chuck Lever
2017-01-07 17:17 ` [PATCH v1 15/22] svcrdma: Clean up RPC-over-RDMA backchannel reply processing Chuck Lever
` (10 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:17 UTC (permalink / raw)
To: linux-rdma, linux-nfs
req_maps are no longer used by the send path and can thus be removed.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 32 -----------
net/sunrpc/xprtrdma/svc_rdma_transport.c | 84 ------------------------------
2 files changed, 1 insertion(+), 115 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 0e681b1..531980b 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -96,23 +96,6 @@ struct svc_rdma_op_ctxt {
struct page *pages[RPCSVC_MAXPAGES];
};
-/*
- * NFS_ requests are mapped on the client side by the chunk lists in
- * the RPCRDMA header. During the fetching of the RPC from the client
- * and the writing of the reply to the client, the memory in the
- * client and the memory in the server must be mapped as contiguous
- * vaddr/len for access by the hardware. These data strucures keep
- * these mappings.
- *
- * For an RDMA_WRITE, the 'sge' maps the RPC REPLY. For RDMA_READ, the
- * 'sge' in the svc_rdma_req_map maps the server side RPC reply and the
- * 'ch' field maps the read-list of the RPCRDMA header to the 'sge'
- * mapping of the reply.
- */
-struct svc_rdma_chunk_sge {
- int start; /* sge no for this chunk */
- int count; /* sge count for this chunk */
-};
struct svc_rdma_fastreg_mr {
struct ib_mr *mr;
struct scatterlist *sg;
@@ -121,15 +104,7 @@ struct svc_rdma_fastreg_mr {
enum dma_data_direction direction;
struct list_head frmr_list;
};
-struct svc_rdma_req_map {
- struct list_head free;
- unsigned long count;
- union {
- struct kvec sge[RPCSVC_MAXPAGES];
- struct svc_rdma_chunk_sge ch[RPCSVC_MAXPAGES];
- unsigned long lkey[RPCSVC_MAXPAGES];
- };
-};
+
#define RDMACTXT_F_LAST_CTXT 2
#define SVCRDMA_DEVCAP_FAST_REG 1 /* fast mr registration */
@@ -162,8 +137,6 @@ struct svcxprt_rdma {
int sc_ctxt_used;
spinlock_t sc_rw_ctxt_lock;
struct list_head sc_rw_ctxts;
- spinlock_t sc_map_lock;
- struct list_head sc_maps;
struct list_head sc_rq_dto_q;
spinlock_t sc_rq_dto_lock;
@@ -275,9 +248,6 @@ extern int svc_rdma_post_send(struct svcxprt_rdma *rdma,
extern struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *);
extern void svc_rdma_put_context(struct svc_rdma_op_ctxt *, int);
extern void svc_rdma_unmap_dma(struct svc_rdma_op_ctxt *ctxt);
-extern struct svc_rdma_req_map *svc_rdma_get_req_map(struct svcxprt_rdma *);
-extern void svc_rdma_put_req_map(struct svcxprt_rdma *,
- struct svc_rdma_req_map *);
extern struct svc_rdma_fastreg_mr *svc_rdma_get_frmr(struct svcxprt_rdma *);
extern void svc_rdma_put_frmr(struct svcxprt_rdma *,
struct svc_rdma_fastreg_mr *);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index d930f70..01119ba 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -271,85 +271,6 @@ static void svc_rdma_destroy_ctxts(struct svcxprt_rdma *xprt)
}
}
-static struct svc_rdma_req_map *alloc_req_map(gfp_t flags)
-{
- struct svc_rdma_req_map *map;
-
- map = kmalloc(sizeof(*map), flags);
- if (map)
- INIT_LIST_HEAD(&map->free);
- return map;
-}
-
-static bool svc_rdma_prealloc_maps(struct svcxprt_rdma *xprt)
-{
- unsigned int i;
-
- /* One for each receive buffer on this connection. */
- i = xprt->sc_max_requests;
-
- while (i--) {
- struct svc_rdma_req_map *map;
-
- map = alloc_req_map(GFP_KERNEL);
- if (!map) {
- dprintk("svcrdma: No memory for request map\n");
- return false;
- }
- list_add(&map->free, &xprt->sc_maps);
- }
- return true;
-}
-
-struct svc_rdma_req_map *svc_rdma_get_req_map(struct svcxprt_rdma *xprt)
-{
- struct svc_rdma_req_map *map = NULL;
-
- spin_lock(&xprt->sc_map_lock);
- if (list_empty(&xprt->sc_maps))
- goto out_empty;
-
- map = list_first_entry(&xprt->sc_maps,
- struct svc_rdma_req_map, free);
- list_del_init(&map->free);
- spin_unlock(&xprt->sc_map_lock);
-
-out:
- map->count = 0;
- return map;
-
-out_empty:
- spin_unlock(&xprt->sc_map_lock);
-
- /* Pre-allocation amount was incorrect */
- map = alloc_req_map(GFP_NOIO);
- if (map)
- goto out;
-
- WARN_ONCE(1, "svcrdma: empty request map list?\n");
- return NULL;
-}
-
-void svc_rdma_put_req_map(struct svcxprt_rdma *xprt,
- struct svc_rdma_req_map *map)
-{
- spin_lock(&xprt->sc_map_lock);
- list_add(&map->free, &xprt->sc_maps);
- spin_unlock(&xprt->sc_map_lock);
-}
-
-static void svc_rdma_destroy_maps(struct svcxprt_rdma *xprt)
-{
- while (!list_empty(&xprt->sc_maps)) {
- struct svc_rdma_req_map *map;
-
- map = list_first_entry(&xprt->sc_maps,
- struct svc_rdma_req_map, free);
- list_del(&map->free);
- kfree(map);
- }
-}
-
/* QP event handler */
static void qp_event_handler(struct ib_event *event, void *context)
{
@@ -543,7 +464,6 @@ static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
INIT_LIST_HEAD(&cma_xprt->sc_frmr_q);
INIT_LIST_HEAD(&cma_xprt->sc_ctxts);
INIT_LIST_HEAD(&cma_xprt->sc_rw_ctxts);
- INIT_LIST_HEAD(&cma_xprt->sc_maps);
init_waitqueue_head(&cma_xprt->sc_send_wait);
spin_lock_init(&cma_xprt->sc_lock);
@@ -551,7 +471,6 @@ static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
spin_lock_init(&cma_xprt->sc_frmr_q_lock);
spin_lock_init(&cma_xprt->sc_ctxt_lock);
spin_lock_init(&cma_xprt->sc_rw_ctxt_lock);
- spin_lock_init(&cma_xprt->sc_map_lock);
if (listener)
set_bit(XPT_LISTENER, &cma_xprt->sc_xprt.xpt_flags);
@@ -1047,8 +966,6 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
if (!svc_rdma_prealloc_ctxts(newxprt))
goto errout;
- if (!svc_rdma_prealloc_maps(newxprt))
- goto errout;
/*
* Limit ORD based on client limit, local device limit, and
@@ -1282,7 +1199,6 @@ static void __svc_rdma_free(struct work_struct *work)
rdma_dealloc_frmr_q(rdma);
svc_rdma_destroy_rw_ctxts(rdma);
svc_rdma_destroy_ctxts(rdma);
- svc_rdma_destroy_maps(rdma);
/* Destroy the QP if present (not a listener) */
if (rdma->sc_qp && !IS_ERR(rdma->sc_qp))
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 15/22] svcrdma: Clean up RPC-over-RDMA backchannel reply processing
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (13 preceding siblings ...)
2017-01-07 17:17 ` [PATCH v1 14/22] svcrdma: Remove the req_map cache Chuck Lever
@ 2017-01-07 17:17 ` Chuck Lever
2017-01-07 17:36 ` Trond Myklebust
2017-01-07 17:17 ` [PATCH v1 16/22] svcrdma: Use generic RDMA R/W API in RPC Call path Chuck Lever
` (9 subsequent siblings)
24 siblings, 1 reply; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:17 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Replace C structure-based XDR decoding with pointer arithmetic.
Pointer arithmetic is considered more portable.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 2 +-
net/sunrpc/xprtrdma/svc_rdma_backchannel.c | 17 +++++++++++++----
net/sunrpc/xprtrdma/svc_rdma_recvfrom.c | 27 +++++++++++++++------------
3 files changed, 29 insertions(+), 17 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 531980b..9866e94 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -184,7 +184,7 @@ static inline void svc_rdma_count_mappings(struct svcxprt_rdma *rdma,
/* svc_rdma_backchannel.c */
extern int svc_rdma_handle_bc_reply(struct rpc_xprt *xprt,
- struct rpcrdma_msg *rmsgp,
+ __be32 *rdma_resp,
struct xdr_buf *rcvbuf);
/* svc_rdma_marshal.c */
diff --git a/net/sunrpc/xprtrdma/svc_rdma_backchannel.c b/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
index 7d49545..6b3bbb6 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_backchannel.c
@@ -11,7 +11,16 @@
#undef SVCRDMA_BACKCHANNEL_DEBUG
-int svc_rdma_handle_bc_reply(struct rpc_xprt *xprt, struct rpcrdma_msg *rmsgp,
+/**
+ * svc_rdma_handle_bc_reply - Process incoming backchannel reply
+ * @xprt: controlling backchannel transport
+ * @rdma_resp: pointer to incoming transport header
+ * @rcvbuf: XDR buffer into which to decode the reply
+ *
+ * Returns zero and fills @rcvbuf if successful. Otherwise a
+ * negative errno is returned.
+ */
+int svc_rdma_handle_bc_reply(struct rpc_xprt *xprt, __be32 *rdma_resp,
struct xdr_buf *rcvbuf)
{
struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
@@ -26,13 +35,13 @@ int svc_rdma_handle_bc_reply(struct rpc_xprt *xprt, struct rpcrdma_msg *rmsgp,
p = (__be32 *)src->iov_base;
len = src->iov_len;
- xid = rmsgp->rm_xid;
+ xid = *rdma_resp;
#ifdef SVCRDMA_BACKCHANNEL_DEBUG
pr_info("%s: xid=%08x, length=%zu\n",
__func__, be32_to_cpu(xid), len);
pr_info("%s: RPC/RDMA: %*ph\n",
- __func__, (int)RPCRDMA_HDRLEN_MIN, rmsgp);
+ __func__, (int)RPCRDMA_HDRLEN_MIN, rdma_resp);
pr_info("%s: RPC: %*ph\n",
__func__, (int)len, p);
#endif
@@ -52,7 +61,7 @@ int svc_rdma_handle_bc_reply(struct rpc_xprt *xprt, struct rpcrdma_msg *rmsgp,
goto out_unlock;
memcpy(dst->iov_base, p, len);
- credits = be32_to_cpu(rmsgp->rm_credit);
+ credits = be32_to_cpup(rdma_resp + 2);
if (credits == 0)
credits = 1; /* don't deadlock */
else if (credits > r_xprt->rx_buf.rb_bc_max_requests)
diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index f1b5bbf..c9ea2e9 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -565,28 +565,30 @@ static void rdma_read_complete(struct svc_rqst *rqstp,
* the RPC/RDMA header small and fixed in size, so it is
* straightforward to check the RPC header's direction field.
*/
-static bool
-svc_rdma_is_backchannel_reply(struct svc_xprt *xprt, struct rpcrdma_msg *rmsgp)
+static bool svc_rdma_is_backchannel_reply(struct svc_xprt *xprt,
+ __be32 *rdma_resp)
{
- __be32 *p = (__be32 *)rmsgp;
+ __be32 *p;
if (!xprt->xpt_bc_xprt)
return false;
- if (rmsgp->rm_type != rdma_msg)
+ p = rdma_resp + 3;
+ if (*p++ != rdma_msg)
return false;
- if (rmsgp->rm_body.rm_chunks[0] != xdr_zero)
+
+ if (*p++ != xdr_zero)
return false;
- if (rmsgp->rm_body.rm_chunks[1] != xdr_zero)
+ if (*p++ != xdr_zero)
return false;
- if (rmsgp->rm_body.rm_chunks[2] != xdr_zero)
+ if (*p++ != xdr_zero)
return false;
- /* sanity */
- if (p[7] != rmsgp->rm_xid)
+ /* XID sanity */
+ if (*p++ != *rdma_resp)
return false;
/* call direction */
- if (p[8] == cpu_to_be32(RPC_CALL))
+ if (*p == cpu_to_be32(RPC_CALL))
return false;
return true;
@@ -652,8 +654,9 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
goto out_drop;
rqstp->rq_xprt_hlen = ret;
- if (svc_rdma_is_backchannel_reply(xprt, rmsgp)) {
- ret = svc_rdma_handle_bc_reply(xprt->xpt_bc_xprt, rmsgp,
+ if (svc_rdma_is_backchannel_reply(xprt, (__be32 *)rmsgp)) {
+ ret = svc_rdma_handle_bc_reply(xprt->xpt_bc_xprt,
+ (__be32 *)rmsgp,
&rqstp->rq_arg);
svc_rdma_put_context(ctxt, 0);
if (ret)
^ permalink raw reply related [flat|nested] 30+ messages in thread* Re: [PATCH v1 15/22] svcrdma: Clean up RPC-over-RDMA backchannel reply processing
2017-01-07 17:17 ` [PATCH v1 15/22] svcrdma: Clean up RPC-over-RDMA backchannel reply processing Chuck Lever
@ 2017-01-07 17:36 ` Trond Myklebust
0 siblings, 0 replies; 30+ messages in thread
From: Trond Myklebust @ 2017-01-07 17:36 UTC (permalink / raw)
To: Chuck Lever; +Cc: Linux RDMA Mailing List, Linux NFS Mailing List
DQo+IE9uIEphbiA3LCAyMDE3LCBhdCAxMjoxNywgQ2h1Y2sgTGV2ZXIgPGNodWNrLmxldmVyQG9y
YWNsZS5jb20+IHdyb3RlOg0KPiANCj4gUmVwbGFjZSBDIHN0cnVjdHVyZS1iYXNlZCBYRFIgZGVj
b2Rpbmcgd2l0aCBwb2ludGVyIGFyaXRobWV0aWMuDQo+IFBvaW50ZXIgYXJpdGhtZXRpYyBpcyBj
b25zaWRlcmVkIG1vcmUgcG9ydGFibGUuDQo+IA0KPiBTaWduZWQtb2ZmLWJ5OiBDaHVjayBMZXZl
ciA8Y2h1Y2subGV2ZXJAb3JhY2xlLmNvbT4NCj4gLS0tDQo+IGluY2x1ZGUvbGludXgvc3VucnBj
L3N2Y19yZG1hLmggICAgICAgICAgICB8ICAgIDIgKy0NCj4gbmV0L3N1bnJwYy94cHJ0cmRtYS9z
dmNfcmRtYV9iYWNrY2hhbm5lbC5jIHwgICAxNyArKysrKysrKysrKysrLS0tLQ0KPiBuZXQvc3Vu
cnBjL3hwcnRyZG1hL3N2Y19yZG1hX3JlY3Zmcm9tLmMgICAgfCAgIDI3ICsrKysrKysrKysrKysr
Ky0tLS0tLS0tLS0tLQ0KPiAzIGZpbGVzIGNoYW5nZWQsIDI5IGluc2VydGlvbnMoKyksIDE3IGRl
bGV0aW9ucygtKQ0KPiANCj4gZGlmZiAtLWdpdCBhL2luY2x1ZGUvbGludXgvc3VucnBjL3N2Y19y
ZG1hLmggYi9pbmNsdWRlL2xpbnV4L3N1bnJwYy9zdmNfcmRtYS5oDQo+IGluZGV4IDUzMTk4MGIu
Ljk4NjZlOTQgMTAwNjQ0DQo+IC0tLSBhL2luY2x1ZGUvbGludXgvc3VucnBjL3N2Y19yZG1hLmgN
Cj4gKysrIGIvaW5jbHVkZS9saW51eC9zdW5ycGMvc3ZjX3JkbWEuaA0KPiBAQCAtMTg0LDcgKzE4
NCw3IEBAIHN0YXRpYyBpbmxpbmUgdm9pZCBzdmNfcmRtYV9jb3VudF9tYXBwaW5ncyhzdHJ1Y3Qg
c3ZjeHBydF9yZG1hICpyZG1hLA0KPiANCj4gLyogc3ZjX3JkbWFfYmFja2NoYW5uZWwuYyAqLw0K
PiBleHRlcm4gaW50IHN2Y19yZG1hX2hhbmRsZV9iY19yZXBseShzdHJ1Y3QgcnBjX3hwcnQgKnhw
cnQsDQo+IC0JCQkJICAgIHN0cnVjdCBycGNyZG1hX21zZyAqcm1zZ3AsDQo+ICsJCQkJICAgIF9f
YmUzMiAqcmRtYV9yZXNwLA0KPiAJCQkJICAgIHN0cnVjdCB4ZHJfYnVmICpyY3ZidWYpOw0KPiAN
Cj4gLyogc3ZjX3JkbWFfbWFyc2hhbC5jICovDQo+IGRpZmYgLS1naXQgYS9uZXQvc3VucnBjL3hw
cnRyZG1hL3N2Y19yZG1hX2JhY2tjaGFubmVsLmMgYi9uZXQvc3VucnBjL3hwcnRyZG1hL3N2Y19y
ZG1hX2JhY2tjaGFubmVsLmMNCj4gaW5kZXggN2Q0OTU0NS4uNmIzYmJiNiAxMDA2NDQNCj4gLS0t
IGEvbmV0L3N1bnJwYy94cHJ0cmRtYS9zdmNfcmRtYV9iYWNrY2hhbm5lbC5jDQo+ICsrKyBiL25l
dC9zdW5ycGMveHBydHJkbWEvc3ZjX3JkbWFfYmFja2NoYW5uZWwuYw0KPiBAQCAtMTEsNyArMTEs
MTYgQEANCj4gDQo+ICN1bmRlZiBTVkNSRE1BX0JBQ0tDSEFOTkVMX0RFQlVHDQo+IA0KPiAtaW50
IHN2Y19yZG1hX2hhbmRsZV9iY19yZXBseShzdHJ1Y3QgcnBjX3hwcnQgKnhwcnQsIHN0cnVjdCBy
cGNyZG1hX21zZyAqcm1zZ3AsDQo+ICsvKioNCj4gKyAqIHN2Y19yZG1hX2hhbmRsZV9iY19yZXBs
eSAtIFByb2Nlc3MgaW5jb21pbmcgYmFja2NoYW5uZWwgcmVwbHkNCj4gKyAqIEB4cHJ0OiBjb250
cm9sbGluZyBiYWNrY2hhbm5lbCB0cmFuc3BvcnQNCj4gKyAqIEByZG1hX3Jlc3A6IHBvaW50ZXIg
dG8gaW5jb21pbmcgdHJhbnNwb3J0IGhlYWRlcg0KPiArICogQHJjdmJ1ZjogWERSIGJ1ZmZlciBp
bnRvIHdoaWNoIHRvIGRlY29kZSB0aGUgcmVwbHkNCj4gKyAqDQo+ICsgKiBSZXR1cm5zIHplcm8g
YW5kIGZpbGxzIEByY3ZidWYgaWYgc3VjY2Vzc2Z1bC4gT3RoZXJ3aXNlIGENCj4gKyAqIG5lZ2F0
aXZlIGVycm5vIGlzIHJldHVybmVkLg0KPiArICovDQo+ICtpbnQgc3ZjX3JkbWFfaGFuZGxlX2Jj
X3JlcGx5KHN0cnVjdCBycGNfeHBydCAqeHBydCwgX19iZTMyICpyZG1hX3Jlc3AsDQo+IAkJCSAg
ICAgc3RydWN0IHhkcl9idWYgKnJjdmJ1ZikNCj4gew0KPiAJc3RydWN0IHJwY3JkbWFfeHBydCAq
cl94cHJ0ID0gcnBjeF90b19yZG1heCh4cHJ0KTsNCj4gQEAgLTI2LDEzICszNSwxMyBAQCBpbnQg
c3ZjX3JkbWFfaGFuZGxlX2JjX3JlcGx5KHN0cnVjdCBycGNfeHBydCAqeHBydCwgc3RydWN0IHJw
Y3JkbWFfbXNnICpybXNncCwNCj4gDQo+IAlwID0gKF9fYmUzMiAqKXNyYy0+aW92X2Jhc2U7DQo+
IAlsZW4gPSBzcmMtPmlvdl9sZW47DQo+IC0JeGlkID0gcm1zZ3AtPnJtX3hpZDsNCj4gKwl4aWQg
PSAqcmRtYV9yZXNwOw0KPiANCj4gI2lmZGVmIFNWQ1JETUFfQkFDS0NIQU5ORUxfREVCVUcNCj4g
CXByX2luZm8oIiVzOiB4aWQ9JTA4eCwgbGVuZ3RoPSV6dVxuIiwNCj4gCQlfX2Z1bmNfXywgYmUz
Ml90b19jcHUoeGlkKSwgbGVuKTsNCj4gCXByX2luZm8oIiVzOiBSUEMvUkRNQTogJSpwaFxuIiwN
Cj4gLQkJX19mdW5jX18sIChpbnQpUlBDUkRNQV9IRFJMRU5fTUlOLCBybXNncCk7DQo+ICsJCV9f
ZnVuY19fLCAoaW50KVJQQ1JETUFfSERSTEVOX01JTiwgcmRtYV9yZXNwKTsNCj4gCXByX2luZm8o
IiVzOiAgICAgIFJQQzogJSpwaFxuIiwNCj4gCQlfX2Z1bmNfXywgKGludClsZW4sIHApOw0KPiAj
ZW5kaWYNCj4gQEAgLTUyLDcgKzYxLDcgQEAgaW50IHN2Y19yZG1hX2hhbmRsZV9iY19yZXBseShz
dHJ1Y3QgcnBjX3hwcnQgKnhwcnQsIHN0cnVjdCBycGNyZG1hX21zZyAqcm1zZ3AsDQo+IAkJZ290
byBvdXRfdW5sb2NrOw0KPiAJbWVtY3B5KGRzdC0+aW92X2Jhc2UsIHAsIGxlbik7DQo+IA0KPiAt
CWNyZWRpdHMgPSBiZTMyX3RvX2NwdShybXNncC0+cm1fY3JlZGl0KTsNCj4gKwljcmVkaXRzID0g
YmUzMl90b19jcHVwKHJkbWFfcmVzcCArIDIpOw0KPiAJaWYgKGNyZWRpdHMgPT0gMCkNCj4gCQlj
cmVkaXRzID0gMTsJLyogZG9uJ3QgZGVhZGxvY2sgKi8NCj4gCWVsc2UgaWYgKGNyZWRpdHMgPiBy
X3hwcnQtPnJ4X2J1Zi5yYl9iY19tYXhfcmVxdWVzdHMpDQo+IGRpZmYgLS1naXQgYS9uZXQvc3Vu
cnBjL3hwcnRyZG1hL3N2Y19yZG1hX3JlY3Zmcm9tLmMgYi9uZXQvc3VucnBjL3hwcnRyZG1hL3N2
Y19yZG1hX3JlY3Zmcm9tLmMNCj4gaW5kZXggZjFiNWJiZi4uYzllYTJlOSAxMDA2NDQNCj4gLS0t
IGEvbmV0L3N1bnJwYy94cHJ0cmRtYS9zdmNfcmRtYV9yZWN2ZnJvbS5jDQo+ICsrKyBiL25ldC9z
dW5ycGMveHBydHJkbWEvc3ZjX3JkbWFfcmVjdmZyb20uYw0KPiBAQCAtNTY1LDI4ICs1NjUsMzAg
QEAgc3RhdGljIHZvaWQgcmRtYV9yZWFkX2NvbXBsZXRlKHN0cnVjdCBzdmNfcnFzdCAqcnFzdHAs
DQo+ICAqIHRoZSBSUEMvUkRNQSBoZWFkZXIgc21hbGwgYW5kIGZpeGVkIGluIHNpemUsIHNvIGl0
IGlzDQo+ICAqIHN0cmFpZ2h0Zm9yd2FyZCB0byBjaGVjayB0aGUgUlBDIGhlYWRlcidzIGRpcmVj
dGlvbiBmaWVsZC4NCj4gICovDQo+IC1zdGF0aWMgYm9vbA0KPiAtc3ZjX3JkbWFfaXNfYmFja2No
YW5uZWxfcmVwbHkoc3RydWN0IHN2Y194cHJ0ICp4cHJ0LCBzdHJ1Y3QgcnBjcmRtYV9tc2cgKnJt
c2dwKQ0KPiArc3RhdGljIGJvb2wgc3ZjX3JkbWFfaXNfYmFja2NoYW5uZWxfcmVwbHkoc3RydWN0
IHN2Y194cHJ0ICp4cHJ0LA0KPiArCQkJCQkgIF9fYmUzMiAqcmRtYV9yZXNwKQ0KPiB7DQo+IC0J
X19iZTMyICpwID0gKF9fYmUzMiAqKXJtc2dwOw0KPiArCV9fYmUzMiAqcDsNCj4gDQo+IAlpZiAo
IXhwcnQtPnhwdF9iY194cHJ0KQ0KPiAJCXJldHVybiBmYWxzZTsNCj4gDQo+IC0JaWYgKHJtc2dw
LT5ybV90eXBlICE9IHJkbWFfbXNnKQ0KPiArCXAgPSByZG1hX3Jlc3AgKyAzOw0KPiArCWlmICgq
cCsrICE9IHJkbWFfbXNnKQ0KPiAJCXJldHVybiBmYWxzZTsNCj4gLQlpZiAocm1zZ3AtPnJtX2Jv
ZHkucm1fY2h1bmtzWzBdICE9IHhkcl96ZXJvKQ0KPiArDQo+ICsJaWYgKCpwKysgIT0geGRyX3pl
cm8pDQo+IAkJcmV0dXJuIGZhbHNlOw0KPiAtCWlmIChybXNncC0+cm1fYm9keS5ybV9jaHVua3Nb
MV0gIT0geGRyX3plcm8pDQo+ICsJaWYgKCpwKysgIT0geGRyX3plcm8pDQo+IAkJcmV0dXJuIGZh
bHNlOw0KPiAtCWlmIChybXNncC0+cm1fYm9keS5ybV9jaHVua3NbMl0gIT0geGRyX3plcm8pDQo+
ICsJaWYgKCpwKysgIT0geGRyX3plcm8pDQo+IAkJcmV0dXJuIGZhbHNlOw0KPiANCj4gLQkvKiBz
YW5pdHkgKi8NCj4gLQlpZiAocFs3XSAhPSBybXNncC0+cm1feGlkKQ0KPiArCS8qIFhJRCBzYW5p
dHkgKi8NCj4gKwlpZiAoKnArKyAhPSAqcmRtYV9yZXNwKQ0KPiAJCXJldHVybiBmYWxzZTsNCj4g
CS8qIGNhbGwgZGlyZWN0aW9uICovDQo+IC0JaWYgKHBbOF0gPT0gY3B1X3RvX2JlMzIoUlBDX0NB
TEwpKQ0KPiArCWlmICgqcCA9PSBjcHVfdG9fYmUzMihSUENfQ0FMTCkpDQo+IAkJcmV0dXJuIGZh
bHNlOw0KPiANCj4gCXJldHVybiB0cnVlOw0KPiBAQCAtNjUyLDggKzY1NCw5IEBAIGludCBzdmNf
cmRtYV9yZWN2ZnJvbShzdHJ1Y3Qgc3ZjX3Jxc3QgKnJxc3RwKQ0KPiAJCWdvdG8gb3V0X2Ryb3A7
DQo+IAlycXN0cC0+cnFfeHBydF9obGVuID0gcmV0Ow0KPiANCj4gLQlpZiAoc3ZjX3JkbWFfaXNf
YmFja2NoYW5uZWxfcmVwbHkoeHBydCwgcm1zZ3ApKSB7DQo+IC0JCXJldCA9IHN2Y19yZG1hX2hh
bmRsZV9iY19yZXBseSh4cHJ0LT54cHRfYmNfeHBydCwgcm1zZ3AsDQo+ICsJaWYgKHN2Y19yZG1h
X2lzX2JhY2tjaGFubmVsX3JlcGx5KHhwcnQsIChfX2JlMzIgKilybXNncCkpIHsNCj4gKwkJcmV0
ID0gc3ZjX3JkbWFfaGFuZGxlX2JjX3JlcGx5KHhwcnQtPnhwdF9iY194cHJ0LA0KPiArCQkJCQkg
ICAgICAgKF9fYmUzMiAqKXJtc2dwLA0KDQpTaG91bGRu4oCZdCB0aGF0IGJlICZybXNncC0+cm1f
eGlkIGluIG9yZGVyIHRvIGJlIHN0cmljdGx5IGVxdWl2YWxlbnQ/DQoNCj4gCQkJCQkgICAgICAg
JnJxc3RwLT5ycV9hcmcpOw0KPiAJCXN2Y19yZG1hX3B1dF9jb250ZXh0KGN0eHQsIDApOw0KPiAJ
CWlmIChyZXQpDQo+IA0KPiAtLQ0KPiBUbyB1bnN1YnNjcmliZSBmcm9tIHRoaXMgbGlzdDogc2Vu
ZCB0aGUgbGluZSAidW5zdWJzY3JpYmUgbGludXgtbmZzIiBpbg0KPiB0aGUgYm9keSBvZiBhIG1l
c3NhZ2UgdG8gbWFqb3Jkb21vQHZnZXIua2VybmVsLm9yZw0KPiBNb3JlIG1ham9yZG9tbyBpbmZv
IGF0ICBodHRwOi8vdmdlci5rZXJuZWwub3JnL21ham9yZG9tby1pbmZvLmh0bWwNCj4gDQoNCg==
^ permalink raw reply [flat|nested] 30+ messages in thread
* [PATCH v1 16/22] svcrdma: Use generic RDMA R/W API in RPC Call path
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (14 preceding siblings ...)
2017-01-07 17:17 ` [PATCH v1 15/22] svcrdma: Clean up RPC-over-RDMA backchannel reply processing Chuck Lever
@ 2017-01-07 17:17 ` Chuck Lever
2017-01-07 17:17 ` [PATCH v1 17/22] svcrdma: Remove unused Read completion handlers Chuck Lever
` (8 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:17 UTC (permalink / raw)
To: linux-rdma, linux-nfs
The current svcrdma recvfrom code path has a lot of detail about
registration mode and the type of port (iWARP, IB, etc).
Instead, use the RDMA core's generic R/W API. This shares code with
other RDMA-enabled ULPs that manages the gory details of buffer
registration and the posting of RDMA Read Work Requests.
Some design notes:
o Since the Read list marshaling code is being
replaced, I took the opportunity to replace some of the C
structure-based XDR encoding code with more portable code that
uses pointer arithmetic. C compilers don't provide a guarantee of
field location in structures.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 13 -
net/sunrpc/xprtrdma/svc_rdma_recvfrom.c | 477 ++++--------------------------
net/sunrpc/xprtrdma/svc_rdma_transport.c | 30 --
3 files changed, 65 insertions(+), 455 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 9866e94..95a3720 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -82,8 +82,6 @@ struct svc_rdma_op_ctxt {
int hdr_count;
struct xdr_buf arg;
struct ib_cqe cqe;
- struct ib_cqe reg_cqe;
- struct ib_cqe inv_cqe;
u32 byte_len;
u32 position;
struct svcxprt_rdma *xprt;
@@ -116,7 +114,6 @@ struct svcxprt_rdma {
struct list_head sc_accept_q; /* Conn. waiting accept */
int sc_ord; /* RDMA read limit */
int sc_max_sge;
- int sc_max_sge_rd; /* max sge for read target */
bool sc_snd_w_inv; /* OK to use Send With Invalidate */
atomic_t sc_sq_avail; /* SQEs ready to be consumed */
@@ -143,10 +140,6 @@ struct svcxprt_rdma {
struct ib_qp *sc_qp;
struct ib_cq *sc_rq_cq;
struct ib_cq *sc_sq_cq;
- int (*sc_reader)(struct svcxprt_rdma *,
- struct svc_rqst *,
- struct svc_rdma_op_ctxt *,
- int *, u32 *, u32, u32, u64, bool);
u32 sc_dev_caps; /* distilled device caps */
unsigned int sc_frmr_pg_list_len;
struct list_head sc_frmr_q;
@@ -205,12 +198,6 @@ extern void svc_rdma_xdr_encode_reply_header(struct svcxprt_rdma *rdma,
/* svc_rdma_recvfrom.c */
extern int svc_rdma_recvfrom(struct svc_rqst *);
-extern int rdma_read_chunk_lcl(struct svcxprt_rdma *, struct svc_rqst *,
- struct svc_rdma_op_ctxt *, int *, u32 *,
- u32, u32, u64, bool);
-extern int rdma_read_chunk_frmr(struct svcxprt_rdma *, struct svc_rqst *,
- struct svc_rdma_op_ctxt *, int *, u32 *,
- u32, u32, u64, bool);
/* svc_rdma_rw.c */
extern void svc_rdma_destroy_rw_ctxts(struct svcxprt_rdma *rdma);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index c9ea2e9..f99f0d7 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -1,4 +1,5 @@
/*
+ * Copyright (c) 2016 Oracle. All rights reserved.
* Copyright (c) 2014 Open Grid Computing, Inc. All rights reserved.
* Copyright (c) 2005-2006 Network Appliance, Inc. All rights reserved.
*
@@ -40,6 +41,39 @@
* Author: Tom Tucker <tom@opengridcomputing.com>
*/
+/* Operation
+ *
+ * The main entry point is svc_rdma_recvfrom. This is called by the
+ * RPC server when the transport indicates there is incoming data to
+ * be read. This occurs when each RDMA Receive completes.
+ *
+ * The passed-in svc_rqst contains a struct xdr_buf which will hold
+ * the incoming XDR-encoded RPC Call message, and a set of pages that
+ * can be used as Read sink buffers.
+ *
+ * Short messages are moved directly into the xdr_buf and the RPC Call
+ * is ready to be processed by the Upper Layer.
+ *
+ * When an incoming message has Read chunks, recvfrom must post RDMA
+ * Reads to pull the chunk payloads to the server. recvfrom returns,
+ * and the svc_rqst is recycled, before the Read I/O is complete. A
+ * second completion and svc_xprt_enqueue are done after the chunk
+ * payloads have become available on the server.
+ *
+ * Page Management
+ *
+ * Pages under I/O must be removed from the svc_rqst before recvfrom
+ * returns.
+ *
+ * The logic here depends on Send Queue and completion ordering. Pages
+ * under I/O are moved from the provided svc_rqst to an svc_rdma_op_ctxt.
+ * The final RDMA Read completion points to this op_ctxt, which is
+ * enqueued on the svc_xprt during that completion.
+ *
+ * A second call to svc_rdma_recvfrom is done to perform final assembly
+ * of the RPC Call message, using the pages in the op_ctxt.
+ */
+
#include <linux/sunrpc/debug.h>
#include <linux/sunrpc/rpc_rdma.h>
#include <linux/spinlock.h>
@@ -115,410 +149,6 @@ static void rdma_build_arg_xdr(struct svc_rqst *rqstp,
rqstp->rq_arg.tail[0].iov_len = 0;
}
-/* Issue an RDMA_READ using the local lkey to map the data sink */
-int rdma_read_chunk_lcl(struct svcxprt_rdma *xprt,
- struct svc_rqst *rqstp,
- struct svc_rdma_op_ctxt *head,
- int *page_no,
- u32 *page_offset,
- u32 rs_handle,
- u32 rs_length,
- u64 rs_offset,
- bool last)
-{
- struct ib_rdma_wr read_wr;
- int pages_needed = PAGE_ALIGN(*page_offset + rs_length) >> PAGE_SHIFT;
- struct svc_rdma_op_ctxt *ctxt = svc_rdma_get_context(xprt);
- int ret, read, pno;
- u32 pg_off = *page_offset;
- u32 pg_no = *page_no;
-
- ctxt->direction = DMA_FROM_DEVICE;
- ctxt->read_hdr = head;
- pages_needed = min_t(int, pages_needed, xprt->sc_max_sge_rd);
- read = min_t(int, (pages_needed << PAGE_SHIFT) - *page_offset,
- rs_length);
-
- for (pno = 0; pno < pages_needed; pno++) {
- int len = min_t(int, rs_length, PAGE_SIZE - pg_off);
-
- head->arg.pages[pg_no] = rqstp->rq_arg.pages[pg_no];
- head->arg.page_len += len;
-
- head->arg.len += len;
- if (!pg_off)
- head->count++;
- rqstp->rq_respages = &rqstp->rq_arg.pages[pg_no+1];
- rqstp->rq_next_page = rqstp->rq_respages + 1;
- ctxt->sge[pno].addr =
- ib_dma_map_page(xprt->sc_cm_id->device,
- head->arg.pages[pg_no], pg_off,
- PAGE_SIZE - pg_off,
- DMA_FROM_DEVICE);
- ret = ib_dma_mapping_error(xprt->sc_cm_id->device,
- ctxt->sge[pno].addr);
- if (ret)
- goto err;
- svc_rdma_count_mappings(xprt, ctxt);
-
- ctxt->sge[pno].lkey = xprt->sc_pd->local_dma_lkey;
- ctxt->sge[pno].length = len;
- ctxt->count++;
-
- /* adjust offset and wrap to next page if needed */
- pg_off += len;
- if (pg_off == PAGE_SIZE) {
- pg_off = 0;
- pg_no++;
- }
- rs_length -= len;
- }
-
- if (last && rs_length == 0)
- set_bit(RDMACTXT_F_LAST_CTXT, &ctxt->flags);
- else
- clear_bit(RDMACTXT_F_LAST_CTXT, &ctxt->flags);
-
- memset(&read_wr, 0, sizeof(read_wr));
- ctxt->cqe.done = svc_rdma_wc_read;
- read_wr.wr.wr_cqe = &ctxt->cqe;
- read_wr.wr.opcode = IB_WR_RDMA_READ;
- read_wr.wr.send_flags = IB_SEND_SIGNALED;
- read_wr.rkey = rs_handle;
- read_wr.remote_addr = rs_offset;
- read_wr.wr.sg_list = ctxt->sge;
- read_wr.wr.num_sge = pages_needed;
-
- ret = svc_rdma_send(xprt, &read_wr.wr);
- if (ret) {
- pr_err("svcrdma: Error %d posting RDMA_READ\n", ret);
- set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
- goto err;
- }
-
- /* return current location in page array */
- *page_no = pg_no;
- *page_offset = pg_off;
- ret = read;
- atomic_inc(&rdma_stat_read);
- return ret;
- err:
- svc_rdma_unmap_dma(ctxt);
- svc_rdma_put_context(ctxt, 0);
- return ret;
-}
-
-/* Issue an RDMA_READ using an FRMR to map the data sink */
-int rdma_read_chunk_frmr(struct svcxprt_rdma *xprt,
- struct svc_rqst *rqstp,
- struct svc_rdma_op_ctxt *head,
- int *page_no,
- u32 *page_offset,
- u32 rs_handle,
- u32 rs_length,
- u64 rs_offset,
- bool last)
-{
- struct ib_rdma_wr read_wr;
- struct ib_send_wr inv_wr;
- struct ib_reg_wr reg_wr;
- u8 key;
- int nents = PAGE_ALIGN(*page_offset + rs_length) >> PAGE_SHIFT;
- struct svc_rdma_op_ctxt *ctxt = svc_rdma_get_context(xprt);
- struct svc_rdma_fastreg_mr *frmr = svc_rdma_get_frmr(xprt);
- int ret, read, pno, dma_nents, n;
- u32 pg_off = *page_offset;
- u32 pg_no = *page_no;
-
- if (IS_ERR(frmr))
- return -ENOMEM;
-
- ctxt->direction = DMA_FROM_DEVICE;
- ctxt->frmr = frmr;
- nents = min_t(unsigned int, nents, xprt->sc_frmr_pg_list_len);
- read = min_t(int, (nents << PAGE_SHIFT) - *page_offset, rs_length);
-
- frmr->direction = DMA_FROM_DEVICE;
- frmr->access_flags = (IB_ACCESS_LOCAL_WRITE|IB_ACCESS_REMOTE_WRITE);
- frmr->sg_nents = nents;
-
- for (pno = 0; pno < nents; pno++) {
- int len = min_t(int, rs_length, PAGE_SIZE - pg_off);
-
- head->arg.pages[pg_no] = rqstp->rq_arg.pages[pg_no];
- head->arg.page_len += len;
- head->arg.len += len;
- if (!pg_off)
- head->count++;
-
- sg_set_page(&frmr->sg[pno], rqstp->rq_arg.pages[pg_no],
- len, pg_off);
-
- rqstp->rq_respages = &rqstp->rq_arg.pages[pg_no+1];
- rqstp->rq_next_page = rqstp->rq_respages + 1;
-
- /* adjust offset and wrap to next page if needed */
- pg_off += len;
- if (pg_off == PAGE_SIZE) {
- pg_off = 0;
- pg_no++;
- }
- rs_length -= len;
- }
-
- if (last && rs_length == 0)
- set_bit(RDMACTXT_F_LAST_CTXT, &ctxt->flags);
- else
- clear_bit(RDMACTXT_F_LAST_CTXT, &ctxt->flags);
-
- dma_nents = ib_dma_map_sg(xprt->sc_cm_id->device,
- frmr->sg, frmr->sg_nents,
- frmr->direction);
- if (!dma_nents) {
- pr_err("svcrdma: failed to dma map sg %p\n",
- frmr->sg);
- return -ENOMEM;
- }
-
- n = ib_map_mr_sg(frmr->mr, frmr->sg, frmr->sg_nents, NULL, PAGE_SIZE);
- if (unlikely(n != frmr->sg_nents)) {
- pr_err("svcrdma: failed to map mr %p (%d/%d elements)\n",
- frmr->mr, n, frmr->sg_nents);
- return n < 0 ? n : -EINVAL;
- }
-
- /* Bump the key */
- key = (u8)(frmr->mr->lkey & 0x000000FF);
- ib_update_fast_reg_key(frmr->mr, ++key);
-
- ctxt->sge[0].addr = frmr->mr->iova;
- ctxt->sge[0].lkey = frmr->mr->lkey;
- ctxt->sge[0].length = frmr->mr->length;
- ctxt->count = 1;
- ctxt->read_hdr = head;
-
- /* Prepare REG WR */
- ctxt->reg_cqe.done = svc_rdma_wc_reg;
- reg_wr.wr.wr_cqe = &ctxt->reg_cqe;
- reg_wr.wr.opcode = IB_WR_REG_MR;
- reg_wr.wr.send_flags = IB_SEND_SIGNALED;
- reg_wr.wr.num_sge = 0;
- reg_wr.mr = frmr->mr;
- reg_wr.key = frmr->mr->lkey;
- reg_wr.access = frmr->access_flags;
- reg_wr.wr.next = &read_wr.wr;
-
- /* Prepare RDMA_READ */
- memset(&read_wr, 0, sizeof(read_wr));
- ctxt->cqe.done = svc_rdma_wc_read;
- read_wr.wr.wr_cqe = &ctxt->cqe;
- read_wr.wr.send_flags = IB_SEND_SIGNALED;
- read_wr.rkey = rs_handle;
- read_wr.remote_addr = rs_offset;
- read_wr.wr.sg_list = ctxt->sge;
- read_wr.wr.num_sge = 1;
- if (xprt->sc_dev_caps & SVCRDMA_DEVCAP_READ_W_INV) {
- read_wr.wr.opcode = IB_WR_RDMA_READ_WITH_INV;
- read_wr.wr.ex.invalidate_rkey = ctxt->frmr->mr->lkey;
- } else {
- read_wr.wr.opcode = IB_WR_RDMA_READ;
- read_wr.wr.next = &inv_wr;
- /* Prepare invalidate */
- memset(&inv_wr, 0, sizeof(inv_wr));
- ctxt->inv_cqe.done = svc_rdma_wc_inv;
- inv_wr.wr_cqe = &ctxt->inv_cqe;
- inv_wr.opcode = IB_WR_LOCAL_INV;
- inv_wr.send_flags = IB_SEND_SIGNALED | IB_SEND_FENCE;
- inv_wr.ex.invalidate_rkey = frmr->mr->lkey;
- }
-
- /* Post the chain */
- ret = svc_rdma_send(xprt, ®_wr.wr);
- if (ret) {
- pr_err("svcrdma: Error %d posting RDMA_READ\n", ret);
- set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
- goto err;
- }
-
- /* return current location in page array */
- *page_no = pg_no;
- *page_offset = pg_off;
- ret = read;
- atomic_inc(&rdma_stat_read);
- return ret;
- err:
- ib_dma_unmap_sg(xprt->sc_cm_id->device,
- frmr->sg, frmr->sg_nents, frmr->direction);
- svc_rdma_put_context(ctxt, 0);
- svc_rdma_put_frmr(xprt, frmr);
- return ret;
-}
-
-static unsigned int
-rdma_rcl_chunk_count(struct rpcrdma_read_chunk *ch)
-{
- unsigned int count;
-
- for (count = 0; ch->rc_discrim != xdr_zero; ch++)
- count++;
- return count;
-}
-
-/* If there was additional inline content, append it to the end of arg.pages.
- * Tail copy has to be done after the reader function has determined how many
- * pages are needed for RDMA READ.
- */
-static int
-rdma_copy_tail(struct svc_rqst *rqstp, struct svc_rdma_op_ctxt *head,
- u32 position, u32 byte_count, u32 page_offset, int page_no)
-{
- char *srcp, *destp;
-
- srcp = head->arg.head[0].iov_base + position;
- byte_count = head->arg.head[0].iov_len - position;
- if (byte_count > PAGE_SIZE) {
- dprintk("svcrdma: large tail unsupported\n");
- return 0;
- }
-
- /* Fit as much of the tail on the current page as possible */
- if (page_offset != PAGE_SIZE) {
- destp = page_address(rqstp->rq_arg.pages[page_no]);
- destp += page_offset;
- while (byte_count--) {
- *destp++ = *srcp++;
- page_offset++;
- if (page_offset == PAGE_SIZE && byte_count)
- goto more;
- }
- goto done;
- }
-
-more:
- /* Fit the rest on the next page */
- page_no++;
- destp = page_address(rqstp->rq_arg.pages[page_no]);
- while (byte_count--)
- *destp++ = *srcp++;
-
- rqstp->rq_respages = &rqstp->rq_arg.pages[page_no+1];
- rqstp->rq_next_page = rqstp->rq_respages + 1;
-
-done:
- byte_count = head->arg.head[0].iov_len - position;
- head->arg.page_len += byte_count;
- head->arg.len += byte_count;
- head->arg.buflen += byte_count;
- return 1;
-}
-
-/* Returns the address of the first read chunk or <nul> if no read chunk
- * is present
- */
-static struct rpcrdma_read_chunk *
-svc_rdma_get_read_chunk(struct rpcrdma_msg *rmsgp)
-{
- struct rpcrdma_read_chunk *ch =
- (struct rpcrdma_read_chunk *)&rmsgp->rm_body.rm_chunks[0];
-
- if (ch->rc_discrim == xdr_zero)
- return NULL;
- return ch;
-}
-
-static int rdma_read_chunks(struct svcxprt_rdma *xprt,
- struct rpcrdma_msg *rmsgp,
- struct svc_rqst *rqstp,
- struct svc_rdma_op_ctxt *head)
-{
- int page_no, ret;
- struct rpcrdma_read_chunk *ch;
- u32 handle, page_offset, byte_count;
- u32 position;
- u64 rs_offset;
- bool last;
-
- /* If no read list is present, return 0 */
- ch = svc_rdma_get_read_chunk(rmsgp);
- if (!ch)
- return 0;
-
- if (rdma_rcl_chunk_count(ch) > RPCSVC_MAXPAGES)
- return -EINVAL;
-
- /* The request is completed when the RDMA_READs complete. The
- * head context keeps all the pages that comprise the
- * request.
- */
- head->arg.head[0] = rqstp->rq_arg.head[0];
- head->arg.tail[0] = rqstp->rq_arg.tail[0];
- head->hdr_count = head->count;
- head->arg.page_base = 0;
- head->arg.page_len = 0;
- head->arg.len = rqstp->rq_arg.len;
- head->arg.buflen = rqstp->rq_arg.buflen;
-
- /* RDMA_NOMSG: RDMA READ data should land just after RDMA RECV data */
- position = be32_to_cpu(ch->rc_position);
- if (position == 0) {
- head->arg.pages = &head->pages[0];
- page_offset = head->byte_len;
- } else {
- head->arg.pages = &head->pages[head->count];
- page_offset = 0;
- }
-
- ret = 0;
- page_no = 0;
- for (; ch->rc_discrim != xdr_zero; ch++) {
- if (be32_to_cpu(ch->rc_position) != position)
- goto err;
-
- handle = be32_to_cpu(ch->rc_target.rs_handle),
- byte_count = be32_to_cpu(ch->rc_target.rs_length);
- xdr_decode_hyper((__be32 *)&ch->rc_target.rs_offset,
- &rs_offset);
-
- while (byte_count > 0) {
- last = (ch + 1)->rc_discrim == xdr_zero;
- ret = xprt->sc_reader(xprt, rqstp, head,
- &page_no, &page_offset,
- handle, byte_count,
- rs_offset, last);
- if (ret < 0)
- goto err;
- byte_count -= ret;
- rs_offset += ret;
- head->arg.buflen += ret;
- }
- }
-
- /* Read list may need XDR round-up (see RFC 5666, s. 3.7) */
- if (page_offset & 3) {
- u32 pad = 4 - (page_offset & 3);
-
- head->arg.tail[0].iov_len += pad;
- head->arg.len += pad;
- head->arg.buflen += pad;
- page_offset += pad;
- }
-
- ret = 1;
- if (position && position < head->arg.head[0].iov_len)
- ret = rdma_copy_tail(rqstp, head, position,
- byte_count, page_offset, page_no);
- head->arg.head[0].iov_len = position;
- head->position = position;
-
- err:
- /* Detach arg pages. svc_recv will replenish them */
- for (page_no = 0;
- &rqstp->rq_pages[page_no] < rqstp->rq_respages; page_no++)
- rqstp->rq_pages[page_no] = NULL;
-
- return ret;
-}
-
static void rdma_read_complete(struct svc_rqst *rqstp,
struct svc_rdma_op_ctxt *head)
{
@@ -594,10 +224,18 @@ static bool svc_rdma_is_backchannel_reply(struct svc_xprt *xprt,
return true;
}
-/*
+/**
+ * svc_rdma_recvfrom - xprt switch API: Receive an RPC call
+ * @rqstp: request structure into which to receive the call
+ *
* Set up the rqstp thread context to point to the RQ buffer. If
- * necessary, pull additional data from the client with an RDMA_READ
+ * necessary, pull additional data from the client with an RDMA Read
* request.
+ *
+ * Returns the number of bytes in the RPC call on success; returns
+ * zero if the incoming RPC call is somehow invalid; or a negative
+ * errno is returned if a problem with the underlying xprt occurred
+ * while receiving the RPC.
*/
int svc_rdma_recvfrom(struct svc_rqst *rqstp)
{
@@ -605,11 +243,12 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
struct svcxprt_rdma *rdma_xprt =
container_of(xprt, struct svcxprt_rdma, sc_xprt);
struct svc_rdma_op_ctxt *ctxt = NULL;
- struct rpcrdma_msg *rmsgp;
- int ret = 0;
+ __be32 *rdma_argp, *ch;
+ int ret;
dprintk("svcrdma: rqstp=%p\n", rqstp);
+ ret = 0;
spin_lock(&rdma_xprt->sc_rq_dto_lock);
if (!list_empty(&rdma_xprt->sc_read_complete_q)) {
ctxt = list_first_entry(&rdma_xprt->sc_read_complete_q,
@@ -638,6 +277,7 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
goto defer;
goto out;
}
+
dprintk("svcrdma: processing ctxt=%p on xprt=%p, rqstp=%p\n",
ctxt, rdma_xprt, rqstp);
atomic_inc(&rdma_stat_recv);
@@ -646,7 +286,7 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
rdma_build_arg_xdr(rqstp, ctxt, ctxt->byte_len);
/* Decode the RDMA header. */
- rmsgp = (struct rpcrdma_msg *)rqstp->rq_arg.head[0].iov_base;
+ rdma_argp = (__be32 *)rqstp->rq_arg.head[0].iov_base;
ret = svc_rdma_xdr_decode_req(&rqstp->rq_arg);
if (ret < 0)
goto out_err;
@@ -654,9 +294,8 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
goto out_drop;
rqstp->rq_xprt_hlen = ret;
- if (svc_rdma_is_backchannel_reply(xprt, (__be32 *)rmsgp)) {
- ret = svc_rdma_handle_bc_reply(xprt->xpt_bc_xprt,
- (__be32 *)rmsgp,
+ if (svc_rdma_is_backchannel_reply(xprt, rdma_argp)) {
+ ret = svc_rdma_handle_bc_reply(xprt->xpt_bc_xprt, rdma_argp,
&rqstp->rq_arg);
svc_rdma_put_context(ctxt, 0);
if (ret)
@@ -664,13 +303,17 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
return ret;
}
- /* Read read-list data. */
- ret = rdma_read_chunks(rdma_xprt, rmsgp, rqstp, ctxt);
+ /* check for Read list */
+ ch = rdma_argp + rpcrdma_fixed_maxsz;
+ if (*ch == xdr_zero)
+ goto complete;
+
+ ret = svc_rdma_recv_read_list(rdma_xprt, ch, ctxt, rqstp);
if (ret > 0) {
- /* read-list posted, defer until data received from client. */
+ /* Read list posted, defer until data received from client. */
goto defer;
} else if (ret < 0) {
- /* Post of read-list failed, free context. */
+ /* Post of Read list failed, free context. */
svc_rdma_put_context(ctxt, 1);
return 0;
}
@@ -691,7 +334,7 @@ int svc_rdma_recvfrom(struct svc_rqst *rqstp)
return ret;
out_err:
- svc_rdma_send_error(rdma_xprt, (__be32 *)rmsgp, ret);
+ svc_rdma_send_error(rdma_xprt, rdma_argp, ret);
svc_rdma_put_context(ctxt, 0);
return 0;
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index 01119ba..bb0281f 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -851,28 +851,13 @@ void svc_rdma_put_frmr(struct svcxprt_rdma *rdma,
}
}
-static unsigned int svc_rdma_read_sqes_per_credit(struct svcxprt_rdma *newxprt)
-{
- struct ib_device_attr *attrs = &newxprt->sc_cm_id->device->attrs;
-
- if (!(attrs->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
- return DIV_ROUND_UP(RPCSVC_MAXPAGES, newxprt->sc_max_sge_rd);
-
- /* FRWR: reg, read, inv */
- return DIV_ROUND_UP(RPCSVC_MAXPAGES,
- attrs->max_fast_reg_page_list_len) * 3;
-}
-
+/* This is the number of RDMA Send operations that can be
+ * outstanding at one time. Forechannel and backchannel
+ * Sends are included.
+ */
static unsigned int svc_rdma_sq_depth(struct svcxprt_rdma *newxprt)
{
- unsigned int sqes_per_credit;
-
- sqes_per_credit = svc_rdma_read_sqes_per_credit(newxprt);
-
- /* RDMA Sends per credit */
- sqes_per_credit += 1;
-
- return sqes_per_credit * newxprt->sc_rq_depth;
+ return newxprt->sc_rq_depth;
}
/* Estimate ctxs per credit assuming a full Read chunk payload
@@ -951,8 +936,6 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
* capabilities of this particular device */
newxprt->sc_max_sge = min((size_t)dev->attrs.max_sge,
(size_t)RPCSVC_MAXPAGES);
- newxprt->sc_max_sge_rd = min_t(size_t, dev->attrs.max_sge_rd,
- RPCSVC_MAXPAGES);
newxprt->sc_max_req_size = svcrdma_max_req_size;
newxprt->sc_max_requests = min_t(u32, dev->attrs.max_qp_wr,
svcrdma_max_requests);
@@ -1043,12 +1026,10 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
* NB: iWARP requires remote write access for the data sink
* of an RDMA_READ. IB does not.
*/
- newxprt->sc_reader = rdma_read_chunk_lcl;
if (dev->attrs.device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS) {
newxprt->sc_frmr_pg_list_len =
dev->attrs.max_fast_reg_page_list_len;
newxprt->sc_dev_caps |= SVCRDMA_DEVCAP_FAST_REG;
- newxprt->sc_reader = rdma_read_chunk_frmr;
} else
newxprt->sc_snd_w_inv = false;
@@ -1101,7 +1082,6 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
sap = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.dst_addr;
dprintk(" remote address : %pIS:%u\n", sap, rpc_get_port(sap));
dprintk(" max_sge : %d\n", newxprt->sc_max_sge);
- dprintk(" max_sge_rd : %d\n", newxprt->sc_max_sge_rd);
dprintk(" sq_depth : %d\n", newxprt->sc_sq_depth);
dprintk(" max_requests : %d\n", newxprt->sc_max_requests);
dprintk(" ord : %d\n", newxprt->sc_ord);
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 17/22] svcrdma: Remove unused Read completion handlers
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (15 preceding siblings ...)
2017-01-07 17:17 ` [PATCH v1 16/22] svcrdma: Use generic RDMA R/W API in RPC Call path Chuck Lever
@ 2017-01-07 17:17 ` Chuck Lever
2017-01-07 17:17 ` [PATCH v1 18/22] svcrdma: Remove frmr cache Chuck Lever
` (7 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:17 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Clean up:
The generic RDMA R/W API conversion of svc_rdma_recvfrom replaced
the Register, Read, and Invalidate completion handlers. Remove the
old ones, which are no longer used.
As an additional clean up, fold the wc_common helper back into the
remaining Send completion handler in svc_rdma_transport.c.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 4 -
net/sunrpc/xprtrdma/svc_rdma_transport.c | 98 ++++--------------------------
2 files changed, 15 insertions(+), 87 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 95a3720..6fae169 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -77,18 +77,16 @@ enum {
*/
struct svc_rdma_op_ctxt {
struct list_head list;
- struct svc_rdma_op_ctxt *read_hdr;
struct svc_rdma_fastreg_mr *frmr;
- int hdr_count;
struct xdr_buf arg;
struct ib_cqe cqe;
u32 byte_len;
u32 position;
struct svcxprt_rdma *xprt;
- unsigned long flags;
enum dma_data_direction direction;
int count;
unsigned int mapped_sges;
+ int hdr_count;
struct ib_send_wr send_wr;
struct ib_sge sge[1 + RPCRDMA_MAX_REQ_SIZE / PAGE_SIZE];
struct page *pages[RPCSVC_MAXPAGES];
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index bb0281f..e77d18b 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -345,36 +345,6 @@ static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
svc_xprt_put(&xprt->sc_xprt);
}
-static void svc_rdma_send_wc_common(struct svcxprt_rdma *xprt,
- struct ib_wc *wc,
- const char *opname)
-{
- if (wc->status != IB_WC_SUCCESS)
- goto err;
-
-out:
- atomic_inc(&xprt->sc_sq_avail);
- wake_up(&xprt->sc_send_wait);
- return;
-
-err:
- set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
- if (wc->status != IB_WC_WR_FLUSH_ERR)
- pr_err("svcrdma: %s: %s (%u/0x%x)\n",
- opname, ib_wc_status_msg(wc->status),
- wc->status, wc->vendor_err);
- goto out;
-}
-
-static void svc_rdma_send_wc_common_put(struct ib_cq *cq, struct ib_wc *wc,
- const char *opname)
-{
- struct svcxprt_rdma *xprt = cq->cq_context;
-
- svc_rdma_send_wc_common(xprt, wc, opname);
- svc_xprt_put(&xprt->sc_xprt);
-}
-
/**
* svc_rdma_wc_send - Invoked by RDMA provider for each polled Send WC
* @cq: completion queue
@@ -383,71 +353,31 @@ static void svc_rdma_send_wc_common_put(struct ib_cq *cq, struct ib_wc *wc,
*/
void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc)
{
- struct ib_cqe *cqe = wc->wr_cqe;
- struct svc_rdma_op_ctxt *ctxt;
-
- svc_rdma_send_wc_common_put(cq, wc, "send");
-
- ctxt = container_of(cqe, struct svc_rdma_op_ctxt, cqe);
- svc_rdma_unmap_dma(ctxt);
- svc_rdma_put_context(ctxt, 1);
-}
-
-/**
- * svc_rdma_wc_reg - Invoked by RDMA provider for each polled FASTREG WC
- * @cq: completion queue
- * @wc: completed WR
- *
- */
-void svc_rdma_wc_reg(struct ib_cq *cq, struct ib_wc *wc)
-{
- svc_rdma_send_wc_common_put(cq, wc, "fastreg");
-}
-
-/**
- * svc_rdma_wc_read - Invoked by RDMA provider for each polled Read WC
- * @cq: completion queue
- * @wc: completed WR
- *
- */
-void svc_rdma_wc_read(struct ib_cq *cq, struct ib_wc *wc)
-{
struct svcxprt_rdma *xprt = cq->cq_context;
struct ib_cqe *cqe = wc->wr_cqe;
struct svc_rdma_op_ctxt *ctxt;
- svc_rdma_send_wc_common(xprt, wc, "read");
+ atomic_inc(&xprt->sc_sq_avail);
+ wake_up(&xprt->sc_send_wait);
ctxt = container_of(cqe, struct svc_rdma_op_ctxt, cqe);
svc_rdma_unmap_dma(ctxt);
- svc_rdma_put_frmr(xprt, ctxt->frmr);
-
- if (test_bit(RDMACTXT_F_LAST_CTXT, &ctxt->flags)) {
- struct svc_rdma_op_ctxt *read_hdr;
-
- read_hdr = ctxt->read_hdr;
- spin_lock(&xprt->sc_rq_dto_lock);
- list_add_tail(&read_hdr->list,
- &xprt->sc_read_complete_q);
- spin_unlock(&xprt->sc_rq_dto_lock);
+ svc_rdma_put_context(ctxt, 1);
- set_bit(XPT_DATA, &xprt->sc_xprt.xpt_flags);
- svc_xprt_enqueue(&xprt->sc_xprt);
- }
+ if (wc->status != IB_WC_SUCCESS)
+ goto flush;
- svc_rdma_put_context(ctxt, 0);
+out:
svc_xprt_put(&xprt->sc_xprt);
-}
+ return;
-/**
- * svc_rdma_wc_inv - Invoked by RDMA provider for each polled LOCAL_INV WC
- * @cq: completion queue
- * @wc: completed WR
- *
- */
-void svc_rdma_wc_inv(struct ib_cq *cq, struct ib_wc *wc)
-{
- svc_rdma_send_wc_common_put(cq, wc, "localInv");
+flush:
+ set_bit(XPT_CLOSE, &xprt->sc_xprt.xpt_flags);
+ if (wc->status != IB_WC_WR_FLUSH_ERR)
+ pr_err("svcrdma: Send: %s (%u/0x%x)\n",
+ ib_wc_status_msg(wc->status),
+ wc->status, wc->vendor_err);
+ goto out;
}
static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 18/22] svcrdma: Remove frmr cache
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (16 preceding siblings ...)
2017-01-07 17:17 ` [PATCH v1 17/22] svcrdma: Remove unused Read completion handlers Chuck Lever
@ 2017-01-07 17:17 ` Chuck Lever
2017-01-07 17:17 ` [PATCH v1 19/22] svcrdma: Clean up after converting svc_rdma_recvfrom to rdma_rw API Chuck Lever
` (6 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:17 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Clean up: Now that the svc_rdma_recvfrom path uses the rdma_rw API,
the details of Read sink buffer registration are dealt with by the
kernel's RDMA core. This cache is no longer used, and can be
removed.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 18 ------
net/sunrpc/xprtrdma/svc_rdma_transport.c | 86 ------------------------------
2 files changed, 104 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 6fae169..2818185 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -77,7 +77,6 @@ enum {
*/
struct svc_rdma_op_ctxt {
struct list_head list;
- struct svc_rdma_fastreg_mr *frmr;
struct xdr_buf arg;
struct ib_cqe cqe;
u32 byte_len;
@@ -92,17 +91,6 @@ struct svc_rdma_op_ctxt {
struct page *pages[RPCSVC_MAXPAGES];
};
-struct svc_rdma_fastreg_mr {
- struct ib_mr *mr;
- struct scatterlist *sg;
- int sg_nents;
- unsigned long access_flags;
- enum dma_data_direction direction;
- struct list_head frmr_list;
-};
-
-#define RDMACTXT_F_LAST_CTXT 2
-
#define SVCRDMA_DEVCAP_FAST_REG 1 /* fast mr registration */
#define SVCRDMA_DEVCAP_READ_W_INV 2 /* read w/ invalidate */
@@ -139,9 +127,6 @@ struct svcxprt_rdma {
struct ib_cq *sc_rq_cq;
struct ib_cq *sc_sq_cq;
u32 sc_dev_caps; /* distilled device caps */
- unsigned int sc_frmr_pg_list_len;
- struct list_head sc_frmr_q;
- spinlock_t sc_frmr_q_lock;
spinlock_t sc_lock; /* transport lock */
@@ -233,9 +218,6 @@ extern int svc_rdma_post_send(struct svcxprt_rdma *rdma,
extern struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *);
extern void svc_rdma_put_context(struct svc_rdma_op_ctxt *, int);
extern void svc_rdma_unmap_dma(struct svc_rdma_op_ctxt *ctxt);
-extern struct svc_rdma_fastreg_mr *svc_rdma_get_frmr(struct svcxprt_rdma *);
-extern void svc_rdma_put_frmr(struct svcxprt_rdma *,
- struct svc_rdma_fastreg_mr *);
extern void svc_sq_reap(struct svcxprt_rdma *);
extern void svc_rq_reap(struct svcxprt_rdma *);
extern void svc_rdma_prep_reply_hdr(struct svc_rqst *);
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index e77d18b..6c52553 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -201,7 +201,6 @@ struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *xprt)
out:
ctxt->count = 0;
ctxt->mapped_sges = 0;
- ctxt->frmr = NULL;
return ctxt;
out_empty:
@@ -391,14 +390,12 @@ static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
INIT_LIST_HEAD(&cma_xprt->sc_accept_q);
INIT_LIST_HEAD(&cma_xprt->sc_rq_dto_q);
INIT_LIST_HEAD(&cma_xprt->sc_read_complete_q);
- INIT_LIST_HEAD(&cma_xprt->sc_frmr_q);
INIT_LIST_HEAD(&cma_xprt->sc_ctxts);
INIT_LIST_HEAD(&cma_xprt->sc_rw_ctxts);
init_waitqueue_head(&cma_xprt->sc_send_wait);
spin_lock_init(&cma_xprt->sc_lock);
spin_lock_init(&cma_xprt->sc_rq_dto_lock);
- spin_lock_init(&cma_xprt->sc_frmr_q_lock);
spin_lock_init(&cma_xprt->sc_ctxt_lock);
spin_lock_init(&cma_xprt->sc_rw_ctxt_lock);
@@ -701,86 +698,6 @@ static struct svc_xprt *svc_rdma_create(struct svc_serv *serv,
return ERR_PTR(ret);
}
-static struct svc_rdma_fastreg_mr *rdma_alloc_frmr(struct svcxprt_rdma *xprt)
-{
- struct ib_mr *mr;
- struct scatterlist *sg;
- struct svc_rdma_fastreg_mr *frmr;
- u32 num_sg;
-
- frmr = kmalloc(sizeof(*frmr), GFP_KERNEL);
- if (!frmr)
- goto err;
-
- num_sg = min_t(u32, RPCSVC_MAXPAGES, xprt->sc_frmr_pg_list_len);
- mr = ib_alloc_mr(xprt->sc_pd, IB_MR_TYPE_MEM_REG, num_sg);
- if (IS_ERR(mr))
- goto err_free_frmr;
-
- sg = kcalloc(RPCSVC_MAXPAGES, sizeof(*sg), GFP_KERNEL);
- if (!sg)
- goto err_free_mr;
-
- sg_init_table(sg, RPCSVC_MAXPAGES);
-
- frmr->mr = mr;
- frmr->sg = sg;
- INIT_LIST_HEAD(&frmr->frmr_list);
- return frmr;
-
- err_free_mr:
- ib_dereg_mr(mr);
- err_free_frmr:
- kfree(frmr);
- err:
- return ERR_PTR(-ENOMEM);
-}
-
-static void rdma_dealloc_frmr_q(struct svcxprt_rdma *xprt)
-{
- struct svc_rdma_fastreg_mr *frmr;
-
- while (!list_empty(&xprt->sc_frmr_q)) {
- frmr = list_entry(xprt->sc_frmr_q.next,
- struct svc_rdma_fastreg_mr, frmr_list);
- list_del_init(&frmr->frmr_list);
- kfree(frmr->sg);
- ib_dereg_mr(frmr->mr);
- kfree(frmr);
- }
-}
-
-struct svc_rdma_fastreg_mr *svc_rdma_get_frmr(struct svcxprt_rdma *rdma)
-{
- struct svc_rdma_fastreg_mr *frmr = NULL;
-
- spin_lock(&rdma->sc_frmr_q_lock);
- if (!list_empty(&rdma->sc_frmr_q)) {
- frmr = list_entry(rdma->sc_frmr_q.next,
- struct svc_rdma_fastreg_mr, frmr_list);
- list_del_init(&frmr->frmr_list);
- frmr->sg_nents = 0;
- }
- spin_unlock(&rdma->sc_frmr_q_lock);
- if (frmr)
- return frmr;
-
- return rdma_alloc_frmr(rdma);
-}
-
-void svc_rdma_put_frmr(struct svcxprt_rdma *rdma,
- struct svc_rdma_fastreg_mr *frmr)
-{
- if (frmr) {
- ib_dma_unmap_sg(rdma->sc_cm_id->device,
- frmr->sg, frmr->sg_nents, frmr->direction);
- spin_lock(&rdma->sc_frmr_q_lock);
- WARN_ON_ONCE(!list_empty(&frmr->frmr_list));
- list_add(&frmr->frmr_list, &rdma->sc_frmr_q);
- spin_unlock(&rdma->sc_frmr_q_lock);
- }
-}
-
/* This is the number of RDMA Send operations that can be
* outstanding at one time. Forechannel and backchannel
* Sends are included.
@@ -957,8 +874,6 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
* of an RDMA_READ. IB does not.
*/
if (dev->attrs.device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS) {
- newxprt->sc_frmr_pg_list_len =
- dev->attrs.max_fast_reg_page_list_len;
newxprt->sc_dev_caps |= SVCRDMA_DEVCAP_FAST_REG;
} else
newxprt->sc_snd_w_inv = false;
@@ -1106,7 +1021,6 @@ static void __svc_rdma_free(struct work_struct *work)
xprt->xpt_bc_xprt = NULL;
}
- rdma_dealloc_frmr_q(rdma);
svc_rdma_destroy_rw_ctxts(rdma);
svc_rdma_destroy_ctxts(rdma);
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 19/22] svcrdma: Clean up after converting svc_rdma_recvfrom to rdma_rw API
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (17 preceding siblings ...)
2017-01-07 17:17 ` [PATCH v1 18/22] svcrdma: Remove frmr cache Chuck Lever
@ 2017-01-07 17:17 ` Chuck Lever
2017-01-07 17:17 ` [PATCH v1 20/22] svcrdma: Clean-up in svc_rdma_post_recv Chuck Lever
` (5 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:17 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Clean up: Registration mode details are now handled by the rdma_rw
API, and thus can be removed from svcrdma.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 4 ---
net/sunrpc/xprtrdma/svc_rdma_transport.c | 39 +++---------------------------
2 files changed, 4 insertions(+), 39 deletions(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 2818185..30c3ae9 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -91,9 +91,6 @@ struct svc_rdma_op_ctxt {
struct page *pages[RPCSVC_MAXPAGES];
};
-#define SVCRDMA_DEVCAP_FAST_REG 1 /* fast mr registration */
-#define SVCRDMA_DEVCAP_READ_W_INV 2 /* read w/ invalidate */
-
struct svcxprt_rdma {
struct svc_xprt sc_xprt; /* SVC transport structure */
struct rdma_cm_id *sc_cm_id; /* RDMA connection id */
@@ -126,7 +123,6 @@ struct svcxprt_rdma {
struct ib_qp *sc_qp;
struct ib_cq *sc_rq_cq;
struct ib_cq *sc_sq_cq;
- u32 sc_dev_caps; /* distilled device caps */
spinlock_t sc_lock; /* transport lock */
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index 6c52553..b357bb2 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -827,7 +827,7 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
memset(&qp_attr, 0, sizeof qp_attr);
qp_attr.event_handler = qp_event_handler;
qp_attr.qp_context = &newxprt->sc_xprt;
- qp_attr.port_num = newxprt->sc_cm_id->port_num;
+ qp_attr.port_num = newxprt->sc_port_num;
qp_attr.cap.max_rdma_ctxs = svc_rdma_max_rdma_ctxs(newxprt);
qp_attr.cap.max_send_wr = newxprt->sc_sq_depth;
qp_attr.cap.max_recv_wr = newxprt->sc_rq_depth;
@@ -851,43 +851,12 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
}
newxprt->sc_qp = newxprt->sc_cm_id->qp;
- /*
- * Use the most secure set of MR resources based on the
- * transport type and available memory management features in
- * the device. Here's the table implemented below:
- *
- * Fast Global DMA Remote WR
- * Reg LKEY MR Access
- * Sup'd Sup'd Needed Needed
- *
- * IWARP N N Y Y
- * N Y Y Y
- * Y N Y N
- * Y Y N -
- *
- * IB N N Y N
- * N Y N -
- * Y N Y N
- * Y Y N -
- *
- * NB: iWARP requires remote write access for the data sink
- * of an RDMA_READ. IB does not.
- */
- if (dev->attrs.device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS) {
- newxprt->sc_dev_caps |= SVCRDMA_DEVCAP_FAST_REG;
- } else
+ if (!(dev->attrs.device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
newxprt->sc_snd_w_inv = false;
-
- /*
- * Determine if a DMA MR is required and if so, what privs are required
- */
- if (!rdma_protocol_iwarp(dev, newxprt->sc_cm_id->port_num) &&
- !rdma_ib_or_roce(dev, newxprt->sc_cm_id->port_num))
+ if (!rdma_protocol_iwarp(dev, newxprt->sc_port_num) &&
+ !rdma_ib_or_roce(dev, newxprt->sc_port_num))
goto errout;
- if (rdma_protocol_iwarp(dev, newxprt->sc_cm_id->port_num))
- newxprt->sc_dev_caps |= SVCRDMA_DEVCAP_READ_W_INV;
-
/* Post receive buffers */
for (i = 0; i < newxprt->sc_max_requests; i++) {
ret = svc_rdma_post_recv(newxprt, GFP_KERNEL);
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 20/22] svcrdma: Clean-up in svc_rdma_post_recv
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (18 preceding siblings ...)
2017-01-07 17:17 ` [PATCH v1 19/22] svcrdma: Clean up after converting svc_rdma_recvfrom to rdma_rw API Chuck Lever
@ 2017-01-07 17:17 ` Chuck Lever
2017-01-07 17:18 ` [PATCH v1 21/22] svcrdma: Clean-up svc_rdma_unmap_dma Chuck Lever
` (4 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:17 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Micro-optimization: Use cached copies of lkey and device instead
of chasing pointers in the RDMA layer.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
net/sunrpc/xprtrdma/svc_rdma_transport.c | 20 ++++++++++++++------
1 file changed, 14 insertions(+), 6 deletions(-)
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index b357bb2..9b60740 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -405,6 +405,14 @@ static struct svcxprt_rdma *rdma_create_xprt(struct svc_serv *serv,
return cma_xprt;
}
+/**
+ * svc_rdma_post_recv - Post a receive WR
+ * @xprt: RPC-over-RDMA transport control block
+ * @flags: GFP flags
+ *
+ * Returns zero if receive WR was posted successfully, or
+ * a negative errno if nothing could be posted.
+ */
int svc_rdma_post_recv(struct svcxprt_rdma *xprt, gfp_t flags)
{
struct ib_recv_wr recv_wr, *bad_recv_wr;
@@ -416,9 +424,10 @@ int svc_rdma_post_recv(struct svcxprt_rdma *xprt, gfp_t flags)
int ret;
ctxt = svc_rdma_get_context(xprt);
- buflen = 0;
ctxt->direction = DMA_FROM_DEVICE;
ctxt->cqe.done = svc_rdma_wc_receive;
+
+ buflen = 0;
for (sge_no = 0; buflen < xprt->sc_max_req_size; sge_no++) {
if (sge_no >= xprt->sc_max_sge) {
pr_err("svcrdma: Too many sges (%d)\n", sge_no);
@@ -428,15 +437,14 @@ int svc_rdma_post_recv(struct svcxprt_rdma *xprt, gfp_t flags)
if (!page)
goto err_put_ctxt;
ctxt->pages[sge_no] = page;
- pa = ib_dma_map_page(xprt->sc_cm_id->device,
- page, 0, PAGE_SIZE,
- DMA_FROM_DEVICE);
- if (ib_dma_mapping_error(xprt->sc_cm_id->device, pa))
+ pa = ib_dma_map_page(xprt->sc_device, page, 0,
+ PAGE_SIZE, DMA_FROM_DEVICE);
+ if (ib_dma_mapping_error(xprt->sc_device, pa))
goto err_put_ctxt;
svc_rdma_count_mappings(xprt, ctxt);
ctxt->sge[sge_no].addr = pa;
ctxt->sge[sge_no].length = PAGE_SIZE;
- ctxt->sge[sge_no].lkey = xprt->sc_pd->local_dma_lkey;
+ ctxt->sge[sge_no].lkey = xprt->sc_lkey;
ctxt->count = sge_no + 1;
buflen += PAGE_SIZE;
}
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 21/22] svcrdma: Clean-up svc_rdma_unmap_dma
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (19 preceding siblings ...)
2017-01-07 17:17 ` [PATCH v1 20/22] svcrdma: Clean-up in svc_rdma_post_recv Chuck Lever
@ 2017-01-07 17:18 ` Chuck Lever
2017-01-07 17:18 ` [PATCH v1 22/22] svcrdma: Re-order fields in svc_rdma_op_ctxt Chuck Lever
` (3 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:18 UTC (permalink / raw)
To: linux-rdma, linux-nfs
svcrdma no longer manages FRMRs, so there's no need to compare each
SGE's lkey with the PD's local_dma_lkey.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
net/sunrpc/xprtrdma/svc_rdma_transport.c | 20 +++++---------------
1 file changed, 5 insertions(+), 15 deletions(-)
diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index 9b60740..4500000 100644
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -223,23 +223,13 @@ struct svc_rdma_op_ctxt *svc_rdma_get_context(struct svcxprt_rdma *xprt)
void svc_rdma_unmap_dma(struct svc_rdma_op_ctxt *ctxt)
{
struct svcxprt_rdma *xprt = ctxt->xprt;
- struct ib_device *device = xprt->sc_cm_id->device;
- u32 lkey = xprt->sc_pd->local_dma_lkey;
unsigned int i;
- for (i = 0; i < ctxt->mapped_sges; i++) {
- /*
- * Unmap the DMA addr in the SGE if the lkey matches
- * the local_dma_lkey, otherwise, ignore it since it is
- * an FRMR lkey and will be unmapped later when the
- * last WR that uses it completes.
- */
- if (ctxt->sge[i].lkey == lkey)
- ib_dma_unmap_page(device,
- ctxt->sge[i].addr,
- ctxt->sge[i].length,
- ctxt->direction);
- }
+ for (i = 0; i < ctxt->mapped_sges; i++)
+ ib_dma_unmap_page(xprt->sc_device,
+ ctxt->sge[i].addr,
+ ctxt->sge[i].length,
+ ctxt->direction);
ctxt->mapped_sges = 0;
}
^ permalink raw reply related [flat|nested] 30+ messages in thread* [PATCH v1 22/22] svcrdma: Re-order fields in svc_rdma_op_ctxt
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (20 preceding siblings ...)
2017-01-07 17:18 ` [PATCH v1 21/22] svcrdma: Clean-up svc_rdma_unmap_dma Chuck Lever
@ 2017-01-07 17:18 ` Chuck Lever
2017-01-08 14:34 ` [PATCH v1 00/22] convert NFS server to new rdma_rw API Christoph Hellwig
` (2 subsequent siblings)
24 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-07 17:18 UTC (permalink / raw)
To: linux-rdma, linux-nfs
Attempt to group fields accessed during completion handling into one
cacheline.
Signed-off-by: Chuck Lever <chuck.lever@oracle.com>
---
include/linux/sunrpc/svc_rdma.h | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/include/linux/sunrpc/svc_rdma.h b/include/linux/sunrpc/svc_rdma.h
index 30c3ae9..6994039 100644
--- a/include/linux/sunrpc/svc_rdma.h
+++ b/include/linux/sunrpc/svc_rdma.h
@@ -77,7 +77,6 @@ enum {
*/
struct svc_rdma_op_ctxt {
struct list_head list;
- struct xdr_buf arg;
struct ib_cqe cqe;
u32 byte_len;
u32 position;
@@ -86,6 +85,7 @@ struct svc_rdma_op_ctxt {
int count;
unsigned int mapped_sges;
int hdr_count;
+ struct xdr_buf arg;
struct ib_send_wr send_wr;
struct ib_sge sge[1 + RPCRDMA_MAX_REQ_SIZE / PAGE_SIZE];
struct page *pages[RPCSVC_MAXPAGES];
^ permalink raw reply related [flat|nested] 30+ messages in thread* Re: [PATCH v1 00/22] convert NFS server to new rdma_rw API
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (21 preceding siblings ...)
2017-01-07 17:18 ` [PATCH v1 22/22] svcrdma: Re-order fields in svc_rdma_op_ctxt Chuck Lever
@ 2017-01-08 14:34 ` Christoph Hellwig
2017-01-08 17:19 ` Chuck Lever
2017-01-13 16:42 ` J. Bruce Fields
2017-01-13 22:04 ` Sagi Grimberg
24 siblings, 1 reply; 30+ messages in thread
From: Christoph Hellwig @ 2017-01-08 14:34 UTC (permalink / raw)
To: Chuck Lever; +Cc: linux-rdma, linux-nfs
On Sat, Jan 07, 2017 at 12:15:15PM -0500, Chuck Lever wrote:
> This series converts the Linux NFS server RPC-over-RDMA
> implementation to use the new core rdma_rw API and to poll its CQs
> in workqueue mode.
>
> Previously published work prototyped only the path that sends RPC
> replies. This series converts both send and receive sides, and
> includes significant clean ups that result from using the new API.
>
> This series has been successfully tested with NFSv3, 4.0, and 4.1;
> with clients that use FRWR and FMR; and with sec=sys, krb5, krb5i,
> and krb5p.
Any performane improvements (or regressions) with it?
> 10 files changed, 1621 insertions(+), 1656 deletions(-)
Hmm, that's not much less code, especially compared to the
other target side drivers where we remove a very substantial amount of
code. I guess I need to spend some time with the individual patches
to understand why.
^ permalink raw reply [flat|nested] 30+ messages in thread* Re: [PATCH v1 00/22] convert NFS server to new rdma_rw API
2017-01-08 14:34 ` [PATCH v1 00/22] convert NFS server to new rdma_rw API Christoph Hellwig
@ 2017-01-08 17:19 ` Chuck Lever
0 siblings, 0 replies; 30+ messages in thread
From: Chuck Lever @ 2017-01-08 17:19 UTC (permalink / raw)
To: Christoph Hellwig; +Cc: linux-rdma, Linux NFS Mailing List
> On Jan 8, 2017, at 9:34 AM, Christoph Hellwig <hch@infradead.org> wrote:
>
> On Sat, Jan 07, 2017 at 12:15:15PM -0500, Chuck Lever wrote:
>> This series converts the Linux NFS server RPC-over-RDMA
>> implementation to use the new core rdma_rw API and to poll its CQs
>> in workqueue mode.
>>
>> Previously published work prototyped only the path that sends RPC
>> replies. This series converts both send and receive sides, and
>> includes significant clean ups that result from using the new API.
>>
>> This series has been successfully tested with NFSv3, 4.0, and 4.1;
>> with clients that use FRWR and FMR; and with sec=sys, krb5, krb5i,
>> and krb5p.
>
> Any performane improvements (or regressions) with it?
NFS WRITE throughput is slightly lower. The maximum is still in
the 25 to 30 Gbps range on FDR. We have previously discussed two
additional major improvements:
- allocating memory and posting RDMA Reads from the Receive
completion handler
- utilizing splice where possible in the NFS server's write path
I'm still thinking about how these should work. IMO it's not a
reason to hold up review and merging what has been done so far.
For NFS READ, I can reach fabric speed. However, the results vary
significantly due to congestion at the client HCA. This is not
a new issue with this patch series.
Some improvement noted in maximum 8KB IOPS.
>> 10 files changed, 1621 insertions(+), 1656 deletions(-)
>
> Hmm, that's not much less code, especially compared to the
> other target side drivers where we remove a very substantial amount of
> code. I guess I need to spend some time with the individual patches
> to understand why.
Some possible reasons:
RPC-over-RDMA is more complex than the other RDMA-enabled storage
protocols, allowing more than one RDMA segment (R_key) per RPC
transaction.
For example, a client that requests a 1MB NFS READ payload is
permitted to split the receive buffers among multiple RDMA segments
with unique R_keys. As I understand the rdma_rw API, each R_key
would need its own rdma_ctx.
Basic FRWR does not support discontiguous segments (one R_key with
a memory region that has gaps). The send path has to transmit xdr_bufs
where the head, page list, and tail are separate memory regions. This
is needed, for example, when sending a whole RPC Reply via RDMA (a
Reply chunk).
Therefore for full generality RDMA segments have to be broken up
across the RPC Reply's xdr_buf, requiring multiple rdma_ctx's. The
RDMA Read logic does not have this constraint: it always reads into a
list of pages, which is straightforward to convert into a single
scatterlist.
There is some clean-up of the use of C structures to access received
messages before they are XDR decoded, and to marshal messages before
they are sent. This has been replaced with the more portable style
of using __be32 pointers, and accounts for a significant amount of
churn.
The new code has more documenting comments that explain the memory
allocation and DMA mapping architecture, and preface each public
function. I estimate this accounts for at least two to three hundred
lines of insertions, maybe more.
--
Chuck Lever
^ permalink raw reply [flat|nested] 30+ messages in thread
* Re: [PATCH v1 00/22] convert NFS server to new rdma_rw API
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (22 preceding siblings ...)
2017-01-08 14:34 ` [PATCH v1 00/22] convert NFS server to new rdma_rw API Christoph Hellwig
@ 2017-01-13 16:42 ` J. Bruce Fields
2017-01-13 16:54 ` Chuck Lever
2017-01-13 22:04 ` Sagi Grimberg
24 siblings, 1 reply; 30+ messages in thread
From: J. Bruce Fields @ 2017-01-13 16:42 UTC (permalink / raw)
To: Chuck Lever; +Cc: linux-rdma, linux-nfs
Can you summarize the motivation for the change, for someone (like me)
that doesn't know much about RDMA? Or does "significant cleanups" cover
it?
Are there any regressions? For example, will it still support all the
same hardware? Is there any change that will be noticeable to NFS/RDMA
users?
--b.
On Sat, Jan 07, 2017 at 12:15:15PM -0500, Chuck Lever wrote:
> This series converts the Linux NFS server RPC-over-RDMA
> implementation to use the new core rdma_rw API and to poll its CQs
> in workqueue mode.
>
> Previously published work prototyped only the path that sends RPC
> replies. This series converts both send and receive sides, and
> includes significant clean ups that result from using the new API.
>
> This series has been successfully tested with NFSv3, 4.0, and 4.1;
> with clients that use FRWR and FMR; and with sec=sys, krb5, krb5i,
> and krb5p.
>
> Available in the "nfsd-rdma-rw-api" topic branch of this git repo:
>
> git://git.linux-nfs.org/projects/cel/cel-2.6.git
>
> And for online review:
>
> http://git.linux-nfs.org/?p=cel/cel-2.6.git;a=shortlog;h=refs/heads/nfsd-rdma-rw-api
>
> ---
>
> Chuck Lever (22):
> svcrdma: Another sendto chunk list parsing update
> svcrdma: Replace RPCRDMA_SQ_DEPTH_MULT
> svcrdma: Remove unused sc_dto_q field
> svcrdma: Combine list fields in struct svc_rdma_op_ctxt
> svcrdma: Poll CQs in "workqueue" mode
> svcrdma: Clean up RPC-over-RDMA Reply header encoder
> svcrdma: Clean up RPC-over-RDMA Call header decoder
> svcrdma: Introduce local rdma_rw API helpers
> svcrdma: Use rdma_rw API in RPC reply path
> svcrdma: Backchannel sendto clean up
> svcrdma: Clean up RDMA_ERROR path
> svcrdma: Reduce size of sge array in struct svc_rdma_op_ctxt
> svcrdma: Remove old RDMA Write completion handlers
> svcrdma: Remove the req_map cache
> svcrdma: Clean up RPC-over-RDMA backchannel reply processing
> svcrdma: Use generic RDMA R/W API in RPC Call path
> svcrdma: Remove unused Read completion handlers
> svcrdma: Remove frmr cache
> svcrdma: Clean up after converting svc_rdma_recvfrom to rdma_rw API
> svcrdma: Clean-up in svc_rdma_post_recv
> svcrdma: Clean-up svc_rdma_unmap_dma
> svcrdma: Re-order fields in svc_rdma_op_ctxt
>
>
> include/linux/sunrpc/rpc_rdma.h | 12
> include/linux/sunrpc/svc_rdma.h | 147 ++---
> net/sunrpc/xprtrdma/Makefile | 2
> net/sunrpc/xprtrdma/svc_rdma.c | 8
> net/sunrpc/xprtrdma/svc_rdma_backchannel.c | 73 +--
> net/sunrpc/xprtrdma/svc_rdma_marshal.c | 478 +++++++++--------
> net/sunrpc/xprtrdma/svc_rdma_recvfrom.c | 518 +++---------------
> net/sunrpc/xprtrdma/svc_rdma_rw.c | 731 ++++++++++++++++++++++++++
> net/sunrpc/xprtrdma/svc_rdma_sendto.c | 802 +++++++++++-----------------
> net/sunrpc/xprtrdma/svc_rdma_transport.c | 506 +++++-------------
> 10 files changed, 1621 insertions(+), 1656 deletions(-)
> create mode 100644 net/sunrpc/xprtrdma/svc_rdma_rw.c
>
> --
> Chuck Lever
> --
> To unsubscribe from this list: send the line "unsubscribe linux-nfs" in
> the body of a message to majordomo@vger.kernel.org
> More majordomo info at http://vger.kernel.org/majordomo-info.html
^ permalink raw reply [flat|nested] 30+ messages in thread* Re: [PATCH v1 00/22] convert NFS server to new rdma_rw API
2017-01-13 16:42 ` J. Bruce Fields
@ 2017-01-13 16:54 ` Chuck Lever
2017-01-13 17:08 ` J. Bruce Fields
0 siblings, 1 reply; 30+ messages in thread
From: Chuck Lever @ 2017-01-13 16:54 UTC (permalink / raw)
To: J. Bruce Fields; +Cc: linux-rdma, Linux NFS Mailing List
Hi Bruce-
> On Jan 13, 2017, at 11:42 AM, bfields@fieldses.org wrote:
>
> Can you summarize the motivation for the change, for someone (like me)
> that doesn't know much about RDMA? Or does "significant cleanups" cover
> it?
Last year, Sagi and Christoph introduced a new set of APIs to the
RDMA core that enable ULPs like iSER and NFS/RDMA to leave the details
of certain aspects of operation to the core, instead of all the ULPs
having to implement the details themselves. Kind of a code refactoring /
de-duplication.
Then, going forward, fixes and improvements made to the logic behind
this new API will be available to all ULPs that use the new API.
This series converts the server-side RPC-over-RDMA implementation
to use the new API.
> Are there any regressions? For example, will it still support all the
> same hardware? Is there any change that will be noticeable to NFS/RDMA
> users?
All the same hardware is supported. The standing policy for ULPs is
that they must work with all in-tree HCA drivers and devices.
The goal of this work is that there is no user-visible change in
functionality or performance.
I know it's a lot to consider, and I'm happy to split up the series
or individual patches to help review.
> --b.
>
> On Sat, Jan 07, 2017 at 12:15:15PM -0500, Chuck Lever wrote:
>> This series converts the Linux NFS server RPC-over-RDMA
>> implementation to use the new core rdma_rw API and to poll its CQs
>> in workqueue mode.
>>
>> Previously published work prototyped only the path that sends RPC
>> replies. This series converts both send and receive sides, and
>> includes significant clean ups that result from using the new API.
>>
>> This series has been successfully tested with NFSv3, 4.0, and 4.1;
>> with clients that use FRWR and FMR; and with sec=sys, krb5, krb5i,
>> and krb5p.
>>
>> Available in the "nfsd-rdma-rw-api" topic branch of this git repo:
>>
>> git://git.linux-nfs.org/projects/cel/cel-2.6.git
>>
>> And for online review:
>>
>> http://git.linux-nfs.org/?p=cel/cel-2.6.git;a=shortlog;h=refs/heads/nfsd-rdma-rw-api
>>
>> ---
>>
>> Chuck Lever (22):
>> svcrdma: Another sendto chunk list parsing update
>> svcrdma: Replace RPCRDMA_SQ_DEPTH_MULT
>> svcrdma: Remove unused sc_dto_q field
>> svcrdma: Combine list fields in struct svc_rdma_op_ctxt
>> svcrdma: Poll CQs in "workqueue" mode
>> svcrdma: Clean up RPC-over-RDMA Reply header encoder
>> svcrdma: Clean up RPC-over-RDMA Call header decoder
>> svcrdma: Introduce local rdma_rw API helpers
>> svcrdma: Use rdma_rw API in RPC reply path
>> svcrdma: Backchannel sendto clean up
>> svcrdma: Clean up RDMA_ERROR path
>> svcrdma: Reduce size of sge array in struct svc_rdma_op_ctxt
>> svcrdma: Remove old RDMA Write completion handlers
>> svcrdma: Remove the req_map cache
>> svcrdma: Clean up RPC-over-RDMA backchannel reply processing
>> svcrdma: Use generic RDMA R/W API in RPC Call path
>> svcrdma: Remove unused Read completion handlers
>> svcrdma: Remove frmr cache
>> svcrdma: Clean up after converting svc_rdma_recvfrom to rdma_rw API
>> svcrdma: Clean-up in svc_rdma_post_recv
>> svcrdma: Clean-up svc_rdma_unmap_dma
>> svcrdma: Re-order fields in svc_rdma_op_ctxt
>>
>>
>> include/linux/sunrpc/rpc_rdma.h | 12
>> include/linux/sunrpc/svc_rdma.h | 147 ++---
>> net/sunrpc/xprtrdma/Makefile | 2
>> net/sunrpc/xprtrdma/svc_rdma.c | 8
>> net/sunrpc/xprtrdma/svc_rdma_backchannel.c | 73 +--
>> net/sunrpc/xprtrdma/svc_rdma_marshal.c | 478 +++++++++--------
>> net/sunrpc/xprtrdma/svc_rdma_recvfrom.c | 518 +++---------------
>> net/sunrpc/xprtrdma/svc_rdma_rw.c | 731 ++++++++++++++++++++++++++
>> net/sunrpc/xprtrdma/svc_rdma_sendto.c | 802 +++++++++++-----------------
>> net/sunrpc/xprtrdma/svc_rdma_transport.c | 506 +++++-------------
>> 10 files changed, 1621 insertions(+), 1656 deletions(-)
>> create mode 100644 net/sunrpc/xprtrdma/svc_rdma_rw.c
>>
>> --
>> Chuck Lever
>> --
>> To unsubscribe from this list: send the line "unsubscribe linux-nfs" in
>> the body of a message to majordomo@vger.kernel.org
>> More majordomo info at http://vger.kernel.org/majordomo-info.html
> --
> To unsubscribe from this list: send the line "unsubscribe linux-rdma" in
> the body of a message to majordomo@vger.kernel.org
> More majordomo info at http://vger.kernel.org/majordomo-info.html
--
Chuck Lever
^ permalink raw reply [flat|nested] 30+ messages in thread
* Re: [PATCH v1 00/22] convert NFS server to new rdma_rw API
2017-01-13 16:54 ` Chuck Lever
@ 2017-01-13 17:08 ` J. Bruce Fields
0 siblings, 0 replies; 30+ messages in thread
From: J. Bruce Fields @ 2017-01-13 17:08 UTC (permalink / raw)
To: Chuck Lever; +Cc: linux-rdma, Linux NFS Mailing List
On Fri, Jan 13, 2017 at 11:54:01AM -0500, Chuck Lever wrote:
> Hi Bruce-
>
>
> > On Jan 13, 2017, at 11:42 AM, bfields@fieldses.org wrote:
> >
> > Can you summarize the motivation for the change, for someone (like me)
> > that doesn't know much about RDMA? Or does "significant cleanups" cover
> > it?
>
> Last year, Sagi and Christoph introduced a new set of APIs to the
> RDMA core that enable ULPs like iSER and NFS/RDMA to leave the details
> of certain aspects of operation to the core, instead of all the ULPs
> having to implement the details themselves. Kind of a code refactoring /
> de-duplication.
>
> Then, going forward, fixes and improvements made to the logic behind
> this new API will be available to all ULPs that use the new API.
>
> This series converts the server-side RPC-over-RDMA implementation
> to use the new API.
>
>
> > Are there any regressions? For example, will it still support all the
> > same hardware? Is there any change that will be noticeable to NFS/RDMA
> > users?
>
> All the same hardware is supported. The standing policy for ULPs is
> that they must work with all in-tree HCA drivers and devices.
>
> The goal of this work is that there is no user-visible change in
> functionality or performance.
>
> I know it's a lot to consider, and I'm happy to split up the series
> or individual patches to help review.
Eh, I'll try to do a little, but realistically I'm mainly trusting you
and Christoph, so it's more important that the process work for you.
--b.
^ permalink raw reply [flat|nested] 30+ messages in thread
* Re: [PATCH v1 00/22] convert NFS server to new rdma_rw API
2017-01-07 17:15 [PATCH v1 00/22] convert NFS server to new rdma_rw API Chuck Lever
` (23 preceding siblings ...)
2017-01-13 16:42 ` J. Bruce Fields
@ 2017-01-13 22:04 ` Sagi Grimberg
24 siblings, 0 replies; 30+ messages in thread
From: Sagi Grimberg @ 2017-01-13 22:04 UTC (permalink / raw)
To: Chuck Lever, linux-rdma, linux-nfs
Hey Chuck,
> This series converts the Linux NFS server RPC-over-RDMA
> implementation to use the new core rdma_rw API and to poll its CQs
> in workqueue mode.
Great to see this happening!
I'll try and have a look early next week.
Hope to see a simpler code with the new API.
^ permalink raw reply [flat|nested] 30+ messages in thread