Merge branch 'feature/raft-bootstrap-part5' into 'master'

Feature/raft bootstrap part5

See merge request picodata/brod/tarantool-module!41

Author: volt0

examples/cluster_node/src/lib.rs

@@ -6,7 +6,7 @@ use std::os::raw::c_int;
 use std::rc::{Rc, Weak};
 
 use tarantool::raft::Node;
-use tarantool::tuple::{FunctionArgs, FunctionCtx};
+use tarantool::tuple::{FunctionArgs, FunctionCtx, Tuple};
 
 #[derive(Default)]
 struct Global {
@@ -21,15 +21,12 @@ lazy_static! {
 }
 
 #[no_mangle]
-pub extern "C" fn run_node(_: FunctionCtx, _: FunctionArgs) -> c_int {
-    let node = Rc::new(
-        Node::new(
-            "libcluster_node.rpc",
-            vec!["127.0.0.1:3301", "127.0.0.1:3302", "127.0.0.1:3303"],
-            Default::default(),
-        )
-        .unwrap(),
-    );
+pub extern "C" fn run_node(_: FunctionCtx, args: FunctionArgs) -> c_int {
+    let args: Tuple = args.into();
+    let (bootstrap_addrs,) = args.into_struct::<(Vec<String>,)>().unwrap();
+
+    let node =
+        Rc::new(Node::new("libcluster_node.rpc", bootstrap_addrs, Default::default()).unwrap());
     GLOBAL.node.replace(Rc::downgrade(&node));
     node.run().unwrap();
     0

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,17 +2,20 @@
 
 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;
+use crate::error::{Error, TarantoolErrorCode};
+use crate::fiber::Cond;
 use crate::net_box::{Conn, ConnOptions, Options};
-use crate::tuple::{FunctionArgs, FunctionCtx};
+use crate::raft::inner::NodeEvent;
+use crate::tuple::{FunctionArgs, FunctionCtx, Tuple};
 
 mod fsm;
 pub mod inner;
@@ -24,6 +27,10 @@ pub struct Node {
     inner: RefCell<NodeInner>,
     connections: RefCell<ConnectionPool>,
     rpc_function: String,
+    events_cond: Cond,
+    events_buffer: RefCell<VecDeque<NodeEvent>>,
+    actions_buffer: RefCell<VecDeque<NodeAction>>,
+    ready_cond: Cond,
     options: NodeOptions,
 }
 
@@ -67,41 +74,81 @@ impl Node {
             inner: RefCell::new(NodeInner::new(id, local_addrs, bootstrap_addrs_cfg)),
             connections: RefCell::new(ConnectionPool::new(options.connection_options.clone())),
             rpc_function: rpc_function.to_string(),
+            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,
         })
     }
 
     pub fn run(&self) -> Result<(), Error> {
-        let mut events = VecDeque::new();
-        let mut actions = VecDeque::new();
         loop {
-            self.inner.borrow_mut().update(&mut events, &mut actions);
-            for action in actions.drain(..) {
+            {
+                let mut actions = self.actions_buffer.borrow_mut();
+                let mut events = self.events_buffer.borrow_mut();
+                self.inner.borrow_mut().update(&mut events, &mut actions);
+            }
+
+            for action in self.actions_buffer.borrow_mut().drain(..) {
                 match action {
+                    NodeAction::Connect(id, addrs) => {
+                        self.connections.borrow_mut().connect(id, &addrs[..])?;
+                    }
                     NodeAction::Request(to, msg) => {
                         let mut conn_pool = self.connections.borrow_mut();
                         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(()),
+                        _ => {}
+                    },
                     _ => {}
                 };
             }
+
+            self.events_cond.wait();
         }
     }
 
     pub fn handle_rpc(&self, ctx: FunctionCtx, args: FunctionArgs) -> i32 {
-        unimplemented!();
+        let args: Tuple = args.into();
+
+        match args.into_struct::<rpc::Request>() {
+            Err(e) => set_error!(TarantoolErrorCode::Protocol, "{}", e),
+            Ok(request) => {
+                match request {
+                    rpc::Request::Bootstrap(msg) => {
+                        self.events_buffer
+                            .borrow_mut()
+                            .push_back(NodeEvent::Request(msg));
+                        self.events_cond.signal();
+                    }
+                    _ => unimplemented!(),
+                };
+
+                ctx.return_mp(&rpc::Response::Ack)
+                    .unwrap_or_else(|e| set_error!(TarantoolErrorCode::ProcC, "{}", e))
+            }
+        }
     }
 
     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;
         }