script/research/crdt: WIP implement P2p for crdt node

script/research/crdt/src/event.rs

@@ -2,7 +2,7 @@ use std::{cmp::Ordering, io};
 
 use darkfi::{
     net,
-    util::serial::{serialize, Decodable, Encodable},
+    util::serial::{Decodable, Encodable},
     Result,
 };
 
@@ -38,8 +38,8 @@ impl Decodable for Event {
 }
 
 impl Event {
-    pub fn new<T: Encodable + Decodable>(value: &T, counter: u64, name: String) -> Self {
-        Self { value: serialize(value), counter, name }
+    pub fn new(value: Vec<u8>, counter: u64, name: String) -> Self {
+        Self { value, counter, name }
     }
 }
 

script/research/crdt/src/lib.rs

@@ -5,65 +5,8 @@ pub mod node;
 
 pub use event::Event;
 pub use gset::GSet;
-pub use net::CrdtP2p;
+pub use net::ProtocolCrdt;
 pub use node::Node;
 
 #[cfg(test)]
-mod tests {
-
-    use super::*;
-
-    fn sync_simulation(mut a: Node, mut b: Node, mut c: Node) -> (Node, Node, Node) {
-        a.gset.merge(&b.gset);
-        a.gset.merge(&c.gset);
-
-        b.gset.merge(&a.gset);
-        b.gset.merge(&c.gset);
-
-        c.gset.merge(&a.gset);
-        c.gset.merge(&b.gset);
-
-        (a, b, c)
-    }
-
-    #[test]
-    fn test_crdt_gset() {
-        let mut a: Node = Node::new("Node A");
-        let mut b: Node = Node::new("Node B");
-        let mut c: Node = Node::new("Node C");
-
-        // node a
-        a.send_event("a_msg1".to_string());
-        a.send_event("a_msg2".to_string());
-
-        // node b
-        b.send_event("b_msg1".to_string());
-
-        // node c
-        c.send_event("c_msg1".to_string());
-
-        // node b
-        b.send_event("b_msg2".to_string());
-
-        let (a, mut b, mut c) = sync_simulation(a, b, c);
-
-        assert_eq!(a.gset.len(), 5);
-        assert_eq!(b.gset.len(), 5);
-        assert_eq!(c.gset.len(), 5);
-
-        // node c
-        c.send_event("c_msg2".to_string());
-        c.send_event("c_msg3".to_string());
-        c.send_event("c_msg4".to_string());
-        c.send_event("c_msg5".to_string());
-
-        // node b
-        b.send_event("b_msg3".to_string());
-
-        let (a, b, c) = sync_simulation(a, b, c);
-
-        assert_eq!(a.gset.len(), 10);
-        assert_eq!(b.gset.len(), 10);
-        assert_eq!(c.gset.len(), 10);
-    }
-}
+mod tests {}

script/research/crdt/src/main.rs

@@ -1,13 +1,64 @@
-use std::sync::Arc;
+use std::{env, sync::Arc};
 
 extern crate clap;
 use async_executor::Executor;
 use easy_parallel::Parallel;
 use simplelog::{ColorChoice, Config, LevelFilter, TermLogger, TerminalMode};
 
-use darkfi::Result;
+use darkfi::{net, Result};
 
-use crdt::{CrdtP2p, Event};
+use crdt::Node;
+
+async fn start(executor: Arc<Executor<'_>>) -> Result<()> {
+    //
+    // XXX THIS for testing purpose
+    //
+
+    let arg = env::args();
+    let arg = arg.last().unwrap();
+
+    // node 1
+    if arg == "1" {
+        let net_settings = net::Settings {
+            outbound_connections: 5,
+            seeds: vec!["127.0.0.1:9999".parse()?],
+            ..Default::default()
+        };
+
+        let node1 = Node::new("node1", net_settings).await;
+
+        executor.spawn(node1.clone().start(executor.clone())).detach();
+
+        darkfi::util::sleep(5).await;
+
+        node1.send_event(String::from("hello")).await?;
+
+        loop {}
+    }
+
+    // node 2
+    if arg == "2" {
+        let net_settings = net::Settings {
+            inbound: Some("127.0.0.1:6666".parse()?),
+            external_addr: Some("127.0.0.1:6666".parse()?),
+            seeds: vec!["127.0.0.1:9999".parse()?],
+            ..Default::default()
+        };
+        let node2 = Node::new("node2", net_settings).await;
+
+        node2.start(executor.clone()).await?;
+    }
+
+    // seed node
+    if arg == "3" {
+        let net_settings =
+            net::Settings { inbound: Some("127.0.0.1:9999".parse()?), ..Default::default() };
+
+        let node3 = Node::new("node3", net_settings).await;
+        node3.start(executor.clone()).await?;
+    }
+    Ok(())
+}
 
 fn main() -> Result<()> {
     let ex = Arc::new(Executor::new());
@@ -25,14 +76,12 @@ fn main() -> Result<()> {
     // let nthreads = num_cpus::get();
     // debug!(target: "IRC DAEMON", "Run {} executor threads", nthreads);
 
-    let (sender, _) = async_channel::unbounded::<Event>();
-
     let (_, result) = Parallel::new()
         .each(0..4, |_| smol::future::block_on(ex.run(shutdown.recv())))
         // Run the main future on the current thread.
         .finish(|| {
             smol::future::block_on(async move {
-                CrdtP2p::start(ex2.clone(), sender).await?;
+                start(ex2.clone()).await?;
                 drop(signal);
                 Ok::<(), darkfi::Error>(())
             })

script/research/crdt/src/net.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use async_std::sync::{Arc, Mutex};
 
 use async_executor::Executor;
 use async_trait::async_trait;
@@ -6,40 +6,14 @@ use log::debug;
 
 use darkfi::{net, Result};
 
-use crate::Event;
+use crate::{Event, GSet};
 
-pub struct CrdtP2p {}
-
-impl CrdtP2p {
-    pub async fn start(
-        executor: Arc<Executor<'_>>,
-        notify_queue_sender: async_channel::Sender<Event>,
-    ) -> Result<()> {
-        let p2p = net::P2p::new(net::Settings::default()).await;
-        let registry = p2p.protocol_registry();
-
-        registry
-            .register(!net::SESSION_SEED, move |channel, p2p| {
-                let sender = notify_queue_sender.clone();
-                async move { ProtocolCrdt::init(channel, sender, p2p).await }
-            })
-            .await;
-
-        //
-        // p2p network main instance
-        //
-        // Performs seed session
-        p2p.clone().start(executor.clone()).await?;
-        // Actual main p2p session
-        p2p.run(executor).await
-    }
-}
-
-struct ProtocolCrdt {
+pub struct ProtocolCrdt {
     jobsman: net::ProtocolJobsManagerPtr,
     notify_queue_sender: async_channel::Sender<Event>,
     event_sub: net::MessageSubscription<Event>,
     p2p: net::P2pPtr,
+    gset: Arc<Mutex<GSet<Event>>>,
 }
 
 impl ProtocolCrdt {
@@ -47,6 +21,7 @@ impl ProtocolCrdt {
         channel: net::ChannelPtr,
         notify_queue_sender: async_channel::Sender<Event>,
         p2p: net::P2pPtr,
+        gset: Arc<Mutex<GSet<Event>>>,
     ) -> net::ProtocolBasePtr {
         let message_subsytem = channel.get_message_subsystem();
         message_subsytem.add_dispatch::<Event>().await;
@@ -58,6 +33,7 @@ impl ProtocolCrdt {
             event_sub,
             jobsman: net::ProtocolJobsManager::new("ProtocolCrdt", channel),
             p2p,
+            gset,
         })
     }
 
@@ -72,6 +48,10 @@ impl ProtocolCrdt {
                 event
             );
 
+            if self.gset.lock().await.contains(&event) {
+                continue
+            }
+
             let event = (*event).clone();
             self.p2p.broadcast(event.clone()).await?;
 

script/research/crdt/src/node.rs

@@ -1,33 +1,87 @@
+use async_std::sync::{Arc, Mutex};
 use std::cmp::max;
 
-use darkfi::util::serial::{Decodable, Encodable};
+use async_executor::Executor;
 
-use crate::{Event, GSet};
+use darkfi::{
+    net,
+    util::serial::{serialize, Decodable, Encodable},
+    Result,
+};
+
+use crate::{Event, GSet, ProtocolCrdt};
 
-#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Node {
     // name to idnetifie the node
     name: String,
     // a grow-only set
-    pub(crate) gset: GSet<Event>,
+    gset: Arc<Mutex<GSet<Event>>>,
     // a counter for the node
-    time: u64,
+    time: Mutex<u64>,
+    p2p: net::P2pPtr,
 }
 
 impl Node {
-    pub fn new(name: &str) -> Self {
-        Self { name: name.into(), gset: GSet::new(), time: 0 }
+    pub async fn new(name: &str, net_settings: net::Settings) -> Arc<Self> {
+        let p2p = net::P2p::new(net_settings).await;
+        Arc::new(Self {
+            name: name.into(),
+            gset: Arc::new(Mutex::new(GSet::new())),
+            time: Mutex::new(0),
+            p2p,
+        })
+    }
+
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        let (snd, rcv) = async_channel::unbounded::<Event>();
+
+        let p2p = self.p2p.clone();
+
+        let registry = p2p.protocol_registry();
+
+        let gset = self.gset.clone();
+
+        registry
+            .register(!net::SESSION_SEED, move |channel, p2p| {
+                let sender = snd.clone();
+                let gset = gset.clone();
+                async move { ProtocolCrdt::init(channel, sender, p2p, gset).await }
+            })
+            .await;
+
+        //
+        // p2p network main instance
+        //
+        // Performs seed session
+        p2p.clone().start(executor.clone()).await?;
+        // Actual main p2p session
+
+        let recv_task = executor.spawn(async move {
+            loop {
+                // XXX remove unwrap
+                let event = rcv.recv().await.unwrap();
+                self.clone().receive_event(&event).await;
+            }
+        });
+
+        p2p.clone().run(executor.clone()).await?;
+
+        recv_task.cancel().await;
+
+        Ok(())
     }
 
-    pub fn receive_event(&mut self, event: &Event) {
-        self.time = max(self.time, event.counter) + 1;
-        self.gset.insert(event);
+    pub async fn receive_event(self: Arc<Self>, event: &Event) {
+        let mut time = self.time.lock().await;
+        *time = max(*time, event.counter) + 1;
+        self.gset.lock().await.insert(event);
     }
 
-    pub fn send_event<T: Decodable + Encodable>(&mut self, value: T) -> Event {
-        self.time += 1;
-        let event = Event::new(&value, self.time, self.name.clone());
-        self.gset.insert(&event);
-        event
+    pub async fn send_event<T: Decodable + Encodable>(self: Arc<Self>, value: T) -> Result<()> {
+        let mut time = self.time.lock().await;
+        *time += 1;
+        let event = Event::new(serialize(&value), *time, self.name.clone());
+        self.gset.lock().await.insert(&event);
+        self.p2p.broadcast(event).await
     }
 }