Merge pull request #1948 from ghostty-org/termio

Termio Refactor

src/Surface.zig

@@ -101,7 +101,7 @@ color_scheme: apprt.ColorScheme = .light,
 last_binding_trigger: u64 = 0,
 
 /// The terminal IO handler.
-io: termio.Impl,
+io: termio.Termio,
 io_thread: termio.Thread,
 io_thr: std.Thread,
 
@@ -396,33 +396,8 @@ pub fn init(
     );
     errdefer render_thread.deinit();
 
-    // Start our IO implementation
-    var io = try termio.Impl.init(alloc, .{
-        .grid_size = grid_size,
-        .screen_size = screen_size,
-        .padding = padding,
-        .full_config = config,
-        .config = try termio.Impl.DerivedConfig.init(alloc, config),
-        .resources_dir = main.state.resources_dir,
-        .renderer_state = &self.renderer_state,
-        .renderer_wakeup = render_thread.wakeup,
-        .renderer_mailbox = render_thread.mailbox,
-        .surface_mailbox = .{ .surface = self, .app = app_mailbox },
-
-        // Get the cgroup if we're on linux and have the decl. I'd love
-        // to change this from a decl to a surface options struct because
-        // then we can do memory management better (don't need to retain
-        // the string around).
-        .linux_cgroup = if (comptime builtin.os.tag == .linux and
-            @hasDecl(apprt.runtime.Surface, "cgroup"))
-            rt_surface.cgroup()
-        else
-            Command.linux_cgroup_default,
-    });
-    errdefer io.deinit();
-
     // Create the IO thread
-    var io_thread = try termio.Thread.init(alloc, &self.io);
+    var io_thread = try termio.Thread.init(alloc);
     errdefer io_thread.deinit();
 
     self.* = .{
@@ -440,7 +415,7 @@ pub fn init(
         },
         .renderer_thr = undefined,
         .mouse = .{},
-        .io = io,
+        .io = undefined,
         .io_thread = io_thread,
         .io_thr = undefined,
         .screen_size = .{ .width = 0, .height = 0 },
@@ -450,6 +425,53 @@ pub fn init(
         .config = derived_config,
     };
 
+    // Start our IO implementation
+    // This separate block ({}) is important because our errdefers must
+    // be scoped here to be valid.
+    {
+        // Initialize our IO backend
+        var io_exec = try termio.Exec.init(alloc, .{
+            .command = config.command,
+            .shell_integration = config.@"shell-integration",
+            .shell_integration_features = config.@"shell-integration-features",
+            .working_directory = config.@"working-directory",
+            .resources_dir = main.state.resources_dir,
+            .term = config.term,
+
+            // Get the cgroup if we're on linux and have the decl. I'd love
+            // to change this from a decl to a surface options struct because
+            // then we can do memory management better (don't need to retain
+            // the string around).
+            .linux_cgroup = if (comptime builtin.os.tag == .linux and
+                @hasDecl(apprt.runtime.Surface, "cgroup"))
+                rt_surface.cgroup()
+            else
+                Command.linux_cgroup_default,
+        });
+        errdefer io_exec.deinit();
+
+        // Initialize our IO mailbox
+        var io_mailbox = try termio.Mailbox.initSPSC(alloc);
+        errdefer io_mailbox.deinit(alloc);
+
+        try termio.Termio.init(&self.io, alloc, .{
+            .grid_size = grid_size,
+            .screen_size = screen_size,
+            .padding = padding,
+            .full_config = config,
+            .config = try termio.Termio.DerivedConfig.init(alloc, config),
+            .backend = .{ .exec = io_exec },
+            .mailbox = io_mailbox,
+            .renderer_state = &self.renderer_state,
+            .renderer_wakeup = render_thread.wakeup,
+            .renderer_mailbox = render_thread.mailbox,
+            .surface_mailbox = .{ .surface = self, .app = app_mailbox },
+        });
+    }
+    // Outside the block, IO has now taken ownership of our temporary state
+    // so we can just defer this and not the subcomponents.
+    errdefer self.io.deinit();
+
     // Report initial cell size on surface creation
     try rt_surface.setCellSize(cell_size.width, cell_size.height);
 
@@ -483,7 +505,7 @@ pub fn init(
     self.io_thr = try std.Thread.spawn(
         .{},
         termio.Thread.threadMain,
-        .{&self.io_thread},
+        .{ &self.io_thread, &self.io },
     );
     self.io_thr.setName("io") catch {};
 
@@ -616,7 +638,7 @@ pub fn activateInspector(self: *Surface) !void {
 
     // Notify our components we have an inspector active
     _ = self.renderer_thread.mailbox.push(.{ .inspector = true }, .{ .forever = {} });
-    _ = self.io_thread.mailbox.push(.{ .inspector = true }, .{ .forever = {} });
+    self.io.queueMessage(.{ .inspector = true }, .unlocked);
 }
 
 /// Deactivate the inspector and stop collecting any information.
@@ -633,7 +655,7 @@ pub fn deactivateInspector(self: *Surface) void {
 
     // Notify our components we have deactivated inspector
     _ = self.renderer_thread.mailbox.push(.{ .inspector = false }, .{ .forever = {} });
-    _ = self.io_thread.mailbox.push(.{ .inspector = false }, .{ .forever = {} });
+    self.io.queueMessage(.{ .inspector = false }, .unlocked);
 
     // Deinit the inspector
     insp.deinit();
@@ -733,8 +755,7 @@ fn reportColorScheme(self: *Surface) !void {
         .dark => "\x1B[?997;1n",
     };
 
-    _ = self.io_thread.mailbox.push(.{ .write_stable = output }, .{ .forever = {} });
+    self.io.queueMessage(.{ .write_stable = output }, .unlocked);
-    try self.io_thread.wakeup.notify();
 }
 
 /// Call this when modifiers change. This is safe to call even if modifiers
@@ -809,26 +830,23 @@ fn changeConfig(self: *Surface, config: *const configpkg.Config) !void {
     // our messages aren't huge.
     var renderer_message = try renderer.Message.initChangeConfig(self.alloc, config);
     errdefer renderer_message.deinit();
-    var termio_config_ptr = try self.alloc.create(termio.Impl.DerivedConfig);
+    var termio_config_ptr = try self.alloc.create(termio.Termio.DerivedConfig);
     errdefer self.alloc.destroy(termio_config_ptr);
-    termio_config_ptr.* = try termio.Impl.DerivedConfig.init(self.alloc, config);
+    termio_config_ptr.* = try termio.Termio.DerivedConfig.init(self.alloc, config);
     errdefer termio_config_ptr.deinit();
 
     _ = self.renderer_thread.mailbox.push(renderer_message, .{ .forever = {} });
-    _ = self.io_thread.mailbox.push(.{
+    self.io.queueMessage(.{
         .change_config = .{
             .alloc = self.alloc,
             .ptr = termio_config_ptr,
         },
-    }, .{ .forever = {} });
+    }, .unlocked);
 
     // With mailbox messages sent, we have to wake them up so they process it.
     self.queueRender() catch |err| {
         log.warn("failed to notify renderer of config change err={}", .{err});
     };
-    self.io_thread.wakeup.notify() catch |err| {
-        log.warn("failed to notify io thread of config change err={}", .{err});
-    };
 }
 
 /// Returns true if the terminal has a selection.
@@ -1066,14 +1084,13 @@ fn setCellSize(self: *Surface, size: renderer.CellSize) !void {
     );
 
     // Notify the terminal
-    _ = self.io_thread.mailbox.push(.{
+    self.io.queueMessage(.{
         .resize = .{
             .grid_size = self.grid_size,
             .screen_size = self.screen_size,
             .padding = self.padding,
         },
-    }, .{ .forever = {} });
+    }, .unlocked);
-    self.io_thread.wakeup.notify() catch {};
 
     // Notify the window
     try self.rt_surface.setCellSize(size.width, size.height);
@@ -1169,14 +1186,13 @@ fn resize(self: *Surface, size: renderer.ScreenSize) !void {
     }
 
     // Mail the IO thread
-    _ = self.io_thread.mailbox.push(.{
+    self.io.queueMessage(.{
         .resize = .{
             .grid_size = self.grid_size,
             .screen_size = self.screen_size,
             .padding = self.padding,
         },
-    }, .{ .forever = {} });
+    }, .unlocked);
-    try self.io_thread.wakeup.notify();
 }
 
 /// Called to set the preedit state for character input. Preedit is used
@@ -1542,12 +1558,11 @@ pub fn keyCallback(
         ev.pty = copy;
     }
 
-    _ = self.io_thread.mailbox.push(switch (write_req) {
+    self.io.queueMessage(switch (write_req) {
         .small => |v| .{ .write_small = v },
         .stable => |v| .{ .write_stable = v },
         .alloc => |v| .{ .write_alloc = v },
-    }, .{ .forever = {} });
+    }, .unlocked);
-    try self.io_thread.wakeup.notify();
 
     // If our event is any keypress that isn't a modifier and we generated
     // some data to send to the pty, then we move the viewport down to the
@@ -1647,11 +1662,7 @@ pub fn focusCallback(self: *Surface, focused: bool) !void {
 
         if (focus_event) {
             const seq = if (focused) "\x1b[I" else "\x1b[O";
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{ .write_stable = seq }, .unlocked);
-                .write_stable = seq,
-            }, .{ .forever = {} });
-
-            try self.io_thread.wakeup.notify();
         }
     }
 }
