Separated disseminnation layer: working, but without the reputation system (#1)

Author: xosmig

raikou/Cargo.toml

@@ -12,6 +12,9 @@ rust-version.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+aptos-consensus = { workspace = true }
+aptos-consensus-types = { workspace = true }
+aptos-crypto = { workspace = true }
 bitvec = { workspace = true }
 defaultmap = "0.5.0"  # Since version 0.6.0, the maps are !Send and !Sync
 env_logger = { workspace = true }

raikou/src/delays.rs

@@ -1,4 +1,4 @@
-use crate::{framework::NodeId, multichain};
+use crate::framework::NodeId;
 use rand::{rngs::SmallRng, thread_rng, Rng, SeedableRng};
 use rand_distr::Distribution;
 use std::{
@@ -8,24 +8,21 @@ use std::{
     sync::Arc,
 };
 
-pub trait DelayFunction<M>: Fn(NodeId, NodeId, &M) -> f64 + Clone + Send + Sync + 'static {}
+pub trait DelayFunction: Fn(NodeId, NodeId) -> f64 + Clone + Send + Sync + 'static {}
 
-impl<M, F> DelayFunction<M> for F where
-    F: Fn(NodeId, NodeId, &M) -> f64 + Clone + Send + Sync + 'static
-{
-}
+impl<F> DelayFunction for F where F: Fn(NodeId, NodeId) -> f64 + Clone + Send + Sync + 'static {}
 
-pub fn uniformly_random_delay<M>(
+pub fn uniformly_random_delay(
     distr: impl Distribution<f64> + Send + Sync + Copy + 'static,
-) -> impl DelayFunction<M> {
-    move |_from, _to, _msg| thread_rng().sample(distr)
+) -> impl DelayFunction {
+    move |_from, _to| thread_rng().sample(distr)
 }
 
-pub fn spacial_delay_2d<M>(
+pub fn spacial_delay_2d(
     max_distr: impl Distribution<f64> + Send + Sync + Copy + 'static,
-) -> impl DelayFunction<M> {
+) -> impl DelayFunction {
     let sqrt2 = f64::sqrt(2.);
-    move |from: NodeId, to, _msg| {
+    move |from: NodeId, to| {
         let from_coordinate = coordinate_2d_from_hash(from);
         let to_coordinate = coordinate_2d_from_hash(to);
         distance_2d(from_coordinate, to_coordinate) / sqrt2 * thread_rng().sample(max_distr)
@@ -35,12 +32,12 @@ pub fn spacial_delay_2d<M>(
 /// `base_distr` is sampled once per pair of nodes.
 /// `mul_noise_distr` and `add_noise_distr` are sampled for each message.
 /// The delay is computed as `base * mul_noise + add_noise`.
-pub fn heterogeneous_symmetric_delay<M>(
+pub fn heterogeneous_symmetric_delay(
     link_base_distr: impl Distribution<f64> + Send + Sync + Copy + 'static,
     mul_noise_distr: impl Distribution<f64> + Send + Sync + Copy + 'static,
     add_noise_distr: impl Distribution<f64> + Send + Sync + Copy + 'static,
-) -> impl DelayFunction<M> {
-    move |from: NodeId, to: NodeId, _msg| {
+) -> impl DelayFunction {
+    move |from: NodeId, to: NodeId| {
         let mut base_seed = [0; 16];
         base_seed[..8].copy_from_slice(&hash((min(from, to), max(from, to))).to_le_bytes());
         let mut base_rng = SmallRng::from_seed(base_seed);
@@ -52,11 +49,11 @@ pub fn heterogeneous_symmetric_delay<M>(
     }
 }
 
-pub fn clustered_delay<M>(
+pub fn clustered_delay(
     within_cluster_distr: impl Distribution<f64> + Send + Sync + Copy + 'static,
     between_cluster_distr: impl Distribution<f64> + Send + Sync + Copy + 'static,
     clusters: Vec<Vec<NodeId>>,
-) -> impl DelayFunction<M> {
+) -> impl DelayFunction {
     // Perform a sanity check that no node is missing or present in multiple clusters.
     let max_id = clusters.iter().flatten().max().unwrap();
     let n_nodes = clusters.iter().map(|cluster| cluster.len()).sum::<usize>();
@@ -72,7 +69,7 @@ pub fn clustered_delay<M>(
             .collect(),
     );
 
-    move |from: NodeId, to: NodeId, _msg| {
+    move |from: NodeId, to: NodeId| {
         let from_cluster = clusters.get(&from).unwrap();
         let to_cluster = clusters.get(&to).unwrap();
         if from_cluster == to_cluster {

raikou/src/framework/mod.rs

@@ -21,7 +21,7 @@ where
 }
 
 pub trait Protocol: Send + Sync {
-    type Message: Send + Sync;
+    type Message: Clone + Send + Sync;
     type TimerEvent: Send + Sync;
 
     fn start_handler<Ctx>(&mut self, ctx: &mut Ctx) -> impl Future<Output = ()> + Send
@@ -50,7 +50,7 @@ pub trait Protocol: Send + Sync {
         Ctx: ContextFor<Self>;
 
     fn run_ctx<Ctx>(
-        protocol: &tokio::sync::Mutex<Self>,
+        protocol: Arc<tokio::sync::Mutex<Self>>,
         ctx: &mut Ctx,
     ) -> impl Future<Output = ()> + Send
     where
@@ -81,7 +81,7 @@ pub trait Protocol: Send + Sync {
     }
 
     fn run<NS, TS>(
-        protocol: &tokio::sync::Mutex<Self>,
+        protocol: Arc<tokio::sync::Mutex<Self>>,
         node_id: NodeId,
         network_service: NS,
         timer: TS,

raikou/src/framework/timer.rs

@@ -44,6 +44,7 @@ impl<E: Send + Sync> TimerService for LocalTimerService<E> {
     }
 }
 
+#[derive(Clone)]
 pub struct InjectedTimerService<U, I> {
     underlying: U,
     injection: I,
@@ -53,6 +54,15 @@ pub trait TimerInjection<E>: Fn(Duration, E) -> (Duration, E) + Send + Sync + 's
 
 impl<I, E> TimerInjection<E> for I where I: Fn(Duration, E) -> (Duration, E) + Send + Sync + 'static {}
 
+pub fn clock_skew_injection<E>(clock_speed: f64) -> impl TimerInjection<E> {
+    move |duration, event| {
+        (
+            Duration::from_secs_f64(duration.as_secs_f64() / clock_speed),
+            event,
+        )
+    }
+}
+
 impl<U, I> InjectedTimerService<U, I>
 where
     U: TimerService,

raikou/src/main.rs

@@ -1,12 +1,13 @@
 use crate::{
-    delays::{heterogeneous_symmetric_delay, spacial_delay_2d, DelayFunction},
+    delays::{heterogeneous_symmetric_delay, DelayFunction},
     framework::{
         network::{InjectedLocalNetwork, Network, NetworkInjection},
-        timer::InjectedTimerService,
+        timer::{clock_skew_injection, InjectedTimerService},
         NodeId, Protocol,
     },
     leader_schedule::round_robin,
     multichain::{Config, MultiChainBft},
+    raikou::{dissemination, dissemination::fake::FakeDisseminationLayer},
 };
 use rand::{seq::SliceRandom, thread_rng, Rng};
 use std::{collections::BTreeMap, iter, sync::Arc, time::Duration};
@@ -29,7 +30,7 @@ const JOLTEON_TIMEOUT: u32 = 3; // in Deltas
 
 // TODO: generalize this to any protocol
 fn multichain_network_injection(
-    delay_function: impl DelayFunction<multichain::Message>,
+    delay_function: impl DelayFunction,
     crashes: Vec<(NodeId, Slot)>,
 ) -> impl NetworkInjection<multichain::Message> {
     let crashes = Arc::new(BTreeMap::from_iter(crashes));
@@ -52,29 +53,29 @@ fn multichain_network_injection(
                 }
             }
 
-            let delay = f64::max(delay_function(from, to, &message), 0.);
+            let delay = f64::max(delay_function(from, to), 0.);
             tokio::time::sleep(Duration::from_secs_f64(delay)).await;
             Some(message)
         }
     }
 }
 fn network_injection<M: Send>(
-    delay_function: impl DelayFunction<M>,
+    delay_function: impl DelayFunction,
     // crashes: Vec<(NodeId, Instant)>,
 ) -> impl NetworkInjection<M> {
     move |from, to, message| {
         let delay_function = delay_function.clone();
 
         async move {
-            let delay = f64::max(delay_function(from, to, &message), 0.);
+            let delay = f64::max(delay_function(from, to), 0.);
             tokio::time::sleep(Duration::from_secs_f64(delay)).await;
             Some(message)
         }
     }
 }
 
 async fn test_multichain(
-    delay_function: impl DelayFunction<multichain::Message>,
+    delay_function: impl DelayFunction,
     n_nodes: usize,
     slots_per_delta: Slot,
     delta: f64,
@@ -158,7 +159,7 @@ async fn test_multichain(
             // network_service.clear_inbox().await;
 
             // println!("Spawning node {node_id}");
-            let node = tokio::sync::Mutex::new(MultiChainBft::new_node(
+            let node = Arc::new(tokio::sync::Mutex::new(MultiChainBft::new_node(
                 node_id,
                 config,
                 start_time,
@@ -169,10 +170,10 @@ async fn test_multichain(
                     batch_commit_time: batch_commit_time_sender,
                     indirectly_committed_slots: indirectly_committed_slots_sender,
                 },
-            ));
+            )));
 
             semaphore.add_permits(1);
-            Protocol::run(&node, node_id, network_service, timer).await
+            Protocol::run(node, node_id, network_service, timer).await
         }));
     }
 
@@ -235,7 +236,7 @@ async fn test_multichain(
 }
 
 async fn test_multichain_with_random_crashes(
-    delay_function: impl DelayFunction<multichain::Message>,
+    delay_function: impl DelayFunction,
     n_nodes: usize,
     slots_per_delta: Slot,
     delta: f64,
@@ -268,7 +269,7 @@ async fn test_multichain_with_random_crashes(
 }
 
 async fn test_multichain_with_consecutive_faulty_leaders_in_a_chain(
-    delay_function: impl DelayFunction<multichain::Message>,
+    delay_function: impl DelayFunction,
     n_nodes: usize,
     slots_per_delta: Slot,
     delta: f64,
@@ -298,7 +299,7 @@ async fn test_multichain_with_consecutive_faulty_leaders_in_a_chain(
 }
 
 async fn test_jolteon(
-    delay_function: impl DelayFunction<jolteon::Message<()>>,
+    delay_function: impl DelayFunction,
     n_nodes: usize,
     delta: f64,
     warmup_duration_in_delta: u32,
@@ -377,16 +378,16 @@ async fn test_jolteon(
             let (_, txns_receiver) = mpsc::channel::<()>(100);
 
             // println!("Spawning node {node_id}");
-            let node = tokio::sync::Mutex::new(jolteon::JolteonNode::new(
+            let node = Arc::new(tokio::sync::Mutex::new(jolteon::JolteonNode::new(
                 node_id,
                 config,
                 txns_receiver,
                 start_time,
                 node_id == monitored_node,
-            ));
+            )));
 
             semaphore.add_permits(1);
-            Protocol::run(&node, node_id, network_service, timer).await
+            Protocol::run(node, node_id, network_service, timer).await
         }));
     }
 
@@ -449,7 +450,7 @@ async fn test_jolteon(
 }
 
 async fn test_jolteon_with_fast_qs(
-    delay_function: impl DelayFunction<jolteon_fast_qs::Message<()>>,
+    delay_function: impl DelayFunction,
     n_nodes: usize,
     delta: f64,
     warmup_duration_in_delta: u32,
@@ -532,7 +533,7 @@ async fn test_jolteon_with_fast_qs(
             let next_txn = || ();
 
             // println!("Spawning node {node_id}");
-            let node = tokio::sync::Mutex::new(jolteon_fast_qs::JolteonNode::new(
+            let node = Arc::new(tokio::sync::Mutex::new(jolteon_fast_qs::JolteonNode::new(
                 node_id,
                 config,
                 next_txn,
@@ -544,10 +545,10 @@ async fn test_jolteon_with_fast_qs(
                     batch_commit_time: batch_commit_time_sender,
                     // indirectly_committed_slots: indirectly_committed_slots_sender,
                 },
-            ));
+            )));
 
             semaphore.add_permits(1);
-            Protocol::run(&node, node_id, network_service, timer).await
+            Protocol::run(node, node_id, network_service, timer).await
         }));
     }
 
@@ -612,7 +613,7 @@ async fn test_jolteon_with_fast_qs(
 }
 
 async fn test_raikou(
-    delay_function: impl DelayFunction<raikou::Message>,
+    delay_function: impl DelayFunction + Clone,
     n_nodes: usize,
     delta: f64,
     spawn_period_in_delta: u32,
@@ -631,6 +632,8 @@ async fn test_raikou(
         rand_distr::Uniform::new(1. * delta, spawn_period_in_delta as f64 * delta);
     let clock_speed_distr = rand_distr::Normal::new(1., 0.01).unwrap();
 
+    let mut diss_network =
+        InjectedLocalNetwork::new(n_nodes, network_injection(delay_function.clone()));
     let mut network = InjectedLocalNetwork::new(n_nodes, network_injection(delay_function));
 
     let f = (n_nodes - 1) / 3;
@@ -663,15 +666,14 @@ async fn test_raikou(
     let start_time = Instant::now();
     for node_id in 0..n_nodes {
         let config = config.clone();
+        let diss_network_service = diss_network.service(node_id);
         let network_service = network.service(node_id);
 
         let clock_speed = { thread_rng().sample(clock_speed_distr) };
-        let timer = InjectedTimerService::local(move |duration, event| {
-            (
-                Duration::from_secs_f64(duration.as_secs_f64() / clock_speed),
-                event,
-            )
-        });
+
+        // introduce artificial clock skew.
+        let diss_timer = InjectedTimerService::local(clock_skew_injection(clock_speed));
+        let timer = InjectedTimerService::local(clock_skew_injection(clock_speed));
 
         // let propose_time_sender = Some(propose_time.new_sender());
         // let enter_time_sender = if node_id == monitored_node {
@@ -698,10 +700,23 @@ async fn test_raikou(
             // // Before starting the node, "drop" all messages sent to it during the spawn delay.
             // network_service.clear_inbox().await;
 
+            let txns_iter = iter::repeat_with(|| vec![]);
+
+            let dissemination = FakeDisseminationLayer::new(
+                node_id,
+                dissemination::fake::Config {
+                    n_nodes,
+                    ac_quorum: 2 * f + 1,
+                    batch_interval: Duration::from_secs_f64(delta * 0.1),
+                },
+                txns_iter,
+            );
+
             // println!("Spawning node {node_id}");
-            let node = tokio::sync::Mutex::new(raikou::RaikouNode::new(
+            let node = Arc::new(tokio::sync::Mutex::new(raikou::RaikouNode::new(
                 node_id,
                 config,
+                dissemination.clone(),
                 start_time,
                 node_id == monitored_node,
                 raikou::Metrics {
@@ -710,10 +725,17 @@ async fn test_raikou(
                     batch_commit_time: batch_commit_time_sender,
                     // indirectly_committed_slots: indirectly_committed_slots_sender,
                 },
+            )));
+
+            spawn(Protocol::run(
+                dissemination.protocol(),
+                node_id,
+                diss_network_service,
+                diss_timer,
             ));
+            spawn(Protocol::run(node, node_id, network_service, timer));
 
             semaphore.add_permits(1);
-            Protocol::run(&node, node_id, network_service, timer).await
         }));
     }