Merge pull request #163 from TimelyDataflow/v0.9-release

V0.9 release

Author: frankmcsherry

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 
 name = "differential-dataflow"
-version = "0.8.0"
+version = "0.9.0"
 authors = ["Frank McSherry <fmcsherry@me.com>"]
 
 description = "An incremental data-parallel dataflow platform"
@@ -12,6 +12,7 @@ homepage = "https://github.com/TimelyDataflow/differential-dataflow"
 repository = "https://github.com/TimelyDataflow/differential-dataflow.git"
 keywords = ["differential", "dataflow"]
 license = "MIT"
+readme = "README.md"
 
 [dev-dependencies]
 indexmap = "1.0.1"
@@ -27,8 +28,8 @@ serde_derive = "1.0"
 abomonation = "0.7"
 abomonation_derive = "0.3"
 timely_sort="0.1.6"
-#timely = "0.8"
-timely = { git = "https://github.com/TimelyDataflow/timely-dataflow" }
+timely = "0.9"
+#timely = { git = "https://github.com/TimelyDataflow/timely-dataflow" }
 #timely = { path = "../timely-dataflow/" }
 fnv="1.0.2"
 

examples/pagerank.rs

@@ -117,14 +117,19 @@ where
              .map(|(node, _degr)| node);
 
         // Propagate surfers along links, blend in reset surfers.
-        let pushed =
+        let mut pushed =
         edges.semijoin(&to_push)
              .map(|(_node, dest)| dest)
              .concat(&reset)
-             .consolidate()
+             .consolidate();
+
+        if iters > 0 {
+            pushed =
+            pushed
              .inner
              .filter(move |(_d,t,_r)| t.inner < iters)
              .as_collection();
+        }
 
         // Bind the recursive variable, return its limit.
         ranks.set(&pushed);

interactive/src/bin/server.rs

@@ -5,9 +5,6 @@ extern crate interactive;
 use timely::synchronization::Sequencer;
 use interactive::{Manager, Command, Value};
 
-use timely::logging::TimelyEvent;
-use differential_dataflow::logging::DifferentialEvent;
-
 fn main() {
 
     let mut args = std::env::args();
@@ -19,77 +16,6 @@ fn main() {
     let command_queue = Arc::new(Mutex::new(VecDeque::<Command<Value>>::new()));
     let command_queue2 = command_queue.clone();
 
-    let guards =
-    timely::execute_from_args(args, move |worker| {
-
-        let timer = ::std::time::Instant::now();
-        let mut manager = Manager::<Value>::new();
-
-        let recv = command_queue.clone();
-
-        use std::rc::Rc;
-        use timely::dataflow::operators::capture::event::link::EventLink;
-        use timely::logging::BatchLogger;
-
-        let timely_events = Rc::new(EventLink::new());
-        let differential_events = Rc::new(EventLink::new());
-
-        manager.publish_timely_logging(worker, Some(timely_events.clone()));
-        manager.publish_differential_logging(worker, Some(differential_events.clone()));
-
-        let mut timely_logger = BatchLogger::new(timely_events.clone());
-        worker
-            .log_register()
-            .insert::<TimelyEvent,_>("timely", move |time, data| timely_logger.publish_batch(time, data));
-
-        let mut differential_logger = BatchLogger::new(differential_events.clone());
-        worker
-            .log_register()
-            .insert::<DifferentialEvent,_>("differential/arrange", move |time, data| differential_logger.publish_batch(time, data));
-
-        let mut sequencer = Sequencer::new(worker, timer);
-
-        let mut done = false;
-        while !done {
-
-            {   // Check out channel status.
-                let mut lock = recv.lock().expect("Mutex poisoned");
-                while let Some(command) = lock.pop_front() {
-                    sequencer.push(command);
-                }
-            }
-
-            // Dequeue and act on commands.
-            // One at a time, so that Shutdown works.
-            if let Some(command) = sequencer.next() {
-                println!("{:?}\tExecuting {:?}", timer.elapsed(), command);
-                if command == Command::Shutdown {
-                    done = true;
-                }
-                command.execute(&mut manager, worker);
-            }
-
-            worker.step();
-        }
-
-        println!("Shutting down");
-
-        // Disable sequencer for shut down.
-        drop(sequencer);
-
-        // Deregister loggers, so that the logging dataflows can shut down.
-        worker
-            .log_register()
-            .insert::<TimelyEvent,_>("timely", move |_time, _data| { });
-
-        worker
-            .log_register()
-            .insert::<DifferentialEvent,_>("differential/arrange", move |_time, _data| { });
-
-    }).expect("Timely computation did not initialize cleanly");
-
-    println!("Now accepting commands");
-
     // Detached thread for client connections.
     std::thread::Builder::new()
         .name("Listener".to_string())
@@ -111,7 +37,41 @@ fn main() {
                     })
                     .expect("failed to create thread");
             }
-
         })
         .expect("Failed to spawn listen thread");
