TransactionView: ReceiveAndBuffer (anza-xyz#3820)

Author: apfitzge

Cargo.lock

@@ -11,7 +11,7 @@ use {
     solana_core::{
         banking_stage::{update_bank_forks_and_poh_recorder_for_new_tpu_bank, BankingStage},
         banking_trace::{BankingTracer, Channels, BANKING_TRACE_DIR_DEFAULT_BYTE_LIMIT},
-        validator::BlockProductionMethod,
+        validator::{BlockProductionMethod, TransactionStructure},
     },
     solana_gossip::cluster_info::{ClusterInfo, Node},
     solana_ledger::{
@@ -290,6 +290,14 @@ fn main() {
                 .possible_values(BlockProductionMethod::cli_names())
                 .help(BlockProductionMethod::cli_message()),
         )
+        .arg(
+            Arg::with_name("transaction_struct")
+                .long("transaction-structure")
+                .value_name("STRUCT")
+                .takes_value(true)
+                .possible_values(TransactionStructure::cli_names())
+                .help(TransactionStructure::cli_message()),
+        )
         .arg(
             Arg::new("num_banking_threads")
                 .long("num-banking-threads")
@@ -320,6 +328,9 @@ fn main() {
     let block_production_method = matches
         .value_of_t::<BlockProductionMethod>("block_production_method")
         .unwrap_or_default();
+    let transaction_struct = matches
+        .value_of_t::<TransactionStructure>("transaction_struct")
+        .unwrap_or_default();
     let num_banking_threads = matches
         .value_of_t::<u32>("num_banking_threads")
         .unwrap_or_else(|_| BankingStage::num_threads());
@@ -470,6 +481,7 @@ fn main() {
     };
     let banking_stage = BankingStage::new_num_threads(
         block_production_method,
+        transaction_struct,
         &cluster_info,
         &poh_recorder,
         non_vote_receiver,

banking-bench/src/main.rs

@@ -15,6 +15,7 @@ codecov = { repository = "solana-labs/solana", branch = "master", service = "git
 
 [dependencies]
 agave-banking-stage-ingress-types = { workspace = true }
+agave-transaction-view = { workspace = true }
 ahash = { workspace = true }
 anyhow = { workspace = true }
 arrayvec = { workspace = true }

core/Cargo.toml

@@ -3,7 +3,10 @@
 
 use {
     agave_banking_stage_ingress_types::BankingPacketBatch,
-    solana_core::{banking_trace::Channels, validator::BlockProductionMethod},
+    solana_core::{
+        banking_trace::Channels,
+        validator::{BlockProductionMethod, TransactionStructure},
+    },
     solana_vote_program::{vote_state::TowerSync, vote_transaction::new_tower_sync_transaction},
 };
 
@@ -193,7 +196,12 @@ enum TransactionType {
     ProgramsAndVotes,
 }
 
-fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
+fn bench_banking(
+    bencher: &mut Bencher,
+    tx_type: TransactionType,
+    block_production_method: BlockProductionMethod,
+    transaction_struct: TransactionStructure,
+) {
     solana_logger::setup();
     let num_threads = BankingStage::num_threads() as usize;
     //   a multiple of packet chunk duplicates to avoid races
@@ -297,7 +305,8 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
     let cluster_info = Arc::new(cluster_info);
     let (s, _r) = unbounded();
     let _banking_stage = BankingStage::new(
-        BlockProductionMethod::CentralScheduler,
+        block_production_method,
+        transaction_struct,
         &cluster_info,
         &poh_recorder,
         non_vote_receiver,
@@ -372,22 +381,82 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
 
 #[bench]
 fn bench_banking_stage_multi_accounts(bencher: &mut Bencher) {
-    bench_banking(bencher, TransactionType::Accounts);
+    bench_banking(
+        bencher,
+        TransactionType::Accounts,
+        BlockProductionMethod::CentralScheduler,
+        TransactionStructure::Sdk,
+    );
 }
 
 #[bench]
 fn bench_banking_stage_multi_programs(bencher: &mut Bencher) {
-    bench_banking(bencher, TransactionType::Programs);
+    bench_banking(
+        bencher,
+        TransactionType::Programs,
+        BlockProductionMethod::CentralScheduler,
+        TransactionStructure::Sdk,
+    );
 }
 
 #[bench]
 fn bench_banking_stage_multi_accounts_with_voting(bencher: &mut Bencher) {
-    bench_banking(bencher, TransactionType::AccountsAndVotes);
+    bench_banking(
+        bencher,
+        TransactionType::AccountsAndVotes,
+        BlockProductionMethod::CentralScheduler,
+        TransactionStructure::Sdk,
+    );
 }
 
 #[bench]
 fn bench_banking_stage_multi_programs_with_voting(bencher: &mut Bencher) {
-    bench_banking(bencher, TransactionType::ProgramsAndVotes);
+    bench_banking(
+        bencher,
+        TransactionType::ProgramsAndVotes,
+        BlockProductionMethod::CentralScheduler,
+        TransactionStructure::Sdk,
+    );
+}
+
+#[bench]
+fn bench_banking_stage_multi_accounts_view(bencher: &mut Bencher) {
+    bench_banking(
+        bencher,
+        TransactionType::Accounts,
+        BlockProductionMethod::CentralScheduler,
+        TransactionStructure::View,
+    );
+}
+
+#[bench]
+fn bench_banking_stage_multi_programs_view(bencher: &mut Bencher) {
+    bench_banking(
+        bencher,
+        TransactionType::Programs,
+        BlockProductionMethod::CentralScheduler,
+        TransactionStructure::View,
+    );
+}
+
+#[bench]
+fn bench_banking_stage_multi_accounts_with_voting_view(bencher: &mut Bencher) {
+    bench_banking(
+        bencher,
+        TransactionType::AccountsAndVotes,
+        BlockProductionMethod::CentralScheduler,
+        TransactionStructure::View,
+    );
+}
+
+#[bench]
+fn bench_banking_stage_multi_programs_with_voting_view(bencher: &mut Bencher) {
+    bench_banking(
+        bencher,
+        TransactionType::ProgramsAndVotes,
+        BlockProductionMethod::CentralScheduler,
+        TransactionStructure::View,
+    );
 }
 
 fn simulate_process_entries(

core/benches/banking_stage.rs

@@ -8,7 +8,7 @@ use {
             BankingTracer, ChannelLabel, Channels, TimedTracedEvent, TracedEvent, TracedSender,
             TracerThread, BANKING_TRACE_DIR_DEFAULT_BYTE_LIMIT, BASENAME,
         },
-        validator::BlockProductionMethod,
+        validator::{BlockProductionMethod, TransactionStructure},
     },
     agave_banking_stage_ingress_types::BankingPacketBatch,
     assert_matches::assert_matches,
@@ -676,6 +676,7 @@ impl BankingSimulator {
         bank_forks: Arc<RwLock<BankForks>>,
         blockstore: Arc<Blockstore>,
         block_production_method: BlockProductionMethod,
+        transaction_struct: TransactionStructure,
     ) -> (SenderLoop, SimulatorLoop, SimulatorThreads) {
         let parent_slot = self.parent_slot().unwrap();
         let mut packet_batches_by_time = self.banking_trace_events.packet_batches_by_time;
@@ -807,6 +808,7 @@ impl BankingSimulator {
         let prioritization_fee_cache = &Arc::new(PrioritizationFeeCache::new(0u64));
         let banking_stage = BankingStage::new_num_threads(
             block_production_method.clone(),
+            transaction_struct.clone(),
             &cluster_info,
             &poh_recorder,
             non_vote_receiver,
@@ -893,12 +895,14 @@ impl BankingSimulator {
         bank_forks: Arc<RwLock<BankForks>>,
         blockstore: Arc<Blockstore>,
         block_production_method: BlockProductionMethod,
+        transaction_struct: TransactionStructure,
     ) -> Result<(), SimulateError> {
         let (sender_loop, simulator_loop, simulator_threads) = self.prepare_simulation(
             genesis_config,
             bank_forks,
             blockstore,
             block_production_method,
+            transaction_struct,
         );
 
         sender_loop.log_starting();