From 1919ec247b708be82cee9b0a76f2f20010ab841b Mon Sep 17 00:00:00 2001
From: Rob Walker <rob@solana.com>
Date: Mon, 25 Jun 2018 15:07:48 -0700
Subject: [PATCH] add a clock to validator windows (part 3 of #309) (#448)

* count entries processed by Bank
 * initialize windows with initial height of Entries
---
 src/bank.rs         | 15 +++++++++++++--
 src/bin/fullnode.rs |  3 ++-
 src/streamer.rs     |  8 +++++---
 src/tvu.rs          |  1 +
 src/window_stage.rs |  2 ++
 5 files changed, 23 insertions(+), 6 deletions(-)

diff --git a/src/bank.rs b/src/bank.rs
index 26fd7d6466..7358b69a1d 100644
--- a/src/bank.rs
+++ b/src/bank.rs
@@ -87,6 +87,11 @@ pub struct Bank {
     /// The number of transactions the bank has processed without error since the
     /// start of the ledger.
     transaction_count: AtomicUsize,
+
+    /// The number of Entries the bank has processed without error since start
+    ///   of the ledger, i.e. poor-man's network synchronization
+    ///   TODO: upgrade to U64 when stable?
+    entry_count: AtomicUsize,
 }
 
 impl Bank {
@@ -100,6 +105,7 @@ impl Bank {
             time_sources: RwLock::new(HashSet::new()),
             last_time: RwLock::new(Utc.timestamp(0, 0)),
             transaction_count: AtomicUsize::new(0),
+            entry_count: AtomicUsize::new(0),
         };
         bank.apply_payment(deposit, &mut bank.balances.write().unwrap());
         bank
@@ -296,11 +302,13 @@ impl Bank {
     }
 
     /// Process an ordered list of entries.
-    pub fn process_entries<I>(&self, entries: I) -> Result<()>
+    pub fn process_entries<I>(&self, entries: I) -> Result<usize>
     where
         I: IntoIterator<Item = Entry>,
     {
         for entry in entries {
+            self.entry_count.fetch_add(1, Ordering::Relaxed);
+
             if !entry.transactions.is_empty() {
                 for result in self.process_transactions(entry.transactions) {
                     result?;
@@ -308,7 +316,7 @@ impl Bank {
             }
             self.register_entry_id(&entry.id);
         }
-        Ok(())
+        Ok(self.entry_count())
     }
 
     /// Process a Witness Signature. Any payment plans waiting on this signature
@@ -422,6 +430,9 @@ impl Bank {
     pub fn transaction_count(&self) -> usize {
         self.transaction_count.load(Ordering::Relaxed)
     }
+    pub fn entry_count(&self) -> usize {
+        self.entry_count.load(Ordering::Relaxed)
+    }
 }
 
 #[cfg(test)]
diff --git a/src/bin/fullnode.rs b/src/bin/fullnode.rs
index b1acf1bdd4..b4f4dbe351 100644
--- a/src/bin/fullnode.rs
+++ b/src/bin/fullnode.rs
@@ -102,7 +102,8 @@ fn main() {
     bank.register_entry_id(&entry1.id);
 
     eprintln!("processing entries...");
-    bank.process_entries(entries).expect("process_entries");
+    let num_entries = bank.process_entries(entries).expect("process_entries");
+    eprintln!("processed {} entries...", num_entries);
 
     eprintln!("creating networking stack...");
 
diff --git a/src/streamer.rs b/src/streamer.rs
index 072fc39d90..9c4e6cb0eb 100644
--- a/src/streamer.rs
+++ b/src/streamer.rs
@@ -395,13 +395,14 @@ pub fn window(
     r: BlobReceiver,
     s: BlobSender,
     retransmit: BlobSender,
+    entry_count: usize,
 ) -> JoinHandle<()> {
     Builder::new()
         .name("solana-window".to_string())
         .spawn(move || {
-            let mut consumed = 0;
-            let mut received = 0;
-            let mut last = 0;
+            let mut consumed = entry_count;
+            let mut received = entry_count;
+            let mut last = entry_count;
             let mut times = 0;
             loop {
                 if exit.load(Ordering::Relaxed) {
@@ -816,6 +817,7 @@ mod test {
             r_reader,
             s_window,
             s_retransmit,
+            0,
         );
         let (s_responder, r_responder) = channel();
         let t_responder = responder(
diff --git a/src/tvu.rs b/src/tvu.rs
index f338d0b1e7..014beb06ec 100644
--- a/src/tvu.rs
+++ b/src/tvu.rs
@@ -86,6 +86,7 @@ impl Tvu {
             exit.clone(),
             blob_recycler.clone(),
             fetch_stage.blob_receiver,
+            bank.entry_count(),
         );
 
         let replicate_stage =
diff --git a/src/window_stage.rs b/src/window_stage.rs
index 36264a88d2..4cc7eceb68 100644
--- a/src/window_stage.rs
+++ b/src/window_stage.rs
@@ -22,6 +22,7 @@ impl WindowStage {
         exit: Arc<AtomicBool>,
         blob_recycler: packet::BlobRecycler,
         fetch_stage_receiver: streamer::BlobReceiver,
+        entry_count: usize,
     ) -> Self {
         let (retransmit_sender, retransmit_receiver) = channel();
 
@@ -41,6 +42,7 @@ impl WindowStage {
             fetch_stage_receiver,
             blob_sender,
             retransmit_sender,
+            entry_count,
         );
         let thread_hdls = vec![t_retransmit, t_window];