OpenSolaris Launch

1 files changed, 1218 insertions, 0 deletions

diff --git a/usr/src/uts/common/io/bufmod.c b/usr/src/uts/common/io/bufmod.c
new file mode 100644
index 0000000000..f553c58aff
--- /dev/null
+++ b/usr/src/uts/common/io/bufmod.c
@@ -0,0 +1,1218 @@
+/*
+ * CDDL HEADER START
+ *
+ * The contents of this file are subject to the terms of the
+ * Common Development and Distribution License, Version 1.0 only
+ * (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 2004 Sun Microsystems, Inc.  All rights reserved.
+ * Use is subject to license terms.
+ */
+
+#pragma ident	"%Z%%M%	%I%	%E% SMI"
+
+/*
+ * STREAMS Buffering module
+ *
+ * This streams module collects incoming messages from modules below
+ * it on the stream and buffers them up into a smaller number of
+ * aggregated messages.  Its main purpose is to reduce overhead by
+ * cutting down on the number of read (or getmsg) calls its client
+ * user process makes.
+ *  - only M_DATA is buffered.
+ *  - multithreading assumes configured as D_MTQPAIR
+ *  - packets are lost only if flag SB_NO_HEADER is clear and buffer
+ *    allocation fails.
+ *  - in order message transmission. This is enforced for messages other
+ *    than high priority messages.
+ *  - zero length messages on the read side are not passed up the
+ *    stream but used internally for synchronization.
+ * FLAGS:
+ * - SB_NO_PROTO_CVT - no conversion of M_PROTO messages to M_DATA.
+ *   (conversion is the default for backwards compatibility
+ *    hence the negative logic).
+ * - SB_NO_HEADER - no headers in buffered data.
+ *   (adding headers is the default for backwards compatibility
+ *    hence the negative logic).
+ * - SB_DEFER_CHUNK - provides improved response time in question-answer
+ *   applications. Buffering is not enabled until the second message
+ *   is received on the read side within the sb_ticks interval.
+ *   This option will often be used in combination with flag SB_SEND_ON_WRITE.
+ * - SB_SEND_ON_WRITE - a write message results in any pending buffered read
+ *   data being immediately sent upstream.
+ * - SB_NO_DROPS - bufmod behaves transparently in flow control and propagates
+ *   the blocked flow condition downstream. If this flag is clear (default)
+ *   messages will be dropped if the upstream flow is blocked.
+ */
+
+
+#include	<sys/types.h>
+#include	<sys/errno.h>
+#include	<sys/debug.h>
+#include	<sys/stropts.h>
+#include	<sys/time.h>
+#include	<sys/stream.h>
+#include	<sys/conf.h>
+#include	<sys/ddi.h>
+#include	<sys/sunddi.h>
+#include	<sys/kmem.h>
+#include	<sys/strsun.h>
+#include	<sys/bufmod.h>
+#include	<sys/modctl.h>
+#include	<sys/isa_defs.h>
+
+/*
+ * Per-Stream state information.
+ *
+ * If sb_ticks is negative, we don't deliver chunks until they're
+ * full.  If it's zero, we deliver every packet as it arrives.  (In
+ * this case we force sb_chunk to zero, to make the implementation
+ * easier.)  Otherwise, sb_ticks gives the number of ticks in a
+ * buffering interval. The interval begins when the a read side data
+ * message is received and a timeout is not active. If sb_snap is
+ * zero, no truncation of the msg is done.
+ */
+struct sb {
+	queue_t	*sb_rq;		/* our rq */
+	mblk_t	*sb_mp;		/* partial chunk */
+	mblk_t  *sb_head;	/* pre-allocated space for the next header */
+	mblk_t	*sb_tail;	/* first mblk of last message appended */
+	uint_t	sb_mlen;	/* sb_mp length */
+	uint_t	sb_mcount;	/* input msg count in sb_mp */
+	uint_t	sb_chunk;	/* max chunk size */
+	clock_t	sb_ticks;	/* timeout interval */
+	timeout_id_t sb_timeoutid; /* qtimeout() id */
+	uint_t	sb_drops;	/* cumulative # discarded msgs */
+	uint_t	sb_snap;	/* snapshot length */
+	uint_t	sb_flags;	/* flags field */
+	uint_t	sb_state;	/* state variable */
+};
+
+/*
+ * Function prototypes.
+ */
+static	int	sbopen(queue_t *, dev_t *, int, int, cred_t *);
+static	int	sbclose(queue_t *, int, cred_t *);
+static	void	sbwput(queue_t *, mblk_t *);
+static	void	sbrput(queue_t *, mblk_t *);
+static	void	sbrsrv(queue_t *);
+static	void	sbioctl(queue_t *, mblk_t *);
+static	void	sbaddmsg(queue_t *, mblk_t *);
+static	void	sbtick(void *);
+static	void	sbclosechunk(struct sb *);
+static	void	sbsendit(queue_t *, mblk_t *);
+
+static struct module_info	sb_minfo = {
+	21,		/* mi_idnum */
+	"bufmod",	/* mi_idname */
+	0,		/* mi_minpsz */
+	INFPSZ,		/* mi_maxpsz */
+	1,		/* mi_hiwat */
+	0		/* mi_lowat */
+};
+
+static struct qinit	sb_rinit = {
+	(int (*)())sbrput,	/* qi_putp */
+	(int (*)())sbrsrv,	/* qi_srvp */
+	sbopen,			/* qi_qopen */
+	sbclose,		/* qi_qclose */
+	NULL,			/* qi_qadmin */
+	&sb_minfo,		/* qi_minfo */
+	NULL			/* qi_mstat */
+};
+
+static struct qinit	sb_winit = {
+	(int (*)())sbwput,	/* qi_putp */
+	NULL,			/* qi_srvp */
+	NULL,			/* qi_qopen */
+	NULL,			/* qi_qclose */
+	NULL,			/* qi_qadmin */
+	&sb_minfo,		/* qi_minfo */
+	NULL			/* qi_mstat */
+};
+
+static struct streamtab	sb_info = {
+	&sb_rinit,	/* st_rdinit */
+	&sb_winit,	/* st_wrinit */
+	NULL,		/* st_muxrinit */
+	NULL		/* st_muxwinit */
+};
+
+
+/*
+ * This is the loadable module wrapper.
+ */
+
+static struct fmodsw fsw = {
+	"bufmod",
+	&sb_info,
+	D_MTQPAIR | D_MP
+};
+
+/*
+ * Module linkage information for the kernel.
+ */
+
+static struct modlstrmod modlstrmod = {
+	&mod_strmodops, "streams buffer mod", &fsw
+};
+
+static struct modlinkage modlinkage = {
+	MODREV_1, &modlstrmod, NULL
+};
+
+
+int
+_init(void)
+{
+	return (mod_install(&modlinkage));
+}
+
+int
+_fini(void)
+{
+	return (mod_remove(&modlinkage));
+}
+
+int
+_info(struct modinfo *modinfop)
+{
+	return (mod_info(&modlinkage, modinfop));
+}
+
+
+/* ARGSUSED */
+static int
+sbopen(queue_t *rq, dev_t *dev, int oflag, int sflag, cred_t *crp)
+{
+	struct sb	*sbp;
+	ASSERT(rq);
+
+	if (sflag != MODOPEN)
+		return (EINVAL);
+
+	if (rq->q_ptr)
+		return (0);
+
+	/*
+	 * Allocate and initialize per-Stream structure.
+	 */
+	sbp = kmem_alloc(sizeof (struct sb), KM_SLEEP);
+	sbp->sb_rq = rq;
+	sbp->sb_ticks = -1;
+	sbp->sb_chunk = SB_DFLT_CHUNK;
+	sbp->sb_tail = sbp->sb_mp = sbp->sb_head = NULL;
+	sbp->sb_mlen = 0;
+	sbp->sb_mcount = 0;
+	sbp->sb_timeoutid = 0;
+	sbp->sb_drops = 0;
+	sbp->sb_snap = 0;
+	sbp->sb_flags = 0;
+	sbp->sb_state = 0;
+
+	rq->q_ptr = WR(rq)->q_ptr = sbp;
+
+	qprocson(rq);
+
+
+	return (0);
+}
+
+/* ARGSUSED1 */
+static int
+sbclose(queue_t *rq, int flag, cred_t *credp)
+{
+	struct	sb	*sbp = (struct sb *)rq->q_ptr;
+
+	ASSERT(sbp);
+
+	qprocsoff(rq);
+	/*
+	 * Cancel an outstanding timeout
+	 */
+	if (sbp->sb_timeoutid != 0) {
+		(void) quntimeout(rq, sbp->sb_timeoutid);
+		sbp->sb_timeoutid = 0;
+	}
+	/*
+	 * Free the current chunk.
+	 */
+	if (sbp->sb_mp) {
+		freemsg(sbp->sb_mp);
+		sbp->sb_tail = sbp->sb_mp = sbp->sb_head = NULL;
+		sbp->sb_mlen = 0;
+	}
+
+	/*
+	 * Free the per-Stream structure.
+	 */
+	kmem_free((caddr_t)sbp, sizeof (struct sb));
+	rq->q_ptr = WR(rq)->q_ptr = NULL;
+
+	return (0);
+}
+
+/*
+ * the correction factor is introduced to compensate for
+ * whatever assumptions the modules below have made about
+ * how much traffic is flowing through the stream and the fact
+ * that bufmod may be snipping messages with the sb_snap length.
+ */
+#define	SNIT_HIWAT(msgsize, fudge)	((4 * msgsize * fudge) + 512)
+#define	SNIT_LOWAT(msgsize, fudge)	((2 * msgsize * fudge) + 256)
+
+
+static void
+sbioc(queue_t *wq, mblk_t *mp)
+{
+	struct iocblk *iocp;
+	struct sb *sbp = (struct sb *)wq->q_ptr;
+	clock_t	ticks;
+	mblk_t	*mop;
+
+	iocp = (struct iocblk *)mp->b_rptr;
+
+	switch (iocp->ioc_cmd) {
+	case SBIOCGCHUNK:
+	case SBIOCGSNAP:
+	case SBIOCGFLAGS:
+	case SBIOCGTIME:
+		miocack(wq, mp, 0, 0);
+		return;
+
+	case SBIOCSTIME:
+#ifdef _SYSCALL32_IMPL
+		if ((iocp->ioc_flag & IOC_MODELS) != IOC_NATIVE) {
+			struct timeval32 *t32;
+
+			t32 = (struct timeval32 *)mp->b_cont->b_rptr;
+			if (t32->tv_sec < 0 || t32->tv_usec < 0) {
+				miocnak(wq, mp, 0, EINVAL);
+				break;
+			}
+			ticks = TIMEVAL_TO_TICK(t32);
+		} else
+#endif /* _SYSCALL32_IMPL */
+		{
+			struct timeval *tb;
+
+			tb = (struct timeval *)mp->b_cont->b_rptr;
+
+			if (tb->tv_sec < 0 || tb->tv_usec < 0) {
+				miocnak(wq, mp, 0, EINVAL);
+				break;
+			}
+			ticks = TIMEVAL_TO_TICK(tb);
+		}
+		sbp->sb_ticks = ticks;
+		if (ticks == 0)
+			sbp->sb_chunk = 0;
+		miocack(wq, mp, 0, 0);
+		sbclosechunk(sbp);
+		return;
+
+	case SBIOCSCHUNK:
+		/*
+		 * set up hi/lo water marks on stream head read queue.
+		 * unlikely to run out of resources. Fix at later date.
+		 */
+		if ((mop = allocb(sizeof (struct stroptions),
+		    BPRI_MED)) != NULL) {
+			struct stroptions *sop;
+			uint_t chunk;
+
+			chunk = *(uint_t *)mp->b_cont->b_rptr;
+			mop->b_datap->db_type = M_SETOPTS;
+			mop->b_wptr += sizeof (struct stroptions);
+			sop = (struct stroptions *)mop->b_rptr;
+			sop->so_flags = SO_HIWAT | SO_LOWAT;
+			sop->so_hiwat = SNIT_HIWAT(chunk, 1);
+			sop->so_lowat = SNIT_LOWAT(chunk, 1);
+			qreply(wq, mop);
+		}
+
+		sbp->sb_chunk = *(uint_t *)mp->b_cont->b_rptr;
+		miocack(wq, mp, 0, 0);
+		sbclosechunk(sbp);
+		return;
+
+	case SBIOCSFLAGS:
+		sbp->sb_flags = *(uint_t *)mp->b_cont->b_rptr;
+		miocack(wq, mp, 0, 0);
+		return;
+
+	case SBIOCSSNAP:
+		/*
+		 * if chunking dont worry about effects of
+		 * snipping of message size on head flow control
+		 * since it has a relatively small bearing on the
+		 * data rate onto the streamn head.
+		 */
+		if (!sbp->sb_chunk) {
+			/*
+			 * set up hi/lo water marks on stream head read queue.
+			 * unlikely to run out of resources. Fix at later date.
+			 */
+			if ((mop = allocb(sizeof (struct stroptions),
+			    BPRI_MED)) != NULL) {
+				struct stroptions *sop;
+				uint_t snap;
+				int fudge;
+
+				snap = *(uint_t *)mp->b_cont->b_rptr;
+				mop->b_datap->db_type = M_SETOPTS;
+				mop->b_wptr += sizeof (struct stroptions);
+				sop = (struct stroptions *)mop->b_rptr;
+				sop->so_flags = SO_HIWAT | SO_LOWAT;
+				fudge = snap <= 100 ?   4 :
+				    snap <= 400 ?   2 :
+				    1;
+				sop->so_hiwat = SNIT_HIWAT(snap, fudge);
+				sop->so_lowat = SNIT_LOWAT(snap, fudge);
+				qreply(wq, mop);
+			}
+		}
+
+		sbp->sb_snap = *(uint_t *)mp->b_cont->b_rptr;
+		miocack(wq, mp, 0, 0);
+		return;
+
+	default:
+		ASSERT(0);
+		return;
+	}
+}
+
+/*
+ * Write-side put procedure.  Its main task is to detect ioctls
+ * for manipulating the buffering state and hand them to sbioctl.
+ * Other message types are passed on through.
+ */
+static void
+sbwput(queue_t *wq, mblk_t *mp)
+{
+	struct	sb	*sbp = (struct sb *)wq->q_ptr;
+	struct copyresp *resp;
+
+	if (sbp->sb_flags & SB_SEND_ON_WRITE)
+		sbclosechunk(sbp);
+	switch (mp->b_datap->db_type) {
+	case M_IOCTL:
+		sbioctl(wq, mp);
+		break;
+
+	case M_IOCDATA:
+		resp = (struct copyresp *)mp->b_rptr;
+		if (resp->cp_rval) {
+			/*
+			 * Just free message on failure.
+			 */
+			freemsg(mp);
+			break;
+		}
+
+		switch (resp->cp_cmd) {
+		case SBIOCSTIME:
+		case SBIOCSCHUNK:
+		case SBIOCSFLAGS:
+		case SBIOCSSNAP:
+		case SBIOCGTIME:
+		case SBIOCGCHUNK:
+		case SBIOCGSNAP:
+		case SBIOCGFLAGS:
+			sbioc(wq, mp);
+			break;
+
+		default:
+			putnext(wq, mp);
+			break;
+		}
+		break;
+
+	default:
+		putnext(wq, mp);
+		break;
+	}
+}
+
+/*
+ * Read-side put procedure.  It's responsible for buffering up incoming
+ * messages and grouping them into aggregates according to the current
+ * buffering parameters.
+ */
+static void
+sbrput(queue_t *rq, mblk_t *mp)
+{
+	struct	sb	*sbp = (struct sb *)rq->q_ptr;
+
+	ASSERT(sbp);
+
+	switch (mp->b_datap->db_type) {
+	case M_PROTO:
+		if (sbp->sb_flags & SB_NO_PROTO_CVT) {
+			sbclosechunk(sbp);
+			sbsendit(rq, mp);
+			break;
+		} else {
+			/*
+			 * Convert M_PROTO to M_DATA.
+			 */
+			mp->b_datap->db_type = M_DATA;
+		}
+		/* FALLTHRU */
+
+	case M_DATA:
+		if ((sbp->sb_flags & SB_DEFER_CHUNK) &&
+		    !(sbp->sb_state & SB_FRCVD)) {
+			sbclosechunk(sbp);
+			sbsendit(rq, mp);
+			sbp->sb_state |= SB_FRCVD;
+		} else
+			sbaddmsg(rq, mp);
+
+		if ((sbp->sb_ticks > 0) && !(sbp->sb_timeoutid))
+			sbp->sb_timeoutid = qtimeout(sbp->sb_rq, sbtick,
+			    sbp, sbp->sb_ticks);
+
+		break;
+
+	case M_FLUSH:
+		if (*mp->b_rptr & FLUSHR) {
+			/*
+			 * Reset timeout, flush the chunk currently in
+			 * progress, and start a new chunk.
+			 */
+			if (sbp->sb_timeoutid) {
+				(void) quntimeout(sbp->sb_rq,
+				    sbp->sb_timeoutid);
+				sbp->sb_timeoutid = 0;
+			}
+			if (sbp->sb_mp) {
+				freemsg(sbp->sb_mp);
+				sbp->sb_tail = sbp->sb_mp = sbp->sb_head = NULL;
+				sbp->sb_mlen = 0;
+				sbp->sb_mcount = 0;
+			}
+			flushq(rq, FLUSHALL);
+		}
+		putnext(rq, mp);
+		break;
+
+	case M_CTL:
+		/*
+		 * Zero-length M_CTL means our timeout() popped.
+		 */
+		if (MBLKL(mp) == 0) {
+			freemsg(mp);
+			sbclosechunk(sbp);
+		} else {
+			sbclosechunk(sbp);
+			sbsendit(rq, mp);
+		}
+		break;
+
+	default:
+		if (mp->b_datap->db_type <= QPCTL) {
+			sbclosechunk(sbp);
+			sbsendit(rq, mp);
+		} else {
+			/* Note: out of band */
+			putnext(rq, mp);
+		}
+		break;
+	}
+}
+
+/*
+ *  read service procedure.
+ */
+/* ARGSUSED */
+static void
+sbrsrv(queue_t *rq)
+{
+	mblk_t	*mp;
+
+	/*
+	 * High priority messages shouldn't get here but if
+	 * one does, jam it through to avoid infinite loop.
+	 */
+	while ((mp = getq(rq)) != NULL) {
+		if (!canputnext(rq) && (mp->b_datap->db_type <= QPCTL)) {
+			/* should only get here if SB_NO_SROPS */
+			(void) putbq(rq, mp);
+			return;
+		}
+		putnext(rq, mp);
+	}
+}
+
+/*
+ * Handle write-side M_IOCTL messages.
+ */
+static void
+sbioctl(queue_t *wq, mblk_t *mp)
+{
+	struct	sb	*sbp = (struct sb *)wq->q_ptr;
+	struct iocblk	*iocp = (struct iocblk *)mp->b_rptr;
+	struct	timeval	*t;
+	clock_t	ticks;
+	mblk_t	*mop;
+	int	transparent = iocp->ioc_count;
+	mblk_t	*datamp;
+	int	error;
+
+	switch (iocp->ioc_cmd) {
+	case SBIOCSTIME:
+		if (iocp->ioc_count == TRANSPARENT) {
+#ifdef _SYSCALL32_IMPL
+			if ((iocp->ioc_flag & IOC_MODELS) != IOC_NATIVE) {
+				mcopyin(mp, NULL, sizeof (struct timeval32),
+				    NULL);
+			} else
+#endif /* _SYSCALL32_IMPL */
+			{
+				mcopyin(mp, NULL, sizeof (*t), NULL);
+			}
+			qreply(wq, mp);
+		} else {
+			/*
+			 * Verify argument length.
+			 */
+#ifdef _SYSCALL32_IMPL
+			if ((iocp->ioc_flag & IOC_MODELS) != IOC_NATIVE) {
+				struct timeval32 *t32;
+
+				error = miocpullup(mp,
+				    sizeof (struct timeval32));
+				if (error != 0) {
+					miocnak(wq, mp, 0, error);
+					break;
+				}
+				t32 = (struct timeval32 *)mp->b_cont->b_rptr;
+				if (t32->tv_sec < 0 || t32->tv_usec < 0) {
+					miocnak(wq, mp, 0, EINVAL);
+					break;
+				}
+				ticks = TIMEVAL_TO_TICK(t32);
+			} else
+#endif /* _SYSCALL32_IMPL */
+			{
+				error = miocpullup(mp, sizeof (struct timeval));
+				if (error != 0) {
+					miocnak(wq, mp, 0, error);
+					break;
+				}
+
+				t = (struct timeval *)mp->b_cont->b_rptr;
+				if (t->tv_sec < 0 || t->tv_usec < 0) {
+					miocnak(wq, mp, 0, EINVAL);
+					break;
+				}
+				ticks = TIMEVAL_TO_TICK(t);
+			}
+			sbp->sb_ticks = ticks;
+			if (ticks == 0)
+				sbp->sb_chunk = 0;
+			miocack(wq, mp, 0, 0);
+			sbclosechunk(sbp);
+		}
+		break;
+
+	case SBIOCGTIME: {
+		struct timeval *t;
+
+		/*
+		 * Verify argument length.
+		 */
+		if (transparent != TRANSPARENT) {
+#ifdef _SYSCALL32_IMPL
+			if ((iocp->ioc_flag & IOC_MODELS) != IOC_NATIVE) {
+				error = miocpullup(mp,
+				    sizeof (struct timeval32));
+				if (error != 0) {
+					miocnak(wq, mp, 0, error);
+					break;
+				}
+			} else
+#endif /* _SYSCALL32_IMPL */
+			error = miocpullup(mp, sizeof (struct timeval));
+			if (error != 0) {
+				miocnak(wq, mp, 0, error);
+				break;
+			}
+		}
+
+		/*
+		 * If infinite timeout, return range error
+		 * for the ioctl.
+		 */
+		if (sbp->sb_ticks < 0) {
+			miocnak(wq, mp, 0, ERANGE);
+			break;
+		}
+
+#ifdef _SYSCALL32_IMPL
+		if ((iocp->ioc_flag & IOC_MODELS) != IOC_NATIVE) {
+			struct timeval32 *t32;
+
+			if (transparent == TRANSPARENT) {
+				datamp = allocb(sizeof (*t32), BPRI_MED);
+				if (datamp == NULL) {
+					miocnak(wq, mp, 0, EAGAIN);
+					break;
+				}
+				mcopyout(mp, NULL, sizeof (*t32), NULL, datamp);
+			}
+
+			t32 = (struct timeval32 *)mp->b_cont->b_rptr;
+			TICK_TO_TIMEVAL32(sbp->sb_ticks, t32);
+
+			if (transparent == TRANSPARENT)
+				qreply(wq, mp);
+			else
+				miocack(wq, mp, sizeof (*t32), 0);
+		} else
+#endif /* _SYSCALL32_IMPL */
+		{
+			if (transparent == TRANSPARENT) {
+				datamp = allocb(sizeof (*t), BPRI_MED);
+				if (datamp == NULL) {
+					miocnak(wq, mp, 0, EAGAIN);
+					break;
+				}
+				mcopyout(mp, NULL, sizeof (*t), NULL, datamp);
+			}
+
+			t = (struct timeval *)mp->b_cont->b_rptr;
+			TICK_TO_TIMEVAL(sbp->sb_ticks, t);
+
+			if (transparent == TRANSPARENT)
+				qreply(wq, mp);
+			else
+				miocack(wq, mp, sizeof (*t), 0);
+		}
+		break;
+	}
+
+	case SBIOCCTIME:
+		sbp->sb_ticks = -1;
+		miocack(wq, mp, 0, 0);
+		break;
+
+	case SBIOCSCHUNK:
+		if (iocp->ioc_count == TRANSPARENT) {
+			mcopyin(mp, NULL, sizeof (uint_t), NULL);
+			qreply(wq, mp);
+		} else {
+			/*
+			 * Verify argument length.
+			 */
+			error = miocpullup(mp, sizeof (uint_t));
+			if (error != 0) {
+				miocnak(wq, mp, 0, error);
+				break;
+			}
+
+			/*
+			 * set up hi/lo water marks on stream head read queue.
+			 * unlikely to run out of resources. Fix at later date.
+			 */
+			if ((mop = allocb(sizeof (struct stroptions),
+			    BPRI_MED)) != NULL) {
+				struct stroptions *sop;
+				uint_t chunk;
+
+				chunk = *(uint_t *)mp->b_cont->b_rptr;
+				mop->b_datap->db_type = M_SETOPTS;
+				mop->b_wptr += sizeof (struct stroptions);
+				sop = (struct stroptions *)mop->b_rptr;
+				sop->so_flags = SO_HIWAT | SO_LOWAT;
+				sop->so_hiwat = SNIT_HIWAT(chunk, 1);
+				sop->so_lowat = SNIT_LOWAT(chunk, 1);
+				qreply(wq, mop);
+			}
+
+			sbp->sb_chunk = *(uint_t *)mp->b_cont->b_rptr;
+			miocack(wq, mp, 0, 0);
+			sbclosechunk(sbp);
+		}
+		break;
+
+	case SBIOCGCHUNK:
+		/*
+		 * Verify argument length.
+		 */
+		if (transparent != TRANSPARENT) {
+			error = miocpullup(mp, sizeof (uint_t));
+			if (error != 0) {
+				miocnak(wq, mp, 0, error);
+				break;
+			}
+		}
+
+		if (transparent == TRANSPARENT) {
+			datamp = allocb(sizeof (uint_t), BPRI_MED);
+			if (datamp == NULL) {
+				miocnak(wq, mp, 0, EAGAIN);
+				break;
+			}
+			mcopyout(mp, NULL, sizeof (uint_t), NULL, datamp);
+		}
+
+		*(uint_t *)mp->b_cont->b_rptr = sbp->sb_chunk;
+
+		if (transparent == TRANSPARENT)
+			qreply(wq, mp);
+		else
+			miocack(wq, mp, sizeof (uint_t), 0);
+		break;
+
+	case SBIOCSSNAP:
+		if (iocp->ioc_count == TRANSPARENT) {
+			mcopyin(mp, NULL, sizeof (uint_t), NULL);
+			qreply(wq, mp);
+		} else {
+			/*
+			 * Verify argument length.
+			 */
+			error = miocpullup(mp, sizeof (uint_t));
+			if (error != 0) {
+				miocnak(wq, mp, 0, error);
+				break;
+			}
+
+			/*
+			 * if chunking dont worry about effects of
+			 * snipping of message size on head flow control
+			 * since it has a relatively small bearing on the
+			 * data rate onto the streamn head.
+			 */
+			if (!sbp->sb_chunk) {
+				/*
+				 * set up hi/lo water marks on stream
+				 * head read queue.  unlikely to run out
+				 * of resources. Fix at later date.
+				 */
+				if ((mop = allocb(sizeof (struct stroptions),
+				    BPRI_MED)) != NULL) {
+					struct stroptions *sop;
+					uint_t snap;
+					int fudge;
+
+					snap = *(uint_t *)mp->b_cont->b_rptr;
+					mop->b_datap->db_type = M_SETOPTS;
+					mop->b_wptr += sizeof (*sop);
+					sop = (struct stroptions *)mop->b_rptr;
+					sop->so_flags = SO_HIWAT | SO_LOWAT;
+					fudge = (snap <= 100) ? 4 :
+					    (snap <= 400) ? 2 : 1;
+					sop->so_hiwat = SNIT_HIWAT(snap, fudge);
+					sop->so_lowat = SNIT_LOWAT(snap, fudge);
+					qreply(wq, mop);
+				}
+			}
+
+			sbp->sb_snap = *(uint_t *)mp->b_cont->b_rptr;
+
+			miocack(wq, mp, 0, 0);
+		}
+		break;
+
+	case SBIOCGSNAP:
+		/*
+		 * Verify argument length
+		 */
+		if (transparent != TRANSPARENT) {
+			error = miocpullup(mp, sizeof (uint_t));
+			if (error != 0) {
+				miocnak(wq, mp, 0, error);
+				break;
+			}
+		}
+
+		if (transparent == TRANSPARENT) {
+			datamp = allocb(sizeof (uint_t), BPRI_MED);
+			if (datamp == NULL) {
+				miocnak(wq, mp, 0, EAGAIN);
+				break;
+			}
+			mcopyout(mp, NULL, sizeof (uint_t), NULL, datamp);
+		}
+
+		*(uint_t *)mp->b_cont->b_rptr = sbp->sb_snap;
+
+		if (transparent == TRANSPARENT)
+			qreply(wq, mp);
+		else
+			miocack(wq, mp, sizeof (uint_t), 0);
+		break;
+
+	case SBIOCSFLAGS:
+		/*
+		 * set the flags.
+		 */
+		if (iocp->ioc_count == TRANSPARENT) {
+			mcopyin(mp, NULL, sizeof (uint_t), NULL);
+			qreply(wq, mp);
+		} else {
+			error = miocpullup(mp, sizeof (uint_t));
+			if (error != 0) {
+				miocnak(wq, mp, 0, error);
+				break;
+			}
+			sbp->sb_flags = *(uint_t *)mp->b_cont->b_rptr;
+			miocack(wq, mp, 0, 0);
+		}
+		break;
+
+	case SBIOCGFLAGS:
+		/*
+		 * Verify argument length
+		 */
+		if (transparent != TRANSPARENT) {
+			error = miocpullup(mp, sizeof (uint_t));
+			if (error != 0) {
+				miocnak(wq, mp, 0, error);
+				break;
+			}
+		}
+
+		if (transparent == TRANSPARENT) {
+			datamp = allocb(sizeof (uint_t), BPRI_MED);
+			if (datamp == NULL) {
+				miocnak(wq, mp, 0, EAGAIN);
+				break;
+			}
+			mcopyout(mp, NULL, sizeof (uint_t), NULL, datamp);
+		}
+
+		*(uint_t *)mp->b_cont->b_rptr = sbp->sb_flags;
+
+		if (transparent == TRANSPARENT)
+			qreply(wq, mp);
+		else
+			miocack(wq, mp, sizeof (uint_t), 0);
+		break;
+
+
+	default:
+		putnext(wq, mp);
+		break;
+	}
+}
+
+/*
+ * Given a length l, calculate the amount of extra storage
+ * required to round it up to the next multiple of the alignment a.
+ */
+#define	RoundUpAmt(l, a)	((l) % (a) ? (a) - ((l) % (a)) : 0)
+/*
+ * Calculate additional amount of space required for alignment.
+ */
+#define	Align(l)		RoundUpAmt(l, sizeof (ulong_t))
+/*
+ * Smallest possible message size when headers are enabled.
+ * This is used to calculate whether a chunk is nearly full.
+ */
+#define	SMALLEST_MESSAGE	sizeof (struct sb_hdr) + _POINTER_ALIGNMENT
+
+/*
+ * Process a read-side M_DATA message.
+ *
+ * If the currently accumulating chunk doesn't have enough room
+ * for the message, close off the chunk, pass it upward, and start
+ * a new one.  Then add the message to the current chunk, taking
+ * account of the possibility that the message's size exceeds the
+ * chunk size.
+ *
+ * If headers are enabled add an sb_hdr header and trailing alignment padding.
+ *
+ * To optimise performance the total number of msgbs should be kept
+ * to a minimum. This is achieved by using any remaining space in message N
+ * for both its own padding as well as the header of message N+1 if possible.
+ * If there's insufficient space we allocate one message to hold this 'wrapper'.
+ * (there's likely to be space beyond message N, since allocb would have
+ * rounded up the required size to one of the dblk_sizes).
+ *
+ */
+static void
+sbaddmsg(queue_t *rq, mblk_t *mp)
+{
+	struct sb	*sbp;
+	struct timeval	t;
+	struct sb_hdr	hp;
+	mblk_t *wrapper;	/* padding for msg N, header for msg N+1 */
+	mblk_t *last;		/* last mblk of current message */
+	size_t wrapperlen;	/* length of header + padding */
+	size_t origlen;		/* data length before truncation */
+	size_t pad;		/* bytes required to align header */
+
+	sbp = (struct sb *)rq->q_ptr;
+
+	origlen = msgdsize(mp);
+
+	/*
+	 * Truncate the message.
+	 */
+	if ((sbp->sb_snap > 0) && (origlen > sbp->sb_snap) &&
+			(adjmsg(mp, -(origlen - sbp->sb_snap)) == 1))
+		hp.sbh_totlen = hp.sbh_msglen = sbp->sb_snap;
+	else
+		hp.sbh_totlen = hp.sbh_msglen = origlen;
+
+	if (sbp->sb_flags & SB_NO_HEADER) {
+
+		/*
+		 * Would the inclusion of this message overflow the current
+		 * chunk? If so close the chunk off and start a new one.
+		 */
+		if ((hp.sbh_totlen + sbp->sb_mlen) > sbp->sb_chunk)
+			sbclosechunk(sbp);
+		/*
+		 * First message too big for chunk - just send it up.
+		 * This will always be true when we're not chunking.
+		 */
+		if (hp.sbh_totlen > sbp->sb_chunk) {
+			sbsendit(rq, mp);
+			return;
+		}
+
+		/*
+		 * We now know that the msg will fit in the chunk.
+		 * Link it onto the end of the chunk.
+		 * Since linkb() walks the entire chain, we keep a pointer to
+		 * the first mblk of the last msgb added and call linkb on that
+		 * that last message, rather than performing the
+		 * O(n) linkb() operation on the whole chain.
+		 * sb_head isn't needed in this SB_NO_HEADER mode.
+		 */
+		if (sbp->sb_mp)
+			linkb(sbp->sb_tail, mp);
+		else
+			sbp->sb_mp = mp;
+
+		sbp->sb_tail = mp;
+		sbp->sb_mlen += hp.sbh_totlen;
+		sbp->sb_mcount++;
+	} else {
+		/* Timestamp must be done immediately */
+		uniqtime(&t);
+		TIMEVAL_TO_TIMEVAL32(&hp.sbh_timestamp, &t);
+
+		pad = Align(hp.sbh_totlen);
+		hp.sbh_totlen += sizeof (hp);
+		hp.sbh_totlen += pad;
+
+		/*
+		 * Would the inclusion of this message overflow the current
+		 * chunk? If so close the chunk off and start a new one.
+		 */
+		if ((hp.sbh_totlen + sbp->sb_mlen) > sbp->sb_chunk)
+				sbclosechunk(sbp);
+
+		if (sbp->sb_head == NULL) {
+			/* Allocate leading header of new chunk */
+			sbp->sb_head = allocb(sizeof (hp), BPRI_MED);
+			if (sbp->sb_head == NULL) {
+				/*
+				 * Memory allocation failure.
+				 * This will need to be revisited
+				 * since using certain flag combinations
+				 * can result in messages being dropped
+				 * silently.
+				 */
+				freemsg(mp);
+				sbp->sb_drops++;
+				return;
+			}
+			sbp->sb_mp = sbp->sb_head;
+		}
+
+		/*
+		 * Copy header into message
+		 */
+		hp.sbh_drops = sbp->sb_drops;
+		hp.sbh_origlen = origlen;
+		(void) memcpy(sbp->sb_head->b_wptr, (char *)&hp, sizeof (hp));
+		sbp->sb_head->b_wptr += sizeof (hp);
+
+		ASSERT(sbp->sb_head->b_wptr <= sbp->sb_head->b_datap->db_lim);
+
+		/*
+		 * Join message to the chunk
+		 */
+		linkb(sbp->sb_head, mp);
+
+		sbp->sb_mcount++;
+		sbp->sb_mlen += hp.sbh_totlen;
+
+		/*
+		 * If the first message alone is too big for the chunk close
+		 * the chunk now.
+		 * If the next message would immediately cause the chunk to
+		 * overflow we may as well close the chunk now. The next
+		 * message is certain to be at least SMALLEST_MESSAGE size.
+		 */
+		if (hp.sbh_totlen + SMALLEST_MESSAGE > sbp->sb_chunk) {
+			sbclosechunk(sbp);
+			return;
+		}
+
+		/*
+		 * Find space for the wrapper. The wrapper consists of:
+		 *
+		 * 1) Padding for this message (this is to ensure each header
+		 * begins on an 8 byte boundary in the userland buffer).
+		 *
+		 * 2) Space for the next message's header, in case the next
+		 * next message will fit in this chunk.
+		 *
+		 * It may be possible to append the wrapper to the last mblk
+		 * of the message, but only if we 'own' the data. If the dblk
+		 * has been shared through dupmsg() we mustn't alter it.
+		 */
+
+		wrapperlen = (sizeof (hp) + pad);
+
+		/* Is there space for the wrapper beyond the message's data ? */
+		for (last = mp; last->b_cont; last = last->b_cont)
+			;
+
+		if ((wrapperlen <= MBLKTAIL(last)) &&
+			(last->b_datap->db_ref == 1)) {
+			if (pad > 0) {
+				/*
+				 * Pad with zeroes to the next pointer boundary
+				 * (we don't want to disclose kernel data to
+				 * users), then advance wptr.
+				 */
+				(void) memset(last->b_wptr, 0, pad);
+				last->b_wptr += pad;
+			}
+			/* Remember where to write the header information */
+			sbp->sb_head = last;
+		} else {
+			/* Have to allocate additional space for the wrapper */
+			wrapper = allocb(wrapperlen, BPRI_MED);
+			if (wrapper == NULL) {
+				sbclosechunk(sbp);
+				return;
+			}
+			if (pad > 0) {
+				/*
+				 * Pad with zeroes (we don't want to disclose
+				 * kernel data to users).
+				 */
+				(void) memset(wrapper->b_wptr, 0, pad);
+				wrapper->b_wptr += pad;
+			}
+			/* Link the wrapper msg onto the end of the chunk */
+			linkb(mp, wrapper);
+			/* Remember to write the next header in this wrapper */
+			sbp->sb_head = wrapper;
+		}
+	}
+}
+
+/*
+ * Called from timeout().
+ * Signal a timeout by passing a zero-length M_CTL msg in the read-side
+ * to synchronize with any active module threads (open, close, wput, rput).
+ */
+static void
+sbtick(void *arg)
+{
+	struct sb *sbp = arg;
+	queue_t	*rq;
+
+	ASSERT(sbp);
+
+	rq = sbp->sb_rq;
+	sbp->sb_timeoutid = 0;		/* timeout has fired */
+
+	if (putctl(rq, M_CTL) == 0)	/* failure */
+		sbp->sb_timeoutid = qtimeout(rq, sbtick, sbp, sbp->sb_ticks);
+}
+
+/*
+ * Close off the currently accumulating chunk and pass
+ * it upward.  Takes care of resetting timers as well.
+ *
+ * This routine is called both directly and as a result
+ * of the chunk timeout expiring.
+ */
+static void
+sbclosechunk(struct sb *sbp)
+{
+	mblk_t	*mp;
+	queue_t	*rq;
+
+	ASSERT(sbp);
+
+	if (sbp->sb_timeoutid) {
+		(void) quntimeout(sbp->sb_rq, sbp->sb_timeoutid);
+		sbp->sb_timeoutid = 0;
+	}
+
+	mp = sbp->sb_mp;
+	rq = sbp->sb_rq;
+
+	/*
+	 * If there's currently a chunk in progress, close it off
+	 * and try to send it up.
+	 */
+	if (mp) {
+		sbsendit(rq, mp);
+	}
+
+	/*
+	 * Clear old chunk.  Ready for new msgs.
+	 */
+	sbp->sb_tail = sbp->sb_mp = sbp->sb_head = NULL;
+	sbp->sb_mlen = 0;
+	sbp->sb_mcount = 0;
+	if (sbp->sb_flags & SB_DEFER_CHUNK)
+		sbp->sb_state &= ~SB_FRCVD;
+
+}
+
+static void
+sbsendit(queue_t *rq, mblk_t *mp)
+{
+	struct	sb	*sbp = (struct sb *)rq->q_ptr;
+
+	if (!canputnext(rq)) {
+		if (sbp->sb_flags & SB_NO_DROPS)
+			(void) putq(rq, mp);
+		else {
+			freemsg(mp);
+			sbp->sb_drops += sbp->sb_mcount;
+		}
+		return;
+	}
+	/*
+	 * If there are messages on the q already, keep
+	 * queueing them since they need to be processed in order.
+	 */
+	if (qsize(rq) > 0) {
+		/* should only get here if SB_NO_DROPS */
+		(void) putq(rq, mp);
+	}
+	else
+		putnext(rq, mp);
+}