Reorganize types and traits to deprioritize Vec streams

Author: frankmcsherry

README.md

@@ -23,6 +23,7 @@ use timely::dataflow::operators::*;
 fn main() {
     timely::example(|scope| {
         (0..10).to_stream(scope)
+               .container::<Vec<_>>()
                .inspect(|x| println!("seen: {:?}", x));
     });
 }
@@ -66,6 +67,7 @@ fn main() {
         // create a new input, exchange data, and inspect its output
         worker.dataflow(|scope| {
             scope.input_from(&mut input)
+                 .container::<Vec<_>>()
                  .exchange(|x| *x)
                  .inspect(move |x| println!("worker {}:\thello {}", index, x))
                  .probe_with(&mut probe);

mdbook/src/chapter_0/chapter_0_0.md

@@ -5,7 +5,7 @@ Let's start with what may be the simplest non-trivial timely dataflow program.
 ```rust
 # extern crate timely;
 
-use timely::dataflow::operators::{ToStream, Inspect};
+use timely::dataflow::operators::{vec::ToStream, Inspect};
 
 timely::example(|scope| {
     (0..10).to_stream(scope)

mdbook/src/chapter_0/chapter_0_1.md

@@ -19,6 +19,7 @@ timely::execute_from_args(std::env::args(), |worker| {
     // create a new input, exchange data, and inspect its output
     let probe = worker.dataflow(|scope|
         scope.input_from(&mut input)
+             .container::<Vec<_>>()
              .exchange(|x| *x)
              .inspect(move |x| println!("worker {}:\thello {}", index, x))
              .probe()

mdbook/src/chapter_1/chapter_1_1.md

@@ -25,6 +25,7 @@ fn main() {
         // create a new input, exchange data, and inspect its output
         let probe = worker.dataflow(|scope|
             scope.input_from(&mut input)
+                 .container::<Vec<_>>()
                  .exchange(|x| *x)
                  .inspect(move |x| println!("worker {}:\thello {}", index, x))
                  .probe()

mdbook/src/chapter_1/chapter_1_3.md

@@ -20,6 +20,7 @@ fn main() {
         // create a new input, exchange data, and inspect its output
         let probe = worker.dataflow(|scope|
             scope.input_from(&mut input)
+                 .container::<Vec<_>>()
                  .exchange(|x| *x)
                  .inspect(move |x| println!("worker {}:\thello {}", index, x))
                  .probe()

mdbook/src/chapter_2/chapter_2.md

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

mdbook/src/chapter_2/chapter_2_1.md

@@ -16,14 +16,13 @@ fn main() {
     // initializes and runs a timely dataflow.
     timely::execute_from_args(std::env::args(), |worker| {
 
-        let mut input = InputHandle::<(), String>::new();
+        let mut input = InputHandle::new();
 
         // define a new dataflow
-        worker.dataflow(|scope| {
-
-            let stream1 = input.to_stream(scope);
-            let stream2 = (0 .. 10).to_stream(scope);
+        worker.dataflow::<u32,_,_>(|scope| {
 
+            let stream1 = input.to_stream(scope).container::<Vec<i32>>();
+            let stream2 = (0 .. 10).to_stream(scope).container::<Vec<_>>();
         });
 
     }).unwrap();

mdbook/src/chapter_2/chapter_2_2.md

@@ -14,6 +14,7 @@ fn main() {
         worker.dataflow::<(),_,_>(|scope| {
             (0 .. 10)
                 .to_stream(scope)
+                .container::<Vec<_>>()
                 .inspect(|x| println!("hello: {}", x));
         });
     }).unwrap();
@@ -36,6 +37,7 @@ fn main() {
         worker.dataflow::<(),_,_>(|scope| {
             (0 .. 10)
                 .to_stream(scope)
+                .container::<Vec<_>>()
                 .inspect_batch(|t, xs| println!("hello: {:?} @ {:?}", xs, t));
         });
     }).unwrap();
@@ -58,8 +60,8 @@ use timely::dataflow::operators::capture::Extract;
 
 fn main() {
     let (data1, data2) = timely::example(|scope| {
-        let data1 = (0..3).to_stream(scope).capture();
-        let data2 = vec![0,1,2].to_stream(scope).capture();
+        let data1 = (0..3).to_stream(scope).container::<Vec<_>>().capture();
+        let data2 = vec![0,1,2].to_stream(scope).container::<Vec<_>>().capture();
         (data1, data2)
     });
 

mdbook/src/chapter_2/chapter_2_3.md

@@ -13,7 +13,7 @@ The following program should print out the numbers one through ten.
 ```rust
 extern crate timely;
 
-use timely::dataflow::operators::{ToStream, Inspect, Map};
+use timely::dataflow::operators::{ToStream, Inspect, vec::Map};
 
 fn main() {
     timely::execute_from_args(std::env::args(), |worker| {
@@ -32,7 +32,7 @@ The closure `map` takes *owned* data as input, which means you are able to mutat
 ```rust
 extern crate timely;
 
-use timely::dataflow::operators::{ToStream, Inspect, Map};
+use timely::dataflow::operators::{ToStream, Inspect, vec::Map};
 
 fn main() {
     timely::execute_from_args(std::env::args(), |worker| {
@@ -56,7 +56,7 @@ For example, the `map_in_place` method takes a closure which receives a mutable
 ```rust
 extern crate timely;
 
-use timely::dataflow::operators::{ToStream, Inspect, Map};
+use timely::dataflow::operators::{ToStream, Inspect, vec::Map};
 
 fn main() {
     timely::execute_from_args(std::env::args(), |worker| {
@@ -76,7 +76,7 @@ Alternately, the `flat_map` method takes input data and allows your closure to t
 ```rust
 extern crate timely;
 
-use timely::dataflow::operators::{ToStream, Inspect, Map};
+use timely::dataflow::operators::{ToStream, Inspect, vec::Map};
 
 fn main() {
     timely::execute_from_args(std::env::args(), |worker| {
@@ -97,7 +97,7 @@ Another fundamental operation is *filtering*, in which a predicate dictates a su
 ```rust
 extern crate timely;
 
-use timely::dataflow::operators::{ToStream, Inspect, Filter};
+use timely::dataflow::operators::{ToStream, Inspect, vec::Filter};
 
 fn main() {
     timely::execute_from_args(std::env::args(), |worker| {
@@ -122,7 +122,7 @@ The `partition` operator takes two arguments, a number of resulting streams to p
 ```rust
 extern crate timely;
 
-use timely::dataflow::operators::{ToStream, Partition, Inspect};
+use timely::dataflow::operators::{ToStream, Inspect, vec::Partition};
 
 fn main() {
     timely::example(|scope| {
@@ -143,7 +143,7 @@ In the other direction, `concat` takes two streams and produces one output strea
 ```rust
 extern crate timely;
 
-use timely::dataflow::operators::{ToStream, Partition, Concat, Inspect};
+use timely::dataflow::operators::{ToStream, Concat, Inspect, vec::Partition};
 
 fn main() {
     timely::example(|scope| {
@@ -163,7 +163,7 @@ There is also a `concatenate` method defined for scopes which collects all strea
 ```rust
 extern crate timely;
 
-use timely::dataflow::operators::{ToStream, Partition, Concatenate, Inspect};
+use timely::dataflow::operators::{ToStream, Concatenate, Inspect, vec::Partition};
 
 fn main() {
     timely::example(|scope| {

mdbook/src/chapter_2/chapter_2_4.md

@@ -17,6 +17,7 @@ fn main() {
     timely::example(|scope| {
         (0u64..10)
             .to_stream(scope)
+            .container::<Vec<_>>()
             .unary(Pipeline, "increment", |capability, info| {
 
                 move |input, output| {
@@ -131,6 +132,7 @@ fn main() {
     timely::example(|scope| {
         (0u64..10)
             .to_stream(scope)
+            .container::<Vec<_>>()
             .unary(Pipeline, "increment", |capability, info| {
 
                 let mut maximum = 0;    // define this here; use in the closure
@@ -179,8 +181,8 @@ use timely::dataflow::channels::pact::Pipeline;
 fn main() {
     timely::example(|scope| {
 
-        let in1 = (0 .. 10).to_stream(scope);
-        let in2 = (0 .. 10).to_stream(scope);
+        let in1 = (0 .. 10).to_stream(scope).container::<Vec<_>>();
+        let in2 = (0 .. 10).to_stream(scope).container::<Vec<_>>();
 
         in1.binary_frontier(in2, Pipeline, Pipeline, "concat_buffer", |capability, info| {
 
@@ -230,8 +232,8 @@ use timely::dataflow::channels::pact::Pipeline;
 fn main() {
     timely::example(|scope| {
 
-        let in1 = (0 .. 10).to_stream(scope);
-        let in2 = (0 .. 10).to_stream(scope);
+        let in1 = (0 .. 10).to_stream(scope).container::<Vec<_>>();
+        let in2 = (0 .. 10).to_stream(scope).container::<Vec<_>>();
 
         in1.binary_frontier(in2, Pipeline, Pipeline, "concat_buffer", |capability, info| {