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-09 20:47:25-08:00
log1ca398b267298b5d03ce854459f2d5234fbf43ee
tree28ae584ded70ef43eb60ecef6b27f40f5678c3ec
parentbe2b93a073246eba1d2f44daeb1962b062096187

std.Io: exploring a different batch API proposal

5 files changed, 313 insertions(+), 149 deletions(-)

lib/std/Io.zig+84-11
......@@ -148,7 +148,10 @@ pub const VTable = struct {
148148 futexWaitUncancelable: *const fn (?*anyopaque, ptr: *const u32, expected: u32) void,
149149 futexWake: *const fn (?*anyopaque, ptr: *const u32, max_waiters: u32) void,
150150
151 operate: *const fn (?*anyopaque, []Operation) void,
151 batch: *const fn (?*anyopaque, []Operation) ConcurrentError!void,
152 batchSubmit: *const fn (?*anyopaque, *Batch) void,
153 batchWait: *const fn (?*anyopaque, *Batch, resubmissions: []const usize, Timeout) Batch.WaitError!usize,
154 batchCancel: *const fn (?*anyopaque, *Batch) void,
152155
153156 dirCreateDir: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirError!void,
154157 dirCreateDirPath: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirPathError!Dir.CreatePathStatus,
......@@ -244,26 +247,96 @@ pub const VTable = struct {
244247};
245248
246249pub const Operation = union(enum) {
247 noop,
250 noop: Noop,
248251 file_read_streaming: FileReadStreaming,
249252
253 pub const Noop = struct {
254 reserved: [2]usize,
255 status: Status(void) = .{ .result = {} },
256 };
257
258 /// Returns 0 on end of stream.
250259 pub const FileReadStreaming = struct {
251260 file: File,
252261 data: []const []u8,
253 /// Causes `result` to return `error.WouldBlock` instead of blocking.
254 nonblocking: bool = false,
255 /// Returns 0 on end of stream.
256 result: File.Reader.Error!usize,
262 status: Status(File.Reader.Error!usize) = .{ .unstarted = {} },
257263 };
264
265 pub fn Status(Result: type) type {
266 return union {
267 unstarted: void,
268 pending: usize,
269 result: Result,
270 };
271 }
258272};
259273
260/// Performs all `operations` in a non-deterministic order. Returns after all
261/// `operations` have been completed. The degree to which the operations are
262/// performed concurrently is determined by the `Io` implementation.
263pub fn operate(io: Io, operations: []Operation) void {
264 return io.vtable.operate(io.userdata, operations);
274/// Performs all `operations` in an unspecified order, concurrently.
275///
276/// Returns after all `operations` have been completed. If the operations could
277/// not be completed concurrently, returns `error.ConcurrencyUnavailable`.
278///
279/// With this API, it is rare for concurrency to not be available. Even a
280/// single-threaded `Io` implementation can, for example, take advantage of
281/// poll() to implement this. Note that poll() is fallible however.
282///
283/// If `operations.len` is one, `error.ConcurrencyUnavailable` is unreachable.
284///
285/// On entry, all operations must already have `.status = .unstarted` except
286/// noops must have `.status = .{ .result = {} }`, to safety check the state
287/// transitions.
288///
289/// On return, all operations have `.status = .{ .result = ... }`.
290pub fn batch(io: Io, operations: []Operation) ConcurrentError!void {
291 return io.vtable.batch(io.userdata, operations);
292}
293
294/// Performs one `Operation`.
295pub fn operate(io: Io, operation: *Operation) void {
296 return io.vtable.batch(io.userdata, (operation)[0..1]) catch unreachable;
265297}
266298
299/// Submits many operations together without waiting for all of them to
300/// complete.
301///
302/// This is a low-level abstraction based on `Operation`. For a higher
303/// level API that operates on `Future`, see `Select`.
304pub const Batch = struct {
305 operations: []Operation,
306 index: usize,
307 reserved: ?*anyopaque,
308
309 pub fn init(operations: []Operation) Batch {
310 return .{ .operations = operations, .index = 0, .reserved = null };
311 }
312
313 /// Submits all non-noop `operations`.
314 pub fn submit(b: *Batch, io: Io) void {
315 return io.vtable.batchSubmit(io.userdata, b);
316 }
317
318 pub const WaitError = ConcurrentError || Cancelable || Timeout.Error;
319
320 /// Resubmits the previously completed or noop-initialized `operations` at
321 /// indexes given by `resubmissions`. This set of indexes typically will be empty
322 /// on the first call to `await` since all operations have already been
323 /// submitted via `async`.
324 ///
325 /// Returns the index of a completed `Operation`, or `operations.len` if
326 /// all operations are completed.
327 ///
328 /// When `error.Canceled` is returned, all operations have already completed.
329 pub fn wait(b: *Batch, io: Io, resubmissions: []const usize, timeout: Timeout) WaitError!usize {
330 return io.vtable.batchWait(io.userdata, b, resubmissions, timeout);
331 }
332
333 /// Returns after all `operations` have completed. Each operation
334 /// independently may or may not have been canceled.
335 pub fn cancel(b: *Batch, io: Io) void {
336 return io.vtable.batchCancel(io.userdata, b);
337 }
338};
339
267340pub const Limit = enum(usize) {
268341 nothing = 0,
269342 unlimited = math.maxInt(usize),
lib/std/Io/File.zig+2-3
......@@ -529,10 +529,9 @@ pub fn readStreaming(file: File, io: Io, buffer: []const []u8) Reader.Error!usiz
529529 var operation: Io.Operation = .{ .file_read_streaming = .{
530530 .file = file,
531531 .data = buffer,
532 .result = undefined,
533532 } };
534 io.vtable.operate(io.userdata, (&operation)[0..1]);
535 return operation.file_read_streaming.result;
533 io.operate(&operation);
534 return operation.file_read_streaming.status.result;
536535}
537536
538537pub const ReadPositionalError = Reader.Error || error{Unseekable};
lib/std/Io/Threaded.zig+171-85
......@@ -1438,7 +1438,10 @@ pub fn io(t: *Threaded) Io {
14381438 .futexWaitUncancelable = futexWaitUncancelable,
14391439 .futexWake = futexWake,
14401440
1441 .operate = operate,
1441 .batch = batch,
1442 .batchSubmit = batchSubmit,
1443 .batchWait = batchWait,
1444 .batchCancel = batchCancel,
14421445
14431446 .dirCreateDir = dirCreateDir,
14441447 .dirCreateDirPath = dirCreateDirPath,
......@@ -1591,7 +1594,10 @@ pub fn ioBasic(t: *Threaded) Io {
15911594 .futexWaitUncancelable = futexWaitUncancelable,
15921595 .futexWake = futexWake,
15931596
1594 .operate = operate,
1597 .batch = batch,
1598 .batchSubmit = batchSubmit,
1599 .batchWait = batchWait,
1600 .batchCancel = batchCancel,
15951601
15961602 .dirCreateDir = dirCreateDir,
15971603 .dirCreateDirPath = dirCreateDirPath,
......@@ -2269,107 +2275,187 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
22692275 Thread.futexWake(ptr, max_waiters);
22702276}
22712277
2272fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {
2278fn batchSubmit(userdata: ?*anyopaque, b: *Io.Batch) void {
22732279 const t: *Threaded = @ptrCast(@alignCast(userdata));
22742280 _ = t;
2281 _ = b;
2282 return;
2283}
2284
2285fn operate(op: *Io.Operation) void {
2286 switch (op.*) {
2287 .noop => {},
2288 .file_read_streaming => |*o| o.status = .{ .result = fileReadStreaming(o.file, o.data) },
2289 }
2290}
22752291
2292fn batchWait(
2293 userdata: ?*anyopaque,
2294 b: *Io.Batch,
2295 resubmissions: []const usize,
2296 timeout: Io.Timeout,
2297) Io.Batch.WaitError!usize {
2298 _ = resubmissions;
2299 const t: *Threaded = @ptrCast(@alignCast(userdata));
2300 const operations = b.operations;
2301 if (operations.len == 1) {
2302 operate(&operations[0]);
2303 return b.operations.len;
2304 }
22762305 if (is_windows) @panic("TODO");
22772306
22782307 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
22792308 var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index
2280 var operation_index: usize = 0;
2281
2282 while (operation_index < operations.len) {
2283 var poll_i: usize = 0;
2284 while (operation_index < operations.len) : (operation_index += 1) {
2285 switch (operations[operation_index]) {
2286 .noop => continue,
2287 .file_read_streaming => |*o| {
2288 if (o.nonblocking) {
2289 o.result = error.WouldBlock;
2290 poll_buffer[poll_i] = .{
2291 .fd = o.file.handle,
2292 .events = posix.POLL.IN,
2293 .revents = 0,
2294 };
2295 if (map_buffer.len - poll_i == 0) break;
2296 map_buffer[poll_i] = @intCast(operation_index);
2297 poll_i += 1;
2298 } else {
2299 o.result = fileReadStreaming(o.file, o.data) catch |err| switch (err) {
2300 error.Canceled => {
2301 setOperationsError(operations[operation_index..], error.Canceled);
2302 return;
2303 },
2304 else => err,
2305 };
2306 }
2307 },
2308 }
2309 }
2309 var poll_i: usize = 0;
2310
2311 for (operations, 0..) |*op, operation_index| switch (op.*) {
2312 .noop => continue,
2313 .file_read_streaming => |*o| {
2314 if (poll_buffer.len - poll_i == 0) return error.ConcurrencyUnavailable;
2315 poll_buffer[poll_i] = .{
2316 .fd = o.file.handle,
2317 .events = posix.POLL.IN,
2318 .revents = 0,
2319 };
2320 map_buffer[poll_i] = @intCast(operation_index);
2321 poll_i += 1;
2322 },
2323 };
23102324
2311 if (poll_i == 0) {
2312 @branchHint(.likely);
2313 return;
2325 if (poll_i == 0) return operations.len;
2326
2327 const t_io = ioBasic(t);
2328 const deadline = timeout.toDeadline(t_io) catch return error.UnsupportedClock;
2329 const max_poll_ms = std.math.maxInt(i32);
2330
2331 while (true) {
2332 const timeout_ms: i32 = if (deadline) |d| t: {
2333 const duration = d.durationFromNow(t_io) catch return error.UnsupportedClock;
2334 if (duration.raw.nanoseconds <= 0) return error.Timeout;
2335 break :t @intCast(@min(max_poll_ms, duration.raw.toMilliseconds()));
2336 } else -1;
2337 const syscall = try Syscall.start();
2338 const rc = posix.system.poll(&poll_buffer, poll_i, timeout_ms);
2339 syscall.finish();
2340 switch (posix.errno(rc)) {
2341 .SUCCESS => {
2342 if (rc == 0) {
2343 // Although spurious timeouts are OK, when no deadline is
2344 // passed we must not return `error.Timeout`.
2345 if (deadline == null) continue;
2346 return error.Timeout;
2347 }
2348 for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, i| {
2349 if (poll_fd.revents == 0) continue;
2350 operate(&operations[i]);
2351 return i;
2352 }
2353 },
2354 .INTR => continue,
2355 else => return error.ConcurrencyUnavailable,
23142356 }
2357 }
2358}
23152359
2316 while (true) {
2317 const syscall = Syscall.start() catch |err| switch (err) {
2318 error.Canceled => {
2319 setPollOperationsError(operations, map_buffer[0..poll_i], error.Canceled);
2320 setOperationsError(operations[operation_index..], error.Canceled);
2321 return;
2322 },
2360fn batchCancel(userdata: ?*anyopaque, b: *Io.Batch) void {
2361 const t: *Threaded = @ptrCast(@alignCast(userdata));
2362 _ = t;
2363 _ = b;
2364 return;
2365}
2366
2367fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!void {
2368 const t: *Threaded = @ptrCast(@alignCast(userdata));
2369 _ = t;
2370
2371 if (operations.len == 1) {
2372 @branchHint(.likely);
2373 return operate(&operations[0]);
2374 }
2375
2376 if (is_windows) @panic("TODO");
2377
2378 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
2379 var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index
2380 var poll_i: usize = 0;
2381
2382 for (operations, 0..) |*op, operation_index| switch (op.*) {
2383 .noop => continue,
2384 .file_read_streaming => |*o| {
2385 if (poll_buffer.len - poll_i == 0) return error.ConcurrencyUnavailable;
2386 poll_buffer[poll_i] = .{
2387 .fd = o.file.handle,
2388 .events = posix.POLL.IN,
2389 .revents = 0,
23232390 };
2324 const poll_rc = posix.system.poll(&poll_buffer, poll_i, -1);
2325 syscall.finish();
2326 switch (posix.errno(poll_rc)) {
2327 .SUCCESS => {
2328 if (poll_rc == 0) {
2329 // Spurious timeout; handle same as INTR.
2330 continue;
2331 }
2332 for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, i| {
2333 if (poll_fd.revents == 0) continue;
2334 switch (operations[i]) {
2335 .noop => unreachable,
2336 .file_read_streaming => |*o| {
2337 o.result = fileReadStreaming(o.file, o.data);
2338 },
2339 }
2340 }
2341 break;
2342 },
2343 .INTR => continue,
2344 .NOMEM => {
2345 setPollOperationsError(operations, map_buffer[0..poll_i], error.SystemResources);
2346 break;
2347 },
2348 else => {
2349 setPollOperationsError(operations, map_buffer[0..poll_i], error.Unexpected);
2350 break;
2351 },
2352 }
2391 map_buffer[poll_i] = @intCast(operation_index);
2392 poll_i += 1;
2393 },
2394 };
2395
2396 const polls = poll_buffer[0..poll_i];
2397 const map = map_buffer[0..poll_i];
2398
2399 var pending = poll_i;
2400 while (pending > 1) {
2401 const syscall = Syscall.start() catch |err| switch (err) {
2402 error.Canceled => {
2403 if (!setOperationsError(operations, polls, map, error.Canceled))
2404 recancelInner();
2405 return;
2406 },
2407 };
2408 const rc = posix.system.poll(polls.ptr, polls.len, -1);
2409 syscall.finish();
2410 switch (posix.errno(rc)) {
2411 .SUCCESS => {
2412 if (rc == 0) {
2413 // Spurious timeout; handle the same as INTR.
2414 continue;
2415 }
2416 for (polls, map) |*poll_fd, i| {
2417 if (poll_fd.revents == 0) continue;
2418 poll_fd.fd = -1;
2419 pending -= 1;
2420 operate(&operations[i]);
2421 }
2422 },
2423 .INTR => continue,
2424 .NOMEM => {
2425 assert(setOperationsError(operations, polls, map, error.SystemResources));
2426 return;
2427 },
2428 else => {
2429 assert(setOperationsError(operations, polls, map, error.Unexpected));
2430 return;
2431 },
23532432 }
23542433 }
2434
2435 if (pending == 1) for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, i| {
2436 if (poll_fd.fd == -1) continue;
2437 operate(&operations[i]);
2438 };
23552439}
23562440
2357fn setPollOperationsError(
2441fn setOperationsError(
23582442 operations: []Io.Operation,
2443 polls: []const posix.pollfd,
23592444 map: []const u8,
23602445 err: error{ Canceled, SystemResources, Unexpected },
2361) void {
2362 for (map) |operation_index| switch (operations[operation_index]) {
2363 .noop => unreachable,
2364 inline else => |*o| o.result = err,
2365 };
2366}
2367
2368fn setOperationsError(operations: []Io.Operation, err: error{ Canceled, SystemResources, Unexpected }) void {
2369 for (operations) |*op| switch (op.*) {
2370 .noop => unreachable,
2371 inline else => |*o| o.result = err,
2372 };
2446) bool {
2447 var marked = false;
2448 for (polls, map) |*poll_fd, i| {
2449 if (poll_fd.fd == -1) continue;
2450 switch (operations[i]) {
2451 .noop => unreachable,
2452 inline else => |*o| {
2453 o.status = .{ .result = err };
2454 marked = true;
2455 },
2456 }
2457 }
2458 return marked;
23732459}
23742460
23752461const dirCreateDir = switch (native_os) {
lib/std/process.zig+9-5
......@@ -465,9 +465,7 @@ pub fn spawnPath(io: Io, dir: Io.Dir, options: SpawnOptions) SpawnError!Child {
465465 return io.vtable.processSpawnPath(io.userdata, dir, options);
466466}
467467
468pub const RunError = posix.GetCwdError || posix.ReadError || SpawnError || posix.PollError || error{
469 StreamTooLong,
470};
468pub const RunError = SpawnError || Child.CollectOutputError;
471469
472470pub const RunOptions = struct {
473471 argv: []const []const u8,
......@@ -545,9 +543,15 @@ pub fn run(gpa: Allocator, io: Io, options: RunOptions) RunError!RunResult {
545543 .stderr_limit = options.stderr_limit,
546544 });
547545
546 const stdout_slice = try stdout.toOwnedSlice(gpa);
547 errdefer gpa.free(stdout_slice);
548
549 const stderr_slice = try stderr.toOwnedSlice(gpa);
550 errdefer gpa.free(stderr_slice);
551
548552 return .{
549 .stdout = try stdout.toOwnedSlice(gpa),
550 .stderr = try stderr.toOwnedSlice(gpa),
553 .stdout = stdout_slice,
554 .stderr = stderr_slice,
551555 .term = try child.wait(io),
552556 };
553557}
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}