authorgravatar for 45520026+kprotty@users.noreply.github.comkprotty <45520026+kprotty@users.noreply.github.com> 2020-10-11 14:18:19-05:00
committergravatar for 45520026+kprotty@users.noreply.github.comkprotty <45520026+kprotty@users.noreply.github.com> 2020-10-11 14:18:19-05:00
logaa53f6d0b5e7fd1904cd24379b77445babf82f96
tree48cee093cb504613870d14b71d66314ebb7a3944
parenta42c0f88e0749a1b17e52bb93d6d31b9dfbca37e

integrate std.time.sleep with the event loop

2 files changed, 158 insertions(+), 1 deletions(-)

lib/std/event/loop.zig+154
...@@ -35,6 +35,9 @@ pub const Loop = struct {...@@ -35,6 +35,9 @@ pub const Loop = struct {
35 /// This is only used by `Loop` for the thread pool and associated resources.35 /// This is only used by `Loop` for the thread pool and associated resources.
36 arena: std.heap.ArenaAllocator,36 arena: std.heap.ArenaAllocator,
3737
38 /// State which manages frames that are sleeping on timers
39 delay_queue: DelayQueue,
40
38 /// Pre-allocated eventfds. All permanently active.41 /// Pre-allocated eventfds. All permanently active.
39 /// This is how `Loop` sends promises to be resumed on other threads.42 /// This is how `Loop` sends promises to be resumed on other threads.
40 available_eventfd_resume_nodes: std.atomic.Stack(ResumeNode.EventFd),43 available_eventfd_resume_nodes: std.atomic.Stack(ResumeNode.EventFd),
...@@ -162,6 +165,7 @@ pub const Loop = struct {...@@ -162,6 +165,7 @@ pub const Loop = struct {
162 .fs_queue = std.atomic.Queue(Request).init(),165 .fs_queue = std.atomic.Queue(Request).init(),
163 .fs_thread = undefined,166 .fs_thread = undefined,
164 .fs_thread_wakeup = std.ResetEvent.init(),167 .fs_thread_wakeup = std.ResetEvent.init(),
168 .delay_queue = undefined,
165 };169 };
166 errdefer self.fs_thread_wakeup.deinit();170 errdefer self.fs_thread_wakeup.deinit();
167 errdefer self.arena.deinit();171 errdefer self.arena.deinit();
...@@ -186,6 +190,9 @@ pub const Loop = struct {...@@ -186,6 +190,9 @@ pub const Loop = struct {
186 self.posixFsRequest(&self.fs_end_request);190 self.posixFsRequest(&self.fs_end_request);
187 self.fs_thread.wait();191 self.fs_thread.wait();
188 };192 };
193
194 if (!std.builtin.single_threaded)
195 try self.delay_queue.init();
189 }196 }
190197
191 pub fn deinit(self: *Loop) void {198 pub fn deinit(self: *Loop) void {
...@@ -645,6 +652,10 @@ pub const Loop = struct {...@@ -645,6 +652,10 @@ pub const Loop = struct {
645 for (self.extra_threads) |extra_thread| {652 for (self.extra_threads) |extra_thread| {
646 extra_thread.wait();653 extra_thread.wait();
647 }654 }
655
656 @atomicStore(bool, &self.delay_queue.is_running, false, .SeqCst);
657 self.delay_queue.event.set();
658 self.delay_queue.thread.wait();
648 }659 }
649660
650 /// Runs the provided function asynchronously. The function's frame is allocated661 /// Runs the provided function asynchronously. The function's frame is allocated
...@@ -748,6 +759,125 @@ pub const Loop = struct {...@@ -748,6 +759,125 @@ pub const Loop = struct {
748 }759 }
749 }760 }
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.delay_queue, 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() ?*Entry {
862 const held = self.entries.mutex.acquire();
863 defer held.release();
864
865 var head = self.entries.head orelse return null;
866
867 var min = head;
868 while (head.next) |node| {
869 const minEntry = @fieldParentPtr(Entry, "node", min);
870 const nodeEntry = @fieldParentPtr(Entry, "node", node);
871 if (nodeEntry.expires < minEntry.expires)
872 min = node;
873 head = node;
874 }
875
876 return @fieldParentPtr(Entry, "node", min);
877 }
878 };
879 };
880
751 /// ------- I/0 APIs -------881 /// ------- I/0 APIs -------
752 pub fn accept(882 pub fn accept(
753 self: *Loop,883 self: *Loop,
...@@ -1550,3 +1680,27 @@ test "std.event.Loop - runDetached" {...@@ -1550,3 +1680,27 @@ test "std.event.Loop - runDetached" {
1550fn testRunDetached() void {1680fn testRunDetached() void {
1551 testRunDetachedData += 1;1681 testRunDetachedData += 1;
1552}1682}
1683
1684test "std.event.Loop - sleep" {
1685 // https://github.com/ziglang/zig/issues/1908
1686 if (builtin.single_threaded) return error.SkipZigTest;
1687 if (!std.io.is_async) return error.SkipZigTest;
1688
1689 const frames = try testing.allocator.alloc(@Frame(testSleep), 10);
1690 defer testing.allocator.free(frames);
1691
1692 const wait_time = 100 * std.time.ns_per_ms;
1693 var sleep_count: usize = 0;
1694
1695 for (frames) |*frame|
1696 frame.* = async testSleep(wait_time, &sleep_count);
1697 for (frames) |*frame|
1698 await frame;
1699
1700 testing.expect(sleep_count == frames.len);
1701}
1702
1703fn testSleep(wait_ns: u64, sleep_count: *usize) void {
1704 Loop.instance.?.sleep(wait_ns);
1705 _ = @atomicRmw(usize, sleep_count, .Add, 1, .SeqCst);
1706}
lib/std/time.zig+4-1
...@@ -14,8 +14,11 @@ const is_windows = std.Target.current.os.tag == .windows;...@@ -14,8 +14,11 @@ const is_windows = std.Target.current.os.tag == .windows;
14pub const epoch = @import("time/epoch.zig");14pub const epoch = @import("time/epoch.zig");
1515
16/// Spurious wakeups are possible and no precision of timing is guaranteed.16/// Spurious wakeups are possible and no precision of timing is guaranteed.
17/// TODO integrate with evented I/O
18pub fn sleep(nanoseconds: u64) void {17pub 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
19 if (is_windows) {22 if (is_windows) {
20 const big_ms_from_ns = nanoseconds / ns_per_ms;23 const big_ms_from_ns = nanoseconds / ns_per_ms;
21 const ms = math.cast(os.windows.DWORD, big_ms_from_ns) catch math.maxInt(os.windows.DWORD);24 const ms = math.cast(os.windows.DWORD, big_ms_from_ns) catch math.maxInt(os.windows.DWORD);