raft: remove p2p code from start function & add p2p get_info() to ircd rpc

bin/ircd/src/main.rs

@@ -14,9 +14,9 @@ use smol::future;
 use structopt_toml::StructOptToml;
 
 use darkfi::{
-    async_daemonize,
+    async_daemonize, net,
     net::transport::{TcpTransport, Transport},
-    raft::Raft,
+    raft::{NetMsg, ProtocolRaft, Raft},
     rpc::rpcserver::{listen_and_serve, RpcServerConfig},
     util::{
         cli::{log_config, spawn_config},
@@ -138,6 +138,32 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     let raft_sender = raft.get_broadcast();
     let raft_receiver = raft.get_commits();
 
+    // P2p setup
+    let (p2p_send_channel, p2p_recv_channel) = async_channel::unbounded::<NetMsg>();
+
+    let p2p = net::P2p::new(net_settings.into()).await;
+    let p2p = p2p.clone();
+
+    let registry = p2p.protocol_registry();
+
+    let seen_net_msg = Arc::new(Mutex::new(vec![]));
+    let raft_node_id = raft.id.clone();
+    registry
+        .register(net::SESSION_ALL, move |channel, p2p| {
+            let raft_node_id = raft_node_id.clone();
+            let sender = p2p_send_channel.clone();
+            let seen_net_msg_cloned = seen_net_msg.clone();
+            async move {
+                ProtocolRaft::init(raft_node_id, channel, sender, p2p, seen_net_msg_cloned).await
+            }
+        })
+        .await;
+
+    p2p.clone().start(executor.clone()).await?;
+
+    let executor_cloned = executor.clone();
+    let p2p_run_task = executor_cloned.spawn(p2p.clone().run(executor.clone()));
+
     //
     // RPC interface
     //
@@ -148,7 +174,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         identity_pass: Default::default(),
     };
     let executor_cloned = executor.clone();
-    let rpc_interface = Arc::new(JsonRpcInterface { addr: settings.rpc_listen });
+    let rpc_interface = Arc::new(JsonRpcInterface { addr: settings.rpc_listen, p2p: p2p.clone() });
     let rpc_task = executor.spawn(async move {
         listen_and_serve(rpc_config, rpc_interface, executor_cloned.clone()).await
     });
@@ -182,6 +208,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         Ok(())
     });
 
+    // Run once receive exit signal
     let (signal, shutdown) = async_channel::bounded::<()>(1);
     ctrlc_async::set_async_handler(async move {
         warn!(target: "ircd", "ircd start Exit Signal");
@@ -189,11 +216,12 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         signal.send(()).await.unwrap();
         rpc_task.cancel().await;
         irc_task.cancel().await;
+        p2p_run_task.cancel().await;
     })
     .unwrap();
 
     // blocking
-    raft.start(net_settings.into(), executor.clone(), shutdown.clone()).await?;
+    raft.start(p2p.clone(), p2p_recv_channel.clone(), executor.clone(), shutdown.clone()).await?;
 
     Ok(())
 }

bin/ircd/src/rpc.rs

@@ -5,14 +5,18 @@ use async_trait::async_trait;
 use log::debug;
 use serde_json::{json, Value};
 
