Merge pull request #115 from frankmcsherry/logging

Logging

Author: frankmcsherry

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 
 name = "differential-dataflow"
-version = "0.6.0"
+version = "0.7.0"
 authors = ["Frank McSherry <fmcsherry@me.com>"]
 
 description = "An incremental data-parallel dataflow platform"
@@ -24,7 +24,7 @@ graph_map = { git = "https://github.com/frankmcsherry/graph-map.git" }
 abomonation = { git = "https://github.com/frankmcsherry/abomonation" }
 abomonation_derive = "0.3"
 timely_sort="0.1.6"
-timely = { git = "https://github.com/frankmcsherry/timely-dataflow" }
+timely = { git = "https://github.com/frankmcsherry/timely-dataflow", tag = "v0.7.0" }
 fnv="1.0.2"
 
 [profile.release]

examples/arrange.rs

@@ -8,21 +8,12 @@ use rand::{Rng, SeedableRng, StdRng};
 
 use differential_dataflow::input::Input;
 use differential_dataflow::AsCollection;
-use differential_dataflow::operators::arrange::{Arrange, ArrangeByKey};
+use differential_dataflow::operators::arrange::ArrangeByKey;
 use differential_dataflow::operators::group::Group;
 use differential_dataflow::operators::join::JoinCore;
 use differential_dataflow::operators::Iterate;
 use differential_dataflow::operators::Consolidate;
 
