sys/kern/subr_workqueue.c

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