common/rpc/rpcmod.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 2008 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */
/* Copyright (c) 1990 Mentat Inc. */

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

#pragma ident	"%Z%%M%	%I%	%E% SMI"

/*
 * Kernel RPC filtering module
 */

#include <sys/param.h>
#include <sys/types.h>
#include <sys/stream.h>
#include <sys/stropts.h>
#include <sys/tihdr.h>
#include <sys/timod.h>
#include <sys/tiuser.h>
#include <sys/debug.h>
#include <sys/signal.h>
#include <sys/pcb.h>
#include <sys/user.h>
#include <sys/errno.h>
#include <sys/cred.h>
#include <sys/policy.h>
#include <sys/inline.h>
#include <sys/cmn_err.h>
#include <sys/kmem.h>
#include <sys/file.h>
#include <sys/sysmacros.h>
#include <sys/systm.h>
#include <sys/t_lock.h>
#include <sys/ddi.h>
#include <sys/vtrace.h>
#include <sys/callb.h>
#include <sys/strsun.h>

#include <sys/strlog.h>
#include <rpc/rpc_com.h>
#include <inet/common.h>
#include <rpc/types.h>
#include <sys/time.h>
#include <rpc/xdr.h>
#include <rpc/auth.h>
#include <rpc/clnt.h>
#include <rpc/rpc_msg.h>
#include <rpc/clnt.h>
#include <rpc/svc.h>
#include <rpc/rpcsys.h>
#include <rpc/rpc_rdma.h>

/*
 * This is the loadable module wrapper.
 */
#include <sys/conf.h>
#include <sys/modctl.h>
#include <sys/syscall.h>

extern struct streamtab rpcinfo;

static struct fmodsw fsw = {
	"rpcmod",
	&rpcinfo,
	D_NEW|D_MP,
};

/*
 * Module linkage information for the kernel.
 */

static struct modlstrmod modlstrmod = {
	&mod_strmodops, "rpc interface str mod", &fsw
};

/*
 * For the RPC system call.
 */
static struct sysent rpcsysent = {
	2,
	SE_32RVAL1 | SE_ARGC | SE_NOUNLOAD,
	rpcsys
};

static struct modlsys modlsys = {
	&mod_syscallops,
	"RPC syscall",
	&rpcsysent
};

#ifdef _SYSCALL32_IMPL
static struct modlsys modlsys32 = {
	&mod_syscallops32,
	"32-bit RPC syscall",
	&rpcsysent
};
#endif /* _SYSCALL32_IMPL */

static struct modlinkage modlinkage = {
	MODREV_1,
	{
		&modlsys,
#ifdef _SYSCALL32_IMPL
		&modlsys32,
#endif
		&modlstrmod,
		NULL
	}
};

int
_init(void)
{
	int error = 0;
	callb_id_t cid;
	int status;

	svc_init();
	clnt_init();
	cid = callb_add(connmgr_cpr_reset, 0, CB_CL_CPR_RPC, "rpc");

	if (error = mod_install(&modlinkage)) {
		/*
		 * Could not install module, cleanup previous
		 * initialization work.
		 */
		clnt_fini();
		if (cid != NULL)
			(void) callb_delete(cid);

		return (error);
	}

	/*
	 * Load up the RDMA plugins and initialize the stats. Even if the
	 * plugins loadup fails, but rpcmod was successfully installed the
	 * counters still get initialized.
	 */
	rw_init(&rdma_lock, NULL, RW_DEFAULT, NULL);
	mutex_init(&rdma_modload_lock, NULL, MUTEX_DEFAULT, NULL);
	mt_kstat_init();

	/*
	 * Get our identification into ldi.  This is used for loading
	 * other modules, e.g. rpcib.
	 */
	status = ldi_ident_from_mod(&modlinkage, &rpcmod_li);
	if (status != 0) {
		cmn_err(CE_WARN, "ldi_ident_from_mod fails with %d", status);
		rpcmod_li = NULL;
	}

	return (error);
}

/*
 * The unload entry point fails, because we advertise entry points into
 * rpcmod from the rest of kRPC: rpcmod_release().
 */
int
_fini(void)
{
	return (EBUSY);
}

int
_info(struct modinfo *modinfop)
{
	return (mod_info(&modlinkage, modinfop));
}

extern int nulldev();

#define	RPCMOD_ID	2049

int rmm_open(), rmm_close();

/*
 * To save instructions, since STREAMS ignores the return value
 * from these functions, they are defined as void here. Kind of icky, but...
 */
