blockworks-foundation
/
solana-with-rpc-optimizations
mirror of https://github.com/blockworks-foundation/solana-with-rpc-optimizations.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor thin client (#30229) 

Make client/thin_client a thin wrapper and forward calls to the thin-client/thin_client
This commit is contained in:
Lijun Wang 2023-02-10 09:55:13 -08:00 committed by GitHub
parent dcb2d6c8ae
commit 84fbf9273b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 238 additions and 446 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
bench-tps/src/main.rs
View File

@ -116,7 +116,7 @@ fn create_client(
rpc_client,
websocket_url,
TpuClientConfig::default(),
Arc::new(cache),
cache,
)
.unwrap_or_else(|err| {
eprintln!("Could not create TpuClient {err:?}");
@ -128,7 +128,7 @@ fn create_client(
rpc_client,
websocket_url,
TpuClientConfig::default(),
Arc::new(cache),
cache,
)
.unwrap_or_else(|err| {
eprintln!("Could not create TpuClient {err:?}");

4
cli/src/program.rs
View File

@ -2168,7 +2168,7 @@ fn send_deploy_messages(
rpc_client.clone(),
&config.websocket_url,
TpuClientConfig::default(),
Arc::new(cache),
cache,
)?
.send_and_confirm_messages_with_spinner(
write_messages,
@ -2178,7 +2178,7 @@ fn send_deploy_messages(
rpc_client.clone(),
&config.websocket_url,
TpuClientConfig::default(),
Arc::new(cache),
cache,
)?
.send_and_confirm_messages_with_spinner(
write_messages,

8
client/src/connection_cache.rs
View File

@ -25,8 +25,8 @@ const DEFAULT_CONNECTION_CACHE_USE_QUIC: bool = true;
/// construction of the ConnectionCache for code dealing both with udp and quic.
/// For the scenario only using udp or quic, use connection-cache/ConnectionCache directly.
pub enum ConnectionCache {
Quic(BackendConnectionCache<QuicPool, QuicConnectionManager, QuicConfig>),
Udp(BackendConnectionCache<UdpPool, UdpConnectionManager, UdpConfig>),
Quic(Arc<BackendConnectionCache<QuicPool, QuicConnectionManager, QuicConfig>>),
Udp(Arc<BackendConnectionCache<UdpPool, UdpConnectionManager, UdpConfig>>),
}
type QuicBaseClientConnection = <QuicPool as ConnectionPool>::BaseClientConnection;
@ -71,7 +71,7 @@ impl ConnectionCache {
}
let connection_manager = QuicConnectionManager::new_with_connection_config(config);
let cache = BackendConnectionCache::new(connection_manager, connection_pool_size).unwrap();
Self::Quic(cache)
Self::Quic(Arc::new(cache))
}
#[deprecated(
@ -102,7 +102,7 @@ impl ConnectionCache {
let connection_pool_size = 1.max(connection_pool_size);
let connection_manager = UdpConnectionManager::default();
let cache = BackendConnectionCache::new(connection_manager, connection_pool_size).unwrap();
Self::Udp(cache)
Self::Udp(Arc::new(cache))
}
pub fn use_quic(&self) -> bool {

2
client/src/nonblocking/tpu_client.rs
View File

@ -85,7 +85,7 @@ impl TpuClient<QuicPool, QuicConnectionManager, QuicConfig> {
config: TpuClientConfig,
) -> Result<Self> {
let connection_cache = match ConnectionCache::default() {
ConnectionCache::Quic(cache) => Arc::new(cache),
ConnectionCache::Quic(cache) => cache,
ConnectionCache::Udp(_) => {
return Err(TpuSenderError::Custom(String::from(
"Invalid default connection cache",

660
client/src/thin_client.rs
View File

@ -5,15 +5,13 @@
use {
crate::connection_cache::ConnectionCache,
log::*,
rayon::iter::{IntoParallelIterator, ParallelIterator},
solana_connection_cache::client_connection::ClientConnection,
solana_quic_client::{QuicConfig, QuicConnectionManager, QuicPool},
solana_rpc_client::rpc_client::RpcClient,
solana_rpc_client_api::{config::RpcProgramAccountsConfig, response::Response},
solana_rpc_client_api::config::RpcProgramAccountsConfig,
solana_sdk::{
account::Account,
client::{AsyncClient, Client, SyncClient},
clock::{Slot, MAX_PROCESSING_AGE},
clock::Slot,
commitment_config::CommitmentConfig,
epoch_info::EpochInfo,
fee_calculator::{FeeCalculator, FeeRateGovernor},
@ -21,28 +19,77 @@ use {
instruction::Instruction,
message::Message,
pubkey::Pubkey,
signature::{Keypair, Signature, Signer},
signature::{Keypair, Signature},
signers::Signers,
system_instruction,
timing::duration_as_ms,
transaction::{self, Transaction, VersionedTransaction},
transport::Result as TransportResult,
},
solana_thin_client::thin_client::temporary_pub::*,
std::{
io,
net::SocketAddr,
sync::Arc,
time::{Duration, Instant},
},
solana_thin_client::thin_client::ThinClient as BackendThinClient,
solana_udp_client::{UdpConfig, UdpConnectionManager, UdpPool},
std::{net::SocketAddr, sync::Arc, time::Duration},
};
/// An object for querying and sending transactions to the network.
pub struct ThinClient {
rpc_clients: Vec<RpcClient>,
tpu_addrs: Vec<SocketAddr>,
optimizer: ClientOptimizer,
connection_cache: Arc<ConnectionCache>,
/// A thin wrapper over thin-client/ThinClient to ease
/// construction of the ThinClient for code dealing both with udp and quic.
/// For the scenario only using udp or quic, use thin-client/ThinClient directly.
pub enum ThinClient {
Quic(BackendThinClient<QuicPool, QuicConnectionManager, QuicConfig>),
Udp(BackendThinClient<UdpPool, UdpConnectionManager, UdpConfig>),
}
/// Macros easing the forwarding calls to the BackendThinClient
macro_rules! dispatch {
/* Regular version */
($vis:vis fn $name:ident(&self $(, $arg:ident : $ty:ty)*) $(-> $out:ty)?) => {
#[inline]
$vis fn $name(&self $(, $arg:$ty)*) $(-> $out)? {
match self {
Self::Quic(this) => this.$name($($arg),*),
Self::Udp(this) => this.$name($($arg),*),
}
}
};
/* The self is a mut */
($vis:vis fn $name:ident(&mut self $(, $arg:ident : $ty:ty)*) $(-> $out:ty)?) => {
#[inline]
$vis fn $name(&mut self $(, $arg:$ty)*) $(-> $out)? {
match self {
Self::Quic(this) => this.$name($($arg),*),
Self::Udp(this) => this.$name($($arg),*),
}
}
};
/* There is a type parameter */
($vis:vis fn $name:ident<T: Signers>(&self $(, $arg:ident : $ty:ty)*) $(-> $out:ty)?) => {
#[inline]
$vis fn $name<T: Signers>(&self $(, $arg:$ty)*) $(-> $out)? {
match self {
Self::Quic(this) => this.$name($($arg),*),
Self::Udp(this) => this.$name($($arg),*),
}
}
};
}
/// Macro forwarding calls to BackendThinClient with deprecated functions
macro_rules! dispatch_allow_deprecated {
($vis:vis fn $name:ident(&self $(, $arg:ident : $ty:ty)*) $(-> $out:ty)?) => {
#[inline]
$vis fn $name(&self $(, $arg:$ty)*) $(-> $out)? {
match self {
Self::Quic(this) => {
#[allow(deprecated)]
this.$name($($arg),*)
}
Self::Udp(this) => {
#[allow(deprecated)]
this.$name($($arg),*)
}
}
}
};
}
impl ThinClient {
@ -54,7 +101,18 @@ impl ThinClient {
tpu_addr: SocketAddr,
connection_cache: Arc<ConnectionCache>,
) -> Self {
Self::new_from_client(RpcClient::new_socket(rpc_addr), tpu_addr, connection_cache)
match &*connection_cache {
ConnectionCache::Quic(connection_cache) => {
let thin_client =
BackendThinClient::new(rpc_addr, tpu_addr, connection_cache.clone());
ThinClient::Quic(thin_client)
}
ConnectionCache::Udp(connection_cache) => {
let thin_client =
BackendThinClient::new(rpc_addr, tpu_addr, connection_cache.clone());
ThinClient::Udp(thin_client)
}
}
}
pub fn new_socket_with_timeout(
@ -63,20 +121,25 @@ impl ThinClient {
timeout: Duration,
connection_cache: Arc<ConnectionCache>,
) -> Self {
let rpc_client = RpcClient::new_socket_with_timeout(rpc_addr, timeout);
Self::new_from_client(rpc_client, tpu_addr, connection_cache)
}
fn new_from_client(
rpc_client: RpcClient,
tpu_addr: SocketAddr,
connection_cache: Arc<ConnectionCache>,
) -> Self {
Self {
rpc_clients: vec![rpc_client],
tpu_addrs: vec![tpu_addr],
optimizer: ClientOptimizer::new(0),
connection_cache,
match &*connection_cache {
ConnectionCache::Quic(connection_cache) => {
let thin_client = BackendThinClient::new_socket_with_timeout(
rpc_addr,
tpu_addr,
timeout,
connection_cache.clone(),
);
ThinClient::Quic(thin_client)
}
ConnectionCache::Udp(connection_cache) => {
let thin_client = BackendThinClient::new_socket_with_timeout(
rpc_addr,
tpu_addr,
timeout,
connection_cache.clone(),
);
ThinClient::Udp(thin_client)
}
}
}
@ -85,470 +148,199 @@ impl ThinClient {
tpu_addrs: Vec<SocketAddr>,
connection_cache: Arc<ConnectionCache>,
) -> Self {
assert!(!rpc_addrs.is_empty());
assert_eq!(rpc_addrs.len(), tpu_addrs.len());
let rpc_clients: Vec<_> = rpc_addrs.into_iter().map(RpcClient::new_socket).collect();
let optimizer = ClientOptimizer::new(rpc_clients.len());
Self {
rpc_clients,
tpu_addrs,
optimizer,
connection_cache,
match &*connection_cache {
ConnectionCache::Quic(connection_cache) => {
let thin_client = BackendThinClient::new_from_addrs(
rpc_addrs,
tpu_addrs,
connection_cache.clone(),
);
ThinClient::Quic(thin_client)
}
ConnectionCache::Udp(connection_cache) => {
let thin_client = BackendThinClient::new_from_addrs(
rpc_addrs,
tpu_addrs,
connection_cache.clone(),
);
ThinClient::Udp(thin_client)
}
}
}
fn tpu_addr(&self) -> &SocketAddr {
&self.tpu_addrs[self.optimizer.best()]
}
dispatch!(pub fn rpc_client(&self) -> &RpcClient);
pub fn rpc_client(&self) -> &RpcClient {
&self.rpc_clients[self.optimizer.best()]
}
dispatch!(pub fn retry_transfer_until_confirmed(&self, keypair: &Keypair, transaction: &mut Transaction, tries: usize, min_confirmed_blocks: usize) -> TransportResult<Signature>);
/// Retry a sending a signed Transaction to the server for processing.
pub fn retry_transfer_until_confirmed(
dispatch!(pub fn retry_transfer(
&self,
keypair: &Keypair,
transaction: &mut Transaction,
tries: usize,
min_confirmed_blocks: usize,
) -> TransportResult<Signature> {
self.send_and_confirm_transaction(&[keypair], transaction, tries, min_confirmed_blocks)
}
tries: usize
) -> TransportResult<Signature>);
/// Retry sending a signed Transaction with one signing Keypair to the server for processing.
pub fn retry_transfer(
&self,
keypair: &Keypair,
transaction: &mut Transaction,
tries: usize,
) -> TransportResult<Signature> {
self.send_and_confirm_transaction(&[keypair], transaction, tries, 0)
}
pub fn send_and_confirm_transaction<T: Signers>(
dispatch!(pub fn send_and_confirm_transaction<T: Signers>(
&self,
keypairs: &T,
transaction: &mut Transaction,
tries: usize,
pending_confirmations: usize,
) -> TransportResult<Signature> {
for x in 0..tries {
let now = Instant::now();
let mut num_confirmed = 0;
let mut wait_time = MAX_PROCESSING_AGE;
// resend the same transaction until the transaction has no chance of succeeding
let wire_transaction =
bincode::serialize(&transaction).expect("transaction serialization failed");
while now.elapsed().as_secs() < wait_time as u64 {
if num_confirmed == 0 {
let conn = self.connection_cache.get_connection(self.tpu_addr());
// Send the transaction if there has been no confirmation (e.g. the first time)
#[allow(clippy::needless_borrow)]
conn.send_data(&wire_transaction)?;
}
pending_confirmations: usize
) -> TransportResult<Signature>);
if let Ok(confirmed_blocks) = self.poll_for_signature_confirmation(
&transaction.signatures[0],
pending_confirmations,
) {
num_confirmed = confirmed_blocks;
if confirmed_blocks >= pending_confirmations {
return Ok(transaction.signatures[0]);
}
// Since network has seen the transaction, wait longer to receive
// all pending confirmations. Resending the transaction could result into
// extra transaction fees
wait_time = wait_time.max(
MAX_PROCESSING_AGE * pending_confirmations.saturating_sub(num_confirmed),
);
}
}
info!("{} tries failed transfer to {}", x, self.tpu_addr());
let blockhash = self.get_latest_blockhash()?;
transaction.sign(keypairs, blockhash);
}
Err(io::Error::new(
io::ErrorKind::Other,
format!("retry_transfer failed in {tries} retries"),
)
.into())
}
dispatch!(pub fn poll_get_balance(&self, pubkey: &Pubkey) -> TransportResult<u64>);
pub fn poll_get_balance(&self, pubkey: &Pubkey) -> TransportResult<u64> {
self.poll_get_balance_with_commitment(pubkey, CommitmentConfig::default())
}
pub fn poll_get_balance_with_commitment(
dispatch!(pub fn poll_get_balance_with_commitment(
&self,
pubkey: &Pubkey,
commitment_config: CommitmentConfig,
) -> TransportResult<u64> {
self.rpc_client()
.poll_get_balance_with_commitment(pubkey, commitment_config)
.map_err(|e| e.into())
}
commitment_config: CommitmentConfig
) -> TransportResult<u64>);
pub fn wait_for_balance(&self, pubkey: &Pubkey, expected_balance: Option<u64>) -> Option<u64> {
self.rpc_client().wait_for_balance_with_commitment(
pubkey,
expected_balance,
CommitmentConfig::default(),
)
}
dispatch!(pub fn wait_for_balance(&self, pubkey: &Pubkey, expected_balance: Option<u64>) -> Option<u64>);
pub fn get_program_accounts_with_config(
dispatch!(pub fn get_program_accounts_with_config(
&self,
pubkey: &Pubkey,
config: RpcProgramAccountsConfig,
) -> TransportResult<Vec<(Pubkey, Account)>> {
self.rpc_client()
.get_program_accounts_with_config(pubkey, config)
.map_err(|e| e.into())
}
config: RpcProgramAccountsConfig
) -> TransportResult<Vec<(Pubkey, Account)>>);
pub fn wait_for_balance_with_commitment(
dispatch!(pub fn wait_for_balance_with_commitment(
&self,
pubkey: &Pubkey,
expected_balance: Option<u64>,
commitment_config: CommitmentConfig,
) -> Option<u64> {
self.rpc_client().wait_for_balance_with_commitment(
pubkey,
expected_balance,
commitment_config,
)
}
commitment_config: CommitmentConfig
) -> Option<u64>);
pub fn poll_for_signature_with_commitment(
dispatch!(pub fn poll_for_signature_with_commitment(
&self,
signature: &Signature,
commitment_config: CommitmentConfig,
) -> TransportResult<()> {
self.rpc_client()
.poll_for_signature_with_commitment(signature, commitment_config)
.map_err(|e| e.into())
}
commitment_config: CommitmentConfig
) -> TransportResult<()>);
pub fn get_num_blocks_since_signature_confirmation(
dispatch!(pub fn get_num_blocks_since_signature_confirmation(
&mut self,
sig: &Signature,
) -> TransportResult<usize> {
self.rpc_client()
.get_num_blocks_since_signature_confirmation(sig)
.map_err(|e| e.into())
}
sig: &Signature
) -> TransportResult<usize>);
}
impl Client for ThinClient {
fn tpu_addr(&self) -> String {
self.tpu_addr().to_string()
}
dispatch!(fn tpu_addr(&self) -> String);
}
impl SyncClient for ThinClient {
fn send_and_confirm_message<T: Signers>(
dispatch!(fn send_and_confirm_message<T: Signers>(
&self,
keypairs: &T,
message: Message,
) -> TransportResult<Signature> {
let blockhash = self.get_latest_blockhash()?;
let mut transaction = Transaction::new(keypairs, message, blockhash);
let signature = self.send_and_confirm_transaction(keypairs, &mut transaction, 5, 0)?;
Ok(signature)
}
message: Message
) -> TransportResult<Signature>);
fn send_and_confirm_instruction(
dispatch!(fn send_and_confirm_instruction(
&self,
keypair: &Keypair,
instruction: Instruction,
) -> TransportResult<Signature> {
let message = Message::new(&[instruction], Some(&keypair.pubkey()));
self.send_and_confirm_message(&[keypair], message)
}
instruction: Instruction
) -> TransportResult<Signature>);
fn transfer_and_confirm(
dispatch!(fn transfer_and_confirm(
&self,
lamports: u64,
keypair: &Keypair,
pubkey: &Pubkey,
) -> TransportResult<Signature> {
let transfer_instruction =
system_instruction::transfer(&keypair.pubkey(), pubkey, lamports);
self.send_and_confirm_instruction(keypair, transfer_instruction)
}
pubkey: &Pubkey
) -> TransportResult<Signature>);
fn get_account_data(&self, pubkey: &Pubkey) -> TransportResult<Option<Vec<u8>>> {
Ok(self.rpc_client().get_account_data(pubkey).ok())
}
dispatch!(fn get_account_data(&self, pubkey: &Pubkey) -> TransportResult<Option<Vec<u8>>>);
fn get_account(&self, pubkey: &Pubkey) -> TransportResult<Option<Account>> {
let account = self.rpc_client().get_account(pubkey);
match account {
Ok(value) => Ok(Some(value)),
Err(_) => Ok(None),
}
}
dispatch!(fn get_account(&self, pubkey: &Pubkey) -> TransportResult<Option<Account>>);
fn get_account_with_commitment(
dispatch!(fn get_account_with_commitment(
&self,
pubkey: &Pubkey,
commitment_config: CommitmentConfig,
) -> TransportResult<Option<Account>> {
self.rpc_client()
.get_account_with_commitment(pubkey, commitment_config)
.map_err(|e| e.into())
.map(|r| r.value)
}
commitment_config: CommitmentConfig
) -> TransportResult<Option<Account>>);
fn get_balance(&self, pubkey: &Pubkey) -> TransportResult<u64> {
self.rpc_client().get_balance(pubkey).map_err(|e| e.into())
}
dispatch!(fn get_balance(&self, pubkey: &Pubkey) -> TransportResult<u64>);
fn get_balance_with_commitment(
dispatch!(fn get_balance_with_commitment(
&self,
pubkey: &Pubkey,
commitment_config: CommitmentConfig,
) -> TransportResult<u64> {
self.rpc_client()
.get_balance_with_commitment(pubkey, commitment_config)
.map_err(|e| e.into())
.map(|r| r.value)
}
commitment_config: CommitmentConfig
) -> TransportResult<u64>);
fn get_minimum_balance_for_rent_exemption(&self, data_len: usize) -> TransportResult<u64> {
self.rpc_client()
.get_minimum_balance_for_rent_exemption(data_len)
.map_err(|e| e.into())
}
dispatch!(fn get_minimum_balance_for_rent_exemption(&self, data_len: usize) -> TransportResult<u64>);
fn get_recent_blockhash(&self) -> TransportResult<(Hash, FeeCalculator)> {
#[allow(deprecated)]
let (blockhash, fee_calculator, _last_valid_slot) =
self.get_recent_blockhash_with_commitment(CommitmentConfig::default())?;
Ok((blockhash, fee_calculator))
}
dispatch_allow_deprecated!(fn get_recent_blockhash(&self) -> TransportResult<(Hash, FeeCalculator)>);
fn get_recent_blockhash_with_commitment(
dispatch_allow_deprecated!(fn get_recent_blockhash_with_commitment(
&self,
commitment_config: CommitmentConfig,
) -> TransportResult<(Hash, FeeCalculator, Slot)> {
let index = self.optimizer.experiment();
let now = Instant::now();
#[allow(deprecated)]
let recent_blockhash =
self.rpc_clients[index].get_recent_blockhash_with_commitment(commitment_config);
match recent_blockhash {
Ok(Response { value, .. }) => {
self.optimizer.report(index, duration_as_ms(&now.elapsed()));
Ok((value.0, value.1, value.2))
}
Err(e) => {
self.optimizer.report(index, std::u64::MAX);
Err(e.into())
}
}
}
commitment_config: CommitmentConfig
) -> TransportResult<(Hash, FeeCalculator, Slot)>);
fn get_fee_calculator_for_blockhash(
dispatch_allow_deprecated!(fn get_fee_calculator_for_blockhash(
&self,
blockhash: &Hash
) -> TransportResult<Option<FeeCalculator>>);
dispatch_allow_deprecated!(fn get_fee_rate_governor(&self) -> TransportResult<FeeRateGovernor>);
dispatch!(fn get_signature_status(
&self,
signature: &Signature
) -> TransportResult<Option<transaction::Result<()>>>);
dispatch!(fn get_signature_status_with_commitment(
&self,
signature: &Signature,
commitment_config: CommitmentConfig
) -> TransportResult<Option<transaction::Result<()>>>);
dispatch!(fn get_slot(&self) -> TransportResult<u64>);
dispatch!(fn get_slot_with_commitment(
&self,
commitment_config: CommitmentConfig
) -> TransportResult<u64>);
dispatch!(fn get_epoch_info(&self) -> TransportResult<EpochInfo>);
dispatch!(fn get_transaction_count(&self) -> TransportResult<u64>);
dispatch!(fn get_transaction_count_with_commitment(
&self,
commitment_config: CommitmentConfig
) -> TransportResult<u64>);
dispatch!(fn poll_for_signature_confirmation(
&self,
signature: &Signature,
min_confirmed_blocks: usize
) -> TransportResult<usize>);
dispatch!(fn poll_for_signature(&self, signature: &Signature) -> TransportResult<()>);
dispatch_allow_deprecated!(fn get_new_blockhash(&self, blockhash: &Hash) -> TransportResult<(Hash, FeeCalculator)>);
dispatch!(fn get_latest_blockhash(&self) -> TransportResult<Hash>);
dispatch!(fn get_latest_blockhash_with_commitment(
&self,
commitment_config: CommitmentConfig
) -> TransportResult<(Hash, u64)>);
dispatch!(fn is_blockhash_valid(
&self,
blockhash: &Hash,
) -> TransportResult<Option<FeeCalculator>> {
#[allow(deprecated)]
self.rpc_client()
.get_fee_calculator_for_blockhash(blockhash)
.map_err(|e| e.into())
}
commitment_config: CommitmentConfig
) -> TransportResult<bool>);
fn get_fee_rate_governor(&self) -> TransportResult<FeeRateGovernor> {
#[allow(deprecated)]
self.rpc_client()
.get_fee_rate_governor()
.map_err(|e| e.into())
.map(|r| r.value)
}
fn get_signature_status(
&self,
signature: &Signature,
) -> TransportResult<Option<transaction::Result<()>>> {
let status = self
.rpc_client()
.get_signature_status(signature)
.map_err(|err| {
io::Error::new(
io::ErrorKind::Other,
format!("send_transaction failed with error {err:?}"),
)
})?;
Ok(status)
}
fn get_signature_status_with_commitment(
&self,
signature: &Signature,
commitment_config: CommitmentConfig,
) -> TransportResult<Option<transaction::Result<()>>> {
let status = self
.rpc_client()
.get_signature_status_with_commitment(signature, commitment_config)
.map_err(|err| {
io::Error::new(
io::ErrorKind::Other,
format!("send_transaction failed with error {err:?}"),
)
})?;
Ok(status)
}
fn get_slot(&self) -> TransportResult<u64> {
self.get_slot_with_commitment(CommitmentConfig::default())
}
fn get_slot_with_commitment(
&self,
commitment_config: CommitmentConfig,
) -> TransportResult<u64> {
let slot = self
.rpc_client()
.get_slot_with_commitment(commitment_config)
.map_err(|err| {
io::Error::new(
io::ErrorKind::Other,
format!("send_transaction failed with error {err:?}"),
)
})?;
Ok(slot)
}
fn get_epoch_info(&self) -> TransportResult<EpochInfo> {
self.rpc_client().get_epoch_info().map_err(|e| e.into())
}
fn get_transaction_count(&self) -> TransportResult<u64> {
let index = self.optimizer.experiment();
let now = Instant::now();
match self.rpc_client().get_transaction_count() {
Ok(transaction_count) => {
self.optimizer.report(index, duration_as_ms(&now.elapsed()));
Ok(transaction_count)
}
Err(e) => {
self.optimizer.report(index, std::u64::MAX);
Err(e.into())
}
}
}
fn get_transaction_count_with_commitment(
&self,
commitment_config: CommitmentConfig,
) -> TransportResult<u64> {
let index = self.optimizer.experiment();
let now = Instant::now();
match self
.rpc_client()
.get_transaction_count_with_commitment(commitment_config)
{
Ok(transaction_count) => {
self.optimizer.report(index, duration_as_ms(&now.elapsed()));
Ok(transaction_count)
}
Err(e) => {
self.optimizer.report(index, std::u64::MAX);
Err(e.into())
}
}
}
/// Poll the server until the signature has been confirmed by at least `min_confirmed_blocks`
fn poll_for_signature_confirmation(
&self,
signature: &Signature,
min_confirmed_blocks: usize,
) -> TransportResult<usize> {
self.rpc_client()
.poll_for_signature_confirmation(signature, min_confirmed_blocks)
.map_err(|e| e.into())
}
fn poll_for_signature(&self, signature: &Signature) -> TransportResult<()> {
self.rpc_client()
.poll_for_signature(signature)
.map_err(|e| e.into())
}
fn get_new_blockhash(&self, blockhash: &Hash) -> TransportResult<(Hash, FeeCalculator)> {
#[allow(deprecated)]
self.rpc_client()
.get_new_blockhash(blockhash)
.map_err(|e| e.into())
}
fn get_latest_blockhash(&self) -> TransportResult<Hash> {
let (blockhash, _) =
self.get_latest_blockhash_with_commitment(CommitmentConfig::default())?;
Ok(blockhash)
}
fn get_latest_blockhash_with_commitment(
&self,
commitment_config: CommitmentConfig,
) -> TransportResult<(Hash, u64)> {
let index = self.optimizer.experiment();
let now = Instant::now();
match self.rpc_clients[index].get_latest_blockhash_with_commitment(commitment_config) {
Ok((blockhash, last_valid_block_height)) => {
self.optimizer.report(index, duration_as_ms(&now.elapsed()));
Ok((blockhash, last_valid_block_height))
}
Err(e) => {
self.optimizer.report(index, std::u64::MAX);
Err(e.into())
}
}
}
fn is_blockhash_valid(
&self,
blockhash: &Hash,
commitment_config: CommitmentConfig,
) -> TransportResult<bool> {
self.rpc_client()
.is_blockhash_valid(blockhash, commitment_config)
.map_err(|e| e.into())
}
fn get_fee_for_message(&self, message: &Message) -> TransportResult<u64> {
self.rpc_client()
.get_fee_for_message(message)
.map_err(|e| e.into())
}
dispatch!(fn get_fee_for_message(&self, message: &Message) -> TransportResult<u64>);
}
impl AsyncClient for ThinClient {
fn async_send_versioned_transaction(
dispatch!(fn async_send_versioned_transaction(
&self,
transaction: VersionedTransaction,
) -> TransportResult<Signature> {
let conn = self.connection_cache.get_connection(self.tpu_addr());
let wire_transaction =
bincode::serialize(&transaction).expect("serialize Transaction in send_batch");
conn.send_data(&wire_transaction)?;
Ok(transaction.signatures[0])
}
transaction: VersionedTransaction
) -> TransportResult<Signature>);
fn async_send_versioned_transaction_batch(
dispatch!(fn async_send_versioned_transaction_batch(
&self,
batch: Vec<VersionedTransaction>,
) -> TransportResult<()> {
let conn = self.connection_cache.get_connection(self.tpu_addr());
let buffers = batch
.into_par_iter()
.map(|tx| bincode::serialize(&tx).expect("serialize Transaction in send_batch"))
.collect::<Vec<_>>();
conn.send_data_batch(&buffers)?;
Ok(())
}
batch: Vec<VersionedTransaction>
) -> TransportResult<()>);
}

2
client/src/tpu_client.rs
View File

@ -77,7 +77,7 @@ impl TpuClient<QuicPool, QuicConnectionManager, QuicConfig> {
config: TpuClientConfig,
) -> Result<Self> {
let connection_cache = match ConnectionCache::default() {
ConnectionCache::Quic(cache) => Arc::new(cache),
ConnectionCache::Quic(cache) => cache,
ConnectionCache::Udp(_) => {
return Err(TpuSenderError::Custom(String::from(
"Invalid default connection cache",

4
rpc-test/tests/rpc.rs
View File

@ -472,7 +472,7 @@ fn run_tpu_send_transaction(tpu_use_quic: bool) {
rpc_client.clone(),
&test_validator.rpc_pubsub_url(),
TpuClientConfig::default(),
Arc::new(cache),
cache,
)
.unwrap()
.send_transaction(&tx),
@ -480,7 +480,7 @@ fn run_tpu_send_transaction(tpu_use_quic: bool) {
rpc_client.clone(),
&test_validator.rpc_pubsub_url(),
TpuClientConfig::default(),
Arc::new(cache),
cache,
)
.unwrap()
.send_transaction(&tx),