diff --git a/zig/src/core/concurrency.zig b/zig/src/core/concurrency.zig index af7a9e1a..1332d982 100644 --- a/zig/src/core/concurrency.zig +++ b/zig/src/core/concurrency.zig @@ -26,6 +26,8 @@ pub const Mutex = switch (builtin.os.tag) { /// /// Waiters must always check their predicate in a loop because a broadcast /// does not imply that the predicate remains true after the mutex is reacquired. +/// `waitUntil` additionally accepts a suspend-inclusive deadline and may return +/// either because it was signalled or because that deadline may have elapsed. pub const Condition = switch (builtin.os.tag) { .windows => WindowsCondition, else => PosixCondition, @@ -135,8 +137,8 @@ pub const RunAfter = struct { /// Callback to notify when this handle reaches a terminal outcome. callback: ?Scheduled.Callback = null, - /// Approximate delay remaining in scheduler ticks. - remaining_ms: u64 = 0, + /// Absolute suspend-inclusive deadline for this callback. + deadline_ns: u64 = 0, /// Private intrusive linkage used by the executor's pending list. next: ?*Scheduled = null, @@ -191,6 +193,13 @@ pub const RunAfter = struct { /// Delay associated with the replaceable callback slot. delay_ms: u64 = 0, + /// Absolute suspend-inclusive deadline for the replaceable callback. + /// + /// A successor scheduled from a running callback leaves this unset until + /// that callback returns, so its delay begins when the worker can service + /// it. + deadline_ns: ?u64 = null, + /// Replaceable callback waiting in, or executing from, the callback slot. callback: ?Callback = null, @@ -247,11 +256,10 @@ pub const RunAfter = struct { /// Appends an independent one-shot callback without replacing earlier ones. /// - /// Delays are best-effort and quantized to `scheduler_resolution_ms`; a - /// callback may run late but is given a full extra tick so it does not run - /// early at a tick boundary. This mode must not be started while a callback - /// submitted with `scheduleReplacing` is pending. The function starts the - /// worker lazily and reports failure to spawn it. + /// The worker waits directly for the earliest pending deadline and is + /// interrupted when an earlier callback is appended. This mode must not be + /// started while a callback submitted with `scheduleReplacing` is pending. + /// The function starts the worker lazily and reports failure to spawn it. pub fn scheduleAppending( self: *RunAfter, scheduled: *Scheduled, @@ -307,9 +315,7 @@ pub const RunAfter = struct { scheduled.* = .{ .context = context, .callback = callback, - // A full extra tick prevents a newly appended job from expiring - // early when it lands immediately before the current tick. - .remaining_ms = delay_ms +| scheduler_resolution_ms, + .deadline_ns = deadlineAfterMs(monotonicNs(), delay_ms), }; if (self.scheduled_tail) |tail| { tail.next = scheduled; @@ -320,7 +326,12 @@ pub const RunAfter = struct { if (!self.scheduling) { self.scheduling = true; - self.replaceLocked(scheduler_resolution_ms, runScheduled, self); + self.replaceAtLocked(scheduled.deadline_ns, runScheduled, self); + } else if ((self.state == .scheduled or + self.state == .running_scheduled) and + scheduled.deadline_ns < self.deadline_ns.?) + { + self.replaceAtLocked(scheduled.deadline_ns, runScheduled, self); } } @@ -349,6 +360,7 @@ pub const RunAfter = struct { const cancelled = self.takeScheduledLocked(); self.scheduling = false; self.generation +%= 1; + self.deadline_ns = null; self.callback = null; self.callback_ctx = null; self.state = .stopping; @@ -365,9 +377,10 @@ pub const RunAfter = struct { /// Blocks until no replaceable callback is pending or running. /// - /// In independent scheduling mode, the internal tick callback keeps the - /// executor non-idle until every scheduled handle has been notified. This - /// function must not race with `deinit` or run from an executor callback. + /// In independent scheduling mode, the internal scheduler callback keeps + /// the executor non-idle until every scheduled handle has been notified. + /// This function must not race with `deinit` or run from an executor + /// callback. pub fn wait(self: *RunAfter) void { self.mutex.lock(); defer self.mutex.unlock(); @@ -379,7 +392,7 @@ pub const RunAfter = struct { /// Worker entry point for the replaceable callback slot. /// - /// The worker sleeps while idle, evaluates delays without holding `mutex`, + /// The worker sleeps while idle, waits interruptibly for absolute deadlines, /// and serializes callback execution. It exits only after entering /// `.stopping`. fn run(self: *RunAfter) void { @@ -394,12 +407,19 @@ pub const RunAfter = struct { } std.debug.assert(self.state == .scheduled); const generation = self.generation; - const delay_ms = self.delay_ms; - self.mutex.unlock(); - - if (self.sleepUntilChangedOrElapsed(generation, delay_ms)) continue; + const deadline_ns = self.deadline_ns orelse deadline: { + const value = deadlineAfterMs(monotonicNs(), self.delay_ms); + self.deadline_ns = value; + break :deadline value; + }; + + while (self.state == .scheduled and + self.generation == generation and + monotonicNs() < deadline_ns) + { + self.cond.waitUntil(&self.mutex, deadline_ns); + } - self.mutex.lock(); if (self.state == .stopping) { self.mutex.unlock(); return; @@ -419,6 +439,7 @@ pub const RunAfter = struct { self.mutex.lock(); switch (self.state) { .running => { + self.deadline_ns = null; self.callback = null; self.callback_ctx = null; self.state = .idle; @@ -435,10 +456,10 @@ pub const RunAfter = struct { } } - /// Processes one scheduler tick for the independent callback list. + /// Processes callbacks at the earliest independent deadline. /// - /// Due handles are detached while holding `mutex`, the next tick is armed - /// if necessary, and notifications are delivered after unlocking. + /// Due handles are detached while holding `mutex`, the next exact deadline + /// is armed if necessary, and notifications are delivered after unlocking. fn runScheduled(context: ?*anyopaque) void { const self: *RunAfter = @ptrCast(@alignCast(context.?)); @@ -452,10 +473,11 @@ pub const RunAfter = struct { var due_tail: ?*Scheduled = null; var previous: ?*Scheduled = null; var current = self.scheduled_head; + var next_deadline_ns: ?u64 = null; + const now_ns = monotonicNs(); while (current) |scheduled| { const next = scheduled.next; - scheduled.remaining_ms -|= scheduler_resolution_ms; - if (scheduled.remaining_ms == 0) { + if (now_ns >= scheduled.deadline_ns) { if (previous) |before| { before.next = next; } else { @@ -472,12 +494,17 @@ pub const RunAfter = struct { due_tail = scheduled; } else { previous = scheduled; + if (next_deadline_ns == null or + scheduled.deadline_ns < next_deadline_ns.?) + { + next_deadline_ns = scheduled.deadline_ns; + } } current = next; } - if (self.scheduled_head != null) { - self.replaceLocked(scheduler_resolution_ms, runScheduled, self); + if (next_deadline_ns) |deadline_ns| { + self.replaceAtLocked(deadline_ns, runScheduled, self); } else { self.scheduling = false; } @@ -496,8 +523,40 @@ pub const RunAfter = struct { callback: Callback, callback_ctx: ?*anyopaque, ) void { + const was_running = switch (self.state) { + .running, .running_scheduled => true, + .idle, .scheduled => false, + .stopping => @panic("cannot schedule a stopping RunAfter"), + }; self.generation +%= 1; self.delay_ms = delay_ms; + self.deadline_ns = if (was_running) + null + else + deadlineAfterMs(monotonicNs(), delay_ms); + self.callback = callback; + self.callback_ctx = callback_ctx; + self.state = switch (self.state) { + .idle, .scheduled => .scheduled, + .running, .running_scheduled => .running_scheduled, + .stopping => unreachable, + }; + self.cond.broadcast(); + } + + /// Arms an absolute deadline while `mutex` is held. + /// + /// This variant is used by independent callbacks, whose deadlines begin at + /// submission time even when the scheduler callback is currently running. + fn replaceAtLocked( + self: *RunAfter, + deadline_ns: u64, + callback: Callback, + callback_ctx: ?*anyopaque, + ) void { + self.generation +%= 1; + self.delay_ms = 0; + self.deadline_ns = deadline_ns; self.callback = callback; self.callback_ctx = callback_ctx; self.state = switch (self.state) { @@ -523,6 +582,7 @@ pub const RunAfter = struct { const cancelled = self.takeScheduledLocked(); self.scheduling = false; self.generation +%= 1; + self.deadline_ns = null; self.callback = null; self.callback_ctx = null; self.state = switch (self.state) { @@ -555,36 +615,6 @@ pub const RunAfter = struct { current = next; } } - - /// Sleeps for `delay_ms` unless the observed callback slot changes. - /// - /// Returns `true` when the generation or state changed and the caller must - /// discard the elapsed-delay result. Polling bounds rescheduling latency to - /// roughly `sleep_step_ms` without requiring timed condition waits. - fn sleepUntilChangedOrElapsed(self: *RunAfter, generation: u64, delay_ms: u64) bool { - // Maximum sleep interval between generation checks. - const sleep_step_ms = 10; - var remaining_ms = delay_ms; - while (remaining_ms > 0) { - if (self.hasChanged(generation)) return true; - - const current_sleep_ms = @min(remaining_ms, sleep_step_ms); - sleepMs(current_sleep_ms); - remaining_ms -= current_sleep_ms; - } - return self.hasChanged(generation); - } - - /// Reports whether `generation` still identifies the scheduled slot. - fn hasChanged(self: *RunAfter, generation: u64) bool { - self.mutex.lock(); - defer self.mutex.unlock(); - - return self.state != .scheduled or self.generation != generation; - } - - /// Tick resolution used for independent scheduled callbacks. - const scheduler_resolution_ms = 10; }; /// POSIX implementation of `Mutex` using `pthread_mutex_t`. @@ -639,6 +669,31 @@ const PosixCondition = struct { } } + /// Atomically releases `mutex` and waits for a signal or an absolute + /// suspend-inclusive deadline. + /// + /// POSIX condition variables use a real-time absolute timeout so the + /// kernel accounts for system suspend. The caller must recheck both its + /// predicate and the continuous clock after this function returns. + pub fn waitUntil( + self: *PosixCondition, + mutex: *Mutex, + deadline_ns: u64, + ) void { + const remaining_ns = nanosecondsUntil(deadline_ns, monotonicNs()); + if (remaining_ns == 0) return; + + var realtime: std.c.timespec = undefined; + if (std.c.clock_gettime(.REALTIME, &realtime) != 0) { + @panic("clock_gettime failed"); + } + const timeout = timespecAfterNs(realtime, remaining_ns); + switch (std.c.pthread_cond_timedwait(&self.raw, &mutex.raw, &timeout)) { + .SUCCESS, .TIMEDOUT => {}, + else => @panic("pthread_cond_timedwait failed"), + } + } + /// Wakes every thread currently waiting on this condition variable. /// /// Waiters still serialize on their associated mutex before returning. @@ -716,6 +771,38 @@ const WindowsCondition = struct { } } + /// Atomically releases `mutex` and waits for a signal or an absolute + /// suspend-inclusive deadline. + /// + /// A positive NT timeout is an absolute system-time deadline and therefore + /// expires across system suspend. The caller validates the corresponding + /// continuous deadline after this function returns. + pub fn waitUntil( + self: *WindowsCondition, + mutex: *Mutex, + deadline_ns: u64, + ) void { + const remaining_ns = nanosecondsUntil(deadline_ns, monotonicNs()); + if (remaining_ns == 0) return; + + const now_100ns: u64 = @intCast(windows.ntdll.RtlGetSystemTimePrecise()); + const remaining_100ns = (remaining_ns - 1) / 100 + 1; + const absolute_100ns = now_100ns +| remaining_100ns; + var timeout: windows.LARGE_INTEGER = @intCast(@min( + absolute_100ns, + @as(u64, std.math.maxInt(windows.LARGE_INTEGER)), + )); + switch (RtlSleepConditionVariableSRW( + &self.raw, + &mutex.raw, + &timeout, + 0, + )) { + .SUCCESS, .TIMEOUT => {}, + else => @panic("RtlSleepConditionVariableSRW failed"), + } + } + /// Wakes every thread currently waiting on this condition variable. pub fn broadcast(self: *WindowsCondition) void { windows.ntdll.RtlWakeAllConditionVariable(&self.raw); @@ -753,6 +840,11 @@ pub fn sleepMs(value: u64) void { } /// Returns nanoseconds from a process-independent monotonic clock. +/// +/// The clock includes time spent in system suspend wherever the host exposes +/// that distinction. Network deadlines therefore remain wall-time deadlines +/// across device sleep without being affected by changes to the real-time +/// clock. pub fn monotonicNs() u64 { if (builtin.os.tag == .windows) { const windows = std.os.windows; @@ -765,9 +857,78 @@ pub fn monotonicNs() u64 { return @intCast((@as(u128, ticks) * std.time.ns_per_s) / ticks_per_second); } + const clock_id = switch (builtin.os.tag) { + // CLOCK_BOOTTIME is monotonic and includes system suspend. + .linux => std.c.CLOCK.BOOTTIME, + + // Darwin's MONOTONIC_RAW is the continuous counterpart to + // UPTIME_RAW and includes time spent asleep. + .driverkit, + .ios, + .maccatalyst, + .macos, + .tvos, + .visionos, + .watchos, + => std.c.CLOCK.MONOTONIC_RAW, + + // These are the closest boot-relative clocks exposed by the hosts. + .dragonfly, .freebsd => std.c.CLOCK.MONOTONIC_FAST, + else => std.c.CLOCK.MONOTONIC, + }; var timestamp: std.c.timespec = undefined; - if (std.c.clock_gettime(.MONOTONIC, ×tamp) != 0) return 0; + if (std.c.clock_gettime(clock_id, ×tamp) != 0) return 0; const seconds: u64 = @intCast(timestamp.sec); const nanoseconds: u64 = @intCast(timestamp.nsec); return seconds *| @as(u64, std.time.ns_per_s) +| nanoseconds; } + +/// Computes a saturating absolute deadline from an integer millisecond delay. +fn deadlineAfterMs(now_ns: u64, delay_ms: u64) u64 { + return now_ns +| delay_ms *| @as(u64, std.time.ns_per_ms); +} + +/// Returns the nanoseconds remaining before an absolute deadline. +fn nanosecondsUntil(deadline_ns: u64, now_ns: u64) u64 { + if (now_ns >= deadline_ns) return 0; + return deadline_ns - now_ns; +} + +/// Returns the positive whole-millisecond sleep needed to reach `deadline_ns`. +/// +/// Rounding up prevents the scheduler from firing before a sub-millisecond +/// remainder elapses. A clock jump past the deadline returns zero immediately. +fn millisecondsUntil(deadline_ns: u64, now_ns: u64) u64 { + const remaining_ns = nanosecondsUntil(deadline_ns, now_ns); + if (remaining_ns == 0) return 0; + return (remaining_ns - 1) / std.time.ns_per_ms + 1; +} + +/// Adds a relative nanosecond duration to a POSIX timestamp with saturation. +fn timespecAfterNs(now: std.c.timespec, duration_ns: u64) std.c.timespec { + const Seconds = @TypeOf(now.sec); + const max_seconds: u64 = @intCast(std.math.maxInt(Seconds)); + const now_seconds: u64 = @intCast(now.sec); + var extra_seconds = duration_ns / std.time.ns_per_s; + var nanoseconds: u64 = @intCast(now.nsec); + nanoseconds += duration_ns % std.time.ns_per_s; + if (nanoseconds >= std.time.ns_per_s) { + nanoseconds -= std.time.ns_per_s; + extra_seconds +|= 1; + } + if (extra_seconds > max_seconds - now_seconds) { + return .{ + .sec = std.math.maxInt(Seconds), + .nsec = std.time.ns_per_s - 1, + }; + } + return .{ + .sec = @intCast(now_seconds + extra_seconds), + .nsec = @intCast(nanoseconds), + }; +} + +pub const testing = struct { + pub const deadlineAfterMilliseconds = deadlineAfterMs; + pub const millisecondsUntilDeadline = millisecondsUntil; +}; diff --git a/zig/tests/core/concurrency.zig b/zig/tests/core/concurrency.zig index 062a9e18..43d82ab4 100644 --- a/zig/tests/core/concurrency.zig +++ b/zig/tests/core/concurrency.zig @@ -14,6 +14,18 @@ fn waitUntil(value: *const AtomicBool) void { } } +fn waitUntilWithin(value: *const AtomicBool, timeout_ms: u64) bool { + const deadline_ns = core.concurrency.testing.deadlineAfterMilliseconds( + core.concurrency.monotonicNs(), + timeout_ms, + ); + while (!value.load(.acquire)) { + if (core.concurrency.monotonicNs() >= deadline_ns) return false; + std.Thread.yield() catch {}; + } + return true; +} + fn settleScheduler() void { for (0..1000) |_| { std.Thread.yield() catch {}; @@ -103,6 +115,23 @@ const ScheduledProbe = struct { } }; +const BlockingScheduledProbe = struct { + started: AtomicBool = AtomicBool.init(false), + release: AtomicBool = AtomicBool.init(false), + + fn record( + scheduled: *core.RunAfter.Scheduled, + outcome: core.RunAfter.Scheduled.Outcome, + ) void { + const self: *BlockingScheduledProbe = @ptrCast( + @alignCast(scheduled.context.?), + ); + if (outcome != .elapsed) return; + self.started.store(true, .release); + waitUntil(&self.release); + } +}; + test "drainer waits until in-flight work completes" { const Worker = struct { mutex: *core.Mutex, @@ -189,6 +218,46 @@ test "RunAfter runs callback after delay" { waitUntil(&did_run); } +test "RunAfter deadlines account for continuous clock jumps" { + const start_ns = 5 * std.time.ns_per_s; + const deadline_ns = core.concurrency.testing.deadlineAfterMilliseconds( + start_ns, + 100, + ); + + try std.testing.expectEqual( + @as(u64, 100), + core.concurrency.testing.millisecondsUntilDeadline( + deadline_ns, + start_ns, + ), + ); + try std.testing.expectEqual( + @as(u64, 1), + core.concurrency.testing.millisecondsUntilDeadline( + deadline_ns, + deadline_ns - 1, + ), + ); + try std.testing.expectEqual( + @as(u64, 0), + core.concurrency.testing.millisecondsUntilDeadline( + deadline_ns, + deadline_ns + 10 * std.time.ns_per_s, + ), + ); +} + +test "RunAfter deadlines saturate without wrapping" { + try std.testing.expectEqual( + std.math.maxInt(u64), + core.concurrency.testing.deadlineAfterMilliseconds( + std.math.maxInt(u64) - 1, + 1, + ), + ); +} + test "RunAfter cancel prevents callback" { var did_run = AtomicBool.init(false); var run_after = core.RunAfter{}; @@ -267,6 +336,33 @@ test "RunAfter schedules independent one-shot callbacks" { try std.testing.expect(!later_probe.cancelled.load(.acquire)); } +test "RunAfter rearms an earlier deadline while notifying callbacks" { + var blocking_probe = BlockingScheduledProbe{}; + var earlier_probe = ScheduledProbe{}; + var later_probe = ScheduledProbe{}; + var blocking = core.RunAfter.Scheduled{}; + var earlier = core.RunAfter.Scheduled{}; + var later = core.RunAfter.Scheduled{}; + var run_after = core.RunAfter{}; + defer run_after.deinit(); + defer blocking_probe.release.store(true, .release); + + try run_after.scheduleAppending(&later, 2_000, ScheduledProbe.record, &later_probe); + try run_after.scheduleAppending( + &blocking, + 1, + BlockingScheduledProbe.record, + &blocking_probe, + ); + waitUntil(&blocking_probe.started); + + try run_after.scheduleAppending(&earlier, 1, ScheduledProbe.record, &earlier_probe); + blocking_probe.release.store(true, .release); + + try std.testing.expect(waitUntilWithin(&earlier_probe.elapsed, 500)); + try std.testing.expect(!later_probe.elapsed.load(.acquire)); +} + test "RunAfter cancel releases scheduled callbacks" { var probe = ScheduledProbe{}; var scheduled = core.RunAfter.Scheduled{};