void rmm_rput(queue_t *, mblk_t *);
void rmm_wput(queue_t *, mblk_t *);
void rmm_rsrv(queue_t *);
void rmm_wsrv(queue_t *);

int rpcmodopen(), rpcmodclose();
void rpcmodrput(), rpcmodwput();
void rpcmodrsrv(), rpcmodwsrv();

static	void	rpcmodwput_other(queue_t *, mblk_t *);
static	int	mir_close(queue_t *q);
static	int	mir_open(queue_t *q, dev_t *devp, int flag, int sflag,
		    cred_t *credp);
static	void	mir_rput(queue_t *q, mblk_t *mp);
static	void	mir_rsrv(queue_t *q);
static	void	mir_wput(queue_t *q, mblk_t *mp);
static	void	mir_wsrv(queue_t *q);

static struct module_info rpcmod_info =
	{RPCMOD_ID, "rpcmod", 0, INFPSZ, 256*1024, 1024};

/*
 * Read side has no service procedure.
 */
static struct qinit rpcmodrinit = {
	(int (*)())rmm_rput,
	(int (*)())rmm_rsrv,
	rmm_open,
	rmm_close,
	nulldev,
	&rpcmod_info,
	NULL
};

/*
 * The write put procedure is simply putnext to conserve stack space.
 * The write service procedure is not used to queue data, but instead to
 * synchronize with flow control.
 */
static struct qinit rpcmodwinit = {
	(int (*)())rmm_wput,
	(int (*)())rmm_wsrv,
	rmm_open,
	rmm_close,
	nulldev,
	&rpcmod_info,
	NULL
};
struct streamtab rpcinfo = { &rpcmodrinit, &rpcmodwinit, NULL, NULL };

struct xprt_style_ops {
	int (*xo_open)();
	int (*xo_close)();
	void (*xo_wput)();
	void (*xo_wsrv)();
	void (*xo_rput)();
	void (*xo_rsrv)();
};

static struct xprt_style_ops xprt_clts_ops = {
	rpcmodopen,
	rpcmodclose,
	rpcmodwput,
	rpcmodwsrv,
	rpcmodrput,
	NULL
};

static struct xprt_style_ops xprt_cots_ops = {
	mir_open,
	mir_close,
	mir_wput,
	mir_wsrv,
	mir_rput,
	mir_rsrv
};

/*
 * Per rpcmod "slot" data structure. q->q_ptr points to one of these.
 */
struct rpcm {
	void		*rm_krpc_cell;	/* Reserved for use by KRPC */
	struct		xprt_style_ops	*rm_ops;
	int		rm_type;	/* Client or server side stream */
#define	RM_CLOSING	0x1		/* somebody is trying to close slot */
	uint_t		rm_state;	/* state of the slot. see above */
	uint_t		rm_ref;		/* cnt of external references to slot */
	kmutex_t	rm_lock;	/* mutex protecting above fields */
	kcondvar_t	rm_cwait;	/* condition for closing */
	zoneid_t	rm_zoneid;	/* zone which pushed rpcmod */
};

struct temp_slot {
	void *cell;
	struct xprt_style_ops *ops;
	int type;
	mblk_t *info_ack;
	kmutex_t lock;
	kcondvar_t wait;
};

