danos/system/services/thread-test/thread-test.zig

//! thread-test — danos's multi-threaded exerciser (docs/threading-plan.md M2, M3).
//!
//! Two modes, chosen by argv[1] (default "spawn"):
//!   spawn — M2: one worker writes a shared global; the main thread observes it, proving
//!           `runtime.Thread.spawn` started a task in the **same** address space.
//!   join  — M3: N workers each do K atomic increments on a shared counter and stamp the
//!           core they ran on; the main thread `join`s all N and checks the total is
//!           exactly N*K (every worker ran, join waited) and that >1 core was used
//!           (genuine parallelism). Then a detached worker proves `detach` runs and
//!           needs no join.
//!
//! Built multi-threaded (`addThreadedUserBinary`) so atomics/shared reads are real.

const std = @import("std");
const runtime = @import("runtime");

fn write(comptime s: []const u8) void {
    _ = runtime.system.write(s);
}

// --- M2: spawn mode ---------------------------------------------------------

var shared_value: u32 = 0;
var spawn_done = std.atomic.Value(u32).init(0);
const sentinel: u32 = 0xA5A5;

fn spawnWorker() void {
    shared_value = sentinel;
    spawn_done.store(1, .release);
}

fn runSpawnMode() void {
    write("thread-test: starting\n");
    _ = runtime.Thread.spawn(.{}, spawnWorker, .{}) catch {
        write("thread-test: FAIL spawn refused\n");
        return;
    };
    var spins: usize = 0;
    while (spawn_done.load(.acquire) == 0 and spins < 50_000_000) : (spins += 1) {
        runtime.system.yield();
    }
    if (spawn_done.load(.acquire) == 1 and shared_value == sentinel) {
        write("thread-test: child ran in shared aspace ok\n");
    } else {
        write("thread-test: FAIL worker did not update shared memory\n");
    }
}

// --- M3: join mode ----------------------------------------------------------

const worker_count: u32 = 4;
const iterations: u64 = 100_000;

var counter = std.atomic.Value(u64).init(0);
var cores_seen = std.atomic.Value(u32).init(0);

fn joinWorker() void {
    var i: u64 = 0;
    while (i < iterations) : (i += 1) {
        _ = counter.fetchAdd(1, .monotonic);
        if (i % 1000 == 0) stampCore(); // periodic: catches cross-core migration too
    }
    stampCore();
}

fn stampCore() void {
    const core = runtime.Thread.currentCore();
    if (core < 32) _ = cores_seen.fetchOr(@as(u32, 1) << @intCast(core), .monotonic);
}

var detach_done = std.atomic.Value(u32).init(0);

fn detachWorker() void {
    detach_done.store(1, .release);
}

fn runJoinMode() void {
    write("thread-test: join mode starting\n");

    var threads: [worker_count]runtime.Thread = undefined;
    var spawned: u32 = 0;
    while (spawned < worker_count) : (spawned += 1) {
        threads[spawned] = runtime.Thread.spawn(.{}, joinWorker, .{}) catch break;
    }
    if (spawned != worker_count) {
        write("thread-test: FAIL could not spawn all workers\n");
        return;
    }
    for (threads[0..spawned]) |t| t.join();

    const total = counter.load(.acquire);
    const cores = @popCount(cores_seen.load(.acquire));
    if (total != worker_count * iterations) {
        write("thread-test: FAIL counter mismatch (a worker was lost or join did not wait)\n");
        return;
    }
    if (cores <= 1) {
        write("thread-test: FAIL workers never ran on more than one core\n");
        return;
    }

    // detach: the worker runs and we never join it.
    const dt = runtime.Thread.spawn(.{}, detachWorker, .{}) catch {
        write("thread-test: FAIL detach spawn refused\n");
        return;
    };
    dt.detach();
    var spins: usize = 0;
    while (detach_done.load(.acquire) == 0 and spins < 50_000_000) : (spins += 1) {
        runtime.system.yield();
    }
    if (detach_done.load(.acquire) != 1) {
        write("thread-test: FAIL detached worker did not run\n");
        return;
    }

    write("thread-test: join ok\n"); // the M3 verdict marker
}

// --- M4: futex mode ---------------------------------------------------------

const Futex = runtime.Thread.Futex;

var futex_word = std.atomic.Value(u32).init(0);
var waiter_parked = std.atomic.Value(u32).init(0);

fn futexWaiter() void {
    write("thread-futex: waiting\n");
    waiter_parked.store(1, .release);
    // Block while the word is still 0; the waker sets it to 1 and wakes us.
    while (futex_word.load(.acquire) == 0) {
        Futex.wait(&futex_word, 0);
    }
    write("thread-futex: woke\n");
}

fn runFutexMode() void {
    write("thread-futex: starting\n");

    const waiter = runtime.Thread.spawn(.{}, futexWaiter, .{}) catch {
        write("thread-futex: FAIL spawn refused\n");
        return;
    };
    // Let the waiter reach its wait, then give it a beat to actually park in-kernel.
    var spins: usize = 0;
    while (waiter_parked.load(.acquire) == 0 and spins < 50_000_000) : (spins += 1) {
        runtime.system.yield();
    }
    runtime.system.sleep(50);

    // The handshake: publish the value, then wake the parked waiter.
    futex_word.store(1, .release);
    write("thread-futex: waking\n");
    Futex.wake(&futex_word, 1);

    waiter.join(); // returns once the waiter woke and printed "woke"

    // Timeout: nobody ever wakes this word, so timedWait must report a timeout.
    var lonely = std.atomic.Value(u32).init(0);
    if (Futex.timedWait(&lonely, 0, 100_000_000)) |_| {
        write("thread-futex: FAIL timedWait did not time out\n");
        return;
    } else |_| {}
    write("thread-futex: timeout ok\n");

    write("thread-futex: ok\n"); // the M4 verdict marker
}

pub fn main(init: runtime.process.Init) void {
    const mode = init.arguments.get(1) orelse "spawn";
    if (std.mem.eql(u8, mode, "join")) {
        runJoinMode();
    } else if (std.mem.eql(u8, mode, "futex")) {
        runFutexMode();
    } else {
        runSpawnMode();
    }
}