tiny.coz.delay

1const std = @import("std");23pub const ThreadState = struct {4    in_use: std.atomic.Value(bool) = .init(false),5    local_delay_ns: std.atomic.Value(u64) = .init(0),6    pre_block_time_ns: u64 = 0,7    is_blocked: std.atomic.Value(bool) = .init(false),89    pub fn setInUse(self: *ThreadState, value: bool) void {10        self.in_use.store(value, .seq_cst);11    }1213    pub fn acquireUse(self: *ThreadState) bool {14        return self.in_use.cmpxchgStrong(false, true, .seq_cst, .seq_cst) == null;15    }1617    pub fn checkInUse(self: *const ThreadState) bool {18        return self.in_use.load(.seq_cst);19    }2021    pub fn localDelay(self: *const ThreadState) u64 {22        return self.local_delay_ns.load(.monotonic);23    }24};2526pub const Coordinator = struct {27    experiment_active: std.atomic.Value(bool) = .init(false),28    global_delay_ns: std.atomic.Value(u64) = .init(0),29    delay_size_ns: std.atomic.Value(u64) = .init(0),30    overshoot_ns: std.atomic.Value(u64) = .init(0),31    capped_waits: bool = false,3233    pub fn startExperiment(self: *Coordinator, delay_size_ns: u64) void {34        self.delay_size_ns.store(delay_size_ns, .monotonic);35        self.overshoot_ns.store(0, .monotonic);36        self.experiment_active.store(true, .release);37    }3839    pub fn finishExperiment(self: *Coordinator) void {40        self.experiment_active.store(false, .release);41    }4243    pub fn active(self: *const Coordinator) bool {44        return self.experiment_active.load(.acquire);45    }4647    pub fn globalDelay(self: *const Coordinator) u64 {48        return self.global_delay_ns.load(.monotonic);49    }5051    pub fn delaySize(self: *const Coordinator) u64 {52        return self.delay_size_ns.load(.monotonic);53    }5455    pub fn overshoot(self: *const Coordinator) u64 {56        return self.overshoot_ns.load(.monotonic);57    }5859    pub fn preBlock(self: *Coordinator, thread: *ThreadState) void {60        thread.is_blocked.store(true, .release);61        thread.pre_block_time_ns = self.globalDelay();62    }6364    pub fn postBlock(self: *Coordinator, thread: *ThreadState, skip_delays: bool) void {65        thread.setInUse(true);66        defer thread.setInUse(false);6768        if (skip_delays) {69            const delta = self.globalDelay() -| thread.pre_block_time_ns;70            _ = thread.local_delay_ns.fetchAdd(delta, .monotonic);71        }72        thread.is_blocked.store(false, .release);73    }7475    pub fn creditSelectedHit(self: *Coordinator, thread: *ThreadState) void {76        if (!self.active()) return;77        _ = thread.local_delay_ns.fetchAdd(self.delaySize(), .monotonic);78    }7980    pub fn creditSelectedThreadAndPush(self: *Coordinator, thread: *ThreadState) u64 {81        if (!self.active()) return self.globalDelay();8283        const delay_size = self.delaySize();84        const new_global = self.global_delay_ns.fetchAdd(delay_size, .monotonic) + delay_size;85        var local = thread.localDelay();86        while (local < new_global) {87            if (thread.local_delay_ns.cmpxchgWeak(local, new_global, .monotonic, .monotonic) == null) break;88            local = thread.localDelay();89        }90        return new_global;91    }9293    pub fn addDelays(self: *Coordinator, thread: *ThreadState, wait: anytype) u64 {94        if (self.active()) {95            if (thread.is_blocked.load(.acquire)) return 0;9697            const global = self.globalDelay();98            const local = thread.localDelay();99            if (local > global) {100                _ = self.global_delay_ns.fetchAdd(local - global, .monotonic);101                return 0;102            }103            if (local < global) {104                const needed = global - local;105                const waited = callWait(wait, needed);106                if (self.capped_waits) {107                    _ = thread.local_delay_ns.fetchAdd(needed, .monotonic);108                    if (waited > needed) _ = self.overshoot_ns.fetchAdd(waited - needed, .monotonic);109                } else {110                    _ = thread.local_delay_ns.fetchAdd(waited, .monotonic);111                }112                return waited;113            }114            return 0;115        }116117        thread.local_delay_ns.store(self.globalDelay(), .monotonic);118        return 0;119    }120121    pub fn inheritDelay(_: *Coordinator, child: *ThreadState, parent_delay_ns: u64) void {122        child.local_delay_ns.store(parent_delay_ns, .monotonic);123    }124};125126fn callWait(wait: anytype, ns: u64) u64 {127    switch (@typeInfo(@TypeOf(wait))) {128        .@"struct", .@"enum", .@"union", .@"opaque" => if (comptime @hasDecl(@TypeOf(wait), "wait")) return wait.wait(ns),129        else => {},130    }131    return wait(ns);132}133134fn exactWait(ns: u64) u64 {135    return ns;136}137138fn overshootingWait(ns: u64) u64 {139    return ns + 3;140}141142const CountingWait = struct {143    calls: *u32,144145    pub fn wait(self: CountingWait, ns: u64) u64 {146        self.calls.* += 1;147        return ns;148    }149};150151test "thread state publishes in-use changes" {152    var thread: ThreadState = .{};153154    try std.testing.expect(!thread.checkInUse());155    try std.testing.expect(thread.acquireUse());156    try std.testing.expect(!thread.acquireUse());157    thread.setInUse(true);158    try std.testing.expect(thread.checkInUse());159    thread.setInUse(false);160    try std.testing.expect(!thread.checkInUse());161}162163test "selected hit advances sampled thread then pushes global delay" {164    var coordinator: Coordinator = .{};165    var sampled: ThreadState = .{};166    var other: ThreadState = .{};167168    coordinator.startExperiment(10);169    coordinator.creditSelectedHit(&sampled);170    try std.testing.expectEqual(@as(u64, 10), sampled.localDelay());171172    try std.testing.expectEqual(@as(u64, 0), coordinator.addDelays(&sampled, exactWait));173    try std.testing.expectEqual(@as(u64, 10), coordinator.globalDelay());174175    try std.testing.expectEqual(@as(u64, 10), coordinator.addDelays(&other, exactWait));176    try std.testing.expectEqual(@as(u64, 10), other.localDelay());177}178179test "inactive coordinator catches thread local delay up without waiting" {180    var coordinator: Coordinator = .{};181    var thread: ThreadState = .{};182    coordinator.global_delay_ns.store(25, .monotonic);183184    try std.testing.expectEqual(@as(u64, 0), coordinator.addDelays(&thread, exactWait));185    try std.testing.expectEqual(@as(u64, 25), thread.localDelay());186}187188test "blocked threads do not wait until post block clears the flag" {189    var coordinator: Coordinator = .{};190    var thread: ThreadState = .{};191192    coordinator.startExperiment(0);193    coordinator.global_delay_ns.store(20, .monotonic);194    coordinator.preBlock(&thread);195196    try std.testing.expectEqual(@as(u64, 0), coordinator.addDelays(&thread, exactWait));197    try std.testing.expectEqual(@as(u64, 0), thread.localDelay());198199    coordinator.postBlock(&thread, false);200    try std.testing.expectEqual(@as(u64, 20), coordinator.addDelays(&thread, exactWait));201    try std.testing.expectEqual(@as(u64, 20), thread.localDelay());202}203204test "post block can skip delays inserted while blocked" {205    var coordinator: Coordinator = .{};206    var thread: ThreadState = .{};207208    coordinator.startExperiment(0);209    coordinator.global_delay_ns.store(10, .monotonic);210    coordinator.preBlock(&thread);211    coordinator.global_delay_ns.store(35, .monotonic);212    coordinator.postBlock(&thread, true);213214    try std.testing.expectEqual(@as(u64, 25), thread.localDelay());215    try std.testing.expectEqual(@as(u64, 10), thread.pre_block_time_ns);216    try std.testing.expectEqual(@as(u64, 10), coordinator.addDelays(&thread, exactWait));217    try std.testing.expectEqual(@as(u64, 35), thread.localDelay());218}219220test "selected sampled thread credit pushes global delay without delaying sampled thread" {221    var coordinator: Coordinator = .{};222    var sampled: ThreadState = .{};223    var other: ThreadState = .{};224225    coordinator.startExperiment(12);226    try std.testing.expectEqual(@as(u64, 12), coordinator.creditSelectedThreadAndPush(&sampled));227    try std.testing.expectEqual(@as(u64, 12), sampled.localDelay());228    try std.testing.expectEqual(@as(u64, 12), coordinator.globalDelay());229    try std.testing.expectEqual(@as(u64, 12), coordinator.addDelays(&other, exactWait));230}231232test "capped waits charge requested delay and record overshoot separately" {233    var coordinator: Coordinator = .{ .capped_waits = true };234    var thread: ThreadState = .{};235236    coordinator.startExperiment(0);237    coordinator.global_delay_ns.store(10, .monotonic);238239    try std.testing.expectEqual(@as(u64, 13), coordinator.addDelays(&thread, overshootingWait));240    try std.testing.expectEqual(@as(u64, 10), thread.localDelay());241    try std.testing.expectEqual(@as(u64, 3), coordinator.overshoot());242}243244test "delay coordinator accepts wait objects" {245    var coordinator: Coordinator = .{};246    var thread: ThreadState = .{};247    var calls: u32 = 0;248249    coordinator.startExperiment(0);250    coordinator.global_delay_ns.store(11, .monotonic);251252    try std.testing.expectEqual(@as(u64, 11), coordinator.addDelays(&thread, CountingWait{ .calls = &calls }));253    try std.testing.expectEqual(@as(u32, 1), calls);254    try std.testing.expectEqual(@as(u64, 11), thread.localDelay());255}256257test "child threads inherit parent delay credit" {258    var coordinator: Coordinator = .{};259    var child: ThreadState = .{};260261    coordinator.inheritDelay(&child, 17);262263    try std.testing.expectEqual(@as(u64, 17), child.localDelay());264}

1pub const delay = @import("delay.zig");