Add batch size CLI flags

beacon_node/beacon_processor/src/lib.rs

@@ -193,8 +193,8 @@ const WORKER_TASK_NAME: &str = "beacon_processor_worker";
 /// Poisoning occurs when an invalid signature is included in a batch of attestations. A single
 /// invalid signature causes the entire batch to fail. When a batch fails, we fall-back to
 /// individually verifying each attestation signature.
-const MAX_GOSSIP_ATTESTATION_BATCH_SIZE: usize = 64;
-const MAX_GOSSIP_AGGREGATE_BATCH_SIZE: usize = 64;
+const DEFAULT_MAX_GOSSIP_ATTESTATION_BATCH_SIZE: usize = 64;
+const DEFAULT_MAX_GOSSIP_AGGREGATE_BATCH_SIZE: usize = 64;
 
 /// Unique IDs used for metrics and testing.
 pub const WORKER_FREED: &str = "worker_freed";
@@ -232,6 +232,8 @@ pub struct BeaconProcessorConfig {
     pub max_workers: usize,
     pub max_work_event_queue_len: usize,
     pub max_scheduled_work_queue_len: usize,
+    pub max_gossip_attestation_batch_size: usize,
+    pub max_gossip_aggregate_batch_size: usize,
     pub enable_backfill_rate_limiting: bool,
 }
 
@@ -241,6 +243,8 @@ impl Default for BeaconProcessorConfig {
             max_workers: cmp::max(1, num_cpus::get()),
             max_work_event_queue_len: DEFAULT_MAX_WORK_EVENT_QUEUE_LEN,
             max_scheduled_work_queue_len: DEFAULT_MAX_SCHEDULED_WORK_QUEUE_LEN,
+            max_gossip_attestation_batch_size: DEFAULT_MAX_GOSSIP_ATTESTATION_BATCH_SIZE,
+            max_gossip_aggregate_batch_size: DEFAULT_MAX_GOSSIP_AGGREGATE_BATCH_SIZE,
             enable_backfill_rate_limiting: true,
         }
     }
@@ -925,8 +929,10 @@ impl<E: EthSpec> BeaconProcessor<E> {
                         // aggregates are more valuable to local validators and effectively give us
                         // more information with less signature verification time.
                         } else if aggregate_queue.len() > 0 {
-                            let batch_size =
-                                cmp::min(aggregate_queue.len(), MAX_GOSSIP_AGGREGATE_BATCH_SIZE);
+                            let batch_size = cmp::min(
+                                aggregate_queue.len(),
+                                self.config.max_gossip_aggregate_batch_size,
+                            );
 
                             if batch_size < 2 {
                                 // One single aggregate is in the queue, process it individually.
@@ -985,7 +991,7 @@ impl<E: EthSpec> BeaconProcessor<E> {
                         } else if attestation_queue.len() > 0 {
                             let batch_size = cmp::min(
                                 attestation_queue.len(),
-                                MAX_GOSSIP_ATTESTATION_BATCH_SIZE,
+                                self.config.max_gossip_attestation_batch_size,
                             );
 
                             if batch_size < 2 {

beacon_node/src/cli.rs

@@ -1163,4 +1163,25 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
                 .default_value("12288")
                 .takes_value(true)
         )
+        .arg(
+            Arg::with_name("beacon-processor-attestation-batch-size")
+                .long("beacon-processor-gossip-attestation-batch")
+                .value_name("INTEGER")
+                .help("Specifies the number of gossip attestations in a signature verification batch. \
+                       Higher values may reduce CPU usage in a healthy network whilst lower values may \
+                       increase CPU usage in an unhealthy network.")
+                .default_value("64")
+                .takes_value(true)
+        )
+        .arg(
+            Arg::with_name("beacon-processor-aggregate-batch-size")
+                .long("beacon-processor-gossip-aggregate-batch")
+                .value_name("INTEGER")
+                .help("Specifies the number of gossip aggregate attestations in a signature \
+                       verification batch. \
+                       Higher values may reduce CPU usage in a healthy network whilst lower values may \
+                       increase CPU usage in an unhealthy network.")
+                .default_value("64")
+                .takes_value(true)
+        )
 }

beacon_node/src/config.rs

@@ -819,6 +819,14 @@ pub fn get_config<E: EthSpec>(
         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")?;
+    client_config
+        .beacon_processor
+        .max_gossip_attestation_batch_size =
+        clap_utils::parse_required(cli_args, "beacon-processor-attestation-batch-size")?;
+    client_config
+        .beacon_processor
+        .max_gossip_aggregate_batch_size =
+        clap_utils::parse_required(cli_args, "beacon-processor-aggregate-batch-size")?;
 
     Ok(client_config)
 }

lighthouse/tests/beacon_node.rs

@@ -2289,6 +2289,8 @@ fn beacon_processor() {
         .flag("beacon-processor-max-workers", Some("1"))
         .flag("beacon-processor-work-queue", Some("2"))
         .flag("beacon-processor-reprocess-queue", Some("3"))
+        .flag("beacon-processor-attestation-batch-size", Some("4"))
+        .flag("beacon-processor-aggregate-batch-size", Some("5"))
         .flag("disable-backfill-rate-limiting", None)
         .run_with_zero_port()
         .with_config(|config| {
@@ -2298,6 +2300,8 @@ fn beacon_processor() {
                     max_workers: 1,
                     max_work_event_queue_len: 2,
                     max_scheduled_work_queue_len: 3,
+                    max_gossip_attestation_batch_size: 4,
+                    max_gossip_aggregate_batch_size: 5,
                     enable_backfill_rate_limiting: false
                 }
             )