multi: make simulation generic over clock with default impl

Add a generic clock parameter to our simulation so that we can
add the ability to mock time.

Author: carlaKC

simln-lib/src/lib.rs

@@ -1,3 +1,4 @@
+use self::clock::{Clock, SystemClock};
 use async_trait::async_trait;
 use bitcoin::secp256k1::PublicKey;
 use bitcoin::Network;
@@ -491,6 +492,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<dyn Clock>,
 }
 
 #[derive(Clone)]
@@ -527,6 +530,7 @@ impl Simulation {
             tasks,
             shutdown_trigger,
             shutdown_listener,
+            clock: Arc::new(SystemClock {}),
         }
     }
 
@@ -701,6 +705,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! {
@@ -709,7 +714,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()
@@ -776,23 +781,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,
@@ -930,11 +939,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:?}.");
@@ -967,6 +977,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();
 
@@ -980,6 +992,7 @@ impl Simulation {
                     executor.source_info,
                     executor.network_generator,
                     executor.payment_generator,
+                    clock,
                     pe_sender,
                     pe_listener,
                 )
@@ -1001,6 +1014,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,
@@ -1022,7 +1036,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 {
@@ -1084,6 +1098,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> {
@@ -1107,7 +1122,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
@@ -1183,13 +1198,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));
@@ -1288,6 +1304,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);
 
@@ -1298,7 +1315,7 @@ async fn run_results_logger(
                 break
             }
 
-            _ = time::sleep(interval) => {
+            _ = clock.sleep(interval) => {
                 log::info!("{}", logger.lock().await)
             }
         }