authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-30 19:10:44-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-30 19:10:44-08:00
log9f6152ae93037b7e69c14dba5691b146f670b3f8
treefa53c32a68d063a771a617ce40bfd36838240780
parenta9778e25496a148839d5de1024213ca96f39c5f8

std.Io.Threaded: rework file reading to observe nonblocking flag

- batchAwaitAsync does blocking reads with NtReadFile (no APC, no event) when the nonblocking flag is unset, but still takes advantage of APCs when nonblocking flag is set. - batchAwaitConcurrent returns error.ConcurrencyUnavailable when it encounters a file_read_streaming operation on a file in blocking mode. - fileReadStreaming avoids pointlessly checking sync cancelation status when nonblocking flag is set, uses an APC with a done flag, and waits on that value to change in NtDelayExecution before returning. - fix incorrect use of NtCancelIoFile (ntdll function prototype was wrong, leading to misuse)

2 files changed, 148 insertions(+), 80 deletions(-)

lib/std/Io/Threaded.zig+147-79
......@@ -1340,8 +1340,6 @@ const AlertableSyscall = struct {
13401340 }
13411341};
13421342
1343fn noopApc(_: ?*anyopaque, _: *windows.IO_STATUS_BLOCK, _: windows.ULONG) callconv(.winapi) void {}
1344
13451343fn waitForApcOrAlert() void {
13461344 const infinite_timeout: windows.LARGE_INTEGER = std.math.minInt(windows.LARGE_INTEGER);
13471345 _ = windows.ntdll.NtDelayExecution(windows.TRUE, &infinite_timeout);
......@@ -2500,7 +2498,10 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper
25002498fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Batch.AwaitAsyncError!void {
25012499 const t: *Threaded = @ptrCast(@alignCast(userdata));
25022500 if (is_windows) {
2503 try batchAwaitWindows(b);
2501 batchAwaitWindows(b, false) catch |err| switch (err) {
2502 error.ConcurrencyUnavailable => unreachable, // passed concurrency=false
2503 else => |e| return e,
2504 };
25042505 const alertable_syscall = try AlertableSyscall.start();
25052506 while (b.pending.head != .none and b.completions.head == .none) waitForApcOrAlert();
25062507 alertable_syscall.finish();
......@@ -2616,7 +2617,7 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout
26162617 },
26172618 error.UnsupportedClock => |e| return e,
26182619 };
2619 try batchAwaitWindows(b);
2620 try batchAwaitWindows(b, true);
26202621 while (b.pending.head != .none and b.completions.head == .none) {
26212622 var delay_interval: windows.LARGE_INTEGER = interval: {
26222623 const d = deadline orelse break :interval std.math.minInt(windows.LARGE_INTEGER);
......@@ -2810,7 +2811,8 @@ fn batchCancel(userdata: ?*anyopaque, b: *Io.Batch) void {
28102811 while (index != .none) {
28112812 const pending = &b.storage[index.toIndex()].pending;
28122813 const context: *WindowsBatchPendingOperationContext = .fromErased(&pending.context);
2813 _ = windows.ntdll.NtCancelIoFile(context.file, &context.iosb);
2814 var cancel_iosb: windows.IO_STATUS_BLOCK = undefined;
2815 _ = windows.ntdll.NtCancelIoFileEx(context.file, &context.iosb, &cancel_iosb);
28142816 index = pending.node.next;
28152817 }
28162818 while (b.pending.head != .none) waitForApcOrAlert();
......@@ -2860,30 +2862,94 @@ fn batchApc(apc_context: ?*anyopaque, iosb: *windows.IO_STATUS_BLOCK, _: windows
28602862 }
28612863}
28622864
2863fn batchAwaitWindows(b: *Io.Batch) Io.Cancelable!void {
2865/// If `concurrency` is false, `error.ConcurrencyUnavailable` is unreachable.
2866fn batchAwaitWindows(b: *Io.Batch, concurrency: bool) error{ Canceled, ConcurrencyUnavailable }!void {
28642867 var index = b.submissions.head;
28652868 errdefer b.submissions.head = index;
28662869 while (index != .none) {
28672870 const storage = &b.storage[index.toIndex()];
28682871 const submission = storage.submission;
2869 errdefer storage.* = .{ .submission = submission };
28702872 storage.* = .{ .pending = .{
28712873 .node = .{ .prev = b.pending.tail, .next = .none },
28722874 .tag = submission.operation,
28732875 .context = undefined,
28742876 } };
2875 const context: *WindowsBatchPendingOperationContext = .fromErased(&storage.pending.context);
2876 switch (submission.operation) {
2877 .file_read_streaming => |o| {
2878 context.file = o.file.handle;
2879 try ntReadFile(o.file.handle, o.data, &batchApc, b, &context.iosb);
2880 },
2881 }
28822877 switch (b.pending.tail) {
28832878 .none => b.pending.head = index,
28842879 else => |tail_index| b.storage[tail_index.toIndex()].pending.node.next = index,
28852880 }
28862881 b.pending.tail = index;
2882 const context: *WindowsBatchPendingOperationContext = .fromErased(&storage.pending.context);
2883 errdefer {
2884 context.iosb.u.Status = .CANCELLED;
2885 batchApc(b, &context.iosb, 0);
2886 }
2887 switch (submission.operation) {
2888 .file_read_streaming => |o| o: {
2889 var data_index: usize = 0;
2890 while (o.data.len - data_index != 0 and o.data[data_index].len == 0) data_index += 1;
2891 if (o.data.len - data_index == 0) {
2892 context.iosb = .{
2893 .u = .{ .Status = .SUCCESS },
2894 .Information = 0,
2895 };
2896 batchApc(b, &context.iosb, 0);
2897 break :o;
2898 }
2899 const buffer = o.data[data_index];
2900 const short_buffer_len = @min(std.math.maxInt(u32), buffer.len);
2901
2902 if (o.file.flags.nonblocking) {
2903 context.file = o.file.handle;
2904 switch (windows.ntdll.NtReadFile(
2905 o.file.handle,
2906 null, // event
2907 &batchApc,
2908 b,
2909 &context.iosb,
2910 buffer.ptr,
2911 short_buffer_len,
2912 null, // byte offset
2913 null, // key
2914 )) {
2915 .PENDING, .SUCCESS => {},
2916 .CANCELLED => unreachable,
2917 else => |status| {
2918 context.iosb.u.Status = status;
2919 batchApc(b, &context.iosb, 0);
2920 },
2921 }
2922 } else {
2923 if (concurrency) return error.ConcurrencyUnavailable;
2924
2925 const syscall: Syscall = try .start();
2926 while (true) switch (windows.ntdll.NtReadFile(
2927 o.file.handle,
2928 null, // event
2929 null, // APC routine
2930 null, // APC context
2931 &context.iosb,
2932 buffer.ptr,
2933 short_buffer_len,
2934 null, // byte offset
2935 null, // key
2936 )) {
2937 .PENDING => unreachable, // unrecoverable: wrong File nonblocking flag
2938 .CANCELLED => {
2939 try syscall.checkCancel();
2940 continue;
2941 },
2942 else => |status| {
2943 syscall.finish();
2944
2945 context.iosb.u.Status = status;
2946 batchApc(b, &context.iosb, 0);
2947 break;
2948 },
2949 };
2950 }
2951 },
2952 }
28872953 index = submission.node.next;
28882954 }
28892955 b.submissions = .{ .head = .none, .tail = .none };
......@@ -8846,28 +8912,76 @@ fn fileReadStreamingPosix(file: File, data: []const []u8) File.ReadStreamingErro
88468912}
88478913
88488914fn fileReadStreamingWindows(file: File, data: []const []u8) File.ReadStreamingError!usize {
8849 var io_status_block: windows.IO_STATUS_BLOCK = .{
8850 .u = .{ .Status = .PENDING },
8851 .Information = undefined,
8852 };
8853 try ntReadFile(file.handle, data, &noopApc, null, &io_status_block);
8854
8855 while (@atomicLoad(windows.NTSTATUS, &io_status_block.u.Status, .acquire) == .PENDING) {
8856 // Once we get here we must not return from the function until the
8857 // operation completes, thereby releasing reference to io_status_block.
8858 const alertable_syscall = AlertableSyscall.start() catch |err| switch (err) {
8859 error.Canceled => |e| {
8860 _ = windows.ntdll.NtCancelIoFile(file.handle, &io_status_block);
8861 while (@atomicLoad(windows.NTSTATUS, &io_status_block.u.Status, .acquire) == .PENDING) {
8862 waitForApcOrAlert();
8863 }
8864 return e;
8915 var index: usize = 0;
8916 while (data.len - index != 0 and data[index].len == 0) index += 1;
8917 if (data.len - index == 0) return 0;
8918 const buffer = data[index];
8919 const short_buffer_len = @min(std.math.maxInt(u32), buffer.len);
8920
8921 var iosb: windows.IO_STATUS_BLOCK = undefined;
8922
8923 if (!file.flags.nonblocking) {
8924 const syscall: Syscall = try .start();
8925 while (true) switch (windows.ntdll.NtReadFile(
8926 file.handle,
8927 null, // event
8928 null, // APC routine
8929 null, // APC context
8930 &iosb,
8931 buffer.ptr,
8932 short_buffer_len,
8933 null, // byte offset
8934 null, // key
8935 )) {
8936 .PENDING => unreachable, // unrecoverable: wrong File nonblocking flag
8937 .CANCELLED => {
8938 try syscall.checkCancel();
8939 continue;
8940 },
8941 else => |status| {
8942 syscall.finish();
8943 iosb.u.Status = status;
8944 return ntReadFileResult(&iosb);
88658945 },
88668946 };
8867 waitForApcOrAlert();
8868 alertable_syscall.finish();
88698947 }
8870 return ntReadFileResult(&io_status_block);
8948
8949 var done: bool = false;
8950
8951 switch (windows.ntdll.NtReadFile(
8952 file.handle,
8953 null, // event
8954 flagApc,
8955 &done, // APC context
8956 &iosb,
8957 buffer.ptr,
8958 short_buffer_len,
8959 null, // byte offset
8960 null, // key
8961 )) {
8962 // We must wait for the APC routine.
8963 .PENDING, .SUCCESS => while (!done) {
8964 // Once we get here we must not return from the function until the
8965 // operation completes, thereby releasing reference to io_status_block.
8966 const alertable_syscall = AlertableSyscall.start() catch |err| switch (err) {
8967 error.Canceled => |e| {
8968 var cancel_iosb: windows.IO_STATUS_BLOCK = undefined;
8969 _ = windows.ntdll.NtCancelIoFileEx(file.handle, &iosb, &cancel_iosb);
8970 while (!done) waitForApcOrAlert();
8971 return e;
8972 },
8973 };
8974 waitForApcOrAlert();
8975 alertable_syscall.finish();
8976 },
8977 else => |status| iosb.u.Status = status,
8978 }
8979 return ntReadFileResult(&iosb);
8980}
8981
8982fn flagApc(userdata: ?*anyopaque, _: *windows.IO_STATUS_BLOCK, _: windows.ULONG) callconv(.winapi) void {
8983 const flag: *bool = @ptrCast(userdata);
8984 flag.* = true;
88718985}
88728986
88738987fn ntReadFileResult(io_status_block: *const windows.IO_STATUS_BLOCK) !usize {
......@@ -8883,52 +8997,6 @@ fn ntReadFileResult(io_status_block: *const windows.IO_STATUS_BLOCK) !usize {
88838997 }
88848998}
88858999
8886fn ntReadFile(
8887 handle: windows.HANDLE,
8888 data: []const []u8,
8889 apcRoutine: ?*const windows.IO_APC_ROUTINE,
8890 apc_context: ?*anyopaque,
8891 iosb: *windows.IO_STATUS_BLOCK,
8892) Io.Cancelable!void {
8893 var index: usize = 0;
8894 while (index < data.len and data[index].len == 0) index += 1;
8895 if (index == data.len) {
8896 iosb.* = .{ .u = .{ .Status = .SUCCESS }, .Information = 0 };
8897 if (apcRoutine) |routine| if (routine != &noopApc) {
8898 _ = windows.ntdll.NtQueueApcThread(windows.current_process, routine, apc_context, iosb, null);
8899 };
8900 return;
8901 }
8902 const buffer = data[index];
8903
8904 const syscall: Syscall = try .start();
8905 while (true) switch (windows.ntdll.NtReadFile(
8906 handle,
8907 null, // event
8908 apcRoutine,
8909 apc_context,
8910 iosb,
8911 buffer.ptr,
8912 @min(std.math.maxInt(u32), buffer.len),
8913 null, // byte offset
8914 null, // key
8915 )) {
8916 .PENDING => {
8917 syscall.finish();
8918 return;
8919 },
8920 .CANCELLED => {
8921 try syscall.checkCancel();
8922 continue;
8923 },
8924 else => |status| {
8925 syscall.finish();
8926 iosb.u.Status = status;
8927 return;
8928 },
8929 };
8930}
8931
89329000fn fileReadPositionalPosix(file: File, data: []const []u8, offset: u64) File.ReadPositionalError!usize {
89339001 if (!have_preadv) @compileError("TODO implement fileReadPositionalPosix for cursed operating systems that don't support preadv (it's only Haiku)");
89349002
lib/std/os/windows/ntdll.zig+1-1
......@@ -614,5 +614,5 @@ pub extern "ntdll" fn NtCancelIoFileEx(
614614
615615pub extern "ntdll" fn NtCancelIoFile(
616616 FileHandle: HANDLE,
617 IoRequestToCancel: ?*IO_STATUS_BLOCK,
617 IoStatusBlock: *IO_STATUS_BLOCK,
618618) callconv(.winapi) NTSTATUS;