RsBundle  Changes On Branch 43abfc4c67ef4e09

use std::io::Write as _;
use std::sync::Arc;

use env_logger::{self, fmt::Color};
use ordered_float::NotNan;
use threadpool::ThreadPool;

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

const Nfac: usize = 3;
const Ncus: usize = 5;
const F: [Real; Nfac] = [1000.0, 1000.0, 1000.0];
const CAP: [Real; Nfac] = [500.0, 500.0, 500.0];

        self.pool = Some(pool)
    }

    fn stop(&mut self) {
        self.pool.take();
    }

    fn evaluate<I>(
        &mut self,
        i: usize,
        y: Arc<DVector>,
        index: I,
        tx: ResultSender<I, Self::Primal, Self::Err>,
    ) -> Result<(), Self::Err>
    fn evaluate<S>(&mut self, i: usize, y: Arc<DVector>, tx: S) -> Result<(), Self::Err>
    where
        I: Send + Copy + 'static,
        S: ResultSender<Self> + 'static,
    {
        if self.pool.is_none() {
            self.start()
        }
        let y = y.clone();
        self.pool.as_ref().unwrap().execute(move || {
            let res = if i < Nfac {
                solve_facility_problem(i, &y)
            } else {
                solve_customer_problem(i - Nfac, &y)
            };
            match res {
                Ok((value, minorant, primal)) => {
                    if tx.send(Ok(EvalResult::ObjectiveValue { index, value })).is_err() {
                    if tx.objective(value).is_err() {
                        return;
                    }
                    if tx
                        .send(Ok(EvalResult::Minorant {
                            index,
                            minorant,
                    if tx.minorant(minorant, primal).is_err() {
                            primal,
                        }))
                        .is_err()
                    {
                        return;
                    }
                }
                Err(err) => if tx.send(Err(err)).is_err() {},
                Err(err) => if tx.error(err).is_err() {},
            };
        });
        Ok(())
    }
}

fn show_primals<F>(f: F) -> Result<(), Box<dyn Error>>

use env_logger::fmt::Color;
use log::{debug, Level};
use rustop::opts;
use std::io::Write;
use std::sync::Arc;
use std::thread;

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

#[derive(Clone)]
struct QuadraticProblem {
    a: [[Real; 2]; 2],
    b: [Real; 2],

        1
    }

    fn start(&mut self) {}

    fn stop(&mut self) {}

    fn evaluate<I>(
        &mut self,
        fidx: usize,
        x: Arc<DVector>,
        index: I,
        tx: ResultSender<I, Self::Primal, Self::Err>,
    ) -> Result<(), Self::Err>
    fn evaluate<S>(&mut self, fidx: usize, x: Arc<DVector>, tx: S) -> Result<(), Self::Err>
    where
        I: Send + Copy + 'static,
        S: ResultSender<Self> + 'static,
    {
        let x = x.clone();
        let p = self.clone();
        thread::spawn(move || {
            assert_eq!(fidx, 0);
            let mut objective = p.c;
            let mut g = dvec![0.0; 2];

            for i in 0..2 {
                g[i] += (0..2).map(|j| p.a[i][j] * x[j]).sum::<Real>();
                objective += x[i] * (g[i] + p.b[i]);
                g[i] = 2.0 * g[i] + p.b[i];
            }

            debug!("Evaluation at {:?}", x);
            debug!("  objective={}", objective);
            debug!("  subgradient={}", g);

            tx.send(Ok(EvalResult::ObjectiveValue {
                index,
                value: objective,
            }))
            .unwrap();
            tx.send(Ok(EvalResult::Minorant {
            tx.objective(objective).unwrap();
            tx.minorant(
                index,
                minorant: Minorant {
                Minorant {
                    constant: objective,
                    linear: g,
                },
                primal: (),
                (),
            }))
            )
            .unwrap();
        });
        Ok(())
    }
}

fn main() -> Result<(), Box<dyn Error>> {

//
// 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 crate::mcf;
use crate::problem::{
    EvalResult, FirstOrderProblem as ParallelProblem, ResultSender, Update as ParallelUpdate, UpdateSender,
    FirstOrderProblem as ParallelProblem, ResultSender, Update as ParallelUpdate, UpdateSender,
    UpdateState as ParallelUpdateState,
};
use crate::{DVector, Minorant, Real};

use itertools::izip;
use log::{debug, warn};
use num_traits::Float;

        self.pool = Some(pool);
    }

    fn stop(&mut self) {
        self.pool.take();
    }

    fn evaluate<I>(
    fn evaluate<S>(&mut self, i: usize, y: Arc<DVector>, tx: S) -> Result<()>
        &mut self,
        i: usize,
        y: Arc<DVector>,
        index: I,
        tx: ResultSender<I, Self::Primal, Self::Err>,
    ) -> Result<()>
    where
        I: Send + Copy + 'static,
        S: ResultSender<Self> + 'static,
    {
        if self.pool.is_none() {
            self.start()
        }
        let y = y.clone();
        let sub = self.subs[i].clone();
        // Attention: `y` might be shorter than the set of active constraints
        // because evaluation and problem update are not synchronized, i.e. the
        // evaluation may still use an older model with less constraints.
        let active_constraints = self.active_constraints.read().unwrap()[0..y.len()].to_vec();
        self.pool
            .as_ref()
            .unwrap()
            .execute(move || match sub.write().unwrap().evaluate(&y, active_constraints) {
                Ok((objective, subg, primal)) => {
                    tx.send(Ok(EvalResult::ObjectiveValue {
                        index,
                        value: objective,
                    }))
                    .unwrap();
                    tx.send(Ok(EvalResult::Minorant {
                    tx.objective(objective).unwrap();
                    tx.minorant(
                        index,
                        minorant: Minorant {
                        Minorant {
                            constant: objective,
                            linear: subg,
                        },
                        primal,
                    }))
                    )
                    .unwrap();
                }
                Err(err) => tx.send(Err(err)).unwrap(),
                Err(err) => tx.error(err).unwrap(),
            });
        Ok(())
    }

    fn update<I, U>(&mut self, state: U, index: I, tx: UpdateSender<I, Self::Primal, Self::Err>) -> Result<()>
    where
        U: ParallelUpdateState<Self::Primal>,

//! An asynchronous first-order oracle.

use crate::{Aggregatable, DVector, Minorant, Real};
use crossbeam::channel::Sender;
use std::sync::Arc;

/// Evaluation result.
/// Channel to send evaluation results to.
///
/// The result of an evaluation is new information to be made
/// available to the solver and the master problem. There are
pub trait ResultSender<P>: Send
where
/// essentially two types of information:
///
///    1. The (exact) function value of a sub-function at some point.
///    2. A minorant of some sub-function.
#[derive(Debug)]
pub enum EvalResult<I, P> {
    /// The objective value at some point.
    ObjectiveValue { index: I, value: Real },
    /// A minorant with an associated primal.
    Minorant { index: I, minorant: Minorant, primal: P },
}

    P: FirstOrderProblem,
{
    type Err: std::error::Error + Send;

    /// Send a new objective `value`.
    fn objective(&self, value: Real) -> Result<(), Self::Err>;

    /// Send a new `minorant` with associated `primal`.
    fn minorant(&self, minorant: Minorant, primal: P::Primal) -> Result<(), Self::Err>;

    /// Send an error message.
    fn error(&self, err: P::Err) -> Result<(), Self::Err>;
}
/// Channel to send evaluation results to.
pub type ResultSender<I, P, E> = Sender<Result<EvalResult<I, P>, E>>;

/// Problem update information.
///
/// The solver calls the `update` method of the problem regularly.
/// This method can modify the problem by adding (or moving)
/// variables. The possible updates are encoded in this type.
pub enum Update<I, P, E> {

    /// Start the evaluation of the i^th subproblem at the given point.
    ///
    /// The results of the evaluation should be passed to the provided channel.
    /// In order to work correctly, the results must contain (an upper bound on)
    /// the objective value at $y$ as well as at least one subgradient centered
    /// at $y$ eventually.
    fn evaluate<I: Send + Copy + 'static>(
        &mut self,
        i: usize,
        y: Arc<DVector>,
        index: I,
        tx: ResultSender<I, Self::Primal, Self::Err>,
    ) -> Result<(), Self::Err>
    fn evaluate<S>(&mut self, i: usize, y: Arc<DVector>, tx: S) -> Result<(), Self::Err>
    where
        I: Send + 'static;
        S: ResultSender<Self> + 'static,
        Self: Sized;

    /// Called to update the problem.
    ///
    /// This method is called regularly by the solver. The problem should send problem update
    /// information (e.g. adding new variables) to the provided channel.
    ///
    /// The updates might be generated asynchronously.

pub mod sync;
pub use sync::{DefaultSolver, NoBundleSolver};

pub mod asyn;

mod masterprocess;

mod channels;

use std::iter::repeat;
use std::sync::Arc;
use std::time::Instant;
use threadpool::ThreadPool;

use crate::{DVector, Real};

use super::channels::{ChannelSender, EvalResult};
use super::masterprocess::{self, MasterConfig, MasterProcess, MasterResponse};
use crate::master::{Builder as MasterBuilder, MasterProblem};
use crate::problem::{EvalResult, FirstOrderProblem, Update, UpdateState};
use crate::problem::{FirstOrderProblem, Update, 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;

/// The default solver.

    /// 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: Into<Box<dyn std::error::Error + Sync + Send>> + 'static,
    P::Err: Into<Box<dyn std::error::Error + Sync + Send>> + 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

        ));

        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(), i, self.client_tx.clone().unwrap())
                .evaluate(i, y.clone(), ChannelSender::new(i, self.client_tx.clone().unwrap()))
                .map_err(Error::Evaluation)?;
        }

        debug!("Initialization complete");

        self.start_time = Instant::now();

            .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(), i, client_tx.clone())
                .evaluate(i, self.data.nxt_y.clone(), ChannelSender::new(i, client_tx.clone()))
                .map_err(Error::Evaluation)?;
        }
        Ok(false)
    }

    /// Update the problem after descent or null `step`.
    ///

