1// SPDX-License-Identifier: GPL-2.0 2/* 3 * Copyright (c) 2015, 2017 Oracle. All rights reserved. 4 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved. 5 */ 6 7/* Lightweight memory registration using Fast Registration Work 8 * Requests (FRWR). 9 * 10 * FRWR features ordered asynchronous registration and invalidation 11 * of arbitrarily-sized memory regions. This is the fastest and safest 12 * but most complex memory registration mode. 13 */ 14 15/* Normal operation 16 * 17 * A Memory Region is prepared for RDMA Read or Write using a FAST_REG 18 * Work Request (frwr_map). When the RDMA operation is finished, this 19 * Memory Region is invalidated using a LOCAL_INV Work Request 20 * (frwr_unmap_async and frwr_unmap_sync). 21 * 22 * Typically FAST_REG Work Requests are not signaled, and neither are 23 * RDMA Send Work Requests (with the exception of signaling occasionally 24 * to prevent provider work queue overflows). This greatly reduces HCA 25 * interrupt workload. 26 */ 27 28/* Transport recovery 29 * 30 * frwr_map and frwr_unmap_* cannot run at the same time the transport 31 * connect worker is running. The connect worker holds the transport 32 * send lock, just as ->send_request does. This prevents frwr_map and 33 * the connect worker from running concurrently. When a connection is 34 * closed, the Receive completion queue is drained before the allowing 35 * the connect worker to get control. This prevents frwr_unmap and the 36 * connect worker from running concurrently. 37 * 38 * When the underlying transport disconnects, MRs that are in flight 39 * are flushed and are likely unusable. Thus all MRs are destroyed. 40 * New MRs are created on demand. 41 */ 42 43#include <linux/sunrpc/svc_rdma.h> 44 45#include "xprt_rdma.h" 46#include <trace/events/rpcrdma.h> 47 48#if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 49# define RPCDBG_FACILITY RPCDBG_TRANS 50#endif 51 52/** 53 * frwr_release_mr - Destroy one MR 54 * @mr: MR allocated by frwr_mr_init 55 * 56 */ 57void frwr_release_mr(struct rpcrdma_mr *mr) 58{ 59 int rc; 60 61 rc = ib_dereg_mr(mr->frwr.fr_mr); 62 if (rc) 63 trace_xprtrdma_frwr_dereg(mr, rc); 64 kfree(mr->mr_sg); 65 kfree(mr); 66} 67 68static void frwr_mr_recycle(struct rpcrdma_mr *mr) 69{ 70 struct rpcrdma_xprt *r_xprt = mr->mr_xprt; 71 72 trace_xprtrdma_mr_recycle(mr); 73 74 if (mr->mr_dir != DMA_NONE) { 75 trace_xprtrdma_mr_unmap(mr); 76 ib_dma_unmap_sg(r_xprt->rx_ep->re_id->device, 77 mr->mr_sg, mr->mr_nents, mr->mr_dir); 78 mr->mr_dir = DMA_NONE; 79 } 80 81 spin_lock(&r_xprt->rx_buf.rb_lock); 82 list_del(&mr->mr_all); 83 r_xprt->rx_stats.mrs_recycled++; 84 spin_unlock(&r_xprt->rx_buf.rb_lock); 85 86 frwr_release_mr(mr); 87} 88 89/* frwr_reset - Place MRs back on the free list 90 * @req: request to reset 91 * 92 * Used after a failed marshal. For FRWR, this means the MRs 93 * don't have to be fully released and recreated. 94 * 95 * NB: This is safe only as long as none of @req's MRs are 96 * involved with an ongoing asynchronous FAST_REG or LOCAL_INV 97 * Work Request. 98 */ 99void frwr_reset(struct rpcrdma_req *req) 100{ 101 struct rpcrdma_mr *mr; 102 103 while ((mr = rpcrdma_mr_pop(&req->rl_registered))) 104 rpcrdma_mr_put(mr); 105} 106 107/** 108 * frwr_mr_init - Initialize one MR 109 * @r_xprt: controlling transport instance 110 * @mr: generic MR to prepare for FRWR 111 * 112 * Returns zero if successful. Otherwise a negative errno 113 * is returned. 114 */ 115int frwr_mr_init(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mr *mr) 116{ 117 struct rpcrdma_ep *ep = r_xprt->rx_ep; 118 unsigned int depth = ep->re_max_fr_depth; 119 struct scatterlist *sg; 120 struct ib_mr *frmr; 121 int rc; 122 123 frmr = ib_alloc_mr(ep->re_pd, ep->re_mrtype, depth); 124 if (IS_ERR(frmr)) 125 goto out_mr_err; 126 127 sg = kmalloc_array(depth, sizeof(*sg), GFP_NOFS); 128 if (!sg) 129 goto out_list_err; 130 131 mr->mr_xprt = r_xprt; 132 mr->frwr.fr_mr = frmr; 133 mr->mr_dir = DMA_NONE; 134 INIT_LIST_HEAD(&mr->mr_list); 135 init_completion(&mr->frwr.fr_linv_done); 136 137 sg_init_table(sg, depth); 138 mr->mr_sg = sg; 139 return 0; 140 141out_mr_err: 142 rc = PTR_ERR(frmr); 143 trace_xprtrdma_frwr_alloc(mr, rc); 144 return rc; 145 146out_list_err: 147 ib_dereg_mr(frmr); 148 return -ENOMEM; 149} 150 151/** 152 * frwr_query_device - Prepare a transport for use with FRWR 153 * @ep: endpoint to fill in 154 * @device: RDMA device to query 155 * 156 * On success, sets: 157 * ep->re_attr 158 * ep->re_max_requests 159 * ep->re_max_rdma_segs 160 * ep->re_max_fr_depth 161 * ep->re_mrtype 162 * 163 * Return values: 164 * On success, returns zero. 165 * %-EINVAL - the device does not support FRWR memory registration 166 * %-ENOMEM - the device is not sufficiently capable for NFS/RDMA 167 */ 168int frwr_query_device(struct rpcrdma_ep *ep, const struct ib_device *device) 169{ 170 const struct ib_device_attr *attrs = &device->attrs; 171 int max_qp_wr, depth, delta; 172 unsigned int max_sge; 173 174 if (!(attrs->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS) || 175 attrs->max_fast_reg_page_list_len == 0) { 176 pr_err("rpcrdma: 'frwr' mode is not supported by device %s\n", 177 device->name); 178 return -EINVAL; 179 } 180 181 max_sge = min_t(unsigned int, attrs->max_send_sge, 182 RPCRDMA_MAX_SEND_SGES); 183 if (max_sge < RPCRDMA_MIN_SEND_SGES) { 184 pr_err("rpcrdma: HCA provides only %u send SGEs\n", max_sge); 185 return -ENOMEM; 186 } 187 ep->re_attr.cap.max_send_sge = max_sge; 188 ep->re_attr.cap.max_recv_sge = 1; 189 190 ep->re_mrtype = IB_MR_TYPE_MEM_REG; 191 if (attrs->device_cap_flags & IB_DEVICE_SG_GAPS_REG) 192 ep->re_mrtype = IB_MR_TYPE_SG_GAPS; 193 194 /* Quirk: Some devices advertise a large max_fast_reg_page_list_len 195 * capability, but perform optimally when the MRs are not larger 196 * than a page. 197 */ 198 if (attrs->max_sge_rd > RPCRDMA_MAX_HDR_SEGS) 199 ep->re_max_fr_depth = attrs->max_sge_rd; 200 else 201 ep->re_max_fr_depth = attrs->max_fast_reg_page_list_len; 202 if (ep->re_max_fr_depth > RPCRDMA_MAX_DATA_SEGS) 203 ep->re_max_fr_depth = RPCRDMA_MAX_DATA_SEGS; 204 205 /* Add room for frwr register and invalidate WRs. 206 * 1. FRWR reg WR for head 207 * 2. FRWR invalidate WR for head 208 * 3. N FRWR reg WRs for pagelist 209 * 4. N FRWR invalidate WRs for pagelist 210 * 5. FRWR reg WR for tail 211 * 6. FRWR invalidate WR for tail 212 * 7. The RDMA_SEND WR 213 */ 214 depth = 7; 215 216 /* Calculate N if the device max FRWR depth is smaller than 217 * RPCRDMA_MAX_DATA_SEGS. 218 */ 219 if (ep->re_max_fr_depth < RPCRDMA_MAX_DATA_SEGS) { 220 delta = RPCRDMA_MAX_DATA_SEGS - ep->re_max_fr_depth; 221 do { 222 depth += 2; /* FRWR reg + invalidate */ 223 delta -= ep->re_max_fr_depth; 224 } while (delta > 0); 225 } 226 227 max_qp_wr = attrs->max_qp_wr; 228 max_qp_wr -= RPCRDMA_BACKWARD_WRS; 229 max_qp_wr -= 1; 230 if (max_qp_wr < RPCRDMA_MIN_SLOT_TABLE) 231 return -ENOMEM; 232 if (ep->re_max_requests > max_qp_wr) 233 ep->re_max_requests = max_qp_wr; 234 ep->re_attr.cap.max_send_wr = ep->re_max_requests * depth; 235 if (ep->re_attr.cap.max_send_wr > max_qp_wr) { 236 ep->re_max_requests = max_qp_wr / depth; 237 if (!ep->re_max_requests) 238 return -ENOMEM; 239 ep->re_attr.cap.max_send_wr = ep->re_max_requests * depth; 240 } 241 ep->re_attr.cap.max_send_wr += RPCRDMA_BACKWARD_WRS; 242 ep->re_attr.cap.max_send_wr += 1; /* for ib_drain_sq */ 243 ep->re_attr.cap.max_recv_wr = ep->re_max_requests; 244 ep->re_attr.cap.max_recv_wr += RPCRDMA_BACKWARD_WRS; 245 ep->re_attr.cap.max_recv_wr += RPCRDMA_MAX_RECV_BATCH; 246 ep->re_attr.cap.max_recv_wr += 1; /* for ib_drain_rq */ 247 248 ep->re_max_rdma_segs = 249 DIV_ROUND_UP(RPCRDMA_MAX_DATA_SEGS, ep->re_max_fr_depth); 250 /* Reply chunks require segments for head and tail buffers */ 251 ep->re_max_rdma_segs += 2; 252 if (ep->re_max_rdma_segs > RPCRDMA_MAX_HDR_SEGS) 253 ep->re_max_rdma_segs = RPCRDMA_MAX_HDR_SEGS; 254 255 /* Ensure the underlying device is capable of conveying the 256 * largest r/wsize NFS will ask for. This guarantees that 257 * failing over from one RDMA device to another will not 258 * break NFS I/O. 259 */ 260 if ((ep->re_max_rdma_segs * ep->re_max_fr_depth) < RPCRDMA_MAX_SEGS) 261 return -ENOMEM; 262 263 return 0; 264} 265 266/** 267 * frwr_map - Register a memory region 268 * @r_xprt: controlling transport 269 * @seg: memory region co-ordinates 270 * @nsegs: number of segments remaining 271 * @writing: true when RDMA Write will be used 272 * @xid: XID of RPC using the registered memory 273 * @mr: MR to fill in 274 * 275 * Prepare a REG_MR Work Request to register a memory region 276 * for remote access via RDMA READ or RDMA WRITE. 277 * 278 * Returns the next segment or a negative errno pointer. 279 * On success, @mr is filled in. 280 */ 281struct rpcrdma_mr_seg *frwr_map(struct rpcrdma_xprt *r_xprt, 282 struct rpcrdma_mr_seg *seg, 283 int nsegs, bool writing, __be32 xid, 284 struct rpcrdma_mr *mr) 285{ 286 struct rpcrdma_ep *ep = r_xprt->rx_ep; 287 struct ib_reg_wr *reg_wr; 288 int i, n, dma_nents; 289 struct ib_mr *ibmr; 290 u8 key; 291 292 if (nsegs > ep->re_max_fr_depth) 293 nsegs = ep->re_max_fr_depth; 294 for (i = 0; i < nsegs;) { 295 if (seg->mr_page) 296 sg_set_page(&mr->mr_sg[i], 297 seg->mr_page, 298 seg->mr_len, 299 offset_in_page(seg->mr_offset)); 300 else 301 sg_set_buf(&mr->mr_sg[i], seg->mr_offset, 302 seg->mr_len); 303 304 ++seg; 305 ++i; 306 if (ep->re_mrtype == IB_MR_TYPE_SG_GAPS) 307 continue; 308 if ((i < nsegs && offset_in_page(seg->mr_offset)) || 309 offset_in_page((seg-1)->mr_offset + (seg-1)->mr_len)) 310 break; 311 } 312 mr->mr_dir = rpcrdma_data_dir(writing); 313 mr->mr_nents = i; 314 315 dma_nents = ib_dma_map_sg(ep->re_id->device, mr->mr_sg, mr->mr_nents, 316 mr->mr_dir); 317 if (!dma_nents) 318 goto out_dmamap_err; 319 320 ibmr = mr->frwr.fr_mr; 321 n = ib_map_mr_sg(ibmr, mr->mr_sg, dma_nents, NULL, PAGE_SIZE); 322 if (n != dma_nents) 323 goto out_mapmr_err; 324 325 ibmr->iova &= 0x00000000ffffffff; 326 ibmr->iova |= ((u64)be32_to_cpu(xid)) << 32; 327 key = (u8)(ibmr->rkey & 0x000000FF); 328 ib_update_fast_reg_key(ibmr, ++key); 329 330 reg_wr = &mr->frwr.fr_regwr; 331 reg_wr->mr = ibmr; 332 reg_wr->key = ibmr->rkey; 333 reg_wr->access = writing ? 334 IB_ACCESS_REMOTE_WRITE | IB_ACCESS_LOCAL_WRITE : 335 IB_ACCESS_REMOTE_READ; 336 337 mr->mr_handle = ibmr->rkey; 338 mr->mr_length = ibmr->length; 339 mr->mr_offset = ibmr->iova; 340 trace_xprtrdma_mr_map(mr); 341 342 return seg; 343 344out_dmamap_err: 345 mr->mr_dir = DMA_NONE; 346 trace_xprtrdma_frwr_sgerr(mr, i); 347 return ERR_PTR(-EIO); 348 349out_mapmr_err: 350 trace_xprtrdma_frwr_maperr(mr, n); 351 return ERR_PTR(-EIO); 352} 353 354/** 355 * frwr_wc_fastreg - Invoked by RDMA provider for a flushed FastReg WC 356 * @cq: completion queue 357 * @wc: WCE for a completed FastReg WR 358 * 359 */ 360static void frwr_wc_fastreg(struct ib_cq *cq, struct ib_wc *wc) 361{ 362 struct ib_cqe *cqe = wc->wr_cqe; 363 struct rpcrdma_frwr *frwr = 364 container_of(cqe, struct rpcrdma_frwr, fr_cqe); 365 366 /* WARNING: Only wr_cqe and status are reliable at this point */ 367 trace_xprtrdma_wc_fastreg(wc, frwr); 368 /* The MR will get recycled when the associated req is retransmitted */ 369 370 rpcrdma_flush_disconnect(cq->cq_context, wc); 371} 372 373/** 374 * frwr_send - post Send WRs containing the RPC Call message 375 * @r_xprt: controlling transport instance 376 * @req: prepared RPC Call 377 * 378 * For FRWR, chain any FastReg WRs to the Send WR. Only a 379 * single ib_post_send call is needed to register memory 380 * and then post the Send WR. 381 * 382 * Returns the return code from ib_post_send. 383 * 384 * Caller must hold the transport send lock to ensure that the 385 * pointers to the transport's rdma_cm_id and QP are stable. 386 */ 387int frwr_send(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req) 388{ 389 struct ib_send_wr *post_wr; 390 struct rpcrdma_mr *mr; 391 392 post_wr = &req->rl_wr; 393 list_for_each_entry(mr, &req->rl_registered, mr_list) { 394 struct rpcrdma_frwr *frwr; 395 396 frwr = &mr->frwr; 397 398 frwr->fr_cqe.done = frwr_wc_fastreg; 399 frwr->fr_regwr.wr.next = post_wr; 400 frwr->fr_regwr.wr.wr_cqe = &frwr->fr_cqe; 401 frwr->fr_regwr.wr.num_sge = 0; 402 frwr->fr_regwr.wr.opcode = IB_WR_REG_MR; 403 frwr->fr_regwr.wr.send_flags = 0; 404 405 post_wr = &frwr->fr_regwr.wr; 406 } 407 408 return ib_post_send(r_xprt->rx_ep->re_id->qp, post_wr, NULL); 409} 410 411/** 412 * frwr_reminv - handle a remotely invalidated mr on the @mrs list 413 * @rep: Received reply 414 * @mrs: list of MRs to check 415 * 416 */ 417void frwr_reminv(struct rpcrdma_rep *rep, struct list_head *mrs) 418{ 419 struct rpcrdma_mr *mr; 420 421 list_for_each_entry(mr, mrs, mr_list) 422 if (mr->mr_handle == rep->rr_inv_rkey) { 423 list_del_init(&mr->mr_list); 424 trace_xprtrdma_mr_reminv(mr); 425 rpcrdma_mr_put(mr); 426 break; /* only one invalidated MR per RPC */ 427 } 428} 429 430static void __frwr_release_mr(struct ib_wc *wc, struct rpcrdma_mr *mr) 431{ 432 if (wc->status != IB_WC_SUCCESS) 433 frwr_mr_recycle(mr); 434 else 435 rpcrdma_mr_put(mr); 436} 437 438/** 439 * frwr_wc_localinv - Invoked by RDMA provider for a LOCAL_INV WC 440 * @cq: completion queue 441 * @wc: WCE for a completed LocalInv WR 442 * 443 */ 444static void frwr_wc_localinv(struct ib_cq *cq, struct ib_wc *wc) 445{ 446 struct ib_cqe *cqe = wc->wr_cqe; 447 struct rpcrdma_frwr *frwr = 448 container_of(cqe, struct rpcrdma_frwr, fr_cqe); 449 struct rpcrdma_mr *mr = container_of(frwr, struct rpcrdma_mr, frwr); 450 451 /* WARNING: Only wr_cqe and status are reliable at this point */ 452 trace_xprtrdma_wc_li(wc, frwr); 453 __frwr_release_mr(wc, mr); 454 455 rpcrdma_flush_disconnect(cq->cq_context, wc); 456} 457 458/** 459 * frwr_wc_localinv_wake - Invoked by RDMA provider for a LOCAL_INV WC 460 * @cq: completion queue 461 * @wc: WCE for a completed LocalInv WR 462 * 463 * Awaken anyone waiting for an MR to finish being fenced. 464 */ 465static void frwr_wc_localinv_wake(struct ib_cq *cq, struct ib_wc *wc) 466{ 467 struct ib_cqe *cqe = wc->wr_cqe; 468 struct rpcrdma_frwr *frwr = 469 container_of(cqe, struct rpcrdma_frwr, fr_cqe); 470 struct rpcrdma_mr *mr = container_of(frwr, struct rpcrdma_mr, frwr); 471 472 /* WARNING: Only wr_cqe and status are reliable at this point */ 473 trace_xprtrdma_wc_li_wake(wc, frwr); 474 __frwr_release_mr(wc, mr); 475 complete(&frwr->fr_linv_done); 476 477 rpcrdma_flush_disconnect(cq->cq_context, wc); 478} 479 480/** 481 * frwr_unmap_sync - invalidate memory regions that were registered for @req 482 * @r_xprt: controlling transport instance 483 * @req: rpcrdma_req with a non-empty list of MRs to process 484 * 485 * Sleeps until it is safe for the host CPU to access the previously mapped 486 * memory regions. This guarantees that registered MRs are properly fenced 487 * from the server before the RPC consumer accesses the data in them. It 488 * also ensures proper Send flow control: waking the next RPC waits until 489 * this RPC has relinquished all its Send Queue entries. 490 */ 491void frwr_unmap_sync(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req) 492{ 493 struct ib_send_wr *first, **prev, *last; 494 const struct ib_send_wr *bad_wr; 495 struct rpcrdma_frwr *frwr; 496 struct rpcrdma_mr *mr; 497 int rc; 498 499 /* ORDER: Invalidate all of the MRs first 500 * 501 * Chain the LOCAL_INV Work Requests and post them with 502 * a single ib_post_send() call. 503 */ 504 frwr = NULL; 505 prev = &first; 506 while ((mr = rpcrdma_mr_pop(&req->rl_registered))) { 507 508 trace_xprtrdma_mr_localinv(mr); 509 r_xprt->rx_stats.local_inv_needed++; 510 511 frwr = &mr->frwr; 512 frwr->fr_cqe.done = frwr_wc_localinv; 513 last = &frwr->fr_invwr; 514 last->next = NULL; 515 last->wr_cqe = &frwr->fr_cqe; 516 last->sg_list = NULL; 517 last->num_sge = 0; 518 last->opcode = IB_WR_LOCAL_INV; 519 last->send_flags = IB_SEND_SIGNALED; 520 last->ex.invalidate_rkey = mr->mr_handle; 521 522 *prev = last; 523 prev = &last->next; 524 } 525 526 /* Strong send queue ordering guarantees that when the 527 * last WR in the chain completes, all WRs in the chain 528 * are complete. 529 */ 530 frwr->fr_cqe.done = frwr_wc_localinv_wake; 531 reinit_completion(&frwr->fr_linv_done); 532 533 /* Transport disconnect drains the receive CQ before it 534 * replaces the QP. The RPC reply handler won't call us 535 * unless re_id->qp is a valid pointer. 536 */ 537 bad_wr = NULL; 538 rc = ib_post_send(r_xprt->rx_ep->re_id->qp, first, &bad_wr); 539 540 /* The final LOCAL_INV WR in the chain is supposed to 541 * do the wake. If it was never posted, the wake will 542 * not happen, so don't wait in that case. 543 */ 544 if (bad_wr != first) 545 wait_for_completion(&frwr->fr_linv_done); 546 if (!rc) 547 return; 548 549 /* Recycle MRs in the LOCAL_INV chain that did not get posted. 550 */ 551 trace_xprtrdma_post_linv(req, rc); 552 while (bad_wr) { 553 frwr = container_of(bad_wr, struct rpcrdma_frwr, 554 fr_invwr); 555 mr = container_of(frwr, struct rpcrdma_mr, frwr); 556 bad_wr = bad_wr->next; 557 558 frwr_mr_recycle(mr); 559 } 560} 561 562/** 563 * frwr_wc_localinv_done - Invoked by RDMA provider for a signaled LOCAL_INV WC 564 * @cq: completion queue 565 * @wc: WCE for a completed LocalInv WR 566 * 567 */ 568static void frwr_wc_localinv_done(struct ib_cq *cq, struct ib_wc *wc) 569{ 570 struct ib_cqe *cqe = wc->wr_cqe; 571 struct rpcrdma_frwr *frwr = 572 container_of(cqe, struct rpcrdma_frwr, fr_cqe); 573 struct rpcrdma_mr *mr = container_of(frwr, struct rpcrdma_mr, frwr); 574 struct rpcrdma_rep *rep = mr->mr_req->rl_reply; 575 576 /* WARNING: Only wr_cqe and status are reliable at this point */ 577 trace_xprtrdma_wc_li_done(wc, frwr); 578 __frwr_release_mr(wc, mr); 579 580 /* Ensure @rep is generated before __frwr_release_mr */ 581 smp_rmb(); 582 rpcrdma_complete_rqst(rep); 583 584 rpcrdma_flush_disconnect(cq->cq_context, wc); 585} 586 587/** 588 * frwr_unmap_async - invalidate memory regions that were registered for @req 589 * @r_xprt: controlling transport instance 590 * @req: rpcrdma_req with a non-empty list of MRs to process 591 * 592 * This guarantees that registered MRs are properly fenced from the 593 * server before the RPC consumer accesses the data in them. It also 594 * ensures proper Send flow control: waking the next RPC waits until 595 * this RPC has relinquished all its Send Queue entries. 596 */ 597void frwr_unmap_async(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req) 598{ 599 struct ib_send_wr *first, *last, **prev; 600 const struct ib_send_wr *bad_wr; 601 struct rpcrdma_frwr *frwr; 602 struct rpcrdma_mr *mr; 603 int rc; 604 605 /* Chain the LOCAL_INV Work Requests and post them with 606 * a single ib_post_send() call. 607 */ 608 frwr = NULL; 609 prev = &first; 610 while ((mr = rpcrdma_mr_pop(&req->rl_registered))) { 611 612 trace_xprtrdma_mr_localinv(mr); 613 r_xprt->rx_stats.local_inv_needed++; 614 615 frwr = &mr->frwr; 616 frwr->fr_cqe.done = frwr_wc_localinv; 617 last = &frwr->fr_invwr; 618 last->next = NULL; 619 last->wr_cqe = &frwr->fr_cqe; 620 last->sg_list = NULL; 621 last->num_sge = 0; 622 last->opcode = IB_WR_LOCAL_INV; 623 last->send_flags = IB_SEND_SIGNALED; 624 last->ex.invalidate_rkey = mr->mr_handle; 625 626 *prev = last; 627 prev = &last->next; 628 } 629 630 /* Strong send queue ordering guarantees that when the 631 * last WR in the chain completes, all WRs in the chain 632 * are complete. The last completion will wake up the 633 * RPC waiter. 634 */ 635 frwr->fr_cqe.done = frwr_wc_localinv_done; 636 637 /* Transport disconnect drains the receive CQ before it 638 * replaces the QP. The RPC reply handler won't call us 639 * unless re_id->qp is a valid pointer. 640 */ 641 bad_wr = NULL; 642 rc = ib_post_send(r_xprt->rx_ep->re_id->qp, first, &bad_wr); 643 if (!rc) 644 return; 645 646 /* Recycle MRs in the LOCAL_INV chain that did not get posted. 647 */ 648 trace_xprtrdma_post_linv(req, rc); 649 while (bad_wr) { 650 frwr = container_of(bad_wr, struct rpcrdma_frwr, fr_invwr); 651 mr = container_of(frwr, struct rpcrdma_mr, frwr); 652 bad_wr = bad_wr->next; 653 654 frwr_mr_recycle(mr); 655 } 656 657 /* The final LOCAL_INV WR in the chain is supposed to 658 * do the wake. If it was never posted, the wake will 659 * not happen, so wake here in that case. 660 */ 661 rpcrdma_complete_rqst(req->rl_reply); 662} 663