Support configurable timestamp resolution in rust simulation (#404)

* sim-rs: add config settings for timestamp resolution

* sim-rs: use u64 instead of duration in timestamp

* sim-rs: support coarse-grained timestamps in network layer

* sim-rs: support configurable timestamp resolution

Author: SupernaviX

data/simulation/config.d.ts

@@ -22,6 +22,10 @@ export interface Config {
   "treat-blocks-as-full": boolean;
   /** Only supported by Haskell simulation. */
   "cleanup-policies": CleanupPolicies;
+  /**
+   * The smallest unit of time to simulate, in fractions of a millisecond.
+   * Only supported by Rust simulation. */
+  "timestamp-resolution-ms": number;
 
   // Leios Protocol Configuration
   "leios-variant": LeiosVariant;

data/simulation/config.default.yaml

@@ -17,6 +17,7 @@ multiplex-mini-protocols: true
 simulate-transactions: true
 treat-blocks-as-full: false
 cleanup-policies: ["cleanup-expired-vote"]
+timestamp-resolution-ms: 0.000001
 
 ################################################################################
 # Leios Protocol Configuration

data/simulation/config.schema.json

@@ -347,6 +347,10 @@
       "description": "Only supported by Haskell simulation.",
       "type": "boolean"
     },
+    "timestamp-resolution-ms": {
+      "description": "The smallest unit of time to simulate, in fractions of a millisecond.\nOnly supported by Rust simulation.",
+      "type": "number"
+    },
     "treat-blocks-as-full": {
       "description": "When `true`, any delays and message sizes are calculated as if\neach block contained as much data as the expected average, rounded up.\nIn particular, for the sake of the above, we consider that:\n  - Each RB includes a certificate.\n  - Certificates contain votes from `vote-threshold` nodes.\n  - Vote bundles vote for `ceil eb-generation-probability` EBs.\n  - EBs reference `ceil (ib-generation-probability * leios-stage-length-slots)` IBs.\nOnly supported by Haskell simulation.",
       "type": "boolean"

sim-rs/sim-cli/src/events/aggregate.rs

@@ -1,7 +1,4 @@
-use std::{
-    collections::{BTreeMap, BTreeSet},
-    time::Duration,
-};
+use std::collections::{BTreeMap, BTreeSet};
 
 use serde::Serialize;
 use sim_core::{
@@ -236,7 +233,7 @@ impl TraceAggregator {
         self.current_time = event.time_s;
         if current_chunk != new_chunk {
             if new_chunk % 4 == 0 {
-                let timestamp = Duration::from_secs((new_chunk / 4) as u64).into();
+                let timestamp = Timestamp::from_secs((new_chunk / 4) as u64);
                 self.tx_counts.push(self.produce_tx_counts(timestamp));
             }
             Some(self.produce_message())

sim-rs/sim-cli/src/main.rs

@@ -140,7 +140,7 @@ async fn main() -> Result<()> {
     let monitor = tokio::spawn(EventMonitor::new(&config, events_source, args.output).run());
     pin!(monitor);
 
-    let clock_coordinator = ClockCoordinator::new();
+    let clock_coordinator = ClockCoordinator::new(config.timestamp_resolution);
     let clock = clock_coordinator.clock();
     let tracker = EventTracker::new(events_sink, clock.clone(), &config.nodes);
     let mut simulation = Simulation::new(config, tracker, clock_coordinator).await?;

sim-rs/sim-core/src/clock.rs

@@ -4,6 +4,7 @@ use std::{
     pin::Pin,
     sync::{atomic::AtomicUsize, Arc},
     task::{Context, Poll},
+    time::Duration,
 };
 
 pub use coordinator::ClockCoordinator;
@@ -48,6 +49,7 @@ impl<T> Ord for FutureEvent<T> {
 
 #[derive(Clone)]
 pub struct Clock {
+    timestamp_resolution: Duration,
     time: Arc<AtomicTimestamp>,
     waiters: Arc<AtomicUsize>,
     tasks: Arc<AtomicUsize>,
@@ -56,12 +58,14 @@ pub struct Clock {
 
 impl Clock {
     fn new(
+        timestamp_resolution: Duration,
         time: Arc<AtomicTimestamp>,
         waiters: Arc<AtomicUsize>,
         tasks: Arc<AtomicUsize>,
         tx: mpsc::UnboundedSender<ClockEvent>,
     ) -> Self {
         Self {
+            timestamp_resolution,
             time,
             waiters,
             tasks,
@@ -79,6 +83,7 @@ impl Clock {
             .fetch_add(1, std::sync::atomic::Ordering::AcqRel);
         ClockBarrier {
             id,
+            timestamp_resolution: self.timestamp_resolution,
             time: self.time.clone(),
             tasks: self.tasks.clone(),
             tx: self.tx.clone(),
@@ -88,6 +93,7 @@ impl Clock {
 
 pub struct ClockBarrier {
     id: usize,
+    timestamp_resolution: Duration,
     time: Arc<AtomicTimestamp>,
     tasks: Arc<AtomicUsize>,
     tx: mpsc::UnboundedSender<ClockEvent>,
@@ -107,7 +113,7 @@ impl ClockBarrier {
     }
 
     pub fn wait_until(&mut self, timestamp: Timestamp) -> Waiter {
-        self.wait(Some(timestamp))
+        self.wait(Some(timestamp.with_resolution(self.timestamp_resolution)))
     }
 
     pub fn wait_forever(&mut self) -> Waiter {

sim-rs/sim-core/src/clock/coordinator.rs

@@ -4,33 +4,30 @@ use std::{
         atomic::{AtomicUsize, Ordering},
         Arc,
     },
+    time::Duration,
 };
 
 use tokio::sync::{mpsc, oneshot};
 
 use super::{timestamp::AtomicTimestamp, Clock, Timestamp};
 
 pub struct ClockCoordinator {
+    timestamp_resolution: Duration,
     time: Arc<AtomicTimestamp>,
     tx: mpsc::UnboundedSender<ClockEvent>,
     rx: mpsc::UnboundedReceiver<ClockEvent>,
     waiter_count: Arc<AtomicUsize>,
     tasks: Arc<AtomicUsize>,
 }
 
-impl Default for ClockCoordinator {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 impl ClockCoordinator {
-    pub fn new() -> Self {
+    pub fn new(timestamp_resolution: Duration) -> Self {
         let time = Arc::new(AtomicTimestamp::new(Timestamp::zero()));
         let (tx, rx) = mpsc::unbounded_channel();
         let waiter_count = Arc::new(AtomicUsize::new(0));
         let tasks = Arc::new(AtomicUsize::new(0));
         Self {
+            timestamp_resolution,
             time,
             tx,
             rx,
@@ -41,6 +38,7 @@ impl ClockCoordinator {
 
     pub fn clock(&self) -> Clock {
         Clock::new(
+            self.timestamp_resolution,
             self.time.clone(),
             self.waiter_count.clone(),
             self.tasks.clone(),
@@ -129,9 +127,11 @@ mod tests {
 
     use super::ClockCoordinator;
 
+    const TIMESTAMP_RESOLUTION: Duration = Duration::from_nanos(1);
+
     #[tokio::test]
     async fn should_wait_once_all_workers_are_waiting() {
-        let mut coordinator = ClockCoordinator::new();
+        let mut coordinator = ClockCoordinator::new(TIMESTAMP_RESOLUTION);
         let clock = coordinator.clock();
         let t0 = clock.now();
         let t1 = t0 + Duration::from_millis(5);
@@ -158,7 +158,7 @@ mod tests {
 
     #[tokio::test]
     async fn should_cancel_wait_when_wait_future_is_dropped() {
-        let mut coordinator = ClockCoordinator::new();
+        let mut coordinator = ClockCoordinator::new(TIMESTAMP_RESOLUTION);
         let clock = coordinator.clock();
         let t0 = clock.now();
         let t1 = t0 + Duration::from_millis(5);
@@ -186,7 +186,7 @@ mod tests {
 
     #[tokio::test]
     async fn should_avoid_race_condition() {
-        let mut coordinator = ClockCoordinator::new();
+        let mut coordinator = ClockCoordinator::new(TIMESTAMP_RESOLUTION);
         let clock = coordinator.clock();
         let t0 = clock.now();
         let t1 = t0 + Duration::from_millis(5);
@@ -219,7 +219,7 @@ mod tests {
 
     #[tokio::test]
     async fn should_allow_time_to_stand_still() {
-        let mut coordinator = ClockCoordinator::new();
+        let mut coordinator = ClockCoordinator::new(TIMESTAMP_RESOLUTION);
         let clock = coordinator.clock();
         let t0 = clock.now();
         let t1 = t0 + Duration::from_millis(5);
@@ -253,7 +253,7 @@ mod tests {
 
     #[tokio::test]
     async fn should_allow_waiting_forever() {
-        let mut coordinator = ClockCoordinator::new();
+        let mut coordinator = ClockCoordinator::new(TIMESTAMP_RESOLUTION);
         let clock = coordinator.clock();
         let t0 = clock.now();
         let t1 = t0 + Duration::from_millis(5);

sim-rs/sim-core/src/clock/timestamp.rs

@@ -6,57 +6,57 @@ use std::{
 
 use serde::Serialize;
 
+const NANOS_PER_SEC: u64 = 1_000_000_000;
+
 /// A timestamp tracks the time from the start of the simulation.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Default)]
-pub struct Timestamp(Duration);
+pub struct Timestamp(u64);
 
 impl Timestamp {
     pub fn zero() -> Self {
-        Self(Duration::from_secs(0))
+        Self(0)
     }
 
     pub fn from_secs(secs: u64) -> Self {
-        Self(Duration::from_secs(secs))
+        Self(secs * NANOS_PER_SEC)
     }
 
     pub fn checked_sub_duration(self, rhs: Duration) -> Option<Self> {
-        Some(Self(self.0.checked_sub(rhs)?))
+        Some(Self(self.0.checked_sub(rhs.as_nanos() as u64)?))
     }
-}
 
-impl From<Duration> for Timestamp {
-    fn from(value: Duration) -> Self {
-        Self(value)
+    pub fn with_resolution(self, delta: Duration) -> Self {
+        Self(self.0.next_multiple_of(delta.as_nanos() as u64))
     }
 }
 
 impl Add<Duration> for Timestamp {
     type Output = Timestamp;
 
     fn add(self, rhs: Duration) -> Self::Output {
-        Timestamp(self.0 + rhs)
+        Self(self.0 + rhs.as_nanos() as u64)
     }
 }
 
 impl AddAssign<Duration> for Timestamp {
     fn add_assign(&mut self, rhs: Duration) {
-        self.0 += rhs;
+        self.0 += rhs.as_nanos() as u64;
     }
 }
 
 impl Sub<Timestamp> for Timestamp {
     type Output = Duration;
 
     fn sub(self, rhs: Timestamp) -> Self::Output {
-        self.0 - rhs.0
+        Duration::from_nanos(self.0 - rhs.0)
     }
 }
 
 impl Sub<Duration> for Timestamp {
     type Output = Timestamp;
 
     fn sub(self, rhs: Duration) -> Self::Output {
-        Self(self.0 - rhs)
+        Self(self.0 - rhs.as_nanos() as u64)
     }
 }
 
@@ -65,30 +65,20 @@ impl Serialize for Timestamp {
     where
         S: serde::Serializer,
     {
-        serializer.serialize_f32(self.0.as_secs_f32())
+        serializer.serialize_f64(self.0 as f64 / NANOS_PER_SEC as f64)
     }
 }
 
 pub struct AtomicTimestamp(AtomicU64);
 impl AtomicTimestamp {
     pub fn new(val: Timestamp) -> Self {
-        Self(AtomicU64::new(duration_to_u64(val.0)))
+        Self(AtomicU64::new(val.0))
     }
     pub fn load(&self, order: Ordering) -> Timestamp {
-        let val = self.0.load(order);
-        Timestamp(u64_to_duration(val))
+        Timestamp(self.0.load(order))
     }
 
     pub fn store(&self, val: Timestamp, order: Ordering) {
-        let val = duration_to_u64(val.0);
-        self.0.store(val, order);
+        self.0.store(val.0, order);
     }
 }
-
-fn u64_to_duration(val: u64) -> Duration {
-    Duration::from_nanos(val)
-}
-
-fn duration_to_u64(val: Duration) -> u64 {
-    val.as_nanos() as u64
-}

sim-rs/sim-core/src/config.rs

@@ -60,6 +60,7 @@ pub struct RawParameters {
     pub leios_variant: LeiosVariant,
     pub relay_strategy: RelayStrategy,
     pub simulate_transactions: bool,
+    pub timestamp_resolution_ms: f64,
 
     // Leios protocol configuration
     pub leios_stage_length_slots: u64,
@@ -447,6 +448,7 @@ impl MockTransactionConfig {
 #[derive(Debug, Clone)]
 pub struct SimConfiguration {
     pub seed: u64,
+    pub timestamp_resolution: Duration,
     pub slots: Option<u64>,
     pub emit_conformance_events: bool,
     pub aggregate_events: bool,
@@ -492,6 +494,7 @@ impl SimConfiguration {
         }
         Ok(Self {
             seed: 0,
+            timestamp_resolution: duration_ms(params.timestamp_resolution_ms),
             slots: None,
             emit_conformance_events: false,
             aggregate_events: false,