RsBundle  Diff

 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see  <http://www.gnu.org/licenses/>
 */

//! An asynchronous parallel bundle solver.

#[cfg(feature = "crossbeam")]
use rs_crossbeam::channel::{unbounded as channel, RecvError};
#[cfg(not(feature = "crossbeam"))]
use std::sync::mpsc::{channel, RecvError};

use log::{debug, info, warn};
use num_cpus;
use num_traits::Float;
use std::sync::Arc;
use std::time::Instant;
use threadpool::ThreadPool;

use crate::{DVector, Real};

use super::channels::{
    ChannelResultSender, ChannelUpdateSender, ClientReceiver, ClientSender, EvalResult, Message, Update,
};
use super::masterprocess::{self, MasterConfig, MasterProcess, MasterResponse, Response};
use crate::master::{Builder as MasterBuilder, MasterProblem};
use crate::problem::{FirstOrderProblem, UpdateState};
use crate::terminator::{StandardTerminatable, StandardTerminator, Terminator};
use crate::weighter::{HKWeightable, HKWeighter, Weighter};

/// The default iteration limit.
pub const DEFAULT_ITERATION_LIMIT: usize = 10_000;

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







#[derive(Debug, Clone)]
pub struct Parameters {
    /// The descent step acceptance factors, must be in (0,1).
    ///
    /// The default value is 0.1.
    acceptance_factor: Real,
}

    /// The algorithm data.
    data: SolverData,

    /// The master problem process.
    master_proc: Option<MasterProcess<P, M::MasterProblem>>,

    /// The channel to receive the evaluation results from subproblems.
    client_tx: Option<ClientSender<usize, P, M::MasterProblem>>,

    /// The channel to receive the evaluation results from subproblems.
    client_rx: Option<ClientReceiver<usize, P, M::MasterProblem>>,

    /// Number of descent steps.
    cnt_descent: usize,

    /// Number of null steps.
    cnt_null: usize,

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

        let (tx, rx) = channel();
        self.client_tx = Some(tx.clone());
        self.client_rx = Some(rx);

        let master_config = MasterConfig {
            num_subproblems: m,
            num_vars: n,
            lower_bounds: self.problem.lower_bounds().map(DVector),
            upper_bounds: self.problem.upper_bounds().map(DVector),

            return Err(Error::Dimension("upper bounds".to_string()));
        }

        debug!("Start master process");
        self.master_proc = Some(MasterProcess::start(
            self.master.build().map_err(Error::BuildMaster)?,
            master_config,
            tx,
            &mut self.threadpool,
        ));

        debug!("Initial problem evaluation");
        // We need an initial evaluation of all oracles for the first center.
        let y = Arc::new(self.data.cur_y.clone());
        for i in 0..m {
            self.problem
                .evaluate(
                    i,
                    y.clone(),
                    ChannelResultSender::new(i, self.client_tx.clone().unwrap()),
                )
                .map_err(Error::Evaluation)?;
        }

        debug!("Initialization complete");

        self.start_time = Instant::now();

    /// 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 {

            let msg = self.client_rx.as_ref().ok_or(Error::NotInitialized)?.recv()?;
            match msg {
                Message::Eval(m) => {
                    if self.handle_client_response(m)? {
                        return Ok(false);



                    }
                }


                Message::Update(_) => warn!("Ignore unexpected problem update message from client"),
                Message::Master(mresponse) => {
                    debug!("Receive master response");
                    // Receive result (new candidate) from the master
                    if self.handle_master_response(mresponse)? {
                        return Ok(true);
                    }

                }
            }
        }
    }

    /// Handle a response from a subproblem evaluation.
    ///
    /// The function returns `Ok(true)` if the final iteration count has been reached.

        // 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<P, M::MasterProblem>) -> Result<bool, P, M> {
        match master_res {
            Response::Error(err) => return Err(err.into()),
            Response::Result {
                nxt_mod,
                sgnorm,
                cnt_updates,
                nxt_d,
                ..
            } => {
                self.data.nxt_d = Arc::new(nxt_d);
                self.data.nxt_mod = nxt_mod;
                self.data.sgnorm = sgnorm;
                self.data.cnt_updates = cnt_updates;
            }
        };

        self.data.expected_progress = self.data.cur_val - self.data.nxt_mod;

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

        // 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)?;

        }

        // Compress bundle
        master.compress()?;

        // Compute new candidate.
        let mut next_y = dvec![];

        next_y.add(&self.data.cur_y, &self.data.nxt_d);
        self.data.nxt_y = Arc::new(next_y);

        // Reset evaluation data.
        self.data.nxt_ubs.clear();
        self.data
            .nxt_ubs
            .resize(self.problem.num_subproblems(), Real::infinity());
        self.data.cnt_remaining_ubs = self.problem.num_subproblems();
        self.data.nxt_cutvals.clear();
        self.data
            .nxt_cutvals
            .resize(self.problem.num_subproblems(), -Real::infinity());
        self.data.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,
                    self.data.nxt_y.clone(),
                    ChannelResultSender::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)?;

                            master_proc
                                .get_aggregated_primal(i)
                                .map_err(|_| "get_aggregated_primal".to_string())
                                .expect("Cannot get aggregated primal from master process")
                        })
                        .collect(),
                },
                ChannelUpdateSender::<_, _, _, M::MasterProblem>::new(self.data.cnt_iter, update_tx),
            )
            .map_err(Error::Update)?;

        let mut have_update = false;
        for msg in update_rx {
            if let Message::Update(update) = msg {
                match update {
                    Update::AddVariables { bounds, sgext, .. } => {
                        have_update = true;
                        let mut newvars = Vec::with_capacity(bounds.len());
                        for (lower, upper) in bounds {
                            if lower > upper {
                                return Err(Error::InvalidBounds { lower, upper });

                            master_proc.add_vars(newvars.iter().map(|v| (v.0, v.1, v.2)).collect(), sgext)?;
                        }
                    }
                    Update::Done { .. } => (), // there's nothing to do
                    Update::Error { err, .. } => return Err(Error::Update(err)),
                }
            } else {
                unreachable!("Only update results allowed during update");
            }
        }

        Ok(have_update)
    }

    /// Return the bound the function value must be below of to enforce a descent step.