diff --git a/src/constraints/compiled_dispatch_tests.rs b/src/constraints/compiled_dispatch_tests.rs index f50d968..7e18513 100644 --- a/src/constraints/compiled_dispatch_tests.rs +++ b/src/constraints/compiled_dispatch_tests.rs @@ -94,7 +94,12 @@ fn compiled_precedence_never_decodes_generic_rows_or_plan_fields() { let baseline = set.initialize_all(&solution); assert_eq!(baseline, DynamicScore::hard_soft(0, -5)); assert!(set.cached_rows.is_none()); - assert_eq!(set.on_retract_all(&solution, 0, 0), DynamicScore::zero()); + // The declared family holds even though this thread never entered a + // solve scope. + assert_eq!( + set.on_retract_all(&solution, 0, 0), + DynamicScore::zero_for_family(ScoreFamily::HardSoft) + ); // A cached delta must not parse even precedence-specific immutable fields. let plan = set.compiled.plans()[0] .python_plan() diff --git a/src/constraints/state.rs b/src/constraints/state.rs index 635c429..177613c 100644 --- a/src/constraints/state.rs +++ b/src/constraints/state.rs @@ -15,7 +15,10 @@ use super::compiled::{ use super::evaluate; use super::list_precedence::{self, ListPrecedenceDeltaMode, ListPrecedenceStateCache}; use crate::error::{panic_with_py_err, py_err}; -use crate::score::{dynamic_score_from_native, DynamicScore}; +use crate::score::{ + dynamic_score_from_native, scoped_dynamic_score_family, score_family_from_name, DynamicScore, + ScoreFamily, +}; use crate::state::PyDynamicSolution; use crate::value::DynamicValue; @@ -333,14 +336,19 @@ pub struct PyDynamicConstraintSet { list_unassigned_count_cache: ListUnassignedCountCache, list_precedence_cache: ListPrecedenceStateCache, retracted_entities: RetractedEntitySet, + /// Declared score family. Every `ConstraintSet` entry point enters it, so + /// scores match on any thread, including Rayon workers scoring evaluator + /// replicas that never entered the solve thread's scope. + family: ScoreFamily, } unsafe impl Send for PyDynamicConstraintSet {} unsafe impl Sync for PyDynamicConstraintSet {} impl PyDynamicConstraintSet { - pub fn new(constraints: Py) -> Self { + pub fn new(constraints: Py, family: ScoreFamily) -> Self { Self { + family, compiled: CompiledConstraintSet::python(constraints), cached_rows: None, join_key_cache: JoinKeyCache::new(), @@ -354,8 +362,13 @@ impl PyDynamicConstraintSet { } pub fn from_solution(py: Python<'_>, solution: &PyDynamicSolution) -> PyResult { + let family = score_family_from_name(&solution.schema().score_family)?; + let compiled = scoped_dynamic_score_family(family, || { + CompiledConstraintSet::from_solution(py, solution) + })?; Ok(Self { - compiled: CompiledConstraintSet::from_solution(py, solution)?, + family, + compiled, cached_rows: None, join_key_cache: JoinKeyCache::new(), join_index_cache: JoinIndexCache::new(), @@ -490,8 +503,10 @@ impl PyDynamicConstraintSet { impl ConstraintSet for PyDynamicConstraintSet { fn evaluate_all(&self, solution: &PyDynamicSolution) -> DynamicScore { - Python::attach(|py| evaluate_compiled_constraints(py, solution, &self.compiled)) - .unwrap_or_else(panic_with_py_err) + scoped_dynamic_score_family(self.family, || { + Python::attach(|py| evaluate_compiled_constraints(py, solution, &self.compiled)) + .unwrap_or_else(panic_with_py_err) + }) } fn constraint_count(&self) -> usize { @@ -517,8 +532,11 @@ impl ConstraintSet for PyDynamicConstraintSet { } fn initialize_all(&mut self, solution: &PyDynamicSolution) -> DynamicScore { - self.initialize_incremental_state(solution) - .unwrap_or_else(panic_with_py_err) + let family = self.family; + scoped_dynamic_score_family(family, || { + self.initialize_incremental_state(solution) + .unwrap_or_else(panic_with_py_err) + }) } fn on_insert_all( @@ -527,21 +545,24 @@ impl ConstraintSet for PyDynamicConstraintSet { entity_index: usize, descriptor_index: usize, ) -> DynamicScore { - let delta = Python::attach(|py| { - self.evaluate_delta( - py, - solution, - DeltaTarget { - entity_index, - descriptor_index, - mode: DeltaMode::Insert, - }, - ) + let family = self.family; + scoped_dynamic_score_family(family, || { + let delta = Python::attach(|py| { + self.evaluate_delta( + py, + solution, + DeltaTarget { + entity_index, + descriptor_index, + mode: DeltaMode::Insert, + }, + ) + }) + .unwrap_or_else(panic_with_py_err); + self.retracted_entities + .remove(&(descriptor_index, entity_index)); + delta }) - .unwrap_or_else(panic_with_py_err); - self.retracted_entities - .remove(&(descriptor_index, entity_index)); - delta } fn on_retract_all( @@ -550,21 +571,24 @@ impl ConstraintSet for PyDynamicConstraintSet { entity_index: usize, descriptor_index: usize, ) -> DynamicScore { - let delta = Python::attach(|py| { - self.evaluate_delta( - py, - solution, - DeltaTarget { - entity_index, - descriptor_index, - mode: DeltaMode::Retract, - }, - ) + let family = self.family; + scoped_dynamic_score_family(family, || { + let delta = Python::attach(|py| { + self.evaluate_delta( + py, + solution, + DeltaTarget { + entity_index, + descriptor_index, + mode: DeltaMode::Retract, + }, + ) + }) + .unwrap_or_else(panic_with_py_err); + self.retracted_entities + .insert((descriptor_index, entity_index)); + delta }) - .unwrap_or_else(panic_with_py_err); - self.retracted_entities - .insert((descriptor_index, entity_index)); - delta } fn reset_all(&mut self) { diff --git a/src/manager/candidate_trace.rs b/src/manager/candidate_trace.rs index f6e80b4..eade17c 100644 --- a/src/manager/candidate_trace.rs +++ b/src/manager/candidate_trace.rs @@ -384,5 +384,6 @@ fn disposition_name(disposition: CandidateTraceDisposition) -> &'static str { CandidateTraceDisposition::ForagerIgnored => "forager_ignored", CandidateTraceDisposition::Selected => "selected", CandidateTraceDisposition::Applied => "applied", + CandidateTraceDisposition::DiscardedSpeculative => "discarded_speculative", } } diff --git a/src/manager/events.rs b/src/manager/events.rs index 0b38198..eb8e7cf 100644 --- a/src/manager/events.rs +++ b/src/manager/events.rs @@ -132,6 +132,7 @@ fn telemetry_to_python(py: Python<'_>, telemetry: &SolverTelemetry) -> PyResult< dict.set_item("step_count", telemetry.step_count)?; dict.set_item("moves_generated", telemetry.moves_generated)?; dict.set_item("moves_evaluated", telemetry.moves_evaluated)?; + dict.set_item("moves_speculative", telemetry.moves_speculative)?; dict.set_item("moves_accepted", telemetry.moves_accepted)?; dict.set_item("moves_applied", telemetry.moves_applied)?; dict.set_item("moves_score_improving", telemetry.moves_score_improving)?; diff --git a/src/solver/api.rs b/src/solver/api.rs index 55f43c6..eddbce4 100644 --- a/src/solver/api.rs +++ b/src/solver/api.rs @@ -12,6 +12,7 @@ use crate::error::{panic_to_py_err, py_err}; use crate::runtime::dynamic_assignment_group::validate_assignment_construction_groups; use crate::schema::compiled::CompiledSchema; use crate::score::{scoped_dynamic_score_family, score_family_from_name, DynamicScorePythonExt}; +use crate::solver::solvable::dynamic_constraints; use crate::state::marshal::{export_solution, import_solution}; use crate::state::PyDynamicSolution; @@ -58,7 +59,6 @@ pub fn solve( let py_solution = solution.bind(py); let mut imported = import_solution(py_solution, Arc::clone(&plan))?; let descriptor = plan.descriptor().clone(); - let constraints = PyDynamicConstraintSet::from_solution(py, &imported)?; validate_assignment_construction_groups(&solver_config, parsed)?; imported.solver_config = solver_config.clone(); let score_family = score_family_from_name(&parsed.score_family)?; @@ -66,7 +66,7 @@ pub fn solve( scoped_dynamic_score_family(score_family, || { try_run_dynamic_solver_with_config_parts( imported, - constraints, + dynamic_constraints, descriptor, dynamic_entity_count, SolverRuntime::detached(), diff --git a/src/solver/solvable.rs b/src/solver/solvable.rs index 3874c87..0f8c925 100644 --- a/src/solver/solvable.rs +++ b/src/solver/solvable.rs @@ -21,15 +21,12 @@ impl Solvable for PyDynamicSolution { let descriptor = runtime_plan.descriptor().clone(); let model = runtime_plan.model().clone(); let default_time_limit_secs = runtime_plan.default_time_limit_secs(); - let constraints = pyo3::Python::attach(|py| { - PyDynamicConstraintSet::from_solution(py, &self).unwrap_or_else(panic_with_py_err) - }); let score_family = score_family_from_name(&runtime_plan.schema().score_family) .unwrap_or_else(panic_with_py_err); let result = scoped_dynamic_score_family(score_family, || { try_run_dynamic_solver_with_config_parts( self, - constraints, + dynamic_constraints, descriptor, dynamic_entity_count, runtime, @@ -44,6 +41,14 @@ impl Solvable for PyDynamicSolution { } } +/// Builds the constraint set for one solution: the working solution and every +/// evaluator replica use this same constructor. +pub(crate) fn dynamic_constraints(solution: &PyDynamicSolution) -> PyDynamicConstraintSet { + pyo3::Python::attach(|py| { + PyDynamicConstraintSet::from_solution(py, solution).unwrap_or_else(panic_with_py_err) + }) +} + fn dynamic_entity_count(solution: &PyDynamicSolution, descriptor_index: usize) -> usize { solution.entity_count(descriptor_index) } diff --git a/tests/python/test_threading.py b/tests/python/test_threading.py index d5e22f4..af5a9ce 100644 --- a/tests/python/test_threading.py +++ b/tests/python/test_threading.py @@ -1,10 +1,72 @@ from concurrent.futures import ThreadPoolExecutor from tests.python.test_scalar_solving import Schedule -from solverforge import Solver +from solverforge import ( + ConstraintFactory, + HardSoftScore, + Solver, + constraint_provider, + planning_entity, + planning_solution, + planning_variable, +) def test_concurrent_python_solves_do_not_share_solution_state() -> None: with ThreadPoolExecutor(max_workers=2) as pool: results = list(pool.map(lambda _: Solver.solve(Schedule()), range(2))) assert results[0] is not results[1] + + +@planning_entity +class ParallelVisit: + nurse = planning_variable(value_range_provider="nurses", allows_unassigned=True) + + def __init__(self, preferred: int) -> None: + self.preferred = preferred + self.nurse = None + + +@constraint_provider +def parallel_constraints(factory: ConstraintFactory): + return [ + factory.for_each(ParallelVisit) + .filter(lambda visit: visit.nurse is None) + .penalize(HardSoftScore.ONE_HARD) + .named("unassigned"), + factory.for_each(ParallelVisit) + .filter(lambda visit: visit.nurse is not None and visit.nurse != visit.preferred) + .penalize(HardSoftScore.ONE_SOFT) + .named("not preferred"), + ] + + +@planning_solution(score=HardSoftScore, constraints=parallel_constraints) +class ParallelPlan: + visits: list[ParallelVisit] + + def __init__(self) -> None: + self.visits = [ParallelVisit((index * 7 + 3) % 5) for index in range(30)] + self.nurses = [0, 1, 2, 3, 4] + self.score = None + + +def _solve_with_threads(count: int) -> ParallelPlan: + return Solver.solve( + ParallelPlan(), + { + "random_seed": 7, + "move_thread_count": {"count": count}, + "phases": [ + {"type": "construction_heuristic"}, + {"type": "local_search", "termination": {"step_count_limit": 60}}, + ], + }, + ) + + +def test_parallel_move_evaluation_matches_sequential_at_fixed_work() -> None: + plans = [_solve_with_threads(count) for count in (1, 2, 4)] + assignments = [[visit.nurse for visit in plan.visits] for plan in plans] + assert assignments[0] == assignments[1] == assignments[2] + assert plans[0].score == plans[1].score == plans[2].score diff --git a/tests/python/test_tracebacks.py b/tests/python/test_tracebacks.py index 355bad5..6e0f9ea 100644 --- a/tests/python/test_tracebacks.py +++ b/tests/python/test_tracebacks.py @@ -183,3 +183,48 @@ def test_dynamic_provider_callback_traceback_surfaces( Solver.solve(plan(), config) frames = traceback.extract_tb(exc_info.value.__traceback__) assert any(frame.name == function_name for frame in frames) + + +@planning_entity +class LateBrokenTask: + worker = planning_variable(value_range_provider="workers", allows_unassigned=True) + + def __init__(self) -> None: + self.worker = 0 + + +@constraint_provider +def late_broken_constraints(factory: ConstraintFactory): + def explode_on_second_worker(task: LateBrokenTask) -> bool: + if task.worker == 1: + raise ValueError("parallel callback failed") + return False + + return [ + factory.for_each(LateBrokenTask) + .filter(explode_on_second_worker) + .penalize(1) + .named("late broken") + ] + + +@planning_solution(constraints=late_broken_constraints) +class LateBrokenPlan: + tasks: list[LateBrokenTask] + + def __init__(self) -> None: + self.tasks = [LateBrokenTask(), LateBrokenTask()] + self.workers = [0, 1] + self.score = None + + +def test_callback_traceback_surfaces_from_parallel_evaluation() -> None: + config = { + "random_seed": 7, + "move_thread_count": {"count": 2}, + "phases": [{"type": "local_search", "termination": {"step_count_limit": 1}}], + } + with pytest.raises(ValueError, match="parallel callback failed") as exc_info: + Solver.solve(LateBrokenPlan(), config) + frames = traceback.extract_tb(exc_info.value.__traceback__) + assert any(frame.name == "explode_on_second_worker" for frame in frames) diff --git a/tests/rust/constraint_set.rs b/tests/rust/constraint_set.rs index 36820bf..06e0544 100644 --- a/tests/rust/constraint_set.rs +++ b/tests/rust/constraint_set.rs @@ -2,7 +2,7 @@ use pyo3::Python; use solverforge_py::constraints::PyDynamicConstraintSet; use solverforge_py::schema::runtime_plan::CompiledRuntimePlan; use solverforge_py::schema::DynamicSchema; -use solverforge_py::score::DynamicScore; +use solverforge_py::score::{DynamicScore, ScoreFamily}; use solverforge_py::state::callback_view::PythonCallbackView; use solverforge_py::state::entity_table::DynamicState; use solverforge_py::state::PyDynamicSolution; @@ -14,7 +14,7 @@ fn dynamic_constraint_set_reports_constraint_count() { crate::initialize_python(); Python::attach(|py| { let constraints = pyo3::types::PyList::empty(py).unbind().into_any(); - let set = PyDynamicConstraintSet::new(constraints); + let set = PyDynamicConstraintSet::new(constraints, ScoreFamily::HardSoft); assert_eq!(set.constraint_count(), 0); let schema = Arc::new(DynamicSchema { solution_type: "Plan".to_string(), @@ -40,6 +40,11 @@ fn dynamic_constraint_set_reports_constraint_count() { solverforge_config::SolverConfig::default(), 0, ); - assert_eq!(set.evaluate_all(&solution), DynamicScore::ZERO); + // The test thread never entered a solve's family scope, so this pins + // that the constraint set scores with its own declared family. + assert_eq!( + set.evaluate_all(&solution), + DynamicScore::zero_for_family(ScoreFamily::HardSoft) + ); }); }