Pass Arc<AtomicBool> by value, not by reference. (solana-labs#31916)

`Arc` is already a reference internally, so it does not seem to be
beneficial to pass a reference to it.  Just adds an extra layer of
indirection.

Functions that need to be able to increment `Arc` reference count need
to take `Arc<AtomicBool>`, but those that just want to read the
`AtomicBool` value can accept `&AtomicBool`, making them a bit more
generic.

This change focuses specifically on `Arc<AtomicBool>`.  There are other
uses of `&Arc<T>` in the code base that could be converted in a similar
manner.  But it would make the change even larger.

bench-tps/src/bench.rs

@@ -248,21 +248,20 @@ where
 
 fn create_sampler_thread<T>(
     client: &Arc<T>,
-    exit_signal: &Arc<AtomicBool>,
+    exit_signal: Arc<AtomicBool>,
     sample_period: u64,
     maxes: &Arc<RwLock<Vec<(String, SampleStats)>>>,
 ) -> JoinHandle<()>
 where
     T: 'static + BenchTpsClient + Send + Sync + ?Sized,
 {
     info!("Sampling TPS every {} second...", sample_period);
-    let exit_signal = exit_signal.clone();
     let maxes = maxes.clone();
     let client = client.clone();
     Builder::new()
         .name("solana-client-sample".to_string())
         .spawn(move || {
-            sample_txs(&exit_signal, &maxes, sample_period, &client);
+            sample_txs(exit_signal, &maxes, sample_period, &client);
         })
         .unwrap()
 }
@@ -325,7 +324,7 @@ fn create_sender_threads<T>(
     thread_batch_sleep_ms: usize,
     total_tx_sent_count: &Arc<AtomicUsize>,
     threads: usize,
-    exit_signal: &Arc<AtomicBool>,
+    exit_signal: Arc<AtomicBool>,
     shared_tx_active_thread_count: &Arc<AtomicIsize>,
 ) -> Vec<JoinHandle<()>>
 where
@@ -407,7 +406,7 @@ where
     // collect the max transaction rate and total tx count seen
     let maxes = Arc::new(RwLock::new(Vec::new()));
     let sample_period = 1; // in seconds
-    let sample_thread = create_sampler_thread(&client, &exit_signal, sample_period, &maxes);
+    let sample_thread = create_sampler_thread(&client, exit_signal.clone(), sample_period, &maxes);
 
     let shared_txs: SharedTransactions = Arc::new(RwLock::new(VecDeque::new()));
 
@@ -439,7 +438,7 @@ where
         thread_batch_sleep_ms,
         &total_tx_sent_count,
         threads,
-        &exit_signal,
+        exit_signal.clone(),
         &shared_tx_active_thread_count,
     );
 
