solverforge_solver/manager/solver_manager/
runtime.rs

1use std::any::Any;
2use std::fmt::{self, Debug};
3use std::sync::atomic::{AtomicBool, Ordering};
4
5use solverforge_core::domain::PlanningSolution;
6
7use super::slot::{
8    JobSlot, SLOT_CANCELLED, SLOT_COMPLETED, SLOT_FAILED, SLOT_PAUSED, SLOT_SOLVING,
9};
10use super::types::{SolverEvent, SolverLifecycleState, SolverTerminalReason};
11use crate::scope::{ProgressCallback, SolverScope};
12use crate::stats::SolverTelemetry;
13
14/// Preserves a foreign-runtime error object while giving retained solving a
15/// stable, displayable failure message.
16pub struct SolverPanicPayload {
17    message: String,
18    payload: Box<dyn Any + Send>,
19}
20
21impl SolverPanicPayload {
22    pub fn new(message: impl Into<String>, payload: impl Any + Send) -> Self {
23        Self {
24            message: message.into(),
25            payload: Box::new(payload),
26        }
27    }
28
29    pub fn message(&self) -> &str {
30        &self.message
31    }
32
33    pub fn into_parts(self) -> (String, Box<dyn Any + Send>) {
34        (self.message, self.payload)
35    }
36}
37
38impl Debug for SolverPanicPayload {
39    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
40        f.debug_struct("SolverPanicPayload")
41            .field("message", &self.message)
42            .finish_non_exhaustive()
43    }
44}
45
46/// Runtime context for a retained solve job.
47///
48/// This is passed into `Solvable::solve()` so the runtime path can publish
49/// lifecycle events, settle exact pauses, and observe cancellation.
50pub struct SolverRuntime<S: PlanningSolution> {
51    job_id: usize,
52    pub(super) slot: &'static JobSlot<S>,
53}
54
55impl<S: PlanningSolution> Clone for SolverRuntime<S> {
56    fn clone(&self) -> Self {
57        *self
58    }
59}
60
61impl<S: PlanningSolution> Copy for SolverRuntime<S> {}
62
63impl<S: PlanningSolution> Debug for SolverRuntime<S> {
64    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
65        f.debug_struct("SolverRuntime")
66            .field("job_id", &self.job_id)
67            .finish()
68    }
69}
70
71#[derive(Debug, Clone, Copy, PartialEq, Eq)]
72enum EventKind {
73    Progress,
74    Resumed,
75    Cancelled,
76}
77
78#[derive(Debug, Clone, Copy, PartialEq, Eq)]
79enum CompletionDisposition {
80    Published,
81    Pause,
82    Cancel,
83    AlreadyTerminal,
84}
85
86impl<S: PlanningSolution> SolverRuntime<S> {
87    pub(super) fn new(job_id: usize, slot: &'static JobSlot<S>) -> Self {
88        Self { job_id, slot }
89    }
90
91    /// Creates a runtime handle for synchronous solves that are not retained by
92    /// a [`SolverManager`](super::SolverManager).
93    ///
94    /// Detached runtimes publish lifecycle state into an internal slot without
95    /// an event receiver. Retained solves should continue to use
96    /// `SolverManager`, which owns reusable slots and event delivery.
97    pub fn detached() -> Self {
98        let slot = Box::leak(Box::new(JobSlot::new()));
99        slot.state.store(SLOT_SOLVING, Ordering::Release);
100        slot.worker_running.store(true, Ordering::Release);
101        Self {
102            job_id: usize::MAX,
103            slot,
104        }
105    }
106
107    pub fn job_id(&self) -> usize {
108        self.job_id
109    }
110
111    pub fn is_cancel_requested(&self) -> bool {
112        self.slot.terminate.load(Ordering::Acquire)
113    }
114
115    pub(crate) fn is_pause_requested(&self) -> bool {
116        self.slot.pause_requested.load(Ordering::Acquire)
117    }
118
119    pub(crate) fn cancel_flag(&self) -> &'static AtomicBool {
120        &self.slot.terminate
121    }
122
123    pub fn emit_progress(
124        &self,
125        current_score: Option<S::Score>,
126        best_score: Option<S::Score>,
127        telemetry: SolverTelemetry,
128    ) {
129        self.emit_non_snapshot_event(current_score, best_score, telemetry, EventKind::Progress);
130    }
131
132    pub fn emit_best_solution(
133        &self,
134        solution: S,
135        current_score: Option<S::Score>,
136        best_score: S::Score,
137        telemetry: SolverTelemetry,
138    ) {
139        self.slot.with_publication(|sender, record| {
140            let state = self.current_state();
141            let terminal_reason = record.terminal_reason;
142            record.current_score = current_score;
143            record.best_score = Some(best_score);
144            record.publish_telemetry(telemetry);
145
146            let snapshot_revision = record.push_snapshot(super::types::SolverSnapshot {
147                job_id: self.job_id,
148                snapshot_revision: 0,
149                lifecycle_state: state,
150                terminal_reason,
151                current_score,
152                best_score: Some(best_score),
153                telemetry: record.telemetry.clone(),
154                solution: solution.clone(),
155            });
156
157            let metadata = record.next_metadata(self.job_id, state, Some(snapshot_revision));
158            if let Some(sender) = sender {
159                let _ = sender.send(SolverEvent::BestSolution { metadata, solution });
160            }
161        });
162    }
163
164    pub fn emit_completed(
165        &self,
166        solution: S,
167        current_score: Option<S::Score>,
168        best_score: S::Score,
169        telemetry: SolverTelemetry,
170        terminal_reason: SolverTerminalReason,
171    ) {
172        loop {
173            let disposition = self.slot.with_publication(|sender, record| {
174                if self.current_state().is_terminal() {
175                    return CompletionDisposition::AlreadyTerminal;
176                }
177                if self.is_cancel_requested() {
178                    return CompletionDisposition::Cancel;
179                }
180                if self.is_pause_requested()
181                    || matches!(
182                        self.current_state(),
183                        SolverLifecycleState::PauseRequested | SolverLifecycleState::Paused
184                    )
185                {
186                    return CompletionDisposition::Pause;
187                }
188
189                self.slot.state.store(SLOT_COMPLETED, Ordering::SeqCst);
190                record.terminal_reason = Some(terminal_reason);
191                record.checkpoint_available = false;
192                record.current_score = current_score;
193                record.best_score = Some(best_score);
194                record.publish_telemetry(telemetry.clone());
195
196                let snapshot_revision = record.push_snapshot(super::types::SolverSnapshot {
197                    job_id: self.job_id,
198                    snapshot_revision: 0,
199                    lifecycle_state: SolverLifecycleState::Completed,
200                    terminal_reason: Some(terminal_reason),
201                    current_score,
202                    best_score: Some(best_score),
203                    telemetry: record.telemetry.clone(),
204                    solution: solution.clone(),
205                });
206
207                let metadata = record.next_metadata(
208                    self.job_id,
209                    SolverLifecycleState::Completed,
210                    Some(snapshot_revision),
211                );
212                if let Some(sender) = sender {
213                    let _ = sender.send(SolverEvent::Completed {
214                        metadata,
215                        solution: solution.clone(),
216                    });
217                }
218                CompletionDisposition::Published
219            });
220
221            match disposition {
222                CompletionDisposition::Published | CompletionDisposition::AlreadyTerminal => {
223                    return;
224                }
225                CompletionDisposition::Cancel => {
226                    self.emit_cancelled(current_score, Some(best_score), telemetry);
227                    return;
228                }
229                CompletionDisposition::Pause => {
230                    if self.pause_with_snapshot(
231                        solution.clone(),
232                        current_score,
233                        Some(best_score),
234                        telemetry.clone(),
235                    ) {
236                        continue;
237                    }
238                    if self.is_cancel_requested() {
239                        self.emit_cancelled(current_score, Some(best_score), telemetry);
240                        return;
241                    }
242                }
243            }
244        }
245    }
246
247    pub fn emit_cancelled(
248        &self,
249        current_score: Option<S::Score>,
250        best_score: Option<S::Score>,
251        telemetry: SolverTelemetry,
252    ) {
253        self.emit_non_snapshot_terminal_event(
254            SolverLifecycleState::Cancelled,
255            SolverTerminalReason::Cancelled,
256            current_score,
257            best_score,
258            telemetry,
259            EventKind::Cancelled,
260        );
261    }
262
263    pub fn emit_failed(&self, error: String) {
264        self.slot.with_publication(|sender, record| {
265            if matches!(
266                self.current_state(),
267                SolverLifecycleState::Completed
268                    | SolverLifecycleState::Cancelled
269                    | SolverLifecycleState::Failed
270            ) {
271                return;
272            }
273            self.slot.state.store(SLOT_FAILED, Ordering::SeqCst);
274            record.terminal_reason = Some(SolverTerminalReason::Failed);
275            record.checkpoint_available = false;
276            record.failure_message = Some(error.clone());
277            record.clear_candidate_trace_detail();
278            let metadata = record.next_metadata(self.job_id, SolverLifecycleState::Failed, None);
279            if let Some(sender) = sender {
280                let _ = sender.send(SolverEvent::Failed { metadata, error });
281            }
282        });
283    }
284
285    pub(crate) fn pause_if_requested<D, ProgressCb>(
286        &self,
287        solver_scope: &mut SolverScope<'_, S, D, ProgressCb>,
288    ) where
289        D: solverforge_scoring::Director<S>,
290        ProgressCb: ProgressCallback<S>,
291    {
292        if !self.slot.pause_requested.load(Ordering::Acquire) || self.is_cancel_requested() {
293            return;
294        }
295
296        solver_scope.pause_timers();
297
298        let solution = solver_scope.score_director().clone_working_solution();
299        let current_score = solver_scope.current_score().copied();
300        let best_score = solver_scope.best_score().copied();
301        let telemetry = solver_scope.stats().snapshot();
302        let _ = self.pause_with_snapshot(solution, current_score, best_score, telemetry);
303        solver_scope.resume_timers();
304    }
305
306    pub fn pause_with_snapshot(
307        &self,
308        solution: S,
309        current_score: Option<S::Score>,
310        best_score: Option<S::Score>,
311        telemetry: SolverTelemetry,
312    ) -> bool {
313        if !self.slot.pause_requested.load(Ordering::Acquire) || self.is_cancel_requested() {
314            return false;
315        }
316
317        let (telemetry, candidate_trace) = telemetry.split_candidate_trace();
318        let published = self.slot.with_publication(|sender, record| {
319            if !self.slot.pause_requested.load(Ordering::Acquire)
320                || self.is_cancel_requested()
321                || self.current_state().is_terminal()
322            {
323                return false;
324            }
325            self.slot.state.store(SLOT_PAUSED, Ordering::SeqCst);
326            let terminal_reason = record.terminal_reason;
327            record.checkpoint_available = true;
328            record.current_score = current_score;
329            record.best_score = best_score;
330            record.publish_telemetry_with_candidate_trace(telemetry.clone(), candidate_trace);
331
332            let snapshot_revision = record.push_snapshot(super::types::SolverSnapshot {
333                job_id: self.job_id,
334                snapshot_revision: 0,
335                lifecycle_state: SolverLifecycleState::Paused,
336                terminal_reason,
337                current_score,
338                best_score,
339                telemetry: record.telemetry.clone(),
340                solution,
341            });
342
343            let metadata = record.next_metadata(
344                self.job_id,
345                SolverLifecycleState::Paused,
346                Some(snapshot_revision),
347            );
348            if let Some(sender) = sender {
349                let _ = sender.send(SolverEvent::Paused { metadata });
350            }
351            true
352        });
353        if !published {
354            return false;
355        }
356
357        let mut guard = self.slot.pause_gate.lock().unwrap();
358        while self.slot.pause_requested.load(Ordering::Acquire) && !self.is_cancel_requested() {
359            guard = self.slot.pause_condvar.wait(guard).unwrap();
360        }
361        drop(guard);
362
363        if self.is_cancel_requested() {
364            return false;
365        }
366
367        self.emit_non_snapshot_event(current_score, best_score, telemetry, EventKind::Resumed);
368        true
369    }
370
371    pub(crate) fn is_terminal(&self) -> bool {
372        self.current_state().is_terminal()
373    }
374
375    fn current_state(&self) -> SolverLifecycleState {
376        self.slot
377            .raw_state()
378            .expect("runtime accessed a freed job slot")
379    }
380
381    fn emit_non_snapshot_event(
382        &self,
383        current_score: Option<S::Score>,
384        best_score: Option<S::Score>,
385        telemetry: SolverTelemetry,
386        kind: EventKind,
387    ) {
388        self.slot.with_publication(|sender, record| {
389            if kind == EventKind::Resumed && self.is_cancel_requested() {
390                return;
391            }
392            let lifecycle_state = match kind {
393                EventKind::Progress => self.current_state(),
394                EventKind::Resumed => {
395                    self.slot.state.store(SLOT_SOLVING, Ordering::SeqCst);
396                    SolverLifecycleState::Solving
397                }
398                EventKind::Cancelled => unreachable!(),
399            };
400            record.current_score = current_score;
401            record.best_score = best_score;
402            record.publish_telemetry(telemetry);
403            if lifecycle_state != SolverLifecycleState::Paused {
404                record.checkpoint_available = false;
405            }
406            let metadata = record.next_metadata(self.job_id, lifecycle_state, None);
407            let event = match kind {
408                EventKind::Progress => SolverEvent::Progress { metadata },
409                EventKind::Resumed => SolverEvent::Resumed { metadata },
410                EventKind::Cancelled => unreachable!(),
411            };
412            if let Some(sender) = sender {
413                let _ = sender.send(event);
414            }
415        });
416    }
417
418    fn emit_non_snapshot_terminal_event(
419        &self,
420        lifecycle_state: SolverLifecycleState,
421        terminal_reason: SolverTerminalReason,
422        current_score: Option<S::Score>,
423        best_score: Option<S::Score>,
424        telemetry: SolverTelemetry,
425        kind: EventKind,
426    ) {
427        self.slot.with_publication(|sender, record| {
428            if self.current_state().is_terminal() {
429                return;
430            }
431            match lifecycle_state {
432                SolverLifecycleState::Cancelled => {
433                    self.slot.state.store(SLOT_CANCELLED, Ordering::SeqCst);
434                }
435                SolverLifecycleState::Failed => {
436                    self.slot.state.store(SLOT_FAILED, Ordering::SeqCst);
437                }
438                _ => {}
439            }
440            record.terminal_reason = Some(terminal_reason);
441            record.checkpoint_available = false;
442            record.current_score = current_score;
443            record.best_score = best_score;
444            record.publish_telemetry(telemetry);
445            let metadata = record.next_metadata(self.job_id, lifecycle_state, None);
446            let event = match kind {
447                EventKind::Cancelled => SolverEvent::Cancelled { metadata },
448                EventKind::Progress | EventKind::Resumed => unreachable!(),
449            };
450            if let Some(sender) = sender {
451                let _ = sender.send(event);
452            }
453        });
454    }
455}
456
457pub(super) fn panic_payload_to_string(payload: Box<dyn Any + Send>) -> String {
458    if let Some(message) = payload.downcast_ref::<&'static str>() {
459        (*message).to_string()
460    } else if let Some(message) = payload.downcast_ref::<String>() {
461        message.clone()
462    } else if let Some(payload) = payload.downcast_ref::<SolverPanicPayload>() {
463        payload.message().to_string()
464    } else {
465        "solver panicked with a non-string payload".to_string()
466    }
467}
468
469#[cfg(test)]
470mod tests {
471    use super::{panic_payload_to_string, SolverPanicPayload};
472
473    #[test]
474    fn retained_failure_uses_foreign_runtime_message() {
475        let payload = SolverPanicPayload::new("foreign traceback", 41_u8);
476        assert_eq!(
477            panic_payload_to_string(Box::new(payload)),
478            "foreign traceback"
479        );
480    }
481
482    #[test]
483    fn foreign_runtime_payload_remains_owned() {
484        let payload = SolverPanicPayload::new("foreign traceback", 41_u8);
485        let (message, payload) = payload.into_parts();
486        assert_eq!(message, "foreign traceback");
487        assert_eq!(*payload.downcast::<u8>().expect("u8 payload"), 41);
488    }
489}
solverforge_solver/manager/solver_manager/runtime.rs

solverforge_solver/manager/solver_manager/
runtime.rs
