RsBundle  Check-in [9198cc93a8]

[package]
name = "bundle"
version = "0.7.0-dev"
edition = "2018"
authors = ["Frank Fischer <frank-fischer@shadow-soft.de>"]

[features]
default = []
blas = ["rs-blas", "openblas-src"]

[dependencies]
either = "^1"
itertools = "^0.8"
libc = "^0.2.6"
log = "^0.4"
c_str_macro = "^1.0"
cplex-sys = "^0.5"
crossbeam = "^0.7"
threadpool = "^1.7"

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see  <http://www.gnu.org/licenses/>
 */



use better_panic;
use bundle::{self, dvec};
use env_logger;
use env_logger::fmt::Color;
use log::{debug, Level};
use rustop::opts;
use std::error::Error;
use std::io::Write;
use std::sync::Arc;
use std::thread;

use bundle::problem::{FirstOrderProblem as ParallelProblem, ResultSender};
use bundle::solver::sync::{DefaultSolver, NoBundleSolver};
use bundle::{DVector, Minorant, Real};

            b: [-12.0, -10.0],
            c: 3.0,
        }
    }
}

impl ParallelProblem for QuadraticProblem {
    type Err = Box<dyn Error + Send>;
    type Primal = ();

    fn num_variables(&self) -> usize {
        2
    }

