authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-13 22:05:18-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-29 06:20:49-07:00
log0732ff22638a0e4b53bd2c42565165679674b3e1
tree3a33d53b091a711be52f53951d31f1d0e5a5d88d
parente8cea8accb7f289fd00ee82063df32882748ac48

std.Io.Threaded: implement connecting to unix sockets

3 files changed, 110 insertions(+), 90 deletions(-)

BRANCH_TODO+2-1
......@@ -5,11 +5,12 @@
55* fix Group.wait not handling cancelation (need to move impl of ResetEvent to Threaded)
66* implement cancelRequest for non-linux posix
77* finish converting all Threaded into directly calling system functions and handling EINTR
8* audit the TODOs
89
910* move max_iovecs_len to std.Io
1011* address the cancelation race condition (signal received between checkCancel and syscall)
1112* update signal values to be an enum
1213* move fs.File.Writer to Io
13* add non-blocking flag to network operations, handle EAGAIN
14* add non-blocking flag to net and fs operations, handle EAGAIN
1415* finish moving std.fs to Io
1516* finish moving all of std.posix into Threaded
lib/std/Io/Threaded.zig+79-87
......@@ -1697,34 +1697,10 @@ fn netListenIpPosix(
16971697) Io.net.IpAddress.ListenError!Io.net.Server {
16981698 const pool: *Pool = @ptrCast(@alignCast(userdata));
16991699 const family = posixAddressFamily(&address);
1700 const protocol: u32 = posix.IPPROTO.TCP;
1701 const socket_fd = while (true) {
1702 try pool.checkCancel();
1703 const flags: u32 = posix.SOCK.STREAM | if (socket_flags_unsupported) 0 else posix.SOCK.CLOEXEC;
1704 const socket_rc = posix.system.socket(family, flags, protocol);
1705 switch (posix.errno(socket_rc)) {
1706 .SUCCESS => {
1707 const fd: posix.fd_t = @intCast(socket_rc);
1708 errdefer posix.close(fd);
1709 if (socket_flags_unsupported) while (true) {
1710 try pool.checkCancel();
1711 switch (posix.errno(posix.system.fcntl(fd, posix.F.SETFD, @as(usize, posix.FD_CLOEXEC)))) {
1712 .SUCCESS => break,
1713 .INTR => continue,
1714 else => |err| return posix.unexpectedErrno(err),
1715 }
1716 };
1717 break fd;
1718 },
1719 .INTR => continue,
1720 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
1721 .MFILE => return error.ProcessFdQuotaExceeded,
1722 .NFILE => return error.SystemFdQuotaExceeded,
1723 .NOBUFS => return error.SystemResources,
1724 .NOMEM => return error.SystemResources,
1725 else => |err| return posix.unexpectedErrno(err),
1726 }
1727 };
1700 const socket_fd = try openSocketPosix(pool, family, .{
1701 .mode = options.mode,
1702 .protocol = options.protocol,
1703 });
17281704 errdefer posix.close(socket_fd);
17291705
17301706 if (options.reuse_address) {
......@@ -1763,53 +1739,49 @@ fn netListenUnix(
17631739) Io.net.UnixAddress.ListenError!Io.net.Socket.Handle {
17641740 if (!Io.net.has_unix_sockets) return error.AddressFamilyUnsupported;
17651741 const pool: *Pool = @ptrCast(@alignCast(userdata));
1766 const protocol: u32 = 0;
1767 const socket_fd = while (true) {
1768 try pool.checkCancel();
1769 const flags: u32 = posix.SOCK.STREAM | if (socket_flags_unsupported) 0 else posix.SOCK.CLOEXEC;
1770 const socket_rc = posix.system.socket(posix.AF.UNIX, flags, protocol);
1771 switch (posix.errno(socket_rc)) {
1772 .SUCCESS => {
1773 const fd: posix.fd_t = @intCast(socket_rc);
1774 errdefer posix.close(fd);
1775 if (socket_flags_unsupported) while (true) {
1776 try pool.checkCancel();
1777 switch (posix.errno(posix.system.fcntl(fd, posix.F.SETFD, @as(usize, posix.FD_CLOEXEC)))) {
1778 .SUCCESS => break,
1779 .INTR => continue,
1780 else => |err| return posix.unexpectedErrno(err),
1781 }
1782 };
1783 break fd;
1784 },
1785 .INTR => continue,
1786 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
1787 .MFILE => return error.ProcessFdQuotaExceeded,
1788 .NFILE => return error.SystemFdQuotaExceeded,
1789 .NOBUFS => return error.SystemResources,
1790 .NOMEM => return error.SystemResources,
1791 else => |err| return posix.unexpectedErrno(err),
1792 }
1742 const socket_fd = openSocketPosix(pool, posix.AF.UNIX, .{ .mode = .stream }) catch |err| switch (err) {
1743 error.ProtocolUnsupportedBySystem => return error.AddressFamilyUnsupported,
1744 error.ProtocolUnsupportedByAddressFamily => return error.AddressFamilyUnsupported,
1745 error.SocketModeUnsupported => return error.AddressFamilyUnsupported,
1746 else => |e| return e,
17931747 };
17941748 errdefer posix.close(socket_fd);
17951749
17961750 var storage: UnixAddress = undefined;
17971751 const addr_len = addressUnixToPosix(address, &storage);
1752 try posixBindUnix(pool, socket_fd, &storage.any, addr_len);
1753
17981754 while (true) {
17991755 try pool.checkCancel();
1800 switch (posix.errno(posix.system.bind(socket_fd, &storage.any, addr_len))) {
1756 switch (posix.errno(posix.system.listen(socket_fd, options.kernel_backlog))) {
1757 .SUCCESS => break,
1758 .ADDRINUSE => return error.AddressInUse,
1759 .BADF => |err| return errnoBug(err),
1760 else => |err| return posix.unexpectedErrno(err),
1761 }
1762 }
1763
1764 return socket_fd;
1765}
1766
1767fn posixBindUnix(pool: *Pool, fd: posix.socket_t, addr: *const posix.sockaddr, addr_len: posix.socklen_t) !void {
1768 while (true) {
1769 try pool.checkCancel();
1770 switch (posix.errno(posix.system.bind(fd, addr, addr_len))) {
18011771 .SUCCESS => break,
18021772 .INTR => continue,
18031773 .ACCES => return error.AccessDenied,
1804 .PERM => return error.PermissionDenied,
18051774 .ADDRINUSE => return error.AddressInUse,
18061775 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
18071776 .ADDRNOTAVAIL => return error.AddressUnavailable,
18081777 .NOMEM => return error.SystemResources,
1778
18091779 .LOOP => return error.SymLinkLoop,
18101780 .NOENT => return error.FileNotFound,
18111781 .NOTDIR => return error.NotDir,
18121782 .ROFS => return error.ReadOnlyFileSystem,
1783 .PERM => return error.PermissionDenied,
1784
18131785 .BADF => |err| return errnoBug(err), // always a race condition if this error is returned
18141786 .INVAL => |err| return errnoBug(err), // invalid parameters
18151787 .NOTSOCK => |err| return errnoBug(err), // invalid `sockfd`
......@@ -1818,18 +1790,6 @@ fn netListenUnix(
18181790 else => |err| return posix.unexpectedErrno(err),
18191791 }
18201792 }
1821
1822 while (true) {
1823 try pool.checkCancel();
1824 switch (posix.errno(posix.system.listen(socket_fd, options.kernel_backlog))) {
1825 .SUCCESS => break,
1826 .ADDRINUSE => return error.AddressInUse,
1827 .BADF => |err| return errnoBug(err),
1828 else => |err| return posix.unexpectedErrno(err),
1829 }
1830 }
1831
1832 return socket_fd;
18331793}
18341794
18351795fn posixBind(pool: *Pool, socket_fd: posix.socket_t, addr: *const posix.sockaddr, addr_len: posix.socklen_t) !void {
......@@ -1857,7 +1817,6 @@ fn posixConnect(pool: *Pool, socket_fd: posix.socket_t, addr: *const posix.socka
18571817 switch (posix.errno(posix.system.connect(socket_fd, addr, addr_len))) {
18581818 .SUCCESS => return,
18591819 .INTR => continue,
1860 .ADDRINUSE => return error.AddressInUse,
18611820 .ADDRNOTAVAIL => return error.AddressUnavailable,
18621821 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
18631822 .AGAIN, .INPROGRESS => return error.WouldBlock,
......@@ -1866,7 +1825,7 @@ fn posixConnect(pool: *Pool, socket_fd: posix.socket_t, addr: *const posix.socka
18661825 .CONNREFUSED => return error.ConnectionRefused,
18671826 .CONNRESET => return error.ConnectionResetByPeer,
18681827 .FAULT => |err| return errnoBug(err),
1869 .ISCONN => return error.AlreadyConnected,
1828 .ISCONN => |err| return errnoBug(err),
18701829 .HOSTUNREACH => return error.HostUnreachable,
18711830 .NETUNREACH => return error.NetworkUnreachable,
18721831 .NOTSOCK => |err| return errnoBug(err),
......@@ -1874,7 +1833,6 @@ fn posixConnect(pool: *Pool, socket_fd: posix.socket_t, addr: *const posix.socka
18741833 .TIMEDOUT => return error.ConnectionTimedOut,
18751834 .CONNABORTED => |err| return errnoBug(err),
18761835 .ACCES => return error.AccessDenied,
1877 // UNIX socket error codes:
18781836 .PERM => |err| return errnoBug(err),
18791837 .NOENT => |err| return errnoBug(err),
18801838 else => |err| return posix.unexpectedErrno(err),
......@@ -1882,6 +1840,35 @@ fn posixConnect(pool: *Pool, socket_fd: posix.socket_t, addr: *const posix.socka
18821840 }
18831841}
18841842
1843fn posixConnectUnix(pool: *Pool, fd: posix.socket_t, addr: *const posix.sockaddr, addr_len: posix.socklen_t) !void {
1844 while (true) {
1845 try pool.checkCancel();
1846 switch (posix.errno(posix.system.connect(fd, addr, addr_len))) {
1847 .SUCCESS => return,
1848 .INTR => continue,
1849
1850 .AFNOSUPPORT => return error.AddressFamilyUnsupported,
1851 .AGAIN => return error.WouldBlock,
1852 .INPROGRESS => return error.WouldBlock,
1853 .ACCES => return error.AccessDenied,
1854
1855 .LOOP => return error.SymLinkLoop,
1856 .NOENT => return error.FileNotFound,
1857 .NOTDIR => return error.NotDir,
1858 .ROFS => return error.ReadOnlyFileSystem,
1859 .PERM => return error.PermissionDenied,
1860
1861 .BADF => |err| return errnoBug(err),
1862 .CONNABORTED => |err| return errnoBug(err),
1863 .FAULT => |err| return errnoBug(err),
1864 .ISCONN => |err| return errnoBug(err),
1865 .NOTSOCK => |err| return errnoBug(err),
1866 .PROTOTYPE => |err| return errnoBug(err),
1867 else => |err| return posix.unexpectedErrno(err),
1868 }
1869 }
1870}
1871
18851872fn posixGetSockName(pool: *Pool, socket_fd: posix.fd_t, addr: *posix.sockaddr, addr_len: *posix.socklen_t) !void {
18861873 while (true) {
18871874 try pool.checkCancel();
......@@ -1926,6 +1913,7 @@ fn netConnectIpPosix(
19261913 .mode = options.mode,
19271914 .protocol = options.protocol,
19281915 });
1916 errdefer posix.close(socket_fd);
19291917 var storage: PosixAddress = undefined;
19301918 var addr_len = addressToPosix(address, &storage);
19311919 try posixConnect(pool, socket_fd, &storage.any, addr_len);
......@@ -1940,10 +1928,14 @@ fn netConnectUnix(
19401928 userdata: ?*anyopaque,
19411929 address: *const Io.net.UnixAddress,
19421930) Io.net.UnixAddress.ConnectError!Io.net.Socket.Handle {
1931 if (!Io.net.has_unix_sockets) return error.AddressFamilyUnsupported;
19431932 const pool: *Pool = @ptrCast(@alignCast(userdata));
1944 _ = pool;
1945 _ = address;
1946 @panic("TODO");
1933 const socket_fd = try openSocketPosix(pool, posix.AF.UNIX, .{ .mode = .stream });
1934 errdefer posix.close(socket_fd);
1935 var storage: UnixAddress = undefined;
1936 const addr_len = addressUnixToPosix(address, &storage);
1937 try posixConnectUnix(pool, socket_fd, &storage.any, addr_len);
1938 return socket_fd;
19471939}
19481940
19491941fn netBindIpPosix(
......@@ -2497,18 +2489,16 @@ fn netInterfaceNameResolve(
24972489 name: *const Io.net.Interface.Name,
24982490) Io.net.Interface.Name.ResolveError!Io.net.Interface {
24992491 const pool: *Pool = @ptrCast(@alignCast(userdata));
2500 try pool.checkCancel();
25012492
25022493 if (native_os == .linux) {
2503 const rc = posix.system.socket(posix.AF.UNIX, posix.SOCK.DGRAM | posix.SOCK.CLOEXEC, 0);
2504 const sock_fd: posix.fd_t = switch (posix.errno(rc)) {
2505 .SUCCESS => @intCast(rc),
2506 .ACCES => return error.AccessDenied,
2507 .MFILE => return error.SystemResources,
2508 .NFILE => return error.SystemResources,
2509 .NOBUFS => return error.SystemResources,
2510 .NOMEM => return error.SystemResources,
2511 else => |err| return posix.unexpectedErrno(err),
2494 const sock_fd = openSocketPosix(pool, posix.AF.UNIX, .{ .mode = .dgram }) catch |err| switch (err) {
2495 error.ProcessFdQuotaExceeded => return error.SystemResources,
2496 error.SystemFdQuotaExceeded => return error.SystemResources,
2497 error.AddressFamilyUnsupported => return error.Unexpected,
2498 error.ProtocolUnsupportedBySystem => return error.Unexpected,
2499 error.ProtocolUnsupportedByAddressFamily => return error.Unexpected,
2500 error.SocketModeUnsupported => return error.Unexpected,
2501 else => |e| return e,
25122502 };
25132503 defer posix.close(sock_fd);
25142504
......@@ -2521,12 +2511,12 @@ fn netInterfaceNameResolve(
25212511 try pool.checkCancel();
25222512 switch (posix.errno(posix.system.ioctl(sock_fd, posix.SIOCGIFINDEX, @intFromPtr(&ifr)))) {
25232513 .SUCCESS => return .{ .index = @bitCast(ifr.ifru.ivalue) },
2514 .INTR => continue,
25242515 .INVAL => |err| return errnoBug(err), // Bad parameters.
25252516 .NOTTY => |err| return errnoBug(err),
25262517 .NXIO => |err| return errnoBug(err),
25272518 .BADF => |err| return errnoBug(err), // Always a race condition.
25282519 .FAULT => |err| return errnoBug(err), // Bad pointer parameter.
2529 .INTR => continue,
25302520 .IO => |err| return errnoBug(err), // sock_fd is not a file descriptor
25312521 .NODEV => return error.InterfaceNotFound,
25322522 else => |err| return posix.unexpectedErrno(err),
......@@ -2535,12 +2525,14 @@ fn netInterfaceNameResolve(
25352525 }
25362526
25372527 if (native_os == .windows) {
2528 try pool.checkCancel();
25382529 const index = std.os.windows.ws2_32.if_nametoindex(&name.bytes);
25392530 if (index == 0) return error.InterfaceNotFound;
25402531 return .{ .index = index };
25412532 }
25422533
25432534 if (builtin.link_libc) {
2535 try pool.checkCancel();
25442536 const index = std.c.if_nametoindex(&name.bytes);
25452537 if (index == 0) return error.InterfaceNotFound;
25462538 return .{ .index = @bitCast(index) };
......@@ -2591,7 +2583,7 @@ fn addressFromPosix(posix_address: *PosixAddress) Io.net.IpAddress {
25912583 return switch (posix_address.any.family) {
25922584 posix.AF.INET => .{ .ip4 = address4FromPosix(&posix_address.in) },
25932585 posix.AF.INET6 => .{ .ip6 = address6FromPosix(&posix_address.in6) },
2594 else => unreachable,
2586 else => .{ .ip4 = .loopback(0) },
25952587 };
25962588}
25972589
lib/std/Io/net.zig+29-2
......@@ -206,6 +206,9 @@ pub const IpAddress = union(enum) {
206206 SystemFdQuotaExceeded,
207207 /// The requested address family (IPv4 or IPv6) is not supported by the operating system.
208208 AddressFamilyUnsupported,
209 ProtocolUnsupportedBySystem,
210 ProtocolUnsupportedByAddressFamily,
211 SocketModeUnsupported,
209212 } || Io.UnexpectedError || Io.Cancelable;
210213
211214 pub const ListenOptions = struct {
......@@ -216,6 +219,16 @@ pub const IpAddress = union(enum) {
216219 /// Sets SO_REUSEADDR and SO_REUSEPORT on POSIX.
217220 /// Sets SO_REUSEADDR on Windows, which is roughly equivalent.
218221 reuse_address: bool = false,
222 /// Only connection-oriented modes may be used here, which includes:
223 /// * `Socket.Mode.stream`
224 /// * `Socket.Mode.seqpacket`
225 mode: Socket.Mode = .stream,
226 /// Only connection-oriented protocols may be used here, which includes:
227 /// * `Protocol.tcp`
228 /// * `Protocol.tp`
229 /// * `Protocol.dccp`
230 /// * `Protocol.sctp`
231 protocol: Protocol = .tcp,
219232 };
220233
221234 /// Waits for a TCP connection. When using this API, `bind` does not need
......@@ -276,7 +289,6 @@ pub const IpAddress = union(enum) {
276289 ConnectionPending,
277290 ConnectionRefused,
278291 ConnectionResetByPeer,
279 AlreadyConnected,
280292 HostUnreachable,
281293 NetworkUnreachable,
282294 ConnectionTimedOut,
......@@ -842,7 +854,22 @@ pub const UnixAddress = struct {
842854 } };
843855 }
844856
845 pub const ConnectError = error{} || Io.Cancelable || Io.UnexpectedError;
857 pub const ConnectError = error{
858 SystemResources,
859 ProcessFdQuotaExceeded,
860 SystemFdQuotaExceeded,
861 AddressFamilyUnsupported,
862 ProtocolUnsupportedBySystem,
863 ProtocolUnsupportedByAddressFamily,
864 SocketModeUnsupported,
865 AccessDenied,
866 PermissionDenied,
867 SymLinkLoop,
868 FileNotFound,
869 NotDir,
870 ReadOnlyFileSystem,
871 WouldBlock,
872 } || Io.Cancelable || Io.UnexpectedError;
846873
847874 pub fn connect(ua: *const UnixAddress, io: Io) ConnectError!Stream {
848875 assert(ua.path.len <= max_len);