std.os: add support for socketpair

lib/std/c.zig

@@ -77,7 +77,7 @@ pub usingnamespace switch (builtin.os.tag) {
         pub extern "c" fn sigfillset(set: ?*c.sigset_t) void;
         pub extern "c" fn sigwait(set: ?*c.sigset_t, sig: ?*c_int) c_int;
 
-        pub extern "c" fn socket(domain: c_uint, sock_type: c_uint, protocol: c_uint) c_int;
+        pub extern "c" fn socket(domain: c_int, sock_type: c_int, protocol: c_int) c_int;
 
         pub extern "c" fn stat(noalias path: [*:0]const u8, noalias buf: *c.Stat) c_int;
 
@@ -185,7 +185,7 @@ pub extern "c" fn uname(buf: *c.utsname) c_int;
 pub extern "c" fn gethostname(name: [*]u8, len: usize) c_int;
 pub extern "c" fn shutdown(socket: c.fd_t, how: c_int) c_int;
 pub extern "c" fn bind(socket: c.fd_t, address: ?*const c.sockaddr, address_len: c.socklen_t) c_int;
-pub extern "c" fn socketpair(domain: c_uint, sock_type: c_uint, protocol: c_uint, sv: *[2]c.fd_t) c_int;
+pub extern "c" fn socketpair(domain: c_int, sock_type: c_int, protocol: c_int, sv: *[2]c.fd_t) c_int;
 pub extern "c" fn listen(sockfd: c.fd_t, backlog: c_uint) c_int;
 pub extern "c" fn getsockname(sockfd: c.fd_t, noalias addr: *c.sockaddr, noalias addrlen: *c.socklen_t) c_int;
 pub extern "c" fn getpeername(sockfd: c.fd_t, noalias addr: *c.sockaddr, noalias addrlen: *c.socklen_t) c_int;

lib/std/net.zig

@@ -1928,7 +1928,7 @@ pub const StreamServer = struct {
     pub fn listen(self: *StreamServer, address: Address) !void {
         const nonblock = if (std.io.is_async) os.SOCK.NONBLOCK else 0;
         const sock_flags = os.SOCK.STREAM | os.SOCK.CLOEXEC | nonblock;
-        var use_sock_flags: u32 = sock_flags;
+        var use_sock_flags: i32 = sock_flags;
         if (self.force_nonblocking) use_sock_flags |= os.SOCK.NONBLOCK;
         const proto = if (address.any.family == os.AF.UNIX) @as(u32, 0) else os.IPPROTO.TCP;
 

lib/std/os.zig

@@ -3377,7 +3377,7 @@ pub const SocketError = error{
     SocketTypeNotSupported,
 } || UnexpectedError;
 
-pub fn socket(domain: u32, socket_type: u32, protocol: u32) SocketError!socket_t {
+pub fn socket(domain: i32, socket_type: i32, protocol: i32) SocketError!socket_t {
     if (builtin.os.tag == .windows) {
         // NOTE: windows translates the SOCK.NONBLOCK/SOCK.CLOEXEC flags into
         // windows-analagous operations
@@ -6671,7 +6671,8 @@ pub fn recvfrom(
     addrlen: ?*socklen_t,
 ) RecvFromError!usize {
     while (true) {
-        const rc = system.recvfrom(sockfd, buf.ptr, buf.len, flags, src_addr, addrlen);
+        const sys = if (builtin.os.tag == .windows) windows else system;
+        const rc = sys.recvfrom(sockfd, buf.ptr, buf.len, flags, src_addr, addrlen);
         if (builtin.os.tag == .windows) {
             if (rc == windows.ws2_32.SOCKET_ERROR) {
                 switch (windows.ws2_32.WSAGetLastError()) {
@@ -7393,3 +7394,29 @@ pub fn ptrace(request: u32, pid: pid_t, addr: usize, signal: usize) PtraceError!
 }
 
 const lfs64_abi = builtin.os.tag == .linux and builtin.link_libc and builtin.abi.isGnu();
+
+const SocketpairError = SocketError || error{OperationNotSupported};
+
+/// On platforms not supporting socketpair (Windows), socketpair is emulated.
+/// On Windows, only `AF_UNIX` and `SOCK_STREAM` are supported.
+pub fn socketpair(domain: i32, sock_type: i32, protocol: i32) SocketpairError![2]socket_t {
+    var pair: [2]socket_t = undefined;
+
+    if (builtin.os.tag == .windows) {
+        @compileError("socketpair is unsupported on Windows");
+    }
+
+    return switch (system.getErrno(system.socketpair(domain, sock_type, protocol, &pair))) {
+        E.SUCCESS => pair,
+        E.AFNOSUPPORT => error.AddressFamilyNotSupported,
+        E.MFILE => error.ProcessFdQuotaExceeded,
+        E.NFILE => error.SystemFdQuotaExceeded,
+        E.OPNOTSUPP => error.OperationNotSupported,
+        E.PROTONOSUPPORT => error.ProtocolNotSupported,
+        E.PROTOTYPE => error.SocketTypeNotSupported,
+        E.NOBUFS => error.SystemResources,
+        E.NOMEM => error.SystemResources,
+        E.FAULT => unreachable, // we own `pair`
+        else => error.Unexpected,
+    };
+}

lib/std/os/linux.zig

@@ -1268,11 +1268,11 @@ pub fn getpeername(fd: i32, noalias addr: *sockaddr, noalias len: *socklen_t) us
     return syscall3(.getpeername, @as(usize, @bitCast(@as(isize, fd))), @intFromPtr(addr), @intFromPtr(len));
 }
 
-pub fn socket(domain: u32, socket_type: u32, protocol: u32) usize {
+pub fn socket(domain: i32, socket_type: i32, protocol: i32) usize {
     if (native_arch == .x86) {
         return socketcall(SC.socket, &[3]usize{ domain, socket_type, protocol });
     }
-    return syscall3(.socket, domain, socket_type, protocol);
+    return syscall3(.socket, @intCast(domain), @intCast(socket_type), @intCast(protocol));
 }
 
 pub fn setsockopt(fd: i32, level: u32, optname: u32, optval: [*]const u8, optlen: socklen_t) usize {

lib/std/os/test.zig

@@ -1230,3 +1230,62 @@ test "fchmodat smoke test" {
     const st = try os.fstatat(tmp.dir.fd, "foo.txt", 0);
     try expectEqual(@as(os.mode_t, 0o755), st.mode & 0b111_111_111);
 }
+
+fn writeSocketTest(socket: os.socket_t, err: *?anyerror) void {
+    const buf = "Hello, World!";
+    var i: usize = 0;
+    while (i < buf.len) {
+        const nb_sent = std.os.send(socket, buf[i..], 0) catch |e| {
+            err.* = e;
+            return;
+        };
+        i += nb_sent;
+    }
+}
+
+fn readSocketTest(socket: os.socket_t, err: *?anyerror) void {
+    var buf: [13]u8 = undefined;
+    var i: usize = 0;
+
+    while (i < buf.len) {
+        const nb_recved = std.os.recv(socket, buf[i..], 0) catch |e| {
+            err.* = e;
+            return;
+        };
+        i += nb_recved;
+    }
+
+    testing.expectEqualSlices(u8, &buf, "Hello, World!") catch |e| {
+        err.* = e;
+    };
+}
+
+test "socketpair" {
+    if (builtin.single_threaded) return error.SkipZigTest;
+
+    const pair = try os.socketpair(os.AF.UNIX, os.SOCK.STREAM, 0);
+    defer os.close(pair[0]);
+    defer os.close(pair[1]);
+
+    const ally = testing.allocator;
+
+    var wr_err: ?anyerror = null;
+    const writer = try std.Thread.spawn(
+        .{ .allocator = ally },
+        writeSocketTest,
+        .{ pair[0], &wr_err },
+    );
+
+    var rd_err: ?anyerror = null;
+    const reader = try std.Thread.spawn(
+        .{ .allocator = ally },
+        readSocketTest,
+        .{ pair[1], &rd_err },
+    );
+
+    writer.join();
+    if (wr_err) |e| return e;
+
+    reader.join();
+    if (rd_err) |e| return e;
+}

lib/std/os/windows/kernel32.zig

@@ -451,3 +451,5 @@ pub extern "kernel32" fn RegOpenKeyExW(
 ) callconv(WINAPI) LSTATUS;
 
 pub extern "kernel32" fn GetPhysicallyInstalledSystemMemory(TotalMemoryInKilobytes: *ULONGLONG) BOOL;
+
+pub extern "kernel32" fn GetTempPathA(nBufferLength: DWORD, lpBuffer: LPSTR) callconv(WINAPI) DWORD;