libraft/
read_only.rs

1use std::collections::{HashMap, HashSet, VecDeque};
2
3use raftpb::Message;
4
5// ReadState provides state for read only query.
6// It's caller's responsibility to call ReadIndex first before getting
7// this state from ready, it's also caller's duty to differentiate if this
8// state is what it requests through request_ctx, eg. given a unique id as
9// request_ctx
10#[derive(Debug, Clone, PartialEq)]
11pub struct ReadState {
12    pub index: u64,
13    pub request_ctx: Vec<u8>,
14}
15
16#[derive(Debug, PartialEq, Clone, Copy)]
17pub enum ReadOnlyOption {
18    /// Safe guarantees the linearizability of the read only request by
19    /// communicating with the quorum. It is the default and suggested option.
20    Safe,
21    /// LeaseBased ensures linearizability of the read only request by
22    /// relying on the leader lease. It can be affected by clock drift.
23    /// If the clock drift is unbounded, leader might keep the lease longer than it
24    /// should (clock can move backward/pause without any bound). ReadIndex is not safe
25    /// in that case.
26    LeaseBased,
27}
28
29impl Default for ReadOnlyOption {
30    fn default() -> ReadOnlyOption {
31        ReadOnlyOption::Safe
32    }
33}
34
35#[derive(Default, Debug, Clone)]
36pub struct ReadIndexStatus {
37    pub req: Message,
38    pub index: u64,
39    acks: HashSet<u64>,
40}
41
42#[derive(Default, Debug, Clone)]
43pub struct ReadOnly {
44    pub option: ReadOnlyOption,
45    pub pending_read_index: HashMap<Vec<u8>, ReadIndexStatus>,
46    pub read_index_queue: VecDeque<Vec<u8>>,
47}
48
49impl ReadOnly {
50    pub(crate) fn new(option: ReadOnlyOption) -> ReadOnly {
51        ReadOnly {
52            option,
53            pending_read_index: HashMap::new(),
54            read_index_queue: VecDeque::new(),
55        }
56    }
57
58    pub(crate) fn last_pending_request_ctx(&mut self) -> Option<Vec<u8>> {
59        self.read_index_queue.back().cloned()
60    }
61
62    // add_request adds a read only reuqest into readonly struct.
63    // `index` is the commit index of the raft state machine when it received
64    // the read only request.
65    // `m` is the original read only request message from the local or remote node.
66    pub(crate) fn add_request(&mut self, index: u64, msg: Message) {
67        let ctx = msg.get_entries()[0].get_data().to_vec();
68        if self.pending_read_index.contains_key(&ctx) {
69            return;
70        }
71        let ris = ReadIndexStatus {
72            index,
73            req: msg,
74            acks: HashSet::new(),
75        };
76        self.pending_read_index.insert(ctx.clone(), ris);
77        self.read_index_queue.push_back(ctx);
78    }
79
80    // recv_ack notifies the readonly struct that the raft state machine received
81    // an acknowledgment of the heartbeat that attached with the read only request
82    // context.
83    pub(crate) fn recv_ack(&mut self, msg: &Message) -> usize {
84        if let Some(rs) = self.pending_read_index.get_mut(msg.get_context()) {
85            rs.acks.insert(msg.get_from());
86            rs.acks.len() + 1
87        } else {
88            0
89        }
90    }
91
92    // advance advances the read only request queue kept by the readonly struct.
93    // It dequeues the requests until it finds the read only request that has
94    // the same context as the given `msg`.
95    pub(crate) fn advance(&mut self, msg: &Message) -> Vec<ReadIndexStatus> {
96        let mut rss = vec![];
97        if let Some(i) = self.read_index_queue.iter().position(|x| {
98            if !self.pending_read_index.contains_key(x) {
99                panic!("cannot find correspond read state from pending map");
100            }
101            *x == msg.get_context()
102        }) {
103            for _ in 0..i + 1 {
104                let rs = self.read_index_queue.pop_front().unwrap();
105                let status = self.pending_read_index.remove(&rs).unwrap();
106                rss.push(status);
107            }
108        }
109        rss
110    }
111}
libraft/read_only.rs

libraft/
read_only.rs
