2 * Copyright (c) 2014 Open Grid Computing, Inc. All rights reserved.
3 * Copyright (c) 2005-2006 Network Appliance, Inc. All rights reserved.
5 * This software is available to you under a choice of one of two
6 * licenses. You may choose to be licensed under the terms of the GNU
7 * General Public License (GPL) Version 2, available from the file
8 * COPYING in the main directory of this source tree, or the BSD-type
11 * Redistribution and use in source and binary forms, with or without
12 * modification, are permitted provided that the following conditions
15 * Redistributions of source code must retain the above copyright
16 * notice, this list of conditions and the following disclaimer.
18 * Redistributions in binary form must reproduce the above
19 * copyright notice, this list of conditions and the following
20 * disclaimer in the documentation and/or other materials provided
21 * with the distribution.
23 * Neither the name of the Network Appliance, Inc. nor the names of
24 * its contributors may be used to endorse or promote products
25 * derived from this software without specific prior written
28 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
29 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
30 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
31 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
32 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
33 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
34 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
35 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
36 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
37 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
38 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
40 * Author: Tom Tucker <tom@opengridcomputing.com>
43 #include <linux/sunrpc/debug.h>
44 #include <linux/sunrpc/rpc_rdma.h>
45 #include <linux/spinlock.h>
46 #include <asm/unaligned.h>
47 #include <rdma/ib_verbs.h>
48 #include <rdma/rdma_cm.h>
49 #include <linux/sunrpc/svc_rdma.h>
51 #define RPCDBG_FACILITY RPCDBG_SVCXPRT
53 static int map_xdr(struct svcxprt_rdma
*xprt
,
55 struct svc_rdma_req_map
*vec
)
64 (xdr
->head
[0].iov_len
+ xdr
->page_len
+ xdr
->tail
[0].iov_len
)) {
65 pr_err("svcrdma: map_xdr: XDR buffer length error\n");
69 /* Skip the first sge, this is for the RPCRDMA header */
73 vec
->sge
[sge_no
].iov_base
= xdr
->head
[0].iov_base
;
74 vec
->sge
[sge_no
].iov_len
= xdr
->head
[0].iov_len
;
79 page_bytes
= xdr
->page_len
;
80 page_off
= xdr
->page_base
;
82 vec
->sge
[sge_no
].iov_base
=
83 page_address(xdr
->pages
[page_no
]) + page_off
;
84 sge_bytes
= min_t(u32
, page_bytes
, (PAGE_SIZE
- page_off
));
85 page_bytes
-= sge_bytes
;
86 vec
->sge
[sge_no
].iov_len
= sge_bytes
;
90 page_off
= 0; /* reset for next time through loop */
94 if (xdr
->tail
[0].iov_len
) {
95 vec
->sge
[sge_no
].iov_base
= xdr
->tail
[0].iov_base
;
96 vec
->sge
[sge_no
].iov_len
= xdr
->tail
[0].iov_len
;
100 dprintk("svcrdma: map_xdr: sge_no %d page_no %d "
101 "page_base %u page_len %u head_len %zu tail_len %zu\n",
102 sge_no
, page_no
, xdr
->page_base
, xdr
->page_len
,
103 xdr
->head
[0].iov_len
, xdr
->tail
[0].iov_len
);
109 static dma_addr_t
dma_map_xdr(struct svcxprt_rdma
*xprt
,
111 u32 xdr_off
, size_t len
, int dir
)
115 if (xdr_off
< xdr
->head
[0].iov_len
) {
116 /* This offset is in the head */
117 xdr_off
+= (unsigned long)xdr
->head
[0].iov_base
& ~PAGE_MASK
;
118 page
= virt_to_page(xdr
->head
[0].iov_base
);
120 xdr_off
-= xdr
->head
[0].iov_len
;
121 if (xdr_off
< xdr
->page_len
) {
122 /* This offset is in the page list */
123 xdr_off
+= xdr
->page_base
;
124 page
= xdr
->pages
[xdr_off
>> PAGE_SHIFT
];
125 xdr_off
&= ~PAGE_MASK
;
127 /* This offset is in the tail */
128 xdr_off
-= xdr
->page_len
;
129 xdr_off
+= (unsigned long)
130 xdr
->tail
[0].iov_base
& ~PAGE_MASK
;
131 page
= virt_to_page(xdr
->tail
[0].iov_base
);
134 dma_addr
= ib_dma_map_page(xprt
->sc_cm_id
->device
, page
, xdr_off
,
135 min_t(size_t, PAGE_SIZE
, len
), dir
);
140 * - The specified write_len can be represented in sc_max_sge * PAGE_SIZE
142 static int send_write(struct svcxprt_rdma
*xprt
, struct svc_rqst
*rqstp
,
144 u32 xdr_off
, int write_len
,
145 struct svc_rdma_req_map
*vec
)
147 struct ib_send_wr write_wr
;
154 struct svc_rdma_op_ctxt
*ctxt
;
156 if (vec
->count
> RPCSVC_MAXPAGES
) {
157 pr_err("svcrdma: Too many pages (%lu)\n", vec
->count
);
161 dprintk("svcrdma: RDMA_WRITE rmr=%x, to=%llx, xdr_off=%d, "
162 "write_len=%d, vec->sge=%p, vec->count=%lu\n",
163 rmr
, (unsigned long long)to
, xdr_off
,
164 write_len
, vec
->sge
, vec
->count
);
166 ctxt
= svc_rdma_get_context(xprt
);
167 ctxt
->direction
= DMA_TO_DEVICE
;
170 /* Find the SGE associated with xdr_off */
171 for (bc
= xdr_off
, xdr_sge_no
= 1; bc
&& xdr_sge_no
< vec
->count
;
173 if (vec
->sge
[xdr_sge_no
].iov_len
> bc
)
175 bc
-= vec
->sge
[xdr_sge_no
].iov_len
;
182 /* Copy the remaining SGE */
184 sge_bytes
= min_t(size_t,
185 bc
, vec
->sge
[xdr_sge_no
].iov_len
-sge_off
);
186 sge
[sge_no
].length
= sge_bytes
;
188 dma_map_xdr(xprt
, &rqstp
->rq_res
, xdr_off
,
189 sge_bytes
, DMA_TO_DEVICE
);
190 xdr_off
+= sge_bytes
;
191 if (ib_dma_mapping_error(xprt
->sc_cm_id
->device
,
194 atomic_inc(&xprt
->sc_dma_used
);
195 sge
[sge_no
].lkey
= xprt
->sc_dma_lkey
;
200 if (xdr_sge_no
> vec
->count
) {
201 pr_err("svcrdma: Too many sges (%d)\n", xdr_sge_no
);
205 if (sge_no
== xprt
->sc_max_sge
)
209 /* Prepare WRITE WR */
210 memset(&write_wr
, 0, sizeof write_wr
);
211 ctxt
->wr_op
= IB_WR_RDMA_WRITE
;
212 write_wr
.wr_id
= (unsigned long)ctxt
;
213 write_wr
.sg_list
= &sge
[0];
214 write_wr
.num_sge
= sge_no
;
215 write_wr
.opcode
= IB_WR_RDMA_WRITE
;
216 write_wr
.send_flags
= IB_SEND_SIGNALED
;
217 write_wr
.wr
.rdma
.rkey
= rmr
;
218 write_wr
.wr
.rdma
.remote_addr
= to
;
221 atomic_inc(&rdma_stat_write
);
222 if (svc_rdma_send(xprt
, &write_wr
))
224 return write_len
- bc
;
226 svc_rdma_unmap_dma(ctxt
);
227 svc_rdma_put_context(ctxt
, 0);
228 /* Fatal error, close transport */
232 static int send_write_chunks(struct svcxprt_rdma
*xprt
,
233 struct rpcrdma_msg
*rdma_argp
,
234 struct rpcrdma_msg
*rdma_resp
,
235 struct svc_rqst
*rqstp
,
236 struct svc_rdma_req_map
*vec
)
238 u32 xfer_len
= rqstp
->rq_res
.page_len
+ rqstp
->rq_res
.tail
[0].iov_len
;
244 struct rpcrdma_write_array
*arg_ary
;
245 struct rpcrdma_write_array
*res_ary
;
248 arg_ary
= svc_rdma_get_write_array(rdma_argp
);
251 res_ary
= (struct rpcrdma_write_array
*)
252 &rdma_resp
->rm_body
.rm_chunks
[1];
254 /* Write chunks start at the pagelist */
255 nchunks
= be32_to_cpu(arg_ary
->wc_nchunks
);
256 for (xdr_off
= rqstp
->rq_res
.head
[0].iov_len
, chunk_no
= 0;
257 xfer_len
&& chunk_no
< nchunks
;
259 struct rpcrdma_segment
*arg_ch
;
262 arg_ch
= &arg_ary
->wc_array
[chunk_no
].wc_target
;
263 write_len
= min(xfer_len
, be32_to_cpu(arg_ch
->rs_length
));
265 /* Prepare the response chunk given the length actually
267 xdr_decode_hyper((__be32
*)&arg_ch
->rs_offset
, &rs_offset
);
268 svc_rdma_xdr_encode_array_chunk(res_ary
, chunk_no
,
274 ret
= send_write(xprt
, rqstp
,
275 be32_to_cpu(arg_ch
->rs_handle
),
276 rs_offset
+ chunk_off
,
281 dprintk("svcrdma: RDMA_WRITE failed, ret=%d\n",
291 /* Update the req with the number of chunks actually used */
292 svc_rdma_xdr_encode_write_list(rdma_resp
, chunk_no
);
294 return rqstp
->rq_res
.page_len
+ rqstp
->rq_res
.tail
[0].iov_len
;
297 static int send_reply_chunks(struct svcxprt_rdma
*xprt
,
298 struct rpcrdma_msg
*rdma_argp
,
299 struct rpcrdma_msg
*rdma_resp
,
300 struct svc_rqst
*rqstp
,
301 struct svc_rdma_req_map
*vec
)
303 u32 xfer_len
= rqstp
->rq_res
.len
;
309 struct rpcrdma_segment
*ch
;
310 struct rpcrdma_write_array
*arg_ary
;
311 struct rpcrdma_write_array
*res_ary
;
314 arg_ary
= svc_rdma_get_reply_array(rdma_argp
);
317 /* XXX: need to fix when reply lists occur with read-list and or
319 res_ary
= (struct rpcrdma_write_array
*)
320 &rdma_resp
->rm_body
.rm_chunks
[2];
322 /* xdr offset starts at RPC message */
323 nchunks
= be32_to_cpu(arg_ary
->wc_nchunks
);
324 for (xdr_off
= 0, chunk_no
= 0;
325 xfer_len
&& chunk_no
< nchunks
;
328 ch
= &arg_ary
->wc_array
[chunk_no
].wc_target
;
329 write_len
= min(xfer_len
, be32_to_cpu(ch
->rs_length
));
331 /* Prepare the reply chunk given the length actually
333 xdr_decode_hyper((__be32
*)&ch
->rs_offset
, &rs_offset
);
334 svc_rdma_xdr_encode_array_chunk(res_ary
, chunk_no
,
335 ch
->rs_handle
, ch
->rs_offset
,
339 ret
= send_write(xprt
, rqstp
,
340 be32_to_cpu(ch
->rs_handle
),
341 rs_offset
+ chunk_off
,
346 dprintk("svcrdma: RDMA_WRITE failed, ret=%d\n",
356 /* Update the req with the number of chunks actually used */
357 svc_rdma_xdr_encode_reply_array(res_ary
, chunk_no
);
359 return rqstp
->rq_res
.len
;
362 /* This function prepares the portion of the RPCRDMA message to be
363 * sent in the RDMA_SEND. This function is called after data sent via
364 * RDMA has already been transmitted. There are three cases:
365 * - The RPCRDMA header, RPC header, and payload are all sent in a
366 * single RDMA_SEND. This is the "inline" case.
367 * - The RPCRDMA header and some portion of the RPC header and data
368 * are sent via this RDMA_SEND and another portion of the data is
370 * - The RPCRDMA header [NOMSG] is sent in this RDMA_SEND and the RPC
371 * header and data are all transmitted via RDMA.
372 * In all three cases, this function prepares the RPCRDMA header in
373 * sge[0], the 'type' parameter indicates the type to place in the
374 * RPCRDMA header, and the 'byte_count' field indicates how much of
375 * the XDR to include in this RDMA_SEND. NB: The offset of the payload
376 * to send is zero in the XDR.
378 static int send_reply(struct svcxprt_rdma
*rdma
,
379 struct svc_rqst
*rqstp
,
381 struct rpcrdma_msg
*rdma_resp
,
382 struct svc_rdma_op_ctxt
*ctxt
,
383 struct svc_rdma_req_map
*vec
,
386 struct ib_send_wr send_wr
;
393 /* Post a recv buffer to handle another request. */
394 ret
= svc_rdma_post_recv(rdma
);
397 "svcrdma: could not post a receive buffer, err=%d."
398 "Closing transport %p.\n", ret
, rdma
);
399 set_bit(XPT_CLOSE
, &rdma
->sc_xprt
.xpt_flags
);
400 svc_rdma_put_context(ctxt
, 0);
404 /* Prepare the context */
405 ctxt
->pages
[0] = page
;
408 /* Prepare the SGE for the RPCRDMA Header */
409 ctxt
->sge
[0].lkey
= rdma
->sc_dma_lkey
;
410 ctxt
->sge
[0].length
= svc_rdma_xdr_get_reply_hdr_len(rdma_resp
);
412 ib_dma_map_page(rdma
->sc_cm_id
->device
, page
, 0,
413 ctxt
->sge
[0].length
, DMA_TO_DEVICE
);
414 if (ib_dma_mapping_error(rdma
->sc_cm_id
->device
, ctxt
->sge
[0].addr
))
416 atomic_inc(&rdma
->sc_dma_used
);
418 ctxt
->direction
= DMA_TO_DEVICE
;
420 /* Map the payload indicated by 'byte_count' */
421 for (sge_no
= 1; byte_count
&& sge_no
< vec
->count
; sge_no
++) {
423 sge_bytes
= min_t(size_t, vec
->sge
[sge_no
].iov_len
, byte_count
);
424 byte_count
-= sge_bytes
;
425 ctxt
->sge
[sge_no
].addr
=
426 dma_map_xdr(rdma
, &rqstp
->rq_res
, xdr_off
,
427 sge_bytes
, DMA_TO_DEVICE
);
428 xdr_off
+= sge_bytes
;
429 if (ib_dma_mapping_error(rdma
->sc_cm_id
->device
,
430 ctxt
->sge
[sge_no
].addr
))
432 atomic_inc(&rdma
->sc_dma_used
);
433 ctxt
->sge
[sge_no
].lkey
= rdma
->sc_dma_lkey
;
434 ctxt
->sge
[sge_no
].length
= sge_bytes
;
436 if (byte_count
!= 0) {
437 pr_err("svcrdma: Could not map %d bytes\n", byte_count
);
441 /* Save all respages in the ctxt and remove them from the
442 * respages array. They are our pages until the I/O
445 pages
= rqstp
->rq_next_page
- rqstp
->rq_respages
;
446 for (page_no
= 0; page_no
< pages
; page_no
++) {
447 ctxt
->pages
[page_no
+1] = rqstp
->rq_respages
[page_no
];
449 rqstp
->rq_respages
[page_no
] = NULL
;
451 * If there are more pages than SGE, terminate SGE
452 * list so that svc_rdma_unmap_dma doesn't attempt to
455 if (page_no
+1 >= sge_no
)
456 ctxt
->sge
[page_no
+1].length
= 0;
458 rqstp
->rq_next_page
= rqstp
->rq_respages
+ 1;
460 if (sge_no
> rdma
->sc_max_sge
) {
461 pr_err("svcrdma: Too many sges (%d)\n", sge_no
);
464 memset(&send_wr
, 0, sizeof send_wr
);
465 ctxt
->wr_op
= IB_WR_SEND
;
466 send_wr
.wr_id
= (unsigned long)ctxt
;
467 send_wr
.sg_list
= ctxt
->sge
;
468 send_wr
.num_sge
= sge_no
;
469 send_wr
.opcode
= IB_WR_SEND
;
470 send_wr
.send_flags
= IB_SEND_SIGNALED
;
472 ret
= svc_rdma_send(rdma
, &send_wr
);
479 svc_rdma_unmap_dma(ctxt
);
480 svc_rdma_put_context(ctxt
, 1);
484 void svc_rdma_prep_reply_hdr(struct svc_rqst
*rqstp
)
488 int svc_rdma_sendto(struct svc_rqst
*rqstp
)
490 struct svc_xprt
*xprt
= rqstp
->rq_xprt
;
491 struct svcxprt_rdma
*rdma
=
492 container_of(xprt
, struct svcxprt_rdma
, sc_xprt
);
493 struct rpcrdma_msg
*rdma_argp
;
494 struct rpcrdma_msg
*rdma_resp
;
495 struct rpcrdma_write_array
*reply_ary
;
496 enum rpcrdma_proc reply_type
;
499 struct page
*res_page
;
500 struct svc_rdma_op_ctxt
*ctxt
;
501 struct svc_rdma_req_map
*vec
;
503 dprintk("svcrdma: sending response for rqstp=%p\n", rqstp
);
505 /* Get the RDMA request header. The receive logic always
506 * places this at the start of page 0.
508 rdma_argp
= page_address(rqstp
->rq_pages
[0]);
510 /* Build an req vec for the XDR */
511 ctxt
= svc_rdma_get_context(rdma
);
512 ctxt
->direction
= DMA_TO_DEVICE
;
513 vec
= svc_rdma_get_req_map();
514 ret
= map_xdr(rdma
, &rqstp
->rq_res
, vec
);
517 inline_bytes
= rqstp
->rq_res
.len
;
519 /* Create the RDMA response header */
520 res_page
= alloc_page(GFP_KERNEL
| __GFP_NOFAIL
);
521 rdma_resp
= page_address(res_page
);
522 reply_ary
= svc_rdma_get_reply_array(rdma_argp
);
524 reply_type
= RDMA_NOMSG
;
526 reply_type
= RDMA_MSG
;
527 svc_rdma_xdr_encode_reply_header(rdma
, rdma_argp
,
528 rdma_resp
, reply_type
);
530 /* Send any write-chunk data and build resp write-list */
531 ret
= send_write_chunks(rdma
, rdma_argp
, rdma_resp
,
534 printk(KERN_ERR
"svcrdma: failed to send write chunks, rc=%d\n",
540 /* Send any reply-list data and update resp reply-list */
541 ret
= send_reply_chunks(rdma
, rdma_argp
, rdma_resp
,
544 printk(KERN_ERR
"svcrdma: failed to send reply chunks, rc=%d\n",
550 ret
= send_reply(rdma
, rqstp
, res_page
, rdma_resp
, ctxt
, vec
,
552 svc_rdma_put_req_map(vec
);
553 dprintk("svcrdma: send_reply returns %d\n", ret
);
559 svc_rdma_put_req_map(vec
);
560 svc_rdma_put_context(ctxt
, 0);
This page took 0.042727 seconds and 5 git commands to generate.