Make tpu own banking tracer thread

Author: ryoqun

banking-bench/src/main.rs

@@ -449,7 +449,6 @@ fn main() {
             None,
             Arc::new(connection_cache),
             bank_forks.clone(),
-            banking_tracer,
         );
         poh_recorder.write().unwrap().set_bank(&bank, false);
 

core/benches/banking_stage.rs

@@ -289,7 +289,6 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
             None,
             Arc::new(ConnectionCache::default()),
             bank_forks,
-            banking_tracer,
         );
         poh_recorder.write().unwrap().set_bank(&bank, false);
 

core/src/banking_stage.rs

@@ -4,7 +4,7 @@
 
 use {
     crate::{
-        banking_trace::{BankingPacketReceiver, BankingTracer, TracerThread},
+        banking_trace::BankingPacketReceiver,
         forward_packet_batches_by_accounts::ForwardPacketBatchesByAccounts,
         immutable_deserialized_packet::ImmutableDeserializedPacket,
         latest_unprocessed_votes::{LatestUnprocessedVotes, VoteSource},
@@ -346,7 +346,6 @@ pub struct BatchedTransactionErrorDetails {
 /// Stores the stage's thread handle and output receiver.
 pub struct BankingStage {
     bank_thread_hdls: Vec<JoinHandle<()>>,
-    tracer_thread_hdl: TracerThread,
 }
 
 #[derive(Debug, Clone)]
@@ -387,7 +386,6 @@ impl BankingStage {
         log_messages_bytes_limit: Option<usize>,
         connection_cache: Arc<ConnectionCache>,
         bank_forks: Arc<RwLock<BankForks>>,
-        banking_tracer: Arc<BankingTracer>,
     ) -> Self {
         Self::new_num_threads(
             cluster_info,
@@ -401,7 +399,6 @@ impl BankingStage {
             log_messages_bytes_limit,
             connection_cache,
             bank_forks,
-            banking_tracer,
         )
     }
 
@@ -418,7 +415,6 @@ impl BankingStage {
         log_messages_bytes_limit: Option<usize>,
         connection_cache: Arc<ConnectionCache>,
         bank_forks: Arc<RwLock<BankForks>>,
-        banking_tracer: Arc<BankingTracer>,
     ) -> Self {
         assert!(num_threads >= MIN_TOTAL_THREADS);
         // Single thread to generate entries from many banks.
@@ -510,10 +506,7 @@ impl BankingStage {
             })
             .collect();
 
-        Self {
-            bank_thread_hdls,
-            tracer_thread_hdl: banking_tracer.take_tracer_thread_join_handle(),
-        }
+        Self { bank_thread_hdls }
     }
 
     /// Forwards all valid, unprocessed packets in the buffer, up to a rate limit. Returns
@@ -1908,11 +1901,6 @@ impl BankingStage {
         for bank_thread_hdl in self.bank_thread_hdls {
             bank_thread_hdl.join()?;
         }
-        if let Some(tracer_thread_hdl) = self.tracer_thread_hdl {
-            if let Err(tracer_result) = tracer_thread_hdl.join()? {
-                error!("tracer thread error: {:?}", tracer_result);
-            }
-        }
         Ok(())
     }
 }
@@ -1963,7 +1951,10 @@ where
 mod tests {
     use {
         super::*,
-        crate::{banking_trace::BankingPacketBatch, unprocessed_packet_batches},
+        crate::{
+            banking_trace::{BankingPacketBatch, BankingTracer},
+            unprocessed_packet_batches,
+        },
         crossbeam_channel::{unbounded, Receiver},
         solana_address_lookup_table_program::state::{AddressLookupTable, LookupTableMeta},
         solana_entry::entry::{next_entry, next_versioned_entry, Entry, EntrySlice},
@@ -2051,7 +2042,6 @@ mod tests {
                 None,
                 Arc::new(ConnectionCache::default()),
                 bank_forks,
-                banking_tracer,
             );
             drop(non_vote_sender);
             drop(tpu_vote_sender);
@@ -2107,7 +2097,6 @@ mod tests {
                 None,
                 Arc::new(ConnectionCache::default()),
                 bank_forks,
-                banking_tracer,
             );
             trace!("sending bank");
             drop(non_vote_sender);
@@ -2188,7 +2177,6 @@ mod tests {
                 None,
                 Arc::new(ConnectionCache::default()),
                 bank_forks,
-                banking_tracer,
             );
 
             // fund another account so we can send 2 good transactions in a single batch.
@@ -2350,7 +2338,6 @@ mod tests {
                     None,
                     Arc::new(ConnectionCache::default()),
                     bank_forks,
-                    banking_tracer,
                 );
 
                 // wait for banking_stage to eat the packets
@@ -4127,7 +4114,6 @@ mod tests {
                 None,
                 Arc::new(ConnectionCache::default()),
                 bank_forks,
-                banking_tracer,
             );
 
             let keypairs = (0..100).map(|_| Keypair::new()).collect_vec();

core/src/tpu.rs

@@ -4,7 +4,7 @@
 use {
     crate::{
         banking_stage::BankingStage,
-        banking_trace::BankingTracer,
+        banking_trace::{BankingTracer, TracerThread},
         broadcast_stage::{BroadcastStage, BroadcastStageType, RetransmitSlotsReceiver},
         cluster_info_vote_listener::{
             ClusterInfoVoteListener, GossipDuplicateConfirmedSlotsSender,
@@ -69,6 +69,7 @@ pub struct Tpu {
     find_packet_sender_stake_stage: FindPacketSenderStakeStage,
     vote_find_packet_sender_stake_stage: FindPacketSenderStakeStage,
     staked_nodes_updater_service: StakedNodesUpdaterService,
+    tracer_thread_hdl: TracerThread,
 }
 
 impl Tpu {
@@ -236,7 +237,6 @@ impl Tpu {
             log_messages_bytes_limit,
             connection_cache.clone(),
             bank_forks.clone(),
-            banking_tracer,
         );
 
         let broadcast_stage = broadcast_type.new_broadcast_stage(
@@ -262,6 +262,7 @@ impl Tpu {
             find_packet_sender_stake_stage,
             vote_find_packet_sender_stake_stage,
             staked_nodes_updater_service,
+            tracer_thread_hdl: banking_tracer.take_tracer_thread_join_handle(),
         }
     }
 
@@ -283,6 +284,14 @@ impl Tpu {
             result?;
         }
         let _ = broadcast_result?;
+        if let Some(tracer_thread_hdl) = self.tracer_thread_hdl {
+            if let Err(tracer_result) = tracer_thread_hdl.join()? {
+                error!(
+                    "banking tracer thread returned error after successful thread join: {:?}",
+                    tracer_result
+                );
+            }
+        }
         Ok(())
     }
 }