diff --git a/AGENTS.md b/AGENTS.md index 1eab4b9c..7ee10c4d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -177,7 +177,12 @@ composition. Clone is acceptable only for: 1. Solution snapshots sent through channels. -2. Partitioned search creating partition copies. +2. Parallel search creating worker-private solution copies: partitioned + search's partition copies, and parallel local-search evaluation's + evaluator replicas — one working-solution copy per evaluator beyond the + main director, made at phase start and again after the working solution is + written outside the parallel loop's own commits. Replicas borrow the solve's + descriptor; they never copy it. 3. Explicit user-owned result snapshots outside retained scoring state, such as `CollectedVec::to_vec()` when the caller deliberately asks for an owned `Vec`. @@ -276,6 +281,9 @@ Use `rayon` for CPU-bound work and `tokio` for async coordination. - Do not add `std::thread::spawn`. - Do not add `std::thread::sleep`. - Prefer ownership transfer over callback APIs that only expose `&Self`. +- Parallel local-search evaluation runs one `rayon::scope` per evaluation + window on the solve thread's current pool; do not add dedicated pools or + blocking waits inside evaluation workers. ### Configured Completion diff --git a/README.md b/README.md index f176bbee..3f5be0c2 100644 --- a/README.md +++ b/README.md @@ -741,6 +741,13 @@ SolverForge leverages Rust's zero-cost abstractions: - **Cache-friendly**: Contiguous memory layouts for hot paths - **Concrete native execution**: Score directors, deciders, bounders, move carriers, and selector carriers remain monomorphized; documented scorer-agnostic `&dyn Director` callbacks are narrow intentional seams +Local search can evaluate candidates on several cores within one solve: +`move_thread_count = { count = 4 }` in `solver.toml` (Python configs through +solverforge-py use `{"move_thread_count": {"count": 4}}`) runs each step's +trials on up to four evaluators while one coordinator keeps the same +acceptance order. Results are reproducible at fixed work, and the default +(`auto`) currently stays sequential. + Absolute throughput depends on the model, constraints, neighborhood, hardware, and termination policy. Use emitted generated/evaluated throughput for one run and the controlled paired benchmark gate below for regression decisions. diff --git a/assets/solverforge-architecture.svg b/assets/solverforge-architecture.svg index 4e97e178..3ded0f32 100644 --- a/assets/solverforge-architecture.svg +++ b/assets/solverforge-architecture.svg @@ -176,6 +176,7 @@ PartitionedSearch — rayon work split ExhaustiveSearch · CustomSearchPhase phases ordered by SolverConfig +LocalSearch — rayon window evaluation on replicas diff --git a/crates/solverforge-bridge/WIREFRAME.md b/crates/solverforge-bridge/WIREFRAME.md index 229095bb..30a78866 100644 --- a/crates/solverforge-bridge/WIREFRAME.md +++ b/crates/solverforge-bridge/WIREFRAME.md @@ -178,6 +178,11 @@ presentation and host-boundary conversion. `scoped_dynamic_score_family(family, callback)` sets the thread-local family used by `DynamicScore::zero()` and `DynamicScore::from_level_numbers(...)` during the callback. +The family is per thread. The solve thread enters it for the whole solve, but +parallel local-search evaluation scores evaluator replicas on Rayon worker +threads, so a binding's constraint set that builds `DynamicScore` values must +also enter its declared family inside its own `ConstraintSet` methods. + ## Public Functions ### Compiled dynamic runner entrypoints @@ -185,7 +190,7 @@ callback)` sets the thread-local family used by `DynamicScore::zero()` and ```rust pub fn try_run_dynamic_solver_with_config_parts( solution: S, - constraints: C, + constraints: fn(&S) -> C, descriptor: SolutionDescriptor, entity_count_by_descriptor: fn(&S, usize) -> usize, runtime: SolverRuntime, @@ -207,6 +212,10 @@ where This is the binding-oriented entrypoint. It consumes frozen dynamic model parts into the same compiled graph and phase runner used by native models. It accepts no phase builder, selector registration, or alternate construction path. +`constraints` builds a constraint set for a given solution; the runtime calls it +for the working solution and for every evaluator replica, so +`move_thread_count` evaluates dynamic models in parallel. Bindings rebuild +their host constraint set from the solution they receive. The optional qualified provenance changes only the candidate-trace header; it does not select another runner. diff --git a/crates/solverforge-bridge/src/runner.rs b/crates/solverforge-bridge/src/runner.rs index d04a18da..9539af20 100644 --- a/crates/solverforge-bridge/src/runner.rs +++ b/crates/solverforge-bridge/src/runner.rs @@ -63,10 +63,14 @@ where /// /// The one model value is consumed by the canonical compiled runner. There is /// no host-language construction branch or deferred alternate execution path. +/// +/// `constraints` builds a constraint set for a given solution. The runtime +/// calls it for the working solution and for every evaluator replica, so +/// `move_thread_count` evaluates dynamic models in parallel. #[allow(clippy::too_many_arguments)] pub fn try_run_dynamic_solver_with_config_parts( solution: S, - constraints: C, + constraints: fn(&S) -> C, descriptor: SolutionDescriptor, entity_count_by_descriptor: fn(&S, usize) -> usize, runtime: SolverRuntime, diff --git a/crates/solverforge-bridge/src/score.rs b/crates/solverforge-bridge/src/score.rs index 5c94dd80..aadf7e33 100644 --- a/crates/solverforge-bridge/src/score.rs +++ b/crates/solverforge-bridge/src/score.rs @@ -27,6 +27,11 @@ thread_local! { /// This keeps the internal static three-level score contract available to the /// solver while allowing binding-driven solves to present zero and derived /// scores with the host model's declared score family. +/// +/// The family is per thread. The solve thread enters it for the whole solve, +/// but parallel evaluation scores evaluator replicas on Rayon worker threads, +/// so a constraint set that builds `DynamicScore` values must also enter its +/// declared family inside its own `ConstraintSet` methods. pub fn scoped_dynamic_score_family( family: DynamicScoreFamily, callback: impl FnOnce() -> T, diff --git a/crates/solverforge-config/WIREFRAME.md b/crates/solverforge-config/WIREFRAME.md index 44af1d6a..311c42d8 100644 --- a/crates/solverforge-config/WIREFRAME.md +++ b/crates/solverforge-config/WIREFRAME.md @@ -720,6 +720,19 @@ Derives: `Debug, Clone, Default, PartialEq, Eq, Deserialize, Serialize`. | `None` | Single-threaded | | `Count(usize)` | Explicit count | +TOML: `move_thread_count = "auto"`, `"none"`, or `{ count = 4 }`. + +`SolverConfig::move_thread_count` drives parallel evaluation in the compiled +runtime's acceptor-forager local search; `PartitionedSearchConfig::thread_count` +sizes partitioned search separately. + +| `move_thread_count` | Acceptor-forager local search | +|---|---| +| `None` | sequential | +| `Auto` | sequential | +| `Count(0)`, `Count(1)` | sequential | +| `Count(n >= 2)` | `min(n, rayon::current_num_threads())` evaluators: the main director plus that many minus one replicas | + ### `PhaseConfig` Derives: `Debug, Clone, Deserialize, Serialize`. Tagged `#[serde(tag = "type", rename_all = "snake_case")]`. diff --git a/crates/solverforge-config/src/tests/parsing.rs b/crates/solverforge-config/src/tests/parsing.rs index 6d97c498..764d81e5 100644 --- a/crates/solverforge-config/src/tests/parsing.rs +++ b/crates/solverforge-config/src/tests/parsing.rs @@ -248,3 +248,20 @@ fn grouped_scalar_move_selector_parses_and_roundtrips() { assert_eq!(selector.max_moves_per_step, Some(64)); assert!(selector.require_hard_improvement); } + +#[test] +fn move_thread_count_parses_every_documented_form() { + let parse = |value: &str| { + SolverConfig::from_toml_str(&format!("move_thread_count = {value}")) + .unwrap() + .move_thread_count + }; + + assert_eq!(parse("\"auto\""), MoveThreadCount::Auto); + assert_eq!(parse("\"none\""), MoveThreadCount::None); + assert_eq!(parse("{ count = 4 }"), MoveThreadCount::Count(4)); + assert_eq!( + SolverConfig::from_toml_str("").unwrap().move_thread_count, + MoveThreadCount::Auto + ); +} diff --git a/crates/solverforge-macros/WIREFRAME.md b/crates/solverforge-macros/WIREFRAME.md index 9a86c9f9..8f947581 100644 --- a/crates/solverforge-macros/WIREFRAME.md +++ b/crates/solverforge-macros/WIREFRAME.md @@ -254,7 +254,7 @@ semantics should omit `domain = "cvrp"` and declare explicit hook paths instead. - `impl SolvableSolution for T` — delegates to `descriptor()` and `entity_count()` - `impl Solvable for T` (when constraints path specified) — `solve(self, runtime: SolverRuntime, qualified_candidate_trace_provenance: Option)` delegates to `solve_internal()` - `impl Analyzable for T` (when constraints path specified) — `analyze()` creates `ScoreDirector` with canonical shadow support and returns `ScoreAnalysis` -- `fn solve_internal(self, runtime: SolverRuntime, qualified_candidate_trace_provenance: Option)` (when constraints path specified) — loads the configured base `SolverConfig` (or embedded `solver.toml`), applies the optional `config = "..."` callback, and calls hidden `try_run_solver_with_config_and_search`. The generated `__solverforge_search_declaration(config, descriptor)` builds one `RuntimeModel` containing scalar slots plus zero or more owner-specific list slots, delegates scalar hook and scalar-group attachment to the `planning_model!` support impl, sorts those variable slots to the descriptor-backed variable order emitted by the macros, constructs `SearchContext` with `try_new`, and returns the selected custom `Search` declaration or canonical defaults. The compiled runtime owns phase preparation, source binding, construction, local search, and terminal lifecycle; no macro-generated phase sequence exists. +- `fn solve_internal(self, runtime: SolverRuntime, qualified_candidate_trace_provenance: Option)` (when constraints path specified) — loads the configured base `SolverConfig` (or embedded `solver.toml`), applies the optional `config = "..."` callback, and calls hidden `try_run_solver_with_config_and_search`, passing the constraints function as the constraint-set constructor (`fn(&Self) -> _`) used for the working solution and every evaluator replica. The generated `__solverforge_search_declaration(config, descriptor)` builds one `RuntimeModel` containing scalar slots plus zero or more owner-specific list slots, delegates scalar hook and scalar-group attachment to the `planning_model!` support impl, sorts those variable slots to the descriptor-backed variable order emitted by the macros, constructs `SearchContext` with `try_new`, and returns the selected custom `Search` declaration or canonical defaults. The compiled runtime owns phase preparation, source binding, construction, local search, and terminal lifecycle; no macro-generated phase sequence exists. - Public solution source methods for all `#[planning_entity_collection]`, `#[problem_fact_collection]`, and streamable `#[planning_list_element_collection]` fields. Each method is inherent on the solution type, for example `Plan::tasks()`, returns the concrete hidden `SourceExtract &[Task]>` wrapper, and carries hidden `ChangeSource::Descriptor(idx)` for planning entities or `ChangeSource::Static` for facts and list elements. User constraints can call the explicit form `ConstraintFactory::new().for_each(Plan::tasks())` or import the generated `PlanConstraintStreams` trait and call `ConstraintFactory::new().tasks()`. Both paths use the same source metadata and concrete stream types. ### Problem Fact Support Derive diff --git a/crates/solverforge-macros/src/planning_solution/runtime/solve.rs b/crates/solverforge-macros/src/planning_solution/runtime/solve.rs index a5240655..afcceef6 100644 --- a/crates/solverforge-macros/src/planning_solution/runtime/solve.rs +++ b/crates/solverforge-macros/src/planning_solution/runtime/solve.rs @@ -39,7 +39,7 @@ pub(super) fn generate_runtime_solve_internal( #config_expr ::solverforge::__internal::try_run_solver_with_config_and_search( self, - #constraints_fn(), + (|_: &Self| #constraints_fn()) as fn(&Self) -> _, Self::descriptor(), Self::entity_count, runtime, diff --git a/crates/solverforge-macros/src/planning_solution_tests.rs b/crates/solverforge-macros/src/planning_solution_tests.rs index 413f8640..ea83a30b 100644 --- a/crates/solverforge-macros/src/planning_solution_tests.rs +++ b/crates/solverforge-macros/src/planning_solution_tests.rs @@ -65,6 +65,9 @@ fn golden_solution_expansion_loads_solver_config_before_config_callback() { expanded.contains("let config = crate :: config :: for_solution (& self , base_config)") ); assert!(expanded.contains("try_run_solver_with_config_and_search")); + assert!(expanded.contains( + "(| _ : & Self | crate :: constraints :: create_constraints ()) as fn (& Self) -> _" + )); } #[test] diff --git a/crates/solverforge-scoring/WIREFRAME.md b/crates/solverforge-scoring/WIREFRAME.md index 8ddc4476..fa2d360d 100644 --- a/crates/solverforge-scoring/WIREFRAME.md +++ b/crates/solverforge-scoring/WIREFRAME.md @@ -212,7 +212,8 @@ src/ │ ├── mod.rs — Re-exports all director types and traits │ ├── traits.rs — Director trait │ ├── score_director.rs — Re-exports ScoreDirector pieces -│ │ ├── score_director/incremental.rs — ScoreDirector (zero-erasure incremental) +│ │ ├── score_director/incremental.rs — ScoreDirector (zero-erasure incremental) +│ │ ├── score_director/descriptor.rs — DirectorDescriptor, ReplicaDescriptor │ │ └── score_director/adapters.rs — Debug and Director trait impls for ScoreDirector │ ├── shadow_aware.rs — SolvableSolution trait and shadow lifecycle notes │ └── tests/ @@ -333,7 +334,7 @@ pub use api::node_sharing::{SharedNodeDiagnostics, SharedNodeId, SharedNodeOpera pub use api::weight_overrides::{ConstraintWeightOverrides, WeightProvider}; // Score Directors -pub use director::score_director::ScoreDirector; +pub use director::score_director::{DirectorDescriptor, ReplicaDescriptor, ScoreDirector}; pub use director::{Director, DirectorScoreState, SolvableSolution}; // Analysis @@ -376,6 +377,7 @@ pub use stream::{ | `snapshot_score_state` | `fn snapshot_score_state(&self) -> DirectorScoreState` | Snapshot committed score state for speculative evaluation | | `restore_score_state` | `fn restore_score_state(&mut self, state: DirectorScoreState)` | Restore a previously snapshotted committed score state | | `reset` | `fn reset(&mut self)` | Default: no-op | +| `fork_replica` | `fn fork_replica(&self) -> Option where Self: Sized` | Independent evaluator replica with initialized incremental state, sharing the director's borrowed descriptor; default: `None` | ### `DirectorScoreState` @@ -502,7 +504,7 @@ Stock collectors include `count()`, `sum()`, `load_balance()`, ### Score Directors -**`ScoreDirector`** where `S: PlanningSolution`, `C: ConstraintSet` +**`ScoreDirector`** where `S: PlanningSolution`, `C: ConstraintSet`, `Desc: DirectorDescriptor` - Primary incremental scoring director. Zero-erasure. - Public methods: `new()`, `with_descriptor()`, `simple()` (convenience for `ScoreDirector`), `simple_zero()` (test helper with empty descriptor), @@ -515,8 +517,17 @@ Stock collectors include `count()`, `sum()`, `load_balance()`, - Returns borrowed constraint metadata views from the monomorphized `ConstraintSet` on demand. - `simple(solution, descriptor, entity_counter)` — creates `ScoreDirector` with empty constraint set - `simple_zero(solution)` — creates `ScoreDirector` with empty descriptor and zero entity counter +- `with_descriptor(solution, constraints, descriptor: Desc, entity_counter)` — `Desc` is an owned `SolutionDescriptor` or a `ReplicaDescriptor`; `new()`, `simple()` and `simple_zero()` build owned-descriptor directors. +- `Director::fork_replica` forks only directors holding a `ReplicaDescriptor`: the replica clones the working solution, builds its constraint set from that clone with the recorded constructor, initializes it, and shares the same borrowed descriptor. Owned-descriptor directors return `None`. - Implements `Director` +**`DirectorDescriptor`** (trait) — how a `ScoreDirector` holds its descriptor. +- `solution_descriptor(&self) -> &SolutionDescriptor` +- `replica(&self) -> Option<(Self, fn(&S) -> C)>` — the handle and constraint-set constructor for an evaluator replica; `None` for an owned `SolutionDescriptor`. + +**`ReplicaDescriptor<'d, S, C>`** — `Copy`; a borrowed `&'d SolutionDescriptor` plus the constraint-set constructor `fn(&S) -> C`. +- `new(descriptor: &'d SolutionDescriptor, constraints: fn(&S) -> C)` + ### Constraint Types All implement `IncrementalConstraint`. diff --git a/crates/solverforge-scoring/src/director/score_director.rs b/crates/solverforge-scoring/src/director/score_director.rs index fa81f575..c53cc414 100644 --- a/crates/solverforge-scoring/src/director/score_director.rs +++ b/crates/solverforge-scoring/src/director/score_director.rs @@ -5,6 +5,8 @@ constraint sets instead of trait-object-based scoring. */ mod adapters; +mod descriptor; mod incremental; +pub use descriptor::{DirectorDescriptor, ReplicaDescriptor}; pub use incremental::ScoreDirector; diff --git a/crates/solverforge-scoring/src/director/score_director/adapters.rs b/crates/solverforge-scoring/src/director/score_director/adapters.rs index cd39bd8c..380fdc0b 100644 --- a/crates/solverforge-scoring/src/director/score_director/adapters.rs +++ b/crates/solverforge-scoring/src/director/score_director/adapters.rs @@ -1,30 +1,37 @@ use solverforge_core::domain::{PlanningSolution, SolutionDescriptor}; use solverforge_core::score::Score; +use super::descriptor::DirectorDescriptor; use super::incremental::ScoreDirector; use crate::api::constraint_set::{ConstraintMetadata, ConstraintSet}; use crate::director::Director; -impl std::fmt::Debug for ScoreDirector +impl std::fmt::Debug for ScoreDirector where S: PlanningSolution + std::fmt::Debug, S::Score: std::fmt::Debug, C: ConstraintSet, + Desc: DirectorDescriptor, { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("ScoreDirector") .field("initialized", &self.is_initialized()) .field("cached_score", &self.get_score()) .field("constraint_count", &self.constraint_count()) + .field( + "forks_replicas", + &self.solution_descriptor.replica().is_some(), + ) .finish() } } -impl Director for ScoreDirector +impl Director for ScoreDirector where S: PlanningSolution, S::Score: Score, C: ConstraintSet + Send, + Desc: DirectorDescriptor + Send, { fn working_solution(&self) -> &S { self.working_solution() @@ -43,7 +50,7 @@ where } fn solution_descriptor(&self) -> &SolutionDescriptor { - &self.solution_descriptor + self.solution_descriptor.solution_descriptor() } fn clone_working_solution(&self) -> S { @@ -66,7 +73,11 @@ where } fn total_entity_count(&self) -> Option { - let count: usize = (0..self.solution_descriptor.entity_descriptors.len()) + let count: usize = (0..self + .solution_descriptor + .solution_descriptor() + .entity_descriptors + .len()) .map(|i| (self.entity_counter)(&self.working_solution, i)) .sum(); Some(count) @@ -91,4 +102,8 @@ where fn reset(&mut self) { self.reset_impl(); } + + fn fork_replica(&self) -> Option { + self.fork_replica_impl() + } } diff --git a/crates/solverforge-scoring/src/director/score_director/descriptor.rs b/crates/solverforge-scoring/src/director/score_director/descriptor.rs new file mode 100644 index 00000000..f9297119 --- /dev/null +++ b/crates/solverforge-scoring/src/director/score_director/descriptor.rs @@ -0,0 +1,71 @@ +use std::marker::PhantomData; + +use solverforge_core::domain::SolutionDescriptor; + +/* How a `ScoreDirector` holds its solution descriptor. + +An owned `SolutionDescriptor` serves a single director. A +`ReplicaDescriptor` borrows the descriptor and carries the constraint-set +constructor, so the director can fork evaluator replicas that share the same +descriptor without copying it. +*/ +pub trait DirectorDescriptor { + fn solution_descriptor(&self) -> &SolutionDescriptor; + + /* A handle for an evaluator replica plus the constructor that builds the + replica's constraint set from its own working solution. `None` when + this director cannot fork replicas. + */ + fn replica(&self) -> Option<(Self, fn(&S) -> C)> + where + Self: Sized; +} + +impl DirectorDescriptor for SolutionDescriptor { + #[inline] + fn solution_descriptor(&self) -> &SolutionDescriptor { + self + } + + fn replica(&self) -> Option<(Self, fn(&S) -> C)> { + None + } +} + +/* A borrowed solution descriptor plus the constraint-set constructor used for +the working solution and every evaluator replica. +*/ +pub struct ReplicaDescriptor<'d, S, C> { + descriptor: &'d SolutionDescriptor, + constraints: fn(&S) -> C, + _phantom: PhantomData (S, C)>, +} + +impl<'d, S, C> ReplicaDescriptor<'d, S, C> { + pub fn new(descriptor: &'d SolutionDescriptor, constraints: fn(&S) -> C) -> Self { + Self { + descriptor, + constraints, + _phantom: PhantomData, + } + } +} + +impl Clone for ReplicaDescriptor<'_, S, C> { + fn clone(&self) -> Self { + *self + } +} + +impl Copy for ReplicaDescriptor<'_, S, C> {} + +impl DirectorDescriptor for ReplicaDescriptor<'_, S, C> { + #[inline] + fn solution_descriptor(&self) -> &SolutionDescriptor { + self.descriptor + } + + fn replica(&self) -> Option<(Self, fn(&S) -> C)> { + Some((*self, self.constraints)) + } +} diff --git a/crates/solverforge-scoring/src/director/score_director/incremental.rs b/crates/solverforge-scoring/src/director/score_director/incremental.rs index 175d1757..e0ea76bc 100644 --- a/crates/solverforge-scoring/src/director/score_director/incremental.rs +++ b/crates/solverforge-scoring/src/director/score_director/incremental.rs @@ -3,6 +3,7 @@ use std::marker::PhantomData; use solverforge_core::domain::{PlanningSolution, SolutionDescriptor}; use solverforge_core::score::Score; +use super::descriptor::DirectorDescriptor; use crate::api::constraint_set::{ConstraintMetadata, ConstraintSet}; /* A zero-erasure score director for incremental scoring. @@ -15,6 +16,9 @@ are known at compile time, enabling complete monomorphization. - `S`: The solution type (must implement `PlanningSolution`) - `C`: The constraint set type (tuple of monomorphized constraints) +- `Desc`: How the solution descriptor is held; an owned `SolutionDescriptor` +by default, or a `ReplicaDescriptor` for directors that fork evaluator +replicas # Example @@ -61,16 +65,17 @@ let score2 = director.calculate_score(); assert_eq!(score, score2); ``` */ -pub struct ScoreDirector +pub struct ScoreDirector where S: PlanningSolution, C: ConstraintSet, + Desc: DirectorDescriptor, { pub(super) working_solution: S, constraints: C, cached_score: S::Score, initialized: bool, - pub(super) solution_descriptor: SolutionDescriptor, + pub(super) solution_descriptor: Desc, /* Entity counter function. Returns the number of entities for the given descriptor index. @@ -103,7 +108,15 @@ where |_, _| 0, ) } +} +impl ScoreDirector +where + S: PlanningSolution, + S::Score: Score, + C: ConstraintSet, + Desc: DirectorDescriptor, +{ /* Creates a new zero-erasure score director with a solution descriptor. This constructor enables the `Director` trait implementation for @@ -113,13 +126,14 @@ where * `solution` - The initial solution * `constraints` - Tuple of monomorphized constraints (e.g., `(C1, C2, C3)`) - * `solution_descriptor` - Metadata for solver infrastructure + * `solution_descriptor` - Metadata for solver infrastructure: an owned + `SolutionDescriptor`, or a `ReplicaDescriptor` to fork evaluator replicas * `entity_counter` - Function returning entity count for descriptor index */ pub fn with_descriptor( solution: S, constraints: C, - solution_descriptor: SolutionDescriptor, + solution_descriptor: Desc, entity_counter: fn(&S, usize) -> usize, ) -> Self { Self { @@ -133,6 +147,27 @@ where } } + /* Builds an initialized evaluator replica: a clone of the working + solution, a constraint set built for that clone, and the same borrowed + descriptor. + + The working-solution clone is the evaluator replica clone boundary: it + happens when replicas are created, never per candidate or per step. + */ + pub(crate) fn fork_replica_impl(&self) -> Option { + let (descriptor, constraints) = self.solution_descriptor.replica()?; + let solution = self.working_solution.clone(); + let replica_constraints = constraints(&solution); + let mut replica = Self::with_descriptor( + solution, + replica_constraints, + descriptor, + self.entity_counter, + ); + replica.calculate_score_impl(); + Some(replica) + } + /* ========================================================================= Private implementation methods (shared between inherent and trait impl) ========================================================================= diff --git a/crates/solverforge-scoring/src/director/tests/score_director.rs b/crates/solverforge-scoring/src/director/tests/score_director.rs index fbec0417..9020c3ad 100644 --- a/crates/solverforge-scoring/src/director/tests/score_director.rs +++ b/crates/solverforge-scoring/src/director/tests/score_director.rs @@ -2,7 +2,7 @@ use crate::api::constraint_set::IncrementalConstraint; use crate::constraint::incremental::IncrementalUniConstraint; -use crate::director::score_director::ScoreDirector; +use crate::director::score_director::{ReplicaDescriptor, ScoreDirector}; use crate::stream::collection_extract::{source, ChangeSource}; use crate::Director; use solverforge_core::domain::{PlanningSolution, SolutionDescriptor}; @@ -307,3 +307,86 @@ fn test_add_then_remove_value() { director.do_change(0, 0, |s| s.values[0] = None); assert_eq!(director.get_score(), SoftScore::of(-2)); } + +fn test_descriptor() -> SolutionDescriptor { + SolutionDescriptor::new("TestSolution", TypeId::of::()) +} + +fn unassigned_constraints( + _solution: &TestSolution, +) -> (impl IncrementalConstraint,) { + (make_unassigned_constraint(),) +} + +#[test] +fn fork_replica_requires_a_replica_descriptor() { + let solution = TestSolution { + values: vec![Some(1), None], + score: None, + }; + let mut director = ScoreDirector::new(solution, (make_unassigned_constraint(),)); + director.calculate_score(); + + assert!(director.fork_replica().is_none()); +} + +#[test] +fn fork_replica_matches_score_and_stays_independent() { + let solution = TestSolution { + values: vec![Some(1), None, None], + score: None, + }; + let descriptor = test_descriptor(); + let mut director = ScoreDirector::with_descriptor( + solution.clone(), + unassigned_constraints(&solution), + ReplicaDescriptor::new(&descriptor, unassigned_constraints), + |_, _| 0, + ); + let main_score = director.calculate_score(); + + let mut replica = director + .fork_replica() + .expect("director with a constraint constructor must fork"); + assert!(replica.is_initialized()); + assert_eq!(replica.get_score(), main_score); + assert!( + std::ptr::eq( + Director::solution_descriptor(&replica), + Director::solution_descriptor(&director) + ), + "a replica must share the descriptor, not copy it" + ); + + replica.before_variable_changed(0, 1); + replica.working_solution_mut().values[1] = Some(7); + replica.after_variable_changed(0, 1); + + assert_eq!(replica.get_score(), SoftScore::of(-1)); + assert_eq!(director.get_score(), SoftScore::of(-2)); + assert_eq!(director.working_solution().values[1], None); +} + +#[test] +fn fork_of_uninitialized_director_initializes_the_replica() { + let solution = TestSolution { + values: vec![None, Some(3), None, None], + score: None, + }; + let descriptor = test_descriptor(); + let director = ScoreDirector::with_descriptor( + solution.clone(), + unassigned_constraints(&solution), + ReplicaDescriptor::new(&descriptor, unassigned_constraints), + |_, _| 0, + ); + assert!(!director.is_initialized()); + + let replica = director + .fork_replica() + .expect("director with a constraint constructor must fork"); + + assert!(replica.is_initialized()); + assert_eq!(replica.get_score(), SoftScore::of(-3)); + assert!(!director.is_initialized()); +} diff --git a/crates/solverforge-scoring/src/director/traits.rs b/crates/solverforge-scoring/src/director/traits.rs index 1bc9de4b..ed55a725 100644 --- a/crates/solverforge-scoring/src/director/traits.rs +++ b/crates/solverforge-scoring/src/director/traits.rs @@ -92,4 +92,16 @@ pub trait Director: Send { // Resets the score director state. fn reset(&mut self) {} + + /* Builds an independent evaluator replica: an owned copy of the working + solution with freshly initialized incremental constraint state. + + Returns `None` when this director cannot construct one. + */ + fn fork_replica(&self) -> Option + where + Self: Sized, + { + None + } } diff --git a/crates/solverforge-scoring/src/lib.rs b/crates/solverforge-scoring/src/lib.rs index 94b596bf..12a9e248 100644 --- a/crates/solverforge-scoring/src/lib.rs +++ b/crates/solverforge-scoring/src/lib.rs @@ -44,7 +44,7 @@ Score Directors ============================================================================ */ -pub use director::score_director::ScoreDirector; +pub use director::score_director::{DirectorDescriptor, ReplicaDescriptor, ScoreDirector}; pub use director::{Director, DirectorScoreState, SolvableSolution}; /* ============================================================================ diff --git a/crates/solverforge-solver/WIREFRAME.md b/crates/solverforge-solver/WIREFRAME.md index 889de513..63a5a5ca 100644 --- a/crates/solverforge-solver/WIREFRAME.md +++ b/crates/solverforge-solver/WIREFRAME.md @@ -266,13 +266,22 @@ src/ │ │ └── grouped_scalar/placer_stream.rs — Concrete candidate and assignment placement stream helpers │ ├── localsearch/ │ │ ├── mod.rs — Acceptor, local-search acceptor/forager, and LocalSearchPhase re-exports -│ │ ├── evaluation.rs — Shared local-search candidate evaluation and hard-delta classification +│ │ ├── evaluation.rs — Trial kernel (run_trial), candidate accounting (account_trial), and hard-delta classification │ │ ├── phase.rs — LocalSearchPhase +│ │ ├── phase/candidates.rs — Candidate loop for one step +│ │ ├── phase/decision.rs — Generated-candidate records and in-order candidate consumption +│ │ ├── phase/step.rs — One step: candidate selection and commit │ │ ├── phase/tests.rs — Tests │ │ ├── forager.rs — LocalSearchForager trait, AcceptedCountForager, FirstAcceptedForager, BestScoreForager, re-exports │ │ ├── forager/any_tests.rs — AnyForager dispatch tests │ │ ├── forager/improving.rs — FirstBestScoreImprovingForager, FirstLastStepScoreImprovingForager │ │ ├── forager/tests.rs — Tests +│ │ ├── parallel/mod.rs — Parallel evaluation module declarations +│ │ ├── parallel/budget.rs — Remaining in-phase evaluation budget for window sizing +│ │ ├── parallel/config.rs — move_thread_count resolution to an evaluator count +│ │ ├── parallel/pool.rs — EvaluatorPool: evaluator replicas, sync, and post-commit advance +│ │ ├── parallel/window.rs — Window evaluation across the main director and replicas +│ │ ├── parallel/tests.rs — Tests │ │ └── acceptor/ │ │ ├── mod.rs — Acceptor module declarations and re-exports │ │ ├── traits.rs — Acceptor trait @@ -1245,6 +1254,39 @@ boundaries report it automatically, and bounded inner candidate scans poll it for prompt progress and interruption; publication is not gated on accumulating a fixed-size candidate batch. +**Parallel evaluation.** Acceptor-forager local search in the compiled runtime +can evaluate candidates on several cores within one solve. Each step produces a +window of candidates serially on the main director, in canonical order; +evaluates the window on the main director plus `n - 1` persistent evaluator +replicas (one `rayon::scope` per window, slots handed out through one atomic +counter); and consumes the results in candidate order through the unchanged +acceptor and forager. The selected move is applied to every replica by borrow +before the owned commit, and each replica must reach the committed step score. +Replicas are forked from the working solution through +`Director::fork_replica`, again whenever the working solution was written +outside the loop's own commits. The solve entry points build the main +`ScoreDirector` with a `ReplicaDescriptor` that borrows the solve's +descriptor, so replicas share it; only the working solution is copied. +Contract: + +- One evaluator runs the sequential loop: same candidates, checks, order, and + telemetry. +- Windows adapt to the step: a step's first window matches the number of + candidates the previous step consumed (at least one per evaluator), and each + further window in the step doubles, up to 32. Candidates are always + consumed in canonical order, so selectors that keep no stream state across + steps reach the sequential trajectory for every evaluator count. Ruin + selectors (cross-step RNG streams) and lazily activated providers see the + speculative pulls, so their trajectory depends on the evaluator count; it is + reproducible for a given count. +- `moves_evaluated`, `score_calculations`, and the move-count and + score-calculation limits count only consumed candidates; the window never + exceeds the remaining in-phase move or score-calculation budget. +- Candidates evaluated ahead of the decision point and then discarded are + counted in `moves_speculative` and traced as `DiscardedSpeculative`. +- `LocalSearchPhase`, VND, construction, exhaustive search, and partitioned + child solves evaluate sequentially. + When runtime phase configuration is omitted, SolverForge always builds state-aware construction. It appends the model-aware streaming local-search default only when the top-level solver termination parses to an effective @@ -1598,7 +1640,10 @@ time, evaluation time, acceptance rate, selector-level telemetry, and exact `Throughput { count, elapsed }` views for generated/evaluated work. Human-facing `moves/s` is derived only at log/console formatting edges. `moves_generated` counts candidates actually yielded by a runtime cursor; it -does not count an unrequested logical tail. Selector `size()` and explicit full +does not count an unrequested logical tail. `moves_speculative` counts +candidates that parallel local-search evaluation evaluated ahead of the +sequential decision point and then discarded; it is zero for sequential +evaluation and never feeds `moves_evaluated` or `score_calculations`. Selector `size()` and explicit full cursor exhaustion cover logical neighborhood size and canonical order. `SolverTelemetry` snapshots expose the same counters plus an optional @@ -1647,7 +1692,8 @@ ordered `CandidateTraceDisposition` transitions. Sources cover construction, local search, VND, generic K-opt, and the specialized list construction/search trial paths. Dispositions distinguish interruption, evaluation, not-doable, hard/score-improvement rejection, acceptor rejection, forager loss, selection, -and application. +application, and `DiscardedSpeculative` (pulled into a parallel evaluation +window, then discarded when the step ended before it was consumed). Logical identities are `CandidateTraceIdentity::Operation` or ordered `Composite`. `CandidateTraceOperationIdentity` carries descriptor, optional @@ -1731,11 +1777,13 @@ partitioned phase starts currently omit a score field. `try_run_solver_with_config_and_search(...) -> RuntimeBuildResult` is the one public configured solve entrypoint used by macro-generated solving. It accepts -the solution/constraints/descriptor, entity counting and logging callbacks, a +the solution, a constraint-set constructor `fn(&S) -> C`, the descriptor, +entity counting and logging callbacks, a `SolverRuntime`, `SolverConfig`, fallback time limit, optional qualified candidate-trace provenance, and a fallible builder for one typed `Search` -declaration. Graph compilation and solve-owned source preparation remain -private; there is no public graph or alternate phase-builder fallback. +declaration. The constructor builds the working solution's constraint set and +every evaluator replica's. Graph compilation and solve-owned source preparation +remain private; there is no public graph or alternate phase-builder fallback. `RuntimeBuildError` is `Declaration { message }`, `Compilation { path, message }`, `Preparation { phase_index, message }`, or diff --git a/crates/solverforge-solver/src/manager/mod_tests_integration/lifecycle_solutions.rs b/crates/solverforge-solver/src/manager/mod_tests_integration/lifecycle_solutions.rs index 8589fb3c..ed623bb0 100644 --- a/crates/solverforge-solver/src/manager/mod_tests_integration/lifecycle_solutions.rs +++ b/crates/solverforge-solver/src/manager/mod_tests_integration/lifecycle_solutions.rs @@ -556,6 +556,8 @@ fn zero_variable_log_scale(solution: &ZeroVariableLifecycleSolution) { solution.gate.wait_for_permit(); } +fn zero_variable_constraints(_solution: &ZeroVariableLifecycleSolution) {} + impl Solvable for ZeroVariableLifecycleSolution { fn solve( self, @@ -564,7 +566,7 @@ impl Solvable for ZeroVariableLifecycleSolution { ) { let result = try_run_solver_with_config_and_search( self, - (), + zero_variable_constraints, zero_variable_solution_descriptor(), zero_variable_entity_count, runtime, diff --git a/crates/solverforge-solver/src/phase/localsearch/evaluation.rs b/crates/solverforge-solver/src/phase/localsearch/evaluation.rs index 9324c79b..ed41a16d 100644 --- a/crates/solverforge-solver/src/phase/localsearch/evaluation.rs +++ b/crates/solverforge-solver/src/phase/localsearch/evaluation.rs @@ -1,4 +1,4 @@ -use std::time::Instant; +use std::time::{Duration, Instant}; use solverforge_core::domain::PlanningSolution; use solverforge_core::score::Score; @@ -17,6 +17,14 @@ pub(crate) enum CandidateEvaluation { RejectedByScoreImprovement(Sc), } +/* Director-only result of one trial transaction. Statistics are recorded by +`account_trial` on the thread that owns the step scope. +*/ +pub(crate) enum TrialOutcome { + NotDoable, + Scored(Sc), +} + #[inline] pub(crate) fn evaluate_candidate( mov: &MoveCandidateRef<'_, S, M>, @@ -32,34 +40,83 @@ where ProgressCb: ProgressCallback, M: Move, { - if move_changes_pinned(mov, step_scope.score_director()) - || !mov.is_doable_with_working_preview(step_scope.score_director_mut()) - { - let move_label = mov.telemetry_label(); - record_evaluated_move(step_scope, selector_index, evaluation_started); - step_scope - .phase_scope_mut() - .record_move_kind_evaluated_unscored(move_label); - if let Some(selector_index) = selector_index { + let outcome = run_trial(mov, step_scope.score_director_mut(), journal_pool); + account_trial( + mov, + step_scope, + reference_score, + selector_index, + evaluation_started.elapsed(), + outcome, + ) +} + +/* Runs the trial transaction on `director`: the pinned/doable check, then +apply, score, undo and restore. Touches nothing but the director and the +undo journal, so it can run on an evaluator replica. +*/ +#[inline] +pub(crate) fn run_trial( + mov: &MoveCandidateRef<'_, S, M>, + director: &mut D, + journal_pool: &mut Vec>, +) -> TrialOutcome +where + S: PlanningSolution, + D: Director, + M: Move, +{ + if move_changes_pinned(mov, director) || !mov.is_doable_with_working_preview(director) { + return TrialOutcome::NotDoable; + } + + let score_state = director.snapshot_score_state(); + let undo = mov.do_move_with_scratch(director, journal_pool); + let move_score = director.calculate_score(); + mov.undo_move_with_scratch(director, undo, journal_pool); + director.restore_score_state(score_state); + TrialOutcome::Scored(move_score) +} + +/* Records the statistics and improvement-gate decisions for one trial +outcome, in the order the sequential evaluation loop has always used. +*/ +#[inline] +pub(crate) fn account_trial( + mov: &MoveCandidateRef<'_, S, M>, + step_scope: &mut StepScope<'_, '_, '_, S, D, ProgressCb>, + reference_score: S::Score, + selector_index: Option, + elapsed: Duration, + outcome: TrialOutcome, +) -> CandidateEvaluation +where + S: PlanningSolution, + D: Director, + ProgressCb: ProgressCallback, + M: Move, +{ + let move_score = match outcome { + TrialOutcome::NotDoable => { + let move_label = mov.telemetry_label(); + record_evaluated_move(step_scope, selector_index, elapsed); step_scope .phase_scope_mut() - .record_selector_move_not_doable(selector_index); - } else { - step_scope.phase_scope_mut().record_move_not_doable(); + .record_move_kind_evaluated_unscored(move_label); + if let Some(selector_index) = selector_index { + step_scope + .phase_scope_mut() + .record_selector_move_not_doable(selector_index); + } else { + step_scope.phase_scope_mut().record_move_not_doable(); + } + step_scope + .phase_scope_mut() + .record_move_kind_not_doable(move_label); + return CandidateEvaluation::NotDoable; } - step_scope - .phase_scope_mut() - .record_move_kind_not_doable(move_label); - return CandidateEvaluation::NotDoable; - } - - let score_state = step_scope.score_director().snapshot_score_state(); - let undo = mov.do_move_with_scratch(step_scope.score_director_mut(), journal_pool); - let move_score = step_scope.score_director_mut().calculate_score(); - mov.undo_move_with_scratch(step_scope.score_director_mut(), undo, journal_pool); - step_scope - .score_director_mut() - .restore_score_state(score_state); + TrialOutcome::Scored(move_score) => move_score, + }; step_scope.phase_scope_mut().record_score_calculation(); @@ -79,7 +136,7 @@ where if mov.requires_hard_improvement() && hard_delta != Some(HardScoreDelta::Improving) { let move_label = mov.telemetry_label(); - record_evaluated_move(step_scope, selector_index, evaluation_started); + record_evaluated_move(step_scope, selector_index, elapsed); step_scope .phase_scope_mut() .record_move_kind_evaluated(move_label, move_score.compare(&reference_score)); @@ -98,7 +155,7 @@ where if mov.requires_score_improvement() && move_score <= reference_score { let move_label = mov.telemetry_label(); - record_evaluated_move(step_scope, selector_index, evaluation_started); + record_evaluated_move(step_scope, selector_index, elapsed); step_scope .phase_scope_mut() .record_move_kind_evaluated(move_label, move_score.compare(&reference_score)); @@ -122,7 +179,7 @@ where pub(crate) fn record_evaluated_move( step_scope: &mut StepScope<'_, '_, '_, S, D, ProgressCb>, selector_index: Option, - evaluation_started: Instant, + elapsed: Duration, ) where S: PlanningSolution, D: Director, @@ -131,10 +188,8 @@ pub(crate) fn record_evaluated_move( if let Some(selector_index) = selector_index { step_scope .phase_scope_mut() - .record_selector_evaluated_move(selector_index, evaluation_started.elapsed()); + .record_selector_evaluated_move(selector_index, elapsed); } else { - step_scope - .phase_scope_mut() - .record_evaluated_move(evaluation_started.elapsed()); + step_scope.phase_scope_mut().record_evaluated_move(elapsed); } } diff --git a/crates/solverforge-solver/src/phase/localsearch/mod.rs b/crates/solverforge-solver/src/phase/localsearch/mod.rs index 8731fa3f..e798f0ef 100644 --- a/crates/solverforge-solver/src/phase/localsearch/mod.rs +++ b/crates/solverforge-solver/src/phase/localsearch/mod.rs @@ -8,6 +8,7 @@ mod acceptor; mod cursor_source; mod evaluation; mod forager; +mod parallel; mod phase; pub(crate) mod vnd; @@ -24,5 +25,6 @@ pub use forager::{ AcceptedCountForager, BestScoreForager, FirstAcceptedForager, FirstBestScoreImprovingForager, FirstLastStepScoreImprovingForager, ForagerDecision, LocalSearchForager, }; +pub(crate) use parallel::resolve_evaluator_count; pub(crate) use phase::solve_local_search_with_resources; pub use phase::LocalSearchPhase; diff --git a/crates/solverforge-solver/src/phase/localsearch/parallel/budget.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/budget.rs new file mode 100644 index 00000000..7e372b78 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/budget.rs @@ -0,0 +1,30 @@ +use solverforge_core::domain::PlanningSolution; +use solverforge_scoring::Director; + +use crate::scope::{ProgressCallback, SolverScope}; + +/* Remaining in-phase evaluation budget: the smaller of the move-count and +score-calculation limits minus the work already counted, or `None` when the +phase sets neither limit. A window never exceeds it, so fixed-work limits stop +at the same candidate as sequential evaluation. +*/ +pub(crate) fn remaining_evaluation_budget( + solver_scope: &SolverScope<'_, S, D, BestCb>, +) -> Option +where + S: PlanningSolution, + D: Director, + BestCb: ProgressCallback, +{ + let stats = solver_scope.stats(); + let moves = solver_scope + .inphase_move_count_limit + .map(|limit| limit.saturating_sub(stats.moves_evaluated)); + let score_calculations = solver_scope + .inphase_score_calc_count_limit + .map(|limit| limit.saturating_sub(stats.score_calculations)); + match (moves, score_calculations) { + (Some(moves), Some(score_calculations)) => Some(moves.min(score_calculations)), + (moves, score_calculations) => moves.or(score_calculations), + } +} diff --git a/crates/solverforge-solver/src/phase/localsearch/parallel/config.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/config.rs new file mode 100644 index 00000000..6de83879 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/config.rs @@ -0,0 +1,12 @@ +use solverforge_config::MoveThreadCount; + +/* Number of evaluators (main director included) for acceptor-forager local +search. `Auto` stays sequential until benchmarks set its policy; an explicit +count is capped by the threads of the pool the solve runs on. +*/ +pub(crate) fn resolve_evaluator_count(count: &MoveThreadCount) -> usize { + match count { + MoveThreadCount::Auto | MoveThreadCount::None => 1, + MoveThreadCount::Count(count) => (*count).clamp(1, rayon::current_num_threads().max(1)), + } +} diff --git a/crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs new file mode 100644 index 00000000..524e9ac3 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs @@ -0,0 +1,19 @@ +/* Parallel local-search evaluation. + +The main director and persistent evaluator replicas run trials for one window +of candidates at a time; the coordinator consumes the results in candidate +order through the unchanged acceptor and forager. +*/ + +mod budget; +mod config; +mod pool; +mod window; + +#[cfg(test)] +mod tests; + +pub(crate) use budget::remaining_evaluation_budget; +pub(crate) use config::resolve_evaluator_count; +pub(crate) use pool::EvaluatorPool; +pub(crate) use window::evaluate_window; diff --git a/crates/solverforge-solver/src/phase/localsearch/parallel/pool.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/pool.rs new file mode 100644 index 00000000..c51811d2 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/pool.rs @@ -0,0 +1,197 @@ +use std::marker::PhantomData; +use std::time::Duration; + +use rayon::prelude::*; +use solverforge_core::domain::PlanningSolution; +use solverforge_scoring::Director; + +use crate::heuristic::r#move::Move; +use crate::heuristic::selector::move_selector::{MoveCandidateRef, MoveCandidateUndo}; +use crate::phase::localsearch::evaluation::TrialOutcome; +use crate::phase::localsearch::phase::WindowSlot; +use crate::scope::{ProgressCallback, SolverScope}; + +/* Largest parallel window. Windows adapt to how many candidates a step +consumes: a step's first window matches the previous step's consumption +(at least one candidate per evaluator), and each further window in the same +step doubles, up to this cap. Candidates past the decision point are +speculative work, so a window much longer than the step wastes evaluations. +*/ +pub(crate) const MAX_PARALLEL_EVALUATION_WINDOW: usize = 32; + +pub(super) type TrialOutputs = Vec<(usize, TrialOutcome, Duration)>; + +/// Trial results of one window, indexed by window slot. +pub(crate) type WindowResults = Vec, Duration)>>; + +/* Evaluators for one local-search phase: the main director plus +`requested - 1` persistent replicas, kept at the committed working solution. +*/ +pub(crate) struct EvaluatorPool +where + S: PlanningSolution, + M: Move, +{ + requested: usize, + replicas: Vec>, + main_outputs: TrialOutputs, + // Window buffers lent to each step, so steps reuse their capacity. + window_slots: Vec, + window_results: WindowResults, + // Size of the next window in the current step. + window: usize, + // Candidates the previous step consumed. + last_step_consumed: usize, + synced_generation: Option, + _phantom: PhantomData (S, M)>, +} + +pub(super) struct EvaluatorReplica +where + S: PlanningSolution, + M: Move, +{ + pub(super) director: D, + pub(super) journal_pool: Vec>, + pub(super) outputs: TrialOutputs, +} + +impl EvaluatorPool +where + S: PlanningSolution, + D: Director, + M: Move, +{ + // `requested` counts evaluators, the main director included. + pub(crate) fn new(requested: usize) -> Self { + Self { + requested: requested.max(1), + replicas: Vec::new(), + main_outputs: Vec::new(), + window_slots: Vec::new(), + window_results: Vec::new(), + window: 1, + last_step_consumed: 0, + synced_generation: None, + _phantom: PhantomData, + } + } + + /// Lends the window buffers to one step; return them with + /// `restore_window_buffers`. + pub(crate) fn take_window_buffers(&mut self) -> (Vec, WindowResults) { + ( + std::mem::take(&mut self.window_slots), + std::mem::take(&mut self.window_results), + ) + } + + pub(crate) fn restore_window_buffers( + &mut self, + mut slots: Vec, + mut results: WindowResults, + ) { + slots.clear(); + results.clear(); + self.window_slots = slots; + self.window_results = results; + } + + pub(crate) fn is_parallel(&self) -> bool { + !self.replicas.is_empty() + } + + /// Starts a step: its first window matches the previous step's + /// consumption, at least one candidate per evaluator. + pub(crate) fn begin_step(&mut self) { + self.window = self.last_step_consumed.clamp( + self.requested.min(MAX_PARALLEL_EVALUATION_WINDOW), + MAX_PARALLEL_EVALUATION_WINDOW, + ); + } + + /// Size of the next window, capped by the remaining evaluation budget. + /// Sequential evaluation always uses a window of one. + pub(crate) fn window_capacity(&self, remaining: Option) -> usize { + if !self.is_parallel() { + return 1; + } + let remaining = remaining.map_or(usize::MAX, |remaining| { + usize::try_from(remaining).unwrap_or(usize::MAX) + }); + self.window.min(remaining).max(1) + } + + /// The step outlasted its window: the next one doubles, up to the cap. + pub(crate) fn grow_window(&mut self) { + self.window = (self.window * 2).min(MAX_PARALLEL_EVALUATION_WINDOW); + } + + pub(crate) fn finish_step(&mut self, consumed: u64) { + self.last_step_consumed = usize::try_from(consumed).unwrap_or(usize::MAX); + } + + /* Rebuilds the replicas when the working solution was written since they + last synced. Each replica clones the working solution once; this is the + evaluator replica clone boundary. + */ + pub(crate) fn sync(&mut self, solver_scope: &SolverScope<'_, S, D, BestCb>) + where + BestCb: ProgressCallback, + { + let generation = solver_scope.working_solution_generation(); + if self.requested <= 1 || self.synced_generation == Some(generation) { + return; + } + self.replicas.clear(); + for _ in 1..self.requested { + let director = solver_scope.score_director().fork_replica().expect( + "parallel evaluation requires a director built with a constraint-set constructor", + ); + self.replicas.push(EvaluatorReplica { + director, + journal_pool: Vec::new(), + outputs: Vec::new(), + }); + } + self.synced_generation = Some(generation); + } + + /* Applies the selected move to every replica by borrow, ahead of the + owned commit on the main director, and checks that each replica reaches + the committed step score. + */ + pub(crate) fn advance(&mut self, selected: &MoveCandidateRef<'_, S, M>, expected: S::Score) { + if !self.is_parallel() { + return; + } + self.replicas.par_iter_mut().for_each(|replica| { + let _undo = selected.do_move(&mut replica.director); + let score = replica.director.calculate_score(); + assert!( + score == expected, + "evaluator replica diverged from the committed step: replica {score}, step {expected}" + ); + }); + } + + pub(crate) fn mark_synced(&mut self, generation: u64) { + if self.is_parallel() { + self.synced_generation = Some(generation); + } + } + + pub(super) fn parts_mut( + &mut self, + ) -> ( + &mut TrialOutputs, + &mut [EvaluatorReplica], + ) { + (&mut self.main_outputs, &mut self.replicas) + } + + #[cfg(test)] + pub(super) fn replicas(&self) -> impl Iterator { + self.replicas.iter().map(|replica| &replica.director) + } +} diff --git a/crates/solverforge-solver/src/phase/localsearch/parallel/tests.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/tests.rs new file mode 100644 index 00000000..1b2ccdbb --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/tests.rs @@ -0,0 +1,330 @@ +// Tests for parallel local-search evaluation. + +use solverforge_core::domain::PlanningSolution; +use solverforge_core::score::SoftScore; +use solverforge_core::{ConstraintRef, ImpactType}; +use solverforge_scoring::api::constraint_set::IncrementalConstraint; +use solverforge_scoring::stream::collection_extract::{source, ChangeSource}; +use solverforge_scoring::{Director, IncrementalUniConstraint, ReplicaDescriptor, ScoreDirector}; + +use super::EvaluatorPool; +use crate::heuristic::r#move::ChangeMove; +use crate::heuristic::selector::ChangeMoveSelector; +use crate::phase::localsearch::{ + solve_local_search_with_resources, AcceptedCountForager, BestScoreForager, + HillClimbingAcceptor, LocalSearchForager, SelectorCursorSource, +}; +use crate::scope::SolverScope; +use crate::test_utils::{ + create_nqueens_descriptor, get_queen_row, nqueens_clone_count, reset_nqueens_clone_count, + set_queen_row, NQueensSolution, Queen, +}; + +type QueenMove = ChangeMove; + +fn queen_weight(queen: &Queen) -> SoftScore { + match queen.row { + Some(row) => SoftScore::of((row * 3 + queen.column * 7) % 11), + None => SoftScore::of(20), + } +} + +fn queen_constraints( + _solution: &NQueensSolution, +) -> (impl IncrementalConstraint,) { + (IncrementalUniConstraint::new( + ConstraintRef::new("", "Queen placement"), + ImpactType::Penalty, + source( + (|s: &NQueensSolution| s.queens.as_slice()) as fn(&NQueensSolution) -> &[Queen], + ChangeSource::Descriptor(0), + ), + |_s: &NQueensSolution, _queen: &Queen| true, + queen_weight, + false, + ),) +} + +fn starting_solution() -> NQueensSolution { + NQueensSolution::with_rows(&[3, 1, 4, 1, 5, 9, 2, 6, 5, 3, 5, 8]) +} + +/* Builds a director that can fork replicas. A macro, because the director's +type names the opaque constraint set and the borrowed descriptor. +*/ +macro_rules! forkable_director { + ($solution:expr, $descriptor:expr) => {{ + let solution: NQueensSolution = $solution; + let constraints = queen_constraints(&solution); + ScoreDirector::with_descriptor( + solution, + constraints, + ReplicaDescriptor::new($descriptor, queen_constraints), + |s: &NQueensSolution, _| s.queens.len(), + ) + }}; +} + +type QueenSelector = ChangeMoveSelector< + NQueensSolution, + i64, + crate::heuristic::selector::FromSolutionEntitySelector, + crate::heuristic::selector::StaticValueSelector, +>; + +fn selector() -> SelectorCursorSource { + SelectorCursorSource::new(ChangeMoveSelector::simple( + get_queen_row, + set_queen_row, + 0, + 0, + "row", + (0..10).collect(), + )) +} + +struct RunResult { + rows: Vec>, + score: Option, + moves_evaluated: u64, + moves_applied: u64, + moves_speculative: u64, + steps: u64, +} + +fn run( + evaluator_count: usize, + mut forager: Fo, + steps: u64, + move_limit: Option, +) -> RunResult +where + Fo: LocalSearchForager, +{ + let descriptor = create_nqueens_descriptor(); + let mut solver_scope = + SolverScope::new(forkable_director!(starting_solution(), &descriptor)).with_seed(7); + solver_scope.start_solving(); + solver_scope.inphase_move_count_limit = move_limit; + let mut source = selector(); + let mut acceptor = HillClimbingAcceptor::new(); + solve_local_search_with_resources( + &mut source, + &mut (), + &mut acceptor, + &mut forager, + Some(steps), + evaluator_count, + &mut solver_scope, + ); + let stats = solver_scope.stats(); + RunResult { + rows: solver_scope + .working_solution() + .queens + .iter() + .map(|queen| queen.row) + .collect(), + score: solver_scope.best_score().copied(), + moves_evaluated: stats.moves_evaluated, + moves_applied: stats.moves_applied, + moves_speculative: stats.moves_speculative, + steps: stats.step_count, + } +} + +#[test] +fn pool_with_one_evaluator_never_forks() { + let director = + ScoreDirector::simple(starting_solution(), create_nqueens_descriptor(), |s, _| { + s.queens.len() + }); + let mut solver_scope = SolverScope::new(director); + solver_scope.start_solving(); + let mut pool = EvaluatorPool::::new(1); + + pool.sync(&solver_scope); + + assert!(!pool.is_parallel()); + assert_eq!(pool.window_capacity(None), 1); +} + +#[test] +#[should_panic(expected = "constraint-set constructor")] +fn pool_requires_a_forkable_director() { + let director = + ScoreDirector::simple(starting_solution(), create_nqueens_descriptor(), |s, _| { + s.queens.len() + }); + let mut solver_scope = SolverScope::new(director); + solver_scope.start_solving(); + let mut pool = EvaluatorPool::::new(2); + + pool.sync(&solver_scope); +} + +#[test] +fn pool_forks_one_replica_per_extra_evaluator_and_resyncs_after_writes() { + let descriptor = create_nqueens_descriptor(); + let mut solver_scope = SolverScope::new(forkable_director!(starting_solution(), &descriptor)); + solver_scope.start_solving(); + solver_scope.calculate_score(); + let mut pool = EvaluatorPool::::new(4); + + reset_nqueens_clone_count(); + pool.sync(&solver_scope); + assert_eq!(nqueens_clone_count(), 3); + assert_eq!(pool.replicas().count(), 3); + + pool.sync(&solver_scope); + assert_eq!( + nqueens_clone_count(), + 3, + "an unchanged generation must not fork again" + ); + + solver_scope.mutate(|director| { + director.before_variable_changed(0, 0); + director.working_solution_mut().queens[0].row = Some(7); + director.after_variable_changed(0, 0); + }); + let main_score = solver_scope.calculate_score(); + pool.sync(&solver_scope); + + assert_eq!(nqueens_clone_count(), 6); + for replica in pool.replicas() { + assert_eq!(replica.working_solution().queens[0].row, Some(7)); + assert_eq!(replica.working_solution().score(), Some(main_score)); + assert!( + std::ptr::eq(Director::solution_descriptor(replica), &descriptor), + "replicas must share the solve's descriptor, not copy it" + ); + } +} + +#[test] +fn sequential_and_parallel_reach_identical_trajectories() { + let sequential = run(1, AcceptedCountForager::new(4, false), 25, None); + for evaluator_count in [2, 4] { + let parallel = run( + evaluator_count, + AcceptedCountForager::new(4, false), + 25, + None, + ); + assert_eq!(parallel.rows, sequential.rows); + assert_eq!(parallel.score, sequential.score); + assert_eq!(parallel.moves_evaluated, sequential.moves_evaluated); + assert_eq!(parallel.moves_applied, sequential.moves_applied); + assert_eq!(parallel.steps, sequential.steps); + } + assert!(sequential.moves_applied > 0); +} + +#[test] +fn speculative_evaluations_are_reported_and_not_counted() { + let sequential = run(1, AcceptedCountForager::new(1, false), 10, None); + let parallel = run(2, AcceptedCountForager::new(1, false), 10, None); + + assert_eq!(parallel.rows, sequential.rows); + assert_eq!(parallel.moves_evaluated, sequential.moves_evaluated); + assert_eq!(sequential.moves_speculative, 0); + assert!(parallel.moves_speculative > 0); +} + +#[test] +fn in_phase_move_limit_reserves_the_window() { + let parallel = run(2, BestScoreForager::new(false), 10, Some(5)); + + assert_eq!(parallel.moves_evaluated, 5); + assert_eq!(parallel.moves_speculative, 0); +} + +#[test] +fn replicas_follow_solution_replacement() { + let descriptor = create_nqueens_descriptor(); + let run_with_replacement = |evaluator_count: usize| { + let mut solver_scope = + SolverScope::new(forkable_director!(starting_solution(), &descriptor)).with_seed(11); + solver_scope.start_solving(); + let mut source = selector(); + let mut acceptor = HillClimbingAcceptor::new(); + let mut forager = AcceptedCountForager::new(3, false); + solve_local_search_with_resources( + &mut source, + &mut (), + &mut acceptor, + &mut forager, + Some(5), + evaluator_count, + &mut solver_scope, + ); + solver_scope.replace_working_solution_and_reinitialize(NQueensSolution::with_rows(&[ + 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, + ])); + solve_local_search_with_resources( + &mut source, + &mut (), + &mut acceptor, + &mut forager, + Some(5), + evaluator_count, + &mut solver_scope, + ); + solver_scope + .working_solution() + .queens + .iter() + .map(|queen| queen.row) + .collect::>() + }; + + assert_eq!(run_with_replacement(2), run_with_replacement(1)); +} + +#[test] +fn evaluator_count_resolution() { + use solverforge_config::MoveThreadCount; + + use super::resolve_evaluator_count; + + assert_eq!(resolve_evaluator_count(&MoveThreadCount::None), 1); + assert_eq!(resolve_evaluator_count(&MoveThreadCount::Auto), 1); + assert_eq!(resolve_evaluator_count(&MoveThreadCount::Count(0)), 1); + assert_eq!(resolve_evaluator_count(&MoveThreadCount::Count(1)), 1); + assert_eq!( + resolve_evaluator_count(&MoveThreadCount::Count(2)), + 2.min(rayon::current_num_threads()) + ); +} + +#[test] +fn windows_track_the_steps_consumption() { + let descriptor = create_nqueens_descriptor(); + let mut solver_scope = SolverScope::new(forkable_director!(starting_solution(), &descriptor)); + solver_scope.start_solving(); + let mut pool = EvaluatorPool::::new(4); + pool.sync(&solver_scope); + + // First step: one candidate per evaluator, doubling while the step lasts. + pool.begin_step(); + assert_eq!(pool.window_capacity(None), 4); + pool.grow_window(); + assert_eq!(pool.window_capacity(None), 8); + pool.grow_window(); + pool.grow_window(); + pool.grow_window(); + assert_eq!(pool.window_capacity(None), 32); + assert_eq!(pool.window_capacity(Some(5)), 5); + pool.finish_step(23); + + // The next step starts at the previous step's consumption. + pool.begin_step(); + assert_eq!(pool.window_capacity(None), 23); + pool.finish_step(2); + pool.begin_step(); + assert_eq!(pool.window_capacity(None), 4); + pool.finish_step(500); + pool.begin_step(); + assert_eq!(pool.window_capacity(None), 32); +} diff --git a/crates/solverforge-solver/src/phase/localsearch/parallel/window.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/window.rs new file mode 100644 index 00000000..8cc1aba6 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/window.rs @@ -0,0 +1,86 @@ +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; + +use solverforge_core::domain::PlanningSolution; +use solverforge_scoring::Director; + +use crate::heuristic::r#move::Move; +use crate::heuristic::selector::move_selector::{MoveCandidateRef, MoveCandidateUndo}; +use crate::phase::localsearch::evaluation::{run_trial, TrialOutcome}; + +use super::pool::{EvaluatorPool, TrialOutputs}; + +/* Runs the trial for every candidate in the window. The main director takes +its share inline on the coordinator thread; each replica takes its share in a +spawned task. Slots are handed out through one atomic counter, and results land +at their slot index, so which evaluator ran a slot never affects the outcome. +*/ +pub(crate) fn evaluate_window( + candidates: &[MoveCandidateRef<'_, S, M>], + main_director: &mut D, + main_journal_pool: &mut Vec>, + evaluators: &mut EvaluatorPool, + results: &mut Vec, Duration)>>, +) where + S: PlanningSolution, + D: Director, + M: Move, +{ + let next = AtomicUsize::new(0); + let (main_outputs, replicas) = evaluators.parts_mut(); + main_outputs.clear(); + rayon::scope(|scope| { + for replica in replicas.iter_mut() { + let next = &next; + replica.outputs.clear(); + scope.spawn(move |_| { + drain_window( + candidates, + next, + &mut replica.director, + &mut replica.journal_pool, + &mut replica.outputs, + ); + }); + } + drain_window( + candidates, + &next, + main_director, + main_journal_pool, + main_outputs, + ); + }); + results.clear(); + results.resize_with(candidates.len(), || None); + for (index, outcome, elapsed) in main_outputs.drain(..) { + results[index] = Some((outcome, elapsed)); + } + for replica in replicas.iter_mut() { + for (index, outcome, elapsed) in replica.outputs.drain(..) { + results[index] = Some((outcome, elapsed)); + } + } +} + +fn drain_window( + candidates: &[MoveCandidateRef<'_, S, M>], + next: &AtomicUsize, + director: &mut D, + journal_pool: &mut Vec>, + outputs: &mut TrialOutputs, +) where + S: PlanningSolution, + D: Director, + M: Move, +{ + loop { + let index = next.fetch_add(1, Ordering::Relaxed); + let Some(candidate) = candidates.get(index) else { + break; + }; + let started = Instant::now(); + let outcome = run_trial(candidate, director, journal_pool); + outputs.push((index, outcome, started.elapsed())); + } +} diff --git a/crates/solverforge-solver/src/phase/localsearch/phase.rs b/crates/solverforge-solver/src/phase/localsearch/phase.rs index c1f6f720..b2b25e17 100644 --- a/crates/solverforge-solver/src/phase/localsearch/phase.rs +++ b/crates/solverforge-solver/src/phase/localsearch/phase.rs @@ -10,6 +10,7 @@ use tracing::info; use crate::heuristic::r#move::Move; use crate::heuristic::selector::move_selector::{CandidateId, MoveCandidateRef}; +use crate::phase::localsearch::parallel::EvaluatorPool; use crate::phase::localsearch::{ Acceptor, LocalSearchForager, MoveCursorSource, SelectorCursorSource, }; @@ -18,8 +19,11 @@ use crate::scope::{PhaseScope, ProgressCallback, SolverScope}; use crate::stats::{format_duration, whole_units_per_second, CandidateTracePullToken}; mod candidates; +mod decision; mod step; +pub(in crate::phase::localsearch) use decision::WindowSlot; + use step::{execute_step, StepOutcome}; const STEP_ACCEPTED_LABEL_LIMIT: usize = 32; @@ -240,6 +244,7 @@ pub(crate) fn solve_local_search_with_resources( acceptor: &mut A, forager: &mut Fo, step_limit: Option, + evaluator_count: usize, solver_scope: &mut SolverScope, ) where S: PlanningSolution, @@ -262,6 +267,7 @@ pub(crate) fn solve_local_search_with_resources( ); acceptor.phase_started(&last_step_score); + let mut evaluators = EvaluatorPool::::new(evaluator_count); let start_time = Instant::now(); loop { if phase_scope.solver_scope_mut().should_terminate() { @@ -278,6 +284,7 @@ pub(crate) fn solve_local_search_with_resources( forager, &mut phase_scope, &mut last_step_score, + &mut evaluators, ) { StepOutcome::Continue | StepOutcome::Restart => continue, StepOutcome::Terminate => break, @@ -344,6 +351,7 @@ where acceptor, forager, *step_limit, + 1, solver_scope, ); } diff --git a/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs b/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs index 6e722336..44a4f6b7 100644 --- a/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs +++ b/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs @@ -1,9 +1,7 @@ //! Candidate streaming for one local-search step. use solverforge_core::domain::PlanningSolution; -use solverforge_core::score::Score; use solverforge_scoring::Director; -use tracing::trace; use crate::heuristic::r#move::Move; use crate::heuristic::selector::move_selector::{ @@ -12,17 +10,18 @@ use crate::heuristic::selector::move_selector::{ use crate::phase::control::{ should_interrupt_before_candidate, should_interrupt_before_evaluation, CONTROL_POLL_INTERVAL, }; -use crate::phase::localsearch::evaluation::{ - evaluate_candidate, record_evaluated_move, CandidateEvaluation, +use crate::phase::localsearch::evaluation::run_trial; +use crate::phase::localsearch::parallel::{ + evaluate_window, remaining_evaluation_budget, EvaluatorPool, }; -use crate::phase::localsearch::{Acceptor, ForagerDecision, LocalSearchForager}; +use crate::phase::localsearch::{Acceptor, LocalSearchForager}; use crate::scope::{ProgressCallback, StepScope}; -use crate::stats::{CandidateTraceDisposition, CandidateTracePullToken, CandidateTraceSource}; +use crate::stats::CandidateTracePullToken; -use super::{candidate_selector_label, take_trace_token, StepMoveLabelCounts}; - -const PROGRESS_POLL_INTERVAL: u64 = 256; -// Candidate iterations between control polls in the step loop. +use super::decision::{ + consume_candidate, discard_window_from, interrupt_window_at, record_generated_candidate, +}; +use super::StepMoveLabelCounts; pub(super) struct CandidateLoopState { pub(super) interrupted: bool, @@ -55,6 +54,7 @@ pub(super) fn evaluate_candidates( acceptor: &mut A, forager: &mut Fo, requires_move_signatures: bool, + evaluators: &mut EvaluatorPool, mut state: CandidateLoopState, ) -> CandidateLoopState where @@ -69,255 +69,147 @@ where // The control poll reads the wall clock; forwarding every candidate // iteration makes it measurable at scale (measured ~9% of wall before // batching). The step loop therefore polls once per batch window; a - // stop stays latched, and per-evaluation interruption still runs on - // its every-64-evaluations schedule below. + // stop stays latched, and per-evaluation interruption still runs before + // every consumed candidate. // Retired undo journals are recycled across the step's trials so // arbitrary-long journals do not grow the heap per trial. + // + // Each pass produces a window of candidates serially on the main + // director, evaluates it (inline for one evaluator, across the evaluator + // pool otherwise), and consumes the results in candidate order. With a + // window of one, the checks, pulls and decisions are exactly the + // sequential loop's. let mut journal_pool: Vec> = Vec::new(); let mut control_polls = 0usize; let mut control_latched = false; + let (mut window, mut results) = evaluators.take_window_buffers(); + let consumed_before = state.evaluated_moves; + evaluators.begin_step(); while !forager.is_quit_early() { if state.interrupted { break; } - control_polls += 1; - if !control_latched && control_polls.is_multiple_of(CONTROL_POLL_INTERVAL) { - control_latched = should_interrupt_before_candidate(step_scope) - || step_scope - .phase_scope_mut() - .solver_scope_mut() - .should_terminate(); - } - if control_latched { - state.interrupted = true; - break; - } - let generation_started = std::time::Instant::now(); - let Some(candidate_id) = - cursor.next_candidate_with_resources(step_scope.score_director_mut(), resources) - else { - break; - }; - let selector_index = cursor.selector_index(candidate_id); - let mov = cursor - .candidate(candidate_id) - .expect("discovered candidate id must remain borrowable"); - let move_label = mov.telemetry_label(); - let selector_label = selector_index.map(|_| candidate_selector_label(&mov)); - let trace_token = step_scope.phase_scope_mut().record_candidate_pull( - CandidateTraceSource::LocalSearch, - selector_index, - candidate_id.index(), - None, - &mov, - ); - let generation_elapsed = generation_started.elapsed(); - state.generated_moves += 1; - step_scope - .phase_scope_mut() - .record_move_kind_generated(move_label); - if let Some(selector_index) = selector_index { - step_scope - .phase_scope_mut() - .record_selector_generated_move_with_label( - selector_index, - selector_label.as_deref().unwrap_or("selector"), - generation_elapsed, - ); - } else { - step_scope - .phase_scope_mut() - .record_generated_move(generation_elapsed); + // Production: serial, canonical order, main director. + let capacity = evaluators.window_capacity(remaining_evaluation_budget( + step_scope.phase_scope().solver_scope(), + )); + window.clear(); + let mut exhausted = false; + while window.len() < capacity { + control_polls += 1; + if !control_latched && control_polls.is_multiple_of(CONTROL_POLL_INTERVAL) { + control_latched = should_interrupt_before_candidate(step_scope) + || step_scope + .phase_scope_mut() + .solver_scope_mut() + .should_terminate(); + } + if control_latched { + state.interrupted = true; + break; + } + + let generation_started = std::time::Instant::now(); + let Some(candidate_id) = + cursor.next_candidate_with_resources(step_scope.score_director_mut(), resources) + else { + exhausted = true; + break; + }; + window.push(record_generated_candidate( + &*cursor, + candidate_id, + generation_started, + step_scope, + &mut state, + )); } - if step_scope.progress_polling_required() - && state.generated_moves.is_multiple_of(PROGRESS_POLL_INTERVAL) - { - step_scope.phase_scope_mut().report_progress_if_due(); + if window.is_empty() { + break; } - if should_interrupt_before_evaluation(step_scope) + // Dispatch: the sequential before-evaluation check for the first slot. + if state.interrupted + || should_interrupt_before_evaluation(step_scope) || step_scope .phase_scope_mut() .solver_scope_mut() .should_terminate() { - if let Some(token) = trace_token { - step_scope - .phase_scope_mut() - .record_candidate_trace_disposition( - token, - CandidateTraceDisposition::InterruptedBeforeEvaluation, - ); - } + interrupt_window_at(cursor, step_scope, &window, 0, false); state.interrupted = true; break; } - state.evaluated_moves += 1; - - let evaluation_started = std::time::Instant::now(); - let move_score = match evaluate_candidate( - &mov, - step_scope, - last_step_score, - selector_index, - evaluation_started, - &mut journal_pool, - ) { - CandidateEvaluation::Scored(score) => { - if let Some(token) = trace_token { - step_scope - .phase_scope_mut() - .record_candidate_trace_disposition( - token, - CandidateTraceDisposition::Evaluated, - ); - } - step_scope - .phase_scope_mut() - .record_move_kind_evaluated(move_label, score.compare(&last_step_score)); - score - } - CandidateEvaluation::NotDoable => { - record_rejected_trace( - step_scope, - trace_token, - CandidateTraceDisposition::NotDoable, - ); - assert!(cursor.release_candidate(candidate_id)); - continue; - } - CandidateEvaluation::RejectedByHardImprovement(_) => { - record_rejected_trace( - step_scope, - trace_token, - CandidateTraceDisposition::RejectedByHardImprovement, - ); - assert!(cursor.release_candidate(candidate_id)); - continue; - } - CandidateEvaluation::RejectedByScoreImprovement(_) => { - record_rejected_trace( - step_scope, - trace_token, - CandidateTraceDisposition::RejectedByScoreImprovement, - ); - assert!(cursor.release_candidate(candidate_id)); - continue; - } - }; - let move_signature = if requires_move_signatures { - match mov.borrowed_tabu_signature() { - Some(borrowed) => Some(borrowed.clone()), - None => Some(mov.tabu_signature(step_scope.score_director())), - } - } else { - None - }; - let accepted = acceptor.is_accepted(&last_step_score, &move_score, move_signature.as_ref()); - - record_evaluated_move(step_scope, selector_index, evaluation_started); - if accepted { - step_scope - .phase_scope_mut() - .record_move_kind_accepted(move_label); - state.accepted_labels.record(move_label); - if let Some(selector_index) = selector_index { - step_scope - .phase_scope_mut() - .record_selector_move_accepted(selector_index); - } else { - step_scope.phase_scope_mut().record_move_accepted(); - } - state.accepted_moves += 1; + // Evaluation. + if evaluators.is_parallel() { + let candidates: Vec<_> = window + .iter() + .map(|slot| { + cursor + .candidate(slot.id) + .expect("window candidate must remain borrowable") + }) + .collect(); + evaluate_window( + &candidates, + step_scope.score_director_mut(), + &mut journal_pool, + evaluators, + &mut results, + ); } else { - if let Some(token) = trace_token { - step_scope - .phase_scope_mut() - .record_candidate_trace_disposition( - token, - CandidateTraceDisposition::AcceptorRejected, - ); - } - if let Some(selector_index) = selector_index { - step_scope - .phase_scope_mut() - .record_selector_move_acceptor_rejected(selector_index); - } else { - step_scope.phase_scope_mut().record_move_acceptor_rejected(); - } - step_scope - .phase_scope_mut() - .record_move_kind_acceptor_rejected( - move_label, - move_score.compare(&last_step_score), - ); + let mov = cursor + .candidate(window[0].id) + .expect("discovered candidate id must remain borrowable"); + let started = std::time::Instant::now(); + let outcome = run_trial(&mov, step_scope.score_director_mut(), &mut journal_pool); + results.clear(); + results.push(Some((outcome, started.elapsed()))); } - trace!( - event = "step", - step = step_scope.step_index(), - move_index = candidate_id.index(), - score = %move_score, - accepted = accepted, - ); - - if accepted { - match forager.add_move_index(candidate_id, move_score) { - ForagerDecision::Keep => { - if let Some(token) = trace_token { - state.accepted_trace_tokens.push((candidate_id, token)); - } - } - ForagerDecision::Release => { - if let Some(token) = trace_token { - step_scope - .phase_scope_mut() - .record_candidate_trace_disposition( - token, - CandidateTraceDisposition::ForagerIgnored, - ); - } - assert!(cursor.release_candidate(candidate_id)); + // Consumption in candidate order. + for index in 0..window.len() { + if index > 0 { + if forager.is_quit_early() { + discard_window_from(cursor, step_scope, &window, index, true); + break; } - ForagerDecision::Replace(replaced_id) => { - if let Some(token) = - take_trace_token(&mut state.accepted_trace_tokens, replaced_id) - { - step_scope - .phase_scope_mut() - .record_candidate_trace_disposition( - token, - CandidateTraceDisposition::ForagerIgnored, - ); - } - if let Some(token) = trace_token { - state.accepted_trace_tokens.push((candidate_id, token)); - } - assert!(cursor.release_candidate(replaced_id)); + if should_interrupt_before_evaluation(step_scope) + || step_scope + .phase_scope_mut() + .solver_scope_mut() + .should_terminate() + { + interrupt_window_at(cursor, step_scope, &window, index, true); + state.interrupted = true; + break; } } - } else { - assert!(cursor.release_candidate(candidate_id)); + let (outcome, elapsed) = results[index] + .take() + .expect("every window slot has a trial result"); + consume_candidate( + cursor, + step_scope, + &window[index], + outcome, + elapsed, + last_step_score, + acceptor, + forager, + requires_move_signatures, + &mut state, + ); + } + if state.interrupted || exhausted { + break; } + evaluators.grow_window(); } + evaluators.finish_step(state.evaluated_moves - consumed_before); + evaluators.restore_window_buffers(window, results); state } - -fn record_rejected_trace( - step_scope: &mut StepScope<'_, '_, '_, S, D, BestCb>, - trace_token: Option, - disposition: CandidateTraceDisposition, -) where - S: PlanningSolution, - D: Director, - BestCb: ProgressCallback, -{ - if let Some(token) = trace_token { - let phase_scope = step_scope.phase_scope_mut(); - phase_scope.record_candidate_trace_disposition(token, CandidateTraceDisposition::Evaluated); - phase_scope.record_candidate_trace_disposition(token, disposition); - } -} diff --git a/crates/solverforge-solver/src/phase/localsearch/phase/decision.rs b/crates/solverforge-solver/src/phase/localsearch/phase/decision.rs new file mode 100644 index 00000000..ba7f71b2 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/phase/decision.rs @@ -0,0 +1,352 @@ +//! Candidate generation records and in-order candidate consumption for one +//! local-search step. + +use std::time::{Duration, Instant}; + +use solverforge_core::domain::PlanningSolution; +use solverforge_core::score::Score; +use solverforge_scoring::Director; +use tracing::trace; + +use crate::heuristic::r#move::Move; +use crate::heuristic::selector::move_selector::{CandidateId, ResourceMoveCursor}; +use crate::phase::localsearch::evaluation::{ + account_trial, record_evaluated_move, CandidateEvaluation, TrialOutcome, +}; +use crate::phase::localsearch::{Acceptor, ForagerDecision, LocalSearchForager}; +use crate::scope::{ProgressCallback, StepScope}; +use crate::stats::{CandidateTraceDisposition, CandidateTracePullToken, CandidateTraceSource}; + +use super::candidates::CandidateLoopState; +use super::{candidate_selector_label, take_trace_token}; + +const PROGRESS_POLL_INTERVAL: u64 = 256; + +/// A generated candidate waiting for its trial result. +pub(in crate::phase::localsearch) struct WindowSlot { + pub(super) id: CandidateId, + pub(super) selector_index: Option, + pub(super) move_label: &'static str, + pub(super) trace_token: Option, +} + +/// Records one pulled candidate: trace pull, generation counters and +/// telemetry, and the periodic progress poll. +pub(super) fn record_generated_candidate( + cursor: &C, + candidate_id: CandidateId, + generation_started: Instant, + step_scope: &mut StepScope<'_, '_, '_, S, D, BestCb>, + state: &mut CandidateLoopState, +) -> WindowSlot +where + S: PlanningSolution, + D: Director, + BestCb: ProgressCallback, + M: Move, + C: ResourceMoveCursor, +{ + let selector_index = cursor.selector_index(candidate_id); + let mov = cursor + .candidate(candidate_id) + .expect("discovered candidate id must remain borrowable"); + let move_label = mov.telemetry_label(); + let selector_label = selector_index.map(|_| candidate_selector_label(&mov)); + let trace_token = step_scope.phase_scope_mut().record_candidate_pull( + CandidateTraceSource::LocalSearch, + selector_index, + candidate_id.index(), + None, + &mov, + ); + let generation_elapsed = generation_started.elapsed(); + state.generated_moves += 1; + step_scope + .phase_scope_mut() + .record_move_kind_generated(move_label); + if let Some(selector_index) = selector_index { + step_scope + .phase_scope_mut() + .record_selector_generated_move_with_label( + selector_index, + selector_label.as_deref().unwrap_or("selector"), + generation_elapsed, + ); + } else { + step_scope + .phase_scope_mut() + .record_generated_move(generation_elapsed); + } + if step_scope.progress_polling_required() + && state.generated_moves.is_multiple_of(PROGRESS_POLL_INTERVAL) + { + step_scope.phase_scope_mut().report_progress_if_due(); + } + WindowSlot { + id: candidate_id, + selector_index, + move_label, + trace_token, + } +} + +/// Feeds one trial result through accounting, the acceptor and the forager, +/// and releases the candidate when the step no longer needs it. +#[allow(clippy::too_many_arguments)] +pub(super) fn consume_candidate( + cursor: &mut C, + step_scope: &mut StepScope<'_, '_, '_, S, D, BestCb>, + slot: &WindowSlot, + outcome: TrialOutcome, + elapsed: Duration, + last_step_score: S::Score, + acceptor: &mut A, + forager: &mut Fo, + requires_move_signatures: bool, + state: &mut CandidateLoopState, +) where + S: PlanningSolution, + D: Director, + BestCb: ProgressCallback, + M: Move, + C: ResourceMoveCursor, + A: Acceptor, + Fo: LocalSearchForager, +{ + let candidate_id = slot.id; + let selector_index = slot.selector_index; + let move_label = slot.move_label; + let trace_token = slot.trace_token; + state.evaluated_moves += 1; + + let mov = cursor + .candidate(candidate_id) + .expect("discovered candidate id must remain borrowable"); + let move_score = match account_trial( + &mov, + step_scope, + last_step_score, + selector_index, + elapsed, + outcome, + ) { + CandidateEvaluation::Scored(score) => { + if let Some(token) = trace_token { + step_scope + .phase_scope_mut() + .record_candidate_trace_disposition( + token, + CandidateTraceDisposition::Evaluated, + ); + } + step_scope + .phase_scope_mut() + .record_move_kind_evaluated(move_label, score.compare(&last_step_score)); + score + } + CandidateEvaluation::NotDoable => { + record_rejected_trace( + step_scope, + trace_token, + CandidateTraceDisposition::NotDoable, + ); + assert!(cursor.release_candidate(candidate_id)); + return; + } + CandidateEvaluation::RejectedByHardImprovement(_) => { + record_rejected_trace( + step_scope, + trace_token, + CandidateTraceDisposition::RejectedByHardImprovement, + ); + assert!(cursor.release_candidate(candidate_id)); + return; + } + CandidateEvaluation::RejectedByScoreImprovement(_) => { + record_rejected_trace( + step_scope, + trace_token, + CandidateTraceDisposition::RejectedByScoreImprovement, + ); + assert!(cursor.release_candidate(candidate_id)); + return; + } + }; + // Moves that own a prebuilt signature lend it; only dynamic builders + // produce one, and it lives for this decision alone. + let built_signature; + let move_signature = if requires_move_signatures { + match mov.borrowed_tabu_signature() { + Some(borrowed) => Some(borrowed), + None => { + built_signature = mov.tabu_signature(step_scope.score_director()); + Some(&built_signature) + } + } + } else { + None + }; + + let accepted = acceptor.is_accepted(&last_step_score, &move_score, move_signature); + + record_evaluated_move(step_scope, selector_index, elapsed); + if accepted { + step_scope + .phase_scope_mut() + .record_move_kind_accepted(move_label); + state.accepted_labels.record(move_label); + if let Some(selector_index) = selector_index { + step_scope + .phase_scope_mut() + .record_selector_move_accepted(selector_index); + } else { + step_scope.phase_scope_mut().record_move_accepted(); + } + state.accepted_moves += 1; + } else { + if let Some(token) = trace_token { + step_scope + .phase_scope_mut() + .record_candidate_trace_disposition( + token, + CandidateTraceDisposition::AcceptorRejected, + ); + } + if let Some(selector_index) = selector_index { + step_scope + .phase_scope_mut() + .record_selector_move_acceptor_rejected(selector_index); + } else { + step_scope.phase_scope_mut().record_move_acceptor_rejected(); + } + step_scope + .phase_scope_mut() + .record_move_kind_acceptor_rejected(move_label, move_score.compare(&last_step_score)); + } + + trace!( + event = "step", + step = step_scope.step_index(), + move_index = candidate_id.index(), + score = %move_score, + accepted = accepted, + ); + + if accepted { + match forager.add_move_index(candidate_id, move_score) { + ForagerDecision::Keep => { + if let Some(token) = trace_token { + state.accepted_trace_tokens.push((candidate_id, token)); + } + } + ForagerDecision::Release => { + if let Some(token) = trace_token { + step_scope + .phase_scope_mut() + .record_candidate_trace_disposition( + token, + CandidateTraceDisposition::ForagerIgnored, + ); + } + assert!(cursor.release_candidate(candidate_id)); + } + ForagerDecision::Replace(replaced_id) => { + if let Some(token) = take_trace_token(&mut state.accepted_trace_tokens, replaced_id) + { + step_scope + .phase_scope_mut() + .record_candidate_trace_disposition( + token, + CandidateTraceDisposition::ForagerIgnored, + ); + } + if let Some(token) = trace_token { + state.accepted_trace_tokens.push((candidate_id, token)); + } + assert!(cursor.release_candidate(replaced_id)); + } + } + } else { + assert!(cursor.release_candidate(candidate_id)); + } +} + +/// Ends the window at `index` because of an interruption: slot `index` gets +/// the sequential loop's `InterruptedBeforeEvaluation` record and stays live, +/// exactly as the sequential loop leaves it; later slots are discarded. +/// `evaluated` tells whether the window's trials already ran. +pub(super) fn interrupt_window_at( + cursor: &mut C, + step_scope: &mut StepScope<'_, '_, '_, S, D, BestCb>, + window: &[WindowSlot], + index: usize, + evaluated: bool, +) where + S: PlanningSolution, + D: Director, + BestCb: ProgressCallback, + M: Move, + C: ResourceMoveCursor, +{ + let slot = &window[index]; + if let Some(token) = slot.trace_token { + step_scope + .phase_scope_mut() + .record_candidate_trace_disposition( + token, + CandidateTraceDisposition::InterruptedBeforeEvaluation, + ); + } + if evaluated { + step_scope.phase_scope_mut().record_move_speculative(); + } + discard_window_from(cursor, step_scope, window, index + 1, evaluated); +} + +/// Discards slots `from..` of a window that the step will not consume: +/// records `DiscardedSpeculative`, releases them, and counts them as +/// speculative work when their trials already ran. +pub(super) fn discard_window_from( + cursor: &mut C, + step_scope: &mut StepScope<'_, '_, '_, S, D, BestCb>, + window: &[WindowSlot], + from: usize, + evaluated: bool, +) where + S: PlanningSolution, + D: Director, + BestCb: ProgressCallback, + M: Move, + C: ResourceMoveCursor, +{ + for slot in window.iter().skip(from) { + if let Some(token) = slot.trace_token { + step_scope + .phase_scope_mut() + .record_candidate_trace_disposition( + token, + CandidateTraceDisposition::DiscardedSpeculative, + ); + } + if evaluated { + step_scope.phase_scope_mut().record_move_speculative(); + } + assert!(cursor.release_candidate(slot.id)); + } +} + +fn record_rejected_trace( + step_scope: &mut StepScope<'_, '_, '_, S, D, BestCb>, + trace_token: Option, + disposition: CandidateTraceDisposition, +) where + S: PlanningSolution, + D: Director, + BestCb: ProgressCallback, +{ + if let Some(token) = trace_token { + let phase_scope = step_scope.phase_scope_mut(); + phase_scope.record_candidate_trace_disposition(token, CandidateTraceDisposition::Evaluated); + phase_scope.record_candidate_trace_disposition(token, disposition); + } +} diff --git a/crates/solverforge-solver/src/phase/localsearch/phase/step.rs b/crates/solverforge-solver/src/phase/localsearch/phase/step.rs index 2d579ca2..08d1e555 100644 --- a/crates/solverforge-solver/src/phase/localsearch/phase/step.rs +++ b/crates/solverforge-solver/src/phase/localsearch/phase/step.rs @@ -13,6 +13,7 @@ use crate::phase::control::{ settle_search_interrupt, should_interrupt_after_step, should_interrupt_before_candidate, StepInterrupt, }; +use crate::phase::localsearch::parallel::EvaluatorPool; use crate::phase::localsearch::{Acceptor, LocalSearchForager, MoveCursorSource}; use crate::scope::{PendingControl, PhaseScope, ProgressCallback, StepScope}; use crate::stats::{AppliedMoveTelemetry, CandidateTraceDisposition}; @@ -34,6 +35,7 @@ pub(super) fn execute_step( forager: &mut Fo, phase_scope: &mut PhaseScope<'_, '_, S, D, BestCb>, last_step_score: &mut S::Score, + evaluators: &mut EvaluatorPool, ) -> StepOutcome where S: PlanningSolution, @@ -45,6 +47,7 @@ where BestCb: ProgressCallback, { let mut step_scope = StepScope::new(phase_scope); + evaluators.sync(step_scope.phase_scope().solver_scope()); /* Reset forager and acceptor for this step. Pass best and last-step scores so foragers that implement @@ -91,6 +94,7 @@ where acceptor, forager, requires_move_signatures, + evaluators, CandidateLoopState::new(interrupted), ); @@ -139,9 +143,16 @@ where }; } let previous_score = *last_step_score; + evaluators.advance(&selected_move, selected_score); step_scope.apply_committed_change(|score_director| { cursor.apply_owned_candidate(selected_index, score_director); }); + evaluators.mark_synced( + step_scope + .phase_scope() + .solver_scope() + .working_solution_generation(), + ); if let Some(token) = selected_trace_token { step_scope .phase_scope_mut() diff --git a/crates/solverforge-solver/src/phase/localsearch/vnd/phase.rs b/crates/solverforge-solver/src/phase/localsearch/vnd/phase.rs index 38d98c48..4a45ab0b 100644 --- a/crates/solverforge-solver/src/phase/localsearch/vnd/phase.rs +++ b/crates/solverforge-solver/src/phase/localsearch/vnd/phase.rs @@ -338,7 +338,7 @@ where } }; - record_evaluated_move(step_scope, selector_index, evaluation_started); + record_evaluated_move(step_scope, selector_index, evaluation_started.elapsed()); progress.record_evaluated(); progress.maybe_report(step_scope); diff --git a/crates/solverforge-solver/src/run.rs b/crates/solverforge-solver/src/run.rs index 9e0223e1..ba1be095 100644 --- a/crates/solverforge-solver/src/run.rs +++ b/crates/solverforge-solver/src/run.rs @@ -12,7 +12,7 @@ use std::path::Path; use solverforge_config::{SolverConfig, TerminationConfig}; use solverforge_core::domain::{PlanningSolution, SolutionDescriptor}; use solverforge_core::score::{ParseableScore, Score}; -use solverforge_scoring::{ConstraintSet, Director, ScoreDirector}; +use solverforge_scoring::{ConstraintSet, Director, ReplicaDescriptor, ScoreDirector}; use tracing::info; use crate::builder::{RuntimeExtensionRegistry, Search}; @@ -207,6 +207,18 @@ where S: PlanningSolution, S::Score: Score + ParseableScore, C: ConstraintSet, +{ + build_director_termination(config, default_secs) +} + +fn build_director_termination( + config: &SolverConfig, + default_secs: u64, +) -> (AnyTermination, Option) +where + S: PlanningSolution, + S::Score: Score + ParseableScore, + D: Director, { let ConfiguredTermination { time_limit: configured_time_limit, @@ -419,10 +431,14 @@ fn load_solver_config_from(path: impl AsRef) -> SolverConfig { /// the compiled graph. Every model, including a zero-work model, follows this /// one lifecycle; the public API never exposes a graph, prepared source /// catalog, or phase-builder fallback. +/// +/// `constraints` builds a constraint set for a given solution. It builds the +/// working solution's constraint set and every evaluator replica's, so parallel +/// local-search evaluation can fork replicas for any model. #[allow(clippy::too_many_arguments)] pub fn try_run_solver_with_config_and_search( solution: S, - constraints: C, + constraints: fn(&S) -> C, descriptor: SolutionDescriptor, entity_count_by_descriptor: fn(&S, usize) -> usize, runtime: SolverRuntime, @@ -470,7 +486,7 @@ where #[allow(clippy::too_many_arguments)] fn try_run_solver_with_config_and_search_request( solution: S, - constraints: C, + constraints: fn(&S) -> C, descriptor: SolutionDescriptor, entity_count_by_descriptor: fn(&S, usize) -> usize, runtime: SolverRuntime, @@ -531,7 +547,7 @@ where #[allow(clippy::too_many_arguments)] fn try_run_solver_with_candidate_trace_request( solution: S, - constraints: C, + constraints: fn(&S) -> C, descriptor: SolutionDescriptor, entity_count_by_descriptor: fn(&S, usize) -> usize, runtime: SolverRuntime, @@ -545,18 +561,24 @@ where S: PlanningSolution, S::Score: Score + ParseableScore, C: ConstraintSet, - Runner: Phase, ChannelProgressCallback> + Send + std::fmt::Debug, + Runner: for<'d> Phase< + S, + ScoreDirector>, + ChannelProgressCallback, + > + Send + + std::fmt::Debug, BuildRunner: FnOnce(&SolverConfig, &SolutionDescriptor) -> RuntimeBuildResult, { log_scale(&solution); + let main_constraints = constraints(&solution); let director = ScoreDirector::with_descriptor( solution, - constraints, - descriptor.clone(), + main_constraints, + ReplicaDescriptor::new(&descriptor, constraints), entity_count_by_descriptor, ); - let (termination, time_limit) = build_termination::(&config, default_time_limit_secs); + let (termination, time_limit) = build_director_termination(&config, default_time_limit_secs); let execution_policy = configured_execution_policy::(&config, default_time_limit_secs, time_limit); diff --git a/crates/solverforge-solver/src/run_tests.rs b/crates/solverforge-solver/src/run_tests.rs index 8e44fb1b..9d97cb62 100644 --- a/crates/solverforge-solver/src/run_tests.rs +++ b/crates/solverforge-solver/src/run_tests.rs @@ -161,13 +161,17 @@ fn test_entity_count(_solution: &TestSolution, _descriptor_index: usize) -> usiz fn noop_log_scale(_solution: &TestSolution) {} +fn test_constraints(_solution: &TestSolution) -> ScoreFromSolutionConstraints { + ScoreFromSolutionConstraints +} + #[test] fn zero_variable_run_builds_the_normal_graph_and_search_declaration() { let declarations = Arc::new(AtomicUsize::new(0)); let observed = Arc::clone(&declarations); let result = try_run_solver_with_config_and_search( TestSolution { score: None }, - ScoreFromSolutionConstraints, + test_constraints, test_descriptor(), test_entity_count, SolverRuntime::detached(), diff --git a/crates/solverforge-solver/src/runtime/compiler/executor/runner/local_search.rs b/crates/solverforge-solver/src/runtime/compiler/executor/runner/local_search.rs index c8798514..c59b9408 100644 --- a/crates/solverforge-solver/src/runtime/compiler/executor/runner/local_search.rs +++ b/crates/solverforge-solver/src/runtime/compiler/executor/runner/local_search.rs @@ -46,6 +46,7 @@ where forager: Option, score_tie_break: ScoreTieBreak, random_seed: Option, + evaluator_count: usize, }, VariableNeighborhoodDescent { phase_termination: Option, @@ -78,6 +79,7 @@ where forager, score_tie_break, random_seed, + evaluator_count, } => { let (mut acceptor, mut forager) = build_acceptor_and_forager::( *components, @@ -94,6 +96,7 @@ where &mut acceptor, &mut forager, None, + *evaluator_count, solver_scope, ); }); diff --git a/crates/solverforge-solver/src/runtime/compiler/executor/runner/lowering.rs b/crates/solverforge-solver/src/runtime/compiler/executor/runner/lowering.rs index c07e2908..fcb86c73 100644 --- a/crates/solverforge-solver/src/runtime/compiler/executor/runner/lowering.rs +++ b/crates/solverforge-solver/src/runtime/compiler/executor/runner/lowering.rs @@ -11,6 +11,7 @@ use solverforge_core::domain::PlanningSolution; use solverforge_core::score::{ParseableScore, Score}; use crate::heuristic::selector::nearby_list_change::CrossEntityDistanceMeter; +use crate::phase::localsearch::resolve_evaluator_count; use crate::runtime::compiler::{ CompiledAcceptorForagerSelector, CompiledLocalSearch, CompiledRuntimeExtension, CompiledRuntimePhase, DefaultLocalSearchEligibility, DefaultRuntimeBindings, @@ -87,6 +88,9 @@ where forager: None, score_tie_break: solverforge_config::ScoreTieBreak::Random, random_seed: solver_config.random_seed, + evaluator_count: resolve_evaluator_count( + &solver_config.move_thread_count, + ), }), Some(plan.candidate_trace_plan()), ) @@ -168,6 +172,7 @@ where forager: config.forager, score_tie_break: config.score_tie_break, random_seed: solver_config.random_seed, + evaluator_count: resolve_evaluator_count(&solver_config.move_thread_count), }, declaration, )) diff --git a/crates/solverforge-solver/src/scope/phase.rs b/crates/solverforge-solver/src/scope/phase.rs index d584c67e..62da0e6f 100644 --- a/crates/solverforge-solver/src/scope/phase.rs +++ b/crates/solverforge-solver/src/scope/phase.rs @@ -488,6 +488,12 @@ impl<'t, 'a, S: PlanningSolution, D: Director, BestCb: ProgressCallback> self.solver_scope.record_score_calculation(); } + /// Records a candidate evaluated ahead of the sequential decision point by + /// parallel evaluation and then discarded. + pub fn record_move_speculative(&mut self) { + self.solver_scope.record_move_speculative(); + } + pub fn record_construction_slot_assigned(&mut self) { self.stats.record_construction_slot_assigned(); self.solver_scope diff --git a/crates/solverforge-solver/src/scope/solver/scope_core.rs b/crates/solverforge-solver/src/scope/solver/scope_core.rs index dcba0e4e..f1017784 100644 --- a/crates/solverforge-solver/src/scope/solver/scope_core.rs +++ b/crates/solverforge-solver/src/scope/solver/scope_core.rs @@ -98,6 +98,11 @@ pub struct SolverScope<'t, S: PlanningSolution, D: Director, ProgressCb = ()> last_best_elapsed: Option, best_solution_revision: Option, solution_revision: u64, + /* Monotonic count of working-solution writes: every committed mutation + and every wholesale replacement. Unlike `solution_revision`, it never + resets, so evaluator replicas can detect any write since they synced. + */ + working_solution_generation: u64, construction_frontier: ConstructionFrontier, phase_budget: Option<&'t PhaseBudget>, pub inphase_step_count_limit: Option, @@ -377,6 +382,7 @@ impl<'t, S: PlanningSolution, D: Director> SolverScope<'t, S, D, ()> { last_best_elapsed: None, best_solution_revision: None, solution_revision: 1, + working_solution_generation: 0, construction_frontier, phase_budget: None, inphase_step_count_limit: None, @@ -422,6 +428,7 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> last_best_elapsed: None, best_solution_revision: None, solution_revision: 1, + working_solution_generation: 0, construction_frontier, phase_budget: None, inphase_step_count_limit: None, @@ -586,6 +593,7 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> last_best_elapsed: self.last_best_elapsed, best_solution_revision: self.best_solution_revision, solution_revision: self.solution_revision, + working_solution_generation: self.working_solution_generation, construction_frontier: self.construction_frontier, phase_budget: self.phase_budget, inphase_step_count_limit: self.inphase_step_count_limit, @@ -675,6 +683,7 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> } pub fn replace_working_solution_and_reinitialize(&mut self, solution: S) -> S::Score { + self.working_solution_generation += 1; *self.score_director.working_solution_mut() = solution; self.score_director.reset(); self.current_score = None; @@ -731,6 +740,10 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> self.solution_revision } + pub(crate) fn working_solution_generation(&self) -> u64 { + self.working_solution_generation + } + pub(crate) fn apply_committed_move(&mut self, mov: &M) where M: Move, diff --git a/crates/solverforge-solver/src/scope/solver/scope_progress.rs b/crates/solverforge-solver/src/scope/solver/scope_progress.rs index 08af659a..1df34a93 100644 --- a/crates/solverforge-solver/src/scope/solver/scope_progress.rs +++ b/crates/solverforge-solver/src/scope/solver/scope_progress.rs @@ -520,6 +520,10 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> } } + pub(crate) fn record_move_speculative(&mut self) { + self.stats.record_move_speculative(); + } + pub(crate) fn record_score_calculation(&mut self) { self.stats.record_score_calculation(); if let Some(phase_budget) = self.phase_budget { @@ -606,6 +610,7 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> self.current_score = None; let output = mutate(&mut self.score_director); self.advance_solution_revision(); + self.working_solution_generation += 1; output } } diff --git a/crates/solverforge-solver/src/scope/tests.rs b/crates/solverforge-solver/src/scope/tests.rs index afadb72c..9157d934 100644 --- a/crates/solverforge-solver/src/scope/tests.rs +++ b/crates/solverforge-solver/src/scope/tests.rs @@ -429,3 +429,36 @@ fn solver_scope_forwards_candidate_trace_plan_finalization() { assert_eq!(trace.header.resolved_phase_plan, terminal_plan); assert!(trace.header.resolved_phase_plan_complete); } + +#[test] +fn working_solution_generation_is_monotonic_across_commits_and_replacement() { + let descriptor = SolutionDescriptor::new("TieSolution", TypeId::of::()); + let director = ScoreDirector::simple( + TieSolution { + marker: 0, + score: None, + }, + descriptor, + |_solution, _descriptor_index| 0, + ); + let mut scope = SolverScope::new(director); + scope.start_solving(); + let initial = scope.working_solution_generation(); + + scope.mutate(|score_director| { + score_director.working_solution_mut().marker = 3; + }); + scope.apply_committed_change(|score_director| { + score_director.working_solution_mut().marker = 4; + }); + let after_commits = scope.working_solution_generation(); + assert_eq!(after_commits, initial + 2); + + scope.replace_working_solution_and_reinitialize(TieSolution { + marker: 9, + score: None, + }); + + assert_eq!(scope.solution_revision(), 1); + assert_eq!(scope.working_solution_generation(), after_commits + 1); +} diff --git a/crates/solverforge-solver/src/stats/candidate_trace.rs b/crates/solverforge-solver/src/stats/candidate_trace.rs index d5649d4d..a8543f8b 100644 --- a/crates/solverforge-solver/src/stats/candidate_trace.rs +++ b/crates/solverforge-solver/src/stats/candidate_trace.rs @@ -525,6 +525,9 @@ pub enum CandidateTraceDisposition { Selected, /// The selected candidate was applied to the working solution. Applied, + /// Parallel evaluation pulled this candidate ahead of the sequential + /// decision point; the step ended before the candidate was consumed. + DiscardedSpeculative, } impl CandidateTraceDisposition { @@ -539,6 +542,7 @@ impl CandidateTraceDisposition { Self::ForagerIgnored => 7, Self::Selected => 8, Self::Applied => 9, + Self::DiscardedSpeculative => 10, } } @@ -552,6 +556,7 @@ impl CandidateTraceDisposition { | Self::AcceptorRejected | Self::ForagerIgnored | Self::Applied + | Self::DiscardedSpeculative ) } } diff --git a/crates/solverforge-solver/src/stats/solver.rs b/crates/solverforge-solver/src/stats/solver.rs index 99c9b117..cf49a2d1 100644 --- a/crates/solverforge-solver/src/stats/solver.rs +++ b/crates/solverforge-solver/src/stats/solver.rs @@ -21,6 +21,9 @@ pub struct SolverStats { pub moves_generated: u64, // Total moves evaluated across all phases. pub moves_evaluated: u64, + /// Candidates evaluated ahead of the sequential decision point by parallel + /// evaluation and then discarded. Zero for sequential evaluation. + pub moves_speculative: u64, // Total moves accepted across all phases. pub moves_accepted: u64, // Total moves applied across all phases. @@ -208,6 +211,12 @@ impl SolverStats { self.score_calculations += 1; } + /// Records a candidate that parallel evaluation evaluated ahead of the + /// sequential decision point and then discarded. + pub fn record_move_speculative(&mut self) { + self.moves_speculative += 1; + } + pub fn record_construction_slot_assigned(&mut self) { self.construction_slots_assigned += 1; } @@ -281,6 +290,7 @@ impl SolverStats { step_count: self.step_count, moves_generated: self.moves_generated, moves_evaluated: self.moves_evaluated, + moves_speculative: self.moves_speculative, moves_accepted: self.moves_accepted, moves_applied: self.moves_applied, moves_score_improving: self.moves_score_improving(), diff --git a/crates/solverforge-solver/src/stats/telemetry.rs b/crates/solverforge-solver/src/stats/telemetry.rs index 85874d14..ebbcd31b 100644 --- a/crates/solverforge-solver/src/stats/telemetry.rs +++ b/crates/solverforge-solver/src/stats/telemetry.rs @@ -120,6 +120,9 @@ pub struct SolverTelemetry { /// Exhaust a cursor or query selector sizing separately for logical size. pub moves_generated: u64, pub moves_evaluated: u64, + /// Candidates evaluated ahead of the sequential decision point by parallel + /// evaluation and then discarded. Zero for sequential evaluation. + pub moves_speculative: u64, pub moves_accepted: u64, pub moves_applied: u64, pub moves_score_improving: u64, @@ -159,6 +162,7 @@ impl SolverTelemetry { step_count: 0, moves_generated: 0, moves_evaluated: 0, + moves_speculative: 0, moves_accepted: 0, moves_applied: 0, moves_score_improving: 0, diff --git a/crates/solverforge-solver/src/stats/tests.rs b/crates/solverforge-solver/src/stats/tests.rs index 2a06a840..dc4f0b97 100644 --- a/crates/solverforge-solver/src/stats/tests.rs +++ b/crates/solverforge-solver/src/stats/tests.rs @@ -355,3 +355,18 @@ fn candidate_trace_plan_finalization_is_one_way() { trace.finalize_resolved_phase_plan(plan.clone()); trace.finalize_resolved_phase_plan(plan); } + +#[test] +fn speculative_moves_reach_the_telemetry_snapshot() { + let mut stats = SolverStats::default(); + stats.start(); + stats.record_evaluated_move(Duration::from_millis(1)); + stats.record_move_speculative(); + stats.record_move_speculative(); + + let snapshot = stats.snapshot(); + + assert_eq!(snapshot.moves_evaluated, 1); + assert_eq!(snapshot.moves_speculative, 2); + assert_eq!(SolverTelemetry::new_const().moves_speculative, 0); +} diff --git a/crates/solverforge/tests/parallel_local_search.rs b/crates/solverforge/tests/parallel_local_search.rs new file mode 100644 index 00000000..edb59255 --- /dev/null +++ b/crates/solverforge/tests/parallel_local_search.rs @@ -0,0 +1,90 @@ +use solverforge::{SolverEvent, SolverManager, SolverTelemetry, SolverTerminalReason}; + +#[path = "parallel_local_search/domain/mod.rs"] +mod domain; + +use domain::{Plan, Resource, Task}; + +fn seeded_plan(move_threads: usize) -> Plan { + let resources = (0..6) + .map(|idx| Resource { + id: format!("resource-{idx}"), + }) + .collect(); + let tasks = (0..40) + .map(|idx| Task { + id: format!("task-{idx}"), + weight: (idx * 7 + 3) % 11, + resource_idx: None, + }) + .collect(); + + Plan { + resources, + tasks, + score: None, + move_threads, + } +} + +fn solve(manager: &'static SolverManager, move_threads: usize) -> (Plan, SolverTelemetry) { + let (job_id, mut receiver) = manager + .solve(seeded_plan(move_threads)) + .expect("job should start"); + let mut completed = None; + while let Some(event) = receiver.blocking_recv() { + if let SolverEvent::Completed { metadata, solution } = event { + assert_eq!( + metadata.terminal_reason, + Some(SolverTerminalReason::Completed) + ); + completed = Some((solution, metadata.telemetry)); + break; + } + } + manager.delete(job_id).expect("delete completed job"); + completed.expect("expected a completed solution") +} + +#[test] +fn fixed_work_trajectory_is_identical_across_thread_counts() { + static MANAGER: SolverManager = SolverManager::new(); + + let runs: Vec<_> = [1, 2, 4] + .into_iter() + .map(|move_threads| solve(&MANAGER, move_threads)) + .collect(); + let (baseline, baseline_telemetry) = &runs[0]; + let baseline_assignment: Vec<_> = baseline + .tasks + .iter() + .map(|task| task.resource_idx) + .collect(); + assert!(baseline_telemetry.moves_applied > 0); + + for (solution, telemetry) in &runs[1..] { + let assignment: Vec<_> = solution + .tasks + .iter() + .map(|task| task.resource_idx) + .collect(); + assert_eq!(assignment, baseline_assignment); + assert_eq!(solution.score, baseline.score); + assert_eq!(telemetry.step_count, baseline_telemetry.step_count); + assert_eq!( + telemetry.moves_evaluated, + baseline_telemetry.moves_evaluated + ); + assert_eq!(telemetry.moves_applied, baseline_telemetry.moves_applied); + } +} + +#[test] +fn sequential_resolution_reports_no_speculation() { + static MANAGER: SolverManager = SolverManager::new(); + + for move_threads in [0, 1] { + let (_, telemetry) = solve(&MANAGER, move_threads); + assert_eq!(telemetry.moves_speculative, 0); + } +} diff --git a/crates/solverforge/tests/parallel_local_search/domain/mod.rs b/crates/solverforge/tests/parallel_local_search/domain/mod.rs new file mode 100644 index 00000000..7151628c --- /dev/null +++ b/crates/solverforge/tests/parallel_local_search/domain/mod.rs @@ -0,0 +1,11 @@ +solverforge::planning_model! { + root = "crates/solverforge/tests/parallel_local_search/domain"; + + mod plan; + mod resource; + mod task; + + pub use plan::Plan; + pub use resource::Resource; + pub use task::Task; +} diff --git a/crates/solverforge/tests/parallel_local_search/domain/plan.rs b/crates/solverforge/tests/parallel_local_search/domain/plan.rs new file mode 100644 index 00000000..2c66c43c --- /dev/null +++ b/crates/solverforge/tests/parallel_local_search/domain/plan.rs @@ -0,0 +1,62 @@ +use solverforge::prelude::*; +use solverforge::stream::ConstraintFactory; +use solverforge::{MoveThreadCount, SolverConfig}; + +use super::{Resource, Task}; + +#[planning_solution(constraints = "define_constraints", config = "parallel_config")] +pub struct Plan { + #[problem_fact_collection] + pub resources: Vec, + + #[planning_entity_collection] + pub tasks: Vec, + + #[planning_score] + pub score: Option, + + pub move_threads: usize, +} + +fn define_constraints() -> impl ConstraintSet { + ( + ConstraintFactory::::new() + .for_each(Plan::tasks()) + .filter(|task: &Task| task.resource_idx.is_none()) + .penalize(HardSoftScore::of(1, 0)) + .named("unassigned task"), + ConstraintFactory::::new() + .for_each(Plan::tasks()) + .filter(|task: &Task| task.resource_idx.is_some()) + .penalize(|task: &Task| { + let resource = task.resource_idx.unwrap_or_default() as i64; + HardSoftScore::of(0, (task.weight * (resource + 3)) % 17) + }) + .named("resource cost"), + ) +} + +const PARALLEL_SOLVER_TOML: &str = r#" +random_seed = 7 + +[[phases]] +type = "construction_heuristic" +construction_heuristic_type = "first_fit" + +[[phases]] +type = "local_search" + +[phases.termination] +step_count_limit = 150 +"#; + +fn parallel_config(solution: &Plan, _base: SolverConfig) -> SolverConfig { + let mut config = SolverConfig::from_toml_str(PARALLEL_SOLVER_TOML) + .expect("parallel local-search fixture config must parse"); + config.move_thread_count = if solution.move_threads == 0 { + MoveThreadCount::Auto + } else { + MoveThreadCount::Count(solution.move_threads) + }; + config +} diff --git a/crates/solverforge/tests/parallel_local_search/domain/resource.rs b/crates/solverforge/tests/parallel_local_search/domain/resource.rs new file mode 100644 index 00000000..9885d81b --- /dev/null +++ b/crates/solverforge/tests/parallel_local_search/domain/resource.rs @@ -0,0 +1,7 @@ +use solverforge::prelude::*; + +#[problem_fact] +pub struct Resource { + #[planning_id] + pub id: String, +} diff --git a/crates/solverforge/tests/parallel_local_search/domain/task.rs b/crates/solverforge/tests/parallel_local_search/domain/task.rs new file mode 100644 index 00000000..d7d81e01 --- /dev/null +++ b/crates/solverforge/tests/parallel_local_search/domain/task.rs @@ -0,0 +1,12 @@ +use solverforge::prelude::*; + +#[planning_entity] +pub struct Task { + #[planning_id] + pub id: String, + + pub weight: i64, + + #[planning_variable(value_range_provider = "resources", allows_unassigned = true)] + pub resource_idx: Option, +} diff --git a/docs/architecture.html b/docs/architecture.html index 853a09b6..746b3b90 100644 --- a/docs/architecture.html +++ b/docs/architecture.html @@ -238,6 +238,7 @@ + @@ -505,6 +506,7 @@ + @@ -772,6 +774,7 @@ + @@ -1039,6 +1042,7 @@ PartitionedSearch — rayon work split ExhaustiveSearch · CustomSearchPhase phases ordered by SolverConfig +LocalSearch — rayon window evaluation on replicas @@ -1306,6 +1310,7 @@ +