RsBundle  Diff

pub type DefaultSolver<P> = Solver<P, StandardTerminator, HKWeighter, crate::master::FullMasterBuilder>;

/// The minimal bundle solver.
pub type NoBundleSolver<P> = Solver<P, StandardTerminator, HKWeighter, crate::master::MinimalMasterBuilder>;

/// Error raised by the parallel bundle [`Solver`].
#[derive(Debug)]
pub enum Error<MErr, PErr> {
    /// An error raised when creating a new master problem solver.
    BuildMaster(MErr),
    /// An error raised by the master problem process.
    Master(MErr),
    /// The iteration limit has been reached.
    IterationLimit { limit: usize },
    /// An error raised by a subproblem evaluation.
    Evaluation(PErr),
    /// An error raised subproblem update.
    Update(PErr),
    /// The dimension of some data is wrong.
    Dimension(String),
    /// Invalid bounds for a variable.
    InvalidBounds { lower: Real, upper: Real },
    /// The value of a variable is outside its bounds.
    ViolatedBounds { lower: Real, upper: Real, value: Real },
    /// The variable index is out of bounds.
    InvalidVariable { index: usize, nvars: usize },
    /// Disconnected channel.
    Disconnected,
    /// An error occurred in a subprocess.
    Process(RecvError),
    /// A method requiring an initialized solver has been called.
    NotInitialized,
    /// The problem has not been solved yet.
    NotSolved,
}

/// The result type of the solver.
///
/// - `T` is the value type,
/// - `P` is the `FirstOrderProblem` associated with the solver,
/// - `M` is the `MasterBuilder` associated with the solver.
pub type Result<T, P, M> = std::result::Result<
    T,
    Error<<<M as MasterBuilder>::MasterProblem as MasterProblem>::Err, <P as FirstOrderProblem>::Err>,
>;

impl<MErr, PErr> std::fmt::Display for Error<MErr, PErr>
where
    MErr: std::fmt::Display,
    PErr: std::fmt::Display,
{
    fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
        use Error::*;
        match self {
            BuildMaster(err) => writeln!(fmt, "Cannot create master problem solver: {}", err),
            Master(err) => writeln!(fmt, "Error in master problem: {}", err),
            IterationLimit { limit } => writeln!(fmt, "The iteration limit has been reached: {}", limit),
            Evaluation(err) => writeln!(fmt, "Error in subproblem evaluation: {}", err),
            Update(err) => writeln!(fmt, "Error in subproblem update: {}", err),
            Dimension(what) => writeln!(fmt, "Wrong dimension for {}", what),
            InvalidBounds { lower, upper } => write!(fmt, "Invalid bounds, lower:{}, upper:{}", lower, upper),
            ViolatedBounds { lower, upper, value } => write!(
                fmt,
                "Violated bounds, lower:{}, upper:{}, value:{}",
                lower, upper, value
            ),
            InvalidVariable { index, nvars } => {
                write!(fmt, "Variable index out of bounds, got:{} must be < {}", index, nvars)
            }
            Disconnected => writeln!(fmt, "A channel got disconnected"),
            Process(err) => writeln!(fmt, "Error in subprocess: {}", err),
            NotInitialized => writeln!(fmt, "The solver must be initialized (called Solver::init()?)"),
            NotSolved => writeln!(fmt, "The problem has not been solved yet"),
        }
    }
}

impl<MErr, PErr> std::error::Error for Error<MErr, PErr>
where
    MErr: std::error::Error + 'static,
    PErr: std::error::Error + 'static,
{
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        use Error::*;
        match self {
            BuildMaster(err) => Some(err),
            Master(err) => Some(err),
            Evaluation(err) => Some(err),
            Process(err) => Some(err),
            _ => None,
        }
    }
}

impl<MErr, PErr> From<masterprocess::Error<MErr, PErr>> for Error<MErr, PErr>
where
    MErr: std::error::Error + 'static,
{
    fn from(err: masterprocess::Error<MErr, PErr>) -> Error<MErr, PErr> {
        use masterprocess::Error::*;
        match err {
            DisconnectedSender => Error::Disconnected,
            DisconnectedReceiver => Error::Disconnected,
            Aggregation(err) => Error::Master(err),
            SubgradientExtension(err) => Error::Update(err),
            Master(err) => Error::Master(err.into()),
        }
    }
}

impl<MErr, PErr> From<RecvError> for Error<MErr, PErr> {
    fn from(err: RecvError) -> Error<MErr, PErr> {
        Error::Process(err.into())
    }
}

type ClientSender<P> =
    Sender<std::result::Result<EvalResult<usize, <P as FirstOrderProblem>::Primal>, <P as FirstOrderProblem>::Err>>;

    /// This is actually the time of the last call to `Solver::init`.
    start_time: Instant,
}

