sim-rs: correct the flow of time in networking code

Author: SupernaviX

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

@@ -50,16 +50,23 @@ impl<T> Ord for FutureEvent<T> {
 pub struct Clock {
     time: Arc<AtomicTimestamp>,
     waiters: Arc<AtomicUsize>,
+    tasks: Arc<AtomicUsize>,
     tx: mpsc::UnboundedSender<ClockEvent>,
 }
 
 impl Clock {
     fn new(
         time: Arc<AtomicTimestamp>,
         waiters: Arc<AtomicUsize>,
+        tasks: Arc<AtomicUsize>,
         tx: mpsc::UnboundedSender<ClockEvent>,
     ) -> Self {
-        Self { time, waiters, tx }
+        Self {
+            time,
+            waiters,
+            tasks,
+            tx,
+        }
     }
 
     pub fn now(&self) -> Timestamp {
@@ -73,6 +80,7 @@ impl Clock {
         ClockBarrier {
             id,
             time: self.time.clone(),
+            tasks: self.tasks.clone(),
             tx: self.tx.clone(),
         }
     }
@@ -81,6 +89,7 @@ impl Clock {
 pub struct ClockBarrier {
     id: usize,
     time: Arc<AtomicTimestamp>,
+    tasks: Arc<AtomicUsize>,
     tx: mpsc::UnboundedSender<ClockEvent>,
 }
 
@@ -90,7 +99,7 @@ impl ClockBarrier {
     }
 
     pub fn start_task(&self) {
-        let _ = self.tx.send(ClockEvent::StartTask);
+        self.tasks.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
     }
 
     pub fn finish_task(&self) {

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

@@ -15,6 +15,7 @@ pub struct ClockCoordinator {
     tx: mpsc::UnboundedSender<ClockEvent>,
     rx: mpsc::UnboundedReceiver<ClockEvent>,
     waiter_count: Arc<AtomicUsize>,
+    tasks: Arc<AtomicUsize>,
 }
 
 impl Default for ClockCoordinator {
@@ -28,18 +29,21 @@ impl ClockCoordinator {
         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 {
             time,
             tx,
             rx,
             waiter_count,
+            tasks,
         }
     }
 
     pub fn clock(&self) -> Clock {
         Clock::new(
             self.time.clone(),
             self.waiter_count.clone(),
+            self.tasks.clone(),
             self.tx.clone(),
         )
     }
@@ -52,18 +56,18 @@ impl ClockCoordinator {
 
         let mut queue: BTreeMap<Timestamp, Vec<usize>> = BTreeMap::new();
         let mut running = waiters.len();
-        let mut open_tasks = 0;
         while let Some(event) = self.rx.recv().await {
             match event {
                 ClockEvent::Wait { actor, until, done } => {
+                    assert!(until.is_none_or(|t| t > self.time.load(Ordering::Acquire)));
                     if waiters[actor].replace(Waiter { until, done }).is_some() {
                         panic!("An actor has somehow managed to wait twice");
                     }
                     running -= 1;
                     if let Some(timestamp) = until {
                         queue.entry(timestamp).or_default().push(actor);
                     }
-                    while running == 0 && open_tasks == 0 {
+                    while running == 0 && self.tasks.load(Ordering::Acquire) == 0 {
                         // advance time
                         let (timestamp, waiter_ids) = queue.pop_first().unwrap();
                         self.time.store(timestamp, Ordering::Release);
@@ -86,11 +90,12 @@ impl ClockCoordinator {
                         running += 1;
                     }
                 }
-                ClockEvent::StartTask => {
-                    open_tasks += 1;
-                }
                 ClockEvent::FinishTask => {
-                    open_tasks -= 1;
+                    self.tasks.fetch_sub(1, Ordering::AcqRel);
+                    assert!(
+                        running != 0,
+                        "All tasks were completed while there were no actors to complete them"
+                    );
                 }
             }
         }
@@ -112,7 +117,6 @@ pub enum ClockEvent {
     CancelWait {
         actor: usize,
     },
-    StartTask,
     FinishTask,
 }
 

sim-rs/sim-core/src/network.rs

@@ -5,7 +5,7 @@ use coordinator::{EdgeConfig, Message, NetworkCoordinator};
 use tokio::sync::mpsc;
 
 use crate::{
-    clock::{Clock, ClockBarrier, Timestamp},
+    clock::{Clock, ClockBarrier},
     config::NodeId,
 };
 
@@ -95,7 +95,6 @@ impl<TProtocol, TMessage> NetworkSink<TProtocol, TMessage> {
         bytes: u64,
         protocol: TProtocol,
         message: TMessage,
-        sent: Timestamp,
     ) -> Result<()> {
         if self
             .sink
@@ -105,7 +104,6 @@ impl<TProtocol, TMessage> NetworkSink<TProtocol, TMessage> {
                 body: message,
                 bytes,
                 protocol,
-                sent,
             })
             .is_err()
         {

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

@@ -79,20 +79,20 @@ impl<TProtocol: Clone + Eq + Hash, TMessage: Debug> NetworkCoordinator<TProtocol
                     }
                 },
                 Some(message) = self.source.recv() => {
-                    self.schedule_message(message);
+                    self.schedule_message(message, clock.now());
                     clock.finish_task();
                 }
             }
         }
     }
 
-    fn schedule_message(&mut self, message: Message<TProtocol, TMessage>) {
+    fn schedule_message(&mut self, message: Message<TProtocol, TMessage>, now: Timestamp) {
         let link = Link {
             from: message.from,
             to: message.to,
         };
         let connection = self.connections.get_mut(&link).unwrap();
-        connection.send(message.body, message.bytes, message.protocol, message.sent);
+        connection.send(message.body, message.bytes, message.protocol, now);
         if let Some(timestamp) = connection.next_arrival_time() {
             self.events.push(link, Reverse(timestamp));
         }
@@ -105,5 +105,4 @@ pub struct Message<TProtocol, TMessage> {
     pub protocol: TProtocol,
     pub body: TMessage,
     pub bytes: u64,
-    pub sent: Timestamp,
 }

sim-rs/sim-core/src/sim/node.rs

@@ -1498,7 +1498,7 @@ impl Node {
         }
         self.clock.start_task();
         self.msg_sink
-            .send_to(to, msg.bytes_size(), msg.protocol(), msg, self.clock.now())
+            .send_to(to, msg.bytes_size(), msg.protocol(), msg)
     }
 
     fn slot_to_pipeline(&self, slot: u64) -> u64 {