feat(raft): readiness waiting + termination

Author: Anton Melnikov

src/raft/inner.rs

@@ -6,18 +6,19 @@ use crate::raft::net::ConnectionId;
 use crate::raft::rpc;
 
 pub struct NodeInner {
-    state: State,
+    state: NodeState,
     local_id: u64,
     peers: BTreeMap<u64, Vec<SocketAddr>>,
     responded_ids: HashSet<u64>,
     bootstrap_addrs: Vec<Vec<SocketAddr>>,
 }
 
-#[derive(Debug, Copy, Clone)]
-enum State {
+#[derive(Debug, Copy, Clone, PartialEq)]
+pub enum NodeState {
     Init,
     ColdBootstrap,
     WarmBootstrap,
+    Ready,
     Offline,
     Done,
 }
@@ -27,6 +28,7 @@ pub enum NodeEvent {
     Request(rpc::BootstrapMsg),
     Response(rpc::BootstrapMsg),
     Timeout,
+    Stop,
 }
 
 #[derive(Debug)]
@@ -35,7 +37,7 @@ pub enum NodeAction {
     UpgradeSeed(ConnectionId, u64),
     Request(ConnectionId, rpc::BootstrapMsg),
     Response(Result<rpc::Response, Error>),
-    Completed,
+    StateChangeNotification(NodeState),
 }
 
 impl NodeInner {
@@ -48,7 +50,7 @@ impl NodeInner {
         peers.insert(local_id, local_addrs);
 
         NodeInner {
-            state: State::Init,
+            state: NodeState::Init,
             local_id,
             peers,
             responded_ids: Default::default(),
@@ -57,7 +59,7 @@ impl NodeInner {
     }
 
     pub fn update(&mut self, events: &mut VecDeque<NodeEvent>, actions: &mut VecDeque<NodeAction>) {
-        if let State::Init = self.state {
+        if let NodeState::Init = self.state {
             self.init(actions);
         }
 
@@ -66,19 +68,23 @@ impl NodeInner {
         }
     }
 
+    pub fn state(&self) -> &NodeState {
+        &self.state
+    }
+
     fn init(&mut self, actions_buf: &mut VecDeque<NodeAction>) {
         for (id, seed_addrs) in self.bootstrap_addrs.clone().into_iter().enumerate() {
             let id = ConnectionId::Seed(id);
             actions_buf.push_back(NodeAction::Connect(id.clone(), seed_addrs));
             self.send_bootstrap_request(id, actions_buf);
         }
 
-        self.state = State::ColdBootstrap;
+        self.state = NodeState::ColdBootstrap;
     }
 
     fn handle_event(&mut self, event: NodeEvent, actions_buf: &mut VecDeque<NodeAction>) {
         use NodeEvent as E;
-        use State as S;
+        use NodeState as S;
 
         let new_state = match (self.state, event) {
             (S::ColdBootstrap, E::Request(req))
@@ -93,19 +99,22 @@ impl NodeInner {
                 let num_peers = self.peers.len();
                 let num_responded = self.responded_ids.len();
                 if num_peers == (num_responded + 1) {
-                    actions_buf.push_back(NodeAction::Completed);
-                    Some(S::Done)
+                    Some(S::Ready)
                 } else {
                     None
                 }
             }
             (S::ColdBootstrap, E::Timeout) => Some(S::Offline),
             (S::Offline, E::Timeout) => None,
+            (_, E::Stop) => Some(S::Done),
             _ => panic!("invalid state"),
         };
 
         if let Some(new_state) = new_state {
-            self.state = new_state;
+            if self.state != new_state {
+                actions_buf.push_back(NodeAction::StateChangeNotification(new_state.clone()));
+                self.state = new_state;
+            }
         }
     }
 

src/raft/mod.rs

@@ -2,12 +2,13 @@
 
 use std::cell::RefCell;
 use std::collections::VecDeque;
+use std::io;
 use std::net::ToSocketAddrs;
 use std::time::Duration;
 
 use rand::random;
 
-use inner::{NodeAction, NodeInner};
+use inner::{NodeAction, NodeInner, NodeState};
 use net::{get_local_addrs, ConnectionPool};
 
 use crate::error::{Error, TarantoolErrorCode};
@@ -29,6 +30,7 @@ pub struct Node {
     events_cond: Cond,
     events_buffer: RefCell<VecDeque<NodeEvent>>,
     actions_buffer: RefCell<VecDeque<NodeAction>>,
+    ready_cond: Cond,
     options: NodeOptions,
 }
 
@@ -75,6 +77,7 @@ impl Node {
             events_cond: Cond::new(),
             events_buffer: RefCell::new(VecDeque::with_capacity(options.recv_queue_size)),
             actions_buffer: RefCell::new(VecDeque::with_capacity(options.send_queue_size)),
+            ready_cond: Cond::new(),
             options,
         })
     }
@@ -97,9 +100,13 @@ impl Node {
                         self.send(conn_pool.get(&to).unwrap(), rpc::Request::Bootstrap(msg))?;
                     }
                     NodeAction::Response(_) => {}
-                    NodeAction::Completed => {
-                        return Ok(());
-                    }
+                    NodeAction::StateChangeNotification(state) => match state {
+                        NodeState::Ready => {
+                            self.ready_cond.signal();
+                        }
+                        NodeState::Done => return Ok(()),
+                        _ => {}
+                    },
                     _ => {}
                 };
             }
@@ -119,7 +126,7 @@ impl Node {
                         self.events_buffer
                             .borrow_mut()
                             .push_back(NodeEvent::Request(msg));
-                        self.events_cond.wait();
+                        self.events_cond.signal();
                     }
                     _ => unimplemented!(),
                 };
