ex: add workqueue

src/ex/init.zig

@@ -11,6 +11,9 @@ const turnstile = ke.private.turnstile;
 var mtx: ke.private.mutex.Mutex = .init();
 var shared_counter: usize = 0;
 
+var work_item: ex.WorkItem = undefined;
+var work_item2: ex.WorkItem = undefined;
+
 fn mutex_test(_: ?*anyopaque) void {
     while (true) {
         mtx.acquire();
@@ -19,6 +22,18 @@ fn mutex_test(_: ?*anyopaque) void {
     }
 }
 
+fn foo(_: ?*anyopaque) void {
+    std.log.info("foo cpu {}", .{ke.cpu.current()});
+    var timer: ke.Timer = undefined;
+    timer.init();
+    ke.timer.set(&timer, std.time.ns_per_ms * 500, null);
+    _ = ke.wait.wait_one(&timer.hdr, null) catch unreachable;
+}
+
+fn bar(_: ?*anyopaque) void {
+    std.log.info("bar cpu {}", .{ke.cpu.current()});
+}
+
 pub fn init(boot_info: *pl.BootInfo) void {
     mm.init(boot_info);
 
@@ -26,6 +41,13 @@ pub fn init(boot_info: *pl.BootInfo) void {
     mm.late_init();
     ke.sched.late_init();
     ps.init();
+    ex.private.workqueue.init() catch @panic("e");
+
+    work_item = .init(.Normal, foo, null);
+    work_item2 = .init(.Normal, bar, null);
+
+    ex.workqueue.enqueue(&work_item);
+    ex.workqueue.enqueue(&work_item2);
 
     if (false) {
         if (boot_info.framebuffer != null) {

src/ex/private.zig

@@ -1,3 +1,3 @@
 pub const init_mod = @import("init.zig");
-
+pub const workqueue = @import("workqueue.zig");
 pub const init = init_mod.init;

src/ex/root.zig

@@ -1,2 +1,10 @@
 pub const fireworks = @import("fireworks.zig");
 pub const private = @import("private.zig");
+
+const p = private;
+
+pub const WorkItem = p.workqueue.WorkItem;
+
+pub const workqueue = struct {
+    pub const enqueue = p.workqueue.enqueue;
+};

src/ex/workqueue.zig

@@ -0,0 +1,157 @@
+//! Work queues and work items.
+//! Work queues are used to enqueue work to be done in a thread context asynchronously.
+//! There are three global work queues for each priority:
+//! - High
+//! - Normal
+//! - Low
+//! High and Normal work queues are per-cpu, and there is one global queue and thread for low-priority work.
+//! Each CPU has one worker thread, until that thread blocks, then another thread is spawned to pick up work on it.
+const std = @import("std");
+const rtl = @import("rtl");
+const r = @import("root");
+const ke = r.ke;
+const ps = r.ps;
+const mm = r.mm;
+
+/// Represents a single work item.
+pub const WorkItem = struct {
+    func: *const fn (arg: ?*anyopaque) void,
+    arg: ?*anyopaque,
+    link: rtl.List.Entry,
+    enqueued: std.atomic.Value(bool),
+    priority: Priority,
+
+    pub const Priority = enum {
+        /// Low priority background work.
+        Low,
+        /// Normal priority work.
+        Normal,
+        /// High priority work.
+        High,
+
+        pub fn to_sched_prio(prio: Priority) u8 {
+            return switch (prio) {
+                .Low => ke.Thread.Priority.low_batch,
+                .Normal => ke.Thread.Priority.high_batch,
+                .High => ke.Thread.Priority.low_realtime,
+            };
+        }
+    };
+
+    pub fn init(priority: Priority, func: *const fn (arg: ?*anyopaque) void, arg: ?*anyopaque) WorkItem {
+        return .{
+            .func = func,
+            .arg = arg,
+            .link = undefined,
+            .enqueued = .init(false),
+            .priority = priority,
+        };
+    }
+};
+
+/// Represents a thread worker pool and its queue.
+const Pool = struct {
+    const Worker = struct {
+        link: rtl.List.Entry,
+        td: *ps.Thread,
+    };
+    /// The enqueued items.
+    queue: ke.Queue,
+    /// Priority of the pool.
+    prio: WorkItem.Priority,
+    /// CPU this pool is on.
+    cpu: ?u32,
+    lock: ke.SpinLock,
+
+    pub fn init(self: *Pool, prio: WorkItem.Priority, cpu: ?u32) !void {
+        self.queue.init(1);
+        self.prio = prio;
+        self.cpu = cpu;
+        self.lock = .init();
+
+        // TODO: dynamic worker pool.
+        // We don't support thread exits yet, so add that first.
+        // Do linux-style where worker threads themselves spawn new threads.
+        const how_many: usize = if (prio == .Low) 4 else 2;
+
+        for (0..how_many) |_| {
+            var td = try ps.thread.create_kernel(
+                WorkItem.Priority.to_sched_prio(self.prio),
+                work_loop,
+                self,
+            );
+
+            if (cpu != null) {
+                td.kern.last_cpu = cpu;
+                td.kern.pinned = true;
+            }
+
+            ke.sched.enqueue(&td.kern);
+        }
+    }
+};
+
+/// Per-CPU work data.
+const PerCpu = struct {
+    /// High-priority work queue.
+    high: Pool,
+    /// Normal-priority work queue.
+    normal: Pool,
+};
+
+const percpu = ke.CpuLocal(PerCpu, undefined);
+
+/// Low-priority work queue.
+var low: Pool = undefined;
+
+fn work_loop(p: ?*anyopaque) void {
+    const pool: *Pool = @ptrCast(@alignCast(p));
+
+    while (true) {
+        const item = pool.queue.remove();
+        const work: *WorkItem = @fieldParentPtr("link", item);
+
+        work.func(work.arg);
+
+        work.enqueued.store(false, .release);
+    }
+}
+
+/// Enqueue a work item to be executed eventually.
+pub fn enqueue(item: *WorkItem) void {
+    // Ensure CPU stays consistent.
+    const ipl = ke.ipl.raise(.Dispatch);
+    defer ke.ipl.lower(ipl);
+
+    var pool: *Pool = undefined;
+
+    switch (item.priority) {
+        .High => {
+            pool = &percpu.local().high;
+        },
+        .Normal => {
+            pool = &percpu.local().normal;
+        },
+        .Low => {
+            pool = &low;
+        },
+    }
+
+    pool.queue.insert(&item.link, .Tail);
+}
+
+/// Initialize the workqueue subsystem.
+pub fn init() !void {
+    // Ensure readying the threads doesn't preempt us.
+    const ipl = ke.ipl.raise(.Dispatch);
+    defer ke.ipl.lower(ipl);
+
+    for (0..ke.ncpus) |cpu| {
+        const c = percpu.remote(@intCast(cpu));
+
+        try c.high.init(.High, @intCast(cpu));
+        try c.normal.init(.Normal, @intCast(cpu));
+    }
+
+    try low.init(.Low, null);
+}