49  * be read. "Data Ready" is signaled when an RDMA Receive completes,
50  * or when a set of RDMA Reads complete.
61  * svc_rdma_recvfrom must post RDMA Reads to pull the RPC Call's
63  * RDMA Reads using pages in svc_rqst::rq_pages, which are
76  * of the RPC Call message, using the RDMA Read sink pages kept in
86  * The first svc_rqst supplies pages for RDMA Reads. These are moved
91  * During the second svc_rdma_recvfrom call, RDMA Read sink pages
98 #include <rdma/ib_verbs.h>
99 #include <rdma/rdma_cm.h>
118 static void svc_rdma_recv_cid_init(struct svcxprt_rdma *rdma,  in svc_rdma_recv_cid_init()  argument
121 	cid->ci_queue_id = rdma->sc_rq_cq->res.id;  in svc_rdma_recv_cid_init()
122 	cid->ci_completion_id = atomic_inc_return(&rdma->sc_completion_ids);  in svc_rdma_recv_cid_init()
126 svc_rdma_recv_ctxt_alloc(struct svcxprt_rdma *rdma)  in svc_rdma_recv_ctxt_alloc()  argument
128 	int node = ibdev_to_node(rdma->sc_cm_id->device);  in svc_rdma_recv_ctxt_alloc()
136 	buffer = kmalloc_node(rdma->sc_max_req_size, GFP_KERNEL, node);  in svc_rdma_recv_ctxt_alloc()
139 	addr = ib_dma_map_single(rdma->sc_pd->device, buffer,  in svc_rdma_recv_ctxt_alloc()
140 				 rdma->sc_max_req_size, DMA_FROM_DEVICE);  in svc_rdma_recv_ctxt_alloc()
141 	if (ib_dma_mapping_error(rdma->sc_pd->device, addr))  in svc_rdma_recv_ctxt_alloc()
144 	svc_rdma_recv_cid_init(rdma, &ctxt->rc_cid);  in svc_rdma_recv_ctxt_alloc()
156 	ctxt->rc_recv_sge.length = rdma->sc_max_req_size;  in svc_rdma_recv_ctxt_alloc()
157 	ctxt->rc_recv_sge.lkey = rdma->sc_pd->local_dma_lkey;  in svc_rdma_recv_ctxt_alloc()
169 static void svc_rdma_recv_ctxt_destroy(struct svcxprt_rdma *rdma,  in svc_rdma_recv_ctxt_destroy()  argument
172 	ib_dma_unmap_single(rdma->sc_pd->device, ctxt->rc_recv_sge.addr,  in svc_rdma_recv_ctxt_destroy()
180  * @rdma: svcxprt_rdma being torn down
183 void svc_rdma_recv_ctxts_destroy(struct svcxprt_rdma *rdma)  in svc_rdma_recv_ctxts_destroy()  argument
188 	while ((node = llist_del_first(&rdma->sc_recv_ctxts))) {  in svc_rdma_recv_ctxts_destroy()
190 		svc_rdma_recv_ctxt_destroy(rdma, ctxt);  in svc_rdma_recv_ctxts_destroy()
196  * @rdma: controlling svcxprt_rdma
200 struct svc_rdma_recv_ctxt *svc_rdma_recv_ctxt_get(struct svcxprt_rdma *rdma)  in svc_rdma_recv_ctxt_get()  argument
205 	node = llist_del_first(&rdma->sc_recv_ctxts);  in svc_rdma_recv_ctxt_get()
215 	ctxt = svc_rdma_recv_ctxt_alloc(rdma);  in svc_rdma_recv_ctxt_get()
223  * @rdma: controlling svcxprt_rdma
227 void svc_rdma_recv_ctxt_put(struct svcxprt_rdma *rdma,  in svc_rdma_recv_ctxt_put()  argument
235 	llist_add(&ctxt->rc_node, &rdma->sc_recv_ctxts);  in svc_rdma_recv_ctxt_put()
250 	struct svcxprt_rdma *rdma =  in svc_rdma_release_ctxt()  local
254 		svc_rdma_recv_ctxt_put(rdma, ctxt);  in svc_rdma_release_ctxt()
257 static bool svc_rdma_refresh_recvs(struct svcxprt_rdma *rdma,  in svc_rdma_refresh_recvs()  argument
265 	if (test_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags))  in svc_rdma_refresh_recvs()
270 		ctxt = svc_rdma_recv_ctxt_get(rdma);  in svc_rdma_refresh_recvs()
277 		rdma->sc_pending_recvs++;  in svc_rdma_refresh_recvs()
282 	ret = ib_post_recv(rdma->sc_qp, recv_chain, &bad_wr);  in svc_rdma_refresh_recvs()
288 	trace_svcrdma_rq_post_err(rdma, ret);  in svc_rdma_refresh_recvs()
293 		svc_rdma_recv_ctxt_put(rdma, ctxt);  in svc_rdma_refresh_recvs()
302  * @rdma: fresh svcxprt_rdma
306 bool svc_rdma_post_recvs(struct svcxprt_rdma *rdma)  in svc_rdma_post_recvs()  argument
308 	return svc_rdma_refresh_recvs(rdma, rdma->sc_max_requests);  in svc_rdma_post_recvs()
312  * svc_rdma_wc_receive - Invoked by RDMA provider for each polled Receive WC
319 	struct svcxprt_rdma *rdma = cq->cq_context;  in svc_rdma_wc_receive()  local
323 	rdma->sc_pending_recvs--;  in svc_rdma_wc_receive()
341 	if (rdma->sc_pending_recvs < rdma->sc_max_requests)  in svc_rdma_wc_receive()
342 		if (!svc_rdma_refresh_recvs(rdma, rdma->sc_recv_batch))  in svc_rdma_wc_receive()
348 	spin_lock(&rdma->sc_rq_dto_lock);  in svc_rdma_wc_receive()
349 	list_add_tail(&ctxt->rc_list, &rdma->sc_rq_dto_q);  in svc_rdma_wc_receive()
351 	set_bit(XPT_DATA, &rdma->sc_xprt.xpt_flags);  in svc_rdma_wc_receive()
352 	spin_unlock(&rdma->sc_rq_dto_lock);  in svc_rdma_wc_receive()
353 	if (!test_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags))  in svc_rdma_wc_receive()
354 		svc_xprt_enqueue(&rdma->sc_xprt);  in svc_rdma_wc_receive()
363 	svc_rdma_recv_ctxt_put(rdma, ctxt);  in svc_rdma_wc_receive()
364 	svc_xprt_deferred_close(&rdma->sc_xprt);  in svc_rdma_wc_receive()
369  * @rdma: svcxprt_rdma being shut down
372 void svc_rdma_flush_recv_queues(struct svcxprt_rdma *rdma)  in svc_rdma_flush_recv_queues()  argument
376 	while ((ctxt = svc_rdma_next_recv_ctxt(&rdma->sc_rq_dto_q))) {  in svc_rdma_flush_recv_queues()
378 		svc_rdma_recv_ctxt_put(rdma, ctxt);  in svc_rdma_flush_recv_queues()
582 /* RPC-over-RDMA Version One private extension: Remote Invalidation.
589 static void svc_rdma_get_inv_rkey(struct svcxprt_rdma *rdma,  in svc_rdma_get_inv_rkey()  argument
598 	if (!rdma->sc_snd_w_inv)  in svc_rdma_get_inv_rkey()
639  * @rq_arg: xdr_buf containing ingress RPC/RDMA message
643  * RPC-over-RDMA transport header.
646  * RPC-over-RDMA header. For RDMA_MSG, this is the RPC message.
648  * The length of the RPC-over-RDMA header is returned.
719 static void svc_rdma_send_error(struct svcxprt_rdma *rdma,  in svc_rdma_send_error()  argument
725 	sctxt = svc_rdma_send_ctxt_get(rdma);  in svc_rdma_send_error()
728 	svc_rdma_send_error_msg(rdma, sctxt, rctxt, status);  in svc_rdma_send_error()
733  * the RPC/RDMA header small and fixed in size, so it is