add an interface for work queues run from a kernel thread.

this type of code is implemented many times already in the kernel, this is
a generic version of all those replicated code bases.

originally from tedu@
ok tedu@ tom@ deraadt@

4 files changed, 247 insertions, 2 deletions

diff --git a/sys/conf/files b/sys/conf/files
index 49115795c4a..8475d8b2303 100644
--- a/sys/conf/files
+++ b/sys/conf/files
@@ -1,4 +1,4 @@
-#	$OpenBSD: files,v 1.403 2007/05/30 08:10:02 uwe Exp $
+#	$OpenBSD: files,v 1.404 2007/05/31 18:16:59 dlg Exp $
 #	$NetBSD: files,v 1.87 1996/05/19 17:17:50 jonathan Exp $
 
 #	@(#)files.newconf	7.5 (Berkeley) 5/10/93
@@ -654,6 +654,7 @@ file kern/kern_tc.c
 file kern/kern_time.c
 file kern/kern_timeout.c
 file kern/kern_watchdog.c		!small_kernel
+file kern/kern_workq.c
 file kern/kern_xxx.c
 file kern/kgdb_stub.c			kgdb
 file kern/sched_bsd.c
diff --git a/sys/kern/init_main.c b/sys/kern/init_main.c
index 46f9faf3ade..91069c58232 100644
--- a/sys/kern/init_main.c
+++ b/sys/kern/init_main.c
@@ -1,4 +1,4 @@
-/*	$OpenBSD: init_main.c,v 1.140 2007/05/16 17:27:30 art Exp $	*/
+/*	$OpenBSD: init_main.c,v 1.141 2007/05/31 18:16:59 dlg Exp $	*/
 /*	$NetBSD: init_main.c,v 1.84.4.1 1996/06/02 09:08:06 mrg Exp $	*/
 
 /*
@@ -73,6 +73,7 @@
 #include <sys/domain.h>
 #include <sys/mbuf.h>
 #include <sys/pipe.h>
+#include <sys/workq.h>
 
 #include <sys/syscall.h>
 #include <sys/syscallargs.h>
@@ -139,6 +140,7 @@ void	start_reaper(void *);
 void	start_crypto(void *);
 void	init_exec(void);
 void	kqueue_init(void);
+void	workq_init(void);
 
 extern char sigcode[], esigcode[];
 #ifdef SYSCALL_DEBUG
@@ -330,6 +332,9 @@ main(void *framep)
 	/* Initialize run queues */
 	rqinit();
 
+	/* Initialize work queues */
+	workq_init();
+
 	/* Configure the devices */
 	cpu_configure();
 