typedef struct mir_s {
	void	*mir_krpc_cell;	/* Reserved for KRPC use. This field */
					/* must be first in the structure. */
	struct xprt_style_ops	*rm_ops;
	int	mir_type;		/* Client or server side stream */

	mblk_t	*mir_head_mp;		/* RPC msg in progress */
		/*
		 * mir_head_mp points the first mblk being collected in
		 * the current RPC message.  Record headers are removed
		 * before data is linked into mir_head_mp.
		 */
	mblk_t	*mir_tail_mp;		/* Last mblk in mir_head_mp */
		/*
		 * mir_tail_mp points to the last mblk in the message
		 * chain starting at mir_head_mp.  It is only valid
		 * if mir_head_mp is non-NULL and is used to add new
		 * data blocks to the end of chain quickly.
		 */

	int32_t	mir_frag_len;		/* Bytes seen in the current frag */
		/*
		 * mir_frag_len starts at -4 for beginning of each fragment.
		 * When this length is negative, it indicates the number of
		 * bytes that rpcmod needs to complete the record marker
		 * header.  When it is positive or zero, it holds the number
		 * of bytes that have arrived for the current fragment and
		 * are held in mir_header_mp.
		 */

	int32_t	mir_frag_header;
		/*
		 * Fragment header as collected for the current fragment.
		 * It holds the last-fragment indicator and the number
		 * of bytes in the fragment.
		 */

	unsigned int
		mir_ordrel_pending : 1,	/* Sent T_ORDREL_REQ */
		mir_hold_inbound : 1,	/* Hold inbound messages on server */
					/* side until outbound flow control */
					/* is relieved. */
		mir_closing : 1,	/* The stream is being closed */
		mir_inrservice : 1,	/* data queued or rd srv proc running */
		mir_inwservice : 1,	/* data queued or wr srv proc running */
		mir_inwflushdata : 1,	/* flush M_DATAs when srv runs */
		/*
		 * On client streams, mir_clntreq is 0 or 1; it is set
		 * to 1 whenever a new request is sent out (mir_wput)
		 * and cleared when the timer fires (mir_timer).  If
		 * the timer fires with this value equal to 0, then the
		 * stream is considered idle and KRPC is notified.
		 */
		mir_clntreq : 1,
		/*
		 * On server streams, stop accepting messages
		 */
		mir_svc_no_more_msgs : 1,
		mir_listen_stream : 1,	/* listen end point */
		mir_unused : 1,	/* no longer used */
		mir_timer_call : 1,
		mir_junk_fill_thru_bit_31 : 21;

	int	mir_setup_complete;	/* server has initialized everything */
	timeout_id_t mir_timer_id;	/* Timer for idle checks */
	clock_t	mir_idle_timeout;	/* Allowed idle time before shutdown */
		/*
		 * This value is copied from clnt_idle_timeout or
		 * svc_idle_timeout during the appropriate ioctl.
		 * Kept in milliseconds
		 */
	clock_t	mir_use_timestamp;	/* updated on client with each use */
		/*
		 * This value is set to lbolt
		 * every time a client stream sends or receives data.
		 * Even if the timer message arrives, we don't shutdown
		 * client unless:
		 *    lbolt >= MSEC_TO_TICK(mir_idle_timeout)+mir_use_timestamp.
		 * This value is kept in HZ.
		 */

	uint_t	*mir_max_msg_sizep;	/* Reference to sanity check size */
		/*
		 * This pointer is set to &clnt_max_msg_size or
		 * &svc_max_msg_size during the appropriate ioctl.
		 */
	zoneid_t mir_zoneid;	/* zone which pushed rpcmod */
	/* Server-side fields. */
	int	mir_ref_cnt;		/* Reference count: server side only */
					/* counts the number of references */
					/* that a kernel RPC server thread */
					/* (see svc_run()) has on this rpcmod */
					/* slot. Effectively, it is the */
					/* number * of unprocessed messages */
					/* that have been passed up to the */
					/* KRPC layer */

	mblk_t	*mir_svc_pend_mp;	/* Pending T_ORDREL_IND or */
					/* T_DISCON_IND */

	/*
	 * these fields are for both client and server, but for debugging,
	 * it is easier to have these last in the structure.
	 */
	kmutex_t	mir_mutex;	/* Mutex and condvar for close */
	kcondvar_t	mir_condvar;	/* synchronization. */
	kcondvar_t	mir_timer_cv;	/* Timer routine sync. */
} mir_t;

void tmp_rput(queue_t *q, mblk_t *mp);

struct xprt_style_ops tmpops = {
	NULL,
	NULL,
	putnext,
	NULL,
	tmp_rput,
	NULL
};

void
tmp_rput(queue_t *q, mblk_t *mp)
{
	struct temp_slot *t = (struct temp_slot *)(q->q_ptr);
	struct T_info_ack *pptr;

	switch (mp->b_datap->db_type) {
	case M_PCPROTO:
		pptr = (struct T_info_ack *)mp->b_rptr;
		switch (pptr->PRIM_type) {
		case T_INFO_ACK:
			mutex_enter(&t->lock);
			t->info_ack = mp;
			cv_signal(&t->wait);
			mutex_exit(&t->lock);
			return;
		default:
			break;
		}
	default:
		break;
	}

	/*
	 * Not an info-ack, so free it. This is ok because we should
	 * not be receiving data until the open finishes: rpcmod
	 * is pushed well before the end-point is bound to an address.
	 */
	freemsg(mp);
}

