authorgravatar for git@vexu.euVeikka Tuominen <git@vexu.eu> 2024-01-28 00:43:46+02:00
committergravatar for git@vexu.euVeikka Tuominen <git@vexu.eu> 2024-02-01 15:22:36+02:00
logb0bea72588c685c1d6439f61d2e842756b5fc496
tree1bc3e8f0a47dc9e62f16fdaaefe610bd485cc18c
parent8d11ade6a769fe498ed20cdb4f80c6acf4ca91de

remove std.event

15 files changed, 0 insertions(+), 3960 deletions(-)

CMakeLists.txt-3
......@@ -233,9 +233,6 @@ set(ZIG_STAGE2_SOURCES
233233 "${CMAKE_SOURCE_DIR}/lib/std/dwarf/OP.zig"
234234 "${CMAKE_SOURCE_DIR}/lib/std/dwarf/TAG.zig"
235235 "${CMAKE_SOURCE_DIR}/lib/std/elf.zig"
236 "${CMAKE_SOURCE_DIR}/lib/std/event.zig"
237 "${CMAKE_SOURCE_DIR}/lib/std/event/batch.zig"
238 "${CMAKE_SOURCE_DIR}/lib/std/event/loop.zig"
239236 "${CMAKE_SOURCE_DIR}/lib/std/fifo.zig"
240237 "${CMAKE_SOURCE_DIR}/lib/std/fmt.zig"
241238 "${CMAKE_SOURCE_DIR}/lib/std/fmt/errol.zig"
lib/std/event.zig deleted-23
......@@ -1,23 +0,0 @@
1pub const Channel = @import("event/channel.zig").Channel;
2pub const Future = @import("event/future.zig").Future;
3pub const Group = @import("event/group.zig").Group;
4pub const Batch = @import("event/batch.zig").Batch;
5pub const Lock = @import("event/lock.zig").Lock;
6pub const Locked = @import("event/locked.zig").Locked;
7pub const RwLock = @import("event/rwlock.zig").RwLock;
8pub const RwLocked = @import("event/rwlocked.zig").RwLocked;
9pub const Loop = @import("event/loop.zig").Loop;
10pub const WaitGroup = @import("event/wait_group.zig").WaitGroup;
11
12test {
13 _ = @import("event/channel.zig");
14 _ = @import("event/future.zig");
15 _ = @import("event/group.zig");
16 _ = @import("event/batch.zig");
17 _ = @import("event/lock.zig");
18 _ = @import("event/locked.zig");
19 _ = @import("event/rwlock.zig");
20 _ = @import("event/rwlocked.zig");
21 _ = @import("event/loop.zig");
22 _ = @import("event/wait_group.zig");
23}
lib/std/event/batch.zig deleted-141
......@@ -1,141 +0,0 @@
1const std = @import("../std.zig");
2const testing = std.testing;
3
4/// Performs multiple async functions in parallel, without heap allocation.
5/// Async function frames are managed externally to this abstraction, and
6/// passed in via the `add` function. Once all the jobs are added, call `wait`.
7/// This API is *not* thread-safe. The object must be accessed from one thread at
8/// a time, however, it need not be the same thread.
9pub fn Batch(
10 /// The return value for each job.
11 /// If a job slot was re-used due to maxed out concurrency, then its result
12 /// value will be overwritten. The values can be accessed with the `results` field.
13 comptime Result: type,
14 /// How many jobs to run in parallel.
15 comptime max_jobs: comptime_int,
16 /// Controls whether the `add` and `wait` functions will be async functions.
17 comptime async_behavior: enum {
18 /// Observe the value of `std.io.is_async` to decide whether `add`
19 /// and `wait` will be async functions. Asserts that the jobs do not suspend when
20 /// `std.options.io_mode == .blocking`. This is a generally safe assumption, and the
21 /// usual recommended option for this parameter.
22 auto_async,
23
24 /// Always uses the `nosuspend` keyword when using `await` on the jobs,
25 /// making `add` and `wait` non-async functions. Asserts that the jobs do not suspend.
26 never_async,
27
28 /// `add` and `wait` use regular `await` keyword, making them async functions.
29 always_async,
30 },
31) type {
32 return struct {
33 jobs: [max_jobs]Job,
34 next_job_index: usize,
35 collected_result: CollectedResult,
36
37 const Job = struct {
38 frame: ?anyframe->Result,
39 result: Result,
40 };
41
42 const Self = @This();
43
44 const CollectedResult = switch (@typeInfo(Result)) {
45 .ErrorUnion => Result,
46 else => void,
47 };
48
49 const async_ok = switch (async_behavior) {
50 .auto_async => std.io.is_async,
51 .never_async => false,
52 .always_async => true,
53 };
54
55 pub fn init() Self {
56 return Self{
57 .jobs = [1]Job{
58 .{
59 .frame = null,
60 .result = undefined,
61 },
62 } ** max_jobs,
63 .next_job_index = 0,
64 .collected_result = {},
65 };
66 }
67
68 /// Add a frame to the Batch. If all jobs are in-flight, then this function
69 /// waits until one completes.
70 /// This function is *not* thread-safe. It must be called from one thread at
71 /// a time, however, it need not be the same thread.
72 /// TODO: "select" language feature to use the next available slot, rather than
73 /// awaiting the next index.
74 pub fn add(self: *Self, frame: anyframe->Result) void {
75 const job = &self.jobs[self.next_job_index];
76 self.next_job_index = (self.next_job_index + 1) % max_jobs;
77 if (job.frame) |existing| {
78 job.result = if (async_ok) await existing else nosuspend await existing;
79 if (CollectedResult != void) {
80 job.result catch |err| {
81 self.collected_result = err;
82 };
83 }
84 }
85 job.frame = frame;
86 }
87
88 /// Wait for all the jobs to complete.
89 /// Safe to call any number of times.
90 /// If `Result` is an error union, this function returns the last error that occurred, if any.
91 /// Unlike the `results` field, the return value of `wait` will report any error that occurred;
92 /// hitting max parallelism will not compromise the result.
93 /// This function is *not* thread-safe. It must be called from one thread at
94 /// a time, however, it need not be the same thread.
95 pub fn wait(self: *Self) CollectedResult {
96 for (self.jobs) |*job|
97 if (job.frame) |f| {
98 job.result = if (async_ok) await f else nosuspend await f;
99 if (CollectedResult != void) {
100 job.result catch |err| {
101 self.collected_result = err;
102 };
103 }
104 job.frame = null;
105 };
106 return self.collected_result;
107 }
108 };
109}
110
111test "std.event.Batch" {
112 if (true) return error.SkipZigTest;
113 var count: usize = 0;
114 var batch = Batch(void, 2, .auto_async).init();
115 batch.add(&async sleepALittle(&count));
116 batch.add(&async increaseByTen(&count));
117 batch.wait();
118 try testing.expect(count == 11);
119
120 var another = Batch(anyerror!void, 2, .auto_async).init();
121 another.add(&async somethingElse());
122 another.add(&async doSomethingThatFails());
123 try testing.expectError(error.ItBroke, another.wait());
124}
125
126fn sleepALittle(count: *usize) void {
127 std.time.sleep(1 * std.time.ns_per_ms);
128 _ = @atomicRmw(usize, count, .Add, 1, .SeqCst);
129}
130
131fn increaseByTen(count: *usize) void {
132 var i: usize = 0;
133 while (i < 10) : (i += 1) {
134 _ = @atomicRmw(usize, count, .Add, 1, .SeqCst);
135 }
136}
137
138fn doSomethingThatFails() anyerror!void {}
139fn somethingElse() anyerror!void {
140 return error.ItBroke;
141}
lib/std/event/channel.zig deleted-334
......@@ -1,334 +0,0 @@
1const std = @import("../std.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const testing = std.testing;
5const Loop = std.event.Loop;
6
7/// Many producer, many consumer, thread-safe, runtime configurable buffer size.
8/// When buffer is empty, consumers suspend and are resumed by producers.
9/// When buffer is full, producers suspend and are resumed by consumers.
10pub fn Channel(comptime T: type) type {
11 return struct {
12 getters: std.atomic.Queue(GetNode),
13 or_null_queue: std.atomic.Queue(*std.atomic.Queue(GetNode).Node),
14 putters: std.atomic.Queue(PutNode),
15 get_count: usize,
16 put_count: usize,
17 dispatch_lock: bool,
18 need_dispatch: bool,
19
20 // simple fixed size ring buffer
21 buffer_nodes: []T,
22 buffer_index: usize,
23 buffer_len: usize,
24
25 const SelfChannel = @This();
26 const GetNode = struct {
27 tick_node: *Loop.NextTickNode,
28 data: Data,
29
30 const Data = union(enum) {
31 Normal: Normal,
32 OrNull: OrNull,
33 };
34
35 const Normal = struct {
36 ptr: *T,
37 };
38
39 const OrNull = struct {
40 ptr: *?T,
41 or_null: *std.atomic.Queue(*std.atomic.Queue(GetNode).Node).Node,
42 };
43 };
44 const PutNode = struct {
45 data: T,
46 tick_node: *Loop.NextTickNode,
47 };
48
49 const global_event_loop = Loop.instance orelse
50 @compileError("std.event.Channel currently only works with event-based I/O");
51
52 /// Call `deinit` to free resources when done.
53 /// `buffer` must live until `deinit` is called.
54 /// For a zero length buffer, use `[0]T{}`.
55 /// TODO https://github.com/ziglang/zig/issues/2765
56 pub fn init(self: *SelfChannel, buffer: []T) void {
57 // The ring buffer implementation only works with power of 2 buffer sizes
58 // because of relying on subtracting across zero. For example (0 -% 1) % 10 == 5
59 assert(buffer.len == 0 or @popCount(buffer.len) == 1);
60
61 self.* = SelfChannel{
62 .buffer_len = 0,
63 .buffer_nodes = buffer,
64 .buffer_index = 0,
65 .dispatch_lock = false,
66 .need_dispatch = false,
67 .getters = std.atomic.Queue(GetNode).init(),
68 .putters = std.atomic.Queue(PutNode).init(),
69 .or_null_queue = std.atomic.Queue(*std.atomic.Queue(GetNode).Node).init(),
70 .get_count = 0,
71 .put_count = 0,
72 };
73 }
74
75 /// Must be called when all calls to put and get have suspended and no more calls occur.
76 /// This can be omitted if caller can guarantee that the suspended putters and getters
77 /// do not need to be run to completion. Note that this may leave awaiters hanging.
78 pub fn deinit(self: *SelfChannel) void {
79 while (self.getters.get()) |get_node| {
80 resume get_node.data.tick_node.data;
81 }
82 while (self.putters.get()) |put_node| {
83 resume put_node.data.tick_node.data;
84 }
85 self.* = undefined;
86 }
87
88 /// puts a data item in the channel. The function returns when the value has been added to the
89 /// buffer, or in the case of a zero size buffer, when the item has been retrieved by a getter.
90 /// Or when the channel is destroyed.
91 pub fn put(self: *SelfChannel, data: T) void {
92 var my_tick_node = Loop.NextTickNode{ .data = @frame() };
93 var queue_node = std.atomic.Queue(PutNode).Node{
94 .data = PutNode{
95 .tick_node = &my_tick_node,
96 .data = data,
97 },
98 };
99
100 suspend {
101 self.putters.put(&queue_node);
102 _ = @atomicRmw(usize, &self.put_count, .Add, 1, .SeqCst);
103
104 self.dispatch();
105 }
106 }
107
108 /// await this function to get an item from the channel. If the buffer is empty, the frame will
109 /// complete when the next item is put in the channel.
110 pub fn get(self: *SelfChannel) callconv(.Async) T {
111 // TODO https://github.com/ziglang/zig/issues/2765
112 var result: T = undefined;
113 var my_tick_node = Loop.NextTickNode{ .data = @frame() };
114 var queue_node = std.atomic.Queue(GetNode).Node{
115 .data = GetNode{
116 .tick_node = &my_tick_node,
117 .data = GetNode.Data{
118 .Normal = GetNode.Normal{ .ptr = &result },
119 },
120 },
121 };
122
123 suspend {
124 self.getters.put(&queue_node);
125 _ = @atomicRmw(usize, &self.get_count, .Add, 1, .SeqCst);
126
127 self.dispatch();
128 }
129 return result;
130 }
131
132 //pub async fn select(comptime EnumUnion: type, channels: ...) EnumUnion {
133 // assert(@memberCount(EnumUnion) == channels.len); // enum union and channels mismatch
134 // assert(channels.len != 0); // enum unions cannot have 0 fields
135 // if (channels.len == 1) {
136 // const result = await (async channels[0].get() catch unreachable);
137 // return @unionInit(EnumUnion, @memberName(EnumUnion, 0), result);
138 // }
139 //}
140
141 /// Get an item from the channel. If the buffer is empty and there are no
142 /// puts waiting, this returns `null`.
143 pub fn getOrNull(self: *SelfChannel) ?T {
144 // TODO integrate this function with named return values
145 // so we can get rid of this extra result copy
146 var result: ?T = null;
147 var my_tick_node = Loop.NextTickNode{ .data = @frame() };
148 var or_null_node = std.atomic.Queue(*std.atomic.Queue(GetNode).Node).Node{ .data = undefined };
149 var queue_node = std.atomic.Queue(GetNode).Node{
150 .data = GetNode{
151 .tick_node = &my_tick_node,
152 .data = GetNode.Data{
153 .OrNull = GetNode.OrNull{
154 .ptr = &result,
155 .or_null = &or_null_node,
156 },
157 },
158 },
159 };
160 or_null_node.data = &queue_node;
161
162 suspend {
163 self.getters.put(&queue_node);
164 _ = @atomicRmw(usize, &self.get_count, .Add, 1, .SeqCst);
165 self.or_null_queue.put(&or_null_node);
166
167 self.dispatch();
168 }
169 return result;
170 }
171
172 fn dispatch(self: *SelfChannel) void {
173 // set the "need dispatch" flag
174 @atomicStore(bool, &self.need_dispatch, true, .SeqCst);
175
176 lock: while (true) {
177 // set the lock flag
178 if (@atomicRmw(bool, &self.dispatch_lock, .Xchg, true, .SeqCst)) return;
179
180 // clear the need_dispatch flag since we're about to do it
181 @atomicStore(bool, &self.need_dispatch, false, .SeqCst);
182
183 while (true) {
184 one_dispatch: {
185 // later we correct these extra subtractions
186 var get_count = @atomicRmw(usize, &self.get_count, .Sub, 1, .SeqCst);
187 var put_count = @atomicRmw(usize, &self.put_count, .Sub, 1, .SeqCst);
188
189 // transfer self.buffer to self.getters
190 while (self.buffer_len != 0) {
191 if (get_count == 0) break :one_dispatch;
192
193 const get_node = &self.getters.get().?.data;
194 switch (get_node.data) {
195 GetNode.Data.Normal => |info| {
196 info.ptr.* = self.buffer_nodes[(self.buffer_index -% self.buffer_len) % self.buffer_nodes.len];
197 },
198 GetNode.Data.OrNull => |info| {
199 _ = self.or_null_queue.remove(info.or_null);
200 info.ptr.* = self.buffer_nodes[(self.buffer_index -% self.buffer_len) % self.buffer_nodes.len];
201 },
202 }
203 global_event_loop.onNextTick(get_node.tick_node);
204 self.buffer_len -= 1;
205
206 get_count = @atomicRmw(usize, &self.get_count, .Sub, 1, .SeqCst);
207 }
208
209 // direct transfer self.putters to self.getters
210 while (get_count != 0 and put_count != 0) {
211 const get_node = &self.getters.get().?.data;
212 const put_node = &self.putters.get().?.data;
213
214 switch (get_node.data) {
215 GetNode.Data.Normal => |info| {
216 info.ptr.* = put_node.data;
217 },
218 GetNode.Data.OrNull => |info| {
219 _ = self.or_null_queue.remove(info.or_null);
220 info.ptr.* = put_node.data;
221 },
222 }
223 global_event_loop.onNextTick(get_node.tick_node);
224 global_event_loop.onNextTick(put_node.tick_node);
225
226 get_count = @atomicRmw(usize, &self.get_count, .Sub, 1, .SeqCst);
227 put_count = @atomicRmw(usize, &self.put_count, .Sub, 1, .SeqCst);
228 }
229
230 // transfer self.putters to self.buffer
231 while (self.buffer_len != self.buffer_nodes.len and put_count != 0) {
232 const put_node = &self.putters.get().?.data;
233
234 self.buffer_nodes[self.buffer_index % self.buffer_nodes.len] = put_node.data;
235 global_event_loop.onNextTick(put_node.tick_node);
236 self.buffer_index +%= 1;
237 self.buffer_len += 1;
238
239 put_count = @atomicRmw(usize, &self.put_count, .Sub, 1, .SeqCst);
240 }
241 }
242
243 // undo the extra subtractions
244 _ = @atomicRmw(usize, &self.get_count, .Add, 1, .SeqCst);
245 _ = @atomicRmw(usize, &self.put_count, .Add, 1, .SeqCst);
246
247 // All the "get or null" functions should resume now.
248 var remove_count: usize = 0;
249 while (self.or_null_queue.get()) |or_null_node| {
250 remove_count += @intFromBool(self.getters.remove(or_null_node.data));
251 global_event_loop.onNextTick(or_null_node.data.data.tick_node);
252 }
253 if (remove_count != 0) {
254 _ = @atomicRmw(usize, &self.get_count, .Sub, remove_count, .SeqCst);
255 }
256
257 // clear need-dispatch flag
258 if (@atomicRmw(bool, &self.need_dispatch, .Xchg, false, .SeqCst)) continue;
259
260 assert(@atomicRmw(bool, &self.dispatch_lock, .Xchg, false, .SeqCst));
261
262 // we have to check again now that we unlocked
263 if (@atomicLoad(bool, &self.need_dispatch, .SeqCst)) continue :lock;
264
265 return;
266 }
267 }
268 }
269 };
270}
271
272test "std.event.Channel" {
273 if (!std.io.is_async) return error.SkipZigTest;
274
275 // https://github.com/ziglang/zig/issues/1908
276 if (builtin.single_threaded) return error.SkipZigTest;
277
278 // https://github.com/ziglang/zig/issues/3251
279 if (builtin.os.tag == .freebsd) return error.SkipZigTest;
280
281 var channel: Channel(i32) = undefined;
282 channel.init(&[0]i32{});
283 defer channel.deinit();
284
285 var handle = async testChannelGetter(&channel);
286 var putter = async testChannelPutter(&channel);
287
288 await handle;
289 await putter;
290}
291
292test "std.event.Channel wraparound" {
293
294 // TODO provide a way to run tests in evented I/O mode
295 if (!std.io.is_async) return error.SkipZigTest;
296
297 const channel_size = 2;
298
299 var buf: [channel_size]i32 = undefined;
300 var channel: Channel(i32) = undefined;
301 channel.init(&buf);
302 defer channel.deinit();
303
304 // add items to channel and pull them out until
305 // the buffer wraps around, make sure it doesn't crash.
306 channel.put(5);
307 try testing.expectEqual(@as(i32, 5), channel.get());
308 channel.put(6);
309 try testing.expectEqual(@as(i32, 6), channel.get());
310 channel.put(7);
311 try testing.expectEqual(@as(i32, 7), channel.get());
312}
313fn testChannelGetter(channel: *Channel(i32)) callconv(.Async) void {
314 const value1 = channel.get();
315 try testing.expect(value1 == 1234);
316
317 const value2 = channel.get();
318 try testing.expect(value2 == 4567);
319
320 const value3 = channel.getOrNull();
321 try testing.expect(value3 == null);
322
323 var last_put = async testPut(channel, 4444);
324 const value4 = channel.getOrNull();
325 try testing.expect(value4.? == 4444);
326 await last_put;
327}
328fn testChannelPutter(channel: *Channel(i32)) callconv(.Async) void {
329 channel.put(1234);
330 channel.put(4567);
331}
332fn testPut(channel: *Channel(i32), value: i32) callconv(.Async) void {
333 channel.put(value);
334}
lib/std/event/future.zig deleted-115
......@@ -1,115 +0,0 @@
1const std = @import("../std.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const testing = std.testing;
5const Lock = std.event.Lock;
6
7/// This is a value that starts out unavailable, until resolve() is called.
8/// While it is unavailable, functions suspend when they try to get() it,
9/// and then are resumed when resolve() is called.
10/// At this point the value remains forever available, and another resolve() is not allowed.
11pub fn Future(comptime T: type) type {
12 return struct {
13 lock: Lock,
14 data: T,
15 available: Available,
16
17 const Available = enum(u8) {
18 NotStarted,
19 Started,
20 Finished,
21 };
22
23 const Self = @This();
24 const Queue = std.atomic.Queue(anyframe);
25
26 pub fn init() Self {
27 return Self{
28 .lock = Lock.initLocked(),
29 .available = .NotStarted,
30 .data = undefined,
31 };
32 }
33
34 /// Obtain the value. If it's not available, wait until it becomes
35 /// available.
36 /// Thread-safe.
37 pub fn get(self: *Self) callconv(.Async) *T {
38 if (@atomicLoad(Available, &self.available, .SeqCst) == .Finished) {
39 return &self.data;
40 }
41 const held = self.lock.acquire();
42 held.release();
43
44 return &self.data;
45 }
46
47 /// Gets the data without waiting for it. If it's available, a pointer is
48 /// returned. Otherwise, null is returned.
49 pub fn getOrNull(self: *Self) ?*T {
50 if (@atomicLoad(Available, &self.available, .SeqCst) == .Finished) {
51 return &self.data;
52 } else {
53 return null;
54 }
55 }
56
57 /// If someone else has started working on the data, wait for them to complete
58 /// and return a pointer to the data. Otherwise, return null, and the caller
59 /// should start working on the data.
60 /// It's not required to call start() before resolve() but it can be useful since
61 /// this method is thread-safe.
62 pub fn start(self: *Self) callconv(.Async) ?*T {
63 const state = @cmpxchgStrong(Available, &self.available, .NotStarted, .Started, .SeqCst, .SeqCst) orelse return null;
64 switch (state) {
65 .Started => {
66 const held = self.lock.acquire();
67 held.release();
68 return &self.data;
69 },
70 .Finished => return &self.data,
71 else => unreachable,
72 }
73 }
74
75 /// Make the data become available. May be called only once.
76 /// Before calling this, modify the `data` property.
77 pub fn resolve(self: *Self) void {
78 const prev = @atomicRmw(Available, &self.available, .Xchg, .Finished, .SeqCst);
79 assert(prev != .Finished); // resolve() called twice
80 Lock.Held.release(Lock.Held{ .lock = &self.lock });
81 }
82 };
83}
84
85test "std.event.Future" {
86 // https://github.com/ziglang/zig/issues/1908
87 if (builtin.single_threaded) return error.SkipZigTest;
88 // https://github.com/ziglang/zig/issues/3251
89 if (builtin.os.tag == .freebsd) return error.SkipZigTest;
90 // TODO provide a way to run tests in evented I/O mode
91 if (!std.io.is_async) return error.SkipZigTest;
92
93 testFuture();
94}
95
96fn testFuture() void {
97 var future = Future(i32).init();
98
99 var a = async waitOnFuture(&future);
100 var b = async waitOnFuture(&future);
101 resolveFuture(&future);
102
103 const result = (await a) + (await b);
104
105 try testing.expect(result == 12);
106}
107
108fn waitOnFuture(future: *Future(i32)) i32 {
109 return future.get().*;
110}
111
112fn resolveFuture(future: *Future(i32)) void {
113 future.data = 6;
114 future.resolve();
115}
lib/std/event/group.zig deleted-160
......@@ -1,160 +0,0 @@
1const std = @import("../std.zig");
2const builtin = @import("builtin");
3const Lock = std.event.Lock;
4const testing = std.testing;
5const Allocator = std.mem.Allocator;
6
7/// ReturnType must be `void` or `E!void`
8/// TODO This API was created back with the old design of async/await, when calling any
9/// async function required an allocator. There is an ongoing experiment to transition
10/// all uses of this API to the simpler and more resource-aware `std.event.Batch` API.
11/// If the transition goes well, all usages of `Group` will be gone, and this API
12/// will be deleted.
13pub fn Group(comptime ReturnType: type) type {
14 return struct {
15 frame_stack: Stack,
16 alloc_stack: AllocStack,
17 lock: Lock,
18 allocator: Allocator,
19
20 const Self = @This();
21
22 const Error = switch (@typeInfo(ReturnType)) {
23 .ErrorUnion => |payload| payload.error_set,
24 else => void,
25 };
26 const Stack = std.atomic.Stack(anyframe->ReturnType);
27 const AllocStack = std.atomic.Stack(Node);
28
29 pub const Node = struct {
30 bytes: []const u8 = &[0]u8{},
31 handle: anyframe->ReturnType,
32 };
33
34 pub fn init(allocator: Allocator) Self {
35 return Self{
36 .frame_stack = Stack.init(),
37 .alloc_stack = AllocStack.init(),
38 .lock = .{},
39 .allocator = allocator,
40 };
41 }
42
43 /// Add a frame to the group. Thread-safe.
44 pub fn add(self: *Self, handle: anyframe->ReturnType) (error{OutOfMemory}!void) {
45 const node = try self.allocator.create(AllocStack.Node);
46 node.* = AllocStack.Node{
47 .next = undefined,
48 .data = Node{
49 .handle = handle,
50 },
51 };
52 self.alloc_stack.push(node);
53 }
54
55 /// Add a node to the group. Thread-safe. Cannot fail.
56 /// `node.data` should be the frame handle to add to the group.
57 /// The node's memory should be in the function frame of
58 /// the handle that is in the node, or somewhere guaranteed to live
59 /// at least as long.
60 pub fn addNode(self: *Self, node: *Stack.Node) void {
61 self.frame_stack.push(node);
62 }
63
64 /// This is equivalent to adding a frame to the group but the memory of its frame is
65 /// allocated by the group and freed by `wait`.
66 /// `func` must be async and have return type `ReturnType`.
67 /// Thread-safe.
68 pub fn call(self: *Self, comptime func: anytype, args: anytype) error{OutOfMemory}!void {
69 const frame = try self.allocator.create(@TypeOf(@call(.{ .modifier = .async_kw }, func, args)));
70 errdefer self.allocator.destroy(frame);
71 const node = try self.allocator.create(AllocStack.Node);
72 errdefer self.allocator.destroy(node);
73 node.* = AllocStack.Node{
74 .next = undefined,
75 .data = Node{
76 .handle = frame,
77 .bytes = std.mem.asBytes(frame),
78 },
79 };
80 frame.* = @call(.{ .modifier = .async_kw }, func, args);
81 self.alloc_stack.push(node);
82 }
83
84 /// Wait for all the calls and promises of the group to complete.
85 /// Thread-safe.
86 /// Safe to call any number of times.
87 pub fn wait(self: *Self) callconv(.Async) ReturnType {
88 const held = self.lock.acquire();
89 defer held.release();
90
91 var result: ReturnType = {};
92
93 while (self.frame_stack.pop()) |node| {
94 if (Error == void) {
95 await node.data;
96 } else {
97 (await node.data) catch |err| {
98 result = err;
99 };
100 }
101 }
102 while (self.alloc_stack.pop()) |node| {
103 const handle = node.data.handle;
104 if (Error == void) {
105 await handle;
106 } else {
107 (await handle) catch |err| {
108 result = err;
109 };
110 }
111 self.allocator.free(node.data.bytes);
112 self.allocator.destroy(node);
113 }
114 return result;
115 }
116 };
117}
118
119test "std.event.Group" {
120 // https://github.com/ziglang/zig/issues/1908
121 if (builtin.single_threaded) return error.SkipZigTest;
122
123 if (!std.io.is_async) return error.SkipZigTest;
124
125 // TODO this file has bit-rotted. repair it
126 if (true) return error.SkipZigTest;
127
128 _ = async testGroup(std.heap.page_allocator);
129}
130fn testGroup(allocator: Allocator) callconv(.Async) void {
131 var count: usize = 0;
132 var group = Group(void).init(allocator);
133 var sleep_a_little_frame = async sleepALittle(&count);
134 group.add(&sleep_a_little_frame) catch @panic("memory");
135 var increase_by_ten_frame = async increaseByTen(&count);
136 group.add(&increase_by_ten_frame) catch @panic("memory");
137 group.wait();
138 try testing.expect(count == 11);
139
140 var another = Group(anyerror!void).init(allocator);
141 var something_else_frame = async somethingElse();
142 another.add(&something_else_frame) catch @panic("memory");
143 var something_that_fails_frame = async doSomethingThatFails();
144 another.add(&something_that_fails_frame) catch @panic("memory");
145 try testing.expectError(error.ItBroke, another.wait());
146}
147fn sleepALittle(count: *usize) callconv(.Async) void {
148 std.time.sleep(1 * std.time.ns_per_ms);
149 _ = @atomicRmw(usize, count, .Add, 1, .SeqCst);
150}
151fn increaseByTen(count: *usize) callconv(.Async) void {
152 var i: usize = 0;
153 while (i < 10) : (i += 1) {
154 _ = @atomicRmw(usize, count, .Add, 1, .SeqCst);
155 }
156}
157fn doSomethingThatFails() callconv(.Async) anyerror!void {}
158fn somethingElse() callconv(.Async) anyerror!void {
159 return error.ItBroke;
160}
lib/std/event/lock.zig deleted-162
......@@ -1,162 +0,0 @@
1const std = @import("../std.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const testing = std.testing;
5const mem = std.mem;
6const Loop = std.event.Loop;
7
8/// Thread-safe async/await lock.
9/// Functions which are waiting for the lock are suspended, and
10/// are resumed when the lock is released, in order.
11/// Allows only one actor to hold the lock.
12/// TODO: make this API also work in blocking I/O mode.
13pub const Lock = struct {
14 mutex: std.Thread.Mutex = std.Thread.Mutex{},
15 head: usize = UNLOCKED,
16
17 const UNLOCKED = 0;
18 const LOCKED = 1;
19
20 const global_event_loop = Loop.instance orelse
21 @compileError("std.event.Lock currently only works with event-based I/O");
22
23 const Waiter = struct {
24 // forced Waiter alignment to ensure it doesn't clash with LOCKED
25 next: ?*Waiter align(2),
26 tail: *Waiter,
27 node: Loop.NextTickNode,
28 };
29
30 pub fn initLocked() Lock {
31 return Lock{ .head = LOCKED };
32 }
33
34 pub fn acquire(self: *Lock) Held {
35 self.mutex.lock();
36
37 // self.head transitions from multiple stages depending on the value:
38 // UNLOCKED -> LOCKED:
39 // acquire Lock ownership when there are no waiters
40 // LOCKED -> <Waiter head ptr>:
41 // Lock is already owned, enqueue first Waiter
42 // <head ptr> -> <head ptr>:
43 // Lock is owned with pending waiters. Push our waiter to the queue.
44
45 if (self.head == UNLOCKED) {
46 self.head = LOCKED;
47 self.mutex.unlock();
48 return Held{ .lock = self };
49 }
50
51 var waiter: Waiter = undefined;
52 waiter.next = null;
53 waiter.tail = &waiter;
54
55 const head = switch (self.head) {
56 UNLOCKED => unreachable,
57 LOCKED => null,
58 else => @as(*Waiter, @ptrFromInt(self.head)),
59 };
60
61 if (head) |h| {
62 h.tail.next = &waiter;
63 h.tail = &waiter;
64 } else {
65 self.head = @intFromPtr(&waiter);
66 }
67
68 suspend {
69 waiter.node = Loop.NextTickNode{
70 .prev = undefined,
71 .next = undefined,
72 .data = @frame(),
73 };
74 self.mutex.unlock();
75 }
76
77 return Held{ .lock = self };
78 }
79
80 pub const Held = struct {
81 lock: *Lock,
82
83 pub fn release(self: Held) void {
84 const waiter = blk: {
85 self.lock.mutex.lock();
86 defer self.lock.mutex.unlock();
87
88 // self.head goes through the reverse transition from acquire():
89 // <head ptr> -> <new head ptr>:
90 // pop a waiter from the queue to give Lock ownership when there are still others pending
91 // <head ptr> -> LOCKED:
92 // pop the laster waiter from the queue, while also giving it lock ownership when awaken
93 // LOCKED -> UNLOCKED:
94 // last lock owner releases lock while no one else is waiting for it
95
96 switch (self.lock.head) {
97 UNLOCKED => {
98 unreachable; // Lock unlocked while unlocking
99 },
100 LOCKED => {
101 self.lock.head = UNLOCKED;
102 break :blk null;
103 },
104 else => {
105 const waiter = @as(*Waiter, @ptrFromInt(self.lock.head));
106 self.lock.head = if (waiter.next == null) LOCKED else @intFromPtr(waiter.next);
107 if (waiter.next) |next|
108 next.tail = waiter.tail;
109 break :blk waiter;
110 },
111 }
112 };
113
114 if (waiter) |w| {
115 global_event_loop.onNextTick(&w.node);
116 }
117 }
118 };
119};
120
121test "std.event.Lock" {
122 if (!std.io.is_async) return error.SkipZigTest;
123
124 // TODO https://github.com/ziglang/zig/issues/1908
125 if (builtin.single_threaded) return error.SkipZigTest;
126
127 // TODO https://github.com/ziglang/zig/issues/3251
128 if (builtin.os.tag == .freebsd) return error.SkipZigTest;
129
130 var lock = Lock{};
131 testLock(&lock);
132
133 const expected_result = [1]i32{3 * @as(i32, @intCast(shared_test_data.len))} ** shared_test_data.len;
134 try testing.expectEqualSlices(i32, &expected_result, &shared_test_data);
135}
136fn testLock(lock: *Lock) void {
137 var handle1 = async lockRunner(lock);
138 var handle2 = async lockRunner(lock);
139 var handle3 = async lockRunner(lock);
140
141 await handle1;
142 await handle2;
143 await handle3;
144}
145
146var shared_test_data = [1]i32{0} ** 10;
147var shared_test_index: usize = 0;
148
149fn lockRunner(lock: *Lock) void {
150 Lock.global_event_loop.yield();
151
152 var i: usize = 0;
153 while (i < shared_test_data.len) : (i += 1) {
154 const handle = lock.acquire();
155 defer handle.release();
156
157 shared_test_index = 0;
158 while (shared_test_index < shared_test_data.len) : (shared_test_index += 1) {
159 shared_test_data[shared_test_index] = shared_test_data[shared_test_index] + 1;
160 }
161 }
162}
lib/std/event/locked.zig deleted-42
......@@ -1,42 +0,0 @@
1const std = @import("../std.zig");
2const Lock = std.event.Lock;
3
4/// Thread-safe async/await lock that protects one piece of data.
5/// Functions which are waiting for the lock are suspended, and
6/// are resumed when the lock is released, in order.
7pub fn Locked(comptime T: type) type {
8 return struct {
9 lock: Lock,
10 private_data: T,
11
12 const Self = @This();
13
14 pub const HeldLock = struct {
15 value: *T,
16 held: Lock.Held,
17
18 pub fn release(self: HeldLock) void {
19 self.held.release();
20 }
21 };
22
23 pub fn init(data: T) Self {
24 return Self{
25 .lock = .{},
26 .private_data = data,
27 };
28 }
29
30 pub fn deinit(self: *Self) void {
31 self.lock.deinit();
32 }
33
34 pub fn acquire(self: *Self) callconv(.Async) HeldLock {
35 return HeldLock{
36 // TODO guaranteed allocation elision
37 .held = self.lock.acquire(),
38 .value = &self.private_data,
39 };
40 }
41 };
42}
lib/std/event/loop.zig deleted-1791
......@@ -1,1791 +0,0 @@
1const std = @import("../std.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const testing = std.testing;
5const mem = std.mem;
6const os = std.os;
7const windows = os.windows;
8const maxInt = std.math.maxInt;
9const Thread = std.Thread;
10
11const is_windows = builtin.os.tag == .windows;
12
13pub const Loop = struct {
14 next_tick_queue: std.atomic.Queue(anyframe),
15 os_data: OsData,
16 final_resume_node: ResumeNode,
17 pending_event_count: usize,
18 extra_threads: []Thread,
19 /// TODO change this to a pool of configurable number of threads
20 /// and rename it to be not file-system-specific. it will become
21 /// a thread pool for turning non-CPU-bound blocking things into
22 /// async things. A fallback for any missing OS-specific API.
23 fs_thread: Thread,
24 fs_queue: std.atomic.Queue(Request),
25 fs_end_request: Request.Node,
26 fs_thread_wakeup: std.Thread.ResetEvent,
27
28 /// For resources that have the same lifetime as the `Loop`.
29 /// This is only used by `Loop` for the thread pool and associated resources.
30 arena: std.heap.ArenaAllocator,
31
32 /// State which manages frames that are sleeping on timers
33 delay_queue: DelayQueue,
34
35 /// Pre-allocated eventfds. All permanently active.
36 /// This is how `Loop` sends promises to be resumed on other threads.
37 available_eventfd_resume_nodes: std.atomic.Stack(ResumeNode.EventFd),
38 eventfd_resume_nodes: []std.atomic.Stack(ResumeNode.EventFd).Node,
39
40 pub const NextTickNode = std.atomic.Queue(anyframe).Node;
41
42 pub const ResumeNode = struct {
43 id: Id,
44 handle: anyframe,
45 overlapped: Overlapped,
46
47 pub const overlapped_init = switch (builtin.os.tag) {
48 .windows => windows.OVERLAPPED{
49 .Internal = 0,
50 .InternalHigh = 0,
51 .DUMMYUNIONNAME = .{
52 .DUMMYSTRUCTNAME = .{
53 .Offset = 0,
54 .OffsetHigh = 0,
55 },
56 },
57 .hEvent = null,
58 },
59 else => {},
60 };
61 pub const Overlapped = @TypeOf(overlapped_init);
62
63 pub const Id = enum {
64 basic,
65 stop,
66 event_fd,
67 };
68
69 pub const EventFd = switch (builtin.os.tag) {
70 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => KEventFd,
71 .linux => struct {
72 base: ResumeNode,
73 epoll_op: u32,
74 eventfd: i32,
75 },
76 .windows => struct {
77 base: ResumeNode,
78 completion_key: usize,
79 },
80 else => struct {},
81 };
82
83 const KEventFd = struct {
84 base: ResumeNode,
85 kevent: os.Kevent,
86 };
87
88 pub const Basic = switch (builtin.os.tag) {
89 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => KEventBasic,
90 .linux => struct {
91 base: ResumeNode,
92 },
93 .windows => struct {
94 base: ResumeNode,
95 },
96 else => @compileError("unsupported OS"),
97 };
98
99 const KEventBasic = struct {
100 base: ResumeNode,
101 kev: os.Kevent,
102 };
103 };
104
105 pub const Instance = switch (std.options.io_mode) {
106 .blocking => @TypeOf(null),
107 .evented => ?*Loop,
108 };
109 pub const instance = std.options.event_loop;
110
111 var global_instance_state: Loop = undefined;
112 pub const default_instance = switch (std.options.io_mode) {
113 .blocking => null,
114 .evented => &global_instance_state,
115 };
116
117 pub const Mode = enum {
118 single_threaded,
119 multi_threaded,
120 };
121 pub const default_mode = .multi_threaded;
122
123 /// TODO copy elision / named return values so that the threads referencing *Loop
124 /// have the correct pointer value.
125 /// https://github.com/ziglang/zig/issues/2761 and https://github.com/ziglang/zig/issues/2765
126 pub fn init(self: *Loop) !void {
127 if (builtin.single_threaded or std.options.event_loop_mode == .single_threaded) {
128 return self.initSingleThreaded();
129 } else {
130 return self.initMultiThreaded();
131 }
132 }
133
134 /// After initialization, call run().
135 /// TODO copy elision / named return values so that the threads referencing *Loop
136 /// have the correct pointer value.
137 /// https://github.com/ziglang/zig/issues/2761 and https://github.com/ziglang/zig/issues/2765
138 pub fn initSingleThreaded(self: *Loop) !void {
139 return self.initThreadPool(1);
140 }
141
142 /// After initialization, call run().
143 /// This is the same as `initThreadPool` using `Thread.getCpuCount` to determine the thread
144 /// pool size.
145 /// TODO copy elision / named return values so that the threads referencing *Loop
146 /// have the correct pointer value.
147 /// https://github.com/ziglang/zig/issues/2761 and https://github.com/ziglang/zig/issues/2765
148 pub fn initMultiThreaded(self: *Loop) !void {
149 if (builtin.single_threaded)
150 @compileError("initMultiThreaded unavailable when building in single-threaded mode");
151 const core_count = try Thread.getCpuCount();
152 return self.initThreadPool(core_count);
153 }
154
155 /// Thread count is the total thread count. The thread pool size will be
156 /// max(thread_count - 1, 0)
157 pub fn initThreadPool(self: *Loop, thread_count: usize) !void {
158 self.* = Loop{
159 .arena = std.heap.ArenaAllocator.init(std.heap.page_allocator),
160 .pending_event_count = 1,
161 .os_data = undefined,
162 .next_tick_queue = std.atomic.Queue(anyframe).init(),
163 .extra_threads = undefined,
164 .available_eventfd_resume_nodes = std.atomic.Stack(ResumeNode.EventFd).init(),
165 .eventfd_resume_nodes = undefined,
166 .final_resume_node = ResumeNode{
167 .id = .stop,
168 .handle = undefined,
169 .overlapped = ResumeNode.overlapped_init,
170 },
171 .fs_end_request = .{ .data = .{ .msg = .end, .finish = .no_action } },
172 .fs_queue = std.atomic.Queue(Request).init(),
173 .fs_thread = undefined,
174 .fs_thread_wakeup = .{},
175 .delay_queue = undefined,
176 };
177 errdefer self.arena.deinit();
178
179 // We need at least one of these in case the fs thread wants to use onNextTick
180 const extra_thread_count = thread_count - 1;
181 const resume_node_count = @max(extra_thread_count, 1);
182 self.eventfd_resume_nodes = try self.arena.allocator().alloc(
183 std.atomic.Stack(ResumeNode.EventFd).Node,
184 resume_node_count,
185 );
186
187 self.extra_threads = try self.arena.allocator().alloc(Thread, extra_thread_count);
188
189 try self.initOsData(extra_thread_count);
190 errdefer self.deinitOsData();
191
192 if (!builtin.single_threaded) {
193 self.fs_thread = try Thread.spawn(.{}, posixFsRun, .{self});
194 }
195 errdefer if (!builtin.single_threaded) {
196 self.posixFsRequest(&self.fs_end_request);
197 self.fs_thread.join();
198 };
199
200 if (!builtin.single_threaded)
201 try self.delay_queue.init();
202 }
203
204 pub fn deinit(self: *Loop) void {
205 self.deinitOsData();
206 self.arena.deinit();
207 self.* = undefined;
208 }
209
210 const InitOsDataError = os.EpollCreateError || mem.Allocator.Error || os.EventFdError ||
211 Thread.SpawnError || os.EpollCtlError || os.KEventError ||
212 windows.CreateIoCompletionPortError;
213
214 const wakeup_bytes = [_]u8{0x1} ** 8;
215
216 fn initOsData(self: *Loop, extra_thread_count: usize) InitOsDataError!void {
217 nosuspend switch (builtin.os.tag) {
218 .linux => {
219 errdefer {
220 while (self.available_eventfd_resume_nodes.pop()) |node| os.close(node.data.eventfd);
221 }
222 for (self.eventfd_resume_nodes) |*eventfd_node| {
223 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
224 .data = ResumeNode.EventFd{
225 .base = ResumeNode{
226 .id = .event_fd,
227 .handle = undefined,
228 .overlapped = ResumeNode.overlapped_init,
229 },
230 .eventfd = try os.eventfd(1, os.linux.EFD.CLOEXEC | os.linux.EFD.NONBLOCK),
231 .epoll_op = os.linux.EPOLL.CTL_ADD,
232 },
233 .next = undefined,
234 };
235 self.available_eventfd_resume_nodes.push(eventfd_node);
236 }
237
238 self.os_data.epollfd = try os.epoll_create1(os.linux.EPOLL.CLOEXEC);
239 errdefer os.close(self.os_data.epollfd);
240
241 self.os_data.final_eventfd = try os.eventfd(0, os.linux.EFD.CLOEXEC | os.linux.EFD.NONBLOCK);
242 errdefer os.close(self.os_data.final_eventfd);
243
244 self.os_data.final_eventfd_event = os.linux.epoll_event{
245 .events = os.linux.EPOLL.IN,
246 .data = os.linux.epoll_data{ .ptr = @intFromPtr(&self.final_resume_node) },
247 };
248 try os.epoll_ctl(
249 self.os_data.epollfd,
250 os.linux.EPOLL.CTL_ADD,
251 self.os_data.final_eventfd,
252 &self.os_data.final_eventfd_event,
253 );
254
255 if (builtin.single_threaded) {
256 assert(extra_thread_count == 0);
257 return;
258 }
259
260 var extra_thread_index: usize = 0;
261 errdefer {
262 // writing 8 bytes to an eventfd cannot fail
263 const amt = os.write(self.os_data.final_eventfd, &wakeup_bytes) catch unreachable;
264 assert(amt == wakeup_bytes.len);
265 while (extra_thread_index != 0) {
266 extra_thread_index -= 1;
267 self.extra_threads[extra_thread_index].join();
268 }
269 }
270 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
271 self.extra_threads[extra_thread_index] = try Thread.spawn(.{}, workerRun, .{self});
272 }
273 },
274 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly => {
275 self.os_data.kqfd = try os.kqueue();
276 errdefer os.close(self.os_data.kqfd);
277
278 const empty_kevs = &[0]os.Kevent{};
279
280 for (self.eventfd_resume_nodes, 0..) |*eventfd_node, i| {
281 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
282 .data = ResumeNode.EventFd{
283 .base = ResumeNode{
284 .id = .event_fd,
285 .handle = undefined,
286 .overlapped = ResumeNode.overlapped_init,
287 },
288 // this one is for sending events
289 .kevent = os.Kevent{
290 .ident = i,
291 .filter = os.system.EVFILT_USER,
292 .flags = os.system.EV_CLEAR | os.system.EV_ADD | os.system.EV_DISABLE,
293 .fflags = 0,
294 .data = 0,
295 .udata = @intFromPtr(&eventfd_node.data.base),
296 },
297 },
298 .next = undefined,
299 };
300 self.available_eventfd_resume_nodes.push(eventfd_node);
301 const kevent_array = @as(*const [1]os.Kevent, &eventfd_node.data.kevent);
302 _ = try os.kevent(self.os_data.kqfd, kevent_array, empty_kevs, null);
303 eventfd_node.data.kevent.flags = os.system.EV_CLEAR | os.system.EV_ENABLE;
304 eventfd_node.data.kevent.fflags = os.system.NOTE_TRIGGER;
305 }
306
307 // Pre-add so that we cannot get error.SystemResources
308 // later when we try to activate it.
309 self.os_data.final_kevent = os.Kevent{
310 .ident = extra_thread_count,
311 .filter = os.system.EVFILT_USER,
312 .flags = os.system.EV_ADD | os.system.EV_DISABLE,
313 .fflags = 0,
314 .data = 0,
315 .udata = @intFromPtr(&self.final_resume_node),
316 };
317 const final_kev_arr = @as(*const [1]os.Kevent, &self.os_data.final_kevent);
318 _ = try os.kevent(self.os_data.kqfd, final_kev_arr, empty_kevs, null);
319 self.os_data.final_kevent.flags = os.system.EV_ENABLE;
320 self.os_data.final_kevent.fflags = os.system.NOTE_TRIGGER;
321
322 if (builtin.single_threaded) {
323 assert(extra_thread_count == 0);
324 return;
325 }
326
327 var extra_thread_index: usize = 0;
328 errdefer {
329 _ = os.kevent(self.os_data.kqfd, final_kev_arr, empty_kevs, null) catch unreachable;
330 while (extra_thread_index != 0) {
331 extra_thread_index -= 1;
332 self.extra_threads[extra_thread_index].join();
333 }
334 }
335 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
336 self.extra_threads[extra_thread_index] = try Thread.spawn(.{}, workerRun, .{self});
337 }
338 },
339 .openbsd => {
340 self.os_data.kqfd = try os.kqueue();
341 errdefer os.close(self.os_data.kqfd);
342
343 const empty_kevs = &[0]os.Kevent{};
344
345 for (self.eventfd_resume_nodes, 0..) |*eventfd_node, i| {
346 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
347 .data = ResumeNode.EventFd{
348 .base = ResumeNode{
349 .id = .event_fd,
350 .handle = undefined,
351 .overlapped = ResumeNode.overlapped_init,
352 },
353 // this one is for sending events
354 .kevent = os.Kevent{
355 .ident = i,
356 .filter = os.system.EVFILT_TIMER,
357 .flags = os.system.EV_CLEAR | os.system.EV_ADD | os.system.EV_DISABLE | os.system.EV_ONESHOT,
358 .fflags = 0,
359 .data = 0,
360 .udata = @intFromPtr(&eventfd_node.data.base),
361 },
362 },
363 .next = undefined,
364 };
365 self.available_eventfd_resume_nodes.push(eventfd_node);
366 const kevent_array = @as(*const [1]os.Kevent, &eventfd_node.data.kevent);
367 _ = try os.kevent(self.os_data.kqfd, kevent_array, empty_kevs, null);
368 eventfd_node.data.kevent.flags = os.system.EV_CLEAR | os.system.EV_ENABLE;
369 }
370
371 // Pre-add so that we cannot get error.SystemResources
372 // later when we try to activate it.
373 self.os_data.final_kevent = os.Kevent{
374 .ident = extra_thread_count,
375 .filter = os.system.EVFILT_TIMER,
376 .flags = os.system.EV_ADD | os.system.EV_ONESHOT | os.system.EV_DISABLE,
377 .fflags = 0,
378 .data = 0,
379 .udata = @intFromPtr(&self.final_resume_node),
380 };
381 const final_kev_arr = @as(*const [1]os.Kevent, &self.os_data.final_kevent);
382 _ = try os.kevent(self.os_data.kqfd, final_kev_arr, empty_kevs, null);
383 self.os_data.final_kevent.flags = os.system.EV_ENABLE;
384
385 if (builtin.single_threaded) {
386 assert(extra_thread_count == 0);
387 return;
388 }
389
390 var extra_thread_index: usize = 0;
391 errdefer {
392 _ = os.kevent(self.os_data.kqfd, final_kev_arr, empty_kevs, null) catch unreachable;
393 while (extra_thread_index != 0) {
394 extra_thread_index -= 1;
395 self.extra_threads[extra_thread_index].join();
396 }
397 }
398 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
399 self.extra_threads[extra_thread_index] = try Thread.spawn(.{}, workerRun, .{self});
400 }
401 },
402 .windows => {
403 self.os_data.io_port = try windows.CreateIoCompletionPort(
404 windows.INVALID_HANDLE_VALUE,
405 null,
406 undefined,
407 maxInt(windows.DWORD),
408 );
409 errdefer windows.CloseHandle(self.os_data.io_port);
410
411 for (self.eventfd_resume_nodes) |*eventfd_node| {
412 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
413 .data = ResumeNode.EventFd{
414 .base = ResumeNode{
415 .id = .event_fd,
416 .handle = undefined,
417 .overlapped = ResumeNode.overlapped_init,
418 },
419 // this one is for sending events
420 .completion_key = @intFromPtr(&eventfd_node.data.base),
421 },
422 .next = undefined,
423 };
424 self.available_eventfd_resume_nodes.push(eventfd_node);
425 }
426
427 if (builtin.single_threaded) {
428 assert(extra_thread_count == 0);
429 return;
430 }
431
432 var extra_thread_index: usize = 0;
433 errdefer {
434 var i: usize = 0;
435 while (i < extra_thread_index) : (i += 1) {
436 while (true) {
437 const overlapped = &self.final_resume_node.overlapped;
438 windows.PostQueuedCompletionStatus(self.os_data.io_port, undefined, undefined, overlapped) catch continue;
439 break;
440 }
441 }
442 while (extra_thread_index != 0) {
443 extra_thread_index -= 1;
444 self.extra_threads[extra_thread_index].join();
445 }
446 }
447 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
448 self.extra_threads[extra_thread_index] = try Thread.spawn(.{}, workerRun, .{self});
449 }
450 },
451 else => {},
452 };
453 }
454
455 fn deinitOsData(self: *Loop) void {
456 nosuspend switch (builtin.os.tag) {
457 .linux => {
458 os.close(self.os_data.final_eventfd);
459 while (self.available_eventfd_resume_nodes.pop()) |node| os.close(node.data.eventfd);
460 os.close(self.os_data.epollfd);
461 },
462 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
463 os.close(self.os_data.kqfd);
464 },
465 .windows => {
466 windows.CloseHandle(self.os_data.io_port);
467 },
468 else => {},
469 };
470 }
471
472 /// resume_node must live longer than the anyframe that it holds a reference to.
473 /// flags must contain EPOLLET
474 pub fn linuxAddFd(self: *Loop, fd: i32, resume_node: *ResumeNode, flags: u32) !void {
475 assert(flags & os.linux.EPOLL.ET == os.linux.EPOLL.ET);
476 self.beginOneEvent();
477 errdefer self.finishOneEvent();
478 try self.linuxModFd(
479 fd,
480 os.linux.EPOLL.CTL_ADD,
481 flags,
482 resume_node,
483 );
484 }
485
486 pub fn linuxModFd(self: *Loop, fd: i32, op: u32, flags: u32, resume_node: *ResumeNode) !void {
487 assert(flags & os.linux.EPOLL.ET == os.linux.EPOLL.ET);
488 var ev = os.linux.epoll_event{
489 .events = flags,
490 .data = os.linux.epoll_data{ .ptr = @intFromPtr(resume_node) },
491 };
492 try os.epoll_ctl(self.os_data.epollfd, op, fd, &ev);
493 }
494
495 pub fn linuxRemoveFd(self: *Loop, fd: i32) void {
496 os.epoll_ctl(self.os_data.epollfd, os.linux.EPOLL.CTL_DEL, fd, null) catch {};
497 self.finishOneEvent();
498 }
499
500 pub fn linuxWaitFd(self: *Loop, fd: i32, flags: u32) void {
501 assert(flags & os.linux.EPOLL.ET == os.linux.EPOLL.ET);
502 assert(flags & os.linux.EPOLL.ONESHOT == os.linux.EPOLL.ONESHOT);
503 var resume_node = ResumeNode.Basic{
504 .base = ResumeNode{
505 .id = .basic,
506 .handle = @frame(),
507 .overlapped = ResumeNode.overlapped_init,
508 },
509 };
510 var need_to_delete = true;
511 defer if (need_to_delete) self.linuxRemoveFd(fd);
512
513 suspend {
514 self.linuxAddFd(fd, &resume_node.base, flags) catch |err| switch (err) {
515 error.FileDescriptorNotRegistered => unreachable,
516 error.OperationCausesCircularLoop => unreachable,
517 error.FileDescriptorIncompatibleWithEpoll => unreachable,
518 error.FileDescriptorAlreadyPresentInSet => unreachable, // evented writes to the same fd is not thread-safe
519
520 error.SystemResources,
521 error.UserResourceLimitReached,
522 error.Unexpected,
523 => {
524 need_to_delete = false;
525 // Fall back to a blocking poll(). Ideally this codepath is never hit, since
526 // epoll should be just fine. But this is better than incorrect behavior.
527 var poll_flags: i16 = 0;
528 if ((flags & os.linux.EPOLL.IN) != 0) poll_flags |= os.POLL.IN;
529 if ((flags & os.linux.EPOLL.OUT) != 0) poll_flags |= os.POLL.OUT;
530 var pfd = [1]os.pollfd{os.pollfd{
531 .fd = fd,
532 .events = poll_flags,
533 .revents = undefined,
534 }};
535 _ = os.poll(&pfd, -1) catch |poll_err| switch (poll_err) {
536 error.NetworkSubsystemFailed => unreachable, // only possible on windows
537
538 error.SystemResources,
539 error.Unexpected,
540 => {
541 // Even poll() didn't work. The best we can do now is sleep for a
542 // small duration and then hope that something changed.
543 std.time.sleep(1 * std.time.ns_per_ms);
544 },
545 };
546 resume @frame();
547 },
548 };
549 }
550 }
551
552 pub fn waitUntilFdReadable(self: *Loop, fd: os.fd_t) void {
553 switch (builtin.os.tag) {
554 .linux => {
555 self.linuxWaitFd(fd, os.linux.EPOLL.ET | os.linux.EPOLL.ONESHOT | os.linux.EPOLL.IN);
556 },
557 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
558 self.bsdWaitKev(@as(usize, @intCast(fd)), os.system.EVFILT_READ, os.system.EV_ONESHOT);
559 },
560 else => @compileError("Unsupported OS"),
561 }
562 }
563
564 pub fn waitUntilFdWritable(self: *Loop, fd: os.fd_t) void {
565 switch (builtin.os.tag) {
566 .linux => {
567 self.linuxWaitFd(fd, os.linux.EPOLL.ET | os.linux.EPOLL.ONESHOT | os.linux.EPOLL.OUT);
568 },
569 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
570 self.bsdWaitKev(@as(usize, @intCast(fd)), os.system.EVFILT_WRITE, os.system.EV_ONESHOT);
571 },
572 else => @compileError("Unsupported OS"),
573 }
574 }
575
576 pub fn waitUntilFdWritableOrReadable(self: *Loop, fd: os.fd_t) void {
577 switch (builtin.os.tag) {
578 .linux => {
579 self.linuxWaitFd(fd, os.linux.EPOLL.ET | os.linux.EPOLL.ONESHOT | os.linux.EPOLL.OUT | os.linux.EPOLL.IN);
580 },
581 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
582 self.bsdWaitKev(@as(usize, @intCast(fd)), os.system.EVFILT_READ, os.system.EV_ONESHOT);
583 self.bsdWaitKev(@as(usize, @intCast(fd)), os.system.EVFILT_WRITE, os.system.EV_ONESHOT);
584 },
585 else => @compileError("Unsupported OS"),
586 }
587 }
588
589 pub fn bsdWaitKev(self: *Loop, ident: usize, filter: i16, flags: u16) void {
590 var resume_node = ResumeNode.Basic{
591 .base = ResumeNode{
592 .id = .basic,
593 .handle = @frame(),
594 .overlapped = ResumeNode.overlapped_init,
595 },
596 .kev = undefined,
597 };
598
599 defer {
600 // If the kevent was set to be ONESHOT, it doesn't need to be deleted manually.
601 if (flags & os.system.EV_ONESHOT != 0) {
602 self.bsdRemoveKev(ident, filter);
603 }
604 }
605
606 suspend {
607 self.bsdAddKev(&resume_node, ident, filter, flags) catch unreachable;
608 }
609 }
610
611 /// resume_node must live longer than the anyframe that it holds a reference to.
612 pub fn bsdAddKev(self: *Loop, resume_node: *ResumeNode.Basic, ident: usize, filter: i16, flags: u16) !void {
613 self.beginOneEvent();
614 errdefer self.finishOneEvent();
615 var kev = [1]os.Kevent{os.Kevent{
616 .ident = ident,
617 .filter = filter,
618 .flags = os.system.EV_ADD | os.system.EV_ENABLE | os.system.EV_CLEAR | flags,
619 .fflags = 0,
620 .data = 0,
621 .udata = @intFromPtr(&resume_node.base),
622 }};
623 const empty_kevs = &[0]os.Kevent{};
624 _ = try os.kevent(self.os_data.kqfd, &kev, empty_kevs, null);
625 }
626
627 pub fn bsdRemoveKev(self: *Loop, ident: usize, filter: i16) void {
628 var kev = [1]os.Kevent{os.Kevent{
629 .ident = ident,
630 .filter = filter,
631 .flags = os.system.EV_DELETE,
632 .fflags = 0,
633 .data = 0,
634 .udata = 0,
635 }};
636 const empty_kevs = &[0]os.Kevent{};
637 _ = os.kevent(self.os_data.kqfd, &kev, empty_kevs, null) catch undefined;
638 self.finishOneEvent();
639 }
640
641 fn dispatch(self: *Loop) void {
642 while (self.available_eventfd_resume_nodes.pop()) |resume_stack_node| {
643 const next_tick_node = self.next_tick_queue.get() orelse {
644 self.available_eventfd_resume_nodes.push(resume_stack_node);
645 return;
646 };
647 const eventfd_node = &resume_stack_node.data;
648 eventfd_node.base.handle = next_tick_node.data;
649 switch (builtin.os.tag) {
650 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
651 const kevent_array = @as(*const [1]os.Kevent, &eventfd_node.kevent);
652 const empty_kevs = &[0]os.Kevent{};
653 _ = os.kevent(self.os_data.kqfd, kevent_array, empty_kevs, null) catch {
654 self.next_tick_queue.unget(next_tick_node);
655 self.available_eventfd_resume_nodes.push(resume_stack_node);
656 return;
657 };
658 },
659 .linux => {
660 // the pending count is already accounted for
661 const epoll_events = os.linux.EPOLL.ONESHOT | os.linux.EPOLL.IN | os.linux.EPOLL.OUT |
662 os.linux.EPOLL.ET;
663 self.linuxModFd(
664 eventfd_node.eventfd,
665 eventfd_node.epoll_op,
666 epoll_events,
667 &eventfd_node.base,
668 ) catch {
669 self.next_tick_queue.unget(next_tick_node);
670 self.available_eventfd_resume_nodes.push(resume_stack_node);
671 return;
672 };
673 },
674 .windows => {
675 windows.PostQueuedCompletionStatus(
676 self.os_data.io_port,
677 undefined,
678 undefined,
679 &eventfd_node.base.overlapped,
680 ) catch {
681 self.next_tick_queue.unget(next_tick_node);
682 self.available_eventfd_resume_nodes.push(resume_stack_node);
683 return;
684 };
685 },
686 else => @compileError("unsupported OS"),
687 }
688 }
689 }
690
691 /// Bring your own linked list node. This means it can't fail.
692 pub fn onNextTick(self: *Loop, node: *NextTickNode) void {
693 self.beginOneEvent(); // finished in dispatch()
694 self.next_tick_queue.put(node);
695 self.dispatch();
696 }
697
698 pub fn cancelOnNextTick(self: *Loop, node: *NextTickNode) void {
699 if (self.next_tick_queue.remove(node)) {
700 self.finishOneEvent();
701 }
702 }
703
704 pub fn run(self: *Loop) void {
705 self.finishOneEvent(); // the reference we start with
706
707 self.workerRun();
708
709 if (!builtin.single_threaded) {
710 switch (builtin.os.tag) {
711 .linux,
712 .macos,
713 .ios,
714 .tvos,
715 .watchos,
716 .freebsd,
717 .netbsd,
718 .dragonfly,
719 .openbsd,
720 => self.fs_thread.join(),
721 else => {},
722 }
723 }
724
725 for (self.extra_threads) |extra_thread| {
726 extra_thread.join();
727 }
728
729 self.delay_queue.deinit();
730 }
731
732 /// Runs the provided function asynchronously. The function's frame is allocated
733 /// with `allocator` and freed when the function returns.
734 /// `func` must return void and it can be an async function.
735 /// Yields to the event loop, running the function on the next tick.
736 pub fn runDetached(self: *Loop, alloc: mem.Allocator, comptime func: anytype, args: anytype) error{OutOfMemory}!void {
737 if (!std.io.is_async) @compileError("Can't use runDetached in non-async mode!");
738 if (@TypeOf(@call(.{}, func, args)) != void) {
739 @compileError("`func` must not have a return value");
740 }
741
742 const Wrapper = struct {
743 const Args = @TypeOf(args);
744 fn run(func_args: Args, loop: *Loop, allocator: mem.Allocator) void {
745 loop.beginOneEvent();
746 loop.yield();
747 @call(.{}, func, func_args); // compile error when called with non-void ret type
748 suspend {
749 loop.finishOneEvent();
750 allocator.destroy(@frame());
751 }
752 }
753 };
754
755 const run_frame = try alloc.create(@Frame(Wrapper.run));
756 run_frame.* = async Wrapper.run(args, self, alloc);
757 }
758
759 /// Yielding lets the event loop run, starting any unstarted async operations.
760 /// Note that async operations automatically start when a function yields for any other reason,
761 /// for example, when async I/O is performed. This function is intended to be used only when
762 /// CPU bound tasks would be waiting in the event loop but never get started because no async I/O
763 /// is performed.
764 pub fn yield(self: *Loop) void {
765 suspend {
766 var my_tick_node = NextTickNode{
767 .prev = undefined,
768 .next = undefined,
769 .data = @frame(),
770 };
771 self.onNextTick(&my_tick_node);
772 }
773 }
774
775 /// If the build is multi-threaded and there is an event loop, then it calls `yield`. Otherwise,
776 /// does nothing.
777 pub fn startCpuBoundOperation() void {
778 if (builtin.single_threaded) {
779 return;
780 } else if (instance) |event_loop| {
781 event_loop.yield();
782 }
783 }
784
785 /// call finishOneEvent when done
786 pub fn beginOneEvent(self: *Loop) void {
787 _ = @atomicRmw(usize, &self.pending_event_count, .Add, 1, .SeqCst);
788 }
789
790 pub fn finishOneEvent(self: *Loop) void {
791 nosuspend {
792 const prev = @atomicRmw(usize, &self.pending_event_count, .Sub, 1, .SeqCst);
793 if (prev != 1) return;
794
795 // cause all the threads to stop
796 self.posixFsRequest(&self.fs_end_request);
797
798 switch (builtin.os.tag) {
799 .linux => {
800 // writing to the eventfd will only wake up one thread, thus multiple writes
801 // are needed to wakeup all the threads
802 var i: usize = 0;
803 while (i < self.extra_threads.len + 1) : (i += 1) {
804 // writing 8 bytes to an eventfd cannot fail
805 const amt = os.write(self.os_data.final_eventfd, &wakeup_bytes) catch unreachable;
806 assert(amt == wakeup_bytes.len);
807 }
808 return;
809 },
810 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
811 const final_kevent = @as(*const [1]os.Kevent, &self.os_data.final_kevent);
812 const empty_kevs = &[0]os.Kevent{};
813 // cannot fail because we already added it and this just enables it
814 _ = os.kevent(self.os_data.kqfd, final_kevent, empty_kevs, null) catch unreachable;
815 return;
816 },
817 .windows => {
818 var i: usize = 0;
819 while (i < self.extra_threads.len + 1) : (i += 1) {
820 while (true) {
821 const overlapped = &self.final_resume_node.overlapped;
822 windows.PostQueuedCompletionStatus(self.os_data.io_port, undefined, undefined, overlapped) catch continue;
823 break;
824 }
825 }
826 return;
827 },
828 else => @compileError("unsupported OS"),
829 }
830 }
831 }
832
833 pub fn sleep(self: *Loop, nanoseconds: u64) void {
834 if (builtin.single_threaded)
835 @compileError("TODO: integrate timers with epoll/kevent/iocp for single-threaded");
836
837 suspend {
838 const now = self.delay_queue.timer.read();
839
840 var entry: DelayQueue.Waiters.Entry = undefined;
841 entry.init(@frame(), now + nanoseconds);
842 self.delay_queue.waiters.insert(&entry);
843
844 // Speculatively wake up the timer thread when we add a new entry.
845 // If the timer thread is sleeping on a longer entry, we need to
846 // interrupt it so that our entry can be expired in time.
847 self.delay_queue.event.set();
848 }
849 }
850
851 const DelayQueue = struct {
852 timer: std.time.Timer,
853 waiters: Waiters,
854 thread: std.Thread,
855 event: std.Thread.ResetEvent,
856 is_running: std.atomic.Value(bool),
857
858 /// Initialize the delay queue by spawning the timer thread
859 /// and starting any timer resources.
860 fn init(self: *DelayQueue) !void {
861 self.* = DelayQueue{
862 .timer = try std.time.Timer.start(),
863 .waiters = DelayQueue.Waiters{
864 .entries = std.atomic.Queue(anyframe).init(),
865 },
866 .thread = undefined,
867 .event = .{},
868 .is_running = std.atomic.Value(bool).init(true),
869 };
870
871 // Must be after init so that it can read the other state, such as `is_running`.
872 self.thread = try std.Thread.spawn(.{}, DelayQueue.run, .{self});
873 }
874
875 fn deinit(self: *DelayQueue) void {
876 self.is_running.store(false, .SeqCst);
877 self.event.set();
878 self.thread.join();
879 }
880
881 /// Entry point for the timer thread
882 /// which waits for timer entries to expire and reschedules them.
883 fn run(self: *DelayQueue) void {
884 const loop = @fieldParentPtr(Loop, "delay_queue", self);
885
886 while (self.is_running.load(.SeqCst)) {
887 self.event.reset();
888 const now = self.timer.read();
889
890 if (self.waiters.popExpired(now)) |entry| {
891 loop.onNextTick(&entry.node);
892 continue;
893 }
894
895 if (self.waiters.nextExpire()) |expires| {
896 if (now >= expires)
897 continue;
898 self.event.timedWait(expires - now) catch {};
899 } else {
900 self.event.wait();
901 }
902 }
903 }
904
905 // TODO: use a tickless hierarchical timer wheel:
906 // https://github.com/wahern/timeout/
907 const Waiters = struct {
908 entries: std.atomic.Queue(anyframe),
909
910 const Entry = struct {
911 node: NextTickNode,
912 expires: u64,
913
914 fn init(self: *Entry, frame: anyframe, expires: u64) void {
915 self.node.data = frame;
916 self.expires = expires;
917 }
918 };
919
920 /// Registers the entry into the queue of waiting frames
921 fn insert(self: *Waiters, entry: *Entry) void {
922 self.entries.put(&entry.node);
923 }
924
925 /// Dequeues one expired event relative to `now`
926 fn popExpired(self: *Waiters, now: u64) ?*Entry {
927 const entry = self.peekExpiringEntry() orelse return null;
928 if (entry.expires > now)
929 return null;
930
931 assert(self.entries.remove(&entry.node));
932 return entry;
933 }
934
935 /// Returns an estimate for the amount of time
936 /// to wait until the next waiting entry expires.
937 fn nextExpire(self: *Waiters) ?u64 {
938 const entry = self.peekExpiringEntry() orelse return null;
939 return entry.expires;
940 }
941
942 fn peekExpiringEntry(self: *Waiters) ?*Entry {
943 self.entries.mutex.lock();
944 defer self.entries.mutex.unlock();
945
946 // starting from the head
947 var head = self.entries.head orelse return null;
948
949 // traverse the list of waiting entries to
950 // find the Node with the smallest `expires` field
951 var min = head;
952 while (head.next) |node| {
953 const minEntry = @fieldParentPtr(Entry, "node", min);
954 const nodeEntry = @fieldParentPtr(Entry, "node", node);
955 if (nodeEntry.expires < minEntry.expires)
956 min = node;
957 head = node;
958 }
959
960 return @fieldParentPtr(Entry, "node", min);
961 }
962 };
963 };
964
965 /// ------- I/0 APIs -------
966 pub fn accept(
967 self: *Loop,
968 /// This argument is a socket that has been created with `socket`, bound to a local address
969 /// with `bind`, and is listening for connections after a `listen`.
970 sockfd: os.socket_t,
971 /// This argument is a pointer to a sockaddr structure. This structure is filled in with the
972 /// address of the peer socket, as known to the communications layer. The exact format of the
973 /// address returned addr is determined by the socket's address family (see `socket` and the
974 /// respective protocol man pages).
975 addr: *os.sockaddr,
976 /// This argument is a value-result argument: the caller must initialize it to contain the
977 /// size (in bytes) of the structure pointed to by addr; on return it will contain the actual size
978 /// of the peer address.
979 ///
980 /// The returned address is truncated if the buffer provided is too small; in this case, `addr_size`
981 /// will return a value greater than was supplied to the call.
982 addr_size: *os.socklen_t,
983 /// The following values can be bitwise ORed in flags to obtain different behavior:
984 /// * `SOCK.CLOEXEC` - Set the close-on-exec (`FD_CLOEXEC`) flag on the new file descriptor. See the
985 /// description of the `O.CLOEXEC` flag in `open` for reasons why this may be useful.
986 flags: u32,
987 ) os.AcceptError!os.socket_t {
988 while (true) {
989 return os.accept(sockfd, addr, addr_size, flags | os.SOCK.NONBLOCK) catch |err| switch (err) {
990 error.WouldBlock => {
991 self.waitUntilFdReadable(sockfd);
992 continue;
993 },
994 else => return err,
995 };
996 }
997 }
998
999 pub fn connect(self: *Loop, sockfd: os.socket_t, sock_addr: *const os.sockaddr, len: os.socklen_t) os.ConnectError!void {
1000 os.connect(sockfd, sock_addr, len) catch |err| switch (err) {
1001 error.WouldBlock => {
1002 self.waitUntilFdWritable(sockfd);
1003 return os.getsockoptError(sockfd);
1004 },
1005 else => return err,
1006 };
1007 }
1008
1009 /// Performs an async `os.open` using a separate thread.
1010 pub fn openZ(self: *Loop, file_path: [*:0]const u8, flags: u32, mode: os.mode_t) os.OpenError!os.fd_t {
1011 var req_node = Request.Node{
1012 .data = .{
1013 .msg = .{
1014 .open = .{
1015 .path = file_path,
1016 .flags = flags,
1017 .mode = mode,
1018 .result = undefined,
1019 },
1020 },
1021 .finish = .{ .tick_node = .{ .data = @frame() } },
1022 },
1023 };
1024 suspend {
1025 self.posixFsRequest(&req_node);
1026 }
1027 return req_node.data.msg.open.result;
1028 }
1029
1030 /// Performs an async `os.opent` using a separate thread.
1031 pub fn openatZ(self: *Loop, fd: os.fd_t, file_path: [*:0]const u8, flags: u32, mode: os.mode_t) os.OpenError!os.fd_t {
1032 var req_node = Request.Node{
1033 .data = .{
1034 .msg = .{
1035 .openat = .{
1036 .fd = fd,
1037 .path = file_path,
1038 .flags = flags,
1039 .mode = mode,
1040 .result = undefined,
1041 },
1042 },
1043 .finish = .{ .tick_node = .{ .data = @frame() } },
1044 },
1045 };
1046 suspend {
1047 self.posixFsRequest(&req_node);
1048 }
1049 return req_node.data.msg.openat.result;
1050 }
1051
1052 /// Performs an async `os.close` using a separate thread.
1053 pub fn close(self: *Loop, fd: os.fd_t) void {
1054 var req_node = Request.Node{
1055 .data = .{
1056 .msg = .{ .close = .{ .fd = fd } },
1057 .finish = .{ .tick_node = .{ .data = @frame() } },
1058 },
1059 };
1060 suspend {
1061 self.posixFsRequest(&req_node);
1062 }
1063 }
1064
1065 /// Performs an async `os.read` using a separate thread.
1066 /// `fd` must block and not return EAGAIN.
1067 pub fn read(self: *Loop, fd: os.fd_t, buf: []u8, simulate_evented: bool) os.ReadError!usize {
1068 if (simulate_evented) {
1069 var req_node = Request.Node{
1070 .data = .{
1071 .msg = .{
1072 .read = .{
1073 .fd = fd,
1074 .buf = buf,
1075 .result = undefined,
1076 },
1077 },
1078 .finish = .{ .tick_node = .{ .data = @frame() } },
1079 },
1080 };
1081 suspend {
1082 self.posixFsRequest(&req_node);
1083 }
1084 return req_node.data.msg.read.result;
1085 } else {
1086 while (true) {
1087 return os.read(fd, buf) catch |err| switch (err) {
1088 error.WouldBlock => {
1089 self.waitUntilFdReadable(fd);
1090 continue;
1091 },
1092 else => return err,
1093 };
1094 }
1095 }
1096 }
1097
1098 /// Performs an async `os.readv` using a separate thread.
1099 /// `fd` must block and not return EAGAIN.
1100 pub fn readv(self: *Loop, fd: os.fd_t, iov: []const os.iovec, simulate_evented: bool) os.ReadError!usize {
1101 if (simulate_evented) {
1102 var req_node = Request.Node{
1103 .data = .{
1104 .msg = .{
1105 .readv = .{
1106 .fd = fd,
1107 .iov = iov,
1108 .result = undefined,
1109 },
1110 },
1111 .finish = .{ .tick_node = .{ .data = @frame() } },
1112 },
1113 };
1114 suspend {
1115 self.posixFsRequest(&req_node);
1116 }
1117 return req_node.data.msg.readv.result;
1118 } else {
1119 while (true) {
1120 return os.readv(fd, iov) catch |err| switch (err) {
1121 error.WouldBlock => {
1122 self.waitUntilFdReadable(fd);
1123 continue;
1124 },
1125 else => return err,
1126 };
1127 }
1128 }
1129 }
1130
1131 /// Performs an async `os.pread` using a separate thread.
1132 /// `fd` must block and not return EAGAIN.
1133 pub fn pread(self: *Loop, fd: os.fd_t, buf: []u8, offset: u64, simulate_evented: bool) os.PReadError!usize {
1134 if (simulate_evented) {
1135 var req_node = Request.Node{
1136 .data = .{
1137 .msg = .{
1138 .pread = .{
1139 .fd = fd,
1140 .buf = buf,
1141 .offset = offset,
1142 .result = undefined,
1143 },
1144 },
1145 .finish = .{ .tick_node = .{ .data = @frame() } },
1146 },
1147 };
1148 suspend {
1149 self.posixFsRequest(&req_node);
1150 }
1151 return req_node.data.msg.pread.result;
1152 } else {
1153 while (true) {
1154 return os.pread(fd, buf, offset) catch |err| switch (err) {
1155 error.WouldBlock => {
1156 self.waitUntilFdReadable(fd);
1157 continue;
1158 },
1159 else => return err,
1160 };
1161 }
1162 }
1163 }
1164
1165 /// Performs an async `os.preadv` using a separate thread.
1166 /// `fd` must block and not return EAGAIN.
1167 pub fn preadv(self: *Loop, fd: os.fd_t, iov: []const os.iovec, offset: u64, simulate_evented: bool) os.ReadError!usize {
1168 if (simulate_evented) {
1169 var req_node = Request.Node{
1170 .data = .{
1171 .msg = .{
1172 .preadv = .{
1173 .fd = fd,
1174 .iov = iov,
1175 .offset = offset,
1176 .result = undefined,
1177 },
1178 },
1179 .finish = .{ .tick_node = .{ .data = @frame() } },
1180 },
1181 };
1182 suspend {
1183 self.posixFsRequest(&req_node);
1184 }
1185 return req_node.data.msg.preadv.result;
1186 } else {
1187 while (true) {
1188 return os.preadv(fd, iov, offset) catch |err| switch (err) {
1189 error.WouldBlock => {
1190 self.waitUntilFdReadable(fd);
1191 continue;
1192 },
1193 else => return err,
1194 };
1195 }
1196 }
1197 }
1198
1199 /// Performs an async `os.write` using a separate thread.
1200 /// `fd` must block and not return EAGAIN.
1201 pub fn write(self: *Loop, fd: os.fd_t, bytes: []const u8, simulate_evented: bool) os.WriteError!usize {
1202 if (simulate_evented) {
1203 var req_node = Request.Node{
1204 .data = .{
1205 .msg = .{
1206 .write = .{
1207 .fd = fd,
1208 .bytes = bytes,
1209 .result = undefined,
1210 },
1211 },
1212 .finish = .{ .tick_node = .{ .data = @frame() } },
1213 },
1214 };
1215 suspend {
1216 self.posixFsRequest(&req_node);
1217 }
1218 return req_node.data.msg.write.result;
1219 } else {
1220 while (true) {
1221 return os.write(fd, bytes) catch |err| switch (err) {
1222 error.WouldBlock => {
1223 self.waitUntilFdWritable(fd);
1224 continue;
1225 },
1226 else => return err,
1227 };
1228 }
1229 }
1230 }
1231
1232 /// Performs an async `os.writev` using a separate thread.
1233 /// `fd` must block and not return EAGAIN.
1234 pub fn writev(self: *Loop, fd: os.fd_t, iov: []const os.iovec_const, simulate_evented: bool) os.WriteError!usize {
1235 if (simulate_evented) {
1236 var req_node = Request.Node{
1237 .data = .{
1238 .msg = .{
1239 .writev = .{
1240 .fd = fd,
1241 .iov = iov,
1242 .result = undefined,
1243 },
1244 },
1245 .finish = .{ .tick_node = .{ .data = @frame() } },
1246 },
1247 };
1248 suspend {
1249 self.posixFsRequest(&req_node);
1250 }
1251 return req_node.data.msg.writev.result;
1252 } else {
1253 while (true) {
1254 return os.writev(fd, iov) catch |err| switch (err) {
1255 error.WouldBlock => {
1256 self.waitUntilFdWritable(fd);
1257 continue;
1258 },
1259 else => return err,
1260 };
1261 }
1262 }
1263 }
1264
1265 /// Performs an async `os.pwrite` using a separate thread.
1266 /// `fd` must block and not return EAGAIN.
1267 pub fn pwrite(self: *Loop, fd: os.fd_t, bytes: []const u8, offset: u64, simulate_evented: bool) os.PerformsWriteError!usize {
1268 if (simulate_evented) {
1269 var req_node = Request.Node{
1270 .data = .{
1271 .msg = .{
1272 .pwrite = .{
1273 .fd = fd,
1274 .bytes = bytes,
1275 .offset = offset,
1276 .result = undefined,
1277 },
1278 },
1279 .finish = .{ .tick_node = .{ .data = @frame() } },
1280 },
1281 };
1282 suspend {
1283 self.posixFsRequest(&req_node);
1284 }
1285 return req_node.data.msg.pwrite.result;
1286 } else {
1287 while (true) {
1288 return os.pwrite(fd, bytes, offset) catch |err| switch (err) {
1289 error.WouldBlock => {
1290 self.waitUntilFdWritable(fd);
1291 continue;
1292 },
1293 else => return err,
1294 };
1295 }
1296 }
1297 }
1298
1299 /// Performs an async `os.pwritev` using a separate thread.
1300 /// `fd` must block and not return EAGAIN.
1301 pub fn pwritev(self: *Loop, fd: os.fd_t, iov: []const os.iovec_const, offset: u64, simulate_evented: bool) os.PWriteError!usize {
1302 if (simulate_evented) {
1303 var req_node = Request.Node{
1304 .data = .{
1305 .msg = .{
1306 .pwritev = .{
1307 .fd = fd,
1308 .iov = iov,
1309 .offset = offset,
1310 .result = undefined,
1311 },
1312 },
1313 .finish = .{ .tick_node = .{ .data = @frame() } },
1314 },
1315 };
1316 suspend {
1317 self.posixFsRequest(&req_node);
1318 }
1319 return req_node.data.msg.pwritev.result;
1320 } else {
1321 while (true) {
1322 return os.pwritev(fd, iov, offset) catch |err| switch (err) {
1323 error.WouldBlock => {
1324 self.waitUntilFdWritable(fd);
1325 continue;
1326 },
1327 else => return err,
1328 };
1329 }
1330 }
1331 }
1332
1333 pub fn sendto(
1334 self: *Loop,
1335 /// The file descriptor of the sending socket.
1336 sockfd: os.fd_t,
1337 /// Message to send.
1338 buf: []const u8,
1339 flags: u32,
1340 dest_addr: ?*const os.sockaddr,
1341 addrlen: os.socklen_t,
1342 ) os.SendToError!usize {
1343 while (true) {
1344 return os.sendto(sockfd, buf, flags, dest_addr, addrlen) catch |err| switch (err) {
1345 error.WouldBlock => {
1346 self.waitUntilFdWritable(sockfd);
1347 continue;
1348 },
1349 else => return err,
1350 };
1351 }
1352 }
1353
1354 pub fn recvfrom(
1355 self: *Loop,
1356 sockfd: os.fd_t,
1357 buf: []u8,
1358 flags: u32,
1359 src_addr: ?*os.sockaddr,
1360 addrlen: ?*os.socklen_t,
1361 ) os.RecvFromError!usize {
1362 while (true) {
1363 return os.recvfrom(sockfd, buf, flags, src_addr, addrlen) catch |err| switch (err) {
1364 error.WouldBlock => {
1365 self.waitUntilFdReadable(sockfd);
1366 continue;
1367 },
1368 else => return err,
1369 };
1370 }
1371 }
1372
1373 /// Performs an async `os.faccessatZ` using a separate thread.
1374 /// `fd` must block and not return EAGAIN.
1375 pub fn faccessatZ(
1376 self: *Loop,
1377 dirfd: os.fd_t,
1378 path_z: [*:0]const u8,
1379 mode: u32,
1380 flags: u32,
1381 ) os.AccessError!void {
1382 var req_node = Request.Node{
1383 .data = .{
1384 .msg = .{
1385 .faccessat = .{
1386 .dirfd = dirfd,
1387 .path = path_z,
1388 .mode = mode,
1389 .flags = flags,
1390 .result = undefined,
1391 },
1392 },
1393 .finish = .{ .tick_node = .{ .data = @frame() } },
1394 },
1395 };
1396 suspend {
1397 self.posixFsRequest(&req_node);
1398 }
1399 return req_node.data.msg.faccessat.result;
1400 }
1401
1402 fn workerRun(self: *Loop) void {
1403 while (true) {
1404 while (true) {
1405 const next_tick_node = self.next_tick_queue.get() orelse break;
1406 self.dispatch();
1407 resume next_tick_node.data;
1408 self.finishOneEvent();
1409 }
1410
1411 switch (builtin.os.tag) {
1412 .linux => {
1413 // only process 1 event so we don't steal from other threads
1414 var events: [1]os.linux.epoll_event = undefined;
1415 const count = os.epoll_wait(self.os_data.epollfd, events[0..], -1);
1416 for (events[0..count]) |ev| {
1417 const resume_node = @as(*ResumeNode, @ptrFromInt(ev.data.ptr));
1418 const handle = resume_node.handle;
1419 const resume_node_id = resume_node.id;
1420 switch (resume_node_id) {
1421 .basic => {},
1422 .stop => return,
1423 .event_fd => {
1424 const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
1425 event_fd_node.epoll_op = os.linux.EPOLL.CTL_MOD;
1426 const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
1427 self.available_eventfd_resume_nodes.push(stack_node);
1428 },
1429 }
1430 resume handle;
1431 if (resume_node_id == .event_fd) {
1432 self.finishOneEvent();
1433 }
1434 }
1435 },
1436 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
1437 var eventlist: [1]os.Kevent = undefined;
1438 const empty_kevs = &[0]os.Kevent{};
1439 const count = os.kevent(self.os_data.kqfd, empty_kevs, eventlist[0..], null) catch unreachable;
1440 for (eventlist[0..count]) |ev| {
1441 const resume_node = @as(*ResumeNode, @ptrFromInt(ev.udata));
1442 const handle = resume_node.handle;
1443 const resume_node_id = resume_node.id;
1444 switch (resume_node_id) {
1445 .basic => {
1446 const basic_node = @fieldParentPtr(ResumeNode.Basic, "base", resume_node);
1447 basic_node.kev = ev;
1448 },
1449 .stop => return,
1450 .event_fd => {
1451 const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
1452 const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
1453 self.available_eventfd_resume_nodes.push(stack_node);
1454 },
1455 }
1456 resume handle;
1457 if (resume_node_id == .event_fd) {
1458 self.finishOneEvent();
1459 }
1460 }
1461 },
1462 .windows => {
1463 var completion_key: usize = undefined;
1464 const overlapped = while (true) {
1465 var nbytes: windows.DWORD = undefined;
1466 var overlapped: ?*windows.OVERLAPPED = undefined;
1467 switch (windows.GetQueuedCompletionStatus(self.os_data.io_port, &nbytes, &completion_key, &overlapped, windows.INFINITE)) {
1468 .Aborted => return,
1469 .Normal => {},
1470 .EOF => {},
1471 .Cancelled => continue,
1472 }
1473 if (overlapped) |o| break o;
1474 };
1475 const resume_node = @fieldParentPtr(ResumeNode, "overlapped", overlapped);
1476 const handle = resume_node.handle;
1477 const resume_node_id = resume_node.id;
1478 switch (resume_node_id) {
1479 .basic => {},
1480 .stop => return,
1481 .event_fd => {
1482 const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
1483 const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
1484 self.available_eventfd_resume_nodes.push(stack_node);
1485 },
1486 }
1487 resume handle;
1488 self.finishOneEvent();
1489 },
1490 else => @compileError("unsupported OS"),
1491 }
1492 }
1493 }
1494
1495 fn posixFsRequest(self: *Loop, request_node: *Request.Node) void {
1496 self.beginOneEvent(); // finished in posixFsRun after processing the msg
1497 self.fs_queue.put(request_node);
1498 self.fs_thread_wakeup.set();
1499 }
1500
1501 fn posixFsCancel(self: *Loop, request_node: *Request.Node) void {
1502 if (self.fs_queue.remove(request_node)) {
1503 self.finishOneEvent();
1504 }
1505 }
1506
1507 fn posixFsRun(self: *Loop) void {
1508 nosuspend while (true) {
1509 self.fs_thread_wakeup.reset();
1510 while (self.fs_queue.get()) |node| {
1511 switch (node.data.msg) {
1512 .end => return,
1513 .read => |*msg| {
1514 msg.result = os.read(msg.fd, msg.buf);
1515 },
1516 .readv => |*msg| {
1517 msg.result = os.readv(msg.fd, msg.iov);
1518 },
1519 .write => |*msg| {
1520 msg.result = os.write(msg.fd, msg.bytes);
1521 },
1522 .writev => |*msg| {
1523 msg.result = os.writev(msg.fd, msg.iov);
1524 },
1525 .pwrite => |*msg| {
1526 msg.result = os.pwrite(msg.fd, msg.bytes, msg.offset);
1527 },
1528 .pwritev => |*msg| {
1529 msg.result = os.pwritev(msg.fd, msg.iov, msg.offset);
1530 },
1531 .pread => |*msg| {
1532 msg.result = os.pread(msg.fd, msg.buf, msg.offset);
1533 },
1534 .preadv => |*msg| {
1535 msg.result = os.preadv(msg.fd, msg.iov, msg.offset);
1536 },
1537 .open => |*msg| {
1538 if (is_windows) unreachable; // TODO
1539 msg.result = os.openZ(msg.path, msg.flags, msg.mode);
1540 },
1541 .openat => |*msg| {
1542 if (is_windows) unreachable; // TODO
1543 msg.result = os.openatZ(msg.fd, msg.path, msg.flags, msg.mode);
1544 },
1545 .faccessat => |*msg| {
1546 msg.result = os.faccessatZ(msg.dirfd, msg.path, msg.mode, msg.flags);
1547 },
1548 .close => |*msg| os.close(msg.fd),
1549 }
1550 switch (node.data.finish) {
1551 .tick_node => |*tick_node| self.onNextTick(tick_node),
1552 .no_action => {},
1553 }
1554 self.finishOneEvent();
1555 }
1556 self.fs_thread_wakeup.wait();
1557 };
1558 }
1559
1560 const OsData = switch (builtin.os.tag) {
1561 .linux => LinuxOsData,
1562 .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => KEventData,
1563 .windows => struct {
1564 io_port: windows.HANDLE,
1565 extra_thread_count: usize,
1566 },
1567 else => struct {},
1568 };
1569
1570 const KEventData = struct {
1571 kqfd: i32,
1572 final_kevent: os.Kevent,
1573 };
1574
1575 const LinuxOsData = struct {
1576 epollfd: i32,
1577 final_eventfd: i32,
1578 final_eventfd_event: os.linux.epoll_event,
1579 };
1580
1581 pub const Request = struct {
1582 msg: Msg,
1583 finish: Finish,
1584
1585 pub const Node = std.atomic.Queue(Request).Node;
1586
1587 pub const Finish = union(enum) {
1588 tick_node: Loop.NextTickNode,
1589 no_action,
1590 };
1591
1592 pub const Msg = union(enum) {
1593 read: Read,
1594 readv: ReadV,
1595 write: Write,
1596 writev: WriteV,
1597 pwrite: PWrite,
1598 pwritev: PWriteV,
1599 pread: PRead,
1600 preadv: PReadV,
1601 open: Open,
1602 openat: OpenAt,
1603 close: Close,
1604 faccessat: FAccessAt,
1605
1606 /// special - means the fs thread should exit
1607 end,
1608
1609 pub const Read = struct {
1610 fd: os.fd_t,
1611 buf: []u8,
1612 result: Error!usize,
1613
1614 pub const Error = os.ReadError;
1615 };
1616
1617 pub const ReadV = struct {
1618 fd: os.fd_t,
1619 iov: []const os.iovec,
1620 result: Error!usize,
1621
1622 pub const Error = os.ReadError;
1623 };
1624
1625 pub const Write = struct {
1626 fd: os.fd_t,
1627 bytes: []const u8,
1628 result: Error!usize,
1629
1630 pub const Error = os.WriteError;
1631 };
1632
1633 pub const WriteV = struct {
1634 fd: os.fd_t,
1635 iov: []const os.iovec_const,
1636 result: Error!usize,
1637
1638 pub const Error = os.WriteError;
1639 };
1640
1641 pub const PWrite = struct {
1642 fd: os.fd_t,
1643 bytes: []const u8,
1644 offset: usize,
1645 result: Error!usize,
1646
1647 pub const Error = os.PWriteError;
1648 };
1649
1650 pub const PWriteV = struct {
1651 fd: os.fd_t,
1652 iov: []const os.iovec_const,
1653 offset: usize,
1654 result: Error!usize,
1655
1656 pub const Error = os.PWriteError;
1657 };
1658
1659 pub const PRead = struct {
1660 fd: os.fd_t,
1661 buf: []u8,
1662 offset: usize,
1663 result: Error!usize,
1664
1665 pub const Error = os.PReadError;
1666 };
1667
1668 pub const PReadV = struct {
1669 fd: os.fd_t,
1670 iov: []const os.iovec,
1671 offset: usize,
1672 result: Error!usize,
1673
1674 pub const Error = os.PReadError;
1675 };
1676
1677 pub const Open = struct {
1678 path: [*:0]const u8,
1679 flags: u32,
1680 mode: os.mode_t,
1681 result: Error!os.fd_t,
1682
1683 pub const Error = os.OpenError;
1684 };
1685
1686 pub const OpenAt = struct {
1687 fd: os.fd_t,
1688 path: [*:0]const u8,
1689 flags: u32,
1690 mode: os.mode_t,
1691 result: Error!os.fd_t,
1692
1693 pub const Error = os.OpenError;
1694 };
1695
1696 pub const Close = struct {
1697 fd: os.fd_t,
1698 };
1699
1700 pub const FAccessAt = struct {
1701 dirfd: os.fd_t,
1702 path: [*:0]const u8,
1703 mode: u32,
1704 flags: u32,
1705 result: Error!void,
1706
1707 pub const Error = os.AccessError;
1708 };
1709 };
1710 };
1711};
1712
1713test "std.event.Loop - basic" {
1714 // https://github.com/ziglang/zig/issues/1908
1715 if (builtin.single_threaded) return error.SkipZigTest;
1716
1717 if (true) {
1718 // https://github.com/ziglang/zig/issues/4922
1719 return error.SkipZigTest;
1720 }
1721
1722 var loop: Loop = undefined;
1723 try loop.initMultiThreaded();
1724 defer loop.deinit();
1725
1726 loop.run();
1727}
1728
1729fn testEventLoop() i32 {
1730 return 1234;
1731}
1732
1733fn testEventLoop2(h: anyframe->i32, did_it: *bool) void {
1734 const value = await h;
1735 try testing.expect(value == 1234);
1736 did_it.* = true;
1737}
1738
1739var testRunDetachedData: usize = 0;
1740test "std.event.Loop - runDetached" {
1741 // https://github.com/ziglang/zig/issues/1908
1742 if (builtin.single_threaded) return error.SkipZigTest;
1743 if (!std.io.is_async) return error.SkipZigTest;
1744 if (true) {
1745 // https://github.com/ziglang/zig/issues/4922
1746 return error.SkipZigTest;
1747 }
1748
1749 var loop: Loop = undefined;
1750 try loop.initMultiThreaded();
1751 defer loop.deinit();
1752
1753 // Schedule the execution, won't actually start until we start the
1754 // event loop.
1755 try loop.runDetached(std.testing.allocator, testRunDetached, .{});
1756
1757 // Now we can start the event loop. The function will return only
1758 // after all tasks have been completed, allowing us to synchronize
1759 // with the previous runDetached.
1760 loop.run();
1761
1762 try testing.expect(testRunDetachedData == 1);
1763}
1764
1765fn testRunDetached() void {
1766 testRunDetachedData += 1;
1767}
1768
1769test "std.event.Loop - sleep" {
1770 // https://github.com/ziglang/zig/issues/1908
1771 if (builtin.single_threaded) return error.SkipZigTest;
1772 if (!std.io.is_async) return error.SkipZigTest;
1773
1774 const frames = try testing.allocator.alloc(@Frame(testSleep), 10);
1775 defer testing.allocator.free(frames);
1776
1777 const wait_time = 100 * std.time.ns_per_ms;
1778 var sleep_count: usize = 0;
1779
1780 for (frames) |*frame|
1781 frame.* = async testSleep(wait_time, &sleep_count);
1782 for (frames) |*frame|
1783 await frame;
1784
1785 try testing.expect(sleep_count == frames.len);
1786}
1787
1788fn testSleep(wait_ns: u64, sleep_count: *usize) void {
1789 Loop.instance.?.sleep(wait_ns);
1790 _ = @atomicRmw(usize, sleep_count, .Add, 1, .SeqCst);
1791}
lib/std/event/rwlock.zig deleted-292
......@@ -1,292 +0,0 @@
1const std = @import("../std.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const testing = std.testing;
5const mem = std.mem;
6const Loop = std.event.Loop;
7const Allocator = std.mem.Allocator;
8
9/// Thread-safe async/await lock.
10/// Functions which are waiting for the lock are suspended, and
11/// are resumed when the lock is released, in order.
12/// Many readers can hold the lock at the same time; however locking for writing is exclusive.
13/// When a read lock is held, it will not be released until the reader queue is empty.
14/// When a write lock is held, it will not be released until the writer queue is empty.
15/// TODO: make this API also work in blocking I/O mode
16pub const RwLock = struct {
17 shared_state: State,
18 writer_queue: Queue,
19 reader_queue: Queue,
20 writer_queue_empty: bool,
21 reader_queue_empty: bool,
22 reader_lock_count: usize,
23
24 const State = enum(u8) {
25 Unlocked,
26 WriteLock,
27 ReadLock,
28 };
29
30 const Queue = std.atomic.Queue(anyframe);
31
32 const global_event_loop = Loop.instance orelse
33 @compileError("std.event.RwLock currently only works with event-based I/O");
34
35 pub const HeldRead = struct {
36 lock: *RwLock,
37
38 pub fn release(self: HeldRead) void {
39 // If other readers still hold the lock, we're done.
40 if (@atomicRmw(usize, &self.lock.reader_lock_count, .Sub, 1, .SeqCst) != 1) {
41 return;
42 }
43
44 @atomicStore(bool, &self.lock.reader_queue_empty, true, .SeqCst);
45 if (@cmpxchgStrong(State, &self.lock.shared_state, .ReadLock, .Unlocked, .SeqCst, .SeqCst) != null) {
46 // Didn't unlock. Someone else's problem.
47 return;
48 }
49
50 self.lock.commonPostUnlock();
51 }
52 };
53
54 pub const HeldWrite = struct {
55 lock: *RwLock,
56
57 pub fn release(self: HeldWrite) void {
58 // See if we can leave it locked for writing, and pass the lock to the next writer
59 // in the queue to grab the lock.
60 if (self.lock.writer_queue.get()) |node| {
61 global_event_loop.onNextTick(node);
62 return;
63 }
64
65 // We need to release the write lock. Check if any readers are waiting to grab the lock.
66 if (!@atomicLoad(bool, &self.lock.reader_queue_empty, .SeqCst)) {
67 // Switch to a read lock.
68 @atomicStore(State, &self.lock.shared_state, .ReadLock, .SeqCst);
69 while (self.lock.reader_queue.get()) |node| {
70 global_event_loop.onNextTick(node);
71 }
72 return;
73 }
74
75 @atomicStore(bool, &self.lock.writer_queue_empty, true, .SeqCst);
76 @atomicStore(State, &self.lock.shared_state, .Unlocked, .SeqCst);
77
78 self.lock.commonPostUnlock();
79 }
80 };
81
82 pub fn init() RwLock {
83 return .{
84 .shared_state = .Unlocked,
85 .writer_queue = Queue.init(),
86 .writer_queue_empty = true,
87 .reader_queue = Queue.init(),
88 .reader_queue_empty = true,
89 .reader_lock_count = 0,
90 };
91 }
92
93 /// Must be called when not locked. Not thread safe.
94 /// All calls to acquire() and release() must complete before calling deinit().
95 pub fn deinit(self: *RwLock) void {
96 assert(self.shared_state == .Unlocked);
97 while (self.writer_queue.get()) |node| resume node.data;
98 while (self.reader_queue.get()) |node| resume node.data;
99 }
100
101 pub fn acquireRead(self: *RwLock) callconv(.Async) HeldRead {
102 _ = @atomicRmw(usize, &self.reader_lock_count, .Add, 1, .SeqCst);
103
104 suspend {
105 var my_tick_node = Loop.NextTickNode{
106 .data = @frame(),
107 .prev = undefined,
108 .next = undefined,
109 };
110
111 self.reader_queue.put(&my_tick_node);
112
113 // At this point, we are in the reader_queue, so we might have already been resumed.
114
115 // We set this bit so that later we can rely on the fact, that if reader_queue_empty == true,
116 // some actor will attempt to grab the lock.
117 @atomicStore(bool, &self.reader_queue_empty, false, .SeqCst);
118
119 // Here we don't care if we are the one to do the locking or if it was already locked for reading.
120 const have_read_lock = if (@cmpxchgStrong(State, &self.shared_state, .Unlocked, .ReadLock, .SeqCst, .SeqCst)) |old_state| old_state == .ReadLock else true;
121 if (have_read_lock) {
122 // Give out all the read locks.
123 if (self.reader_queue.get()) |first_node| {
124 while (self.reader_queue.get()) |node| {
125 global_event_loop.onNextTick(node);
126 }
127 resume first_node.data;
128 }
129 }
130 }
131 return HeldRead{ .lock = self };
132 }
133
134 pub fn acquireWrite(self: *RwLock) callconv(.Async) HeldWrite {
135 suspend {
136 var my_tick_node = Loop.NextTickNode{
137 .data = @frame(),
138 .prev = undefined,
139 .next = undefined,
140 };
141
142 self.writer_queue.put(&my_tick_node);
143
144 // At this point, we are in the writer_queue, so we might have already been resumed.
145
146 // We set this bit so that later we can rely on the fact, that if writer_queue_empty == true,
147 // some actor will attempt to grab the lock.
148 @atomicStore(bool, &self.writer_queue_empty, false, .SeqCst);
149
150 // Here we must be the one to acquire the write lock. It cannot already be locked.
151 if (@cmpxchgStrong(State, &self.shared_state, .Unlocked, .WriteLock, .SeqCst, .SeqCst) == null) {
152 // We now have a write lock.
153 if (self.writer_queue.get()) |node| {
154 // Whether this node is us or someone else, we tail resume it.
155 resume node.data;
156 }
157 }
158 }
159 return HeldWrite{ .lock = self };
160 }
161
162 fn commonPostUnlock(self: *RwLock) void {
163 while (true) {
164 // There might be a writer_queue item or a reader_queue item
165 // If we check and both are empty, we can be done, because the other actors will try to
166 // obtain the lock.
167 // But if there's a writer_queue item or a reader_queue item,
168 // we are the actor which must loop and attempt to grab the lock again.
169 if (!@atomicLoad(bool, &self.writer_queue_empty, .SeqCst)) {
170 if (@cmpxchgStrong(State, &self.shared_state, .Unlocked, .WriteLock, .SeqCst, .SeqCst) != null) {
171 // We did not obtain the lock. Great, the queues are someone else's problem.
172 return;
173 }
174 // If there's an item in the writer queue, give them the lock, and we're done.
175 if (self.writer_queue.get()) |node| {
176 global_event_loop.onNextTick(node);
177 return;
178 }
179 // Release the lock again.
180 @atomicStore(bool, &self.writer_queue_empty, true, .SeqCst);
181 @atomicStore(State, &self.shared_state, .Unlocked, .SeqCst);
182 continue;
183 }
184
185 if (!@atomicLoad(bool, &self.reader_queue_empty, .SeqCst)) {
186 if (@cmpxchgStrong(State, &self.shared_state, .Unlocked, .ReadLock, .SeqCst, .SeqCst) != null) {
187 // We did not obtain the lock. Great, the queues are someone else's problem.
188 return;
189 }
190 // If there are any items in the reader queue, give out all the reader locks, and we're done.
191 if (self.reader_queue.get()) |first_node| {
192 global_event_loop.onNextTick(first_node);
193 while (self.reader_queue.get()) |node| {
194 global_event_loop.onNextTick(node);
195 }
196 return;
197 }
198 // Release the lock again.
199 @atomicStore(bool, &self.reader_queue_empty, true, .SeqCst);
200 if (@cmpxchgStrong(State, &self.shared_state, .ReadLock, .Unlocked, .SeqCst, .SeqCst) != null) {
201 // Didn't unlock. Someone else's problem.
202 return;
203 }
204 continue;
205 }
206 return;
207 }
208 }
209};
210
211test "std.event.RwLock" {
212 // https://github.com/ziglang/zig/issues/2377
213 if (true) return error.SkipZigTest;
214
215 // https://github.com/ziglang/zig/issues/1908
216 if (builtin.single_threaded) return error.SkipZigTest;
217
218 // TODO provide a way to run tests in evented I/O mode
219 if (!std.io.is_async) return error.SkipZigTest;
220
221 var lock = RwLock.init();
222 defer lock.deinit();
223
224 _ = testLock(std.heap.page_allocator, &lock);
225
226 const expected_result = [1]i32{shared_it_count * @as(i32, @intCast(shared_test_data.len))} ** shared_test_data.len;
227 try testing.expectEqualSlices(i32, expected_result, shared_test_data);
228}
229fn testLock(allocator: Allocator, lock: *RwLock) callconv(.Async) void {
230 var read_nodes: [100]Loop.NextTickNode = undefined;
231 for (&read_nodes) |*read_node| {
232 const frame = allocator.create(@Frame(readRunner)) catch @panic("memory");
233 read_node.data = frame;
234 frame.* = async readRunner(lock);
235 Loop.instance.?.onNextTick(read_node);
236 }
237
238 var write_nodes: [shared_it_count]Loop.NextTickNode = undefined;
239 for (&write_nodes) |*write_node| {
240 const frame = allocator.create(@Frame(writeRunner)) catch @panic("memory");
241 write_node.data = frame;
242 frame.* = async writeRunner(lock);
243 Loop.instance.?.onNextTick(write_node);
244 }
245
246 for (&write_nodes) |*write_node| {
247 const casted = @as(*const @Frame(writeRunner), @ptrCast(write_node.data));
248 await casted;
249 allocator.destroy(casted);
250 }
251 for (&read_nodes) |*read_node| {
252 const casted = @as(*const @Frame(readRunner), @ptrCast(read_node.data));
253 await casted;
254 allocator.destroy(casted);
255 }
256}
257
258const shared_it_count = 10;
259var shared_test_data = [1]i32{0} ** 10;
260var shared_test_index: usize = 0;
261var shared_count: usize = 0;
262fn writeRunner(lock: *RwLock) callconv(.Async) void {
263 suspend {} // resumed by onNextTick
264
265 var i: usize = 0;
266 while (i < shared_test_data.len) : (i += 1) {
267 std.time.sleep(100 * std.time.microsecond);
268 const lock_promise = async lock.acquireWrite();
269 const handle = await lock_promise;
270 defer handle.release();
271
272 shared_count += 1;
273 while (shared_test_index < shared_test_data.len) : (shared_test_index += 1) {
274 shared_test_data[shared_test_index] = shared_test_data[shared_test_index] + 1;
275 }
276 shared_test_index = 0;
277 }
278}
279fn readRunner(lock: *RwLock) callconv(.Async) void {
280 suspend {} // resumed by onNextTick
281 std.time.sleep(1);
282
283 var i: usize = 0;
284 while (i < shared_test_data.len) : (i += 1) {
285 const lock_promise = async lock.acquireRead();
286 const handle = await lock_promise;
287 defer handle.release();
288
289 try testing.expect(shared_test_index == 0);
290 try testing.expect(shared_test_data[i] == @as(i32, @intCast(shared_count)));
291 }
292}
lib/std/event/rwlocked.zig deleted-57
......@@ -1,57 +0,0 @@
1const std = @import("../std.zig");
2const RwLock = std.event.RwLock;
3
4/// Thread-safe async/await RW lock that protects one piece of data.
5/// Functions which are waiting for the lock are suspended, and
6/// are resumed when the lock is released, in order.
7pub fn RwLocked(comptime T: type) type {
8 return struct {
9 lock: RwLock,
10 locked_data: T,
11
12 const Self = @This();
13
14 pub const HeldReadLock = struct {
15 value: *const T,
16 held: RwLock.HeldRead,
17
18 pub fn release(self: HeldReadLock) void {
19 self.held.release();
20 }
21 };
22
23 pub const HeldWriteLock = struct {
24 value: *T,
25 held: RwLock.HeldWrite,
26
27 pub fn release(self: HeldWriteLock) void {
28 self.held.release();
29 }
30 };
31
32 pub fn init(data: T) Self {
33 return Self{
34 .lock = RwLock.init(),
35 .locked_data = data,
36 };
37 }
38
39 pub fn deinit(self: *Self) void {
40 self.lock.deinit();
41 }
42
43 pub fn acquireRead(self: *Self) callconv(.Async) HeldReadLock {
44 return HeldReadLock{
45 .held = self.lock.acquireRead(),
46 .value = &self.locked_data,
47 };
48 }
49
50 pub fn acquireWrite(self: *Self) callconv(.Async) HeldWriteLock {
51 return HeldWriteLock{
52 .held = self.lock.acquireWrite(),
53 .value = &self.locked_data,
54 };
55 }
56 };
57}
lib/std/event/wait_group.zig deleted-115
......@@ -1,115 +0,0 @@
1const std = @import("../std.zig");
2const builtin = @import("builtin");
3const Loop = std.event.Loop;
4
5/// A WaitGroup keeps track and waits for a group of async tasks to finish.
6/// Call `begin` when creating new tasks, and have tasks call `finish` when done.
7/// You can provide a count for both operations to perform them in bulk.
8/// Call `wait` to suspend until all tasks are completed.
9/// Multiple waiters are supported.
10///
11/// WaitGroup is an instance of WaitGroupGeneric, which takes in a bitsize
12/// for the internal counter. WaitGroup defaults to a `usize` counter.
13/// It's also possible to define a max value for the counter so that
14/// `begin` will return error.Overflow when the limit is reached, even
15/// if the integer type has not has not overflowed.
16/// By default `max_value` is set to std.math.maxInt(CounterType).
17pub const WaitGroup = WaitGroupGeneric(@bitSizeOf(usize));
18
19pub fn WaitGroupGeneric(comptime counter_size: u16) type {
20 const CounterType = std.meta.Int(.unsigned, counter_size);
21
22 const global_event_loop = Loop.instance orelse
23 @compileError("std.event.WaitGroup currently only works with event-based I/O");
24
25 return struct {
26 counter: CounterType = 0,
27 max_counter: CounterType = std.math.maxInt(CounterType),
28 mutex: std.Thread.Mutex = .{},
29 waiters: ?*Waiter = null,
30 const Waiter = struct {
31 next: ?*Waiter,
32 tail: *Waiter,
33 node: Loop.NextTickNode,
34 };
35
36 const Self = @This();
37 pub fn begin(self: *Self, count: CounterType) error{Overflow}!void {
38 self.mutex.lock();
39 defer self.mutex.unlock();
40
41 const new_counter = try std.math.add(CounterType, self.counter, count);
42 if (new_counter > self.max_counter) return error.Overflow;
43 self.counter = new_counter;
44 }
45
46 pub fn finish(self: *Self, count: CounterType) void {
47 var waiters = blk: {
48 self.mutex.lock();
49 defer self.mutex.unlock();
50 self.counter = std.math.sub(CounterType, self.counter, count) catch unreachable;
51 if (self.counter == 0) {
52 const temp = self.waiters;
53 self.waiters = null;
54 break :blk temp;
55 }
56 break :blk null;
57 };
58
59 // We don't need to hold the lock to reschedule any potential waiter.
60 while (waiters) |w| {
61 const temp_w = w;
62 waiters = w.next;
63 global_event_loop.onNextTick(&temp_w.node);
64 }
65 }
66
67 pub fn wait(self: *Self) void {
68 self.mutex.lock();
69
70 if (self.counter == 0) {
71 self.mutex.unlock();
72 return;
73 }
74
75 var self_waiter: Waiter = undefined;
76 self_waiter.node.data = @frame();
77 if (self.waiters) |head| {
78 head.tail.next = &self_waiter;
79 head.tail = &self_waiter;
80 } else {
81 self.waiters = &self_waiter;
82 self_waiter.tail = &self_waiter;
83 self_waiter.next = null;
84 }
85 suspend {
86 self.mutex.unlock();
87 }
88 }
89 };
90}
91
92test "basic WaitGroup usage" {
93 if (!std.io.is_async) return error.SkipZigTest;
94
95 // TODO https://github.com/ziglang/zig/issues/1908
96 if (builtin.single_threaded) return error.SkipZigTest;
97
98 // TODO https://github.com/ziglang/zig/issues/3251
99 if (builtin.os.tag == .freebsd) return error.SkipZigTest;
100
101 var initial_wg = WaitGroup{};
102 var final_wg = WaitGroup{};
103
104 try initial_wg.begin(1);
105 try final_wg.begin(1);
106 var task_frame = async task(&initial_wg, &final_wg);
107 initial_wg.finish(1);
108 final_wg.wait();
109 await task_frame;
110}
111
112fn task(wg_i: *WaitGroup, wg_f: *WaitGroup) void {
113 wg_i.wait();
114 wg_f.finish(1);
115}
lib/std/fs.zig-3
......@@ -31,8 +31,6 @@ pub const realpathW = os.realpathW;
3131pub const getAppDataDir = @import("fs/get_app_data_dir.zig").getAppDataDir;
3232pub const GetAppDataDirError = @import("fs/get_app_data_dir.zig").GetAppDataDirError;
3333
34pub const Watch = @import("fs/watch.zig").Watch;
35
3634/// This represents the maximum size of a UTF-8 encoded file path that the
3735/// operating system will accept. Paths, including those returned from file
3836/// system operations, may be longer than this length, but such paths cannot
......@@ -641,5 +639,4 @@ test {
641639 _ = &path;
642640 _ = @import("fs/test.zig");
643641 _ = @import("fs/get_app_data_dir.zig");
644 _ = @import("fs/watch.zig");
645642}
lib/std/fs/watch.zig deleted-719
......@@ -1,719 +0,0 @@
1const std = @import("std");
2const builtin = @import("builtin");
3const event = std.event;
4const assert = std.debug.assert;
5const testing = std.testing;
6const os = std.os;
7const mem = std.mem;
8const windows = os.windows;
9const Loop = event.Loop;
10const fd_t = os.fd_t;
11const File = std.fs.File;
12const Allocator = mem.Allocator;
13
14const global_event_loop = Loop.instance orelse
15 @compileError("std.fs.Watch currently only works with event-based I/O");
16
17const WatchEventId = enum {
18 CloseWrite,
19 Delete,
20};
21
22const WatchEventError = error{
23 UserResourceLimitReached,
24 SystemResources,
25 AccessDenied,
26 Unexpected, // TODO remove this possibility
27};
28
29pub fn Watch(comptime V: type) type {
30 return struct {
31 channel: event.Channel(Event.Error!Event),
32 os_data: OsData,
33 allocator: Allocator,
34
35 const OsData = switch (builtin.os.tag) {
36 // TODO https://github.com/ziglang/zig/issues/3778
37 .macos, .freebsd, .netbsd, .dragonfly, .openbsd => KqOsData,
38 .linux => LinuxOsData,
39 .windows => WindowsOsData,
40
41 else => @compileError("Unsupported OS"),
42 };
43
44 const KqOsData = struct {
45 table_lock: event.Lock,
46 file_table: FileTable,
47
48 const FileTable = std.StringHashMapUnmanaged(*Put);
49 const Put = struct {
50 putter_frame: @Frame(kqPutEvents),
51 cancelled: bool = false,
52 value: V,
53 };
54 };
55
56 const WindowsOsData = struct {
57 table_lock: event.Lock,
58 dir_table: DirTable,
59 cancelled: bool = false,
60
61 const DirTable = std.StringHashMapUnmanaged(*Dir);
62 const FileTable = std.StringHashMapUnmanaged(V);
63
64 const Dir = struct {
65 putter_frame: @Frame(windowsDirReader),
66 file_table: FileTable,
67 dir_handle: os.windows.HANDLE,
68 };
69 };
70
71 const LinuxOsData = struct {
72 putter_frame: @Frame(linuxEventPutter),
73 inotify_fd: i32,
74 wd_table: WdTable,
75 table_lock: event.Lock,
76 cancelled: bool = false,
77
78 const WdTable = std.AutoHashMapUnmanaged(i32, Dir);
79 const FileTable = std.StringHashMapUnmanaged(V);
80
81 const Dir = struct {
82 dirname: []const u8,
83 file_table: FileTable,
84 };
85 };
86
87 const Self = @This();
88
89 pub const Event = struct {
90 id: Id,
91 data: V,
92 dirname: []const u8,
93 basename: []const u8,
94
95 pub const Id = WatchEventId;
96 pub const Error = WatchEventError;
97 };
98
99 pub fn init(allocator: Allocator, event_buf_count: usize) !*Self {
100 const self = try allocator.create(Self);
101 errdefer allocator.destroy(self);
102
103 switch (builtin.os.tag) {
104 .linux => {
105 const inotify_fd = try os.inotify_init1(os.linux.IN_NONBLOCK | os.linux.IN_CLOEXEC);
106 errdefer os.close(inotify_fd);
107
108 self.* = Self{
109 .allocator = allocator,
110 .channel = undefined,
111 .os_data = OsData{
112 .putter_frame = undefined,
113 .inotify_fd = inotify_fd,
114 .wd_table = OsData.WdTable.init(allocator),
115 .table_lock = event.Lock{},
116 },
117 };
118
119 const buf = try allocator.alloc(Event.Error!Event, event_buf_count);
120 self.channel.init(buf);
121 self.os_data.putter_frame = async self.linuxEventPutter();
122 return self;
123 },
124
125 .windows => {
126 self.* = Self{
127 .allocator = allocator,
128 .channel = undefined,
129 .os_data = OsData{
130 .table_lock = event.Lock{},
131 .dir_table = OsData.DirTable.init(allocator),
132 },
133 };
134
135 const buf = try allocator.alloc(Event.Error!Event, event_buf_count);
136 self.channel.init(buf);
137 return self;
138 },
139
140 .macos, .freebsd, .netbsd, .dragonfly, .openbsd => {
141 self.* = Self{
142 .allocator = allocator,
143 .channel = undefined,
144 .os_data = OsData{
145 .table_lock = event.Lock{},
146 .file_table = OsData.FileTable.init(allocator),
147 },
148 };
149
150 const buf = try allocator.alloc(Event.Error!Event, event_buf_count);
151 self.channel.init(buf);
152 return self;
153 },
154 else => @compileError("Unsupported OS"),
155 }
156 }
157
158 pub fn deinit(self: *Self) void {
159 switch (builtin.os.tag) {
160 .macos, .freebsd, .netbsd, .dragonfly, .openbsd => {
161 var it = self.os_data.file_table.iterator();
162 while (it.next()) |entry| {
163 const key = entry.key_ptr.*;
164 const value = entry.value_ptr.*;
165 value.cancelled = true;
166 // @TODO Close the fd here?
167 await value.putter_frame;
168 self.allocator.free(key);
169 self.allocator.destroy(value);
170 }
171 },
172 .linux => {
173 self.os_data.cancelled = true;
174 {
175 // Remove all directory watches linuxEventPutter will take care of
176 // cleaning up the memory and closing the inotify fd.
177 var dir_it = self.os_data.wd_table.keyIterator();
178 while (dir_it.next()) |wd_key| {
179 const rc = os.linux.inotify_rm_watch(self.os_data.inotify_fd, wd_key.*);
180 // Errno can only be EBADF, EINVAL if either the inotify fs or the wd are invalid
181 std.debug.assert(rc == 0);
182 }
183 }
184 await self.os_data.putter_frame;
185 },
186 .windows => {
187 self.os_data.cancelled = true;
188 var dir_it = self.os_data.dir_table.iterator();
189 while (dir_it.next()) |dir_entry| {
190 if (windows.kernel32.CancelIoEx(dir_entry.value.dir_handle, null) != 0) {
191 // We canceled the pending ReadDirectoryChangesW operation, but our
192 // frame is still suspending, now waiting indefinitely.
193 // Thus, it is safe to resume it ourslves
194 resume dir_entry.value.putter_frame;
195 } else {
196 std.debug.assert(windows.kernel32.GetLastError() == .NOT_FOUND);
197 // We are at another suspend point, we can await safely for the
198 // function to exit the loop
199 await dir_entry.value.putter_frame;
200 }
201
202 self.allocator.free(dir_entry.key_ptr.*);
203 var file_it = dir_entry.value.file_table.keyIterator();
204 while (file_it.next()) |file_entry| {
205 self.allocator.free(file_entry.*);
206 }
207 dir_entry.value.file_table.deinit(self.allocator);
208 self.allocator.destroy(dir_entry.value_ptr.*);
209 }
210 self.os_data.dir_table.deinit(self.allocator);
211 },
212 else => @compileError("Unsupported OS"),
213 }
214 self.allocator.free(self.channel.buffer_nodes);
215 self.channel.deinit();
216 self.allocator.destroy(self);
217 }
218
219 pub fn addFile(self: *Self, file_path: []const u8, value: V) !?V {
220 switch (builtin.os.tag) {
221 .macos, .freebsd, .netbsd, .dragonfly, .openbsd => return addFileKEvent(self, file_path, value),
222 .linux => return addFileLinux(self, file_path, value),
223 .windows => return addFileWindows(self, file_path, value),
224 else => @compileError("Unsupported OS"),
225 }
226 }
227
228 fn addFileKEvent(self: *Self, file_path: []const u8, value: V) !?V {
229 var realpath_buf: [std.fs.MAX_PATH_BYTES]u8 = undefined;
230 const realpath = try os.realpath(file_path, &realpath_buf);
231
232 const held = self.os_data.table_lock.acquire();
233 defer held.release();
234
235 const gop = try self.os_data.file_table.getOrPut(self.allocator, realpath);
236 errdefer assert(self.os_data.file_table.remove(realpath));
237 if (gop.found_existing) {
238 const prev_value = gop.value_ptr.value;
239 gop.value_ptr.value = value;
240 return prev_value;
241 }
242
243 gop.key_ptr.* = try self.allocator.dupe(u8, realpath);
244 errdefer self.allocator.free(gop.key_ptr.*);
245 gop.value_ptr.* = try self.allocator.create(OsData.Put);
246 errdefer self.allocator.destroy(gop.value_ptr.*);
247 gop.value_ptr.* = .{
248 .putter_frame = undefined,
249 .value = value,
250 };
251
252 // @TODO Can I close this fd and get an error from bsdWaitKev?
253 const flags = if (comptime builtin.target.isDarwin()) os.O.SYMLINK | os.O.EVTONLY else 0;
254 const fd = try os.open(realpath, flags, 0);
255 gop.value_ptr.putter_frame = async self.kqPutEvents(fd, gop.key_ptr.*, gop.value_ptr.*);
256 return null;
257 }
258
259 fn kqPutEvents(self: *Self, fd: os.fd_t, file_path: []const u8, put: *OsData.Put) void {
260 global_event_loop.beginOneEvent();
261 defer {
262 global_event_loop.finishOneEvent();
263 // @TODO: Remove this if we force close otherwise
264 os.close(fd);
265 }
266
267 // We need to manually do a bsdWaitKev to access the fflags.
268 var resume_node = event.Loop.ResumeNode.Basic{
269 .base = .{
270 .id = .Basic,
271 .handle = @frame(),
272 .overlapped = event.Loop.ResumeNode.overlapped_init,
273 },
274 .kev = undefined,
275 };
276
277 var kevs = [1]os.Kevent{undefined};
278 const kev = &kevs[0];
279
280 while (!put.cancelled) {
281 kev.* = os.Kevent{
282 .ident = @as(usize, @intCast(fd)),
283 .filter = os.EVFILT_VNODE,
284 .flags = os.EV_ADD | os.EV_ENABLE | os.EV_CLEAR | os.EV_ONESHOT |
285 os.NOTE_WRITE | os.NOTE_DELETE | os.NOTE_REVOKE,
286 .fflags = 0,
287 .data = 0,
288 .udata = @intFromPtr(&resume_node.base),
289 };
290 suspend {
291 global_event_loop.beginOneEvent();
292 errdefer global_event_loop.finishOneEvent();
293
294 const empty_kevs = &[0]os.Kevent{};
295 _ = os.kevent(global_event_loop.os_data.kqfd, &kevs, empty_kevs, null) catch |err| switch (err) {
296 error.EventNotFound,
297 error.ProcessNotFound,
298 error.Overflow,
299 => unreachable,
300 error.AccessDenied, error.SystemResources => |e| {
301 self.channel.put(e);
302 continue;
303 },
304 };
305 }
306
307 if (kev.flags & os.EV_ERROR != 0) {
308 self.channel.put(os.unexpectedErrno(os.errno(kev.data)));
309 continue;
310 }
311
312 if (kev.fflags & os.NOTE_DELETE != 0 or kev.fflags & os.NOTE_REVOKE != 0) {
313 self.channel.put(Self.Event{
314 .id = .Delete,
315 .data = put.value,
316 .dirname = std.fs.path.dirname(file_path) orelse "/",
317 .basename = std.fs.path.basename(file_path),
318 });
319 } else if (kev.fflags & os.NOTE_WRITE != 0) {
320 self.channel.put(Self.Event{
321 .id = .CloseWrite,
322 .data = put.value,
323 .dirname = std.fs.path.dirname(file_path) orelse "/",
324 .basename = std.fs.path.basename(file_path),
325 });
326 }
327 }
328 }
329
330 fn addFileLinux(self: *Self, file_path: []const u8, value: V) !?V {
331 const dirname = std.fs.path.dirname(file_path) orelse if (file_path[0] == '/') "/" else ".";
332 const basename = std.fs.path.basename(file_path);
333
334 const wd = try os.inotify_add_watch(
335 self.os_data.inotify_fd,
336 dirname,
337 os.linux.IN_CLOSE_WRITE | os.linux.IN_ONLYDIR | os.linux.IN_DELETE | os.linux.IN_EXCL_UNLINK,
338 );
339 // wd is either a newly created watch or an existing one.
340
341 const held = self.os_data.table_lock.acquire();
342 defer held.release();
343
344 const gop = try self.os_data.wd_table.getOrPut(self.allocator, wd);
345 errdefer assert(self.os_data.wd_table.remove(wd));
346 if (!gop.found_existing) {
347 gop.value_ptr.* = OsData.Dir{
348 .dirname = try self.allocator.dupe(u8, dirname),
349 .file_table = OsData.FileTable.init(self.allocator),
350 };
351 }
352
353 const dir = gop.value_ptr;
354 const file_table_gop = try dir.file_table.getOrPut(self.allocator, basename);
355 errdefer assert(dir.file_table.remove(basename));
356 if (file_table_gop.found_existing) {
357 const prev_value = file_table_gop.value_ptr.*;
358 file_table_gop.value_ptr.* = value;
359 return prev_value;
360 } else {
361 file_table_gop.key_ptr.* = try self.allocator.dupe(u8, basename);
362 file_table_gop.value_ptr.* = value;
363 return null;
364 }
365 }
366
367 fn addFileWindows(self: *Self, file_path: []const u8, value: V) !?V {
368 // TODO we might need to convert dirname and basename to canonical file paths ("short"?)
369 const dirname = std.fs.path.dirname(file_path) orelse if (file_path[0] == '/') "/" else ".";
370 var dirname_path_space: windows.PathSpace = undefined;
371 dirname_path_space.len = try std.unicode.utf8ToUtf16Le(&dirname_path_space.data, dirname);
372 dirname_path_space.data[dirname_path_space.len] = 0;
373
374 const basename = std.fs.path.basename(file_path);
375 var basename_path_space: windows.PathSpace = undefined;
376 basename_path_space.len = try std.unicode.utf8ToUtf16Le(&basename_path_space.data, basename);
377 basename_path_space.data[basename_path_space.len] = 0;
378
379 const held = self.os_data.table_lock.acquire();
380 defer held.release();
381
382 const gop = try self.os_data.dir_table.getOrPut(self.allocator, dirname);
383 errdefer assert(self.os_data.dir_table.remove(dirname));
384 if (gop.found_existing) {
385 const dir = gop.value_ptr.*;
386
387 const file_gop = try dir.file_table.getOrPut(self.allocator, basename);
388 errdefer assert(dir.file_table.remove(basename));
389 if (file_gop.found_existing) {
390 const prev_value = file_gop.value_ptr.*;
391 file_gop.value_ptr.* = value;
392 return prev_value;
393 } else {
394 file_gop.value_ptr.* = value;
395 file_gop.key_ptr.* = try self.allocator.dupe(u8, basename);
396 return null;
397 }
398 } else {
399 const dir_handle = try windows.OpenFile(dirname_path_space.span(), .{
400 .dir = std.fs.cwd().fd,
401 .access_mask = windows.FILE_LIST_DIRECTORY,
402 .creation = windows.FILE_OPEN,
403 .io_mode = .evented,
404 .filter = .dir_only,
405 });
406 errdefer windows.CloseHandle(dir_handle);
407
408 const dir = try self.allocator.create(OsData.Dir);
409 errdefer self.allocator.destroy(dir);
410
411 gop.key_ptr.* = try self.allocator.dupe(u8, dirname);
412 errdefer self.allocator.free(gop.key_ptr.*);
413
414 dir.* = OsData.Dir{
415 .file_table = OsData.FileTable.init(self.allocator),
416 .putter_frame = undefined,
417 .dir_handle = dir_handle,
418 };
419 gop.value_ptr.* = dir;
420 try dir.file_table.put(self.allocator, try self.allocator.dupe(u8, basename), value);
421 dir.putter_frame = async self.windowsDirReader(dir, gop.key_ptr.*);
422 return null;
423 }
424 }
425
426 fn windowsDirReader(self: *Self, dir: *OsData.Dir, dirname: []const u8) void {
427 defer os.close(dir.dir_handle);
428 var resume_node = Loop.ResumeNode.Basic{
429 .base = Loop.ResumeNode{
430 .id = .Basic,
431 .handle = @frame(),
432 .overlapped = windows.OVERLAPPED{
433 .Internal = 0,
434 .InternalHigh = 0,
435 .DUMMYUNIONNAME = .{
436 .DUMMYSTRUCTNAME = .{
437 .Offset = 0,
438 .OffsetHigh = 0,
439 },
440 },
441 .hEvent = null,
442 },
443 },
444 };
445
446 var event_buf: [4096]u8 align(@alignOf(windows.FILE_NOTIFY_INFORMATION)) = undefined;
447
448 global_event_loop.beginOneEvent();
449 defer global_event_loop.finishOneEvent();
450
451 while (!self.os_data.cancelled) main_loop: {
452 suspend {
453 _ = windows.kernel32.ReadDirectoryChangesW(
454 dir.dir_handle,
455 &event_buf,
456 event_buf.len,
457 windows.FALSE, // watch subtree
458 windows.FILE_NOTIFY_CHANGE_FILE_NAME | windows.FILE_NOTIFY_CHANGE_DIR_NAME |
459 windows.FILE_NOTIFY_CHANGE_ATTRIBUTES | windows.FILE_NOTIFY_CHANGE_SIZE |
460 windows.FILE_NOTIFY_CHANGE_LAST_WRITE | windows.FILE_NOTIFY_CHANGE_LAST_ACCESS |
461 windows.FILE_NOTIFY_CHANGE_CREATION | windows.FILE_NOTIFY_CHANGE_SECURITY,
462 null, // number of bytes transferred (unused for async)
463 &resume_node.base.overlapped,
464 null, // completion routine - unused because we use IOCP
465 );
466 }
467
468 var bytes_transferred: windows.DWORD = undefined;
469 if (windows.kernel32.GetOverlappedResult(
470 dir.dir_handle,
471 &resume_node.base.overlapped,
472 &bytes_transferred,
473 windows.FALSE,
474 ) == 0) {
475 const potential_error = windows.kernel32.GetLastError();
476 const err = switch (potential_error) {
477 .OPERATION_ABORTED, .IO_INCOMPLETE => err_blk: {
478 if (self.os_data.cancelled)
479 break :main_loop
480 else
481 break :err_blk windows.unexpectedError(potential_error);
482 },
483 else => |err| windows.unexpectedError(err),
484 };
485 self.channel.put(err);
486 } else {
487 var ptr: [*]u8 = &event_buf;
488 const end_ptr = ptr + bytes_transferred;
489 while (@intFromPtr(ptr) < @intFromPtr(end_ptr)) {
490 const ev = @as(*const windows.FILE_NOTIFY_INFORMATION, @ptrCast(ptr));
491 const emit = switch (ev.Action) {
492 windows.FILE_ACTION_REMOVED => WatchEventId.Delete,
493 windows.FILE_ACTION_MODIFIED => .CloseWrite,
494 else => null,
495 };
496 if (emit) |id| {
497 const basename_ptr = @as([*]u16, @ptrCast(ptr + @sizeOf(windows.FILE_NOTIFY_INFORMATION)));
498 const basename_utf16le = basename_ptr[0 .. ev.FileNameLength / 2];
499 var basename_data: [std.fs.MAX_PATH_BYTES]u8 = undefined;
500 const basename = basename_data[0 .. std.unicode.utf16leToUtf8(&basename_data, basename_utf16le) catch unreachable];
501
502 if (dir.file_table.getEntry(basename)) |entry| {
503 self.channel.put(Event{
504 .id = id,
505 .data = entry.value_ptr.*,
506 .dirname = dirname,
507 .basename = entry.key_ptr.*,
508 });
509 }
510 }
511
512 if (ev.NextEntryOffset == 0) break;
513 ptr = @alignCast(ptr + ev.NextEntryOffset);
514 }
515 }
516 }
517 }
518
519 pub fn removeFile(self: *Self, file_path: []const u8) !?V {
520 switch (builtin.os.tag) {
521 .linux => {
522 const dirname = std.fs.path.dirname(file_path) orelse if (file_path[0] == '/') "/" else ".";
523 const basename = std.fs.path.basename(file_path);
524
525 const held = self.os_data.table_lock.acquire();
526 defer held.release();
527
528 const dir = self.os_data.wd_table.get(dirname) orelse return null;
529 if (dir.file_table.fetchRemove(basename)) |file_entry| {
530 self.allocator.free(file_entry.key);
531 return file_entry.value;
532 }
533 return null;
534 },
535 .windows => {
536 const dirname = std.fs.path.dirname(file_path) orelse if (file_path[0] == '/') "/" else ".";
537 const basename = std.fs.path.basename(file_path);
538
539 const held = self.os_data.table_lock.acquire();
540 defer held.release();
541
542 const dir = self.os_data.dir_table.get(dirname) orelse return null;
543 if (dir.file_table.fetchRemove(basename)) |file_entry| {
544 self.allocator.free(file_entry.key);
545 return file_entry.value;
546 }
547 return null;
548 },
549 .macos, .freebsd, .netbsd, .dragonfly, .openbsd => {
550 var realpath_buf: [std.fs.MAX_PATH_BYTES]u8 = undefined;
551 const realpath = try os.realpath(file_path, &realpath_buf);
552
553 const held = self.os_data.table_lock.acquire();
554 defer held.release();
555
556 const entry = self.os_data.file_table.getEntry(realpath) orelse return null;
557 entry.value_ptr.cancelled = true;
558 // @TODO Close the fd here?
559 await entry.value_ptr.putter_frame;
560 self.allocator.free(entry.key_ptr.*);
561 self.allocator.destroy(entry.value_ptr.*);
562
563 assert(self.os_data.file_table.remove(realpath));
564 },
565 else => @compileError("Unsupported OS"),
566 }
567 }
568
569 fn linuxEventPutter(self: *Self) void {
570 global_event_loop.beginOneEvent();
571
572 defer {
573 std.debug.assert(self.os_data.wd_table.count() == 0);
574 self.os_data.wd_table.deinit(self.allocator);
575 os.close(self.os_data.inotify_fd);
576 self.allocator.free(self.channel.buffer_nodes);
577 self.channel.deinit();
578 global_event_loop.finishOneEvent();
579 }
580
581 var event_buf: [4096]u8 align(@alignOf(os.linux.inotify_event)) = undefined;
582
583 while (!self.os_data.cancelled) {
584 const bytes_read = global_event_loop.read(self.os_data.inotify_fd, &event_buf, false) catch unreachable;
585
586 var ptr: [*]u8 = &event_buf;
587 const end_ptr = ptr + bytes_read;
588 while (@intFromPtr(ptr) < @intFromPtr(end_ptr)) {
589 const ev = @as(*const os.linux.inotify_event, @ptrCast(ptr));
590 if (ev.mask & os.linux.IN_CLOSE_WRITE == os.linux.IN_CLOSE_WRITE) {
591 const basename_ptr = ptr + @sizeOf(os.linux.inotify_event);
592 const basename = std.mem.span(@as([*:0]u8, @ptrCast(basename_ptr)));
593
594 const dir = &self.os_data.wd_table.get(ev.wd).?;
595 if (dir.file_table.getEntry(basename)) |file_value| {
596 self.channel.put(Event{
597 .id = .CloseWrite,
598 .data = file_value.value_ptr.*,
599 .dirname = dir.dirname,
600 .basename = file_value.key_ptr.*,
601 });
602 }
603 } else if (ev.mask & os.linux.IN_IGNORED == os.linux.IN_IGNORED) {
604 // Directory watch was removed
605 const held = self.os_data.table_lock.acquire();
606 defer held.release();
607 if (self.os_data.wd_table.fetchRemove(ev.wd)) |wd_entry| {
608 var file_it = wd_entry.value.file_table.keyIterator();
609 while (file_it.next()) |file_entry| {
610 self.allocator.free(file_entry.*);
611 }
612 self.allocator.free(wd_entry.value.dirname);
613 wd_entry.value.file_table.deinit(self.allocator);
614 }
615 } else if (ev.mask & os.linux.IN_DELETE == os.linux.IN_DELETE) {
616 // File or directory was removed or deleted
617 const basename_ptr = ptr + @sizeOf(os.linux.inotify_event);
618 const basename = std.mem.span(@as([*:0]u8, @ptrCast(basename_ptr)));
619
620 const dir = &self.os_data.wd_table.get(ev.wd).?;
621 if (dir.file_table.getEntry(basename)) |file_value| {
622 self.channel.put(Event{
623 .id = .Delete,
624 .data = file_value.value_ptr.*,
625 .dirname = dir.dirname,
626 .basename = file_value.key_ptr.*,
627 });
628 }
629 }
630
631 ptr = @alignCast(ptr + @sizeOf(os.linux.inotify_event) + ev.len);
632 }
633 }
634 }
635 };
636}
637
638const test_tmp_dir = "std_event_fs_test";
639
640test "write a file, watch it, write it again, delete it" {
641 if (!std.io.is_async) return error.SkipZigTest;
642 // TODO https://github.com/ziglang/zig/issues/1908
643 if (builtin.single_threaded) return error.SkipZigTest;
644
645 try std.fs.cwd().makePath(test_tmp_dir);
646 defer std.fs.cwd().deleteTree(test_tmp_dir) catch {};
647
648 return testWriteWatchWriteDelete(std.testing.allocator);
649}
650
651fn testWriteWatchWriteDelete(allocator: Allocator) !void {
652 const file_path = try std.fs.path.join(allocator, &[_][]const u8{ test_tmp_dir, "file.txt" });
653 defer allocator.free(file_path);
654
655 const contents =
656 \\line 1
657 \\line 2
658 ;
659 const line2_offset = 7;
660
661 // first just write then read the file
662 try std.fs.cwd().writeFile(file_path, contents);
663
664 const read_contents = try std.fs.cwd().readFileAlloc(allocator, file_path, 1024 * 1024);
665 defer allocator.free(read_contents);
666 try testing.expectEqualSlices(u8, contents, read_contents);
667
668 // now watch the file
669 var watch = try Watch(void).init(allocator, 0);
670 defer watch.deinit();
671
672 try testing.expect((try watch.addFile(file_path, {})) == null);
673
674 var ev = async watch.channel.get();
675 var ev_consumed = false;
676 defer if (!ev_consumed) {
677 _ = await ev;
678 };
679
680 // overwrite line 2
681 const file = try std.fs.cwd().openFile(file_path, .{ .mode = .read_write });
682 {
683 defer file.close();
684 const write_contents = "lorem ipsum";
685 var iovec = [_]os.iovec_const{.{
686 .iov_base = write_contents,
687 .iov_len = write_contents.len,
688 }};
689 _ = try file.pwritevAll(&iovec, line2_offset);
690 }
691
692 switch ((try await ev).id) {
693 .CloseWrite => {
694 ev_consumed = true;
695 },
696 .Delete => @panic("wrong event"),
697 }
698
699 const contents_updated = try std.fs.cwd().readFileAlloc(allocator, file_path, 1024 * 1024);
700 defer allocator.free(contents_updated);
701
702 try testing.expectEqualSlices(u8,
703 \\line 1
704 \\lorem ipsum
705 , contents_updated);
706
707 ev = async watch.channel.get();
708 ev_consumed = false;
709
710 try std.fs.cwd().deleteFile(file_path);
711 switch ((try await ev).id) {
712 .Delete => {
713 ev_consumed = true;
714 },
715 .CloseWrite => @panic("wrong event"),
716 }
717}
718
719// TODO Test: Add another file watch, remove the old file watch, get an event in the new
lib/std/std.zig-3
......@@ -92,9 +92,6 @@ pub const elf = @import("elf.zig");
9292/// Enum-related metaprogramming helpers.
9393pub const enums = @import("enums.zig");
9494
95/// Evented I/O data structures.
96pub const event = @import("event.zig");
97
9895/// First in, first out data structures.
9996pub const fifo = @import("fifo.zig");
10097