From 7c7b9335d91772b2204bf760c18bfc8e91051b9c Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 19:40:50 +0200 Subject: [PATCH 1/9] feat(scoring): fork evaluator replicas from ScoreDirector Parallel local-search evaluation needs independent score directors that start from the committed working solution. Director gains a defaulted fork_replica capability, and ScoreDirector implements it when it carries a constraint-set constructor (fn(&S) -> C): the replica clones the working solution, builds its own constraint set for that clone, and initializes it, so incremental state is never cloned or shared. A constructor that receives the solution serves static models and dynamic host-language models alike, since a binding can rebuild its constraint set from the runtime model the solution carries. Directors without a constructor keep returning None. --- crates/solverforge-scoring/WIREFRAME.md | 4 +- .../src/director/score_director/adapters.rs | 5 ++ .../director/score_director/incremental.rs | 35 +++++++++++ .../src/director/tests/score_director.rs | 62 +++++++++++++++++++ .../src/director/traits.rs | 12 ++++ 5 files changed, 117 insertions(+), 1 deletion(-) diff --git a/crates/solverforge-scoring/WIREFRAME.md b/crates/solverforge-scoring/WIREFRAME.md index 8ddc4476..91b39b5d 100644 --- a/crates/solverforge-scoring/WIREFRAME.md +++ b/crates/solverforge-scoring/WIREFRAME.md @@ -376,6 +376,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; default: `None` | ### `DirectorScoreState` @@ -504,7 +505,7 @@ Stock collectors include `count()`, `sum()`, `load_balance()`, **`ScoreDirector`** where `S: PlanningSolution`, `C: ConstraintSet` - Primary incremental scoring director. Zero-erasure. -- Public methods: `new()`, `with_descriptor()`, `simple()` (convenience for +- Public methods: `new()`, `with_descriptor()`, `with_replica_constraints()`, `simple()` (convenience for `ScoreDirector`), `simple_zero()` (test helper with empty descriptor), `working_solution()`, `working_solution_mut()`, `into_working_solution()`, `calculate_score()`, `before_variable_changed()`, @@ -515,6 +516,7 @@ 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_replica_constraints(constraints: fn(&S) -> C)` — records the constraint-set constructor used by `Director::fork_replica`; a replica clones the working solution, builds its constraint set from that clone, and initializes it. Directors without a constructor do not fork. - Implements `Director` ### Constraint Types diff --git a/crates/solverforge-scoring/src/director/score_director/adapters.rs b/crates/solverforge-scoring/src/director/score_director/adapters.rs index cd39bd8c..781135e0 100644 --- a/crates/solverforge-scoring/src/director/score_director/adapters.rs +++ b/crates/solverforge-scoring/src/director/score_director/adapters.rs @@ -16,6 +16,7 @@ where .field("initialized", &self.is_initialized()) .field("cached_score", &self.get_score()) .field("constraint_count", &self.constraint_count()) + .field("forks_replicas", &self.replica_constraints.is_some()) .finish() } } @@ -91,4 +92,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/incremental.rs b/crates/solverforge-scoring/src/director/score_director/incremental.rs index 175d1757..0ee52e56 100644 --- a/crates/solverforge-scoring/src/director/score_director/incremental.rs +++ b/crates/solverforge-scoring/src/director/score_director/incremental.rs @@ -78,6 +78,11 @@ where throughout the solver pipeline. */ pub(super) entity_counter: fn(&S, usize) -> usize, + /* Constraint-set constructor used to build evaluator replicas. + + `None` means this director cannot fork replicas. + */ + pub(super) replica_constraints: Option C>, _phantom: PhantomData S>, } @@ -129,10 +134,40 @@ where initialized: false, solution_descriptor, entity_counter, + replica_constraints: None, _phantom: PhantomData, } } + /* Records the constructor that builds each evaluator replica's constraint + set. The constructor receives the replica's own working solution. + */ + pub fn with_replica_constraints(mut self, constraints: fn(&S) -> C) -> Self { + self.replica_constraints = Some(constraints); + self + } + + /* Builds an initialized evaluator replica from a clone of the working + solution and a constraint set built for that clone. + + 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 constraints = self.replica_constraints?; + let solution = self.working_solution.clone(); + let replica_constraints = constraints(&solution); + let mut replica = Self::with_descriptor( + solution, + replica_constraints, + self.solution_descriptor.clone(), + self.entity_counter, + ); + replica.replica_constraints = Some(constraints); + 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..0df83c5e 100644 --- a/crates/solverforge-scoring/src/director/tests/score_director.rs +++ b/crates/solverforge-scoring/src/director/tests/score_director.rs @@ -307,3 +307,65 @@ 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 unassigned_constraints( + _solution: &TestSolution, +) -> (impl IncrementalConstraint,) { + (make_unassigned_constraint(),) +} + +#[test] +fn fork_replica_requires_replica_constraints() { + 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 mut director = ScoreDirector::new(solution.clone(), unassigned_constraints(&solution)) + .with_replica_constraints(unassigned_constraints); + 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); + + 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 director = ScoreDirector::new(solution.clone(), unassigned_constraints(&solution)) + .with_replica_constraints(unassigned_constraints); + 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 + } } From bd818aac0dfd08732d6476f721c9a2a90ffa1bb6 Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 19:44:20 +0200 Subject: [PATCH 2/9] feat(solver): build constraint sets from a constructor at every solve entry The configured solve entry and the bridge's dynamic runner now take a constraint-set constructor, fn(&S) -> C, in place of a ready-made constraint set. The runtime builds the working solution's constraint set with it and records it on the score director, so every solve can fork evaluator replicas whose constraint sets are built the same way as the main one. Generated solves wrap the model's zero-argument constraints function; bindings rebuild their host constraint set from the solution they receive, which is what lets dynamic models take part in parallel evaluation. The bridge documents that its score-family scope is per thread, so binding constraint sets that build DynamicScore values must enter their family themselves when replicas score on worker threads. --- crates/solverforge-bridge/WIREFRAME.md | 11 ++++++++++- crates/solverforge-bridge/src/runner.rs | 6 +++++- crates/solverforge-bridge/src/score.rs | 5 +++++ crates/solverforge-macros/WIREFRAME.md | 2 +- .../src/planning_solution/runtime/solve.rs | 2 +- .../src/planning_solution_tests.rs | 3 +++ crates/solverforge-solver/WIREFRAME.md | 8 +++++--- .../mod_tests_integration/lifecycle_solutions.rs | 4 +++- crates/solverforge-solver/src/run.rs | 16 +++++++++++----- crates/solverforge-solver/src/run_tests.rs | 6 +++++- 10 files changed, 49 insertions(+), 14 deletions(-) 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-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-solver/WIREFRAME.md b/crates/solverforge-solver/WIREFRAME.md index 889de513..0d461197 100644 --- a/crates/solverforge-solver/WIREFRAME.md +++ b/crates/solverforge-solver/WIREFRAME.md @@ -1731,11 +1731,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/run.rs b/crates/solverforge-solver/src/run.rs index 9e0223e1..4e9014ac 100644 --- a/crates/solverforge-solver/src/run.rs +++ b/crates/solverforge-solver/src/run.rs @@ -419,10 +419,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 +474,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 +535,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, @@ -549,12 +553,14 @@ where BuildRunner: FnOnce(&SolverConfig, &SolutionDescriptor) -> RuntimeBuildResult, { log_scale(&solution); + let main_constraints = constraints(&solution); let director = ScoreDirector::with_descriptor( solution, - constraints, + main_constraints, descriptor.clone(), entity_count_by_descriptor, - ); + ) + .with_replica_constraints(constraints); let (termination, time_limit) = build_termination::(&config, default_time_limit_secs); let execution_policy = 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(), From 845a57134eaec03b200af3685476e7cb9629b720 Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 19:46:51 +0200 Subject: [PATCH 3/9] refactor(solver): split the local-search trial kernel from its accounting evaluate_candidate now composes two pieces: run_trial, which performs the pinned/doable check and the apply-score-undo-restore transaction against a director and nothing else, and account_trial, which records statistics and applies the hard- and score-improvement gates on the thread that owns the step scope. Parallel evaluation needs exactly this boundary: trials run on evaluator replicas while all accounting stays on the coordinator in candidate order. Evaluated-move timing is now passed in as a Duration so the accounting side no longer needs the trial's start instant. Sequential behaviour is unchanged. --- .../src/phase/localsearch/evaluation.rs | 121 +++++++++++++----- .../src/phase/localsearch/phase/candidates.rs | 2 +- .../src/phase/localsearch/vnd/phase.rs | 2 +- 3 files changed, 90 insertions(+), 35 deletions(-) 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/phase/candidates.rs b/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs index 6e722336..bac10329 100644 --- a/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs +++ b/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs @@ -217,7 +217,7 @@ where let accepted = acceptor.is_accepted(&last_step_score, &move_score, move_signature.as_ref()); - record_evaluated_move(step_scope, selector_index, evaluation_started); + record_evaluated_move(step_scope, selector_index, evaluation_started.elapsed()); if accepted { 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); From c5584eacf34576cb0edf83782b5813ddcc01e90c Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 19:48:58 +0200 Subject: [PATCH 4/9] refactor(solver): extract local-search candidate generation and consumption The step's candidate loop is now three explicit stages: record_generated_candidate records a pull (trace, generation telemetry, progress poll) and returns a WindowSlot, run_trial evaluates it, and consume_candidate feeds the outcome through accounting, the acceptor and the forager and releases what the step no longer needs. Parallel evaluation produces several slots before consuming them in order, so these stages must be callable separately. Decisions, checks and their order are unchanged. The evaluated-move duration recorded for a scored candidate now covers the trial itself rather than also including the acceptor call, which is the quantity a trial run on another thread can report. --- crates/solverforge-solver/WIREFRAME.md | 5 +- .../src/phase/localsearch/phase.rs | 1 + .../src/phase/localsearch/phase/candidates.rs | 233 ++------------- .../src/phase/localsearch/phase/decision.rs | 282 ++++++++++++++++++ 4 files changed, 314 insertions(+), 207 deletions(-) create mode 100644 crates/solverforge-solver/src/phase/localsearch/phase/decision.rs diff --git a/crates/solverforge-solver/WIREFRAME.md b/crates/solverforge-solver/WIREFRAME.md index 0d461197..4a5ec46f 100644 --- a/crates/solverforge-solver/WIREFRAME.md +++ b/crates/solverforge-solver/WIREFRAME.md @@ -266,8 +266,11 @@ 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 diff --git a/crates/solverforge-solver/src/phase/localsearch/phase.rs b/crates/solverforge-solver/src/phase/localsearch/phase.rs index c1f6f720..61c7e3fa 100644 --- a/crates/solverforge-solver/src/phase/localsearch/phase.rs +++ b/crates/solverforge-solver/src/phase/localsearch/phase.rs @@ -18,6 +18,7 @@ use crate::scope::{PhaseScope, ProgressCallback, SolverScope}; use crate::stats::{format_duration, whole_units_per_second, CandidateTracePullToken}; mod candidates; +mod decision; mod step; use step::{execute_step, StepOutcome}; diff --git a/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs b/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs index bac10329..6d3a7b57 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,13 @@ 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::{Acceptor, ForagerDecision, LocalSearchForager}; +use crate::phase::localsearch::evaluation::run_trial; +use crate::phase::localsearch::{Acceptor, LocalSearchForager}; use crate::scope::{ProgressCallback, StepScope}; -use crate::stats::{CandidateTraceDisposition, CandidateTracePullToken, CandidateTraceSource}; - -use super::{candidate_selector_label, take_trace_token, StepMoveLabelCounts}; +use crate::stats::{CandidateTraceDisposition, CandidateTracePullToken}; -const PROGRESS_POLL_INTERVAL: u64 = 256; -// Candidate iterations between control polls in the step loop. +use super::decision::{consume_candidate, record_generated_candidate}; +use super::StepMoveLabelCounts; pub(super) struct CandidateLoopState { pub(super) interrupted: bool, @@ -99,42 +93,13 @@ where 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 slot = record_generated_candidate( + &*cursor, + candidate_id, + generation_started, + step_scope, + &mut state, ); - 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(); - } if should_interrupt_before_evaluation(step_scope) || step_scope @@ -142,7 +107,7 @@ where .solver_scope_mut() .should_terminate() { - if let Some(token) = trace_token { + if let Some(token) = slot.trace_token { step_scope .phase_scope_mut() .record_candidate_trace_disposition( @@ -153,171 +118,27 @@ where state.interrupted = true; break; } - state.evaluated_moves += 1; let evaluation_started = std::time::Instant::now(); - let move_score = match evaluate_candidate( - &mov, + let outcome = { + let mov = cursor + .candidate(slot.id) + .expect("discovered candidate id must remain borrowable"); + run_trial(&mov, step_scope.score_director_mut(), &mut journal_pool) + }; + consume_candidate( + cursor, step_scope, + &slot, + outcome, + evaluation_started.elapsed(), 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.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, + acceptor, + forager, + requires_move_signatures, + &mut state, ); - - 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)); - } } 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..c27e5263 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/phase/decision.rs @@ -0,0 +1,282 @@ +//! 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(super) 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; + } + }; + 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, 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)); + } +} + +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); + } +} From 8cece00eea6be25fc63eca66ff29288542de138d Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 19:50:47 +0200 Subject: [PATCH 5/9] feat(solver): report speculative evaluation in telemetry and candidate traces Parallel evaluation evaluates a window of candidates before the coordinator consumes them in order, so a step can end with evaluated candidates that the sequential loop would never have reached. That work must be visible without distorting the logical counters that drive fixed-work limits. SolverStats and SolverTelemetry gain moves_speculative, recorded through PhaseScope::record_move_speculative, and the candidate trace gains the terminal DiscardedSpeculative disposition. Sequential evaluation never produces either, so existing telemetry and traces are unchanged. --- crates/solverforge-solver/WIREFRAME.md | 8 ++++++-- crates/solverforge-solver/src/scope/phase.rs | 6 ++++++ .../src/scope/solver/scope_progress.rs | 4 ++++ .../src/stats/candidate_trace.rs | 5 +++++ crates/solverforge-solver/src/stats/solver.rs | 10 ++++++++++ crates/solverforge-solver/src/stats/telemetry.rs | 4 ++++ crates/solverforge-solver/src/stats/tests.rs | 15 +++++++++++++++ 7 files changed, 50 insertions(+), 2 deletions(-) diff --git a/crates/solverforge-solver/WIREFRAME.md b/crates/solverforge-solver/WIREFRAME.md index 4a5ec46f..3444f44d 100644 --- a/crates/solverforge-solver/WIREFRAME.md +++ b/crates/solverforge-solver/WIREFRAME.md @@ -1601,7 +1601,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 @@ -1650,7 +1653,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 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_progress.rs b/crates/solverforge-solver/src/scope/solver/scope_progress.rs index 08af659a..da1308cb 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 { 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); +} From c01efacf4c17fb81c36e3d7937c7732227d3b130 Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 19:55:56 +0200 Subject: [PATCH 6/9] feat(solver): parallel evaluation windows for acceptor-forager local search Local search can now evaluate a step's candidates on several cores without changing the search. Each pass produces a window of candidates serially on the main director in canonical order, runs their trials on the main director plus persistent evaluator replicas inside one rayon::scope, and consumes the results in candidate order through the unchanged accounting, acceptor and forager. Before consuming each later slot the coordinator repeats the forager quit check and the interruption check the sequential loop makes, so with a window of one the loop is exactly the sequential one. Replicas are forked from the working solution and resynchronised whenever its new working_solution_generation counter shows a write outside the loop's own commits; the counter never resets, unlike solution_revision. The selected move is applied to every replica by borrow before the owned commit, and each replica must reach the committed step score. Windows hold 32 candidates regardless of the evaluator count and never exceed the remaining in-phase move or score-calculation budget, so fixed-work runs are reproducible for every count of two or more and logical counters match sequential evaluation; discarded work is reported as speculative. AGENTS.md records the replica copies as the same clone class as partition copies and the one-scope-per-window threading rule. The runtime still passes one evaluator; move_thread_count is wired next. --- AGENTS.md | 9 +- crates/solverforge-solver/WIREFRAME.md | 32 ++ .../src/phase/localsearch/mod.rs | 1 + .../src/phase/localsearch/parallel/budget.rs | 30 ++ .../src/phase/localsearch/parallel/mod.rs | 17 ++ .../src/phase/localsearch/parallel/pool.rs | 139 +++++++++ .../src/phase/localsearch/parallel/tests.rs | 274 ++++++++++++++++++ .../src/phase/localsearch/parallel/window.rs | 86 ++++++ .../src/phase/localsearch/phase.rs | 5 + .../src/phase/localsearch/phase/candidates.rs | 174 +++++++---- .../src/phase/localsearch/phase/decision.rs | 64 ++++ .../src/phase/localsearch/phase/step.rs | 11 + .../compiler/executor/runner/local_search.rs | 1 + .../src/scope/solver/scope_core.rs | 13 + .../src/scope/solver/scope_progress.rs | 1 + crates/solverforge-solver/src/scope/tests.rs | 33 +++ 16 files changed, 836 insertions(+), 54 deletions(-) create mode 100644 crates/solverforge-solver/src/phase/localsearch/parallel/budget.rs create mode 100644 crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs create mode 100644 crates/solverforge-solver/src/phase/localsearch/parallel/pool.rs create mode 100644 crates/solverforge-solver/src/phase/localsearch/parallel/tests.rs create mode 100644 crates/solverforge-solver/src/phase/localsearch/parallel/window.rs diff --git a/AGENTS.md b/AGENTS.md index 1eab4b9c..3586e1d0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -177,7 +177,11 @@ 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 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. 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 +280,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/crates/solverforge-solver/WIREFRAME.md b/crates/solverforge-solver/WIREFRAME.md index 3444f44d..ceb2126c 100644 --- a/crates/solverforge-solver/WIREFRAME.md +++ b/crates/solverforge-solver/WIREFRAME.md @@ -276,6 +276,11 @@ src/ │ │ ├── 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/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 @@ -1248,6 +1253,33 @@ 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. Contract: + +- One evaluator runs the sequential loop: same candidates, checks, order, and + telemetry. +- The window holds 32 candidates whenever two or more evaluators run, so the + trajectory at fixed work is identical for every evaluator count of two or + more. Selectors that keep no stream state across steps also match the + sequential trajectory; ruin selectors (cross-step RNG streams) and lazily + activated providers follow a different, equally reproducible trajectory. +- `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 diff --git a/crates/solverforge-solver/src/phase/localsearch/mod.rs b/crates/solverforge-solver/src/phase/localsearch/mod.rs index 8731fa3f..db34c473 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; 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/mod.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs new file mode 100644 index 00000000..34d391f3 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs @@ -0,0 +1,17 @@ +/* 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 pool; +mod window; + +#[cfg(test)] +mod tests; + +pub(crate) use budget::remaining_evaluation_budget; +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..f1a07a61 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/pool.rs @@ -0,0 +1,139 @@ +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::scope::{ProgressCallback, SolverScope}; + +/* Candidates per parallel window. Fixed so the trajectory does not depend on +the number of evaluators. +*/ +pub(crate) const PARALLEL_EVALUATION_WINDOW: usize = 32; + +pub(super) type TrialOutputs = Vec<(usize, TrialOutcome, 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, + 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(), + synced_generation: None, + _phantom: PhantomData, + } + } + + pub(crate) fn is_parallel(&self) -> bool { + !self.replicas.is_empty() + } + + 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) + }); + PARALLEL_EVALUATION_WINDOW.min(remaining).max(1) + } + + /* 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..0d344a0a --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/tests.rs @@ -0,0 +1,274 @@ +// 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::{ConstraintSet, IncrementalUniConstraint, 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]) +} + +fn forkable_director( + solution: NQueensSolution, +) -> ScoreDirector> { + let constraints = queen_constraints(&solution); + ScoreDirector::with_descriptor( + solution, + constraints, + create_nqueens_descriptor(), + |s, _| s.queens.len(), + ) + .with_replica_constraints(queen_constraints) +} + +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 mut solver_scope = SolverScope::new(forkable_director(starting_solution())).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 mut solver_scope = SolverScope::new(forkable_director(starting_solution())); + 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); + assert_eq!(pool.window_capacity(None), 32); + assert_eq!(pool.window_capacity(Some(5)), 5); + + 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)); + } +} + +#[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 run_with_replacement = |evaluator_count: usize| { + let mut solver_scope = + SolverScope::new(forkable_director(starting_solution())).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)); +} 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 61c7e3fa..cd6cb12d 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, }; @@ -241,6 +242,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, @@ -263,6 +265,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() { @@ -279,6 +282,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, @@ -345,6 +349,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 6d3a7b57..777447dc 100644 --- a/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs +++ b/crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs @@ -11,11 +11,17 @@ use crate::phase::control::{ should_interrupt_before_candidate, should_interrupt_before_evaluation, CONTROL_POLL_INTERVAL, }; use crate::phase::localsearch::evaluation::run_trial; +use crate::phase::localsearch::parallel::{ + evaluate_window, remaining_evaluation_budget, EvaluatorPool, +}; use crate::phase::localsearch::{Acceptor, LocalSearchForager}; use crate::scope::{ProgressCallback, StepScope}; -use crate::stats::{CandidateTraceDisposition, CandidateTracePullToken}; +use crate::stats::CandidateTracePullToken; -use super::decision::{consume_candidate, record_generated_candidate}; +use super::decision::{ + consume_candidate, discard_window_from, interrupt_window_at, record_generated_candidate, + WindowSlot, +}; use super::StepMoveLabelCounts; pub(super) struct CandidateLoopState { @@ -49,6 +55,7 @@ pub(super) fn evaluate_candidates( acceptor: &mut A, forager: &mut Fo, requires_move_signatures: bool, + evaluators: &mut EvaluatorPool, mut state: CandidateLoopState, ) -> CandidateLoopState where @@ -63,81 +70,142 @@ 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: Vec = Vec::new(); + let mut results = Vec::new(); 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(); + + // 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 control_latched { - state.interrupted = true; + if window.is_empty() { 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 slot = record_generated_candidate( - &*cursor, - candidate_id, - generation_started, - step_scope, - &mut state, - ); - - 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) = slot.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; } - let evaluation_started = std::time::Instant::now(); - let outcome = { + // 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 { let mov = cursor - .candidate(slot.id) + .candidate(window[0].id) .expect("discovered candidate id must remain borrowable"); - run_trial(&mov, step_scope.score_director_mut(), &mut journal_pool) - }; - consume_candidate( - cursor, - step_scope, - &slot, - outcome, - evaluation_started.elapsed(), - last_step_score, - acceptor, - forager, - requires_move_signatures, - &mut state, - ); + 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()))); + } + + // 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; + } + 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; + } + } + 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; + } } state diff --git a/crates/solverforge-solver/src/phase/localsearch/phase/decision.rs b/crates/solverforge-solver/src/phase/localsearch/phase/decision.rs index c27e5263..e1da3dc6 100644 --- a/crates/solverforge-solver/src/phase/localsearch/phase/decision.rs +++ b/crates/solverforge-solver/src/phase/localsearch/phase/decision.rs @@ -265,6 +265,70 @@ pub(super) fn consume_candidate( } } +/// 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, 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/runtime/compiler/executor/runner/local_search.rs b/crates/solverforge-solver/src/runtime/compiler/executor/runner/local_search.rs index c8798514..78d153e7 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 @@ -94,6 +94,7 @@ where &mut acceptor, &mut forager, None, + 1, 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 da1308cb..1df34a93 100644 --- a/crates/solverforge-solver/src/scope/solver/scope_progress.rs +++ b/crates/solverforge-solver/src/scope/solver/scope_progress.rs @@ -610,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); +} From 6f882a63696c44125c491fbb70f6e6fb8c296abb Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 19:57:36 +0200 Subject: [PATCH 7/9] feat(solver): drive acceptor-forager evaluation from move_thread_count SolverConfig::move_thread_count was parsed but only documented, so there was no way to ask for parallel evaluation. The compiled runtime now resolves it into the evaluator count of every acceptor-forager local-search phase: an explicit count of two or more runs that many evaluators, capped by the threads of the Rayon pool the solve runs on; None, zero and one stay sequential. Auto also stays sequential for now, so default behaviour and memory use do not change until benchmarks decide its policy. A config test pins the documented TOML forms. --- crates/solverforge-config/WIREFRAME.md | 13 +++++++++++++ crates/solverforge-config/src/tests/parsing.rs | 17 +++++++++++++++++ crates/solverforge-solver/WIREFRAME.md | 1 + .../src/phase/localsearch/mod.rs | 1 + .../src/phase/localsearch/parallel/config.rs | 12 ++++++++++++ .../src/phase/localsearch/parallel/mod.rs | 2 ++ .../src/phase/localsearch/parallel/tests.rs | 16 ++++++++++++++++ .../compiler/executor/runner/local_search.rs | 4 +++- .../compiler/executor/runner/lowering.rs | 5 +++++ 9 files changed, 70 insertions(+), 1 deletion(-) create mode 100644 crates/solverforge-solver/src/phase/localsearch/parallel/config.rs 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-solver/WIREFRAME.md b/crates/solverforge-solver/WIREFRAME.md index ceb2126c..cf5c4ee8 100644 --- a/crates/solverforge-solver/WIREFRAME.md +++ b/crates/solverforge-solver/WIREFRAME.md @@ -278,6 +278,7 @@ src/ │ │ ├── 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 diff --git a/crates/solverforge-solver/src/phase/localsearch/mod.rs b/crates/solverforge-solver/src/phase/localsearch/mod.rs index db34c473..e798f0ef 100644 --- a/crates/solverforge-solver/src/phase/localsearch/mod.rs +++ b/crates/solverforge-solver/src/phase/localsearch/mod.rs @@ -25,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/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 index 34d391f3..524e9ac3 100644 --- a/crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs @@ -6,6 +6,7 @@ order through the unchanged acceptor and forager. */ mod budget; +mod config; mod pool; mod window; @@ -13,5 +14,6 @@ mod window; 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/tests.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/tests.rs index 0d344a0a..338e404a 100644 --- a/crates/solverforge-solver/src/phase/localsearch/parallel/tests.rs +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/tests.rs @@ -272,3 +272,19 @@ fn replicas_follow_solution_replacement() { 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()) + ); +} 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 78d153e7..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,7 +96,7 @@ where &mut acceptor, &mut forager, None, - 1, + *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, )) From f05fc15c20fe441639de32e99853a61bf1b77971 Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 19:58:40 +0200 Subject: [PATCH 8/9] test(solverforge): parallel local-search parity through the generated runtime Pins the end-to-end contract through SolverManager and macro-generated solving: with move_thread_count 1, 2 and 4, a fixed-work solve (seeded construction plus 150 local-search steps) reaches the same assignment, score, step count, evaluated-move count and applied-move count. The plain-scalar default selectors keep no cross-step stream state, so sequential and parallel runs must match exactly. A second test pins that Auto and an explicit count of one stay sequential and report no speculative work. Speculation under parallel evaluation is asserted in the solver's unit tests instead, because an explicit count is capped by the runner's thread count. --- .../tests/parallel_local_search.rs | 90 +++++++++++++++++++ .../tests/parallel_local_search/domain/mod.rs | 11 +++ .../parallel_local_search/domain/plan.rs | 62 +++++++++++++ .../parallel_local_search/domain/resource.rs | 7 ++ .../parallel_local_search/domain/task.rs | 12 +++ 5 files changed, 182 insertions(+) create mode 100644 crates/solverforge/tests/parallel_local_search.rs create mode 100644 crates/solverforge/tests/parallel_local_search/domain/mod.rs create mode 100644 crates/solverforge/tests/parallel_local_search/domain/plan.rs create mode 100644 crates/solverforge/tests/parallel_local_search/domain/resource.rs create mode 100644 crates/solverforge/tests/parallel_local_search/domain/task.rs 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, +} From e7a296f0433819cadea9522f754ec200aa50950d Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 19:59:17 +0200 Subject: [PATCH 9/9] docs: describe parallel local-search evaluation README's performance section now says how to turn on parallel local-search evaluation for Rust and Python models, that results are reproducible at fixed work, and that the auto default stays sequential for now. The architecture map adds the window evaluation on replicas to the search-phase box, regenerated from its SVG source. --- README.md | 7 +++++++ assets/solverforge-architecture.svg | 1 + docs/architecture.html | 5 +++++ 3 files changed, 13 insertions(+) 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/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 @@ +