+
+    // Initiate timely computation.
+    timely::execute_from_args(args, move |worker| {
+
+        let timer = ::std::time::Instant::now();
+        let recv = command_queue.clone();
+
+        let mut manager = Manager::<Value>::new();
+        let mut sequencer = Some(Sequencer::new(worker, timer));
+
+        while sequencer.is_some() {
+
+            // Check out channel status.
+            while let Some(command) = recv.lock().expect("Mutex poisoned").pop_front() {
+                sequencer
+                    .as_mut()
+                    .map(|s| s.push(command));
+            }
+
+            // Dequeue and act on commands.
+            // Once per iteration, so that Shutdown works "immediately".
+            if let Some(command) = sequencer.as_mut().and_then(|s| s.next()) {
+                println!("{:?}\tExecuting {:?}", timer.elapsed(), command);
+                if command == Command::Shutdown {
+                    sequencer = None;
+                }
+                command.execute(&mut manager, worker);
+            }
+
+            worker.step();
+        }
+
+        println!("Shutting down");
+
+    }).expect("Timely computation did not initialize cleanly");
 }

interactive/src/command.rs

@@ -95,7 +95,7 @@ impl<Value: Data+Hash> Command<Value> {
 
             Command::Shutdown => {
                 println!("Shutdown received");
-                manager.shutdown();
+                manager.shutdown(worker);
             }
         }
 

interactive/src/manager.rs

@@ -56,11 +56,49 @@ impl<Value: Data+Hash> Manager<Value> {
         }
     }
 
+    /// Enables logging of timely and differential events.
+    pub fn enable_logging<A: Allocate>(&mut self, worker: &mut Worker<A>)
+    where
+        TimelyEvent: AsVector<Value>,
+        DifferentialEvent: AsVector<Value>,
+    {
+
+        use std::rc::Rc;
+        use timely::dataflow::operators::capture::event::link::EventLink;
+        use timely::logging::BatchLogger;
+
+        let timely_events = Rc::new(EventLink::new());
+        let differential_events = Rc::new(EventLink::new());
+
+        self.publish_timely_logging(worker, Some(timely_events.clone()));
+        self.publish_differential_logging(worker, Some(differential_events.clone()));
+
+        let mut timely_logger = BatchLogger::new(timely_events.clone());
+        worker
+            .log_register()
+            .insert::<TimelyEvent,_>("timely", move |time, data| timely_logger.publish_batch(time, data));
+
+        let mut differential_logger = BatchLogger::new(differential_events.clone());
+        worker
+            .log_register()
+            .insert::<DifferentialEvent,_>("differential/arrange", move |time, data| differential_logger.publish_batch(time, data));
+
+    }
+
     /// Clear the managed inputs and traces.
-    pub fn shutdown(&mut self) {
+    pub fn shutdown<A: Allocate>(&mut self, worker: &mut Worker<A>) {
         self.inputs.sessions.clear();
         self.traces.inputs.clear();
         self.traces.arrangements.clear();
+
+        // Deregister loggers, so that the logging dataflows can shut down.
+        worker
+            .log_register()
+            .insert::<TimelyEvent,_>("timely", move |_time, _data| { });
+
+        worker
+            .log_register()
+            .insert::<DifferentialEvent,_>("differential/arrange", move |_time, _data| { });
     }
 
     /// Inserts a new input session by name.

interactive/src/plan/sfw.rs

@@ -0,0 +1,61 @@
+//! Multi-way equijoin expression plan.
+//!
+//! This plan provides us the opportunity to map out a non-trivial differential
+//! implementation for a complex join query. In particular, we are able to invoke
+//! delta-query and worst-case optimal join plans, which avoid any intermediate
+//! materialization.
+//!
+//! Each `MultiwayJoin` indicates several source collections, equality constraints
+//! among their attributes, and then the set of attributes to produce as results.
+//!
+//! One naive implementation would take each input collection in order, and develop
+//! the join restricted to the prefix of relations so far. Ideally the order would
+//! be such that joined collections have equality constraints and prevent Cartesian
+//! explosion. At each step, a new collection picks out some of the attributes and
+//! instantiates a primitive binary join between the accumulated collection and the
+//! next collection.
+//!
+//! A more sophisticated implementation establishes delta queries for each input
+//! collection, which responds to changes in that input collection against the
+//! current other input collections. For each input collection we may choose very
+//! different join orders, as the order must follow equality constraints.
+//!
+//! A further implementation could develop the results attribute-by-attribute, as
+//! opposed to collection-by-collection, which gives us the ability to use column
+//! indices rather than whole-collection indices.
+
+use std::hash::Hash;
+
+use timely::dataflow::Scope;
+
+use differential_dataflow::operators::JoinCore;
+
+use differential_dataflow::{Collection, Data};
+use plan::{Plan, Render};
+use {TraceManager, Time, Diff};
+
+/// A plan stage joining two source relations on the specified
+/// symbols. Throws if any of the join symbols isn't bound by both
+/// sources.
+#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
+pub struct MultiwayJoin<Value> {
+    /// Attributes to extract as `(attr, input)`.
+    pub results: Vec<(usize, usize)>,
+    /// Source collections.
+    pub sources: Vec<Box<Plan<Value>>>,
+    /// Equality constraints (as lists of equal `(attr, input)` pairs).
+    pub equalities: Vec<Vec<(usize, usize)>>,
+}
+
+impl<V: Data+Hash> Render for MultiwayJoin<V> {
+
+    type Value = V;
+
+    fn render<S: Scope<Timestamp = Time>>(
+        &self,
+        scope: &mut S,
+        arrangements: &mut TraceManager<Self::Value>) -> Collection<S, Vec<Self::Value>, Diff>
+    {
+        unimplemented!()
+    }
+}