Merge pull request #32 from Abso1ut3Zer0/investigate-channel-notification-bug

replace deduplication set with a notification queue for reliable waking

Author: Abso1ut3Zer0

wavelet/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "wavelet"
-version = "0.4.6"
+version = "0.5.0"
 edition = "2024"
 description = "High-performance graph-based stream processing runtime"
 license = "MIT OR Apache-2.0"
@@ -27,6 +27,7 @@ exclude = [
 ahash = "0.8.12"
 crossbeam-queue = "0.3.12"
 crossbeam-utils = "0.8.21"
+derive_builder = "0.20.2"
 enum-as-inner = "0.6.1"
 mio = { version = "1.0.4", features = ["os-poll", "net"] }
 petgraph = "0.8.2"

wavelet/src/channel.rs

@@ -168,7 +168,7 @@ impl<T> Channel<T> {
 #[cfg(test)]
 mod tests {
     use crate::Control;
-    use crate::channel::{Receiver, TryReceiveError};
+    use crate::channel::{Receiver, TryReceiveError, TrySendError};
     use crate::prelude::{Executor, TestClock};
     use crate::runtime::NodeBuilder;
     use std::time::Duration;
@@ -511,4 +511,138 @@ mod tests {
         assert!(executor.has_mutated(&node));
         assert_eq!(*node.borrow(), "1");
     }
+
+    // spurious wakeups are normal, this is just to examine them
+    #[ignore]
+    #[test]
+    fn test_single_producer_spurious_wakeup() {
+        use std::sync::Arc;
+        use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+        use std::thread;
+        use std::time::{Duration, Instant};
+
+        let mut executor = Executor::new();
+        let mut clock = TestClock::new();
+
+        // The key metric: how many times the node wakes up to find nothing
+        let empty_wakeups = Arc::new(AtomicUsize::new(0));
+        let successful_receives = Arc::new(AtomicUsize::new(0));
+        let messages_sent = Arc::new(AtomicUsize::new(0));
+        let still_sending = Arc::new(AtomicBool::new(true));
+
+        let empty = empty_wakeups.clone();
+        let success = successful_receives.clone();
+
+        let (_node, tx) = NodeBuilder::new(0usize)
+            .build_with_channel(&mut executor, 64, move |state, _ctx, rx| {
+                let mut got_any = false;
+
+                // Try to receive messages
+                for _ in 0..16 {
+                    // Process up to 16 messages per cycle
+                    match rx.try_receive() {
+                        Ok(value) => {
+                            got_any = true;
+                            // Verify sequential ordering from single producer
+                            assert_eq!(value, *state, "Expected {}, got {}", *state, value);
+                            *state += 1;
+                            success.fetch_add(1, Ordering::Relaxed);
+                        }
+                        Err(TryReceiveError::Empty) => {
+                            break;
+                        }
+                        Err(TryReceiveError::ChannelClosed) => {
+                            break;
+                        }
+                    }
+                }
+
+                // THE BUG: If we were woken up but got nothing, that's wrong!
+                if !got_any {
+                    empty.fetch_add(1, Ordering::Relaxed);
+                    println!("BUG at message {}: Node woke but channel empty!", *state);
+                }
+
+                Control::Unchanged
+            })
+            .unwrap();
+
+        let sent = messages_sent.clone();
+        let running = still_sending.clone();
+
+        // Single producer thread sending sequential messages
+        let sender = thread::spawn(move || {
+            for i in 0..10000 {
+                // Send sequential numbers
+                loop {
+                    match tx.try_send(i) {
+                        Ok(_) => {
+                            sent.fetch_add(1, Ordering::Relaxed);
+                            break;
+                        }
+                        Err(TrySendError::ChannelFull(_)) => {
+                            // Channel full, yield and retry
+                            thread::yield_now();
+                            continue;
+                        }
+                        Err(TrySendError::ChannelClosed(_)) => {
+                            return;
+                        }
+                    }
+                }
+
+                // Vary the timing slightly
+                if i % 100 == 0 {
+                    thread::yield_now();
+                }
+            }
+            running.store(false, Ordering::Relaxed);
+        });
+
+        // Run executor on main thread
+        let start = Instant::now();
+        let mut cycles = 0;
+
+        while still_sending.load(Ordering::Relaxed)
+            || successful_receives.load(Ordering::Relaxed) < messages_sent.load(Ordering::Relaxed)
+        {
+            executor.cycle(&mut clock, Some(Duration::ZERO)).unwrap();
+            cycles += 1;
+
+            // Occasionally yield to create scheduling variance
+            if cycles % 50 == 0 {
+                thread::yield_now();
+            }
+
+            if start.elapsed() > Duration::from_secs(5) {
+                println!("Timeout!");
+                break;
+            }
+        }
+
+        sender.join().unwrap();
+
+        // Final drain
+        for _ in 0..10 {
+            executor.cycle(&mut clock, Some(Duration::ZERO)).unwrap();
+        }
+
+        let empty = empty_wakeups.load(Ordering::Relaxed);
+        let success = successful_receives.load(Ordering::Relaxed);
+        let sent = messages_sent.load(Ordering::Relaxed);
+
+        println!(
+            "Sent: {}, Received: {}, Empty wakeups: {}",
+            sent, success, empty
+        );
+        println!("Ran {} cycles in {:?}", cycles, start.elapsed());
+
+        // Critical assertion: should NEVER wake up to empty channel
+        assert_eq!(
+            empty, 0,
+            "Memory ordering bug detected: {} empty wakeups",
+            empty
+        );
+        assert_eq!(sent, success, "Lost {} messages", sent - success);
+    }
 }

wavelet/src/runtime/event_driver.rs

@@ -8,42 +8,45 @@ pub use crate::runtime::event_driver::yield_driver::*;
 use crate::runtime::graph::Graph;
 use crate::runtime::scheduler::Scheduler;
 use crate::runtime::{Clock, CycleTime};
-use ahash::{HashSet, HashSetExt};
+use crossbeam_queue::ArrayQueue;
+use derive_builder::Builder;
 use petgraph::prelude::NodeIndex;
 use std::io;
 use std::sync::Arc;
 use std::time::Duration;
 
-const MINIMUM_TIMER_PRECISION: std::time::Duration = std::time::Duration::from_millis(1);
-const IO_CAPACITY: usize = 1024;
-const EVENT_CAPACITY: usize = 1024;
+const MINIMUM_TIMER_PRECISION: Duration = Duration::from_millis(1);
 
 /// A handle for waking a node from external threads or contexts.
 ///
 /// `Notifier` provides a thread-safe way to schedule a node for execution
 /// from outside the main event loop. Useful for integrating with external
 /// libraries, user input, or cross-thread communication.
 pub struct Notifier {
-    raw_events: Arc<spin::Mutex<HashSet<NodeIndex>>>,
+    notifications: Arc<ArrayQueue<NodeIndex>>,
     waker: Arc<mio::Waker>,
     node_index: NodeIndex,
 }
 
 impl Clone for Notifier {
     fn clone(&self) -> Self {
-        Self::new(self.raw_events.clone(), self.waker.clone(), self.node_index)
+        Self::new(
+            self.notifications.clone(),
+            self.waker.clone(),
+            self.node_index,
+        )
     }
 }
 
 impl Notifier {
     /// Creates a new notifier handle (internal use only).
     const fn new(
-        raw_events: Arc<spin::Mutex<HashSet<NodeIndex>>>,
+        notifications: Arc<ArrayQueue<NodeIndex>>,
         waker: Arc<mio::Waker>,
         node_index: NodeIndex,
     ) -> Self {
         Self {
-            raw_events,
+            notifications,
             waker,
             node_index,
         }
@@ -53,13 +56,27 @@ impl Notifier {
     ///
     /// This method is thread-safe and can be called from any thread.
     /// The node will be scheduled on the next polling cycle.
+    ///
+    /// Note: we assume that notification events will coalesce,
+    /// so we only attempt to write to the notification queue.
     #[inline(always)]
     pub fn notify(&self) {
-        self.raw_events.lock().insert(self.node_index);
+        self.notifications.push(self.node_index).ok();
         self.waker.wake().ok();
     }
 }
 
+/// Driver configuration options.
+#[derive(Builder)]
+pub struct EventDriverConfig {
+    #[builder(default = 256)]
+    pub notification_capacity: usize,
+    #[builder(default = 1024)]
+    pub io_capacity: usize,
+    #[builder(default = 16)]
+    pub poll_limit: usize,
+}
+
 /// Unified event management system that coordinates all event sources.
 ///
 /// The `EventDriver` orchestrates the three core event types in the runtime:
@@ -80,23 +97,32 @@ pub struct EventDriver {
     /// Processes immediate yield requests
     yield_driver: YieldDriver,
 
-    /// Tracks deduplicated raw events that have been received
-    raw_events: Arc<spin::Mutex<HashSet<NodeIndex>>>,
+    /// Notification queue tracking events on node indices
+    notifications: Arc<ArrayQueue<NodeIndex>>,
+
+    /// Poll limit for pulling node indices off of
+    /// the notifications queue
+    poll_limit: usize,
 }
 
 impl EventDriver {
     /// Creates a new event driver with default I/O capacity.
     pub(crate) fn new() -> Self {
-        Self::with_capacity(IO_CAPACITY)
+        Self::with_config(
+            EventDriverConfigBuilder::default()
+                .build()
+                .expect("expected default builder"),
+        )
     }
 
-    /// Creates a new event driver with the specified I/O event capacity.
-    pub(crate) fn with_capacity(capacity: usize) -> Self {
+    /// Creates a new event driver with the specified notification capacity.
+    pub(crate) fn with_config(cfg: EventDriverConfig) -> Self {
         Self {
-            io_driver: IoDriver::with_capacity(capacity),
+            io_driver: IoDriver::with_capacity(cfg.io_capacity),
             timer_driver: TimerDriver::new(),
             yield_driver: YieldDriver::new(),
-            raw_events: Arc::new(spin::Mutex::new(HashSet::with_capacity(EVENT_CAPACITY))),
+            notifications: Arc::new(ArrayQueue::new(cfg.notification_capacity)),
+            poll_limit: cfg.poll_limit,
         }
     }
 
@@ -118,7 +144,11 @@ impl EventDriver {
     /// Creates a new `Notifier` to inform the event driver of a raw event.
     #[inline(always)]
     pub fn register_notifier(&self, node_index: NodeIndex) -> Notifier {
-        Notifier::new(self.raw_events.clone(), self.io_driver.waker(), node_index)
+        Notifier::new(
+            self.notifications.clone(),
+            self.io_driver.waker(),
+            node_index,
+        )
     }
 
     /// Polls all event sources and schedules ready nodes.
@@ -140,13 +170,15 @@ impl EventDriver {
         timeout: Option<Duration>,
         epoch: usize,
     ) -> io::Result<CycleTime> {
-        {
-            let mut raw_events = self.raw_events.lock();
-            raw_events.drain().for_each(|node_idx| {
-                if let Some(depth) = graph.can_schedule(node_idx, epoch) {
-                    scheduler.schedule(node_idx, depth).ok();
+        for _ in 0..self.poll_limit {
+            match self.notifications.pop() {
+                None => break,
+                Some(node_idx) => {
+                    if let Some(depth) = graph.can_schedule(node_idx, epoch) {
+                        scheduler.schedule(node_idx, depth).ok();
+                    }
                 }
-            });
+            }
         }
 
         let cycle_time = clock.cycle_time();