xref: /titanic_41/usr/src/uts/common/rpc/clnt_clts.c (revision 1b7435644ebe3112993d73003e61f1bf7125440b)

/*
 * 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 2009 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

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


/*
 * Implements a kernel based, client side RPC.
 */

#include <sys/param.h>
#include <sys/types.h>
#include <sys/systm.h>
#include <sys/sysmacros.h>
#include <sys/stream.h>
#include <sys/strsubr.h>
#include <sys/ddi.h>
#include <sys/tiuser.h>
#include <sys/tihdr.h>
#include <sys/t_kuser.h>
#include <sys/errno.h>
#include <sys/kmem.h>
#include <sys/debug.h>
#include <sys/kstat.h>
#include <sys/t_lock.h>
#include <sys/cmn_err.h>
#include <sys/conf.h>
#include <sys/disp.h>
#include <sys/taskq.h>
#include <sys/list.h>
#include <sys/atomic.h>
#include <sys/zone.h>
#include <netinet/in.h>
#include <rpc/types.h>
#include <rpc/xdr.h>
#include <rpc/auth.h>
#include <rpc/clnt.h>
#include <rpc/rpc_msg.h>

#include <sys/sdt.h>

static enum clnt_stat clnt_clts_kcallit(CLIENT *, rpcproc_t, xdrproc_t,
		    caddr_t, xdrproc_t, caddr_t, struct timeval);
static void	clnt_clts_kabort(CLIENT *);
static void	clnt_clts_kerror(CLIENT *, struct rpc_err *);
static bool_t	clnt_clts_kfreeres(CLIENT *, xdrproc_t, caddr_t);
static bool_t	clnt_clts_kcontrol(CLIENT *, int, char *);
static void	clnt_clts_kdestroy(CLIENT *);
static int	clnt_clts_ksettimers(CLIENT *, struct rpc_timers *,
		    struct rpc_timers *, int, void (*)(), caddr_t, uint32_t);

/*
 * Operations vector for CLTS based RPC
 */
static struct clnt_ops clts_ops = {
	clnt_clts_kcallit,	/* do rpc call */
	clnt_clts_kabort,	/* abort call */
	clnt_clts_kerror,	/* return error status */
	clnt_clts_kfreeres,	/* free results */
	clnt_clts_kdestroy,	/* destroy rpc handle */
	clnt_clts_kcontrol,	/* the ioctl() of rpc */
	clnt_clts_ksettimers	/* set retry timers */
};

/*
 * Endpoint for CLTS (INET, INET6, loopback, etc.)
 */
typedef struct endpnt_type {
	struct endpnt_type *e_next;	/* pointer to next endpoint type */
	list_t		e_pool;		/* list of available endpoints */
	list_t		e_ilist;	/* list of idle endpoints */
	struct endpnt	*e_pcurr;	/* pointer to current endpoint */
	char		e_protofmly[KNC_STRSIZE];	/* protocol family */
	dev_t		e_rdev;		/* device */
	kmutex_t	e_plock;	/* pool lock */
	kmutex_t	e_ilock;	/* idle list lock */
	timeout_id_t	e_itimer;	/* timer to dispatch the taskq */
	uint_t		e_cnt;		/* number of endpoints in the pool */
	zoneid_t	e_zoneid;	/* zoneid of endpoint type */
	kcondvar_t	e_async_cv;	/* cv for asynchronous reap threads */
	uint_t		e_async_count;	/* count of asynchronous reap threads */
} endpnt_type_t;

typedef struct endpnt {
	list_node_t	e_node;		/* link to the pool */
	list_node_t	e_idle;		/* link to the idle list */
	endpnt_type_t	*e_type;	/* back pointer to endpoint type */
	TIUSER		*e_tiptr;	/* pointer to transport endpoint */
	queue_t		*e_wq;		/* write queue */
	uint_t		e_flags;	/* endpoint flags */
	uint_t		e_ref;		/* ref count on endpoint */
	kcondvar_t	e_cv;		/* condition variable */
	kmutex_t	e_lock;		/* protects cv and flags */
	time_t		e_itime;	/* time when rele'd */
} endpnt_t;

#define	ENDPNT_ESTABLISHED	0x1	/* endpoint is established */
#define	ENDPNT_WAITING		0x2	/* thread waiting for endpoint */
#define	ENDPNT_BOUND		0x4	/* endpoint is bound */
#define	ENDPNT_STALE		0x8	/* endpoint is dead */
#define	ENDPNT_ONIDLE		0x10	/* endpoint is on the idle list */

static krwlock_t	endpnt_type_lock; /* protects endpnt_type_list */
static endpnt_type_t	*endpnt_type_list = NULL; /* list of CLTS endpoints */
static struct kmem_cache	*endpnt_cache; /* cache of endpnt_t's */
static taskq_t			*endpnt_taskq; /* endpnt_t reaper thread */
static bool_t			taskq_created; /* flag for endpnt_taskq */
static kmutex_t			endpnt_taskq_lock; /* taskq lock */
static zone_key_t		endpnt_destructor_key;

#define	DEFAULT_ENDPOINT_REAP_INTERVAL 60 /* 1 minute */
#define	DEFAULT_INTERVAL_SHIFT 30 /* 30 seconds */

/*
 * Endpoint tunables
 */
static int	clnt_clts_max_endpoints = -1;
static int	clnt_clts_hash_size = DEFAULT_HASH_SIZE;
static time_t	clnt_clts_endpoint_reap_interval = -1;
static clock_t	clnt_clts_taskq_dispatch_interval;

/*
 * Response completion hash queue
 */
static call_table_t *clts_call_ht;

/*
 * Routines for the endpoint manager
 */
static struct endpnt_type *endpnt_type_create(struct knetconfig *);
static void endpnt_type_free(struct endpnt_type *);
static int check_endpnt(struct endpnt *, struct endpnt **);
static struct endpnt *endpnt_get(struct knetconfig *, int);
static void endpnt_rele(struct endpnt *);
static void endpnt_reap_settimer(endpnt_type_t *);
static void endpnt_reap(endpnt_type_t *);
static void endpnt_reap_dispatch(void *);
static void endpnt_reclaim(zoneid_t);


/*
 * Request dipatching function.
 */
static int clnt_clts_dispatch_send(queue_t *q, mblk_t *, struct netbuf *addr,
					calllist_t *, uint_t, cred_t *);

/*
 * The size of the preserialized RPC header information.
 */
#define	CKU_HDRSIZE	20
/*
 * The initial allocation size.  It is small to reduce space requirements.
 */
#define	CKU_INITSIZE	2048
/*
 * The size of additional allocations, if required.  It is larger to
 * reduce the number of actual allocations.
 */
#define	CKU_ALLOCSIZE	8192

/*
 * Private data per rpc handle.  This structure is allocated by
 * clnt_clts_kcreate, and freed by clnt_clts_kdestroy.
 */
struct cku_private {
	CLIENT			 cku_client;	/* client handle */
	int			 cku_retrys;	/* request retrys */
	calllist_t		 cku_call;
	struct endpnt		*cku_endpnt;	/* open end point */
	struct knetconfig	 cku_config;
	struct netbuf		 cku_addr;	/* remote address */
	struct rpc_err		 cku_err;	/* error status */
	XDR			 cku_outxdr;	/* xdr stream for output */
	XDR			 cku_inxdr;	/* xdr stream for input */
	char			 cku_rpchdr[CKU_HDRSIZE + 4]; /* rpc header */
	struct cred		*cku_cred;	/* credentials */
	struct rpc_timers	*cku_timers;	/* for estimating RTT */
	struct rpc_timers	*cku_timeall;	/* for estimating RTT */
	void			 (*cku_feedback)(int, int, caddr_t);
						/* ptr to feedback rtn */
	caddr_t			 cku_feedarg;	/* argument for feedback func */
	uint32_t		 cku_xid;	/* current XID */
	bool_t			 cku_bcast;	/* RPC broadcast hint */
	int			cku_useresvport; /* Use reserved port */
	struct rpc_clts_client	*cku_stats;	/* counters for the zone */
};

static const struct rpc_clts_client {
	kstat_named_t	rccalls;
	kstat_named_t	rcbadcalls;
	kstat_named_t	rcretrans;
	kstat_named_t	rcbadxids;
	kstat_named_t	rctimeouts;
	kstat_named_t	rcnewcreds;
	kstat_named_t	rcbadverfs;
	kstat_named_t	rctimers;
	kstat_named_t	rcnomem;
	kstat_named_t	rccantsend;
} clts_rcstat_tmpl = {
	{ "calls",	KSTAT_DATA_UINT64 },
	{ "badcalls",	KSTAT_DATA_UINT64 },
	{ "retrans",	KSTAT_DATA_UINT64 },
	{ "badxids",	KSTAT_DATA_UINT64 },
	{ "timeouts",	KSTAT_DATA_UINT64 },
	{ "newcreds",	KSTAT_DATA_UINT64 },
	{ "badverfs",	KSTAT_DATA_UINT64 },
	{ "timers",	KSTAT_DATA_UINT64 },
	{ "nomem",	KSTAT_DATA_UINT64 },
	{ "cantsend",	KSTAT_DATA_UINT64 },
};

static uint_t clts_rcstat_ndata =
	sizeof (clts_rcstat_tmpl) / sizeof (kstat_named_t);

#define	RCSTAT_INCR(s, x)			\
	atomic_inc_64(&(s)->x.value.ui64)

#define	ptoh(p)		(&((p)->cku_client))
#define	htop(h)		((struct cku_private *)((h)->cl_private))

/*
 * Times to retry
 */
#define	SNDTRIES	4
#define	REFRESHES	2	/* authentication refreshes */

/*
 * The following is used to determine the global default behavior for
 * CLTS when binding to a local port.
 *
 * If the value is set to 1 the default will be to select a reserved
 * (aka privileged) port, if the value is zero the default will be to
 * use non-reserved ports.  Users of kRPC may override this by using
 * CLNT_CONTROL() and CLSET_BINDRESVPORT.
 */
