std.Io: introduce batching and operations API, satisfying the "poll" use case #30743

andrewrk commented 2026年01月08日 04:41:10 +01:00

This branch introduces Io.Operation, std.Io.operate, std.Io.Batch, and implements them for FileReadStreaming. In std.Io.Threaded, the implementation is based on poll().

The idea here is that every VTable function that makes sense to have error{Canceled,Timeout} in its error set would be moved into this Operation tagged union. Note that the higher level API, e.g. std.Io.File.readFileStreaming is unchanged. We expect this to be lowerable in a reasonable fashion using IoUring, Kqueue, and Windows.

Motivation

• general support for batching, timeouts, and non-blocking across all I/O operations
• ability to migrate code that used poll() without causing error.ConcurrencyUnavailable on -fsingle-threaded builds
  • for example, processing stdout and stderr of child processes in a single-threaded application

Demonstration

Using std.process.run to collect both stdout and stderr in a single-threaded program using std.Threaded.Io. The point of this example is, if the parent naively reads from either stdout or stderr non-concurrently, it will deadlock.

child.zig:

conststd=@import("std");constIo=std.Io;pubfnmain(init:std.process.Init)!void{constio=init.io;varstdout:Io.File.Writer=.initStreaming(.stdout(),io,&.{});varstderr:Io.File.Writer=.initStreaming(.stderr(),io,&.{});trystdout.interface.splatByteAll('A',5000);trystderr.interface.splatByteAll('B',5000);trystdout.interface.splatByteAll('C',5000);trystderr.interface.splatByteAll('D',5000);}

parent.zig

conststd=@import("std");constIo=std.Io;pubfnmain(init:std.process.Init)!void{constio=init.io;constarena=init.arena.allocator();constresult=trystd.process.run(arena,io,.{.argv=&.{"./child"},});std.debug.print("stdout:\n{s}\nstderr:\n{s}",.{result.stdout,result.stderr});}

$ stage3/bin/zig build-exe parent.zig -fsingle-threaded
$ ./parent
stdout:
AAAAAA...(x5000)CCCCC....(x5000)
stderr:
BBBBBB...(x5000)DDDDD....(x5000)
$ strace ./parent
...
pipe2([8, 9], O_CLOEXEC) = 0
fork() = 3305035
close(9) = 0
close(4) = 0
close(6) = 0
munmap(0x7fb1cec93000, 53248) = 0
munmap(0x7fb1ceb40000, 131072) = 0
read(8, "", 8) = 0
close(8) = 0
mmap(0x7fb1ceca0000, 4096, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fb1cf009000
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}])
readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=129}], 1) = 129
readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=129}], 1) = 129
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}])
readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=194}], 1) = 194
readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=194}], 1) = 194
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}])
readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=291}], 1) = 291
readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=291}], 1) = 291
mremap(0x7fb1cf009000, 4096, 8192, 0) = -1 ENOMEM (Cannot allocate memory)
mmap(0x7fb1cf00a000, 8192, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fb1cf007000
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}])
readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=436}], 1) = 436
readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=436}], 1) = 436
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}])
readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=654}], 1) = 654
readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=654}], 1) = 654
mremap(0x7fb1cf007000, 8192, 12288, 0) = -1 ENOMEM (Cannot allocate memory)
mmap(0x7fb1cf009000, 16384, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fb1cf003000
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}])
readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=981}], 1) = 981
readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=981}], 1) = 981
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}])
readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=1472}], 1) = 1472
readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=1472}], 1) = 1472
mremap(0x7fb1cf003000, 16384, 20480, 0) = -1 ENOMEM (Cannot allocate memory)
mmap(0x7fb1cf007000, 32768, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fb1ced42000
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}])
readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=2208}], 1) = 2208
readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=2208}], 1) = 2208
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}])
readv(3, [{iov_base="CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC"..., iov_len=3312}], 1) = 3312
readv(5, [{iov_base="DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD"..., iov_len=3312}], 1) = 3312
mremap(0x7fb1ced42000, 32768, 49152, 0) = -1 ENOMEM (Cannot allocate memory)
mmap(0x7fb1ced4a000, 73728, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fb1cec8e000
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}])
readv(3, [{iov_base="CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC"..., iov_len=4968}], 1) = 323
readv(5, [{iov_base="DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD"..., iov_len=4968}], 1) = 323
poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLHUP}, {fd=5, revents=POLLHUP}])
--- SIGCHLD {si_signo=SIGCHLD, si_code=CLD_EXITED, si_pid=3305035, si_uid=1000, si_status=0, si_utime=0, si_stime=0} ---
...

