lib: Add workqueues

include/zinnia/workqueue.h

@@ -0,0 +1,36 @@
+// SPDX-License-Identifier: ISC
+
+#pragma once
+
+#include <zinnia/atomic.h>
+#include <zinnia/errno.h>
+#include <zinnia/initgraph.h>
+#include <zinnia/tailq.h>
+#include <stdint.h>
+
+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);

lib/meson.build

@@ -25,4 +25,5 @@ kernel_sources += files(
     'string.c',
     'timer.c',
     'turnstile.c',
+    'workqueue.c',
 )

lib/workqueue.c

@@ -0,0 +1,184 @@
+// SPDX-License-Identifier: ISC
+
+#include <zinnia/alloc.h>
+#include <zinnia/assert.h>
+#include <zinnia/atomic.h>
+#include <zinnia/percpu.h>
+#include <zinnia/process.h>
+#include <zinnia/sched.h>
+#include <zinnia/spin.h>
+#include <zinnia/workqueue.h>
+
+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);