authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-31 14:58:02-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-02-01 02:49:36+01:00
log3abc96a601d2349cc1743774f0cebb2eb0ea0c61
tree63312874e31a82f649d116c5cfcb0fc686fb0df6
parente60ba21114bc4514acc535be7106ce6c16e9bf04

std.Io: add test for batchAwaitAsync

and make it always work for all targets including WASI This function guarantees no additional failure modes introduced.

3 files changed, 171 insertions(+), 93 deletions(-)

lib/std/Io.zig+12-2
...@@ -369,7 +369,9 @@ pub const Batch = struct {...@@ -369,7 +369,9 @@ pub const Batch = struct {
369 context: ?*anyopaque,369 context: ?*anyopaque,
370370
371 /// After calling this, it is safe to unconditionally defer a call to371 /// After calling this, it is safe to unconditionally defer a call to
372 /// `cancel`.372 /// `cancel`. `storage` is a pre-allocated buffer of undefined memory that
373 /// determines the maximum number of active operations that can be
374 /// submitted via `add` and `addAt`.
373 pub fn init(storage: []Operation.Storage) Batch {375 pub fn init(storage: []Operation.Storage) Batch {
374 var prev: Operation.OptionalIndex = .none;376 var prev: Operation.OptionalIndex = .none;
375 for (storage, 0..) |*operation, index| {377 for (storage, 0..) |*operation, index| {
...@@ -422,12 +424,20 @@ pub const Batch = struct {...@@ -422,12 +424,20 @@ pub const Batch = struct {
422 b.submissions.tail = .fromIndex(index);424 b.submissions.tail = .fromIndex(index);
423 }425 }
424426
427 pub const Completion = struct {
428 /// The element within the provided operation storage that completed.
429 /// `addAt` can be used to re-arm the `Batch` using this `index`.
430 index: u32,
431 /// The return value of the operation.
432 result: Operation.Result,
433 };
434
425 /// After calling `awaitAsync`, `awaitConcurrent`, or `cancel`, this435 /// After calling `awaitAsync`, `awaitConcurrent`, or `cancel`, this
426 /// function iterates over the completed operations.436 /// function iterates over the completed operations.
427 ///437 ///
428 /// Each completion returned from this function dequeues from the `Batch`.438 /// Each completion returned from this function dequeues from the `Batch`.
429 /// It is not required to dequeue all completions before awaiting again.439 /// It is not required to dequeue all completions before awaiting again.
430 pub fn next(b: *Batch) ?struct { index: u32, result: Operation.Result } {440 pub fn next(b: *Batch) ?Completion {
431 const index = b.completions.head;441 const index = b.completions.head;
432 if (index == .none) return null;442 if (index == .none) return null;
433 const storage = &b.storage[index.toIndex()];443 const storage = &b.storage[index.toIndex()];
lib/std/Io/Threaded.zig+99-91
...@@ -1938,6 +1938,10 @@ const have_mmap = switch (native_os) {...@@ -1938,6 +1938,10 @@ const have_mmap = switch (native_os) {
1938 .wasi, .windows => false,1938 .wasi, .windows => false,
1939 else => true,1939 else => true,
1940};1940};
1941const have_poll = switch (native_os) {
1942 .wasi, .windows => false,
1943 else => true,
1944};
19411945
1942const open_sym = if (posix.lfs64_abi) posix.system.open64 else posix.system.open;1946const open_sym = if (posix.lfs64_abi) posix.system.open64 else posix.system.open;
1943const openat_sym = if (posix.lfs64_abi) posix.system.openat64 else posix.system.openat;1947const openat_sym = if (posix.lfs64_abi) posix.system.openat64 else posix.system.openat;
...@@ -2507,104 +2511,104 @@ fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void {...@@ -2507,104 +2511,104 @@ fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void {
2507 alertable_syscall.finish();2511 alertable_syscall.finish();
2508 return;2512 return;
2509 }2513 }
2510 if (native_os == .wasi and !builtin.link_libc) @panic("TODO");2514 if (have_poll) {
2511 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;2515 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
2512 var poll_len: u32 = 0;2516 var poll_len: u32 = 0;
2513 {2517 {
2514 var index = b.submissions.head;2518 var index = b.submissions.head;
2515 while (index != .none and poll_len < poll_buffer_len) {2519 while (index != .none and poll_len < poll_buffer_len) {
2516 const submission = &b.storage[index.toIndex()].submission;2520 const submission = &b.storage[index.toIndex()].submission;
2517 switch (submission.operation) {2521 switch (submission.operation) {
2518 .file_read_streaming => |o| {2522 .file_read_streaming => |o| {
2519 poll_buffer[poll_len] = .{ .fd = o.file.handle, .events = posix.POLL.IN, .revents = 0 };2523 poll_buffer[poll_len] = .{ .fd = o.file.handle, .events = posix.POLL.IN, .revents = 0 };
2520 poll_len += 1;2524 poll_len += 1;
2521 },2525 },
2526 }
2527 index = submission.node.next;
2522 }2528 }
2523 index = submission.node.next;
2524 }2529 }
2525 }2530 switch (poll_len) {
2526 switch (poll_len) {2531 0 => return,
2527 0 => return,2532 1 => {},
2528 1 => {},2533 else => while (true) {
2529 else => while (true) {2534 const timeout_ms: i32 = t: {
2530 const timeout_ms: i32 = t: {2535 if (b.completions.head != .none) {
2531 if (b.completions.head != .none) {2536 // It is legal to call batchWait with already completed
2532 // It is legal to call batchWait with already completed2537 // operations in the ring. In such case, we need to avoid
2533 // operations in the ring. In such case, we need to avoid2538 // blocking in the poll syscall, but we can still take this
2534 // blocking in the poll syscall, but we can still take this2539 // opportunity to find additional ready operations.
2535 // opportunity to find additional ready operations.2540 break :t 0;
2536 break :t 0;
2537 }
2538 const max_poll_ms = std.math.maxInt(i32);
2539 break :t max_poll_ms;
2540 };
2541 const syscall = try Syscall.start();
2542 const rc = posix.system.poll(&poll_buffer, poll_len, timeout_ms);
2543 syscall.finish();
2544 switch (posix.errno(rc)) {
2545 .SUCCESS => {
2546 if (rc == 0) {
2547 if (b.completions.head != .none) {
2548 // Since there are already completions available in the
2549 // queue, this is neither a timeout nor a case for
2550 // retrying.
2551 return;
2552 }
2553 continue;
2554 }2541 }
2555 var prev_index: Io.Operation.OptionalIndex = .none;2542 const max_poll_ms = std.math.maxInt(i32);
2556 var index = b.submissions.head;2543 break :t max_poll_ms;
2557 for (poll_buffer[0..poll_len]) |poll_entry| {2544 };
2558 const storage = &b.storage[index.toIndex()];2545 const syscall = try Syscall.start();
2559 const submission = &storage.submission;2546 const rc = posix.system.poll(&poll_buffer, poll_len, timeout_ms);
2560 const next_index = submission.node.next;2547 syscall.finish();
2561 if (poll_entry.revents != 0) {2548 switch (posix.errno(rc)) {
2562 const result = try operate(t, submission.operation);2549 .SUCCESS => {
25632550 if (rc == 0) {
2564 switch (prev_index) {2551 if (b.completions.head != .none) {
2565 .none => b.submissions.head = next_index,2552 // Since there are already completions available in the
2566 else => b.storage[prev_index.toIndex()].submission.node.next = next_index,2553 // queue, this is neither a timeout nor a case for
2567 }2554 // retrying.
2568 if (next_index == .none) b.submissions.tail = prev_index;2555 return;
2569
2570 switch (b.completions.tail) {
2571 .none => b.completions.head = index,
2572 else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
2573 }2556 }
2574 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };2557 continue;
2575 b.completions.tail = index;2558 }
2576 } else prev_index = index;2559 var prev_index: Io.Operation.OptionalIndex = .none;
2577 index = next_index;2560 var index = b.submissions.head;
2578 }2561 for (poll_buffer[0..poll_len]) |poll_entry| {
2579 assert(index == .none);2562 const storage = &b.storage[index.toIndex()];
2580 return;2563 const submission = &storage.submission;
2581 },2564 const next_index = submission.node.next;
2582 .INTR => continue,2565 if (poll_entry.revents != 0) {
2583 else => break,2566 const result = try operate(t, submission.operation);
2584 }2567
2585 },2568 switch (prev_index) {
2569 .none => b.submissions.head = next_index,
2570 else => b.storage[prev_index.toIndex()].submission.node.next = next_index,
2571 }
2572 if (next_index == .none) b.submissions.tail = prev_index;
2573
2574 switch (b.completions.tail) {
2575 .none => b.completions.head = index,
2576 else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
2577 }
2578 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
2579 b.completions.tail = index;
2580 } else prev_index = index;
2581 index = next_index;
2582 }
2583 assert(index == .none);
2584 return;
2585 },
2586 .INTR => continue,
2587 else => break,
2588 }
2589 },
2590 }
2586 }2591 }
2587 {
2588 var tail_index = b.completions.tail;
2589 defer b.completions.tail = tail_index;
2590 var index = b.submissions.head;
2591 errdefer b.submissions.head = index;
2592 while (index != .none) {
2593 const storage = &b.storage[index.toIndex()];
2594 const submission = &storage.submission;
2595 const next_index = submission.node.next;
2596 const result = try operate(t, submission.operation);
25972592
2598 switch (tail_index) {2593 var tail_index = b.completions.tail;
2599 .none => b.completions.head = index,2594 defer b.completions.tail = tail_index;
2600 else => b.storage[tail_index.toIndex()].completion.node.next = index,2595 var index = b.submissions.head;
2601 }2596 errdefer b.submissions.head = index;
2602 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };2597 while (index != .none) {
2603 tail_index = index;2598 const storage = &b.storage[index.toIndex()];
2604 index = next_index;2599 const submission = &storage.submission;
2600 const next_index = submission.node.next;
2601 const result = try operate(t, submission.operation);
2602
2603 switch (tail_index) {
2604 .none => b.completions.head = index,
2605 else => b.storage[tail_index.toIndex()].completion.node.next = index,
2605 }2606 }
2606 b.submissions = .{ .head = .none, .tail = .none };2607 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
2608 tail_index = index;
2609 index = next_index;
2607 }2610 }
2611 b.submissions = .{ .head = .none, .tail = .none };
2608}2612}
26092613
2610fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.AwaitConcurrentError!void {2614fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.AwaitConcurrentError!void {
...@@ -2644,7 +2648,11 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout...@@ -2644,7 +2648,11 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout
2644 }2648 }
2645 return;2649 return;
2646 }2650 }
2647 if (native_os == .wasi and !builtin.link_libc) @panic("TODO");2651 if (native_os == .wasi) {
2652 // TODO call poll_oneoff
2653 return error.ConcurrencyUnavailable;
2654 }
2655 if (!have_poll) return error.ConcurrencyUnavailable;
2648 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;2656 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
2649 var poll_storage: struct {2657 var poll_storage: struct {
2650 gpa: std.mem.Allocator,2658 gpa: std.mem.Allocator,
lib/std/Io/test.zig+60
...@@ -656,3 +656,63 @@ test "memory mapping" {...@@ -656,3 +656,63 @@ test "memory mapping" {
656 try expectEqualStrings("this9is9my data123\x00\x00", mm.memory[0.."this9is9my data123\x00\x00".len]);656 try expectEqualStrings("this9is9my data123\x00\x00", mm.memory[0.."this9is9my data123\x00\x00".len]);
657 }657 }
658}658}
659
660test "read from a file using Batch.awaitAsync API" {
661 const io = testing.io;
662
663 var tmp = tmpDir(.{});
664 defer tmp.cleanup();
665
666 try tmp.dir.writeFile(io, .{
667 .sub_path = "eyes.txt",
668 .data = "Heaven's been cheating the Hell out of me",
669 });
670 try tmp.dir.writeFile(io, .{
671 .sub_path = "saviour.txt",
672 .data = "Burn your thoughts, erase your will / to gods of suffering and tears",
673 });
674
675 var eyes_file = try tmp.dir.openFile(io, "eyes.txt", .{});
676 defer eyes_file.close(io);
677
678 var saviour_file = try tmp.dir.openFile(io, "saviour.txt", .{});
679 defer saviour_file.close(io);
680
681 var eyes_buf: [100]u8 = undefined;
682 var saviour_buf: [100]u8 = undefined;
683 var storage: [2]Io.Operation.Storage = undefined;
684 var batch: Io.Batch = .init(&storage);
685
686 batch.addAt(0, .{ .file_read_streaming = .{
687 .file = eyes_file,
688 .data = &.{&eyes_buf},
689 } });
690 batch.addAt(1, .{ .file_read_streaming = .{
691 .file = saviour_file,
692 .data = &.{&saviour_buf},
693 } });
694
695 // This API is supposed to *always* work even if the target has no
696 // concurrency primitives available.
697 try batch.awaitAsync(io);
698
699 while (batch.next()) |completion| {
700 switch (completion.index) {
701 0 => {
702 const n = try completion.result.file_read_streaming;
703 try expectEqualStrings(
704 "Heaven's been cheating the Hell out of me"[0..n],
705 eyes_buf[0..n],
706 );
707 },
708 1 => {
709 const n = try completion.result.file_read_streaming;
710 try expectEqualStrings(
711 "Burn your thoughts, erase your will / to gods of suffering and tears"[0..n],
712 saviour_buf[0..n],
713 );
714 },
715 else => return error.TestFailure,
716 }
717 }
718}