    fn num_subproblems(&self) -> usize {

            )
            .unwrap();
        });
        Ok(())
    }
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    better_panic::install();
    env_logger::builder()
        .format(|buf, record| {
            let mut style = buf.style();
            let color = match record.level() {
                Level::Error | Level::Warn => Color::Red,
                Level::Trace => Color::Blue,

pub mod unconstrained;
use self::unconstrained::UnconstrainedMasterProblem;

use super::MasterProblem;
pub use super::SubgradientExtension;
use crate::{DVector, Minorant, Real};

use either::Either;
use itertools::multizip;
use log::debug;
use std::f64::{EPSILON, INFINITY, NEG_INFINITY};

/**
 * Turn unconstrained master problem into box-constrained one.
 *

        self.master.weight()
    }

    fn set_weight(&mut self, weight: Real) -> Result<(), Self::Err> {
        self.master.set_weight(weight)
    }

    fn add_vars<S>(
        &mut self,
        bounds: &[(Option<usize>, Real, Real)],
        extend_subgradient: S,
    ) -> Result<(), Either<Self::Err, S::Err>>
    where
        S: SubgradientExtension<Self::MinorantIndex>,
    {
        if !bounds.is_empty() {
            for (index, l, u) in bounds.iter().filter_map(|v| v.0.map(|i| (i, v.1, v.2))) {
                self.lb[index] = l;
                self.ub[index] = u;
            }
            self.lb.extend(bounds.iter().filter(|v| v.0.is_none()).map(|x| x.1));
            self.ub.extend(bounds.iter().filter(|v| v.0.is_none()).map(|x| x.2));

pub mod cpx;
pub mod minimal;

use crate::{DVector, Minorant, Real};

pub use super::SubgradientExtension;

use either::Either;
use std::error::Error;

/**
 * Trait for master problems without box constraints.
 *
 * Implementors of this trait are supposed to solve quadratic
 * optimization problems of the form

    fn add_minorant(&mut self, fidx: usize, minorant: Minorant) -> Result<Self::MinorantIndex, Self::Err>;

    /// Add or move some variables.
    ///
    /// The variables in `changed` have been changed, so the subgradient
    /// information must be updated. Furthermore, `nnew` new variables
    /// are added.
    fn add_vars<S>(
        &mut self,
        nnew: usize,
        changed: &[usize],
        extend_subgradient: S,
    ) -> Result<(), Either<Self::Err, S::Err>>
    where
        S: SubgradientExtension<Self::MinorantIndex>;

    /// Solve the master problem.
    fn solve(&mut self, eta: &DVector, fbound: Real, augbound: Real, relprec: Real) -> Result<(), Self::Err>;

    /// Return the current dual optimal solution.
    fn dualopt(&self) -> &DVector;

use super::{SubgradientExtension, UnconstrainedMasterProblem};
use crate::{Aggregatable, DVector, Minorant, Real};

use c_str_macro::c_str;
use cplex_sys as cpx;
use cplex_sys::trycpx;
use either::Either;
use log::{debug, warn};

use std;
use std::f64::NEG_INFINITY;
use std::iter::{once, repeat};
use std::ops::{Deref, DerefMut};
use std::os::raw::{c_char, c_int};
use std::ptr;
use std::sync::Arc;

#[derive(Debug)]
pub enum CplexMasterError {
    Cplex(cpx::CplexError),

    NoMinorants,
}

impl std::fmt::Display for CplexMasterError {
    fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
        use CplexMasterError::*;
        match self {
            Cplex(err) => err.fmt(fmt),

            NoMinorants => write!(fmt, "Master problem contains no minorants"),
        }
    }
}

impl std::error::Error for CplexMasterError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        use CplexMasterError::*;
        match self {
            Cplex(err) => Some(err),

            NoMinorants => None,
        }
    }
}

impl From<cpx::CplexError> for CplexMasterError {
    fn from(err: cpx::CplexError) -> Self {

        // store minorant
        self.minorants[fidx].push(MinorantInfo { minorant, index });
        self.opt_mults[fidx].push(0.0);

        Ok(index)
    }

    fn add_vars<S>(
        &mut self,
        nnew: usize,
        changed: &[usize],
        mut extend_subgradient: S,
    ) -> std::result::Result<(), Either<CplexMasterError, S::Err>>
    where
        S: SubgradientExtension<Self::MinorantIndex>,

    {
        debug_assert!(!self.minorants[0].is_empty());
        if changed.is_empty() && nnew == 0 {
            return Ok(());
        }
        let noldvars = self.minorants[0][0].linear.len();
        let nnewvars = noldvars + nnew;

        let mut changedvars = vec![];
        changedvars.extend_from_slice(changed);
        changedvars.extend(noldvars..nnewvars);
        for (fidx, mins) in self.minorants.iter_mut().enumerate() {
            for m in &mut mins[..] {
                let new_subg = extend_subgradient
                    .subgradient_extension(fidx, m.index, &changedvars)
                    .map_err(Either::Right)?;
                for (&j, &g) in changed.iter().zip(new_subg.iter()) {
                    m.linear[j] = g;
                }
                m.linear.extend_from_slice(&new_subg[changed.len()..]);
            }
        }

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

use super::{SubgradientExtension, UnconstrainedMasterProblem};
use crate::{Aggregatable, DVector, Minorant, Real};

use either::Either;
use log::debug;

use std::error::Error;
use std::f64::NEG_INFINITY;
use std::fmt;

/// Minimal master problem error.
#[derive(Debug)]
pub enum MinimalMasterError {
    NoMinorants,
    MaxMinorants { subproblem: usize },

}

impl fmt::Display for MinimalMasterError {
    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
        use self::MinimalMasterError::*;
        match self {
            MaxMinorants { subproblem } => write!(
                fmt,
                "The minimal master problem allows at most two minorants (subproblem: {})",
                subproblem
            ),
            NoMinorants => write!(fmt, "The master problem does not contain a minorant"),

        }
    }
}

impl Error for MinimalMasterError {}









/**
 * A minimal master problem with only two minorants.
 *
 * This is the simplest possible master problem for bundle methods. It
 * has only two minorants and only one function model. The advantage
 * is that this model can be solved explicitely and very quickly, but

                }
                Ok(2 * fidx + 1)
            }
            _ => unreachable!("Invalid number of minorants in subproblem {}", fidx),
        }
    }

    fn add_vars<S>(
        &mut self,
        nnew: usize,
        changed: &[usize],
        mut extend_subgradient: S,
    ) -> Result<(), Either<Self::Err, S::Err>>
    where
        S: SubgradientExtension<Self::MinorantIndex>,
    {
        if self.num_subproblems_with_1 == 0 {
            return Ok(());
        }

        let noldvars = self.minorants[0][self
            .num_minorants_of
            .iter()
            .position(|&n| n > 0)
            .ok_or(MinimalMasterError::NoMinorants)
            .map_err(Either::Left)?]
        .linear
        .len();
        let mut changedvars = vec![];
        changedvars.extend_from_slice(changed);
        changedvars.extend(noldvars..noldvars + nnew);

        for fidx in 0..self.num_subproblems {
            for i in 0..self.num_minorants_of[fidx] {
                let new_subg = extend_subgradient
                    .subgradient_extension(fidx, 2 * fidx + i, &changedvars)
                    .map_err(Either::Right)?;
                let m = &mut self.minorants[i][fidx];
                for (&j, &g) in changed.iter().zip(new_subg.iter()) {
                    m.linear[j] = g;
                }
                m.linear.extend_from_slice(&new_subg[changed.len()..]);
            }
        }

pub mod boxed;
pub use self::boxed::BoxedMasterProblem;

pub(crate) mod primalmaster;

use crate::{DVector, Minorant, Real};
use either::Either;
use std::error::Error;
use std::result::Result;

/// Callback for subgradient extensions.
pub trait SubgradientExtension<I> {
    type Err;

    fn subgradient_extension(
        &mut self,
        fidx: usize,
        minorant_index: I,
        changed: &[usize],
    ) -> Result<DVector, Self::Err>;
}

impl<F, I, E> SubgradientExtension<I> for F
where
    F: FnMut(usize, I, &[usize]) -> Result<DVector, E>,
{
    type Err = E;

    fn subgradient_extension(
        &mut self,
        fidx: usize,
        minorant_index: I,
        changed: &[usize],
    ) -> Result<DVector, Self::Err> {
        (self)(fidx, minorant_index, changed)
    }
}

/// Trait for master problems of a proximal bundle methods.
///
/// Note that solvers are allowed to create multiple master problems potentially
/// running them in different threads. Because of this they must implement `Send
/// + 'static`, i.e. they must be quite self-contained and independent from
/// everything else.

    /// Return the current number of inner iterations.
    fn cnt_updates(&self) -> usize;

    /// Add or move some variables with bounds.
    ///
    /// If an index is specified, existing variables are moved,
    /// otherwise new variables are generated.
    fn add_vars<S>(
        &mut self,
        bounds: &[(Option<usize>, Real, Real)],
        extend_subgradient: S,
    ) -> Result<(), Either<Self::Err, S::Err>>
    where
        S: SubgradientExtension<Self::MinorantIndex>;

    /// Add a new minorant to the model.
    ///
    /// The function returns a unique (among all minorants of all
    /// subproblems) index of the minorant. This index must remain
    /// valid until the minorant is aggregated.
    fn add_minorant(&mut self, fidx: usize, minorant: Minorant) -> Result<Self::MinorantIndex, Self::Err>;

//! A wrapper around master problems to handle primal information.

use super::MasterProblem;
use crate::problem::SubgradientExtender;
use crate::{Aggregatable, Minorant, Real};

use either::Either;
use std::collections::HashMap;
use std::ops::{Deref, DerefMut};

/// A wrapper around `MasterProblem` to handle primal information.
///
/// For each minorant there can be an associated (generating) primal data.
/// Typically this data arises from a Lagrangian Relaxation approach. The primal

        }
    }

    pub fn add_vars<E>(
        &mut self,
        vars: Vec<(Option<usize>, Real, Real)>,
        sgext: &mut SubgradientExtender<P, E>,
    ) -> Result<(), Either<M::Err, E>> {



        let primals = &self.primals;
        self.master.add_vars(&vars, |fidx, minidx, vars: &[usize]| {
            sgext(
                fidx,
                primals
                    .get(&minidx)
                    .expect("Extension for non-existing primal requested"),
                vars,
            )

        })
    }

    pub fn add_minorant(&mut self, fidx: usize, minorant: Minorant, primal: P) -> Result<(), M::Err> {
        let index = self.master.add_minorant(fidx, minorant)?;
        let old = self.primals.insert(index, primal);
        debug_assert!(old.is_none(), "Minorant index already used");

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::result::Result<(), std::fmt::Error> {
        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.into()),
            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<P>> + Default,
    W: Weighter<SolverData<P>> + 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.nxt_d = Arc::new(dvec![0.0; num_variables]);
        self.data.nxt_y = Arc::new(dvec![]);
        self.data.update_rx = None;
        self.data.need_update = 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 {
            let client_index;
            let master_index;

    /// The function returns
    ///   - `Ok(true)` if the final iteration count has been reached,
    ///   - `Ok(false)` if the final iteration count has not been reached,
    ///   - `Err(_)` on error.
    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;
                }

    /// Finally the master problem starts the evaluation of all subproblems at
    /// the new candidate.
    ///
    /// Return values
    ///   - `Ok(true)` if the termination criterion has been satisfied,
    ///   - `Ok(false)` if the termination criterion has not been satisfied,
    ///   - `Err(_)` on error.
    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;

    /// variables. This method starts the update process.
    ///
    /// Return values
    ///   - `Ok(true)` if a new update process has been started,
    ///   - `Ok(false)` if there is already a running update process (only one
    ///     is allowed at the same time),
    ///   - `Err(_)` on error.
    fn update_problem(&mut self, step: Step) -> Result<bool, P, M> {
        // only one update may be running at the same time
        if self.data.update_rx.is_some() {
            return Ok(false);
        }

        // Ok, we are doing a new update now ...
        self.data.need_update = false;

    }

    /// Handles an update response `update` of the problem.
    ///
    /// The method is called if the problem informs the solver about a change of
    /// the problem, e.g. adding a new variable. This method updates the other
    /// parts of the solver, e.g. the master problem, about the modification.
    fn handle_update_response(&mut self, update: Update<usize, P::Primal, P::Err>) -> Result<(), P, M> {
        match update {
            Update::AddVariables { bounds, sgext, .. } => {
                if !bounds.is_empty() {
                    // add new variables
                    self.master_proc
                        .as_mut()
                        .unwrap()

            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)?)
    }
}

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

//! Asynchronous process solving a master problem.

use crossbeam::channel::{unbounded as channel, Receiver, RecvError, SendError, Sender};
use either::Either;
use log::{debug, warn};
use std::sync::Arc;
use threadpool::ThreadPool;

use crate::master::primalmaster::PrimalMaster;
use crate::master::MasterProblem;
use crate::problem::{FirstOrderProblem, SubgradientExtender};
use crate::{DVector, Minorant, Real};

#[derive(Debug)]
pub enum Error<MErr, PErr> {


    /// The communication channel for sending requests has been disconnected.
    DisconnectedSender,
    /// The communication channel for sending responds has been disconnected.
    DisconnectedReceiver,
    /// An error occurred when computing an aggregated primal.
    Aggregation(MErr),
    /// An error occurred during the subgradient extension.
    SubgradientExtension(PErr),
    /// An error has been raised by the underlying master problem solver.
    Master(MErr),
}

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) -> Result<(), std::fmt::Error> {
        use Error::*;
        match self {

            DisconnectedSender => writeln!(fmt, "Communication channel to master process has been disconnected"),
            DisconnectedReceiver => writeln!(fmt, "Communication channel from master process has been disconnected"),
            Aggregation(err) => writeln!(fmt, "Computation of primal aggregate failed: {}", err),
            SubgradientExtension(err) => writeln!(fmt, "Subgradient extension failed: {}", err),
            Master(err) => writeln!(fmt, "Error in master problem solver: {}", err),
        }
    }
}

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 {

            DisconnectedSender => None,
            DisconnectedReceiver => None,
            Aggregation(err) => Some(err),
            SubgradientExtension(err) => Some(err),
            Master(err) => Some(err),
        }
    }
}

impl<MErr, PErr, T> From<SendError<T>> for Error<MErr, PErr>
where
    T: Send + 'static,
{
    fn from(_err: SendError<T>) -> Error<MErr, PErr> {
        Error::DisconnectedSender
    }
}

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

/// Configuration information for setting up a master problem.
pub struct MasterConfig {
    /// The number of subproblems.

    pub cnt_updates: usize,
}

type ToMasterSender<P, M> = Sender<MasterTask<<P as FirstOrderProblem>::Primal, <P as FirstOrderProblem>::Err, M>>;

type ToMasterReceiver<P, M> = Receiver<MasterTask<<P as FirstOrderProblem>::Primal, <P as FirstOrderProblem>::Err, M>>;

type MasterSender<MErr, PErr> = Sender<Result<MasterResponse, Error<MErr, PErr>>>;

pub type MasterReceiver<MErr, PErr> = Receiver<Result<MasterResponse, Error<MErr, PErr>>>;

pub struct MasterProcess<P, M>
where
    P: FirstOrderProblem,
    M: MasterProblem,
{
    /// The channel to transmit new tasks to the master problem.
    tx: ToMasterSender<P, M>,

    /// The channel to receive solutions from the master problem.
    pub rx: MasterReceiver<M::Err, P::Err>,

    phantom: std::marker::PhantomData<M>,
}

impl<P, M> MasterProcess<P, M>
where
    P: FirstOrderProblem,
    P::Primal: Send + 'static,
    P::Err: Send + 'static,
    M: MasterProblem + Send + 'static,
    M::MinorantIndex: std::hash::Hash,
    M::Err: Send + 'static,
{
    pub fn start(master: M, master_config: MasterConfig, threadpool: &mut ThreadPool) -> Self {
        // Create a pair of communication channels.
        let (to_master_tx, to_master_rx) = channel();

    }

    /// Add new variables to the master problem.
    pub fn add_vars(
        &mut self,
        vars: Vec<(Option<usize>, Real, Real)>,
        sgext: Box<SubgradientExtender<P::Primal, P::Err>>,
    ) -> Result<(), Error<M::Err, P::Err>>
    where
        P::Err: 'static,
    {
        Ok(self.tx.send(MasterTask::AddVariables(vars, sgext))?)
    }

    /// Add a new minorant to the master problem model.
    ///
    /// This adds the specified `minorant` with associated `primal` data to the
    /// model of subproblem `i`.
    pub fn add_minorant(
        &mut self,
        i: usize,
        minorant: Minorant,
        primal: P::Primal,
    ) -> Result<(), Error<M::Err, P::Err>> {
        Ok(self.tx.send(MasterTask::AddMinorant(i, minorant, primal))?)
    }

    /// Move the center of the master problem.
    ///
    /// This moves the master problem's center in direction $\\alpha \\cdot d$.
    pub fn move_center(&mut self, alpha: Real, d: Arc<DVector>) -> Result<(), Error<M::Err, P::Err>> {
        Ok(self.tx.send(MasterTask::MoveCenter(alpha, d))?)
    }

    /// Solve the master problem.
    ///
    /// The current function value in the center `center_value`.
    /// Once the master problem is solved the process will send a
    /// [`MasterResponse`] message to the `tx` channel.
    pub fn solve(&mut self, center_value: Real) -> Result<(), Error<M::Err, P::Err>> {
        Ok(self.tx.send(MasterTask::Solve { center_value })?)
    }

    /// Compresses the model.
    pub fn compress(&mut self) -> Result<(), Error<M::Err, P::Err>> {
        Ok(self.tx.send(MasterTask::Compress)?)
    }

    /// Sets the new weight of the proximal term in the master problem.
    pub fn set_weight(&mut self, weight: Real) -> Result<(), Error<M::Err, P::Err>> {
        Ok(self.tx.send(MasterTask::SetWeight { weight })?)
    }

    /// Get the current aggregated primal for a certain subproblem.
    pub fn get_aggregated_primal(&self, subproblem: usize) -> Result<P::Primal, Error<M::Err, P::Err>> {
        let (tx, rx) = channel();
        self.tx.send(MasterTask::GetAggregatedPrimal { subproblem, tx })?;
        rx.recv()?.map_err(Error::Aggregation)
    }

    /// The main loop of the master process.
    fn master_main(
        master: M,
        master_config: MasterConfig,
        tx: &mut MasterSender<M::Err, P::Err>,
        rx: ToMasterReceiver<P, M>,
        subgradient_extender: &mut Option<Box<SubgradientExtender<P::Primal, P::Err>>>,
    ) -> Result<(), Error<M::Err, P::Err>> {
        let mut master = PrimalMaster::<_, P::Primal>::new(master);

        // Initialize the master problem.
        master
            .set_num_subproblems(master_config.num_subproblems)
            .map_err(Error::Master)?;
        master
            .set_vars(
                master_config.num_vars,
                master_config.lower_bounds,
                master_config.upper_bounds,
            )
            .map_err(Error::Master)?;

        // The main iteration: wait for new tasks.
        for m in rx {
            match m {
                MasterTask::AddVariables(vars, mut sgext) => {
                    debug!("master: add {} variables to the subproblem", vars.len());
                    master.add_vars(vars, &mut sgext).map_err(|err| match err {
                        Either::Left(err) => Error::Master(err),
                        Either::Right(err) => Error::SubgradientExtension(err),
                    })?;
                    *subgradient_extender = Some(sgext);
                }
                MasterTask::AddMinorant(i, mut m, primal) => {
                    debug!("master: add minorant to subproblem {}", i);
                    // It may happen the number new minorant belongs to an earlier evaluation
                    // with less variables (i.e. new variables have been added
                    // after the start of the evaluation but before the new
                    // minorant is added, i.e. now). In this case we must add
                    // extend the minorant accordingly.
                    if m.linear.len() < master.num_variables() {
                        if let Some(ref mut sgext) = subgradient_extender {
                            let newinds = (m.linear.len()..master.num_variables()).collect::<Vec<_>>();
                            let new_subg = sgext(i, &primal, &newinds)
                                .map_err(|err| Error::<M::Err, P::Err>::SubgradientExtension(err))?;
                            m.linear.extend(new_subg);
                        }
                    }
                    master.add_minorant(i, m, primal).map_err(Error::Master)?;
                }
                MasterTask::MoveCenter(alpha, d) => {
                    debug!("master: move center");

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)?)
    }
}