Merge feature/ticket31-detectprofiles-parallel-probes into develop (#31)
Sondes de DetectProfiles::execute parallélisées (une tâche Tokio par candidat, JoinHandle attendus dans l'ordre) : coût total ~1×timeout au lieu de N×800ms, ordre de sortie déterministe préservé. QA vert 20/20. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -74,9 +74,18 @@ impl DetectProfiles {
|
|||||||
input: DetectProfilesInput,
|
input: DetectProfilesInput,
|
||||||
) -> Result<DetectProfilesOutput, AppError> {
|
) -> Result<DetectProfilesOutput, AppError> {
|
||||||
let mut results = Vec::with_capacity(input.candidates.len());
|
let mut results = Vec::with_capacity(input.candidates.len());
|
||||||
|
let mut probes = Vec::with_capacity(input.candidates.len());
|
||||||
for profile in input.candidates {
|
for profile in input.candidates {
|
||||||
let available = self.runtime.detect(&profile).await.unwrap_or(false);
|
let runtime = Arc::clone(&self.runtime);
|
||||||
results.push(ProfileAvailability { profile, available });
|
probes.push(tokio::spawn(async move {
|
||||||
|
let available = runtime.detect(&profile).await.unwrap_or(false);
|
||||||
|
ProfileAvailability { profile, available }
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
for probe in probes {
|
||||||
|
results.push(probe.await.map_err(|err| {
|
||||||
|
AppError::Process(format!("profile detection task failed: {err}"))
|
||||||
|
})?);
|
||||||
}
|
}
|
||||||
Ok(DetectProfilesOutput { results })
|
Ok(DetectProfilesOutput { results })
|
||||||
}
|
}
|
||||||
|
|||||||
@ -7,7 +7,9 @@
|
|||||||
//! command → result (including an error case to prove graceful degradation).
|
//! command → result (including an error case to prove graceful degradation).
|
||||||
|
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
|
|
||||||
@ -90,6 +92,13 @@ struct StubRuntime {
|
|||||||
|
|
||||||
struct YieldingRuntime;
|
struct YieldingRuntime;
|
||||||
|
|
||||||
|
struct SlowRuntime {
|
||||||
|
delay: Duration,
|
||||||
|
started: Arc<AtomicUsize>,
|
||||||
|
active: Arc<AtomicUsize>,
|
||||||
|
max_active: Arc<AtomicUsize>,
|
||||||
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
impl AgentRuntime for StubRuntime {
|
impl AgentRuntime for StubRuntime {
|
||||||
async fn detect(&self, profile: &AgentProfile) -> Result<bool, RuntimeError> {
|
async fn detect(&self, profile: &AgentProfile) -> Result<bool, RuntimeError> {
|
||||||
@ -129,6 +138,28 @@ impl AgentRuntime for YieldingRuntime {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl AgentRuntime for SlowRuntime {
|
||||||
|
async fn detect(&self, _profile: &AgentProfile) -> Result<bool, RuntimeError> {
|
||||||
|
self.started.fetch_add(1, Ordering::SeqCst);
|
||||||
|
let active = self.active.fetch_add(1, Ordering::SeqCst) + 1;
|
||||||
|
self.max_active.fetch_max(active, Ordering::SeqCst);
|
||||||
|
tokio::time::sleep(self.delay).await;
|
||||||
|
self.active.fetch_sub(1, Ordering::SeqCst);
|
||||||
|
Ok(false)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn prepare_invocation(
|
||||||
|
&self,
|
||||||
|
_profile: &AgentProfile,
|
||||||
|
_ctx: &PreparedContext,
|
||||||
|
_cwd: &ProjectPath,
|
||||||
|
_session: &SessionPlan,
|
||||||
|
) -> Result<SpawnSpec, RuntimeError> {
|
||||||
|
unreachable!("not used in these tests")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
struct SeqIds(Mutex<Vec<uuid::Uuid>>);
|
struct SeqIds(Mutex<Vec<uuid::Uuid>>);
|
||||||
|
|
||||||
impl SeqIds {
|
impl SeqIds {
|
||||||
@ -217,6 +248,55 @@ async fn detect_execute_awaits_async_runtime_without_nested_runtime_panic() {
|
|||||||
assert!(out.results[0].available);
|
assert!(out.results[0].available);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test(flavor = "multi_thread")]
|
||||||
|
async fn detect_profiles_runs_candidate_probes_concurrently_and_keeps_order() {
|
||||||
|
let started = Arc::new(AtomicUsize::new(0));
|
||||||
|
let active = Arc::new(AtomicUsize::new(0));
|
||||||
|
let max_active = Arc::new(AtomicUsize::new(0));
|
||||||
|
let delay = Duration::from_millis(100);
|
||||||
|
let runtime: Arc<dyn AgentRuntime> = Arc::new(SlowRuntime {
|
||||||
|
delay,
|
||||||
|
started: Arc::clone(&started),
|
||||||
|
active,
|
||||||
|
max_active: Arc::clone(&max_active),
|
||||||
|
});
|
||||||
|
let candidates = vec![
|
||||||
|
profile(1, "One", "one"),
|
||||||
|
profile(2, "Two", "two"),
|
||||||
|
profile(3, "Three", "three"),
|
||||||
|
profile(4, "Four", "four"),
|
||||||
|
profile(5, "Five", "five"),
|
||||||
|
];
|
||||||
|
let detect = DetectProfiles::new(runtime);
|
||||||
|
|
||||||
|
let started_at = Instant::now();
|
||||||
|
let out = detect
|
||||||
|
.execute(DetectProfilesInput { candidates })
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let elapsed = started_at.elapsed();
|
||||||
|
|
||||||
|
assert_eq!(started.load(Ordering::SeqCst), 5, "all candidates probed");
|
||||||
|
assert_eq!(
|
||||||
|
max_active.load(Ordering::SeqCst),
|
||||||
|
5,
|
||||||
|
"all probes should overlap instead of running sequentially"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
elapsed < Duration::from_millis(300),
|
||||||
|
"parallel probes should take roughly one probe duration, got {elapsed:?}"
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
out.results
|
||||||
|
.iter()
|
||||||
|
.map(|entry| entry.profile.command.as_str())
|
||||||
|
.collect::<Vec<_>>(),
|
||||||
|
vec!["one", "two", "three", "four", "five"],
|
||||||
|
"result order stays aligned with candidate order"
|
||||||
|
);
|
||||||
|
assert!(out.results.iter().all(|entry| !entry.available));
|
||||||
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
// ConfigureProfiles
|
// ConfigureProfiles
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|||||||
Reference in New Issue
Block a user