[wip] replicator

Cargo.toml

@@ -3,11 +3,12 @@ name = "solana"
 description = "High Performance Blockchain"
 version = "0.4.0"
 documentation = "https://docs.rs/solana"
-homepage = "http://loomprotocol.com/"
+homepage = "http://solana.io/"
 repository = "https://github.com/solana-labs/solana"
 authors = [
-    "Anatoly Yakovenko <[email protected]>",
-    "Greg Fitzgerald <[email protected]>",
+    "Anatoly Yakovenko <[email protected]>",
+    "Greg Fitzgerald <[email protected]>",
+    "Stephen Akridge <[email protected]>",
 ]
 license = "Apache-2.0"
 
@@ -42,6 +43,7 @@ codecov = { repository = "solana-labs/solana", branch = "master", service = "git
 unstable = []
 ipv6 = []
 cuda = []
+erasure = []
 
 [dependencies]
 rayon = "1.0.0"
@@ -60,4 +62,4 @@ matches = "^0.1.6"
 byteorder = "^1.2.1"
 libc = "^0.2.1"
 getopts = "^0.2"
-
+isatty = "0.1"

LICENSE

@@ -1,4 +1,4 @@
-Copyright 2018 Anatoly Yakovenko <[email protected]> and Greg Fitzgerald <[email protected]>
+Copyright 2018 Anatoly Yakovenko, Greg Fitzgerald and Stephen Akridge
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

README.md

@@ -14,6 +14,11 @@ Solana: High Performance Blockchain
 Solana™ is a new architecture for a high performance blockchain. It aims to support
 over 700 thousand transactions per second on a gigabit network.
 
+Introduction
+===
+
+It's possible for a centralized database to process 710,000 transactions per second on a standard gigabit network if the transactions are, on average, no more than 178 bytes. A centralized database can also replicate itself and maintain high availability without significantly compromising that transaction rate using the distributed system technique known as Optimistic Concurrency Control [H.T.Kung, J.T.Robinson (1981)]. At Solana, we're demonstrating that these same theoretical limits apply just as well to blockchain on an adversarial network. The key ingredient? Finding a way to share time when nodes can't trust one-another. Once nodes can trust time, suddenly ~40 years of distributed systems research becomes applicable to blockchain! Furthermore, and much to our surprise, it can implemented using a mechanism that has existed in Bitcoin since day one. The Bitcoin feature is called nLocktime and it can be used to postdate transactions using block height instead of a timestamp. As a Bitcoin client, you'd use block height instead of a timestamp if you don't trust the network. Block height turns out to be an instance of what's being called a Verifiable Delay Function in cryptography circles. It's a cryptographically secure way to say time has passed. In Solana, we use a far more granular verifiable delay function, a SHA 256 hash chain, to checkpoint the ledger and coordinate consensus. With it, we implement Optimistic Concurrency Control and are now well in route towards that theoretical limit of 710,000 transactions per second.
+
 Running the demo
 ===
 
@@ -24,6 +29,13 @@ $ curl https://sh.rustup.rs -sSf | sh
 $ source $HOME/.cargo/env
 ```
 
+Now checkout the code from github:
+
+```bash
+$ git clone https://github.com/solana-labs/solana.git 
+$ cd solana
+```
+
 The testnode server is initialized with a ledger from stdin and
 generates new ledger entries on stdout. To create the input ledger, we'll need
 to create *the mint* and use it to generate a *genesis ledger*. It's done in

build.rs

@@ -1,12 +1,15 @@
 use std::env;
 
 fn main() {
+    println!("cargo:rustc-link-search=native=.");
     if !env::var("CARGO_FEATURE_CUDA").is_err() {
-        println!("cargo:rustc-link-search=native=.");
         println!("cargo:rustc-link-lib=static=cuda_verify_ed25519");
         println!("cargo:rustc-link-search=native=/usr/local/cuda/lib64");
         println!("cargo:rustc-link-lib=dylib=cudart");
         println!("cargo:rustc-link-lib=dylib=cuda");
         println!("cargo:rustc-link-lib=dylib=cudadevrt");
     }
+    if !env::var("CARGO_FEATURE_ERASURE").is_err() {
+        println!("cargo:rustc-link-lib=dylib=Jerasure");
+    }
 }

src/accountant_skel.rs

@@ -22,11 +22,15 @@ use std::io::Write;
 use std::net::{SocketAddr, UdpSocket};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc::{channel, Receiver, Sender};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, RwLock};
 use std::thread::{spawn, JoinHandle};
 use std::time::Duration;
 use streamer;
 use transaction::Transaction;
+use subscribers::Subscribers;
+
+use subscribers;
+use std::mem::size_of;
 
 pub struct AccountantSkel<W: Write + Send + 'static> {
     acc: Accountant,
@@ -245,8 +249,32 @@ impl<W: Write + Send + 'static> AccountantSkel<W> {
         }
         Ok(())
     }
+    /// Process verified blobs, already in order
+    /// Respond with a signed hash of the state
+    fn replicate_state(
+        obj: &Arc<Mutex<AccountantSkel<W>>>,
+        verified_receiver: &BlobReceiver,
+        blob_sender: &streamer::BlobSender,
+        blob_recycler: &packet::BlobRecycler,
+    ) -> Result<()> {
+        let timer = Duration::new(1, 0);
+        let blobs = verified_receiver.recv_timeout(timer)?;
+        for msgs in blobs {
+            let entries:Vec<Entry> = b.read().unwrap().data.deserialize()?;
+            for e in entries {
+                obj.lock().unwrap().acc.process_verified_events(e.events)?;
+            }
+            //TODO respond back to leader with hash of the state
+        }
+        for blob in blobs {
+            blob_recycler.recycle(blob);
+        }
+        Ok(())
+    }
+
 
     /// Create a UDP microservice that forwards messages the given AccountantSkel.
+    /// This service is the network leader
     /// Set `exit` to shutdown its threads.
     pub fn serve(
         obj: &Arc<Mutex<AccountantSkel<W>>>,
@@ -279,7 +307,7 @@ impl<W: Write + Send + 'static> AccountantSkel<W> {
 
         let skel = obj.clone();
         let t_server = spawn(move || loop {
-            let e = AccountantSkel::process(
+            let e = Self::process(
                 &skel,
                 &verified_receiver,
                 &blob_sender,
@@ -292,6 +320,71 @@ impl<W: Write + Send + 'static> AccountantSkel<W> {
         });
         Ok(vec![t_receiver, t_responder, t_server, t_verifier])
     }
+
+    /// This service receives messages from a leader in the network and processes the transactions
+    /// on the accountant state.
+    /// # Arguments
+    /// * `obj` - The accountant state.
+    /// * `rsubs` - The subscribers.
+    /// * `exit` - The exit signal.
+    /// # Remarks
+    /// The pipeline is constructed as follows:
+    /// 1. receive blobs from the network, these are out of order
+    /// 2. verify blobs, PoH, signatures (TODO)
+    /// 3. reconstruct contiguous window
+    ///     a. order the blobs
+    ///     b. use erasure coding to reconstruct missing blobs
+    ///     c. ask the network for missing blobs, if erasure coding is insufficient
+    ///     d. make sure that the blobs PoH sequences connect (TODO)
+    /// 4. process the transaction state machine
+    /// 5. respond with the hash of the state back to the leader
+    pub fn replicate(
+        obj: &Arc<Mutex<AccountantSkel<W>>>,
+        rsubs: Subscribers,
+        exit: Arc<AtomicBool>,
+    ) -> Result<Vec<JoinHandle<()>>> {
+        let read = UdpSocket::bind(rsubs.me.addr)?;
+        // make sure we are on the same interface
+        let mut local = read.local_addr()?;
+        local.set_port(0);
+        let write = UdpSocket::bind(local)?;
+
+        let blob_recycler = packet::BlobRecycler::default();
+        let (blob_sender, blob_receiver) = channel();
+        let t_blob_receiver =
+            streamer::blob_receiver(exit.clone(), blob_recycler.clone(), read, blob_sender)?;
+        let (window_sender, window_receiver) = channel();
+        let (retransmit_sender, retransmit_receiver) = channel();
+
+        let subs = Arc::new(RwLock::new(rsubs));
+        let t_retransmit = streamer::retransmitter(
+            write,
+            exit.clone(),
+            subs,
+            blob_recycler.clone(),
+            retransmit_receiver,
+        );
+        //TODO
+        //the packets comming out of blob_receiver need to be sent to the GPU and verified
+        //then sent to the window, which does the erasure coding reconstruction
+        let t_window = streamer::window(
+            exit.clone(),
+            subs,
+            blob_recycler.clone(),
+            blob_receiver,
+            window_sender,
+            retransmit_sender,
+        );
+
+        let skel = obj.clone();
+        let t_server = spawn(move || loop {
+            let e = Self::replicate_state(&skel, &window_receiver, &blob_sender, &blob_recycler);
+            if e.is_err() && exit.load(Ordering::Relaxed) {
+                break;
+            }
+        });
+        Ok(vec![t_blob_receiver, t_retransmit, t_window, t_server])
+    }
 }
 
 #[cfg(test)]
@@ -319,7 +412,7 @@ mod tests {
     use accountant_skel::{to_packets, Request};
     use bincode::serialize;
     use ecdsa;
-    use packet::{PacketRecycler, NUM_PACKETS};
+    use packet::{BlobRecycler, PacketRecycler, NUM_PACKETS};
     use transaction::{memfind, test_tx};
 
     use accountant::Accountant;
@@ -339,6 +432,12 @@ mod tests {
     use std::time::Duration;
     use transaction::Transaction;
 
+    use subscribers::{Node, Subscribers};
+    use streamer;
+    use std::sync::mpsc::channel;
+    use std::collections::VecDeque;
+    use packet::{PACKET_DATA_SIZE};
+
     #[test]
     fn test_layout() {
         let tr = test_tx();
@@ -443,6 +542,54 @@ mod tests {
         exit.store(true, Ordering::Relaxed);
     }
 
+    #[test]
+    fn test_replicate() {
+        let read = UdpSocket::bind("127.0.0.1:0").expect("bind");
+        let addr = read.local_addr().unwrap();
+        let send = UdpSocket::bind("127.0.0.1:0").expect("bind");
+        let exit = Arc::new(AtomicBool::new(false));
+
+        let node_me = Node::default();
+        let node_leader = Node::new([0; 8], 0, send.local_addr().unwrap());
+        let subs = Subscribers::new(node_me, node_leader, &[]);
+
+        let recv_recycler = PacketRecycler::default();
+        let resp_recycler = BlobRecycler::default();
+        let (s_reader, r_reader) = channel();
+        let t_receiver = streamer::receiver(read, exit.clone(), recv_recycler.clone(), s_reader).unwrap();
+        let (s_responder, r_responder) = channel();
+        let t_responder = streamer::responder(send, exit.clone(), resp_recycler.clone(), r_responder);
+
+        let alice = Mint::new(10_000);
+        let acc = Accountant::new(&alice);
+        let bob_pubkey = KeyPair::new().pubkey();
+        let historian = Historian::new(&alice.last_id(), Some(30));
+        let acc = Arc::new(Mutex::new(AccountantSkel::new(
+            acc,
+            alice.last_id(),
+            sink(),
+            historian,
+        )));
+
+        let _threads = AccountantSkel::replicate(&acc, subs, exit.clone()).unwrap();
+
+        let mut msgs = VecDeque::new();
+        for i in 0..10 {
+            let b = resp_recycler.allocate();
+            let b_ = b.clone();
+            let mut w = b.write().unwrap();
+            w.data[0] = i as u8;
+            w.meta.size = PACKET_DATA_SIZE;
+            w.meta.set_addr(&addr);
+            msgs.push_back(b_);
+        }
+        s_responder.send(msgs).expect("send");
+
+        exit.store(true, Ordering::Relaxed);
+        t_receiver.join().expect("join");
+        t_responder.join().expect("join");
+    }
+
 }
 
 #[cfg(all(feature = "unstable", test))]

src/bin/client-demo.rs

@@ -1,20 +1,32 @@
 extern crate getopts;
+extern crate isatty;
 extern crate rayon;
 extern crate serde_json;
 extern crate solana;
 
 use getopts::Options;
+use isatty::stdin_isatty;
 use rayon::prelude::*;
 use solana::accountant_stub::AccountantStub;
 use solana::mint::Mint;
 use solana::signature::{KeyPair, KeyPairUtil};
 use solana::transaction::Transaction;
 use std::env;
-use std::io::stdin;
+use std::io::{stdin, Read};
 use std::net::UdpSocket;
+use std::process::exit;
 use std::thread::sleep;
 use std::time::{Duration, Instant};
 
+fn print_usage(program: &str, opts: Options) {
+    let mut brief = format!("Usage: cat <mint.json> | {} [options]\n\n", program);
+    brief += "  Solana client demo creates a number of transactions and\n";
+    brief += "  sends them to a target node.";
+    brief += "  Takes json formatted mint file to stdin.";
+
+    print!("{}", opts.usage(&brief));
+}
+
 fn main() {
     let mut threads = 4usize;
     let mut addr: String = "127.0.0.1:8000".to_string();
@@ -24,12 +36,21 @@ fn main() {
     opts.optopt("s", "", "server address", "host:port");
     opts.optopt("c", "", "client address", "host:port");
     opts.optopt("t", "", "number of threads", "4");
+    opts.optflag("h", "help", "print help");
     let args: Vec<String> = env::args().collect();
     let matches = match opts.parse(&args[1..]) {
         Ok(m) => m,
-        Err(f) => panic!(f.to_string()),
+        Err(e) => {
+            eprintln!("{}", e);
+            exit(1);
+        }
     };
 
+    if matches.opt_present("h") {
+        let program = args[0].clone();
+        print_usage(&program, opts);
+        return;
+    }
     if matches.opt_present("s") {
         addr = matches.opt_str("s").unwrap();
     }
@@ -39,7 +60,23 @@ fn main() {
     if matches.opt_present("t") {
         threads = matches.opt_str("t").unwrap().parse().expect("integer");
     }
-    let mint: Mint = serde_json::from_reader(stdin()).unwrap();
+
+    if stdin_isatty() {
+        eprintln!("nothing found on stdin, expected a json file");
+        exit(1);
+    }
+
+    let mut buffer = String::new();
+    let num_bytes = stdin().read_to_string(&mut buffer).unwrap();
+    if num_bytes == 0 {
+        eprintln!("empty file on stdin, expected a json file");
+        exit(1);
+    }
+
+    let mint: Mint = serde_json::from_str(&buffer).unwrap_or_else(|e| {
+        eprintln!("failed to parse json: {}", e);
+        exit(1);
+    });
     let mint_keypair = mint.keypair();
     let mint_pubkey = mint.pubkey();