authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2026-01-10 15:34:36-05:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-30 12:10:01-08:00
logdea653fdb97c62968138c28496f25510b5d50024
tree037776338497fd9675c83c327ba38316cfedb66f
parent5456f953fab3254bd87a1be8d7c42b15d4bb8138

Io: add ring to `Batch` API

5 files changed, 258 insertions(+), 158 deletions(-)

lib/std/Io.zig+121-48
......@@ -149,9 +149,8 @@ pub const VTable = struct {
149149 futexWaitUncancelable: *const fn (?*anyopaque, ptr: *const u32, expected: u32) void,
150150 futexWake: *const fn (?*anyopaque, ptr: *const u32, max_waiters: u32) void,
151151
152 batch: *const fn (?*anyopaque, []Operation) ConcurrentError!void,
153 batchSubmit: *const fn (?*anyopaque, *Batch) void,
154 batchWait: *const fn (?*anyopaque, *Batch, resubmissions: []const usize, Timeout) Batch.WaitError!usize,
152 operate: *const fn (?*anyopaque, *Operation) Cancelable!void,
153 batchWait: *const fn (?*anyopaque, *Batch, Timeout) Batch.WaitError!void,
155154 batchCancel: *const fn (?*anyopaque, *Batch) void,
156155
157156 dirCreateDir: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirError!void,
......@@ -261,48 +260,50 @@ pub const Operation = union(enum) {
261260
262261 pub const Noop = struct {
263262 reserved: [2]usize = .{ 0, 0 },
264 status: Status(void) = .{ .result = {} },
263 status: Status(void) = .{ .unstarted = {} },
265264 };
266265
267266 /// Returns 0 on end of stream.
268267 pub const FileReadStreaming = struct {
269268 file: File,
270269 data: []const []u8,
271 status: Status(File.Reader.Error!usize) = .{ .unstarted = {} },
270 status: Status(Error!usize) = .{ .unstarted = {} },
271
272 pub const Error = error{
273 InputOutput,
274 SystemResources,
275 /// Trying to read a directory file descriptor as if it were a file.
276 IsDir,
277 BrokenPipe,
278 ConnectionResetByPeer,
279 /// File was not opened with read capability.
280 NotOpenForReading,
281 SocketUnconnected,
282 /// Non-blocking has been enabled, and reading from the file descriptor
283 /// would block.
284 WouldBlock,
285 /// In WASI, this error occurs when the file descriptor does
286 /// not hold the required rights to read from it.
287 AccessDenied,
288 /// Unable to read file due to lock. Depending on the `Io` implementation,
289 /// reading from a locked file may return this error, or may ignore the
290 /// lock.
291 LockViolation,
292 } || Io.UnexpectedError;
272293 };
273294
274295 pub fn Status(Result: type) type {
275296 return union {
276297 unstarted: void,
277 pending: usize,
298 pending: *Batch,
278299 result: Result,
279300 };
280301 }
281302};
282303
283/// Performs all `operations` in an unspecified order, concurrently.
284///
285/// Returns after all `operations` have been completed. If the operations could
286/// not be completed concurrently, returns `error.ConcurrencyUnavailable`.
287///
288/// With this API, it is rare for concurrency to not be available. Even a
289/// single-threaded `Io` implementation can, for example, take advantage of
290/// poll() to implement this. Note that poll() is fallible however.
291///
292/// If `operations.len` is one, `error.ConcurrencyUnavailable` is unreachable.
293///
294/// On entry, all operations must already have `.status = .unstarted` except
295/// noops must have `.status = .{ .result = {} }`, to safety check the state
296/// transitions.
297///
298/// On return, all operations have `.status = .{ .result = ... }`.
299pub fn batch(io: Io, operations: []Operation) ConcurrentError!void {
300 return io.vtable.batch(io.userdata, operations);
301}
302
303304/// Performs one `Operation`.
304pub fn operate(io: Io, operation: *Operation) void {
305 return io.vtable.batch(io.userdata, (operation)[0..1]) catch unreachable;
305pub fn operate(io: Io, operation: *Operation) Cancelable!void {
306 return io.vtable.operate(io.userdata, operation) catch unreachable;
306307}
307308
308309/// Submits many operations together without waiting for all of them to
......@@ -312,35 +313,107 @@ pub fn operate(io: Io, operation: *Operation) void {
312313/// level API that operates on `Future`, see `Select`.
313314pub const Batch = struct {
314315 operations: []Operation,
315 index: usize,
316 reserved: ?*anyopaque,
316 ring: [*]u32,
317 user: struct {
318 submit_tail: RingIndex,
319 complete_head: RingIndex,
320 complete_tail: RingIndex,
321 },
322 impl: struct {
323 submit_head: RingIndex,
324 submit_tail: RingIndex,
325 complete_tail: RingIndex,
326 reserved: ?*anyopaque,
327 },
328
329 pub const RingIndex = enum(u32) {
330 _,
331
332 pub fn index(ri: RingIndex, len: u31) u31 {
333 const i = @intFromEnum(ri);
334 assert(i < @as(u32, len) * 2);
335 return @intCast(if (i < len) i else i - len);
336 }
337
338 pub fn prev(ri: RingIndex, len: u31) RingIndex {
339 const i = @intFromEnum(ri);
340 const double_len = @as(u32, len) * 2;
341 assert(i <= double_len);
342 return @enumFromInt((if (i > 0) i else double_len) - 1);
343 }
344
345 pub fn next(ri: RingIndex, len: u31) RingIndex {
346 const i = @intFromEnum(ri) + 1;
347 const double_len = @as(u32, len) * 2;
348 assert(i <= double_len);
349 return @enumFromInt(if (i < double_len) i else 0);
350 }
351 };
352
353 pub const WaitError = ConcurrentError || Cancelable || Timeout.Error;
317354
318 pub fn init(operations: []Operation) Batch {
319 return .{ .operations = operations, .index = 0, .reserved = null };
355 pub fn init(operations: []Operation, ring: []u32) Batch {
356 const len: u31 = @intCast(operations.len);
357 assert(ring.len == len);
358 return .{
359 .operations = operations,
360 .ring = ring.ptr,
361 .user = .{
362 .submit_tail = @enumFromInt(0),
363 .complete_head = @enumFromInt(0),
364 .complete_tail = @enumFromInt(0),
365 },
366 .impl = .{
367 .submit_head = @enumFromInt(0),
368 .submit_tail = @enumFromInt(0),
369 .complete_tail = @enumFromInt(0),
370 .reserved = null,
371 },
372 };
320373 }
321374
322 /// Submits all non-noop `operations`.
323 pub fn submit(b: *Batch, io: Io) void {
324 return io.vtable.batchSubmit(io.userdata, b);
375 /// Adds `b.operations[operation]` to the list of submitted operations
376 /// that will be performed when `wait` is called.
377 pub fn add(b: *Batch, operation: usize) void {
378 const tail = b.user.submit_tail;
379 const len: u31 = @intCast(b.operations.len);
380 b.user.submit_tail = tail.next(len);
381 b.ring[0..len][tail.index(len)] = @intCast(operation);
325382 }
326383
327 pub const WaitError = ConcurrentError || Cancelable || Timeout.Error;
384 fn flush(b: *Batch) void {
385 @atomicStore(RingIndex, &b.impl.submit_tail, b.user.submit_tail, .release);
386 }
328387
329 /// Resubmits the previously completed or noop-initialized `operations` at
330 /// indexes given by `resubmissions`. This set of indexes typically will be empty
331 /// on the first call to `await` since all operations have already been
332 /// submitted via `async`.
333 ///
334 /// Returns the index of a completed `Operation`, or `operations.len` if
335 /// all operations are completed.
388 /// Returns `operation` such that `b.operations[operation]` has completed.
389 /// Returns `null` when `wait` should be called.
390 pub fn next(b: *Batch) ?u32 {
391 const head = b.user.complete_head;
392 if (head == b.user.complete_tail) {
393 @branchHint(.unlikely);
394 b.flush();
395 const tail = @atomicLoad(RingIndex, &b.impl.complete_tail, .acquire);
396 if (head == tail) {
397 @branchHint(.unlikely);
398 return null;
399 }
400 assert(head != tail);
401 b.user.complete_tail = tail;
402 }
403 const len: u31 = @intCast(b.operations.len);
404 b.user.complete_head = head.next(len);
405 return b.ring[0..len][head.index(len)];
406 }
407
408 /// Starts work on any submitted operations and returns when at least one has completeed.
336409 ///
337 /// When `error.Canceled` is returned, all operations have already completed.
338 pub fn wait(b: *Batch, io: Io, resubmissions: []const usize, timeout: Timeout) WaitError!usize {
339 return io.vtable.batchWait(io.userdata, b, resubmissions, timeout);
410 /// Returns `error.Timeout` if `timeout` expires first.
411 pub fn wait(b: *Batch, io: Io, timeout: Timeout) WaitError!void {
412 return io.vtable.batchWait(io.userdata, b, timeout);
340413 }
341414
342 /// Returns after all `operations` have completed. Each operation
343 /// independently may or may not have been canceled.
415 /// Returns after all `operations` have completed. Operations which have not completed
416 /// after this function returns were successfully dropped and had no side effects.
344417 pub fn cancel(b: *Batch, io: Io) void {
345418 return io.vtable.batchCancel(io.userdata, b);
346419 }
lib/std/Io/File.zig+1-1
......@@ -558,7 +558,7 @@ pub fn readStreaming(file: File, io: Io, buffer: []const []u8) Reader.Error!usiz
558558 .file = file,
559559 .data = buffer,
560560 } };
561 io.operate(&operation);
561 try io.operate(&operation);
562562 return operation.file_read_streaming.status.result;
563563}
564564
lib/std/Io/File/Reader.zig+1-21
......@@ -26,27 +26,7 @@ size_err: ?SizeError = null,
2626seek_err: ?SeekError = null,
2727interface: Io.Reader,
2828
29pub const Error = error{
30 InputOutput,
31 SystemResources,
32 /// Trying to read a directory file descriptor as if it were a file.
33 IsDir,
34 BrokenPipe,
35 ConnectionResetByPeer,
36 /// File was not opened with read capability.
37 NotOpenForReading,
38 SocketUnconnected,
39 /// Non-blocking has been enabled, and reading from the file descriptor
40 /// would block.
41 WouldBlock,
42 /// In WASI, this error occurs when the file descriptor does
43 /// not hold the required rights to read from it.
44 AccessDenied,
45 /// Unable to read file due to lock. Depending on the `Io` implementation,
46 /// reading from a locked file may return this error, or may ignore the
47 /// lock.
48 LockViolation,
49} || Io.Cancelable || Io.UnexpectedError;
29pub const Error = Io.Operation.FileReadStreaming.Error || Io.Cancelable;
5030
5131pub const SizeError = File.StatError || error{
5232 /// Occurs if, for example, the file handle is a network socket and therefore does not have a size.
lib/std/Io/Threaded.zig+102-49
......@@ -1587,8 +1587,7 @@ pub fn io(t: *Threaded) Io {
15871587 .futexWaitUncancelable = futexWaitUncancelable,
15881588 .futexWake = futexWake,
15891589
1590 .batch = batch,
1591 .batchSubmit = batchSubmit,
1590 .operate = operate,
15921591 .batchWait = batchWait,
15931592 .batchCancel = batchCancel,
15941593
......@@ -1751,8 +1750,7 @@ pub fn ioBasic(t: *Threaded) Io {
17511750 .futexWaitUncancelable = futexWaitUncancelable,
17521751 .futexWake = futexWake,
17531752
1754 .batch = batch,
1755 .batchSubmit = batchSubmit,
1753 .operate = operate,
17561754 .batchWait = batchWait,
17571755 .batchCancel = batchCancel,
17581756
......@@ -2456,59 +2454,82 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
24562454 Thread.futexWake(ptr, max_waiters);
24572455}
24582456
2459fn batchSubmit(userdata: ?*anyopaque, b: *Io.Batch) void {
2457fn operate(userdata: ?*anyopaque, op: *Io.Operation) Io.Cancelable!void {
24602458 const t: *Threaded = @ptrCast(@alignCast(userdata));
24612459 _ = t;
2462 _ = b;
2463 return;
2464}
2465
2466fn operate(op: *Io.Operation) void {
24672460 switch (op.*) {
2468 .noop => {},
2469 .file_read_streaming => |*o| o.status = .{ .result = fileReadStreaming(o.file, o.data) },
2461 .noop => |*o| {
2462 _ = o.status.unstarted;
2463 o.status = .{ .result = {} };
2464 },
2465 .file_read_streaming => |*o| {
2466 _ = o.status.unstarted;
2467 o.status = .{ .result = fileReadStreaming(o.file, o.data) catch |err| switch (err) {
2468 error.Canceled => return error.Canceled,
2469 else => |e| e,
2470 } };
2471 },
24702472 }
24712473}
24722474
2473fn batchWait(
2474 userdata: ?*anyopaque,
2475 b: *Io.Batch,
2476 resubmissions: []const usize,
2477 timeout: Io.Timeout,
2478) Io.Batch.WaitError!usize {
2479 _ = resubmissions;
2475fn batchWait(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.WaitError!void {
24802476 const t: *Threaded = @ptrCast(@alignCast(userdata));
24812477 const operations = b.operations;
2482 if (operations.len == 1) {
2483 operate(&operations[0]);
2484 return b.operations.len;
2478 const len: u31 = @intCast(operations.len);
2479 const ring = b.ring[0..len];
2480 var submit_head = b.impl.submit_head;
2481 const submit_tail = b.user.submit_tail;
2482 b.impl.submit_tail = submit_tail;
2483 var complete_tail = b.impl.complete_tail;
2484 var map_buffer: [poll_buffer_len]u32 = undefined; // poll_buffer index to operations index
2485 var poll_i: usize = 0;
2486 defer {
2487 for (map_buffer[0..poll_i]) |op| {
2488 submit_head = submit_head.prev(len);
2489 ring[submit_head.index(len)] = op;
2490 }
2491 b.impl.submit_head = submit_head;
2492 b.impl.complete_tail = complete_tail;
2493 b.user.complete_tail = complete_tail;
24852494 }
24862495 if (is_windows) @panic("TODO");
2487
24882496 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
2489 var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index
2490 var poll_i: usize = 0;
2491
2492 for (operations, 0..) |*op, operation_index| switch (op.*) {
2493 .noop => continue,
2494 .file_read_streaming => |*o| {
2495 if (poll_buffer.len - poll_i == 0) return error.ConcurrencyUnavailable;
2496 poll_buffer[poll_i] = .{
2497 .fd = o.file.handle,
2498 .events = posix.POLL.IN,
2499 .revents = 0,
2500 };
2501 map_buffer[poll_i] = @intCast(operation_index);
2502 poll_i += 1;
2497 while (submit_head != submit_tail) : (submit_head = submit_head.next(len)) {
2498 const op = ring[submit_head.index(len)];
2499 const operation = &operations[op];
2500 switch (operation.*) {
2501 else => {
2502 try operate(t, operation);
2503 ring[complete_tail.index(len)] = op;
2504 complete_tail = complete_tail.next(len);
2505 },
2506 .file_read_streaming => |*o| {
2507 _ = o.status.unstarted;
2508 if (poll_buffer.len - poll_i == 0) return error.ConcurrencyUnavailable;
2509 poll_buffer[poll_i] = .{
2510 .fd = o.file.handle,
2511 .events = posix.POLL.IN,
2512 .revents = 0,
2513 };
2514 map_buffer[poll_i] = op;
2515 poll_i += 1;
2516 },
2517 }
2518 }
2519 switch (poll_i) {
2520 0 => return,
2521 1 => if (timeout == .none) {
2522 const op = map_buffer[0];
2523 try operate(t, &operations[op]);
2524 ring[complete_tail.index(len)] = op;
2525 complete_tail = complete_tail.next(len);
2526 return;
25032527 },
2504 };
2505
2506 if (poll_i == 0) return operations.len;
2507
2528 else => {},
2529 }
25082530 const t_io = ioBasic(t);
25092531 const deadline = timeout.toDeadline(t_io) catch return error.UnsupportedClock;
25102532 const max_poll_ms = std.math.maxInt(i32);
2511
25122533 while (true) {
25132534 const timeout_ms: i32 = if (deadline) |d| t: {
25142535 const duration = d.durationFromNow(t_io) catch return error.UnsupportedClock;
......@@ -2526,11 +2547,24 @@ fn batchWait(
25262547 if (deadline == null) continue;
25272548 return error.Timeout;
25282549 }
2529 for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, i| {
2530 if (poll_fd.revents == 0) continue;
2531 operate(&operations[i]);
2532 return i;
2550 var canceled = false;
2551 for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, op| {
2552 if (poll_fd.revents == 0) {
2553 submit_head = submit_head.prev(len);
2554 ring[submit_head.index(len)] = op;
2555 } else {
2556 operate(t, &operations[op]) catch |err| switch (err) {
2557 error.Canceled => {
2558 canceled = true;
2559 continue;
2560 },
2561 };
2562 ring[complete_tail.index(len)] = op;
2563 complete_tail = complete_tail.next(len);
2564 }
25332565 }
2566 poll_i = 0;
2567 return if (canceled) error.Canceled;
25342568 },
25352569 .INTR => continue,
25362570 else => return error.ConcurrencyUnavailable,
......@@ -2540,9 +2574,27 @@ fn batchWait(
25402574
25412575fn batchCancel(userdata: ?*anyopaque, b: *Io.Batch) void {
25422576 const t: *Threaded = @ptrCast(@alignCast(userdata));
2543 _ = t;
2544 _ = b;
2545 return;
2577 const operations = b.operations;
2578 const len: u31 = @intCast(operations.len);
2579 const ring = b.ring[0..len];
2580 var submit_head = b.impl.submit_head;
2581 const submit_tail = b.user.submit_tail;
2582 b.impl.submit_tail = submit_tail;
2583 var complete_tail = b.impl.complete_tail;
2584 while (submit_head != submit_tail) : (submit_head = submit_head.next(len)) {
2585 const op = ring[submit_head.index(len)];
2586 switch (operations[op]) {
2587 .noop => {
2588 operate(t, &operations[op]) catch unreachable;
2589 ring[complete_tail.index(len)] = op;
2590 complete_tail = complete_tail.next(len);
2591 },
2592 .file_read_streaming => |*o| _ = o.status.unstarted,
2593 }
2594 }
2595 b.impl.submit_head = submit_tail;
2596 b.impl.complete_tail = complete_tail;
2597 b.user.complete_tail = complete_tail;
25462598}
25472599
25482600fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!void {
......@@ -10352,6 +10404,7 @@ fn nowWasi(clock: Io.Clock) Io.Clock.Error!Io.Timestamp {
1035210404
1035310405fn sleep(userdata: ?*anyopaque, timeout: Io.Timeout) Io.SleepError!void {
1035410406 const t: *Threaded = @ptrCast(@alignCast(userdata));
10407 if (timeout == .none) return;
1035510408 if (use_parking_sleep) return parking_sleep.sleep(try timeout.toDeadline(ioBasic(t)));
1035610409 if (native_os == .wasi) return sleepWasi(t, timeout);
1035710410 if (@TypeOf(posix.system.clock_nanosleep) != void) return sleepPosix(timeout);
lib/std/process/Child.zig+33-39
......@@ -149,51 +149,45 @@ pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions)
149149 const files: [2]Io.File = .{ child.stdout.?, child.stderr.? };
150150 const lists: [2]*std.ArrayList(u8) = .{ options.stdout, options.stderr };
151151 const limits: [2]Io.Limit = .{ options.stdout_limit, options.stderr_limit };
152 var dones: [2]bool = .{ false, false };
153152 var reads: [2]Io.Operation = undefined;
154153 var vecs: [2][1][]u8 = undefined;
155 while (true) {
156 for (&reads, &lists, &files, dones, &vecs) |*read, list, file, done, *vec| {
157 if (done) {
158 read.* = .{ .noop = .{} };
159 continue;
160 }
161 if (options.allocator) |gpa| try list.ensureUnusedCapacity(gpa, 1);
162 const cap = list.unusedCapacitySlice();
163 if (cap.len == 0) return error.StreamTooLong;
164 vec[0] = cap;
165 read.* = .{ .file_read_streaming = .{
166 .file = file,
167 .data = vec,
168 } };
154 var ring: [2]u32 = undefined;
155 var batch: Io.Batch = .init(&reads, &ring);
156 defer {
157 batch.cancel(io);
158 while (batch.next()) |op| {
159 lists[op].items.len += reads[op].file_read_streaming.status.result catch continue;
169160 }
170 var all_done = true;
171 var any_canceled = false;
172 var other_err: (error{StreamTooLong} || Io.File.Reader.Error)!void = {};
173 try io.vtable.batch(io.userdata, &reads);
174 for (&reads, &lists, &limits, &dones) |*read, list, limit, *done| {
175 if (done.*) continue;
176 const n = read.file_read_streaming.status.result catch |err| switch (err) {
177 error.Canceled => {
178 any_canceled = true;
179 continue;
180 },
181 error.WouldBlock => continue,
182 else => |e| {
183 other_err = e;
184 continue;
185 },
186 };
161 }
162 var remaining: usize = 0;
163 for (0.., &reads, &lists, &files, &vecs) |op, *read, list, file, *vec| {
164 if (options.allocator) |gpa| try list.ensureUnusedCapacity(gpa, 1);
165 const cap = list.unusedCapacitySlice();
166 if (cap.len == 0) return error.StreamTooLong;
167 vec[0] = cap;
168 read.* = .{ .file_read_streaming = .{
169 .file = file,
170 .data = vec,
171 } };
172 batch.add(op);
173 remaining += 1;
174 }
175 while (remaining > 0) {
176 try batch.wait(io, .none);
177 while (batch.next()) |op| {
178 const n = try reads[op].file_read_streaming.status.result;
187179 if (n == 0) {
188 done.* = true;
180 remaining -= 1;
189181 } else {
190 all_done = false;
182 lists[op].items.len += n;
183 if (lists[op].items.len > @intFromEnum(limits[op])) return error.StreamTooLong;
184 if (options.allocator) |gpa| try lists[op].ensureUnusedCapacity(gpa, 1);
185 const cap = lists[op].unusedCapacitySlice();
186 if (cap.len == 0) return error.StreamTooLong;
187 vecs[op][0] = cap;
188 reads[op].file_read_streaming.status = .{ .unstarted = {} };
189 batch.add(op);
191190 }
192 list.items.len += n;
193 if (list.items.len > @intFromEnum(limit)) other_err = error.StreamTooLong;
194191 }
195 if (any_canceled) return error.Canceled;
196 try other_err;
197 if (all_done) return;
198192 }
199193}