Skip to content

Commit

Permalink
remove duplicates from fanout leaders (#5109)
Browse files Browse the repository at this point in the history
Currently, when sending transactions to future leaders, duplicate leaders may appear in consecutive slots, resulting in unnecessary spam.
This PRs removes duplicates from the fanout leaders set.
  • Loading branch information
KirillLykov authored Mar 4, 2025
1 parent 561023d commit 62acd94
Showing 1 changed file with 25 additions and 24 deletions.
49 changes: 25 additions & 24 deletions tpu-client-next/src/connection_workers_scheduler.rs
Original file line number Diff line number Diff line change
Expand Up @@ -217,29 +217,28 @@ impl ConnectionWorkersScheduler {
}
};

let updated_leaders = leader_updater.next_leaders(leaders_fanout.connect);
let connect_leaders = leader_updater.next_leaders(leaders_fanout.connect);
let send_leaders = extract_send_leaders(&connect_leaders, leaders_fanout.send);

let (fanout_leaders, connect_leaders) =
split_leaders(&updated_leaders, &leaders_fanout);
// add future leaders to the cache to hide the latency of opening
// the connection.
for peer in connect_leaders {
if !workers.contains(peer) {
if !workers.contains(&peer) {
let stats = send_stats_per_addr.entry(peer.ip()).or_default();
let worker = Self::spawn_worker(
&endpoint,
peer,
&peer,
worker_channel_size,
skip_check_transaction_age,
max_reconnect_attempts,
stats.clone(),
);
maybe_shutdown_worker(workers.push(*peer, worker));
maybe_shutdown_worker(workers.push(peer, worker));
}
}

if let Err(error) =
Broadcaster::send_to_workers(&mut workers, fanout_leaders, transaction_batch).await
Broadcaster::send_to_workers(&mut workers, &send_leaders, transaction_batch).await
{
last_error = Some(error);
break;
Expand Down Expand Up @@ -338,21 +337,23 @@ impl WorkersBroadcaster for NonblockingBroadcaster {
}
}

/// Splits `leaders` into two slices based on the `fanout` configuration:
/// * the first slice contains the leaders to which transactions will be sent,
/// * the second vector contains the leaders, used to warm up connections. This
/// slice includes the first set.
fn split_leaders<'leaders>(
leaders: &'leaders [SocketAddr],
fanout: &Fanout,
) -> (&'leaders [SocketAddr], &'leaders [SocketAddr]) {
let Fanout { send, connect } = fanout;
assert!(send <= connect);
let send_count = (*send).min(leaders.len());
let connect_count = (*connect).min(leaders.len());

let send_slice = &leaders[..send_count];
let connect_slice = &leaders[..connect_count];

(send_slice, connect_slice)
/// Extracts a list of unique leader addresses to which transactions will be sent.
///
/// This function selects up to `send_fanout` addresses from the `leaders` list, ensuring that
/// only unique addresses are included while maintaining their original order.
fn extract_send_leaders(leaders: &[SocketAddr], send_fanout: usize) -> Vec<SocketAddr> {
let send_count = send_fanout.min(leaders.len());
remove_duplicates(&leaders[..send_count])
}

/// Removes duplicate `SocketAddr` elements from the given slice while
/// preserving their original order.
fn remove_duplicates(input: &[SocketAddr]) -> Vec<SocketAddr> {
let mut res = Vec::with_capacity(input.len());
for address in input {
if !res.contains(address) {
res.push(*address);
}
}
res
}

0 comments on commit 62acd94

Please sign in to comment.