diff --git a/lib/std/Io.zig b/lib/std/Io.zig index 14c7306ca6673d99bad8f3b8b5eafc150f935f74..9bd7ec5a0dc76ed8628b7a9933b7c1bf41203ec3 100644 --- a/lib/std/Io.zig +++ b/lib/std/Io.zig @@ -229,7 +229,6 @@ pub const VTable = struct { randomSecure: *const fn (?*anyopaque, buffer: []u8) RandomSecureError!void, netListenIp: *const fn (?*anyopaque, address: *const net.IpAddress, net.IpAddress.ListenOptions) net.IpAddress.ListenError!net.Socket, - netAccept: *const fn (?*anyopaque, server: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket, netBindIp: *const fn (?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.BindOptions) net.IpAddress.BindError!net.Socket, netConnectIp: *const fn (?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.ConnectOptions) net.IpAddress.ConnectError!net.Socket, netListenUnix: *const fn (?*anyopaque, *const net.UnixAddress, net.UnixAddress.ListenOptions) net.UnixAddress.ListenError!net.Socket.Handle, @@ -251,6 +250,7 @@ pub const Operation = union(enum) { device_io_control: DeviceIoControl, net_receive: NetReceive, net_send: NetSend, + net_accept: NetAccept, net_read: NetRead, net_write: NetWrite, @@ -342,6 +342,35 @@ pub const Operation = union(enum) { }, }; + pub const NetAccept = struct { + socket_handle: net.Socket.Handle, + options: net.Server.AcceptOptions, + + pub const Error = error{ + /// The per-process limit on the number of open file descriptors has been reached. + ProcessFdQuotaExceeded, + /// The system-wide limit on the total number of open files has been reached. + SystemFdQuotaExceeded, + /// Not enough free memory. This often means that the memory allocation is limited + /// by the socket buffer limits, not by the system memory. + SystemResources, + /// Either `listen` was never called, or `shutdown` was called (possibly while + /// this call was blocking). This allows `shutdown` to be used as a concurrent + /// cancellation mechanism. + SocketNotListening, + /// The network subsystem has failed. + NetworkDown, + /// An incoming connection was indicated, but was subsequently terminated by the + /// remote peer prior to accepting the call. + ConnectionAborted, + /// Firewall rules forbid connection. + BlockedByFirewall, + ProtocolFailure, + } || Io.UnexpectedError; + + pub const Result = Error!net.Socket; + }; + pub const NetReceive = struct { socket_handle: net.Socket.Handle, message_buffer: []net.IncomingMessage, @@ -2819,7 +2848,6 @@ pub const failing: std.Io = .{ .sleep = noSleep, .netListenIp = failingNetListenIp, - .netAccept = failingNetAccept, .netBindIp = failingNetBindIp, .netConnectIp = failingNetConnectIp, .netListenUnix = failingNetListenUnix, @@ -2972,6 +3000,7 @@ pub fn failingOperate(userdata: ?*anyopaque, operation: Operation) Cancelable!Op .device_io_control => unreachable, .net_receive => .{ .net_receive = .{ error.NetworkDown, 0 } }, .net_send => .{ .net_send = .{ error.NetworkDown, 0 } }, + .net_accept => .{ .net_accept = error.NetworkDown }, .net_read => .{ .net_read = error.NetworkDown }, .net_write => .{ .net_write = error.NetworkDown }, }; diff --git a/lib/std/Io/Dispatch.zig b/lib/std/Io/Dispatch.zig index 99ad98ca71ce921ebc0e6edadfa1456a7f672e84..e79c79dd356262dc1272e379d1d0bedcb5c78d50 100644 --- a/lib/std/Io/Dispatch.zig +++ b/lib/std/Io/Dispatch.zig @@ -452,7 +452,6 @@ pub fn io(ev: *Evented) Io { .randomSecure = randomSecure, .netListenIp = netListenIpUnavailable, - .netAccept = netAcceptUnavailable, .netBindIp = netBindIpUnavailable, .netConnectIp = netConnectIpUnavailable, .netListenUnix = netListenUnixUnavailable, @@ -1712,6 +1711,7 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper .device_io_control => |*o| return .{ .device_io_control = try deviceIoControl(o) }, .net_receive => @panic("TODO implement net_receive operation"), .net_send => @panic("TODO implement net_send operation"), + .net_accept => @panic("TODO implement net_accept operation"), .net_read => @panic("TODO implement net_read operation"), .net_write => @panic("TODO implement net_write operation"), } @@ -4803,18 +4803,6 @@ fn netListenIpUnavailable( return error.NetworkDown; } -fn netAcceptUnavailable( - userdata: ?*anyopaque, - listen_handle: net.Socket.Handle, - options: net.Server.AcceptOptions, -) net.Server.AcceptError!net.Socket { - const ev: *Evented = @ptrCast(@alignCast(userdata)); - _ = ev; - _ = listen_handle; - _ = options; - return error.NetworkDown; -} - fn netBindIpUnavailable( userdata: ?*anyopaque, address: *const net.IpAddress, diff --git a/lib/std/Io/Kqueue.zig b/lib/std/Io/Kqueue.zig index 8f8be91d59bd3f3131ad5c08955da9de09b90e26..20e7820d0611c944701b599f703ddcd8424e541d 100644 --- a/lib/std/Io/Kqueue.zig +++ b/lib/std/Io/Kqueue.zig @@ -646,7 +646,6 @@ pub fn io(k: *Kqueue) Io { .netListenIp = netListenIp, .netListenUnix = netListenUnix, - .netAccept = netAccept, .netBindIp = netBindIp, .netConnectIp = netConnectIp, .netConnectUnix = netConnectUnix, @@ -1008,13 +1007,6 @@ fn netListenIp( _ = options; @panic("TODO"); } -fn netAccept(userdata: ?*anyopaque, server: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket { - const k: *Kqueue = @ptrCast(@alignCast(userdata)); - _ = k; - _ = server; - _ = options; - @panic("TODO"); -} fn netBindIp( userdata: ?*anyopaque, address: *const net.IpAddress, diff --git a/lib/std/Io/Threaded.zig b/lib/std/Io/Threaded.zig index dae5f07f95685e5d4e6a8b883f3a392b65d8bd54..c81ef369fb203d8f22e98b6b4a5e6472d642fcbe 100644 --- a/lib/std/Io/Threaded.zig +++ b/lib/std/Io/Threaded.zig @@ -1969,10 +1969,6 @@ pub fn io(t: *Threaded) Io { .windows => netListenUnixWindows, else => netListenUnixPosix, }, - .netAccept = switch (native_os) { - .windows => netAcceptWindows, - else => netAcceptPosix, - }, .netBindIp = switch (native_os) { .windows => netBindIpWindows, else => netBindIpPosix, @@ -2625,6 +2621,12 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper break :o .{ null, sent }; }, }, + .net_accept => |o| return .{ + .net_accept = netAccept(t, o.socket_handle, o.options) catch |err| switch (err) { + error.Canceled => |e| return e, + else => |e| e, + }, + }, .net_read => |o| return .{ .net_read = netRead(o.socket_handle, o.data, o.control) catch |err| switch (err) { error.Canceled => |e| return e, @@ -2698,6 +2700,14 @@ fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void { }; poll_len += 1; }, + .net_accept => |o| { + poll_buffer[poll_len] = .{ + .fd = o.socket_handle, + .events = posix.POLL.IN | posix.POLL.ERR, + .revents = 0, + }; + poll_len += 1; + }, .net_read => |o| { poll_buffer[poll_len] = .{ .fd = o.socket_handle, @@ -2917,6 +2927,7 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } }; b.completed.tail = index; }, + .net_accept => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR), .net_read => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR), .net_write => |o| try poll_storage.add(o.socket_handle, posix.POLL.OUT | posix.POLL.ERR), } @@ -3115,6 +3126,7 @@ fn batchApc( .device_io_control => .{ .device_io_control = iosb.* }, .net_receive => unreachable, .net_send => unreachable, + .net_accept => unreachable, .net_read => unreachable, .net_write => unreachable, }; @@ -3332,6 +3344,16 @@ fn batchDrainSubmittedWindows(t: *Threaded, b: *Io.Batch, concurrency: bool) (Io .net_send = netSendWindows(t, o.socket_handle, o.messages, o.flags), }); }, + .net_accept => |o| { + // TODO integrate with overlapped I/O or equivalent to avoid this error + if (concurrency) return error.ConcurrencyUnavailable; + batchCompleteBlockingWindows(b, operation_userdata, .{ + .net_accept = netAccept(t, o.socket_handle, o.options) catch |err| switch (err) { + error.Canceled => |e| return e, + else => |e| e, + }, + }); + }, .net_read => |*o| { // TODO integrate with overlapped I/O or equivalent to avoid this error if (concurrency) return error.ConcurrencyUnavailable; @@ -12813,11 +12835,15 @@ fn bindSocketUnixAfd(socket_handle: net.Socket.Handle, address: *const net.UnixA } } -fn netAcceptPosix(userdata: ?*anyopaque, listen_fd: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket { +fn netAccept(t: *Threaded, listen_fd: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket { if (!have_networking) return error.NetworkDown; - const t: *Threaded = @ptrCast(@alignCast(userdata)); - _ = t; - options; + return if (is_windows) + netAcceptWindows(t, listen_fd, options) + else + netAcceptPosix(listen_fd); +} + +fn netAcceptPosix(listen_fd: net.Socket.Handle) net.Server.AcceptError!net.Socket { var storage: PosixAddress = undefined; var addr_len: posix.socklen_t = @sizeOf(PosixAddress); const syscall: Syscall = try .start(); @@ -12862,9 +12888,7 @@ fn netAcceptPosix(userdata: ?*anyopaque, listen_fd: net.Socket.Handle, options: return .{ .handle = fd, .address = addressFromPosix(&storage) }; } -fn netAcceptWindows(userdata: ?*anyopaque, listen_handle: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket { - if (!have_networking) return error.NetworkDown; - const t: *Threaded = @ptrCast(@alignCast(userdata)); +fn netAcceptWindows(t: *Threaded, listen_handle: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket { const Storage = extern struct { Info: windows.AFD.LISTEN_RESPONSE_INFO, RemoteAddress: extern union { posix: PosixAddress, unix: UnixAddress }, diff --git a/lib/std/Io/Uring.zig b/lib/std/Io/Uring.zig index 907d96c6fc7d12264d6a1615ff1f46133dc5914a..e4edcb03cfe59a32dbd7ee6fc6786397dfddef24 100644 --- a/lib/std/Io/Uring.zig +++ b/lib/std/Io/Uring.zig @@ -772,7 +772,6 @@ pub fn io(ev: *Evented) Io { .randomSecure = randomSecure, .netListenIp = netListenIpUnavailable, - .netAccept = netAcceptUnavailable, .netBindIp = netBindIp, .netConnectIp = netConnectIpUnavailable, .netListenUnix = netListenUnixUnavailable, @@ -2111,6 +2110,12 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper break :r .{ error.NetworkDown, 0 }; // TODO }, }, + .net_accept => |o| .{ + .net_accept = r: { + _ = o; + break :r error.NetworkDown; // TODO + }, + }, .net_read => |o| .{ .net_read = r: { _ = o; @@ -4995,18 +5000,6 @@ fn netListenIpUnavailable( return error.NetworkDown; } -fn netAcceptUnavailable( - userdata: ?*anyopaque, - listen_handle: net.Socket.Handle, - options: net.Server.AcceptOptions, -) net.Server.AcceptError!net.Socket { - const ev: *Evented = @ptrCast(@alignCast(userdata)); - _ = ev; - _ = listen_handle; - _ = options; - return error.NetworkDown; -} - fn netBindIp( userdata: ?*anyopaque, address: *const net.IpAddress, diff --git a/lib/std/Io/net.zig b/lib/std/Io/net.zig index d128f2d2b37f0f8383e72953c0e345e0a0d6f7e9..6d2524427cb6fe37f9a74b6175cdfbfc58e0493b 100644 --- a/lib/std/Io/net.zig +++ b/lib/std/Io/net.zig @@ -1556,30 +1556,7 @@ pub const Server = struct { s.* = undefined; } - pub const AcceptError = error{ - /// The per-process limit on the number of open file descriptors has been reached. - ProcessFdQuotaExceeded, - /// The system-wide limit on the total number of open files has been reached. - SystemFdQuotaExceeded, - /// Not enough free memory. This often means that the memory allocation is limited - /// by the socket buffer limits, not by the system memory. - SystemResources, - /// Either `listen` was never called, or `shutdown` was called (possibly while - /// this call was blocking). This allows `shutdown` to be used as a concurrent - /// cancellation mechanism. - SocketNotListening, - /// The network subsystem has failed. - NetworkDown, - /// No connection is already queued and ready to be accepted, and - /// the socket is configured as non-blocking. - WouldBlock, - /// An incoming connection was indicated, but was subsequently terminated by the - /// remote peer prior to accepting the call. - ConnectionAborted, - /// Firewall rules forbid connection. - BlockedByFirewall, - ProtocolFailure, - } || Io.UnexpectedError || Io.Cancelable; + pub const AcceptError = Io.Operation.NetAccept.Error || Io.Cancelable; pub const AcceptOptions = switch (native_os) { .windows => struct { mode: Socket.Mode, protocol: ?Protocol }, @@ -1588,7 +1565,10 @@ pub const Server = struct { /// Blocks until a client connects to the server. pub fn accept(s: *Server, io: Io) AcceptError!Stream { - return .{ .socket = try io.vtable.netAccept(io.userdata, s.socket.handle, s.options) }; + return .{ .socket = try (try io.operate(.{ .net_accept = .{ + .socket_handle = s.socket.handle, + .options = s.options, + } })).net_accept }; } };