diff --git a/docs/reference/cli/native.md b/docs/reference/cli/native.md index 26ef93c..53175dd 100644 --- a/docs/reference/cli/native.md +++ b/docs/reference/cli/native.md @@ -9,6 +9,13 @@ The CLI uses `HOLLOW_COMMAND_ADDR` when available. Otherwise it discovers the running host through `%LOCALAPPDATA%\hollow\command-ipc-address`, so a Windows CLI invoked from an unrelated shell or from WSL works without inherited state. +The command server binds only to loopback addresses. It handles up to eight +connections concurrently, while application mutations remain serialized on the +frame thread. Incomplete frames are disconnected after a total five-second +operation deadline, including on Windows. Native client timeouts also cover +connection establishment. Loopback binding does not authenticate local users; +protect access to the local session accordingly. + ## Running it ```bash diff --git a/docs/reference/lua/process.md b/docs/reference/lua/process.md index a3e9b78..318b05d 100644 --- a/docs/reference/lua/process.md +++ b/docs/reference/lua/process.md @@ -1,69 +1,75 @@ # `hollow.process` -Run child processes from Lua. -The simple tuple helpers are the recommended path today; -`spawn` and `exec` are placeholders for a future richer API. +Run host processes from Lua. Prefer `spawn` or `exec` in interactive callbacks; +the older `run` and `run_child_process` helpers wait synchronously. -## Functions +## Asynchronous processes ```lua -hollow.process.run_child_process(args, opts?) -- returns (ok, stdout, stderr) -hollow.process.run(cmd, args?) -- returns { code, stdout, stderr } -hollow.term.run_domain_process(args, domain?, opts?) -- runs through a domain shell -``` +local job = hollow.process.spawn({ + cmd = { "git", "status", "--short" }, + cwd = "/path/to/project", + env = { GIT_OPTIONAL_LOCKS = "0" }, + timeout_ms = 30000, + output_limit = 1024 * 1024, + on_complete = function(result) + print(result.code, result.stdout, result.stderr, result.error) + end, +}) -`run_child_process` is the WezTerm-style tuple helper. -`run` returns a structured table with `code`, `stdout`, `stderr`. -`run_domain_process` resolves the configured domain shell and runs -the argv through it; if `domain` is omitted it uses the current -pane's domain. +job:status() -- "running" or "finished" (completion delivered) +job:result() -- nil until completion, then the result table +job:cancel() -- request cancellation; job:kill() is an alias +job:next(function(result) print(result.code) end) +``` -## `opts` +`cmd` is an argv array, or a single executable name. Strings are not parsed as +shell commands; pass a shell explicitly when required. `env` overrides inherited +variables. `cwd` defaults to the application's working directory. -```lua -{ - hide_window = true, -- default true; suppresses a console window on Windows -} -``` +Up to 16 jobs may be outstanding per Lua runtime. Output is collected separately +for stdout and stderr, with a default limit of 1 MiB each and maximum of 16 MiB. +Exceeding a limit terminates the job with an error. The default timeout is 30 seconds, +with a maximum of 24 hours. Both limits must be positive integers. -## Examples +Completion is polled by a deferred callback every 10 ms; delivery depends on the +application tick. `on_complete` runs on the Lua callback thread, never in a worker. +Config reload and runtime shutdown cancel outstanding jobs and release their resources. +Cancellation applies to the direct child; it does not promise process-tree termination. -```lua -local ok, out, err = hollow.process.run_child_process({ - "git", "rev-parse", "--show-toplevel", -}) +Results contain `code`, `stdout`, `stderr`, optional `error`, and `canceled`. +A nonzero process exit is a normal result. Spawn failures, timeouts, output-limit +errors, and cancellation use `code = -1` and an `error` string. Output may be absent +on these failures. This API collects output at completion; it does not expose +streaming readers, stdin writers, or a process PID. -if ok then - print("top-level:", out) -else - print("git failed:", err) -end -``` +## Promises and coroutines -Run a process through the active pane's domain: +`exec(opts)` returns a promise. It resolves with the result even for a nonzero +exit code; infrastructure failures reject with a result table (or a validation +error string). `spawn` supports cancellation through its returned handle. ```lua -local ok, out, err = hollow.term.run_domain_process({ - "ls", "-la", -}) +hollow.process.exec({ cmd = { "git", "branch", "--show-current" } }) + :next(function(result) print(result.stdout) end) + :catch(function(err) print("process failed", err) end) + +hollow.async.run(function() + local job = hollow.process.spawn({ cmd = { "git", "status", "--short" } }) + local result = job:wait() -- yields the coroutine + print(result.stdout) +end) ``` -Run a process through a specific domain: +## Synchronous compatibility helpers ```lua -local ok, out, err = hollow.term.run_domain_process({ - "uname", "-a", -}, "UbuntuWSL") +hollow.process.run_child_process(args, opts?) -- (ok, stdout, stderr) +hollow.process.run(cmd, args?) -- { code, stdout, stderr } +hollow.term.run_domain_process(args, domain?, opts?) -- through a domain shell ``` -## Placeholders - -`hollow.process.spawn(opts)` and `hollow.process.exec(opts)` are -declared in the API surface but are not implemented yet. -Use the helpers above for now. - -## See also - -- [`hollow.term.run_domain_process`](term.md#run-a-process-in-a-domain) -- [`hollow.fs`](fs.md) — filesystem helpers -- [Plugins](../../plugins.md) — uses `hollow.process.run` to clone repos +`opts.hide_window` defaults to true on Windows. These helpers retain their existing +50 KiB per-stream output limits. `run_domain_process` uses the active pane's domain +when none is supplied. Asynchronous jobs currently run on the host; pass an explicit +WSL or SSH executable when required. diff --git a/src/app.zig b/src/app.zig index f9a4943..8345aca 100644 --- a/src/app.zig +++ b/src/app.zig @@ -57,6 +57,7 @@ const quick_select = @import("app/quick_select.zig"); const htp = @import("app/htp.zig"); const HtpCodec = @import("htp/codec.zig").Codec; const input = @import("app/input.zig"); +const PtyBudget = @import("app/pty_budget.zig").Budget; const ActionQueue = @import("app/action_queue.zig").ActionQueue; const Lifecycle = @import("app/lifecycle.zig").Lifecycle; const hyperlinks = @import("app/hyperlinks.zig"); @@ -387,6 +388,7 @@ pub const App = struct { command_ready: std.Io.Condition = .init, command_done: std.Io.Condition = .init, pending_command: ?*cmd_ipc.PendingCommandRequest = null, + command_shutting_down: bool = false, automation_mutex: std.Io.Mutex = .init, automation_changed: std.Io.Condition = .init, automation_revision: u64 = 1, @@ -934,6 +936,7 @@ pub const App = struct { self.automation_changed.broadcast(io.get()); self.automation_mutex.unlock(io.get()); + cmd_ipc.shutdownPendingCommands(self); if (self.command_ipc_server) |*server| { server.deinit(); self.command_ipc_server = null; @@ -2228,19 +2231,25 @@ pub const App = struct { var next_idle_render_poll_ns: i128 = 0; if (self.mux) |*mux| { const active_pane = mux.activePane(); - var inactive_panes_remaining: usize = 0; + var visible_buf: [MAX_LAYOUT_LEAVES]LayoutLeaf = undefined; + const visible = self.computeActiveLayout(&visible_buf); + const now = io.nanoTimestamp(); + const interacting = (self.last_input_activity_ns != 0 and now - self.last_input_activity_ns < PTY_RECENT_ACTIVITY_NS) or + (self.last_resize_activity_ns != 0 and now - self.last_resize_activity_ns < PTY_RECENT_ACTIVITY_NS) or self.previous_frame_slow; + var visible_budget = PtyBudget{ .bytes = 256 * 1024, .nanoseconds = if (interacting) 500_000 else 2_000_000, .panes = 0 }; + var hidden_budget = PtyBudget{ + .bytes = if (interacting) (if (self.frame_count % 2 == 0) @as(usize, 16 * 1024) else 0) else 256 * 1024, + .nanoseconds = if (interacting) 500_000 else 1_000_000, + .panes = 0, + }; var count_panes = mux.paneIterator(); while (count_panes.next()) |pane| { - if (pane != active_pane) inactive_panes_remaining += 1; + if (pane == active_pane) continue; + const is_visible = for (visible) |leaf| { + if (leaf.pane == pane) break true; + } else false; + if (is_visible) visible_budget.panes += 1 else hidden_budget.panes += 1; } - // Heavy background output otherwise rebuilds and redraws terminal - // state every frame. Leave alternating frames free for input and - // active-pane rendering. - var inactive_pty_budget: usize = if (self.frame_count % 2 == 0) 16 * 1024 else 0; - const inactive_deadline_ns: i128 = if (inactive_pty_budget > 0) - io.nanoTimestamp() + PTY_INTERACTIVE_BUDGET_NS - else - 0; var panes = mux.paneIteratorActiveFirst(); var pane_idx: usize = 0; var total_pty_read_ns: i128 = 0; @@ -2258,18 +2267,12 @@ pub const App = struct { while (panes.next()) |pane| { const pane_is_active = active_pane == pane; const active_screen_before = pane.active_screen; - const pty_read_loops: usize = if (pane_is_active) - (pane_mod.PTY_HARD_BYTE_LIMIT + pane_mod.PTY_PARSE_CHUNK_BYTES - 1) / pane_mod.PTY_PARSE_CHUNK_BYTES - else - 2; - // Inactive panes share one frame budget so work remains bounded - // regardless of pane count. Idle panes donate quota to later panes. - const pty_read_bytes: usize = if (pane_is_active) - pane_mod.PTY_HARD_BYTE_LIMIT - else if (inactive_panes_remaining > 0 and inactive_pty_budget > 0) - (inactive_pty_budget + inactive_panes_remaining - 1) / inactive_panes_remaining - else - 0; + const pane_is_visible = for (visible) |leaf| { + if (leaf.pane == pane) break true; + } else false; + const background_budget = if (pane_is_visible) &visible_budget else &hidden_budget; + const pty_read_loops: usize = (pane_mod.PTY_HARD_BYTE_LIMIT + pane_mod.PTY_PARSE_CHUNK_BYTES - 1) / pane_mod.PTY_PARSE_CHUNK_BYTES; + const pty_read_bytes: usize = if (pane_is_active) pane_mod.PTY_HARD_BYTE_LIMIT else background_budget.byteQuota(); const pending_output_bytes: usize = if (pane_is_active) blk: { const deferred_output_bytes = pane.boot_output.items.len +| pane.pending_terminal_inject.items.len; break :blk pane.pendingPtyOutputBytes() +| deferred_output_bytes; @@ -2286,17 +2289,17 @@ pub const App = struct { pane_mod.PTY_PARSE_CHUNK_BYTES; const pty_budget_ns: i128 = if (pane_is_active) selectPtyBudgetNs(pty_now_ns, self.last_input_activity_ns, self.last_resize_activity_ns, pane.last_pty_output_ns, self.previous_frame_slow, pending_output_bytes) - else if (inactive_deadline_ns != 0 and io.nanoTimestamp() < inactive_deadline_ns) - inactive_deadline_ns - io.nanoTimestamp() + else if (pty_read_bytes > 0) + background_budget.timeQuota() else 0; + const poll_start_ns = io.nanoTimestamp(); const pty_bytes_read = pane.pollPty(runtime, pty_read_loops, pty_read_bytes, pty_budget_ns, pty_parse_chunk_bytes, self.config.debug_overlay) catch |err| result: { std.log.err("pane pollPty error: {s}", .{@errorName(err)}); break :result 0; }; if (!pane_is_active) { - inactive_pty_budget -|= pty_bytes_read; - inactive_panes_remaining -= 1; + background_budget.consume(pty_bytes_read, io.nanoTimestamp() - poll_start_ns); } const sync_now_ns = io.nanoTimestamp(); const sync_mode_active = self.config.synchronized_output and @@ -2780,13 +2783,13 @@ test "PTY budget adapts to interaction, backlog, pressure, and slow frames" { test "jsonObjectIndex accepts non-negative integers and whole floats" { var object = try std.json.ObjectMap.init(std.testing.allocator, &.{}, &.{}); - defer object.deinit(); + defer object.deinit(std.testing.allocator); - try object.put("int", .{ .integer = 7 }); - try object.put("float", .{ .float = 3.0 }); - try object.put("negative", .{ .integer = -1 }); - try object.put("fraction", .{ .float = 2.5 }); - try object.put("text", .{ .string = "4" }); + try object.put(std.testing.allocator, "int", .{ .integer = 7 }); + try object.put(std.testing.allocator, "float", .{ .float = 3.0 }); + try object.put(std.testing.allocator, "negative", .{ .integer = -1 }); + try object.put(std.testing.allocator, "fraction", .{ .float = 2.5 }); + try object.put(std.testing.allocator, "text", .{ .string = "4" }); try std.testing.expectEqual(@as(?usize, 7), jsonObjectIndex(object, "int")); try std.testing.expectEqual(@as(?usize, 3), jsonObjectIndex(object, "float")); @@ -2806,7 +2809,7 @@ test "cloneJsonValue deep copies nested JSON values" { var nested = std.json.Array.init(std.testing.allocator); try nested.append(.{ .string = try std.testing.allocator.dupe(u8, "alpha") }); try nested.append(.{ .integer = 9 }); - try source.put(try std.testing.allocator.dupe(u8, "list"), .{ .array = nested }); + try source.put(std.testing.allocator, try std.testing.allocator.dupe(u8, "list"), .{ .array = nested }); const clone = try cloneJsonValue(std.testing.allocator, .{ .object = source }); defer deinitJsonValue(std.testing.allocator, clone); diff --git a/src/app/command_dispatcher.zig b/src/app/command_dispatcher.zig index 0969763..3ec10ed 100644 --- a/src/app/command_dispatcher.zig +++ b/src/app/command_dispatcher.zig @@ -154,7 +154,7 @@ pub fn drainPendingCommand(self: *App) void { std.log.info("command-ipc: dispatch_ms={d:.3} kind={s}", .{ elapsedMs(start_ns), @tagName(pending.request.kind) }); } pending.done = true; - self.command_done.signal(io.get()); + self.command_done.broadcast(io.get()); } pub fn runCommandSync(self: *App, request: command_mod.Request) command_mod.Response { @@ -163,10 +163,12 @@ pub fn runCommandSync(self: *App, request: command_mod.Request) command_mod.Resp self.command_mutex.lockUncancelable(io.get()); defer self.command_mutex.unlock(io.get()); - while (self.pending_command != null) { + while (self.pending_command != null and !self.command_shutting_down) { self.command_done.waitUncancelable(io.get(), &self.command_mutex); } + if (self.command_shutting_down) return command_mod.Response.fail("shutting_down", "Hollow is shutting down"); + self.pending_command = &pending; self.signalWake(); self.command_ready.signal(io.get()); @@ -978,3 +980,43 @@ fn execEmit(self: *App, request: command_mod.Request) command_mod.Response { if (!result.success) return command_mod.Response.fail("error", result.error_message orelse "htp emit failed"); return okNull(); } + +/// Wake every connection waiting for the frame thread before joining workers. +pub fn shutdownPendingCommands(self: *App) void { + self.command_mutex.lockUncancelable(io.get()); + defer self.command_mutex.unlock(io.get()); + self.command_shutting_down = true; + if (self.pending_command) |pending| { + if (!pending.done) { + pending.response = command_mod.Response.fail("shutting_down", "Hollow is shutting down"); + pending.done = true; + } + } + self.command_done.broadcast(io.get()); +} + +test "shutdown releases commands waiting for the frame thread" { + const app = try std.testing.allocator.create(App); + defer std.testing.allocator.destroy(app); + app.* = App.init(std.testing.allocator); + defer app.deinit(); + const Waiter = struct { + fn run(target: *App) void { + var response = runCommandSync(target, .{ .kind = .get_revision }); + defer response.deinit(target.allocator); + std.debug.assert(!response.success); + std.debug.assert(std.mem.eql(u8, response.status, "shutting_down")); + } + }; + const worker = try std.Thread.spawn(.{}, Waiter.run, .{app}); + defer worker.join(); + defer shutdownPendingCommands(app); + { + app.command_mutex.lockUncancelable(io.get()); + defer app.command_mutex.unlock(io.get()); + while (app.pending_command == null) { + try io.waitTimeout(&app.command_ready, &app.command_mutex, std.time.ns_per_s); + } + } + // The defer order cancels the pending request before joining the worker. +} diff --git a/src/app/copy_mode.zig b/src/app/copy_mode.zig index 6af9386..907c37b 100644 --- a/src/app/copy_mode.zig +++ b/src/app/copy_mode.zig @@ -941,6 +941,6 @@ test "copy mode regex finder supports simple regexp operators" { try std.testing.expectEqual(@as(usize, 0), anchored_both.start); try std.testing.expectEqual(@as(usize, 3), anchored_both.end); - try std.testing.expectEqual(@as(?struct { start: usize, end: usize }, null), copyModeRegexFind("^foo", "xxfoo", 0)); - try std.testing.expectEqual(@as(?struct { start: usize, end: usize }, null), copyModeRegexFind("foo$", "foobar", 0)); + try std.testing.expect(copyModeRegexFind("^foo", "xxfoo", 0) == null); + try std.testing.expect(copyModeRegexFind("foo$", "foobar", 0) == null); } diff --git a/src/app/pty_budget.zig b/src/app/pty_budget.zig new file mode 100644 index 0000000..a7b9131 --- /dev/null +++ b/src/app/pty_budget.zig @@ -0,0 +1,40 @@ +const std = @import("std"); + +/// Each class gets its own parsing budget. Time is charged only while polling +/// that class, so active-pane work cannot spend a hidden pane's allowance. +pub const Budget = struct { + bytes: usize, + nanoseconds: i128, + panes: usize, + + pub fn byteQuota(self: Budget) usize { + if (self.panes == 0) return 0; + return std.math.divCeil(usize, self.bytes, self.panes) catch 0; + } + + pub fn timeQuota(self: Budget) i128 { + if (self.panes == 0) return 0; + return @divFloor(self.nanoseconds, @as(i128, @intCast(self.panes))); + } + + pub fn consume(self: *Budget, bytes: usize, elapsed_ns: i128) void { + self.bytes -|= bytes; + self.nanoseconds = @max(0, self.nanoseconds - @max(0, elapsed_ns)); + self.panes -|= 1; + } +}; + +test "busy panes share time and idle panes donate unused quota" { + var budget = Budget{ .bytes = 300, .nanoseconds = 3000, .panes = 3 }; + try std.testing.expectEqual(@as(usize, 100), budget.byteQuota()); + try std.testing.expectEqual(@as(i128, 1000), budget.timeQuota()); + budget.consume(0, 0); + try std.testing.expectEqual(@as(usize, 150), budget.byteQuota()); + try std.testing.expectEqual(@as(i128, 1500), budget.timeQuota()); + budget.consume(150, 1500); + try std.testing.expectEqual(@as(usize, 150), budget.byteQuota()); + try std.testing.expectEqual(@as(i128, 1500), budget.timeQuota()); + budget.consume(200, 2000); + try std.testing.expectEqual(@as(usize, 0), budget.byteQuota()); + try std.testing.expectEqual(@as(i128, 0), budget.timeQuota()); +} diff --git a/src/bench/renderer_bench.zig b/src/bench/renderer_bench.zig index 43c4a41..deaf5ae 100644 --- a/src/bench/renderer_bench.zig +++ b/src/bench/renderer_bench.zig @@ -473,11 +473,10 @@ fn buildCorpus(allocator: std.mem.Allocator, options: Options) ![]u8 { while (frame < options.frames) : (frame += 1) { var row: usize = 0; while (row < options.rows) : (row += 1) { - try corpus.appendSlice(allocator, - "\x1b[31m16 \x1b[38;5;196m256 \x1b[38;2;120;80;220mtruecolor\x1b[0m " ++ - "\x1b[1mbold\x1b[22m \x1b[3mitalic\x1b[23m \x1b[4munderline\x1b[24m " ++ - "\x1b[9mstrike\x1b[29m \x1b[7minverse\x1b[27m \u{250c}\u{2500}\u{2510} " ++ - "cafe\u{301}\u{754c} ligature ffi\n"); + try corpus.appendSlice(allocator, "\x1b[31m16 \x1b[38;5;196m256 \x1b[38;2;120;80;220mtruecolor\x1b[0m " ++ + "\x1b[1mbold\x1b[22m \x1b[3mitalic\x1b[23m \x1b[4munderline\x1b[24m " ++ + "\x1b[9mstrike\x1b[29m \x1b[7minverse\x1b[27m \u{250c}\u{2500}\u{2510} " ++ + "cafe\u{301}\u{754c} ligature ffi\n"); } } }, @@ -685,16 +684,7 @@ fn parseOptions(allocator: std.mem.Allocator, args: [][]u8) !Options { if (i + 1 >= args.len) return error.MissingOptionValue; const value = args[i + 1]; i += 1; - if (std.mem.eql(u8, arg, "--scenario")) options.scenario = try parseScenario(value) - else if (std.mem.eql(u8, arg, "--input")) options.input_path = try allocator.dupe(u8, value) - else if (std.mem.eql(u8, arg, "--frames")) options.frames = try parseUnsigned(value) - else if (std.mem.eql(u8, arg, "--rows")) options.rows = try parseUnsigned(value) - else if (std.mem.eql(u8, arg, "--cols")) options.cols = try parseUnsigned(value) - else if (std.mem.eql(u8, arg, "--chunk-bytes")) options.chunk_bytes = try parseUnsigned(value) - else if (std.mem.eql(u8, arg, "--warmup")) options.warmup = try parseUnsigned(value) - else if (std.mem.eql(u8, arg, "--iterations")) options.iterations = try parseUnsigned(value) - else if (std.mem.eql(u8, arg, "--mode")) options.mode = try parseMode(value) - else return error.UnknownOption; + if (std.mem.eql(u8, arg, "--scenario")) options.scenario = try parseScenario(value) else if (std.mem.eql(u8, arg, "--input")) options.input_path = try allocator.dupe(u8, value) else if (std.mem.eql(u8, arg, "--frames")) options.frames = try parseUnsigned(value) else if (std.mem.eql(u8, arg, "--rows")) options.rows = try parseUnsigned(value) else if (std.mem.eql(u8, arg, "--cols")) options.cols = try parseUnsigned(value) else if (std.mem.eql(u8, arg, "--chunk-bytes")) options.chunk_bytes = try parseUnsigned(value) else if (std.mem.eql(u8, arg, "--warmup")) options.warmup = try parseUnsigned(value) else if (std.mem.eql(u8, arg, "--iterations")) options.iterations = try parseUnsigned(value) else if (std.mem.eql(u8, arg, "--mode")) options.mode = try parseMode(value) else return error.UnknownOption; } if (options.rows == 0 or options.cols == 0 or options.rows > std.math.maxInt(u16) or options.cols > std.math.maxInt(u16)) return error.InvalidGrid; if (options.chunk_bytes == 0 or options.iterations == 0) return error.InvalidCount; @@ -726,9 +716,8 @@ pub fn main(init: std.process.Init) !void { if (options.json) { const chunks = (corpus.len + options.chunk_bytes - 1) / options.chunk_bytes; try output.interface.print("{{\"scenario\":\"{s}\",\"mode\":\"{s}\",\"rows\":{d},\"cols\":{d},\"frames\":{d},\"bytes\":{d},\"chunk_bytes\":{d},\"chunks\":{d},\"iterations\":{d},\"input_checksum\":\"{x}\",\"render_state_rows\":{d},\"render_state_cols\":{d},\"dirty_level\":\"{s}\",\"cursor_row\":{d},\"cursor_col\":{d},\"cells_visited\":{d},\"glyph_runs\":{d},\"bg_rects\":{d},\"atlas_flushed\":{},\"glyph_verts_count\":{d},\"stages\":{{", .{ - @tagName(options.scenario), @tagName(options.mode), options.rows, options.cols, options.frames, corpus.len, options.chunk_bytes, chunks, options.iterations, checksum, - harness.render_state_rows, harness.render_state_cols, @tagName(harness.dirty_level), harness.final_cursor_row, harness.final_cursor_col, - harness.last_cells_visited, harness.last_glyph_runs, harness.last_bg_rects, harness.last_atlas_flushed, harness.last_glyph_verts, + @tagName(options.scenario), @tagName(options.mode), options.rows, options.cols, options.frames, corpus.len, options.chunk_bytes, chunks, options.iterations, checksum, + harness.render_state_rows, harness.render_state_cols, @tagName(harness.dirty_level), harness.final_cursor_row, harness.final_cursor_col, harness.last_cells_visited, harness.last_glyph_runs, harness.last_bg_rects, harness.last_atlas_flushed, harness.last_glyph_verts, }); var first = true; if (results.parse.values.items.len > 0) { @@ -765,13 +754,12 @@ pub fn main(init: std.process.Init) !void { } else { const chunks = (corpus.len + options.chunk_bytes - 1) / options.chunk_bytes; try output.interface.print("scenario: {s}\nmode: {s}\ngrid: {d}x{d}\nframes: {d}\nbytes: {d}\nchunk_bytes: {d}\nchunks: {d}\niterations: {d}\ninput_checksum: {x}\nrender_state: {d}x{d}\ndirty_level: {s}\ncursor: {d},{d}\nrenderer_counters: rows_rendered={d} rows_skipped={d} cells_visited={d} glyph_runs={d} bg_rects={d} atlas_flushed={} glyph_verts_count={d}\n", .{ - @tagName(options.scenario), @tagName(options.mode), options.cols, options.rows, options.frames, corpus.len, options.chunk_bytes, chunks, options.iterations, checksum, - harness.render_state_cols, harness.render_state_rows, @tagName(harness.dirty_level), harness.final_cursor_row, harness.final_cursor_col, - harness.last_rows_rendered, harness.last_rows_skipped, harness.last_cells_visited, harness.last_glyph_runs, - harness.last_bg_rects, harness.last_atlas_flushed, harness.last_glyph_verts, - }); - if (results.cold_render.values.items.len > 0) try output.interface.print("cold_render_cpu_ms: {d:.3}\n", .{@as(f64, @floatFromInt(results.cold_render.values.items[0])) / 1_000_000.0}); - if (results.cold_pipeline.values.items.len > 0) try output.interface.print("cold_pipeline_ms: {d:.3}\n", .{@as(f64, @floatFromInt(results.cold_pipeline.values.items[0])) / 1_000_000.0}); + @tagName(options.scenario), @tagName(options.mode), options.cols, options.rows, options.frames, corpus.len, options.chunk_bytes, chunks, options.iterations, checksum, + harness.render_state_cols, harness.render_state_rows, @tagName(harness.dirty_level), harness.final_cursor_row, harness.final_cursor_col, harness.last_rows_rendered, harness.last_rows_skipped, harness.last_cells_visited, harness.last_glyph_runs, harness.last_bg_rects, + harness.last_atlas_flushed, harness.last_glyph_verts, + }); + if (results.cold_render.values.items.len > 0) try output.interface.print("cold_render_cpu_ms: {d:.3}\n", .{@as(f64, @floatFromInt(results.cold_render.values.items[0])) / 1_000_000.0}); + if (results.cold_pipeline.values.items.len > 0) try output.interface.print("cold_pipeline_ms: {d:.3}\n", .{@as(f64, @floatFromInt(results.cold_pipeline.values.items[0])) / 1_000_000.0}); if (results.parse.values.items.len > 0) try printStats(&output, "parse", results.parse.values.items, corpus.len); if (results.render_state.values.items.len > 0) try printStats(&output, "render_state", results.render_state.values.items, 0); if (results.render.values.items.len > 0) try printStats(&output, "render_cpu", results.render.values.items, 0); @@ -822,3 +810,24 @@ test "renderer benchmark checksum remains deterministic" { const allocator = std.testing.allocator; try std.testing.expectEqual(@as(u64, 0x652825d1a05e7565), try deterministicRepaintChecksum(allocator)); } + +pub fn runCachePressureTest(allocator: std.mem.Allocator) !void { + var harness = try Harness.init(allocator, .{}); + defer harness.deinit(); + const renderer = &harness.renderer; + renderer.text_cache_limit_bytes = 4096; + var buffer: [64]u8 = undefined; + for (0..2000) |i| { + const text = try std.fmt.bufPrint(&buffer, "unique-{d}", .{i}); + renderer.beginFrame(); + const shaped = renderer.getOrShape(text, 0) orelse return error.ShapeFailed; + const prepared = renderer.prepareShapedGlyphs(shaped, .terminal) orelse return error.PrepareFailed; + renderer.putPreparedCache(text, 0, .terminal, prepared.glyphs); + try std.testing.expect(renderer.shape_cache_bytes <= renderer.text_cache_limit_bytes); + try std.testing.expect(renderer.prepared_cache_bytes <= renderer.text_cache_limit_bytes); + // Exercise the borrowed-pointer fast path after repeated evictions. + try std.testing.expect(renderer.getPreparedCache(text, 0, .terminal) != null); + } + try std.testing.expect(renderer.shape_cache_evictions > 0); + try std.testing.expect(renderer.prepared_cache_evictions > 0); +} diff --git a/src/ipc.zig b/src/ipc.zig index 3d203ce..ee30156 100644 --- a/src/ipc.zig +++ b/src/ipc.zig @@ -14,13 +14,18 @@ const windows = if (builtin.os.tag == .windows) std.os.windows else void; extern "kernel32" fn MoveFileExW(lpExistingFileName: [*:0]const u16, lpNewFileName: [*:0]const u16, dwFlags: windows.DWORD) callconv(.winapi) windows.BOOL; pub const Server = struct { + const Connection = struct { + stream: ?std.Io.net.Stream = null, + thread: ?std.Thread = null, + }; + allocator: std.mem.Allocator, app: *anyopaque, handler: *const fn (app: *anyopaque, request: command.Request) command.Response, thread: ?std.Thread = null, stop_flag: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), active_mutex: io.Mutex = .{}, - active_stream: ?std.Io.net.Stream = null, + connections: [8]Connection = [_]Connection{.{}} ** 8, wake_stream: ?std.Io.net.Stream = null, listen_address: ?std.Io.net.IpAddress = null, listen_address_text: ?[]u8 = null, @@ -43,6 +48,7 @@ pub const Server = struct { pub fn start(self: *Server) !void { if (self.started) return; + self.stop_flag.store(false, .release); const configured_addr = io.getEnvVarOwned(self.allocator, EnvVar) catch null; defer if (configured_addr) |value| self.allocator.free(value); @@ -52,6 +58,8 @@ pub const Server = struct { else std.Io.net.IpAddress{ .ip4 = .loopback(0) }; + if (!isLoopback(bind_address)) return error.NonLoopbackCommandAddress; + var listener = try bind_address.listen(io.get(), .{ .reuse_address = true }); errdefer listener.deinit(io.get()); @@ -74,7 +82,9 @@ pub const Server = struct { self.stop_flag.store(true, .release); self.active_mutex.lock(); - if (self.active_stream) |stream| stream.shutdown(io.get(), .both) catch {}; + for (&self.connections) |*connection| { + if (connection.stream) |stream| stream.shutdown(io.get(), .both) catch {}; + } self.active_mutex.unlock(); self.wakeAcceptLoop(); if (self.wake_stream) |stream| { @@ -82,6 +92,10 @@ pub const Server = struct { self.wake_stream = null; } if (self.thread) |thread| thread.join(); + for (&self.connections) |*connection| { + if (connection.thread) |thread| thread.join(); + connection.thread = null; + } self.unpublishAddress(); self.thread = null; self.listen_address = null; @@ -150,21 +164,50 @@ pub const Server = struct { stream.close(io.get()); break; } - self.active_stream = stream; - self.active_mutex.unlock(); - std.log.info("command-ipc: accepted connection from {f}", .{stream.socket.address}); - handleConnection(self, stream) catch |err| { - std.log.warn("command-ipc: request failed: {s}", .{@errorName(err)}); - }; - self.active_mutex.lock(); - stream.close(io.get()); - self.active_stream = null; - self.active_mutex.unlock(); + var available: ?usize = null; + for (&self.connections, 0..) |*connection, index| { + if (connection.stream == null) { + available = index; + break; + } + } + if (available) |index| { + const connection = &self.connections[index]; + // A cleared stream means the worker has released the mutex and + // will not touch its slot again. Reap before reusing the slot. + const previous = connection.thread; + connection.stream = stream; + self.active_mutex.unlock(); + if (previous) |thread| thread.join(); + connection.thread = std.Thread.spawn(.{}, connectionLoop, .{ self, index, stream }) catch { + self.active_mutex.lock(); + stream.close(io.get()); + connection.stream = null; + connection.thread = null; + self.active_mutex.unlock(); + continue; + }; + } else { + self.active_mutex.unlock(); + stream.close(io.get()); + } } } + fn connectionLoop(self: *Server, index: usize, stream: std.Io.net.Stream) void { + handleConnection(self, stream) catch |err| { + std.log.warn("command-ipc: request failed: {s}", .{@errorName(err)}); + }; + self.active_mutex.lock(); + stream.close(io.get()); + self.connections[index].stream = null; + self.active_mutex.unlock(); + } + fn handleConnection(self: *Server, stream: std.Io.net.Stream) !void { - try setTimeouts(stream, server_timeout_ms); + var deadline = SocketDeadline{ .stream = stream, .timeout_ms = server_timeout_ms }; + try deadline.start(); + defer deadline.finish(); const frame = try readFrame(self.allocator, stream); defer self.allocator.free(frame); @@ -230,11 +273,17 @@ pub fn send(allocator: std.mem.Allocator, request: command.Request, timeout_ms: const connect_start_ns = if (timing_enabled) io.nanoTimestamp() else 0; const remote_addr = try std.Io.net.IpAddress.parseLiteral(addr_text); - const stream = try remote_addr.connect(io.get(), .{ .mode = .stream, .protocol = .tcp }); + const stream = try remote_addr.connect(io.get(), .{ + .mode = .stream, + .protocol = .tcp, + .timeout = if (timeout_ms == 0) .none else .{ .duration = .{ .clock = .awake, .raw = .fromMilliseconds(@intCast(@min(timeout_ms, std.math.maxInt(i64)))) } }, + }); defer stream.close(io.get()); if (timing_enabled) clientTraceFmt("connect_ms={d:.3}", .{elapsedMs(connect_start_ns)}); - try setTimeouts(stream, timeout_ms); + var deadline = SocketDeadline{ .stream = stream, .timeout_ms = timeout_ms }; + try deadline.start(); + defer deadline.finish(); const encode_start_ns = if (timing_enabled) io.nanoTimestamp() else 0; const payload = try encodeRequest(allocator, request); @@ -413,22 +462,37 @@ fn readSocket(stream: std.Io.net.Stream, buffer: []u8) !usize { return reader.interface.readSliceShort(buffer) catch return reader.err orelse error.ConnectionClosed; } -fn setTimeouts(stream: std.Io.net.Stream, timeout_ms: u64) !void { - if (timeout_ms == 0) return; +/// A total operation deadline, including slow trickle reads. AFD handles on +/// Windows do not support Winsock timeouts, but stream shutdown wakes readers. +const SocketDeadline = struct { + stream: std.Io.net.Stream, + timeout_ms: u64, + mutex: std.Io.Mutex = .init, + done_condition: std.Io.Condition = .init, + done: bool = false, + thread: ?std.Thread = null, - if (builtin.os.tag == .windows) { - // std.Io.net uses Windows AFD handles, not Winsock sockets. - // Winsock socket options cannot be applied to these handles. - return; + fn start(self: *SocketDeadline) !void { + if (self.timeout_ms != 0) self.thread = try std.Thread.spawn(.{}, watch, .{self}); } - var value = std.posix.timeval{ - .sec = @intCast(timeout_ms / std.time.ms_per_s), - .usec = @intCast((timeout_ms % std.time.ms_per_s) * std.time.us_per_ms), - }; - if (std.c.setsockopt(stream.socket.handle, std.posix.SOL.SOCKET, std.c.SO.RCVTIMEO, &value, @sizeOf(@TypeOf(value))) != 0) return error.SetSocketTimeoutFailed; - if (std.c.setsockopt(stream.socket.handle, std.posix.SOL.SOCKET, std.c.SO.SNDTIMEO, &value, @sizeOf(@TypeOf(value))) != 0) return error.SetSocketTimeoutFailed; -} + fn finish(self: *SocketDeadline) void { + self.mutex.lockUncancelable(io.get()); + self.done = true; + self.done_condition.signal(io.get()); + self.mutex.unlock(io.get()); + if (self.thread) |thread| thread.join(); + } + + fn watch(self: *SocketDeadline) void { + self.mutex.lockUncancelable(io.get()); + defer self.mutex.unlock(io.get()); + if (self.done) return; + io.waitTimeout(&self.done_condition, &self.mutex, self.timeout_ms *| std.time.ns_per_ms) catch { + if (!self.done) self.stream.shutdown(io.get(), .both) catch {}; + }; + } +}; fn jsonObjectString(object: std.json.ObjectMap, key: []const u8) ?[]const u8 { const value = object.get(key) orelse return null; @@ -452,3 +516,63 @@ test "encoded request round trips absent automation fields" { try std.testing.expect(parsed.request.revision == null); try std.testing.expect(parsed.request.generation == null); } + +fn isLoopback(address: std.Io.net.IpAddress) bool { + return switch (address) { + .ip4 => |value| value.bytes[0] == 127, + .ip6 => |value| value.isLoopBack(), + }; +} + +test "command transport accepts only loopback bindings" { + try std.testing.expect(isLoopback(try std.Io.net.IpAddress.parseLiteral("127.0.0.1:0"))); + try std.testing.expect(isLoopback(try std.Io.net.IpAddress.parseLiteral("[::1]:0"))); + try std.testing.expect(!isLoopback(try std.Io.net.IpAddress.parseLiteral("0.0.0.0:0"))); + try std.testing.expect(!isLoopback(try std.Io.net.IpAddress.parseLiteral("192.168.1.2:0"))); +} + +fn testHandler(_: *anyopaque, _: command.Request) command.Response { + return .{}; +} + +test "idle IPC client does not block another request" { + var context: u8 = 0; + var server = Server.init(std.testing.allocator, &context, testHandler); + // Do not publish a test address over a running user's discovery file. + const listener = try (std.Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io.get(), .{}); + server.listen_address = listener.socket.address; + server.thread = try std.Thread.spawn(.{}, Server.acceptLoop, .{ &server, listener }); + server.started = true; + defer server.deinit(); + const idle = try server.listen_address.?.connect(io.get(), .{ .mode = .stream }); + defer idle.close(io.get()); + const active = try server.listen_address.?.connect(io.get(), .{ .mode = .stream }); + defer active.close(io.get()); + var deadline = SocketDeadline{ .stream = active, .timeout_ms = 1000 }; + try deadline.start(); + defer deadline.finish(); + try writeFrame(active, "{\"kind\":\"get_revision\"}"); + const reply = try readFrame(std.testing.allocator, active); + defer std.testing.allocator.free(reply); + var result = try decodeResponse(std.testing.allocator, reply); + defer result.deinit(std.testing.allocator); + try std.testing.expect(result.success); +} + +test "socket deadline interrupts an incomplete frame" { + var listener = try (std.Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io.get(), .{}); + defer listener.deinit(io.get()); + const client = try listener.socket.address.connect(io.get(), .{ .mode = .stream }); + defer client.close(io.get()); + const accepted = try listener.accept(io.get()); + defer accepted.close(io.get()); + var deadline = SocketDeadline{ .stream = accepted, .timeout_ms = 20 }; + try deadline.start(); + defer deadline.finish(); + // A partial header must not keep the worker alive indefinitely. + try writeAllSocket(client, &.{1}); + if (readFrame(std.testing.allocator, accepted)) |frame| { + std.testing.allocator.free(frame); + return error.ExpectedDeadline; + } else |_| {} +} diff --git a/src/lua/core.lua b/src/lua/core.lua index c2d7460..f7f3742 100644 --- a/src/lua/core.lua +++ b/src/lua/core.lua @@ -334,10 +334,60 @@ function hollow.process.run(cmd, args) return host_api.run_process(cmd, args or {}) end -function hollow.process.spawn(_opts) - util.unsupported("hollow.process.spawn") +function hollow.process.spawn(opts) + if type(opts) ~= "table" then error("process.spawn expects options") end + local cmd = opts.cmd + if type(cmd) == "string" then cmd = { cmd } end + if type(cmd) ~= "table" or #cmd == 0 then error("process.spawn requires cmd argv") end + for _, arg in ipairs(cmd) do + if type(arg) ~= "string" or arg:find("%z") then error("cmd items must be strings without NUL") end + end + if opts.env ~= nil then + if type(opts.env) ~= "table" then error("env must be a table") end + for key, value in pairs(opts.env) do + if type(key) ~= "string" or key == "" or key:find("[=%z]") or type(value) ~= "string" or value:find("%z") then + error("env must contain valid string keys and values") + end + end + end + if opts.cwd ~= nil and (type(opts.cwd) ~= "string" or opts.cwd:find("%z")) then error("cwd must be a string without NUL") end + for _, name in ipairs({ "timeout_ms", "output_limit" }) do + local value = opts[name] + if value ~= nil and (type(value) ~= "number" or value ~= value or value <= 0 or value == math.huge or value % 1 ~= 0) then + error(name .. " must be a positive integer") + end + end + if opts.on_complete ~= nil and type(opts.on_complete) ~= "function" then error("on_complete must be a function") end + local id, err = host_api.process_start({ cmd = cmd, cwd = opts.cwd, env = opts.env, timeout_ms = opts.timeout_ms, output_limit = opts.output_limit }) + if not id then error("process.spawn: " .. tostring(err)) end + local result + local resolve_result + local promise = hollow.async.promise(function(resolve) resolve_result = resolve end) + local handle = {} + function handle:status() return result and "finished" or "running" end + function handle:result() return result end + function handle:cancel() + if not result then host_api.process_cancel(id) end + end + handle.kill = handle.cancel + function handle:wait() return promise:await() end + function handle:next(fn) return promise:next(fn) end + local function poll() + result = host_api.process_poll(id) + if not result then host_api.defer(poll, 10); return end + resolve_result(result) + if opts.on_complete then opts.on_complete(result) end + end + host_api.defer(poll, 10) + return handle end -function hollow.process.exec(_opts) - util.unsupported("hollow.process.exec") +function hollow.process.exec(opts) + return hollow.async.promise(function(resolve, reject) + local ok, handle = pcall(hollow.process.spawn, opts) + if not ok then reject(handle); return end + handle:next(function(result) + if result.error then reject(result) else resolve(result) end + end) + end) end diff --git a/src/lua/tests/test_process.lua b/src/lua/tests/test_process.lua new file mode 100644 index 0000000..4a9f778 --- /dev/null +++ b/src/lua/tests/test_process.lua @@ -0,0 +1,67 @@ +local harness = require("tests.harness") + +describe("asynchronous processes", function() + local env + before_each(function() env = harness.boot() end) + + it("returns immediately, polls pending work and completes once", function() + local polls, completed = 0, 0 + env.host_api.process_start = function(opts) + assert.are.same({ "git", "status" }, opts.cmd) + assert.are.equal("/tmp", opts.cwd) + return 1 + end + env.host_api.process_poll = function() + polls = polls + 1 + if polls == 1 then return nil end + return { code = 0, stdout = "clean", stderr = "" } + end + local handle = env.hollow.process.spawn({ + cmd = { "git", "status" }, cwd = "/tmp", + on_complete = function() completed = completed + 1 end, + }) + assert.are.equal("running", handle:status()) + assert.are.equal(0, polls) + local received + handle:next(function(result) received = result end) + env.flush_deferred() + assert.are.equal("finished", handle:status()) + assert.are.equal("clean", received.stdout) + assert.are.equal(1, completed) + assert.are.equal(received, handle:result()) + end) + + it("supports cancellation without canceling a reused native slot", function() + local canceled = 0 + env.host_api.process_start = function() return 1 end + env.host_api.process_cancel = function() canceled = canceled + 1 end + env.host_api.process_poll = function() return { code = -1, error = "Canceled", canceled = true } end + local handle = env.hollow.process.spawn({ cmd = "sleep" }) + handle:cancel() + env.flush_deferred() + handle:cancel() + assert.are.equal(1, canceled) + assert.is_true(handle:result().canceled) + end) + + it("exec rejects infrastructure failures and preserves nonzero exit codes", function() + env.host_api.process_start = function() return 1 end + env.host_api.process_poll = function() return { code = 7, stdout = "", stderr = "failed" } end + local value + env.hollow.process.exec({ cmd = "tool" }):next(function(result) value = result end) + env.flush_deferred() + assert.are.equal(7, value.code) + env.host_api.process_poll = function() return { code = -1, error = "Timeout" } end + local failure + env.hollow.process.exec({ cmd = "tool" }):catch(function(result) failure = result end) + env.flush_deferred() + assert.are.equal("Timeout", failure.error) + end) + + it("validates options before starting native work", function() + env.host_api.process_start = function() error("must not start") end + assert.has_error(function() env.hollow.process.spawn({ cmd = {} }) end, "process.spawn requires cmd argv") + assert.has_error(function() env.hollow.process.spawn({ cmd = "tool", timeout_ms = -1 }) end, "timeout_ms must be a positive integer") + assert.has_error(function() env.hollow.process.spawn({ cmd = "tool", env = { BAD = 1 } }) end, "env must contain valid string keys and values") + end) +end) diff --git a/src/lua_bridge.zig b/src/lua_bridge.zig index 8be3455..7468b58 100644 --- a/src/lua_bridge.zig +++ b/src/lua_bridge.zig @@ -1,3 +1,4 @@ +const ProcessJob = @import("process_jobs.zig").Job; const std = @import("std"); const io = @import("io.zig"); const config = @import("config.zig"); @@ -839,6 +840,7 @@ pub const Runtime = struct { } pub fn deinit(self: *Runtime) void { + for (self.context.process_jobs) |job| if (job) |value| value.destroy(); if (self.lua_sources) |*sources| sources.deinit(self.allocator); if (self.context.pending_workspace_name) |name| self.allocator.free(name); if (self.context.on_key_ref != LUA_NOREF) self.context.api.unref(self.state, LUA_REGISTRYINDEX, self.context.on_key_ref); @@ -1853,6 +1855,16 @@ pub const Runtime = struct { api.push_cclosure(self.state, l_run_child_process, 1); api.set_field(self.state, -2, "run_child_process"); + api.push_light_userdata(self.state, self.context); + api.push_cclosure(self.state, l_process_start, 1); + api.set_field(self.state, -2, "process_start"); + api.push_light_userdata(self.state, self.context); + api.push_cclosure(self.state, l_process_poll, 1); + api.set_field(self.state, -2, "process_poll"); + api.push_light_userdata(self.state, self.context); + api.push_cclosure(self.state, l_process_cancel, 1); + api.set_field(self.state, -2, "process_cancel"); + api.push_light_userdata(self.state, self.context); api.push_cclosure(self.state, l_run_process, 1); api.set_field(self.state, -2, "run_process"); @@ -2343,6 +2355,7 @@ const BridgeContext = struct { quick_select_match_ref: c_int = -1, quick_select_action_ref: c_int = -1, gui_ready_fired: bool = false, + process_jobs: [16]?*ProcessJob = [_]?*ProcessJob{null} ** 16, deferred_callback_refs: std.ArrayListUnmanaged(c_int) = .empty, timed_callback_refs: std.ArrayListUnmanaged(TimedCallback) = .empty, }; @@ -6817,3 +6830,130 @@ test "drainExpiredTimedCallbacks preserves timer order" { try std.testing.expectEqualSlices(c_int, &.{ 11, 33 }, &.{ pending.items[0].ref, pending.items[1].ref }); try std.testing.expectEqualSlices(c_int, &.{ 22, 44 }, &.{ timed_callback_refs.items[0].ref, timed_callback_refs.items[1].ref }); } + +fn startProcess(ctx: *BridgeContext, state: *State) !usize { + const api = ctx.api; + const slot = for (ctx.process_jobs, 0..) |job, i| { + if (job == null) break i; + } else return error.ProcessJobLimit; + if (@as(LuaType, @enumFromInt(api.value_type(state, 1))) != .table) return error.ExpectedOptions; + api.get_field(state, 1, "cmd"); + if (@as(LuaType, @enumFromInt(api.value_type(state, -1))) != .table) return error.ExpectedArgv; + var argv: std.ArrayList([]const u8) = .empty; + defer argv.deinit(std.heap.page_allocator); + var i: c_int = 1; + while (true) : (i += 1) { + api.rawgeti(state, -1, i); + if (@as(LuaType, @enumFromInt(api.value_type(state, -1))) == .nil_type) { + pop(api, state, 1); + break; + } + var len: usize = 0; + const ptr = api.to_lstring(state, -1, &len) orelse return error.InvalidArgument; + argv.append(std.heap.page_allocator, ptr[0..len]) catch return error.OutOfMemory; + pop(api, state, 1); + } + const cwd = luaStringField(api, state, 1, "cwd"); + const timeout = @min(luaNonNegativeIntegerField(api, state, 1, "timeout_ms") orelse 30_000, 24 * 60 * 60 * 1000); + const limit = @min(luaNonNegativeIntegerField(api, state, 1, "output_limit") orelse 1024 * 1024, 16 * 1024 * 1024); + var environment = io.environ().createMap(std.heap.page_allocator) catch return error.EnvironmentUnavailable; + defer environment.deinit(); + api.get_field(state, 1, "env"); + if (@as(LuaType, @enumFromInt(api.value_type(state, -1))) == .table) { + const env_idx = absoluteIndex(api, state, -1); + api.push_nil(state); + while (api.next(state, env_idx) != 0) { + if (@as(LuaType, @enumFromInt(api.value_type(state, -2))) != .string or + @as(LuaType, @enumFromInt(api.value_type(state, -1))) != .string) return error.InvalidEnvironment; + var key_len: usize = 0; + var value_len: usize = 0; + const key = api.to_lstring(state, -2, &key_len) orelse return error.InvalidEnvironment; + const value = api.to_lstring(state, -1, &value_len) orelse return error.InvalidEnvironment; + environment.put(key[0..key_len], value[0..value_len]) catch return error.InvalidEnvironment; + pop(api, state, 1); + } + } + const job = try ProcessJob.create(ctx.allocator, argv.items, cwd, timeout, limit, &environment); + ctx.process_jobs[slot] = job; + return slot + 1; +} + +fn l_process_start(state: *State) callconv(.c) c_int { + const ctx = bridgeContext(state); + const id = startProcess(ctx, state) catch |err| { + ctx.api.push_nil(state); + const name = @errorName(err); + ctx.api.push_lstring(state, name.ptr, name.len); + return 2; + }; + ctx.api.push_integer(state, @intCast(id)); + return 1; +} + +fn processSlot(ctx: *BridgeContext, state: *State) ?usize { + const number = ctx.api.to_number(state, 1); + if (!std.math.isFinite(number) or number != @floor(number) or number < 1 or number > ctx.process_jobs.len) return null; + const id: usize = @intFromFloat(number); + return @intCast(id - 1); +} + +fn l_process_cancel(state: *State) callconv(.c) c_int { + const ctx = bridgeContext(state); + if (processSlot(ctx, state)) |slot| if (ctx.process_jobs[slot]) |job| { + job.cancel(); + }; + return 0; +} + +fn l_process_poll(state: *State) callconv(.c) c_int { + const ctx = bridgeContext(state); + const api = ctx.api; + const slot = processSlot(ctx, state) orelse return 0; + const job = ctx.process_jobs[slot] orelse return 0; + if (!job.ready()) return 0; + defer { + job.destroy(); + ctx.process_jobs[slot] = null; + } + api.create_table(state, 0, 5); + api.push_integer(state, if (job.failure != null) -1 else if (job.result) |result| childExitCode(result.term) else -1); + api.set_field(state, -2, "code"); + const stdout = if (job.result) |result| result.stdout else ""; + api.push_lstring(state, stdout.ptr, stdout.len); + api.set_field(state, -2, "stdout"); + const stderr = if (job.result) |result| result.stderr else ""; + api.push_lstring(state, stderr.ptr, stderr.len); + api.set_field(state, -2, "stderr"); + if (job.failure) |err| { + const name = @errorName(err); + api.push_lstring(state, name.ptr, name.len); + api.set_field(state, -2, "error"); + } + api.push_boolean(state, if (job.canceled) 1 else 0); + api.set_field(state, -2, "canceled"); + return 1; +} + +test "native process bridge owns argv and returns structured results" { + if (@import("builtin").os.tag == .windows) return error.SkipZigTest; + var cfg = config.Config.init(std.testing.allocator); + defer cfg.deinit(); + var runtime = try Runtime.init(std.testing.allocator, &cfg); + defer runtime.deinit(); + try runtime.runString( + \\job_id = assert(host_api.process_start({ + \\ cmd = {"/bin/sh", "-c", "printf '%s' \"$HOLLOW_JOB_TEST\"; printf problem >&2; exit 7"}, + \\ cwd = "/tmp", env = {HOLLOW_JOB_TEST = "owned-value"}, timeout_ms = 2000, + \\})) + \\collectgarbage("collect") + ); + runtime.context.process_jobs[0].?.future.?.await(io.get()); + try runtime.runString( + \\local result = host_api.process_poll(job_id) + \\assert(result.code == 7) + \\assert(result.stdout == "owned-value") + \\assert(result.stderr == "problem") + \\assert(result.error == nil) + ); + try std.testing.expect(runtime.context.process_jobs[0] == null); +} diff --git a/src/main.zig b/src/main.zig index fe0022c..05a0796 100644 --- a/src/main.zig +++ b/src/main.zig @@ -619,6 +619,13 @@ fn windowsStdHandle(stream_id: win32.DWORD) ?std.Io.File { } test { + _ = @import("ipc.zig"); + _ = @import("process_jobs.zig"); + _ = @import("app/pty_budget.zig"); + _ = @import("app/command_dispatcher.zig"); + _ = @import("app/action_queue.zig"); + _ = @import("render/sokol_runtime.zig"); + if (builtin.os.tag != .windows) _ = @import("pty/pty_posix.zig"); _ = @import("config.zig"); _ = @import("platform.zig"); _ = @import("lua_bridge.zig"); diff --git a/src/process_jobs.zig b/src/process_jobs.zig new file mode 100644 index 0000000..10575f9 --- /dev/null +++ b/src/process_jobs.zig @@ -0,0 +1,122 @@ +const std = @import("std"); +const io = @import("io.zig"); + +/// Owned process work. Only the owner accesses Future; workers never enter Lua. +pub const Job = struct { + allocator: std.mem.Allocator, + arena: std.heap.ArenaAllocator, + argv: []const []const u8, + cwd: ?[]const u8, + environment: ?std.process.Environ.Map, + output_limit: usize, + deadline_ns: i128, + done: std.atomic.Value(bool) = .init(false), + future: ?std.Io.Future(void) = null, + result: ?std.process.RunResult = null, + failure: ?anyerror = null, + canceled: bool = false, + + pub fn create(allocator: std.mem.Allocator, argv: []const []const u8, cwd: ?[]const u8, timeout_ms: usize, output_limit: usize, environment: ?*const std.process.Environ.Map) !*Job { + if (argv.len == 0 or argv[0].len == 0) return error.EmptyCommand; + const job = try allocator.create(Job); + errdefer allocator.destroy(job); + var arena = std.heap.ArenaAllocator.init(allocator); + errdefer arena.deinit(); + const owned = try arena.allocator().alloc([]const u8, argv.len); + for (argv, 0..) |arg, i| { + if (std.mem.indexOfScalar(u8, arg, 0) != null) return error.InvalidArgument; + owned[i] = try arena.allocator().dupe(u8, arg); + } + const owned_cwd = if (cwd) |path| try arena.allocator().dupe(u8, path) else null; + var owned_environment = if (environment) |value| try value.clone(allocator) else null; + errdefer if (owned_environment) |*value| value.deinit(); + job.* = .{ + .allocator = allocator, + .arena = arena, + .argv = owned, + .cwd = owned_cwd, + .environment = owned_environment, + .output_limit = output_limit, + .deadline_ns = io.nanoTimestamp() + @as(i128, @intCast(timeout_ms)) * std.time.ns_per_ms, + }; + job.future = try io.get().concurrent(run, .{job}); + return job; + } + + fn run(self: *Job) void { + self.result = std.process.run(self.allocator, io.get(), .{ + .argv = self.argv, + .cwd = if (self.cwd) |path| .{ .path = path } else .inherit, + .environ_map = if (self.environment) |*value| value else null, + .stdout_limit = .limited(self.output_limit), + .stderr_limit = .limited(self.output_limit), + .timeout = .{ .deadline = .{ .clock = .awake, .raw = .fromNanoseconds(@intCast(self.deadline_ns)) } }, + .create_no_window = true, + }) catch |err| blk: { + self.failure = err; + break :blk null; + }; + self.done.store(true, .release); + } + + pub fn ready(self: *Job) bool { + if (self.done.load(.acquire)) return true; + if (io.nanoTimestamp() >= self.deadline_ns) { + if (self.future) |*future| future.cancel(io.get()); + self.failure = error.Timeout; + return true; + } + return false; + } + + pub fn cancel(self: *Job) void { + if (self.done.load(.acquire)) return; + self.canceled = true; + if (self.future) |*future| future.cancel(io.get()); + self.failure = error.Canceled; + } + + pub fn destroy(self: *Job) void { + if (self.future) |*future| future.cancel(io.get()); + if (self.result) |result| { + self.allocator.free(result.stdout); + self.allocator.free(result.stderr); + } + if (self.environment) |*value| value.deinit(); + self.arena.deinit(); + self.allocator.destroy(self); + } +}; + +test "process jobs collect output without blocking the owner" { + if (@import("builtin").os.tag == .windows) return error.SkipZigTest; + const job = try Job.create(std.testing.allocator, &.{ "/bin/sh", "-c", "printf hello; printf problem >&2; exit 7" }, null, 2000, 4096, null); + defer job.destroy(); + job.future.?.await(io.get()); + try std.testing.expect(job.ready()); + try std.testing.expect(job.failure == null); + try std.testing.expectEqualStrings("hello", job.result.?.stdout); + try std.testing.expectEqualStrings("problem", job.result.?.stderr); + try std.testing.expectEqual(@as(u8, 7), job.result.?.term.exited); +} + +test "cancel process with closed output pipes" { + if (@import("builtin").os.tag == .windows) return error.SkipZigTest; + const job = try Job.create(std.testing.allocator, &.{ "/bin/sh", "-c", "exec 1>&- 2>&-; exec sleep 30" }, null, 2000, 4096, null); + defer job.destroy(); + job.cancel(); + try std.testing.expect(job.ready()); + try std.testing.expectEqual(error.Canceled, job.failure.?); +} + +test "process output limits and deadlines terminate work" { + if (@import("builtin").os.tag == .windows) return error.SkipZigTest; + const verbose = try Job.create(std.testing.allocator, &.{ "/bin/sh", "-c", "printf '12345678901234567890'" }, null, 2000, 4, null); + defer verbose.destroy(); + verbose.future.?.await(io.get()); + try std.testing.expectEqual(error.StreamTooLong, verbose.failure.?); + const slow = try Job.create(std.testing.allocator, &.{ "/bin/sh", "-c", "exec sleep 30" }, null, 20, 4096, null); + defer slow.destroy(); + slow.future.?.await(io.get()); + try std.testing.expectEqual(error.Timeout, slow.failure.?); +} diff --git a/src/pty/pty_posix.zig b/src/pty/pty_posix.zig index 7710955..fc75d17 100644 --- a/src/pty/pty_posix.zig +++ b/src/pty/pty_posix.zig @@ -19,6 +19,17 @@ const c = @cImport({ const READER_HIGH_WATER_BYTES = 4 * 1024 * 1024; +const WRITER_HIGH_WATER_BYTES = 4 * 1024 * 1024; + +const WriterState = struct { + mutex: io.Mutex = .{}, + ready: io.Condition = .{}, + buf: std.ArrayListUnmanaged(u8) = .empty, + start: usize = 0, + closing: bool = false, + failed: bool = false, +}; + const ReaderState = struct { mutex: io.Mutex = .{}, ready: io.Condition = .{}, @@ -37,6 +48,8 @@ pub const PosixPty = struct { pid: c.pid_t, reader_state: *ReaderState, reader_thread: ?std.Thread = null, + writer_state: *WriterState, + writer_thread: ?std.Thread = null, alive: bool = true, closed: bool = false, @@ -70,7 +83,7 @@ pub const PosixPty = struct { var env_arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer env_arena.deinit(); const envp = if (env_block) |env| buildEnvp(env_arena.allocator(), env) catch c._exit(1) else null; - execWithPath(shell_path, argv, if (envp) |items| @constCast(@ptrCast(items.ptr)) else null); + execWithPath(shell_path, argv, if (envp) |items| @ptrCast(@constCast(items.ptr)) else null); c._exit(1); } if (pid < 0) return error.ForkPtyFailed; @@ -84,12 +97,29 @@ pub const PosixPty = struct { reader_state.* = .{ .wake = wake }; errdefer allocator.destroy(reader_state); + // Both workers share a nonblocking master. Never toggle descriptor flags + // around individual writes: that races the reader thread. + const flags = c.fcntl(master, c.F_GETFL, @as(c_int, 0)); + if (flags < 0 or c.fcntl(master, c.F_SETFL, flags | c.O_NONBLOCK) < 0) return error.WriteFailed; + const writer_state = try allocator.create(WriterState); + writer_state.* = .{}; + errdefer allocator.destroy(writer_state); + var pty = PosixPty{ .allocator = allocator, .fd = master, .pid = pid, .reader_state = reader_state, + .writer_state = writer_state, }; + pty.writer_thread = try std.Thread.spawn(.{}, writerLoop, .{ pty.fd, writer_state }); + errdefer { + writer_state.mutex.lock(); + writer_state.closing = true; + writer_state.ready.broadcast(); + writer_state.mutex.unlock(); + pty.writer_thread.?.join(); + } pty.reader_thread = try std.Thread.spawn(.{}, readerLoop, .{ pty.fd, pty.reader_state }); return pty; @@ -255,52 +285,31 @@ pub const PosixPty = struct { return self.reader_state.eof or self.reader_state.out_of_memory or self.reader_state.buf.items.len > self.reader_state.start; } + /// Enqueue input without waiting for the child to read it. Admission is + /// all-or-nothing so callers never retry a partially accepted paste. pub fn writeAll(self: *PosixPty, bytes: []const u8) !void { - var offset: usize = 0; - while (offset < bytes.len) { - const written = c.write(self.fd, bytes.ptr + offset, bytes.len - offset); - if (written < 0) { - switch (std.posix.errno(-1)) { - .AGAIN => continue, - else => return error.WriteFailed, - } - } - offset += @intCast(written); + const state = self.writer_state; + state.mutex.lock(); + defer state.mutex.unlock(); + if (state.closing or state.failed) return error.WriteFailed; + const pending = state.buf.items.len - state.start; + if (bytes.len > WRITER_HIGH_WATER_BYTES - pending) return error.InputQueueFull; + if (state.start > 0 and state.buf.capacity - state.buf.items.len < bytes.len) { + std.mem.copyForwards(u8, state.buf.items[0..pending], state.buf.items[state.start..]); + state.buf.items.len = pending; + state.start = 0; } + try state.buf.appendSlice(std.heap.page_allocator, bytes); + state.ready.signal(); } pub fn writeAllUntil(self: *PosixPty, bytes: []const u8, deadline_ns: i128) !usize { - const flags = c.fcntl(self.fd, c.F_GETFL, @as(c_int, 0)); - if (flags < 0 or c.fcntl(self.fd, c.F_SETFL, flags | c.O_NONBLOCK) < 0) return error.WriteFailed; - defer _ = c.fcntl(self.fd, c.F_SETFL, flags); - - var offset: usize = 0; - while (offset < bytes.len) { - if (io.nanoTimestamp() >= deadline_ns) break; - const result = c.write(self.fd, bytes.ptr + offset, @min(bytes.len - offset, 4096)); - if (result > 0) { - offset += @intCast(result); - continue; - } - if (result == 0) return error.WriteFailed; - switch (std.posix.errno(-1)) { - .AGAIN => { - var poll_fd = c.struct_pollfd{ - .fd = self.fd, - .events = c.POLLOUT, - .revents = 0, - }; - const remaining_ns = deadline_ns - io.nanoTimestamp(); - if (remaining_ns <= 0) break; - const timeout_ms: c_int = @intCast(@divFloor(remaining_ns, std.time.ns_per_ms)); - const ready = c.poll(&poll_fd, 1, timeout_ms); - if (ready == 0) break; - if (ready < 0 and std.posix.errno(-1) != .INTR) return error.WriteFailed; - }, - else => return error.WriteFailed, - } - } - return offset; + if (io.nanoTimestamp() >= deadline_ns) return 0; + self.writeAll(bytes) catch |err| { + if (err == error.InputQueueFull) return 0; + return err; + }; + return bytes.len; } pub fn resize(self: *PosixPty, cols: u16, rows: u16) void { @@ -321,8 +330,15 @@ pub const PosixPty = struct { self.reader_state.ready.broadcast(); self.reader_state.mutex.unlock(); if (self.isAlive()) _ = c.kill(self.pid, c.SIGTERM); - _ = c.close(self.fd); + self.writer_state.mutex.lock(); + self.writer_state.closing = true; + self.writer_state.ready.broadcast(); + self.writer_state.mutex.unlock(); + if (self.writer_thread) |thread| thread.join(); if (self.reader_thread) |thread| thread.join(); + _ = c.close(self.fd); + self.writer_state.buf.deinit(std.heap.page_allocator); + self.allocator.destroy(self.writer_state); self.reader_state.buf.deinit(std.heap.page_allocator); self.allocator.destroy(self.reader_state); self.closed = true; @@ -497,3 +513,77 @@ fn readerLoop(fd: c_int, reader_state: *ReaderState) void { } } } + +fn writerLoop(fd: c_int, state: *WriterState) void { + var chunk: [16 * 1024]u8 = undefined; + while (true) { + state.mutex.lock(); + while (!state.closing and state.start == state.buf.items.len) state.ready.wait(&state.mutex); + if (state.closing) { + state.mutex.unlock(); + return; + } + const count = @min(chunk.len, state.buf.items.len - state.start); + @memcpy(chunk[0..count], state.buf.items[state.start..][0..count]); + state.mutex.unlock(); + + var poll_fd = c.struct_pollfd{ .fd = fd, .events = c.POLLOUT, .revents = 0 }; + const ready = c.poll(&poll_fd, 1, 25); + if (ready == 0) continue; + if (ready < 0 and std.posix.errno(-1) == .INTR) continue; + const written = if (ready > 0) c.write(fd, &chunk, count) else -1; + if (written < 0) switch (std.posix.errno(-1)) { + .AGAIN, .INTR => continue, + else => {}, + }; + state.mutex.lock(); + if (written <= 0) { + state.failed = true; + state.mutex.unlock(); + return; + } + state.start += @intCast(written); + if (state.start == state.buf.items.len) { + state.buf.clearRetainingCapacity(); + state.start = 0; + } + state.mutex.unlock(); + } +} + +test "queued input is bounded and admission preserves existing bytes" { + var state = WriterState{}; + defer state.buf.deinit(std.heap.page_allocator); + var pty = PosixPty{ .allocator = std.testing.allocator, .fd = -1, .pid = 0, .reader_state = undefined, .writer_state = &state }; + try pty.writeAll("first"); + const oversized = try std.testing.allocator.alloc(u8, WRITER_HIGH_WATER_BYTES); + defer std.testing.allocator.free(oversized); + try std.testing.expectError(error.InputQueueFull, pty.writeAll(oversized)); + try pty.writeAll("second"); + try std.testing.expectEqualStrings("firstsecond", state.buf.items); + try std.testing.expectEqual(@as(usize, 0), try pty.writeAllUntil("late", io.nanoTimestamp() - 1)); + state.closing = true; + try std.testing.expectError(error.WriteFailed, pty.writeAll("closed")); +} + +test "writer shutdown completes when child pipe is full" { + var fds: [2]c_int = undefined; + try std.testing.expectEqual(@as(c_int, 0), c.pipe(&fds)); + defer _ = c.close(fds[0]); + defer _ = c.close(fds[1]); + const flags = c.fcntl(fds[1], c.F_GETFL, @as(c_int, 0)); + try std.testing.expect(c.fcntl(fds[1], c.F_SETFL, flags | c.O_NONBLOCK) >= 0); + const filler = [_]u8{0} ** 4096; + while (c.write(fds[1], &filler, filler.len) > 0) {} + var state = WriterState{}; + defer state.buf.deinit(std.heap.page_allocator); + var pty = PosixPty{ .allocator = std.testing.allocator, .fd = fds[1], .pid = 0, .reader_state = undefined, .writer_state = &state }; + try pty.writeAll("queued behind blocked pipe"); + const worker = try std.Thread.spawn(.{}, writerLoop, .{ fds[1], &state }); + state.mutex.lock(); + state.closing = true; + state.ready.broadcast(); + state.mutex.unlock(); + worker.join(); + try std.testing.expectEqualStrings("queued behind blocked pipe", state.buf.items); +} diff --git a/src/render/atlas.zig b/src/render/atlas.zig index 9a6dc19..aeddaf6 100644 --- a/src/render/atlas.zig +++ b/src/render/atlas.zig @@ -341,6 +341,8 @@ fn clearUvCaches(self: *FtRenderer) void { self.allocator.free(val.glyphs); } self.prepared_cache.clearRetainingCapacity(); + self.prepared_cache_bytes = 0; + self.prepared_cache_fifo.clear(); self.recent_prepared = [_]?RecentPreparedEntry{null} ** RECENT_PREPARED_CACHE_LEN; self.ascii_glyphs = [_][256]?Glyph{[_]?Glyph{null} ** 256} ** 4; self.prepared_glyphs.clearRetainingCapacity(); diff --git a/src/render/cache_fifo.zig b/src/render/cache_fifo.zig new file mode 100644 index 0000000..5ad46f0 --- /dev/null +++ b/src/render/cache_fifo.zig @@ -0,0 +1,57 @@ +const std = @import("std"); + +/// Insertion order for bounded caches. Amortized constant-time eviction avoids +/// rescanning the hash table from its first bucket on every cache miss. +pub fn Fifo(comptime Key: type) type { + return struct { + keys: std.ArrayListUnmanaged(Key) = .empty, + head: usize = 0, + + pub fn reserve(self: *@This(), allocator: std.mem.Allocator) !void { + if (self.head > 0 and self.head >= self.keys.items.len / 2) { + const remaining = self.keys.items.len - self.head; + std.mem.copyForwards(Key, self.keys.items[0..remaining], self.keys.items[self.head..]); + self.keys.items.len = remaining; + self.head = 0; + } + try self.keys.ensureUnusedCapacity(allocator, 1); + } + + pub fn push(self: *@This(), key: Key) void { + self.keys.appendAssumeCapacity(key); + } + + pub fn pop(self: *@This()) ?Key { + if (self.head == self.keys.items.len) return null; + const key = self.keys.items[self.head]; + self.head += 1; + return key; + } + + pub fn clear(self: *@This()) void { + self.keys.clearRetainingCapacity(); + self.head = 0; + } + + pub fn deinit(self: *@This(), allocator: std.mem.Allocator) void { + self.keys.deinit(allocator); + } + }; +} + +test "cache insertion order survives repeated compaction" { + var fifo = Fifo(usize){}; + defer fifo.deinit(std.testing.allocator); + for (0..8) |i| { + try fifo.reserve(std.testing.allocator); + fifo.push(i); + } + for (8..10000) |i| { + try std.testing.expectEqual(i - 8, fifo.pop().?); + try fifo.reserve(std.testing.allocator); + fifo.push(i); + } + try std.testing.expect(fifo.keys.items.len <= 16); + fifo.clear(); + try std.testing.expect(fifo.pop() == null); +} diff --git a/src/render/ft_renderer.zig b/src/render/ft_renderer.zig index 21150f7..fea78ba 100644 --- a/src/render/ft_renderer.zig +++ b/src/render/ft_renderer.zig @@ -209,6 +209,14 @@ pub const FtRenderer = struct { glyph_cache: std.HashMap(GlyphKey, Glyph, GlyphCacheContext, std.hash_map.default_max_load_percentage), // Shaping cache + // Owned glyph storage plus a conservative allowance for hash tables and FIFO keys. + shape_cache_fifo: @import("cache_fifo.zig").Fifo(ShapeKey) = .{}, + prepared_cache_fifo: @import("cache_fifo.zig").Fifo(PreparedKey) = .{}, + shape_cache_bytes: usize = 0, + prepared_cache_bytes: usize = 0, + text_cache_limit_bytes: usize = 8 * 1024 * 1024, + shape_cache_evictions: usize = 0, + prepared_cache_evictions: usize = 0, shape_cache: std.HashMap(ShapeKey, ShapeResult, ShapeCacheContext, std.hash_map.default_max_load_percentage), // Prepared run cache @@ -738,11 +746,13 @@ pub const FtRenderer = struct { self.allocator.free(val.glyphs); } self.prepared_cache.deinit(); + self.prepared_cache_fifo.deinit(self.allocator); var it = self.shape_cache.valueIterator(); while (it.next()) |val| { self.allocator.free(val.glyphs); } self.shape_cache.deinit(); + self.shape_cache_fifo.deinit(self.allocator); self.glyph_cache.deinit(); atlas_mod.deinitPages(self); c.sg_destroy_sampler(self.atlas_smp); diff --git a/src/render/shaping.zig b/src/render/shaping.zig index a7cc7e9..d958587 100644 --- a/src/render/shaping.zig +++ b/src/render/shaping.zig @@ -70,12 +70,27 @@ pub fn putPreparedCache(self: *FtRenderer, utf8: []const u8, face_idx: u8, raste self.putRecentPrepared(key, fingerprint, entry.glyphs); return; } + const cost = glyphs.len * @sizeOf(PreparedGlyph) + 6 * @sizeOf(PreparedKey); + if (cost > self.text_cache_limit_bytes) return; + while (self.prepared_cache_bytes + cost > self.text_cache_limit_bytes) { + const victim_key = self.prepared_cache_fifo.pop() orelse break; + const victim = self.prepared_cache.fetchRemove(victim_key) orelse continue; + const victim_glyphs = victim.value.glyphs; + self.prepared_cache_bytes -= victim_glyphs.len * @sizeOf(PreparedGlyph) + 6 * @sizeOf(PreparedKey); + // Recent entries borrow storage from this map. + self.recent_prepared = [_]?RecentPreparedEntry{null} ** RECENT_PREPARED_CACHE_LEN; + self.allocator.free(victim_glyphs); + self.prepared_cache_evictions += 1; + } + self.prepared_cache_fifo.reserve(self.allocator) catch return; const owned = self.allocator.alloc(PreparedGlyph, glyphs.len) catch return; fastmem.copy(PreparedGlyph, owned, glyphs); self.prepared_cache.put(key, .{ .glyphs = owned }) catch { self.allocator.free(owned); return; }; + self.prepared_cache_fifo.push(key); + self.prepared_cache_bytes += cost; self.putRecentPrepared(key, fingerprint, owned); } @@ -211,6 +226,17 @@ pub fn getOrShape(self: *FtRenderer, utf8: []const u8, face_idx: u8) ?ShapeResul const positions = ft.hb_buffer_get_glyph_positions(buf, &pos_len); if (infos == null or positions == null) return null; + const cost = @as(usize, info_len) * @sizeOf(GlyphInstance) + 6 * @sizeOf(ShapeKey); + // Keep at least one run even when a diagnostic budget is very small. + while (self.shape_cache_bytes + cost > self.text_cache_limit_bytes) { + const victim_key = self.shape_cache_fifo.pop() orelse break; + const victim = self.shape_cache.fetchRemove(victim_key) orelse continue; + const victim_glyphs = victim.value.glyphs; + self.shape_cache_bytes -= victim_glyphs.len * @sizeOf(GlyphInstance) + 6 * @sizeOf(ShapeKey); + self.allocator.free(victim_glyphs); + self.shape_cache_evictions += 1; + } + self.shape_cache_fifo.reserve(self.allocator) catch return null; const glyphs = self.allocator.alloc(GlyphInstance, info_len) catch return null; var i: usize = 0; while (i < info_len) : (i += 1) { @@ -227,6 +253,8 @@ pub fn getOrShape(self: *FtRenderer, utf8: []const u8, face_idx: u8) ?ShapeResul self.allocator.free(glyphs); return null; }; + self.shape_cache_fifo.push(key); + self.shape_cache_bytes += cost; return res; } diff --git a/src/render/sokol_runtime.zig b/src/render/sokol_runtime.zig index 4b02015..7338b3e 100644 --- a/src/render/sokol_runtime.zig +++ b/src/render/sokol_runtime.zig @@ -4789,7 +4789,7 @@ fn handleChar(app: *App, event: c.sapp_event) void { c.sapp_consume_event(); return; } - const mods = ghosttyMods(event.modifiers); + const mods = altGrAdjustedMods(ghosttyMods(event.modifiers)); if (copy_mode.copyModeActive(app) and (mods & ghostty.Mods.ctrl) != 0 and utf8.len == 1 and utf8[0] == 0x16) { _ = app.enqueueMouse(.{ .copy_mode_begin_selection = true }); c.sapp_consume_event(); @@ -5476,6 +5476,21 @@ fn modifierBitForKey(key: ghostty.Key) u32 { }; } +/// Strips the synthetic Ctrl+Alt bits that Windows reports while AltGr is +/// held (and the Alt bit some X11 layouts put on ISO_Level3_Shift), so +/// AltGr-produced characters are delivered as plain text. Only applies +/// when the physical right-hand Alt key is down; real Ctrl/Alt chords with +/// the left-hand keys are untouched. +pub fn altGrAdjustedModsFor(mods: u32, alt_right_down: bool) u32 { + if (!alt_right_down) return mods; + if ((mods & (ghostty.Mods.ctrl | ghostty.Mods.alt)) == 0) return mods; + return mods & ~@as(u32, ghostty.Mods.ctrl | ghostty.Mods.alt); +} + +fn altGrAdjustedMods(mods: u32) u32 { + return altGrAdjustedModsFor(mods, g_right_alt_down); +} + fn ghosttyMods(modifiers: u32) u32 { var mods: u32 = ghostty.Mods.none; if ((modifiers & c.SAPP_MODIFIER_SHIFT) != 0) mods |= ghostty.Mods.shift; @@ -5675,3 +5690,17 @@ fn drawBorderRect(x: f32, y: f32, w: f32, h: f32, r: u8, g: u8, b: u8, a: u8) vo c.sgl_end(); c.sgl_load_default_pipeline(); } + +test "altGrAdjustedModsFor strips synthetic ctrl+alt only while right alt is down" { + const ctrl_alt = ghostty.Mods.ctrl | ghostty.Mods.alt; + const ctrl_shift = ghostty.Mods.ctrl | ghostty.Mods.shift; + const shift_ctrl_alt = ghostty.Mods.shift | ctrl_alt; + + try std.testing.expectEqual(@as(u32, 0), altGrAdjustedModsFor(ctrl_alt, true)); + try std.testing.expectEqual(@as(u32, ghostty.Mods.shift), altGrAdjustedModsFor(shift_ctrl_alt, true)); + try std.testing.expectEqual(@as(u32, ghostty.Mods.shift | ghostty.Mods.ctrl), altGrAdjustedModsFor(ctrl_shift, false)); + + try std.testing.expectEqual(@as(u32, ghostty.Mods.shift), altGrAdjustedModsFor(ghostty.Mods.shift, true)); + try std.testing.expectEqual(@as(u32, 0), altGrAdjustedModsFor(0, true)); + try std.testing.expectEqual(@as(u32, 0), altGrAdjustedModsFor(0, false)); +} diff --git a/src/renderer_bench_test.zig b/src/renderer_bench_test.zig index b4f8780..87496ab 100644 --- a/src/renderer_bench_test.zig +++ b/src/renderer_bench_test.zig @@ -9,3 +9,7 @@ test "renderer benchmark corpus integration" { test "renderer handles oversized grapheme clusters" { try benchmark.runUnicodeGraphemeTest(std.testing.allocator); } + +test "text caches remain bounded under unique output" { + try benchmark.runCachePressureTest(std.testing.allocator); +} diff --git a/types/hollow.lua b/types/hollow.lua index a708b6a..9b78c8d 100644 --- a/types/hollow.lua +++ b/types/hollow.lua @@ -1112,34 +1112,28 @@ ---@field pane HollowPane ---@field payload any ----@class HollowProcessWriter ----@field write fun(data: string) - ----@class HollowProcessReader ----@field read fun(): string|nil - ---@class HollowProcess ----@field pid integer ----@field stdin HollowProcessWriter ----@field stdout HollowProcessReader ----@field stderr HollowProcessReader ----@field wait fun(): integer ----@field kill fun() - ----@class HollowExecResult ----@field exit_code integer ----@field stdout string ----@field stderr string +---@field status fun(self: HollowProcess): "running"|"finished" +---@field result fun(self: HollowProcess): HollowProcessRunResult|nil +---@field cancel fun(self: HollowProcess) +---@field kill fun(self: HollowProcess) Alias for cancel +---@field wait fun(self: HollowProcess): HollowProcessRunResult Coroutine only +---@field next fun(self: HollowProcess, callback: fun(result: HollowProcessRunResult)): HollowPromise ---@class HollowProcessRunResult ---@field code integer ---@field stdout string ---@field stderr string +---@field error? string Spawn, timeout, cancellation, or output limit failure +---@field canceled? boolean ---@class HollowProcessOpts ----@field cmd string|string[] +---@field cmd string|string[] Executable or argv; no implicit shell parsing ---@field cwd? string ----@field env? table +---@field env? table Overrides inherited environment +---@field timeout_ms? integer Default 30000; maximum 86400000 +---@field output_limit? integer Per stream bytes; default 1 MiB, maximum 16 MiB +---@field on_complete? fun(result: HollowProcessRunResult) ---@class HollowProcessRunOpts ---@field hide_window? boolean Defaults to true on the host bridge @@ -1884,7 +1878,7 @@ local process = {} function process.spawn(opts) end ---@param opts HollowProcessOpts ----@return HollowExecResult +---@return HollowPromise function process.exec(opts) end ---@param args string[]