fs/sockfs/sockcommon_subr.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 (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
 */
/*
 * Copyright 2014, OmniTI Computer Consulting, Inc. All rights reserved.
 * Copyright 2019 Joyent, Inc.
 */

#include <sys/types.h>
#include <sys/param.h>
#include <sys/signal.h>
#include <sys/cmn_err.h>

#include <sys/stropts.h>
#include <sys/socket.h>
#include <sys/socketvar.h>
#include <sys/sockio.h>
#include <sys/strsubr.h>
#include <sys/strsun.h>
#include <sys/atomic.h>
#include <sys/tihdr.h>

#include <fs/sockfs/sockcommon.h>
#include <fs/sockfs/sockfilter_impl.h>
#include <fs/sockfs/socktpi.h>
#include <fs/sockfs/sodirect.h>
#include <sys/ddi.h>
#include <inet/ip.h>
#include <sys/time.h>
#include <sys/cmn_err.h>

#ifdef SOCK_TEST
extern int do_useracc;
extern clock_t sock_test_timelimit;
#endif /* SOCK_TEST */

#define	MBLK_PULL_LEN 64
uint32_t so_mblk_pull_len = MBLK_PULL_LEN;

#ifdef DEBUG
boolean_t so_debug_length = B_FALSE;
static boolean_t so_check_length(sonode_t *so);
#endif

static int
so_acceptq_dequeue_locked(struct sonode *so, boolean_t dontblock,
    struct sonode **nsop)
{
	struct sonode *nso = NULL;

	*nsop = NULL;
	ASSERT(MUTEX_HELD(&so->so_acceptq_lock));
	while ((nso = list_remove_head(&so->so_acceptq_list)) == NULL) {
		/*
		 * No need to check so_error here, because it is not
		 * possible for a listening socket to be reset or otherwise
		 * disconnected.
		 *
		 * So now we just need check if it's ok to wait.
		 */
		if (dontblock)
			return (EWOULDBLOCK);
		if (so->so_state & (SS_CLOSING | SS_FALLBACK_PENDING))
			return (EINTR);

		if (cv_wait_sig_swap(&so->so_acceptq_cv,
		    &so->so_acceptq_lock) == 0)
			return (EINTR);
	}

	ASSERT(nso != NULL);
	ASSERT(so->so_acceptq_len > 0);
	so->so_acceptq_len--;
	nso->so_listener = NULL;

	*nsop = nso;

	return (0);
}

/*
 * int so_acceptq_dequeue(struct sonode *, boolean_t, struct sonode **)
 *
 * Pulls a connection off of the accept queue.
 *
 * Arguments:
 *   so	       - listening socket
 *   dontblock - indicate whether it's ok to sleep if there are no
 *		 connections on the queue
 *   nsop      - Value-return argument
 *
 * Return values:
 *   0 when a connection is successfully dequeued, in which case nsop
 *   is set to point to the new connection. Upon failure a non-zero
 *   value is returned, and the value of nsop is set to NULL.
 *
 * Note:
 *   so_acceptq_dequeue() may return prematurly if the socket is falling
 *   back to TPI.
 */
int
so_acceptq_dequeue(struct sonode *so, boolean_t dontblock,
    struct sonode **nsop)
{
	int error;

	mutex_enter(&so->so_acceptq_lock);
	error = so_acceptq_dequeue_locked(so, dontblock, nsop);
	mutex_exit(&so->so_acceptq_lock);

	return (error);
}

static void
so_acceptq_flush_impl(struct sonode *so, list_t *list, boolean_t doclose)
{
	struct sonode *nso;

	while ((nso = list_remove_head(list)) != NULL) {
		nso->so_listener = NULL;
		if (doclose) {
			(void) socket_close(nso, 0, CRED());
		} else {
			/*
			 * Only used for fallback - not possible when filters
			 * are present.
			 */
			ASSERT(so->so_filter_active == 0);
			/*
			 * Since the socket is on the accept queue, there can
			 * only be one reference. We drop the reference and
			 * just blow off the socket.
			 */
			ASSERT(nso->so_count == 1);
			nso->so_count--;
			/* drop the proto ref */
			VN_RELE(SOTOV(nso));
		}
		socket_destroy(nso);
	}
}
/*
 * void so_acceptq_flush(struct sonode *so)
 *
 * Removes all pending connections from a listening socket, and
 * frees the associated resources.
 *
 * Arguments
 *   so	     - listening socket
 *   doclose - make a close downcall for each socket on the accept queue
 *
 * Return values:
 *   None.
 *
 * Note:
 *   The caller has to ensure that no calls to so_acceptq_enqueue() or
 *   so_acceptq_dequeue() occur while the accept queue is being flushed.
 *   So either the socket needs to be in a state where no operations
 *   would come in, or so_lock needs to be obtained.
 */
void
so_acceptq_flush(struct sonode *so, boolean_t doclose)
{
	so_acceptq_flush_impl(so, &so->so_acceptq_list, doclose);
	so_acceptq_flush_impl(so, &so->so_acceptq_defer, doclose);

	so->so_acceptq_len = 0;
}

int
so_wait_connected_locked(struct sonode *so, boolean_t nonblock,
    sock_connid_t id)
{
	ASSERT(MUTEX_HELD(&so->so_lock));

	/*
	 * The protocol has notified us that a connection attempt is being
	 * made, so before we wait for a notification to arrive we must
	 * clear out any errors associated with earlier connection attempts.
	 */
	if (so->so_error != 0 && SOCK_CONNID_LT(so->so_proto_connid, id))
		so->so_error = 0;

	while (SOCK_CONNID_LT(so->so_proto_connid, id)) {
		if (nonblock)
			return (EINPROGRESS);

		if (so->so_state & (SS_CLOSING | SS_FALLBACK_PENDING))
			return (EINTR);

		if (cv_wait_sig_swap(&so->so_state_cv, &so->so_lock) == 0)
			return (EINTR);
	}

	if (so->so_error != 0)
		return (sogeterr(so, B_TRUE));
	/*
	 * Under normal circumstances, so_error should contain an error
	 * in case the connect failed. However, it is possible for another
	 * thread to come in a consume the error, so generate a sensible
	 * error in that case.
	 */
	if ((so->so_state & SS_ISCONNECTED) == 0)
		return (ECONNREFUSED);

	return (0);
}

/*
 * int so_wait_connected(struct sonode *so, boolean_t nonblock,
 *    sock_connid_t id)
 *
 * Wait until the socket is connected or an error has occured.
 *
 * Arguments:
 *   so	      - socket
 *   nonblock - indicate whether it's ok to sleep if the connection has
 *		not yet been established
 *   gen      - generation number that was returned by the protocol
 *		when the operation was started
 *
 * Returns:
 *   0 if the connection attempt was successful, or an error indicating why
 *   the connection attempt failed.
 */
int
so_wait_connected(struct sonode *so, boolean_t nonblock, sock_connid_t id)
{
	int error;

	mutex_enter(&so->so_lock);
	error = so_wait_connected_locked(so, nonblock, id);
	mutex_exit(&so->so_lock);

	return (error);
}

int
so_snd_wait_qnotfull_locked(struct sonode *so, boolean_t dontblock)
{
	int error;

	ASSERT(MUTEX_HELD(&so->so_lock));
	while (SO_SND_FLOWCTRLD(so)) {
		if (so->so_state & SS_CANTSENDMORE)
			return (EPIPE);
		if (dontblock)
			return (EWOULDBLOCK);

		if (so->so_state & (SS_CLOSING | SS_FALLBACK_PENDING))
			return (EINTR);

		if (so->so_sndtimeo == 0) {
			/*
			 * Zero means disable timeout.
			 */
			error = cv_wait_sig(&so->so_snd_cv, &so->so_lock);
		} else {
			error = cv_reltimedwait_sig(&so->so_snd_cv,
			    &so->so_lock, so->so_sndtimeo, TR_CLOCK_TICK);
		}
		if (error == 0)
			return (EINTR);
		else if (error == -1)
			return (EAGAIN);
	}
	return (0);
}

/*
 * int so_wait_sendbuf(struct sonode *so, boolean_t dontblock)
 *
 * Wait for the transport to notify us about send buffers becoming
 * available.
 */
int
so_snd_wait_qnotfull(struct sonode *so, boolean_t dontblock)
{
	int error = 0;

	mutex_enter(&so->so_lock);
	so->so_snd_wakeup = B_TRUE;
	error = so_snd_wait_qnotfull_locked(so, dontblock);
	so->so_snd_wakeup = B_FALSE;
	mutex_exit(&so->so_lock);

	return (error);
}

void
so_snd_qfull(struct sonode *so)
{
	mutex_enter(&so->so_lock);
	so->so_snd_qfull = B_TRUE;
	mutex_exit(&so->so_lock);
}

void
so_snd_qnotfull(struct sonode *so)
{
	mutex_enter(&so->so_lock);
	so->so_snd_qfull = B_FALSE;
	/* wake up everyone waiting for buffers */
	cv_broadcast(&so->so_snd_cv);
	mutex_exit(&so->so_lock);
}

/*
 * Change the process/process group to which SIGIO is sent.
 */
int
socket_chgpgrp(struct sonode *so, pid_t pid)
{
	int error;

	ASSERT(MUTEX_HELD(&so->so_lock));
	if (pid != 0) {
		/*
		 * Permissions check by sending signal 0.
		 * Note that when kill fails it does a
		 * set_errno causing the system call to fail.
		 */
		error = kill(pid, 0);
		if (error != 0) {
			return (error);
		}
	}
	so->so_pgrp = pid;
	return (0);
}


/*
 * Generate a SIGIO, for 'writable' events include siginfo structure,
 * for read events just send the signal.
 */
/*ARGSUSED*/
static void
socket_sigproc(proc_t *proc, int event)
{
	k_siginfo_t info;

	ASSERT(event & (SOCKETSIG_WRITE | SOCKETSIG_READ | SOCKETSIG_URG));

	if (event & SOCKETSIG_WRITE) {
		info.si_signo = SIGPOLL;
		info.si_code = POLL_OUT;
		info.si_errno = 0;
		info.si_fd = 0;
		info.si_band = 0;
		sigaddq(proc, NULL, &info, KM_NOSLEEP);
	}
	if (event & SOCKETSIG_READ) {
		sigtoproc(proc, NULL, SIGPOLL);
	}
	if (event & SOCKETSIG_URG) {
		sigtoproc(proc, NULL, SIGURG);
	}
}

void
socket_sendsig(struct sonode *so, int event)
{
	proc_t *proc;

	ASSERT(MUTEX_HELD(&so->so_lock));

	if (so->so_pgrp == 0 || (!(so->so_state & SS_ASYNC) &&
	    event != SOCKETSIG_URG)) {
		return;
	}

	dprint(3, ("sending sig %d to %d\n", event, so->so_pgrp));

	if (so->so_pgrp > 0) {
		/*
		 * XXX This unfortunately still generates
		 * a signal when a fd is closed but
		 * the proc is active.
		 */
		mutex_enter(&pidlock);
		/*
		 * Even if the thread started in another zone, we're receiving
		 * on behalf of this socket's zone, so find the proc using the
		 * socket's zone ID.
		 */
		proc = prfind_zone(so->so_pgrp, so->so_zoneid);
		if (proc == NULL) {
			mutex_exit(&pidlock);
			return;
		}
		mutex_enter(&proc->p_lock);
		mutex_exit(&pidlock);
		socket_sigproc(proc, event);
		mutex_exit(&proc->p_lock);
	} else {
		/*
		 * Send to process group. Hold pidlock across
		 * calls to socket_sigproc().
		 */
		pid_t pgrp = -so->so_pgrp;

		mutex_enter(&pidlock);
		/*
		 * Even if the thread started in another zone, we're receiving
		 * on behalf of this socket's zone, so find the pgrp using the
		 * socket's zone ID.
		 */
		proc = pgfind_zone(pgrp, so->so_zoneid);
		while (proc != NULL) {
			mutex_enter(&proc->p_lock);
			socket_sigproc(proc, event);
			mutex_exit(&proc->p_lock);
			proc = proc->p_pglink;
		}
		mutex_exit(&pidlock);
	}
}

#define	MIN(a, b) ((a) < (b) ? (a) : (b))
/* Copy userdata into a new mblk_t */
mblk_t *
socopyinuio(uio_t *uiop, ssize_t iosize, size_t wroff, ssize_t maxblk,
    size_t tail_len, int *errorp)
{
	mblk_t	*head = NULL, **tail = &head;

	ASSERT(iosize == INFPSZ || iosize > 0);

	if (iosize == INFPSZ || iosize > uiop->uio_resid)
		iosize = uiop->uio_resid;

	if (maxblk == INFPSZ)
		maxblk = iosize;

	/* Nothing to do in these cases, so we're done */
	if (iosize < 0 || maxblk < 0 || (maxblk == 0 && iosize > 0))
		goto done;

	/*
	 * We will enter the loop below if iosize is 0; it will allocate an
	 * empty message block and call uiomove(9F) which will just return.
	 * We could avoid that with an extra check but would only slow
	 * down the much more likely case where iosize is larger than 0.
	 */
	do {
		ssize_t blocksize;
		mblk_t	*mp;

		blocksize = MIN(iosize, maxblk);
		ASSERT(blocksize >= 0);
		mp = allocb(wroff + blocksize + tail_len, BPRI_MED);
		if (mp == NULL) {
			*errorp = ENOMEM;
			return (head);
		}
		mp->b_rptr += wroff;
		mp->b_wptr = mp->b_rptr + blocksize;

		*tail = mp;
		tail = &mp->b_cont;

		/* uiomove(9F) either returns 0 or EFAULT */
		if ((*errorp = uiomove(mp->b_rptr, (size_t)blocksize,
		    UIO_WRITE, uiop)) != 0) {
			ASSERT(*errorp != ENOMEM);
			freemsg(head);
			return (NULL);
		}

		iosize -= blocksize;
	} while (iosize > 0);

done:
	*errorp = 0;
	return (head);
}

mblk_t *
socopyoutuio(mblk_t *mp, struct uio *uiop, ssize_t max_read, int *errorp)
{
	int error;
	ptrdiff_t n;
	mblk_t *nmp;

	ASSERT(mp->b_wptr >= mp->b_rptr);

	/*
	 * max_read is the offset of the oobmark and read can not go pass
	 * the oobmark.
	 */
	if (max_read == INFPSZ || max_read > uiop->uio_resid)
		max_read = uiop->uio_resid;

	do {
		if ((n = MIN(max_read, MBLKL(mp))) != 0) {
			ASSERT(n > 0);

			error = uiomove(mp->b_rptr, n, UIO_READ, uiop);
			if (error != 0) {
				freemsg(mp);
				*errorp = error;
				return (NULL);
			}
		}

		mp->b_rptr += n;
		max_read -= n;
		while (mp != NULL && (mp->b_rptr >= mp->b_wptr)) {
			/*
			 * get rid of zero length mblks
			 */
			nmp = mp;
			mp = mp->b_cont;
			freeb(nmp);
		}
	} while (mp != NULL && max_read > 0);

	*errorp = 0;
	return (mp);
}

static void
so_prepend_msg(struct sonode *so, mblk_t *mp, mblk_t *last_tail)
{
	ASSERT(last_tail != NULL);
	mp->b_next = so->so_rcv_q_head;
	mp->b_prev = last_tail;
	ASSERT(!(DB_FLAGS(mp) & DBLK_UIOA));

	if (so->so_rcv_q_head == NULL) {
		ASSERT(so->so_rcv_q_last_head == NULL);
		so->so_rcv_q_last_head = mp;
#ifdef DEBUG
	} else {
		ASSERT(!(DB_FLAGS(so->so_rcv_q_head) & DBLK_UIOA));
#endif
	}
	so->so_rcv_q_head = mp;

#ifdef DEBUG
	if (so_debug_length) {
		mutex_enter(&so->so_lock);
		ASSERT(so_check_length(so));
		mutex_exit(&so->so_lock);
	}
#endif
}

/*
 * Move a mblk chain (mp_head, mp_last_head) to the sonode's rcv queue so it
 * can be processed by so_dequeue_msg().
 */
void
so_process_new_message(struct sonode *so, mblk_t *mp_head, mblk_t *mp_last_head)
{
	if (so->so_filter_active > 0 &&
	    (mp_head = sof_filter_data_in_proc(so, mp_head,
	    &mp_last_head)) == NULL)
		return;

	ASSERT(mp_head->b_prev != NULL);
	if (so->so_rcv_q_head == NULL) {
		so->so_rcv_q_head = mp_head;
		so->so_rcv_q_last_head = mp_last_head;
		ASSERT(so->so_rcv_q_last_head->b_prev != NULL);
	} else {
		boolean_t flag_equal = ((DB_FLAGS(mp_head) & DBLK_UIOA) ==
		    (DB_FLAGS(so->so_rcv_q_last_head) & DBLK_UIOA));

		if (mp_head->b_next == NULL &&
		    DB_TYPE(mp_head) == M_DATA &&
		    DB_TYPE(so->so_rcv_q_last_head) == M_DATA && flag_equal) {
			so->so_rcv_q_last_head->b_prev->b_cont = mp_head;
			so->so_rcv_q_last_head->b_prev = mp_head->b_prev;
			mp_head->b_prev = NULL;
		} else if (flag_equal && (DB_FLAGS(mp_head) & DBLK_UIOA)) {
			/*
			 * Append to last_head if more than one mblks, and both
			 * mp_head and last_head are I/OAT mblks.
			 */
			ASSERT(mp_head->b_next != NULL);
			so->so_rcv_q_last_head->b_prev->b_cont = mp_head;
			so->so_rcv_q_last_head->b_prev = mp_head->b_prev;
			mp_head->b_prev = NULL;

			so->so_rcv_q_last_head->b_next = mp_head->b_next;
			mp_head->b_next = NULL;
			so->so_rcv_q_last_head = mp_last_head;
		} else {
#ifdef DEBUG
			{
				mblk_t *tmp_mblk;
				tmp_mblk = mp_head;
				while (tmp_mblk != NULL) {
					ASSERT(tmp_mblk->b_prev != NULL);
					tmp_mblk = tmp_mblk->b_next;
				}
			}
#endif
			so->so_rcv_q_last_head->b_next = mp_head;
			so->so_rcv_q_last_head = mp_last_head;
		}
	}
}

/*
 * Check flow control on a given sonode.  Must have so_lock held, and
 * this function will release the hold.  Return true if flow control
 * is cleared.
 */
boolean_t
so_check_flow_control(struct sonode *so)
{
	ASSERT(MUTEX_HELD(&so->so_lock));

	if (so->so_flowctrld && (so->so_rcv_queued < so->so_rcvlowat &&
	    !(so->so_state & SS_FIL_RCV_FLOWCTRL))) {
		so->so_flowctrld = B_FALSE;
		mutex_exit(&so->so_lock);
		/*
		 * Open up flow control. SCTP does not have any downcalls, and
		 * it will clr flow ctrl in sosctp_recvmsg().
		 */
		if (so->so_downcalls != NULL &&
		    so->so_downcalls->sd_clr_flowctrl != NULL) {
			(*so->so_downcalls->sd_clr_flowctrl)
			    (so->so_proto_handle);
		}
		/* filters can start injecting data */
		sof_sonode_notify_filters(so, SOF_EV_INJECT_DATA_IN_OK, 0);
		return (B_TRUE);
	} else {
		mutex_exit(&so->so_lock);
		return (B_FALSE);
	}
}

int
so_dequeue_msg(struct sonode *so, mblk_t **mctlp, struct uio *uiop,
    rval_t *rvalp, int flags)
{
	mblk_t	*mp, *nmp;
	mblk_t	*savemp, *savemptail;
	mblk_t	*new_msg_head;
	mblk_t	*new_msg_last_head;
	mblk_t	*last_tail;
	boolean_t partial_read;
	boolean_t reset_atmark = B_FALSE;
	int more = 0;
	int error;
	ssize_t oobmark;
	sodirect_t *sodp = so->so_direct;

	partial_read = B_FALSE;
	*mctlp = NULL;
again:
	mutex_enter(&so->so_lock);
again1:
#ifdef DEBUG
	if (so_debug_length) {
		ASSERT(so_check_length(so));
	}
#endif
	if (so->so_state & SS_RCVATMARK) {
		/* Check whether the caller is OK to read past the mark */
		if (flags & MSG_NOMARK) {
			mutex_exit(&so->so_lock);
			return (EWOULDBLOCK);
		}
		reset_atmark = B_TRUE;
	}
	/*
	 * First move messages from the dump area to processing area
	 */
	if (sodp != NULL) {
		if (sodp->sod_enabled) {
			if (sodp->sod_uioa.uioa_state & UIOA_ALLOC) {
				/* nothing to uioamove */
				sodp = NULL;
			} else if (sodp->sod_uioa.uioa_state & UIOA_INIT) {
				sodp->sod_uioa.uioa_state &= UIOA_CLR;
				sodp->sod_uioa.uioa_state |= UIOA_ENABLED;
				/*
				 * try to uioamove() the data that
				 * has already queued.
				 */
				sod_uioa_so_init(so, sodp, uiop);
			}
		} else {
			sodp = NULL;
		}
	}
	new_msg_head = so->so_rcv_head;
	new_msg_last_head = so->so_rcv_last_head;
	so->so_rcv_head = NULL;
	so->so_rcv_last_head = NULL;
	oobmark = so->so_oobmark;
	/*
	 * We can release the lock as there can only be one reader
	 */
	mutex_exit(&so->so_lock);

	if (new_msg_head != NULL) {
		so_process_new_message(so, new_msg_head, new_msg_last_head);
	}
	savemp = savemptail = NULL;
	rvalp->r_vals = 0;
	error = 0;
	mp = so->so_rcv_q_head;

	if (mp != NULL &&
	    (so->so_rcv_timer_tid == 0 ||
	    so->so_rcv_queued >= so->so_rcv_thresh)) {
		partial_read = B_FALSE;

		if (flags & MSG_PEEK) {
			if ((nmp = dupmsg(mp)) == NULL &&
			    (nmp = copymsg(mp)) == NULL) {
				size_t size = msgsize(mp);

				error = strwaitbuf(size, BPRI_HI);
				if (error) {
					return (error);
				}
				goto again;
			}
			mp = nmp;
		} else {
			ASSERT(mp->b_prev != NULL);
			last_tail = mp->b_prev;
			mp->b_prev = NULL;
			so->so_rcv_q_head = mp->b_next;
			if (so->so_rcv_q_head == NULL) {
				so->so_rcv_q_last_head = NULL;
			}
			mp->b_next = NULL;
		}

		ASSERT(mctlp != NULL);
		/*
		 * First process PROTO or PCPROTO blocks, if any.
		 */
		if (DB_TYPE(mp) != M_DATA) {
			*mctlp = mp;
			savemp = mp;
			savemptail = mp;
			ASSERT(DB_TYPE(mp) == M_PROTO ||
			    DB_TYPE(mp) == M_PCPROTO);
			while (mp->b_cont != NULL &&
			    DB_TYPE(mp->b_cont) != M_DATA) {
				ASSERT(DB_TYPE(mp->b_cont) == M_PROTO ||
				    DB_TYPE(mp->b_cont) == M_PCPROTO);
				mp = mp->b_cont;
				savemptail = mp;
			}
			mp = savemptail->b_cont;
			savemptail->b_cont = NULL;
		}

		ASSERT(DB_TYPE(mp) == M_DATA);
		/*
		 * Now process DATA blocks, if any. Note that for sodirect
		 * enabled socket, uio_resid can be 0.
		 */
		if (uiop->uio_resid >= 0) {
			ssize_t copied = 0;

			if (sodp != NULL && (DB_FLAGS(mp) & DBLK_UIOA)) {
				mutex_enter(&so->so_lock);
				ASSERT(uiop == (uio_t *)&sodp->sod_uioa);
				copied = sod_uioa_mblk(so, mp);
				if (copied > 0)
					partial_read = B_TRUE;
				mutex_exit(&so->so_lock);
				/* mark this mblk as processed */
				mp = NULL;
			} else {
				ssize_t oldresid = uiop->uio_resid;

				if (MBLKL(mp) < so_mblk_pull_len) {
					if (pullupmsg(mp, -1) == 1) {
						last_tail = mp;
					}
				}
				/*
				 * Can not read beyond the oobmark
				 */
				mp = socopyoutuio(mp, uiop,
				    oobmark == 0 ? INFPSZ : oobmark, &error);
				if (error != 0) {
					freemsg(*mctlp);
					*mctlp = NULL;
					more = 0;
					goto done;
				}
				ASSERT(oldresid >= uiop->uio_resid);
				copied = oldresid - uiop->uio_resid;
				if (oldresid > uiop->uio_resid)
					partial_read = B_TRUE;
			}
			ASSERT(copied >= 0);
			if (copied > 0 && !(flags & MSG_PEEK)) {
				mutex_enter(&so->so_lock);
				so->so_rcv_queued -= copied;
				ASSERT(so->so_oobmark >= 0);
				if (so->so_oobmark > 0) {
					so->so_oobmark -= copied;
					ASSERT(so->so_oobmark >= 0);
					if (so->so_oobmark == 0) {
						ASSERT(so->so_state &
						    SS_OOBPEND);
						so->so_oobmark = 0;
						so->so_state |= SS_RCVATMARK;
					}
				}
				/*
				 * so_check_flow_control() will drop
				 * so->so_lock.
				 */
				rvalp->r_val2 = so_check_flow_control(so);
			}
		}
		if (mp != NULL) { /* more data blocks in msg */
			more |= MOREDATA;
			if ((flags & (MSG_PEEK|MSG_TRUNC))) {
				if (flags & MSG_PEEK) {
					freemsg(mp);
				} else {
					unsigned int msize = msgdsize(mp);

					freemsg(mp);
					mutex_enter(&so->so_lock);
					so->so_rcv_queued -= msize;
					/*
					 * so_check_flow_control() will drop
					 * so->so_lock.
					 */
					rvalp->r_val2 =
					    so_check_flow_control(so);
				}
			} else if (partial_read && !somsghasdata(mp)) {
				/*
				 * Avoid queuing a zero-length tail part of
				 * a message. partial_read == 1 indicates that
				 * we read some of the message.
				 */
				freemsg(mp);
				more &= ~MOREDATA;
			} else {
				if (savemp != NULL &&
				    (flags & MSG_DUPCTRL)) {
					mblk_t *nmp;
					/*
					 * There should only be non data mblks
					 */
					ASSERT(DB_TYPE(savemp) != M_DATA &&
					    DB_TYPE(savemptail) != M_DATA);
try_again:
					if ((nmp = dupmsg(savemp)) == NULL &&
					    (nmp = copymsg(savemp)) == NULL) {

						size_t size = msgsize(savemp);

						error = strwaitbuf(size,
						    BPRI_HI);
						if (error != 0) {
							/*
							 * In case we
							 * cannot copy
							 * control data
							 * free the remaining
							 * data.
							 */
							freemsg(mp);
							goto done;
						}
						goto try_again;
					}

					ASSERT(nmp != NULL);
					ASSERT(DB_TYPE(nmp) != M_DATA);
					savemptail->b_cont = mp;
					*mctlp = nmp;
					mp = savemp;
				}
				/*
				 * putback mp
				 */
				so_prepend_msg(so, mp, last_tail);
			}
		}

		/* fast check so_rcv_head if there is more data */
		if (partial_read && !(so->so_state & SS_RCVATMARK) &&
		    *mctlp == NULL && uiop->uio_resid > 0 &&
		    !(flags & MSG_PEEK) && so->so_rcv_head != NULL) {
			goto again;
		}
	} else if (!partial_read) {
		mutex_enter(&so->so_lock);
		if (so->so_error != 0) {
			error = sogeterr(so, !(flags & MSG_PEEK));
			mutex_exit(&so->so_lock);
			return (error);
		}

		/* See if new data has arrived in the meantime */
		if (so->so_rcv_head != NULL)
			goto again1;

		/*
		 * No pending data. Return right away for nonblocking
		 * socket, otherwise sleep waiting for data.
		 */
		if (!(so->so_state & SS_CANTRCVMORE) && uiop->uio_resid > 0) {
			if ((uiop->uio_fmode & (FNDELAY|FNONBLOCK)) ||
			    (flags & MSG_DONTWAIT)) {
				error = EWOULDBLOCK;
			} else {
				if (so->so_state & (SS_CLOSING |
				    SS_FALLBACK_PENDING)) {
					mutex_exit(&so->so_lock);
					error = EINTR;
					goto done;
				}

				so->so_rcv_wakeup = B_TRUE;
				so->so_rcv_wanted = uiop->uio_resid;
				if (so->so_rcvtimeo == 0) {
					/*
					 * Zero means disable timeout.
					 */
					error = cv_wait_sig(&so->so_rcv_cv,
					    &so->so_lock);
				} else {
					error = cv_reltimedwait_sig(
					    &so->so_rcv_cv, &so->so_lock,
					    so->so_rcvtimeo, TR_CLOCK_TICK);
				}
				so->so_rcv_wakeup = B_FALSE;
				so->so_rcv_wanted = 0;

				if (error == 0) {
					error = EINTR;
				} else if (error == -1) {
					error = EAGAIN;
				} else {
					goto again1;
				}
			}
		}
		mutex_exit(&so->so_lock);
	}
	if (reset_atmark && partial_read && !(flags & MSG_PEEK)) {
		/*
		 * We are passed the mark, update state
		 * 4.3BSD and 4.4BSD clears the mark when peeking across it.
		 * The draft Posix socket spec states that the mark should
		 * not be cleared when peeking. We follow the latter.
		 */
		mutex_enter(&so->so_lock);
		ASSERT(so_verify_oobstate(so));
		so->so_state &= ~(SS_OOBPEND|SS_HAVEOOBDATA|SS_RCVATMARK);
		freemsg(so->so_oobmsg);
		so->so_oobmsg = NULL;
		ASSERT(so_verify_oobstate(so));
		mutex_exit(&so->so_lock);
	}
	ASSERT(so->so_rcv_wakeup == B_FALSE);
done:
	if (sodp != NULL) {
		mutex_enter(&so->so_lock);
		if (sodp->sod_enabled &&
		    (sodp->sod_uioa.uioa_state & UIOA_ENABLED)) {
			SOD_UIOAFINI(sodp);
			if (sodp->sod_uioa.uioa_mbytes > 0) {
				ASSERT(so->so_rcv_q_head != NULL ||
				    so->so_rcv_head != NULL);
				so->so_rcv_queued -= sod_uioa_mblk(so, NULL);
				if (error == EWOULDBLOCK)
					error = 0;
			}
		}
		mutex_exit(&so->so_lock);
	}
#ifdef DEBUG
	if (so_debug_length) {
		mutex_enter(&so->so_lock);
		ASSERT(so_check_length(so));
		mutex_exit(&so->so_lock);
	}
#endif
	rvalp->r_val1 = more;
	ASSERT(MUTEX_NOT_HELD(&so->so_lock));
	return (error);
}

/*
 * Enqueue data from the protocol on the socket's rcv queue.
 *
 * We try to hook new M_DATA mblks onto an existing chain, however,
 * that cannot be done if the existing chain has already been
 * processed by I/OAT. Non-M_DATA mblks are just linked together via
 * b_next. In all cases the b_prev of the enqueued mblk is set to
 * point to the last mblk in its b_cont chain.
 */
void
so_enqueue_msg(struct sonode *so, mblk_t *mp, size_t msg_size)
{
	ASSERT(MUTEX_HELD(&so->so_lock));

#ifdef DEBUG
	if (so_debug_length) {
		ASSERT(so_check_length(so));
	}
#endif
	so->so_rcv_queued += msg_size;

	if (so->so_rcv_head == NULL) {
		ASSERT(so->so_rcv_last_head == NULL);
		so->so_rcv_head = mp;
		so->so_rcv_last_head = mp;
	} else if ((DB_TYPE(mp) == M_DATA &&
	    DB_TYPE(so->so_rcv_last_head) == M_DATA) &&
	    ((DB_FLAGS(mp) & DBLK_UIOA) ==
	    (DB_FLAGS(so->so_rcv_last_head) & DBLK_UIOA))) {
		/* Added to the end */
		ASSERT(so->so_rcv_last_head != NULL);
		ASSERT(so->so_rcv_last_head->b_prev != NULL);
		so->so_rcv_last_head->b_prev->b_cont = mp;
	} else {
		/* Start a new end */
		so->so_rcv_last_head->b_next = mp;
		so->so_rcv_last_head = mp;
	}
	while (mp->b_cont != NULL)
		mp = mp->b_cont;

	so->so_rcv_last_head->b_prev = mp;
#ifdef DEBUG
	if (so_debug_length) {
		ASSERT(so_check_length(so));
	}
#endif
}

/*
 * Return B_TRUE if there is data in the message, B_FALSE otherwise.
 */
boolean_t
somsghasdata(mblk_t *mp)
{
	for (; mp; mp = mp->b_cont)
		if (mp->b_datap->db_type == M_DATA) {
			ASSERT(mp->b_wptr >= mp->b_rptr);
			if (mp->b_wptr > mp->b_rptr)
				return (B_TRUE);
		}
	return (B_FALSE);
}

/*
 * Flush the read side of sockfs.
 *
 * The caller must be sure that a reader is not already active when the
 * buffer is being flushed.
 */
void
so_rcv_flush(struct sonode *so)
{
	mblk_t  *mp;

	ASSERT(MUTEX_HELD(&so->so_lock));

	if (so->so_oobmsg != NULL) {
		freemsg(so->so_oobmsg);
		so->so_oobmsg = NULL;
		so->so_oobmark = 0;
		so->so_state &=
		    ~(SS_OOBPEND|SS_HAVEOOBDATA|SS_HADOOBDATA|SS_RCVATMARK);
	}

	/*
	 * Free messages sitting in the recv queues
	 */
	while (so->so_rcv_q_head != NULL) {
		mp = so->so_rcv_q_head;
		so->so_rcv_q_head = mp->b_next;
		mp->b_next = mp->b_prev = NULL;
		freemsg(mp);
	}
	while (so->so_rcv_head != NULL) {
		mp = so->so_rcv_head;
		so->so_rcv_head = mp->b_next;
		mp->b_next = mp->b_prev = NULL;
		freemsg(mp);
	}
	so->so_rcv_queued = 0;
	so->so_rcv_q_head = NULL;
	so->so_rcv_q_last_head = NULL;
	so->so_rcv_head = NULL;
	so->so_rcv_last_head = NULL;
}

/*
 * Handle recv* calls that set MSG_OOB or MSG_OOB together with MSG_PEEK.
 */
int
sorecvoob(struct sonode *so, struct nmsghdr *msg, struct uio *uiop, int flags,
    boolean_t oob_inline)
{
	mblk_t		*mp, *nmp;
	int		error;

	dprintso(so, 1, ("sorecvoob(%p, %p, 0x%x)\n", (void *)so, (void *)msg,
	    flags));

	if (msg != NULL) {
		/*
		 * There is never any oob data with addresses or control since
		 * the T_EXDATA_IND does not carry any options.
		 */
		msg->msg_controllen = 0;
		msg->msg_namelen = 0;
		msg->msg_flags = 0;
	}

	mutex_enter(&so->so_lock);
	ASSERT(so_verify_oobstate(so));
	if (oob_inline ||
	    (so->so_state & (SS_OOBPEND|SS_HADOOBDATA)) != SS_OOBPEND) {
		dprintso(so, 1, ("sorecvoob: inline or data consumed\n"));
		mutex_exit(&so->so_lock);
		return (EINVAL);
	}
	if (!(so->so_state & SS_HAVEOOBDATA)) {
		dprintso(so, 1, ("sorecvoob: no data yet\n"));
		mutex_exit(&so->so_lock);
		return (EWOULDBLOCK);
	}
	ASSERT(so->so_oobmsg != NULL);
	mp = so->so_oobmsg;
	if (flags & MSG_PEEK) {
		/*
		 * Since recv* can not return ENOBUFS we can not use dupmsg.
		 * Instead we revert to the consolidation private
		 * allocb_wait plus bcopy.
		 */
		mblk_t *mp1;

		mp1 = allocb_wait(msgdsize(mp), BPRI_MED, STR_NOSIG, NULL);
		ASSERT(mp1);

		while (mp != NULL) {
			ssize_t size;

			size = MBLKL(mp);
			bcopy(mp->b_rptr, mp1->b_wptr, size);
			mp1->b_wptr += size;
			ASSERT(mp1->b_wptr <= mp1->b_datap->db_lim);
			mp = mp->b_cont;
		}
		mp = mp1;
	} else {
		/*
		 * Update the state indicating that the data has been consumed.
		 * Keep SS_OOBPEND set until data is consumed past the mark.
		 */
		so->so_oobmsg = NULL;
		so->so_state ^= SS_HAVEOOBDATA|SS_HADOOBDATA;
	}
	ASSERT(so_verify_oobstate(so));
	mutex_exit(&so->so_lock);

	error = 0;
	nmp = mp;
	while (nmp != NULL && uiop->uio_resid > 0) {
		ssize_t n = MBLKL(nmp);

		n = MIN(n, uiop->uio_resid);
		if (n > 0)
			error = uiomove(nmp->b_rptr, n,
			    UIO_READ, uiop);
		if (error)
			break;
		nmp = nmp->b_cont;
	}
	ASSERT(mp->b_next == NULL && mp->b_prev == NULL);
	freemsg(mp);
	return (error);
}

/*
 * Allocate and initializ sonode
 */
/* ARGSUSED */
struct sonode *
socket_sonode_create(struct sockparams *sp, int family, int type,
    int protocol, int version, int sflags, int *errorp, struct cred *cr)
{
	sonode_t *so;
	int	kmflags;

	/*
	 * Choose the right set of sonodeops based on the upcall and
	 * down call version that the protocol has provided
	 */
	if (SOCK_UC_VERSION != sp->sp_smod_info->smod_uc_version ||
	    SOCK_DC_VERSION != sp->sp_smod_info->smod_dc_version) {
		/*
		 * mismatch
		 */
#ifdef DEBUG
		cmn_err(CE_CONT, "protocol and socket module version mismatch");
#endif
		*errorp = EINVAL;
		return (NULL);
	}

	kmflags = (sflags & SOCKET_NOSLEEP) ? KM_NOSLEEP : KM_SLEEP;

	so = kmem_cache_alloc(socket_cache, kmflags);
	if (so == NULL) {
		*errorp = ENOMEM;
		return (NULL);
	}

	sonode_init(so, sp, family, type, protocol, &so_sonodeops);

	if (version == SOV_DEFAULT)
		version = so_default_version;

	so->so_version = (short)version;

	/*
	 * set the default values to be INFPSZ
	 * if a protocol desires it can change the value later
	 */
	so->so_proto_props.sopp_rxhiwat = SOCKET_RECVHIWATER;
	so->so_proto_props.sopp_rxlowat = SOCKET_RECVLOWATER;
	so->so_proto_props.sopp_maxpsz = INFPSZ;
	so->so_proto_props.sopp_maxblk = INFPSZ;

	return (so);
}

int
socket_init_common(struct sonode *so, struct sonode *pso, int flags, cred_t *cr)
{
	int error = 0;

	if (pso != NULL) {
		/*
		 * We have a passive open, so inherit basic state from
		 * the parent (listener).
		 *
		 * No need to grab the new sonode's lock, since there is no
		 * one that can have a reference to it.
		 */
		mutex_enter(&pso->so_lock);

		so->so_state |= SS_ISCONNECTED | (pso->so_state & SS_ASYNC);
		so->so_pgrp = pso->so_pgrp;
		so->so_rcvtimeo = pso->so_rcvtimeo;
		so->so_sndtimeo = pso->so_sndtimeo;
		so->so_xpg_rcvbuf = pso->so_xpg_rcvbuf;
		/*
		 * Make note of the socket level options. TCP and IP level
		 * options are already inherited. We could do all this after
		 * accept is successful but doing it here simplifies code and
		 * no harm done for error case.
		 */
		so->so_options = pso->so_options & (SO_DEBUG|SO_REUSEADDR|
		    SO_KEEPALIVE|SO_DONTROUTE|SO_BROADCAST|SO_USELOOPBACK|
		    SO_OOBINLINE|SO_DGRAM_ERRIND|SO_LINGER);
		so->so_proto_props = pso->so_proto_props;
		so->so_mode = pso->so_mode;
		so->so_pollev = pso->so_pollev & SO_POLLEV_ALWAYS;

		mutex_exit(&pso->so_lock);

		/*
		 * If the parent has any filters, try to inherit them.
		 */
		if (pso->so_filter_active > 0 &&
		    (error = sof_sonode_inherit_filters(so, pso)) != 0)
			return (error);

	} else {
		struct sockparams *sp = so->so_sockparams;
		sock_upcalls_t *upcalls_to_use;

		/*
		 * Attach automatic filters, if there are any.
		 */
		if (!list_is_empty(&sp->sp_auto_filters) &&
		    (error = sof_sonode_autoattach_filters(so, cr)) != 0)
			return (error);

		/* OK to attach filters */
		so->so_state |= SS_FILOP_OK;

		/*
		 * Based on the version number select the right upcalls to
		 * pass down. Currently we only have one version so choose
		 * default
		 */
		upcalls_to_use = &so_upcalls;

		/* active open, so create a lower handle */
		so->so_proto_handle =
		    sp->sp_smod_info->smod_proto_create_func(so->so_family,
		    so->so_type, so->so_protocol, &so->so_downcalls,
		    &so->so_mode, &error, flags, cr);

		if (so->so_proto_handle == NULL) {
			ASSERT(error != 0);
			/*
			 * To be safe; if a lower handle cannot be created, and
			 * the proto does not give a reason why, assume there
			 * was a lack of memory.
			 */
			return ((error == 0) ? ENOMEM : error);
		}
		ASSERT(so->so_downcalls != NULL);
		ASSERT(so->so_downcalls->sd_send != NULL ||
		    so->so_downcalls->sd_send_uio != NULL);
		if (so->so_downcalls->sd_recv_uio != NULL) {
			ASSERT(so->so_downcalls->sd_poll != NULL);
			so->so_pollev |= SO_POLLEV_ALWAYS;
		}

		(*so->so_downcalls->sd_activate)(so->so_proto_handle,
		    (sock_upper_handle_t)so, upcalls_to_use, 0, cr);

		/* Wildcard */

		/*
		 * FIXME No need for this, the protocol can deal with it in
		 * sd_create(). Should update ICMP.
		 */
		if (so->so_protocol != so->so_sockparams->sp_protocol) {
			int protocol = so->so_protocol;
			int error;
			/*
			 * Issue SO_PROTOTYPE setsockopt.
			 */
			error = socket_setsockopt(so, SOL_SOCKET, SO_PROTOTYPE,
			    &protocol, (t_uscalar_t)sizeof (protocol), cr);
			if (error) {
				(void) (*so->so_downcalls->sd_close)
				    (so->so_proto_handle, 0, cr);

				mutex_enter(&so->so_lock);
				so_rcv_flush(so);
				mutex_exit(&so->so_lock);
				/*
				 * Setsockopt often fails with ENOPROTOOPT but
				 * socket() should fail with
				 * EPROTONOSUPPORT/EPROTOTYPE.
				 */
				return (EPROTONOSUPPORT);
			}
		}
	}

	if (uioasync.enabled)
		sod_sock_init(so);

	/* put an extra reference on the socket for the protocol */
	VN_HOLD(SOTOV(so));

	return (0);
}

/*
 * int socket_ioctl_common(struct sonode *so, int cmd, intptr_t arg, int mode,
 *         struct cred *cr, int32_t *rvalp)
 *
 * Handle ioctls that manipulate basic socket state; non-blocking,
 * async, etc.
 *
 * Returns:
 *   < 0  - ioctl was not handle
 *  >= 0  - ioctl was handled, if > 0, then it is an errno
 *
 * Notes:
 *   Assumes the standard receive buffer is used to obtain info for
 *   NREAD.
 */
/* ARGSUSED */
int
socket_ioctl_common(struct sonode *so, int cmd, intptr_t arg, int mode,
    struct cred *cr, int32_t *rvalp)
{
	switch (cmd) {
	case SIOCSQPTR:
		/*
		 * SIOCSQPTR is valid only when helper stream is created
		 * by the protocol.
		 */

		return (EOPNOTSUPP);
	case FIONBIO: {
		int32_t value;

		if (so_copyin((void *)arg, &value, sizeof (int32_t),
		    (mode & (int)FKIOCTL)))
			return (EFAULT);

		mutex_enter(&so->so_lock);
		if (value) {
			so->so_state |= SS_NDELAY;
		} else {
			so->so_state &= ~SS_NDELAY;
		}
		mutex_exit(&so->so_lock);
		return (0);
	}
	case FIOASYNC: {
		int32_t value;

		if (so_copyin((void *)arg, &value, sizeof (int32_t),
		    (mode & (int)FKIOCTL)))
			return (EFAULT);

		mutex_enter(&so->so_lock);

		if (value) {
			/* Turn on SIGIO */
			so->so_state |= SS_ASYNC;
		} else {
			/* Turn off SIGIO */
			so->so_state &= ~SS_ASYNC;
		}
		mutex_exit(&so->so_lock);

		return (0);
	}

	case SIOCSPGRP:
	case FIOSETOWN: {
		int error;
		pid_t pid;

		if (so_copyin((void *)arg, &pid, sizeof (pid_t),
		    (mode & (int)FKIOCTL)))
			return (EFAULT);

		mutex_enter(&so->so_lock);
		error = (pid != so->so_pgrp) ? socket_chgpgrp(so, pid) : 0;
		mutex_exit(&so->so_lock);
		return (error);
	}
	case SIOCGPGRP:
	case FIOGETOWN:
		if (so_copyout(&so->so_pgrp, (void *)arg,
		    sizeof (pid_t), (mode & (int)FKIOCTL)))
			return (EFAULT);

		return (0);
	case SIOCATMARK: {
		int retval;

		/*
		 * Only protocols that support urgent data can handle ATMARK.
		 */
		if ((so->so_mode & SM_EXDATA) == 0)
			return (EINVAL);

		/*
		 * If the protocol is maintaining its own buffer, then the
		 * request must be passed down.
		 */
		if (so->so_downcalls->sd_recv_uio != NULL)
			return (-1);

		retval = (so->so_state & SS_RCVATMARK) != 0;

		if (so_copyout(&retval, (void *)arg, sizeof (int),
		    (mode & (int)FKIOCTL))) {
			return (EFAULT);
		}
		return (0);
	}

	case FIONREAD: {
		int retval;

		/*
		 * If the protocol is maintaining its own buffer, then the
		 * request must be passed down.
		 */
		if (so->so_downcalls->sd_recv_uio != NULL)
			return (-1);

		retval = MIN(so->so_rcv_queued, INT_MAX);

		if (so_copyout(&retval, (void *)arg,
		    sizeof (retval), (mode & (int)FKIOCTL))) {
			return (EFAULT);
		}
		return (0);
	}

	case _I_GETPEERCRED: {
		int error = 0;

		if ((mode & FKIOCTL) == 0)
			return (EINVAL);

		mutex_enter(&so->so_lock);
		if ((so->so_mode & SM_CONNREQUIRED) == 0) {
			error = ENOTSUP;
		} else if ((so->so_state & SS_ISCONNECTED) == 0) {
			error = ENOTCONN;
		} else if (so->so_peercred != NULL) {
			k_peercred_t *kp = (k_peercred_t *)arg;
			kp->pc_cr = so->so_peercred;
			kp->pc_cpid = so->so_cpid;
			crhold(so->so_peercred);
		} else {
			error = EINVAL;
		}
		mutex_exit(&so->so_lock);
		return (error);
	}
	default:
		return (-1);
	}
}

/*
 * Handle the I_NREAD STREAM ioctl.
 */
static int
so_strioc_nread(struct sonode *so, intptr_t arg, int mode, int32_t *rvalp)
{
	size_t size = 0;
	int retval;
	int count = 0;
	mblk_t *mp;
	clock_t wakeup = drv_usectohz(10);

	if (so->so_downcalls == NULL ||
	    so->so_downcalls->sd_recv_uio != NULL)
		return (EINVAL);

	mutex_enter(&so->so_lock);
	/* Wait for reader to get out of the way. */
	while (so->so_flag & SOREADLOCKED) {
		/*
		 * If reader is waiting for data, then there should be nothing
		 * on the rcv queue.
		 */
		if (so->so_rcv_wakeup)
			goto out;

		/* Do a timed sleep, in case the reader goes to sleep. */
		(void) cv_reltimedwait(&so->so_read_cv, &so->so_lock, wakeup,
		    TR_CLOCK_TICK);
	}

	/*
	 * Since we are holding so_lock no new reader will come in, and the
	 * protocol will not be able to enqueue data. So it's safe to walk
	 * both rcv queues.
	 */
	mp = so->so_rcv_q_head;
	if (mp != NULL) {
		size = msgdsize(so->so_rcv_q_head);
		for (; mp != NULL; mp = mp->b_next)
			count++;
	} else {
		/*
		 * In case the processing list was empty, get the size of the
		 * next msg in line.
		 */
		size = msgdsize(so->so_rcv_head);
	}

	for (mp = so->so_rcv_head; mp != NULL; mp = mp->b_next)
		count++;
out:
	mutex_exit(&so->so_lock);

	/*
	 * Drop down from size_t to the "int" required by the
	 * interface.  Cap at INT_MAX.
	 */
	retval = MIN(size, INT_MAX);
	if (so_copyout(&retval, (void *)arg, sizeof (retval),
	    (mode & (int)FKIOCTL))) {
		return (EFAULT);
	} else {
		*rvalp = count;
		return (0);
	}
}

/*
 * Process STREAM ioctls.
 *
 * Returns:
 *   < 0  - ioctl was not handle
 *  >= 0  - ioctl was handled, if > 0, then it is an errno
 */
int
socket_strioc_common(struct sonode *so, int cmd, intptr_t arg, int mode,
    struct cred *cr, int32_t *rvalp)
{
	int retval;

	/* Only STREAM iotcls are handled here */
	if ((cmd & 0xffffff00U) != STR)
		return (-1);

	switch (cmd) {
	case I_CANPUT:
		/*
		 * We return an error for I_CANPUT so that isastream(3C) will
		 * not report the socket as being a STREAM.
		 */
		return (EOPNOTSUPP);
	case I_NREAD:
		/* Avoid doing a fallback for I_NREAD. */
		return (so_strioc_nread(so, arg, mode, rvalp));
	case I_LOOK:
		/* Avoid doing a fallback for I_LOOK. */
		if (so_copyout("sockmod", (void *)arg, strlen("sockmod") + 1,
		    (mode & (int)FKIOCTL))) {
			return (EFAULT);
		}
		return (0);
	default:
		break;
	}

	/*
	 * Try to fall back to TPI, and if successful, reissue the ioctl.
	 */
	if ((retval = so_tpi_fallback(so, cr)) == 0) {
		/* Reissue the ioctl */
		ASSERT(so->so_rcv_q_head == NULL);
		return (SOP_IOCTL(so, cmd, arg, mode, cr, rvalp));
	} else {
		return (retval);
	}
}

/*
 * This is called for all socket types to verify that the buffer size is large
 * enough for the option, and if we can, handle the request as well. Most
 * options will be forwarded to the protocol.
 */
int
socket_getopt_common(struct sonode *so, int level, int option_name,
    void *optval, socklen_t *optlenp, int flags)
{
	if (level != SOL_SOCKET)
		return (-1);

	switch (option_name) {
	case SO_ERROR:
	case SO_DOMAIN:
	case SO_TYPE:
	case SO_ACCEPTCONN: {
		int32_t value;
		socklen_t optlen = *optlenp;

		if (optlen < (t_uscalar_t)sizeof (int32_t)) {
			return (EINVAL);
		}

		switch (option_name) {
		case SO_ERROR:
			mutex_enter(&so->so_lock);
			value = sogeterr(so, B_TRUE);
			mutex_exit(&so->so_lock);
			break;
		case SO_DOMAIN:
			value = so->so_family;
			break;
		case SO_TYPE:
			value = so->so_type;
			break;
		case SO_ACCEPTCONN:
			if (so->so_state & SS_ACCEPTCONN)
				value = SO_ACCEPTCONN;
			else
				value = 0;
			break;
		}

		bcopy(&value, optval, sizeof (value));
		*optlenp = sizeof (value);

		return (0);
	}
	case SO_SNDTIMEO:
	case SO_RCVTIMEO: {
		clock_t value;
		socklen_t optlen = *optlenp;

		if (get_udatamodel() == DATAMODEL_NONE ||
		    get_udatamodel() == DATAMODEL_NATIVE) {
			if (optlen < sizeof (struct timeval))
				return (EINVAL);
		} else {
			if (optlen < sizeof (struct timeval32))
				return (EINVAL);
		}
		if (option_name == SO_RCVTIMEO)
			value = drv_hztousec(so->so_rcvtimeo);
		else
			value = drv_hztousec(so->so_sndtimeo);

		if (get_udatamodel() == DATAMODEL_NONE ||
		    get_udatamodel() == DATAMODEL_NATIVE) {
			((struct timeval *)(optval))->tv_sec =
			    value / (1000 * 1000);
			((struct timeval *)(optval))->tv_usec =
			    value % (1000 * 1000);
			*optlenp = sizeof (struct timeval);
		} else {
			((struct timeval32 *)(optval))->tv_sec =
			    value / (1000 * 1000);
			((struct timeval32 *)(optval))->tv_usec =
			    value % (1000 * 1000);
			*optlenp = sizeof (struct timeval32);
		}
		return (0);
	}
	case SO_DEBUG:
	case SO_REUSEADDR:
	case SO_KEEPALIVE:
	case SO_DONTROUTE:
	case SO_BROADCAST:
	case SO_USELOOPBACK:
	case SO_OOBINLINE:
	case SO_SNDBUF:
#ifdef notyet
	case SO_SNDLOWAT:
	case SO_RCVLOWAT:
#endif /* notyet */
	case SO_DGRAM_ERRIND: {
		socklen_t optlen = *optlenp;

		if (optlen < (t_uscalar_t)sizeof (int32_t))
			return (EINVAL);
		break;
	}
	case SO_RCVBUF: {
		socklen_t optlen = *optlenp;

		if (optlen < (t_uscalar_t)sizeof (int32_t))
			return (EINVAL);

		if ((flags & _SOGETSOCKOPT_XPG4_2) && so->so_xpg_rcvbuf != 0) {
			/*
			 * XXX If SO_RCVBUF has been set and this is an
			 * XPG 4.2 application then do not ask the transport
			 * since the transport might adjust the value and not
			 * return exactly what was set by the application.
			 * For non-XPG 4.2 application we return the value
			 * that the transport is actually using.
			 */
			*(int32_t *)optval = so->so_xpg_rcvbuf;
			*optlenp = sizeof (so->so_xpg_rcvbuf);
			return (0);
		}
		/*
		 * If the option has not been set then get a default
		 * value from the transport.
		 */
		break;
	}
	case SO_LINGER: {
		socklen_t optlen = *optlenp;

		if (optlen < (t_uscalar_t)sizeof (struct linger))
			return (EINVAL);
		break;
	}
	case SO_SND_BUFINFO: {
		socklen_t optlen = *optlenp;

		if (optlen < (t_uscalar_t)sizeof (struct so_snd_bufinfo))
			return (EINVAL);
		((struct so_snd_bufinfo *)(optval))->sbi_wroff =
		    (so->so_proto_props).sopp_wroff;
		((struct so_snd_bufinfo *)(optval))->sbi_maxblk =
		    (so->so_proto_props).sopp_maxblk;
		((struct so_snd_bufinfo *)(optval))->sbi_maxpsz =
		    (so->so_proto_props).sopp_maxpsz;
		((struct so_snd_bufinfo *)(optval))->sbi_tail =
		    (so->so_proto_props).sopp_tail;
		*optlenp = sizeof (struct so_snd_bufinfo);
		return (0);
	}
	case SO_SND_COPYAVOID: {
		sof_instance_t *inst;

		/*
		 * Avoid zero-copy if there is a filter with a data_out
		 * callback. We could let the operation succeed, but then
		 * the filter would have to copy the data anyway.
		 */
		for (inst = so->so_filter_top; inst != NULL;
		    inst = inst->sofi_next) {
			if (SOF_INTERESTED(inst, data_out))
				return (EOPNOTSUPP);
		}
		break;
	}

	default:
		break;
	}

	/* Unknown Option */
	return (-1);
}

void
socket_sonode_destroy(struct sonode *so)
{
	sonode_fini(so);
	kmem_cache_free(socket_cache, so);
}

int
so_zcopy_wait(struct sonode *so)
{
	int error = 0;

	mutex_enter(&so->so_lock);
	while (!(so->so_copyflag & STZCNOTIFY)) {
		if (so->so_state & SS_CLOSING) {
			mutex_exit(&so->so_lock);
			return (EINTR);
		}
		if (cv_wait_sig(&so->so_copy_cv, &so->so_lock) == 0) {
			error = EINTR;
			break;
		}
	}
	so->so_copyflag &= ~STZCNOTIFY;
	mutex_exit(&so->so_lock);
	return (error);
}

void
so_timer_callback(void *arg)
{
	struct sonode *so = (struct sonode *)arg;

	mutex_enter(&so->so_lock);

	so->so_rcv_timer_tid = 0;
	if (so->so_rcv_queued > 0) {
		so_notify_data(so, so->so_rcv_queued);
	} else {
		mutex_exit(&so->so_lock);
	}
}

#ifdef DEBUG
/*
 * Verify that the length stored in so_rcv_queued and the length of data blocks
 * queued is same.
 */
static boolean_t
so_check_length(sonode_t *so)
{
	mblk_t *mp = so->so_rcv_q_head;
	int len = 0;

	ASSERT(MUTEX_HELD(&so->so_lock));

	if (mp != NULL) {
		len = msgdsize(mp);
		while ((mp = mp->b_next) != NULL)
			len += msgdsize(mp);
	}
	mp = so->so_rcv_head;
	if (mp != NULL) {
		len += msgdsize(mp);
		while ((mp = mp->b_next) != NULL)
			len += msgdsize(mp);
	}
	return ((len == so->so_rcv_queued) ? B_TRUE : B_FALSE);
}
#endif

int
so_get_mod_version(struct sockparams *sp)
{
	ASSERT(sp != NULL && sp->sp_smod_info != NULL);
	return (sp->sp_smod_info->smod_version);
}

/*
 * so_start_fallback()
 *
 * Block new socket operations from coming in, and wait for active operations
 * to complete. Threads that are sleeping will be woken up so they can get
 * out of the way.
 *
 * The caller must be a reader on so_fallback_rwlock.
 */
static boolean_t
so_start_fallback(struct sonode *so)
{
	ASSERT(RW_READ_HELD(&so->so_fallback_rwlock));

	mutex_enter(&so->so_lock);
	if (so->so_state & SS_FALLBACK_PENDING) {
		mutex_exit(&so->so_lock);
		return (B_FALSE);
	}
	so->so_state |= SS_FALLBACK_PENDING;
	/*
	 * Poke all threads that might be sleeping. Any operation that comes
	 * in after the cv_broadcast will observe the fallback pending flag
	 * which cause the call to return where it would normally sleep.
	 */
	cv_broadcast(&so->so_state_cv);		/* threads in connect() */
	cv_broadcast(&so->so_rcv_cv);		/* threads in recvmsg() */
	cv_broadcast(&so->so_snd_cv);		/* threads in sendmsg() */
	mutex_enter(&so->so_acceptq_lock);
	cv_broadcast(&so->so_acceptq_cv);	/* threads in accept() */
	mutex_exit(&so->so_acceptq_lock);
	mutex_exit(&so->so_lock);

	/*
	 * The main reason for the rw_tryupgrade call is to provide
	 * observability during the fallback process. We want to
	 * be able to see if there are pending operations.
	 */
	if (rw_tryupgrade(&so->so_fallback_rwlock) == 0) {
		/*
		 * It is safe to drop and reaquire the fallback lock, because
		 * we are guaranteed that another fallback cannot take place.
		 */
		rw_exit(&so->so_fallback_rwlock);
		DTRACE_PROBE1(pending__ops__wait, (struct sonode *), so);
		rw_enter(&so->so_fallback_rwlock, RW_WRITER);
		DTRACE_PROBE1(pending__ops__complete, (struct sonode *), so);
	}

	return (B_TRUE);
}

/*
 * so_end_fallback()
 *
 * Allow socket opertions back in.
 *
 * The caller must be a writer on so_fallback_rwlock.
 */
static void
so_end_fallback(struct sonode *so)
{
	ASSERT(RW_ISWRITER(&so->so_fallback_rwlock));

	mutex_enter(&so->so_lock);
	so->so_state &= ~(SS_FALLBACK_PENDING|SS_FALLBACK_DRAIN);
	mutex_exit(&so->so_lock);

	rw_downgrade(&so->so_fallback_rwlock);
}

/*
 * so_quiesced_cb()
 *
 * Callback passed to the protocol during fallback. It is called once
 * the endpoint is quiescent.
 *
 * No requests from the user, no notifications from the protocol, so it
 * is safe to synchronize the state. Data can also be moved without
 * risk for reordering.
 *
 * We do not need to hold so_lock, since there can be only one thread
 * operating on the sonode.
 */
static mblk_t *
so_quiesced_cb(sock_upper_handle_t sock_handle, sock_quiesce_arg_t *arg,
    struct T_capability_ack *tcap,
    struct sockaddr *laddr, socklen_t laddrlen,
    struct sockaddr *faddr, socklen_t faddrlen, short opts)
{
	struct sonode *so = (struct sonode *)sock_handle;
	boolean_t atmark;
	mblk_t *retmp = NULL, **tailmpp = &retmp;

	if (tcap != NULL)
		sotpi_update_state(so, tcap, laddr, laddrlen, faddr, faddrlen,
		    opts);

	/*
	 * Some protocols do not quiece the data path during fallback. Once
	 * we set the SS_FALLBACK_DRAIN flag any attempt to queue data will
	 * fail and the protocol is responsible for saving the data for later
	 * delivery (i.e., once the fallback has completed).
	 */
	mutex_enter(&so->so_lock);
	so->so_state |= SS_FALLBACK_DRAIN;
	SOCKET_TIMER_CANCEL(so);
	mutex_exit(&so->so_lock);

	if (so->so_rcv_head != NULL) {
		if (so->so_rcv_q_last_head == NULL)
			so->so_rcv_q_head = so->so_rcv_head;
		else
			so->so_rcv_q_last_head->b_next = so->so_rcv_head;
		so->so_rcv_q_last_head = so->so_rcv_last_head;
	}

	atmark = (so->so_state & SS_RCVATMARK) != 0;
	/*
	 * Clear any OOB state having to do with pending data. The TPI
	 * code path will set the appropriate oob state when we move the
	 * oob data to the STREAM head. We leave SS_HADOOBDATA since the oob
	 * data has already been consumed.
	 */
	so->so_state &= ~(SS_RCVATMARK|SS_OOBPEND|SS_HAVEOOBDATA);

	ASSERT(so->so_oobmsg != NULL || so->so_oobmark <= so->so_rcv_queued);

	/*
	 * Move data to the STREAM head.
	 */
	while (so->so_rcv_q_head != NULL) {
		mblk_t *mp = so->so_rcv_q_head;
		size_t mlen = msgdsize(mp);

		so->so_rcv_q_head = mp->b_next;
		mp->b_next = NULL;
		mp->b_prev = NULL;

		/*
		 * Send T_EXDATA_IND if we are at the oob mark.
		 */
		if (atmark) {
			struct T_exdata_ind *tei;
			mblk_t *mp1 = arg->soqa_exdata_mp;

			arg->soqa_exdata_mp = NULL;
			ASSERT(mp1 != NULL);
			mp1->b_datap->db_type = M_PROTO;
			tei = (struct T_exdata_ind *)mp1->b_rptr;
			tei->PRIM_type = T_EXDATA_IND;
			tei->MORE_flag = 0;
			mp1->b_wptr = (uchar_t *)&tei[1];

			if (IS_SO_OOB_INLINE(so)) {
				mp1->b_cont = mp;
			} else {
				ASSERT(so->so_oobmsg != NULL);
				mp1->b_cont = so->so_oobmsg;
				so->so_oobmsg = NULL;

				/* process current mp next time around */
				mp->b_next = so->so_rcv_q_head;
				so->so_rcv_q_head = mp;
				mlen = 0;
			}
			mp = mp1;

			/* we have consumed the oob mark */
			atmark = B_FALSE;
		} else if (so->so_oobmark > 0) {
			/*
			 * Check if the OOB mark is within the current
			 * mblk chain. In that case we have to split it up.
			 */
			if (so->so_oobmark < mlen) {
				mblk_t *urg_mp = mp;

				atmark = B_TRUE;
				mp = NULL;
				mlen = so->so_oobmark;

				/*
				 * It is assumed that the OOB mark does
				 * not land within a mblk.
				 */
				do {
					so->so_oobmark -= MBLKL(urg_mp);
					mp = urg_mp;
					urg_mp = urg_mp->b_cont;
				} while (so->so_oobmark > 0);
				mp->b_cont = NULL;
				if (urg_mp != NULL) {
					urg_mp->b_next = so->so_rcv_q_head;
					so->so_rcv_q_head = urg_mp;
				}
			} else {
				so->so_oobmark -= mlen;
				if (so->so_oobmark == 0)
					atmark = B_TRUE;
			}
		}

		/*
		 * Queue data on the STREAM head.
		 */
		so->so_rcv_queued -= mlen;
		*tailmpp = mp;
		tailmpp = &mp->b_next;
	}
	so->so_rcv_head = NULL;
	so->so_rcv_last_head = NULL;
	so->so_rcv_q_head = NULL;
	so->so_rcv_q_last_head = NULL;

	/*
	 * Check if the oob byte is at the end of the data stream, or if the
	 * oob byte has not yet arrived. In the latter case we have to send a
	 * SIGURG and a mark indicator to the STREAM head. The mark indicator
	 * is needed to guarantee correct behavior for SIOCATMARK. See block
	 * comment in socktpi.h for more details.
	 */
	if (atmark || so->so_oobmark > 0) {
		mblk_t *mp;

		if (atmark && so->so_oobmsg != NULL) {
			struct T_exdata_ind *tei;

			mp = arg->soqa_exdata_mp;
			arg->soqa_exdata_mp = NULL;
			ASSERT(mp != NULL);
			mp->b_datap->db_type = M_PROTO;
			tei = (struct T_exdata_ind *)mp->b_rptr;
			tei->PRIM_type = T_EXDATA_IND;
			tei->MORE_flag = 0;
			mp->b_wptr = (uchar_t *)&tei[1];

			mp->b_cont = so->so_oobmsg;
			so->so_oobmsg = NULL;

			*tailmpp = mp;
			tailmpp = &mp->b_next;
		} else {
			/* Send up the signal */
			mp = arg->soqa_exdata_mp;
			arg->soqa_exdata_mp = NULL;
			ASSERT(mp != NULL);
			DB_TYPE(mp) = M_PCSIG;
			*mp->b_wptr++ = (uchar_t)SIGURG;
			*tailmpp = mp;
			tailmpp = &mp->b_next;

			/* Send up the mark indicator */
			mp = arg->soqa_urgmark_mp;
			arg->soqa_urgmark_mp = NULL;
			mp->b_flag = atmark ? MSGMARKNEXT : MSGNOTMARKNEXT;
			*tailmpp = mp;
			tailmpp = &mp->b_next;

			so->so_oobmark = 0;
		}
	}
	ASSERT(so->so_oobmark == 0);
	ASSERT(so->so_rcv_queued == 0);

	return (retmp);
}

#ifdef DEBUG
/*
 * Do an integrity check of the sonode. This should be done if a
 * fallback fails after sonode has initially been converted to use
 * TPI and subsequently have to be reverted.
 *
 * Failure to pass the integrity check will panic the system.
 */
void
so_integrity_check(struct sonode *cur, struct sonode *orig)
{
	VERIFY(cur->so_vnode == orig->so_vnode);
	VERIFY(cur->so_ops == orig->so_ops);
	/*
	 * For so_state we can only VERIFY the state flags in CHECK_STATE.
	 * The other state flags might be affected by a notification from the
	 * protocol.
	 */
#define	CHECK_STATE	(SS_CANTRCVMORE|SS_CANTSENDMORE|SS_NDELAY|SS_NONBLOCK| \
	SS_ASYNC|SS_ACCEPTCONN|SS_SAVEDEOR|SS_RCVATMARK|SS_OOBPEND| \
	SS_HAVEOOBDATA|SS_HADOOBDATA|SS_SENTLASTREADSIG|SS_SENTLASTWRITESIG)
	VERIFY((cur->so_state & (orig->so_state & CHECK_STATE)) ==
	    (orig->so_state & CHECK_STATE));
	VERIFY(cur->so_mode == orig->so_mode);
	VERIFY(cur->so_flag == orig->so_flag);
	VERIFY(cur->so_count == orig->so_count);
	/* Cannot VERIFY so_proto_connid; proto can update it */
	VERIFY(cur->so_sockparams == orig->so_sockparams);
	/* an error might have been recorded, but it can not be lost */
	VERIFY(cur->so_error != 0 || orig->so_error == 0);
	VERIFY(cur->so_family == orig->so_family);
	VERIFY(cur->so_type == orig->so_type);
	VERIFY(cur->so_protocol == orig->so_protocol);
	VERIFY(cur->so_version == orig->so_version);
	/* New conns might have arrived, but none should have been lost */
	VERIFY(cur->so_acceptq_len >= orig->so_acceptq_len);
	VERIFY(list_head(&cur->so_acceptq_list) ==
	    list_head(&orig->so_acceptq_list));
	VERIFY(cur->so_backlog == orig->so_backlog);
	/* New OOB migth have arrived, but mark should not have been lost */
	VERIFY(cur->so_oobmark >= orig->so_oobmark);
	/* Cannot VERIFY so_oobmsg; the proto might have sent up a new one */
	VERIFY(cur->so_pgrp == orig->so_pgrp);
	VERIFY(cur->so_peercred == orig->so_peercred);
	VERIFY(cur->so_cpid == orig->so_cpid);
	VERIFY(cur->so_zoneid == orig->so_zoneid);
	/* New data migth have arrived, but none should have been lost */
	VERIFY(cur->so_rcv_queued >= orig->so_rcv_queued);
	VERIFY(cur->so_rcv_q_head == orig->so_rcv_q_head);
	VERIFY(cur->so_rcv_head == orig->so_rcv_head);
	VERIFY(cur->so_proto_handle == orig->so_proto_handle);
	VERIFY(cur->so_downcalls == orig->so_downcalls);
	/* Cannot VERIFY so_proto_props; they can be updated by proto */
}
#endif

