Bench-tps: flush tx queue when too old (#6201) (#6208)

automerge

bench-tps/src/bench.rs

@@ -10,25 +10,33 @@ use solana_drone::drone::request_airdrop_transaction;
 use solana_librapay_api::{create_genesis, upload_mint_program, upload_payment_program};
 use solana_measure::measure::Measure;
 use solana_metrics::datapoint_info;
-use solana_sdk::client::Client;
+use solana_sdk::{
-use solana_sdk::fee_calculator::FeeCalculator;
+    client::Client,
-use solana_sdk::hash::Hash;
+    clock::{DEFAULT_TICKS_PER_SLOT, MAX_RECENT_BLOCKHASHES},
-use solana_sdk::pubkey::Pubkey;
+    fee_calculator::FeeCalculator,
-use solana_sdk::signature::{Keypair, KeypairUtil};
+    hash::Hash,
-use solana_sdk::system_instruction;
+    pubkey::Pubkey,
-use solana_sdk::system_transaction;
+    signature::{Keypair, KeypairUtil},
-use solana_sdk::timing::timestamp;
+    system_instruction, system_transaction,
-use solana_sdk::timing::{duration_as_ms, duration_as_s};
+    timing::{duration_as_ms, duration_as_s, timestamp},
-use solana_sdk::transaction::Transaction;
+    transaction::Transaction,
-use std::cmp;
+};
-use std::collections::VecDeque;
+use std::{
-use std::net::SocketAddr;
+    cmp,
-use std::sync::atomic::{AtomicBool, AtomicIsize, AtomicUsize, Ordering};
+    collections::VecDeque,
-use std::sync::{Arc, RwLock};
+    net::SocketAddr,
-use std::thread::sleep;
+    sync::{
-use std::thread::Builder;
+        atomic::{AtomicBool, AtomicIsize, AtomicUsize, Ordering},
-use std::time::Duration;
+        Arc, RwLock,
-use std::time::Instant;
+    },
+    thread::{sleep, Builder},
+    time::{Duration, Instant},
+};
+
+// The point at which transactions become "too old", in seconds. The cluster keeps blockhashes for
+// approximately MAX_RECENT_BLOCKHASHES/DEFAULT_TICKS_PER_SLOT seconds. The adjustment of 5sec
+// seems about right to minimize BlockhashNotFound errors, based on empirical testing.
+const MAX_TX_QUEUE_AGE: u64 = MAX_RECENT_BLOCKHASHES as u64 / DEFAULT_TICKS_PER_SLOT - 5;
 
 use solana_librapay_api::librapay_transaction;
 
@@ -385,15 +393,23 @@ fn do_tx_transfers<T: Client>(
             );
             let tx_len = txs0.len();
             let transfer_start = Instant::now();
+            let mut old_transactions = false;
             for tx in txs0 {
                 let now = timestamp();
-                if now > tx.1 && now - tx.1 > 1000 * 30 {
+                // Transactions that are too old will be rejected by the cluster Don't bother
+                // sending them.
+                if now > tx.1 && now - tx.1 > 1000 * MAX_TX_QUEUE_AGE {
+                    old_transactions = true;
                     continue;
                 }
                 client
                     .async_send_transaction(tx.0)
                     .expect("async_send_transaction in do_tx_transfers");
             }
+            if old_transactions {
+                let mut shared_txs_wl = shared_txs.write().expect("write lock in do_tx_transfers");
+                shared_txs_wl.clear();
+            }
             shared_tx_thread_count.fetch_add(-1, Ordering::Relaxed);
             total_tx_sent_count.fetch_add(tx_len, Ordering::Relaxed);
             println!(