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 2008 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 (*rmod->rdma_ops->rdma_svc_listen)(svcdata); 364 } 365 366 void 367 svc_rdma_kstop(SVCMASTERXPRT *xprt) 368 { 369 struct rdma_svc_data *svcdata; 370 rdma_mod_t *rmod; 371 372 svcdata = &((struct rdma_data *)xprt->xp_p2)->rd_data; 373 rmod = ((struct rdma_data *)xprt->xp_p2)->r_mod; 374 375 /* 376 * Call the stop listener routine for each plugin. 377 */ 378 (*rmod->rdma_ops->rdma_svc_stop)(svcdata); 379 if (svcdata->active) 380 DTRACE_PROBE(krpc__e__svcrdma__kstop); 381 } 382 383 /* ARGSUSED */ 384 static void 385 svc_rdma_kclone_destroy(SVCXPRT *clone_xprt) 386 { 387 } 388 389 static bool_t 390 svc_rdma_krecv(SVCXPRT *clone_xprt, mblk_t *mp, struct rpc_msg *msg) 391 { 392 XDR *xdrs; 393 CONN *conn; 394 395 rdma_recv_data_t *rdp = (rdma_recv_data_t *)mp->b_rptr; 396 struct clone_rdma_data *crdp; 397 struct clist *cl = NULL; 398 struct clist *wcl = NULL; 399 struct clist *cllong = NULL; 400 401 rdma_stat status; 402 uint32_t vers, op, pos, xid; 403 uint32_t rdma_credit; 404 uint32_t wcl_total_length = 0; 405 bool_t wwl = FALSE; 406 407 crdp = (struct clone_rdma_data *)clone_xprt->xp_p2buf; 408 RSSTAT_INCR(rscalls); 409 conn = rdp->conn; 410 411 status = rdma_svc_postrecv(conn); 412 if (status != RDMA_SUCCESS) { 413 DTRACE_PROBE(krpc__e__svcrdma__krecv__postrecv); 414 goto badrpc_call; 415 } 416 417 xdrs = &clone_xprt->xp_xdrin; 418 xdrmem_create(xdrs, rdp->rpcmsg.addr, rdp->rpcmsg.len, XDR_DECODE); 419 xid = *(uint32_t *)rdp->rpcmsg.addr; 420 XDR_SETPOS(xdrs, sizeof (uint32_t)); 421 422 if (! xdr_u_int(xdrs, &vers) || 423 ! xdr_u_int(xdrs, &rdma_credit) || 424 ! xdr_u_int(xdrs, &op)) { 425 DTRACE_PROBE(krpc__e__svcrdma__krecv__uint); 426 goto xdr_err; 427 } 428 429 /* Checking if the status of the recv operation was normal */ 430 if (rdp->status != 0) { 431 DTRACE_PROBE1(krpc__e__svcrdma__krecv__invalid__status, 432 int, rdp->status); 433 goto badrpc_call; 434 } 435 436 if (! xdr_do_clist(xdrs, &cl)) { 437 DTRACE_PROBE(krpc__e__svcrdma__krecv__do__clist); 438 goto xdr_err; 439 } 440 441 if (!xdr_decode_wlist_svc(xdrs, &wcl, &wwl, &wcl_total_length, conn)) { 442 DTRACE_PROBE(krpc__e__svcrdma__krecv__decode__wlist); 443 if (cl) 444 clist_free(cl); 445 goto xdr_err; 446 } 447 crdp->cl_wlist = wcl; 448 449 crdp->cl_reply = NULL; 450 (void) xdr_decode_reply_wchunk(xdrs, &crdp->cl_reply); 451 452 /* 453 * A chunk at 0 offset indicates that the RPC call message 454 * is in a chunk. Get the RPC call message chunk. 455 */ 456 if (cl != NULL && op == RDMA_NOMSG) { 457 458 /* Remove RPC call message chunk from chunklist */ 459 cllong = cl; 460 cl = cl->c_next; 461 cllong->c_next = NULL; 462 463 464 /* Allocate and register memory for the RPC call msg chunk */ 465 cllong->rb_longbuf.type = RDMA_LONG_BUFFER; 466 cllong->rb_longbuf.len = cllong->c_len > LONG_REPLY_LEN ? 467 cllong->c_len : LONG_REPLY_LEN; 468 469 if (rdma_buf_alloc(conn, &cllong->rb_longbuf)) { 470 clist_free(cllong); 471 goto cll_malloc_err; 472 } 473 474 cllong->u.c_daddr3 = cllong->rb_longbuf.addr; 475 476 if (cllong->u.c_daddr == NULL) { 477 DTRACE_PROBE(krpc__e__svcrdma__krecv__nomem); 478 rdma_buf_free(conn, &cllong->rb_longbuf); 479 clist_free(cllong); 480 goto cll_malloc_err; 481 } 482 483 status = clist_register(conn, cllong, CLIST_REG_DST); 484 if (status) { 485 DTRACE_PROBE(krpc__e__svcrdma__krecv__clist__reg); 486 rdma_buf_free(conn, &cllong->rb_longbuf); 487 clist_free(cllong); 488 goto cll_malloc_err; 489 } 490 491 /* 492 * Now read the RPC call message in 493 */ 494 status = RDMA_READ(conn, cllong, WAIT); 495 if (status) { 496 DTRACE_PROBE(krpc__e__svcrdma__krecv__read); 497 (void) clist_deregister(conn, cllong, CLIST_REG_DST); 498 rdma_buf_free(conn, &cllong->rb_longbuf); 499 clist_free(cllong); 500 goto cll_malloc_err; 501 } 502 503 status = clist_syncmem(conn, cllong, CLIST_REG_DST); 504 (void) clist_deregister(conn, cllong, CLIST_REG_DST); 505 506 xdrrdma_create(xdrs, (caddr_t)(uintptr_t)cllong->u.c_daddr3, 507 cllong->c_len, 0, cl, XDR_DECODE, conn); 508 509 crdp->rpcbuf = cllong->rb_longbuf; 510 crdp->rpcbuf.len = cllong->c_len; 511 clist_free(cllong); 512 RDMA_BUF_FREE(conn, &rdp->rpcmsg); 513 } else { 514 pos = XDR_GETPOS(xdrs); 515 xdrrdma_create(xdrs, rdp->rpcmsg.addr + pos, 516 rdp->rpcmsg.len - pos, 0, cl, XDR_DECODE, conn); 517 crdp->rpcbuf = rdp->rpcmsg; 518 519 /* Use xdrrdmablk_ops to indicate there is a read chunk list */ 520 if (cl != NULL) { 521 int32_t flg = XDR_RDMA_RLIST_REG; 522 523 XDR_CONTROL(xdrs, XDR_RDMA_SET_FLAGS, &flg); 524 xdrs->x_ops = &xdrrdmablk_ops; 525 } 526 } 527 528 if (crdp->cl_wlist) { 529 int32_t flg = XDR_RDMA_WLIST_REG; 530 531 XDR_CONTROL(xdrs, XDR_RDMA_SET_WLIST, crdp->cl_wlist); 532 XDR_CONTROL(xdrs, XDR_RDMA_SET_FLAGS, &flg); 533 } 534 535 if (! xdr_callmsg(xdrs, msg)) { 536 DTRACE_PROBE(krpc__e__svcrdma__krecv__callmsg); 537 RSSTAT_INCR(rsxdrcall); 538 goto callmsg_err; 539 } 540 541 /* 542 * Point the remote transport address in the service_transport 543 * handle at the address in the request. 544 */ 545 clone_xprt->xp_rtaddr.buf = conn->c_raddr.buf; 546 clone_xprt->xp_rtaddr.len = conn->c_raddr.len; 547 clone_xprt->xp_rtaddr.maxlen = conn->c_raddr.len; 548 clone_xprt->xp_xid = xid; 549 crdp->conn = conn; 550 551 freeb(mp); 552 553 return (TRUE); 554 555 callmsg_err: 556 rdma_buf_free(conn, &crdp->rpcbuf); 557 558 cll_malloc_err: 559 if (cl) 560 clist_free(cl); 561 xdr_err: 562 XDR_DESTROY(xdrs); 563 564 badrpc_call: 565 RDMA_BUF_FREE(conn, &rdp->rpcmsg); 566 RDMA_REL_CONN(conn); 567 freeb(mp); 568 RSSTAT_INCR(rsbadcalls); 569 return (FALSE); 570 } 571 572 static int 573 svc_process_long_reply(SVCXPRT * clone_xprt, 574 xdrproc_t xdr_results, caddr_t xdr_location, 575 struct rpc_msg *msg, bool_t has_args, int *msglen, 576 int *freelen, int *numchunks, unsigned int *final_len) 577 { 578 int status; 579 XDR xdrslong; 580 struct clist *wcl = NULL; 581 int count = 0; 582 int alloc_len; 583 char *memp; 584 rdma_buf_t long_rpc = {0}; 585 struct clone_rdma_data *crdp; 586 587 crdp = (struct clone_rdma_data *)clone_xprt->xp_p2buf; 588 589 bzero(&xdrslong, sizeof (xdrslong)); 590 591 /* Choose a size for the long rpc response */ 592 if (MSG_IS_RPCSEC_GSS(msg)) { 593 alloc_len = RNDUP(MAX_AUTH_BYTES + *msglen); 594 } else { 595 alloc_len = RNDUP(*msglen); 596 } 597 598 if (alloc_len <= 64 * 1024) { 599 if (alloc_len > 32 * 1024) { 600 alloc_len = 64 * 1024; 601 } else { 602 if (alloc_len > 16 * 1024) { 603 alloc_len = 32 * 1024; 604 } else { 605 alloc_len = 16 * 1024; 606 } 607 } 608 } 609 610 long_rpc.type = RDMA_LONG_BUFFER; 611 long_rpc.len = alloc_len; 612 if (rdma_buf_alloc(crdp->conn, &long_rpc)) { 613 return (SVC_RDMA_FAIL); 614 } 615 616 memp = long_rpc.addr; 617 xdrmem_create(&xdrslong, memp, alloc_len, XDR_ENCODE); 618 619 msg->rm_xid = clone_xprt->xp_xid; 620 621 if (!(xdr_replymsg(&xdrslong, msg) && 622 (!has_args || SVCAUTH_WRAP(&clone_xprt->xp_auth, &xdrslong, 623 xdr_results, xdr_location)))) { 624 rdma_buf_free(crdp->conn, &long_rpc); 625 DTRACE_PROBE(krpc__e__svcrdma__longrep__authwrap); 626 return (SVC_RDMA_FAIL); 627 } 628 629 *final_len = XDR_GETPOS(&xdrslong); 630 631 *numchunks = 0; 632 *freelen = 0; 633 634 wcl = crdp->cl_reply; 635 wcl->rb_longbuf = long_rpc; 636 637 count = *final_len; 638 while (wcl != NULL) { 639 if (wcl->c_dmemhandle.mrc_rmr == 0) 640 break; 641 642 if (wcl->c_len > count) { 643 wcl->c_len = count; 644 } 645 wcl->w.c_saddr3 = (caddr_t)memp; 646 647 count -= wcl->c_len; 648 *numchunks += 1; 649 if (count == 0) 650 break; 651 memp += wcl->c_len; 652 wcl = wcl->c_next; 653 } 654 655 wcl = crdp->cl_reply; 656 657 /* 658 * MUST fail if there are still more data 659 */ 660 if (count > 0) { 661 rdma_buf_free(crdp->conn, &long_rpc); 662 DTRACE_PROBE(krpc__e__svcrdma__longrep__dlen__clist); 663 return (SVC_RDMA_FAIL); 664 } 665 666 if (clist_register(crdp->conn, wcl, CLIST_REG_SOURCE) != RDMA_SUCCESS) { 667 rdma_buf_free(crdp->conn, &long_rpc); 668 DTRACE_PROBE(krpc__e__svcrdma__longrep__clistreg); 669 return (SVC_RDMA_FAIL); 670 } 671 672 status = clist_syncmem(crdp->conn, wcl, CLIST_REG_SOURCE); 673 674 if (status) { 675 (void) clist_deregister(crdp->conn, wcl, CLIST_REG_SOURCE); 676 rdma_buf_free(crdp->conn, &long_rpc); 677 DTRACE_PROBE(krpc__e__svcrdma__longrep__syncmem); 678 return (SVC_RDMA_FAIL); 679 } 680 681 status = RDMA_WRITE(crdp->conn, wcl, WAIT); 682 683 (void) clist_deregister(crdp->conn, wcl, CLIST_REG_SOURCE); 684 rdma_buf_free(crdp->conn, &wcl->rb_longbuf); 685 686 if (status != RDMA_SUCCESS) { 687 DTRACE_PROBE(krpc__e__svcrdma__longrep__write); 688 return (SVC_RDMA_FAIL); 689 } 690 691 return (SVC_RDMA_SUCCESS); 692 } 693 694 695 static int 696 svc_compose_rpcmsg(SVCXPRT * clone_xprt, CONN * conn, xdrproc_t xdr_results, 697 caddr_t xdr_location, rdma_buf_t *rpcreply, XDR ** xdrs, 698 struct rpc_msg *msg, bool_t has_args, uint_t *len) 699 { 700 /* 701 * Get a pre-allocated buffer for rpc reply 702 */ 703 rpcreply->type = SEND_BUFFER; 704 if (rdma_buf_alloc(conn, rpcreply)) { 705 DTRACE_PROBE(krpc__e__svcrdma__rpcmsg__reply__nofreebufs); 706 return (SVC_RDMA_FAIL); 707 } 708 709 xdrrdma_create(*xdrs, rpcreply->addr, rpcreply->len, 710 0, NULL, XDR_ENCODE, conn); 711 712 msg->rm_xid = clone_xprt->xp_xid; 713 714 if (has_args) { 715 if (!(xdr_replymsg(*xdrs, msg) && 716 (!has_args || 717 SVCAUTH_WRAP(&clone_xprt->xp_auth, *xdrs, 718 xdr_results, xdr_location)))) { 719 rdma_buf_free(conn, rpcreply); 720 DTRACE_PROBE( 721 krpc__e__svcrdma__rpcmsg__reply__authwrap1); 722 return (SVC_RDMA_FAIL); 723 } 724 } else { 725 if (!xdr_replymsg(*xdrs, msg)) { 726 rdma_buf_free(conn, rpcreply); 727 DTRACE_PROBE( 728 krpc__e__svcrdma__rpcmsg__reply__authwrap2); 729 return (SVC_RDMA_FAIL); 730 } 731 } 732 733 *len = XDR_GETPOS(*xdrs); 734 735 return (SVC_RDMA_SUCCESS); 736 } 737 738 /* 739 * Send rpc reply. 740 */ 741 static bool_t 742 svc_rdma_ksend(SVCXPRT * clone_xprt, struct rpc_msg *msg) 743 { 744 XDR *xdrs_rpc = &(clone_xprt->xp_xdrout); 745 XDR xdrs_rhdr; 746 CONN *conn = NULL; 747 rdma_buf_t rbuf_resp = {0}, rbuf_rpc_resp = {0}; 748 749 struct clone_rdma_data *crdp; 750 struct clist *cl_read = NULL; 751 struct clist *cl_send = NULL; 752 struct clist *cl_write = NULL; 753 xdrproc_t xdr_results; /* results XDR encoding function */ 754 caddr_t xdr_location; /* response results pointer */ 755 756 int retval = FALSE; 757 int status, msglen, num_wreply_segments = 0; 758 uint32_t rdma_credit = 0; 759 int freelen = 0; 760 bool_t has_args; 761 uint_t final_resp_len, rdma_response_op, vers; 762 763 bzero(&xdrs_rhdr, sizeof (XDR)); 764 crdp = (struct clone_rdma_data *)clone_xprt->xp_p2buf; 765 conn = crdp->conn; 766 767 /* 768 * If there is a result procedure specified in the reply message, 769 * it will be processed in the xdr_replymsg and SVCAUTH_WRAP. 770 * We need to make sure it won't be processed twice, so we null 771 * it for xdr_replymsg here. 772 */ 773 has_args = FALSE; 774 if (msg->rm_reply.rp_stat == MSG_ACCEPTED && 775 msg->rm_reply.rp_acpt.ar_stat == SUCCESS) { 776 if ((xdr_results = msg->acpted_rply.ar_results.proc) != NULL) { 777 has_args = TRUE; 778 xdr_location = msg->acpted_rply.ar_results.where; 779 msg->acpted_rply.ar_results.proc = xdr_void; 780 msg->acpted_rply.ar_results.where = NULL; 781 } 782 } 783 784 /* 785 * Given the limit on the inline response size (RPC_MSG_SZ), 786 * there is a need to make a guess as to the overall size of 787 * the response. If the resultant size is beyond the inline 788 * size, then the server needs to use the "reply chunk list" 789 * provided by the client (if the client provided one). An 790 * example of this type of response would be a READDIR 791 * response (e.g. a small directory read would fit in RPC_MSG_SZ 792 * and that is the preference but it may not fit) 793 * 794 * Combine the encoded size and the size of the true results 795 * and then make the decision about where to encode and send results. 796 * 797 * One important note, this calculation is ignoring the size 798 * of the encoding of the authentication overhead. The reason 799 * for this is rooted in the complexities of access to the 800 * encoded size of RPCSEC_GSS related authentiation, 801 * integrity, and privacy. 802 * 803 * If it turns out that the encoded authentication bumps the 804 * response over the RPC_MSG_SZ limit, then it may need to 805 * attempt to encode for the reply chunk list. 806 */ 807 808 /* 809 * Calculating the "sizeof" the RPC response header and the 810 * encoded results. 811 */ 812 msglen = xdr_sizeof(xdr_replymsg, msg); 813 814 if (msglen > 0) { 815 RSSTAT_INCR(rstotalreplies); 816 } 817 if (has_args) 818 msglen += xdrrdma_sizeof(xdr_results, xdr_location, 819 rdma_minchunk, NULL, NULL); 820 821 DTRACE_PROBE1(krpc__i__svcrdma__ksend__msglen, int, msglen); 822 823 status = SVC_RDMA_SUCCESS; 824 825 if (msglen < RPC_MSG_SZ) { 826 /* 827 * Looks like the response will fit in the inline 828 * response; let's try 829 */ 830 RSSTAT_INCR(rstotalinlinereplies); 831 832 rdma_response_op = RDMA_MSG; 833 834 status = svc_compose_rpcmsg(clone_xprt, conn, xdr_results, 835 xdr_location, &rbuf_rpc_resp, &xdrs_rpc, msg, 836 has_args, &final_resp_len); 837 838 DTRACE_PROBE1(krpc__i__srdma__ksend__compose_status, 839 int, status); 840 DTRACE_PROBE1(krpc__i__srdma__ksend__compose_len, 841 int, final_resp_len); 842 843 if (status == SVC_RDMA_SUCCESS && crdp->cl_reply) { 844 clist_free(crdp->cl_reply); 845 crdp->cl_reply = NULL; 846 } 847 } 848 849 /* 850 * If the encode failed (size?) or the message really is 851 * larger than what is allowed, try the response chunk list. 852 */ 853 if (status != SVC_RDMA_SUCCESS || msglen >= RPC_MSG_SZ) { 854 /* 855 * attempting to use a reply chunk list when there 856 * isn't one won't get very far... 857 */ 858 if (crdp->cl_reply == NULL) { 859 DTRACE_PROBE(krpc__e__svcrdma__ksend__noreplycl); 860 goto out; 861 } 862 863 RSSTAT_INCR(rstotallongreplies); 864 865 msglen = xdr_sizeof(xdr_replymsg, msg); 866 msglen += xdrrdma_sizeof(xdr_results, xdr_location, 0, 867 NULL, NULL); 868 869 status = svc_process_long_reply(clone_xprt, xdr_results, 870 xdr_location, msg, has_args, &msglen, &freelen, 871 &num_wreply_segments, &final_resp_len); 872 873 DTRACE_PROBE1(krpc__i__svcrdma__ksend__longreplen, 874 int, final_resp_len); 875 876 if (status != SVC_RDMA_SUCCESS) { 877 DTRACE_PROBE(krpc__e__svcrdma__ksend__compose__failed); 878 goto out; 879 } 880 881 rdma_response_op = RDMA_NOMSG; 882 } 883 884 DTRACE_PROBE1(krpc__i__svcrdma__ksend__rdmamsg__len, 885 int, final_resp_len); 886 887 rbuf_resp.type = SEND_BUFFER; 888 if (rdma_buf_alloc(conn, &rbuf_resp)) { 889 rdma_buf_free(conn, &rbuf_rpc_resp); 890 DTRACE_PROBE(krpc__e__svcrdma__ksend__nofreebufs); 891 goto out; 892 } 893 894 rdma_credit = rdma_bufs_granted; 895 896 vers = RPCRDMA_VERS; 897 xdrmem_create(&xdrs_rhdr, rbuf_resp.addr, rbuf_resp.len, XDR_ENCODE); 898 (*(uint32_t *)rbuf_resp.addr) = msg->rm_xid; 899 /* Skip xid and set the xdr position accordingly. */ 900 XDR_SETPOS(&xdrs_rhdr, sizeof (uint32_t)); 901 if (!xdr_u_int(&xdrs_rhdr, &vers) || 902 !xdr_u_int(&xdrs_rhdr, &rdma_credit) || 903 !xdr_u_int(&xdrs_rhdr, &rdma_response_op)) { 904 rdma_buf_free(conn, &rbuf_rpc_resp); 905 rdma_buf_free(conn, &rbuf_resp); 906 DTRACE_PROBE(krpc__e__svcrdma__ksend__uint); 907 goto out; 908 } 909 910 /* 911 * Now XDR the read chunk list, actually always NULL 912 */ 913 (void) xdr_encode_rlist_svc(&xdrs_rhdr, cl_read); 914 915 /* 916 * encode write list -- we already drove RDMA_WRITEs 917 */ 918 cl_write = crdp->cl_wlist; 919 if (!xdr_encode_wlist(&xdrs_rhdr, cl_write)) { 920 DTRACE_PROBE(krpc__e__svcrdma__ksend__enc__wlist); 921 rdma_buf_free(conn, &rbuf_rpc_resp); 922 rdma_buf_free(conn, &rbuf_resp); 923 goto out; 924 } 925 926 /* 927 * XDR encode the RDMA_REPLY write chunk 928 */ 929 if (!xdr_encode_reply_wchunk(&xdrs_rhdr, crdp->cl_reply, 930 num_wreply_segments)) { 931 rdma_buf_free(conn, &rbuf_rpc_resp); 932 rdma_buf_free(conn, &rbuf_resp); 933 goto out; 934 } 935 936 clist_add(&cl_send, 0, XDR_GETPOS(&xdrs_rhdr), &rbuf_resp.handle, 937 rbuf_resp.addr, NULL, NULL); 938 939 if (rdma_response_op == RDMA_MSG) { 940 clist_add(&cl_send, 0, final_resp_len, &rbuf_rpc_resp.handle, 941 rbuf_rpc_resp.addr, NULL, NULL); 942 } 943 944 status = RDMA_SEND(conn, cl_send, msg->rm_xid); 945 946 if (status == RDMA_SUCCESS) { 947 retval = TRUE; 948 } 949 950 out: 951 /* 952 * Free up sendlist chunks 953 */ 954 if (cl_send != NULL) 955 clist_free(cl_send); 956 957 /* 958 * Destroy private data for xdr rdma 959 */ 960 if (clone_xprt->xp_xdrout.x_ops != NULL) { 961 XDR_DESTROY(&(clone_xprt->xp_xdrout)); 962 } 963 964 if (crdp->cl_reply) { 965 clist_free(crdp->cl_reply); 966 crdp->cl_reply = NULL; 967 } 968 969 /* 970 * This is completely disgusting. If public is set it is 971 * a pointer to a structure whose first field is the address 972 * of the function to free that structure and any related 973 * stuff. (see rrokfree in nfs_xdr.c). 974 */ 975 if (xdrs_rpc->x_public) { 976 /* LINTED pointer alignment */ 977 (**((int (**)()) xdrs_rpc->x_public)) (xdrs_rpc->x_public); 978 } 979 980 if (xdrs_rhdr.x_ops != NULL) { 981 XDR_DESTROY(&xdrs_rhdr); 982 } 983 984 return (retval); 985 } 986 987 /* 988 * Deserialize arguments. 989 */ 990 static bool_t 991 svc_rdma_kgetargs(SVCXPRT *clone_xprt, xdrproc_t xdr_args, caddr_t args_ptr) 992 { 993 if ((SVCAUTH_UNWRAP(&clone_xprt->xp_auth, &clone_xprt->xp_xdrin, 994 xdr_args, args_ptr)) != TRUE) 995 return (FALSE); 996 return (TRUE); 997 } 998 999 static bool_t 1000 svc_rdma_kfreeargs(SVCXPRT *clone_xprt, xdrproc_t xdr_args, 1001 caddr_t args_ptr) 1002 { 1003 struct clone_rdma_data *crdp; 1004 bool_t retval; 1005 1006 crdp = (struct clone_rdma_data *)clone_xprt->xp_p2buf; 1007 1008 /* 1009 * Free the args if needed then XDR_DESTROY 1010 */ 1011 if (args_ptr) { 1012 XDR *xdrs = &clone_xprt->xp_xdrin; 1013 1014 xdrs->x_op = XDR_FREE; 1015 retval = (*xdr_args)(xdrs, args_ptr); 1016 } 1017 1018 XDR_DESTROY(&(clone_xprt->xp_xdrin)); 1019 rdma_buf_free(crdp->conn, &crdp->rpcbuf); 1020 if (crdp->cl_reply) { 1021 clist_free(crdp->cl_reply); 1022 crdp->cl_reply = NULL; 1023 } 1024 RDMA_REL_CONN(crdp->conn); 1025 1026 return (retval); 1027 } 1028 1029 /* ARGSUSED */ 1030 static int32_t * 1031 svc_rdma_kgetres(SVCXPRT *clone_xprt, int size) 1032 { 1033 return (NULL); 1034 } 1035 1036 /* ARGSUSED */ 1037 static void 1038 svc_rdma_kfreeres(SVCXPRT *clone_xprt) 1039 { 1040 } 1041 1042 /* 1043 * the dup cacheing routines below provide a cache of non-failure 1044 * transaction id's. rpc service routines can use this to detect 1045 * retransmissions and re-send a non-failure response. 1046 */ 1047 1048 /* 1049 * MAXDUPREQS is the number of cached items. It should be adjusted 1050 * to the service load so that there is likely to be a response entry 1051 * when the first retransmission comes in. 1052 */ 1053 #define MAXDUPREQS 1024 1054 1055 /* 1056 * This should be appropriately scaled to MAXDUPREQS. 1057 */ 1058 #define DRHASHSZ 257 1059 1060 #if ((DRHASHSZ & (DRHASHSZ - 1)) == 0) 1061 #define XIDHASH(xid) ((xid) & (DRHASHSZ - 1)) 1062 #else 1063 #define XIDHASH(xid) ((xid) % DRHASHSZ) 1064 #endif 1065 #define DRHASH(dr) XIDHASH((dr)->dr_xid) 1066 #define REQTOXID(req) ((req)->rq_xprt->xp_xid) 1067 1068 static int rdmandupreqs = 0; 1069 int rdmamaxdupreqs = MAXDUPREQS; 1070 static kmutex_t rdmadupreq_lock; 1071 static struct dupreq *rdmadrhashtbl[DRHASHSZ]; 1072 static int rdmadrhashstat[DRHASHSZ]; 1073 1074 static void unhash(struct dupreq *); 1075 1076 /* 1077 * rdmadrmru points to the head of a circular linked list in lru order. 1078 * rdmadrmru->dr_next == drlru 1079 */ 1080 struct dupreq *rdmadrmru; 1081 1082 /* 1083 * svc_rdma_kdup searches the request cache and returns 0 if the 1084 * request is not found in the cache. If it is found, then it 1085 * returns the state of the request (in progress or done) and 1086 * the status or attributes that were part of the original reply. 1087 */ 1088 static int 1089 svc_rdma_kdup(struct svc_req *req, caddr_t res, int size, struct dupreq **drpp, 1090 bool_t *dupcachedp) 1091 { 1092 struct dupreq *dr; 1093 uint32_t xid; 1094 uint32_t drhash; 1095 int status; 1096 1097 xid = REQTOXID(req); 1098 mutex_enter(&rdmadupreq_lock); 1099 RSSTAT_INCR(rsdupchecks); 1100 /* 1101 * Check to see whether an entry already exists in the cache. 1102 */ 1103 dr = rdmadrhashtbl[XIDHASH(xid)]; 1104 while (dr != NULL) { 1105 if (dr->dr_xid == xid && 1106 dr->dr_proc == req->rq_proc && 1107 dr->dr_prog == req->rq_prog && 1108 dr->dr_vers == req->rq_vers && 1109 dr->dr_addr.len == req->rq_xprt->xp_rtaddr.len && 1110 bcmp((caddr_t)dr->dr_addr.buf, 1111 (caddr_t)req->rq_xprt->xp_rtaddr.buf, 1112 dr->dr_addr.len) == 0) { 1113 status = dr->dr_status; 1114 if (status == DUP_DONE) { 1115 bcopy(dr->dr_resp.buf, res, size); 1116 if (dupcachedp != NULL) 1117 *dupcachedp = (dr->dr_resfree != NULL); 1118 } else { 1119 dr->dr_status = DUP_INPROGRESS; 1120 *drpp = dr; 1121 } 1122 RSSTAT_INCR(rsdupreqs); 1123 mutex_exit(&rdmadupreq_lock); 1124 return (status); 1125 } 1126 dr = dr->dr_chain; 1127 } 1128 1129 /* 1130 * There wasn't an entry, either allocate a new one or recycle 1131 * an old one. 1132 */ 1133 if (rdmandupreqs < rdmamaxdupreqs) { 1134 dr = kmem_alloc(sizeof (*dr), KM_NOSLEEP); 1135 if (dr == NULL) { 1136 mutex_exit(&rdmadupreq_lock); 1137 return (DUP_ERROR); 1138 } 1139 dr->dr_resp.buf = NULL; 1140 dr->dr_resp.maxlen = 0; 1141 dr->dr_addr.buf = NULL; 1142 dr->dr_addr.maxlen = 0; 1143 if (rdmadrmru) { 1144 dr->dr_next = rdmadrmru->dr_next; 1145 rdmadrmru->dr_next = dr; 1146 } else { 1147 dr->dr_next = dr; 1148 } 1149 rdmandupreqs++; 1150 } else { 1151 dr = rdmadrmru->dr_next; 1152 while (dr->dr_status == DUP_INPROGRESS) { 1153 dr = dr->dr_next; 1154 if (dr == rdmadrmru->dr_next) { 1155 mutex_exit(&rdmadupreq_lock); 1156 return (DUP_ERROR); 1157 } 1158 } 1159 unhash(dr); 1160 if (dr->dr_resfree) { 1161 (*dr->dr_resfree)(dr->dr_resp.buf); 1162 } 1163 } 1164 dr->dr_resfree = NULL; 1165 rdmadrmru = dr; 1166 1167 dr->dr_xid = REQTOXID(req); 1168 dr->dr_prog = req->rq_prog; 1169 dr->dr_vers = req->rq_vers; 1170 dr->dr_proc = req->rq_proc; 1171 if (dr->dr_addr.maxlen < req->rq_xprt->xp_rtaddr.len) { 1172 if (dr->dr_addr.buf != NULL) 1173 kmem_free(dr->dr_addr.buf, dr->dr_addr.maxlen); 1174 dr->dr_addr.maxlen = req->rq_xprt->xp_rtaddr.len; 1175 dr->dr_addr.buf = kmem_alloc(dr->dr_addr.maxlen, KM_NOSLEEP); 1176 if (dr->dr_addr.buf == NULL) { 1177 dr->dr_addr.maxlen = 0; 1178 dr->dr_status = DUP_DROP; 1179 mutex_exit(&rdmadupreq_lock); 1180 return (DUP_ERROR); 1181 } 1182 } 1183 dr->dr_addr.len = req->rq_xprt->xp_rtaddr.len; 1184 bcopy(req->rq_xprt->xp_rtaddr.buf, dr->dr_addr.buf, dr->dr_addr.len); 1185 if (dr->dr_resp.maxlen < size) { 1186 if (dr->dr_resp.buf != NULL) 1187 kmem_free(dr->dr_resp.buf, dr->dr_resp.maxlen); 1188 dr->dr_resp.maxlen = (unsigned int)size; 1189 dr->dr_resp.buf = kmem_alloc(size, KM_NOSLEEP); 1190 if (dr->dr_resp.buf == NULL) { 1191 dr->dr_resp.maxlen = 0; 1192 dr->dr_status = DUP_DROP; 1193 mutex_exit(&rdmadupreq_lock); 1194 return (DUP_ERROR); 1195 } 1196 } 1197 dr->dr_status = DUP_INPROGRESS; 1198 1199 drhash = (uint32_t)DRHASH(dr); 1200 dr->dr_chain = rdmadrhashtbl[drhash]; 1201 rdmadrhashtbl[drhash] = dr; 1202 rdmadrhashstat[drhash]++; 1203 mutex_exit(&rdmadupreq_lock); 1204 *drpp = dr; 1205 return (DUP_NEW); 1206 } 1207 1208 /* 1209 * svc_rdma_kdupdone marks the request done (DUP_DONE or DUP_DROP) 1210 * and stores the response. 1211 */ 1212 static void 1213 svc_rdma_kdupdone(struct dupreq *dr, caddr_t res, void (*dis_resfree)(), 1214 int size, int status) 1215 { 1216 ASSERT(dr->dr_resfree == NULL); 1217 if (status == DUP_DONE) { 1218 bcopy(res, dr->dr_resp.buf, size); 1219 dr->dr_resfree = dis_resfree; 1220 } 1221 dr->dr_status = status; 1222 } 1223 1224 /* 1225 * This routine expects that the mutex, rdmadupreq_lock, is already held. 1226 */ 1227 static void 1228 unhash(struct dupreq *dr) 1229 { 1230 struct dupreq *drt; 1231 struct dupreq *drtprev = NULL; 1232 uint32_t drhash; 1233 1234 ASSERT(MUTEX_HELD(&rdmadupreq_lock)); 1235 1236 drhash = (uint32_t)DRHASH(dr); 1237 drt = rdmadrhashtbl[drhash]; 1238 while (drt != NULL) { 1239 if (drt == dr) { 1240 rdmadrhashstat[drhash]--; 1241 if (drtprev == NULL) { 1242 rdmadrhashtbl[drhash] = drt->dr_chain; 1243 } else { 1244 drtprev->dr_chain = drt->dr_chain; 1245 } 1246 return; 1247 } 1248 drtprev = drt; 1249 drt = drt->dr_chain; 1250 } 1251 } 1252 1253 bool_t 1254 rdma_get_wchunk(struct svc_req *req, iovec_t *iov, struct clist *wlist) 1255 { 1256 struct clist *clist; 1257 uint32_t tlen; 1258 1259 if (req->rq_xprt->xp_type != T_RDMA) { 1260 return (FALSE); 1261 } 1262 1263 tlen = 0; 1264 clist = wlist; 1265 while (clist) { 1266 tlen += clist->c_len; 1267 clist = clist->c_next; 1268 } 1269 1270 /* 1271 * set iov to addr+len of first segment of first wchunk of 1272 * wlist sent by client. krecv() already malloc'd a buffer 1273 * large enough, but registration is deferred until we write 1274 * the buffer back to (NFS) client using RDMA_WRITE. 1275 */ 1276 iov->iov_base = (caddr_t)(uintptr_t)wlist->w.c_saddr; 1277 iov->iov_len = tlen; 1278 1279 return (TRUE); 1280 } 1281