/**
 * ch_msg: - process message from queue
 * @q: queue
 * @mp: the message to process
 *
 * This simply flows data messages through with flow control and intercepts
 * all other messages.
 */
static inline fastcall __hot int
ch_msg(queue_t *q, mblk_t *mp)
{
	struct ch *ch = CH_PRIV(q);

	if (test_bit(CH_ENABLE_BIT, &ch->flags)) {
		if (likely(DB_TYPE(mp) == M_DATA)) {
			if (likely(bcanputnext(q, mp->b_band))) {
				putnext(q, mp);
				return (0);
			}
			return (-EBUSY);
		} else if (DB_TYPE(mp) == M_PROTO) {
			if (*(uint32_t *) mp->b_rptr = CH_DATA_REQ) {
				if (likely(bcanputnext(q, mp->b_band))) {
					*(uint32_t *) mp->b_rptr = CH_DATA_IND;
					putnext(q, mp);
					return (0);
				}
				return (-EBUSY);
			}
			/* go slow */
		}
	}
	return ch_msg_slow(ch, q, mp);
}

/*
 *  -------------------------------------------------------------------------
 *
 *  M_FLUSH Handling
 *
 *  -------------------------------------------------------------------------
 */
STATIC inline void
sscop_w_flush(queue_t *q, mblk_t *mp)
{
	if (*mp->b_rptr & FLUSHW) {
		if (*mp->b_rptr & FLUSHBAND)
			flushband(q, mp->b_rptr[1], FLUSHDATA);
		else
			flushq(q, FLUSHDATA);
		if (q - q_next) {
			putnext(q, mp);
			return;
		}
		*mp->b_rptr &= ~FLUSHW;
	}
	if (*mp->b_rptr & FLUSHR) {
		if (*mp->b_rptr & FLUSHBAND)
			flushband(RD(q), mp->b_rptr[1], FLUSHDATA);
		else
			flushq(RD(q), FLUSHDATA);
		qreply(q, mp);
		return;
	}
	if (q->q_next) {
		putnext(q, mp);
		return;
	}
}

/**
 * ptem_rput - read put procedure
 * @q: read queue
 * @mp: message to put
 */
static streamscall __hot_get int
ptem_rput(queue_t *q, mblk_t *mp)
{
	/* fast path for data */
	if (likely(mp->b_datap->db_type == M_DATA)) {
	      m_data:
		/* no read side service procedure, do not check for flow control. */
		putnext(q, mp);
		return (0);
	}
	switch (mp->b_datap->db_type) {
	case M_DATA:
		goto m_data;
	case M_IOCDATA:
	case M_IOCACK:
		/* should not happen */
		swerr();
	case M_READ:
	case M_DELAY:
	case M_IOCNAK:
		freemsg(mp);
		break;
	default:
		swerr();
	case M_HANGUP:
	case M_FLUSH:		/* we do not queue on the read side */
		putnext(q, mp);
		break;
	case M_IOCTL:
		ptem_r_msg(q, mp);
		break;
	}
	return (0);
}

/*
 * wput(9E) is symmetric for master and slave sides, so this handles both
 * without splitting the codepath.
 *
 * zc_wput() looks at the other side; if there is no process holding that
 * side open, it frees the message.  This prevents processes from hanging
 * if no one is holding open the console.  Otherwise, it putnext's high
 * priority messages, putnext's normal messages if possible, and otherwise
 * enqueues the messages; in the case that something is enqueued, wsrv(9E)
 * will take care of eventually shuttling I/O to the other side.
 */
static void
zc_wput(queue_t *qp, mblk_t *mp)
{
	unsigned char type = mp->b_datap->db_type;

	ASSERT(qp->q_ptr);

	DBG1("entering zc_wput, %s side", zc_side(qp));

	if (zc_switch(RD(qp)) == NULL) {
		DBG1("wput to %s side (no one listening)", zc_side(qp));
		switch (type) {
		case M_FLUSH:
			handle_mflush(qp, mp);
			break;
		case M_IOCTL:
			miocnak(qp, mp, 0, 0);
			break;
		default:
			freemsg(mp);
			break;
		}
		return;
	}

	if (type >= QPCTL) {
		DBG1("(hipri) wput, %s side", zc_side(qp));
		switch (type) {
		case M_READ:		/* supposedly from ldterm? */
			DBG("zc_wput: tossing M_READ\n");
			freemsg(mp);
			break;
		case M_FLUSH:
			handle_mflush(qp, mp);
			break;
		default:
			/*
			 * Put this to the other side.
			 */
			ASSERT(zc_switch(RD(qp)) != NULL);
			putnext(zc_switch(RD(qp)), mp);
			break;
		}
		DBG1("done (hipri) wput, %s side", zc_side(qp));
		return;
	}

	/*
	 * Only putnext if there isn't already something in the queue.
	 * otherwise things would wind up out of order.
	 */
	if (qp->q_first == NULL && bcanputnext(RD(zc_switch(qp)), mp->b_band)) {
		DBG("wput: putting message to other side\n");
		putnext(RD(zc_switch(qp)), mp);
	} else {
		DBG("wput: putting msg onto queue\n");
		(void) putq(qp, mp);
	}
	DBG1("done wput, %s side", zc_side(qp));
}

