From 87ce0d27acd82c9592803f135d1eea5eedaec985 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Timo=20Pr=C3=BC=C3=9Fe?= Date: Sat, 5 Sep 2026 16:02:48 +0200 Subject: [PATCH] feat: deepen async concurrency seams MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move sync host work (config load, shell conditions, clone fs ops) off Tokio workers via host_blocking; add Runner cancellation and JoinSet for parallel layers. Gate::admit acquires exclusive lanes, per-Task sub-quotas for parallel tasks, then work permits. ChannelSink uses bounded mpsc with block_in_place backpressure — never blocking_send directly on a worker (that path panics). Intern command_desc as Arc alongside task_name. Update ADR-0003/0005 and CONTEXT to match. --- CONTEXT.md | 55 ++++--- docs/adr/0003-concurrency-gate-global.md | 15 +- docs/adr/0005-task-event-sink.md | 14 +- src/engine/commands/catalog.rs | 12 +- src/engine/commands/clone.rs | 6 +- src/engine/commands/copy.rs | 1 + src/engine/commands/setup.rs | 10 +- src/engine/commands/symlink.rs | 3 +- src/engine/concurrency.rs | 96 +++++++++++ src/engine/conditions.rs | 86 ++++++---- src/engine/context.rs | 3 + src/engine/event.rs | 8 +- src/engine/host_blocking.rs | 8 + src/engine/mod.rs | 1 + src/engine/runner.rs | 194 +++++++++++++++++------ src/engine/sink.rs | 85 +++++++++- src/main.rs | 4 +- src/tui/event_loop.rs | 4 +- src/tui/mod.rs | 2 +- src/tui/plain.rs | 2 +- 20 files changed, 478 insertions(+), 131 deletions(-) create mode 100644 src/engine/host_blocking.rs diff --git a/CONTEXT.md b/CONTEXT.md index d9ccc13..9ea8d69 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -70,7 +70,10 @@ _Avoid_: action, command, operation. **Runner**: The component that orders tasks, applies skip rules, and drives each task's -command entries to completion, emitting events as it goes (`TaskRunner`). +command entries to completion, emitting events as it goes (`TaskRunner`). Holds +a `CancellationToken` shared with nested Sub-config Runners; cancel aborts +in-flight Tokio tasks (OS subprocess teardown is a follow-up). Uses +`tokio::task::JoinSet` for parallel layer and parallel Command admission. _Avoid_: engine (too broad), executor (means something narrower here). **Command executor**: @@ -99,10 +102,10 @@ justifies a real plugin seam — see ADR-0006). **Task event**: A message describing execution progress (`TaskEvent`) — lifecycle and -per-line/per-file output alike. `task_name` is an `Arc` interned once per -Task so per-line output clones a refcount, not a new allocation. Emitted -through the **Task event sink**; the TUI and plain logger consume events from -the channel-backed adapter. +per-line/per-file output alike. `task_name` and `command_desc` are `Arc` +interned once per Task / Command entry so per-line output clones refcounts, +not new allocations. Emitted through the **Task event sink**; the TUI and +plain logger consume events from the channel-backed adapter. _Avoid_: message, log, signal. **History**: @@ -128,8 +131,8 @@ Declarative gate on whether a Task runs — `only_if` (all must pass) and `skip_if` (any triggers skip). Each field accepts a path string, a list of path strings (backward compatible), or rich objects: `{ path }`, `{ env }`, `{ command }`, `{ mode }`. Evaluated in `engine::conditions` together with OS -filter and install History skip; the Runner calls `evaluate_skip` before -spawning work. +filter and install History skip; the Runner calls async `evaluate_skip` before +spawning work. Shell `{ command }` conditions run via **Host blocking**. _Avoid_: when clause, if guard, predicate. **Task graph**: @@ -142,24 +145,30 @@ _Avoid_: dependency resolver, DAG, scheduler (do not reuse "scheduler" for concurrency — see **Concurrency gate**). **Task event sink**: -The seam for emitting **Task events**. Adapters: **ChannelSink** (mpsc → -TUI/plain) and **NullSink** (benches). The Runner and `CommandContext` both -emit through this interface — not a raw sender. Subprocess line readers may -coalesce stdout/stderr into a **`CommandOutputBatch`** Task event before -emit (single-line `CommandOutput` remains for sparse progress); batching is -an engine policy, not a third sink adapter. +The seam for emitting **Task events**. Adapters: **ChannelSink** (bounded +mpsc, capacity 8192, backpressure via `block_in_place` + `blocking_send` on +the multi-thread runtime → TUI/plain) and **NullSink** +(benches). The Runner and `CommandContext` both emit through this interface — +not a raw sender. Subprocess line readers may coalesce stdout/stderr into a +**`CommandOutputBatch`** Task event before emit (single-line `CommandOutput` +remains for sparse progress); batching is an engine policy, not a third sink +adapter. _Avoid_: logger, event bus, observer. **Concurrency gate**: The Runner's global cap on in-flight leaf Command executor work, driven by `num_threads` (default: physical CPUs − 1). Permits are per Command entry; `machine_setup` does not hold a permit so nested Sub-configs can share the -gate. Sync File ops work runs via `spawn_blocking` so it does not block Tokio -workers. Owns a shared Rayon FS apply pool (same size as the permit limit), -created lazily on first tree-apply use. Also owns **Exclusive lanes** (package -managers) and a separate **tree-apply** K=1 semaphore so only one -`pool.install` runs at a time (tree-apply does not reuse Exclusive lanes). -Does **not** order Tasks by dependency — that remains the **Task graph**. +gate. **`admit`** acquires an Exclusive lane (if any), an optional per-Task +sub-quota for `parallel: true` Tasks (half the gate limit, ceil, minimum 1), +then a work permit. Sync File ops work runs via `spawn_blocking` so it does +not block Tokio workers; sync config load, shell conditions, and clone fs ops +use **Host blocking** (`engine::host_blocking`). Owns a shared Rayon FS apply +pool (same size as the permit limit), created lazily on first tree-apply use. +Also owns **Exclusive lanes** (package managers) and a separate **tree-apply** +K=1 semaphore so only one `pool.install` runs at a time (tree-apply does not +reuse Exclusive lanes). Does **not** order Tasks by dependency — that remains +the **Task graph**. _Avoid_: scheduler (do not call the FS pool the "scheduler"). **Exclusive lane**: @@ -216,6 +225,14 @@ thin per-kind Command executors that supply a per-file strategy. Does not move privilege planning into File ops (ADR-0002). _Avoid_: unified tree command, generic file command. +**Host blocking**: +The seam for running synchronous host work off Tokio worker threads via +`tokio::task::spawn_blocking` — config load in Sub-config, shell condition +evaluation, clone directory create/remove, and similar blocking I/O. Tree-op +already uses its own `spawn_blocking` path; main-thread pre-runtime config load +in `main.rs` stays synchronous. +_Avoid_: blocking the async runtime, inline fs in async executors. + **Command bench**: The measurement module for Command executor / Tree materialization / Runner wall-clock speed — Criterion microbenches plus thin Runner smoke over diff --git a/docs/adr/0003-concurrency-gate-global.md b/docs/adr/0003-concurrency-gate-global.md index 7a443c7..d4e9c10 100644 --- a/docs/adr/0003-concurrency-gate-global.md +++ b/docs/adr/0003-concurrency-gate-global.md @@ -11,13 +11,15 @@ Task — so sequential and `parallel: true` Tasks share the same rule. The Sub-config Runners share the parent's gate and their leaf commands acquire normally. That avoids a deadlock when `num_threads: 1` and a parent Task would otherwise hold the only permit across nested work. Sync File ops (`copy` / -`symlink`) run via `spawn_blocking`. We accepted that a fat `parallel: true` -Task can starve siblings; per-Task sub-quotas are a future fix. The gate also +`symlink`) run via `spawn_blocking`. Per-Task sub-quotas are implemented for +`parallel: true` Tasks: each such Task gets a semaphore of +`max(1, gate_limit.div_ceil(2))` so parallel commands within one Task cannot +consume the entire gate. Sequential Tasks pass no sub-quota. The gate also owns a shared Rayon pool of the same size for in-tree DirectFs file apply (ADR-0004) so sibling commands do not each spawn a private worker set. The pool is created lazily on first tree-apply use, not when the gate is built. -## Tree-apply admission (accepted 2026-09-05 — not yet implemented) +## Tree-apply admission (accepted 2026-09-05 — implemented) A second semaphore on the Concurrency gate admits **at most one** concurrent tree `pool.install` (K=1). Leaf Command permits and Exclusive lanes are @@ -25,3 +27,10 @@ unchanged. This prevents sibling tree Command entries from oversubscribing the shared Rayon pool. Splitting pool width or reusing package-manager Exclusive lanes for trees was rejected. K>1 remains a possible later knob; default is exclusive tree-apply. + +## Per-Task sub-quotas (implemented 2026-09-05) + +When a Task has `parallel: true`, the Runner creates a per-Task semaphore of +`max(1, gate_limit.div_ceil(2))`. Each parallel Command entry acquires through +`ConcurrencyGate::admit` with that quota before the global work permit. +Sequential Tasks omit the sub-quota. diff --git a/docs/adr/0005-task-event-sink.md b/docs/adr/0005-task-event-sink.md index 59d3dbb..2637c17 100644 --- a/docs/adr/0005-task-event-sink.md +++ b/docs/adr/0005-task-event-sink.md @@ -7,11 +7,19 @@ wall-clock from event clone/channel cost and keeps one seam for lifecycle and CommandOutput alike. Fan-out to multiple simultaneous consumers was rejected as overbuilt for a single UI consumer. -## Coalesced subprocess output (accepted 2026-09-05 — not yet implemented) +## Bounded ChannelSink (implemented 2026-09-05) + +**ChannelSink** uses a bounded `mpsc` channel with capacity **8192**. +`emit` applies backpressure (`try_send`, then on `Full`: `block_in_place` + +`blocking_send` on the multi-thread runtime, plain `blocking_send` outside a +runtime). Direct `blocking_send` on a Tokio worker panics — do not restore +that path. Closed receivers are ignored. **NullSink** is unchanged. + +## Coalesced subprocess output (implemented 2026-09-05) Subprocess line reading may buffer stdout/stderr and emit a **`CommandOutputBatch`** Task event (multiple lines per flush) instead of one `CommandOutput` per line. Single-line `CommandOutput` stays for sparse progress (e.g. FileProgress). Coalescing lives in the engine line reader, not in a new -sink adapter. TUI/plain must expand batches. Bench the ChannelSink path (not -only NullSink) when landing. +sink adapter. TUI/plain expand batches. Bench the ChannelSink path (not only +NullSink) when landing. diff --git a/src/engine/commands/catalog.rs b/src/engine/commands/catalog.rs index 5529bb6..f63756d 100644 --- a/src/engine/commands/catalog.rs +++ b/src/engine/commands/catalog.rs @@ -48,13 +48,13 @@ pub fn create_executor(entry: &CommandEntry) -> Executor { } /// Display label for a Command entry (used by `Display` and Task events). -pub fn description(entry: &CommandEntry) -> String { +pub fn description(entry: &CommandEntry) -> std::sync::Arc { match entry { - CommandEntry::Copy(args) => args.to_string(), - CommandEntry::Symlink(args) => args.to_string(), - CommandEntry::Clone(args) => args.to_string(), - CommandEntry::Run(args) => args.to_string(), - CommandEntry::MachineSetup(args) => args.to_string(), + CommandEntry::Copy(args) => std::sync::Arc::from(args.to_string()), + CommandEntry::Symlink(args) => std::sync::Arc::from(args.to_string()), + CommandEntry::Clone(args) => std::sync::Arc::from(args.to_string()), + CommandEntry::Run(args) => std::sync::Arc::from(args.to_string()), + CommandEntry::MachineSetup(args) => std::sync::Arc::from(args.to_string()), } } diff --git a/src/engine/commands/clone.rs b/src/engine/commands/clone.rs index 2f087b4..eae8d72 100644 --- a/src/engine/commands/clone.rs +++ b/src/engine/commands/clone.rs @@ -69,7 +69,8 @@ impl CloneCommand { )); if let Some(parent) = target.parent() { - std::fs::create_dir_all(parent)?; + let parent = parent.to_path_buf(); + crate::engine::host_blocking::run(move || std::fs::create_dir_all(&parent)).await??; } run_git_command( @@ -95,7 +96,8 @@ impl CloneCommand { if target.exists() { ctx.log_progress(format!("remove {}", display_path(&target))); - std::fs::remove_dir_all(&target)?; + let target = target.to_path_buf(); + crate::engine::host_blocking::run(move || std::fs::remove_dir_all(&target)).await??; } Ok(()) diff --git a/src/engine/commands/copy.rs b/src/engine/commands/copy.rs index f249e3a..88580f0 100644 --- a/src/engine/commands/copy.rs +++ b/src/engine/commands/copy.rs @@ -206,6 +206,7 @@ mod tests { default_shell: crate::config::types::Shell::Bash, task_name: Arc::::from("t"), depth: 0, + cancel: tokio_util::sync::CancellationToken::new(), } } diff --git a/src/engine/commands/setup.rs b/src/engine/commands/setup.rs index 971461d..32af546 100644 --- a/src/engine/commands/setup.rs +++ b/src/engine/commands/setup.rs @@ -49,7 +49,10 @@ async fn run_sub_config(args: &MachineSetupArgs, ctx: &CommandContext) -> Result ctx.log_info(format!("Loading sub-config: {config_str}")); - let config = crate::config::load_config(&config_str)?; + let config_str = config_str.into_owned(); + let sub_config_dir = crate::config::resolve_config_dir(&config_str, &ctx.config_dir); + let config = crate::engine::host_blocking::run(move || crate::config::load_config(&config_str)) + .await??; let run_set = match (&args.task, args.with_deps) { (Some(task_name), true) => Some(crate::config::selection::expand_for_mode( @@ -63,12 +66,11 @@ async fn run_sub_config(args: &MachineSetupArgs, ctx: &CommandContext) -> Result // Resolve the sub-config's directory for its own relative paths. URLs // and unresolvable paths fall back to the parent's config_dir. - let sub_config_dir = crate::config::resolve_config_dir(&config_str, &ctx.config_dir); - let runner = crate::engine::runner::TaskRunner::new(config, ctx.mode, Arc::clone(&ctx.events)) .with_gate(Arc::clone(&ctx.gate)) .with_config_dir(sub_config_dir) - .with_depth(ctx.depth + 1); + .with_depth(ctx.depth + 1) + .with_cancel(ctx.cancel.clone()); match (run_set, &args.task) { (Some(tasks), _) => runner.run_tasks(&tasks, args.force).await, diff --git a/src/engine/commands/symlink.rs b/src/engine/commands/symlink.rs index ead5ab5..5aa8ac0 100644 --- a/src/engine/commands/symlink.rs +++ b/src/engine/commands/symlink.rs @@ -164,7 +164,7 @@ mod tests { dir: &Path, ) -> ( CommandContext, - tokio::sync::mpsc::UnboundedReceiver, + tokio::sync::mpsc::Receiver, ) { let (events, rx) = crate::engine::sink::ChannelSink::channel(); let ctx = CommandContext { @@ -178,6 +178,7 @@ mod tests { default_shell: crate::config::types::Shell::Bash, task_name: std::sync::Arc::::from("t"), depth: 0, + cancel: tokio_util::sync::CancellationToken::new(), }; (ctx, rx) } diff --git a/src/engine/concurrency.rs b/src/engine/concurrency.rs index 7926ac2..0be50ad 100644 --- a/src/engine/concurrency.rs +++ b/src/engine/concurrency.rs @@ -107,6 +107,62 @@ impl ConcurrencyGate { .expect("ConcurrencyGate semaphore is never closed") } + /// Per-Task sub-quota for `parallel: true` Tasks — half the gate limit (ceil), at least 1. + pub fn task_quota_limit(&self) -> usize { + std::cmp::max(1, self.limit.div_ceil(2)) + } + + /// Lane first, then optional per-Task quota, then work permit (ADR-0010 + sub-quotas). + /// + /// Calls `on_lane_wait(lane)` only when the lane is already held (contended path). + pub async fn admit( + self: &Arc, + lane: Option, + occupies_slot: bool, + task_quota: Option<&Arc>, + on_lane_wait: impl FnOnce(ExclusiveLane), + ) -> AdmitPermits { + let lane_permit = if let Some(lane) = lane { + match self.try_acquire_lane(lane) { + Some(permit) => Some(permit), + None => { + on_lane_wait(lane); + Some(self.acquire_lane(lane).await) + } + } + } else { + None + }; + + let task_permit = if let Some(quota) = task_quota { + #[expect( + clippy::expect_used, + reason = "per-Task quota semaphore is never closed" + )] + Some( + quota + .clone() + .acquire_owned() + .await + .expect("per-Task quota semaphore is never closed"), + ) + } else { + None + }; + + let work_permit = if occupies_slot { + Some(self.acquire().await) + } else { + None + }; + + AdmitPermits { + _lane: lane_permit, + _task: task_permit, + _work: work_permit, + } + } + /// Try to take an Exclusive lane without waiting. pub fn try_acquire_lane(&self, lane: ExclusiveLane) -> Option { self.lanes[lane.index()].clone().try_acquire_owned().ok() @@ -139,6 +195,13 @@ impl ConcurrencyGate { } } +/// Permits held for one Command entry admission (lane, optional Task quota, work). +pub struct AdmitPermits { + _lane: Option, + _task: Option, + _work: Option, +} + fn build_fs_pool(limit: usize) -> ThreadPool { #[expect( clippy::expect_used, @@ -168,6 +231,39 @@ pub fn resolve_limit(num_threads: Option) -> usize { mod tests { use super::*; + #[test] + fn task_quota_limit_is_half_ceil_min_one() { + let gate = ConcurrencyGate::from_num_threads(Some(2)); + assert_eq!(gate.task_quota_limit(), 1); + let gate = ConcurrencyGate::from_num_threads(Some(5)); + assert_eq!(gate.task_quota_limit(), 3); + let gate = ConcurrencyGate::from_num_threads(Some(1)); + assert_eq!(gate.task_quota_limit(), 1); + } + + #[tokio::test] + async fn task_quota_serializes_parallel_commands_when_one() { + let gate = Arc::new(ConcurrencyGate::from_num_threads(Some(2))); + assert_eq!(gate.task_quota_limit(), 1); + let quota = Arc::new(Semaphore::new(gate.task_quota_limit())); + let gate2 = Arc::clone(&gate); + let quota2 = Arc::clone("a); + let first = tokio::spawn(async move { + let _p = gate2.admit(None, true, Some("a2), |_| {}).await; + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + }); + tokio::task::yield_now().await; + let gate3 = Arc::clone(&gate); + let quota3 = Arc::clone("a); + let second = tokio::spawn(async move { + gate3.admit(None, true, Some("a3), |_| {}).await; + }); + tokio::task::yield_now().await; + assert!(!second.is_finished()); + first.await.unwrap(); + second.await.unwrap(); + } + #[test] fn resolve_limit_respects_explicit() { assert_eq!(resolve_limit(Some(4)), 4); diff --git a/src/engine/conditions.rs b/src/engine/conditions.rs index bcbd3ef..798ffbf 100644 --- a/src/engine/conditions.rs +++ b/src/engine/conditions.rs @@ -2,13 +2,14 @@ use std::path::Path; use crate::config::history::History; use crate::config::types::{Condition, Shell, TaskConfig}; +use crate::engine::host_blocking; use crate::engine::mode::Mode; use crate::utils::path::expand_path; use crate::utils::shell::shell_binary; /// Decide whether a task should be skipped. Returns `Some(reason)` when the /// task must not run (OS filter, conditions, or install History). -pub fn evaluate_skip( +pub async fn evaluate_skip( task: &TaskConfig, name: &str, mode: Mode, @@ -22,13 +23,13 @@ pub fn evaluate_skip( } for cond in task.only_if.iter() { - if let Some(reason) = evaluate_only_if(cond, mode, config_dir, default_shell) { + if let Some(reason) = evaluate_only_if(cond, mode, config_dir, default_shell).await { return Some(reason); } } for cond in task.skip_if.iter() { - if let Some(reason) = evaluate_skip_if(cond, mode, config_dir, default_shell) { + if let Some(reason) = evaluate_skip_if(cond, mode, config_dir, default_shell).await { return Some(reason); } } @@ -40,7 +41,7 @@ pub fn evaluate_skip( None } -fn evaluate_only_if( +async fn evaluate_only_if( cond: &Condition, mode: Mode, config_dir: &Path, @@ -60,10 +61,16 @@ fn evaluate_only_if( _ => Some(format!("Condition not met: env '{var}' is unset or empty")), }, Condition::Command(command) => { - if command_succeeds(command, default_shell) { + let display = command.clone(); + let shell = default_shell.clone(); + let cmd = display.clone(); + let ok = host_blocking::run(move || command_succeeds(&cmd, &shell)) + .await + .unwrap_or(false); + if ok { None } else { - Some(format!("Condition not met: command failed: '{command}'")) + Some(format!("Condition not met: command failed: '{display}'")) } } Condition::Mode(modes) => { @@ -83,7 +90,7 @@ fn evaluate_only_if( } } -fn evaluate_skip_if( +async fn evaluate_skip_if( cond: &Condition, mode: Mode, config_dir: &Path, @@ -103,8 +110,14 @@ fn evaluate_skip_if( _ => None, }, Condition::Command(command) => { - if command_succeeds(command, default_shell) { - Some(format!("Skipped: command succeeded: '{command}'")) + let display = command.clone(); + let shell = default_shell.clone(); + let cmd = display.clone(); + let ok = host_blocking::run(move || command_succeeds(&cmd, &shell)) + .await + .unwrap_or(false); + if ok { + Some(format!("Skipped: command succeeded: '{display}'")) } else { None } @@ -167,8 +180,8 @@ mod tests { } } - #[test] - fn only_if_path_missing_skips() { + #[tokio::test] + async fn only_if_path_missing_skips() { let dir = tempdir().unwrap(); let task = task_with( vec![Condition::Path("/nonexistent/path".into())].into(), @@ -183,11 +196,12 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_some()); } - #[test] - fn only_if_path_exists_runs() { + #[tokio::test] + async fn only_if_path_exists_runs() { let dir = tempdir().unwrap(); let marker = dir.path().join("marker"); std::fs::write(&marker, "").unwrap(); @@ -204,11 +218,12 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_none()); } - #[test] - fn skip_if_path_exists_skips() { + #[tokio::test] + async fn skip_if_path_exists_skips() { let dir = tempdir().unwrap(); let marker = dir.path().join("marker"); std::fs::write(&marker, "").unwrap(); @@ -225,11 +240,12 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_some()); } - #[test] - fn only_if_env_set_runs() { + #[tokio::test] + async fn only_if_env_set_runs() { let dir = tempdir().unwrap(); env::set_var("MACHINE_SETUP_TEST_COND_VAR", "yes"); let task = task_with( @@ -245,12 +261,13 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_none()); env::remove_var("MACHINE_SETUP_TEST_COND_VAR"); } - #[test] - fn only_if_env_unset_skips() { + #[tokio::test] + async fn only_if_env_unset_skips() { let dir = tempdir().unwrap(); env::remove_var("MACHINE_SETUP_TEST_COND_UNSET"); let task = task_with( @@ -266,11 +283,12 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_some()); } - #[test] - fn only_if_mode_matches_runs() { + #[tokio::test] + async fn only_if_mode_matches_runs() { let dir = tempdir().unwrap(); let task = task_with( vec![Condition::Mode(vec![Mode::Install, Mode::Update])].into(), @@ -285,11 +303,12 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_none()); } - #[test] - fn only_if_mode_mismatch_skips() { + #[tokio::test] + async fn only_if_mode_mismatch_skips() { let dir = tempdir().unwrap(); let task = task_with( vec![Condition::Mode(vec![Mode::Update])].into(), @@ -304,11 +323,12 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_some()); } - #[test] - fn skip_if_mode_matches_skips() { + #[tokio::test] + async fn skip_if_mode_matches_skips() { let dir = tempdir().unwrap(); let task = task_with( Conditions::default(), @@ -323,11 +343,12 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_some()); } - #[test] - fn only_if_command_true_runs() { + #[tokio::test] + async fn only_if_command_true_runs() { let dir = tempdir().unwrap(); let task = task_with( vec![Condition::Command("true".into())].into(), @@ -342,11 +363,12 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_none()); } - #[test] - fn only_if_command_false_skips() { + #[tokio::test] + async fn only_if_command_false_skips() { let dir = tempdir().unwrap(); let task = task_with( vec![Condition::Command("false".into())].into(), @@ -361,11 +383,12 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_some()); } - #[test] - fn skip_if_command_true_skips() { + #[tokio::test] + async fn skip_if_command_true_skips() { let dir = tempdir().unwrap(); let task = task_with( Conditions::default(), @@ -380,6 +403,7 @@ mod tests { dir.path(), &Shell::Bash, ) + .await .is_some()); } } diff --git a/src/engine/context.rs b/src/engine/context.rs index 3a0ed8f..d14f46c 100644 --- a/src/engine/context.rs +++ b/src/engine/context.rs @@ -35,6 +35,9 @@ pub struct CommandContext { /// Nesting depth (0 = top-level, 1 = sub-config, etc.) pub depth: usize, + + /// Shared cancellation token (parent and nested Sub-config runs). + pub cancel: tokio_util::sync::CancellationToken, } impl CommandContext { diff --git a/src/engine/event.rs b/src/engine/event.rs index 872f891..4904ace 100644 --- a/src/engine/event.rs +++ b/src/engine/event.rs @@ -37,7 +37,7 @@ pub enum TaskEvent { /// A command within a task started. CommandStarted { task_name: Arc, - command_desc: String, + command_desc: Arc, /// 1-based index within the task's command list. command_index: usize, command_total: usize, @@ -46,7 +46,7 @@ pub enum TaskEvent { /// A command is waiting on an Exclusive lane already held in this run. CommandWaiting { task_name: Arc, - command_desc: String, + command_desc: Arc, command_index: usize, command_total: usize, lane: ExclusiveLane, @@ -55,7 +55,7 @@ pub enum TaskEvent { /// A command within a task completed successfully. CommandCompleted { task_name: Arc, - command_desc: String, + command_desc: Arc, command_index: usize, command_total: usize, }, @@ -63,7 +63,7 @@ pub enum TaskEvent { /// A command within a task failed. CommandFailed { task_name: Arc, - command_desc: String, + command_desc: Arc, command_index: usize, command_total: usize, error: String, diff --git a/src/engine/host_blocking.rs b/src/engine/host_blocking.rs new file mode 100644 index 0000000..02d6f45 --- /dev/null +++ b/src/engine/host_blocking.rs @@ -0,0 +1,8 @@ +use crate::error::{Error, Result}; + +/// Run sync host work off Tokio worker threads. +pub async fn run(f: impl FnOnce() -> T + Send + 'static) -> Result { + tokio::task::spawn_blocking(f) + .await + .map_err(|e| Error::TaskJoin(e.to_string())) +} diff --git a/src/engine/mod.rs b/src/engine/mod.rs index f87d527..ba3a9cd 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -3,6 +3,7 @@ pub mod concurrency; pub mod conditions; pub mod context; pub mod event; +pub mod host_blocking; pub mod mode; pub mod output; pub mod runner; diff --git a/src/engine/runner.rs b/src/engine/runner.rs index b968bba..db09d6a 100644 --- a/src/engine/runner.rs +++ b/src/engine/runner.rs @@ -2,6 +2,10 @@ use std::collections::HashMap; use std::path::PathBuf; use std::sync::{Arc, Mutex}; +use tokio::sync::Semaphore; +use tokio::task::JoinSet; +use tokio_util::sync::CancellationToken; + use crate::config::graph::TaskGraph; use crate::config::history::History; use crate::config::types::{AppConfig, TaskConfig}; @@ -23,6 +27,7 @@ pub struct TaskRunner { gate: Arc, config_dir: Arc, depth: usize, + cancel: CancellationToken, /// Lazily built executors per task name — one `Arc` per Command entry. executor_cache: Mutex]>>>, } @@ -45,6 +50,7 @@ impl TaskRunner { gate, config_dir: Arc::new(std::env::current_dir().unwrap_or_default()), depth: 0, + cancel: CancellationToken::new(), executor_cache: Mutex::new(HashMap::new()), } } @@ -65,6 +71,11 @@ impl TaskRunner { self } + pub fn with_cancel(mut self, cancel: CancellationToken) -> Self { + self.cancel = cancel; + self + } + /// Run all tasks (respecting parallel config). pub async fn run_all(&self, force: bool) -> Result<()> { let task_names: Vec = self.config.tasks.keys().cloned().collect(); @@ -81,6 +92,10 @@ impl TaskRunner { /// Run specific tasks by name. pub async fn run_tasks(&self, task_names: &[String], force: bool) -> Result<()> { + if self.cancel.is_cancelled() { + return Err(Error::Aborted); + } + // Resolve dependency order via the task graph. When no task has // dependencies, this borrows `task_names` instead of cloning it. let graph = TaskGraph::new(&self.config.tasks); @@ -107,6 +122,9 @@ impl TaskRunner { let mut tally = Tally::default(); for layer in &layers { + if self.cancel.is_cancelled() { + break; + } self.run_layer( layer, force, @@ -115,6 +133,9 @@ impl TaskRunner { &mut tally, ) .await; + if self.cancel.is_cancelled() { + break; + } } // Save history @@ -128,6 +149,10 @@ impl TaskRunner { skipped: tally.skipped, }); + if self.cancel.is_cancelled() { + return Err(Error::Aborted); + } + if tally.failed > 0 { Err(Error::TasksFailed(tally.failed)) } else { @@ -151,9 +176,13 @@ impl TaskRunner { history: &mut History, tally: &mut Tally, ) { - let mut handles = Vec::new(); + let mut join_set = JoinSet::new(); for name in layer { + if self.cancel.is_cancelled() { + break; + } + let task_config = Arc::clone(&self.config.tasks[name]); if let Some(reason) = evaluate_skip( @@ -164,7 +193,9 @@ impl TaskRunner { history, self.config_dir.as_path(), &self.config.default_shell, - ) { + ) + .await + { self.send(TaskEvent::TaskSkipped { task_name: Arc::::from(name.as_str()), reason, @@ -174,24 +205,38 @@ impl TaskRunner { } let ctx = self.create_context(name, Arc::clone(&temp_dir)); - let name = Arc::clone(&ctx.task_name); - let executors = self.executors_for_task(name.as_ref(), task_config.as_ref()); - handles.push(tokio::spawn(async move { - let result = run_task_with_retry(&task_config, &ctx, executors).await; - (name, result) - })); + let name_arc = Arc::clone(&ctx.task_name); + let executors = self.executors_for_task(name, task_config.as_ref()); + let cancel = self.cancel.clone(); + join_set.spawn(async move { + tokio::select! { + result = run_task_with_retry(&task_config, &ctx, executors) => { + (name_arc, result) + } + _ = cancel.cancelled() => { + (name_arc, Err(Error::Aborted)) + } + } + }); } - for handle in handles { - match handle.await { + while let Some(result) = join_set.join_next().await { + if self.cancel.is_cancelled() { + join_set.abort_all(); + while join_set.join_next().await.is_some() {} + break; + } + + match result { Ok((name, Ok(()))) => { - self.update_history(history, &name); + self.update_history(history, name.as_ref()); tally.succeeded += 1; } Ok((name, Err(e))) => { + let error = e.to_string(); self.send(TaskEvent::TaskFailed { task_name: name, - error: e.to_string(), + error, }); tally.failed += 1; } @@ -224,6 +269,7 @@ impl TaskRunner { default_shell: self.config.default_shell.clone(), task_name: Arc::::from(task_name), depth: self.depth, + cancel: self.cancel.clone(), } } @@ -284,6 +330,10 @@ async fn run_task_with_retry( let max_attempts = task.retry + 1; for attempt in 1..=max_attempts { + if ctx.cancel.is_cancelled() { + return Err(Error::Aborted); + } + match run_task(task, ctx, Arc::clone(&executors)).await { Ok(()) => return Ok(()), Err(e) if attempt < max_attempts => { @@ -293,7 +343,10 @@ async fn run_task_with_retry( max_attempts, error: e.to_string(), }); - tokio::time::sleep(std::time::Duration::from_secs(task.retry_delay_secs)).await; + tokio::select! { + _ = tokio::time::sleep(std::time::Duration::from_secs(task.retry_delay_secs)) => {} + _ = ctx.cancel.cancelled() => return Err(Error::Aborted), + } } Err(e) => return Err(e), } @@ -307,6 +360,10 @@ async fn run_task( ctx: &CommandContext, executors: Arc<[Arc]>, ) -> Result<()> { + if ctx.cancel.is_cancelled() { + return Err(Error::Aborted); + } + ctx.emit(TaskEvent::TaskStarted { task_name: Arc::clone(&ctx.task_name), command_count: task.commands.len(), @@ -314,34 +371,55 @@ async fn run_task( }); let command_total = task.commands.len(); + let task_quota = task + .parallel + .then(|| Arc::new(Semaphore::new(ctx.gate.task_quota_limit()))); if task.parallel { - let mut handles = Vec::new(); + let mut join_set = JoinSet::new(); for (i, entry) in task.commands.iter().enumerate() { + if ctx.cancel.is_cancelled() { + break; + } + let command_index = i + 1; let executor = Arc::clone(&executors[i]); let desc = catalog::description(entry); let lane = catalog::exclusive_lane(entry, ctx.mode); ctx.emit(TaskEvent::CommandStarted { task_name: Arc::clone(&ctx.task_name), - command_desc: desc.clone(), + command_desc: Arc::clone(&desc), command_index, command_total, }); let ctx = ctx.clone(); - handles.push(tokio::spawn(async move { - let result = - execute_with_gate(lane, &executor, &ctx, &desc, command_index, command_total) - .await; + let quota = task_quota.as_ref().map(Arc::clone); + join_set.spawn(async move { + let result = execute_with_gate( + lane, + &executor, + &ctx, + desc.clone(), + command_index, + command_total, + quota.as_ref(), + ) + .await; (desc, command_index, result) - })); + }); } - for handle in handles { - let (desc, command_index, result) = - handle.await.map_err(|e| Error::TaskJoin(e.to_string()))?; - match result { + while let Some(result) = join_set.join_next().await { + if ctx.cancel.is_cancelled() { + join_set.abort_all(); + while join_set.join_next().await.is_some() {} + return Err(Error::Aborted); + } + + let (desc, command_index, cmd_result) = + result.map_err(|e| Error::TaskJoin(e.to_string()))?; + match cmd_result { Ok(()) => { ctx.emit(TaskEvent::CommandCompleted { task_name: Arc::clone(&ctx.task_name), @@ -358,24 +436,39 @@ async fn run_task( command_total, error: e.to_string(), }); + join_set.abort_all(); + while join_set.join_next().await.is_some() {} return Err(e); } } } } else { for (i, entry) in task.commands.iter().enumerate() { + if ctx.cancel.is_cancelled() { + return Err(Error::Aborted); + } + let command_index = i + 1; let executor = &executors[i]; let desc = catalog::description(entry); let lane = catalog::exclusive_lane(entry, ctx.mode); ctx.emit(TaskEvent::CommandStarted { task_name: Arc::clone(&ctx.task_name), - command_desc: desc.clone(), + command_desc: Arc::clone(&desc), command_index, command_total, }); - match execute_with_gate(lane, executor, ctx, &desc, command_index, command_total).await + match execute_with_gate( + lane, + executor, + ctx, + desc.clone(), + command_index, + command_total, + None, + ) + .await { Ok(()) => { ctx.emit(TaskEvent::CommandCompleted { @@ -406,39 +499,46 @@ async fn run_task( Ok(()) } -/// Admit a Command entry: Exclusive lane (if any) first, then work permit. +/// Admit a Command entry via the Concurrency gate, then execute. async fn execute_with_gate( lane: Option, executor: &Executor, ctx: &CommandContext, - command_desc: &str, + command_desc: Arc, command_index: usize, command_total: usize, + task_quota: Option<&Arc>, ) -> Result<()> { - let _lane_permit = if let Some(lane) = lane { - match ctx.gate.try_acquire_lane(lane) { - Some(permit) => Some(permit), - None => { - ctx.emit(TaskEvent::CommandWaiting { - task_name: ctx.task_name.clone(), - command_desc: command_desc.to_string(), + if ctx.cancel.is_cancelled() { + return Err(Error::Aborted); + } + + let occupies = executor.occupies_concurrency_slot(); + let events = Arc::clone(&ctx.events); + let task_name = Arc::clone(&ctx.task_name); + let desc_for_wait = Arc::clone(&command_desc); + + let _permits = ctx + .gate + .admit(lane, occupies, task_quota, |lane| { + TaskEventSink::emit( + events.as_ref(), + TaskEvent::CommandWaiting { + task_name: Arc::clone(&task_name), + command_desc: desc_for_wait, command_index, command_total, lane, - }); - Some(ctx.gate.acquire_lane(lane).await) - } - } - } else { - None - }; + }, + ); + }) + .await; - if executor.occupies_concurrency_slot() { - let _permit = ctx.gate.acquire().await; - executor.execute(ctx).await - } else { - executor.execute(ctx).await + if ctx.cancel.is_cancelled() { + return Err(Error::Aborted); } + + executor.execute(ctx).await } #[cfg(test)] diff --git a/src/engine/sink.rs b/src/engine/sink.rs index c732f2e..0b0da5b 100644 --- a/src/engine/sink.rs +++ b/src/engine/sink.rs @@ -4,10 +4,14 @@ use std::sync::Arc; +use tokio::runtime::RuntimeFlavor; use tokio::sync::mpsc; use super::event::TaskEvent; +/// Bounded channel capacity for production event delivery (ADR-0005). +pub const CHANNEL_CAPACITY: usize = 8192; + /// Emit Task events without callers knowing about channels or UI. pub trait TaskEventSink: Send + Sync { fn emit(&self, event: TaskEvent); @@ -16,26 +20,64 @@ pub trait TaskEventSink: Send + Sync { /// Shared handle used by the Runner and CommandContext. pub type SharedSink = Arc; -/// Forwards events onto an unbounded mpsc channel (TUI / plain logger). +/// Forwards events onto a bounded mpsc channel (TUI / plain logger). pub struct ChannelSink { - tx: mpsc::UnboundedSender, + tx: mpsc::Sender, } impl ChannelSink { - pub fn from_sender(tx: mpsc::UnboundedSender) -> SharedSink { + pub fn from_sender(tx: mpsc::Sender) -> SharedSink { Arc::new(Self { tx }) } /// Create a channel pair wrapped as a sink + receiver. - pub fn channel() -> (SharedSink, mpsc::UnboundedReceiver) { - let (tx, rx) = mpsc::unbounded_channel(); + pub fn channel() -> (SharedSink, mpsc::Receiver) { + Self::channel_with_capacity(CHANNEL_CAPACITY) + } + + /// Like [`Self::channel`] with an explicit capacity (tests / benches). + pub fn channel_with_capacity(capacity: usize) -> (SharedSink, mpsc::Receiver) { + let (tx, rx) = mpsc::channel(capacity.max(1)); (Self::from_sender(tx), rx) } } impl TaskEventSink for ChannelSink { fn emit(&self, event: TaskEvent) { - let _ = self.tx.send(event); + if self.tx.is_closed() { + return; + } + match self.tx.try_send(event) { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(event)) => { + send_when_full(&self.tx, event); + } + Err(mpsc::error::TrySendError::Closed(_)) => {} + } + } +} + +/// Apply backpressure without panicking on a Tokio worker thread. +/// +/// `Sender::blocking_send` panics inside an async runtime. On the multi-thread +/// runtime (production), park the worker via `block_in_place`. Outside a +/// runtime (e.g. sync tests, `spawn_blocking`), call `blocking_send` directly. +/// On the current-thread runtime, blocking is impossible — drop the event. +fn send_when_full(tx: &mpsc::Sender, event: TaskEvent) { + match tokio::runtime::Handle::try_current() { + Ok(handle) if handle.runtime_flavor() == RuntimeFlavor::CurrentThread => { + // Cannot block; prefer losing a progress event over panicking. + let _ = event; + } + Ok(_) => { + let tx = tx.clone(); + tokio::task::block_in_place(move || { + let _ = tx.blocking_send(event); + }); + } + Err(_) => { + let _ = tx.blocking_send(event); + } } } @@ -77,4 +119,35 @@ mod tests { skipped: 0, }); } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn channel_sink_backpressure_on_worker_does_not_panic() { + let (sink, mut rx) = ChannelSink::channel_with_capacity(1); + + let drained = tokio::spawn(async move { + let first = rx.recv().await.expect("first event"); + let second = rx.recv().await.expect("second event"); + (first, second) + }); + + sink.emit(TaskEvent::TaskCompleted { + task_name: "a".into(), + }); + // Second emit hits Full and must block_in_place rather than panic. + sink.emit(TaskEvent::TaskCompleted { + task_name: "b".into(), + }); + + let (first, second) = drained.await.expect("join drain"); + match (first, second) { + ( + TaskEvent::TaskCompleted { task_name: a }, + TaskEvent::TaskCompleted { task_name: b }, + ) => { + assert_eq!(a.as_ref(), "a"); + assert_eq!(b.as_ref(), "b"); + } + other => panic!("unexpected events: {other:?}"), + } + } } diff --git a/src/main.rs b/src/main.rs index 99d408c..702bf96 100644 --- a/src/main.rs +++ b/src/main.rs @@ -323,7 +323,9 @@ async fn run_execution( let mode = Mode::from_command(&cli.command) .expect("non-execution verbs are handled before this point"); - let runner = TaskRunner::new(app_config, mode, events).with_config_dir(config_dir); + let runner = TaskRunner::new(app_config, mode, events) + .with_config_dir(config_dir) + .with_cancel(cancel.clone()); let force = cli.force; let task_names_clone = task_names.clone(); diff --git a/src/tui/event_loop.rs b/src/tui/event_loop.rs index f8c923d..5a7e39e 100644 --- a/src/tui/event_loop.rs +++ b/src/tui/event_loop.rs @@ -19,7 +19,7 @@ use super::state::UiState; pub async fn run_loop( terminal: &mut Terminal>, mut state: UiState, - mut event_rx: mpsc::UnboundedReceiver, + mut event_rx: mpsc::Receiver, cancel: CancellationToken, ) -> anyhow::Result { let (key_tx, mut key_rx) = mpsc::unbounded_channel::(); @@ -110,7 +110,7 @@ pub async fn run_loop( } fn apply_engine_batch( - event_rx: &mut mpsc::UnboundedReceiver, + event_rx: &mut mpsc::Receiver, mut state: UiState, first: TaskEvent, ) -> UiState { diff --git a/src/tui/mod.rs b/src/tui/mod.rs index fce009a..05d26ba 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -32,7 +32,7 @@ pub(crate) fn restore_terminal() { /// Run the TUI, consuming events from the engine until all tasks are done. pub async fn run( - event_rx: mpsc::UnboundedReceiver, + event_rx: mpsc::Receiver, task_names: Vec, mode: crate::engine::mode::Mode, cancel: CancellationToken, diff --git a/src/tui/plain.rs b/src/tui/plain.rs index 1918235..38c380b 100644 --- a/src/tui/plain.rs +++ b/src/tui/plain.rs @@ -4,7 +4,7 @@ use crate::engine::event::TaskEvent; use crate::tui::log_display; /// Plain text event consumer for --no-tui / CI environments. -pub async fn run(mut event_rx: mpsc::UnboundedReceiver) { +pub async fn run(mut event_rx: mpsc::Receiver) { while let Some(event) = event_rx.recv().await { match &event { TaskEvent::TaskStarted {