verbs.c (36604B)
1// SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause 2/* 3 * Copyright (c) 2014-2017 Oracle. All rights reserved. 4 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved. 5 * 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 10 * license below: 11 * 12 * Redistribution and use in source and binary forms, with or without 13 * modification, are permitted provided that the following conditions 14 * are met: 15 * 16 * Redistributions of source code must retain the above copyright 17 * notice, this list of conditions and the following disclaimer. 18 * 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. 23 * 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 27 * permission. 28 * 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. 40 */ 41 42/* 43 * verbs.c 44 * 45 * Encapsulates the major functions managing: 46 * o adapters 47 * o endpoints 48 * o connections 49 * o buffer memory 50 */ 51 52#include <linux/interrupt.h> 53#include <linux/slab.h> 54#include <linux/sunrpc/addr.h> 55#include <linux/sunrpc/svc_rdma.h> 56#include <linux/log2.h> 57 58#include <asm-generic/barrier.h> 59#include <asm/bitops.h> 60 61#include <rdma/ib_cm.h> 62 63#include "xprt_rdma.h" 64#include <trace/events/rpcrdma.h> 65 66static int rpcrdma_sendctxs_create(struct rpcrdma_xprt *r_xprt); 67static void rpcrdma_sendctxs_destroy(struct rpcrdma_xprt *r_xprt); 68static void rpcrdma_sendctx_put_locked(struct rpcrdma_xprt *r_xprt, 69 struct rpcrdma_sendctx *sc); 70static int rpcrdma_reqs_setup(struct rpcrdma_xprt *r_xprt); 71static void rpcrdma_reqs_reset(struct rpcrdma_xprt *r_xprt); 72static void rpcrdma_rep_destroy(struct rpcrdma_rep *rep); 73static void rpcrdma_reps_unmap(struct rpcrdma_xprt *r_xprt); 74static void rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt); 75static void rpcrdma_mrs_destroy(struct rpcrdma_xprt *r_xprt); 76static void rpcrdma_ep_get(struct rpcrdma_ep *ep); 77static int rpcrdma_ep_put(struct rpcrdma_ep *ep); 78static struct rpcrdma_regbuf * 79rpcrdma_regbuf_alloc(size_t size, enum dma_data_direction direction, 80 gfp_t flags); 81static void rpcrdma_regbuf_dma_unmap(struct rpcrdma_regbuf *rb); 82static void rpcrdma_regbuf_free(struct rpcrdma_regbuf *rb); 83 84/* Wait for outstanding transport work to finish. ib_drain_qp 85 * handles the drains in the wrong order for us, so open code 86 * them here. 87 */ 88static void rpcrdma_xprt_drain(struct rpcrdma_xprt *r_xprt) 89{ 90 struct rpcrdma_ep *ep = r_xprt->rx_ep; 91 struct rdma_cm_id *id = ep->re_id; 92 93 /* Wait for rpcrdma_post_recvs() to leave its critical 94 * section. 95 */ 96 if (atomic_inc_return(&ep->re_receiving) > 1) 97 wait_for_completion(&ep->re_done); 98 99 /* Flush Receives, then wait for deferred Reply work 100 * to complete. 101 */ 102 ib_drain_rq(id->qp); 103 104 /* Deferred Reply processing might have scheduled 105 * local invalidations. 106 */ 107 ib_drain_sq(id->qp); 108 109 rpcrdma_ep_put(ep); 110} 111 112/* Ensure xprt_force_disconnect() is invoked exactly once when a 113 * connection is closed or lost. (The important thing is it needs 114 * to be invoked "at least" once). 115 */ 116void rpcrdma_force_disconnect(struct rpcrdma_ep *ep) 117{ 118 if (atomic_add_unless(&ep->re_force_disconnect, 1, 1)) 119 xprt_force_disconnect(ep->re_xprt); 120} 121 122/** 123 * rpcrdma_flush_disconnect - Disconnect on flushed completion 124 * @r_xprt: transport to disconnect 125 * @wc: work completion entry 126 * 127 * Must be called in process context. 128 */ 129void rpcrdma_flush_disconnect(struct rpcrdma_xprt *r_xprt, struct ib_wc *wc) 130{ 131 if (wc->status != IB_WC_SUCCESS) 132 rpcrdma_force_disconnect(r_xprt->rx_ep); 133} 134 135/** 136 * rpcrdma_wc_send - Invoked by RDMA provider for each polled Send WC 137 * @cq: completion queue 138 * @wc: WCE for a completed Send WR 139 * 140 */ 141static void rpcrdma_wc_send(struct ib_cq *cq, struct ib_wc *wc) 142{ 143 struct ib_cqe *cqe = wc->wr_cqe; 144 struct rpcrdma_sendctx *sc = 145 container_of(cqe, struct rpcrdma_sendctx, sc_cqe); 146 struct rpcrdma_xprt *r_xprt = cq->cq_context; 147 148 /* WARNING: Only wr_cqe and status are reliable at this point */ 149 trace_xprtrdma_wc_send(wc, &sc->sc_cid); 150 rpcrdma_sendctx_put_locked(r_xprt, sc); 151 rpcrdma_flush_disconnect(r_xprt, wc); 152} 153 154/** 155 * rpcrdma_wc_receive - Invoked by RDMA provider for each polled Receive WC 156 * @cq: completion queue 157 * @wc: WCE for a completed Receive WR 158 * 159 */ 160static void rpcrdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc) 161{ 162 struct ib_cqe *cqe = wc->wr_cqe; 163 struct rpcrdma_rep *rep = container_of(cqe, struct rpcrdma_rep, 164 rr_cqe); 165 struct rpcrdma_xprt *r_xprt = cq->cq_context; 166 167 /* WARNING: Only wr_cqe and status are reliable at this point */ 168 trace_xprtrdma_wc_receive(wc, &rep->rr_cid); 169 --r_xprt->rx_ep->re_receive_count; 170 if (wc->status != IB_WC_SUCCESS) 171 goto out_flushed; 172 173 /* status == SUCCESS means all fields in wc are trustworthy */ 174 rpcrdma_set_xdrlen(&rep->rr_hdrbuf, wc->byte_len); 175 rep->rr_wc_flags = wc->wc_flags; 176 rep->rr_inv_rkey = wc->ex.invalidate_rkey; 177 178 ib_dma_sync_single_for_cpu(rdmab_device(rep->rr_rdmabuf), 179 rdmab_addr(rep->rr_rdmabuf), 180 wc->byte_len, DMA_FROM_DEVICE); 181 182 rpcrdma_reply_handler(rep); 183 return; 184 185out_flushed: 186 rpcrdma_flush_disconnect(r_xprt, wc); 187 rpcrdma_rep_put(&r_xprt->rx_buf, rep); 188} 189 190static void rpcrdma_update_cm_private(struct rpcrdma_ep *ep, 191 struct rdma_conn_param *param) 192{ 193 const struct rpcrdma_connect_private *pmsg = param->private_data; 194 unsigned int rsize, wsize; 195 196 /* Default settings for RPC-over-RDMA Version One */ 197 rsize = RPCRDMA_V1_DEF_INLINE_SIZE; 198 wsize = RPCRDMA_V1_DEF_INLINE_SIZE; 199 200 if (pmsg && 201 pmsg->cp_magic == rpcrdma_cmp_magic && 202 pmsg->cp_version == RPCRDMA_CMP_VERSION) { 203 rsize = rpcrdma_decode_buffer_size(pmsg->cp_send_size); 204 wsize = rpcrdma_decode_buffer_size(pmsg->cp_recv_size); 205 } 206 207 if (rsize < ep->re_inline_recv) 208 ep->re_inline_recv = rsize; 209 if (wsize < ep->re_inline_send) 210 ep->re_inline_send = wsize; 211 212 rpcrdma_set_max_header_sizes(ep); 213} 214 215/** 216 * rpcrdma_cm_event_handler - Handle RDMA CM events 217 * @id: rdma_cm_id on which an event has occurred 218 * @event: details of the event 219 * 220 * Called with @id's mutex held. Returns 1 if caller should 221 * destroy @id, otherwise 0. 222 */ 223static int 224rpcrdma_cm_event_handler(struct rdma_cm_id *id, struct rdma_cm_event *event) 225{ 226 struct sockaddr *sap = (struct sockaddr *)&id->route.addr.dst_addr; 227 struct rpcrdma_ep *ep = id->context; 228 229 might_sleep(); 230 231 switch (event->event) { 232 case RDMA_CM_EVENT_ADDR_RESOLVED: 233 case RDMA_CM_EVENT_ROUTE_RESOLVED: 234 ep->re_async_rc = 0; 235 complete(&ep->re_done); 236 return 0; 237 case RDMA_CM_EVENT_ADDR_ERROR: 238 ep->re_async_rc = -EPROTO; 239 complete(&ep->re_done); 240 return 0; 241 case RDMA_CM_EVENT_ROUTE_ERROR: 242 ep->re_async_rc = -ENETUNREACH; 243 complete(&ep->re_done); 244 return 0; 245 case RDMA_CM_EVENT_DEVICE_REMOVAL: 246 pr_info("rpcrdma: removing device %s for %pISpc\n", 247 ep->re_id->device->name, sap); 248 fallthrough; 249 case RDMA_CM_EVENT_ADDR_CHANGE: 250 ep->re_connect_status = -ENODEV; 251 goto disconnected; 252 case RDMA_CM_EVENT_ESTABLISHED: 253 rpcrdma_ep_get(ep); 254 ep->re_connect_status = 1; 255 rpcrdma_update_cm_private(ep, &event->param.conn); 256 trace_xprtrdma_inline_thresh(ep); 257 wake_up_all(&ep->re_connect_wait); 258 break; 259 case RDMA_CM_EVENT_CONNECT_ERROR: 260 ep->re_connect_status = -ENOTCONN; 261 goto wake_connect_worker; 262 case RDMA_CM_EVENT_UNREACHABLE: 263 ep->re_connect_status = -ENETUNREACH; 264 goto wake_connect_worker; 265 case RDMA_CM_EVENT_REJECTED: 266 ep->re_connect_status = -ECONNREFUSED; 267 if (event->status == IB_CM_REJ_STALE_CONN) 268 ep->re_connect_status = -ENOTCONN; 269wake_connect_worker: 270 wake_up_all(&ep->re_connect_wait); 271 return 0; 272 case RDMA_CM_EVENT_DISCONNECTED: 273 ep->re_connect_status = -ECONNABORTED; 274disconnected: 275 rpcrdma_force_disconnect(ep); 276 return rpcrdma_ep_put(ep); 277 default: 278 break; 279 } 280 281 return 0; 282} 283 284static struct rdma_cm_id *rpcrdma_create_id(struct rpcrdma_xprt *r_xprt, 285 struct rpcrdma_ep *ep) 286{ 287 unsigned long wtimeout = msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1; 288 struct rpc_xprt *xprt = &r_xprt->rx_xprt; 289 struct rdma_cm_id *id; 290 int rc; 291 292 init_completion(&ep->re_done); 293 294 id = rdma_create_id(xprt->xprt_net, rpcrdma_cm_event_handler, ep, 295 RDMA_PS_TCP, IB_QPT_RC); 296 if (IS_ERR(id)) 297 return id; 298 299 ep->re_async_rc = -ETIMEDOUT; 300 rc = rdma_resolve_addr(id, NULL, (struct sockaddr *)&xprt->addr, 301 RDMA_RESOLVE_TIMEOUT); 302 if (rc) 303 goto out; 304 rc = wait_for_completion_interruptible_timeout(&ep->re_done, wtimeout); 305 if (rc < 0) 306 goto out; 307 308 rc = ep->re_async_rc; 309 if (rc) 310 goto out; 311 312 ep->re_async_rc = -ETIMEDOUT; 313 rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT); 314 if (rc) 315 goto out; 316 rc = wait_for_completion_interruptible_timeout(&ep->re_done, wtimeout); 317 if (rc < 0) 318 goto out; 319 rc = ep->re_async_rc; 320 if (rc) 321 goto out; 322 323 return id; 324 325out: 326 rdma_destroy_id(id); 327 return ERR_PTR(rc); 328} 329 330static void rpcrdma_ep_destroy(struct kref *kref) 331{ 332 struct rpcrdma_ep *ep = container_of(kref, struct rpcrdma_ep, re_kref); 333 334 if (ep->re_id->qp) { 335 rdma_destroy_qp(ep->re_id); 336 ep->re_id->qp = NULL; 337 } 338 339 if (ep->re_attr.recv_cq) 340 ib_free_cq(ep->re_attr.recv_cq); 341 ep->re_attr.recv_cq = NULL; 342 if (ep->re_attr.send_cq) 343 ib_free_cq(ep->re_attr.send_cq); 344 ep->re_attr.send_cq = NULL; 345 346 if (ep->re_pd) 347 ib_dealloc_pd(ep->re_pd); 348 ep->re_pd = NULL; 349 350 kfree(ep); 351 module_put(THIS_MODULE); 352} 353 354static noinline void rpcrdma_ep_get(struct rpcrdma_ep *ep) 355{ 356 kref_get(&ep->re_kref); 357} 358 359/* Returns: 360 * %0 if @ep still has a positive kref count, or 361 * %1 if @ep was destroyed successfully. 362 */ 363static noinline int rpcrdma_ep_put(struct rpcrdma_ep *ep) 364{ 365 return kref_put(&ep->re_kref, rpcrdma_ep_destroy); 366} 367 368static int rpcrdma_ep_create(struct rpcrdma_xprt *r_xprt) 369{ 370 struct rpcrdma_connect_private *pmsg; 371 struct ib_device *device; 372 struct rdma_cm_id *id; 373 struct rpcrdma_ep *ep; 374 int rc; 375 376 ep = kzalloc(sizeof(*ep), GFP_KERNEL); 377 if (!ep) 378 return -ENOTCONN; 379 ep->re_xprt = &r_xprt->rx_xprt; 380 kref_init(&ep->re_kref); 381 382 id = rpcrdma_create_id(r_xprt, ep); 383 if (IS_ERR(id)) { 384 kfree(ep); 385 return PTR_ERR(id); 386 } 387 __module_get(THIS_MODULE); 388 device = id->device; 389 ep->re_id = id; 390 reinit_completion(&ep->re_done); 391 392 ep->re_max_requests = r_xprt->rx_xprt.max_reqs; 393 ep->re_inline_send = xprt_rdma_max_inline_write; 394 ep->re_inline_recv = xprt_rdma_max_inline_read; 395 rc = frwr_query_device(ep, device); 396 if (rc) 397 goto out_destroy; 398 399 r_xprt->rx_buf.rb_max_requests = cpu_to_be32(ep->re_max_requests); 400 401 ep->re_attr.srq = NULL; 402 ep->re_attr.cap.max_inline_data = 0; 403 ep->re_attr.sq_sig_type = IB_SIGNAL_REQ_WR; 404 ep->re_attr.qp_type = IB_QPT_RC; 405 ep->re_attr.port_num = ~0; 406 407 ep->re_send_batch = ep->re_max_requests >> 3; 408 ep->re_send_count = ep->re_send_batch; 409 init_waitqueue_head(&ep->re_connect_wait); 410 411 ep->re_attr.send_cq = ib_alloc_cq_any(device, r_xprt, 412 ep->re_attr.cap.max_send_wr, 413 IB_POLL_WORKQUEUE); 414 if (IS_ERR(ep->re_attr.send_cq)) { 415 rc = PTR_ERR(ep->re_attr.send_cq); 416 ep->re_attr.send_cq = NULL; 417 goto out_destroy; 418 } 419 420 ep->re_attr.recv_cq = ib_alloc_cq_any(device, r_xprt, 421 ep->re_attr.cap.max_recv_wr, 422 IB_POLL_WORKQUEUE); 423 if (IS_ERR(ep->re_attr.recv_cq)) { 424 rc = PTR_ERR(ep->re_attr.recv_cq); 425 ep->re_attr.recv_cq = NULL; 426 goto out_destroy; 427 } 428 ep->re_receive_count = 0; 429 430 /* Initialize cma parameters */ 431 memset(&ep->re_remote_cma, 0, sizeof(ep->re_remote_cma)); 432 433 /* Prepare RDMA-CM private message */ 434 pmsg = &ep->re_cm_private; 435 pmsg->cp_magic = rpcrdma_cmp_magic; 436 pmsg->cp_version = RPCRDMA_CMP_VERSION; 437 pmsg->cp_flags |= RPCRDMA_CMP_F_SND_W_INV_OK; 438 pmsg->cp_send_size = rpcrdma_encode_buffer_size(ep->re_inline_send); 439 pmsg->cp_recv_size = rpcrdma_encode_buffer_size(ep->re_inline_recv); 440 ep->re_remote_cma.private_data = pmsg; 441 ep->re_remote_cma.private_data_len = sizeof(*pmsg); 442 443 /* Client offers RDMA Read but does not initiate */ 444 ep->re_remote_cma.initiator_depth = 0; 445 ep->re_remote_cma.responder_resources = 446 min_t(int, U8_MAX, device->attrs.max_qp_rd_atom); 447 448 /* Limit transport retries so client can detect server 449 * GID changes quickly. RPC layer handles re-establishing 450 * transport connection and retransmission. 451 */ 452 ep->re_remote_cma.retry_count = 6; 453 454 /* RPC-over-RDMA handles its own flow control. In addition, 455 * make all RNR NAKs visible so we know that RPC-over-RDMA 456 * flow control is working correctly (no NAKs should be seen). 457 */ 458 ep->re_remote_cma.flow_control = 0; 459 ep->re_remote_cma.rnr_retry_count = 0; 460 461 ep->re_pd = ib_alloc_pd(device, 0); 462 if (IS_ERR(ep->re_pd)) { 463 rc = PTR_ERR(ep->re_pd); 464 ep->re_pd = NULL; 465 goto out_destroy; 466 } 467 468 rc = rdma_create_qp(id, ep->re_pd, &ep->re_attr); 469 if (rc) 470 goto out_destroy; 471 472 r_xprt->rx_ep = ep; 473 return 0; 474 475out_destroy: 476 rpcrdma_ep_put(ep); 477 rdma_destroy_id(id); 478 return rc; 479} 480 481/** 482 * rpcrdma_xprt_connect - Connect an unconnected transport 483 * @r_xprt: controlling transport instance 484 * 485 * Returns 0 on success or a negative errno. 486 */ 487int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt) 488{ 489 struct rpc_xprt *xprt = &r_xprt->rx_xprt; 490 struct rpcrdma_ep *ep; 491 int rc; 492 493 rc = rpcrdma_ep_create(r_xprt); 494 if (rc) 495 return rc; 496 ep = r_xprt->rx_ep; 497 498 xprt_clear_connected(xprt); 499 rpcrdma_reset_cwnd(r_xprt); 500 501 /* Bump the ep's reference count while there are 502 * outstanding Receives. 503 */ 504 rpcrdma_ep_get(ep); 505 rpcrdma_post_recvs(r_xprt, 1, true); 506 507 rc = rdma_connect(ep->re_id, &ep->re_remote_cma); 508 if (rc) 509 goto out; 510 511 if (xprt->reestablish_timeout < RPCRDMA_INIT_REEST_TO) 512 xprt->reestablish_timeout = RPCRDMA_INIT_REEST_TO; 513 wait_event_interruptible(ep->re_connect_wait, 514 ep->re_connect_status != 0); 515 if (ep->re_connect_status <= 0) { 516 rc = ep->re_connect_status; 517 goto out; 518 } 519 520 rc = rpcrdma_sendctxs_create(r_xprt); 521 if (rc) { 522 rc = -ENOTCONN; 523 goto out; 524 } 525 526 rc = rpcrdma_reqs_setup(r_xprt); 527 if (rc) { 528 rc = -ENOTCONN; 529 goto out; 530 } 531 rpcrdma_mrs_create(r_xprt); 532 frwr_wp_create(r_xprt); 533 534out: 535 trace_xprtrdma_connect(r_xprt, rc); 536 return rc; 537} 538 539/** 540 * rpcrdma_xprt_disconnect - Disconnect underlying transport 541 * @r_xprt: controlling transport instance 542 * 543 * Caller serializes. Either the transport send lock is held, 544 * or we're being called to destroy the transport. 545 * 546 * On return, @r_xprt is completely divested of all hardware 547 * resources and prepared for the next ->connect operation. 548 */ 549void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt) 550{ 551 struct rpcrdma_ep *ep = r_xprt->rx_ep; 552 struct rdma_cm_id *id; 553 int rc; 554 555 if (!ep) 556 return; 557 558 id = ep->re_id; 559 rc = rdma_disconnect(id); 560 trace_xprtrdma_disconnect(r_xprt, rc); 561 562 rpcrdma_xprt_drain(r_xprt); 563 rpcrdma_reps_unmap(r_xprt); 564 rpcrdma_reqs_reset(r_xprt); 565 rpcrdma_mrs_destroy(r_xprt); 566 rpcrdma_sendctxs_destroy(r_xprt); 567 568 if (rpcrdma_ep_put(ep)) 569 rdma_destroy_id(id); 570 571 r_xprt->rx_ep = NULL; 572} 573 574/* Fixed-size circular FIFO queue. This implementation is wait-free and 575 * lock-free. 576 * 577 * Consumer is the code path that posts Sends. This path dequeues a 578 * sendctx for use by a Send operation. Multiple consumer threads 579 * are serialized by the RPC transport lock, which allows only one 580 * ->send_request call at a time. 581 * 582 * Producer is the code path that handles Send completions. This path 583 * enqueues a sendctx that has been completed. Multiple producer 584 * threads are serialized by the ib_poll_cq() function. 585 */ 586 587/* rpcrdma_sendctxs_destroy() assumes caller has already quiesced 588 * queue activity, and rpcrdma_xprt_drain has flushed all remaining 589 * Send requests. 590 */ 591static void rpcrdma_sendctxs_destroy(struct rpcrdma_xprt *r_xprt) 592{ 593 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 594 unsigned long i; 595 596 if (!buf->rb_sc_ctxs) 597 return; 598 for (i = 0; i <= buf->rb_sc_last; i++) 599 kfree(buf->rb_sc_ctxs[i]); 600 kfree(buf->rb_sc_ctxs); 601 buf->rb_sc_ctxs = NULL; 602} 603 604static struct rpcrdma_sendctx *rpcrdma_sendctx_create(struct rpcrdma_ep *ep) 605{ 606 struct rpcrdma_sendctx *sc; 607 608 sc = kzalloc(struct_size(sc, sc_sges, ep->re_attr.cap.max_send_sge), 609 GFP_KERNEL); 610 if (!sc) 611 return NULL; 612 613 sc->sc_cqe.done = rpcrdma_wc_send; 614 sc->sc_cid.ci_queue_id = ep->re_attr.send_cq->res.id; 615 sc->sc_cid.ci_completion_id = 616 atomic_inc_return(&ep->re_completion_ids); 617 return sc; 618} 619 620static int rpcrdma_sendctxs_create(struct rpcrdma_xprt *r_xprt) 621{ 622 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 623 struct rpcrdma_sendctx *sc; 624 unsigned long i; 625 626 /* Maximum number of concurrent outstanding Send WRs. Capping 627 * the circular queue size stops Send Queue overflow by causing 628 * the ->send_request call to fail temporarily before too many 629 * Sends are posted. 630 */ 631 i = r_xprt->rx_ep->re_max_requests + RPCRDMA_MAX_BC_REQUESTS; 632 buf->rb_sc_ctxs = kcalloc(i, sizeof(sc), GFP_KERNEL); 633 if (!buf->rb_sc_ctxs) 634 return -ENOMEM; 635 636 buf->rb_sc_last = i - 1; 637 for (i = 0; i <= buf->rb_sc_last; i++) { 638 sc = rpcrdma_sendctx_create(r_xprt->rx_ep); 639 if (!sc) 640 return -ENOMEM; 641 642 buf->rb_sc_ctxs[i] = sc; 643 } 644 645 buf->rb_sc_head = 0; 646 buf->rb_sc_tail = 0; 647 return 0; 648} 649 650/* The sendctx queue is not guaranteed to have a size that is a 651 * power of two, thus the helpers in circ_buf.h cannot be used. 652 * The other option is to use modulus (%), which can be expensive. 653 */ 654static unsigned long rpcrdma_sendctx_next(struct rpcrdma_buffer *buf, 655 unsigned long item) 656{ 657 return likely(item < buf->rb_sc_last) ? item + 1 : 0; 658} 659 660/** 661 * rpcrdma_sendctx_get_locked - Acquire a send context 662 * @r_xprt: controlling transport instance 663 * 664 * Returns pointer to a free send completion context; or NULL if 665 * the queue is empty. 666 * 667 * Usage: Called to acquire an SGE array before preparing a Send WR. 668 * 669 * The caller serializes calls to this function (per transport), and 670 * provides an effective memory barrier that flushes the new value 671 * of rb_sc_head. 672 */ 673struct rpcrdma_sendctx *rpcrdma_sendctx_get_locked(struct rpcrdma_xprt *r_xprt) 674{ 675 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 676 struct rpcrdma_sendctx *sc; 677 unsigned long next_head; 678 679 next_head = rpcrdma_sendctx_next(buf, buf->rb_sc_head); 680 681 if (next_head == READ_ONCE(buf->rb_sc_tail)) 682 goto out_emptyq; 683 684 /* ORDER: item must be accessed _before_ head is updated */ 685 sc = buf->rb_sc_ctxs[next_head]; 686 687 /* Releasing the lock in the caller acts as a memory 688 * barrier that flushes rb_sc_head. 689 */ 690 buf->rb_sc_head = next_head; 691 692 return sc; 693 694out_emptyq: 695 /* The queue is "empty" if there have not been enough Send 696 * completions recently. This is a sign the Send Queue is 697 * backing up. Cause the caller to pause and try again. 698 */ 699 xprt_wait_for_buffer_space(&r_xprt->rx_xprt); 700 r_xprt->rx_stats.empty_sendctx_q++; 701 return NULL; 702} 703 704/** 705 * rpcrdma_sendctx_put_locked - Release a send context 706 * @r_xprt: controlling transport instance 707 * @sc: send context to release 708 * 709 * Usage: Called from Send completion to return a sendctxt 710 * to the queue. 711 * 712 * The caller serializes calls to this function (per transport). 713 */ 714static void rpcrdma_sendctx_put_locked(struct rpcrdma_xprt *r_xprt, 715 struct rpcrdma_sendctx *sc) 716{ 717 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 718 unsigned long next_tail; 719 720 /* Unmap SGEs of previously completed but unsignaled 721 * Sends by walking up the queue until @sc is found. 722 */ 723 next_tail = buf->rb_sc_tail; 724 do { 725 next_tail = rpcrdma_sendctx_next(buf, next_tail); 726 727 /* ORDER: item must be accessed _before_ tail is updated */ 728 rpcrdma_sendctx_unmap(buf->rb_sc_ctxs[next_tail]); 729 730 } while (buf->rb_sc_ctxs[next_tail] != sc); 731 732 /* Paired with READ_ONCE */ 733 smp_store_release(&buf->rb_sc_tail, next_tail); 734 735 xprt_write_space(&r_xprt->rx_xprt); 736} 737 738static void 739rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt) 740{ 741 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 742 struct rpcrdma_ep *ep = r_xprt->rx_ep; 743 unsigned int count; 744 745 for (count = 0; count < ep->re_max_rdma_segs; count++) { 746 struct rpcrdma_mr *mr; 747 int rc; 748 749 mr = kzalloc(sizeof(*mr), GFP_KERNEL); 750 if (!mr) 751 break; 752 753 rc = frwr_mr_init(r_xprt, mr); 754 if (rc) { 755 kfree(mr); 756 break; 757 } 758 759 spin_lock(&buf->rb_lock); 760 rpcrdma_mr_push(mr, &buf->rb_mrs); 761 list_add(&mr->mr_all, &buf->rb_all_mrs); 762 spin_unlock(&buf->rb_lock); 763 } 764 765 r_xprt->rx_stats.mrs_allocated += count; 766 trace_xprtrdma_createmrs(r_xprt, count); 767} 768 769static void 770rpcrdma_mr_refresh_worker(struct work_struct *work) 771{ 772 struct rpcrdma_buffer *buf = container_of(work, struct rpcrdma_buffer, 773 rb_refresh_worker); 774 struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt, 775 rx_buf); 776 777 rpcrdma_mrs_create(r_xprt); 778 xprt_write_space(&r_xprt->rx_xprt); 779} 780 781/** 782 * rpcrdma_mrs_refresh - Wake the MR refresh worker 783 * @r_xprt: controlling transport instance 784 * 785 */ 786void rpcrdma_mrs_refresh(struct rpcrdma_xprt *r_xprt) 787{ 788 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 789 struct rpcrdma_ep *ep = r_xprt->rx_ep; 790 791 /* If there is no underlying connection, it's no use 792 * to wake the refresh worker. 793 */ 794 if (ep->re_connect_status == 1) { 795 /* The work is scheduled on a WQ_MEM_RECLAIM 796 * workqueue in order to prevent MR allocation 797 * from recursing into NFS during direct reclaim. 798 */ 799 queue_work(xprtiod_workqueue, &buf->rb_refresh_worker); 800 } 801} 802 803/** 804 * rpcrdma_req_create - Allocate an rpcrdma_req object 805 * @r_xprt: controlling r_xprt 806 * @size: initial size, in bytes, of send and receive buffers 807 * @flags: GFP flags passed to memory allocators 808 * 809 * Returns an allocated and fully initialized rpcrdma_req or NULL. 810 */ 811struct rpcrdma_req *rpcrdma_req_create(struct rpcrdma_xprt *r_xprt, size_t size, 812 gfp_t flags) 813{ 814 struct rpcrdma_buffer *buffer = &r_xprt->rx_buf; 815 struct rpcrdma_req *req; 816 817 req = kzalloc(sizeof(*req), flags); 818 if (req == NULL) 819 goto out1; 820 821 req->rl_sendbuf = rpcrdma_regbuf_alloc(size, DMA_TO_DEVICE, flags); 822 if (!req->rl_sendbuf) 823 goto out2; 824 825 req->rl_recvbuf = rpcrdma_regbuf_alloc(size, DMA_NONE, flags); 826 if (!req->rl_recvbuf) 827 goto out3; 828 829 INIT_LIST_HEAD(&req->rl_free_mrs); 830 INIT_LIST_HEAD(&req->rl_registered); 831 spin_lock(&buffer->rb_lock); 832 list_add(&req->rl_all, &buffer->rb_allreqs); 833 spin_unlock(&buffer->rb_lock); 834 return req; 835 836out3: 837 kfree(req->rl_sendbuf); 838out2: 839 kfree(req); 840out1: 841 return NULL; 842} 843 844/** 845 * rpcrdma_req_setup - Per-connection instance setup of an rpcrdma_req object 846 * @r_xprt: controlling transport instance 847 * @req: rpcrdma_req object to set up 848 * 849 * Returns zero on success, and a negative errno on failure. 850 */ 851int rpcrdma_req_setup(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req) 852{ 853 struct rpcrdma_regbuf *rb; 854 size_t maxhdrsize; 855 856 /* Compute maximum header buffer size in bytes */ 857 maxhdrsize = rpcrdma_fixed_maxsz + 3 + 858 r_xprt->rx_ep->re_max_rdma_segs * rpcrdma_readchunk_maxsz; 859 maxhdrsize *= sizeof(__be32); 860 rb = rpcrdma_regbuf_alloc(__roundup_pow_of_two(maxhdrsize), 861 DMA_TO_DEVICE, GFP_KERNEL); 862 if (!rb) 863 goto out; 864 865 if (!__rpcrdma_regbuf_dma_map(r_xprt, rb)) 866 goto out_free; 867 868 req->rl_rdmabuf = rb; 869 xdr_buf_init(&req->rl_hdrbuf, rdmab_data(rb), rdmab_length(rb)); 870 return 0; 871 872out_free: 873 rpcrdma_regbuf_free(rb); 874out: 875 return -ENOMEM; 876} 877 878/* ASSUMPTION: the rb_allreqs list is stable for the duration, 879 * and thus can be walked without holding rb_lock. Eg. the 880 * caller is holding the transport send lock to exclude 881 * device removal or disconnection. 882 */ 883static int rpcrdma_reqs_setup(struct rpcrdma_xprt *r_xprt) 884{ 885 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 886 struct rpcrdma_req *req; 887 int rc; 888 889 list_for_each_entry(req, &buf->rb_allreqs, rl_all) { 890 rc = rpcrdma_req_setup(r_xprt, req); 891 if (rc) 892 return rc; 893 } 894 return 0; 895} 896 897static void rpcrdma_req_reset(struct rpcrdma_req *req) 898{ 899 /* Credits are valid for only one connection */ 900 req->rl_slot.rq_cong = 0; 901 902 rpcrdma_regbuf_free(req->rl_rdmabuf); 903 req->rl_rdmabuf = NULL; 904 905 rpcrdma_regbuf_dma_unmap(req->rl_sendbuf); 906 rpcrdma_regbuf_dma_unmap(req->rl_recvbuf); 907 908 frwr_reset(req); 909} 910 911/* ASSUMPTION: the rb_allreqs list is stable for the duration, 912 * and thus can be walked without holding rb_lock. Eg. the 913 * caller is holding the transport send lock to exclude 914 * device removal or disconnection. 915 */ 916static void rpcrdma_reqs_reset(struct rpcrdma_xprt *r_xprt) 917{ 918 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 919 struct rpcrdma_req *req; 920 921 list_for_each_entry(req, &buf->rb_allreqs, rl_all) 922 rpcrdma_req_reset(req); 923} 924 925static noinline 926struct rpcrdma_rep *rpcrdma_rep_create(struct rpcrdma_xprt *r_xprt, 927 bool temp) 928{ 929 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 930 struct rpcrdma_rep *rep; 931 932 rep = kzalloc(sizeof(*rep), GFP_KERNEL); 933 if (rep == NULL) 934 goto out; 935 936 rep->rr_rdmabuf = rpcrdma_regbuf_alloc(r_xprt->rx_ep->re_inline_recv, 937 DMA_FROM_DEVICE, GFP_KERNEL); 938 if (!rep->rr_rdmabuf) 939 goto out_free; 940 941 if (!rpcrdma_regbuf_dma_map(r_xprt, rep->rr_rdmabuf)) 942 goto out_free_regbuf; 943 944 rep->rr_cid.ci_completion_id = 945 atomic_inc_return(&r_xprt->rx_ep->re_completion_ids); 946 947 xdr_buf_init(&rep->rr_hdrbuf, rdmab_data(rep->rr_rdmabuf), 948 rdmab_length(rep->rr_rdmabuf)); 949 rep->rr_cqe.done = rpcrdma_wc_receive; 950 rep->rr_rxprt = r_xprt; 951 rep->rr_recv_wr.next = NULL; 952 rep->rr_recv_wr.wr_cqe = &rep->rr_cqe; 953 rep->rr_recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov; 954 rep->rr_recv_wr.num_sge = 1; 955 rep->rr_temp = temp; 956 957 spin_lock(&buf->rb_lock); 958 list_add(&rep->rr_all, &buf->rb_all_reps); 959 spin_unlock(&buf->rb_lock); 960 return rep; 961 962out_free_regbuf: 963 rpcrdma_regbuf_free(rep->rr_rdmabuf); 964out_free: 965 kfree(rep); 966out: 967 return NULL; 968} 969 970static void rpcrdma_rep_free(struct rpcrdma_rep *rep) 971{ 972 rpcrdma_regbuf_free(rep->rr_rdmabuf); 973 kfree(rep); 974} 975 976static void rpcrdma_rep_destroy(struct rpcrdma_rep *rep) 977{ 978 struct rpcrdma_buffer *buf = &rep->rr_rxprt->rx_buf; 979 980 spin_lock(&buf->rb_lock); 981 list_del(&rep->rr_all); 982 spin_unlock(&buf->rb_lock); 983 984 rpcrdma_rep_free(rep); 985} 986 987static struct rpcrdma_rep *rpcrdma_rep_get_locked(struct rpcrdma_buffer *buf) 988{ 989 struct llist_node *node; 990 991 /* Calls to llist_del_first are required to be serialized */ 992 node = llist_del_first(&buf->rb_free_reps); 993 if (!node) 994 return NULL; 995 return llist_entry(node, struct rpcrdma_rep, rr_node); 996} 997 998/** 999 * rpcrdma_rep_put - Release rpcrdma_rep back to free list 1000 * @buf: buffer pool 1001 * @rep: rep to release 1002 * 1003 */ 1004void rpcrdma_rep_put(struct rpcrdma_buffer *buf, struct rpcrdma_rep *rep) 1005{ 1006 llist_add(&rep->rr_node, &buf->rb_free_reps); 1007} 1008 1009/* Caller must ensure the QP is quiescent (RQ is drained) before 1010 * invoking this function, to guarantee rb_all_reps is not 1011 * changing. 1012 */ 1013static void rpcrdma_reps_unmap(struct rpcrdma_xprt *r_xprt) 1014{ 1015 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1016 struct rpcrdma_rep *rep; 1017 1018 list_for_each_entry(rep, &buf->rb_all_reps, rr_all) { 1019 rpcrdma_regbuf_dma_unmap(rep->rr_rdmabuf); 1020 rep->rr_temp = true; /* Mark this rep for destruction */ 1021 } 1022} 1023 1024static void rpcrdma_reps_destroy(struct rpcrdma_buffer *buf) 1025{ 1026 struct rpcrdma_rep *rep; 1027 1028 spin_lock(&buf->rb_lock); 1029 while ((rep = list_first_entry_or_null(&buf->rb_all_reps, 1030 struct rpcrdma_rep, 1031 rr_all)) != NULL) { 1032 list_del(&rep->rr_all); 1033 spin_unlock(&buf->rb_lock); 1034 1035 rpcrdma_rep_free(rep); 1036 1037 spin_lock(&buf->rb_lock); 1038 } 1039 spin_unlock(&buf->rb_lock); 1040} 1041 1042/** 1043 * rpcrdma_buffer_create - Create initial set of req/rep objects 1044 * @r_xprt: transport instance to (re)initialize 1045 * 1046 * Returns zero on success, otherwise a negative errno. 1047 */ 1048int rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt) 1049{ 1050 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1051 int i, rc; 1052 1053 buf->rb_bc_srv_max_requests = 0; 1054 spin_lock_init(&buf->rb_lock); 1055 INIT_LIST_HEAD(&buf->rb_mrs); 1056 INIT_LIST_HEAD(&buf->rb_all_mrs); 1057 INIT_WORK(&buf->rb_refresh_worker, rpcrdma_mr_refresh_worker); 1058 1059 INIT_LIST_HEAD(&buf->rb_send_bufs); 1060 INIT_LIST_HEAD(&buf->rb_allreqs); 1061 INIT_LIST_HEAD(&buf->rb_all_reps); 1062 1063 rc = -ENOMEM; 1064 for (i = 0; i < r_xprt->rx_xprt.max_reqs; i++) { 1065 struct rpcrdma_req *req; 1066 1067 req = rpcrdma_req_create(r_xprt, RPCRDMA_V1_DEF_INLINE_SIZE * 2, 1068 GFP_KERNEL); 1069 if (!req) 1070 goto out; 1071 list_add(&req->rl_list, &buf->rb_send_bufs); 1072 } 1073 1074 init_llist_head(&buf->rb_free_reps); 1075 1076 return 0; 1077out: 1078 rpcrdma_buffer_destroy(buf); 1079 return rc; 1080} 1081 1082/** 1083 * rpcrdma_req_destroy - Destroy an rpcrdma_req object 1084 * @req: unused object to be destroyed 1085 * 1086 * Relies on caller holding the transport send lock to protect 1087 * removing req->rl_all from buf->rb_all_reqs safely. 1088 */ 1089void rpcrdma_req_destroy(struct rpcrdma_req *req) 1090{ 1091 struct rpcrdma_mr *mr; 1092 1093 list_del(&req->rl_all); 1094 1095 while ((mr = rpcrdma_mr_pop(&req->rl_free_mrs))) { 1096 struct rpcrdma_buffer *buf = &mr->mr_xprt->rx_buf; 1097 1098 spin_lock(&buf->rb_lock); 1099 list_del(&mr->mr_all); 1100 spin_unlock(&buf->rb_lock); 1101 1102 frwr_mr_release(mr); 1103 } 1104 1105 rpcrdma_regbuf_free(req->rl_recvbuf); 1106 rpcrdma_regbuf_free(req->rl_sendbuf); 1107 rpcrdma_regbuf_free(req->rl_rdmabuf); 1108 kfree(req); 1109} 1110 1111/** 1112 * rpcrdma_mrs_destroy - Release all of a transport's MRs 1113 * @r_xprt: controlling transport instance 1114 * 1115 * Relies on caller holding the transport send lock to protect 1116 * removing mr->mr_list from req->rl_free_mrs safely. 1117 */ 1118static void rpcrdma_mrs_destroy(struct rpcrdma_xprt *r_xprt) 1119{ 1120 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1121 struct rpcrdma_mr *mr; 1122 1123 cancel_work_sync(&buf->rb_refresh_worker); 1124 1125 spin_lock(&buf->rb_lock); 1126 while ((mr = list_first_entry_or_null(&buf->rb_all_mrs, 1127 struct rpcrdma_mr, 1128 mr_all)) != NULL) { 1129 list_del(&mr->mr_list); 1130 list_del(&mr->mr_all); 1131 spin_unlock(&buf->rb_lock); 1132 1133 frwr_mr_release(mr); 1134 1135 spin_lock(&buf->rb_lock); 1136 } 1137 spin_unlock(&buf->rb_lock); 1138} 1139 1140/** 1141 * rpcrdma_buffer_destroy - Release all hw resources 1142 * @buf: root control block for resources 1143 * 1144 * ORDERING: relies on a prior rpcrdma_xprt_drain : 1145 * - No more Send or Receive completions can occur 1146 * - All MRs, reps, and reqs are returned to their free lists 1147 */ 1148void 1149rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf) 1150{ 1151 rpcrdma_reps_destroy(buf); 1152 1153 while (!list_empty(&buf->rb_send_bufs)) { 1154 struct rpcrdma_req *req; 1155 1156 req = list_first_entry(&buf->rb_send_bufs, 1157 struct rpcrdma_req, rl_list); 1158 list_del(&req->rl_list); 1159 rpcrdma_req_destroy(req); 1160 } 1161} 1162 1163/** 1164 * rpcrdma_mr_get - Allocate an rpcrdma_mr object 1165 * @r_xprt: controlling transport 1166 * 1167 * Returns an initialized rpcrdma_mr or NULL if no free 1168 * rpcrdma_mr objects are available. 1169 */ 1170struct rpcrdma_mr * 1171rpcrdma_mr_get(struct rpcrdma_xprt *r_xprt) 1172{ 1173 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1174 struct rpcrdma_mr *mr; 1175 1176 spin_lock(&buf->rb_lock); 1177 mr = rpcrdma_mr_pop(&buf->rb_mrs); 1178 spin_unlock(&buf->rb_lock); 1179 return mr; 1180} 1181 1182/** 1183 * rpcrdma_reply_put - Put reply buffers back into pool 1184 * @buffers: buffer pool 1185 * @req: object to return 1186 * 1187 */ 1188void rpcrdma_reply_put(struct rpcrdma_buffer *buffers, struct rpcrdma_req *req) 1189{ 1190 if (req->rl_reply) { 1191 rpcrdma_rep_put(buffers, req->rl_reply); 1192 req->rl_reply = NULL; 1193 } 1194} 1195 1196/** 1197 * rpcrdma_buffer_get - Get a request buffer 1198 * @buffers: Buffer pool from which to obtain a buffer 1199 * 1200 * Returns a fresh rpcrdma_req, or NULL if none are available. 1201 */ 1202struct rpcrdma_req * 1203rpcrdma_buffer_get(struct rpcrdma_buffer *buffers) 1204{ 1205 struct rpcrdma_req *req; 1206 1207 spin_lock(&buffers->rb_lock); 1208 req = list_first_entry_or_null(&buffers->rb_send_bufs, 1209 struct rpcrdma_req, rl_list); 1210 if (req) 1211 list_del_init(&req->rl_list); 1212 spin_unlock(&buffers->rb_lock); 1213 return req; 1214} 1215 1216/** 1217 * rpcrdma_buffer_put - Put request/reply buffers back into pool 1218 * @buffers: buffer pool 1219 * @req: object to return 1220 * 1221 */ 1222void rpcrdma_buffer_put(struct rpcrdma_buffer *buffers, struct rpcrdma_req *req) 1223{ 1224 rpcrdma_reply_put(buffers, req); 1225 1226 spin_lock(&buffers->rb_lock); 1227 list_add(&req->rl_list, &buffers->rb_send_bufs); 1228 spin_unlock(&buffers->rb_lock); 1229} 1230 1231/* Returns a pointer to a rpcrdma_regbuf object, or NULL. 1232 * 1233 * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for 1234 * receiving the payload of RDMA RECV operations. During Long Calls 1235 * or Replies they may be registered externally via frwr_map. 1236 */ 1237static struct rpcrdma_regbuf * 1238rpcrdma_regbuf_alloc(size_t size, enum dma_data_direction direction, 1239 gfp_t flags) 1240{ 1241 struct rpcrdma_regbuf *rb; 1242 1243 rb = kmalloc(sizeof(*rb), flags); 1244 if (!rb) 1245 return NULL; 1246 rb->rg_data = kmalloc(size, flags); 1247 if (!rb->rg_data) { 1248 kfree(rb); 1249 return NULL; 1250 } 1251 1252 rb->rg_device = NULL; 1253 rb->rg_direction = direction; 1254 rb->rg_iov.length = size; 1255 return rb; 1256} 1257 1258/** 1259 * rpcrdma_regbuf_realloc - re-allocate a SEND/RECV buffer 1260 * @rb: regbuf to reallocate 1261 * @size: size of buffer to be allocated, in bytes 1262 * @flags: GFP flags 1263 * 1264 * Returns true if reallocation was successful. If false is 1265 * returned, @rb is left untouched. 1266 */ 1267bool rpcrdma_regbuf_realloc(struct rpcrdma_regbuf *rb, size_t size, gfp_t flags) 1268{ 1269 void *buf; 1270 1271 buf = kmalloc(size, flags); 1272 if (!buf) 1273 return false; 1274 1275 rpcrdma_regbuf_dma_unmap(rb); 1276 kfree(rb->rg_data); 1277 1278 rb->rg_data = buf; 1279 rb->rg_iov.length = size; 1280 return true; 1281} 1282 1283/** 1284 * __rpcrdma_regbuf_dma_map - DMA-map a regbuf 1285 * @r_xprt: controlling transport instance 1286 * @rb: regbuf to be mapped 1287 * 1288 * Returns true if the buffer is now DMA mapped to @r_xprt's device 1289 */ 1290bool __rpcrdma_regbuf_dma_map(struct rpcrdma_xprt *r_xprt, 1291 struct rpcrdma_regbuf *rb) 1292{ 1293 struct ib_device *device = r_xprt->rx_ep->re_id->device; 1294 1295 if (rb->rg_direction == DMA_NONE) 1296 return false; 1297 1298 rb->rg_iov.addr = ib_dma_map_single(device, rdmab_data(rb), 1299 rdmab_length(rb), rb->rg_direction); 1300 if (ib_dma_mapping_error(device, rdmab_addr(rb))) { 1301 trace_xprtrdma_dma_maperr(rdmab_addr(rb)); 1302 return false; 1303 } 1304 1305 rb->rg_device = device; 1306 rb->rg_iov.lkey = r_xprt->rx_ep->re_pd->local_dma_lkey; 1307 return true; 1308} 1309 1310static void rpcrdma_regbuf_dma_unmap(struct rpcrdma_regbuf *rb) 1311{ 1312 if (!rb) 1313 return; 1314 1315 if (!rpcrdma_regbuf_is_mapped(rb)) 1316 return; 1317 1318 ib_dma_unmap_single(rb->rg_device, rdmab_addr(rb), rdmab_length(rb), 1319 rb->rg_direction); 1320 rb->rg_device = NULL; 1321} 1322 1323static void rpcrdma_regbuf_free(struct rpcrdma_regbuf *rb) 1324{ 1325 rpcrdma_regbuf_dma_unmap(rb); 1326 if (rb) 1327 kfree(rb->rg_data); 1328 kfree(rb); 1329} 1330 1331/** 1332 * rpcrdma_post_recvs - Refill the Receive Queue 1333 * @r_xprt: controlling transport instance 1334 * @needed: current credit grant 1335 * @temp: mark Receive buffers to be deleted after one use 1336 * 1337 */ 1338void rpcrdma_post_recvs(struct rpcrdma_xprt *r_xprt, int needed, bool temp) 1339{ 1340 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1341 struct rpcrdma_ep *ep = r_xprt->rx_ep; 1342 struct ib_recv_wr *wr, *bad_wr; 1343 struct rpcrdma_rep *rep; 1344 int count, rc; 1345 1346 rc = 0; 1347 count = 0; 1348 1349 if (likely(ep->re_receive_count > needed)) 1350 goto out; 1351 needed -= ep->re_receive_count; 1352 if (!temp) 1353 needed += RPCRDMA_MAX_RECV_BATCH; 1354 1355 if (atomic_inc_return(&ep->re_receiving) > 1) 1356 goto out; 1357 1358 /* fast path: all needed reps can be found on the free list */ 1359 wr = NULL; 1360 while (needed) { 1361 rep = rpcrdma_rep_get_locked(buf); 1362 if (rep && rep->rr_temp) { 1363 rpcrdma_rep_destroy(rep); 1364 continue; 1365 } 1366 if (!rep) 1367 rep = rpcrdma_rep_create(r_xprt, temp); 1368 if (!rep) 1369 break; 1370 1371 rep->rr_cid.ci_queue_id = ep->re_attr.recv_cq->res.id; 1372 trace_xprtrdma_post_recv(rep); 1373 rep->rr_recv_wr.next = wr; 1374 wr = &rep->rr_recv_wr; 1375 --needed; 1376 ++count; 1377 } 1378 if (!wr) 1379 goto out; 1380 1381 rc = ib_post_recv(ep->re_id->qp, wr, 1382 (const struct ib_recv_wr **)&bad_wr); 1383 if (rc) { 1384 trace_xprtrdma_post_recvs_err(r_xprt, rc); 1385 for (wr = bad_wr; wr;) { 1386 struct rpcrdma_rep *rep; 1387 1388 rep = container_of(wr, struct rpcrdma_rep, rr_recv_wr); 1389 wr = wr->next; 1390 rpcrdma_rep_put(buf, rep); 1391 --count; 1392 } 1393 } 1394 if (atomic_dec_return(&ep->re_receiving) > 0) 1395 complete(&ep->re_done); 1396 1397out: 1398 trace_xprtrdma_post_recvs(r_xprt, count); 1399 ep->re_receive_count += count; 1400 return; 1401}