16683 pthread_cond_timedwait broken when using static initializer

[illumos-gate.git] / usr / src / uts / common / rpc / svc.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */

/*
 * Copyright 2012 Marcel Telka <marcel@telka.sk>
 * Copyright 2015 Nexenta Systems, Inc.  All rights reserved.
 * Copyright 2018 OmniOS Community Edition (OmniOSce) Association.
 * Copyright 2021 Racktop Systems, Inc.
 */

/*
 * Copyright 2010 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

/*
 * Copyright 1993 OpenVision Technologies, Inc., All Rights Reserved.
 */

/*      Copyright (c) 1983, 1984, 1985,  1986, 1987, 1988, 1989 AT&T    */
/*      All Rights Reserved     */

/*
 * Portions of this source code were derived from Berkeley 4.3 BSD
 * under license from the Regents of the University of California.
 */

/*
 * Server-side remote procedure call interface.
 *
 * Master transport handle (SVCMASTERXPRT).
 *   The master transport handle structure is shared among service
 *   threads processing events on the transport. Some fields in the
 *   master structure are protected by locks
 *   - xp_req_lock protects the request queue:
 *      xp_req_head, xp_req_tail, xp_reqs, xp_size, xp_full, xp_enable
 *   - xp_thread_lock protects the thread (clone) counts
 *      xp_threads, xp_detached_threads, xp_wq
 *   Each master transport is registered to exactly one thread pool.
 *
 * Clone transport handle (SVCXPRT)
 *   The clone transport handle structure is a per-service-thread handle
 *   to the transport. The structure carries all the fields/buffers used
 *   for request processing. A service thread or, in other words, a clone
 *   structure, can be linked to an arbitrary master structure to process
 *   requests on this transport. The master handle keeps track of reference
 *   counts of threads (clones) linked to it. A service thread can switch
 *   to another transport by unlinking its clone handle from the current
 *   transport and linking to a new one. Switching is relatively inexpensive
 *   but it involves locking (master's xprt->xp_thread_lock).
 *
 * Pools.
 *   A pool represents a kernel RPC service (NFS, Lock Manager, etc.).
 *   Transports related to the service are registered to the service pool.
 *   Service threads can switch between different transports in the pool.
 *   Thus, each service has its own pool of service threads. The maximum
 *   number of threads in a pool is pool->p_maxthreads. This limit allows
 *   to restrict resource usage by the service. Some fields are protected
 *   by locks:
 *   - p_req_lock protects several counts and flags:
 *      p_reqs, p_size, p_walkers, p_asleep, p_drowsy, p_req_cv
 *   - p_thread_lock governs other thread counts:
 *      p_threads, p_detached_threads, p_reserved_threads, p_closing
 *
 *   In addition, each pool contains a doubly-linked list of transports,
 *   an `xprt-ready' queue and a creator thread (see below). Threads in
 *   the pool share some other parameters such as stack size and
 *   polling timeout.
 *
 *   Pools are initialized through the svc_pool_create() function called from
 *   the nfssys() system call. However, thread creation must be done by
 *   the userland agent. This is done by using SVCPOOL_WAIT and
 *   SVCPOOL_RUN arguments to nfssys(), which call svc_wait() and
 *   svc_do_run(), respectively. Once the pool has been initialized,
 *   the userland process must set up a 'creator' thread. This thread
 *   should park itself in the kernel by calling svc_wait(). If
 *   svc_wait() returns successfully, it should fork off a new worker
 *   thread, which then calls svc_do_run() in order to get work. When
 *   that thread is complete, svc_do_run() will return, and the user
 *   program should call thr_exit().
 *
 *   When we try to register a new pool and there is an old pool with
 *   the same id in the doubly linked pool list (this happens when we kill
 *   and restart nfsd or lockd), then we unlink the old pool from the list
 *   and mark its state as `closing'. After that the transports can still
 *   process requests but new transports won't be registered. When all the
 *   transports and service threads associated with the pool are gone the
 *   creator thread (see below) will clean up the pool structure and exit.
 *
 * svc_queuereq() and svc_run().
 *   The kernel RPC server is interrupt driven. The svc_queuereq() interrupt
 *   routine is called to deliver an RPC request. The service threads
 *   loop in svc_run(). The interrupt function queues a request on the
 *   transport's queue and it makes sure that the request is serviced.
 *   It may either wake up one of sleeping threads, or ask for a new thread
 *   to be created, or, if the previous request is just being picked up, do
 *   nothing. In the last case the service thread that is picking up the
 *   previous request will wake up or create the next thread. After a service
 *   thread processes a request and sends a reply it returns to svc_run()
 *   and svc_run() calls svc_poll() to find new input.
 *
 * svc_poll().
 *   In order to avoid unnecessary locking, which causes performance
 *   problems, we always look for a pending request on the current transport.
 *   If there is none we take a hint from the pool's `xprt-ready' queue.
 *   If the queue had an overflow we switch to the `drain' mode checking
 *   each transport  in the pool's transport list. Once we find a
 *   master transport handle with a pending request we latch the request
 *   lock on this transport and return to svc_run(). If the request
 *   belongs to a transport different than the one the service thread is
 *   linked to we need to unlink and link again.
 *
 *   A service thread goes asleep when there are no pending
 *   requests on the transports registered on the pool's transports.
 *   All the pool's threads sleep on the same condition variable.
 *   If a thread has been sleeping for too long period of time
 *   (by default 5 seconds) it wakes up and exits.  Also when a transport
 *   is closing sleeping threads wake up to unlink from this transport.
 *
 * The `xprt-ready' queue.
 *   If a service thread finds no request on a transport it is currently linked
 *   to it will find another transport with a pending request. To make
 *   this search more efficient each pool has an `xprt-ready' queue.
 *   The queue is a FIFO. When the interrupt routine queues a request it also
 *   inserts a pointer to the transport into the `xprt-ready' queue. A
 *   thread looking for a transport with a pending request can pop up a
 *   transport and check for a request. The request can be already gone
 *   since it could be taken by a thread linked to that transport. In such a
 *   case we try the next hint. The `xprt-ready' queue has fixed size (by
 *   default 256 nodes). If it overflows svc_poll() has to switch to the
 *   less efficient but safe `drain' mode and walk through the pool's
 *   transport list.
 *
 *   Both the svc_poll() loop and the `xprt-ready' queue are optimized
 *   for the peak load case that is for the situation when the queue is not
 *   empty, there are all the time few pending requests, and a service
 *   thread which has just processed a request does not go asleep but picks
 *   up immediately the next request.
 *
 * Thread creator.
 *   Each pool has a thread creator associated with it. The creator thread
 *   sleeps on a condition variable and waits for a signal to create a
 *   service thread. The actual thread creation is done in userland by
 *   the method described in "Pools" above.
 *
 *   Signaling threads should turn on the `creator signaled' flag, and
 *   can avoid sending signals when the flag is on. The flag is cleared
 *   when the thread is created.
 *
 *   When the pool is in closing state (ie it has been already unregistered
 *   from the pool list) the last thread on the last transport in the pool
 *   should turn the p_creator_exit flag on. The creator thread will
 *   clean up the pool structure and exit.
 *
 * Thread reservation; Detaching service threads.
 *   A service thread can detach itself to block for an extended amount
 *   of time. However, to keep the service active we need to guarantee
 *   at least pool->p_redline non-detached threads that can process incoming
 *   requests. This, the maximum number of detached and reserved threads is
 *   p->p_maxthreads - p->p_redline. A service thread should first acquire
 *   a reservation, and if the reservation was granted it can detach itself.
 *   If a reservation was granted but the thread does not detach itself
 *   it should cancel the reservation before it returns to svc_run().
 */

#include <sys/param.h>
#include <sys/types.h>
#include <rpc/types.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/tiuser.h>
#include <sys/t_kuser.h>
#include <netinet/in.h>
#include <rpc/xdr.h>
#include <rpc/auth.h>
#include <rpc/clnt.h>
#include <rpc/rpc_msg.h>
#include <rpc/svc.h>
#include <sys/proc.h>
#include <sys/user.h>
#include <sys/stream.h>
#include <sys/strsubr.h>
#include <sys/strsun.h>
#include <sys/tihdr.h>
#include <sys/debug.h>
#include <sys/cmn_err.h>
#include <sys/file.h>
#include <sys/systm.h>
#include <sys/callb.h>
#include <sys/vtrace.h>
#include <sys/zone.h>
#include <nfs/nfs.h>
#include <sys/tsol/label_macro.h>

/*
 * Defines for svc_poll()
 */
#define SVC_EXPRTGONE ((SVCMASTERXPRT *)1)      /* Transport is closing */
#define SVC_ETIMEDOUT ((SVCMASTERXPRT *)2)      /* Timeout */
#define SVC_EINTR ((SVCMASTERXPRT *)3)          /* Interrupted by signal */

/*
 * Default stack size for service threads.
 */
#define DEFAULT_SVC_RUN_STKSIZE         (0)     /* default kernel stack */

int    svc_default_stksize = DEFAULT_SVC_RUN_STKSIZE;

/*
 * Default polling timeout for service threads.
 * Multiplied by hz when used.
 */
#define DEFAULT_SVC_POLL_TIMEOUT        (5)     /* seconds */

clock_t svc_default_timeout = DEFAULT_SVC_POLL_TIMEOUT;

/*
 * Size of the `xprt-ready' queue.
 */
#define DEFAULT_SVC_QSIZE               (256)   /* qnodes */

size_t svc_default_qsize = DEFAULT_SVC_QSIZE;

/*
 * Default limit for the number of service threads.
 */
#define DEFAULT_SVC_MAXTHREADS          (INT16_MAX)

int    svc_default_maxthreads = DEFAULT_SVC_MAXTHREADS;

/*
 * Maximum number of requests from the same transport (in `drain' mode).
 */
#define DEFAULT_SVC_MAX_SAME_XPRT       (8)

int    svc_default_max_same_xprt = DEFAULT_SVC_MAX_SAME_XPRT;


/*
 * Default `Redline' of non-detached threads.
 * Total number of detached and reserved threads in an RPC server
 * thread pool is limited to pool->p_maxthreads - svc_redline.
 */
#define DEFAULT_SVC_REDLINE             (1)

int    svc_default_redline = DEFAULT_SVC_REDLINE;

/*
 * A node for the `xprt-ready' queue.
 * See below.
 */
struct __svcxprt_qnode {
        __SVCXPRT_QNODE *q_next;
        SVCMASTERXPRT   *q_xprt;
};

/*
 * Global SVC variables (private).
 */
struct svc_globals {
        SVCPOOL         *svc_pools;
        kmutex_t        svc_plock;
};

/*
 * Debug variable to check for rdma based
 * transport startup and cleanup. Contorlled
 * through /etc/system. Off by default.
 */
int rdma_check = 0;

/*
 * This allows disabling flow control in svc_queuereq().
 */
volatile int svc_flowcontrol_disable = 0;

/*
 * Authentication parameters list.
 */
static caddr_t rqcred_head;
static kmutex_t rqcred_lock;

/*
 * If true, then keep quiet about version mismatch.
 * This macro is for broadcast RPC only. We have no broadcast RPC in
 * kernel now but one may define a flag in the transport structure
 * and redefine this macro.
 */
#define version_keepquiet(xprt) (FALSE)

/*
 * ZSD key used to retrieve zone-specific svc globals
 */
static zone_key_t svc_zone_key;

static void svc_callout_free(SVCMASTERXPRT *);
static void svc_xprt_qinit(SVCPOOL *, size_t);
static void svc_xprt_qdestroy(SVCPOOL *);
static void svc_thread_creator(SVCPOOL *);
static void svc_creator_signal(SVCPOOL *);
static void svc_creator_signalexit(SVCPOOL *);
static void svc_pool_unregister(struct svc_globals *, SVCPOOL *);
static int svc_run(SVCPOOL *);

