From 244618dbeb544d99350d70b6cc558c35eeedbd11 Mon Sep 17 00:00:00 2001 From: Marvin Friedrich Date: Sun, 25 Oct 2026 19:15:00 +0100 Subject: [PATCH] lib: Add workqueues --- include/zinnia/workqueue.h | 36 ++++++++ lib/meson.build | 1 + lib/workqueue.c | 184 +++++++++++++++++++++++++++++++++++++ 3 files changed, 221 insertions(+) create mode 100644 include/zinnia/workqueue.h create mode 100644 lib/workqueue.c diff --git a/include/zinnia/workqueue.h b/include/zinnia/workqueue.h new file mode 100644 index 0000000..b9c6eee --- /dev/null +++ b/include/zinnia/workqueue.h @@ -0,0 +1,36 @@ +// SPDX-License-Identifier: ISC + +#pragma once + +#include +#include +#include +#include +#include + +struct work { + void (*func)(struct work* work); + tailq_link(struct work) link; + atomic_bool pending; +}; + +struct workqueue; + +// Workqueue for regular prios +extern struct workqueue* workqueue_system; +// Workqueue for high prios +extern struct workqueue* workqueue_high; + +void work_init(struct work* work, void (*func)(struct work* work)); + +errno_t workqueue_create(const char* name, uint8_t priority, struct workqueue** out); + +void workqueue_destroy(struct workqueue* wq); + +bool work_queue(struct workqueue* wq, struct work* work); + +bool work_cancel(struct workqueue* wq, struct work* work); + +void workqueue_flush(struct workqueue* wq); + +INIT_EXTERN(workqueue_stage); diff --git a/lib/meson.build b/lib/meson.build index b56e508..6bc9317 100644 --- a/lib/meson.build +++ b/lib/meson.build @@ -25,4 +25,5 @@ kernel_sources += files( 'string.c', 'timer.c', 'turnstile.c', + 'workqueue.c', ) diff --git a/lib/workqueue.c b/lib/workqueue.c new file mode 100644 index 0000000..24dcaaf --- /dev/null +++ b/lib/workqueue.c @@ -0,0 +1,184 @@ +// SPDX-License-Identifier: ISC + +#include +#include +#include +#include +#include +#include +#include +#include + +struct workqueue { + struct spinlock lock; + tailq_head(struct work) queue; + struct task* worker; + bool idle; + bool stopping; + struct task* exit_waiter; + atomic_bool exited; +}; + +struct work_barrier { + struct work work; + struct task* waiter; + atomic_bool done; +}; + +struct workqueue* workqueue_system; +struct workqueue* workqueue_high; + +void work_init(struct work* work, void (*func)(struct work* work)) { + work->func = func; + atomic_init(&work->pending, false); +} + +static struct task* current_task() { + return percpu_get()->sched.current; +} + +static void wait_for(atomic_bool* flag) { + while (!atomic_load(flag, ATOMIC_ACQUIRE)) + sched_block(&percpu_get()->sched); +} + +static void signal(atomic_bool* flag, struct task* waiter) { + atomic_store(flag, true, ATOMIC_RELEASE); + sched_wake(waiter); +} + +static struct work* next_work(struct workqueue* wq, void (**out_func)(struct work* work)) { + spin_lock(&wq->lock); + while (true) { + struct work* work = tailq_first(&wq->queue); + if (work) { + tailq_remove(&wq->queue, work, link); + *out_func = work->func; + atomic_store(&work->pending, false, ATOMIC_RELEASE); + spin_unlock(&wq->lock); + return work; + } + if (wq->stopping) { + spin_unlock(&wq->lock); + return nullptr; + } + + wq->idle = true; + spin_unlock(&wq->lock); + sched_block(&percpu_get()->sched); + spin_lock(&wq->lock); + } +} + +// Worker task main loop. +static void worker_fn(uintptr_t arg) { + struct workqueue* wq = (struct workqueue*)arg; + + struct work* work; + void (*func)(struct work* work); + while ((work = next_work(wq, &func))) + func(work); + + struct task* waiter = wq->exit_waiter; + signal(&wq->exited, waiter); + sched_exit(); +} + +errno_t workqueue_create(const char* name, uint8_t priority, struct workqueue** out) { + struct workqueue* wq = mem_alloc(sizeof(*wq), 0); + if (!wq) + return ENOMEM; + + spin_init(&wq->lock); + tailq_init(&wq->queue); + atomic_init(&wq->exited, false); + + errno_t status = task_create(name, process_kernel(), worker_fn, (uintptr_t)wq, &wq->worker); + if (status) { + mem_free(wq); + return status; + } + + sched_set_priority(wq->worker, priority); + sched_add_task(&percpu_get()->sched, wq->worker); + + *out = wq; + return 0; +} + +static void wake_worker(struct workqueue* wq, bool stop) { + spin_lock(&wq->lock); + if (stop) + wq->stopping = true; + const bool was_idle = wq->idle; + wq->idle = false; + spin_unlock(&wq->lock); + + if (was_idle) + sched_wake(wq->worker); +} + +void workqueue_destroy(struct workqueue* wq) { + wq->exit_waiter = current_task(); + wake_worker(wq, true); + wait_for(&wq->exited); + mem_free(wq); +} + +bool work_queue(struct workqueue* wq, struct work* work) { + if (atomic_exchange(&work->pending, true, ATOMIC_ACQ_REL)) + return false; + + spin_lock(&wq->lock); + assert(!wq->stopping, "Queued work on a workqueue that is being destroyed"); + tailq_insert_tail(&wq->queue, work, link); + const bool was_idle = wq->idle; + wq->idle = false; + spin_unlock(&wq->lock); + + if (was_idle) + sched_wake(wq->worker); + return true; +} + +bool work_cancel(struct workqueue* wq, struct work* work) { + bool removed = false; + spin_lock(&wq->lock); + for (struct work* it = tailq_first(&wq->queue); it; it = tailq_next(it, link)) { + if (it == work) { + tailq_remove(&wq->queue, work, link); + atomic_store(&work->pending, false, ATOMIC_RELEASE); + removed = true; + break; + } + } + spin_unlock(&wq->lock); + return removed; +} + +static void barrier_fn(struct work* work) { + struct work_barrier* barrier = containerof(work, struct work_barrier, work); + struct task* waiter = barrier->waiter; + signal(&barrier->done, waiter); +} + +void workqueue_flush(struct workqueue* wq) { + assert(current_task() != wq->worker, "Tried to flush a workqueue from its own worker"); + + struct work_barrier barrier = {.waiter = current_task()}; + work_init(&barrier.work, barrier_fn); + atomic_init(&barrier.done, false); + work_queue(wq, &barrier.work); + wait_for(&barrier.done); +} + +static void workqueue_stage_init(struct boot_info* info) { + errno_t status = workqueue_create("kworker", SCHED_PRIO_DEFAULT, &workqueue_system); + assert(status == 0, "Unable to create a workqueue"); + + status = workqueue_create("kworker/high", SCHED_PRIO_WORK_HIGH, &workqueue_high); + assert(status == 0, "Unable to create a workqueue"); +} + +INIT_TASK(workqueue_stage, "core.workqueue", workqueue_stage_init); +INIT_DEPENDS(workqueue_stage, sched_stage);