sys/rpc/svc.c

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