/* ARGSUSED */
static void *
svc_zoneinit(zoneid_t zoneid)
{
        struct svc_globals *svc;

        svc = kmem_alloc(sizeof (*svc), KM_SLEEP);
        mutex_init(&svc->svc_plock, NULL, MUTEX_DEFAULT, NULL);
        svc->svc_pools = NULL;
        return (svc);
}

/* ARGSUSED */
static void
svc_zoneshutdown(zoneid_t zoneid, void *arg)
{
        struct svc_globals *svc = arg;
        SVCPOOL *pool;

        mutex_enter(&svc->svc_plock);
        while ((pool = svc->svc_pools) != NULL) {
                svc_pool_unregister(svc, pool);
        }
        mutex_exit(&svc->svc_plock);
}

/* ARGSUSED */
static void
svc_zonefini(zoneid_t zoneid, void *arg)
{
        struct svc_globals *svc = arg;

        ASSERT(svc->svc_pools == NULL);
        mutex_destroy(&svc->svc_plock);
        kmem_free(svc, sizeof (*svc));
}

/*
 * Global SVC init routine.
 * Initialize global generic and transport type specific structures
 * used by the kernel RPC server side. This routine is called only
 * once when the module is being loaded.
 */
void
svc_init()
{
        zone_key_create(&svc_zone_key, svc_zoneinit, svc_zoneshutdown,
            svc_zonefini);
        svc_cots_init();
        svc_clts_init();
}

/*
 * Destroy the SVCPOOL structure.
 */
static void
svc_pool_cleanup(SVCPOOL *pool)
{
        ASSERT(pool->p_threads + pool->p_detached_threads == 0);
        ASSERT(pool->p_lcount == 0);
        ASSERT(pool->p_closing);

        /*
         * Call the user supplied shutdown function.  This is done
         * here so the user of the pool will be able to cleanup
         * service related resources.
         */
        if (pool->p_shutdown != NULL)
                (pool->p_shutdown)();

        /* Destroy `xprt-ready' queue */
        svc_xprt_qdestroy(pool);

        /* Destroy transport list */
        rw_destroy(&pool->p_lrwlock);

        /* Destroy locks and condition variables */
        mutex_destroy(&pool->p_thread_lock);
        mutex_destroy(&pool->p_req_lock);
        cv_destroy(&pool->p_req_cv);

        /* Destroy creator's locks and condition variables */
        mutex_destroy(&pool->p_creator_lock);
        cv_destroy(&pool->p_creator_cv);
        mutex_destroy(&pool->p_user_lock);
        cv_destroy(&pool->p_user_cv);

        /* Free pool structure */
        kmem_free(pool, sizeof (SVCPOOL));
}

/*
 * If all the transports and service threads are already gone
 * signal the creator thread to clean up and exit.
 */
static bool_t
svc_pool_tryexit(SVCPOOL *pool)
{
        ASSERT(MUTEX_HELD(&pool->p_thread_lock));
        ASSERT(pool->p_closing);

        if (pool->p_threads + pool->p_detached_threads == 0) {
                rw_enter(&pool->p_lrwlock, RW_READER);
                if (pool->p_lcount == 0) {
                        /*
                         * Release the locks before sending a signal.
                         */
                        rw_exit(&pool->p_lrwlock);
                        mutex_exit(&pool->p_thread_lock);

                        /*
                         * Notify the creator thread to clean up and exit
                         *
                         * NOTICE: No references to the pool beyond this point!
                         *                 The pool is being destroyed.
                         */
                        ASSERT(!MUTEX_HELD(&pool->p_thread_lock));
                        svc_creator_signalexit(pool);

                        return (TRUE);
                }
                rw_exit(&pool->p_lrwlock);
        }

        ASSERT(MUTEX_HELD(&pool->p_thread_lock));
        return (FALSE);
}

/*
 * Find a pool with a given id.
 */
static SVCPOOL *
svc_pool_find(struct svc_globals *svc, int id)
{
        SVCPOOL *pool;

        ASSERT(MUTEX_HELD(&svc->svc_plock));

        /*
         * Search the list for a pool with a matching id
         * and register the transport handle with that pool.
         */
        for (pool = svc->svc_pools; pool; pool = pool->p_next)
                if (pool->p_id == id)
                        return (pool);

        return (NULL);
}

/*
 * PSARC 2003/523 Contract Private Interface
 * svc_do_run
 * Changes must be reviewed by Solaris File Sharing
 * Changes must be communicated to contract-2003-523@sun.com
 */
int
svc_do_run(int id)
{
        SVCPOOL *pool;
        int err = 0;
        struct svc_globals *svc;

        svc = zone_getspecific(svc_zone_key, curproc->p_zone);
        mutex_enter(&svc->svc_plock);

        pool = svc_pool_find(svc, id);

        mutex_exit(&svc->svc_plock);

        if (pool == NULL)
                return (ENOENT);

        /*
         * Increment counter of pool threads now
         * that a thread has been created.
         */
        mutex_enter(&pool->p_thread_lock);
        pool->p_threads++;
        mutex_exit(&pool->p_thread_lock);

        /* Give work to the new thread. */
        err = svc_run(pool);

        return (err);
}

/*
 * Unregister a pool from the pool list.
 * Set the closing state. If all the transports and service threads
 * are already gone signal the creator thread to clean up and exit.
 */
static void
svc_pool_unregister(struct svc_globals *svc, SVCPOOL *pool)
{
        SVCPOOL *next = pool->p_next;
        SVCPOOL *prev = pool->p_prev;

        ASSERT(MUTEX_HELD(&svc->svc_plock));

        /* Remove from the list */
        if (pool == svc->svc_pools)
                svc->svc_pools = next;
        if (next)
                next->p_prev = prev;
        if (prev)
                prev->p_next = next;
        pool->p_next = pool->p_prev = NULL;

        /*
         * Offline the pool. Mark the pool as closing.
         * If there are no transports in this pool notify
         * the creator thread to clean it up and exit.
         */
        mutex_enter(&pool->p_thread_lock);
        if (pool->p_offline != NULL)
                (pool->p_offline)();
        pool->p_closing = TRUE;
        if (svc_pool_tryexit(pool))
                return;
        mutex_exit(&pool->p_thread_lock);
}

/*
 * Register a pool with a given id in the global doubly linked pool list.
 * - if there is a pool with the same id in the list then unregister it
 * - insert the new pool into the list.
 */
static void
svc_pool_register(struct svc_globals *svc, SVCPOOL *pool, int id)
{
        SVCPOOL *old_pool;

        /*
         * If there is a pool with the same id then remove it from
         * the list and mark the pool as closing.
         */
        mutex_enter(&svc->svc_plock);

        if (old_pool = svc_pool_find(svc, id))
                svc_pool_unregister(svc, old_pool);

        /* Insert into the doubly linked list */
        pool->p_id = id;
        pool->p_next = svc->svc_pools;
        pool->p_prev = NULL;
        if (svc->svc_pools)
                svc->svc_pools->p_prev = pool;
        svc->svc_pools = pool;

        mutex_exit(&svc->svc_plock);
}

/*
 * Initialize a newly created pool structure
 */
static int
svc_pool_init(SVCPOOL *pool, uint_t maxthreads, uint_t redline,
    uint_t qsize, uint_t timeout, uint_t stksize, uint_t max_same_xprt)
{
        klwp_t *lwp = ttolwp(curthread);

        ASSERT(pool);

        if (maxthreads == 0)
                maxthreads = svc_default_maxthreads;
        if (redline == 0)
                redline = svc_default_redline;
        if (qsize == 0)
                qsize = svc_default_qsize;
        if (timeout == 0)
                timeout = svc_default_timeout;
        if (stksize == 0)
                stksize = svc_default_stksize;
        if (max_same_xprt == 0)
                max_same_xprt = svc_default_max_same_xprt;

        if (maxthreads < redline)
                return (EINVAL);

        /* Allocate and initialize the `xprt-ready' queue */
        svc_xprt_qinit(pool, qsize);

        /* Initialize doubly-linked xprt list */
        rw_init(&pool->p_lrwlock, NULL, RW_DEFAULT, NULL);

        /*
         * Setting lwp_childstksz on the current lwp so that
         * descendants of this lwp get the modified stacksize, if
         * it is defined. It is important that either this lwp or
         * one of its descendants do the actual servicepool thread
         * creation to maintain the stacksize inheritance.
         */
        if (lwp != NULL)
                lwp->lwp_childstksz = stksize;

        /* Initialize thread limits, locks and condition variables */
        pool->p_maxthreads = maxthreads;
        pool->p_redline = redline;
        pool->p_timeout = timeout * hz;
        pool->p_stksize = stksize;
        pool->p_max_same_xprt = max_same_xprt;
        mutex_init(&pool->p_thread_lock, NULL, MUTEX_DEFAULT, NULL);
        mutex_init(&pool->p_req_lock, NULL, MUTEX_DEFAULT, NULL);
        cv_init(&pool->p_req_cv, NULL, CV_DEFAULT, NULL);

        /* Initialize userland creator */
        pool->p_user_exit = FALSE;
        pool->p_signal_create_thread = FALSE;
        pool->p_user_waiting = FALSE;
        mutex_init(&pool->p_user_lock, NULL, MUTEX_DEFAULT, NULL);
        cv_init(&pool->p_user_cv, NULL, CV_DEFAULT, NULL);

        /* Initialize the creator and start the creator thread */
        pool->p_creator_exit = FALSE;
        mutex_init(&pool->p_creator_lock, NULL, MUTEX_DEFAULT, NULL);
        cv_init(&pool->p_creator_cv, NULL, CV_DEFAULT, NULL);

        (void) zthread_create(NULL, pool->p_stksize, svc_thread_creator,
            pool, 0, minclsyspri);

        return (0);
}

/*
 * PSARC 2003/523 Contract Private Interface
 * svc_pool_create
 * Changes must be reviewed by Solaris File Sharing
 * Changes must be communicated to contract-2003-523@sun.com
 *
 * Create an kernel RPC server-side thread/transport pool.
 *
 * This is public interface for creation of a server RPC thread pool
 * for a given service provider. Transports registered with the pool's id
 * will be served by a pool's threads. This function is called from the
 * nfssys() system call.
 */
int
svc_pool_create(struct svcpool_args *args)
{
        SVCPOOL *pool;
        int error;
        struct svc_globals *svc;

        /*
         * Caller should check credentials in a way appropriate
         * in the context of the call.
         */

        svc = zone_getspecific(svc_zone_key, curproc->p_zone);
        /* Allocate a new pool */
        pool = kmem_zalloc(sizeof (SVCPOOL), KM_SLEEP);

        /*
         * Initialize the pool structure and create a creator thread.
         */
        error = svc_pool_init(pool, args->maxthreads, args->redline,
            args->qsize, args->timeout, args->stksize, args->max_same_xprt);

        if (error) {
                kmem_free(pool, sizeof (SVCPOOL));
                return (error);
        }

        /* Register the pool with the global pool list */
        svc_pool_register(svc, pool, args->id);

        return (0);
}

int
svc_pool_control(int id, int cmd, void *arg)
{
        SVCPOOL *pool;
        struct svc_globals *svc;

        svc = zone_getspecific(svc_zone_key, curproc->p_zone);

        switch (cmd) {
        case SVCPSET_SHUTDOWN_PROC:
                /*
                 * Search the list for a pool with a matching id
                 * and register the transport handle with that pool.
                 */
                mutex_enter(&svc->svc_plock);

                if ((pool = svc_pool_find(svc, id)) == NULL) {
                        mutex_exit(&svc->svc_plock);
                        return (ENOENT);
                }
                /*
                 * Grab the transport list lock before releasing the
                 * pool list lock
                 */
                rw_enter(&pool->p_lrwlock, RW_WRITER);
                mutex_exit(&svc->svc_plock);

                pool->p_shutdown = *((void (*)())arg);

                rw_exit(&pool->p_lrwlock);

                return (0);
        case SVCPSET_UNREGISTER_PROC:
                /*
                 * Search the list for a pool with a matching id
                 * and register the unregister callback handle with that pool.
                 */
                mutex_enter(&svc->svc_plock);

                if ((pool = svc_pool_find(svc, id)) == NULL) {
                        mutex_exit(&svc->svc_plock);
                        return (ENOENT);
                }
                /*
                 * Grab the transport list lock before releasing the
                 * pool list lock
                 */
                rw_enter(&pool->p_lrwlock, RW_WRITER);
                mutex_exit(&svc->svc_plock);

                pool->p_offline = *((void (*)())arg);

                rw_exit(&pool->p_lrwlock);

                return (0);
        default:
                return (EINVAL);
        }
}

