runtime: Collect stake delegations only once during epoch activation

Processing new epoch (`Bank::process_new_epoch`) involves collecting
stake delegations twice:

1) In `Bank::compute_new_epoch_caches_and_rewards`, to create a stake
   history entry and refresh vote accounts.
2) In `Bank::get_epoch_reward_calculate_param_info`, which is then used
   in `Bank::calculate_stake_vote_rewards` to calculate rewards for
   stakers and voters.

The overall time of crossing the epoch boundary is ~519ms:

```
update_epoch_us=519953i
```

Where the two heaviest operations are `collect()`` calls on stake
delegations, each of them taking ~200-220ms.

Reduce that to just one collect by passing the vector 1) with freshly
computed stake history and vote accounts to `Bank::begin_partitioned_rewards`.
This way, we can avoid calling `Bank::get_epoch_reward_calculate_param_info`.

The new time of crossing the epoch boundary is ~337ms:

```
update_epoch_us=337371i
```

Making that change possible required several refactors:

* Tale `&PointValue` in `Bank::create_epoch_rewards_sysvar`. That makes
  it easier to operate on references of `PartitionedRewardsCalculation`.
  Copying integers from `PointValue` is cheap and has no visible
  performance impact.
* Split `Stakes::activate_epoch`, that was performing calculations and
  mutating the cache at the same time. The calculations got split to
  `Stakes::calculate_activated_stake` that takes `&self`.
* Add `Stakes::stake_delegations_ves` method. Stake delegations are
  stored as hash array mapped trie (HAMT)[0], which means that inserts,
  deletions and lookups are average-case O(1) and worst-case O(log n).
  However, the performance of iterations is poor due to depth-first
  traversal and jumps. Currently it's also impossible to iterate over it
  with rayon. That issue is known and handled by converting the HAMT to
  a vector with `stakes.stake_delegations.iter().collect()`. Move that
  trick to a dedicated method that describes the performance
  consequences.
* Add `FilteredStakeDelegation` wrapper type, that wraps a vector of
  stake delegations and acts as a lazy iterator that filters out ones
  with insufficient stake.
* Split the code dealing with rewards calculation and vote rewards
  distribution into separate methods:
  * `Bank::calculate_rewards` that takes `&self` and does not acquire
    any locks.
  * `Bank::begin_partitioned_rewards` that takes `&mut self`, sets
    calculation status and creates a sysvar.
  * `Bank::distribute_vote_rewards` that stores partitioned rewards and
    increases capitalization.

[0] https://en.wikipedia.org/wiki/Hash_array_mapped_trie

Author: vadorovsky

runtime/src/bank.rs

@@ -47,7 +47,7 @@ use {
         rent_collector::RentCollector,
         runtime_config::RuntimeConfig,
         stake_account::StakeAccount,
-        stake_utils,
+        stake_history::StakeHistory as CowStakeHistory,
         stake_weighted_timestamp::{
             calculate_stake_weighted_timestamp, MaxAllowableDrift,
             MAX_ALLOWABLE_DRIFT_PERCENTAGE_FAST, MAX_ALLOWABLE_DRIFT_PERCENTAGE_SLOW_V2,
@@ -112,7 +112,6 @@ use {
     solana_lattice_hash::lt_hash::LtHash,
     solana_measure::{measure::Measure, measure_time, measure_us},
     solana_message::{inner_instruction::InnerInstructions, AccountKeys, SanitizedMessage},
-    solana_native_token::LAMPORTS_PER_SOL,
     solana_packet::PACKET_DATA_SIZE,
     solana_precompile_error::PrecompileError,
     solana_program_runtime::{
@@ -167,7 +166,7 @@ use {
         transaction_accounts::KeyedAccountSharedData, TransactionReturnData,
     },
     solana_transaction_error::{TransactionError, TransactionResult as Result},
-    solana_vote::vote_account::{VoteAccount, VoteAccountsHashMap},
+    solana_vote::vote_account::{VoteAccount, VoteAccounts, VoteAccountsHashMap},
     std::{
         collections::{HashMap, HashSet},
         fmt,
@@ -1050,6 +1049,14 @@ impl AtomicBankHashStats {
     }
 }
 
+struct NewEpochBundle {
+    stake_history: CowStakeHistory,
+    vote_accounts: VoteAccounts,
+    rewards_calculation: Arc<PartitionedRewardsCalculation>,
+    calculate_activated_stake_time_us: u64,
+    update_rewards_with_thread_pool_time_us: u64,
+}
+
 impl Bank {
     fn default_with_accounts(accounts: Accounts) -> Self {
         let mut bank = Self {
@@ -1591,6 +1598,47 @@ impl Bank {
             .new_warmup_cooldown_rate_epoch(&self.epoch_schedule)
     }
 
+    /// Returns updated stake history and vote accounts that includes new
+    /// activated stake from the last epoch.
+    fn compute_new_epoch_caches_and_rewards(
+        &self,
+        thread_pool: &ThreadPool,
+        parent_epoch: Epoch,
+        reward_calc_tracer: Option<impl RewardCalcTracer>,
+        rewards_metrics: &mut RewardsMetrics,
+    ) -> NewEpochBundle {
+        // Add new entry to stakes.stake_history, set appropriate epoch and
+        // update vote accounts with warmed up stakes before saving a
+        // snapshot of stakes in epoch stakes
+        let stakes = self.stakes_cache.stakes();
+        let stake_delegations = stakes.stake_delegations_vec();
+        let ((stake_history, vote_accounts), calculate_activated_stake_time_us) =
+            measure_us!(stakes.calculate_activated_stake(
+                self.epoch(),
+                thread_pool,
+                self.new_warmup_cooldown_rate_epoch(),
+                &stake_delegations
+            ));
+        // Apply stake rewards and commission using new snapshots.
+        let (rewards_calculation, update_rewards_with_thread_pool_time_us) = measure_us!(self
+            .calculate_rewards(
+                &stake_history,
+                stake_delegations,
+                &vote_accounts,
+                parent_epoch,
+                reward_calc_tracer,
+                thread_pool,
+                rewards_metrics,
+            ));
+        NewEpochBundle {
+            stake_history,
+            vote_accounts,
+            rewards_calculation,
+            calculate_activated_stake_time_us,
+            update_rewards_with_thread_pool_time_us,
+        }
+    }
+
     /// process for the start of a new epoch
     fn process_new_epoch(
         &mut self,
@@ -1610,31 +1658,37 @@ impl Bank {
             thread_pool.install(|| { self.compute_and_apply_new_feature_activations() })
         );
 
-        // Add new entry to stakes.stake_history, set appropriate epoch and
-        // update vote accounts with warmed up stakes before saving a
-        // snapshot of stakes in epoch stakes
-        let (_, activate_epoch_time_us) = measure_us!(self.stakes_cache.activate_epoch(
-            epoch,
+        let mut rewards_metrics = RewardsMetrics::default();
+        let NewEpochBundle {
+            stake_history,
+            vote_accounts,
+            rewards_calculation,
+            calculate_activated_stake_time_us,
+            update_rewards_with_thread_pool_time_us,
+        } = self.compute_new_epoch_caches_and_rewards(
             &thread_pool,
-            self.new_warmup_cooldown_rate_epoch()
-        ));
+            parent_epoch,
+            reward_calc_tracer,
+            &mut rewards_metrics,
+        );
+
+        self.stakes_cache
+            .activate_epoch(epoch, stake_history, vote_accounts);
 
         // Save a snapshot of stakes for use in consensus and stake weighted networking
         let leader_schedule_epoch = self.epoch_schedule.get_leader_schedule_epoch(slot);
         let (_, update_epoch_stakes_time_us) =
             measure_us!(self.update_epoch_stakes(leader_schedule_epoch));
 
-        let mut rewards_metrics = RewardsMetrics::default();
-        // After saving a snapshot of stakes, apply stake rewards and commission
-        let (_, update_rewards_with_thread_pool_time_us) = measure_us!(self
-            .begin_partitioned_rewards(
-                reward_calc_tracer,
-                &thread_pool,
-                parent_epoch,
-                parent_slot,
-                parent_height,
-                &mut rewards_metrics,
-            ));
+        // Distribute rewards commission to vote accounts and cache stake rewards
+        // for partitioned distribution in the upcoming slots.
+        self.begin_partitioned_rewards(
+            parent_slot,
+            parent_height,
+            parent_epoch,
+            &rewards_calculation,
+            &rewards_metrics,
+        );
 
         report_new_epoch_metrics(
             epoch,
@@ -1643,7 +1697,7 @@ impl Bank {
             NewEpochTimings {
                 thread_pool_time_us,
                 apply_feature_activations_time_us,
-                activate_epoch_time_us,
+                calculate_activated_stake_time_us,
                 update_epoch_stakes_time_us,
                 update_rewards_with_thread_pool_time_us,
             },
@@ -2301,41 +2355,6 @@ impl Bank {
         }
     }
 
-    fn filter_stake_delegations<'a>(
-        &self,
-        stakes: &'a Stakes<StakeAccount<Delegation>>,
-    ) -> Vec<(&'a Pubkey, &'a StakeAccount<Delegation>)> {
-        if self
-            .feature_set
-            .is_active(&feature_set::stake_minimum_delegation_for_rewards::id())
-        {
-            let num_stake_delegations = stakes.stake_delegations().len();
-            let min_stake_delegation = stake_utils::get_minimum_delegation(
-                self.feature_set
-                    .is_active(&agave_feature_set::stake_raise_minimum_delegation_to_1_sol::id()),
-            )
-            .max(LAMPORTS_PER_SOL);
-
-            let (stake_delegations, filter_time_us) = measure_us!(stakes
-                .stake_delegations()
-                .iter()
-                .filter(|(_stake_pubkey, cached_stake_account)| {
-                    cached_stake_account.delegation().stake >= min_stake_delegation
-                })
-                .collect::<Vec<_>>());
-
-            datapoint_info!(
-                "stake_account_filter_time",
-                ("filter_time_us", filter_time_us, i64),
-                ("num_stake_delegations_before", num_stake_delegations, i64),
-                ("num_stake_delegations_after", stake_delegations.len(), i64)
-            );
-            stake_delegations
-        } else {
-            stakes.stake_delegations().iter().collect()
-        }
-    }
-
     /// Convert computed VoteRewards to VoteRewardsAccounts for storing.
     ///
     /// This function processes vote rewards and consolidates them into a single

runtime/src/bank/metrics.rs

@@ -11,7 +11,7 @@ use {
 pub(crate) struct NewEpochTimings {
     pub(crate) thread_pool_time_us: u64,
     pub(crate) apply_feature_activations_time_us: u64,
-    pub(crate) activate_epoch_time_us: u64,
+    pub(crate) calculate_activated_stake_time_us: u64,
     pub(crate) update_epoch_stakes_time_us: u64,
     pub(crate) update_rewards_with_thread_pool_time_us: u64,
 }
@@ -63,7 +63,11 @@ pub(crate) fn report_new_epoch_metrics(
             timings.apply_feature_activations_time_us,
             i64
         ),
-        ("activate_epoch_us", timings.activate_epoch_time_us, i64),
+        (
+            "calculate_activated_stake_us",
+            timings.calculate_activated_stake_time_us,
+            i64
+        ),
         (
             "update_epoch_stakes_us",
             timings.update_epoch_stakes_time_us,