/*
 * telmodwsrv - module write service procedure
 */
static void
telmodwsrv(queue_t *q)
{
	mblk_t	*mp, *savemp;

	struct	telmod_info    *tmip = (struct telmod_info *)q->q_ptr;

	while ((mp = getq(q)) != NULL) {
		if (!canputnext(q)) {
			ASSERT(mp->b_datap->db_type < QPCTL);
			(void) putbq(q, mp);
			return;
		}
		switch (mp->b_datap->db_type) {

		case M_DATA:
			if (tmip->flags & TEL_STOPPED) {
				(void) putbq(q, mp);
				return;
			}
			/*
			 * Insert a null character if carraige return
			 * is not followed by line feed
			 */
			if (!snd_parse(q, mp)) {
				return;
			}
			break;

		case M_CTL:
			if (((mp->b_wptr - mp->b_rptr) == 1) &&
				(*(mp->b_rptr) == M_CTL_MAGIC_NUMBER)) {
				savemp = mp->b_cont;
				freeb(mp);
				mp = savemp;
			}
			putnext(q, mp);
			break;

		case M_PROTO:
			putnext(q, mp);
			break;

		default:
#ifdef DEBUG
			cmn_err(CE_NOTE,
			    "telmodwsrv: unexpected msg type 0x%x",
			    mp->b_datap->db_type);
#endif
			freemsg(mp);
		}

	}
}

/*
 * dm2s_rsrv - Streams read side service procedure.
 *
 * All messages are received in the service procedure
 * only. This is done to simplify the streams synchronization.
 */
int
dm2s_rsrv(queue_t *rq)
{
	mblk_t *mp;
	dm2s_t *dm2sp = (dm2s_t *)rq->q_ptr;

	DPRINTF(DBG_DRV, ("dm2s_rsrv: called\n"));
	ASSERT(dm2sp != NULL);
	mutex_enter(&dm2sp->ms_lock);

	/* Receive if there are any messages waiting in the mailbox. */
	dm2s_receive(dm2sp);
	mutex_exit(&dm2sp->ms_lock);

	/* Send the received messages up the stream. */
	while ((mp = getq(rq)) != NULL) {
		if (canputnext(rq)) {
			putnext(rq, mp);
		} else {
			putbq(rq, mp);
			break;
		}
	}
	DPRINTF(DBG_DRV, ("dm2s_rsrv: return\n"));
	return (0);
}

static int
setup_stream(
	     queue_t *q,
	     int mode
	     )
{
	register mblk_t *mp;

	pprintf(DD_OPEN,"parse: SETUP_STREAM - setting up stream for q=%x\n", q);

	mp = allocb(sizeof(struct stroptions), BPRI_MED);
	if (mp)
	{
		struct stroptions *str = (void *)mp->b_wptr;

		str->so_flags   = SO_READOPT|SO_HIWAT|SO_LOWAT|SO_ISNTTY;
		str->so_readopt = (mode == M_PARSE) ? RMSGD : RNORM;
		str->so_hiwat   = (mode == M_PARSE) ? sizeof(parsetime_t) : 256;
		str->so_lowat   = 0;
		mp->b_datap->db_type = M_SETOPTS;
		mp->b_wptr     += sizeof(struct stroptions);
		if (!q)
		    panic("NULL q - strange");
		putnext(q, mp);
		return putctl1(WR(q)->q_next, M_CTL, (mode == M_PARSE) ? MC_SERVICEIMM :
			       MC_SERVICEDEF);
	}
	else
	{
		pprintf(DD_OPEN, "parse: setup_stream - FAILED - no MEMORY for allocb\n"); 
		return 0;
	}
}