/*
 * Pool's transport list manipulation routines.
 * - svc_xprt_register()
 * - svc_xprt_unregister()
 *
 * svc_xprt_register() is called from svc_tli_kcreate() to
 * insert a new master transport handle into the doubly linked
 * list of server transport handles (one list per pool).
 *
 * The list is used by svc_poll(), when it operates in `drain'
 * mode, to search for a next transport with a pending request.
 */

int
svc_xprt_register(SVCMASTERXPRT *xprt, int id)
{
        SVCMASTERXPRT *prev, *next;
        SVCPOOL *pool;
        struct svc_globals *svc;

        svc = zone_getspecific(svc_zone_key, curproc->p_zone);
        /*
         * Search the list for a pool with a matching id
         * and register the transport handle with that pool.
         */
        mutex_enter(&svc->svc_plock);

        if ((pool = svc_pool_find(svc, id)) == NULL) {
                mutex_exit(&svc->svc_plock);
                return (ENOENT);
        }

        /* Grab the transport list lock before releasing the pool list lock */
        rw_enter(&pool->p_lrwlock, RW_WRITER);
        mutex_exit(&svc->svc_plock);

        /* Don't register new transports when the pool is in closing state */
        if (pool->p_closing) {
                rw_exit(&pool->p_lrwlock);
                return (EBUSY);
        }

        /*
         * Initialize xp_pool to point to the pool.
         * We don't want to go through the pool list every time.
         */
        xprt->xp_pool = pool;

        /*
         * Insert a transport handle into the list.
         * The list head points to the most recently inserted transport.
         */
        if (pool->p_lhead == NULL)
                pool->p_lhead = xprt->xp_prev = xprt->xp_next = xprt;
        else {
                next = pool->p_lhead;
                prev = pool->p_lhead->xp_prev;

                xprt->xp_next = next;
                xprt->xp_prev = prev;

                pool->p_lhead = prev->xp_next = next->xp_prev = xprt;
        }

        /* Increment the transports count */
        pool->p_lcount++;

        rw_exit(&pool->p_lrwlock);
        return (0);
}

/*
 * Called from svc_xprt_cleanup() to remove a master transport handle
 * from the pool's list of server transports (when a transport is
 * being destroyed).
 */
void
svc_xprt_unregister(SVCMASTERXPRT *xprt)
{
        SVCPOOL *pool = xprt->xp_pool;

        /*
         * Unlink xprt from the list.
         * If the list head points to this xprt then move it
         * to the next xprt or reset to NULL if this is the last
         * xprt in the list.
         */
        rw_enter(&pool->p_lrwlock, RW_WRITER);

        if (xprt == xprt->xp_next)
                pool->p_lhead = NULL;
        else {
                SVCMASTERXPRT *next = xprt->xp_next;
                SVCMASTERXPRT *prev = xprt->xp_prev;

                next->xp_prev = prev;
                prev->xp_next = next;

                if (pool->p_lhead == xprt)
                        pool->p_lhead = next;
        }

        xprt->xp_next = xprt->xp_prev = NULL;

        /* Decrement list count */
        pool->p_lcount--;

        rw_exit(&pool->p_lrwlock);
}

static void
svc_xprt_qdestroy(SVCPOOL *pool)
{
        mutex_destroy(&pool->p_qend_lock);
        kmem_free(pool->p_qbody, pool->p_qsize * sizeof (__SVCXPRT_QNODE));
}

/*
 * Initialize an `xprt-ready' queue for a given pool.
 */
static void
svc_xprt_qinit(SVCPOOL *pool, size_t qsize)
{
        int i;

        pool->p_qsize = qsize;
        pool->p_qbody = kmem_zalloc(pool->p_qsize * sizeof (__SVCXPRT_QNODE),
            KM_SLEEP);

        for (i = 0; i < pool->p_qsize - 1; i++)
                pool->p_qbody[i].q_next = &(pool->p_qbody[i+1]);

        pool->p_qbody[pool->p_qsize-1].q_next = &(pool->p_qbody[0]);
        pool->p_qtop = &(pool->p_qbody[0]);
        pool->p_qend = &(pool->p_qbody[0]);

        mutex_init(&pool->p_qend_lock, NULL, MUTEX_DEFAULT, NULL);
}

/*
 * Called from the svc_queuereq() interrupt routine to queue
 * a hint for svc_poll() which transport has a pending request.
 * - insert a pointer to xprt into the xprt-ready queue (FIFO)
 * - if the xprt-ready queue is full turn the overflow flag on.
 *
 * NOTICE: pool->p_qtop is protected by the pool's request lock
 * and the caller (svc_queuereq()) must hold the lock.
 */
static void
svc_xprt_qput(SVCPOOL *pool, SVCMASTERXPRT *xprt)
{
        ASSERT(MUTEX_HELD(&pool->p_req_lock));

        /* If the overflow flag is on there is nothing we can do */
        if (pool->p_qoverflow)
                return;

        /* If the queue is full turn the overflow flag on and exit */
        if (pool->p_qtop->q_next == pool->p_qend) {
                mutex_enter(&pool->p_qend_lock);
                if (pool->p_qtop->q_next == pool->p_qend) {
                        pool->p_qoverflow = TRUE;
                        mutex_exit(&pool->p_qend_lock);
                        return;
                }
                mutex_exit(&pool->p_qend_lock);
        }

        /* Insert a hint and move pool->p_qtop */
        pool->p_qtop->q_xprt = xprt;
        pool->p_qtop = pool->p_qtop->q_next;
}

/*
 * Called from svc_poll() to get a hint which transport has a
 * pending request. Returns a pointer to a transport or NULL if the
 * `xprt-ready' queue is empty.
 *
 * Since we do not acquire the pool's request lock while checking if
 * the queue is empty we may miss a request that is just being delivered.
 * However this is ok since svc_poll() will retry again until the
 * count indicates that there are pending requests for this pool.
 */
static SVCMASTERXPRT *
svc_xprt_qget(SVCPOOL *pool)
{
        SVCMASTERXPRT *xprt;

        mutex_enter(&pool->p_qend_lock);
        do {
                /*
                 * If the queue is empty return NULL.
                 * Since we do not acquire the pool's request lock which
                 * protects pool->p_qtop this is not exact check. However,
                 * this is safe - if we miss a request here svc_poll()
                 * will retry again.
                 */
                if (pool->p_qend == pool->p_qtop) {
                        mutex_exit(&pool->p_qend_lock);
                        return (NULL);
                }

                /* Get a hint and move pool->p_qend */
                xprt = pool->p_qend->q_xprt;
                pool->p_qend = pool->p_qend->q_next;

                /* Skip fields deleted by svc_xprt_qdelete()     */
        } while (xprt == NULL);
        mutex_exit(&pool->p_qend_lock);

        return (xprt);
}

/*
 * Delete all the references to a transport handle that
 * is being destroyed from the xprt-ready queue.
 * Deleted pointers are replaced with NULLs.
 */
static void
svc_xprt_qdelete(SVCPOOL *pool, SVCMASTERXPRT *xprt)
{
        __SVCXPRT_QNODE *q;

        mutex_enter(&pool->p_req_lock);
        for (q = pool->p_qend; q != pool->p_qtop; q = q->q_next) {
                if (q->q_xprt == xprt)
                        q->q_xprt = NULL;
        }
        mutex_exit(&pool->p_req_lock);
}

/*
 * Destructor for a master server transport handle.
 * - if there are no more non-detached threads linked to this transport
 *   then, if requested, call xp_closeproc (we don't wait for detached
 *   threads linked to this transport to complete).
 * - if there are no more threads linked to this
 *   transport then
 *   a) remove references to this transport from the xprt-ready queue
 *   b) remove a reference to this transport from the pool's transport list
 *   c) call a transport specific `destroy' function
 *   d) cancel remaining thread reservations.
 *
 * NOTICE: Caller must hold the transport's thread lock.
 */
static void
svc_xprt_cleanup(SVCMASTERXPRT *xprt, bool_t detached)
{
        ASSERT(MUTEX_HELD(&xprt->xp_thread_lock));
        ASSERT(xprt->xp_wq == NULL);

        /*
         * If called from the last non-detached thread
         * it should call the closeproc on this transport.
         */
        if (!detached && xprt->xp_threads == 0 && xprt->xp_closeproc) {
                (*(xprt->xp_closeproc)) (xprt);
        }

        if (xprt->xp_threads + xprt->xp_detached_threads > 0)
                mutex_exit(&xprt->xp_thread_lock);
        else {
                /* Remove references to xprt from the `xprt-ready' queue */
                svc_xprt_qdelete(xprt->xp_pool, xprt);

                /* Unregister xprt from the pool's transport list */
                svc_xprt_unregister(xprt);
                svc_callout_free(xprt);
                SVC_DESTROY(xprt);
        }
}

/*
 * Find a dispatch routine for a given prog/vers pair.
 * This function is called from svc_getreq() to search the callout
 * table for an entry with a matching RPC program number `prog'
 * and a version range that covers `vers'.
 * - if it finds a matching entry it returns pointer to the dispatch routine
 * - otherwise it returns NULL and fills both vers_min and vers_max
 *   with, respectively, lowest version and highest version
 *   supported for the program `prog'
 */
static SVC_DISPATCH *
svc_callout_find(SVCXPRT *xprt, rpcprog_t prog, rpcvers_t vers,
    rpcvers_t *vers_min, rpcvers_t *vers_max)
{
        SVC_CALLOUT_TABLE *sct = xprt->xp_sct;
        int i;

        *vers_min = ~(rpcvers_t)0;
        *vers_max = 0;

        for (i = 0; i < sct->sct_size; i++) {
                SVC_CALLOUT *sc = &sct->sct_sc[i];

                if (prog == sc->sc_prog) {
                        if (vers >= sc->sc_versmin && vers <= sc->sc_versmax)
                                return (sc->sc_dispatch);

                        if (*vers_max < sc->sc_versmax)
                                *vers_max = sc->sc_versmax;
                        if (*vers_min > sc->sc_versmin)
                                *vers_min = sc->sc_versmin;
                }
        }

        return (NULL);
}

/*
 * Optionally free callout table allocated for this transport by
 * the service provider.
 */
static void
svc_callout_free(SVCMASTERXPRT *xprt)
{
        SVC_CALLOUT_TABLE *sct = xprt->xp_sct;

        if (sct->sct_free) {
                kmem_free(sct->sct_sc, sct->sct_size * sizeof (SVC_CALLOUT));
                kmem_free(sct, sizeof (SVC_CALLOUT_TABLE));
        }
}

/*
 * Send a reply to an RPC request
 *
 * PSARC 2003/523 Contract Private Interface
 * svc_sendreply
 * Changes must be reviewed by Solaris File Sharing
 * Changes must be communicated to contract-2003-523@sun.com
 */
bool_t
svc_sendreply(const SVCXPRT *clone_xprt, const xdrproc_t xdr_results,
    const caddr_t xdr_location)
{
        struct rpc_msg rply;

        rply.rm_direction = REPLY;
        rply.rm_reply.rp_stat = MSG_ACCEPTED;
        rply.acpted_rply.ar_verf = clone_xprt->xp_verf;
        rply.acpted_rply.ar_stat = SUCCESS;
        rply.acpted_rply.ar_results.where = xdr_location;
        rply.acpted_rply.ar_results.proc = xdr_results;

        return (SVC_REPLY((SVCXPRT *)clone_xprt, &rply));
}