-use darkfi::rpc::{
-    jsonrpc,
-    jsonrpc::{ErrorCode, JsonRequest, JsonResult},
-    rpcserver::RequestHandler,
+use darkfi::{
+    net,
+    rpc::{
+        jsonrpc,
+        jsonrpc::{ErrorCode, JsonRequest, JsonResult},
+        rpcserver::RequestHandler,
+    },
 };
 
 pub struct JsonRpcInterface {
     pub addr: SocketAddr,
+    pub p2p: net::P2pPtr,
 }
 
 #[async_trait]
@@ -26,7 +30,7 @@ impl RequestHandler for JsonRpcInterface {
 
         match req.method.as_str() {
             Some("ping") => self.pong(req.id, req.params).await,
-            //Some("get_info") => self.get_info(req.id, req.params).await,
+            Some("get_info") => self.get_info(req.id, req.params).await,
             Some(_) | None => jsonrpc::error(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
@@ -40,4 +44,13 @@ impl JsonRpcInterface {
     async fn pong(&self, id: Value, _params: Value) -> JsonResult {
         jsonrpc::response(json!("pong"), id).into()
     }
+
+    // RPCAPI:
+    // Retrieves P2P network information.
+    // --> {"jsonrpc": "2.0", "method": "get_info", "params": [], "id": 42}
+    // <-- {"jsonrpc": "2.0", result": {"nodeID": [], "nodeinfo": [], "id": 42}
+    async fn get_info(&self, id: Value, _params: Value) -> JsonResult {
+        let resp = self.p2p.get_info().await;
+        jsonrpc::response(resp, id).into()
+    }
 }

bin/tau/taud/src/main.rs

@@ -1,4 +1,4 @@
-use async_std::sync::Arc;
+use async_std::sync::{Arc, Mutex};
 use std::fs::create_dir_all;
 
 use async_executor::Executor;
@@ -11,8 +11,8 @@ use smol::future;
 use structopt_toml::StructOptToml;
 
 use darkfi::{
-    async_daemonize,
-    raft::Raft,
+    async_daemonize, net,
+    raft::{NetMsg, ProtocolRaft, Raft},
     rpc::rpcserver::{listen_and_serve, RpcServerConfig},
     util::{
         cli::{log_config, spawn_config},
@@ -189,6 +189,32 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         }
     });
 
+    // P2p setup
+    let (p2p_send_channel, p2p_recv_channel) = async_channel::unbounded::<NetMsg>();
+
+    let p2p = net::P2p::new(net_settings.into()).await;
+    let p2p = p2p.clone();
+
+    let registry = p2p.protocol_registry();
+
+    let seen_net_msg = Arc::new(Mutex::new(vec![]));
+    let raft_node_id = raft.id.clone();
+    registry
+        .register(net::SESSION_ALL, move |channel, p2p| {
+            let raft_node_id = raft_node_id.clone();
+            let sender = p2p_send_channel.clone();
+            let seen_net_msg_cloned = seen_net_msg.clone();
+            async move {
+                ProtocolRaft::init(raft_node_id, channel, sender, p2p, seen_net_msg_cloned).await
+            }
+        })
+        .await;
+
+    p2p.clone().start(executor.clone()).await?;
+
+    let executor_cloned = executor.clone();
+    let p2p_run_task = executor_cloned.spawn(p2p.clone().run(executor.clone()));
+
     let (signal, shutdown) = async_channel::bounded::<()>(1);
     ctrlc_async::set_async_handler(async move {
         warn!(target: "tau", "taud start() Exit Signal");
@@ -196,11 +222,12 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         signal.send(()).await.unwrap();
         rpc_listener_taks.cancel().await;
         recv_update.cancel().await;
+        p2p_run_task.cancel().await;
     })
     .unwrap();
 
     // blocking
-    raft.start(net_settings.into(), executor.clone(), shutdown.clone()).await?;
+    raft.start(p2p.clone(), p2p_recv_channel.clone(), executor.clone(), shutdown.clone()).await?;
 
     Ok(())
 }

src/raft/consensus.rs

@@ -20,7 +20,7 @@ use super::{
         Broadcast, BroadcastMsgRequest, Log, LogRequest, LogResponse, Logs, MapLength, NetMsg,
         NetMsgMethod, NodeId, Role, Sender, SyncRequest, SyncResponse, VoteRequest, VoteResponse,
     },
-    DataStore, ProtocolRaft,
+    DataStore,
 };
 
 const HEARTBEATTIMEOUT: u64 = 100;
@@ -48,10 +48,7 @@ async fn load_node_ids_loop(
     }
 }
 
