Introduce FlatContainer, container function for hinting

Author: frankmcsherry

container/Cargo.toml

@@ -7,4 +7,5 @@ license = "MIT"
 
 [dependencies]
 columnation = { git = "https://github.com/frankmcsherry/columnation" }
+flatcontainer = "0.1"
 serde = { version = "1.0"}

container/src/flatcontainer.rs

@@ -0,0 +1,50 @@
+//! Present a [`FlatStack`] as a timely container.
+
+pub use flatcontainer::*;
+use crate::{buffer, Container, PushContainer, PushInto};
+
+impl<R: Region + Clone + 'static> Container for FlatStack<R> {
+    type ItemRef<'a> = R::ReadItem<'a>  where Self: 'a;
+    type Item<'a> = R::ReadItem<'a> where Self: 'a;
+
+    fn len(&self) -> usize {
+        self.len()
+    }
+
+    fn clear(&mut self) {
+        self.clear()
+    }
+
+    type Iter<'a> = <&'a Self as IntoIterator>::IntoIter;
+
+    fn iter<'a>(&'a self) -> Self::Iter<'a> {
+        IntoIterator::into_iter(self)
+    }
+
+    type DrainIter<'a> = Self::Iter<'a>;
+
+    fn drain<'a>(&'a mut self) -> Self::DrainIter<'a> {
+        IntoIterator::into_iter(&*self)
+    }
+}
+
+impl<R: Region + Clone + 'static> PushContainer for FlatStack<R> {
+    fn capacity(&self) -> usize {
+        self.capacity()
+    }
+
+    fn preferred_capacity() -> usize {
+        buffer::default_capacity::<R::Index>()
+    }
+
+    fn reserve(&mut self, additional: usize) {
+        self.reserve(additional);
+    }
+}
+
+impl<R: Region + Clone + 'static, T: CopyOnto<R>> PushInto<FlatStack<R>> for T {
+    #[inline]
+    fn push_into(self, target: &mut FlatStack<R>) {
+        target.copy(self);
+    }
+}

container/src/lib.rs

@@ -3,6 +3,7 @@
 #![forbid(missing_docs)]
 
 pub mod columnation;
+pub mod flatcontainer;
 
 /// A container transferring data through dataflow edges
 ///

timely/examples/distinct.rs

@@ -35,6 +35,7 @@ fn main() {
                             }
                         })
                     })
+                .container::<Vec<_>>()
                 .inspect(move |x| println!("worker {}:\tvalue {}", index, x))
                 .probe_with(&mut probe);
         });

timely/examples/flatcontainer.rs

@@ -0,0 +1,93 @@
+//! Wordcount based on flatcontainer.
+
+#[cfg(feature = "bincode")]
+use {
+    std::collections::HashMap,
+    timely::container::flatcontainer::{Containerized, FlatStack},
+    timely::dataflow::channels::pact::{ExchangeCore, Pipeline},
+    timely::dataflow::operators::core::InputHandle,
+    timely::dataflow::operators::{Inspect, Operator, Probe},
+    timely::dataflow::ProbeHandle,
+};
+
+#[cfg(feature = "bincode")]
+fn main() {
+    // initializes and runs a timely dataflow.
+    timely::execute_from_args(std::env::args(), |worker| {
+        let mut input =
+            <InputHandle<_, FlatStack<<(String, i64) as Containerized>::Region>>>::new();
+        let mut probe = ProbeHandle::new();
+
+        // create a new input, exchange data, and inspect its output
+        worker.dataflow::<usize, _, _>(|scope| {
+            input
+                .to_stream(scope)
+                .unary::<FlatStack<<(String, i64) as Containerized>::Region>, _, _, _>(
+                    Pipeline,
+                    "Split",
+                    |_cap, _info| {
+                        move |input, output| {
+                            while let Some((time, data)) = input.next() {
+                                let mut session = output.session(&time);
+                                for (text, diff) in data.iter().flat_map(|(text, diff)| {
+                                    text.split_whitespace().map(move |s| (s, diff))
+                                }) {
+                                    session.give((text, diff));
+                                }
+                            }
+                        }
+                    },
+                )
+                .unary_frontier::<FlatStack<<(String, i64) as Containerized>::Region>, _, _, _>(
+                    ExchangeCore::new(|(s, _): &(&str, _)| s.len() as u64),
+                    "WordCount",
+                    |_capability, _info| {
+                        let mut queues = HashMap::new();
+                        let mut counts = HashMap::new();
+
+                        move |input, output| {
+                            while let Some((time, data)) = input.next() {
+                                queues
+                                    .entry(time.retain())
+                                    .or_insert(Vec::new())
+                                    .push(data.take());
+                            }
+
+                            for (key, val) in queues.iter_mut() {
+                                if !input.frontier().less_equal(key.time()) {
+                                    let mut session = output.session(key);
+                                    for batch in val.drain(..) {
+                                        for (word, diff) in batch.iter() {
+                                            let entry =
+                                                counts.entry(word.to_string()).or_insert(0i64);
+                                            *entry += diff;
+                                            session.give((word, *entry));
+                                        }
+                                    }
+                                }
+                            }
+
+                            queues.retain(|_key, val| !val.is_empty());
+                        }
+                    },
+                )
+                .inspect(|x| println!("seen: {:?}", x))
+                .probe_with(&mut probe);
+        });
+
+        // introduce data and watch!
+        for round in 0..10 {
+            input.send(("flat container", 1));
+            input.advance_to(round + 1);
+            while probe.less_than(input.time()) {
+                worker.step();
+            }
+        }
+    })
+    .unwrap();
+}
+
+#[cfg(not(feature = "bincode"))]
+fn main() {
+    eprintln!("Example requires feature bincode.");
+}

