ircd: WIP removing raft

bin/ircd/Cargo.toml

@@ -9,7 +9,7 @@ license = "AGPL-3.0-only"
 edition = "2021"
 
 [dependencies]
-darkfi = {path = "../../", features = ["net", "rpc", "raft"]}
+darkfi = {path = "../../", features = ["net", "rpc"]}
 # Async
 smol = "1.2.5"
 futures = "0.3.21"

bin/ircd/src/main.rs

@@ -1,11 +1,12 @@
-use std::{net::SocketAddr, sync::atomic::Ordering};
-
-use async_channel::{Receiver, Sender};
-use async_executor::Executor;
 use async_std::{
     net::{TcpListener, TcpStream},
     sync::{Arc, Mutex},
 };
+use std::{net::SocketAddr, sync::atomic::Ordering};
+
+use async_channel::Receiver;
+use async_executor::Executor;
+
 use futures::{io::BufReader, AsyncBufReadExt, AsyncReadExt, FutureExt};
 use fxhash::FxHashMap;
 use log::{debug, error, info, warn};
@@ -15,31 +16,30 @@ use structopt_toml::StructOptToml;
 
 use darkfi::{
     async_daemonize, net,
-    raft::{NetMsg, ProtocolRaft, Raft},
     rpc::server::listen_and_serve,
     util::{
         cli::{get_log_config, get_log_level, spawn_config},
-        path::{expand_path, get_config_path},
+        path::get_config_path,
     },
     Error, Result,
 };
 
 pub mod crypto;
 pub mod privmsg;
+pub mod protocol_privmsg;
 pub mod rpc;
 pub mod server;
 pub mod settings;
 
 use crate::{
     crypto::try_decrypt_message,
-    privmsg::Privmsg,
+    privmsg::{Privmsg, SeenMsgIds},
+    protocol_privmsg::ProtocolPrivmsg,
     rpc::JsonRpcInterface,
     server::IrcServerConnection,
     settings::{parse_configured_channels, Args, ChannelInfo, CONFIG_FILE, CONFIG_FILE_CONTENTS},
 };
 
