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); 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); 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 DTRACE_PROBE1(krpc__i__replylen, uint_t, *final_len); 639 *numchunks = 0; 640 *freelen = 0; 641 642 wcl = crdp->cl_reply; 643 wcl->rb_longbuf = long_rpc; 644 645 count = *final_len; 646 while ((wcl != NULL) && (count > 0)) { 647 648 if (wcl->c_dmemhandle.mrc_rmr == 0) 649 break; 650 651 DTRACE_PROBE2(krpc__i__write__chunks, uint32_t, count, 652 uint32_t, wcl->c_len); 653 654 if (wcl->c_len > count) { 655 wcl->c_len = count; 656 } 657 wcl->w.c_saddr3 = (caddr_t)memp; 658 659 count -= wcl->c_len; 660 *numchunks += 1; 661 memp += wcl->c_len; 662 wcl = wcl->c_next; 663 } 664 665 /* 666 * Make rest of the chunks 0-len 667 */ 668 while (wcl != NULL) { 669 if (wcl->c_dmemhandle.mrc_rmr == 0) 670 break; 671 wcl->c_len = 0; 672 wcl = wcl->c_next; 673 } 674 675 wcl = crdp->cl_reply; 676 677 /* 678 * MUST fail if there are still more data 679 */ 680 if (count > 0) { 681 rdma_buf_free(crdp->conn, &long_rpc); 682 DTRACE_PROBE(krpc__e__svcrdma__longrep__dlen__clist); 683 return (SVC_RDMA_FAIL); 684 } 685 686 if (clist_register(crdp->conn, wcl, CLIST_REG_SOURCE) != RDMA_SUCCESS) { 687 rdma_buf_free(crdp->conn, &long_rpc); 688 DTRACE_PROBE(krpc__e__svcrdma__longrep__clistreg); 689 return (SVC_RDMA_FAIL); 690 } 691 692 status = clist_syncmem(crdp->conn, wcl, CLIST_REG_SOURCE); 693 694 if (status) { 695 (void) clist_deregister(crdp->conn, wcl); 696 rdma_buf_free(crdp->conn, &long_rpc); 697 DTRACE_PROBE(krpc__e__svcrdma__longrep__syncmem); 698 return (SVC_RDMA_FAIL); 699 } 700 701 status = RDMA_WRITE(crdp->conn, wcl, WAIT); 702 703 (void) clist_deregister(crdp->conn, wcl); 704 rdma_buf_free(crdp->conn, &wcl->rb_longbuf); 705 706 if (status != RDMA_SUCCESS) { 707 DTRACE_PROBE(krpc__e__svcrdma__longrep__write); 708 return (SVC_RDMA_FAIL); 709 } 710 711 return (SVC_RDMA_SUCCESS); 712 } 713 714 715 static int 716 svc_compose_rpcmsg(SVCXPRT * clone_xprt, CONN * conn, xdrproc_t xdr_results, 717 caddr_t xdr_location, rdma_buf_t *rpcreply, XDR ** xdrs, 718 struct rpc_msg *msg, bool_t has_args, uint_t *len) 719 { 720 /* 721 * Get a pre-allocated buffer for rpc reply 722 */ 723 rpcreply->type = SEND_BUFFER; 724 if (rdma_buf_alloc(conn, rpcreply)) { 725 DTRACE_PROBE(krpc__e__svcrdma__rpcmsg__reply__nofreebufs); 726 return (SVC_RDMA_FAIL); 727 } 728 729 xdrrdma_create(*xdrs, rpcreply->addr, rpcreply->len, 730 0, NULL, XDR_ENCODE, conn); 731 732 msg->rm_xid = clone_xprt->xp_xid; 733 734 if (has_args) { 735 if (!(xdr_replymsg(*xdrs, msg) && 736 (!has_args || 737 SVCAUTH_WRAP(&clone_xprt->xp_auth, *xdrs, 738 xdr_results, xdr_location)))) { 739 rdma_buf_free(conn, rpcreply); 740 DTRACE_PROBE( 741 krpc__e__svcrdma__rpcmsg__reply__authwrap1); 742 return (SVC_RDMA_FAIL); 743 } 744 } else { 745 if (!xdr_replymsg(*xdrs, msg)) { 746 rdma_buf_free(conn, rpcreply); 747 DTRACE_PROBE( 748 krpc__e__svcrdma__rpcmsg__reply__authwrap2); 749 return (SVC_RDMA_FAIL); 750 } 751 } 752 753 *len = XDR_GETPOS(*xdrs); 754 755 return (SVC_RDMA_SUCCESS); 756 } 757 758 /* 759 * Send rpc reply. 760 */ 761 static bool_t 762 svc_rdma_ksend(SVCXPRT * clone_xprt, struct rpc_msg *msg) 763 { 764 XDR *xdrs_rpc = &(clone_xprt->xp_xdrout); 765 XDR xdrs_rhdr; 766 CONN *conn = NULL; 767 rdma_buf_t rbuf_resp = {0}, rbuf_rpc_resp = {0}; 768 769 struct clone_rdma_data *crdp; 770 struct clist *cl_read = NULL; 771 struct clist *cl_send = NULL; 772 struct clist *cl_write = NULL; 773 xdrproc_t xdr_results; /* results XDR encoding function */ 774 caddr_t xdr_location; /* response results pointer */ 775 776 int retval = FALSE; 777 int status, msglen, num_wreply_segments = 0; 778 uint32_t rdma_credit = 0; 779 int freelen = 0; 780 bool_t has_args; 781 uint_t final_resp_len, rdma_response_op, vers; 782 783 bzero(&xdrs_rhdr, sizeof (XDR)); 784 crdp = (struct clone_rdma_data *)clone_xprt->xp_p2buf; 785 conn = crdp->conn; 786 787 /* 788 * If there is a result procedure specified in the reply message, 789 * it will be processed in the xdr_replymsg and SVCAUTH_WRAP. 790 * We need to make sure it won't be processed twice, so we null 791 * it for xdr_replymsg here. 792 */ 793 has_args = FALSE; 794 if (msg->rm_reply.rp_stat == MSG_ACCEPTED && 795 msg->rm_reply.rp_acpt.ar_stat == SUCCESS) { 796 if ((xdr_results = msg->acpted_rply.ar_results.proc) != NULL) { 797 has_args = TRUE; 798 xdr_location = msg->acpted_rply.ar_results.where; 799 msg->acpted_rply.ar_results.proc = xdr_void; 800 msg->acpted_rply.ar_results.where = NULL; 801 } 802 } 803 804 /* 805 * Given the limit on the inline response size (RPC_MSG_SZ), 806 * there is a need to make a guess as to the overall size of 807 * the response. If the resultant size is beyond the inline 808 * size, then the server needs to use the "reply chunk list" 809 * provided by the client (if the client provided one). An 810 * example of this type of response would be a READDIR 811 * response (e.g. a small directory read would fit in RPC_MSG_SZ 812 * and that is the preference but it may not fit) 813 * 814 * Combine the encoded size and the size of the true results 815 * and then make the decision about where to encode and send results. 816 * 817 * One important note, this calculation is ignoring the size 818 * of the encoding of the authentication overhead. The reason 819 * for this is rooted in the complexities of access to the 820 * encoded size of RPCSEC_GSS related authentiation, 821 * integrity, and privacy. 822 * 823 * If it turns out that the encoded authentication bumps the 824 * response over the RPC_MSG_SZ limit, then it may need to 825 * attempt to encode for the reply chunk list. 826 */ 827 828 /* 829 * Calculating the "sizeof" the RPC response header and the 830 * encoded results. 831 */ 832 msglen = xdr_sizeof(xdr_replymsg, msg); 833 834 if (msglen > 0) { 835 RSSTAT_INCR(rstotalreplies); 836 } 837 if (has_args) 838 msglen += xdrrdma_sizeof(xdr_results, xdr_location, 839 rdma_minchunk, NULL, NULL); 840 841 DTRACE_PROBE1(krpc__i__svcrdma__ksend__msglen, int, msglen); 842 843 status = SVC_RDMA_SUCCESS; 844 845 if (msglen < RPC_MSG_SZ) { 846 /* 847 * Looks like the response will fit in the inline 848 * response; let's try 849 */ 850 RSSTAT_INCR(rstotalinlinereplies); 851 852 rdma_response_op = RDMA_MSG; 853 854 status = svc_compose_rpcmsg(clone_xprt, conn, xdr_results, 855 xdr_location, &rbuf_rpc_resp, &xdrs_rpc, msg, 856 has_args, &final_resp_len); 857 858 DTRACE_PROBE1(krpc__i__srdma__ksend__compose_status, 859 int, status); 860 DTRACE_PROBE1(krpc__i__srdma__ksend__compose_len, 861 int, final_resp_len); 862 863 if (status == SVC_RDMA_SUCCESS && crdp->cl_reply) { 864 clist_free(crdp->cl_reply); 865 crdp->cl_reply = NULL; 866 } 867 } 868 869 /* 870 * If the encode failed (size?) or the message really is 871 * larger than what is allowed, try the response chunk list. 872 */ 873 if (status != SVC_RDMA_SUCCESS || msglen >= RPC_MSG_SZ) { 874 /* 875 * attempting to use a reply chunk list when there 876 * isn't one won't get very far... 877 */ 878 if (crdp->cl_reply == NULL) { 879 DTRACE_PROBE(krpc__e__svcrdma__ksend__noreplycl); 880 goto out; 881 } 882 883 RSSTAT_INCR(rstotallongreplies); 884 885 msglen = xdr_sizeof(xdr_replymsg, msg); 886 msglen += xdrrdma_sizeof(xdr_results, xdr_location, 0, 887 NULL, NULL); 888 889 status = svc_process_long_reply(clone_xprt, xdr_results, 890 xdr_location, msg, has_args, &msglen, &freelen, 891 &num_wreply_segments, &final_resp_len); 892 893 DTRACE_PROBE1(krpc__i__svcrdma__ksend__longreplen, 894 int, final_resp_len); 895 896 if (status != SVC_RDMA_SUCCESS) { 897 DTRACE_PROBE(krpc__e__svcrdma__ksend__compose__failed); 898 goto out; 899 } 900 901 rdma_response_op = RDMA_NOMSG; 902 } 903 904 DTRACE_PROBE1(krpc__i__svcrdma__ksend__rdmamsg__len, 905 int, final_resp_len); 906 907 rbuf_resp.type = SEND_BUFFER; 908 if (rdma_buf_alloc(conn, &rbuf_resp)) { 909 rdma_buf_free(conn, &rbuf_rpc_resp); 910 DTRACE_PROBE(krpc__e__svcrdma__ksend__nofreebufs); 911 goto out; 912 } 913 914 rdma_credit = rdma_bufs_granted; 915 916 vers = RPCRDMA_VERS; 917 xdrmem_create(&xdrs_rhdr, rbuf_resp.addr, rbuf_resp.len, XDR_ENCODE); 918 (*(uint32_t *)rbuf_resp.addr) = msg->rm_xid; 919 /* Skip xid and set the xdr position accordingly. */ 920 XDR_SETPOS(&xdrs_rhdr, sizeof (uint32_t)); 921 if (!xdr_u_int(&xdrs_rhdr, &vers) || 922 !xdr_u_int(&xdrs_rhdr, &rdma_credit) || 923 !xdr_u_int(&xdrs_rhdr, &rdma_response_op)) { 924 rdma_buf_free(conn, &rbuf_rpc_resp); 925 rdma_buf_free(conn, &rbuf_resp); 926 DTRACE_PROBE(krpc__e__svcrdma__ksend__uint); 927 goto out; 928 } 929 930 /* 931 * Now XDR the read chunk list, actually always NULL 932 */ 933 (void) xdr_encode_rlist_svc(&xdrs_rhdr, cl_read); 934 935 /* 936 * encode write list -- we already drove RDMA_WRITEs 937 */ 938 cl_write = crdp->cl_wlist; 939 if (!xdr_encode_wlist(&xdrs_rhdr, cl_write)) { 940 DTRACE_PROBE(krpc__e__svcrdma__ksend__enc__wlist); 941 rdma_buf_free(conn, &rbuf_rpc_resp); 942 rdma_buf_free(conn, &rbuf_resp); 943 goto out; 944 } 945 946 /* 947 * XDR encode the RDMA_REPLY write chunk 948 */ 949 if (!xdr_encode_reply_wchunk(&xdrs_rhdr, crdp->cl_reply, 950 num_wreply_segments)) { 951 rdma_buf_free(conn, &rbuf_rpc_resp); 952 rdma_buf_free(conn, &rbuf_resp); 953 goto out; 954 } 955 956 clist_add(&cl_send, 0, XDR_GETPOS(&xdrs_rhdr), &rbuf_resp.handle, 957 rbuf_resp.addr, NULL, NULL); 958 959 if (rdma_response_op == RDMA_MSG) { 960 clist_add(&cl_send, 0, final_resp_len, &rbuf_rpc_resp.handle, 961 rbuf_rpc_resp.addr, NULL, NULL); 962 } 963 964 status = RDMA_SEND(conn, cl_send, msg->rm_xid); 965 966 if (status == RDMA_SUCCESS) { 967 retval = TRUE; 968 } 969 970 out: 971 /* 972 * Free up sendlist chunks 973 */ 974 if (cl_send != NULL) 975 clist_free(cl_send); 976 977 /* 978 * Destroy private data for xdr rdma 979 */ 980 if (clone_xprt->xp_xdrout.x_ops != NULL) { 981 XDR_DESTROY(&(clone_xprt->xp_xdrout)); 982 } 983 984 if (crdp->cl_reply) { 985 clist_free(crdp->cl_reply); 986 crdp->cl_reply = NULL; 987 } 988 989 /* 990 * This is completely disgusting. If public is set it is 991 * a pointer to a structure whose first field is the address 992 * of the function to free that structure and any related 993 * stuff. (see rrokfree in nfs_xdr.c). 994 */ 995 if (xdrs_rpc->x_public) { 996 /* LINTED pointer alignment */ 997 (**((int (**)()) xdrs_rpc->x_public)) (xdrs_rpc->x_public); 998 } 999 1000 if (xdrs_rhdr.x_ops != NULL) { 1001 XDR_DESTROY(&xdrs_rhdr); 1002 } 1003 1004 return (retval); 1005 } 1006 1007 /* 1008 * Deserialize arguments. 1009 */ 1010 static bool_t 1011 svc_rdma_kgetargs(SVCXPRT *clone_xprt, xdrproc_t xdr_args, caddr_t args_ptr) 1012 { 1013 if ((SVCAUTH_UNWRAP(&clone_xprt->xp_auth, &clone_xprt->xp_xdrin, 1014 xdr_args, args_ptr)) != TRUE) 1015 return (FALSE); 1016 return (TRUE); 1017 } 1018 1019 static bool_t 1020 svc_rdma_kfreeargs(SVCXPRT *clone_xprt, xdrproc_t xdr_args, 1021 caddr_t args_ptr) 1022 { 1023 struct clone_rdma_data *crdp; 1024 bool_t retval; 1025 1026 crdp = (struct clone_rdma_data *)clone_xprt->xp_p2buf; 1027 1028 /* 1029 * Free the args if needed then XDR_DESTROY 1030 */ 1031 if (args_ptr) { 1032 XDR *xdrs = &clone_xprt->xp_xdrin; 1033 1034 xdrs->x_op = XDR_FREE; 1035 retval = (*xdr_args)(xdrs, args_ptr); 1036 } 1037 1038 XDR_DESTROY(&(clone_xprt->xp_xdrin)); 1039 rdma_buf_free(crdp->conn, &crdp->rpcbuf); 1040 if (crdp->cl_reply) { 1041 clist_free(crdp->cl_reply); 1042 crdp->cl_reply = NULL; 1043 } 1044 RDMA_REL_CONN(crdp->conn); 1045 1046 return (retval); 1047 } 1048 1049 /* ARGSUSED */ 1050 static int32_t * 1051 svc_rdma_kgetres(SVCXPRT *clone_xprt, int size) 1052 { 1053 return (NULL); 1054 } 1055 1056 /* ARGSUSED */ 1057 static void 1058 svc_rdma_kfreeres(SVCXPRT *clone_xprt) 1059 { 1060 } 1061 1062 /* 1063 * the dup cacheing routines below provide a cache of non-failure 1064 * transaction id's. rpc service routines can use this to detect 1065 * retransmissions and re-send a non-failure response. 1066 */ 1067 1068 /* 1069 * MAXDUPREQS is the number of cached items. It should be adjusted 1070 * to the service load so that there is likely to be a response entry 1071 * when the first retransmission comes in. 1072 */ 1073 #define MAXDUPREQS 1024 1074 1075 /* 1076 * This should be appropriately scaled to MAXDUPREQS. 1077 */ 1078 #define DRHASHSZ 257 1079 1080 #if ((DRHASHSZ & (DRHASHSZ - 1)) == 0) 1081 #define XIDHASH(xid) ((xid) & (DRHASHSZ - 1)) 1082 #else 1083 #define XIDHASH(xid) ((xid) % DRHASHSZ) 1084 #endif 1085 #define DRHASH(dr) XIDHASH((dr)->dr_xid) 1086 #define REQTOXID(req) ((req)->rq_xprt->xp_xid) 1087 1088 static int rdmandupreqs = 0; 1089 int rdmamaxdupreqs = MAXDUPREQS; 1090 static kmutex_t rdmadupreq_lock; 1091 static struct dupreq *rdmadrhashtbl[DRHASHSZ]; 1092 static int rdmadrhashstat[DRHASHSZ]; 1093 1094 static void unhash(struct dupreq *); 1095 1096 /* 1097 * rdmadrmru points to the head of a circular linked list in lru order. 1098 * rdmadrmru->dr_next == drlru 1099 */ 1100 struct dupreq *rdmadrmru; 1101 1102 /* 1103 * svc_rdma_kdup searches the request cache and returns 0 if the 1104 * request is not found in the cache. If it is found, then it 1105 * returns the state of the request (in progress or done) and 1106 * the status or attributes that were part of the original reply. 1107 */ 1108 static int 1109 svc_rdma_kdup(struct svc_req *req, caddr_t res, int size, struct dupreq **drpp, 1110 bool_t *dupcachedp) 1111 { 1112 struct dupreq *dr; 1113 uint32_t xid; 1114 uint32_t drhash; 1115 int status; 1116 1117 xid = REQTOXID(req); 1118 mutex_enter(&rdmadupreq_lock); 1119 RSSTAT_INCR(rsdupchecks); 1120 /* 1121 * Check to see whether an entry already exists in the cache. 1122 */ 1123 dr = rdmadrhashtbl[XIDHASH(xid)]; 1124 while (dr != NULL) { 1125 if (dr->dr_xid == xid && 1126 dr->dr_proc == req->rq_proc && 1127 dr->dr_prog == req->rq_prog && 1128 dr->dr_vers == req->rq_vers && 1129 dr->dr_addr.len == req->rq_xprt->xp_rtaddr.len && 1130 bcmp((caddr_t)dr->dr_addr.buf, 1131 (caddr_t)req->rq_xprt->xp_rtaddr.buf, 1132 dr->dr_addr.len) == 0) { 1133 status = dr->dr_status; 1134 if (status == DUP_DONE) { 1135 bcopy(dr->dr_resp.buf, res, size); 1136 if (dupcachedp != NULL) 1137 *dupcachedp = (dr->dr_resfree != NULL); 1138 } else { 1139 dr->dr_status = DUP_INPROGRESS; 1140 *drpp = dr; 1141 } 1142 RSSTAT_INCR(rsdupreqs); 1143 mutex_exit(&rdmadupreq_lock); 1144 return (status); 1145 } 1146 dr = dr->dr_chain; 1147 } 1148 1149 /* 1150 * There wasn't an entry, either allocate a new one or recycle 1151 * an old one. 1152 */ 1153 if (rdmandupreqs < rdmamaxdupreqs) { 1154 dr = kmem_alloc(sizeof (*dr), KM_NOSLEEP); 1155 if (dr == NULL) { 1156 mutex_exit(&rdmadupreq_lock); 1157 return (DUP_ERROR); 1158 } 1159 dr->dr_resp.buf = NULL; 1160 dr->dr_resp.maxlen = 0; 1161 dr->dr_addr.buf = NULL; 1162 dr->dr_addr.maxlen = 0; 1163 if (rdmadrmru) { 1164 dr->dr_next = rdmadrmru->dr_next; 1165 rdmadrmru->dr_next = dr; 1166 } else { 1167 dr->dr_next = dr; 1168 } 1169 rdmandupreqs++; 1170 } else { 1171 dr = rdmadrmru->dr_next; 1172 while (dr->dr_status == DUP_INPROGRESS) { 1173 dr = dr->dr_next; 1174 if (dr == rdmadrmru->dr_next) { 1175 mutex_exit(&rdmadupreq_lock); 1176 return (DUP_ERROR); 1177 } 1178 } 1179 unhash(dr); 1180 if (dr->dr_resfree) { 1181 (*dr->dr_resfree)(dr->dr_resp.buf); 1182 } 1183 } 1184 dr->dr_resfree = NULL; 1185 rdmadrmru = dr; 1186 1187 dr->dr_xid = REQTOXID(req); 1188 dr->dr_prog = req->rq_prog; 1189 dr->dr_vers = req->rq_vers; 1190 dr->dr_proc = req->rq_proc; 1191 if (dr->dr_addr.maxlen < req->rq_xprt->xp_rtaddr.len) { 1192 if (dr->dr_addr.buf != NULL) 1193 kmem_free(dr->dr_addr.buf, dr->dr_addr.maxlen); 1194 dr->dr_addr.maxlen = req->rq_xprt->xp_rtaddr.len; 1195 dr->dr_addr.buf = kmem_alloc(dr->dr_addr.maxlen, KM_NOSLEEP); 1196 if (dr->dr_addr.buf == NULL) { 1197 dr->dr_addr.maxlen = 0; 1198 dr->dr_status = DUP_DROP; 1199 mutex_exit(&rdmadupreq_lock); 1200 return (DUP_ERROR); 1201 } 1202 } 1203 dr->dr_addr.len = req->rq_xprt->xp_rtaddr.len; 1204 bcopy(req->rq_xprt->xp_rtaddr.buf, dr->dr_addr.buf, dr->dr_addr.len); 1205 if (dr->dr_resp.maxlen < size) { 1206 if (dr->dr_resp.buf != NULL) 1207 kmem_free(dr->dr_resp.buf, dr->dr_resp.maxlen); 1208 dr->dr_resp.maxlen = (unsigned int)size; 1209 dr->dr_resp.buf = kmem_alloc(size, KM_NOSLEEP); 1210 if (dr->dr_resp.buf == NULL) { 1211 dr->dr_resp.maxlen = 0; 1212 dr->dr_status = DUP_DROP; 1213 mutex_exit(&rdmadupreq_lock); 1214 return (DUP_ERROR); 1215 } 1216 } 1217 dr->dr_status = DUP_INPROGRESS; 1218 1219 drhash = (uint32_t)DRHASH(dr); 1220 dr->dr_chain = rdmadrhashtbl[drhash]; 1221 rdmadrhashtbl[drhash] = dr; 1222 rdmadrhashstat[drhash]++; 1223 mutex_exit(&rdmadupreq_lock); 1224 *drpp = dr; 1225 return (DUP_NEW); 1226 } 1227 1228 /* 1229 * svc_rdma_kdupdone marks the request done (DUP_DONE or DUP_DROP) 1230 * and stores the response. 1231 */ 1232 static void 1233 svc_rdma_kdupdone(struct dupreq *dr, caddr_t res, void (*dis_resfree)(), 1234 int size, int status) 1235 { 1236 ASSERT(dr->dr_resfree == NULL); 1237 if (status == DUP_DONE) { 1238 bcopy(res, dr->dr_resp.buf, size); 1239 dr->dr_resfree = dis_resfree; 1240 } 1241 dr->dr_status = status; 1242 } 1243 1244 /* 1245 * This routine expects that the mutex, rdmadupreq_lock, is already held. 1246 */ 1247 static void 1248 unhash(struct dupreq *dr) 1249 { 1250 struct dupreq *drt; 1251 struct dupreq *drtprev = NULL; 1252 uint32_t drhash; 1253 1254 ASSERT(MUTEX_HELD(&rdmadupreq_lock)); 1255 1256 drhash = (uint32_t)DRHASH(dr); 1257 drt = rdmadrhashtbl[drhash]; 1258 while (drt != NULL) { 1259 if (drt == dr) { 1260 rdmadrhashstat[drhash]--; 1261 if (drtprev == NULL) { 1262 rdmadrhashtbl[drhash] = drt->dr_chain; 1263 } else { 1264 drtprev->dr_chain = drt->dr_chain; 1265 } 1266 return; 1267 } 1268 drtprev = drt; 1269 drt = drt->dr_chain; 1270 } 1271 } 1272 1273 bool_t 1274 rdma_get_wchunk(struct svc_req *req, iovec_t *iov, struct clist *wlist) 1275 { 1276 struct clist *clist; 1277 uint32_t tlen; 1278 1279 if (req->rq_xprt->xp_type != T_RDMA) { 1280 return (FALSE); 1281 } 1282 1283 tlen = 0; 1284 clist = wlist; 1285 while (clist) { 1286 tlen += clist->c_len; 1287 clist = clist->c_next; 1288 } 1289 1290 /* 1291 * set iov to addr+len of first segment of first wchunk of 1292 * wlist sent by client. krecv() already malloc'd a buffer 1293 * large enough, but registration is deferred until we write 1294 * the buffer back to (NFS) client using RDMA_WRITE. 1295 */ 1296 iov->iov_base = (caddr_t)(uintptr_t)wlist->w.c_saddr; 1297 iov->iov_len = tlen; 1298 1299 return (TRUE); 1300 } 1301 1302 /* 1303 * routine to setup the read chunk lists 1304 */ 1305 1306 int 1307 rdma_setup_read_chunks(struct clist *wcl, uint32_t count, int *wcl_len) 1308 { 1309 int data_len, avail_len; 1310 uint_t round_len; 1311 1312 data_len = avail_len = 0; 1313 1314 while (wcl != NULL && count > 0) { 1315 if (wcl->c_dmemhandle.mrc_rmr == 0) 1316 break; 1317 1318 if (wcl->c_len < count) { 1319 data_len += wcl->c_len; 1320 avail_len = 0; 1321 } else { 1322 data_len += count; 1323 avail_len = wcl->c_len - count; 1324 wcl->c_len = count; 1325 } 1326 count -= wcl->c_len; 1327 1328 if (count == 0) 1329 break; 1330 1331 wcl = wcl->c_next; 1332 } 1333 1334 /* 1335 * MUST fail if there are still more data 1336 */ 1337 if (count > 0) { 1338 DTRACE_PROBE2(krpc__e__rdma_setup_read_chunks_clist_len, 1339 int, data_len, int, count); 1340 return (FALSE); 1341 } 1342 1343 /* 1344 * Round up the last chunk to 4-byte boundary 1345 */ 1346 *wcl_len = roundup(data_len, BYTES_PER_XDR_UNIT); 1347 round_len = *wcl_len - data_len; 1348 1349 if (round_len) { 1350 1351 /* 1352 * If there is space in the current chunk, 1353 * add the roundup to the chunk. 1354 */ 1355 if (avail_len >= round_len) { 1356 wcl->c_len += round_len; 1357 } else { 1358 /* 1359 * try the next one. 1360 */ 1361 wcl = wcl->c_next; 1362 if ((wcl == NULL) || (wcl->c_len < round_len)) { 1363 DTRACE_PROBE1( 1364 krpc__e__rdma_setup_read_chunks_rndup, 1365 int, round_len); 1366 return (FALSE); 1367 } 1368 wcl->c_len = round_len; 1369 } 1370 } 1371 1372 wcl = wcl->c_next; 1373 1374 /* 1375 * Make rest of the chunks 0-len 1376 */ 1377 1378 clist_zero_len(wcl); 1379 1380 return (TRUE); 1381 } 1382