Reduce load on validator subscription channels (#5311)

* Fix tests

* Merge branch 'unstable' into unclog-channels

* Avoid reallocations

* Reduce subscription load on beacon node

Author: pawanjay176

beacon_node/http_api/src/lib.rs

@@ -3448,34 +3448,34 @@ pub fn serve<T: BeaconChainTypes>(
              chain: Arc<BeaconChain<T>>,
              log: Logger| {
                 task_spawner.blocking_json_task(Priority::P0, move || {
-                    for subscription in &subscriptions {
-                        chain
-                            .validator_monitor
-                            .write()
-                            .auto_register_local_validator(subscription.validator_index);
-
-                        let validator_subscription = api_types::ValidatorSubscription {
-                            validator_index: subscription.validator_index,
-                            attestation_committee_index: subscription.committee_index,
-                            slot: subscription.slot,
-                            committee_count_at_slot: subscription.committees_at_slot,
-                            is_aggregator: subscription.is_aggregator,
-                        };
-
-                        let message = ValidatorSubscriptionMessage::AttestationSubscribe {
-                            subscriptions: vec![validator_subscription],
-                        };
-                        if let Err(e) = validator_subscription_tx.try_send(message) {
-                            warn!(
-                                log,
-                                "Unable to process committee subscriptions";
-                                "info" => "the host may be overloaded or resource-constrained",
-                                "error" => ?e,
-                            );
-                            return Err(warp_utils::reject::custom_server_error(
-                                "unable to queue subscription, host may be overloaded or shutting down".to_string(),
-                            ));
-                        }
+                    let subscriptions: std::collections::BTreeSet<_> = subscriptions
+                        .iter()
+                        .map(|subscription| {
+                            chain
+                                .validator_monitor
+                                .write()
+                                .auto_register_local_validator(subscription.validator_index);
+                            api_types::ValidatorSubscription {
+                                attestation_committee_index: subscription.committee_index,
+                                slot: subscription.slot,
+                                committee_count_at_slot: subscription.committees_at_slot,
+                                is_aggregator: subscription.is_aggregator,
+                            }
+                        })
+                        .collect();
+                    let message =
+                        ValidatorSubscriptionMessage::AttestationSubscribe { subscriptions };
+                    if let Err(e) = validator_subscription_tx.try_send(message) {
+                        warn!(
+                            log,
+                            "Unable to process committee subscriptions";
+                            "info" => "the host may be overloaded or resource-constrained",
+                            "error" => ?e,
+                        );
+                        return Err(warp_utils::reject::custom_server_error(
+                            "unable to queue subscription, host may be overloaded or shutting down"
+                                .to_string(),
+                        ));
                     }
 
                     Ok(())

beacon_node/network/src/service.rs

@@ -27,6 +27,7 @@ use lighthouse_network::{
     MessageId, NetworkEvent, NetworkGlobals, PeerId,
 };
 use slog::{crit, debug, error, info, o, trace, warn};
+use std::collections::BTreeSet;
 use std::{collections::HashSet, pin::Pin, sync::Arc, time::Duration};
 use store::HotColdDB;
 use strum::IntoStaticStr;
@@ -119,7 +120,7 @@ pub enum NetworkMessage<T: EthSpec> {
 pub enum ValidatorSubscriptionMessage {
     /// Subscribes a list of validators to specific slots for attestation duties.
     AttestationSubscribe {
-        subscriptions: Vec<ValidatorSubscription>,
+        subscriptions: BTreeSet<ValidatorSubscription>,
     },
     SyncCommitteeSubscribe {
         subscriptions: Vec<SyncCommitteeSubscription>,
@@ -783,7 +784,7 @@ impl<T: BeaconChainTypes> NetworkService<T> {
             ValidatorSubscriptionMessage::AttestationSubscribe { subscriptions } => {
                 if let Err(e) = self
                     .attestation_service
-                    .validator_subscriptions(subscriptions)
+                    .validator_subscriptions(subscriptions.into_iter())
                 {
                     warn!(self.log, "Attestation validator subscription failed"; "error" => e);
                 }

beacon_node/network/src/subnet_service/attestation_subnets.rs

@@ -196,7 +196,7 @@ impl<T: BeaconChainTypes> AttestationService<T> {
     /// safely dropped.
     pub fn validator_subscriptions(
         &mut self,
-        subscriptions: Vec<ValidatorSubscription>,
+        subscriptions: impl Iterator<Item = ValidatorSubscription>,
     ) -> Result<(), String> {
         // If the node is in a proposer-only state, we ignore all subnet subscriptions.
         if self.proposer_only {
@@ -227,7 +227,6 @@ impl<T: BeaconChainTypes> AttestationService<T> {
                     warn!(self.log,
                         "Failed to compute subnet id for validator subscription";
                         "error" => ?e,
-                        "validator_index" => subscription.validator_index
                     );
                     continue;
                 }
@@ -257,13 +256,11 @@ impl<T: BeaconChainTypes> AttestationService<T> {
                     warn!(self.log,
                         "Subscription to subnet error";
                         "error" => e,
-                        "validator_index" => subscription.validator_index,
                     );
                 } else {
                     trace!(self.log,
                         "Subscribed to subnet for aggregator duties";
                         "exact_subnet" => ?exact_subnet,
-                        "validator_index" => subscription.validator_index
                     );
                 }
             }

beacon_node/network/src/subnet_service/tests/mod.rs

@@ -180,14 +180,12 @@ mod attestation_service {
     use super::*;
 
     fn get_subscription(
-        validator_index: u64,
         attestation_committee_index: CommitteeIndex,
         slot: Slot,
         committee_count_at_slot: u64,
         is_aggregator: bool,
     ) -> ValidatorSubscription {
         ValidatorSubscription {
-            validator_index,
             attestation_committee_index,
             slot,
             committee_count_at_slot,
@@ -204,7 +202,6 @@ mod attestation_service {
         (0..validator_count)
             .map(|validator_index| {
                 get_subscription(
-                    validator_index,
                     validator_index,
                     slot,
                     committee_count_at_slot,
@@ -217,7 +214,6 @@ mod attestation_service {
     #[tokio::test]
     async fn subscribe_current_slot_wait_for_unsubscribe() {
         // subscription config
-        let validator_index = 1;
         let committee_index = 1;
         // Keep a low subscription slot so that there are no additional subnet discovery events.
         let subscription_slot = 0;
@@ -233,7 +229,6 @@ mod attestation_service {
             .expect("Could not get current slot");
 
         let subscriptions = vec![get_subscription(
-            validator_index,
             committee_index,
             current_slot + Slot::new(subscription_slot),
             committee_count,
@@ -242,7 +237,7 @@ mod attestation_service {
 
         // submit the subscriptions
         attestation_service
-            .validator_subscriptions(subscriptions)
+            .validator_subscriptions(subscriptions.into_iter())
             .unwrap();
 
         // not enough time for peer discovery, just subscribe, unsubscribe
@@ -293,7 +288,6 @@ mod attestation_service {
     #[tokio::test]
     async fn test_same_subnet_unsubscription() {
         // subscription config
-        let validator_index = 1;
         let committee_count = 1;
         let subnets_per_node = MainnetEthSpec::default_spec().subnets_per_node as usize;
 
@@ -313,15 +307,13 @@ mod attestation_service {
             .expect("Could not get current slot");
 
         let sub1 = get_subscription(
-            validator_index,
             com1,
             current_slot + Slot::new(subscription_slot1),
             committee_count,
             true,
         );
 
         let sub2 = get_subscription(
-            validator_index,
             com2,
             current_slot + Slot::new(subscription_slot2),
             committee_count,
@@ -350,7 +342,7 @@ mod attestation_service {
 
         // submit the subscriptions
         attestation_service
-            .validator_subscriptions(vec![sub1, sub2])
+            .validator_subscriptions(vec![sub1, sub2].into_iter())
             .unwrap();
 
         // Unsubscription event should happen at slot 2 (since subnet id's are the same, unsubscription event should be at higher slot + 1)
@@ -431,7 +423,7 @@ mod attestation_service {
 
         // submit the subscriptions
         attestation_service
-            .validator_subscriptions(subscriptions)
+            .validator_subscriptions(subscriptions.into_iter())
             .unwrap();
 
         let events = get_events(&mut attestation_service, Some(131), 10).await;
@@ -501,7 +493,7 @@ mod attestation_service {
 
         // submit the subscriptions
         attestation_service
-            .validator_subscriptions(subscriptions)
+            .validator_subscriptions(subscriptions.into_iter())
             .unwrap();
 
         let events = get_events(&mut attestation_service, None, 3).await;
@@ -538,7 +530,6 @@ mod attestation_service {
     #[tokio::test]
     async fn test_subscribe_same_subnet_several_slots_apart() {
         // subscription config
-        let validator_index = 1;
         let committee_count = 1;
         let subnets_per_node = MainnetEthSpec::default_spec().subnets_per_node as usize;
 
@@ -558,15 +549,13 @@ mod attestation_service {
             .expect("Could not get current slot");
 
         let sub1 = get_subscription(
-            validator_index,
             com1,
             current_slot + Slot::new(subscription_slot1),
             committee_count,
             true,
         );
 
         let sub2 = get_subscription(
-            validator_index,
             com2,
             current_slot + Slot::new(subscription_slot2),
             committee_count,
@@ -595,7 +584,7 @@ mod attestation_service {
 
         // submit the subscriptions
         attestation_service
-            .validator_subscriptions(vec![sub1, sub2])
+            .validator_subscriptions(vec![sub1, sub2].into_iter())
             .unwrap();
 
         // Unsubscription event should happen at the end of the slot.
@@ -668,7 +657,7 @@ mod attestation_service {
 
         // submit the subscriptions
         attestation_service
-            .validator_subscriptions(subscriptions)
+            .validator_subscriptions(subscriptions.into_iter())
             .unwrap();
 
         // There should only be the same subscriptions as there are in the specification,

consensus/types/src/validator_subscription.rs

@@ -4,10 +4,8 @@ use ssz_derive::{Decode, Encode};
 
 /// A validator subscription, created when a validator subscribes to a slot to perform optional aggregation
 /// duties.
-#[derive(PartialEq, Debug, Serialize, Deserialize, Clone, Encode, Decode)]
+#[derive(PartialEq, Debug, Serialize, Deserialize, Clone, Encode, Decode, Eq, PartialOrd, Ord)]
 pub struct ValidatorSubscription {
-    /// The validators index.
-    pub validator_index: u64,
     /// The index of the committee within `slot` of which the validator is a member. Used by the
     /// beacon node to quickly evaluate the associated `SubnetId`.
     pub attestation_committee_index: CommitteeIndex,