add raft framework, implement trigger election

Author: chyezh

build.rs

@@ -0,0 +1,4 @@
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tonic_build::compile_protos("proto/raftrpc.proto")?;
+    Ok(())
+}

proto/raftrpc.proto

@@ -0,0 +1,30 @@
+syntax = "proto3";
+
+package raftrpc;
+
+service Raft {
+    // Invoked by candidate gather votes.
+    rpc RequestVote (RequestVoteArgs) returns (RequestVoteReply);
+
+    // Invoked by leader to replicate log entries, also used as heartbeat.
+    // Do a total-order broadcast.
+    rpc AppendEntries (AppendEntriesArgs) returns (AppendEntriesReply);
+}
+
+message RequestVoteArgs {
+    uint64 term = 1;
+    uint64 candidate_id = 2;
+    uint64 last_log_index = 3;
+    uint64 last_log_term = 4;
+}
+
+message RequestVoteReply {
+    uint64 term = 1;
+    bool vote_granted = 2;
+}
+
+message AppendEntriesArgs {
+}
+
+message AppendEntriesReply {
+}

src/lib.rs

@@ -1,5 +1,6 @@
 mod cmd;
 mod engine;
+mod raft;
 mod resp;
 mod server;
 mod util;

src/raft/error.rs

@@ -0,0 +1,2 @@
+#[derive(thiserror::Error, Debug)]
+pub enum Error {}

src/raft/mod.rs

@@ -0,0 +1,4 @@
+mod error;
+mod node;
+mod rpc;
+mod state;

src/raft/node.rs

@@ -0,0 +1,111 @@
+use super::rpc::*;
+use super::state::State;
+use std::sync::{Arc, Mutex};
+use tokio::{sync::mpsc::channel, time::Instant};
+use tonic::{Request, Response, Status};
+
+pub struct RafterInner {
+    // Manage all peers rpc client
+    peers: Vec<RaftClient>,
+
+    // State of this node
+    state: Arc<Mutex<State>>,
+
+    // next triggering election time
+    election_instant: Instant,
+}
+
+impl RafterInner {
+    // Try to start a new election
+    fn trigger_election(&mut self) {
+        let mut state = self.state.lock().unwrap();
+        // Leader always don't trigger election.
+        if state.is_leader() {
+            return;
+        }
+
+        // Set as candidate and vote for myself
+        state.set_as_candidate();
+        let current_term = state.term();
+        let me = state.me();
+
+        // Send request to all peers except this node
+        let (tx, mut rx) = channel(self.peers.len() - 1);
+        for (node_id, peer) in self.peers.iter().enumerate() {
+            if node_id != me {
+                let mut peer = peer.clone();
+                let tx = tx.clone();
+
+                let args = RequestVoteArgs {
+                    term: current_term,
+                    candidate_id: me as u64,
+                    // TODO: check last log index
+                    last_log_index: 0,
+                    last_log_term: 0,
+                };
+
+                tokio::spawn(async move {
+                    let vote_reply = peer.request_vote(args).await;
+                    tx.send(vote_reply).await.ok();
+                });
+            }
+        }
+
+        // Async receiving the vote reply, and handle the election
+        let voted_threshold = self.peers.len() / 2;
+        let state = Arc::clone(&self.state);
+        tokio::spawn(async move {
+            // Candidates always voted for myself
+            let mut voted = 1;
+
+            while let Some(Ok(vote_reply)) = rx.recv().await {
+                let vote_reply = vote_reply.get_ref();
+                if vote_reply.term == current_term && vote_reply.vote_granted {
+                    voted += 1;
+                    if voted > voted_threshold {
+                        // Quorum of peers voted for this node, break to set this node as leader
+                        break;
+                    }
+                } else if current_term < vote_reply.term {
+                    // Find a new term
+                    // Set this node into a follower, give up election process in this term
+                    state
+                        .lock()
+                        .unwrap()
+                        .set_as_follower_in_new_term(vote_reply.term);
+                    return;
+                }
+            }
+
+            if voted > voted_threshold {
+                // Election success, set this peer as leader if current term is not changed
+                state
+                    .lock()
+                    .unwrap()
+                    .set_as_leader_in_this_term(current_term);
+            }
+        });
+    }
+}
+
+// Implementation for Raft algorithm
+pub struct Rafter {}
+
+#[tonic::async_trait]
+impl Raft for Rafter {
+    // Implement server side of request-vote rpc for raft
+    async fn request_vote(
+        &self,
+        request: Request<RequestVoteArgs>,
+    ) -> Result<Response<RequestVoteReply>, Status> {
+        todo!()
+    }
+
+    // Implement server side of append-entries rpc for raft
+    async fn append_entries(
+        &self,
+        request: Request<AppendEntriesArgs>,
+    ) -> Result<Response<AppendEntriesReply>, Status> {
+        todo!()
+    }
+}

