Fix timers started from a callback firing early on epoll and IOCP

src/backend/epoll.zig

@@ -122,6 +122,10 @@ pub const Loop = struct {
 
         /// Whether our loop is in a stopped state or not.
         stopped: bool = false,
+
+        /// True once epoll_wait returns, until "now" is updated again: the
+        /// wait took time that the cached time doesn't count.
+        now_outdated: bool = false,
     } = .{},
 
     pub fn init(options: Options) !Loop {
@@ -241,6 +245,7 @@ pub const Loop = struct {
         const rc = std.os.linux.clock_gettime(std.os.linux.CLOCK.MONOTONIC, &ts);
         if (linux.errno(rc) == .SUCCESS) {
             self.cached_now = ts;
+            self.flags.now_outdated = false;
         }
     }
 
@@ -324,6 +329,9 @@ pub const Loop = struct {
             (next_ms % std.time.ms_per_s) * std.time.ns_per_ms,
         ) orelse return max;
 
+        // "next_ms" is from now, not from before the last wait.
+        if (self.flags.now_outdated) self.update_now();
+
         return .{
             .sec = std.math.add(isize, self.cached_now.sec, next_s) catch
                 return max,
@@ -472,6 +480,7 @@ pub const Loop = struct {
             };
 
             const n = epoll_helper.epoll_wait(self.fd, &events, timeout);
+            self.flags.now_outdated = true;
 
             // Process all our events and invoke their completion handlers
             for (events[0..n]) |ev| {

src/backend/iocp.zig

@@ -70,6 +70,11 @@ pub const Loop = struct {
 
         /// Whether our loop is in a stopped state or not.
         stopped: bool = false,
+
+        /// True once GetQueuedCompletionStatusEx returns, until "now" is
+        /// updated again: the wait took time that the cached time doesn't
+        /// count.
+        now_outdated: bool = false,
     } = .{},
 
     /// Initialize a new IOCP-backed event loop. See the Options docs
@@ -336,6 +341,7 @@ pub const Loop = struct {
 
                 else => return err,
             };
+            self.flags.now_outdated = true;
 
             // Go through the entries and perform completions callbacks.
             for (entries[0..count]) |entry| {
@@ -401,6 +407,7 @@ pub const Loop = struct {
         // Compute the current timestamp in ms by multiplying the QueryPerfomanceCounter value in
         // ticks by the duration of a tick.
         self.cached_now = windows.QueryPerformanceCounter() * self.qpc_duration;
+        self.flags.now_outdated = false;
     }
 
     /// Add a timer to the loop. The timer will execute in "next_ms". This is oneshot: the timer
@@ -467,6 +474,8 @@ pub const Loop = struct {
 
     // Get the absolute timestamp corresponding to the given "next_ms".
     pub fn timer_next(self: *Loop, next_ms: u64) u64 {
+        // "next_ms" is from now, not from before the last wait.
+        if (self.flags.now_outdated) self.update_now();
         return self.cached_now + next_ms * std.time.ns_per_ms;
     }
 

src/watcher/timer.zig

@@ -637,5 +637,78 @@ fn TimerTests(
             try testing.expect(canceled);
             try testing.expect(cancel_confirm);
         }
+
+        test "timer started after a wait counts from then" {
+            // No threads to wake the loop with.
+            if (builtin.os.tag == .wasi) return error.SkipZigTest;
+
+            const testing = std.testing;
+            const wait_ms = 50;
+            const timer_ms = 20;
+
+            var loop = try xev.Loop.init(.{});
+            defer loop.deinit();
+
+            var notifier = try xev.Async.init();
+            defer notifier.deinit();
+
+            const State = struct {
+                timer: Impl,
+                c_timer: xev.Completion = .{},
+                started_ns: i96 = 0,
+                fired_ns: i96 = 0,
+
+                fn nowNs() i96 {
+                    const io = std.Io.Threaded.global_single_threaded.io();
+                    return std.Io.Clock.awake.now(io).nanoseconds;
+                }
+            };
+            var state: State = .{ .timer = try Impl.init() };
+            defer state.timer.deinit();
+
+            // The loop has waited wait_ms when this runs; the timer it starts
+            // must still take timer_ms.
+            var c_wait: xev.Completion = .{};
+            notifier.wait(&loop, &c_wait, State, &state, (struct {
+                fn callback(
+                    ud: ?*State,
+                    l: *xev.Loop,
+                    _: *xev.Completion,
+                    r: xev.Async.WaitError!void,
+                ) xev.CallbackAction {
+                    _ = r catch unreachable;
+                    const s = ud.?;
+                    s.started_ns = State.nowNs();
+                    s.timer.run(l, &s.c_timer, timer_ms, State, s, (struct {
+                        fn callback(
+                            ud_inner: ?*State,
+                            _: *xev.Loop,
+                            _: *xev.Completion,
+                            r_inner: Impl.RunError!void,
+                        ) xev.CallbackAction {
+                            _ = r_inner catch unreachable;
+                            ud_inner.?.fired_ns = State.nowNs();
+                            return .disarm;
+                        }
+                    }).callback);
+                    return .disarm;
+                }
+            }).callback);
+
+            const sender = try std.Thread.spawn(.{}, (struct {
+                fn run(n: *xev.Async) void {
+                    const io = std.Io.Threaded.global_single_threaded.io();
+                    std.Io.sleep(io, .fromMilliseconds(wait_ms), .awake) catch unreachable;
+                    n.notify() catch unreachable;
+                }
+            }).run, .{&notifier});
+            defer sender.join();
+
+            try loop.run(.until_done);
+
+            // A millisecond's slack for the backend's clock not being this one.
+            const elapsed_ns = state.fired_ns - state.started_ns;
+            try testing.expect(elapsed_ns >= (timer_ms - 1) * std.time.ns_per_ms);
+        }
     };
 }