/*
 * No procedure error reply
 *
 * PSARC 2003/523 Contract Private Interface
 * svcerr_noproc
 * Changes must be reviewed by Solaris File Sharing
 * Changes must be communicated to contract-2003-523@sun.com
 */
void
svcerr_noproc(const SVCXPRT *clone_xprt)
{
        struct rpc_msg rply;

        rply.rm_direction = REPLY;
        rply.rm_reply.rp_stat = MSG_ACCEPTED;
        rply.acpted_rply.ar_verf = clone_xprt->xp_verf;
        rply.acpted_rply.ar_stat = PROC_UNAVAIL;
        SVC_FREERES((SVCXPRT *)clone_xprt);
        SVC_REPLY((SVCXPRT *)clone_xprt, &rply);
}

/*
 * Can't decode arguments error reply
 *
 * PSARC 2003/523 Contract Private Interface
 * svcerr_decode
 * Changes must be reviewed by Solaris File Sharing
 * Changes must be communicated to contract-2003-523@sun.com
 */
void
svcerr_decode(const SVCXPRT *clone_xprt)
{
        struct rpc_msg rply;

        rply.rm_direction = REPLY;
        rply.rm_reply.rp_stat = MSG_ACCEPTED;
        rply.acpted_rply.ar_verf = clone_xprt->xp_verf;
        rply.acpted_rply.ar_stat = GARBAGE_ARGS;
        SVC_FREERES((SVCXPRT *)clone_xprt);
        SVC_REPLY((SVCXPRT *)clone_xprt, &rply);
}

/*
 * Some system error
 */
void
svcerr_systemerr(const SVCXPRT *clone_xprt)
{
        struct rpc_msg rply;

        rply.rm_direction = REPLY;
        rply.rm_reply.rp_stat = MSG_ACCEPTED;
        rply.acpted_rply.ar_verf = clone_xprt->xp_verf;
        rply.acpted_rply.ar_stat = SYSTEM_ERR;
        SVC_FREERES((SVCXPRT *)clone_xprt);
        SVC_REPLY((SVCXPRT *)clone_xprt, &rply);
}

/*
 * Authentication error reply
 */
void
svcerr_auth(const SVCXPRT *clone_xprt, const enum auth_stat why)
{
        struct rpc_msg rply;

        rply.rm_direction = REPLY;
        rply.rm_reply.rp_stat = MSG_DENIED;
        rply.rjcted_rply.rj_stat = AUTH_ERROR;
        rply.rjcted_rply.rj_why = why;
        SVC_FREERES((SVCXPRT *)clone_xprt);
        SVC_REPLY((SVCXPRT *)clone_xprt, &rply);
}

/*
 * Authentication too weak error reply
 */
void
svcerr_weakauth(const SVCXPRT *clone_xprt)
{
        svcerr_auth((SVCXPRT *)clone_xprt, AUTH_TOOWEAK);
}

/*
 * Authentication error; bad credentials
 */
void
svcerr_badcred(const SVCXPRT *clone_xprt)
{
        struct rpc_msg rply;

        rply.rm_direction = REPLY;
        rply.rm_reply.rp_stat = MSG_DENIED;
        rply.rjcted_rply.rj_stat = AUTH_ERROR;
        rply.rjcted_rply.rj_why = AUTH_BADCRED;
        SVC_FREERES((SVCXPRT *)clone_xprt);
        SVC_REPLY((SVCXPRT *)clone_xprt, &rply);
}

/*
 * Program unavailable error reply
 *
 * PSARC 2003/523 Contract Private Interface
 * svcerr_noprog
 * Changes must be reviewed by Solaris File Sharing
 * Changes must be communicated to contract-2003-523@sun.com
 */
void
svcerr_noprog(const SVCXPRT *clone_xprt)
{
        struct rpc_msg rply;

        rply.rm_direction = REPLY;
        rply.rm_reply.rp_stat = MSG_ACCEPTED;
        rply.acpted_rply.ar_verf = clone_xprt->xp_verf;
        rply.acpted_rply.ar_stat = PROG_UNAVAIL;
        SVC_FREERES((SVCXPRT *)clone_xprt);
        SVC_REPLY((SVCXPRT *)clone_xprt, &rply);
}

/*
 * Program version mismatch error reply
 *
 * PSARC 2003/523 Contract Private Interface
 * svcerr_progvers
 * Changes must be reviewed by Solaris File Sharing
 * Changes must be communicated to contract-2003-523@sun.com
 */
void
svcerr_progvers(const SVCXPRT *clone_xprt,
    const rpcvers_t low_vers, const rpcvers_t high_vers)
{
        struct rpc_msg rply;

        rply.rm_direction = REPLY;
        rply.rm_reply.rp_stat = MSG_ACCEPTED;
        rply.acpted_rply.ar_verf = clone_xprt->xp_verf;
        rply.acpted_rply.ar_stat = PROG_MISMATCH;
        rply.acpted_rply.ar_vers.low = low_vers;
        rply.acpted_rply.ar_vers.high = high_vers;
        SVC_FREERES((SVCXPRT *)clone_xprt);
        SVC_REPLY((SVCXPRT *)clone_xprt, &rply);
}

/*
 * Get server side input from some transport.
 *
 * Statement of authentication parameters management:
 * This function owns and manages all authentication parameters, specifically
 * the "raw" parameters (msg.rm_call.cb_cred and msg.rm_call.cb_verf) and
 * the "cooked" credentials (rqst->rq_clntcred).
 * However, this function does not know the structure of the cooked
 * credentials, so it make the following assumptions:
 *   a) the structure is contiguous (no pointers), and
 *   b) the cred structure size does not exceed RQCRED_SIZE bytes.
 * In all events, all three parameters are freed upon exit from this routine.
 * The storage is trivially managed on the call stack in user land, but
 * is malloced in kernel land.
 *
 * Note: the xprt's xp_svc_lock is not held while the service's dispatch
 * routine is running.  If we decide to implement svc_unregister(), we'll
 * need to decide whether it's okay for a thread to unregister a service
 * while a request is being processed.  If we decide that this is a
 * problem, we can probably use some sort of reference counting scheme to
 * keep the callout entry from going away until the request has completed.
 */
static void
svc_getreq(
        SVCXPRT *clone_xprt,    /* clone transport handle */
        mblk_t *mp)
{
        struct rpc_msg msg;
        struct svc_req r;
        char  *cred_area;       /* too big to allocate on call stack */

        TRACE_0(TR_FAC_KRPC, TR_SVC_GETREQ_START,
            "svc_getreq_start:");

        ASSERT(clone_xprt->xp_master != NULL);
        ASSERT(!is_system_labeled() || msg_getcred(mp, NULL) != NULL ||
            mp->b_datap->db_type != M_DATA);

        /*
         * Firstly, allocate the authentication parameters' storage
         */
        mutex_enter(&rqcred_lock);
        if (rqcred_head) {
                cred_area = rqcred_head;

                /* LINTED pointer alignment */
                rqcred_head = *(caddr_t *)rqcred_head;
                mutex_exit(&rqcred_lock);
        } else {
                mutex_exit(&rqcred_lock);
                cred_area = kmem_alloc(2 * MAX_AUTH_BYTES + RQCRED_SIZE,
                    KM_SLEEP);
        }
        msg.rm_call.cb_cred.oa_base = cred_area;
        msg.rm_call.cb_verf.oa_base = &(cred_area[MAX_AUTH_BYTES]);
        r.rq_clntcred = &(cred_area[2 * MAX_AUTH_BYTES]);

        /*
         * underlying transport recv routine may modify mblk data
         * and make it difficult to extract label afterwards. So
         * get the label from the raw mblk data now.
         */
        if (is_system_labeled()) {
                cred_t *cr;

                r.rq_label = kmem_alloc(sizeof (bslabel_t), KM_SLEEP);
                cr = msg_getcred(mp, NULL);
                ASSERT(cr != NULL);

                bcopy(label2bslabel(crgetlabel(cr)), r.rq_label,
                    sizeof (bslabel_t));
        } else {
                r.rq_label = NULL;
        }

        /*
         * Now receive a message from the transport.
         */
        if (SVC_RECV(clone_xprt, mp, &msg)) {
                void (*dispatchroutine) (struct svc_req *, SVCXPRT *);
                rpcvers_t vers_min;
                rpcvers_t vers_max;
                bool_t no_dispatch;
                enum auth_stat why;

                /*
                 * Find the registered program and call its
                 * dispatch routine.
                 */
                r.rq_xprt = clone_xprt;
                r.rq_prog = msg.rm_call.cb_prog;
                r.rq_vers = msg.rm_call.cb_vers;
                r.rq_proc = msg.rm_call.cb_proc;
                r.rq_cred = msg.rm_call.cb_cred;

                /*
                 * First authenticate the message.
                 */
                TRACE_0(TR_FAC_KRPC, TR_SVC_GETREQ_AUTH_START,
                    "svc_getreq_auth_start:");
                if ((why = sec_svc_msg(&r, &msg, &no_dispatch)) != AUTH_OK) {
                        TRACE_1(TR_FAC_KRPC, TR_SVC_GETREQ_AUTH_END,
                            "svc_getreq_auth_end:(%S)", "failed");
                        svcerr_auth(clone_xprt, why);
                        /*
                         * Free the arguments.
                         */
                        (void) SVC_FREEARGS(clone_xprt, NULL, NULL);
                } else if (no_dispatch) {
                        /*
                         * XXX - when bug id 4053736 is done, remove
                         * the SVC_FREEARGS() call.
                         */
                        (void) SVC_FREEARGS(clone_xprt, NULL, NULL);
                } else {
                        TRACE_1(TR_FAC_KRPC, TR_SVC_GETREQ_AUTH_END,
                            "svc_getreq_auth_end:(%S)", "good");

                        dispatchroutine = svc_callout_find(clone_xprt,
                            r.rq_prog, r.rq_vers, &vers_min, &vers_max);

                        if (dispatchroutine) {
                                (*dispatchroutine) (&r, clone_xprt);
                        } else {
                                /*
                                 * If we got here, the program or version
                                 * is not served ...
                                 */
                                if (vers_max == 0 ||
                                    version_keepquiet(clone_xprt))
                                        svcerr_noprog(clone_xprt);
                                else
                                        svcerr_progvers(clone_xprt, vers_min,
                                            vers_max);

                                /*
                                 * Free the arguments. For successful calls
                                 * this is done by the dispatch routine.
                                 */
                                (void) SVC_FREEARGS(clone_xprt, NULL, NULL);
                                /* Fall through to ... */
                        }
                        /*
                         * Call cleanup procedure for RPCSEC_GSS.
                         * This is a hack since there is currently no
                         * op, such as SVC_CLEANAUTH. rpc_gss_cleanup
                         * should only be called for a non null proc.
                         * Null procs in RPC GSS are overloaded to
                         * provide context setup and control. The main
                         * purpose of rpc_gss_cleanup is to decrement the
                         * reference count associated with the cached
                         * GSS security context. We should never get here
                         * for an RPCSEC_GSS null proc since *no_dispatch
                         * would have been set to true from sec_svc_msg above.
                         */
                        if (r.rq_cred.oa_flavor == RPCSEC_GSS)
                                rpc_gss_cleanup(clone_xprt);
                }
        }

        if (r.rq_label != NULL)
                kmem_free(r.rq_label, sizeof (bslabel_t));

        /*
         * Free authentication parameters' storage
         */
        mutex_enter(&rqcred_lock);
        /* LINTED pointer alignment */
        *(caddr_t *)cred_area = rqcred_head;
        rqcred_head = cred_area;
        mutex_exit(&rqcred_lock);
}

/*
 * Allocate new clone transport handle.
 */
