diff options
Diffstat (limited to 'src/9p_io.zig')
| -rw-r--r-- | src/9p_io.zig | 40 |
1 files changed, 34 insertions, 6 deletions
diff --git a/src/9p_io.zig b/src/9p_io.zig index 2ca11a87..651b3d0a 100644 --- a/src/9p_io.zig +++ b/src/9p_io.zig @@ -64,7 +64,9 @@ const prefix = "pardes-9p-"; pub const msize = ninep.msize; -pub const max_conns = 4; +/// Scripts, a mount and an agent or two at once; each slot holds its +/// buffers (a few msize) whether used or not. +pub const max_conns = 16; extern "c" fn inet_pton(family: c_int, src: [*:0]const u8, dst: *anyopaque) c_int; @@ -176,6 +178,8 @@ pub const Listener = struct { stopping: std.atomic.Value(bool) = .init(false), /// A connection wrote the Restore and has yet to answer it. restore_writer: std.atomic.Value(bool) = .init(false), + /// Clients turned away with every slot taken, since the log last said so. + refused: std.atomic.Value(u32) = .init(0), /// Connections accepted so far, and the count each slot's connection /// was accepted at: a Restore cuts those accepted before it (`reset`), /// and a new client in a freed slot has a later stamp. Written on the @@ -217,6 +221,14 @@ pub const Listener = struct { /// answers. A request that would change a pane while the editor is out /// in a syscall mid-step is parked in the engine instead, and retried /// when the turn is next given up between steps (`wakeParked`). + /// On the accepting task, without the turn: counted, and logged with + /// the turn (`answerHeld`). + fn onRefused(ctx: ?*anyopaque) void { + const l = of(ctx); + _ = l.refused.fetchAdd(1, .acq_rel); + l.kick(); + } + fn onOpened(ctx: ?*anyopaque, conn: *Runner.Conn) void { const l = of(ctx); l.accepted[conn.index].store(l.accepts.fetchAdd(1, .acq_rel) + 1, .release); @@ -296,7 +308,15 @@ pub const Listener = struct { // Set with the turn still held, so `reset` sees it and waits. if (restoring) l.restore_writer.store(true, .release); defer if (restoring) l.restore_writer.store(false, .release); + core.fs.late_failure_len = 0; if (core.effects_len != 0) pardes.turn.awaitSettled(epoch); + // ponytail: one slot, so a failure of another client's effects that + // settle in the same wait is told to this write too. + if (l.core == core and core.fs.late_failure_len != 0) { + const failed = pardes.ctlfs.failText(req.tag, pardes.ctlfs.E.IO, core.fs.late_failure[0..core.fs.late_failure_len]); + pardes.ctlfs.events.noteError(core, req, failed); + return conn.reply(&failed, ""); + } // The Restore's own write is answered once it is done, and before // `reset` hangs every connection up, this one too, so the writer // hears that it happened rather than a cut; a failed one leaves the @@ -323,6 +343,11 @@ pub const Listener = struct { const l = of(ctx); if (l.stopping.load(.acquire)) return; const core = l.core; + const refused = l.refused.swap(0, .acq_rel); + if (refused != 0) { + var why: [64]u8 = undefined; + pardes.ctlfs.events.notePath(core, "err -", std.fmt.bufPrint(&why, "9p: too many connections ({d} turned away)", .{refused}) catch "9p: too many connections"); + } if (!core.fs.news) return; core.fs.news = false; for (&core.fs.opens) |*o| { @@ -816,7 +841,7 @@ pub fn listen(io: std.Io, gpa: std.mem.Allocator, core: *pardes.Pardes, named: [ l.runner.init(.{ .io = io, .root = pardes.ctlfs.root, - .handler = .{ .ctx = l, .serve = Listener.onServe, .opened = Listener.onOpened }, + .handler = .{ .ctx = l, .serve = Listener.onServe, .opened = Listener.onOpened, .refused = Listener.onRefused }, .greet_timeout_ms = Listener.greet_deadline_ms, }); const entry_name = if (named.len != 0) named else fallback; @@ -1278,7 +1303,8 @@ test "Unix TCP and QUIC share one listener through reads writes reconnects and r var directory: [64:0]u8 = undefined; _ = try std.fmt.bufPrintSentinel(&directory, "/tmp/pardes-tcp-XXXXXX", .{}, 0); if (mkdtemp(&directory) == null) return error.TempDirectoryFailed; - defer _ = rmdir(&directory); + // What the listener posted and left inside goes with it. + defer std.Io.Dir.cwd().deleteTree(testing.io, std.mem.sliceTo(&directory, 0)) catch |err| std.debug.print("deleteTree: {s}\n", .{@errorName(err)}); const old_runtime = if (libc.getenv("XDG_RUNTIME_DIR")) |v| try gpa.dupeZ(u8, std.mem.span(v)) else null; defer { if (old_runtime) |v| { @@ -1348,7 +1374,7 @@ test "Unix TCP and QUIC share one listener through reads writes reconnects and r } try testing.expect(l.path().len != 0); try testing.expectEqual(tcp, l.tcp_address != null); - try testing.expectEqual(@as(usize, 4), l.runner.conns.len); + try testing.expectEqual(@as(usize, max_conns), l.runner.conns.len); try testing.expect(l.watcher == null); const port = (if (tcp) l.tcp_address else l.quic_address).?.getPort(); try testing.expect(port != 0); @@ -1389,7 +1415,8 @@ test "a change waits while the editor is out mid-step, a read does not, and the var directory: [64:0]u8 = undefined; _ = try std.fmt.bufPrintSentinel(&directory, "/tmp/pardes-park-XXXXXX", .{}, 0); if (mkdtemp(&directory) == null) return error.TempDirectoryFailed; - defer _ = rmdir(&directory); + // What the listener posted and left inside goes with it. + defer std.Io.Dir.cwd().deleteTree(testing.io, std.mem.sliceTo(&directory, 0)) catch |err| std.debug.print("deleteTree: {s}\n", .{@errorName(err)}); const old_runtime = if (libc.getenv("XDG_RUNTIME_DIR")) |v| try gpa.dupeZ(u8, std.mem.span(v)) else null; defer { if (old_runtime) |v| { @@ -1458,7 +1485,8 @@ test "a held read is answered when the log has news, and a flushed one spends no var directory: [64:0]u8 = undefined; _ = try std.fmt.bufPrintSentinel(&directory, "/tmp/pardes-held-XXXXXX", .{}, 0); if (mkdtemp(&directory) == null) return error.TempDirectoryFailed; - defer _ = rmdir(&directory); + // What the listener posted and left inside goes with it. + defer std.Io.Dir.cwd().deleteTree(testing.io, std.mem.sliceTo(&directory, 0)) catch |err| std.debug.print("deleteTree: {s}\n", .{@errorName(err)}); const old_runtime = if (libc.getenv("XDG_RUNTIME_DIR")) |v| try gpa.dupeZ(u8, std.mem.span(v)) else null; defer { if (old_runtime) |v| { |