Skip to content
This repository was archived by the owner on Jan 22, 2025. It is now read-only.

Commit f51d29f

Browse files
mergify[bot]lijunwangs
authored and
yihau
committed
v1.17: Parameterize max streams per ms (backport of #707) (#1105)
* Parameterize max streams per ms (#707) Make PPS a parameter instead of the hard coded Co-authored-by: Lijun Wang <[email protected]>
1 parent 808b2f5 commit f51d29f

File tree

5 files changed

+62
-17
lines changed

5 files changed

+62
-17
lines changed

‎client/src/connection_cache.rs

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -217,7 +217,8 @@ mod tests {
217217
crossbeam_channel::unbounded,
218218
solana_sdk::{net::DEFAULT_TPU_COALESCE, signature::Keypair},
219219
solana_streamer::{
220-
nonblocking::quic::DEFAULT_WAIT_FOR_CHUNK_TIMEOUT, streamer::StakedNodes,
220+
nonblocking::quic::{DEFAULT_MAX_STREAMS_PER_MS, DEFAULT_WAIT_FOR_CHUNK_TIMEOUT},
221+
streamer::StakedNodes,
221222
},
222223
std::{
223224
net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket},
@@ -256,6 +257,7 @@ mod tests {
256257
staked_nodes,
257258
10,
258259
10,
260+
DEFAULT_MAX_STREAMS_PER_MS,
259261
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
260262
DEFAULT_TPU_COALESCE,
261263
)

‎core/src/tpu.rs

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ use {
3333
solana_runtime::{bank_forks::BankForks, prioritization_fee_cache::PrioritizationFeeCache},
3434
solana_sdk::{clock::Slot, pubkey::Pubkey, signature::Keypair},
3535
solana_streamer::{
36-
nonblocking::quic::DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
36+
nonblocking::quic::{DEFAULT_MAX_STREAMS_PER_MS, DEFAULT_WAIT_FOR_CHUNK_TIMEOUT},
3737
quic::{spawn_server, MAX_STAKED_CONNECTIONS, MAX_UNSTAKED_CONNECTIONS},
3838
streamer::StakedNodes,
3939
},
@@ -163,6 +163,7 @@ impl Tpu {
163163
staked_nodes.clone(),
164164
MAX_STAKED_CONNECTIONS,
165165
MAX_UNSTAKED_CONNECTIONS,
166+
DEFAULT_MAX_STREAMS_PER_MS,
166167
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
167168
tpu_coalesce,
168169
)
@@ -183,6 +184,7 @@ impl Tpu {
183184
staked_nodes.clone(),
184185
MAX_STAKED_CONNECTIONS.saturating_add(MAX_UNSTAKED_CONNECTIONS),
185186
0, // Prevent unstaked nodes from forwarding transactions
187+
DEFAULT_MAX_STREAMS_PER_MS,
186188
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
187189
tpu_coalesce,
188190
)

‎quic-client/tests/quic_client.rs

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,8 @@ mod tests {
1010
},
1111
solana_sdk::{net::DEFAULT_TPU_COALESCE, packet::PACKET_DATA_SIZE, signature::Keypair},
1212
solana_streamer::{
13-
nonblocking::quic::DEFAULT_WAIT_FOR_CHUNK_TIMEOUT, streamer::StakedNodes,
13+
nonblocking::quic::{DEFAULT_MAX_STREAMS_PER_MS, DEFAULT_WAIT_FOR_CHUNK_TIMEOUT},
14+
streamer::StakedNodes,
1415
tls_certificates::new_self_signed_tls_certificate,
1516
},
1617
std::{
@@ -79,6 +80,7 @@ mod tests {
7980
staked_nodes,
8081
10,
8182
10,
83+
DEFAULT_MAX_STREAMS_PER_MS,
8284
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
8385
DEFAULT_TPU_COALESCE,
8486
)
@@ -159,6 +161,7 @@ mod tests {
159161
staked_nodes,
160162
10,
161163
10,
164+
DEFAULT_MAX_STREAMS_PER_MS,
162165
Duration::from_secs(1), // wait_for_chunk_timeout
163166
DEFAULT_TPU_COALESCE,
164167
)
@@ -217,6 +220,7 @@ mod tests {
217220
staked_nodes.clone(),
218221
10,
219222
10,
223+
DEFAULT_MAX_STREAMS_PER_MS,
220224
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
221225
DEFAULT_TPU_COALESCE,
222226
)
@@ -241,6 +245,7 @@ mod tests {
241245
staked_nodes,
242246
10,
243247
10,
248+
DEFAULT_MAX_STREAMS_PER_MS,
244249
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
245250
DEFAULT_TPU_COALESCE,
246251
)

‎streamer/src/nonblocking/quic.rs

Lines changed: 42 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -53,9 +53,8 @@ use {
5353
},
5454
};
5555

56-
/// Limit to 500K PPS
57-
const MAX_STREAMS_PER_100MS: u64 = 500_000 / 10;
5856
const MAX_UNSTAKED_STREAMS_PERCENT: u64 = 20;
57+
const STREAM_THROTTLING_INTERVAL_MS: u64 = 100;
5958
const STREAM_THROTTLING_INTERVAL: Duration = Duration::from_millis(100);
6059
const WAIT_FOR_STREAM_TIMEOUT: Duration = Duration::from_millis(100);
6160
pub const DEFAULT_WAIT_FOR_CHUNK_TIMEOUT: Duration = Duration::from_secs(10);
@@ -75,6 +74,9 @@ const CONNECTION_CLOSE_CODE_TOO_MANY: u32 = 4;
7574
const CONNECTION_CLOSE_REASON_TOO_MANY: &[u8] = b"too_many";
7675
const STREAM_STOP_CODE_THROTTLING: u32 = 15;
7776

77+
/// Limit to 500K PPS
78+
pub const DEFAULT_MAX_STREAMS_PER_MS: u64 = 500;
79+
7880
// A sequence of bytes that is part of a packet
7981
// along with where in the packet it is
8082
struct PacketChunk {
@@ -111,6 +113,7 @@ pub fn spawn_server(
111113
staked_nodes: Arc<RwLock<StakedNodes>>,
112114
max_staked_connections: usize,
113115
max_unstaked_connections: usize,
116+
max_streams_per_ms: u64,
114117
wait_for_chunk_timeout: Duration,
115118
coalesce: Duration,
116119
) -> Result<(Endpoint, Arc<StreamStats>, JoinHandle<()>), QuicServerError> {
@@ -136,6 +139,7 @@ pub fn spawn_server(
136139
staked_nodes,
137140
max_staked_connections,
138141
max_unstaked_connections,
142+
max_streams_per_ms,
139143
stats.clone(),
140144
wait_for_chunk_timeout,
141145
coalesce,
@@ -153,6 +157,7 @@ async fn run_server(
153157
staked_nodes: Arc<RwLock<StakedNodes>>,
154158
max_staked_connections: usize,
155159
max_unstaked_connections: usize,
160+
max_streams_per_ms: u64,
156161
stats: Arc<StreamStats>,
157162
wait_for_chunk_timeout: Duration,
158163
coalesce: Duration,
@@ -192,6 +197,7 @@ async fn run_server(
192197
staked_nodes.clone(),
193198
max_staked_connections,
194199
max_unstaked_connections,
200+
max_streams_per_ms,
195201
stats.clone(),
196202
wait_for_chunk_timeout,
197203
));
@@ -328,6 +334,7 @@ fn handle_and_cache_new_connection(
328334
params: &NewConnectionHandlerParams,
329335
wait_for_chunk_timeout: Duration,
330336
max_unstaked_connections: usize,
337+
max_streams_per_ms: u64,
331338
) -> Result<(), ConnectionHandlerError> {
332339
if let Ok(max_uni_streams) = VarInt::from_u64(compute_max_allowed_uni_streams(
333340
connection_table_l.peer_type,
@@ -379,6 +386,7 @@ fn handle_and_cache_new_connection(
379386
peer_type,
380387
wait_for_chunk_timeout,
381388
max_unstaked_connections,
389+
max_streams_per_ms,
382390
));
383391
Ok(())
384392
} else {
@@ -407,6 +415,7 @@ async fn prune_unstaked_connections_and_add_new_connection(
407415
max_connections: usize,
408416
params: &NewConnectionHandlerParams,
409417
wait_for_chunk_timeout: Duration,
418+
max_streams_per_ms: u64,
410419
) -> Result<(), ConnectionHandlerError> {
411420
let stats = params.stats.clone();
412421
if max_connections > 0 {
@@ -420,6 +429,7 @@ async fn prune_unstaked_connections_and_add_new_connection(
420429
params,
421430
wait_for_chunk_timeout,
422431
max_connections,
432+
max_streams_per_ms,
423433
)
424434
} else {
425435
connection.close(
@@ -484,6 +494,7 @@ async fn setup_connection(
484494
staked_nodes: Arc<RwLock<StakedNodes>>,
485495
max_staked_connections: usize,
486496
max_unstaked_connections: usize,
497+
max_streams_per_ms: u64,
487498
stats: Arc<StreamStats>,
488499
wait_for_chunk_timeout: Duration,
489500
) {
@@ -502,8 +513,9 @@ async fn setup_connection(
502513
),
503514
|(pubkey, stake, total_stake, max_stake, min_stake)| {
504515
// The heuristic is that the stake should be large engouh to have 1 stream pass throuh within one throttle
505-
// interval during which we allow max MAX_STREAMS_PER_100MS streams.
506-
let min_stake_ratio = 1_f64 / MAX_STREAMS_PER_100MS as f64;
516+
// interval during which we allow max (MAX_STREAMS_PER_MS * STREAM_THROTTLING_INTERVAL_MS) streams.
517+
let min_stake_ratio =
518+
1_f64 / (max_streams_per_ms * STREAM_THROTTLING_INTERVAL_MS) as f64;
507519
let stake_ratio = stake as f64 / total_stake as f64;
508520
let stake = if stake_ratio < min_stake_ratio {
509521
// If it is a staked connection with ultra low stake ratio, treat it as unstaked.
@@ -541,6 +553,7 @@ async fn setup_connection(
541553
&params,
542554
wait_for_chunk_timeout,
543555
max_unstaked_connections,
556+
max_streams_per_ms,
544557
) {
545558
stats
546559
.connection_added_from_staked_peer
@@ -556,6 +569,7 @@ async fn setup_connection(
556569
max_unstaked_connections,
557570
&params,
558571
wait_for_chunk_timeout,
572+
max_streams_per_ms,
559573
)
560574
.await
561575
{
@@ -577,6 +591,7 @@ async fn setup_connection(
577591
max_unstaked_connections,
578592
&params,
579593
wait_for_chunk_timeout,
594+
max_streams_per_ms,
580595
)
581596
.await
582597
{
@@ -725,12 +740,14 @@ fn max_streams_for_connection_in_100ms(
725740
stake: u64,
726741
total_stake: u64,
727742
max_unstaked_connections: usize,
743+
max_streams_per_ms: u64,
728744
) -> u64 {
745+
let max_streams_per_interval = max_streams_per_ms.saturating_mul(STREAM_THROTTLING_INTERVAL_MS);
729746
let max_unstaked_streams_per_100ms = if max_unstaked_connections == 0 {
730747
0
731748
} else {
732749
Percentage::from(MAX_UNSTAKED_STREAMS_PERCENT)
733-
.apply_to(MAX_STREAMS_PER_100MS)
750+
.apply_to(max_streams_per_interval)
734751
.saturating_div(MAX_UNSTAKED_CONNECTIONS as u64)
735752
};
736753

@@ -744,8 +761,8 @@ fn max_streams_for_connection_in_100ms(
744761
if matches!(connection_type, ConnectionPeerType::Unstaked) || stake == 0 {
745762
max_unstaked_streams_per_100ms
746763
} else {
747-
let max_total_staked_streams: u64 = MAX_STREAMS_PER_100MS
748-
- Percentage::from(MAX_UNSTAKED_STREAMS_PERCENT).apply_to(MAX_STREAMS_PER_100MS);
764+
let max_total_staked_streams: u64 = max_streams_per_interval
765+
- Percentage::from(MAX_UNSTAKED_STREAMS_PERCENT).apply_to(max_streams_per_interval);
749766
std::cmp::max(
750767
min_staked_streams_per_100ms,
751768
((max_total_staked_streams as f64 / total_stake as f64) * stake as f64) as u64,
@@ -762,6 +779,7 @@ fn reset_throttling_params_if_needed(last_instant: &mut tokio::time::Instant) ->
762779
}
763780
}
764781

782+
#[allow(clippy::too_many_arguments)]
765783
async fn handle_connection(
766784
connection: Connection,
767785
remote_addr: SocketAddr,
@@ -772,6 +790,7 @@ async fn handle_connection(
772790
peer_type: ConnectionPeerType,
773791
wait_for_chunk_timeout: Duration,
774792
max_unstaked_connections: usize,
793+
max_streams_per_ms: u64,
775794
) {
776795
let stats = params.stats;
777796
debug!(
@@ -787,6 +806,7 @@ async fn handle_connection(
787806
params.stake,
788807
params.total_stake,
789808
max_unstaked_connections,
809+
max_streams_per_ms,
790810
);
791811
let mut last_throttling_instant = tokio::time::Instant::now();
792812
let mut streams_in_current_interval = 0;
@@ -1307,6 +1327,7 @@ pub mod test {
13071327
staked_nodes,
13081328
MAX_STAKED_CONNECTIONS,
13091329
MAX_UNSTAKED_CONNECTIONS,
1330+
DEFAULT_MAX_STREAMS_PER_MS,
13101331
Duration::from_secs(2),
13111332
DEFAULT_TPU_COALESCE,
13121333
)
@@ -1743,6 +1764,7 @@ pub mod test {
17431764
staked_nodes,
17441765
MAX_STAKED_CONNECTIONS,
17451766
0, // Do not allow any connection from unstaked clients/nodes
1767+
DEFAULT_MAX_STREAMS_PER_MS,
17461768
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
17471769
DEFAULT_TPU_COALESCE,
17481770
)
@@ -1774,6 +1796,7 @@ pub mod test {
17741796
staked_nodes,
17751797
MAX_STAKED_CONNECTIONS,
17761798
MAX_UNSTAKED_CONNECTIONS,
1799+
DEFAULT_MAX_STREAMS_PER_MS,
17771800
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
17781801
DEFAULT_TPU_COALESCE,
17791802
)
@@ -2120,7 +2143,8 @@ pub mod test {
21202143
ConnectionPeerType::Unstaked,
21212144
0,
21222145
10000,
2123-
MAX_UNSTAKED_CONNECTIONS
2146+
MAX_UNSTAKED_CONNECTIONS,
2147+
DEFAULT_MAX_STREAMS_PER_MS,
21242148
),
21252149
20
21262150
);
@@ -2131,7 +2155,8 @@ pub mod test {
21312155
ConnectionPeerType::Unstaked,
21322156
10,
21332157
10000,
2134-
MAX_UNSTAKED_CONNECTIONS
2158+
MAX_UNSTAKED_CONNECTIONS,
2159+
DEFAULT_MAX_STREAMS_PER_MS,
21352160
),
21362161
20
21372162
);
@@ -2143,7 +2168,8 @@ pub mod test {
21432168
ConnectionPeerType::Staked,
21442169
0,
21452170
10000,
2146-
MAX_UNSTAKED_CONNECTIONS
2171+
MAX_UNSTAKED_CONNECTIONS,
2172+
DEFAULT_MAX_STREAMS_PER_MS,
21472173
),
21482174
20
21492175
);
@@ -2155,7 +2181,8 @@ pub mod test {
21552181
ConnectionPeerType::Staked,
21562182
15,
21572183
10000,
2158-
MAX_UNSTAKED_CONNECTIONS
2184+
MAX_UNSTAKED_CONNECTIONS,
2185+
DEFAULT_MAX_STREAMS_PER_MS
21592186
),
21602187
60
21612188
);
@@ -2167,7 +2194,8 @@ pub mod test {
21672194
ConnectionPeerType::Staked,
21682195
1000,
21692196
10000,
2170-
MAX_UNSTAKED_CONNECTIONS
2197+
MAX_UNSTAKED_CONNECTIONS,
2198+
DEFAULT_MAX_STREAMS_PER_MS
21712199
),
21722200
4000
21732201
);
@@ -2179,7 +2207,8 @@ pub mod test {
21792207
ConnectionPeerType::Staked,
21802208
1,
21812209
50000,
2182-
MAX_UNSTAKED_CONNECTIONS
2210+
MAX_UNSTAKED_CONNECTIONS,
2211+
DEFAULT_MAX_STREAMS_PER_MS
21832212
),
21842213
21
21852214
);

‎streamer/src/quic.rs

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -432,6 +432,7 @@ pub fn spawn_server(
432432
staked_nodes: Arc<RwLock<StakedNodes>>,
433433
max_staked_connections: usize,
434434
max_unstaked_connections: usize,
435+
max_streams_per_ms: u64,
435436
wait_for_chunk_timeout: Duration,
436437
coalesce: Duration,
437438
) -> Result<(Endpoint, thread::JoinHandle<()>), QuicServerError> {
@@ -449,6 +450,7 @@ pub fn spawn_server(
449450
staked_nodes,
450451
max_staked_connections,
451452
max_unstaked_connections,
453+
max_streams_per_ms,
452454
wait_for_chunk_timeout,
453455
coalesce,
454456
)
@@ -468,7 +470,9 @@ pub fn spawn_server(
468470
mod test {
469471
use {
470472
super::*,
471-
crate::nonblocking::quic::{test::*, DEFAULT_WAIT_FOR_CHUNK_TIMEOUT},
473+
crate::nonblocking::quic::{
474+
test::*, DEFAULT_MAX_STREAMS_PER_MS, DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
475+
},
472476
crossbeam_channel::unbounded,
473477
solana_sdk::net::DEFAULT_TPU_COALESCE,
474478
std::net::SocketAddr,
@@ -498,6 +502,7 @@ mod test {
498502
staked_nodes,
499503
MAX_STAKED_CONNECTIONS,
500504
MAX_UNSTAKED_CONNECTIONS,
505+
DEFAULT_MAX_STREAMS_PER_MS,
501506
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
502507
DEFAULT_TPU_COALESCE,
503508
)
@@ -554,6 +559,7 @@ mod test {
554559
staked_nodes,
555560
MAX_STAKED_CONNECTIONS,
556561
MAX_UNSTAKED_CONNECTIONS,
562+
DEFAULT_MAX_STREAMS_PER_MS,
557563
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
558564
DEFAULT_TPU_COALESCE,
559565
)
@@ -597,6 +603,7 @@ mod test {
597603
staked_nodes,
598604
MAX_STAKED_CONNECTIONS,
599605
0, // Do not allow any connection from unstaked clients/nodes
606+
DEFAULT_MAX_STREAMS_PER_MS,
600607
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
601608
DEFAULT_TPU_COALESCE,
602609
)

0 commit comments

Comments
 (0)