int
rmm_open(queue_t *q, dev_t *devp, int flag, int sflag, cred_t *crp)
{
	mblk_t *bp;
	struct temp_slot ts, *t;
	struct T_info_ack *pptr;
	int error = 0;

	ASSERT(q != NULL);
	/*
	 * Check for re-opens.
	 */
	if (q->q_ptr) {
		TRACE_1(TR_FAC_KRPC, TR_RPCMODOPEN_END,
		    "rpcmodopen_end:(%s)", "q->qptr");
		return (0);
	}

	t = &ts;
	bzero(t, sizeof (*t));
	q->q_ptr = (void *)t;
	WR(q)->q_ptr = (void *)t;

	/*
	 * Allocate the required messages upfront.
	 */
	if ((bp = allocb(sizeof (struct T_info_req) +
	    sizeof (struct T_info_ack), BPRI_LO)) == (mblk_t *)NULL) {
		return (ENOBUFS);
	}

	mutex_init(&t->lock, NULL, MUTEX_DEFAULT, NULL);
	cv_init(&t->wait, NULL, CV_DEFAULT, NULL);

	t->ops = &tmpops;

	qprocson(q);
	bp->b_datap->db_type = M_PCPROTO;
	*(int32_t *)bp->b_wptr = (int32_t)T_INFO_REQ;
	bp->b_wptr += sizeof (struct T_info_req);
	putnext(WR(q), bp);

	mutex_enter(&t->lock);
	while (t->info_ack == NULL) {
		if (cv_wait_sig(&t->wait, &t->lock) == 0) {
			error = EINTR;
			break;
		}
	}
	mutex_exit(&t->lock);

	if (error)
		goto out;

	pptr = (struct T_info_ack *)t->info_ack->b_rptr;

	if (pptr->SERV_type == T_CLTS) {
		if ((error = rpcmodopen(q, devp, flag, sflag, crp)) == 0)
			((struct rpcm *)q->q_ptr)->rm_ops = &xprt_clts_ops;
	} else {
		if ((error = mir_open(q, devp, flag, sflag, crp)) == 0)
			((mir_t *)q->q_ptr)->rm_ops = &xprt_cots_ops;
	}

out:
	if (error)
		qprocsoff(q);

	freemsg(t->info_ack);
	mutex_destroy(&t->lock);
	cv_destroy(&t->wait);

	return (error);
}

void
rmm_rput(queue_t *q, mblk_t  *mp)
{
	(*((struct temp_slot *)q->q_ptr)->ops->xo_rput)(q, mp);
}

void
rmm_rsrv(queue_t *q)
{
	(*((struct temp_slot *)q->q_ptr)->ops->xo_rsrv)(q);
}

void
rmm_wput(queue_t *q, mblk_t *mp)
{
	(*((struct temp_slot *)q->q_ptr)->ops->xo_wput)(q, mp);
}

void
rmm_wsrv(queue_t *q)
{
	(*((struct temp_slot *)q->q_ptr)->ops->xo_wsrv)(q);
}

int
rmm_close(queue_t *q, int flag, cred_t *crp)
{
	return ((*((struct temp_slot *)q->q_ptr)->ops->xo_close)(q, flag, crp));
}

/*
 * rpcmodopen -	open routine gets called when the module gets pushed
 *		onto the stream.
 */
/*ARGSUSED*/
int
rpcmodopen(queue_t *q, dev_t *devp, int flag, int sflag, cred_t *crp)
{
	struct rpcm *rmp;

	extern void (*rpc_rele)(queue_t *, mblk_t *);
	static void rpcmod_release(queue_t *, mblk_t *);

	TRACE_0(TR_FAC_KRPC, TR_RPCMODOPEN_START, "rpcmodopen_start:");

	/*
	 * Initialize entry points to release a rpcmod slot (and an input
	 * message if supplied) and to send an output message to the module
	 * below rpcmod.
	 */
	if (rpc_rele == NULL)
		rpc_rele = rpcmod_release;

	/*
	 * Only sufficiently privileged users can use this module, and it
	 * is assumed that they will use this module properly, and NOT send
	 * bulk data from downstream.
	 */
	if (secpolicy_rpcmod_open(crp) != 0)
		return (EPERM);

	/*
	 * Allocate slot data structure.
	 */
	rmp = kmem_zalloc(sizeof (*rmp), KM_SLEEP);

	mutex_init(&rmp->rm_lock, NULL, MUTEX_DEFAULT, NULL);
	cv_init(&rmp->rm_cwait, NULL, CV_DEFAULT, NULL);
	rmp->rm_zoneid = rpc_zoneid();
	/*
	 * slot type will be set by kRPC client and server ioctl's
	 */
	rmp->rm_type = 0;

	q->q_ptr = (void *)rmp;
	WR(q)->q_ptr = (void *)rmp;

	TRACE_1(TR_FAC_KRPC, TR_RPCMODOPEN_END, "rpcmodopen_end:(%s)", "end");
	return (0);
}

