authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-02-02 00:05:51-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-03-08 19:20:06-07:00
logbe0f188990d8aedc3f5bea5c9ab436a7f778d59d
treef5a230e008c34fb38098485eaa3d3305a6ce29c7
parent6be202f46633d02e20d0f068a32296113ecb95ca

std.Io: move netReceive to become an Operation

Notably, the timeout becomes provided by the general-purpose Batch API rather than being special-purposed.

3 files changed, 145 insertions(+), 144 deletions(-)

lib/std/Io.zig+44-2
...@@ -243,7 +243,6 @@ pub const VTable = struct {...@@ -243,7 +243,6 @@ pub const VTable = struct {
243 netConnectUnix: *const fn (?*anyopaque, *const net.UnixAddress) net.UnixAddress.ConnectError!net.Socket.Handle,243 netConnectUnix: *const fn (?*anyopaque, *const net.UnixAddress) net.UnixAddress.ConnectError!net.Socket.Handle,
244 netSocketCreatePair: *const fn (?*anyopaque, net.Socket.CreatePairOptions) net.Socket.CreatePairError![2]net.Socket,244 netSocketCreatePair: *const fn (?*anyopaque, net.Socket.CreatePairOptions) net.Socket.CreatePairError![2]net.Socket,
245 netSend: *const fn (?*anyopaque, net.Socket.Handle, []net.OutgoingMessage, net.SendFlags) struct { ?net.Socket.SendError, usize },245 netSend: *const fn (?*anyopaque, net.Socket.Handle, []net.OutgoingMessage, net.SendFlags) struct { ?net.Socket.SendError, usize },
246 netReceive: *const fn (?*anyopaque, net.Socket.Handle, message_buffer: []net.IncomingMessage, data_buffer: []u8, net.ReceiveFlags, Timeout) struct { ?net.Socket.ReceiveTimeoutError, usize },
247 /// Returns 0 on end of stream.246 /// Returns 0 on end of stream.
248 netRead: *const fn (?*anyopaque, src: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usize,247 netRead: *const fn (?*anyopaque, src: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usize,
249 netWrite: *const fn (?*anyopaque, dest: net.Socket.Handle, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize,248 netWrite: *const fn (?*anyopaque, dest: net.Socket.Handle, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize,
...@@ -261,6 +260,7 @@ pub const Operation = union(enum) {...@@ -261,6 +260,7 @@ pub const Operation = union(enum) {
261 /// On Windows this is NtDeviceIoControlFile. On POSIX this is ioctl. On260 /// On Windows this is NtDeviceIoControlFile. On POSIX this is ioctl. On
262 /// other systems this tag is unreachable.261 /// other systems this tag is unreachable.
263 device_io_control: DeviceIoControl,262 device_io_control: DeviceIoControl,
263 net_receive: NetReceive,
264264
265 pub const Tag = @typeInfo(Operation).@"union".tag_type.?;265 pub const Tag = @typeInfo(Operation).@"union".tag_type.?;
266266
...@@ -350,6 +350,35 @@ pub const Operation = union(enum) {...@@ -350,6 +350,35 @@ pub const Operation = union(enum) {
350 },350 },
351 };351 };
352352
353 pub const NetReceive = struct {
354 socket_handle: net.Socket.Handle,
355 message_buffer: []net.IncomingMessage,
356 data_buffer: []u8,
357 flags: net.ReceiveFlags,
358
359 pub const Error = error{
360 /// Insufficient memory or other resource internal to the operating system.
361 SystemResources,
362 /// Per-process limit on the number of open file descriptors has been reached.
363 ProcessFdQuotaExceeded,
364 /// System-wide limit on the total number of open files has been reached.
365 SystemFdQuotaExceeded,
366 /// Local end has been shut down on a connection-oriented socket, or
367 /// the socket was never connected.
368 SocketUnconnected,
369 /// The socket type requires that message be sent atomically, and the
370 /// size of the message to be sent made this impossible. The message
371 /// was not transmitted, or was partially transmitted.
372 MessageOversize,
373 /// Network connection was unexpectedly closed by sender.
374 ConnectionResetByPeer,
375 /// The local network interface used to reach the destination is offline.
376 NetworkDown,
377 } || Io.UnexpectedError;
378
379 pub const Result = struct { ?net.Socket.ReceiveError, usize };
380 };
381
353 pub const Result = Result: {382 pub const Result = Result: {
354 const operation_fields = @typeInfo(Operation).@"union".fields;383 const operation_fields = @typeInfo(Operation).@"union".fields;
355 var field_names: [operation_fields.len][]const u8 = undefined;384 var field_names: [operation_fields.len][]const u8 = undefined;
...@@ -417,6 +446,19 @@ pub fn operate(io: Io, operation: Operation) Cancelable!Operation.Result {...@@ -417,6 +446,19 @@ pub fn operate(io: Io, operation: Operation) Cancelable!Operation.Result {
417 return io.vtable.operate(io.userdata, operation);446 return io.vtable.operate(io.userdata, operation);
418}447}
419448
449pub const OperateTimeoutError = Cancelable || Timeout.Error || ConcurrentError;
450
451/// Performs one `Operation` with provided `timeout`.
452pub fn operateTimeout(io: Io, operation: Operation, timeout: Timeout) OperateTimeoutError!Operation.Result {
453 var storage: [1]Operation.Storage = undefined;
454 var batch: Batch = .init(&storage);
455 batch.addAt(0, operation);
456 try batch.awaitConcurrent(io, timeout);
457 const completion = batch.next().?;
458 assert(completion.index == 0);
459 return completion.result;
460}
461
420/// Submits many operations together without waiting for all of them to462/// Submits many operations together without waiting for all of them to
421/// complete.463/// complete.
422///464///
...@@ -1716,7 +1758,7 @@ pub const Event = enum(u32) {...@@ -1716,7 +1758,7 @@ pub const Event = enum(u32) {
1716 }1758 }
17171759
1718 /// Blocks until the logical boolean is `true`.1760 /// Blocks until the logical boolean is `true`.
1719 pub fn wait(event: *Event, io: Io) Io.Cancelable!void {1761 pub fn wait(event: *Event, io: Io) Cancelable!void {
1720 if (@cmpxchgStrong(Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) {1762 if (@cmpxchgStrong(Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) {
1721 .unset => unreachable,1763 .unset => unreachable,
1722 .waiting => {},1764 .waiting => {},
lib/std/Io/Threaded.zig+86-120
...@@ -61,7 +61,7 @@ disable_memory_mapping: bool,...@@ -61,7 +61,7 @@ disable_memory_mapping: bool,
6161
62stderr_writer: File.Writer = .{62stderr_writer: File.Writer = .{
63 .io = undefined,63 .io = undefined,
64 .interface = Io.File.Writer.initInterface(&.{}),64 .interface = File.Writer.initInterface(&.{}),
65 .file = if (is_windows) undefined else .stderr(),65 .file = if (is_windows) undefined else .stderr(),
66 .mode = .streaming,66 .mode = .streaming,
67},67},
...@@ -160,7 +160,7 @@ pub const Environ = struct {...@@ -160,7 +160,7 @@ pub const Environ = struct {
160 },160 },
161 };161 };
162162
163 pub fn scan(environ: *Environ, allocator: std.mem.Allocator) void {163 pub fn scan(environ: *Environ, allocator: Allocator) void {
164 if (is_windows) {164 if (is_windows) {
165 // This value expires with any call that modifies the environment,165 // This value expires with any call that modifies the environment,
166 // which is outside of this Io implementation's control, so references166 // which is outside of this Io implementation's control, so references
...@@ -1901,10 +1901,6 @@ pub fn io(t: *Threaded) Io {...@@ -1901,10 +1901,6 @@ pub fn io(t: *Threaded) Io {
1901 .windows => netSendWindows,1901 .windows => netSendWindows,
1902 else => netSendPosix,1902 else => netSendPosix,
1903 },1903 },
1904 .netReceive = switch (native_os) {
1905 .windows => netReceiveWindows,
1906 else => netReceivePosix,
1907 },
1908 .netInterfaceNameResolve = netInterfaceNameResolve,1904 .netInterfaceNameResolve = netInterfaceNameResolve,
1909 .netInterfaceName = netInterfaceName,1905 .netInterfaceName = netInterfaceName,
1910 .netLookup = netLookup,1906 .netLookup = netLookup,
...@@ -2037,7 +2033,6 @@ pub fn ioBasic(t: *Threaded) Io {...@@ -2037,7 +2033,6 @@ pub fn ioBasic(t: *Threaded) Io {
2037 .netWrite = netWriteUnavailable,2033 .netWrite = netWriteUnavailable,
2038 .netWriteFile = netWriteFileUnavailable,2034 .netWriteFile = netWriteFileUnavailable,
2039 .netSend = netSendUnavailable,2035 .netSend = netSendUnavailable,
2040 .netReceive = netReceiveUnavailable,
2041 .netInterfaceNameResolve = netInterfaceNameResolveUnavailable,2036 .netInterfaceNameResolve = netInterfaceNameResolveUnavailable,
2042 .netInterfaceName = netInterfaceNameUnavailable,2037 .netInterfaceName = netInterfaceNameUnavailable,
2043 .netLookup = netLookupUnavailable,2038 .netLookup = netLookupUnavailable,
...@@ -2638,6 +2633,15 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper...@@ -2638,6 +2633,15 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper
2638 },2633 },
2639 },2634 },
2640 .device_io_control => |*o| return .{ .device_io_control = try deviceIoControl(o) },2635 .device_io_control => |*o| return .{ .device_io_control = try deviceIoControl(o) },
2636 .net_receive => |*o| return .{ .net_receive = o: {
2637 if (!have_networking) break :o .{ error.NetworkDown, 0 };
2638 if (is_windows) break :o netReceiveWindows(t, o.socket_handle, o.message_buffer, o.data_buffer, o.flags);
2639 netReceivePosix(o.socket_handle, &o.message_buffer[0], o.data_buffer, o.flags) catch |err| switch (err) {
2640 error.Canceled => |e| return e,
2641 else => |e| break :o .{ e, 0 },
2642 };
2643 break :o .{ null, 1 };
2644 } },
2641 }2645 }
2642}2646}
26432647
...@@ -2662,11 +2666,19 @@ fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void {...@@ -2662,11 +2666,19 @@ fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void {
2662 const submission = &b.storage[index.toIndex()].submission;2666 const submission = &b.storage[index.toIndex()].submission;
2663 switch (submission.operation) {2667 switch (submission.operation) {
2664 .file_read_streaming => |o| {2668 .file_read_streaming => |o| {
2665 poll_buffer[poll_len] = .{ .fd = o.file.handle, .events = posix.POLL.IN, .revents = 0 };2669 poll_buffer[poll_len] = .{
2670 .fd = o.file.handle,
2671 .events = posix.POLL.IN | posix.POLL.ERR,
2672 .revents = 0,
2673 };
2666 poll_len += 1;2674 poll_len += 1;
2667 },2675 },
2668 .file_write_streaming => |o| {2676 .file_write_streaming => |o| {
2669 poll_buffer[poll_len] = .{ .fd = o.file.handle, .events = posix.POLL.OUT, .revents = 0 };2677 poll_buffer[poll_len] = .{
2678 .fd = o.file.handle,
2679 .events = posix.POLL.OUT | posix.POLL.ERR,
2680 .revents = 0,
2681 };
2670 poll_len += 1;2682 poll_len += 1;
2671 },2683 },
2672 .device_io_control => |o| {2684 .device_io_control => |o| {
...@@ -2677,6 +2689,14 @@ fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void {...@@ -2677,6 +2689,14 @@ fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void {
2677 };2689 };
2678 poll_len += 1;2690 poll_len += 1;
2679 },2691 },
2692 .net_receive => |*o| {
2693 poll_buffer[poll_len] = .{
2694 .fd = o.socket_handle,
2695 .events = posix.POLL.IN | posix.POLL.ERR,
2696 .revents = 0,
2697 };
2698 poll_len += 1;
2699 },
2680 }2700 }
2681 index = submission.node.next;2701 index = submission.node.next;
2682 }2702 }
...@@ -2796,12 +2816,12 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout...@@ -2796,12 +2816,12 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout
2796 if (!have_poll) return error.ConcurrencyUnavailable;2816 if (!have_poll) return error.ConcurrencyUnavailable;
2797 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;2817 var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
2798 var poll_storage: struct {2818 var poll_storage: struct {
2799 gpa: std.mem.Allocator,2819 gpa: Allocator,
2800 batch: *Io.Batch,2820 batch: *Io.Batch,
2801 slice: []posix.pollfd,2821 slice: []posix.pollfd,
2802 len: u32,2822 len: u32,
28032823
2804 fn add(storage: *@This(), file: Io.File, events: @FieldType(posix.pollfd, "events")) Io.ConcurrentError!void {2824 fn add(storage: *@This(), fd: File.Handle, events: @FieldType(posix.pollfd, "events")) Io.ConcurrentError!void {
2805 const len = storage.len;2825 const len = storage.len;
2806 if (len == poll_buffer_len) {2826 if (len == poll_buffer_len) {
2807 const slice: []posix.pollfd = if (storage.batch.userdata) |batch_userdata|2827 const slice: []posix.pollfd = if (storage.batch.userdata) |batch_userdata|
...@@ -2816,7 +2836,7 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout...@@ -2816,7 +2836,7 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout
2816 storage.slice = slice;2836 storage.slice = slice;
2817 }2837 }
2818 storage.slice[len] = .{2838 storage.slice[len] = .{
2819 .fd = file.handle,2839 .fd = fd,
2820 .events = events,2840 .events = events,
2821 .revents = 0,2841 .revents = 0,
2822 };2842 };
...@@ -2828,9 +2848,10 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout...@@ -2828,9 +2848,10 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout
2828 while (index != .none) {2848 while (index != .none) {
2829 const submission = &b.storage[index.toIndex()].submission;2849 const submission = &b.storage[index.toIndex()].submission;
2830 switch (submission.operation) {2850 switch (submission.operation) {
2831 .file_read_streaming => |o| try poll_storage.add(o.file, posix.POLL.IN),2851 .file_read_streaming => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.ERR),
2832 .file_write_streaming => |o| try poll_storage.add(o.file, posix.POLL.OUT),2852 .file_write_streaming => |o| try poll_storage.add(o.file.handle, posix.POLL.OUT | posix.POLL.ERR),
2833 .device_io_control => |o| try poll_storage.add(o.file, posix.POLL.IN | posix.POLL.OUT | posix.POLL.ERR),2853 .device_io_control => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.OUT | posix.POLL.ERR),
2854 .net_receive => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR),
2834 }2855 }
2835 index = submission.node.next;2856 index = submission.node.next;
2836 }2857 }
...@@ -3000,6 +3021,7 @@ fn batchApc(...@@ -3000,6 +3021,7 @@ fn batchApc(
3000 .file_read_streaming => .{ .file_read_streaming = ntReadFileResult(iosb) },3021 .file_read_streaming => .{ .file_read_streaming = ntReadFileResult(iosb) },
3001 .file_write_streaming => .{ .file_write_streaming = ntWriteFileResult(iosb) },3022 .file_write_streaming => .{ .file_write_streaming = ntWriteFileResult(iosb) },
3002 .device_io_control => .{ .device_io_control = iosb.* },3023 .device_io_control => .{ .device_io_control = iosb.* },
3024 .net_receive => unreachable, // TODO
3003 };3025 };
3004 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };3026 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
3005 },3027 },
...@@ -3201,6 +3223,11 @@ fn batchDrainSubmittedWindows(b: *Io.Batch, concurrency: bool) (Io.ConcurrentErr...@@ -3201,6 +3223,11 @@ fn batchDrainSubmittedWindows(b: *Io.Batch, concurrency: bool) (Io.ConcurrentErr
3201 };3223 };
3202 }3224 }
3203 },3225 },
3226 .net_receive => |o| {
3227 if (concurrency) return error.ConcurrencyUnavailable;
3228 _ = o;
3229 @panic("TODO implement Batch NetReceive on Windows");
3230 },
3204 }3231 }
3205 index = submission.node.next;3232 index = submission.node.next;
3206 }3233 }
...@@ -13190,70 +13217,42 @@ fn netSendMany(...@@ -13190,70 +13217,42 @@ fn netSendMany(
13190}13217}
1319113218
13192fn netReceivePosix(13219fn netReceivePosix(
13193 userdata: ?*anyopaque,13220 socket_handle: net.Socket.Handle,
13194 handle: net.Socket.Handle,13221 message: *net.IncomingMessage,
13195 message_buffer: []net.IncomingMessage,
13196 data_buffer: []u8,13222 data_buffer: []u8,
13197 flags: net.ReceiveFlags,13223 flags: net.ReceiveFlags,
13198 timeout: Io.Timeout,13224) net.Socket.ReceiveError!void {
13199) struct { ?net.Socket.ReceiveTimeoutError, usize } {
13200 if (!have_networking) return .{ error.NetworkDown, 0 };
13201 const t: *Threaded = @ptrCast(@alignCast(userdata));
13202 const t_io = io(t);
13203
13204 // recvmmsg is useless, here's why:13225 // recvmmsg is useless, here's why:
13205 // * [timeout bug](https://bugzilla.kernel.org/show_bug.cgi?id=75371)13226 // * [timeout bug](https://bugzilla.kernel.org/show_bug.cgi?id=75371)
13206 // * it wants iovecs for each message but we have a better API: one data13227 // * it wants iovecs for each message but we have a better API: one data
13207 // buffer to handle all the messages. The better API cannot be lowered to13228 // buffer to handle all the messages. The better API cannot be lowered to
13208 // the split vectors though because reducing the buffer size might make13229 // the split vectors though because reducing the buffer size might make
13209 // some messages unreceivable.13230 // some messages unreceivable.
13210
13211 // So the strategy instead is to use non-blocking recvmsg calls, calling
13212 // poll() with timeout if the first one returns EAGAIN.
13213 const posix_flags: u32 =13231 const posix_flags: u32 =
13214 @as(u32, if (flags.oob) posix.MSG.OOB else 0) |13232 @as(u32, if (flags.oob) posix.MSG.OOB else 0) |
13215 @as(u32, if (flags.peek) posix.MSG.PEEK else 0) |13233 @as(u32, if (flags.peek) posix.MSG.PEEK else 0) |
13216 @as(u32, if (flags.trunc) posix.MSG.TRUNC else 0) |13234 @as(u32, if (flags.trunc) posix.MSG.TRUNC else 0) |
13217 posix.MSG.DONTWAIT | posix.MSG.NOSIGNAL;13235 posix.MSG.NOSIGNAL;
1321813236
13219 var poll_fds: [1]posix.pollfd = .{13237 var storage: PosixAddress = undefined;
13220 .{13238 var iov: posix.iovec = .{ .base = data_buffer.ptr, .len = data_buffer.len };
13221 .fd = handle,13239 var msg: posix.msghdr = .{
13222 .events = posix.POLL.IN,13240 .name = &storage.any,
13223 .revents = undefined,13241 .namelen = @sizeOf(PosixAddress),
13224 },13242 .iov = (&iov)[0..1],
13243 .iovlen = 1,
13244 .control = message.control.ptr,
13245 .controllen = @intCast(message.control.len),
13246 .flags = undefined,
13225 };13247 };
13226 var message_i: usize = 0;
13227 var data_i: usize = 0;
13228
13229 const deadline = timeout.toTimestamp(t_io);
1323013248
13231 recv: while (true) {13249 const syscall = try Syscall.start();
13232 if (message_buffer.len - message_i == 0) return .{ null, message_i };13250 while (true) {
13233 const message = &message_buffer[message_i];13251 const rc = posix.system.recvmsg(socket_handle, &msg, posix_flags);
13234 const remaining_data_buffer = data_buffer[data_i..];13252 switch (posix.errno(rc)) {
13235 var storage: PosixAddress = undefined;
13236 var iov: posix.iovec = .{ .base = remaining_data_buffer.ptr, .len = remaining_data_buffer.len };
13237 var msg: posix.msghdr = .{
13238 .name = &storage.any,
13239 .namelen = @sizeOf(PosixAddress),
13240 .iov = (&iov)[0..1],
13241 .iovlen = 1,
13242 .control = message.control.ptr,
13243 .controllen = @intCast(message.control.len),
13244 .flags = undefined,
13245 };
13246
13247 const recv_rc = rc: {
13248 const syscall = Syscall.start() catch |err| return .{ err, message_i };
13249 const rc = posix.system.recvmsg(handle, &msg, posix_flags);
13250 syscall.finish();
13251 break :rc rc;
13252 };
13253 switch (posix.errno(recv_rc)) {
13254 .SUCCESS => {13253 .SUCCESS => {
13255 const data = remaining_data_buffer[0..@intCast(recv_rc)];13254 syscall.finish();
13256 data_i += data.len;13255 const data = data_buffer[0..@intCast(rc)];
13257 message.* = .{13256 message.* = .{
13258 .from = addressFromPosix(&storage),13257 .from = addressFromPosix(&storage),
13259 .data = data,13258 .data = data,
...@@ -13266,78 +13265,45 @@ fn netReceivePosix(...@@ -13266,78 +13265,45 @@ fn netReceivePosix(
13266 .errqueue = if (@hasDecl(posix.MSG, "ERRQUEUE")) (msg.flags & posix.MSG.ERRQUEUE) != 0 else false,13265 .errqueue = if (@hasDecl(posix.MSG, "ERRQUEUE")) (msg.flags & posix.MSG.ERRQUEUE) != 0 else false,
13267 },13266 },
13268 };13267 };
13269 message_i += 1;13268 return;
13270 continue;
13271 },13269 },
13272 .AGAIN => while (true) {13270 .INTR => {
13273 if (message_i != 0) return .{ null, message_i };13271 try syscall.checkCancel();
1327413272 continue;
13275 const max_poll_ms = std.math.maxInt(u31);
13276 const timeout_ms: u31 = if (deadline) |d| t: {
13277 const duration = d.durationFromNow(t_io);
13278 if (duration.raw.nanoseconds <= 0) return .{ error.Timeout, message_i };
13279 break :t @intCast(@min(max_poll_ms, duration.raw.toMilliseconds()));
13280 } else max_poll_ms;
13281
13282 const syscall = Syscall.start() catch |err| return .{ err, message_i };
13283 const poll_rc = posix.system.poll(&poll_fds, poll_fds.len, timeout_ms);
13284 syscall.finish();
13285
13286 switch (posix.errno(poll_rc)) {
13287 .SUCCESS => {
13288 if (poll_rc == 0) {
13289 // Although spurious timeouts are OK, when no deadline
13290 // is passed we must not return `error.Timeout`.
13291 if (deadline == null) continue;
13292 return .{ error.Timeout, message_i };
13293 }
13294 continue :recv;
13295 },
13296 .INTR => continue,
13297
13298 .FAULT => |err| return .{ errnoBug(err), message_i },
13299 .INVAL => |err| return .{ errnoBug(err), message_i },
13300 .NOMEM => return .{ error.SystemResources, message_i },
13301 else => |err| return .{ posix.unexpectedErrno(err), message_i },
13302 }
13303 },13273 },
13304 .INTR => continue,13274 .NFILE => return syscall.fail(error.SystemFdQuotaExceeded),
1330513275 .MFILE => return syscall.fail(error.ProcessFdQuotaExceeded),
13306 .BADF => |err| return .{ errnoBug(err), message_i },13276 .NOBUFS => return syscall.fail(error.SystemResources),
13307 .NFILE => return .{ error.SystemFdQuotaExceeded, message_i },13277 .NOMEM => return syscall.fail(error.SystemResources),
13308 .MFILE => return .{ error.ProcessFdQuotaExceeded, message_i },13278 .NOTCONN => return syscall.fail(error.SocketUnconnected),
13309 .FAULT => |err| return .{ errnoBug(err), message_i },13279 .MSGSIZE => return syscall.fail(error.MessageOversize),
13310 .INVAL => |err| return .{ errnoBug(err), message_i },13280 .PIPE => return syscall.fail(error.SocketUnconnected),
13311 .NOBUFS => return .{ error.SystemResources, message_i },13281 .CONNRESET => return syscall.fail(error.ConnectionResetByPeer),
13312 .NOMEM => return .{ error.SystemResources, message_i },13282 .NETDOWN => return syscall.fail(error.NetworkDown),
13313 .NOTCONN => return .{ error.SocketUnconnected, message_i },13283 .AGAIN => |err| return syscall.errnoBug(err),
13314 .NOTSOCK => |err| return .{ errnoBug(err), message_i },13284 .BADF => |err| return syscall.errnoBug(err),
13315 .MSGSIZE => return .{ error.MessageOversize, message_i },13285 .FAULT => |err| return syscall.errnoBug(err),
13316 .PIPE => return .{ error.SocketUnconnected, message_i },13286 .INVAL => |err| return syscall.errnoBug(err),
13317 .OPNOTSUPP => |err| return .{ errnoBug(err), message_i },13287 .NOTSOCK => |err| return syscall.errnoBug(err),
13318 .CONNRESET => return .{ error.ConnectionResetByPeer, message_i },13288 .OPNOTSUPP => |err| return syscall.errnoBug(err),
13319 .NETDOWN => return .{ error.NetworkDown, message_i },13289 else => |err| return syscall.unexpectedErrno(err),
13320 else => |err| return .{ posix.unexpectedErrno(err), message_i },
13321 }13290 }
13322 }13291 }
13323}13292}
1332413293
13325fn netReceiveWindows(13294fn netReceiveWindows(
13326 userdata: ?*anyopaque,13295 t: *Threaded,
13327 handle: net.Socket.Handle,13296 socket_handle: net.Socket.Handle,
13328 message_buffer: []net.IncomingMessage,13297 message_buffer: []net.IncomingMessage,
13329 data_buffer: []u8,13298 data_buffer: []u8,
13330 flags: net.ReceiveFlags,13299 flags: net.ReceiveFlags,
13331 timeout: Io.Timeout,13300) struct { ?net.Socket.ReceiveError, usize } {
13332) struct { ?net.Socket.ReceiveTimeoutError, usize } {
13333 if (!have_networking) return .{ error.NetworkDown, 0 };13301 if (!have_networking) return .{ error.NetworkDown, 0 };
13334 const t: *Threaded = @ptrCast(@alignCast(userdata));
13335 _ = t;13302 _ = t;
13336 _ = handle;13303 _ = socket_handle;
13337 _ = message_buffer;13304 _ = message_buffer;
13338 _ = data_buffer;13305 _ = data_buffer;
13339 _ = flags;13306 _ = flags;
13340 _ = timeout;
13341 @panic("TODO implement netReceiveWindows");13307 @panic("TODO implement netReceiveWindows");
13342}13308}
1334313309
lib/std/Io/net.zig+15-22
...@@ -1109,25 +1109,7 @@ pub const Socket = struct {...@@ -1109,25 +1109,7 @@ pub const Socket = struct {
1109 if (n != messages.len) return err.?;1109 if (n != messages.len) return err.?;
1110 }1110 }
11111111
1112 pub const ReceiveError = error{1112 pub const ReceiveError = Io.Operation.NetReceive.Error || Io.Cancelable;
1113 /// Insufficient memory or other resource internal to the operating system.
1114 SystemResources,
1115 /// Per-process limit on the number of open file descriptors has been reached.
1116 ProcessFdQuotaExceeded,
1117 /// System-wide limit on the total number of open files has been reached.
1118 SystemFdQuotaExceeded,
1119 /// Local end has been shut down on a connection-oriented socket, or
1120 /// the socket was never connected.
1121 SocketUnconnected,
1122 /// The socket type requires that message be sent atomically, and the
1123 /// size of the message to be sent made this impossible. The message
1124 /// was not transmitted, or was partially transmitted.
1125 MessageOversize,
1126 /// Network connection was unexpectedly closed by sender.
1127 ConnectionResetByPeer,
1128 /// The local network interface used to reach the destination is offline.
1129 NetworkDown,
1130 } || Io.UnexpectedError || Io.Cancelable;
11311113
1132 /// Waits for data. Connectionless.1114 /// Waits for data. Connectionless.
1133 ///1115 ///
...@@ -1145,7 +1127,7 @@ pub const Socket = struct {...@@ -1145,7 +1127,7 @@ pub const Socket = struct {
1145 return message;1127 return message;
1146 }1128 }
11471129
1148 pub const ReceiveTimeoutError = ReceiveError || Io.Timeout.Error;1130 pub const ReceiveTimeoutError = ReceiveError || Io.Timeout.Error || Io.ConcurrentError;
11491131
1150 /// Waits for data. Connectionless.1132 /// Waits for data. Connectionless.
1151 ///1133 ///
...@@ -1161,7 +1143,12 @@ pub const Socket = struct {...@@ -1161,7 +1143,12 @@ pub const Socket = struct {
1161 timeout: Io.Timeout,1143 timeout: Io.Timeout,
1162 ) ReceiveTimeoutError!IncomingMessage {1144 ) ReceiveTimeoutError!IncomingMessage {
1163 var message: IncomingMessage = .init;1145 var message: IncomingMessage = .init;
1164 const maybe_err, const count = io.vtable.netReceive(io.userdata, s.handle, (&message)[0..1], buffer, .{}, timeout);1146 const maybe_err, const count = (try io.operateTimeout(.{ .net_receive = .{
1147 .socket = s.handle,
1148 .message_buffer = (&message)[0..1],
1149 .data_buffer = buffer,
1150 .flags = .{},
1151 } }, timeout)).net_receive;
1165 if (maybe_err) |err| return err;1152 if (maybe_err) |err| return err;
1166 assert(1 == count);1153 assert(1 == count);
1167 return message;1154 return message;
...@@ -1186,7 +1173,13 @@ pub const Socket = struct {...@@ -1186,7 +1173,13 @@ pub const Socket = struct {
1186 flags: ReceiveFlags,1173 flags: ReceiveFlags,
1187 timeout: Io.Timeout,1174 timeout: Io.Timeout,
1188 ) struct { ?ReceiveTimeoutError, usize } {1175 ) struct { ?ReceiveTimeoutError, usize } {
1189 return io.vtable.netReceive(io.userdata, s.handle, message_buffer, data_buffer, flags, timeout);1176 const result = io.operateTimeout(.{ .net_receive = .{
1177 .socket_handle = s.handle,
1178 .message_buffer = message_buffer,
1179 .data_buffer = data_buffer,
1180 .flags = flags,
1181 } }, timeout) catch |err| return .{ err, 0 };
1182 return result.net_receive;
1190 }1183 }
11911184
1192 pub const CreatePairError = error{1185 pub const CreatePairError = error{