renderer: hand off state mutex to avoid starving frames (#13265)

The mutex is unfair. From my understanding (after a brief convo with
@rockorager), it's a race between the parse thread and the renderer
thread.

The parse thread is unlocking it then locking it faster than the
renderer thread can get a lock/unlock. Under sustained pty output the
parse thread never sleeps between batches, so the renderer's frame
snapshot can starve for as long as the output lasts.

The fix here makes the parse thread voluntarily stay off the lock until
the renderer has had one turn by introducing two atomics:

- `demand`: a waiter count. The renderer increments it before locking
and decrements it after acquiring, so demand > 0 means "the renderer is
queued on the lock or about to be."
- `handoff_gen`: a generation counter. The renderer bumps it (with a
futex wake) after unlocking, meaning "a waiter completed one full turn."

At each batch boundary the parse thread checks demand (a single relaxed
load in the common case, so this **should** cost nothing when the
renderer isn't waiting). If a waiter exists, it futex-waits on
handoff_gen until the renderer has taken and released the lock, bounded
by a 1ms timeout, trying to ensure a lost wake can't stall IO.

This is inspired from `parking_lot` form rust land, which has eventual
fairness, but applied only at the one site that misbehaves.

## LLM Note

I used Fable 5 in Amp to write the code, but only after I identified the
problem myself from previous experience with mutex fairness (and the
convo above with rockorager).

The diagnosis — the parse thread barging the unfair mutex and starving
the renderer — came first.Fable then traced the exact loop in
`Exec.zig/generic.zig` and implemented the handoff protocol.

I've reviewed the changes.

src/renderer/State.zig

@@ -1,6 +1,9 @@
 //! This is the render state that is given to a renderer.
 
+const State = @This();
+
 const std = @import("std");
+const assert = std.debug.assert;
 const Allocator = std.mem.Allocator;
 const Inspector = @import("../inspector/main.zig").Inspector;
 const terminalpkg = @import("../terminal/main.zig");
@@ -30,6 +33,76 @@ preedit: ?Preedit = null,
 /// need about the mouse.
 mouse: Mouse = .{},
 
+/// The number of threads currently waiting to acquire `mutex` via
+/// `lockDemand`. This is not protected by the mutex; it is read by
+/// hot lock/unlock loops (the IO parse thread) in `yieldToDemand` to
+/// decide whether to hand the mutex off before relocking it.
+demand: std.atomic.Value(u32) = .init(0),
+
+/// Handoff generation counter. Incremented (with a futex wake) by
+/// `unlockDemand` after a demanding waiter releases the mutex, so that
+/// `yieldToDemand` knows the waiter had its turn.
+handoff_gen: std.atomic.Value(u32) = .init(0),
+
+/// How long `yieldToDemand` sleeps waiting for a demanding waiter to
+/// take its turn before giving up. This bounds how long the IO parse
+/// thread can stall if a wake is lost or the waiter is descheduled; a
+/// demanding critical section (the renderer's frame snapshot) is
+/// microseconds, so one millisecond is generous.
+const handoff_timeout_ns = 1 * std.time.ns_per_ms;
+
+/// Acquire `mutex` while signaling demand for it. Use this instead of
+/// locking the mutex directly on threads that must not be starved by
+/// a hot lock/unlock loop (the renderer's frame snapshot). Must be
+/// released with `unlockDemand`; releasing with `mutex.unlock` keeps
+/// the data safe but makes parked `yieldToDemand` callers wait out
+/// their full timeout.
+///
+/// Both `std.Thread.Mutex` and os_unfair_lock are unfair: a running
+/// thread that unlocks and immediately relocks beats a sleeping
+/// waiter every time, because the waiter must first be woken and
+/// scheduled. Under sustained pty output the IO parse thread is
+/// exactly such a loop, so without this signal the renderer can
+/// starve for as long as the output lasts.
+pub fn lockDemand(self: *State) void {
+    _ = self.demand.fetchAdd(1, .monotonic);
+    self.mutex.lock();
+    const prev = self.demand.fetchSub(1, .monotonic);
+    assert(prev > 0);
+}
+
+/// Release `mutex` acquired via `lockDemand` and notify hot loops
+/// parked in `yieldToDemand` that the demanding waiter had its turn.
+pub fn unlockDemand(self: *State) void {
+    self.mutex.unlock();
+    _ = self.handoff_gen.fetchAdd(1, .monotonic);
+    std.Thread.Futex.wake(&self.handoff_gen, 1);
+}
+
+/// Called by hot lock/unlock loops between critical sections, with
+/// `mutex` NOT held: if a `lockDemand` waiter exists, sleep until it
+/// has acquired and released the mutex (or the timeout passes). This
+/// is the handoff that unfair mutexes never do on their own.
+///
+/// The orderings here are all monotonic because these atomics are a
+/// scheduling heuristic, not a synchronization boundary: the mutex
+/// itself orders the protected data, and the timeout bounds any
+/// staleness.
+pub fn yieldToDemand(self: *State) void {
+    if (self.demand.load(.monotonic) == 0) return;
+
+    // Snapshot the generation before rechecking demand: if the waiter
+    // acquires and releases between our check and the wait below, the
+    // generation no longer matches and timedWait returns immediately.
+    const gen = self.handoff_gen.load(.monotonic);
+    if (self.demand.load(.monotonic) == 0) return;
+    std.Thread.Futex.timedWait(
+        &self.handoff_gen,
+        gen,
+        handoff_timeout_ns,
+    ) catch {};
+}
+
 pub const Mouse = struct {
     /// The point on the viewport where the mouse currently is. We use
     /// viewport points to avoid the complexity of mapping the mouse to

src/renderer/generic.zig

@@ -1170,8 +1170,10 @@ pub fn Renderer(comptime GraphicsAPI: type) type {
                 //     std.log.err("[updateFrame critical time] start={}\tduration={} us", .{ start_micro, end.since(start) / std.time.ns_per_us });
                 // }
 
-                state.mutex.lock();
-                defer state.mutex.unlock();
+                // Lock while signaling demand so the IO parse thread
+                // can't starve us. See renderer.State.lockDemand.
+                state.lockDemand();
+                defer state.unlockDemand();
 
                 // If we're in a synchronized output state, we pause all rendering.
                 if (state.terminal.modes.get(.synchronized_output)) {

src/termio/Exec.zig

@@ -1486,6 +1486,10 @@ pub const ReadThread = struct {
                     _ = posix.write(pipeline.idle_write_fd, "i") catch {};
                 }
             }
+
+            // Batch boundary: hand the renderer state mutex off if
+            // the renderer is waiting. See renderer.State.lockDemand.
+            io.renderer_state.yieldToDemand();
         }
     }
 
@@ -1776,6 +1780,11 @@ pub const ReadThread = struct {
                 }
 
                 @call(.always_inline, termio.Termio.processOutput, .{ io, buf[0..n] });
+
+                // See threadMainPosix: hand the renderer state mutex
+                // off if the renderer is waiting, since this loop
+                // would otherwise starve it under heavy output.
+                io.renderer_state.yieldToDemand();
             }
 
             var quit_bytes: windows.DWORD = 0;