authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2020-10-14 21:49:45-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2020-10-14 21:49:45-04:00
log3b4432d9a6ae7f01f66e5ddcc15bf579d2c47460
treed24b87f33bf868ad6b4857bb500a80441744cc4a
parent2f52f95b928c1dcb42afb22243c27fb9661cbd1b
parent12508025a4b3a841819b913e4ed88810ca04ba79
signaturebadge-question-mark Signed by PGP key 4AEE18F83AFDEB23

Merge pull request #6655 from kprotty/timers

Integrate std.time.sleep with the event loop

4 files changed, 391 insertions(+), 1 deletions(-)

lib/std/auto_reset_event.zig created+229
......@@ -0,0 +1,229 @@
1// SPDX-License-Identifier: MIT
2// Copyright (c) 2015-2020 Zig Contributors
3// This file is part of [zig](https://ziglang.org/), which is MIT licensed.
4// The MIT license requires this copyright notice to be included in all copies
5// and substantial portions of the software.
6const std = @import("std.zig");
7const builtin = @import("builtin");
8const testing = std.testing;
9const assert = std.debug.assert;
10
11/// Similar to std.ResetEvent but on `set()` it also (atomically) does `reset()`.
12/// Unlike std.ResetEvent, `wait()` can only be called by one thread (MPSC-like).
13pub const AutoResetEvent = struct {
14 // AutoResetEvent has 3 possible states:
15 // - UNSET: the AutoResetEvent is currently unset
16 // - SET: the AutoResetEvent was notified before a wait() was called
17 // - <std.ResetEvent pointer>: there is an active waiter waiting for a notification.
18 //
19 // When attempting to wait:
20 // if the event is unset, it registers a ResetEvent pointer to be notified when the event is set
21 // if the event is already set, then it consumes the notification and resets the event.
22 //
23 // When attempting to notify:
24 // if the event is unset, then we set the event
25 // if theres a waiting ResetEvent, then we unset the event and notify the ResetEvent
26 //
27 // This ensures that the event is automatically reset after a wait() has been issued
28 // and avoids the race condition when using std.ResetEvent in the following scenario:
29 // thread 1 | thread 2
30 // std.ResetEvent.wait() |
31 // | std.ResetEvent.set()
32 // | std.ResetEvent.set()
33 // std.ResetEvent.reset() |
34 // std.ResetEvent.wait() | (missed the second .set() notification above)
35
36
37 state: usize = UNSET,
38
39 const UNSET = 0;
40 const SET = 1;
41
42 // the minimum alignment for the `*std.ResetEvent` created by wait*()
43 const event_align = std.math.max(@alignOf(std.ResetEvent), 2);
44
45 pub fn wait(self: *AutoResetEvent) void {
46 self.waitFor(null) catch unreachable;
47 }
48
49 pub fn timedWait(self: *AutoResetEvent, timeout: u64) error{TimedOut}!void {
50 return self.waitFor(timeout);
51 }
52
53 fn waitFor(self: *AutoResetEvent, timeout: ?u64) error{TimedOut}!void {
54 // lazily initialized std.ResetEvent
55 var reset_event: std.ResetEvent align(event_align) = undefined;
56 var has_reset_event = false;
57 defer if (has_reset_event) {
58 reset_event.deinit();
59 };
60
61 var state = @atomicLoad(usize, &self.state, .SeqCst);
62 while (true) {
63 // consume a notification if there is any
64 if (state == SET) {
65 @atomicStore(usize, &self.state, UNSET, .SeqCst);
66 return;
67 }
68
69 // check if theres currently a pending ResetEvent pointer already registered
70 if (state != UNSET) {
71 unreachable; // multiple waiting threads on the same AutoResetEvent
72 }
73
74 // lazily initialize the ResetEvent if it hasn't been already
75 if (!has_reset_event) {
76 has_reset_event = true;
77 reset_event = std.ResetEvent.init();
78 }
79
80 // Since the AutoResetEvent currently isnt set,
81 // try to register our ResetEvent on it to wait
82 // for a set() call from another thread.
83 if (@cmpxchgWeak(
84 usize,
85 &self.state,
86 UNSET,
87 @ptrToInt(&reset_event),
88 .SeqCst,
89 .SeqCst,
90 )) |new_state| {
91 state = new_state;
92 continue;
93 }
94
95 // if no timeout was specified, then just wait forever
96 const timeout_ns = timeout orelse {
97 reset_event.wait();
98 return;
99 };
100
101 // wait with a timeout and return if signalled via set()
102 if (reset_event.timedWait(timeout_ns)) |_| {
103 return;
104 } else |timed_out| {}
105
106 // If we timed out, we need to transition the AutoResetEvent back to UNSET.
107 // If we don't, then when we return, a set() thread could observe a pointer to an invalid ResetEvent.
108 state = @cmpxchgStrong(
109 usize,
110 &self.state,
111 @ptrToInt(&reset_event),
112 UNSET,
113 .SeqCst,
114 .SeqCst,
115 ) orelse return error.TimedOut;
116
117 // We didn't manage to unregister ourselves from the state.
118 if (state == SET) {
119 unreachable; // AutoResetEvent notified without waking up the waiting thread
120 } else if (state != UNSET) {
121 unreachable; // multiple waiting threads on the same AutoResetEvent observed when timing out
122 }
123
124 // This menas a set() thread saw our ResetEvent pointer, acquired it, and is trying to wake it up.
125 // We need to wait for it to wake up our ResetEvent before we can return and invalidate it.
126 // We don't return error.TimedOut here as it technically notified us while we were "timing out".
127 reset_event.wait();
128 return;
129 }
130 }
131
132 pub fn set(self: *AutoResetEvent) void {
133 var state = @atomicLoad(usize, &self.state, .SeqCst);
134 while (true) {
135 // If the AutoResetEvent is already set, there is nothing else left to do
136 if (state == SET) {
137 return;
138 }
139
140 // If the AutoResetEvent isn't set,
141 // then try to leave a notification for the wait() thread that we set() it.
142 if (state == UNSET) {
143 state = @cmpxchgWeak(
144 usize,
145 &self.state,
146 UNSET,
147 SET,
148 .SeqCst,
149 .SeqCst,
150 ) orelse return;
151 continue;
152 }
153
154 // There is a ResetEvent pointer registered on the AutoResetEvent event thats waiting.
155 // Try to acquire ownership of it so that we can wake it up.
156 // This also resets the AutoResetEvent so that there is no race condition as defined above.
157 if (@cmpxchgWeak(
158 usize,
159 &self.state,
160 state,
161 UNSET,
162 .SeqCst,
163 .SeqCst,
164 )) |new_state| {
165 state = new_state;
166 continue;
167 }
168
169 const reset_event = @intToPtr(*align(event_align) std.ResetEvent, state);
170 reset_event.set();
171 return;
172 }
173 }
174};
175
176test "std.AutoResetEvent" {
177 // test local code paths
178 {
179 var event = AutoResetEvent{};
180 testing.expectError(error.TimedOut, event.timedWait(1));
181 event.set();
182 event.wait();
183 }
184
185 // test cross-thread signaling
186 if (builtin.single_threaded)
187 return;
188
189 const Context = struct {
190 value: u128 = 0,
191 in: AutoResetEvent = AutoResetEvent{},
192 out: AutoResetEvent = AutoResetEvent{},
193
194 const Self = @This();
195
196 fn sender(self: *Self) void {
197 testing.expect(self.value == 0);
198 self.value = 1;
199 self.out.set();
200
201 self.in.wait();
202 testing.expect(self.value == 2);
203 self.value = 3;
204 self.out.set();
205
206 self.in.wait();
207 testing.expect(self.value == 4);
208 }
209
210 fn receiver(self: *Self) void {
211 self.out.wait();
212 testing.expect(self.value == 1);
213 self.value = 2;
214 self.in.set();
215
216 self.out.wait();
217 testing.expect(self.value == 3);
218 self.value = 4;
219 self.in.set();
220 }
221 };
222
223 var context = Context{};
224 const send_thread = try std.Thread.spawn(&context, Context.sender);
225 const recv_thread = try std.Thread.spawn(&context, Context.receiver);
226
227 send_thread.wait();
228 recv_thread.wait();
229}
\ No newline at end of file
lib/std/event/loop.zig+157
......@@ -35,6 +35,9 @@ pub const Loop = struct {
3535 /// This is only used by `Loop` for the thread pool and associated resources.
3636 arena: std.heap.ArenaAllocator,
3737
38 /// State which manages frames that are sleeping on timers
39 delay_queue: DelayQueue,
40
3841 /// Pre-allocated eventfds. All permanently active.
3942 /// This is how `Loop` sends promises to be resumed on other threads.
4043 available_eventfd_resume_nodes: std.atomic.Stack(ResumeNode.EventFd),
......@@ -162,6 +165,7 @@ pub const Loop = struct {
162165 .fs_queue = std.atomic.Queue(Request).init(),
163166 .fs_thread = undefined,
164167 .fs_thread_wakeup = std.ResetEvent.init(),
168 .delay_queue = undefined,
165169 };
166170 errdefer self.fs_thread_wakeup.deinit();
167171 errdefer self.arena.deinit();
......@@ -186,6 +190,9 @@ pub const Loop = struct {
186190 self.posixFsRequest(&self.fs_end_request);
187191 self.fs_thread.wait();
188192 };
193
194 if (!std.builtin.single_threaded)
195 try self.delay_queue.init();
189196 }
190197
191198 pub fn deinit(self: *Loop) void {
......@@ -645,6 +652,10 @@ pub const Loop = struct {
645652 for (self.extra_threads) |extra_thread| {
646653 extra_thread.wait();
647654 }
655
656 @atomicStore(bool, &self.delay_queue.is_running, false, .SeqCst);
657 self.delay_queue.event.set();
658 self.delay_queue.thread.wait();
648659 }
649660
650661 /// Runs the provided function asynchronously. The function's frame is allocated
......@@ -748,6 +759,128 @@ pub const Loop = struct {
748759 }
749760 }
750761
762 pub fn sleep(self: *Loop, nanoseconds: u64) void {
763 if (std.builtin.single_threaded)
764 @compileError("TODO: integrate timers with epoll/kevent/iocp for single-threaded");
765
766 suspend {
767 const now = self.delay_queue.timer.read();
768
769 var entry: DelayQueue.Waiters.Entry = undefined;
770 entry.init(@frame(), now + nanoseconds);
771 self.delay_queue.waiters.insert(&entry);
772
773 // Speculatively wake up the timer thread when we add a new entry.
774 // If the timer thread is sleeping on a longer entry, we need to
775 // interrupt it so that our entry can be expired in time.
776 self.delay_queue.event.set();
777 }
778 }
779
780 const DelayQueue = struct {
781 timer: std.time.Timer,
782 waiters: Waiters,
783 thread: *std.Thread,
784 event: std.AutoResetEvent,
785 is_running: bool,
786
787 /// Initialize the delay queue by spawning the timer thread
788 /// and starting any timer resources.
789 fn init(self: *DelayQueue) !void {
790 self.* = DelayQueue{
791 .timer = try std.time.Timer.start(),
792 .waiters = DelayQueue.Waiters{
793 .entries = std.atomic.Queue(anyframe).init(),
794 },
795 .thread = try std.Thread.spawn(self, DelayQueue.run),
796 .event = std.AutoResetEvent{},
797 .is_running = true,
798 };
799 }
800
801 /// Entry point for the timer thread
802 /// which waits for timer entries to expire and reschedules them.
803 fn run(self: *DelayQueue) void {
804 const loop = @fieldParentPtr(Loop, "delay_queue", self);
805
806 while (@atomicLoad(bool, &self.is_running, .SeqCst)) {
807 const now = self.timer.read();
808
809 if (self.waiters.popExpired(now)) |entry| {
810 loop.onNextTick(&entry.node);
811 continue;
812 }
813
814 if (self.waiters.nextExpire()) |expires| {
815 if (now >= expires)
816 continue;
817 self.event.timedWait(expires - now) catch {};
818 } else {
819 self.event.wait();
820 }
821 }
822 }
823
824 // TODO: use a tickless heirarchical timer wheel:
825 // https://github.com/wahern/timeout/
826 const Waiters = struct {
827 entries: std.atomic.Queue(anyframe),
828
829 const Entry = struct {
830 node: NextTickNode,
831 expires: u64,
832
833 fn init(self: *Entry, frame: anyframe, expires: u64) void {
834 self.node.data = frame;
835 self.expires = expires;
836 }
837 };
838
839 /// Registers the entry into the queue of waiting frames
840 fn insert(self: *Waiters, entry: *Entry) void {
841 self.entries.put(&entry.node);
842 }
843
844 /// Dequeues one expired event relative to `now`
845 fn popExpired(self: *Waiters, now: u64) ?*Entry {
846 const entry = self.peekExpiringEntry() orelse return null;
847 if (entry.expires > now)
848 return null;
849
850 assert(self.entries.remove(&entry.node));
851 return entry;
852 }
853
854 /// Returns an estimate for the amount of time
855 /// to wait until the next waiting entry expires.
856 fn nextExpire(self: *Waiters) ?u64 {
857 const entry = self.peekExpiringEntry() orelse return null;
858 return entry.expires;
859 }
860
861 fn peekExpiringEntry(self: *Waiters) ?*Entry {
862 const held = self.entries.mutex.acquire();
863 defer held.release();
864
865 // starting from the head
866 var head = self.entries.head orelse return null;
867
868 // traverse the list of waiting entires to
869 // find the Node with the smallest `expires` field
870 var min = head;
871 while (head.next) |node| {
872 const minEntry = @fieldParentPtr(Entry, "node", min);
873 const nodeEntry = @fieldParentPtr(Entry, "node", node);
874 if (nodeEntry.expires < minEntry.expires)
875 min = node;
876 head = node;
877 }
878
879 return @fieldParentPtr(Entry, "node", min);
880 }
881 };
882 };
883
751884 /// ------- I/0 APIs -------
752885 pub fn accept(
753886 self: *Loop,
......@@ -1550,3 +1683,27 @@ test "std.event.Loop - runDetached" {
15501683fn testRunDetached() void {
15511684 testRunDetachedData += 1;
15521685}
1686
1687test "std.event.Loop - sleep" {
1688 // https://github.com/ziglang/zig/issues/1908
1689 if (builtin.single_threaded) return error.SkipZigTest;
1690 if (!std.io.is_async) return error.SkipZigTest;
1691
1692 const frames = try testing.allocator.alloc(@Frame(testSleep), 10);
1693 defer testing.allocator.free(frames);
1694
1695 const wait_time = 100 * std.time.ns_per_ms;
1696 var sleep_count: usize = 0;
1697
1698 for (frames) |*frame|
1699 frame.* = async testSleep(wait_time, &sleep_count);
1700 for (frames) |*frame|
1701 await frame;
1702
1703 testing.expect(sleep_count == frames.len);
1704}
1705
1706fn testSleep(wait_ns: u64, sleep_count: *usize) void {
1707 Loop.instance.?.sleep(wait_ns);
1708 _ = @atomicRmw(usize, sleep_count, .Add, 1, .SeqCst);
1709}
lib/std/std.zig+1
......@@ -14,6 +14,7 @@ pub const AutoArrayHashMap = array_hash_map.AutoArrayHashMap;
1414pub const AutoArrayHashMapUnmanaged = array_hash_map.AutoArrayHashMapUnmanaged;
1515pub const AutoHashMap = hash_map.AutoHashMap;
1616pub const AutoHashMapUnmanaged = hash_map.AutoHashMapUnmanaged;
17pub const AutoResetEvent = @import("auto_reset_event.zig").AutoResetEvent;
1718pub const BufMap = @import("buf_map.zig").BufMap;
1819pub const BufSet = @import("buf_set.zig").BufSet;
1920pub const ChildProcess = @import("child_process.zig").ChildProcess;
lib/std/time.zig+4-1
......@@ -14,8 +14,11 @@ const is_windows = std.Target.current.os.tag == .windows;
1414pub const epoch = @import("time/epoch.zig");
1515
1616/// Spurious wakeups are possible and no precision of timing is guaranteed.
17/// TODO integrate with evented I/O
1817pub fn sleep(nanoseconds: u64) void {
18 // TODO: opting out of async sleeping?
19 if (std.io.is_async)
20 return std.event.Loop.instance.?.sleep(nanoseconds);
21
1922 if (is_windows) {
2023 const big_ms_from_ns = nanoseconds / ns_per_ms;
2124 const ms = math.cast(os.windows.DWORD, big_ms_from_ns) catch math.maxInt(os.windows.DWORD);