Unleashed v1.4

[unleashed.git] / kernel / os / putnext.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 2009 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

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


/*
 *              UNIX Device Driver Interface functions
 *      This file contains the C-versions of putnext() and put().
 *      Assembly language versions exist for some architectures.
 */

#include <sys/types.h>
#include <sys/systm.h>
#include <sys/cpuvar.h>
#include <sys/debug.h>
#include <sys/t_lock.h>
#include <sys/stream.h>
#include <sys/thread.h>
#include <sys/strsubr.h>
#include <sys/ddi.h>
#include <sys/vtrace.h>
#include <sys/cmn_err.h>
#include <sys/strft.h>
#include <sys/stack.h>
#include <sys/archsystm.h>

/*
 * Streams with many modules may create long chains of calls via putnext() which
 * may exhaust stack space. When putnext detects that the stack space left is
 * too small (less then PUT_STACK_NEEDED), the call chain is broken and
 * further processing is delegated to the background thread via call to
 * putnext_tail(). Unfortunately there is no generic solution with fixed stack
 * size, and putnext() is recursive function, so this hack is a necessary evil.
 *
 * The redzone value is chosen dependent on the default stack size which is 8K
 * on 32-bit kernels and on x86 and 16K on 64-bit kernels. The values are chosen
 * empirically. For 64-bit kernels it is 5000 and for 32-bit kernels it is 3000.
 * Experiments showed that 2500 is not enough for either 32-bit or 64-bit
 * kernels.
 *
 * The redzone value is a tuneable rather then a constant to allow adjustments
 * in the field.
 *
 * The check in PUT_STACK_NOTENOUGH is taken from segkp_map_red() function. It
 * is possible to define it as a generic function exported by seg_kp, but
 *
 * a) It may sound like an open invitation to use the facility indiscriminately.
 * b) It adds extra function call in putnext path.
 *
 * We keep a global counter `put_stack_notenough' which keeps track how many
 * times the stack switching hack was used.
 */

static ulong_t put_stack_notenough;

#ifdef  _LP64
#define PUT_STACK_NEEDED 5000
#else
#define PUT_STACK_NEEDED 3000
#endif

int put_stack_needed = PUT_STACK_NEEDED;

#if defined(STACK_GROWTH_DOWN)
#define PUT_STACK_NOTENOUGH()                                   \
        (((STACK_BIAS + (uintptr_t)getfp() -                    \
            (uintptr_t)curthread->t_stkbase) < put_stack_needed) && \
        ++put_stack_notenough)
#else
#error  "STACK_GROWTH_DOWN undefined"
#endif

boolean_t       UseFastlocks = B_FALSE;

/*
 * function: putnext()
 * purpose:  call the put routine of the queue linked to qp
 *
 * Note: this function is written to perform well on modern computer
 * architectures by e.g. preloading values into registers and "smearing" out
 * code.
 *
 * A note on the fastput mechanism.  The most significant bit of a
 * putcount is considered the "FASTPUT" bit.  If set, then there is
 * nothing stoping a concurrent put from occuring (note that putcounts
 * are only allowed on CIPUT perimiters).  If, however, it is cleared,
 * then we need to take the normal lock path by aquiring the SQLOCK.
 * This is a slowlock.  When a thread starts exclusiveness, e.g. wants
 * writer access, it will clear the FASTPUT bit, causing new threads
 * to take the slowlock path.  This assures that putcounts will not
 * increase in value, so the want-writer does not need to constantly
 * aquire the putlocks to sum the putcounts.  This does have the
 * possibility of having the count drop right after reading, but that
 * is no different than aquiring, reading and then releasing.  However,
 * in this mode, it cannot go up, so eventually they will drop to zero
 * and the want-writer can proceed.
 *
 * If the FASTPUT bit is set, or in the slowlock path we see that there
 * are no writers or want-writers, we make the choice of calling the
 * putproc, or a "fast-fill_syncq".  The fast-fill is a fill with
 * immediate intention to drain.  This is done because there are
 * messages already at the queue waiting to drain.  To preserve message
 * ordering, we need to put this message at the end, and pickup the
 * messages at the beginning.  We call the macro that actually
 * enqueues the message on the queue, and then call qdrain_syncq.  If
 * there is already a drainer, we just return.  We could make that
 * check before calling qdrain_syncq, but it is a little more clear
 * to have qdrain_syncq do this (we might try the above optimization
 * as this behavior evolves).  qdrain_syncq assumes that SQ_EXCL is set
 * already if this is a non-CIPUT perimiter, and that an appropriate
 * claim has been made.  So we do all that work before dropping the
 * SQLOCK with our claim.
 *
 * If we cannot proceed with the putproc/fast-fill, we just fall
 * through to the qfill_syncq, and then tail processing.  If state
 * has changed in that cycle, or wakeups are needed, it will occur
 * there.
 */
