Encapsulate transaction storage filtering better

solana-labs · Sep 30, 2022 · 4e8c730 · 4e8c730
1 parent 082b762
commit 4e8c730
Show file tree

Hide file tree

Showing 12 changed files with 597 additions and 594 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/core/benches/unprocessed_packet_batches.rs b/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,
-            &mut unprocessed_packet_batches,
             &mut forward_packet_batches_by_accounts,
-            128usize,
         );
     }
 }