static int
log_rsrv(queue_t *q)
{
	mblk_t *mp;
	char *msg, *msgid_start, *msgid_end;
	size_t idlen;

	while (canputnext(q) && (mp = getq(q)) != NULL) {
		if (log_msgid == 0) {
			/*
			 * Strip out the message ID.  If it's a kernel
			 * SL_CONSOLE message, replace msgid with "unix: ".
			 */
			msg = (char *)mp->b_cont->b_rptr;
			if ((msgid_start = strstr(msg, "[ID ")) != NULL &&
			    (msgid_end = strstr(msgid_start, "] ")) != NULL) {
				log_ctl_t *lc = (log_ctl_t *)mp->b_rptr;
				if ((lc->flags & SL_CONSOLE) &&
				    (lc->pri & LOG_FACMASK) == LOG_KERN)
					msgid_start = msg + snprintf(msg,
					    7, "unix: ");
				idlen = msgid_end + 2 - msgid_start;
				ovbcopy(msg, msg + idlen, msgid_start - msg);
				mp->b_cont->b_rptr += idlen;
			}
		}
		mp->b_band = 0;
		putnext(q, mp);
	}
	return (0);
}

static void
t_rput(queue_t *q, mblk_t *mp)
{
	int err = EOPNOTSUPP;

	trace();
	if (q->q_count && mp->b_datap->db_type < QPCTL) {
		putq(q, mp);
		return;
	}
	switch (mp->b_datap->db_type) {
	case M_DATA:
		if ((err = t_m_data(q, mp)))
			break;
		return;
	case M_CTL:
	case M_PROTO:
	case M_PCPROTO:
		if ((err = t_m_proto(q, mp)))
			break;
		return;
	}
	switch (err) {
	case EAGAIN:
		putq(q, mp);
		return;
	case EOPNOTSUPP:
		if (q->q_next) {
			putnext(q, mp);
			return;
		}
	}
	freemsg(mp);
}

static fastcall __unlikely int
ch_m_flush(queue_t *q, mblk_t *mp)
{
	if (mp->b_rptr[0] & ((q->q_flag & QREADR) ? FLUSHR : FLUSHW)) {
		if (mp->b_rptr[0] & FLUSHBAND)
			flushband(q, mp->b_rptr[1], FLUSHDATA);
		else
			flushq(q, FLUSHDATA);
	}
	/* do a pipemod flush sense reversal too */
	switch (mp->b_rptr[0] & FLUSHRW) {
	case FLUSHR:
		mp->b_rptr[0] &= ~FLUSHRW;
		mp->b_rptr[0] |= FLUSHW;
		break;
	case FLUSHW:
		mp->b_rptr[0] &= ~FLUSHRW;
		mp->b_rptr[0] |= FLUSHR;
		break;
	case FLUSHRW:
		mp->b_rptr[0] &= ~FLUSHRW;
		mp->b_rptr[0] |= FLUSHRW;
		break;
	}
	putnext(q, mp);
	return (0);
}

static int
setup_stream(
	     queue_t *q,
	     int mode
	     )
{
	mblk_t *mp;

	mp = allocb(sizeof(struct stroptions), BPRI_MED);
	if (mp)
	{
		struct stroptions *str = (struct stroptions *)(void *)mp->b_rptr;

		str->so_flags   = SO_READOPT|SO_HIWAT|SO_LOWAT;
		str->so_readopt = (mode == M_PARSE) ? RMSGD : RNORM;
		str->so_hiwat   = (mode == M_PARSE) ? sizeof(parsetime_t) : 256;
		str->so_lowat   = 0;
		mp->b_datap->db_type = M_SETOPTS;
		mp->b_wptr += sizeof(struct stroptions);
		putnext(q, mp);
		return putctl1(WR(q)->q_next, M_CTL, (mode == M_PARSE) ? MC_SERVICEIMM :
			       MC_SERVICEDEF);
	}
	else
	{
		parseprintf(DD_OPEN,("parse: setup_stream - FAILED - no MEMORY for allocb\n"));
		return 0;
	}
}

/************************************************************************
 *
 * Function Name: ip2xinet_lwsrv
 * Title: IP2XINET Lower Write Service routine
 *
 * Description:
 *      Send all of the messages on this queue down to the next driver.
 *      If we discover that we can't do a put, then stop the Linux
 *      devices from sending us stuff.
 *
 ************************************************************************/