src/raft/rpc.rs

@@ -0,0 +1,10 @@
+pub use raftrpc::{
+    raft_server::{Raft, RaftServer},
+    AppendEntriesArgs, AppendEntriesReply, RequestVoteArgs, RequestVoteReply,
+};
+
+pub type RaftClient = raftrpc::raft_client::RaftClient<tonic::transport::Channel>;
+
+pub mod raftrpc {
+    tonic::include_proto!("raftrpc");
+}

src/raft/state.rs

@@ -0,0 +1,98 @@
+#[derive(Default, Copy, Clone, Debug, PartialEq, Eq)]
+enum Role {
+    #[default]
+    Follower,
+    Candidate,
+    Leader,
+}
+
+#[derive(Debug, Default)]
+struct Peer {
+    // Index of the next log entry to send to that server
+    // Initialized to leader last log index + 1
+    next_index: u64,
+
+    // Index of highest log entry known to be replicated on server
+    // Initialized to 0, increases monotonically
+    match_index: u64,
+}
+
+#[derive(Debug, Default)]
+pub struct State {
+    // Role of this node
+    role: Role,
+
+    // This node's index in peers
+    me_index: usize,
+
+    // Latest term server has seen, increases monotonically
+    current_term: u64,
+
+    // Candidate id (index of peers array) that received voted in current term
+    // None if not voted.
+    voted_for: Option<usize>,
+
+    // Log entries; each entry contains command for state machine,
+    // and term when entry was received by leader
+    log: Vec<u64>,
+
+    // Index of highest log entry known to be committed
+    commit_index: u64,
+
+    // Index of highest log entry applied to state machine
+    last_applied: u64,
+
+    // List of Peer meta data, reinitialized after election
+    // Only available on leader node
+    peers: Vec<Peer>,
+}
+
+impl State {
+    #[inline]
+    pub fn is_leader(&self) -> bool {
+        matches!(self.role, Role::Leader)
+    }
+
+    #[inline]
+    pub fn term(&self) -> u64 {
+        self.current_term
+    }
+
+    // Perform a CAS, set this node as leader, if term is matched.
+    #[inline]
+    pub fn set_as_leader_in_this_term(&mut self, term: u64) {
+        if self.current_term == term {
+            self.role = Role::Leader;
+        }
+    }
+
+    // Set this node as follower and update the term if term is gte than current term
+    #[inline]
+    pub fn set_as_follower_in_new_term(&mut self, term: u64) {
+        if term >= self.current_term {
+            self.current_term = term;
+            self.role = Role::Follower;
+        }
+    }
+
+    // Set this node as candidate, increment current term
+    // Always vote for myself
+    #[inline]
+    pub fn set_as_candidate(&mut self) {
+        self.current_term += 1;
+        self.role = Role::Candidate;
+        self.voted_for = Some(self.me_index);
+    }
+
+    // Vote for peer at peers[candidate_id]
+    #[inline]
+    pub fn vote_for(&mut self, candidate_id: u64) {
+        self.voted_for = Some(candidate_id as usize);
+    }
+
+    // Access me index
+    #[inline]
+    pub fn me(&self) -> usize {
+        self.me_index
+    }
+}