/*
 * so_tpi_fallback()
 *
 * This is the fallback initation routine; things start here.
 *
 * Basic strategy:
 *   o Block new socket operations from coming in
 *   o Allocate/initate info needed by TPI
 *   o Quiesce the connection, at which point we sync
 *     state and move data
 *   o Change operations (sonodeops) associated with the socket
 *   o Unblock threads waiting for the fallback to finish
 */
int
so_tpi_fallback(struct sonode *so, struct cred *cr)
{
	int error;
	queue_t *q;
	struct sockparams *sp;
	struct sockparams *newsp = NULL;
	so_proto_fallback_func_t fbfunc;
	const char *devpath;
	boolean_t direct;
	struct sonode *nso;
	sock_quiesce_arg_t arg = { NULL, NULL };
#ifdef DEBUG
	struct sonode origso;
#endif
	error = 0;
	sp = so->so_sockparams;
	fbfunc = sp->sp_smod_info->smod_proto_fallback_func;

	/*
	 * Cannot fallback if the socket has active filters or a krecv callback.
	 */
	if (so->so_filter_active > 0 || so->so_krecv_cb != NULL)
		return (EINVAL);

	switch (so->so_family) {
	case AF_INET:
		devpath = sp->sp_smod_info->smod_fallback_devpath_v4;
		break;
	case AF_INET6:
		devpath = sp->sp_smod_info->smod_fallback_devpath_v6;
		break;
	default:
		return (EINVAL);
	}

	/*
	 * Fallback can only happen if the socket module has a TPI device
	 * and fallback function.
	 */
	if (devpath == NULL || fbfunc == NULL)
		return (EINVAL);

	/*
	 * Initiate fallback; upon success we know that no new requests
	 * will come in from the user.
	 */
	if (!so_start_fallback(so))
		return (EAGAIN);
#ifdef DEBUG
	/*
	 * Make a copy of the sonode in case we need to make an integrity
	 * check later on.
	 */
	bcopy(so, &origso, sizeof (*so));
#endif

	sp->sp_stats.sps_nfallback.value.ui64++;

	newsp = sockparams_hold_ephemeral_bydev(so->so_family, so->so_type,
	    so->so_protocol, devpath, KM_SLEEP, &error);
	if (error != 0)
		goto out;

	if (so->so_direct != NULL) {
		sodirect_t *sodp = so->so_direct;
		mutex_enter(&so->so_lock);

		so->so_direct->sod_enabled = B_FALSE;
		so->so_state &= ~SS_SODIRECT;
		ASSERT(sodp->sod_uioafh == NULL);
		mutex_exit(&so->so_lock);
	}

	/* Turn sonode into a TPI socket */
	error = sotpi_convert_sonode(so, newsp, &direct, &q, cr);
	if (error != 0)
		goto out;
	/*
	 * When it comes to urgent data we have two cases to deal with;
	 * (1) The oob byte has already arrived, or (2) the protocol has
	 * notified that oob data is pending, but it has not yet arrived.
	 *
	 * For (1) all we need to do is send a T_EXDATA_IND to indicate were
	 * in the byte stream the oob byte is. For (2) we have to send a
	 * SIGURG (M_PCSIG), followed by a zero-length mblk indicating whether
	 * the oob byte will be the next byte from the protocol.
	 *
	 * So in the worst case we need two mblks, one for the signal, another
	 * for mark indication. In that case we use the exdata_mp for the sig.
	 */
	arg.soqa_exdata_mp = allocb_wait(sizeof (struct T_exdata_ind),
	    BPRI_MED, STR_NOSIG, NULL);
	arg.soqa_urgmark_mp = allocb_wait(0, BPRI_MED, STR_NOSIG, NULL);

	/*
	 * Now tell the protocol to start using TPI. so_quiesced_cb be
	 * called once it's safe to synchronize state.
	 */
	DTRACE_PROBE1(proto__fallback__begin, struct sonode *, so);
	error = (*fbfunc)(so->so_proto_handle, q, direct, so_quiesced_cb,
	    &arg);
	DTRACE_PROBE1(proto__fallback__end, struct sonode *, so);

	if (error != 0) {
		/* protocol was unable to do a fallback, revert the sonode */
		sotpi_revert_sonode(so, cr);
		goto out;
	}

	/*
	 * Walk the accept queue and notify the proto that they should
	 * fall back to TPI. The protocol will send up the T_CONN_IND.
	 */
	nso = list_head(&so->so_acceptq_list);
	while (nso != NULL) {
		int rval;
		struct sonode *next;

		if (arg.soqa_exdata_mp == NULL) {
			arg.soqa_exdata_mp =
			    allocb_wait(sizeof (struct T_exdata_ind),
			    BPRI_MED, STR_NOSIG, NULL);
		}
		if (arg.soqa_urgmark_mp == NULL) {
			arg.soqa_urgmark_mp = allocb_wait(0, BPRI_MED,
			    STR_NOSIG, NULL);
		}

		DTRACE_PROBE1(proto__fallback__begin, struct sonode *, nso);
		rval = (*fbfunc)(nso->so_proto_handle, NULL, direct,
		    so_quiesced_cb, &arg);
		DTRACE_PROBE1(proto__fallback__end, struct sonode *, nso);
		if (rval != 0) {
			/* Abort the connection */
			zcmn_err(getzoneid(), CE_WARN,
			    "Failed to convert socket in accept queue to TPI. "
			    "Pid = %d\n", curproc->p_pid);
			next = list_next(&so->so_acceptq_list, nso);
			list_remove(&so->so_acceptq_list, nso);
			so->so_acceptq_len--;

			(void) socket_close(nso, 0, CRED());
			socket_destroy(nso);
			nso = next;
		} else {
			nso = list_next(&so->so_acceptq_list, nso);
		}
	}

	/*
	 * Now flush the acceptq, this will destroy all sockets. They will
	 * be recreated in sotpi_accept().
	 */
	so_acceptq_flush(so, B_FALSE);

	mutex_enter(&so->so_lock);
	so->so_state |= SS_FALLBACK_COMP;
	mutex_exit(&so->so_lock);

	/*
	 * Swap the sonode ops. Socket opertations that come in once this
	 * is done will proceed without blocking.
	 */
	so->so_ops = &sotpi_sonodeops;

	/*
	 * Wake up any threads stuck in poll. This is needed since the poll
	 * head changes when the fallback happens (moves from the sonode to
	 * the STREAMS head).
	 */
	pollwakeup(&so->so_poll_list, POLLERR);

	/*
	 * When this non-STREAM socket was created we placed an extra ref on
	 * the associated vnode to support asynchronous close. Drop that ref
	 * here.
	 */
	ASSERT(SOTOV(so)->v_count >= 2);
	VN_RELE(SOTOV(so));
out:
	so_end_fallback(so);

	if (error != 0) {
#ifdef DEBUG
		so_integrity_check(so, &origso);
#endif
		zcmn_err(getzoneid(), CE_WARN,
		    "Failed to convert socket to TPI (err=%d). Pid = %d\n",
		    error, curproc->p_pid);
		if (newsp != NULL)
			SOCKPARAMS_DEC_REF(newsp);
	}
	if (arg.soqa_exdata_mp != NULL)
		freemsg(arg.soqa_exdata_mp);
	if (arg.soqa_urgmark_mp != NULL)
		freemsg(arg.soqa_urgmark_mp);

	return (error);
}

