From f010006a668352536afee3f7c6dc6c215d5834b7 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Wed, 19 Aug 2026 00:09:00 +0800 Subject: [PATCH] =?UTF-8?q?fix(concurrency):=20=E6=89=AB=E6=8F=8F=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=E2=80=94=E2=80=94bash=20=E5=8F=96=E6=B6=88=E6=B3=84?= =?UTF-8?q?=E6=BC=8F=E4=B8=8E=E5=AD=90=E4=BB=A3=E7=90=86=E9=98=9F=E5=88=97?= =?UTF-8?q?=E9=98=BB=E5=A1=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - bash: Command 补 kill_on_drop(true)。原先 /stop 的 select! 竞速 drop run_command future 时，Child detach 不杀进程、read_stream 任务因管道永不 EOF 而永久存活，每次取消泄漏孤儿进程+2任务；进程被杀后管道 EOF 同时解决读取任务泄漏 - runtime: sub_done send().await 改 5s 超时保护。队列满且无消费者时永久阻塞会占住全局 Semaphore permit，8 permit 耗尽后整个子代理子系统停摆；超时放弃结果由 DB 状态 + wait 对账兜底 - runtime: cancel_pending_for_topic / reap_orphan_subagents 的 parking_lot registry 锁不再跨 SQLite 调用持有（锁内收集、锁外落库），SQLite 卡顿时不再放大 /stop 延迟 --- src/tools/bash.rs | 5 ++ src/tools/task/runtime.rs | 158 +++++++++++++++++++++++--------------- 2 files changed, 101 insertions(+), 62 deletions(-) diff --git a/src/tools/bash.rs b/src/tools/bash.rs index a665391..357d961 100644 --- a/src/tools/bash.rs +++ b/src/tools/bash.rs @@ -465,6 +465,11 @@ impl BashTool { .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) + // 外部取消安全：run_command future 被 drop 时（/stop 的 select! 竞速、 + // 子代理超时等），Child drop 会终止 OS 进程。否则子进程变孤儿、 + // read_stream 任务因管道永不 EOF 而永久存活，每次取消泄漏一组资源。 + // 进程被杀后管道到达 EOF，read_stream 任务也随之自然退出。 + .kill_on_drop(true) .current_dir(cwd); let mut child = cmd.spawn().map_err(|e| format!("Failed to spawn: {}", e))?; diff --git a/src/tools/task/runtime.rs b/src/tools/task/runtime.rs index 05e654f..7a08b42 100644 --- a/src/tools/task/runtime.rs +++ b/src/tools/task/runtime.rs @@ -51,6 +51,37 @@ impl Drop for CancelRegistryGuard { } } +/// 超时保护地发送 SubagentResult。 +/// +/// sub_done_q 容量有限（32）且仅由 wait 协调器消费。若主代理从不调用 wait， +/// 无超时的 `send().await` 会永久阻塞——而本函数运行在持有全局并发许可 +/// （Semaphore permit）的 spawn 任务末尾，阻塞会永久占住 permit，8 个 permit +/// 耗尽后整个子代理子系统停摆。超时后放弃结果：DB 状态照常更新， +/// wait 侧另有 query_pending_task_ids 对账兜底，不会丢失完成事实。 +async fn send_sub_done_timeout( + sender: &tokio::sync::mpsc::Sender, + result: SubagentResult, +) { + let task_id = result.task_id.clone(); + match tokio::time::timeout(Duration::from_secs(5), sender.send(result)).await { + Ok(Ok(())) => {} + Ok(Err(e)) => { + tracing::warn!( + error = %e, + task_id = %task_id, + "Failed to send SubagentResult to sub_done_q (receiver dropped?)" + ); + } + Err(_) => { + tracing::warn!( + task_id = %task_id, + "Timed out sending SubagentResult (queue full, no consumer); \ + dropping result — DB status update + wait reconciliation cover this" + ); + } + } +} + use crate::agent::{ AgentLoop, AgentRuntimeConfig, EmittedMessageHandler, PersistingEmittedMessageHandler, SystemPrompt, SystemPromptContext, SystemPromptProvider, @@ -1118,7 +1149,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime { }) .unwrap_or_default(), }; - let _ = sub_done_sender.send(result).await; + let _ = send_sub_done_timeout(&sub_done_sender, result).await; let _ = store.update_pending_subagent_status(&task_id_for_spawn, "failed"); // _registry_guard drop 时清理 registry 条目 return; @@ -1183,13 +1214,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime { output, pending_task_ids, }; - if let Err(e) = sub_done_sender.send(result).await { - tracing::warn!( - error = %e, - task_id = %task_id_for_spawn, - "Failed to send SubagentResult to sub_done_q (receiver dropped?)" - ); - } + send_sub_done_timeout(&sub_done_sender, result).await; // UPDATE pending_subagents 状态 let status_str = match status { @@ -1419,47 +1444,55 @@ impl SubAgentRuntime for DefaultSubAgentRuntime { "Cancelling pending subagents for topic" ); - // 触发每个子代理的 CancellationToken - let registry = self.cancel_registry.lock(); - for record in &running { - if let Some(token) = registry.get(&record.task_id) { - token.cancel(); - tracing::info!( - task_id = %record.task_id, - "Cancelled subagent token" - ); - } else { - // token 不在 registry 中（可能已完成但 DB 状态未更新，或进程重启后丢失） - // 不变量 1：条件 UPDATE，仅在 status='running' 时转为 cancelled， - // 避免 spawn 已完成的终态被覆盖（completed → cancelled 是非法转换） - match self.store.try_update_pending_subagent_status( - &record.task_id, - "running", - "cancelled", - ) { - Ok(true) => { - tracing::info!( - task_id = %record.task_id, - "Marked subagent as cancelled in DB (token not in registry)" - ); - } - Ok(false) => { - tracing::info!( - task_id = %record.task_id, - "Subagent status already updated by another path, skip cancel" - ); - } - Err(e) => { - tracing::warn!( - error = %e, - task_id = %record.task_id, - "Failed to mark subagent as cancelled in DB" - ); - } + // 触发每个子代理的 CancellationToken。 + // 锁内只做内存操作（cancel + 收集），SQLite 调用移到锁外—— + // registry 是 parking_lot 同步锁，持锁跨 DB IO 会阻塞所有并发的 + // 注册/Guard::drop，SQLite 卡顿时放大 /stop 延迟。 + let mut missing_from_registry: Vec = Vec::new(); + { + let registry = self.cancel_registry.lock(); + for record in &running { + if let Some(token) = registry.get(&record.task_id) { + token.cancel(); + tracing::info!( + task_id = %record.task_id, + "Cancelled subagent token" + ); + } else { + missing_from_registry.push(record.task_id.clone()); + } + } + } + + for task_id in &missing_from_registry { + // token 不在 registry 中（可能已完成但 DB 状态未更新，或进程重启后丢失） + // 不变量 1：条件 UPDATE，仅在 status='running' 时转为 cancelled， + // 避免 spawn 已完成的终态被覆盖（completed → cancelled 是非法转换） + match self + .store + .try_update_pending_subagent_status(task_id, "running", "cancelled") + { + Ok(true) => { + tracing::info!( + task_id = %task_id, + "Marked subagent as cancelled in DB (token not in registry)" + ); + } + Ok(false) => { + tracing::info!( + task_id = %task_id, + "Subagent status already updated by another path, skip cancel" + ); + } + Err(e) => { + tracing::warn!( + error = %e, + task_id = %task_id, + "Failed to mark subagent as cancelled in DB" + ); } } } - drop(registry); count } @@ -1481,25 +1514,27 @@ impl SubAgentRuntime for DefaultSubAgentRuntime { } let mut reaped = 0; - let registry = self.cancel_registry.lock(); - for record in &running { - // 执行任务仍在（registry 有 token）→ 真正在跑，保留 - if registry.contains_key(&record.task_id) { - continue; - } + // 锁内只收集僵尸 task_id，SQLite 条件 UPDATE 移到锁外（理由同 + // cancel_pending_for_topic：避免持 parking_lot 锁跨 DB IO）。 + let zombie_ids: Vec = { + let registry = self.cancel_registry.lock(); + running + .iter() + .filter(|record| !registry.contains_key(&record.task_id)) + .map(|record| record.task_id.clone()) + .collect() + }; + for task_id in &zombie_ids { // 僵尸：DB=running 但执行任务已消失。 // 条件 UPDATE（仅 running→interrupted），不覆盖已终态。 - match self.store.try_update_pending_subagent_status( - &record.task_id, - "running", - "interrupted", - ) { + match self + .store + .try_update_pending_subagent_status(task_id, "running", "interrupted") + { Ok(true) => { reaped += 1; tracing::warn!( - task_id = %record.task_id, - def_name = ?record.def_name, - spawned_at = ?record.spawned_at, + task_id = %task_id, "Reaped zombie subagent (DB running but executor gone); marked interrupted" ); } @@ -1507,13 +1542,12 @@ impl SubAgentRuntime for DefaultSubAgentRuntime { Err(e) => { tracing::warn!( error = %e, - task_id = %record.task_id, + task_id = %task_id, "Failed to reap zombie subagent" ); } } } - drop(registry); if reaped > 0 { tracing::info!(