Merge Blockers

• fix bugs and failures
• Windows implementation

Followup Issues

• move more stuff from VTable to Operation

---

closes https://github.com/ziglang/zig/issues/25753

This branch introduces `Io.Operation`, `std.Io.operate`, `std.Io.Batch`, and implements them for `FileReadStreaming`. In `std.Io.Threaded`, the implementation is based on poll(). The idea here is that every VTable function that makes sense to have `error{Canceled,Timeout}` in its error set would be moved into this `Operation` tagged union. Note that the higher level API, e.g. `std.Io.File.readFileStreaming` is unchanged. We expect this to be lowerable in a reasonable fashion using IoUring, Kqueue, and Windows. ## Motivation * general support for batching, timeouts, and non-blocking across all I/O operations * ability to migrate code that used `poll()` without causing `error.ConcurrencyUnavailable` on `-fsingle-threaded` builds - for example, processing stdout and stderr of child processes in a single-threaded application ## Demonstration Using `std.process.run` to collect both stdout and stderr in a single-threaded program using `std.Threaded.Io`. The point of this example is, if the parent naively reads from either stdout or stderr non-concurrently, it will deadlock. child.zig: ```zig const std = @import("std"); const Io = std.Io; pub fn main(init: std.process.Init) !void { const io = init.io; var stdout: Io.File.Writer = .initStreaming(.stdout(), io, &.{}); var stderr: Io.File.Writer = .initStreaming(.stderr(), io, &.{}); try stdout.interface.splatByteAll('A', 5000); try stderr.interface.splatByteAll('B', 5000); try stdout.interface.splatByteAll('C', 5000); try stderr.interface.splatByteAll('D', 5000); } ``` parent.zig ```zig const std = @import("std"); const Io = std.Io; pub fn main(init: std.process.Init) !void { const io = init.io; const arena = init.arena.allocator(); const result = try std.process.run(arena, io, .{ .argv = &.{"./child"}, }); std.debug.print("stdout:\n{s}\nstderr:\n{s}", .{ result.stdout, result.stderr }); } ``` ``` $ stage3/bin/zig build-exe parent.zig -fsingle-threaded $ ./parent stdout: AAAAAA...(x5000)CCCCC....(x5000) stderr: BBBBBB...(x5000)DDDDD....(x5000) $ strace ./parent ... pipe2([8, 9], O_CLOEXEC) = 0 fork() = 3305035 close(9) = 0 close(4) = 0 close(6) = 0 munmap(0x7fb1cec93000, 53248) = 0 munmap(0x7fb1ceb40000, 131072) = 0 read(8, "", 8) = 0 close(8) = 0 mmap(0x7fb1ceca0000, 4096, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fb1cf009000 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}]) readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=129}], 1) = 129 readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=129}], 1) = 129 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}]) readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=194}], 1) = 194 readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=194}], 1) = 194 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}]) readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=291}], 1) = 291 readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=291}], 1) = 291 mremap(0x7fb1cf009000, 4096, 8192, 0) = -1 ENOMEM (Cannot allocate memory) mmap(0x7fb1cf00a000, 8192, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fb1cf007000 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}]) readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=436}], 1) = 436 readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=436}], 1) = 436 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}]) readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=654}], 1) = 654 readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=654}], 1) = 654 mremap(0x7fb1cf007000, 8192, 12288, 0) = -1 ENOMEM (Cannot allocate memory) mmap(0x7fb1cf009000, 16384, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fb1cf003000 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}]) readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=981}], 1) = 981 readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=981}], 1) = 981 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}]) readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=1472}], 1) = 1472 readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=1472}], 1) = 1472 mremap(0x7fb1cf003000, 16384, 20480, 0) = -1 ENOMEM (Cannot allocate memory) mmap(0x7fb1cf007000, 32768, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fb1ced42000 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}]) readv(3, [{iov_base="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"..., iov_len=2208}], 1) = 2208 readv(5, [{iov_base="BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"..., iov_len=2208}], 1) = 2208 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}]) readv(3, [{iov_base="CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC"..., iov_len=3312}], 1) = 3312 readv(5, [{iov_base="DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD"..., iov_len=3312}], 1) = 3312 mremap(0x7fb1ced42000, 32768, 49152, 0) = -1 ENOMEM (Cannot allocate memory) mmap(0x7fb1ced4a000, 73728, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fb1cec8e000 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLIN}, {fd=5, revents=POLLIN}]) readv(3, [{iov_base="CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC"..., iov_len=4968}], 1) = 323 readv(5, [{iov_base="DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD"..., iov_len=4968}], 1) = 323 poll([{fd=3, events=POLLIN}, {fd=5, events=POLLIN}], 2, -1) = 2 ([{fd=3, revents=POLLHUP}, {fd=5, revents=POLLHUP}]) --- SIGCHLD {si_signo=SIGCHLD, si_code=CLD_EXITED, si_pid=3305035, si_uid=1000, si_status=0, si_utime=0, si_stime=0} --- ... ``` ## Merge Blockers * fix bugs and failures * Windows implementation ## Followup Issues * move more stuff from VTable to Operation ---- closes https://github.com/ziglang/zig/issues/25753

