xref: /netvirt/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 2007 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.
 */

#pragma ident	"@(#)svc.c	1.104	07/06/15 SMI"

/*
 * 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_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_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.
 *
 *   There is no longer an "inconsistent" but "safe" optimization in the
 *   svc_queuereq() code. This "inconsistent" state was leading to
 *   inconsistencies between the actual number of requests and the value
 *   of p_reqs (the total number of requests). Because of this, hangs were
 *   occurring in svc_poll() where p_reqs was greater than one and no
 *   requests were found on the request queues.
 *
 * 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/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>

#define	RQCRED_SIZE	400	/* this size is excessive */

/*
 * 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;

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

/*
 * Pointers to transport specific `rele' routines in rpcmod (set from rpcmod).
 */
void	(*rpc_rele)(queue_t *, mblk_t *) = NULL;
void	(*mir_rele)(queue_t *, mblk_t *) = NULL;

/* ARGSUSED */
void
rpc_rdma_rele(queue_t *q, mblk_t *mp)
{
}
void    (*rdma_rele)(queue_t *, mblk_t *) = rpc_rdma_rele;


/*
 * This macro picks which `rele' routine to use, based on the transport type.
 */
#define	RELE_PROC(xprt) \
	((xprt)->xp_type == T_RDMA ? rdma_rele : \
	(((xprt)->xp_type == T_CLTS) ? rpc_rele : mir_rele))

/*
 * 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 (at) 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 (at) 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 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 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 = pool->p_qend;
	__SVCXPRT_QNODE *qtop = pool->p_qtop;

	/*
	 * Delete all the references to xprt between the current
	 * position of pool->p_qend and current pool->p_qtop.
	 */
	for (;;) {
		if (q->q_xprt == xprt)
			q->q_xprt = NULL;
		if (q == qtop)
			return;
		q = q->q_next;
	}
}

/*
 * 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, if `minp' or `maxp' are not NULL,
 *   fills them 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 (at) 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 (at) 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 (at) 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((