authorgravatar for bblack@wikimedia.orgBrandon Black <bblack@wikimedia.org> 2026-05-12 05:14:38-05:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-10-10 00:47:09+02:00
log71f6780c98c0f84668385a901bb591d19aa3944d
tree713f57dd6759464f9235e8b1c6ff5e3aba6b54cd
parentf46b6501a2626dae6b6419e904972ab702fab27e

Move netAccept to an Operation

6 files changed, 78 insertions(+), 72 deletions(-)

lib/std/Io.zig+31-2
......@@ -229,7 +229,6 @@ pub const VTable = struct {
229229 randomSecure: *const fn (?*anyopaque, buffer: []u8) RandomSecureError!void,
230230
231231 netListenIp: *const fn (?*anyopaque, address: *const net.IpAddress, net.IpAddress.ListenOptions) net.IpAddress.ListenError!net.Socket,
232 netAccept: *const fn (?*anyopaque, server: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket,
233232 netBindIp: *const fn (?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.BindOptions) net.IpAddress.BindError!net.Socket,
234233 netConnectIp: *const fn (?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.ConnectOptions) net.IpAddress.ConnectError!net.Socket,
235234 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) {
251250 device_io_control: DeviceIoControl,
252251 net_receive: NetReceive,
253252 net_send: NetSend,
253 net_accept: NetAccept,
254254 net_read: NetRead,
255255 net_write: NetWrite,
256256
......@@ -342,6 +342,35 @@ pub const Operation = union(enum) {
342342 },
343343 };
344344
345 pub const NetAccept = struct {
346 socket_handle: net.Socket.Handle,
347 options: net.Server.AcceptOptions,
348
349 pub const Error = error{
350 /// The per-process limit on the number of open file descriptors has been reached.
351 ProcessFdQuotaExceeded,
352 /// The system-wide limit on the total number of open files has been reached.
353 SystemFdQuotaExceeded,
354 /// Not enough free memory. This often means that the memory allocation is limited
355 /// by the socket buffer limits, not by the system memory.
356 SystemResources,
357 /// Either `listen` was never called, or `shutdown` was called (possibly while
358 /// this call was blocking). This allows `shutdown` to be used as a concurrent
359 /// cancellation mechanism.
360 SocketNotListening,
361 /// The network subsystem has failed.
362 NetworkDown,
363 /// An incoming connection was indicated, but was subsequently terminated by the
364 /// remote peer prior to accepting the call.
365 ConnectionAborted,
366 /// Firewall rules forbid connection.
367 BlockedByFirewall,
368 ProtocolFailure,
369 } || Io.UnexpectedError;
370
371 pub const Result = Error!net.Socket;
372 };
373
345374 pub const NetReceive = struct {
346375 socket_handle: net.Socket.Handle,
347376 message_buffer: []net.IncomingMessage,
......@@ -2819,7 +2848,6 @@ pub const failing: std.Io = .{
28192848 .sleep = noSleep,
28202849
28212850 .netListenIp = failingNetListenIp,
2822 .netAccept = failingNetAccept,
28232851 .netBindIp = failingNetBindIp,
28242852 .netConnectIp = failingNetConnectIp,
28252853 .netListenUnix = failingNetListenUnix,
......@@ -2972,6 +3000,7 @@ pub fn failingOperate(userdata: ?*anyopaque, operation: Operation) Cancelable!Op
29723000 .device_io_control => unreachable,
29733001 .net_receive => .{ .net_receive = .{ error.NetworkDown, 0 } },
29743002 .net_send => .{ .net_send = .{ error.NetworkDown, 0 } },
3003 .net_accept => .{ .net_accept = error.NetworkDown },
29753004 .net_read => .{ .net_read = error.NetworkDown },
29763005 .net_write => .{ .net_write = error.NetworkDown },
29773006 };
lib/std/Io/Dispatch.zig+1-13
......@@ -452,7 +452,6 @@ pub fn io(ev: *Evented) Io {
452452 .randomSecure = randomSecure,
453453
454454 .netListenIp = netListenIpUnavailable,
455 .netAccept = netAcceptUnavailable,
456455 .netBindIp = netBindIpUnavailable,
457456 .netConnectIp = netConnectIpUnavailable,
458457 .netListenUnix = netListenUnixUnavailable,
......@@ -1712,6 +1711,7 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper
17121711 .device_io_control => |*o| return .{ .device_io_control = try deviceIoControl(o) },
17131712 .net_receive => @panic("TODO implement net_receive operation"),
17141713 .net_send => @panic("TODO implement net_send operation"),
1714 .net_accept => @panic("TODO implement net_accept operation"),
17151715 .net_read => @panic("TODO implement net_read operation"),
17161716 .net_write => @panic("TODO implement net_write operation"),
17171717 }
......@@ -4803,18 +4803,6 @@ fn netListenIpUnavailable(
48034803 return error.NetworkDown;
48044804}
48054805
4806fn netAcceptUnavailable(
4807 userdata: ?*anyopaque,
4808 listen_handle: net.Socket.Handle,
4809 options: net.Server.AcceptOptions,
4810) net.Server.AcceptError!net.Socket {
4811 const ev: *Evented = @ptrCast(@alignCast(userdata));
4812 _ = ev;
4813 _ = listen_handle;
4814 _ = options;
4815 return error.NetworkDown;
4816}
4817
48184806fn netBindIpUnavailable(
48194807 userdata: ?*anyopaque,
48204808 address: *const net.IpAddress,
lib/std/Io/Kqueue.zig-8
......@@ -646,7 +646,6 @@ pub fn io(k: *Kqueue) Io {
646646
647647 .netListenIp = netListenIp,
648648 .netListenUnix = netListenUnix,
649 .netAccept = netAccept,
650649 .netBindIp = netBindIp,
651650 .netConnectIp = netConnectIp,
652651 .netConnectUnix = netConnectUnix,
......@@ -1008,13 +1007,6 @@ fn netListenIp(
10081007 _ = options;
10091008 @panic("TODO");
10101009}
1011fn netAccept(userdata: ?*anyopaque, server: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket {
1012 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1013 _ = k;
1014 _ = server;
1015 _ = options;
1016 @panic("TODO");
1017}
10181010fn netBindIp(
10191011 userdata: ?*anyopaque,
10201012 address: *const net.IpAddress,
lib/std/Io/Threaded.zig+35-11
......@@ -1969,10 +1969,6 @@ pub fn io(t: *Threaded) Io {
19691969 .windows => netListenUnixWindows,
19701970 else => netListenUnixPosix,
19711971 },
1972 .netAccept = switch (native_os) {
1973 .windows => netAcceptWindows,
1974 else => netAcceptPosix,
1975 },
19761972 .netBindIp = switch (native_os) {
19771973 .windows => netBindIpWindows,
19781974 else => netBindIpPosix,
......@@ -2625,6 +2621,12 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper
26252621 break :o .{ null, sent };
26262622 },
26272623 },
2624 .net_accept => |o| return .{
2625 .net_accept = netAccept(t, o.socket_handle, o.options) catch |err| switch (err) {
2626 error.Canceled => |e| return e,
2627 else => |e| e,
2628 },
2629 },
26282630 .net_read => |o| return .{
26292631 .net_read = netRead(o.socket_handle, o.data, o.control) catch |err| switch (err) {
26302632 error.Canceled => |e| return e,
......@@ -2698,6 +2700,14 @@ fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void {
26982700 };
26992701 poll_len += 1;
27002702 },
2703 .net_accept => |o| {
2704 poll_buffer[poll_len] = .{
2705 .fd = o.socket_handle,
2706 .events = posix.POLL.IN | posix.POLL.ERR,
2707 .revents = 0,
2708 };
2709 poll_len += 1;
2710 },
27012711 .net_read => |o| {
27022712 poll_buffer[poll_len] = .{
27032713 .fd = o.socket_handle,
......@@ -2917,6 +2927,7 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout
29172927 storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
29182928 b.completed.tail = index;
29192929 },
2930 .net_accept => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR),
29202931 .net_read => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR),
29212932 .net_write => |o| try poll_storage.add(o.socket_handle, posix.POLL.OUT | posix.POLL.ERR),
29222933 }
......@@ -3115,6 +3126,7 @@ fn batchApc(
31153126 .device_io_control => .{ .device_io_control = iosb.* },
31163127 .net_receive => unreachable,
31173128 .net_send => unreachable,
3129 .net_accept => unreachable,
31183130 .net_read => unreachable,
31193131 .net_write => unreachable,
31203132 };
......@@ -3332,6 +3344,16 @@ fn batchDrainSubmittedWindows(t: *Threaded, b: *Io.Batch, concurrency: bool) (Io
33323344 .net_send = netSendWindows(t, o.socket_handle, o.messages, o.flags),
33333345 });
33343346 },
3347 .net_accept => |o| {
3348 // TODO integrate with overlapped I/O or equivalent to avoid this error
3349 if (concurrency) return error.ConcurrencyUnavailable;
3350 batchCompleteBlockingWindows(b, operation_userdata, .{
3351 .net_accept = netAccept(t, o.socket_handle, o.options) catch |err| switch (err) {
3352 error.Canceled => |e| return e,
3353 else => |e| e,
3354 },
3355 });
3356 },
33353357 .net_read => |*o| {
33363358 // TODO integrate with overlapped I/O or equivalent to avoid this error
33373359 if (concurrency) return error.ConcurrencyUnavailable;
......@@ -12813,11 +12835,15 @@ fn bindSocketUnixAfd(socket_handle: net.Socket.Handle, address: *const net.UnixA
1281312835 }
1281412836}
1281512837
12816fn netAcceptPosix(userdata: ?*anyopaque, listen_fd: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket {
12838fn netAccept(t: *Threaded, listen_fd: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket {
1281712839 if (!have_networking) return error.NetworkDown;
12818 const t: *Threaded = @ptrCast(@alignCast(userdata));
12819 _ = t;
12820 options;
12840 return if (is_windows)
12841 netAcceptWindows(t, listen_fd, options)
12842 else
12843 netAcceptPosix(listen_fd);
12844}
12845
12846fn netAcceptPosix(listen_fd: net.Socket.Handle) net.Server.AcceptError!net.Socket {
1282112847 var storage: PosixAddress = undefined;
1282212848 var addr_len: posix.socklen_t = @sizeOf(PosixAddress);
1282312849 const syscall: Syscall = try .start();
......@@ -12862,9 +12888,7 @@ fn netAcceptPosix(userdata: ?*anyopaque, listen_fd: net.Socket.Handle, options:
1286212888 return .{ .handle = fd, .address = addressFromPosix(&storage) };
1286312889}
1286412890
12865fn netAcceptWindows(userdata: ?*anyopaque, listen_handle: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket {
12866 if (!have_networking) return error.NetworkDown;
12867 const t: *Threaded = @ptrCast(@alignCast(userdata));
12891fn netAcceptWindows(t: *Threaded, listen_handle: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket {
1286812892 const Storage = extern struct {
1286912893 Info: windows.AFD.LISTEN_RESPONSE_INFO,
1287012894 RemoteAddress: extern union { posix: PosixAddress, unix: UnixAddress },
lib/std/Io/Uring.zig+6-13
......@@ -772,7 +772,6 @@ pub fn io(ev: *Evented) Io {
772772 .randomSecure = randomSecure,
773773
774774 .netListenIp = netListenIpUnavailable,
775 .netAccept = netAcceptUnavailable,
776775 .netBindIp = netBindIp,
777776 .netConnectIp = netConnectIpUnavailable,
778777 .netListenUnix = netListenUnixUnavailable,
......@@ -2111,6 +2110,12 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper
21112110 break :r .{ error.NetworkDown, 0 }; // TODO
21122111 },
21132112 },
2113 .net_accept => |o| .{
2114 .net_accept = r: {
2115 _ = o;
2116 break :r error.NetworkDown; // TODO
2117 },
2118 },
21142119 .net_read => |o| .{
21152120 .net_read = r: {
21162121 _ = o;
......@@ -4995,18 +5000,6 @@ fn netListenIpUnavailable(
49955000 return error.NetworkDown;
49965001}
49975002
4998fn netAcceptUnavailable(
4999 userdata: ?*anyopaque,
5000 listen_handle: net.Socket.Handle,
5001 options: net.Server.AcceptOptions,
5002) net.Server.AcceptError!net.Socket {
5003 const ev: *Evented = @ptrCast(@alignCast(userdata));
5004 _ = ev;
5005 _ = listen_handle;
5006 _ = options;
5007 return error.NetworkDown;
5008}
5009
50105003fn netBindIp(
50115004 userdata: ?*anyopaque,
50125005 address: *const net.IpAddress,
lib/std/Io/net.zig+5-25
......@@ -1556,30 +1556,7 @@ pub const Server = struct {
15561556 s.* = undefined;
15571557 }
15581558
1559 pub const AcceptError = error{
1560 /// The per-process limit on the number of open file descriptors has been reached.
1561 ProcessFdQuotaExceeded,
1562 /// The system-wide limit on the total number of open files has been reached.
1563 SystemFdQuotaExceeded,
1564 /// Not enough free memory. This often means that the memory allocation is limited
1565 /// by the socket buffer limits, not by the system memory.
1566 SystemResources,
1567 /// Either `listen` was never called, or `shutdown` was called (possibly while
1568 /// this call was blocking). This allows `shutdown` to be used as a concurrent
1569 /// cancellation mechanism.
1570 SocketNotListening,
1571 /// The network subsystem has failed.
1572 NetworkDown,
1573 /// No connection is already queued and ready to be accepted, and
1574 /// the socket is configured as non-blocking.
1575 WouldBlock,
1576 /// An incoming connection was indicated, but was subsequently terminated by the
1577 /// remote peer prior to accepting the call.
1578 ConnectionAborted,
1579 /// Firewall rules forbid connection.
1580 BlockedByFirewall,
1581 ProtocolFailure,
1582 } || Io.UnexpectedError || Io.Cancelable;
1559 pub const AcceptError = Io.Operation.NetAccept.Error || Io.Cancelable;
15831560
15841561 pub const AcceptOptions = switch (native_os) {
15851562 .windows => struct { mode: Socket.Mode, protocol: ?Protocol },
......@@ -1588,7 +1565,10 @@ pub const Server = struct {
15881565
15891566 /// Blocks until a client connects to the server.
15901567 pub fn accept(s: *Server, io: Io) AcceptError!Stream {
1591 return .{ .socket = try io.vtable.netAccept(io.userdata, s.socket.handle, s.options) };
1568 return .{ .socket = try (try io.operate(.{ .net_accept = .{
1569 .socket_handle = s.socket.handle,
1570 .options = s.options,
1571 } })).net_accept };
15921572 }
15931573};
15941574