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-09 20:47:24-08:00
log31fadc0eee2dc25983f493eca7e843629b917b2b
treec56296a7e657a37b47862b9cfbbe96437caa4b94
parentec25b138481e16f2d73d12ed6ffacd8e7ba65e94

std.Io: delete the poll API

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

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