clnt_vc.c revision 184877
1/* $NetBSD: clnt_vc.c,v 1.4 2000/07/14 08:40:42 fvdl Exp $ */ 2 3/* 4 * Sun RPC is a product of Sun Microsystems, Inc. and is provided for 5 * unrestricted use provided that this legend is included on all tape 6 * media and as a part of the software program in whole or part. Users 7 * may copy or modify Sun RPC without charge, but are not authorized 8 * to license or distribute it to anyone else except as part of a product or 9 * program developed by the user. 10 * 11 * SUN RPC IS PROVIDED AS IS WITH NO WARRANTIES OF ANY KIND INCLUDING THE 12 * WARRANTIES OF DESIGN, MERCHANTIBILITY AND FITNESS FOR A PARTICULAR 13 * PURPOSE, OR ARISING FROM A COURSE OF DEALING, USAGE OR TRADE PRACTICE. 14 * 15 * Sun RPC is provided with no support and without any obligation on the 16 * part of Sun Microsystems, Inc. to assist in its use, correction, 17 * modification or enhancement. 18 * 19 * SUN MICROSYSTEMS, INC. SHALL HAVE NO LIABILITY WITH RESPECT TO THE 20 * INFRINGEMENT OF COPYRIGHTS, TRADE SECRETS OR ANY PATENTS BY SUN RPC 21 * OR ANY PART THEREOF. 22 * 23 * In no event will Sun Microsystems, Inc. be liable for any lost revenue 24 * or profits or other special, indirect and consequential damages, even if 25 * Sun has been advised of the possibility of such damages. 26 * 27 * Sun Microsystems, Inc. 28 * 2550 Garcia Avenue 29 * Mountain View, California 94043 30 */ 31 32#if defined(LIBC_SCCS) && !defined(lint) 33static char *sccsid2 = "@(#)clnt_tcp.c 1.37 87/10/05 Copyr 1984 Sun Micro"; 34static char *sccsid = "@(#)clnt_tcp.c 2.2 88/08/01 4.0 RPCSRC"; 35static char sccsid3[] = "@(#)clnt_vc.c 1.19 89/03/16 Copyr 1988 Sun Micro"; 36#endif 37#include <sys/cdefs.h> 38__FBSDID("$FreeBSD: head/sys/rpc/clnt_vc.c 184877 2008-11-12 12:21:18Z dfr $"); 39 40/* 41 * clnt_tcp.c, Implements a TCP/IP based, client side RPC. 42 * 43 * Copyright (C) 1984, Sun Microsystems, Inc. 44 * 45 * TCP based RPC supports 'batched calls'. 46 * A sequence of calls may be batched-up in a send buffer. The rpc call 47 * return immediately to the client even though the call was not necessarily 48 * sent. The batching occurs if the results' xdr routine is NULL (0) AND 49 * the rpc timeout value is zero (see clnt.h, rpc). 50 * 51 * Clients should NOT casually batch calls that in fact return results; that is, 52 * the server side should be aware that a call is batched and not produce any 53 * return message. Batched calls that produce many result messages can 54 * deadlock (netlock) the client and the server.... 55 * 56 * Now go hang yourself. 57 */ 58 59#include <sys/param.h> 60#include <sys/systm.h> 61#include <sys/lock.h> 62#include <sys/malloc.h> 63#include <sys/mbuf.h> 64#include <sys/mutex.h> 65#include <sys/pcpu.h> 66#include <sys/proc.h> 67#include <sys/protosw.h> 68#include <sys/socket.h> 69#include <sys/socketvar.h> 70#include <sys/syslog.h> 71#include <sys/time.h> 72#include <sys/uio.h> 73#include <netinet/tcp.h> 74 75#include <rpc/rpc.h> 76#include <rpc/rpc_com.h> 77 78#define MCALL_MSG_SIZE 24 79 80struct cmessage { 81 struct cmsghdr cmsg; 82 struct cmsgcred cmcred; 83}; 84 85static enum clnt_stat clnt_vc_call(CLIENT *, struct rpc_callextra *, 86 rpcproc_t, struct mbuf *, struct mbuf **, struct timeval); 87static void clnt_vc_geterr(CLIENT *, struct rpc_err *); 88static bool_t clnt_vc_freeres(CLIENT *, xdrproc_t, void *); 89static void clnt_vc_abort(CLIENT *); 90static bool_t clnt_vc_control(CLIENT *, u_int, void *); 91static void clnt_vc_close(CLIENT *); 92static void clnt_vc_destroy(CLIENT *); 93static bool_t time_not_ok(struct timeval *); 94static void clnt_vc_soupcall(struct socket *so, void *arg, int waitflag); 95 96static struct clnt_ops clnt_vc_ops = { 97 .cl_call = clnt_vc_call, 98 .cl_abort = clnt_vc_abort, 99 .cl_geterr = clnt_vc_geterr, 100 .cl_freeres = clnt_vc_freeres, 101 .cl_close = clnt_vc_close, 102 .cl_destroy = clnt_vc_destroy, 103 .cl_control = clnt_vc_control 104}; 105 106/* 107 * A pending RPC request which awaits a reply. Requests which have 108 * received their reply will have cr_xid set to zero and cr_mrep to 109 * the mbuf chain of the reply. 110 */ 111struct ct_request { 112 TAILQ_ENTRY(ct_request) cr_link; 113 uint32_t cr_xid; /* XID of request */ 114 struct mbuf *cr_mrep; /* reply received by upcall */ 115 int cr_error; /* any error from upcall */ 116 char cr_verf[MAX_AUTH_BYTES]; /* reply verf */ 117}; 118 119TAILQ_HEAD(ct_request_list, ct_request); 120 121struct ct_data { 122 struct mtx ct_lock; 123 int ct_threads; /* number of threads in clnt_vc_call */ 124 bool_t ct_closing; /* TRUE if we are closing */ 125 bool_t ct_closed; /* TRUE if we are closed */ 126 struct socket *ct_socket; /* connection socket */ 127 bool_t ct_closeit; /* close it on destroy */ 128 struct timeval ct_wait; /* wait interval in milliseconds */ 129 struct sockaddr_storage ct_addr; /* remote addr */ 130 struct rpc_err ct_error; 131 uint32_t ct_xid; 132 char ct_mcallc[MCALL_MSG_SIZE]; /* marshalled callmsg */ 133 size_t ct_mpos; /* pos after marshal */ 134 const char *ct_waitchan; 135 int ct_waitflag; 136 struct mbuf *ct_record; /* current reply record */ 137 size_t ct_record_resid; /* how much left of reply to read */ 138 bool_t ct_record_eor; /* true if reading last fragment */ 139 struct ct_request_list ct_pending; 140}; 141 142static const char clnt_vc_errstr[] = "%s : %s"; 143static const char clnt_vc_str[] = "clnt_vc_create"; 144static const char clnt_read_vc_str[] = "read_vc"; 145static const char __no_mem_str[] = "out of memory"; 146 147/* 148 * Create a client handle for a connection. 149 * Default options are set, which the user can change using clnt_control()'s. 150 * The rpc/vc package does buffering similar to stdio, so the client 151 * must pick send and receive buffer sizes, 0 => use the default. 152 * NB: fd is copied into a private area. 153 * NB: The rpch->cl_auth is set null authentication. Caller may wish to 154 * set this something more useful. 155 * 156 * fd should be an open socket 157 */ 158CLIENT * 159clnt_vc_create( 160 struct socket *so, /* open file descriptor */ 161 struct sockaddr *raddr, /* servers address */ 162 const rpcprog_t prog, /* program number */ 163 const rpcvers_t vers, /* version number */ 164 size_t sendsz, /* buffer recv size */ 165 size_t recvsz) /* buffer send size */ 166{ 167 CLIENT *cl; /* client handle */ 168 struct ct_data *ct = NULL; /* client handle */ 169 struct timeval now; 170 struct rpc_msg call_msg; 171 static uint32_t disrupt; 172 struct __rpc_sockinfo si; 173 XDR xdrs; 174 int error, interrupted, one = 1; 175 struct sockopt sopt; 176 177 if (disrupt == 0) 178 disrupt = (uint32_t)(long)raddr; 179 180 cl = (CLIENT *)mem_alloc(sizeof (*cl)); 181 ct = (struct ct_data *)mem_alloc(sizeof (*ct)); 182 183 mtx_init(&ct->ct_lock, "ct->ct_lock", NULL, MTX_DEF); 184 ct->ct_threads = 0; 185 ct->ct_closing = FALSE; 186 ct->ct_closed = FALSE; 187 188 if ((so->so_state & (SS_ISCONNECTED|SS_ISCONFIRMING)) == 0) { 189 error = soconnect(so, raddr, curthread); 190 SOCK_LOCK(so); 191 interrupted = 0; 192 while ((so->so_state & SS_ISCONNECTING) 193 && so->so_error == 0) { 194 error = msleep(&so->so_timeo, SOCK_MTX(so), 195 PSOCK | PCATCH, "connec", 0); 196 if (error) { 197 if (error == EINTR || error == ERESTART) 198 interrupted = 1; 199 break; 200 } 201 } 202 if (error == 0) { 203 error = so->so_error; 204 so->so_error = 0; 205 } 206 SOCK_UNLOCK(so); 207 if (error) { 208 if (!interrupted) 209 so->so_state &= ~SS_ISCONNECTING; 210 rpc_createerr.cf_stat = RPC_SYSTEMERROR; 211 rpc_createerr.cf_error.re_errno = error; 212 goto err; 213 } 214 } 215 216 if (!__rpc_socket2sockinfo(so, &si)) 217 goto err; 218 219 if (so->so_proto->pr_flags & PR_CONNREQUIRED) { 220 bzero(&sopt, sizeof(sopt)); 221 sopt.sopt_dir = SOPT_SET; 222 sopt.sopt_level = SOL_SOCKET; 223 sopt.sopt_name = SO_KEEPALIVE; 224 sopt.sopt_val = &one; 225 sopt.sopt_valsize = sizeof(one); 226 sosetopt(so, &sopt); 227 } 228 229 if (so->so_proto->pr_protocol == IPPROTO_TCP) { 230 bzero(&sopt, sizeof(sopt)); 231 sopt.sopt_dir = SOPT_SET; 232 sopt.sopt_level = IPPROTO_TCP; 233 sopt.sopt_name = TCP_NODELAY; 234 sopt.sopt_val = &one; 235 sopt.sopt_valsize = sizeof(one); 236 sosetopt(so, &sopt); 237 } 238 239 ct->ct_closeit = FALSE; 240 241 /* 242 * Set up private data struct 243 */ 244 ct->ct_socket = so; 245 ct->ct_wait.tv_sec = -1; 246 ct->ct_wait.tv_usec = -1; 247 memcpy(&ct->ct_addr, raddr, raddr->sa_len); 248 249 /* 250 * Initialize call message 251 */ 252 getmicrotime(&now); 253 ct->ct_xid = ((uint32_t)++disrupt) ^ __RPC_GETXID(&now); 254 call_msg.rm_xid = ct->ct_xid; 255 call_msg.rm_direction = CALL; 256 call_msg.rm_call.cb_rpcvers = RPC_MSG_VERSION; 257 call_msg.rm_call.cb_prog = (uint32_t)prog; 258 call_msg.rm_call.cb_vers = (uint32_t)vers; 259 260 /* 261 * pre-serialize the static part of the call msg and stash it away 262 */ 263 xdrmem_create(&xdrs, ct->ct_mcallc, MCALL_MSG_SIZE, 264 XDR_ENCODE); 265 if (! xdr_callhdr(&xdrs, &call_msg)) { 266 if (ct->ct_closeit) { 267 soclose(ct->ct_socket); 268 } 269 goto err; 270 } 271 ct->ct_mpos = XDR_GETPOS(&xdrs); 272 XDR_DESTROY(&xdrs); 273 ct->ct_waitchan = "rpcrecv"; 274 ct->ct_waitflag = 0; 275 276 /* 277 * Create a client handle which uses xdrrec for serialization 278 * and authnone for authentication. 279 */ 280 cl->cl_refs = 1; 281 cl->cl_ops = &clnt_vc_ops; 282 cl->cl_private = ct; 283 cl->cl_auth = authnone_create(); 284 sendsz = __rpc_get_t_size(si.si_af, si.si_proto, (int)sendsz); 285 recvsz = __rpc_get_t_size(si.si_af, si.si_proto, (int)recvsz); 286 soreserve(ct->ct_socket, sendsz, recvsz); 287 288 SOCKBUF_LOCK(&ct->ct_socket->so_rcv); 289 ct->ct_socket->so_upcallarg = ct; 290 ct->ct_socket->so_upcall = clnt_vc_soupcall; 291 ct->ct_socket->so_rcv.sb_flags |= SB_UPCALL; 292 SOCKBUF_UNLOCK(&ct->ct_socket->so_rcv); 293 294 ct->ct_record = NULL; 295 ct->ct_record_resid = 0; 296 TAILQ_INIT(&ct->ct_pending); 297 return (cl); 298 299err: 300 if (cl) { 301 if (ct) { 302 mtx_destroy(&ct->ct_lock); 303 mem_free(ct, sizeof (struct ct_data)); 304 } 305 if (cl) 306 mem_free(cl, sizeof (CLIENT)); 307 } 308 return ((CLIENT *)NULL); 309} 310 311static enum clnt_stat 312clnt_vc_call( 313 CLIENT *cl, /* client handle */ 314 struct rpc_callextra *ext, /* call metadata */ 315 rpcproc_t proc, /* procedure number */ 316 struct mbuf *args, /* pointer to args */ 317 struct mbuf **resultsp, /* pointer to results */ 318 struct timeval utimeout) 319{ 320 struct ct_data *ct = (struct ct_data *) cl->cl_private; 321 AUTH *auth; 322 struct rpc_err *errp; 323 enum clnt_stat stat; 324 XDR xdrs; 325 struct rpc_msg reply_msg; 326 bool_t ok; 327 int nrefreshes = 2; /* number of times to refresh cred */ 328 struct timeval timeout; 329 uint32_t xid; 330 struct mbuf *mreq = NULL, *results; 331 struct ct_request *cr; 332 int error; 333 334 cr = malloc(sizeof(struct ct_request), M_RPC, M_WAITOK); 335 336 mtx_lock(&ct->ct_lock); 337 338 if (ct->ct_closing || ct->ct_closed) { 339 mtx_unlock(&ct->ct_lock); 340 free(cr, M_RPC); 341 return (RPC_CANTSEND); 342 } 343 ct->ct_threads++; 344 345 if (ext) { 346 auth = ext->rc_auth; 347 errp = &ext->rc_err; 348 } else { 349 auth = cl->cl_auth; 350 errp = &ct->ct_error; 351 } 352 353 cr->cr_mrep = NULL; 354 cr->cr_error = 0; 355 356 if (ct->ct_wait.tv_usec == -1) { 357 timeout = utimeout; /* use supplied timeout */ 358 } else { 359 timeout = ct->ct_wait; /* use default timeout */ 360 } 361 362call_again: 363 mtx_assert(&ct->ct_lock, MA_OWNED); 364 365 ct->ct_xid++; 366 xid = ct->ct_xid; 367 368 mtx_unlock(&ct->ct_lock); 369 370 /* 371 * Leave space to pre-pend the record mark. 372 */ 373 MGETHDR(mreq, M_WAIT, MT_DATA); 374 mreq->m_data += sizeof(uint32_t); 375 KASSERT(ct->ct_mpos + sizeof(uint32_t) <= MHLEN, 376 ("RPC header too big")); 377 bcopy(ct->ct_mcallc, mreq->m_data, ct->ct_mpos); 378 mreq->m_len = ct->ct_mpos; 379 380 /* 381 * The XID is the first thing in the request. 382 */ 383 *mtod(mreq, uint32_t *) = htonl(xid); 384 385 xdrmbuf_create(&xdrs, mreq, XDR_ENCODE); 386 387 errp->re_status = stat = RPC_SUCCESS; 388 389 if ((! XDR_PUTINT32(&xdrs, &proc)) || 390 (! AUTH_MARSHALL(auth, xid, &xdrs, 391 m_copym(args, 0, M_COPYALL, M_WAITOK)))) { 392 errp->re_status = stat = RPC_CANTENCODEARGS; 393 mtx_lock(&ct->ct_lock); 394 goto out; 395 } 396 mreq->m_pkthdr.len = m_length(mreq, NULL); 397 398 /* 399 * Prepend a record marker containing the packet length. 400 */ 401 M_PREPEND(mreq, sizeof(uint32_t), M_WAIT); 402 *mtod(mreq, uint32_t *) = 403 htonl(0x80000000 | (mreq->m_pkthdr.len - sizeof(uint32_t))); 404 405 cr->cr_xid = xid; 406 mtx_lock(&ct->ct_lock); 407 TAILQ_INSERT_TAIL(&ct->ct_pending, cr, cr_link); 408 mtx_unlock(&ct->ct_lock); 409 410 /* 411 * sosend consumes mreq. 412 */ 413 error = sosend(ct->ct_socket, NULL, NULL, mreq, NULL, 0, curthread); 414 mreq = NULL; 415 if (error == EMSGSIZE) { 416 SOCKBUF_LOCK(&ct->ct_socket->so_snd); 417 sbwait(&ct->ct_socket->so_snd); 418 SOCKBUF_UNLOCK(&ct->ct_socket->so_snd); 419 AUTH_VALIDATE(auth, xid, NULL, NULL); 420 mtx_lock(&ct->ct_lock); 421 TAILQ_REMOVE(&ct->ct_pending, cr, cr_link); 422 goto call_again; 423 } 424 425 reply_msg.acpted_rply.ar_verf.oa_flavor = AUTH_NULL; 426 reply_msg.acpted_rply.ar_verf.oa_base = cr->cr_verf; 427 reply_msg.acpted_rply.ar_verf.oa_length = 0; 428 reply_msg.acpted_rply.ar_results.where = NULL; 429 reply_msg.acpted_rply.ar_results.proc = (xdrproc_t)xdr_void; 430 431 mtx_lock(&ct->ct_lock); 432 if (error) { 433 TAILQ_REMOVE(&ct->ct_pending, cr, cr_link); 434 errp->re_errno = error; 435 errp->re_status = stat = RPC_CANTSEND; 436 goto out; 437 } 438 439 /* 440 * Check to see if we got an upcall while waiting for the 441 * lock. In both these cases, the request has been removed 442 * from ct->ct_pending. 443 */ 444 if (cr->cr_error) { 445 TAILQ_REMOVE(&ct->ct_pending, cr, cr_link); 446 errp->re_errno = cr->cr_error; 447 errp->re_status = stat = RPC_CANTRECV; 448 goto out; 449 } 450 if (cr->cr_mrep) { 451 TAILQ_REMOVE(&ct->ct_pending, cr, cr_link); 452 goto got_reply; 453 } 454 455 /* 456 * Hack to provide rpc-based message passing 457 */ 458 if (timeout.tv_sec == 0 && timeout.tv_usec == 0) { 459 TAILQ_REMOVE(&ct->ct_pending, cr, cr_link); 460 errp->re_status = stat = RPC_TIMEDOUT; 461 goto out; 462 } 463 464 error = msleep(cr, &ct->ct_lock, ct->ct_waitflag, ct->ct_waitchan, 465 tvtohz(&timeout)); 466 467 TAILQ_REMOVE(&ct->ct_pending, cr, cr_link); 468 469 if (error) { 470 /* 471 * The sleep returned an error so our request is still 472 * on the list. Turn the error code into an 473 * appropriate client status. 474 */ 475 errp->re_errno = error; 476 switch (error) { 477 case EINTR: 478 stat = RPC_INTR; 479 break; 480 case EWOULDBLOCK: 481 stat = RPC_TIMEDOUT; 482 break; 483 default: 484 stat = RPC_CANTRECV; 485 } 486 errp->re_status = stat; 487 goto out; 488 } else { 489 /* 490 * We were woken up by the upcall. If the 491 * upcall had a receive error, report that, 492 * otherwise we have a reply. 493 */ 494 if (cr->cr_error) { 495 errp->re_errno = cr->cr_error; 496 errp->re_status = stat = RPC_CANTRECV; 497 goto out; 498 } 499 } 500 501got_reply: 502 /* 503 * Now decode and validate the response. We need to drop the 504 * lock since xdr_replymsg may end up sleeping in malloc. 505 */ 506 mtx_unlock(&ct->ct_lock); 507 508 if (ext && ext->rc_feedback) 509 ext->rc_feedback(FEEDBACK_OK, proc, ext->rc_feedback_arg); 510 511 xdrmbuf_create(&xdrs, cr->cr_mrep, XDR_DECODE); 512 ok = xdr_replymsg(&xdrs, &reply_msg); 513 cr->cr_mrep = NULL; 514 515 if (ok) { 516 if ((reply_msg.rm_reply.rp_stat == MSG_ACCEPTED) && 517 (reply_msg.acpted_rply.ar_stat == SUCCESS)) 518 errp->re_status = stat = RPC_SUCCESS; 519 else 520 stat = _seterr_reply(&reply_msg, errp); 521 522 if (stat == RPC_SUCCESS) { 523 results = xdrmbuf_getall(&xdrs); 524 if (!AUTH_VALIDATE(auth, xid, 525 &reply_msg.acpted_rply.ar_verf, 526 &results)) { 527 errp->re_status = stat = RPC_AUTHERROR; 528 errp->re_why = AUTH_INVALIDRESP; 529 } else { 530 KASSERT(results, 531 ("auth validated but no result")); 532 *resultsp = results; 533 } 534 } /* end successful completion */ 535 /* 536 * If unsuccesful AND error is an authentication error 537 * then refresh credentials and try again, else break 538 */ 539 else if (stat == RPC_AUTHERROR) 540 /* maybe our credentials need to be refreshed ... */ 541 if (nrefreshes > 0 && 542 AUTH_REFRESH(auth, &reply_msg)) { 543 nrefreshes--; 544 XDR_DESTROY(&xdrs); 545 mtx_lock(&ct->ct_lock); 546 goto call_again; 547 } 548 /* end of unsuccessful completion */ 549 } /* end of valid reply message */ 550 else { 551 errp->re_status = stat = RPC_CANTDECODERES; 552 } 553 XDR_DESTROY(&xdrs); 554 mtx_lock(&ct->ct_lock); 555out: 556 mtx_assert(&ct->ct_lock, MA_OWNED); 557 558 KASSERT(stat != RPC_SUCCESS || *resultsp, 559 ("RPC_SUCCESS without reply")); 560 561 if (mreq) 562 m_freem(mreq); 563 if (cr->cr_mrep) 564 m_freem(cr->cr_mrep); 565 566 ct->ct_threads--; 567 if (ct->ct_closing) 568 wakeup(ct); 569 570 mtx_unlock(&ct->ct_lock); 571 572 if (auth && stat != RPC_SUCCESS) 573 AUTH_VALIDATE(auth, xid, NULL, NULL); 574 575 free(cr, M_RPC); 576 577 return (stat); 578} 579 580static void 581clnt_vc_geterr(CLIENT *cl, struct rpc_err *errp) 582{ 583 struct ct_data *ct = (struct ct_data *) cl->cl_private; 584 585 *errp = ct->ct_error; 586} 587 588static bool_t 589clnt_vc_freeres(CLIENT *cl, xdrproc_t xdr_res, void *res_ptr) 590{ 591 XDR xdrs; 592 bool_t dummy; 593 594 xdrs.x_op = XDR_FREE; 595 dummy = (*xdr_res)(&xdrs, res_ptr); 596 597 return (dummy); 598} 599 600/*ARGSUSED*/ 601static void 602clnt_vc_abort(CLIENT *cl) 603{ 604} 605 606static bool_t 607clnt_vc_control(CLIENT *cl, u_int request, void *info) 608{ 609 struct ct_data *ct = (struct ct_data *)cl->cl_private; 610 void *infop = info; 611 612 mtx_lock(&ct->ct_lock); 613 614 switch (request) { 615 case CLSET_FD_CLOSE: 616 ct->ct_closeit = TRUE; 617 mtx_unlock(&ct->ct_lock); 618 return (TRUE); 619 case CLSET_FD_NCLOSE: 620 ct->ct_closeit = FALSE; 621 mtx_unlock(&ct->ct_lock); 622 return (TRUE); 623 default: 624 break; 625 } 626 627 /* for other requests which use info */ 628 if (info == NULL) { 629 mtx_unlock(&ct->ct_lock); 630 return (FALSE); 631 } 632 switch (request) { 633 case CLSET_TIMEOUT: 634 if (time_not_ok((struct timeval *)info)) { 635 mtx_unlock(&ct->ct_lock); 636 return (FALSE); 637 } 638 ct->ct_wait = *(struct timeval *)infop; 639 break; 640 case CLGET_TIMEOUT: 641 *(struct timeval *)infop = ct->ct_wait; 642 break; 643 case CLGET_SERVER_ADDR: 644 (void) memcpy(info, &ct->ct_addr, (size_t)ct->ct_addr.ss_len); 645 break; 646 case CLGET_SVC_ADDR: 647 /* 648 * Slightly different semantics to userland - we use 649 * sockaddr instead of netbuf. 650 */ 651 memcpy(info, &ct->ct_addr, ct->ct_addr.ss_len); 652 break; 653 case CLSET_SVC_ADDR: /* set to new address */ 654 mtx_unlock(&ct->ct_lock); 655 return (FALSE); 656 case CLGET_XID: 657 *(uint32_t *)info = ct->ct_xid; 658 break; 659 case CLSET_XID: 660 /* This will set the xid of the NEXT call */ 661 /* decrement by 1 as clnt_vc_call() increments once */ 662 ct->ct_xid = *(uint32_t *)info - 1; 663 break; 664 case CLGET_VERS: 665 /* 666 * This RELIES on the information that, in the call body, 667 * the version number field is the fifth field from the 668 * begining of the RPC header. MUST be changed if the 669 * call_struct is changed 670 */ 671 *(uint32_t *)info = 672 ntohl(*(uint32_t *)(void *)(ct->ct_mcallc + 673 4 * BYTES_PER_XDR_UNIT)); 674 break; 675 676 case CLSET_VERS: 677 *(uint32_t *)(void *)(ct->ct_mcallc + 678 4 * BYTES_PER_XDR_UNIT) = 679 htonl(*(uint32_t *)info); 680 break; 681 682 case CLGET_PROG: 683 /* 684 * This RELIES on the information that, in the call body, 685 * the program number field is the fourth field from the 686 * begining of the RPC header. MUST be changed if the 687 * call_struct is changed 688 */ 689 *(uint32_t *)info = 690 ntohl(*(uint32_t *)(void *)(ct->ct_mcallc + 691 3 * BYTES_PER_XDR_UNIT)); 692 break; 693 694 case CLSET_PROG: 695 *(uint32_t *)(void *)(ct->ct_mcallc + 696 3 * BYTES_PER_XDR_UNIT) = 697 htonl(*(uint32_t *)info); 698 break; 699 700 case CLSET_WAITCHAN: 701 ct->ct_waitchan = (const char *)info; 702 break; 703 704 case CLGET_WAITCHAN: 705 *(const char **) info = ct->ct_waitchan; 706 break; 707 708 case CLSET_INTERRUPTIBLE: 709 if (*(int *) info) 710 ct->ct_waitflag = PCATCH; 711 else 712 ct->ct_waitflag = 0; 713 break; 714 715 case CLGET_INTERRUPTIBLE: 716 if (ct->ct_waitflag) 717 *(int *) info = TRUE; 718 else 719 *(int *) info = FALSE; 720 break; 721 722 default: 723 mtx_unlock(&ct->ct_lock); 724 return (FALSE); 725 } 726 727 mtx_unlock(&ct->ct_lock); 728 return (TRUE); 729} 730 731static void 732clnt_vc_close(CLIENT *cl) 733{ 734 struct ct_data *ct = (struct ct_data *) cl->cl_private; 735 struct ct_request *cr; 736 737 mtx_lock(&ct->ct_lock); 738 739 if (ct->ct_closed) { 740 mtx_unlock(&ct->ct_lock); 741 return; 742 } 743 744 if (ct->ct_closing) { 745 while (ct->ct_closing) 746 msleep(ct, &ct->ct_lock, 0, "rpcclose", 0); 747 KASSERT(ct->ct_closed, ("client should be closed")); 748 mtx_unlock(&ct->ct_lock); 749 return; 750 } 751 752 if (ct->ct_socket) { 753 SOCKBUF_LOCK(&ct->ct_socket->so_rcv); 754 ct->ct_socket->so_upcallarg = NULL; 755 ct->ct_socket->so_upcall = NULL; 756 ct->ct_socket->so_rcv.sb_flags &= ~SB_UPCALL; 757 SOCKBUF_UNLOCK(&ct->ct_socket->so_rcv); 758 759 /* 760 * Abort any pending requests and wait until everyone 761 * has finished with clnt_vc_call. 762 */ 763 ct->ct_closing = TRUE; 764 TAILQ_FOREACH(cr, &ct->ct_pending, cr_link) { 765 cr->cr_xid = 0; 766 cr->cr_error = ESHUTDOWN; 767 wakeup(cr); 768 } 769 770 while (ct->ct_threads) 771 msleep(ct, &ct->ct_lock, 0, "rpcclose", 0); 772 } 773 774 ct->ct_closing = FALSE; 775 ct->ct_closed = TRUE; 776 mtx_unlock(&ct->ct_lock); 777 wakeup(ct); 778} 779 780static void 781clnt_vc_destroy(CLIENT *cl) 782{ 783 struct ct_data *ct = (struct ct_data *) cl->cl_private; 784 struct socket *so = NULL; 785 786 clnt_vc_close(cl); 787 788 mtx_lock(&ct->ct_lock); 789 790 if (ct->ct_socket) { 791 if (ct->ct_closeit) { 792 so = ct->ct_socket; 793 } 794 } 795 796 mtx_unlock(&ct->ct_lock); 797 798 mtx_destroy(&ct->ct_lock); 799 if (so) { 800 soshutdown(so, SHUT_WR); 801 soclose(so); 802 } 803 mem_free(ct, sizeof(struct ct_data)); 804 mem_free(cl, sizeof(CLIENT)); 805} 806 807/* 808 * Make sure that the time is not garbage. -1 value is disallowed. 809 * Note this is different from time_not_ok in clnt_dg.c 810 */ 811static bool_t 812time_not_ok(struct timeval *t) 813{ 814 return (t->tv_sec <= -1 || t->tv_sec > 100000000 || 815 t->tv_usec <= -1 || t->tv_usec > 1000000); 816} 817 818void 819clnt_vc_soupcall(struct socket *so, void *arg, int waitflag) 820{ 821 struct ct_data *ct = (struct ct_data *) arg; 822 struct uio uio; 823 struct mbuf *m; 824 struct ct_request *cr; 825 int error, rcvflag, foundreq; 826 uint32_t xid, header; 827 bool_t do_read; 828 829 uio.uio_td = curthread; 830 do { 831 /* 832 * If ct_record_resid is zero, we are waiting for a 833 * record mark. 834 */ 835 if (ct->ct_record_resid == 0) { 836 837 /* 838 * Make sure there is either a whole record 839 * mark in the buffer or there is some other 840 * error condition 841 */ 842 do_read = FALSE; 843 SOCKBUF_LOCK(&so->so_rcv); 844 if (so->so_rcv.sb_cc >= sizeof(uint32_t) 845 || (so->so_rcv.sb_state & SBS_CANTRCVMORE) 846 || so->so_error) 847 do_read = TRUE; 848 SOCKBUF_UNLOCK(&so->so_rcv); 849 850 if (!do_read) 851 return; 852 853 uio.uio_resid = sizeof(uint32_t); 854 m = NULL; 855 rcvflag = MSG_DONTWAIT | MSG_SOCALLBCK; 856 error = soreceive(so, NULL, &uio, &m, NULL, &rcvflag); 857 858 if (error == EWOULDBLOCK) 859 break; 860 861 /* 862 * If there was an error, wake up all pending 863 * requests. 864 */ 865 if (error || uio.uio_resid > 0) { 866 wakeup_all: 867 mtx_lock(&ct->ct_lock); 868 if (!error) { 869 /* 870 * We must have got EOF trying 871 * to read from the stream. 872 */ 873 error = ECONNRESET; 874 } 875 ct->ct_error.re_status = RPC_CANTRECV; 876 ct->ct_error.re_errno = error; 877 TAILQ_FOREACH(cr, &ct->ct_pending, cr_link) { 878 cr->cr_error = error; 879 wakeup(cr); 880 } 881 mtx_unlock(&ct->ct_lock); 882 break; 883 } 884 bcopy(mtod(m, uint32_t *), &header, sizeof(uint32_t)); 885 header = ntohl(header); 886 ct->ct_record = NULL; 887 ct->ct_record_resid = header & 0x7fffffff; 888 ct->ct_record_eor = ((header & 0x80000000) != 0); 889 m_freem(m); 890 } else { 891 /* 892 * Wait until the socket has the whole record 893 * buffered. 894 */ 895 do_read = FALSE; 896 SOCKBUF_LOCK(&so->so_rcv); 897 if (so->so_rcv.sb_cc >= ct->ct_record_resid 898 || (so->so_rcv.sb_state & SBS_CANTRCVMORE) 899 || so->so_error) 900 do_read = TRUE; 901 SOCKBUF_UNLOCK(&so->so_rcv); 902 903 if (!do_read) 904 return; 905 906 /* 907 * We have the record mark. Read as much as 908 * the socket has buffered up to the end of 909 * this record. 910 */ 911 uio.uio_resid = ct->ct_record_resid; 912 m = NULL; 913 rcvflag = MSG_DONTWAIT | MSG_SOCALLBCK; 914 error = soreceive(so, NULL, &uio, &m, NULL, &rcvflag); 915 916 if (error == EWOULDBLOCK) 917 break; 918 919 if (error || uio.uio_resid == ct->ct_record_resid) 920 goto wakeup_all; 921 922 /* 923 * If we have part of the record already, 924 * chain this bit onto the end. 925 */ 926 if (ct->ct_record) 927 m_last(ct->ct_record)->m_next = m; 928 else 929 ct->ct_record = m; 930 931 ct->ct_record_resid = uio.uio_resid; 932 933 /* 934 * If we have the entire record, see if we can 935 * match it to a request. 936 */ 937 if (ct->ct_record_resid == 0 938 && ct->ct_record_eor) { 939 /* 940 * The XID is in the first uint32_t of 941 * the reply. 942 */ 943 if (ct->ct_record->m_len < sizeof(xid)) 944 ct->ct_record = 945 m_pullup(ct->ct_record, 946 sizeof(xid)); 947 if (!ct->ct_record) 948 break; 949 bcopy(mtod(ct->ct_record, uint32_t *), 950 &xid, sizeof(uint32_t)); 951 xid = ntohl(xid); 952 953 mtx_lock(&ct->ct_lock); 954 foundreq = 0; 955 TAILQ_FOREACH(cr, &ct->ct_pending, cr_link) { 956 if (cr->cr_xid == xid) { 957 /* 958 * This one 959 * matches. We leave 960 * the reply mbuf in 961 * cr->cr_mrep. Set 962 * the XID to zero so 963 * that we will ignore 964 * any duplicaed 965 * replies. 966 */ 967 cr->cr_xid = 0; 968 cr->cr_mrep = ct->ct_record; 969 cr->cr_error = 0; 970 foundreq = 1; 971 wakeup(cr); 972 break; 973 } 974 } 975 mtx_unlock(&ct->ct_lock); 976 977 if (!foundreq) 978 m_freem(ct->ct_record); 979 ct->ct_record = NULL; 980 } 981 } 982 } while (m); 983} 984