STATIC streamscall int
ip2xinet_lwsrv(queue_t *q)
{
	mblk_t *mp;
	int allsent = 1;
	int i;
	struct ip2xinet_dev *dev = ip2xinet_devs;
	struct ip2xinet_priv *privp;

	while ((mp = getq(q))) {
		/* M_PROTO's should be last on the list.  If it is something else, then it should
		   be ahead, and we can just go ahead and put it down. */
		if (mp->b_datap->db_type == M_PROTO) {
			if (canputnext(q)) {
				putnext(q, mp);
			} else {
				noenable(q);
				if (!putbq(q, mp))
					freemsg(mp);	/* FIXME */
				enableok(q);
				allsent = 0;
				break;
			}
		} else {
			putnext(q, mp);
		}
	}

	/* Handle the flow control.  If we were able to send everything then it is ok for the
	   kernel to send us more stuff.  Otherwise it is not ok.  Go through all of the devices
	   and set the appropriate state. */
	spin_lock(&ip2xinet_lock);
	for (i = 0; i < NUMIP2XINET; i++, dev++) {
		privp = &dev->priv;
		if (privp->state == 1 && ip2xinet_status.ip2x_dlstate == DL_IDLE) {
			if (allsent) {
				netif_start_queue(&dev->dev);	/* kernel can transmit */
			} else {
				netif_stop_queue(&dev->dev);	/* We are flow controlled. */
			}
		}
	}
	spin_unlock(&ip2xinet_lock);

	return (0);
}

/**
 * pckt_wput - write-side put procedure
 * @q: write queue
 * @mp: message to put
 *
 * When the user process writes data, the pckt module passes the message on to
 * the next module or driver unchanged.
 */
static streamscall int
pckt_wput(queue_t *q, mblk_t *mp)
{
	/* No write-side service procedure, no need to check for flow control, pass the message
	   along. */
	putnext(q, mp);
	return (0);
}

int kz_put_json (kz_t *kz, const char *json_str)
{
    if (!(kz->flags & KZ_FLAGS_RAW)) {
        errno = EINVAL;
        return -1;
    }
    return putnext (kz, json_str);
}

/* ARGSUSED */
static void
cvc_input_daemon(void)
{
	char		linebuf[MAX_XFER_CINPUT + 1];
	char		*cp;
	mblk_t		*mbp;
	int		c;
	int		dropped_read = 0;

	for (;;) {
		cvc_getstr(linebuf);

		mbp = allocb(strlen(linebuf), BPRI_MED);
		if (mbp == NULL) {	/* drop it & go on if no buffer */
			if (!dropped_read) {
				cmn_err(CE_WARN, "cvc_input_daemon: "
				    "dropping IOSRAM reads");
			}
			dropped_read++;
			continue;
		}

		if (dropped_read) {
			cmn_err(CE_WARN,
			    "cvc_input_daemon: dropped %d IOSRAM reads",
			    dropped_read);
			dropped_read = 0;
		}

		for (cp = linebuf; *cp != '\0'; cp++) {
			c = (int)*cp;
			if (c == '\r')
				c = '\n';
			c &= 0177;
			*mbp->b_wptr = (char)c;
			mbp->b_wptr++;
		}
		mutex_enter(&cvcmutex);
		if (input_ok) {
			if (cvcinput_q == NULL) {
				cmn_err(CE_WARN,
				    "cvc_input_daemon: cvcinput_q is NULL!");
			} else {
				/*
				 * XXX - should canputnext be called here?
				 * Starfire's cvc doesn't do that, and it
				 * appears to work anyway.
				 */
				(void) putnext(cvcinput_q, mbp);
			}
		} else {
			freemsg(mbp);
		}
		mutex_exit(&cvcmutex);
	}

	/* NOTREACHED */
}

