Home | History | Annotate | Line # | Download | only in kern
subr_workqueue.c revision 1.22.8.1 
      1 /*	$NetBSD: subr_workqueue.c,v 1.22.8.1 2008/03/24 07:16:14 keiichi 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.22.8.1 2008/03/24 07:16:14 keiichi 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_worker(void *cookie)
    116 {
    117 	struct workqueue *wq = cookie;
    118 	struct workqueue_queue *q;
    119 
    120 	/* find the workqueue of this kthread */
    121 	q = workqueue_queue_lookup(wq, curlwp->l_cpu);
    122 
    123 	for (;;) {
    124 		struct workqhead tmp;
    125 
    126 		/*
    127 		 * we violate abstraction of SIMPLEQ.
    128 		 */
    129 
    130 #if defined(DIAGNOSTIC)
    131 		tmp.sqh_last = (void *)POISON;
    132 #endif /* defined(DIAGNOSTIC) */
    133 
    134 		mutex_enter(&q->q_mutex);
    135 		while (SIMPLEQ_EMPTY(&q->q_queue))
    136 			cv_wait(&q->q_cv, &q->q_mutex);
    137 		tmp.sqh_first = q->q_queue.sqh_first; /* XXX */
    138 		SIMPLEQ_INIT(&q->q_queue);
    139 		mutex_exit(&q->q_mutex);
    140 
    141 		workqueue_runlist(wq, &tmp);
    142 	}
    143 }
    144 
    145 static void
    146 workqueue_init(struct workqueue *wq, const char *name,
    147     void (*callback_func)(struct work *, void *), void *callback_arg,
    148     pri_t prio, int ipl)
    149 {
    150 
    151 	wq->wq_prio = prio;
    152 	wq->wq_name = name;
    153 	wq->wq_func = callback_func;
    154 	wq->wq_arg = callback_arg;
    155 }
    156 
    157 static int
    158 workqueue_initqueue(struct workqueue *wq, struct workqueue_queue *q,
    159     int ipl, struct cpu_info *ci)
    160 {
    161 	int error, ktf;
    162 
    163 	KASSERT(q->q_worker == NULL);
    164 
    165 	mutex_init(&q->q_mutex, MUTEX_DEFAULT, ipl);
    166 	cv_init(&q->q_cv, wq->wq_name);
    167 	SIMPLEQ_INIT(&q->q_queue);
    168 	ktf = ((wq->wq_flags & WQ_MPSAFE) != 0 ? KTHREAD_MPSAFE : 0);
    169 	if (ci) {
    170 		error = kthread_create(wq->wq_prio, ktf, ci, workqueue_worker,
    171 		    wq, &q->q_worker, "%s/%u", wq->wq_name, ci->ci_index);
    172 	} else {
    173 		error = kthread_create(wq->wq_prio, ktf, ci, workqueue_worker,
    174 		    wq, &q->q_worker, "%s", wq->wq_name);
    175 	}
    176 	if (error != 0) {
    177 		mutex_destroy(&q->q_mutex);
    178 		cv_destroy(&q->q_cv);
    179 		KASSERT(q->q_worker == NULL);
    180 	}
    181 	return error;
    182 }
    183 
    184 struct workqueue_exitargs {
    185 	work_impl_t wqe_wk;
    186 	struct workqueue_queue *wqe_q;
    187 };
    188 
    189 static void
    190 workqueue_exit(struct work *wk, void *arg)
    191 {
    192 	struct workqueue_exitargs *wqe = (void *)wk;
    193 	struct workqueue_queue *q = wqe->wqe_q;
    194 
    195 	/*
    196 	 * only competition at this point is workqueue_finiqueue.
    197 	 */
    198 
    199 	KASSERT(q->q_worker == curlwp);
    200 	KASSERT(SIMPLEQ_EMPTY(&q->q_queue));
    201 	mutex_enter(&q->q_mutex);
    202 	q->q_worker = NULL;
    203 	cv_signal(&q->q_cv);
    204 	mutex_exit(&q->q_mutex);
    205 	kthread_exit(0);
    206 }
    207 
    208 static void
    209 workqueue_finiqueue(struct workqueue *wq, struct workqueue_queue *q)
    210 {
    211 	struct workqueue_exitargs wqe;
    212 
    213 	KASSERT(wq->wq_func == workqueue_exit);
    214 
    215 	wqe.wqe_q = q;
    216 	KASSERT(SIMPLEQ_EMPTY(&q->q_queue));
    217 	KASSERT(q->q_worker != NULL);
    218 	mutex_enter(&q->q_mutex);
    219 	SIMPLEQ_INSERT_TAIL(&q->q_queue, &wqe.wqe_wk, wk_entry);
    220 	cv_signal(&q->q_cv);
    221 	while (q->q_worker != NULL) {
    222 		cv_wait(&q->q_cv, &q->q_mutex);
    223 	}
    224 	mutex_exit(&q->q_mutex);
    225 	mutex_destroy(&q->q_mutex);
    226 	cv_destroy(&q->q_cv);
    227 }
    228 
    229 /* --- */
    230 
    231 int
    232 workqueue_create(struct workqueue **wqp, const char *name,
    233     void (*callback_func)(struct work *, void *), void *callback_arg,
    234     pri_t prio, int ipl, int flags)
    235 {
    236 	struct workqueue *wq;
    237 	struct workqueue_queue *q;
    238 	void *ptr;
    239 	int error = 0;
    240 
    241 	KASSERT(sizeof(work_impl_t) <= sizeof(struct work));
    242 
    243 	ptr = kmem_zalloc(workqueue_size(flags), KM_SLEEP);
    244 	wq = (void *)roundup2((intptr_t)ptr, CPU_ALIGN_SIZE);
    245 	wq->wq_ptr = ptr;
    246 	wq->wq_flags = flags;
    247 
    248 	workqueue_init(wq, name, callback_func, callback_arg, prio, ipl);
    249 
    250 	if (flags & WQ_PERCPU) {
    251 		struct cpu_info *ci;
    252 		CPU_INFO_ITERATOR cii;
    253 
    254 		/* create the work-queue for each CPU */
    255 		for (CPU_INFO_FOREACH(cii, ci)) {
    256 			q = workqueue_queue_lookup(wq, ci);
    257 			error = workqueue_initqueue(wq, q, ipl, ci);
    258 			if (error) {
    259 				break;
    260 			}
    261 		}
    262 	} else {
    263 		/* initialize a work-queue */
    264 		q = workqueue_queue_lookup(wq, NULL);
    265 		error = workqueue_initqueue(wq, q, ipl, NULL);
    266 	}
    267 
    268 	if (error != 0) {
    269 		workqueue_destroy(wq);
    270 	} else {
    271 		*wqp = wq;
    272 	}
    273 
    274 	return error;
    275 }
    276 
    277 void
    278 workqueue_destroy(struct workqueue *wq)
    279 {
    280 	struct workqueue_queue *q;
    281 	struct cpu_info *ci;
    282 	CPU_INFO_ITERATOR cii;
    283 
    284 	wq->wq_func = workqueue_exit;
    285 	for (CPU_INFO_FOREACH(cii, ci)) {
    286 		q = workqueue_queue_lookup(wq, ci);
    287 		if (q->q_worker != NULL) {
    288 			workqueue_finiqueue(wq, q);
    289 		}
    290 	}
    291 	kmem_free(wq->wq_ptr, workqueue_size(wq->wq_flags));
    292 }
    293 
    294 void
    295 workqueue_enqueue(struct workqueue *wq, struct work *wk0, struct cpu_info *ci)
    296 {
    297 	struct workqueue_queue *q;
    298 	work_impl_t *wk = (void *)wk0;
    299 
    300 	KASSERT(wq->wq_flags & WQ_PERCPU || ci == NULL);
    301 	q = workqueue_queue_lookup(wq, ci);
    302 
    303 	mutex_enter(&q->q_mutex);
    304 	SIMPLEQ_INSERT_TAIL(&q->q_queue, wk, wk_entry);
    305 	cv_signal(&q->q_cv);
    306 	mutex_exit(&q->q_mutex);
    307 }
    308