perf(core): use quanta::Instant for hot-path timing (#2384)

Replace std::time::Instant with quanta::Instant on per-packet, per-RPC,
and per-session paths. TSC-based, ~5ns vs ~25ns per now() call.

Reuses the existing `extern crate self as hotpath` alias so
`use hotpath::instant::Instant;` resolves to the same quanta type with
or without the hotpath feature. Leaves tokio::time::Instant and
smoltcp::time::Instant untouched.
This commit is contained in:
fanyang
2026-06-28 10:43:55 +08:00
committed by GitHub
parent be2034dd06
commit 7205517160
32 changed files with 123 additions and 80 deletions
Generated
+1
View File
@@ -2362,6 +2362,7 @@ dependencies = [
"prost-reflect", "prost-reflect",
"prost-reflect-build", "prost-reflect-build",
"prost-wkt-types", "prost-wkt-types",
"quanta",
"quinn", "quinn",
"quinn-proto", "quinn-proto",
"quote", "quote",
+1
View File
@@ -53,6 +53,7 @@ chrono = { version = "0.4.37", features = ["serde"] }
guarden = "0.2" guarden = "0.2"
hotpath = { version = "0.18", default-features = false, optional = true } hotpath = { version = "0.18", default-features = false, optional = true }
quanta = "0.12"
delegate = "0.13.5" delegate = "0.13.5"
+7 -5
View File
@@ -3,9 +3,11 @@ use std::{
net::{IpAddr, SocketAddr}, net::{IpAddr, SocketAddr},
str::FromStr as _, str::FromStr as _,
sync::Arc, sync::Arc,
time::{Duration, Instant, SystemTime, UNIX_EPOCH}, time::{Duration, SystemTime, UNIX_EPOCH},
}; };
use hotpath::instant::Instant;
use crate::common::{config::ConfigLoader, global_ctx::ArcGlobalCtx, token_bucket::TokenBucket}; use crate::common::{config::ConfigLoader, global_ctx::ArcGlobalCtx, token_bucket::TokenBucket};
use crate::proto::acl::*; use crate::proto::acl::*;
use anyhow::Context as _; use anyhow::Context as _;
@@ -107,10 +109,10 @@ impl AclCacheKey {
// Cache entry with timestamp for LRU cleanup // Cache entry with timestamp for LRU cleanup
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct AclCacheEntry { pub(crate) struct AclCacheEntry {
pub action: Action, pub action: Action,
pub matched_rule: RuleId, pub matched_rule: RuleId,
pub last_access: std::time::Instant, pub last_access: Instant,
// New fields to track rule characteristics for proper cache behavior // New fields to track rule characteristics for proper cache behavior
pub conn_track_key: Option<String>, pub conn_track_key: Option<String>,
pub rate_limit_keys: Vec<RateLimitKey>, pub rate_limit_keys: Vec<RateLimitKey>,
@@ -410,7 +412,7 @@ impl AclProcessor {
} }
// Remove oldest entries (LRU cleanup) // Remove oldest entries (LRU cleanup)
let mut entries: Vec<(AclCacheKey, std::time::Instant)> = cache let mut entries: Vec<(AclCacheKey, Instant)> = cache
.iter() .iter()
.map(|entry| (entry.key().clone(), entry.value().last_access)) .map(|entry| (entry.key().clone(), entry.value().last_access))
.collect(); .collect();
@@ -431,7 +433,7 @@ impl AclProcessor {
); );
} }
pub fn process_packet_with_cache_entry( pub(crate) fn process_packet_with_cache_entry(
&self, &self,
packet_info: &PacketInfo, packet_info: &PacketInfo,
cache_entry: &AclCacheEntry, cache_entry: &AclCacheEntry,
+2 -1
View File
@@ -1,9 +1,10 @@
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::cell::UnsafeCell; use std::cell::UnsafeCell;
use std::fmt; use std::fmt;
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, Instant}; use std::time::Duration;
use tokio::time::interval; use tokio::time::interval;
use tokio_util::task::AbortOnDropHandle; use tokio_util::task::AbortOnDropHandle;
+3 -2
View File
@@ -2,12 +2,13 @@ use std::collections::BTreeSet;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicBool;
use std::sync::{Arc, RwLock}; use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant}; use std::time::Duration;
use crate::proto::common::{NatType, StunInfo}; use crate::proto::common::{NatType, StunInfo};
use anyhow::Context; use anyhow::Context;
use chrono::Local; use chrono::Local;
use crossbeam::atomic::AtomicCell; use crossbeam::atomic::AtomicCell;
use hotpath::instant::Instant;
use rand::seq::IteratorRandom; use rand::seq::IteratorRandom;
use socket2::{SockAddr, SockRef}; use socket2::{SockAddr, SockRef};
use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -1312,7 +1313,7 @@ impl StunInfoCollectorTrait for MockStunInfoCollector {
StunInfo { StunInfo {
udp_nat_type: self.udp_nat_type as i32, udp_nat_type: self.udp_nat_type as i32,
tcp_nat_type: NatType::Unknown as i32, tcp_nat_type: NatType::Unknown as i32,
last_update_time: std::time::Instant::now().elapsed().as_secs() as i64, last_update_time: Local::now().timestamp(),
min_port: 100, min_port: 100,
max_port: 200, max_port: 200,
public_ip: vec!["127.0.0.1".to_string(), "::1".to_string()], public_ip: vec!["127.0.0.1".to_string(), "::1".to_string()],
+3 -1
View File
@@ -8,9 +8,11 @@ use std::{
Arc, Arc,
atomic::{AtomicBool, Ordering}, atomic::{AtomicBool, Ordering},
}, },
time::{Duration, Instant}, time::Duration,
}; };
use hotpath::instant::Instant;
use crate::{ use crate::{
common::{ common::{
PeerId, dns::socket_addrs, error::Error, global_ctx::ArcGlobalCtx, PeerId, dns::socket_addrs, error::Error, global_ctx::ArcGlobalCtx,
+2 -1
View File
@@ -2,10 +2,11 @@ use std::{
collections::BTreeSet, collections::BTreeSet,
future::Future, future::Future,
sync::{Arc, Weak}, sync::{Arc, Weak},
time::{Duration, Instant}, time::Duration,
}; };
use dashmap::DashSet; use dashmap::DashSet;
use hotpath::instant::Instant;
use tokio::{sync::mpsc, task::JoinSet, time::timeout}; use tokio::{sync::mpsc, task::JoinSet, time::timeout};
use crate::{ use crate::{
+2 -1
View File
@@ -1,10 +1,11 @@
use std::{ use std::{
net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
sync::Arc, sync::Arc,
time::{Duration, Instant}, time::Duration,
}; };
use anyhow::{Context, Error}; use anyhow::{Context, Error};
use hotpath::instant::Instant;
use rand::Rng as _; use rand::Rng as _;
use tokio::task::JoinSet; use tokio::task::JoinSet;
@@ -1,10 +1,11 @@
use std::{ use std::{
net::{IpAddr, SocketAddr, SocketAddrV4}, net::{IpAddr, SocketAddr, SocketAddrV4},
sync::Arc, sync::Arc,
time::{Duration, Instant}, time::Duration,
}; };
use anyhow::Context; use anyhow::Context;
use hotpath::instant::Instant;
use tokio::sync::Mutex; use tokio::sync::Mutex;
use tokio_util::task::AbortOnDropHandle; use tokio_util::task::AbortOnDropHandle;
@@ -7,6 +7,7 @@ use std::{
use crossbeam::atomic::AtomicCell; use crossbeam::atomic::AtomicCell;
use dashmap::{DashMap, DashSet}; use dashmap::{DashMap, DashSet};
use guarden::defer; use guarden::defer;
use hotpath::instant::Instant;
use rand::seq::SliceRandom as _; use rand::seq::SliceRandom as _;
use tokio::{net::UdpSocket, sync::Mutex, task::JoinSet}; use tokio::{net::UdpSocket, sync::Mutex, task::JoinSet};
use tracing::{Instrument, Level, instrument}; use tracing::{Instrument, Level, instrument};
@@ -356,9 +357,9 @@ pub(crate) struct UdpHolePunchListener {
_port_mapping_lease: Option<upnp::UdpPortMappingLease>, _port_mapping_lease: Option<upnp::UdpPortMappingLease>,
conn_counter: Arc<Box<dyn TunnelConnCounter>>, conn_counter: Arc<Box<dyn TunnelConnCounter>>,
listen_time: std::time::Instant, listen_time: Instant,
last_select_time: AtomicCell<std::time::Instant>, last_select_time: AtomicCell<Instant>,
last_active_time: Arc<AtomicCell<std::time::Instant>>, last_active_time: Arc<AtomicCell<Instant>>,
} }
impl UdpHolePunchListener { impl UdpHolePunchListener {
@@ -421,14 +422,14 @@ impl UdpHolePunchListener {
running_clone.store(false); running_clone.store(false);
}); });
let last_active_time = Arc::new(AtomicCell::new(std::time::Instant::now())); let last_active_time = Arc::new(AtomicCell::new(Instant::now()));
let conn_counter_clone = conn_counter.clone(); let conn_counter_clone = conn_counter.clone();
let last_active_time_clone = last_active_time.clone(); let last_active_time_clone = last_active_time.clone();
tasks.spawn(async move { tasks.spawn(async move {
loop { loop {
tokio::time::sleep(std::time::Duration::from_secs(5)).await; tokio::time::sleep(std::time::Duration::from_secs(5)).await;
if conn_counter_clone.get().unwrap_or(0) != 0 { if conn_counter_clone.get().unwrap_or(0) != 0 {
last_active_time_clone.store(std::time::Instant::now()); last_active_time_clone.store(Instant::now());
} }
} }
}); });
@@ -444,14 +445,14 @@ impl UdpHolePunchListener {
_port_mapping_lease: port_mapping_lease, _port_mapping_lease: port_mapping_lease,
conn_counter, conn_counter,
listen_time: std::time::Instant::now(), listen_time: Instant::now(),
last_select_time: AtomicCell::new(std::time::Instant::now()), last_select_time: AtomicCell::new(Instant::now()),
last_active_time, last_active_time,
}) })
} }
pub async fn get_socket(&self) -> Arc<UdpSocket> { pub async fn get_socket(&self) -> Arc<UdpSocket> {
self.last_select_time.store(std::time::Instant::now()); self.last_select_time.store(Instant::now());
self.socket.clone() self.socket.clone()
} }
@@ -1,9 +1,7 @@
use std::{ use std::{sync::Arc, time::Duration};
sync::Arc,
time::{Duration, Instant},
};
use anyhow::Context; use anyhow::Context;
use hotpath::instant::Instant;
use tokio::net::UdpSocket; use tokio::net::UdpSocket;
use tokio_util::task::AbortOnDropHandle; use tokio_util::task::AbortOnDropHandle;
+2 -1
View File
@@ -1,6 +1,6 @@
use std::{ use std::{
sync::{Arc, atomic::AtomicBool}, sync::{Arc, atomic::AtomicBool},
time::{Duration, Instant}, time::Duration,
}; };
use anyhow::{Context, Error}; use anyhow::{Context, Error};
@@ -8,6 +8,7 @@ use both_easy_sym::{PunchBothEasySymHoleClient, PunchBothEasySymHoleServer};
use common::{PunchHoleServerCommon, UdpNatType, UdpPunchClientMethod}; use common::{PunchHoleServerCommon, UdpNatType, UdpPunchClientMethod};
use cone::{PunchConeHoleClient, PunchConeHoleServer}; use cone::{PunchConeHoleClient, PunchConeHoleServer};
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use once_cell::sync::Lazy; use once_cell::sync::Lazy;
use sym_to_cone::{PunchSymToConeHoleClient, PunchSymToConeHoleServer}; use sym_to_cone::{PunchSymToConeHoleClient, PunchSymToConeHoleServer};
use tokio::{sync::Mutex, task::JoinHandle}; use tokio::{sync::Mutex, task::JoinHandle};
@@ -5,11 +5,12 @@ use std::{
Arc, Arc,
atomic::{AtomicBool, Ordering}, atomic::{AtomicBool, Ordering},
}, },
time::{Duration, Instant}, time::Duration,
}; };
use anyhow::Context; use anyhow::Context;
use guarden::defer; use guarden::defer;
use hotpath::instant::Instant;
use rand::{Rng, seq::SliceRandom}; use rand::{Rng, seq::SliceRandom};
use tokio::{net::UdpSocket, sync::RwLock}; use tokio::{net::UdpSocket, sync::RwLock};
use tokio_util::task::AbortOnDropHandle; use tokio_util::task::AbortOnDropHandle;
+3 -2
View File
@@ -7,6 +7,7 @@ use std::{
}; };
use anyhow::Context; use anyhow::Context;
use hotpath::instant::Instant;
use pnet::packet::{ use pnet::packet::{
Packet, Packet,
icmp::{self, IcmpCode, IcmpTypes, MutableIcmpPacket, echo_reply::MutableEchoReplyPacket}, icmp::{self, IcmpCode, IcmpTypes, MutableIcmpPacket, echo_reply::MutableEchoReplyPacket},
@@ -45,7 +46,7 @@ struct IcmpNatEntry {
src_peer_id: PeerId, src_peer_id: PeerId,
my_peer_id: PeerId, my_peer_id: PeerId,
src_ip: IpAddr, src_ip: IpAddr,
start_time: std::time::Instant, start_time: Instant,
mapped_dst_ip: std::net::Ipv4Addr, mapped_dst_ip: std::net::Ipv4Addr,
} }
@@ -60,7 +61,7 @@ impl IcmpNatEntry {
src_peer_id, src_peer_id,
my_peer_id, my_peer_id,
src_ip, src_ip,
start_time: std::time::Instant::now(), start_time: Instant::now(),
mapped_dst_ip, mapped_dst_ip,
}) })
} }
+2 -1
View File
@@ -1,9 +1,10 @@
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use pnet::packet::Packet; use pnet::packet::Packet;
use pnet::packet::ip::IpNextHeaderProtocol; use pnet::packet::ip::IpNextHeaderProtocol;
use pnet::packet::ipv4::{self, Ipv4Flags, Ipv4Packet, MutableIpv4Packet}; use pnet::packet::ipv4::{self, Ipv4Flags, Ipv4Packet, MutableIpv4Packet};
use std::net::Ipv4Addr; use std::net::Ipv4Addr;
use std::time::{Duration, Instant}; use std::time::Duration;
use crate::common::error::Error; use crate::common::error::Error;
+5 -4
View File
@@ -1018,6 +1018,7 @@ impl TcpProxyRpc for QuicProxyDstRpcService {
mod tests { mod tests {
use super::*; use super::*;
use bytes::Buf; use bytes::Buf;
use hotpath::instant::Instant;
/// Helper function: Create a pair of interconnected QuicSockets. /// Helper function: Create a pair of interconnected QuicSockets.
/// Data sent by socket_a will enter socket_b's rx, and vice versa. /// Data sent by socket_a will enter socket_b's rx, and vice versa.
@@ -1197,7 +1198,7 @@ mod tests {
// Accept unidirectional stream // Accept unidirectional stream
let mut recv = connection.accept_uni().await.unwrap(); let mut recv = connection.accept_uni().await.unwrap();
let start = std::time::Instant::now(); let start = Instant::now();
let mut received = 0; let mut received = 0;
// Loop read until the stream ends // Loop read until the stream ends
@@ -1234,7 +1235,7 @@ mod tests {
let bytes_data = Bytes::from(data_chunk); // Use Bytes to avoid repeated allocation let bytes_data = Bytes::from(data_chunk); // Use Bytes to avoid repeated allocation
println!("Client: Start sending {} MB...", TOTAL_SIZE / 1024 / 1024); println!("Client: Start sending {} MB...", TOTAL_SIZE / 1024 / 1024);
let start_send = std::time::Instant::now(); let start_send = Instant::now();
let chunks = TOTAL_SIZE / CHUNK_SIZE; let chunks = TOTAL_SIZE / CHUNK_SIZE;
for _ in 0..chunks { for _ in 0..chunks {
@@ -1276,7 +1277,7 @@ mod tests {
println!("Server: Accepted connection"); println!("Server: Accepted connection");
let mut stream_handles = Vec::new(); let mut stream_handles = Vec::new();
let start = std::time::Instant::now(); let start = Instant::now();
// Accept an expected number of streams // Accept an expected number of streams
for i in 0..STREAM_COUNT { for i in 0..STREAM_COUNT {
@@ -1346,7 +1347,7 @@ mod tests {
STREAM_COUNT STREAM_COUNT
); );
let start_send = std::time::Instant::now(); let start_send = Instant::now();
let mut client_tasks = Vec::new(); let mut client_tasks = Vec::new();
// Start sending tasks concurrently // Start sending tasks concurrently
+2 -1
View File
@@ -5,10 +5,11 @@ use std::{
Arc, Weak, Arc, Weak,
atomic::{AtomicBool, AtomicUsize, Ordering}, atomic::{AtomicBool, AtomicUsize, Ordering},
}, },
time::{Duration, Instant}, time::Duration,
}; };
use crossbeam::atomic::AtomicCell; use crossbeam::atomic::AtomicCell;
use hotpath::instant::Instant;
#[cfg(feature = "kcp")] #[cfg(feature = "kcp")]
use kcp_sys::{endpoint::KcpEndpoint, stream::KcpStream}; use kcp_sys::{endpoint::KcpEndpoint, stream::KcpStream};
use tokio_util::sync::{CancellationToken, DropGuard}; use tokio_util::sync::{CancellationToken, DropGuard};
+2 -1
View File
@@ -22,11 +22,12 @@ use std::{
atomic::{AtomicUsize, Ordering}, atomic::{AtomicUsize, Ordering},
}, },
task::{Context, Poll}, task::{Context, Poll},
time::{Duration, Instant}, time::Duration,
}; };
use anyhow::Context as _; use anyhow::Context as _;
use dashmap::mapref::entry::Entry; use dashmap::mapref::entry::Entry;
use hotpath::instant::Instant;
use tokio::io::{AsyncRead, AsyncWrite}; use tokio::io::{AsyncRead, AsyncWrite};
use crate::{common::error::Error, gateway::fast_socks5::server::AsyncTcpConnector}; use crate::{common::error::Error, gateway::fast_socks5::server::AsyncTcpConnector};
+2 -1
View File
@@ -3,6 +3,7 @@ use cidr::Ipv4Inet;
use core::panic; use core::panic;
use crossbeam::atomic::AtomicCell; use crossbeam::atomic::AtomicCell;
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use pnet::packet::MutablePacket; use pnet::packet::MutablePacket;
use pnet::packet::Packet; use pnet::packet::Packet;
use pnet::packet::ip::IpNextHeaderProtocols; use pnet::packet::ip::IpNextHeaderProtocols;
@@ -12,7 +13,7 @@ use socket2::{SockRef, TcpKeepalive};
use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4}; use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::atomic::{AtomicBool, AtomicU16}; use std::sync::atomic::{AtomicBool, AtomicU16};
use std::sync::{Arc, Weak}; use std::sync::{Arc, Weak};
use std::time::{Duration, Instant}; use std::time::Duration;
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, copy_bidirectional}; use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, copy_bidirectional};
use tokio::net::{TcpListener, TcpSocket, TcpStream}; use tokio::net::{TcpListener, TcpSocket, TcpStream};
use tokio::sync::{Mutex, mpsc}; use tokio::sync::{Mutex, mpsc};
+6 -5
View File
@@ -8,6 +8,7 @@ use bytes::{BufMut, BytesMut};
use cidr::Ipv4Inet; use cidr::Ipv4Inet;
use crossbeam::atomic::AtomicCell; use crossbeam::atomic::AtomicCell;
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use pnet::packet::{ use pnet::packet::{
Packet, Packet,
ip::IpNextHeaderProtocols, ip::IpNextHeaderProtocols,
@@ -60,8 +61,8 @@ struct UdpNatEntry {
socket: Option<UdpSocket>, socket: Option<UdpSocket>,
forward_task: Mutex<Option<JoinHandle<()>>>, forward_task: Mutex<Option<JoinHandle<()>>>,
stopped: AtomicBool, stopped: AtomicBool,
start_time: std::time::Instant, start_time: Instant,
last_active_time: AtomicCell<std::time::Instant>, last_active_time: AtomicCell<Instant>,
denied: bool, denied: bool,
} }
@@ -85,8 +86,8 @@ impl UdpNatEntry {
socket, socket,
forward_task: Mutex::new(None), forward_task: Mutex::new(None),
stopped: AtomicBool::new(false), stopped: AtomicBool::new(false),
start_time: std::time::Instant::now(), start_time: Instant::now(),
last_active_time: AtomicCell::new(std::time::Instant::now()), last_active_time: AtomicCell::new(Instant::now()),
denied, denied,
}) })
} }
@@ -255,7 +256,7 @@ impl UdpNatEntry {
} }
fn mark_active(&self) { fn mark_active(&self) {
self.last_active_time.store(std::time::Instant::now()); self.last_active_time.store(Instant::now());
} }
fn is_active(&self) -> bool { fn is_active(&self) -> bool {
+1 -1
View File
@@ -1,9 +1,9 @@
use std::collections::BTreeSet; use std::collections::BTreeSet;
use std::sync::{Arc, Weak}; use std::sync::{Arc, Weak};
use std::time::Instant;
use crate::common::global_ctx::{ArcGlobalCtx, GlobalCtxEvent}; use crate::common::global_ctx::{ArcGlobalCtx, GlobalCtxEvent};
use crate::peers::peer_manager::PeerManager; use crate::peers::peer_manager::PeerManager;
use hotpath::instant::Instant;
use tokio_util::task::AbortOnDropHandle; use tokio_util::task::AbortOnDropHandle;
/// ProxyCidrsMonitor monitors changes in proxy CIDRs from peer routes /// ProxyCidrsMonitor monitors changes in proxy CIDRs from peer routes
+15
View File
@@ -14,6 +14,21 @@ extern crate self as hotpath;
#[cfg(not(feature = "hotpath"))] #[cfg(not(feature = "hotpath"))]
mod hotpath_off; mod hotpath_off;
// When the `hotpath` feature is off, expose a local `instant` module backed by
// `quanta::Instant` so call sites can uniformly write `use hotpath::instant::Instant;`
// regardless of whether the feature is enabled. With the feature on, the real
// `hotpath` crate provides the same path (also `quanta::Instant` on Linux), so
// the two modes resolve to the identical type.
#[cfg(not(feature = "hotpath"))]
pub mod instant {
pub type Instant = quanta::Instant;
}
// Re-export `Instant` at the crate root so public APIs that expose it
// (e.g. `Route::get_peer_info_last_update_time`) reference a deliberate
// public type rather than leaking an inaccessible one.
pub use hotpath::instant::Instant;
mod arch; mod arch;
mod gateway; mod gateway;
pub mod instance; pub mod instance;
+3 -2
View File
@@ -1,6 +1,5 @@
use std::net::{Ipv4Addr, Ipv6Addr}; use std::net::{Ipv4Addr, Ipv6Addr};
use std::sync::atomic::Ordering; use std::sync::atomic::Ordering;
use std::time::Instant;
use std::{ use std::{
net::IpAddr, net::IpAddr,
sync::{Arc, atomic::AtomicBool}, sync::{Arc, atomic::AtomicBool},
@@ -8,6 +7,7 @@ use std::{
use arc_swap::ArcSwap; use arc_swap::ArcSwap;
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use pnet::packet::ipv6::Ipv6Packet; use pnet::packet::ipv6::Ipv6Packet;
use pnet::packet::{ use pnet::packet::{
Packet as _, ip::IpNextHeaderProtocols, ipv4::Ipv4Packet, tcp::TcpPacket, udp::UdpPacket, Packet as _, ip::IpNextHeaderProtocols, ipv4::Ipv4Packet, tcp::TcpPacket, udp::UdpPacket,
@@ -402,9 +402,10 @@ mod tests {
use std::{ use std::{
net::{IpAddr, Ipv4Addr, Ipv6Addr}, net::{IpAddr, Ipv4Addr, Ipv6Addr},
sync::Arc, sync::Arc,
time::Instant,
}; };
use hotpath::instant::Instant;
use crate::{ use crate::{
common::acl_processor::PacketInfo, common::acl_processor::PacketInfo,
proto::acl::{Acl, ChainType, Protocol}, proto::acl::{Acl, ChainType, Protocol},
+2 -1
View File
@@ -6,6 +6,7 @@ use std::{
time::Duration, time::Duration,
}; };
use hotpath::instant::Instant;
use rand::{Rng, thread_rng}; use rand::{Rng, thread_rng};
use tokio::{ use tokio::{
sync::broadcast, sync::broadcast,
@@ -177,7 +178,7 @@ impl PeerConnPinger {
sink.send(req).await?; sink.send(req).await?;
control_metrics.record_tx(req_len); control_metrics.record_tx(req_len);
let now = std::time::Instant::now(); let now = Instant::now();
// wait until we get a pong packet in ctrl_resp_receiver // wait until we get a pong packet in ctrl_resp_receiver
let resp = timeout(Duration::from_secs(2), async { let resp = timeout(Duration::from_secs(2), async {
loop { loop {
+5 -7
View File
@@ -2,12 +2,13 @@ use anyhow::Context;
use async_trait::async_trait; use async_trait::async_trait;
use cidr::{Ipv4Cidr, Ipv6Cidr}; use cidr::{Ipv4Cidr, Ipv6Cidr};
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use std::collections::BTreeSet; use std::collections::BTreeSet;
use std::{ use std::{
fmt::Debug, fmt::Debug,
net::{IpAddr, Ipv4Addr, Ipv6Addr}, net::{IpAddr, Ipv4Addr, Ipv6Addr},
sync::{Arc, Weak, atomic::AtomicBool}, sync::{Arc, Weak, atomic::AtomicBool},
time::{Duration, Instant, SystemTime}, time::{Duration, SystemTime},
}; };
#[cfg(feature = "hotpath")] #[cfg(feature = "hotpath")]
@@ -2204,12 +2205,9 @@ impl PeerManager {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use base64::Engine; use base64::Engine;
use std::{ use std::{collections::HashMap, fmt::Debug, sync::Arc, time::Duration};
collections::HashMap,
fmt::Debug, use hotpath::instant::Instant;
sync::Arc,
time::{Duration, Instant},
};
use crate::{ use crate::{
common::{ common::{
+9 -8
View File
@@ -6,13 +6,14 @@ use std::{
Arc, Weak, Arc, Weak,
atomic::{AtomicBool, AtomicU32, Ordering}, atomic::{AtomicBool, AtomicU32, Ordering},
}, },
time::{Duration, Instant, SystemTime}, time::{Duration, SystemTime},
}; };
use arc_swap::ArcSwap; use arc_swap::ArcSwap;
use cidr::{IpCidr, Ipv4Cidr, Ipv6Cidr, Ipv6Inet}; use cidr::{IpCidr, Ipv4Cidr, Ipv6Cidr, Ipv6Inet};
use crossbeam::atomic::AtomicCell; use crossbeam::atomic::AtomicCell;
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use ordered_hash_map::OrderedHashMap; use ordered_hash_map::OrderedHashMap;
use parking_lot::{RwLock, lock_api::RwLockUpgradableReadGuard}; use parking_lot::{RwLock, lock_api::RwLockUpgradableReadGuard};
use petgraph::{ use petgraph::{
@@ -2170,9 +2171,9 @@ struct PeerRouteServiceImpl {
interface_peers_generation: AtomicU64, interface_peers_generation: AtomicU64,
applied_interface_peers_generation: AtomicU64, applied_interface_peers_generation: AtomicU64,
last_update_my_foreign_network: AtomicCell<Option<std::time::Instant>>, last_update_my_foreign_network: AtomicCell<Option<Instant>>,
peer_info_last_update: AtomicCell<std::time::Instant>, peer_info_last_update: AtomicCell<Instant>,
} }
impl Debug for PeerRouteServiceImpl { impl Debug for PeerRouteServiceImpl {
@@ -2239,7 +2240,7 @@ impl PeerRouteServiceImpl {
last_update_my_foreign_network: AtomicCell::new(None), last_update_my_foreign_network: AtomicCell::new(None),
peer_info_last_update: AtomicCell::new(std::time::Instant::now()), peer_info_last_update: AtomicCell::new(Instant::now()),
} }
} }
@@ -2435,7 +2436,7 @@ impl PeerRouteServiceImpl {
} }
self.last_update_my_foreign_network self.last_update_my_foreign_network
.store(Some(std::time::Instant::now())); .store(Some(Instant::now()));
let foreign_networks = self let foreign_networks = self
.interface .interface
@@ -3156,12 +3157,12 @@ impl PeerRouteServiceImpl {
"update_peer_info_last_update, my_peer_id: {:?}, prev: {:?}, new: {:?}", "update_peer_info_last_update, my_peer_id: {:?}, prev: {:?}, new: {:?}",
self.my_peer_id, self.my_peer_id,
self.peer_info_last_update.load(), self.peer_info_last_update.load(),
std::time::Instant::now() Instant::now()
); );
self.peer_info_last_update.store(std::time::Instant::now()); self.peer_info_last_update.store(Instant::now());
} }
fn get_peer_info_last_update(&self) -> std::time::Instant { fn get_peer_info_last_update(&self) -> Instant {
self.peer_info_last_update.load() self.peer_info_last_update.load()
} }
+2 -1
View File
@@ -1,6 +1,7 @@
use std::{sync::Arc, time::Instant}; use std::sync::Arc;
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use prost::Message; use prost::Message;
use snow::params::NoiseParams; use snow::params::NoiseParams;
use tokio::sync::{Mutex, OwnedMutexGuard, oneshot}; use tokio::sync::{Mutex, OwnedMutexGuard, oneshot};
+3 -2
View File
@@ -1,6 +1,7 @@
use cidr::Ipv6Inet; use cidr::Ipv6Inet;
use cidr::{Ipv4Cidr, Ipv6Cidr}; use cidr::{Ipv4Cidr, Ipv6Cidr};
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use std::{ use std::{
collections::BTreeSet, collections::BTreeSet,
net::{Ipv4Addr, Ipv6Addr}, net::{Ipv4Addr, Ipv6Addr},
@@ -157,7 +158,7 @@ pub trait Route {
async fn get_peer_info(&self, peer_id: PeerId) -> Option<RoutePeerInfo>; async fn get_peer_info(&self, peer_id: PeerId) -> Option<RoutePeerInfo>;
async fn get_peer_info_last_update_time(&self) -> std::time::Instant; async fn get_peer_info_last_update_time(&self) -> Instant;
fn get_peer_groups(&self, peer_id: PeerId) -> Arc<Vec<String>>; fn get_peer_groups(&self, peer_id: PeerId) -> Arc<Vec<String>>;
@@ -226,7 +227,7 @@ impl Route for MockRoute {
panic!("mock route") panic!("mock route")
} }
async fn get_peer_info_last_update_time(&self) -> std::time::Instant { async fn get_peer_info_last_update_time(&self) -> Instant {
panic!("mock route") panic!("mock route")
} }
+9 -8
View File
@@ -5,6 +5,7 @@ use std::sync::{Arc, Mutex};
use bytes::Bytes; use bytes::Bytes;
use dashmap::DashMap; use dashmap::DashMap;
use guarden::defer; use guarden::defer;
use hotpath::instant::Instant;
use prost::Message; use prost::Message;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio::task::JoinSet; use tokio::task::JoinSet;
@@ -52,7 +53,7 @@ struct InflightRequestKey {
struct InflightRequest { struct InflightRequest {
sender: RpcPacketSender, sender: RpcPacketSender,
merger: PacketMerger, merger: PacketMerger,
start_time: std::time::Instant, start_time: Instant,
} }
impl std::fmt::Debug for InflightRequest { impl std::fmt::Debug for InflightRequest {
@@ -65,14 +66,14 @@ impl std::fmt::Debug for InflightRequest {
} }
#[derive(Debug, Clone, Default)] #[derive(Debug, Clone, Default)]
pub struct PeerInfo { pub(crate) struct PeerInfo {
pub peer_id: PeerId, pub peer_id: PeerId,
pub compression_info: RpcCompressionInfo, pub compression_info: RpcCompressionInfo,
pub last_active: Option<std::time::Instant>, pub last_active: Option<Instant>,
} }
type InflightRequestTable = Arc<DashMap<InflightRequestKey, InflightRequest>>; type InflightRequestTable = Arc<DashMap<InflightRequestKey, InflightRequest>>;
pub type PeerInfoTable = Arc<DashMap<PeerId, PeerInfo>>; pub(crate) type PeerInfoTable = Arc<DashMap<PeerId, PeerInfo>>;
pub struct Client { pub struct Client {
mpsc: Mutex<MpscTunnel<Box<dyn Tunnel>>>, mpsc: Mutex<MpscTunnel<Box<dyn Tunnel>>>,
@@ -123,7 +124,7 @@ impl Client {
tasks.spawn(async move { tasks.spawn(async move {
loop { loop {
tokio::time::sleep(std::time::Duration::from_secs(30)).await; tokio::time::sleep(std::time::Duration::from_secs(30)).await;
let now = std::time::Instant::now(); let now = Instant::now();
peer_infos.retain(|_, v| { peer_infos.retain(|_, v| {
if let Some(last_active) = v.last_active { if let Some(last_active) = v.last_active {
return now.duration_since(last_active) return now.duration_since(last_active)
@@ -230,7 +231,7 @@ impl Client {
method: <Self::Descriptor as ServiceDescriptor>::Method, method: <Self::Descriptor as ServiceDescriptor>::Method,
input: bytes::Bytes, input: bytes::Bytes,
) -> Result<bytes::Bytes> { ) -> Result<bytes::Bytes> {
let start_time = std::time::Instant::now(); let start_time = Instant::now();
let transaction_id = CUR_TID.fetch_add(1, std::sync::atomic::Ordering::Relaxed); let transaction_id = CUR_TID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let (tx, mut rx) = mpsc::unbounded_channel(); let (tx, mut rx) = mpsc::unbounded_channel();
let key = InflightRequestKey { let key = InflightRequestKey {
@@ -314,7 +315,7 @@ impl Client {
PeerInfo { PeerInfo {
peer_id: self.to_peer_id, peer_id: self.to_peer_id,
compression_info, compression_info,
last_active: Some(std::time::Instant::now()), last_active: Some(Instant::now()),
}, },
); );
@@ -385,7 +386,7 @@ impl Client {
self.inflight_requests.len() self.inflight_requests.len()
} }
pub fn peer_info_table(&self) -> PeerInfoTable { pub(crate) fn peer_info_table(&self) -> PeerInfoTable {
self.peer_info.clone() self.peer_info.clone()
} }
} }
+7 -5
View File
@@ -1,5 +1,7 @@
use prost::{Message as _, length_delimiter_len}; use prost::{Message as _, length_delimiter_len};
use hotpath::instant::Instant;
use crate::{ use crate::{
common::{PeerId, compressor::DefaultCompressor}, common::{PeerId, compressor::DefaultCompressor},
proto::{ proto::{
@@ -42,10 +44,10 @@ pub async fn decompress_packet(
Ok(decompressed) Ok(decompressed)
} }
pub struct PacketMerger { pub(crate) struct PacketMerger {
first_piece: Option<RpcPacket>, first_piece: Option<RpcPacket>,
pieces: Vec<RpcPacket>, pieces: Vec<RpcPacket>,
last_updated: std::time::Instant, last_updated: Instant,
} }
impl Default for PacketMerger { impl Default for PacketMerger {
@@ -59,7 +61,7 @@ impl PacketMerger {
Self { Self {
first_piece: None, first_piece: None,
pieces: Vec::new(), pieces: Vec::new(),
last_updated: std::time::Instant::now(), last_updated: Instant::now(),
} }
} }
@@ -132,12 +134,12 @@ impl PacketMerger {
.resize(total_pieces as usize, Default::default()); .resize(total_pieces as usize, Default::default());
self.pieces[piece_idx as usize] = rpc_packet; self.pieces[piece_idx as usize] = rpc_packet;
self.last_updated = std::time::Instant::now(); self.last_updated = Instant::now();
Ok(self.try_merge_pieces()) Ok(self.try_merge_pieces())
} }
pub fn last_updated(&self) -> std::time::Instant { pub(crate) fn last_updated(&self) -> Instant {
self.last_updated self.last_updated
} }
} }
+2 -1
View File
@@ -5,6 +5,7 @@ use std::{
use bytes::Bytes; use bytes::Bytes;
use dashmap::DashMap; use dashmap::DashMap;
use hotpath::instant::Instant;
use prost::Message; use prost::Message;
use tokio::{task::JoinSet, time::timeout}; use tokio::{task::JoinSet, time::timeout};
use tokio_stream::StreamExt; use tokio_stream::StreamExt;
@@ -233,7 +234,7 @@ impl Server {
} }
let mut resp_msg = RpcResponse::default(); let mut resp_msg = RpcResponse::default();
let now = std::time::Instant::now(); let now = Instant::now();
let compression_info = packet.compression_info; let compression_info = packet.compression_info;
let resp_bytes = Self::handle_rpc_request(packet, reg, tunnel_info).await; let resp_bytes = Self::handle_rpc_request(packet, reg, tunnel_info).await;
+5 -3
View File
@@ -6,6 +6,8 @@ use std::{
time::Duration, time::Duration,
}; };
use hotpath::instant::Instant;
use super::{ use super::{
FromUrl, IpVersion, Tunnel, TunnelError, TunnelInfo, TunnelListener, TunnelUrl, ZCPacketSink, FromUrl, IpVersion, Tunnel, TunnelError, TunnelInfo, TunnelListener, TunnelUrl, ZCPacketSink,
ZCPacketStream, ZCPacketStream,
@@ -346,7 +348,7 @@ struct WgPeer {
data: Option<WgPeerData>, data: Option<WgPeerData>,
tasks: JoinSet<()>, tasks: JoinSet<()>,
access_time: AtomicCell<std::time::Instant>, access_time: AtomicCell<Instant>,
} }
impl WgPeer { impl WgPeer {
@@ -369,7 +371,7 @@ impl WgPeer {
data: None, data: None,
tasks: JoinSet::new(), tasks: JoinSet::new(),
access_time: AtomicCell::new(std::time::Instant::now()), access_time: AtomicCell::new(Instant::now()),
} }
} }
@@ -385,7 +387,7 @@ impl WgPeer {
} }
async fn handle_packet_from_peer(&self, packet: &[u8]) { async fn handle_packet_from_peer(&self, packet: &[u8]) {
self.access_time.store(std::time::Instant::now()); self.access_time.store(Instant::now());
tracing::trace!("Received {} bytes from peer", packet.len()); tracing::trace!("Received {} bytes from peer", packet.len());
let data = self.data.as_ref().unwrap(); let data = self.data.as_ref().unwrap();
// TODO: improve this // TODO: improve this