int
so_krecv_set(sonode_t *so, so_krecv_f cb, void *arg)
{
	int ret;

	if (cb == NULL && arg != NULL)
		return (EINVAL);

	SO_BLOCK_FALLBACK(so, so_krecv_set(so, cb, arg));

	mutex_enter(&so->so_lock);
	if (so->so_state & SS_FALLBACK_COMP) {
		mutex_exit(&so->so_lock);
		SO_UNBLOCK_FALLBACK(so);
		return (ENOTSUP);
	}

	ret = so_lock_read(so, 0);
	VERIFY(ret == 0);
	/*
	 * Other consumers may actually care about getting extant data delivered
	 * to them, when they come along, they should figure out the best API
	 * for that.
	 */
	so_rcv_flush(so);

	so->so_krecv_cb = cb;
	so->so_krecv_arg = arg;

	so_unlock_read(so);
	mutex_exit(&so->so_lock);
	SO_UNBLOCK_FALLBACK(so);

	return (0);
}

void
so_krecv_unblock(sonode_t *so)
{
	mutex_enter(&so->so_lock);
	VERIFY(so->so_krecv_cb != NULL);

	so->so_rcv_queued = 0;
	(void) so_check_flow_control(so);
	/*
	 * so_check_flow_control() always drops so->so_lock, so we won't
	 * need to drop it ourselves.
	 */
}