Multi packet size support - WIP

accountsdb-plugin-postgres/src/postgres_client/postgres_client_transaction.rs

@@ -1294,7 +1294,7 @@ pub(crate) mod tests {
         let transaction = VersionedTransaction::from(transaction);
 
         let transaction =
-            SanitizedTransaction::try_create(transaction, message_hash, Some(true), |_| {
+            SanitizedTransaction::try_create(transaction, message_hash, Some(true), None, |_| {
                 Err(TransactionError::UnsupportedVersion)
             })
             .unwrap();
@@ -1333,7 +1333,7 @@ pub(crate) mod tests {
         transaction.sanitize().unwrap();
 
         let transaction =
-            SanitizedTransaction::try_create(transaction, message_hash, Some(true), |_message| {
+            SanitizedTransaction::try_create(transaction, message_hash, Some(true), None, |_message| {
                 Ok(LoadedAddresses {
                     writable: vec![Pubkey::new_unique(), Pubkey::new_unique()],
                     readonly: vec![Pubkey::new_unique(), Pubkey::new_unique()],

banking-bench/src/main.rs

@@ -13,14 +13,15 @@ use {
         get_tmp_ledger_path,
     },
     solana_measure::measure::Measure,
-    solana_perf::packet::to_packet_batches,
+    solana_perf::packet::{to_packet_batches, PacketBatch},
     solana_poh::poh_recorder::{create_test_recorder, PohRecorder, WorkingBankEntry},
     solana_runtime::{
         accounts_background_service::AbsRequestSender, bank::Bank, bank_forks::BankForks,
         cost_model::CostModel,
     },
     solana_sdk::{
         hash::Hash,
+        packet::Packet,
         signature::{Keypair, Signature},
         system_transaction,
         timing::{duration_as_us, timestamp},
@@ -167,6 +168,8 @@ fn main() {
     } = create_genesis_config(mint_total);
 
     let (verified_sender, verified_receiver) = unbounded();
+    // TODO: Actually pump some stuff into _verified_extended_sender like verified_sender ?
+    let (verified_extended_sender, verified_extended_receiver) = unbounded();
     let (vote_sender, vote_receiver) = unbounded();
     let (tpu_vote_sender, tpu_vote_receiver) = unbounded();
     let (replay_vote_sender, _replay_vote_receiver) = unbounded();
@@ -212,7 +215,8 @@ fn main() {
         bank.clear_signatures();
     }
 
-    let mut verified: Vec<_> = to_packet_batches(&transactions, packets_per_chunk);
+    let mut verified: Vec<PacketBatch<Packet>> =
+        to_packet_batches(&transactions, packets_per_chunk);
     let ledger_path = get_tmp_ledger_path!();
     {
         let blockstore = Arc::new(
@@ -231,6 +235,7 @@ fn main() {
             &cluster_info,
             &poh_recorder,
             verified_receiver,
+            verified_extended_receiver,
             tpu_vote_receiver,
             vote_receiver,
             None,
@@ -389,6 +394,7 @@ fn main() {
         );
 
         drop(verified_sender);
+        drop(verified_extended_sender);
         drop(tpu_vote_sender);
         drop(vote_sender);
         exit.store(true, Ordering::Relaxed);

banks-server/src/banks_server.rs

@@ -373,6 +373,7 @@ pub async fn start_local_server(
 pub async fn start_tcp_server(
     listen_addr: SocketAddr,
     tpu_addr: SocketAddr,
+    tpu_extended_addr: SocketAddr,
     bank_forks: Arc<RwLock<BankForks>>,
     block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
 ) -> io::Result<()> {
@@ -396,6 +397,7 @@ pub async fn start_tcp_server(
 
             SendTransactionService::new::<NullTpuInfo>(
                 tpu_addr,
+                tpu_extended_addr,
                 &bank_forks,
                 None,
                 receiver,

banks-server/src/rpc_banks_service.rs

@@ -27,13 +27,15 @@ pub struct RpcBanksService {
 async fn start_abortable_tcp_server(
     listen_addr: SocketAddr,
     tpu_addr: SocketAddr,
+    tpu_extended_addr: SocketAddr,
     bank_forks: Arc<RwLock<BankForks>>,
     block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
     exit: Arc<AtomicBool>,
 ) {
     let server = start_tcp_server(
         listen_addr,
         tpu_addr,
+        tpu_extended_addr,
         bank_forks.clone(),
         block_commitment_cache.clone(),
     )
@@ -56,13 +58,15 @@ impl RpcBanksService {
     fn run(
         listen_addr: SocketAddr,
         tpu_addr: SocketAddr,
+        tpu_extended_addr: SocketAddr,
         bank_forks: Arc<RwLock<BankForks>>,
         block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
         exit: Arc<AtomicBool>,
     ) {
         let server = start_abortable_tcp_server(
             listen_addr,
             tpu_addr,
+            tpu_extended_addr,
             bank_forks,
             block_commitment_cache,
             exit,
@@ -73,6 +77,7 @@ impl RpcBanksService {
     pub fn new(
         listen_addr: SocketAddr,
         tpu_addr: SocketAddr,
+        tpu_extended_addr: SocketAddr,
         bank_forks: &Arc<RwLock<BankForks>>,
         block_commitment_cache: &Arc<RwLock<BlockCommitmentCache>>,
         exit: &Arc<AtomicBool>,
@@ -86,6 +91,7 @@ impl RpcBanksService {
                 Self::run(
                     listen_addr,
                     tpu_addr,
+                    tpu_extended_addr,
                     bank_forks,
                     block_commitment_cache,
                     exit,

bench-streamer/src/main.rs

@@ -3,7 +3,7 @@ use {
     clap::{crate_description, crate_name, App, Arg},
     solana_streamer::{
         packet::{Packet, PacketBatch, PacketBatchRecycler, PACKET_DATA_SIZE},
-        streamer::{receiver, PacketBatchReceiver},
+        streamer::{receiver, StandardPacketReceiver},
     },
     std::{
         cmp::max,
@@ -42,7 +42,8 @@ fn producer(addr: &SocketAddr, exit: Arc<AtomicBool>) -> JoinHandle<()> {
     })
 }
 
-fn sink(exit: Arc<AtomicBool>, rvs: Arc<AtomicUsize>, r: PacketBatchReceiver) -> JoinHandle<()> {
+// TODO: Revisit whether this should be generic
+fn sink(exit: Arc<AtomicBool>, rvs: Arc<AtomicUsize>, r: StandardPacketReceiver) -> JoinHandle<()> {
     spawn(move || loop {
         if exit.load(Ordering::Relaxed) {
             return;

bench-tps/Cargo.toml

@@ -27,7 +27,7 @@ solana-runtime = { path = "../runtime", version = "=1.10.0" }
 solana-sdk = { path = "../sdk", version = "=1.10.0" }
 solana-streamer = { path = "../streamer", version = "=1.10.0" }
 solana-version = { path = "../version", version = "=1.10.0" }
-
+bincode = "1.3.3"
 [dev-dependencies]
 serial_test = "0.5.1"
 solana-local-cluster = { path = "../local-cluster", version = "=1.10.0" }

bench-tps/src/bench.rs

@@ -18,7 +18,7 @@ use {
         signature::{Keypair, Signer},
         system_instruction, system_transaction,
         timing::{duration_as_ms, duration_as_s, duration_as_us, timestamp},
-        transaction::Transaction,
+        transaction::{Transaction, TransactionError},
     },
     std::{
         collections::{HashSet, VecDeque},
@@ -324,6 +324,29 @@ fn metrics_submit_lamport_balance(lamport_balance: u64) {
     );
 }
 
+/// Create and sign new large transaction of many system_instruction::Transfer
+pub fn big_transfer(
+    from_keypair: &Keypair,
+    to: &Pubkey,
+    lamports: u64,
+    recent_blockhash: Hash,
+) -> Transaction {
+    let from_pubkey = from_keypair.pubkey();
+
+    let instructions = (0..128).map(|_|
+        system_instruction::transfer(&from_pubkey, &to, lamports)
+        ).collect::<Vec<_>>();
+
+    let tx = Transaction::new(&[from_keypair],
+        Message::new(
+        &instructions,
+        Some(&from_pubkey),
+    ),
+    recent_blockhash);
+    info!("transaction size: {}", bincode::serialized_size(&tx).map_err(|_| 0 as u64).unwrap());
+    tx
+}
+
 fn generate_system_txs(
     source: &[&Keypair],
     dest: &VecDeque<&Keypair>,
@@ -340,7 +363,7 @@ fn generate_system_txs(
         .par_iter()
         .map(|(from, to)| {
             (
-                system_transaction::transfer(from, &to.pubkey(), 1, *blockhash),
+                big_transfer(from, &to.pubkey(), 1, *blockhash),
                 timestamp(),
             )
         })

bench-tps/src/main.rs

@@ -48,6 +48,7 @@ fn main() {
         info!("Generating {} keypairs", keypair_count);
         let (keypairs, _) = generate_keypairs(id, keypair_count as u64);
         let num_accounts = keypairs.len() as u64;
+        //let num_lamports_per_account = num_lamports_per_account*200;
         let max_fee =
             FeeRateGovernor::new(*target_lamports_per_signature, 0).max_lamports_per_signature;
         let num_lamports_per_account = (num_accounts - 1 + NUM_SIGNATURES_FOR_TXS * max_fee)