impl<P, T, W, M> Solver<P, T, W, M>
where
    P: FirstOrderProblem,
    P::Err: Send + 'static,
    T: Terminator<SolverData> + Default,
    W: Weighter<SolverData> + Default,
    M: MasterBuilder,
    <M::MasterProblem as MasterProblem>::MinorantIndex: std::hash::Hash,
{
    /// Create a new parallel bundle solver.
    pub fn new(problem: P) -> Self

    ///
    /// This will reset the internal data structures so that a new fresh
    /// solution process can be started.
    ///
    /// It will also setup all worker processes.
    ///
    /// This function is automatically called by [`Solver::solve`].
    pub fn init(&mut self) -> Result<(), P, M> {
        debug!("Initialize solver");

        let n = self.problem.num_variables();
        let m = self.problem.num_subproblems();

        self.data.init(dvec![0.0; n]);
        self.cnt_descent = 0;

        self.data.cnt_remaining_mins = num_subproblems;
        self.data.nxt_d = Arc::new(dvec![0.0; num_variables]);
        self.data.nxt_y = Arc::new(dvec![]);
        self.data.updated = true;
    }

    /// Solve the problem with the default maximal iteration limit [`DEFAULT_ITERATION_LIMIT`].
    pub fn solve(&mut self) -> Result<(), P, M> {
        self.solve_with_limit(DEFAULT_ITERATION_LIMIT)
    }

    /// Solve the problem with a maximal iteration limit.
    pub fn solve_with_limit(&mut self, limit: usize) -> Result<(), P, M> {
        // First initialize the internal data structures.
        self.init()?;

        if self.solve_iter(limit)? {
            Ok(())
        } else {
            Err(Error::IterationLimit { limit })
        }
    }

    /// Solve the problem but stop after at most `niter` iterations.
    ///
    /// The function returns `Ok(true)` if the termination criterion
    /// has been satisfied. Otherwise it returns `Ok(false)` or an
    /// error code.
    ///
    /// If this function is called again, the solution process is
    /// continued from the previous point. Because of this one *must*
    /// call `init()` before the first call to this function.
    pub fn solve_iter(&mut self, niter: usize) -> Result<bool, P, M> {
        debug!("Start solving up to {} iterations", niter);

        self.reset_iteration_data(niter);
        loop {
            select! {
                recv(self.client_rx.as_ref().ok_or(Error::NotInitialized)?) -> msg => {
                    let msg = msg?.map_err(Error::Evaluation)?;

    /// Handle a response from a subproblem evaluation.
    ///
    /// The function returns `Ok(true)` if the final iteration count has been reached.
    fn handle_client_response(
        &mut self,
        msg: EvalResult<usize, <P as FirstOrderProblem>::Primal>,
    ) -> Result<bool, P, M> {
        let master = self.master_proc.as_mut().ok_or(Error::NotInitialized)?;
        match msg {
            EvalResult::ObjectiveValue { index, value } => {
                debug!("Receive objective from subproblem {}: {}", index, value);
                if self.data.nxt_ubs[index].is_infinite() {
                    self.data.cnt_remaining_ubs -= 1;
                }

        // Compute the new candidate. The main loop will wait for the result of
        // this solution process of the master problem.
        self.master_proc.as_mut().unwrap().solve(self.data.cur_val)?;

        Ok(self.data.cnt_iter >= self.data.max_iter)
    }

    fn handle_master_response(&mut self, master_res: MasterResponse) -> Result<bool, P, M> {
        let master = self.master_proc.as_mut().ok_or(Error::NotInitialized)?;

        self.data.nxt_mod = master_res.nxt_mod;
        self.data.sgnorm = master_res.sgnorm;
        self.data.expected_progress = self.data.cur_val - self.data.nxt_mod;
        self.data.cnt_updates = master_res.cnt_updates;

            self.problem
                .evaluate(i, self.data.nxt_y.clone(), ChannelSender::new(i, client_tx.clone()))
                .map_err(Error::Evaluation)?;
        }
        Ok(false)
    }

    fn update_problem(&mut self, step: Step) -> Result<bool, P, M> {
        let master_proc = self.master_proc.as_mut().ok_or(Error::NotInitialized)?;
        let (update_tx, update_rx) = channel();
        self.problem
            .update(
                UpdateData {
                    cur_y: Arc::new(self.data.cur_y.clone()),
                    nxt_y: self.data.nxt_y.clone(),

            self.data.nxt_mod,
            self.data.nxt_val,
            self.data.cur_val
        );
    }

    /// Return the aggregated primal of the given subproblem.
    pub fn aggregated_primal(&self, subproblem: usize) -> Result<P::Primal, P, M> {
        Ok(self
            .master_proc
            .as_ref()
            .ok_or(Error::NotSolved)?
            .get_aggregated_primal(subproblem)?)
    }
}