static int clnt_clts_do_bindresvport = 1;

#define	BINDRESVPORT_RETRIES 5

void
clnt_clts_stats_init(zoneid_t zoneid, struct rpc_clts_client **statsp)
{
	kstat_t *ksp;
	kstat_named_t *knp;

	knp = rpcstat_zone_init_common(zoneid, "unix", "rpc_clts_client",
	    (const kstat_named_t *)&clts_rcstat_tmpl,
	    sizeof (clts_rcstat_tmpl));
	/*
	 * Backwards compatibility for old kstat clients
	 */
	ksp = kstat_create_zone("unix", 0, "rpc_client", "rpc",
	    KSTAT_TYPE_NAMED, clts_rcstat_ndata,
	    KSTAT_FLAG_VIRTUAL | KSTAT_FLAG_WRITABLE, zoneid);
	if (ksp) {
		ksp->ks_data = knp;
		kstat_install(ksp);
	}
	*statsp = (struct rpc_clts_client *)knp;
}

void
clnt_clts_stats_fini(zoneid_t zoneid, struct rpc_clts_client **statsp)
{
	rpcstat_zone_fini_common(zoneid, "unix", "rpc_clts_client");
	kstat_delete_byname_zone("unix", 0, "rpc_client", zoneid);
	kmem_free(*statsp, sizeof (clts_rcstat_tmpl));
}

/*
 * Create an rpc handle for a clts rpc connection.
 * Allocates space for the handle structure and the private data.
 */
/* ARGSUSED */
int
clnt_clts_kcreate(struct knetconfig *config, struct netbuf *addr,
	rpcprog_t pgm, rpcvers_t vers, int retrys, struct cred *cred,
	CLIENT **cl)
{
	CLIENT *h;
	struct cku_private *p;
	struct rpc_msg call_msg;
	int error;
	int plen;

	if (cl == NULL)
		return (EINVAL);

	*cl = NULL;
	error = 0;

	p = kmem_zalloc(sizeof (*p), KM_SLEEP);

	h = ptoh(p);

	/* handle */
	h->cl_ops = &clts_ops;
	h->cl_private = (caddr_t)p;
	h->cl_auth = authkern_create();

	/* call message, just used to pre-serialize below */
	call_msg.rm_xid = 0;
	call_msg.rm_direction = CALL;
	call_msg.rm_call.cb_rpcvers = RPC_MSG_VERSION;
	call_msg.rm_call.cb_prog = pgm;
	call_msg.rm_call.cb_vers = vers;

	/* private */
	clnt_clts_kinit(h, addr, retrys, cred);

	xdrmem_create(&p->cku_outxdr, p->cku_rpchdr, CKU_HDRSIZE, XDR_ENCODE);

	/* pre-serialize call message header */
	if (!xdr_callhdr(&p->cku_outxdr, &call_msg)) {
		error = EINVAL;		/* XXX */
		goto bad;
	}

	p->cku_config.knc_rdev = config->knc_rdev;
	p->cku_config.knc_semantics = config->knc_semantics;
	plen = strlen(config->knc_protofmly) + 1;
	p->cku_config.knc_protofmly = kmem_alloc(plen, KM_SLEEP);
	bcopy(config->knc_protofmly, p->cku_config.knc_protofmly, plen);
	p->cku_useresvport = -1; /* value is has not been set */

	cv_init(&p->cku_call.call_cv, NULL, CV_DEFAULT, NULL);
	mutex_init(&p->cku_call.call_lock, NULL, MUTEX_DEFAULT, NULL);

	*cl = h;
	return (0);

bad:
	auth_destroy(h->cl_auth);
	kmem_free(p->cku_addr.buf, addr->maxlen);
	kmem_free(p, sizeof (struct cku_private));

	return (error);
}

void
clnt_clts_kinit(CLIENT *h, struct netbuf *addr, int retrys, cred_t *cred)
{
	/* LINTED pointer alignment */
	struct cku_private *p = htop(h);
	struct rpcstat *rsp;

	rsp = zone_getspecific(rpcstat_zone_key, rpc_zone());
	ASSERT(rsp != NULL);

	p->cku_retrys = retrys;

	if (p->cku_addr.maxlen < addr->len) {
		if (p->cku_addr.maxlen != 0 && p->cku_addr.buf != NULL)
			kmem_free(p->cku_addr.buf, p->cku_addr.maxlen);

		p->cku_addr.buf = kmem_zalloc(addr->maxlen, KM_SLEEP);
		p->cku_addr.maxlen = addr->maxlen;
	}

	p->cku_addr.len = addr->len;
	bcopy(addr->buf, p->cku_addr.buf, addr->len);

	p->cku_cred = cred;
	p->cku_xid = 0;
	p->cku_timers = NULL;
	p->cku_timeall = NULL;
	p->cku_feedback = NULL;
	p->cku_bcast = FALSE;
	p->cku_call.call_xid = 0;
	p->cku_call.call_hash = 0;
	p->cku_call.call_notified = FALSE;
	p->cku_call.call_next = NULL;
	p->cku_call.call_prev = NULL;
	p->cku_call.call_reply = NULL;
	p->cku_call.call_wq = NULL;
	p->cku_stats = rsp->rpc_clts_client;
}

/*
 * set the timers.  Return current retransmission timeout.
 */
static int
clnt_clts_ksettimers(CLIENT *h, struct rpc_timers *t, struct rpc_timers *all,
	int minimum, void (*feedback)(int, int, caddr_t), caddr_t arg,
	uint32_t xid)
{
	/* LINTED pointer alignment */
	struct cku_private *p = htop(h);
	int value;

	p->cku_feedback = feedback;
	p->cku_feedarg = arg;
	p->cku_timers = t;
	p->cku_timeall = all;
	if (xid)
		p->cku_xid = xid;
	value = all->rt_rtxcur;
	value += t->rt_rtxcur;
	if (value < minimum)
		return (minimum);
	RCSTAT_INCR(p->cku_stats, rctimers);
	return (value);
}

/*
 * Time out back off function. tim is in HZ
 */
#define	MAXTIMO	(20 * hz)
#define	backoff(tim)	(((tim) < MAXTIMO) ? dobackoff(tim) : (tim))
#define	dobackoff(tim)	((((tim) << 1) > MAXTIMO) ? MAXTIMO : ((tim) << 1))

#define	RETRY_POLL_TIMO	30

/*
 * Call remote procedure.
 * Most of the work of rpc is done here.  We serialize what is left
 * of the header (some was pre-serialized in the handle), serialize
 * the arguments, and send it off.  We wait for a reply or a time out.
 * Timeout causes an immediate return, other packet problems may cause
 * a retry on the receive.  When a good packet is received we deserialize
 * it, and check verification.  A bad reply code will cause one retry
 * with full (longhand) credentials.
 */
enum clnt_stat
clnt_clts_kcallit_addr(CLIENT *h, rpcproc_t procnum, xdrproc_t xdr_args,
	caddr_t argsp, xdrproc_t xdr_results, caddr_t resultsp,
	struct timeval wait, struct netbuf *sin)
{
	/* LINTED pointer alignment */
	struct cku_private *p = htop(h);
	XDR *xdrs;
	int stries = p->cku_retrys;
	int refreshes = REFRESHES;	/* number of times to refresh cred */
	int round_trip;			/* time the RPC */
	int error;
	mblk_t *mp;
	mblk_t *mpdup;
	mblk_t *resp = NULL;
	mblk_t *tmp;
	calllist_t *call = &p->cku_call;
	clock_t ori_timout, timout;
	bool_t interrupted;
	enum clnt_stat status;
	struct rpc_msg reply_msg;
	enum clnt_stat re_status;
	endpnt_t *endpt;

	RCSTAT_INCR(p->cku_stats, rccalls);

	RPCLOG(2, "clnt_clts_kcallit_addr: wait.tv_sec: %ld\n", wait.tv_sec);
	RPCLOG(2, "clnt_clts_kcallit_addr: wait.tv_usec: %ld\n", wait.tv_usec);

	timout = TIMEVAL_TO_TICK(&wait);
	ori_timout = timout;

	if (p->cku_xid == 0) {
		p->cku_xid = alloc_xid();
		if (p->cku_endpnt != NULL)
			endpnt_rele(p->cku_endpnt);
		p->cku_endpnt = NULL;
	}
	call->call_zoneid = rpc_zoneid();

	mpdup = NULL;
call_again:

	if (mpdup == NULL) {

		while ((mp = allocb(CKU_INITSIZE, BPRI_LO)) == NULL) {
			if (strwaitbuf(CKU_INITSIZE, BPRI_LO)) {
				p->cku_err.re_status = RPC_SYSTEMERROR;
				p->cku_err.re_errno = ENOSR;
				goto done;
			}
		}

		xdrs = &p->cku_outxdr;
		xdrmblk_init(xdrs, mp, XDR_ENCODE, CKU_ALLOCSIZE);

		if (h->cl_auth->ah_cred.oa_flavor != RPCSEC_GSS) {
			/*
			 * Copy in the preserialized RPC header
			 * information.
			 */
			bcopy(p->cku_rpchdr, mp->b_rptr, CKU_HDRSIZE);

			/*
			 * transaction id is the 1st thing in the output
			 * buffer.
			 */
			/* LINTED pointer alignment */
			(*(uint32_t *)(mp->b_rptr)) = p->cku_xid;

			/* Skip the preserialized stuff. */
			XDR_SETPOS(xdrs, CKU_HDRSIZE);

			/* Serialize dynamic stuff into the output buffer. */
			if ((!XDR_PUTINT32(xdrs, (int32_t *)&procnum)) ||
			    (!AUTH_MARSHALL(h->cl_auth, xdrs, p->cku_cred)) ||
			    (!(*xdr_args)(xdrs, argsp))) {
				freemsg(mp);
				p->cku_err.re_status = RPC_CANTENCODEARGS;
				p->cku_err.re_errno = EIO;
				goto done;
			}
		} else {
			uint32_t *uproc = (uint32_t *)
			    &p->cku_rpchdr[CKU_HDRSIZE];
			IXDR_PUT_U_INT32(uproc, procnum);

			(*(uint32_t *)(&p->cku_rpchdr[0])) = p->cku_xid;
			XDR_SETPOS(xdrs, 0);

			/* Serialize the procedure number and the arguments. */
			if (!AUTH_WRAP(h->cl_auth, (caddr_t)p->cku_rpchdr,
			    CKU_HDRSIZE+4, xdrs, xdr_args, argsp)) {
				freemsg(mp);
				p->cku_err.re_status = RPC_CANTENCODEARGS;
				p->cku_err.re_errno = EIO;
				goto done;
			}
		}
	} else
		mp = mpdup;

