svc.c revision 261053
1/* $NetBSD: svc.c,v 1.21 2000/07/06 03:10:35 christos Exp $ */ 2 3/*- 4 * Copyright (c) 2009, Sun Microsystems, Inc. 5 * All rights reserved. 6 * 7 * Redistribution and use in source and binary forms, with or without 8 * modification, are permitted provided that the following conditions are met: 9 * - Redistributions of source code must retain the above copyright notice, 10 * this list of conditions and the following disclaimer. 11 * - Redistributions in binary form must reproduce the above copyright notice, 12 * this list of conditions and the following disclaimer in the documentation 13 * and/or other materials provided with the distribution. 14 * - Neither the name of Sun Microsystems, Inc. nor the names of its 15 * contributors may be used to endorse or promote products derived 16 * from this software without specific prior written permission. 17 * 18 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 19 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 20 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 21 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE 22 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 23 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 24 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 25 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 26 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 27 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 28 * POSSIBILITY OF SUCH DAMAGE. 29 */ 30 31#if defined(LIBC_SCCS) && !defined(lint) 32static char *sccsid2 = "@(#)svc.c 1.44 88/02/08 Copyr 1984 Sun Micro"; 33static char *sccsid = "@(#)svc.c 2.4 88/08/11 4.0 RPCSRC"; 34#endif 35#include <sys/cdefs.h> 36__FBSDID("$FreeBSD: stable/10/sys/rpc/svc.c 261053 2014-01-22 23:51:12Z mav $"); 37 38/* 39 * svc.c, Server-side remote procedure call interface. 40 * 41 * There are two sets of procedures here. The xprt routines are 42 * for handling transport handles. The svc routines handle the 43 * list of service routines. 44 * 45 * Copyright (C) 1984, Sun Microsystems, Inc. 46 */ 47 48#include <sys/param.h> 49#include <sys/lock.h> 50#include <sys/kernel.h> 51#include <sys/kthread.h> 52#include <sys/malloc.h> 53#include <sys/mbuf.h> 54#include <sys/mutex.h> 55#include <sys/proc.h> 56#include <sys/queue.h> 57#include <sys/socketvar.h> 58#include <sys/systm.h> 59#include <sys/ucred.h> 60 61#include <rpc/rpc.h> 62#include <rpc/rpcb_clnt.h> 63#include <rpc/replay.h> 64 65#include <rpc/rpc_com.h> 66 67#define SVC_VERSQUIET 0x0001 /* keep quiet about vers mismatch */ 68#define version_keepquiet(xp) (SVC_EXT(xp)->xp_flags & SVC_VERSQUIET) 69 70static struct svc_callout *svc_find(SVCPOOL *pool, rpcprog_t, rpcvers_t, 71 char *); 72static void svc_new_thread(SVCPOOL *pool); 73static void xprt_unregister_locked(SVCXPRT *xprt); 74 75/* *************** SVCXPRT related stuff **************** */ 76 77static int svcpool_minthread_sysctl(SYSCTL_HANDLER_ARGS); 78static int svcpool_maxthread_sysctl(SYSCTL_HANDLER_ARGS); 79 80SVCPOOL* 81svcpool_create(const char *name, struct sysctl_oid_list *sysctl_base) 82{ 83 SVCPOOL *pool; 84 85 pool = malloc(sizeof(SVCPOOL), M_RPC, M_WAITOK|M_ZERO); 86 87 mtx_init(&pool->sp_lock, "sp_lock", NULL, MTX_DEF); 88 pool->sp_name = name; 89 pool->sp_state = SVCPOOL_INIT; 90 pool->sp_proc = NULL; 91 TAILQ_INIT(&pool->sp_xlist); 92 TAILQ_INIT(&pool->sp_active); 93 TAILQ_INIT(&pool->sp_callouts); 94 LIST_INIT(&pool->sp_threads); 95 LIST_INIT(&pool->sp_idlethreads); 96 pool->sp_minthreads = 1; 97 pool->sp_maxthreads = 1; 98 pool->sp_threadcount = 0; 99 100 /* 101 * Don't use more than a quarter of mbuf clusters or more than 102 * 45Mb buffering requests. 103 */ 104 pool->sp_space_high = nmbclusters * MCLBYTES / 4; 105 if (pool->sp_space_high > 45 << 20) 106 pool->sp_space_high = 45 << 20; 107 pool->sp_space_low = 2 * pool->sp_space_high / 3; 108 109 sysctl_ctx_init(&pool->sp_sysctl); 110 if (sysctl_base) { 111 SYSCTL_ADD_PROC(&pool->sp_sysctl, sysctl_base, OID_AUTO, 112 "minthreads", CTLTYPE_INT | CTLFLAG_RW, 113 pool, 0, svcpool_minthread_sysctl, "I", ""); 114 SYSCTL_ADD_PROC(&pool->sp_sysctl, sysctl_base, OID_AUTO, 115 "maxthreads", CTLTYPE_INT | CTLFLAG_RW, 116 pool, 0, svcpool_maxthread_sysctl, "I", ""); 117 SYSCTL_ADD_INT(&pool->sp_sysctl, sysctl_base, OID_AUTO, 118 "threads", CTLFLAG_RD, &pool->sp_threadcount, 0, ""); 119 120 SYSCTL_ADD_UINT(&pool->sp_sysctl, sysctl_base, OID_AUTO, 121 "request_space_used", CTLFLAG_RD, 122 &pool->sp_space_used, 0, 123 "Space in parsed but not handled requests."); 124 125 SYSCTL_ADD_UINT(&pool->sp_sysctl, sysctl_base, OID_AUTO, 126 "request_space_used_highest", CTLFLAG_RD, 127 &pool->sp_space_used_highest, 0, 128 "Highest space used since reboot."); 129 130 SYSCTL_ADD_UINT(&pool->sp_sysctl, sysctl_base, OID_AUTO, 131 "request_space_high", CTLFLAG_RW, 132 &pool->sp_space_high, 0, 133 "Maximum space in parsed but not handled requests."); 134 135 SYSCTL_ADD_UINT(&pool->sp_sysctl, sysctl_base, OID_AUTO, 136 "request_space_low", CTLFLAG_RW, 137 &pool->sp_space_low, 0, 138 "Low water mark for request space."); 139 140 SYSCTL_ADD_INT(&pool->sp_sysctl, sysctl_base, OID_AUTO, 141 "request_space_throttled", CTLFLAG_RD, 142 &pool->sp_space_throttled, 0, 143 "Whether nfs requests are currently throttled"); 144 145 SYSCTL_ADD_INT(&pool->sp_sysctl, sysctl_base, OID_AUTO, 146 "request_space_throttle_count", CTLFLAG_RD, 147 &pool->sp_space_throttle_count, 0, 148 "Count of times throttling based on request space has occurred"); 149 } 150 151 return pool; 152} 153 154void 155svcpool_destroy(SVCPOOL *pool) 156{ 157 SVCXPRT *xprt, *nxprt; 158 struct svc_callout *s; 159 struct svcxprt_list cleanup; 160 161 TAILQ_INIT(&cleanup); 162 mtx_lock(&pool->sp_lock); 163 164 while (TAILQ_FIRST(&pool->sp_xlist)) { 165 xprt = TAILQ_FIRST(&pool->sp_xlist); 166 xprt_unregister_locked(xprt); 167 TAILQ_INSERT_TAIL(&cleanup, xprt, xp_link); 168 } 169 170 while (TAILQ_FIRST(&pool->sp_callouts)) { 171 s = TAILQ_FIRST(&pool->sp_callouts); 172 mtx_unlock(&pool->sp_lock); 173 svc_unreg(pool, s->sc_prog, s->sc_vers); 174 mtx_lock(&pool->sp_lock); 175 } 176 mtx_unlock(&pool->sp_lock); 177 178 TAILQ_FOREACH_SAFE(xprt, &cleanup, xp_link, nxprt) { 179 SVC_RELEASE(xprt); 180 } 181 182 mtx_destroy(&pool->sp_lock); 183 184 if (pool->sp_rcache) 185 replay_freecache(pool->sp_rcache); 186 187 sysctl_ctx_free(&pool->sp_sysctl); 188 free(pool, M_RPC); 189} 190 191static bool_t 192svcpool_active(SVCPOOL *pool) 193{ 194 enum svcpool_state state = pool->sp_state; 195 196 if (state == SVCPOOL_INIT || state == SVCPOOL_CLOSING) 197 return (FALSE); 198 return (TRUE); 199} 200 201/* 202 * Sysctl handler to set the minimum thread count on a pool 203 */ 204static int 205svcpool_minthread_sysctl(SYSCTL_HANDLER_ARGS) 206{ 207 SVCPOOL *pool; 208 int newminthreads, error, n; 209 210 pool = oidp->oid_arg1; 211 newminthreads = pool->sp_minthreads; 212 error = sysctl_handle_int(oidp, &newminthreads, 0, req); 213 if (error == 0 && newminthreads != pool->sp_minthreads) { 214 if (newminthreads > pool->sp_maxthreads) 215 return (EINVAL); 216 mtx_lock(&pool->sp_lock); 217 if (newminthreads > pool->sp_minthreads 218 && svcpool_active(pool)) { 219 /* 220 * If the pool is running and we are 221 * increasing, create some more threads now. 222 */ 223 n = newminthreads - pool->sp_threadcount; 224 if (n > 0) { 225 mtx_unlock(&pool->sp_lock); 226 while (n--) 227 svc_new_thread(pool); 228 mtx_lock(&pool->sp_lock); 229 } 230 } 231 pool->sp_minthreads = newminthreads; 232 mtx_unlock(&pool->sp_lock); 233 } 234 return (error); 235} 236 237/* 238 * Sysctl handler to set the maximum thread count on a pool 239 */ 240static int 241svcpool_maxthread_sysctl(SYSCTL_HANDLER_ARGS) 242{ 243 SVCPOOL *pool; 244 SVCTHREAD *st; 245 int newmaxthreads, error; 246 247 pool = oidp->oid_arg1; 248 newmaxthreads = pool->sp_maxthreads; 249 error = sysctl_handle_int(oidp, &newmaxthreads, 0, req); 250 if (error == 0 && newmaxthreads != pool->sp_maxthreads) { 251 if (newmaxthreads < pool->sp_minthreads) 252 return (EINVAL); 253 mtx_lock(&pool->sp_lock); 254 if (newmaxthreads < pool->sp_maxthreads 255 && svcpool_active(pool)) { 256 /* 257 * If the pool is running and we are 258 * decreasing, wake up some idle threads to 259 * encourage them to exit. 260 */ 261 LIST_FOREACH(st, &pool->sp_idlethreads, st_ilink) 262 cv_signal(&st->st_cond); 263 } 264 pool->sp_maxthreads = newmaxthreads; 265 mtx_unlock(&pool->sp_lock); 266 } 267 return (error); 268} 269 270/* 271 * Activate a transport handle. 272 */ 273void 274xprt_register(SVCXPRT *xprt) 275{ 276 SVCPOOL *pool = xprt->xp_pool; 277 278 SVC_ACQUIRE(xprt); 279 mtx_lock(&pool->sp_lock); 280 xprt->xp_registered = TRUE; 281 xprt->xp_active = FALSE; 282 TAILQ_INSERT_TAIL(&pool->sp_xlist, xprt, xp_link); 283 mtx_unlock(&pool->sp_lock); 284} 285 286/* 287 * De-activate a transport handle. Note: the locked version doesn't 288 * release the transport - caller must do that after dropping the pool 289 * lock. 290 */ 291static void 292xprt_unregister_locked(SVCXPRT *xprt) 293{ 294 SVCPOOL *pool = xprt->xp_pool; 295 296 mtx_assert(&pool->sp_lock, MA_OWNED); 297 KASSERT(xprt->xp_registered == TRUE, 298 ("xprt_unregister_locked: not registered")); 299 xprt_inactive_locked(xprt); 300 TAILQ_REMOVE(&pool->sp_xlist, xprt, xp_link); 301 xprt->xp_registered = FALSE; 302} 303 304void 305xprt_unregister(SVCXPRT *xprt) 306{ 307 SVCPOOL *pool = xprt->xp_pool; 308 309 mtx_lock(&pool->sp_lock); 310 if (xprt->xp_registered == FALSE) { 311 /* Already unregistered by another thread */ 312 mtx_unlock(&pool->sp_lock); 313 return; 314 } 315 xprt_unregister_locked(xprt); 316 mtx_unlock(&pool->sp_lock); 317 318 SVC_RELEASE(xprt); 319} 320 321/* 322 * Attempt to assign a service thread to this transport. 323 */ 324static int 325xprt_assignthread(SVCXPRT *xprt) 326{ 327 SVCPOOL *pool = xprt->xp_pool; 328 SVCTHREAD *st; 329 330 mtx_assert(&pool->sp_lock, MA_OWNED); 331 st = LIST_FIRST(&pool->sp_idlethreads); 332 if (st) { 333 LIST_REMOVE(st, st_ilink); 334 st->st_idle = FALSE; 335 SVC_ACQUIRE(xprt); 336 xprt->xp_thread = st; 337 st->st_xprt = xprt; 338 cv_signal(&st->st_cond); 339 return (TRUE); 340 } else { 341 /* 342 * See if we can create a new thread. The 343 * actual thread creation happens in 344 * svc_run_internal because our locking state 345 * is poorly defined (we are typically called 346 * from a socket upcall). Don't create more 347 * than one thread per second. 348 */ 349 if (pool->sp_state == SVCPOOL_ACTIVE 350 && pool->sp_lastcreatetime < time_uptime 351 && pool->sp_threadcount < pool->sp_maxthreads) { 352 pool->sp_state = SVCPOOL_THREADWANTED; 353 } 354 } 355 return (FALSE); 356} 357 358void 359xprt_active(SVCXPRT *xprt) 360{ 361 SVCPOOL *pool = xprt->xp_pool; 362 363 mtx_lock(&pool->sp_lock); 364 365 if (!xprt->xp_registered) { 366 /* 367 * Race with xprt_unregister - we lose. 368 */ 369 mtx_unlock(&pool->sp_lock); 370 return; 371 } 372 373 if (!xprt->xp_active) { 374 xprt->xp_active = TRUE; 375 if (xprt->xp_thread == NULL) { 376 if (!xprt_assignthread(xprt)) 377 TAILQ_INSERT_TAIL(&pool->sp_active, xprt, 378 xp_alink); 379 } 380 } 381 382 mtx_unlock(&pool->sp_lock); 383} 384 385void 386xprt_inactive_locked(SVCXPRT *xprt) 387{ 388 SVCPOOL *pool = xprt->xp_pool; 389 390 mtx_assert(&pool->sp_lock, MA_OWNED); 391 if (xprt->xp_active) { 392 if (xprt->xp_thread == NULL) 393 TAILQ_REMOVE(&pool->sp_active, xprt, xp_alink); 394 xprt->xp_active = FALSE; 395 } 396} 397 398void 399xprt_inactive(SVCXPRT *xprt) 400{ 401 SVCPOOL *pool = xprt->xp_pool; 402 403 mtx_lock(&pool->sp_lock); 404 xprt_inactive_locked(xprt); 405 mtx_unlock(&pool->sp_lock); 406} 407 408/* 409 * Variant of xprt_inactive() for use only when sure that port is 410 * assigned to thread. For example, withing receive handlers. 411 */ 412void 413xprt_inactive_self(SVCXPRT *xprt) 414{ 415 416 KASSERT(xprt->xp_thread != NULL, 417 ("xprt_inactive_self(%p) with NULL xp_thread", xprt)); 418 xprt->xp_active = FALSE; 419} 420 421/* 422 * Add a service program to the callout list. 423 * The dispatch routine will be called when a rpc request for this 424 * program number comes in. 425 */ 426bool_t 427svc_reg(SVCXPRT *xprt, const rpcprog_t prog, const rpcvers_t vers, 428 void (*dispatch)(struct svc_req *, SVCXPRT *), 429 const struct netconfig *nconf) 430{ 431 SVCPOOL *pool = xprt->xp_pool; 432 struct svc_callout *s; 433 char *netid = NULL; 434 int flag = 0; 435 436/* VARIABLES PROTECTED BY svc_lock: s, svc_head */ 437 438 if (xprt->xp_netid) { 439 netid = strdup(xprt->xp_netid, M_RPC); 440 flag = 1; 441 } else if (nconf && nconf->nc_netid) { 442 netid = strdup(nconf->nc_netid, M_RPC); 443 flag = 1; 444 } /* must have been created with svc_raw_create */ 445 if ((netid == NULL) && (flag == 1)) { 446 return (FALSE); 447 } 448 449 mtx_lock(&pool->sp_lock); 450 if ((s = svc_find(pool, prog, vers, netid)) != NULL) { 451 if (netid) 452 free(netid, M_RPC); 453 if (s->sc_dispatch == dispatch) 454 goto rpcb_it; /* he is registering another xptr */ 455 mtx_unlock(&pool->sp_lock); 456 return (FALSE); 457 } 458 s = malloc(sizeof (struct svc_callout), M_RPC, M_NOWAIT); 459 if (s == NULL) { 460 if (netid) 461 free(netid, M_RPC); 462 mtx_unlock(&pool->sp_lock); 463 return (FALSE); 464 } 465 466 s->sc_prog = prog; 467 s->sc_vers = vers; 468 s->sc_dispatch = dispatch; 469 s->sc_netid = netid; 470 TAILQ_INSERT_TAIL(&pool->sp_callouts, s, sc_link); 471 472 if ((xprt->xp_netid == NULL) && (flag == 1) && netid) 473 ((SVCXPRT *) xprt)->xp_netid = strdup(netid, M_RPC); 474 475rpcb_it: 476 mtx_unlock(&pool->sp_lock); 477 /* now register the information with the local binder service */ 478 if (nconf) { 479 bool_t dummy; 480 struct netconfig tnc; 481 struct netbuf nb; 482 tnc = *nconf; 483 nb.buf = &xprt->xp_ltaddr; 484 nb.len = xprt->xp_ltaddr.ss_len; 485 dummy = rpcb_set(prog, vers, &tnc, &nb); 486 return (dummy); 487 } 488 return (TRUE); 489} 490 491/* 492 * Remove a service program from the callout list. 493 */ 494void 495svc_unreg(SVCPOOL *pool, const rpcprog_t prog, const rpcvers_t vers) 496{ 497 struct svc_callout *s; 498 499 /* unregister the information anyway */ 500 (void) rpcb_unset(prog, vers, NULL); 501 mtx_lock(&pool->sp_lock); 502 while ((s = svc_find(pool, prog, vers, NULL)) != NULL) { 503 TAILQ_REMOVE(&pool->sp_callouts, s, sc_link); 504 if (s->sc_netid) 505 mem_free(s->sc_netid, sizeof (s->sc_netid) + 1); 506 mem_free(s, sizeof (struct svc_callout)); 507 } 508 mtx_unlock(&pool->sp_lock); 509} 510 511/* ********************** CALLOUT list related stuff ************* */ 512 513/* 514 * Search the callout list for a program number, return the callout 515 * struct. 516 */ 517static struct svc_callout * 518svc_find(SVCPOOL *pool, rpcprog_t prog, rpcvers_t vers, char *netid) 519{ 520 struct svc_callout *s; 521 522 mtx_assert(&pool->sp_lock, MA_OWNED); 523 TAILQ_FOREACH(s, &pool->sp_callouts, sc_link) { 524 if (s->sc_prog == prog && s->sc_vers == vers 525 && (netid == NULL || s->sc_netid == NULL || 526 strcmp(netid, s->sc_netid) == 0)) 527 break; 528 } 529 530 return (s); 531} 532 533/* ******************* REPLY GENERATION ROUTINES ************ */ 534 535static bool_t 536svc_sendreply_common(struct svc_req *rqstp, struct rpc_msg *rply, 537 struct mbuf *body) 538{ 539 SVCXPRT *xprt = rqstp->rq_xprt; 540 bool_t ok; 541 542 if (rqstp->rq_args) { 543 m_freem(rqstp->rq_args); 544 rqstp->rq_args = NULL; 545 } 546 547 if (xprt->xp_pool->sp_rcache) 548 replay_setreply(xprt->xp_pool->sp_rcache, 549 rply, svc_getrpccaller(rqstp), body); 550 551 if (!SVCAUTH_WRAP(&rqstp->rq_auth, &body)) 552 return (FALSE); 553 554 ok = SVC_REPLY(xprt, rply, rqstp->rq_addr, body); 555 if (rqstp->rq_addr) { 556 free(rqstp->rq_addr, M_SONAME); 557 rqstp->rq_addr = NULL; 558 } 559 560 return (ok); 561} 562 563/* 564 * Send a reply to an rpc request 565 */ 566bool_t 567svc_sendreply(struct svc_req *rqstp, xdrproc_t xdr_results, void * xdr_location) 568{ 569 struct rpc_msg rply; 570 struct mbuf *m; 571 XDR xdrs; 572 bool_t ok; 573 574 rply.rm_xid = rqstp->rq_xid; 575 rply.rm_direction = REPLY; 576 rply.rm_reply.rp_stat = MSG_ACCEPTED; 577 rply.acpted_rply.ar_verf = rqstp->rq_verf; 578 rply.acpted_rply.ar_stat = SUCCESS; 579 rply.acpted_rply.ar_results.where = NULL; 580 rply.acpted_rply.ar_results.proc = (xdrproc_t) xdr_void; 581 582 m = m_getcl(M_WAITOK, MT_DATA, 0); 583 xdrmbuf_create(&xdrs, m, XDR_ENCODE); 584 ok = xdr_results(&xdrs, xdr_location); 585 XDR_DESTROY(&xdrs); 586 587 if (ok) { 588 return (svc_sendreply_common(rqstp, &rply, m)); 589 } else { 590 m_freem(m); 591 return (FALSE); 592 } 593} 594 595bool_t 596svc_sendreply_mbuf(struct svc_req *rqstp, struct mbuf *m) 597{ 598 struct rpc_msg rply; 599 600 rply.rm_xid = rqstp->rq_xid; 601 rply.rm_direction = REPLY; 602 rply.rm_reply.rp_stat = MSG_ACCEPTED; 603 rply.acpted_rply.ar_verf = rqstp->rq_verf; 604 rply.acpted_rply.ar_stat = SUCCESS; 605 rply.acpted_rply.ar_results.where = NULL; 606 rply.acpted_rply.ar_results.proc = (xdrproc_t) xdr_void; 607 608 return (svc_sendreply_common(rqstp, &rply, m)); 609} 610 611/* 612 * No procedure error reply 613 */ 614void 615svcerr_noproc(struct svc_req *rqstp) 616{ 617 SVCXPRT *xprt = rqstp->rq_xprt; 618 struct rpc_msg rply; 619 620 rply.rm_xid = rqstp->rq_xid; 621 rply.rm_direction = REPLY; 622 rply.rm_reply.rp_stat = MSG_ACCEPTED; 623 rply.acpted_rply.ar_verf = rqstp->rq_verf; 624 rply.acpted_rply.ar_stat = PROC_UNAVAIL; 625 626 if (xprt->xp_pool->sp_rcache) 627 replay_setreply(xprt->xp_pool->sp_rcache, 628 &rply, svc_getrpccaller(rqstp), NULL); 629 630 svc_sendreply_common(rqstp, &rply, NULL); 631} 632 633/* 634 * Can't decode args error reply 635 */ 636void 637svcerr_decode(struct svc_req *rqstp) 638{ 639 SVCXPRT *xprt = rqstp->rq_xprt; 640 struct rpc_msg rply; 641 642 rply.rm_xid = rqstp->rq_xid; 643 rply.rm_direction = REPLY; 644 rply.rm_reply.rp_stat = MSG_ACCEPTED; 645 rply.acpted_rply.ar_verf = rqstp->rq_verf; 646 rply.acpted_rply.ar_stat = GARBAGE_ARGS; 647 648 if (xprt->xp_pool->sp_rcache) 649 replay_setreply(xprt->xp_pool->sp_rcache, 650 &rply, (struct sockaddr *) &xprt->xp_rtaddr, NULL); 651 652 svc_sendreply_common(rqstp, &rply, NULL); 653} 654 655/* 656 * Some system error 657 */ 658void 659svcerr_systemerr(struct svc_req *rqstp) 660{ 661 SVCXPRT *xprt = rqstp->rq_xprt; 662 struct rpc_msg rply; 663 664 rply.rm_xid = rqstp->rq_xid; 665 rply.rm_direction = REPLY; 666 rply.rm_reply.rp_stat = MSG_ACCEPTED; 667 rply.acpted_rply.ar_verf = rqstp->rq_verf; 668 rply.acpted_rply.ar_stat = SYSTEM_ERR; 669 670 if (xprt->xp_pool->sp_rcache) 671 replay_setreply(xprt->xp_pool->sp_rcache, 672 &rply, svc_getrpccaller(rqstp), NULL); 673 674 svc_sendreply_common(rqstp, &rply, NULL); 675} 676 677/* 678 * Authentication error reply 679 */ 680void 681svcerr_auth(struct svc_req *rqstp, enum auth_stat why) 682{ 683 SVCXPRT *xprt = rqstp->rq_xprt; 684 struct rpc_msg rply; 685 686 rply.rm_xid = rqstp->rq_xid; 687 rply.rm_direction = REPLY; 688 rply.rm_reply.rp_stat = MSG_DENIED; 689 rply.rjcted_rply.rj_stat = AUTH_ERROR; 690 rply.rjcted_rply.rj_why = why; 691 692 if (xprt->xp_pool->sp_rcache) 693 replay_setreply(xprt->xp_pool->sp_rcache, 694 &rply, svc_getrpccaller(rqstp), NULL); 695 696 svc_sendreply_common(rqstp, &rply, NULL); 697} 698 699/* 700 * Auth too weak error reply 701 */ 702void 703svcerr_weakauth(struct svc_req *rqstp) 704{ 705 706 svcerr_auth(rqstp, AUTH_TOOWEAK); 707} 708 709/* 710 * Program unavailable error reply 711 */ 712void 713svcerr_noprog(struct svc_req *rqstp) 714{ 715 SVCXPRT *xprt = rqstp->rq_xprt; 716 struct rpc_msg rply; 717 718 rply.rm_xid = rqstp->rq_xid; 719 rply.rm_direction = REPLY; 720 rply.rm_reply.rp_stat = MSG_ACCEPTED; 721 rply.acpted_rply.ar_verf = rqstp->rq_verf; 722 rply.acpted_rply.ar_stat = PROG_UNAVAIL; 723 724 if (xprt->xp_pool->sp_rcache) 725 replay_setreply(xprt->xp_pool->sp_rcache, 726 &rply, svc_getrpccaller(rqstp), NULL); 727 728 svc_sendreply_common(rqstp, &rply, NULL); 729} 730 731/* 732 * Program version mismatch error reply 733 */ 734void 735svcerr_progvers(struct svc_req *rqstp, rpcvers_t low_vers, rpcvers_t high_vers) 736{ 737 SVCXPRT *xprt = rqstp->rq_xprt; 738 struct rpc_msg rply; 739 740 rply.rm_xid = rqstp->rq_xid; 741 rply.rm_direction = REPLY; 742 rply.rm_reply.rp_stat = MSG_ACCEPTED; 743 rply.acpted_rply.ar_verf = rqstp->rq_verf; 744 rply.acpted_rply.ar_stat = PROG_MISMATCH; 745 rply.acpted_rply.ar_vers.low = (uint32_t)low_vers; 746 rply.acpted_rply.ar_vers.high = (uint32_t)high_vers; 747 748 if (xprt->xp_pool->sp_rcache) 749 replay_setreply(xprt->xp_pool->sp_rcache, 750 &rply, svc_getrpccaller(rqstp), NULL); 751 752 svc_sendreply_common(rqstp, &rply, NULL); 753} 754 755/* 756 * Allocate a new server transport structure. All fields are 757 * initialized to zero and xp_p3 is initialized to point at an 758 * extension structure to hold various flags and authentication 759 * parameters. 760 */ 761SVCXPRT * 762svc_xprt_alloc() 763{ 764 SVCXPRT *xprt; 765 SVCXPRT_EXT *ext; 766 767 xprt = mem_alloc(sizeof(SVCXPRT)); 768 memset(xprt, 0, sizeof(SVCXPRT)); 769 ext = mem_alloc(sizeof(SVCXPRT_EXT)); 770 memset(ext, 0, sizeof(SVCXPRT_EXT)); 771 xprt->xp_p3 = ext; 772 refcount_init(&xprt->xp_refs, 1); 773 774 return (xprt); 775} 776 777/* 778 * Free a server transport structure. 779 */ 780void 781svc_xprt_free(xprt) 782 SVCXPRT *xprt; 783{ 784 785 mem_free(xprt->xp_p3, sizeof(SVCXPRT_EXT)); 786 mem_free(xprt, sizeof(SVCXPRT)); 787} 788 789/* ******************* SERVER INPUT STUFF ******************* */ 790 791/* 792 * Read RPC requests from a transport and queue them to be 793 * executed. We handle authentication and replay cache replies here. 794 * Actually dispatching the RPC is deferred till svc_executereq. 795 */ 796static enum xprt_stat 797svc_getreq(SVCXPRT *xprt, struct svc_req **rqstp_ret) 798{ 799 SVCPOOL *pool = xprt->xp_pool; 800 struct svc_req *r; 801 struct rpc_msg msg; 802 struct mbuf *args; 803 enum xprt_stat stat; 804 805 /* now receive msgs from xprtprt (support batch calls) */ 806 r = malloc(sizeof(*r), M_RPC, M_WAITOK|M_ZERO); 807 808 msg.rm_call.cb_cred.oa_base = r->rq_credarea; 809 msg.rm_call.cb_verf.oa_base = &r->rq_credarea[MAX_AUTH_BYTES]; 810 r->rq_clntcred = &r->rq_credarea[2*MAX_AUTH_BYTES]; 811 if (SVC_RECV(xprt, &msg, &r->rq_addr, &args)) { 812 enum auth_stat why; 813 814 /* 815 * Handle replays and authenticate before queuing the 816 * request to be executed. 817 */ 818 SVC_ACQUIRE(xprt); 819 r->rq_xprt = xprt; 820 if (pool->sp_rcache) { 821 struct rpc_msg repmsg; 822 struct mbuf *repbody; 823 enum replay_state rs; 824 rs = replay_find(pool->sp_rcache, &msg, 825 svc_getrpccaller(r), &repmsg, &repbody); 826 switch (rs) { 827 case RS_NEW: 828 break; 829 case RS_DONE: 830 SVC_REPLY(xprt, &repmsg, r->rq_addr, 831 repbody); 832 if (r->rq_addr) { 833 free(r->rq_addr, M_SONAME); 834 r->rq_addr = NULL; 835 } 836 m_freem(args); 837 goto call_done; 838 839 default: 840 m_freem(args); 841 goto call_done; 842 } 843 } 844 845 r->rq_xid = msg.rm_xid; 846 r->rq_prog = msg.rm_call.cb_prog; 847 r->rq_vers = msg.rm_call.cb_vers; 848 r->rq_proc = msg.rm_call.cb_proc; 849 r->rq_size = sizeof(*r) + m_length(args, NULL); 850 r->rq_args = args; 851 if ((why = _authenticate(r, &msg)) != AUTH_OK) { 852 /* 853 * RPCSEC_GSS uses this return code 854 * for requests that form part of its 855 * context establishment protocol and 856 * should not be dispatched to the 857 * application. 858 */ 859 if (why != RPCSEC_GSS_NODISPATCH) 860 svcerr_auth(r, why); 861 goto call_done; 862 } 863 864 if (!SVCAUTH_UNWRAP(&r->rq_auth, &r->rq_args)) { 865 svcerr_decode(r); 866 goto call_done; 867 } 868 869 /* 870 * Everything checks out, return request to caller. 871 */ 872 *rqstp_ret = r; 873 r = NULL; 874 } 875call_done: 876 if (r) { 877 svc_freereq(r); 878 r = NULL; 879 } 880 if ((stat = SVC_STAT(xprt)) == XPRT_DIED) { 881 xprt_unregister(xprt); 882 } 883 884 return (stat); 885} 886 887static void 888svc_executereq(struct svc_req *rqstp) 889{ 890 SVCXPRT *xprt = rqstp->rq_xprt; 891 SVCPOOL *pool = xprt->xp_pool; 892 int prog_found; 893 rpcvers_t low_vers; 894 rpcvers_t high_vers; 895 struct svc_callout *s; 896 897 /* now match message with a registered service*/ 898 prog_found = FALSE; 899 low_vers = (rpcvers_t) -1L; 900 high_vers = (rpcvers_t) 0L; 901 TAILQ_FOREACH(s, &pool->sp_callouts, sc_link) { 902 if (s->sc_prog == rqstp->rq_prog) { 903 if (s->sc_vers == rqstp->rq_vers) { 904 /* 905 * We hand ownership of r to the 906 * dispatch method - they must call 907 * svc_freereq. 908 */ 909 (*s->sc_dispatch)(rqstp, xprt); 910 return; 911 } /* found correct version */ 912 prog_found = TRUE; 913 if (s->sc_vers < low_vers) 914 low_vers = s->sc_vers; 915 if (s->sc_vers > high_vers) 916 high_vers = s->sc_vers; 917 } /* found correct program */ 918 } 919 920 /* 921 * if we got here, the program or version 922 * is not served ... 923 */ 924 if (prog_found) 925 svcerr_progvers(rqstp, low_vers, high_vers); 926 else 927 svcerr_noprog(rqstp); 928 929 svc_freereq(rqstp); 930} 931 932static void 933svc_checkidle(SVCPOOL *pool) 934{ 935 SVCXPRT *xprt, *nxprt; 936 time_t timo; 937 struct svcxprt_list cleanup; 938 939 TAILQ_INIT(&cleanup); 940 TAILQ_FOREACH_SAFE(xprt, &pool->sp_xlist, xp_link, nxprt) { 941 /* 942 * Only some transports have idle timers. Don't time 943 * something out which is just waking up. 944 */ 945 if (!xprt->xp_idletimeout || xprt->xp_thread) 946 continue; 947 948 timo = xprt->xp_lastactive + xprt->xp_idletimeout; 949 if (time_uptime > timo) { 950 xprt_unregister_locked(xprt); 951 TAILQ_INSERT_TAIL(&cleanup, xprt, xp_link); 952 } 953 } 954 955 mtx_unlock(&pool->sp_lock); 956 TAILQ_FOREACH_SAFE(xprt, &cleanup, xp_link, nxprt) { 957 SVC_RELEASE(xprt); 958 } 959 mtx_lock(&pool->sp_lock); 960 961} 962 963static void 964svc_assign_waiting_sockets(SVCPOOL *pool) 965{ 966 SVCXPRT *xprt; 967 968 while ((xprt = TAILQ_FIRST(&pool->sp_active)) != NULL) { 969 if (xprt_assignthread(xprt)) 970 TAILQ_REMOVE(&pool->sp_active, xprt, xp_alink); 971 else 972 break; 973 } 974} 975 976static bool_t 977svc_request_space_available(SVCPOOL *pool) 978{ 979 980 mtx_assert(&pool->sp_lock, MA_OWNED); 981 982 if (pool->sp_space_throttled) { 983 /* 984 * Below the low-water yet? If so, assign any waiting sockets. 985 */ 986 if (pool->sp_space_used < pool->sp_space_low) { 987 pool->sp_space_throttled = FALSE; 988 svc_assign_waiting_sockets(pool); 989 return TRUE; 990 } 991 992 return FALSE; 993 } else { 994 if (pool->sp_space_used 995 >= pool->sp_space_high) { 996 pool->sp_space_throttled = TRUE; 997 pool->sp_space_throttle_count++; 998 return FALSE; 999 } 1000 1001 return TRUE; 1002 } 1003} 1004 1005static void 1006svc_run_internal(SVCPOOL *pool, bool_t ismaster) 1007{ 1008 SVCTHREAD *st, *stpref; 1009 SVCXPRT *xprt; 1010 enum xprt_stat stat; 1011 struct svc_req *rqstp; 1012 int error; 1013 1014 st = mem_alloc(sizeof(*st)); 1015 st->st_xprt = NULL; 1016 STAILQ_INIT(&st->st_reqs); 1017 cv_init(&st->st_cond, "rpcsvc"); 1018 1019 mtx_lock(&pool->sp_lock); 1020 LIST_INSERT_HEAD(&pool->sp_threads, st, st_link); 1021 1022 /* 1023 * If we are a new thread which was spawned to cope with 1024 * increased load, set the state back to SVCPOOL_ACTIVE. 1025 */ 1026 if (pool->sp_state == SVCPOOL_THREADSTARTING) 1027 pool->sp_state = SVCPOOL_ACTIVE; 1028 1029 while (pool->sp_state != SVCPOOL_CLOSING) { 1030 /* 1031 * Create new thread if requested. 1032 */ 1033 if (pool->sp_state == SVCPOOL_THREADWANTED) { 1034 pool->sp_state = SVCPOOL_THREADSTARTING; 1035 pool->sp_lastcreatetime = time_uptime; 1036 mtx_unlock(&pool->sp_lock); 1037 svc_new_thread(pool); 1038 mtx_lock(&pool->sp_lock); 1039 continue; 1040 } 1041 1042 /* 1043 * Check for idle transports once per second. 1044 */ 1045 if (time_uptime > pool->sp_lastidlecheck) { 1046 pool->sp_lastidlecheck = time_uptime; 1047 svc_checkidle(pool); 1048 } 1049 1050 xprt = st->st_xprt; 1051 if (!xprt && STAILQ_EMPTY(&st->st_reqs)) { 1052 /* 1053 * Enforce maxthreads count. 1054 */ 1055 if (pool->sp_threadcount > pool->sp_maxthreads) 1056 break; 1057 1058 /* 1059 * Before sleeping, see if we can find an 1060 * active transport which isn't being serviced 1061 * by a thread. 1062 */ 1063 if (svc_request_space_available(pool) && 1064 (xprt = TAILQ_FIRST(&pool->sp_active)) != NULL) { 1065 TAILQ_REMOVE(&pool->sp_active, xprt, xp_alink); 1066 SVC_ACQUIRE(xprt); 1067 xprt->xp_thread = st; 1068 st->st_xprt = xprt; 1069 continue; 1070 } 1071 1072 LIST_INSERT_HEAD(&pool->sp_idlethreads, st, st_ilink); 1073 st->st_idle = TRUE; 1074 if (ismaster || (!ismaster && 1075 pool->sp_threadcount > pool->sp_minthreads)) 1076 error = cv_timedwait_sig(&st->st_cond, 1077 &pool->sp_lock, 5 * hz); 1078 else 1079 error = cv_wait_sig(&st->st_cond, 1080 &pool->sp_lock); 1081 if (st->st_idle) { 1082 LIST_REMOVE(st, st_ilink); 1083 st->st_idle = FALSE; 1084 } 1085 1086 /* 1087 * Reduce worker thread count when idle. 1088 */ 1089 if (error == EWOULDBLOCK) { 1090 if (!ismaster 1091 && (pool->sp_threadcount 1092 > pool->sp_minthreads) 1093 && !st->st_xprt 1094 && STAILQ_EMPTY(&st->st_reqs)) 1095 break; 1096 } else if (error) { 1097 mtx_unlock(&pool->sp_lock); 1098 svc_exit(pool); 1099 mtx_lock(&pool->sp_lock); 1100 break; 1101 } 1102 continue; 1103 } 1104 1105 if (xprt) { 1106 /* 1107 * Drain the transport socket and queue up any 1108 * RPCs. 1109 */ 1110 xprt->xp_lastactive = time_uptime; 1111 stat = XPRT_IDLE; 1112 do { 1113 if (!svc_request_space_available(pool)) 1114 break; 1115 rqstp = NULL; 1116 mtx_unlock(&pool->sp_lock); 1117 stat = svc_getreq(xprt, &rqstp); 1118 mtx_lock(&pool->sp_lock); 1119 if (rqstp) { 1120 /* 1121 * See if the application has 1122 * a preference for some other 1123 * thread. 1124 */ 1125 stpref = st; 1126 if (pool->sp_assign) 1127 stpref = pool->sp_assign(st, 1128 rqstp); 1129 1130 pool->sp_space_used += 1131 rqstp->rq_size; 1132 if (pool->sp_space_used 1133 > pool->sp_space_used_highest) 1134 pool->sp_space_used_highest = 1135 pool->sp_space_used; 1136 rqstp->rq_thread = stpref; 1137 STAILQ_INSERT_TAIL(&stpref->st_reqs, 1138 rqstp, rq_link); 1139 stpref->st_reqcount++; 1140 1141 /* 1142 * If we assigned the request 1143 * to another thread, make 1144 * sure its awake and continue 1145 * reading from the 1146 * socket. Otherwise, try to 1147 * find some other thread to 1148 * read from the socket and 1149 * execute the request 1150 * immediately. 1151 */ 1152 if (stpref == st) 1153 break; 1154 if (stpref->st_idle) { 1155 LIST_REMOVE(stpref, st_ilink); 1156 stpref->st_idle = FALSE; 1157 cv_signal(&stpref->st_cond); 1158 } 1159 } 1160 } while (stat == XPRT_MOREREQS 1161 && pool->sp_state != SVCPOOL_CLOSING); 1162 1163 /* 1164 * Move this transport to the end of the 1165 * active list to ensure fairness when 1166 * multiple transports are active. If this was 1167 * the last queued request, svc_getreq will 1168 * end up calling xprt_inactive to remove from 1169 * the active list. 1170 */ 1171 xprt->xp_thread = NULL; 1172 st->st_xprt = NULL; 1173 if (xprt->xp_active) { 1174 if (!xprt_assignthread(xprt)) 1175 TAILQ_INSERT_TAIL(&pool->sp_active, 1176 xprt, xp_alink); 1177 } 1178 mtx_unlock(&pool->sp_lock); 1179 SVC_RELEASE(xprt); 1180 mtx_lock(&pool->sp_lock); 1181 } 1182 1183 /* 1184 * Execute what we have queued. 1185 */ 1186 while ((rqstp = STAILQ_FIRST(&st->st_reqs)) != NULL) { 1187 size_t sz = rqstp->rq_size; 1188 mtx_unlock(&pool->sp_lock); 1189 svc_executereq(rqstp); 1190 mtx_lock(&pool->sp_lock); 1191 pool->sp_space_used -= sz; 1192 } 1193 } 1194 1195 if (st->st_xprt) { 1196 xprt = st->st_xprt; 1197 st->st_xprt = NULL; 1198 SVC_RELEASE(xprt); 1199 } 1200 1201 KASSERT(STAILQ_EMPTY(&st->st_reqs), ("stray reqs on exit")); 1202 LIST_REMOVE(st, st_link); 1203 pool->sp_threadcount--; 1204 1205 mtx_unlock(&pool->sp_lock); 1206 1207 cv_destroy(&st->st_cond); 1208 mem_free(st, sizeof(*st)); 1209 1210 if (!ismaster) 1211 wakeup(pool); 1212} 1213 1214static void 1215svc_thread_start(void *arg) 1216{ 1217 1218 svc_run_internal((SVCPOOL *) arg, FALSE); 1219 kthread_exit(); 1220} 1221 1222static void 1223svc_new_thread(SVCPOOL *pool) 1224{ 1225 struct thread *td; 1226 1227 pool->sp_threadcount++; 1228 kthread_add(svc_thread_start, pool, 1229 pool->sp_proc, &td, 0, 0, 1230 "%s: service", pool->sp_name); 1231} 1232 1233void 1234svc_run(SVCPOOL *pool) 1235{ 1236 int i; 1237 struct proc *p; 1238 struct thread *td; 1239 1240 p = curproc; 1241 td = curthread; 1242 snprintf(td->td_name, sizeof(td->td_name), 1243 "%s: master", pool->sp_name); 1244 pool->sp_state = SVCPOOL_ACTIVE; 1245 pool->sp_proc = p; 1246 pool->sp_lastcreatetime = time_uptime; 1247 pool->sp_threadcount = 1; 1248 1249 for (i = 1; i < pool->sp_minthreads; i++) { 1250 svc_new_thread(pool); 1251 } 1252 1253 svc_run_internal(pool, TRUE); 1254 1255 mtx_lock(&pool->sp_lock); 1256 while (pool->sp_threadcount > 0) 1257 msleep(pool, &pool->sp_lock, 0, "svcexit", 0); 1258 mtx_unlock(&pool->sp_lock); 1259} 1260 1261void 1262svc_exit(SVCPOOL *pool) 1263{ 1264 SVCTHREAD *st; 1265 1266 mtx_lock(&pool->sp_lock); 1267 1268 if (pool->sp_state != SVCPOOL_CLOSING) { 1269 pool->sp_state = SVCPOOL_CLOSING; 1270 LIST_FOREACH(st, &pool->sp_idlethreads, st_ilink) 1271 cv_signal(&st->st_cond); 1272 } 1273 1274 mtx_unlock(&pool->sp_lock); 1275} 1276 1277bool_t 1278svc_getargs(struct svc_req *rqstp, xdrproc_t xargs, void *args) 1279{ 1280 struct mbuf *m; 1281 XDR xdrs; 1282 bool_t stat; 1283 1284 m = rqstp->rq_args; 1285 rqstp->rq_args = NULL; 1286 1287 xdrmbuf_create(&xdrs, m, XDR_DECODE); 1288 stat = xargs(&xdrs, args); 1289 XDR_DESTROY(&xdrs); 1290 1291 return (stat); 1292} 1293 1294bool_t 1295svc_freeargs(struct svc_req *rqstp, xdrproc_t xargs, void *args) 1296{ 1297 XDR xdrs; 1298 1299 if (rqstp->rq_addr) { 1300 free(rqstp->rq_addr, M_SONAME); 1301 rqstp->rq_addr = NULL; 1302 } 1303 1304 xdrs.x_op = XDR_FREE; 1305 return (xargs(&xdrs, args)); 1306} 1307 1308void 1309svc_freereq(struct svc_req *rqstp) 1310{ 1311 SVCTHREAD *st; 1312 SVCXPRT *xprt; 1313 SVCPOOL *pool; 1314 1315 st = rqstp->rq_thread; 1316 xprt = rqstp->rq_xprt; 1317 if (xprt) 1318 pool = xprt->xp_pool; 1319 else 1320 pool = NULL; 1321 if (st) { 1322 mtx_lock(&pool->sp_lock); 1323 KASSERT(rqstp == STAILQ_FIRST(&st->st_reqs), 1324 ("Freeing request out of order")); 1325 STAILQ_REMOVE_HEAD(&st->st_reqs, rq_link); 1326 st->st_reqcount--; 1327 if (pool->sp_done) 1328 pool->sp_done(st, rqstp); 1329 mtx_unlock(&pool->sp_lock); 1330 } 1331 1332 if (rqstp->rq_auth.svc_ah_ops) 1333 SVCAUTH_RELEASE(&rqstp->rq_auth); 1334 1335 if (rqstp->rq_xprt) { 1336 SVC_RELEASE(rqstp->rq_xprt); 1337 } 1338 1339 if (rqstp->rq_addr) 1340 free(rqstp->rq_addr, M_SONAME); 1341 1342 if (rqstp->rq_args) 1343 m_freem(rqstp->rq_args); 1344 1345 free(rqstp, M_RPC); 1346} 1347