@@ -131,11 +138,17 @@ impl Node {
     }
 
     pub fn wait_ready(&self, timeout: Duration) -> Result<(), Error> {
-        unimplemented!();
+        if self.inner.borrow().state() != &NodeState::Ready {
+            if !self.ready_cond.wait_timeout(timeout) {
+                return Err(Error::IO(io::ErrorKind::TimedOut.into()));
+            }
+        }
+        Ok(())
     }
 
     pub fn close(&self) {
-        unimplemented!();
+        self.events_buffer.borrow_mut().push_back(NodeEvent::Stop);
+        self.events_cond.signal();
     }
 
     fn send(&self, conn: &Conn, request: rpc::Request) -> Result<Option<rpc::Response>, Error> {

tests/src/test_raft.rs

@@ -1,6 +1,6 @@
 use std::collections::VecDeque;
 
-use tarantool::raft::inner::{NodeAction, NodeEvent, NodeInner};
+use tarantool::raft::inner::{NodeAction, NodeEvent, NodeInner, NodeState};
 use tarantool::raft::net::ConnectionId;
 use tarantool::raft::rpc;
 
@@ -34,7 +34,7 @@ pub fn test_bootstrap_solo() {
     }));
     node.update(&mut events, &mut actions);
 
-    assert_eq!(actions.len(), 2);
+    assert_eq!(actions.len(), 3);
     assert!(matches!(
         &actions[0],
         NodeAction::Connect(ConnectionId::Peer(2), addrs) if addrs == &remote_addrs
@@ -58,7 +58,10 @@ pub fn test_bootstrap_solo() {
     node.update(&mut events, &mut actions);
 
     assert_eq!(actions.len(), 1);
-    assert!(matches!(&actions[0], NodeAction::Completed))
+    assert!(matches!(
+        &actions[0],
+        NodeAction::StateChangeNotification(NodeState::Ready)
+    ))
 }
 
 pub fn test_bootstrap_2n() {
@@ -91,7 +94,7 @@ pub fn test_bootstrap_2n() {
 fn communicate(from: &mut VecDeque<NodeAction>, to: &mut VecDeque<NodeEvent>) -> bool {
     let mut is_completed = false;
     for action in from.drain(..) {
-        if let NodeAction::Completed = action {
+        if let NodeAction::StateChangeNotification(NodeState::Ready) = action {
             is_completed = true;
         }