Disable RpcCompletedSlotsService when unused (#1474)

Adjust RpcCompletedSlotsService to only be created if --full-rpc-api
is passed into the CLI. This service is unused for non-RPC nodes

core/src/validator.rs

@@ -54,7 +54,8 @@ use {
     solana_ledger::{
         bank_forks_utils,
         blockstore::{
-            Blockstore, BlockstoreError, BlockstoreSignals, CompletedSlotsReceiver, PurgeType,
+            Blockstore, BlockstoreError, PurgeType, MAX_COMPLETED_SLOTS_IN_CHANNEL,
+            MAX_REPLAY_WAKE_UP_SIGNALS,
         },
         blockstore_metric_report_service::BlockstoreMetricReportService,
         blockstore_options::{BlockstoreOptions, BlockstoreRecoveryMode, LedgerColumnOptions},
@@ -459,7 +460,7 @@ pub struct Validator {
     validator_exit: Arc<RwLock<Exit>>,
     json_rpc_service: Option<JsonRpcService>,
     pubsub_service: Option<PubSubService>,
-    rpc_completed_slots_service: JoinHandle<()>,
+    rpc_completed_slots_service: Option<JoinHandle<()>>,
     optimistically_confirmed_bank_tracker: Option<OptimisticallyConfirmedBankTracker>,
     transaction_status_service: Option<TransactionStatusService>,
     rewards_recorder_service: Option<RewardsRecorderService>,
@@ -701,7 +702,6 @@ impl Validator {
             blockstore,
             original_blockstore_root,
             ledger_signal_receiver,
-            completed_slots_receiver,
             leader_schedule_cache,
             starting_snapshot_hashes,
             TransactionHistoryServices {
@@ -981,6 +981,7 @@ impl Validator {
             pubsub_service,
             completed_data_sets_sender,
             completed_data_sets_service,
+            rpc_completed_slots_service,
             optimistically_confirmed_bank_tracker,
             bank_notification_sender,
         ) = if let Some((rpc_addr, rpc_pubsub_addr)) = config.rpc_addrs {
@@ -1062,6 +1063,20 @@ impl Validator {
                     )
                 };
 
+            let rpc_completed_slots_service = if !config.rpc_config.full_api {
+                None
+            } else {
+                let (completed_slots_sender, completed_slots_receiver) =
+                    bounded(MAX_COMPLETED_SLOTS_IN_CHANNEL);
+                blockstore.add_completed_slots_signal(completed_slots_sender);
+
+                Some(RpcCompletedSlotsService::spawn(
+                    completed_slots_receiver,
+                    rpc_subscriptions.clone(),
+                    exit.clone(),
+                ))
+            };
+
             let optimistically_confirmed_bank_tracker =
                 Some(OptimisticallyConfirmedBankTracker::new(
                     bank_notification_receiver,
@@ -1081,11 +1096,12 @@ impl Validator {
                 pubsub_service,
                 completed_data_sets_sender,
                 completed_data_sets_service,
+                rpc_completed_slots_service,
                 optimistically_confirmed_bank_tracker,
                 bank_notification_sender_config,
             )
         } else {
-            (None, None, None, None, None, None)
+            (None, None, None, None, None, None, None)
         };
 
         if config.halt_at_slot.is_some() {
@@ -1181,12 +1197,6 @@ impl Validator {
         let (gossip_verified_vote_hash_sender, gossip_verified_vote_hash_receiver) = unbounded();
         let (duplicate_confirmed_slot_sender, duplicate_confirmed_slots_receiver) = unbounded();
 
-        let rpc_completed_slots_service = RpcCompletedSlotsService::spawn(
-            completed_slots_receiver,
-            rpc_subscriptions.clone(),
-            exit.clone(),
-        );
-
         let (banking_tracer, tracer_thread) =
             BankingTracer::new((config.banking_trace_dir_byte_limit > 0).then_some((
                 &blockstore.banking_trace_path(),
@@ -1541,9 +1551,11 @@ impl Validator {
             pubsub_service.join().expect("pubsub_service");
         }
 
-        self.rpc_completed_slots_service
-            .join()
-            .expect("rpc_completed_slots_service");
+        if let Some(rpc_completed_slots_service) = self.rpc_completed_slots_service {
+            rpc_completed_slots_service
+                .join()
+                .expect("rpc_completed_slots_service");
+        }
 
         if let Some(optimistically_confirmed_bank_tracker) =
             self.optimistically_confirmed_bank_tracker
@@ -1807,7 +1819,6 @@ fn load_blockstore(
         Arc<Blockstore>,
         Slot,
         Receiver<bool>,
-        CompletedSlotsReceiver,
         LeaderScheduleCache,
         Option<StartingSnapshotHashes>,
         TransactionHistoryServices,
@@ -1845,13 +1856,12 @@ fn load_blockstore(
         check_poh_speed(&genesis_config, None)?;
     }
 
-    let BlockstoreSignals {
-        mut blockstore,
-        ledger_signal_receiver,
-        completed_slots_receiver,
-        ..
-    } = Blockstore::open_with_signal(ledger_path, blockstore_options_from_config(config))
-        .map_err(|err| format!("Failed to open Blockstore: {err:?}"))?;
+    let mut blockstore =
+        Blockstore::open_with_options(ledger_path, blockstore_options_from_config(config))
+            .map_err(|err| format!("Failed to open Blockstore: {err:?}"))?;
+
+    let (ledger_signal_sender, ledger_signal_receiver) = bounded(MAX_REPLAY_WAKE_UP_SIGNALS);
+    blockstore.add_new_shred_signal(ledger_signal_sender);
 
     blockstore.shred_timing_point_sender = poh_timing_point_sender;
     // following boot sequence (esp BankForks) could set root. so stash the original value
@@ -1938,7 +1948,6 @@ fn load_blockstore(
         blockstore,
         original_blockstore_root,
         ledger_signal_receiver,
-        completed_slots_receiver,
         leader_schedule_cache,
         starting_snapshot_hashes,
         transaction_history_services,