openraft 0.9.24

Advanced Raft consensus
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

use std::fmt;

use crate::display_ext::DisplayInstantExt;
use crate::display_ext::DisplayOptionExt;
use crate::progress::Progress;
use crate::progress::VecProgress;
use crate::proposer::Leader;
use crate::quorum::QuorumSet;
use crate::type_config::alias::InstantOf;
use crate::LogId;
use crate::RaftTypeConfig;
use crate::Vote;

/// Candidate: voting state.
#[derive(Clone, Debug)]
#[derive(PartialEq, Eq)]
pub(crate) struct Candidate<C, QS>
where
    C: RaftTypeConfig,
    QS: QuorumSet<C::NodeId>,
{
    /// When the voting is started.
    starting_time: InstantOf<C>,

    /// The vote.
    vote: Vote<C::NodeId>,

    last_log_id: Option<LogId<C::NodeId>>,

    /// Which nodes have granted the the vote at certain time point.
    progress: VecProgress<C::NodeId, bool, bool, QS>,

    quorum_set: QS,

    learner_ids: Vec<C::NodeId>,
}

impl<C, QS> fmt::Display for Candidate<C, QS>
where
    C: RaftTypeConfig,
    QS: QuorumSet<C::NodeId> + fmt::Debug + 'static,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "{{{}@{}, last_log_id:{} progress:{}}}",
            self.vote,
            self.starting_time.display(),
            self.last_log_id.display(),
            self.progress
        )
    }
}

impl<C, QS> Candidate<C, QS>
where
    C: RaftTypeConfig,
    QS: QuorumSet<C::NodeId> + fmt::Debug + Clone + 'static,
{
    pub(crate) fn new(
        starting_time: InstantOf<C>,
        vote: Vote<C::NodeId>,
        last_log_id: Option<LogId<C::NodeId>>,
        quorum_set: QS,
        learner_ids: impl IntoIterator<Item = C::NodeId>,
    ) -> Self {
        Self {
            starting_time,
            vote,
            last_log_id,
            progress: VecProgress::new(quorum_set.clone(), [], false),
            quorum_set,
            learner_ids: learner_ids.into_iter().collect::<Vec<_>>(),
        }
    }

    pub(crate) fn vote_ref(&self) -> &Vote<C::NodeId> {
        &self.vote
    }

    pub(crate) fn last_log_id(&self) -> Option<&LogId<C::NodeId>> {
        self.last_log_id.as_ref()
    }

    pub(crate) fn progress(&self) -> &VecProgress<C::NodeId, bool, bool, QS> {
        &self.progress
    }

    /// Grant the vote by a node.
    pub(crate) fn grant_by(&mut self, target: &C::NodeId) -> bool {
        let granted = *self.progress.update(target, true).expect("target not in quorum set");

        tracing::info!(voting = display(&self), "{}", func_name!());

        granted
    }

    /// Return the node ids that has granted this vote.
    #[allow(dead_code)]
    pub(crate) fn granters(&self) -> impl Iterator<Item = C::NodeId> + '_ {
        self.progress().iter().filter(|(_, granted)| *granted).map(|(target, _)| target.clone())
    }

    pub(crate) fn into_leader(self) -> Leader<C, QS> {
        // Mark the vote as committed, i.e., being granted and saved by a quorum.
        let vote = {
            let mut vote = self.vote_ref().clone();
            debug_assert!(!vote.is_committed());
            vote.commit();
            vote
        };

        let last_leader_log_ids = self.last_log_id().cloned().into_iter().collect::<Vec<_>>();

        Leader::new(vote, self.quorum_set.clone(), self.learner_ids, &last_leader_log_ids)
    }
}