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/README.md b/README.md index f176bbee..3f5be0c2 100644 --- a/README.md +++ b/README.md @@ -741,6 +741,13 @@ SolverForge leverages Rust's zero-cost abstractions: - **Cache-friendly**: Contiguous memory layouts for hot paths - **Concrete native execution**: Score directors, deciders, bounders, move carriers, and selector carriers remain monomorphized; documented scorer-agnostic `&dyn Director` callbacks are narrow intentional seams +Local search can evaluate candidates on several cores within one solve: +`move_thread_count = { count = 4 }` in `solver.toml` (Python configs through +solverforge-py use `{"move_thread_count": {"count": 4}}`) runs each step's +trials on up to four evaluators while one coordinator keeps the same +acceptance order. Results are reproducible at fixed work, and the default +(`auto`) currently stays sequential. + Absolute throughput depends on the model, constraints, neighborhood, hardware, and termination policy. Use emitted generated/evaluated throughput for one run and the controlled paired benchmark gate below for regression decisions. diff --git a/assets/solverforge-architecture.svg b/assets/solverforge-architecture.svg index 4e97e178..3ded0f32 100644 --- a/assets/solverforge-architecture.svg +++ b/assets/solverforge-architecture.svg @@ -176,6 +176,7 @@ PartitionedSearch — rayon work split ExhaustiveSearch · CustomSearchPhase phases ordered by SolverConfig +LocalSearch — rayon window evaluation on replicas diff --git a/crates/solverforge-bridge/WIREFRAME.md b/crates/solverforge-bridge/WIREFRAME.md index 229095bb..30a78866 100644 --- a/crates/solverforge-bridge/WIREFRAME.md +++ b/crates/solverforge-bridge/WIREFRAME.md @@ -178,6 +178,11 @@ presentation and host-boundary conversion. `scoped_dynamic_score_family(family, callback)` sets the thread-local family used by `DynamicScore::zero()` and `DynamicScore::from_level_numbers(...)` during the callback. +The family is per thread. The solve thread enters it for the whole solve, but +parallel local-search evaluation scores evaluator replicas on Rayon worker +threads, so a binding's constraint set that builds `DynamicScore` values must +also enter its declared family inside its own `ConstraintSet` methods. + ## Public Functions ### Compiled dynamic runner entrypoints @@ -185,7 +190,7 @@ callback)` sets the thread-local family used by `DynamicScore::zero()` and ```rust pub fn try_run_dynamic_solver_with_config_parts( solution: S, - constraints: C, + constraints: fn(&S) -> C, descriptor: SolutionDescriptor, entity_count_by_descriptor: fn(&S, usize) -> usize, runtime: SolverRuntime, @@ -207,6 +212,10 @@ where This is the binding-oriented entrypoint. It consumes frozen dynamic model parts into the same compiled graph and phase runner used by native models. It accepts no phase builder, selector registration, or alternate construction path. +`constraints` builds a constraint set for a given solution; the runtime calls it +for the working solution and for every evaluator replica, so +`move_thread_count` evaluates dynamic models in parallel. Bindings rebuild +their host constraint set from the solution they receive. The optional qualified provenance changes only the candidate-trace header; it does not select another runner. diff --git a/crates/solverforge-bridge/src/runner.rs b/crates/solverforge-bridge/src/runner.rs index d04a18da..9539af20 100644 --- a/crates/solverforge-bridge/src/runner.rs +++ b/crates/solverforge-bridge/src/runner.rs @@ -63,10 +63,14 @@ where /// /// The one model value is consumed by the canonical compiled runner. There is /// no host-language construction branch or deferred alternate execution path. +/// +/// `constraints` builds a constraint set for a given solution. The runtime +/// calls it for the working solution and for every evaluator replica, so +/// `move_thread_count` evaluates dynamic models in parallel. #[allow(clippy::too_many_arguments)] pub fn try_run_dynamic_solver_with_config_parts( solution: S, - constraints: C, + constraints: fn(&S) -> C, descriptor: SolutionDescriptor, entity_count_by_descriptor: fn(&S, usize) -> usize, runtime: SolverRuntime, diff --git a/crates/solverforge-bridge/src/score.rs b/crates/solverforge-bridge/src/score.rs index 5c94dd80..aadf7e33 100644 --- a/crates/solverforge-bridge/src/score.rs +++ b/crates/solverforge-bridge/src/score.rs @@ -27,6 +27,11 @@ thread_local! { /// This keeps the internal static three-level score contract available to the /// solver while allowing binding-driven solves to present zero and derived /// scores with the host model's declared score family. +/// +/// The family is per thread. The solve thread enters it for the whole solve, +/// but parallel evaluation scores evaluator replicas on Rayon worker threads, +/// so a constraint set that builds `DynamicScore` values must also enter its +/// declared family inside its own `ConstraintSet` methods. pub fn scoped_dynamic_score_family( family: DynamicScoreFamily, callback: impl FnOnce() -> T, diff --git a/crates/solverforge-config/WIREFRAME.md b/crates/solverforge-config/WIREFRAME.md index 44af1d6a..311c42d8 100644 --- a/crates/solverforge-config/WIREFRAME.md +++ b/crates/solverforge-config/WIREFRAME.md @@ -720,6 +720,19 @@ Derives: `Debug, Clone, Default, PartialEq, Eq, Deserialize, Serialize`. | `None` | Single-threaded | | `Count(usize)` | Explicit count | +TOML: `move_thread_count = "auto"`, `"none"`, or `{ count = 4 }`. + +`SolverConfig::move_thread_count` drives parallel evaluation in the compiled +runtime's acceptor-forager local search; `PartitionedSearchConfig::thread_count` +sizes partitioned search separately. + +| `move_thread_count` | Acceptor-forager local search | +|---|---| +| `None` | sequential | +| `Auto` | sequential | +| `Count(0)`, `Count(1)` | sequential | +| `Count(n >= 2)` | `min(n, rayon::current_num_threads())` evaluators: the main director plus that many minus one replicas | + ### `PhaseConfig` Derives: `Debug, Clone, Deserialize, Serialize`. Tagged `#[serde(tag = "type", rename_all = "snake_case")]`. diff --git a/crates/solverforge-config/src/tests/parsing.rs b/crates/solverforge-config/src/tests/parsing.rs index 6d97c498..764d81e5 100644 --- a/crates/solverforge-config/src/tests/parsing.rs +++ b/crates/solverforge-config/src/tests/parsing.rs @@ -248,3 +248,20 @@ fn grouped_scalar_move_selector_parses_and_roundtrips() { assert_eq!(selector.max_moves_per_step, Some(64)); assert!(selector.require_hard_improvement); } + +#[test] +fn move_thread_count_parses_every_documented_form() { + let parse = |value: &str| { + SolverConfig::from_toml_str(&format!("move_thread_count = {value}")) + .unwrap() + .move_thread_count + }; + + assert_eq!(parse("\"auto\""), MoveThreadCount::Auto); + assert_eq!(parse("\"none\""), MoveThreadCount::None); + assert_eq!(parse("{ count = 4 }"), MoveThreadCount::Count(4)); + assert_eq!( + SolverConfig::from_toml_str("").unwrap().move_thread_count, + MoveThreadCount::Auto + ); +} diff --git a/crates/solverforge-macros/WIREFRAME.md b/crates/solverforge-macros/WIREFRAME.md index 9a86c9f9..8f947581 100644 --- a/crates/solverforge-macros/WIREFRAME.md +++ b/crates/solverforge-macros/WIREFRAME.md @@ -254,7 +254,7 @@ semantics should omit `domain = "cvrp"` and declare explicit hook paths instead. - `impl SolvableSolution for T` — delegates to `descriptor()` and `entity_count()` - `impl Solvable for T` (when constraints path specified) — `solve(self, runtime: SolverRuntime, qualified_candidate_trace_provenance: Option)` delegates to `solve_internal()` - `impl Analyzable for T` (when constraints path specified) — `analyze()` creates `ScoreDirector` with canonical shadow support and returns `ScoreAnalysis` -- `fn solve_internal(self, runtime: SolverRuntime, qualified_candidate_trace_provenance: Option)` (when constraints path specified) — loads the configured base `SolverConfig` (or embedded `solver.toml`), applies the optional `config = "..."` callback, and calls hidden `try_run_solver_with_config_and_search`. The generated `__solverforge_search_declaration(config, descriptor)` builds one `RuntimeModel` containing scalar slots plus zero or more owner-specific list slots, delegates scalar hook and scalar-group attachment to the `planning_model!` support impl, sorts those variable slots to the descriptor-backed variable order emitted by the macros, constructs `SearchContext` with `try_new`, and returns the selected custom `Search` declaration or canonical defaults. The compiled runtime owns phase preparation, source binding, construction, local search, and terminal lifecycle; no macro-generated phase sequence exists. +- `fn solve_internal(self, runtime: SolverRuntime, qualified_candidate_trace_provenance: Option)` (when constraints path specified) — loads the configured base `SolverConfig` (or embedded `solver.toml`), applies the optional `config = "..."` callback, and calls hidden `try_run_solver_with_config_and_search`, passing the constraints function as the constraint-set constructor (`fn(&Self) -> _`) used for the working solution and every evaluator replica. The generated `__solverforge_search_declaration(config, descriptor)` builds one `RuntimeModel` containing scalar slots plus zero or more owner-specific list slots, delegates scalar hook and scalar-group attachment to the `planning_model!` support impl, sorts those variable slots to the descriptor-backed variable order emitted by the macros, constructs `SearchContext` with `try_new`, and returns the selected custom `Search` declaration or canonical defaults. The compiled runtime owns phase preparation, source binding, construction, local search, and terminal lifecycle; no macro-generated phase sequence exists. - Public solution source methods for all `#[planning_entity_collection]`, `#[problem_fact_collection]`, and streamable `#[planning_list_element_collection]` fields. Each method is inherent on the solution type, for example `Plan::tasks()`, returns the concrete hidden `SourceExtract &[Task]>` wrapper, and carries hidden `ChangeSource::Descriptor(idx)` for planning entities or `ChangeSource::Static` for facts and list elements. User constraints can call the explicit form `ConstraintFactory::new().for_each(Plan::tasks())` or import the generated `PlanConstraintStreams` trait and call `ConstraintFactory::new().tasks()`. Both paths use the same source metadata and concrete stream types. ### Problem Fact Support Derive diff --git a/crates/solverforge-macros/src/planning_solution/runtime/solve.rs b/crates/solverforge-macros/src/planning_solution/runtime/solve.rs index a5240655..afcceef6 100644 --- a/crates/solverforge-macros/src/planning_solution/runtime/solve.rs +++ b/crates/solverforge-macros/src/planning_solution/runtime/solve.rs @@ -39,7 +39,7 @@ pub(super) fn generate_runtime_solve_internal( #config_expr ::solverforge::__internal::try_run_solver_with_config_and_search( self, - #constraints_fn(), + (|_: &Self| #constraints_fn()) as fn(&Self) -> _, Self::descriptor(), Self::entity_count, runtime, diff --git a/crates/solverforge-macros/src/planning_solution_tests.rs b/crates/solverforge-macros/src/planning_solution_tests.rs index 413f8640..ea83a30b 100644 --- a/crates/solverforge-macros/src/planning_solution_tests.rs +++ b/crates/solverforge-macros/src/planning_solution_tests.rs @@ -65,6 +65,9 @@ fn golden_solution_expansion_loads_solver_config_before_config_callback() { expanded.contains("let config = crate :: config :: for_solution (& self , base_config)") ); assert!(expanded.contains("try_run_solver_with_config_and_search")); + assert!(expanded.contains( + "(| _ : & Self | crate :: constraints :: create_constraints ()) as fn (& Self) -> _" + )); } #[test] diff --git a/crates/solverforge-scoring/WIREFRAME.md b/crates/solverforge-scoring/WIREFRAME.md index 8ddc4476..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 + } } diff --git a/crates/solverforge-solver/WIREFRAME.md b/crates/solverforge-solver/WIREFRAME.md index 889de513..cf5c4ee8 100644 --- a/crates/solverforge-solver/WIREFRAME.md +++ b/crates/solverforge-solver/WIREFRAME.md @@ -266,13 +266,22 @@ src/ │ │ └── grouped_scalar/placer_stream.rs — Concrete candidate and assignment placement stream helpers │ ├── localsearch/ │ │ ├── mod.rs — Acceptor, local-search acceptor/forager, and LocalSearchPhase re-exports -│ │ ├── evaluation.rs — Shared local-search candidate evaluation and hard-delta classification +│ │ ├── evaluation.rs — Trial kernel (run_trial), candidate accounting (account_trial), and hard-delta classification │ │ ├── phase.rs — LocalSearchPhase +│ │ ├── phase/candidates.rs — Candidate loop for one step +│ │ ├── phase/decision.rs — Generated-candidate records and in-order candidate consumption +│ │ ├── phase/step.rs — One step: candidate selection and commit │ │ ├── phase/tests.rs — Tests │ │ ├── forager.rs — LocalSearchForager trait, AcceptedCountForager, FirstAcceptedForager, BestScoreForager, re-exports │ │ ├── forager/any_tests.rs — AnyForager dispatch tests │ │ ├── forager/improving.rs — FirstBestScoreImprovingForager, FirstLastStepScoreImprovingForager │ │ ├── forager/tests.rs — Tests +│ │ ├── parallel/mod.rs — Parallel evaluation module declarations +│ │ ├── parallel/budget.rs — Remaining in-phase evaluation budget for window sizing +│ │ ├── parallel/config.rs — move_thread_count resolution to an evaluator count +│ │ ├── parallel/pool.rs — EvaluatorPool: evaluator replicas, sync, and post-commit advance +│ │ ├── parallel/window.rs — Window evaluation across the main director and replicas +│ │ ├── parallel/tests.rs — Tests │ │ └── acceptor/ │ │ ├── mod.rs — Acceptor module declarations and re-exports │ │ ├── traits.rs — Acceptor trait @@ -1245,6 +1254,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 @@ -1598,7 +1634,10 @@ time, evaluation time, acceptance rate, selector-level telemetry, and exact `Throughput { count, elapsed }` views for generated/evaluated work. Human-facing `moves/s` is derived only at log/console formatting edges. `moves_generated` counts candidates actually yielded by a runtime cursor; it -does not count an unrequested logical tail. Selector `size()` and explicit full +does not count an unrequested logical tail. `moves_speculative` counts +candidates that parallel local-search evaluation evaluated ahead of the +sequential decision point and then discarded; it is zero for sequential +evaluation and never feeds `moves_evaluated` or `score_calculations`. Selector `size()` and explicit full cursor exhaustion cover logical neighborhood size and canonical order. `SolverTelemetry` snapshots expose the same counters plus an optional @@ -1647,7 +1686,8 @@ ordered `CandidateTraceDisposition` transitions. Sources cover construction, local search, VND, generic K-opt, and the specialized list construction/search trial paths. Dispositions distinguish interruption, evaluation, not-doable, hard/score-improvement rejection, acceptor rejection, forager loss, selection, -and application. +application, and `DiscardedSpeculative` (pulled into a parallel evaluation +window, then discarded when the step ended before it was consumed). Logical identities are `CandidateTraceIdentity::Operation` or ordered `Composite`. `CandidateTraceOperationIdentity` carries descriptor, optional @@ -1731,11 +1771,13 @@ partitioned phase starts currently omit a score field. `try_run_solver_with_config_and_search(...) -> RuntimeBuildResult` is the one public configured solve entrypoint used by macro-generated solving. It accepts -the solution/constraints/descriptor, entity counting and logging callbacks, a +the solution, a constraint-set constructor `fn(&S) -> C`, the descriptor, +entity counting and logging callbacks, a `SolverRuntime`, `SolverConfig`, fallback time limit, optional qualified candidate-trace provenance, and a fallible builder for one typed `Search` -declaration. Graph compilation and solve-owned source preparation remain -private; there is no public graph or alternate phase-builder fallback. +declaration. The constructor builds the working solution's constraint set and +every evaluator replica's. Graph compilation and solve-owned source preparation +remain private; there is no public graph or alternate phase-builder fallback. `RuntimeBuildError` is `Declaration { message }`, `Compilation { path, message }`, `Preparation { phase_index, message }`, or diff --git a/crates/solverforge-solver/src/manager/mod_tests_integration/lifecycle_solutions.rs b/crates/solverforge-solver/src/manager/mod_tests_integration/lifecycle_solutions.rs index 8589fb3c..ed623bb0 100644 --- a/crates/solverforge-solver/src/manager/mod_tests_integration/lifecycle_solutions.rs +++ b/crates/solverforge-solver/src/manager/mod_tests_integration/lifecycle_solutions.rs @@ -556,6 +556,8 @@ fn zero_variable_log_scale(solution: &ZeroVariableLifecycleSolution) { solution.gate.wait_for_permit(); } +fn zero_variable_constraints(_solution: &ZeroVariableLifecycleSolution) {} + impl Solvable for ZeroVariableLifecycleSolution { fn solve( self, @@ -564,7 +566,7 @@ impl Solvable for ZeroVariableLifecycleSolution { ) { let result = try_run_solver_with_config_and_search( self, - (), + zero_variable_constraints, zero_variable_solution_descriptor(), zero_variable_entity_count, runtime, diff --git a/crates/solverforge-solver/src/phase/localsearch/evaluation.rs b/crates/solverforge-solver/src/phase/localsearch/evaluation.rs index 9324c79b..ed41a16d 100644 --- a/crates/solverforge-solver/src/phase/localsearch/evaluation.rs +++ b/crates/solverforge-solver/src/phase/localsearch/evaluation.rs @@ -1,4 +1,4 @@ -use std::time::Instant; +use std::time::{Duration, Instant}; use solverforge_core::domain::PlanningSolution; use solverforge_core::score::Score; @@ -17,6 +17,14 @@ pub(crate) enum CandidateEvaluation { RejectedByScoreImprovement(Sc), } +/* Director-only result of one trial transaction. Statistics are recorded by +`account_trial` on the thread that owns the step scope. +*/ +pub(crate) enum TrialOutcome { + NotDoable, + Scored(Sc), +} + #[inline] pub(crate) fn evaluate_candidate( mov: &MoveCandidateRef<'_, S, M>, @@ -32,34 +40,83 @@ where ProgressCb: ProgressCallback, M: Move, { - if move_changes_pinned(mov, step_scope.score_director()) - || !mov.is_doable_with_working_preview(step_scope.score_director_mut()) - { - let move_label = mov.telemetry_label(); - record_evaluated_move(step_scope, selector_index, evaluation_started); - step_scope - .phase_scope_mut() - .record_move_kind_evaluated_unscored(move_label); - if let Some(selector_index) = selector_index { + let outcome = run_trial(mov, step_scope.score_director_mut(), journal_pool); + account_trial( + mov, + step_scope, + reference_score, + selector_index, + evaluation_started.elapsed(), + outcome, + ) +} + +/* Runs the trial transaction on `director`: the pinned/doable check, then +apply, score, undo and restore. Touches nothing but the director and the +undo journal, so it can run on an evaluator replica. +*/ +#[inline] +pub(crate) fn run_trial( + mov: &MoveCandidateRef<'_, S, M>, + director: &mut D, + journal_pool: &mut Vec>, +) -> TrialOutcome +where + S: PlanningSolution, + D: Director, + M: Move, +{ + if move_changes_pinned(mov, director) || !mov.is_doable_with_working_preview(director) { + return TrialOutcome::NotDoable; + } + + let score_state = director.snapshot_score_state(); + let undo = mov.do_move_with_scratch(director, journal_pool); + let move_score = director.calculate_score(); + mov.undo_move_with_scratch(director, undo, journal_pool); + director.restore_score_state(score_state); + TrialOutcome::Scored(move_score) +} + +/* Records the statistics and improvement-gate decisions for one trial +outcome, in the order the sequential evaluation loop has always used. +*/ +#[inline] +pub(crate) fn account_trial( + mov: &MoveCandidateRef<'_, S, M>, + step_scope: &mut StepScope<'_, '_, '_, S, D, ProgressCb>, + reference_score: S::Score, + selector_index: Option, + elapsed: Duration, + outcome: TrialOutcome, +) -> CandidateEvaluation +where + S: PlanningSolution, + D: Director, + ProgressCb: ProgressCallback, + M: Move, +{ + let move_score = match outcome { + TrialOutcome::NotDoable => { + let move_label = mov.telemetry_label(); + record_evaluated_move(step_scope, selector_index, elapsed); step_scope .phase_scope_mut() - .record_selector_move_not_doable(selector_index); - } else { - step_scope.phase_scope_mut().record_move_not_doable(); + .record_move_kind_evaluated_unscored(move_label); + if let Some(selector_index) = selector_index { + step_scope + .phase_scope_mut() + .record_selector_move_not_doable(selector_index); + } else { + step_scope.phase_scope_mut().record_move_not_doable(); + } + step_scope + .phase_scope_mut() + .record_move_kind_not_doable(move_label); + return CandidateEvaluation::NotDoable; } - step_scope - .phase_scope_mut() - .record_move_kind_not_doable(move_label); - return CandidateEvaluation::NotDoable; - } - - let score_state = step_scope.score_director().snapshot_score_state(); - let undo = mov.do_move_with_scratch(step_scope.score_director_mut(), journal_pool); - let move_score = step_scope.score_director_mut().calculate_score(); - mov.undo_move_with_scratch(step_scope.score_director_mut(), undo, journal_pool); - step_scope - .score_director_mut() - .restore_score_state(score_state); + TrialOutcome::Scored(move_score) => move_score, + }; step_scope.phase_scope_mut().record_score_calculation(); @@ -79,7 +136,7 @@ where if mov.requires_hard_improvement() && hard_delta != Some(HardScoreDelta::Improving) { let move_label = mov.telemetry_label(); - record_evaluated_move(step_scope, selector_index, evaluation_started); + record_evaluated_move(step_scope, selector_index, elapsed); step_scope .phase_scope_mut() .record_move_kind_evaluated(move_label, move_score.compare(&reference_score)); @@ -98,7 +155,7 @@ where if mov.requires_score_improvement() && move_score <= reference_score { let move_label = mov.telemetry_label(); - record_evaluated_move(step_scope, selector_index, evaluation_started); + record_evaluated_move(step_scope, selector_index, elapsed); step_scope .phase_scope_mut() .record_move_kind_evaluated(move_label, move_score.compare(&reference_score)); @@ -122,7 +179,7 @@ where pub(crate) fn record_evaluated_move( step_scope: &mut StepScope<'_, '_, '_, S, D, ProgressCb>, selector_index: Option, - evaluation_started: Instant, + elapsed: Duration, ) where S: PlanningSolution, D: Director, @@ -131,10 +188,8 @@ pub(crate) fn record_evaluated_move( if let Some(selector_index) = selector_index { step_scope .phase_scope_mut() - .record_selector_evaluated_move(selector_index, evaluation_started.elapsed()); + .record_selector_evaluated_move(selector_index, elapsed); } else { - step_scope - .phase_scope_mut() - .record_evaluated_move(evaluation_started.elapsed()); + step_scope.phase_scope_mut().record_evaluated_move(elapsed); } } diff --git a/crates/solverforge-solver/src/phase/localsearch/mod.rs b/crates/solverforge-solver/src/phase/localsearch/mod.rs index 8731fa3f..e798f0ef 100644 --- a/crates/solverforge-solver/src/phase/localsearch/mod.rs +++ b/crates/solverforge-solver/src/phase/localsearch/mod.rs @@ -8,6 +8,7 @@ mod acceptor; mod cursor_source; mod evaluation; mod forager; +mod parallel; mod phase; pub(crate) mod vnd; @@ -24,5 +25,6 @@ pub use forager::{ AcceptedCountForager, BestScoreForager, FirstAcceptedForager, FirstBestScoreImprovingForager, FirstLastStepScoreImprovingForager, ForagerDecision, LocalSearchForager, }; +pub(crate) use parallel::resolve_evaluator_count; pub(crate) use phase::solve_local_search_with_resources; pub use phase::LocalSearchPhase; diff --git a/crates/solverforge-solver/src/phase/localsearch/parallel/budget.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/budget.rs new file mode 100644 index 00000000..7e372b78 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/budget.rs @@ -0,0 +1,30 @@ +use solverforge_core::domain::PlanningSolution; +use solverforge_scoring::Director; + +use crate::scope::{ProgressCallback, SolverScope}; + +/* Remaining in-phase evaluation budget: the smaller of the move-count and +score-calculation limits minus the work already counted, or `None` when the +phase sets neither limit. A window never exceeds it, so fixed-work limits stop +at the same candidate as sequential evaluation. +*/ +pub(crate) fn remaining_evaluation_budget( + solver_scope: &SolverScope<'_, S, D, BestCb>, +) -> Option +where + S: PlanningSolution, + D: Director, + BestCb: ProgressCallback, +{ + let stats = solver_scope.stats(); + let moves = solver_scope + .inphase_move_count_limit + .map(|limit| limit.saturating_sub(stats.moves_evaluated)); + let score_calculations = solver_scope + .inphase_score_calc_count_limit + .map(|limit| limit.saturating_sub(stats.score_calculations)); + match (moves, score_calculations) { + (Some(moves), Some(score_calculations)) => Some(moves.min(score_calculations)), + (moves, score_calculations) => moves.or(score_calculations), + } +} diff --git a/crates/solverforge-solver/src/phase/localsearch/parallel/config.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/config.rs new file mode 100644 index 00000000..6de83879 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/config.rs @@ -0,0 +1,12 @@ +use solverforge_config::MoveThreadCount; + +/* Number of evaluators (main director included) for acceptor-forager local +search. `Auto` stays sequential until benchmarks set its policy; an explicit +count is capped by the threads of the pool the solve runs on. +*/ +pub(crate) fn resolve_evaluator_count(count: &MoveThreadCount) -> usize { + match count { + MoveThreadCount::Auto | MoveThreadCount::None => 1, + MoveThreadCount::Count(count) => (*count).clamp(1, rayon::current_num_threads().max(1)), + } +} diff --git a/crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs new file mode 100644 index 00000000..524e9ac3 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/mod.rs @@ -0,0 +1,19 @@ +/* Parallel local-search evaluation. + +The main director and persistent evaluator replicas run trials for one window +of candidates at a time; the coordinator consumes the results in candidate +order through the unchanged acceptor and forager. +*/ + +mod budget; +mod config; +mod pool; +mod window; + +#[cfg(test)] +mod tests; + +pub(crate) use budget::remaining_evaluation_budget; +pub(crate) use config::resolve_evaluator_count; +pub(crate) use pool::EvaluatorPool; +pub(crate) use window::evaluate_window; diff --git a/crates/solverforge-solver/src/phase/localsearch/parallel/pool.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/pool.rs new file mode 100644 index 00000000..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..338e404a --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/tests.rs @@ -0,0 +1,290 @@ +// 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)); +} + +#[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/phase/localsearch/parallel/window.rs b/crates/solverforge-solver/src/phase/localsearch/parallel/window.rs new file mode 100644 index 00000000..8cc1aba6 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/parallel/window.rs @@ -0,0 +1,86 @@ +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; + +use solverforge_core::domain::PlanningSolution; +use solverforge_scoring::Director; + +use crate::heuristic::r#move::Move; +use crate::heuristic::selector::move_selector::{MoveCandidateRef, MoveCandidateUndo}; +use crate::phase::localsearch::evaluation::{run_trial, TrialOutcome}; + +use super::pool::{EvaluatorPool, TrialOutputs}; + +/* Runs the trial for every candidate in the window. The main director takes +its share inline on the coordinator thread; each replica takes its share in a +spawned task. Slots are handed out through one atomic counter, and results land +at their slot index, so which evaluator ran a slot never affects the outcome. +*/ +pub(crate) fn evaluate_window( + candidates: &[MoveCandidateRef<'_, S, M>], + main_director: &mut D, + main_journal_pool: &mut Vec>, + evaluators: &mut EvaluatorPool, + results: &mut Vec, Duration)>>, +) where + S: PlanningSolution, + D: Director, + M: Move, +{ + let next = AtomicUsize::new(0); + let (main_outputs, replicas) = evaluators.parts_mut(); + main_outputs.clear(); + rayon::scope(|scope| { + for replica in replicas.iter_mut() { + let next = &next; + replica.outputs.clear(); + scope.spawn(move |_| { + drain_window( + candidates, + next, + &mut replica.director, + &mut replica.journal_pool, + &mut replica.outputs, + ); + }); + } + drain_window( + candidates, + &next, + main_director, + main_journal_pool, + main_outputs, + ); + }); + results.clear(); + results.resize_with(candidates.len(), || None); + for (index, outcome, elapsed) in main_outputs.drain(..) { + results[index] = Some((outcome, elapsed)); + } + for replica in replicas.iter_mut() { + for (index, outcome, elapsed) in replica.outputs.drain(..) { + results[index] = Some((outcome, elapsed)); + } + } +} + +fn drain_window( + candidates: &[MoveCandidateRef<'_, S, M>], + next: &AtomicUsize, + director: &mut D, + journal_pool: &mut Vec>, + outputs: &mut TrialOutputs, +) where + S: PlanningSolution, + D: Director, + M: Move, +{ + loop { + let index = next.fetch_add(1, Ordering::Relaxed); + let Some(candidate) = candidates.get(index) else { + break; + }; + let started = Instant::now(); + let outcome = run_trial(candidate, director, journal_pool); + outputs.push((index, outcome, started.elapsed())); + } +} diff --git a/crates/solverforge-solver/src/phase/localsearch/phase.rs b/crates/solverforge-solver/src/phase/localsearch/phase.rs index c1f6f720..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, }; @@ -18,6 +19,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}; @@ -240,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, @@ -262,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() { @@ -278,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, @@ -344,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 6e722336..777447dc 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,19 @@ use crate::heuristic::selector::move_selector::{ use crate::phase::control::{ should_interrupt_before_candidate, should_interrupt_before_evaluation, CONTROL_POLL_INTERVAL, }; -use crate::phase::localsearch::evaluation::{ - evaluate_candidate, record_evaluated_move, CandidateEvaluation, +use crate::phase::localsearch::evaluation::run_trial; +use crate::phase::localsearch::parallel::{ + evaluate_window, remaining_evaluation_budget, EvaluatorPool, }; -use crate::phase::localsearch::{Acceptor, ForagerDecision, LocalSearchForager}; +use crate::phase::localsearch::{Acceptor, LocalSearchForager}; use crate::scope::{ProgressCallback, StepScope}; -use crate::stats::{CandidateTraceDisposition, CandidateTracePullToken, CandidateTraceSource}; +use crate::stats::CandidateTracePullToken; -use super::{candidate_selector_label, take_trace_token, StepMoveLabelCounts}; - -const PROGRESS_POLL_INTERVAL: u64 = 256; -// Candidate iterations between control polls in the step loop. +use super::decision::{ + consume_candidate, discard_window_from, interrupt_window_at, record_generated_candidate, + WindowSlot, +}; +use super::StepMoveLabelCounts; pub(super) struct CandidateLoopState { pub(super) interrupted: bool, @@ -55,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 @@ -69,255 +70,143 @@ 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(); - } - if control_latched { - state.interrupted = true; - break; - } - let generation_started = std::time::Instant::now(); - let Some(candidate_id) = - cursor.next_candidate_with_resources(step_scope.score_director_mut(), resources) - else { - break; - }; - let selector_index = cursor.selector_index(candidate_id); - let mov = cursor - .candidate(candidate_id) - .expect("discovered candidate id must remain borrowable"); - let move_label = mov.telemetry_label(); - let selector_label = selector_index.map(|_| candidate_selector_label(&mov)); - let trace_token = step_scope.phase_scope_mut().record_candidate_pull( - CandidateTraceSource::LocalSearch, - selector_index, - candidate_id.index(), - None, - &mov, - ); - let generation_elapsed = generation_started.elapsed(); - state.generated_moves += 1; - step_scope - .phase_scope_mut() - .record_move_kind_generated(move_label); - if let Some(selector_index) = selector_index { - step_scope - .phase_scope_mut() - .record_selector_generated_move_with_label( - selector_index, - selector_label.as_deref().unwrap_or("selector"), - generation_elapsed, - ); - } else { - step_scope - .phase_scope_mut() - .record_generated_move(generation_elapsed); + // Production: serial, canonical order, main director. + let capacity = evaluators.window_capacity(remaining_evaluation_budget( + step_scope.phase_scope().solver_scope(), + )); + window.clear(); + let mut exhausted = false; + while window.len() < capacity { + control_polls += 1; + if !control_latched && control_polls.is_multiple_of(CONTROL_POLL_INTERVAL) { + control_latched = should_interrupt_before_candidate(step_scope) + || step_scope + .phase_scope_mut() + .solver_scope_mut() + .should_terminate(); + } + if control_latched { + state.interrupted = true; + break; + } + + let generation_started = std::time::Instant::now(); + let Some(candidate_id) = + cursor.next_candidate_with_resources(step_scope.score_director_mut(), resources) + else { + exhausted = true; + break; + }; + window.push(record_generated_candidate( + &*cursor, + candidate_id, + generation_started, + step_scope, + &mut state, + )); } - if step_scope.progress_polling_required() - && state.generated_moves.is_multiple_of(PROGRESS_POLL_INTERVAL) - { - step_scope.phase_scope_mut().report_progress_if_due(); + if window.is_empty() { + break; } - if should_interrupt_before_evaluation(step_scope) + // Dispatch: the sequential before-evaluation check for the first slot. + if state.interrupted + || should_interrupt_before_evaluation(step_scope) || step_scope .phase_scope_mut() .solver_scope_mut() .should_terminate() { - if let Some(token) = trace_token { - step_scope - .phase_scope_mut() - .record_candidate_trace_disposition( - token, - CandidateTraceDisposition::InterruptedBeforeEvaluation, - ); - } + interrupt_window_at(cursor, step_scope, &window, 0, false); state.interrupted = true; break; } - state.evaluated_moves += 1; - - let evaluation_started = std::time::Instant::now(); - let move_score = match evaluate_candidate( - &mov, - step_scope, - last_step_score, - selector_index, - evaluation_started, - &mut journal_pool, - ) { - CandidateEvaluation::Scored(score) => { - if let Some(token) = trace_token { - step_scope - .phase_scope_mut() - .record_candidate_trace_disposition( - token, - CandidateTraceDisposition::Evaluated, - ); - } - step_scope - .phase_scope_mut() - .record_move_kind_evaluated(move_label, score.compare(&last_step_score)); - score - } - CandidateEvaluation::NotDoable => { - record_rejected_trace( - step_scope, - trace_token, - CandidateTraceDisposition::NotDoable, - ); - assert!(cursor.release_candidate(candidate_id)); - continue; - } - CandidateEvaluation::RejectedByHardImprovement(_) => { - record_rejected_trace( - step_scope, - trace_token, - CandidateTraceDisposition::RejectedByHardImprovement, - ); - assert!(cursor.release_candidate(candidate_id)); - continue; - } - CandidateEvaluation::RejectedByScoreImprovement(_) => { - record_rejected_trace( - step_scope, - trace_token, - CandidateTraceDisposition::RejectedByScoreImprovement, - ); - assert!(cursor.release_candidate(candidate_id)); - continue; - } - }; - let move_signature = if requires_move_signatures { - match mov.borrowed_tabu_signature() { - Some(borrowed) => Some(borrowed.clone()), - None => Some(mov.tabu_signature(step_scope.score_director())), - } - } else { - None - }; - let accepted = acceptor.is_accepted(&last_step_score, &move_score, move_signature.as_ref()); - - record_evaluated_move(step_scope, selector_index, evaluation_started); - if accepted { - step_scope - .phase_scope_mut() - .record_move_kind_accepted(move_label); - state.accepted_labels.record(move_label); - if let Some(selector_index) = selector_index { - step_scope - .phase_scope_mut() - .record_selector_move_accepted(selector_index); - } else { - step_scope.phase_scope_mut().record_move_accepted(); - } - state.accepted_moves += 1; + // Evaluation. + if evaluators.is_parallel() { + let candidates: Vec<_> = window + .iter() + .map(|slot| { + cursor + .candidate(slot.id) + .expect("window candidate must remain borrowable") + }) + .collect(); + evaluate_window( + &candidates, + step_scope.score_director_mut(), + &mut journal_pool, + evaluators, + &mut results, + ); } else { - if let Some(token) = trace_token { - step_scope - .phase_scope_mut() - .record_candidate_trace_disposition( - token, - CandidateTraceDisposition::AcceptorRejected, - ); - } - if let Some(selector_index) = selector_index { - step_scope - .phase_scope_mut() - .record_selector_move_acceptor_rejected(selector_index); - } else { - step_scope.phase_scope_mut().record_move_acceptor_rejected(); - } - step_scope - .phase_scope_mut() - .record_move_kind_acceptor_rejected( - move_label, - move_score.compare(&last_step_score), - ); + let mov = cursor + .candidate(window[0].id) + .expect("discovered candidate id must remain borrowable"); + let started = std::time::Instant::now(); + let outcome = run_trial(&mov, step_scope.score_director_mut(), &mut journal_pool); + results.clear(); + results.push(Some((outcome, started.elapsed()))); } - trace!( - event = "step", - step = step_scope.step_index(), - move_index = candidate_id.index(), - score = %move_score, - accepted = accepted, - ); - - if accepted { - match forager.add_move_index(candidate_id, move_score) { - ForagerDecision::Keep => { - if let Some(token) = trace_token { - state.accepted_trace_tokens.push((candidate_id, token)); - } - } - ForagerDecision::Release => { - if let Some(token) = trace_token { - step_scope - .phase_scope_mut() - .record_candidate_trace_disposition( - token, - CandidateTraceDisposition::ForagerIgnored, - ); - } - assert!(cursor.release_candidate(candidate_id)); + // Consumption in candidate order. + for index in 0..window.len() { + if index > 0 { + if forager.is_quit_early() { + discard_window_from(cursor, step_scope, &window, index, true); + break; } - ForagerDecision::Replace(replaced_id) => { - if let Some(token) = - take_trace_token(&mut state.accepted_trace_tokens, replaced_id) - { - step_scope - .phase_scope_mut() - .record_candidate_trace_disposition( - token, - CandidateTraceDisposition::ForagerIgnored, - ); - } - if let Some(token) = trace_token { - state.accepted_trace_tokens.push((candidate_id, token)); - } - assert!(cursor.release_candidate(replaced_id)); + if should_interrupt_before_evaluation(step_scope) + || step_scope + .phase_scope_mut() + .solver_scope_mut() + .should_terminate() + { + interrupt_window_at(cursor, step_scope, &window, index, true); + state.interrupted = true; + break; } } - } else { - assert!(cursor.release_candidate(candidate_id)); + let (outcome, elapsed) = results[index] + .take() + .expect("every window slot has a trial result"); + consume_candidate( + cursor, + step_scope, + &window[index], + outcome, + elapsed, + last_step_score, + acceptor, + forager, + requires_move_signatures, + &mut state, + ); + } + if state.interrupted || exhausted { + break; } } 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..e1da3dc6 --- /dev/null +++ b/crates/solverforge-solver/src/phase/localsearch/phase/decision.rs @@ -0,0 +1,346 @@ +//! 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)); + } +} + +/// Ends the window at `index` because of an interruption: slot `index` gets +/// the sequential loop's `InterruptedBeforeEvaluation` record and stays live, +/// exactly as the sequential loop leaves it; later slots are discarded. +/// `evaluated` tells whether the window's trials already ran. +pub(super) fn interrupt_window_at( + cursor: &mut C, + step_scope: &mut StepScope<'_, '_, '_, S, D, BestCb>, + window: &[WindowSlot], + index: usize, + evaluated: bool, +) where + S: PlanningSolution, + D: Director, + BestCb: ProgressCallback, + M: Move, + C: ResourceMoveCursor, +{ + let slot = &window[index]; + if let Some(token) = slot.trace_token { + step_scope + .phase_scope_mut() + .record_candidate_trace_disposition( + token, + CandidateTraceDisposition::InterruptedBeforeEvaluation, + ); + } + if evaluated { + step_scope.phase_scope_mut().record_move_speculative(); + } + discard_window_from(cursor, step_scope, window, index + 1, evaluated); +} + +/// Discards slots `from..` of a window that the step will not consume: +/// records `DiscardedSpeculative`, releases them, and counts them as +/// speculative work when their trials already ran. +pub(super) fn discard_window_from( + cursor: &mut C, + step_scope: &mut StepScope<'_, '_, '_, S, D, BestCb>, + window: &[WindowSlot], + from: usize, + evaluated: bool, +) where + S: PlanningSolution, + D: Director, + BestCb: ProgressCallback, + M: Move, + C: ResourceMoveCursor, +{ + for slot in window.iter().skip(from) { + if let Some(token) = slot.trace_token { + step_scope + .phase_scope_mut() + .record_candidate_trace_disposition( + token, + CandidateTraceDisposition::DiscardedSpeculative, + ); + } + if evaluated { + step_scope.phase_scope_mut().record_move_speculative(); + } + assert!(cursor.release_candidate(slot.id)); + } +} + +fn record_rejected_trace( + step_scope: &mut StepScope<'_, '_, '_, S, D, BestCb>, + trace_token: Option, + disposition: CandidateTraceDisposition, +) where + S: PlanningSolution, + D: Director, + BestCb: ProgressCallback, +{ + if let Some(token) = trace_token { + let phase_scope = step_scope.phase_scope_mut(); + phase_scope.record_candidate_trace_disposition(token, CandidateTraceDisposition::Evaluated); + phase_scope.record_candidate_trace_disposition(token, disposition); + } +} diff --git a/crates/solverforge-solver/src/phase/localsearch/phase/step.rs b/crates/solverforge-solver/src/phase/localsearch/phase/step.rs index 2d579ca2..08d1e555 100644 --- a/crates/solverforge-solver/src/phase/localsearch/phase/step.rs +++ b/crates/solverforge-solver/src/phase/localsearch/phase/step.rs @@ -13,6 +13,7 @@ use crate::phase::control::{ settle_search_interrupt, should_interrupt_after_step, should_interrupt_before_candidate, StepInterrupt, }; +use crate::phase::localsearch::parallel::EvaluatorPool; use crate::phase::localsearch::{Acceptor, LocalSearchForager, MoveCursorSource}; use crate::scope::{PendingControl, PhaseScope, ProgressCallback, StepScope}; use crate::stats::{AppliedMoveTelemetry, CandidateTraceDisposition}; @@ -34,6 +35,7 @@ pub(super) fn execute_step( forager: &mut Fo, phase_scope: &mut PhaseScope<'_, '_, S, D, BestCb>, last_step_score: &mut S::Score, + evaluators: &mut EvaluatorPool, ) -> StepOutcome where S: PlanningSolution, @@ -45,6 +47,7 @@ where BestCb: ProgressCallback, { let mut step_scope = StepScope::new(phase_scope); + evaluators.sync(step_scope.phase_scope().solver_scope()); /* Reset forager and acceptor for this step. Pass best and last-step scores so foragers that implement @@ -91,6 +94,7 @@ where acceptor, forager, requires_move_signatures, + evaluators, CandidateLoopState::new(interrupted), ); @@ -139,9 +143,16 @@ where }; } let previous_score = *last_step_score; + evaluators.advance(&selected_move, selected_score); step_scope.apply_committed_change(|score_director| { cursor.apply_owned_candidate(selected_index, score_director); }); + evaluators.mark_synced( + step_scope + .phase_scope() + .solver_scope() + .working_solution_generation(), + ); if let Some(token) = selected_trace_token { step_scope .phase_scope_mut() diff --git a/crates/solverforge-solver/src/phase/localsearch/vnd/phase.rs b/crates/solverforge-solver/src/phase/localsearch/vnd/phase.rs index 38d98c48..4a45ab0b 100644 --- a/crates/solverforge-solver/src/phase/localsearch/vnd/phase.rs +++ b/crates/solverforge-solver/src/phase/localsearch/vnd/phase.rs @@ -338,7 +338,7 @@ where } }; - record_evaluated_move(step_scope, selector_index, evaluation_started); + record_evaluated_move(step_scope, selector_index, evaluation_started.elapsed()); progress.record_evaluated(); progress.maybe_report(step_scope); diff --git a/crates/solverforge-solver/src/run.rs b/crates/solverforge-solver/src/run.rs index 9e0223e1..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(), diff --git a/crates/solverforge-solver/src/runtime/compiler/executor/runner/local_search.rs b/crates/solverforge-solver/src/runtime/compiler/executor/runner/local_search.rs index c8798514..c59b9408 100644 --- a/crates/solverforge-solver/src/runtime/compiler/executor/runner/local_search.rs +++ b/crates/solverforge-solver/src/runtime/compiler/executor/runner/local_search.rs @@ -46,6 +46,7 @@ where forager: Option, score_tie_break: ScoreTieBreak, random_seed: Option, + evaluator_count: usize, }, VariableNeighborhoodDescent { phase_termination: Option, @@ -78,6 +79,7 @@ where forager, score_tie_break, random_seed, + evaluator_count, } => { let (mut acceptor, mut forager) = build_acceptor_and_forager::( *components, @@ -94,6 +96,7 @@ where &mut acceptor, &mut forager, None, + *evaluator_count, solver_scope, ); }); diff --git a/crates/solverforge-solver/src/runtime/compiler/executor/runner/lowering.rs b/crates/solverforge-solver/src/runtime/compiler/executor/runner/lowering.rs index c07e2908..fcb86c73 100644 --- a/crates/solverforge-solver/src/runtime/compiler/executor/runner/lowering.rs +++ b/crates/solverforge-solver/src/runtime/compiler/executor/runner/lowering.rs @@ -11,6 +11,7 @@ use solverforge_core::domain::PlanningSolution; use solverforge_core::score::{ParseableScore, Score}; use crate::heuristic::selector::nearby_list_change::CrossEntityDistanceMeter; +use crate::phase::localsearch::resolve_evaluator_count; use crate::runtime::compiler::{ CompiledAcceptorForagerSelector, CompiledLocalSearch, CompiledRuntimeExtension, CompiledRuntimePhase, DefaultLocalSearchEligibility, DefaultRuntimeBindings, @@ -87,6 +88,9 @@ where forager: None, score_tie_break: solverforge_config::ScoreTieBreak::Random, random_seed: solver_config.random_seed, + evaluator_count: resolve_evaluator_count( + &solver_config.move_thread_count, + ), }), Some(plan.candidate_trace_plan()), ) @@ -168,6 +172,7 @@ where forager: config.forager, score_tie_break: config.score_tie_break, random_seed: solver_config.random_seed, + evaluator_count: resolve_evaluator_count(&solver_config.move_thread_count), }, declaration, )) diff --git a/crates/solverforge-solver/src/scope/phase.rs b/crates/solverforge-solver/src/scope/phase.rs index d584c67e..62da0e6f 100644 --- a/crates/solverforge-solver/src/scope/phase.rs +++ b/crates/solverforge-solver/src/scope/phase.rs @@ -488,6 +488,12 @@ impl<'t, 'a, S: PlanningSolution, D: Director, BestCb: ProgressCallback> self.solver_scope.record_score_calculation(); } + /// Records a candidate evaluated ahead of the sequential decision point by + /// parallel evaluation and then discarded. + pub fn record_move_speculative(&mut self) { + self.solver_scope.record_move_speculative(); + } + pub fn record_construction_slot_assigned(&mut self) { self.stats.record_construction_slot_assigned(); self.solver_scope diff --git a/crates/solverforge-solver/src/scope/solver/scope_core.rs b/crates/solverforge-solver/src/scope/solver/scope_core.rs index dcba0e4e..f1017784 100644 --- a/crates/solverforge-solver/src/scope/solver/scope_core.rs +++ b/crates/solverforge-solver/src/scope/solver/scope_core.rs @@ -98,6 +98,11 @@ pub struct SolverScope<'t, S: PlanningSolution, D: Director, ProgressCb = ()> last_best_elapsed: Option, best_solution_revision: Option, solution_revision: u64, + /* Monotonic count of working-solution writes: every committed mutation + and every wholesale replacement. Unlike `solution_revision`, it never + resets, so evaluator replicas can detect any write since they synced. + */ + working_solution_generation: u64, construction_frontier: ConstructionFrontier, phase_budget: Option<&'t PhaseBudget>, pub inphase_step_count_limit: Option, @@ -377,6 +382,7 @@ impl<'t, S: PlanningSolution, D: Director> SolverScope<'t, S, D, ()> { last_best_elapsed: None, best_solution_revision: None, solution_revision: 1, + working_solution_generation: 0, construction_frontier, phase_budget: None, inphase_step_count_limit: None, @@ -422,6 +428,7 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> last_best_elapsed: None, best_solution_revision: None, solution_revision: 1, + working_solution_generation: 0, construction_frontier, phase_budget: None, inphase_step_count_limit: None, @@ -586,6 +593,7 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> last_best_elapsed: self.last_best_elapsed, best_solution_revision: self.best_solution_revision, solution_revision: self.solution_revision, + working_solution_generation: self.working_solution_generation, construction_frontier: self.construction_frontier, phase_budget: self.phase_budget, inphase_step_count_limit: self.inphase_step_count_limit, @@ -675,6 +683,7 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> } pub fn replace_working_solution_and_reinitialize(&mut self, solution: S) -> S::Score { + self.working_solution_generation += 1; *self.score_director.working_solution_mut() = solution; self.score_director.reset(); self.current_score = None; @@ -731,6 +740,10 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> self.solution_revision } + pub(crate) fn working_solution_generation(&self) -> u64 { + self.working_solution_generation + } + pub(crate) fn apply_committed_move(&mut self, mov: &M) where M: Move, diff --git a/crates/solverforge-solver/src/scope/solver/scope_progress.rs b/crates/solverforge-solver/src/scope/solver/scope_progress.rs index 08af659a..1df34a93 100644 --- a/crates/solverforge-solver/src/scope/solver/scope_progress.rs +++ b/crates/solverforge-solver/src/scope/solver/scope_progress.rs @@ -520,6 +520,10 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> } } + pub(crate) fn record_move_speculative(&mut self) { + self.stats.record_move_speculative(); + } + pub(crate) fn record_score_calculation(&mut self) { self.stats.record_score_calculation(); if let Some(phase_budget) = self.phase_budget { @@ -606,6 +610,7 @@ impl<'t, S: PlanningSolution, D: Director, ProgressCb: ProgressCallback> self.current_score = None; let output = mutate(&mut self.score_director); self.advance_solution_revision(); + self.working_solution_generation += 1; output } } diff --git a/crates/solverforge-solver/src/scope/tests.rs b/crates/solverforge-solver/src/scope/tests.rs index afadb72c..9157d934 100644 --- a/crates/solverforge-solver/src/scope/tests.rs +++ b/crates/solverforge-solver/src/scope/tests.rs @@ -429,3 +429,36 @@ fn solver_scope_forwards_candidate_trace_plan_finalization() { assert_eq!(trace.header.resolved_phase_plan, terminal_plan); assert!(trace.header.resolved_phase_plan_complete); } + +#[test] +fn working_solution_generation_is_monotonic_across_commits_and_replacement() { + let descriptor = SolutionDescriptor::new("TieSolution", TypeId::of::()); + let director = ScoreDirector::simple( + TieSolution { + marker: 0, + score: None, + }, + descriptor, + |_solution, _descriptor_index| 0, + ); + let mut scope = SolverScope::new(director); + scope.start_solving(); + let initial = scope.working_solution_generation(); + + scope.mutate(|score_director| { + score_director.working_solution_mut().marker = 3; + }); + scope.apply_committed_change(|score_director| { + score_director.working_solution_mut().marker = 4; + }); + let after_commits = scope.working_solution_generation(); + assert_eq!(after_commits, initial + 2); + + scope.replace_working_solution_and_reinitialize(TieSolution { + marker: 9, + score: None, + }); + + assert_eq!(scope.solution_revision(), 1); + assert_eq!(scope.working_solution_generation(), after_commits + 1); +} diff --git a/crates/solverforge-solver/src/stats/candidate_trace.rs b/crates/solverforge-solver/src/stats/candidate_trace.rs index d5649d4d..a8543f8b 100644 --- a/crates/solverforge-solver/src/stats/candidate_trace.rs +++ b/crates/solverforge-solver/src/stats/candidate_trace.rs @@ -525,6 +525,9 @@ pub enum CandidateTraceDisposition { Selected, /// The selected candidate was applied to the working solution. Applied, + /// Parallel evaluation pulled this candidate ahead of the sequential + /// decision point; the step ended before the candidate was consumed. + DiscardedSpeculative, } impl CandidateTraceDisposition { @@ -539,6 +542,7 @@ impl CandidateTraceDisposition { Self::ForagerIgnored => 7, Self::Selected => 8, Self::Applied => 9, + Self::DiscardedSpeculative => 10, } } @@ -552,6 +556,7 @@ impl CandidateTraceDisposition { | Self::AcceptorRejected | Self::ForagerIgnored | Self::Applied + | Self::DiscardedSpeculative ) } } diff --git a/crates/solverforge-solver/src/stats/solver.rs b/crates/solverforge-solver/src/stats/solver.rs index 99c9b117..cf49a2d1 100644 --- a/crates/solverforge-solver/src/stats/solver.rs +++ b/crates/solverforge-solver/src/stats/solver.rs @@ -21,6 +21,9 @@ pub struct SolverStats { pub moves_generated: u64, // Total moves evaluated across all phases. pub moves_evaluated: u64, + /// Candidates evaluated ahead of the sequential decision point by parallel + /// evaluation and then discarded. Zero for sequential evaluation. + pub moves_speculative: u64, // Total moves accepted across all phases. pub moves_accepted: u64, // Total moves applied across all phases. @@ -208,6 +211,12 @@ impl SolverStats { self.score_calculations += 1; } + /// Records a candidate that parallel evaluation evaluated ahead of the + /// sequential decision point and then discarded. + pub fn record_move_speculative(&mut self) { + self.moves_speculative += 1; + } + pub fn record_construction_slot_assigned(&mut self) { self.construction_slots_assigned += 1; } @@ -281,6 +290,7 @@ impl SolverStats { step_count: self.step_count, moves_generated: self.moves_generated, moves_evaluated: self.moves_evaluated, + moves_speculative: self.moves_speculative, moves_accepted: self.moves_accepted, moves_applied: self.moves_applied, moves_score_improving: self.moves_score_improving(), diff --git a/crates/solverforge-solver/src/stats/telemetry.rs b/crates/solverforge-solver/src/stats/telemetry.rs index 85874d14..ebbcd31b 100644 --- a/crates/solverforge-solver/src/stats/telemetry.rs +++ b/crates/solverforge-solver/src/stats/telemetry.rs @@ -120,6 +120,9 @@ pub struct SolverTelemetry { /// Exhaust a cursor or query selector sizing separately for logical size. pub moves_generated: u64, pub moves_evaluated: u64, + /// Candidates evaluated ahead of the sequential decision point by parallel + /// evaluation and then discarded. Zero for sequential evaluation. + pub moves_speculative: u64, pub moves_accepted: u64, pub moves_applied: u64, pub moves_score_improving: u64, @@ -159,6 +162,7 @@ impl SolverTelemetry { step_count: 0, moves_generated: 0, moves_evaluated: 0, + moves_speculative: 0, moves_accepted: 0, moves_applied: 0, moves_score_improving: 0, diff --git a/crates/solverforge-solver/src/stats/tests.rs b/crates/solverforge-solver/src/stats/tests.rs index 2a06a840..dc4f0b97 100644 --- a/crates/solverforge-solver/src/stats/tests.rs +++ b/crates/solverforge-solver/src/stats/tests.rs @@ -355,3 +355,18 @@ fn candidate_trace_plan_finalization_is_one_way() { trace.finalize_resolved_phase_plan(plan.clone()); trace.finalize_resolved_phase_plan(plan); } + +#[test] +fn speculative_moves_reach_the_telemetry_snapshot() { + let mut stats = SolverStats::default(); + stats.start(); + stats.record_evaluated_move(Duration::from_millis(1)); + stats.record_move_speculative(); + stats.record_move_speculative(); + + let snapshot = stats.snapshot(); + + assert_eq!(snapshot.moves_evaluated, 1); + assert_eq!(snapshot.moves_speculative, 2); + assert_eq!(SolverTelemetry::new_const().moves_speculative, 0); +} diff --git a/crates/solverforge/tests/parallel_local_search.rs b/crates/solverforge/tests/parallel_local_search.rs new file mode 100644 index 00000000..edb59255 --- /dev/null +++ b/crates/solverforge/tests/parallel_local_search.rs @@ -0,0 +1,90 @@ +use solverforge::{SolverEvent, SolverManager, SolverTelemetry, SolverTerminalReason}; + +#[path = "parallel_local_search/domain/mod.rs"] +mod domain; + +use domain::{Plan, Resource, Task}; + +fn seeded_plan(move_threads: usize) -> Plan { + let resources = (0..6) + .map(|idx| Resource { + id: format!("resource-{idx}"), + }) + .collect(); + let tasks = (0..40) + .map(|idx| Task { + id: format!("task-{idx}"), + weight: (idx * 7 + 3) % 11, + resource_idx: None, + }) + .collect(); + + Plan { + resources, + tasks, + score: None, + move_threads, + } +} + +fn solve(manager: &'static SolverManager, move_threads: usize) -> (Plan, SolverTelemetry) { + let (job_id, mut receiver) = manager + .solve(seeded_plan(move_threads)) + .expect("job should start"); + let mut completed = None; + while let Some(event) = receiver.blocking_recv() { + if let SolverEvent::Completed { metadata, solution } = event { + assert_eq!( + metadata.terminal_reason, + Some(SolverTerminalReason::Completed) + ); + completed = Some((solution, metadata.telemetry)); + break; + } + } + manager.delete(job_id).expect("delete completed job"); + completed.expect("expected a completed solution") +} + +#[test] +fn fixed_work_trajectory_is_identical_across_thread_counts() { + static MANAGER: SolverManager = SolverManager::new(); + + let runs: Vec<_> = [1, 2, 4] + .into_iter() + .map(|move_threads| solve(&MANAGER, move_threads)) + .collect(); + let (baseline, baseline_telemetry) = &runs[0]; + let baseline_assignment: Vec<_> = baseline + .tasks + .iter() + .map(|task| task.resource_idx) + .collect(); + assert!(baseline_telemetry.moves_applied > 0); + + for (solution, telemetry) in &runs[1..] { + let assignment: Vec<_> = solution + .tasks + .iter() + .map(|task| task.resource_idx) + .collect(); + assert_eq!(assignment, baseline_assignment); + assert_eq!(solution.score, baseline.score); + assert_eq!(telemetry.step_count, baseline_telemetry.step_count); + assert_eq!( + telemetry.moves_evaluated, + baseline_telemetry.moves_evaluated + ); + assert_eq!(telemetry.moves_applied, baseline_telemetry.moves_applied); + } +} + +#[test] +fn sequential_resolution_reports_no_speculation() { + static MANAGER: SolverManager = SolverManager::new(); + + for move_threads in [0, 1] { + let (_, telemetry) = solve(&MANAGER, move_threads); + assert_eq!(telemetry.moves_speculative, 0); + } +} diff --git a/crates/solverforge/tests/parallel_local_search/domain/mod.rs b/crates/solverforge/tests/parallel_local_search/domain/mod.rs new file mode 100644 index 00000000..7151628c --- /dev/null +++ b/crates/solverforge/tests/parallel_local_search/domain/mod.rs @@ -0,0 +1,11 @@ +solverforge::planning_model! { + root = "crates/solverforge/tests/parallel_local_search/domain"; + + mod plan; + mod resource; + mod task; + + pub use plan::Plan; + pub use resource::Resource; + pub use task::Task; +} diff --git a/crates/solverforge/tests/parallel_local_search/domain/plan.rs b/crates/solverforge/tests/parallel_local_search/domain/plan.rs new file mode 100644 index 00000000..2c66c43c --- /dev/null +++ b/crates/solverforge/tests/parallel_local_search/domain/plan.rs @@ -0,0 +1,62 @@ +use solverforge::prelude::*; +use solverforge::stream::ConstraintFactory; +use solverforge::{MoveThreadCount, SolverConfig}; + +use super::{Resource, Task}; + +#[planning_solution(constraints = "define_constraints", config = "parallel_config")] +pub struct Plan { + #[problem_fact_collection] + pub resources: Vec, + + #[planning_entity_collection] + pub tasks: Vec, + + #[planning_score] + pub score: Option, + + pub move_threads: usize, +} + +fn define_constraints() -> impl ConstraintSet { + ( + ConstraintFactory::::new() + .for_each(Plan::tasks()) + .filter(|task: &Task| task.resource_idx.is_none()) + .penalize(HardSoftScore::of(1, 0)) + .named("unassigned task"), + ConstraintFactory::::new() + .for_each(Plan::tasks()) + .filter(|task: &Task| task.resource_idx.is_some()) + .penalize(|task: &Task| { + let resource = task.resource_idx.unwrap_or_default() as i64; + HardSoftScore::of(0, (task.weight * (resource + 3)) % 17) + }) + .named("resource cost"), + ) +} + +const PARALLEL_SOLVER_TOML: &str = r#" +random_seed = 7 + +[[phases]] +type = "construction_heuristic" +construction_heuristic_type = "first_fit" + +[[phases]] +type = "local_search" + +[phases.termination] +step_count_limit = 150 +"#; + +fn parallel_config(solution: &Plan, _base: SolverConfig) -> SolverConfig { + let mut config = SolverConfig::from_toml_str(PARALLEL_SOLVER_TOML) + .expect("parallel local-search fixture config must parse"); + config.move_thread_count = if solution.move_threads == 0 { + MoveThreadCount::Auto + } else { + MoveThreadCount::Count(solution.move_threads) + }; + config +} diff --git a/crates/solverforge/tests/parallel_local_search/domain/resource.rs b/crates/solverforge/tests/parallel_local_search/domain/resource.rs new file mode 100644 index 00000000..9885d81b --- /dev/null +++ b/crates/solverforge/tests/parallel_local_search/domain/resource.rs @@ -0,0 +1,7 @@ +use solverforge::prelude::*; + +#[problem_fact] +pub struct Resource { + #[planning_id] + pub id: String, +} diff --git a/crates/solverforge/tests/parallel_local_search/domain/task.rs b/crates/solverforge/tests/parallel_local_search/domain/task.rs new file mode 100644 index 00000000..d7d81e01 --- /dev/null +++ b/crates/solverforge/tests/parallel_local_search/domain/task.rs @@ -0,0 +1,12 @@ +use solverforge::prelude::*; + +#[planning_entity] +pub struct Task { + #[planning_id] + pub id: String, + + pub weight: i64, + + #[planning_variable(value_range_provider = "resources", allows_unassigned = true)] + pub resource_idx: Option, +} diff --git a/docs/architecture.html b/docs/architecture.html index 853a09b6..746b3b90 100644 --- a/docs/architecture.html +++ b/docs/architecture.html @@ -238,6 +238,7 @@ + @@ -505,6 +506,7 @@ + @@ -772,6 +774,7 @@ + @@ -1039,6 +1042,7 @@ PartitionedSearch — rayon work split ExhaustiveSearch · CustomSearchPhase phases ordered by SolverConfig +LocalSearch — rayon window evaluation on replicas @@ -1306,6 +1310,7 @@ +