/// Parses rates of Timely operators from a CSV file and stores them in a topology
///
/// # Arguments
///
/// `file`: The path to the CSV file
///
/// `topology`: The topology struct to store the rate information
///
/// `source_rates_file`: The path to an optional CSV file containing true output rates of source operators
///
/// `verbose`: A boolean flag to print further information to standard output
///
/// # Comments
///
/// * Each line in `file` contains the true and observed rates of an operator instance aggregated per time window. It has the form:
//
///   `window(Rates(worker_id,operator_id,true_processing_rate,true_output_rate,observed_processing_rate,observed_output_rate))`
///
/// * Each line in `source_rates_file` contains the true output rates of source operators. It has the form:
//
///   `operator_id,true_output_rate_per_instance`
///
/// * The true output rates in `source_rates_file` overwrite the respective rates from `file` (if any)
Parses rates of Timely operators from a CSV file and stores them in a topology
Parses rates of Timely operators from a CSV file and stores them in a topology
pub fn read_timely_rates(filepath: &Path, topology: &mut Topology, source_rates_filepath: Option<&Path>, verbose: bool)
{
    let file = File::open(&filepath).expect("Cannot open rates file.");
    let reader = BufReader::new(file);
    let mut logs: HashMap<OperatorId,HashMap<WorkerId,Vec<Log>>> = HashMap::new();
    let rates_re = Regex::new(r"replayed converted:\s+(\d+)\s+\(Rates\(Op\((\d+),\s+(\d+)\)\),\s+\((\d+\.*\d+),\s+(\d+\.*\d+),\s+(\d+\.*\d+),\s+(\d+\.*\d+)\)\)").unwrap();
    for line in reader.lines()
    {
        let line = line.unwrap();
        if line.trim().len() == 0 || line.chars().next()==Some('#') {continue;}
        for cap in rates_re.captures_iter(&line[..])
        { // println!("Timestamp: {}, Worker id: {}, Operator id: {} Rates: {}, {}, {}, {}",&cap[1],&cap[2],&cap[3],&cap[4],&cap[5],&cap[6],&cap[7]);
            let operator_slot = logs.entry(cap[3].to_string()).or_insert_with(|| HashMap::new());
            let instance_slot = operator_slot.entry(cap[2].to_string()).or_insert_with(|| Vec::new());
            instance_slot.push((u64::from_str(&cap[1]).expect("Cannot extract epoch timestamp."),
                                f64::from_str(&cap[4]).expect("Cannot extract true processing rate."),
                                f64::from_str(&cap[5]).expect("Cannot extract true output rate."),
                                f64::from_str(&cap[6]).expect("Cannot extract observed processing rate."),
                                f64::from_str(&cap[7]).expect("Cannot extract observed output rate."))
            			);
        }
    }
    // Aggregate rate per logical operator
    for (op_id,instance_logs) in logs.iter_mut()
    {
        let mut op_logs: HashMap<Epoch,Rates> = HashMap::new();
        for (_,mut logs) in instance_logs.iter_mut()
        {
            logs.sort_by(|a:&Log,b:&Log| a.0.cmp(&b.0));
            let mut epoch = 0;
            for &(_,true_proc_rate,true_out_rate,obs_proc_rate,obs_out_rate) in logs.iter()
            {
                let op_log = op_logs.entry(epoch).or_insert_with(|| Rates::default());
                op_log.0 += true_proc_rate;
                op_log.1 += true_out_rate;
                op_log.2 += obs_proc_rate;
                op_log.3 += obs_out_rate;
                epoch += 1;
            }
        }
    // Update topology with aggregated logs
    let &(idx,_) = topology.dictionary.get(op_id).expect("Operator not found in dictionary.");
    match topology.logical_graph.node_weight_mut(idx)
    {
        Some(op_info) => op_info.rates = op_logs,
        None => panic!("Operator not found in topology.")
    }
    }
    // Set true output source rates explicitly
    match source_rates_filepath
    {
        Some(fp) =>
        {
            let source_rates = parse_operator_rates(fp,verbose);
            set_true_output_rates(topology,source_rates,verbose);
        },
        None => {}
    }
}
pub fn read_timely_rates ( filepath : & Path , topology : & mut Topology , source_rates_filepath : Option < & Path > , verbose : bool ) { let file = File :: open ( & filepath ) . expect ( " " ) ; let reader = BufReader :: new ( file ) ; let mut logs : HashMap < OperatorId , HashMap < WorkerId , Vec < Log > > > = HashMap :: new ( ) ; let rates_re = Regex :: new ( r"replayed converted:\s+(\d+)\s+\(Rates\(Op\((\d+),\s+(\d+)\)\),\s+\((\d+\.*\d+),\s+(\d+\.*\d+),\s+(\d+\.*\d+),\s+(\d+\.*\d+)\)\)" ) . unwrap ( ) ; for line in reader . lines ( ) { let line = line . unwrap ( ) ; if line . trim ( ) . len ( ) == 0 || line . chars ( ) . next ( ) == Some ( '#' ) { continue ; } for cap in rates_re . captures_iter ( & line [ .. ] ) { // println!("Timestamp: {}, Worker id: {}, Operator id: {} Rates: {}, {}, {}, {}",&cap[1],&cap[2],&cap[3],&cap[4],&cap[5],&cap[6],&cap[7]); let operator_slot = logs . entry ( cap [ 3 ] . to_string ( ) ) . or_insert_with ( | | HashMap :: new ( ) ) ; let instance_slot = operator_slot . entry ( cap [ 2 ] . to_string ( ) ) . or_insert_with ( | | Vec :: new ( ) ) ; instance_slot . push ( ( u64 :: from_str ( & cap [ 1 ] ) . expect ( " " ) , f64 :: from_str ( & cap [ 4 ] ) . expect ( " " ) , f64 :: from_str ( & cap [ 5 ] ) . expect ( " " ) , f64 :: from_str ( & cap [ 6 ] ) . expect ( " " ) , f64 :: from_str ( & cap [ 7 ] ) . expect ( " " ) ) ) ; } } // Aggregate rate per logical operator for ( op_id , instance_logs ) in logs . iter_mut ( ) { let mut op_logs : HashMap < Epoch , Rates > = HashMap :: new ( ) ; for ( _ , mut logs ) in instance_logs . iter_mut ( ) { logs . sort_by ( | a : & Log , b : & Log | a . 0 . cmp ( & b . 0 ) ) ; let mut epoch = 0 ; for & ( _ , true_proc_rate , true_out_rate , obs_proc_rate , obs_out_rate ) in logs . iter ( ) { let op_log = op_logs . entry ( epoch ) . or_insert_with ( | | Rates :: default ( ) ) ; op_log . 0 += true_proc_rate ; op_log . 1 += true_out_rate ; op_log . 2 += obs_proc_rate ; op_log . 3 += obs_out_rate ; epoch += 1 ; } } // Update topology with aggregated logs let & ( idx , _ ) = topology . dictionary . get ( op_id ) . expect ( " " ) ; match topology . logical_graph . node_weight_mut ( idx ) { Some ( op_info ) => op_info . rates = op_logs , None => panic ! ( " " ) } } // Set true output source rates explicitly match source_rates_filepath { Some ( fp ) => { let source_rates = parse_operator_rates ( fp , verbose ) ; set_true_output_rates ( topology , source_rates , verbose ) ; } , None => { } } }
Parses rates of Timely operators from a CSV file and stores them in a topology
Parses rates of Timely operators from a CSV file and stores them in a topology
filepath
&Path
topology
&mut Topology
source_rates_filepath
Option<&Path>
verbose
bool
comments
Each line in `file` contains the true and observed rates of an operator instance aggregated per time window. It has the form.

`window(Rates(worker_id,operator_id,true_processing_rate,true_output_rate,observed_processing_rate,observed_output_rate))`

Each line in `source_rates_file` contains the true output rates of source operators. It has the form.



The true output rates in `source_rates_file` overwrite the respective rates from `file` (if any)
Each line in ` file ` contains the true and observed rates of an operator instance aggregated per time window . It has the form . ` window ( Rates ( worker_id operator_id true_processing_rate true_output_rate observed_processing_rate observed_output_rate )) ` Each line in ` source_rates_file ` contains the true output rates of source operators . It has the form . The true output rates in ` source_rates_file ` overwrite the respective rates from ` file ` ( if any )