test(async): add concurrency verification tests for optimize_parallel
- Timing-based proof that trials run in parallel, not sequentially - Atomic counter proof that multiple workers overlap simultaneously - Panic propagation returns TaskError with panic message - Partial failures with parallel path store only successful trials
This commit is contained in:
@@ -4,6 +4,8 @@
|
|||||||
|
|
||||||
#![cfg(feature = "async")]
|
#![cfg(feature = "async")]
|
||||||
|
|
||||||
|
use std::sync::Arc;
|
||||||
|
|
||||||
use optimizer::parameter::{FloatParam, Parameter};
|
use optimizer::parameter::{FloatParam, Parameter};
|
||||||
use optimizer::sampler::random::RandomSampler;
|
use optimizer::sampler::random::RandomSampler;
|
||||||
use optimizer::sampler::tpe::TpeSampler;
|
use optimizer::sampler::tpe::TpeSampler;
|
||||||
@@ -192,3 +194,118 @@ async fn test_optimize_parallel_single_concurrency() {
|
|||||||
|
|
||||||
assert_eq!(study.n_trials(), 10);
|
assert_eq!(study.n_trials(), 10);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_parallel_executes_concurrently() {
|
||||||
|
let sampler = RandomSampler::with_seed(42);
|
||||||
|
let study: Study<f64> = Study::with_sampler(Direction::Minimize, sampler);
|
||||||
|
|
||||||
|
let x_param = FloatParam::new(0.0, 10.0);
|
||||||
|
|
||||||
|
let start = tokio::time::Instant::now();
|
||||||
|
study
|
||||||
|
.optimize_parallel(4, 4, move |trial: &mut optimizer::Trial| {
|
||||||
|
let x = x_param.suggest(trial)?;
|
||||||
|
std::thread::sleep(std::time::Duration::from_millis(100));
|
||||||
|
Ok::<_, Error>(x)
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.expect("parallel optimization should succeed");
|
||||||
|
|
||||||
|
let elapsed = start.elapsed();
|
||||||
|
assert_eq!(study.n_trials(), 4);
|
||||||
|
// Sequential would take ~400ms; parallel with concurrency=4 should be ~100ms
|
||||||
|
assert!(
|
||||||
|
elapsed < std::time::Duration::from_millis(350),
|
||||||
|
"expected parallel execution under 350ms, took {elapsed:?}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_parallel_max_concurrency_reached() {
|
||||||
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
|
|
||||||
|
let sampler = RandomSampler::with_seed(42);
|
||||||
|
let study: Study<f64> = Study::with_sampler(Direction::Minimize, sampler);
|
||||||
|
|
||||||
|
let x_param = FloatParam::new(0.0, 10.0);
|
||||||
|
let active = Arc::new(AtomicUsize::new(0));
|
||||||
|
let max_active = Arc::new(AtomicUsize::new(0));
|
||||||
|
|
||||||
|
let active_c = Arc::clone(&active);
|
||||||
|
let max_active_c = Arc::clone(&max_active);
|
||||||
|
|
||||||
|
study
|
||||||
|
.optimize_parallel(8, 4, move |trial: &mut optimizer::Trial| {
|
||||||
|
let x = x_param.suggest(trial)?;
|
||||||
|
|
||||||
|
let current = active_c.fetch_add(1, Ordering::SeqCst) + 1;
|
||||||
|
// Update max_active if this is the highest seen so far
|
||||||
|
max_active_c.fetch_max(current, Ordering::SeqCst);
|
||||||
|
|
||||||
|
std::thread::sleep(std::time::Duration::from_millis(50));
|
||||||
|
|
||||||
|
active_c.fetch_sub(1, Ordering::SeqCst);
|
||||||
|
Ok::<_, Error>(x)
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.expect("parallel optimization should succeed");
|
||||||
|
|
||||||
|
assert_eq!(study.n_trials(), 8);
|
||||||
|
let max = max_active.load(Ordering::SeqCst);
|
||||||
|
assert!(
|
||||||
|
max >= 2,
|
||||||
|
"expected at least 2 concurrent workers, but max was {max}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_parallel_panic_returns_task_error() {
|
||||||
|
let study: Study<f64> = Study::new(Direction::Minimize);
|
||||||
|
|
||||||
|
let result = study
|
||||||
|
.optimize_parallel(3, 2, |_trial: &mut optimizer::Trial| {
|
||||||
|
panic!("boom");
|
||||||
|
#[allow(unreachable_code)]
|
||||||
|
Ok::<_, Error>(0.0)
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
|
||||||
|
match result {
|
||||||
|
Err(Error::TaskError(msg)) => {
|
||||||
|
assert!(
|
||||||
|
msg.contains("panic"),
|
||||||
|
"expected error message to contain 'panic', got: {msg}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
other => panic!("expected TaskError, got {other:?}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_parallel_partial_failures_trial_count() {
|
||||||
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
|
|
||||||
|
let sampler = RandomSampler::with_seed(42);
|
||||||
|
let study: Study<f64> = Study::with_sampler(Direction::Minimize, sampler);
|
||||||
|
|
||||||
|
let x_param = FloatParam::new(0.0, 10.0);
|
||||||
|
let counter = Arc::new(AtomicUsize::new(0));
|
||||||
|
|
||||||
|
let counter_c = Arc::clone(&counter);
|
||||||
|
|
||||||
|
study
|
||||||
|
.optimize_parallel(10, 3, move |trial: &mut optimizer::Trial| {
|
||||||
|
let idx = counter_c.fetch_add(1, Ordering::SeqCst);
|
||||||
|
if idx % 2 == 1 {
|
||||||
|
return Err(Error::NoCompletedTrials);
|
||||||
|
}
|
||||||
|
let x = x_param.suggest(trial)?;
|
||||||
|
Ok::<_, Error>(x)
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.expect("should succeed with partial failures");
|
||||||
|
|
||||||
|
// Even indices succeed (0, 2, 4, 6, 8), odd indices fail
|
||||||
|
assert_eq!(study.n_trials(), 5);
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user