timely/dataflow/operators/core/
enterleave.rs

1//! Extension traits to move a `Stream` between an outer `Scope` and inner `Scope`.
2//!
3//! Each `Stream` indicates its containing `Scope` as part of its type signature. To create a new
4//! stream with the same contents in another scope, one must explicit use the methods `enter` and
5//! `leave`, to clearly indicate the transition to the timely dataflow progress tracking logic.
6//!
7//! # Examples
8//! ```
9//! use timely::dataflow::scope::Scope;
10//! use timely::dataflow::operators::{Enter, Leave, ToStream, Inspect};
11//!
12//! timely::example(|outer| {
13//!     let stream = (0..9).to_stream(outer).container::<Vec<_>>();
14//!     let output = outer.region(|inner| {
15//!         stream.enter(inner)
16//!               .inspect(|x| println!("in nested scope: {:?}", x))
17//!               .leave(outer)
18//!     });
19//! });
20//! ```
21
22use std::marker::PhantomData;
23use std::rc::Rc;
24
25use crate::logging::{TimelyLogger, MessagesEvent};
26use crate::progress::Timestamp;
27use crate::progress::timestamp::Refines;
28use crate::progress::{Source, Target};
29use crate::{Accountable, Container};
30use crate::communication::Push;
31use crate::dataflow::channels::pushers::{Counter, Tee};
32use crate::dataflow::channels::Message;
33use crate::dataflow::{Stream, Scope};
34
35/// Extension trait to move a `Stream` into a child of its current `Scope`.
36pub trait Enter<'outer, TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, C> {
37    /// Moves the `Stream` argument into a child of its current `Scope`.
38    ///
39    /// The destination scope must be a child of the stream's scope.
40    /// The method checks this property at runtime, and will panic if not respected.
41    ///
42    /// # Examples
43    /// ```
44    /// use timely::dataflow::operators::{Enter, Leave, ToStream};
45    ///
46    /// timely::example(|outer| {
47    ///     let stream = (0..9).to_stream(outer).container::<Vec<_>>();
48    ///     let output = outer.region(|inner| {
49    ///         stream.enter(inner).leave(outer)
50    ///     });
51    /// });
52    /// ```
53    fn enter<'inner>(self, inner: Scope<'inner, TInner>) -> Stream<'inner, TInner, C>;
54}
55
56impl<'outer, TOuter, TInner, C> Enter<'outer, TOuter, TInner, C> for Stream<'outer, TOuter, C>
57where
58    TOuter: Timestamp,
59    TInner: Timestamp + Refines<TOuter>,
60    C: Container,
61{
62    fn enter<'inner>(self, inner: Scope<'inner, TInner>) -> Stream<'inner, TInner, C> {
63
64        // Validate that `inner` is a child of `self`'s scope.
65        let inner_addr = inner.addr();
66        let outer_addr = self.scope().addr();
67        assert!(
68            inner_addr.len() == outer_addr.len() + 1
69                && inner_addr[..outer_addr.len()] == outer_addr[..],
70            "Enter::enter: `inner` is not a child of the stream's scope \
71            (inner addr: {:?}, outer addr: {:?})",
72            inner_addr,
73            outer_addr,
74        );
75
76        let (targets, registrar) = Tee::<TInner, C>::new();
77        let ingress = IngressNub {
78            targets: Counter::new(targets),
79            phantom: PhantomData,
80            activator: inner.activator_for(inner_addr),
81            active: false,
82        };
83        let produced = Rc::clone(ingress.targets.produced());
84        let input = inner.subgraph.borrow_mut().new_input(produced);
85        let channel_id = inner.worker().new_identifier();
86
87        if let Some(logger) = inner.worker().logging() {
88            let pusher = LogPusher::new(ingress, channel_id, inner.index(), logger);
89            self.connect_to(input, pusher, channel_id);
90        } else {
91            self.connect_to(input, ingress, channel_id);
92        }
93
94        Stream::new(Source::new(0, input.port), registrar, inner)
95    }
96}
97
98/// Extension trait to move a `Stream` to the parent of its current `Scope`.
99pub trait Leave<'inner, TInner: Timestamp, C> {
100    /// Moves a `Stream` to the parent of its current `Scope`.
101    ///
102    /// The parent scope must be supplied as an argument.
103    ///
104    /// The destination scope must be the parent of the stream's scope.
105    /// The method checks this property at runtime, and will panic if not respected.
106    ///
107    /// # Examples
108    /// ```
109    /// use timely::dataflow::operators::{Enter, Leave, ToStream};
110    ///
111    /// timely::example(|outer| {
112    ///     let stream = (0..9).to_stream(outer).container::<Vec<_>>();
113    ///     let output = outer.region(|inner| {
114    ///         stream.enter(inner).leave(outer)
115    ///     });
116    /// });
117    /// ```
118    fn leave<'outer, TOuter: Timestamp>(self, outer: Scope<'outer, TOuter>) -> Stream<'outer, TOuter, C> where TInner: Refines<TOuter>;
119}
120
121impl<'inner, TInner, C> Leave<'inner, TInner, C> for Stream<'inner, TInner, C>
122where
123    TInner: Timestamp,
124    C: Container,
125{
126    fn leave<'outer, TOuter: Timestamp>(self, outer: Scope<'outer, TOuter>) -> Stream<'outer, TOuter, C> where TInner: Refines<TOuter> {
127
128        let scope = self.scope();
129
130        // Validate that `self`'s scope is a child of `outer`.
131        let inner_addr = scope.addr();
132        let outer_addr = outer.addr();
133        assert!(
134            inner_addr.len() == outer_addr.len() + 1
135                && inner_addr[..outer_addr.len()] == outer_addr[..],
136            "Leave::leave: `outer` is not the parent of the stream's scope \
137            (stream addr: {:?}, outer addr: {:?})",
138            inner_addr,
139            outer_addr,
140        );
141
142        let output = scope.subgraph.borrow_mut().new_output();
143        let target = Target::new(0, output.port);
144        let (targets, registrar) = Tee::<TOuter, C>::new();
145        let egress = EgressNub { targets, phantom: PhantomData };
146        let channel_id = scope.worker().new_identifier();
147
148        if let Some(logger) = scope.worker().logging() {
149            let pusher = LogPusher::new(egress, channel_id, scope.index(), logger);
150            self.connect_to(target, pusher, channel_id);
151        } else {
152            self.connect_to(target, egress, channel_id);
153        }
154
155        Stream::new(output, registrar, outer)
156    }
157}
158
159
160struct IngressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container> {
161    targets: Counter<TInner, Tee<TInner, TContainer>>,
162    phantom: PhantomData<TOuter>,
163    activator: crate::scheduling::Activator,
164    active: bool,
165}
166
167impl<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer: Container> Push<Message<TOuter, TContainer>> for IngressNub<TOuter, TInner, TContainer> {
168    fn push(&mut self, element: &mut Option<Message<TOuter, TContainer>>) {
169        if let Some(outer_message) = element {
170            let data = ::std::mem::take(&mut outer_message.data);
171            let stamp = outer_message.stamp.map_pointwise(|time| TInner::to_inner(time.clone()));
172            let mut inner_message = Some(Message::new(stamp, data));
173            self.targets.push(&mut inner_message);
174            if let Some(inner_message) = inner_message {
175                outer_message.data = inner_message.data;
176            }
177            self.active = true;
178        }
179        else {
180            if self.active {
181                self.activator.activate();
182                self.active = false;
183            }
184            self.targets.done();
185        }
186    }
187}
188
189
190struct EgressNub<TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, TContainer> {
191    targets: Tee<TOuter, TContainer>,
192    phantom: PhantomData<TInner>,
193}
194
195impl<TOuter, TInner, TContainer: Container> Push<Message<TInner, TContainer>> for EgressNub<TOuter, TInner, TContainer>
196where TOuter: Timestamp, TInner: Timestamp+Refines<TOuter>, {
197    fn push(&mut self, message: &mut Option<Message<TInner, TContainer>>) {
198        if let Some(inner_message) = message {
199            let data = ::std::mem::take(&mut inner_message.data);
200            let stamp = inner_message.stamp.map_pointwise(|time| time.clone().to_outer());
201            let mut outer_message = Some(Message::new(stamp, data));
202            self.targets.push(&mut outer_message);
203            if let Some(outer_message) = outer_message {
204                inner_message.data = outer_message.data;
205            }
206        }
207        else { self.targets.done(); }
208    }
209}
210
211/// A pusher that logs messages passing through it.
212///
213/// This type performs the same function as the `LogPusher` and `LogPuller` types in
214/// `timely::dataflow::channels::pact`. We need a special implementation for `enter`/`leave`
215/// channels because those don't have a puller connected. Thus, this pusher needs to log both the
216/// send and the receive `MessageEvent`.
217struct LogPusher<P> {
218    pusher: P,
219    channel: usize,
220    counter: usize,
221    index: usize,
222    logger: TimelyLogger,
223}
224
225impl<P> LogPusher<P> {
226    fn new(pusher: P, channel: usize, index: usize, logger: TimelyLogger) -> Self {
227        Self {
228            pusher,
229            channel,
230            counter: 0,
231            index,
232            logger,
233        }
234    }
235}
236
237impl<T, C, P> Push<Message<T, C>> for LogPusher<P>
238where
239    C: Accountable,
240    P: Push<Message<T, C>>,
241{
242    fn push(&mut self, element: &mut Option<Message<T, C>>) {
243        if let Some(bundle) = element {
244            let send_event = MessagesEvent {
245                is_send: true,
246                channel: self.channel,
247                source: self.index,
248                target: self.index,
249                seq_no: self.counter,
250                record_count: bundle.data.record_count(),
251            };
252            let recv_event = MessagesEvent {
253                is_send: false,
254                ..send_event
255            };
256
257            self.logger.log(send_event);
258            self.logger.log(recv_event);
259            self.counter += 1;
260        }
261
262        self.pusher.push(element);
263    }
264}
265
266#[cfg(test)]
267mod test {
268    /// Test that nested scopes with pass-through edges (no operators) correctly communicate progress.
269    ///
270    /// This is for issue: https://github.com/TimelyDataflow/timely-dataflow/issues/377
271    #[test]
272    fn test_nested() {
273
274        use crate::dataflow::{InputHandle, ProbeHandle};
275        use crate::dataflow::operators::{vec::Input, Inspect, Probe};
276
277        use crate::dataflow::operators::{Enter, Leave};
278
279        // initializes and runs a timely dataflow.
280        crate::execute(crate::Config::process(3), |worker| {
281
282            let index = worker.index();
283            let mut input = InputHandle::new();
284            let probe = ProbeHandle::new();
285
286            // create a new input, exchange data, and inspect its output
287            worker.dataflow(|scope| {
288                let data = scope.input_from(&mut input);
289
290                scope.region(|inner| {
291
292                    let data = data.enter(inner);
293                    inner.region(|inner2| data.enter(inner2).leave(inner)).leave(scope)
294                })
295                    .inspect(move |x| println!("worker {}:\thello {}", index, x))
296                    .probe_with(&probe);
297            });
298
299            // introduce data and watch!
300            input.advance_to(0);
301            for round in 0..10 {
302                if index == 0 {
303                    input.send(round);
304                }
305                input.advance_to(round + 1);
306                while probe.less_than(input.time()) {
307                    worker.step_or_park(None);
308                }
309            }
310        }).unwrap();
311    }
312
313}
timely/dataflow/operators/core/enterleave.rs

timely/dataflow/operators/core/
enterleave.rs
