1 // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
3 * Copyright (c) 2014-2017 Oracle. All rights reserved.
4 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
6 * This software is available to you under a choice of one of two
7 * licenses. You may choose to be licensed under the terms of the GNU
8 * General Public License (GPL) Version 2, available from the file
9 * COPYING in the main directory of this source tree, or the BSD-type
12 * Redistribution and use in source and binary forms, with or without
13 * modification, are permitted provided that the following conditions
16 * Redistributions of source code must retain the above copyright
17 * notice, this list of conditions and the following disclaimer.
19 * Redistributions in binary form must reproduce the above
20 * copyright notice, this list of conditions and the following
21 * disclaimer in the documentation and/or other materials provided
22 * with the distribution.
24 * Neither the name of the Network Appliance, Inc. nor the names of
25 * its contributors may be used to endorse or promote products
26 * derived from this software without specific prior written
29 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
30 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
31 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
32 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
33 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
34 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
35 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
36 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
37 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
38 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
39 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
45 * This file contains the guts of the RPC RDMA protocol, and
46 * does marshaling/unmarshaling, etc. It is also where interfacing
47 * to the Linux RPC framework lives.
50 #include <linux/highmem.h>
52 #include <linux/sunrpc/svc_rdma.h>
54 #include "xprt_rdma.h"
55 #include <trace/events/rpcrdma.h>
57 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
58 # define RPCDBG_FACILITY RPCDBG_TRANS
61 /* Returns size of largest RPC-over-RDMA header in a Call message
63 * The largest Call header contains a full-size Read list and a
64 * minimal Reply chunk.
66 static unsigned int rpcrdma_max_call_header_size(unsigned int maxsegs
)
70 /* Fixed header fields and list discriminators */
71 size
= RPCRDMA_HDRLEN_MIN
;
73 /* Maximum Read list size */
74 size
= maxsegs
* rpcrdma_readchunk_maxsz
* sizeof(__be32
);
76 /* Minimal Read chunk size */
77 size
+= sizeof(__be32
); /* segment count */
78 size
+= rpcrdma_segment_maxsz
* sizeof(__be32
);
79 size
+= sizeof(__be32
); /* list discriminator */
81 dprintk("RPC: %s: max call header size = %u\n",
86 /* Returns size of largest RPC-over-RDMA header in a Reply message
88 * There is only one Write list or one Reply chunk per Reply
89 * message. The larger list is the Write list.
91 static unsigned int rpcrdma_max_reply_header_size(unsigned int maxsegs
)
95 /* Fixed header fields and list discriminators */
96 size
= RPCRDMA_HDRLEN_MIN
;
98 /* Maximum Write list size */
99 size
= sizeof(__be32
); /* segment count */
100 size
+= maxsegs
* rpcrdma_segment_maxsz
* sizeof(__be32
);
101 size
+= sizeof(__be32
); /* list discriminator */
103 dprintk("RPC: %s: max reply header size = %u\n",
108 void rpcrdma_set_max_header_sizes(struct rpcrdma_xprt
*r_xprt
)
110 struct rpcrdma_create_data_internal
*cdata
= &r_xprt
->rx_data
;
111 struct rpcrdma_ia
*ia
= &r_xprt
->rx_ia
;
112 unsigned int maxsegs
= ia
->ri_max_segs
;
114 ia
->ri_max_inline_write
= cdata
->inline_wsize
-
115 rpcrdma_max_call_header_size(maxsegs
);
116 ia
->ri_max_inline_read
= cdata
->inline_rsize
-
117 rpcrdma_max_reply_header_size(maxsegs
);
120 /* The client can send a request inline as long as the RPCRDMA header
121 * plus the RPC call fit under the transport's inline limit. If the
122 * combined call message size exceeds that limit, the client must use
123 * a Read chunk for this operation.
125 * A Read chunk is also required if sending the RPC call inline would
126 * exceed this device's max_sge limit.
128 static bool rpcrdma_args_inline(struct rpcrdma_xprt
*r_xprt
,
129 struct rpc_rqst
*rqst
)
131 struct xdr_buf
*xdr
= &rqst
->rq_snd_buf
;
132 unsigned int count
, remaining
, offset
;
134 if (xdr
->len
> r_xprt
->rx_ia
.ri_max_inline_write
)
138 remaining
= xdr
->page_len
;
139 offset
= offset_in_page(xdr
->page_base
);
140 count
= RPCRDMA_MIN_SEND_SGES
;
142 remaining
-= min_t(unsigned int,
143 PAGE_SIZE
- offset
, remaining
);
145 if (++count
> r_xprt
->rx_ia
.ri_max_send_sges
)
153 /* The client can't know how large the actual reply will be. Thus it
154 * plans for the largest possible reply for that particular ULP
155 * operation. If the maximum combined reply message size exceeds that
156 * limit, the client must provide a write list or a reply chunk for
159 static bool rpcrdma_results_inline(struct rpcrdma_xprt
*r_xprt
,
160 struct rpc_rqst
*rqst
)
162 struct rpcrdma_ia
*ia
= &r_xprt
->rx_ia
;
164 return rqst
->rq_rcv_buf
.buflen
<= ia
->ri_max_inline_read
;
167 /* Split @vec on page boundaries into SGEs. FMR registers pages, not
168 * a byte range. Other modes coalesce these SGEs into a single MR
171 * Returns pointer to next available SGE, and bumps the total number
174 static struct rpcrdma_mr_seg
*
175 rpcrdma_convert_kvec(struct kvec
*vec
, struct rpcrdma_mr_seg
*seg
,
178 u32 remaining
, page_offset
;
181 base
= vec
->iov_base
;
182 page_offset
= offset_in_page(base
);
183 remaining
= vec
->iov_len
;
186 seg
->mr_offset
= base
;
187 seg
->mr_len
= min_t(u32
, PAGE_SIZE
- page_offset
, remaining
);
188 remaining
-= seg
->mr_len
;
197 /* Convert @xdrbuf into SGEs no larger than a page each. As they
198 * are registered, these SGEs are then coalesced into RDMA segments
199 * when the selected memreg mode supports it.
201 * Returns positive number of SGEs consumed, or a negative errno.
205 rpcrdma_convert_iovs(struct rpcrdma_xprt
*r_xprt
, struct xdr_buf
*xdrbuf
,
206 unsigned int pos
, enum rpcrdma_chunktype type
,
207 struct rpcrdma_mr_seg
*seg
)
209 unsigned long page_base
;
211 struct page
**ppages
;
215 seg
= rpcrdma_convert_kvec(&xdrbuf
->head
[0], seg
, &n
);
217 len
= xdrbuf
->page_len
;
218 ppages
= xdrbuf
->pages
+ (xdrbuf
->page_base
>> PAGE_SHIFT
);
219 page_base
= offset_in_page(xdrbuf
->page_base
);
221 if (unlikely(!*ppages
)) {
222 /* XXX: Certain upper layer operations do
223 * not provide receive buffer pages.
225 *ppages
= alloc_page(GFP_ATOMIC
);
229 seg
->mr_page
= *ppages
;
230 seg
->mr_offset
= (char *)page_base
;
231 seg
->mr_len
= min_t(u32
, PAGE_SIZE
- page_base
, len
);
239 /* When encoding a Read chunk, the tail iovec contains an
240 * XDR pad and may be omitted.
242 if (type
== rpcrdma_readch
&& r_xprt
->rx_ia
.ri_implicit_roundup
)
245 /* When encoding a Write chunk, some servers need to see an
246 * extra segment for non-XDR-aligned Write chunks. The upper
247 * layer provides space in the tail iovec that may be used
250 if (type
== rpcrdma_writech
&& r_xprt
->rx_ia
.ri_implicit_roundup
)
253 if (xdrbuf
->tail
[0].iov_len
)
254 seg
= rpcrdma_convert_kvec(&xdrbuf
->tail
[0], seg
, &n
);
257 if (unlikely(n
> RPCRDMA_MAX_SEGS
))
263 encode_item_present(struct xdr_stream
*xdr
)
267 p
= xdr_reserve_space(xdr
, sizeof(*p
));
276 encode_item_not_present(struct xdr_stream
*xdr
)
280 p
= xdr_reserve_space(xdr
, sizeof(*p
));
289 xdr_encode_rdma_segment(__be32
*iptr
, struct rpcrdma_mr
*mr
)
291 *iptr
++ = cpu_to_be32(mr
->mr_handle
);
292 *iptr
++ = cpu_to_be32(mr
->mr_length
);
293 xdr_encode_hyper(iptr
, mr
->mr_offset
);
297 encode_rdma_segment(struct xdr_stream
*xdr
, struct rpcrdma_mr
*mr
)
301 p
= xdr_reserve_space(xdr
, 4 * sizeof(*p
));
305 xdr_encode_rdma_segment(p
, mr
);
310 encode_read_segment(struct xdr_stream
*xdr
, struct rpcrdma_mr
*mr
,
315 p
= xdr_reserve_space(xdr
, 6 * sizeof(*p
));
319 *p
++ = xdr_one
; /* Item present */
320 *p
++ = cpu_to_be32(position
);
321 xdr_encode_rdma_segment(p
, mr
);
325 /* Register and XDR encode the Read list. Supports encoding a list of read
326 * segments that belong to a single read chunk.
328 * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
330 * Read chunklist (a linked list):
331 * N elements, position P (same P for all chunks of same arg!):
332 * 1 - PHLOO - 1 - PHLOO - ... - 1 - PHLOO - 0
334 * Returns zero on success, or a negative errno if a failure occurred.
335 * @xdr is advanced to the next position in the stream.
337 * Only a single @pos value is currently supported.
340 rpcrdma_encode_read_list(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
341 struct rpc_rqst
*rqst
, enum rpcrdma_chunktype rtype
)
343 struct xdr_stream
*xdr
= &req
->rl_stream
;
344 struct rpcrdma_mr_seg
*seg
;
345 struct rpcrdma_mr
*mr
;
349 pos
= rqst
->rq_snd_buf
.head
[0].iov_len
;
350 if (rtype
== rpcrdma_areadch
)
352 seg
= req
->rl_segments
;
353 nsegs
= rpcrdma_convert_iovs(r_xprt
, &rqst
->rq_snd_buf
, pos
,
359 seg
= r_xprt
->rx_ia
.ri_ops
->ro_map(r_xprt
, seg
, nsegs
,
363 rpcrdma_mr_push(mr
, &req
->rl_registered
);
365 if (encode_read_segment(xdr
, mr
, pos
) < 0)
368 trace_xprtrdma_read_chunk(rqst
->rq_task
, pos
, mr
, nsegs
);
369 r_xprt
->rx_stats
.read_chunk_count
++;
370 nsegs
-= mr
->mr_nents
;
376 /* Register and XDR encode the Write list. Supports encoding a list
377 * containing one array of plain segments that belong to a single
380 * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
382 * Write chunklist (a list of (one) counted array):
384 * 1 - N - HLOO - HLOO - ... - HLOO - 0
386 * Returns zero on success, or a negative errno if a failure occurred.
387 * @xdr is advanced to the next position in the stream.
389 * Only a single Write chunk is currently supported.
392 rpcrdma_encode_write_list(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
393 struct rpc_rqst
*rqst
, enum rpcrdma_chunktype wtype
)
395 struct xdr_stream
*xdr
= &req
->rl_stream
;
396 struct rpcrdma_mr_seg
*seg
;
397 struct rpcrdma_mr
*mr
;
401 seg
= req
->rl_segments
;
402 nsegs
= rpcrdma_convert_iovs(r_xprt
, &rqst
->rq_rcv_buf
,
403 rqst
->rq_rcv_buf
.head
[0].iov_len
,
408 if (encode_item_present(xdr
) < 0)
410 segcount
= xdr_reserve_space(xdr
, sizeof(*segcount
));
411 if (unlikely(!segcount
))
413 /* Actual value encoded below */
417 seg
= r_xprt
->rx_ia
.ri_ops
->ro_map(r_xprt
, seg
, nsegs
,
421 rpcrdma_mr_push(mr
, &req
->rl_registered
);
423 if (encode_rdma_segment(xdr
, mr
) < 0)
426 trace_xprtrdma_write_chunk(rqst
->rq_task
, mr
, nsegs
);
427 r_xprt
->rx_stats
.write_chunk_count
++;
428 r_xprt
->rx_stats
.total_rdma_request
+= mr
->mr_length
;
430 nsegs
-= mr
->mr_nents
;
433 /* Update count of segments in this Write chunk */
434 *segcount
= cpu_to_be32(nchunks
);
439 /* Register and XDR encode the Reply chunk. Supports encoding an array
440 * of plain segments that belong to a single write (reply) chunk.
442 * Encoding key for single-list chunks (HLOO = Handle32 Length32 Offset64):
444 * Reply chunk (a counted array):
446 * 1 - N - HLOO - HLOO - ... - HLOO
448 * Returns zero on success, or a negative errno if a failure occurred.
449 * @xdr is advanced to the next position in the stream.
452 rpcrdma_encode_reply_chunk(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
453 struct rpc_rqst
*rqst
, enum rpcrdma_chunktype wtype
)
455 struct xdr_stream
*xdr
= &req
->rl_stream
;
456 struct rpcrdma_mr_seg
*seg
;
457 struct rpcrdma_mr
*mr
;
461 seg
= req
->rl_segments
;
462 nsegs
= rpcrdma_convert_iovs(r_xprt
, &rqst
->rq_rcv_buf
, 0, wtype
, seg
);
466 if (encode_item_present(xdr
) < 0)
468 segcount
= xdr_reserve_space(xdr
, sizeof(*segcount
));
469 if (unlikely(!segcount
))
471 /* Actual value encoded below */
475 seg
= r_xprt
->rx_ia
.ri_ops
->ro_map(r_xprt
, seg
, nsegs
,
479 rpcrdma_mr_push(mr
, &req
->rl_registered
);
481 if (encode_rdma_segment(xdr
, mr
) < 0)
484 trace_xprtrdma_reply_chunk(rqst
->rq_task
, mr
, nsegs
);
485 r_xprt
->rx_stats
.reply_chunk_count
++;
486 r_xprt
->rx_stats
.total_rdma_request
+= mr
->mr_length
;
488 nsegs
-= mr
->mr_nents
;
491 /* Update count of segments in the Reply chunk */
492 *segcount
= cpu_to_be32(nchunks
);
498 * rpcrdma_unmap_sendctx - DMA-unmap Send buffers
499 * @sc: sendctx containing SGEs to unmap
503 rpcrdma_unmap_sendctx(struct rpcrdma_sendctx
*sc
)
505 struct rpcrdma_ia
*ia
= &sc
->sc_xprt
->rx_ia
;
509 /* The first two SGEs contain the transport header and
510 * the inline buffer. These are always left mapped so
511 * they can be cheaply re-used.
513 sge
= &sc
->sc_sges
[2];
514 for (count
= sc
->sc_unmap_count
; count
; ++sge
, --count
)
515 ib_dma_unmap_page(ia
->ri_device
,
516 sge
->addr
, sge
->length
, DMA_TO_DEVICE
);
518 if (test_and_clear_bit(RPCRDMA_REQ_F_TX_RESOURCES
, &sc
->sc_req
->rl_flags
)) {
519 smp_mb__after_atomic();
520 wake_up_bit(&sc
->sc_req
->rl_flags
, RPCRDMA_REQ_F_TX_RESOURCES
);
524 /* Prepare an SGE for the RPC-over-RDMA transport header.
527 rpcrdma_prepare_hdr_sge(struct rpcrdma_ia
*ia
, struct rpcrdma_req
*req
,
530 struct rpcrdma_sendctx
*sc
= req
->rl_sendctx
;
531 struct rpcrdma_regbuf
*rb
= req
->rl_rdmabuf
;
532 struct ib_sge
*sge
= sc
->sc_sges
;
534 if (!rpcrdma_dma_map_regbuf(ia
, rb
))
536 sge
->addr
= rdmab_addr(rb
);
538 sge
->lkey
= rdmab_lkey(rb
);
540 ib_dma_sync_single_for_device(rdmab_device(rb
), sge
->addr
,
541 sge
->length
, DMA_TO_DEVICE
);
546 pr_err("rpcrdma: failed to DMA map a Send buffer\n");
550 /* Prepare the Send SGEs. The head and tail iovec, and each entry
551 * in the page list, gets its own SGE.
554 rpcrdma_prepare_msg_sges(struct rpcrdma_ia
*ia
, struct rpcrdma_req
*req
,
555 struct xdr_buf
*xdr
, enum rpcrdma_chunktype rtype
)
557 struct rpcrdma_sendctx
*sc
= req
->rl_sendctx
;
558 unsigned int sge_no
, page_base
, len
, remaining
;
559 struct rpcrdma_regbuf
*rb
= req
->rl_sendbuf
;
560 struct ib_device
*device
= ia
->ri_device
;
561 struct ib_sge
*sge
= sc
->sc_sges
;
562 u32 lkey
= ia
->ri_pd
->local_dma_lkey
;
563 struct page
*page
, **ppages
;
565 /* The head iovec is straightforward, as it is already
566 * DMA-mapped. Sync the content that has changed.
568 if (!rpcrdma_dma_map_regbuf(ia
, rb
))
571 sge
[sge_no
].addr
= rdmab_addr(rb
);
572 sge
[sge_no
].length
= xdr
->head
[0].iov_len
;
573 sge
[sge_no
].lkey
= rdmab_lkey(rb
);
574 ib_dma_sync_single_for_device(rdmab_device(rb
), sge
[sge_no
].addr
,
575 sge
[sge_no
].length
, DMA_TO_DEVICE
);
577 /* If there is a Read chunk, the page list is being handled
578 * via explicit RDMA, and thus is skipped here. However, the
579 * tail iovec may include an XDR pad for the page list, as
580 * well as additional content, and may not reside in the
581 * same page as the head iovec.
583 if (rtype
== rpcrdma_readch
) {
584 len
= xdr
->tail
[0].iov_len
;
586 /* Do not include the tail if it is only an XDR pad */
590 page
= virt_to_page(xdr
->tail
[0].iov_base
);
591 page_base
= offset_in_page(xdr
->tail
[0].iov_base
);
593 /* If the content in the page list is an odd length,
594 * xdr_write_pages() has added a pad at the beginning
595 * of the tail iovec. Force the tail's non-pad content
596 * to land at the next XDR position in the Send message.
598 page_base
+= len
& 3;
603 /* If there is a page list present, temporarily DMA map
604 * and prepare an SGE for each page to be sent.
607 ppages
= xdr
->pages
+ (xdr
->page_base
>> PAGE_SHIFT
);
608 page_base
= offset_in_page(xdr
->page_base
);
609 remaining
= xdr
->page_len
;
612 if (sge_no
> RPCRDMA_MAX_SEND_SGES
- 2)
613 goto out_mapping_overflow
;
615 len
= min_t(u32
, PAGE_SIZE
- page_base
, remaining
);
616 sge
[sge_no
].addr
= ib_dma_map_page(device
, *ppages
,
619 if (ib_dma_mapping_error(device
, sge
[sge_no
].addr
))
620 goto out_mapping_err
;
621 sge
[sge_no
].length
= len
;
622 sge
[sge_no
].lkey
= lkey
;
624 sc
->sc_unmap_count
++;
631 /* The tail iovec is not always constructed in the same
632 * page where the head iovec resides (see, for example,
633 * gss_wrap_req_priv). To neatly accommodate that case,
634 * DMA map it separately.
636 if (xdr
->tail
[0].iov_len
) {
637 page
= virt_to_page(xdr
->tail
[0].iov_base
);
638 page_base
= offset_in_page(xdr
->tail
[0].iov_base
);
639 len
= xdr
->tail
[0].iov_len
;
643 sge
[sge_no
].addr
= ib_dma_map_page(device
, page
,
646 if (ib_dma_mapping_error(device
, sge
[sge_no
].addr
))
647 goto out_mapping_err
;
648 sge
[sge_no
].length
= len
;
649 sge
[sge_no
].lkey
= lkey
;
650 sc
->sc_unmap_count
++;
654 sc
->sc_wr
.num_sge
+= sge_no
;
655 if (sc
->sc_unmap_count
)
656 __set_bit(RPCRDMA_REQ_F_TX_RESOURCES
, &req
->rl_flags
);
660 pr_err("rpcrdma: failed to DMA map a Send buffer\n");
663 out_mapping_overflow
:
664 rpcrdma_unmap_sendctx(sc
);
665 pr_err("rpcrdma: too many Send SGEs (%u)\n", sge_no
);
669 rpcrdma_unmap_sendctx(sc
);
670 pr_err("rpcrdma: Send mapping error\n");
675 * rpcrdma_prepare_send_sges - Construct SGEs for a Send WR
676 * @r_xprt: controlling transport
677 * @req: context of RPC Call being marshalled
678 * @hdrlen: size of transport header, in bytes
679 * @xdr: xdr_buf containing RPC Call
680 * @rtype: chunk type being encoded
682 * Returns 0 on success; otherwise a negative errno is returned.
685 rpcrdma_prepare_send_sges(struct rpcrdma_xprt
*r_xprt
,
686 struct rpcrdma_req
*req
, u32 hdrlen
,
687 struct xdr_buf
*xdr
, enum rpcrdma_chunktype rtype
)
689 req
->rl_sendctx
= rpcrdma_sendctx_get_locked(&r_xprt
->rx_buf
);
690 if (!req
->rl_sendctx
)
692 req
->rl_sendctx
->sc_wr
.num_sge
= 0;
693 req
->rl_sendctx
->sc_unmap_count
= 0;
694 req
->rl_sendctx
->sc_req
= req
;
695 __clear_bit(RPCRDMA_REQ_F_TX_RESOURCES
, &req
->rl_flags
);
697 if (!rpcrdma_prepare_hdr_sge(&r_xprt
->rx_ia
, req
, hdrlen
))
700 if (rtype
!= rpcrdma_areadch
)
701 if (!rpcrdma_prepare_msg_sges(&r_xprt
->rx_ia
, req
, xdr
, rtype
))
708 * rpcrdma_marshal_req - Marshal and send one RPC request
709 * @r_xprt: controlling transport
710 * @rqst: RPC request to be marshaled
712 * For the RPC in "rqst", this function:
713 * - Chooses the transfer mode (eg., RDMA_MSG or RDMA_NOMSG)
714 * - Registers Read, Write, and Reply chunks
715 * - Constructs the transport header
716 * - Posts a Send WR to send the transport header and request
719 * %0 if the RPC was sent successfully,
720 * %-ENOTCONN if the connection was lost,
721 * %-EAGAIN if the caller should call again with the same arguments,
722 * %-ENOBUFS if the caller should call again after a delay,
723 * %-EMSGSIZE if the transport header is too small,
724 * %-EIO if a permanent problem occurred while marshaling.
727 rpcrdma_marshal_req(struct rpcrdma_xprt
*r_xprt
, struct rpc_rqst
*rqst
)
729 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
730 struct xdr_stream
*xdr
= &req
->rl_stream
;
731 enum rpcrdma_chunktype rtype
, wtype
;
736 rpcrdma_set_xdrlen(&req
->rl_hdrbuf
, 0);
737 xdr_init_encode(xdr
, &req
->rl_hdrbuf
,
738 req
->rl_rdmabuf
->rg_base
);
740 /* Fixed header fields */
742 p
= xdr_reserve_space(xdr
, 4 * sizeof(*p
));
746 *p
++ = rpcrdma_version
;
747 *p
++ = cpu_to_be32(r_xprt
->rx_buf
.rb_max_requests
);
749 /* When the ULP employs a GSS flavor that guarantees integrity
750 * or privacy, direct data placement of individual data items
753 ddp_allowed
= !(rqst
->rq_cred
->cr_auth
->au_flags
&
754 RPCAUTH_AUTH_DATATOUCH
);
757 * Chunks needed for results?
759 * o If the expected result is under the inline threshold, all ops
761 * o Large read ops return data as write chunk(s), header as
763 * o Large non-read ops return as a single reply chunk.
765 if (rpcrdma_results_inline(r_xprt
, rqst
))
766 wtype
= rpcrdma_noch
;
767 else if (ddp_allowed
&& rqst
->rq_rcv_buf
.flags
& XDRBUF_READ
)
768 wtype
= rpcrdma_writech
;
770 wtype
= rpcrdma_replych
;
773 * Chunks needed for arguments?
775 * o If the total request is under the inline threshold, all ops
776 * are sent as inline.
777 * o Large write ops transmit data as read chunk(s), header as
779 * o Large non-write ops are sent with the entire message as a
780 * single read chunk (protocol 0-position special case).
782 * This assumes that the upper layer does not present a request
783 * that both has a data payload, and whose non-data arguments
784 * by themselves are larger than the inline threshold.
786 if (rpcrdma_args_inline(r_xprt
, rqst
)) {
788 rtype
= rpcrdma_noch
;
789 } else if (ddp_allowed
&& rqst
->rq_snd_buf
.flags
& XDRBUF_WRITE
) {
791 rtype
= rpcrdma_readch
;
793 r_xprt
->rx_stats
.nomsg_call_count
++;
795 rtype
= rpcrdma_areadch
;
798 /* If this is a retransmit, discard previously registered
799 * chunks. Very likely the connection has been replaced,
800 * so these registrations are invalid and unusable.
802 while (unlikely(!list_empty(&req
->rl_registered
))) {
803 struct rpcrdma_mr
*mr
;
805 mr
= rpcrdma_mr_pop(&req
->rl_registered
);
806 rpcrdma_mr_recycle(mr
);
809 /* This implementation supports the following combinations
810 * of chunk lists in one RPC-over-RDMA Call message:
815 * - Read list + Reply chunk
817 * It might not yet support the following combinations:
819 * - Read list + Write list
821 * It does not support the following combinations:
823 * - Write list + Reply chunk
824 * - Read list + Write list + Reply chunk
826 * This implementation supports only a single chunk in each
827 * Read or Write list. Thus for example the client cannot
828 * send a Call message with a Position Zero Read chunk and a
829 * regular Read chunk at the same time.
831 if (rtype
!= rpcrdma_noch
) {
832 ret
= rpcrdma_encode_read_list(r_xprt
, req
, rqst
, rtype
);
836 ret
= encode_item_not_present(xdr
);
840 if (wtype
== rpcrdma_writech
) {
841 ret
= rpcrdma_encode_write_list(r_xprt
, req
, rqst
, wtype
);
845 ret
= encode_item_not_present(xdr
);
849 if (wtype
!= rpcrdma_replych
)
850 ret
= encode_item_not_present(xdr
);
852 ret
= rpcrdma_encode_reply_chunk(r_xprt
, req
, rqst
, wtype
);
856 trace_xprtrdma_marshal(rqst
, xdr_stream_pos(xdr
), rtype
, wtype
);
858 ret
= rpcrdma_prepare_send_sges(r_xprt
, req
, xdr_stream_pos(xdr
),
859 &rqst
->rq_snd_buf
, rtype
);
867 xprt_wait_for_buffer_space(rqst
->rq_xprt
);
872 r_xprt
->rx_stats
.failed_marshal_count
++;
878 * rpcrdma_inline_fixup - Scatter inline received data into rqst's iovecs
879 * @rqst: controlling RPC request
880 * @srcp: points to RPC message payload in receive buffer
881 * @copy_len: remaining length of receive buffer content
882 * @pad: Write chunk pad bytes needed (zero for pure inline)
884 * The upper layer has set the maximum number of bytes it can
885 * receive in each component of rq_rcv_buf. These values are set in
886 * the head.iov_len, page_len, tail.iov_len, and buflen fields.
888 * Unlike the TCP equivalent (xdr_partial_copy_from_skb), in
889 * many cases this function simply updates iov_base pointers in
890 * rq_rcv_buf to point directly to the received reply data, to
891 * avoid copying reply data.
893 * Returns the count of bytes which had to be memcopied.
896 rpcrdma_inline_fixup(struct rpc_rqst
*rqst
, char *srcp
, int copy_len
, int pad
)
898 unsigned long fixup_copy_count
;
899 int i
, npages
, curlen
;
901 struct page
**ppages
;
904 /* The head iovec is redirected to the RPC reply message
905 * in the receive buffer, to avoid a memcopy.
907 rqst
->rq_rcv_buf
.head
[0].iov_base
= srcp
;
908 rqst
->rq_private_buf
.head
[0].iov_base
= srcp
;
910 /* The contents of the receive buffer that follow
911 * head.iov_len bytes are copied into the page list.
913 curlen
= rqst
->rq_rcv_buf
.head
[0].iov_len
;
914 if (curlen
> copy_len
)
916 trace_xprtrdma_fixup(rqst
, copy_len
, curlen
);
920 ppages
= rqst
->rq_rcv_buf
.pages
+
921 (rqst
->rq_rcv_buf
.page_base
>> PAGE_SHIFT
);
922 page_base
= offset_in_page(rqst
->rq_rcv_buf
.page_base
);
923 fixup_copy_count
= 0;
924 if (copy_len
&& rqst
->rq_rcv_buf
.page_len
) {
927 pagelist_len
= rqst
->rq_rcv_buf
.page_len
;
928 if (pagelist_len
> copy_len
)
929 pagelist_len
= copy_len
;
930 npages
= PAGE_ALIGN(page_base
+ pagelist_len
) >> PAGE_SHIFT
;
931 for (i
= 0; i
< npages
; i
++) {
932 curlen
= PAGE_SIZE
- page_base
;
933 if (curlen
> pagelist_len
)
934 curlen
= pagelist_len
;
936 trace_xprtrdma_fixup_pg(rqst
, i
, srcp
,
938 destp
= kmap_atomic(ppages
[i
]);
939 memcpy(destp
+ page_base
, srcp
, curlen
);
940 flush_dcache_page(ppages
[i
]);
941 kunmap_atomic(destp
);
944 fixup_copy_count
+= curlen
;
945 pagelist_len
-= curlen
;
951 /* Implicit padding for the last segment in a Write
952 * chunk is inserted inline at the front of the tail
953 * iovec. The upper layer ignores the content of
954 * the pad. Simply ensure inline content in the tail
955 * that follows the Write chunk is properly aligned.
961 /* The tail iovec is redirected to the remaining data
962 * in the receive buffer, to avoid a memcopy.
964 if (copy_len
|| pad
) {
965 rqst
->rq_rcv_buf
.tail
[0].iov_base
= srcp
;
966 rqst
->rq_private_buf
.tail
[0].iov_base
= srcp
;
969 return fixup_copy_count
;
972 /* By convention, backchannel calls arrive via rdma_msg type
973 * messages, and never populate the chunk lists. This makes
974 * the RPC/RDMA header small and fixed in size, so it is
975 * straightforward to check the RPC header's direction field.
978 rpcrdma_is_bcall(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_rep
*rep
)
979 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
981 struct xdr_stream
*xdr
= &rep
->rr_stream
;
984 if (rep
->rr_proc
!= rdma_msg
)
987 /* Peek at stream contents without advancing. */
988 p
= xdr_inline_decode(xdr
, 0);
991 if (*p
++ != xdr_zero
)
993 if (*p
++ != xdr_zero
)
995 if (*p
++ != xdr_zero
)
999 if (*p
++ != rep
->rr_xid
)
1001 if (*p
!= cpu_to_be32(RPC_CALL
))
1004 /* Now that we are sure this is a backchannel call,
1005 * advance to the RPC header.
1007 p
= xdr_inline_decode(xdr
, 3 * sizeof(*p
));
1011 rpcrdma_bc_receive_call(r_xprt
, rep
);
1015 pr_warn("RPC/RDMA short backward direction call\n");
1018 #else /* CONFIG_SUNRPC_BACKCHANNEL */
1022 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
1024 static int decode_rdma_segment(struct xdr_stream
*xdr
, u32
*length
)
1030 p
= xdr_inline_decode(xdr
, 4 * sizeof(*p
));
1034 handle
= be32_to_cpup(p
++);
1035 *length
= be32_to_cpup(p
++);
1036 xdr_decode_hyper(p
, &offset
);
1038 trace_xprtrdma_decode_seg(handle
, *length
, offset
);
1042 static int decode_write_chunk(struct xdr_stream
*xdr
, u32
*length
)
1044 u32 segcount
, seglength
;
1047 p
= xdr_inline_decode(xdr
, sizeof(*p
));
1052 segcount
= be32_to_cpup(p
);
1053 while (segcount
--) {
1054 if (decode_rdma_segment(xdr
, &seglength
))
1056 *length
+= seglength
;
1062 /* In RPC-over-RDMA Version One replies, a Read list is never
1063 * expected. This decoder is a stub that returns an error if
1064 * a Read list is present.
1066 static int decode_read_list(struct xdr_stream
*xdr
)
1070 p
= xdr_inline_decode(xdr
, sizeof(*p
));
1073 if (unlikely(*p
!= xdr_zero
))
1078 /* Supports only one Write chunk in the Write list
1080 static int decode_write_list(struct xdr_stream
*xdr
, u32
*length
)
1089 p
= xdr_inline_decode(xdr
, sizeof(*p
));
1097 if (decode_write_chunk(xdr
, &chunklen
))
1099 *length
+= chunklen
;
1105 static int decode_reply_chunk(struct xdr_stream
*xdr
, u32
*length
)
1109 p
= xdr_inline_decode(xdr
, sizeof(*p
));
1115 if (decode_write_chunk(xdr
, length
))
1121 rpcrdma_decode_msg(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_rep
*rep
,
1122 struct rpc_rqst
*rqst
)
1124 struct xdr_stream
*xdr
= &rep
->rr_stream
;
1125 u32 writelist
, replychunk
, rpclen
;
1128 /* Decode the chunk lists */
1129 if (decode_read_list(xdr
))
1131 if (decode_write_list(xdr
, &writelist
))
1133 if (decode_reply_chunk(xdr
, &replychunk
))
1136 /* RDMA_MSG sanity checks */
1137 if (unlikely(replychunk
))
1140 /* Build the RPC reply's Payload stream in rqst->rq_rcv_buf */
1141 base
= (char *)xdr_inline_decode(xdr
, 0);
1142 rpclen
= xdr_stream_remaining(xdr
);
1143 r_xprt
->rx_stats
.fixup_copy_count
+=
1144 rpcrdma_inline_fixup(rqst
, base
, rpclen
, writelist
& 3);
1146 r_xprt
->rx_stats
.total_rdma_reply
+= writelist
;
1147 return rpclen
+ xdr_align_size(writelist
);
1151 rpcrdma_decode_nomsg(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_rep
*rep
)
1153 struct xdr_stream
*xdr
= &rep
->rr_stream
;
1154 u32 writelist
, replychunk
;
1156 /* Decode the chunk lists */
1157 if (decode_read_list(xdr
))
1159 if (decode_write_list(xdr
, &writelist
))
1161 if (decode_reply_chunk(xdr
, &replychunk
))
1164 /* RDMA_NOMSG sanity checks */
1165 if (unlikely(writelist
))
1167 if (unlikely(!replychunk
))
1170 /* Reply chunk buffer already is the reply vector */
1171 r_xprt
->rx_stats
.total_rdma_reply
+= replychunk
;
1176 rpcrdma_decode_error(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_rep
*rep
,
1177 struct rpc_rqst
*rqst
)
1179 struct xdr_stream
*xdr
= &rep
->rr_stream
;
1182 p
= xdr_inline_decode(xdr
, sizeof(*p
));
1188 p
= xdr_inline_decode(xdr
, 2 * sizeof(*p
));
1191 dprintk("RPC: %5u: %s: server reports version error (%u-%u)\n",
1192 rqst
->rq_task
->tk_pid
, __func__
,
1193 be32_to_cpup(p
), be32_to_cpu(*(p
+ 1)));
1196 dprintk("RPC: %5u: %s: server reports header decoding error\n",
1197 rqst
->rq_task
->tk_pid
, __func__
);
1200 dprintk("RPC: %5u: %s: server reports unrecognized error %d\n",
1201 rqst
->rq_task
->tk_pid
, __func__
, be32_to_cpup(p
));
1204 r_xprt
->rx_stats
.bad_reply_count
++;
1208 /* Perform XID lookup, reconstruction of the RPC reply, and
1209 * RPC completion while holding the transport lock to ensure
1210 * the rep, rqst, and rq_task pointers remain stable.
1212 void rpcrdma_complete_rqst(struct rpcrdma_rep
*rep
)
1214 struct rpcrdma_xprt
*r_xprt
= rep
->rr_rxprt
;
1215 struct rpc_xprt
*xprt
= &r_xprt
->rx_xprt
;
1216 struct rpc_rqst
*rqst
= rep
->rr_rqst
;
1219 xprt
->reestablish_timeout
= 0;
1221 switch (rep
->rr_proc
) {
1223 status
= rpcrdma_decode_msg(r_xprt
, rep
, rqst
);
1226 status
= rpcrdma_decode_nomsg(r_xprt
, rep
);
1229 status
= rpcrdma_decode_error(r_xprt
, rep
, rqst
);
1238 spin_lock(&xprt
->queue_lock
);
1239 xprt_complete_rqst(rqst
->rq_task
, status
);
1240 xprt_unpin_rqst(rqst
);
1241 spin_unlock(&xprt
->queue_lock
);
1244 /* If the incoming reply terminated a pending RPC, the next
1245 * RPC call will post a replacement receive buffer as it is
1249 trace_xprtrdma_reply_hdr(rep
);
1250 r_xprt
->rx_stats
.bad_reply_count
++;
1255 void rpcrdma_release_rqst(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
)
1257 /* Invalidate and unmap the data payloads before waking
1258 * the waiting application. This guarantees the memory
1259 * regions are properly fenced from the server before the
1260 * application accesses the data. It also ensures proper
1261 * send flow control: waking the next RPC waits until this
1262 * RPC has relinquished all its Send Queue entries.
1264 if (!list_empty(&req
->rl_registered
))
1265 r_xprt
->rx_ia
.ri_ops
->ro_unmap_sync(r_xprt
,
1266 &req
->rl_registered
);
1268 /* Ensure that any DMA mapped pages associated with
1269 * the Send of the RPC Call have been unmapped before
1270 * allowing the RPC to complete. This protects argument
1271 * memory not controlled by the RPC client from being
1272 * re-used before we're done with it.
1274 if (test_bit(RPCRDMA_REQ_F_TX_RESOURCES
, &req
->rl_flags
)) {
1275 r_xprt
->rx_stats
.reply_waits_for_send
++;
1276 out_of_line_wait_on_bit(&req
->rl_flags
,
1277 RPCRDMA_REQ_F_TX_RESOURCES
,
1279 TASK_UNINTERRUPTIBLE
);
1283 /* Reply handling runs in the poll worker thread. Anything that
1284 * might wait is deferred to a separate workqueue.
1286 void rpcrdma_deferred_completion(struct work_struct
*work
)
1288 struct rpcrdma_rep
*rep
=
1289 container_of(work
, struct rpcrdma_rep
, rr_work
);
1290 struct rpcrdma_req
*req
= rpcr_to_rdmar(rep
->rr_rqst
);
1291 struct rpcrdma_xprt
*r_xprt
= rep
->rr_rxprt
;
1293 trace_xprtrdma_defer_cmp(rep
);
1294 if (rep
->rr_wc_flags
& IB_WC_WITH_INVALIDATE
)
1295 r_xprt
->rx_ia
.ri_ops
->ro_reminv(rep
, &req
->rl_registered
);
1296 rpcrdma_release_rqst(r_xprt
, req
);
1297 rpcrdma_complete_rqst(rep
);
1300 /* Process received RPC/RDMA messages.
1302 * Errors must result in the RPC task either being awakened, or
1303 * allowed to timeout, to discover the errors at that time.
1305 void rpcrdma_reply_handler(struct rpcrdma_rep
*rep
)
1307 struct rpcrdma_xprt
*r_xprt
= rep
->rr_rxprt
;
1308 struct rpc_xprt
*xprt
= &r_xprt
->rx_xprt
;
1309 struct rpcrdma_buffer
*buf
= &r_xprt
->rx_buf
;
1310 struct rpcrdma_req
*req
;
1311 struct rpc_rqst
*rqst
;
1315 /* Fixed transport header fields */
1316 xdr_init_decode(&rep
->rr_stream
, &rep
->rr_hdrbuf
,
1317 rep
->rr_hdrbuf
.head
[0].iov_base
);
1318 p
= xdr_inline_decode(&rep
->rr_stream
, 4 * sizeof(*p
));
1320 goto out_shortreply
;
1322 rep
->rr_vers
= *p
++;
1323 credits
= be32_to_cpu(*p
++);
1324 rep
->rr_proc
= *p
++;
1326 if (rep
->rr_vers
!= rpcrdma_version
)
1327 goto out_badversion
;
1329 if (rpcrdma_is_bcall(r_xprt
, rep
))
1332 /* Match incoming rpcrdma_rep to an rpcrdma_req to
1333 * get context for handling any incoming chunks.
1335 spin_lock(&xprt
->queue_lock
);
1336 rqst
= xprt_lookup_rqst(xprt
, rep
->rr_xid
);
1339 xprt_pin_rqst(rqst
);
1340 spin_unlock(&xprt
->queue_lock
);
1343 credits
= 1; /* don't deadlock */
1344 else if (credits
> buf
->rb_max_requests
)
1345 credits
= buf
->rb_max_requests
;
1346 if (buf
->rb_credits
!= credits
) {
1347 spin_lock_bh(&xprt
->transport_lock
);
1348 buf
->rb_credits
= credits
;
1349 xprt
->cwnd
= credits
<< RPC_CWNDSHIFT
;
1350 spin_unlock_bh(&xprt
->transport_lock
);
1353 req
= rpcr_to_rdmar(rqst
);
1354 req
->rl_reply
= rep
;
1355 rep
->rr_rqst
= rqst
;
1356 clear_bit(RPCRDMA_REQ_F_PENDING
, &req
->rl_flags
);
1358 trace_xprtrdma_reply(rqst
->rq_task
, rep
, req
, credits
);
1359 queue_work(buf
->rb_completion_wq
, &rep
->rr_work
);
1363 trace_xprtrdma_reply_vers(rep
);
1367 spin_unlock(&xprt
->queue_lock
);
1368 trace_xprtrdma_reply_rqst(rep
);
1372 trace_xprtrdma_reply_short(rep
);
1375 rpcrdma_recv_buffer_put(rep
);