timely/examples/hashjoin.rs

@@ -76,6 +76,7 @@ fn main() {
                         });
                     }
                 })
+                .container::<Vec<_>>()
                 .probe_with(&mut probe);
         });
 

timely/examples/simple.rs

@@ -5,6 +5,7 @@ use timely::dataflow::operators::*;
 fn main() {
     timely::example(|scope| {
         (0..10).to_stream(scope)
+               .container::<Vec<_>>()
                .inspect(|x| println!("seen: {:?}", x));
     });
 }

timely/examples/wordcount.rs

@@ -24,6 +24,7 @@ fn main() {
                         .map(move |word| (word.to_owned(), diff))
                         .collect::<Vec<_>>()
                  )
+                 .container::<Vec<_>>()
                  .unary_frontier(exchange, "WordCount", |_capability, _info| {
 
                     let mut queues = HashMap::new();
@@ -51,6 +52,7 @@ fn main() {
 
                         queues.retain(|_key, val| !val.is_empty());
                     }})
+                 .container::<Vec<_>>()
                  .inspect(|x| println!("seen: {:?}", x))
                  .probe_with(&mut probe);
         });

timely/src/dataflow/operators/capability.rs

@@ -426,7 +426,8 @@ impl<T: Timestamp> CapabilitySet<T> {
     ///                     output.session(a_cap).give(());
     ///                 }
     ///             }
-    ///         });
+    ///         })
+    ///         .container::<Vec<_>>();
     /// });
     /// ```
     pub fn from_elem(cap: Capability<T>) -> Self {

timely/src/dataflow/operators/core/input.rs

@@ -81,6 +81,7 @@ pub trait Input : Scope {
     ///     let mut input = Handle::new();
     ///     worker.dataflow(|scope| {
     ///         scope.input_from(&mut input)
+    ///              .container::<Vec<_>>()
     ///              .inspect(|x| println!("hello {:?}", x));
     ///     });
     ///
@@ -198,6 +199,7 @@ impl<T: Timestamp, C: Container> Handle<T, C> {
     ///     let mut input = Handle::new();
     ///     worker.dataflow(|scope| {
     ///         scope.input_from(&mut input)
+    ///              .container::<Vec<_>>()
     ///              .inspect(|x| println!("hello {:?}", x));
     ///     });
     ///
@@ -235,6 +237,7 @@ impl<T: Timestamp, C: Container> Handle<T, C> {
     ///     let mut input = Handle::new();
     ///     worker.dataflow(|scope| {
     ///         input.to_stream(scope)
+    ///              .container::<Vec<_>>()
     ///              .inspect(|x| println!("hello {:?}", x));
     ///     });
     ///
@@ -404,6 +407,7 @@ impl<T: Timestamp, C: PushContainer> Handle<T, C> {
     ///     let mut input = Handle::new();
     ///     worker.dataflow(|scope| {
     ///         scope.input_from(&mut input)
+    ///              .container::<Vec<_>>()
     ///              .inspect(|x| println!("hello {:?}", x));
     ///     });
     ///