matklad commented 2026年01月08日 13:44:27 +01:00

+1 to the overall shape. Not only collectOutput now works with a single thread, I'd argue that its implementation is just more natural as a hand-coded state machine, and not as a pair of asyncConcurrent tasks.

Specific interface I am not happy with (but maybe I just misunderstand something). The issue I have with it is that it assumes readiness based IO, not completion based:

pub fn operate(io: Io, operations: []Operation, n_wait: usize, timeout: Timeout) OperateError!void {

If n_wait < operations.len, the interface needs to guarantee that each operation is "atomic" --- either it is completed, or otherwise not started. This works for poll, but doesn't work for io_uring, and arguably, "real world".

E.g, suppose I write low-level code that issues 5 write operations to an actual physical device. The way this works is that I ask the device to start the corresponding DMA operations, and then wait for interrupt which notifies me about completions.

I can wait for first 2 out of 5 operations to complete, but I still need to wait for other 3, I can't just drop them on the floor. The operate interface prevents this for two reasons:

• there isn't affordance to wait for perviously submitted operations
• If we fix that, the problem of memory fragmentation appears ---- If I submit a slice of 1000 ops, 999 complete, but one is slow, I'll still have to keep the entire slice pinned, I can't trivially re-use 999 completed slots.

+1 to the overall shape. Not only `collectOutput` now works with a single thread, I'd argue that its implementation is just more natural as a hand-coded state machine, and not as a pair of `asyncConcurrent` tasks. Specific interface I am not happy with (but maybe I just misunderstand something). The issue I have with it is that it assumes readiness based IO, not completion based: ``` pub fn operate(io: Io, operations: []Operation, n_wait: usize, timeout: Timeout) OperateError!void { ``` If `n_wait < operations.len`, the interface needs to guarantee that each operation is "atomic" --- either it is completed, or otherwise not started. This works for `poll`, but doesn't work for `io_uring`, and arguably, "real world". E.g, suppose I write low-level code that issues 5 write operations to an actual physical device. The way this works is that I ask the device to start the corresponding DMA operations, and then wait for interrupt which notifies me about completions. I can wait for first 2 out of 5 operations to complete, but I still need to wait for other 3, I can't just drop them on the floor. The `operate` interface prevents this for two reasons: * there isn't affordance to wait for _perviously_ submitted operations * If we fix that, the problem of memory fragmentation appears ---- If I submit a slice of 1000 ops, 999 complete, but one is slow, I'll still have to keep the entire slice pinned, I can't trivially re-use 999 completed slots.

matklad commented 2026年01月08日 13:48:21 +01:00

Oh, there's also the poll-vs-epoll big-O problem here: if operations.len is 10^6, but n_wait it 1, you still have to trawl though the million to find out which single op did complete.

Oh, there's also the poll-vs-epoll big-O problem here: if `operations.len` is 10^6, but n_wait it 1, you still have to trawl though the million to find out which single op did complete.

mlugg commented 2026年01月08日 17:23:38 +01:00

I agree with matklad's concerns, but can't really figure out a nice solution to them. While trying to find one, I realised that (IMO) the natural conclusion of this PR's line of thinking is to literally just expose an io_uring style API---it lets you handle completions as soon as they're available, it works with any primitive you can think of, and it allows you to reuse buffer memory. However, I strongly believe that would be a misstep: the whole point of the Io API is to abstract over these low-level event loop mechanisms, so straight-up exposing an io_uring-esque API would seriously muddy the intention behind the interface (and, of course, introduces a lot of extra complexity).

Also, I get the feeling that the use cases for this operate feature are actually pretty rare. I honestly am struggling to think of many cases where I just want to dispatch a bunch of IO operations and don't want non-trivial handling of their individual completions as they come in (if I do want that, then I should just have an async or concurrent task per operation---that's kinda the whole point of the future/group API!). However, I think that there is one very clear counterexample, which is exactly what you've done here: polling child process output streams. That's a real use case, which we really want to be able to use with a dumb single-threaded Io implementation, and it's what every call to std.Io.poll in the Zig repository is doing right now---but I think other use cases where operate would be preferable to async/concurrent are pretty damn rare. (Of course, if someone has a counterexample where they think operate is clearly preferable for something else too, do bring that up! A counterexample might make me rethink my conclusions here).

