add input stream for parsolver

examples/sat/main.rs

@@ -129,6 +129,13 @@ fn solve_multi_threads(model: Model, opt: &Opt, num_threads: usize) -> Result<()
                 std::process::exit(1);
             }
         }
+        SolverResult::Interrupt(_) => {
+            println!("INTERRUPT");
+            if opt.expected_satisfiability.is_some() {
+                eprintln!("Error: could not conclude on SAT or UNSAT within the allocated time");
+                std::process::exit(1);
+            }
+        }
     }
     par_solver.print_stats();
 

examples/scheduling/src/main.rs

@@ -136,6 +136,15 @@ fn solve(kind: ProblemKind, instance: &str, opt: &Opt) {
             solver.print_stats();
             println!("TIMEOUT (best solution cost {best_cost}");
         }
+        SolverResult::Interrupt(None) => {
+            solver.print_stats();
+            println!("INTERRUPT (not solution found)");
+        }
+        SolverResult::Interrupt(Some(solution)) => {
+            let best_cost = solution.var_domain(makespan).lb;
+            solver.print_stats();
+            println!("INTERRUPT (best solution cost {best_cost}");
+        }
     }
     println!("TOTAL RUNTIME: {:.6}", start_time.elapsed().as_secs_f64());
 }

planning/grpc/server/src/bin/server.rs

@@ -130,6 +130,7 @@ pub fn solve(
                 engine: Some(engine()),
             })
         }
+
         SolverResult::Timeout(opt_plan) => {
             println!("************* TIMEOUT **************");
             let opt_plan = if let Some((finite_problem, plan)) = opt_plan {
@@ -153,6 +154,30 @@ pub fn solve(
                 engine: Some(engine()),
             })
         }
+
+        SolverResult::Interrupt(opt_plan) => {
+            println!("************* INTERRUPT **************");
+            let opt_plan = if let Some((finite_problem, plan)) = opt_plan {
+                println!("\n{}", solver::format_plan(&finite_problem, &plan, htn_mode)?);
+                Some(serialize_plan(problem, &finite_problem, &plan)?)
+            } else {
+                None
+            };
+
+            let status = if opt_plan.is_none() {
+                up::plan_generation_result::Status::Timeout
+            } else {
+                up::plan_generation_result::Status::SolvedSatisficing
+            };
+
+            Ok(up::PlanGenerationResult {
+                status: status as i32,
+                plan: opt_plan,
+                metrics: Default::default(),
+                log_messages: vec![],
+                engine: Some(engine()),
+            })
+        }
     }
 }
 #[derive(Default)]

planning/planners/src/bin/lcp.rs

@@ -135,6 +135,7 @@ fn main() -> Result<()> {
             println!("\nNo plan found");
         }
         SolverResult::Timeout(_) => println!("\nTimeout"),
+        SolverResult::Interrupt(_) => println!("\nInterrupt"),
     }
 
     Ok(())

planning/planners/src/solver.rs

@@ -151,7 +151,7 @@ pub fn solve(
 fn propagate_and_print(pb: &FiniteProblem) -> bool {
     let Ok(EncodedProblem { model, .. }) = encode(pb, None) else {
         println!("==> Invalid model");
-        return false
+        return false;
     };
     let mut solver = init_solver(model);
 
@@ -292,8 +292,13 @@ fn solve_finite_problem(
     if PRINT_INITIAL_PROPAGATION.get() {
         propagate_and_print(&pb);
     }
-    let Ok( EncodedProblem { mut model, objective: metric, encoding }) = encode(&pb, metric) else {
-        return SolverResult::Unsat
+    let Ok(EncodedProblem {
+        mut model,
+        objective: metric,
+        encoding,
+    }) = encode(&pb, metric)
+    else {
+        return SolverResult::Unsat;
     };
     if let Some(metric) = metric {
         model.enforce(metric.le_lit(cost_upper_bound), []);

solver/src/solver/parallel/parallel_solver.rs

@@ -1,7 +1,7 @@
 use crate::model::extensions::{AssignmentExt, SavedAssignment, Shaped};
 use crate::model::lang::IAtom;
 use crate::model::{Label, ModelShape};
-use crate::solver::parallel::signals::{InputSignal, InputStream, OutputSignal, SolverOutput, ThreadID};
+use crate::solver::parallel::signals::{InputSignal, InputStream, OutputSignal, SolverOutput, Synchro, ThreadID};
 use crate::solver::{Exit, Solver};
 use crossbeam_channel::{select, Receiver, Sender};
 use std::sync::Arc;
@@ -11,6 +11,7 @@ use std::time::{Duration, Instant};
 pub struct ParSolver<Lbl> {
     base_model: ModelShape<Lbl>,
     solvers: Vec<Worker<Lbl>>,
+    synchro: Synchro,
 }
 
 pub type Solution = Arc<SavedAssignment>;
@@ -20,6 +21,7 @@ pub enum SolverResult<Solution> {
     Sol(Solution),
     /// The solver terminated, without a finding a solution
     Unsat,
+    Interrupt(Option<Solution>),
     /// Teh solver was interrupted due to a timeout.
     /// It may have found a suboptimal solution.
     Timeout(Option<Solution>),
@@ -31,6 +33,7 @@ impl<Sol> SolverResult<Sol> {
             SolverResult::Sol(s) => SolverResult::Sol(f(s)),
             SolverResult::Unsat => SolverResult::Unsat,
             SolverResult::Timeout(opt_sol) => SolverResult::Timeout(opt_sol.map(f)),
+            SolverResult::Interrupt(opt_sol) => SolverResult::Interrupt(opt_sol.map(f)),
         }
     }
 }
@@ -87,6 +90,7 @@ impl<Lbl: Label> ParSolver<Lbl> {
         let mut solver = ParSolver {
             base_model: base_solver.model.shape.clone(),
             solvers: Vec::with_capacity(num_workers),
+            synchro: Default::default(),
         };
         for i in 0..(num_workers - 1) {
             let mut s = base_solver.clone();
@@ -114,6 +118,10 @@ impl<Lbl: Label> ParSolver<Lbl> {
         rcv
     }
 
+    pub fn input_stream(&mut self) -> InputStream {
+        self.synchro.input_stream()
+    }
+
     /// Solve the problem that was given on initialization using all available solvers.
     pub fn solve(&mut self, deadline: Option<Instant>) -> SolverResult<Solution> {
         self.race_solvers(|s| s.solve(), |_| {}, deadline)
@@ -216,6 +224,26 @@ impl<Lbl: Label> ParSolver<Lbl> {
                 Duration::MAX
             };
             select! {
+                recv(self.synchro.signals) -> res => {
+                    if let Ok(signal) = res {
+                        match signal {
+                            InputSignal::Interrupt => {
+                                for s in &mut self.solvers {
+                                // notify all threads that they should stop ASAP
+                                    s.interrupt()
+                                }
+                                let result = match status {
+                                    SolverStatus::Pending => SolverResult::Interrupt(None),
+                                    SolverStatus::Intermediate(sol) => SolverResult::Interrupt(Some(sol)),
+                                    SolverStatus::Final(result) => result,
+                                };
+                                status = SolverStatus::Final(result);
+                            }
+                            _ => unreachable!()
+
+                        }
+                    }
+                }
                 recv(result_rcv) -> res => { // solver termination
                     let WorkerResult {
                         id: worker_id,