13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
*
* 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, Sender};
use log::{debug, info};
use num_cpus;
use num_traits::Float;
use std::sync::Arc;
use std::time::Instant;
use threadpool::ThreadPool;
|
|
|
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
*
* 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};
use num_cpus;
use num_traits::Float;
use std::sync::Arc;
use std::time::Instant;
use threadpool::ThreadPool;
|
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
where
MErr: std::error::Error + 'static,
{
fn from(err: masterprocess::Error<MErr>) -> Error<E> {
Error::Master(err.into())
}
}
type ClientSender<P> =
Sender<std::result::Result<EvalResult<usize, <P as FirstOrderProblem>::Primal>, <P as FirstOrderProblem>::Err>>;
type ClientReceiver<P> =
Receiver<std::result::Result<EvalResult<usize, <P as FirstOrderProblem>::Primal>, <P as FirstOrderProblem>::Err>>;
|
>
>
>
>
>
>
|
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
where
MErr: std::error::Error + 'static,
{
fn from(err: masterprocess::Error<MErr>) -> Error<E> {
Error::Master(err.into())
}
}
impl<E> From<RecvError> for Error<E> {
fn from(err: RecvError) -> Error<E> {
Error::Process(err.into())
}
}
type ClientSender<P> =
Sender<std::result::Result<EvalResult<usize, <P as FirstOrderProblem>::Primal>, <P as FirstOrderProblem>::Err>>;
type ClientReceiver<P> =
Receiver<std::result::Result<EvalResult<usize, <P as FirstOrderProblem>::Primal>, <P as FirstOrderProblem>::Err>>;
|
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
|
debug!("Start solving up to {} iterations", niter);
let mut itdata = IterData::new(self.problem.num_subproblems(), self.problem.num_variables(), niter);
loop {
select! {
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)?;
if self.handle_client_response(msg, &mut itdata)? {
return Ok(false);
}
},
recv(self.master_proc.as_ref().ok_or(Error::NotInitialized)?.rx) -> msg => {
debug!("Receive master response");
// Receive result (new candidate) from the master
let master_res = msg
.map_err(|err| Error::Process(err.into()))?
|
|
<
<
>
|
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
|
debug!("Start solving up to {} iterations", niter);
let mut itdata = IterData::new(self.problem.num_subproblems(), self.problem.num_variables(), niter);
loop {
select! {
recv(self.client_rx.as_ref().ok_or(Error::NotInitialized)?) -> msg => {
let msg = msg?
.map_err(Error::Evaluation)?;
if self.handle_client_response(msg, &mut itdata)? {
return Ok(false);
}
},
recv(self.master_proc.as_ref().ok_or(Error::NotInitialized)?.rx) -> msg => {
debug!("Receive master response");
// Receive result (new candidate) from the master
let master_res = msg
?
.map_err(|err| Error::Master(err.into()))?;
if self.handle_master_response(master_res, &mut itdata)? {
return Ok(true);
}
},
}
|