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.

use crossbeam::channel::{select, unbounded as channel, Receiver, RecvError, Sender};
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::{ChannelSender, ChannelUpdateSender, ClientMessage, EvalResult, Update};
use super::masterprocess::{self, MasterConfig, MasterProcess, MasterResponse};
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.

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

type ClientSender<P> = Sender<ClientMessage<usize, <P as FirstOrderProblem>::Primal, <P as FirstOrderProblem>::Err>>;

type ClientReceiver<P> =
    Receiver<ClientMessage<usize, <P as FirstOrderProblem>::Primal, <P as FirstOrderProblem>::Err>>;

/// Parameters for tuning the solver.
#[derive(Debug, Clone)]
pub struct Parameters {
    /// The descent step acceptance factors, must be in (0,1).
    ///
    /// The default value is 0.1.

    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 => {
                    match msg? {
                        ClientMessage::Eval(m) =>
                            if self.handle_client_response(m)? {
                                return Ok(false);
                            },
                        ClientMessage::Update(_) => {
                            warn!("Ignore unexpected problem update message from client")
                        }
                    }
                },
                recv(self.master_proc.as_ref().ok_or(Error::NotInitialized)?.rx) -> msg => {
                    debug!("Receive master response");
                    // Receive result (new candidate) from the master
                    if self.handle_master_response(msg??)? {
                        return Ok(true);

                        .collect(),
                },
                ChannelUpdateSender::new(self.data.cnt_iter, update_tx),
            )
            .map_err(Error::Update)?;

        let mut have_update = false;
        for msg in update_rx {
            if let ClientMessage::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 });
                            }
                            let value = if lower > 0.0 {
                                lower
                            } else if upper < 0.0 {
                                upper
                            } else {
                                0.0
                            };
                            //self.bounds.push((lower, upper));
                            newvars.push((None, lower - value, upper - value, value));
                        }
                        if !newvars.is_empty() {
                            // modify moved variables
                            for (index, val) in newvars.iter().filter_map(|v| v.0.map(|i| (i, v.3))) {
                                self.data.cur_y[index] = val;
                            }

                            // add new variables
                            self.data
                                .cur_y
                                .extend(newvars.iter().filter(|v| v.0.is_none()).map(|v| v.3));

                            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!("No evaluation result during update");
            }
        }

        Ok(have_update)
    }

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