wral 0.2.0

Write ahead logging for durability
Documentation 
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

use arbitrary::Unstructured;
use rand::{prelude::random, rngs::SmallRng, Rng, SeedableRng};

use super::*;

#[test]
fn test_wal() {
    let seeds: Vec<u128> = vec![275868558029936601763097336595517926704, random()];
    let seed = seeds[random::<usize>() % seeds.len()];
    // let seed: u128 = 275868558029936601763097336595517926704;
    println!("test_wal {}", seed);
    let mut rng = SmallRng::from_seed(seed.to_le_bytes());

    let mut config: Config = {
        let bytes = rng.gen::<[u8; 32]>();
        let mut uns = Unstructured::new(&bytes);
        uns.arbitrary().unwrap()
    };
    config.name = "test-wal".to_string();
    let dir = tempfile::tempdir().unwrap();
    config.dir = dir.path().into();

    println!("{:?}", config);
    let val = Wal::create(config, state::NoState).unwrap();

    let n_threads = 1;

    let mut writers = vec![];
    for id in 0..n_threads {
        let wal = val.clone();
        writers.push(std::thread::spawn(move || writer(id, wal, 1000, seed + id)));
    }

    let mut entries: Vec<Vec<entry::Entry>> = vec![];
    for handle in writers {
        entries.push(handle.join().unwrap());
    }
    let entries: Vec<entry::Entry> = entries.into_iter().flatten().collect();

    let n = entries.len() as u64;
    let sum = entries.iter().map(|e| e.to_seqno()).sum::<u64>();
    assert_eq!(sum, (n * (n + 1)) / 2);

    let mut readers = vec![];
    for id in 0..n_threads {
        let wal = val.clone();
        let entries = entries.clone();
        readers.push(std::thread::spawn(move || {
            reader(id, wal, 10, seed + id, entries)
        }));
    }

    for handle in readers {
        handle.join().unwrap();
    }

    val.close(true).unwrap();
}

fn writer(_id: u128, wal: Wal, ops: usize, seed: u128) -> Vec<entry::Entry> {
    let mut rng = SmallRng::from_seed(seed.to_le_bytes());

    let mut entries = vec![];
    for _i in 1..ops {
        let op: Vec<u8> = {
            let bytes = rng.gen::<[u8; 32]>();
            let mut uns = Unstructured::new(&bytes);
            uns.arbitrary().unwrap()
        };
        let seqno = wal.add_op(&op).unwrap();
        entries.push(entry::Entry::new(seqno, op));
    }

    entries
}

fn reader(_id: u128, wal: Wal, ops: usize, seed: u128, entries: Vec<entry::Entry>) {
    let mut rng = SmallRng::from_seed(seed.to_le_bytes());

    for _i in 0..ops {
        match rng.gen::<u8>() % 2 {
            0 => {
                let items: Vec<entry::Entry> =
                    wal.iter().unwrap().map(|x| x.unwrap()).collect();
                assert_eq!(items, entries);
            }
            1 => {
                let start = rng.gen::<usize>() % entries.len();
                let end = start + (rng.gen::<usize>() % (entries.len() - start));
                let (x, y) = (entries[start].to_seqno(), entries[end].to_seqno());
                let items: Vec<entry::Entry> =
                    wal.range(x..y).unwrap().map(|x| x.unwrap()).collect();
                assert_eq!(items, entries[start..end]);
            }
            _ => unreachable!(),
        }
    }
}