subr_workqueue.c revision 1.37.6.1
1 /* $NetBSD: subr_workqueue.c,v 1.37.6.1 2024/04/18 15:51:35 martin 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.37.6.1 2024/04/18 15:51:35 martin Exp $"); 31 32 #include <sys/param.h> 33 34 #include <sys/condvar.h> 35 #include <sys/cpu.h> 36 #include <sys/kmem.h> 37 #include <sys/kthread.h> 38 #include <sys/mutex.h> 39 #include <sys/proc.h> 40 #include <sys/queue.h> 41 #include <sys/sdt.h> 42 #include <sys/systm.h> 43 #include <sys/workqueue.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_pending; 55 uint64_t q_gen; 56 lwp_t *q_worker; 57 }; 58 59 struct workqueue { 60 void (*wq_func)(struct work *, void *); 61 void *wq_arg; 62 int wq_flags; 63 64 char wq_name[MAXCOMLEN]; 65 pri_t wq_prio; 66 void *wq_ptr; 67 }; 68 69 #define WQ_SIZE (roundup2(sizeof(struct workqueue), coherency_unit)) 70 #define WQ_QUEUE_SIZE (roundup2(sizeof(struct workqueue_queue), coherency_unit)) 71 72 #define POISON 0xaabbccdd 73 74 SDT_PROBE_DEFINE7(sdt, kernel, workqueue, create, 75 "struct workqueue *"/*wq*/, 76 "const char *"/*name*/, 77 "void (*)(struct work *, void *)"/*func*/, 78 "void *"/*arg*/, 79 "pri_t"/*prio*/, 80 "int"/*ipl*/, 81 "int"/*flags*/); 82 SDT_PROBE_DEFINE1(sdt, kernel, workqueue, destroy, 83 "struct workqueue *"/*wq*/); 84 85 SDT_PROBE_DEFINE3(sdt, kernel, workqueue, enqueue, 86 "struct workqueue *"/*wq*/, 87 "struct work *"/*wk*/, 88 "struct cpu_info *"/*ci*/); 89 SDT_PROBE_DEFINE4(sdt, kernel, workqueue, entry, 90 "struct workqueue *"/*wq*/, 91 "struct work *"/*wk*/, 92 "void (*)(struct work *, void *)"/*func*/, 93 "void *"/*arg*/); 94 SDT_PROBE_DEFINE4(sdt, kernel, workqueue, return, 95 "struct workqueue *"/*wq*/, 96 "struct work *"/*wk*/, 97 "void (*)(struct work *, void *)"/*func*/, 98 "void *"/*arg*/); 99 SDT_PROBE_DEFINE2(sdt, kernel, workqueue, wait__start, 100 "struct workqueue *"/*wq*/, 101 "struct work *"/*wk*/); 102 SDT_PROBE_DEFINE2(sdt, kernel, workqueue, wait__self, 103 "struct workqueue *"/*wq*/, 104 "struct work *"/*wk*/); 105 SDT_PROBE_DEFINE2(sdt, kernel, workqueue, wait__hit, 106 "struct workqueue *"/*wq*/, 107 "struct work *"/*wk*/); 108 SDT_PROBE_DEFINE2(sdt, kernel, workqueue, wait__done, 109 "struct workqueue *"/*wq*/, 110 "struct work *"/*wk*/); 111 112 SDT_PROBE_DEFINE1(sdt, kernel, workqueue, exit__start, 113 "struct workqueue *"/*wq*/); 114 SDT_PROBE_DEFINE1(sdt, kernel, workqueue, exit__done, 115 "struct workqueue *"/*wq*/); 116 117 static size_t 118 workqueue_size(int flags) 119 { 120 121 return WQ_SIZE 122 + ((flags & WQ_PERCPU) != 0 ? ncpu : 1) * WQ_QUEUE_SIZE 123 + coherency_unit; 124 } 125 126 static struct workqueue_queue * 127 workqueue_queue_lookup(struct workqueue *wq, struct cpu_info *ci) 128 { 129 u_int idx = 0; 130 131 if (wq->wq_flags & WQ_PERCPU) { 132 idx = ci ? cpu_index(ci) : cpu_index(curcpu()); 133 } 134 135 return (void *)((uintptr_t)(wq) + WQ_SIZE + (idx * WQ_QUEUE_SIZE)); 136 } 137 138 static void 139 workqueue_runlist(struct workqueue *wq, struct workqhead *list) 140 { 141 work_impl_t *wk; 142 work_impl_t *next; 143 144 for (wk = SIMPLEQ_FIRST(list); wk != NULL; wk = next) { 145 next = SIMPLEQ_NEXT(wk, wk_entry); 146 SDT_PROBE4(sdt, kernel, workqueue, entry, 147 wq, wk, wq->wq_func, wq->wq_arg); 148 (*wq->wq_func)((void *)wk, wq->wq_arg); 149 SDT_PROBE4(sdt, kernel, workqueue, return, 150 wq, wk, wq->wq_func, wq->wq_arg); 151 } 152 } 153 154 static void 155 workqueue_worker(void *cookie) 156 { 157 struct workqueue *wq = cookie; 158 struct workqueue_queue *q; 159 160 /* find the workqueue of this kthread */ 161 q = workqueue_queue_lookup(wq, curlwp->l_cpu); 162 163 mutex_enter(&q->q_mutex); 164 for (;;) { 165 struct workqhead tmp; 166 167 SIMPLEQ_INIT(&tmp); 168 169 while (SIMPLEQ_EMPTY(&q->q_queue_pending)) 170 cv_wait(&q->q_cv, &q->q_mutex); 171 SIMPLEQ_CONCAT(&tmp, &q->q_queue_pending); 172 SIMPLEQ_INIT(&q->q_queue_pending); 173 174 /* 175 * Mark the queue as actively running a batch of work 176 * by setting the generation number odd. 177 */ 178 q->q_gen |= 1; 179 mutex_exit(&q->q_mutex); 180 181 workqueue_runlist(wq, &tmp); 182 183 /* 184 * Notify workqueue_wait that we have completed a batch 185 * of work by incrementing the generation number. 186 */ 187 mutex_enter(&q->q_mutex); 188 KASSERTMSG(q->q_gen & 1, "q=%p gen=%"PRIu64, q, q->q_gen); 189 q->q_gen++; 190 cv_broadcast(&q->q_cv); 191 } 192 mutex_exit(&q->q_mutex); 193 } 194 195 static void 196 workqueue_init(struct workqueue *wq, const char *name, 197 void (*callback_func)(struct work *, void *), void *callback_arg, 198 pri_t prio, int ipl) 199 { 200 201 KASSERT(sizeof(wq->wq_name) > strlen(name)); 202 strncpy(wq->wq_name, name, sizeof(wq->wq_name)); 203 204 wq->wq_prio = prio; 205 wq->wq_func = callback_func; 206 wq->wq_arg = callback_arg; 207 } 208 209 static int 210 workqueue_initqueue(struct workqueue *wq, struct workqueue_queue *q, 211 int ipl, struct cpu_info *ci) 212 { 213 int error, ktf; 214 215 KASSERT(q->q_worker == NULL); 216 217 mutex_init(&q->q_mutex, MUTEX_DEFAULT, ipl); 218 cv_init(&q->q_cv, wq->wq_name); 219 SIMPLEQ_INIT(&q->q_queue_pending); 220 q->q_gen = 0; 221 ktf = ((wq->wq_flags & WQ_MPSAFE) != 0 ? KTHREAD_MPSAFE : 0); 222 if (wq->wq_prio < PRI_KERNEL) 223 ktf |= KTHREAD_TS; 224 if (ci) { 225 error = kthread_create(wq->wq_prio, ktf, ci, workqueue_worker, 226 wq, &q->q_worker, "%s/%u", wq->wq_name, ci->ci_index); 227 } else { 228 error = kthread_create(wq->wq_prio, ktf, ci, workqueue_worker, 229 wq, &q->q_worker, "%s", wq->wq_name); 230 } 231 if (error != 0) { 232 mutex_destroy(&q->q_mutex); 233 cv_destroy(&q->q_cv); 234 KASSERT(q->q_worker == NULL); 235 } 236 return error; 237 } 238 239 struct workqueue_exitargs { 240 work_impl_t wqe_wk; 241 struct workqueue_queue *wqe_q; 242 }; 243 244 static void 245 workqueue_exit(struct work *wk, void *arg) 246 { 247 struct workqueue_exitargs *wqe = (void *)wk; 248 struct workqueue_queue *q = wqe->wqe_q; 249 250 /* 251 * only competition at this point is workqueue_finiqueue. 252 */ 253 254 KASSERT(q->q_worker == curlwp); 255 KASSERT(SIMPLEQ_EMPTY(&q->q_queue_pending)); 256 mutex_enter(&q->q_mutex); 257 q->q_worker = NULL; 258 cv_broadcast(&q->q_cv); 259 mutex_exit(&q->q_mutex); 260 kthread_exit(0); 261 } 262 263 static void 264 workqueue_finiqueue(struct workqueue *wq, struct workqueue_queue *q) 265 { 266 struct workqueue_exitargs wqe; 267 268 KASSERT(wq->wq_func == workqueue_exit); 269 270 wqe.wqe_q = q; 271 KASSERT(SIMPLEQ_EMPTY(&q->q_queue_pending)); 272 KASSERT(q->q_worker != NULL); 273 mutex_enter(&q->q_mutex); 274 SIMPLEQ_INSERT_TAIL(&q->q_queue_pending, &wqe.wqe_wk, wk_entry); 275 cv_broadcast(&q->q_cv); 276 while (q->q_worker != NULL) { 277 cv_wait(&q->q_cv, &q->q_mutex); 278 } 279 mutex_exit(&q->q_mutex); 280 mutex_destroy(&q->q_mutex); 281 cv_destroy(&q->q_cv); 282 } 283 284 /* --- */ 285 286 int 287 workqueue_create(struct workqueue **wqp, const char *name, 288 void (*callback_func)(struct work *, void *), void *callback_arg, 289 pri_t prio, int ipl, int flags) 290 { 291 struct workqueue *wq; 292 struct workqueue_queue *q; 293 void *ptr; 294 int error = 0; 295 296 CTASSERT(sizeof(work_impl_t) <= sizeof(struct work)); 297 298 ptr = kmem_zalloc(workqueue_size(flags), KM_SLEEP); 299 wq = (void *)roundup2((uintptr_t)ptr, coherency_unit); 300 wq->wq_ptr = ptr; 301 wq->wq_flags = flags; 302 303 workqueue_init(wq, name, callback_func, callback_arg, prio, ipl); 304 305 if (flags & WQ_PERCPU) { 306 struct cpu_info *ci; 307 CPU_INFO_ITERATOR cii; 308 309 /* create the work-queue for each CPU */ 310 for (CPU_INFO_FOREACH(cii, ci)) { 311 q = workqueue_queue_lookup(wq, ci); 312 error = workqueue_initqueue(wq, q, ipl, ci); 313 if (error) { 314 break; 315 } 316 } 317 } else { 318 /* initialize a work-queue */ 319 q = workqueue_queue_lookup(wq, NULL); 320 error = workqueue_initqueue(wq, q, ipl, NULL); 321 } 322 323 if (error != 0) { 324 workqueue_destroy(wq); 325 } else { 326 *wqp = wq; 327 } 328 329 return error; 330 } 331 332 static bool 333 workqueue_q_wait(struct workqueue *wq, struct workqueue_queue *q, 334 work_impl_t *wk_target) 335 { 336 work_impl_t *wk; 337 bool found = false; 338 uint64_t gen; 339 340 mutex_enter(&q->q_mutex); 341 342 /* 343 * Avoid a deadlock scenario. We can't guarantee that 344 * wk_target has completed at this point, but we can't wait for 345 * it either, so do nothing. 346 * 347 * XXX Are there use-cases that require this semantics? 348 */ 349 if (q->q_worker == curlwp) { 350 SDT_PROBE2(sdt, kernel, workqueue, wait__self, wq, wk_target); 351 goto out; 352 } 353 354 /* 355 * Wait until the target is no longer pending. If we find it 356 * on this queue, the caller can stop looking in other queues. 357 * If we don't find it in this queue, however, we can't skip 358 * waiting -- it may be hidden in the running queue which we 359 * have no access to. 360 */ 361 again: 362 SIMPLEQ_FOREACH(wk, &q->q_queue_pending, wk_entry) { 363 if (wk == wk_target) { 364 SDT_PROBE2(sdt, kernel, workqueue, wait__hit, wq, wk); 365 found = true; 366 cv_wait(&q->q_cv, &q->q_mutex); 367 goto again; 368 } 369 } 370 371 /* 372 * The target may be in the batch of work currently running, 373 * but we can't touch that queue. So if there's anything 374 * running, wait until the generation changes. 375 */ 376 gen = q->q_gen; 377 if (gen & 1) { 378 do 379 cv_wait(&q->q_cv, &q->q_mutex); 380 while (gen == q->q_gen); 381 } 382 383 out: 384 mutex_exit(&q->q_mutex); 385 386 return found; 387 } 388 389 /* 390 * Wait for a specified work to finish. The caller must ensure that no new 391 * work will be enqueued before calling workqueue_wait. Note that if the 392 * workqueue is WQ_PERCPU, the caller can enqueue a new work to another queue 393 * other than the waiting queue. 394 */ 395 void 396 workqueue_wait(struct workqueue *wq, struct work *wk) 397 { 398 struct workqueue_queue *q; 399 bool found; 400 401 ASSERT_SLEEPABLE(); 402 403 SDT_PROBE2(sdt, kernel, workqueue, wait__start, wq, wk); 404 if (ISSET(wq->wq_flags, WQ_PERCPU)) { 405 struct cpu_info *ci; 406 CPU_INFO_ITERATOR cii; 407 for (CPU_INFO_FOREACH(cii, ci)) { 408 q = workqueue_queue_lookup(wq, ci); 409 found = workqueue_q_wait(wq, q, (work_impl_t *)wk); 410 if (found) 411 break; 412 } 413 } else { 414 q = workqueue_queue_lookup(wq, NULL); 415 (void)workqueue_q_wait(wq, q, (work_impl_t *)wk); 416 } 417 SDT_PROBE2(sdt, kernel, workqueue, wait__done, wq, wk); 418 } 419 420 void 421 workqueue_destroy(struct workqueue *wq) 422 { 423 struct workqueue_queue *q; 424 struct cpu_info *ci; 425 CPU_INFO_ITERATOR cii; 426 427 ASSERT_SLEEPABLE(); 428 429 SDT_PROBE1(sdt, kernel, workqueue, exit__start, wq); 430 wq->wq_func = workqueue_exit; 431 for (CPU_INFO_FOREACH(cii, ci)) { 432 q = workqueue_queue_lookup(wq, ci); 433 if (q->q_worker != NULL) { 434 workqueue_finiqueue(wq, q); 435 } 436 } 437 SDT_PROBE1(sdt, kernel, workqueue, exit__done, wq); 438 kmem_free(wq->wq_ptr, workqueue_size(wq->wq_flags)); 439 } 440 441 #ifdef DEBUG 442 static void 443 workqueue_check_duplication(struct workqueue_queue *q, work_impl_t *wk) 444 { 445 work_impl_t *_wk; 446 447 SIMPLEQ_FOREACH(_wk, &q->q_queue_pending, wk_entry) { 448 if (_wk == wk) 449 panic("%s: tried to enqueue a queued work", __func__); 450 } 451 } 452 #endif 453 454 void 455 workqueue_enqueue(struct workqueue *wq, struct work *wk0, struct cpu_info *ci) 456 { 457 struct workqueue_queue *q; 458 work_impl_t *wk = (void *)wk0; 459 460 SDT_PROBE3(sdt, kernel, workqueue, enqueue, wq, wk0, ci); 461 462 KASSERT(wq->wq_flags & WQ_PERCPU || ci == NULL); 463 q = workqueue_queue_lookup(wq, ci); 464 465 mutex_enter(&q->q_mutex); 466 #ifdef DEBUG 467 workqueue_check_duplication(q, wk); 468 #endif 469 SIMPLEQ_INSERT_TAIL(&q->q_queue_pending, wk, wk_entry); 470 cv_broadcast(&q->q_cv); 471 mutex_exit(&q->q_mutex); 472 } 473
Indexes created Tue Sep 23 07:09:52 GMT 2025