1 /* 2 * CDDL HEADER START 3 * 4 * The contents of this file are subject to the terms of the 5 * Common Development and Distribution License (the "License"). 6 * You may not use this file except in compliance with the License. 7 * 8 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE 9 * or http://www.opensolaris.org/os/licensing. 10 * See the License for the specific language governing permissions 11 * and limitations under the License. 12 * 13 * When distributing Covered Code, include this CDDL HEADER in each 14 * file and include the License file at usr/src/OPENSOLARIS.LICENSE. 15 * If applicable, add the following below this CDDL HEADER, with the 16 * fields enclosed by brackets "[]" replaced with your own identifying 17 * information: Portions Copyright [yyyy] [name of copyright owner] 18 * 19 * CDDL HEADER END 20 */ 21 /* 22 * Copyright 2009 Sun Microsystems, Inc. All rights reserved. 23 * Use is subject to license terms. 24 */ 25 /* Copyright (c) 1983, 1984, 1985, 1986, 1987, 1988, 1989 AT&T */ 26 /* All Rights Reserved */ 27 /* 28 * Portions of this source code were derived from Berkeley 29 * 4.3 BSD under license from the Regents of the University of 30 * California. 31 */ 32 33 /* 34 * Server side of RPC over RDMA in the kernel. 35 */ 36 37 #include <sys/param.h> 38 #include <sys/types.h> 39 #include <sys/user.h> 40 #include <sys/sysmacros.h> 41 #include <sys/proc.h> 42 #include <sys/file.h> 43 #include <sys/errno.h> 44 #include <sys/kmem.h> 45 #include <sys/debug.h> 46 #include <sys/systm.h> 47 #include <sys/cmn_err.h> 48 #include <sys/kstat.h> 49 #include <sys/vtrace.h> 50 #include <sys/debug.h> 51 52 #include <rpc/types.h> 53 #include <rpc/xdr.h> 54 #include <rpc/auth.h> 55 #include <rpc/clnt.h> 56 #include <rpc/rpc_msg.h> 57 #include <rpc/svc.h> 58 #include <rpc/rpc_rdma.h> 59 #include <sys/ddi.h> 60 #include <sys/sunddi.h> 61 62 #include <inet/common.h> 63 #include <inet/ip.h> 64 #include <inet/ip6.h> 65 66 #include <nfs/nfs.h> 67 #include <sys/sdt.h> 68 69 #define SVC_RDMA_SUCCESS 0 70 #define SVC_RDMA_FAIL -1 71 72 #define SVC_CREDIT_FACTOR (0.5) 73 74 #define MSG_IS_RPCSEC_GSS(msg) \ 75 ((msg)->rm_reply.rp_acpt.ar_verf.oa_flavor == RPCSEC_GSS) 76 77 78 uint32_t rdma_bufs_granted = RDMA_BUFS_GRANT; 79 80 /* 81 * RDMA transport specific data associated with SVCMASTERXPRT 82 */ 83 struct rdma_data { 84 SVCMASTERXPRT *rd_xprt; /* back ptr to SVCMASTERXPRT */ 85 struct rdma_svc_data rd_data; /* rdma data */ 86 rdma_mod_t *r_mod; /* RDMA module containing ops ptr */ 87 }; 88 89 /* 90 * Plugin connection specific data stashed away in clone SVCXPRT 91 */ 92 struct clone_rdma_data { 93 CONN *conn; /* RDMA connection */ 94 rdma_buf_t rpcbuf; /* RPC req/resp buffer */ 95 struct clist *cl_reply; /* reply chunk buffer info */ 96 struct clist *cl_wlist; /* write list clist */ 97 }; 98 99 #define MAXADDRLEN 128 /* max length for address mask */ 100 101 /* 102 * Routines exported through ops vector. 103 */ 104 static bool_t svc_rdma_krecv(SVCXPRT *, mblk_t *, struct rpc_msg *); 105 static bool_t svc_rdma_ksend(SVCXPRT *, struct rpc_msg *); 106 static bool_t svc_rdma_kgetargs(SVCXPRT *, xdrproc_t, caddr_t); 107 static bool_t svc_rdma_kfreeargs(SVCXPRT *, xdrproc_t, caddr_t); 108 void svc_rdma_kdestroy(SVCMASTERXPRT *); 109 static int svc_rdma_kdup(struct svc_req *, caddr_t, int, 110 struct dupreq **, bool_t *); 111 static void svc_rdma_kdupdone(struct dupreq *, caddr_t, 112 void (*)(), int, int); 113 static int32_t *svc_rdma_kgetres(SVCXPRT *, int); 114 static void svc_rdma_kfreeres(SVCXPRT *); 115 static void svc_rdma_kclone_destroy(SVCXPRT *); 116 static void svc_rdma_kstart(SVCMASTERXPRT *); 117 void svc_rdma_kstop(SVCMASTERXPRT *); 118 119 static int svc_process_long_reply(SVCXPRT *, xdrproc_t, 120 caddr_t, struct rpc_msg *, bool_t, int *, 121 int *, int *, unsigned int *); 122 123 static int svc_compose_rpcmsg(SVCXPRT *, CONN *, xdrproc_t, 124 caddr_t, rdma_buf_t *, XDR **, struct rpc_msg *, 125 bool_t, uint_t *); 126 static bool_t rpcmsg_length(xdrproc_t, 127 caddr_t, 128 struct rpc_msg *, bool_t, int); 129 130 /* 131 * Server transport operations vector. 132 */ 133 struct svc_ops rdma_svc_ops = { 134 svc_rdma_krecv, /* Get requests */ 135 svc_rdma_kgetargs, /* Deserialize arguments */ 136 svc_rdma_ksend, /* Send reply */ 137 svc_rdma_kfreeargs, /* Free argument data space */ 138 svc_rdma_kdestroy, /* Destroy transport handle */ 139 svc_rdma_kdup, /* Check entry in dup req cache */ 140 svc_rdma_kdupdone, /* Mark entry in dup req cache as done */ 141 svc_rdma_kgetres, /* Get pointer to response buffer */ 142 svc_rdma_kfreeres, /* Destroy pre-serialized response header */ 143 svc_rdma_kclone_destroy, /* Destroy a clone xprt */ 144 svc_rdma_kstart /* Tell `ready-to-receive' to rpcmod */ 145 }; 146 147 /* 148 * Server statistics 149 * NOTE: This structure type is duplicated in the NFS fast path. 150 */ 151 struct { 152 kstat_named_t rscalls; 153 kstat_named_t rsbadcalls; 154 kstat_named_t rsnullrecv; 155 kstat_named_t rsbadlen; 156 kstat_named_t rsxdrcall; 157 kstat_named_t rsdupchecks; 158 kstat_named_t rsdupreqs; 159 kstat_named_t rslongrpcs; 160 kstat_named_t rstotalreplies; 161 kstat_named_t rstotallongreplies; 162 kstat_named_t rstotalinlinereplies; 163 } rdmarsstat = { 164 { "calls", KSTAT_DATA_UINT64 }, 165 { "badcalls", KSTAT_DATA_UINT64 }, 166 { "nullrecv", KSTAT_DATA_UINT64 }, 167 { "badlen", KSTAT_DATA_UINT64 }, 168 { "xdrcall", KSTAT_DATA_UINT64 }, 169 { "dupchecks", KSTAT_DATA_UINT64 }, 170 { "dupreqs", KSTAT_DATA_UINT64 }, 171 { "longrpcs", KSTAT_DATA_UINT64 }, 172 { "totalreplies", KSTAT_DATA_UINT64 }, 173 { "totallongreplies", KSTAT_DATA_UINT64 }, 174 { "totalinlinereplies", KSTAT_DATA_UINT64 }, 175 }; 176 177 kstat_named_t *rdmarsstat_ptr = (kstat_named_t *)&rdmarsstat; 178 uint_t rdmarsstat_ndata = sizeof (rdmarsstat) / sizeof (kstat_named_t); 179 180 #define RSSTAT_INCR(x) atomic_add_64(&rdmarsstat.x.value.ui64, 1) 181 /* 182 * Create a transport record. 183 * The transport record, output buffer, and private data structure 184 * are allocated. The output buffer is serialized into using xdrmem. 185 * There is one transport record per user process which implements a 186 * set of services. 187 */ 188 /* ARGSUSED */ 189 int 190 svc_rdma_kcreate(char *netid, SVC_CALLOUT_TABLE *sct, int id, 191 rdma_xprt_group_t *started_xprts) 192 { 193 int error; 194 SVCMASTERXPRT *xprt; 195 struct rdma_data *rd; 196 rdma_registry_t *rmod; 197 rdma_xprt_record_t *xprt_rec; 198 queue_t *q; 199 /* 200 * modload the RDMA plugins is not already done. 201 */ 202 if (!rdma_modloaded) { 203 /*CONSTANTCONDITION*/ 204 ASSERT(sizeof (struct clone_rdma_data) <= SVC_P2LEN); 205 206 mutex_enter(&rdma_modload_lock); 207 if (!rdma_modloaded) { 208 error = rdma_modload(); 209 } 210 mutex_exit(&rdma_modload_lock); 211 212 if (error) 213 return (error); 214 } 215 216 /* 217 * master_xprt_count is the count of master transport handles 218 * that were successfully created and are ready to recieve for 219 * RDMA based access. 220 */ 221 error = 0; 222 xprt_rec = NULL; 223 rw_enter(&rdma_lock, RW_READER); 224 if (rdma_mod_head == NULL) { 225 started_xprts->rtg_count = 0; 226 rw_exit(&rdma_lock); 227 if (rdma_dev_available) 228 return (EPROTONOSUPPORT); 229 else 230 return (ENODEV); 231 } 232 233 /* 234 * If we have reached here, then atleast one RDMA plugin has loaded. 235 * Create a master_xprt, make it start listenining on the device, 236 * if an error is generated, record it, we might need to shut 237 * the master_xprt. 238 * SVC_START() calls svc_rdma_kstart which calls plugin binding 239 * routines. 240 */ 241 for (rmod = rdma_mod_head; rmod != NULL; rmod = rmod->r_next) { 242 243 /* 244 * One SVCMASTERXPRT per RDMA plugin. 245 */ 246 xprt = kmem_zalloc(sizeof (*xprt), KM_SLEEP); 247 xprt->xp_ops = &rdma_svc_ops; 248 xprt->xp_sct = sct; 249 xprt->xp_type = T_RDMA; 250 mutex_init(&xprt->xp_req_lock, NULL, MUTEX_DEFAULT, NULL); 251 mutex_init(&xprt->xp_thread_lock, NULL, MUTEX_DEFAULT, NULL); 252 xprt->xp_req_head = (mblk_t *)0; 253 xprt->xp_req_tail = (mblk_t *)0; 254 xprt->xp_threads = 0; 255 xprt->xp_detached_threads = 0; 256 257 rd = kmem_zalloc(sizeof (*rd), KM_SLEEP); 258 xprt->xp_p2 = (caddr_t)rd; 259 rd->rd_xprt = xprt; 260 rd->r_mod = rmod->r_mod; 261 262 q = &rd->rd_data.q; 263 xprt->xp_wq = q; 264 q->q_ptr = &rd->rd_xprt; 265 xprt->xp_netid = NULL; 266 267 if (netid != NULL) { 268 xprt->xp_netid = kmem_alloc(strlen(netid) + 1, 269 KM_SLEEP); 270 (void) strcpy(xprt->xp_netid, netid); 271 } 272 273 xprt->xp_addrmask.maxlen = 274 xprt->xp_addrmask.len = sizeof (struct sockaddr_in); 275 xprt->xp_addrmask.buf = 276 kmem_zalloc(xprt->xp_addrmask.len, KM_SLEEP); 277 ((struct sockaddr_in *)xprt->xp_addrmask.buf)->sin_addr.s_addr = 278 (uint32_t)~0; 279 ((struct sockaddr_in *)xprt->xp_addrmask.buf)->sin_family = 280 (ushort_t)~0; 281 282 /* 283 * Each of the plugins will have their own Service ID 284 * to listener specific mapping, like port number for VI 285 * and service name for IB. 286 */ 287 rd->rd_data.svcid = id; 288 error = svc_xprt_register(xprt, id); 289 if (error) { 290 DTRACE_PROBE(krpc__e__svcrdma__xprt__reg); 291 goto cleanup; 292 } 293 294 SVC_START(xprt); 295 if (!rd->rd_data.active) { 296 svc_xprt_unregister(xprt); 297 error = rd->rd_data.err_code; 298 goto cleanup; 299 } 300 301 /* 302 * This is set only when there is atleast one or more 303 * transports successfully created. We insert the pointer 304 * to the created RDMA master xprt into a separately maintained 305 * list. This way we can easily reference it later to cleanup, 306 * when NFS kRPC service pool is going away/unregistered. 307 */ 308 started_xprts->rtg_count ++; 309 xprt_rec = kmem_alloc(sizeof (*xprt_rec), KM_SLEEP); 310 xprt_rec->rtr_xprt_ptr = xprt; 311 xprt_rec->rtr_next = started_xprts->rtg_listhead; 312 started_xprts->rtg_listhead = xprt_rec; 313 continue; 314 cleanup: 315 SVC_DESTROY(xprt); 316 if (error == RDMA_FAILED) 317 error = EPROTONOSUPPORT; 318 } 319 320 rw_exit(&rdma_lock); 321 322 /* 323 * Don't return any error even if a single plugin was started 324 * successfully. 325 */ 326 if (started_xprts->rtg_count == 0) 327 return (error); 328 return (0); 329 } 330 331 /* 332 * Cleanup routine for freeing up memory allocated by 333 * svc_rdma_kcreate() 334 */ 335 void 336 svc_rdma_kdestroy(SVCMASTERXPRT *xprt) 337 { 338 struct rdma_data *rd = (struct rdma_data *)xprt->xp_p2; 339 340 341 mutex_destroy(&xprt->xp_req_lock); 342 mutex_destroy(&xprt->xp_thread_lock); 343 kmem_free(xprt->xp_netid, strlen(xprt->xp_netid) + 1); 344 kmem_free(rd, sizeof (*rd)); 345 kmem_free(xprt->xp_addrmask.buf, xprt->xp_addrmask.maxlen); 346 kmem_free(xprt, sizeof (*xprt)); 347 } 348 349 350 static void 351 svc_rdma_kstart(SVCMASTERXPRT *xprt) 352 { 353 struct rdma_svc_data *svcdata; 354 rdma_mod_t *rmod; 355 356 svcdata = &((struct rdma_data *)xprt->xp_p2)->rd_data; 357 rmod = ((struct rdma_data *)xprt->xp_p2)->r_mod; 358 359 /* 360 * Create a listener for module at this port 361 */ 362 363 if (rmod->rdma_count != 0) 364 (*rmod->rdma_ops->rdma_svc_listen)(svcdata); 365 else 366 svcdata->err_code = RDMA_FAILED; 367 } 368 369 void 370 svc_rdma_kstop(SVCMASTERXPRT *xprt) 371 { 372 struct rdma_svc_data *svcdata; 373 rdma_mod_t *rmod; 374 375 svcdata = &((struct rdma_data *)xprt->xp_p2)->rd_data; 376 rmod = ((struct rdma_data *)xprt->xp_p2)->r_mod; 377 378 /* 379 * Call the stop listener routine for each plugin. If rdma_count is 380 * already zero set active to zero. 381 */ 382 if (rmod->rdma_count != 0) 383 (*rmod->rdma_ops->rdma_svc_stop)(svcdata); 384 else 385 svcdata->active = 0; 386 if (svcdata->active) 387 DTRACE_PROBE(krpc__e__svcrdma__kstop); 388 } 389 390 /* ARGSUSED */ 391 static void 392 svc_rdma_kclone_destroy(SVCXPRT *clone_xprt) 393 { 394 } 395 396 static bool_t 397 svc_rdma_krecv(SVCXPRT *clone_xprt, mblk_t *mp, struct rpc_msg *msg) 398 { 399 XDR *xdrs; 400 CONN *conn; 401 402 rdma_recv_data_t *rdp = (rdma_recv_data_t *)mp->b_rptr; 403 struct clone_rdma_data *crdp; 404 struct clist *cl = NULL; 405 struct clist *wcl = NULL; 406 struct clist *cllong = NULL; 407 408 rdma_stat status; 409 uint32_t vers, op, pos, xid; 410 uint32_t rdma_credit; 411 uint32_t wcl_total_length = 0; 412 bool_t wwl = FALSE; 413 414 crdp = (struct clone_rdma_data *)clone_xprt->xp_p2buf; 415 RSSTAT_INCR(rscalls); 416 conn = rdp->conn; 417 418 status = rdma_svc_postrecv(conn); 419 if (status != RDMA_SUCCESS) { 420 DTRACE_PROBE(krpc__e__svcrdma__krecv__postrecv); 421 goto badrpc_call; 422 } 423 424 xdrs = &clone_xprt->xp_xdrin; 425 xdrmem_create(xdrs, rdp->rpcmsg.addr, rdp->rpcmsg.len, XDR_DECODE); 426 xid = *(uint32_t *)rdp->rpcmsg.addr; 427 XDR_SETPOS(xdrs, sizeof (uint32_t)); 428 429 if (! xdr_u_int(xdrs, &vers) || 430 ! xdr_u_int(xdrs, &rdma_credit) || 431 ! xdr_u_int(xdrs, &op)) { 432 DTRACE_PROBE(krpc__e__svcrdma__krecv__uint); 433 goto xdr_err; 434 } 435 436 /* Checking if the status of the recv operation was normal */ 437 if (rdp->status != 0) { 438 DTRACE_PROBE1(krpc__e__svcrdma__krecv__invalid__status, 439 int, rdp->status); 440 goto badrpc_call; 441 } 442 443 if (! xdr_do_clist(xdrs, &cl)) { 444 DTRACE_PROBE(krpc__e__svcrdma__krecv__do__clist); 445 goto xdr_err; 446 } 447 448 if (!xdr_decode_wlist_svc(xdrs, &wcl, &wwl, &wcl_total_length, conn)) { 449 DTRACE_PROBE(krpc__e__svcrdma__krecv__decode__wlist); 450 if (cl) 451 clist_free(cl); 452 goto xdr_err; 453 } 454 crdp->cl_wlist = wcl; 455 456 crdp->cl_reply = NULL; 457 (void) xdr_decode_reply_wchunk(xdrs, &crdp->cl_reply); 458 459 /* 460 * A chunk at 0 offset indicates that the RPC call message 461 * is in a chunk. Get the RPC call message chunk. 462 */ 463 if (cl != NULL && op == RDMA_NOMSG) { 464 465 /* Remove RPC call message chunk from chunklist */ 466 cllong = cl; 467 cl = cl->c_next; 468 cllong->c_next = NULL; 469 470 471 /* Allocate and register memory for the RPC call msg chunk */ 472 cllong->rb_longbuf.type = RDMA_LONG_BUFFER; 473 cllong->rb_longbuf.len = cllong->c_len > LONG_REPLY_LEN ? 474 cllong->c_len : LONG_REPLY_LEN; 475 476 if (rdma_buf_alloc(conn, &cllong->rb_longbuf)) { 477 clist_free(cllong); 478 goto cll_malloc_err; 479 } 480 481 cllong->u.c_daddr3 = cllong->rb_longbuf.addr; 482 483 if (cllong->u.c_daddr == NULL) { 484 DTRACE_PROBE(krpc__e__svcrdma__krecv__nomem); 485 rdma_buf_free(conn, &cllong->rb_longbuf); 486 clist_free(cllong); 487 goto cll_malloc_err; 488 } 489 490 status = clist_register(conn, cllong, CLIST_REG_DST); 491 if (status) { 492 DTRACE_PROBE(krpc__e__svcrdma__krecv__clist__reg); 493 rdma_buf_free(conn, &cllong->rb_longbuf); 494 clist_free(cllong); 495 goto cll_malloc_err; 496 } 497 498 /* 499 * Now read the RPC call message in 500 */ 501 status = RDMA_READ(conn, cllong, WAIT); 502 if (status) { 503 DTRACE_PROBE(krpc__e__svcrdma__krecv__read); 504 (void) clist_deregister(conn, cllong, CLIST_REG_DST); 505 rdma_buf_free(conn, &cllong->rb_longbuf); 506 clist_free(cllong); 507 goto cll_malloc_err; 508 } 509 510 status = clist_syncmem(conn, cllong, CLIST_REG_DST); 511 (void) clist_deregister(conn, cllong, CLIST_REG_DST); 512 513 xdrrdma_create(xdrs, (caddr_t)(uintptr_t)cllong->u.c_daddr3, 514 cllong->c_len, 0, cl, XDR_DECODE, conn); 515 516 crdp->rpcbuf = cllong->rb_longbuf; 517 crdp->rpcbuf.len = cllong->c_len; 518 clist_free(cllong); 519 RDMA_BUF_FREE(conn, &rdp->rpcmsg); 520 } else { 521 pos = XDR_GETPOS(xdrs); 522 xdrrdma_create(xdrs, rdp->rpcmsg.addr + pos, 523 rdp->rpcmsg.len - pos, 0, cl, XDR_DECODE, conn); 524 crdp->rpcbuf = rdp->rpcmsg; 525 526 /* Use xdrrdmablk_ops to indicate there is a read chunk list */ 527 if (cl != NULL) { 528 int32_t flg = XDR_RDMA_RLIST_REG; 529 530 XDR_CONTROL(xdrs, XDR_RDMA_SET_FLAGS, &flg); 531 xdrs->x_ops = &xdrrdmablk_ops; 532 } 533 } 534 535 if (crdp->cl_wlist) { 536 int32_t flg = XDR_RDMA_WLIST_REG; 537 538 XDR_CONTROL(xdrs, XDR_RDMA_SET_WLIST, crdp->cl_wlist); 539 XDR_CONTROL(xdrs, XDR_RDMA_SET_FLAGS, &flg); 540 } 541 542 if (! xdr_callmsg(xdrs, msg)) { 543 DTRACE_PROBE(krpc__e__svcrdma__krecv__callmsg); 544 RSSTAT_INCR(rsxdrcall); 545 goto callmsg_err; 546 } 547 548 /* 549 * Point the remote transport address in the service_transport 550 * handle at the address in the request. 551 */ 552 clone_xprt->xp_rtaddr.buf = conn->c_raddr.buf; 553 clone_xprt->xp_rtaddr.len = conn->c_raddr.len; 554 clone_xprt->xp_rtaddr.maxlen = conn->c_raddr.len; 555 clone_xprt->xp_xid = xid; 556 crdp->conn = conn; 557 558 freeb(mp); 559 560 return (TRUE); 561 562 callmsg_err: 563 rdma_buf_free(conn, &crdp->rpcbuf); 564 565 cll_malloc_err: 566 if (cl) 567 clist_free(cl); 568 xdr_err: 569 XDR_DESTROY(xdrs); 570 571 badrpc_call: 572 RDMA_BUF_FREE(conn, &rdp->rpcmsg); 573 RDMA_REL_CONN(conn); 574 freeb(mp); 575 RSSTAT_INCR(rsbadcalls); 576 return (FALSE); 577 } 578 579 static int 580 svc_process_long_reply(SVCXPRT * clone_xprt, 581 xdrproc_t xdr_results, caddr_t xdr_location, 582 struct rpc_msg *msg, bool_t has_args, int *msglen, 583 int *freelen, int *numchunks, unsigned int *final_len) 584 { 585 int status; 586 XDR xdrslong; 587 struct clist *wcl = NULL; 588 int count = 0; 589 int alloc_len; 590 char *memp; 591 rdma_buf_t long_rpc = {0}; 592 struct clone_rdma_data *crdp; 593 594 crdp = (struct clone_rdma_data *)clone_xprt->xp_p2buf; 595 596 bzero(&xdrslong, sizeof (xdrslong)); 597 598 /* Choose a size for the long rpc response */ 599 if (MSG_IS_RPCSEC_GSS(msg)) { 600 alloc_len = RNDUP(MAX_AUTH_BYTES + *msglen); 601 } else { 602 alloc_len = RNDUP(*msglen); 603 } 604 605 if (alloc_len <= 64 * 1024) { 606 if (alloc_len > 32 * 1024) { 607 alloc_len = 64 * 1024; 608 } else { 609 if (alloc_len > 16 * 1024) { 610 alloc_len = 32 * 1024; 611 } else { 612 alloc_len = 16 * 1024; 613 } 614 } 615 } 616 617 long_rpc.type = RDMA_LONG_BUFFER; 618 long_rpc.len = alloc_len; 619 if (rdma_buf_alloc(crdp->conn, &long_rpc)) { 620 return (SVC_RDMA_FAIL); 621 } 622 623 memp = long_rpc.addr; 624 xdrmem_create(&xdrslong, memp, alloc_len, XDR_ENCODE); 625 626 msg->rm_xid = clone_xprt->xp_xid; 627 628 if (!(xdr_replymsg(&xdrslong, msg) && 629 (!has_args || SVCAUTH_WRAP(&clone_xprt->xp_auth, &xdrslong, 630 xdr_results, xdr_location)))) { 631 rdma_buf_free(crdp->conn, &long_rpc); 632 DTRACE_PROBE(krpc__e__svcrdma__longrep__authwrap); 633 return (SVC_RDMA_FAIL); 634 } 635 636 *final_len = XDR_GETPOS(&xdrslong); 637 638 *numchunks = 0; 639 *freelen = 0; 640 641 wcl = crdp->cl_reply; 642 wcl->rb_longbuf = long_rpc; 643 644 count = *final_len; 645 while (wcl != NULL) { 646 if (wcl->c_dmemhandle.mrc_rmr == 0) 647 break; 648 649 if (wcl->c_len > count) { 650 wcl->c_len = count; 651 } 652 wcl->w.c_saddr3 = (caddr_t)memp; 653 654 count -= wcl->c_len; 655 *numchunks += 1; 656 if (count == 0) 657 break; 658 memp += wcl->c_len; 659 wcl = wcl->c_next; 660 } 661 662 wcl = crdp->cl_reply; 663 664 /* 665 * MUST fail if there are still more data 666 */ 667 if (count > 0) { 668 rdma_buf_free(crdp->conn, &long_rpc); 669 DTRACE_PROBE(krpc__e__svcrdma__longrep__dlen__clist); 670 return (SVC_RDMA_FAIL); 671 } 672 673 if (clist_register(crdp->conn, wcl, CLIST_REG_SOURCE) != RDMA_SUCCESS) { 674 rdma_buf_free(crdp->conn, &long_rpc); 675 DTRACE_PROBE(krpc__e__svcrdma__longrep__clistreg); 676 return (SVC_RDMA_FAIL); 677 } 678 679 status = clist_syncmem(crdp->conn, wcl, CLIST_REG_SOURCE); 680 681 if (status) { 682 (void) clist_deregister(crdp->conn, wcl, CLIST_REG_SOURCE); 683 rdma_buf_free(crdp->conn, &long_rpc); 684 DTRACE_PROBE(krpc__e__svcrdma__longrep__syncmem); 685 return (SVC_RDMA_FAIL); 686 } 687 688 status = RDMA_WRITE(crdp->conn, wcl, WAIT); 689 690 (void) clist_deregister(crdp->conn, wcl, CLIST_REG_SOURCE); 691 rdma_buf_free(crdp->conn, &wcl->rb_longbuf); 692 693 if (status != RDMA_SUCCESS) { 694 DTRACE_PROBE(krpc__e__svcrdma__longrep__write); 695 return (SVC_RDMA_FAIL); 696 } 697 698 return (SVC_RDMA_SUCCESS); 699 } 700 701 702 static int 703 svc_compose_rpcmsg(SVCXPRT * clone_xprt, CONN * conn, xdrproc_t xdr_results, 704 caddr_t xdr_location, rdma_buf_t *rpcreply, XDR ** xdrs, 705 struct rpc_msg *msg, bool_t has_args, uint_t *len) 706 { 707 /* 708 * Get a pre-allocated buffer for rpc reply 709 */ 710 rpcreply->type = SEND_BUFFER; 711 if (rdma_buf_alloc(conn, rpcreply)) { 712 DTRACE_PROBE(krpc__e__svcrdma__rpcmsg__reply__nofreebufs); 713 return (SVC_RDMA_FAIL); 714 } 715 716 xdrrdma_create(*xdrs, rpcreply->addr, rpcreply->len, 717 0, NULL, XDR_ENCODE, conn); 718 719 msg->rm_xid = clone_xprt->xp_xid; 720 721 if (has_args) { 722 if (!(xdr_replymsg(*xdrs, msg) && 723 (!has_args || 724 SVCAUTH_WRAP(&clone_xprt->xp_auth, *xdrs, 725 xdr_results, xdr_location)))) { 726 rdma_buf_free(conn, rpcreply); 727 DTRACE_PROBE( 728 krpc__e__svcrdma__rpcmsg__reply__authwrap1); 729 return (SVC_RDMA_FAIL); 730 } 731 } else { 732 if (!xdr_replymsg(*xdrs, msg)) { 733 rdma_buf_free(conn, rpcreply); 734 DTRACE_PROBE( 735 krpc__e__svcrdma__rpcmsg__reply__authwrap2); 736 return (SVC_RDMA_FAIL); 737 } 738 } 739 740 *len = XDR_GETPOS(*xdrs); 741 742 return (SVC_RDMA_SUCCESS); 743 } 744 745 /* 746 * Send rpc reply. 747 */ 748 static bool_t 749 svc_rdma_ksend(SVCXPRT * clone_xprt, struct rpc_msg *msg) 750 { 751 XDR *xdrs_rpc = &(clone_xprt->xp_xdrout); 752 XDR xdrs_rhdr; 753 CONN *conn = NULL; 754 rdma_buf_t rbuf_resp = {0}, rbuf_rpc_resp = {0}; 755 756 struct clone_rdma_data *crdp; 757 struct clist *cl_read = NULL; 758 struct clist *cl_send = NULL; 759 struct clist *cl_write = NULL; 760 xdrproc_t xdr_results; /* results XDR encoding function */ 761 caddr_t xdr_location; /* response results pointer */ 762 763 int retval = FALSE; 764 int status, msglen, num_wreply_segments = 0; 765 uint32_t rdma_credit = 0; 766 int freelen = 0; 767 bool_t has_args; 768 uint_t final_resp_len, rdma_response_op, vers; 769 770 bzero(&xdrs_rhdr, sizeof (XDR)); 771 crdp = (struct clone_rdma_data *)clone_xprt->xp_p2buf; 772 conn = crdp->conn; 773 774 /* 775 * If there is a result procedure specified in the reply message, 776 * it will be processed in the xdr_replymsg and SVCAUTH_WRAP. 777 * We need to make sure it won't be processed twice, so we null 778 * it for xdr_replymsg here. 779 */ 780 has_args = FALSE; 781 if (msg->rm_reply.rp_stat == MSG_ACCEPTED && 782 msg->rm_reply.rp_acpt.ar_stat == SUCCESS) { 783 if ((xdr_results = msg->acpted_rply.ar_results.proc) != NULL) { 784 has_args = TRUE; 785 xdr_location = msg->acpted_rply.ar_results.where; 786 msg->acpted_rply.ar_results.proc = xdr_void; 787 msg->acpted_rply.ar_results.where = NULL; 788 } 789 } 790 791 /* 792 * Given the limit on the inline response size (RPC_MSG_SZ), 793 * there is a need to make a guess as to the overall size of 794 * the response. If the resultant size is beyond the inline 795 * size, then the server needs to use the "reply chunk list" 796 * provided by the client (if the client provided one). An 797 * example of this type of response would be a READDIR 798 * response (e.g. a small directory read would fit in RPC_MSG_SZ 799 * and that is the preference but it may not fit) 800 * 801 * Combine the encoded size and the size of the true results 802 * and then make the decision about where to encode and send results. 803 * 804 * One important note, this calculation is ignoring the size 805 * of the encoding of the authentication overhead. The reason 806 * for this is rooted in the complexities of access to the 807 * encoded size of RPCSEC_GSS related authentiation, 808 * integrity, and privacy. 809 * 810 * If it turns out that the encoded authentication bumps the 811 * response over the RPC_MSG_SZ limit, then it may need to 812 * attempt to encode for the reply chunk list. 813 */ 814 815 /* 816 * Calculating the "sizeof" the RPC response header and the 817 * encoded results. 818 */ 819 msglen = xdr_sizeof(xdr_replymsg, msg); 820 821 if (msglen > 0) { 822 RSSTAT_INCR(rstotalreplies); 823 } 824 if (has_args) 825 msglen += xdrrdma_sizeof(xdr_results, xdr_location, 826 rdma_minchunk, NULL, NULL); 827 828 DTRACE_PROBE1(krpc__i__svcrdma__ksend__msglen, int, msglen); 829 830 status = SVC_RDMA_SUCCESS; 831 832 if (msglen < RPC_MSG_SZ) { 833 /* 834 * Looks like the response will fit in the inline 835 * response; let's try 836 */ 837 RSSTAT_INCR(rstotalinlinereplies); 838 839 rdma_response_op = RDMA_MSG; 840 841 status = svc_compose_rpcmsg(clone_xprt, conn, xdr_results, 842 xdr_location, &rbuf_rpc_resp, &xdrs_rpc, msg, 843 has_args, &final_resp_len); 844 845 DTRACE_PROBE1(krpc__i__srdma__ksend__compose_status, 846 int, status); 847 DTRACE_PROBE1(krpc__i__srdma__ksend__compose_len, 848 int, final_resp_len); 849 850 if (status == SVC_RDMA_SUCCESS && crdp->cl_reply) { 851 clist_free(crdp->cl_reply); 852 crdp->cl_reply = NULL; 853 } 854 } 855 856 /* 857 * If the encode failed (size?) or the message really is 858 * larger than what is allowed, try the response chunk list. 859 */ 860 if (status != SVC_RDMA_SUCCESS || msglen >= RPC_MSG_SZ) { 861 /* 862 * attempting to use a reply chunk list when there 863 * isn't one won't get very far... 864 */ 865 if (crdp->cl_reply == NULL) { 866 DTRACE_PROBE(krpc__e__svcrdma__ksend__noreplycl); 867 goto out; 868 } 869 870 RSSTAT_INCR(rstotallongreplies); 871 872 msglen = xdr_sizeof(xdr_replymsg, msg); 873 msglen += xdrrdma_sizeof(xdr_results, xdr_location, 0, 874 NULL, NULL); 875 876 status = svc_process_long_reply(clone_xprt, xdr_results, 877 xdr_location, msg, has_args, &msglen, &freelen, 878 &num_wreply_segments, &final_resp_len); 879 880 DTRACE_PROBE1(krpc__i__svcrdma__ksend__longreplen, 881 int, final_resp_len); 882 883 if (status != SVC_RDMA_SUCCESS) { 884 DTRACE_PROBE(krpc__e__svcrdma__ksend__compose__failed); 885 goto out; 886 } 887 888 rdma_response_op = RDMA_NOMSG; 889 } 890 891 DTRACE_PROBE1(krpc__i__svcrdma__ksend__rdmamsg__len, 892 int, final_resp_len); 893 894 rbuf_resp.type = SEND_BUFFER; 895 if (rdma_buf_alloc(conn, &rbuf_resp)) { 896 rdma_buf_free(conn, &rbuf_rpc_resp); 897 DTRACE_PROBE(krpc__e__svcrdma__ksend__nofreebufs); 898 goto out; 899 } 900 901 rdma_credit = rdma_bufs_granted; 902 903 vers = RPCRDMA_VERS; 904 xdrmem_create(&xdrs_rhdr, rbuf_resp.addr, rbuf_resp.len, XDR_ENCODE); 905 (*(uint32_t *)rbuf_resp.addr) = msg->rm_xid; 906 /* Skip xid and set the xdr position accordingly. */ 907 XDR_SETPOS(&xdrs_rhdr, sizeof (uint32_t)); 908 if (!xdr_u_int(&xdrs_rhdr, &vers) || 909 !xdr_u_int(&xdrs_rhdr, &rdma_credit) || 910 !xdr_u_int(&xdrs_rhdr, &rdma_response_op)) { 911 rdma_buf_free(conn, &rbuf_rpc_resp); 912 rdma_buf_free(conn, &rbuf_resp); 913 DTRACE_PROBE(krpc__e__svcrdma__ksend__uint); 914 goto out; 915 } 916 917 /* 918 * Now XDR the read chunk list, actually always NULL 919 */ 920 (void) xdr_encode_rlist_svc(&xdrs_rhdr, cl_read); 921 922 /* 923 * encode write list -- we already drove RDMA_WRITEs 924 */ 925 cl_write = crdp->cl_wlist; 926 if (!xdr_encode_wlist(&xdrs_rhdr, cl_write)) { 927 DTRACE_PROBE(krpc__e__svcrdma__ksend__enc__wlist); 928 rdma_buf_free(conn, &rbuf_rpc_resp); 929 rdma_buf_free(conn, &rbuf_resp); 930 goto out; 931 } 932 933 /* 934 * XDR encode the RDMA_REPLY write chunk 935 */ 936 if (!xdr_encode_reply_wchunk(&xdrs_rhdr, crdp->cl_reply, 937 num_wreply_segments)) { 938 rdma_buf_free(conn, &rbuf_rpc_resp); 939 rdma_buf_free(conn, &rbuf_resp); 940 goto out; 941 } 942 943 clist_add(&cl_send, 0, XDR_GETPOS(&xdrs_rhdr), &rbuf_resp.handle, 944 rbuf_resp.addr, NULL, NULL); 945 946 if (rdma_response_op == RDMA_MSG) { 947 clist_add(&cl_send, 0, final_resp_len, &rbuf_rpc_resp.handle, 948 rbuf_rpc_resp.addr, NULL, NULL); 949 } 950 951 status = RDMA_SEND(conn, cl_send, msg->rm_xid); 952 953 if (status == RDMA_SUCCESS) { 954 retval = TRUE; 955 } 956 957 out: 958 /* 959 * Free up sendlist chunks 960 */ 961 if (cl_send != NULL) 962 clist_free(cl_send); 963 964 /* 965 * Destroy private data for xdr rdma 966 */ 967 if (clone_xprt->xp_xdrout.x_ops != NULL) { 968 XDR_DESTROY(&(clone_xprt->xp_xdrout)); 969 } 970 971 if (crdp->cl_reply) { 972 clist_free(crdp->cl_reply); 973 crdp->cl_reply = NULL; 974 } 975 976 /* 977 * This is completely disgusting. If public is set it is 978 * a pointer to a structure whose first field is the address 979 * of the function to free that structure and any related 980 * stuff. (see rrokfree in nfs_xdr.c). 981 */ 982 if (xdrs_rpc->x_public) { 983 /* LINTED pointer alignment */ 984 (**((int (**)()) xdrs_rpc->x_public)) (xdrs_rpc->x_public); 985 } 986 987 if (xdrs_rhdr.x_ops != NULL) { 988 XDR_DESTROY(&xdrs_rhdr); 989 } 990 991 return (retval); 992 } 993 994 /* 995 * Deserialize arguments. 996 */ 997 static bool_t 998 svc_rdma_kgetargs(SVCXPRT *clone_xprt, xdrproc_t xdr_args, caddr_t args_ptr) 999 { 1000 if ((SVCAUTH_UNWRAP(&clone_xprt->xp_auth, &clone_xprt->xp_xdrin, 1001 xdr_args, args_ptr)) != TRUE) 1002 return (FALSE); 1003 return (TRUE); 1004 } 1005 1006 static bool_t 1007 svc_rdma_kfreeargs(SVCXPRT *clone_xprt, xdrproc_t xdr_args, 1008 caddr_t args_ptr) 1009 { 1010 struct clone_rdma_data *crdp; 1011 bool_t retval; 1012 1013 crdp = (struct clone_rdma_data *)clone_xprt->xp_p2buf; 1014 1015 /* 1016 * Free the args if needed then XDR_DESTROY 1017 */ 1018 if (args_ptr) { 1019 XDR *xdrs = &clone_xprt->xp_xdrin; 1020 1021 xdrs->x_op = XDR_FREE; 1022 retval = (*xdr_args)(xdrs, args_ptr); 1023 } 1024 1025 XDR_DESTROY(&(clone_xprt->xp_xdrin)); 1026 rdma_buf_free(crdp->conn, &crdp->rpcbuf); 1027 if (crdp->cl_reply) { 1028 clist_free(crdp->cl_reply); 1029 crdp->cl_reply = NULL; 1030 } 1031 RDMA_REL_CONN(crdp->conn); 1032 1033 return (retval); 1034 } 1035 1036 /* ARGSUSED */ 1037 static int32_t * 1038 svc_rdma_kgetres(SVCXPRT *clone_xprt, int size) 1039 { 1040 return (NULL); 1041 } 1042 1043 /* ARGSUSED */ 1044 static void 1045 svc_rdma_kfreeres(SVCXPRT *clone_xprt) 1046 { 1047 } 1048 1049 /* 1050 * the dup cacheing routines below provide a cache of non-failure 1051 * transaction id's. rpc service routines can use this to detect 1052 * retransmissions and re-send a non-failure response. 1053 */ 1054 1055 /* 1056 * MAXDUPREQS is the number of cached items. It should be adjusted 1057 * to the service load so that there is likely to be a response entry 1058 * when the first retransmission comes in. 1059 */ 1060 #define MAXDUPREQS 1024 1061 1062 /* 1063 * This should be appropriately scaled to MAXDUPREQS. 1064 */ 1065 #define DRHASHSZ 257 1066 1067 #if ((DRHASHSZ & (DRHASHSZ - 1)) == 0) 1068 #define XIDHASH(xid) ((xid) & (DRHASHSZ - 1)) 1069 #else 1070 #define XIDHASH(xid) ((xid) % DRHASHSZ) 1071 #endif 1072 #define DRHASH(dr) XIDHASH((dr)->dr_xid) 1073 #define REQTOXID(req) ((req)->rq_xprt->xp_xid) 1074 1075 static int rdmandupreqs = 0; 1076 int rdmamaxdupreqs = MAXDUPREQS; 1077 static kmutex_t rdmadupreq_lock; 1078 static struct dupreq *rdmadrhashtbl[DRHASHSZ]; 1079 static int rdmadrhashstat[DRHASHSZ]; 1080 1081 static void unhash(struct dupreq *); 1082 1083 /* 1084 * rdmadrmru points to the head of a circular linked list in lru order. 1085 * rdmadrmru->dr_next == drlru 1086 */ 1087 struct dupreq *rdmadrmru; 1088 1089 /* 1090 * svc_rdma_kdup searches the request cache and returns 0 if the 1091 * request is not found in the cache. If it is found, then it 1092 * returns the state of the request (in progress or done) and 1093 * the status or attributes that were part of the original reply. 1094 */ 1095 static int 1096 svc_rdma_kdup(struct svc_req *req, caddr_t res, int size, struct dupreq **drpp, 1097 bool_t *dupcachedp) 1098 { 1099 struct dupreq *dr; 1100 uint32_t xid; 1101 uint32_t drhash; 1102 int status; 1103 1104 xid = REQTOXID(req); 1105 mutex_enter(&rdmadupreq_lock); 1106 RSSTAT_INCR(rsdupchecks); 1107 /* 1108 * Check to see whether an entry already exists in the cache. 1109 */ 1110 dr = rdmadrhashtbl[XIDHASH(xid)]; 1111 while (dr != NULL) { 1112 if (dr->dr_xid == xid && 1113 dr->dr_proc == req->rq_proc && 1114 dr->dr_prog == req->rq_prog && 1115 dr->dr_vers == req->rq_vers && 1116 dr->dr_addr.len == req->rq_xprt->xp_rtaddr.len && 1117 bcmp((caddr_t)dr->dr_addr.buf, 1118 (caddr_t)req->rq_xprt->xp_rtaddr.buf, 1119 dr->dr_addr.len) == 0) { 1120 status = dr->dr_status; 1121 if (status == DUP_DONE) { 1122 bcopy(dr->dr_resp.buf, res, size); 1123 if (dupcachedp != NULL) 1124 *dupcachedp = (dr->dr_resfree != NULL); 1125 } else { 1126 dr->dr_status = DUP_INPROGRESS; 1127 *drpp = dr; 1128 } 1129 RSSTAT_INCR(rsdupreqs); 1130 mutex_exit(&rdmadupreq_lock); 1131 return (status); 1132 } 1133 dr = dr->dr_chain; 1134 } 1135 1136 /* 1137 * There wasn't an entry, either allocate a new one or recycle 1138 * an old one. 1139 */ 1140 if (rdmandupreqs < rdmamaxdupreqs) { 1141 dr = kmem_alloc(sizeof (*dr), KM_NOSLEEP); 1142 if (dr == NULL) { 1143 mutex_exit(&rdmadupreq_lock); 1144 return (DUP_ERROR); 1145 } 1146 dr->dr_resp.buf = NULL; 1147 dr->dr_resp.maxlen = 0; 1148 dr->dr_addr.buf = NULL; 1149 dr->dr_addr.maxlen = 0; 1150 if (rdmadrmru) { 1151 dr->dr_next = rdmadrmru->dr_next; 1152 rdmadrmru->dr_next = dr; 1153 } else { 1154 dr->dr_next = dr; 1155 } 1156 rdmandupreqs++; 1157 } else { 1158 dr = rdmadrmru->dr_next; 1159 while (dr->dr_status == DUP_INPROGRESS) { 1160 dr = dr->dr_next; 1161 if (dr == rdmadrmru->dr_next) { 1162 mutex_exit(&rdmadupreq_lock); 1163 return (DUP_ERROR); 1164 } 1165 } 1166 unhash(dr); 1167 if (dr->dr_resfree) { 1168 (*dr->dr_resfree)(dr->dr_resp.buf); 1169 } 1170 } 1171 dr->dr_resfree = NULL; 1172 rdmadrmru = dr; 1173 1174 dr->dr_xid = REQTOXID(req); 1175 dr->dr_prog = req->rq_prog; 1176 dr->dr_vers = req->rq_vers; 1177 dr->dr_proc = req->rq_proc; 1178 if (dr->dr_addr.maxlen < req->rq_xprt->xp_rtaddr.len) { 1179 if (dr->dr_addr.buf != NULL) 1180 kmem_free(dr->dr_addr.buf, dr->dr_addr.maxlen); 1181 dr->dr_addr.maxlen = req->rq_xprt->xp_rtaddr.len; 1182 dr->dr_addr.buf = kmem_alloc(dr->dr_addr.maxlen, KM_NOSLEEP); 1183 if (dr->dr_addr.buf == NULL) { 1184 dr->dr_addr.maxlen = 0; 1185 dr->dr_status = DUP_DROP; 1186 mutex_exit(&rdmadupreq_lock); 1187 return (DUP_ERROR); 1188 } 1189 } 1190 dr->dr_addr.len = req->rq_xprt->xp_rtaddr.len; 1191 bcopy(req->rq_xprt->xp_rtaddr.buf, dr->dr_addr.buf, dr->dr_addr.len); 1192 if (dr->dr_resp.maxlen < size) { 1193 if (dr->dr_resp.buf != NULL) 1194 kmem_free(dr->dr_resp.buf, dr->dr_resp.maxlen); 1195 dr->dr_resp.maxlen = (unsigned int)size; 1196 dr->dr_resp.buf = kmem_alloc(size, KM_NOSLEEP); 1197 if (dr->dr_resp.buf == NULL) { 1198 dr->dr_resp.maxlen = 0; 1199 dr->dr_status = DUP_DROP; 1200 mutex_exit(&rdmadupreq_lock); 1201 return (DUP_ERROR); 1202 } 1203 } 1204 dr->dr_status = DUP_INPROGRESS; 1205 1206 drhash = (uint32_t)DRHASH(dr); 1207 dr->dr_chain = rdmadrhashtbl[drhash]; 1208 rdmadrhashtbl[drhash] = dr; 1209 rdmadrhashstat[drhash]++; 1210 mutex_exit(&rdmadupreq_lock); 1211 *drpp = dr; 1212 return (DUP_NEW); 1213 } 1214 1215 /* 1216 * svc_rdma_kdupdone marks the request done (DUP_DONE or DUP_DROP) 1217 * and stores the response. 1218 */ 1219 static void 1220 svc_rdma_kdupdone(struct dupreq *dr, caddr_t res, void (*dis_resfree)(), 1221 int size, int status) 1222 { 1223 ASSERT(dr->dr_resfree == NULL); 1224 if (status == DUP_DONE) { 1225 bcopy(res, dr->dr_resp.buf, size); 1226 dr->dr_resfree = dis_resfree; 1227 } 1228 dr->dr_status = status; 1229 } 1230 1231 /* 1232 * This routine expects that the mutex, rdmadupreq_lock, is already held. 1233 */ 1234 static void 1235 unhash(struct dupreq *dr) 1236 { 1237 struct dupreq *drt; 1238 struct dupreq *drtprev = NULL; 1239 uint32_t drhash; 1240 1241 ASSERT(MUTEX_HELD(&rdmadupreq_lock)); 1242 1243 drhash = (uint32_t)DRHASH(dr); 1244 drt = rdmadrhashtbl[drhash]; 1245 while (drt != NULL) { 1246 if (drt == dr) { 1247 rdmadrhashstat[drhash]--; 1248 if (drtprev == NULL) { 1249 rdmadrhashtbl[drhash] = drt->dr_chain; 1250 } else { 1251 drtprev->dr_chain = drt->dr_chain; 1252 } 1253 return; 1254 } 1255 drtprev = drt; 1256 drt = drt->dr_chain; 1257 } 1258 } 1259 1260 bool_t 1261 rdma_get_wchunk(struct svc_req *req, iovec_t *iov, struct clist *wlist) 1262 { 1263 struct clist *clist; 1264 uint32_t tlen; 1265 1266 if (req->rq_xprt->xp_type != T_RDMA) { 1267 return (FALSE); 1268 } 1269 1270 tlen = 0; 1271 clist = wlist; 1272 while (clist) { 1273 tlen += clist->c_len; 1274 clist = clist->c_next; 1275 } 1276 1277 /* 1278 * set iov to addr+len of first segment of first wchunk of 1279 * wlist sent by client. krecv() already malloc'd a buffer 1280 * large enough, but registration is deferred until we write 1281 * the buffer back to (NFS) client using RDMA_WRITE. 1282 */ 1283 iov->iov_base = (caddr_t)(uintptr_t)wlist->w.c_saddr; 1284 iov->iov_len = tlen; 1285 1286 return (TRUE); 1287 } 1288