RsBundle  Diff

use num_traits::Float;
use std::sync::Arc;
use std::time::Instant;
use threadpool::ThreadPool;

use crate::{DVector, Real};

use super::masterprocess::{MasterConfig, MasterProcess};
use super::masterprocess::{MasterConfig, MasterProcess, MasterResponse};
use super::problem::{EvalResult, FirstOrderProblem};
use crate::master::{self, MasterProblem};
use crate::solver::{SolverParams, Step};
use crate::terminator::{StandardTerminatable, StandardTerminator, Terminator};
use crate::weighter::{HKWeightable, HKWeighter, Weighter};

/// The default iteration limit.

        self.new_cutval
    }

    fn sgnorm(&self) -> Real {
        self.sgnorm
    }
}

/// Internal data used during the main iteration loop.
struct IterData {
    /// Maximal number of iterations.
    max_iter: usize,
    cnt_iter: usize,
    cnt_updates: usize,
    nxt_ubs: Vec<Real>,
    cnt_remaining_ubs: usize,
    nxt_cutvals: Vec<Real>,
    cnt_remaining_mins: usize,
    nxt_d: Arc<DVector>,
    nxt_y: Arc<DVector>,
}

impl IterData {
    fn new(num_subproblems: usize, num_variables: usize, max_iter: usize) -> Self {
        IterData {
            max_iter,
            cnt_iter: 0,
            cnt_updates: 0,
            nxt_ubs: vec![Real::infinity(); num_subproblems],
            cnt_remaining_ubs: num_subproblems,
            nxt_cutvals: vec![-Real::infinity(); num_subproblems],
            cnt_remaining_mins: num_subproblems,
            nxt_d: Arc::new(dvec![0.0; num_variables]),
            nxt_y: Arc::new(dvec![]),
        }
    }
}

/// The default master problem builder.
///
/// This is a fully disaggregatable master problem.
pub type FullMasterBuilder = master::boxed::Builder<master::cpx::Builder>;