	mpdup = dupmsg(mp);
	if (mpdup == NULL) {
		freemsg(mp);
		p->cku_err.re_status = RPC_SYSTEMERROR;
		p->cku_err.re_errno = ENOSR;
		goto done;
	}

	/*
	 * Grab an endpnt only if the endpoint is NULL.  We could be retrying
	 * the request and in this case we want to go through the same
	 * source port, so that the duplicate request cache may detect a
	 * retry.
	 */

	if (p->cku_endpnt == NULL)
		p->cku_endpnt = endpnt_get(&p->cku_config, p->cku_useresvport);

	if (p->cku_endpnt == NULL) {
		freemsg(mp);
		p->cku_err.re_status = RPC_SYSTEMERROR;
		p->cku_err.re_errno = ENOSR;
		goto done;
	}

	round_trip = ddi_get_lbolt();

	error = clnt_clts_dispatch_send(p->cku_endpnt->e_wq, mp,
	    &p->cku_addr, call, p->cku_xid, p->cku_cred);

	if (error != 0) {
		freemsg(mp);
		p->cku_err.re_status = RPC_CANTSEND;
		p->cku_err.re_errno = error;
		RCSTAT_INCR(p->cku_stats, rccantsend);
		goto done1;
	}

	RPCLOG(64, "clnt_clts_kcallit_addr: sent call for xid 0x%x\n",
	    p->cku_xid);

	/*
	 * There are two reasons for which we go back to to tryread.
	 *
	 * a) In case the status is RPC_PROCUNAVAIL and we sent out a
	 *    broadcast we should not get any invalid messages with the
	 *    RPC_PROCUNAVAIL error back. Some broken RPC implementations
	 *    send them and for this we have to ignore them ( as we would
	 *    have never received them ) and look for another message
	 *    which might contain the valid response because we don't know
	 *    how many broken implementations are in the network. So we are
	 *    going to loop until
	 *    - we received a valid response
	 *    - we have processed all invalid responses and
	 *	got a time out when we try to receive again a
	 *	message.
	 *
	 * b) We will jump back to tryread also in case we failed
	 *    within the AUTH_VALIDATE. In this case we should move
	 *    on and loop until we received a valid response or we
	 *    have processed all responses with broken authentication
	 *    and we got a time out when we try to receive a message.
	 */
tryread:
	mutex_enter(&call->call_lock);
	interrupted = FALSE;
	if (call->call_notified == FALSE) {
		klwp_t *lwp = ttolwp(curthread);
		clock_t cv_wait_ret = 1; /* init to > 0 */
		clock_t cv_timout = timout;

		if (lwp != NULL)
			lwp->lwp_nostop++;

		cv_timout += ddi_get_lbolt();

		if (h->cl_nosignal)
			while ((cv_wait_ret =
			    cv_timedwait(&call->call_cv,
			    &call->call_lock, cv_timout)) > 0 &&
			    call->call_notified == FALSE)
				;
		else
			while ((cv_wait_ret =
			    cv_timedwait_sig(&call->call_cv,
			    &call->call_lock, cv_timout)) > 0 &&
			    call->call_notified == FALSE)
				;

		if (cv_wait_ret == 0)
			interrupted = TRUE;

		if (lwp != NULL)
			lwp->lwp_nostop--;
	}
	resp = call->call_reply;
	call->call_reply = NULL;
	status = call->call_status;
	/*
	 * We have to reset the call_notified here. In case we have
	 * to do a retry ( e.g. in case we got a RPC_PROCUNAVAIL
	 * error ) we need to set this to false to ensure that
	 * we will wait for the next message. When the next message
	 * is going to arrive the function clnt_clts_dispatch_notify
	 * will set this to true again.
	 */
	call->call_notified = FALSE;
	call->call_status = RPC_TIMEDOUT;
	mutex_exit(&call->call_lock);

	if (status == RPC_TIMEDOUT) {
		if (interrupted) {
			/*
			 * We got interrupted, bail out
			 */
			p->cku_err.re_status = RPC_INTR;
			p->cku_err.re_errno = EINTR;
			goto done1;
		} else {
			RPCLOG(8, "clnt_clts_kcallit_addr: "
			    "request w/xid 0x%x timedout "
			    "waiting for reply\n", p->cku_xid);
#if 0 /* XXX not yet */
			/*
			 * Timeout may be due to a dead gateway. Send
			 * an ioctl downstream advising deletion of
			 * route when we reach the half-way point to
			 * timing out.
			 */
			if (stries == p->cku_retrys/2) {
				t_kadvise(p->cku_endpnt->e_tiptr,
				    (uchar_t *)p->cku_addr.buf,
				    p->cku_addr.len);
			}
#endif /* not yet */
			p->cku_err.re_status = RPC_TIMEDOUT;
			p->cku_err.re_errno = ETIMEDOUT;
			RCSTAT_INCR(p->cku_stats, rctimeouts);
			goto done1;
		}
	}

	ASSERT(resp != NULL);

	/*
	 * Prepare the message for further processing.  We need to remove
	 * the datagram header and copy the source address if necessary.  No
	 * need to verify the header since rpcmod took care of that.
	 */
	/*
	 * Copy the source address if the caller has supplied a netbuf.
	 */
	if (sin != NULL) {
		union T_primitives *pptr;

		pptr = (union T_primitives *)resp->b_rptr;
		bcopy(resp->b_rptr + pptr->unitdata_ind.SRC_offset, sin->buf,
		    pptr->unitdata_ind.SRC_length);
		sin->len = pptr->unitdata_ind.SRC_length;
	}

	/*
	 * Pop off the datagram header.
	 * It was retained in rpcmodrput().
	 */
	tmp = resp;
	resp = resp->b_cont;
	tmp->b_cont = NULL;
	freeb(tmp);

	round_trip = ddi_get_lbolt() - round_trip;
	/*
	 * Van Jacobson timer algorithm here, only if NOT a retransmission.
	 */
	if (p->cku_timers != NULL && stries == p->cku_retrys) {
		int rt;

		rt = round_trip;
		rt -= (p->cku_timers->rt_srtt >> 3);
		p->cku_timers->rt_srtt += rt;
		if (rt < 0)
			rt = - rt;
		rt -= (p->cku_timers->rt_deviate >> 2);
		p->cku_timers->rt_deviate += rt;
		p->cku_timers->rt_rtxcur =
		    (clock_t)((p->cku_timers->rt_srtt >> 2) +
		    p->cku_timers->rt_deviate) >> 1;

		rt = round_trip;
		rt -= (p->cku_timeall->rt_srtt >> 3);
		p->cku_timeall->rt_srtt += rt;
		if (rt < 0)
			rt = - rt;
		rt -= (p->cku_timeall->rt_deviate >> 2);
		p->cku_timeall->rt_deviate += rt;
		p->cku_timeall->rt_rtxcur =
		    (clock_t)((p->cku_timeall->rt_srtt >> 2) +
		    p->cku_timeall->rt_deviate) >> 1;
		if (p->cku_feedback != NULL) {
			(*p->cku_feedback)(FEEDBACK_OK, procnum,
			    p->cku_feedarg);
		}
	}

	/*
	 * Process reply
	 */
	xdrs = &(p->cku_inxdr);
	xdrmblk_init(xdrs, resp, XDR_DECODE, 0);

	reply_msg.rm_direction = REPLY;
	reply_msg.rm_reply.rp_stat = MSG_ACCEPTED;
	reply_msg.acpted_rply.ar_stat = SUCCESS;
	reply_msg.acpted_rply.ar_verf = _null_auth;
	/*
	 *  xdr_results will be done in AUTH_UNWRAP.
	 */
	reply_msg.acpted_rply.ar_results.where = NULL;
	reply_msg.acpted_rply.ar_results.proc = xdr_void;

	/*
	 * Decode and validate the response.
	 */
	if (!xdr_replymsg(xdrs, &reply_msg)) {
		p->cku_err.re_status = RPC_CANTDECODERES;
		p->cku_err.re_errno = EIO;
		(void) xdr_rpc_free_verifier(xdrs, &reply_msg);
		goto done1;
	}

	_seterr_reply(&reply_msg, &(p->cku_err));

