From 1dedb5dba1b93f3d587a4006ee3589a8c45b533d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Gr=C3=A9goire?= Date: Fri, 24 Jul 2026 19:56:25 -0400 Subject: [PATCH] tests: add tests --- src/ex/init.zig | 15 +- src/ex/root.zig | 1 - src/root.zig | 1 + src/{ex => tests}/fireworks.zig | 0 src/tests/mutex.zig | 183 +++++++++++++++++ src/tests/root.zig | 15 ++ src/{ex => tests}/sintab.zig | 0 src/tests/smr.zig | 171 ++++++++++++++++ src/tests/smr_pi.zig | 345 ++++++++++++++++++++++++++++++++ src/tests/turnstile.zig | 230 +++++++++++++++++++++ 10 files changed, 957 insertions(+), 4 deletions(-) rename src/{ex => tests}/fireworks.zig (100%) create mode 100644 src/tests/mutex.zig create mode 100644 src/tests/root.zig rename src/{ex => tests}/sintab.zig (100%) create mode 100644 src/tests/smr.zig create mode 100644 src/tests/smr_pi.zig create mode 100644 src/tests/turnstile.zig diff --git a/src/ex/init.zig b/src/ex/init.zig index db70333..71f5d5f 100644 --- a/src/ex/init.zig +++ b/src/ex/init.zig @@ -22,11 +22,20 @@ export fn kmain(boot_info: *r.BootInfo) callconv(.c) void { exp.work.init() catch @panic("e"); exp.console.init(); + if (boot_info.cmdline) |cline| { + if (rtl.cmdline.get_string(cline, "test")) |tst| { + // Run a test if specified. + if (r.tests.tests.get(tst)) |func| { + const t = ps.thread.create_kernel(ke.Thread.Priority.default, func, boot_info) catch @panic("oom"); + ke.sched.enqueue(&t.kern); + } else { + std.log.info("invalid test: \"{s}\"", .{tst}); + } + } + } + if (boot_info.framebuffer != null) { // 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); } while (true) { diff --git a/src/ex/root.zig b/src/ex/root.zig index 152a71b..5614e58 100644 --- a/src/ex/root.zig +++ b/src/ex/root.zig @@ -1,4 +1,3 @@ -pub const fireworks = @import("fireworks.zig"); pub const private = @import("private.zig"); const p = private; diff --git a/src/root.zig b/src/root.zig index 8161318..01d63fc 100644 --- a/src/root.zig +++ b/src/root.zig @@ -8,6 +8,7 @@ pub const arch = @import("arch"); pub const mm = @import("mm/root.zig"); pub const ps = @import("proc/root.zig"); pub const BootInfo = @import("loader/info.zig"); +pub const tests = @import("tests/root.zig"); const config = @import("config"); diff --git a/src/ex/fireworks.zig b/src/tests/fireworks.zig similarity index 100% rename from src/ex/fireworks.zig rename to src/tests/fireworks.zig diff --git a/src/tests/mutex.zig b/src/tests/mutex.zig new file mode 100644 index 0000000..576efec --- /dev/null +++ b/src/tests/mutex.zig @@ -0,0 +1,183 @@ +//! Mutex stress test. +const std = @import("std"); +const r = @import("root"); +const ke = r.ke; +const ps = r.ps; + +const nmutexes = 8; +const max_threads = 32; + +const Guarded = struct { + mutex: ke.Mutex, + guard: usize, + count: u64, +}; + +var locks: [nmutexes]Guarded = undefined; +var totals: [max_threads]u64 = undefined; +var tds: [max_threads]*ps.Thread = undefined; +var is_parked: [max_threads]std.atomic.Value(bool) = undefined; +var nthreads: usize = 0; + +var paused = std.atomic.Value(bool).init(false); +var parked = std.atomic.Value(usize).init(0); + +fn rand(state: *u64) u64 { + var x = state.*; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + state.* = x; + return x; +} + +fn delay_ms(ms: usize) void { + var timer: ke.Timer = undefined; + timer.init(); + ke.timer.set(&timer, std.time.ns_per_ms * ms, null); + _ = ke.wait.wait_one(&timer.hdr, null) catch unreachable; +} + +fn worker(param: ?*anyopaque) void { + const id = @intFromPtr(param.?) - 1; + var seed = (ke.time.read_time() ^ (id *% 0x9e3779b97f4a7c15)) | 1; + var iter: u64 = 0; + + while (true) : (iter += 1) { + if (paused.load(.acquire)) { + _ = parked.fetchAdd(1, .acq_rel); + is_parked[id].store(true, .release); + while (paused.load(.acquire)) delay_ms(1); + is_parked[id].store(false, .release); + _ = parked.fetchSub(1, .acq_rel); + } + + const first: usize = @intCast(rand(&seed) % nmutexes); + const last = @min(first + 1 + rand(&seed) % 2, nmutexes); + + for (first..last) |i| locks[i].mutex.acquire(); + + for (first..last) |i| { + if (@atomicLoad(usize, &locks[i].guard, .monotonic) != 0) { + @panic("mutex: guard set on entry, exclusion violated"); + } + + @atomicStore(usize, &locks[i].guard, id + 1, .monotonic); + } + if (rand(&seed) % 16 == 0) { + delay_ms(1); + } else { + for (0..rand(&seed) % 256) |_| std.atomic.spinLoopHint(); + } + + for (first..last) |i| { + const c = @atomicLoad(u64, &locks[i].count, .monotonic); + + for (0..rand(&seed) % 64) |_| std.atomic.spinLoopHint(); + + @atomicStore(u64, &locks[i].count, c + 1, .monotonic); + + if (@atomicLoad(usize, &locks[i].guard, .monotonic) != id + 1) { + @panic("mutex: guard changed while held, exclusion violated"); + } + + @atomicStore(usize, &locks[i].guard, 0, .monotonic); + } + + var i = last; + while (i > first) : (i -= 1) locks[i - 1].mutex.release(); + + totals[id] += last - first; + + if (iter % 64 == 0) delay_ms(1); + } +} + +fn dump_stuck_workers() void { + for (0..nthreads) |i| { + if (is_parked[i].load(.acquire)) continue; + + const td = &tds[i].kern; + const waiting = @atomicLoad(?*anyopaque, &td.waiting_on, .monotonic); + var lock_idx: isize = -1; + + for (&locks, 0..) |*l, j| { + if (waiting == @as(?*anyopaque, @ptrCast(&l.mutex))) lock_idx = @intCast(j); + } + + std.log.err("mutex: worker {} stuck: state {s}, prio {}/{}, inherited {}, waiting_on {x} (lock {})", .{ + i, + @tagName(td.state.load(.monotonic)), + @atomicLoad(u8, &td.priority, .monotonic), + @atomicLoad(u8, &td.base_priority, .monotonic), + @atomicLoad(u8, &td.inherited_prio, .monotonic), + if (waiting) |w| @intFromPtr(w) else 0, + lock_idx, + }); + } +} + +/// Park all workers (they hold nothing at the loop top), then verify that +/// no increments were lost and no guard is stuck. +fn quiesce_verify() u64 { + paused.store(true, .release); + + var spins: usize = 0; + while (parked.load(.acquire) != nthreads) : (spins += 1) { + if (spins > 10_000) { + dump_stuck_workers(); + @panic("mutex: workers failed to park"); + } + delay_ms(1); + } + + var counted: u64 = 0; + var expected: u64 = 0; + + for (&locks) |*l| { + if (@atomicLoad(usize, &l.guard, .monotonic) != 0) { + @panic("mutex: guard set at quiescence"); + } + + counted += @atomicLoad(u64, &l.count, .monotonic); + } + + for (0..nthreads) |i| expected += @atomicLoad(u64, &totals[i], .monotonic); + + if (counted != expected) { + std.log.err("mutex: lost updates: counted {}, expected {}", .{ counted, expected }); + @panic("mutex: lost updates, exclusion or ordering violated"); + } + + paused.store(false, .release); + while (parked.load(.acquire) != 0) delay_ms(1); + + return counted; +} + +pub fn start(_: ?*anyopaque) void { + for (&locks) |*l| { + l.mutex = .init(); + l.guard = 0; + l.count = 0; + } + + nthreads = @max(4, @min(ke.ncpus * 2, max_threads)); + + for (0..nthreads) |i| { + totals[i] = 0; + is_parked[i] = .init(false); + + const prio: u8 = @intCast(2 + (i * 5) % 29); + const t = ps.thread.create_kernel(prio, &worker, @ptrFromInt(i + 1)) catch @panic("oom"); + tds[i] = t; + ke.sched.enqueue(&t.kern); + } + + while (true) { + delay_ms(2000); + + const count = quiesce_verify(); + std.log.info("mutex: {} sections, no lost updates", .{count}); + } +} diff --git a/src/tests/root.zig b/src/tests/root.zig new file mode 100644 index 0000000..f33ff4b --- /dev/null +++ b/src/tests/root.zig @@ -0,0 +1,15 @@ +//! kernel-mode stress tests +const std = @import("std"); +const fireworks = @import("fireworks.zig"); +const smr = @import("smr.zig"); +const smr_pi = @import("smr_pi.zig"); +const turnstile = @import("turnstile.zig"); +const mutex = @import("mutex.zig"); + +pub const tests: std.StaticStringMap(*const fn (?*anyopaque) void) = .initComptime(.{ + .{ "fireworks", fireworks.start }, + .{ "smr", smr.start }, + .{ "smr_pi", smr_pi.start }, + .{ "turnstile", turnstile.start }, + .{ "mutex", mutex.start }, +}); diff --git a/src/ex/sintab.zig b/src/tests/sintab.zig similarity index 100% rename from src/ex/sintab.zig rename to src/tests/sintab.zig diff --git a/src/tests/smr.zig b/src/tests/smr.zig new file mode 100644 index 0000000..b4e7557 --- /dev/null +++ b/src/tests/smr.zig @@ -0,0 +1,171 @@ +//! SMR stress test. +const std = @import("std"); +const r = @import("root"); +const ke = r.ke; +const mm = r.mm; +const ps = r.ps; + +const ki = ke.private; + +const nslots = 128; +const obj_words = 8; +const reads_per_section = 16; + +const Object = struct { + vals: [obj_words]u64, +}; + +var dom: *ke.smr.Domain = undefined; +var zone: mm.zone.Zone = undefined; + +var slots = [_]std.atomic.Value(?*Object){.init(null)} ** nslots; + +var next_val = std.atomic.Value(u64).init(1); +var reads = std.atomic.Value(u64).init(0); +var writes = std.atomic.Value(u64).init(0); +var polls = std.atomic.Value(u64).init(0); + +fn rand(state: *u64) u64 { + var x = state.*; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + state.* = x; + return x; +} + +fn seed_from(param: ?*anyopaque) u64 { + return (ke.time.read_time() ^ (@intFromPtr(param.?) *% 0x9e3779b97f4a7c15)) | 1; +} + +fn delay_ms(ms: usize) void { + var timer: ke.Timer = undefined; + timer.init(); + ke.timer.set(&timer, std.time.ns_per_ms * ms, null); + _ = ke.wait.wait_one(&timer.hdr, null) catch unreachable; +} + +fn reader(param: ?*anyopaque) void { + var seed = seed_from(param); + var iter: u64 = 0; + + while (true) : (iter += 1) { + const ipl = ki.smr.enter(dom); + var done: u64 = 0; + + for (0..reads_per_section) |_| { + const idx: usize = @intCast(rand(&seed) % nslots); + const obj = slots[idx].load(.acquire) orelse continue; + + const v = @atomicLoad(u64, &obj.vals[0], .monotonic); + + for (1..obj_words) |i| { + if (@atomicLoad(u64, &obj.vals[i], .monotonic) != v) { + @panic("smr stress: torn read, object reused during read section"); + } + } + + done += 1; + } + + ki.smr.exit(dom, ipl); + + _ = reads.fetchAdd(done, .monotonic); + + if (iter % 512 == 0) ke.sched.yield(); + } +} + +fn writer(param: ?*anyopaque) void { + var seed = seed_from(param); + var iter: u64 = 0; + + while (true) : (iter += 1) { + const idx: usize = @intCast(rand(&seed) % nslots); + var new_obj: ?*Object = null; + + // Every so often publish null instead to exercise empty slots. + if (rand(&seed) % 8 != 0) { + if (zone.alloc(.{})) |buf| { + const obj: *Object = @ptrCast(@alignCast(buf)); + const v = next_val.fetchAdd(1, .monotonic); + + // Fill word by word with pauses so a reader still holding + // this memory has a wide window to observe a torn state. + for (&obj.vals) |*w| { + @atomicStore(u64, w, v, .monotonic); + std.atomic.spinLoopHint(); + } + + new_obj = obj; + } else |_| {} + } + + const old = slots[idx].swap(new_obj, .acq_rel); + + if (old) |o| { + zone.free(o); + } + + _ = writes.fetchAdd(1, .monotonic); + + if (iter % 256 == 0) ke.sched.yield(); + } +} + +fn poller(_: ?*anyopaque) void { + var iter: u64 = 0; + + while (true) : (iter += 1) { + if (iter % 2 == 0) { + _ = ke.smr.poll(dom, ke.smr.advance(dom), true); + } else { + const goal = ke.smr.deferred_advance(dom); + + if (!ke.smr.poll(dom, goal, false)) { + ke.smr.deferred_advance_commit(dom, goal); + _ = ke.smr.poll(dom, goal, true); + } + } + + _ = polls.fetchAdd(1, .monotonic); + + ke.sched.yield(); + } +} + +fn spawn(entry: *const fn (?*anyopaque) void, id: usize) void { + const t = ps.thread.create_kernel(ke.Thread.Priority.default, entry, @ptrFromInt(id + 1)) catch @panic("oom"); + ke.sched.enqueue(&t.kern); +} + +pub fn start(_: ?*anyopaque) void { + dom = mm.zone.smr_domain_create(false) catch @panic("failed to create SMR domain"); + zone.init("smr-stress", @sizeOf(Object), .{ + .alignment = @alignOf(Object), + .smr = dom, + }); + + const nreaders = ke.ncpus; + const nwriters = @max(1, ke.ncpus / 2); + + for (0..nreaders) |i| spawn(&reader, i); + for (0..nwriters) |i| spawn(&writer, nreaders + i); + spawn(&poller, nreaders + nwriters); + + var last: usize = 0; + while (true) { + delay_ms(1000); + + std.log.info("smr: reads: {}, writes: {}, polls: {}, read_seq: {}, write_seq: {} scans/s={}", .{ + reads.load(.monotonic), + writes.load(.monotonic), + polls.load(.monotonic), + dom.clock.read_seq.load(.monotonic), + dom.clock.write_seq.load(.monotonic), + ke.private.smr.full_scans.load(.monotonic) - last, + }); + + last = ke.private.smr.full_scans.load(.monotonic); + } +} diff --git a/src/tests/smr_pi.zig b/src/tests/smr_pi.zig new file mode 100644 index 0000000..799cf08 --- /dev/null +++ b/src/tests/smr_pi.zig @@ -0,0 +1,345 @@ +//! Preemptible SMR priority-inheritance test. +//! Same shape as the SMR stress test, but the read sections +//! are long enough to be preempted and the pollers run at a high priority so +//! they end up blocked on the stalled readers. +const std = @import("std"); +const r = @import("root"); +const ke = r.ke; +const mm = r.mm; +const ps = r.ps; +const rtl = @import("rtl"); + +const ki = ke.private; + +const prio_reader: u8 = 4; +const prio_writer: u8 = 6; +const prio_poller: u8 = 30; + +const nslots = 128; +const obj_words = 8; +const reads_per_section = 16; +const readers_per_cpu = 4; +const section_us = 3000; + +const Object = struct { + vals: [obj_words]u64, +}; + +var dom: *ke.smr.Domain = undefined; +var zone: mm.zone.Zone = undefined; + +var slots = [_]std.atomic.Value(?*Object){.init(null)} ** nslots; + +var next_val = std.atomic.Value(u64).init(1); +var reads = std.atomic.Value(u64).init(0); +var writes = std.atomic.Value(u64).init(0); +var polls = std.atomic.Value(u64).init(0); + +/// Number of pairs seen boosted. +var boosts_seen = std.atomic.Value(u64).init(0); +/// Largest number of readers ever seen stalled on a single CPU at once. +var max_co_stalled = std.atomic.Value(usize).init(0); + +var paused = std.atomic.Value(bool).init(false); +var parked = std.atomic.Value(usize).init(0); +var nworkers: usize = 0; + +var reader_tds: [64]*ps.Thread = undefined; +var nreaders: usize = 0; + +fn rand(state: *u64) u64 { + var x = state.*; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + state.* = x; + return x; +} + +fn seed_from(param: ?*anyopaque) u64 { + return (ke.time.read_time() ^ (@intFromPtr(param.?) *% 0x9e3779b97f4a7c15)) | 1; +} + +fn delay_ms(ms: usize) void { + var timer: ke.Timer = undefined; + timer.init(); + ke.timer.set(&timer, std.time.ns_per_ms * ms, null); + _ = ke.wait.wait_one(&timer.hdr, null) catch unreachable; +} + +/// Park here while the checker is verifying quiescence. Only ever called +/// outside a read section. +fn park_point() void { + if (!paused.load(.acquire)) return; + + _ = parked.fetchAdd(1, .acq_rel); + while (paused.load(.acquire)) delay_ms(1); + _ = parked.fetchSub(1, .acq_rel); +} + +fn spin_us(us: u64) void { + const began = ke.time.read_time(); + while (ke.time.read_time() - began < us * std.time.ns_per_us) { + std.atomic.spinLoopHint(); + } +} + +fn reader(param: ?*anyopaque) void { + var seed = seed_from(param); + + while (true) { + park_point(); + + var tracker: ki.smr.Tracker = undefined; + ki.smr.enter_preempt(dom, &tracker); + var done: u64 = 0; + + for (0..reads_per_section) |_| { + const idx: usize = @intCast(rand(&seed) % nslots); + const obj = slots[idx].load(.acquire) orelse continue; + + const v = @atomicLoad(u64, &obj.vals[0], .monotonic); + + for (1..obj_words) |i| { + if (@atomicLoad(u64, &obj.vals[i], .monotonic) != v) { + @panic("smr pi: torn read, object reused during read section"); + } + } + + done += 1; + } + + // Stay in the section well past a timeslice so we get preempted and + // land on the stalled list. A blocked poller must boost us out of here. + spin_us(section_us); + + ki.smr.exit_preempt(dom, &tracker); + + _ = reads.fetchAdd(done, .monotonic); + } +} + +fn writer(param: ?*anyopaque) void { + var seed = seed_from(param); + + while (true) { + park_point(); + + const idx: usize = @intCast(rand(&seed) % nslots); + var new_obj: ?*Object = null; + + if (rand(&seed) % 8 != 0) { + if (zone.alloc(.{})) |buf| { + const obj: *Object = @ptrCast(@alignCast(buf)); + const v = next_val.fetchAdd(1, .monotonic); + + for (&obj.vals) |*w| { + @atomicStore(u64, w, v, .monotonic); + std.atomic.spinLoopHint(); + } + + new_obj = obj; + } else |_| {} + } + + const old = slots[idx].swap(new_obj, .acq_rel); + + if (old) |o| { + zone.free(o); + } + + _ = writes.fetchAdd(1, .monotonic); + + ke.sched.yield(); + } +} + +fn poller(_: ?*anyopaque) void { + while (true) { + park_point(); + + _ = ke.smr.poll(dom, ke.smr.advance(dom), true); + + _ = polls.fetchAdd(1, .monotonic); + + ke.sched.yield(); + } +} + +fn cur_prio(td: *ke.Thread) u8 { + return @atomicLoad(u8, &td.priority, .monotonic); +} + +fn cur_inherited(td: *ke.Thread) u8 { + return @atomicLoad(u8, &td.inherited_prio, .monotonic); +} + +/// Highest priority among the threads blocked on `ts`, or null if none. +fn blocked_prio(ts: *ki.turnstile.Turnstile) ?u8 { + var best: ?u8 = null; + + for (&ts.queues) |*q| { + var it = q.iterator(); + while (it.next()) : (it.advance()) { + const w: *ki.turnstile.Waiter = @fieldParentPtr("link", it.get()); + best = @max(best orelse 0, cur_prio(w.thread)); + } + } + + return best; +} + +/// Walk one domain CPU and check every stalled reader against the priority of +/// whoever is blocked waiting for them. +fn check_cpu(cpu: *ke.smr.Cpu) void { + const ipl = ke.ipl.raise(.Dispatch); + cpu.stall_lock.acquire_no_ipl(); + + const turnstile = ki.turnstile.lookup(cpu); + + if (turnstile) |ts| { + if (blocked_prio(ts)) |want| { + var stalled: usize = 0; + var it = cpu.stalled.iterator(); + + while (it.next()) : (it.advance()) { + const owner: *ki.turnstile.Owner = @fieldParentPtr("link", it.get()); + const tracker: *ki.smr.Tracker = @fieldParentPtr("owner", owner); + const td = tracker.owner.thread; + stalled += 1; + + if (cur_inherited(td) < want or cur_prio(td) < want) { + std.log.err( + "smr pi: stalled reader {} of this cpu: thread {*} prio {}, inherited {}, want {}", + .{ stalled, td, cur_prio(td), cur_inherited(td), want }, + ); + @panic("smr pi: stalled reader not boosted to waiter priority"); + } + } + + _ = boosts_seen.fetchAdd(stalled, .monotonic); + + var max = max_co_stalled.load(.monotonic); + while (stalled > max) { + max = max_co_stalled.cmpxchgWeak(max, stalled, .monotonic, .monotonic) orelse break; + } + } + } + + ki.turnstile.exit(cpu); + cpu.stall_lock.release_no_ipl(); + ke.ipl.lower(ipl); +} + +fn check_boosts() void { + for (0..ke.ncpus) |i| check_cpu(&dom.cpus[i]); +} + +/// Park every worker and verify that no donation survived. +fn quiesce_check() void { + paused.store(true, .release); + + var spins: usize = 0; + while (parked.load(.acquire) != nworkers) : (spins += 1) { + if (spins > 10_000) @panic("smr pi: workers failed to park"); + delay_ms(1); + } + + for (0..ke.ncpus) |i| { + const cpu = &dom.cpus[i]; + const ipl = ke.ipl.raise(.Dispatch); + cpu.stall_lock.acquire_no_ipl(); + + if (!cpu.stalled.is_empty()) @panic("smr pi: stalled readers left at quiescence"); + if (cpu.stall_seq.load(.monotonic) != ke.smr.seq_invalid) { + @panic("smr pi: stall sequence not reset at quiescence"); + } + + cpu.stall_lock.release_no_ipl(); + ke.ipl.lower(ipl); + } + + for (0..nreaders) |i| { + const td = &reader_tds[i].kern; + const base = @atomicLoad(u8, &td.base_priority, .monotonic); + + if (cur_inherited(td) != 0 or cur_prio(td) != base) { + std.log.err("smr pi: reader {}: prio {}, inherited {}, base {}", .{ + i, + cur_prio(td), + cur_inherited(td), + base, + }); + @panic("smr pi: boost not revoked at quiescence"); + } + } + + paused.store(false, .release); + while (parked.load(.acquire) != 0) delay_ms(1); +} + +fn spawn(prio: u8, entry: *const fn (?*anyopaque) void, id: usize) *ps.Thread { + const t = ps.thread.create_kernel(prio, entry, @ptrFromInt(id + 1)) catch @panic("oom"); + ke.sched.enqueue(&t.kern); + return t; +} + +pub fn start(_: ?*anyopaque) void { + std.log.info("{}", .{@sizeOf(ki.smr.Tracker)}); + + dom = mm.zone.smr_domain_create(true) catch @panic("failed to create SMR domain"); + zone.init("smr-pi", @sizeOf(Object), .{ + .alignment = @alignOf(Object), + .smr = dom, + }); + + nreaders = @min(ke.ncpus * readers_per_cpu, reader_tds.len); + const nwriters = @max(1, ke.ncpus / 2); + const npollers = @max(1, ke.ncpus / 2); + + nworkers = nreaders + nwriters + npollers; + + var id: usize = 0; + + for (0..nreaders) |i| { + reader_tds[i] = spawn(prio_reader, &reader, id); + id += 1; + } + + for (0..nwriters) |_| { + _ = spawn(prio_writer, &writer, id); + id += 1; + } + + for (0..npollers) |_| { + _ = spawn(prio_poller, &poller, id); + id += 1; + } + + var round: usize = 0; + + while (true) : (round += 1) { + for (0..50) |_| { + check_boosts(); + delay_ms(15); + } + + quiesce_check(); + + const seen = boosts_seen.load(.monotonic); + const max_co = max_co_stalled.load(.monotonic); + + std.log.info("smr pi: reads: {}, writes: {}, polls: {}, boosts seen: {}, max co-stalled: {}", .{ + reads.load(.monotonic), + writes.load(.monotonic), + polls.load(.monotonic), + seen, + max_co, + }); + + if (round >= 2) { + if (seen == 0) @panic("smr pi: never observed a boosted stalled reader"); + if (max_co < 2) @panic("smr pi: never observed two readers stalled on one cpu"); + } + } +} diff --git a/src/tests/turnstile.zig b/src/tests/turnstile.zig new file mode 100644 index 0000000..987137a --- /dev/null +++ b/src/tests/turnstile.zig @@ -0,0 +1,230 @@ +//! Turnstile / priority inheritance test. +const std = @import("std"); +const r = @import("root"); +const ke = r.ke; +const ps = r.ps; + +const prio_low: u8 = 4; +const prio_mid: u8 = 16; +const prio_high: u8 = 30; + +const nmutexes = 8; +const max_stress = 32; + +var m1: ke.Mutex = undefined; +var m2: ke.Mutex = undefined; + +var low_td: *ps.Thread = undefined; +var mid_td: *ps.Thread = undefined; +var high_td: *ps.Thread = undefined; + +var low_locked = std.atomic.Value(bool).init(false); +var release_low = std.atomic.Value(bool).init(false); +var release_mid = std.atomic.Value(bool).init(false); +var finished = std.atomic.Value(bool).init(false); +var chain_done = std.atomic.Value(usize).init(0); + +var locks: [nmutexes]ke.Mutex = undefined; +var stress_tds: [max_stress]*ps.Thread = undefined; +var stress_prio: [max_stress]u8 = undefined; +var nstress: usize = 0; + +var paused = std.atomic.Value(bool).init(false); +var parked = std.atomic.Value(usize).init(0); +var acquisitions = std.atomic.Value(u64).init(0); + +fn rand(state: *u64) u64 { + var x = state.*; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + state.* = x; + return x; +} + +fn delay_ms(ms: usize) void { + var timer: ke.Timer = undefined; + timer.init(); + ke.timer.set(&timer, std.time.ns_per_ms * ms, null); + _ = ke.wait.wait_one(&timer.hdr, null) catch unreachable; +} + +fn wait_flag(flag: *std.atomic.Value(bool)) void { + while (!flag.load(.acquire)) delay_ms(1); +} + +fn cur_prio(td: *ps.Thread) u8 { + return @atomicLoad(u8, &td.kern.priority, .monotonic); +} + +fn cur_inherited(td: *ps.Thread) u8 { + return @atomicLoad(u8, &td.kern.inherited_prio, .monotonic); +} + +/// Poll until `td` reaches priority `want`, panic on timeout. +fn expect_prio(td: *ps.Thread, want: u8, what: []const u8) void { + for (0..2000) |_| { + if (cur_prio(td) == want) return; + delay_ms(1); + } + + std.log.err("pi: {s}: want priority {}, got {}", .{ what, want, cur_prio(td) }); + @panic("pi: priority check failed"); +} + +fn chain_low(_: ?*anyopaque) void { + m1.acquire(); + low_locked.store(true, .release); + wait_flag(&release_low); + m1.release(); + + _ = chain_done.fetchAdd(1, .release); + wait_flag(&finished); + ps.thread.exit(); +} + +fn chain_mid(_: ?*anyopaque) void { + m2.acquire(); + m1.acquire(); + m1.release(); + wait_flag(&release_mid); + m2.release(); + + _ = chain_done.fetchAdd(1, .release); + wait_flag(&finished); + ps.thread.exit(); +} + +fn chain_high(_: ?*anyopaque) void { + m2.acquire(); + m2.release(); + + _ = chain_done.fetchAdd(1, .release); + wait_flag(&finished); + ps.thread.exit(); +} + +fn spawn(prio: u8, entry: *const fn (?*anyopaque) void, arg: ?*anyopaque) *ps.Thread { + const t = ps.thread.create_kernel(prio, entry, arg) catch @panic("oom"); + ke.sched.enqueue(&t.kern); + return t; +} + +fn chain_test() void { + m1 = .init(); + m2 = .init(); + + low_td = spawn(prio_low, &chain_low, null); + wait_flag(&low_locked); + + // Mid takes M2 then blocks on M1, donating to low. + mid_td = spawn(prio_mid, &chain_mid, null); + expect_prio(low_td, prio_mid, "low boosted by mid"); + + // High blocks on M2; the donation must propagate through mid down to low. + high_td = spawn(prio_high, &chain_high, null); + expect_prio(mid_td, prio_high, "mid boosted by high"); + expect_prio(low_td, prio_high, "boost propagated to low"); + + release_low.store(true, .release); + expect_prio(low_td, prio_low, "low boost revoked after release"); + expect_prio(mid_td, prio_high, "mid still boosted while high waits"); + + release_mid.store(true, .release); + expect_prio(mid_td, prio_mid, "mid boost revoked after release"); + + while (chain_done.load(.acquire) != 3) delay_ms(1); + + for ([_]*ps.Thread{ low_td, mid_td, high_td }) |td| { + if (cur_inherited(td) != 0) @panic("pi: leftover inherited priority after chain test"); + } + + std.log.info("turnstile: chain test passed", .{}); + finished.store(true, .release); +} + +fn stress(param: ?*anyopaque) void { + const id = @intFromPtr(param.?) - 1; + var seed = (ke.time.read_time() ^ (id *% 0x9e3779b97f4a7c15)) | 1; + var iter: u64 = 0; + + while (true) : (iter += 1) { + if (paused.load(.acquire)) { + _ = parked.fetchAdd(1, .acq_rel); + while (paused.load(.acquire)) delay_ms(1); + _ = parked.fetchSub(1, .acq_rel); + } + + const first: usize = @intCast(rand(&seed) % nmutexes); + const last = @min(first + 1 + rand(&seed) % 3, nmutexes); + + for (first..last) |i| locks[i].acquire(); + + if (rand(&seed) % 16 == 0) { + delay_ms(1); + } else { + for (0..rand(&seed) % 512) |_| std.atomic.spinLoopHint(); + } + + var i = last; + while (i > first) : (i -= 1) locks[i - 1].release(); + + if (cur_prio(stress_tds[id]) < stress_prio[id]) { + @panic("pi: priority dropped below base"); + } + + _ = acquisitions.fetchAdd(1, .monotonic); + + if (iter % 64 == 0) delay_ms(1); + } +} + +/// Park all threads, then +/// verify every donation has been revoked. +fn quiesce_check() void { + paused.store(true, .release); + + var spins: usize = 0; + while (parked.load(.acquire) != nstress) : (spins += 1) { + if (spins > 10_000) @panic("pi: stress threads failed to park"); + delay_ms(1); + } + + for (0..nstress) |i| { + if (cur_prio(stress_tds[i]) != stress_prio[i] or cur_inherited(stress_tds[i]) != 0) { + std.log.err("pi: thread {}: base {}, prio {}, inherited {}", .{ + i, + stress_prio[i], + cur_prio(stress_tds[i]), + cur_inherited(stress_tds[i]), + }); + @panic("pi: boost not revoked at quiescence"); + } + } + + paused.store(false, .release); + while (parked.load(.acquire) != 0) delay_ms(1); +} + +pub fn start(_: ?*anyopaque) void { + chain_test(); + + for (&locks) |*l| l.* = .init(); + + nstress = @max(4, @min(ke.ncpus * 2, max_stress)); + + for (0..nstress) |i| { + const prio: u8 = @intCast(2 + (i * 5) % 29); + stress_prio[i] = prio; + + const t = ps.thread.create_kernel(prio, &stress, @ptrFromInt(i + 1)) catch @panic("oom"); + stress_tds[i] = t; + ke.sched.enqueue(&t.kern); + } + + while (true) { + delay_ms(2000); + quiesce_check(); + std.log.info("turnstile: acquisitions: {}, quiescence OK", .{acquisitions.load(.monotonic)}); + } +}