void
putnext(queue_t *qp, mblk_t *mp)
{
        queue_t         *fqp = qp; /* For strft tracing */
        syncq_t         *sq;
        uint16_t        flags;
        uint16_t        drain_mask;
        struct qinit    *qi;
        int             (*putproc)();
        struct stdata   *stp;
        int             ix;
        boolean_t       queued = B_FALSE;
        kmutex_t        *sdlock = NULL;
        kmutex_t        *sqciplock = NULL;
        ushort_t        *sqcipcount = NULL;

        TRACE_2(TR_FAC_STREAMS_FR, TR_PUTNEXT_START,
            "putnext_start:(%p, %p)", qp, mp);

        ASSERT(mp->b_datap->db_ref != 0);
        ASSERT(mp->b_next == NULL && mp->b_prev == NULL);
        stp = STREAM(qp);
        ASSERT(stp != NULL);
        if (stp->sd_ciputctrl != NULL) {
                ix = CPU->cpu_seqid & stp->sd_nciputctrl;
                sdlock = &stp->sd_ciputctrl[ix].ciputctrl_lock;
                mutex_enter(sdlock);
        } else {
                mutex_enter(sdlock = &stp->sd_lock);
        }
        qp = qp->q_next;
        sq = qp->q_syncq;
        ASSERT(sq != NULL);
        ASSERT(MUTEX_NOT_HELD(SQLOCK(sq)));
        qi = qp->q_qinfo;

        if (sq->sq_ciputctrl != NULL) {
                /* fastlock: */
                ASSERT(sq->sq_flags & SQ_CIPUT);
                ix = CPU->cpu_seqid & sq->sq_nciputctrl;
                sqciplock = &sq->sq_ciputctrl[ix].ciputctrl_lock;
                sqcipcount = &sq->sq_ciputctrl[ix].ciputctrl_count;
                mutex_enter(sqciplock);
                if (!((*sqcipcount) & SQ_FASTPUT) ||
                    (sq->sq_flags & (SQ_STAYAWAY|SQ_EXCL|SQ_EVENTS))) {
                        mutex_exit(sqciplock);
                        sqciplock = NULL;
                        goto slowlock;
                }
                mutex_exit(sdlock);
                (*sqcipcount)++;
                ASSERT(*sqcipcount != 0);
                queued = qp->q_sqflags & Q_SQQUEUED;
                mutex_exit(sqciplock);
        } else {
        slowlock:
                ASSERT(sqciplock == NULL);
                mutex_enter(SQLOCK(sq));
                mutex_exit(sdlock);
                flags = sq->sq_flags;
                /*
                 * We are going to drop SQLOCK, so make a claim to prevent syncq
                 * from closing.
                 */
                sq->sq_count++;
                ASSERT(sq->sq_count != 0);              /* Wraparound */
                /*
                 * If there are writers or exclusive waiters, there is not much
                 * we can do.  Place the message on the syncq and schedule a
                 * background thread to drain it.
                 *
                 * Also if we are approaching end of stack, fill the syncq and
                 * switch processing to a background thread - see comments on
                 * top.
                 */
                if ((flags & (SQ_STAYAWAY|SQ_EXCL|SQ_EVENTS)) ||
                    (sq->sq_needexcl != 0) || PUT_STACK_NOTENOUGH()) {

                        TRACE_3(TR_FAC_STREAMS_FR, TR_PUTNEXT_END,
                            "putnext_end:(%p, %p, %p) SQ_EXCL fill",
                            qp, mp, sq);

                        /*
                         * NOTE: qfill_syncq will need QLOCK. It is safe to drop
                         * SQLOCK because positive sq_count keeps the syncq from
                         * closing.
                         */
                        mutex_exit(SQLOCK(sq));

                        qfill_syncq(sq, qp, mp);
                        /*
                         * NOTE: after the call to qfill_syncq() qp may be
                         * closed, both qp and sq should not be referenced at
                         * this point.
                         *
                         * This ASSERT is located here to prevent stack frame
                         * consumption in the DEBUG code.
                         */
                        ASSERT(sqciplock == NULL);
                        return;
                }

                queued = qp->q_sqflags & Q_SQQUEUED;
                /*
                 * If not a concurrent perimiter, we need to acquire
                 * it exclusively.  It could not have been previously
                 * set since we held the SQLOCK before testing
                 * SQ_GOAWAY above (which includes SQ_EXCL).
                 * We do this here because we hold the SQLOCK, and need
                 * to make this state change BEFORE dropping it.
                 */
                if (!(flags & SQ_CIPUT)) {
                        ASSERT((sq->sq_flags & SQ_EXCL) == 0);
                        ASSERT(!(sq->sq_type & SQ_CIPUT));
                        sq->sq_flags |= SQ_EXCL;
                }
                mutex_exit(SQLOCK(sq));
        }

        ASSERT((sq->sq_flags & (SQ_EXCL|SQ_CIPUT)));
        ASSERT(MUTEX_NOT_HELD(SQLOCK(sq)));

        /*
         * We now have a claim on the syncq, we are either going to
         * put the message on the syncq and then drain it, or we are
         * going to call the putproc().
         */
        putproc = qi->qi_putp;
        if (!queued) {
                STR_FTEVENT_MSG(mp, fqp, FTEV_PUTNEXT, mp->b_rptr -
                    mp->b_datap->db_base);
                (*putproc)(qp, mp);
                ASSERT(MUTEX_NOT_HELD(SQLOCK(sq)));
                ASSERT(MUTEX_NOT_HELD(QLOCK(qp)));
        } else {
                mutex_enter(QLOCK(qp));
                /*
                 * If there are no messages in front of us, just call putproc(),
                 * otherwise enqueue the message and drain the queue.
                 */
                if (qp->q_syncqmsgs == 0) {
                        mutex_exit(QLOCK(qp));
                        STR_FTEVENT_MSG(mp, fqp, FTEV_PUTNEXT, mp->b_rptr -
                            mp->b_datap->db_base);
                        (*putproc)(qp, mp);
                        ASSERT(MUTEX_NOT_HELD(SQLOCK(sq)));
                } else {
                        /*
                         * We are doing a fill with the intent to
                         * drain (meaning we are filling because
                         * there are messages in front of us ane we
                         * need to preserve message ordering)
                         * Therefore, put the message on the queue
                         * and call qdrain_syncq (must be done with
                         * the QLOCK held).
                         */
                        STR_FTEVENT_MSG(mp, fqp, FTEV_PUTNEXT,
                            mp->b_rptr - mp->b_datap->db_base);

#ifdef DEBUG
                        /*
                         * These two values were in the original code for
                         * all syncq messages.  This is unnecessary in
                         * the current implementation, but was retained
                         * in debug mode as it is usefull to know where
                         * problems occur.
                         */
                        mp->b_queue = qp;
                        mp->b_prev = (mblk_t *)putproc;
#endif
                        SQPUT_MP(qp, mp);
                        qdrain_syncq(sq, qp);
                        ASSERT(MUTEX_NOT_HELD(QLOCK(qp)));
                }
        }
        /*
         * Before we release our claim, we need to see if any
         * events were posted. If the syncq is SQ_EXCL && SQ_QUEUED,
         * we were responsible for going exclusive and, therefore,
         * are resposible for draining.
         */
        if (sq->sq_flags & (SQ_EXCL)) {
                drain_mask = 0;
        } else {
                drain_mask = SQ_QUEUED;
        }

        if (sqciplock != NULL) {
                mutex_enter(sqciplock);
                flags = sq->sq_flags;
                ASSERT(flags & SQ_CIPUT);
                /* SQ_EXCL could have been set by qwriter_inner */
                if ((flags & (SQ_EXCL|SQ_TAIL)) || sq->sq_needexcl) {
                        /*
                         * we need SQLOCK to handle
                         * wakeups/drains/flags change.  sqciplock
                         * is needed to decrement sqcipcount.
                         * SQLOCK has to be grabbed before sqciplock
                         * for lock ordering purposes.
                         * after sqcipcount is decremented some lock
                         * still needs to be held to make sure
                         * syncq won't get freed on us.
                         *
                         * To prevent deadlocks we try to grab SQLOCK and if it
                         * is held already we drop sqciplock, acquire SQLOCK and
                         * reacqwire sqciplock again.
                         */
                        if (mutex_tryenter(SQLOCK(sq)) == 0) {
                                mutex_exit(sqciplock);
                                mutex_enter(SQLOCK(sq));
                                mutex_enter(sqciplock);
                        }
                        flags = sq->sq_flags;
                        ASSERT(*sqcipcount != 0);
                        (*sqcipcount)--;
                        mutex_exit(sqciplock);
                } else {
                        ASSERT(*sqcipcount != 0);
                        (*sqcipcount)--;
                        mutex_exit(sqciplock);
                        TRACE_3(TR_FAC_STREAMS_FR, TR_PUTNEXT_END,
                        "putnext_end:(%p, %p, %p) done", qp, mp, sq);
                        return;
                }
        } else {
                mutex_enter(SQLOCK(sq));
                flags = sq->sq_flags;
                ASSERT(sq->sq_count != 0);
                sq->sq_count--;
        }
        if ((flags & (SQ_TAIL)) || sq->sq_needexcl) {
                putnext_tail(sq, qp, (flags & ~drain_mask));
                /*
                 * The only purpose of this ASSERT is to preserve calling stack
                 * in DEBUG kernel.
                 */
                ASSERT(sq != NULL);
                return;
        }
        ASSERT((sq->sq_flags & (SQ_EXCL|SQ_CIPUT)) || queued);
        ASSERT((flags & (SQ_EXCL|SQ_CIPUT)) || queued);
        /*
         * Safe to always drop SQ_EXCL:
         *      Not SQ_CIPUT means we set SQ_EXCL above
         *      For SQ_CIPUT SQ_EXCL will only be set if the put
         *      procedure did a qwriter(INNER) in which case
         *      nobody else is in the inner perimeter and we
         *      are exiting.
         *
         * I would like to make the following assertion:
         *
         * ASSERT((flags & (SQ_EXCL|SQ_CIPUT)) != (SQ_EXCL|SQ_CIPUT) ||
         *      sq->sq_count == 0);
         *
         * which indicates that if we are both putshared and exclusive,
         * we became exclusive while executing the putproc, and the only
         * claim on the syncq was the one we dropped a few lines above.
         * But other threads that enter putnext while the syncq is exclusive
         * need to make a claim as they may need to drop SQLOCK in the
         * has_writers case to avoid deadlocks.  If these threads are
         * delayed or preempted, it is possible that the writer thread can
         * find out that there are other claims making the (sq_count == 0)
         * test invalid.
         */

        sq->sq_flags = flags & ~SQ_EXCL;
        mutex_exit(SQLOCK(sq));
        TRACE_3(TR_FAC_STREAMS_FR, TR_PUTNEXT_END,
            "putnext_end:(%p, %p, %p) done", qp, mp, sq);
}


