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