@@ -786,7 +785,7 @@ fn get_new_latest_blockhash<T: BenchTpsClient + ?Sized>(
 }
 
 fn poll_blockhash<T: BenchTpsClient + ?Sized>(
-    exit_signal: &Arc<AtomicBool>,
+    exit_signal: &AtomicBool,
     blockhash: &Arc<RwLock<Hash>>,
     client: &Arc<T>,
     id: &Pubkey,
@@ -836,7 +835,7 @@ fn poll_blockhash<T: BenchTpsClient + ?Sized>(
 }
 
 fn do_tx_transfers<T: BenchTpsClient + ?Sized>(
-    exit_signal: &Arc<AtomicBool>,
+    exit_signal: &AtomicBool,
     shared_txs: &SharedTransactions,
     shared_tx_thread_count: &Arc<AtomicIsize>,
     total_tx_sent_count: &Arc<AtomicUsize>,

bench-tps/src/perf_utils.rs

@@ -23,7 +23,7 @@ pub struct SampleStats {
 }
 
 pub fn sample_txs<T>(
-    exit_signal: &Arc<AtomicBool>,
+    exit_signal: Arc<AtomicBool>,
     sample_stats: &Arc<RwLock<Vec<(String, SampleStats)>>>,
     sample_period: u64,
     client: &Arc<T>,

client-test/tests/client.rs

@@ -142,7 +142,7 @@ fn test_account_subscription() {
     let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
     let max_complete_rewards_slot = Arc::new(AtomicU64::default());
     let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
-        &exit,
+        exit.clone(),
         max_complete_transaction_status_slot,
         max_complete_rewards_slot,
         bank_forks.clone(),
@@ -261,7 +261,7 @@ fn test_block_subscription() {
     let max_complete_rewards_slot = Arc::new(AtomicU64::default());
     // setup RpcSubscriptions && PubSubService
     let subscriptions = Arc::new(RpcSubscriptions::new_for_tests_with_blockstore(
-        &exit,
+        exit.clone(),
         max_complete_transaction_status_slot,
         max_complete_rewards_slot,
         blockstore.clone(),
@@ -348,7 +348,7 @@ fn test_program_subscription() {
     let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
     let max_complete_rewards_slot = Arc::new(AtomicU64::default());
     let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
-        &exit,
+        exit.clone(),
         max_complete_transaction_status_slot,
         max_complete_rewards_slot,
         bank_forks.clone(),
@@ -434,7 +434,7 @@ fn test_root_subscription() {
     let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
     let max_complete_rewards_slot = Arc::new(AtomicU64::default());
     let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
-        &exit,
+        exit.clone(),
         max_complete_transaction_status_slot,
         max_complete_rewards_slot,
         bank_forks.clone(),
@@ -485,7 +485,7 @@ fn test_slot_subscription() {
     let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
     let max_complete_rewards_slot = Arc::new(AtomicU64::default());
     let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
-        &exit,
+        exit.clone(),
         max_complete_transaction_status_slot,
         max_complete_rewards_slot,
         bank_forks,
@@ -561,7 +561,7 @@ async fn test_slot_subscription_async() {
         let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
         let max_complete_rewards_slot = Arc::new(AtomicU64::default());
         let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
-            &exit,
+            exit.clone(),
             max_complete_transaction_status_slot,
             max_complete_rewards_slot,
             bank_forks,

client/src/transaction_executor.rs

@@ -51,7 +51,7 @@ impl TransactionExecutor {
         let sigs = Arc::new(RwLock::new(Vec::new()));
         let cleared = Arc::new(RwLock::new(Vec::new()));
         let exit = Arc::new(AtomicBool::new(false));
-        let sig_clear_t = Self::start_sig_clear_thread(&exit, &sigs, &cleared, &client);
+        let sig_clear_t = Self::start_sig_clear_thread(exit.clone(), &sigs, &cleared, &client);
         Self {
             sigs,
             cleared,
@@ -96,13 +96,12 @@ impl TransactionExecutor {
     }
 
     fn start_sig_clear_thread(
-        exit: &Arc<AtomicBool>,
+        exit: Arc<AtomicBool>,
         sigs: &Arc<RwLock<PendingQueue>>,
         cleared: &Arc<RwLock<Vec<u64>>>,
         client: &Arc<RpcClient>,
     ) -> JoinHandle<()> {
         let sigs = sigs.clone();
-        let exit = exit.clone();
         let cleared = cleared.clone();
         let client = client.clone();
         Builder::new()

core/src/banking_stage/consumer.rs

@@ -1611,7 +1611,7 @@ mod tests {
                 None,
                 blockstore.clone(),
                 false,
-                &Arc::new(AtomicBool::new(false)),
+                Arc::new(AtomicBool::new(false)),
             );
 
             let (replay_vote_sender, _replay_vote_receiver) = unbounded();
@@ -1749,7 +1749,7 @@ mod tests {
                 None,
                 blockstore.clone(),
                 false,
-                &Arc::new(AtomicBool::new(false)),
+                Arc::new(AtomicBool::new(false)),
             );
 
             let (replay_vote_sender, _replay_vote_receiver) = unbounded();

core/src/cache_block_meta_service.rs

@@ -26,9 +26,8 @@ impl CacheBlockMetaService {
     pub fn new(
         cache_block_meta_receiver: CacheBlockMetaReceiver,
         blockstore: Arc<Blockstore>,
-        exit: &Arc<AtomicBool>,
+        exit: Arc<AtomicBool>,
     ) -> Self {
-        let exit = exit.clone();
         let thread_hdl = Builder::new()
             .name("solCacheBlkTime".to_string())
             .spawn(move || loop {

core/src/cluster_info_vote_listener.rs

@@ -264,18 +264,20 @@ impl ClusterInfoVoteListener {
                 })
                 .unwrap()
         };
-        let exit_ = exit.clone();
-        let bank_send_thread = Builder::new()
-            .name("solCiBankSend".to_string())
-            .spawn(move || {
-                let _ = Self::bank_send_loop(
-                    exit_,
-                    verified_vote_label_packets_receiver,
-                    poh_recorder,
-                    &verified_packets_sender,
-                );
-            })
-            .unwrap();
+        let bank_send_thread = {
+            let exit = exit.clone();
+            Builder::new()
+                .name("solCiBankSend".to_string())
+                .spawn(move || {
+                    let _ = Self::bank_send_loop(
+                        exit,
+                        verified_vote_label_packets_receiver,
+                        poh_recorder,
+                        &verified_packets_sender,
+                    );
+                })
+                .unwrap()
+        };
 
         let send_thread = Builder::new()
             .name("solCiProcVotes".to_string())
@@ -1447,7 +1449,7 @@ mod tests {
         let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
         let max_complete_rewards_slot = Arc::new(AtomicU64::default());
         let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
-            &exit,
+            exit,
             max_complete_transaction_status_slot,
             max_complete_rewards_slot,
             bank_forks,
@@ -1563,7 +1565,7 @@ mod tests {
         let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
         let max_complete_rewards_slot = Arc::new(AtomicU64::default());
         let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
-            &exit,
+            exit,
             max_complete_transaction_status_slot,
             max_complete_rewards_slot,
             bank_forks,

core/src/commitment_service.rs

@@ -56,27 +56,26 @@ pub struct AggregateCommitmentService {
 
 impl AggregateCommitmentService {
     pub fn new(
-        exit: &Arc<AtomicBool>,
+        exit: Arc<AtomicBool>,
         block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
         subscriptions: Arc<RpcSubscriptions>,
     ) -> (Sender<CommitmentAggregationData>, Self) {
         let (sender, receiver): (
             Sender<CommitmentAggregationData>,
             Receiver<CommitmentAggregationData>,
         ) = unbounded();
-        let exit_ = exit.clone();
         (
             sender,
             Self {
                 t_commitment: Builder::new()
                     .name("solAggCommitSvc".to_string())
                     .spawn(move || loop {
-                        if exit_.load(Ordering::Relaxed) {
+                        if exit.load(Ordering::Relaxed) {
                             break;
                         }
 
                         if let Err(RecvTimeoutError::Disconnected) =
-                            Self::run(&receiver, &block_commitment_cache, &subscriptions, &exit_)
+                            Self::run(&receiver, &block_commitment_cache, &subscriptions, &exit)
                         {
                             break;
                         }
@@ -90,7 +89,7 @@ impl AggregateCommitmentService {
         receiver: &Receiver<CommitmentAggregationData>,
         block_commitment_cache: &RwLock<BlockCommitmentCache>,
         subscriptions: &Arc<RpcSubscriptions>,
-        exit: &Arc<AtomicBool>,
+        exit: &AtomicBool,
     ) -> Result<(), RecvTimeoutError> {
         loop {
             if exit.load(Ordering::Relaxed) {

core/src/completed_data_sets_service.rs

@@ -32,10 +32,9 @@ impl CompletedDataSetsService {
         completed_sets_receiver: CompletedDataSetsReceiver,
         blockstore: Arc<Blockstore>,
         rpc_subscriptions: Arc<RpcSubscriptions>,
-        exit: &Arc<AtomicBool>,
+        exit: Arc<AtomicBool>,
         max_slots: Arc<MaxSlots>,
     ) -> Self {
-        let exit = exit.clone();
         let thread_hdl = Builder::new()
             .name("solComplDataSet".to_string())
             .spawn(move || loop {

core/src/fetch_stage.rs

@@ -34,7 +34,7 @@ impl FetchStage {
         sockets: Vec<UdpSocket>,
         tpu_forwards_sockets: Vec<UdpSocket>,
         tpu_vote_sockets: Vec<UdpSocket>,
-        exit: &Arc<AtomicBool>,
+        exit: Arc<AtomicBool>,
         poh_recorder: &Arc<RwLock<PohRecorder>>,
         coalesce: Duration,
     ) -> (Self, PacketBatchReceiver, PacketBatchReceiver) {
@@ -66,7 +66,7 @@ impl FetchStage {
         sockets: Vec<UdpSocket>,
         tpu_forwards_sockets: Vec<UdpSocket>,
         tpu_vote_sockets: Vec<UdpSocket>,
-        exit: &Arc<AtomicBool>,
+        exit: Arc<AtomicBool>,
         sender: &PacketBatchSender,
         vote_sender: &PacketBatchSender,
         forward_sender: &PacketBatchSender,
@@ -142,7 +142,7 @@ impl FetchStage {
         tpu_sockets: Vec<Arc<UdpSocket>>,
         tpu_forwards_sockets: Vec<Arc<UdpSocket>>,
         tpu_vote_sockets: Vec<Arc<UdpSocket>>,
-        exit: &Arc<AtomicBool>,
+        exit: Arc<AtomicBool>,
         sender: &PacketBatchSender,
         vote_sender: &PacketBatchSender,
         forward_sender: &PacketBatchSender,
@@ -234,7 +234,6 @@ impl FetchStage {
             })
             .unwrap();
 
-        let exit = exit.clone();
         let metrics_thread_hdl = Builder::new()
             .name("solFetchStgMetr".to_string())
             .spawn(move || loop {

core/src/ledger_cleanup_service.rs

@@ -49,9 +49,8 @@ impl LedgerCleanupService {
         new_root_receiver: Receiver<Slot>,
         blockstore: Arc<Blockstore>,
         max_ledger_shreds: u64,
-        exit: &Arc<AtomicBool>,
+        exit: Arc<AtomicBool>,
     ) -> Self {
-        let exit = exit.clone();
         let mut last_purge_slot = 0;
 
         info!(

core/src/ledger_metric_report_service.rs

@@ -23,12 +23,11 @@ pub struct LedgerMetricReportService {
 }
 
 impl LedgerMetricReportService {
-    pub fn new(blockstore: Arc<Blockstore>, exit: &Arc<AtomicBool>) -> Self {
-        let exit_signal = exit.clone();
+    pub fn new(blockstore: Arc<Blockstore>, exit: Arc<AtomicBool>) -> Self {
         let t_cf_metric = Builder::new()
             .name("solRocksCfMtrcs".to_string())
             .spawn(move || loop {
-                if exit_signal.load(Ordering::Relaxed) {
+                if exit.load(Ordering::Relaxed) {
                     break;
                 }
                 thread::sleep(Duration::from_millis(

core/src/poh_timing_report_service.rs

@@ -24,13 +24,12 @@ pub struct PohTimingReportService {
 }
 
 impl PohTimingReportService {
-    pub fn new(receiver: PohTimingReceiver, exit: &Arc<AtomicBool>) -> Self {
-        let exit_signal = exit.clone();
+    pub fn new(receiver: PohTimingReceiver, exit: Arc<AtomicBool>) -> Self {
         let mut poh_timing_reporter = PohTimingReporter::default();
         let t_poh_timing = Builder::new()
             .name("solPohTimingRpt".to_string())
             .spawn(move || loop {
-                if exit_signal.load(Ordering::Relaxed) {
+                if exit.load(Ordering::Relaxed) {
                     break;
                 }
                 if let Ok(SlotPohTimingInfo {
@@ -65,7 +64,7 @@ mod test {
         let exit = Arc::new(AtomicBool::new(false));
         // Create the service
         let poh_timing_report_service =
-            PohTimingReportService::new(poh_timing_point_receiver, &exit);
+            PohTimingReportService::new(poh_timing_point_receiver, exit.clone());
 
         // Send SlotPohTimingPoint
         let _ = poh_timing_point_sender.send(SlotPohTimingInfo::new_slot_start_poh_time_point(