	re_status = p->cku_err.re_status;
	if (re_status == RPC_SUCCESS) {
		/*
		 * Reply is good, check auth.
		 */
		if (!AUTH_VALIDATE(h->cl_auth,
		    &reply_msg.acpted_rply.ar_verf)) {
			p->cku_err.re_status = RPC_AUTHERROR;
			p->cku_err.re_why = AUTH_INVALIDRESP;
			RCSTAT_INCR(p->cku_stats, rcbadverfs);
			(void) xdr_rpc_free_verifier(xdrs, &reply_msg);
			goto tryread;
		}
		if (!AUTH_UNWRAP(h->cl_auth, xdrs, xdr_results, resultsp)) {
			p->cku_err.re_status = RPC_CANTDECODERES;
			p->cku_err.re_errno = EIO;
		}
		(void) xdr_rpc_free_verifier(xdrs, &reply_msg);
		goto done1;
	}
	/* set errno in case we can't recover */
	if (re_status != RPC_VERSMISMATCH &&
	    re_status != RPC_AUTHERROR && re_status != RPC_PROGVERSMISMATCH)
		p->cku_err.re_errno = EIO;
	/*
	 * Determine whether or not we're doing an RPC
	 * broadcast. Some server implementations don't
	 * follow RFC 1050, section 7.4.2 in that they
	 * don't remain silent when they see a proc
	 * they don't support. Therefore we keep trying
	 * to receive on RPC_PROCUNAVAIL, hoping to get
	 * a valid response from a compliant server.
	 */
	if (re_status == RPC_PROCUNAVAIL && p->cku_bcast) {
		(void) xdr_rpc_free_verifier(xdrs, &reply_msg);
		goto tryread;
	}
	if (re_status == RPC_AUTHERROR) {

		(void) xdr_rpc_free_verifier(xdrs, &reply_msg);
		call_table_remove(call);
		if (call->call_reply != NULL) {
			freemsg(call->call_reply);
			call->call_reply = NULL;
		}

		/*
		 * Maybe our credential need to be refreshed
		 */
		if (refreshes > 0 &&
		    AUTH_REFRESH(h->cl_auth, &reply_msg, p->cku_cred)) {
			/*
			 * The credential is refreshed. Try the request again.
			 * Even if stries == 0, we still retry as long as
			 * refreshes > 0. This prevents a soft authentication
			 * error turning into a hard one at an upper level.
			 */
			refreshes--;
			RCSTAT_INCR(p->cku_stats, rcbadcalls);
			RCSTAT_INCR(p->cku_stats, rcnewcreds);

			freemsg(mpdup);
			mpdup = NULL;
			freemsg(resp);
			resp = NULL;
			goto call_again;
		}
		/*
		 * We have used the client handle to do an AUTH_REFRESH
		 * and the RPC status may be set to RPC_SUCCESS;
		 * Let's make sure to set it to RPC_AUTHERROR.
		 */
		p->cku_err.re_status = RPC_CANTDECODERES;

		/*
		 * Map recoverable and unrecoverable
		 * authentication errors to appropriate errno
		 */
		switch (p->cku_err.re_why) {
		case AUTH_TOOWEAK:
			/*
			 * Could be an nfsportmon failure, set
			 * useresvport and try again.
			 */
			if (p->cku_useresvport != 1) {
				p->cku_useresvport = 1;

				freemsg(mpdup);
				mpdup = NULL;
				freemsg(resp);
				resp = NULL;

				endpt = p->cku_endpnt;
				if (endpt->e_tiptr != NULL) {
					mutex_enter(&endpt->e_lock);
					endpt->e_flags &= ~ENDPNT_BOUND;
					(void) t_kclose(endpt->e_tiptr, 1);
					endpt->e_tiptr = NULL;
					mutex_exit(&endpt->e_lock);

				}

				p->cku_xid = alloc_xid();
				endpnt_rele(p->cku_endpnt);
				p->cku_endpnt = NULL;
				goto call_again;
			}
			/* FALLTHRU */
		case AUTH_BADCRED:
		case AUTH_BADVERF:
		case AUTH_INVALIDRESP:
		case AUTH_FAILED:
		case RPCSEC_GSS_NOCRED:
		case RPCSEC_GSS_FAILED:
			p->cku_err.re_errno = EACCES;
			break;
		case AUTH_REJECTEDCRED:
		case AUTH_REJECTEDVERF:
		default:
			p->cku_err.re_errno = EIO;
			break;
		}
		RPCLOG(1, "clnt_clts_kcallit : authentication failed "
		    "with RPC_AUTHERROR of type %d\n",
		    p->cku_err.re_why);
		goto done;
	}

	(void) xdr_rpc_free_verifier(xdrs, &reply_msg);

done1:
	call_table_remove(call);
	if (call->call_reply != NULL) {
		freemsg(call->call_reply);
		call->call_reply = NULL;
	}
	RPCLOG(64, "clnt_clts_kcallit_addr: xid 0x%x taken off dispatch list",
	    p->cku_xid);

done:
	if (resp != NULL) {
		freemsg(resp);
		resp = NULL;
	}

	if ((p->cku_err.re_status != RPC_SUCCESS) &&
	    (p->cku_err.re_status != RPC_INTR) &&
	    (p->cku_err.re_status != RPC_UDERROR) &&
	    !IS_UNRECOVERABLE_RPC(p->cku_err.re_status)) {
		if (p->cku_feedback != NULL && stries == p->cku_retrys) {
			(*p->cku_feedback)(FEEDBACK_REXMIT1, procnum,
			    p->cku_feedarg);
		}

		timout = backoff(timout);
		if (p->cku_timeall != (struct rpc_timers *)0)
			p->cku_timeall->rt_rtxcur = timout;

		if (p->cku_err.re_status == RPC_SYSTEMERROR ||
		    p->cku_err.re_status == RPC_CANTSEND) {
			/*
			 * Errors due to lack of resources, wait a bit
			 * and try again.
			 */
			(void) delay(hz/10);
		}
		if (stries-- > 0) {
			RCSTAT_INCR(p->cku_stats, rcretrans);
			goto call_again;
		}
	}

	if (mpdup != NULL)
		freemsg(mpdup);

	if (p->cku_err.re_status != RPC_SUCCESS) {
		RCSTAT_INCR(p->cku_stats, rcbadcalls);
	}

	/*
	 * Allow the endpoint to be held by the client handle in case this
	 * RPC was not successful.  A retry may occur at a higher level and
	 * in this case we may want to send the request over the same
	 * source port.
	 * Endpoint is also released for one-way RPC: no reply, nor retransmit
	 * is expected.
	 */
	if ((p->cku_err.re_status == RPC_SUCCESS ||
	    (p->cku_err.re_status == RPC_TIMEDOUT && ori_timout == 0)) &&
	    p->cku_endpnt != NULL) {
		endpnt_rele(p->cku_endpnt);
		p->cku_endpnt = NULL;
	} else {
		DTRACE_PROBE2(clnt_clts_kcallit_done, int, p->cku_err.re_status,
		    struct endpnt *, p->cku_endpnt);
	}

	return (p->cku_err.re_status);
}

static enum clnt_stat
clnt_clts_kcallit(CLIENT *h, rpcproc_t procnum, xdrproc_t xdr_args,
	caddr_t argsp, xdrproc_t xdr_results, caddr_t resultsp,
	struct timeval wait)
{
	return (clnt_clts_kcallit_addr(h, procnum, xdr_args, argsp,
	    xdr_results, resultsp, wait, NULL));
}

/*
 * Return error info on this handle.
 */
static void
clnt_clts_kerror(CLIENT *h, struct rpc_err *err)
{
	/* LINTED pointer alignment */
	struct cku_private *p = htop(h);

	*err = p->cku_err;
}

static bool_t
clnt_clts_kfreeres(CLIENT *h, xdrproc_t xdr_res, caddr_t res_ptr)
{
	/* LINTED pointer alignment */
	struct cku_private *p = htop(h);
	XDR *xdrs;

	xdrs = &(p->cku_outxdr);
	xdrs->x_op = XDR_FREE;
	return ((*xdr_res)(xdrs, res_ptr));
}

/*ARGSUSED*/
static void
clnt_clts_kabort(CLIENT *h)
{
}

static bool_t
clnt_clts_kcontrol(CLIENT *h, int cmd, char *arg)
{
	/* LINTED pointer alignment */
	struct cku_private *p = htop(h);

	switch (cmd) {
	case CLSET_XID:
		p->cku_xid = *((uint32_t *)arg);
		return (TRUE);

	case CLGET_XID:
		*((uint32_t *)arg) = p->cku_xid;
		return (TRUE);

	case CLSET_BCAST:
		p->cku_bcast = *((uint32_t *)arg);
		return (TRUE);

	case CLGET_BCAST:
		*((uint32_t *)arg) = p->cku_bcast;
		return (TRUE);
	case CLSET_BINDRESVPORT:
		if (arg == NULL)
			return (FALSE);

		if (*(int *)arg != 1 && *(int *)arg != 0)
			return (FALSE);

		p->cku_useresvport = *(int *)arg;

		return (TRUE);

	case CLGET_BINDRESVPORT:
		if (arg == NULL)
			return (FALSE);

		*(int *)arg = p->cku_useresvport;

		return (TRUE);

	default:
		return (FALSE);
	}
}

/*
 * Destroy rpc handle.
 * Frees the space used for output buffer, private data, and handle
 * structure, and the file pointer/TLI data on last reference.
 */
static void
clnt_clts_kdestroy(CLIENT *h)
{
	/* LINTED pointer alignment */
	struct cku_private *p = htop(h);
	calllist_t *call = &p->cku_call;

	int plen;

	RPCLOG(8, "clnt_clts_kdestroy h: %p\n", (void *)h);
	RPCLOG(8, "clnt_clts_kdestroy h: xid=0x%x\n", p->cku_xid);

	if (p->cku_endpnt != NULL)
		endpnt_rele(p->cku_endpnt);

	cv_destroy(&call->call_cv);
	mutex_destroy(&call->call_lock);

	plen = strlen(p->cku_config.knc_protofmly) + 1;
	kmem_free(p->cku_config.knc_protofmly, plen);
	kmem_free(p->cku_addr.buf, p->cku_addr.maxlen);
	kmem_free(p, sizeof (*p));
}