-pub type SeenMsgIds = Arc<Mutex<Vec<u32>>>;
-
 fn build_irc_msg(msg: &Privmsg) -> String {
     debug!("ABOUT TO SEND: {:?}", msg);
     let irc_msg =
@@ -81,11 +81,11 @@ async fn broadcast_msg(
 }
 
 async fn process(
-    raft_receiver: Receiver<Privmsg>,
+    p2p_receiver: Receiver<Privmsg>,
     stream: TcpStream,
     peer_addr: SocketAddr,
-    raft_sender: Sender<Privmsg>,
-    seen_msg_id: SeenMsgIds,
+    p2p: net::P2pPtr,
+    seen_msg_ids: SeenMsgIds,
     autojoin_chans: Vec<String>,
     configured_chans: FxHashMap<String, ChannelInfo>,
 ) -> Result<()> {
@@ -94,8 +94,8 @@ async fn process(
     let mut reader = BufReader::new(reader);
     let mut conn = IrcServerConnection::new(
         writer,
-        seen_msg_id.clone(),
-        raft_sender,
+        seen_msg_ids.clone(),
+        p2p.clone(),
         autojoin_chans,
         configured_chans,
     );
@@ -103,17 +103,10 @@ async fn process(
     loop {
         let mut line = String::new();
         futures::select! {
-            privmsg = raft_receiver.recv().fuse() => {
+            privmsg = p2p_receiver.recv().fuse() => {
                 let mut msg = privmsg?;
                 info!("Received msg from Raft: {:?}", msg);
 
-                let mut smi = seen_msg_id.lock().await;
-                if smi.contains(&msg.id) {
-                    continue
-                }
-                smi.push(msg.id);
-                drop(smi);
-
                 // Try to potentially decrypt the incoming message.
                 if conn.configured_chans.contains_key(&msg.channel) {
                     let chan_info = conn.configured_chans.get(&msg.channel).unwrap();
@@ -156,47 +149,35 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         return Ok(())
     }
 
-    let seen_msg_id: SeenMsgIds = Arc::new(Mutex::new(vec![]));
-
     // Pick up channel settings from the TOML configuration
     let cfg_path = get_config_path(settings.config, CONFIG_FILE)?;
     let configured_chans = parse_configured_channels(&cfg_path)?;
 
     //
-    //Raft
+    // P2p setup
     //
-    let datastore_path = expand_path(&settings.datastore)?;
     let net_settings = settings.net;
-    let datastore_raft = datastore_path.join("ircd.db");
-    let mut raft = Raft::<Privmsg>::new(net_settings.inbound.clone(), datastore_raft)?;
-    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_send_channel, p2p_recv_channel) = async_channel::unbounded::<Privmsg>();
 
     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();
+    let seen_msg_ids = Arc::new(Mutex::new(vec![]));
+    let seen_msg_ids_cloned = seen_msg_ids.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
-            }
+            let seen_msg_ids_cloned = seen_msg_ids_cloned.clone();
+            async move { ProtocolPrivmsg::init(channel, sender, p2p, seen_msg_ids_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()));
+    executor_cloned.spawn(p2p.clone().run(executor.clone())).detach();
 
     //
     // RPC interface
@@ -204,8 +185,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     let rpc_listen_addr = settings.rpc_listen.clone();
     let rpc_interface =
         Arc::new(JsonRpcInterface { addr: rpc_listen_addr.clone(), p2p: p2p.clone() });
-    let rpc_task =
-        executor.spawn(async move { listen_and_serve(rpc_listen_addr, rpc_interface).await });
+    executor.spawn(async move { listen_and_serve(rpc_listen_addr, rpc_interface).await }).detach();
 
     //
     // IRC instance
@@ -214,33 +194,35 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     let listener = TcpListener::bind(irc_listen_addr).await?;
     let local_addr = listener.local_addr()?;
     info!("IRC listening on {}", local_addr);
+
     let executor_cloned = executor.clone();
-    let raft_receiver_cloned = raft_receiver.clone();
-    let irc_task: smol::Task<Result<()>> = executor.spawn(async move {
-        loop {
-            let (stream, peer_addr) = match listener.accept().await {
-                Ok((s, a)) => (s, a),
-                Err(e) => {
-                    error!("Failed listening for connections: {}", e);
-                    return Err(Error::ServiceStopped)
-                }
-            };
-
-            info!("IRC Accepted client: {}", peer_addr);
-
-            executor_cloned
-                .spawn(process(
-                    raft_receiver_cloned.clone(),
-                    stream,
-                    peer_addr,
-                    raft_sender.clone(),
-                    seen_msg_id.clone(),
-                    settings.autojoin.clone(),
-                    configured_chans.clone(),
-                ))
-                .detach();
-        }
-    });
+    executor
+        .spawn(async move {
+            loop {
+                let (stream, peer_addr) = match listener.accept().await {
+                    Ok((s, a)) => (s, a),
+                    Err(e) => {
+                        error!("Failed accepting new connections: {}", e);
+                        continue
+                    }
+                };
+
+                info!("IRC Accepted client: {}", peer_addr);
+
+                executor_cloned
+                    .spawn(process(
+                        p2p_recv_channel.clone(),
+                        stream,
+                        peer_addr,
+                        p2p.clone(),
+                        seen_msg_ids.clone(),
+                        settings.autojoin.clone(),
+                        configured_chans.clone(),
+                    ))
+                    .detach();
+            }
+        })
+        .detach();
 
     // Run once receive exit signal
     let (signal, shutdown) = async_channel::bounded::<()>(1);
@@ -248,14 +230,13 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         warn!(target: "ircd", "ircd start Exit Signal");
         // cleaning up tasks running in the background
         signal.send(()).await.unwrap();
-        rpc_task.cancel().await;
-        irc_task.cancel().await;
-        p2p_run_task.cancel().await;
     })
     .unwrap();
 
-    // blocking
-    raft.start(p2p.clone(), p2p_recv_channel.clone(), executor.clone(), shutdown.clone()).await?;
+    // Wait for SIGINT
+    shutdown.recv().await?;
+    print!("\r");
+    info!("Caught termination signal, cleaning up and exiting...");
 
     Ok(())
 }

bin/ircd/src/privmsg.rs

@@ -1,6 +1,10 @@
+use async_std::sync::{Arc, Mutex};
+
 use darkfi::util::serial::{SerialDecodable, SerialEncodable};
 
-pub type PrivmsgId = u32;
+pub type PrivmsgId = u64;
+
+pub type SeenMsgIds = Arc<Mutex<Vec<u64>>>;
 
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct Privmsg {

bin/ircd/src/server.rs

@@ -6,9 +6,13 @@ use fxhash::FxHashMap;
 use log::{debug, info, warn};
 use rand::{rngs::OsRng, RngCore};
 
-use darkfi::{Error, Result};
+use darkfi::{net::P2pPtr, Error, Result};
 
-use crate::{crypto::encrypt_message, privmsg::Privmsg, ChannelInfo, SeenMsgIds};
+use crate::{
+    crypto::encrypt_message,
+    privmsg::{Privmsg, SeenMsgIds},
+    ChannelInfo,
+};
 
 const RPL_NOTOPIC: u32 = 331;
 const RPL_TOPIC: u32 = 332;
@@ -19,8 +23,8 @@ pub struct IrcServerConnection {
     is_user_init: bool,
     is_registered: bool,
     nickname: String,
-    seen_msg_id: SeenMsgIds,
-    p2p_sender: async_channel::Sender<Privmsg>,
+    seen_msg_ids: SeenMsgIds,
+    p2p: P2pPtr,
     auto_channels: Vec<String>,
     pub configured_chans: FxHashMap<String, ChannelInfo>,
 }
@@ -28,8 +32,8 @@ pub struct IrcServerConnection {
 impl IrcServerConnection {
     pub fn new(
         write_stream: WriteHalf<TcpStream>,
-        seen_msg_id: SeenMsgIds,
-        p2p_sender: async_channel::Sender<Privmsg>,
+        seen_msg_ids: SeenMsgIds,
+        p2p: P2pPtr,
         auto_channels: Vec<String>,
         configured_chans: FxHashMap<String, ChannelInfo>,
     ) -> Self {
@@ -39,8 +43,8 @@ impl IrcServerConnection {
             is_user_init: false,
             is_registered: false,
             nickname: "anon".to_string(),
-            seen_msg_id,
-            p2p_sender,
+            seen_msg_ids,
+            p2p,
             auto_channels,
             configured_chans,
         }
@@ -153,7 +157,7 @@ impl IrcServerConnection {
                             message.to_string()
                         };
 
-                        let random_id = OsRng.next_u32();
+                        let random_id = OsRng.next_u64();
 
                         let protocol_msg = Privmsg {
                             id: random_id,
@@ -162,12 +166,12 @@ impl IrcServerConnection {
                             message,
                         };
 
-                        let mut smi = self.seen_msg_id.lock().await;
+                        let mut smi = self.seen_msg_ids.lock().await;
                         smi.push(random_id);
                         drop(smi);
 
                         debug!(target: "ircd", "PRIVMSG to be sent: {:?}", protocol_msg);
-                        self.p2p_sender.send(protocol_msg).await?;
+                        self.p2p.broadcast(protocol_msg).await?;
                     }
                 }
             }