diff --git a/Cargo.lock b/Cargo.lock index 9381ec07..4c455e6b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2362,6 +2362,7 @@ dependencies = [ "prost-reflect", "prost-reflect-build", "prost-wkt-types", + "quanta", "quinn", "quinn-proto", "quote", diff --git a/easytier/Cargo.toml b/easytier/Cargo.toml index c61cf02b..6e483d61 100644 --- a/easytier/Cargo.toml +++ b/easytier/Cargo.toml @@ -53,6 +53,7 @@ chrono = { version = "0.4.37", features = ["serde"] } guarden = "0.2" hotpath = { version = "0.18", default-features = false, optional = true } +quanta = "0.12" delegate = "0.13.5" diff --git a/easytier/src/common/acl_processor.rs b/easytier/src/common/acl_processor.rs index 4a414603..bfd364a0 100644 --- a/easytier/src/common/acl_processor.rs +++ b/easytier/src/common/acl_processor.rs @@ -3,9 +3,11 @@ use std::{ net::{IpAddr, SocketAddr}, str::FromStr as _, 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::proto::acl::*; use anyhow::Context as _; @@ -107,10 +109,10 @@ impl AclCacheKey { // Cache entry with timestamp for LRU cleanup #[derive(Debug, Clone)] -pub struct AclCacheEntry { +pub(crate) struct AclCacheEntry { pub action: Action, pub matched_rule: RuleId, - pub last_access: std::time::Instant, + pub last_access: Instant, // New fields to track rule characteristics for proper cache behavior pub conn_track_key: Option, pub rate_limit_keys: Vec, @@ -410,7 +412,7 @@ impl AclProcessor { } // Remove oldest entries (LRU cleanup) - let mut entries: Vec<(AclCacheKey, std::time::Instant)> = cache + let mut entries: Vec<(AclCacheKey, Instant)> = cache .iter() .map(|entry| (entry.key().clone(), entry.value().last_access)) .collect(); @@ -431,7 +433,7 @@ impl AclProcessor { ); } - pub fn process_packet_with_cache_entry( + pub(crate) fn process_packet_with_cache_entry( &self, packet_info: &PacketInfo, cache_entry: &AclCacheEntry, diff --git a/easytier/src/common/stats_manager.rs b/easytier/src/common/stats_manager.rs index ecbde860..5d344e1a 100644 --- a/easytier/src/common/stats_manager.rs +++ b/easytier/src/common/stats_manager.rs @@ -1,9 +1,10 @@ use dashmap::DashMap; +use hotpath::instant::Instant; use serde::{Deserialize, Serialize}; use std::cell::UnsafeCell; use std::fmt; use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::Duration; use tokio::time::interval; use tokio_util::task::AbortOnDropHandle; diff --git a/easytier/src/common/stun.rs b/easytier/src/common/stun.rs index 008a63f8..925e820b 100644 --- a/easytier/src/common/stun.rs +++ b/easytier/src/common/stun.rs @@ -2,12 +2,13 @@ use std::collections::BTreeSet; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::atomic::AtomicBool; use std::sync::{Arc, RwLock}; -use std::time::{Duration, Instant}; +use std::time::Duration; use crate::proto::common::{NatType, StunInfo}; use anyhow::Context; use chrono::Local; use crossbeam::atomic::AtomicCell; +use hotpath::instant::Instant; use rand::seq::IteratorRandom; use socket2::{SockAddr, SockRef}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -1312,7 +1313,7 @@ impl StunInfoCollectorTrait for MockStunInfoCollector { StunInfo { udp_nat_type: self.udp_nat_type 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, max_port: 200, public_ip: vec!["127.0.0.1".to_string(), "::1".to_string()], diff --git a/easytier/src/connector/direct.rs b/easytier/src/connector/direct.rs index e6296f11..dbde5223 100644 --- a/easytier/src/connector/direct.rs +++ b/easytier/src/connector/direct.rs @@ -8,9 +8,11 @@ use std::{ Arc, atomic::{AtomicBool, Ordering}, }, - time::{Duration, Instant}, + time::Duration, }; +use hotpath::instant::Instant; + use crate::{ common::{ PeerId, dns::socket_addrs, error::Error, global_ctx::ArcGlobalCtx, diff --git a/easytier/src/connector/manual.rs b/easytier/src/connector/manual.rs index c797c50c..c70db5fc 100644 --- a/easytier/src/connector/manual.rs +++ b/easytier/src/connector/manual.rs @@ -2,10 +2,11 @@ use std::{ collections::BTreeSet, future::Future, sync::{Arc, Weak}, - time::{Duration, Instant}, + time::Duration, }; use dashmap::DashSet; +use hotpath::instant::Instant; use tokio::{sync::mpsc, task::JoinSet, time::timeout}; use crate::{ diff --git a/easytier/src/connector/tcp_hole_punch.rs b/easytier/src/connector/tcp_hole_punch.rs index 9baaeaa0..4945f1b4 100644 --- a/easytier/src/connector/tcp_hole_punch.rs +++ b/easytier/src/connector/tcp_hole_punch.rs @@ -1,10 +1,11 @@ use std::{ net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, sync::Arc, - time::{Duration, Instant}, + time::Duration, }; use anyhow::{Context, Error}; +use hotpath::instant::Instant; use rand::Rng as _; use tokio::task::JoinSet; diff --git a/easytier/src/connector/udp_hole_punch/both_easy_sym.rs b/easytier/src/connector/udp_hole_punch/both_easy_sym.rs index 03f30723..e96d9dba 100644 --- a/easytier/src/connector/udp_hole_punch/both_easy_sym.rs +++ b/easytier/src/connector/udp_hole_punch/both_easy_sym.rs @@ -1,10 +1,11 @@ use std::{ net::{IpAddr, SocketAddr, SocketAddrV4}, sync::Arc, - time::{Duration, Instant}, + time::Duration, }; use anyhow::Context; +use hotpath::instant::Instant; use tokio::sync::Mutex; use tokio_util::task::AbortOnDropHandle; diff --git a/easytier/src/connector/udp_hole_punch/common.rs b/easytier/src/connector/udp_hole_punch/common.rs index e82b0663..15e1d60f 100644 --- a/easytier/src/connector/udp_hole_punch/common.rs +++ b/easytier/src/connector/udp_hole_punch/common.rs @@ -7,6 +7,7 @@ use std::{ use crossbeam::atomic::AtomicCell; use dashmap::{DashMap, DashSet}; use guarden::defer; +use hotpath::instant::Instant; use rand::seq::SliceRandom as _; use tokio::{net::UdpSocket, sync::Mutex, task::JoinSet}; use tracing::{Instrument, Level, instrument}; @@ -356,9 +357,9 @@ pub(crate) struct UdpHolePunchListener { _port_mapping_lease: Option, conn_counter: Arc>, - listen_time: std::time::Instant, - last_select_time: AtomicCell, - last_active_time: Arc>, + listen_time: Instant, + last_select_time: AtomicCell, + last_active_time: Arc>, } impl UdpHolePunchListener { @@ -421,14 +422,14 @@ impl UdpHolePunchListener { 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 last_active_time_clone = last_active_time.clone(); tasks.spawn(async move { loop { tokio::time::sleep(std::time::Duration::from_secs(5)).await; 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, conn_counter, - listen_time: std::time::Instant::now(), - last_select_time: AtomicCell::new(std::time::Instant::now()), + listen_time: Instant::now(), + last_select_time: AtomicCell::new(Instant::now()), last_active_time, }) } pub async fn get_socket(&self) -> Arc { - self.last_select_time.store(std::time::Instant::now()); + self.last_select_time.store(Instant::now()); self.socket.clone() } diff --git a/easytier/src/connector/udp_hole_punch/cone.rs b/easytier/src/connector/udp_hole_punch/cone.rs index bb738e09..da2838dd 100644 --- a/easytier/src/connector/udp_hole_punch/cone.rs +++ b/easytier/src/connector/udp_hole_punch/cone.rs @@ -1,9 +1,7 @@ -use std::{ - sync::Arc, - time::{Duration, Instant}, -}; +use std::{sync::Arc, time::Duration}; use anyhow::Context; +use hotpath::instant::Instant; use tokio::net::UdpSocket; use tokio_util::task::AbortOnDropHandle; diff --git a/easytier/src/connector/udp_hole_punch/mod.rs b/easytier/src/connector/udp_hole_punch/mod.rs index e1c2e14b..90d90e24 100644 --- a/easytier/src/connector/udp_hole_punch/mod.rs +++ b/easytier/src/connector/udp_hole_punch/mod.rs @@ -1,6 +1,6 @@ use std::{ sync::{Arc, atomic::AtomicBool}, - time::{Duration, Instant}, + time::Duration, }; use anyhow::{Context, Error}; @@ -8,6 +8,7 @@ use both_easy_sym::{PunchBothEasySymHoleClient, PunchBothEasySymHoleServer}; use common::{PunchHoleServerCommon, UdpNatType, UdpPunchClientMethod}; use cone::{PunchConeHoleClient, PunchConeHoleServer}; use dashmap::DashMap; +use hotpath::instant::Instant; use once_cell::sync::Lazy; use sym_to_cone::{PunchSymToConeHoleClient, PunchSymToConeHoleServer}; use tokio::{sync::Mutex, task::JoinHandle}; diff --git a/easytier/src/connector/udp_hole_punch/sym_to_cone.rs b/easytier/src/connector/udp_hole_punch/sym_to_cone.rs index b527a508..406a6fec 100644 --- a/easytier/src/connector/udp_hole_punch/sym_to_cone.rs +++ b/easytier/src/connector/udp_hole_punch/sym_to_cone.rs @@ -5,11 +5,12 @@ use std::{ Arc, atomic::{AtomicBool, Ordering}, }, - time::{Duration, Instant}, + time::Duration, }; use anyhow::Context; use guarden::defer; +use hotpath::instant::Instant; use rand::{Rng, seq::SliceRandom}; use tokio::{net::UdpSocket, sync::RwLock}; use tokio_util::task::AbortOnDropHandle; diff --git a/easytier/src/gateway/icmp_proxy.rs b/easytier/src/gateway/icmp_proxy.rs index 0e0182d2..20fa6efb 100644 --- a/easytier/src/gateway/icmp_proxy.rs +++ b/easytier/src/gateway/icmp_proxy.rs @@ -7,6 +7,7 @@ use std::{ }; use anyhow::Context; +use hotpath::instant::Instant; use pnet::packet::{ Packet, icmp::{self, IcmpCode, IcmpTypes, MutableIcmpPacket, echo_reply::MutableEchoReplyPacket}, @@ -45,7 +46,7 @@ struct IcmpNatEntry { src_peer_id: PeerId, my_peer_id: PeerId, src_ip: IpAddr, - start_time: std::time::Instant, + start_time: Instant, mapped_dst_ip: std::net::Ipv4Addr, } @@ -60,7 +61,7 @@ impl IcmpNatEntry { src_peer_id, my_peer_id, src_ip, - start_time: std::time::Instant::now(), + start_time: Instant::now(), mapped_dst_ip, }) } diff --git a/easytier/src/gateway/ip_reassembler.rs b/easytier/src/gateway/ip_reassembler.rs index 543429aa..d0ad8f7f 100644 --- a/easytier/src/gateway/ip_reassembler.rs +++ b/easytier/src/gateway/ip_reassembler.rs @@ -1,9 +1,10 @@ use dashmap::DashMap; +use hotpath::instant::Instant; use pnet::packet::Packet; use pnet::packet::ip::IpNextHeaderProtocol; use pnet::packet::ipv4::{self, Ipv4Flags, Ipv4Packet, MutableIpv4Packet}; use std::net::Ipv4Addr; -use std::time::{Duration, Instant}; +use std::time::Duration; use crate::common::error::Error; diff --git a/easytier/src/gateway/quic_proxy.rs b/easytier/src/gateway/quic_proxy.rs index 9414458e..6fc03192 100644 --- a/easytier/src/gateway/quic_proxy.rs +++ b/easytier/src/gateway/quic_proxy.rs @@ -1018,6 +1018,7 @@ impl TcpProxyRpc for QuicProxyDstRpcService { mod tests { use super::*; use bytes::Buf; + use hotpath::instant::Instant; /// Helper function: Create a pair of interconnected QuicSockets. /// Data sent by socket_a will enter socket_b's rx, and vice versa. @@ -1197,7 +1198,7 @@ mod tests { // Accept unidirectional stream let mut recv = connection.accept_uni().await.unwrap(); - let start = std::time::Instant::now(); + let start = Instant::now(); let mut received = 0; // 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 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; for _ in 0..chunks { @@ -1276,7 +1277,7 @@ mod tests { println!("Server: Accepted connection"); let mut stream_handles = Vec::new(); - let start = std::time::Instant::now(); + let start = Instant::now(); // Accept an expected number of streams for i in 0..STREAM_COUNT { @@ -1346,7 +1347,7 @@ mod tests { STREAM_COUNT ); - let start_send = std::time::Instant::now(); + let start_send = Instant::now(); let mut client_tasks = Vec::new(); // Start sending tasks concurrently diff --git a/easytier/src/gateway/socks5.rs b/easytier/src/gateway/socks5.rs index fe98fd0b..3d92d3e2 100644 --- a/easytier/src/gateway/socks5.rs +++ b/easytier/src/gateway/socks5.rs @@ -5,10 +5,11 @@ use std::{ Arc, Weak, atomic::{AtomicBool, AtomicUsize, Ordering}, }, - time::{Duration, Instant}, + time::Duration, }; use crossbeam::atomic::AtomicCell; +use hotpath::instant::Instant; #[cfg(feature = "kcp")] use kcp_sys::{endpoint::KcpEndpoint, stream::KcpStream}; use tokio_util::sync::{CancellationToken, DropGuard}; diff --git a/easytier/src/gateway/socks5/dataplane.rs b/easytier/src/gateway/socks5/dataplane.rs index 5dc4aacd..ca895a40 100644 --- a/easytier/src/gateway/socks5/dataplane.rs +++ b/easytier/src/gateway/socks5/dataplane.rs @@ -22,11 +22,12 @@ use std::{ atomic::{AtomicUsize, Ordering}, }, task::{Context, Poll}, - time::{Duration, Instant}, + time::Duration, }; use anyhow::Context as _; use dashmap::mapref::entry::Entry; +use hotpath::instant::Instant; use tokio::io::{AsyncRead, AsyncWrite}; use crate::{common::error::Error, gateway::fast_socks5::server::AsyncTcpConnector}; diff --git a/easytier/src/gateway/tcp_proxy.rs b/easytier/src/gateway/tcp_proxy.rs index 6e252268..39d86505 100644 --- a/easytier/src/gateway/tcp_proxy.rs +++ b/easytier/src/gateway/tcp_proxy.rs @@ -3,6 +3,7 @@ use cidr::Ipv4Inet; use core::panic; use crossbeam::atomic::AtomicCell; use dashmap::DashMap; +use hotpath::instant::Instant; use pnet::packet::MutablePacket; use pnet::packet::Packet; use pnet::packet::ip::IpNextHeaderProtocols; @@ -12,7 +13,7 @@ use socket2::{SockRef, TcpKeepalive}; use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4}; use std::sync::atomic::{AtomicBool, AtomicU16}; use std::sync::{Arc, Weak}; -use std::time::{Duration, Instant}; +use std::time::Duration; use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, copy_bidirectional}; use tokio::net::{TcpListener, TcpSocket, TcpStream}; use tokio::sync::{Mutex, mpsc}; diff --git a/easytier/src/gateway/udp_proxy.rs b/easytier/src/gateway/udp_proxy.rs index c4cab103..bd9bca05 100644 --- a/easytier/src/gateway/udp_proxy.rs +++ b/easytier/src/gateway/udp_proxy.rs @@ -8,6 +8,7 @@ use bytes::{BufMut, BytesMut}; use cidr::Ipv4Inet; use crossbeam::atomic::AtomicCell; use dashmap::DashMap; +use hotpath::instant::Instant; use pnet::packet::{ Packet, ip::IpNextHeaderProtocols, @@ -60,8 +61,8 @@ struct UdpNatEntry { socket: Option, forward_task: Mutex>>, stopped: AtomicBool, - start_time: std::time::Instant, - last_active_time: AtomicCell, + start_time: Instant, + last_active_time: AtomicCell, denied: bool, } @@ -85,8 +86,8 @@ impl UdpNatEntry { socket, forward_task: Mutex::new(None), stopped: AtomicBool::new(false), - start_time: std::time::Instant::now(), - last_active_time: AtomicCell::new(std::time::Instant::now()), + start_time: Instant::now(), + last_active_time: AtomicCell::new(Instant::now()), denied, }) } @@ -255,7 +256,7 @@ impl UdpNatEntry { } 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 { diff --git a/easytier/src/instance/proxy_cidrs_monitor.rs b/easytier/src/instance/proxy_cidrs_monitor.rs index 2efbf915..c9782d46 100644 --- a/easytier/src/instance/proxy_cidrs_monitor.rs +++ b/easytier/src/instance/proxy_cidrs_monitor.rs @@ -1,9 +1,9 @@ use std::collections::BTreeSet; use std::sync::{Arc, Weak}; -use std::time::Instant; use crate::common::global_ctx::{ArcGlobalCtx, GlobalCtxEvent}; use crate::peers::peer_manager::PeerManager; +use hotpath::instant::Instant; use tokio_util::task::AbortOnDropHandle; /// ProxyCidrsMonitor monitors changes in proxy CIDRs from peer routes diff --git a/easytier/src/lib.rs b/easytier/src/lib.rs index 2d8f6c7b..8a95e5e7 100644 --- a/easytier/src/lib.rs +++ b/easytier/src/lib.rs @@ -14,6 +14,21 @@ extern crate self as hotpath; #[cfg(not(feature = "hotpath"))] 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 gateway; pub mod instance; diff --git a/easytier/src/peers/acl_filter.rs b/easytier/src/peers/acl_filter.rs index 58446206..bfcc55ef 100644 --- a/easytier/src/peers/acl_filter.rs +++ b/easytier/src/peers/acl_filter.rs @@ -1,6 +1,5 @@ use std::net::{Ipv4Addr, Ipv6Addr}; use std::sync::atomic::Ordering; -use std::time::Instant; use std::{ net::IpAddr, sync::{Arc, atomic::AtomicBool}, @@ -8,6 +7,7 @@ use std::{ use arc_swap::ArcSwap; use dashmap::DashMap; +use hotpath::instant::Instant; use pnet::packet::ipv6::Ipv6Packet; use pnet::packet::{ Packet as _, ip::IpNextHeaderProtocols, ipv4::Ipv4Packet, tcp::TcpPacket, udp::UdpPacket, @@ -402,9 +402,10 @@ mod tests { use std::{ net::{IpAddr, Ipv4Addr, Ipv6Addr}, sync::Arc, - time::Instant, }; + use hotpath::instant::Instant; + use crate::{ common::acl_processor::PacketInfo, proto::acl::{Acl, ChainType, Protocol}, diff --git a/easytier/src/peers/peer_conn_ping.rs b/easytier/src/peers/peer_conn_ping.rs index 1a3cde30..bc64ea86 100644 --- a/easytier/src/peers/peer_conn_ping.rs +++ b/easytier/src/peers/peer_conn_ping.rs @@ -6,6 +6,7 @@ use std::{ time::Duration, }; +use hotpath::instant::Instant; use rand::{Rng, thread_rng}; use tokio::{ sync::broadcast, @@ -177,7 +178,7 @@ impl PeerConnPinger { sink.send(req).await?; 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 let resp = timeout(Duration::from_secs(2), async { loop { diff --git a/easytier/src/peers/peer_manager.rs b/easytier/src/peers/peer_manager.rs index f7028962..aa716e00 100644 --- a/easytier/src/peers/peer_manager.rs +++ b/easytier/src/peers/peer_manager.rs @@ -2,12 +2,13 @@ use anyhow::Context; use async_trait::async_trait; use cidr::{Ipv4Cidr, Ipv6Cidr}; use dashmap::DashMap; +use hotpath::instant::Instant; use std::collections::BTreeSet; use std::{ fmt::Debug, net::{IpAddr, Ipv4Addr, Ipv6Addr}, sync::{Arc, Weak, atomic::AtomicBool}, - time::{Duration, Instant, SystemTime}, + time::{Duration, SystemTime}, }; #[cfg(feature = "hotpath")] @@ -2204,12 +2205,9 @@ impl PeerManager { #[cfg(test)] mod tests { use base64::Engine; - use std::{ - collections::HashMap, - fmt::Debug, - sync::Arc, - time::{Duration, Instant}, - }; + use std::{collections::HashMap, fmt::Debug, sync::Arc, time::Duration}; + + use hotpath::instant::Instant; use crate::{ common::{ diff --git a/easytier/src/peers/peer_ospf_route.rs b/easytier/src/peers/peer_ospf_route.rs index 7c5916d5..10aa1ac6 100644 --- a/easytier/src/peers/peer_ospf_route.rs +++ b/easytier/src/peers/peer_ospf_route.rs @@ -6,13 +6,14 @@ use std::{ Arc, Weak, atomic::{AtomicBool, AtomicU32, Ordering}, }, - time::{Duration, Instant, SystemTime}, + time::{Duration, SystemTime}, }; use arc_swap::ArcSwap; use cidr::{IpCidr, Ipv4Cidr, Ipv6Cidr, Ipv6Inet}; use crossbeam::atomic::AtomicCell; use dashmap::DashMap; +use hotpath::instant::Instant; use ordered_hash_map::OrderedHashMap; use parking_lot::{RwLock, lock_api::RwLockUpgradableReadGuard}; use petgraph::{ @@ -2170,9 +2171,9 @@ struct PeerRouteServiceImpl { interface_peers_generation: AtomicU64, applied_interface_peers_generation: AtomicU64, - last_update_my_foreign_network: AtomicCell>, + last_update_my_foreign_network: AtomicCell>, - peer_info_last_update: AtomicCell, + peer_info_last_update: AtomicCell, } impl Debug for PeerRouteServiceImpl { @@ -2239,7 +2240,7 @@ impl PeerRouteServiceImpl { 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 - .store(Some(std::time::Instant::now())); + .store(Some(Instant::now())); let foreign_networks = self .interface @@ -3156,12 +3157,12 @@ impl PeerRouteServiceImpl { "update_peer_info_last_update, my_peer_id: {:?}, prev: {:?}, new: {:?}", self.my_peer_id, 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() } diff --git a/easytier/src/peers/relay_peer_map.rs b/easytier/src/peers/relay_peer_map.rs index db7dd855..984cd684 100644 --- a/easytier/src/peers/relay_peer_map.rs +++ b/easytier/src/peers/relay_peer_map.rs @@ -1,6 +1,7 @@ -use std::{sync::Arc, time::Instant}; +use std::sync::Arc; use dashmap::DashMap; +use hotpath::instant::Instant; use prost::Message; use snow::params::NoiseParams; use tokio::sync::{Mutex, OwnedMutexGuard, oneshot}; diff --git a/easytier/src/peers/route_trait.rs b/easytier/src/peers/route_trait.rs index e6427ced..3f70a2e0 100644 --- a/easytier/src/peers/route_trait.rs +++ b/easytier/src/peers/route_trait.rs @@ -1,6 +1,7 @@ use cidr::Ipv6Inet; use cidr::{Ipv4Cidr, Ipv6Cidr}; use dashmap::DashMap; +use hotpath::instant::Instant; use std::{ collections::BTreeSet, net::{Ipv4Addr, Ipv6Addr}, @@ -157,7 +158,7 @@ pub trait Route { async fn get_peer_info(&self, peer_id: PeerId) -> Option; - 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>; @@ -226,7 +227,7 @@ impl Route for MockRoute { 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") } diff --git a/easytier/src/proto/rpc_impl/client.rs b/easytier/src/proto/rpc_impl/client.rs index d7960e41..2ffaef1a 100644 --- a/easytier/src/proto/rpc_impl/client.rs +++ b/easytier/src/proto/rpc_impl/client.rs @@ -5,6 +5,7 @@ use std::sync::{Arc, Mutex}; use bytes::Bytes; use dashmap::DashMap; use guarden::defer; +use hotpath::instant::Instant; use prost::Message; use tokio::sync::mpsc; use tokio::task::JoinSet; @@ -52,7 +53,7 @@ struct InflightRequestKey { struct InflightRequest { sender: RpcPacketSender, merger: PacketMerger, - start_time: std::time::Instant, + start_time: Instant, } impl std::fmt::Debug for InflightRequest { @@ -65,14 +66,14 @@ impl std::fmt::Debug for InflightRequest { } #[derive(Debug, Clone, Default)] -pub struct PeerInfo { +pub(crate) struct PeerInfo { pub peer_id: PeerId, pub compression_info: RpcCompressionInfo, - pub last_active: Option, + pub last_active: Option, } type InflightRequestTable = Arc>; -pub type PeerInfoTable = Arc>; +pub(crate) type PeerInfoTable = Arc>; pub struct Client { mpsc: Mutex>>, @@ -123,7 +124,7 @@ impl Client { tasks.spawn(async move { loop { tokio::time::sleep(std::time::Duration::from_secs(30)).await; - let now = std::time::Instant::now(); + let now = Instant::now(); peer_infos.retain(|_, v| { if let Some(last_active) = v.last_active { return now.duration_since(last_active) @@ -230,7 +231,7 @@ impl Client { method: ::Method, input: bytes::Bytes, ) -> Result { - 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 (tx, mut rx) = mpsc::unbounded_channel(); let key = InflightRequestKey { @@ -314,7 +315,7 @@ impl Client { PeerInfo { peer_id: self.to_peer_id, compression_info, - last_active: Some(std::time::Instant::now()), + last_active: Some(Instant::now()), }, ); @@ -385,7 +386,7 @@ impl Client { self.inflight_requests.len() } - pub fn peer_info_table(&self) -> PeerInfoTable { + pub(crate) fn peer_info_table(&self) -> PeerInfoTable { self.peer_info.clone() } } diff --git a/easytier/src/proto/rpc_impl/packet.rs b/easytier/src/proto/rpc_impl/packet.rs index a03085df..c54978f0 100644 --- a/easytier/src/proto/rpc_impl/packet.rs +++ b/easytier/src/proto/rpc_impl/packet.rs @@ -1,5 +1,7 @@ use prost::{Message as _, length_delimiter_len}; +use hotpath::instant::Instant; + use crate::{ common::{PeerId, compressor::DefaultCompressor}, proto::{ @@ -42,10 +44,10 @@ pub async fn decompress_packet( Ok(decompressed) } -pub struct PacketMerger { +pub(crate) struct PacketMerger { first_piece: Option, pieces: Vec, - last_updated: std::time::Instant, + last_updated: Instant, } impl Default for PacketMerger { @@ -59,7 +61,7 @@ impl PacketMerger { Self { first_piece: None, 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()); 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()) } - pub fn last_updated(&self) -> std::time::Instant { + pub(crate) fn last_updated(&self) -> Instant { self.last_updated } } diff --git a/easytier/src/proto/rpc_impl/server.rs b/easytier/src/proto/rpc_impl/server.rs index 4b251202..cde46f6d 100644 --- a/easytier/src/proto/rpc_impl/server.rs +++ b/easytier/src/proto/rpc_impl/server.rs @@ -5,6 +5,7 @@ use std::{ use bytes::Bytes; use dashmap::DashMap; +use hotpath::instant::Instant; use prost::Message; use tokio::{task::JoinSet, time::timeout}; use tokio_stream::StreamExt; @@ -233,7 +234,7 @@ impl Server { } let mut resp_msg = RpcResponse::default(); - let now = std::time::Instant::now(); + let now = Instant::now(); let compression_info = packet.compression_info; let resp_bytes = Self::handle_rpc_request(packet, reg, tunnel_info).await; diff --git a/easytier/src/tunnel/wireguard.rs b/easytier/src/tunnel/wireguard.rs index 443fe2f2..57a5b72f 100644 --- a/easytier/src/tunnel/wireguard.rs +++ b/easytier/src/tunnel/wireguard.rs @@ -6,6 +6,8 @@ use std::{ time::Duration, }; +use hotpath::instant::Instant; + use super::{ FromUrl, IpVersion, Tunnel, TunnelError, TunnelInfo, TunnelListener, TunnelUrl, ZCPacketSink, ZCPacketStream, @@ -346,7 +348,7 @@ struct WgPeer { data: Option, tasks: JoinSet<()>, - access_time: AtomicCell, + access_time: AtomicCell, } impl WgPeer { @@ -369,7 +371,7 @@ impl WgPeer { data: None, 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]) { - self.access_time.store(std::time::Instant::now()); + self.access_time.store(Instant::now()); tracing::trace!("Received {} bytes from peer", packet.len()); let data = self.data.as_ref().unwrap(); // TODO: improve this