sim-lib/feat: make Simulation generic over Clock

Author: carlaKC

sim-cli/src/parsing.rs

@@ -3,6 +3,7 @@ use bitcoin::secp256k1::PublicKey;
 use clap::{builder::TypedValueParser, Parser};
 use log::LevelFilter;
 use serde::{Deserialize, Serialize};
+use simln_lib::clock::SimulationClock;
 use simln_lib::sim_node::{
     ln_node_from_graph, populate_network_graph, ChannelPolicy, SimGraph, SimulatedChannel,
 };
@@ -198,7 +199,7 @@ pub async fn create_simulation_with_network(
     cli: &Cli,
     sim_params: &SimParams,
     tasks: TaskTracker,
-) -> Result<(Simulation, Vec<ActivityDefinition>), anyhow::Error> {
+) -> Result<(Simulation<SimulationClock>, Vec<ActivityDefinition>), anyhow::Error> {
     let cfg: SimulationCfg = SimulationCfg::try_from(cli)?;
     let SimParams {
         nodes: _,
@@ -241,7 +242,14 @@ pub async fn create_simulation_with_network(
         get_validated_activities(&nodes, nodes_info, sim_params.activity.clone()).await?;
 
     Ok((
-        Simulation::new(cfg, nodes, tasks, shutdown_trigger, shutdown_listener),
+        Simulation::new(
+            cfg,
+            nodes,
+            tasks,
+            Arc::new(SimulationClock::new(1)?),
+            shutdown_trigger,
+            shutdown_listener,
+        ),
         validated_activities,
     ))
 }
@@ -252,7 +260,7 @@ pub async fn create_simulation(
     cli: &Cli,
     sim_params: &SimParams,
     tasks: TaskTracker,
-) -> Result<(Simulation, Vec<ActivityDefinition>), anyhow::Error> {
+) -> Result<(Simulation<SimulationClock>, Vec<ActivityDefinition>), anyhow::Error> {
     let cfg: SimulationCfg = SimulationCfg::try_from(cli)?;
     let SimParams {
         nodes,
@@ -267,7 +275,14 @@ pub async fn create_simulation(
         get_validated_activities(&clients, clients_info, sim_params.activity.clone()).await?;
 
     Ok((
-        Simulation::new(cfg, clients, tasks, shutdown_trigger, shutdown_listener),
+        Simulation::new(
+            cfg,
+            clients,
+            tasks,
+            Arc::new(SimulationClock::new(1)?),
+            shutdown_trigger,
+            shutdown_listener,
+        ),
         validated_activities,
     ))
 }

simln-lib/src/lib.rs

@@ -1,5 +1,6 @@
 #![deny(rustdoc::broken_intra_doc_links)]
 
+use self::clock::Clock;
 use async_trait::async_trait;
 use bitcoin::secp256k1::PublicKey;
 use bitcoin::Network;
@@ -545,7 +546,7 @@ impl SimulationCfg {
 /// The simulator can execute both predefined payment patterns and generate random payment activity
 /// based on configuration parameters.
 #[derive(Clone)]
-pub struct Simulation {
+pub struct Simulation<C: Clock + 'static> {
     /// Config for the simulation itself.
     cfg: SimulationCfg,
     /// The lightning node that is being simulated.
@@ -558,6 +559,8 @@ pub struct Simulation {
     /// High level triggers used to manage simulation tasks and shutdown.
     shutdown_trigger: Trigger,
     shutdown_listener: Listener,
+    /// Clock for the simulation.
+    clock: Arc<C>,
 }
 
 /// Configuration for writing simulation results to CSV files.
@@ -579,11 +582,12 @@ struct ExecutorKit {
     payment_generator: Box<dyn PaymentGenerator>,
 }
 
-impl Simulation {
+impl<C: Clock + 'static> Simulation<C> {
     pub fn new(
         cfg: SimulationCfg,
         nodes: HashMap<PublicKey, Arc<Mutex<dyn LightningNode>>>,
         tasks: TaskTracker,
+        clock: Arc<C>,
         shutdown_trigger: Trigger,
         shutdown_listener: Listener,
     ) -> Self {
@@ -594,6 +598,7 @@ impl Simulation {
             tasks,
             shutdown_trigger,
             shutdown_listener,
+            clock,
         }
     }
 
@@ -771,6 +776,7 @@ impl Simulation {
         if let Some(total_time) = self.cfg.total_time {
             let shutdown = self.shutdown_trigger.clone();
             let listener = self.shutdown_listener.clone();
+            let clock = self.clock.clone();
 
             self.tasks.spawn(async move {
                 select! {
@@ -779,7 +785,7 @@ impl Simulation {
                         log::debug!("Timeout task exited on listener signal");
                     }
 
-                    _ = time::sleep(total_time) => {
+                    _ = clock.sleep(total_time) => {
                         log::info!(
                             "Simulation run for {}s. Shutting down.",
                             total_time.as_secs()
@@ -846,23 +852,27 @@ impl Simulation {
 
         let result_logger_clone = result_logger.clone();
         let result_logger_listener = listener.clone();
+        let clock = self.clock.clone();
         tasks.spawn(async move {
             log::debug!("Starting results logger.");
             run_results_logger(
                 result_logger_listener,
                 result_logger_clone,
                 Duration::from_secs(60),
+                clock,
             )
             .await;
             log::debug!("Exiting results logger.");
         });
 
         // csr: consume simulation results
         let csr_write_results = self.cfg.write_results.clone();
+        let clock = self.clock.clone();
         tasks.spawn(async move {
             log::debug!("Starting simulation results consumer.");
             if let Err(e) = consume_simulation_results(
                 result_logger,
+                clock,
                 results_receiver,
                 listener,
                 csr_write_results,
@@ -1003,11 +1013,12 @@ impl Simulation {
             let ce_shutdown = self.shutdown_trigger.clone();
             let ce_output_sender = output_sender.clone();
             let ce_node = node.clone();
+            let clock = self.clock.clone();
             tasks.spawn(async move {
                 let node_info = ce_node.lock().await.get_info().clone();
                 log::debug!("Starting events consumer for {}.", node_info);
                 if let Err(e) =
-                    consume_events(ce_node, receiver, ce_output_sender, ce_listener).await
+                    consume_events(ce_node, clock, receiver, ce_output_sender, ce_listener).await
                 {
                     ce_shutdown.trigger();
                     log::error!("Event consumer for node {node_info} exited with error: {e:?}.");
@@ -1040,6 +1051,8 @@ impl Simulation {
             let pe_shutdown = self.shutdown_trigger.clone();
             let pe_listener = self.shutdown_listener.clone();
             let pe_sender = sender.clone();
+            let clock = self.clock.clone();
+
             tasks.spawn(async move {
                 let source = executor.source_info.clone();
 
@@ -1053,6 +1066,7 @@ impl Simulation {
                     executor.source_info,
                     executor.network_generator,
                     executor.payment_generator,
+                    clock,
                     pe_sender,
                     pe_listener,
                 )
@@ -1074,6 +1088,7 @@ impl Simulation {
 /// event being executed is piped into a channel to handle the result of the event.
 async fn consume_events(
     node: Arc<Mutex<dyn LightningNode>>,
+    clock: Arc<dyn Clock>,
     mut receiver: Receiver<SimulationEvent>,
     sender: Sender<SimulationOutput>,
     listener: Listener,
@@ -1095,7 +1110,7 @@ async fn consume_events(
                                 hash: None,
                                 amount_msat: amt_msat,
                                 destination: dest.pubkey,
-                                dispatch_time: SystemTime::now(),
+                                dispatch_time: clock.now(),
                             };
 
                             let outcome = match node.send_payment(dest.pubkey, amt_msat).await {
@@ -1157,6 +1172,7 @@ async fn produce_events<N: DestinationGenerator + ?Sized, A: PaymentGenerator +
     source: NodeInfo,
     network_generator: Arc<Mutex<N>>,
     node_generator: Box<A>,
+    clock: Arc<dyn Clock>,
     sender: Sender<SimulationEvent>,
     listener: Listener,
 ) -> Result<(), SimulationError> {
@@ -1180,7 +1196,7 @@ async fn produce_events<N: DestinationGenerator + ?Sized, A: PaymentGenerator +
             },
             // Wait until our time to next payment has elapsed then execute a random amount payment to a random
             // destination.
-            _ = time::sleep(wait) => {
+            _ = clock.sleep(wait) => {
                 let (destination, capacity) = network_generator.lock().await.choose_destination(source.pubkey).map_err(SimulationError::DestinationGenerationError)?;
 
                 // Only proceed with a payment if the amount is non-zero, otherwise skip this round. If we can't get
@@ -1256,13 +1272,14 @@ fn get_payment_delay<A: PaymentGenerator + ?Sized>(
 
 async fn consume_simulation_results(
     logger: Arc<Mutex<PaymentResultLogger>>,
+    clock: Arc<dyn Clock>,
     mut receiver: Receiver<(Payment, PaymentResult)>,
     listener: Listener,
     write_results: Option<WriteResults>,
 ) -> Result<(), SimulationError> {
     let mut writer = match write_results {
         Some(res) => {
-            let duration = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)?;
+            let duration = clock.now().duration_since(SystemTime::UNIX_EPOCH)?;
             let file = res
                 .results_dir
                 .join(format!("simulation_{:?}.csv", duration));
@@ -1361,6 +1378,7 @@ async fn run_results_logger(
     listener: Listener,
     logger: Arc<Mutex<PaymentResultLogger>>,
     interval: Duration,
+    clock: Arc<dyn Clock>,
 ) {
     log::info!("Summary of results will be reported every {:?}.", interval);
 
@@ -1371,7 +1389,7 @@ async fn run_results_logger(
                 break
             }
 
-            _ = time::sleep(interval) => {
+            _ = clock.sleep(interval) => {
                 log::info!("{}", logger.lock().await)
             }
         }

simln-lib/src/test_utils.rs

@@ -10,6 +10,7 @@ use std::{collections::HashMap, fmt, sync::Arc, time::Duration};
 use tokio::sync::Mutex;
 use tokio_util::task::TaskTracker;
 
+use crate::clock::SystemClock;
 use crate::{
     ActivityDefinition, LightningError, LightningNode, NodeInfo, PaymentGenerationError,
     PaymentGenerator, Simulation, SimulationCfg, ValueOrRange,
@@ -195,12 +196,15 @@ impl LightningTestNodeBuilder {
 
 /// Creates a new simulation with the given clients and activity definitions.
 /// Note: This sets a runtime for the simulation of 0, so run() will exit immediately.
-pub fn create_simulation(clients: HashMap<PublicKey, Arc<Mutex<dyn LightningNode>>>) -> Simulation {
+pub fn create_simulation(
+    clients: HashMap<PublicKey, Arc<Mutex<dyn LightningNode>>>,
+) -> Simulation<SystemClock> {
     let (shutdown_trigger, shutdown_listener) = triggered::trigger();
     Simulation::new(
         SimulationCfg::new(Some(0), 0, 0.0, None, None),
         clients,
         TaskTracker::new(),
+        Arc::new(SystemClock {}),
         shutdown_trigger,
         shutdown_listener,
     )