Persistent log

Cargo.toml

@@ -9,9 +9,9 @@ b_memcached = ["memcached-rs"]
 b_postgresql = ["postgres"]
 b_mysql = ["mysql"]
 b_mssql = ["futures", "futures-state-stream", "tiberius", "tokio-core"]
-b_netsoup = ["futures", "tokio-core", "tarpc", "tarpc-plugins", "serde", "serde_derive"]
+b_netsoup = ["futures", "tokio-core", "tarpc", "tarpc-plugins"]
 b_hybrid = ["mysql", "memcached-rs"]
-default = ["web", "b_netsoup", "b_memcached", "b_mysql", "b_mssql", "b_hybrid"]
+default = ["web", "b_netsoup", "b_memcached", "b_mysql", "b_mssql", "b_hybrid", "serde", "serde_derive"]
 profiling = ["timekeeper/default"]
 binaries = ["default"]
 
@@ -59,6 +59,12 @@ serde_derive = { version = "0.9", optional = true }
 # for web
 rustc-serialize = { version = "0.3", optional = true }
 
+# durable log
+snowflake = "1.2.0"
+serde_json = "0.9.8"
+buf_redux = "0.6.1"
+time = "0.1"
+
 [dependencies.rustful]
 version = "0.9"
 default-features = false

benchmarks/piazza/piazza.rs

@@ -10,7 +10,7 @@ use std::fs::{OpenOptions, File};
 use std::io::Write;
 use slog::DrainExt;
 
-use distributary::{DataType, Join, JoinType, Blender, Base, NodeAddress, Filter, Mutator, Index};
+use distributary::{DataType, Join, JoinType, Blender, Base, BaseDurabilityLevel, NodeAddress, Filter, Mutator, Index};
 
 pub struct Piazza {
     pub soup: Blender,
@@ -53,7 +53,7 @@ impl Piazza {
             // add a post base table
             post = mig.add_ingredient("post",
                                       &["pid", "cid", "author", "content"],
-                                      Base::new(vec![1]));
+                                      Base::new(vec![1], BaseDurabilityLevel::None));
 
             // add a class base table
             class = mig.add_ingredient("class", &["cid", "classname"], Base::default());

src/flow/core/data.rs

@@ -11,8 +11,7 @@ use std::fmt;
 ///
 /// Having this be an enum allows for our code to be agnostic about the types of user data except
 /// when type information is specifically necessary.
-#[derive(Eq, PartialOrd, Ord, Hash, Debug, Clone)]
-#[cfg_attr(feature="b_netsoup", derive(Serialize, Deserialize))]
+#[derive(Eq, PartialOrd, Ord, Hash, Debug, Clone, Serialize, Deserialize)]
 pub enum DataType {
     /// An empty value.
     None,
@@ -184,7 +183,7 @@ impl fmt::Display for DataType {
 }
 
 /// A record is a single positive or negative data record with an associated time stamp.
-#[derive(Clone, PartialEq, Eq, Debug)]
+#[derive(Clone, PartialEq, Eq, Debug, Serialize, Deserialize)]
 pub enum Record {
     Positive(sync::Arc<Vec<DataType>>),
     Negative(sync::Arc<Vec<DataType>>),
@@ -293,7 +292,7 @@ impl IntoIterator for Records {
 /// Represents a set of records returned from a query.
 pub type Datas = Vec<Vec<DataType>>;
 
-#[derive(Clone, Default, PartialEq, Debug)]
+#[derive(Clone, Default, PartialEq, Debug, Serialize, Deserialize)]
 pub struct Records(Vec<Record>);
 
 impl Deref for Records {

src/flow/core/processing.rs

@@ -84,7 +84,7 @@ pub trait Ingredient
                 data: prelude::Records,
                 domain: &prelude::DomainNodes,
                 states: &prelude::StateMap)
-                -> prelude::Records;
+                -> Option<prelude::Records>;
 
     fn can_query_through(&self) -> bool {
         false

src/flow/domain/mod.rs

@@ -871,10 +871,3 @@ impl Domain {
             .unwrap()
     }
 }
-
-#[cfg(test)]
-impl Drop for Domain {
-    fn drop(&mut self) {
-        //println!("Dropping Domain!")
-    }
-}

src/flow/domain/single.rs

@@ -161,7 +161,7 @@ impl NodeDescriptor {
             }
             flow::node::Type::Internal(ref mut i) => {
                 let from = m.link().src;
-                m.map_data(|data| i.on_input(from, data, nodes, state));
+                m.map_data(|data| i.on_input(from, data, nodes, state).unwrap());
                 materialize(m.data(), state.get_mut(&addr));
                 m
             }

src/flow/mod.rs

@@ -463,7 +463,7 @@ impl<'a> Migration<'a> {
                 self.mainline.ingredients.add_edge(*parent.as_global(), ni, false);
             }
         }
-        // and tell the caller its id
+
         ni.into()
     }
 
@@ -889,18 +889,13 @@ impl<'a> Migration<'a> {
 
 impl Drop for Blender {
     fn drop(&mut self) {
-        //println!("Blender started dropping.");
-
         for (_, tx) in &mut self.txs {
             // don't unwrap, because given domain may already have terminated
             drop(tx.send(payload::Packet::Quit));
         }
         for d in self.domains.drain(..) {
-            //println!("Waiting for domain thread to join.");
             d.join().unwrap();
         }
-
-        //println!("Blender is done dropping.")
     }
 }
 

src/flow/payload.rs

@@ -74,6 +74,9 @@ pub enum Packet {
         parents: Vec<LocalNodeIndex>,
     },
 
+    /// Instruct base nodes to flush any buffered writes.
+    //FlushBufferedWrites,
+
     /// Set up a fresh, empty state for a node, indexed by a particular column.
     ///
     /// This is done in preparation of a subsequent state replay.

src/lib.rs

@@ -175,7 +175,8 @@
 //! # let mut g = Blender::new();
 //! # let article = {
 //! # let mut mig = g.start_migration();
-//! # let article = mig.add_ingredient("article", &["id", "title"], Base::default());
+//! # let article = mig.add_ingredient("article", &["id", "title"],
+//! #                                  Base::default());
 //! # mig.commit();
 //! # article
 //! # };
@@ -356,6 +357,11 @@ extern crate futures;
 #[cfg(feature="b_netsoup")]
 extern crate tokio_core;
 
+extern crate buf_redux;
+extern crate serde_json;
+extern crate snowflake;
+extern crate time;
+
 mod backlog;
 mod checktable;
 mod flow;
@@ -369,7 +375,7 @@ pub use flow::{Blender, Migration, Mutator};
 pub use flow::core::{NodeAddress, DataType, Datas};
 pub use flow::node::StreamUpdate;
 pub use flow::domain::Index;
-pub use ops::base::Base;
+pub use ops::base::{Base, BaseDurabilityLevel};
 pub use ops::grouped::aggregate::{Aggregator, Aggregation};
 pub use ops::grouped::concat::{GroupConcat, TextComponent};
 pub use ops::grouped::extremum::{Extremum, ExtremumOperator};

src/mir/mod.rs

@@ -772,7 +772,8 @@ fn make_base_node(name: &str,
             .collect();
         mig.add_ingredient(name,
                            column_names.as_slice(),
-                           ops::base::Base::new(pkey_column_ids))
+                           ops::base::Base::new(pkey_column_ids,
+                                                ops::base::BaseDurabilityLevel::None))
     } else {
         mig.add_ingredient(name, column_names.as_slice(), ops::base::Base::default())
     };