timely/dataflow/operators/core/
feedback.rs

1//! Create cycles in a timely dataflow graph.
2
3use crate::Container;
4use crate::dataflow::channels::pact::Pipeline;
5use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
6use crate::dataflow::scope::Iterative;
7use crate::dataflow::{Stream, Scope};
8use crate::order::Product;
9use crate::progress::frontier::Antichain;
10use crate::progress::{Timestamp, PathSummary};
11
12/// Creates a `Stream` and a `Handle` to later bind the source of that `Stream`.
13pub trait Feedback<'scope, T: Timestamp> {
14
15    /// Creates a [Stream] and a [Handle] to later bind the source of that `Stream`.
16    ///
17    /// The resulting `Stream` will have its data defined by a future call to `connect_loop` with
18    /// its `Handle` passed as an argument. Messages passed through the stream will have the
19    /// elements of their stamps advanced by `summary`. Container contents are not inspected or
20    /// changed; containers whose contents carry times must advance them before the loop.
21    ///
22    /// A message is dropped if its advanced stamp is empty, which includes messages that
23    /// arrive with empty stamps.
24    ///
25    /// # Examples
26    /// ```
27    /// use timely::dataflow::Scope;
28    /// use timely::dataflow::operators::{Feedback, ConnectLoop, ToStream, Concat, Inspect};
29    /// use timely::dataflow::operators::vec::{Map, Filter};
30    ///
31    /// timely::example(|scope| {
32    ///     // circulate 0..10, incrementing each, until each reaches 100.
33    ///     let (handle, cycle) = scope.feedback(1);
34    ///     (0..10).to_stream(scope)
35    ///            .container::<Vec<_>>()
36    ///            .concat(cycle)
37    ///            .inspect(|x| println!("seen: {:?}", x))
38    ///            .map(|x| x + 1)
39    ///            .filter(|x| *x < 100)
40    ///            .connect_loop(handle);
41    /// });
42    /// ```
43    fn feedback<C: Container>(&self, summary: <T as Timestamp>::Summary) -> (Handle<'scope, T, C>, Stream<'scope, T, C>);
44}
45
46/// Creates a `Stream` and a `Handle` to later bind the source of that `Stream`.
47pub trait LoopVariable<'scope, TOuter: Timestamp, TInner: Timestamp> {
48    /// Creates a `Stream` and a `Handle` to later bind the source of that `Stream`.
49    ///
50    /// The resulting `Stream` will have its data defined by a future call to `connect_loop` with
51    /// its `Handle` passed as an argument. Messages passed through the stream will have the
52    /// elements of their stamps advanced by `summary`. Container contents are not inspected or
53    /// changed; containers whose contents carry times must advance them before the loop.
54    ///
55    /// A message is dropped if its advanced stamp is empty, which includes messages that
56    /// arrive with empty stamps.
57    ///
58    /// # Examples
59    /// ```
60    /// use timely::dataflow::Scope;
61    /// use timely::dataflow::operators::{LoopVariable, ConnectLoop, ToStream, Concat, Inspect};
62    /// use timely::dataflow::operators::vec::{Map, Filter};
63    ///
64    /// timely::example(|scope| {
65    ///     // circulate 0..10, incrementing each, until each reaches 100.
66    ///     scope.iterative::<usize,_,_>(|inner| {
67    ///         let (handle, cycle) = inner.loop_variable(1);
68    ///         (0..10).to_stream(inner)
69    ///                .container::<Vec<_>>()
70    ///                .concat(cycle)
71    ///                .inspect(|x| println!("seen: {:?}", x))
72    ///                .map(|x| x + 1)
73    ///                .filter(|x| *x < 100)
74    ///                .connect_loop(handle);
75    ///     });
76    /// });
77    /// ```
78    fn loop_variable<C: Container>(&self, summary: TInner::Summary) -> (Handle<'scope, Product<TOuter, TInner>, C>, Stream<'scope, Product<TOuter, TInner>, C>);
79}
80
81impl<'scope, T: Timestamp> Feedback<'scope, T> for Scope<'scope, T> {
82
83    fn feedback<C: Container>(&self, summary: <T as Timestamp>::Summary) -> (Handle<'scope, T, C>, Stream<'scope, T, C>) {
84
85        let mut builder = OperatorBuilder::new("Feedback".to_owned(), *self);
86        let (output, stream) = builder.new_output();
87
88        (Handle { builder, summary, output }, stream)
89    }
90}
91
92impl<'scope, TOuter: Timestamp, TInner: Timestamp> LoopVariable<'scope, TOuter, TInner> for Iterative<'scope, TOuter, TInner> {
93    fn loop_variable<C: Container>(&self, summary: TInner::Summary) -> (Handle<'scope, Product<TOuter, TInner>, C>, Stream<'scope, Product<TOuter, TInner>, C>) {
94        self.feedback(Product::new(Default::default(), summary))
95    }
96}
97
98/// Connect a `Stream` to the input of a loop variable.
99pub trait ConnectLoop<'scope, T: Timestamp, C: Container> {
100    /// Connect a `Stream` to be the input of a loop variable.
101    ///
102    /// # Examples
103    /// ```
104    /// use timely::dataflow::Scope;
105    /// use timely::dataflow::operators::{Feedback, ConnectLoop, ToStream, Concat, Inspect};
106    /// use timely::dataflow::operators::vec::{Map, Filter};
107    ///
108    /// timely::example(|scope| {
109    ///     // circulate 0..10, incrementing each, until each reaches 100.
110    ///     let (handle, cycle) = scope.feedback(1);
111    ///     (0..10).to_stream(scope)
112    ///            .container::<Vec<_>>()
113    ///            .concat(cycle)
114    ///            .inspect(|x| println!("seen: {:?}", x))
115    ///            .map(|x| x + 1)
116    ///            .filter(|x| *x < 100)
117    ///            .connect_loop(handle);
118    /// });
119    /// ```
120    fn connect_loop(self, handle: Handle<'scope, T, C>);
121}
122
123impl<'scope, T: Timestamp, C: Container> ConnectLoop<'scope, T, C> for Stream<'scope, T, C> {
124    fn connect_loop(self, handle: Handle<'scope, T, C>) {
125
126        let mut builder = handle.builder;
127        let summary = handle.summary;
128        let mut output = handle.output;
129
130        let mut input = builder.new_input_connection(self, Pipeline, [(0, Antichain::from_elem(summary.clone()))]);
131        builder.set_notify_for(0, crate::progress::operate::FrontierInterest::Never);
132
133        builder.build(move |_capability| move |_frontier| {
134            let mut output = output.activate();
135            input.for_each(|cap, data| {
136                // Advance each stamp element by the summary, discarding elements that
137                // cannot traverse the feedback edge and restoring minimality. The edge
138                // maps capabilities and does not inspect contents: advancing any times
139                // the contents carry, and discarding those that cannot advance, is the
140                // container's responsibility. A message whose advanced stamp is empty
141                // is dropped.
142                let new_caps = cap.stamp()
143                    .map_into(|time| summary.results_in(time))
144                    .iter()
145                    .map(|time| cap.delayed(time, output.output_index()))
146                    .collect::<crate::dataflow::operators::CapabilitySet<_>>();
147                if !new_caps.is_empty() {
148                    output.give(&new_caps, data);
149                }
150            });
151        });
152    }
153}
154
155/// A handle used to bind the source of a loop variable.
156#[derive(Debug)]
157pub struct Handle<'scope, T: Timestamp, C: Container> {
158    builder: OperatorBuilder<'scope, T>,
159    summary: <T as Timestamp>::Summary,
160    output: crate::dataflow::channels::pushers::Output<T, C>,
161}
timely/dataflow/operators/core/feedback.rs

timely/dataflow/operators/core/
feedback.rs