/*
 * rpcmodclose - This routine gets called when the module gets popped
 * off of the stream.
 */
/*ARGSUSED*/
int
rpcmodclose(queue_t *q, int flag, cred_t *crp)
{
	struct rpcm *rmp;

	ASSERT(q != NULL);
	rmp = (struct rpcm *)q->q_ptr;

	/*
	 * Mark our state as closing.
	 */
	mutex_enter(&rmp->rm_lock);
	rmp->rm_state |= RM_CLOSING;

	/*
	 * Check and see if there are any messages on the queue.  If so, send
	 * the messages, regardless whether the downstream module is ready to
	 * accept data.
	 */
	if (rmp->rm_type == RPC_SERVER) {
		flushq(q, FLUSHDATA);

		qenable(WR(q));

		if (rmp->rm_ref) {
			mutex_exit(&rmp->rm_lock);
			/*
			 * call into SVC to clean the queue
			 */
			svc_queueclean(q);
			mutex_enter(&rmp->rm_lock);

			/*
			 * Block while there are kRPC threads with a reference
			 * to this message.
			 */
			while (rmp->rm_ref)
				cv_wait(&rmp->rm_cwait, &rmp->rm_lock);
		}

		mutex_exit(&rmp->rm_lock);

		/*
		 * It is now safe to remove this queue from the stream. No kRPC
		 * threads have a reference to the stream, and none ever will,
		 * because RM_CLOSING is set.
		 */
		qprocsoff(q);

		/* Notify kRPC that this stream is going away. */
		svc_queueclose(q);
	} else {
		mutex_exit(&rmp->rm_lock);
		qprocsoff(q);
	}

	q->q_ptr = NULL;
	WR(q)->q_ptr = NULL;
	mutex_destroy(&rmp->rm_lock);
	cv_destroy(&rmp->rm_cwait);
	kmem_free(rmp, sizeof (*rmp));
	return (0);
}

#ifdef	DEBUG
int	rpcmod_send_msg_up = 0;
int	rpcmod_send_uderr = 0;
int	rpcmod_send_dup = 0;
int	rpcmod_send_dup_cnt = 0;
#endif

/*
 * rpcmodrput -	Module read put procedure.  This is called from
 *		the module, driver, or stream head downstream.
 */
