raft-log 0.3.0

Raft log implementation
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

use std::io;
use std::io::Write;
use std::sync::Arc;
use std::thread::sleep;
use std::time::Duration;

use goldenfile::Mint;
use raft_log::Config;
use raft_log::RaftLog;
use raft_log::Types;
use raft_log::api::raft_log_writer::RaftLogWriter;
use tempfile::TempDir;

// Define the types for our test
#[derive(Debug, Clone, PartialEq, Eq, Default)]
struct TestTypes;

impl Types for TestTypes {
    type LogId = (u64, u64); // (term, index)
    type LogPayload = String;
    type Vote = (u64, u64); // (term, node_id)
    type Callback = std::sync::mpsc::SyncSender<Result<(), io::Error>>;
    type UserData = Vec<u8>; // For snapshot data

    fn log_index(log_id: &Self::LogId) -> u64 {
        log_id.1 // Return the index part
    }

    fn payload_size(payload: &Self::LogPayload) -> u64 {
        payload.len() as u64
    }
}

#[test]
fn test_massive_load() -> std::io::Result<()> {
    let mut mint = Mint::new("tests/massive");
    let file = &mut mint.new_goldenfile("periodical-read.txt")?;

    let temp_dir = TempDir::new()?;
    let path = temp_dir.path().to_path_buf().to_str().unwrap().to_string();
    // let path = "tests/foo".to_string();

    let config = Arc::new(Config {
        dir: path,
        log_cache_max_items: Some(200),
        chunk_max_records: Some(100),
        ..Default::default()
    });

    let mut log_index = 0u64;
    let mut purge_index = 0u64;

    // reopen 3 times
    for reopen_index in 0..3 {
        let mut log = RaftLog::<TestTypes>::open(config.clone())?;

        for i in 1u64..500 {
            log_index += 1;
            let log_id = (log_index, log_index);

            // Append new entries
            let entries =
                vec![(log_id, format!("data-{}-{}", reopen_index, log_index))];
            log.append(entries)?;

            // Commit and purge periodically
            if i % 11 == 0 {
                log.commit(log_id)?;
            }

            if i % 13 == 0 {
                log.purge((purge_index, purge_index))?;
                purge_index += 7;
            }

            // Save vote occasionally
            if i % 17 == 0 {
                log.save_vote((log_index, 1u64))?;
            }

            // Truncate the log
            if i % 29 == 0 {
                log.truncate(log_index - 5)?;
                log_index -= 6;
            }

            // Read and verify periodically
            if i % 23 == 0 {
                // deterministic random index between purged and log_index
                let random_index =
                    purge_index + (i * 40503 % (log_index - purge_index + 1));
                let entries = log
                    .read(random_index, random_index + 6)
                    .collect::<Result<Vec<_>, _>>()?;

                let state = log.log_state();
                writeln!(file, "{:?}", state)?;

                log.wait_worker_idle();
                log.drain_cache_evictable();

                let stat = log.stat();
                writeln!(file, "{:#}", stat)?;

                for entry in entries {
                    writeln!(file, "{:?}", entry)?;
                }
            }
        }

        log.wait_worker_idle();
        log.drain_cache_evictable();

        let stat = log.stat();
        writeln!(file, "sync start: {:#}", stat)?;

        let (tx, rx) = std::sync::mpsc::sync_channel(1);
        log.flush(Some(tx))?;
        rx.recv().unwrap()?;

        let stat = log.stat();
        writeln!(file, "sync done: {:#}", stat)?;

        sleep(Duration::from_millis(200));
    }

    Ok(())
}