cmux-tui: a daemon start adopts its terminal hosts in parallel (perf100, cx-p00l3.1)

cmux-tui/crates/cmux-tui-core/src/mux/kitty_budget.rs

@@ -216,7 +216,7 @@ impl Mux {
             }
         }
         budget.capacity = kitty_image_budget_capacity(
-            Self::kitty_image_budget_owner_count(budget),
+            Self::kitty_image_budget_owner_count(budget).max(budget.owner_floor),
             budget.capacity,
         );
     }

cmux-tui/crates/cmux-tui-core/src/mux/kitty_budget_state.rs

@@ -101,6 +101,12 @@ pub(super) struct KittyImageBudgetState {
     pub(super) capacity: usize,
     pub(super) worker_running: bool,
     pub(super) expansion_in_flight: bool,
+    /// While a daemon start adopts many hosts at once: the owner count the
+    /// capacity is sized for at least, so reservations made side by side
+    /// get their final share at once. A reservation not yet committed
+    /// cannot be shrunk, and each one that changed the bucket used to wait
+    /// out KITTY_RESERVATION_WAIT and start with graphics disabled.
+    pub(super) owner_floor: usize,
 }
 
 pub(super) struct PendingKittyImageBudgetOperation {

cmux-tui/crates/cmux-tui-core/src/mux/kitty_reservation.rs

@@ -12,6 +12,23 @@ use super::*;
 const KITTY_RESERVATION_WAIT: Duration = Duration::from_millis(500);
 
 impl Mux {
+    /// Size the Kitty budget for `extra` more owners before reserving them
+    /// side by side ([`KittyImageBudgetState::owner_floor`]); `0` ends it.
+    pub(super) fn set_kitty_image_owner_floor(self: &Arc<Self>, extra: usize) {
+        {
+            let mut budget = self.kitty_image_budget.lock().unwrap();
+            budget.owner_floor = if extra == 0 {
+                0
+            } else {
+                (Self::kitty_image_budget_owner_count(&budget) + extra)
+                    .min(KITTY_IMAGE_BUDGET_OWNER_LIMIT)
+            };
+            Self::rebalance_kitty_image_budget_owners(&mut budget);
+            self.kitty_image_budget_changed.notify_all();
+        }
+        self.start_kitty_image_budget_worker();
+    }
+
     pub(crate) fn reserve_kitty_image_surface(
         self: &Arc<Self>,
         surface: SurfaceId,
@@ -27,7 +44,10 @@ impl Mux {
             let owns_quota =
                 budget.blocked_surfaces.is_empty() && owner_count < KITTY_IMAGE_BUDGET_OWNER_LIMIT;
             if owns_quota {
-                budget.capacity = kitty_image_budget_capacity(owner_count + 1, budget.capacity);
+                budget.capacity = kitty_image_budget_capacity(
+                    (owner_count + 1).max(budget.owner_floor),
+                    budget.capacity,
+                );
             }
             budget.entries.insert(
                 surface,

cmux-tui/crates/cmux-tui-core/src/mux/terminal_adoption.rs

@@ -3,6 +3,20 @@
 
 use super::*;
 
+/// Most terminal-host handshakes a daemon start runs at once.
+#[cfg(any(unix, windows))]
+const MAX_PARALLEL_HANDSHAKES: usize = 16;
+
+/// A live host whose durable row passed every startup check: its handshake
+/// runs in [`Mux::adopt_handshakes_in_parallel`].
+#[cfg(any(unix, windows))]
+struct PendingHandshake {
+    terminal_id: String,
+    record: crate::terminal_host_runtime::TerminalHostRecord,
+    record_path: std::path::PathBuf,
+    restored_binding: Option<RestoredTerminalBinding>,
+}
+
 impl Mux {
     #[cfg(any(unix, windows))]
     pub(super) fn restored_terminal_binding(
@@ -87,6 +101,7 @@ impl Mux {
             None => Vec::new(),
         };
         let mut handled_terminals = HashSet::new();
+        let mut handshakes = Vec::new();
         // At most one warm snapshot host becomes the first terminal of a
         // fresh registry (SurfaceOptions::adopt_template_terminal).
         let mut template_claimed = false;
@@ -267,13 +282,40 @@ impl Mux {
                 continue;
             }
             let restored_binding = self.restored_terminal_binding(&terminal_id)?;
-            // Bound startup head-of-line blocking per host. One handshake is
-            // enough for healthy hosts; live or indeterminate failures
-            // continue on the asynchronous adoption loop instead of retrying
-            // serially ahead of every later terminal.
-            let adopted = self
-                .adopt_restored_terminal(restored_binding, &options, &record, &record_path)
-                .ok();
+            // Every path after the handshake marks it handled; marking it now
+            // keeps a second record of the same terminal out of this pass.
+            handled_terminals.insert(terminal_id.clone());
+            handshakes.push(PendingHandshake { terminal_id, record, record_path, restored_binding });
+        }
+
+        // The handshakes (connect, then replay the host's snapshot into the
+        // mirror) are independent per host and are most of a handoff's time:
+        // run them at once on a few threads (perf100: 113 terminals adopted
+        // one at a time kept every terminal blank for 8-20 s after an
+        // update). Their results then commit in record order, exactly as the
+        // serial loop did, so every registry and topology decision is
+        // unchanged. Bound startup head-of-line blocking per host: one
+        // handshake is enough for healthy hosts; live or indeterminate
+        // failures continue on the asynchronous adoption loop instead of
+        // retrying serially ahead of every later terminal.
+        // Ends the Kitty owner floor when the commits are done, also on an
+        // early error.
+        struct KittyOwnerFloor(Option<Arc<Mux>>);
+        impl Drop for KittyOwnerFloor {
+            fn drop(&mut self) {
+                if let Some(mux) = self.0.take() {
+                    mux.set_kitty_image_owner_floor(0);
+                }
+            }
+        }
+        let mut kitty_floor = KittyOwnerFloor(None);
+        if handshakes.len() > 1 {
+            self.set_kitty_image_owner_floor(handshakes.len());
+            kitty_floor.0 = Some(self.clone());
+        }
+        let adopted = self.adopt_handshakes_in_parallel(&options, &mut handshakes);
+        for (handshake, adopted) in handshakes.into_iter().zip(adopted) {
+            let PendingHandshake { terminal_id, record, record_path, .. } = handshake;
             let surface = match adopted {
                 Some(surface) => surface,
                 None => {
@@ -335,6 +377,9 @@ impl Mux {
             self.reap_if_dead(&surface);
         }
 
+        // Every adopted surface is committed (or gone): the budget follows
+        // the real owners again.
+        drop(kitty_floor);
         self.mark_unadoptable_terminal_hosts(&options, &mut handled_terminals)?;
 
         // No launcher survives a daemon restart. A durable lifecycle row
@@ -362,6 +407,72 @@ impl Mux {
         Ok(())
     }
 
+    /// Adopt every pending handshake's host, several at once, and return
+    /// the results in the same order (`None`: the handshake failed).
+    #[cfg(any(unix, windows))]
+    fn adopt_handshakes_in_parallel(
+        self: &Arc<Self>,
+        options: &SurfaceOptions,
+        handshakes: &mut [PendingHandshake],
+    ) -> Vec<Option<Arc<Surface>>> {
+        let count = handshakes.len();
+        // A handshake mostly waits on its host (another process builds and
+        // sends the snapshot), so the worker count does not follow the cores.
+        let workers = MAX_PARALLEL_HANDSHAKES.min(count);
+        // Test seam: a slow host handshake (debug builds only).
+        #[cfg(debug_assertions)]
+        let test_delay = std::env::var("CMUX_TUI_TEST_ADOPT_HANDSHAKE_DELAY_MS")
+            .ok()
+            .and_then(|value| value.parse::<u64>().ok())
+            .map(|ms| Duration::from_millis(ms.min(5_000)));
+        type Job<'a> = Mutex<Option<(Option<RestoredTerminalBinding>, &'a PendingHandshake)>>;
+        let jobs: Vec<Job<'_>> =
+            handshakes
+                .iter_mut()
+                .map(|handshake| {
+                    let binding = handshake.restored_binding.take();
+                    Mutex::new(Some((binding, &*handshake)))
+                })
+                .collect();
+        let results: Vec<Mutex<Option<Arc<Surface>>>> =
+            (0..count).map(|_| Mutex::new(None)).collect();
+        let next = AtomicUsize::new(0);
+        let work = || {
+            loop {
+                let index = next.fetch_add(1, Ordering::Relaxed);
+                let Some(job) = jobs.get(index) else { return };
+                let Some((binding, handshake)) = job.lock().unwrap().take() else { continue };
+                #[cfg(debug_assertions)]
+                if let Some(delay) = test_delay {
+                    std::thread::sleep(delay);
+                }
+                let adopted = self
+                    .adopt_restored_terminal(
+                        binding,
+                        options,
+                        &handshake.record,
+                        &handshake.record_path,
+                    )
+                    .ok();
+                *results[index].lock().unwrap() = adopted;
+            }
+        };
+        if workers <= 1 {
+            work();
+        } else {
+            std::thread::scope(|scope| {
+                for _ in 1..workers {
+                    // A worker that cannot start leaves its jobs to the others.
+                    let _ = std::thread::Builder::new()
+                        .name("terminal-adopt-handshake".into())
+                        .spawn_scoped(scope, work);
+                }
+                work();
+            });
+        }
+        results.into_iter().map(|result| result.into_inner().unwrap()).collect()
+    }
+
     /// Complete an adopted template terminal (complete_template_adoption),
     /// retrying in the background until it succeeds or the daemon shuts down.
     /// Adoption itself has already committed, so a failure here must neither

cmux-tui/crates/cmux-tui/tests/terminal_host_recovery.rs

@@ -5311,6 +5311,9 @@ mod host_scrollback;
 #[path = "terminal_host_recovery/output_coalescing.rs"]
 mod output_coalescing;
 
+#[path = "terminal_host_recovery/parallel_handoff.rs"]
+mod parallel_handoff;
+
 #[cfg(target_os = "macos")]
 #[path = "terminal_host_recovery/slim_host.rs"]
 mod slim_host;

cmux-tui/crates/cmux-tui/tests/terminal_host_recovery/parallel_handoff.rs

@@ -0,0 +1,109 @@
+//! A daemon handoff (an update, a crash restart) adopts its terminal hosts
+//! at once, not one after another (perf100 tui lane, bead cx-p00l3.1). The
+//! new daemon answers only after the startup adoption pass, and each host's
+//! handshake (connect, then replay its snapshot) used to run in turn: 113
+//! terminals kept every terminal blank for 8 to 20 s after an update. The
+//! handshakes now run at once and commit in record order.
+//!
+//! A debug seam makes every handshake take HANDSHAKE_MS, so the serial pass
+//! would take TERMINALS x HANDSHAKE_MS. The test measures from the new
+//! daemon's spawn to its socket answering, checks every terminal is running
+//! with its screen and its input intact, and that nothing was reordered.
+
+use super::*;
+
+const TERMINALS: usize = 16;
+const HANDSHAKE_MS: u64 = 400;
+/// Serial adoption would take TERMINALS x HANDSHAKE_MS = 6.4 s.
+const MAX_HANDOFF: Duration = Duration::from_millis(3_000);
+
+#[test]
+fn daemon_handoff_adopts_terminal_hosts_in_parallel_with_screens_and_input_intact() {
+    let mut harness = RecoveryHarness::start_unstarted("parallel-handoff");
+    harness.extra_env.push(("CMUX_TUI_TEST_ADOPT_HANDSHAKE_DELAY_MS".into(), "0".into()));
+    harness.restart();
+    let mut terminals = Vec::new();
+    for index in 0..TERMINALS {
+        let created = request(
+            &harness.socket,
+            serde_json::json!({
+                "id": 100 + index,
+                "cmd": "run",
+                "argv": ["/bin/cat"],
+                "new_workspace": true,
+                "name": format!("handoff-{index}"),
+            }),
+        );
+        let surface = created["surface"].as_u64().unwrap();
+        let terminal_id = created["terminal_id"].as_str().unwrap().to_string();
+        let marker = format!("before-handoff-{index}-{}", std::process::id());
+        request(
+            &harness.socket,
+            serde_json::json!({"id": 200 + index, "cmd": "send", "surface": surface, "text": format!("{marker}\n")}),
+        );
+        terminals.push((terminal_id, marker, surface));
+    }
+    for (_, marker, surface) in &terminals {
+        assert!(wait_for_screen(&harness.socket, *surface, marker).contains(marker.as_str()));
+    }
+    let hosts = wait_for_host_records(&harness.host_root(), TERMINALS);
+
+    harness.sigkill();
+    harness.extra_env = vec![(
+        "CMUX_TUI_TEST_ADOPT_HANDSHAKE_DELAY_MS".into(),
+        HANDSHAKE_MS.to_string(),
+    )];
+    let started = Instant::now();
+    harness.restart();
+    // Until its answer: the socket may accept before the daemon serves.
+    request(&harness.socket, serde_json::json!({"id": 299, "cmd": "identify"}));
+    let handoff = started.elapsed();
+    eprintln!(
+        "handoff: {TERMINALS} terminals with {HANDSHAKE_MS} ms handshakes answered in {:.2} s \
+         (serial: {:.1} s)",
+        handoff.as_secs_f64(),
+        (TERMINALS as u64 * HANDSHAKE_MS) as f64 / 1000.0
+    );
+
+    // Every terminal is running on its own host, with its screen.
+    for (index, (terminal_id, marker, _)) in terminals.iter().enumerate() {
+        let deadline = Instant::now() + test_timeout(Duration::from_secs(15));
+        let surface = loop {
+            let resolved = request(
+                &harness.socket,
+                serde_json::json!({"id": 300 + index, "cmd": "resolve-terminal", "terminal_id": terminal_id}),
+            );
+            if resolved["lifecycle"] == "running"
+                && let Some(surface) = resolved["surface"].as_u64()
+            {
+                break surface;
+            }
+            assert!(Instant::now() < deadline, "terminal {terminal_id} not adopted: {resolved}");
+            std::thread::sleep(Duration::from_millis(25));
+        };
+        assert!(wait_for_screen(&harness.socket, surface, marker).contains(marker.as_str()));
+        // Input after the handoff lands after the old screen, in order.
+        let after = format!("after-handoff-{index}-{}", std::process::id());
+        request(
+            &harness.socket,
+            serde_json::json!({"id": 400 + index, "cmd": "send", "surface": surface, "text": format!("{after}\n")}),
+        );
+        let screen = wait_for_screen(&harness.socket, surface, &after);
+        let before_at = screen.find(marker.as_str()).expect("old screen kept");
+        let after_at = screen.find(&after).expect("new input shown");
+        assert!(before_at < after_at, "output reordered on {terminal_id}: {screen}");
+    }
+    let adopted = wait_for_host_records(&harness.host_root(), TERMINALS);
+    let mut before_pids: Vec<u32> = hosts.iter().map(|(_, record)| record.host_pid).collect();
+    let mut after_pids: Vec<u32> = adopted.iter().map(|(_, record)| record.host_pid).collect();
+    before_pids.sort_unstable();
+    after_pids.sort_unstable();
+    assert_eq!(before_pids, after_pids, "the handoff replaced terminal hosts");
+    assert!(
+        handoff <= test_timeout(MAX_HANDOFF),
+        "the handoff took {:.2} s for {TERMINALS} terminals (budget {:.1} s; serial is {:.1} s)",
+        handoff.as_secs_f64(),
+        MAX_HANDOFF.as_secs_f64(),
+        (TERMINALS as u64 * HANDSHAKE_MS) as f64 / 1000.0
+    );
+}