gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-06 10:11:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'electra-epoch-proc' of https://github.com/sigp/lighthouse into electra-engine-api

Browse Source
This commit is contained in:
realbigsean
2024-05-10 10:00:42 -04:00
parent 9b5ea9d867 08e045875f
commit 6d2c396ef2
4 changed files with 369 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
consensus/state_processing/src/epoch_cache.rs
View File

@@ -9,6 +9,7 @@ use types::{ActivationQueue, BeaconState, ChainSpec, EthSpec, ForkName, Hash256}
pub struct PreEpochCache {
epoch_key: EpochCacheKey,
effective_balances: Vec<u64>,
total_active_balance: u64,
}
impl PreEpochCache {
@@ -36,20 +37,51 @@ impl PreEpochCache {
Ok(Self {
epoch_key,
effective_balances: Vec::with_capacity(state.validators().len()),
total_active_balance: 0,
})
}
pub fn push_effective_balance(&mut self, effective_balance: u64) {
self.effective_balances.push(effective_balance);
pub fn update_effective_balance(
&mut self,
validator_index: usize,
effective_balance: u64,
is_active_next_epoch: bool,
) -> Result<(), EpochCacheError> {
if validator_index == self.effective_balances.len() {
self.effective_balances.push(effective_balance);
if is_active_next_epoch {
self.total_active_balance
.safe_add_assign(effective_balance)?;
}
Ok(())
} else if let Some(existing_balance) = self.effective_balances.get_mut(validator_index) {
// Update total active balance for a late change in effective balance. This happens when
// processing consolidations.
if is_active_next_epoch {
self.total_active_balance
.safe_add_assign(effective_balance)?;
self.total_active_balance
.safe_sub_assign(*existing_balance)?;
}
*existing_balance = effective_balance;
Ok(())
} else {
Err(EpochCacheError::ValidatorIndexOutOfBounds { validator_index })
}
}
pub fn get_total_active_balance(&self) -> u64 {
self.total_active_balance
}
pub fn into_epoch_cache(
self,
total_active_balance: u64,
activation_queue: ActivationQueue,
spec: &ChainSpec,
) -> Result<EpochCache, EpochCacheError> {
let epoch = self.epoch_key.epoch;
let total_active_balance = self.total_active_balance;
let sqrt_total_active_balance = SqrtTotalActiveBalance::new(total_active_balance);
let base_reward_per_increment = BaseRewardPerIncrement::new(total_active_balance, spec)?;
@@ -131,9 +163,9 @@ pub fn initialize_epoch_cache<E: EthSpec>(
decision_block_root,
},
effective_balances,
total_active_balance,
};
*state.epoch_cache_mut() =
pre_epoch_cache.into_epoch_cache(total_active_balance, activation_queue, spec)?;
*state.epoch_cache_mut() = pre_epoch_cache.into_epoch_cache(activation_queue, spec)?;
Ok(())
}

1
consensus/state_processing/src/per_epoch_processing/errors.rs
View File

@@ -25,6 +25,7 @@ pub enum EpochProcessingError {
InvalidFlagIndex(usize),
MilhouseError(milhouse::Error),
EpochCache(EpochCacheError),
SinglePassMissingActivationQueue,
}
impl From<InclusionError> for EpochProcessingError {

338
consensus/state_processing/src/per_epoch_processing/single_pass.rs
View File

@@ -1,20 +1,24 @@
use crate::{
common::update_progressive_balances_cache::initialize_progressive_balances_cache,
common::{
decrease_balance, increase_balance,
update_progressive_balances_cache::initialize_progressive_balances_cache,
},
epoch_cache::{initialize_epoch_cache, PreEpochCache},
per_epoch_processing::{Delta, Error, ParticipationEpochSummary},
};
use itertools::izip;
use safe_arith::{SafeArith, SafeArithIter};
use std::cmp::{max, min};
use std::collections::BTreeSet;
use std::collections::{BTreeSet, HashMap};
use types::{
consts::altair::{
NUM_FLAG_INDICES, PARTICIPATION_FLAG_WEIGHTS, TIMELY_HEAD_FLAG_INDEX,
TIMELY_TARGET_FLAG_INDEX, WEIGHT_DENOMINATOR,
},
milhouse::Cow,
ActivationQueue, BeaconState, BeaconStateError, ChainSpec, Epoch, EthSpec, ExitCache, ForkName,
ParticipationFlags, ProgressiveBalancesCache, RelativeEpoch, Unsigned, Validator,
ActivationQueue, BeaconState, BeaconStateError, ChainSpec, Checkpoint, Epoch, EthSpec,
ExitCache, ForkName, List, ParticipationFlags, ProgressiveBalancesCache, RelativeEpoch,
Unsigned, Validator,
};
pub struct SinglePassConfig {
@@ -22,6 +26,8 @@ pub struct SinglePassConfig {
pub rewards_and_penalties: bool,
pub registry_updates: bool,
pub slashings: bool,
pub pending_balance_deposits: bool,
pub pending_consolidations: bool,
pub effective_balance_updates: bool,
}
@@ -38,6 +44,8 @@ impl SinglePassConfig {
rewards_and_penalties: true,
registry_updates: true,
slashings: true,
pending_balance_deposits: true,
pending_consolidations: true,
effective_balance_updates: true,
}
}
@@ -48,6 +56,8 @@ impl SinglePassConfig {
rewards_and_penalties: false,
registry_updates: false,
slashings: false,
pending_balance_deposits: false,
pending_consolidations: false,
effective_balance_updates: false,
}
}
@@ -57,6 +67,7 @@ impl SinglePassConfig {
struct StateContext {
current_epoch: Epoch,
next_epoch: Epoch,
finalized_checkpoint: Checkpoint,
is_in_inactivity_leak: bool,
total_active_balance: u64,
churn_limit: u64,
@@ -73,6 +84,15 @@ struct SlashingsContext {
target_withdrawable_epoch: Epoch,
}
struct PendingBalanceDepositsContext {
/// The value to set `next_deposit_index` to *after* processing completes.
next_deposit_index: usize,
/// The value to set `deposit_balance_to_consume` to *after* processing completes.
deposit_balance_to_consume: u64,
/// Total balance increases for each validator due to pending balance deposits.
validator_deposits_to_process: HashMap<usize, u64>,
}
struct EffectiveBalancesContext {
downward_threshold: u64,
upward_threshold: u64,
@@ -129,6 +149,7 @@ pub fn process_epoch_single_pass<E: EthSpec>(
let state_ctxt = &StateContext {
current_epoch,
next_epoch,
finalized_checkpoint,
is_in_inactivity_leak,
total_active_balance,
churn_limit,
@@ -139,6 +160,13 @@ pub fn process_epoch_single_pass<E: EthSpec>(
let slashings_ctxt = &SlashingsContext::new(state, state_ctxt, spec)?;
let mut next_epoch_cache = PreEpochCache::new_for_next_epoch(state)?;
let pending_balance_deposits_ctxt =
if fork_name >= ForkName::Electra && conf.pending_balance_deposits {
Some(PendingBalanceDepositsContext::new(state, spec)?)
} else {
None
};
// Split the state into several disjoint mutable borrows.
let (
validators,
@@ -165,12 +193,19 @@ pub fn process_epoch_single_pass<E: EthSpec>(
// Compute shared values required for different parts of epoch processing.
let rewards_ctxt = &RewardsAndPenaltiesContext::new(progressive_balances, state_ctxt, spec)?;
let activation_queue = &epoch_cache
.activation_queue()?
.get_validators_eligible_for_activation(
finalized_checkpoint.epoch,
activation_churn_limit as usize,
);
let mut activation_queues = if fork_name < ForkName::Electra {
let activation_queue = epoch_cache
.activation_queue()?
.get_validators_eligible_for_activation(
finalized_checkpoint.epoch,
activation_churn_limit as usize,
);
let next_epoch_activation_queue = ActivationQueue::default();
Some((activation_queue, next_epoch_activation_queue))
} else {
None
};
let effective_balances_ctxt = &EffectiveBalancesContext::new(spec)?;
// Iterate over the validators and related fields in one pass.
@@ -178,10 +213,6 @@ pub fn process_epoch_single_pass<E: EthSpec>(
let mut balances_iter = balances.iter_cow();
let mut inactivity_scores_iter = inactivity_scores.iter_cow();
// Values computed for the next epoch transition.
let mut next_epoch_total_active_balance = 0;
let mut next_epoch_activation_queue = ActivationQueue::default();
for (index, &previous_epoch_participation, &current_epoch_participation) in izip!(
0..num_validators,
previous_epoch_participation.iter(),
@@ -246,12 +277,14 @@ pub fn process_epoch_single_pass<E: EthSpec>(
// `process_registry_updates`
if conf.registry_updates {
let activation_queue_refs = activation_queues
.as_mut()
.map(|(current_queue, next_queue)| (&*current_queue, next_queue));
process_single_registry_update(
&mut validator,
validator_info,
exit_cache,
activation_queue,
&mut next_epoch_activation_queue,
activation_queue_refs,
state_ctxt,
spec,
)?;
@@ -262,13 +295,22 @@ pub fn process_epoch_single_pass<E: EthSpec>(
process_single_slashing(&mut balance, &validator, slashings_ctxt, state_ctxt, spec)?;
}
// `process_pending_balance_deposits`
if let Some(pending_balance_deposits_ctxt) = &pending_balance_deposits_ctxt {
process_pending_balance_deposits_for_validator(
&mut balance,
validator_info,
pending_balance_deposits_ctxt,
)?;
}
// `process_effective_balance_updates`
if conf.effective_balance_updates {
process_single_effective_balance_update(
validator_info.index,
*balance,
&mut validator,
validator_info,
&mut next_epoch_total_active_balance,
validator_info.current_epoch_participation,
&mut next_epoch_cache,
progressive_balances,
effective_balances_ctxt,
@@ -278,15 +320,45 @@ pub fn process_epoch_single_pass<E: EthSpec>(
}
}
if conf.effective_balance_updates {
state.set_total_active_balance(next_epoch, next_epoch_total_active_balance, spec);
*state.epoch_cache_mut() = next_epoch_cache.into_epoch_cache(
next_epoch_total_active_balance,
next_epoch_activation_queue,
// Finish processing pending balance deposits if relevant.
//
// This *could* be reordered after `process_pending_consolidations` which pushes only to the end
// of the `pending_balance_deposits` list. But we may as well preserve the write ordering used
// by the spec and do this first.
if let Some(ctxt) = pending_balance_deposits_ctxt {
let new_pending_balance_deposits = List::try_from_iter(
state
.pending_balance_deposits()?
.iter_from(ctxt.next_deposit_index)?
.cloned(),
)?;
*state.pending_balance_deposits_mut()? = new_pending_balance_deposits;
*state.deposit_balance_to_consume_mut()? = ctxt.deposit_balance_to_consume;
}
// Process consolidations outside the single-pass loop, as they depend on balances for multiple
// validators and cannot be computed accurately inside the loop.
if fork_name >= ForkName::Electra && conf.pending_consolidations {
process_pending_consolidations(
state,
&mut next_epoch_cache,
effective_balances_ctxt,
state_ctxt,
spec,
)?;
}
// Finally, finish updating effective balance caches. We need this to happen *after* processing
// of pending consolidations, which recomputes some effective balances.
if conf.effective_balance_updates {
let next_epoch_total_active_balance = next_epoch_cache.get_total_active_balance();
state.set_total_active_balance(next_epoch, next_epoch_total_active_balance, spec);
let next_epoch_activation_queue =
activation_queues.map_or_else(ActivationQueue::default, |(_, queue)| queue);
*state.epoch_cache_mut() =
next_epoch_cache.into_epoch_cache(next_epoch_activation_queue, spec)?;
}
Ok(summary)
}
@@ -456,6 +528,31 @@ impl RewardsAndPenaltiesContext {
}
fn process_single_registry_update(
validator: &mut Cow<Validator>,
validator_info: &ValidatorInfo,
exit_cache: &mut ExitCache,
activation_queues: Option<(&BTreeSet<usize>, &mut ActivationQueue)>,
state_ctxt: &StateContext,
spec: &ChainSpec,
) -> Result<(), Error> {
if state_ctxt.fork_name < ForkName::Electra {
let (activation_queue, next_epoch_activation_queue) =
activation_queues.ok_or(Error::SinglePassMissingActivationQueue)?;
process_single_registry_update_pre_electra(
validator,
validator_info,
exit_cache,
activation_queue,
next_epoch_activation_queue,
state_ctxt,
spec,
)
} else {
process_single_registry_update_post_electra(validator, exit_cache, state_ctxt, spec)
}
}
fn process_single_registry_update_pre_electra(
validator: &mut Cow<Validator>,
validator_info: &ValidatorInfo,
exit_cache: &mut ExitCache,
@@ -491,6 +588,35 @@ fn process_single_registry_update(
Ok(())
}
fn process_single_registry_update_post_electra(
validator: &mut Cow<Validator>,
exit_cache: &mut ExitCache,
state_ctxt: &StateContext,
spec: &ChainSpec,
) -> Result<(), Error> {
let current_epoch = state_ctxt.current_epoch;
if validator.is_eligible_for_activation_queue(spec, state_ctxt.fork_name) {
validator.make_mut()?.activation_eligibility_epoch = current_epoch.safe_add(1)?;
}
if validator.is_active_at(current_epoch) && validator.effective_balance <= spec.ejection_balance
{
// TODO(electra): make sure initiate_validator_exit is updated
initiate_validator_exit(validator, exit_cache, state_ctxt, spec)?;
}
if validator.is_eligible_for_activation_with_finalized_checkpoint(
&state_ctxt.finalized_checkpoint,
spec,
) {
validator.make_mut()?.activation_epoch =
spec.compute_activation_exit_epoch(current_epoch)?;
}
Ok(())
}
fn initiate_validator_exit(
validator: &mut Cow<Validator>,
exit_cache: &mut ExitCache,
@@ -568,6 +694,146 @@ fn process_single_slashing(
Ok(())
}
impl PendingBalanceDepositsContext {
fn new<E: EthSpec>(state: &BeaconState<E>, spec: &ChainSpec) -> Result<Self, Error> {
let available_for_processing = state
.deposit_balance_to_consume()?
.safe_add(state.get_activation_exit_churn_limit(spec)?)?;
let mut processed_amount = 0;
let mut next_deposit_index = 0;
let mut validator_deposits_to_process = HashMap::new();
let pending_balance_deposits = state.pending_balance_deposits()?;
for deposit in pending_balance_deposits.iter() {
if processed_amount.safe_add(deposit.amount)? > available_for_processing {
break;
}
validator_deposits_to_process
.entry(deposit.index as usize)
.or_insert(0)
.safe_add_assign(deposit.amount)?;
processed_amount.safe_add_assign(deposit.amount)?;
next_deposit_index.safe_add_assign(1)?;
}
let deposit_balance_to_consume = if next_deposit_index == pending_balance_deposits.len() {
0
} else {
available_for_processing.safe_sub(processed_amount)?
};
Ok(Self {
next_deposit_index,
deposit_balance_to_consume,
validator_deposits_to_process,
})
}
}
fn process_pending_balance_deposits_for_validator(
balance: &mut Cow<u64>,
validator_info: &ValidatorInfo,
pending_balance_deposits_ctxt: &PendingBalanceDepositsContext,
) -> Result<(), Error> {
if let Some(deposit_amount) = pending_balance_deposits_ctxt
.validator_deposits_to_process
.get(&validator_info.index)
{
balance.make_mut()?.safe_add_assign(*deposit_amount)?;
}
Ok(())
}
/// We process pending consolidations after all of single-pass epoch processing, and then patch up
/// the effective balances for affected validators.
///
/// This is safe because processing consolidations does not depend on the `effective_balance`.
fn process_pending_consolidations<E: EthSpec>(
state: &mut BeaconState<E>,
next_epoch_cache: &mut PreEpochCache,
effective_balances_ctxt: &EffectiveBalancesContext,
state_ctxt: &StateContext,
spec: &ChainSpec,
) -> Result<(), Error> {
let mut next_pending_consolidation: usize = 0;
let current_epoch = state.current_epoch();
let pending_consolidations = state.pending_consolidations()?.clone();
let mut affected_validators = BTreeSet::new();
for pending_consolidation in &pending_consolidations {
let source_index = pending_consolidation.source_index as usize;
let target_index = pending_consolidation.target_index as usize;
let source_validator = state.get_validator(source_index)?;
if source_validator.slashed {
next_pending_consolidation.safe_add_assign(1)?;
continue;
}
if source_validator.withdrawable_epoch > current_epoch {
break;
}
// Calculate the active balance while we have the source validator loaded. This is a safe
// reordering.
let source_balance = *state
.balances()
.get(source_index)
.ok_or(BeaconStateError::UnknownValidator(source_index))?;
let active_balance =
source_validator.get_active_balance(source_balance, spec, state_ctxt.fork_name);
// Churn any target excess active balance of target and raise its max.
state.switch_to_compounding_validator(target_index, spec)?;
// Move active balance to target. Excess balance is withdrawable.
decrease_balance(state, source_index, active_balance)?;
increase_balance(state, target_index, active_balance)?;
affected_validators.insert(source_index);
affected_validators.insert(target_index);
next_pending_consolidation.safe_add_assign(1)?;
}
let new_pending_consolidations = List::try_from_iter(
state
.pending_consolidations()?
.iter_from(next_pending_consolidation)?
.cloned(),
)?;
*state.pending_consolidations_mut()? = new_pending_consolidations;
// Re-process effective balance updates for validators affected by consolidations.
let (validators, balances, _, current_epoch_participation, _, progressive_balances, _, _) =
state.mutable_validator_fields()?;
for validator_index in affected_validators {
let balance = *balances
.get(validator_index)
.ok_or(BeaconStateError::UnknownValidator(validator_index))?;
let mut validator = validators
.get_cow(validator_index)
.ok_or(BeaconStateError::UnknownValidator(validator_index))?;
let validator_current_epoch_participation = *current_epoch_participation
.get(validator_index)
.ok_or(BeaconStateError::UnknownValidator(validator_index))?;
process_single_effective_balance_update(
validator_index,
balance,
&mut validator,
validator_current_epoch_participation,
next_epoch_cache,
progressive_balances,
effective_balances_ctxt,
state_ctxt,
spec,
)?;
}
Ok(())
}
impl EffectiveBalancesContext {
fn new(spec: &ChainSpec) -> Result<Self, Error> {
let hysteresis_increment = spec
@@ -584,18 +850,24 @@ impl EffectiveBalancesContext {
}
}
/// This function abstracts over phase0 and Electra effective balance processing.
#[allow(clippy::too_many_arguments)]
fn process_single_effective_balance_update(
validator_index: usize,
balance: u64,
validator: &mut Cow<Validator>,
validator_info: &ValidatorInfo,
next_epoch_total_active_balance: &mut u64,
validator_current_epoch_participation: ParticipationFlags,
next_epoch_cache: &mut PreEpochCache,
progressive_balances: &mut ProgressiveBalancesCache,
eb_ctxt: &EffectiveBalancesContext,
state_ctxt: &StateContext,
spec: &ChainSpec,
) -> Result<(), Error> {
// Use the higher effective balance limit if post-Electra and compounding withdrawal credentials
// are set.
let effective_balance_limit =
validator.get_validator_max_effective_balance(spec, state_ctxt.fork_name);
let old_effective_balance = validator.effective_balance;
let new_effective_balance = if balance.safe_add(eb_ctxt.downward_threshold)?
< validator.effective_balance
@@ -606,15 +878,13 @@ fn process_single_effective_balance_update(
{
min(
balance.safe_sub(balance.safe_rem(spec.effective_balance_increment)?)?,
spec.max_effective_balance,
effective_balance_limit,
)
} else {
validator.effective_balance
};
if validator.is_active_at(state_ctxt.next_epoch) {
next_epoch_total_active_balance.safe_add_assign(new_effective_balance)?;
}
let is_active_next_epoch = validator.is_active_at(state_ctxt.next_epoch);
if new_effective_balance != old_effective_balance {
validator.make_mut()?.effective_balance = new_effective_balance;
@@ -623,14 +893,18 @@ fn process_single_effective_balance_update(
// previous epoch once the epoch transition completes.
progressive_balances.on_effective_balance_change(
validator.slashed,
validator_info.current_epoch_participation,
validator_current_epoch_participation,
old_effective_balance,
new_effective_balance,
)?;
}
// Caching: update next epoch effective balances.
next_epoch_cache.push_effective_balance(new_effective_balance);
// Caching: update next epoch effective balances and total active balance.
next_epoch_cache.update_effective_balance(
validator_index,
new_effective_balance,
is_active_next_epoch,
)?;
Ok(())
}

30
consensus/types/src/validator.rs
View File

@@ -1,6 +1,6 @@
use crate::{
test_utils::TestRandom, Address, BeaconState, ChainSpec, Epoch, EthSpec, ForkName, Hash256,
PublicKeyBytes,
test_utils::TestRandom, Address, BeaconState, ChainSpec, Checkpoint, Epoch, EthSpec, ForkName,
Hash256, PublicKeyBytes,
};
use serde::{Deserialize, Serialize};
use ssz_derive::{Decode, Encode};
@@ -87,15 +87,25 @@ impl Validator {
}
/// Returns `true` if the validator is eligible to be activated.
///
/// Spec v0.12.1
pub fn is_eligible_for_activation<E: EthSpec>(
&self,
state: &BeaconState<E>,
spec: &ChainSpec,
) -> bool {
self.is_eligible_for_activation_with_finalized_checkpoint(
&state.finalized_checkpoint(),
spec,
)
}
/// Returns `true` if the validator is eligible to be activated.
pub fn is_eligible_for_activation_with_finalized_checkpoint(
&self,
finalized_checkpoint: &Checkpoint,
spec: &ChainSpec,
) -> bool {
// Placement in queue is finalized
self.activation_eligibility_epoch <= state.finalized_checkpoint().epoch
self.activation_eligibility_epoch <= finalized_checkpoint.epoch
// Has not yet been activated
&& self.activation_epoch == spec.far_future_epoch
}
@@ -267,6 +277,16 @@ impl Validator {
spec.max_effective_balance
}
}
pub fn get_active_balance(
&self,
validator_balance: u64,
spec: &ChainSpec,
current_fork: ForkName,
) -> u64 {
let max_effective_balance = self.get_validator_max_effective_balance(spec, current_fork);
std::cmp::min(validator_balance, max_effective_balance)
}
}
impl Default for Validator {