timely/dataflow/operators/core/capture/
event.rs

1//! Traits and types describing timely dataflow events.
2//!
3//! The `Event` type describes the information an operator can observe about a timely dataflow
4//! stream. There are two types of events, (i) the receipt of data and (ii) reports of progress
5//! of timestamps.
6
7use columnar::Columnar;
8use serde::{Deserialize, Serialize};
9
10use crate::progress::Stamp;
11
12/// Data and progress events of the captured stream.
13#[derive(Debug, Clone, Hash, Ord, PartialOrd, Eq, PartialEq, Deserialize, Serialize, Columnar)]
14#[serde(bound(deserialize = "T: Deserialize<'de> + Ord, C: Deserialize<'de>"))]
15pub enum Event<T, C> {
16    /// Progress received via `push_external_progress`.
17    Progress(Vec<(T, i64)>),
18    /// Messages received via the data stream, with their stamps.
19    ///
20    /// Stamps are sorted multisets, not antichains: stamps that have crossed a scope
21    /// boundary may contain comparable or duplicate elements.
22    Messages(Stamp<T>, C),
23}
24
25/// Iterates over contained `Event<T, C>`.
26///
27/// The `EventIterator` trait describes types that can iterate over `Cow`s of events,
28/// and which can be used to replay a stream into a new timely dataflow computation.
29///
30/// This method is not simply an iterator because of the lifetime in the result.
31pub trait EventIterator<T: Clone, C: Clone> {
32    /// Iterates over `Cow<Event<T, C>>` elements.
33    fn next(&mut self) -> Option<std::borrow::Cow<'_, Event<T, C>>>;
34}
35
36/// Receives `Event<T, C>` events.
37pub trait EventPusher<T, C> {
38    /// Provides a new `Event<T, D>` to the pusher.
39    fn push(&mut self, event: Event<T, C>);
40}
41
42// implementation for the linked list behind a `Handle`.
43impl<T, C> EventPusher<T, C> for ::std::sync::mpsc::Sender<Event<T, C>> {
44    fn push(&mut self, event: Event<T, C>) {
45        // NOTE: An Err(x) result just means "data not accepted" most likely
46        //       because the receiver is gone. No need to panic.
47        let _ = self.send(event);
48    }
49}
50
51/// A linked-list event pusher and iterator.
52pub mod link {
53
54    use std::borrow::Cow;
55    use std::rc::Rc;
56    use std::cell::RefCell;
57
58    use super::{Event, EventPusher, EventIterator};
59
60    /// A linked list of Event<T, C>.
61    pub struct EventLink<T, C> {
62        /// An event, if one exists.
63        ///
64        /// An event might not exist, if either we want to insert a `None` and have the output iterator pause,
65        /// or in the case of the very first linked list element, which has no event when constructed.
66        pub event: Option<Event<T, C>>,
67        /// The next event, if it exists.
68        pub next: RefCell<Option<Rc<EventLink<T, C>>>>,
69    }
70
71    impl<T, C> EventLink<T, C> {
72        /// Allocates a new `EventLink`.
73        pub fn new() -> EventLink<T, C> {
74            EventLink { event: None, next: RefCell::new(None) }
75        }
76    }
77
78    // implementation for the linked list behind a `Handle`.
79    impl<T, C> EventPusher<T, C> for Rc<EventLink<T, C>> {
80        fn push(&mut self, event: Event<T, C>) {
81            *self.next.borrow_mut() = Some(Rc::new(EventLink { event: Some(event), next: RefCell::new(None) }));
82            let next = Rc::clone(self.next.borrow().as_ref().unwrap());
83            *self = next;
84        }
85    }
86
87    impl<T: Clone, C: Clone> EventIterator<T, C> for Rc<EventLink<T, C>> {
88        fn next(&mut self) -> Option<Cow<'_, Event<T, C>>> {
89            let is_some = self.next.borrow().is_some();
90            if is_some {
91                let next = Rc::clone(self.next.borrow().as_ref().unwrap());
92                *self = next;
93                if let Some(this) = Rc::get_mut(self) {
94                    this.event.take().map(Cow::Owned)
95                }
96                else {
97                    self.event.as_ref().map(Cow::Borrowed)
98                }
99            }
100            else {
101                None
102            }
103        }
104    }
105
106    // Drop implementation to prevent stack overflow through naive drop impl.
107    impl<T, C> Drop for EventLink<T, C> {
108        fn drop(&mut self) {
109            while let Some(link) = self.next.replace(None) {
110                if let Ok(head) = Rc::try_unwrap(link) {
111                    *self = head;
112                }
113            }
114        }
115    }
116
117    impl<T, C> Default for EventLink<T, C> {
118        fn default() -> Self {
119            Self::new()
120        }
121    }
122
123    #[test]
124    fn avoid_stack_overflow_in_drop() {
125        #[cfg(miri)]
126        let limit = 1_000;
127        #[cfg(not(miri))]
128        let limit = 1_000_000;
129        let mut event1 = Rc::new(EventLink::<(),()>::new());
130        let _event2 = Rc::clone(&event1);
131        for _ in 0 .. limit {
132            event1.push(Event::Progress(vec![]));
133        }
134    }
135}
136
137/// A thread-safe linked-list event pusher and iterator.
138pub mod link_sync {
139
140    use std::borrow::Cow;
141    use std::sync::{Arc, Mutex};
142
143    use super::{Event, EventPusher, EventIterator};
144
145    /// A linked list of Event<T, C> usable across threads.
146    pub struct EventLink<T, C> {
147        /// An event, if one exists.
148        ///
149        /// An event might not exist, if either we want to insert a `None` and have the output iterator pause,
150        /// or in the case of the very first linked list element, which has no event when constructed.
151        pub event: Option<Event<T, C>>,
152        /// The next event, if it exists.
153        pub next: Mutex<Option<Arc<EventLink<T, C>>>>,
154    }
155
156    impl<T, C> EventLink<T, C> {
157        /// Allocates a new `EventLink`.
158        pub fn new() -> EventLink<T, C> {
159            EventLink { event: None, next: Mutex::new(None) }
160        }
161    }
162
163    impl<T, C> EventPusher<T, C> for Arc<EventLink<T, C>> {
164        fn push(&mut self, event: Event<T, C>) {
165            let mut guard = self.next.lock().unwrap();
166            *guard = Some(Arc::new(EventLink { event: Some(event), next: Mutex::new(None) }));
167            let next = Arc::clone(guard.as_ref().unwrap());
168            drop(guard);
169            *self = next;
170        }
171    }
172
173    impl<T: Clone, C: Clone> EventIterator<T, C> for Arc<EventLink<T, C>> {
174        fn next(&mut self) -> Option<Cow<'_, Event<T, C>>> {
175            let is_some = self.next.lock().unwrap().is_some();
176            if is_some {
177                let next = Arc::clone(self.next.lock().unwrap().as_ref().unwrap());
178                *self = next;
179                if let Some(this) = Arc::get_mut(self) {
180                    this.event.take().map(Cow::Owned)
181                }
182                else {
183                    self.event.as_ref().map(Cow::Borrowed)
184                }
185            }
186            else {
187                None
188            }
189        }
190    }
191
192    // Drop implementation to prevent stack overflow through naive drop impl.
193    impl<T, C> Drop for EventLink<T, C> {
194        fn drop(&mut self) {
195            while let Some(link) = self.next.get_mut().unwrap().take() {
196                if let Ok(head) = Arc::try_unwrap(link) {
197                    *self = head;
198                }
199            }
200        }
201    }
202
203    impl<T, C> Default for EventLink<T, C> {
204        fn default() -> Self {
205            Self::new()
206        }
207    }
208
209    #[test]
210    fn avoid_stack_overflow_in_drop() {
211        #[cfg(miri)]
212        let limit = 1_000;
213        #[cfg(not(miri))]
214        let limit = 1_000_000;
215        let mut event1 = Arc::new(EventLink::<(),()>::new());
216        let _event2 = Arc::clone(&event1);
217        for _ in 0 .. limit {
218            event1.push(Event::Progress(vec![]));
219        }
220    }
221}
222
223/// A binary event pusher and iterator.
224pub mod binary {
225
226    use std::borrow::Cow;
227    use std::io::ErrorKind;
228    use std::ops::DerefMut;
229    use std::sync::Arc;
230
231    use serde::{de::DeserializeOwned, Serialize};
232    use timely_communication::allocator::zero_copy::bytes_slab::{BytesRefill, BytesSlab};
233
234    use super::{Event, EventPusher, EventIterator};
235
236    /// A wrapper for `W: Write` implementing `EventPusher<T, C>`.
237    pub struct EventWriter<T, C, W: ::std::io::Write> {
238        stream: W,
239        phant: ::std::marker::PhantomData<(T, C)>,
240    }
241
242    impl<T, C, W: ::std::io::Write> EventWriter<T, C, W> {
243        /// Allocates a new `EventWriter` wrapping a supplied writer.
244        pub fn new(w: W) -> Self {
245            Self {
246                stream: w,
247                phant: ::std::marker::PhantomData,
248            }
249        }
250    }
251
252    impl<T: Serialize, C: Serialize, W: ::std::io::Write> EventPusher<T, C> for EventWriter<T, C, W> {
253        fn push(&mut self, event: Event<T, C>) {
254            // TODO: `push` has no mechanism to report errors, so we `unwrap`.
255            let len = ::bincode::serialized_size(&event).expect("Event bincode failed");
256            self.stream.write_all(&len.to_le_bytes()).expect("Event write failed");
257            ::bincode::serialize_into(&mut self.stream, &event).expect("Event bincode failed");
258        }
259    }
260
261    /// A Wrapper for `R: Read` implementing `EventIterator<T, D>`.
262    pub struct EventReader<T, C, R: ::std::io::Read> {
263        reader: R,
264        buf: BytesSlab,
265        phant: ::std::marker::PhantomData<(T, C)>,
266    }
267
268    impl<T, C, R: ::std::io::Read> EventReader<T, C, R> {
269        /// Allocates a new `EventReader` wrapping a supplied reader.
270        pub fn new(r: R) -> Self {
271            let refill = BytesRefill {
272                logic: Arc::new(|size| {
273                    Box::new(vec![0_u8; size]) as Box<dyn DerefMut<Target = [u8]> + Send>
274                }),
275                limit: None,
276            };
277            Self {
278                reader: r,
279                buf: BytesSlab::new(20, refill),
280                phant: ::std::marker::PhantomData,
281            }
282        }
283    }
284
285    impl<T: DeserializeOwned + Ord + Clone, C: DeserializeOwned + Clone, R: ::std::io::Read> EventIterator<T, C> for EventReader<T, C, R> {
286        fn next(&mut self) -> Option<Cow<'_, Event<T, C>>> {
287            self.buf.ensure_capacity(1);
288            // Attempt to read some more bytes into self.buffer.
289            match self.reader.read(self.buf.empty()) {
290                Ok(n) => self.buf.make_valid(n),
291                Err(e) if e.kind() == ErrorKind::WouldBlock => {}
292                Err(e) => panic!("read failed: {e}"),
293            };
294
295            let valid = self.buf.valid();
296            if valid.len() >= 8 {
297                let event_len = u64::from_le_bytes([
298                    valid[0], valid[1], valid[2], valid[3], valid[4], valid[5], valid[6], valid[7],
299                ]);
300                let required_bytes = (event_len + 8) as usize;
301                if valid.len() >= required_bytes {
302                    let bytes = self.buf.extract(required_bytes);
303                    let event = ::bincode::deserialize(&bytes[8..]).expect("Event decode failed");
304                    Some(Cow::Owned(event))
305                } else {
306                    None
307                }
308            } else {
309                None
310            }
311        }
312    }
313}
timely/dataflow/operators/core/capture/event.rs

timely/dataflow/operators/core/capture/
event.rs