/*
 * wrapper for qi_putp entry in module ops vec.
 * implements asynchronous putnext().
 * Note, that unlike putnext(), this routine is NOT optimized for the
 * fastpath.  Calling this routine will grab whatever locks are necessary
 * to protect the stream head, q_next, and syncq's.
 * And since it is in the normal locks path, we do not use putlocks if
 * they exist (though this can be changed by swapping the value of
 * UseFastlocks).
 */
void
put(queue_t *qp, mblk_t *mp)
{
        queue_t         *fqp = qp; /* For strft tracing */
        syncq_t         *sq;
        uint16_t        flags;
        uint16_t        drain_mask;
        struct qinit    *qi;
        int             (*putproc)();
        int             ix;
        boolean_t       queued = B_FALSE;
        kmutex_t        *sqciplock = NULL;
        ushort_t        *sqcipcount = NULL;

        TRACE_2(TR_FAC_STREAMS_FR, TR_PUT_START,
            "put:(%X, %X)", qp, mp);
        ASSERT(mp->b_datap->db_ref != 0);
        ASSERT(mp->b_next == NULL && mp->b_prev == NULL);

        sq = qp->q_syncq;
        ASSERT(sq != NULL);
        qi = qp->q_qinfo;

        if (UseFastlocks && sq->sq_ciputctrl != NULL) {
                /* fastlock: */
                ASSERT(sq->sq_flags & SQ_CIPUT);
                ix = CPU->cpu_seqid & sq->sq_nciputctrl;
                sqciplock = &sq->sq_ciputctrl[ix].ciputctrl_lock;
                sqcipcount = &sq->sq_ciputctrl[ix].ciputctrl_count;
                mutex_enter(sqciplock);
                if (!((*sqcipcount) & SQ_FASTPUT) ||
                    (sq->sq_flags & (SQ_STAYAWAY|SQ_EXCL|SQ_EVENTS))) {
                        mutex_exit(sqciplock);
                        sqciplock = NULL;
                        goto slowlock;
                }
                (*sqcipcount)++;
                ASSERT(*sqcipcount != 0);
                queued = qp->q_sqflags & Q_SQQUEUED;
                mutex_exit(sqciplock);
        } else {
        slowlock:
                ASSERT(sqciplock == NULL);
                mutex_enter(SQLOCK(sq));
                flags = sq->sq_flags;
                /*
                 * We are going to drop SQLOCK, so make a claim to prevent syncq
                 * from closing.
                 */
                sq->sq_count++;
                ASSERT(sq->sq_count != 0);              /* Wraparound */
                /*
                 * If there are writers or exclusive waiters, there is not much
                 * we can do.  Place the message on the syncq and schedule a
                 * background thread to drain it.
                 *
                 * Also if we are approaching end of stack, fill the syncq and
                 * switch processing to a background thread - see comments on
                 * top.
                 */
                if ((flags & (SQ_STAYAWAY|SQ_EXCL|SQ_EVENTS)) ||
                    (sq->sq_needexcl != 0) || PUT_STACK_NOTENOUGH()) {

                        TRACE_3(TR_FAC_STREAMS_FR, TR_PUTNEXT_END,
                            "putnext_end:(%p, %p, %p) SQ_EXCL fill",
                            qp, mp, sq);

                        /*
                         * NOTE: qfill_syncq will need QLOCK. It is safe to drop
                         * SQLOCK because positive sq_count keeps the syncq from
                         * closing.
                         */
                        mutex_exit(SQLOCK(sq));

                        qfill_syncq(sq, qp, mp);
                        /*
                         * NOTE: after the call to qfill_syncq() qp may be
                         * closed, both qp and sq should not be referenced at
                         * this point.
                         *
                         * This ASSERT is located here to prevent stack frame
                         * consumption in the DEBUG code.
                         */
                        ASSERT(sqciplock == NULL);
                        return;
                }

                queued = qp->q_sqflags & Q_SQQUEUED;
                /*
                 * If not a concurrent perimiter, we need to acquire
                 * it exclusively.  It could not have been previously
                 * set since we held the SQLOCK before testing
                 * SQ_GOAWAY above (which includes SQ_EXCL).
                 * We do this here because we hold the SQLOCK, and need
                 * to make this state change BEFORE dropping it.
                 */
                if (!(flags & SQ_CIPUT)) {
                        ASSERT((sq->sq_flags & SQ_EXCL) == 0);
                        ASSERT(!(sq->sq_type & SQ_CIPUT));
                        sq->sq_flags |= SQ_EXCL;
                }
                mutex_exit(SQLOCK(sq));
        }

        ASSERT((sq->sq_flags & (SQ_EXCL|SQ_CIPUT)));
        ASSERT(MUTEX_NOT_HELD(SQLOCK(sq)));

        /*
         * We now have a claim on the syncq, we are either going to
         * put the message on the syncq and then drain it, or we are
         * going to call the putproc().
         */
        putproc = qi->qi_putp;
        if (!queued) {
                STR_FTEVENT_MSG(mp, fqp, FTEV_PUTNEXT, mp->b_rptr -
                    mp->b_datap->db_base);
                (*putproc)(qp, mp);
                ASSERT(MUTEX_NOT_HELD(SQLOCK(sq)));
                ASSERT(MUTEX_NOT_HELD(QLOCK(qp)));
        } else {
                mutex_enter(QLOCK(qp));
                /*
                 * If there are no messages in front of us, just call putproc(),
                 * otherwise enqueue the message and drain the queue.
                 */
                if (qp->q_syncqmsgs == 0) {
                        mutex_exit(QLOCK(qp));
                        STR_FTEVENT_MSG(mp, fqp, FTEV_PUTNEXT, mp->b_rptr -
                            mp->b_datap->db_base);
                        (*putproc)(qp, mp);
                        ASSERT(MUTEX_NOT_HELD(SQLOCK(sq)));
                } else {
                        /*
                         * We are doing a fill with the intent to
                         * drain (meaning we are filling because
                         * there are messages in front of us ane we
                         * need to preserve message ordering)
                         * Therefore, put the message on the queue
                         * and call qdrain_syncq (must be done with
                         * the QLOCK held).
                         */
                        STR_FTEVENT_MSG(mp, fqp, FTEV_PUTNEXT,
                            mp->b_rptr - mp->b_datap->db_base);

#ifdef DEBUG
                        /*
                         * These two values were in the original code for
                         * all syncq messages.  This is unnecessary in
                         * the current implementation, but was retained
                         * in debug mode as it is usefull to know where
                         * problems occur.
                         */
                        mp->b_queue = qp;
                        mp->b_prev = (mblk_t *)putproc;
#endif
                        SQPUT_MP(qp, mp);
                        qdrain_syncq(sq, qp);
                        ASSERT(MUTEX_NOT_HELD(QLOCK(qp)));
                }
        }
        /*
         * Before we release our claim, we need to see if any
         * events were posted. If the syncq is SQ_EXCL && SQ_QUEUED,
         * we were responsible for going exclusive and, therefore,
         * are resposible for draining.
         */
        if (sq->sq_flags & (SQ_EXCL)) {
                drain_mask = 0;
        } else {
                drain_mask = SQ_QUEUED;
        }

        if (sqciplock != NULL) {
                mutex_enter(sqciplock);
                flags = sq->sq_flags;
                ASSERT(flags & SQ_CIPUT);
                /* SQ_EXCL could have been set by qwriter_inner */
                if ((flags & (SQ_EXCL|SQ_TAIL)) || sq->sq_needexcl) {
                        /*
                         * we need SQLOCK to handle
                         * wakeups/drains/flags change.  sqciplock
                         * is needed to decrement sqcipcount.
                         * SQLOCK has to be grabbed before sqciplock
                         * for lock ordering purposes.
                         * after sqcipcount is decremented some lock
                         * still needs to be held to make sure
                         * syncq won't get freed on us.
                         *
                         * To prevent deadlocks we try to grab SQLOCK and if it
                         * is held already we drop sqciplock, acquire SQLOCK and
                         * reacqwire sqciplock again.
                         */
                        if (mutex_tryenter(SQLOCK(sq)) == 0) {
                                mutex_exit(sqciplock);
                                mutex_enter(SQLOCK(sq));
                                mutex_enter(sqciplock);
                        }
                        flags = sq->sq_flags;
                        ASSERT(*sqcipcount != 0);
                        (*sqcipcount)--;
                        mutex_exit(sqciplock);
                } else {
                        ASSERT(*sqcipcount != 0);
                        (*sqcipcount)--;
                        mutex_exit(sqciplock);
                        TRACE_3(TR_FAC_STREAMS_FR, TR_PUTNEXT_END,
                        "putnext_end:(%p, %p, %p) done", qp, mp, sq);
                        return;
                }
        } else {
                mutex_enter(SQLOCK(sq));
                flags = sq->sq_flags;
                ASSERT(sq->sq_count != 0);
                sq->sq_count--;
        }
        if ((flags & (SQ_TAIL)) || sq->sq_needexcl) {
                putnext_tail(sq, qp, (flags & ~drain_mask));
                /*
                 * The only purpose of this ASSERT is to preserve calling stack
                 * in DEBUG kernel.
                 */
                ASSERT(sq != NULL);
                return;
        }
        ASSERT((sq->sq_flags & (SQ_EXCL|SQ_CIPUT)) || queued);
        ASSERT((flags & (SQ_EXCL|SQ_CIPUT)) || queued);
        /*
         * Safe to always drop SQ_EXCL:
         *      Not SQ_CIPUT means we set SQ_EXCL above
         *      For SQ_CIPUT SQ_EXCL will only be set if the put
         *      procedure did a qwriter(INNER) in which case
         *      nobody else is in the inner perimeter and we
         *      are exiting.
         *
         * I would like to make the following assertion:
         *
         * ASSERT((flags & (SQ_EXCL|SQ_CIPUT)) != (SQ_EXCL|SQ_CIPUT) ||
         *      sq->sq_count == 0);
         *
         * which indicates that if we are both putshared and exclusive,
         * we became exclusive while executing the putproc, and the only
         * claim on the syncq was the one we dropped a few lines above.
         * But other threads that enter putnext while the syncq is exclusive
         * need to make a claim as they may need to drop SQLOCK in the
         * has_writers case to avoid deadlocks.  If these threads are
         * delayed or preempted, it is possible that the writer thread can
         * find out that there are other claims making the (sq_count == 0)
         * test invalid.
         */

        sq->sq_flags = flags & ~SQ_EXCL;
        mutex_exit(SQLOCK(sq));
        TRACE_3(TR_FAC_STREAMS_FR, TR_PUTNEXT_END,
            "putnext_end:(%p, %p, %p) done", qp, mp, sq);
}