/*
 * Copyright (c) 2019 Frank Fischer <frank-fischer@shadow-soft.de>
 *
 * This program is free software: you can redistribute it and/or
 * modify it under the terms of the GNU General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * 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/>
 */

//! Implementation for `ResultSender` using channels.

use crate::problem::{FirstOrderProblem, ResultSender};
use crate::{Minorant, Real};

use crossbeam::channel::{SendError, Sender};

/// Evaluation result.
///
/// The result of an evaluation is new information to be made
/// available to the solver and the master problem. There are
/// essentially two types of information:
///
///    1. The (exact) function value of a sub-function at some point.
///    2. A minorant of some sub-function.
#[derive(Debug)]
pub enum EvalResult<I, P> {
    /// The objective value at some point.
    ObjectiveValue { index: I, value: Real },
    /// A minorant with an associated primal.
    Minorant { index: I, minorant: Minorant, primal: P },
}

pub struct ChannelSender<I, P, E> {
    index: I,
    tx: Sender<Result<EvalResult<I, P>, E>>,
}

impl<I, P, E> ChannelSender<I, P, E> {
    pub fn new(index: I, tx: Sender<Result<EvalResult<I, P>, E>>) -> Self {
        ChannelSender { index, tx }
    }
}

impl<I, P> ResultSender<P> for ChannelSender<I, P::Primal, P::Err>
where
    I: Clone + Send,
    P: FirstOrderProblem,
    P::Err: Send,
{
    type Err = SendError<Result<EvalResult<I, P::Primal>, P::Err>>;

    fn objective(&self, value: Real) -> Result<(), SendError<Result<EvalResult<I, P::Primal>, P::Err>>> {
        self.tx.send(Ok(EvalResult::ObjectiveValue {
            index: self.index.clone(),
            value,
        }))
    }

    fn minorant(
        &self,
        minorant: Minorant,
        primal: P::Primal,
    ) -> Result<(), SendError<Result<EvalResult<I, P::Primal>, P::Err>>> {
        self.tx.send(Ok(EvalResult::Minorant {
            index: self.index.clone(),
            minorant,
            primal,
        }))
    }

    fn error(&self, err: P::Err) -> Result<(), SendError<Result<EvalResult<I, P::Primal>, P::Err>>> {
        self.tx.send(Err(err))
    }
}

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

use crate::{DVector, Real};

use super::channels::{ChannelSender, EvalResult};
use super::masterprocess::{self, MasterConfig, MasterProcess, MasterResponse};
use crate::master::{Builder as MasterBuilder, MasterProblem};
use crate::problem::{EvalResult, FirstOrderProblem, Update, UpdateState};
use crate::problem::{FirstOrderProblem, Update, 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;

/// The default solver.

    /// 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: Into<Box<dyn std::error::Error + Sync + Send>> + 'static,
    P::Err: Into<Box<dyn std::error::Error + Sync + Send>> + 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

        ));

        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(), i, self.client_tx.clone().unwrap())
                .evaluate(i, y.clone(), ChannelSender::new(i, self.client_tx.clone().unwrap()))
                .map_err(Error::Evaluation)?;
        }

        debug!("Initialization complete");

        self.start_time = Instant::now();

            .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(), i, client_tx.clone())
                .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, Error<P::Err>> {
        let master_proc = self.master_proc.as_mut().ok_or(Error::NotInitialized)?;