Permit testnets without a GPU

- Also use rayon to parallelize to_blobs() to maximize CPU usage

ci/testnet-deploy.sh

@@ -9,6 +9,7 @@ validatorNodeCount=10
 publicNetwork=false
 snapChannel=edge
 delete=false
+enableGpu=false
 
 usage() {
   exitcode=0
@@ -30,6 +31,7 @@ Deploys a CD testnet
    -n [number]          - Number of validator nodes (default: $validatorNodeCount)
    -c [number]          - Number of client nodes (default: $clientNodeCount)
    -P                   - Use public network IP addresses (default: $publicNetwork)
+   -g                   - Enable GPU (default: $enableGpu)
    -a [address]         - Set the leader node's external IP address to this GCE address
    -d                   - Delete the network
 
@@ -45,7 +47,7 @@ zone=$2
 [[ -n $zone ]] || usage "Zone not specified"
 shift 2
 
-while getopts "h?p:Pn:c:s:a:d" opt; do
+while getopts "h?p:Pn:c:s:ga:d" opt; do
   case $opt in
   h | \?)
     usage
@@ -69,6 +71,9 @@ while getopts "h?p:Pn:c:s:a:d" opt; do
       ;;
     esac
     ;;
+  g)
+    enableGpu=true
+    ;;
   a)
     leaderAddress=$OPTARG
     ;;
@@ -86,11 +91,14 @@ gce_create_args=(
   -a "$leaderAddress"
   -c "$clientNodeCount"
   -n "$validatorNodeCount"
-  -g
   -p "$netName"
   -z "$zone"
 )
 
+if $enableGpu; then
+  gce_create_args+=(-g)
+fi
+
 if $publicNetwork; then
   gce_create_args+=(-P)
 fi

src/bin/keygen.rs

@@ -21,6 +21,7 @@ fn main() -> Result<(), Box<error::Error>> {
                 .long("outfile")
                 .value_name("PATH")
                 .takes_value(true)
+                .required(true)
                 .help("path to generated file"),
         ).get_matches();
 

src/broadcast_stage.rs

@@ -2,20 +2,23 @@
 //!
 use counter::Counter;
 use crdt::{Crdt, CrdtError, NodeInfo};
+use entry::Entry;
 #[cfg(feature = "erasure")]
 use erasure;
+use ledger::Block;
 use log::Level;
-use packet::BlobRecycler;
+use packet::{BlobRecycler, SharedBlobs};
+use rayon::prelude::*;
 use result::{Error, Result};
 use service::Service;
 use std::mem;
 use std::net::UdpSocket;
 use std::sync::atomic::AtomicUsize;
-use std::sync::mpsc::RecvTimeoutError;
+use std::sync::mpsc::{Receiver, RecvTimeoutError};
 use std::sync::{Arc, RwLock};
 use std::thread::{self, Builder, JoinHandle};