@@ -1786,14 +1797,10 @@ pub fn scrollCallback(
                     break :seq if (y.delta < 0) "\x1b[A" else "\x1b[B";
                 };
                 for (0..y.delta_unsigned) |_| {
-                    _ = self.io_thread.mailbox.push(.{
+                    self.io.queueMessage(.{ .write_stable = seq }, .locked);
-                        .write_stable = seq,
-                    }, .{ .instant = {} });
                 }
             }
 
-            // After sending all our messages we have to notify our IO thread
-            try self.io_thread.wakeup.notify();
             return;
         }
 
@@ -1995,12 +2002,10 @@ fn mouseReport(
             data[5] = 32 + @as(u8, @intCast(viewport_point.y)) + 1;
 
             // Ask our IO thread to write the data
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{ .write_small = .{
-                .write_small = .{
+                .data = data,
-                    .data = data,
+                .len = 6,
-                    .len = 6,
+            } }, .locked);
-                },
-            }, .{ .forever = {} });
         },
 
         .utf8 => {
@@ -2020,12 +2025,10 @@ fn mouseReport(
             i += try std.unicode.utf8Encode(@intCast(32 + viewport_point.y + 1), data[i..]);
 
             // Ask our IO thread to write the data
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{ .write_small = .{
-                .write_small = .{
+                .data = data,
-                    .data = data,
+                .len = @intCast(i),
-                    .len = @intCast(i),
+            } }, .locked);
-                },
-            }, .{ .forever = {} });
         },
 
         .sgr => {
@@ -2043,12 +2046,10 @@ fn mouseReport(
             });
 
             // Ask our IO thread to write the data
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{ .write_small = .{
-                .write_small = .{
+                .data = data,
-                    .data = data,
+                .len = @intCast(resp.len),
-                    .len = @intCast(resp.len),
+            } }, .locked);
-                },
-            }, .{ .forever = {} });
         },
 
         .urxvt => {
@@ -2062,12 +2063,10 @@ fn mouseReport(
             });
 
             // Ask our IO thread to write the data
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{ .write_small = .{
-                .write_small = .{
+                .data = data,
-                    .data = data,
+                .len = @intCast(resp.len),
-                    .len = @intCast(resp.len),
+            } }, .locked);
-                },
-            }, .{ .forever = {} });
         },
 
         .sgr_pixels => {
@@ -2085,17 +2084,12 @@ fn mouseReport(
             });
 
             // Ask our IO thread to write the data
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{ .write_small = .{
-                .write_small = .{
+                .data = data,
-                    .data = data,
+                .len = @intCast(resp.len),
-                    .len = @intCast(resp.len),
+            } }, .locked);
-                },
-            }, .{ .forever = {} });
         },
     }
-
-    // After sending all our messages we have to notify our IO thread
-    try self.io_thread.wakeup.notify();
 }
 
 /// Returns true if the shift modifier is allowed to be captured by modifier
@@ -2496,9 +2490,7 @@ fn clickMoveCursor(self: *Surface, to: terminal.Pin) !void {
             break :arrow if (t.modes.get(.cursor_keys)) "\x1bOB" else "\x1b[B";
         };
         for (0..@abs(path.y)) |_| {
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{ .write_stable = arrow }, .locked);
-                .write_stable = arrow,
-            }, .{ .instant = {} });
         }
     }
     if (path.x != 0) {
@@ -2508,13 +2500,9 @@ fn clickMoveCursor(self: *Surface, to: terminal.Pin) !void {
             break :arrow if (t.modes.get(.cursor_keys)) "\x1bOC" else "\x1b[C";
         };
         for (0..@abs(path.x)) |_| {
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{ .write_stable = arrow }, .locked);
-                .write_stable = arrow,
-            }, .{ .instant = {} });
         }
     }
-
-    try self.io_thread.wakeup.notify();
 }
 
 /// Returns the link at the given cursor position, if any.
@@ -3188,11 +3176,10 @@ pub fn performBindingAction(self: *Surface, action: input.Binding.Action) !bool
                 .esc => try std.fmt.bufPrint(&buf, "\x1b{s}", .{data}),
                 else => unreachable,
             };
-            _ = self.io_thread.mailbox.push(try termio.Message.writeReq(
+            self.io.queueMessage(try termio.Message.writeReq(
                 self.alloc,
                 full_data,
-            ), .{ .forever = {} });
+            ), .unlocked);
-            try self.io_thread.wakeup.notify();
 
             // CSI/ESC triggers a scroll.
             {
@@ -3216,11 +3203,10 @@ pub fn performBindingAction(self: *Surface, action: input.Binding.Action) !bool
                 );
                 return true;
             };
-            _ = self.io_thread.mailbox.push(try termio.Message.writeReq(
+            self.io.queueMessage(try termio.Message.writeReq(
                 self.alloc,
                 text,
-            ), .{ .forever = {} });
+            ), .unlocked);
-            try self.io_thread.wakeup.notify();
 
             // Text triggers a scroll.
             {
@@ -3250,16 +3236,10 @@ pub fn performBindingAction(self: *Surface, action: input.Binding.Action) !bool
             };
 
             if (normal) {
-                _ = self.io_thread.mailbox.push(.{
+                self.io.queueMessage(.{ .write_stable = ck.normal }, .unlocked);
-                    .write_stable = ck.normal,
-                }, .{ .forever = {} });
             } else {
-                _ = self.io_thread.mailbox.push(.{
+                self.io.queueMessage(.{ .write_stable = ck.application }, .unlocked);
-                    .write_stable = ck.application,
-                }, .{ .forever = {} });
             }
-
-            try self.io_thread.wakeup.notify();
         },
 
         .reset => {
@@ -3341,63 +3321,55 @@ pub fn performBindingAction(self: *Surface, action: input.Binding.Action) !bool
                 if (self.io.terminal.active_screen == .alternate) return false;
             }
 
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{
                 .clear_screen = .{ .history = true },
-            }, .{ .forever = {} });
+            }, .unlocked);
-            try self.io_thread.wakeup.notify();
         },
 
         .scroll_to_top => {
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{
                 .scroll_viewport = .{ .top = {} },
-            }, .{ .forever = {} });
+            }, .unlocked);
-            try self.io_thread.wakeup.notify();
         },
 
         .scroll_to_bottom => {
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{
                 .scroll_viewport = .{ .bottom = {} },
-            }, .{ .forever = {} });
+            }, .unlocked);
-            try self.io_thread.wakeup.notify();
         },
 
         .scroll_page_up => {
             const rows: isize = @intCast(self.grid_size.rows);
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{
                 .scroll_viewport = .{ .delta = -1 * rows },
-            }, .{ .forever = {} });
+            }, .unlocked);
-            try self.io_thread.wakeup.notify();
         },
 
         .scroll_page_down => {
             const rows: isize = @intCast(self.grid_size.rows);
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{
                 .scroll_viewport = .{ .delta = rows },
-            }, .{ .forever = {} });
+            }, .unlocked);
-            try self.io_thread.wakeup.notify();
         },
 
         .scroll_page_fractional => |fraction| {
             const rows: f32 = @floatFromInt(self.grid_size.rows);
             const delta: isize = @intFromFloat(@floor(fraction * rows));
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{
                 .scroll_viewport = .{ .delta = delta },
-            }, .{ .forever = {} });
+            }, .unlocked);
-            try self.io_thread.wakeup.notify();
         },
 
         .scroll_page_lines => |lines| {
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{
                 .scroll_viewport = .{ .delta = lines },
-            }, .{ .forever = {} });
+            }, .unlocked);
-            try self.io_thread.wakeup.notify();
         },
 
         .jump_to_prompt => |delta| {
-            _ = self.io_thread.mailbox.push(.{
+            self.io.queueMessage(.{
                 .jump_to_prompt = @intCast(delta),
-            }, .{ .forever = {} });
+            }, .unlocked);
-            try self.io_thread.wakeup.notify();
         },
 
         .write_scrollback_file => write_scrollback_file: {
@@ -3441,11 +3413,10 @@ pub fn performBindingAction(self: *Surface, action: input.Binding.Action) !bool
             var path_buf: [std.fs.MAX_PATH_BYTES]u8 = undefined;
             const path = try tmp_dir.dir.realpath("scrollback", &path_buf);
 
-            _ = self.io_thread.mailbox.push(try termio.Message.writeReq(
+            self.io.queueMessage(try termio.Message.writeReq(
                 self.alloc,
                 path,
-            ), .{ .forever = {} });
+            ), .unlocked);
-            try self.io_thread.wakeup.notify();
         },
 
         .new_window => try self.app.newWindow(self.rt_app, .{ .parent = self }),