/*
 * The connectionless (CLTS) kRPC endpoint management subsystem.
 *
 * Because endpoints are potentially shared among threads making RPC calls,
 * they are managed in a pool according to type (endpnt_type_t).  Each
 * endpnt_type_t points to a list of usable endpoints through the e_pool
 * field, which is of type list_t.  list_t is a doubly-linked list.
 * The number of endpoints in the pool is stored in the e_cnt field of
 * endpnt_type_t and the endpoints are reference counted using the e_ref field
 * in the endpnt_t structure.
 *
 * As an optimization, endpoints that have no references are also linked
 * to an idle list via e_ilist which is also of type list_t.  When a thread
 * calls endpnt_get() to obtain a transport endpoint, the idle list is first
 * consulted and if such an endpoint exists, it is removed from the idle list
 * and returned to the caller.
 *
 * If the idle list is empty, then a check is made to see if more endpoints
 * can be created.  If so, we proceed and create a new endpoint which is added
 * to the pool and returned to the caller.  If we have reached the limit and
 * cannot make a new endpoint then one is returned to the caller via round-
 * robin policy.
 *
 * When an endpoint is placed on the idle list by a thread calling
 * endpnt_rele(), it is timestamped and then a reaper taskq is scheduled to
 * be dispatched if one hasn't already been.  When the timer fires, the
 * taskq traverses the idle list and checks to see which endpoints are
 * eligible to be closed.  It determines this by checking if the timestamp
 * when the endpoint was released has exceeded the the threshold for how long
 * it should stay alive.
 *
 * endpnt_t structures remain persistent until the memory reclaim callback,
 * endpnt_reclaim(), is invoked.
 *
 * Here is an example of how the data structures would be laid out by the
 * subsystem:
 *
 *       endpnt_type_t
 *
 *	 loopback		                  inet
 *	 _______________	                  ______________
 *	| e_next        |----------------------->| e_next       |---->>
 *	| e_pool        |<---+                   | e_pool       |<----+
 *	| e_ilist       |<---+--+                | e_ilist      |<----+--+
 *   +->| e_pcurr       |----+--+--+	      +->| e_pcurr      |-----+--+--+
 *   |	| ...           |    |  |  |	      |	 | ...	        |     |  |  |
 *   |	| e_itimer (90) |    |  |  |	      |	 | e_itimer (0) |     |  |  |
 *   |	| e_cnt (1)     |    |  |  |	      |	 | e_cnt (3)    |     |  |  |
 *   |	+---------------+    |  |  |	      |	 +--------------+     |  |  |
 *   |			     |  |  |	      |			      |  |  |
 *   |   endpnt_t            |  |  |          |	                      |  |  |
 *   |	 ____________        |  |  |	      |	  ____________        |  |  |
 *   |	| e_node     |<------+  |  |	      |	 | e_node     |<------+  |  |
 *   |	| e_idle     |<---------+  |	      |	 | e_idle     |       |  |  |
 *   +--| e_type     |<------------+	      +--| e_type     |       |  |  |
 *	| e_tiptr    |                        |  | e_tiptr    |       |  |  |
 *      | ...	     |		              |	 | ...	      |       |  |  |
 *	| e_lock     |		              |	 | e_lock     |       |  |  |
 *	| ...        |		              |	 | ...	      |       |  |  |
 *      | e_ref (0)  |		              |	 | e_ref (2)  |       |  |  |
 *	| e_itime    |	                      |	 | e_itime    |       |  |  |
 *	+------------+		              |	 +------------+       |  |  |
 *					      |			      |  |  |
 *					      |			      |  |  |
 *					      |	  ____________        |  |  |
 *					      |	 | e_node     |<------+  |  |
 *					      |	 | e_idle     |<------+--+  |
 *					      +--| e_type     |       |     |
 *					      |	 | e_tiptr    |       |     |
 *					      |	 | ...	      |       |     |
 *					      |	 | e_lock     |       |     |
 *					      |	 | ...	      |       |     |
 *					      |	 | e_ref (0)  |       |     |
 *					      |	 | e_itime    |       |     |
 *					      |	 +------------+       |     |
 *					      |			      |     |
 *					      |			      |     |
 *					      |	  ____________        |     |
 *					      |	 | e_node     |<------+     |
 *					      |	 | e_idle     |             |
 *					      +--| e_type     |<------------+
 *						 | e_tiptr    |
 *						 | ...	      |
 *						 | e_lock     |
 *						 | ...	      |
 *						 | e_ref (1)  |
 *						 | e_itime    |
 *						 +------------+
 *
 * Endpoint locking strategy:
 *
 * The following functions manipulate lists which hold the endpoint and the
 * endpoints themselves:
 *
 * endpnt_get()/check_endpnt()/endpnt_rele()/endpnt_reap()/do_endpnt_reclaim()
 *
 * Lock description follows:
 *
 * endpnt_type_lock: Global reader/writer lock which protects accesses to the
 *		     endpnt_type_list.
 *
 * e_plock: Lock defined in the endpnt_type_t.  It is intended to
 *	    protect accesses to the pool of endopints (e_pool) for a given
 *	    endpnt_type_t.
 *
 * e_ilock: Lock defined in endpnt_type_t.  It is intended to protect accesses
 *	    to the idle list (e_ilist) of available endpoints for a given
 *	    endpnt_type_t.  It also protects access to the e_itimer, e_async_cv,
 *	    and e_async_count fields in endpnt_type_t.
 *
 * e_lock: Lock defined in the endpnt structure.  It is intended to protect
 *	   flags, cv, and ref count.
 *
 * The order goes as follows so as not to induce deadlock.
 *
 * endpnt_type_lock -> e_plock -> e_ilock -> e_lock
 *
 * Interaction with Zones and shutting down:
 *
 * endpnt_type_ts are uniquely identified by the (e_zoneid, e_rdev, e_protofmly)
 * tuple, which means that a zone may not reuse another zone's idle endpoints
 * without first doing a t_kclose().
 *
 * A zone's endpnt_type_ts are destroyed when a zone is shut down; e_async_cv
 * and e_async_count are used to keep track of the threads in endpnt_taskq
 * trying to reap endpnt_ts in the endpnt_type_t.
 */

/*
 * Allocate and initialize an endpnt_type_t
 */
static struct endpnt_type *
endpnt_type_create(struct knetconfig *config)
{
	struct endpnt_type	*etype;

	/*
	 * Allocate a new endpoint type to hang a list of
	 * endpoints off of it.
	 */
	etype = kmem_alloc(sizeof (struct endpnt_type), KM_SLEEP);
	etype->e_next = NULL;
	etype->e_pcurr = NULL;
	etype->e_itimer = 0;
	etype->e_cnt = 0;

	(void) strncpy(etype->e_protofmly, config->knc_protofmly, KNC_STRSIZE);
	mutex_init(&etype->e_plock, NULL, MUTEX_DEFAULT, NULL);
	mutex_init(&etype->e_ilock, NULL, MUTEX_DEFAULT, NULL);
	etype->e_rdev = config->knc_rdev;
	etype->e_zoneid = rpc_zoneid();
	etype->e_async_count = 0;
	cv_init(&etype->e_async_cv, NULL, CV_DEFAULT, NULL);

	list_create(&etype->e_pool, sizeof (endpnt_t),
	    offsetof(endpnt_t, e_node));
	list_create(&etype->e_ilist, sizeof (endpnt_t),
	    offsetof(endpnt_t, e_idle));

	/*
	 * Check to see if we need to create a taskq for endpoint
	 * reaping
	 */
	mutex_enter(&endpnt_taskq_lock);
	if (taskq_created == FALSE) {
		taskq_created = TRUE;
		mutex_exit(&endpnt_taskq_lock);
		ASSERT(endpnt_taskq == NULL);
		endpnt_taskq = taskq_create("clts_endpnt_taskq", 1,
		    minclsyspri, 200, INT_MAX, 0);
	} else
		mutex_exit(&endpnt_taskq_lock);

	return (etype);
}

/*
 * Free an endpnt_type_t
 */
static void
endpnt_type_free(struct endpnt_type *etype)
{
	mutex_destroy(&etype->e_plock);
	mutex_destroy(&etype->e_ilock);
	list_destroy(&etype->e_pool);
	list_destroy(&etype->e_ilist);
	kmem_free(etype, sizeof (endpnt_type_t));
}

/*
 * Check the endpoint to ensure that it is suitable for use.
 *
 * Possible return values:
 *
 * return (1) - Endpoint is established, but needs to be re-opened.
 * return (0) && *newp == NULL - Endpoint is established, but unusable.
 * return (0) && *newp != NULL - Endpoint is established and usable.
 */
static int
check_endpnt(struct endpnt *endp, struct endpnt **newp)
{
	*newp = endp;

	mutex_enter(&endp->e_lock);
	ASSERT(endp->e_ref >= 1);

	/*
	 * The first condition we check for is if the endpoint has been
	 * allocated, but is unusable either because it has been closed or
	 * has been marked stale.  Only *one* thread will be allowed to
	 * execute the then clause.  This is enforced because the first thread
	 * to check this condition will clear the flags, so that subsequent
	 * thread(s) checking this endpoint will move on.
	 */
	if ((endp->e_flags & ENDPNT_ESTABLISHED) &&
	    (!(endp->e_flags & ENDPNT_BOUND) ||
	    (endp->e_flags & ENDPNT_STALE))) {
		/*
		 * Clear the flags here since they will be
		 * set again by this thread.  They need to be
		 * individually cleared because we want to maintain
		 * the state for ENDPNT_ONIDLE.
		 */
		endp->e_flags &= ~(ENDPNT_ESTABLISHED |
		    ENDPNT_WAITING | ENDPNT_BOUND | ENDPNT_STALE);
		mutex_exit(&endp->e_lock);
		return (1);
	}

	/*
	 * The second condition is meant for any thread that is waiting for
	 * an endpoint to become established.  It will cv_wait() until
	 * the condition for the endpoint has been changed to ENDPNT_BOUND or
	 * ENDPNT_STALE.
	 */
	while (!(endp->e_flags & ENDPNT_BOUND) &&
	    !(endp->e_flags & ENDPNT_STALE)) {
		endp->e_flags |= ENDPNT_WAITING;
		cv_wait(&endp->e_cv, &endp->e_lock);
	}

	ASSERT(endp->e_flags & ENDPNT_ESTABLISHED);

	/*
	 * The last case we check for is if the endpoint has been marked stale.
	 * If this is the case then set *newp to NULL and return, so that the
	 * caller is notified of the error and can take appropriate action.
	 */
	if (endp->e_flags & ENDPNT_STALE) {
		endp->e_ref--;
		*newp = NULL;
	}
	mutex_exit(&endp->e_lock);
	return (0);
}

