script/research/crdt: impl Decodable + Encodable for Event

script/research/crdt/src/event.rs

@@ -1,11 +1,15 @@
-use std::cmp::Ordering;
+use std::{cmp::Ordering, io};
 
-use serde::{Deserialize, Serialize};
+use darkfi::{
+    net,
+    util::serial::{serialize, Decodable, Encodable},
+    Result,
+};
 
-#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, PartialOrd)]
-pub struct Event<T: Clone> {
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd)]
+pub struct Event {
     // the msg in the event
-    pub value: T,
+    pub value: Vec<u8>,
     // the counter for lamport clock
     pub counter: u64,
     // It might be necessary to attach the node's name to the timestamp
@@ -13,13 +17,33 @@ pub struct Event<T: Clone> {
     pub name: String,
 }
 
-impl<T: Clone> Event<T> {
-    pub fn new(value: &T, counter: u64, name: String) -> Self {
-        Self { value: value.clone(), counter, name }
+impl Encodable for Event {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.value.encode(&mut s)?;
+        len += self.counter.encode(&mut s)?;
+        len += self.name.encode(&mut s)?;
+        Ok(len)
     }
 }
 
-impl<T: Eq + PartialOrd + Clone> Ord for Event<T> {
+impl Decodable for Event {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self {
+            value: Decodable::decode(&mut d)?,
+            counter: Decodable::decode(&mut d)?,
+            name: Decodable::decode(&mut d)?,
+        })
+    }
+}
+
+impl Event {
+    pub fn new<T: Encodable + Decodable>(value: &T, counter: u64, name: String) -> Self {
+        Self { value: serialize(value), counter, name }
+    }
+}
+
+impl Ord for Event {
     fn cmp(&self, other: &Self) -> Ordering {
         let ord = self.counter.cmp(&other.counter);
         if ord == Ordering::Equal {
@@ -28,3 +52,9 @@ impl<T: Eq + PartialOrd + Clone> Ord for Event<T> {
         ord
     }
 }
+
+impl net::Message for Event {
+    fn name() -> &'static str {
+        "event"
+    }
+}

script/research/crdt/src/gset.rs

@@ -1,10 +1,6 @@
 use std::collections::BTreeSet;
 
-use serde::{Deserialize, Serialize};
-
-// CRDT using gset and lamport clock
-
-#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct GSet<T: Ord> {
     set: BTreeSet<T>,
 }

script/research/crdt/src/lib.rs

@@ -13,11 +13,7 @@ mod tests {
 
     use super::*;
 
-    fn sync_simulation(
-        mut a: Node<String>,
-        mut b: Node<String>,
-        mut c: Node<String>,
-    ) -> (Node<String>, Node<String>, Node<String>) {
+    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);
 
@@ -32,22 +28,22 @@ mod tests {
 
     #[test]
     fn test_crdt_gset() {
-        let mut a: Node<String> = Node::new("Node A");
-        let mut b: Node<String> = Node::new("Node B");
-        let mut c: Node<String> = Node::new("Node C");
+        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());
+        a.send_event("a_msg1".to_string());
+        a.send_event("a_msg2".to_string());
 
         // node b
-        b.send_event(&"b_msg1".to_string());
+        b.send_event("b_msg1".to_string());
 
         // node c
-        c.send_event(&"c_msg1".to_string());
+        c.send_event("c_msg1".to_string());
 
         // node b
-        b.send_event(&"b_msg2".to_string());
+        b.send_event("b_msg2".to_string());
 
         let (a, mut b, mut c) = sync_simulation(a, b, c);
 
@@ -56,13 +52,13 @@ mod tests {
         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());
+        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());
+        b.send_event("b_msg3".to_string());
 
         let (a, b, c) = sync_simulation(a, b, c);
 

script/research/crdt/src/main.rs

@@ -7,7 +7,7 @@ use simplelog::{ColorChoice, Config, LevelFilter, TermLogger, TerminalMode};
 
 use darkfi::Result;
 
-use crdt::CrdtP2p;
+use crdt::{CrdtP2p, Event};
 
 fn main() -> Result<()> {
     let ex = Arc::new(Executor::new());
@@ -25,7 +25,7 @@ fn main() -> Result<()> {
     // let nthreads = num_cpus::get();
     // debug!(target: "IRC DAEMON", "Run {} executor threads", nthreads);
 
-    let (sender, _) = async_channel::unbounded::<crdt::net::Event>();
+    let (sender, _) = async_channel::unbounded::<Event>();
 
     let (_, result) = Parallel::new()
         .each(0..4, |_| smol::future::block_on(ex.run(shutdown.recv())))

script/research/crdt/src/net.rs

@@ -1,13 +1,12 @@
+use std::sync::Arc;
+
 use async_executor::Executor;
 use async_trait::async_trait;
-
-use darkfi::{
-    net,
-    util::serial::{Decodable, Encodable},
-    Result,
-};
 use log::debug;
-use std::{io, sync::Arc};
+
+use darkfi::{net, Result};
+
+use crate::Event;
 
 pub struct CrdtP2p {}
 
@@ -36,28 +35,6 @@ impl CrdtP2p {
     }
 }
 
-#[derive(Debug, Clone)]
-pub struct Event {}
-
-impl net::Message for Event {
-    fn name() -> &'static str {
-        "event"
-    }
-}
-
-impl Encodable for Event {
-    fn encode<S: io::Write>(&self, mut _s: S) -> Result<usize> {
-        let len = 0;
-        Ok(len)
-    }
-}
-
-impl Decodable for Event {
-    fn decode<D: io::Read>(mut _d: D) -> Result<Self> {
-        Ok(Self {})
-    }
-}
-
 struct ProtocolCrdt {
     jobsman: net::ProtocolJobsManagerPtr,
     notify_queue_sender: async_channel::Sender<Event>,

script/research/crdt/src/node.rs

@@ -1,32 +1,32 @@
 use std::cmp::max;
 
-use serde::{Deserialize, Serialize};
+use darkfi::util::serial::{Decodable, Encodable};
 
 use crate::{Event, GSet};
 
-#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
-pub struct Node<T: Ord + Clone> {
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Node {
     // name to idnetifie the node
     name: String,
     // a grow-only set
-    pub(crate) gset: GSet<Event<T>>,
+    pub(crate) gset: GSet<Event>,
     // a counter for the node
     time: u64,
 }
 
-impl<T: Ord + Clone> Node<T> {
+impl Node {
     pub fn new(name: &str) -> Self {
         Self { name: name.into(), gset: GSet::new(), time: 0 }
     }
 
-    pub fn receive_event(&mut self, event: &Event<T>) {
+    pub fn receive_event(&mut self, event: &Event) {
         self.time = max(self.time, event.counter) + 1;
         self.gset.insert(event);
     }
 
-    pub fn send_event(&mut self, value: &T) -> Event<T> {
+    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());
+        let event = Event::new(&value, self.time, self.name.clone());
         self.gset.insert(&event);
         event
     }