common/os/serializer.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */
/*
 * Copyright 2004 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

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

/*
 * Kernel protection serializers: general purpose synchronization mechanism.
 *
 * Serializers provide a simple way to serialize access to some resource. They
 * can be used as an alternative to locks or STREAMS perimeters. They scale
 * much better than STREAMS outer serializers.
 *
 * Serializer is an abstraction that guarantees that all functions executed
 * within the serializer are serialized: they are executed in the order they
 * entered serializer one at a time.
 *
 * INTERFACES:
 *
 * serializer_t *serializer_create(flags);
 *
 *	Create a serializer. The flags may be either SER_SLEEP or SER_NOSLEEP
 *	which are the same as KM_SLEEP and KM_NOSLEEP respectively.
 *
 * serializer_enter(serializer, proc, mblk, arg);
 *
 *	Execute 'proc(mblk, arg)' within the serializer.
 *
 * serializer_wait(serializer);
 *
 *	Wait for pending serializer jobs to complete. This function should never
 *	be called within the serializer or it will deadlock.
 *
 * serializer_destroy(serializer);
 *
 *	Destroy serializer.
 *
 * Serializers export three DTrace SDT probes:
 *
 *	serializer-enqueue(serializer, mblk, arg, proc)
 *
 *		The probe triggers when serializer is busy and the request is
 *		queued.
 *
 *	serializer-exec-start(serializer, mblk, arg, proc)
 *
 *		The probe triggers before the request is executed
 *
 *	serializer-exec-end(serializer, mblk, arg, proc)
 *
 *		The probe triggers after the request is executed
 *
 *
 * IMPLEMENTATION.
 *
 * Serializer consists of a "owner" and a list of queued jobs. The first thread
 * entering serializer sets the owner and executes its job directly without
 * context switch. Then it processes jobs which may have been enqueued while it
 * was executing a job and drops the owner, leaving the serializer empty.  Any
 * thread entering an owned serializer enqueues its job and returns immediately.
 *
 * Serializer data structure holds several fields used for debugging only. They
 * are not relevant for the proper serializer functioning.
 *
 * When new requests arrive faster then they are processed it is possible that a
 * thread that started processing serializer will continue doing so for a long
 * time. To avoid such pathological behavior the amount of requests drained by
 * serializer_enter() is limited by `serializer_credit' value. After the credit
 * is expired serializer_enter() schedules a taskq request to continue draining.
 * The taskq thread draining is not limited by serializer_credit. Note that it
 * is possible that another serializer_enter() will drain the serializer before
 * a taskq thread will get to it.
 */

#include <sys/types.h>
#include <sys/kmem.h>
#include <sys/thread.h>
#include <sys/mutex.h>
#include <sys/systm.h>
#include <sys/debug.h>
#include <sys/taskq.h>
#include <sys/sdt.h>
#include <sys/serializer.h>

#define	SERIALIZER_NAMELEN 31

/*
 * Serializer abstraction.
 * Fields marked (D) are used for debugging purposes only.
 */
struct serializer_s {
	kmutex_t	ser_lock;	/* Protects state and the list */
	kthread_t	*ser_owner;	/* Thread executing serializer */
	ushort_t	ser_taskq;	/* Serializer scheduled for taskq */
	kcondvar_t	ser_cv;		/* For serializer-wait */
	uint_t		ser_count;	/* # of queued requests (D) */
	mblk_t		*ser_first;	/* First message in the queue */
	mblk_t		*ser_last;	/* Last message in the queue */
	srproc_t	*ser_proc;	/* Currently executing proc (D) */
	mblk_t		*ser_curr;	/* Currently executing msg (D) */
	void		*ser_arg;	/* Currently executing arg (D) */
};

static kmem_cache_t *serializer_cache;

/*
 * How many drains are allowed before we switch to taskq processing.
 */
#define	SERIALIZER_CREDIT 10
static int serializer_credit = SERIALIZER_CREDIT;

/* Statistics for debugging */
static int perim_context_swtch = 0;

static int serializer_constructor(void *, void *, int);
static void serializer_destructor(void *, void *);
static void serializer_exec(serializer_t *, srproc_t, mblk_t *, void *);
static void serializer_enqueue(serializer_t *, srproc_t, mblk_t *, void *);
static void serializer_drain(serializer_t *, int);
static void serializer_drain_completely(serializer_t *);

