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

extern crate alloc;
extern crate std;

use crate::kairos::{BackoffStrategy, Clock, ClockConfig, ExponentialBackoff, TickCounter};
use alloc::vec::Vec;
use std::sync::{Arc, Barrier};
use std::thread;

#[test]
fn test_concurrency_with_backoff() {
    let config = ClockConfig {
        max_skew_warning_threshold: 100_000_000,
        #[cfg(feature = "rand")]
        backoff: BackoffStrategy::Exponential(ExponentialBackoff::new(
            4,       // base spins
            64,      // max spins
            Some(8), // up to 8 extra spins of jitter
        )),
        #[cfg(not(feature = "rand"))]
        backoff: BackoffStrategy::Exponential(ExponentialBackoff::new(4, 64, Some((123, 8)))),
    };
    let clock = Arc::new(Clock::new(TickCounter::new(), 123, config).unwrap());
    // Miri interprets every instruction and emulates weak memory, so the same
    // race coverage costs orders of magnitude more per stamp; scale the volume
    // down (S86) rather than skip the crate's one threaded CAS exercise.
    let num_threads = if cfg!(miri) { 4 } else { 8 };
    let num_timestamps_per_thread = if cfg!(miri) { 25 } else { 1000 };
    let barrier = Arc::new(Barrier::new(num_threads));
    let mut handles = Vec::new();

    for _ in 0..num_threads {
        let clock_clone = Arc::clone(&clock);
        let barrier_clone = Arc::clone(&barrier);
        handles.push(thread::spawn(move || {
            let mut timestamps = Vec::with_capacity(num_timestamps_per_thread);
            let _ = barrier_clone.wait();
            for _ in 0..num_timestamps_per_thread {
                timestamps.push(clock_clone.now(0u16));
            }
            timestamps
        }));
    }

    let mut all_timestamps = Vec::new();
    for handle in handles {
        all_timestamps.extend(handle.join().expect("Thread failed to complete"));
    }

    assert_eq!(
        all_timestamps.len(),
        num_threads * num_timestamps_per_thread
    );
    let mut sorted_timestamps = all_timestamps.clone();
    sorted_timestamps.sort();
    let original_len = sorted_timestamps.len();
    sorted_timestamps.dedup();
    assert_eq!(
        original_len,
        sorted_timestamps.len(),
        "All concurrent timestamps must be unique"
    );

    // The real CAS retry path ran under contention; peak_attempts records it.
    assert!(clock.stats().peak_attempts >= 1);
}