/// Implementation of a parallel bundle method.

    ///
    /// 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, Error<P::Err>> {
        debug!("Start solving up to {} iterations", niter);

        let master = self.master_proc.as_mut().ok_or(Error::NotInitialized)?;
        let client_tx = self.client_tx.as_ref().ok_or(Error::NotInitialized)?;
        let client_rx = self.client_rx.as_ref().ok_or(Error::NotInitialized)?;

        let mut cnt_iter = 0;
        let mut cnt_updates = 0;
        let mut nxt_ubs = vec![Real::infinity(); self.problem.num_subproblems()];
        let mut cnt_remaining_ubs = self.problem.num_subproblems();
        let mut itdata = IterData::new(self.problem.num_subproblems(), self.problem.num_variables(), niter);
        let mut nxt_cutvals = vec![-Real::infinity(); self.problem.num_subproblems()];
        let mut cnt_remaining_mins = self.problem.num_subproblems();
        let mut nxt_d = Arc::new(dvec![0.0; self.problem.num_variables()]);
        let mut nxt_y = Arc::new(dvec![]);

        loop {
            select! {
                recv(client_rx) -> msg => {
                recv(self.client_rx.as_ref().ok_or(Error::NotInitialized)?) -> msg => {
                    let msg = msg
                        .map_err(|err| Error::Process(err.into()))?
                        .map_err(Error::Evaluation)?;
                    match msg {
                        EvalResult::ObjectiveValue { index, value } => {
                                debug!("Receive objective from subproblem {}: {}", index, value);
                            if nxt_ubs[index].is_infinite() {
                                cnt_remaining_ubs -= 1;
                            }
                            nxt_ubs[index] = nxt_ubs[index].min(value);
                        }
                        EvalResult::Minorant { index, mut minorant, primal } => {
                            debug!("Receive minorant from subproblem {}", index);
                            if nxt_cutvals[index].is_infinite() {
                                cnt_remaining_mins -= 1;
                            }
                            // move center of minorant to cur_y
                            minorant.move_center(-1.0, &nxt_d);
                            nxt_cutvals[index] = nxt_cutvals[index].max(minorant.constant);
                            // add minorant to master problem
                            master.add_minorant(index, minorant, primal)?;
                        }
                    }

                    if self.handle_client_response(msg, &mut itdata)? {
                    if cnt_remaining_ubs == 0 && cnt_remaining_mins == 0 {
                        // All subproblems have been evaluated, do a step.

                        return Ok(false);
                        let nxt_ub = nxt_ubs.iter().sum::<Real>();
                        let descent_bnd = Self::get_descent_bound(self.params.acceptance_factor, &self.data);

                    }
                        self.data.nxt_val = nxt_ub;
                        self.data.new_cutval = nxt_cutvals.iter().sum::<Real>();

                        debug!("Step");
                        debug!("  cur_val    ={}", self.data.cur_val);
                        debug!("  nxt_mod    ={}", self.data.nxt_mod);
                        debug!("  nxt_ub     ={}", nxt_ub);
                        debug!("  descent_bnd={}", descent_bnd);


                        let step;
                        if self.data.cur_val.is_infinite() {
                            // This is the first evaluation. We effectively get
                            // the function value at the current center but
                            // we do not have a model estimate yet. Hence, we do not know
                            // a good guess for the weight.
                            step = Step::Descent;
                            self.data.cur_val = nxt_ub;
                            self.data.cur_weight = Real::infinity();
                            master.set_weight(1.0)?;

                            debug!("First Step");
                            debug!("  cur_val={}", self.data.cur_val);
                            debug!("  cur_y={}", self.data.cur_y);
                        } else if nxt_ub <= descent_bnd {
                            step = Step::Descent;
                            self.cnt_descent += 1;

                            // Note that we must update the weight *before* we
                            // change the internal data, so the old information
                            // that caused the descent step is still available.
                            self.data.cur_weight = self.weighter.descent_weight(&self.data);
                            self.data.cur_y = nxt_y.as_ref().clone();
                            self.data.cur_val = nxt_ub;

                            master.move_center(1.0, nxt_d.clone())?;
                            master.set_weight(self.data.cur_weight)?;

                            debug!("Descent Step");
                            debug!("  dir ={}", nxt_d);
                            debug!("  newy={}", self.data.cur_y);
                        } else {
                            step = Step::Null;
                            self.cnt_null += 1;
                            self.data.cur_weight = self.weighter.null_weight(&self.data);
                            master.set_weight(self.data.cur_weight)?;
                        }

                },
                recv(self.master_proc.as_ref().ok_or(Error::NotInitialized)?.rx) -> msg => {
                        Self::show_info(&self.start_time, step, &self.data, self.cnt_descent, self.cnt_null, cnt_updates);
                        cnt_iter += 1;
                        if cnt_iter >= niter { break }

                        master.solve(self.data.cur_val)?;
                    }
                },
                recv(master.rx) -> msg => {
                    debug!("Receive master response");
                    // Receive result (new candidate) from the master
                    let master_res = msg
                        .map_err(|err| Error::Process(err.into()))?
                        .map_err(|err| Error::Master(err.into()))?;

                    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;
                    cnt_updates = master_res.cnt_updates;

                    if self.handle_master_response(master_res, &mut itdata)? {
                    // If this is the very first solution of the model,
                    // we use its result as to make a good guess for the initial weight
                    // of the proximal term and resolve.
                    if self.data.cur_weight.is_infinite() {
                        self.data.cur_weight = self.weighter.initial_weight(&self.data);
                        master.set_weight(self.data.cur_weight)?;
                        master.solve(self.data.cur_val)?;
                        continue;
                    }

                    if self.terminator.terminate(&self.data) {
                        Self::show_info(&self.start_time, Step::Term, &self.data, self.cnt_descent, self.cnt_null, cnt_updates);
                        info!("Termination criterion satisfied");
                        return Ok(true)
                        return Ok(true);
                    }

                    // Compress bundle
                    master.compress()?;

                    // Compute new candidate.
                    let mut next_y = dvec![];
                    nxt_d = Arc::new(master_res.nxt_d);
                    next_y.add(&self.data.cur_y, &nxt_d);
                    nxt_y = Arc::new(next_y);

                    // Reset evaluation data.
                    nxt_ubs.clear();
                    nxt_ubs.resize(self.problem.num_subproblems(), Real::infinity());
                    cnt_remaining_ubs = self.problem.num_subproblems();
                    nxt_cutvals.clear();
                    nxt_cutvals.resize(self.problem.num_subproblems(), -Real::infinity());
                    cnt_remaining_mins = self.problem.num_subproblems();

                    // Start evaluation of all subproblems at the new candidate.
                    for i in 0..self.problem.num_subproblems() {
                        self.problem.evaluate(i, nxt_y.clone(), i, client_tx.clone()) .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>,
        itdata: &mut IterData,
    ) -> Result<bool, Error<P::Err>> {
        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 itdata.nxt_ubs[index].is_infinite() {
                    itdata.cnt_remaining_ubs -= 1;
                }
                itdata.nxt_ubs[index] = itdata.nxt_ubs[index].min(value);
            }
            EvalResult::Minorant {
                index,
                mut minorant,
                primal,
            } => {
                debug!("Receive minorant from subproblem {}", index);
                if itdata.nxt_cutvals[index].is_infinite() {
                    itdata.cnt_remaining_mins -= 1;
                }
                // move center of minorant to cur_y
                minorant.move_center(-1.0, &itdata.nxt_d);
                itdata.nxt_cutvals[index] = itdata.nxt_cutvals[index].max(minorant.constant);
                // add minorant to master problem
                master.add_minorant(index, minorant, primal)?;
            }
        }

        if itdata.cnt_remaining_ubs > 0 || itdata.cnt_remaining_mins > 0 {
            // Haven't received data from all subproblems, yet.
            return Ok(false);
        }

        // All subproblems have been evaluated, do a step.
        let nxt_ub = itdata.nxt_ubs.iter().sum::<Real>();
        let descent_bnd = Self::get_descent_bound(self.params.acceptance_factor, &self.data);

        self.data.nxt_val = nxt_ub;
        self.data.new_cutval = itdata.nxt_cutvals.iter().sum::<Real>();

        debug!("Step");
        debug!("  cur_val    ={}", self.data.cur_val);
        debug!("  nxt_mod    ={}", self.data.nxt_mod);
        debug!("  nxt_ub     ={}", nxt_ub);
        debug!("  descent_bnd={}", descent_bnd);

        let step;
        if self.data.cur_val.is_infinite() {
            // This is the first evaluation. We effectively get
            // the function value at the current center but
            // we do not have a model estimate yet. Hence, we do not know
            // a good guess for the weight.
            step = Step::Descent;
            self.data.cur_val = nxt_ub;
            self.data.cur_weight = Real::infinity();
            master.set_weight(1.0)?;

            debug!("First Step");
            debug!("  cur_val={}", self.data.cur_val);
            debug!("  cur_y={}", self.data.cur_y);
        } else if nxt_ub <= descent_bnd {
            step = Step::Descent;
            self.cnt_descent += 1;

            // Note that we must update the weight *before* we
            // change the internal data, so the old information
            // that caused the descent step is still available.
            self.data.cur_weight = self.weighter.descent_weight(&self.data);
            self.data.cur_y = itdata.nxt_y.as_ref().clone();
            self.data.cur_val = nxt_ub;

            master.move_center(1.0, itdata.nxt_d.clone())?;
            master.set_weight(self.data.cur_weight)?;

            debug!("Descent Step");
            debug!("  dir ={}", itdata.nxt_d);
            debug!("  newy={}", self.data.cur_y);
        } else {
            step = Step::Null;
            self.cnt_null += 1;
            self.data.cur_weight = self.weighter.null_weight(&self.data);
            master.set_weight(self.data.cur_weight)?;
        }

        Self::show_info(
            &self.start_time,
            step,
            &self.data,
            self.cnt_descent,
            self.cnt_null,
            itdata.cnt_updates,
        );
        itdata.cnt_iter += 1;

        if itdata.cnt_iter >= itdata.max_iter {
            return Ok(true);
        }

        master.solve(self.data.cur_val)?;

        Ok(false)
    }

    fn handle_master_response(
        &mut self,
        master_res: MasterResponse,
        itdata: &mut IterData,
    ) -> Result<bool, Error<P::Err>> {
        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;
        itdata.cnt_updates = master_res.cnt_updates;

        // If this is the very first solution of the model,
        // we use its result as to make a good guess for the initial weight
        // of the proximal term and resolve.
        if self.data.cur_weight.is_infinite() {
            self.data.cur_weight = self.weighter.initial_weight(&self.data);
            master.set_weight(self.data.cur_weight)?;
            master.solve(self.data.cur_val)?;
            return Ok(false);
        }

        if self.terminator.terminate(&self.data) {
            Self::show_info(
                &self.start_time,
                Step::Term,
                &self.data,
                self.cnt_descent,
                self.cnt_null,
                itdata.cnt_updates,
            );
            info!("Termination criterion satisfied");
            return Ok(true);
        }

        // Compress bundle
        master.compress()?;

        // Compute new candidate.
        let mut next_y = dvec![];
        itdata.nxt_d = Arc::new(master_res.nxt_d);
        next_y.add(&self.data.cur_y, &itdata.nxt_d);
        itdata.nxt_y = Arc::new(next_y);

        // Reset evaluation data.
        itdata.nxt_ubs.clear();
        itdata.nxt_ubs.resize(self.problem.num_subproblems(), Real::infinity());
        itdata.cnt_remaining_ubs = self.problem.num_subproblems();
        itdata.nxt_cutvals.clear();
        itdata
            .nxt_cutvals
            .resize(self.problem.num_subproblems(), -Real::infinity());
        itdata.cnt_remaining_mins = self.problem.num_subproblems();

        // Start evaluation of all subproblems at the new candidate.
        let client_tx = self.client_tx.as_ref().ok_or(Error::NotInitialized)?;
        for i in 0..self.problem.num_subproblems() {
            self.problem
                .evaluate(i, itdata.nxt_y.clone(), i, client_tx.clone())
                .map_err(Error::Evaluation)?;
        }
        Ok(false)
    }

    /// Return the bound the function value must be below of to enforce a descent step.
    ///
    /// If the oracle guarantees that $f(\bar{y}) \le$ this bound, the
    /// bundle method will perform a descent step.