Having pondered this a little, my current feeling is that the operate API sadly doesn't really fit. It feels like it's trying to bolt on to Io a completely different design for asynchronous operations, and tries to be useful for several things at once, therefore ending up kind of mediocre at them all. I believe that a small, focused poll API (vaguely akin to the old one) would probably be a neater and simpler addition to the interface, which handles the key use case of child processes just as well; reflects the primitives actually available on OSes so that implementations don't have to do anything too complicated; and avoids muddying the waters wrt how you're "supposed" to do certain stuff asynchronously.

I agree with matklad's concerns, but can't really figure out a nice solution to them. While trying to find one, I realised that (IMO) the natural conclusion of this PR's line of thinking is to literally just expose an io_uring style API---it lets you handle completions as soon as they're available, it works with any primitive you can think of, and it allows you to reuse buffer memory. However, I strongly believe that would be a misstep: the whole point of the `Io` API is to abstract over these low-level event loop mechanisms, so straight-up exposing an io_uring-esque API would seriously muddy the intention behind the interface (and, of course, introduces a lot of extra complexity). Also, I get the feeling that the use cases for this `operate` feature are actually pretty rare. I honestly am struggling to think of many cases where I just want to dispatch a bunch of IO operations and *don't* want non-trivial handling of their individual completions as they come in (if I do want that, then I should just have an `async` or `concurrent` task per operation---that's kinda the whole point of the future/group API!). However, I think that there is one very clear counterexample, which is exactly what you've done here: polling child process output streams. That's a real use case, which we really want to be able to use with a dumb single-threaded `Io` implementation, and it's what *every* call to `std.Io.poll` in the Zig repository is doing right now---but I think other use cases where `operate` would be preferable to `async`/`concurrent` are pretty damn rare. (Of course, if someone has a counterexample where they think `operate` is clearly preferable for something else too, do bring that up! A counterexample might make me rethink my conclusions here). Having pondered this a little, my current feeling is that the `operate` API sadly doesn't really fit. It feels like it's trying to bolt on to `Io` a completely different design for asynchronous operations, and tries to be useful for several things at once, therefore ending up kind of mediocre at them all. I believe that a small, focused `poll` API (vaguely akin to the old one) would probably be a neater and simpler addition to the interface, which handles the key use case of child processes just as well; reflects the primitives actually available on OSes so that implementations don't have to do anything too complicated; and avoids muddying the waters wrt how you're "supposed" to do certain stuff asynchronously.

blblack commented 2026年01月08日 18:46:54 +01:00

I'm likely being naive as someone who's never used the current poller directly, but what benefit does this bring that wouldn't be better solved by making any higher level "poller" interface just use io.concurrent, and then working down from there to support other use-cases: to meet a single-threaded requirement, caller can use Io.Evented, and for platforms that don't have a working Io.Evented, implement possibly-somewhat-limited Io.Evented backends for Windows and posixy poll(2) that suffice at least for this use-case, etc?

I'm likely being naive as someone who's never used the current poller directly, but what benefit does this bring that wouldn't be better solved by making any higher level "poller" interface just use `io.concurrent`, and then working down from there to support other use-cases: to meet a single-threaded requirement, caller can use `Io.Evented`, and for platforms that don't have a working `Io.Evented`, implement possibly-somewhat-limited `Io.Evented` backends for Windows and posixy `poll(2)` that suffice at least for this use-case, etc?

matklad commented 2026年01月08日 22:09:03 +01:00

counterexample

I think there might be two classes of objections here: performance and programming model.

For performance, I don't know an example, but that has to be something with hight natural concurrent
"width" with relatively few syscalls per unit of concurrency, so perhaps concurrent du?

For programming model, my central example would be compaction code (compaction.zig). Compaction is similar to merge,
where you turn sorted A and sorted B into sorted combined union C, except that all of A, B, and C
are on disk, and chunked. So the code here is a loop that reads chunk A, chunk B, merges them
in-memory, and writes new chunk C, with complications:

