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