authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2026-01-10 15:34:36-05:00
committergravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2026-01-12 20:05:15-05:00
logf633a4845aeea5f390d4ea1b05e5d957efb02966
treebcc9fb652281816f12f6aae1bc7b72d3cc7addbb
parent078a19cf3111726f4a3861da9d4e297eac7dc585

Io: add ring to `Batch` API

5 files changed, 259 insertions(+), 159 deletions(-)

lib/std/Io.zig+122-48
......@@ -148,9 +148,8 @@ 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 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,
151 operate: *const fn (?*anyopaque, *Operation) Cancelable!void,
152 batchWait: *const fn (?*anyopaque, *Batch, Timeout) Batch.WaitError!void,
154153 batchCancel: *const fn (?*anyopaque, *Batch) void,
155154
156155 dirCreateDir: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirError!void,
......@@ -252,48 +251,51 @@ pub const Operation = union(enum) {
252251
253252 pub const Noop = struct {
254253 reserved: [2]usize = .{ 0, 0 },
255 status: Status(void) = .{ .result = {} },
254 status: Status(void) = .{ .unstarted = {} },
256255 };
257256
258257 /// Returns 0 on end of stream.
259258 pub const FileReadStreaming = struct {
260259 file: File,
261260 data: []const []u8,
262 status: Status(File.Reader.Error!usize) = .{ .unstarted = {} },
261 status: Status(Error!usize) = .{ .unstarted = {} },
262
263 pub const Error = error{
264 InputOutput,
265 SystemResources,
266 IsDir,
267 BrokenPipe,
268 ConnectionResetByPeer,
269 Timeout,
270 /// In WASI, EBADF is mapped to this error because it is returned when
271 /// trying to read a directory file descriptor as if it were a file.
272 NotOpenForReading,
273 SocketUnconnected,
274 /// Non-blocking has been enabled, and reading from the file descriptor
275 /// would block.
276 WouldBlock,
277 /// In WASI, this error occurs when the file descriptor does
278 /// not hold the required rights to read from it.
279 AccessDenied,
280 /// Unable to read file due to lock. Depending on the `Io` implementation,
281 /// reading from a locked file may return this error, or may ignore the
282 /// lock.
283 LockViolation,
284 } || Io.UnexpectedError;
263285 };
264286
265287 pub fn Status(Result: type) type {
266288 return union {
267289 unstarted: void,
268 pending: usize,
290 pending: *Batch,
269291 result: Result,
270292 };
271293 }
272294};
273295
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
294296/// Performs one `Operation`.
295pub fn operate(io: Io, operation: *Operation) void {
296 return io.vtable.batch(io.userdata, (operation)[0..1]) catch unreachable;
297pub fn operate(io: Io, operation: *Operation) Cancelable!void {
298 return io.vtable.operate(io.userdata, operation) catch unreachable;
297299}
298300
299301/// Submits many operations together without waiting for all of them to
......@@ -303,35 +305,107 @@ pub fn operate(io: Io, operation: *Operation) void {
303305/// level API that operates on `Future`, see `Select`.
304306pub const Batch = struct {
305307 operations: []Operation,
306 index: usize,
307 reserved: ?*anyopaque,
308 ring: [*]u32,
309 user: struct {
310 submit_tail: RingIndex,
311 complete_head: RingIndex,
312 complete_tail: RingIndex,
313 },
314 impl: struct {
315 submit_head: RingIndex,
316 submit_tail: RingIndex,
317 complete_tail: RingIndex,
318 reserved: ?*anyopaque,
319 },
320
321 pub const RingIndex = enum(u32) {
322 _,
323
324 pub fn index(ri: RingIndex, len: u31) u31 {
325 const i = @intFromEnum(ri);
326 assert(i < @as(u32, len) * 2);
327 return @intCast(if (i < len) i else i - len);
328 }
329
330 pub fn prev(ri: RingIndex, len: u31) RingIndex {
331 const i = @intFromEnum(ri);
332 const double_len = @as(u32, len) * 2;
333 assert(i <= double_len);
334 return @enumFromInt((if (i > 0) i else double_len) - 1);
335 }
336
337 pub fn next(ri: RingIndex, len: u31) RingIndex {
338 const i = @intFromEnum(ri) + 1;
339 const double_len = @as(u32, len) * 2;
340 assert(i <= double_len);
341 return @enumFromInt(if (i < double_len) i else 0);
342 }
343 };
344
345 pub const WaitError = ConcurrentError || Cancelable || Timeout.Error;
308346
309 pub fn init(operations: []Operation) Batch {
310 return .{ .operations = operations, .index = 0, .reserved = null };
347 pub fn init(operations: []Operation, ring: []u32) Batch {
348 const len: u31 = @intCast(operations.len);
349 assert(ring.len == len);
350 return .{
351 .operations = operations,
352 .ring = ring.ptr,
353 .user = .{
354 .submit_tail = @enumFromInt(0),
355 .complete_head = @enumFromInt(0),
356 .complete_tail = @enumFromInt(0),
357 },
358 .impl = .{
359 .submit_head = @enumFromInt(0),
360 .submit_tail = @enumFromInt(0),
361 .complete_tail = @enumFromInt(0),
362 .reserved = null,
363 },
364 };
311365 }
312366
313 /// Submits all non-noop `operations`.
314 pub fn submit(b: *Batch, io: Io) void {
315 return io.vtable.batchSubmit(io.userdata, b);
367 /// Adds `b.operations[operation]` to the list of submitted operations
368 /// that will be performed when `wait` is called.
369 pub fn add(b: *Batch, operation: usize) void {
370 const tail = b.user.submit_tail;
371 const len: u31 = @intCast(b.operations.len);
372 b.user.submit_tail = tail.next(len);
373 b.ring[0..len][tail.index(len)] = @intCast(operation);
316374 }
317375
318 pub const WaitError = ConcurrentError || Cancelable || Timeout.Error;
376 fn flush(b: *Batch) void {
377 @atomicStore(RingIndex, &b.impl.submit_tail, b.user.submit_tail, .release);
378 }
319379
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.
380 /// Returns `operation` such that `b.operations[operation]` has completed.
381 /// Returns `null` when `wait` should be called.
382 pub fn next(b: *Batch) ?u32 {
383 const head = b.user.complete_head;
384 if (head == b.user.complete_tail) {
385 @branchHint(.unlikely);
386 b.flush();
387 const tail = @atomicLoad(RingIndex, &b.impl.complete_tail, .acquire);
388 if (head == tail) {
389 @branchHint(.unlikely);
390 return null;
391 }
392 assert(head != tail);
393 b.user.complete_tail = tail;
394 }
395 const len: u31 = @intCast(b.operations.len);
396 b.user.complete_head = head.next(len);
397 return b.ring[0..len][head.index(len)];
398 }
399
400 /// Starts work on any submitted operations and returns when at least one has completeed.
327401 ///
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);
402 /// Returns `error.Timeout` if `timeout` expires first.
403 pub fn wait(b: *Batch, io: Io, timeout: Timeout) WaitError!void {
404 return io.vtable.batchWait(io.userdata, b, timeout);
331405 }
332406
333 /// Returns after all `operations` have completed. Each operation
334 /// independently may or may not have been canceled.
407 /// Returns after all `operations` have completed. Operations which have not completed
408 /// after this function returns were successfully dropped and had no side effects.
335409 pub fn cancel(b: *Batch, io: Io) void {
336410 return io.vtable.batchCancel(io.userdata, b);
337411 }
lib/std/Io/File.zig+1-1
......@@ -530,7 +530,7 @@ pub fn readStreaming(file: File, io: Io, buffer: []const []u8) Reader.Error!usiz
530530 .file = file,
531531 .data = buffer,
532532 } };
533 io.operate(&operation);
533 try io.operate(&operation);
534534 return operation.file_read_streaming.status.result;
535535}
536536
lib/std/Io/File/Reader.zig+1-22
......@@ -26,28 +26,7 @@ size_err: ?SizeError = null,
2626seek_err: ?SeekError = null,
2727interface: Io.Reader,
2828
29pub const Error = error{
30 InputOutput,
31 SystemResources,
32 IsDir,
33 BrokenPipe,
34 ConnectionResetByPeer,
35 Timeout,
36 /// In WASI, EBADF is mapped to this error because it is returned when
37 /// trying to read a directory file descriptor as if it were a file.
38 NotOpenForReading,
39 SocketUnconnected,
40 /// Non-blocking has been enabled, and reading from the file descriptor
41 /// would block.
42 WouldBlock,
43 /// In WASI, this error occurs when the file descriptor does
44 /// not hold the required rights to read from it.
45 AccessDenied,
46 /// Unable to read file due to lock. Depending on the `Io` implementation,
47 /// reading from a locked file may return this error, or may ignore the
48 /// lock.
49 LockViolation,
50} || Io.Cancelable || Io.UnexpectedError;
29pub const Error = Io.Operation.FileReadStreaming.Error || Io.Cancelable;
5130
5231pub const SizeError = std.os.windows.GetFileSizeError || File.StatError || error{
5332 /// 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
......@@ -1438,8 +1438,7 @@ pub fn io(t: *Threaded) Io {
14381438 .futexWaitUncancelable = futexWaitUncancelable,
14391439 .futexWake = futexWake,
14401440
1441 .batch = batch,
1442 .batchSubmit = batchSubmit,
1441 .operate = operate,
14431442 .batchWait = batchWait,
14441443 .batchCancel = batchCancel,
14451444
......@@ -1594,8 +1593,7 @@ pub fn ioBasic(t: *Threaded) Io {
15941593 .futexWaitUncancelable = futexWaitUncancelable,
15951594 .futexWake = futexWake,
15961595
1597 .batch = batch,
1598 .batchSubmit = batchSubmit,
1596 .operate = operate,
15991597 .batchWait = batchWait,
16001598 .batchCancel = batchCancel,
16011599
......@@ -2275,59 +2273,82 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
22752273 Thread.futexWake(ptr, max_waiters);
22762274}
22772275
2278fn batchSubmit(userdata: ?*anyopaque, b: *Io.Batch) void {
2276fn operate(userdata: ?*anyopaque, op: *Io.Operation) Io.Cancelable!void {
22792277 const t: *Threaded = @ptrCast(@alignCast(userdata));
22802278 _ = t;
2281 _ = b;
2282 return;
2283}
2284
2285fn operate(op: *Io.Operation) void {
22862279 switch (op.*) {
2287 .noop => {},
2288 .file_read_streaming => |*o| o.status = .{ .result = fileReadStreaming(o.file, o.data) },
2280 .noop => |*o| {
2281 _ = o.status.unstarted;
2282 o.status = .{ .result = {} };
2283 },
2284 .file_read_streaming => |*o| {
2285 _ = o.status.unstarted;
2286 o.status = .{ .result = fileReadStreaming(o.file, o.data) catch |err| switch (err) {
2287 error.Canceled => return error.Canceled,
2288 else => |e| e,
2289 } };
2290 },
22892291 }
22902292}
22912293
2292fn batchWait(
2293 userdata: ?*anyopaque,
2294 b: *Io.Batch,
2295 resubmissions: []const usize,
2296 timeout: Io.Timeout,
2297) Io.Batch.WaitError!usize {
2298 _ = resubmissions;
2294fn batchWait(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.WaitError!void {
22992295 const t: *Threaded = @ptrCast(@alignCast(userdata));
23002296 const operations = b.operations;
2301 if (operations.len == 1) {
2302 operate(&operations[0]);
2303 return b.operations.len;
2297 const len: u31 = @intCast(operations.len);
2298 const ring = b.ring[0..len];
2299 var submit_head = b.impl.submit_head;
2300 const submit_tail = b.user.submit_tail;
2301 b.impl.submit_tail = submit_tail;
2302 var complete_tail = b.impl.complete_tail;
2303 var map_buffer: [poll_buffer_len]u32 = undefined; // poll_buffer index to operations index
2304 var poll_i: usize = 0;
2305 defer {
2306 for (map_buffer[0..poll_i]) |op| {
2307 submit_head = submit_head.prev(len);
2308 ring[submit_head.index(len)] = op;
2309 }
2310 b.impl.submit_head = submit_head;
2311 b.impl.complete_tail = complete_tail;
2312 b.user.complete_tail = complete_tail;
23042313 }
23052314 if (is_windows) @panic("TODO");
2306
23072315 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
2308 var map_buffer: [poll_buffer_len]u8 = undefined; // poll_buffer index to operations index
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;
2316 while (submit_head != submit_tail) : (submit_head = submit_head.next(len)) {
2317 const op = ring[submit_head.index(len)];
2318 const operation = &operations[op];
2319 switch (operation.*) {
2320 else => {
2321 try operate(t, operation);
2322 ring[complete_tail.index(len)] = op;
2323 complete_tail = complete_tail.next(len);
2324 },
2325 .file_read_streaming => |*o| {
2326 _ = o.status.unstarted;
2327 if (poll_buffer.len - poll_i == 0) return error.ConcurrencyUnavailable;
2328 poll_buffer[poll_i] = .{
2329 .fd = o.file.handle,
2330 .events = posix.POLL.IN,
2331 .revents = 0,
2332 };
2333 map_buffer[poll_i] = op;
2334 poll_i += 1;
2335 },
2336 }
2337 }
2338 switch (poll_i) {
2339 0 => return,
2340 1 => if (timeout == .none) {
2341 const op = map_buffer[0];
2342 try operate(t, &operations[op]);
2343 ring[complete_tail.index(len)] = op;
2344 complete_tail = complete_tail.next(len);
2345 return;
23222346 },
2323 };
2324
2325 if (poll_i == 0) return operations.len;
2326
2347 else => {},
2348 }
23272349 const t_io = ioBasic(t);
23282350 const deadline = timeout.toDeadline(t_io) catch return error.UnsupportedClock;
23292351 const max_poll_ms = std.math.maxInt(i32);
2330
23312352 while (true) {
23322353 const timeout_ms: i32 = if (deadline) |d| t: {
23332354 const duration = d.durationFromNow(t_io) catch return error.UnsupportedClock;
......@@ -2345,11 +2366,24 @@ fn batchWait(
23452366 if (deadline == null) continue;
23462367 return error.Timeout;
23472368 }
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;
2369 var canceled = false;
2370 for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, op| {
2371 if (poll_fd.revents == 0) {
2372 submit_head = submit_head.prev(len);
2373 ring[submit_head.index(len)] = op;
2374 } else {
2375 operate(t, &operations[op]) catch |err| switch (err) {
2376 error.Canceled => {
2377 canceled = true;
2378 continue;
2379 },
2380 };
2381 ring[complete_tail.index(len)] = op;
2382 complete_tail = complete_tail.next(len);
2383 }
23522384 }
2385 poll_i = 0;
2386 return if (canceled) error.Canceled;
23532387 },
23542388 .INTR => continue,
23552389 else => return error.ConcurrencyUnavailable,
......@@ -2359,9 +2393,27 @@ fn batchWait(
23592393
23602394fn batchCancel(userdata: ?*anyopaque, b: *Io.Batch) void {
23612395 const t: *Threaded = @ptrCast(@alignCast(userdata));
2362 _ = t;
2363 _ = b;
2364 return;
2396 const operations = b.operations;
2397 const len: u31 = @intCast(operations.len);
2398 const ring = b.ring[0..len];
2399 var submit_head = b.impl.submit_head;
2400 const submit_tail = b.user.submit_tail;
2401 b.impl.submit_tail = submit_tail;
2402 var complete_tail = b.impl.complete_tail;
2403 while (submit_head != submit_tail) : (submit_head = submit_head.next(len)) {
2404 const op = ring[submit_head.index(len)];
2405 switch (operations[op]) {
2406 .noop => {
2407 operate(t, &operations[op]) catch unreachable;
2408 ring[complete_tail.index(len)] = op;
2409 complete_tail = complete_tail.next(len);
2410 },
2411 .file_read_streaming => |*o| _ = o.status.unstarted,
2412 }
2413 }
2414 b.impl.submit_head = submit_tail;
2415 b.impl.complete_tail = complete_tail;
2416 b.user.complete_tail = complete_tail;
23652417}
23662418
23672419fn batch(userdata: ?*anyopaque, operations: []Io.Operation) Io.ConcurrentError!void {
......@@ -9910,6 +9962,7 @@ fn nowWasi(clock: Io.Clock) Io.Clock.Error!Io.Timestamp {
99109962
99119963fn sleep(userdata: ?*anyopaque, timeout: Io.Timeout) Io.SleepError!void {
99129964 const t: *Threaded = @ptrCast(@alignCast(userdata));
9965 if (timeout == .none) return;
99139966 if (use_parking_sleep) return parking_sleep.sleep(timeout);
99149967 if (native_os == .wasi) return sleepWasi(t, timeout);
99159968 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}