Merge pull request #309 from carlaKC/des-tokio

Use time-paused tokio for zero sleeps

Author: carlaKC

sim-cli/Cargo.toml

@@ -18,7 +18,8 @@ triggered = "0.1.2"
 serde = "1.0.183"
 serde_json = "1.0.104"
 simple_logger = "4.2.0"
-simln-lib = { path = "../simln-lib" }
+# The virtual-time feature is required for the --virtual-time flag, which runs simulations on a paused runtime.
+simln-lib = { path = "../simln-lib", features = ["virtual-time"] }
 tokio = { version = "1.26.0", features = ["full"] }
 bitcoin = { version = "0.30.1" }
 ctrlc = "3.4.0"

sim-cli/src/main.rs

@@ -1,27 +1,27 @@
 use std::sync::Arc;
+use std::time::SystemTime;
 
 use clap::Parser;
 use log::LevelFilter;
 use sim_cli::parsing::{create_simulation, create_simulation_with_network, parse_sim_params, Cli};
+use simln_lib::latency_interceptor::LatencyIntercepor;
+use simln_lib::sim_node::Interceptor;
 use simln_lib::{
-    clock::SimulationClock,
-    latency_interceptor::LatencyIntercepor,
-    sim_node::{CustomRecords, Interceptor},
-    SimulationCfg,
+    clock::SimulationClock, runtime::block_on_virtual_time, sim_node::CustomRecords,
+    ActivityDefinition, Simulation, SimulationCfg,
 };
 use simple_logger::SimpleLogger;
 use tokio_util::task::TaskTracker;
 
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
+fn main() -> anyhow::Result<()> {
     // Enable tracing if building in developer mode.
     #[cfg(feature = "dev")]
     {
         console_subscriber::init();
     }
 
     let cli = Cli::parse();
-    let sim_params = parse_sim_params(&cli).await?;
+    let sim_params = parse_sim_params(&cli)?;
 
     SimpleLogger::new()
         .with_level(LevelFilter::Warn)
@@ -32,35 +32,52 @@ async fn main() -> anyhow::Result<()> {
 
     cli.validate(&sim_params)?;
 
-    let tasks = TaskTracker::new();
-
-    let (sim, validated_activities) = if sim_params.sim_network.is_empty() {
-        create_simulation(&cli, &sim_params, tasks.clone()).await?
-    } else {
-        let latency = cli.latency_ms.unwrap_or(0);
-        let interceptors = if latency > 0 {
-            vec![Arc::new(LatencyIntercepor::new_poisson(
-                latency as f32,
-                cli.fix_seed,
-            )?) as Arc<dyn Interceptor>]
+    let cfg = SimulationCfg::try_from(&cli)?;
+    let latency = cli.latency_ms.unwrap_or(0);
+    let build_and_run = move |clock: Arc<SimulationClock>| async move {
+        let (sim, activities) = if sim_params.sim_network.is_empty() {
+            create_simulation(cfg, &sim_params, clock, TaskTracker::new()).await?
         } else {
-            vec![]
+            let (sim, activities, _) = create_simulation_with_network(
+                cfg,
+                &sim_params,
+                clock,
+                TaskTracker::new(),
+                // Create an interceptor to add latency to payments, otherwise none.
+                if latency > 0 {
+                    vec![Arc::new(LatencyIntercepor::new_poisson(
+                        latency as f32,
+                        cli.fix_seed,
+                    )?) as Arc<dyn Interceptor>]
+                } else {
+                    vec![]
+                },
+                CustomRecords::default(),
+            )
+            .await?;
+            (sim, activities)
         };
-        let sim_cfg: SimulationCfg = SimulationCfg::try_from(&cli)?;
-        let clock = Arc::new(SimulationClock::new(cli.speedup_clock.unwrap_or(1))?);
-        let (sim, validated_activities, _) = create_simulation_with_network(
-            sim_cfg,
-            &sim_params,
-            clock,
-            tasks.clone(),
-            interceptors,
-            CustomRecords::default(),
-        )
-        .await?;
-        (sim, validated_activities)
+
+        run_simulation(sim, activities).await
     };
-    let sim2 = sim.clone();
 
+    // For virtual time, our helper will build the clock we need and pass it to build_and_run. If
+    // running with regular time, we can just pass a clock right in.
+    if cli.virtual_time {
+        block_on_virtual_time(SystemTime::now(), build_and_run)?
+    } else {
+        tokio::runtime::Runtime::new()?.block_on(build_and_run(Arc::new(SimulationClock::new(
+            SystemTime::now(),
+        ))))
+    }
+}
+
+/// Drives a fully-configured simulation to completion, wiring up a ctrl-c handler that triggers a clean shutdown.
+async fn run_simulation(
+    sim: Simulation<SimulationClock>,
+    validated_activities: Vec<ActivityDefinition>,
+) -> anyhow::Result<()> {
+    let sim2 = sim.clone();
     ctrlc::set_handler(move || {
         log::info!("Shutting down simulation.");
         sim2.shutdown();

sim-cli/src/parsing.rs

@@ -84,14 +84,15 @@ pub struct Cli {
     /// Seed to run random activity generator deterministically
     #[clap(long, short)]
     pub fix_seed: Option<u64>,
-    /// A multiplier to wall time to speed up the simulation's clock. Only available when when running on a network of
-    /// simulated nodes.
-    #[clap(long)]
-    pub speedup_clock: Option<u16>,
     /// Latency to optionally introduce for payments in a simulated network expressed in
     /// milliseconds.
     #[clap(long)]
     pub latency_ms: Option<u32>,
+    /// Run the simulated network on virtual time: time advances instantly to the next event instead of sleeping on the
+    /// wall clock, so the run finishes as fast as the CPU allows and is reproducible for a given seed. Only available on
+    /// a simulated network, and requires --total-time to bound the run.
+    #[clap(long, default_value_t = false)]
+    pub virtual_time: bool,
 }
 
 impl Cli {
@@ -111,18 +112,25 @@ impl Cli {
                 nodes or sim_graph to run with simulated nodes"
             ));
         }
-        if !sim_params.nodes.is_empty() && self.speedup_clock.is_some() {
-            return Err(anyhow!(
-                "Clock speedup is only allowed when running on a simulated network"
-            ));
-        }
-
         if !sim_params.nodes.is_empty() && self.latency_ms.is_some() {
             return Err(anyhow!(
                 "Latency for payments is only allowed when running on a simulated network"
             ));
         }
 
+        if self.virtual_time {
+            if !sim_params.nodes.is_empty() {
+                return Err(anyhow!(
+                    "Virtual time is only allowed when running on a simulated network; real nodes run on wall time"
+                ));
+            }
+            if self.total_time.is_none() {
+                return Err(anyhow!(
+                    "Virtual time requires --total-time, otherwise it advances forever"
+                ));
+            }
+        }
+
         if !sim_params.exclude.is_empty() {
             if sim_params.sim_network.is_empty() {
                 return Err(anyhow!(
@@ -339,14 +347,15 @@ pub async fn create_simulation_with_network(
     ))
 }
 
-/// Parses the cli options provided and creates a simulation to be run, connecting to lightning nodes and validating
-/// any activity described in the simulation file.
+/// Creates a simulation to be run against a set of real lightning nodes, connecting to the nodes described in
+/// `sim_params` and validating any activity described in the simulation file. The simulation is driven by `clock`,
+/// which must be constructed on the runtime that will run the simulation.
 pub async fn create_simulation(
-    cli: &Cli,
+    cfg: SimulationCfg,
     sim_params: &SimParams,
+    clock: Arc<SimulationClock>,
     tasks: TaskTracker,
 ) -> Result<(Simulation<SimulationClock>, Vec<ActivityDefinition>), anyhow::Error> {
-    let cfg: SimulationCfg = SimulationCfg::try_from(cli)?;
     let SimParams {
         nodes,
         sim_network: _sim_network,
@@ -365,9 +374,7 @@ pub async fn create_simulation(
             cfg,
             clients,
             tasks,
-            // When running on a real network, the underlying node may use wall time so we always use a clock with no
-            // speedup.
-            Arc::new(SimulationClock::new(1)?),
+            clock,
             shutdown_trigger,
             shutdown_listener,
         ),
@@ -531,7 +538,7 @@ async fn validate_activities(
     Ok(validated_activities)
 }
 
-async fn read_sim_path(data_dir: PathBuf, sim_file: PathBuf) -> anyhow::Result<PathBuf> {
+fn read_sim_path(data_dir: PathBuf, sim_file: PathBuf) -> anyhow::Result<PathBuf> {
     if sim_file.exists() {
         Ok(sim_file)
     } else if sim_file.is_relative() {
@@ -540,15 +547,15 @@ async fn read_sim_path(data_dir: PathBuf, sim_file: PathBuf) -> anyhow::Result<P
             Ok(sim_path)
         } else {
             log::info!("Simulation file '{}' does not exist.", sim_path.display());
-            select_sim_file(data_dir).await
+            select_sim_file(data_dir)
         }
     } else {
         log::info!("Simulation file '{}' does not exist.", sim_file.display());
-        select_sim_file(data_dir).await
+        select_sim_file(data_dir)
     }
 }
 
-pub async fn select_sim_file(data_dir: PathBuf) -> anyhow::Result<PathBuf> {
+pub fn select_sim_file(data_dir: PathBuf) -> anyhow::Result<PathBuf> {
     let sim_files = std::fs::read_dir(data_dir.clone())?
         .filter_map(|f| {
             f.ok().and_then(|f| {
@@ -585,8 +592,8 @@ fn mkdir(dir: PathBuf) -> anyhow::Result<PathBuf> {
     Ok(dir)
 }
 
-pub async fn parse_sim_params(cli: &Cli) -> anyhow::Result<SimParams> {
-    let sim_path = read_sim_path(cli.data_dir.clone(), cli.sim_file.clone()).await?;
+pub fn parse_sim_params(cli: &Cli) -> anyhow::Result<SimParams> {
+    let sim_path = read_sim_path(cli.data_dir.clone(), cli.sim_file.clone())?;
     let sim_params = serde_json::from_str(&std::fs::read_to_string(sim_path)?).map_err(|e| {
         anyhow!(
             "Could not deserialize node connection data or activity description from simulation file (line {}, col {}, err: {}).",

simln-lib/Cargo.toml

@@ -24,7 +24,7 @@ thiserror = "1.0.45"
 log = "0.4.20"
 triggered = "0.1.2"
 mpsc = "0.2.0"
-tokio = "1.31.0"
+tokio = { version = "1.31.0", features = ["rt", "sync", "time", "macros", "fs", "io-util"] }
 rand = "0.8.5"
 hex = "0.4.3"
 csv = "1.2.2"
@@ -35,8 +35,12 @@ reqwest = { version = "0.12", features = ["json", "multipart"] }
 tokio-util = { version = "0.7.13", features = ["rt"] }
 ldk-server-client = { git = "https://github.com/lightningdevkit/ldk-server", rev = "8163f4fe139368613959bf4f10b19ee6a5b9b4ab" }
 
+[features]
+virtual-time = ["tokio/test-util"]
+
 [dev-dependencies]
 ntest = "0.9.0"
 mockall = "0.13.1"
 futures = "0.3.31"
 tempfile = "3"
+tokio = { version = "1.31.0", features = ["macros", "rt-multi-thread"] }