diff --git a/Cargo.lock b/Cargo.lock index 166c9274a2..58da613ee9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4708,6 +4708,7 @@ dependencies = [ "account_manager", "account_utils", "beacon_node", + "beacon_processor", "bls", "boot_node", "clap", diff --git a/beacon_node/beacon_chain/src/chain_config.rs b/beacon_node/beacon_chain/src/chain_config.rs index cc7a957ecc..017d2d6035 100644 --- a/beacon_node/beacon_chain/src/chain_config.rs +++ b/beacon_node/beacon_chain/src/chain_config.rs @@ -79,8 +79,6 @@ pub struct ChainConfig { /// /// This is useful for block builders and testing. pub always_prepare_payload: bool, - /// Whether backfill sync processing should be rate-limited. - pub enable_backfill_rate_limiting: bool, /// Whether to use `ProgressiveBalancesCache` in unrealized FFG progression calculation. pub progressive_balances_mode: ProgressiveBalancesMode, } @@ -112,7 +110,6 @@ impl Default for ChainConfig { shuffling_cache_size: crate::shuffling_cache::DEFAULT_CACHE_SIZE, genesis_backfill: false, always_prepare_payload: false, - enable_backfill_rate_limiting: true, progressive_balances_mode: ProgressiveBalancesMode::Checked, } } diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index b8d5db568e..1383dd1f78 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -68,16 +68,21 @@ struct TestRig { impl Drop for TestRig { fn drop(&mut self) { // Causes the beacon processor to shutdown. - self.beacon_processor_tx = BeaconProcessorSend(mpsc::channel(MAX_WORK_EVENT_QUEUE_LEN).0); + let len = BeaconProcessorConfig::default().max_work_event_queue_len; + self.beacon_processor_tx = BeaconProcessorSend(mpsc::channel(len).0); } } impl TestRig { pub async fn new(chain_length: u64) -> Self { - Self::new_with_chain_config(chain_length, ChainConfig::default()).await + Self::new_parametric( + chain_length, + BeaconProcessorConfig::default().enable_backfill_rate_limiting, + ) + .await } - pub async fn new_with_chain_config(chain_length: u64, chain_config: ChainConfig) -> Self { + pub async fn new_parametric(chain_length: u64, enable_backfill_rate_limiting: bool) -> Self { // This allows for testing voluntary exits without building out a massive chain. let mut spec = E::default_spec(); spec.shard_committee_period = 2; @@ -86,7 +91,7 @@ impl TestRig { .spec(spec) .deterministic_keypairs(VALIDATOR_COUNT) .fresh_ephemeral_store() - .chain_config(chain_config) + .chain_config(<_>::default()) .build(); harness.advance_slot(); @@ -172,8 +177,15 @@ impl TestRig { let log = harness.logger().clone(); - let (beacon_processor_tx, beacon_processor_rx) = mpsc::channel(MAX_WORK_EVENT_QUEUE_LEN); - let beacon_processor_tx = BeaconProcessorSend(beacon_processor_tx); + let mut beacon_processor_config = BeaconProcessorConfig::default(); + beacon_processor_config.enable_backfill_rate_limiting = enable_backfill_rate_limiting; + let BeaconProcessorChannels { + beacon_processor_tx, + beacon_processor_rx, + work_reprocessing_tx, + work_reprocessing_rx, + } = BeaconProcessorChannels::new(&beacon_processor_config); + let (sync_tx, _sync_rx) = mpsc::unbounded_channel(); // Default metadata @@ -196,8 +208,6 @@ impl TestRig { let executor = harness.runtime.task_executor.clone(); - let (work_reprocessing_tx, work_reprocessing_rx) = - mpsc::channel(MAX_SCHEDULED_WORK_QUEUE_LEN); let (work_journal_tx, work_journal_rx) = mpsc::channel(16_364); let duplicate_cache = DuplicateCache::default(); @@ -220,7 +230,7 @@ impl TestRig { executor, max_workers: cmp::max(1, num_cpus::get()), current_workers: 0, - enable_backfill_rate_limiting: harness.chain.config.enable_backfill_rate_limiting, + config: beacon_processor_config, log: log.clone(), } .spawn_manager( @@ -940,11 +950,8 @@ async fn test_backfill_sync_processing() { /// Ensure that backfill batches get processed as fast as they can when rate-limiting is disabled. #[tokio::test] async fn test_backfill_sync_processing_rate_limiting_disabled() { - let chain_config = ChainConfig { - enable_backfill_rate_limiting: false, - ..Default::default() - }; - let mut rig = TestRig::new_with_chain_config(SMALL_CHAIN, chain_config).await; + let enable_backfill_rate_limiting = false; + let mut rig = TestRig::new_parametric(SMALL_CHAIN, enable_backfill_rate_limiting).await; for _ in 0..3 { rig.enqueue_backfill_batch(); diff --git a/beacon_node/src/cli.rs b/beacon_node/src/cli.rs index fc53f57888..b96c8bc018 100644 --- a/beacon_node/src/cli.rs +++ b/beacon_node/src/cli.rs @@ -1131,4 +1131,36 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> { .takes_value(true) .possible_values(ProgressiveBalancesMode::VARIANTS) ) + .arg( + Arg::with_name("beacon-processor-max-workers") + .long("beacon-processor-max-workers") + .value_name("INTEGER") + .help("Specifies the maximum concurrent tasks for the task scheduler. Increasing \ + this value may increase resource consumption. Reducing the value \ + may result in decreased resource usage and diminished performance. The \ + default value is the number of logical CPU cores on the host.") + .takes_value(true) + ) + .arg( + Arg::with_name("beacon-processor-work-queue") + .long("beacon-processor-work-queue") + .value_name("INTEGER") + .help("Specifies the length of the inbound event queue. Increasing this value \ + may prevent messages from being dropped at risk of overwhelming the \ + host resources. Decreasing this value may cause messages to be dropped but \ + may help resource-constrained hosts.") + .default_value("16384") + .takes_value(true) + ) + .arg( + Arg::with_name("beacon-processor-reprocess-queue") + .long("beacon-processor-reprocess-queue") + .value_name("INTEGER") + .help("Specifies the length of the queue for messages requiring delayed processing. \ + Increasing this value may prevent messages from being dropped at risk of \ + overwhelming the host resources. Decreasing this value may cause messages \ + to be dropped but may help resource-constrained hosts.") + .default_value("12288") + .takes_value(true) + ) } diff --git a/beacon_node/src/config.rs b/beacon_node/src/config.rs index 948c70dd41..8f30e50d68 100644 --- a/beacon_node/src/config.rs +++ b/beacon_node/src/config.rs @@ -792,7 +792,7 @@ pub fn get_config( } // Backfill sync rate-limiting - client_config.chain.enable_backfill_rate_limiting = + client_config.beacon_processor.enable_backfill_rate_limiting = !cli_args.is_present("disable-backfill-rate-limiting"); if let Some(path) = clap_utils::parse_optional(cli_args, "invalid-gossip-verified-blocks-path")? @@ -806,6 +806,20 @@ pub fn get_config( client_config.chain.progressive_balances_mode = progressive_balances_mode; } + if let Some(max_workers) = clap_utils::parse_optional(cli_args, "beacon-processor-max-workers")? + { + client_config.beacon_processor.max_workers = max_workers; + } + + if client_config.beacon_processor.max_workers == 0 { + return Err("--beacon-processor-max-workers must be a non-zero value".to_string()); + } + + client_config.beacon_processor.max_work_event_queue_len = + clap_utils::parse_required(cli_args, "beacon-processor-work-queue")?; + client_config.beacon_processor.max_scheduled_work_queue_len = + clap_utils::parse_required(cli_args, "beacon-processor-reprocess-queue")?; + Ok(client_config) } diff --git a/lighthouse/Cargo.toml b/lighthouse/Cargo.toml index e7746a2db9..675b37192b 100644 --- a/lighthouse/Cargo.toml +++ b/lighthouse/Cargo.toml @@ -64,6 +64,7 @@ slashing_protection = { path = "../validator_client/slashing_protection" } lighthouse_network = { path = "../beacon_node/lighthouse_network" } sensitive_url = { path = "../common/sensitive_url" } eth1 = { path = "../beacon_node/eth1" } +beacon_processor = { path = "../beacon_node/beacon_processor" } [[test]] name = "lighthouse_tests" diff --git a/lighthouse/tests/beacon_node.rs b/lighthouse/tests/beacon_node.rs index 4badcef3cf..8e4aab6062 100644 --- a/lighthouse/tests/beacon_node.rs +++ b/lighthouse/tests/beacon_node.rs @@ -5,6 +5,7 @@ use beacon_node::beacon_chain::chain_config::{ DisallowedReOrgOffsets, DEFAULT_RE_ORG_CUTOFF_DENOMINATOR, DEFAULT_RE_ORG_MAX_EPOCHS_SINCE_FINALIZATION, DEFAULT_RE_ORG_THRESHOLD, }; +use beacon_processor::BeaconProcessorConfig; use eth1::Eth1Endpoint; use lighthouse_network::PeerId; use std::fs::File; @@ -1118,13 +1119,13 @@ fn disable_backfill_rate_limiting_flag() { CommandLineTest::new() .flag("disable-backfill-rate-limiting", None) .run_with_zero_port() - .with_config(|config| assert!(!config.chain.enable_backfill_rate_limiting)); + .with_config(|config| assert!(!config.beacon_processor.enable_backfill_rate_limiting)); } #[test] fn default_backfill_rate_limiting_flag() { CommandLineTest::new() .run_with_zero_port() - .with_config(|config| assert!(config.chain.enable_backfill_rate_limiting)); + .with_config(|config| assert!(config.beacon_processor.enable_backfill_rate_limiting)); } #[test] fn default_boot_nodes() { @@ -2277,3 +2278,36 @@ fn progressive_balances_fast() { ) }); } + +#[test] +fn beacon_processor() { + CommandLineTest::new() + .run_with_zero_port() + .with_config(|config| assert_eq!(config.beacon_processor, <_>::default())); + + CommandLineTest::new() + .flag("beacon-processor-max-workers", Some("1")) + .flag("beacon-processor-work-queue", Some("2")) + .flag("beacon-processor-reprocess-queue", Some("3")) + .flag("disable-backfill-rate-limiting", None) + .run_with_zero_port() + .with_config(|config| { + assert_eq!( + config.beacon_processor, + BeaconProcessorConfig { + max_workers: 1, + max_work_event_queue_len: 2, + max_scheduled_work_queue_len: 3, + enable_backfill_rate_limiting: false + } + ) + }); +} + +#[test] +#[should_panic] +fn beacon_processor_zero_workers() { + CommandLineTest::new() + .flag("beacon-processor-max-workers", Some("0")) + .run_with_zero_port(); +}