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

Commit f1ebc5b

Browse files
behzadnouribehzadnouri
authored
separates out quic streamer connection stats from different servers (#31797)
1 parent 9281ab7 commit f1ebc5b

File tree

5 files changed

+43
-75
lines changed

5 files changed

+43
-75
lines changed

‎client/src/connection_cache.rs

+4-18
Original file line numberDiff line numberDiff line change
@@ -217,8 +217,7 @@ 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, quic::StreamStats,
221-
streamer::StakedNodes,
220+
nonblocking::quic::DEFAULT_WAIT_FOR_CHUNK_TIMEOUT, streamer::StakedNodes,
222221
},
223222
std::{
224223
net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket},
@@ -229,37 +228,25 @@ mod tests {
229228
},
230229
};
231230

232-
fn server_args() -> (
233-
UdpSocket,
234-
Arc<AtomicBool>,
235-
Keypair,
236-
IpAddr,
237-
Arc<StreamStats>,
238-
) {
231+
fn server_args() -> (UdpSocket, Arc<AtomicBool>, Keypair, IpAddr) {
239232
(
240233
UdpSocket::bind("127.0.0.1:0").unwrap(),
241234
Arc::new(AtomicBool::new(false)),
242235
Keypair::new(),
243236
"127.0.0.1".parse().unwrap(),
244-
Arc::new(StreamStats::default()),
245237
)
246238
}
247239

248240
#[test]
249241
fn test_connection_with_specified_client_endpoint() {
250242
// Start a response receiver:
251-
let (
252-
response_recv_socket,
253-
response_recv_exit,
254-
keypair2,
255-
response_recv_ip,
256-
response_recv_stats,
257-
) = server_args();
243+
let (response_recv_socket, response_recv_exit, keypair2, response_recv_ip) = server_args();
258244
let (sender2, _receiver2) = unbounded();
259245

260246
let staked_nodes = Arc::new(RwLock::new(StakedNodes::default()));
261247

262248
let (response_recv_endpoint, response_recv_thread) = solana_streamer::quic::spawn_server(
249+
"quic_streamer_test",
263250
response_recv_socket,
264251
&keypair2,
265252
response_recv_ip,
@@ -269,7 +256,6 @@ mod tests {
269256
staked_nodes,
270257
10,
271258
10,
272-
response_recv_stats,
273259
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
274260
DEFAULT_TPU_COALESCE,
275261
)

‎core/src/tpu.rs

+3-4
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ use {
3737
solana_sdk::{pubkey::Pubkey, signature::Keypair},
3838
solana_streamer::{
3939
nonblocking::quic::DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
40-
quic::{spawn_server, StreamStats, MAX_STAKED_CONNECTIONS, MAX_UNSTAKED_CONNECTIONS},
40+
quic::{spawn_server, MAX_STAKED_CONNECTIONS, MAX_UNSTAKED_CONNECTIONS},
4141
streamer::StakedNodes,
4242
},
4343
std::{
@@ -145,8 +145,8 @@ impl Tpu {
145145

146146
let (non_vote_sender, non_vote_receiver) = banking_tracer.create_channel_non_vote();
147147

148-
let stats = Arc::new(StreamStats::default());
149148
let (_, tpu_quic_t) = spawn_server(
149+
"quic_streamer_tpu",
150150
transactions_quic_sockets,
151151
keypair,
152152
cluster_info
@@ -160,13 +160,13 @@ impl Tpu {
160160
staked_nodes.clone(),
161161
MAX_STAKED_CONNECTIONS,
162162
MAX_UNSTAKED_CONNECTIONS,
163-
stats.clone(),
164163
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
165164
tpu_coalesce,
166165
)
167166
.unwrap();
168167

169168
let (_, tpu_forwards_quic_t) = spawn_server(
169+
"quic_streamer_tpu_forwards",
170170
transactions_forwards_quic_sockets,
171171
keypair,
172172
cluster_info
@@ -180,7 +180,6 @@ impl Tpu {
180180
staked_nodes.clone(),
181181
MAX_STAKED_CONNECTIONS.saturating_add(MAX_UNSTAKED_CONNECTIONS),
182182
0, // Prevent unstaked nodes from forwarding transactions
183-
stats,
184183
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
185184
tpu_coalesce,
186185
)

‎quic-client/tests/quic_client.rs

+12-26
Original file line numberDiff line numberDiff line change
@@ -10,8 +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, quic::StreamStats,
14-
streamer::StakedNodes, tls_certificates::new_self_signed_tls_certificate,
13+
nonblocking::quic::DEFAULT_WAIT_FOR_CHUNK_TIMEOUT, streamer::StakedNodes,
14+
tls_certificates::new_self_signed_tls_certificate,
1515
},
1616
std::{
1717
net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket},
@@ -49,19 +49,12 @@ mod tests {
4949
assert_eq!(total_packets, num_expected_packets);
5050
}
5151

52-
fn server_args() -> (
53-
UdpSocket,
54-
Arc<AtomicBool>,
55-
Keypair,
56-
IpAddr,
57-
Arc<StreamStats>,
58-
) {
52+
fn server_args() -> (UdpSocket, Arc<AtomicBool>, Keypair, IpAddr) {
5953
(
6054
UdpSocket::bind("127.0.0.1:0").unwrap(),
6155
Arc::new(AtomicBool::new(false)),
6256
Keypair::new(),
6357
"127.0.0.1".parse().unwrap(),
64-
Arc::new(StreamStats::default()),
6558
)
6659
}
6760

@@ -74,8 +67,9 @@ mod tests {
7467
solana_logger::setup();
7568
let (sender, receiver) = unbounded();
7669
let staked_nodes = Arc::new(RwLock::new(StakedNodes::default()));
77-
let (s, exit, keypair, ip, stats) = server_args();
70+
let (s, exit, keypair, ip) = server_args();
7871
let (_, t) = solana_streamer::quic::spawn_server(
72+
"quic_streamer_test",
7973
s.try_clone().unwrap(),
8074
&keypair,
8175
ip,
@@ -85,7 +79,6 @@ mod tests {
8579
staked_nodes,
8680
10,
8781
10,
88-
stats,
8982
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
9083
DEFAULT_TPU_COALESCE,
9184
)
@@ -154,8 +147,9 @@ mod tests {
154147
solana_logger::setup();
155148
let (sender, receiver) = unbounded();
156149
let staked_nodes = Arc::new(RwLock::new(StakedNodes::default()));
157-
let (s, exit, keypair, ip, stats) = server_args();
158-
let (_, t) = solana_streamer::nonblocking::quic::spawn_server(
150+
let (s, exit, keypair, ip) = server_args();
151+
let (_, _, t) = solana_streamer::nonblocking::quic::spawn_server(
152+
"quic_streamer_test",
159153
s.try_clone().unwrap(),
160154
&keypair,
161155
ip,
@@ -165,7 +159,6 @@ mod tests {
165159
staked_nodes,
166160
10,
167161
10,
168-
stats,
169162
Duration::from_secs(1), // wait_for_chunk_timeout
170163
DEFAULT_TPU_COALESCE,
171164
)
@@ -210,9 +203,9 @@ mod tests {
210203
// Request Receiver
211204
let (sender, receiver) = unbounded();
212205
let staked_nodes = Arc::new(RwLock::new(StakedNodes::default()));
213-
let (request_recv_socket, request_recv_exit, keypair, request_recv_ip, request_recv_stats) =
214-
server_args();
206+
let (request_recv_socket, request_recv_exit, keypair, request_recv_ip) = server_args();
215207
let (request_recv_endpoint, request_recv_thread) = solana_streamer::quic::spawn_server(
208+
"quic_streamer_test",
216209
request_recv_socket.try_clone().unwrap(),
217210
&keypair,
218211
request_recv_ip,
@@ -222,27 +215,21 @@ mod tests {
222215
staked_nodes.clone(),
223216
10,
224217
10,
225-
request_recv_stats,
226218
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
227219
DEFAULT_TPU_COALESCE,
228220
)
229221
.unwrap();
230222

231223
drop(request_recv_endpoint);
232224
// Response Receiver:
233-
let (
234-
response_recv_socket,
235-
response_recv_exit,
236-
keypair2,
237-
response_recv_ip,
238-
response_recv_stats,
239-
) = server_args();
225+
let (response_recv_socket, response_recv_exit, keypair2, response_recv_ip) = server_args();
240226
let (sender2, receiver2) = unbounded();
241227

242228
let addr = response_recv_socket.local_addr().unwrap().ip();
243229
let port = response_recv_socket.local_addr().unwrap().port();
244230
let server_addr = SocketAddr::new(addr, port);
245231
let (response_recv_endpoint, response_recv_thread) = solana_streamer::quic::spawn_server(
232+
"quic_streamer_test",
246233
response_recv_socket,
247234
&keypair2,
248235
response_recv_ip,
@@ -252,7 +239,6 @@ mod tests {
252239
staked_nodes,
253240
10,
254241
10,
255-
response_recv_stats,
256242
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
257243
DEFAULT_TPU_COALESCE,
258244
)

‎streamer/src/nonblocking/quic.rs

+16-16
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ struct PacketAccumulator {
8585

8686
#[allow(clippy::too_many_arguments)]
8787
pub fn spawn_server(
88+
name: &'static str,
8889
sock: UdpSocket,
8990
keypair: &Keypair,
9091
gossip_host: IpAddr,
@@ -94,32 +95,34 @@ pub fn spawn_server(
9495
staked_nodes: Arc<RwLock<StakedNodes>>,
9596
max_staked_connections: usize,
9697
max_unstaked_connections: usize,
97-
stats: Arc<StreamStats>,
9898
wait_for_chunk_timeout: Duration,
9999
coalesce: Duration,
100-
) -> Result<(Endpoint, JoinHandle<()>), QuicServerError> {
101-
info!("Start quic server on {:?}", sock);
100+
) -> Result<(Endpoint, Arc<StreamStats>, JoinHandle<()>), QuicServerError> {
101+
info!("Start {name} quic server on {sock:?}");
102102
let (config, _cert) = configure_server(keypair, gossip_host)?;
103103

104104
let endpoint = Endpoint::new(EndpointConfig::default(), Some(config), sock, TokioRuntime)
105105
.map_err(QuicServerError::EndpointFailed)?;
106+
let stats = Arc::<StreamStats>::default();
106107
let handle = tokio::spawn(run_server(
108+
name,
107109
endpoint.clone(),
108110
packet_sender,
109111
exit,
110112
max_connections_per_peer,
111113
staked_nodes,
112114
max_staked_connections,
113115
max_unstaked_connections,
114-
stats,
116+
stats.clone(),
115117
wait_for_chunk_timeout,
116118
coalesce,
117119
));
118-
Ok((endpoint, handle))
120+
Ok((endpoint, stats, handle))
119121
}
120122

121123
#[allow(clippy::too_many_arguments)]
122-
pub async fn run_server(
124+
async fn run_server(
125+
name: &'static str,
123126
incoming: Endpoint,
124127
packet_sender: Sender<PacketBatch>,
125128
exit: Arc<AtomicBool>,
@@ -152,7 +155,7 @@ pub async fn run_server(
152155
let timeout_connection = timeout(WAIT_FOR_CONNECTION_TIMEOUT, incoming.accept()).await;
153156

154157
if last_datapoint.elapsed().as_secs() >= 5 {
155-
stats.report();
158+
stats.report(name);
156159
last_datapoint = Instant::now();
157160
}
158161

@@ -1166,8 +1169,8 @@ pub mod test {
11661169
let ip = "127.0.0.1".parse().unwrap();
11671170
let server_address = s.local_addr().unwrap();
11681171
let staked_nodes = Arc::new(RwLock::new(option_staked_nodes.unwrap_or_default()));
1169-
let stats = Arc::new(StreamStats::default());
1170-
let (_, t) = spawn_server(
1172+
let (_, stats, t) = spawn_server(
1173+
"quic_streamer_test",
11711174
s,
11721175
&keypair,
11731176
ip,
@@ -1177,7 +1180,6 @@ pub mod test {
11771180
staked_nodes,
11781181
MAX_STAKED_CONNECTIONS,
11791182
MAX_UNSTAKED_CONNECTIONS,
1180-
stats.clone(),
11811183
Duration::from_secs(2),
11821184
DEFAULT_TPU_COALESCE,
11831185
)
@@ -1597,8 +1599,8 @@ pub mod test {
15971599
let ip = "127.0.0.1".parse().unwrap();
15981600
let server_address = s.local_addr().unwrap();
15991601
let staked_nodes = Arc::new(RwLock::new(StakedNodes::default()));
1600-
let stats = Arc::new(StreamStats::default());
1601-
let (_, t) = spawn_server(
1602+
let (_, _, t) = spawn_server(
1603+
"quic_streamer_test",
16021604
s,
16031605
&keypair,
16041606
ip,
@@ -1608,7 +1610,6 @@ pub mod test {
16081610
staked_nodes,
16091611
MAX_STAKED_CONNECTIONS,
16101612
0, // Do not allow any connection from unstaked clients/nodes
1611-
stats,
16121613
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
16131614
DEFAULT_TPU_COALESCE,
16141615
)
@@ -1629,8 +1630,8 @@ pub mod test {
16291630
let ip = "127.0.0.1".parse().unwrap();
16301631
let server_address = s.local_addr().unwrap();
16311632
let staked_nodes = Arc::new(RwLock::new(StakedNodes::default()));
1632-
let stats = Arc::new(StreamStats::default());
1633-
let (_, t) = spawn_server(
1633+
let (_, stats, t) = spawn_server(
1634+
"quic_streamer_test",
16341635
s,
16351636
&keypair,
16361637
ip,
@@ -1640,7 +1641,6 @@ pub mod test {
16401641
staked_nodes,
16411642
MAX_STAKED_CONNECTIONS,
16421643
MAX_UNSTAKED_CONNECTIONS,
1643-
stats.clone(),
16441644
DEFAULT_WAIT_FOR_CHUNK_TIMEOUT,
16451645
DEFAULT_TPU_COALESCE,
16461646
)

0 commit comments

Comments
 (0)