Use par_iter in network module (#116)

* Use par_iter in network module

* Pass rng from filter function

* Remove locking for network messages

simulations/src/network/mod.rs

@@ -6,14 +6,15 @@ use std::{
 };
 // crates
 use crossbeam::channel::{self, Receiver, Sender};
-use rand::Rng;
+use rand::{rngs::ThreadRng, Rng};
+use rayon::prelude::*;
 // internal
 use crate::node::NodeId;
 
 pub mod behaviour;
 pub mod regions;
 
-pub type NetworkTime = Instant;
+type NetworkTime = Instant;
 
 pub struct Network<M> {
     pub regions: regions::RegionsData,
@@ -25,7 +26,7 @@ pub struct Network<M> {
 
 impl<M> Network<M>
 where
-    M: Clone,
+    M: Send + Sync + Clone,
 {
     pub fn new(regions: regions::RegionsData) -> Self {
         Self {
@@ -62,39 +63,64 @@ where
     }
 
     /// Collects and dispatches messages to connected interfaces.
-    pub fn step<R: Rng>(&mut self, rng: &mut R, time_passed: Duration) {
+    pub fn step(&mut self, time_passed: Duration) {
         self.collect_messages();
-        self.dispatch_after(rng, time_passed);
+        self.dispatch_after(time_passed);
     }
 
     /// Receive and store all messages from nodes.
     pub fn collect_messages(&mut self) {
-        self.from_node_receivers.iter().for_each(|(_, from_node)| {
-            while let Ok(message) = from_node.try_recv() {
-                self.messages.push((self.network_time, message));
-            }
-        });
+        let mut new_messages = self
+            .from_node_receivers
+            .par_iter()
+            .flat_map(|(_, from_node)| {
+                from_node
+                    .try_iter()
+                    .map(|msg| (self.network_time, msg))
+                    .collect::<Vec<_>>()
+            })
+            .collect();
+
+        self.messages.append(&mut new_messages);
     }
 
     /// Reiterate all messages and send to appropriate nodes if simulated
     /// delay has passed.
-    pub fn dispatch_after<R: Rng>(&mut self, rng: &mut R, time_passed: Duration) {
+    pub fn dispatch_after(&mut self, time_passed: Duration) {
         self.network_time += time_passed;
 
-        let mut delayed = vec![];
-        while let Some((network_time, message)) = self.messages.pop() {
-            // If cost is None, message won't get sent nor it will be
-            // readded to the pending messages list.
+        let delayed = self
+            .messages
+            .par_iter()
+            .filter(|(network_time, message)| {
+                let mut rng = ThreadRng::default();
+                self.send_or_drop_message(&mut rng, network_time, message)
+            })
+            .cloned()
+            .collect();
+
+        self.messages = delayed;
+    }
+
+    /// Returns true if message needs to be delayed and be dispatched in future.
+    fn send_or_drop_message<R: Rng>(
+        &self,
+        rng: &mut R,
+        network_time: &NetworkTime,
+        message: &NetworkMessage<M>,
+    ) -> bool {
         if let Some(delay) = self.send_message_cost(rng, message.from, message.to) {
             if network_time.add(delay) <= self.network_time {
                 let to_node = self.to_node_senders.get(&message.to).unwrap();
-                    to_node.send(message).expect("Node should have connection");
+                to_node
+                    .send(message.clone())
+                    .expect("Node should have connection");
+                return false;
             } else {
-                    delayed.push((network_time, message));
+                return true;
             }
         }
-        }
-        self.messages = delayed;
+        false
     }
 }
 
@@ -127,7 +153,6 @@ mod tests {
     };
     use crate::node::NodeId;
     use crossbeam::channel::{self, Receiver, Sender};
-    use rand::rngs::mock::StepRng;
     use std::{collections::HashMap, time::Duration};
 
     struct MockNetworkInterface {
@@ -165,7 +190,6 @@ mod tests {
 
     #[test]
     fn send_receive_messages() {
-        let mut rng = StepRng::new(1, 0);
         let node_a = 0.into();
         let node_b = 1.into();
 
@@ -191,15 +215,15 @@ mod tests {
         assert_eq!(a.receive_messages().len(), 0);
         assert_eq!(b.receive_messages().len(), 0);
 
-        network.step(&mut rng, Duration::from_millis(0));
+        network.step(Duration::from_millis(0));
         assert_eq!(a.receive_messages().len(), 0);
         assert_eq!(b.receive_messages().len(), 0);
 
-        network.step(&mut rng, Duration::from_millis(100));
+        network.step(Duration::from_millis(100));
         assert_eq!(a.receive_messages().len(), 0);
         assert_eq!(b.receive_messages().len(), 1);
 
-        network.step(&mut rng, Duration::from_millis(100));
+        network.step(Duration::from_millis(100));
         assert_eq!(a.receive_messages().len(), 0);
         assert_eq!(b.receive_messages().len(), 0);
 
@@ -208,14 +232,13 @@ mod tests {
         b.send_message(node_a, ());
         network.collect_messages();
 
-        network.dispatch_after(&mut rng, Duration::from_millis(100));
+        network.dispatch_after(Duration::from_millis(100));
         assert_eq!(a.receive_messages().len(), 3);
         assert_eq!(b.receive_messages().len(), 0);
     }
 
     #[test]
     fn regions_send_receive_messages() {
-        let mut rng = StepRng::new(1, 0);
         let node_a = 0.into();
         let node_b = 1.into();
         let node_c = 2.into();
@@ -265,7 +288,7 @@ mod tests {
         c.send_message(node_b, ());
         network.collect_messages();
 
-        network.dispatch_after(&mut rng, Duration::from_millis(100));
+        network.dispatch_after(Duration::from_millis(100));
         assert_eq!(a.receive_messages().len(), 1);
         assert_eq!(b.receive_messages().len(), 1);
         assert_eq!(c.receive_messages().len(), 0);
@@ -274,12 +297,12 @@ mod tests {
         b.send_message(node_c, ());
         network.collect_messages();
 
-        network.dispatch_after(&mut rng, Duration::from_millis(400));
+        network.dispatch_after(Duration::from_millis(400));
         assert_eq!(a.receive_messages().len(), 1); // c to a
         assert_eq!(b.receive_messages().len(), 2); // c to b && a to b
         assert_eq!(c.receive_messages().len(), 2); // a to c && b to c
 
-        network.dispatch_after(&mut rng, Duration::from_millis(100));
+        network.dispatch_after(Duration::from_millis(100));
         assert_eq!(a.receive_messages().len(), 0);
         assert_eq!(b.receive_messages().len(), 0);
         assert_eq!(c.receive_messages().len(), 1); // b to c

simulations/src/node/dummy.rs

@@ -315,9 +315,6 @@ impl DummyNode {
 
     fn handle_leaf(&mut self, message: &NetworkMessage<DummyMessage>) {
         if let DummyMessage::Proposal(block) = &message.payload {
-            if block.view > self.current_view() {
-                self.update_view(block.view);
-            }
             if !self.is_vote_sent(block.view) {
                 let parents = &self.local_view.parents.as_ref().expect("leaf has parents");
                 parents.iter().for_each(|node_id| {
@@ -601,7 +598,7 @@ mod tests {
         }
 
         // 1. Leaders receive vote and broadcast new Proposal(Block) to all nodes.
-        network.dispatch_after(&mut rng, Duration::from_millis(100));
+        network.dispatch_after(Duration::from_millis(100));
         nodes.iter_mut().for_each(|(_, node)| {
             node.step();
         });
@@ -609,7 +606,7 @@ mod tests {
 
         // 2. a) All nodes received proposal block.
         //    b) Leaf nodes send vote to internal nodes.
-        network.dispatch_after(&mut rng, Duration::from_millis(100));
+        network.dispatch_after(Duration::from_millis(100));
         nodes.iter_mut().for_each(|(_, node)| {
             node.step();
         });
@@ -632,7 +629,7 @@ mod tests {
         assert!(nodes[&6.into()].state().view_state[&1].vote_sent); // Leaf
 
         // 3. Internal nodes send vote to root node.
-        network.dispatch_after(&mut rng, Duration::from_millis(100));
+        network.dispatch_after(Duration::from_millis(100));
         nodes.iter_mut().for_each(|(_, node)| {
             node.step();
         });
@@ -650,7 +647,7 @@ mod tests {
         assert!(nodes[&6.into()].state().view_state[&1].vote_sent); // Leaf
 
         // 4. Root node send vote to next view leader nodes.
-        network.dispatch_after(&mut rng, Duration::from_millis(100));
+        network.dispatch_after(Duration::from_millis(100));
         nodes.iter_mut().for_each(|(_, node)| {
             node.step();
         });
@@ -666,7 +663,7 @@ mod tests {
         assert!(nodes[&6.into()].state().view_state[&1].vote_sent); // Leaf
 
         // 5. Leaders receive vote and broadcast new Proposal(Block) to all nodes.
-        network.dispatch_after(&mut rng, Duration::from_millis(100));
+        network.dispatch_after(Duration::from_millis(100));
         nodes.iter_mut().for_each(|(_, node)| {
             node.step();
         });
@@ -679,7 +676,7 @@ mod tests {
 
         // 6. a) All nodes received proposal block.
         //    b) Leaf nodes send vote to internal nodes.
-        network.dispatch_after(&mut rng, Duration::from_millis(100));
+        network.dispatch_after(Duration::from_millis(100));
         nodes.iter_mut().for_each(|(_, node)| {
             node.step();
         });
@@ -740,7 +737,7 @@ mod tests {
         }
 
         for _ in 0..7 {
-            network.dispatch_after(&mut rng, Duration::from_millis(100));
+            network.dispatch_after(Duration::from_millis(100));
             nodes.iter_mut().for_each(|(_, node)| {
                 node.step();
             });
@@ -789,7 +786,7 @@ mod tests {
         }
 
         for _ in 0..7 {
-            network.dispatch_after(&mut rng, Duration::from_millis(100));
+            network.dispatch_after(Duration::from_millis(100));
             nodes.iter_mut().for_each(|(_, node)| {
                 node.step();
             });
@@ -835,7 +832,7 @@ mod tests {
 
         let nodes = Arc::new(RwLock::new(nodes));
         for _ in 0..9 {
-            network.dispatch_after(&mut rng, Duration::from_millis(100));
+            network.dispatch_after(Duration::from_millis(100));
             nodes.write().unwrap().par_iter_mut().for_each(|(_, node)| {
                 node.step();
             });

simulations/src/runner/async_runner.rs

@@ -15,7 +15,7 @@ pub fn simulate<M, N: Node, O: Overlay>(
     mut out_data: Option<&mut Vec<OutData>>,
 ) -> anyhow::Result<()>
 where
-    M: Clone,
+    M: Send + Sync + Clone,
     N::Settings: Clone,
     N: Send + Sync,
     N::State: Serialize,

simulations/src/runner/glauber_runner.rs

@@ -16,7 +16,7 @@ pub fn simulate<M, N: Node, O: Overlay>(
     mut out_data: Option<&mut Vec<OutData>>,
 ) -> anyhow::Result<()>
 where
-    M: Clone,
+    M: Send + Sync + Clone,
     N: Send + Sync,
     N::Settings: Clone,
     N::State: Serialize,

simulations/src/runner/layered_runner.rs

@@ -50,7 +50,7 @@ pub fn simulate<M, N: Node, O: Overlay>(
     mut out_data: Option<&mut Vec<OutData>>,
 ) -> anyhow::Result<()>
 where
-    M: Clone,
+    M: Send + Sync + Clone,
     N: Send + Sync,
     N::Settings: Clone,
     N::State: Serialize,

simulations/src/runner/mod.rs

@@ -37,7 +37,7 @@ where
 
 impl<M, N: Node, O: Overlay> SimulationRunner<M, N, O>
 where
-    M: Clone,
+    M: Send + Sync + Clone,
     N: Send + Sync,
     N::Settings: Clone,
     N::State: Serialize,
@@ -101,8 +101,7 @@ where
     }
 
     fn step(&mut self) {
-        self.network
-            .dispatch_after(&mut self.rng, Duration::from_millis(100));
+        self.network.dispatch_after(Duration::from_millis(100));
         self.nodes
             .write()
             .expect("Single access to nodes vector")

simulations/src/runner/sync_runner.rs

@@ -13,7 +13,7 @@ pub fn simulate<M, N: Node, O: Overlay>(
     mut out_data: Option<&mut Vec<OutData>>,
 ) -> anyhow::Result<()>
 where
-    M: Clone,
+    M: Send + Sync + Clone,
     N: Send + Sync,
     N::Settings: Clone,
     N::State: Serialize,