• Elements from A and B might also cancel out, so in general it's hard to predict what happens
  first: running out of chunk A, chunk B, or in-progress chunk C.
• To hide IO latency, we want to read ahead several chunks of A and B. Similarly, several chunks
  from C might be in flight, if in-memory merge is faster than chunk write latency.
• But we don't want to read too far ahead, so there's a shared budget of in-memory chunks we can use
  for IO.
• But we need to be smart with prioritizing which chunk we read next, depending whether A or B is
  consumed faster.
• And we actually run several independent copies of the process, but they all share budget.
• And, the kicker, the whole process is suspendable, we pause at certain well-defined safe-points
  and resume later.
• And safe points are defined dynmaically based on the progress so far.

This adds up to a ridiculous 2k lines choreographic dance around a 10-line hot loop that actually
does all the real work.

And it does seem to me that writing that in the explicit state machine style "I've read/merged/wrote
a chunk, what should I do next, given global constraints?" is the least bad approach. Filing each
individual read as a separate async and than updating the state under mutex feels horrible. Having
a central "dispatch" async function with a mailbox where completions are posted is better, but feels
like an extra abstraction overhead over raw concurrent operations.

operate won't help there though, because each individual "read_chunk" operation isn't a single
read syscall, but rather a readAll loop.

Which I guess is another issue with operate --- not only its a separate way to express
concurrency, it also is not composable, and works only for operations that IO deems "primitive". Hm,
I wonder if we can actually flip this around? Instead of operations beeing structs, we could
submit arbitrary functions, and then IO is allowed to optimize certain operations? E.g.,
single_threaded could document that, if all ops are reads and writes, it goes via epoll without
spending concurrency budget. Not a sericous suggestion :)

> counterexample I think there might be two classes of objections here: performance and programming model. For performance, I don't know an example, but that has to be something with hight natural concurrent "width" with relatively few syscalls per unit of concurrency, so perhaps concurrent `du`? For programming model, my central example would be compaction code ([compaction.zig](https://github.com/tigerbeetle/tigerbeetle/blob/46941baa27667ec964e302954d8f2eb342b03cdc/src/lsm/compaction.zig)). Compaction is similar to merge, where you turn sorted A and sorted B into sorted combined union C, except that all of A, B, and C are on disk, and chunked. So the code here is a loop that reads chunk A, chunk B, merges them in-memory, and writes new chunk C, with complications: * Elements from A and B might also cancel out, so in general it's hard to predict what happens first: running out of chunk A, chunk B, or in-progress chunk C. * To hide IO latency, we want to read ahead several chunks of A and B. Similarly, several chunks from C might be in flight, if in-memory merge is faster than chunk write latency. * But we don't want to read too far ahead, so there's a shared budget of in-memory chunks we can use for IO. * But we need to be smart with prioritizing which chunk we read next, depending whether A or B is consumed faster. * And we actually run several independent copies of the process, but they all share budget. * And, the kicker, the whole process is suspendable, we pause at certain well-defined safe-points and resume later. * And safe points are defined dynmaically based on the progress so far. This adds up to a ridiculous 2k lines choreographic dance around a 10-line hot loop that actually does all the real work. And it does seem to me that writing that in the explicit state machine style "I've read/merged/wrote a chunk, what should I do next, given global constraints?" is the least bad approach. Filing each individual read as a separate `async` and than updating the state under mutex feels horrible. Having a central "dispatch" async function with a mailbox where completions are posted is better, but feels like an extra abstraction overhead over raw concurrent operations. `operate` won't help there though, because each individual "read_chunk" operation isn't a single read syscall, but rather a `readAll` loop. Which I guess is another issue with `operate` --- not only its a separate way to express concurrency, it also is not composable, and works only for operations that IO deems "primitive". Hm, I wonder if we can actually flip this around? Instead of `operations` beeing structs, we could submit arbitrary functions, and then IO is allowed to optimize certain operations? E.g., single_threaded could document that, if all ops are reads and writes, it goes via epoll without spending concurrency budget. Not a sericous suggestion :)

andrewrk commented 2026年01月08日 22:19:05 +01:00

Thank you, both of you, for taking a look!

Please have a look at the commit I just pushed. I simplified the interface and implementation to eliminate the n_wait and timeout parameters. Timeout can maybe be brought back in the future but it's a separate concern.

