1 /* 2 * Copyright (c) 2014 Open Grid Computing, Inc. All rights reserved. 3 * Copyright (c) 2005-2006 Network Appliance, Inc. All rights reserved. 4 * 5 * This software is available to you under a choice of one of two 6 * licenses. You may choose to be licensed under the terms of the GNU 7 * General Public License (GPL) Version 2, available from the file 8 * COPYING in the main directory of this source tree, or the BSD-type 9 * license below: 10 * 11 * Redistribution and use in source and binary forms, with or without 12 * modification, are permitted provided that the following conditions 13 * are met: 14 * 15 * Redistributions of source code must retain the above copyright 16 * notice, this list of conditions and the following disclaimer. 17 * 18 * Redistributions in binary form must reproduce the above 19 * copyright notice, this list of conditions and the following 20 * disclaimer in the documentation and/or other materials provided 21 * with the distribution. 22 * 23 * Neither the name of the Network Appliance, Inc. nor the names of 24 * its contributors may be used to endorse or promote products 25 * derived from this software without specific prior written 26 * permission. 27 * 28 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 29 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 30 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 31 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 32 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 33 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT 34 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 35 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 36 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 37 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 38 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 39 * 40 * Author: Tom Tucker <tom@opengridcomputing.com> 41 */ 42 43 #include <linux/sunrpc/debug.h> 44 #include <linux/sunrpc/rpc_rdma.h> 45 #include <linux/spinlock.h> 46 #include <asm/unaligned.h> 47 #include <rdma/ib_verbs.h> 48 #include <rdma/rdma_cm.h> 49 #include <linux/sunrpc/svc_rdma.h> 50 51 #define RPCDBG_FACILITY RPCDBG_SVCXPRT 52 53 static u32 xdr_padsize(u32 len) 54 { 55 return (len & 3) ? (4 - (len & 3)) : 0; 56 } 57 58 int svc_rdma_map_xdr(struct svcxprt_rdma *xprt, 59 struct xdr_buf *xdr, 60 struct svc_rdma_req_map *vec, 61 bool write_chunk_present) 62 { 63 int sge_no; 64 u32 sge_bytes; 65 u32 page_bytes; 66 u32 page_off; 67 int page_no; 68 69 if (xdr->len != 70 (xdr->head[0].iov_len + xdr->page_len + xdr->tail[0].iov_len)) { 71 pr_err("svcrdma: %s: XDR buffer length error\n", __func__); 72 return -EIO; 73 } 74 75 /* Skip the first sge, this is for the RPCRDMA header */ 76 sge_no = 1; 77 78 /* Head SGE */ 79 vec->sge[sge_no].iov_base = xdr->head[0].iov_base; 80 vec->sge[sge_no].iov_len = xdr->head[0].iov_len; 81 sge_no++; 82 83 /* pages SGE */ 84 page_no = 0; 85 page_bytes = xdr->page_len; 86 page_off = xdr->page_base; 87 while (page_bytes) { 88 vec->sge[sge_no].iov_base = 89 page_address(xdr->pages[page_no]) + page_off; 90 sge_bytes = min_t(u32, page_bytes, (PAGE_SIZE - page_off)); 91 page_bytes -= sge_bytes; 92 vec->sge[sge_no].iov_len = sge_bytes; 93 94 sge_no++; 95 page_no++; 96 page_off = 0; /* reset for next time through loop */ 97 } 98 99 /* Tail SGE */ 100 if (xdr->tail[0].iov_len) { 101 unsigned char *base = xdr->tail[0].iov_base; 102 size_t len = xdr->tail[0].iov_len; 103 u32 xdr_pad = xdr_padsize(xdr->page_len); 104 105 if (write_chunk_present && xdr_pad) { 106 base += xdr_pad; 107 len -= xdr_pad; 108 } 109 110 if (len) { 111 vec->sge[sge_no].iov_base = base; 112 vec->sge[sge_no].iov_len = len; 113 sge_no++; 114 } 115 } 116 117 dprintk("svcrdma: %s: sge_no %d page_no %d " 118 "page_base %u page_len %u head_len %zu tail_len %zu\n", 119 __func__, sge_no, page_no, xdr->page_base, xdr->page_len, 120 xdr->head[0].iov_len, xdr->tail[0].iov_len); 121 122 vec->count = sge_no; 123 return 0; 124 } 125 126 static dma_addr_t dma_map_xdr(struct svcxprt_rdma *xprt, 127 struct xdr_buf *xdr, 128 u32 xdr_off, size_t len, int dir) 129 { 130 struct page *page; 131 dma_addr_t dma_addr; 132 if (xdr_off < xdr->head[0].iov_len) { 133 /* This offset is in the head */ 134 xdr_off += (unsigned long)xdr->head[0].iov_base & ~PAGE_MASK; 135 page = virt_to_page(xdr->head[0].iov_base); 136 } else { 137 xdr_off -= xdr->head[0].iov_len; 138 if (xdr_off < xdr->page_len) { 139 /* This offset is in the page list */ 140 xdr_off += xdr->page_base; 141 page = xdr->pages[xdr_off >> PAGE_SHIFT]; 142 xdr_off &= ~PAGE_MASK; 143 } else { 144 /* This offset is in the tail */ 145 xdr_off -= xdr->page_len; 146 xdr_off += (unsigned long) 147 xdr->tail[0].iov_base & ~PAGE_MASK; 148 page = virt_to_page(xdr->tail[0].iov_base); 149 } 150 } 151 dma_addr = ib_dma_map_page(xprt->sc_cm_id->device, page, xdr_off, 152 min_t(size_t, PAGE_SIZE, len), dir); 153 return dma_addr; 154 } 155 156 /* Parse the RPC Call's transport header. 157 */ 158 static void svc_rdma_get_write_arrays(struct rpcrdma_msg *rmsgp, 159 struct rpcrdma_write_array **write, 160 struct rpcrdma_write_array **reply) 161 { 162 __be32 *p; 163 164 p = (__be32 *)&rmsgp->rm_body.rm_chunks[0]; 165 166 /* Read list */ 167 while (*p++ != xdr_zero) 168 p += 5; 169 170 /* Write list */ 171 if (*p != xdr_zero) { 172 *write = (struct rpcrdma_write_array *)p; 173 while (*p++ != xdr_zero) 174 p += 1 + be32_to_cpu(*p) * 4; 175 } else { 176 *write = NULL; 177 p++; 178 } 179 180 /* Reply chunk */ 181 if (*p != xdr_zero) 182 *reply = (struct rpcrdma_write_array *)p; 183 else 184 *reply = NULL; 185 } 186 187 /* RPC-over-RDMA Version One private extension: Remote Invalidation. 188 * Responder's choice: requester signals it can handle Send With 189 * Invalidate, and responder chooses one rkey to invalidate. 190 * 191 * Find a candidate rkey to invalidate when sending a reply. Picks the 192 * first rkey it finds in the chunks lists. 193 * 194 * Returns zero if RPC's chunk lists are empty. 195 */ 196 static u32 svc_rdma_get_inv_rkey(struct rpcrdma_msg *rdma_argp, 197 struct rpcrdma_write_array *wr_ary, 198 struct rpcrdma_write_array *rp_ary) 199 { 200 struct rpcrdma_read_chunk *rd_ary; 201 struct rpcrdma_segment *arg_ch; 202 203 rd_ary = (struct rpcrdma_read_chunk *)&rdma_argp->rm_body.rm_chunks[0]; 204 if (rd_ary->rc_discrim != xdr_zero) 205 return be32_to_cpu(rd_ary->rc_target.rs_handle); 206 207 if (wr_ary && be32_to_cpu(wr_ary->wc_nchunks)) { 208 arg_ch = &wr_ary->wc_array[0].wc_target; 209 return be32_to_cpu(arg_ch->rs_handle); 210 } 211 212 if (rp_ary && be32_to_cpu(rp_ary->wc_nchunks)) { 213 arg_ch = &rp_ary->wc_array[0].wc_target; 214 return be32_to_cpu(arg_ch->rs_handle); 215 } 216 217 return 0; 218 } 219 220 /* Assumptions: 221 * - The specified write_len can be represented in sc_max_sge * PAGE_SIZE 222 */ 223 static int send_write(struct svcxprt_rdma *xprt, struct svc_rqst *rqstp, 224 u32 rmr, u64 to, 225 u32 xdr_off, int write_len, 226 struct svc_rdma_req_map *vec) 227 { 228 struct ib_rdma_wr write_wr; 229 struct ib_sge *sge; 230 int xdr_sge_no; 231 int sge_no; 232 int sge_bytes; 233 int sge_off; 234 int bc; 235 struct svc_rdma_op_ctxt *ctxt; 236 237 if (vec->count > RPCSVC_MAXPAGES) { 238 pr_err("svcrdma: Too many pages (%lu)\n", vec->count); 239 return -EIO; 240 } 241 242 dprintk("svcrdma: RDMA_WRITE rmr=%x, to=%llx, xdr_off=%d, " 243 "write_len=%d, vec->sge=%p, vec->count=%lu\n", 244 rmr, (unsigned long long)to, xdr_off, 245 write_len, vec->sge, vec->count); 246 247 ctxt = svc_rdma_get_context(xprt); 248 ctxt->direction = DMA_TO_DEVICE; 249 sge = ctxt->sge; 250 251 /* Find the SGE associated with xdr_off */ 252 for (bc = xdr_off, xdr_sge_no = 1; bc && xdr_sge_no < vec->count; 253 xdr_sge_no++) { 254 if (vec->sge[xdr_sge_no].iov_len > bc) 255 break; 256 bc -= vec->sge[xdr_sge_no].iov_len; 257 } 258 259 sge_off = bc; 260 bc = write_len; 261 sge_no = 0; 262 263 /* Copy the remaining SGE */ 264 while (bc != 0) { 265 sge_bytes = min_t(size_t, 266 bc, vec->sge[xdr_sge_no].iov_len-sge_off); 267 sge[sge_no].length = sge_bytes; 268 sge[sge_no].addr = 269 dma_map_xdr(xprt, &rqstp->rq_res, xdr_off, 270 sge_bytes, DMA_TO_DEVICE); 271 xdr_off += sge_bytes; 272 if (ib_dma_mapping_error(xprt->sc_cm_id->device, 273 sge[sge_no].addr)) 274 goto err; 275 svc_rdma_count_mappings(xprt, ctxt); 276 sge[sge_no].lkey = xprt->sc_pd->local_dma_lkey; 277 ctxt->count++; 278 sge_off = 0; 279 sge_no++; 280 xdr_sge_no++; 281 if (xdr_sge_no > vec->count) { 282 pr_err("svcrdma: Too many sges (%d)\n", xdr_sge_no); 283 goto err; 284 } 285 bc -= sge_bytes; 286 if (sge_no == xprt->sc_max_sge) 287 break; 288 } 289 290 /* Prepare WRITE WR */ 291 memset(&write_wr, 0, sizeof write_wr); 292 ctxt->cqe.done = svc_rdma_wc_write; 293 write_wr.wr.wr_cqe = &ctxt->cqe; 294 write_wr.wr.sg_list = &sge[0]; 295 write_wr.wr.num_sge = sge_no; 296 write_wr.wr.opcode = IB_WR_RDMA_WRITE; 297 write_wr.wr.send_flags = IB_SEND_SIGNALED; 298 write_wr.rkey = rmr; 299 write_wr.remote_addr = to; 300 301 /* Post It */ 302 atomic_inc(&rdma_stat_write); 303 if (svc_rdma_send(xprt, &write_wr.wr)) 304 goto err; 305 return write_len - bc; 306 err: 307 svc_rdma_unmap_dma(ctxt); 308 svc_rdma_put_context(ctxt, 0); 309 return -EIO; 310 } 311 312 noinline 313 static int send_write_chunks(struct svcxprt_rdma *xprt, 314 struct rpcrdma_write_array *wr_ary, 315 struct rpcrdma_msg *rdma_resp, 316 struct svc_rqst *rqstp, 317 struct svc_rdma_req_map *vec) 318 { 319 u32 xfer_len = rqstp->rq_res.page_len; 320 int write_len; 321 u32 xdr_off; 322 int chunk_off; 323 int chunk_no; 324 int nchunks; 325 struct rpcrdma_write_array *res_ary; 326 int ret; 327 328 res_ary = (struct rpcrdma_write_array *) 329 &rdma_resp->rm_body.rm_chunks[1]; 330 331 /* Write chunks start at the pagelist */ 332 nchunks = be32_to_cpu(wr_ary->wc_nchunks); 333 for (xdr_off = rqstp->rq_res.head[0].iov_len, chunk_no = 0; 334 xfer_len && chunk_no < nchunks; 335 chunk_no++) { 336 struct rpcrdma_segment *arg_ch; 337 u64 rs_offset; 338 339 arg_ch = &wr_ary->wc_array[chunk_no].wc_target; 340 write_len = min(xfer_len, be32_to_cpu(arg_ch->rs_length)); 341 342 /* Prepare the response chunk given the length actually 343 * written */ 344 xdr_decode_hyper((__be32 *)&arg_ch->rs_offset, &rs_offset); 345 svc_rdma_xdr_encode_array_chunk(res_ary, chunk_no, 346 arg_ch->rs_handle, 347 arg_ch->rs_offset, 348 write_len); 349 chunk_off = 0; 350 while (write_len) { 351 ret = send_write(xprt, rqstp, 352 be32_to_cpu(arg_ch->rs_handle), 353 rs_offset + chunk_off, 354 xdr_off, 355 write_len, 356 vec); 357 if (ret <= 0) 358 goto out_err; 359 chunk_off += ret; 360 xdr_off += ret; 361 xfer_len -= ret; 362 write_len -= ret; 363 } 364 } 365 /* Update the req with the number of chunks actually used */ 366 svc_rdma_xdr_encode_write_list(rdma_resp, chunk_no); 367 368 return rqstp->rq_res.page_len; 369 370 out_err: 371 pr_err("svcrdma: failed to send write chunks, rc=%d\n", ret); 372 return -EIO; 373 } 374 375 noinline 376 static int send_reply_chunks(struct svcxprt_rdma *xprt, 377 struct rpcrdma_write_array *rp_ary, 378 struct rpcrdma_msg *rdma_resp, 379 struct svc_rqst *rqstp, 380 struct svc_rdma_req_map *vec) 381 { 382 u32 xfer_len = rqstp->rq_res.len; 383 int write_len; 384 u32 xdr_off; 385 int chunk_no; 386 int chunk_off; 387 int nchunks; 388 struct rpcrdma_segment *ch; 389 struct rpcrdma_write_array *res_ary; 390 int ret; 391 392 /* XXX: need to fix when reply lists occur with read-list and or 393 * write-list */ 394 res_ary = (struct rpcrdma_write_array *) 395 &rdma_resp->rm_body.rm_chunks[2]; 396 397 /* xdr offset starts at RPC message */ 398 nchunks = be32_to_cpu(rp_ary->wc_nchunks); 399 for (xdr_off = 0, chunk_no = 0; 400 xfer_len && chunk_no < nchunks; 401 chunk_no++) { 402 u64 rs_offset; 403 ch = &rp_ary->wc_array[chunk_no].wc_target; 404 write_len = min(xfer_len, be32_to_cpu(ch->rs_length)); 405 406 /* Prepare the reply chunk given the length actually 407 * written */ 408 xdr_decode_hyper((__be32 *)&ch->rs_offset, &rs_offset); 409 svc_rdma_xdr_encode_array_chunk(res_ary, chunk_no, 410 ch->rs_handle, ch->rs_offset, 411 write_len); 412 chunk_off = 0; 413 while (write_len) { 414 ret = send_write(xprt, rqstp, 415 be32_to_cpu(ch->rs_handle), 416 rs_offset + chunk_off, 417 xdr_off, 418 write_len, 419 vec); 420 if (ret <= 0) 421 goto out_err; 422 chunk_off += ret; 423 xdr_off += ret; 424 xfer_len -= ret; 425 write_len -= ret; 426 } 427 } 428 /* Update the req with the number of chunks actually used */ 429 svc_rdma_xdr_encode_reply_array(res_ary, chunk_no); 430 431 return rqstp->rq_res.len; 432 433 out_err: 434 pr_err("svcrdma: failed to send reply chunks, rc=%d\n", ret); 435 return -EIO; 436 } 437 438 /* This function prepares the portion of the RPCRDMA message to be 439 * sent in the RDMA_SEND. This function is called after data sent via 440 * RDMA has already been transmitted. There are three cases: 441 * - The RPCRDMA header, RPC header, and payload are all sent in a 442 * single RDMA_SEND. This is the "inline" case. 443 * - The RPCRDMA header and some portion of the RPC header and data 444 * are sent via this RDMA_SEND and another portion of the data is 445 * sent via RDMA. 446 * - The RPCRDMA header [NOMSG] is sent in this RDMA_SEND and the RPC 447 * header and data are all transmitted via RDMA. 448 * In all three cases, this function prepares the RPCRDMA header in 449 * sge[0], the 'type' parameter indicates the type to place in the 450 * RPCRDMA header, and the 'byte_count' field indicates how much of 451 * the XDR to include in this RDMA_SEND. NB: The offset of the payload 452 * to send is zero in the XDR. 453 */ 454 static int send_reply(struct svcxprt_rdma *rdma, 455 struct svc_rqst *rqstp, 456 struct page *page, 457 struct rpcrdma_msg *rdma_resp, 458 struct svc_rdma_req_map *vec, 459 int byte_count, 460 u32 inv_rkey) 461 { 462 struct svc_rdma_op_ctxt *ctxt; 463 struct ib_send_wr send_wr; 464 u32 xdr_off; 465 int sge_no; 466 int sge_bytes; 467 int page_no; 468 int pages; 469 int ret = -EIO; 470 471 /* Prepare the context */ 472 ctxt = svc_rdma_get_context(rdma); 473 ctxt->direction = DMA_TO_DEVICE; 474 ctxt->pages[0] = page; 475 ctxt->count = 1; 476 477 /* Prepare the SGE for the RPCRDMA Header */ 478 ctxt->sge[0].lkey = rdma->sc_pd->local_dma_lkey; 479 ctxt->sge[0].length = 480 svc_rdma_xdr_get_reply_hdr_len((__be32 *)rdma_resp); 481 ctxt->sge[0].addr = 482 ib_dma_map_page(rdma->sc_cm_id->device, page, 0, 483 ctxt->sge[0].length, DMA_TO_DEVICE); 484 if (ib_dma_mapping_error(rdma->sc_cm_id->device, ctxt->sge[0].addr)) 485 goto err; 486 svc_rdma_count_mappings(rdma, ctxt); 487 488 ctxt->direction = DMA_TO_DEVICE; 489 490 /* Map the payload indicated by 'byte_count' */ 491 xdr_off = 0; 492 for (sge_no = 1; byte_count && sge_no < vec->count; sge_no++) { 493 sge_bytes = min_t(size_t, vec->sge[sge_no].iov_len, byte_count); 494 byte_count -= sge_bytes; 495 ctxt->sge[sge_no].addr = 496 dma_map_xdr(rdma, &rqstp->rq_res, xdr_off, 497 sge_bytes, DMA_TO_DEVICE); 498 xdr_off += sge_bytes; 499 if (ib_dma_mapping_error(rdma->sc_cm_id->device, 500 ctxt->sge[sge_no].addr)) 501 goto err; 502 svc_rdma_count_mappings(rdma, ctxt); 503 ctxt->sge[sge_no].lkey = rdma->sc_pd->local_dma_lkey; 504 ctxt->sge[sge_no].length = sge_bytes; 505 } 506 if (byte_count != 0) { 507 pr_err("svcrdma: Could not map %d bytes\n", byte_count); 508 goto err; 509 } 510 511 /* Save all respages in the ctxt and remove them from the 512 * respages array. They are our pages until the I/O 513 * completes. 514 */ 515 pages = rqstp->rq_next_page - rqstp->rq_respages; 516 for (page_no = 0; page_no < pages; page_no++) { 517 ctxt->pages[page_no+1] = rqstp->rq_respages[page_no]; 518 ctxt->count++; 519 rqstp->rq_respages[page_no] = NULL; 520 } 521 rqstp->rq_next_page = rqstp->rq_respages + 1; 522 523 if (sge_no > rdma->sc_max_sge) { 524 pr_err("svcrdma: Too many sges (%d)\n", sge_no); 525 goto err; 526 } 527 memset(&send_wr, 0, sizeof send_wr); 528 ctxt->cqe.done = svc_rdma_wc_send; 529 send_wr.wr_cqe = &ctxt->cqe; 530 send_wr.sg_list = ctxt->sge; 531 send_wr.num_sge = sge_no; 532 if (inv_rkey) { 533 send_wr.opcode = IB_WR_SEND_WITH_INV; 534 send_wr.ex.invalidate_rkey = inv_rkey; 535 } else 536 send_wr.opcode = IB_WR_SEND; 537 send_wr.send_flags = IB_SEND_SIGNALED; 538 539 ret = svc_rdma_send(rdma, &send_wr); 540 if (ret) 541 goto err; 542 543 return 0; 544 545 err: 546 svc_rdma_unmap_dma(ctxt); 547 svc_rdma_put_context(ctxt, 1); 548 return ret; 549 } 550 551 void svc_rdma_prep_reply_hdr(struct svc_rqst *rqstp) 552 { 553 } 554 555 int svc_rdma_sendto(struct svc_rqst *rqstp) 556 { 557 struct svc_xprt *xprt = rqstp->rq_xprt; 558 struct svcxprt_rdma *rdma = 559 container_of(xprt, struct svcxprt_rdma, sc_xprt); 560 struct rpcrdma_msg *rdma_argp; 561 struct rpcrdma_msg *rdma_resp; 562 struct rpcrdma_write_array *wr_ary, *rp_ary; 563 int ret; 564 int inline_bytes; 565 struct page *res_page; 566 struct svc_rdma_req_map *vec; 567 u32 inv_rkey; 568 __be32 *p; 569 570 dprintk("svcrdma: sending response for rqstp=%p\n", rqstp); 571 572 /* Get the RDMA request header. The receive logic always 573 * places this at the start of page 0. 574 */ 575 rdma_argp = page_address(rqstp->rq_pages[0]); 576 svc_rdma_get_write_arrays(rdma_argp, &wr_ary, &rp_ary); 577 578 inv_rkey = 0; 579 if (rdma->sc_snd_w_inv) 580 inv_rkey = svc_rdma_get_inv_rkey(rdma_argp, wr_ary, rp_ary); 581 582 /* Build an req vec for the XDR */ 583 vec = svc_rdma_get_req_map(rdma); 584 ret = svc_rdma_map_xdr(rdma, &rqstp->rq_res, vec, wr_ary != NULL); 585 if (ret) 586 goto err0; 587 inline_bytes = rqstp->rq_res.len; 588 589 /* Create the RDMA response header. xprt->xpt_mutex, 590 * acquired in svc_send(), serializes RPC replies. The 591 * code path below that inserts the credit grant value 592 * into each transport header runs only inside this 593 * critical section. 594 */ 595 ret = -ENOMEM; 596 res_page = alloc_page(GFP_KERNEL); 597 if (!res_page) 598 goto err0; 599 rdma_resp = page_address(res_page); 600 601 p = &rdma_resp->rm_xid; 602 *p++ = rdma_argp->rm_xid; 603 *p++ = rdma_argp->rm_vers; 604 *p++ = rdma->sc_fc_credits; 605 *p++ = rp_ary ? rdma_nomsg : rdma_msg; 606 607 /* Start with empty chunks */ 608 *p++ = xdr_zero; 609 *p++ = xdr_zero; 610 *p = xdr_zero; 611 612 /* Send any write-chunk data and build resp write-list */ 613 if (wr_ary) { 614 ret = send_write_chunks(rdma, wr_ary, rdma_resp, rqstp, vec); 615 if (ret < 0) 616 goto err1; 617 inline_bytes -= ret + xdr_padsize(ret); 618 } 619 620 /* Send any reply-list data and update resp reply-list */ 621 if (rp_ary) { 622 ret = send_reply_chunks(rdma, rp_ary, rdma_resp, rqstp, vec); 623 if (ret < 0) 624 goto err1; 625 inline_bytes -= ret; 626 } 627 628 /* Post a fresh Receive buffer _before_ sending the reply */ 629 ret = svc_rdma_post_recv(rdma, GFP_KERNEL); 630 if (ret) 631 goto err1; 632 633 ret = send_reply(rdma, rqstp, res_page, rdma_resp, vec, 634 inline_bytes, inv_rkey); 635 if (ret < 0) 636 goto err0; 637 638 svc_rdma_put_req_map(rdma, vec); 639 dprintk("svcrdma: send_reply returns %d\n", ret); 640 return ret; 641 642 err1: 643 put_page(res_page); 644 err0: 645 svc_rdma_put_req_map(rdma, vec); 646 pr_err("svcrdma: Could not send reply, err=%d. Closing transport.\n", 647 ret); 648 set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags); 649 return -ENOTCONN; 650 } 651 652 void svc_rdma_send_error(struct svcxprt_rdma *xprt, struct rpcrdma_msg *rmsgp, 653 int status) 654 { 655 struct ib_send_wr err_wr; 656 struct page *p; 657 struct svc_rdma_op_ctxt *ctxt; 658 enum rpcrdma_errcode err; 659 __be32 *va; 660 int length; 661 int ret; 662 663 ret = svc_rdma_repost_recv(xprt, GFP_KERNEL); 664 if (ret) 665 return; 666 667 p = alloc_page(GFP_KERNEL); 668 if (!p) 669 return; 670 va = page_address(p); 671 672 /* XDR encode an error reply */ 673 err = ERR_CHUNK; 674 if (status == -EPROTONOSUPPORT) 675 err = ERR_VERS; 676 length = svc_rdma_xdr_encode_error(xprt, rmsgp, err, va); 677 678 ctxt = svc_rdma_get_context(xprt); 679 ctxt->direction = DMA_TO_DEVICE; 680 ctxt->count = 1; 681 ctxt->pages[0] = p; 682 683 /* Prepare SGE for local address */ 684 ctxt->sge[0].lkey = xprt->sc_pd->local_dma_lkey; 685 ctxt->sge[0].length = length; 686 ctxt->sge[0].addr = ib_dma_map_page(xprt->sc_cm_id->device, 687 p, 0, length, DMA_TO_DEVICE); 688 if (ib_dma_mapping_error(xprt->sc_cm_id->device, ctxt->sge[0].addr)) { 689 dprintk("svcrdma: Error mapping buffer for protocol error\n"); 690 svc_rdma_put_context(ctxt, 1); 691 return; 692 } 693 svc_rdma_count_mappings(xprt, ctxt); 694 695 /* Prepare SEND WR */ 696 memset(&err_wr, 0, sizeof(err_wr)); 697 ctxt->cqe.done = svc_rdma_wc_send; 698 err_wr.wr_cqe = &ctxt->cqe; 699 err_wr.sg_list = ctxt->sge; 700 err_wr.num_sge = 1; 701 err_wr.opcode = IB_WR_SEND; 702 err_wr.send_flags = IB_SEND_SIGNALED; 703 704 /* Post It */ 705 ret = svc_rdma_send(xprt, &err_wr); 706 if (ret) { 707 dprintk("svcrdma: Error %d posting send for protocol error\n", 708 ret); 709 svc_rdma_unmap_dma(ctxt); 710 svc_rdma_put_context(ctxt, 1); 711 } 712 } 713