authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-05 22:19:08-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-30 12:10:01-08:00
logdef22c2b63022b0c06dac8b5f7554deb644ca089
tree788b5a92ba6015a422d2c2eb8fd069477f2bb3b8
parentb5174455f8049b926b9e712ed8f77c13327ffe8c

std.Io: delete the poll API

1 files changed, 8 insertions(+), 458 deletions(-)

lib/std/Io.zig+8-458
......@@ -15,463 +15,13 @@
1515const Io = @This();
1616
1717const builtin = @import("builtin");
18const is_windows = builtin.os.tag == .windows;
1918
2019const std = @import("std.zig");
21const windows = std.os.windows;
22const posix = std.posix;
2320const math = std.math;
2421const assert = std.debug.assert;
2522const Allocator = std.mem.Allocator;
2623const Alignment = std.mem.Alignment;
2724
28pub fn poll(
29 gpa: Allocator,
30 comptime StreamEnum: type,
31 files: PollFiles(StreamEnum),
32) Poller(StreamEnum) {
33 const enum_fields = @typeInfo(StreamEnum).@"enum".fields;
34 var result: Poller(StreamEnum) = .{
35 .gpa = gpa,
36 .readers = @splat(.failing),
37 .poll_fds = undefined,
38 .windows = if (is_windows) .{
39 .first_read_done = false,
40 .overlapped = [1]windows.OVERLAPPED{
41 std.mem.zeroes(windows.OVERLAPPED),
42 } ** enum_fields.len,
43 .small_bufs = undefined,
44 .active = .{
45 .count = 0,
46 .handles_buf = undefined,
47 .stream_map = undefined,
48 },
49 } else {},
50 };
51
52 inline for (enum_fields, 0..) |field, i| {
53 if (is_windows) {
54 result.windows.active.handles_buf[i] = @field(files, field.name).handle;
55 } else {
56 result.poll_fds[i] = .{
57 .fd = @field(files, field.name).handle,
58 .events = posix.POLL.IN,
59 .revents = undefined,
60 };
61 }
62 }
63
64 return result;
65}
66
67pub fn Poller(comptime StreamEnum: type) type {
68 return struct {
69 const enum_fields = @typeInfo(StreamEnum).@"enum".fields;
70 const PollFd = if (is_windows) void else posix.pollfd;
71
72 gpa: Allocator,
73 readers: [enum_fields.len]Reader,
74 poll_fds: [enum_fields.len]PollFd,
75 windows: if (is_windows) struct {
76 first_read_done: bool,
77 overlapped: [enum_fields.len]windows.OVERLAPPED,
78 small_bufs: [enum_fields.len][128]u8,
79 active: struct {
80 count: math.IntFittingRange(0, enum_fields.len),
81 handles_buf: [enum_fields.len]windows.HANDLE,
82 stream_map: [enum_fields.len]StreamEnum,
83
84 pub fn removeAt(self: *@This(), index: u32) void {
85 assert(index < self.count);
86 for (index + 1..self.count) |i| {
87 self.handles_buf[i - 1] = self.handles_buf[i];
88 self.stream_map[i - 1] = self.stream_map[i];
89 }
90 self.count -= 1;
91 }
92 },
93 } else void,
94
95 const Self = @This();
96
97 pub fn deinit(self: *Self) void {
98 const gpa = self.gpa;
99 if (is_windows) {
100 // cancel any pending IO to prevent clobbering OVERLAPPED value
101 for (self.windows.active.handles_buf[0..self.windows.active.count]) |h| {
102 _ = windows.kernel32.CancelIo(h);
103 }
104 }
105 inline for (&self.readers) |*r| gpa.free(r.buffer);
106 self.* = undefined;
107 }
108
109 pub fn poll(self: *Self) !bool {
110 if (is_windows) {
111 return pollWindows(self, null);
112 } else {
113 return pollPosix(self, null);
114 }
115 }
116
117 pub fn pollTimeout(self: *Self, nanoseconds: u64) !bool {
118 if (is_windows) {
119 return pollWindows(self, nanoseconds);
120 } else {
121 return pollPosix(self, nanoseconds);
122 }
123 }
124
125 pub fn reader(self: *Self, which: StreamEnum) *Reader {
126 return &self.readers[@intFromEnum(which)];
127 }
128
129 pub fn toOwnedSlice(self: *Self, which: StreamEnum) error{OutOfMemory}![]u8 {
130 const gpa = self.gpa;
131 const r = reader(self, which);
132 if (r.seek == 0) {
133 const new = try gpa.realloc(r.buffer, r.end);
134 r.buffer = &.{};
135 r.end = 0;
136 return new;
137 }
138 const new = try gpa.dupe(u8, r.buffered());
139 gpa.free(r.buffer);
140 r.buffer = &.{};
141 r.seek = 0;
142 r.end = 0;
143 return new;
144 }
145
146 fn pollWindows(self: *Self, nanoseconds: ?u64) !bool {
147 const bump_amt = 512;
148 const gpa = self.gpa;
149
150 if (!self.windows.first_read_done) {
151 var already_read_data = false;
152 for (0..enum_fields.len) |i| {
153 const handle = self.windows.active.handles_buf[i];
154 switch (try windowsAsyncReadToFifoAndQueueSmallRead(
155 gpa,
156 handle,
157 &self.windows.overlapped[i],
158 &self.readers[i],
159 &self.windows.small_bufs[i],
160 bump_amt,
161 )) {
162 .populated, .empty => |state| {
163 if (state == .populated) already_read_data = true;
164 self.windows.active.handles_buf[self.windows.active.count] = handle;
165 self.windows.active.stream_map[self.windows.active.count] = @as(StreamEnum, @enumFromInt(i));
166 self.windows.active.count += 1;
167 },
168 .closed => {}, // don't add to the wait_objects list
169 .closed_populated => {
170 // don't add to the wait_objects list, but we did already get data
171 already_read_data = true;
172 },
173 }
174 }
175 self.windows.first_read_done = true;
176 if (already_read_data) return true;
177 }
178
179 while (true) {
180 if (self.windows.active.count == 0) return false;
181
182 const status = windows.kernel32.WaitForMultipleObjects(
183 self.windows.active.count,
184 &self.windows.active.handles_buf,
185 0,
186 if (nanoseconds) |ns|
187 @min(std.math.cast(u32, ns / std.time.ns_per_ms) orelse (windows.INFINITE - 1), windows.INFINITE - 1)
188 else
189 windows.INFINITE,
190 );
191 if (status == windows.WAIT_FAILED)
192 return windows.unexpectedError(windows.GetLastError());
193 if (status == windows.WAIT_TIMEOUT)
194 return true;
195
196 if (status < windows.WAIT_OBJECT_0 or status > windows.WAIT_OBJECT_0 + enum_fields.len - 1)
197 unreachable;
198
199 const active_idx = status - windows.WAIT_OBJECT_0;
200
201 const stream_idx = @intFromEnum(self.windows.active.stream_map[active_idx]);
202 const handle = self.windows.active.handles_buf[active_idx];
203
204 const overlapped = &self.windows.overlapped[stream_idx];
205 const stream_reader = &self.readers[stream_idx];
206 const small_buf = &self.windows.small_bufs[stream_idx];
207
208 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {
209 .success => |n| n,
210 .closed => {
211 self.windows.active.removeAt(active_idx);
212 continue;
213 },
214 .aborted => unreachable,
215 };
216 const buf = small_buf[0..num_bytes_read];
217 const dest = try writableSliceGreedyAlloc(stream_reader, gpa, buf.len);
218 @memcpy(dest[0..buf.len], buf);
219 advanceBufferEnd(stream_reader, buf.len);
220
221 switch (try windowsAsyncReadToFifoAndQueueSmallRead(
222 gpa,
223 handle,
224 overlapped,
225 stream_reader,
226 small_buf,
227 bump_amt,
228 )) {
229 .empty => {}, // irrelevant, we already got data from the small buffer
230 .populated => {},
231 .closed,
232 .closed_populated, // identical, since we already got data from the small buffer
233 => self.windows.active.removeAt(active_idx),
234 }
235 return true;
236 }
237 }
238
239 fn pollPosix(self: *Self, nanoseconds: ?u64) !bool {
240 const gpa = self.gpa;
241 // We ask for ensureUnusedCapacity with this much extra space. This
242 // has more of an effect on small reads because once the reads
243 // start to get larger the amount of space an ArrayList will
244 // allocate grows exponentially.
245 const bump_amt = 512;
246
247 const err_mask = posix.POLL.ERR | posix.POLL.NVAL | posix.POLL.HUP;
248
249 const events_len = try posix.poll(&self.poll_fds, if (nanoseconds) |ns|
250 std.math.cast(i32, ns / std.time.ns_per_ms) orelse std.math.maxInt(i32)
251 else
252 -1);
253 if (events_len == 0) {
254 for (self.poll_fds) |poll_fd| {
255 if (poll_fd.fd != -1) return true;
256 } else return false;
257 }
258
259 var keep_polling = false;
260 for (&self.poll_fds, &self.readers) |*poll_fd, *r| {
261 // Try reading whatever is available before checking the error
262 // conditions.
263 // It's still possible to read after a POLL.HUP is received,
264 // always check if there's some data waiting to be read first.
265 if (poll_fd.revents & posix.POLL.IN != 0) {
266 const buf = try writableSliceGreedyAlloc(r, gpa, bump_amt);
267 const amt = posix.read(poll_fd.fd, buf) catch |err| switch (err) {
268 error.BrokenPipe => 0, // Handle the same as EOF.
269 else => |e| return e,
270 };
271 advanceBufferEnd(r, amt);
272 if (amt == 0) {
273 // Remove the fd when the EOF condition is met.
274 poll_fd.fd = -1;
275 } else {
276 keep_polling = true;
277 }
278 } else if (poll_fd.revents & err_mask != 0) {
279 // Exclude the fds that signaled an error.
280 poll_fd.fd = -1;
281 } else if (poll_fd.fd != -1) {
282 keep_polling = true;
283 }
284 }
285 return keep_polling;
286 }
287
288 /// Returns a slice into the unused capacity of `buffer` with at least
289 /// `min_len` bytes, extending `buffer` by resizing it with `gpa` as necessary.
290 ///
291 /// After calling this function, typically the caller will follow up with a
292 /// call to `advanceBufferEnd` to report the actual number of bytes buffered.
293 fn writableSliceGreedyAlloc(r: *Reader, allocator: Allocator, min_len: usize) Allocator.Error![]u8 {
294 {
295 const unused = r.buffer[r.end..];
296 if (unused.len >= min_len) return unused;
297 }
298 if (r.seek > 0) {
299 const data = r.buffer[r.seek..r.end];
300 @memmove(r.buffer[0..data.len], data);
301 r.seek = 0;
302 r.end = data.len;
303 }
304 {
305 var list: std.ArrayList(u8) = .{
306 .items = r.buffer[0..r.end],
307 .capacity = r.buffer.len,
308 };
309 defer r.buffer = list.allocatedSlice();
310 try list.ensureUnusedCapacity(allocator, min_len);
311 }
312 const unused = r.buffer[r.end..];
313 assert(unused.len >= min_len);
314 return unused;
315 }
316
317 /// After writing directly into the unused capacity of `buffer`, this function
318 /// updates `end` so that users of `Reader` can receive the data.
319 fn advanceBufferEnd(r: *Reader, n: usize) void {
320 assert(n <= r.buffer.len - r.end);
321 r.end += n;
322 }
323
324 /// The `ReadFile` docuementation states that `lpNumberOfBytesRead` does not have a meaningful
325 /// result when using overlapped I/O, but also that it cannot be `null` on Windows 7. For
326 /// compatibility, we point it to this dummy variables, which we never otherwise access.
327 /// See: https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-readfile
328 var win_dummy_bytes_read: u32 = undefined;
329
330 /// Read as much data as possible from `handle` with `overlapped`, and write it to the FIFO. Before
331 /// returning, queue a read into `small_buf` so that `WaitForMultipleObjects` returns when more data
332 /// is available. `handle` must have no pending asynchronous operation.
333 fn windowsAsyncReadToFifoAndQueueSmallRead(
334 gpa: Allocator,
335 handle: windows.HANDLE,
336 overlapped: *windows.OVERLAPPED,
337 r: *Reader,
338 small_buf: *[128]u8,
339 bump_amt: usize,
340 ) !enum { empty, populated, closed_populated, closed } {
341 var read_any_data = false;
342 while (true) {
343 const fifo_read_pending = while (true) {
344 const buf = try writableSliceGreedyAlloc(r, gpa, bump_amt);
345 const buf_len = math.cast(u32, buf.len) orelse math.maxInt(u32);
346
347 if (0 == windows.kernel32.ReadFile(
348 handle,
349 buf.ptr,
350 buf_len,
351 &win_dummy_bytes_read,
352 overlapped,
353 )) switch (windows.GetLastError()) {
354 .IO_PENDING => break true,
355 .BROKEN_PIPE => return if (read_any_data) .closed_populated else .closed,
356 else => |err| return windows.unexpectedError(err),
357 };
358
359 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {
360 .success => |n| n,
361 .closed => return if (read_any_data) .closed_populated else .closed,
362 .aborted => unreachable,
363 };
364
365 read_any_data = true;
366 advanceBufferEnd(r, num_bytes_read);
367
368 if (num_bytes_read == buf_len) {
369 // We filled the buffer, so there's probably more data available.
370 continue;
371 } else {
372 // We didn't fill the buffer, so assume we're out of data.
373 // There is no pending read.
374 break false;
375 }
376 };
377
378 if (fifo_read_pending) cancel_read: {
379 // Cancel the pending read into the FIFO.
380 _ = windows.kernel32.CancelIo(handle);
381
382 // We have to wait for the handle to be signalled, i.e. for the cancelation to complete.
383 switch (windows.kernel32.WaitForSingleObject(handle, windows.INFINITE)) {
384 windows.WAIT_OBJECT_0 => {},
385 windows.WAIT_FAILED => return windows.unexpectedError(windows.GetLastError()),
386 else => unreachable,
387 }
388
389 // If it completed before we canceled, make sure to tell the FIFO!
390 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, true)) {
391 .success => |n| n,
392 .closed => return if (read_any_data) .closed_populated else .closed,
393 .aborted => break :cancel_read,
394 };
395 read_any_data = true;
396 advanceBufferEnd(r, num_bytes_read);
397 }
398
399 // Try to queue the 1-byte read.
400 if (0 == windows.kernel32.ReadFile(
401 handle,
402 small_buf,
403 small_buf.len,
404 &win_dummy_bytes_read,
405 overlapped,
406 )) switch (windows.GetLastError()) {
407 .IO_PENDING => {
408 // 1-byte read pending as intended
409 return if (read_any_data) .populated else .empty;
410 },
411 .BROKEN_PIPE => return if (read_any_data) .closed_populated else .closed,
412 else => |err| return windows.unexpectedError(err),
413 };
414
415 // We got data back this time. Write it to the FIFO and run the main loop again.
416 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {
417 .success => |n| n,
418 .closed => return if (read_any_data) .closed_populated else .closed,
419 .aborted => unreachable,
420 };
421 const buf = small_buf[0..num_bytes_read];
422 const dest = try writableSliceGreedyAlloc(r, gpa, buf.len);
423 @memcpy(dest[0..buf.len], buf);
424 advanceBufferEnd(r, buf.len);
425 read_any_data = true;
426 }
427 }
428
429 /// Simple wrapper around `GetOverlappedResult` to determine the result of a `ReadFile` operation.
430 /// If `!allow_aborted`, then `aborted` is never returned (`OPERATION_ABORTED` is considered unexpected).
431 ///
432 /// The `ReadFile` documentation states that the number of bytes read by an overlapped `ReadFile` must be determined using `GetOverlappedResult`, even if the
433 /// operation immediately returns data:
434 /// "Use NULL for [lpNumberOfBytesRead] if this is an asynchronous operation to avoid potentially
435 /// erroneous results."
436 /// "If `hFile` was opened with `FILE_FLAG_OVERLAPPED`, the following conditions are in effect: [...]
437 /// The lpNumberOfBytesRead parameter should be set to NULL. Use the GetOverlappedResult function to
438 /// get the actual number of bytes read."
439 /// See: https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-readfile
440 fn windowsGetReadResult(
441 handle: windows.HANDLE,
442 overlapped: *windows.OVERLAPPED,
443 allow_aborted: bool,
444 ) !union(enum) {
445 success: u32,
446 closed,
447 aborted,
448 } {
449 var num_bytes_read: u32 = undefined;
450 if (0 == windows.kernel32.GetOverlappedResult(
451 handle,
452 overlapped,
453 &num_bytes_read,
454 0,
455 )) switch (windows.GetLastError()) {
456 .BROKEN_PIPE => return .closed,
457 .OPERATION_ABORTED => |err| if (allow_aborted) {
458 return .aborted;
459 } else {
460 return windows.unexpectedError(err);
461 },
462 else => |err| return windows.unexpectedError(err),
463 };
464 return .{ .success = num_bytes_read };
465 }
466 };
467}
468
469/// Given an enum, returns a struct with fields of that enum, each field
470/// representing an I/O stream for polling.
471pub fn PollFiles(comptime StreamEnum: type) type {
472 return @Struct(.auto, null, std.meta.fieldNames(StreamEnum), &@splat(Io.File), &@splat(.{}));
473}
474
47525userdata: ?*anyopaque,
47626vtable: *const VTable,
47727
......@@ -704,18 +254,18 @@ pub const VTable = struct {
704254
705255pub const Limit = enum(usize) {
706256 nothing = 0,
707 unlimited = std.math.maxInt(usize),
257 unlimited = math.maxInt(usize),
708258 _,
709259
710 /// `std.math.maxInt(usize)` is interpreted to mean `.unlimited`.
260 /// `math.maxInt(usize)` is interpreted to mean `.unlimited`.
711261 pub fn limited(n: usize) Limit {
712262 return @enumFromInt(n);
713263 }
714264
715 /// Any value grater than `std.math.maxInt(usize)` is interpreted to mean
265 /// Any value grater than `math.maxInt(usize)` is interpreted to mean
716266 /// `.unlimited`.
717267 pub fn limited64(n: u64) Limit {
718 return @enumFromInt(@min(n, std.math.maxInt(usize)));
268 return @enumFromInt(@min(n, math.maxInt(usize)));
719269 }
720270
721271 pub fn countVec(data: []const []const u8) Limit {
......@@ -929,9 +479,9 @@ pub const Clock = enum {
929479 };
930480 }
931481
932 pub fn compare(lhs: Clock.Timestamp, op: std.math.CompareOperator, rhs: Clock.Timestamp) bool {
482 pub fn compare(lhs: Clock.Timestamp, op: math.CompareOperator, rhs: Clock.Timestamp) bool {
933483 assert(lhs.clock == rhs.clock);
934 return std.math.compare(lhs.raw.nanoseconds, op, rhs.raw.nanoseconds);
484 return math.compare(lhs.raw.nanoseconds, op, rhs.raw.nanoseconds);
935485 }
936486 };
937487
......@@ -996,7 +546,7 @@ pub const Duration = struct {
996546 nanoseconds: i96,
997547
998548 pub const zero: Duration = .{ .nanoseconds = 0 };
999 pub const max: Duration = .{ .nanoseconds = std.math.maxInt(i96) };
549 pub const max: Duration = .{ .nanoseconds = math.maxInt(i96) };
1000550
1001551 pub fn fromNanoseconds(x: i96) Duration {
1002552 return .{ .nanoseconds = x };
......@@ -1652,7 +1202,7 @@ pub const Event = enum(u32) {
16521202 pub fn set(e: *Event, io: Io) void {
16531203 switch (@atomicRmw(Event, e, .Xchg, .is_set, .release)) {
16541204 .unset, .is_set => {},
1655 .waiting => io.futexWake(Event, e, std.math.maxInt(u32)),
1205 .waiting => io.futexWake(Event, e, math.maxInt(u32)),
16561206 }
16571207 }
16581208