9p: use cloud9 protocol sessions and transports

1 files changed, 46 insertions, 184 deletions

diff --git a/src/9p_io.zig b/src/9p_io.zig
index 89611c2b..943561eb 100644
--- a/src/9p_io.zig
+++ b/src/9p_io.zig
@@ -2,6 +2,7 @@ const std = @import("std");
 const libc = std.c;
 const builtin = @import("builtin");
 const ninep = @import("9p.zig");
+const transport = @import("cloud9").transport;
 const pardes = @import("pardes.zig");
 const limits = @import("memory.zig").limits;
 pub const quic_enabled = @import("9p_options").quic;
@@ -93,21 +94,6 @@ fn canonicalIp(address: std.Io.net.IpAddress) std.Io.net.IpAddress {
     return address;
 }
 
-fn ipSockaddr(address: std.Io.net.IpAddress, out: *libc.sockaddr.storage) libc.socklen_t {
-    return switch (address) {
-        .ip4 => |ip| blk: {
-            const addr: *libc.sockaddr.in = @ptrCast(out);
-            addr.* = .{ .port = std.mem.nativeToBig(u16, ip.port), .addr = @bitCast(ip.bytes) };
-            break :blk @sizeOf(libc.sockaddr.in);
-        },
-        .ip6 => |ip| blk: {
-            const addr: *libc.sockaddr.in6 = @ptrCast(out);
-            addr.* = .{ .port = std.mem.nativeToBig(u16, ip.port), .addr = ip.bytes, .flowinfo = 0, .scope_id = 0 };
-            break :blk @sizeOf(libc.sockaddr.in6);
-        },
-    };
-}
-
 fn sockaddrIp(address: *const libc.sockaddr) ?std.Io.net.IpAddress {
     return switch (address.family) {
         libc.AF.INET => blk: {
@@ -207,25 +193,11 @@ pub const Listener = struct {
     }
 
     fn listenTcp(l: *Listener, address: std.Io.net.IpAddress) !void {
+        const fd = try transport.listenFd(.{ .tcp = address }, max_conns);
+        errdefer transport.close(fd);
         var addr: libc.sockaddr.storage = undefined;
-        const addr_len = ipSockaddr(address, &addr);
-        const fd = libc.socket(addr.family, libc.SOCK.STREAM, 0);
-        if (fd < 0) return error.SocketFailed;
-        errdefer _ = libc.close(fd);
-        setCloexec(fd);
-        setNonblock(fd);
-        const on: c_int = 1;
-        if (libc.setsockopt(fd, libc.SOL.SOCKET, libc.SO.REUSEADDR, &on, @sizeOf(c_int)) != 0)
-            return error.SocketOptionFailed;
-        if (address == .ip6) {
-            const v6only = if (darwin) 27 else std.os.linux.IPV6.V6ONLY;
-            if (libc.setsockopt(fd, libc.IPPROTO.IPV6, v6only, &on, @sizeOf(c_int)) != 0)
-                return error.SocketOptionFailed;
-        }
-        if (libc.bind(fd, @ptrCast(&addr), addr_len) != 0) return error.BindFailed;
-        if (libc.listen(fd, max_conns) != 0) return error.ListenFailed;
-        var actual_len: libc.socklen_t = @sizeOf(libc.sockaddr.storage);
-        if (libc.getsockname(fd, @ptrCast(&addr), &actual_len) != 0) return error.SocketAddressFailed;
+        var len: libc.socklen_t = @sizeOf(@TypeOf(addr));
+        if (libc.getsockname(fd, @ptrCast(&addr), &len) != 0) return error.SocketAddressFailed;
         l.tcp_address = sockaddrIp(@ptrCast(&addr)) orelse return error.SocketAddressFailed;
         l.tcp_fd = fd;
         log.info("serving 9P2000 over TCP on {f}", .{l.tcp_address.?});
@@ -236,26 +208,11 @@ pub const Listener = struct {
         for ([_]c_int{ l.fd, l.tcp_fd }) |listener_fd| {
             if (listener_fd < 0) continue;
             for (0..max_conns + 1) |_| {
-                const fd = libc.accept(listener_fd, null, null);
-                if (fd < 0) switch (libc.errno(fd)) {
-                    .AGAIN => break,
-                    .INTR, .CONNABORTED => continue,
-                    else => {
-                        l.paused_ms = nowMs() +| accept_pause_ms;
-                        log.warn("accept failed; pausing the listener for {d} ms", .{accept_pause_ms});
-                        return;
-                    },
-                };
-                setCloexec(fd);
-                setNonblock(fd);
-                if (listener_fd == l.tcp_fd) {
-                    const on: c_int = 1;
-                    _ = libc.setsockopt(fd, libc.IPPROTO.TCP, libc.TCP.NODELAY, &on, @sizeOf(c_int));
-                }
-                if (comptime darwin) {
-                    const on: c_int = 1;
-                    _ = libc.setsockopt(fd, libc.SOL.SOCKET, libc.SO.NOSIGPIPE, &on, @sizeOf(c_int));
-                }
+                const fd = (transport.acceptFd(listener_fd, listener_fd == l.tcp_fd) catch {
+                    l.paused_ms = nowMs() +| accept_pause_ms;
+                    log.warn("accept failed; pausing the listener for {d} ms", .{accept_pause_ms});
+                    return;
+                }) orelse break;
                 const c = for (&l.conns, 0..) |*cand, i| {
                     if (!l.live(@intCast(i)) and !cand.draining) break cand;
                 } else {
@@ -301,7 +258,7 @@ pub const Listener = struct {
         const now = nowMs();
         if (now == 0) return;
         for (&l.conns, 0..) |*c, i| {
-            if (!l.live(@intCast(i)) or c.srv.msize != 0) continue;
+            if (!l.live(@intCast(i)) or c.srv.protocol.msize != 0) continue;
             if (now - c.accepted_ms < greet_deadline_ms) continue;
             log.debug("slot {d} never sent Tversion; taking it back", .{i});
             l.drop(@intCast(i));
@@ -333,7 +290,7 @@ pub const Listener = struct {
             if (comptime quic_enabled) {
                 if (c.quic) |*connection| if (connection.pending()) return 0;
             }
-            if (c.srv.msize != 0) continue;
+            if (c.srv.protocol.msize != 0) continue;
             const at = c.accepted_ms + greet_deadline_ms;
             due = if (due) |d| @min(d, at) else at;
         }
@@ -341,30 +298,22 @@ pub const Listener = struct {
         return @intCast(@max(0, at - now));
     }
 
-    fn nowMs() i64 {
-        var ts: libc.timespec = undefined;
-        if (libc.clock_gettime(.MONOTONIC, &ts) != 0) return 0;
-        return @as(i64, ts.sec) * std.time.ms_per_s + @divTrunc(ts.nsec, std.time.ns_per_ms);
-    }
+    const nowMs = transport.nowMs;
 
     pub fn fill(l: *Listener, i: u8) void {
         if (comptime !supported) return;
         const c = &l.conns[i];
-        if (c.srv.dead) return l.drop(i);
-        const room = c.srv.in.len - c.srv.in_len;
+        if (c.srv.protocol.dead) return l.drop(i);
+        const room = c.srv.protocol.in.len - c.srv.protocol.in_len;
         if (room == 0) return;
         var buf: [msize]u8 = undefined;
-        const got: isize = if (quic_enabled and c.quic != null)
-            @intCast((c.quic.?.read(buf[0..@min(room, buf.len)]) catch return l.drop(i)) orelse return)
+        const got = if (quic_enabled and c.quic != null)
+            (c.quic.?.read(buf[0..@min(room, buf.len)]) catch return l.drop(i)) orelse return
         else
-            libc.read(c.fd, &buf, @min(room, buf.len));
+            (transport.read(c.fd, buf[0..@min(room, buf.len)]) catch return l.drop(i)) orelse return;
         if (got == 0) return l.drop(i);
-        if (got < 0) return switch (libc.errno(got)) {
-            .INTR, .AGAIN => {},
-            else => l.drop(i),
-        };
         const n = c.srv.push(buf[0..@intCast(got)]);
-        if (c.srv.dead) return l.drop(i);
+        if (c.srv.protocol.dead) return l.drop(i);
         std.debug.assert(n == @as(usize, @intCast(got)));
     }
 
@@ -375,15 +324,10 @@ pub const Listener = struct {
         while (true) {
             const bytes = c.srv.output();
             if (bytes.len == 0) return;
-            const n: isize = if (quic_enabled and c.quic != null)
-                @intCast(c.quic.?.write(bytes) catch return l.drop(i))
+            const n = if (quic_enabled and c.quic != null)
+                c.quic.?.write(bytes) catch return l.drop(i)
             else
-                libc.send(c.fd, bytes.ptr, bytes.len, nosignal);
-            if (n < 0) switch (libc.errno(n)) {
-                .INTR => continue,
-                .AGAIN => return,
-                else => return l.drop(i),
-            };
+                (transport.write(c.fd, bytes) catch return l.drop(i)) orelse return;
             if (n == 0) return;
             c.srv.wrote(@intCast(n));
         }
@@ -610,37 +554,29 @@ pub fn listen(gpa: std.mem.Allocator, named: []const u8, fallback: []const u8, t
         gpa.destroy(l);
         return null;
     };
-    var addr: libc.sockaddr.un = .{ .path = @splat(0) };
-    @memcpy(addr.path[0 .. p.len + 1], p[0 .. p.len + 1]);
-    const fd = libc.socket(libc.AF.UNIX, libc.SOCK.STREAM, 0);
-    if (fd < 0) {
-        gpa.destroy(l);
-        return null;
-    }
-    setCloexec(fd);
-    if (libc.bind(fd, @ptrCast(&addr), @sizeOf(@TypeOf(addr))) != 0) {
-        const bind_error = libc.errno(-1);
+    const fd = transport.listenFd(.{ .unix = p }, max_conns) catch |err| retry: {
         const io = std.Io.Threaded.global_single_threaded.io();
         const existing = std.Io.Dir.cwd().statFile(io, p, .{ .follow_symlinks = false }) catch null;
-        if (bind_error != .ADDRINUSE or existing == null or existing.?.kind != .unix_domain_socket or alive(p)) {
+        if (err != error.Bind or existing == null or existing.?.kind != .unix_domain_socket or alive(p)) {
             log.warn("something is already listening on {s}", .{p});
-            _ = libc.close(fd);
             gpa.destroy(l);
             return null;
         }
-        if (libc.unlink(p) != 0 or libc.bind(fd, @ptrCast(&addr), @sizeOf(@TypeOf(addr))) != 0) {
-            _ = libc.close(fd);
+        if (libc.unlink(p) != 0) {
             gpa.destroy(l);
             return null;
         }
-    }
-    if (libc.chmod(p, 0o600) != 0 or libc.listen(fd, max_conns) != 0) {
-        _ = libc.close(fd);
+        break :retry transport.listenFd(.{ .unix = p }, max_conns) catch {
+            gpa.destroy(l);
+            return null;
+        };
+    };
+    if (libc.chmod(p, 0o600) != 0) {
+        transport.close(fd);
         _ = libc.unlink(p);
         gpa.destroy(l);
         return null;
     }
-    setNonblock(fd);
     l.fd = fd;
     l.path_len = p.len;
     if (tcp_dial) |dial| {
@@ -687,18 +623,7 @@ pub fn listen(gpa: std.mem.Allocator, named: []const u8, fallback: []const u8, t
     return l;
 }
 
-fn alive(path: [:0]const u8) bool {
-    var addr: libc.sockaddr.un = .{ .path = @splat(0) };
-    if (path.len + 1 > sun_path_len) return true; // cannot ask; assume occupied
-    @memcpy(addr.path[0 .. path.len + 1], path[0 .. path.len + 1]);
-    const fd = libc.socket(libc.AF.UNIX, libc.SOCK.STREAM, 0);
-    if (fd < 0) return true;
-    defer _ = libc.close(fd);
-    setCloexec(fd);
-    setNonblock(fd);
-    if (libc.connect(fd, @ptrCast(&addr), @sizeOf(@TypeOf(addr))) == 0) return true;
-    return libc.errno(-1) != .CONNREFUSED;
-}
+const alive = transport.isListening;
 
 fn setNonblock(fd: c_int) void {
     const flags = libc.fcntl(fd, libc.F.GETFL, @as(c_int, 0));
@@ -708,8 +633,6 @@ fn setNonblock(fd: c_int) void {
     _ = libc.fcntl(fd, libc.F.SETFL, @as(c_int, @bitCast(@as(u32, @bitCast(o)))));
 }
 
-const nosignal: u32 = if (darwin) 0 else libc.MSG.NOSIGNAL;
-
 const testing = std.testing;
 
 test "the socket name is a third prefix in the shared directory" {
@@ -1115,15 +1038,10 @@ pub const Client = struct {
                         return;
                     }
                 }
-                var fds = [1]libc.pollfd{.{ .fd = s.fd, .events = events, .revents = 0 }};
-                const ready = libc.poll(&fds, 1, @intCast(@min(left, budget_ms)));
-                if (ready < 0) {
-                    if (libc.errno(ready) == .INTR) continue;
-                    return Error.Hangup;
-                }
-                if (ready == 0 or nowMs() >= s.deadline) return Error.Timeout;
-                if (fds[0].revents & events != 0) return;
-                return Error.Hangup;
+                return transport.wait(s.fd, events, s.deadline) catch |err| switch (err) {
+                    error.Timeout => Error.Timeout,
+                    else => Error.Hangup,
+                };
             }
         }
 
@@ -1143,12 +1061,7 @@ pub const Client = struct {
                     }
                 }
                 try s.wait(poll_out);
-                const sent = libc.send(s.fd, bytes.ptr, bytes.len, nosignal);
-                if (sent < 0) switch (libc.errno(sent)) {
-                    .INTR, .AGAIN => continue,
-                    else => return Error.Hangup,
-                };
-                if (sent == 0) return Error.Hangup;
+                const sent = (transport.write(s.fd, bytes) catch return Error.Hangup) orelse continue;
                 s.cl.wrote(@intCast(sent));
             }
         }
@@ -1174,12 +1087,8 @@ pub const Client = struct {
                     }
                 }
                 try s.wait(poll_in);
-                const got = libc.read(s.fd, &s.stage, @min(room, s.stage.len));
+                const got = (transport.read(s.fd, s.stage[0..@min(room, s.stage.len)]) catch return Error.Hangup) orelse continue;
                 if (got == 0) return Error.Hangup;
-                if (got < 0) switch (libc.errno(got)) {
-                    .INTR, .AGAIN => continue,
-                    else => return Error.Hangup,
-                };
                 const n = s.cl.push(s.stage[0..@intCast(got)]);
                 std.debug.assert(n == @as(usize, @intCast(got)));
             }
@@ -1317,62 +1226,15 @@ pub const Client = struct {
     }
 
     fn connect(sock: Dial, deadline: i64) Error!c_int {
-        var addr: libc.sockaddr.storage = undefined;
-        const addr_len: libc.socklen_t = switch (sock) {
-            .unix => |path| blk: {
-                if (path.len + 1 > sun_path_len) return Error.BadDial;
-                const un: *libc.sockaddr.un = @ptrCast(&addr);
-                un.* = .{ .path = @splat(0) };
-                @memcpy(un.path[0 .. path.len + 1], path[0 .. path.len + 1]);
-                break :blk @sizeOf(libc.sockaddr.un);
-            },
-            .tcp => |ip| ipSockaddr(ip, &addr),
+        const address: transport.Address = switch (sock) {
+            .unix => |path| .{ .unix = path },
+            .tcp => |ip| .{ .tcp = ip },
             .quic => return Error.QuicUnavailable,
         };
-        const fd = libc.socket(addr.family, libc.SOCK.STREAM, 0);
-        if (fd < 0) return Error.Dial;
-        setCloexec(fd);
-        setNonblock(fd);
-        errdefer _ = libc.close(fd);
-        if (sock == .tcp) {
-            const on: c_int = 1;
-            _ = libc.setsockopt(fd, libc.IPPROTO.TCP, libc.TCP.NODELAY, &on, @sizeOf(c_int));
-        }
-        while (true) {
-            if (libc.connect(fd, @ptrCast(&addr), addr_len) == 0) break;
-            switch (libc._errno().*) {
-                @intFromEnum(libc.E.AGAIN), @intFromEnum(libc.E.INTR) => {
-                    if (nowMs() >= deadline) return Error.Dial;
-                    nap(2);
-                },
-                @intFromEnum(libc.E.INPROGRESS), @intFromEnum(libc.E.ALREADY) => {
-                    const left = deadline - nowMs();
-                    if (left <= 0) return Error.Dial;
-                    var pfd: [1]libc.pollfd = .{.{ .fd = fd, .events = poll_out, .revents = 0 }};
-                    if (libc.poll(&pfd, 1, @intCast(@min(left, 1000))) <= 0) continue;
-                    var err: c_int = 0;
-                    var len: libc.socklen_t = @sizeOf(c_int);
-                    if (libc.getsockopt(fd, libc.SOL.SOCKET, libc.SO.ERROR, @ptrCast(&err), &len) != 0)
-                        return Error.Dial;
-                    if (err == 0) break;
-                    return Error.Dial;
-                },
-                @intFromEnum(libc.E.ISCONN) => break,
-                else => return Error.Dial,
-            }
-        }
-        if (comptime darwin) {
-            const on: c_int = 1;
-            _ = libc.setsockopt(fd, libc.SOL.SOCKET, libc.SO.NOSIGPIPE, &on, @sizeOf(c_int));
-        }
-        return fd;
+        return transport.connectFd(address, deadline) catch return Error.Dial;
     }
 
-    fn nowMs() i64 {
-        var ts: libc.timespec = undefined;
-        if (libc.clock_gettime(.MONOTONIC, &ts) != 0) return std.math.maxInt(i64);
-        return @as(i64, ts.sec) * std.time.ms_per_s + @divTrunc(ts.nsec, std.time.ns_per_ms);
-    }
+    const nowMs = transport.nowMs;
 
     const poll_in: i16 = @intCast(libc.POLL.IN);
     const poll_out: i16 = @intCast(libc.POLL.OUT);
@@ -1436,8 +1298,8 @@ pub const Client = struct {
         const transport_bytes = if (quic_enabled) @sizeOf(?quic.Connection) else 0;
         try testing.expectEqual(@as(usize, msize), @as(usize, (Session{ .fd = -1, .deadline = 0, .display_path = "" }).in.len));
         try testing.expect(transport_bytes <= 64);
-        try testing.expect(@sizeOf(Session) <= 3 * msize + 256 + transport_bytes);
-        try testing.expect(@sizeOf(ninep.Client) <= 256);
+        try testing.expect(@sizeOf(Session) <= 3 * msize + @sizeOf(ninep.Client) + 128 + transport_bytes);
+        try testing.expect(@sizeOf(ninep.Client) <= 512);
     }
 
     test "expired sessions do not send or consume buffered protocol work" {