authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-10-21 22:01:36+02:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-10-21 22:01:36+02:00
log9ed5e259611028b49718a65c2b87ae9fc4a2dbd2
tree91ebafe86b4e6808f32f5556027d72edf415d2ea
parentae7b2ba9e3d43fc058ef2faadfceda77e2018f19
parent416ed78779a98e29ccbd8737266e56b0a0043de8

Merge pull request 'std.Io.Batch: fix operation not evicted from submitted list in netReceive fast path in awaitConcurrent' (#36191) from gBillal/zig:fix-io-batch-await-conc-fast-path into master

Reviewed-on: https://codeberg.org/ziglang/zig/pulls/36191

1 files changed, 69 insertions(+), 55 deletions(-)

lib/std/Io/Threaded.zig+69-55
...@@ -2869,69 +2869,83 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout...@@ -2869,69 +2869,83 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout
2869 } = .{ .gpa = t.allocator, .batch = b, .slice = &poll_buffer, .len = 0 };2869 } = .{ .gpa = t.allocator, .batch = b, .slice = &poll_buffer, .len = 0 };
2870 {2870 {
2871 var index = b.submitted.head;2871 var index = b.submitted.head;
2872 var prev_index: Io.Operation.OptionalIndex = .none;
2872 while (index != .none) {2873 while (index != .none) {
2873 const storage = &b.storage[index.toIndex()];2874 const storage = &b.storage[index.toIndex()];
2874 const submission = storage.submission;2875 const submission = storage.submission;
2875 switch (submission.operation) {2876 const next_index = submission.node.next;
2876 .file_read_streaming => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.ERR),2877 const result: Io.Operation.Result = eager: {
2877 .file_write_streaming => |o| try poll_storage.add(o.file.handle, posix.POLL.OUT | posix.POLL.ERR),2878 // If control flow makes it to the end of this switch, it is handled as a
2878 .device_io_control => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.OUT | posix.POLL.ERR),2879 // non-blocking operation and stays in the submission queue. The other possibility
2879 .net_receive => |*o| nb: {2880 // is breaking from the eager block, making the operation handled immediately.
2880 var data_i: usize = 0;2881 switch (submission.operation) {
2881 const result: Io.Operation.Result = .{ .net_receive = for (o.message_buffer, 0..) |*msg, msg_i| {2882 .file_read_streaming => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.ERR),
2882 const remaining_data_buffer = o.data_buffer[data_i..];2883 .file_write_streaming => |o| try poll_storage.add(o.file.handle, posix.POLL.OUT | posix.POLL.ERR),
2883 netReceivePosix(o.socket_handle, msg, remaining_data_buffer, o.flags, true) catch |err| switch (err) {2884 .device_io_control => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.OUT | posix.POLL.ERR),
2884 error.Canceled => |e| return e,2885 .net_receive => |*o| nb: {
2885 error.WouldBlock => {2886 var data_i: usize = 0;
2886 if (msg_i != 0) break .{ null, msg_i };2887 break :eager .{ .net_receive = for (o.message_buffer, 0..) |*msg, msg_i| {
2887 try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR);2888 const remaining_data_buffer = o.data_buffer[data_i..];
2888 break :nb;2889 netReceivePosix(o.socket_handle, msg, remaining_data_buffer, o.flags, true) catch |err| switch (err) {
2889 },2890 error.Canceled => |e| return e,
2890 else => |e| break .{ e, msg_i },
2891 };
2892 data_i += msg.data.len;
2893 } else .{ null, o.message_buffer.len } };
2894 switch (b.completed.tail) {
2895 .none => b.completed.head = index,
2896 else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
2897 }
2898 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
2899 b.completed.tail = index;
2900 },
2901 .net_send => |*o| nb: {
2902 const result: Io.Operation.Result = .{
2903 .net_send = o: {
2904 const send_err, const sent = netSendPosix(t, o.socket_handle, o.messages, o.flags, true);
2905 if (send_err) |err| switch (err) {
2906 error.Canceled => |e| if (sent == 0) {
2907 return e;
2908 } else {
2909 // Leave the `error.Canceled` for later, but don't try to send any more messages.
2910 recancelInner();
2911 break :o .{ null, sent };
2912 },
2913 error.WouldBlock => {2891 error.WouldBlock => {
2914 if (sent != 0) break :o .{ null, sent };2892 if (msg_i != 0) break .{ null, msg_i };
2915 try poll_storage.add(o.socket_handle, posix.POLL.OUT | posix.POLL.ERR);2893 try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR);
2916 break :nb;2894 break :nb;
2917 },2895 },
2918 else => |e| break :o .{ e, sent },2896 else => |e| break .{ e, msg_i },
2919 };2897 };
2920 break :o .{ null, sent };2898 data_i += msg.data.len;
2921 },2899 } else .{ null, o.message_buffer.len } };
2922 };2900 },
2923 switch (b.completed.tail) {2901 .net_send => |*o| nb: {
2924 .none => b.completed.head = index,2902 break :eager .{
2925 else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,2903 .net_send = o: {
2926 }2904 const send_err, const sent = netSendPosix(t, o.socket_handle, o.messages, o.flags, true);
2927 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };2905 if (send_err) |err| switch (err) {
2928 b.completed.tail = index;2906 error.Canceled => |e| if (sent == 0) {
2929 },2907 return e;
2930 .net_accept => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR),2908 } else {
2931 .net_read => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR),2909 // Leave the `error.Canceled` for later, but don't try to send any more messages.
2932 .net_write => |o| try poll_storage.add(o.socket_handle, posix.POLL.OUT | posix.POLL.ERR),2910 recancelInner();
2911 break :o .{ null, sent };
2912 },
2913 error.WouldBlock => {
2914 if (sent != 0) break :o .{ null, sent };
2915 try poll_storage.add(o.socket_handle, posix.POLL.OUT | posix.POLL.ERR);
2916 break :nb;
2917 },
2918 else => |e| break :o .{ e, sent },
2919 };
2920 break :o .{ null, sent };
2921 },
2922 };
2923 },
2924 .net_accept => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR),
2925 .net_read => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR),
2926 .net_write => |o| try poll_storage.add(o.socket_handle, posix.POLL.OUT | posix.POLL.ERR),
2927 }
2928 // Non-blocking case. Submission entry stays alive.
2929 prev_index = index;
2930 index = next_index;
2931 continue;
2932 };
2933 // The operation completed without blocking so we must evict it from the submission list
2934 // and add it to the completion list.
2935 switch (prev_index) {
2936 .none => b.submitted.head = next_index,
2937 else => b.storage[prev_index.toIndex()].submission.node.next = next_index,
2933 }2938 }
2934 index = submission.node.next;2939 if (next_index == .none) b.submitted.tail = prev_index;
2940
2941 switch (b.completed.tail) {
2942 .none => b.completed.head = index,
2943 else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
2944 }
2945 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
2946 b.completed.tail = index;
2947
2948 index = next_index;
2935 }2949 }
2936 }2950 }
2937 switch (poll_storage.len) {2951 switch (poll_storage.len) {