Merge branch 'unstable' of https://github.com/sigp/lighthouse into merge-unstable-deneb-aug-9

beacon_node/client/src/builder.rs

@@ -14,10 +14,8 @@ use beacon_chain::{
     store::{HotColdDB, ItemStore, LevelDB, StoreConfig},
     BeaconChain, BeaconChainTypes, Eth1ChainBackend, MigratorConfig, 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::{
@@ -39,7 +37,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,
@@ -79,11 +77,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>
@@ -99,10 +95,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,
@@ -121,10 +113,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,
         }
     }
 
@@ -140,6 +130,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
@@ -479,6 +475,7 @@ where
                         chain: None,
                         network_senders: None,
                         network_globals: None,
+                        beacon_processor_send: None,
                         eth1_service: Some(genesis_service.eth1_service.clone()),
                         log: context.log().clone(),
                         sse_logging_components: runtime_context.sse_logging_components.clone(),
@@ -556,6 +553,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 {
@@ -571,8 +572,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))?;
@@ -695,6 +696,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 {
@@ -704,6 +713,7 @@ where
                 network_senders: self.network_senders.clone(),
                 network_globals: self.network_globals.clone(),
                 eth1_service: self.eth1_service.clone(),
+                beacon_processor_send: Some(beacon_processor_channels.beacon_processor_tx.clone()),
                 sse_logging_components: runtime_context.sse_logging_components.clone(),
                 log: log.clone(),
             });
@@ -767,15 +777,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_chain.spec.maximum_gossip_clock_disparity(),