xref: /netbsd-src/sys/kern/subr_workqueue.c (revision 88ab7da936c561aba9ff7492617ffb08f8d08ece)
1 /*	$NetBSD: subr_workqueue.c,v 1.13 2007/07/09 21:10:55 ad Exp $	*/
2 
3 /*-
4  * Copyright (c)2002, 2005 YAMAMOTO Takashi,
5  * All rights reserved.
6  *
7  * Redistribution and use in source and binary forms, with or without
8  * modification, are permitted provided that the following conditions
9  * are met:
10  * 1. Redistributions of source code must retain the above copyright
11  *    notice, this list of conditions and the following disclaimer.
12  * 2. Redistributions in binary form must reproduce the above copyright
13  *    notice, this list of conditions and the following disclaimer in the
14  *    documentation and/or other materials provided with the distribution.
15  *
16  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
17  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
18  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
19  * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
20  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
21  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
22  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
23  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
24  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
25  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
26  * SUCH DAMAGE.
27  */
28 
29 #include <sys/cdefs.h>
30 __KERNEL_RCSID(0, "$NetBSD: subr_workqueue.c,v 1.13 2007/07/09 21:10:55 ad Exp $");
31 
32 #include <sys/param.h>
33 #include <sys/systm.h>
34 #include <sys/kthread.h>
35 #include <sys/kmem.h>
36 #include <sys/proc.h>
37 #include <sys/workqueue.h>
38 #include <sys/mutex.h>
39 #include <sys/condvar.h>
40 
41 SIMPLEQ_HEAD(workqhead, work);
42 
43 struct workqueue_queue {
44 	kmutex_t q_mutex;
45 	kcondvar_t q_cv;
46 	struct workqhead q_queue;
47 	struct lwp *q_worker;
48 };
49 
50 struct workqueue {
51 	struct workqueue_queue wq_queue; /* todo: make this per-cpu */
52 
53 	void (*wq_func)(struct work *, void *);
54 	void *wq_arg;
55 	const char *wq_name;
56 	pri_t wq_prio;
57 	ipl_cookie_t wq_ipl;
58 };
59 
60 #define	POISON	0xaabbccdd
61 
62 static void
63 workqueue_runlist(struct workqueue *wq, struct workqhead *list)
64 {
65 	struct work *wk;
66 	struct work *next;
67 
68 	/*
69 	 * note that "list" is not a complete SIMPLEQ.
70 	 */
71 
72 	for (wk = SIMPLEQ_FIRST(list); wk != NULL; wk = next) {
73 		next = SIMPLEQ_NEXT(wk, wk_entry);
74 		(*wq->wq_func)(wk, wq->wq_arg);
75 	}
76 }
77 
78 static void
79 workqueue_run(struct workqueue *wq)
80 {
81 	struct workqueue_queue *q = &wq->wq_queue;
82 
83 	for (;;) {
84 		struct workqhead tmp;
85 
86 		/*
87 		 * we violate abstraction of SIMPLEQ.
88 		 */
89 
90 #if defined(DIAGNOSTIC)
91 		tmp.sqh_last = (void *)POISON;
92 #endif /* defined(DIAGNOSTIC) */
93 
94 		mutex_enter(&q->q_mutex);
95 		while (SIMPLEQ_EMPTY(&q->q_queue))
96 			cv_wait(&q->q_cv, &q->q_mutex);
97 		tmp.sqh_first = q->q_queue.sqh_first; /* XXX */
98 		SIMPLEQ_INIT(&q->q_queue);
99 		mutex_exit(&q->q_mutex);
100 
101 		workqueue_runlist(wq, &tmp);
102 	}
103 }
104 
105 static void
106 workqueue_worker(void *arg)
107 {
108 	struct workqueue *wq = arg;
109 	struct lwp *l;
110 
111 	l = curlwp;
112 	lwp_lock(l);
113 	l->l_priority = wq->wq_prio;
114 	l->l_usrpri = wq->wq_prio;
115 	lwp_unlock(l);
116 
117 	workqueue_run(wq);
118 }
119 
120 static void
121 workqueue_init(struct workqueue *wq, const char *name,
122     void (*callback_func)(struct work *, void *), void *callback_arg,
123     pri_t prio, int ipl)
124 {
125 
126 	wq->wq_ipl = makeiplcookie(ipl);
127 	wq->wq_prio = prio;
128 	wq->wq_name = name;
129 	wq->wq_func = callback_func;
130 	wq->wq_arg = callback_arg;
131 }
132 
133 static int
134 workqueue_initqueue(struct workqueue *wq, int ipl, int flags)
135 {
136 	struct workqueue_queue *q = &wq->wq_queue;
137 	int error, ktf;
138 
139 	mutex_init(&q->q_mutex, MUTEX_DRIVER, ipl);
140 	cv_init(&q->q_cv, wq->wq_name);
141 	SIMPLEQ_INIT(&q->q_queue);
142 	ktf = ((flags & WQ_MPSAFE) != 0 ? KTHREAD_MPSAFE : 0);
143 	error = kthread_create(wq->wq_prio, ktf, NULL, workqueue_worker,
144 	    wq, &q->q_worker, wq->wq_name);
145 
146 	return error;
147 }
148 
149 struct workqueue_exitargs {
150 	struct work wqe_wk;
151 	struct workqueue_queue *wqe_q;
152 };
153 
154 static void
155 workqueue_exit(struct work *wk, void *arg)
156 {
157 	struct workqueue_exitargs *wqe = (void *)wk;
158 	struct workqueue_queue *q = wqe->wqe_q;
159 
160 	/*
161 	 * only competition at this point is workqueue_finiqueue.
162 	 */
163 
164 	KASSERT(q->q_worker == curlwp);
165 	mutex_enter(&q->q_mutex);
166 	q->q_worker = NULL;
167 	cv_signal(&q->q_cv);
168 	mutex_exit(&q->q_mutex);
169 	kthread_exit(0);
170 }
171 
172 static void
173 workqueue_finiqueue(struct workqueue *wq)
174 {
175 	struct workqueue_queue *q = &wq->wq_queue;
176 	struct workqueue_exitargs wqe;
177 
178 	wq->wq_func = workqueue_exit;
179 
180 	wqe.wqe_q = q;
181 	KASSERT(SIMPLEQ_EMPTY(&q->q_queue));
182 	KASSERT(q->q_worker != NULL);
183 	mutex_enter(&q->q_mutex);
184 	SIMPLEQ_INSERT_TAIL(&q->q_queue, &wqe.wqe_wk, wk_entry);
185 	cv_signal(&q->q_cv);
186 	while (q->q_worker != NULL) {
187 		cv_wait(&q->q_cv, &q->q_mutex);
188 	}
189 	mutex_exit(&q->q_mutex);
190 	mutex_destroy(&q->q_mutex);
191 	cv_destroy(&q->q_cv);
192 }
193 
194 /* --- */
195 
196 int
197 workqueue_create(struct workqueue **wqp, const char *name,
198     void (*callback_func)(struct work *, void *), void *callback_arg,
199     pri_t prio, int ipl, int flags)
200 {
201 	struct workqueue *wq;
202 	int error;
203 
204 	wq = kmem_alloc(sizeof(*wq), KM_SLEEP);
205 	if (wq == NULL) {
206 		return ENOMEM;
207 	}
208 
209 	workqueue_init(wq, name, callback_func, callback_arg, prio, ipl);
210 
211 	error = workqueue_initqueue(wq, ipl, flags);
212 	if (error) {
213 		kmem_free(wq, sizeof(*wq));
214 		return error;
215 	}
216 
217 	*wqp = wq;
218 	return 0;
219 }
220 
221 void
222 workqueue_destroy(struct workqueue *wq)
223 {
224 
225 	workqueue_finiqueue(wq);
226 	kmem_free(wq, sizeof(*wq));
227 }
228 
229 void
230 workqueue_enqueue(struct workqueue *wq, struct work *wk)
231 {
232 	struct workqueue_queue *q = &wq->wq_queue;
233 
234 	mutex_enter(&q->q_mutex);
235 	SIMPLEQ_INSERT_TAIL(&q->q_queue, wk, wk_entry);
236 	cv_signal(&q->q_cv);
237 	mutex_exit(&q->q_mutex);
238 }
239