Integrate cluster info voting into the TPU and ReplayStage

aeyakovenko · aeyakovenko · commit 91336a190340 · 2019-01-26T12:22:40.000-08:00
diff --git a/src/broadcast_service.rs b/src/broadcast_service.rs
@@ -227,7 +227,7 @@ impl BroadcastService {
             ) {
                 match e {
                     Error::RecvTimeoutError(RecvTimeoutError::Disconnected) => {
-                        return BroadcastServiceReturnType::ChannelDisconnected
+                        return BroadcastServiceReturnType::ChannelDisconnected;
                     }
                     Error::RecvTimeoutError(RecvTimeoutError::Timeout) => (),
                     Error::ClusterInfoError(ClusterInfoError::NoPeers) => (), // TODO: Why are the unit-tests throwing hundreds of these?
diff --git a/src/cluster_info.rs b/src/cluster_info.rs
@@ -56,7 +56,7 @@ pub const NEIGHBORHOOD_SIZE: usize = DATA_PLANE_FANOUT;
 pub const GROW_LAYER_CAPACITY: bool = false;
 
 /// milliseconds we sleep for between gossip requests
-const GOSSIP_SLEEP_MILLIS: u64 = 100;
+pub const GOSSIP_SLEEP_MILLIS: u64 = 100;
 
 #[derive(Debug, PartialEq, Eq)]
 pub enum ClusterInfoError {
diff --git a/src/cluster_info_vote_listener.rs b/src/cluster_info_vote_listener.rs
@@ -0,0 +1,70 @@
+use crate::cluster_info::{ClusterInfo, GOSSIP_SLEEP_MILLIS};
+use crate::counter::Counter;
+use crate::packet;
+use crate::result::Result;
+use crate::service::Service;
+use crate::streamer::PacketSender;
+use log::Level;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::{Arc, RwLock};
+use std::thread::{self, sleep, Builder, JoinHandle};
+use std::time::Duration;
+
+pub struct ClusterInfoVoteListener {
+    exit: Arc<AtomicBool>,
+    thread_hdls: Vec<JoinHandle<()>>,
+}
+
+impl ClusterInfoVoteListener {
+    pub fn new(
+        exit: Arc<AtomicBool>,
+        cluster_info: Arc<RwLock<ClusterInfo>>,
+        sender: PacketSender,
+    ) -> Self {
+        let exit1 = exit.clone();
+        let thread = Builder::new()
+            .name("solana-cluster_info_vote_listener".to_string())
+            .spawn(move || {
+                let _ = Self::recv_loop(&exit1, &cluster_info, &sender);
+            })
+            .unwrap();
+        Self {
+            exit,
+            thread_hdls: vec![thread],
+        }
+    }
+    fn recv_loop(
+        exit: &Arc<AtomicBool>,
+        cluster_info: &Arc<RwLock<ClusterInfo>>,
+        sender: &PacketSender,
+    ) -> Result<()> {
+        let mut last_ts = 0;
+        loop {
+            if exit.load(Ordering::Relaxed) {
+                return Ok(());
+            }
+            let (votes, new_ts) = cluster_info.read().unwrap().get_votes(last_ts);
+            last_ts = new_ts;
+            inc_new_counter_info!("cluster_info_vote_listener-recv_count", votes.len());
+            let msgs = packet::to_packets(&votes);
+            for m in msgs {
+                sender.send(m)?;
+            }
+            sleep(Duration::from_millis(GOSSIP_SLEEP_MILLIS));
+        }
+    }
+    pub fn close(&self) {
+        self.exit.store(true, Ordering::Relaxed);
+    }
+}
+
+impl Service for ClusterInfoVoteListener {
+    type JoinReturnType = ();
+
+    fn join(self) -> thread::Result<()> {
+        for thread_hdl in self.thread_hdls {
+            thread_hdl.join()?;
+        }
+        Ok(())
+    }
+}
diff --git a/src/fetch_stage.rs b/src/fetch_stage.rs
@@ -1,7 +1,7 @@
 //! The `fetch_stage` batches input from a UDP socket and sends it to a channel.
 
 use crate::service::Service;
-use crate::streamer::{self, PacketReceiver};
+use crate::streamer::{self, PacketReceiver, PacketSender};
 use std::net::UdpSocket;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc::channel;
@@ -16,20 +16,28 @@ pub struct FetchStage {
 impl FetchStage {
     #[allow(clippy::new_ret_no_self)]
     pub fn new(sockets: Vec<UdpSocket>, exit: Arc<AtomicBool>) -> (Self, PacketReceiver) {
+        let (sender, receiver) = channel();
+        (Self::new_with_sender(sockets, exit, sender), receiver)
+    }
+    pub fn new_with_sender(
+        sockets: Vec<UdpSocket>,
+        exit: Arc<AtomicBool>,
+        sender: PacketSender,
+    ) -> Self {
         let tx_sockets = sockets.into_iter().map(Arc::new).collect();
-        Self::new_multi_socket(tx_sockets, exit)
+        Self::new_multi_socket(tx_sockets, exit, sender)
     }
     fn new_multi_socket(
         sockets: Vec<Arc<UdpSocket>>,
         exit: Arc<AtomicBool>,
-    ) -> (Self, PacketReceiver) {
-        let (sender, receiver) = channel();
+        sender: PacketSender,
+    ) -> Self {
         let thread_hdls: Vec<_> = sockets
             .into_iter()
             .map(|socket| streamer::receiver(socket, exit.clone(), sender.clone(), "fetch-stage"))
             .collect();
 
-        (Self { exit, thread_hdls }, receiver)
+        Self { exit, thread_hdls }
     }
 
     pub fn close(&self) {
diff --git a/src/fullnode.rs b/src/fullnode.rs
@@ -326,6 +326,7 @@ impl Fullnode {
             // Start in leader mode.
             let (tpu, entry_receiver, tpu_exit) = Tpu::new(
                 &bank,
+                cluster_info.clone(),
                 Default::default(),
                 node.sockets
                     .tpu
@@ -483,6 +484,7 @@ impl Fullnode {
 
         let (tpu, blob_receiver, tpu_exit) = Tpu::new(
             &self.bank,
+            self.cluster_info.clone(),
             Default::default(),
             self.tpu_sockets
                 .iter()
diff --git a/src/lib.rs b/src/lib.rs
@@ -21,6 +21,7 @@ pub mod chacha;
 pub mod chacha_cuda;
 pub mod checkpoint;
 pub mod client;
+pub mod cluster_info_vote_listener;
 pub mod crds;
 pub mod crds_gossip;
 pub mod crds_gossip_error;
diff --git a/src/replay_stage.rs b/src/replay_stage.rs
@@ -11,13 +11,11 @@ use crate::leader_scheduler::TICKS_PER_BLOCK;
 use crate::packet::BlobError;
 use crate::result::{Error, Result};
 use crate::service::Service;
-use crate::streamer::{responder, BlobSender};
 use crate::vote_signer_proxy::VoteSignerProxy;
 use log::Level;
 use solana_metrics::{influxdb, submit};
 use solana_sdk::signature::{Keypair, KeypairUtil};
 use solana_sdk::timing::duration_as_ms;
-use std::net::UdpSocket;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::mpsc::channel;
 use std::sync::mpsc::RecvTimeoutError;
@@ -52,7 +50,6 @@ impl Drop for Finalizer {
 }
 
 pub struct ReplayStage {
-    t_responder: JoinHandle<()>,
     t_replay: JoinHandle<Option<ReplayStageReturnType>>,
 }
 
@@ -65,7 +62,6 @@ impl ReplayStage {
         window_receiver: &EntryReceiver,
         keypair: &Arc<Keypair>,
         vote_signer: Option<&Arc<VoteSignerProxy>>,
-        vote_blob_sender: Option<&BlobSender>,
         ledger_entry_sender: &EntrySender,
         entry_height: &mut u64,
         last_entry_id: &mut Hash,
@@ -138,11 +134,8 @@ impl ReplayStage {
 
                 if 0 == num_ticks_to_next_vote {
                     if let Some(signer) = vote_signer {
-                        if let Some(sender) = vote_blob_sender {
-                            signer
-                                .send_validator_vote(bank, &cluster_info, sender)
-                                .unwrap();
-                        }
+                        let vote = signer.validator_vote(bank);
+                        cluster_info.write().unwrap().push_vote(vote);
                     }
                 }
                 let (scheduled_leader, _) = bank
@@ -204,10 +197,7 @@ impl ReplayStage {
         entry_height: u64,
         last_entry_id: Hash,
     ) -> (Self, EntryReceiver) {
-        let (vote_blob_sender, vote_blob_receiver) = channel();
         let (ledger_entry_sender, ledger_entry_receiver) = channel();
-        let send = UdpSocket::bind("0.0.0.0:0").expect("bind");
-        let t_responder = responder("replay_stage", Arc::new(send), vote_blob_receiver);
 
         let keypair = Arc::new(keypair);
         let t_replay = Builder::new()
@@ -243,7 +233,6 @@ impl ReplayStage {
                         &window_receiver,
                         &keypair,
                         vote_signer.as_ref(),
-                        Some(&vote_blob_sender),
                         &ledger_entry_sender,
                         &mut entry_height_,
                         &mut last_entry_id,
@@ -259,21 +248,14 @@ impl ReplayStage {
             })
             .unwrap();
 
-        (
-            Self {
-                t_responder,
-                t_replay,
-            },
-            ledger_entry_receiver,
-        )
+        (Self { t_replay }, ledger_entry_receiver)
     }
 }
 
 impl Service for ReplayStage {
     type JoinReturnType = Option<ReplayStageReturnType>;
 
     fn join(self) -> thread::Result<Option<ReplayStageReturnType>> {
-        self.t_responder.join()?;
         self.t_replay.join()
     }
 }
@@ -486,10 +468,8 @@ mod test {
             last_entry_id,
         );
 
-        // Vote sender should error because no leader contact info is found in the
-        // ClusterInfo
-        let (mock_sender, _mock_receiver) = channel();
-        let _vote_err = vote_signer.send_validator_vote(&bank, &cluster_info_me, &mock_sender);
+        let vote = vote_signer.validator_vote(&bank);
+        cluster_info_me.write().unwrap().push_vote(vote);
 
         // Send ReplayStage an entry, should see it on the ledger writer receiver
         let next_tick = create_ticks(
@@ -602,10 +582,8 @@ mod test {
             last_entry_id,
         );
 
-        // Vote sender should error because no leader contact info is found in the
-        // ClusterInfo
-        let (mock_sender, _mock_receiver) = channel();
-        let _vote_err = signer_proxy.send_validator_vote(&bank, &cluster_info_me, &mock_sender);
+        let vote = signer_proxy.validator_vote(&bank);
+        cluster_info_me.write().unwrap().push_vote(vote);
 
         // Send enough ticks to trigger leader rotation
         let total_entries_to_send = (bootstrap_height - initial_tick_height) as usize;
@@ -685,7 +663,6 @@ mod test {
             &entry_receiver,
             &my_keypair,
             Some(&vote_signer),
-            None,
             &ledger_entry_sender,
             &mut entry_height,
             &mut last_entry_id,
@@ -711,7 +688,6 @@ mod test {
             &entry_receiver,
             &Arc::new(Keypair::new()),
             Some(&vote_signer),
-            None,
             &ledger_entry_sender,
             &mut entry_height,
             &mut last_entry_id,
diff --git a/src/tpu.rs b/src/tpu.rs
@@ -3,6 +3,8 @@
 
 use crate::bank::Bank;
 use crate::banking_stage::{BankingStage, BankingStageReturnType};
+use crate::cluster_info::ClusterInfo;
+use crate::cluster_info_vote_listener::ClusterInfoVoteListener;
 use crate::entry::Entry;
 use crate::fetch_stage::FetchStage;
 use crate::poh_service::Config;
@@ -12,8 +14,9 @@ use solana_sdk::hash::Hash;
 use solana_sdk::pubkey::Pubkey;
 use std::net::UdpSocket;
 use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::mpsc::channel;
 use std::sync::mpsc::Receiver;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 use std::thread;
 
 pub enum TpuReturnType {
@@ -24,13 +27,15 @@ pub struct Tpu {
     fetch_stage: FetchStage,
     sigverify_stage: SigVerifyStage,
     banking_stage: BankingStage,
+    cluster_info_vote_listener: ClusterInfoVoteListener,
     exit: Arc<AtomicBool>,
 }
 
 impl Tpu {
     #[allow(clippy::new_ret_no_self)]
     pub fn new(
         bank: &Arc<Bank>,
+        cluster_info: Arc<RwLock<ClusterInfo>>,
         tick_duration: Config,
         transactions_sockets: Vec<UdpSocket>,
         sigverify_disabled: bool,
@@ -40,7 +45,11 @@ impl Tpu {
     ) -> (Self, Receiver<Vec<Entry>>, Arc<AtomicBool>) {
         let exit = Arc::new(AtomicBool::new(false));
 
-        let (fetch_stage, packet_receiver) = FetchStage::new(transactions_sockets, exit.clone());
+        let (packet_sender, packet_receiver) = channel();
+        let fetch_stage =
+            FetchStage::new_with_sender(transactions_sockets, exit.clone(), packet_sender.clone());
+        let cluster_info_vote_listener =
+            ClusterInfoVoteListener::new(exit.clone(), cluster_info, packet_sender);
 
         let (sigverify_stage, verified_receiver) =
             SigVerifyStage::new(packet_receiver, sigverify_disabled);
@@ -58,6 +67,7 @@ impl Tpu {
             fetch_stage,
             sigverify_stage,
             banking_stage,
+            cluster_info_vote_listener,
             exit: exit.clone(),
         };
 
@@ -84,6 +94,7 @@ impl Service for Tpu {
     fn join(self) -> thread::Result<(Option<TpuReturnType>)> {
         self.fetch_stage.join()?;
         self.sigverify_stage.join()?;
+        self.cluster_info_vote_listener.join()?;
         match self.banking_stage.join()? {
             Some(BankingStageReturnType::LeaderRotation) => Ok(Some(TpuReturnType::LeaderRotation)),
             _ => Ok(None),
diff --git a/src/vote_signer_proxy.rs b/src/vote_signer_proxy.rs

Original file line number	Diff line number	Diff line change
`@@ -227,7 +227,7 @@ impl BroadcastService {`
`227`	`227`	`) {`
`228`	`228`	`match e {`
`229`	`229`	`Error::RecvTimeoutError(RecvTimeoutError::Disconnected) => {`
`230`		`- return BroadcastServiceReturnType::ChannelDisconnected`
	`230`	`+ return BroadcastServiceReturnType::ChannelDisconnected;`
`231`	`231`	`}`
`232`	`232`	`Error::RecvTimeoutError(RecvTimeoutError::Timeout) => (),`
`233`	`233`	`Error::ClusterInfoError(ClusterInfoError::NoPeers) => (), // TODO: Why are the unit-tests throwing hundreds of these?`