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

Cache object: d6f6b1816b2208f102731bd03e0dcdd4


[ source navigation ] [ diff markup ] [ identifier search ] [ freetext search ] [ file search ] [ list types ] [ track identifier ]


This page is part of the FreeBSD/Linux Linux Kernel Cross-Reference, and was automatically generated using a modified version of the LXR engine.