/// Parses a Timely dataflow from a Timely log file and returns a topology
///
/// # Arguments
///
/// `file`: The path to the Timely log file
///
/// # Comments
///
/// * The Timely dataflow is extracted from Timely logs of the following form:
///
///   `Operates(OperatesEvent { id: OperatorId, addr: [WorkerId, X, LocalOperatorId], name: OperatorName }))`
///
///   `Channels(ChannelsEvent { id: ChannelId, scope_addr: [Y, W], source: (LocalOperatorId, Z), target: (LocalOperatorId, P) }))`
Parses a Timely dataflow from a Timely log file and returns a topology
Parses a Timely dataflow from a Timely log file and returns a topology
pub fn create_timely_topology(file: &Path) -> Topology
{
    let mut topology = Topology::new();
    let file = File::open(&file).expect("Cannot open topology file.");
    let reader = BufReader::new(file);
    let op_re = Regex::new(r####"Operates\(OperatesEvent\s+\{\s+id:\s+(\d+),\s+addr:\s\[(\d+),.+,\s(\d+)\],\sname:\s"(\s*(\w)+\s*)+""####).unwrap();
    let ch_re = Regex::new(r"Channels\(ChannelsEvent\s\{\sid:\s(\d+),\s.+source:\s\((\d+),.+target:\s\((\d+).+").unwrap();
    let mut operators = HashMap::new();
    let mut edges = HashSet::new();
    let mut op_id_mapping = HashMap::new();
    // Number of instances for each operator
    let mut instances = usize::max_value();
    for line in reader.lines()
    {
        let line = line.unwrap();
        if line.trim().len() == 0 || line.chars().next()==Some('#') {continue;}
        for cap in op_re.captures_iter(&line[..])
        { // println!("Unique operator id: {}, Worker id: {} Operator id: {} Name: {}", &cap[1], &cap[2], &cap[3], &cap[4]);
            op_id_mapping.entry(cap[3].to_string()).or_insert(cap[1].to_string());
            operators.entry(cap[1].to_string())
                    .or_insert_with(|| (cap[4].to_string(),Vec::new()))
                    .1.push(cap[2].to_string());
            continue;
        }
        for cap in ch_re.captures_iter(&line[..])
        { // println!("Channel id: {} Source operator id: {} Target operator id: {}", &cap[1], &cap[2], &cap[3]);
            edges.insert((cap[1].to_string(),cap[2].to_string(),cap[3].to_string()));
        }
    }
    for (op_id,(name,workers)) in operators.drain()
    {
        match instances == usize::max_value()
        {
            false => assert_eq!(instances,workers.len()),	// All operators in Timely have the same number of instances
            true => instances = workers.len()
        }
        let _ = topology.add_unique_node(&op_id[..],&name[..],workers.len() as u32);
    }
    for (c,s,t) in edges.drain()
    { // Map local operator ids to global ones
        let s_op_uid = op_id_mapping.get(&s).expect("No unique id found for source operator.");
        let t_op_uid = op_id_mapping.get(&t).expect("No unique id found for target operator.");
        let &(n1,_) = topology.dictionary.get(&s_op_uid[..]).expect("Source operator was not found in dictionary.");
        let &(n2,_) = topology.dictionary.get(&t_op_uid[..]).expect("Target operator was not found in dictionary.");
        topology.add_unique_edge(n1,n2);
        topology.channel_dictionary.insert(c,(s_op_uid.clone(),t_op_uid.clone()));
    }
    topology
}
pub fn create_timely_topology ( file : & Path ) -> Topology { let mut topology = Topology :: new ( ) ; let file = File :: open ( & file ) . expect ( " " ) ; let reader = BufReader :: new ( file ) ; let op_re = Regex :: new ( r####"Operates\(OperatesEvent\s+\{\s+id:\s+(\d+),\s+addr:\s\[(\d+),.+,\s(\d+)\],\sname:\s"(\s*(\w)+\s*)+""#### ) . unwrap ( ) ; let ch_re = Regex :: new ( r"Channels\(ChannelsEvent\s\{\sid:\s(\d+),\s.+source:\s\((\d+),.+target:\s\((\d+).+" ) . unwrap ( ) ; let mut operators = HashMap :: new ( ) ; let mut edges = HashSet :: new ( ) ; let mut op_id_mapping = HashMap :: new ( ) ; // Number of instances for each operator let mut instances = usize :: max_value ( ) ; for line in reader . lines ( ) { let line = line . unwrap ( ) ; if line . trim ( ) . len ( ) == 0 || line . chars ( ) . next ( ) == Some ( '#' ) { continue ; } for cap in op_re . captures_iter ( & line [ .. ] ) { // println!("Unique operator id: {}, Worker id: {} Operator id: {} Name: {}", &cap[1], &cap[2], &cap[3], &cap[4]); op_id_mapping . entry ( cap [ 3 ] . to_string ( ) ) . or_insert ( cap [ 1 ] . to_string ( ) ) ; operators . entry ( cap [ 1 ] . to_string ( ) ) . or_insert_with ( | | ( cap [ 4 ] . to_string ( ) , Vec :: new ( ) ) ) . 1 . push ( cap [ 2 ] . to_string ( ) ) ; continue ; } for cap in ch_re . captures_iter ( & line [ .. ] ) { // println!("Channel id: {} Source operator id: {} Target operator id: {}", &cap[1], &cap[2], &cap[3]); edges . insert ( ( cap [ 1 ] . to_string ( ) , cap [ 2 ] . to_string ( ) , cap [ 3 ] . to_string ( ) ) ) ; } } for ( op_id , ( name , workers ) ) in operators . drain ( ) { match instances == usize :: max_value ( ) { false => assert_eq ! ( instances workers len ( ) ) , // All operators in Timely have the same number of instances true => instances = workers . len ( ) } let _ = topology . add_unique_node ( & op_id [ .. ] , & name [ .. ] , workers . len ( ) as u32 ) ; } for ( c , s , t ) in edges . drain ( ) { // Map local operator ids to global ones let s_op_uid = op_id_mapping . get ( & s ) . expect ( " " ) ; let t_op_uid = op_id_mapping . get ( & t ) . expect ( " " ) ; let & ( n1 , _ ) = topology . dictionary . get ( & s_op_uid [ .. ] ) . expect ( " " ) ; let & ( n2 , _ ) = topology . dictionary . get ( & t_op_uid [ .. ] ) . expect ( " " ) ; topology . add_unique_edge ( n1 , n2 ) ; topology . channel_dictionary . insert ( c , ( s_op_uid . clone ( ) , t_op_uid . clone ( ) ) ) ; } topology }
Parses a Timely dataflow from a Timely log file and returns a topology
Parses a Timely dataflow from a Timely log file and returns a topology
file
&Path
comments
The Timely dataflow is extracted from Timely logs of the following form.

`Operates(OperatesEvent { id: OperatorId, addr: [WorkerId, X, LocalOperatorId], name: OperatorName }))`

`Channels(ChannelsEvent { id: ChannelId, scope_addr: [Y, W], source: (LocalOperatorId, Z), target: (LocalOperatorId, P) }))`
The Timely dataflow is extracted from Timely logs of the following form . ` Operates ( OperatesEvent { id : OperatorId addr : [ WorkerId X LocalOperatorId ] name : OperatorName } )) ` ` Channels ( ChannelsEvent { id : ChannelId scope_addr : [ Y W ] source : ( LocalOperatorId Z ) target : ( LocalOperatorId P ) } )) `