Skip to content

Reduce load on validator subscription channels #5311

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Mar 7, 2024
Merged

Reduce load on validator subscription channels #5311

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6dd7547
Reduce subscription load on beacon node
pawanjay176 Feb 26, 2024
370e65d
Avoid reallocations
pawanjay176 Mar 7, 2024
67d5520
Merge branch 'unstable' into unclog-channels
pawanjay176 Mar 7, 2024
dcaa2dc
Fix tests
pawanjay176 Mar 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 28 additions & 28 deletions beacon_node/http_api/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
Expand Down
5 changes: 3 additions & 2 deletions beacon_node/network/src/service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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>,
Expand Down Expand Up @@ -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);
}
Expand Down
5 changes: 1 addition & 4 deletions beacon_node/network/src/subnet_service/attestation_subnets.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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
);
}
}
Expand Down
23 changes: 6 additions & 17 deletions beacon_node/network/src/subnet_service/tests/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -204,7 +202,6 @@ mod attestation_service {
(0..validator_count)
.map(|validator_index| {
get_subscription(
validator_index,
validator_index,
slot,
committee_count_at_slot,
Expand All @@ -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;
Expand All @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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;

Expand All @@ -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,
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand All @@ -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,
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down
4 changes: 1 addition & 3 deletions consensus/types/src/validator_subscription.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down