Specific interface I am not happy with (but maybe I just misunderstand something). The issue I have with it is that it assumes readiness based IO, not completion based

This is the main thing I addressed with the new commit. The std.Io.Threaded implementation based on poll() does up to 1 poll() and then always returns. With this API definition, completion-based implementations such as IoUring can send the operations to the event loop like normal and wait on them like normal, returning after all have completed. This is a valid definition of "nonblocking = true" in the operation (i.e. it does not block other operations).

I'm not concerned about operations.len being very large. Applications suffering from this problem should be solved by using a more appropriate Io implementation. Meanwhile, if you're stuck with std.Io.Threaded (i.e. poll), this batching interface is going to be strictly better than doing these operations in a loop. Much, much better, in fact, because it will prevent I/O operations from unnecessarily blocking other ones. Also the implementation has a fixed buffer for the poll set, so it's O(1) technically ;)

@blblack - the problem statement is this:

ability to migrate code that used poll() without causing error.ConcurrencyUnavailable on -fsingle-threaded builds

In other words, if you use io.concurrent() in single-threaded builds using std.Io.Threaded, you receive error.ConcurrencyUnavailable. Therefore, if users update their Zig code from using poll() to using std.Io under these conditions, their software will now fail, when it worked perfectly fine before.

Thank you, both of you, for taking a look! Please have a look at the commit I just pushed. I simplified the interface and implementation to eliminate the `n_wait` and `timeout` parameters. Timeout can maybe be brought back in the future but it's a separate concern. > Specific interface I am not happy with (but maybe I just misunderstand something). The issue I have with it is that it assumes readiness based IO, not completion based This is the main thing I addressed with the new commit. The `std.Io.Threaded` implementation based on poll() does up to 1 poll() and then always returns. With this API definition, completion-based implementations such as IoUring can send the operations to the event loop like normal and wait on them like normal, returning after all have completed. This is a valid definition of "nonblocking = true" in the operation (i.e. it does not block other operations). I'm not concerned about operations.len being very large. Applications suffering from this problem should be solved by using a more appropriate Io implementation. Meanwhile, if you're stuck with `std.Io.Threaded` (i.e. `poll`), this batching interface is going to be strictly better than doing these operations in a loop. Much, much better, in fact, because it will prevent I/O operations from unnecessarily blocking other ones. Also the implementation has a fixed buffer for the poll set, so it's O(1) technically ;) @blblack - the problem statement is this: > ability to migrate code that used poll() without causing error.ConcurrencyUnavailable on -fsingle-threaded builds In other words, if you use `io.concurrent()` in single-threaded builds using `std.Io.Threaded`, you receive `error.ConcurrencyUnavailable`. Therefore, if users update their Zig code from using poll() to using `std.Io` under these conditions, their software will now fail, when it worked perfectly fine before.

andrewrk commented 2026年01月09日 01:41:48 +01:00

I'd argue that its implementation is just more natural as a hand-coded state machine, and not as a pair of asyncConcurrent tasks.

I have to strong disagree with you there. If you're willing to take error.ConcurrencyUnavailable into the error set, then this is what you can write:

