/*

 * 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) 1984, 1986, 1987, 1988, 1989 AT&T	*/

/*	  All Rights Reserved  	*/

/*

 * Copyright 2007 Sun Microsystems, Inc.  All rights reserved.

 * Use is subject to license terms.

 */

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

#include <sys/types.h>

#include <sys/sysmacros.h>

#include <sys/param.h>

#include <sys/errno.h>

#include <sys/signal.h>

#include <sys/proc.h>

#include <sys/conf.h>

#include <sys/cred.h>

#include <sys/user.h>

#include <sys/vnode.h>

#include <sys/file.h>

#include <sys/session.h>

#include <sys/stream.h>

#include <sys/strsubr.h>

#include <sys/stropts.h>

#include <sys/poll.h>

#include <sys/systm.h>

#include <sys/cpuvar.h>

#include <sys/uio.h>

#include <sys/cmn_err.h>

#include <sys/priocntl.h>

#include <sys/procset.h>

#include <sys/vmem.h>

#include <sys/bitmap.h>

#include <sys/kmem.h>

#include <sys/siginfo.h>

#include <sys/vtrace.h>

#include <sys/callb.h>

#include <sys/debug.h>

#include <sys/modctl.h>

#include <sys/vmsystm.h>

#include <vm/page.h>

#include <sys/atomic.h>

#include <sys/suntpi.h>

#include <sys/strlog.h>

#include <sys/promif.h>

#include <sys/project.h>

#include <sys/vm.h>

#include <sys/taskq.h>

#include <sys/sunddi.h>

#include <sys/sunldi_impl.h>

#include <sys/strsun.h>

#include <sys/isa_defs.h>

#include <sys/multidata.h>

#include <sys/pattr.h>

#include <sys/strft.h>

#include <sys/fs/snode.h>

#include <sys/zone.h>

#include <sys/open.h>

#include <sys/sunldi.h>

#include <sys/sad.h>

#include <sys/netstack.h>

#define	O_SAMESTR(q)	(((q)->q_next) && \

	(((q)->q_flag & QREADR) == ((q)->q_next->q_flag & QREADR)))

/*

 * WARNING:

 * The variables and routines in this file are private, belonging

 * to the STREAMS subsystem. These should not be used by modules

 * or drivers. Compatibility will not be guaranteed.

 */

/*

 * Id value used to distinguish between different multiplexor links.

 */

static int32_t lnk_id = 0;

#define	STREAMS_LOPRI MINCLSYSPRI

static pri_t streams_lopri = STREAMS_LOPRI;

#define	STRSTAT(x)	(str_statistics.x.value.ui64++)

typedef struct str_stat {

	kstat_named_t	sqenables;

	kstat_named_t	stenables;

	kstat_named_t	syncqservice;

	kstat_named_t	freebs;

	kstat_named_t	qwr_outer;

	kstat_named_t	rservice;

	kstat_named_t	strwaits;

	kstat_named_t	taskqfails;

	kstat_named_t	bufcalls;

	kstat_named_t	qhelps;

	kstat_named_t	qremoved;

	kstat_named_t	sqremoved;

	kstat_named_t	bcwaits;

	kstat_named_t	sqtoomany;

} str_stat_t;

static str_stat_t str_statistics = {

	{ "sqenables",		KSTAT_DATA_UINT64 },

	{ "stenables",		KSTAT_DATA_UINT64 },

	{ "syncqservice",	KSTAT_DATA_UINT64 },

	{ "freebs",		KSTAT_DATA_UINT64 },

	{ "qwr_outer",		KSTAT_DATA_UINT64 },

	{ "rservice",		KSTAT_DATA_UINT64 },

	{ "strwaits",		KSTAT_DATA_UINT64 },

	{ "taskqfails",		KSTAT_DATA_UINT64 },

	{ "bufcalls",		KSTAT_DATA_UINT64 },

	{ "qhelps",		KSTAT_DATA_UINT64 },

	{ "qremoved",		KSTAT_DATA_UINT64 },

	{ "sqremoved",		KSTAT_DATA_UINT64 },

	{ "bcwaits",		KSTAT_DATA_UINT64 },

	{ "sqtoomany",		KSTAT_DATA_UINT64 },

};

static kstat_t *str_kstat;

/*

 * qrunflag was used previously to control background scheduling of queues. It

 * is not used anymore, but kept here in case some module still wants to access

 * it via qready() and setqsched macros.

 */

char qrunflag;			/*  Unused */

