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