Add BeaconProcessorConfig

beacon_node/client/src/builder.rs

@@ -13,10 +13,8 @@ use beacon_chain::{
     store::{HotColdDB, ItemStore, LevelDB, StoreConfig},
     BeaconChain, BeaconChainTypes, Eth1ChainBackend, ServerSentEventHandler,
 };
-use beacon_processor::{
-    work_reprocessing_queue::ReprocessQueueMessage, BeaconProcessor, BeaconProcessorSend,
-    WorkEvent, MAX_SCHEDULED_WORK_QUEUE_LEN, MAX_WORK_EVENT_QUEUE_LEN,
-};
+use beacon_processor::BeaconProcessorConfig;
+use beacon_processor::{BeaconProcessor, BeaconProcessorChannels};
 use environment::RuntimeContext;
 use eth1::{Config as Eth1Config, Service as Eth1Service};
 use eth2::{
@@ -37,7 +35,7 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
 use timer::spawn_timer;
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::oneshot;
 use types::{
     test_utils::generate_deterministic_keypairs, BeaconState, ChainSpec, EthSpec,
     ExecutionBlockHash, Hash256, SignedBeaconBlock,
@@ -76,11 +74,9 @@ pub struct ClientBuilder<T: BeaconChainTypes> {
     http_api_config: http_api::Config,
     http_metrics_config: http_metrics::Config,
     slasher: Option<Arc<Slasher<T::EthSpec>>>,
+    beacon_processor_config: Option<BeaconProcessorConfig>,
+    beacon_processor_channels: Option<BeaconProcessorChannels<T::EthSpec>>,
     eth_spec_instance: T::EthSpec,
-    beacon_processor_send: BeaconProcessorSend<T::EthSpec>,
-    beacon_processor_receive: mpsc::Receiver<WorkEvent<T::EthSpec>>,
-    work_reprocessing_tx: mpsc::Sender<ReprocessQueueMessage>,
-    work_reprocessing_rx: mpsc::Receiver<ReprocessQueueMessage>,
 }
 
 impl<TSlotClock, TEth1Backend, TEthSpec, THotStore, TColdStore>
@@ -96,10 +92,6 @@ where
     ///
     /// The `eth_spec_instance` parameter is used to concretize `TEthSpec`.
     pub fn new(eth_spec_instance: TEthSpec) -> Self {
-        let (beacon_processor_send, beacon_processor_receive) =
-            mpsc::channel(MAX_WORK_EVENT_QUEUE_LEN);
-        let (work_reprocessing_tx, work_reprocessing_rx) =
-            mpsc::channel(MAX_SCHEDULED_WORK_QUEUE_LEN);
         Self {
             slot_clock: None,
             store: None,
@@ -117,10 +109,8 @@ where
             http_metrics_config: <_>::default(),
             slasher: None,
             eth_spec_instance,
-            beacon_processor_send: BeaconProcessorSend(beacon_processor_send),
-            beacon_processor_receive,
-            work_reprocessing_tx,
-            work_reprocessing_rx,
+            beacon_processor_config: None,
+            beacon_processor_channels: None,
         }
     }
 
@@ -136,6 +126,12 @@ where
         self
     }
 
+    pub fn beacon_processor(mut self, config: BeaconProcessorConfig) -> Self {
+        self.beacon_processor_channels = Some(BeaconProcessorChannels::new(&config));
+        self.beacon_processor_config = Some(config);
+        self
+    }
+
     pub fn slasher(mut self, slasher: Arc<Slasher<TEthSpec>>) -> Self {
         self.slasher = Some(slasher);
         self
@@ -571,6 +567,10 @@ where
             .as_ref()
             .ok_or("network requires a runtime_context")?
             .clone();
+        let beacon_processor_channels = self
+            .beacon_processor_channels
+            .as_ref()
+            .ok_or("network requires beacon_processor_channels")?;
 
         // If gossipsub metrics are required we build a registry to record them
         let mut gossipsub_registry = if config.metrics_enabled {
@@ -586,8 +586,8 @@ where
             gossipsub_registry
                 .as_mut()
                 .map(|registry| registry.sub_registry_with_prefix("gossipsub")),
-            self.beacon_processor_send.clone(),
-            self.work_reprocessing_tx.clone(),
+            beacon_processor_channels.beacon_processor_tx.clone(),
+            beacon_processor_channels.work_reprocessing_tx.clone(),
         )
         .await
         .map_err(|e| format!("Failed to start network: {:?}", e))?;
@@ -710,6 +710,14 @@ where
             .runtime_context
             .as_ref()
             .ok_or("build requires a runtime context")?;
+        let beacon_processor_channels = self
+            .beacon_processor_channels
+            .take()
+            .ok_or("build requires beacon_processor_channels")?;
+        let beacon_processor_config = self
+            .beacon_processor_config
+            .take()
+            .ok_or("build requires a beacon_processor_config")?;
         let log = runtime_context.log().clone();
 
         let http_api_listen_addr = if self.http_api_config.enabled {
@@ -719,7 +727,7 @@ where
                 network_senders: self.network_senders.clone(),
                 network_globals: self.network_globals.clone(),
                 eth1_service: self.eth1_service.clone(),
-                beacon_processor_send: Some(self.beacon_processor_send.clone()),
+                beacon_processor_send: Some(beacon_processor_channels.beacon_processor_tx.clone()),
                 sse_logging_components: runtime_context.sse_logging_components.clone(),
                 log: log.clone(),
             });
@@ -783,15 +791,13 @@ where
                     executor: beacon_processor_context.executor.clone(),
                     max_workers: cmp::max(1, num_cpus::get()),
                     current_workers: 0,
-                    enable_backfill_rate_limiting: beacon_chain
-                        .config
-                        .enable_backfill_rate_limiting,
+                    config: beacon_processor_config,
                     log: beacon_processor_context.log().clone(),
                 }
                 .spawn_manager(
-                    self.beacon_processor_receive,
-                    self.work_reprocessing_tx,
-                    self.work_reprocessing_rx,
+                    beacon_processor_channels.beacon_processor_rx,
+                    beacon_processor_channels.work_reprocessing_tx,
+                    beacon_processor_channels.work_reprocessing_rx,
                     None,
                     beacon_chain.slot_clock.clone(),
                 );

beacon_node/client/src/config.rs

@@ -1,4 +1,5 @@
 use beacon_chain::validator_monitor::DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD;
+use beacon_processor::BeaconProcessorConfig;
 use directory::DEFAULT_ROOT_DIR;
 use environment::LoggerConfig;
 use network::NetworkConfig;
@@ -80,6 +81,7 @@ pub struct Config {
     pub slasher: Option<slasher::Config>,
     pub logger_config: LoggerConfig,
     pub always_prefer_builder_payload: bool,
+    pub beacon_processor: BeaconProcessorConfig,
 }
 
 impl Default for Config {
@@ -107,6 +109,7 @@ impl Default for Config {
             validator_monitor_individual_tracking_threshold: DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD,
             logger_config: LoggerConfig::default(),
             always_prefer_builder_payload: false,
+            beacon_processor: <_>::default(),
         }
     }
 }