/// Creates a topology with two sources, add rates for some epochs,
/// evaluates policy, and validates the results
Creates a topology with two sources, add rates for some epochs,
evaluates policy, and validates the results
Creates a topology with two sources add rates for some epochs evaluates policy and validates the results
fn evaluate_scaling_policy_two_sources()
{
    let mut topo = Topology::new();
    let s_1 = topo.add_unique_node("Source1","Source1",1);
    let s_2 = topo.add_unique_node("Source2","Source2",1);
    let op_1 = topo.add_unique_node("Op1","Op1",1);
    let op_2 = topo.add_unique_node("Op2","Op2",1);
    topo.add_unique_edge(s_1,op_1);
    topo.add_unique_edge(s_2,op_1);
    topo.add_unique_edge(op_1,op_2);
    { // Set aggregated rates for epoch 0
        let lg = &mut topo.logical_graph;
        lg[s_1].rates.insert(0,(0.0,100.0,0.0,90.0));
        lg[s_2].rates.insert(0,(0.0,100.0,0.0,90.0));
        lg[op_1].rates.insert(0,(50.0,400.0,45.0,380.0));
        lg[op_2].rates.insert(0,(100.0,200.0,100.0,200.0));
    }
    let c0 = evaluate_scaling_policy_at_epoch(&mut topo,f64::MAX,0.0,0,true);
    { // Verify results for epoch 0
        let lg = &mut topo.logical_graph;
        assert_eq!(*lg[op_1].optimal_parallelism_per_epoch.get(&0).expect("Did not find optimal parallelism."),4);
        assert_eq!(*lg[op_2].optimal_parallelism_per_epoch.get(&0).expect("Did not find optimal parallelism."),16);
    }
    let mut c = as_vec(&c0);
    c.sort();
    let mut expected = vec![("Op1".to_string(),4),("Op2".to_string(),16),("Source2".to_string(),1),("Source1".to_string(),1)];
    expected.sort();
    assert_eq!(c,expected);
    { // Set aggregated rates for epoch 1
        let lg = &mut topo.logical_graph;
        lg[s_1].rates.insert(1,(0.0,200.0,0.0,90.0));
        lg[s_2].rates.insert(1,(0.0,200.0,0.0,90.0));
        lg[op_1].rates.insert(1,(40.0,700.0,15.0,380.0));
        lg[op_2].rates.insert(1,(99.0,200.0,100.0,200.0));
    }
    // Re-evaluate policy for epoch 0
    let c0 = evaluate_scaling_policy_at_epoch(&mut topo,f64::MAX,0.0,0,true);
    { // Verify results for epoch 0
        let lg = &mut topo.logical_graph;
        assert_eq!(*lg[op_1].optimal_parallelism_per_epoch.get(&0).expect("Did not find optimal parallelism."),4);
        assert_eq!(*lg[op_2].optimal_parallelism_per_epoch.get(&0).expect("Did not find optimal parallelism."),16);
    }
    let mut c = as_vec(&c0);
    c.sort();
    let mut expected = vec![("Op1".to_string(),4),("Op2".to_string(),16),("Source2".to_string(),1),("Source1".to_string(),1)];
    expected.sort();
    assert_eq!(c,expected);
    // Now evaluate policy for epoch 1
    let c1 = evaluate_scaling_policy_at_epoch(&mut topo,f64::MAX,0.0,1,true);
    { // Verify results for epoch 1
        let lg = &mut topo.logical_graph;
        assert_eq!(*lg[op_1].optimal_parallelism_per_epoch.get(&1).expect("Did not find optimal parallelism."),10);
        assert_eq!(*lg[op_2].optimal_parallelism_per_epoch.get(&1).expect("Did not find optimal parallelism."),71);
    }
    let mut c = as_vec(&c1);
    c.sort();
    let mut expected = vec![("Op1".to_string(),10),("Op2".to_string(),71),("Source2".to_string(),1),("Source1".to_string(),1)];
    expected.sort();
    assert_eq!(c,expected);
    { // Set aggregated rates for epoch 7
        let lg = &mut topo.logical_graph;
        lg[s_1].rates.insert(7,(0.0,200.0,0.0,90.0));
        lg[s_2].rates.insert(7,(0.0,200.0,0.0,90.0));
        lg[op_1].rates.insert(7,(10.0,700.0,15.0,380.0));
        lg[op_2].rates.insert(7,(201.0,200.0,100.0,200.0));
    }
    // Now evaluate policy for epoch 7
    let c7 = evaluate_scaling_policy_at_epoch(&mut topo,f64::MAX,0.0,7,true);
    { // Verify results for epoch 7
        let lg = &mut topo.logical_graph;
        assert_eq!(*lg[op_1].optimal_parallelism_per_epoch.get(&7).expect("Did not find optimal parallelism."),40);
        assert_eq!(*lg[op_2].optimal_parallelism_per_epoch.get(&7).expect("Did not find optimal parallelism."),140);
    }
    let mut conf = as_vec(&c7);
    conf.sort();
    let mut expected = vec![("Op1".to_string(),40),("Op2".to_string(),140),("Source2".to_string(),1),("Source1".to_string(),1)];
    expected.sort();
    assert_eq!(conf,expected);
}
fn evaluate_scaling_policy_two_sources ( ) { let mut topo = Topology :: new ( ) ; let s_1 = topo . add_unique_node ( " " , " " , 1 ) ; let s_2 = topo . add_unique_node ( " " , " " , 1 ) ; let op_1 = topo . add_unique_node ( " " , " " , 1 ) ; let op_2 = topo . add_unique_node ( " " , " " , 1 ) ; topo . add_unique_edge ( s_1 , op_1 ) ; topo . add_unique_edge ( s_2 , op_1 ) ; topo . add_unique_edge ( op_1 , op_2 ) ; { // Set aggregated rates for epoch 0 let lg = & mut topo . logical_graph ; lg [ s_1 ] . rates . insert ( 0 , ( 0.0 , 100.0 , 0.0 , 90.0 ) ) ; lg [ s_2 ] . rates . insert ( 0 , ( 0.0 , 100.0 , 0.0 , 90.0 ) ) ; lg [ op_1 ] . rates . insert ( 0 , ( 50.0 , 400.0 , 45.0 , 380.0 ) ) ; lg [ op_2 ] . rates . insert ( 0 , ( 100.0 , 200.0 , 100.0 , 200.0 ) ) ; } let c0 = evaluate_scaling_policy_at_epoch ( & mut topo , f64 :: MAX , 0.0 , 0 , true ) ; { // Verify results for epoch 0 let lg = & mut topo . logical_graph ; assert_eq ! ( lg [ op_1 ] optimal_parallelism_per_epoch get ( 0 ) expect ( " " ) 4 ) ; assert_eq ! ( lg [ op_2 ] optimal_parallelism_per_epoch get ( 0 ) expect ( " " ) 16 ) ; } let mut c = as_vec ( & c0 ) ; c . sort ( ) ; let mut expected = vec ! [ ( " " to_string ( ) 4 ) ( " " to_string ( ) 16 ) ( " " to_string ( ) 1 ) ( " " to_string ( ) 1 ) ] ; expected . sort ( ) ; assert_eq ! ( c expected ) ; { // Set aggregated rates for epoch 1 let lg = & mut topo . logical_graph ; lg [ s_1 ] . rates . insert ( 1 , ( 0.0 , 200.0 , 0.0 , 90.0 ) ) ; lg [ s_2 ] . rates . insert ( 1 , ( 0.0 , 200.0 , 0.0 , 90.0 ) ) ; lg [ op_1 ] . rates . insert ( 1 , ( 40.0 , 700.0 , 15.0 , 380.0 ) ) ; lg [ op_2 ] . rates . insert ( 1 , ( 99.0 , 200.0 , 100.0 , 200.0 ) ) ; } // Re-evaluate policy for epoch 0 let c0 = evaluate_scaling_policy_at_epoch ( & mut topo , f64 :: MAX , 0.0 , 0 , true ) ; { // Verify results for epoch 0 let lg = & mut topo . logical_graph ; assert_eq ! ( lg [ op_1 ] optimal_parallelism_per_epoch get ( 0 ) expect ( " " ) 4 ) ; assert_eq ! ( lg [ op_2 ] optimal_parallelism_per_epoch get ( 0 ) expect ( " " ) 16 ) ; } let mut c = as_vec ( & c0 ) ; c . sort ( ) ; let mut expected = vec ! [ ( " " to_string ( ) 4 ) ( " " to_string ( ) 16 ) ( " " to_string ( ) 1 ) ( " " to_string ( ) 1 ) ] ; expected . sort ( ) ; assert_eq ! ( c expected ) ; // Now evaluate policy for epoch 1 let c1 = evaluate_scaling_policy_at_epoch ( & mut topo , f64 :: MAX , 0.0 , 1 , true ) ; { // Verify results for epoch 1 let lg = & mut topo . logical_graph ; assert_eq ! ( lg [ op_1 ] optimal_parallelism_per_epoch get ( 1 ) expect ( " " ) 10 ) ; assert_eq ! ( lg [ op_2 ] optimal_parallelism_per_epoch get ( 1 ) expect ( " " ) 71 ) ; } let mut c = as_vec ( & c1 ) ; c . sort ( ) ; let mut expected = vec ! [ ( " " to_string ( ) 10 ) ( " " to_string ( ) 71 ) ( " " to_string ( ) 1 ) ( " " to_string ( ) 1 ) ] ; expected . sort ( ) ; assert_eq ! ( c expected ) ; { // Set aggregated rates for epoch 7 let lg = & mut topo . logical_graph ; lg [ s_1 ] . rates . insert ( 7 , ( 0.0 , 200.0 , 0.0 , 90.0 ) ) ; lg [ s_2 ] . rates . insert ( 7 , ( 0.0 , 200.0 , 0.0 , 90.0 ) ) ; lg [ op_1 ] . rates . insert ( 7 , ( 10.0 , 700.0 , 15.0 , 380.0 ) ) ; lg [ op_2 ] . rates . insert ( 7 , ( 201.0 , 200.0 , 100.0 , 200.0 ) ) ; } // Now evaluate policy for epoch 7 let c7 = evaluate_scaling_policy_at_epoch ( & mut topo , f64 :: MAX , 0.0 , 7 , true ) ; { // Verify results for epoch 7 let lg = & mut topo . logical_graph ; assert_eq ! ( lg [ op_1 ] optimal_parallelism_per_epoch get ( 7 ) expect ( " " ) 40 ) ; assert_eq ! ( lg [ op_2 ] optimal_parallelism_per_epoch get ( 7 ) expect ( " " ) 140 ) ; } let mut conf = as_vec ( & c7 ) ; conf . sort ( ) ; let mut expected = vec ! [ ( " " to_string ( ) 40 ) ( " " to_string ( ) 140 ) ( " " to_string ( ) 1 ) ( " " to_string ( ) 1 ) ] ; expected . sort ( ) ; assert_eq ! ( conf expected ) ; }
Creates a topology with two sources, add rates for some epochs,
evaluates policy, and validates the results
Creates a topology with two sources add rates for some epochs evaluates policy and validates the results