perf(core): cap shards in per-network and per-peer maps

Public relays create many mostly empty maps for each foreign network.
DashMap defaults to 128 shards on a 24-CPU host, allocating about
1.26 GiB of shard arrays for 2,421 networks before peer/session state.

Cap these maps at eight shards and retain the four-shard default on
single-CPU hosts. Cache the shard count once and keep global indexes
on their existing default. Apply the same cap to peer connection maps
and route synchronization version maps.

Replace nested peer/network DashSets with HashSets protected by their
existing outer map or an entry mutex. Keep cross-index updates under
the manager lock. Use a plain HashMap for the local route graph index.

Validation: 245 peer/RPC tests and eight three-node integration tests
passed, along with workspace fmt and clippy with full features.
This commit is contained in:
KKRainbow
2026-10-12 23:58:22 +08:00
parent d31396f04c
commit ef3e6b418a
11 changed files with 84 additions and 59 deletions
+12
View File
@@ -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<usize> = std::sync::LazyLock::new(|| {
(std::thread::available_parallelism().map_or(1, usize::from) * 4)
.next_power_of_two()
.min(8)
});
*SHARDS
}
+2 -1
View File
@@ -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));
+2 -1
View File
@@ -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())),
+2 -1
View File
@@ -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()),
}
}
+29 -25
View File
@@ -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<TrafficMetricRecorder>,
event_handler_started: AtomicBool,
indexed_peers: DashSet<PeerId>,
indexed_peers: parking_lot::Mutex<HashSet<PeerId>>,
tasks: Mutex<JoinSet<()>>,
@@ -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<String, Arc<ForeignNetworkEntry>>,
peer_network_map: DashMap<PeerId, DashSet<String>>,
peer_network_map: DashMap<PeerId, HashSet<String>>,
network_peer_last_update: DashMap<String, SystemTime>,
global_peer_map: Weak<PeerMap>,
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::<Vec<_>>();
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<DashSet<String>> {
fn get_peer_network(&self, peer_id: PeerId) -> Option<HashSet<String>> {
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::<Vec<_>>(),
[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 {
+5 -4
View File
@@ -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,
})
}
@@ -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<PeerId, usize, Directed>;
type PeerIdToNodeIdxMap = DashMap<PeerId, NodeIndex>;
type PeerIdToNodeIdxMap = HashMap<PeerId, NodeIndex>;
#[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()),
+2 -1
View File
@@ -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()),
}
}
+3 -2
View File
@@ -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,
}
+2 -1
View File
@@ -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,
}
}
+2 -1
View File
@@ -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()),
}
}