feat(raft): single entry point for inner (update method)

Author: Anton Melnikov

src/raft/inner.rs

@@ -21,6 +21,7 @@ enum State {
     Done,
 }
 
+#[derive(Debug)]
 pub enum NodeEvent {
     Request(rpc::BootstrapMsg),
     Response(rpc::BootstrapMsg),
@@ -56,13 +57,17 @@ impl NodeInner {
         node_inner
     }
 
-    pub fn pending_actions(&mut self) -> Vec<NodeAction> {
-        self.pending_actions_buffer
-            .drain(..)
-            .collect::<Vec<NodeAction>>()
+    pub fn update(&mut self, events: &mut VecDeque<NodeEvent>, actions: &mut VecDeque<NodeAction>) {
+        while let Some(event) = events.pop_front() {
+            self.handle_event(event);
+        }
+
+        for action in self.pending_actions_buffer.drain(..) {
+            actions.push_back(action);
+        }
     }
 
-    pub fn handle_event(&mut self, event: NodeEvent) {
+    fn handle_event(&mut self, event: NodeEvent) {
         use NodeEvent as E;
         use State as S;
 

src/raft/mod.rs

@@ -1,6 +1,7 @@
 #![cfg(feature = "raft_node")]
 
 use std::cell::RefCell;
+use std::collections::VecDeque;
 use std::net::ToSocketAddrs;
 use std::time::Duration;
 
@@ -71,8 +72,11 @@ impl Node {
     }
 
     pub fn run(&self) -> Result<(), Error> {
+        let mut events = VecDeque::new();
+        let mut actions = VecDeque::new();
         loop {
-            for action in self.inner.borrow_mut().pending_actions() {
+            self.inner.borrow_mut().update(&mut events, &mut actions);
+            for action in actions.drain(..) {
                 match action {
                     NodeAction::Request(to, msg) => {
                         let mut conn_pool = self.connections.borrow_mut();

tests/src/test_raft.rs

@@ -1,3 +1,5 @@
+use std::collections::VecDeque;
+
 use tarantool::raft::inner::{NodeAction, NodeEvent, NodeInner};
 use tarantool::raft::net::ConnectionId;
 use tarantool::raft::rpc;
@@ -6,9 +8,12 @@ pub fn test_bootstrap_solo() {
     let local_addrs = vec!["127.0.0.1:3301".parse().unwrap()];
     let remote_addrs = vec!["127.0.0.1:3302".parse().unwrap()];
 
+    let mut events = VecDeque::new();
+    let mut actions = VecDeque::new();
+
     let mut node = NodeInner::new(1, local_addrs.clone(), vec![remote_addrs.clone()]);
+    node.update(&mut events, &mut actions);
 
-    let actions = node.pending_actions();
     assert_eq!(actions.len(), 2);
     assert!(matches!(
         &actions[0],
@@ -21,13 +26,14 @@ pub fn test_bootstrap_solo() {
             nodes: vec![(1, local_addrs.clone())],
         }
     ));
+    actions.clear();
 
-    node.handle_event(NodeEvent::Response(rpc::BootstrapMsg {
+    events.push_back(NodeEvent::Response(rpc::BootstrapMsg {
         from_id: 2,
         nodes: vec![(2, remote_addrs.clone())],
     }));
+    node.update(&mut events, &mut actions);
 
-    let actions = node.pending_actions();
     assert_eq!(actions.len(), 2);
     assert!(matches!(
         &actions[0],
@@ -43,13 +49,14 @@ pub fn test_bootstrap_solo() {
             ],
         }
     ));
+    actions.clear();
 
-    node.handle_event(NodeEvent::Response(rpc::BootstrapMsg {
+    events.push_back(NodeEvent::Response(rpc::BootstrapMsg {
         from_id: 2,
         nodes: vec![(1, local_addrs.clone()), (2, remote_addrs.clone())],
     }));
+    node.update(&mut events, &mut actions);
 
-    let actions = node.pending_actions();
     assert_eq!(actions.len(), 1);
     assert!(matches!(&actions[0], NodeAction::Completed))
 }
@@ -58,44 +65,50 @@ pub fn test_bootstrap_2n() {
     let n1_addrs = vec!["127.0.0.1:3301".parse().unwrap()];
     let n2_addrs = vec!["127.0.0.1:3302".parse().unwrap()];
 
+    let mut n1_events = VecDeque::new();
+    let mut n1_actions = VecDeque::new();
     let mut n1_ctrl = NodeInner::new(1, n1_addrs.clone(), vec![n2_addrs.clone()]);
+
+    let mut n2_events = VecDeque::new();
+    let mut n2_actions = VecDeque::new();
     let mut n2_ctrl = NodeInner::new(2, n2_addrs.clone(), vec![n1_addrs.clone()]);
 
-    assert_eq!(communicate(&mut n1_ctrl, &mut n2_ctrl), (false, false));
-    assert_eq!(communicate(&mut n1_ctrl, &mut n2_ctrl), (false, false));
-    assert_eq!(communicate(&mut n1_ctrl, &mut n2_ctrl), (true, true));
-}
+    let mut n1_is_completed = false;
+    let mut n2_is_completed = false;
 
-fn communicate(n1_ctrl: &mut NodeInner, n2_ctrl: &mut NodeInner) -> (bool, bool) {
-    let n1_actions = n1_ctrl.pending_actions();
-    let n2_actions = n2_ctrl.pending_actions();
+    for _ in 0..3 {
+        n1_ctrl.update(&mut n1_events, &mut n1_actions);
+        n2_ctrl.update(&mut n2_events, &mut n2_actions);
 
-    let mut n1_is_completed = false;
-    for action in n1_actions {
-        if let NodeAction::Completed = action {
-            n1_is_completed = true;
-        }
-        forward_action(action, n2_ctrl);
+        n1_is_completed = n1_is_completed || communicate(&mut n1_actions, &mut n2_events);
+        n2_is_completed = n2_is_completed || communicate(&mut n2_actions, &mut n1_events);
     }
 
-    let mut n2_is_completed = false;
-    for action in n2_actions {
+    assert!(n1_is_completed);
+    assert!(n2_is_completed);
+}
+
+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 {
-            n2_is_completed = true;
+            is_completed = true;
         }
-        forward_action(action, n1_ctrl);
-    }
 
-    (n1_is_completed, n2_is_completed)
+        if let Some(event) = forward_action(action) {
+            to.push_back(event);
+        }
+    }
+    is_completed
 }
 
-fn forward_action(action: NodeAction, node_ctrl: &mut NodeInner) {
+fn forward_action(action: NodeAction) -> Option<NodeEvent> {
     match action {
-        NodeAction::Request(_, msg) => node_ctrl.handle_event(NodeEvent::Request(msg)),
+        NodeAction::Request(_, msg) => Some(NodeEvent::Request(msg)),
         NodeAction::Response(resp) => match resp.unwrap() {
-            rpc::Response::Bootstrap(msg) => node_ctrl.handle_event(NodeEvent::Response(msg)),
-            _ => {}
+            rpc::Response::Bootstrap(msg) => Some(NodeEvent::Response(msg)),
+            _ => None,
         },
-        _ => {}
-    };
+        _ => None,
+    }
 }