diff --git a/sys/kern/kern_workq.c b/sys/kern/kern_workq.c
new file mode 100644
index 00000000000..e3435bb8fca
--- /dev/null
+++ b/sys/kern/kern_workq.c
@@ -0,0 +1,206 @@
+/*	$OpenBSD: kern_workq.c,v 1.1 2007/05/31 18:16:59 dlg Exp $ */
+
+/*
+ * Copyright (c) 2007 David Gwynne <dlg@openbsd.org>
+ * Copyright (c) 2007 Ted Unangst <tedu@openbsd.org>
+ *
+ * Permission to use, copy, modify, and distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+#include <sys/param.h>
+#include <sys/systm.h>
+#include <sys/malloc.h>
+#include <sys/pool.h>
+#include <sys/queue.h>
+#include <sys/kthread.h>
+#include <sys/stdarg.h>
+#include <sys/workq.h>
+
+struct workq_task {
+	int		wqt_flags;
+	void		(*wqt_func)(void *, void *);
+	void		*wqt_arg1;
+	void		*wqt_arg2;
+
+	SIMPLEQ_ENTRY(workq_task) wqt_entry;
+};
+
+struct workq {
+	int		wq_flags;
+#define WQ_F_RUNNING		(1<<0)
+	int		wq_running;
+	int		wq_busy;
+	int		wq_max;
+	const char	*wq_name;
+
+	SIMPLEQ_HEAD(, workq_task) wq_tasklist;
+};
+
+struct pool	workq_task_pool;
+struct workq	workq_syswq = {
+	WQ_F_RUNNING,
+	0,
+	0,
+	1,
+	"syswq"
+};
+
+void		workq_init(void); /* called in init_main.c */
+void		workq_init_syswq(void *);
+void		workq_create_thread(void *);
+void		workq_thread(void *);
+
+void
+workq_init(void)
+{
+	pool_init(&workq_task_pool, sizeof(struct workq_task), 0, 0,
+	    0, "wqtasks", NULL);
+
+	SIMPLEQ_INIT(&workq_syswq.wq_tasklist);
+	kthread_create_deferred(workq_init_syswq, NULL);
+}
+
+void
+workq_init_syswq(void *arg)
+{
+	if (kthread_create(workq_thread, &workq_syswq, NULL, "%s",
+	    workq_syswq.wq_name) != 0)
+		panic("unable to create system work queue thread");
+
+	workq_syswq.wq_running++;
+}
+
+struct workq *
+workq_create(const char *name, int maxqs)
+{
+	struct workq *wq;
+
+	wq = malloc(sizeof(*wq), M_DEVBUF, M_NOWAIT);
+	if (wq == NULL)
+		return (NULL);
+
+	wq->wq_flags = WQ_F_RUNNING;
+	wq->wq_running = 0;
+	wq->wq_busy = 0;
+	wq->wq_max = maxqs;
+	wq->wq_name = name;
+	SIMPLEQ_INIT(&wq->wq_tasklist);
+
+	/* try to create a thread to guarantee that tasks will be serviced */
+	kthread_create_deferred(workq_create_thread, wq);
+
+	return (wq);
+}
+
+void
+workq_destroy(struct workq *wq)
+{
+	int			s;
+
+	s = splhigh();
+
+	wq->wq_flags &= ~WQ_F_RUNNING;
+	while (wq->wq_running != 0) {
+		wakeup(wq);
+		tsleep(&wq->wq_running, PWAIT, "wqdestroy", 0);
+	}
+
+	splx(s);
+
+	free(wq, M_DEVBUF);
+}
+
+int
+workq_add_task(struct workq *wq, void (*func)(void *, void *),
+    void *a1, void *a2, int flags)
+{
+	struct workq_task	*wqt;
+	int			wake = 1;
+	int			s;
+
+	if (wq == NULL)
+		wq = &workq_syswq;
+	
+	s = splhigh();
+	wqt = pool_get(&workq_task_pool, (flags & WQ_WAITOK) ?
+	    PR_WAITOK : PR_NOWAIT);
+	splx(s);
+	if (!wqt)
+		return (ENOMEM);
+
+	wqt->wqt_flags = flags;
+	wqt->wqt_func = func;
+	wqt->wqt_arg1 = a1;
+	wqt->wqt_arg2 = a2;
+
+	s = splhigh();
+	SIMPLEQ_INSERT_TAIL(&wq->wq_tasklist, wqt, wqt_entry);
+	if ((wq->wq_running < wq->wq_max) && (wq->wq_running == wq->wq_busy)) {
+		kthread_create_deferred(workq_create_thread, wq);
+		wake = 0;
+	}
+	splx(s);
+
+	if (wake)
+		wakeup_one(wq);
+
+	return (0);
+}
+
+void
+workq_create_thread(void *arg)
+{
+	struct workq		*wq = arg;
+	int			rv;
+
+	rv = kthread_create(workq_thread, wq, NULL, "%s", wq->wq_name);
+	if (rv != 0) {
+		printf("unable to create \"%s\" workq thread\n", wq->wq_name);
+		return;
+	}
+
+	wq->wq_running++;
+}
+
+void
+workq_thread(void *arg)
+{
+	struct workq		*wq = arg;
+	struct workq_task	*wqt;
+	int			s;
+
+	s = splhigh();
+	while (wq->wq_flags & WQ_F_RUNNING) {
+		while ((wqt = SIMPLEQ_FIRST(&wq->wq_tasklist)) != NULL) {
+			SIMPLEQ_REMOVE_HEAD(&wq->wq_tasklist, wqt_entry);
+			wq->wq_busy++;
+			splx(s);
+
+			if (wqt->wqt_flags & WQ_MPSAFE)
+				;
+			wqt->wqt_func(wqt->wqt_arg1, wqt->wqt_arg2);
+			if (wqt->wqt_flags & WQ_MPSAFE)
+				;
+
+			s = splhigh();
+			wq->wq_busy--;
+			pool_put(&workq_task_pool, wqt);
+		}
+		tsleep(wq, PWAIT, "bored", 0);
+	}
+	wq->wq_running--;
+	splx(s);
+	wakeup(&wq->wq_running);
+
+	kthread_exit(0);
+}
diff --git a/sys/sys/workq.h b/sys/sys/workq.h
new file mode 100644
index 00000000000..0059b4a4c33
--- /dev/null
+++ b/sys/sys/workq.h
@@ -0,0 +1,33 @@
+/*	$OpenBSD: workq.h,v 1.1 2007/05/31 18:16:59 dlg Exp $ */
+
+/*
+ * Copyright (c) 2007 David Gwynne <dlg@openbsd.org>
+ * Copyright (c) 2007 Ted Unangst <tedu@openbsd.org>
+ *
+ * Permission to use, copy, modify, and distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+#ifndef _SYS_WORKQ_H_
+#define _SYS_WORKQ_H_
+
+struct workq;
+
+#define WQ_WAITOK	(1<<0)
+#define WQ_MPSAFE	(1<<1)
+
+struct workq	*workq_create(const char *, int);
+int		workq_add_task(struct workq *, void (*func)(void *, void *),
+		    void *, void *, int);
+void		workq_destroy(struct workq *);
+
+#endif /* _SYS_WORKQ_H_ */