rpc-send-tx-svc: add with_config constructor

Cargo.lock

@@ -4664,6 +4664,7 @@ dependencies = [
  "solana-rpc",
  "solana-runtime",
  "solana-sdk",
+ "solana-send-transaction-service",
  "solana-stake-program",
  "solana-streamer",
  "solana-transaction-status",
@@ -5423,6 +5424,7 @@ dependencies = [
  "solana-rpc",
  "solana-runtime",
  "solana-sdk",
+ "solana-send-transaction-service",
  "solana-streamer",
  "solana-validator",
  "solana-version",
@@ -5836,6 +5838,7 @@ dependencies = [
  "solana-rpc",
  "solana-runtime",
  "solana-sdk",
+ "solana-send-transaction-service",
  "solana-streamer",
  "solana-version",
  "solana-vote-program",

core/Cargo.toml

@@ -54,6 +54,7 @@ solana-runtime = { path = "../runtime", version = "=1.9.0" }
 solana-sdk = { path = "../sdk", version = "=1.9.0" }
 solana-frozen-abi = { path = "../frozen-abi", version = "=1.9.0" }
 solana-frozen-abi-macro = { path = "../frozen-abi/macro", version = "=1.9.0" }
+solana-send-transaction-service = { path = "../send-transaction-service", version = "=1.9.0" }
 solana-streamer = { path = "../streamer", version = "=1.9.0" }
 solana-transaction-status = { path = "../transaction-status", version = "=1.9.0" }
 solana-vote-program = { path = "../programs/vote", version = "=1.9.0" }

core/src/validator.rs

@@ -88,6 +88,7 @@ use {
         signature::{Keypair, Signer},
         timing::timestamp,
     },
+    solana_send_transaction_service::send_transaction_service,
     solana_streamer::socket::SocketAddrSpace,
     solana_vote_program::vote_state::VoteState,
     std::{
@@ -148,8 +149,7 @@ pub struct ValidatorConfig {
     pub contact_debug_interval: u64,
     pub contact_save_interval: u64,
     pub bpf_jit: bool,
-    pub send_transaction_retry_ms: u64,
+    pub send_transaction_service_config: send_transaction_service::Config,
-    pub send_transaction_leader_forward_count: u64,
     pub no_poh_speed_test: bool,
     pub poh_pinned_cpu_core: usize,
     pub poh_hashes_per_batch: u64,
@@ -209,8 +209,7 @@ impl Default for ValidatorConfig {
             contact_debug_interval: DEFAULT_CONTACT_DEBUG_INTERVAL_MILLIS,
             contact_save_interval: DEFAULT_CONTACT_SAVE_INTERVAL_MILLIS,
             bpf_jit: false,
-            send_transaction_retry_ms: 2000,
+            send_transaction_service_config: send_transaction_service::Config::default(),
-            send_transaction_leader_forward_count: 2,
             no_poh_speed_test: true,
             poh_pinned_cpu_core: poh_service::DEFAULT_PINNED_CPU_CORE,
             poh_hashes_per_batch: poh_service::DEFAULT_HASHES_PER_BATCH,
@@ -616,8 +615,7 @@ impl Validator {
                     config.trusted_validators.clone(),
                     rpc_override_health_check.clone(),
                     optimistically_confirmed_bank.clone(),
-                    config.send_transaction_retry_ms,
+                    config.send_transaction_service_config.clone(),
-                    config.send_transaction_leader_forward_count,
                     max_slots.clone(),
                     leader_schedule_cache.clone(),
                     max_complete_transaction_status_slot,

local-cluster/src/validator_configs.rs

@@ -44,8 +44,7 @@ pub fn safe_clone_config(config: &ValidatorConfig) -> ValidatorConfig {
         contact_debug_interval: config.contact_debug_interval,
         contact_save_interval: config.contact_save_interval,
         bpf_jit: config.bpf_jit,
-        send_transaction_retry_ms: config.send_transaction_retry_ms,
+        send_transaction_service_config: config.send_transaction_service_config.clone(),
-        send_transaction_leader_forward_count: config.send_transaction_leader_forward_count,
         no_poh_speed_test: config.no_poh_speed_test,
         poh_pinned_cpu_core: config.poh_pinned_cpu_core,
         account_indexes: config.account_indexes.clone(),

replica-node/Cargo.toml

@@ -25,6 +25,7 @@ solana-rpc = { path = "../rpc", version = "=1.9.0" }
 solana-replica-lib = { path = "../replica-lib", version = "=1.9.0" }
 solana-runtime = { path = "../runtime", version = "=1.9.0" }
 solana-sdk = { path = "../sdk", version = "=1.9.0" }
+solana-send-transaction-service = { path = "../send-transaction-service", version = "=1.9.0" }
 solana-streamer = { path = "../streamer", version = "=1.9.0" }
 solana-version = { path = "../version", version = "=1.9.0" }
 solana-validator = { path = "../validator", version = "=1.9.0" }

replica-node/src/replica_node.rs

@@ -26,6 +26,7 @@ use {
         snapshot_config::SnapshotConfig, snapshot_package::SnapshotType, snapshot_utils,
     },
     solana_sdk::{clock::Slot, exit::Exit, genesis_config::GenesisConfig, hash::Hash},
+    solana_send_transaction_service::send_transaction_service,
     solana_streamer::socket::SocketAddrSpace,
     std::{
         fs,
@@ -237,8 +238,11 @@ fn start_client_rpc_services(
             None,
             rpc_override_health_check,
             optimistically_confirmed_bank.clone(),
-            0,
+            send_transaction_service::Config {
-            0,
+                retry_rate_ms: 0,
+                leader_forward_count: 0,
+                ..send_transaction_service::Config::default()
+            },
             max_slots,
             leader_schedule_cache.clone(),
             max_complete_transaction_status_slot,

rpc/src/rpc_service.rs

@@ -34,7 +34,7 @@ use {
         exit::Exit, genesis_config::DEFAULT_GENESIS_DOWNLOAD_PATH, hash::Hash,
         native_token::lamports_to_sol, pubkey::Pubkey,
     },
-    solana_send_transaction_service::send_transaction_service::SendTransactionService,
+    solana_send_transaction_service::send_transaction_service::{self, SendTransactionService},
     std::{
         collections::HashSet,
         net::SocketAddr,
@@ -297,8 +297,7 @@ impl JsonRpcService {
         trusted_validators: Option<HashSet<Pubkey>>,
         override_health_check: Arc<AtomicBool>,
         optimistically_confirmed_bank: Arc<RwLock<OptimisticallyConfirmedBank>>,
-        send_transaction_retry_ms: u64,
+        send_transaction_service_config: send_transaction_service::Config,
-        send_transaction_leader_forward_count: u64,
         max_slots: Arc<MaxSlots>,
         leader_schedule_cache: Arc<LeaderScheduleCache>,
         current_transaction_status_slot: Arc<AtomicU64>,
@@ -395,13 +394,12 @@ impl JsonRpcService {
 
         let leader_info =
             poh_recorder.map(|recorder| ClusterTpuInfo::new(cluster_info.clone(), recorder));
-        let _send_transaction_service = Arc::new(SendTransactionService::new(
+        let _send_transaction_service = Arc::new(SendTransactionService::new_with_config(
             tpu_address,
             &bank_forks,
             leader_info,
             receiver,
-            send_transaction_retry_ms,
+            send_transaction_service_config,
-            send_transaction_leader_forward_count,
         ));
 
         #[cfg(test)]
@@ -557,8 +555,11 @@ mod tests {
             None,
             Arc::new(AtomicBool::new(false)),
             optimistically_confirmed_bank,
-            1000,
+            send_transaction_service::Config {
-            1,
+                retry_rate_ms: 1000,
+                leader_forward_count: 1,
+                ..send_transaction_service::Config::default()
+            },
             Arc::new(MaxSlots::default()),
             Arc::new(LeaderScheduleCache::default()),
             Arc::new(AtomicU64::default()),

send-transaction-service/src/send_transaction_service.rs

@@ -18,6 +18,10 @@ use {
 
 /// Maximum size of the transaction queue
 const MAX_TRANSACTION_QUEUE_SIZE: usize = 10_000; // This seems like a lot but maybe it needs to be bigger one day
+/// Default retry interval
+const DEFAULT_RETRY_RATE_MS: u64 = 2_000;
+/// Default number of leaders to forward transactions to
+const DEFAULT_LEADER_FORWARD_COUNT: u64 = 2;
 
 pub struct SendTransactionService {
     thread: JoinHandle<()>,
@@ -61,6 +65,21 @@ struct ProcessTransactionsResult {
     retained: u64,
 }
 
+#[derive(Clone, Debug)]
+pub struct Config {
+    pub retry_rate_ms: u64,
+    pub leader_forward_count: u64,
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            retry_rate_ms: DEFAULT_RETRY_RATE_MS,
+            leader_forward_count: DEFAULT_LEADER_FORWARD_COUNT,
+        }
+    }
+}
+
 impl SendTransactionService {
     pub fn new<T: TpuInfo + std::marker::Send + 'static>(
         tpu_address: SocketAddr,
@@ -69,14 +88,28 @@ impl SendTransactionService {
         receiver: Receiver<TransactionInfo>,
         retry_rate_ms: u64,
         leader_forward_count: u64,
+    ) -> Self {
+        let config = Config {
+            retry_rate_ms,
+            leader_forward_count,
+            ..Config::default()
+        };
+        Self::new_with_config(tpu_address, bank_forks, leader_info, receiver, config)
+    }
+
+    pub fn new_with_config<T: TpuInfo + std::marker::Send + 'static>(
+        tpu_address: SocketAddr,
+        bank_forks: &Arc<RwLock<BankForks>>,
+        leader_info: Option<T>,
+        receiver: Receiver<TransactionInfo>,
+        config: Config,
     ) -> Self {
         let thread = Self::retry_thread(
             tpu_address,
             receiver,
             bank_forks.clone(),
             leader_info,
-            retry_rate_ms,
+            config,
-            leader_forward_count,
         );
         Self { thread }
     }
@@ -86,8 +119,7 @@ impl SendTransactionService {
         receiver: Receiver<TransactionInfo>,
         bank_forks: Arc<RwLock<BankForks>>,
         mut leader_info: Option<T>,
-        retry_rate_ms: u64,
+        config: Config,
-        leader_forward_count: u64,
     ) -> JoinHandle<()> {
         let mut last_status_check = Instant::now();
         let mut last_leader_refresh = Instant::now();
@@ -101,13 +133,13 @@ impl SendTransactionService {
         Builder::new()
             .name("send-tx-sv2".to_string())
             .spawn(move || loop {
-                match receiver.recv_timeout(Duration::from_millis(1000.min(retry_rate_ms))) {
+                match receiver.recv_timeout(Duration::from_millis(1000.min(config.retry_rate_ms))) {
                     Err(RecvTimeoutError::Disconnected) => break,
                     Err(RecvTimeoutError::Timeout) => {}
                     Ok(transaction_info) => {
-                        let addresses = leader_info
+                        let addresses = leader_info.as_ref().map(|leader_info| {
-                            .as_ref()
+                            leader_info.get_leader_tpus(config.leader_forward_count)
-                            .map(|leader_info| leader_info.get_leader_tpus(leader_forward_count));
+                        });
                         let addresses = addresses
                             .map(|address_list| {
                                 if address_list.is_empty() {
@@ -132,7 +164,7 @@ impl SendTransactionService {
                     }
                 }
 
-                if last_status_check.elapsed().as_millis() as u64 >= retry_rate_ms {
+                if last_status_check.elapsed().as_millis() as u64 >= config.retry_rate_ms {
                     if !transactions.is_empty() {
                         datapoint_info!(
                             "send_transaction_service-queue-size",
@@ -153,7 +185,7 @@ impl SendTransactionService {
                             &tpu_address,
                             &mut transactions,
                             &leader_info,
-                            leader_forward_count,
+                            &config,
                         );
                     }
                     last_status_check = Instant::now();
@@ -175,7 +207,7 @@ impl SendTransactionService {
         tpu_address: &SocketAddr,
         transactions: &mut HashMap<Signature, TransactionInfo>,
         leader_info: &Option<T>,
-        leader_forward_count: u64,
+        config: &Config,
     ) -> ProcessTransactionsResult {
         let mut result = ProcessTransactionsResult::default();
 
@@ -223,9 +255,9 @@ impl SendTransactionService {
                     result.retried += 1;
                     transaction_info.retries += 1;
                     inc_new_counter_info!("send_transaction_service-retry", 1);
-                    let addresses = leader_info
+                    let addresses = leader_info.as_ref().map(|leader_info| {
-                        .as_ref()
+                        leader_info.get_leader_tpus(config.leader_forward_count)
-                        .map(|leader_info| leader_info.get_leader_tpus(leader_forward_count));
+                    });
                     let addresses = addresses
                         .map(|address_list| {
                             if address_list.is_empty() {
@@ -318,7 +350,10 @@ mod test {
         let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
         let send_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
         let tpu_address = "127.0.0.1:0".parse().unwrap();
-        let leader_forward_count = 1;
+        let config = Config {
+            leader_forward_count: 1,
+            ..Config::default()
+        };
 
         let root_bank = Arc::new(Bank::new_from_parent(
             &bank_forks.read().unwrap().working_bank(),
@@ -364,7 +399,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert!(transactions.is_empty());
         assert_eq!(
@@ -393,7 +428,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert!(transactions.is_empty());
         assert_eq!(
@@ -422,7 +457,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert!(transactions.is_empty());
         assert_eq!(
@@ -451,7 +486,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert_eq!(transactions.len(), 1);
         assert_eq!(
@@ -481,7 +516,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert_eq!(transactions.len(), 1);
         assert_eq!(
@@ -521,7 +556,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert_eq!(transactions.len(), 1);
         assert_eq!(
@@ -539,7 +574,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert!(transactions.is_empty());
         assert_eq!(
@@ -560,7 +595,10 @@ mod test {
         let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
         let send_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
         let tpu_address = "127.0.0.1:0".parse().unwrap();
-        let leader_forward_count = 1;
+        let config = Config {
+            leader_forward_count: 1,
+            ..Config::default()
+        };
 
         let root_bank = Arc::new(Bank::new_from_parent(
             &bank_forks.read().unwrap().working_bank(),
@@ -619,7 +657,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert!(transactions.is_empty());
         assert_eq!(
@@ -647,7 +685,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert!(transactions.is_empty());
         assert_eq!(
@@ -677,7 +715,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert!(transactions.is_empty());
         assert_eq!(
@@ -705,7 +743,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert!(transactions.is_empty());
         assert_eq!(
@@ -734,7 +772,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert!(transactions.is_empty());
         assert_eq!(
@@ -763,7 +801,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert_eq!(transactions.len(), 1);
         assert_eq!(
@@ -793,7 +831,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert_eq!(transactions.len(), 1);
         assert_eq!(
@@ -821,7 +859,7 @@ mod test {
             &tpu_address,
             &mut transactions,
             &None,
-            leader_forward_count,
+            &config,
         );
         assert_eq!(transactions.len(), 0);
         assert_eq!(

validator/Cargo.toml

@@ -44,6 +44,7 @@ solana-replica-lib = { path = "../replica-lib", version = "=1.9.0" }
 solana-rpc = { path = "../rpc", version = "=1.9.0" }
 solana-runtime = { path = "../runtime", version = "=1.9.0" }
 solana-sdk = { path = "../sdk", version = "=1.9.0" }
+solana-send-transaction-service = { path = "../send-transaction-service", version = "=1.9.0" }
 solana-streamer = { path = "../streamer", version = "=1.9.0" }
 solana-version = { path = "../version", version = "=1.9.0" }
 solana-vote-program = { path = "../programs/vote", version = "=1.9.0" }

validator/src/main.rs

@@ -59,6 +59,7 @@ use {
         pubkey::Pubkey,
         signature::{Keypair, Signer},
     },
+    solana_send_transaction_service::send_transaction_service,
     solana_streamer::socket::SocketAddrSpace,
     solana_validator::{
         admin_rpc_service, bootstrap, dashboard::Dashboard, ledger_lockfile, lock_ledger,
@@ -422,11 +423,12 @@ pub fn main() {
         PubSubConfig::default().queue_capacity_items.to_string();
     let default_rpc_pubsub_queue_capacity_bytes =
         PubSubConfig::default().queue_capacity_bytes.to_string();
-    let default_rpc_send_transaction_retry_ms = ValidatorConfig::default()
+    let default_send_transaction_service_config = send_transaction_service::Config::default();
-        .send_transaction_retry_ms
+    let default_rpc_send_transaction_retry_ms = default_send_transaction_service_config
+        .retry_rate_ms
         .to_string();
-    let default_rpc_send_transaction_leader_forward_count = ValidatorConfig::default()
+    let default_rpc_send_transaction_leader_forward_count = default_send_transaction_service_config
-        .send_transaction_leader_forward_count
+        .leader_forward_count
         .to_string();
     let default_rpc_threads = num_cpus::get().to_string();
     let default_accountsdb_repl_threads = num_cpus::get().to_string();
@@ -2093,12 +2095,14 @@ pub fn main() {
         debug_keys,
         contact_debug_interval,
         bpf_jit: !matches.is_present("no_bpf_jit"),
-        send_transaction_retry_ms: value_t_or_exit!(matches, "rpc_send_transaction_retry_ms", u64),
+        send_transaction_service_config: send_transaction_service::Config {
-        send_transaction_leader_forward_count: value_t_or_exit!(
+            retry_rate_ms: value_t_or_exit!(matches, "rpc_send_transaction_retry_ms", u64),
+            leader_forward_count: value_t_or_exit!(
                 matches,
                 "rpc_send_transaction_leader_forward_count",
                 u64
             ),
+        },
         no_poh_speed_test: matches.is_present("no_poh_speed_test"),
         poh_pinned_cpu_core: value_of(&matches, "poh_pinned_cpu_core")
             .unwrap_or(poh_service::DEFAULT_PINNED_CPU_CORE),