2 * Copyright (c) 2005-2006 Network Appliance, Inc. All rights reserved.
4 * This software is available to you under a choice of one of two
5 * licenses. You may choose to be licensed under the terms of the GNU
6 * General Public License (GPL) Version 2, available from the file
7 * COPYING in the main directory of this source tree, or the BSD-type
10 * Redistribution and use in source and binary forms, with or without
11 * modification, are permitted provided that the following conditions
14 * Redistributions of source code must retain the above copyright
15 * notice, this list of conditions and the following disclaimer.
17 * Redistributions in binary form must reproduce the above
18 * copyright notice, this list of conditions and the following
19 * disclaimer in the documentation and/or other materials provided
20 * with the distribution.
22 * Neither the name of the Network Appliance, Inc. nor the names of
23 * its contributors may be used to endorse or promote products
24 * derived from this software without specific prior written
27 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
28 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
29 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
30 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
31 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
32 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
33 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
34 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
35 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
36 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
37 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
39 * Author: Tom Tucker <tom@opengridcomputing.com>
42 #include <linux/sunrpc/debug.h>
43 #include <linux/sunrpc/rpc_rdma.h>
44 #include <linux/spinlock.h>
45 #include <asm/unaligned.h>
46 #include <rdma/ib_verbs.h>
47 #include <rdma/rdma_cm.h>
48 #include <linux/sunrpc/svc_rdma.h>
50 #define RPCDBG_FACILITY RPCDBG_SVCXPRT
53 * Replace the pages in the rq_argpages array with the pages from the SGE in
54 * the RDMA_RECV completion. The SGL should contain full pages up until the
57 static void rdma_build_arg_xdr(struct svc_rqst
*rqstp
,
58 struct svc_rdma_op_ctxt
*ctxt
,
65 /* Swap the page in the SGE with the page in argpages */
66 page
= ctxt
->pages
[0];
67 put_page(rqstp
->rq_pages
[0]);
68 rqstp
->rq_pages
[0] = page
;
70 /* Set up the XDR head */
71 rqstp
->rq_arg
.head
[0].iov_base
= page_address(page
);
72 rqstp
->rq_arg
.head
[0].iov_len
= min(byte_count
, ctxt
->sge
[0].length
);
73 rqstp
->rq_arg
.len
= byte_count
;
74 rqstp
->rq_arg
.buflen
= byte_count
;
76 /* Compute bytes past head in the SGL */
77 bc
= byte_count
- rqstp
->rq_arg
.head
[0].iov_len
;
79 /* If data remains, store it in the pagelist */
80 rqstp
->rq_arg
.page_len
= bc
;
81 rqstp
->rq_arg
.page_base
= 0;
82 rqstp
->rq_arg
.pages
= &rqstp
->rq_pages
[1];
84 while (bc
&& sge_no
< ctxt
->count
) {
85 page
= ctxt
->pages
[sge_no
];
86 put_page(rqstp
->rq_pages
[sge_no
]);
87 rqstp
->rq_pages
[sge_no
] = page
;
88 bc
-= min(bc
, ctxt
->sge
[sge_no
].length
);
89 rqstp
->rq_arg
.buflen
+= ctxt
->sge
[sge_no
].length
;
92 rqstp
->rq_respages
= &rqstp
->rq_pages
[sge_no
];
94 /* We should never run out of SGE because the limit is defined to
95 * support the max allowed RPC data length
97 BUG_ON(bc
&& (sge_no
== ctxt
->count
));
98 BUG_ON((rqstp
->rq_arg
.head
[0].iov_len
+ rqstp
->rq_arg
.page_len
)
100 BUG_ON(rqstp
->rq_arg
.len
!= byte_count
);
102 /* If not all pages were used from the SGL, free the remaining ones */
104 while (sge_no
< ctxt
->count
) {
105 page
= ctxt
->pages
[sge_no
++];
111 rqstp
->rq_arg
.tail
[0].iov_base
= NULL
;
112 rqstp
->rq_arg
.tail
[0].iov_len
= 0;
116 int start
; /* sge no for this chunk */
117 int count
; /* sge count for this chunk */
120 /* Encode a read-chunk-list as an array of IB SGE
123 * - chunk[0]->position points to pages[0] at an offset of 0
124 * - pages[] is not physically or virtually contigous and consists of
125 * PAGE_SIZE elements.
128 * - sge array pointing into pages[] array.
129 * - chunk_sge array specifying sge index and count for each
130 * chunk in the read list
133 static int rdma_rcl_to_sge(struct svcxprt_rdma
*xprt
,
134 struct svc_rqst
*rqstp
,
135 struct svc_rdma_op_ctxt
*head
,
136 struct rpcrdma_msg
*rmsgp
,
138 struct chunk_sge
*ch_sge_ary
,
148 struct rpcrdma_read_chunk
*ch
;
153 ch
= (struct rpcrdma_read_chunk
*)&rmsgp
->rm_body
.rm_chunks
[0];
155 ch_bytes
= ch
->rc_target
.rs_length
;
156 head
->arg
.head
[0] = rqstp
->rq_arg
.head
[0];
157 head
->arg
.tail
[0] = rqstp
->rq_arg
.tail
[0];
158 head
->arg
.pages
= &head
->pages
[head
->count
];
159 head
->sge
[0].length
= head
->count
; /* save count of hdr pages */
160 head
->arg
.page_base
= 0;
161 head
->arg
.page_len
= ch_bytes
;
162 head
->arg
.len
= rqstp
->rq_arg
.len
+ ch_bytes
;
163 head
->arg
.buflen
= rqstp
->rq_arg
.buflen
+ ch_bytes
;
165 ch_sge_ary
[0].start
= 0;
167 sge_bytes
= min_t(int, PAGE_SIZE
-page_off
, ch_bytes
);
169 ib_dma_map_page(xprt
->sc_cm_id
->device
,
170 rqstp
->rq_arg
.pages
[page_no
],
173 sge
[sge_no
].length
= sge_bytes
;
174 sge
[sge_no
].lkey
= xprt
->sc_phys_mr
->lkey
;
176 * Don't bump head->count here because the same page
177 * may be used by multiple SGE.
179 head
->arg
.pages
[page_no
] = rqstp
->rq_arg
.pages
[page_no
];
180 rqstp
->rq_respages
= &rqstp
->rq_arg
.pages
[page_no
+1];
182 byte_count
-= sge_bytes
;
183 ch_bytes
-= sge_bytes
;
186 * If all bytes for this chunk have been mapped to an
187 * SGE, move to the next SGE
190 ch_sge_ary
[ch_no
].count
=
191 sge_no
- ch_sge_ary
[ch_no
].start
;
194 ch_sge_ary
[ch_no
].start
= sge_no
;
195 ch_bytes
= ch
->rc_target
.rs_length
;
196 /* If bytes remaining account for next chunk */
198 head
->arg
.page_len
+= ch_bytes
;
199 head
->arg
.len
+= ch_bytes
;
200 head
->arg
.buflen
+= ch_bytes
;
204 * If this SGE consumed all of the page, move to the
207 if ((sge_bytes
+ page_off
) == PAGE_SIZE
) {
211 * If there are still bytes left to map, bump
217 page_off
+= sge_bytes
;
219 BUG_ON(byte_count
!= 0);
223 static void rdma_set_ctxt_sge(struct svc_rdma_op_ctxt
*ctxt
,
231 for (i
= 0; i
< count
; i
++) {
232 ctxt
->sge
[i
].addr
= sge
[i
].addr
;
233 ctxt
->sge
[i
].length
= sge
[i
].length
;
234 *sgl_offset
= *sgl_offset
+ sge
[i
].length
;
238 static int rdma_read_max_sge(struct svcxprt_rdma
*xprt
, int sge_count
)
240 if ((RDMA_TRANSPORT_IWARP
==
241 rdma_node_get_transport(xprt
->sc_cm_id
->
246 return min_t(int, sge_count
, xprt
->sc_max_sge
);
250 * Use RDMA_READ to read data from the advertised client buffer into the
251 * XDR stream starting at rq_arg.head[0].iov_base.
252 * Each chunk in the array
253 * contains the following fields:
254 * discrim - '1', This isn't used for data placement
255 * position - The xdr stream offset (the same for every chunk)
256 * handle - RMR for client memory region
257 * length - data transfer length
258 * offset - 64 bit tagged offset in remote memory region
260 * On our side, we need to read into a pagelist. The first page immediately
261 * follows the RPC header.
263 * This function returns 1 to indicate success. The data is not yet in
264 * the pagelist and therefore the RPC request must be deferred. The
265 * I/O completion will enqueue the transport again and
266 * svc_rdma_recvfrom will complete the request.
268 * NOTE: The ctxt must not be touched after the last WR has been posted
269 * because the I/O completion processing may occur on another
270 * processor and free / modify the context. Ne touche pas!
272 static int rdma_read_xdr(struct svcxprt_rdma
*xprt
,
273 struct rpcrdma_msg
*rmsgp
,
274 struct svc_rqst
*rqstp
,
275 struct svc_rdma_op_ctxt
*hdr_ctxt
)
277 struct ib_send_wr read_wr
;
285 struct rpcrdma_read_chunk
*ch
;
286 struct svc_rdma_op_ctxt
*ctxt
= NULL
;
287 struct svc_rdma_op_ctxt
*head
;
288 struct svc_rdma_op_ctxt
*tmp_sge_ctxt
;
289 struct svc_rdma_op_ctxt
*tmp_ch_ctxt
;
290 struct chunk_sge
*ch_sge_ary
;
292 /* If no read list is present, return 0 */
293 ch
= svc_rdma_get_read_chunk(rmsgp
);
297 /* Allocate temporary contexts to keep SGE */
298 BUG_ON(sizeof(struct ib_sge
) < sizeof(struct chunk_sge
));
299 tmp_sge_ctxt
= svc_rdma_get_context(xprt
);
300 sge
= tmp_sge_ctxt
->sge
;
301 tmp_ch_ctxt
= svc_rdma_get_context(xprt
);
302 ch_sge_ary
= (struct chunk_sge
*)tmp_ch_ctxt
->sge
;
304 svc_rdma_rcl_chunk_counts(ch
, &ch_count
, &byte_count
);
305 sge_count
= rdma_rcl_to_sge(xprt
, rqstp
, hdr_ctxt
, rmsgp
,
307 ch_count
, byte_count
);
308 head
= svc_rdma_get_context(xprt
);
312 for (ch
= (struct rpcrdma_read_chunk
*)&rmsgp
->rm_body
.rm_chunks
[0];
313 ch
->rc_discrim
!= 0; ch
++, ch_no
++) {
318 ctxt
->next
= svc_rdma_get_context(xprt
);
322 ctxt
->direction
= DMA_FROM_DEVICE
;
323 clear_bit(RDMACTXT_F_READ_DONE
, &ctxt
->flags
);
324 clear_bit(RDMACTXT_F_LAST_CTXT
, &ctxt
->flags
);
325 if ((ch
+1)->rc_discrim
== 0) {
327 * Checked in sq_cq_reap to see if we need to
330 set_bit(RDMACTXT_F_LAST_CTXT
, &ctxt
->flags
);
331 ctxt
->next
= hdr_ctxt
;
332 hdr_ctxt
->next
= head
;
335 /* Prepare READ WR */
336 memset(&read_wr
, 0, sizeof read_wr
);
337 ctxt
->wr_op
= IB_WR_RDMA_READ
;
338 read_wr
.wr_id
= (unsigned long)ctxt
;
339 read_wr
.opcode
= IB_WR_RDMA_READ
;
340 read_wr
.send_flags
= IB_SEND_SIGNALED
;
341 read_wr
.wr
.rdma
.rkey
= ch
->rc_target
.rs_handle
;
342 read_wr
.wr
.rdma
.remote_addr
=
343 get_unaligned(&(ch
->rc_target
.rs_offset
)) +
345 read_wr
.sg_list
= &sge
[ch_sge_ary
[ch_no
].start
];
347 rdma_read_max_sge(xprt
, ch_sge_ary
[ch_no
].count
);
348 rdma_set_ctxt_sge(ctxt
, &sge
[ch_sge_ary
[ch_no
].start
],
353 err
= svc_rdma_send(xprt
, &read_wr
);
355 printk(KERN_ERR
"svcrdma: Error posting send = %d\n",
358 * Break the circular list so free knows when
359 * to stop if the error happened to occur on
365 atomic_inc(&rdma_stat_read
);
367 if (read_wr
.num_sge
< ch_sge_ary
[ch_no
].count
) {
368 ch_sge_ary
[ch_no
].count
-= read_wr
.num_sge
;
369 ch_sge_ary
[ch_no
].start
+= read_wr
.num_sge
;
377 svc_rdma_put_context(tmp_sge_ctxt
, 0);
378 svc_rdma_put_context(tmp_ch_ctxt
, 0);
380 /* Detach arg pages. svc_recv will replenish them */
381 for (ch_no
= 0; &rqstp
->rq_pages
[ch_no
] < rqstp
->rq_respages
; ch_no
++)
382 rqstp
->rq_pages
[ch_no
] = NULL
;
385 * Detach res pages. svc_release must see a resused count of
386 * zero or it will attempt to put them.
388 while (rqstp
->rq_resused
)
389 rqstp
->rq_respages
[--rqstp
->rq_resused
] = NULL
;
392 printk(KERN_ERR
"svcrdma : RDMA_READ error = %d\n", err
);
393 set_bit(XPT_CLOSE
, &xprt
->sc_xprt
.xpt_flags
);
394 /* Free the linked list of read contexts */
395 while (head
!= NULL
) {
397 svc_rdma_put_context(head
, 1);
406 static int rdma_read_complete(struct svc_rqst
*rqstp
,
407 struct svc_rdma_op_ctxt
*data
)
409 struct svc_rdma_op_ctxt
*head
= data
->next
;
416 for (page_no
= 0; page_no
< head
->count
; page_no
++) {
417 put_page(rqstp
->rq_pages
[page_no
]);
418 rqstp
->rq_pages
[page_no
] = head
->pages
[page_no
];
420 /* Point rq_arg.pages past header */
421 rqstp
->rq_arg
.pages
= &rqstp
->rq_pages
[head
->sge
[0].length
];
422 rqstp
->rq_arg
.page_len
= head
->arg
.page_len
;
423 rqstp
->rq_arg
.page_base
= head
->arg
.page_base
;
425 /* rq_respages starts after the last arg page */
426 rqstp
->rq_respages
= &rqstp
->rq_arg
.pages
[page_no
];
427 rqstp
->rq_resused
= 0;
429 /* Rebuild rq_arg head and tail. */
430 rqstp
->rq_arg
.head
[0] = head
->arg
.head
[0];
431 rqstp
->rq_arg
.tail
[0] = head
->arg
.tail
[0];
432 rqstp
->rq_arg
.len
= head
->arg
.len
;
433 rqstp
->rq_arg
.buflen
= head
->arg
.buflen
;
435 /* XXX: What should this be? */
436 rqstp
->rq_prot
= IPPROTO_MAX
;
439 * Free the contexts we used to build the RDMA_READ. We have
440 * to be careful here because the context list uses the same
441 * next pointer used to chain the contexts associated with the
444 data
->next
= NULL
; /* terminate circular list */
447 svc_rdma_put_context(head
, 0);
449 } while (head
!= NULL
);
451 ret
= rqstp
->rq_arg
.head
[0].iov_len
452 + rqstp
->rq_arg
.page_len
453 + rqstp
->rq_arg
.tail
[0].iov_len
;
454 dprintk("svcrdma: deferred read ret=%d, rq_arg.len =%d, "
455 "rq_arg.head[0].iov_base=%p, rq_arg.head[0].iov_len = %zd\n",
456 ret
, rqstp
->rq_arg
.len
, rqstp
->rq_arg
.head
[0].iov_base
,
457 rqstp
->rq_arg
.head
[0].iov_len
);
459 /* Indicate that we've consumed an RQ credit */
460 rqstp
->rq_xprt_ctxt
= rqstp
->rq_xprt
;
461 svc_xprt_received(rqstp
->rq_xprt
);
466 * Set up the rqstp thread context to point to the RQ buffer. If
467 * necessary, pull additional data from the client with an RDMA_READ
470 int svc_rdma_recvfrom(struct svc_rqst
*rqstp
)
472 struct svc_xprt
*xprt
= rqstp
->rq_xprt
;
473 struct svcxprt_rdma
*rdma_xprt
=
474 container_of(xprt
, struct svcxprt_rdma
, sc_xprt
);
475 struct svc_rdma_op_ctxt
*ctxt
= NULL
;
476 struct rpcrdma_msg
*rmsgp
;
480 dprintk("svcrdma: rqstp=%p\n", rqstp
);
483 * The rq_xprt_ctxt indicates if we've consumed an RQ credit
484 * or not. It is used in the rdma xpo_release_rqst function to
485 * determine whether or not to return an RQ WQE to the RQ.
487 rqstp
->rq_xprt_ctxt
= NULL
;
489 spin_lock_bh(&rdma_xprt
->sc_read_complete_lock
);
490 if (!list_empty(&rdma_xprt
->sc_read_complete_q
)) {
491 ctxt
= list_entry(rdma_xprt
->sc_read_complete_q
.next
,
492 struct svc_rdma_op_ctxt
,
494 list_del_init(&ctxt
->dto_q
);
496 spin_unlock_bh(&rdma_xprt
->sc_read_complete_lock
);
498 return rdma_read_complete(rqstp
, ctxt
);
500 spin_lock_bh(&rdma_xprt
->sc_rq_dto_lock
);
501 if (!list_empty(&rdma_xprt
->sc_rq_dto_q
)) {
502 ctxt
= list_entry(rdma_xprt
->sc_rq_dto_q
.next
,
503 struct svc_rdma_op_ctxt
,
505 list_del_init(&ctxt
->dto_q
);
507 atomic_inc(&rdma_stat_rq_starve
);
508 clear_bit(XPT_DATA
, &xprt
->xpt_flags
);
511 spin_unlock_bh(&rdma_xprt
->sc_rq_dto_lock
);
513 /* This is the EAGAIN path. The svc_recv routine will
514 * return -EAGAIN, the nfsd thread will go to call into
515 * svc_recv again and we shouldn't be on the active
518 if (test_bit(XPT_CLOSE
, &xprt
->xpt_flags
))
524 dprintk("svcrdma: processing ctxt=%p on xprt=%p, rqstp=%p, status=%d\n",
525 ctxt
, rdma_xprt
, rqstp
, ctxt
->wc_status
);
526 BUG_ON(ctxt
->wc_status
!= IB_WC_SUCCESS
);
527 atomic_inc(&rdma_stat_recv
);
529 /* Build up the XDR from the receive buffers. */
530 rdma_build_arg_xdr(rqstp
, ctxt
, ctxt
->byte_len
);
532 /* Decode the RDMA header. */
533 len
= svc_rdma_xdr_decode_req(&rmsgp
, rqstp
);
534 rqstp
->rq_xprt_hlen
= len
;
536 /* If the request is invalid, reply with an error */
539 (void)svc_rdma_send_error(rdma_xprt
, rmsgp
, ERR_VERS
);
543 /* Read read-list data. If we would need to wait, defer
544 * it. Not that in this case, we don't return the RQ credit
545 * until after the read completes.
547 if (rdma_read_xdr(rdma_xprt
, rmsgp
, rqstp
, ctxt
)) {
548 svc_xprt_received(xprt
);
552 /* Indicate we've consumed an RQ credit */
553 rqstp
->rq_xprt_ctxt
= rqstp
->rq_xprt
;
555 ret
= rqstp
->rq_arg
.head
[0].iov_len
556 + rqstp
->rq_arg
.page_len
557 + rqstp
->rq_arg
.tail
[0].iov_len
;
558 svc_rdma_put_context(ctxt
, 0);
560 dprintk("svcrdma: ret = %d, rq_arg.len =%d, "
561 "rq_arg.head[0].iov_base=%p, rq_arg.head[0].iov_len = %zd\n",
562 ret
, rqstp
->rq_arg
.len
,
563 rqstp
->rq_arg
.head
[0].iov_base
,
564 rqstp
->rq_arg
.head
[0].iov_len
);
565 rqstp
->rq_prot
= IPPROTO_MAX
;
566 svc_xprt_copy_addrs(rqstp
, xprt
);
567 svc_xprt_received(xprt
);
572 svc_rdma_put_context(ctxt
, 1);
573 /* Indicate we've consumed an RQ credit */
574 rqstp
->rq_xprt_ctxt
= rqstp
->rq_xprt
;
576 dprintk("svcrdma: transport %p is closing\n", xprt
);
578 * Set the close bit and enqueue it. svc_recv will see the
579 * close bit and call svc_xprt_delete
581 set_bit(XPT_CLOSE
, &xprt
->xpt_flags
);
582 svc_xprt_received(xprt
);
This page took 0.042337 seconds and 5 git commands to generate.