feat(raft): refactor bootstrap fsm. Part 1

Author: Anton Melnikov

src/raft/bootstrap.rs

@@ -4,6 +4,7 @@ use std::net::SocketAddr;
 
 use crate::error::Error;
 
+use super::net::ConnectionId;
 use super::rpc;
 
 pub struct BoostrapController {
@@ -30,7 +31,9 @@ pub enum BootstrapEvent {
 
 #[derive(Debug)]
 pub enum BootstrapAction {
-    Request(rpc::BootstrapMsg, Vec<SocketAddr>),
+    Connect(ConnectionId, Vec<SocketAddr>),
+    UpgradeSeed(ConnectionId, u64),
+    Request(ConnectionId, rpc::BootstrapMsg),
     Response(Result<rpc::Response, Error>),
     Completed,
 }
@@ -51,7 +54,7 @@ impl BoostrapController {
             responded_ids: Default::default(),
             pending_actions_buffer: Default::default(),
         };
-        bootstrap_controller.broadcast(bootstrap_addrs.into_iter());
+        bootstrap_controller.poll_seeds(bootstrap_addrs.into_iter());
         bootstrap_controller
     }
 
@@ -103,38 +106,39 @@ impl BoostrapController {
         let mut responded_ids = self.responded_ids.borrow_mut();
         if !responded_ids.contains(&req.from_id) {
             let new_nodes = self.merge_nodes_list(&req.nodes);
-            let new_nodes = new_nodes.iter().map(|(_, addrs)| addrs.clone());
-            self.broadcast(new_nodes);
-
+            for (id, addrs) in new_nodes {
+                let id = ConnectionId::Peer(id);
+                self.send(BootstrapAction::Connect(id.clone(), addrs));
+                self.send_bootstrap_request(id);
+            }
             responded_ids.insert(req.from_id);
         }
     }
 
     #[inline]
-    fn broadcast(&self, addrs: impl Iterator<Item = Vec<SocketAddr>>) {
-        for peer_addrs in addrs {
-            self.send_request(peer_addrs.clone());
+    fn poll_seeds(&self, addrs: impl Iterator<Item = Vec<SocketAddr>>) {
+        for (id, seed_addrs) in addrs.enumerate() {
+            let id = ConnectionId::Seed(id);
+            self.send(BootstrapAction::Connect(id.clone(), seed_addrs));
+            self.send_bootstrap_request(id);
         }
     }
 
     #[inline]
-    fn send_request(&self, to: Vec<SocketAddr>) {
+    fn send_bootstrap_request(&self, to: ConnectionId) {
         let nodes = self
             .peers
             .borrow()
             .iter()
-            .map(|(a, b)| (*a, b.clone()))
+            .map(|(id, addrs)| (*id, addrs.clone()))
             .collect();
 
-        let peers = self.peers.borrow();
-        let local_addrs = peers.get(&self.local_id).unwrap();
         self.send(BootstrapAction::Request(
+            to,
             rpc::BootstrapMsg {
                 from_id: self.local_id,
-                from_addrs: local_addrs.clone(),
                 nodes,
             },
-            to,
         ));
     }
 

src/raft/mod.rs

@@ -1,28 +1,22 @@
 #![cfg(feature = "raft_node")]
 
-use std::cell::{Cell, RefCell};
-use std::io;
+use std::cell::RefCell;
 use std::net::ToSocketAddrs;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
-use protobuf::Message as _;
-use raft::prelude::Message;
 use rand::random;
 
 use bootstrap::{BoostrapController, BootstrapAction};
 use net::{get_local_addrs, ConnectionPoll};
 
-use crate::error::{Error, TarantoolErrorCode};
-use crate::fiber::{Cond, Latch};
+use crate::error::Error;
 use crate::net_box::{Conn, ConnOptions, Options};
-use crate::tuple::{FunctionArgs, FunctionCtx, Tuple};
-
-use self::inner::NodeInner;
+use crate::tuple::{FunctionArgs, FunctionCtx};
 
 pub mod bootstrap;
 mod fsm;
 mod inner;
-mod net;
+pub mod net;
 pub mod rpc;
 mod storage;
 
@@ -34,11 +28,7 @@ pub enum NodeState {
 }
 
 pub struct Node {
-    state: Cell<NodeState>,
-    state_lock: Latch,
-    state_cond: Cond,
     bootstrap_ctrl: BoostrapController,
-    inner: RefCell<NodeInner>,
     connections: RefCell<ConnectionPoll>,
     rpc_function: String,
     options: NodeOptions,
@@ -81,111 +71,41 @@ impl Node {
         }
 
         Ok(Node {
-            state: Cell::new(NodeState::Bootstrapping),
-            state_lock: Latch::new(),
-            state_cond: Cond::new(),
             bootstrap_ctrl: BoostrapController::new(id, local_addrs, bootstrap_addrs_cfg),
-            inner: RefCell::new(NodeInner::new(id, &options)?),
-            connections: Default::default(),
+            connections: RefCell::new(ConnectionPoll::new(options.connection_options.clone())),
             rpc_function: rpc_function.to_string(),
             options,
         })
     }
 
     pub fn run(&self) -> Result<(), Error> {
         loop {
-            let mut is_state_changed = false;
-            {
-                let _lock = self.state_lock.lock();
-                let next_state = match self.state.get() {
-                    NodeState::Bootstrapping => {
-                        for action in self.bootstrap_ctrl.pending_actions() {
-                            match action {
-                                BootstrapAction::Request(msg, to) => {
-                                    let mut conn_pool = self.connections.borrow_mut();
-                                    self.send(
-                                        conn_pool.connect_or_get(None, &to)?,
-                                        rpc::Request::Bootstrap(msg),
-                                    )?;
-                                }
-                                BootstrapAction::Response(_) => {}
-                                BootstrapAction::Completed => {}
-                            };
-                        }
-
-                        Some(NodeState::Closed)
+            for action in self.bootstrap_ctrl.pending_actions() {
+                match action {
+                    BootstrapAction::Request(to, msg) => {
+                        let mut conn_pool = self.connections.borrow_mut();
+                        self.send(conn_pool.get(&to).unwrap(), rpc::Request::Bootstrap(msg))?;
                     }
-                    NodeState::Active => {
-                        unimplemented!()
+                    BootstrapAction::Response(_) => {}
+                    BootstrapAction::Completed => {
+                        return Ok(());
                     }
-                    NodeState::Closed => break,
+                    _ => {}
                 };
-
-                if let Some(next_state) = next_state {
-                    self.state.replace(next_state);
-                    is_state_changed = true;
-                }
-            }
-
-            if is_state_changed {
-                self.state_cond.broadcast();
             }
         }
-
-        Ok(())
     }
 
     pub fn handle_rpc(&self, ctx: FunctionCtx, args: FunctionArgs) -> i32 {
-        let args: Tuple = args.into();
-
-        match args.into_struct::<rpc::Request>() {
-            Err(e) => set_error!(TarantoolErrorCode::Protocol, "{}", e),
-            Ok(request) => {
-                let response = match request {
-                    rpc::Request::Bootstrap(msg) => self.recv_bootstrap_request(msg),
-                    rpc::Request::Raft { data: msg_data } => {
-                        let mut msg = Message::default();
-                        match msg.merge_from_bytes(&msg_data) {
-                            Err(e) => {
-                                return set_error!(TarantoolErrorCode::Protocol, "{}", e);
-                            }
-                            Ok(()) => {
-                                let _lock = self.state_lock.lock();
-                                if let NodeState::Active = self.state.get() {
-                                    self.inner.borrow().handle_msg(msg);
-                                }
-                            }
-                        }
-                        rpc::Response::Ack
-                    }
-                    _ => unimplemented!(),
-                };
-
-                ctx.return_mp(&response)
-                    .unwrap_or_else(|e| set_error!(TarantoolErrorCode::ProcC, "{}", e))
-            }
-        }
+        unimplemented!();
     }
 
     pub fn wait_ready(&self, timeout: Duration) -> Result<(), Error> {
-        let started_at = Instant::now();
-        while self.state.get() != NodeState::Active {
-            let is_timeout = !match timeout.checked_sub(started_at.elapsed()) {
-                None => false,
-                Some(timeout) => self.state_cond.wait_timeout(timeout),
-            };
-
-            if is_timeout {
-                return Err(io::Error::from(io::ErrorKind::TimedOut).into());
-            }
-        }
-
-        Ok(())
+        unimplemented!();
     }
 
     pub fn close(&self) {
-        let _lock = self.state_lock.lock();
-        self.state.replace(NodeState::Closed);
+        unimplemented!();
     }
 
     fn send(&self, conn: &Conn, request: rpc::Request) -> Result<Option<rpc::Response>, Error> {
@@ -207,12 +127,4 @@ impl Node {
             },
         }
     }
-
-    fn recv_bootstrap_request(&self, request: rpc::BootstrapMsg) -> rpc::Response {
-        unimplemented!()
-    }
-
-    fn send_raft_batch(&self, msgs: &mut dyn Iterator<Item = Message>) -> Result<(), Error> {
-        unimplemented!()
-    }
 }

src/raft/net.rs

@@ -7,56 +7,36 @@ use std::ptr::null_mut;
 use ipnetwork::{Ipv4Network, Ipv6Network};
 
 use crate::error::Error;
-
-use super::{Conn, ConnOptions};
+use crate::net_box::{Conn, ConnOptions};
 
 #[derive(Default)]
 pub struct ConnectionPoll {
-    connections: Vec<Conn>,
-    ids_index: HashMap<u64, usize>,
-    addrs_index: HashMap<SocketAddr, usize>,
     options: ConnOptions,
+    inner: HashMap<ConnectionId, Conn>,
 }
 
-impl ConnectionPoll {
-    pub fn connect_or_get(
-        &mut self,
-        id: Option<u64>,
-        addrs: &Vec<SocketAddr>,
-    ) -> Result<&Conn, Error> {
-        let inner_id = if let Some(id) = id {
-            self.ids_index.get(&id)
-        } else {
-            let mut inner_id = None;
-            for addr in addrs {
-                inner_id = self.addrs_index.get(addr);
-                if inner_id.is_some() {
-                    break;
-                }
-            }
-            inner_id
-        };
-
-        let inner_id = match inner_id {
-            Some(inner_id) => *inner_id,
-            None => {
-                let inner_id = self.connections.len();
-                self.connections
-                    .push(Conn::new(addrs.as_slice(), self.options.clone(), None)?);
-
-                if let Some(id) = id {
-                    self.ids_index.insert(id, inner_id);
-                }
+#[derive(Clone, Debug, Hash, Eq, PartialEq)]
+pub enum ConnectionId {
+    Seed(usize),
+    Peer(u64),
+}
 
-                for addr in addrs {
-                    self.addrs_index.insert(addr.clone(), inner_id);
-                }
+impl ConnectionPoll {
+    pub fn new(options: ConnOptions) -> Self {
+        ConnectionPoll {
+            options,
+            inner: HashMap::new(),
+        }
+    }
 
-                inner_id
-            }
-        };
+    pub fn connect(&mut self, id: ConnectionId, addrs: impl ToSocketAddrs) -> Result<(), Error> {
+        let conn = Conn::new(addrs, self.options.clone(), None)?;
+        self.inner.insert(id, conn);
+        Ok(())
+    }
 
-        Ok(self.connections.get(inner_id).unwrap())
+    pub fn get(&mut self, id: &ConnectionId) -> Option<&mut Conn> {
+        self.inner.get_mut(id)
     }
 }
 

src/raft/rpc.rs

@@ -34,7 +34,6 @@ impl AsTuple for Response {}
 #[derive(Debug, PartialEq, Serialize, Deserialize)]
 pub struct BootstrapMsg {
     pub from_id: u64,
-    pub from_addrs: Vec<SocketAddr>,
     pub nodes: Vec<(u64, Vec<SocketAddr>)>,
 }