blitz_traits/
handoff.rs

1//! Consumer-owned event batches with typed pointer publication.
2
3use std::collections::VecDeque;
4use std::sync::{Arc, Mutex, Weak};
5
6struct Segment<T> {
7    producer: spsc::Producer<Arc<T>>,
8    consumer: spsc::Consumer<Arc<T>>,
9}
10
11struct Pending<T> {
12    segments: VecDeque<Segment<T>>,
13    closed: bool,
14}
15
16struct Queue<T> {
17    pending: Mutex<Pending<T>>,
18    wake: Arc<dyn Fn() + Send + Sync>,
19}
20
21/// The single owner of an unbounded FIFO of typed event pointers.
22///
23/// Publication schedules the owner's wake callback on nagoya only when an
24/// empty batch becomes nonempty. Producers share one serialized ring lane,
25/// following the debug-control handoff. Full rings append another segment so
26/// publication stays unbounded. Taking a batch transfers the ring handles in
27/// one operation, releasing the lock before any event is handled. There is no
28/// idle receive loop and no payload copying.
29pub struct EventQueue<T>(Arc<Queue<T>>);
30
31impl<T> EventQueue<T> {
32    pub fn new(wake: impl Fn() + Send + Sync + 'static) -> Self {
33        Self(Arc::new(Queue {
34            pending: Mutex::new(Pending {
35                segments: VecDeque::new(),
36                closed: false,
37            }),
38            wake: Arc::new(wake),
39        }))
40    }
41
42    /// A publication capability that does not keep the consumer alive.
43    pub fn target(&self) -> EventTarget<T> {
44        EventTarget(Arc::downgrade(&self.0))
45    }
46
47    /// Take the current batch in publication order, releasing the lock before
48    /// the caller handles any event or drops any payload.
49    pub fn take_pending(&mut self) -> EventBatch<T> {
50        EventBatch(std::mem::take(&mut self.0.pending.lock().unwrap().segments))
51    }
52}
53
54impl<T> Drop for EventQueue<T> {
55    fn drop(&mut self) {
56        let pending = {
57            let mut state = self.0.pending.lock().unwrap();
58            state.closed = true;
59            std::mem::take(&mut state.segments)
60        };
61        drop(pending);
62    }
63}
64
65/// A finite batch of already published event pointers, owned by the consumer.
66pub struct EventBatch<T>(VecDeque<Segment<T>>);
67
68impl<T> Iterator for EventBatch<T> {
69    type Item = Arc<T>;
70
71    fn next(&mut self) -> Option<Self::Item> {
72        while let Some(segment) = self.0.front() {
73            if let Some(value) = segment.consumer.try_pop() {
74                return Some(value);
75            }
76            self.0.pop_front();
77        }
78        None
79    }
80}
81
82/// Weak capability for handing an Arc to its consumer's event queue.
83pub struct EventTarget<T>(Weak<Queue<T>>);
84
85impl<T> Clone for EventTarget<T> {
86    fn clone(&self) -> Self {
87        Self(Weak::clone(&self.0))
88    }
89}
90
91impl<T> EventTarget<T> {
92    /// Returns the original pointer if the consumer has been dropped.
93    /// Callers that later move the payload out must transfer a unique Arc.
94    pub fn publish(&self, value: Arc<T>) -> Result<(), Arc<T>> {
95        let Some(queue) = self.0.upgrade() else {
96            return Err(value);
97        };
98        let needs_wake = {
99            let mut pending = queue.pending.lock().unwrap();
100            if pending.closed {
101                return Err(value);
102            }
103            let empty = pending.segments.is_empty();
104            let value = match pending.segments.back() {
105                Some(segment) => match segment.producer.try_push(value) {
106                    None => return Ok(()),
107                    Some(value) => value,
108                },
109                None => value,
110            };
111            let (producer, consumer) = spsc::make(64);
112            assert!(producer.try_push(value).is_none());
113            pending.segments.push_back(Segment { producer, consumer });
114            empty
115        };
116        if needs_wake {
117            let wake = Arc::clone(&queue.wake);
118            // Dropping nagoya's JoinHandle detaches this finite wake task.
119            drop(nagoya::spawn(async move { wake() }));
120        }
121        Ok(())
122    }
123}
blitz_traits/handoff.rs

blitz_traits/
handoff.rs