/*
 * SERIALIZER Implementation.
 */

/*
 * Record debugging information and execute single request.
 */
static void
serializer_exec(serializer_t *s, srproc_t proc, mblk_t *mp, void *arg)
{
	ASSERT(MUTEX_NOT_HELD(&s->ser_lock));
	ASSERT(s->ser_owner == curthread);

	ASSERT(proc != NULL);
	ASSERT(mp != NULL);

	s->ser_curr = mp;
	s->ser_arg = arg;
	s->ser_proc = proc;
	proc(mp, arg);
}

/*
 * Enqueue a single request on serializer.
 */
static void
serializer_enqueue(serializer_t *s, srproc_t proc, mblk_t *mp, void *arg)
{
	ASSERT(MUTEX_HELD(&s->ser_lock));

	DTRACE_PROBE4(serializer__enqueue, serializer_t *, s,
	    mblk_t *, mp, void *, arg, srproc_t, proc);
	s->ser_count++;
	mp->b_queue = (queue_t *)proc;
	mp->b_prev = (mblk_t *)arg;
	if (s->ser_last != NULL)
		s->ser_last->b_next = mp;
	else
		s->ser_first = mp;
	s->ser_last = mp;
}

/*
 * Drain serializer, limiting drain to `credit' requests at most.
 */
static void
serializer_drain(serializer_t *s, int credit)
{
	mblk_t *mp = s->ser_first;

	ASSERT(MUTEX_HELD(&s->ser_lock));
	ASSERT(s->ser_owner == curthread);

	for (; mp != NULL && credit-- != 0; mp = s->ser_first) {
		srproc_t *proc = (srproc_t *)mp->b_queue;
		void *arg = mp->b_prev;

		if ((s->ser_first = s->ser_first->b_next) == NULL) {
			s->ser_last = NULL;
		} else {
			mp->b_next = NULL;
		}
		ASSERT(s->ser_count != 0);
		s->ser_count--;
		mp->b_queue = NULL;
		mp->b_prev = NULL;
		mutex_exit(&s->ser_lock);

		DTRACE_PROBE4(serializer__exec__start, serializer_t *, s,
		    mblk_t *, mp, void *, arg, srproc_t, proc);
		serializer_exec(s, proc, mp, arg);
		DTRACE_PROBE4(serializer__exec__end, serializer_t *, s,
		    mblk_t *, mp, void *, arg, srproc_t, proc);

		mutex_enter(&s->ser_lock);
	}
}

/*
 * Drain serializer completely if serializer is free.
 */
static void
serializer_drain_completely(serializer_t *s)
{
	mutex_enter(&s->ser_lock);
	ASSERT(s->ser_taskq);
	if (s->ser_owner == NULL) {
		s->ser_owner = curthread;
		while (s->ser_first != NULL)
			serializer_drain(s, INT_MAX);
		s->ser_owner = NULL;
		s->ser_curr = NULL;
		s->ser_proc = NULL;
		s->ser_arg = NULL;
	}
	s->ser_taskq = B_FALSE;
	/*
	 * Wake up serializer_wait().
	 */
	cv_signal(&s->ser_cv);
	mutex_exit(&s->ser_lock);
}

/*
 * Call proc(mp, arg) within serializer.
 *
 * If serializer is empty and not owned, proc(mp, arg) is called right
 * away. Otherwise the request is queued.
 */