/*

 * Most of the streams scheduling is done via task queues. Task queues may fail

 * for non-sleep dispatches, so there are two backup threads servicing failed

 * requests for queues and syncqs. Both of these threads also service failed

 * dispatches freebs requests. Queues are put in the list specified by `qhead'

 * and `qtail' pointers, syncqs use `sqhead' and `sqtail' pointers and freebs

 * requests are put into `freebs_list' which has no tail pointer. All three

 * lists are protected by a single `service_queue' lock and use

 * `services_to_run' condition variable for signaling background threads. Use of

 * a single lock should not be a problem because it is only used under heavy

 * loads when task queues start to fail and at that time it may be a good idea

 * to throttle scheduling requests.

 *

 * NOTE: queues and syncqs should be scheduled by two separate threads because

 * queue servicing may be blocked waiting for a syncq which may be also

 * scheduled for background execution. This may create a deadlock when only one

 * thread is used for both.

 */

static taskq_t *streams_taskq;		/* Used for most STREAMS scheduling */

static kmutex_t service_queue;		/* protects all of servicing vars */

static kcondvar_t services_to_run;	/* wake up background service thread */

static kcondvar_t syncqs_to_run;	/* wake up background service thread */

/*

 * List of queues scheduled for background processing dueue to lack of resources

 * in the task queues. Protected by service_queue lock;

 */

static struct queue *qhead;

static struct queue *qtail;

/*

 * Same list for syncqs

 */

static syncq_t *sqhead;

static syncq_t *sqtail;

static mblk_t *freebs_list;	/* list of buffers to free */

/*

 * Backup threads for servicing queues and syncqs

 */

kthread_t *streams_qbkgrnd_thread;

kthread_t *streams_sqbkgrnd_thread;

/*

 * Bufcalls related variables.

 */

struct bclist	strbcalls;	/* list of waiting bufcalls */

kmutex_t	strbcall_lock;	/* protects bufcall list (strbcalls) */

kcondvar_t	strbcall_cv;	/* Signaling when a bufcall is added */

kmutex_t	bcall_monitor;	/* sleep/wakeup style monitor */

kcondvar_t	bcall_cv;	/* wait 'till executing bufcall completes */

kthread_t	*bc_bkgrnd_thread; /* Thread to service bufcall requests */

kmutex_t	strresources;	/* protects global resources */

kmutex_t	muxifier;	/* single-threads multiplexor creation */

static void	*str_stack_init(netstackid_t stackid, netstack_t *ns);

static void	str_stack_shutdown(netstackid_t stackid, void *arg);

static void	str_stack_fini(netstackid_t stackid, void *arg);

extern void	time_to_wait(clock_t *, clock_t);

/*

 * run_queues is no longer used, but is kept in case some 3-d party

 * module/driver decides to use it.

 */

int run_queues = 0;

/*

 * sq_max_size is the depth of the syncq (in number of messages) before

 * qfill_syncq() starts QFULL'ing destination queues. As its primary

 * consumer - IP is no longer D_MTPERMOD, but there may be other

 * modules/drivers depend on this syncq flow control, we prefer to

 * choose a large number as the default value. For potential

 * performance gain, this value is tunable in /etc/system.

 */

int sq_max_size = 10000;

/*

 * the number of ciputctrl structures per syncq and stream we create when

 * needed.

 */

int n_ciputctrl;

int max_n_ciputctrl = 16;

/*

 * if n_ciputctrl is < min_n_ciputctrl don't even create ciputctrl_cache.

 */

int min_n_ciputctrl = 2;

/*

 * Per-driver/module syncqs

 * ========================

 *

 * For drivers/modules that use PERMOD or outer syncqs we keep a list of

 * perdm structures, new entries being added (and new syncqs allocated) when

 * setq() encounters a module/driver with a streamtab that it hasn't seen

 * before.

 * The reason for this mechanism is that some modules and drivers share a

 * common streamtab and it is necessary for those modules and drivers to also

 * share a common PERMOD syncq.

 *

 * perdm_list --> dm_str == streamtab_1

 *                dm_sq == syncq_1

 *                dm_ref

 *                dm_next --> dm_str == streamtab_2

 *                            dm_sq == syncq_2

 *                            dm_ref

 *                            dm_next --> ... NULL

 *

 * The dm_ref field is incremented for each new driver/module that takes

 * a reference to the perdm structure and hence shares the syncq.

 * References are held in the fmodsw_impl_t structure for each STREAMS module

 * or the dev_impl array (indexed by device major number) for each driver.

 *

 * perdm_list -> [dm_ref == 1] -> [dm_ref == 2] -> [dm_ref == 1] -> NULL

 *		     ^                 ^ ^               ^

 *                   |  ______________/  |               |

 *                   | /                 |               |

 * dev_impl:     ...|x|y|...          module A	      module B

 *

 * When a module/driver is unloaded the reference count is decremented and,

 * when it falls to zero, the perdm structure is removed from the list and

 * the syncq is freed (see rele_dm()).

 */

