Add get_finality request and use it from multinode test (#941)

src/bank.rs

@@ -15,6 +15,7 @@ use log::Level;
 use mint::Mint;
 use payment_plan::{Payment, PaymentPlan, Witness};
 use signature::{Keypair, Pubkey, Signature};
+use std;
 use std::collections::hash_map::Entry::Occupied;
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::result;
@@ -91,6 +92,9 @@ pub struct Bank {
     /// This bool allows us to submit metrics that are specific for leaders or validators
     /// It is set to `true` by fullnode before creating the bank.
     pub is_leader: bool,
+
+    // The latest finality time for the network
+    finality_time: AtomicUsize,
 }
 
 impl Default for Bank {
@@ -102,6 +106,7 @@ impl Default for Bank {
             last_ids_sigs: RwLock::new(HashMap::new()),
             transaction_count: AtomicUsize::new(0),
             is_leader: true,
+            finality_time: AtomicUsize::new(std::usize::MAX),
         }
     }
 }
@@ -582,6 +587,14 @@ impl Bank {
         }
         false
     }
+
+    pub fn finality(&self) -> usize {
+        self.finality_time.load(Ordering::Relaxed)
+    }
+
+    pub fn set_finality(&self, finality: usize) {
+        self.finality_time.store(finality, Ordering::Relaxed);
+    }
 }
 
 #[cfg(test)]
@@ -595,6 +608,7 @@ mod tests {
     use ledger;
     use packet::BLOB_DATA_SIZE;
     use signature::KeypairUtil;
+    use std;
     use std::io::{BufReader, Cursor, Seek, SeekFrom};
     use std::mem::size_of;
 
@@ -982,5 +996,12 @@ mod tests {
         let validator_bank = Bank::new_default(false);
         assert!(!validator_bank.is_leader);
     }
+    #[test]
+    fn test_finality() {
+        let def_bank = Bank::default();
+        assert_eq!(def_bank.finality(), std::usize::MAX);
+        def_bank.set_finality(90);
+        assert_eq!(def_bank.finality(), 90);
+    }
 
 }

src/request.rs

@@ -10,6 +10,7 @@ pub enum Request {
     GetLastId,
     GetTransactionCount,
     GetSignature { signature: Signature },
+    GetFinality,
 }
 
 impl Request {
@@ -25,4 +26,5 @@ pub enum Response {
     LastId { id: Hash },
     TransactionCount { transaction_count: u64 },
     SignatureStatus { signature_status: bool },
+    Finality { time: usize },
 }

src/request_processor.rs

@@ -46,6 +46,12 @@ impl RequestProcessor {
                 info!("Response::Signature {:?}", rsp);
                 Some(rsp)
             }
+            Request::GetFinality => {
+                let time = self.bank.finality();
+                let rsp = (Response::Finality { time }, rsp_addr);
+                info!("Response::Finality {:?}", rsp);
+                Some(rsp)
+            }
         }
     }
 

src/thin_client.rs

@@ -29,6 +29,7 @@ pub struct ThinClient {
     transaction_count: u64,
     balances: HashMap<Pubkey, i64>,
     signature_status: bool,
+    finality: Option<usize>,
 }
 
 impl ThinClient {
@@ -50,6 +51,7 @@ impl ThinClient {
             transaction_count: 0,
             balances: HashMap::new(),
             signature_status: false,
+            finality: None,
         }
     }
 
@@ -83,6 +85,10 @@ impl ThinClient {
                     trace!("Response signature not found");
                 }
             }
+            Response::Finality { time } => {
+                trace!("Response finality {:?}", time);
+                self.finality = Some(time);
+            }
         }
     }
 
@@ -143,6 +149,33 @@ impl ThinClient {
             .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "nokey"))
     }
 
+    /// Request the finality from the leader node
+    pub fn get_finality(&mut self) -> usize {
+        trace!("get_finality");
+        let req = Request::GetFinality;
+        let data = serialize(&req).expect("serialize GetFinality in pub fn get_finality");
+        let mut done = false;
+        while !done {
+            debug!("get_finality send_to {}", &self.requests_addr);
+            self.requests_socket
+                .send_to(&data, &self.requests_addr)
+                .expect("buffer error in pub fn get_finality");
+
+            match self.recv_response() {
+                Ok(resp) => {
+                    if let Response::Finality { .. } = resp {
+                        done = true;
+                    }
+                    self.process_response(&resp);
+                }
+                Err(e) => {
+                    debug!("thin_client get_finality error: {}", e);
+                }
+            }
+        }
+        self.finality.expect("some finality")
+    }
+
     /// Request the transaction count.  If the response packet is dropped by the network,
     /// this method will hang.
     pub fn transaction_count(&mut self) -> u64 {

src/vote_stage.rs

@@ -152,6 +152,8 @@ pub fn send_leader_vote(
                 );
                 inc_new_counter_info!("vote_stage-leader_sent_vote", 1);
 
+                bank.set_finality((now - *last_valid_validator_timestamp) as usize);
+
                 metrics::submit(
                     influxdb::Point::new(&"leader-finality")
                         .add_field("duration_ms", influxdb::Value::Integer(finality_ms as i64))

tests/multinode.rs

@@ -695,12 +695,13 @@ fn test_multi_node_dynamic_network() {
                 info!("{} nodes are lagging behind leader", num_nodes_behind);
             }
             info!(
-                "SUCCESS[{}] {} out of {} distance: {} max_distance: {}",
+                "SUCCESS[{}] {} out of {} distance: {} max_distance: {}  finality: {}",
                 i,
                 success,
                 validators.len(),
                 total_distance,
-                max_distance
+                max_distance,
+                get_finality(&leader_data)
             );
             if success == validators.len() && total_distance == 0 {
                 consecutive_success += 1;
@@ -820,3 +821,9 @@ fn retry_send_tx_and_retry_get_balance(
     }
     None
 }
+
+fn get_finality(leader: &NodeInfo) -> usize {
+    let mut client = mk_client(leader);
+    trace!("getting leader finality");
+    client.get_finality()
+}