Persistent log

Cargo.toml

@@ -8,7 +8,7 @@ web = ["rustful", "rustc-serialize"]
 b_memcached = ["memcached-rs"]
 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"]
 profiling = ["timekeeper/default"]
@@ -20,6 +20,8 @@ itertools = "0.5"
 petgraph = "0.4"
 regex = "0.1"
 fnv = "1.0"
+serde = "0.9"
+serde_derive = "0.9"
 slog = "1.5.2"
 #slog = { version = "1.5.2", features = ["max_level_trace", "release_max_level_debug"] }
 slog-term = "1.5.0"
@@ -51,12 +53,16 @@ futures = { version ="0.1.9", optional = true }
 tokio-core = { version = "0.1", optional = true }
 tarpc = {git="https://github.com/ms705/tarpc.git", branch = "hack", optional = true}
 tarpc-plugins = { git = "https://github.com/ms705/tarpc", branch = "hack", optional = true }
-serde = { version = "0.9", optional = true }
-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/bank/bank.rs

@@ -16,7 +16,7 @@ use std::time;
 
 use std::collections::HashMap;
 
-use distributary::{Blender, Base, Aggregation, Join, JoinType, Datas, DataType, Token, Mutator};
+use distributary::{Blender, Base, BaseDurabilityLevel, Aggregation, Join, JoinType, Datas, DataType, Token, Mutator};
 
 use rand::Rng;
 
@@ -58,9 +58,10 @@ pub fn setup(num_putters: usize) -> Box<Bank> {
         let mut mig = g.start_migration();
 
         // add transfers base table
+        let d = BaseDurabilityLevel::SyncImmediately;  // Buffered makes assert on getter fail.
         transfers = mig.add_transactional_base("transfers",
                                                &["src_acct", "dst_acct", "amount"],
-                                               Base::default());
+                                               Base::default().with_durability(d));
 
         // add all debits
         debits = mig.add_ingredient("debits",
@@ -317,10 +318,13 @@ fn client(i: usize,
         let rl: u64 = read_latencies.iter().sum();
         let wl: u64 = write_latencies.iter().sum();
         let sl: u64 = settle_latencies.iter().sum();
+        let wsl: u64 = write_start_to_txn_end_latencies.iter().sum();
+
         let n = write_latencies.len() as f64;
         println!("read latency: {:.3} μs", rl as f64 / n * 0.001);
         println!("write latency: {:.3} μs", wl as f64 / n * 0.001);
         println!("settle latency: {:.3} μs", sl as f64 / n * 0.001);
+        println!("write + settle latency: {:.3} μs", wsl as f64 / n * 0.001);
 
         let mut latencies_hist = Histogram::<i64>::new_with_bounds(10, 10000000, 4).unwrap();
         for sample_nanos in write_start_to_txn_end_latencies {
@@ -417,7 +421,7 @@ fn main() {
     };
 
     // let system settle
-    // thread::sleep(time::Duration::new(1, 0));
+    // thread::sleep(time::Duration::new(2, 0));
     let start = time::Instant::now();
 
     // benchmark
@@ -432,6 +436,10 @@ fn main() {
 
                      if i == 0 {
                          populate(naccounts, &mut transfers_put);
+
+                         //println!("Let those accounts settle...");
+                         //thread::sleep(time::Duration::new(2, 0));
+                         //println!("Done!");
                      }
 
                      thread::Builder::new()

benchmarks/piazza/piazza.rs

@@ -53,7 +53,7 @@ impl Piazza {
             // add a post base table
             post = mig.add_ingredient("post",
                                       &["pid", "cid", "author", "content"],
-                                      Base::with_key(vec![1], vec![]));
+                                      Base::default().with_key(vec![1]));
 
             // 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,
@@ -195,7 +194,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>>),
@@ -321,7 +320,7 @@ impl<'a> IntoIterator for &'a 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/domain/mod.rs

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

src/flow/mod.rs

@@ -1,7 +1,7 @@
 use petgraph;
 use petgraph::graph::NodeIndex;
 use checktable;
-use ops::base::Base;
+use ops::base::{Base, BaseDurabilityLevel};
 use vec_map::VecMap;
 
 use std::sync::mpsc;
@@ -593,7 +593,9 @@ impl<'a> Migration<'a> {
               S2: ToString,
               FS: IntoIterator<Item = S2>
     {
-        let mut i: node::Type = b.into();
+        // transactions require a base node with SyncImmediately durability
+        let d = BaseDurabilityLevel::SyncImmediately;
+        let mut i: node::Type = b.with_durability(d).into();
         i.on_connected(&self.mainline.ingredients);
 
         // add to the graph
@@ -1109,18 +1111,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/lib.rs

@@ -358,6 +358,11 @@ extern crate vec_map;
 #[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;
@@ -371,7 +376,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

@@ -998,7 +998,7 @@ fn make_base_node(name: &str,
             .collect();
         mig.add_ingredient(name,
                            column_names.as_slice(),
-                           ops::base::Base::with_key(pkey_column_ids, vec![]))
+                           ops::base::Base::new(vec![]).with_key(pkey_column_ids))
     } else {
         mig.add_ingredient(name, column_names.as_slice(), ops::base::Base::new(vec![]))
     };