SVCXPRT *
svc_clone_init(void)
{
        SVCXPRT *clone_xprt;

        clone_xprt = kmem_zalloc(sizeof (SVCXPRT), KM_SLEEP);
        clone_xprt->xp_cred = crget();
        return (clone_xprt);
}

/*
 * Free memory allocated by svc_clone_init.
 */
void
svc_clone_free(SVCXPRT *clone_xprt)
{
        /* Fre credentials from crget() */
        if (clone_xprt->xp_cred)
                crfree(clone_xprt->xp_cred);
        kmem_free(clone_xprt, sizeof (SVCXPRT));
}

/*
 * Link a per-thread clone transport handle to a master
 * - increment a thread reference count on the master
 * - copy some of the master's fields to the clone
 * - call a transport specific clone routine.
 */
void
svc_clone_link(SVCMASTERXPRT *xprt, SVCXPRT *clone_xprt, SVCXPRT *clone_xprt2)
{
        cred_t *cred = clone_xprt->xp_cred;

        ASSERT(cred);

        /*
         * Bump up master's thread count.
         * Linking a per-thread clone transport handle to a master
         * associates a service thread with the master.
         */
        mutex_enter(&xprt->xp_thread_lock);
        xprt->xp_threads++;
        mutex_exit(&xprt->xp_thread_lock);

        /* Clear everything */
        bzero(clone_xprt, sizeof (SVCXPRT));

        /* Set pointer to the master transport stucture */
        clone_xprt->xp_master = xprt;

        /* Structure copy of all the common fields */
        clone_xprt->xp_xpc = xprt->xp_xpc;

        /* Restore per-thread fields (xp_cred) */
        clone_xprt->xp_cred = cred;

        if (clone_xprt2)
                SVC_CLONE_XPRT(clone_xprt2, clone_xprt);
}

/*
 * Unlink a non-detached clone transport handle from a master
 * - decrement a thread reference count on the master
 * - if the transport is closing (xp_wq is NULL) call svc_xprt_cleanup();
 *   if this is the last non-detached/absolute thread on this transport
 *   then it will close/destroy the transport
 * - call transport specific function to destroy the clone handle
 * - clear xp_master to avoid recursion.
 */
void
svc_clone_unlink(SVCXPRT *clone_xprt)
{
        SVCMASTERXPRT *xprt = clone_xprt->xp_master;

        /* This cannot be a detached thread */
        ASSERT(!clone_xprt->xp_detached);
        ASSERT(xprt->xp_threads > 0);

        /* Decrement a reference count on the transport */
        mutex_enter(&xprt->xp_thread_lock);
        xprt->xp_threads--;

        /* svc_xprt_cleanup() unlocks xp_thread_lock or destroys xprt */
        if (xprt->xp_wq)
                mutex_exit(&xprt->xp_thread_lock);
        else
                svc_xprt_cleanup(xprt, FALSE);

        /* Call a transport specific clone `destroy' function */
        SVC_CLONE_DESTROY(clone_xprt);

        /* Clear xp_master */
        clone_xprt->xp_master = NULL;
}

/*
 * Unlink a detached clone transport handle from a master
 * - decrement the thread count on the master
 * - if the transport is closing (xp_wq is NULL) call svc_xprt_cleanup();
 *   if this is the last thread on this transport then it will destroy
 *   the transport.
 * - call a transport specific function to destroy the clone handle
 * - clear xp_master to avoid recursion.
 */
static void
svc_clone_unlinkdetached(SVCXPRT *clone_xprt)
{
        SVCMASTERXPRT *xprt = clone_xprt->xp_master;

        /* This must be a detached thread */
        ASSERT(clone_xprt->xp_detached);
        ASSERT(xprt->xp_detached_threads > 0);
        ASSERT(xprt->xp_threads + xprt->xp_detached_threads > 0);

        /* Grab xprt->xp_thread_lock and decrement link counts */
        mutex_enter(&xprt->xp_thread_lock);
        xprt->xp_detached_threads--;

        /* svc_xprt_cleanup() unlocks xp_thread_lock or destroys xprt */
        if (xprt->xp_wq)
                mutex_exit(&xprt->xp_thread_lock);
        else
                svc_xprt_cleanup(xprt, TRUE);

        /* Call transport specific clone `destroy' function */
        SVC_CLONE_DESTROY(clone_xprt);

        /* Clear xp_master */
        clone_xprt->xp_master = NULL;
}

/*
 * Try to exit a non-detached service thread
 * - check if there are enough threads left
 * - if this thread (ie its clone transport handle) are linked
 *   to a master transport then unlink it
 * - free the clone structure
 * - return to userland for thread exit
 *
 * If this is the last non-detached or the last thread on this
 * transport then the call to svc_clone_unlink() will, respectively,
 * close and/or destroy the transport.
 */
static void
svc_thread_exit(SVCPOOL *pool, SVCXPRT *clone_xprt)
{
        if (clone_xprt->xp_master)
                svc_clone_unlink(clone_xprt);
        svc_clone_free(clone_xprt);

        mutex_enter(&pool->p_thread_lock);
        pool->p_threads--;
        if (pool->p_closing && svc_pool_tryexit(pool))
                /* return -  thread exit will be handled at user level */
                return;
        mutex_exit(&pool->p_thread_lock);

        /* return -  thread exit will be handled at user level */
}

/*
 * Exit a detached service thread that returned to svc_run
 * - decrement the `detached thread' count for the pool
 * - unlink the detached clone transport handle from the master
 * - free the clone structure
 * - return to userland for thread exit
 *
 * If this is the last thread on this transport then the call
 * to svc_clone_unlinkdetached() will destroy the transport.
 */
static void
svc_thread_exitdetached(SVCPOOL *pool, SVCXPRT *clone_xprt)
{
        /* This must be a detached thread */
        ASSERT(clone_xprt->xp_master);
        ASSERT(clone_xprt->xp_detached);
        ASSERT(!MUTEX_HELD(&pool->p_thread_lock));

        svc_clone_unlinkdetached(clone_xprt);
        svc_clone_free(clone_xprt);

        mutex_enter(&pool->p_thread_lock);

        ASSERT(pool->p_reserved_threads >= 0);
        ASSERT(pool->p_detached_threads > 0);

        pool->p_detached_threads--;
        if (pool->p_closing && svc_pool_tryexit(pool))
                /* return -  thread exit will be handled at user level */
                return;
        mutex_exit(&pool->p_thread_lock);

        /* return -  thread exit will be handled at user level */
}

/*
 * PSARC 2003/523 Contract Private Interface
 * svc_wait
 * Changes must be reviewed by Solaris File Sharing
 * Changes must be communicated to contract-2003-523@sun.com
 */
int
svc_wait(int id)
{
        SVCPOOL *pool;
        int     err = 0;
        struct svc_globals *svc;

        svc = zone_getspecific(svc_zone_key, curproc->p_zone);
        mutex_enter(&svc->svc_plock);
        pool = svc_pool_find(svc, id);
        mutex_exit(&svc->svc_plock);

        if (pool == NULL)
                return (ENOENT);

        mutex_enter(&pool->p_user_lock);

        /* Check if there's already a user thread waiting on this pool */
        if (pool->p_user_waiting) {
                mutex_exit(&pool->p_user_lock);
                return (EBUSY);
        }

        pool->p_user_waiting = TRUE;

        /* Go to sleep, waiting for the signaled flag. */
        while (!pool->p_signal_create_thread && !pool->p_user_exit) {
                if (cv_wait_sig(&pool->p_user_cv, &pool->p_user_lock) == 0) {
                        /* Interrupted, return to handle exit or signal */
                        pool->p_user_waiting = FALSE;
                        pool->p_signal_create_thread = FALSE;
                        mutex_exit(&pool->p_user_lock);

                        /*
                         * Thread has been interrupted and therefore
                         * the service daemon is leaving as well so
                         * let's go ahead and remove the service
                         * pool at this time.
                         */
                        mutex_enter(&svc->svc_plock);
                        svc_pool_unregister(svc, pool);
                        mutex_exit(&svc->svc_plock);

                        return (EINTR);
                }
        }

        pool->p_signal_create_thread = FALSE;
        pool->p_user_waiting = FALSE;

        /*
         * About to exit the service pool. Set return value
         * to let the userland code know our intent. Signal
         * svc_thread_creator() so that it can clean up the
         * pool structure.
         */
        if (pool->p_user_exit) {
                err = ECANCELED;
                cv_signal(&pool->p_user_cv);
        }

        mutex_exit(&pool->p_user_lock);

        /* Return to userland with error code, for possible thread creation. */
        return (err);
}

/*
 * `Service threads' creator thread.
 * The creator thread waits for a signal to create new thread.
 */
static void
svc_thread_creator(SVCPOOL *pool)
{
        callb_cpr_t cpr_info;   /* CPR info for the creator thread */

        CALLB_CPR_INIT(&cpr_info, &pool->p_creator_lock, callb_generic_cpr,
            "svc_thread_creator");

        for (;;) {
                mutex_enter(&pool->p_creator_lock);

                /* Check if someone set the exit flag */
                if (pool->p_creator_exit)
                        break;

                /* Clear the `signaled' flag and go asleep */
                pool->p_creator_signaled = FALSE;

                CALLB_CPR_SAFE_BEGIN(&cpr_info);
                cv_wait(&pool->p_creator_cv, &pool->p_creator_lock);
                CALLB_CPR_SAFE_END(&cpr_info, &pool->p_creator_lock);

                /* Check if someone signaled to exit */
                if (pool->p_creator_exit)
                        break;

                mutex_exit(&pool->p_creator_lock);

                mutex_enter(&pool->p_thread_lock);

                /*
                 * When the pool is in closing state and all the transports
                 * are gone the creator should not create any new threads.
                 */
                if (pool->p_closing) {
                        rw_enter(&pool->p_lrwlock, RW_READER);
                        if (pool->p_lcount == 0) {
                                rw_exit(&pool->p_lrwlock);
                                mutex_exit(&pool->p_thread_lock);
                                continue;
                        }
                        rw_exit(&pool->p_lrwlock);
                }

                /*
                 * Create a new service thread now.
                 */
                ASSERT(pool->p_reserved_threads >= 0);
                ASSERT(pool->p_detached_threads >= 0);

                if (pool->p_threads + pool->p_detached_threads <
                    pool->p_maxthreads) {
                        /*
                         * Signal the service pool wait thread
                         * only if it hasn't already been signaled.
                         */
                        mutex_enter(&pool->p_user_lock);
                        if (pool->p_signal_create_thread == FALSE) {
                                pool->p_signal_create_thread = TRUE;
                                cv_signal(&pool->p_user_cv);
                        }
                        mutex_exit(&pool->p_user_lock);

                }

                mutex_exit(&pool->p_thread_lock);
        }

        /*
         * Pool is closed. Cleanup and exit.
         */

        /* Signal userland creator thread that it can stop now. */
        mutex_enter(&pool->p_user_lock);
        pool->p_user_exit = TRUE;
        cv_broadcast(&pool->p_user_cv);
        mutex_exit(&pool->p_user_lock);

        /* Wait for svc_wait() to be done with the pool */
        mutex_enter(&pool->p_user_lock);
        while (pool->p_user_waiting) {
                CALLB_CPR_SAFE_BEGIN(&cpr_info);
                cv_wait(&pool->p_user_cv, &pool->p_user_lock);
                CALLB_CPR_SAFE_END(&cpr_info, &pool->p_creator_lock);
        }
        mutex_exit(&pool->p_user_lock);

        CALLB_CPR_EXIT(&cpr_info);
        svc_pool_cleanup(pool);
        zthread_exit();
}

/*
 * If the creator thread  is idle signal it to create
 * a new service thread.
 */
static void
svc_creator_signal(SVCPOOL *pool)
{
        mutex_enter(&pool->p_creator_lock);
        if (pool->p_creator_signaled == FALSE) {
                pool->p_creator_signaled = TRUE;
                cv_signal(&pool->p_creator_cv);
        }
        mutex_exit(&pool->p_creator_lock);
}