perdm_t *perdm_list = NULL;

static krwlock_t perdm_rwlock;

cdevsw_impl_t *devimpl;

extern struct qinit strdata;

extern struct qinit stwdata;

static void runservice(queue_t *);

static void streams_bufcall_service(void);

static void streams_qbkgrnd_service(void);

static void streams_sqbkgrnd_service(void);

static syncq_t *new_syncq(void);

static void free_syncq(syncq_t *);

static void outer_insert(syncq_t *, syncq_t *);

static void outer_remove(syncq_t *, syncq_t *);

static void write_now(syncq_t *);

static void clr_qfull(queue_t *);

static void enable_svc(queue_t *);

static void runbufcalls(void);

static void sqenable(syncq_t *);

static void sqfill_events(syncq_t *, queue_t *, mblk_t *, void (*)());

static void wait_q_syncq(queue_t *);

static void backenable_insertedq(queue_t *);

static void queue_service(queue_t *);

static void stream_service(stdata_t *);

static void syncq_service(syncq_t *);

static void qwriter_outer_service(syncq_t *);

static void mblk_free(mblk_t *);

#ifdef DEBUG

static int qprocsareon(queue_t *);

#endif

static void set_nfsrv_ptr(queue_t *, queue_t *, queue_t *, queue_t *);

static void reset_nfsrv_ptr(queue_t *, queue_t *);

static void sq_run_events(syncq_t *);

static int propagate_syncq(queue_t *);

static void	blocksq(syncq_t *, ushort_t, int);

static void	unblocksq(syncq_t *, ushort_t, int);

static int	dropsq(syncq_t *, uint16_t);

static void	emptysq(syncq_t *);

static sqlist_t *sqlist_alloc(struct stdata *, int);

static void	sqlist_free(sqlist_t *);

static sqlist_t	*sqlist_build(queue_t *, struct stdata *, boolean_t);

static void	sqlist_insert(sqlist_t *, syncq_t *);

static void	sqlist_insertall(sqlist_t *, queue_t *);

static void	strsetuio(stdata_t *);

struct kmem_cache *stream_head_cache;

struct kmem_cache *queue_cache;

struct kmem_cache *syncq_cache;

struct kmem_cache *qband_cache;

struct kmem_cache *linkinfo_cache;

struct kmem_cache *ciputctrl_cache = NULL;

static linkinfo_t *linkinfo_list;

/* global esballoc throttling queue */

static esb_queue_t	system_esbq;

/*

 * esballoc tunable parameters.

 */

int		esbq_max_qlen = 0x16;	/* throttled queue length */

clock_t		esbq_timeout = 0x8;	/* timeout to process esb queue */

/*

 * routines to handle esballoc queuing.

 */

static void esballoc_process_queue(esb_queue_t *);

static void esballoc_enqueue_mblk(mblk_t *);

static void esballoc_timer(void *);

static void esballoc_set_timer(esb_queue_t *, clock_t);

static void esballoc_mblk_free(mblk_t *);

/*

 *  Qinit structure and Module_info structures

 *	for passthru read and write queues

 */

static void pass_wput(queue_t *, mblk_t *);

static queue_t *link_addpassthru(stdata_t *);

static void link_rempassthru(queue_t *);

struct  module_info passthru_info = {

	0,

	"passthru",

	0,

	INFPSZ,

	STRHIGH,

	STRLOW

};

struct  qinit passthru_rinit = {

	(int (*)())putnext,

	NULL,

	NULL,

	NULL,

	NULL,

	&passthru_info,

	NULL

};

struct  qinit passthru_winit = {

	(int (*)()) pass_wput,

	NULL,

	NULL,

	NULL,

	NULL,

	&passthru_info,

	NULL

};

/*

 * Special form of assertion: verify that X implies Y i.e. when X is true Y

 * should also be true.

 */

#define	IMPLY(X, Y)	ASSERT(!(X) || (Y))

/*

 * Logical equivalence. Verify that both X and Y are either TRUE or FALSE.

 */

#define	EQUIV(X, Y)	{ IMPLY(X, Y); IMPLY(Y, X); }

/*

 * Verify correctness of list head/tail pointers.

 */

