diff --git a/src/ex/init.zig b/src/ex/init.zig index b6e0e4a..e24757d 100644 --- a/src/ex/init.zig +++ b/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) { diff --git a/src/ex/private.zig b/src/ex/private.zig index 345149e..b3406fd 100644 --- a/src/ex/private.zig +++ b/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; diff --git a/src/ex/root.zig b/src/ex/root.zig index 9a94df9..d372565 100644 --- a/src/ex/root.zig +++ b/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; +}; diff --git a/src/ex/workqueue.zig b/src/ex/workqueue.zig new file mode 100644 index 0000000..dd9dd86 --- /dev/null +++ b/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); +}