fn unchanged_or_scaled_down(before: HashMap<OperatorInstanceId,(NodeIndex,u32)>, after: HashMap<OperatorInstanceId,Vec<u32>>, sources: usize) -> Vec<Epoch>
{
    let mut min_epoch = usize::max_value();
    after.iter().for_each(|(_,instances)|
                    {
                        if min_epoch > instances.len() { min_epoch = instances.len(); }
                    });
    let mut epochs = Vec::new();
    for i in 0..min_epoch
    {
        let mut ops = 0;
        for (op1,instances_per_epoch) in after.iter()
        {
            let &(_,instances_before) = before.get(op1).expect("Operators before and after don't match");
            if instances_per_epoch[i] <= instances_before { ops += 1; }
        }
        if ops == before.len()-sources { epochs.push(i as Epoch); }
    }
    epochs
}
/// Collects all observation windows (epochs) that did not include any scale-up decision
///
/// # Arguments
///
/// `before`: The given dataflow configuration
///
/// `after`: The estimated optimal configuration
///
/// `sources`: The number of source operators in the dataflow
///
/// # Returns
///
/// A vector of epochs (identified by their start timestamps)
Collects all observation windows (epochs) that did not include any scale-up decision
Collects all observation windows ( epochs ) that did not include any scale - up decision
fn unchanged_or_scaled_down(before: HashMap<OperatorInstanceId,(NodeIndex,u32)>, after: HashMap<OperatorInstanceId,Vec<u32>>, sources: usize) -> Vec<Epoch>
{
    let mut min_epoch = usize::max_value();
    after.iter().for_each(|(_,instances)|
                    {
                        if min_epoch > instances.len() { min_epoch = instances.len(); }
                    });
    let mut epochs = Vec::new();
    for i in 0..min_epoch
    {
        let mut ops = 0;
        for (op1,instances_per_epoch) in after.iter()
        {
            let &(_,instances_before) = before.get(op1).expect("Operators before and after don't match");
            if instances_per_epoch[i] <= instances_before { ops += 1; }
        }
        if ops == before.len()-sources { epochs.push(i as Epoch); }
    }
    epochs
}
fn unchanged_or_scaled_down ( before : HashMap < OperatorInstanceId , ( NodeIndex , u32 ) > , after : HashMap < OperatorInstanceId , Vec < u32 > > , sources : usize ) -> Vec < Epoch > { let mut min_epoch = usize :: max_value ( ) ; after . iter ( ) . for_each ( | ( _ , instances ) | { if min_epoch > instances . len ( ) { min_epoch = instances . len ( ) ; } } ) ; let mut epochs = Vec :: new ( ) ; for i in 0 .. min_epoch { let mut ops = 0 ; for ( op1 , instances_per_epoch ) in after . iter ( ) { let & ( _ , instances_before ) = before . get ( op1 ) . expect ( " " ) ; if instances_per_epoch [ i ] <= instances_before { ops += 1 ; } } if ops == before . len ( ) - sources { epochs . push ( i as Epoch ) ; } } epochs }
Collects all observation windows (epochs) that did not include any scale-up decision
Collects all observation windows ( epochs ) that did not include any scale - up decision
before
HashMap<OperatorInstanceId,(NodeIndex,u32)>
after
HashMap<OperatorInstanceId,Vec<u32>>
sources
usize
A vector of epochs (identified by their start timestamps)
A vector of epochs ( identified by their start timestamps )
before
HashMap<OperatorInstanceId,(NodeIndex,u32)>
after
HashMap<OperatorInstanceId,Vec<u32>>
sources
usize