/// Evaluates the scaling model on a topology with collected metrics for a given epoch
///
/// # Arguments
///
/// `Topology`: The topology struct with the collected operator rates
///
/// `threshold`: The target rate ratio threshold
///
/// `factor`: The scaling factor
///
/// `epoch`: The epoch for which to evaluate the policy
///
/// `verbose`: A boolean flag to print further information to standard output
///
/// # Returns
///
/// A string of the form:
///
/// `operator_id1,number_of_instances1,operator_id2,number_of_instances2,...`
///
/// containing the optimal dataflow configuration, i.e. the optimal level of parallelism for each dataflow operator
///
/// # Comments
///
/// * The `threshold >= 1 (default: f64::MAX)` defines the maximum allowed ratio of a target source rate to the respective observed source rate after the application of a scaling decision.
/// When DS2 converges and the aforementioned ratio for a source operator is above the given threshold, a final configuration adjustment is triggered in an attempt to meet the throughput
///
/// * The `factor >= 0 (default: 0)` is used to deliberately over-provision each dataflow operator by `factor * 100%` more instances than the optimal
Evaluates the scaling model on a topology with collected metrics for a given epoch
Evaluates the scaling model on a topology with collected metrics for a given epoch
pub fn evaluate_scaling_policy_at_epoch(topo: &mut Topology, threshold: f64, factor: f64, epoch: Epoch, verbose: bool) -> String
{
    // Used in configuration adjustments
    let sources = topo.get_sources_idx();
    let mut source_rates_ratios = Vec::new();
    // Useful in Timely experiments
    let mut workers = 0f64;
    let mut optimal_config = String::new(); // The result
    // Sort operators in topological order -- Works only with DAG topologies
    let operators_to_visit = toposort(&topo.logical_graph,None).expect("Cannot sort operators topologically.");
    let graph = &mut topo.logical_graph;    // The dataflow topology
    if verbose
    {
        print!("\nOperators in topological order: ");
        operators_to_visit.iter()
                        .for_each(|&op_id|
                        {
                            let op_info = graph.node_weight(op_id).unwrap();
                            print!("{}, ",op_info.name);
                        });
    }
    // A mapping from operators/nodes to their true output rates per observation window (epoch)
    let mut true_output_rate: HashMap<NodeIndex,Rate> = HashMap::new();
    for &idx in operators_to_visit.iter()
    { // Traverse operators/nodes in topological order, from the sources to the sinks
      // At each step, compare the true processing rate of the current operator with the
      // true output rate of the upstream operator(s) to estimate its optimal level of parallelism
        let mut true_processing_rate = Rate::default();
        { // Step 0: Collect the aggregated true processing rate for the current operator per epoch
            let op_info = &graph[idx];
            if verbose { println!("\n\nOperator '{}' with ID: {}",op_info.name,op_info.id); }
            let rates = op_info.rates.get(&epoch).expect("No rates found for the given epoch.");
            true_processing_rate = rates.0;
            if verbose { println!("Aggregated true processing rate for epoch {} is {}",epoch,true_processing_rate); }
        }
        let mut estimated_true_output_rate = Rate::default();
        let mut observed_output_rate = Rate::default();	// Used only in estimating source rate ratios
        { // Step 1: Collect the aggregated true output rate for the current operator per epoch
            let op_info = &graph[idx];
            let rates = op_info.rates.get(&epoch).expect("No rates found for the given epoch.");
            estimated_true_output_rate = rates.1;
            // Compute 'true output rate / observed output rate' for each source operator per epoch
            if sources.contains(&idx)
            { // Keep source rate info
                let rates = op_info.rates.get(&epoch).expect("No rates found for the given epoch.");
                observed_output_rate = rates.3;
                let ratio = estimated_true_output_rate/observed_output_rate;
                source_rates_ratios.push(ratio);
            }
        }
        // Step 2: Estimate the optimal level of parallelism for the current operator
        let op_instances = graph[idx].instances;
        if let Some(mut input_rate) = true_output_rate.remove(&idx)
        { // Initially, 'true_output_rates' is empty since the source operators do not have inputs
            if verbose { println!("Aggregated target rate for epoch {} is {}",epoch,input_rate); }
            // Estimate optimal level of parallelism assuming there is no computation skew across current operator instances
            let optimal_instances = input_rate/(true_processing_rate / op_instances as f64);
            // Update the total number of instances in the optimal dataflow configuration per epoch -- Useful in Timely experiments
            if !optimal_instances.is_nan() && !optimal_instances.is_infinite() // Ignore partially logged epochs
            {
                workers += optimal_instances;
            }
            // Store optimal configuration for the current operator in the topology
            graph[idx].optimal_parallelism_per_epoch.insert(epoch, (optimal_instances * (1.0 + factor)).ceil() as u32);
            // Estimate the aggregated true output rate of the current operator when configured with the optimal number of instances
            estimated_true_output_rate *= optimal_instances / (op_instances as f64);
        }
        let adjacent_operators = graph.neighbors(idx).collect::<Vec<_>>();	// The downstream operators of the current operator
        for &operator in adjacent_operators.iter()
        { // Step 3: Update the estimated true output rate, which amounts to the estimated input rate of the downstream operators
            let entry = true_output_rate.entry(operator).or_insert(Rate::default());
            *entry += estimated_true_output_rate;
        }
        // Step 4: Add the optimal configuration of the current operator for the given epoch to the result
        let op_info = &graph[idx];
        match op_info.optimal_parallelism_per_epoch.get(&epoch)
        {
            Some(level_of_parallelism) =>
            {
                if verbose { println!("Optimal parallelism for epoch {} is {}",epoch,level_of_parallelism); }
                optimal_config.push_str(&format!("{},{},",op_info.id,level_of_parallelism))
            },
            None => { if !sources.contains(&idx) { panic!("No optimal parallelism found."); } }
        }
    }
    // Useful in Timely experiments
    if verbose { println!("Epoch {}: {} -> {} workers",epoch,workers,workers.ceil()); }
    // Add source parallelism to the result
    for &idx in sources.iter()
    {
        let src_id = &graph[idx].id;
        let level_of_parallelism = &graph[idx].instances;
        optimal_config.push_str(&format!("{},{},",src_id,level_of_parallelism))
    }
    let l = optimal_config.len();
    optimal_config.truncate(l-1);
    optimal_config
}
pub fn evaluate_scaling_policy_at_epoch ( topo : & mut Topology , threshold : f64 , factor : f64 , epoch : Epoch , verbose : bool ) -> String { // Used in configuration adjustments let sources = topo . get_sources_idx ( ) ; let mut source_rates_ratios = Vec :: new ( ) ; // Useful in Timely experiments let mut workers = 0f64 ; let mut optimal_config = String :: new ( ) ; // The result // Sort operators in topological order -- Works only with DAG topologies let operators_to_visit = toposort ( & topo . logical_graph , None ) . expect ( " " ) ; let graph = & mut topo . logical_graph ; // The dataflow topology if verbose { print ! ( " \n " ) ; operators_to_visit . iter ( ) . for_each ( | & op_id | { let op_info = graph . node_weight ( op_id ) . unwrap ( ) ; print ! ( " " op_info name ) ; } ) ; } // A mapping from operators/nodes to their true output rates per observation window (epoch) let mut true_output_rate : HashMap < NodeIndex , Rate > = HashMap :: new ( ) ; for & idx in operators_to_visit . iter ( ) { // Traverse operators/nodes in topological order, from the sources to the sinks // At each step, compare the true processing rate of the current operator with the // true output rate of the upstream operator(s) to estimate its optimal level of parallelism let mut true_processing_rate = Rate :: default ( ) ; { // Step 0: Collect the aggregated true processing rate for the current operator per epoch let op_info = & graph [ idx ] ; if verbose { println ! ( " \n \n " op_info name op_info id ) ; } let rates = op_info . rates . get ( & epoch ) . expect ( " " ) ; true_processing_rate = rates . 0 ; if verbose { println ! ( " " epoch true_processing_rate ) ; } } let mut estimated_true_output_rate = Rate :: default ( ) ; let mut observed_output_rate = Rate :: default ( ) ; // Used only in estimating source rate ratios { // Step 1: Collect the aggregated true output rate for the current operator per epoch let op_info = & graph [ idx ] ; let rates = op_info . rates . get ( & epoch ) . expect ( " " ) ; estimated_true_output_rate = rates . 1 ; // Compute 'true output rate / observed output rate' for each source operator per epoch if sources . contains ( & idx ) { // Keep source rate info let rates = op_info . rates . get ( & epoch ) . expect ( " " ) ; observed_output_rate = rates . 3 ; let ratio = estimated_true_output_rate / observed_output_rate ; source_rates_ratios . push ( ratio ) ; } } // Step 2: Estimate the optimal level of parallelism for the current operator let op_instances = graph [ idx ] . instances ; if let Some ( mut input_rate ) = true_output_rate . remove ( & idx ) { // Initially, 'true_output_rates' is empty since the source operators do not have inputs if verbose { println ! ( " " epoch input_rate ) ; } // Estimate optimal level of parallelism assuming there is no computation skew across current operator instances let optimal_instances = input_rate / ( true_processing_rate / op_instances as f64 ) ; // Update the total number of instances in the optimal dataflow configuration per epoch -- Useful in Timely experiments if ! optimal_instances . is_nan ( ) && ! optimal_instances . is_infinite ( ) // Ignore partially logged epochs { workers += optimal_instances ; } // Store optimal configuration for the current operator in the topology graph [ idx ] . optimal_parallelism_per_epoch . insert ( epoch , ( optimal_instances * ( 1.0 + factor ) ) . ceil ( ) as u32 ) ; // Estimate the aggregated true output rate of the current operator when configured with the optimal number of instances estimated_true_output_rate *= optimal_instances / ( op_instances as f64 ) ; } let adjacent_operators = graph . neighbors ( idx ) . collect :: < Vec < _ > > ( ) ; // The downstream operators of the current operator for & operator in adjacent_operators . iter ( ) { // Step 3: Update the estimated true output rate, which amounts to the estimated input rate of the downstream operators let entry = true_output_rate . entry ( operator ) . or_insert ( Rate :: default ( ) ) ; * entry += estimated_true_output_rate ; } // Step 4: Add the optimal configuration of the current operator for the given epoch to the result let op_info = & graph [ idx ] ; match op_info . optimal_parallelism_per_epoch . get ( & epoch ) { Some ( level_of_parallelism ) => { if verbose { println ! ( " " epoch level_of_parallelism ) ; } optimal_config . push_str ( & format ! ( " " op_info id level_of_parallelism ) ) } , None => { if ! sources . contains ( & idx ) { panic ! ( " " ) ; } } } } // Useful in Timely experiments if verbose { println ! ( " " epoch workers workers ceil ( ) ) ; } // Add source parallelism to the result for & idx in sources . iter ( ) { let src_id = & graph [ idx ] . id ; let level_of_parallelism = & graph [ idx ] . instances ; optimal_config . push_str ( & format ! ( " " src_id level_of_parallelism ) ) } let l = optimal_config . len ( ) ; optimal_config . truncate ( l - 1 ) ; optimal_config }
Evaluates the scaling model on a topology with collected metrics for a given epoch
Evaluates the scaling model on a topology with collected metrics for a given epoch
A string of the form.



containing the optimal dataflow configuration, i.e. the optimal level of parallelism for each dataflow operator
A string of the form . containing the optimal dataflow configuration i . e . the optimal level of parallelism for each dataflow operator
topo
&mut Topology
threshold
f64
factor
f64
epoch
Epoch
verbose
bool
comments
The `threshold >= 1 (default: f64::MAX)` defines the maximum allowed ratio of a target source rate to the respective observed source rate after the application of a scaling decision.
When DS2 converges and the aforementioned ratio for a source operator is above the given threshold, a final configuration adjustment is triggered in an attempt to meet the throughput

The `factor >= 0 (default: 0)` is used to deliberately over-provision each dataflow operator by `factor * 100%` more instances than the optimal
The ` threshold > = 1 ( default : f64 :: MAX ) ` defines the maximum allowed ratio of a target source rate to the respective observed source rate after the application of a scaling decision . When DS2 converges and the aforementioned ratio for a source operator is above the given threshold a final configuration adjustment is triggered in an attempt to meet the throughput The ` factor > = 0 ( default : 0 ) ` is used to deliberately over - provision each dataflow operator by ` factor * 100% ` more instances than the optimal