/*
 * Notify the creator thread to clean up and exit.
 */
static void
svc_creator_signalexit(SVCPOOL *pool)
{
        mutex_enter(&pool->p_creator_lock);
        pool->p_creator_exit = TRUE;
        cv_signal(&pool->p_creator_cv);
        mutex_exit(&pool->p_creator_lock);
}

/*
 * Polling part of the svc_run().
 * - search for a transport with a pending request
 * - when one is found then latch the request lock and return to svc_run()
 * - if there is no request go asleep and wait for a signal
 * - handle two exceptions:
 *   a) current transport is closing
 *   b) timeout waiting for a new request
 *   in both cases return to svc_run()
 */
static SVCMASTERXPRT *
svc_poll(SVCPOOL *pool, SVCMASTERXPRT *xprt, SVCXPRT *clone_xprt)
{
        /*
         * Main loop iterates until
         * a) we find a pending request,
         * b) detect that the current transport is closing
         * c) time out waiting for a new request.
         */
        for (;;) {
                SVCMASTERXPRT *next;
                clock_t timeleft;

                /*
                 * Step 1.
                 * Check if there is a pending request on the current
                 * transport handle so that we can avoid cloning.
                 * If so then decrement the `pending-request' count for
                 * the pool and return to svc_run().
                 *
                 * We need to prevent a potential starvation. When
                 * a selected transport has all pending requests coming in
                 * all the time then the service threads will never switch to
                 * another transport. With a limited number of service
                 * threads some transports may be never serviced.
                 * To prevent such a scenario we pick up at most
                 * pool->p_max_same_xprt requests from the same transport
                 * and then take a hint from the xprt-ready queue or walk
                 * the transport list.
                 */
                if (xprt && xprt->xp_req_head && (!pool->p_qoverflow ||
                    clone_xprt->xp_same_xprt++ < pool->p_max_same_xprt)) {
                        mutex_enter(&xprt->xp_req_lock);
                        if (xprt->xp_req_head)
                                return (xprt);
                        mutex_exit(&xprt->xp_req_lock);
                }
                clone_xprt->xp_same_xprt = 0;

                /*
                 * Step 2.
                 * If there is no request on the current transport try to
                 * find another transport with a pending request.
                 */
                mutex_enter(&pool->p_req_lock);
                pool->p_walkers++;
                mutex_exit(&pool->p_req_lock);

                /*
                 * Make sure that transports will not be destroyed just
                 * while we are checking them.
                 */
                rw_enter(&pool->p_lrwlock, RW_READER);

                for (;;) {
                        SVCMASTERXPRT *hint;

                        /*
                         * Get the next transport from the xprt-ready queue.
                         * This is a hint. There is no guarantee that the
                         * transport still has a pending request since it
                         * could be picked up by another thread in step 1.
                         *
                         * If the transport has a pending request then keep
                         * it locked. Decrement the `pending-requests' for
                         * the pool and `walking-threads' counts, and return
                         * to svc_run().
                         */
                        hint = svc_xprt_qget(pool);

                        if (hint && hint->xp_req_head) {
                                mutex_enter(&hint->xp_req_lock);
                                if (hint->xp_req_head) {
                                        rw_exit(&pool->p_lrwlock);

                                        mutex_enter(&pool->p_req_lock);
                                        pool->p_walkers--;
                                        mutex_exit(&pool->p_req_lock);

                                        return (hint);
                                }
                                mutex_exit(&hint->xp_req_lock);
                        }

                        /*
                         * If there was no hint in the xprt-ready queue then
                         * - if there is less pending requests than polling
                         *   threads go asleep
                         * - otherwise check if there was an overflow in the
                         *   xprt-ready queue; if so, then we need to break
                         *   the `drain' mode
                         */
                        if (hint == NULL) {
                                if (pool->p_reqs < pool->p_walkers) {
                                        mutex_enter(&pool->p_req_lock);
                                        if (pool->p_reqs < pool->p_walkers)
                                                goto sleep;
                                        mutex_exit(&pool->p_req_lock);
                                }
                                if (pool->p_qoverflow) {
                                        break;
                                }
                        }
                }

                /*
                 * If there was an overflow in the xprt-ready queue then we
                 * need to switch to the `drain' mode, i.e. walk through the
                 * pool's transport list and search for a transport with a
                 * pending request. If we manage to drain all the pending
                 * requests then we can clear the overflow flag. This will
                 * switch svc_poll() back to taking hints from the xprt-ready
                 * queue (which is generally more efficient).
                 *
                 * If there are no registered transports simply go asleep.
                 */
                if (xprt == NULL && pool->p_lhead == NULL) {
                        mutex_enter(&pool->p_req_lock);
                        goto sleep;
                }

                /*
                 * `Walk' through the pool's list of master server
                 * transport handles. Continue to loop until there are less
                 * looping threads then pending requests.
                 */
                next = xprt ? xprt->xp_next : pool->p_lhead;

                for (;;) {
                        /*
                         * Check if there is a request on this transport.
                         *
                         * Since blocking on a locked mutex is very expensive
                         * check for a request without a lock first. If we miss
                         * a request that is just being delivered but this will
                         * cost at most one full walk through the list.
                         */
                        if (next->xp_req_head) {
                                /*
                                 * Check again, now with a lock.
                                 */
                                mutex_enter(&next->xp_req_lock);
                                if (next->xp_req_head) {
                                        rw_exit(&pool->p_lrwlock);

                                        mutex_enter(&pool->p_req_lock);
                                        pool->p_walkers--;
                                        mutex_exit(&pool->p_req_lock);

                                        return (next);
                                }
                                mutex_exit(&next->xp_req_lock);
                        }

                        /*
                         * Continue to `walk' through the pool's
                         * transport list until there is less requests
                         * than walkers. Check this condition without
                         * a lock first to avoid contention on a mutex.
                         */
                        if (pool->p_reqs < pool->p_walkers) {
                                /* Check again, now with the lock. */
                                mutex_enter(&pool->p_req_lock);
                                if (pool->p_reqs < pool->p_walkers)
                                        break;  /* goto sleep */
                                mutex_exit(&pool->p_req_lock);
                        }

                        next = next->xp_next;
                }

        sleep:
                /*
                 * No work to do. Stop the `walk' and go asleep.
                 * Decrement the `walking-threads' count for the pool.
                 */
                pool->p_walkers--;
                rw_exit(&pool->p_lrwlock);

                /*
                 * Count us as asleep, mark this thread as safe
                 * for suspend and wait for a request.
                 */
                pool->p_asleep++;
                timeleft = cv_reltimedwait_sig(&pool->p_req_cv,
                    &pool->p_req_lock, pool->p_timeout, TR_CLOCK_TICK);

                /*
                 * If the drowsy flag is on this means that
                 * someone has signaled a wakeup. In such a case
                 * the `asleep-threads' count has already updated
                 * so just clear the flag.
                 *
                 * If the drowsy flag is off then we need to update
                 * the `asleep-threads' count.
                 */
                if (pool->p_drowsy) {
                        pool->p_drowsy = FALSE;
                        /*
                         * If the thread is here because it timedout,
                         * instead of returning SVC_ETIMEDOUT, it is
                         * time to do some more work.
                         */
                        if (timeleft == -1)
                                timeleft = 1;
                } else {
                        pool->p_asleep--;
                }
                mutex_exit(&pool->p_req_lock);

                /*
                 * If we received a signal while waiting for a
                 * request, inform svc_run(), so that we can return
                 * to user level and exit.
                 */
                if (timeleft == 0)
                        return (SVC_EINTR);

                /*
                 * If the current transport is gone then notify
                 * svc_run() to unlink from it.
                 */
                if (xprt && xprt->xp_wq == NULL)
                        return (SVC_EXPRTGONE);

                /*
                 * If we have timed out waiting for a request inform
                 * svc_run() that we probably don't need this thread.
                 */
                if (timeleft == -1)
                        return (SVC_ETIMEDOUT);
        }
}

/*
 * calculate memory space used by message
 */
static size_t
svc_msgsize(mblk_t *mp)
{
        size_t count = 0;

        for (; mp; mp = mp->b_cont)
                count += MBLKSIZE(mp);

        return (count);
}

/*
 * svc_flowcontrol() attempts to turn the flow control on or off for the
 * transport.
 *
 * On input the xprt->xp_full determines whether the flow control is currently
 * off (FALSE) or on (TRUE).  If it is off we do tests to see whether we should
 * turn it on, and vice versa.
 *
 * There are two conditions considered for the flow control.  Both conditions
 * have the low and the high watermark.  Once the high watermark is reached in
 * EITHER condition the flow control is turned on.  For turning the flow
 * control off BOTH conditions must be below the low watermark.
 *
 * Condition #1 - Number of requests queued:
 *
 * The max number of threads working on the pool is roughly pool->p_maxthreads.
 * Every thread could handle up to pool->p_max_same_xprt requests from one
 * transport before it moves to another transport.  See svc_poll() for details.
 * In case all threads in the pool are working on a transport they will handle
 * no more than enough_reqs (pool->p_maxthreads * pool->p_max_same_xprt)
 * requests in one shot from that transport.  We are turning the flow control
 * on once the high watermark is reached for a transport so that the underlying
 * queue knows the rate of incoming requests is higher than we are able to
 * handle.
 *
 * The high watermark: 2 * enough_reqs
 * The low watermark: enough_reqs
 *
 * Condition #2 - Length of the data payload for the queued messages/requests:
 *
 * We want to prevent a particular pool exhausting the memory, so once the
 * total length of queued requests for the whole pool reaches the high
 * watermark we start to turn on the flow control for significant memory
 * consumers (individual transports).  To keep the implementation simple
 * enough, this condition is not exact, because we count only the data part of
 * the queued requests and we ignore the overhead.  For our purposes this
 * should be enough.  We should also consider that up to pool->p_maxthreads
 * threads for the pool might work on large requests (this is not counted for
 * this condition).  We need to leave some space for rest of the system and for
 * other big memory consumers (like ZFS).  Also, after the flow control is
 * turned on (on cots transports) we can start to accumulate a few megabytes in
 * queues for each transport.
 *
 * Usually, the big memory consumers are NFS WRITE requests, so we do not
 * expect to see this condition met for other than NFS pools.
 *
 * The high watermark: 1/5 of available memory
 * The low watermark: 1/6 of available memory
 *
 * Once the high watermark is reached we turn the flow control on only for
 * transports exceeding a per-transport memory limit.  The per-transport
 * fraction of memory is calculated as:
 *
 * the high watermark / number of transports
 *
 * For transports with less than the per-transport fraction of memory consumed,
 * the flow control is not turned on, so they are not blocked by a few "hungry"
 * transports.  Because of this, the total memory consumption for the
 * particular pool might grow up to 2 * the high watermark.
 *
 * The individual transports are unblocked once their consumption is below:
 *
 * per-transport fraction of memory / 2
 *
 * or once the total memory consumption for the whole pool falls below the low
 * watermark.
 *
 */
static void
svc_flowcontrol(SVCMASTERXPRT *xprt)
{
        SVCPOOL *pool = xprt->xp_pool;
        size_t totalmem = ptob(physmem);
        int enough_reqs = pool->p_maxthreads * pool->p_max_same_xprt;

        ASSERT(MUTEX_HELD(&xprt->xp_req_lock));

        /* Should we turn the flow control on? */
        if (xprt->xp_full == FALSE) {
                /* Is flow control disabled? */
                if (svc_flowcontrol_disable != 0)
                        return;

                /* Is there enough requests queued? */
                if (xprt->xp_reqs >= enough_reqs * 2) {
                        xprt->xp_full = TRUE;
                        return;
                }

                /*
                 * If this pool uses over 20% of memory and this transport is
                 * significant memory consumer then we are full
                 */
                if (pool->p_size >= totalmem / 5 &&
                    xprt->xp_size >= totalmem / 5 / pool->p_lcount)
                        xprt->xp_full = TRUE;

                return;
        }

        /* We might want to turn the flow control off */

        /* Do we still have enough requests? */
        if (xprt->xp_reqs > enough_reqs)
                return;

        /*
         * If this pool still uses over 16% of memory and this transport is
         * still significant memory consumer then we are still full
         */
        if (pool->p_size >= totalmem / 6 &&
            xprt->xp_size >= totalmem / 5 / pool->p_lcount / 2)
                return;

        /* Turn the flow control off and make sure rpcmod is notified */
        xprt->xp_full = FALSE;
        xprt->xp_enable = TRUE;
}