-use std::time::Duration;
-use streamer::BlobReceiver;
+use std::time::{Duration, Instant};
+use timing::duration_as_ms;
 use window::{self, SharedWindow, WindowIndex, WindowUtil, WINDOW_SIZE};
 
 fn broadcast(
@@ -23,27 +26,42 @@ fn broadcast(
     broadcast_table: &[NodeInfo],
     window: &SharedWindow,
     recycler: &BlobRecycler,
-    receiver: &BlobReceiver,
+    receiver: &Receiver<Vec<Entry>>,
     sock: &UdpSocket,
     transmit_index: &mut WindowIndex,
     receive_index: &mut u64,
 ) -> Result<()> {
     let id = node_info.id;
     let timer = Duration::new(1, 0);
-    let mut dq = receiver.recv_timeout(timer)?;
-    while let Ok(mut nq) = receiver.try_recv() {
-        dq.append(&mut nq);
+    let entries = receiver.recv_timeout(timer)?;
+    let mut num_entries = entries.len();
+    let mut ventries = Vec::new();
+    ventries.push(entries);
+    while let Ok(entries) = receiver.try_recv() {
+        num_entries += entries.len();
+        ventries.push(entries);
     }
 
+    let to_blobs_start = Instant::now();
+    let dq: SharedBlobs = ventries
+        .into_par_iter()
+        .flat_map(|p| p.to_blobs(recycler))
+        .collect();
+
+    let to_blobs_elapsed = duration_as_ms(&to_blobs_start.elapsed());
+
     // flatten deque to vec
     let blobs_vec: Vec<_> = dq.into_iter().collect();
 
+    let blobs_chunking = Instant::now();
     // We could receive more blobs than window slots so
     // break them up into window-sized chunks to process
     let blobs_chunked = blobs_vec.chunks(WINDOW_SIZE as usize).map(|x| x.to_vec());
+    let chunking_elapsed = duration_as_ms(&blobs_chunking.elapsed());
 
     trace!("{}", window.read().unwrap().print(&id, *receive_index));
 
+    let broadcast_start = Instant::now();
     for mut blobs in blobs_chunked {
         let blobs_len = blobs.len();
         trace!("{}: broadcast blobs.len: {}", id, blobs_len);
@@ -115,6 +133,13 @@ fn broadcast(
             *receive_index,
         )?;
     }
+    let broadcast_elapsed = duration_as_ms(&broadcast_start.elapsed());
+
+    info!(
+        "broadcast: {} entries, blob time {} chunking time {} broadcast time {}",
+        num_entries, to_blobs_elapsed, chunking_elapsed, broadcast_elapsed
+    );
+
     Ok(())
 }
 
@@ -129,7 +154,7 @@ impl BroadcastStage {
         window: &SharedWindow,
         entry_height: u64,
         recycler: &BlobRecycler,
-        receiver: &BlobReceiver,
+        receiver: &Receiver<Vec<Entry>>,
     ) {
         let mut transmit_index = WindowIndex {
             data: entry_height,
@@ -177,7 +202,7 @@ impl BroadcastStage {
         window: SharedWindow,
         entry_height: u64,
         recycler: BlobRecycler,
-        receiver: BlobReceiver,
+        receiver: Receiver<Vec<Entry>>,
     ) -> Self {
         let thread_hdl = Builder::new()
             .name("solana-broadcaster".to_string())

src/fullnode.rs

@@ -243,7 +243,7 @@ impl Fullnode {
                 // TODO: To light up PoH, uncomment the following line:
                 //let tick_duration = Some(Duration::from_millis(1000));
 
-                let (tpu, blob_receiver) = Tpu::new(
+                let (tpu, entry_receiver) = Tpu::new(
                     keypair,
                     &bank,
                     &crdt,
@@ -262,7 +262,7 @@ impl Fullnode {
                     shared_window,
                     entry_height,
                     blob_recycler.clone(),
-                    blob_receiver,
+                    entry_receiver,
                 );
                 thread_hdls.extend(broadcast_stage.thread_hdls());
             }

src/record_stage.rs

@@ -5,11 +5,12 @@
 //! Transaction, the latest hash, and the number of hashes since the last transaction.
 //! The resulting stream of entries represents ordered transactions in time.
 
+use bank::Bank;
 use entry::Entry;
-use hash::Hash;
 use recorder::Recorder;
 use service::Service;
 use std::sync::mpsc::{channel, Receiver, RecvError, Sender, TryRecvError};
+use std::sync::Arc;
 use std::thread::{self, Builder, JoinHandle};
 use std::time::{Duration, Instant};
 use transaction::Transaction;
@@ -27,18 +28,15 @@ pub struct RecordStage {
 impl RecordStage {
     /// A background thread that will continue tagging received Transaction messages and
     /// sending back Entry messages until either the receiver or sender channel is closed.
-    pub fn new(
-        signal_receiver: Receiver<Signal>,
-        start_hash: &Hash,
-    ) -> (Self, Receiver<Vec<Entry>>) {
+    pub fn new(signal_receiver: Receiver<Signal>, bank: Arc<Bank>) -> (Self, Receiver<Vec<Entry>>) {
         let (entry_sender, entry_receiver) = channel();
-        let start_hash = *start_hash;
+        let start_hash = bank.last_id();
 
         let thread_hdl = Builder::new()
             .name("solana-record-stage".to_string())
             .spawn(move || {
                 let mut recorder = Recorder::new(start_hash);
-                let _ = Self::process_signals(&mut recorder, &signal_receiver, &entry_sender);
+                let _ = Self::process_signals(&mut recorder, &signal_receiver, bank, &entry_sender);
             }).unwrap();
 
         (RecordStage { thread_hdl }, entry_receiver)
@@ -47,11 +45,11 @@ impl RecordStage {
     /// Same as `RecordStage::new`, but will automatically produce entries every `tick_duration`.
     pub fn new_with_clock(
         signal_receiver: Receiver<Signal>,
-        start_hash: &Hash,
+        bank: Arc<Bank>,
         tick_duration: Duration,
     ) -> (Self, Receiver<Vec<Entry>>) {
         let (entry_sender, entry_receiver) = channel();
-        let start_hash = *start_hash;
+        let start_hash = bank.last_id();
 
         let thread_hdl = Builder::new()
             .name("solana-record-stage".to_string())
@@ -64,6 +62,7 @@ impl RecordStage {
                         start_time,
                         tick_duration,
                         &signal_receiver,
+                        bank.clone(),
                         &entry_sender,
                     ).is_err()
                     {
@@ -78,6 +77,7 @@ impl RecordStage {
 
     fn process_signal(
         signal: Signal,
+        bank: &Arc<Bank>,
         recorder: &mut Recorder,
         sender: &Sender<Vec<Entry>>,
     ) -> Result<(), ()> {
@@ -87,6 +87,13 @@ impl RecordStage {
             vec![]
         };
         let entries = recorder.record(txs);
+
+        for entry in entries.iter() {
+            if !entry.has_more {
+                bank.register_entry_id(&entry.id);
+            }
+        }
+
         sender.send(entries).or(Err(()))?;
         Ok(())
     }
@@ -94,11 +101,12 @@ impl RecordStage {
     fn process_signals(
         recorder: &mut Recorder,
         receiver: &Receiver<Signal>,
+        bank: Arc<Bank>,
         sender: &Sender<Vec<Entry>>,
     ) -> Result<(), ()> {
         loop {
             match receiver.recv() {
-                Ok(signal) => Self::process_signal(signal, recorder, sender)?,
+                Ok(signal) => Self::process_signal(signal, &bank, recorder, sender)?,
                 Err(RecvError) => return Err(()),
             }
         }
@@ -109,6 +117,7 @@ impl RecordStage {
         start_time: Instant,
         tick_duration: Duration,
         receiver: &Receiver<Signal>,
+        bank: Arc<Bank>,
         sender: &Sender<Vec<Entry>>,
     ) -> Result<(), ()> {
         loop {
@@ -116,7 +125,7 @@ impl RecordStage {
                 sender.send(vec![entry]).or(Err(()))?;
             }
             match receiver.try_recv() {
-                Ok(signal) => Self::process_signal(signal, recorder, sender)?,
+                Ok(signal) => Self::process_signal(signal, &bank, recorder, sender)?,
                 Err(TryRecvError::Empty) => return Ok(()),
                 Err(TryRecvError::Disconnected) => return Err(()),
             };
@@ -137,16 +146,21 @@ impl Service for RecordStage {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use bank::Bank;
     use ledger::Block;
+    use mint::Mint;
     use signature::{Keypair, KeypairUtil};
     use std::sync::mpsc::channel;
+    use std::sync::Arc;
     use std::thread::sleep;
 
     #[test]
     fn test_historian() {
         let (tx_sender, tx_receiver) = channel();
-        let zero = Hash::default();
-        let (record_stage, entry_receiver) = RecordStage::new(tx_receiver, &zero);
+        let mint = Mint::new(1234);
+        let bank = Arc::new(Bank::new(&mint));
+        let zero = bank.last_id();
+        let (record_stage, entry_receiver) = RecordStage::new(tx_receiver, bank);
 
         tx_sender.send(Signal::Tick).unwrap();
         sleep(Duration::new(0, 1_000_000));
@@ -171,8 +185,9 @@ mod tests {
     #[test]
     fn test_historian_closed_sender() {
         let (tx_sender, tx_receiver) = channel();
-        let zero = Hash::default();
-        let (record_stage, entry_receiver) = RecordStage::new(tx_receiver, &zero);
+        let mint = Mint::new(1234);
+        let bank = Arc::new(Bank::new(&mint));
+        let (record_stage, entry_receiver) = RecordStage::new(tx_receiver, bank);
         drop(entry_receiver);
         tx_sender.send(Signal::Tick).unwrap();
         assert_eq!(record_stage.thread_hdl.join().unwrap(), ());
@@ -181,8 +196,10 @@ mod tests {
     #[test]
     fn test_transactions() {
         let (tx_sender, signal_receiver) = channel();
-        let zero = Hash::default();
-        let (_record_stage, entry_receiver) = RecordStage::new(signal_receiver, &zero);
+        let mint = Mint::new(1234);
+        let bank = Arc::new(Bank::new(&mint));
+        let zero = bank.last_id();
+        let (_record_stage, entry_receiver) = RecordStage::new(signal_receiver, bank);
         let alice_keypair = Keypair::new();
         let bob_pubkey = Keypair::new().pubkey();
         let tx0 = Transaction::new(&alice_keypair, bob_pubkey, 1, zero);
@@ -198,9 +215,11 @@ mod tests {
     #[test]
     fn test_clock() {
         let (tx_sender, tx_receiver) = channel();
-        let zero = Hash::default();
+        let mint = Mint::new(1234);
+        let bank = Arc::new(Bank::new(&mint));
+        let zero = bank.last_id();
         let (_record_stage, entry_receiver) =
-            RecordStage::new_with_clock(tx_receiver, &zero, Duration::from_millis(20));
+            RecordStage::new_with_clock(tx_receiver, bank, Duration::from_millis(20));
         sleep(Duration::from_millis(900));
         tx_sender.send(Signal::Tick).unwrap();
         drop(tx_sender);

src/tpu.rs

@@ -28,6 +28,7 @@
 use bank::Bank;
 use banking_stage::BankingStage;
 use crdt::Crdt;
+use entry::Entry;
 use fetch_stage::FetchStage;
 use packet::{BlobRecycler, PacketRecycler};
 use record_stage::RecordStage;
@@ -36,10 +37,10 @@ use signature::Keypair;
 use sigverify_stage::SigVerifyStage;
 use std::net::UdpSocket;
 use std::sync::atomic::AtomicBool;
+use std::sync::mpsc::Receiver;
 use std::sync::{Arc, RwLock};
 use std::thread::{self, JoinHandle};
 use std::time::Duration;
-use streamer::BlobReceiver;
 use write_stage::WriteStage;
 
 pub struct Tpu {
@@ -61,7 +62,7 @@ impl Tpu {
         exit: Arc<AtomicBool>,
         ledger_path: &str,
         sigverify_disabled: bool,
-    ) -> (Self, BlobReceiver) {
+    ) -> (Self, Receiver<Vec<Entry>>) {
         let packet_recycler = PacketRecycler::default();
 
         let (fetch_stage, packet_receiver) =
@@ -75,12 +76,12 @@ impl Tpu {
 
         let (record_stage, entry_receiver) = match tick_duration {
             Some(tick_duration) => {
-                RecordStage::new_with_clock(signal_receiver, &bank.last_id(), tick_duration)
+                RecordStage::new_with_clock(signal_receiver, bank.clone(), tick_duration)
             }
-            None => RecordStage::new(signal_receiver, &bank.last_id()),
+            None => RecordStage::new(signal_receiver, bank.clone()),
         };
 
-        let (write_stage, blob_receiver) = WriteStage::new(
+        let (write_stage, entry_forwarder) = WriteStage::new(
             keypair,
             bank.clone(),
             crdt.clone(),
@@ -96,7 +97,7 @@ impl Tpu {
             record_stage,
             write_stage,
         };
-        (tpu, blob_receiver)
+        (tpu, entry_forwarder)
     }
 
     pub fn close(self) -> thread::Result<()> {

src/write_stage.rs

@@ -14,11 +14,12 @@ use service::Service;
 use signature::Keypair;
 use std::net::UdpSocket;
 use std::sync::atomic::AtomicUsize;
-use std::sync::mpsc::{channel, Receiver, RecvTimeoutError};
+use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender};
 use std::sync::{Arc, RwLock};
 use std::thread::{self, Builder, JoinHandle};
-use std::time::Duration;
-use streamer::{responder, BlobReceiver, BlobSender};
+use std::time::{Duration, Instant};
+use streamer::responder;
+use timing::{duration_as_ms, duration_as_s};
 use vote_stage::send_leader_vote;
 
 pub struct WriteStage {
@@ -27,41 +28,72 @@ pub struct WriteStage {
 
 impl WriteStage {
     /// Process any Entry items that have been published by the RecordStage.
-    /// continuosly broadcast blobs of entries out
+    /// continuosly send entries out
     pub fn write_and_send_entries(
         crdt: &Arc<RwLock<Crdt>>,
-        bank: &Arc<Bank>,
         ledger_writer: &mut LedgerWriter,
-        blob_sender: &BlobSender,
-        blob_recycler: &BlobRecycler,
+        entry_sender: &Sender<Vec<Entry>>,
         entry_receiver: &Receiver<Vec<Entry>>,
     ) -> Result<()> {
+        let mut ventries = Vec::new();
         let entries = entry_receiver.recv_timeout(Duration::new(1, 0))?;
+        let mut num_entries = entries.len();
+        let mut num_txs = 0;
 
-        let votes = &entries.votes();
-        crdt.write().unwrap().insert_votes(&votes);
+        ventries.push(entries);
+        while let Ok(more) = entry_receiver.try_recv() {
+            num_entries += more.len();
+            ventries.push(more);
+        }
 
-        ledger_writer.write_entries(entries.clone())?;
+        info!("write_stage entries: {}", num_entries);
 
-        for entry in &entries {
-            if !entry.has_more {
-                bank.register_entry_id(&entry.id);
+        let to_blobs_total = 0;
+        let mut blob_send_total = 0;
+        let mut register_entry_total = 0;
+        let mut crdt_votes_total = 0;
+
+        let start = Instant::now();
+        for _ in 0..ventries.len() {
+            let entries = ventries.pop().unwrap();
+            for e in entries.iter() {
+                num_txs += e.transactions.len();
             }
+            let crdt_votes_start = Instant::now();
+            let votes = &entries.votes();
+            crdt.write().unwrap().insert_votes(&votes);
+            crdt_votes_total += duration_as_ms(&crdt_votes_start.elapsed());
+
+            ledger_writer.write_entries(entries.clone())?;
+
+            let register_entry_start = Instant::now();
+            register_entry_total += duration_as_ms(&register_entry_start.elapsed());
+
+            inc_new_counter_info!("write_stage-write_entries", entries.len());
+
+            //TODO(anatoly): real stake based voting needs to change this
+            //leader simply votes if the current set of validators have voted
+            //on a valid last id
+
+            trace!("New entries? {}", entries.len());
+            let blob_send_start = Instant::now();
+            if !entries.is_empty() {
+                inc_new_counter_info!("write_stage-recv_vote", votes.len());
+                inc_new_counter_info!("write_stage-broadcast_entries", entries.len());
+                trace!("broadcasting {}", entries.len());
+                entry_sender.send(entries)?;
+            }
+
+            blob_send_total += duration_as_ms(&blob_send_start.elapsed());
         }
+        info!("done write_stage txs: {} time {} ms txs/s: {} to_blobs_total: {} register_entry_total: {} blob_send_total: {} crdt_votes_total: {}",
+              num_txs, duration_as_ms(&start.elapsed()),
+              num_txs as f32 / duration_as_s(&start.elapsed()),
+              to_blobs_total,
+              register_entry_total,
+              blob_send_total,
+              crdt_votes_total);
 
-        //TODO(anatoly): real stake based voting needs to change this
-        //leader simply votes if the current set of validators have voted
-        //on a valid last id
-
-        trace!("New blobs? {}", entries.len());
-        let blobs = entries.to_blobs(blob_recycler);
-
-        if !blobs.is_empty() {
-            inc_new_counter_info!("write_stage-recv_vote", votes.len());
-            inc_new_counter_info!("write_stage-broadcast_blobs", blobs.len());
-            trace!("broadcasting {}", blobs.len());
-            blob_sender.send(blobs)?;
-        }
         Ok(())
     }
 
@@ -73,7 +105,7 @@ impl WriteStage {
         blob_recycler: BlobRecycler,
         ledger_path: &str,
         entry_receiver: Receiver<Vec<Entry>>,
-    ) -> (Self, BlobReceiver) {
+    ) -> (Self, Receiver<Vec<Entry>>) {
         let (vote_blob_sender, vote_blob_receiver) = channel();
         let send = UdpSocket::bind("0.0.0.0:0").expect("bind");
         let t_responder = responder(
@@ -82,7 +114,7 @@ impl WriteStage {
             blob_recycler.clone(),
             vote_blob_receiver,
         );
-        let (blob_sender, blob_receiver) = channel();
+        let (entry_sender, entry_receiver_forward) = channel();
         let mut ledger_writer = LedgerWriter::recover(ledger_path).unwrap();
 
         let thread_hdl = Builder::new()
@@ -94,10 +126,8 @@ impl WriteStage {
                 loop {
                     if let Err(e) = Self::write_and_send_entries(
                         &crdt,
-                        &bank,
                         &mut ledger_writer,
-                        &blob_sender,
-                        &blob_recycler,
+                        &entry_sender,
                         &entry_receiver,
                     ) {
                         match e {
@@ -129,7 +159,7 @@ impl WriteStage {
             }).unwrap();
 
         let thread_hdls = vec![t_responder, thread_hdl];
-        (WriteStage { thread_hdls }, blob_receiver)
+        (WriteStage { thread_hdls }, entry_receiver_forward)
     }
 }