feat(async): AsyncProblem trait + run_async on RandomSearch and DifferentialEvolution

Adds the headline async/await capability for IO-bound evaluations
(HTTP services, RPC clients, spawned subprocesses) — the
differentiator vs pymoo / hyperopt / MOEA Framework.

No public-API breaks for synchronous users. The new surface is
gated behind a new `async` feature flag.

- core::async_problem::AsyncProblem trait (async fn evaluate_async).
- algorithms::parallel_eval_async::evaluate_batch_async helper using
  futures::stream::FuturesOrdered with concurrency-bounded chunks;
  preserves input order so seeded determinism holds when evaluations
  are themselves deterministic.
- run_async on RandomSearch and DifferentialEvolution.
- examples/async_eval.rs: simulated 20 ms remote service. concurrency=1
  → 4.2 s, concurrency=4 → 2.1 s (2× speedup).

Bumps Cargo.toml to 0.8.0; CHANGELOG entry covers the above plus a
note that 0.6.0/0.7.0 on crates.io are yanked experimentals and 0.8
picks up cleanly from 0.5.
This commit is contained in:
2026-05-06 07:55:56 -06:00
parent fa3f2e8fb0
commit 6368ca5f3d
10 changed files with 396 additions and 2 deletions
+39 -1
View File
@@ -7,6 +7,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [0.8.0] — 2026-05-06
Theme: async evaluation. heuropt now supports problems where each
evaluation is a `.await`-able operation — HTTP services, RPC clients,
spawned subprocesses. This is the differentiating capability vs.
pymoo / hyperopt / MOEA Framework, none of which ship first-class
async support.
No public-API breaks for synchronous users. The new surface is
gated behind a new `async` feature flag.
> **Note on version numbers.** Versions 0.6.0 and 0.7.0 were
> published on crates.io but contained experimental observability
> APIs and metrics that were rolled back. Both are yanked. 0.8.0
> picks up cleanly from 0.5.0 with just the async additions; if
> you were on 0.5.x, upgrading to 0.8 is a feature-additive bump.
### Added
- New optional feature `async`, gated on
[`futures`](https://crates.io/crates/futures).
- `core::async_problem::AsyncProblem` trait — mirrors `Problem` but
with `async fn evaluate_async(&self, decision)`. Adapt an
existing sync `Problem` with a one-line wrapper.
- Per-algorithm `run_async(&problem, concurrency).await` methods on
`RandomSearch` and `DifferentialEvolution` — drives evaluations
through whichever async runtime the caller is using (typically
tokio). `concurrency` bounds in-flight evaluations.
- Internal `algorithms::parallel_eval_async::evaluate_batch_async`
helper — uses `futures::stream::FuturesOrdered` with concurrency-
bounded chunks, preserves input order so seeded determinism is
preserved when evaluations are themselves deterministic.
- `examples/async_eval.rs` — worked example with a simulated 20 ms
remote service. At concurrency = 1 it's serial; at concurrency = 4
it's 2× faster; demonstrates DifferentialEvolution under tokio.
[0.8.0]: https://github.com/swaits/heuropt/releases/tag/v0.8.0
## [0.5.0] — 2026-05-05
Theme: comprehensive documentation and project polish. No public-API
@@ -469,5 +507,5 @@ Initial release.
`RandomSearch`, `Nsga2`, and `DifferentialEvolution`. Seeded runs stay
bit-identical to serial mode.
[Unreleased]: https://github.com/swaits/heuropt/compare/v0.5.0...HEAD
[Unreleased]: https://github.com/swaits/heuropt/compare/v0.8.0...HEAD
[0.1.0]: https://github.com/swaits/heuropt/releases/tag/v0.1.0
+8 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "heuropt"
version = "0.5.0"
version = "0.8.0"
edition = "2024"
rust-version = "1.85"
authors = ["Stephen Waits <steve@waits.net>"]
@@ -17,8 +17,10 @@ categories = ["algorithms", "science", "mathematics", "simulation"]
default = []
serde = ["dep:serde"]
parallel = ["dep:rayon"]
async = ["dep:futures"]
[dependencies]
futures = { version = "0.3", optional = true, default-features = false, features = ["std", "async-await"] }
rand = "0.9"
rand_distr = "0.5"
rayon = { version = "1", optional = true }
@@ -27,11 +29,16 @@ serde = { version = "1", features = ["derive"], optional = true }
[dev-dependencies]
gungraun = "0.18"
proptest = "1"
tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] }
[[bench]]
name = "hot_paths"
harness = false
[[example]]
name = "async_eval"
required-features = ["async"]
# Tighten release codegen for the compare harness and downstream binaries
# that build heuropt directly (i.e. when this crate is the workspace root).
# When heuropt is used as a dependency the consumer's profile wins.
+84
View File
@@ -0,0 +1,84 @@
//! Async evaluation example: optimize hyperparameters where each
//! evaluation is an awaitable (simulated HTTP) call.
//!
//! Demonstrates:
//! - Implementing [`AsyncProblem`].
//! - Driving the optimizer through `tokio` with bounded concurrency.
//! - Comparing wall-clock time at concurrency = 1 vs 8.
//!
//! Run with: `cargo run --release --features async --example async_eval`
use std::time::Instant;
use heuropt::core::async_problem::AsyncProblem;
use heuropt::prelude::*;
struct RemoteService;
impl AsyncProblem for RemoteService {
type Decision = Vec<f64>;
fn objectives(&self) -> ObjectiveSpace {
ObjectiveSpace::new(vec![Objective::minimize("loss")])
}
async fn evaluate_async(&self, x: &Vec<f64>) -> Evaluation {
// Simulate a 20 ms remote-service round-trip per evaluation.
// The compute itself is ~free; the latency is the bottleneck.
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
let loss: f64 = x.iter().map(|v| v * v).sum();
Evaluation::new(vec![loss])
}
}
#[tokio::main]
async fn main() {
let bounds = vec![(-1.0_f64, 1.0_f64); 4];
let problem = RemoteService;
println!("RandomSearch with 200 evaluations (20 ms each)");
println!();
for &concurrency in &[1_usize, 4, 16] {
let mut opt = RandomSearch::new(
RandomSearchConfig {
iterations: 100,
batch_size: 2,
seed: 42,
},
RealBounds::new(bounds.clone()),
);
let started = Instant::now();
let result = opt.run_async(&problem, concurrency).await;
let elapsed = started.elapsed();
println!(
"concurrency = {:>2} elapsed = {:>5} ms best loss = {:>8.5} evaluations = {}",
concurrency,
elapsed.as_millis(),
result.best.unwrap().evaluation.objectives[0],
result.evaluations,
);
}
println!();
println!("DifferentialEvolution at concurrency=8");
let started = Instant::now();
let mut de = DifferentialEvolution::new(
DifferentialEvolutionConfig {
population_size: 8,
generations: 10,
differential_weight: 0.5,
crossover_probability: 0.9,
seed: 42,
},
RealBounds::new(bounds.clone()),
);
let result = de.run_async(&problem, 8).await;
let elapsed = started.elapsed();
println!(
"elapsed = {:>5} ms best loss = {:>8.5} evaluations = {}",
elapsed.as_millis(),
result.best.unwrap().evaluation.objectives[0],
result.evaluations,
);
}
+100
View File
@@ -184,6 +184,106 @@ where
}
}
#[cfg(feature = "async")]
impl DifferentialEvolution {
/// Async version of [`Optimizer::run`] — drives evaluations through
/// the user-chosen async runtime. Available only with the `async`
/// feature.
///
/// `concurrency` bounds in-flight evaluations per batch (initial
/// population and per-generation trials).
pub async fn run_async<P>(
&mut self,
problem: &P,
concurrency: usize,
) -> OptimizationResult<Vec<f64>>
where
P: crate::core::async_problem::AsyncProblem<Decision = Vec<f64>>,
{
use rand::Rng as _;
use crate::algorithms::parallel_eval_async::evaluate_batch_async;
use crate::traits::Initializer as _;
assert!(
self.config.population_size >= 4,
"DifferentialEvolution requires population_size >= 4",
);
assert!(
(0.0..=1.0).contains(&self.config.crossover_probability),
"DifferentialEvolution crossover_probability must be in [0.0, 1.0]",
);
let objectives = problem.objectives();
assert!(
objectives.is_single_objective(),
"DifferentialEvolution only supports single-objective problems",
);
let direction = objectives.objectives[0].direction;
let dim = self.bounds.bounds.len();
let n = self.config.population_size;
let mut rng = rng_from_seed(self.config.seed);
let mut decisions: Vec<Vec<f64>> = self.bounds.initialize(n, &mut rng);
let initial_pop = evaluate_batch_async(problem, decisions.clone(), concurrency).await;
let mut evaluations = initial_pop.len();
let mut current_pop = initial_pop;
let mut evals: Vec<f64> = current_pop
.iter()
.map(|c| c.evaluation.objectives[0])
.collect();
for _generation in 0..self.config.generations {
let trials: Vec<Vec<f64>> = (0..n)
.map(|i| {
let (r1, r2, r3) = pick_three_distinct(n, i, &mut rng);
let j_rand = rng.random_range(0..dim);
let mut trial = decisions[i].clone();
for j in 0..dim {
let take_donor =
rng.random_bool(self.config.crossover_probability) || j == j_rand;
if take_donor {
let mutant = decisions[r1][j]
+ self.config.differential_weight
* (decisions[r2][j] - decisions[r3][j]);
let (lo, hi) = self.bounds.bounds[j];
trial[j] = mutant.clamp(lo, hi);
}
}
trial
})
.collect();
let trial_cands: Vec<Candidate<Vec<f64>>> =
evaluate_batch_async(problem, trials, concurrency).await;
evaluations += trial_cands.len();
for (i, trial_cand) in trial_cands.into_iter().enumerate() {
let trial_obj = trial_cand.evaluation.objectives[0];
let target_obj = evals[i];
let trial_better = match direction {
Direction::Minimize => trial_obj <= target_obj,
Direction::Maximize => trial_obj >= target_obj,
};
if trial_better {
decisions[i] = trial_cand.decision.clone();
evals[i] = trial_obj;
current_pop[i] = trial_cand;
}
}
}
let front = pareto_front(&current_pop, &objectives);
let best = best_candidate(&current_pop, &objectives);
OptimizationResult::new(
Population::new(current_pop),
front,
best,
evaluations,
self.config.generations,
)
}
}
fn pick_three_distinct(
n: usize,
exclude: usize,
+2
View File
@@ -22,6 +22,8 @@ pub mod nsga3;
pub mod one_plus_one_es;
pub mod paes;
pub(crate) mod parallel_eval;
#[cfg(feature = "async")]
pub(crate) mod parallel_eval_async;
pub mod particle_swarm;
pub mod pesa2;
pub mod random_search;
+58
View File
@@ -0,0 +1,58 @@
//! Async population evaluator.
//!
//! Available only with the `async` feature. Used by the `run_async`
//! method on algorithms that support async problems.
use futures::stream::{FuturesOrdered, StreamExt};
use crate::core::async_problem::AsyncProblem;
use crate::core::candidate::Candidate;
/// Evaluate every decision concurrently against `problem`, preserving
/// input order in the returned vector. Concurrency is bounded by
/// `concurrency` (≥ 1) — too high a value wastes memory and may
/// overload downstream services; too low forfeits parallelism.
///
/// Returns a future that the caller drives via their preferred
/// runtime (typically tokio).
pub async fn evaluate_batch_async<P>(
problem: &P,
decisions: Vec<P::Decision>,
concurrency: usize,
) -> Vec<Candidate<P::Decision>>
where
P: AsyncProblem,
{
assert!(
concurrency >= 1,
"evaluate_batch_async concurrency must be >= 1"
);
let mut out: Vec<Candidate<P::Decision>> = Vec::with_capacity(decisions.len());
// Process in concurrency-bounded chunks to keep peak memory low
// and avoid blasting downstream services. Each chunk uses
// FuturesOrdered to preserve per-chunk order, and chunks are
// emitted in their natural order.
let mut iter = decisions.into_iter();
loop {
let mut futs = FuturesOrdered::new();
for _ in 0..concurrency {
match iter.next() {
Some(d) => {
futs.push_back(async move {
let e = problem.evaluate_async(&d).await;
Candidate::new(d, e)
});
}
None => break,
}
}
if futs.is_empty() {
break;
}
while let Some(c) = futs.next().await {
out.push(c);
}
}
out
}
+45
View File
@@ -113,6 +113,51 @@ where
}
}
#[cfg(feature = "async")]
impl<I> RandomSearch<I> {
/// Async version of [`Optimizer::run`] — drives evaluations through
/// the user-chosen async runtime (typically tokio). Useful when
/// `evaluate` is IO-bound (HTTP, RPC, subprocess).
///
/// `concurrency` bounds how many evaluations are in-flight at once;
/// `1` is sequential, larger values push more load to the
/// downstream service.
///
/// Available only with the `async` feature.
pub async fn run_async<P>(
&mut self,
problem: &P,
concurrency: usize,
) -> OptimizationResult<P::Decision>
where
P: crate::core::async_problem::AsyncProblem,
I: Initializer<P::Decision>,
{
use crate::algorithms::parallel_eval_async::evaluate_batch_async;
let objectives = problem.objectives();
let mut rng = rng_from_seed(self.config.seed);
let mut all: Vec<Candidate<P::Decision>> = Vec::new();
let mut evaluations = 0usize;
for _ in 0..self.config.iterations {
let decisions = self
.initializer
.initialize(self.config.batch_size, &mut rng);
evaluations += decisions.len();
let cands = evaluate_batch_async(problem, decisions, concurrency).await;
all.extend(cands);
}
let front = pareto_front(&all, &objectives);
let best = best_candidate(&all, &objectives);
OptimizationResult::new(
Population::new(all),
front,
best,
evaluations,
self.config.iterations,
)
}
}
#[cfg(test)]
mod tests {
use super::*;
+54
View File
@@ -0,0 +1,54 @@
//! Async-evaluable problems for IO-bound workloads.
//!
//! Most heuropt algorithms operate synchronously: their `Problem::evaluate`
//! returns immediately. For workloads where evaluation is *IO-bound* — calling
//! an HTTP service, querying a remote model, spawning a subprocess —
//! awaiting an async fn is much more efficient than blocking a worker
//! thread.
//!
//! [`AsyncProblem`] mirrors [`Problem`](crate::core::Problem) but its
//! `evaluate_async` returns a future. Algorithms that support async
//! evaluation (NSGA-II, DE, RandomSearch as of v0.7.0; others land
//! incrementally) expose a `run_async` method that drives evaluations
//! through a user-chosen async runtime (typically tokio).
//!
//! Available only with the `async` feature.
use std::future::Future;
use crate::core::evaluation::Evaluation;
use crate::core::objective::ObjectiveSpace;
/// A problem whose evaluation is async — useful when `evaluate` does
/// IO (HTTP, RPC, subprocess) rather than pure CPU work.
///
/// Mirrors [`Problem`](crate::core::Problem) one-for-one except that
/// `evaluate_async` returns a future. The returned future must be
/// `Send` so the algorithm can run many evaluations concurrently
/// across a runtime's worker pool.
///
/// Implementors who already have a synchronous `Problem` can adapt
/// to `AsyncProblem` with a one-line wrapper:
///
/// ```ignore
/// impl AsyncProblem for MyProblem {
/// type Decision = <Self as Problem>::Decision;
/// fn objectives(&self) -> ObjectiveSpace { Problem::objectives(self) }
/// async fn evaluate_async(&self, x: &Self::Decision) -> Evaluation {
/// Problem::evaluate(self, x)
/// }
/// }
/// ```
pub trait AsyncProblem: Sync {
/// The thing the optimizer changes. Same constraints as
/// [`Problem::Decision`](crate::core::Problem::Decision).
type Decision: Clone + Send + Sync;
/// Return the objectives for this problem.
fn objectives(&self) -> ObjectiveSpace;
/// Evaluate `decision` asynchronously. The returned future is
/// driven by whichever runtime the algorithm's `run_async` is
/// invoked from.
fn evaluate_async(&self, decision: &Self::Decision) -> impl Future<Output = Evaluation> + Send;
}
+4
View File
@@ -1,5 +1,7 @@
//! Concrete data types and the `Problem` trait that the rest of the crate is built on.
#[cfg(feature = "async")]
pub mod async_problem;
pub mod candidate;
pub mod evaluation;
pub mod objective;
@@ -9,6 +11,8 @@ pub mod problem;
pub mod result;
pub mod rng;
#[cfg(feature = "async")]
pub use async_problem::AsyncProblem;
pub use candidate::*;
pub use evaluation::*;
pub use objective::*;
+2
View File
@@ -8,6 +8,8 @@ pub use crate::core::{
Candidate, Direction, Evaluation, Objective, ObjectiveSpace, OptimizationResult,
PartialProblem, Population, Problem, Rng, rng_from_seed,
};
#[cfg(feature = "async")]
pub use crate::core::async_problem::AsyncProblem;
pub use crate::traits::{Initializer, Optimizer, Repair, Variation};