/*
 * Main loop of the kernel RPC server
 * - wait for input (find a transport with a pending request).
 * - dequeue the request
 * - call a registered server routine to process the requests
 *
 * There can many threads running concurrently in this loop
 * on the same or on different transports.
 */
static int
svc_run(SVCPOOL *pool)
{
        SVCMASTERXPRT *xprt = NULL;     /* master transport handle  */
        SVCXPRT *clone_xprt;    /* clone for this thread    */
        proc_t *p = ttoproc(curthread);

        /* Allocate a clone transport handle for this thread */
        clone_xprt = svc_clone_init();

        /*
         * The loop iterates until the thread becomes
         * idle too long or the transport is gone.
         */
        for (;;) {
                SVCMASTERXPRT *next;
                mblk_t *mp;
                bool_t enable;
                size_t size;

                TRACE_0(TR_FAC_KRPC, TR_SVC_RUN, "svc_run");

                /*
                 * If the process is exiting/killed, return
                 * immediately without processing any more
                 * requests.
                 */
                if (p->p_flag & (SEXITING | SKILLED)) {
                        svc_thread_exit(pool, clone_xprt);
                        return (EINTR);
                }

                /* Find a transport with a pending request */
                next = svc_poll(pool, xprt, clone_xprt);

                /*
                 * If svc_poll() finds a transport with a request
                 * it latches xp_req_lock on it. Therefore we need
                 * to dequeue the request and release the lock as
                 * soon as possible.
                 */
                ASSERT(next != NULL &&
                    (next == SVC_EXPRTGONE ||
                    next == SVC_ETIMEDOUT ||
                    next == SVC_EINTR ||
                    MUTEX_HELD(&next->xp_req_lock)));

                /* Ooops! Current transport is closing. Unlink now */
                if (next == SVC_EXPRTGONE) {
                        svc_clone_unlink(clone_xprt);
                        xprt = NULL;
                        continue;
                }

                /* Ooops! Timeout while waiting for a request. Exit */
                if (next == SVC_ETIMEDOUT) {
                        svc_thread_exit(pool, clone_xprt);
                        return (0);
                }

                /*
                 * Interrupted by a signal while waiting for a
                 * request. Return to userspace and exit.
                 */
                if (next == SVC_EINTR) {
                        svc_thread_exit(pool, clone_xprt);
                        return (EINTR);
                }

                /*
                 * De-queue the request and release the request lock
                 * on this transport (latched by svc_poll()).
                 */
                mp = next->xp_req_head;
                next->xp_req_head = mp->b_next;
                mp->b_next = (mblk_t *)0;
                size = svc_msgsize(mp);

                mutex_enter(&pool->p_req_lock);
                pool->p_reqs--;
                if (pool->p_reqs == 0)
                        pool->p_qoverflow = FALSE;
                pool->p_size -= size;
                mutex_exit(&pool->p_req_lock);

                next->xp_reqs--;
                next->xp_size -= size;

                if (next->xp_full)
                        svc_flowcontrol(next);

                TRACE_2(TR_FAC_KRPC, TR_NFSFP_QUE_REQ_DEQ,
                    "rpc_que_req_deq:pool %p mp %p", pool, mp);
                mutex_exit(&next->xp_req_lock);

                /*
                 * If this is a new request on a current transport then
                 * the clone structure is already properly initialized.
                 * Otherwise, if the request is on a different transport,
                 * unlink from the current master and link to
                 * the one we got a request on.
                 */
                if (next != xprt) {
                        if (xprt)
                                svc_clone_unlink(clone_xprt);
                        svc_clone_link(next, clone_xprt, NULL);
                        xprt = next;
                }

                /*
                 * If there are more requests and req_cv hasn't
                 * been signaled yet then wake up one more thread now.
                 *
                 * We avoid signaling req_cv until the most recently
                 * signaled thread wakes up and gets CPU to clear
                 * the `drowsy' flag.
                 */
                if (!(pool->p_drowsy || pool->p_reqs <= pool->p_walkers ||
                    pool->p_asleep == 0)) {
                        mutex_enter(&pool->p_req_lock);

                        if (pool->p_drowsy || pool->p_reqs <= pool->p_walkers ||
                            pool->p_asleep == 0)
                                mutex_exit(&pool->p_req_lock);
                        else {
                                pool->p_asleep--;
                                pool->p_drowsy = TRUE;

                                cv_signal(&pool->p_req_cv);
                                mutex_exit(&pool->p_req_lock);
                        }
                }

                /*
                 * If there are no asleep/signaled threads, we are
                 * still below pool->p_maxthreads limit, and no thread is
                 * currently being created then signal the creator
                 * for one more service thread.
                 *
                 * The asleep and drowsy checks are not protected
                 * by a lock since it hurts performance and a wrong
                 * decision is not essential.
                 */
                if (pool->p_asleep == 0 && !pool->p_drowsy &&
                    pool->p_threads + pool->p_detached_threads <
                    pool->p_maxthreads)
                        svc_creator_signal(pool);

                /*
                 * Process the request.
                 */
                svc_getreq(clone_xprt, mp);

                /* If thread had a reservation it should have been canceled */
                ASSERT(!clone_xprt->xp_reserved);

                /*
                 * If the clone is marked detached then exit.
                 * The rpcmod slot has already been released
                 * when we detached this thread.
                 */
                if (clone_xprt->xp_detached) {
                        svc_thread_exitdetached(pool, clone_xprt);
                        return (0);
                }

                /*
                 * Release our reference on the rpcmod
                 * slot attached to xp_wq->q_ptr.
                 */
                mutex_enter(&xprt->xp_req_lock);
                enable = xprt->xp_enable;
                if (enable)
                        xprt->xp_enable = FALSE;
                mutex_exit(&xprt->xp_req_lock);
                SVC_RELE(clone_xprt, NULL, enable);
        }
        /* NOTREACHED */
}

/*
 * Flush any pending requests for the queue and
 * free the associated mblks.
 */
void
svc_queueclean(queue_t *q)
{
        SVCMASTERXPRT *xprt = ((void **) q->q_ptr)[0];
        mblk_t *mp;
        SVCPOOL *pool;

        /*
         * clean up the requests
         */
        mutex_enter(&xprt->xp_req_lock);
        pool = xprt->xp_pool;
        while ((mp = xprt->xp_req_head) != NULL) {
                /* remove the request from the list */
                xprt->xp_req_head = mp->b_next;
                mp->b_next = (mblk_t *)0;
                SVC_RELE(xprt, mp, FALSE);
        }

        mutex_enter(&pool->p_req_lock);
        pool->p_reqs -= xprt->xp_reqs;
        pool->p_size -= xprt->xp_size;
        mutex_exit(&pool->p_req_lock);

        xprt->xp_reqs = 0;
        xprt->xp_size = 0;
        xprt->xp_full = FALSE;
        xprt->xp_enable = FALSE;
        mutex_exit(&xprt->xp_req_lock);
}

/*
 * This routine is called by rpcmod to inform kernel RPC that a
 * queue is closing. It is called after all the requests have been
 * picked up (that is after all the slots on the queue have
 * been released by kernel RPC). It is also guaranteed that no more
 * request will be delivered on this transport.
 *
 * - clear xp_wq to mark the master server transport handle as closing
 * - if there are no more threads on this transport close/destroy it
 * - otherwise, leave the linked threads to close/destroy the transport
 *   later.
 */
void
svc_queueclose(queue_t *q)
{
        SVCMASTERXPRT *xprt = ((void **) q->q_ptr)[0];

        if (xprt == NULL) {
                /*
                 * If there is no master xprt associated with this stream,
                 * then there is nothing to do.  This happens regularly
                 * with connection-oriented listening streams created by
                 * nfsd.
                 */
                return;
        }

        mutex_enter(&xprt->xp_thread_lock);

        ASSERT(xprt->xp_req_head == NULL);
        ASSERT(xprt->xp_wq != NULL);

        xprt->xp_wq = NULL;

        if (xprt->xp_threads == 0) {
                SVCPOOL *pool = xprt->xp_pool;

                /*
                 * svc_xprt_cleanup() destroys the transport
                 * or releases the transport thread lock
                 */
                svc_xprt_cleanup(xprt, FALSE);

                mutex_enter(&pool->p_thread_lock);

                /*
                 * If the pool is in closing state and this was
                 * the last transport in the pool then signal the creator
                 * thread to clean up and exit.
                 */
                if (pool->p_closing && svc_pool_tryexit(pool)) {
                        return;
                }
                mutex_exit(&pool->p_thread_lock);
        } else {
                /*
                 * There are still some threads linked to the transport.  They
                 * are very likely sleeping in svc_poll().  We could wake up
                 * them by broadcasting on the p_req_cv condition variable, but
                 * that might give us a performance penalty if there are too
                 * many sleeping threads.
                 *
                 * Instead, we do nothing here.  The linked threads will unlink
                 * themselves and destroy the transport once they are woken up
                 * on timeout, or by new request.  There is no reason to hurry
                 * up now with the thread wake up.
                 */

                /*
                 *  NOTICE: No references to the master transport structure
                 *          beyond this point!
                 */
                mutex_exit(&xprt->xp_thread_lock);
        }
}

/*
 * Interrupt `request delivery' routine called from rpcmod
 * - put a request at the tail of the transport request queue
 * - insert a hint for svc_poll() into the xprt-ready queue
 * - increment the `pending-requests' count for the pool
 * - handle flow control
 * - wake up a thread sleeping in svc_poll() if necessary
 * - if all the threads are running ask the creator for a new one.
 */
bool_t
svc_queuereq(queue_t *q, mblk_t *mp, bool_t flowcontrol)
{
        SVCMASTERXPRT *xprt = ((void **) q->q_ptr)[0];
        SVCPOOL *pool = xprt->xp_pool;
        size_t size;

        TRACE_0(TR_FAC_KRPC, TR_SVC_QUEUEREQ_START, "svc_queuereq_start");

        ASSERT(!is_system_labeled() || msg_getcred(mp, NULL) != NULL ||
            mp->b_datap->db_type != M_DATA);

        /*
         * Step 1.
         * Grab the transport's request lock and the
         * pool's request lock so that when we put
         * the request at the tail of the transport's
         * request queue, possibly put the request on
         * the xprt ready queue and increment the
         * pending request count it looks atomic.
         */
        mutex_enter(&xprt->xp_req_lock);
        if (flowcontrol && xprt->xp_full) {
                mutex_exit(&xprt->xp_req_lock);

                return (FALSE);
        }
        ASSERT(xprt->xp_full == FALSE);
        mutex_enter(&pool->p_req_lock);
        if (xprt->xp_req_head == NULL)
                xprt->xp_req_head = mp;
        else
                xprt->xp_req_tail->b_next = mp;
        xprt->xp_req_tail = mp;

        /*
         * Step 2.
         * Insert a hint into the xprt-ready queue, increment
         * counters, handle flow control, and wake up
         * a thread sleeping in svc_poll() if necessary.
         */

        /* Insert pointer to this transport into the xprt-ready queue */
        svc_xprt_qput(pool, xprt);

        /* Increment counters */
        pool->p_reqs++;
        xprt->xp_reqs++;

        size = svc_msgsize(mp);
        xprt->xp_size += size;
        pool->p_size += size;

        /* Handle flow control */
        if (flowcontrol)
                svc_flowcontrol(xprt);

        TRACE_2(TR_FAC_KRPC, TR_NFSFP_QUE_REQ_ENQ,
            "rpc_que_req_enq:pool %p mp %p", pool, mp);

        /*
         * If there are more requests and req_cv hasn't
         * been signaled yet then wake up one more thread now.
         *
         * We avoid signaling req_cv until the most recently
         * signaled thread wakes up and gets CPU to clear
         * the `drowsy' flag.
         */
        if (pool->p_drowsy || pool->p_reqs <= pool->p_walkers ||
            pool->p_asleep == 0) {
                mutex_exit(&pool->p_req_lock);
        } else {
                pool->p_drowsy = TRUE;
                pool->p_asleep--;

                /*
                 * Signal wakeup and drop the request lock.
                 */
                cv_signal(&pool->p_req_cv);
                mutex_exit(&pool->p_req_lock);
        }
        mutex_exit(&xprt->xp_req_lock);

        /*
         * Step 3.
         * If there are no asleep/signaled threads, we are
         * still below pool->p_maxthreads limit, and no thread is
         * currently being created then signal the creator
         * for one more service thread.
         *
         * The asleep and drowsy checks are not not protected
         * by a lock since it hurts performance and a wrong
         * decision is not essential.
         */
        if (pool->p_asleep == 0 && !pool->p_drowsy &&
            pool->p_threads + pool->p_detached_threads < pool->p_maxthreads)
                svc_creator_signal(pool);

        TRACE_1(TR_FAC_KRPC, TR_SVC_QUEUEREQ_END,
            "svc_queuereq_end:(%S)", "end");

        return (TRUE);
}

