fix storage (#32)

Signed-off-by: dierbei <[email protected]>

Author: dierbei

examples/simple_run.rs

@@ -9,8 +9,6 @@ use slog::{error, info};
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::str::FromStr;
-use std::thread;
-use tokio::runtime::Runtime;
 use tokio::time::Duration;
 
 use raft_rs::network::{NetworkLayer, TCPManager};
@@ -34,7 +32,7 @@ async fn main() {
         .clone()
         .iter()
         .map(|n| ServerConfig {
-            election_timeout: Duration::from_millis(200),
+            election_timeout: Duration::from_millis(1000),
             address: n.address,
             default_leader: Some(1u32),
             leadership_preferences: HashMap::new(),
@@ -47,20 +45,18 @@ async fn main() {
     for (i, config) in configs.into_iter().enumerate() {
         let id = cluster_nodes[i];
         let cc = cluster_config.clone();
-        handles.push(thread::spawn(move || {
-            let rt = Runtime::new().unwrap();
-            let mut server = Server::new(id, config, cc);
-            rt.block_on(server.start());
+        handles.push(tokio::spawn(async move {
+            let mut server = Server::new(id, config, cc).await;
+            server.start().await;
         }));
     }
 
     // Simulate a client request after some delay
     tokio::time::sleep(Duration::from_secs(20)).await;
     client_request(1, 42u32).await;
     tokio::time::sleep(Duration::from_secs(2)).await;
-    // Join all server threads
     for handle in handles {
-        handle.join().unwrap();
+        handle.await.unwrap();
     }
 }
 

examples/simulate_add_node.rs

@@ -7,8 +7,6 @@ use slog::error;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::str::FromStr;
-use std::thread;
-use tokio::runtime::Runtime;
 use tokio::time::Duration;
 
 use raft_rs::network::{NetworkLayer, TCPManager};
@@ -17,15 +15,15 @@ use raft_rs::server::{Server, ServerConfig};
 #[tokio::main]
 async fn main() {
     // Define cluster configuration
-    let mut cluster_nodes = vec![1, 2, 3, 4, 5];
+    let cluster_nodes = vec![1, 2, 3, 4, 5];
     let peers = vec![
         NodeMeta::from((1, SocketAddr::from_str("127.0.0.1:5001").unwrap())),
         NodeMeta::from((2, SocketAddr::from_str("127.0.0.1:5002").unwrap())),
         NodeMeta::from((3, SocketAddr::from_str("127.0.0.1:5003").unwrap())),
         NodeMeta::from((4, SocketAddr::from_str("127.0.0.1:5004").unwrap())),
         NodeMeta::from((5, SocketAddr::from_str("127.0.0.1:5005").unwrap())),
     ];
-    let mut cluster_config = ClusterConfig::new(peers.clone());
+    let cluster_config = ClusterConfig::new(peers.clone());
     // Create server configs
     let configs: Vec<_> = peers
         .clone()
@@ -44,10 +42,9 @@ async fn main() {
     for (i, config) in configs.into_iter().enumerate() {
         let id = cluster_nodes[i];
         let cc = cluster_config.clone();
-        handles.push(thread::spawn(move || {
-            let rt = Runtime::new().unwrap();
-            let mut server = Server::new(id, config, cc);
-            rt.block_on(server.start());
+        handles.push(tokio::spawn(async move {
+            let mut server = Server::new(id, config, cc).await;
+            server.start().await;
         }));
     }
 
@@ -66,28 +63,25 @@ async fn main() {
     };
 
     // Launching a new node
-    handles.push(thread::spawn(move || {
-        let rt = Runtime::new().unwrap();
-        let mut server = Server::new(new_node_id, new_node_conf, cluster_config);
-        rt.block_on(server.start());
+    handles.push(tokio::spawn(async move {
+        let mut server = Server::new(new_node_id, new_node_conf, cluster_config).await;
+        server.start().await;
     }));
 
     // Simulate sending a Raft Join request after a few seconds
     // Because we need to wait until the new node has started
     tokio::time::sleep(Duration::from_secs(3)).await;
     add_node_request(new_node_id, new_node_address).await;
 
-    // Wait for all servers to finish
     for handle in handles {
-        handle.join().unwrap();
+        handle.await.unwrap();
     }
 }
 
 async fn add_node_request(new_node_id: u32, addr: SocketAddr) {
     let log = get_logger();
 
-    let server_address = addr;
-    let network_manager = TCPManager::new(server_address);
+    let network_manager = TCPManager::new(addr);
 
     let request_data = vec![
         new_node_id.to_be_bytes().to_vec(),
@@ -98,7 +92,13 @@ async fn add_node_request(new_node_id: u32, addr: SocketAddr) {
     .concat();
 
     // Let's assume that 5001 is the port of the leader node.
-    if let Err(e) = network_manager.send(&server_address, &request_data).await {
+    if let Err(e) = network_manager
+        .send(
+            &SocketAddr::from_str("127.0.0.1:5001").unwrap(),
+            &request_data,
+        )
+        .await
+    {
         error!(log, "Failed to send client request: {}", e);
     }
 }

examples/simulate_node_failure.rs

@@ -45,7 +45,7 @@ async fn main() {
         let id = cluster_nodes[i];
         let cc = cluster_config.clone();
         let server_handle = tokio::spawn(async move {
-            let mut server = Server::new(id, config, cc);
+            let mut server = Server::new(id, config, cc).await;
             server.start().await;
         });
         server_handles.push(server_handle);
@@ -77,7 +77,7 @@ async fn main() {
         };
         let cc = cluster_config.clone();
         let server_handle = tokio::spawn(async move {
-            let mut server = Server::new(server_to_stop.try_into().unwrap(), config, cc);
+            let mut server = Server::new(server_to_stop.try_into().unwrap(), config, cc).await;
             server.start().await;
         });
         server_handles[server_to_stop - 1] = server_handle;

src/error.rs

@@ -33,8 +33,8 @@ pub enum NetworkError {
     ConnectError(SocketAddr),
     #[error("Failed binding to {0}")]
     BindError(SocketAddr),
-    #[error("Broadcast failed")]
-    BroadcastError,
+    #[error("Broadcast failed, errmsg: {0}")]
+    BroadcastError(String),
 }
 
 #[derive(Error, Debug)]

src/network.rs

@@ -79,7 +79,7 @@ impl NetworkLayer for TCPManager {
             .into_iter()
             .collect::<std::result::Result<_, _>>()
             // FIXME: We should let client decide what to do with the errors
-            .map_err(|_e| NetworkError::BroadcastError)?;
+            .map_err(|e| NetworkError::BroadcastError(e.to_string()))?;
         Ok(())
     }