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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

extern crate alloc;

use crate::kairos::Kairos;
use crate::metis::{Event, FifoIdeal, Ideal};
use alloc::vec::Vec;
use core::num::NonZeroUsize;
use proptest::prelude::*;

/// A scalar [`crate::metis::Fifo`] event: the stamp names the sender, `deps` is the
/// sender's own sequence number.
fn fifo_event(physical: u64, station: u32, seq: u64, payload: u32) -> Event<u32, u64> {
    Event {
        stamp: Kairos::new(physical, 0, station, 0u16),
        deps: seq,
        payload,
    }
}

fn drain_fifo(buf: &mut FifoIdeal<u32>) -> Vec<u32> {
    let mut out = Vec::new();
    while let Some(p) = buf.pop_ready() {
        out.push(p);
    }
    out
}

#[test]
fn test_fifo_has_no_cross_source_dependency() {
    // The scalar gate has no completeness clause, so one source's gap does not
    // block another source's contiguous event.
    let mut buf: FifoIdeal<u32> = Ideal::default();
    buf.insert(fifo_event(20, 1, 2, 12));
    buf.insert(fifo_event(10, 2, 1, 21));
    assert_eq!(buf.pop_ready(), Some(21));
    assert_eq!(buf.pop_ready(), None);
    assert_eq!(buf.pending_len(), 1);
    buf.insert(fifo_event(5, 1, 1, 11));
    assert_eq!(buf.pop_ready(), Some(11));
    assert_eq!(buf.pop_ready(), Some(12));
    assert_eq!(buf.pop_ready(), None);
}

#[test]
fn test_fifo_orders_concurrent_sources_by_kairos() {
    // When several sources' next events are ready at once, the machine's Kairos
    // linearization breaks the tie, as it does under the causal gate.
    let mut buf: FifoIdeal<u32> = Ideal::default();
    buf.insert(fifo_event(20, 1, 2, 12));
    buf.insert(fifo_event(30, 2, 1, 21));
    buf.insert(fifo_event(10, 1, 1, 11));
    assert_eq!(drain_fifo(&mut buf), [11u32, 12, 21]);
}

#[test]
fn test_fifo_drops_stale_replay() {
    let mut buf: FifoIdeal<u32> = Ideal::default();
    buf.insert(fifo_event(10, 1, 1, 11));
    assert_eq!(buf.pop_ready(), Some(11));
    buf.insert(fifo_event(10, 1, 1, 11));
    assert_eq!(buf.pending_len(), 0);
    assert_eq!(buf.pop_ready(), None);
}

#[test]
fn test_try_insert_fifo_rejects_at_capacity() {
    // The bounded path is gate-generic. Source 1 seq 2 is a live gap, so it holds
    // the only slot; a second live event is rejected.
    let cap = NonZeroUsize::new(1).unwrap();
    let mut buf: FifoIdeal<u32> = Ideal::default();
    assert!(buf.try_insert(fifo_event(20, 1, 2, 12), cap).is_ok());
    let rejected = buf.try_insert(fifo_event(10, 2, 1, 21), cap).unwrap_err();
    assert_eq!(rejected.event.payload, 21);
    assert_eq!(rejected.event.deps, 1);
    assert_eq!(buf.pending_len(), 1);
}

proptest! {
    /// The scalar gate delivers each source's events in strict sequence order, whatever
    /// the arrival order and however sources interleave.
    #[test]
    fn prop_fifo_releases_each_source_in_seq_order(
        spec in prop::collection::vec(1u64..5, 1..4),
    ) {
        let mut events: Vec<Event<(u32, u64), u64>> = Vec::new();
        for (i, &count) in spec.iter().enumerate() {
            let station = u32::try_from(i + 1).unwrap();
            for seq in 1..=count {
                let physical = u64::from(station) * 100 + seq;
                events.push(Event {
                    stamp: Kairos::new(physical, 0, station, 0u16),
                    deps: seq,
                    payload: (station, seq),
                });
            }
        }
        let n = events.len();
        let mut buf: FifoIdeal<(u32, u64)> = Ideal::default();
        for event in events.into_iter().rev() {
            buf.insert(event);
        }
        let mut released = Vec::new();
        while let Some(p) = buf.pop_ready() {
            released.push(p);
        }
        prop_assert_eq!(released.len(), n);
        for (i, &count) in spec.iter().enumerate() {
            let station = u32::try_from(i + 1).unwrap();
            let seqs: Vec<u64> = released
                .iter()
                .filter(|(s, _)| *s == station)
                .map(|(_, q)| *q)
                .collect();
            prop_assert_eq!(seqs, (1..=count).collect::<Vec<u64>>());
        }
    }
}