@@ -3700,16 +3671,16 @@ fn completeClipboardPaste(
     if (critical.bracketed) {
         // If we're bracketd we write the data as-is to the terminal with
         // the bracketed paste escape codes around it.
-        _ = self.io_thread.mailbox.push(.{
+        self.io.queueMessage(.{
             .write_stable = "\x1B[200~",
-        }, .{ .forever = {} });
+        }, .unlocked);
-        _ = self.io_thread.mailbox.push(try termio.Message.writeReq(
+        self.io.queueMessage(try termio.Message.writeReq(
             self.alloc,
             data,
-        ), .{ .forever = {} });
+        ), .unlocked);
-        _ = self.io_thread.mailbox.push(.{
+        self.io.queueMessage(.{
             .write_stable = "\x1B[201~",
-        }, .{ .forever = {} });
+        }, .unlocked);
     } else {
         // If its not bracketed the input bytes are indistinguishable from
         // keystrokes, so we must be careful. For example, we must replace
@@ -3736,13 +3707,11 @@ fn completeClipboardPaste(
             len += 1;
         }
 
-        _ = self.io_thread.mailbox.push(try termio.Message.writeReq(
+        self.io.queueMessage(try termio.Message.writeReq(
             self.alloc,
             buf[0..len],
-        ), .{ .forever = {} });
+        ), .unlocked);
     }
-
-    try self.io_thread.wakeup.notify();
 }
 
 fn completeClipboardReadOSC52(
@@ -3784,11 +3753,10 @@ fn completeClipboardReadOSC52(
     const encoded = enc.encode(buf[prefix.len..], data);
     assert(encoded.len == size);
 
-    _ = self.io_thread.mailbox.push(try termio.Message.writeReq(
+    self.io.queueMessage(try termio.Message.writeReq(
         self.alloc,
         buf,
-    ), .{ .forever = {} });
+    ), .unlocked);
-    self.io_thread.wakeup.notify() catch {};
 }
 
 fn showDesktopNotification(self: *Surface, title: [:0]const u8, body: [:0]const u8) !void {

src/termio.zig

@@ -1,18 +1,35 @@
-//! IO implementation and utilities. The IO implementation is responsible
+//! Termio is responsible for "terminal IO." Specifically, this is the
-//! for taking the config, spinning up a child process, and handling IO
+//! reading and writing of bytes for the underlying pty or pty-like device.
-//! with the terminal.
+//!
+//! Termio is constructed of a few components:
+//!   - Termio - The main shared struct that has common logic across all
+//!     backends and mailboxes (defined below).
+//!   - Backend - Responsible for the actual physical IO. For example, one
+//!     implementation creates a subprocess, allocates and assigns a pty,
+//!     and sets up a read thread on the pty.
+//!   - Mailbox - Responsible for storing/dispensing event messages to
+//!     the backend. This exists separately from backends because termio
+//!     is built to be both single and multi-threaded.
+//!
+//! Termio supports (and recommends) multi-threaded operation. Multi-threading
+//! enables the read/writes to generally happen on separate threads and
+//! almost always improves throughput and latency under heavy IO load. To
+//! enable threading, use the Thread struct. This wraps a Termio, requires
+//! specific backend/mailbox capabilities, and sets up the necessary threads.
+
+const stream_handler = @import("termio/stream_handler.zig");
 
 pub usingnamespace @import("termio/message.zig");
+pub const backend = @import("termio/backend.zig");
+pub const mailbox = @import("termio/mailbox.zig");
 pub const Exec = @import("termio/Exec.zig");
 pub const Options = @import("termio/Options.zig");
+pub const Termio = @import("termio/Termio.zig");
 pub const Thread = @import("termio/Thread.zig");
-pub const Mailbox = Thread.Mailbox;
+pub const Backend = backend.Backend;
-
+pub const DerivedConfig = Termio.DerivedConfig;
-/// The implementation to use for the IO. This is just "exec" for now but
+pub const Mailbox = mailbox.Mailbox;
-/// this is somewhat pluggable so that in the future we can introduce other
+pub const StreamHandler = stream_handler.StreamHandler;
-/// options for other platforms (i.e. wasm) or even potentially a vtable
-/// implementation for runtime polymorphism.
-pub const Impl = Exec;
 
 test {
     @import("std").testing.refAllDecls(@This());

src/termio/Options.zig

@@ -23,10 +23,14 @@ padding: renderer.Padding,
 full_config: *const Config,
 
 /// The derived configuration for this termio implementation.
-config: termio.Impl.DerivedConfig,
+config: termio.Termio.DerivedConfig,
 
-/// The application resources directory.
+/// The backend for termio that implements where reads/writes are sourced.
-resources_dir: ?[]const u8,
+backend: termio.Backend,
+
+/// The mailbox for the terminal. This is how messages are delivered.
+/// If you're using termio.Thread this MUST be "mailbox".
+mailbox: termio.Mailbox,
 
 /// The render state. The IO implementation can modify anything here. The
 /// surface thread will setup the initial "terminal" pointer but the IO impl
@@ -43,7 +47,3 @@ renderer_mailbox: *renderer.Thread.Mailbox,
 
 /// The mailbox for sending the surface messages.
 surface_mailbox: apprt.surface.Mailbox,
-
-/// The cgroup to apply to the started termio process, if able by
-/// the termio implementation. This only applies to Linux.
-linux_cgroup: Command.LinuxCgroup = Command.linux_cgroup_default,

src/termio/Termio.zig

@@ -0,0 +1,548 @@
+//! Primary terminal IO ("termio") state. This maintains the terminal state,
+//! pty, subprocess, etc. This is flexible enough to be used in environments
+//! that don't have a pty and simply provides the input/output using raw
+//! bytes.
+pub const Termio = @This();
+
+const std = @import("std");
+const builtin = @import("builtin");
+const build_config = @import("../build_config.zig");
+const assert = std.debug.assert;
+const Allocator = std.mem.Allocator;
+const ArenaAllocator = std.heap.ArenaAllocator;
+const EnvMap = std.process.EnvMap;
+const posix = std.posix;
+const termio = @import("../termio.zig");
+const Command = @import("../Command.zig");
+const Pty = @import("../pty.zig").Pty;
+const SegmentedPool = @import("../segmented_pool.zig").SegmentedPool;
+const StreamHandler = @import("stream_handler.zig").StreamHandler;
+const terminal = @import("../terminal/main.zig");
+const terminfo = @import("../terminfo/main.zig");
+const xev = @import("xev");
+const renderer = @import("../renderer.zig");
+const apprt = @import("../apprt.zig");
+const fastmem = @import("../fastmem.zig");
+const internal_os = @import("../os/main.zig");
+const windows = internal_os.windows;
+const configpkg = @import("../config.zig");
+const shell_integration = @import("shell_integration.zig");
+
+const log = std.log.scoped(.io_exec);
+
+/// Allocator
+alloc: Allocator,
+
+/// This is the implementation responsible for io.
+backend: termio.Backend,
+
+/// The derived configuration for this termio implementation.
+config: DerivedConfig,
+
+/// The terminal emulator internal state. This is the abstract "terminal"
+/// that manages input, grid updating, etc. and is renderer-agnostic. It
+/// just stores internal state about a grid.
+terminal: terminal.Terminal,
+
+/// The shared render state
+renderer_state: *renderer.State,
+
+/// A handle to wake up the renderer. This hints to the renderer that that
+/// a repaint should happen.
+renderer_wakeup: xev.Async,
+
+/// The mailbox for notifying the renderer of things.
+renderer_mailbox: *renderer.Thread.Mailbox,
+
+/// The mailbox for communicating with the surface.
+surface_mailbox: apprt.surface.Mailbox,
+
+/// The cached grid size whenever a resize is called.
+grid_size: renderer.GridSize,
+
+/// The mailbox implementation to use.
+mailbox: termio.Mailbox,
+
+/// The stream parser. This parses the stream of escape codes and so on
+/// from the child process and calls callbacks in the stream handler.
+terminal_stream: terminal.Stream(StreamHandler),
+
+/// Last time the cursor was reset. This is used to prevent message
+/// flooding with cursor resets.
+last_cursor_reset: ?std.time.Instant = null,
+
+/// The configuration for this IO that is derived from the main
+/// configuration. This must be exported so that we don't need to
+/// pass around Config pointers which makes memory management a pain.
+pub const DerivedConfig = struct {
+    arena: ArenaAllocator,
+
+    palette: terminal.color.Palette,
+    image_storage_limit: usize,
+    cursor_style: terminal.CursorStyle,
+    cursor_blink: ?bool,
+    cursor_color: ?configpkg.Config.Color,
+    foreground: configpkg.Config.Color,
+    background: configpkg.Config.Color,
+    osc_color_report_format: configpkg.Config.OSCColorReportFormat,
+    abnormal_runtime_threshold_ms: u32,
+    wait_after_command: bool,
+    enquiry_response: []const u8,
+
+    pub fn init(
+        alloc_gpa: Allocator,
+        config: *const configpkg.Config,
+    ) !DerivedConfig {
+        var arena = ArenaAllocator.init(alloc_gpa);
+        errdefer arena.deinit();
+        const alloc = arena.allocator();
+
+        return .{
+            .palette = config.palette.value,
+            .image_storage_limit = config.@"image-storage-limit",
+            .cursor_style = config.@"cursor-style",
+            .cursor_blink = config.@"cursor-style-blink",
+            .cursor_color = config.@"cursor-color",
+            .foreground = config.foreground,
+            .background = config.background,
+            .osc_color_report_format = config.@"osc-color-report-format",
+            .abnormal_runtime_threshold_ms = config.@"abnormal-command-exit-runtime",
+            .wait_after_command = config.@"wait-after-command",
+            .enquiry_response = try alloc.dupe(u8, config.@"enquiry-response"),
+
+            // This has to be last so that we copy AFTER the arena allocations
+            // above happen (Zig assigns in order).
+            .arena = arena,
+        };
+    }
+
+    pub fn deinit(self: *DerivedConfig) void {
+        self.arena.deinit();
+    }
+};
+
+/// Initialize the termio state.
+///
+/// This will also start the child process if the termio is configured
+/// to run a child process.
+pub fn init(self: *Termio, alloc: Allocator, opts: termio.Options) !void {
+    // Create our terminal
+    var term = try terminal.Terminal.init(alloc, .{
+        .cols = opts.grid_size.columns,
+        .rows = opts.grid_size.rows,
+        .max_scrollback = opts.full_config.@"scrollback-limit",
+    });
+    errdefer term.deinit(alloc);
+    term.default_palette = opts.config.palette;
+    term.color_palette.colors = opts.config.palette;
+
+    // Setup our initial grapheme cluster support if enabled. We use a
+    // switch to ensure we get a compiler error if more cases are added.
+    switch (opts.full_config.@"grapheme-width-method") {
+        .unicode => term.modes.set(.grapheme_cluster, true),
+        .legacy => {},
+    }
+
+    // Set the image size limits
+    try term.screen.kitty_images.setLimit(
+        alloc,
+        &term.screen,
+        opts.config.image_storage_limit,
+    );
+    try term.secondary_screen.kitty_images.setLimit(
+        alloc,
+        &term.secondary_screen,
+        opts.config.image_storage_limit,
+    );
+
+    // Set default cursor blink settings
+    term.modes.set(
+        .cursor_blinking,
+        opts.config.cursor_blink orelse true,
+    );
+
+    // Set our default cursor style
+    term.screen.cursor.cursor_style = opts.config.cursor_style;
+
+    // Setup our backend.
+    var backend = opts.backend;
+    backend.initTerminal(&term);
+
+    // Setup our terminal size in pixels for certain requests.
+    const screen_size = opts.screen_size.subPadding(opts.padding);
+    term.width_px = screen_size.width;
+    term.height_px = screen_size.height;
+
+    // Create our stream handler. This points to memory in self so it
+    // isn't safe to use until self.* is set.
+    const handler: StreamHandler = handler: {
+        const default_cursor_color = if (opts.config.cursor_color) |col|
+            col.toTerminalRGB()
+        else
+            null;
+
+        break :handler .{
+            .alloc = alloc,
+            .termio_mailbox = &self.mailbox,
+            .surface_mailbox = opts.surface_mailbox,
+            .renderer_state = opts.renderer_state,
+            .renderer_wakeup = opts.renderer_wakeup,
+            .renderer_mailbox = opts.renderer_mailbox,
+            .grid_size = &self.grid_size,
+            .terminal = &self.terminal,
+            .osc_color_report_format = opts.config.osc_color_report_format,
+            .enquiry_response = opts.config.enquiry_response,
+            .default_foreground_color = opts.config.foreground.toTerminalRGB(),
+            .default_background_color = opts.config.background.toTerminalRGB(),
+            .default_cursor_style = opts.config.cursor_style,
+            .default_cursor_blink = opts.config.cursor_blink,
+            .default_cursor_color = default_cursor_color,
+            .cursor_color = default_cursor_color,
+            .foreground_color = opts.config.foreground.toTerminalRGB(),
+            .background_color = opts.config.background.toTerminalRGB(),
+        };
+    };
+
+    self.* = .{
+        .alloc = alloc,
+        .terminal = term,
+        .config = opts.config,
+        .renderer_state = opts.renderer_state,
+        .renderer_wakeup = opts.renderer_wakeup,
+        .renderer_mailbox = opts.renderer_mailbox,
+        .surface_mailbox = opts.surface_mailbox,
+        .grid_size = opts.grid_size,
+        .backend = opts.backend,
+        .mailbox = opts.mailbox,
+        .terminal_stream = .{
+            .handler = handler,
+            .parser = .{
+                .osc_parser = .{
+                    // Populate the OSC parser allocator (optional) because
+                    // we want to support large OSC payloads such as OSC 52.
+                    .alloc = alloc,
+                },
+            },
+        },
+    };
+}
+
+pub fn deinit(self: *Termio) void {
+    self.backend.deinit();
+    self.terminal.deinit(self.alloc);
+    self.config.deinit();
+    self.mailbox.deinit(self.alloc);
+
+    // Clear any StreamHandler state
+    self.terminal_stream.handler.deinit();
+    self.terminal_stream.deinit();
+}
+
+pub fn threadEnter(self: *Termio, thread: *termio.Thread, data: *ThreadData) !void {
+    data.* = .{
+        .alloc = self.alloc,
+        .loop = &thread.loop,
+        .renderer_state = self.renderer_state,
+        .surface_mailbox = self.surface_mailbox,
+        .mailbox = &self.mailbox,
+        .backend = undefined, // Backend must replace this on threadEnter
+    };
+
+    // Setup our backend
+    try self.backend.threadEnter(self.alloc, self, data);
+}
+
+pub fn threadExit(self: *Termio, data: *ThreadData) void {
+    self.backend.threadExit(data);
+}
+
+/// Send a message to the the mailbox. Depending on the mailbox type in
+/// use this may process now or it may just enqueue and process later.
+///
+/// This will also notify the mailbox thread to process the message. If
+/// you're sending a lot of messages, it may be more efficient to use
+/// the mailbox directly and then call notify separately.
+pub fn queueMessage(
+    self: *Termio,
+    msg: termio.Message,
+    mutex: enum { locked, unlocked },
+) void {
+    self.mailbox.send(msg, switch (mutex) {
+        .locked => self.renderer_state.mutex,
+        .unlocked => null,
+    });
+    self.mailbox.notify();
+}
+
+/// Queue a write directly to the pty.
+///
+/// If you're using termio.Thread, this must ONLY be called from the
+/// mailbox thread. If you're not on the thread, use queueMessage with
+/// mailbox messages instead.
+///
+/// If you're not using termio.Thread, this is not threadsafe.
+pub inline fn queueWrite(
+    self: *Termio,
+    td: *ThreadData,
+    data: []const u8,
+    linefeed: bool,
+) !void {
+    try self.backend.queueWrite(self.alloc, td, data, linefeed);
+}
+
+/// Update the configuration.
+pub fn changeConfig(self: *Termio, td: *ThreadData, config: *DerivedConfig) !void {
+    // The remainder of this function is modifying terminal state or
+    // the read thread data, all of which requires holding the renderer
+    // state lock.
+    self.renderer_state.mutex.lock();
+    defer self.renderer_state.mutex.unlock();
+
+    // Deinit our old config. We do this in the lock because the
+    // stream handler may be referencing the old config (i.e. enquiry resp)
+    self.config.deinit();
+    self.config = config.*;
+
+    // Update our stream handler. The stream handler uses the same
+    // renderer mutex so this is safe to do despite being executed
+    // from another thread.
+    self.terminal_stream.handler.changeConfig(&self.config);
+    td.backend.changeConfig(&self.config);
+
+    // Update the configuration that we know about.
+    //
+    // Specific things we don't update:
+    //   - command, working-directory: we never restart the underlying
+    //   process so we don't care or need to know about these.
+
+    // Update the default palette. Note this will only apply to new colors drawn
+    // since we decode all palette colors to RGB on usage.
+    self.terminal.default_palette = config.palette;
+
+    // Update the active palette, except for any colors that were modified with
+    // OSC 4
+    for (0..config.palette.len) |i| {
+        if (!self.terminal.color_palette.mask.isSet(i)) {
+            self.terminal.color_palette.colors[i] = config.palette[i];
+            self.terminal.flags.dirty.palette = true;
+        }
+    }
+
+    // Set the image size limits
+    try self.terminal.screen.kitty_images.setLimit(
+        self.alloc,
+        &self.terminal.screen,
+        config.image_storage_limit,
+    );
+    try self.terminal.secondary_screen.kitty_images.setLimit(
+        self.alloc,
+        &self.terminal.secondary_screen,
+        config.image_storage_limit,
+    );
+}
+
+/// Resize the terminal.
+pub fn resize(
+    self: *Termio,
+    grid_size: renderer.GridSize,
+    screen_size: renderer.ScreenSize,
+    padding: renderer.Padding,
+) !void {
+    // Update the size of our pty.
+    const padded_size = screen_size.subPadding(padding);
+    try self.backend.resize(grid_size, padded_size);
+
+    // Update our cached grid size
+    self.grid_size = grid_size;
+
+    // Enter the critical area that we want to keep small
+    {
+        self.renderer_state.mutex.lock();
+        defer self.renderer_state.mutex.unlock();
+
+        // Update the size of our terminal state
+        try self.terminal.resize(
+            self.alloc,
+            grid_size.columns,
+            grid_size.rows,
+        );
+
+        // Update our pixel sizes
+        self.terminal.width_px = padded_size.width;
+        self.terminal.height_px = padded_size.height;
+
+        // Disable synchronized output mode so that we show changes
+        // immediately for a resize. This is allowed by the spec.
+        self.terminal.modes.set(.synchronized_output, false);
+
+        // Wake up our renderer so any changes will be shown asap
+        self.renderer_wakeup.notify() catch {};
+    }
+}
+
+/// Reset the synchronized output mode. This is usually called by timer
+/// expiration from the termio thread.
+pub fn resetSynchronizedOutput(self: *Termio) void {
+    self.renderer_state.mutex.lock();
+    defer self.renderer_state.mutex.unlock();
+    self.terminal.modes.set(.synchronized_output, false);
+    self.renderer_wakeup.notify() catch {};
+}
+
+/// Clear the screen.
+pub fn clearScreen(self: *Termio, td: *ThreadData, history: bool) !void {
+    {
+        self.renderer_state.mutex.lock();
+        defer self.renderer_state.mutex.unlock();
+
+        // If we're on the alternate screen, we do not clear. Since this is an
+        // emulator-level screen clear, this messes up the running programs
+        // knowledge of where the cursor is and causes rendering issues. So,
+        // for alt screen, we do nothing.
+        if (self.terminal.active_screen == .alternate) return;
+
+        // Clear our scrollback
+        if (history) self.terminal.eraseDisplay(.scrollback, false);
+
+        // If we're not at a prompt, we just delete above the cursor.
+        if (!self.terminal.cursorIsAtPrompt()) {
+            if (self.terminal.screen.cursor.y > 0) {
+                self.terminal.screen.eraseRows(
+                    .{ .active = .{ .y = 0 } },
+                    .{ .active = .{ .y = self.terminal.screen.cursor.y - 1 } },
+                );
+            }
+
+            return;
+        }
+
+        // At a prompt, we want to first fully clear the screen, and then after
+        // send a FF (0x0C) to the shell so that it can repaint the screen.
+        // Mark the current row as a not a prompt so we can properly
+        // clear the full screen in the next eraseDisplay call.
+        self.terminal.markSemanticPrompt(.command);
+        assert(!self.terminal.cursorIsAtPrompt());
+        self.terminal.eraseDisplay(.complete, false);
+    }
+
+    // If we reached here it means we're at a prompt, so we send a form-feed.
+    try self.queueWrite(td, &[_]u8{0x0C}, false);
+}
+
+/// Scroll the viewport
+pub fn scrollViewport(self: *Termio, scroll: terminal.Terminal.ScrollViewport) !void {
+    self.renderer_state.mutex.lock();
+    defer self.renderer_state.mutex.unlock();
+    try self.terminal.scrollViewport(scroll);
+}
+
+/// Jump the viewport to the prompt.
+pub fn jumpToPrompt(self: *Termio, delta: isize) !void {
+    {
+        self.renderer_state.mutex.lock();
+        defer self.renderer_state.mutex.unlock();
+        self.terminal.screen.scroll(.{ .delta_prompt = delta });
+    }
+
+    try self.renderer_wakeup.notify();
+}
+
+/// Called when the child process exited abnormally but before
+/// the surface is notified.
+pub fn childExitedAbnormally(self: *Termio, exit_code: u32, runtime_ms: u64) !void {
+    self.renderer_state.mutex.lock();
+    defer self.renderer_state.mutex.unlock();
+    const t = self.renderer_state.terminal;
+    try self.backend.childExitedAbnormally(self.alloc, t, exit_code, runtime_ms);
+}
+
+/// Process output from the pty. This is the manual API that users can
+/// call with pty data but it is also called by the read thread when using
+/// an exec subprocess.
+pub fn processOutput(self: *Termio, buf: []const u8) void {
+    // We are modifying terminal state from here on out and we need
+    // the lock to grab our read data.
+    self.renderer_state.mutex.lock();
+    defer self.renderer_state.mutex.unlock();
+    self.processOutputLocked(buf);
+}
+
+/// Process output from readdata but the lock is already held.
+fn processOutputLocked(self: *Termio, buf: []const u8) void {
+    // Schedule a render. We can call this first because we have the lock.
+    self.terminal_stream.handler.queueRender() catch unreachable;
+
+    // Whenever a character is typed, we ensure the cursor is in the
+    // non-blink state so it is rendered if visible. If we're under
+    // HEAVY read load, we don't want to send a ton of these so we
+    // use a timer under the covers
+    if (std.time.Instant.now()) |now| cursor_reset: {
+        if (self.last_cursor_reset) |last| {
+            if (now.since(last) <= (500 * std.time.ns_per_ms)) {
+                break :cursor_reset;
+            }
+        }
+
+        self.last_cursor_reset = now;
+        _ = self.renderer_mailbox.push(.{
+            .reset_cursor_blink = {},
+        }, .{ .instant = {} });
+    } else |err| {
+        log.warn("failed to get current time err={}", .{err});
+    }
+
+    // If we have an inspector, we enter SLOW MODE because we need to
+    // process a byte at a time alternating between the inspector handler
+    // and the termio handler. This is very slow compared to our optimizations
+    // below but at least users only pay for it if they're using the inspector.
+    if (self.renderer_state.inspector) |insp| {
+        for (buf, 0..) |byte, i| {
+            insp.recordPtyRead(buf[i .. i + 1]) catch |err| {
+                log.err("error recording pty read in inspector err={}", .{err});
+            };
+
+            self.terminal_stream.next(byte) catch |err|
+                log.err("error processing terminal data: {}", .{err});
+        }
+    } else {
+        self.terminal_stream.nextSlice(buf) catch |err|
+            log.err("error processing terminal data: {}", .{err});
+    }
+
+    // If our stream handling caused messages to be sent to the mailbox
+    // thread, then we need to wake it up so that it processes them.
+    if (self.terminal_stream.handler.termio_messaged) {
+        self.terminal_stream.handler.termio_messaged = false;
+        self.mailbox.notify();
+    }
+}
+
+/// ThreadData is the data created and stored in the termio thread
+/// when the thread is started and destroyed when the thread is
+/// stopped.
+///
+/// All of the fields in this struct should only be read/written by
+/// the termio thread. As such, a lock is not necessary.
+pub const ThreadData = struct {
+    /// Allocator used for the event data
+    alloc: Allocator,
+
+    /// The event loop associated with this thread. This is owned by
+    /// the Thread but we have a pointer so we can queue new work to it.
+    loop: *xev.Loop,
+
+    /// The shared render state
+    renderer_state: *renderer.State,
+
+    /// Mailboxes for different threads
+    surface_mailbox: apprt.surface.Mailbox,
+
+    /// Data associated with the backend implementation (i.e. pty/exec state)
+    backend: termio.backend.ThreadData,
+    mailbox: *termio.Mailbox,
+
+    pub fn deinit(self: *ThreadData) void {
+        self.backend.deinit(self.alloc);
+        self.* = undefined;
+    }
+};

src/termio/Thread.zig

@@ -1,5 +1,14 @@
-//! Represents the IO thread logic. The IO thread is responsible for
+//! Represents the "writer" thread for terminal IO. The reader side is
-//! the child process and pty management.
+//! handled by the Termio struct itself and dependent on the underlying
+//! implementation (i.e. if its a pty, manual, etc.).
+//!
+//! The writer thread does handle writing bytes to the pty but also handles
+//! different events such as starting synchronized output, changing some
+//! modes (like linefeed), etc. The goal is to offload as much from the
+//! reader thread as possible since it is the hot path in parsing VT
+//! sequences and updating terminal state.
+//!
+//! This thread state can only be used by one thread at a time.
 pub const Thread = @This();
 
 const std = @import("std");
@@ -12,11 +21,6 @@ const BlockingQueue = @import("../blocking_queue.zig").BlockingQueue;
 const Allocator = std.mem.Allocator;
 const log = std.log.scoped(.io_thread);
 
-/// The type used for sending messages to the IO thread. For now this is
-/// hardcoded with a capacity. We can make this a comptime parameter in
-/// the future if we want it configurable.
-pub const Mailbox = BlockingQueue(termio.Message, 64);
-
 /// This stores the information that is coalesced.
 const Coalesce = struct {
     /// The number of milliseconds to coalesce certain messages like resize for.
@@ -38,8 +42,8 @@ alloc: std.mem.Allocator,
 /// so that users of the loop always have an allocator.
 loop: xev.Loop,
 
-/// This can be used to wake up the thread.
+/// The completion to use for the wakeup async handle that is present
-wakeup: xev.Async,
+/// on the termio.Writer.
 wakeup_c: xev.Completion = .{},
 
 /// This can be used to stop the thread on the next loop iteration.
@@ -58,13 +62,6 @@ sync_reset: xev.Timer,
 sync_reset_c: xev.Completion = .{},
 sync_reset_cancel_c: xev.Completion = .{},
 
-/// The underlying IO implementation.
-impl: *termio.Impl,
-
-/// The mailbox that can be used to send this thread messages. Note
-/// this is a blocking queue so if it is full you will get errors (or block).
-mailbox: *Mailbox,
-
 flags: packed struct {
     /// This is set to true only when an abnormal exit is detected. It
     /// tells our mailbox system to drain and ignore all messages.
@@ -83,16 +80,11 @@ flags: packed struct {
 /// is up to the caller to start the thread with the threadMain entrypoint.
 pub fn init(
     alloc: Allocator,
-    impl: *termio.Impl,
 ) !Thread {
     // Create our event loop.
     var loop = try xev.Loop.init(.{});
     errdefer loop.deinit();
 
-    // This async handle is used to "wake up" the renderer and force a render.
-    var wakeup_h = try xev.Async.init();
-    errdefer wakeup_h.deinit();
-
     // This async handle is used to stop the loop and force the thread to end.
     var stop_h = try xev.Async.init();
     errdefer stop_h.deinit();
@@ -105,19 +97,12 @@ pub fn init(
     var sync_reset_h = try xev.Timer.init();
     errdefer sync_reset_h.deinit();
 
-    // The mailbox for messaging this thread
-    var mailbox = try Mailbox.create(alloc);
-    errdefer mailbox.destroy(alloc);
-
     return Thread{
         .alloc = alloc,
         .loop = loop,
-        .wakeup = wakeup_h,
         .stop = stop_h,
         .coalesce = coalesce_h,
         .sync_reset = sync_reset_h,
-        .impl = impl,
-        .mailbox = mailbox,
     };
 }
 
@@ -127,17 +112,13 @@ pub fn deinit(self: *Thread) void {
     self.coalesce.deinit();
     self.sync_reset.deinit();
     self.stop.deinit();
-    self.wakeup.deinit();
     self.loop.deinit();
-
-    // Nothing can possibly access the mailbox anymore, destroy it.
-    self.mailbox.destroy(self.alloc);
 }
 
 /// The main entrypoint for the thread.
-pub fn threadMain(self: *Thread) void {
+pub fn threadMain(self: *Thread, io: *termio.Termio) void {
     // Call child function so we can use errors...
-    self.threadMain_() catch |err| {
+    self.threadMain_(io) catch |err| {
         log.warn("error in io thread err={}", .{err});
 
         // Use an arena to simplify memory management below
@@ -150,9 +131,9 @@ pub fn threadMain(self: *Thread) void {
         // the error to the surface thread and let the apprt deal with it
         // in some way but this works for now. Without this, the user would
         // just see a blank terminal window.
-        self.impl.renderer_state.mutex.lock();
+        io.renderer_state.mutex.lock();
-        defer self.impl.renderer_state.mutex.unlock();
+        defer io.renderer_state.mutex.unlock();
-        const t = self.impl.renderer_state.terminal;
+        const t = io.renderer_state.terminal;
 
         // Hide the cursor
         t.modes.set(.cursor_visible, false);
@@ -216,19 +197,30 @@ pub fn threadMain(self: *Thread) void {
     }
 }
 
-fn threadMain_(self: *Thread) !void {
+fn threadMain_(self: *Thread, io: *termio.Termio) !void {
     defer log.debug("IO thread exited", .{});
 
-    // Start the async handlers. We start these first so that they're
+    // Get the mailbox. This must be an SPSC mailbox for threading.
-    // registered even if anything below fails so we can drain the mailbox.
+    const mailbox = switch (io.mailbox) {
-    self.wakeup.wait(&self.loop, &self.wakeup_c, Thread, self, wakeupCallback);
+        .spsc => |*v| v,
-    self.stop.wait(&self.loop, &self.stop_c, Thread, self, stopCallback);
+        // else => return error.TermioUnsupportedMailbox,
+    };
+
+    // This is the data sent to xev callbacks. We want a pointer to both
+    // ourselves and the thread data so we can thread that through (pun intended).
+    var cb: CallbackData = .{ .self = self, .io = io };
 
     // Run our thread start/end callbacks. This allows the implementation
-    // to hook into the event loop as needed.
+    // to hook into the event loop as needed. The thread data is created
-    var data = try self.impl.threadEnter(self);
+    // on the stack here so that it has a stable pointer throughout the
-    defer data.deinit();
+    // lifetime of the thread.
-    defer self.impl.threadExit(data);
+    try io.threadEnter(self, &cb.data);
+    defer cb.data.deinit();
+    defer io.threadExit(&cb.data);
+
+    // Start the async handlers.
+    mailbox.wakeup.wait(&self.loop, &self.wakeup_c, CallbackData, &cb, wakeupCallback);
+    self.stop.wait(&self.loop, &self.stop_c, CallbackData, &cb, stopCallback);
 
     // Run
     log.debug("starting IO thread", .{});
@@ -236,11 +228,26 @@ fn threadMain_(self: *Thread) !void {
     try self.loop.run(.until_done);
 }
 
+/// This is the data passed to xev callbacks on the thread.
+const CallbackData = struct {
+    self: *Thread,
+    io: *termio.Termio,
+    data: termio.Termio.ThreadData = undefined,
+};
+
 /// Drain the mailbox, handling all the messages in our terminal implementation.
-fn drainMailbox(self: *Thread) !void {
+fn drainMailbox(
+    self: *Thread,
+    cb: *CallbackData,
+) !void {
+    // We assert when starting the thread that this is the state
+    const mailbox = cb.io.mailbox.spsc.queue;
+    const io = cb.io;
+    const data = &cb.data;
+
     // If we're draining, we just drain the mailbox and return.
     if (self.flags.drain) {
-        while (self.mailbox.pop()) |_| {}
+        while (mailbox.pop()) |_| {}
         return;
     }
 
@@ -248,7 +255,7 @@ fn drainMailbox(self: *Thread) !void {
     // expectation is that all our message handlers will be non-blocking
     // ENOUGH to not mess up throughput on producers.
     var redraw: bool = false;
-    while (self.mailbox.pop()) |message| {
+    while (mailbox.pop()) |message| {
         // If we have a message we always redraw
         redraw = true;
 
@@ -256,21 +263,33 @@ fn drainMailbox(self: *Thread) !void {
         switch (message) {
             .change_config => |config| {
                 defer config.alloc.destroy(config.ptr);
-                try self.impl.changeConfig(config.ptr);
+                try io.changeConfig(data, config.ptr);
             },
             .inspector => |v| self.flags.has_inspector = v,
-            .resize => |v| self.handleResize(v),
+            .resize => |v| self.handleResize(cb, v),
-            .clear_screen => |v| try self.impl.clearScreen(v.history),
+            .clear_screen => |v| try io.clearScreen(data, v.history),
-            .scroll_viewport => |v| try self.impl.scrollViewport(v),
+            .scroll_viewport => |v| try io.scrollViewport(v),
-            .jump_to_prompt => |v| try self.impl.jumpToPrompt(v),
+            .jump_to_prompt => |v| try io.jumpToPrompt(v),
-            .start_synchronized_output => self.startSynchronizedOutput(),
+            .start_synchronized_output => self.startSynchronizedOutput(cb),
             .linefeed_mode => |v| self.flags.linefeed_mode = v,
-            .child_exited_abnormally => |v| try self.impl.childExitedAbnormally(v.exit_code, v.runtime_ms),
+            .child_exited_abnormally => |v| try io.childExitedAbnormally(v.exit_code, v.runtime_ms),
-            .write_small => |v| try self.impl.queueWrite(v.data[0..v.len], self.flags.linefeed_mode),
+            .write_small => |v| try io.queueWrite(
-            .write_stable => |v| try self.impl.queueWrite(v, self.flags.linefeed_mode),
+                data,
+                v.data[0..v.len],
+                self.flags.linefeed_mode,
+            ),
+            .write_stable => |v| try io.queueWrite(
+                data,
+                v,
+                self.flags.linefeed_mode,
+            ),
             .write_alloc => |v| {
                 defer v.alloc.free(v.data);
-                try self.impl.queueWrite(v.data, self.flags.linefeed_mode);
+                try io.queueWrite(
+                    data,
+                    v.data,
+                    self.flags.linefeed_mode,
+                );
             },
         }
     }
@@ -278,23 +297,23 @@ fn drainMailbox(self: *Thread) !void {
     // Trigger a redraw after we've drained so we don't waste cyces
     // messaging a redraw.
     if (redraw) {
-        try self.impl.renderer_wakeup.notify();
+        try io.renderer_wakeup.notify();
     }
 }
 
-fn startSynchronizedOutput(self: *Thread) void {
+fn startSynchronizedOutput(self: *Thread, cb: *CallbackData) void {
     self.sync_reset.reset(
         &self.loop,
         &self.sync_reset_c,
         &self.sync_reset_cancel_c,
         sync_reset_ms,
-        Thread,
+        CallbackData,
-        self,
+        cb,
         syncResetCallback,
     );
 }
 
-fn handleResize(self: *Thread, resize: termio.Message.Resize) void {
+fn handleResize(self: *Thread, cb: *CallbackData, resize: termio.Message.Resize) void {
     self.coalesce_data.resize = resize;
 
     // If the timer is already active we just return. In the future we want
@@ -307,14 +326,14 @@ fn handleResize(self: *Thread, resize: termio.Message.Resize) void {
         &self.coalesce_c,
         &self.coalesce_cancel_c,
         Coalesce.min_ms,
-        Thread,
+        CallbackData,
-        self,
+        cb,
         coalesceCallback,
     );
 }
 
 fn syncResetCallback(
-    self_: ?*Thread,
+    cb_: ?*CallbackData,
     _: *xev.Loop,
     _: *xev.Completion,
     r: xev.Timer.RunError!void,
@@ -327,13 +346,13 @@ fn syncResetCallback(
         },
     };
 
-    const self = self_ orelse return .disarm;
+    const cb = cb_ orelse return .disarm;
-    self.impl.resetSynchronizedOutput();
+    cb.io.resetSynchronizedOutput();
     return .disarm;
 }
 
 fn coalesceCallback(
-    self_: ?*Thread,
+    cb_: ?*CallbackData,
     _: *xev.Loop,
     _: *xev.Completion,
     r: xev.Timer.RunError!void,
@@ -346,11 +365,11 @@ fn coalesceCallback(
         },
     };
 
-    const self = self_ orelse return .disarm;
+    const cb = cb_ orelse return .disarm;
 
-    if (self.coalesce_data.resize) |v| {
+    if (cb.self.coalesce_data.resize) |v| {
-        self.coalesce_data.resize = null;
+        cb.self.coalesce_data.resize = null;
-        self.impl.resize(v.grid_size, v.screen_size, v.padding) catch |err| {
+        cb.io.resize(v.grid_size, v.screen_size, v.padding) catch |err| {
             log.warn("error during resize err={}", .{err});
         };
     }
@@ -359,7 +378,7 @@ fn coalesceCallback(
 }
 
 fn wakeupCallback(
-    self_: ?*Thread,
+    cb_: ?*CallbackData,
     _: *xev.Loop,
     _: *xev.Completion,
     r: xev.Async.WaitError!void,
@@ -369,23 +388,22 @@ fn wakeupCallback(
         return .rearm;
     };
 
-    const t = self_.?;
-
     // When we wake up, we check the mailbox. Mailbox producers should
     // wake up our thread after publishing.
-    t.drainMailbox() catch |err|
+    const cb = cb_ orelse return .rearm;
+    cb.self.drainMailbox(cb) catch |err|
         log.err("error draining mailbox err={}", .{err});
 
     return .rearm;
 }
 
 fn stopCallback(
-    self_: ?*Thread,
+    cb_: ?*CallbackData,
     _: *xev.Loop,
     _: *xev.Completion,
     r: xev.Async.WaitError!void,
 ) xev.CallbackAction {
     _ = r catch unreachable;
-    self_.?.loop.stop();
+    cb_.?.self.loop.stop();
     return .disarm;
 }

src/termio/backend.zig

@@ -0,0 +1,123 @@
+const std = @import("std");
+const builtin = @import("builtin");
+const assert = std.debug.assert;
+const Allocator = std.mem.Allocator;
+const posix = std.posix;
+const xev = @import("xev");
+const build_config = @import("../build_config.zig");
+const configpkg = @import("../config.zig");
+const internal_os = @import("../os/main.zig");
+const renderer = @import("../renderer.zig");
+const shell_integration = @import("shell_integration.zig");
+const terminal = @import("../terminal/main.zig");
+const termio = @import("../termio.zig");
+const Command = @import("../Command.zig");
+const SegmentedPool = @import("../segmented_pool.zig").SegmentedPool;
+const Pty = @import("../pty.zig").Pty;
+
+// The preallocation size for the write request pool. This should be big
+// enough to satisfy most write requests. It must be a power of 2.
+const WRITE_REQ_PREALLOC = std.math.pow(usize, 2, 5);
+
+/// The kinds of backends.
+pub const Kind = enum { exec };
+
+/// Configuration for the various backend types.
+pub const Config = union(Kind) {
+    /// Exec uses posix exec to run a command with a pty.
+    exec: termio.Exec.Config,
+};
+
+/// Backend implementations. A backend is responsible for owning the pty
+/// behavior and providing read/write capabilities.
+pub const Backend = union(Kind) {
+    exec: termio.Exec,
+
+    pub fn deinit(self: *Backend) void {
+        switch (self.*) {
+            .exec => |*exec| exec.deinit(),
+        }
+    }
+
+    pub fn initTerminal(self: *Backend, t: *terminal.Terminal) void {
+        switch (self.*) {
+            .exec => |*exec| exec.initTerminal(t),
+        }
+    }
+
+    pub fn threadEnter(
+        self: *Backend,
+        alloc: Allocator,
+        io: *termio.Termio,
+        td: *termio.Termio.ThreadData,
+    ) !void {
+        switch (self.*) {
+            .exec => |*exec| try exec.threadEnter(alloc, io, td),
+        }
+    }
+
+    pub fn threadExit(self: *Backend, td: *termio.Termio.ThreadData) void {
+        switch (self.*) {
+            .exec => |*exec| exec.threadExit(td),
+        }
+    }
+
+    pub fn resize(
+        self: *Backend,
+        grid_size: renderer.GridSize,
+        screen_size: renderer.ScreenSize,
+    ) !void {
+        switch (self.*) {
+            .exec => |*exec| try exec.resize(grid_size, screen_size),
+        }
+    }
+
+    pub fn queueWrite(
+        self: *Backend,
+        alloc: Allocator,
+        td: *termio.Termio.ThreadData,
+        data: []const u8,
+        linefeed: bool,
+    ) !void {
+        switch (self.*) {
+            .exec => |*exec| try exec.queueWrite(alloc, td, data, linefeed),
+        }
+    }
+
+    pub fn childExitedAbnormally(
+        self: *Backend,
+        gpa: Allocator,
+        t: *terminal.Terminal,
+        exit_code: u32,
+        runtime_ms: u64,
+    ) !void {
+        switch (self.*) {
+            .exec => |*exec| try exec.childExitedAbnormally(
+                gpa,
+                t,
+                exit_code,
+                runtime_ms,
+            ),
+        }
+    }
+};
+
+/// Termio thread data. See termio.ThreadData for docs.
+pub const ThreadData = union(Kind) {
+    exec: termio.Exec.ThreadData,
+
+    pub fn deinit(self: *ThreadData, alloc: Allocator) void {
+        switch (self.*) {
+            .exec => |*exec| exec.deinit(alloc),
+        }
+    }
+
+    pub fn changeConfig(self: *ThreadData, config: *termio.DerivedConfig) void {
+        switch (self.*) {
+            .exec => |*exec| {
+                exec.abnormal_runtime_threshold_ms = config.abnormal_runtime_threshold_ms;
+                exec.wait_after_command = config.wait_after_command;
+            },
+        }
+    }
+};

src/termio/mailbox.zig

@@ -0,0 +1,108 @@
+const std = @import("std");
+const builtin = @import("builtin");
+const assert = std.debug.assert;
+const Allocator = std.mem.Allocator;
+const xev = @import("xev");
+const renderer = @import("../renderer.zig");
+const termio = @import("../termio.zig");
+const BlockingQueue = @import("../blocking_queue.zig").BlockingQueue;
+
+const log = std.log.scoped(.io_writer);
+
+/// A queue used for storing messages that is periodically drained.
+/// Typically used by a multi-threaded application. The capacity is
+/// hardcoded to a value that empirically has made sense for Ghostty usage
+/// but I'm open to changing it with good arguments.
+const Queue = BlockingQueue(termio.Message, 64);
+
+/// The location to where write-related messages are sent.
+pub const Mailbox = union(enum) {
+    // /// Write messages to an unbounded list backed by an allocator.
+    // /// This is useful for single-threaded applications where you're not
+    // /// afraid of running out of memory. You should be careful that you're
+    // /// processing this in a timely manner though since some heavy workloads
+    // /// will produce a LOT of messages.
+    // ///
+    // /// At the time of authoring this, the primary use case for this is
+    // /// testing more than anything, but it probably will have a use case
+    // /// in libghostty eventually.
+    // unbounded: std.ArrayList(termio.Message),
+
+    /// Write messages to a SPSC queue for multi-threaded applications.
+    spsc: struct {
+        queue: *Queue,
+        wakeup: xev.Async,
+    },
+
+    /// Init the SPSC writer.
+    pub fn initSPSC(alloc: Allocator) !Mailbox {
+        var queue = try Queue.create(alloc);
+        errdefer queue.destroy(alloc);
+
+        var wakeup = try xev.Async.init();
+        errdefer wakeup.deinit();
+
+        return .{ .spsc = .{ .queue = queue, .wakeup = wakeup } };
+    }
+
+    pub fn deinit(self: *Mailbox, alloc: Allocator) void {
+        switch (self.*) {
+            .spsc => |*v| {
+                v.queue.destroy(alloc);
+                v.wakeup.deinit();
+            },
+        }
+    }
+
+    /// Sends the given message without notifying there are messages.
+    ///
+    /// If the optional mutex is given, it must already be LOCKED. If the
+    /// send would block, we'll unlock this mutex, resend the message, and
+    /// lock it again. This handles an edge case where queues are full.
+    /// This may not apply to all writer types.
+    pub fn send(
+        self: *Mailbox,
+        msg: termio.Message,
+        mutex: ?*std.Thread.Mutex,
+    ) void {
+        switch (self.*) {
+            .spsc => |*mb| send: {
+                // Try to write to the queue with an instant timeout. This is the
+                // fast path because we can queue without a lock.
+                if (mb.queue.push(msg, .{ .instant = {} }) > 0) break :send;
+
+                // If we enter this conditional, the queue is full. We wake up
+                // the writer thread so that it can process messages to clear up
+                // space. However, the writer thread may require the renderer
+                // lock so we need to unlock.
+                mb.wakeup.notify() catch |err| {
+                    log.warn("failed to wake up writer, data will be dropped err={}", .{err});
+                    return;
+                };
+
+                // Unlock the renderer state so the writer thread can acquire it.
+                // Then try to queue our message before continuing. This is a very
+                // slow path because we are having a lot of contention for data.
+                // But this only gets triggered in certain pathological cases.
+                //
+                // Note that writes themselves don't require a lock, but there
+                // are other messages in the writer queue (resize, focus) that
+                // could acquire the lock. This is why we have to release our lock
+                // here.
+                if (mutex) |m| m.unlock();
+                defer if (mutex) |m| m.lock();
+                _ = mb.queue.push(msg, .{ .forever = {} });
+            },
+        }
+    }
+
+    /// Notify that there are new messages. This may be a noop depending
+    /// on the writer type.
+    pub fn notify(self: *Mailbox) void {
+        switch (self.*) {
+            .spsc => |*v| v.wakeup.notify() catch |err| {
+                log.warn("failed to notify writer, data will be dropped err={}", .{err});
+            },
+        }
+    }
+};

src/termio/message.zig

@@ -33,7 +33,7 @@ pub const Message = union(enum) {
     /// is allocated via the allocator and is expected to be freed when done.
     change_config: struct {
         alloc: Allocator,
-        ptr: *termio.Impl.DerivedConfig,
+        ptr: *termio.Termio.DerivedConfig,
     },
 
     /// Activate or deactivate the inspector.