void
serializer_enter(serializer_t *s, srproc_t proc, mblk_t *mp, void *arg)
{
	ASSERT(proc != NULL);
	ASSERT(mp != NULL);
	ASSERT(mp->b_next == NULL);
	ASSERT(mp->b_prev == NULL);

	ASSERT(MUTEX_NOT_HELD(&s->ser_lock));

	mutex_enter(&s->ser_lock);
	if (s->ser_owner != NULL) {
		/*
		 * Serializer is owned. Enqueue and return.
		 */
		serializer_enqueue(s, proc, mp, arg);
	} else {
		taskqid_t tid = TASKQID_INVALID;

		/*
		 * If the request list is empty, can process right away,
		 * otherwise enqueue and process.
		 */
		s->ser_owner = curthread;

		if (s->ser_first != NULL) {
			ASSERT(s->ser_count != 0);
			serializer_enqueue(s, proc, mp, arg);
		} else {
			ASSERT(s->ser_count == 0);
			mutex_exit(&s->ser_lock);
			/*
			 * Execute request
			 */
			DTRACE_PROBE4(serializer__exec__start,
			    serializer_t *, s, mblk_t *, mp,
			    void *, arg, srproc_t, proc);
			serializer_exec(s, proc, mp, arg);
			DTRACE_PROBE4(serializer__exec__end,
			    serializer_t *, s, mblk_t *, mp,
			    void *, arg, srproc_t, proc);
			mutex_enter(&s->ser_lock);
		}

		/*
		 * Drain whatever has arrived in the meantime.
		 * If we spend too much time draining, continue draining by the
		 * taskq thread.
		 */
		while ((s->ser_first != NULL) && (tid == 0)) {
			serializer_drain(s, serializer_credit);
			if (s->ser_first != NULL) {
				perim_context_swtch++;
				/*
				 * If there is a taskq pending for this
				 * serializer, no need to schedule a new one.
				 */
				if (s->ser_taskq) {
					break;
				} else {
					tid = taskq_dispatch(system_taskq,
					    (task_func_t *)
					    serializer_drain_completely,
					    s, TQ_NOSLEEP | TQ_NOQUEUE);
					if (tid != TASKQID_INVALID)
						s->ser_taskq = B_TRUE;
				}
			}
		}
		s->ser_owner = NULL;
		s->ser_curr = NULL;
		s->ser_proc = NULL;
		s->ser_arg = NULL;
	}
	/*
	 * Wakeup serializer_wait().
	 */
	cv_signal(&s->ser_cv);
	mutex_exit(&s->ser_lock);
}

/*
 * Wait for pending serializer jobs to complete. This function should never be
 * called within the serializer or it will deadlock.
 */
void
serializer_wait(serializer_t *s)
{
	mutex_enter(&s->ser_lock);

	ASSERT(s->ser_owner != curthread);

	while ((s->ser_owner != NULL) || s->ser_taskq || (s->ser_first != NULL))
		cv_wait(&s->ser_cv, &s->ser_lock);
	ASSERT((s->ser_first == NULL) && (s->ser_last == NULL));
	/*
	 * Wakeup other potential waiters.
	 */
	cv_signal(&s->ser_cv);
	mutex_exit(&s->ser_lock);
}

/*
 * Create a new serializer.
 */
serializer_t *
serializer_create(int flags)
{
	return (kmem_cache_alloc(serializer_cache, flags));
}

/*
 * Wait for all pending entries to drain and then destroy serializer.
 */
void
serializer_destroy(serializer_t *s)
{
	serializer_wait(s);

	ASSERT(s->ser_owner == NULL);
	ASSERT(s->ser_taskq == 0);
	ASSERT(s->ser_count == 0);
	ASSERT(s->ser_first == NULL);
	ASSERT(s->ser_last == NULL);

	kmem_cache_free(serializer_cache, s);
}

/*ARGSUSED*/
static int
serializer_constructor(void *buf, void *cdrarg, int kmflags)
{
	serializer_t *s = buf;

	mutex_init(&s->ser_lock, NULL, MUTEX_DEFAULT, NULL);
	cv_init(&s->ser_cv, NULL, CV_DEFAULT, NULL);

	s->ser_taskq = 0;
	s->ser_count = 0;
	s->ser_first = s->ser_last = s->ser_curr = NULL;
	s->ser_proc = NULL;
	s->ser_arg = NULL;
	s->ser_owner = NULL;
	return (0);
}

/*ARGSUSED*/
static void
serializer_destructor(void *buf, void *cdrarg)
{
	serializer_t *s = buf;

	mutex_destroy(&s->ser_lock);
	cv_destroy(&s->ser_cv);
}

void
serializer_init(void)
{
	serializer_cache = kmem_cache_create("serializer_cache",
	    sizeof (serializer_t), 0, serializer_constructor,
	    serializer_destructor, NULL, NULL, NULL, 0);
}