From a9c933114de52715a30587a695cc7f70f3814003 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Gr=C3=A9goire?= Date: Mon, 29 Jun 2026 17:12:46 -0400 Subject: [PATCH] ps+ke: add thread reaping --- src/ex/fireworks.zig | 7 ++-- src/ex/init.zig | 6 +-- src/ex/workqueue.zig | 4 +- src/kern/root.zig | 4 ++ src/kern/sched.zig | 97 +++++++++++++++++++++++++++----------------- src/kern/thread.zig | 27 ++++++++++++ src/proc/private.zig | 1 + src/proc/root.zig | 1 + src/proc/thread.zig | 37 +++++++++++++++++ src/rtl/handoff.zig | 2 + 10 files changed, 140 insertions(+), 46 deletions(-) diff --git a/src/ex/fireworks.zig b/src/ex/fireworks.zig index 352ea02..4529a98 100644 --- a/src/ex/fireworks.zig +++ b/src/ex/fireworks.zig @@ -150,8 +150,7 @@ fn particle(param: ?*anyopaque) void { data.vel_y += @divTrunc(int_to_fp(10) * delay, 1000); } - // Now block. - ke.sched.block(); + ps.thread.exit(); } fn spawn_particle(arg: ?*anyopaque) void { @@ -214,7 +213,7 @@ fn explodeable(_: ?*anyopaque) void { spawn_particle(param); } - ke.sched.block(); + ps.thread.exit(); } pub fn start(param: ?*anyopaque) void { @@ -239,6 +238,8 @@ pub fn start(param: ?*anyopaque) void { spawn_explodeable(); } + std.log.info("async: {}, sync: {}, usable memory: {} KiB", .{ mm.private.tlb.async_shootdowns.load(.monotonic), mm.private.tlb.sync_shootdowns.load(.monotonic), mm.private.phys.usable_memory.load(.monotonic) / 1024 }); + perform_delay(2000); } } diff --git a/src/ex/init.zig b/src/ex/init.zig index 771f49f..c252756 100644 --- a/src/ex/init.zig +++ b/src/ex/init.zig @@ -21,10 +21,10 @@ pub fn init(boot_info: *pl.BootInfo) void { exp.console.init(); if (boot_info.framebuffer != null) { - fbconsole.init(boot_info); + // fbconsole.init(boot_info); - // const t = ps.thread.create_kernel(ke.Thread.Priority.default, ex.fireworks.start, boot_info) catch @panic("oom"); - // ke.sched.enqueue(&t.kern); + const t = ps.thread.create_kernel(ke.Thread.Priority.default, ex.fireworks.start, boot_info) catch @panic("oom"); + ke.sched.enqueue(&t.kern); } while (true) { diff --git a/src/ex/workqueue.zig b/src/ex/workqueue.zig index 5ea3dc2..e9b2296 100644 --- a/src/ex/workqueue.zig +++ b/src/ex/workqueue.zig @@ -38,8 +38,8 @@ pub const WorkItem = struct { } }; - pub fn init(priority: Priority, func: *const fn (arg: ?*anyopaque) void, arg: ?*anyopaque) WorkItem { - return .{ + pub fn init(self: *WorkItem, priority: Priority, func: *const fn (arg: ?*anyopaque) void, arg: ?*anyopaque) void { + self.* = .{ .func = func, .arg = arg, .link = undefined, diff --git a/src/kern/root.zig b/src/kern/root.zig index f163984..1a54990 100644 --- a/src/kern/root.zig +++ b/src/kern/root.zig @@ -33,6 +33,9 @@ pub const ipl = struct { pub const thread = struct { pub const init = p.thread.init; + pub const exit = p.thread.exit; + pub const current = p.thread.current; + pub const reaper_list = &p.thread.reaper_list; }; pub const dpc = struct { @@ -51,6 +54,7 @@ pub const sched = struct { pub const block = p.sched.block; pub const unblock = p.sched.unblock; pub const late_init = p.sched.late_init; + pub const yield = p.sched.yield; }; pub const timer = struct { diff --git a/src/kern/sched.zig b/src/kern/sched.zig index 27f1b5f..005a7d8 100644 --- a/src/kern/sched.zig +++ b/src/kern/sched.zig @@ -231,6 +231,27 @@ pub fn enqueue(td: *ke.Thread) void { td.lock.release(ipl); } +/// Yield on the current CPU. +pub fn yield() void { + const ipl = ke.ipl.raise(.Dispatch); + const td = percpu.local().current_thread.?; + + td.lock.acquire_no_ipl(); + + const cpu = percpu.local(); + if (td != cpu.idle_thread and td.state.load(.monotonic) == .Running) { + cpu.queues_lock.acquire_no_ipl(); + insert_in_queue(cpu, td, false); + td.cpu = ke.cpu.current(); + cpu.queues_lock.release_no_ipl(); + } + + yield_locked(); + + // td lock released + ke.ipl.lower(ipl); +} + /// Block the currently running thread. pub fn block() void { const ipl = ke.ipl.raise(.Dispatch); @@ -249,7 +270,44 @@ pub fn block_locked(curtd: *ke.Thread) void { curtd.sleep_start = ke.time.read_time(); curtd.runq = null; - yield(ke.cpu.current()); + yield_locked(); +} + +/// Yield the current CPU with the current thread already locked at Dispatch IPL. +pub fn yield_locked() void { + const sched_cpu = percpu.local(); + + sched_cpu.queues_lock.acquire_no_ipl(); + + const cur = sched_cpu.current_thread; + var next = sched_cpu.next_thread; + + if (next != null) { + sched_cpu.next_thread = null; + } else { + // Pick a new thread to run. + next = select_thread(null, sched_cpu, false); + } + + if (next == null) { + // Nothing to run, go idle. + next = sched_cpu.idle_thread; + sched_cpu.steal_work = true; + } + + sched_cpu.queues_lock.release_no_ipl(); + + if (next != null and cur.? != next.?) { + // Switch into the thread. Take its lock here to ensure it is not still on its stack. + next.?.lock.acquire_no_ipl(); + next.?.lock.release_no_ipl(); + + do_switch(sched_cpu, cur.?, next.?); + } else { + cur.?.lock.release_no_ipl(); + } + + // cur lock dropped } pub fn unblock_locked(td: *ke.Thread) void { @@ -856,43 +914,6 @@ fn do_switch(cpu: *PerCpu, cur: *ke.Thread, next: *ke.Thread) void { cur.context.switch_to(&next.context); } -// Yield on `cpu` -fn yield(cpu: u32) void { - const sched_cpu = percpu.remote(cpu); - - sched_cpu.queues_lock.acquire_no_ipl(); - - const cur = sched_cpu.current_thread; - var next = sched_cpu.next_thread; - - if (next != null) { - sched_cpu.next_thread = null; - } else { - // Pick a new thread to run. - next = select_thread(null, sched_cpu, false); - } - - if (next == null) { - // Nothing to run, go idle. - next = sched_cpu.idle_thread; - sched_cpu.steal_work = true; - } - - sched_cpu.queues_lock.release_no_ipl(); - - if (next != null and cur.? != next.?) { - // Switch into the thread. Take its lock here to ensure it is not still on its stack. - next.?.lock.acquire_no_ipl(); - next.?.lock.release_no_ipl(); - - do_switch(sched_cpu, cur.?, next.?); - } else { - cur.?.lock.release_no_ipl(); - } - - // cur lock dropped -} - fn find_most_loaded(exclude: ?*ke.CpuMask) ?u32 { var most: ?u32 = null; var most_load: usize = 0; diff --git a/src/kern/thread.zig b/src/kern/thread.zig index 32b6efd..31f0121 100644 --- a/src/kern/thread.zig +++ b/src/kern/thread.zig @@ -108,6 +108,8 @@ pub const Thread = struct { /// Queue this thread is associated with. queue: ?*ke.Queue, queue_item: ?*rtl.List.Entry, + /// Base of the stack. + stack: r.VAddr, /// Initialize a thread. /// - `stack`: Address of the **base** of the stack on which the initial context for the thread is built @@ -142,6 +144,7 @@ pub const Thread = struct { .waiting_on = null, .queue = null, .queue_item = null, + .stack = stack, }; thread.turnstiles_owned.init(); @@ -167,3 +170,27 @@ pub const Thread = struct { return self.priority >= Priority.low_interactive and self.priority <= Priority.high_interactive; } }; + +/// HandoffList of threads waiting to be reaped. +/// This is managed by the process subsystem. +pub var reaper_list: rtl.HandoffList = undefined; + +/// Terminate the currently running thread. +/// This does not return. +pub fn exit() void { + _ = ke.ipl.raise(.Dispatch); + const curtd = current(); + + curtd.lock.acquire_no_ipl(); + curtd.state.store(.Zombie, .monotonic); + + // Reuse the runq linkage to put on reaper list. + reaper_list.insert(@ptrCast(&curtd.runq_link.next)); + + ki.sched.yield_locked(); +} + +/// Return the currently running thread. +pub fn current() *ke.Thread { + return ki.sched.percpu.local().current_thread.?; +} diff --git a/src/proc/private.zig b/src/proc/private.zig index eb90a19..0c56ea0 100644 --- a/src/proc/private.zig +++ b/src/proc/private.zig @@ -25,4 +25,5 @@ fn turnstile_ctor(ts: *ki.turnstile.Turnstile) void { /// Initialize the process subsystem. pub fn init() void { turnstile_zone.init("turnstiles", .{ .ctor = turnstile_ctor }); + thread.init(); } diff --git a/src/proc/root.zig b/src/proc/root.zig index 2bfa1b2..99532e0 100644 --- a/src/proc/root.zig +++ b/src/proc/root.zig @@ -11,6 +11,7 @@ pub const Thread = p.thread.Thread; // === Exported interfaces === pub const thread = struct { pub const create_kernel = p.thread.create_kernel; + pub const exit = p.thread.exit; }; pub const init = p.init; diff --git a/src/proc/thread.zig b/src/proc/thread.zig index ab29970..605a27a 100644 --- a/src/proc/thread.zig +++ b/src/proc/thread.zig @@ -1,5 +1,8 @@ //! Higher-level thread management. +const rtl = @import("rtl"); +const std = @import("std"); const r = @import("root"); +const ex = r.ex; const psp = r.ps.private; const ke = r.ke; const ki = ke.private; @@ -7,12 +10,40 @@ const mm = r.mm; const kernel_thread_stack_size = r.kib(16); +var reaper_item: ex.WorkItem = undefined; + /// Higher-level wrapper over a thread. pub const Thread = struct { /// Lower-level part of the thread. kern: ke.Thread, }; +fn reap_thread(obj: *anyopaque, _: ?*anyopaque) void { + const link: **rtl.List.Entry = @ptrCast(@alignCast(obj)); + + // Bleh + const entry: *rtl.List.Entry = @fieldParentPtr("next", link); + const ketd: *ke.Thread = @fieldParentPtr("runq_link", entry); + const td: *Thread = @fieldParentPtr("kern", ketd); + + psp.turnstile_zone.destroy(td.kern.turnstile); + mm.heap.free(td.kern.stack, kernel_thread_stack_size); + mm.zone.gpa.destroy(td); +} + +fn reaper_fn(_: ?*anyopaque) void { + ke.thread.reaper_list.process(reap_thread, null); +} + +fn activation(_: *rtl.HandoffList) void { + ex.workqueue.enqueue(&reaper_item); +} + +pub fn init() void { + ke.thread.reaper_list.* = .init(activation); + reaper_item.init(.High, reaper_fn, null); +} + /// Create a kernel thread. /// - `prio`: base priority of the thread. /// - `entry`: entry point. @@ -26,3 +57,9 @@ pub fn create_kernel(prio: u8, entry: *const fn (arg: ?*anyopaque) void, arg: ?* return td; } + +/// Exit the currently running thread. +/// For now, this is just a wrapper over the kernel function. +pub fn exit() void { + ke.thread.exit(); +} diff --git a/src/rtl/handoff.zig b/src/rtl/handoff.zig index 530d22f..1a3ba86 100644 --- a/src/rtl/handoff.zig +++ b/src/rtl/handoff.zig @@ -41,6 +41,8 @@ pub const HandoffList = struct { } /// Pop the entire list and process it. + /// The callback takes the object pointer as its first parameter + /// and ctx as its second. pub fn process(self: *Self, callback: *const fn (*anyopaque, ?*anyopaque) void, ctx: ?*anyopaque) void { while (true) { const head = self.head.swap(processing_tag, .acquire);