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