Skip to content

Commit

Permalink
v1.17: Make the quic server connection table use an async lock, reduc…
Browse files Browse the repository at this point in the history
…ing thrashing (backport of solana-labs#293) (solana-labs#299)

Make the quic server connection table use an async lock
  • Loading branch information
mergify[bot] authored Mar 18, 2024
1 parent af153cb commit 29c004c
Showing 1 changed file with 26 additions and 8 deletions.
34 changes: 26 additions & 8 deletions streamer/src/nonblocking/quic.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,13 +30,26 @@ use {
std::{
iter::repeat_with,
net::{IpAddr, SocketAddr, UdpSocket},
// CAUTION: be careful not to introduce any awaits while holding an RwLock.
sync::{
atomic::{AtomicBool, AtomicU64, Ordering},
Arc, Mutex, MutexGuard, RwLock,
Arc, RwLock,
},
time::{Duration, Instant},
},
tokio::{task::JoinHandle, time::timeout},
tokio::{
// CAUTION: It's kind of sketch that we're mixing async and sync locks (see the RwLock above).
// This is done so that sync code can also access the stake table.
// Make sure we don't hold a sync lock across an await - including the await to
// lock an async Mutex. This does not happen now and should not happen as long as we
// don't hold an async Mutex and sync RwLock at the same time (currently true)
// but if we do, the scope of the RwLock must always be a subset of the async Mutex
// (i.e. lock order is always async Mutex -> RwLock). Also, be careful not to
// introduce any other awaits while holding the RwLock.
sync::{Mutex, MutexGuard},
task::JoinHandle,
time::timeout,
},
};

/// Limit to 500K PPS
Expand Down Expand Up @@ -383,7 +396,7 @@ fn handle_and_cache_new_connection(
}
}

fn prune_unstaked_connections_and_add_new_connection(
async fn prune_unstaked_connections_and_add_new_connection(
connection: Connection,
connection_table: Arc<Mutex<ConnectionTable>>,
max_connections: usize,
Expand All @@ -393,7 +406,7 @@ fn prune_unstaked_connections_and_add_new_connection(
let stats = params.stats.clone();
if max_connections > 0 {
let connection_table_clone = connection_table.clone();
let mut connection_table = connection_table.lock().unwrap();
let mut connection_table = connection_table.lock().await;
prune_unstaked_connection_table(&mut connection_table, max_connections, stats);
handle_and_cache_new_connection(
connection,
Expand Down Expand Up @@ -496,7 +509,8 @@ async fn setup_connection(
);

if params.stake > 0 {
let mut connection_table_l = staked_connection_table.lock().unwrap();
let mut connection_table_l = staked_connection_table.lock().await;

if connection_table_l.total_size >= max_staked_connections {
let num_pruned =
connection_table_l.prune_random(PRUNE_RANDOM_SAMPLE_SIZE, params.stake);
Expand Down Expand Up @@ -525,7 +539,9 @@ async fn setup_connection(
max_unstaked_connections,
&params,
wait_for_chunk_timeout,
) {
)
.await
{
stats
.connection_added_from_staked_peer
.fetch_add(1, Ordering::Relaxed);
Expand All @@ -544,7 +560,9 @@ async fn setup_connection(
max_unstaked_connections,
&params,
wait_for_chunk_timeout,
) {
)
.await
{
stats
.connection_added_from_unstaked_peer
.fetch_add(1, Ordering::Relaxed);
Expand Down Expand Up @@ -807,7 +825,7 @@ async fn handle_connection(
}
}

let removed_connection_count = connection_table.lock().unwrap().remove_connection(
let removed_connection_count = connection_table.lock().await.remove_connection(
ConnectionTableKey::new(remote_addr.ip(), params.remote_pubkey),
remote_addr.port(),
stable_id,
Expand Down

0 comments on commit 29c004c

Please sign in to comment.