Skip to content

Commit 273a1fc

Browse files
fffonionfffonion
committed
chore: clippy happy
1 parent dade582 commit 273a1fc

9 files changed

Lines changed: 98 additions & 76 deletions

File tree

‎pd-edge/examples/http_proxy_perf_framework.rs‎

Lines changed: 38 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -807,17 +807,29 @@ struct ProxyProcess {
807807
child: Child,
808808
}
809809

810+
struct ProxyProcessConfig<'a> {
811+
binary_path: &'a Path,
812+
data_addr: SocketAddr,
813+
https_addr: Option<SocketAddr>,
814+
http3_addr: Option<SocketAddr>,
815+
admin_addr: SocketAddr,
816+
vm_fuel: Option<u64>,
817+
vm_fuel_check_interval: u32,
818+
vm_execution_mode: VmExecutionModeArg,
819+
}
820+
810821
impl ProxyProcess {
811-
fn spawn(
812-
binary_path: &Path,
813-
data_addr: SocketAddr,
814-
https_addr: Option<SocketAddr>,
815-
http3_addr: Option<SocketAddr>,
816-
admin_addr: SocketAddr,
817-
vm_fuel: Option<u64>,
818-
vm_fuel_check_interval: u32,
819-
vm_execution_mode: VmExecutionModeArg,
820-
) -> Result<Self, Box<dyn std::error::Error>> {
822+
fn spawn(config: ProxyProcessConfig<'_>) -> Result<Self, Box<dyn std::error::Error>> {
823+
let ProxyProcessConfig {
824+
binary_path,
825+
data_addr,
826+
https_addr,
827+
http3_addr,
828+
admin_addr,
829+
vm_fuel,
830+
vm_fuel_check_interval,
831+
vm_execution_mode,
832+
} = config;
821833
let rust_log = env::var("PD_EDGE_PROXY_RUST_LOG").unwrap_or_else(|_| "error".to_string());
822834
let inherit_stdout = env_flag("PD_EDGE_PROXY_STDOUT_INHERIT");
823835
let mut command = Command::new(binary_path);
@@ -1599,16 +1611,16 @@ async fn run_scenario(
15991611
None
16001612
};
16011613
let admin_addr = reserve_loopback_addr()?;
1602-
let mut proxy = ProxyProcess::spawn(
1614+
let mut proxy = ProxyProcess::spawn(ProxyProcessConfig {
16031615
binary_path,
16041616
data_addr,
16051617
https_addr,
16061618
http3_addr,
16071619
admin_addr,
1608-
config.vm_fuel,
1609-
config.vm_fuel_check_interval,
1610-
config.vm_execution_mode,
1611-
)?;
1620+
vm_fuel: config.vm_fuel,
1621+
vm_fuel_check_interval: config.vm_fuel_check_interval,
1622+
vm_execution_mode: config.vm_execution_mode,
1623+
})?;
16121624

16131625
wait_until_proxy_ready(
16141626
&clients.http,
@@ -2357,7 +2369,7 @@ impl BenchmarkUpstreamResponse {
23572369
response
23582370
}
23592371

2360-
fn into_http3_head(&self) -> HyperResponse<()> {
2372+
fn to_http3_head(&self) -> HyperResponse<()> {
23612373
let mut response = HyperResponse::builder()
23622374
.status(HttpStatusCode::OK)
23632375
.body(())
@@ -2518,15 +2530,15 @@ fn build_http3_bench_client_config() -> quinn::ClientConfig {
25182530
#[cfg(feature = "http3")]
25192531
fn build_http3_transport_config() -> quinn::TransportConfig {
25202532
let mut transport = quinn::TransportConfig::default();
2521-
transport.max_concurrent_bidi_streams(
2522-
quinn::VarInt::from_u32(BENCH_HTTP3_MAX_CONCURRENT_BIDI_STREAMS).into(),
2523-
);
2524-
transport.stream_receive_window(
2525-
quinn::VarInt::from_u32(BENCH_HTTP3_STREAM_RECEIVE_WINDOW_BYTES).into(),
2526-
);
2527-
transport.receive_window(
2528-
quinn::VarInt::from_u32(BENCH_HTTP3_CONNECTION_RECEIVE_WINDOW_BYTES).into(),
2529-
);
2533+
transport.max_concurrent_bidi_streams(quinn::VarInt::from_u32(
2534+
BENCH_HTTP3_MAX_CONCURRENT_BIDI_STREAMS,
2535+
));
2536+
transport.stream_receive_window(quinn::VarInt::from_u32(
2537+
BENCH_HTTP3_STREAM_RECEIVE_WINDOW_BYTES,
2538+
));
2539+
transport.receive_window(quinn::VarInt::from_u32(
2540+
BENCH_HTTP3_CONNECTION_RECEIVE_WINDOW_BYTES,
2541+
));
25302542
transport.send_window(BENCH_HTTP3_SEND_WINDOW_BYTES);
25312543
transport.keep_alive_interval(Some(Duration::from_millis(
25322544
BENCH_HTTP3_KEEPALIVE_INTERVAL_MS,
@@ -2687,7 +2699,7 @@ async fn write_http3_bench_response(
26872699
) {
26882700
let body = response.body.clone();
26892701
stream
2690-
.send_response(response.into_http3_head())
2702+
.send_response(response.to_http3_head())
26912703
.await
26922704
.expect("http3 upstream benchmark response head should send");
26932705
if !body.is_empty() {

‎pd-edge/src/abi_impl/http/state.rs‎

Lines changed: 20 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -704,7 +704,7 @@ pub(crate) fn upstream_reqwest_client_builder() -> reqwest::ClientBuilder {
704704
let builder = reqwest::Client::builder().tls_info(true).tcp_nodelay(true);
705705
#[cfg(feature = "http2")]
706706
{
707-
return builder.http2_adaptive_window(true);
707+
builder.http2_adaptive_window(true)
708708
}
709709
#[cfg(not(feature = "http2"))]
710710
{
@@ -794,10 +794,7 @@ impl HttpOutboundRequestNode {
794794
}
795795
}
796796

797-
fn headers_or_request_head<'a>(
798-
&'a self,
799-
request_head: &'a HttpRequestHead,
800-
) -> &'a HeaderMap {
797+
fn headers_or_request_head<'a>(&'a self, request_head: &'a HttpRequestHead) -> &'a HeaderMap {
801798
if self.inherits_request_head {
802799
request_head.headers()
803800
} else {
@@ -1284,8 +1281,6 @@ pub(crate) struct RuntimeServices {
12841281
upstream_http_sessions: Option<SharedHttpUpstreamSessions>,
12851282
upstream_http3_sessions: Option<SharedHttp3UpstreamSessions>,
12861283
downstream_http_sessions: Option<http2::SharedHttpDownstreamSessions>,
1287-
#[cfg_attr(not(feature = "http3"), allow(dead_code))]
1288-
downstream_http3_sessions: Option<http3::SharedHttp3DownstreamSessions>,
12891284
#[cfg(feature = "tls")]
12901285
downstream_tls_termination: Option<Arc<tokio_rustls::rustls::ServerConfig>>,
12911286
rate_limiter: SharedRateLimiter,
@@ -1302,7 +1297,6 @@ impl RuntimeServices {
13021297
upstream_http_sessions: None,
13031298
upstream_http3_sessions: None,
13041299
downstream_http_sessions: None,
1305-
downstream_http3_sessions: None,
13061300
#[cfg(feature = "tls")]
13071301
downstream_tls_termination: None,
13081302
rate_limiter,
@@ -1359,7 +1353,6 @@ pub(crate) fn new_shared_http_plane_runtime_services(
13591353
upstream_http_sessions: SharedHttpUpstreamSessions,
13601354
upstream_http3_sessions: SharedHttp3UpstreamSessions,
13611355
downstream_http_sessions: http2::SharedHttpDownstreamSessions,
1362-
downstream_http3_sessions: http3::SharedHttp3DownstreamSessions,
13631356
) -> SharedRuntimeServices {
13641357
Arc::new(RuntimeServices {
13651358
upstream_client: Some(upstream_client),
@@ -1368,7 +1361,6 @@ pub(crate) fn new_shared_http_plane_runtime_services(
13681361
upstream_http_sessions: Some(upstream_http_sessions),
13691362
upstream_http3_sessions: Some(upstream_http3_sessions),
13701363
downstream_http_sessions: Some(downstream_http_sessions),
1371-
downstream_http3_sessions: Some(downstream_http3_sessions),
13721364
#[cfg(feature = "tls")]
13731365
downstream_tls_termination: None,
13741366
rate_limiter,
@@ -2001,7 +1993,8 @@ impl ProxyVmContext {
20011993
client_ip,
20021994
headers,
20031995
};
2004-
let downstream_websocket = WebSocketConnectionState::for_http_request(request_head.headers());
1996+
let downstream_websocket =
1997+
WebSocketConnectionState::for_http_request(request_head.headers());
20051998
*self.lock_request_head() = request_head;
20061999

20072000
{
@@ -2794,10 +2787,22 @@ fn prepared_upstream_request(
27942787
let services = context.services();
27952788
let (method, path, query, headers) = context.with_request_head(|request_head| {
27962789
(
2797-
exchange.request.method_or_request_head(request_head).clone(),
2798-
exchange.request.path_or_request_head(request_head).to_string(),
2799-
exchange.request.query_or_request_head(request_head).to_string(),
2800-
exchange.request.headers_or_request_head(request_head).clone(),
2790+
exchange
2791+
.request
2792+
.method_or_request_head(request_head)
2793+
.clone(),
2794+
exchange
2795+
.request
2796+
.path_or_request_head(request_head)
2797+
.to_string(),
2798+
exchange
2799+
.request
2800+
.query_or_request_head(request_head)
2801+
.to_string(),
2802+
exchange
2803+
.request
2804+
.headers_or_request_head(request_head)
2805+
.clone(),
28012806
)
28022807
});
28032808
Ok(PreparedUpstreamRequest {

‎pd-edge/src/abi_impl/mod.rs‎

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -47,9 +47,7 @@ pub(crate) use self::http3::{
4747
pub(crate) use self::quic::{build_quic_server_config, tune_udp_socket_buffers};
4848
#[cfg(feature = "tls")]
4949
pub(crate) use self::transport::build_default_self_signed_server_config;
50-
pub(crate) use self::transport::{
51-
ReplayPrefixedIo, new_shared_tls_session_cache,
52-
};
50+
pub(crate) use self::transport::{ReplayPrefixedIo, new_shared_tls_session_cache};
5351

5452
pub type SharedRateLimiter = Arc<RateLimiterStore>;
5553
pub type SharedVmAsyncOps = Arc<LazyVmAsyncOps>;
@@ -230,7 +228,11 @@ pub struct LazyVmAsyncOps {
230228
impl LazyVmAsyncOps {
231229
fn lock_ops(&self) -> ProfiledMutexGuard<'_, VmAsyncOps> {
232230
let inner = self.inner.get_or_init(|| Mutex::new(VmAsyncOps::new()));
233-
lock_metrics::lock(inner, LockMetricKey::VmAsyncOps, "vm async ops lock poisoned")
231+
lock_metrics::lock(
232+
inner,
233+
LockMetricKey::VmAsyncOps,
234+
"vm async ops lock poisoned",
235+
)
234236
}
235237
}
236238

@@ -261,7 +263,7 @@ pub fn enter_edge_host_context(
261263
) -> EdgeHostContextGuard {
262264
#[cfg(feature = "console")]
263265
{
264-
return enter_edge_host_context_inner(vm_context, async_ops, None);
266+
enter_edge_host_context_inner(vm_context, async_ops, None)
265267
}
266268
#[cfg(not(feature = "console"))]
267269
{
@@ -660,6 +662,12 @@ impl RateLimiterStore {
660662
}
661663
}
662664

665+
impl Default for RateLimiterStore {
666+
fn default() -> Self {
667+
Self::new()
668+
}
669+
}
670+
663671
#[derive(Clone, Debug)]
664672
pub(crate) enum EdgeVirtualIoHandle {
665673
BufferedRead { text: String, offset: usize },

‎pd-edge/src/abi_impl/quic/mod.rs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -148,12 +148,12 @@ pub(crate) fn tune_udp_socket_buffers(socket: &std::net::UdpSocket) -> io::Resul
148148
#[cfg(feature = "http3")]
149149
fn default_quic_transport_config() -> quinn::TransportConfig {
150150
let mut transport = quinn::TransportConfig::default();
151-
transport.max_concurrent_bidi_streams(
152-
quinn::VarInt::from_u32(QUIC_MAX_CONCURRENT_BIDI_STREAMS).into(),
153-
);
154151
transport
155-
.stream_receive_window(quinn::VarInt::from_u32(QUIC_STREAM_RECEIVE_WINDOW_BYTES).into());
156-
transport.receive_window(quinn::VarInt::from_u32(QUIC_CONNECTION_RECEIVE_WINDOW_BYTES).into());
152+
.max_concurrent_bidi_streams(quinn::VarInt::from_u32(QUIC_MAX_CONCURRENT_BIDI_STREAMS));
153+
transport.stream_receive_window(quinn::VarInt::from_u32(QUIC_STREAM_RECEIVE_WINDOW_BYTES));
154+
transport.receive_window(quinn::VarInt::from_u32(
155+
QUIC_CONNECTION_RECEIVE_WINDOW_BYTES,
156+
));
157157
transport.send_window(QUIC_SEND_WINDOW_BYTES);
158158
transport.keep_alive_interval(Some(std::time::Duration::from_millis(
159159
QUIC_KEEPALIVE_INTERVAL_MS,

‎pd-edge/src/cache.rs‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ where
4040
}
4141
}
4242

43+
#[cfg(test)]
4344
pub(crate) fn get(&mut self, key: &K) -> Option<&V> {
4445
if !self.values.contains_key(key) {
4546
return None;
@@ -124,6 +125,7 @@ where
124125
self.values.len()
125126
}
126127

128+
#[cfg(test)]
127129
pub(crate) fn capacity(&self) -> usize {
128130
self.capacity
129131
}
@@ -139,7 +141,6 @@ pub(crate) struct ShardedRwLruStore<K, V>
139141
where
140142
K: Clone + Eq + Hash,
141143
{
142-
capacity: usize,
143144
shards: Box<[RwLock<BoundedLruStore<K, V>>]>,
144145
}
145146

@@ -158,7 +159,7 @@ where
158159
})
159160
.collect::<Vec<_>>()
160161
.into_boxed_slice();
161-
Self { capacity, shards }
162+
Self { shards }
162163
}
163164

164165
pub(crate) fn peek_cloned(
@@ -219,7 +220,15 @@ where
219220

220221
#[cfg(test)]
221222
pub(crate) fn capacity(&self) -> usize {
222-
self.capacity
223+
self.shards
224+
.iter()
225+
.map(|shard| {
226+
shard
227+
.read()
228+
.expect("sharded lru store lock poisoned")
229+
.capacity
230+
})
231+
.sum()
223232
}
224233

225234
#[cfg(test)]

‎pd-edge/src/debug_session.rs‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -347,12 +347,9 @@ pub fn run_vm_with_optional_debugger(
347347
};
348348

349349
if let Some(session) = session {
350-
let matched = context
351-
.with_request_head(|request_head| request_matches_session(
352-
request_head.headers(),
353-
request_head.path(),
354-
&session,
355-
));
350+
let matched = context.with_request_head(|request_head| {
351+
request_matches_session(request_head.headers(), request_head.path(), &session)
352+
});
356353
if !matched {
357354
return vm.run();
358355
}

‎pd-edge/src/runtime.rs‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -232,7 +232,6 @@ impl SharedState {
232232
upstream_http_sessions.clone(),
233233
upstream_http3_sessions.clone(),
234234
downstream_http2_sessions.clone(),
235-
downstream_http3_sessions.clone(),
236235
);
237236
Self {
238237
active_program: Arc::new(ArcSwapOption::from(None::<Arc<LoadedProgram>>)),

‎pd-edge/src/runtime/http_plane/proxy_path.rs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -500,11 +500,11 @@ async fn promote_captured_downstream_transport_into_http_request(
500500
let captured = captured.map_err(VmError::HostError)?;
501501
context
502502
.promote_downstream_http_request(
503-
captured.request,
504-
captured.http2_attachment,
505-
captured.http1_upgrade,
506-
)
507-
.await;
503+
captured.request,
504+
captured.http2_attachment,
505+
captured.http1_upgrade,
506+
)
507+
.await;
508508
Ok(())
509509
}
510510

‎pd-edge/src/runtime/vm_runner.rs‎

Lines changed: 2 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -212,11 +212,7 @@ async fn run_vm_async(
212212
);
213213
if debug.attach_debugger {
214214
let vm_context = vm_store.data().vm_context.clone();
215-
run_vm_with_optional_debugger(
216-
&debug_session,
217-
vm_context,
218-
vm_store.vm_mut(),
219-
)
215+
run_vm_with_optional_debugger(&debug_session, vm_context, vm_store.vm_mut())
220216
} else {
221217
vm_store.run()
222218
}
@@ -267,11 +263,7 @@ fn run_vm_threading(
267263
);
268264
if debug.attach_debugger {
269265
let vm_context = vm_store.data().vm_context.clone();
270-
run_vm_with_optional_debugger(
271-
&debug_session,
272-
vm_context,
273-
vm_store.vm_mut(),
274-
)
266+
run_vm_with_optional_debugger(&debug_session, vm_context, vm_store.vm_mut())
275267
} else {
276268
vm_store.run()
277269
}

0 commit comments

Comments
 (0)