diff --git a/src/constraints/compiled_dispatch_tests.rs b/src/constraints/compiled_dispatch_tests.rs index f50d968..3a7cf4e 100644 --- a/src/constraints/compiled_dispatch_tests.rs +++ b/src/constraints/compiled_dispatch_tests.rs @@ -94,7 +94,8 @@ 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()); + // Scoring reads no family, so this thread needs no solve scope. + assert_eq!(set.on_retract_all(&solution, 0, 0), DynamicScore::ZERO); // A cached delta must not parse even precedence-specific immutable fields. let plan = set.compiled.plans()[0] .python_plan() diff --git a/src/constraints/list_precedence.rs b/src/constraints/list_precedence.rs index 5dcfcb7..28f1b8e 100644 --- a/src/constraints/list_precedence.rs +++ b/src/constraints/list_precedence.rs @@ -1013,8 +1013,7 @@ fn precedence_score_for_family( let soft = makespan.saturating_neg(); match score_family { "soft" => DynamicScore::soft(soft.saturating_add(hard)), - "hard_soft_decimal" => DynamicScore::hard_soft_decimal(hard, soft), - "hard_medium_soft" => DynamicScore::hard_medium_soft(hard, 0, soft), + "hard_medium_soft" => DynamicScore::of(hard, 0, soft), _ => DynamicScore::hard_soft(hard, soft), } } diff --git a/src/constraints/state.rs b/src/constraints/state.rs index 635c429..db9decf 100644 --- a/src/constraints/state.rs +++ b/src/constraints/state.rs @@ -15,7 +15,7 @@ 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, score_family_from_name, DynamicScore}; use crate::state::PyDynamicSolution; use crate::value::DynamicValue; @@ -354,8 +354,11 @@ impl PyDynamicConstraintSet { } pub fn from_solution(py: Python<'_>, solution: &PyDynamicSolution) -> PyResult { + // Rejects an undeclared family before any constraint is compiled. + score_family_from_name(&solution.schema().score_family)?; + let compiled = CompiledConstraintSet::from_solution(py, solution)?; Ok(Self { - compiled: CompiledConstraintSet::from_solution(py, solution)?, + compiled, cached_rows: None, join_key_cache: JoinKeyCache::new(), join_index_cache: JoinIndexCache::new(), @@ -806,7 +809,6 @@ fn key_from_value(value: &DynamicValue) -> PyResult> { fn score_times(score: DynamicScore, count: usize) -> DynamicScore { let count = i64::try_from(count).unwrap_or(i64::MAX); DynamicScore { - family: score.family, hard: score.hard.saturating_mul(count), medium: score.medium.saturating_mul(count), soft: score.soft.saturating_mul(count), 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/score.rs b/src/score.rs index c80b126..a8f7ab6 100644 --- a/src/score.rs +++ b/src/score.rs @@ -1,6 +1,5 @@ use pyo3::prelude::*; use pyo3::types::{PyDict, PyList, PyTuple}; -use solverforge_core::score::Score; use crate::error::py_err; @@ -42,7 +41,7 @@ pub fn dynamic_score_from_dict(dict: &Bound<'_, PyDict>) -> PyResult()) .collect::>>()?; - score_from_levels(family.as_deref(), numbers.as_slice(), None) + score_from_levels(family.as_deref(), numbers.as_slice()) } fn dynamic_score_from_sequence<'py>( @@ -51,21 +50,17 @@ fn dynamic_score_from_sequence<'py>( let numbers = items .map(|item| item.extract::()) .collect::>>()?; - score_from_levels(None, numbers.as_slice(), Some(DynamicScore::zero().family)) + score_from_levels(None, numbers.as_slice()) } -fn score_from_levels( - family: Option<&str>, - numbers: &[i64], - preferred_family: Option, -) -> PyResult { +fn score_from_levels(family: Option<&str>, numbers: &[i64]) -> PyResult { Ok(match (family, numbers) { (Some("soft"), [soft]) => DynamicScore::soft(*soft), - (Some("hard_soft"), [hard, soft]) => DynamicScore::hard_soft(*hard, *soft), - (Some("hard_soft_decimal"), [hard, soft]) => DynamicScore::hard_soft_decimal(*hard, *soft), - (Some("hard_medium_soft"), [hard, medium, soft]) => { - DynamicScore::hard_medium_soft(*hard, *medium, *soft) + // Decimal levels arrive already scaled; they share the two-level value. + (Some("hard_soft" | "hard_soft_decimal"), [hard, soft]) => { + DynamicScore::hard_soft(*hard, *soft) } + (Some("hard_medium_soft"), [hard, medium, soft]) => DynamicScore::of(*hard, *medium, *soft), (Some(family), _) => { return Err(py_err(format!( "score family `{family}` is incompatible with {} levels", @@ -73,11 +68,8 @@ fn score_from_levels( ))); } (None, [soft]) => DynamicScore::soft(*soft), - (None, [hard, soft]) if preferred_family == Some(ScoreFamily::HardSoftDecimal) => { - DynamicScore::hard_soft_decimal(*hard, *soft) - } (None, [hard, soft]) => DynamicScore::hard_soft(*hard, *soft), - (None, [hard, medium, soft]) => DynamicScore::hard_medium_soft(*hard, *medium, *soft), + (None, [hard, medium, soft]) => DynamicScore::of(*hard, *medium, *soft), (None, other) => { return Err(py_err(format!( "unsupported dynamic score level count {}", diff --git a/src/solver/api.rs b/src/solver/api.rs index 55f43c6..97ebedb 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,28 +59,35 @@ 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)?; - let mut solved = catch_unwind(AssertUnwindSafe(|| { - scoped_dynamic_score_family(score_family, || { - try_run_dynamic_solver_with_config_parts( - imported, - constraints, - descriptor, - dynamic_entity_count, - SolverRuntime::detached(), - solver_config, - 30, - dynamic_log_scale, - None, - plan.model().clone(), - ) - .map_err(|error| py_err(error.to_string())) + let model = plan.model().clone(); + // The solve runs without the GIL: parallel evaluation scores evaluator + // replicas on Rayon workers that attach to Python themselves, and a + // calling thread that kept the GIL while waiting on them would never let + // them in. + let mut solved = py + .detach(|| { + catch_unwind(AssertUnwindSafe(|| { + scoped_dynamic_score_family(score_family, || { + try_run_dynamic_solver_with_config_parts( + imported, + dynamic_constraints, + descriptor, + dynamic_entity_count, + SolverRuntime::detached(), + solver_config, + 30, + dynamic_log_scale, + None, + model, + ) + .map_err(|error| py_err(error.to_string())) + }) + })) }) - })) - .map_err(panic_to_py_err)??; + .map_err(panic_to_py_err)??; solved.refresh_all_shadows()?; export_solution(py_solution, &solved)?; if let Some(score) = solved.score { 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..d6396cf 100644 --- a/tests/python/test_threading.py +++ b/tests/python/test_threading.py @@ -1,10 +1,190 @@ from concurrent.futures import ThreadPoolExecutor +from typing import Any + +import pytest from tests.python.test_scalar_solving import Schedule -from solverforge import Solver +from solverforge import ( + ConstraintFactory, + HardMediumSoftScore, + HardSoftDecimalScore, + HardSoftScore, + SoftScore, + Solver, + constraint_provider, + planning_entity, + planning_list_variable, + 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 _fixed_work_config(count: int) -> dict[str, Any]: + return { + "random_seed": 7, + "move_thread_count": {"count": count}, + "phases": [ + {"type": "construction_heuristic"}, + {"type": "local_search", "termination": {"step_count_limit": 60}}, + ], + } + + +def _solve_with_threads(count: int) -> ParallelPlan: + return Solver.solve(ParallelPlan(), _fixed_work_config(count)) + + +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 + + +# Each family's declared name and its (unassigned, not preferred) penalties. +FAMILY_WEIGHTS: dict[type, tuple[str, object, object]] = { + SoftScore: ("soft", SoftScore.of(10), SoftScore.ONE_SOFT), + HardSoftScore: ("hard_soft", HardSoftScore.ONE_HARD, HardSoftScore.ONE_SOFT), + HardSoftDecimalScore: ( + "hard_soft_decimal", + HardSoftDecimalScore.ONE_HARD, + HardSoftDecimalScore.ONE_SOFT, + ), + HardMediumSoftScore: ( + "hard_medium_soft", + HardMediumSoftScore.ONE_HARD, + HardMediumSoftScore.ONE_MEDIUM, + ), +} + + +def _family_plan(score_cls: type) -> type: + _, unassigned, not_preferred = FAMILY_WEIGHTS[score_cls] + + @planning_entity + class FamilyVisit: + 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 family_constraints(factory: ConstraintFactory): + return [ + factory.for_each(FamilyVisit) + .filter(lambda visit: visit.nurse is None) + .penalize(unassigned) + .named("unassigned"), + factory.for_each(FamilyVisit) + .filter(lambda visit: visit.nurse is not None and visit.nurse != visit.preferred) + .penalize(not_preferred) + .named("not preferred"), + ] + + @planning_solution(score=score_cls, constraints=family_constraints) + class FamilyPlan: + visits: list[FamilyVisit] + + def __init__(self) -> None: + self.visits = [FamilyVisit((index * 7 + 3) % 5) for index in range(12)] + self.nurses = [0, 1, 2, 3, 4] + self.score = None + + return FamilyPlan + + +@pytest.mark.parametrize("score_cls", list(FAMILY_WEIGHTS), ids=lambda cls: cls.__name__) +def test_parallel_evaluation_keeps_every_score_family(score_cls: type) -> None: + plan_cls = _family_plan(score_cls) + plans = [Solver.solve(plan_cls(), _fixed_work_config(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 + family = FAMILY_WEIGHTS[score_cls][0] + assert all(plan.score["family"] == family for plan in plans) + + +@planning_entity +class ParallelRoute: + visits = planning_list_variable(element_collection="stops") + + def __init__(self) -> None: + self.visits: list[int] = [] + + +def _route_cost(route: ParallelRoute) -> SoftScore: + # Position-dependent cost, so the order inside a route matters as well + # as which route owns a stop. + return SoftScore.of( + sum((stop * (position + 3)) % 7 for position, stop in enumerate(route.visits)) + ) + + +@constraint_provider +def parallel_route_constraints(factory: ConstraintFactory): + return [ + factory.for_each(ParallelRoute).penalize(_route_cost).named("route cost"), + factory.for_each(ParallelRoute) + .penalize(lambda route: SoftScore.of(len(route.visits) ** 2)) + .named("balance"), + ] + + +@planning_solution(score=SoftScore, constraints=parallel_route_constraints) +class ParallelRoutePlan: + routes: list[ParallelRoute] + + def __init__(self) -> None: + self.routes = [ParallelRoute() for _ in range(3)] + self.stops = list(range(14)) + self.score = None + + +def test_parallel_list_evaluation_matches_sequential_at_fixed_work() -> None: + plans = [Solver.solve(ParallelRoutePlan(), _fixed_work_config(count)) for count in (1, 2, 4)] + + routes = [[route.visits for route in plan.routes] for plan in plans] + assert routes[0] == routes[1] == routes[2] + assert plans[0].score == plans[1].score == plans[2].score + assert sorted(stop for route in routes[0] for stop in route) == list(range(14)) 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..118fcff 100644 --- a/tests/rust/constraint_set.rs +++ b/tests/rust/constraint_set.rs @@ -40,6 +40,8 @@ fn dynamic_constraint_set_reports_constraint_count() { solverforge_config::SolverConfig::default(), 0, ); + // The test thread never entered a solve's family scope; scoring + // reads no family, so an empty constraint set scores plain zero. assert_eq!(set.evaluate_all(&solution), DynamicScore::ZERO); }); }