mirror of
https://github.com/rdmsr/zag.git
synced 2026-10-27 13:58:20 +00:00
tests: add tests
This commit is contained in:
Loïc Grégoire
+12
-3
@@ -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) {
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
pub const fireworks = @import("fireworks.zig");
|
||||
pub const private = @import("private.zig");
|
||||
|
||||
const p = private;
|
||||
|
||||
@@ -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");
|
||||
|
||||
|
||||
@@ -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});
|
||||
}
|
||||
}
|
||||
@@ -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 },
|
||||
});
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user