-use differential_dataflow::trace::Trace;
-use differential_dataflow::trace::implementations::ord::OrdValSpineAbom;
-
-// use differential_dataflow::trace::implementations::ord::OrdValSpine;
-// use differential_dataflow::trace::{Cursor, Trace};
-// use differential_dataflow::trace::Batch;
-// use differential_dataflow::hashable::OrdWrapper;
-// use differential_dataflow::trace::TraceReader;
-
 fn main() {
 
     let nodes: u32 = std::env::args().nth(1).unwrap().parse().unwrap();
@@ -32,7 +23,7 @@ fn main() {
     let inspect: bool = std::env::args().nth(5).unwrap() == "inspect";
 
 
-    // define a new timely dataflow computation. 
+    // define a new timely dataflow computation.
     timely::execute_from_args(std::env::args().skip(6), move |worker| {
 
         let timer = ::std::time::Instant::now();
@@ -99,8 +90,8 @@ fn main() {
             })
             .probe_with(&mut probe)
             .as_collection()
-            // .arrange_by_key()
-            .arrange(OrdValSpineAbom::new())
+            .arrange_by_key()
+            // .arrange::<OrdValSpineAbom>()
             .trace
         });
 

examples/bfs.rs

@@ -11,6 +11,7 @@ use differential_dataflow::input::Input;
 use differential_dataflow::Collection;
 use differential_dataflow::operators::*;
 use differential_dataflow::lattice::Lattice;
+use differential_dataflow::logging::DifferentialEvent;
 
 type Node = u32;
 type Edge = (Node, Node);
@@ -26,6 +27,22 @@ fn main() {
     // define a new computational scope, in which to run BFS
     timely::execute_from_args(std::env::args(), move |worker| {
 
+        if let Ok(addr) = ::std::env::var("DIFFERENTIAL_LOG_ADDR") {
+
+            eprintln!("enabled DIFFERENTIAL logging to {}", addr);
+
+            if let Ok(stream) = ::std::net::TcpStream::connect(&addr) {
+                let writer = ::timely::dataflow::operators::capture::EventWriter::new(stream);
+                let mut logger = ::timely::logging::BatchLogger::new(writer);
+                worker.log_register().insert::<DifferentialEvent,_>("differential/arrange", move |time, data|
+                    logger.publish_batch(time, data)
+                );
+            }
+            else {
+                panic!("Could not connect to differential log address: {:?}", addr);
+            }
+        }
+
         let timer = ::std::time::Instant::now();
 
         // define BFS dataflow; return handles to roots and edges inputs

examples/bijkstra.rs

@@ -26,7 +26,7 @@ fn main() {
 
     // define a new computational scope, in which to run BFS
     timely::execute_from_args(std::env::args().skip(6), move |worker| {
-        
+
         let timer = ::std::time::Instant::now();
 
         // define BFS dataflow; return handles to roots and edges inputs
@@ -80,7 +80,7 @@ fn main() {
                     graph.insert((rng1.gen_range(0, nodes), rng1.gen_range(0, nodes)));
                     graph.remove((rng2.gen_range(0, nodes), rng2.gen_range(0, nodes)));
                 }
-                graph.advance_to(3 + round * batch + element);                
+                graph.advance_to(3 + round * batch + element);
             }
             graph.flush();
 
@@ -100,15 +100,15 @@ fn main() {
 fn bidijkstra<G: Scope>(edges: &Collection<G, Edge>, goals: &Collection<G, (Node, Node)>) -> Collection<G, ((Node, Node), u32)>
 where G::Timestamp: Lattice+Ord {
 
-    edges.scope().scoped(|inner| {
+    edges.scope().scoped::<u64,_,_>(|inner| {
 
-        // Our plan is to start evolving distances from both sources and destinations. 
+        // Our plan is to start evolving distances from both sources and destinations.
         // The evolution from a source or destination should continue as long as there
         // is a corresponding destination or source that has not yet been reached.
 
         // forward and reverse (node, (root, dist))
-        let forward = Variable::from(goals.map(|(x,_)| (x,(x,0))).enter(inner));
-        let reverse = Variable::from(goals.map(|(_,y)| (y,(y,0))).enter(inner));
+        let forward = Variable::new_from(goals.map(|(x,_)| (x,(x,0))).enter(inner), u64::max_value(), 1);
+        let reverse = Variable::new_from(goals.map(|(_,y)| (y,(y,0))).enter(inner), u64::max_value(), 1);
 
         let goals = goals.enter(inner);
         let edges = edges.enter(inner);
@@ -118,7 +118,7 @@ where G::Timestamp: Lattice+Ord {
         //   done(src, dst) := forward(src, med), reverse(dst, med), goal(src, dst).
         //
         // This is a cyclic join, which should scare us a bunch.
-        let reached = 
+        let reached =
         forward
             .join_map(&reverse, |_, &(src,d1), &(dst,d2)| ((src, dst), d1 + d2))
             .group(|_key, s, t| t.push((*s[0].0, 1)));
@@ -133,7 +133,7 @@ where G::Timestamp: Lattice+Ord {
 
         // Let's expand out forward queries that are active.
         let forward_active = active.map(|(x,_y)| x).distinct();
-        let forward_next = 
+        let forward_next =
         forward
             .map(|(med, (src, dist))| (src, (med, dist)))
             .semijoin(&forward_active)
@@ -148,7 +148,7 @@ where G::Timestamp: Lattice+Ord {
 
         // Let's expand out reverse queries that are active.
         let reverse_active = active.map(|(_x,y)| y).distinct();
-        let reverse_next = 
+        let reverse_next =
         reverse
             .map(|(med, (rev, dist))| (rev, (med, dist)))
             .semijoin(&reverse_active)

examples/graspan.rs

@@ -10,10 +10,11 @@ use indexmap::IndexMap;
 use timely::communication::Allocate;
 
 use timely::dataflow::Scope;
-use timely::dataflow::scopes::{ScopeParent, Root, Child};
+use timely::dataflow::scopes::{ScopeParent, Child};
 use timely::progress::timestamp::RootTimestamp;
 use timely::progress::nested::product::Product;
 use timely::progress::Timestamp;
+use timely::worker::Worker;
 
 use differential_dataflow::Collection;
 use differential_dataflow::lattice::Lattice;
@@ -88,7 +89,7 @@ type Arrange<G, K, V, R> = Arranged<G, K, V, R, TraceValHandle<K, V, <G as Scope
 /// An edge variable provides arranged representations of its contents, even before they are
 /// completely defined, in support of recursively defined productions.
 pub struct EdgeVariable<'a, G: Scope> where G::Timestamp : Lattice {
-    variable: Variable<'a, G, Edge, isize>,
+    variable: Variable<'a, G, Edge, u64, isize>,
     current: Collection<Child<'a, G, u64>, Edge, isize>,
     forward: Option<Arrange<Child<'a, G, u64>, Node, Node, isize>>,
     reverse: Option<Arrange<Child<'a, G, u64>, Node, Node, isize>>,
@@ -97,7 +98,7 @@ pub struct EdgeVariable<'a, G: Scope> where G::Timestamp : Lattice {
 impl<'a, G: Scope> EdgeVariable<'a, G> where G::Timestamp : Lattice {
     /// Creates a new variable initialized with `source`.
     pub fn from(source: &Collection<Child<'a, G, u64>, Edge>) -> Self {
-        let variable = Variable::from(source.filter(|_| false));
+        let variable = Variable::new_from(source.filter(|_| false), u64::max_value(), 1);
         EdgeVariable {
             variable: variable,
             current: source.clone(),
@@ -157,7 +158,7 @@ impl Query {
     }
 
     /// Creates a dataflow implementing the query, and returns input and trace handles.
-    pub fn render_in<'a, A, T>(&self, scope: &mut Child<'a, Root<A>, T>) -> IndexMap<String, RelationHandles<T>>
+    pub fn render_in<'a, A, T>(&self, scope: &mut Child<'a, Worker<A>, T>) -> IndexMap<String, RelationHandles<T>>
     where A: Allocate, T: Timestamp+Lattice {
 
         // Create new input (handle, stream) pairs