diff --git a/easytier-core/src/foundation/mod.rs b/easytier-core/src/foundation/mod.rs index e1293aed..5683448b 100644 --- a/easytier-core/src/foundation/mod.rs +++ b/easytier-core/src/foundation/mod.rs @@ -14,3 +14,15 @@ pub mod stats; pub(crate) mod task; pub(crate) mod time; pub(crate) mod token_bucket; + +// Each network and peer owns several small maps. Cap their shard arrays to +// avoid multiplying empty-map overhead on public servers, while retaining +// DashMap's smaller default on single-CPU machines. +pub(crate) fn small_map_shard_amount() -> usize { + static SHARDS: std::sync::LazyLock = std::sync::LazyLock::new(|| { + (std::thread::available_parallelism().map_or(1, usize::from) * 4) + .next_power_of_two() + .min(8) + }); + *SHARDS +} diff --git a/easytier-core/src/peers/conn/peer.rs b/easytier-core/src/peers/conn/peer.rs index d60fd693..aadc9b76 100644 --- a/easytier-core/src/peers/conn/peer.rs +++ b/easytier-core/src/peers/conn/peer.rs @@ -17,6 +17,7 @@ use crate::peers::{ }; use crate::{ config::PeerId, + foundation::small_map_shard_amount, packet::ZCPacket, peers::error::Error, proto::{core_peer::peer::PeerConnInfo, peer_rpc::PeerIdentityType}, @@ -57,7 +58,7 @@ impl Peer { packet_recv_chan: PacketRecvChan, context: ArcPeerContext, ) -> Self { - let conns: ConnMap = Arc::new(DashMap::new()); + let conns: ConnMap = Arc::new(DashMap::with_shard_amount(small_map_shard_amount())); let (close_event_sender, mut close_event_receiver) = mpsc::channel(10); let shutdown_notifier = Arc::new(tokio::sync::Notify::new()); let peer_identity_type = Arc::new(AtomicCell::new(None)); diff --git a/easytier-core/src/peers/conn/peer_map.rs b/easytier-core/src/peers/conn/peer_map.rs index 00a6432e..bc1cab6a 100644 --- a/easytier-core/src/peers/conn/peer_map.rs +++ b/easytier-core/src/peers/conn/peer_map.rs @@ -11,6 +11,7 @@ use tokio::sync::RwLock; use crate::{ config::PeerId, + foundation::small_map_shard_amount, packet::ZCPacket, peers::{ context::{ArcPeerContext, NetworkIdentity, PeerEvent}, @@ -53,7 +54,7 @@ impl PeerMap { PeerMap { context, my_peer_id, - peer_map: DashMap::new(), + peer_map: DashMap::with_shard_amount(small_map_shard_amount()), packet_send, routes: RwLock::new(Vec::new()), alive_client_urls: Arc::new(Mutex::new(HashMap::new())), diff --git a/easytier-core/src/peers/context.rs b/easytier-core/src/peers/context.rs index d7b892e5..7e9bf51d 100644 --- a/easytier-core/src/peers/context.rs +++ b/easytier-core/src/peers/context.rs @@ -28,6 +28,7 @@ use crate::{ RouteConfig, TrafficConfig, }, events::{CoreEvent, CoreEventSink}, + foundation::small_map_shard_amount, foundation::stats::{LabelSet, LabelType, MetricName, StatsManager}, foundation::token_bucket::{ArcByteLimiter, BucketConfig, TokenBucketManager}, peers::{ @@ -438,7 +439,7 @@ pub(crate) struct TrustedKeyMapManager { impl TrustedKeyMapManager { pub fn new() -> Self { Self { - network_trusted_keys: DashMap::new(), + network_trusted_keys: DashMap::with_shard_amount(small_map_shard_amount()), } } diff --git a/easytier-core/src/peers/foreign_network/mod.rs b/easytier-core/src/peers/foreign_network/mod.rs index 34e703f7..83f6bc94 100644 --- a/easytier-core/src/peers/foreign_network/mod.rs +++ b/easytier-core/src/peers/foreign_network/mod.rs @@ -5,12 +5,13 @@ pub(crate) mod client; use std::sync::atomic::{AtomicBool, Ordering}; use std::{ + collections::HashSet, future, sync::{Arc, Weak}, time::SystemTime, }; -use dashmap::{DashMap, DashSet}; +use dashmap::DashMap; use easytier_proto::common::Flags; use guarden::defer; use tokio::sync::{ @@ -358,7 +359,7 @@ struct ForeignNetworkEntry { traffic_metrics: Arc, event_handler_started: AtomicBool, - indexed_peers: DashSet, + indexed_peers: parking_lot::Mutex>, tasks: Mutex>, @@ -499,7 +500,7 @@ impl ForeignNetworkEntry { traffic_metrics, event_handler_started: AtomicBool::new(false), - indexed_peers: DashSet::new(), + indexed_peers: parking_lot::Mutex::new(HashSet::new()), tasks: Mutex::new(JoinSet::new()), @@ -624,7 +625,7 @@ impl Drop for ForeignNetworkEntry { struct ForeignNetworkManagerData { network_peer_maps: DashMap>, - peer_network_map: DashMap>, + peer_network_map: DashMap>, network_peer_last_update: DashMap, global_peer_map: Weak, lock: std::sync::Mutex<()>, @@ -638,19 +639,21 @@ impl ForeignNetworkManagerData { ) { let indexed_peers = entry .indexed_peers + .lock() .iter() - .map(|peer_id| *peer_id) + .copied() .collect::>(); for peer_id in indexed_peers { - self.peer_network_map.remove_if(&peer_id, |_, networks| { - networks.remove(network_name); - networks.is_empty() - }); + self.peer_network_map + .remove_if_mut(&peer_id, |_, networks| { + networks.remove(network_name); + networks.is_empty() + }); } - entry.indexed_peers.clear(); + entry.indexed_peers.lock().clear(); } - fn get_peer_network(&self, peer_id: PeerId) -> Option> { + fn get_peer_network(&self, peer_id: PeerId) -> Option> { self.peer_network_map.get(&peer_id).map(|v| v.clone()) } @@ -697,11 +700,12 @@ impl ForeignNetworkManagerData { self.remove_entry_from_peer_index_locked(network_name, &expected_entry); self.network_peer_last_update.remove(network_name); } else { - self.peer_network_map.remove_if(&peer_id, |_, networks| { - networks.remove(network_name); - networks.is_empty() - }); - expected_entry.indexed_peers.remove(&peer_id); + self.peer_network_map + .remove_if_mut(&peer_id, |_, networks| { + networks.remove(network_name); + networks.is_empty() + }); + expected_entry.indexed_peers.lock().remove(&peer_id); } shrink_dashmap(&self.peer_network_map, None); shrink_dashmap(&self.network_peer_maps, None); @@ -738,7 +742,7 @@ impl ForeignNetworkManagerData { .entry(peer_id) .or_default() .insert(network_name.clone()); - expected_entry.indexed_peers.insert(peer_id); + expected_entry.indexed_peers.lock().insert(peer_id); self.network_peer_last_update .insert(network_name.clone(), SystemTime::now()); true @@ -803,7 +807,7 @@ impl ForeignNetworkManagerData { .remove_if(network_name, |_, entry| { Arc::ptr_eq(entry, &expected_entry) && entry.peer_map.is_empty() - && entry.indexed_peers.is_empty() + && entry.indexed_peers.lock().is_empty() }) .is_some() { @@ -1669,10 +1673,10 @@ mod tests { .entry(peer_id) .or_default() .insert(network_name.clone()); - entry.indexed_peers.insert(peer_id); + entry.indexed_peers.lock().insert(peer_id); } data.peer_network_map - .get(&20) + .get_mut(&20) .unwrap() .insert(other_network.clone()); @@ -1697,11 +1701,11 @@ mod tests { .get(&20) .unwrap() .iter() - .map(|network| network.clone()) + .cloned() .collect::>(), [other_network] ); - assert!(entry.indexed_peers.is_empty()); + assert!(entry.indexed_peers.lock().is_empty()); assert!(!data.network_peer_last_update.contains_key(&network_name)); } @@ -1730,7 +1734,7 @@ mod tests { .entry(peer_id) .or_default() .insert(network_name.clone()); - current_entry.indexed_peers.insert(peer_id); + current_entry.indexed_peers.lock().insert(peer_id); } assert!( data.peer_network_map @@ -1738,7 +1742,7 @@ mod tests { .unwrap() .contains(&network_name) ); - assert!(current_entry.indexed_peers.contains(&10)); + assert!(current_entry.indexed_peers.lock().contains(&10)); data.remove_peer(10, &network_name, &Arc::downgrade(&old_entry)); assert!( @@ -1759,7 +1763,7 @@ mod tests { assert!(!data.network_peer_maps.contains_key(&network_name)); assert!(!data.peer_network_map.contains_key(&10)); assert!(!data.peer_network_map.contains_key(&11)); - assert!(current_entry.indexed_peers.is_empty()); + assert!(current_entry.indexed_peers.lock().is_empty()); } impl ForeignNetworkManager { diff --git a/easytier-core/src/peers/relay_peer_map.rs b/easytier-core/src/peers/relay_peer_map.rs index dc835a49..a53e4425 100644 --- a/easytier-core/src/peers/relay_peer_map.rs +++ b/easytier-core/src/peers/relay_peer_map.rs @@ -8,6 +8,7 @@ use tokio::sync::{Mutex, OwnedMutexGuard, oneshot}; use crate::{ config::PeerId, + foundation::small_map_shard_amount, foundation::time::{Duration, timeout}, packet::{PacketType, ZCPacket}, peers::{ @@ -146,10 +147,10 @@ impl RelayPeerMap { metric_network_name, my_peer_id, peer_session_store, - states: DashMap::new(), - pending_handshakes: DashMap::new(), - handshake_locks: DashMap::new(), - pending_packets: DashMap::new(), + states: DashMap::with_shard_amount(small_map_shard_amount()), + pending_handshakes: DashMap::with_shard_amount(small_map_shard_amount()), + handshake_locks: DashMap::with_shard_amount(small_map_shard_amount()), + pending_packets: DashMap::with_shard_amount(small_map_shard_amount()), is_secure_mode_enabled, }) } diff --git a/easytier-core/src/peers/route/peer_ospf_route.rs b/easytier-core/src/peers/route/peer_ospf_route.rs index b375c261..d0547390 100644 --- a/easytier-core/src/peers/route/peer_ospf_route.rs +++ b/easytier-core/src/peers/route/peer_ospf_route.rs @@ -34,6 +34,7 @@ use tokio::{ use crate::{ config::PeerId, config::peers::PeerGroupIdentity, + foundation::small_map_shard_amount, peers::{ PeerPacketFilter, context::{ @@ -200,7 +201,7 @@ impl OspfRouteSnapshotLookup<'_> { } type PeerGraph = Graph; -type PeerIdToNodeIdxMap = DashMap; +type PeerIdToNodeIdxMap = HashMap; #[derive(Debug, Clone, Copy)] pub struct OspfNextHopInfo { @@ -228,11 +229,11 @@ pub struct OspfRouteTable { impl OspfRouteTable { pub fn new() -> Self { OspfRouteTable { - peer_infos: DashMap::new(), - next_hop_map: DashMap::new(), - suppressed_peer_ids: DashMap::new(), - ipv4_peer_id_map: DashMap::new(), - ipv6_peer_id_map: DashMap::new(), + peer_infos: DashMap::with_shard_amount(small_map_shard_amount()), + next_hop_map: DashMap::with_shard_amount(small_map_shard_amount()), + suppressed_peer_ids: DashMap::with_shard_amount(small_map_shard_amount()), + ipv4_peer_id_map: DashMap::with_shard_amount(small_map_shard_amount()), + ipv6_peer_id_map: DashMap::with_shard_amount(small_map_shard_amount()), cidr_peer_id_map: ArcSwap::new(Arc::new(PrefixMap::new())), cidr_v6_peer_id_map: ArcSwap::new(Arc::new(PrefixMap::new())), next_hop_map_version: AtomicVersion::new(), @@ -301,7 +302,7 @@ impl OspfRouteTable { let mut graph: PeerGraph = PeerGraph::new(); let mut start_node_idx = None; - let peer_id_to_node_index: PeerIdToNodeIdxMap = DashMap::new(); + let mut peer_id_to_node_index = PeerIdToNodeIdxMap::new(); for entry in &snapshot.peer_infos { let peer_id = entry.peer_id; @@ -321,12 +322,10 @@ impl OspfRouteTable { return (graph, NodeIndex::end()); } - for item in peer_id_to_node_index.iter() { - let src_peer_id = *item.key(); + for (&src_peer_id, src_node_idx) in &peer_id_to_node_index { if src_peer_id != my_peer_id && snapshot.suppressed_peer_ids.contains(&src_peer_id) { continue; } - let src_node_idx = item.value(); let connected_peers: BTreeSet<_> = lookup.connected_peers(src_peer_id).unwrap_or_default(); @@ -2035,9 +2034,9 @@ impl SyncRouteSession { SyncRouteSession { my_peer_id, dst_peer_id, - dst_saved_peer_info_versions: DashMap::new(), - dst_saved_conn_info_version: DashMap::new(), - dst_saved_foreign_network_versions: DashMap::new(), + dst_saved_peer_info_versions: DashMap::with_shard_amount(small_map_shard_amount()), + dst_saved_conn_info_version: DashMap::with_shard_amount(small_map_shard_amount()), + dst_saved_foreign_network_versions: DashMap::with_shard_amount(small_map_shard_amount()), unreachable_peers_for_peer_info: parking_lot::Mutex::new(BTreeMap::new()), unreachable_peers_for_conn_info: parking_lot::Mutex::new(BTreeMap::new()), @@ -2412,7 +2411,7 @@ impl PeerRouteServiceImpl { my_peer_id, my_peer_route_id: rand::random(), context: context.clone(), - sessions: DashMap::new(), + sessions: DashMap::with_shard_amount(small_map_shard_amount()), stopped: AtomicBool::new(false), session_operation: AsyncRwLock::new(()), @@ -2422,21 +2421,23 @@ impl PeerRouteServiceImpl { route_table: OspfRouteTable::new(), route_table_with_cost: OspfRouteTable::new(), - foreign_network_owner_map: DashMap::new(), - foreign_network_my_peer_id_map: DashMap::new(), + foreign_network_owner_map: DashMap::with_shard_amount(small_map_shard_amount()), + foreign_network_my_peer_id_map: DashMap::with_shard_amount(small_map_shard_amount()), synced_route_info: SyncedRouteInfo { default_easytier_version: context.easytier_version(), peer_infos: RwLock::new(OrderedHashMap::new()), - raw_peer_infos: DashMap::new(), + raw_peer_infos: DashMap::with_shard_amount(small_map_shard_amount()), conn_map: RwLock::new(OrderedHashMap::new()), - foreign_network: DashMap::new(), - group_trust_map: DashMap::new(), - group_trust_map_cache: DashMap::new(), + foreign_network: DashMap::with_shard_amount(small_map_shard_amount()), + group_trust_map: DashMap::with_shard_amount(small_map_shard_amount()), + group_trust_map_cache: DashMap::with_shard_amount(small_map_shard_amount()), group_trust_update_lock: parking_lot::Mutex::new(()), trusted_credential_pubkeys: RwLock::new(HashMap::new()), - non_reusable_credential_owners: DashMap::new(), - suppressed_non_reusable_credential_peers: DashMap::new(), + non_reusable_credential_owners: DashMap::with_shard_amount(small_map_shard_amount()), + suppressed_non_reusable_credential_peers: DashMap::with_shard_amount( + small_map_shard_amount(), + ), version: AtomicVersion::new(), }, public_ipv6_service: std::sync::Mutex::new(Weak::new()), diff --git a/easytier-core/src/peers/traffic_metrics.rs b/easytier-core/src/peers/traffic_metrics.rs index 724bed6c..8f0d6dc1 100644 --- a/easytier-core/src/peers/traffic_metrics.rs +++ b/easytier-core/src/peers/traffic_metrics.rs @@ -4,6 +4,7 @@ use dashmap::DashMap; use futures::future::BoxFuture; use crate::config::PeerId; +use crate::foundation::small_map_shard_amount; use crate::foundation::stats::{CounterHandle, LabelSet, LabelType, MetricName, StatsManager}; use crate::packet::{PacketType, ZCPacket}; use crate::peers::util::shrink_dashmap; @@ -88,7 +89,7 @@ impl LogicalTrafficMetrics { instance_bytes_metric, instance_packets_metric, label_kind, - per_peer: DashMap::new(), + per_peer: DashMap::with_shard_amount(small_map_shard_amount()), } } diff --git a/easytier-core/src/rpc/client.rs b/easytier-core/src/rpc/client.rs index f35b51b8..52defac1 100644 --- a/easytier-core/src/rpc/client.rs +++ b/easytier-core/src/rpc/client.rs @@ -14,6 +14,7 @@ use tokio::{sync::mpsc, task::JoinSet}; use crate::{ config::PeerId, foundation::{ + small_map_shard_amount, stats::{ArcRpcMetrics, RpcMetricLabels, RpcMetricsProvider}, time::timeout, }, @@ -109,8 +110,8 @@ impl Client { Self { mpsc: Mutex::new(MpscTunnel::new(ring_a, None)), transport: Mutex::new(MpscTunnel::new(ring_b, None)), - inflight_requests: Arc::new(DashMap::new()), - peer_info: Arc::new(DashMap::new()), + inflight_requests: Arc::new(DashMap::with_shard_amount(small_map_shard_amount())), + peer_info: Arc::new(DashMap::with_shard_amount(small_map_shard_amount())), tasks: Mutex::new(JoinSet::new()), metrics: None, } diff --git a/easytier-core/src/rpc/server.rs b/easytier-core/src/rpc/server.rs index 796cd397..74f75908 100644 --- a/easytier-core/src/rpc/server.rs +++ b/easytier-core/src/rpc/server.rs @@ -14,6 +14,7 @@ use tokio::task::JoinSet; use crate::{ foundation::{ + small_map_shard_amount, stats::{ArcRpcMetrics, RpcMetricLabels, RpcMetricsProvider}, task::reap_joinset_background, }, @@ -77,7 +78,7 @@ impl Server { tasks: Arc::new(Mutex::new(JoinSet::new())), handler_tasks: Arc::new(Mutex::new(JoinSet::new())), stopped: Arc::new(AtomicBool::new(false)), - packet_mergers: Arc::new(DashMap::new()), + packet_mergers: Arc::new(DashMap::with_shard_amount(small_map_shard_amount())), metrics: None, } } diff --git a/easytier-core/src/rpc/service_registry.rs b/easytier-core/src/rpc/service_registry.rs index 60b47a83..8b3d060f 100644 --- a/easytier-core/src/rpc/service_registry.rs +++ b/easytier-core/src/rpc/service_registry.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use dashmap::DashMap; +use crate::foundation::small_map_shard_amount; use crate::proto::common::RpcDescriptor; use crate::proto::rpc_types; #[cfg(all(feature = "management-rpc", any(test, target_os = "wasi")))] @@ -86,7 +87,7 @@ impl Default for ServiceRegistry { impl ServiceRegistry { pub fn new() -> Self { Self { - table: DashMap::new(), + table: DashMap::with_shard_amount(small_map_shard_amount()), } }