Zig Version
0.16.0-dev.2987+1311880ee
Steps to Reproduce and Observed Behavior
Below are two examples of where Io.Batch.awaitConcurrent does not work properly using Io.Threaded.
This first example combines data from two streams and writes it to stdout:
conststd=@import("std");pubfnmain(init:std.process.Init)!void{constio=init.io;constargs=tryinit.minimal.args.toSlice(init.arena.allocator());if(args.len<3)returnerror.BadArgs;conststdout:std.Io.File=.stdout();constcwd:std.Io.Dir=.cwd();varstorage:[2]std.Io.Operation.Storage=undefined;varbatch:std.Io.Batch=.init(&storage);varstreams:[2]struct{file:std.Io.File,buf:[4096]u8,vec:[1][]u8,}=undefined;for(0..,args[1..][0..2],&streams)|i,path,*stream|{errdeferfor(streams[0..i])|s|s.file.close(io);stream.*=.{.file=trycwd.openFile(io,path,.{}),.buf=undefined,.vec=.{&stream.buf},};batch.addAt(@intCast(i),.{.file_read_streaming=.{.file=stream.file,.data=&stream.vec,}});}defer{for(streams)|s|s.file.close(io);batch.cancel(io);}varremaining:usize=streams.len;while(remaining!=0){trybatch.awaitConcurrent(io,.none);while(batch.next())|completion|{constn=completion.result.file_read_streamingcatch|e|switch(e){error.EndOfStream=>{remaining-=1;continue;},else=>returne,};consts=streams[completion.index];stdout.writeStreamingAll(io,s.buf[0..n])catch|e|switch(e){error.BrokenPipe=>return,else=>returne,};batch.addAt(completion.index,.{.file_read_streaming=.{.file=s.file,.data=&s.vec,}});}}}Build the executable and create two pipes:
zig build-exe stream-combine.zig
mkfifo a b
When the below is run it should output 1 and 2 at about the same time, however it (usually) waits until the first pipe is closed to output 2 for reasons explained in the comments.
./stream-combine a b &
cat a &
# Open the write-end of both pipes so the program can open them and
# begin polling.
sleep 4 >> a &
sleep 1 >> b &
sleep 0.5 # Give the program a moment to open and start polling them
# Write data to both pipes
echo 1 >> a # cat will almost always read this before the program,
# however the program will still see POLLIN and block on a read from
# the pipe until it is closed and EndOfStream is indicated.
echo 2 >> b
wait
The following code should finish (on a unix-based system):
This next example creates a child process which performs a blocking write to stdout and then reads from stdin. The main process concurrently reads the child's stdout and writes to stdin.
When run, the program gets deadlocked by both processes blocking on writes since pipes do not support short writes; POLLOUT only indicates one byte of data can written without blocking, and also has the same race-condition demonstrated above.
conststd=@import("std");pubfnmain(init:std.process.Init)!void{constargs=tryinit.minimal.args.toSlice(init.arena.allocator());if(args.len<1)returnerror.BadArgs;constblocking_write_size:usize=65537;// (for an empty pipe on x86_64 linux;// can be less if there is data already buffered)constblocking_write_data:[blocking_write_size]u8=@splat(0);if(args.len==1){// main processvarchild=trystd.process.spawn(init.io,.{.argv=&.{args[0],"options"},.stdin=.pipe,.stdout=.pipe,});errdeferchild.kill(init.io);conststdin=child.stdin.?;conststdout=child.stdout.?;varsend_vec:[1][]constu8=.{&blocking_write_data};varrecv_buf:[4096]u8=undefined;constrecv_vec:[1][]u8=.{&recv_buf};varrecieved:usize=0;varstorage:[2]std.Io.Operation.Storage=undefined;varbatch:std.Io.Batch=.init(&storage);deferbatch.cancel(init.io);batch.addAt(0,.{.file_write_streaming=.{.file=stdin,.data=&send_vec}});batch.addAt(1,.{.file_read_streaming=.{.file=stdout,.data=&recv_vec}});while(send_vec[0].len!=0orrecieved!=blocking_write_size){trybatch.awaitConcurrent(init.io,.none);while(batch.next())|completion|switch(completion.index){0=>{send_vec[0]=send_vec[0][trycompletion.result.file_write_streaming..];if(send_vec[0].len==0)continue;batch.addAt(0,.{.file_write_streaming=.{.file=stdin,.data=&send_vec}});},1=>{recieved+=trycompletion.result.file_read_streaming;if(recieved==blocking_write_size)continue;batch.addAt(1,.{.file_read_streaming=.{.file=stdout,.data=&recv_vec}});},else=>unreachable,};}stdin.close(init.io);child.stdin=null;_=trychild.wait(init.io);}else{// sub processconststdin=std.Io.File.stdin();conststdout=std.Io.File.stdout();trystdout.writeStreamingAll(init.io,&blocking_write_data);varstdin_buf:[4096]u8=undefined;varstdin_r=stdin.readerStreaming(init.io,&stdin_buf);_=stdin_r.interface.discardRemaining()catch|e|switch(e){error.ReadFailed=>returnstdin_r.err.?,};}}### Zig Version
0.16.0-dev.2987+1311880ee
### Steps to Reproduce and Observed Behavior
Below are two examples of where `Io.Batch.awaitConcurrent` does not work properly using `Io.Threaded`.
---
This first example combines data from two streams and writes it to stdout:
```zig
const std = @import("std");
pub fn main(init: std.process.Init) !void {
const io = init.io;
const args = try init.minimal.args.toSlice(init.arena.allocator());
if (args.len < 3) return error.BadArgs;
const stdout: std.Io.File = .stdout();
const cwd: std.Io.Dir = .cwd();
var storage: [2]std.Io.Operation.Storage = undefined;
var batch: std.Io.Batch = .init(&storage);
var streams: [2]struct {
file: std.Io.File,
buf: [4096]u8,
vec: [1][]u8,
} = undefined;
for (0.., args[1..][0..2], &streams) |i, path, *stream| {
errdefer for (streams[0..i]) |s| s.file.close(io);
stream.* = .{
.file = try cwd.openFile(io, path, .{}),
.buf = undefined,
.vec = .{&stream.buf},
};
batch.addAt(@intCast(i), .{ .file_read_streaming = .{
.file = stream.file,
.data = &stream.vec,
} });
}
defer {
for (streams) |s| s.file.close(io);
batch.cancel(io);
}
var remaining: usize = streams.len;
while (remaining != 0) {
try batch.awaitConcurrent(io, .none);
while (batch.next()) |completion| {
const n = completion.result.file_read_streaming catch |e| switch (e) {
error.EndOfStream => {
remaining -= 1;
continue;
},
else => return e,
};
const s = streams[completion.index];
stdout.writeStreamingAll(io, s.buf[0..n]) catch |e| switch (e) {
error.BrokenPipe => return,
else => return e,
};
batch.addAt(completion.index, .{ .file_read_streaming = .{
.file = s.file,
.data = &s.vec,
} });
}
}
}
```
Build the executable and create two pipes:
```sh
zig build-exe stream-combine.zig
mkfifo a b
```
When the below is run it should output 1 and 2 at about the same time, however it (usually) waits until the first pipe is closed to output 2 for reasons explained in the comments.
```sh
./stream-combine a b &
cat a &
# Open the write-end of both pipes so the program can open them and
# begin polling.
sleep 4 >> a &
sleep 1 >> b &
sleep 0.5 # Give the program a moment to open and start polling them
# Write data to both pipes
echo 1 >> a # cat will almost always read this before the program,
# however the program will still see POLLIN and block on a read from
# the pipe until it is closed and EndOfStream is indicated.
echo 2 >> b
wait
```
---
The following code should finish (on a unix-based system):
This next example creates a child process which performs a blocking write to stdout and then reads from stdin. The main process concurrently reads the child's stdout and writes to stdin.
When run, the program gets deadlocked by both processes blocking on writes since pipes do not support short writes; POLLOUT only indicates one byte of data can written without blocking, and also has the same race-condition demonstrated above.
```zig
const std = @import("std");
pub fn main(init: std.process.Init) !void {
const args = try init.minimal.args.toSlice(init.arena.allocator());
if (args.len < 1) return error.BadArgs;
const blocking_write_size: usize = 65537; // (for an empty pipe on x86_64 linux;
// can be less if there is data already buffered)
const blocking_write_data: [blocking_write_size]u8 = @splat(0);
if (args.len == 1) {
// main process
var child = try std.process.spawn(init.io, .{
.argv = &.{ args[0], "options" },
.stdin = .pipe,
.stdout = .pipe,
});
errdefer child.kill(init.io);
const stdin = child.stdin.?;
const stdout = child.stdout.?;
var send_vec: [1][]const u8 = .{&blocking_write_data};
var recv_buf: [4096]u8 = undefined;
const recv_vec: [1][]u8 = .{&recv_buf};
var recieved: usize = 0;
var storage: [2]std.Io.Operation.Storage = undefined;
var batch: std.Io.Batch = .init(&storage);
defer batch.cancel(init.io);
batch.addAt(0, .{ .file_write_streaming = .{ .file = stdin, .data = &send_vec } });
batch.addAt(1, .{ .file_read_streaming = .{ .file = stdout, .data = &recv_vec } });
while (send_vec[0].len != 0 or recieved != blocking_write_size) {
try batch.awaitConcurrent(init.io, .none);
while (batch.next()) |completion| switch (completion.index) {
0 => {
send_vec[0] = send_vec[0][try completion.result.file_write_streaming..];
if (send_vec[0].len == 0) continue;
batch.addAt(0, .{ .file_write_streaming = .{ .file = stdin, .data = &send_vec } });
},
1 => {
recieved += try completion.result.file_read_streaming;
if (recieved == blocking_write_size) continue;
batch.addAt(1, .{ .file_read_streaming = .{ .file = stdout, .data = &recv_vec } });
},
else => unreachable,
};
}
stdin.close(init.io);
child.stdin = null;
_ = try child.wait(init.io);
} else {
// sub process
const stdin = std.Io.File.stdin();
const stdout = std.Io.File.stdout();
try stdout.writeStreamingAll(init.io, &blocking_write_data);
var stdin_buf: [4096]u8 = undefined;
var stdin_r = stdin.readerStreaming(init.io, &stdin_buf);
_ = stdin_r.interface.discardRemaining() catch |e| switch (e) {
error.ReadFailed => return stdin_r.err.?,
};
}
}
```