void
rpcmodrput(queue_t *q, mblk_t *mp)
{
	struct rpcm *rmp;
	union T_primitives *pptr;
	int hdrsz;

	TRACE_0(TR_FAC_KRPC, TR_RPCMODRPUT_START, "rpcmodrput_start:");

	ASSERT(q != NULL);
	rmp = (struct rpcm *)q->q_ptr;

	if (rmp->rm_type == 0) {
		freemsg(mp);
		return;
	}

#ifdef DEBUG
	if (rpcmod_send_msg_up > 0) {
		mblk_t *nmp = copymsg(mp);
		if (nmp) {
			putnext(q, nmp);
			rpcmod_send_msg_up--;
		}
	}
	if ((rpcmod_send_uderr > 0) && mp->b_datap->db_type == M_PROTO) {
		mblk_t *nmp;
		struct T_unitdata_ind *data;
		struct T_uderror_ind *ud;
		int d;
		data = (struct T_unitdata_ind *)mp->b_rptr;
		if (data->PRIM_type == T_UNITDATA_IND) {
			d = sizeof (*ud) - sizeof (*data);
			nmp = allocb(mp->b_wptr - mp->b_rptr + d, BPRI_HI);
			if (nmp) {
				ud = (struct T_uderror_ind *)nmp->b_rptr;
				ud->PRIM_type = T_UDERROR_IND;
				ud->DEST_length = data->SRC_length;
				ud->DEST_offset = data->SRC_offset + d;
				ud->OPT_length = data->OPT_length;
				ud->OPT_offset = data->OPT_offset + d;
				ud->ERROR_type = ENETDOWN;
				if (data->SRC_length) {
					bcopy(mp->b_rptr +
					    data->SRC_offset,
					    nmp->b_rptr +
					    ud->DEST_offset,
					    data->SRC_length);
				}
				if (data->OPT_length) {
					bcopy(mp->b_rptr +
					    data->OPT_offset,
					    nmp->b_rptr +
					    ud->OPT_offset,
					    data->OPT_length);
				}
				nmp->b_wptr += d;
				nmp->b_wptr += (mp->b_wptr - mp->b_rptr);
				nmp->b_datap->db_type = M_PROTO;
				putnext(q, nmp);
				rpcmod_send_uderr--;
			}
		}
	}
#endif
	switch (mp->b_datap->db_type) {
	default:
		putnext(q, mp);
		break;

	case M_PROTO:
	case M_PCPROTO:
		ASSERT((mp->b_wptr - mp->b_rptr) >= sizeof (int32_t));
		pptr = (union T_primitives *)mp->b_rptr;

		/*
		 * Forward this message to krpc if it is data.
		 */
		if (pptr->type == T_UNITDATA_IND) {
			mblk_t *nmp;

		/*
		 * Check if the module is being popped.
		 */
			mutex_enter(&rmp->rm_lock);
			if (rmp->rm_state & RM_CLOSING) {
				mutex_exit(&rmp->rm_lock);
				putnext(q, mp);
				break;
			}

			switch (rmp->rm_type) {
			case RPC_CLIENT:
				mutex_exit(&rmp->rm_lock);
				hdrsz = mp->b_wptr - mp->b_rptr;

				/*
				 * Make sure the header is sane.
				 */
				if (hdrsz < TUNITDATAINDSZ ||
				    hdrsz < (pptr->unitdata_ind.OPT_length +
				    pptr->unitdata_ind.OPT_offset) ||
				    hdrsz < (pptr->unitdata_ind.SRC_length +
				    pptr->unitdata_ind.SRC_offset)) {
					freemsg(mp);
					return;
				}

				/*
				 * Call clnt_clts_dispatch_notify, so that it
				 * can pass the message to the proper caller.
				 * Don't discard the header just yet since the
				 * client may need the sender's address.
				 */
				clnt_clts_dispatch_notify(mp, hdrsz,
				    rmp->rm_zoneid);
				return;
			case RPC_SERVER:
				/*
				 * rm_krpc_cell is exclusively used by the kRPC
				 * CLTS server
				 */
				if (rmp->rm_krpc_cell) {
#ifdef DEBUG
					/*
					 * Test duplicate request cache and
					 * rm_ref count handling by sending a
					 * duplicate every so often, if
					 * desired.
					 */
					if (rpcmod_send_dup &&
					    rpcmod_send_dup_cnt++ %
					    rpcmod_send_dup)
						nmp = copymsg(mp);
					else
						nmp = NULL;
#endif
					/*
					 * Raise the reference count on this
					 * module to prevent it from being
					 * popped before krpc generates the
					 * reply.
					 */
					rmp->rm_ref++;
					mutex_exit(&rmp->rm_lock);

					/*
					 * Submit the message to krpc.
					 */
					svc_queuereq(q, mp);
#ifdef DEBUG
					/*
					 * Send duplicate if we created one.
					 */
					if (nmp) {
						mutex_enter(&rmp->rm_lock);
						rmp->rm_ref++;
						mutex_exit(&rmp->rm_lock);
						svc_queuereq(q, nmp);
					}
#endif
				} else {
					mutex_exit(&rmp->rm_lock);
					freemsg(mp);
				}
				return;
			default:
				mutex_exit(&rmp->rm_lock);
				freemsg(mp);
				return;
			} /* end switch(rmp->rm_type) */
		} else if (pptr->type == T_UDERROR_IND) {
			mutex_enter(&rmp->rm_lock);
			hdrsz = mp->b_wptr - mp->b_rptr;

			/*
			 * Make sure the header is sane
			 */
			if (hdrsz < TUDERRORINDSZ ||
			    hdrsz < (pptr->uderror_ind.OPT_length +
			    pptr->uderror_ind.OPT_offset) ||
			    hdrsz < (pptr->uderror_ind.DEST_length +
			    pptr->uderror_ind.DEST_offset)) {
				mutex_exit(&rmp->rm_lock);
				freemsg(mp);
				return;
			}

			/*
			 * In the case where a unit data error has been
			 * received, all we need to do is clear the message from
			 * the queue.
			 */
			mutex_exit(&rmp->rm_lock);
			freemsg(mp);
			RPCLOG(32, "rpcmodrput: unitdata error received at "
			    "%ld\n", gethrestime_sec());
			return;
		} /* end else if (pptr->type == T_UDERROR_IND) */

		putnext(q, mp);
		break;
	} /* end switch (mp->b_datap->db_type) */

	TRACE_0(TR_FAC_KRPC, TR_RPCMODRPUT_END,
	    "rpcmodrput_end:");
	/*
	 * Return codes are not looked at by the STREAMS framework.
	 */
}

/*
 * write put procedure
 */