#ifdef DEBUG
/*
 * Provide a fault injection setting to test error conditions.
 */
static int endpnt_get_return_null = 0;
#endif

/*
 * Returns a handle (struct endpnt *) to an open and bound endpoint
 * specified by the knetconfig passed in.  Returns NULL if no valid endpoint
 * can be obtained.
 */
static struct endpnt *
endpnt_get(struct knetconfig *config, int useresvport)
{
	struct endpnt_type	*n_etype = NULL;
	struct endpnt_type	*np = NULL;
	struct endpnt		*new = NULL;
	struct endpnt		*endp = NULL;
	struct endpnt		*next = NULL;
	TIUSER			*tiptr = NULL;
	int			rtries = BINDRESVPORT_RETRIES;
	int			i = 0;
	int			error;
	int			retval;
	zoneid_t		zoneid = rpc_zoneid();
	cred_t			*cr;

	RPCLOG(1, "endpnt_get: protofmly %s, ", config->knc_protofmly);
	RPCLOG(1, "rdev %ld\n", config->knc_rdev);

#ifdef DEBUG
	/*
	 * Inject fault if desired.  Pretend we have a stale endpoint
	 * and return NULL.
	 */
	if (endpnt_get_return_null > 0) {
		endpnt_get_return_null--;
		return (NULL);
	}
#endif
	rw_enter(&endpnt_type_lock, RW_READER);

top:
	for (np = endpnt_type_list; np != NULL; np = np->e_next)
		if ((np->e_zoneid == zoneid) &&
		    (np->e_rdev == config->knc_rdev) &&
		    (strcmp(np->e_protofmly,
		    config->knc_protofmly) == 0))
			break;

	if (np == NULL && n_etype != NULL) {
		ASSERT(rw_write_held(&endpnt_type_lock));

		/*
		 * Link the endpoint type onto the list
		 */
		n_etype->e_next = endpnt_type_list;
		endpnt_type_list = n_etype;
		np = n_etype;
		n_etype = NULL;
	}

	if (np == NULL) {
		/*
		 * The logic here is that we were unable to find an
		 * endpnt_type_t that matched our criteria, so we allocate a
		 * new one.  Because kmem_alloc() needs to be called with
		 * KM_SLEEP, we drop our locks so that we don't induce
		 * deadlock.  After allocating and initializing the
		 * endpnt_type_t, we reaquire the lock and go back to check
		 * if this entry needs to be added to the list.  Since we do
		 * some operations without any locking other threads may
		 * have been looking for the same endpnt_type_t and gone
		 * through this code path.  We check for this case and allow
		 * one thread to link its endpnt_type_t to the list and the
		 * other threads will simply free theirs.
		 */
		rw_exit(&endpnt_type_lock);
		n_etype = endpnt_type_create(config);

		/*
		 * We need to reaquire the lock with RW_WRITER here so that
		 * we can safely link the new endpoint type onto the list.
		 */
		rw_enter(&endpnt_type_lock, RW_WRITER);
		goto top;
	}

	rw_exit(&endpnt_type_lock);
	/*
	 * If n_etype is not NULL, then another thread was able to
	 * insert an endpnt_type_t of this type  onto the list before
	 * we did.  Go ahead and free ours.
	 */
	if (n_etype != NULL)
		endpnt_type_free(n_etype);

	mutex_enter(&np->e_ilock);
	/*
	 * The algorithm to hand out endpoints is to first
	 * give out those that are idle if such endpoints
	 * exist.  Otherwise, create a new one if we haven't
	 * reached the max threshold.  Finally, we give out
	 * endpoints in a pseudo LRU fashion (round-robin).
	 *
	 * Note:  The idle list is merely a hint of those endpoints
	 * that should be idle.  There exists a window after the
	 * endpoint is released and before it is linked back onto the
	 * idle list where a thread could get a reference to it and
	 * use it.  This is okay, since the reference counts will
	 * still be consistent.
	 */
	if ((endp = (endpnt_t *)list_head(&np->e_ilist)) != NULL) {
		timeout_id_t t_id = 0;

		mutex_enter(&endp->e_lock);
		endp->e_ref++;
		endp->e_itime = 0;
		endp->e_flags &= ~ENDPNT_ONIDLE;
		mutex_exit(&endp->e_lock);

		/*
		 * Pop the endpoint off the idle list and hand it off
		 */
		list_remove(&np->e_ilist, endp);

		if (np->e_itimer != 0) {
			t_id = np->e_itimer;
			np->e_itimer = 0;
		}
		mutex_exit(&np->e_ilock);
		/*
		 * Reset the idle timer if it has been set
		 */
		if (t_id != (timeout_id_t)0)
			(void) untimeout(t_id);

		if (check_endpnt(endp, &new) == 0)
			return (new);
	} else if (np->e_cnt >= clnt_clts_max_endpoints) {
		/*
		 * There are no idle endpoints currently, so
		 * create a new one if we have not reached the maximum or
		 * hand one out in round-robin.
		 */
		mutex_exit(&np->e_ilock);
		mutex_enter(&np->e_plock);
		endp = np->e_pcurr;
		mutex_enter(&endp->e_lock);
		endp->e_ref++;
		mutex_exit(&endp->e_lock);

		ASSERT(endp != NULL);
		/*
		 * Advance the pointer to the next eligible endpoint, if
		 * necessary.
		 */
		if (np->e_cnt > 1) {
			next = (endpnt_t *)list_next(&np->e_pool, np->e_pcurr);
			if (next == NULL)
				next = (endpnt_t *)list_head(&np->e_pool);
			np->e_pcurr = next;
		}

		mutex_exit(&np->e_plock);

		/*
		 * We need to check to see if this endpoint is bound or
		 * not.  If it is in progress then just wait until
		 * the set up is complete
		 */
		if (check_endpnt(endp, &new) == 0)
			return (new);
	} else {
		mutex_exit(&np->e_ilock);
		mutex_enter(&np->e_plock);

		/*
		 * Allocate a new endpoint to use.  If we can't allocate any
		 * more memory then use one that is already established if any
		 * such endpoints exist.
		 */
		new = kmem_cache_alloc(endpnt_cache, KM_NOSLEEP);
		if (new == NULL) {
			RPCLOG0(1, "endpnt_get: kmem_cache_alloc failed\n");
			/*
			 * Try to recover by using an existing endpoint.
			 */
			if (np->e_cnt <= 0) {
				mutex_exit(&np->e_plock);
				return (NULL);
			}
			endp = np->e_pcurr;
			if ((next = list_next(&np->e_pool, np->e_pcurr)) !=
			    NULL)
				np->e_pcurr = next;
			ASSERT(endp != NULL);
			mutex_enter(&endp->e_lock);
			endp->e_ref++;
			mutex_exit(&endp->e_lock);
			mutex_exit(&np->e_plock);

			if (check_endpnt(endp, &new) == 0)
				return (new);
		} else {
			/*
			 * Partially init an endpoint structure and put
			 * it on the list, so that other interested threads
			 * know that one is being created
			 */
			bzero(new, sizeof (struct endpnt));

			cv_init(&new->e_cv, NULL, CV_DEFAULT, NULL);
			mutex_init(&new->e_lock, NULL, MUTEX_DEFAULT, NULL);
			new->e_ref = 1;
			new->e_type = np;

			/*
			 * Link the endpoint into the pool.
			 */
			list_insert_head(&np->e_pool, new);
			np->e_cnt++;
			if (np->e_pcurr == NULL)
				np->e_pcurr = new;
			mutex_exit(&np->e_plock);
		}
	}

	/*
	 * The transport should be opened with sufficient privs
	 */
	cr = zone_kcred();
	error = t_kopen(NULL, config->knc_rdev, FREAD|FWRITE|FNDELAY, &tiptr,
	    cr);
	if (error) {
		RPCLOG(1, "endpnt_get: t_kopen: %d\n", error);
		goto bad;
	}

	new->e_tiptr = tiptr;
	rpc_poptimod(tiptr->fp->f_vnode);

	/*
	 * Allow the kernel to push the module on behalf of the user.
	 */
	error = strioctl(tiptr->fp->f_vnode, I_PUSH, (intptr_t)"rpcmod", 0,
	    K_TO_K, cr, &retval);
	if (error) {
		RPCLOG(1, "endpnt_get: kstr_push on rpcmod failed %d\n", error);
		goto bad;
	}

	error = strioctl(tiptr->fp->f_vnode, RPC_CLIENT, 0, 0, K_TO_K,
	    cr, &retval);
	if (error) {
		RPCLOG(1, "endpnt_get: strioctl failed %d\n", error);
		goto bad;
	}

	/*
	 * Connectionless data flow should bypass the stream head.
	 */
	new->e_wq = tiptr->fp->f_vnode->v_stream->sd_wrq->q_next;

	error = strioctl(tiptr->fp->f_vnode, I_PUSH, (intptr_t)"timod", 0,
	    K_TO_K, cr, &retval);
	if (error) {
		RPCLOG(1, "endpnt_get: kstr_push on timod failed %d\n", error);
		goto bad;
	}

	/*
	 * Attempt to bind the endpoint.  If we fail then propogate
	 * error back to calling subsystem, so that it can be handled
	 * appropriately.
	 * If the caller has not specified reserved port usage then
	 * take the system default.
	 */
	if (useresvport == -1)
		useresvport = clnt_clts_do_bindresvport;

	if (useresvport &&
	    (strcmp(config->knc_protofmly, NC_INET) == 0 ||
	    strcmp(config->knc_protofmly, NC_INET6) == 0)) {

		while ((error =
		    bindresvport(new->e_tiptr, NULL, NULL, FALSE)) != 0) {
			RPCLOG(1,
			    "endpnt_get: bindresvport error %d\n", error);
			if (error != EPROTO) {
				if (rtries-- <= 0)
					goto bad;

				delay(hz << i++);
				continue;
			}

			(void) t_kclose(new->e_tiptr, 1);
			/*
			 * reopen with all privileges
			 */
			error = t_kopen(NULL, config->knc_rdev,
			    FREAD|FWRITE|FNDELAY,
			    &new->e_tiptr, cr);
			if (error) {
				RPCLOG(1, "endpnt_get: t_kopen: %d\n", error);
					new->e_tiptr = NULL;
					goto bad;
			}
		}
	} else if ((error = t_kbind(new->e_tiptr, NULL, NULL)) != 0) {
		RPCLOG(1, "endpnt_get: t_kbind failed: %d\n", error);
		goto bad;
	}

	/*
	 * Set the flags and notify and waiters that we have an established
	 * endpoint.
	 */
	mutex_enter(&new->e_lock);
	new->e_flags |= ENDPNT_ESTABLISHED;
	new->e_flags |= ENDPNT_BOUND;
	if (new->e_flags & ENDPNT_WAITING) {
		cv_broadcast(&new->e_cv);
		new->e_flags &= ~ENDPNT_WAITING;
	}
	mutex_exit(&new->e_lock);

	return (new);

bad:
	ASSERT(new != NULL);
	/*
	 * mark this endpoint as stale and notify any threads waiting
	 * on this endpoint that it will be going away.
	 */
	mutex_enter(&new->e_lock);
	if (new->e_ref > 0) {
		new->e_flags |= ENDPNT_ESTABLISHED;
		new->e_flags |= ENDPNT_STALE;
		if (new->e_flags & ENDPNT_WAITING) {
			cv_broadcast(&new->e_cv);
			new->e_flags &= ~ENDPNT_WAITING;
		}
	}
	new->e_ref--;
	new->e_tiptr = NULL;
	mutex_exit(&new->e_lock);

	/*
	 * If there was a transport endopoint opened, then close it.
	 */
	if (tiptr != NULL)
		(void) t_kclose(tiptr, 1);

	return (NULL);
}

