omnipaxos 0.2.3

A distributed log library written in Rust
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

pub mod utils;

use kompact::prelude::{promise, Ask, FutureCollection};
use omnipaxos::ballot_leader_election::Ballot;
use serial_test::serial;
use std::{thread, time::Duration};
use utils::{TestConfig, TestSystem};

/// Test case for batching.
#[test]
#[serial]
fn batching_test() {
    let wait_time_between_propose = Duration::from_millis(2);

    let cfg = TestConfig::load("batching_test").expect("Test config loaded");
    let mut sys = TestSystem::with(cfg);
    let first_node = sys.nodes.get(&1).unwrap();
    let (kprom, kfuture) = promise::<Ballot>();
    first_node.on_definition(|x| x.election_futures.push(Ask::new(kprom, ())));
    sys.start_all_nodes();
    // Wait for initial leader to become elected. Note: needed so that initial propsals don't get
    // bunched and throw off alignment with cfg.num_proposals.
    let _leader_id = kfuture
        .wait_timeout(cfg.wait_timeout)
        .expect("No leader has been elected in the allocated time!")
        .pid;

    let mut futures = vec![];
    let mut last_decided_idx = 0;
    let proposals = utils::create_proposals(1, cfg.num_proposals);
    for v in proposals {
        let (kprom, kfuture) = promise::<()>();
        first_node.on_definition(|x| {
            x.insert_decided_future(Ask::new(kprom, v.clone()));
            x.paxos.append(v.clone()).expect("Failed to append");
        });
        futures.push(kfuture);
        thread::sleep(wait_time_between_propose);
        // check batching
        first_node.on_definition(|x| {
            let decided_idx = x.paxos.get_decided_idx();
            check_batching(decided_idx, last_decided_idx, cfg.batch_size);
            last_decided_idx = decided_idx;
        });
    }

    match FutureCollection::collect_with_timeout::<Vec<_>>(futures, cfg.wait_timeout) {
        Ok(_) => {}
        Err(e) => panic!("Error on collecting futures of decided proposals: {}", e),
    }
    sys.shutdown();
}

fn check_batching(decided_idx: usize, last_decided_idx: usize, batch_size: usize) {
    let idx_diff = decided_idx - last_decided_idx;
    if idx_diff != 0 {
        assert!(decided_idx - last_decided_idx >= batch_size);
    }
}