minerva 0.2.0

Causal ordering for distributed systems
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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

extern crate alloc;

use super::super::support::{arb_history, perm_of};
use crate::metis::{BoundedInsert, CausalIdeal, Event};
use alloc::vec::Vec;
use core::num::NonZeroUsize;
use proptest::prelude::*;

proptest! {
    /// After every `try_insert`, the live backlog never exceeds `capacity`.
    #[test]
    fn prop_try_insert_never_exceeds_capacity(
        (history, perm) in arb_history()
            .prop_flat_map(|history| {
                let perm = perm_of(history.len());
                (Just(history), perm)
            }),
        cap in 1usize..6,
    ) {
        let cap = NonZeroUsize::new(cap).unwrap();
        let mut buf = CausalIdeal::new();
        for &index in &perm {
            if let Err(at) = buf.try_insert(history[index].clone(), cap) {
                prop_assert_eq!(at.event.payload, history[index].payload);
            }
            prop_assert!(buf.pending_len() <= cap.get());
        }
    }
}

proptest! {
    /// Atomic replacement preserves the bound, progress, and resident causal closure.
    #[test]
    fn prop_try_insert_evicting_where_preserves_causal_closure(
        (history, perm) in arb_history()
            .prop_flat_map(|history| {
                let perm = perm_of(history.len());
                (Just(history), perm)
            }),
        cap in 1usize..6,
    ) {
        let cap = NonZeroUsize::new(cap).unwrap();
        let mut buf = CausalIdeal::new();
        let mut pending = Vec::new();
        for &index in &perm {
            let arrival = history[index].clone();
            let before = buf.delivered().clone();
            let outcome = buf.try_insert_evicting_where(
                arrival.clone(),
                cap,
                |_| true,
            );
            prop_assert!(buf.pending_len() <= cap.get());
            prop_assert_eq!(buf.delivered(), &before);
            match outcome {
                Ok(BoundedInsert::Buffered) => pending.push(arrival),
                Ok(BoundedInsert::DroppedStale) => {
                    prop_assert!(false, "a valid unreleased history event is not stale");
                }
                Ok(BoundedInsert::Evicted { evicted }) => {
                    let victim = pending
                        .iter()
                        .position(|resident| resident.stamp == evicted.stamp)
                        .expect("the returned victim was pending");
                    prop_assert!(is_causally_maximal(victim, &pending, &arrival));
                    let _ = pending.swap_remove(victim);
                    pending.push(arrival);
                }
                Err(refused) => {
                    prop_assert_eq!(refused.event.stamp, arrival.stamp);
                    prop_assert!(pending
                        .iter()
                        .enumerate()
                        .all(|(candidate, _)| !is_causally_maximal(candidate, &pending, &arrival)));
                }
            }
            prop_assert_eq!(buf.pending_len(), pending.len());
        }
    }
}

fn is_causally_maximal(candidate: usize, pending: &[Event<u32>], arrival: &Event<u32>) -> bool {
    let victim = &pending[candidate];
    !depends_on(arrival, victim)
        && pending
            .iter()
            .enumerate()
            .all(|(index, event)| index == candidate || !depends_on(event, victim))
}

fn depends_on(event: &Event<u32>, predecessor: &Event<u32>) -> bool {
    let station = predecessor.stamp.station_id();
    let sequence = predecessor.deps.get(station);
    let event_dot = (
        event.stamp.station_id(),
        event.deps.get(event.stamp.station_id()),
    );

    event_dot != (station, sequence) && event.deps.get(station) >= sequence
}