Process async BankClient transactions in batches (#3738)

* Process async transactions in batches

This aims to process transactions at least as fast as LocalCluster

* Add benchmark

runtime/benches/bank.rs

@@ -3,20 +3,19 @@
 extern crate test;
 
 use solana_runtime::bank::*;
+use solana_runtime::bank_client::BankClient;
+use solana_sdk::client::AsyncClient;
 use solana_sdk::genesis_block::GenesisBlock;
 use solana_sdk::hash::hash;
 use solana_sdk::signature::{Keypair, KeypairUtil};
 use solana_sdk::system_transaction;
 use solana_sdk::timing::{DEFAULT_TICKS_PER_SLOT, MAX_RECENT_BLOCKHASHES};
+use solana_sdk::transaction::Transaction;
 use test::Bencher;
 
-#[bench]
+// Create transactions between unrelated parties.
-fn bench_process_transaction(bencher: &mut Bencher) {
+pub fn create_sample_transactions(bank: &Bank, mint_keypair: &Keypair) -> Vec<Transaction> {
-    let (genesis_block, mint_keypair) = GenesisBlock::new(100_000_000);
+    (0..4096)
-    let bank = Bank::new(&genesis_block);
-
-    // Create transactions between unrelated parties.
-    let transactions: Vec<_> = (0..4096)
         .into_iter()
         .map(|_| {
             // Seed the 'from' account.
@@ -32,22 +31,22 @@ fn bench_process_transaction(bencher: &mut Bencher) {
 
             // Seed the 'to' account and a cell for its signature.
             let rando1 = Keypair::new();
-            let tx = system_transaction::transfer(
+            system_transaction::transfer(&rando0, &rando1.pubkey(), 1, bank.last_blockhash(), 0)
-                &rando0,
-                &rando1.pubkey(),
-                1,
-                bank.last_blockhash(),
-                0,
-            );
-            assert_eq!(bank.process_transaction(&tx), Ok(()));
-
-            // Finally, return the transaction to the benchmark.
-            tx
         })
-        .collect();
+        .collect()
+}
+
+#[bench]
+fn bench_process_transaction(bencher: &mut Bencher) {
+    let (genesis_block, mint_keypair) = GenesisBlock::new(100_000_000);
+    let bank = Bank::new(&genesis_block);
+    let transactions = create_sample_transactions(&bank, &mint_keypair);
+
+    // Run once to create all the 'to' accounts.
+    let results = bank.process_transactions(&transactions);
+    assert!(results.iter().all(Result::is_ok));
 
     let mut id = bank.last_blockhash();
-
     for _ in 0..(MAX_RECENT_BLOCKHASHES * DEFAULT_TICKS_PER_SLOT as usize) {
         bank.register_tick(&id);
         id = hash(&id.as_ref())
@@ -60,3 +59,18 @@ fn bench_process_transaction(bencher: &mut Bencher) {
         assert!(results.iter().all(Result::is_ok));
     })
 }
+
+#[bench]
+fn bench_bank_client(bencher: &mut Bencher) {
+    let (genesis_block, mint_keypair) = GenesisBlock::new(100_000_000);
+    let bank = Bank::new(&genesis_block);
+    let transactions = create_sample_transactions(&bank, &mint_keypair);
+
+    bencher.iter(|| {
+        let bank = Bank::new(&genesis_block);
+        let bank_client = BankClient::new(bank);
+        for transaction in transactions.clone() {
+            bank_client.async_send_transaction(transaction).unwrap();
+        }
+    })
+}

runtime/src/bank_client.rs

@@ -1,5 +1,5 @@
 use crate::bank::Bank;
-use solana_sdk::client::{AsyncClient, SyncClient};
+use solana_sdk::client::{AsyncClient, Client, SyncClient};
 use solana_sdk::hash::Hash;
 use solana_sdk::instruction::Instruction;
 use solana_sdk::message::Message;
@@ -10,17 +10,22 @@ use solana_sdk::system_instruction;
 use solana_sdk::transaction::{self, Transaction};
 use solana_sdk::transport::Result;
 use std::io;
+use std::sync::mpsc::{channel, Receiver, Sender};
+use std::sync::Arc;
+use std::thread::Builder;
 
 pub struct BankClient {
-    bank: Bank,
+    bank: Arc<Bank>,
+    transaction_sender: Sender<Transaction>,
 }
 
+impl Client for BankClient {}
+
 impl AsyncClient for BankClient {
     fn async_send_transaction(&self, transaction: Transaction) -> io::Result<Signature> {
-        // Ignore the result. Client must use get_signature_status() instead.
+        let signature = transaction.signatures.get(0).cloned().unwrap_or_default();
-        let _ = self.bank.process_transaction(&transaction);
+        self.transaction_sender.send(transaction).unwrap();
-
+        Ok(signature)
-        Ok(transaction.signatures.get(0).cloned().unwrap_or_default())
     }
 
     fn async_send_message(
@@ -104,8 +109,29 @@ impl SyncClient for BankClient {
 }
 
 impl BankClient {
+    fn run(bank: &Bank, transaction_receiver: Receiver<Transaction>) {
+        while let Ok(tx) = transaction_receiver.recv() {
+            let mut transactions = vec![tx];
+            while let Ok(tx) = transaction_receiver.try_recv() {
+                transactions.push(tx);
+            }
+            let _ = bank.process_transactions(&transactions);
+        }
+    }
+
     pub fn new(bank: Bank) -> Self {
-        Self { bank }
+        let bank = Arc::new(bank);
+        let (transaction_sender, transaction_receiver) = channel();
+        let thread_bank = bank.clone();
+        let bank = bank.clone();
+        Builder::new()
+            .name("solana-bank-client".to_string())
+            .spawn(move || Self::run(&thread_bank, transaction_receiver))
+            .unwrap();
+        Self {
+            bank,
+            transaction_sender,
+        }
     }
 }