/*ARGSUSED*/
static int
zc_master_open(zc_state_t *zcs,
    queue_t	*rqp,	/* pointer to the read side queue */
    dev_t	*devp,	/* pointer to stream tail's dev */
    int		oflag,	/* the user open(2) supplied flags */
    int		sflag,	/* open state flag */
    cred_t	*credp)	/* credentials */
{
	mblk_t *mop;
	struct stroptions *sop;

	/*
	 * Enforce exclusivity on the master side; the only consumer should
	 * be the zoneadmd for the zone.
	 */
	if ((zcs->zc_state & ZC_STATE_MOPEN) != 0)
		return (EBUSY);

	if ((mop = allocb(sizeof (struct stroptions), BPRI_MED)) == NULL) {
		DBG("zc_master_open(): mop allocation failed\n");
		return (ENOMEM);
	}

	zcs->zc_state |= ZC_STATE_MOPEN;

	/*
	 * q_ptr stores driver private data; stash the soft state data on both
	 * read and write sides of the queue.
	 */
	WR(rqp)->q_ptr = rqp->q_ptr = zcs;
	qprocson(rqp);

	/*
	 * Following qprocson(), the master side is fully plumbed into the
	 * STREAM and may send/receive messages.  Setting zcs->zc_master_rdq
	 * will allow the slave to send messages to us (the master).
	 * This cannot occur before qprocson() because the master is not
	 * ready to process them until that point.
	 */
	zcs->zc_master_rdq = rqp;

	/*
	 * set up hi/lo water marks on stream head read queue and add
	 * controlling tty as needed.
	 */
	mop->b_datap->db_type = M_SETOPTS;
	mop->b_wptr += sizeof (struct stroptions);
	sop = (struct stroptions *)(void *)mop->b_rptr;
	if (oflag & FNOCTTY)
		sop->so_flags = SO_HIWAT | SO_LOWAT;
	else
		sop->so_flags = SO_HIWAT | SO_LOWAT | SO_ISTTY;
	sop->so_hiwat = _TTY_BUFSIZ;
	sop->so_lowat = 256;
	putnext(rqp, mop);

	return (0);
}

/*
 * Upstream messages are passed unchanged.
 * If a hangup occurs the target is no longer usable, so deprecate it.
 */
static int
wcmrput(queue_t *q, mblk_t *mp)
{
	if (DB_TYPE(mp) == M_HANGUP)
		/* Don't block waiting for outstanding operations to complete */
		srpop(q->q_stream->sd_vnode, B_FALSE);
	putnext(q, mp);
	return (0);
}

static void
t_wput(queue_t *q, mblk_t *mp)
{
	trace();
	if (mp->b_datap->db_type < QPCTL && (q->q_count || !canputnext(q)))
		putq(q, mp);
	else
		putnext(q, mp);
}

/*
 *  -------------------------------------------------------------------------
 *
 *  WRITE QUEUE PUT and SRV routines
 *
 *  -------------------------------------------------------------------------
 *
 *  SSCOP WPUT - Message from above.
 *
 *  If the message is priority message we attempt to process it immediately.
 *  If the message is non-priority message, but there are no messages on the
 *  queue yet, we attempt to process it immediately.  If the message is not
 *  supported, we pass it down-queue if possible.  If the message cannot be
 *  processed immediately, we place it on the queue.
 */
STATIC int
sscop_wput(queue_t *q, mblk_t *mp)
{
	mblk_t *mp;
	int err = -EOPNOTSUPP;

	if (q->q_count && mp->b_datap->db_type < QPCTL) {
		putq(q, mp);
		/* 
		 *  NOTE:- after placing messages on the queue here, I should
		 *  check for transmit congestion.  I should check if placing
		 *  the message on the queue crosses a band threshold for
		 *  congestion onset and abatement.  When crossing congestion
		 *  thresholds, I should notify MTP3.
		 */
		return (0);
	}
	switch (mp->b_datap->db_type) {
	case M_DATA:
		if ((err = sscop_w_data(q, mp)))
			break;
		return (0);
	case M_PROTO:
		if ((err = sscop_w_proto(q, mp)))
			break;
		return (0);
	case M_PCPROTO:
		if ((err = sscop_w_pcproto(q, mp)))
			break;
		return (0);
	case M_CTL:
		if ((err = sscop_w_ctl(q, mp)))
			break;
		return (0);
	case M_IOCTL:
		if ((err = sscop_w_ioctl(q, mp)))
			break;
		return (0);
	case M_FLUSH:
		sscop_w_flush(q, mp);
		return (0);
	}
	switch (err) {
	case -EAGAIN:
		if (mp->b_datap->db_type < QPCTL) {
			putq(q, mp);
			return (0);
		}
	case -EOPNOTSUPP:
		if (q->q_next) {
			putnext(q, mp);
			return (0);
		}
	}
	freemsg(mp);
	return (err);
}

/*
 *  SDL_DISCONNECT_IND
 *  -----------------------------------
 */
STATIC INLINE int
sdl_disconnect_ind(queue_t *q, mblk_t *mp)
{
	union SDL_primitives *p = (union SDL_primitives *) mp->b_rptr;

	p->sdl_primitive = SDL_DISCONNECT_IND;
	putnext(q, mp);
	return (QR_ABSORBED);
}