Merge pull request #341 from pragma-org/stevan/concurrent-requests

Stevan/concurrent requests

Author: stevana

simulation/amaru-sim/src/simulator/generate.rs

@@ -12,21 +12,23 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use proptest::prelude::*;
 use pure_stage::Instant;
 use rand::rngs::StdRng;
 use rand::Rng;
-use rand::SeedableRng;
 use rand_distr::{Distribution, Exp};
 use serde::Deserialize;
 use serde_json::Result;
+use std::cmp::Reverse;
 use std::collections::{BTreeMap, BTreeSet};
 use std::fmt;
 use std::fs;
 use std::path::PathBuf;
 use std::time::Duration;
 
+use crate::echo::Envelope;
+
 use super::bytes::Bytes;
+use super::simulate::Entry;
 use super::sync::ChainSyncMessage;
 use slot_arithmetic::Slot;
 
@@ -298,18 +300,6 @@ pub fn generate_inputs<R: Rng>(rng: &mut R, file_path: &PathBuf) -> Result<Vec<C
     }
 }
 
-pub fn generate_inputs_strategy(
-    file_path: &PathBuf,
-    seed: Option<u64>,
-) -> impl Strategy<Value = Vec<ChainSyncMessage>> + use<'_> {
-    any::<u64>().prop_map(move |s| {
-        let seed = seed.unwrap_or(s);
-        let mut rng = StdRng::seed_from_u64(seed);
-        println!("seed {}", seed);
-        generate_inputs(&mut rng, file_path).unwrap()
-    })
-}
-
 pub fn generate_arrival_times<R: Rng>(
     rng: &mut R,
     start_time: Instant,
@@ -329,12 +319,89 @@ pub fn generate_arrival_times<R: Rng>(
     arrival_times
 }
 
+pub fn generate_entries<R: Rng>(
+    file_path: &PathBuf,
+    start_time: Instant,
+    mean_millis: f64,
+    number_of_upstream_peers: u8,
+) -> impl Fn(&mut R) -> Vec<Reverse<Entry<ChainSyncMessage>>> + use<'_, R> {
+    move |rng| {
+        let mut entries: Vec<Reverse<Entry<ChainSyncMessage>>> = vec![];
+        for client in 1..=number_of_upstream_peers {
+            let messages =
+                generate_inputs(rng, file_path).expect("Failed to generate inputs from chain file");
+            let arrival_times =
+                generate_arrival_times(rng, start_time, mean_millis, messages.len());
+            entries.extend(
+                messages
+                    .into_iter()
+                    .enumerate()
+                    .map(|(idx, msg)| {
+                        Reverse(Entry {
+                            arrival_time: arrival_times[idx],
+                            envelope: Envelope {
+                                src: "c".to_owned() + &client.to_string(),
+                                dest: "n1".to_string(),
+                                body: msg,
+                            },
+                        })
+                    })
+                    .collect::<Vec<_>>(),
+            );
+        }
+        entries
+    }
+}
+
+pub fn generate_vec<A>(
+    size: usize,
+    generator: impl Fn(&mut StdRng) -> A,
+) -> impl Fn(&mut StdRng) -> Vec<A> {
+    move |rng| {
+        let mut result = Vec::<A>::with_capacity(size);
+        for _ in 0..size {
+            result.push(generator(rng));
+        }
+        result
+    }
+}
+
+pub fn generate_u8_then<A>(low: u8, high: u8, then: impl Fn(u8) -> A) -> impl Fn(&mut StdRng) -> A {
+    move |rng| {
+        let x = rng.random_range(low..=high);
+        then(x)
+    }
+}
+
+pub fn generate_u8(low: u8, high: u8) -> impl Fn(&mut StdRng) -> u8 {
+    generate_u8_then(low, high, |x| x)
+}
+
 #[cfg(test)]
 mod test {
+    use rand::rngs::StdRng;
+    use rand::SeedableRng;
     use std::path::Path;
 
     use crate::simulator::generate::*;
 
+    #[test]
+    fn test_generate_u8() {
+        let seed = 1234;
+        let mut rng = StdRng::seed_from_u64(seed);
+        let mut counts = std::collections::BTreeMap::new();
+
+        for _i in 0..1000 {
+            let x = generate_u8(1, 6)(&mut rng);
+            *counts.entry(x).or_insert(0) += 1;
+        }
+
+        assert_eq!(counts.len(), 6);
+        for i in 1..=6 {
+            assert!(counts.contains_key(&i), "value {} was never generated", i);
+        }
+    }
+
     #[test]
     fn test_ancestors() {
         let a = Bytes::try_from("aa").unwrap();

simulation/amaru-sim/src/simulator/mod.rs

@@ -37,12 +37,13 @@ use amaru_stores::rocksdb::consensus::InMemConsensusStore;
 use anyhow::Error;
 use bytes::Bytes;
 use clap::Parser;
-use generate::{generate_inputs_strategy, parse_json, read_chain_json};
+use generate::{generate_entries, parse_json, read_chain_json};
 use ledger::{populate_chain_store, FakeStakeDistribution};
-use proptest::test_runner::Config;
 use pure_stage::{simulation::SimulationBuilder, trace_buffer::TraceBuffer, StageRef};
-use pure_stage::{Receiver, StageGraph, Void};
-use simulate::{pure_stage_node_handle, simulate, Trace};
+use pure_stage::{Instant, Receiver, StageGraph, Void};
+use rand::Rng;
+use simulate::{pure_stage_node_handle, simulate, History, SimulateConfig};
+use std::time::Duration;
 use std::{path::PathBuf, sync::Arc};
 use tokio::sync::Mutex;
 use tracing::{info, Span};
@@ -81,6 +82,18 @@ pub struct Args {
     #[arg(long, default_value_t = Hash::from([0; 32]))]
     pub start_header: Hash<32>,
 
+    /// Number of tests to run in simulation
+    #[arg(long, default_value = "50")]
+    pub number_of_tests: Option<u32>,
+
+    /// Number of nodes in simulation.
+    #[arg(long, default_value = "1")]
+    pub number_of_nodes: Option<u8>,
+
+    /// Number of upstream peers to simulate
+    #[arg(long, default_value = "2")]
+    pub number_of_upstream_peers: Option<u8>,
+
     /// Seed for simulation testing.
     #[arg(long)]
     pub seed: Option<u64>,
@@ -108,8 +121,9 @@ fn init_node(args: &Args) -> (GlobalParameters, SelectChain, ValidateHeader) {
     let select_chain = SelectChain::new(make_chain_selector(
         Origin,
         &chain_store,
-        // FIXME: Shouldn't be hardcoded!
-        &vec![Peer::new("c1")],
+        &(1..=args.number_of_upstream_peers.unwrap_or(2))
+            .map(|i| Peer::new(&format!("c{}", i)))
+            .collect::<Vec<_>>(),
     ));
     let chain_ref = Arc::new(Mutex::new(chain_store));
     let validate_header = ValidateHeader::new(Arc::new(stake_distribution), chain_ref.clone());
@@ -334,7 +348,9 @@ fn spawn_node(
 }
 
 pub fn run(rt: tokio::runtime::Runtime, args: Args) {
-    let number_of_nodes = 1;
+    let number_of_tests = args.number_of_tests.unwrap_or(50);
+    let number_of_nodes = args.number_of_nodes.unwrap_or(1);
+    let number_of_upstream_peers = args.number_of_upstream_peers.unwrap_or(2);
     let trace_buffer = Arc::new(parking_lot::Mutex::new(TraceBuffer::new(42, 1_000_000_000)));
 
     let spawn = || {
@@ -344,11 +360,24 @@ pub fn run(rt: tokio::runtime::Runtime, args: Args) {
         pure_stage_node_handle(rx, receive, running).unwrap()
     };
 
+    let seed = args.seed.unwrap_or({
+        let mut rng = rand::rng();
+        rng.random::<u64>()
+    });
+
     simulate(
-        Config::default(),
-        number_of_nodes,
+        SimulateConfig {
+            number_of_tests,
+            seed,
+            number_of_nodes,
+        },
         spawn,
-        generate_inputs_strategy(&args.block_tree_file, args.seed),
+        generate_entries(
+            &args.block_tree_file,
+            Instant::at_offset(Duration::from_secs(0)),
+            200.0,
+            number_of_upstream_peers,
+        ),
         chain_property(&args.block_tree_file),
         trace_buffer.clone(),
         args.persist_on_success,
@@ -357,15 +386,13 @@ pub fn run(rt: tokio::runtime::Runtime, args: Args) {
 
 fn chain_property(
     chain_data_path: &PathBuf,
-) -> impl Fn(Trace<ChainSyncMessage>) -> Result<(), String> + use<'_> {
-    move |trace| {
-        match trace.0.last() {
-            None => Err("impossible, no last entry in trace".to_string()),
+) -> impl Fn(History<ChainSyncMessage>) -> Result<(), String> + use<'_> {
+    move |history| {
+        match history.0.last() {
+            None => Err("impossible, no last entry in history".to_string()),
             Some(entry) => {
-                assert_eq!(entry.src, "n1", "entry: {:?}, trace: {:?}", entry, trace);
-                assert_eq!(entry.dest, "c1");
                 // FIXME: the property is wrong, we should check the property
-                // that the output message trace is a prefix of the read chain
+                // that the output message history is a prefix of the read chain
                 let data = read_chain_json(chain_data_path);
                 let blocks = parse_json(data.as_bytes()).map_err(|err| err.to_string())?;
                 match &entry.body {
@@ -376,20 +403,13 @@ fn chain_property(
                             .map(|block| (block.hash.clone(), Slot::from(block.slot)))
                             .expect("empty chain data");
                         if actual != expected {
-                            panic!(
+                            return Err(format!(
                                 "tip of chains don't match, expected {:?}, got {:?}",
                                 expected, actual
-                            );
-                        }
-                        info!("Success!")
-                    }
-                    _ => {
-                        info!("TRACE:");
-                        for entry in &trace.0 {
-                            info!("{:?}", entry);
+                            ));
                         }
-                        panic!("Last entry in trace isn't a forward")
                     }
+                    _ => return Err("Last entry in history isn't a forward".to_string()),
                 }
                 Ok(())
             }