pub fn run(
    timely_configuration: timely::Configuration,
    replay_source: ReplaySource,
    with_waiting: bool,
    epoch: Option<u64>) -> Result<(), STError> {

    println!("running cp metric with_waiting: {:?}", with_waiting);

    timely::execute(timely_configuration, move |worker| {
        let index = worker.index();

        // read replayers from file (offline) or TCP stream (online)
        let readers = connect::make_readers(replay_source.clone(), worker.index(), worker.peers()).expect("couldn't create readers");

        worker.dataflow(|scope| {
            let pag: Stream<_, (PagEdge, Pair<u64, Duration>, isize)>  = pag::create_pag(scope, readers, index, 1);

            pag
                .filter(move |(edge, _t, _diff)| if let Some(e) = epoch {
                    edge.source.epoch == e
                } else {
                    true
                })
                .cpmetric(with_waiting)
                .filter(|(_, c)| c > &1)
                .inspect_time(|t, (x, c)| println!("CP {} | ep {} | dur {:?} | w{}@{:?} -> w{}@{:?} | type {:?}",
                                                   c, t.first - 1,
                                                   if x.destination.timestamp > x.source.timestamp {
                                                       (x.destination.timestamp - x.source.timestamp)
                                                   } else {
                                                       Default::default()
                                                   },
                                                   x.source.worker_id,
                                                   x.source.timestamp,
                                                   x.destination.worker_id,
                                                   x.destination.timestamp,
                                                   x.edge_type));
        });
    })
        .map_err(|x| STError(format!("error in the timely computation: {}", x)))?;

    Ok(())
}
/// Runs graph algorithms on ST2.
Runs graph algorithms on ST2.
Runs graph algorithms on ST2 .
pub fn run(
    timely_configuration: timely::Configuration,
    replay_source: ReplaySource,
    with_waiting: bool,
    epoch: Option<u64>) -> Result<(), STError> {
    println!("running cp metric with_waiting: {:?}", with_waiting);
    timely::execute(timely_configuration, move |worker| {
        let index = worker.index();
        // read replayers from file (offline) or TCP stream (online)
        let readers = connect::make_readers(replay_source.clone(), worker.index(), worker.peers()).expect("couldn't create readers");
        worker.dataflow(|scope| {
            let pag: Stream<_, (PagEdge, Pair<u64, Duration>, isize)>  = pag::create_pag(scope, readers, index, 1);
            pag
                .filter(move |(edge, _t, _diff)| if let Some(e) = epoch {
                    edge.source.epoch == e
                } else {
                    true
                })
                .cpmetric(with_waiting)
                .filter(|(_, c)| c > &1)
                .inspect_time(|t, (x, c)| println!("CP {} | ep {} | dur {:?} | w{}@{:?} -> w{}@{:?} | type {:?}",
                                                   c, t.first - 1,
                                                   if x.destination.timestamp > x.source.timestamp {
                                                       (x.destination.timestamp - x.source.timestamp)
                                                   } else {
                                                       Default::default()
                                                   },
                                                   x.source.worker_id,
                                                   x.source.timestamp,
                                                   x.destination.worker_id,
                                                   x.destination.timestamp,
                                                   x.edge_type));
        });
    })
        .map_err(|x| STError(format!("error in the timely computation: {}", x)))?;
    Ok(())
}
pub fn run ( timely_configuration : timely :: Configuration , replay_source : ReplaySource , with_waiting : bool , epoch : Option < u64 > ) -> Result < ( ) , STError > { println ! ( " " with_waiting ) ; timely :: execute ( timely_configuration , move | worker | { let index = worker . index ( ) ; // read replayers from file (offline) or TCP stream (online) let readers = connect :: make_readers ( replay_source . clone ( ) , worker . index ( ) , worker . peers ( ) ) . expect ( " " ) ; worker . dataflow ( | scope | { let pag : Stream < _ , ( PagEdge , Pair < u64 , Duration > , isize ) > = pag :: create_pag ( scope , readers , index , 1 ) ; pag . filter ( move | ( edge , _t , _diff ) | if let Some ( e ) = epoch { edge . source . epoch == e } else { true } ) . cpmetric ( with_waiting ) . filter ( | ( _ , c ) | c > & 1 ) . inspect_time ( | t , ( x , c ) | println ! ( " " c t first 1 if x destination timestamp x source timestamp { ( x destination timestamp x source timestamp ) } else { Default default ( ) } x source worker_id x source timestamp x destination worker_id x destination timestamp x edge_type ) ) ; } ) ; } ) . map_err ( | x | STError ( format ! ( " " x ) ) ) ? ; Ok ( ( ) ) }
Runs graph algorithms on ST2.
Runs graph algorithms on ST2 .
timely_configuration
timely::Configuration
replay_source
ReplaySource
with_waiting
bool
epoch
Option<u64>
timely_configuration
timely::Configuration
replay_source
ReplaySource
with_waiting
bool
epoch
Option<u64>