perf(maestro): run tmux subprocess calls off the async runtime (NAP-256)
ci / rust (push) Failing after 6m50s
ci / rust (push) Failing after 6m50s
Every `SessionManager` tmux call shelled out via blocking `std::process::Command::output()` on the tokio worker — including `send_keys` on the per-keystroke hot path. Under concurrent typing the small worker pool stalls on `tmux` fork+waits (tens of ms on a loaded host), starving HTTP handlers, the registry WS, the liveness watcher, and PTY readers. Wrap each call site (create/kill/recover/liveness/send_keys/resize/pipe_pane) in `tokio::task::spawn_blocking` via a small `tmux_blocking` helper, so the blocking fork+wait runs on the blocking pool and never parks an async worker. The `Tmux` API stays synchronous (lower blast radius than an async migration; its tests are unchanged). Full daemon suite + send_keys/broker e2e green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -228,12 +228,14 @@ impl SessionManager {
|
||||
return Err(e.into());
|
||||
}
|
||||
};
|
||||
if let Err(e) = self.inner.tmux.new_session_piped(
|
||||
id,
|
||||
&effective_cwd,
|
||||
&req.command,
|
||||
&pipe_pane_command(&fifo_path),
|
||||
) {
|
||||
let spawned = {
|
||||
let tmux = self.inner.tmux.clone();
|
||||
let cwd = effective_cwd.clone();
|
||||
let command = req.command.clone();
|
||||
let sink = pipe_pane_command(&fifo_path);
|
||||
tmux_blocking(move || tmux.new_session_piped(id, &cwd, &command, &sink)).await
|
||||
};
|
||||
if let Err(e) = spawned {
|
||||
// `bridge` drops at the end of this block, closing the FIFO reader.
|
||||
let _ = std::fs::remove_file(&fifo_path);
|
||||
self.rollback_worktree(&worktree_base_repo, &worktree_path);
|
||||
@@ -254,7 +256,8 @@ impl SessionManager {
|
||||
if let Err(e) = self.inner.sidecar.write(&session) {
|
||||
// tmux already spawned the session; kill it so no ghost survives a
|
||||
// failed sidecar write, then unwind the FIFO + worktree.
|
||||
let _ = self.inner.tmux.kill_session(id);
|
||||
let tmux = self.inner.tmux.clone();
|
||||
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
|
||||
let _ = std::fs::remove_file(&fifo_path);
|
||||
self.rollback_worktree(&worktree_base_repo, &worktree_path);
|
||||
return Err(SessionError::Sidecar(e));
|
||||
@@ -297,7 +300,8 @@ impl SessionManager {
|
||||
};
|
||||
let record = removed.ok_or(SessionError::NotFound(id))?;
|
||||
|
||||
let _ = self.inner.tmux.kill_session(id);
|
||||
let tmux = self.inner.tmux.clone();
|
||||
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
|
||||
let _ = std::fs::remove_file(&record.fifo_path);
|
||||
self.inner
|
||||
.sidecar
|
||||
@@ -346,7 +350,10 @@ impl SessionManager {
|
||||
/// us to repair them.
|
||||
pub async fn recover(&self) -> Result<usize, SessionError> {
|
||||
tokio::fs::create_dir_all(&self.inner.fifo_dir).await?;
|
||||
let live = self.inner.tmux.list_sessions()?;
|
||||
let live = {
|
||||
let tmux = self.inner.tmux.clone();
|
||||
tmux_blocking(move || tmux.list_sessions()).await?
|
||||
};
|
||||
let mut adopted = 0usize;
|
||||
for id in live {
|
||||
let mut session = match self.inner.sidecar.read(id) {
|
||||
@@ -381,10 +388,11 @@ impl SessionManager {
|
||||
// Re-pipe so output starts flowing again. `Tmux::pipe_pane` runs
|
||||
// without tmux's `-o` flag, so any stale pipe the previous daemon
|
||||
// left running to a now-unread FIFO is replaced (NAP-228).
|
||||
let _ = self
|
||||
.inner
|
||||
.tmux
|
||||
.pipe_pane(id, &pipe_pane_command(&fifo_path));
|
||||
let _ = {
|
||||
let tmux = self.inner.tmux.clone();
|
||||
let sink = pipe_pane_command(&fifo_path);
|
||||
tmux_blocking(move || tmux.pipe_pane(id, &sink)).await
|
||||
};
|
||||
let mut sessions = self.inner.sessions.lock().await;
|
||||
sessions.insert(
|
||||
id,
|
||||
@@ -407,9 +415,10 @@ impl SessionManager {
|
||||
if !self.has(id).await {
|
||||
return Err(SessionError::NotFound(id));
|
||||
}
|
||||
self.inner
|
||||
.tmux
|
||||
.send_keys(id, bytes)
|
||||
let tmux = self.inner.tmux.clone();
|
||||
let bytes = bytes.to_vec();
|
||||
tmux_blocking(move || tmux.send_keys(id, &bytes))
|
||||
.await
|
||||
.map_err(SessionError::Tmux)
|
||||
}
|
||||
|
||||
@@ -417,13 +426,28 @@ impl SessionManager {
|
||||
if !self.has(id).await {
|
||||
return Err(SessionError::NotFound(id));
|
||||
}
|
||||
self.inner
|
||||
.tmux
|
||||
.resize_window(id, cols, rows)
|
||||
let tmux = self.inner.tmux.clone();
|
||||
tmux_blocking(move || tmux.resize_window(id, cols, rows))
|
||||
.await
|
||||
.map_err(SessionError::Tmux)
|
||||
}
|
||||
}
|
||||
|
||||
/// Run a blocking `tmux` subprocess call on the blocking thread pool so it
|
||||
/// never stalls a tokio worker. `tmux` is a fork+exec that can take tens of
|
||||
/// milliseconds on a loaded host, and `send_keys` rides the per-keystroke hot
|
||||
/// path — running it inline would starve HTTP handlers, the registry WS, the
|
||||
/// liveness watcher, and PTY readers (NAP-256).
|
||||
async fn tmux_blocking<T, F>(f: F) -> T
|
||||
where
|
||||
F: FnOnce() -> T + Send + 'static,
|
||||
T: Send + 'static,
|
||||
{
|
||||
tokio::task::spawn_blocking(f)
|
||||
.await
|
||||
.expect("tmux blocking task panicked")
|
||||
}
|
||||
|
||||
/// One pass of the liveness watcher: diff the in-memory session set against
|
||||
/// the tmux-reported one, and reap anything we still track but tmux has lost.
|
||||
async fn run_liveness_pass(inner: &Arc<Inner>) {
|
||||
@@ -439,12 +463,15 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
|
||||
sessions.keys().copied().collect()
|
||||
};
|
||||
|
||||
let live_ids = match inner.tmux.list_sessions() {
|
||||
let live_ids = {
|
||||
let tmux = inner.tmux.clone();
|
||||
match tmux_blocking(move || tmux.list_sessions()).await {
|
||||
Ok(ids) => ids,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = ?e, "liveness watcher: tmux list-sessions failed");
|
||||
return;
|
||||
}
|
||||
}
|
||||
};
|
||||
let live: HashSet<Uuid> = live_ids.into_iter().collect();
|
||||
|
||||
|
||||
Reference in New Issue
Block a user