1 /* $NetBSD: clnt_dg.c,v 1.4 2000/07/14 08:40:41 fvdl Exp $ */ 2 3 /*- 4 * SPDX-License-Identifier: BSD-3-Clause 5 * 6 * Copyright (c) 2009, Sun Microsystems, Inc. 7 * All rights reserved. 8 * 9 * Redistribution and use in source and binary forms, with or without 10 * modification, are permitted provided that the following conditions are met: 11 * - Redistributions of source code must retain the above copyright notice, 12 * this list of conditions and the following disclaimer. 13 * - Redistributions in binary form must reproduce the above copyright notice, 14 * this list of conditions and the following disclaimer in the documentation 15 * and/or other materials provided with the distribution. 16 * - Neither the name of Sun Microsystems, Inc. nor the names of its 17 * contributors may be used to endorse or promote products derived 18 * from this software without specific prior written permission. 19 * 20 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 21 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 22 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 23 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE 24 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 25 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 26 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 27 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 28 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 29 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 30 * POSSIBILITY OF SUCH DAMAGE. 31 */ 32 /* 33 * Copyright (c) 1986-1991 by Sun Microsystems Inc. 34 */ 35 36 #if defined(LIBC_SCCS) && !defined(lint) 37 #ident "@(#)clnt_dg.c 1.23 94/04/22 SMI" 38 static char sccsid[] = "@(#)clnt_dg.c 1.19 89/03/16 Copyr 1988 Sun Micro"; 39 #endif 40 #include <sys/cdefs.h> 41 /* 42 * Implements a connectionless client side RPC. 43 */ 44 45 #include <sys/param.h> 46 #include <sys/systm.h> 47 #include <sys/kernel.h> 48 #include <sys/lock.h> 49 #include <sys/malloc.h> 50 #include <sys/mbuf.h> 51 #include <sys/mutex.h> 52 #include <sys/pcpu.h> 53 #include <sys/proc.h> 54 #include <sys/socket.h> 55 #include <sys/socketvar.h> 56 #include <sys/time.h> 57 #include <sys/uio.h> 58 59 #include <net/vnet.h> 60 61 #include <rpc/rpc.h> 62 #include <rpc/rpc_com.h> 63 64 65 #ifdef _FREEFALL_CONFIG 66 /* 67 * Disable RPC exponential back-off for FreeBSD.org systems. 68 */ 69 #define RPC_MAX_BACKOFF 1 /* second */ 70 #else 71 #define RPC_MAX_BACKOFF 30 /* seconds */ 72 #endif 73 74 static bool_t time_not_ok(struct timeval *); 75 static enum clnt_stat clnt_dg_call(CLIENT *, struct rpc_callextra *, 76 rpcproc_t, struct mbuf *, struct mbuf **, struct timeval); 77 static void clnt_dg_geterr(CLIENT *, struct rpc_err *); 78 static bool_t clnt_dg_freeres(CLIENT *, xdrproc_t, void *); 79 static void clnt_dg_abort(CLIENT *); 80 static bool_t clnt_dg_control(CLIENT *, u_int, void *); 81 static void clnt_dg_close(CLIENT *); 82 static void clnt_dg_destroy(CLIENT *); 83 static int clnt_dg_soupcall(struct socket *so, void *arg, int waitflag); 84 85 static const struct clnt_ops clnt_dg_ops = { 86 .cl_call = clnt_dg_call, 87 .cl_abort = clnt_dg_abort, 88 .cl_geterr = clnt_dg_geterr, 89 .cl_freeres = clnt_dg_freeres, 90 .cl_close = clnt_dg_close, 91 .cl_destroy = clnt_dg_destroy, 92 .cl_control = clnt_dg_control 93 }; 94 95 static volatile uint32_t rpc_xid = 0; 96 97 /* 98 * A pending RPC request which awaits a reply. Requests which have 99 * received their reply will have cr_xid set to zero and cr_mrep to 100 * the mbuf chain of the reply. 101 */ 102 struct cu_request { 103 TAILQ_ENTRY(cu_request) cr_link; 104 CLIENT *cr_client; /* owner */ 105 uint32_t cr_xid; /* XID of request */ 106 struct mbuf *cr_mrep; /* reply received by upcall */ 107 int cr_error; /* any error from upcall */ 108 char cr_verf[MAX_AUTH_BYTES]; /* reply verf */ 109 }; 110 111 TAILQ_HEAD(cu_request_list, cu_request); 112 113 #define MCALL_MSG_SIZE 24 114 115 /* 116 * This structure is pointed to by the socket buffer's sb_upcallarg 117 * member. It is separate from the client private data to facilitate 118 * multiple clients sharing the same socket. The cs_lock mutex is used 119 * to protect all fields of this structure, the socket's receive 120 * buffer SOCKBUF_LOCK is used to ensure that exactly one of these 121 * structures is installed on the socket. 122 */ 123 struct cu_socket { 124 struct mtx cs_lock; 125 int cs_refs; /* Count of clients */ 126 struct cu_request_list cs_pending; /* Requests awaiting replies */ 127 int cs_upcallrefs; /* Refcnt of upcalls in prog.*/ 128 }; 129 130 static void clnt_dg_upcallsdone(struct socket *, struct cu_socket *); 131 132 /* 133 * Private data kept per client handle 134 */ 135 struct cu_data { 136 int cu_threads; /* # threads in clnt_vc_call */ 137 bool_t cu_closing; /* TRUE if we are closing */ 138 bool_t cu_closed; /* TRUE if we are closed */ 139 struct socket *cu_socket; /* connection socket */ 140 bool_t cu_closeit; /* opened by library */ 141 struct sockaddr_storage cu_raddr; /* remote address */ 142 int cu_rlen; 143 struct timeval cu_wait; /* retransmit interval */ 144 struct timeval cu_total; /* total time for the call */ 145 struct rpc_err cu_error; 146 uint32_t cu_xid; 147 char cu_mcallc[MCALL_MSG_SIZE]; /* marshalled callmsg */ 148 size_t cu_mcalllen; 149 size_t cu_sendsz; /* send size */ 150 size_t cu_recvsz; /* recv size */ 151 int cu_async; 152 int cu_connect; /* Use connect(). */ 153 int cu_connected; /* Have done connect(). */ 154 const char *cu_waitchan; 155 int cu_waitflag; 156 int cu_cwnd; /* congestion window */ 157 int cu_sent; /* number of in-flight RPCs */ 158 bool_t cu_cwnd_wait; 159 }; 160 161 #define CWNDSCALE 256 162 #define MAXCWND (32 * CWNDSCALE) 163 164 /* 165 * Connection less client creation returns with client handle parameters. 166 * Default options are set, which the user can change using clnt_control(). 167 * fd should be open and bound. 168 * NB: The rpch->cl_auth is initialized to null authentication. 169 * Caller may wish to set this something more useful. 170 * 171 * sendsz and recvsz are the maximum allowable packet sizes that can be 172 * sent and received. Normally they are the same, but they can be 173 * changed to improve the program efficiency and buffer allocation. 174 * If they are 0, use the transport default. 175 * 176 * If svcaddr is NULL, returns NULL. 177 */ 178 CLIENT * 179 clnt_dg_create( 180 struct socket *so, 181 struct sockaddr *svcaddr, /* servers address */ 182 rpcprog_t program, /* program number */ 183 rpcvers_t version, /* version number */ 184 size_t sendsz, /* buffer recv size */ 185 size_t recvsz) /* buffer send size */ 186 { 187 CLIENT *cl = NULL; /* client handle */ 188 struct cu_data *cu = NULL; /* private data */ 189 struct cu_socket *cs = NULL; 190 struct sockbuf *sb; 191 struct timeval now; 192 struct rpc_msg call_msg; 193 struct __rpc_sockinfo si; 194 XDR xdrs; 195 int error; 196 uint32_t newxid; 197 198 if (svcaddr == NULL) { 199 rpc_createerr.cf_stat = RPC_UNKNOWNADDR; 200 return (NULL); 201 } 202 203 if (!__rpc_socket2sockinfo(so, &si)) { 204 rpc_createerr.cf_stat = RPC_TLIERROR; 205 rpc_createerr.cf_error.re_errno = 0; 206 return (NULL); 207 } 208 209 /* 210 * Find the receive and the send size 211 */ 212 sendsz = __rpc_get_t_size(si.si_af, si.si_proto, (int)sendsz); 213 recvsz = __rpc_get_t_size(si.si_af, si.si_proto, (int)recvsz); 214 if ((sendsz == 0) || (recvsz == 0)) { 215 rpc_createerr.cf_stat = RPC_TLIERROR; /* XXX */ 216 rpc_createerr.cf_error.re_errno = 0; 217 return (NULL); 218 } 219 220 cl = mem_alloc(sizeof (CLIENT)); 221 222 /* 223 * Should be multiple of 4 for XDR. 224 */ 225 sendsz = rounddown(sendsz + 3, 4); 226 recvsz = rounddown(recvsz + 3, 4); 227 cu = mem_alloc(sizeof (*cu)); 228 cu->cu_threads = 0; 229 cu->cu_closing = FALSE; 230 cu->cu_closed = FALSE; 231 (void) memcpy(&cu->cu_raddr, svcaddr, (size_t)svcaddr->sa_len); 232 cu->cu_rlen = svcaddr->sa_len; 233 /* Other values can also be set through clnt_control() */ 234 cu->cu_wait.tv_sec = 3; /* heuristically chosen */ 235 cu->cu_wait.tv_usec = 0; 236 cu->cu_total.tv_sec = -1; 237 cu->cu_total.tv_usec = -1; 238 cu->cu_sendsz = sendsz; 239 cu->cu_recvsz = recvsz; 240 cu->cu_async = FALSE; 241 cu->cu_connect = FALSE; 242 cu->cu_connected = FALSE; 243 cu->cu_waitchan = "rpcrecv"; 244 cu->cu_waitflag = 0; 245 cu->cu_cwnd = MAXCWND / 2; 246 cu->cu_sent = 0; 247 cu->cu_cwnd_wait = FALSE; 248 (void) getmicrotime(&now); 249 /* Clip at 28bits so that it will not wrap around. */ 250 newxid = __RPC_GETXID(&now) & 0xfffffff; 251 atomic_cmpset_32(&rpc_xid, 0, newxid); 252 call_msg.rm_xid = atomic_fetchadd_32(&rpc_xid, 1); 253 call_msg.rm_call.cb_prog = program; 254 call_msg.rm_call.cb_vers = version; 255 xdrmem_create(&xdrs, cu->cu_mcallc, MCALL_MSG_SIZE, XDR_ENCODE); 256 if (! xdr_callhdr(&xdrs, &call_msg)) { 257 rpc_createerr.cf_stat = RPC_CANTENCODEARGS; /* XXX */ 258 rpc_createerr.cf_error.re_errno = 0; 259 goto err2; 260 } 261 cu->cu_mcalllen = XDR_GETPOS(&xdrs); 262 263 /* 264 * By default, closeit is always FALSE. It is users responsibility 265 * to do a close on it, else the user may use clnt_control 266 * to let clnt_destroy do it for him/her. 267 */ 268 cu->cu_closeit = FALSE; 269 cu->cu_socket = so; 270 error = soreserve(so, (u_long)sendsz, (u_long)recvsz); 271 if (error != 0) { 272 rpc_createerr.cf_stat = RPC_FAILED; 273 rpc_createerr.cf_error.re_errno = error; 274 goto err2; 275 } 276 277 sb = &so->so_rcv; 278 SOCKBUF_LOCK(&so->so_rcv); 279 recheck_socket: 280 if (sb->sb_upcall) { 281 if (sb->sb_upcall != clnt_dg_soupcall) { 282 SOCKBUF_UNLOCK(&so->so_rcv); 283 printf("clnt_dg_create(): socket already has an incompatible upcall\n"); 284 goto err2; 285 } 286 cs = (struct cu_socket *) sb->sb_upcallarg; 287 mtx_lock(&cs->cs_lock); 288 cs->cs_refs++; 289 mtx_unlock(&cs->cs_lock); 290 } else { 291 /* 292 * We are the first on this socket - allocate the 293 * structure and install it in the socket. 294 */ 295 SOCKBUF_UNLOCK(&so->so_rcv); 296 cs = mem_alloc(sizeof(*cs)); 297 SOCKBUF_LOCK(&so->so_rcv); 298 if (sb->sb_upcall) { 299 /* 300 * We have lost a race with some other client. 301 */ 302 mem_free(cs, sizeof(*cs)); 303 goto recheck_socket; 304 } 305 mtx_init(&cs->cs_lock, "cs->cs_lock", NULL, MTX_DEF); 306 cs->cs_refs = 1; 307 cs->cs_upcallrefs = 0; 308 TAILQ_INIT(&cs->cs_pending); 309 soupcall_set(so, SO_RCV, clnt_dg_soupcall, cs); 310 } 311 SOCKBUF_UNLOCK(&so->so_rcv); 312 313 cl->cl_refs = 1; 314 cl->cl_ops = &clnt_dg_ops; 315 cl->cl_private = (caddr_t)(void *)cu; 316 cl->cl_auth = authnone_create(); 317 cl->cl_tp = NULL; 318 cl->cl_netid = NULL; 319 return (cl); 320 err2: 321 mem_free(cl, sizeof (CLIENT)); 322 mem_free(cu, sizeof (*cu)); 323 324 return (NULL); 325 } 326 327 static enum clnt_stat 328 clnt_dg_call( 329 CLIENT *cl, /* client handle */ 330 struct rpc_callextra *ext, /* call metadata */ 331 rpcproc_t proc, /* procedure number */ 332 struct mbuf *args, /* pointer to args */ 333 struct mbuf **resultsp, /* pointer to results */ 334 struct timeval utimeout) /* seconds to wait before giving up */ 335 { 336 struct cu_data *cu = (struct cu_data *)cl->cl_private; 337 struct cu_socket *cs; 338 struct rpc_timers *rt; 339 AUTH *auth; 340 struct rpc_err *errp; 341 enum clnt_stat stat; 342 XDR xdrs; 343 struct rpc_msg reply_msg; 344 bool_t ok; 345 int retrans; /* number of re-transmits so far */ 346 int nrefreshes = 2; /* number of times to refresh cred */ 347 struct timeval *tvp; 348 int timeout; 349 int retransmit_time; 350 int next_sendtime, starttime, rtt, time_waited, tv = 0; 351 struct sockaddr *sa; 352 uint32_t xid = 0; 353 struct mbuf *mreq = NULL, *results; 354 struct cu_request *cr; 355 int error; 356 357 cs = cu->cu_socket->so_rcv.sb_upcallarg; 358 cr = malloc(sizeof(struct cu_request), M_RPC, M_WAITOK); 359 360 mtx_lock(&cs->cs_lock); 361 362 if (cu->cu_closing || cu->cu_closed) { 363 mtx_unlock(&cs->cs_lock); 364 free(cr, M_RPC); 365 return (RPC_CANTSEND); 366 } 367 cu->cu_threads++; 368 369 if (ext) { 370 auth = ext->rc_auth; 371 errp = &ext->rc_err; 372 } else { 373 auth = cl->cl_auth; 374 errp = &cu->cu_error; 375 } 376 377 cr->cr_client = cl; 378 cr->cr_mrep = NULL; 379 cr->cr_error = 0; 380 381 if (cu->cu_total.tv_usec == -1) { 382 tvp = &utimeout; /* use supplied timeout */ 383 } else { 384 tvp = &cu->cu_total; /* use default timeout */ 385 } 386 if (tvp->tv_sec || tvp->tv_usec) 387 timeout = tvtohz(tvp); 388 else 389 timeout = 0; 390 391 if (cu->cu_connect && !cu->cu_connected) { 392 mtx_unlock(&cs->cs_lock); 393 error = soconnect(cu->cu_socket, 394 (struct sockaddr *)&cu->cu_raddr, curthread); 395 mtx_lock(&cs->cs_lock); 396 if (error) { 397 errp->re_errno = error; 398 errp->re_status = stat = RPC_CANTSEND; 399 goto out; 400 } 401 cu->cu_connected = 1; 402 } 403 if (cu->cu_connected) 404 sa = NULL; 405 else 406 sa = (struct sockaddr *)&cu->cu_raddr; 407 time_waited = 0; 408 retrans = 0; 409 if (ext && ext->rc_timers) { 410 rt = ext->rc_timers; 411 if (!rt->rt_rtxcur) 412 rt->rt_rtxcur = tvtohz(&cu->cu_wait); 413 retransmit_time = next_sendtime = rt->rt_rtxcur; 414 } else { 415 rt = NULL; 416 retransmit_time = next_sendtime = tvtohz(&cu->cu_wait); 417 } 418 419 starttime = ticks; 420 421 call_again: 422 mtx_assert(&cs->cs_lock, MA_OWNED); 423 424 xid = atomic_fetchadd_32(&rpc_xid, 1); 425 426 send_again: 427 mtx_unlock(&cs->cs_lock); 428 429 mreq = m_gethdr(M_WAITOK, MT_DATA); 430 KASSERT(cu->cu_mcalllen <= MHLEN, ("RPC header too big")); 431 bcopy(cu->cu_mcallc, mreq->m_data, cu->cu_mcalllen); 432 mreq->m_len = cu->cu_mcalllen; 433 434 /* 435 * The XID is the first thing in the request. 436 */ 437 *mtod(mreq, uint32_t *) = htonl(xid); 438 439 xdrmbuf_create(&xdrs, mreq, XDR_ENCODE); 440 441 if (cu->cu_async == TRUE && args == NULL) 442 goto get_reply; 443 444 if ((! XDR_PUTINT32(&xdrs, &proc)) || 445 (! AUTH_MARSHALL(auth, xid, &xdrs, 446 m_copym(args, 0, M_COPYALL, M_WAITOK)))) { 447 errp->re_status = stat = RPC_CANTENCODEARGS; 448 mtx_lock(&cs->cs_lock); 449 goto out; 450 } 451 mreq->m_pkthdr.len = m_length(mreq, NULL); 452 453 cr->cr_xid = xid; 454 mtx_lock(&cs->cs_lock); 455 456 /* 457 * Try to get a place in the congestion window. 458 */ 459 while (cu->cu_sent >= cu->cu_cwnd) { 460 cu->cu_cwnd_wait = TRUE; 461 error = msleep(&cu->cu_cwnd_wait, &cs->cs_lock, 462 cu->cu_waitflag, "rpccwnd", 0); 463 if (error) { 464 errp->re_errno = error; 465 if (error == EINTR || error == ERESTART) 466 errp->re_status = stat = RPC_INTR; 467 else 468 errp->re_status = stat = RPC_CANTSEND; 469 goto out; 470 } 471 } 472 cu->cu_sent += CWNDSCALE; 473 474 TAILQ_INSERT_TAIL(&cs->cs_pending, cr, cr_link); 475 mtx_unlock(&cs->cs_lock); 476 477 /* 478 * sosend consumes mreq. 479 */ 480 error = sosend(cu->cu_socket, sa, NULL, mreq, NULL, 0, curthread); 481 mreq = NULL; 482 483 /* 484 * sub-optimal code appears here because we have 485 * some clock time to spare while the packets are in flight. 486 * (We assume that this is actually only executed once.) 487 */ 488 reply_msg.acpted_rply.ar_verf.oa_flavor = AUTH_NULL; 489 reply_msg.acpted_rply.ar_verf.oa_base = cr->cr_verf; 490 reply_msg.acpted_rply.ar_verf.oa_length = 0; 491 reply_msg.acpted_rply.ar_results.where = NULL; 492 reply_msg.acpted_rply.ar_results.proc = (xdrproc_t)xdr_void; 493 494 mtx_lock(&cs->cs_lock); 495 if (error) { 496 TAILQ_REMOVE(&cs->cs_pending, cr, cr_link); 497 errp->re_errno = error; 498 errp->re_status = stat = RPC_CANTSEND; 499 cu->cu_sent -= CWNDSCALE; 500 if (cu->cu_cwnd_wait) { 501 cu->cu_cwnd_wait = FALSE; 502 wakeup(&cu->cu_cwnd_wait); 503 } 504 goto out; 505 } 506 507 /* 508 * Check to see if we got an upcall while waiting for the 509 * lock. 510 */ 511 if (cr->cr_error) { 512 TAILQ_REMOVE(&cs->cs_pending, cr, cr_link); 513 errp->re_errno = cr->cr_error; 514 errp->re_status = stat = RPC_CANTRECV; 515 cu->cu_sent -= CWNDSCALE; 516 if (cu->cu_cwnd_wait) { 517 cu->cu_cwnd_wait = FALSE; 518 wakeup(&cu->cu_cwnd_wait); 519 } 520 goto out; 521 } 522 if (cr->cr_mrep) { 523 TAILQ_REMOVE(&cs->cs_pending, cr, cr_link); 524 cu->cu_sent -= CWNDSCALE; 525 if (cu->cu_cwnd_wait) { 526 cu->cu_cwnd_wait = FALSE; 527 wakeup(&cu->cu_cwnd_wait); 528 } 529 goto got_reply; 530 } 531 532 /* 533 * Hack to provide rpc-based message passing 534 */ 535 if (timeout == 0) { 536 TAILQ_REMOVE(&cs->cs_pending, cr, cr_link); 537 errp->re_status = stat = RPC_TIMEDOUT; 538 cu->cu_sent -= CWNDSCALE; 539 if (cu->cu_cwnd_wait) { 540 cu->cu_cwnd_wait = FALSE; 541 wakeup(&cu->cu_cwnd_wait); 542 } 543 goto out; 544 } 545 546 get_reply: 547 for (;;) { 548 /* Decide how long to wait. */ 549 if (next_sendtime < timeout) 550 tv = next_sendtime; 551 else 552 tv = timeout; 553 tv -= time_waited; 554 555 if (tv > 0) { 556 if (cu->cu_closing || cu->cu_closed) { 557 error = 0; 558 cr->cr_error = ESHUTDOWN; 559 } else { 560 error = msleep(cr, &cs->cs_lock, 561 cu->cu_waitflag, cu->cu_waitchan, tv); 562 } 563 } else { 564 error = EWOULDBLOCK; 565 } 566 567 TAILQ_REMOVE(&cs->cs_pending, cr, cr_link); 568 cu->cu_sent -= CWNDSCALE; 569 if (cu->cu_cwnd_wait) { 570 cu->cu_cwnd_wait = FALSE; 571 wakeup(&cu->cu_cwnd_wait); 572 } 573 574 if (!error) { 575 /* 576 * We were woken up by the upcall. If the 577 * upcall had a receive error, report that, 578 * otherwise we have a reply. 579 */ 580 if (cr->cr_error) { 581 errp->re_errno = cr->cr_error; 582 errp->re_status = stat = RPC_CANTRECV; 583 goto out; 584 } 585 586 cu->cu_cwnd += (CWNDSCALE * CWNDSCALE 587 + cu->cu_cwnd / 2) / cu->cu_cwnd; 588 if (cu->cu_cwnd > MAXCWND) 589 cu->cu_cwnd = MAXCWND; 590 591 if (rt) { 592 /* 593 * Add one to the time since a tick 594 * count of N means that the actual 595 * time taken was somewhere between N 596 * and N+1. 597 */ 598 rtt = ticks - starttime + 1; 599 600 /* 601 * Update our estimate of the round 602 * trip time using roughly the 603 * algorithm described in RFC 604 * 2988. Given an RTT sample R: 605 * 606 * RTTVAR = (1-beta) * RTTVAR + beta * |SRTT-R| 607 * SRTT = (1-alpha) * SRTT + alpha * R 608 * 609 * where alpha = 0.125 and beta = 0.25. 610 * 611 * The initial retransmit timeout is 612 * SRTT + 4*RTTVAR and doubles on each 613 * retransmision. 614 */ 615 if (rt->rt_srtt == 0) { 616 rt->rt_srtt = rtt; 617 rt->rt_deviate = rtt / 2; 618 } else { 619 int32_t error = rtt - rt->rt_srtt; 620 rt->rt_srtt += error / 8; 621 error = abs(error) - rt->rt_deviate; 622 rt->rt_deviate += error / 4; 623 } 624 rt->rt_rtxcur = rt->rt_srtt + 4*rt->rt_deviate; 625 } 626 627 break; 628 } 629 630 /* 631 * The sleep returned an error so our request is still 632 * on the list. If we got EWOULDBLOCK, we may want to 633 * re-send the request. 634 */ 635 if (error != EWOULDBLOCK) { 636 errp->re_errno = error; 637 if (error == EINTR || error == ERESTART) 638 errp->re_status = stat = RPC_INTR; 639 else 640 errp->re_status = stat = RPC_CANTRECV; 641 goto out; 642 } 643 644 time_waited = ticks - starttime; 645 646 /* Check for timeout. */ 647 if (time_waited > timeout) { 648 errp->re_errno = EWOULDBLOCK; 649 errp->re_status = stat = RPC_TIMEDOUT; 650 goto out; 651 } 652 653 /* Retransmit if necessary. */ 654 if (time_waited >= next_sendtime) { 655 cu->cu_cwnd /= 2; 656 if (cu->cu_cwnd < CWNDSCALE) 657 cu->cu_cwnd = CWNDSCALE; 658 if (ext && ext->rc_feedback) { 659 mtx_unlock(&cs->cs_lock); 660 if (retrans == 0) 661 ext->rc_feedback(FEEDBACK_REXMIT1, 662 proc, ext->rc_feedback_arg); 663 else 664 ext->rc_feedback(FEEDBACK_REXMIT2, 665 proc, ext->rc_feedback_arg); 666 mtx_lock(&cs->cs_lock); 667 } 668 if (cu->cu_closing || cu->cu_closed) { 669 errp->re_errno = ESHUTDOWN; 670 errp->re_status = stat = RPC_CANTRECV; 671 goto out; 672 } 673 retrans++; 674 /* update retransmit_time */ 675 if (retransmit_time < RPC_MAX_BACKOFF * hz) 676 retransmit_time = 2 * retransmit_time; 677 next_sendtime += retransmit_time; 678 goto send_again; 679 } 680 cu->cu_sent += CWNDSCALE; 681 TAILQ_INSERT_TAIL(&cs->cs_pending, cr, cr_link); 682 } 683 684 got_reply: 685 /* 686 * Now decode and validate the response. We need to drop the 687 * lock since xdr_replymsg may end up sleeping in malloc. 688 */ 689 mtx_unlock(&cs->cs_lock); 690 691 if (ext && ext->rc_feedback) 692 ext->rc_feedback(FEEDBACK_OK, proc, ext->rc_feedback_arg); 693 694 xdrmbuf_create(&xdrs, cr->cr_mrep, XDR_DECODE); 695 ok = xdr_replymsg(&xdrs, &reply_msg); 696 cr->cr_mrep = NULL; 697 698 if (ok) { 699 if ((reply_msg.rm_reply.rp_stat == MSG_ACCEPTED) && 700 (reply_msg.acpted_rply.ar_stat == SUCCESS)) 701 errp->re_status = stat = RPC_SUCCESS; 702 else 703 stat = _seterr_reply(&reply_msg, &(cu->cu_error)); 704 705 if (errp->re_status == RPC_SUCCESS) { 706 results = xdrmbuf_getall(&xdrs); 707 if (! AUTH_VALIDATE(auth, xid, 708 &reply_msg.acpted_rply.ar_verf, 709 &results)) { 710 errp->re_status = stat = RPC_AUTHERROR; 711 errp->re_why = AUTH_INVALIDRESP; 712 if (retrans && 713 auth->ah_cred.oa_flavor == RPCSEC_GSS) { 714 /* 715 * If we retransmitted, its 716 * possible that we will 717 * receive a reply for one of 718 * the earlier transmissions 719 * (which will use an older 720 * RPCSEC_GSS sequence 721 * number). In this case, just 722 * go back and listen for a 723 * new reply. We could keep a 724 * record of all the seq 725 * numbers we have transmitted 726 * so far so that we could 727 * accept a reply for any of 728 * them here. 729 */ 730 XDR_DESTROY(&xdrs); 731 mtx_lock(&cs->cs_lock); 732 cu->cu_sent += CWNDSCALE; 733 TAILQ_INSERT_TAIL(&cs->cs_pending, 734 cr, cr_link); 735 cr->cr_mrep = NULL; 736 goto get_reply; 737 } 738 } else { 739 *resultsp = results; 740 } 741 } /* end successful completion */ 742 /* 743 * If unsuccessful AND error is an authentication error 744 * then refresh credentials and try again, else break 745 */ 746 else if (stat == RPC_AUTHERROR) 747 /* maybe our credentials need to be refreshed ... */ 748 if (nrefreshes > 0 && 749 AUTH_REFRESH(auth, &reply_msg)) { 750 nrefreshes--; 751 XDR_DESTROY(&xdrs); 752 mtx_lock(&cs->cs_lock); 753 goto call_again; 754 } 755 /* end of unsuccessful completion */ 756 } /* end of valid reply message */ 757 else { 758 errp->re_status = stat = RPC_CANTDECODERES; 759 760 } 761 XDR_DESTROY(&xdrs); 762 mtx_lock(&cs->cs_lock); 763 out: 764 mtx_assert(&cs->cs_lock, MA_OWNED); 765 766 if (mreq) 767 m_freem(mreq); 768 if (cr->cr_mrep) 769 m_freem(cr->cr_mrep); 770 771 cu->cu_threads--; 772 if (cu->cu_closing) 773 wakeup(cu); 774 775 mtx_unlock(&cs->cs_lock); 776 777 if (auth && stat != RPC_SUCCESS) 778 AUTH_VALIDATE(auth, xid, NULL, NULL); 779 780 free(cr, M_RPC); 781 782 return (stat); 783 } 784 785 static void 786 clnt_dg_geterr(CLIENT *cl, struct rpc_err *errp) 787 { 788 struct cu_data *cu = (struct cu_data *)cl->cl_private; 789 790 *errp = cu->cu_error; 791 } 792 793 static bool_t 794 clnt_dg_freeres(CLIENT *cl, xdrproc_t xdr_res, void *res_ptr) 795 { 796 XDR xdrs; 797 bool_t dummy; 798 799 xdrs.x_op = XDR_FREE; 800 dummy = (*xdr_res)(&xdrs, res_ptr); 801 802 return (dummy); 803 } 804 805 /*ARGSUSED*/ 806 static void 807 clnt_dg_abort(CLIENT *h) 808 { 809 } 810 811 static bool_t 812 clnt_dg_control(CLIENT *cl, u_int request, void *info) 813 { 814 struct cu_data *cu = (struct cu_data *)cl->cl_private; 815 struct cu_socket *cs; 816 struct sockaddr *addr; 817 818 cs = cu->cu_socket->so_rcv.sb_upcallarg; 819 mtx_lock(&cs->cs_lock); 820 821 switch (request) { 822 case CLSET_FD_CLOSE: 823 cu->cu_closeit = TRUE; 824 mtx_unlock(&cs->cs_lock); 825 return (TRUE); 826 case CLSET_FD_NCLOSE: 827 cu->cu_closeit = FALSE; 828 mtx_unlock(&cs->cs_lock); 829 return (TRUE); 830 } 831 832 /* for other requests which use info */ 833 if (info == NULL) { 834 mtx_unlock(&cs->cs_lock); 835 return (FALSE); 836 } 837 switch (request) { 838 case CLSET_TIMEOUT: 839 if (time_not_ok((struct timeval *)info)) { 840 mtx_unlock(&cs->cs_lock); 841 return (FALSE); 842 } 843 cu->cu_total = *(struct timeval *)info; 844 break; 845 case CLGET_TIMEOUT: 846 *(struct timeval *)info = cu->cu_total; 847 break; 848 case CLSET_RETRY_TIMEOUT: 849 if (time_not_ok((struct timeval *)info)) { 850 mtx_unlock(&cs->cs_lock); 851 return (FALSE); 852 } 853 cu->cu_wait = *(struct timeval *)info; 854 break; 855 case CLGET_RETRY_TIMEOUT: 856 *(struct timeval *)info = cu->cu_wait; 857 break; 858 case CLGET_SVC_ADDR: 859 /* 860 * Slightly different semantics to userland - we use 861 * sockaddr instead of netbuf. 862 */ 863 memcpy(info, &cu->cu_raddr, cu->cu_raddr.ss_len); 864 break; 865 case CLSET_SVC_ADDR: /* set to new address */ 866 addr = (struct sockaddr *)info; 867 (void) memcpy(&cu->cu_raddr, addr, addr->sa_len); 868 break; 869 case CLGET_XID: 870 *(uint32_t *)info = atomic_load_32(&rpc_xid); 871 break; 872 873 case CLSET_XID: 874 /* This will set the xid of the NEXT call */ 875 /* decrement by 1 as clnt_dg_call() increments once */ 876 atomic_store_32(&rpc_xid, *(uint32_t *)info - 1); 877 break; 878 879 case CLGET_VERS: 880 /* 881 * This RELIES on the information that, in the call body, 882 * the version number field is the fifth field from the 883 * beginning of the RPC header. MUST be changed if the 884 * call_struct is changed 885 */ 886 *(uint32_t *)info = 887 ntohl(*(uint32_t *)(void *)(cu->cu_mcallc + 888 4 * BYTES_PER_XDR_UNIT)); 889 break; 890 891 case CLSET_VERS: 892 *(uint32_t *)(void *)(cu->cu_mcallc + 4 * BYTES_PER_XDR_UNIT) 893 = htonl(*(uint32_t *)info); 894 break; 895 896 case CLGET_PROG: 897 /* 898 * This RELIES on the information that, in the call body, 899 * the program number field is the fourth field from the 900 * beginning of the RPC header. MUST be changed if the 901 * call_struct is changed 902 */ 903 *(uint32_t *)info = 904 ntohl(*(uint32_t *)(void *)(cu->cu_mcallc + 905 3 * BYTES_PER_XDR_UNIT)); 906 break; 907 908 case CLSET_PROG: 909 *(uint32_t *)(void *)(cu->cu_mcallc + 3 * BYTES_PER_XDR_UNIT) 910 = htonl(*(uint32_t *)info); 911 break; 912 case CLSET_ASYNC: 913 cu->cu_async = *(int *)info; 914 break; 915 case CLSET_CONNECT: 916 cu->cu_connect = *(int *)info; 917 break; 918 case CLSET_WAITCHAN: 919 cu->cu_waitchan = (const char *)info; 920 break; 921 case CLGET_WAITCHAN: 922 *(const char **) info = cu->cu_waitchan; 923 break; 924 case CLSET_INTERRUPTIBLE: 925 if (*(int *) info) 926 cu->cu_waitflag = PCATCH; 927 else 928 cu->cu_waitflag = 0; 929 break; 930 case CLGET_INTERRUPTIBLE: 931 if (cu->cu_waitflag) 932 *(int *) info = TRUE; 933 else 934 *(int *) info = FALSE; 935 break; 936 default: 937 mtx_unlock(&cs->cs_lock); 938 return (FALSE); 939 } 940 mtx_unlock(&cs->cs_lock); 941 return (TRUE); 942 } 943 944 static void 945 clnt_dg_close(CLIENT *cl) 946 { 947 struct cu_data *cu = (struct cu_data *)cl->cl_private; 948 struct cu_socket *cs; 949 struct cu_request *cr; 950 951 cs = cu->cu_socket->so_rcv.sb_upcallarg; 952 mtx_lock(&cs->cs_lock); 953 954 if (cu->cu_closed) { 955 mtx_unlock(&cs->cs_lock); 956 return; 957 } 958 959 if (cu->cu_closing) { 960 while (cu->cu_closing) 961 msleep(cu, &cs->cs_lock, 0, "rpcclose", 0); 962 KASSERT(cu->cu_closed, ("client should be closed")); 963 mtx_unlock(&cs->cs_lock); 964 return; 965 } 966 967 /* 968 * Abort any pending requests and wait until everyone 969 * has finished with clnt_vc_call. 970 */ 971 cu->cu_closing = TRUE; 972 TAILQ_FOREACH(cr, &cs->cs_pending, cr_link) { 973 if (cr->cr_client == cl) { 974 cr->cr_xid = 0; 975 cr->cr_error = ESHUTDOWN; 976 wakeup(cr); 977 } 978 } 979 980 while (cu->cu_threads) 981 msleep(cu, &cs->cs_lock, 0, "rpcclose", 0); 982 983 cu->cu_closing = FALSE; 984 cu->cu_closed = TRUE; 985 986 mtx_unlock(&cs->cs_lock); 987 wakeup(cu); 988 } 989 990 static void 991 clnt_dg_destroy(CLIENT *cl) 992 { 993 struct cu_data *cu = (struct cu_data *)cl->cl_private; 994 struct cu_socket *cs; 995 struct socket *so = NULL; 996 bool_t lastsocketref; 997 998 cs = cu->cu_socket->so_rcv.sb_upcallarg; 999 clnt_dg_close(cl); 1000 1001 SOCKBUF_LOCK(&cu->cu_socket->so_rcv); 1002 mtx_lock(&cs->cs_lock); 1003 1004 cs->cs_refs--; 1005 if (cs->cs_refs == 0) { 1006 mtx_unlock(&cs->cs_lock); 1007 soupcall_clear(cu->cu_socket, SO_RCV); 1008 clnt_dg_upcallsdone(cu->cu_socket, cs); 1009 SOCKBUF_UNLOCK(&cu->cu_socket->so_rcv); 1010 mtx_destroy(&cs->cs_lock); 1011 mem_free(cs, sizeof(*cs)); 1012 lastsocketref = TRUE; 1013 } else { 1014 mtx_unlock(&cs->cs_lock); 1015 SOCKBUF_UNLOCK(&cu->cu_socket->so_rcv); 1016 lastsocketref = FALSE; 1017 } 1018 1019 if (cu->cu_closeit && lastsocketref) { 1020 so = cu->cu_socket; 1021 cu->cu_socket = NULL; 1022 } 1023 1024 if (so) 1025 soclose(so); 1026 1027 if (cl->cl_netid && cl->cl_netid[0]) 1028 mem_free(cl->cl_netid, strlen(cl->cl_netid) +1); 1029 if (cl->cl_tp && cl->cl_tp[0]) 1030 mem_free(cl->cl_tp, strlen(cl->cl_tp) +1); 1031 mem_free(cu, sizeof (*cu)); 1032 mem_free(cl, sizeof (CLIENT)); 1033 } 1034 1035 /* 1036 * Make sure that the time is not garbage. -1 value is allowed. 1037 */ 1038 static bool_t 1039 time_not_ok(struct timeval *t) 1040 { 1041 return (t->tv_sec < -1 || t->tv_sec > 100000000 || 1042 t->tv_usec < -1 || t->tv_usec > 1000000); 1043 } 1044 1045 int 1046 clnt_dg_soupcall(struct socket *so, void *arg, int waitflag) 1047 { 1048 struct cu_socket *cs = (struct cu_socket *) arg; 1049 struct uio uio; 1050 struct mbuf *m; 1051 struct mbuf *control; 1052 struct cu_request *cr; 1053 int error, rcvflag, foundreq; 1054 uint32_t xid; 1055 1056 cs->cs_upcallrefs++; 1057 uio.uio_resid = 1000000000; 1058 uio.uio_td = curthread; 1059 do { 1060 SOCKBUF_UNLOCK(&so->so_rcv); 1061 m = NULL; 1062 control = NULL; 1063 rcvflag = MSG_DONTWAIT; 1064 error = soreceive(so, NULL, &uio, &m, &control, &rcvflag); 1065 if (control) 1066 m_freem(control); 1067 SOCKBUF_LOCK(&so->so_rcv); 1068 1069 if (error == EWOULDBLOCK) 1070 break; 1071 1072 /* 1073 * If there was an error, wake up all pending 1074 * requests. 1075 */ 1076 if (error) { 1077 mtx_lock(&cs->cs_lock); 1078 TAILQ_FOREACH(cr, &cs->cs_pending, cr_link) { 1079 cr->cr_xid = 0; 1080 cr->cr_error = error; 1081 wakeup(cr); 1082 } 1083 mtx_unlock(&cs->cs_lock); 1084 break; 1085 } 1086 1087 /* 1088 * The XID is in the first uint32_t of the reply. 1089 */ 1090 if (m->m_len < sizeof(xid) && m_length(m, NULL) < sizeof(xid)) { 1091 /* 1092 * Should never happen. 1093 */ 1094 m_freem(m); 1095 continue; 1096 } 1097 1098 m_copydata(m, 0, sizeof(xid), (char *)&xid); 1099 xid = ntohl(xid); 1100 1101 /* 1102 * Attempt to match this reply with a pending request. 1103 */ 1104 mtx_lock(&cs->cs_lock); 1105 foundreq = 0; 1106 TAILQ_FOREACH(cr, &cs->cs_pending, cr_link) { 1107 if (cr->cr_xid == xid) { 1108 /* 1109 * This one matches. We leave the 1110 * reply mbuf in cr->cr_mrep. Set the 1111 * XID to zero so that we will ignore 1112 * any duplicated replies that arrive 1113 * before clnt_dg_call removes it from 1114 * the queue. 1115 */ 1116 cr->cr_xid = 0; 1117 cr->cr_mrep = m; 1118 cr->cr_error = 0; 1119 foundreq = 1; 1120 wakeup(cr); 1121 break; 1122 } 1123 } 1124 mtx_unlock(&cs->cs_lock); 1125 1126 /* 1127 * If we didn't find the matching request, just drop 1128 * it - its probably a repeated reply. 1129 */ 1130 if (!foundreq) 1131 m_freem(m); 1132 } while (m); 1133 cs->cs_upcallrefs--; 1134 if (cs->cs_upcallrefs < 0) 1135 panic("rpcdg upcall refcnt"); 1136 if (cs->cs_upcallrefs == 0) 1137 wakeup(&cs->cs_upcallrefs); 1138 return (SU_OK); 1139 } 1140 1141 /* 1142 * Wait for all upcalls in progress to complete. 1143 */ 1144 static void 1145 clnt_dg_upcallsdone(struct socket *so, struct cu_socket *cs) 1146 { 1147 1148 SOCKBUF_LOCK_ASSERT(&so->so_rcv); 1149 1150 while (cs->cs_upcallrefs > 0) 1151 (void) msleep(&cs->cs_upcallrefs, SOCKBUF_MTX(&so->so_rcv), 0, 1152 "rpcdgup", 0); 1153 } 1154