-async fn p2p_recv_send_loop(
-    p2p_recv: async_channel::Receiver<NetMsg>,
-    p2p: net::P2pPtr,
-) -> Result<()> {
+async fn p2p_send_loop(p2p_recv: async_channel::Receiver<NetMsg>, p2p: net::P2pPtr) -> Result<()> {
     loop {
         let msg: NetMsg = match p2p_recv.recv().await {
             Ok(m) => m,
@@ -74,7 +71,7 @@ pub struct Raft<T> {
     // this will be derived from the ip
     // if the node doesn't have an id then will become a listener and doesn't have the right
     // to request/response votes or response a confirmation for log
-    id: Option<NodeId>,
+    pub id: Option<NodeId>,
 
     // these five vars should be on local storage
     current_term: u64,
@@ -159,39 +156,13 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
 
     pub async fn start(
         &mut self,
-        net_settings: net::Settings,
+        p2p: net::P2pPtr,
+        p2p_recv_channel: async_channel::Receiver<NetMsg>,
         executor: Arc<Executor<'_>>,
         stop_signal: async_channel::Receiver<()>,
     ) -> Result<()> {
-        // P2p setup
-        let (p2p_snd, receive_queues) = async_channel::unbounded::<NetMsg>();
-
-        let p2p = net::P2p::new(net_settings).await;
-        let p2p = p2p.clone();
-
-        let registry = p2p.protocol_registry();
-
-        let seen_net_msg = Arc::new(Mutex::new(vec![]));
-        let self_id = self.id.clone();
-        registry
-            .register(net::SESSION_ALL, move |channel, p2p| {
-                let self_id = self_id.clone();
-                let sender = p2p_snd.clone();
-                let seen_net_msg_cloned = seen_net_msg.clone();
-                async move {
-                    ProtocolRaft::init(self_id, channel, sender, p2p, seen_net_msg_cloned).await
-                }
-            })
-            .await;
-
-        // P2p start
-        p2p.clone().start(executor.clone()).await?;
-
-        let executor_cloned = executor.clone();
-        let p2p_task = executor_cloned.spawn(p2p.clone().run(executor.clone()));
-
-        let p2p_recv = self.sender.1.clone();
-        let p2p_recv_task = executor.spawn(p2p_recv_send_loop(p2p_recv.clone(), p2p.clone()));
+        let receiver = self.sender.1.clone();
+        let p2p_send_task = executor.spawn(p2p_send_loop(receiver.clone(), p2p.clone()));
 
         let load_ips_task =
             executor.spawn(load_node_ids_loop(self.nodes.clone(), p2p.clone(), self.role.clone()));
@@ -206,7 +177,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             info!("send sync request");
             self.send(None, &serialize(&sync_request), NetMsgMethod::SyncRequest, None).await?;
 
-            self.waiting_for_sync(receive_queues.clone(), stop_signal.clone()).await?;
+            self.waiting_for_sync(p2p_recv_channel.clone(), stop_signal.clone()).await?;
         }
 
         let mut rng = rand::thread_rng();
@@ -223,7 +194,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             let result: Result<()>;
 
             select! {
-                m =  receive_queues.recv().fuse() => result = self.handle_method(m?).await,
+                m =  p2p_recv_channel.recv().fuse() => result = self.handle_method(m?).await,
                 m =  broadcast_msg_rv.recv().fuse() => result = self.broadcast_msg(&m?,None).await,
                 _ = task::sleep(timeout).fuse() => {
                     result = if self.role == Role::Leader {
@@ -243,8 +214,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
 
         warn!(target: "raft", "Raft start() Exit Signal");
         load_ips_task.cancel().await;
-        p2p_recv_task.cancel().await;
-        p2p_task.cancel().await;
+        p2p_send_task.cancel().await;
         self.datastore.cancel().await?;
         Ok(())
     }