From 33b6c74e04730e4a034601bce8608fd293bce3b2 Mon Sep 17 00:00:00 2001 From: hanafish <1106510024@qq.com> Date: Fri, 7 Aug 2026 22:32:33 +0800 Subject: [PATCH] perf(benchmark): bound terminal run/batch registries --- src-tauri/src/benchmark/agent_batch.rs | 28 +- src-tauri/src/benchmark/commands.rs | 6 +- src-tauri/src/benchmark/mod.rs | 29 ++ src-tauri/src/benchmark/process.rs | 36 +++ src-tauri/src/benchmark/retention.rs | 368 +++++++++++++++++++++++++ src-tauri/src/benchmark/run.rs | 2 + src-tauri/src/benchmark/swe_bench.rs | 9 +- src-tauri/src/lib.rs | 3 + 8 files changed, 463 insertions(+), 18 deletions(-) create mode 100644 src-tauri/src/benchmark/retention.rs diff --git a/src-tauri/src/benchmark/agent_batch.rs b/src-tauri/src/benchmark/agent_batch.rs index 2fb8224e16..a6f8f68c25 100644 --- a/src-tauri/src/benchmark/agent_batch.rs +++ b/src-tauri/src/benchmark/agent_batch.rs @@ -14,6 +14,7 @@ use super::dto::{ use super::history::{load_agent_batch_history, persist_agent_batch_status}; use super::launch::{benchmark_agent_prompt, benchmark_launch_params}; use super::paths::benchmark_agent_submission_patch_path; +use super::retention::prune_terminal_agent_batches; use super::run::trim_logs; use super::{ BENCHMARK_AGENT_BATCHES, BENCHMARK_AGENT_BATCH_STATUS_CANCELLED, @@ -29,10 +30,9 @@ pub(super) async fn load_agent_batch_for_update( return Ok(status); } let status = load_agent_batch_history(batch_id)?; - BENCHMARK_AGENT_BATCHES - .lock() - .await - .insert(batch_id.to_string(), status.clone()); + let mut batches = BENCHMARK_AGENT_BATCHES.lock().await; + batches.insert(batch_id.to_string(), status.clone()); + prune_terminal_agent_batches(&mut batches); Ok(status) } @@ -40,10 +40,10 @@ pub(super) async fn persist_updated_agent_batch( mut batch: BenchmarkAgentBatchStatus, ) -> Result { refresh_agent_batch_counts(&mut batch); - BENCHMARK_AGENT_BATCHES - .lock() - .await - .insert(batch.batch_id.clone(), batch.clone()); + let mut batches = BENCHMARK_AGENT_BATCHES.lock().await; + batches.insert(batch.batch_id.clone(), batch.clone()); + prune_terminal_agent_batches(&mut batches); + drop(batches); persist_agent_batch_status(&batch)?; Ok(batch) } @@ -71,10 +71,10 @@ pub(super) async fn refresh_agent_batch_evaluations( } drop(runs); if changed { - BENCHMARK_AGENT_BATCHES - .lock() - .await - .insert(batch.batch_id.clone(), batch.clone()); + let mut batches = BENCHMARK_AGENT_BATCHES.lock().await; + batches.insert(batch.batch_id.clone(), batch.clone()); + prune_terminal_agent_batches(&mut batches); + drop(batches); persist_agent_batch_status(&batch)?; } Ok(batch) @@ -237,7 +237,9 @@ where update(item); } refresh_agent_batch_counts(batch); - batch.clone() + let status = batch.clone(); + prune_terminal_agent_batches(&mut batches); + status }; if let Err(error) = persist_agent_batch_status(&status) { tracing::warn!( diff --git a/src-tauri/src/benchmark/commands.rs b/src-tauri/src/benchmark/commands.rs index 45a5daedfa..75917db904 100644 --- a/src-tauri/src/benchmark/commands.rs +++ b/src-tauri/src/benchmark/commands.rs @@ -43,6 +43,7 @@ use super::launch::{ use super::paths::benchmark_agent_submission_patch_path; use super::preflight::{run_swe_bench_preflight, run_terminal_bench_preflight}; use super::process::terminate_process; +use super::retention::{prune_terminal_agent_batches, prune_terminal_runs}; use super::run::trim_logs; use super::swe_bench::{ build_swe_bench_run_plan, run_swe_bench_patch_only_worktree, run_swe_bench_process, @@ -215,7 +216,9 @@ pub async fn benchmark_cancel_run( .push("Cancel requested for evaluator process.".to_string()); trim_logs(&mut status.logs); } - status.process_id + let process_id = status.process_id; + prune_terminal_runs(&mut runs); + process_id }; if let Some(pid) = process_id { @@ -623,6 +626,7 @@ pub async fn benchmark_cancel_agent_batch( batch.finished_at = Some(now); refresh_agent_batch_counts(batch); let status = batch.clone(); + prune_terminal_agent_batches(&mut batches); drop(batches); persist_agent_batch_status(&status)?; Ok(status) diff --git a/src-tauri/src/benchmark/mod.rs b/src-tauri/src/benchmark/mod.rs index 4bdba39fe2..485adf0f14 100644 --- a/src-tauri/src/benchmark/mod.rs +++ b/src-tauri/src/benchmark/mod.rs @@ -22,6 +22,7 @@ mod launch; mod paths; mod preflight; mod process; +mod retention; mod run; mod swe_bench; @@ -74,3 +75,31 @@ static BENCHMARK_RUNS: LazyLock>>> LazyLock::new(|| Arc::new(Mutex::new(HashMap::new()))); static BENCHMARK_AGENT_BATCHES: LazyLock>>> = LazyLock::new(|| Arc::new(Mutex::new(HashMap::new()))); + +/// Best-effort termination of evaluator processes that are still running +/// when the app exits. Called from the `ExitRequested` handler alongside the +/// other subprocess cleanup; without it the spawned Python/Docker evaluators +/// would outlive the app as orphans. +pub fn terminate_running_evaluators_sync() { + let Ok(runs) = BENCHMARK_RUNS.try_lock() else { + tracing::warn!( + "[benchmark] runs registry locked during shutdown; skipping evaluator cleanup" + ); + return; + }; + let process_ids: Vec = runs + .values() + .filter(|run| run.status == BENCHMARK_RUN_STATUS_RUNNING) + .filter_map(|run| run.process_id) + .collect(); + drop(runs); + for process_id in process_ids { + if let Err(error) = process::terminate_process_sync(process_id) { + tracing::warn!( + process_id, + error = %error, + "[benchmark] failed to terminate evaluator during shutdown" + ); + } + } +} diff --git a/src-tauri/src/benchmark/process.rs b/src-tauri/src/benchmark/process.rs index c22aeb94b4..c888e62d63 100644 --- a/src-tauri/src/benchmark/process.rs +++ b/src-tauri/src/benchmark/process.rs @@ -190,6 +190,42 @@ pub(super) async fn run_python_import( }) } +/// Synchronous variant of [`terminate_process`] for shutdown paths that run +/// outside the async runtime (the Tauri `ExitRequested` handler). +pub(super) fn terminate_process_sync(process_id: u32) -> Result<(), String> { + #[cfg(unix)] + let output = std::process::Command::new("kill") + .arg("-TERM") + .arg(process_id.to_string()) + .output(); + + #[cfg(windows)] + let output = { + use std::os::windows::process::CommandExt; + let mut cmd = std::process::Command::new("taskkill"); + cmd.arg("/PID") + .arg(process_id.to_string()) + .arg("/T") + .arg("/F"); + // Suppress the console window on Windows. + cmd.creation_flags(app_platform::CREATE_NO_WINDOW); + cmd.output() + }; + + let output = + output.map_err(|error| format!("Failed to terminate process {process_id}: {error}"))?; + if output.status.success() { + Ok(()) + } else { + let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string(); + Err(if stderr.is_empty() { + format!("Process termination exited with {}", output.status) + } else { + stderr + }) + } +} + pub(super) async fn terminate_process(process_id: u32) -> Result<(), String> { #[cfg(unix)] let output = Command::new("kill") diff --git a/src-tauri/src/benchmark/retention.rs b/src-tauri/src/benchmark/retention.rs new file mode 100644 index 0000000000..231be767d7 --- /dev/null +++ b/src-tauri/src/benchmark/retention.rs @@ -0,0 +1,368 @@ +//! Bounded retention for the in-memory benchmark registries. +//! +//! `BENCHMARK_RUNS` and `BENCHMARK_AGENT_BATCHES` are process-wide statics +//! that previously only ever grew. Active entries must stay resident for as +//! long as the run/batch is in flight, but terminal entries only serve status +//! polls issued shortly after completion, so we keep the most recently +//! finished `MAX_TERMINAL_*` terminal entries per registry and evict the rest. +//! +//! A count-based budget was chosen over a TTL because it bounds memory even +//! when many runs finish inside a short window, needs no sweep timer, and is +//! deterministic to test. Agent batches are persisted to disk on every +//! mutation and are reloaded from history on demand, so evicting them from +//! memory is lossless. Runs have no persistence layer; evicting a terminal +//! run only affects `benchmark_get_run_status` lookups for run ids older +//! than the retention budget. + +use std::collections::HashMap; + +use super::dto::{BenchmarkAgentBatchStatus, BenchmarkRunStatus}; +use super::{ + BENCHMARK_AGENT_BATCH_STATUS_CANCELLED, BENCHMARK_AGENT_BATCH_STATUS_FAILED, + BENCHMARK_AGENT_BATCH_STATUS_LAUNCHED, BENCHMARK_RUN_STATUS_APPLIED, + BENCHMARK_RUN_STATUS_CANCELLED, BENCHMARK_RUN_STATUS_FAILED, BENCHMARK_RUN_STATUS_PASSED, +}; + +/// Terminal run entries kept in memory (most recently finished first). +pub(super) const MAX_TERMINAL_BENCHMARK_RUNS: usize = 50; +/// Terminal agent batch entries kept in memory (most recently finished first). +pub(super) const MAX_TERMINAL_BENCHMARK_AGENT_BATCHES: usize = 50; + +/// Snapshot of one registry entry, as consumed by the eviction policy. +pub(super) struct RetentionEntry { + pub key: String, + /// Terminal entries are eviction candidates; active entries never are. + pub terminal: bool, + /// RFC3339 completion timestamp. Every writer uses + /// `Utc::now().to_rfc3339()`, so lexicographic order matches + /// chronological order. `None` sorts as oldest. + pub finished_at: Option, +} + +/// Pure eviction policy: keep every active entry, keep the `budget` most +/// recently finished terminal entries, and return the keys of the remaining +/// terminal entries. Ties on `finished_at` break by key so the result is +/// deterministic and repeated applications are idempotent. +pub(super) fn terminal_keys_to_evict(entries: &[RetentionEntry], budget: usize) -> Vec { + let mut terminal: Vec<&RetentionEntry> = + entries.iter().filter(|entry| entry.terminal).collect(); + if terminal.len() <= budget { + return Vec::new(); + } + terminal.sort_by(|left, right| { + right + .finished_at + .cmp(&left.finished_at) + .then_with(|| right.key.cmp(&left.key)) + }); + terminal[budget..] + .iter() + .map(|entry| entry.key.clone()) + .collect() +} + +/// A run is terminal once it can no longer transition or hold a live +/// process. Unknown statuses are treated as active so they are never +/// evicted by mistake. +pub(super) fn is_terminal_run_status(status: &str) -> bool { + matches!( + status, + BENCHMARK_RUN_STATUS_PASSED + | BENCHMARK_RUN_STATUS_FAILED + | BENCHMARK_RUN_STATUS_CANCELLED + | BENCHMARK_RUN_STATUS_APPLIED + ) +} + +/// A batch is terminal once no item is queued or running. Unknown statuses +/// are treated as active so they are never evicted by mistake. +pub(super) fn is_terminal_agent_batch_status(status: &str) -> bool { + matches!( + status, + BENCHMARK_AGENT_BATCH_STATUS_LAUNCHED + | BENCHMARK_AGENT_BATCH_STATUS_FAILED + | BENCHMARK_AGENT_BATCH_STATUS_CANCELLED + ) +} + +/// Evict terminal runs beyond the retention budget. Callers must hold the +/// `BENCHMARK_RUNS` lock and invoke this whenever an entry is inserted in or +/// transitions to a terminal status. +pub(super) fn prune_terminal_runs(runs: &mut HashMap) { + let entries: Vec = runs + .iter() + .map(|(key, status)| RetentionEntry { + key: key.clone(), + terminal: is_terminal_run_status(&status.status), + finished_at: status.finished_at.clone(), + }) + .collect(); + for key in terminal_keys_to_evict(&entries, MAX_TERMINAL_BENCHMARK_RUNS) { + runs.remove(&key); + } +} + +/// Evict terminal agent batches beyond the retention budget. Callers must +/// hold the `BENCHMARK_AGENT_BATCHES` lock and invoke this whenever an entry +/// is inserted in or transitions to a terminal status. Evicted batches stay +/// available through the on-disk batch history. +pub(super) fn prune_terminal_agent_batches( + batches: &mut HashMap, +) { + let entries: Vec = batches + .iter() + .map(|(key, status)| RetentionEntry { + key: key.clone(), + terminal: is_terminal_agent_batch_status(&status.status), + finished_at: status.finished_at.clone(), + }) + .collect(); + for key in terminal_keys_to_evict(&entries, MAX_TERMINAL_BENCHMARK_AGENT_BATCHES) { + batches.remove(&key); + } +} + +#[cfg(test)] +mod tests { + use super::super::{ + BENCHMARK_AGENT_BATCH_STATUS_QUEUED, BENCHMARK_AGENT_BATCH_STATUS_RUNNING, + BENCHMARK_RUN_STATUS_RUNNING, + }; + use super::*; + + fn entry(key: &str, terminal: bool, finished_at: Option<&str>) -> RetentionEntry { + RetentionEntry { + key: key.to_string(), + terminal, + finished_at: finished_at.map(ToOwned::to_owned), + } + } + + fn timestamp(index: usize) -> String { + // Same shape as `Utc::now().to_rfc3339()`; later index == more recent. + format!("2026-08-07T10:{:02}:00+00:00", index) + } + + #[test] + fn keeps_all_active_entries_even_over_budget() { + let entries: Vec = (0..10) + .map(|index| entry(&format!("run-{index}"), false, Some(×tamp(index)))) + .collect(); + assert!(terminal_keys_to_evict(&entries, 3).is_empty()); + } + + #[test] + fn evicts_oldest_terminal_entries_beyond_budget() { + let entries: Vec = (0..5) + .map(|index| entry(&format!("run-{index}"), true, Some(×tamp(index)))) + .collect(); + let mut evicted = terminal_keys_to_evict(&entries, 3); + evicted.sort(); + assert_eq!(evicted, vec!["run-0".to_string(), "run-1".to_string()]); + } + + #[test] + fn active_entries_do_not_count_toward_terminal_budget() { + let mut entries: Vec = (0..10) + .map(|index| entry(&format!("active-{index}"), false, None)) + .collect(); + entries.extend( + (0..4).map(|index| entry(&format!("done-{index}"), true, Some(×tamp(index)))), + ); + let evicted = terminal_keys_to_evict(&entries, 3); + assert_eq!(evicted, vec!["done-0".to_string()]); + assert!(!evicted.iter().any(|key| key.starts_with("active-"))); + } + + #[test] + fn is_idempotent_when_reapplied() { + let mut entries: Vec = (0..6) + .map(|index| entry(&format!("run-{index}"), true, Some(×tamp(index)))) + .collect(); + let evicted = terminal_keys_to_evict(&entries, 2); + assert_eq!(evicted.len(), 4); + entries.retain(|entry| !evicted.contains(&entry.key)); + assert!(terminal_keys_to_evict(&entries, 2).is_empty()); + } + + #[test] + fn missing_timestamps_are_evicted_first() { + let entries = vec![ + entry("no-timestamp", true, None), + entry("recent", true, Some(×tamp(2))), + entry("old", true, Some(×tamp(1))), + ]; + let evicted = terminal_keys_to_evict(&entries, 2); + assert_eq!(evicted, vec!["no-timestamp".to_string()]); + } + + #[test] + fn equal_timestamps_break_ties_by_key_deterministically() { + let shared = timestamp(1); + let entries = vec![ + entry("b", true, Some(&shared)), + entry("a", true, Some(&shared)), + entry("c", true, Some(&shared)), + ]; + assert_eq!(terminal_keys_to_evict(&entries, 2), vec!["a".to_string()]); + assert_eq!( + terminal_keys_to_evict(&entries, 2), + terminal_keys_to_evict(&entries, 2) + ); + } + + #[test] + fn zero_budget_evicts_every_terminal_entry() { + let entries = vec![ + entry("done", true, Some(×tamp(1))), + entry("active", false, None), + ]; + assert_eq!( + terminal_keys_to_evict(&entries, 0), + vec!["done".to_string()] + ); + } + + #[test] + fn run_status_classification_matches_lifecycle() { + assert!(!is_terminal_run_status(BENCHMARK_RUN_STATUS_RUNNING)); + assert!(is_terminal_run_status(BENCHMARK_RUN_STATUS_PASSED)); + assert!(is_terminal_run_status(BENCHMARK_RUN_STATUS_FAILED)); + assert!(is_terminal_run_status(BENCHMARK_RUN_STATUS_CANCELLED)); + assert!(is_terminal_run_status(BENCHMARK_RUN_STATUS_APPLIED)); + // Unknown statuses stay resident rather than risking eviction of an + // entry that may still transition. + assert!(!is_terminal_run_status("mystery")); + } + + #[test] + fn agent_batch_status_classification_matches_lifecycle() { + assert!(!is_terminal_agent_batch_status( + BENCHMARK_AGENT_BATCH_STATUS_QUEUED + )); + assert!(!is_terminal_agent_batch_status( + BENCHMARK_AGENT_BATCH_STATUS_RUNNING + )); + assert!(is_terminal_agent_batch_status( + BENCHMARK_AGENT_BATCH_STATUS_LAUNCHED + )); + assert!(is_terminal_agent_batch_status( + BENCHMARK_AGENT_BATCH_STATUS_FAILED + )); + assert!(is_terminal_agent_batch_status( + BENCHMARK_AGENT_BATCH_STATUS_CANCELLED + )); + assert!(!is_terminal_agent_batch_status("mystery")); + } + + fn run_status(run_id: &str, status: &str, finished_at: Option<&str>) -> BenchmarkRunStatus { + BenchmarkRunStatus { + run_id: run_id.to_string(), + benchmark_kind: "swe_bench_pro".to_string(), + evaluation_mode: "local_docker".to_string(), + task_id: "task".to_string(), + status: status.to_string(), + source_path: String::new(), + repo_path: None, + patch_path: String::new(), + output_dir: String::new(), + worktree_path: None, + started_at: None, + finished_at: finished_at.map(ToOwned::to_owned), + exit_code: None, + process_id: None, + logs: Vec::new(), + result: None, + error: None, + } + } + + #[test] + fn prune_terminal_runs_keeps_active_and_recent_terminal_entries() { + let mut runs = HashMap::new(); + for index in 0..MAX_TERMINAL_BENCHMARK_RUNS + 5 { + let run_id = format!("done-{index}"); + runs.insert( + run_id.clone(), + run_status( + &run_id, + BENCHMARK_RUN_STATUS_PASSED, + Some(&format!( + "2026-08-07T{:02}:{:02}:00+00:00", + index / 60, + index % 60 + )), + ), + ); + } + runs.insert( + "active".to_string(), + run_status("active", BENCHMARK_RUN_STATUS_RUNNING, None), + ); + + prune_terminal_runs(&mut runs); + + assert!(runs.contains_key("active")); + assert_eq!(runs.len(), MAX_TERMINAL_BENCHMARK_RUNS + 1); + assert!(!runs.contains_key("done-0")); + assert!(runs.contains_key(&format!("done-{}", MAX_TERMINAL_BENCHMARK_RUNS + 4))); + } + + fn batch_status( + batch_id: &str, + status: &str, + finished_at: Option<&str>, + ) -> BenchmarkAgentBatchStatus { + BenchmarkAgentBatchStatus { + batch_id: batch_id.to_string(), + benchmark_kind: "swe_bench_pro".to_string(), + source_path: String::new(), + launch: None, + master_session_id: String::new(), + master_session_name: String::new(), + status: status.to_string(), + total_tasks: 0, + queued: 0, + running: 0, + launched: 0, + failed: 0, + cancelled: 0, + created_at: String::new(), + started_at: None, + finished_at: finished_at.map(ToOwned::to_owned), + concurrency: 1, + items: Vec::new(), + error: None, + } + } + + #[test] + fn prune_terminal_agent_batches_keeps_active_and_recent_terminal_entries() { + let mut batches = HashMap::new(); + for index in 0..MAX_TERMINAL_BENCHMARK_AGENT_BATCHES + 3 { + let batch_id = format!("done-{index}"); + batches.insert( + batch_id.clone(), + batch_status( + &batch_id, + BENCHMARK_AGENT_BATCH_STATUS_LAUNCHED, + Some(&format!( + "2026-08-07T{:02}:{:02}:00+00:00", + index / 60, + index % 60 + )), + ), + ); + } + batches.insert( + "active".to_string(), + batch_status("active", BENCHMARK_AGENT_BATCH_STATUS_RUNNING, None), + ); + + prune_terminal_agent_batches(&mut batches); + + assert!(batches.contains_key("active")); + assert_eq!(batches.len(), MAX_TERMINAL_BENCHMARK_AGENT_BATCHES + 1); + assert!(!batches.contains_key("done-0")); + } +} diff --git a/src-tauri/src/benchmark/run.rs b/src-tauri/src/benchmark/run.rs index aa6537523f..cbf19d4642 100644 --- a/src-tauri/src/benchmark/run.rs +++ b/src-tauri/src/benchmark/run.rs @@ -3,6 +3,7 @@ use chrono::Utc; use serde_json::Value; +use super::retention::prune_terminal_runs; use super::{BENCHMARK_RUNS, BENCHMARK_RUN_STATUS_CANCELLED, MAX_RUN_LOG_LINES}; pub(super) async fn set_run_process_id(run_id: &str, process_id: u32) { @@ -60,6 +61,7 @@ pub(super) async fn finish_run_with_result( .logs .push(format!("Run finished with status: {status_value}")); trim_logs(&mut status.logs); + prune_terminal_runs(&mut runs); } } diff --git a/src-tauri/src/benchmark/swe_bench.rs b/src-tauri/src/benchmark/swe_bench.rs index 4343f44b6f..0401b353eb 100644 --- a/src-tauri/src/benchmark/swe_bench.rs +++ b/src-tauri/src/benchmark/swe_bench.rs @@ -22,6 +22,7 @@ use super::paths::{ }; use super::preflight::run_swe_bench_preflight; use super::process::{command_version_in_dir, ensure_benchmark_python_env}; +use super::retention::prune_terminal_runs; use super::run::{append_run_log, finish_run, finish_run_with_result, set_run_process_id}; use super::{ BENCHMARK_RUNS, BENCHMARK_RUN_STATUS_APPLIED, BENCHMARK_RUN_STATUS_FAILED, @@ -200,10 +201,10 @@ pub(super) async fn run_swe_bench_patch_only_worktree( }, }; - BENCHMARK_RUNS - .lock() - .await - .insert(plan.run_id.clone(), status_value.clone()); + let mut runs = BENCHMARK_RUNS.lock().await; + runs.insert(plan.run_id.clone(), status_value.clone()); + prune_terminal_runs(&mut runs); + drop(runs); Ok(status_value) } diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 544b863989..503d92b74f 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1177,6 +1177,9 @@ pub fn run() { app_handle .state::<::terminal::pty_commands::pty::PtyState>() .shutdown_kill_all(); + // Terminate benchmark evaluator subprocesses still running so + // they don't outlive the app as orphans. + benchmark::terminate_running_evaluators_sync(); } _ => {} }