authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-09 15:06:50-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-30 12:10:01-08:00
log45bc4b4e36bc34a2e246e4433a21030ee961fe71
tree20bcb11cfff3baadad5810bd7f91610af61d82fd
parentd1d39cb3fe97ad5273222a6a0e530bb1b949518f

std.Io: exploring a different batch API proposal

5 files changed, 312 insertions(+), 152 deletions(-)

lib/std/Io.zig+84-11
......@@ -149,7 +149,10 @@ 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 operate: *const fn (?*anyopaque, []Operation) void,
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,
155 batchCancel: *const fn (?*anyopaque, *Batch) void,
153156
154157 dirCreateDir: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirError!void,
155158 dirCreateDirPath: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirPathError!Dir.CreatePathStatus,
......@@ -253,26 +256,96 @@ pub const VTable = struct {
253256};
254257
255258pub const Operation = union(enum) {
256 noop,
259 noop: Noop,
257260 file_read_streaming: FileReadStreaming,
258261
262 pub const Noop = struct {
263 reserved: [2]usize,
264 status: Status(void) = .{ .result = {} },
265 };
266
267 /// Returns 0 on end of stream.
259268 pub const FileReadStreaming = struct {
260269 file: File,
261270 data: []const []u8,
262 /// Causes `result` to return `error.WouldBlock` instead of blocking.
263 nonblocking: bool = false,
264 /// Returns 0 on end of stream.
265 result: File.Reader.Error!usize,
271 status: Status(File.Reader.Error!usize) = .{ .unstarted = {} },
266272 };
273
274 pub fn Status(Result: type) type {
275 return union {
276 unstarted: void,
277 pending: usize,
278 result: Result,
279 };
280 }
267281};
268282
269/// Performs all `operations` in a non-deterministic order. Returns after all
270/// `operations` have been completed. The degree to which the operations are
271/// performed concurrently is determined by the `Io` implementation.
272pub fn operate(io: Io, operations: []Operation) void {
273 return io.vtable.operate(io.userdata, operations);
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
303/// Performs one `Operation`.
304pub fn operate(io: Io, operation: *Operation) void {
305 return io.vtable.batch(io.userdata, (operation)[0..1]) catch unreachable;
274306}
275307
308/// Submits many operations together without waiting for all of them to
309/// complete.
310///
311/// This is a low-level abstraction based on `Operation`. For a higher
312/// level API that operates on `Future`, see `Select`.
313pub const Batch = struct {
314 operations: []Operation,
315 index: usize,
316 reserved: ?*anyopaque,
317
318 pub fn init(operations: []Operation) Batch {
319 return .{ .operations = operations, .index = 0, .reserved = null };
320 }
321
322 /// Submits all non-noop `operations`.
323 pub fn submit(b: *Batch, io: Io) void {
324 return io.vtable.batchSubmit(io.userdata, b);
325 }
326
327 pub const WaitError = ConcurrentError || Cancelable || Timeout.Error;
328
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.
336 ///
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);
340 }
341
342 /// Returns after all `operations` have completed. Each operation
343 /// independently may or may not have been canceled.
344 pub fn cancel(b: *Batch, io: Io) void {
345 return io.vtable.batchCancel(io.userdata, b);
346 }
347};
348
276349pub const Limit = enum(usize) {
277350 nothing = 0,
278351 unlimited = math.maxInt(usize),
lib/std/Io/File.zig+2-3
......@@ -557,10 +557,9 @@ pub fn readStreaming(file: File, io: Io, buffer: []const []u8) Reader.Error!usiz
557557 var operation: Io.Operation = .{ .file_read_streaming = .{
558558 .file = file,
559559 .data = buffer,
560 .result = undefined,
561560 } };
562 io.vtable.operate(io.userdata, (&operation)[0..1]);
563 return operation.file_read_streaming.result;
561 io.operate(&operation);
562 return operation.file_read_streaming.status.result;
564563}
565564
566565pub const ReadPositionalError = error{
lib/std/Io/Threaded.zig+171-85
......@@ -1587,7 +1587,10 @@ pub fn io(t: *Threaded) Io {
15871587 .futexWaitUncancelable = futexWaitUncancelable,
15881588 .futexWake = futexWake,
15891589
1590 .operate = operate,
1590 .batch = batch,
1591 .batchSubmit = batchSubmit,
1592 .batchWait = batchWait,
1593 .batchCancel = batchCancel,
15911594
15921595 .dirCreateDir = dirCreateDir,
15931596 .dirCreateDirPath = dirCreateDirPath,
......@@ -1748,7 +1751,10 @@ pub fn ioBasic(t: *Threaded) Io {
17481751 .futexWaitUncancelable = futexWaitUncancelable,
17491752 .futexWake = futexWake,
17501753
1751 .operate = operate,
1754 .batch = batch,
1755 .batchSubmit = batchSubmit,
1756 .batchWait = batchWait,
1757 .batchCancel = batchCancel,
17521758
17531759 .dirCreateDir = dirCreateDir,
17541760 .dirCreateDirPath = dirCreateDirPath,
......@@ -2450,107 +2456,187 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
24502456 Thread.futexWake(ptr, max_waiters);
24512457}
24522458
2453fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {
2459fn batchSubmit(userdata: ?*anyopaque, b: *Io.Batch) void {
24542460 const t: *Threaded = @ptrCast(@alignCast(userdata));
24552461 _ = t;
2462 _ = b;
2463 return;
2464}
2465
2466fn operate(op: *Io.Operation) void {
2467 switch (op.*) {
2468 .noop => {},
2469 .file_read_streaming => |*o| o.status = .{ .result = fileReadStreaming(o.file, o.data) },
2470 }
2471}
24562472
2473fn batchWait(
2474 userdata: ?*anyopaque,
2475 b: *Io.Batch,
2476 resubmissions: []const usize,
2477 timeout: Io.Timeout,
2478) Io.Batch.WaitError!usize {
2479 _ = resubmissions;
2480 const t: *Threaded = @ptrCast(@alignCast(userdata));
2481 const operations = b.operations;
2482 if (operations.len == 1) {
2483 operate(&operations[0]);
2484 return b.operations.len;
2485 }
24572486 if (is_windows) @panic("TODO");
24582487
24592488 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
24602489 var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index
2461 var operation_index: usize = 0;
2462
2463 while (operation_index < operations.len) {
2464 var poll_i: usize = 0;
2465 while (operation_index < operations.len) : (operation_index += 1) {
2466 switch (operations[operation_index]) {
2467 .noop => continue,
2468 .file_read_streaming => |*o| {
2469 if (o.nonblocking) {
2470 o.result = error.WouldBlock;
2471 poll_buffer[poll_i] = .{
2472 .fd = o.file.handle,
2473 .events = posix.POLL.IN,
2474 .revents = 0,
2475 };
2476 if (map_buffer.len - poll_i == 0) break;
2477 map_buffer[poll_i] = @intCast(operation_index);
2478 poll_i += 1;
2479 } else {
2480 o.result = fileReadStreaming(o.file, o.data) catch |err| switch (err) {
2481 error.Canceled => {
2482 setOperationsError(operations[operation_index..], error.Canceled);
2483 return;
2484 },
2485 else => err,
2486 };
2487 }
2488 },
2489 }
2490 }
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;
2503 },
2504 };
24912505
2492 if (poll_i == 0) {
2493 @branchHint(.likely);
2494 return;
2506 if (poll_i == 0) return operations.len;
2507
2508 const t_io = ioBasic(t);
2509 const deadline = timeout.toDeadline(t_io) catch return error.UnsupportedClock;
2510 const max_poll_ms = std.math.maxInt(i32);
2511
2512 while (true) {
2513 const timeout_ms: i32 = if (deadline) |d| t: {
2514 const duration = d.durationFromNow(t_io) catch return error.UnsupportedClock;
2515 if (duration.raw.nanoseconds <= 0) return error.Timeout;
2516 break :t @intCast(@min(max_poll_ms, duration.raw.toMilliseconds()));
2517 } else -1;
2518 const syscall = try Syscall.start();
2519 const rc = posix.system.poll(&poll_buffer, poll_i, timeout_ms);
2520 syscall.finish();
2521 switch (posix.errno(rc)) {
2522 .SUCCESS => {
2523 if (rc == 0) {
2524 // Although spurious timeouts are OK, when no deadline is
2525 // passed we must not return `error.Timeout`.
2526 if (deadline == null) continue;
2527 return error.Timeout;
2528 }
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;
2533 }
2534 },
2535 .INTR => continue,
2536 else => return error.ConcurrencyUnavailable,
24952537 }
2538 }
2539}
24962540
2497 while (true) {
2498 const syscall = Syscall.start() catch |err| switch (err) {
2499 error.Canceled => {
2500 setPollOperationsError(operations, map_buffer[0..poll_i], error.Canceled);
2501 setOperationsError(operations[operation_index..], error.Canceled);
2502 return;
2503 },
2541fn batchCancel(userdata: ?*anyopaque, b: *Io.Batch) void {
2542 const t: *Threaded = @ptrCast(@alignCast(userdata));
2543 _ = t;
2544 _ = b;
2545 return;
2546}
2547
2548fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!void {
2549 const t: *Threaded = @ptrCast(@alignCast(userdata));
2550 _ = t;
2551
2552 if (operations.len == 1) {
2553 @branchHint(.likely);
2554 return operate(&operations[0]);
2555 }
2556
2557 if (is_windows) @panic("TODO");
2558
2559 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
2560 var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index
2561 var poll_i: usize = 0;
2562
2563 for (operations, 0..) |*op, operation_index| switch (op.*) {
2564 .noop => continue,
2565 .file_read_streaming => |*o| {
2566 if (poll_buffer.len - poll_i == 0) return error.ConcurrencyUnavailable;
2567 poll_buffer[poll_i] = .{
2568 .fd = o.file.handle,
2569 .events = posix.POLL.IN,
2570 .revents = 0,
25042571 };
2505 const poll_rc = posix.system.poll(&poll_buffer, poll_i, -1);
2506 syscall.finish();
2507 switch (posix.errno(poll_rc)) {
2508 .SUCCESS => {
2509 if (poll_rc == 0) {
2510 // Spurious timeout; handle same as INTR.
2511 continue;
2512 }
2513 for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, i| {
2514 if (poll_fd.revents == 0) continue;
2515 switch (operations[i]) {
2516 .noop => unreachable,
2517 .file_read_streaming => |*o| {
2518 o.result = fileReadStreaming(o.file, o.data);
2519 },
2520 }
2521 }
2522 break;
2523 },
2524 .INTR => continue,
2525 .NOMEM => {
2526 setPollOperationsError(operations, map_buffer[0..poll_i], error.SystemResources);
2527 break;
2528 },
2529 else => {
2530 setPollOperationsError(operations, map_buffer[0..poll_i], error.Unexpected);
2531 break;
2532 },
2533 }
2572 map_buffer[poll_i] = @intCast(operation_index);
2573 poll_i += 1;
2574 },
2575 };
2576
2577 const polls = poll_buffer[0..poll_i];
2578 const map = map_buffer[0..poll_i];
2579
2580 var pending = poll_i;
2581 while (pending > 1) {
2582 const syscall = Syscall.start() catch |err| switch (err) {
2583 error.Canceled => {
2584 if (!setOperationsError(operations, polls, map, error.Canceled))
2585 recancelInner();
2586 return;
2587 },
2588 };
2589 const rc = posix.system.poll(polls.ptr, polls.len, -1);
2590 syscall.finish();
2591 switch (posix.errno(rc)) {
2592 .SUCCESS => {
2593 if (rc == 0) {
2594 // Spurious timeout; handle the same as INTR.
2595 continue;
2596 }
2597 for (polls, map) |*poll_fd, i| {
2598 if (poll_fd.revents == 0) continue;
2599 poll_fd.fd = -1;
2600 pending -= 1;
2601 operate(&operations[i]);
2602 }
2603 },
2604 .INTR => continue,
2605 .NOMEM => {
2606 assert(setOperationsError(operations, polls, map, error.SystemResources));
2607 return;
2608 },
2609 else => {
2610 assert(setOperationsError(operations, polls, map, error.Unexpected));
2611 return;
2612 },
25342613 }
25352614 }
2615
2616 if (pending == 1) for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, i| {
2617 if (poll_fd.fd == -1) continue;
2618 operate(&operations[i]);
2619 };
25362620}
25372621
2538fn setPollOperationsError(
2622fn setOperationsError(
25392623 operations: []Io.Operation,
2624 polls: []const posix.pollfd,
25402625 map: []const u8,
25412626 err: error{ Canceled, SystemResources, Unexpected },
2542) void {
2543 for (map) |operation_index| switch (operations[operation_index]) {
2544 .noop => unreachable,
2545 inline else => |*o| o.result = err,
2546 };
2547}
2548
2549fn setOperationsError(operations: []Io.Operation, err: error{ Canceled, SystemResources, Unexpected }) void {
2550 for (operations) |*op| switch (op.*) {
2551 .noop => unreachable,
2552 inline else => |*o| o.result = err,
2553 };
2627) bool {
2628 var marked = false;
2629 for (polls, map) |*poll_fd, i| {
2630 if (poll_fd.fd == -1) continue;
2631 switch (operations[i]) {
2632 .noop => unreachable,
2633 inline else => |*o| {
2634 o.status = .{ .result = err };
2635 marked = true;
2636 },
2637 }
2638 }
2639 return marked;
25542640}
25552641
25562642const dirCreateDir = switch (native_os) {
lib/std/process.zig+8-8
......@@ -453,9 +453,7 @@ pub fn spawnPath(io: Io, dir: Io.Dir, options: SpawnOptions) SpawnError!Child {
453453 return io.vtable.processSpawnPath(io.userdata, dir, options);
454454}
455455
456pub const RunError = CurrentPathError || posix.ReadError || SpawnError || posix.PollError || error{
457 StreamTooLong,
458};
456pub const RunError = SpawnError || Child.CollectOutputError;
459457
460458pub const RunOptions = struct {
461459 argv: []const []const u8,
......@@ -535,13 +533,15 @@ pub fn run(gpa: Allocator, io: Io, options: RunOptions) RunError!RunResult {
535533
536534 const term = try child.wait(io);
537535
538 const owned_stdout = try stdout.toOwnedSlice(gpa);
539 errdefer gpa.free(owned_stdout);
540 const owned_stderr = try stderr.toOwnedSlice(gpa);
536 const stdout_slice = try stdout.toOwnedSlice(gpa);
537 errdefer gpa.free(stdout_slice);
538
539 const stderr_slice = try stderr.toOwnedSlice(gpa);
540 errdefer gpa.free(stderr_slice);
541541
542542 return .{
543 .stdout = owned_stdout,
544 .stderr = owned_stderr,
543 .stdout = stdout_slice,
544 .stderr = stderr_slice,
545545 .term = term,
546546 };
547547}
lib/std/process/Child.zig+47-45
......@@ -125,7 +125,9 @@ pub fn wait(child: *Child, io: Io) WaitError!Term {
125125 return io.vtable.childWait(io.userdata, child);
126126}
127127
128pub const CollectOutputError = error{StreamTooLong} || Allocator.Error || Io.File.Reader.Error;
128pub const CollectOutputError = error{
129 StreamTooLong,
130} || Io.ConcurrentError || Allocator.Error || Io.File.Reader.Error || Io.Timeout.Error;
129131
130132pub const CollectOutputOptions = struct {
131133 stdout: *std.ArrayList(u8),
......@@ -135,6 +137,7 @@ pub const CollectOutputOptions = struct {
135137 allocator: ?Allocator = null,
136138 stdout_limit: Io.Limit = .unlimited,
137139 stderr_limit: Io.Limit = .unlimited,
140 timeout: Io.Timeout = .none,
138141};
139142
140143/// Collect the output from the process's stdout and stderr. Will return once
......@@ -144,56 +147,55 @@ pub const CollectOutputOptions = struct {
144147/// The process must have been started with stdout and stderr set to
145148/// `process.SpawnOptions.StdIo.pipe`.
146149pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions) CollectOutputError!void {
147 const files: [2]Io.File = .{ child.stdout.?, child.stderr.? };
148150 const lists: [2]*std.ArrayList(u8) = .{ options.stdout, options.stderr };
149151 const limits: [2]Io.Limit = .{ options.stdout_limit, options.stderr_limit };
150 var dones: [2]bool = .{ false, false };
151 var reads: [2]Io.Operation = undefined;
152
153 if (options.allocator) |gpa| {
154 for (lists) |list| try list.ensureUnusedCapacity(gpa, 1);
155 } else {
156 for (lists) |list| {
157 if (list.unusedCapacitySlice().len == 0)
158 return error.StreamTooLong;
159 }
160 }
161
152162 var vecs: [2][1][]u8 = undefined;
153 while (true) {
154 for (&reads, &lists, &files, dones, &vecs) |*read, list, file, done, *vec| {
155 if (done) {
156 read.* = .noop;
157 continue;
158 }
163 for (lists, &vecs) |list, *vec|
164 vec[0] = list.unusedCapacitySlice();
165
166 var operations: [2]Io.Operation = .{
167 .{ .file_read_streaming = .{
168 .file = child.stdout.?,
169 .data = &vecs[0],
170 } },
171 .{ .file_read_streaming = .{
172 .file = child.stderr.?,
173 .data = &vecs[1],
174 } },
175 };
176
177 var batch: Io.Batch = .init(&operations);
178 batch.submit(io);
179 defer batch.cancel(io);
180
181 var pending = operations.len;
182 var retry_index: ?usize = null;
183 while (pending > 0) {
184 const resubmissions: []const usize = if (retry_index) |i| &.{i} else &.{};
185 const index = try batch.wait(io, resubmissions, options.timeout);
186 const n = try operations[index].file_read_streaming.status.result;
187 if (n == 0) {
188 pending -= 1;
189 } else {
190 retry_index = index;
191 const list = lists[index];
192 const limit = limits[index];
193 list.items.len += n;
194 if (list.items.len >= @intFromEnum(limit)) return error.StreamTooLong;
159195 if (options.allocator) |gpa| try list.ensureUnusedCapacity(gpa, 1);
160196 const cap = list.unusedCapacitySlice();
161197 if (cap.len == 0) return error.StreamTooLong;
162 vec[0] = cap;
163 read.* = .{ .file_read_streaming = .{
164 .file = file,
165 .data = vec,
166 .nonblocking = true,
167 .result = undefined,
168 } };
169 }
170 var all_done = true;
171 var any_canceled = false;
172 var other_err: (error{StreamTooLong} || Io.File.Reader.Error)!void = {};
173 io.vtable.operate(io.userdata, &reads);
174 for (&reads, &lists, &limits, &dones) |*read, list, limit, *done| {
175 if (done.*) continue;
176 const n = read.file_read_streaming.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 };
187 if (n == 0) {
188 done.* = true;
189 } else {
190 all_done = false;
191 }
192 list.items.len += n;
193 if (list.items.len > @intFromEnum(limit)) other_err = error.StreamTooLong;
198 vecs[index][0] = cap;
194199 }
195 if (any_canceled) return error.Canceled;
196 try other_err;
197 if (all_done) return;
198200 }
199201}