/*
 * Release a referece to the endpoint
 */
static void
endpnt_rele(struct endpnt *sp)
{
	mutex_enter(&sp->e_lock);
	ASSERT(sp->e_ref > 0);
	sp->e_ref--;
	/*
	 * If the ref count is zero, then start the idle timer and link
	 * the endpoint onto the idle list.
	 */
	if (sp->e_ref == 0) {
		sp->e_itime = gethrestime_sec();

		/*
		 * Check to see if the endpoint is already linked to the idle
		 * list, so that we don't try to reinsert it.
		 */
		if (sp->e_flags & ENDPNT_ONIDLE) {
			mutex_exit(&sp->e_lock);
			mutex_enter(&sp->e_type->e_ilock);
			endpnt_reap_settimer(sp->e_type);
			mutex_exit(&sp->e_type->e_ilock);
			return;
		}

		sp->e_flags |= ENDPNT_ONIDLE;
		mutex_exit(&sp->e_lock);
		mutex_enter(&sp->e_type->e_ilock);
		list_insert_tail(&sp->e_type->e_ilist, sp);
		endpnt_reap_settimer(sp->e_type);
		mutex_exit(&sp->e_type->e_ilock);
	} else
		mutex_exit(&sp->e_lock);
}

static void
endpnt_reap_settimer(endpnt_type_t *etp)
{
	if (etp->e_itimer == (timeout_id_t)0)
		etp->e_itimer = timeout(endpnt_reap_dispatch, (void *)etp,
		    clnt_clts_taskq_dispatch_interval);
}

static void
endpnt_reap_dispatch(void *a)
{
	endpnt_type_t *etp = a;

	/*
	 * The idle timer has fired, so dispatch the taskq to close the
	 * endpoint.
	 */
	if (taskq_dispatch(endpnt_taskq, (task_func_t *)endpnt_reap, etp,
	    TQ_NOSLEEP) == NULL)
		return;
	mutex_enter(&etp->e_ilock);
	etp->e_async_count++;
	mutex_exit(&etp->e_ilock);
}

/*
 * Traverse the idle list and close those endpoints that have reached their
 * timeout interval.
 */
static void
endpnt_reap(endpnt_type_t *etp)
{
	struct endpnt *e;
	struct endpnt *next_node = NULL;

	mutex_enter(&etp->e_ilock);
	e = list_head(&etp->e_ilist);
	while (e != NULL) {
		next_node = list_next(&etp->e_ilist, e);

		mutex_enter(&e->e_lock);
		if (e->e_ref > 0) {
			mutex_exit(&e->e_lock);
			e = next_node;
			continue;
		}

		ASSERT(e->e_ref == 0);
		if (e->e_itime > 0 &&
		    (e->e_itime + clnt_clts_endpoint_reap_interval) <
		    gethrestime_sec()) {
			e->e_flags &= ~ENDPNT_BOUND;
			(void) t_kclose(e->e_tiptr, 1);
			e->e_tiptr = NULL;
			e->e_itime = 0;
		}
		mutex_exit(&e->e_lock);
		e = next_node;
	}
	etp->e_itimer = 0;
	if (--etp->e_async_count == 0)
		cv_signal(&etp->e_async_cv);
	mutex_exit(&etp->e_ilock);
}

static void
endpnt_reclaim(zoneid_t zoneid)
{
	struct endpnt_type *np;
	struct endpnt *e;
	struct endpnt *next_node = NULL;
	list_t free_list;
	int rcnt = 0;

	list_create(&free_list, sizeof (endpnt_t), offsetof(endpnt_t, e_node));

	RPCLOG0(1, "endpnt_reclaim: reclaim callback started\n");
	rw_enter(&endpnt_type_lock, RW_READER);
	for (np = endpnt_type_list; np != NULL; np = np->e_next) {
		if (zoneid != ALL_ZONES && zoneid != np->e_zoneid)
			continue;

		mutex_enter(&np->e_plock);
		RPCLOG(1, "endpnt_reclaim: protofmly %s, ",
		    np->e_protofmly);
		RPCLOG(1, "rdev %ld\n", np->e_rdev);
		RPCLOG(1, "endpnt_reclaim: found %d endpoint(s)\n",
		    np->e_cnt);

		if (np->e_cnt == 0) {
			mutex_exit(&np->e_plock);
			continue;
		}

		/*
		 * The nice thing about maintaining an idle list is that if
		 * there are any endpoints to reclaim, they are going to be
		 * on this list.  Just go through and reap the one's that
		 * have ref counts of zero.
		 */
		mutex_enter(&np->e_ilock);
		e = list_head(&np->e_ilist);
		while (e != NULL) {
			next_node = list_next(&np->e_ilist, e);
			mutex_enter(&e->e_lock);
			if (e->e_ref > 0) {
				mutex_exit(&e->e_lock);
				e = next_node;
				continue;
			}
			ASSERT(e->e_ref == 0);
			mutex_exit(&e->e_lock);

			list_remove(&np->e_ilist, e);
			list_remove(&np->e_pool, e);
			list_insert_head(&free_list, e);

			rcnt++;
			np->e_cnt--;
			e = next_node;
		}
		mutex_exit(&np->e_ilock);
		/*
		 * Reset the current pointer to be safe
		 */
		if ((e = (struct endpnt *)list_head(&np->e_pool)) != NULL)
			np->e_pcurr = e;
		else {
			ASSERT(np->e_cnt == 0);
			np->e_pcurr = NULL;
		}

		mutex_exit(&np->e_plock);
	}
	rw_exit(&endpnt_type_lock);

	while ((e = list_head(&free_list)) != NULL) {
		list_remove(&free_list, e);
		if (e->e_tiptr != NULL)
			(void) t_kclose(e->e_tiptr, 1);

		cv_destroy(&e->e_cv);
		mutex_destroy(&e->e_lock);
		kmem_cache_free(endpnt_cache, e);
	}
	list_destroy(&free_list);
	RPCLOG(1, "endpnt_reclaim: reclaimed %d endpoint(s)\n", rcnt);
}

/*
 * Endpoint reclaim zones destructor callback routine.
 *
 * After reclaiming any cached entries, we basically go through the endpnt_type
 * list, canceling outstanding timeouts and free'ing data structures.
 */
/* ARGSUSED */
static void
endpnt_destructor(zoneid_t zoneid, void *a)
{
	struct endpnt_type **npp;
	struct endpnt_type *np;
	struct endpnt_type *free_list = NULL;
	timeout_id_t t_id = 0;
	extern void clcleanup_zone(zoneid_t);
	extern void clcleanup4_zone(zoneid_t);

	/* Make sure NFS client handles are released. */
	clcleanup_zone(zoneid);
	clcleanup4_zone(zoneid);

	endpnt_reclaim(zoneid);
	/*
	 * We don't need to be holding on to any locks across the call to
	 * endpnt_reclaim() and the code below; we know that no-one can
	 * be holding open connections for this zone (all processes and kernel
	 * threads are gone), so nothing could be adding anything to the list.
	 */
	rw_enter(&endpnt_type_lock, RW_WRITER);
	npp = &endpnt_type_list;
	while ((np = *npp) != NULL) {
		if (np->e_zoneid != zoneid) {
			npp = &np->e_next;
			continue;
		}
		mutex_enter(&np->e_plock);
		mutex_enter(&np->e_ilock);
		if (np->e_itimer != 0) {
			t_id = np->e_itimer;
			np->e_itimer = 0;
		}
		ASSERT(np->e_cnt == 0);
		ASSERT(list_head(&np->e_pool) == NULL);
		ASSERT(list_head(&np->e_ilist) == NULL);

		mutex_exit(&np->e_ilock);
		mutex_exit(&np->e_plock);

		/*
		 * untimeout() any outstanding timers that have not yet fired.
		 */
		if (t_id != (timeout_id_t)0)
			(void) untimeout(t_id);
		*npp = np->e_next;
		np->e_next = free_list;
		free_list = np;
	}
	rw_exit(&endpnt_type_lock);

	while (free_list != NULL) {
		np = free_list;
		free_list = free_list->e_next;
		/*
		 * Wait for threads in endpnt_taskq trying to reap endpnt_ts in
		 * the endpnt_type_t.
		 */
		mutex_enter(&np->e_ilock);
		while (np->e_async_count > 0)
			cv_wait(&np->e_async_cv, &np->e_ilock);
		cv_destroy(&np->e_async_cv);
		mutex_destroy(&np->e_plock);
		mutex_destroy(&np->e_ilock);
		list_destroy(&np->e_pool);
		list_destroy(&np->e_ilist);
		kmem_free(np, sizeof (endpnt_type_t));
	}
}

