Add replicator support to multinode demo (#4221)

automerge

core/src/gossip_service.rs

@@ -7,6 +7,7 @@ use crate::cluster_info::FULLNODE_PORT_RANGE;
 use crate::contact_info::ContactInfo;
 use crate::service::Service;
 use crate::streamer;
+use rand::{thread_rng, Rng};
 use solana_client::thin_client::{create_client, ThinClient};
 use solana_sdk::pubkey::Pubkey;
 use solana_sdk::signature::{Keypair, KeypairUtil};
@@ -107,6 +108,7 @@ pub fn discover(
     ))
 }
 
+/// Creates a ThinClient per valid node
 pub fn get_clients(nodes: &[ContactInfo]) -> Vec<ThinClient> {
     nodes
         .iter()
@@ -115,6 +117,16 @@ pub fn get_clients(nodes: &[ContactInfo]) -> Vec<ThinClient> {
         .collect()
 }
 
+/// Creates a ThinClient by selecting a valid node at random
+pub fn get_client(nodes: &[ContactInfo]) -> ThinClient {
+    let nodes: Vec<_> = nodes
+        .iter()
+        .filter_map(ContactInfo::valid_client_facing_addr)
+        .collect();
+    let select = thread_rng().gen_range(0, nodes.len());
+    create_client(nodes[select], FULLNODE_PORT_RANGE)
+}
+
 fn spy(
     spy_ref: Arc<RwLock<ClusterInfo>>,
     num_nodes: Option<usize>,

core/src/replicator.rs

@@ -206,8 +206,9 @@ impl Replicator {
             &exit,
         );
 
-        info!("Looking for leader at {:?}", cluster_entrypoint);
-        crate::gossip_service::discover_nodes(&cluster_entrypoint.gossip, 1)?;
+        info!("Connecting to the cluster via {:?}", cluster_entrypoint);
+        let nodes = crate::gossip_service::discover_nodes(&cluster_entrypoint.gossip, 1)?;
+        let client = crate::gossip_service::get_client(&nodes);
 
         let (storage_blockhash, storage_slot) = Self::poll_for_blockhash_and_slot(&cluster_info)?;
 
@@ -242,30 +243,33 @@ impl Replicator {
             &Hash::default(),
         );
 
-        let client = create_client(cluster_entrypoint.client_facing_addr(), FULLNODE_PORT_RANGE);
         Self::setup_mining_account(&client, &keypair, &storage_keypair)?;
-
         let mut thread_handles =
             create_request_processor(node.sockets.storage.unwrap(), &exit, slot);
 
         // receive blobs from retransmit and drop them.
-        let exit2 = exit.clone();
-        let t_retransmit = spawn(move || loop {
-            let _ = retransmit_receiver.recv_timeout(Duration::from_secs(1));
-            if exit2.load(Ordering::Relaxed) {
-                break;
-            }
-        });
+        let t_retransmit = {
+            let exit = exit.clone();
+            spawn(move || loop {
+                let _ = retransmit_receiver.recv_timeout(Duration::from_secs(1));
+                if exit.load(Ordering::Relaxed) {
+                    break;
+                }
+            })
+        };
         thread_handles.push(t_retransmit);
 
-        let exit3 = exit.clone();
-        let blocktree1 = blocktree.clone();
-        let t_replicate = spawn(move || loop {
-            Self::wait_for_ledger_download(slot, &blocktree1, &exit3, &node_info, &cluster_info);
-            if exit3.load(Ordering::Relaxed) {
-                break;
-            }
-        });
+        let t_replicate = {
+            let exit = exit.clone();
+            let blocktree = blocktree.clone();
+            spawn(move || loop {
+                Self::wait_for_ledger_download(slot, &blocktree, &exit, &node_info, &cluster_info);
+                if exit.load(Ordering::Relaxed) {
+                    break;
+                }
+            })
+        };
+        //always push this last
         thread_handles.push(t_replicate);
 
         Ok(Self {
@@ -291,6 +295,8 @@ impl Replicator {
     }
 
     pub fn run(&mut self) {
+        info!("waiting for ledger download");
+        self.thread_handles.pop().unwrap().join().unwrap();
         self.encrypt_ledger()
             .expect("ledger encrypt not successful");
         loop {
@@ -310,7 +316,7 @@ impl Replicator {
         node_info: &ContactInfo,
         cluster_info: &Arc<RwLock<ClusterInfo>>,
     ) {
-        info!("window created, waiting for ledger download done");
+        info!("window created, waiting for ledger download");
         let mut _received_so_far = 0;
 
         let mut current_slot = start_slot;