Lines Matching +full:no +full:- +full:memory +full:- +full:wc
1 // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
3 * Copyright (c) 2016-2018 Oracle. All rights reserved.
5 * Copyright (c) 2005-2006 Network Appliance, Inc. All rights reserved.
10 * COPYING in the main directory of this source tree, or the BSD-type
33 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
109 static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc);
121 int node = ibdev_to_node(rdma->sc_cm_id->device);
129 buffer = kmalloc_node(rdma->sc_max_req_size, GFP_KERNEL, node);
132 addr = ib_dma_map_single(rdma->sc_pd->device, buffer,
133 rdma->sc_max_req_size, DMA_FROM_DEVICE);
134 if (ib_dma_mapping_error(rdma->sc_pd->device, addr))
137 svc_rdma_recv_cid_init(rdma, &ctxt->rc_cid);
138 pcl_init(&ctxt->rc_call_pcl);
139 pcl_init(&ctxt->rc_read_pcl);
140 pcl_init(&ctxt->rc_write_pcl);
141 pcl_init(&ctxt->rc_reply_pcl);
143 ctxt->rc_recv_wr.next = NULL;
144 ctxt->rc_recv_wr.wr_cqe = &ctxt->rc_cqe;
145 ctxt->rc_recv_wr.sg_list = &ctxt->rc_recv_sge;
146 ctxt->rc_recv_wr.num_sge = 1;
147 ctxt->rc_cqe.done = svc_rdma_wc_receive;
148 ctxt->rc_recv_sge.addr = addr;
149 ctxt->rc_recv_sge.length = rdma->sc_max_req_size;
150 ctxt->rc_recv_sge.lkey = rdma->sc_pd->local_dma_lkey;
151 ctxt->rc_recv_buf = buffer;
152 svc_rdma_cc_init(rdma, &ctxt->rc_cc);
166 ib_dma_unmap_single(rdma->sc_pd->device, ctxt->rc_recv_sge.addr,
167 ctxt->rc_recv_sge.length, DMA_FROM_DEVICE);
168 kfree(ctxt->rc_recv_buf);
173 * svc_rdma_recv_ctxts_destroy - Release all recv_ctxt's for an xprt
182 while ((node = llist_del_first(&rdma->sc_recv_ctxts))) {
189 * svc_rdma_recv_ctxt_get - Allocate a recv_ctxt
199 node = llist_del_first(&rdma->sc_recv_ctxts);
204 ctxt->rc_page_count = 0;
209 * svc_rdma_recv_ctxt_put - Return recv_ctxt to free list
217 svc_rdma_cc_release(rdma, &ctxt->rc_cc, DMA_FROM_DEVICE);
222 release_pages(ctxt->rc_pages, ctxt->rc_page_count);
224 pcl_free(&ctxt->rc_call_pcl);
225 pcl_free(&ctxt->rc_read_pcl);
226 pcl_free(&ctxt->rc_write_pcl);
227 pcl_free(&ctxt->rc_reply_pcl);
229 llist_add(&ctxt->rc_node, &rdma->sc_recv_ctxts);
233 * svc_rdma_release_ctxt - Release transport-specific per-rqst resources
235 * @vctxt: the context from rqstp->rq_xprt_ctxt or dr->xprt_ctxt
259 if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags))
263 while (wanted--) {
268 trace_svcrdma_post_recv(&ctxt->rc_cid);
269 ctxt->rc_recv_wr.next = recv_chain;
270 recv_chain = &ctxt->rc_recv_wr;
271 rdma->sc_pending_recvs++;
276 ret = ib_post_recv(rdma->sc_qp, recv_chain, &bad_wr);
286 bad_wr = bad_wr->next;
289 /* Since we're destroying the xprt, no need to reset
295 * svc_rdma_post_recvs - Post initial set of Recv WRs
300 * %false: memory allocation or DMA error
309 total = (rdma->sc_max_requests * 2) + rdma->sc_recv_batch;
310 while (total--) {
316 llist_add(&ctxt->rc_node, &rdma->sc_recv_ctxts);
319 return svc_rdma_refresh_recvs(rdma, rdma->sc_max_requests);
323 * svc_rdma_wc_receive - Invoked by RDMA provider for each polled Receive WC
325 * @wc: Work Completion object
328 static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
330 struct svcxprt_rdma *rdma = cq->cq_context;
331 struct ib_cqe *cqe = wc->wr_cqe;
334 rdma->sc_pending_recvs--;
336 /* WARNING: Only wc->wr_cqe and wc->status are reliable */
339 if (wc->status != IB_WC_SUCCESS)
341 trace_svcrdma_wc_recv(wc, &ctxt->rc_cid);
352 if (rdma->sc_pending_recvs < rdma->sc_max_requests)
353 if (!svc_rdma_refresh_recvs(rdma, rdma->sc_recv_batch))
356 /* All wc fields are now known to be valid */
357 ctxt->rc_byte_len = wc->byte_len;
359 spin_lock(&rdma->sc_rq_dto_lock);
360 list_add_tail(&ctxt->rc_list, &rdma->sc_rq_dto_q);
362 set_bit(XPT_DATA, &rdma->sc_xprt.xpt_flags);
363 spin_unlock(&rdma->sc_rq_dto_lock);
364 if (!test_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags))
365 svc_xprt_enqueue(&rdma->sc_xprt);
369 if (wc->status == IB_WC_WR_FLUSH_ERR)
370 trace_svcrdma_wc_recv_flush(wc, &ctxt->rc_cid);
372 trace_svcrdma_wc_recv_err(wc, &ctxt->rc_cid);
375 svc_xprt_deferred_close(&rdma->sc_xprt);
379 * svc_rdma_flush_recv_queues - Drain pending Receive work
387 while ((ctxt = svc_rdma_next_recv_ctxt(&rdma->sc_read_complete_q))) {
388 list_del(&ctxt->rc_list);
391 while ((ctxt = svc_rdma_next_recv_ctxt(&rdma->sc_rq_dto_q))) {
392 list_del(&ctxt->rc_list);
400 struct xdr_buf *arg = &rqstp->rq_arg;
402 arg->head[0].iov_base = ctxt->rc_recv_buf;
403 arg->head[0].iov_len = ctxt->rc_byte_len;
404 arg->tail[0].iov_base = NULL;
405 arg->tail[0].iov_len = 0;
406 arg->page_len = 0;
407 arg->page_base = 0;
408 arg->buflen = ctxt->rc_byte_len;
409 arg->len = ctxt->rc_byte_len;
413 * xdr_count_read_segments - Count number of Read segments in Read list
415 * @p: Start of an un-decoded Read list
434 rctxt->rc_call_pcl.cl_count = 0;
435 rctxt->rc_read_pcl.cl_count = 0;
440 p = xdr_inline_decode(&rctxt->rc_stream,
450 ++rctxt->rc_read_pcl.cl_count;
452 ++rctxt->rc_call_pcl.cl_count;
455 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
465 * - Read list does not overflow Receive buffer.
466 * - Chunk size limited by largest NFS data payload.
478 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
493 if (xdr_stream_decode_u32(&rctxt->rc_stream, &segcount))
503 p = xdr_inline_decode(&rctxt->rc_stream,
509 * xdr_count_write_chunks - Count number of Write chunks in Write list
511 * @p: start of an un-decoded Write list
525 rctxt->rc_write_pcl.cl_count = 0;
529 ++rctxt->rc_write_pcl.cl_count;
530 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
540 * - This implementation currently supports only one Write chunk.
543 * - Write list does not overflow Receive buffer.
544 * - Chunk size limited by largest NFS data payload.
556 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
561 if (!pcl_alloc_write(rctxt, &rctxt->rc_write_pcl, p))
564 rctxt->rc_cur_result_payload = pcl_first_chunk(&rctxt->rc_write_pcl);
571 * - Reply chunk does not overflow Receive buffer.
572 * - Chunk size limited by largest NFS data payload.
584 p = xdr_inline_decode(&rctxt->rc_stream, sizeof(*p));
593 rctxt->rc_reply_pcl.cl_count = 1;
594 return pcl_alloc_write(rctxt, &rctxt->rc_reply_pcl, p);
597 /* RPC-over-RDMA Version One private extension: Remote Invalidation.
611 ctxt->rc_inv_rkey = 0;
613 if (!rdma->sc_snd_w_inv)
617 pcl_for_each_chunk(chunk, &ctxt->rc_call_pcl) {
620 inv_rkey = segment->rs_handle;
621 else if (inv_rkey != segment->rs_handle)
625 pcl_for_each_chunk(chunk, &ctxt->rc_read_pcl) {
628 inv_rkey = segment->rs_handle;
629 else if (inv_rkey != segment->rs_handle)
633 pcl_for_each_chunk(chunk, &ctxt->rc_write_pcl) {
636 inv_rkey = segment->rs_handle;
637 else if (inv_rkey != segment->rs_handle)
641 pcl_for_each_chunk(chunk, &ctxt->rc_reply_pcl) {
644 inv_rkey = segment->rs_handle;
645 else if (inv_rkey != segment->rs_handle)
649 ctxt->rc_inv_rkey = inv_rkey;
653 * svc_rdma_xdr_decode_req - Decode the transport header
657 * On entry, xdr->head[0].iov_base points to first byte of the
658 * RPC-over-RDMA transport header.
661 * RPC-over-RDMA header. For RDMA_MSG, this is the RPC message.
663 * The length of the RPC-over-RDMA header is returned.
666 * - The transport header is entirely contained in the head iovec.
674 rdma_argp = rq_arg->head[0].iov_base;
675 xdr_init_decode(&rctxt->rc_stream, rq_arg, rdma_argp, NULL);
677 p = xdr_inline_decode(&rctxt->rc_stream,
685 rctxt->rc_msgtype = *p;
686 switch (rctxt->rc_msgtype) {
706 rq_arg->head[0].iov_base = rctxt->rc_stream.p;
707 hdr_len = xdr_stream_pos(&rctxt->rc_stream);
708 rq_arg->head[0].iov_len -= hdr_len;
709 rq_arg->len -= hdr_len;
714 trace_svcrdma_decode_short_err(rctxt, rq_arg->len);
715 return -EINVAL;
719 return -EPROTONOSUPPORT;
727 return -EINVAL;
731 return -EINVAL;
754 __be32 *p = rctxt->rc_recv_buf;
756 if (!xprt->xpt_bc_xprt)
759 if (rctxt->rc_msgtype != rdma_msg)
762 if (!pcl_is_empty(&rctxt->rc_call_pcl))
764 if (!pcl_is_empty(&rctxt->rc_read_pcl))
766 if (!pcl_is_empty(&rctxt->rc_write_pcl))
768 if (!pcl_is_empty(&rctxt->rc_reply_pcl))
787 struct svc_rdma_chunk *chunk = pcl_first_chunk(&ctxt->rc_read_pcl);
788 struct xdr_buf *buf = &rqstp->rq_arg;
796 buf->tail[0].iov_base = buf->head[0].iov_base + chunk->ch_position;
797 buf->tail[0].iov_len = buf->head[0].iov_len - chunk->ch_position;
798 buf->head[0].iov_len = chunk->ch_position;
804 * list is increased to include XDR round-up.
807 * thus the rounded-up length never crosses a page boundary.
809 buf->pages = &rqstp->rq_pages[0];
810 length = xdr_align_size(chunk->ch_length);
811 buf->page_len = length;
812 buf->len += length;
813 buf->buflen += length;
819 * with payload in multiple Read chunks and no PZRC.
824 struct xdr_buf *buf = &rqstp->rq_arg;
826 buf->len += ctxt->rc_readbytes;
827 buf->buflen += ctxt->rc_readbytes;
829 buf->head[0].iov_base = page_address(rqstp->rq_pages[0]);
830 buf->head[0].iov_len = min_t(size_t, PAGE_SIZE, ctxt->rc_readbytes);
831 buf->pages = &rqstp->rq_pages[1];
832 buf->page_len = ctxt->rc_readbytes - buf->head[0].iov_len;
843 struct xdr_buf *buf = &rqstp->rq_arg;
845 buf->len += ctxt->rc_readbytes;
846 buf->buflen += ctxt->rc_readbytes;
848 buf->head[0].iov_base = page_address(rqstp->rq_pages[0]);
849 buf->head[0].iov_len = min_t(size_t, PAGE_SIZE, ctxt->rc_readbytes);
850 buf->pages = &rqstp->rq_pages[1];
851 buf->page_len = ctxt->rc_readbytes - buf->head[0].iov_len;
862 release_pages(rqstp->rq_respages, ctxt->rc_page_count);
863 for (i = 0; i < ctxt->rc_page_count; i++)
864 rqstp->rq_pages[i] = ctxt->rc_pages[i];
869 rqstp->rq_respages = &rqstp->rq_pages[ctxt->rc_page_count];
870 rqstp->rq_next_page = rqstp->rq_respages + 1;
875 ctxt->rc_page_count = 0;
881 rqstp->rq_arg = ctxt->rc_saved_arg;
882 if (pcl_is_empty(&ctxt->rc_call_pcl)) {
883 if (ctxt->rc_read_pcl.cl_count == 1)
891 trace_svcrdma_read_finished(&ctxt->rc_cid);
895 * svc_rdma_recvfrom - Receive an RPC call
900 * %0 if there were no Calls ready to return,
901 * %-EINVAL if the Read chunk data is too large,
902 * %-ENOMEM if rdma_rw context pool was exhausted,
903 * %-ENOTCONN if posting failed (connection is lost),
904 * %-EIO if rdma_rw initialization failed (DMA mapping, etc).
907 * when there are no remaining ctxt's to process.
911 * - If the ctxt completes a Receive, then construct the Call
914 * - If there are no Read chunks in this message, then finish
918 * - If there are Read chunks in this message, post Read WRs to
924 struct svc_xprt *xprt = rqstp->rq_xprt;
933 rqstp->rq_respages = rqstp->rq_pages;
934 rqstp->rq_next_page = rqstp->rq_respages;
936 rqstp->rq_xprt_ctxt = NULL;
938 spin_lock(&rdma_xprt->sc_rq_dto_lock);
939 ctxt = svc_rdma_next_recv_ctxt(&rdma_xprt->sc_read_complete_q);
941 list_del(&ctxt->rc_list);
942 spin_unlock(&rdma_xprt->sc_rq_dto_lock);
947 ctxt = svc_rdma_next_recv_ctxt(&rdma_xprt->sc_rq_dto_q);
949 list_del(&ctxt->rc_list);
951 /* No new incoming requests, terminate the loop */
952 clear_bit(XPT_DATA, &xprt->xpt_flags);
953 spin_unlock(&rdma_xprt->sc_rq_dto_lock);
961 ib_dma_sync_single_for_cpu(rdma_xprt->sc_pd->device,
962 ctxt->rc_recv_sge.addr, ctxt->rc_byte_len,
966 ret = svc_rdma_xdr_decode_req(&rqstp->rq_arg, ctxt);
977 if (!pcl_is_empty(&ctxt->rc_read_pcl) ||
978 !pcl_is_empty(&ctxt->rc_call_pcl))
982 rqstp->rq_xprt_ctxt = ctxt;
983 rqstp->rq_prot = IPPROTO_MAX;
985 set_bit(RQ_SECURE, &rqstp->rq_flags);
986 return rqstp->rq_arg.len;
999 ctxt->rc_saved_arg = rqstp->rq_arg;
1003 if (ret == -EINVAL)