/*
 * Endpoint reclaim kmem callback routine.
 */
/* ARGSUSED */
static void
endpnt_repossess(void *a)
{
	/*
	 * Reclaim idle endpnt's from all zones.
	 */
	if (endpnt_taskq != NULL)
		(void) taskq_dispatch(endpnt_taskq,
		    (task_func_t *)endpnt_reclaim, (void *)ALL_ZONES,
		    TQ_NOSLEEP);
}

/*
 * RPC request dispatch routine.  Constructs a datagram message and wraps it
 * around the RPC request to pass downstream.
 */
static int
clnt_clts_dispatch_send(queue_t *q, mblk_t *mp, struct netbuf *addr,
    calllist_t *cp, uint_t xid, cred_t *cr)
{
	mblk_t *bp;
	int msgsz;
	struct T_unitdata_req *udreq;

	/*
	 * Set up the call record.
	 */
	cp->call_wq = q;
	cp->call_xid = xid;
	cp->call_status = RPC_TIMEDOUT;
	cp->call_notified = FALSE;
	RPCLOG(64,
	    "clnt_clts_dispatch_send: putting xid 0x%x on "
	    "dispatch list\n", xid);
	cp->call_hash = call_hash(xid, clnt_clts_hash_size);
	cp->call_bucket = &clts_call_ht[cp->call_hash];
	call_table_enter(cp);

	/*
	 * Construct the datagram
	 */
	msgsz = (int)TUNITDATAREQSZ;
	/*
	 * Note: if the receiver uses SCM_UCRED/getpeerucred the pid will
	 * appear as -1.
	 */
	while (!(bp = allocb_cred(msgsz + addr->len, cr, NOPID))) {
		if (strwaitbuf(msgsz + addr->len, BPRI_LO))
			return (ENOSR);
	}

	udreq = (struct T_unitdata_req *)bp->b_wptr;
	udreq->PRIM_type = T_UNITDATA_REQ;
	udreq->DEST_length = addr->len;

	if (addr->len) {
		bcopy(addr->buf, bp->b_wptr + msgsz, addr->len);
		udreq->DEST_offset = (t_scalar_t)msgsz;
		msgsz += addr->len;
	} else
		udreq->DEST_offset = 0;
	udreq->OPT_length = 0;
	udreq->OPT_offset = 0;

	bp->b_datap->db_type = M_PROTO;
	bp->b_wptr += msgsz;

	/*
	 * Link the datagram header with the actual data
	 */
	linkb(bp, mp);

	/*
	 * Send downstream.
	 */
	if (canput(cp->call_wq)) {
		put(cp->call_wq, bp);
		return (0);
	}

	return (EIO);
}

/*
 * RPC response delivery routine.  Deliver the response to the waiting
 * thread by matching the xid.
 */
void
clnt_clts_dispatch_notify(mblk_t *mp, int resp_off, zoneid_t zoneid)
{
	calllist_t *e = NULL;
	call_table_t *chtp;
	uint32_t xid;
	uint_t hash;
	unsigned char *hdr_offset;
	mblk_t *resp;

	/*
	 * If the RPC response is not contained in the same mblk as the
	 * datagram header, then move to the next mblk.
	 */
	hdr_offset = mp->b_rptr;
	resp = mp;
	if ((mp->b_wptr - (mp->b_rptr + resp_off)) == 0)
		resp = mp->b_cont;
	else
		resp->b_rptr += resp_off;

	ASSERT(resp != NULL);

	if ((IS_P2ALIGNED(resp->b_rptr, sizeof (uint32_t))) &&
	    (resp->b_wptr - resp->b_rptr) >= sizeof (xid))
		xid = *((uint32_t *)resp->b_rptr);
	else {
		int i = 0;
		unsigned char *p = (unsigned char *)&xid;
		unsigned char *rptr;
		mblk_t *tmp = resp;

		/*
		 * Copy the xid, byte-by-byte into xid.
		 */
		while (tmp) {
			rptr = tmp->b_rptr;
			while (rptr < tmp->b_wptr) {
				*p++ = *rptr++;
				if (++i >= sizeof (xid))
					goto done_xid_copy;
			}
			tmp = tmp->b_cont;
		}

		/*
		 * If we got here, we ran out of mblk space before the
		 * xid could be copied.
		 */
		ASSERT(tmp == NULL && i < sizeof (xid));

		RPCLOG0(1,
		    "clnt_dispatch_notify(clts): message less than "
		    "size of xid\n");

		freemsg(mp);
		return;
	}

done_xid_copy:

	/*
	 * Reset the read pointer back to the beginning of the protocol
	 * header if we moved it.
	 */
	if (mp->b_rptr != hdr_offset)
		mp->b_rptr = hdr_offset;

	hash = call_hash(xid, clnt_clts_hash_size);
	chtp = &clts_call_ht[hash];
	/* call_table_find returns with the hash bucket locked */
	call_table_find(chtp, xid, e);

	if (e != NULL) {
		mutex_enter(&e->call_lock);

		/*
		 * verify that the reply is coming in on
		 * the same zone that it was sent from.
		 */
		if (e->call_zoneid != zoneid) {
			mutex_exit(&e->call_lock);
			mutex_exit(&chtp->ct_lock);
			RPCLOG0(8, "clnt_dispatch_notify (clts): incorrect "
			    "zoneid\n");
			freemsg(mp);
			return;
		}

		/*
		 * found thread waiting for this reply.
		 */
		if (e->call_reply) {
			RPCLOG(8,
			    "clnt_dispatch_notify (clts): discarding old "
			    "reply for xid 0x%x\n",
			    xid);
			freemsg(e->call_reply);
		}
		e->call_notified = TRUE;
		e->call_reply = mp;
		e->call_status = RPC_SUCCESS;
		cv_signal(&e->call_cv);
		mutex_exit(&e->call_lock);
		mutex_exit(&chtp->ct_lock);
	} else {
		zone_t *zone;
		struct rpcstat *rpcstat;

		mutex_exit(&chtp->ct_lock);
		RPCLOG(8, "clnt_dispatch_notify (clts): no caller for reply "
		    "0x%x\n", xid);
		freemsg(mp);
		/*
		 * This is unfortunate, but we need to lookup the zone so we
		 * can increment its "rcbadxids" counter.
		 */
		zone = zone_find_by_id(zoneid);
		if (zone == NULL) {
			/*
			 * The zone went away...
			 */
			return;
		}
		rpcstat = zone_getspecific(rpcstat_zone_key, zone);
		if (zone_status_get(zone) >= ZONE_IS_SHUTTING_DOWN) {
			/*
			 * Not interested
			 */
			zone_rele(zone);
			return;
		}
		RCSTAT_INCR(rpcstat->rpc_clts_client, rcbadxids);
		zone_rele(zone);
	}
}

/*
 * Init routine.  Called when rpcmod is loaded.
 */
void
clnt_clts_init(void)
{
	endpnt_cache = kmem_cache_create("clnt_clts_endpnt_cache",
	    sizeof (struct endpnt), 0, NULL, NULL, endpnt_repossess, NULL,
	    NULL, 0);

	rw_init(&endpnt_type_lock, NULL, RW_DEFAULT, NULL);

	/*
	 * Perform simple bounds checking to make sure that the setting is
	 * reasonable
	 */
	if (clnt_clts_max_endpoints <= 0) {
		if (clnt_clts_do_bindresvport)
			clnt_clts_max_endpoints = RESERVED_PORTSPACE;
		else
			clnt_clts_max_endpoints = NONRESERVED_PORTSPACE;
	}

	if (clnt_clts_do_bindresvport &&
	    clnt_clts_max_endpoints > RESERVED_PORTSPACE)
		clnt_clts_max_endpoints = RESERVED_PORTSPACE;
	else if (clnt_clts_max_endpoints > NONRESERVED_PORTSPACE)
		clnt_clts_max_endpoints = NONRESERVED_PORTSPACE;

	if (clnt_clts_hash_size < DEFAULT_MIN_HASH_SIZE)
		clnt_clts_hash_size = DEFAULT_MIN_HASH_SIZE;

	/*
	 * Defer creating the taskq until rpcmod gets pushed.  If we are
	 * in diskless boot mode, rpcmod will get loaded early even before
	 * thread_create() is available.
	 */
	endpnt_taskq = NULL;
	taskq_created = FALSE;
	mutex_init(&endpnt_taskq_lock, NULL, MUTEX_DEFAULT, NULL);

	if (clnt_clts_endpoint_reap_interval < DEFAULT_ENDPOINT_REAP_INTERVAL)
		clnt_clts_endpoint_reap_interval =
		    DEFAULT_ENDPOINT_REAP_INTERVAL;

	/*
	 * Dispatch the taskq at an interval which is offset from the
	 * interval that the endpoints should be reaped.
	 */
	clnt_clts_taskq_dispatch_interval =
	    (clnt_clts_endpoint_reap_interval + DEFAULT_INTERVAL_SHIFT) * hz;

	/*
	 * Initialize the completion queue
	 */
	clts_call_ht = call_table_init(clnt_clts_hash_size);
	/*
	 * Initialize the zone destructor callback.
	 */
	zone_key_create(&endpnt_destructor_key, NULL, NULL, endpnt_destructor);
}

void
clnt_clts_fini(void)
{
	(void) zone_key_delete(endpnt_destructor_key);
}