Home | History | Annotate | Line # | Download | only in kern
subr_workqueue.c revision 1.20 
      1 /*	$NetBSD: subr_workqueue.c,v 1.20 2007/08/07 10:42:22 yamt Exp $	*/
      2 
      3 /*-
      4  * Copyright (c)2002, 2005, 2006, 2007 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.20 2007/08/07 10:42:22 yamt Exp $");
     31 
     32 #include <sys/param.h>
     33 #include <sys/cpu.h>
     34 #include <sys/systm.h>
     35 #include <sys/kthread.h>
     36 #include <sys/kmem.h>
     37 #include <sys/proc.h>
     38 #include <sys/workqueue.h>
     39 #include <sys/mutex.h>
     40 #include <sys/condvar.h>
     41 #include <sys/queue.h>
     42 
     43 typedef struct work_impl {
     44 	SIMPLEQ_ENTRY(work_impl) wk_entry;
     45 } work_impl_t;
     46 
     47 SIMPLEQ_HEAD(workqhead, work_impl);
     48 
     49 struct workqueue_queue {
     50 	kmutex_t q_mutex;
     51 	kcondvar_t q_cv;
     52 	struct workqhead q_queue;
     53 	struct lwp *q_worker;
     54 };
     55 
     56 struct workqueue {
     57 	void (*wq_func)(struct work *, void *);
     58 	void *wq_arg;
     59 	int wq_flags;
     60 
     61 	const char *wq_name;
     62 	pri_t wq_prio;
     63 	void *wq_ptr;
     64 };
     65 
     66 #ifdef MULTIPROCESSOR
     67 #define	CPU_ALIGN_SIZE		CACHE_LINE_SIZE
     68 #else
     69 #define	CPU_ALIGN_SIZE		(ALIGNBYTES + 1)
     70 #endif
     71 
     72 #define	WQ_SIZE		(roundup2(sizeof(struct workqueue), CPU_ALIGN_SIZE))
     73 #define	WQ_QUEUE_SIZE	(roundup2(sizeof(struct workqueue_queue), CPU_ALIGN_SIZE))
     74 
     75 #define	POISON	0xaabbccdd
     76 
     77 static size_t
     78 workqueue_size(int flags)
     79 {
     80 
     81 	return WQ_SIZE
     82 	    + ((flags & WQ_PERCPU) != 0 ? ncpu : 1) * WQ_QUEUE_SIZE
     83 	    + CPU_ALIGN_SIZE;
     84 }
     85 
     86 static struct workqueue_queue *
     87 workqueue_queue_lookup(struct workqueue *wq, struct cpu_info *ci)
     88 {
     89 	u_int idx = 0;
     90 
     91 	if (wq->wq_flags & WQ_PERCPU) {
     92 		idx = ci ? cpu_index(ci) : cpu_index(curcpu());
     93 	}
     94 
     95 	return (void *)((intptr_t)(wq) + WQ_SIZE + (idx * WQ_QUEUE_SIZE));
     96 }
     97 
     98 static void
     99 workqueue_runlist(struct workqueue *wq, struct workqhead *list)
    100 {
    101 	work_impl_t *wk;
    102 	work_impl_t *next;
    103 
    104 	/*
    105 	 * note that "list" is not a complete SIMPLEQ.
    106 	 */
    107 
    108 	for (wk = SIMPLEQ_FIRST(list); wk != NULL; wk = next) {
    109 		next = SIMPLEQ_NEXT(wk, wk_entry);
    110 		(*wq->wq_func)((void *)wk, wq->wq_arg);
    111 	}
    112 }
    113 
    114 static void
    115 workqueue_run(struct workqueue *wq)
    116 {
    117 	struct workqueue_queue *q;
    118 
    119 	/* find the workqueue of this kthread */
    120 	q = workqueue_queue_lookup(wq, curlwp->l_cpu);
    121 
    122 	for (;;) {
    123 		struct workqhead tmp;
    124 
    125 		/*
    126 		 * we violate abstraction of SIMPLEQ.
    127 		 */
    128 
    129 #if defined(DIAGNOSTIC)
    130 		tmp.sqh_last = (void *)POISON;
    131 #endif /* defined(DIAGNOSTIC) */
    132 
    133 		mutex_enter(&q->q_mutex);
    134 		while (SIMPLEQ_EMPTY(&q->q_queue))
    135 			cv_wait(&q->q_cv, &q->q_mutex);
    136 		tmp.sqh_first = q->q_queue.sqh_first; /* XXX */
    137 		SIMPLEQ_INIT(&q->q_queue);
    138 		mutex_exit(&q->q_mutex);
    139 
    140 		workqueue_runlist(wq, &tmp);
    141 	}
    142 }
    143 
    144 static void
    145 workqueue_worker(void *arg)
    146 {
    147 	struct workqueue *wq = arg;
    148 	struct lwp *l;
    149 
    150 	l = curlwp;
    151 	lwp_lock(l);
    152 	l->l_priority = wq->wq_prio;
    153 	l->l_usrpri = wq->wq_prio;
    154 	lwp_unlock(l);
    155 
    156 	workqueue_run(wq);
    157 }
    158 
    159 static void
    160 workqueue_init(struct workqueue *wq, const char *name,
    161     void (*callback_func)(struct work *, void *), void *callback_arg,
    162     pri_t prio, int ipl)
    163 {
    164 
    165 	wq->wq_prio = prio;
    166 	wq->wq_name = name;
    167 	wq->wq_func = callback_func;
    168 	wq->wq_arg = callback_arg;
    169 }
    170 
    171 static int
    172 workqueue_initqueue(struct workqueue *wq, struct workqueue_queue *q,
    173     int ipl, struct cpu_info *ci)
    174 {
    175 	int error, ktf;
    176 
    177 	KASSERT(q->q_worker == NULL);
    178 
    179 	mutex_init(&q->q_mutex, MUTEX_DRIVER, ipl);
    180 	cv_init(&q->q_cv, wq->wq_name);
    181 	SIMPLEQ_INIT(&q->q_queue);
    182 	ktf = ((wq->wq_flags & WQ_MPSAFE) != 0 ? KTHREAD_MPSAFE : 0);
    183 	if (ci) {
    184 		error = kthread_create(wq->wq_prio, ktf, ci, workqueue_worker,
    185 		    wq, &q->q_worker, "%s/%u", wq->wq_name, (u_int)ci->ci_cpuid);
    186 	} else {
    187 		error = kthread_create(wq->wq_prio, ktf, ci, workqueue_worker,
    188 		    wq, &q->q_worker, "%s", wq->wq_name);
    189 	}
    190 	if (error != 0) {
    191 		mutex_destroy(&q->q_mutex);
    192 		cv_destroy(&q->q_cv);
    193 		KASSERT(q->q_worker == NULL);
    194 	}
    195 	return error;
    196 }
    197 
    198 struct workqueue_exitargs {
    199 	work_impl_t wqe_wk;
    200 	struct workqueue_queue *wqe_q;
    201 };
    202 
    203 static void
    204 workqueue_exit(struct work *wk, void *arg)
    205 {
    206 	struct workqueue_exitargs *wqe = (void *)wk;
    207 	struct workqueue_queue *q = wqe->wqe_q;
    208 
    209 	/*
    210 	 * only competition at this point is workqueue_finiqueue.
    211 	 */
    212 
    213 	KASSERT(q->q_worker == curlwp);
    214 	KASSERT(SIMPLEQ_EMPTY(&q->q_queue));
    215 	mutex_enter(&q->q_mutex);
    216 	q->q_worker = NULL;
    217 	cv_signal(&q->q_cv);
    218 	mutex_exit(&q->q_mutex);
    219 	kthread_exit(0);
    220 }
    221 
    222 static void
    223 workqueue_finiqueue(struct workqueue *wq, struct workqueue_queue *q)
    224 {
    225 	struct workqueue_exitargs wqe;
    226 
    227 	KASSERT(wq->wq_func == workqueue_exit);
    228 
    229 	wqe.wqe_q = q;
    230 	KASSERT(SIMPLEQ_EMPTY(&q->q_queue));
    231 	KASSERT(q->q_worker != NULL);
    232 	mutex_enter(&q->q_mutex);
    233 	SIMPLEQ_INSERT_TAIL(&q->q_queue, &wqe.wqe_wk, wk_entry);
    234 	cv_signal(&q->q_cv);
    235 	while (q->q_worker != NULL) {
    236 		cv_wait(&q->q_cv, &q->q_mutex);
    237 	}
    238 	mutex_exit(&q->q_mutex);
    239 	mutex_destroy(&q->q_mutex);
    240 	cv_destroy(&q->q_cv);
    241 }
    242 
    243 /* --- */
    244 
    245 int
    246 workqueue_create(struct workqueue **wqp, const char *name,
    247     void (*callback_func)(struct work *, void *), void *callback_arg,
    248     pri_t prio, int ipl, int flags)
    249 {
    250 	struct workqueue *wq;
    251 	struct workqueue_queue *q;
    252 	void *ptr;
    253 	int error = 0;
    254 
    255 	KASSERT(sizeof(work_impl_t) <= sizeof(struct work));
    256 
    257 	ptr = kmem_zalloc(workqueue_size(flags), KM_SLEEP);
    258 	wq = (void *)roundup2((intptr_t)ptr, CPU_ALIGN_SIZE);
    259 	wq->wq_ptr = ptr;
    260 	wq->wq_flags = flags;
    261 
    262 	workqueue_init(wq, name, callback_func, callback_arg, prio, ipl);
    263 
    264 	if (flags & WQ_PERCPU) {
    265 		struct cpu_info *ci;
    266 		CPU_INFO_ITERATOR cii;
    267 
    268 		/* create the work-queue for each CPU */
    269 		for (CPU_INFO_FOREACH(cii, ci)) {
    270 			q = workqueue_queue_lookup(wq, ci);
    271 			error = workqueue_initqueue(wq, q, ipl, ci);
    272 			if (error) {
    273 				break;
    274 			}
    275 		}
    276 	} else {
    277 		/* initialize a work-queue */
    278 		q = workqueue_queue_lookup(wq, NULL);
    279 		error = workqueue_initqueue(wq, q, ipl, NULL);
    280 	}
    281 
    282 	if (error != 0) {
    283 		workqueue_destroy(wq);
    284 	} else {
    285 		*wqp = wq;
    286 	}
    287 
    288 	return error;
    289 }
    290 
    291 void
    292 workqueue_destroy(struct workqueue *wq)
    293 {
    294 	struct workqueue_queue *q;
    295 	struct cpu_info *ci;
    296 	CPU_INFO_ITERATOR cii;
    297 
    298 	wq->wq_func = workqueue_exit;
    299 	for (CPU_INFO_FOREACH(cii, ci)) {
    300 		q = workqueue_queue_lookup(wq, ci);
    301 		if (q->q_worker != NULL) {
    302 			workqueue_finiqueue(wq, q);
    303 		}
    304 	}
    305 	kmem_free(wq->wq_ptr, workqueue_size(wq->wq_flags));
    306 }
    307 
    308 void
    309 workqueue_enqueue(struct workqueue *wq, struct work *wk0, struct cpu_info *ci)
    310 {
    311 	struct workqueue_queue *q;
    312 	work_impl_t *wk = (void *)wk0;
    313 
    314 	KASSERT(wq->wq_flags & WQ_PERCPU || ci == NULL);
    315 	q = workqueue_queue_lookup(wq, ci);
    316 
    317 	mutex_enter(&q->q_mutex);
    318 	SIMPLEQ_INSERT_TAIL(&q->q_queue, wk, wk_entry);
    319 	cv_signal(&q->q_cv);
    320 	mutex_exit(&q->q_mutex);
    321 }
    322