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