sys/rpc/clnt_vc.c

   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)
  33 static char *sccsid2 = "@(#)clnt_tcp.c 1.37 87/10/05 Copyr 1984 Sun Micro";
  34 static char *sccsid = "@(#)clnt_tcp.c   2.2 88/08/01 4.0 RPCSRC";
  35 static char sccsid3[] = "@(#)clnt_vc.c 1.19 89/03/16 Copyr 1988 Sun Micro";
  36 #endif
  37 #include <sys/cdefs.h>
  38 __FBSDID("$FreeBSD$");
  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/socket.h>
  68 #include <sys/socketvar.h>
  69 #include <sys/syslog.h>
  70 #include <sys/time.h>
  71 #include <sys/uio.h>
  72
  73 #include <rpc/rpc.h>
  74 #include <rpc/rpc_com.h>
  75
  76 #define MCALL_MSG_SIZE 24
  77
  78 struct cmessage {
  79         struct cmsghdr cmsg;
  80         struct cmsgcred cmcred;
  81 };
  82
  83 static enum clnt_stat clnt_vc_call(CLIENT *, struct rpc_callextra *,
  84     rpcproc_t, xdrproc_t, void *, xdrproc_t, void *, struct timeval);
  85 static void clnt_vc_geterr(CLIENT *, struct rpc_err *);
  86 static bool_t clnt_vc_freeres(CLIENT *, xdrproc_t, void *);
  87 static void clnt_vc_abort(CLIENT *);
  88 static bool_t clnt_vc_control(CLIENT *, u_int, void *);
  89 static void clnt_vc_destroy(CLIENT *);
  90 static bool_t time_not_ok(struct timeval *);
  91 static void clnt_vc_soupcall(struct socket *so, void *arg, int waitflag);
  92
  93 static struct clnt_ops clnt_vc_ops = {
  94         .cl_call =      clnt_vc_call,
  95         .cl_abort =     clnt_vc_abort,
  96         .cl_geterr =    clnt_vc_geterr,
  97         .cl_freeres =   clnt_vc_freeres,
  98         .cl_destroy =   clnt_vc_destroy,
  99         .cl_control =   clnt_vc_control
 100 };
 101
 102 /*
 103  * A pending RPC request which awaits a reply. Requests which have
 104  * received their reply will have cr_xid set to zero and cr_mrep to
 105  * the mbuf chain of the reply.
 106  */
 107 struct ct_request {
 108         TAILQ_ENTRY(ct_request) cr_link;
 109         uint32_t                cr_xid;         /* XID of request */
 110         struct mbuf             *cr_mrep;       /* reply received by upcall */
 111         int                     cr_error;       /* any error from upcall */
 112 };
 113
 114 TAILQ_HEAD(ct_request_list, ct_request);
 115
 116 struct ct_data {
 117         struct mtx      ct_lock;
 118         int             ct_threads;     /* number of threads in clnt_vc_call */
 119         bool_t          ct_closing;     /* TRUE if we are destroying client */
 120         struct socket   *ct_socket;     /* connection socket */
 121         bool_t          ct_closeit;     /* close it on destroy */
 122         struct timeval  ct_wait;        /* wait interval in milliseconds */
 123         struct sockaddr_storage ct_addr; /* remote addr */
 124         struct rpc_err  ct_error;
 125         uint32_t        ct_xid;
 126         char            ct_mcallc[MCALL_MSG_SIZE]; /* marshalled callmsg */
 127         size_t          ct_mpos;        /* pos after marshal */
 128         const char      *ct_waitchan;
 129         int             ct_waitflag;
 130         struct mbuf     *ct_record;     /* current reply record */
 131         size_t          ct_record_resid; /* how much left of reply to read */
 132         bool_t          ct_record_eor;   /* true if reading last fragment */
 133         struct ct_request_list ct_pending;
 134 };
 135
 136 static const char clnt_vc_errstr[] = "%s : %s";
 137 static const char clnt_vc_str[] = "clnt_vc_create";
 138 static const char clnt_read_vc_str[] = "read_vc";
 139 static const char __no_mem_str[] = "out of memory";
 140
 141 /*
 142  * Create a client handle for a connection.
 143  * Default options are set, which the user can change using clnt_control()'s.
 144  * The rpc/vc package does buffering similar to stdio, so the client
 145  * must pick send and receive buffer sizes, 0 => use the default.
 146  * NB: fd is copied into a private area.
 147  * NB: The rpch->cl_auth is set null authentication. Caller may wish to
 148  * set this something more useful.
 149  *
 150  * fd should be an open socket
 151  */
 152 CLIENT *
 153 clnt_vc_create(
 154         struct socket *so,              /* open file descriptor */
 155         struct sockaddr *raddr,         /* servers address */
 156         const rpcprog_t prog,           /* program number */
 157         const rpcvers_t vers,           /* version number */
 158         size_t sendsz,                  /* buffer recv size */
 159         size_t recvsz)                  /* buffer send size */
 160 {
 161         CLIENT *cl;                     /* client handle */
 162         struct ct_data *ct = NULL;      /* client handle */
 163         struct timeval now;
 164         struct rpc_msg call_msg;
 165         static uint32_t disrupt;
 166         struct __rpc_sockinfo si;
 167         XDR xdrs;
 168         int error, interrupted;
 169
 170         if (disrupt == 0)
 171                 disrupt = (uint32_t)(long)raddr;
 172
 173         cl = (CLIENT *)mem_alloc(sizeof (*cl));
 174         ct = (struct ct_data *)mem_alloc(sizeof (*ct));
 175
 176         mtx_init(&ct->ct_lock, "ct->ct_lock", NULL, MTX_DEF);
 177         ct->ct_threads = 0;
 178         ct->ct_closing = FALSE;
 179
 180         if ((so->so_state & (SS_ISCONNECTED|SS_ISCONFIRMING)) == 0) {
 181                 error = soconnect(so, raddr, curthread);
 182                 SOCK_LOCK(so);
 183                 interrupted = 0;
 184                 while ((so->so_state & SS_ISCONNECTING)
 185                     && so->so_error == 0) {
 186                         error = msleep(&so->so_timeo, SOCK_MTX(so),
 187                             PSOCK | PCATCH, "connec", 0);
 188                         if (error) {
 189                                 if (error == EINTR || error == ERESTART)
 190                                         interrupted = 1;
 191                                 break;
 192                         }
 193                 }
 194                 if (error == 0) {
 195                         error = so->so_error;
 196                         so->so_error = 0;
 197                 }
 198                 SOCK_UNLOCK(so);
 199                 if (error) {
 200                         if (!interrupted)
 201                                 so->so_state &= ~SS_ISCONNECTING;
 202                         rpc_createerr.cf_stat = RPC_SYSTEMERROR;
 203                         rpc_createerr.cf_error.re_errno = error;
 204                         goto err;
 205                 }
 206         }
 207
 208         if (!__rpc_socket2sockinfo(so, &si))
 209                 goto err;
 210
 211         ct->ct_closeit = FALSE;
 212
 213         /*
 214          * Set up private data struct
 215          */
 216         ct->ct_socket = so;
 217         ct->ct_wait.tv_sec = -1;
 218         ct->ct_wait.tv_usec = -1;
 219         memcpy(&ct->ct_addr, raddr, raddr->sa_len);
 220
 221         /*
 222          * Initialize call message
 223          */
 224         getmicrotime(&now);
 225         ct->ct_xid = ((uint32_t)++disrupt) ^ __RPC_GETXID(&now);
 226         call_msg.rm_xid = ct->ct_xid;
 227         call_msg.rm_direction = CALL;
 228         call_msg.rm_call.cb_rpcvers = RPC_MSG_VERSION;
 229         call_msg.rm_call.cb_prog = (uint32_t)prog;
 230         call_msg.rm_call.cb_vers = (uint32_t)vers;
 231
 232         /*
 233          * pre-serialize the static part of the call msg and stash it away
 234          */
 235         xdrmem_create(&xdrs, ct->ct_mcallc, MCALL_MSG_SIZE,
 236             XDR_ENCODE);
 237         if (! xdr_callhdr(&xdrs, &call_msg)) {
 238                 if (ct->ct_closeit) {
 239                         soclose(ct->ct_socket);
 240                 }
 241                 goto err;
 242         }
 243         ct->ct_mpos = XDR_GETPOS(&xdrs);
 244         XDR_DESTROY(&xdrs);
 245         ct->ct_waitchan = "rpcrecv";
 246         ct->ct_waitflag = 0;
 247
 248         /*
 249          * Create a client handle which uses xdrrec for serialization
 250          * and authnone for authentication.
 251          */
 252         cl->cl_refs = 1;
 253         cl->cl_ops = &clnt_vc_ops;
 254         cl->cl_private = ct;
 255         cl->cl_auth = authnone_create();
 256         sendsz = __rpc_get_t_size(si.si_af, si.si_proto, (int)sendsz);
 257         recvsz = __rpc_get_t_size(si.si_af, si.si_proto, (int)recvsz);
 258
 259         SOCKBUF_LOCK(&ct->ct_socket->so_rcv);
 260         ct->ct_socket->so_upcallarg = ct;
 261         ct->ct_socket->so_upcall = clnt_vc_soupcall;
 262         ct->ct_socket->so_rcv.sb_flags |= SB_UPCALL;
 263         SOCKBUF_UNLOCK(&ct->ct_socket->so_rcv);
 264
 265         ct->ct_record = NULL;
 266         ct->ct_record_resid = 0;
 267         TAILQ_INIT(&ct->ct_pending);
 268         return (cl);
 269
 270 err:
 271         if (cl) {
 272                 if (ct) {
 273                         mem_free(ct, sizeof (struct ct_data));
 274                 }
 275                 if (cl)
 276                         mem_free(cl, sizeof (CLIENT));
 277         }
 278         return ((CLIENT *)NULL);
 279 }
 280
 281 static enum clnt_stat
 282 clnt_vc_call(
 283         CLIENT *cl,
 284         struct rpc_callextra *ext,
 285         rpcproc_t proc,
 286         xdrproc_t xdr_args,
 287         void *args_ptr,
 288         xdrproc_t xdr_results,
 289         void *results_ptr,
 290         struct timeval utimeout)
 291 {
 292         struct ct_data *ct = (struct ct_data *) cl->cl_private;
 293         AUTH *auth;
 294         XDR xdrs;
 295         struct rpc_msg reply_msg;
 296         bool_t ok;
 297         int nrefreshes = 2;             /* number of times to refresh cred */
 298         struct timeval timeout;
 299         uint32_t xid;
 300         struct mbuf *mreq = NULL;
 301         struct ct_request *cr;
 302         int error;
 303
 304         cr = malloc(sizeof(struct ct_request), M_RPC, M_WAITOK);
 305
 306         mtx_lock(&ct->ct_lock);
 307
 308         if (ct->ct_closing) {
 309                 mtx_unlock(&ct->ct_lock);
 310                 free(cr, M_RPC);
 311                 return (RPC_CANTSEND);
 312         }
 313         ct->ct_threads++;
 314
 315         if (ext)
 316                 auth = ext->rc_auth;
 317         else
 318                 auth = cl->cl_auth;
 319
 320         cr->cr_mrep = NULL;
 321         cr->cr_error = 0;
 322
 323         if (ct->ct_wait.tv_usec == -1) {
 324                 timeout = utimeout;     /* use supplied timeout */
 325         } else {
 326                 timeout = ct->ct_wait;  /* use default timeout */
 327         }
 328
 329 call_again:
 330         mtx_assert(&ct->ct_lock, MA_OWNED);
 331
 332         ct->ct_xid++;
 333         xid = ct->ct_xid;
 334
 335         mtx_unlock(&ct->ct_lock);
 336
 337         /*
 338          * Leave space to pre-pend the record mark.
 339          */
 340         MGETHDR(mreq, M_WAIT, MT_DATA);
 341         MCLGET(mreq, M_WAIT);
 342         mreq->m_len = 0;
 343         mreq->m_data += sizeof(uint32_t);
 344         m_append(mreq, ct->ct_mpos, ct->ct_mcallc);
 345
 346         /*
 347          * The XID is the first thing in the request.
 348          */
 349         *mtod(mreq, uint32_t *) = htonl(xid);
 350
 351         xdrmbuf_create(&xdrs, mreq, XDR_ENCODE);
 352
 353         ct->ct_error.re_status = RPC_SUCCESS;
 354
 355         if ((! XDR_PUTINT32(&xdrs, &proc)) ||
 356             (! AUTH_MARSHALL(auth, &xdrs)) ||
 357             (! (*xdr_args)(&xdrs, args_ptr))) {
 358                 if (ct->ct_error.re_status == RPC_SUCCESS)
 359                         ct->ct_error.re_status = RPC_CANTENCODEARGS;
 360                 mtx_lock(&ct->ct_lock);
 361                 goto out;
 362         }
 363         m_fixhdr(mreq);
 364
 365         /*
 366          * Prepend a record marker containing the packet length.
 367          */
 368         M_PREPEND(mreq, sizeof(uint32_t), M_WAIT);
 369         *mtod(mreq, uint32_t *) =
 370                 htonl(0x80000000 | (mreq->m_pkthdr.len - sizeof(uint32_t)));
 371
 372         cr->cr_xid = xid;
 373         mtx_lock(&ct->ct_lock);
 374         TAILQ_INSERT_TAIL(&ct->ct_pending, cr, cr_link);
 375         mtx_unlock(&ct->ct_lock);
 376
 377         /*
 378          * sosend consumes mreq.
 379          */
 380         error = sosend(ct->ct_socket, NULL, NULL, mreq, NULL, 0, curthread);
 381         mreq = NULL;
 382
 383         reply_msg.acpted_rply.ar_verf = _null_auth;
 384         reply_msg.acpted_rply.ar_results.where = results_ptr;
 385         reply_msg.acpted_rply.ar_results.proc = xdr_results;
 386
 387         mtx_lock(&ct->ct_lock);
 388         if (error) {
 389                 TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
 390                 ct->ct_error.re_errno = error;
 391                 ct->ct_error.re_status = RPC_CANTSEND;
 392                 goto out;
 393         }
 394
 395         /*
 396          * Check to see if we got an upcall while waiting for the
 397          * lock. In both these cases, the request has been removed
 398          * from ct->ct_pending.
 399          */
 400         if (cr->cr_error) {
 401                 TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
 402                 ct->ct_error.re_errno = cr->cr_error;
 403                 ct->ct_error.re_status = RPC_CANTRECV;
 404                 goto out;
 405         }
 406         if (cr->cr_mrep) {
 407                 TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
 408                 goto got_reply;
 409         }
 410
 411         /*
 412          * Hack to provide rpc-based message passing
 413          */
 414         if (timeout.tv_sec == 0 && timeout.tv_usec == 0) {
 415                 TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
 416                 ct->ct_error.re_status = RPC_TIMEDOUT;
 417                 goto out;
 418         }
 419
 420         error = msleep(cr, &ct->ct_lock, ct->ct_waitflag, ct->ct_waitchan,
 421             tvtohz(&timeout));
 422
 423         TAILQ_REMOVE(&ct->ct_pending, cr, cr_link);
 424
 425         if (error) {
 426                 /*
 427                  * The sleep returned an error so our request is still
 428                  * on the list. Turn the error code into an
 429                  * appropriate client status.
 430                  */
 431                 ct->ct_error.re_errno = error;
 432                 switch (error) {
 433                 case EINTR:
 434                         ct->ct_error.re_status = RPC_INTR;
 435                         break;
 436                 case EWOULDBLOCK:
 437                         ct->ct_error.re_status = RPC_TIMEDOUT;
 438                         break;
 439                 default:
 440                         ct->ct_error.re_status = RPC_CANTRECV;
 441                 }
 442                 goto out;
 443         } else {
 444                 /*
 445                  * We were woken up by the upcall.  If the
 446                  * upcall had a receive error, report that,
 447                  * otherwise we have a reply.
 448                  */
 449                 if (cr->cr_error) {
 450                         ct->ct_error.re_errno = cr->cr_error;
 451                         ct->ct_error.re_status = RPC_CANTRECV;
 452                         goto out;
 453                 }
 454         }
 455
 456 got_reply:
 457         /*
 458          * Now decode and validate the response. We need to drop the
 459          * lock since xdr_replymsg may end up sleeping in malloc.
 460          */
 461         mtx_unlock(&ct->ct_lock);
 462
 463         xdrmbuf_create(&xdrs, cr->cr_mrep, XDR_DECODE);
 464         ok = xdr_replymsg(&xdrs, &reply_msg);
 465         XDR_DESTROY(&xdrs);
 466         cr->cr_mrep = NULL;
 467
 468         mtx_lock(&ct->ct_lock);
 469
 470         if (ok) {
 471                 if ((reply_msg.rm_reply.rp_stat == MSG_ACCEPTED) &&
 472                         (reply_msg.acpted_rply.ar_stat == SUCCESS))
 473                         ct->ct_error.re_status = RPC_SUCCESS;
 474                 else
 475                         _seterr_reply(&reply_msg, &(ct->ct_error));
 476
 477                 if (ct->ct_error.re_status == RPC_SUCCESS) {
 478                         if (! AUTH_VALIDATE(cl->cl_auth,
 479                                             &reply_msg.acpted_rply.ar_verf)) {
 480                                 ct->ct_error.re_status = RPC_AUTHERROR;
 481                                 ct->ct_error.re_why = AUTH_INVALIDRESP;
 482                         }
 483                         if (reply_msg.acpted_rply.ar_verf.oa_base != NULL) {
 484                                 xdrs.x_op = XDR_FREE;
 485                                 (void) xdr_opaque_auth(&xdrs,
 486                                         &(reply_msg.acpted_rply.ar_verf));
 487                         }
 488                 }               /* end successful completion */
 489                 /*
 490                  * If unsuccesful AND error is an authentication error
 491                  * then refresh credentials and try again, else break
 492                  */
 493                 else if (ct->ct_error.re_status == RPC_AUTHERROR)
 494                         /* maybe our credentials need to be refreshed ... */
 495                         if (nrefreshes > 0 &&
 496                             AUTH_REFRESH(cl->cl_auth, &reply_msg)) {
 497                                 nrefreshes--;
 498                                 goto call_again;
 499                         }
 500                 /* end of unsuccessful completion */
 501         }       /* end of valid reply message */
 502         else {
 503                 ct->ct_error.re_status = RPC_CANTDECODERES;
 504         }
 505 out:
 506         mtx_assert(&ct->ct_lock, MA_OWNED);
 507
 508         if (mreq)
 509                 m_freem(mreq);
 510         if (cr->cr_mrep)
 511                 m_freem(cr->cr_mrep);
 512
 513         ct->ct_threads--;
 514         if (ct->ct_closing)
 515                 wakeup(ct);
 516
 517         mtx_unlock(&ct->ct_lock);
 518
 519         free(cr, M_RPC);
 520
 521         return (ct->ct_error.re_status);
 522 }
 523
 524 static void
 525 clnt_vc_geterr(CLIENT *cl, struct rpc_err *errp)
 526 {
 527         struct ct_data *ct = (struct ct_data *) cl->cl_private;
 528
 529         *errp = ct->ct_error;
 530 }
 531
 532 static bool_t
 533 clnt_vc_freeres(CLIENT *cl, xdrproc_t xdr_res, void *res_ptr)
 534 {
 535         XDR xdrs;
 536         bool_t dummy;
 537
 538         xdrs.x_op = XDR_FREE;
 539         dummy = (*xdr_res)(&xdrs, res_ptr);
 540
 541         return (dummy);
 542 }
 543
 544 /*ARGSUSED*/
 545 static void
 546 clnt_vc_abort(CLIENT *cl)
 547 {
 548 }
 549
 550 static bool_t
 551 clnt_vc_control(CLIENT *cl, u_int request, void *info)
 552 {
 553         struct ct_data *ct = (struct ct_data *)cl->cl_private;
 554         void *infop = info;
 555
 556         mtx_lock(&ct->ct_lock);
 557
 558         switch (request) {
 559         case CLSET_FD_CLOSE:
 560                 ct->ct_closeit = TRUE;
 561                 mtx_unlock(&ct->ct_lock);
 562                 return (TRUE);
 563         case CLSET_FD_NCLOSE:
 564                 ct->ct_closeit = FALSE;
 565                 mtx_unlock(&ct->ct_lock);
 566                 return (TRUE);
 567         default:
 568                 break;
 569         }
 570
 571         /* for other requests which use info */
 572         if (info == NULL) {
 573                 mtx_unlock(&ct->ct_lock);
 574                 return (FALSE);
 575         }
 576         switch (request) {
 577         case CLSET_TIMEOUT:
 578                 if (time_not_ok((struct timeval *)info)) {
 579                         mtx_unlock(&ct->ct_lock);
 580                         return (FALSE);
 581                 }
 582                 ct->ct_wait = *(struct timeval *)infop;
 583                 break;
 584         case CLGET_TIMEOUT:
 585                 *(struct timeval *)infop = ct->ct_wait;
 586                 break;
 587         case CLGET_SERVER_ADDR:
 588                 (void) memcpy(info, &ct->ct_addr, (size_t)ct->ct_addr.ss_len);
 589                 break;
 590         case CLGET_SVC_ADDR:
 591                 /*
 592                  * Slightly different semantics to userland - we use
 593                  * sockaddr instead of netbuf.
 594                  */
 595                 memcpy(info, &ct->ct_addr, ct->ct_addr.ss_len);
 596                 break;
 597         case CLSET_SVC_ADDR:            /* set to new address */
 598                 mtx_unlock(&ct->ct_lock);
 599                 return (FALSE);
 600         case CLGET_XID:
 601                 *(uint32_t *)info = ct->ct_xid;
 602                 break;
 603         case CLSET_XID:
 604                 /* This will set the xid of the NEXT call */
 605                 /* decrement by 1 as clnt_vc_call() increments once */
 606                 ct->ct_xid = *(uint32_t *)info - 1;
 607                 break;
 608         case CLGET_VERS:
 609                 /*
 610                  * This RELIES on the information that, in the call body,
 611                  * the version number field is the fifth field from the
 612                  * begining of the RPC header. MUST be changed if the
 613                  * call_struct is changed
 614                  */
 615                 *(uint32_t *)info =
 616                     ntohl(*(uint32_t *)(void *)(ct->ct_mcallc +
 617                     4 * BYTES_PER_XDR_UNIT));
 618                 break;
 619
 620         case CLSET_VERS:
 621                 *(uint32_t *)(void *)(ct->ct_mcallc +
 622                     4 * BYTES_PER_XDR_UNIT) =
 623                     htonl(*(uint32_t *)info);
 624                 break;
 625
 626         case CLGET_PROG:
 627                 /*
 628                  * This RELIES on the information that, in the call body,
 629                  * the program number field is the fourth field from the
 630                  * begining of the RPC header. MUST be changed if the
 631                  * call_struct is changed
 632                  */
 633                 *(uint32_t *)info =
 634                     ntohl(*(uint32_t *)(void *)(ct->ct_mcallc +
 635                     3 * BYTES_PER_XDR_UNIT));
 636                 break;
 637
 638         case CLSET_PROG:
 639                 *(uint32_t *)(void *)(ct->ct_mcallc +
 640                     3 * BYTES_PER_XDR_UNIT) =
 641                     htonl(*(uint32_t *)info);
 642                 break;
 643
 644         case CLSET_WAITCHAN:
 645                 ct->ct_waitchan = *(const char **)info;
 646                 break;
 647
 648         case CLGET_WAITCHAN:
 649                 *(const char **) info = ct->ct_waitchan;
 650                 break;
 651
 652         case CLSET_INTERRUPTIBLE:
 653                 if (*(int *) info)
 654                         ct->ct_waitflag = PCATCH;
 655                 else
 656                         ct->ct_waitflag = 0;
 657                 break;
 658
 659         case CLGET_INTERRUPTIBLE:
 660                 if (ct->ct_waitflag)
 661                         *(int *) info = TRUE;
 662                 else
 663                         *(int *) info = FALSE;
 664                 break;
 665
 666         default:
 667                 mtx_unlock(&ct->ct_lock);
 668                 return (FALSE);
 669         }
 670
 671         mtx_unlock(&ct->ct_lock);
 672         return (TRUE);
 673 }
 674
 675 static void
 676 clnt_vc_destroy(CLIENT *cl)
 677 {
 678         struct ct_data *ct = (struct ct_data *) cl->cl_private;
 679         struct ct_request *cr;
 680         struct socket *so = NULL;
 681
 682         mtx_lock(&ct->ct_lock);
 683
 684         if (ct->ct_socket) {
 685                 SOCKBUF_LOCK(&ct->ct_socket->so_rcv);
 686                 ct->ct_socket->so_upcallarg = NULL;
 687                 ct->ct_socket->so_upcall = NULL;
 688                 ct->ct_socket->so_rcv.sb_flags &= ~SB_UPCALL;
 689                 SOCKBUF_UNLOCK(&ct->ct_socket->so_rcv);
 690
 691                 /*
 692                  * Abort any pending requests and wait until everyone
 693                  * has finished with clnt_vc_call.
 694                  */
 695                 ct->ct_closing = TRUE;
 696                 TAILQ_FOREACH(cr, &ct->ct_pending, cr_link) {
 697                         cr->cr_xid = 0;
 698                         cr->cr_error = ESHUTDOWN;
 699                         wakeup(cr);
 700                 }
 701
 702                 while (ct->ct_threads)
 703                         msleep(ct, &ct->ct_lock, 0, "rpcclose", 0);
 704
 705                 if (ct->ct_closeit) {
 706                         so = ct->ct_socket;
 707                 }
 708         }
 709
 710         mtx_unlock(&ct->ct_lock);
 711
 712         mtx_destroy(&ct->ct_lock);
 713         if (so) {
 714                 soshutdown(so, SHUT_WR);
 715                 soclose(so);
 716         }
 717         mem_free(ct, sizeof(struct ct_data));
 718         mem_free(cl, sizeof(CLIENT));
 719 }
 720
 721 /*
 722  * Make sure that the time is not garbage.   -1 value is disallowed.
 723  * Note this is different from time_not_ok in clnt_dg.c
 724  */
 725 static bool_t
 726 time_not_ok(struct timeval *t)
 727 {
 728         return (t->tv_sec <= -1 || t->tv_sec > 100000000 ||
 729                 t->tv_usec <= -1 || t->tv_usec > 1000000);
 730 }
 731
 732 void
 733 clnt_vc_soupcall(struct socket *so, void *arg, int waitflag)
 734 {
 735         struct ct_data *ct = (struct ct_data *) arg;
 736         struct uio uio;
 737         struct mbuf *m;
 738         struct ct_request *cr;
 739         int error, rcvflag, foundreq;
 740         uint32_t xid, header;
 741
 742         uio.uio_td = curthread;
 743         do {
 744                 /*
 745                  * If ct_record_resid is zero, we are waiting for a
 746                  * record mark.
 747                  */
 748                 if (ct->ct_record_resid == 0) {
 749                         bool_t do_read;
 750
 751                         /*
 752                          * Make sure there is either a whole record
 753                          * mark in the buffer or there is some other
 754                          * error condition
 755                          */
 756                         do_read = FALSE;
 757                         SOCKBUF_LOCK(&so->so_rcv);
 758                         if (so->so_rcv.sb_cc >= sizeof(uint32_t)
 759                             || (so->so_rcv.sb_state & SBS_CANTRCVMORE)
 760                             || so->so_error)
 761                                 do_read = TRUE;
 762                         SOCKBUF_UNLOCK(&so->so_rcv);
 763
 764                         if (!do_read)
 765                                 return;
 766
 767                         uio.uio_resid = sizeof(uint32_t);
 768                         m = NULL;
 769                         rcvflag = MSG_DONTWAIT | MSG_SOCALLBCK;
 770                         error = soreceive(so, NULL, &uio, &m, NULL, &rcvflag);
 771
 772                         if (error == EWOULDBLOCK)
 773                                 break;
 774
 775                         /*
 776                          * If there was an error, wake up all pending
 777                          * requests.
 778                          */
 779                         if (error || uio.uio_resid > 0) {
 780                         wakeup_all:
 781                                 mtx_lock(&ct->ct_lock);
 782                                 if (!error) {
 783                                         /*
 784                                          * We must have got EOF trying
 785                                          * to read from the stream.
 786                                          */
 787                                         error = ECONNRESET;
 788                                 }
 789                                 ct->ct_error.re_status = RPC_CANTRECV;
 790                                 ct->ct_error.re_errno = error;
 791                                 TAILQ_FOREACH(cr, &ct->ct_pending, cr_link) {
 792                                         cr->cr_error = error;
 793                                         wakeup(cr);
 794                                 }
 795                                 mtx_unlock(&ct->ct_lock);
 796                                 break;
 797                         }
 798                         memcpy(&header, mtod(m, uint32_t *), sizeof(uint32_t));
 799                         header = ntohl(header);
 800                         ct->ct_record = NULL;
 801                         ct->ct_record_resid = header & 0x7fffffff;
 802                         ct->ct_record_eor = ((header & 0x80000000) != 0);
 803                         m_freem(m);
 804                 } else {
 805                         /*
 806                          * We have the record mark. Read as much as
 807                          * the socket has buffered up to the end of
 808                          * this record.
 809                          */
 810                         uio.uio_resid = ct->ct_record_resid;
 811                         m = NULL;
 812                         rcvflag = MSG_DONTWAIT | MSG_SOCALLBCK;
 813                         error = soreceive(so, NULL, &uio, &m, NULL, &rcvflag);
 814
 815                         if (error == EWOULDBLOCK)
 816                                 break;
 817
 818                         if (error || uio.uio_resid == ct->ct_record_resid)
 819                                 goto wakeup_all;
 820
 821                         /*
 822                          * If we have part of the record already,
 823                          * chain this bit onto the end.
 824                          */
 825                         if (ct->ct_record)
 826                                 m_last(ct->ct_record)->m_next = m;
 827                         else
 828                                 ct->ct_record = m;
 829
 830                         ct->ct_record_resid = uio.uio_resid;
 831
 832                         /*
 833                          * If we have the entire record, see if we can
 834                          * match it to a request.
 835                          */
 836                         if (ct->ct_record_resid == 0
 837                             && ct->ct_record_eor) {
 838                                 /*
 839                                  * The XID is in the first uint32_t of
 840                                  * the reply.
 841                                  */
 842                                 ct->ct_record =
 843                                         m_pullup(ct->ct_record, sizeof(xid));
 844                                 if (!ct->ct_record)
 845                                         break;
 846                                 memcpy(&xid,
 847                                     mtod(ct->ct_record, uint32_t *),
 848                                     sizeof(uint32_t));
 849                                 xid = ntohl(xid);
 850
 851                                 mtx_lock(&ct->ct_lock);
 852                                 foundreq = 0;
 853                                 TAILQ_FOREACH(cr, &ct->ct_pending, cr_link) {
 854                                         if (cr->cr_xid == xid) {
 855                                                 /*
 856                                                  * This one
 857                                                  * matches. We leave
 858                                                  * the reply mbuf in
 859                                                  * cr->cr_mrep. Set
 860                                                  * the XID to zero so
 861                                                  * that we will ignore
 862                                                  * any duplicaed
 863                                                  * replies.
 864                                                  */
 865                                                 cr->cr_xid = 0;
 866                                                 cr->cr_mrep = ct->ct_record;
 867                                                 cr->cr_error = 0;
 868                                                 foundreq = 1;
 869                                                 wakeup(cr);
 870                                                 break;
 871                                         }
 872                                 }
 873                                 mtx_unlock(&ct->ct_lock);
 874
 875                                 if (!foundreq)
 876                                         m_freem(ct->ct_record);
 877                                 ct->ct_record = NULL;
 878                         }
 879                 }
 880         } while (m);
 881 }