Split out voting and banking threads in banking stage (#27931)

* Split out voting and banking threads in banking stage

Additionally this allows us to aggressively prune the buffer for voting threads
as with the new vote state only the latest vote from each validator is
necessary.

* Update local cluster test to use new Vote ix

* Encapsulate transaction storage filtering better

* Address pr comments

* Commit cargo lock change

* clippy

* Remove unsafe impls

* pr comments

* compute_sanitized_transaction -> build_sanitized_transaction

* &Arc -> Arc

* Move test

* Refactor metrics enums

* clippy

Cargo.lock

@@ -6159,6 +6159,7 @@ dependencies = [
  "solana-logger 1.15.0",
  "solana-measure",
  "solana-metrics",
+ "solana-perf",
  "solana-program-runtime",
  "solana-rayon-threadlimit",
  "solana-sdk 1.15.0",

core/benches/banking_stage.rs

@@ -13,6 +13,7 @@ use {
         leader_slot_banking_stage_metrics::LeaderSlotMetricsTracker,
         qos_service::QosService,
         unprocessed_packet_batches::*,
+        unprocessed_transaction_storage::{ThreadType, UnprocessedTransactionStorage},
     },
     solana_entry::entry::{next_hash, Entry},
     solana_gossip::cluster_info::{ClusterInfo, Node},
@@ -83,8 +84,10 @@ fn bench_consume_buffered(bencher: &mut Bencher) {
         let transactions = vec![tx; 4194304];
         let batches = transactions_to_deserialized_packets(&transactions).unwrap();
         let batches_len = batches.len();
-        let mut transaction_buffer =
+        let mut transaction_buffer = UnprocessedTransactionStorage::new_transaction_storage(
-            UnprocessedPacketBatches::from_iter(batches.into_iter(), 2 * batches_len);
+            UnprocessedPacketBatches::from_iter(batches.into_iter(), 2 * batches_len),
+            ThreadType::Transactions,
+        );
         let (s, _r) = unbounded();
         // This tests the performance of buffering packets.
         // If the packet buffers are copied, performance will be poor.
@@ -94,7 +97,7 @@ fn bench_consume_buffered(bencher: &mut Bencher) {
                 std::u128::MAX,
                 &poh_recorder,
                 &mut transaction_buffer,
-                None,
+                &None,
                 &s,
                 None::<Box<dyn Fn()>>,
                 &BankingStageStats::default(),

core/benches/unprocessed_packet_batches.rs

@@ -6,8 +6,11 @@ extern crate test;
 use {
     rand::distributions::{Distribution, Uniform},
     solana_core::{
-        banking_stage::*, forward_packet_batches_by_accounts::ForwardPacketBatchesByAccounts,
+        forward_packet_batches_by_accounts::ForwardPacketBatchesByAccounts,
         unprocessed_packet_batches::*,
+        unprocessed_transaction_storage::{
+            ThreadType, UnprocessedTransactionStorage, UNPROCESSED_BUFFER_STEP_SIZE,
+        },
     },
     solana_measure::measure::Measure,
     solana_perf::packet::{Packet, PacketBatch},
@@ -104,7 +107,7 @@ fn insert_packet_batches(
 #[allow(clippy::unit_arg)]
 fn bench_packet_clone(bencher: &mut Bencher) {
     let batch_count = 1000;
-    let packet_per_batch_count = 128;
+    let packet_per_batch_count = UNPROCESSED_BUFFER_STEP_SIZE;
 
     let packet_batches: Vec<PacketBatch> = (0..batch_count)
         .map(|_| build_packet_batch(packet_per_batch_count, None).0)
@@ -134,9 +137,9 @@ fn bench_packet_clone(bencher: &mut Bencher) {
 #[bench]
 #[ignore]
 fn bench_unprocessed_packet_batches_within_limit(bencher: &mut Bencher) {
-    let buffer_capacity = 1_000 * 128;
+    let buffer_capacity = 1_000 * UNPROCESSED_BUFFER_STEP_SIZE;
     let batch_count = 1_000;
-    let packet_per_batch_count = 128;
+    let packet_per_batch_count = UNPROCESSED_BUFFER_STEP_SIZE;
 
     bencher.iter(|| {
         insert_packet_batches(buffer_capacity, batch_count, packet_per_batch_count, false);
@@ -148,9 +151,9 @@ fn bench_unprocessed_packet_batches_within_limit(bencher: &mut Bencher) {
 #[bench]
 #[ignore]
 fn bench_unprocessed_packet_batches_beyond_limit(bencher: &mut Bencher) {
-    let buffer_capacity = 1_000 * 128;
+    let buffer_capacity = 1_000 * UNPROCESSED_BUFFER_STEP_SIZE;
     let batch_count = 1_100;
-    let packet_per_batch_count = 128;
+    let packet_per_batch_count = UNPROCESSED_BUFFER_STEP_SIZE;
 
     // this is the worst scenario testing: all batches are uniformly populated with packets from
     // priority 100..228, so in order to drop a batch, algo will have to drop all packets that has
@@ -167,9 +170,9 @@ fn bench_unprocessed_packet_batches_beyond_limit(bencher: &mut Bencher) {
 #[bench]
 #[ignore]
 fn bench_unprocessed_packet_batches_randomized_within_limit(bencher: &mut Bencher) {
-    let buffer_capacity = 1_000 * 128;
+    let buffer_capacity = 1_000 * UNPROCESSED_BUFFER_STEP_SIZE;
     let batch_count = 1_000;
-    let packet_per_batch_count = 128;
+    let packet_per_batch_count = UNPROCESSED_BUFFER_STEP_SIZE;
 
     bencher.iter(|| {
         insert_packet_batches(buffer_capacity, batch_count, packet_per_batch_count, true);
@@ -181,9 +184,9 @@ fn bench_unprocessed_packet_batches_randomized_within_limit(bencher: &mut Benche
 #[bench]
 #[ignore]
 fn bench_unprocessed_packet_batches_randomized_beyond_limit(bencher: &mut Bencher) {
-    let buffer_capacity = 1_000 * 128;
+    let buffer_capacity = 1_000 * UNPROCESSED_BUFFER_STEP_SIZE;
     let batch_count = 1_100;
-    let packet_per_batch_count = 128;
+    let packet_per_batch_count = UNPROCESSED_BUFFER_STEP_SIZE;
 
     bencher.iter(|| {
         insert_packet_batches(buffer_capacity, batch_count, packet_per_batch_count, true);
@@ -198,7 +201,6 @@ fn buffer_iter_desc_and_forward(
 ) {
     solana_logger::setup();
     let mut unprocessed_packet_batches = UnprocessedPacketBatches::with_capacity(buffer_max_size);
-
     let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(10_000);
     let bank = Bank::new_for_tests(&genesis_config);
     let bank_forks = BankForks::new(bank);
@@ -226,13 +228,15 @@ fn buffer_iter_desc_and_forward(
 
     // forward whole buffer
     {
+        let mut transaction_storage = UnprocessedTransactionStorage::new_transaction_storage(
+            unprocessed_packet_batches,
+            ThreadType::Transactions,
+        );
         let mut forward_packet_batches_by_accounts =
             ForwardPacketBatchesByAccounts::new_with_default_batch_limits();
-        let _ = BankingStage::filter_and_forward_with_account_limits(
+        let _ = transaction_storage.filter_forwardable_packets_and_add_batches(
-            &current_bank,
+            current_bank,
-            &mut unprocessed_packet_batches,
             &mut forward_packet_batches_by_accounts,
-            128usize,
         );
     }
 }

core/src/fetch_stage.rs

@@ -1,16 +1,13 @@
 //! The `fetch_stage` batches input from a UDP socket and sends it to a channel.
 
 use {
-    crate::{
+    crate::result::{Error, Result},
-        banking_stage::HOLD_TRANSACTIONS_SLOT_OFFSET,
-        result::{Error, Result},
-    },
     crossbeam_channel::{unbounded, RecvTimeoutError},
     solana_metrics::{inc_new_counter_debug, inc_new_counter_info},
     solana_perf::{packet::PacketBatchRecycler, recycler::Recycler},
     solana_poh::poh_recorder::PohRecorder,
     solana_sdk::{
-        clock::DEFAULT_TICKS_PER_SLOT,
+        clock::{DEFAULT_TICKS_PER_SLOT, HOLD_TRANSACTIONS_SLOT_OFFSET},
         packet::{Packet, PacketFlags},
     },
     solana_streamer::streamer::{

core/src/forward_packet_batches_by_accounts.rs

@@ -182,7 +182,7 @@ impl ForwardPacketBatchesByAccounts {
 mod tests {
     use {
         super::*,
-        crate::unprocessed_packet_batches::{self, DeserializedPacket},
+        crate::unprocessed_packet_batches::DeserializedPacket,
         solana_runtime::transaction_priority_details::TransactionPriorityDetails,
         solana_sdk::{
             feature_set::FeatureSet, hash::Hash, signature::Keypair, system_transaction,
@@ -352,13 +352,14 @@ mod tests {
         // assert it is added, and buffer still accepts more packets
         {
             let packet = build_deserialized_packet_for_test(10, &hot_account, requested_cu);
-            let tx = unprocessed_packet_batches::transaction_from_deserialized_packet(
+            let tx = packet
-                packet.immutable_section(),
+                .immutable_section()
-                &Arc::new(FeatureSet::default()),
+                .build_sanitized_transaction(
-                false, //votes_only,
+                    &Arc::new(FeatureSet::default()),
-                SimpleAddressLoader::Disabled,
+                    false, //votes_only,
-            )
+                    SimpleAddressLoader::Disabled,
-            .unwrap();
+                )
+                .unwrap();
             assert!(forward_packet_batches_by_accounts
                 .try_add_packet(&tx, packet.immutable_section().clone()));
 
@@ -372,13 +373,14 @@ mod tests {
         {
             let packet =
                 build_deserialized_packet_for_test(100, &hot_account, 1 /*requested_cu*/);
-            let tx = unprocessed_packet_batches::transaction_from_deserialized_packet(
+            let tx = packet
-                packet.immutable_section(),
+                .immutable_section()
-                &Arc::new(FeatureSet::default()),
+                .build_sanitized_transaction(
-                false, //votes_only,
+                    &Arc::new(FeatureSet::default()),
-                SimpleAddressLoader::Disabled,
+                    false, //votes_only,
-            )
+                    SimpleAddressLoader::Disabled,
-            .unwrap();
+                )
+                .unwrap();
             assert!(!forward_packet_batches_by_accounts
                 .try_add_packet(&tx, packet.immutable_section().clone()));
 
@@ -392,13 +394,14 @@ mod tests {
         {
             let packet =
                 build_deserialized_packet_for_test(100, &other_account, 1 /*requested_cu*/);
-            let tx = unprocessed_packet_batches::transaction_from_deserialized_packet(
+            let tx = packet
-                packet.immutable_section(),
+                .immutable_section()
-                &Arc::new(FeatureSet::default()),
+                .build_sanitized_transaction(
-                false, //votes_only,
+                    &Arc::new(FeatureSet::default()),
-                SimpleAddressLoader::Disabled,
+                    false, //votes_only,
-            )
+                    SimpleAddressLoader::Disabled,
-            .unwrap();
+                )
+                .unwrap();
             assert!(!forward_packet_batches_by_accounts
                 .try_add_packet(&tx, packet.immutable_section().clone()));
 

core/src/immutable_deserialized_packet.rs

@@ -4,14 +4,18 @@ use {
         GetTransactionPriorityDetails, TransactionPriorityDetails,
     },
     solana_sdk::{
+        feature_set,
         hash::Hash,
         message::Message,
         sanitize::SanitizeError,
         short_vec::decode_shortu16_len,
         signature::Signature,
-        transaction::{SanitizedVersionedTransaction, VersionedTransaction},
+        transaction::{
+            AddressLoader, SanitizedTransaction, SanitizedVersionedTransaction,
+            VersionedTransaction,
+        },
     },
-    std::{cmp::Ordering, mem::size_of},
+    std::{cmp::Ordering, mem::size_of, sync::Arc},
     thiserror::Error,
 };
 
@@ -94,6 +98,28 @@ impl ImmutableDeserializedPacket {
     pub fn compute_unit_limit(&self) -> u64 {
         self.priority_details.compute_unit_limit
     }
+
+    // This function deserializes packets into transactions, computes the blake3 hash of transaction
+    // messages, and verifies secp256k1 instructions.
+    pub fn build_sanitized_transaction(
+        &self,
+        feature_set: &Arc<feature_set::FeatureSet>,
+        votes_only: bool,
+        address_loader: impl AddressLoader,
+    ) -> Option<SanitizedTransaction> {
+        if votes_only && !self.is_simple_vote() {
+            return None;
+        }
+        let tx = SanitizedTransaction::try_new(
+            self.transaction().clone(),
+            *self.message_hash(),
+            self.is_simple_vote(),
+            address_loader,
+        )
+        .ok()?;
+        tx.verify_precompiles(feature_set).ok()?;
+        Some(tx)
+    }
 }
 
 impl PartialOrd for ImmutableDeserializedPacket {

core/src/latest_unprocessed_votes.rs

@@ -1,13 +1,11 @@
-#![allow(dead_code)]
 use {
     crate::{
         forward_packet_batches_by_accounts::ForwardPacketBatchesByAccounts,
         immutable_deserialized_packet::{DeserializedPacketError, ImmutableDeserializedPacket},
-        unprocessed_packet_batches,
     },
     itertools::Itertools,
     rand::{thread_rng, Rng},
-    solana_perf::packet::{Packet, PacketBatch},
+    solana_perf::packet::Packet,
     solana_runtime::bank::Bank,
     solana_sdk::{clock::Slot, program_utils::limited_deserialize, pubkey::Pubkey},
     solana_vote_program::vote_instruction::VoteInstruction,
@@ -56,7 +54,9 @@ impl LatestValidatorVotePacket {
 
         match limited_deserialize::<VoteInstruction>(&instruction.data) {
             Ok(VoteInstruction::UpdateVoteState(vote_state_update))
-            | Ok(VoteInstruction::UpdateVoteStateSwitch(vote_state_update, _)) => {
+            | Ok(VoteInstruction::UpdateVoteStateSwitch(vote_state_update, _))
+            | Ok(VoteInstruction::CompactUpdateVoteState(vote_state_update))
+            | Ok(VoteInstruction::CompactUpdateVoteStateSwitch(vote_state_update, _)) => {
                 let &pubkey = message
                     .message
                     .static_account_keys()
@@ -102,16 +102,6 @@ impl LatestValidatorVotePacket {
     }
 }
 
-pub fn deserialize_packets<'a>(
-    packet_batch: &'a PacketBatch,
-    packet_indexes: &'a [usize],
-    vote_source: VoteSource,
-) -> impl Iterator<Item = LatestValidatorVotePacket> + 'a {
-    packet_indexes.iter().filter_map(move |packet_index| {
-        LatestValidatorVotePacket::new(packet_batch[*packet_index].clone(), vote_source).ok()
-    })
-}
-
 // TODO: replace this with rand::seq::index::sample_weighted once we can update rand to 0.8+
 // This requires updating dependencies of ed25519-dalek as rand_core is not compatible cross
 // version https://github.com/dalek-cryptography/ed25519-dalek/pull/214
@@ -135,7 +125,8 @@ pub(crate) fn weighted_random_order_by_stake<'a>(
     pubkey_with_weight.into_iter().map(|(_, pubkey)| pubkey)
 }
 
-pub struct VoteBatchInsertionMetrics {
+#[derive(Default, Debug)]
+pub(crate) struct VoteBatchInsertionMetrics {
     pub(crate) num_dropped_gossip: usize,
     pub(crate) num_dropped_tpu: usize,
 }
@@ -164,7 +155,7 @@ impl LatestUnprocessedVotes {
         self.len() == 0
     }
 
-    pub fn insert_batch(
+    pub(crate) fn insert_batch(
         &self,
         votes: impl Iterator<Item = LatestValidatorVotePacket>,
     ) -> VoteBatchInsertionMetrics {
@@ -259,9 +250,8 @@ impl LatestUnprocessedVotes {
                     let mut vote = lock.write().unwrap();
                     if !vote.is_vote_taken() && !vote.is_forwarded() {
                         let deserialized_vote_packet = vote.vote.as_ref().unwrap().clone();
-                        if let Some(sanitized_vote_transaction) =
+                        if let Some(sanitized_vote_transaction) = deserialized_vote_packet
-                            unprocessed_packet_batches::transaction_from_deserialized_packet(
+                            .build_sanitized_transaction(
-                                &deserialized_vote_packet,
                                 &bank.feature_set,
                                 bank.vote_only_bank(),
                                 bank.as_ref(),
@@ -329,7 +319,7 @@ mod tests {
         super::*,
         itertools::Itertools,
         rand::{thread_rng, Rng},
-        solana_perf::packet::{Packet, PacketFlags},
+        solana_perf::packet::{Packet, PacketBatch, PacketFlags},
         solana_runtime::{
             bank::Bank,
             genesis_utils::{self, ValidatorVoteKeypairs},
@@ -361,6 +351,16 @@ mod tests {
         LatestValidatorVotePacket::new(packet, vote_source).unwrap()
     }
 
+    fn deserialize_packets<'a>(
+        packet_batch: &'a PacketBatch,
+        packet_indexes: &'a [usize],
+        vote_source: VoteSource,
+    ) -> impl Iterator<Item = LatestValidatorVotePacket> + 'a {
+        packet_indexes.iter().filter_map(move |packet_index| {
+            LatestValidatorVotePacket::new(packet_batch[*packet_index].clone(), vote_source).ok()
+        })
+    }
+
     #[test]
     fn test_deserialize_vote_packets() {
         let keypairs = ValidatorVoteKeypairs::new_rand();

core/src/leader_slot_banking_stage_metrics.rs

@@ -1,5 +1,8 @@
 use {
-    crate::leader_slot_banking_stage_timing_metrics::*,
+    crate::{
+        leader_slot_banking_stage_timing_metrics::*,
+        unprocessed_transaction_storage::InsertPacketBatchSummary,
+    },
     solana_poh::poh_recorder::BankStart,
     solana_runtime::transaction_error_metrics::*,
     solana_sdk::{clock::Slot, saturating_add_assign},
@@ -270,6 +273,8 @@ pub(crate) struct LeaderSlotMetrics {
 
     transaction_error_metrics: TransactionErrorMetrics,
 
+    vote_packet_count_metrics: VotePacketCountMetrics,
+
     timing_metrics: LeaderSlotTimingMetrics,
 
     // Used by tests to check if the `self.report()` method was called
@@ -283,6 +288,7 @@ impl LeaderSlotMetrics {
             slot,
             packet_count_metrics: LeaderSlotPacketCountMetrics::new(),
             transaction_error_metrics: TransactionErrorMetrics::new(),
+            vote_packet_count_metrics: VotePacketCountMetrics::new(),
             timing_metrics: LeaderSlotTimingMetrics::new(bank_creation_time),
             is_reported: false,
         }
@@ -294,6 +300,7 @@ impl LeaderSlotMetrics {
         self.timing_metrics.report(self.id, self.slot);
         self.transaction_error_metrics.report(self.id, self.slot);
         self.packet_count_metrics.report(self.id, self.slot);
+        self.vote_packet_count_metrics.report(self.id, self.slot);
     }
 
     /// Returns `Some(self.slot)` if the metrics have been reported, otherwise returns None
@@ -310,6 +317,33 @@ impl LeaderSlotMetrics {
     }
 }
 
+// Metrics describing vote tx packets that were processed in the tpu vote thread as well as
+// extraneous votes that were filtered out
+#[derive(Debug, Default)]
+pub(crate) struct VotePacketCountMetrics {
+    // How many votes ingested from gossip were dropped
+    dropped_gossip_votes: u64,
+
+    // How many votes ingested from tpu were dropped
+    dropped_tpu_votes: u64,
+}
+
+impl VotePacketCountMetrics {
+    fn new() -> Self {
+        Self { ..Self::default() }
+    }
+
+    fn report(&self, id: u32, slot: Slot) {
+        datapoint_info!(
+            "banking_stage-vote_packet_counts",
+            ("id", id, i64),
+            ("slot", slot, i64),
+            ("dropped_gossip_votes", self.dropped_gossip_votes, i64),
+            ("dropped_tpu_votes", self.dropped_tpu_votes, i64)
+        );
+    }
+}
+
 #[derive(Debug)]
 pub(crate) enum MetricsTrackerAction {
     Noop,
@@ -498,6 +532,21 @@ impl LeaderSlotMetricsTracker {
         }
     }
 
+    pub(crate) fn accumulate_insert_packet_batches_summary(
+        &mut self,
+        insert_packet_batches_summary: &InsertPacketBatchSummary,
+    ) {
+        self.increment_exceeded_buffer_limit_dropped_packets_count(
+            insert_packet_batches_summary.total_dropped_packets() as u64,
+        );
+        self.increment_dropped_gossip_vote_count(
+            insert_packet_batches_summary.dropped_gossip_packets() as u64,
+        );
+        self.increment_dropped_tpu_vote_count(
+            insert_packet_batches_summary.dropped_tpu_packets() as u64
+        );
+    }
+
     pub(crate) fn accumulate_transaction_errors(
         &mut self,
         error_metrics: &TransactionErrorMetrics,
@@ -780,6 +829,28 @@ impl LeaderSlotMetricsTracker {
             );
         }
     }
+
+    pub(crate) fn increment_dropped_gossip_vote_count(&mut self, count: u64) {
+        if let Some(leader_slot_metrics) = &mut self.leader_slot_metrics {
+            saturating_add_assign!(
+                leader_slot_metrics
+                    .vote_packet_count_metrics
+                    .dropped_gossip_votes,
+                count
+            );
+        }
+    }
+
+    pub(crate) fn increment_dropped_tpu_vote_count(&mut self, count: u64) {
+        if let Some(leader_slot_metrics) = &mut self.leader_slot_metrics {
+            saturating_add_assign!(
+                leader_slot_metrics
+                    .vote_packet_count_metrics
+                    .dropped_tpu_votes,
+                count
+            );
+        }
+    }
 }
 
 #[cfg(test)]

core/src/unprocessed_packet_batches.rs

@@ -3,11 +3,7 @@ use {
     min_max_heap::MinMaxHeap,
     solana_perf::packet::{Packet, PacketBatch},
     solana_runtime::transaction_priority_details::TransactionPriorityDetails,
-    solana_sdk::{
+    solana_sdk::{hash::Hash, transaction::Transaction},
-        feature_set,
-        hash::Hash,
-        transaction::{AddressLoader, SanitizedTransaction, Transaction},
-    },
     std::{
         cmp::Ordering,
         collections::{hash_map::Entry, HashMap},
@@ -74,6 +70,12 @@ impl Ord for DeserializedPacket {
     }
 }
 
+#[derive(Debug)]
+pub struct PacketBatchInsertionMetrics {
+    pub(crate) num_dropped_packets: usize,
+    pub(crate) num_dropped_tracer_packets: usize,
+}
+
 /// Currently each banking_stage thread has a `UnprocessedPacketBatches` buffer to store
 /// PacketBatch's received from sigverify. Banking thread continuously scans the buffer
 /// to pick proper packets to add to the block.
@@ -115,7 +117,7 @@ impl UnprocessedPacketBatches {
     pub fn insert_batch(
         &mut self,
         deserialized_packets: impl Iterator<Item = DeserializedPacket>,
-    ) -> (usize, usize) {
+    ) -> PacketBatchInsertionMetrics {
         let mut num_dropped_packets = 0;
         let mut num_dropped_tracer_packets = 0;
         for deserialized_packet in deserialized_packets {
@@ -131,7 +133,10 @@ impl UnprocessedPacketBatches {
                 }
             }
         }
-        (num_dropped_packets, num_dropped_tracer_packets)
+        PacketBatchInsertionMetrics {
+            num_dropped_packets,
+            num_dropped_tracer_packets,
+        }
     }
 
     /// Pushes a new `deserialized_packet` into the unprocessed packet batches if it does not already
@@ -283,7 +288,7 @@ impl UnprocessedPacketBatches {
     }
 
     pub fn mark_accepted_packets_as_forwarded(
-        buffered_packet_batches: &mut UnprocessedPacketBatches,
+        &mut self,
         packets_to_process: &[Arc<ImmutableDeserializedPacket>],
         accepted_packet_indexes: &[usize],
     ) {
@@ -291,7 +296,7 @@ impl UnprocessedPacketBatches {
             .iter()
             .for_each(|accepted_packet_index| {
                 let accepted_packet = packets_to_process[*accepted_packet_index].clone();
-                if let Some(deserialized_packet) = buffered_packet_batches
+                if let Some(deserialized_packet) = self
                     .message_hash_to_transaction
                     .get_mut(accepted_packet.message_hash())
                 {
@@ -322,30 +327,6 @@ pub fn transactions_to_deserialized_packets(
         .collect()
 }
 
-// This function deserializes packets into transactions, computes the blake3 hash of transaction
-// messages, and verifies secp256k1 instructions. A list of sanitized transactions are returned
-// with their packet indexes.
-#[allow(clippy::needless_collect)]
-pub fn transaction_from_deserialized_packet(
-    deserialized_packet: &ImmutableDeserializedPacket,
-    feature_set: &Arc<feature_set::FeatureSet>,
-    votes_only: bool,
-    address_loader: impl AddressLoader,
-) -> Option<SanitizedTransaction> {
-    if votes_only && !deserialized_packet.is_simple_vote() {
-        return None;
-    }
-    let tx = SanitizedTransaction::try_new(
-        deserialized_packet.transaction().clone(),
-        *deserialized_packet.message_hash(),
-        deserialized_packet.is_simple_vote(),
-        address_loader,
-    )
-    .ok()?;
-    tx.verify_precompiles(feature_set).ok()?;
-    Some(tx)
-}
-
 #[cfg(test)]
 mod tests {
     use {
@@ -357,6 +338,7 @@ mod tests {
             transaction::{SimpleAddressLoader, Transaction},
         },
         solana_vote_program::vote_transaction,
+        std::sync::Arc,
     };
 
     fn simple_deserialized_packet() -> DeserializedPacket {
@@ -529,8 +511,7 @@ mod tests {
 
             let mut votes_only = false;
             let txs = packet_vector.iter().filter_map(|tx| {
-                transaction_from_deserialized_packet(
+                tx.immutable_section().build_sanitized_transaction(
-                    tx.immutable_section(),
                     &Arc::new(FeatureSet::default()),
                     votes_only,
                     SimpleAddressLoader::Disabled,
@@ -540,8 +521,7 @@ mod tests {
 
             votes_only = true;
             let txs = packet_vector.iter().filter_map(|tx| {
-                transaction_from_deserialized_packet(
+                tx.immutable_section().build_sanitized_transaction(
-                    tx.immutable_section(),
                     &Arc::new(FeatureSet::default()),
                     votes_only,
                     SimpleAddressLoader::Disabled,
@@ -560,8 +540,7 @@ mod tests {
 
             let mut votes_only = false;
             let txs = packet_vector.iter().filter_map(|tx| {
-                transaction_from_deserialized_packet(
+                tx.immutable_section().build_sanitized_transaction(
-                    tx.immutable_section(),
                     &Arc::new(FeatureSet::default()),
                     votes_only,
                     SimpleAddressLoader::Disabled,
@@ -571,8 +550,7 @@ mod tests {
 
             votes_only = true;
             let txs = packet_vector.iter().filter_map(|tx| {
-                transaction_from_deserialized_packet(
+                tx.immutable_section().build_sanitized_transaction(
-                    tx.immutable_section(),
                     &Arc::new(FeatureSet::default()),
                     votes_only,
                     SimpleAddressLoader::Disabled,
@@ -591,8 +569,7 @@ mod tests {
 
             let mut votes_only = false;
             let txs = packet_vector.iter().filter_map(|tx| {
-                transaction_from_deserialized_packet(
+                tx.immutable_section().build_sanitized_transaction(
-                    tx.immutable_section(),
                     &Arc::new(FeatureSet::default()),
                     votes_only,
                     SimpleAddressLoader::Disabled,
@@ -602,8 +579,7 @@ mod tests {
 
             votes_only = true;
             let txs = packet_vector.iter().filter_map(|tx| {
-                transaction_from_deserialized_packet(
+                tx.immutable_section().build_sanitized_transaction(
-                    tx.immutable_section(),
                     &Arc::new(FeatureSet::default()),
                     votes_only,
                     SimpleAddressLoader::Disabled,

core/src/unprocessed_transaction_storage.rs

@@ -1,22 +1,28 @@
-#![allow(dead_code)]
 use {
     crate::{
-        banking_stage::{self, BankingStage, FilterForwardingResults, ForwardOption},
+        banking_stage::{FilterForwardingResults, ForwardOption},
         forward_packet_batches_by_accounts::ForwardPacketBatchesByAccounts,
         immutable_deserialized_packet::ImmutableDeserializedPacket,
         latest_unprocessed_votes::{
-            self, LatestUnprocessedVotes, LatestValidatorVotePacket, VoteBatchInsertionMetrics,
+            LatestUnprocessedVotes, LatestValidatorVotePacket, VoteBatchInsertionMetrics,
             VoteSource,
         },
-        unprocessed_packet_batches::{self, DeserializedPacket, UnprocessedPacketBatches},
+        unprocessed_packet_batches::{
+            DeserializedPacket, PacketBatchInsertionMetrics, UnprocessedPacketBatches,
+        },
     },
     itertools::Itertools,
     min_max_heap::MinMaxHeap,
-    solana_perf::packet::PacketBatch,
+    solana_measure::measure,
     solana_runtime::bank::Bank,
+    solana_sdk::{
+        clock::FORWARD_TRANSACTIONS_TO_LEADER_AT_SLOT_OFFSET, saturating_add_assign,
+        transaction::SanitizedTransaction,
+    },
     std::sync::Arc,
 };
 
+pub const UNPROCESSED_BUFFER_STEP_SIZE: usize = 128;
 const MAX_STAKED_VALIDATORS: usize = 10_000;
 
 #[derive(Debug)]
@@ -43,12 +49,54 @@ pub enum ThreadType {
     Transactions,
 }
 
-#[derive(Debug, Default)]
+#[derive(Debug)]
-pub struct InsertPacketBatchesSummary {
+pub(crate) enum InsertPacketBatchSummary {
-    pub(crate) num_dropped_packets: usize,
+    VoteBatchInsertionMetrics(VoteBatchInsertionMetrics),
-    pub(crate) num_dropped_gossip_vote_packets: usize,
+    PacketBatchInsertionMetrics(PacketBatchInsertionMetrics),
-    pub(crate) num_dropped_tpu_vote_packets: usize,
+}
-    pub(crate) num_dropped_tracer_packets: usize,
+
+impl InsertPacketBatchSummary {
+    pub fn total_dropped_packets(&self) -> usize {
+        match self {
+            Self::VoteBatchInsertionMetrics(metrics) => {
+                metrics.num_dropped_gossip + metrics.num_dropped_tpu
+            }
+            Self::PacketBatchInsertionMetrics(metrics) => metrics.num_dropped_packets,
+        }
+    }
+
+    pub fn dropped_gossip_packets(&self) -> usize {
+        match self {
+            Self::VoteBatchInsertionMetrics(metrics) => metrics.num_dropped_gossip,
+            _ => 0,
+        }
+    }
+
+    pub fn dropped_tpu_packets(&self) -> usize {
+        match self {
+            Self::VoteBatchInsertionMetrics(metrics) => metrics.num_dropped_tpu,
+            _ => 0,
+        }
+    }
+
+    pub fn dropped_tracer_packets(&self) -> usize {
+        match self {
+            Self::PacketBatchInsertionMetrics(metrics) => metrics.num_dropped_tracer_packets,
+            _ => 0,
+        }
+    }
+}
+
+impl From<VoteBatchInsertionMetrics> for InsertPacketBatchSummary {
+    fn from(metrics: VoteBatchInsertionMetrics) -> Self {
+        Self::VoteBatchInsertionMetrics(metrics)
+    }
+}
+
+impl From<PacketBatchInsertionMetrics> for InsertPacketBatchSummary {
+    fn from(metrics: PacketBatchInsertionMetrics) -> Self {
+        Self::PacketBatchInsertionMetrics(metrics)
+    }
 }
 
 fn filter_processed_packets<'a, F>(
@@ -145,33 +193,17 @@ impl UnprocessedTransactionStorage {
         }
     }
 
-    pub fn deserialize_and_insert_batch(
+    pub(crate) fn insert_batch(
         &mut self,
-        packet_batch: &PacketBatch,
+        deserialized_packets: Vec<ImmutableDeserializedPacket>,
-        packet_indexes: &[usize],
+    ) -> InsertPacketBatchSummary {
-    ) -> InsertPacketBatchesSummary {
         match self {
             Self::VoteStorage(vote_storage) => {
-                let VoteBatchInsertionMetrics {
+                InsertPacketBatchSummary::from(vote_storage.insert_batch(deserialized_packets))
-                    num_dropped_gossip,
-                    num_dropped_tpu,
-                } = vote_storage.deserialize_and_insert_batch(packet_batch, packet_indexes);
-                InsertPacketBatchesSummary {
-                    num_dropped_packets: num_dropped_gossip + num_dropped_tpu,
-                    num_dropped_gossip_vote_packets: num_dropped_gossip,
-                    num_dropped_tpu_vote_packets: num_dropped_tpu,
-                    ..InsertPacketBatchesSummary::default()
-                }
-            }
-            Self::LocalTransactionStorage(transaction_storage) => {
-                let (num_dropped_packets, num_dropped_tracer_packets) =
-                    transaction_storage.deserialize_and_insert_batch(packet_batch, packet_indexes);
-                InsertPacketBatchesSummary {
-                    num_dropped_packets,
-                    num_dropped_tracer_packets,
-                    ..InsertPacketBatchesSummary::default()
-                }
             }
+            Self::LocalTransactionStorage(transaction_storage) => InsertPacketBatchSummary::from(
+                transaction_storage.insert_batch(deserialized_packets),
+            ),
         }
     }
 
@@ -241,17 +273,22 @@ impl VoteStorage {
         self.latest_unprocessed_votes.clear_forwarded_packets();
     }
 
-    fn deserialize_and_insert_batch(
+    fn insert_batch(
         &mut self,
-        packet_batch: &PacketBatch,
+        deserialized_packets: Vec<ImmutableDeserializedPacket>,
-        packet_indexes: &[usize],
     ) -> VoteBatchInsertionMetrics {
         self.latest_unprocessed_votes
-            .insert_batch(latest_unprocessed_votes::deserialize_packets(
+            .insert_batch(
-                packet_batch,
+                deserialized_packets
-                packet_indexes,
+                    .into_iter()
-                self.vote_source,
+                    .filter_map(|deserialized_packet| {
-            ))
+                        LatestValidatorVotePacket::new_from_immutable(
+                            Arc::new(deserialized_packet),
+                            self.vote_source,
+                        )
+                        .ok()
+                    }),
+            )
     }
 
     fn filter_forwardable_packets_and_add_batches(
@@ -345,13 +382,14 @@ impl ThreadLocalUnprocessedPackets {
         self.unprocessed_packet_batches.clear();
     }
 
-    fn deserialize_and_insert_batch(
+    fn insert_batch(
         &mut self,
-        packet_batch: &PacketBatch,
+        deserialized_packets: Vec<ImmutableDeserializedPacket>,
-        packet_indexes: &[usize],
+    ) -> PacketBatchInsertionMetrics {
-    ) -> (usize, usize) {
         self.unprocessed_packet_batches.insert_batch(
-            unprocessed_packet_batches::deserialize_packets(packet_batch, packet_indexes),
+            deserialized_packets
+                .into_iter()
+                .map(DeserializedPacket::from_immutable_section),
         )
     }
 
@@ -360,11 +398,298 @@ impl ThreadLocalUnprocessedPackets {
         bank: Arc<Bank>,
         forward_packet_batches_by_accounts: &mut ForwardPacketBatchesByAccounts,
     ) -> FilterForwardingResults {
-        BankingStage::filter_and_forward_with_account_limits(
+        self.filter_and_forward_with_account_limits(
-            &bank,
+            bank,
-            &mut self.unprocessed_packet_batches,
             forward_packet_batches_by_accounts,
-            banking_stage::UNPROCESSED_BUFFER_STEP_SIZE,
+            UNPROCESSED_BUFFER_STEP_SIZE,
+        )
+    }
+
+    /// Filter out packets that fail to sanitize, or are no longer valid (could be
+    /// too old, a duplicate of something already processed). Doing this in batches to avoid
+    /// checking bank's blockhash and status cache per transaction which could be bad for performance.
+    /// Added valid and sanitized packets to forwarding queue.
+    fn filter_and_forward_with_account_limits(
+        &mut self,
+        bank: Arc<Bank>,
+        forward_buffer: &mut ForwardPacketBatchesByAccounts,
+        batch_size: usize,
+    ) -> FilterForwardingResults {
+        let mut total_forwardable_tracer_packets: usize = 0;
+        let mut total_tracer_packets_in_buffer: usize = 0;
+        let mut total_forwardable_packets: usize = 0;
+        let mut total_packet_conversion_us: u64 = 0;
+        let mut total_filter_packets_us: u64 = 0;
+        let mut dropped_tx_before_forwarding_count: usize = 0;
+
+        let mut original_priority_queue = self.swap_priority_queue();
+
+        // indicates if `forward_buffer` still accept more packets, see details at
+        // `ForwardPacketBatchesByAccounts.rs`.
+        let mut accepting_packets = true;
+        // batch iterate through self.unprocessed_packet_batches in desc priority order
+        let retained_priority_queue: MinMaxHeap<Arc<ImmutableDeserializedPacket>> =
+            original_priority_queue
+                .drain_desc()
+                .chunks(batch_size)
+                .into_iter()
+                .flat_map(|packets_to_process| {
+                    let packets_to_process = packets_to_process.into_iter().collect_vec();
+
+                    // Vec<bool> of same size of `packets_to_process`, each indicates
+                    // corresponding packet is tracer packet.
+                    let tracer_packet_indexes = packets_to_process
+                        .iter()
+                        .map(|deserialized_packet| {
+                            deserialized_packet
+                                .original_packet()
+                                .meta
+                                .is_tracer_packet()
+                        })
+                        .collect::<Vec<_>>();
+                    saturating_add_assign!(
+                        total_tracer_packets_in_buffer,
+                        tracer_packet_indexes
+                            .iter()
+                            .filter(|is_tracer| **is_tracer)
+                            .count()
+                    );
+
+                    if accepting_packets {
+                        let (
+                            (sanitized_transactions, transaction_to_packet_indexes),
+                            packet_conversion_time,
+                        ): ((Vec<SanitizedTransaction>, Vec<usize>), _) = measure!(
+                            self.sanitize_unforwarded_packets(&packets_to_process, &bank,),
+                            "sanitize_packet",
+                        );
+                        saturating_add_assign!(
+                            total_packet_conversion_us,
+                            packet_conversion_time.as_us()
+                        );
+
+                        let (forwardable_transaction_indexes, filter_packets_time) = measure!(
+                            Self::filter_invalid_transactions(&sanitized_transactions, &bank,),
+                            "filter_packets",
+                        );
+                        saturating_add_assign!(
+                            total_filter_packets_us,
+                            filter_packets_time.as_us()
+                        );
+
+                        for forwardable_transaction_index in &forwardable_transaction_indexes {
+                            saturating_add_assign!(total_forwardable_packets, 1);
+                            let forwardable_packet_index =
+                                transaction_to_packet_indexes[*forwardable_transaction_index];
+                            if tracer_packet_indexes[forwardable_packet_index] {
+                                saturating_add_assign!(total_forwardable_tracer_packets, 1);
+                            }
+                        }
+
+                        let accepted_packet_indexes = Self::add_filtered_packets_to_forward_buffer(
+                            forward_buffer,
+                            &packets_to_process,
+                            &sanitized_transactions,
+                            &transaction_to_packet_indexes,
+                            &forwardable_transaction_indexes,
+                            &mut dropped_tx_before_forwarding_count,
+                        );
+                        accepting_packets =
+                            accepted_packet_indexes.len() == forwardable_transaction_indexes.len();
+
+                        self.unprocessed_packet_batches
+                            .mark_accepted_packets_as_forwarded(
+                                &packets_to_process,
+                                &accepted_packet_indexes,
+                            );
+
+                        self.collect_retained_packets(
+                            &packets_to_process,
+                            &Self::prepare_filtered_packet_indexes(
+                                &transaction_to_packet_indexes,
+                                &forwardable_transaction_indexes,
+                            ),
+                        )
+                    } else {
+                        // skip sanitizing and filtering if not longer able to add more packets for forwarding
+                        saturating_add_assign!(
+                            dropped_tx_before_forwarding_count,
+                            packets_to_process.len()
+                        );
+                        packets_to_process
+                    }
+                })
+                .collect();
+
+        // replace packet priority queue
+        self.unprocessed_packet_batches.packet_priority_queue = retained_priority_queue;
+
+        inc_new_counter_info!(
+            "banking_stage-dropped_tx_before_forwarding",
+            dropped_tx_before_forwarding_count
+        );
+
+        FilterForwardingResults {
+            total_forwardable_packets,
+            total_tracer_packets_in_buffer,
+            total_forwardable_tracer_packets,
+            total_packet_conversion_us,
+            total_filter_packets_us,
+        }
+    }
+
+    /// Take self.unprocessed_packet_batches's priority_queue out, leave empty MinMaxHeap in its place.
+    fn swap_priority_queue(&mut self) -> MinMaxHeap<Arc<ImmutableDeserializedPacket>> {
+        let capacity = self.unprocessed_packet_batches.capacity();
+        std::mem::replace(
+            &mut self.unprocessed_packet_batches.packet_priority_queue,
+            MinMaxHeap::with_capacity(capacity),
+        )
+    }
+
+    /// sanitize un-forwarded packet into SanitizedTransaction for validation and forwarding.
+    fn sanitize_unforwarded_packets(
+        &mut self,
+        packets_to_process: &[Arc<ImmutableDeserializedPacket>],
+        bank: &Arc<Bank>,
+    ) -> (Vec<SanitizedTransaction>, Vec<usize>) {
+        // Get ref of ImmutableDeserializedPacket
+        let deserialized_packets = packets_to_process.iter().map(|p| &**p);
+        let (transactions, transaction_to_packet_indexes): (Vec<SanitizedTransaction>, Vec<usize>) =
+            deserialized_packets
+                .enumerate()
+                .filter_map(|(packet_index, deserialized_packet)| {
+                    if !self
+                        .unprocessed_packet_batches
+                        .is_forwarded(deserialized_packet)
+                    {
+                        deserialized_packet
+                            .build_sanitized_transaction(
+                                &bank.feature_set,
+                                bank.vote_only_bank(),
+                                bank.as_ref(),
+                            )
+                            .map(|transaction| (transaction, packet_index))
+                    } else {
+                        None
+                    }
+                })
+                .unzip();
+
+        // report metrics
+        inc_new_counter_info!("banking_stage-packet_conversion", 1);
+        let unsanitized_packets_filtered_count =
+            packets_to_process.len().saturating_sub(transactions.len());
+        inc_new_counter_info!(
+            "banking_stage-dropped_tx_before_forwarding",
+            unsanitized_packets_filtered_count
+        );
+
+        (transactions, transaction_to_packet_indexes)
+    }
+
+    /// Checks sanitized transactions against bank, returns valid transaction indexes
+    fn filter_invalid_transactions(
+        transactions: &[SanitizedTransaction],
+        bank: &Arc<Bank>,
+    ) -> Vec<usize> {
+        let filter = vec![Ok(()); transactions.len()];
+        let results = bank.check_transactions_with_forwarding_delay(
+            transactions,
+            &filter,
+            FORWARD_TRANSACTIONS_TO_LEADER_AT_SLOT_OFFSET,
+        );
+        // report metrics
+        let filtered_out_transactions_count = transactions.len().saturating_sub(results.len());
+        inc_new_counter_info!(
+            "banking_stage-dropped_tx_before_forwarding",
+            filtered_out_transactions_count
+        );
+
+        results
+            .iter()
+            .enumerate()
+            .filter_map(
+                |(tx_index, (result, _))| if result.is_ok() { Some(tx_index) } else { None },
+            )
+            .collect_vec()
+    }
+
+    fn prepare_filtered_packet_indexes(
+        transaction_to_packet_indexes: &[usize],
+        retained_transaction_indexes: &[usize],
+    ) -> Vec<usize> {
+        retained_transaction_indexes
+            .iter()
+            .map(|tx_index| transaction_to_packet_indexes[*tx_index])
+            .collect_vec()
+    }
+
+    /// try to add filtered forwardable and valid packets to forward buffer;
+    /// returns vector of packet indexes that were accepted for forwarding.
+    fn add_filtered_packets_to_forward_buffer(
+        forward_buffer: &mut ForwardPacketBatchesByAccounts,
+        packets_to_process: &[Arc<ImmutableDeserializedPacket>],
+        transactions: &[SanitizedTransaction],
+        transaction_to_packet_indexes: &[usize],
+        forwardable_transaction_indexes: &[usize],
+        dropped_tx_before_forwarding_count: &mut usize,
+    ) -> Vec<usize> {
+        let mut added_packets_count: usize = 0;
+        let mut accepted_packet_indexes = Vec::with_capacity(transaction_to_packet_indexes.len());
+        for forwardable_transaction_index in forwardable_transaction_indexes {
+            let sanitized_transaction = &transactions[*forwardable_transaction_index];
+            let forwardable_packet_index =
+                transaction_to_packet_indexes[*forwardable_transaction_index];
+            let immutable_deserialized_packet =
+                packets_to_process[forwardable_packet_index].clone();
+            if !forward_buffer.try_add_packet(sanitized_transaction, immutable_deserialized_packet)
+            {
+                break;
+            }
+            accepted_packet_indexes.push(forwardable_packet_index);
+            saturating_add_assign!(added_packets_count, 1);
+        }
+
+        // count the packets not being forwarded in this batch
+        saturating_add_assign!(
+            *dropped_tx_before_forwarding_count,
+            forwardable_transaction_indexes.len() - added_packets_count
+        );
+
+        accepted_packet_indexes
+    }
+
+    fn collect_retained_packets(
+        &mut self,
+        packets_to_process: &[Arc<ImmutableDeserializedPacket>],
+        retained_packet_indexes: &[usize],
+    ) -> Vec<Arc<ImmutableDeserializedPacket>> {
+        self.remove_non_retained_packets(packets_to_process, retained_packet_indexes);
+        retained_packet_indexes
+            .iter()
+            .map(|i| packets_to_process[*i].clone())
+            .collect_vec()
+    }
+
+    /// remove packets from UnprocessedPacketBatches.message_hash_to_transaction after they have
+    /// been removed from UnprocessedPacketBatches.packet_priority_queue
+    fn remove_non_retained_packets(
+        &mut self,
+        packets_to_process: &[Arc<ImmutableDeserializedPacket>],
+        retained_packet_indexes: &[usize],
+    ) {
+        filter_processed_packets(
+            retained_packet_indexes
+                .iter()
+                .chain(std::iter::once(&packets_to_process.len())),
+            |start, end| {
+                for processed_packet in &packets_to_process[start..end] {
+                    self.unprocessed_packet_batches
+                        .message_hash_to_transaction
+                        .remove(processed_packet.message_hash());
+                }
+            },
         )
     }
 
@@ -372,13 +697,7 @@ impl ThreadLocalUnprocessedPackets {
     where
         F: FnMut(&Vec<Arc<ImmutableDeserializedPacket>>) -> Option<Vec<usize>>,
     {
-        let mut retryable_packets = {
+        let mut retryable_packets = self.swap_priority_queue();
-            let capacity = self.unprocessed_packet_batches.capacity();
-            std::mem::replace(
-                &mut self.unprocessed_packet_batches.packet_priority_queue,
-                MinMaxHeap::with_capacity(capacity),
-            )
-        };
         let retryable_packets: MinMaxHeap<Arc<ImmutableDeserializedPacket>> = retryable_packets
             .drain_desc()
             .chunks(batch_size)
@@ -388,25 +707,10 @@ impl ThreadLocalUnprocessedPackets {
                 if let Some(retryable_transaction_indexes) =
                     processing_function(&packets_to_process)
                 {
-                    // Remove the non-retryable packets, packets that were either:
+                    self.collect_retained_packets(
-                    // 1) Successfully processed
+                        &packets_to_process,
-                    // 2) Failed but not retryable
+                        &retryable_transaction_indexes,
-                    filter_processed_packets(
+                    )
-                        retryable_transaction_indexes
-                            .iter()
-                            .chain(std::iter::once(&packets_to_process.len())),
-                        |start, end| {
-                            for processed_packet in &packets_to_process[start..end] {
-                                self.unprocessed_packet_batches
-                                    .message_hash_to_transaction
-                                    .remove(processed_packet.message_hash());
-                            }
-                        },
-                    );
-                    retryable_transaction_indexes
-                        .iter()
-                        .map(|i| packets_to_process[*i].clone())
-                        .collect_vec()
                 } else {
                     packets_to_process
                 }
@@ -427,7 +731,21 @@ impl ThreadLocalUnprocessedPackets {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use {
+        super::*,
+        solana_ledger::genesis_utils::{create_genesis_config, GenesisConfigInfo},
+        solana_perf::packet::{Packet, PacketFlags},
+        solana_sdk::{
+            hash::Hash,
+            signature::{Keypair, Signer},
+            system_transaction,
+            transaction::Transaction,
+        },
+        solana_vote_program::{
+            vote_state::VoteStateUpdate, vote_transaction::new_vote_state_update_transaction,
+        },
+        std::error::Error,
+    };
 
     #[test]
     fn test_filter_processed_packets() {
@@ -479,4 +797,214 @@ mod tests {
         filter_processed_packets(retryable_indexes.iter(), f);
         assert_eq!(non_retryable_indexes, vec![(0, 1), (4, 5), (6, 8)]);
     }
+
+    #[test]
+    fn test_filter_and_forward_with_account_limits() {
+        solana_logger::setup();
+        let GenesisConfigInfo {
+            genesis_config,
+            mint_keypair,
+            ..
+        } = create_genesis_config(10);
+        let current_bank = Arc::new(Bank::new_for_tests(&genesis_config));
+
+        let simple_transactions: Vec<Transaction> = (0..256)
+            .map(|_id| {
+                // packets are deserialized upon receiving, failed packets will not be
+                // forwarded; Therefore we need to create real packets here.
+                let key1 = Keypair::new();
+                system_transaction::transfer(
+                    &mint_keypair,
+                    &key1.pubkey(),
+                    genesis_config.rent.minimum_balance(0),
+                    genesis_config.hash(),
+                )
+            })
+            .collect_vec();
+
+        let mut packets: Vec<DeserializedPacket> = simple_transactions
+            .iter()
+            .enumerate()
+            .map(|(packets_id, transaction)| {
+                let mut p = Packet::from_data(None, transaction).unwrap();
+                p.meta.port = packets_id as u16;
+                p.meta.set_tracer(true);
+                DeserializedPacket::new(p).unwrap()
+            })
+            .collect_vec();
+
+        // all packets are forwarded
+        {
+            let buffered_packet_batches: UnprocessedPacketBatches =
+                UnprocessedPacketBatches::from_iter(packets.clone().into_iter(), packets.len());
+            let mut transaction_storage = UnprocessedTransactionStorage::new_transaction_storage(
+                buffered_packet_batches,
+                ThreadType::Transactions,
+            );
+            let mut forward_packet_batches_by_accounts =
+                ForwardPacketBatchesByAccounts::new_with_default_batch_limits();
+
+            let FilterForwardingResults {
+                total_forwardable_packets,
+                total_tracer_packets_in_buffer,
+                total_forwardable_tracer_packets,
+                ..
+            } = transaction_storage.filter_forwardable_packets_and_add_batches(
+                current_bank.clone(),
+                &mut forward_packet_batches_by_accounts,
+            );
+            assert_eq!(total_forwardable_packets, 256);
+            assert_eq!(total_tracer_packets_in_buffer, 256);
+            assert_eq!(total_forwardable_tracer_packets, 256);
+
+            // packets in a batch are forwarded in arbitrary order; verify the ports match after
+            // sorting
+            let expected_ports: Vec<_> = (0..256).collect();
+            let mut forwarded_ports: Vec<_> = forward_packet_batches_by_accounts
+                .iter_batches()
+                .flat_map(|batch| {
+                    batch
+                        .get_forwardable_packets()
+                        .into_iter()
+                        .map(|p| p.meta.port)
+                })
+                .collect();
+            forwarded_ports.sort_unstable();
+            assert_eq!(expected_ports, forwarded_ports);
+        }
+
+        // some packets are forwarded
+        {
+            let num_already_forwarded = 16;
+            for packet in &mut packets[0..num_already_forwarded] {
+                packet.forwarded = true;
+            }
+            let buffered_packet_batches: UnprocessedPacketBatches =
+                UnprocessedPacketBatches::from_iter(packets.clone().into_iter(), packets.len());
+            let mut transaction_storage = UnprocessedTransactionStorage::new_transaction_storage(
+                buffered_packet_batches,
+                ThreadType::Transactions,
+            );
+            let mut forward_packet_batches_by_accounts =
+                ForwardPacketBatchesByAccounts::new_with_default_batch_limits();
+            let FilterForwardingResults {
+                total_forwardable_packets,
+                total_tracer_packets_in_buffer,
+                total_forwardable_tracer_packets,
+                ..
+            } = transaction_storage.filter_forwardable_packets_and_add_batches(
+                current_bank.clone(),
+                &mut forward_packet_batches_by_accounts,
+            );
+            assert_eq!(
+                total_forwardable_packets,
+                packets.len() - num_already_forwarded
+            );
+            assert_eq!(total_tracer_packets_in_buffer, packets.len());
+            assert_eq!(
+                total_forwardable_tracer_packets,
+                packets.len() - num_already_forwarded
+            );
+        }
+
+        // some packets are invalid (already processed)
+        {
+            let num_already_processed = 16;
+            for tx in &simple_transactions[0..num_already_processed] {
+                assert_eq!(current_bank.process_transaction(tx), Ok(()));
+            }
+            let buffered_packet_batches: UnprocessedPacketBatches =
+                UnprocessedPacketBatches::from_iter(packets.clone().into_iter(), packets.len());
+            let mut transaction_storage = UnprocessedTransactionStorage::new_transaction_storage(
+                buffered_packet_batches,
+                ThreadType::Transactions,
+            );
+            let mut forward_packet_batches_by_accounts =
+                ForwardPacketBatchesByAccounts::new_with_default_batch_limits();
+            let FilterForwardingResults {
+                total_forwardable_packets,
+                total_tracer_packets_in_buffer,
+                total_forwardable_tracer_packets,
+                ..
+            } = transaction_storage.filter_forwardable_packets_and_add_batches(
+                current_bank,
+                &mut forward_packet_batches_by_accounts,
+            );
+            assert_eq!(
+                total_forwardable_packets,
+                packets.len() - num_already_processed
+            );
+            assert_eq!(total_tracer_packets_in_buffer, packets.len());
+            assert_eq!(
+                total_forwardable_tracer_packets,
+                packets.len() - num_already_processed
+            );
+        }
+    }
+
+    #[test]
+    fn test_unprocessed_transaction_storage_insert() -> Result<(), Box<dyn Error>> {
+        let keypair = Keypair::new();
+        let vote_keypair = Keypair::new();
+        let pubkey = solana_sdk::pubkey::new_rand();
+
+        let small_transfer = Packet::from_data(
+            None,
+            system_transaction::transfer(&keypair, &pubkey, 1, Hash::new_unique()),
+        )?;
+        let mut vote = Packet::from_data(
+            None,
+            new_vote_state_update_transaction(
+                VoteStateUpdate::default(),
+                Hash::new_unique(),
+                &keypair,
+                &vote_keypair,
+                &vote_keypair,
+                None,
+            ),
+        )?;
+        vote.meta.flags.set(PacketFlags::SIMPLE_VOTE_TX, true);
+        let big_transfer = Packet::from_data(
+            None,
+            system_transaction::transfer(&keypair, &pubkey, 1000000, Hash::new_unique()),
+        )?;
+
+        for thread_type in [
+            ThreadType::Transactions,
+            ThreadType::Voting(VoteSource::Gossip),
+            ThreadType::Voting(VoteSource::Tpu),
+        ] {
+            let mut transaction_storage = UnprocessedTransactionStorage::new_transaction_storage(
+                UnprocessedPacketBatches::with_capacity(100),
+                thread_type,
+            );
+            transaction_storage.insert_batch(vec![
+                ImmutableDeserializedPacket::new(small_transfer.clone(), None)?,
+                ImmutableDeserializedPacket::new(vote.clone(), None)?,
+                ImmutableDeserializedPacket::new(big_transfer.clone(), None)?,
+            ]);
+            let deserialized_packets = transaction_storage
+                .iter()
+                .map(|packet| packet.immutable_section().original_packet().clone())
+                .collect_vec();
+            assert_eq!(3, deserialized_packets.len());
+            assert!(deserialized_packets.contains(&small_transfer));
+            assert!(deserialized_packets.contains(&vote));
+            assert!(deserialized_packets.contains(&big_transfer));
+        }
+
+        for vote_source in [VoteSource::Gossip, VoteSource::Tpu] {
+            let mut transaction_storage = UnprocessedTransactionStorage::new_vote_storage(
+                Arc::new(LatestUnprocessedVotes::new()),
+                vote_source,
+            );
+            transaction_storage.insert_batch(vec![
+                ImmutableDeserializedPacket::new(small_transfer.clone(), None)?,
+                ImmutableDeserializedPacket::new(vote.clone(), None)?,
+                ImmutableDeserializedPacket::new(big_transfer.clone(), None)?,
+            ]);
+            assert_eq!(1, transaction_storage.len());
+        }
+        Ok(())
+    }
 }

local-cluster/tests/local_cluster_slow_1.rs

@@ -30,6 +30,7 @@ use {
         hash::Hash,
         pubkey::Pubkey,
         signature::Signer,
+        vote::state::VoteStateUpdate,
     },
     solana_streamer::socket::SocketAddrSpace,
     solana_vote_program::{vote_state::MAX_LOCKOUT_HISTORY, vote_transaction},
@@ -541,22 +542,31 @@ fn test_duplicate_shreds_broadcast_leader() {
                             // root by this validator, but we're not concerned with lockout violations
                             // by this validator so it's fine.
                             let leader_blockstore = open_blockstore(&bad_leader_ledger_path);
-                            let mut vote_slots: Vec<Slot> = AncestorIterator::new_inclusive(
+                            let mut vote_slots: Vec<(Slot, u32)> = AncestorIterator::new_inclusive(
                                 latest_vote_slot,
                                 &leader_blockstore,
                             )
                             .take(MAX_LOCKOUT_HISTORY)
+                            .zip(1..)
                             .collect();
                             vote_slots.reverse();
-                            let vote_tx = vote_transaction::new_vote_transaction(
+                            let mut vote = VoteStateUpdate::from(vote_slots);
-                                vote_slots,
+                            let root = AncestorIterator::new_inclusive(
-                                vote_hash,
+                                latest_vote_slot,
-                                leader_vote_tx.message.recent_blockhash,
+                                &leader_blockstore,
-                                &node_keypair,
+                            )
-                                &vote_keypair,
+                            .nth(MAX_LOCKOUT_HISTORY);
-                                &vote_keypair,
+                            vote.root = root;
-                                None,
+                            vote.hash = vote_hash;
-                            );
+                            let vote_tx =
+                                vote_transaction::new_compact_vote_state_update_transaction(
+                                    vote,
+                                    leader_vote_tx.message.recent_blockhash,
+                                    &node_keypair,
+                                    &vote_keypair,
+                                    &vote_keypair,
+                                    None,
+                                );
                             gossip_vote_index += 1;
                             gossip_vote_index %= MAX_LOCKOUT_HISTORY;
                             cluster_info.push_vote_at_index(vote_tx, gossip_vote_index as u8)

programs/sbf/Cargo.lock

@@ -5088,6 +5088,7 @@ dependencies = [
  "solana-frozen-abi-macro 1.15.0",
  "solana-measure",
  "solana-metrics",
+ "solana-perf",
  "solana-program-runtime",
  "solana-rayon-threadlimit",
  "solana-sdk 1.15.0",

programs/vote/src/vote_transaction.rs

@@ -72,3 +72,33 @@ pub fn new_vote_state_update_transaction(
     vote_tx.partial_sign(&[authorized_voter_keypair], blockhash);
     vote_tx
 }
+
+pub fn new_compact_vote_state_update_transaction(
+    vote_state_update: VoteStateUpdate,
+    blockhash: Hash,
+    node_keypair: &Keypair,
+    vote_keypair: &Keypair,
+    authorized_voter_keypair: &Keypair,
+    switch_proof_hash: Option<Hash>,
+) -> Transaction {
+    let vote_ix = if let Some(switch_proof_hash) = switch_proof_hash {
+        vote::instruction::compact_update_vote_state_switch(
+            &vote_keypair.pubkey(),
+            &authorized_voter_keypair.pubkey(),
+            vote_state_update,
+            switch_proof_hash,
+        )
+    } else {
+        vote::instruction::compact_update_vote_state(
+            &vote_keypair.pubkey(),
+            &authorized_voter_keypair.pubkey(),
+            vote_state_update,
+        )
+    };
+
+    let mut vote_tx = Transaction::new_with_payer(&[vote_ix], Some(&node_keypair.pubkey()));
+
+    vote_tx.partial_sign(&[node_keypair], blockhash);
+    vote_tx.partial_sign(&[authorized_voter_keypair], blockhash);
+    vote_tx
+}

runtime/Cargo.toml

@@ -49,6 +49,7 @@ solana-frozen-abi = { path = "../frozen-abi", version = "=1.15.0" }
 solana-frozen-abi-macro = { path = "../frozen-abi/macro", version = "=1.15.0" }
 solana-measure = { path = "../measure", version = "=1.15.0" }
 solana-metrics = { path = "../metrics", version = "=1.15.0" }
+solana-perf = { path = "../perf", version = "=1.15.0" }
 solana-program-runtime = { path = "../program-runtime", version = "=1.15.0" }
 solana-rayon-threadlimit = { path = "../rayon-threadlimit", version = "=1.15.0" }
 solana-sdk = { path = "../sdk", version = "=1.15.0" }

runtime/src/bank.rs

@@ -83,6 +83,7 @@ use {
     },
     solana_measure::{measure, measure::Measure},
     solana_metrics::{inc_new_counter_debug, inc_new_counter_info},
+    solana_perf::perf_libs,
     solana_program_runtime::{
         accounts_data_meter::MAX_ACCOUNTS_DATA_LEN,
         compute_budget::{self, ComputeBudget},
@@ -105,7 +106,7 @@ use {
         clock::{
             BankId, Epoch, Slot, SlotCount, SlotIndex, UnixTimestamp, DEFAULT_TICKS_PER_SECOND,
             INITIAL_RENT_EPOCH, MAX_PROCESSING_AGE, MAX_TRANSACTION_FORWARDING_DELAY,
-            SECONDS_PER_DAY,
+            MAX_TRANSACTION_FORWARDING_DELAY_GPU, SECONDS_PER_DAY,
         },
         ed25519_program,
         epoch_info::EpochInfo,
@@ -141,7 +142,7 @@ use {
         sysvar::{self, Sysvar, SysvarId},
         timing::years_as_slots,
         transaction::{
-            MessageHash, Result, SanitizedTransaction, Transaction, TransactionError,
+            self, MessageHash, Result, SanitizedTransaction, Transaction, TransactionError,
             TransactionVerificationMode, VersionedTransaction, MAX_TX_ACCOUNT_LOCKS,
         },
         transaction_context::{
@@ -7736,6 +7737,36 @@ impl Bank {
             .epoch_accounts_hash_manager
             .try_get_epoch_accounts_hash()
     }
+
+    /// Checks a batch of sanitized transactions again bank for age and status
+    pub fn check_transactions_with_forwarding_delay(
+        &self,
+        transactions: &[SanitizedTransaction],
+        filter: &[transaction::Result<()>],
+        forward_transactions_to_leader_at_slot_offset: u64,
+    ) -> Vec<TransactionCheckResult> {
+        let mut error_counters = TransactionErrorMetrics::default();
+        // The following code also checks if the blockhash for a transaction is too old
+        // The check accounts for
+        //  1. Transaction forwarding delay
+        //  2. The slot at which the next leader will actually process the transaction
+        // Drop the transaction if it will expire by the time the next node receives and processes it
+        let api = perf_libs::api();
+        let max_tx_fwd_delay = if api.is_none() {
+            MAX_TRANSACTION_FORWARDING_DELAY
+        } else {
+            MAX_TRANSACTION_FORWARDING_DELAY_GPU
+        };
+
+        self.check_transactions(
+            transactions,
+            filter,
+            (MAX_PROCESSING_AGE)
+                .saturating_sub(max_tx_fwd_delay)
+                .saturating_sub(forward_transactions_to_leader_at_slot_offset as usize),
+            &mut error_counters,
+        )
+    }
 }
 
 /// Compute how much an account has changed size.  This function is useful when the data size delta

sdk/program/src/clock.rs

@@ -108,6 +108,10 @@ pub const MAX_TRANSACTION_FORWARDING_DELAY_GPU: usize = 2;
 /// More delay is expected if CUDA is not enabled (as signature verification takes longer)
 pub const MAX_TRANSACTION_FORWARDING_DELAY: usize = 6;
 
+/// Transaction forwarding, which leader to forward to and how long to hold
+pub const FORWARD_TRANSACTIONS_TO_LEADER_AT_SLOT_OFFSET: u64 = 2;
+pub const HOLD_TRANSACTIONS_SLOT_OFFSET: u64 = 20;
+
 /// The unit of time given to a leader for encoding a block.
 ///
 /// It is some some number of _ticks_ long.