diff --git a/lib/std/process.zig b/lib/std/process.zig
index b5fc8541d8..15815354c8 100644
--- a/lib/std/process.zig
+++ b/lib/std/process.zig
@@ -467,6 +467,7 @@ pub fn spawnPath(io: Io, dir: Io.Dir, options: SpawnOptions) SpawnError!Child {
 
 pub const RunError = posix.GetCwdError || posix.ReadError || SpawnError || posix.PollError || error{
 StreamTooLong,
+ ConcurrencyUnavailable,
 };
 
 pub const RunOptions = struct {
diff --git a/lib/std/process/Child.zig b/lib/std/process/Child.zig
index fc31014520..e64f6106fa 100644
--- a/lib/std/process/Child.zig
+++ b/lib/std/process/Child.zig
@@ -125,14 +125,15 @@ pub fn wait(child: *Child, io: Io) WaitError!Term {
 return io.vtable.childWait(io.userdata, child);
 }
 
-pub const CollectOutputError = error{StreamTooLong} || Allocator.Error || Io.File.Reader.Error;
+pub const CollectOutputError = error{
+ StreamTooLong,
+ ConcurrencyUnavailable,
+} || Allocator.Error || Io.File.Reader.Error;
 
 pub const CollectOutputOptions = struct {
 stdout: *std.ArrayList(u8),
 stderr: *std.ArrayList(u8),
- /// Used for `stdout` and `stderr`. If not provided, only the existing
- /// capacity will be used.
- allocator: ?Allocator = null,
+ allocator: Allocator,
 stdout_limit: Io.Limit = .unlimited,
 stderr_limit: Io.Limit = .unlimited,
 };
@@ -144,56 +145,24 @@ pub const CollectOutputOptions = struct {
 /// The process must have been started with stdout and stderr set to
 /// `process.SpawnOptions.StdIo.pipe`.
 pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions) CollectOutputError!void {
- const files: [2]Io.File = .{ child.stdout.?, child.stderr.? };
- const lists: [2]*std.ArrayList(u8) = .{ options.stdout, options.stderr };
- const limits: [2]Io.Limit = .{ options.stdout_limit, options.stderr_limit };
- var dones: [2]bool = .{ false, false };
- var reads: [2]Io.Operation = undefined;
- var vecs: [2][1][]u8 = undefined;
- while (true) {
- for (&reads, &lists, &files, dones, &vecs) |*read, list, file, done, *vec| {
- if (done) {
- read.* = .noop;
- continue;
- }
- if (options.allocator) |gpa| try list.ensureUnusedCapacity(gpa, 1);
- const cap = list.unusedCapacitySlice();
- if (cap.len == 0) return error.StreamTooLong;
- vec[0] = cap;
- read.* = .{ .file_read_streaming = .{
- .file = file,
- .data = vec,
- .nonblocking = true,
- .result = undefined,
- } };
- }
- var all_done = true;
- var any_canceled = false;
- var other_err: (error{StreamTooLong} || Io.File.Reader.Error)!void = {};
- io.vtable.operate(io.userdata, &reads);
- for (&reads, &lists, &limits, &dones) |*read, list, limit, *done| {
- if (done.*) continue;
- const n = read.file_read_streaming.result catch |err| switch (err) {
- error.Canceled => {
- any_canceled = true;
- continue;
- },
- error.WouldBlock => continue,
- else => |e| {
- other_err = e;
- continue;
- },
- };
- if (n == 0) {
- done.* = true;
- } else {
- all_done = false;
- }
- list.items.len += n;
- if (list.items.len > @intFromEnum(limit)) other_err = error.StreamTooLong;
- }
- if (any_canceled) return error.Canceled;
- try other_err;
- if (all_done) return;
- }
+ var stdout = try io.concurrent(collectStream, .{
+ io, options.allocator, child.stdout.?, options.stdout, options.stdout_limit,
+ });
+ defer stdout.cancel(io) catch {};
+
+ var stderr = try io.concurrent(collectStream, .{
+ io, options.allocator, child.stderr.?, options.stderr, options.stderr_limit,
+ });
+ defer stderr.cancel(io) catch {};
+
+ try stdout.await(io);
+ try stderr.await(io);
+}
+
+fn collectStream(io: Io, gpa: Allocator, file: File, list: *std.ArrayList(u8), limit: Io.Limit) CollectOutputError!void {
+ var fr = file.readerStreaming(io, &.{});
+ fr.interface.appendRemaining(gpa, list, limit) catch |err| switch (err) {
+ error.ReadFailed => return fr.err.?,
+ else => |e| return e,
+ };
 }

This is sooo much nicer. The only reason to suffer through the other way is to avoid possibility of failure via error.ConcurrencyUnavailable. Or, avoiding the tough question of, "after upgrading to Zig 0.16.0, why does my program spawn an extra thread now when it didn't before?"

> I'd argue that its implementation is just more natural as a hand-coded state machine, and not as a pair of asyncConcurrent tasks. I have to strong disagree with you there. If you're willing to take `error.ConcurrencyUnavailable` into the error set, then this is what you can write: ```diff diff --git a/lib/std/process.zig b/lib/std/process.zig index b5fc8541d8..15815354c8 100644 --- a/lib/std/process.zig +++ b/lib/std/process.zig @@ -467,6 +467,7 @@ pub fn spawnPath(io: Io, dir: Io.Dir, options: SpawnOptions) SpawnError!Child { pub const RunError = posix.GetCwdError || posix.ReadError || SpawnError || posix.PollError || error{ StreamTooLong, + ConcurrencyUnavailable, }; pub const RunOptions = struct { diff --git a/lib/std/process/Child.zig b/lib/std/process/Child.zig index fc31014520..e64f6106fa 100644 --- a/lib/std/process/Child.zig +++ b/lib/std/process/Child.zig @@ -125,14 +125,15 @@ pub fn wait(child: *Child, io: Io) WaitError!Term { return io.vtable.childWait(io.userdata, child); } -pub const CollectOutputError = error{StreamTooLong} || Allocator.Error || Io.File.Reader.Error; +pub const CollectOutputError = error{ + StreamTooLong, + ConcurrencyUnavailable, +} || Allocator.Error || Io.File.Reader.Error; pub const CollectOutputOptions = struct { stdout: *std.ArrayList(u8), stderr: *std.ArrayList(u8), - /// Used for `stdout` and `stderr`. If not provided, only the existing - /// capacity will be used. - allocator: ?Allocator = null, + allocator: Allocator, stdout_limit: Io.Limit = .unlimited, stderr_limit: Io.Limit = .unlimited, }; @@ -144,56 +145,24 @@ pub const CollectOutputOptions = struct { /// The process must have been started with stdout and stderr set to /// `process.SpawnOptions.StdIo.pipe`. pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions) CollectOutputError!void { - const files: [2]Io.File = .{ child.stdout.?, child.stderr.? }; - const lists: [2]*std.ArrayList(u8) = .{ options.stdout, options.stderr }; - const limits: [2]Io.Limit = .{ options.stdout_limit, options.stderr_limit }; - var dones: [2]bool = .{ false, false }; - var reads: [2]Io.Operation = undefined; - var vecs: [2][1][]u8 = undefined; - while (true) { - for (&reads, &lists, &files, dones, &vecs) |*read, list, file, done, *vec| { - if (done) { - read.* = .noop; - continue; - } - if (options.allocator) |gpa| try list.ensureUnusedCapacity(gpa, 1); - const cap = list.unusedCapacitySlice(); - if (cap.len == 0) return error.StreamTooLong; - vec[0] = cap; - read.* = .{ .file_read_streaming = .{ - .file = file, - .data = vec, - .nonblocking = true, - .result = undefined, - } }; - } - var all_done = true; - var any_canceled = false; - var other_err: (error{StreamTooLong} || Io.File.Reader.Error)!void = {}; - io.vtable.operate(io.userdata, &reads); - for (&reads, &lists, &limits, &dones) |*read, list, limit, *done| { - if (done.*) continue; - const n = read.file_read_streaming.result catch |err| switch (err) { - error.Canceled => { - any_canceled = true; - continue; - }, - error.WouldBlock => continue, - else => |e| { - other_err = e; - continue; - }, - }; - if (n == 0) { - done.* = true; - } else { - all_done = false; - } - list.items.len += n; - if (list.items.len > @intFromEnum(limit)) other_err = error.StreamTooLong; - } - if (any_canceled) return error.Canceled; - try other_err; - if (all_done) return; - } + var stdout = try io.concurrent(collectStream, .{ + io, options.allocator, child.stdout.?, options.stdout, options.stdout_limit, + }); + defer stdout.cancel(io) catch {}; + + var stderr = try io.concurrent(collectStream, .{ + io, options.allocator, child.stderr.?, options.stderr, options.stderr_limit, + }); + defer stderr.cancel(io) catch {}; + + try stdout.await(io); + try stderr.await(io); +} + +fn collectStream(io: Io, gpa: Allocator, file: File, list: *std.ArrayList(u8), limit: Io.Limit) CollectOutputError!void { + var fr = file.readerStreaming(io, &.{}); + fr.interface.appendRemaining(gpa, list, limit) catch |err| switch (err) { + error.ReadFailed => return fr.err.?, + else => |e| return e, + }; } ``` This is sooo much nicer. The only reason to suffer through the other way is to avoid possibility of failure via `error.ConcurrencyUnavailable`. Or, avoiding the tough question of, "after upgrading to Zig 0.16.0, why does my program spawn an extra thread now when it didn't before?"

matklad commented 2026年01月12日 12:46:35 +01:00

I have to strong disagree with you there

Thanks for calling me out on this one!

Please have a look at the commit I just pushed.

Yeah, I like the new version, don't see any footguns, it solves the read2 use-case, and it feels like it might be useful more generally.

>I have to strong disagree with you there Thanks for calling me out on this one! >Please have a look at the commit I just pushed. Yeah, I like the new version, don't see any footguns, it solves the [`read2`](https://github.com/rust-lang/cargo/blob/2e5dd3484ec047c5f825ccdc13ae188394b8708b/crates/cargo-util/src/read2.rs) use-case, and it feels like it might be useful more generally.

This pull request has changes conflicting with the target branch.