#define	LISTCHECK(head, tail, link) {				\

	EQUIV(head, tail);					\

	IMPLY(tail != NULL, tail->link == NULL);		\

}

/*

 * Enqueue a list element `el' in the end of a list denoted by `head' and `tail'

 * using a `link' field.

 */

#define	ENQUEUE(el, head, tail, link) {				\

	ASSERT(el->link == NULL);				\

	LISTCHECK(head, tail, link);				\

	if (head == NULL)					\

		head = el;					\

	else							\

		tail->link = el;				\

	tail = el;						\

}

/*

 * Dequeue the first element of the list denoted by `head' and `tail' pointers

 * using a `link' field and put result into `el'.

 */

#define	DQ(el, head, tail, link) {				\

	LISTCHECK(head, tail, link);				\

	el = head;						\

	if (head != NULL) {					\

		head = head->link;				\

		if (head == NULL)				\

			tail = NULL;				\

		el->link = NULL;				\

	}							\

}

/*

 * Remove `el' from the list using `chase' and `curr' pointers and return result

 * in `succeed'.

 */

#define	RMQ(el, head, tail, link, chase, curr, succeed) {	\

	LISTCHECK(head, tail, link);				\

	chase = NULL;						\

	succeed = 0;						\

	for (curr = head; (curr != el) && (curr != NULL); curr = curr->link) \

		chase = curr;					\

	if (curr != NULL) {					\

		succeed = 1;					\

		ASSERT(curr == el);				\

		if (chase != NULL)				\

			chase->link = curr->link;		\

		else						\

			head = curr->link;			\

		curr->link = NULL;				\

		if (curr == tail)				\

			tail = chase;				\

	}							\

	LISTCHECK(head, tail, link);				\

}

/* Handling of delayed messages on the inner syncq. */

/*

 * DEBUG versions should use function versions (to simplify tracing) and

 * non-DEBUG kernels should use macro versions.

 */

/*

 * Put a queue on the syncq list of queues.

 * Assumes SQLOCK held.

 */

#define	SQPUT_Q(sq, qp)							\

{									\

	ASSERT(MUTEX_HELD(SQLOCK(sq)));					\

	if (!(qp->q_sqflags & Q_SQQUEUED)) {				\

		/* The queue should not be linked anywhere */		\

		ASSERT((qp->q_sqprev == NULL) && (qp->q_sqnext == NULL)); \

		/* Head and tail may only be NULL simultaneously */	\

		EQUIV(sq->sq_head, sq->sq_tail);			\

		/* Queue may be only enqueyed on its syncq */		\

		ASSERT(sq == qp->q_syncq);				\

		/* Check the correctness of SQ_MESSAGES flag */		\

		EQUIV(sq->sq_head, (sq->sq_flags & SQ_MESSAGES));	\

		/* Sanity check first/last elements of the list */	\

		IMPLY(sq->sq_head != NULL, sq->sq_head->q_sqprev == NULL);\

		IMPLY(sq->sq_tail != NULL, sq->sq_tail->q_sqnext == NULL);\

		/*							\

		 * Sanity check of priority field: empty queue should	\

		 * have zero priority					\

		 * and nqueues equal to zero.				\

		 */							\

		IMPLY(sq->sq_head == NULL, sq->sq_pri == 0);		\

		/* Sanity check of sq_nqueues field */			\

		EQUIV(sq->sq_head, sq->sq_nqueues);			\

		if (sq->sq_head == NULL) {				\

			sq->sq_head = sq->sq_tail = qp;			\

			sq->sq_flags |= SQ_MESSAGES;			\

		} else if (qp->q_spri == 0) {				\

			qp->q_sqprev = sq->sq_tail;			\

			sq->sq_tail->q_sqnext = qp;			\

			sq->sq_tail = qp;				\

		} else {						\

			/*						\

			 * Put this queue in priority order: higher	\

			 * priority gets closer to the head.		\

			 */						\

			queue_t **qpp = &sq->sq_tail;			\

			queue_t *qnext = NULL;				\

									\

			while (*qpp != NULL && qp->q_spri > (*qpp)->q_spri) { \

				qnext = *qpp;				\

				qpp = &(*qpp)->q_sqprev;		\

			}						\

			qp->q_sqnext = qnext;				\

			qp->q_sqprev = *qpp;				\

			if (*qpp != NULL) {				\

				(*qpp)->q_sqnext = qp;			\

			} else {					\

				sq->sq_head = qp;			\

				sq->sq_pri = sq->sq_head->q_spri;	\

			}						\

			*qpp = qp;					\

		}							\

		qp->q_sqflags |= Q_SQQUEUED;				\