void
rpcmodwput(queue_t *q, mblk_t *mp)
{
	struct rpcm	*rmp;

	ASSERT(q != NULL);

	switch (mp->b_datap->db_type) {
		case M_PROTO:
		case M_PCPROTO:
			break;
		default:
			rpcmodwput_other(q, mp);
			return;
	}

	/*
	 * Check to see if we can send the message downstream.
	 */
	if (canputnext(q)) {
		putnext(q, mp);
		return;
	}

	rmp = (struct rpcm *)q->q_ptr;
	ASSERT(rmp != NULL);

	/*
	 * The first canputnext failed.  Try again except this time with the
	 * lock held, so that we can check the state of the stream to see if
	 * it is closing.  If either of these conditions evaluate to true
	 * then send the meesage.
	 */
	mutex_enter(&rmp->rm_lock);
	if (canputnext(q) || (rmp->rm_state & RM_CLOSING)) {
		mutex_exit(&rmp->rm_lock);
		putnext(q, mp);
	} else {
		/*
		 * canputnext failed again and the stream is not closing.
		 * Place the message on the queue and let the service
		 * procedure handle the message.
		 */
		mutex_exit(&rmp->rm_lock);
		(void) putq(q, mp);
	}
}

static void
rpcmodwput_other(queue_t *q, mblk_t *mp)
{
	struct rpcm	*rmp;
	struct iocblk	*iocp;

	rmp = (struct rpcm *)q->q_ptr;
	ASSERT(rmp != NULL);

	switch (mp->b_datap->db_type) {
		case M_IOCTL:
			iocp = (struct iocblk *)mp->b_rptr;
			ASSERT(iocp != NULL);
			switch (iocp->ioc_cmd) {
				case RPC_CLIENT:
				case RPC_SERVER:
					mutex_enter(&rmp->rm_lock);
					rmp->rm_type = iocp->ioc_cmd;
					mutex_exit(&rmp->rm_lock);
					mp->b_datap->db_type = M_IOCACK;
					qreply(q, mp);
					return;
				default:
				/*
				 * pass the ioctl downstream and hope someone
				 * down there knows how to handle it.
				 */
					putnext(q, mp);
					return;
			}
		default:
			break;
	}
	/*
	 * This is something we definitely do not know how to handle, just
	 * pass the message downstream
	 */
	putnext(q, mp);
}

/*
 * Module write service procedure. This is called by downstream modules
 * for back enabling during flow control.
 */
void
rpcmodwsrv(queue_t *q)
{
	struct rpcm	*rmp;
	mblk_t		*mp = NULL;

	rmp = (struct rpcm *)q->q_ptr;
	ASSERT(rmp != NULL);

	/*
	 * Get messages that may be queued and send them down stream
	 */
	while ((mp = getq(q)) != NULL) {
		/*
		 * Optimize the service procedure for the server-side, by
		 * avoiding a call to canputnext().
		 */
		if (rmp->rm_type == RPC_SERVER || canputnext(q)) {
			putnext(q, mp);
			continue;
		}
		(void) putbq(q, mp);
		return;
	}
}

static void
rpcmod_release(queue_t *q, mblk_t *bp)
{
	struct rpcm *rmp;

	/*
	 * For now, just free the message.
	 */
	if (bp)
		freemsg(bp);
	rmp = (struct rpcm *)q->q_ptr;

	mutex_enter(&rmp->rm_lock);
	rmp->rm_ref--;

	if (rmp->rm_ref == 0 && (rmp->rm_state & RM_CLOSING)) {
		cv_broadcast(&rmp->rm_cwait);
	}

	mutex_exit(&rmp->rm_lock);
}

/*
 * This part of rpcmod is pushed on a connection-oriented transport for use
 * by RPC.  It serves to bypass the Stream head, implements
 * the record marking protocol, and dispatches incoming RPC messages.
 */

/* Default idle timer values */
#define	MIR_CLNT_IDLE_TIMEOUT	(5 * (60 * 1000L))	/* 5 minutes */
#define	MIR_SVC_IDLE_TIMEOUT	(6 * (60 * 1000L))	/* 6 minutes */
#define	MIR_SVC_ORDREL_TIMEOUT	(10 * (60 * 1000L))	/* 10 minutes */
#define	MIR_LASTFRAG	0x80000000	/* Record marker */

#define	DLEN(mp) (mp->b_cont ? msgdsize(mp) : (mp->b_wptr - mp->b_rptr))

#define	MIR_SVC_QUIESCED(mir)	\
	(mir->mir_ref_cnt == 0 && mir->mir_inrservice == 0)