/*
 * Reserve a service thread so that it can be detached later.
 * This reservation is required to make sure that when it tries to
 * detach itself the total number of detached threads does not exceed
 * pool->p_maxthreads - pool->p_redline (i.e. that we can have
 * up to pool->p_redline non-detached threads).
 *
 * If the thread does not detach itself later, it should cancel the
 * reservation before returning to svc_run().
 *
 * - check if there is room for more reserved/detached threads
 * - if so, then increment the `reserved threads' count for the pool
 * - mark the thread as reserved (setting the flag in the clone transport
 *   handle for this thread
 * - returns 1 if the reservation succeeded, 0 if it failed.
 */
int
svc_reserve_thread(SVCXPRT *clone_xprt)
{
        SVCPOOL *pool = clone_xprt->xp_master->xp_pool;

        /* Recursive reservations are not allowed */
        ASSERT(!clone_xprt->xp_reserved);
        ASSERT(!clone_xprt->xp_detached);

        /* Check pool counts if there is room for reservation */
        mutex_enter(&pool->p_thread_lock);
        if (pool->p_reserved_threads + pool->p_detached_threads >=
            pool->p_maxthreads - pool->p_redline) {
                mutex_exit(&pool->p_thread_lock);
                return (0);
        }
        pool->p_reserved_threads++;
        mutex_exit(&pool->p_thread_lock);

        /* Mark the thread (clone handle) as reserved */
        clone_xprt->xp_reserved = TRUE;

        return (1);
}

/*
 * Cancel a reservation for a thread.
 * - decrement the `reserved threads' count for the pool
 * - clear the flag in the clone transport handle for this thread.
 */
void
svc_unreserve_thread(SVCXPRT *clone_xprt)
{
        SVCPOOL *pool = clone_xprt->xp_master->xp_pool;

        /* Thread must have a reservation */
        ASSERT(clone_xprt->xp_reserved);
        ASSERT(!clone_xprt->xp_detached);

        /* Decrement global count */
        mutex_enter(&pool->p_thread_lock);
        pool->p_reserved_threads--;
        mutex_exit(&pool->p_thread_lock);

        /* Clear reservation flag */
        clone_xprt->xp_reserved = FALSE;
}

/*
 * Detach a thread from its transport, so that it can block for an
 * extended time.  Because the transport can be closed after the thread is
 * detached, the thread should have already sent off a reply if it was
 * going to send one.
 *
 * - decrement `non-detached threads' count and increment `detached threads'
 *   counts for the transport
 * - decrement the  `non-detached threads' and `reserved threads'
 *   counts and increment the `detached threads' count for the pool
 * - release the rpcmod slot
 * - mark the clone (thread) as detached.
 *
 * No need to return a pointer to the thread's CPR information, since
 * the thread has a userland identity.
 *
 * NOTICE: a thread must not detach itself without making a prior reservation
 *         through svc_thread_reserve().
 */
callb_cpr_t *
svc_detach_thread(SVCXPRT *clone_xprt)
{
        SVCMASTERXPRT *xprt = clone_xprt->xp_master;
        SVCPOOL *pool = xprt->xp_pool;
        bool_t enable;

        /* Thread must have a reservation */
        ASSERT(clone_xprt->xp_reserved);
        ASSERT(!clone_xprt->xp_detached);

        /* Bookkeeping for this transport */
        mutex_enter(&xprt->xp_thread_lock);
        xprt->xp_threads--;
        xprt->xp_detached_threads++;
        mutex_exit(&xprt->xp_thread_lock);

        /* Bookkeeping for the pool */
        mutex_enter(&pool->p_thread_lock);
        pool->p_threads--;
        pool->p_reserved_threads--;
        pool->p_detached_threads++;
        mutex_exit(&pool->p_thread_lock);

        /* Release an rpcmod slot for this request */
        mutex_enter(&xprt->xp_req_lock);
        enable = xprt->xp_enable;
        if (enable)
                xprt->xp_enable = FALSE;
        mutex_exit(&xprt->xp_req_lock);
        SVC_RELE(clone_xprt, NULL, enable);

        /* Mark the clone (thread) as detached */
        clone_xprt->xp_reserved = FALSE;
        clone_xprt->xp_detached = TRUE;

        return (NULL);
}

/*
 * This routine is responsible for extracting RDMA plugin master XPRT,
 * unregister from the SVCPOOL and initiate plugin specific cleanup.
 * It is passed a list/group of rdma transports as records which are
 * active in a given registered or unregistered kRPC thread pool. Its shuts
 * all active rdma transports in that pool. If the thread active on the trasport
 * happens to be last thread for that pool, it will signal the creater thread
 * to cleanup the pool and destroy the xprt in svc_queueclose()
 */
void
rdma_stop(rdma_xprt_group_t *rdma_xprts)
{
        SVCMASTERXPRT *xprt;
        rdma_xprt_record_t *curr_rec;
        queue_t *q;
        mblk_t *mp;
        int i, rtg_count;
        SVCPOOL *pool;

        if (rdma_xprts->rtg_count == 0)
                return;

        rtg_count = rdma_xprts->rtg_count;

        for (i = 0; i < rtg_count; i++) {
                curr_rec = rdma_xprts->rtg_listhead;
                rdma_xprts->rtg_listhead = curr_rec->rtr_next;
                rdma_xprts->rtg_count--;
                curr_rec->rtr_next = NULL;
                xprt = curr_rec->rtr_xprt_ptr;
                q = xprt->xp_wq;
                svc_rdma_kstop(xprt);

                mutex_enter(&xprt->xp_req_lock);
                pool = xprt->xp_pool;
                while ((mp = xprt->xp_req_head) != NULL) {
                        rdma_recv_data_t *rdp = (rdma_recv_data_t *)mp->b_rptr;

                        /* remove the request from the list */
                        xprt->xp_req_head = mp->b_next;
                        mp->b_next = (mblk_t *)0;

                        RDMA_BUF_FREE(rdp->conn, &rdp->rpcmsg);
                        RDMA_REL_CONN(rdp->conn);
                        freemsg(mp);
                }
                mutex_enter(&pool->p_req_lock);
                pool->p_reqs -= xprt->xp_reqs;
                pool->p_size -= xprt->xp_size;
                mutex_exit(&pool->p_req_lock);
                xprt->xp_reqs = 0;
                xprt->xp_size = 0;
                xprt->xp_full = FALSE;
                xprt->xp_enable = FALSE;
                mutex_exit(&xprt->xp_req_lock);
                svc_queueclose(q);
#ifdef  DEBUG
                if (rdma_check)
                        cmn_err(CE_NOTE, "rdma_stop: Exited svc_queueclose\n");
#endif
                /*
                 * Free the rdma transport record for the expunged rdma
                 * based master transport handle.
                 */
                kmem_free(curr_rec, sizeof (rdma_xprt_record_t));
                if (!rdma_xprts->rtg_listhead)
                        break;
        }
}


/*
 * rpc_msg_dup/rpc_msg_free
 * Currently only used by svc_rpcsec_gss.c but put in this file as it
 * may be useful to others in the future.
 * But future consumers should be careful cuz so far
 *   - only tested/used for call msgs (not reply)
 *   - only tested/used with call verf oa_length==0
 */
struct rpc_msg *
rpc_msg_dup(struct rpc_msg *src)
{
        struct rpc_msg *dst;
        struct opaque_auth oa_src, oa_dst;

        dst = kmem_alloc(sizeof (*dst), KM_SLEEP);

        dst->rm_xid = src->rm_xid;
        dst->rm_direction = src->rm_direction;

        dst->rm_call.cb_rpcvers = src->rm_call.cb_rpcvers;
        dst->rm_call.cb_prog = src->rm_call.cb_prog;
        dst->rm_call.cb_vers = src->rm_call.cb_vers;
        dst->rm_call.cb_proc = src->rm_call.cb_proc;

        /* dup opaque auth call body cred */
        oa_src = src->rm_call.cb_cred;

        oa_dst.oa_flavor = oa_src.oa_flavor;
        oa_dst.oa_base = kmem_alloc(oa_src.oa_length, KM_SLEEP);

        bcopy(oa_src.oa_base, oa_dst.oa_base, oa_src.oa_length);
        oa_dst.oa_length = oa_src.oa_length;

        dst->rm_call.cb_cred = oa_dst;

        /* dup or just alloc opaque auth call body verifier */
        if (src->rm_call.cb_verf.oa_length > 0) {
                oa_src = src->rm_call.cb_verf;

                oa_dst.oa_flavor = oa_src.oa_flavor;
                oa_dst.oa_base = kmem_alloc(oa_src.oa_length, KM_SLEEP);

                bcopy(oa_src.oa_base, oa_dst.oa_base, oa_src.oa_length);
                oa_dst.oa_length = oa_src.oa_length;

                dst->rm_call.cb_verf = oa_dst;
        } else {
                oa_dst.oa_flavor = -1;  /* will be set later */
                oa_dst.oa_base = kmem_alloc(MAX_AUTH_BYTES, KM_SLEEP);

                oa_dst.oa_length = 0;   /* will be set later */

                dst->rm_call.cb_verf = oa_dst;
        }
        return (dst);

error:
        kmem_free(dst->rm_call.cb_cred.oa_base, dst->rm_call.cb_cred.oa_length);
        kmem_free(dst, sizeof (*dst));
        return (NULL);
}

void
rpc_msg_free(struct rpc_msg **msg, int cb_verf_oa_length)
{
        struct rpc_msg *m = *msg;

        kmem_free(m->rm_call.cb_cred.oa_base, m->rm_call.cb_cred.oa_length);
        m->rm_call.cb_cred.oa_base = NULL;
        m->rm_call.cb_cred.oa_length = 0;

        kmem_free(m->rm_call.cb_verf.oa_base, cb_verf_oa_length);
        m->rm_call.cb_verf.oa_base = NULL;
        m->rm_call.cb_verf.oa_length = 0;

        kmem_free(m, sizeof (*m));
        m = NULL;
}

/*
 * Generally 'cr_ref' should be 1, otherwise reference is kept
 * in underlying calls, so reset it.
 */
cred_t *
svc_xprt_cred(SVCXPRT *xprt)
{
        cred_t *cr = xprt->xp_cred;

        ASSERT(cr != NULL);

        if (crgetref(cr) != 1) {
                crfree(cr);
                cr = crget();
                xprt->xp_cred = cr;
        }
        return (cr);
}