Skip to content

Commit

Permalink
v1.18: client: Start resending sooner during `send_and_confirm_transa…
Browse files Browse the repository at this point in the history
…ctions_in_parallel` (backport of #348) (#357)

client: Start resending sooner during `send_and_confirm_transactions_in_parallel` (#348)

client: Confirm sooner during send_and_confirm_in_parallel
(cherry picked from commit b2f4fb3)

Co-authored-by: Jon C <[email protected]>
  • Loading branch information
mergify[bot] and joncinque authored Mar 22, 2024
1 parent 7606adf commit 12b8083
Showing 1 changed file with 57 additions and 51 deletions.
108 changes: 57 additions & 51 deletions client/src/send_and_confirm_transactions_in_parallel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ use {
},
bincode::serialize,
dashmap::DashMap,
futures_util::future::{join_all, TryFutureExt},
futures_util::future::{join_all, FutureExt},
solana_quic_client::{QuicConfig, QuicConnectionManager, QuicPool},
solana_rpc_client::spinner::{self, SendTransactionProgress},
solana_rpc_client_api::{
Expand Down Expand Up @@ -188,9 +188,7 @@ async fn send_transaction_with_rpc_fallback(
serialized_transaction: Vec<u8>,
context: &SendingContext,
index: usize,
counter: usize,
) -> Result<()> {
tokio::time::sleep(SEND_INTERVAL.saturating_mul(counter as u32)).await;
let send_over_rpc = if let Some(tpu_client) = tpu_client {
!tpu_client
.send_wire_transaction(serialized_transaction.clone())
Expand Down Expand Up @@ -261,44 +259,42 @@ async fn sign_all_messages_and_send<T: Signers + ?Sized>(
.expect("Transaction should be signable");
let serialized_transaction = serialize(&transaction).expect("Transaction should serialize");
let signature = transaction.signatures[0];
futures.push(
futures.push(async move {
tokio::time::sleep(SEND_INTERVAL.saturating_mul(counter as u32)).await;
// send to confirm the transaction
context.unconfirmed_transaction_map.insert(
signature,
TransactionData {
index: *index,
serialized_transaction: serialized_transaction.clone(),
last_valid_block_height: blockhashdata.last_valid_block_height,
message: message.clone(),
},
);
if let Some(progress_bar) = progress_bar {
let progress = progress_from_context_and_block_height(
context,
blockhashdata.last_valid_block_height,
);
progress.set_message_for_confirmed_transactions(
progress_bar,
&format!(
"Sending {}/{} transactions",
counter + 1,
current_transaction_count,
),
);
}
send_transaction_with_rpc_fallback(
rpc_client,
tpu_client,
transaction,
serialized_transaction.clone(),
serialized_transaction,
context,
*index,
counter,
)
.and_then(move |_| async move {
// send to confirm the transaction
context.unconfirmed_transaction_map.insert(
signature,
TransactionData {
index: *index,
serialized_transaction,
last_valid_block_height: blockhashdata.last_valid_block_height,
message: message.clone(),
},
);
if let Some(progress_bar) = progress_bar {
let progress = progress_from_context_and_block_height(
context,
blockhashdata.last_valid_block_height,
);
progress.set_message_for_confirmed_transactions(
progress_bar,
&format!(
"Sending {}/{} transactions",
counter + 1,
current_transaction_count,
),
);
}
Ok(())
}),
);
.await
});
}
// collect to convert Vec<Result<_>> to Result<Vec<_>>
join_all(futures).await.into_iter().collect::<Result<_>>()?;
Expand Down Expand Up @@ -477,23 +473,33 @@ pub async fn send_and_confirm_transactions_in_parallel<T: Signers + ?Sized>(
// clear the map so that we can start resending
unconfirmed_transasction_map.clear();

sign_all_messages_and_send(
&progress_bar,
&rpc_client,
&tpu_client,
messages_with_index,
signers,
&context,
)
.await?;

// wait until all the transactions are confirmed or expired
confirm_transactions_till_block_height_and_resend_unexpired_transaction_over_tpu(
&progress_bar,
&tpu_client,
&context,
)
.await;
let futures = [
sign_all_messages_and_send(
&progress_bar,
&rpc_client,
&tpu_client,
messages_with_index,
signers,
&context,
)
.boxed_local(),
async {
// Give the signing and sending a head start before trying to
// confirm and resend
tokio::time::sleep(TPU_RESEND_REFRESH_RATE).await;
confirm_transactions_till_block_height_and_resend_unexpired_transaction_over_tpu(
&progress_bar,
&tpu_client,
&context,
)
.await;
// Infallible, but required to have the same return type as
// `sign_all_messages_and_send`
Ok(())
}
.boxed_local(),
];
join_all(futures).await.into_iter().collect::<Result<_>>()?;

if unconfirmed_transasction_map.is_empty() {
break;
Expand Down

0 comments on commit 12b8083

Please sign in to comment.