lolraft 0.10.2

A Multi-Raft implementation in Rust language.
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

use super::*;

use std::collections::BTreeMap;

pub struct Query {
    pub message: Bytes,
    pub user_completion: completion::UserCompletion,
}

pub struct Inner {
    app: Ref<App>,
    q: spin::Mutex<Impl>,
}

#[derive(shrinkwraprs::Shrinkwrap, Clone)]
pub struct QueryQueue(Arc<Inner>);
impl QueryQueue {
    pub fn new(app: Ref<App>) -> Self {
        let inner = Inner {
            app,
            q: Impl::new().into(),
        };
        Self(inner.into())
    }
}

impl QueryQueue {
    /// Register a query to be executed when the read index reaches `read_index`.
    /// `read_index` is the index of the commit pointer of when the query is submitted.
    pub fn register(&self, read_index: Index, query: Query) {
        let mut q = self.q.lock();
        q.register(read_index, query);
    }

    /// Execute awaiting queries whose `read_index` ∈ `[, index]` in parallel.
    pub fn execute(&self, index: Index) -> bool {
        let mut q = self.q.lock();
        q.execute(index, &self.app)
    }
}

pub struct Impl {
    reserved: BTreeMap<Index, Vec<Query>>,
}

impl Impl {
    fn new() -> Self {
        Self {
            reserved: BTreeMap::new(),
        }
    }

    fn register(&mut self, read_index: Index, query: Query) {
        self.reserved
            .entry(read_index)
            .or_insert(vec![])
            .push(query);
    }

    fn execute(&mut self, index: Index, app: &App) -> bool {
        let queries = {
            let mut out = vec![];
            let tgt_indexes: Vec<Index> = self.reserved.range(..=index).map(|(k, _)| *k).collect();
            for idx in tgt_indexes {
                if let Some(queries) = self.reserved.remove(&idx) {
                    for query in queries {
                        out.push((query, app.clone()));
                    }
                }
            }
            out
        };

        if queries.is_empty() {
            return false;
        }

        let futs = queries.into_iter().map(
            |(
                Query {
                    message,
                    user_completion,
                },
                app,
            )| async move {
                // The `completion` of the failed queries are dropped
                // which just results in failing on the client side.
                if let Ok(resp) = app.process_read(&message).await {
                    user_completion.complete_with(resp).ok();
                }
            },
        );

        for fut in futs {
            tokio::spawn(fut);
        }

        true
    }
}