#define	MIR_CLEAR_INRSRV(mir_ptr)	{	\
	(mir_ptr)->mir_inrservice = 0;	\
	if ((mir_ptr)->mir_type == RPC_SERVER &&	\
		(mir_ptr)->mir_closing)	\
		cv_signal(&(mir_ptr)->mir_condvar);	\
}

/*
 * Don't block service procedure (and mir_close) if
 * we are in the process of closing.
 */
#define	MIR_WCANPUTNEXT(mir_ptr, write_q)	\
	(canputnext(write_q) || ((mir_ptr)->mir_svc_no_more_msgs == 1))

static int	mir_clnt_dup_request(queue_t *q, mblk_t *mp);
static void	mir_rput_proto(queue_t *q, mblk_t *mp);
static int	mir_svc_policy_notify(queue_t *q, int event);
static void	mir_svc_release(queue_t *wq, mblk_t *mp);
static void	mir_svc_start(queue_t *wq);
static void	mir_svc_idle_start(queue_t *, mir_t *);
static void	mir_svc_idle_stop(queue_t *, mir_t *);
static void	mir_svc_start_close(queue_t *, mir_t *);
static void	mir_clnt_idle_do_stop(queue_t *);
static void	mir_clnt_idle_stop(queue_t *, mir_t *);
static void	mir_clnt_idle_start(queue_t *, mir_t *);
static void	mir_wput(queue_t *q, mblk_t *mp);
static void	mir_wput_other(queue_t *q, mblk_t *mp);
static void	mir_wsrv(queue_t *q);
static	void	mir_disconnect(queue_t *, mir_t *ir);
static	int	mir_check_len(queue_t *, int32_t, mblk_t *);
static	void	mir_timer(void *);

extern void	(*mir_rele)(queue_t *, mblk_t *);
extern void	(*mir_start)(queue_t *);
extern void	(*clnt_stop_idle)(queue_t *);

clock_t	clnt_idle_timeout = MIR_CLNT_IDLE_TIMEOUT;
clock_t	svc_idle_timeout = MIR_SVC_IDLE_TIMEOUT;

/*
 * Timeout for subsequent notifications of idle connection.  This is
 * typically used to clean up after a wedged orderly release.
 */
clock_t	svc_ordrel_timeout = MIR_SVC_ORDREL_TIMEOUT; /* milliseconds */

extern	uint_t	*clnt_max_msg_sizep;
extern	uint_t	*svc_max_msg_sizep;
uint_t	clnt_max_msg_size = RPC_MAXDATASIZE;
uint_t	svc_max_msg_size = RPC_MAXDATASIZE;
uint_t	mir_krpc_cell_null;

static void
mir_timer_stop(mir_t *mir)
{
	timeout_id_t tid;

	ASSERT(MUTEX_HELD(&mir->mir_mutex));

	/*
	 * Since the mir_mutex lock needs to be released to call
	 * untimeout(), we need to make sure that no other thread
	 * can start/stop the timer (changing mir_timer_id) during
	 * that time.  The mir_timer_call bit and the mir_timer_cv
	 * condition variable are used to synchronize this.  Setting
	 * mir_timer_call also tells mir_timer() (refer to the comments
	 * in mir_timer()) that it does not need to do anything.
	 */
	while (mir->mir_timer_call)
		cv_wait(&mir->mir_timer_cv, &mir->mir_mutex);
	mir->mir_timer_call = B_TRUE;

	if ((tid = mir->mir_timer_id) != 0) {
		mir->mir_timer_id = 0;
		mutex_exit(&mir->mir_mutex);
		(void) untimeout(tid);
		mutex_enter(&mir->mir_mutex);
	}
	mir->mir_timer_call = B_FALSE;
	cv_broadcast(&mir->mir_timer_cv);
}

static void
mir_timer_start(queue_t *q, mir_t *mir, clock_t intrvl)
{
	timeout_id_t tid;

	ASSERT(MUTEX_HELD(&mir->mir_mutex));

	while (mir->mir_timer_call)
		cv_wait(&mir->mir_timer_cv, &mir->mir_mutex);
	mir->mir_timer_call = B_TRUE;

	if ((tid = mir->mir_timer_id) != 0) {
		mutex_exit(&mir->mir_mutex);
		(void) untimeout(tid);
		mutex_enter(&mir->mir_mutex);
	}
	/* Only start the timer when it is not closing. */
	if (!mir->mir_closing) {
		mir->mir_timer_id = timeout