mirror of
https://github.com/EasyTier/EasyTier.git
synced 2026-10-29 08:03:54 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 fanyang89
|
c12b73e1ae | ||
|
fanyang
|
1375cd1832 | ||
|
fanyang
|
ebb97fd4f4 | ||
|
fanyang
|
7d249979ea | ||
|
fanyang
|
8fa161bc1b | ||
|
fanyang
|
2624b2740e | ||
|
fanyang
|
bf8cff60bf | ||
|
fanyang
|
a247358ec1 | ||
|
fanyang
|
c94d106714 | ||
|
fanyang
|
721b863547 | ||
|
fanyang
|
034f5066cd | ||
|
fanyang
|
9869ddaa4b | ||
 Luna Yao
|
5ea6766238 | ||
|
Luna Yao
|
5efbc8587f | ||
 HYec
|
7632cd64da | ||
 韩嘉乐
|
16b666ad25 | ||
|
Luna Yao
|
8909e88484 | ||
|
Luna Yao
|
5edc4cb1cd | ||
|
韩嘉乐
|
e7709f1cb5 |
Generated
+220
-26
@@ -129,6 +129,12 @@ dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "anes"
|
||||
version = "0.1.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299"
|
||||
|
||||
[[package]]
|
||||
name = "anstream"
|
||||
version = "0.6.15"
|
||||
@@ -241,6 +247,16 @@ dependencies = [
|
||||
"password-hash",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ariadne"
|
||||
version = "0.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "36f5e3dca4e09a6f340a61a0e9c7b61e030c69fc27bf29d73218f7e5e3b7638f"
|
||||
dependencies = [
|
||||
"unicode-width 0.1.11",
|
||||
"yansi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrayvec"
|
||||
version = "0.7.6"
|
||||
@@ -915,7 +931,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bfcfdc083699101d5a7965e49925975f2f55060f94f9a05e7187be95d530ca59"
|
||||
dependencies = [
|
||||
"once_cell",
|
||||
"proc-macro-crate 3.2.0",
|
||||
"proc-macro-crate 3.5.0",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
@@ -1129,6 +1145,12 @@ dependencies = [
|
||||
"toml 0.9.12+spec-1.1.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "cast"
|
||||
version = "0.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"
|
||||
|
||||
[[package]]
|
||||
name = "cc"
|
||||
version = "1.2.10"
|
||||
@@ -1238,6 +1260,33 @@ dependencies = [
|
||||
"windows-targets 0.52.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ciborium"
|
||||
version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e"
|
||||
dependencies = [
|
||||
"ciborium-io",
|
||||
"ciborium-ll",
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ciborium-io"
|
||||
version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757"
|
||||
|
||||
[[package]]
|
||||
name = "ciborium-ll"
|
||||
version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9"
|
||||
dependencies = [
|
||||
"ciborium-io",
|
||||
"half",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "cidr"
|
||||
version = "0.3.1"
|
||||
@@ -1583,6 +1632,42 @@ dependencies = [
|
||||
"cfg-if",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "criterion"
|
||||
version = "0.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f"
|
||||
dependencies = [
|
||||
"anes",
|
||||
"cast",
|
||||
"ciborium",
|
||||
"clap",
|
||||
"criterion-plot",
|
||||
"is-terminal",
|
||||
"itertools 0.10.5",
|
||||
"num-traits",
|
||||
"once_cell",
|
||||
"oorandom",
|
||||
"plotters",
|
||||
"rayon",
|
||||
"regex",
|
||||
"serde",
|
||||
"serde_derive",
|
||||
"serde_json",
|
||||
"tinytemplate",
|
||||
"walkdir",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "criterion-plot"
|
||||
version = "0.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1"
|
||||
dependencies = [
|
||||
"cast",
|
||||
"itertools 0.10.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "critical-section"
|
||||
version = "1.2.0"
|
||||
@@ -2234,6 +2319,7 @@ dependencies = [
|
||||
"aes-gcm",
|
||||
"anyhow",
|
||||
"arc-swap",
|
||||
"ariadne",
|
||||
"async-recursion",
|
||||
"async-ringbuf",
|
||||
"async-stream",
|
||||
@@ -2255,6 +2341,7 @@ dependencies = [
|
||||
"clap_complete",
|
||||
"clap_complete_nushell",
|
||||
"console-subscriber",
|
||||
"criterion",
|
||||
"crossbeam",
|
||||
"ctor 0.8.0",
|
||||
"dashmap",
|
||||
@@ -2265,6 +2352,7 @@ dependencies = [
|
||||
"derive_builder",
|
||||
"derive_more 2.1.1",
|
||||
"encoding",
|
||||
"fastant",
|
||||
"flume 0.12.0",
|
||||
"forwarded-header-value",
|
||||
"futures",
|
||||
@@ -2272,7 +2360,7 @@ dependencies = [
|
||||
"gethostname 0.5.0",
|
||||
"git-version",
|
||||
"globwalk",
|
||||
"guarden",
|
||||
"guarden 0.2.0",
|
||||
"hickory-client",
|
||||
"hickory-proto",
|
||||
"hickory-resolver",
|
||||
@@ -2318,7 +2406,7 @@ dependencies = [
|
||||
"prost-reflect-build",
|
||||
"prost-wkt-types",
|
||||
"quinn",
|
||||
"quinn-plaintext",
|
||||
"quinn-proto",
|
||||
"quote",
|
||||
"rand 0.8.5",
|
||||
"rcgen",
|
||||
@@ -2330,6 +2418,7 @@ dependencies = [
|
||||
"rstest",
|
||||
"rust-i18n",
|
||||
"rustls",
|
||||
"seahash",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"serial_test",
|
||||
@@ -2457,7 +2546,7 @@ dependencies = [
|
||||
"dashmap",
|
||||
"easytier",
|
||||
"futures",
|
||||
"guarden",
|
||||
"guarden 0.1.2",
|
||||
"jsonwebtoken",
|
||||
"mimalloc",
|
||||
"mockall",
|
||||
@@ -2836,6 +2925,16 @@ dependencies = [
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "fastant"
|
||||
version = "0.1.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2e825441bfb2d831c47c97d05821552db8832479f44c571b97fededbf0099c07"
|
||||
dependencies = [
|
||||
"small_ctor",
|
||||
"web-time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "fastbloom"
|
||||
version = "0.9.0"
|
||||
@@ -3593,7 +3692,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ca87812d87fa82896df1adfb5c111cdeaae3edb6da028f5df002dcbd7df71454"
|
||||
dependencies = [
|
||||
"futures",
|
||||
"guarden-macros",
|
||||
"guarden-macros 0.1.2",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "guarden"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b8408903291a7d0cc74169d5de4dd1919a9a402a2f67fcd7df3303ed045fae73"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"guarden-macros 0.2.0",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
@@ -3608,6 +3718,18 @@ dependencies = [
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "guarden-macros"
|
||||
version = "0.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1e0ef28f1077c259f9e7e238e234a78ce18cedbf0251fd2135f5fc23c40e79fe"
|
||||
dependencies = [
|
||||
"proc-macro-crate 3.5.0",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "h2"
|
||||
version = "0.4.7"
|
||||
@@ -4460,6 +4582,17 @@ dependencies = [
|
||||
"once_cell",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "is-terminal"
|
||||
version = "0.4.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
|
||||
dependencies = [
|
||||
"hermit-abi",
|
||||
"libc",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "is-wsl"
|
||||
version = "0.4.0"
|
||||
@@ -5578,7 +5711,7 @@ version = "0.7.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "680998035259dcfcafe653688bf2aa6d3e2dc05e98be6ab46afb089dc84f1df8"
|
||||
dependencies = [
|
||||
"proc-macro-crate 3.2.0",
|
||||
"proc-macro-crate 3.5.0",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
@@ -5820,6 +5953,12 @@ dependencies = [
|
||||
"portable-atomic",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "oorandom"
|
||||
version = "11.1.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
|
||||
|
||||
[[package]]
|
||||
name = "opaque-debug"
|
||||
version = "0.3.1"
|
||||
@@ -6450,6 +6589,34 @@ dependencies = [
|
||||
"time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "plotters"
|
||||
version = "0.3.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747"
|
||||
dependencies = [
|
||||
"num-traits",
|
||||
"plotters-backend",
|
||||
"plotters-svg",
|
||||
"wasm-bindgen",
|
||||
"web-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "plotters-backend"
|
||||
version = "0.3.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a"
|
||||
|
||||
[[package]]
|
||||
name = "plotters-svg"
|
||||
version = "0.3.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670"
|
||||
dependencies = [
|
||||
"plotters-backend",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pnet"
|
||||
version = "0.35.0"
|
||||
@@ -6710,11 +6877,11 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "proc-macro-crate"
|
||||
version = "3.2.0"
|
||||
version = "3.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8ecf48c7ca261d60b74ab1a7b20da18bede46776b2e55535cb958eb595c5fa7b"
|
||||
checksum = "e67ba7e9b2b56446f1d419b1d807906278ffa1a658a8a5d8a39dcb1f5a78614f"
|
||||
dependencies = [
|
||||
"toml_edit 0.22.20",
|
||||
"toml_edit 0.25.12+spec-1.1.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -7022,18 +7189,6 @@ dependencies = [
|
||||
"web-time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quinn-plaintext"
|
||||
version = "0.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f3e617feaeb6493018fa35fc47ae8b630ac8903d8159e9e747018841b99bad3d"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"quinn-proto",
|
||||
"seahash",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quinn-proto"
|
||||
version = "0.11.12"
|
||||
@@ -7608,7 +7763,7 @@ checksum = "1f168d99749d307be9de54d23fd226628d99768225ef08f6ffb52e0182a27746"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"glob",
|
||||
"proc-macro-crate 3.2.0",
|
||||
"proc-macro-crate 3.5.0",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"regex",
|
||||
@@ -8655,6 +8810,12 @@ dependencies = [
|
||||
"autocfg",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "small_ctor"
|
||||
version = "0.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "88414a5ca1f85d82cc34471e975f0f74f6aa54c40f062efa42c0080e7f763f81"
|
||||
|
||||
[[package]]
|
||||
name = "smallvec"
|
||||
version = "1.13.2"
|
||||
@@ -9813,6 +9974,16 @@ dependencies = [
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tinytemplate"
|
||||
version = "1.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc"
|
||||
dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tinyvec"
|
||||
version = "1.8.0"
|
||||
@@ -9932,8 +10103,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "tokio-websockets"
|
||||
version = "0.13.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "dad543404f98bfc969aeb71994105c592acfc6c43323fddcd016bb208d1c65cb"
|
||||
source = "git+https://github.com/EasyTier/tokio-websockets#dc9771c7c215882349c3cb328877550a3593df21"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"bytes",
|
||||
@@ -10008,6 +10178,15 @@ dependencies = [
|
||||
"serde_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "toml_datetime"
|
||||
version = "1.1.1+spec-1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7"
|
||||
dependencies = [
|
||||
"serde_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "toml_edit"
|
||||
version = "0.19.15"
|
||||
@@ -10045,6 +10224,18 @@ dependencies = [
|
||||
"winnow 0.6.18",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "toml_edit"
|
||||
version = "0.25.12+spec-1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d2153edc6955a6c354fad8f5efd38b6a8769bdccf9fe50f8e1329f81b0baa5d7"
|
||||
dependencies = [
|
||||
"indexmap 2.14.0",
|
||||
"toml_datetime 1.1.1+spec-1.1.0",
|
||||
"toml_parser",
|
||||
"winnow 1.0.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "toml_parser"
|
||||
version = "1.1.2+spec-1.1.0"
|
||||
@@ -11892,6 +12083,9 @@ name = "winnow"
|
||||
version = "1.0.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "09dac053f1cd375980747450bfc7250c264eaae0583872e845c0c7cd578872b5"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "winreg"
|
||||
@@ -12273,7 +12467,7 @@ version = "5.14.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "897e79616e84aac4b2c46e9132a4f63b93105d54fe8c0e8f6bffc21fa8d49222"
|
||||
dependencies = [
|
||||
"proc-macro-crate 3.2.0",
|
||||
"proc-macro-crate 3.5.0",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
@@ -12510,7 +12704,7 @@ version = "5.10.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5b59b012ebe9c46656f9cc08d8da8b4c726510aef12559da3e5f1bf72780752c"
|
||||
dependencies = [
|
||||
"proc-macro-crate 3.2.0",
|
||||
"proc-macro-crate 3.5.0",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
|
||||
@@ -118,11 +118,7 @@ pub(crate) fn set_tun_fd(
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn get_runtime_snapshot() -> RuntimeAggregateState {
|
||||
get_runtime_snapshot_inner()
|
||||
}
|
||||
|
||||
pub(crate) fn get_runtime_snapshot_inner() -> RuntimeAggregateState {
|
||||
pub(crate) fn collect_runtime_state() -> RuntimeAggregateState {
|
||||
let infos = match ASYNC_RUNTIME.block_on(INSTANCE_MANAGER.collect_network_infos()) {
|
||||
Ok(infos) => infos,
|
||||
Err(err) => {
|
||||
|
||||
@@ -3,6 +3,4 @@ mod routing;
|
||||
mod socket_server;
|
||||
|
||||
pub(crate) use routing::aggregate_requested_tun_routes;
|
||||
pub use socket_server::{
|
||||
set_snapshot_broadcast_enabled, start_local_socket_server, stop_local_socket_server,
|
||||
};
|
||||
pub use socket_server::{start_local_socket_server, stop_local_socket_server};
|
||||
|
||||
@@ -32,6 +32,13 @@ pub(crate) fn send_local_socket_message(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn shrink_clients_if_sparse(clients: &mut Vec<UnixStream>) {
|
||||
let sparse_limit = clients.len().saturating_mul(2).max(4);
|
||||
if clients.capacity() > sparse_limit {
|
||||
clients.shrink_to_fit();
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn broadcast_local_socket_message(
|
||||
clients: &mut Vec<UnixStream>,
|
||||
message_type: &str,
|
||||
@@ -45,6 +52,7 @@ pub(crate) fn broadcast_local_socket_message(
|
||||
active_clients.push(client);
|
||||
}
|
||||
}
|
||||
shrink_clients_if_sparse(&mut active_clients);
|
||||
*clients = active_clients;
|
||||
delivered
|
||||
}
|
||||
@@ -79,6 +87,7 @@ pub(crate) fn broadcast_local_socket_json_payload_message(
|
||||
active_clients.push(client);
|
||||
}
|
||||
}
|
||||
shrink_clients_if_sparse(&mut active_clients);
|
||||
*clients = active_clients;
|
||||
delivered
|
||||
}
|
||||
|
||||
@@ -1,13 +1,20 @@
|
||||
use super::protocol::{
|
||||
TunRequestPayload, broadcast_local_socket_json_payload_message, broadcast_local_socket_message,
|
||||
};
|
||||
use crate::INSTANCE_MANAGER;
|
||||
use crate::collect_runtime_state_inner;
|
||||
use crate::config::repository::kernel_socket_path;
|
||||
use crate::get_runtime_snapshot_inner;
|
||||
use crate::kernel_bridge::routing::aggregate_tun_routes;
|
||||
use crate::runtime::state::runtime_state::{
|
||||
PeerConnInfo as RuntimePeerConnInfo, RuntimeAggregateState, peer_conn_to_view,
|
||||
};
|
||||
use crate::{ASYNC_RUNTIME, INSTANCE_MANAGER};
|
||||
use easytier::common::global_ctx::{EventBusSubscriber, GlobalCtxEvent};
|
||||
use easytier::proto::api::instance::ListPeerRequest;
|
||||
use easytier::proto::rpc_types::controller::BaseController;
|
||||
use once_cell::sync::Lazy;
|
||||
use serde::Serialize;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::hash::Hash;
|
||||
use std::io::ErrorKind;
|
||||
use std::os::unix::net::{UnixListener, UnixStream};
|
||||
use std::path::PathBuf;
|
||||
@@ -23,13 +30,75 @@ struct LocalSocketState {
|
||||
}
|
||||
|
||||
static LOCAL_SOCKET_STATE: Lazy<Mutex<Option<LocalSocketState>>> = Lazy::new(|| Mutex::new(None));
|
||||
static SNAPSHOT_BROADCAST_ENABLED: AtomicBool = AtomicBool::new(true);
|
||||
const SOCKET_TICK_INTERVAL: Duration = Duration::from_millis(250);
|
||||
const TRAFFIC_STATS_INTERVAL: Duration = Duration::from_secs(1);
|
||||
const INSTANCE_POLL_INTERVAL: Duration = Duration::from_secs(1);
|
||||
const TUN_FAST_CHECK_WINDOW: Duration = Duration::from_secs(8);
|
||||
const EVENT_RECEIVER_SYNC_INTERVAL: Duration = Duration::from_secs(1);
|
||||
|
||||
pub fn set_snapshot_broadcast_enabled(enabled: bool) {
|
||||
SNAPSHOT_BROADCAST_ENABLED.store(enabled, Ordering::Relaxed);
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct TrafficStatsPayload {
|
||||
instances: Vec<InstanceTrafficStats>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct InstanceTrafficStats {
|
||||
config_id: String,
|
||||
instance_id: String,
|
||||
rx_bytes: i64,
|
||||
tx_bytes: i64,
|
||||
peers: Vec<PeerTrafficStats>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct PeerTrafficStats {
|
||||
peer_id: i64,
|
||||
rx_bytes: i64,
|
||||
tx_bytes: i64,
|
||||
total_bytes: i64,
|
||||
latency_us: i64,
|
||||
loss_rate: f64,
|
||||
}
|
||||
|
||||
struct PendingPeerEvent {
|
||||
event: &'static str,
|
||||
instance_id: String,
|
||||
peer_id: i64,
|
||||
conn: Option<RuntimePeerConnInfo>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct DrainedKernelEvents {
|
||||
tun_refresh: bool,
|
||||
topology_lost: bool,
|
||||
peer_events: Vec<PendingPeerEvent>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct RuntimePeerEventPayload {
|
||||
event: &'static str,
|
||||
config_id: String,
|
||||
instance_id: String,
|
||||
peer_id: i64,
|
||||
conn: Option<RuntimePeerConnInfo>,
|
||||
}
|
||||
|
||||
fn shrink_hash_map_if_sparse<K: Eq + Hash, V>(map: &mut HashMap<K, V>) {
|
||||
let sparse_limit = map.len().saturating_mul(2).max(8);
|
||||
if map.capacity() > sparse_limit {
|
||||
map.shrink_to_fit();
|
||||
}
|
||||
}
|
||||
|
||||
fn shrink_hash_set_if_sparse<T: Eq + Hash>(set: &mut HashSet<T>) {
|
||||
let sparse_limit = set.len().saturating_mul(2).max(8);
|
||||
if set.capacity() > sparse_limit {
|
||||
set.shrink_to_fit();
|
||||
}
|
||||
}
|
||||
|
||||
fn sync_tun_event_receivers(receivers: &mut HashMap<String, EventBusSubscriber>) {
|
||||
@@ -44,36 +113,67 @@ fn sync_tun_event_receivers(receivers: &mut HashMap<String, EventBusSubscriber>)
|
||||
}
|
||||
}
|
||||
receivers.retain(|instance_id, _| active_instance_ids.contains(instance_id));
|
||||
shrink_hash_map_if_sparse(receivers);
|
||||
}
|
||||
|
||||
fn event_needs_tun_refresh(event: &GlobalCtxEvent) -> bool {
|
||||
matches!(
|
||||
event,
|
||||
GlobalCtxEvent::DhcpIpv4Changed(_, _)
|
||||
| GlobalCtxEvent::DhcpIpv4Conflicted(_)
|
||||
| GlobalCtxEvent::PublicIpv6Changed(_, _)
|
||||
| GlobalCtxEvent::PublicIpv6RoutesUpdated(_, _)
|
||||
| GlobalCtxEvent::ProxyCidrsUpdated(_, _)
|
||||
| GlobalCtxEvent::ConfigPatched(_)
|
||||
| GlobalCtxEvent::PeerAdded(_)
|
||||
| GlobalCtxEvent::PeerRemoved(_)
|
||||
| GlobalCtxEvent::PeerConnAdded(_)
|
||||
| GlobalCtxEvent::PeerConnRemoved(_)
|
||||
| GlobalCtxEvent::PublicIpv6RoutesUpdated(_, _)
|
||||
)
|
||||
}
|
||||
|
||||
fn drain_tun_refresh_events(receivers: &mut HashMap<String, EventBusSubscriber>) -> bool {
|
||||
let mut refresh_needed = false;
|
||||
fn drain_kernel_events(receivers: &mut HashMap<String, EventBusSubscriber>) -> DrainedKernelEvents {
|
||||
let mut drained = DrainedKernelEvents::default();
|
||||
let mut closed_receivers = Vec::new();
|
||||
for (instance_id, receiver) in receivers.iter_mut() {
|
||||
loop {
|
||||
match receiver.try_recv() {
|
||||
Ok(event) => {
|
||||
refresh_needed = event_needs_tun_refresh(&event) || refresh_needed;
|
||||
drained.tun_refresh = event_needs_tun_refresh(&event) || drained.tun_refresh;
|
||||
match event {
|
||||
GlobalCtxEvent::PeerAdded(peer_id) => {
|
||||
drained.peer_events.push(PendingPeerEvent {
|
||||
event: "peer_added",
|
||||
instance_id: instance_id.clone(),
|
||||
peer_id: peer_id as i64,
|
||||
conn: None,
|
||||
});
|
||||
}
|
||||
GlobalCtxEvent::PeerRemoved(peer_id) => {
|
||||
drained.peer_events.push(PendingPeerEvent {
|
||||
event: "peer_removed",
|
||||
instance_id: instance_id.clone(),
|
||||
peer_id: peer_id as i64,
|
||||
conn: None,
|
||||
});
|
||||
}
|
||||
GlobalCtxEvent::PeerConnAdded(conn_info) => {
|
||||
let peer_id = conn_info.peer_id as i64;
|
||||
drained.peer_events.push(PendingPeerEvent {
|
||||
event: "peer_conn_added",
|
||||
instance_id: instance_id.clone(),
|
||||
peer_id,
|
||||
conn: Some(peer_conn_to_view(conn_info)),
|
||||
});
|
||||
}
|
||||
GlobalCtxEvent::PeerConnRemoved(conn_info) => {
|
||||
let peer_id = conn_info.peer_id as i64;
|
||||
drained.peer_events.push(PendingPeerEvent {
|
||||
event: "peer_conn_removed",
|
||||
instance_id: instance_id.clone(),
|
||||
peer_id,
|
||||
conn: Some(peer_conn_to_view(conn_info)),
|
||||
});
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::TryRecvError::Empty) => break,
|
||||
Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) => {
|
||||
refresh_needed = true;
|
||||
drained.topology_lost = true;
|
||||
continue;
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::TryRecvError::Closed) => {
|
||||
@@ -86,7 +186,124 @@ fn drain_tun_refresh_events(receivers: &mut HashMap<String, EventBusSubscriber>)
|
||||
for instance_id in closed_receivers {
|
||||
receivers.remove(&instance_id);
|
||||
}
|
||||
refresh_needed
|
||||
drained
|
||||
}
|
||||
|
||||
fn broadcast_runtime_peer_events(
|
||||
clients: &mut Vec<UnixStream>,
|
||||
peer_events: Vec<PendingPeerEvent>,
|
||||
) {
|
||||
for event in peer_events {
|
||||
let payload = RuntimePeerEventPayload {
|
||||
event: event.event,
|
||||
config_id: event.instance_id.clone(),
|
||||
instance_id: event.instance_id,
|
||||
peer_id: event.peer_id,
|
||||
conn: event.conn,
|
||||
};
|
||||
match serde_json::to_string(&payload) {
|
||||
Ok(json) => {
|
||||
let _ = broadcast_local_socket_json_payload_message(
|
||||
clients,
|
||||
"runtime_peer_event",
|
||||
&json,
|
||||
);
|
||||
}
|
||||
Err(err) => {
|
||||
ohrs_log_error!("[Rust] serialize runtime peer event failed: {}", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn tun_candidate_ids(snapshot: &RuntimeAggregateState) -> HashSet<String> {
|
||||
snapshot
|
||||
.instances
|
||||
.iter()
|
||||
.filter(|instance| instance.running && instance.tun_required)
|
||||
.map(|instance| instance.instance_id.clone())
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn collect_traffic_stats() -> TrafficStatsPayload {
|
||||
let services = INSTANCE_MANAGER
|
||||
.iter()
|
||||
.filter_map(|instance| {
|
||||
instance
|
||||
.value()
|
||||
.get_api_service()
|
||||
.map(|api_service| (instance.key().to_string(), api_service))
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
let instances = ASYNC_RUNTIME.block_on(async {
|
||||
let mut instances = Vec::new();
|
||||
for (instance_id, api_service) in services {
|
||||
let peers = match api_service
|
||||
.get_peer_manage_service()
|
||||
.list_peer(BaseController::default(), ListPeerRequest::default())
|
||||
.await
|
||||
{
|
||||
Ok(response) => response.peer_infos,
|
||||
Err(err) => {
|
||||
ohrs_log_debug!(
|
||||
"[Rust] collect traffic stats list_peer failed instance={}: {}",
|
||||
instance_id,
|
||||
err
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let mut instance_rx_bytes = 0i64;
|
||||
let mut instance_tx_bytes = 0i64;
|
||||
let mut peer_stats = Vec::with_capacity(peers.len());
|
||||
|
||||
for peer in peers {
|
||||
let mut peer_rx_bytes = 0i64;
|
||||
let mut peer_tx_bytes = 0i64;
|
||||
let mut latency_us = i64::MAX;
|
||||
let mut loss_rate = 0f64;
|
||||
|
||||
for conn in peer.conns {
|
||||
if let Some(stats) = conn.stats {
|
||||
let rx_bytes = stats.rx_bytes as i64;
|
||||
let tx_bytes = stats.tx_bytes as i64;
|
||||
peer_rx_bytes += rx_bytes;
|
||||
peer_tx_bytes += tx_bytes;
|
||||
latency_us = latency_us.min(stats.latency_us as i64);
|
||||
}
|
||||
loss_rate = loss_rate.max(conn.loss_rate as f64);
|
||||
}
|
||||
|
||||
instance_rx_bytes += peer_rx_bytes;
|
||||
instance_tx_bytes += peer_tx_bytes;
|
||||
peer_stats.push(PeerTrafficStats {
|
||||
peer_id: peer.peer_id as i64,
|
||||
rx_bytes: peer_rx_bytes,
|
||||
tx_bytes: peer_tx_bytes,
|
||||
total_bytes: peer_rx_bytes + peer_tx_bytes,
|
||||
latency_us: if latency_us == i64::MAX {
|
||||
-1
|
||||
} else {
|
||||
latency_us
|
||||
},
|
||||
loss_rate,
|
||||
});
|
||||
}
|
||||
|
||||
instances.push(InstanceTrafficStats {
|
||||
config_id: instance_id.clone(),
|
||||
instance_id,
|
||||
rx_bytes: instance_rx_bytes,
|
||||
tx_bytes: instance_tx_bytes,
|
||||
peers: peer_stats,
|
||||
});
|
||||
}
|
||||
instances
|
||||
});
|
||||
|
||||
TrafficStatsPayload { instances }
|
||||
}
|
||||
|
||||
pub fn start_local_socket_server() -> bool {
|
||||
@@ -131,21 +348,25 @@ pub fn start_local_socket_server() -> bool {
|
||||
let stop_flag = std::sync::Arc::new(AtomicBool::new(false));
|
||||
let worker_stop_flag = stop_flag.clone();
|
||||
let worker = thread::spawn(move || {
|
||||
let mut last_snapshot_json = String::new();
|
||||
let mut last_topology_json = String::new();
|
||||
let mut delivered_tun_requests = HashSet::new();
|
||||
let mut last_tun_route_signatures = HashMap::<String, String>::new();
|
||||
let mut tun_fast_until = Instant::now() + TUN_FAST_CHECK_WINDOW;
|
||||
let mut tun_bootstrap_done = false;
|
||||
let mut last_event_receiver_sync_at: Option<Instant> = None;
|
||||
let mut last_traffic_stats_at: Option<Instant> = None;
|
||||
let mut last_instance_poll_at: Option<Instant> = None;
|
||||
let mut tun_event_receivers = HashMap::<String, EventBusSubscriber>::new();
|
||||
let mut clients = Vec::<UnixStream>::new();
|
||||
|
||||
while !worker_stop_flag.load(Ordering::Relaxed) {
|
||||
let mut full_topology_dirty = false;
|
||||
let mut accepted_client = false;
|
||||
loop {
|
||||
match listener.accept() {
|
||||
Ok((stream, _addr)) => {
|
||||
accepted_client = true;
|
||||
full_topology_dirty = true;
|
||||
clients.push(stream);
|
||||
tun_fast_until = Instant::now() + TUN_FAST_CHECK_WINDOW;
|
||||
tun_bootstrap_done = false;
|
||||
@@ -158,15 +379,21 @@ pub fn start_local_socket_server() -> bool {
|
||||
}
|
||||
}
|
||||
|
||||
let snapshot_enabled = SNAPSHOT_BROADCAST_ENABLED.load(Ordering::Relaxed);
|
||||
if clients.is_empty() {
|
||||
if !last_snapshot_json.is_empty() {
|
||||
last_snapshot_json.clear();
|
||||
if !last_topology_json.is_empty() {
|
||||
last_topology_json.clear();
|
||||
last_topology_json.shrink_to_fit();
|
||||
}
|
||||
delivered_tun_requests.clear();
|
||||
shrink_hash_set_if_sparse(&mut delivered_tun_requests);
|
||||
last_tun_route_signatures.clear();
|
||||
shrink_hash_map_if_sparse(&mut last_tun_route_signatures);
|
||||
tun_event_receivers.clear();
|
||||
shrink_hash_map_if_sparse(&mut tun_event_receivers);
|
||||
clients.shrink_to_fit();
|
||||
last_event_receiver_sync_at = None;
|
||||
last_traffic_stats_at = None;
|
||||
last_instance_poll_at = None;
|
||||
tun_bootstrap_done = false;
|
||||
thread::sleep(SOCKET_TICK_INTERVAL);
|
||||
continue;
|
||||
@@ -181,115 +408,143 @@ pub fn start_local_socket_server() -> bool {
|
||||
sync_tun_event_receivers(&mut tun_event_receivers);
|
||||
last_event_receiver_sync_at = Some(now);
|
||||
}
|
||||
if drain_tun_refresh_events(&mut tun_event_receivers) {
|
||||
let drained_events = drain_kernel_events(&mut tun_event_receivers);
|
||||
let tun_refresh = drained_events.tun_refresh;
|
||||
let topology_lost = drained_events.topology_lost;
|
||||
let peer_events = drained_events.peer_events;
|
||||
if topology_lost {
|
||||
full_topology_dirty = true;
|
||||
}
|
||||
if tun_refresh {
|
||||
tun_bootstrap_done = false;
|
||||
tun_fast_until = now + TUN_FAST_CHECK_WINDOW;
|
||||
}
|
||||
let should_collect_snapshot = snapshot_enabled
|
||||
|| accepted_client
|
||||
|| (!tun_bootstrap_done && now < tun_fast_until);
|
||||
if !should_collect_snapshot {
|
||||
if !last_snapshot_json.is_empty() {
|
||||
last_snapshot_json.clear();
|
||||
if !peer_events.is_empty() {
|
||||
broadcast_runtime_peer_events(&mut clients, peer_events);
|
||||
}
|
||||
let should_collect_traffic_stats = last_traffic_stats_at
|
||||
.map(|last| now.duration_since(last) >= TRAFFIC_STATS_INTERVAL)
|
||||
.unwrap_or(true);
|
||||
if should_collect_traffic_stats {
|
||||
last_traffic_stats_at = Some(now);
|
||||
match serde_json::to_string(&collect_traffic_stats()) {
|
||||
Ok(json) => {
|
||||
let _ = broadcast_local_socket_json_payload_message(
|
||||
&mut clients,
|
||||
"traffic_stats",
|
||||
&json,
|
||||
);
|
||||
}
|
||||
Err(err) => {
|
||||
ohrs_log_error!("[Rust] serialize traffic stats failed: {}", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
let should_poll_instance = last_instance_poll_at
|
||||
.map(|last| now.duration_since(last) >= INSTANCE_POLL_INTERVAL)
|
||||
.unwrap_or(true);
|
||||
let should_collect_topology = accepted_client
|
||||
|| full_topology_dirty
|
||||
|| tun_refresh
|
||||
|| should_poll_instance
|
||||
|| (!tun_bootstrap_done && now < tun_fast_until);
|
||||
if !should_collect_topology {
|
||||
thread::sleep(SOCKET_TICK_INTERVAL);
|
||||
continue;
|
||||
}
|
||||
|
||||
let snapshot = get_runtime_snapshot_inner();
|
||||
if snapshot_enabled {
|
||||
let snapshot_json = match serde_json::to_string(&snapshot) {
|
||||
Ok(json) => json,
|
||||
Err(err) => {
|
||||
ohrs_log_error!("[Rust] serialize runtime snapshot failed: {}", err);
|
||||
thread::sleep(SOCKET_TICK_INTERVAL);
|
||||
continue;
|
||||
let snapshot = collect_runtime_state_inner();
|
||||
last_instance_poll_at = Some(now);
|
||||
match serde_json::to_string(&snapshot) {
|
||||
Ok(json) => {
|
||||
if accepted_client || full_topology_dirty || json != last_topology_json {
|
||||
let _ = broadcast_local_socket_json_payload_message(
|
||||
&mut clients,
|
||||
"runtime_topology",
|
||||
&json,
|
||||
);
|
||||
last_topology_json = json;
|
||||
}
|
||||
};
|
||||
|
||||
if accepted_client || snapshot_json != last_snapshot_json {
|
||||
let _ = broadcast_local_socket_json_payload_message(
|
||||
&mut clients,
|
||||
"runtime_snapshot",
|
||||
&snapshot_json,
|
||||
);
|
||||
last_snapshot_json = snapshot_json;
|
||||
}
|
||||
} else if !last_snapshot_json.is_empty() {
|
||||
last_snapshot_json.clear();
|
||||
Err(err) => {
|
||||
ohrs_log_error!("[Rust] serialize runtime topology failed: {}", err);
|
||||
}
|
||||
}
|
||||
|
||||
let active_tun_candidate_ids = tun_candidate_ids(&snapshot);
|
||||
delivered_tun_requests
|
||||
.retain(|instance_id| active_tun_candidate_ids.contains(instance_id));
|
||||
last_tun_route_signatures
|
||||
.retain(|instance_id, _| active_tun_candidate_ids.contains(instance_id));
|
||||
shrink_hash_set_if_sparse(&mut delivered_tun_requests);
|
||||
shrink_hash_map_if_sparse(&mut last_tun_route_signatures);
|
||||
let mut saw_running_instance = false;
|
||||
let mut saw_tun_candidate = false;
|
||||
for instance in snapshot.instances.iter() {
|
||||
if instance.running {
|
||||
saw_running_instance = true;
|
||||
}
|
||||
if instance.running && instance.tun_required {
|
||||
saw_tun_candidate = true;
|
||||
let virtual_ipv4 = instance
|
||||
.my_node_info
|
||||
.as_ref()
|
||||
.and_then(|info| info.virtual_ipv4.clone());
|
||||
let virtual_ipv4_cidr = instance
|
||||
.my_node_info
|
||||
.as_ref()
|
||||
.and_then(|info| info.virtual_ipv4_cidr.clone());
|
||||
if clients.is_empty() {
|
||||
if !(instance.running && instance.tun_required) {
|
||||
continue;
|
||||
}
|
||||
|
||||
saw_tun_candidate = true;
|
||||
let virtual_ipv4 = instance
|
||||
.my_node_info
|
||||
.as_ref()
|
||||
.and_then(|info| info.virtual_ipv4.clone());
|
||||
let virtual_ipv4_cidr = instance
|
||||
.my_node_info
|
||||
.as_ref()
|
||||
.and_then(|info| info.virtual_ipv4_cidr.clone());
|
||||
if clients.is_empty() {
|
||||
continue;
|
||||
}
|
||||
if virtual_ipv4.is_none() || virtual_ipv4_cidr.is_none() {
|
||||
continue;
|
||||
}
|
||||
let aggregated_routes = aggregate_tun_routes(instance);
|
||||
let route_signature = serde_json::to_string(&(
|
||||
&virtual_ipv4,
|
||||
&virtual_ipv4_cidr,
|
||||
&aggregated_routes,
|
||||
instance.magic_dns_enabled,
|
||||
instance.need_exit_node,
|
||||
))
|
||||
.unwrap_or_else(|_| "[]".to_string());
|
||||
let should_send = !delivered_tun_requests.contains(&instance.instance_id)
|
||||
|| last_tun_route_signatures
|
||||
.get(&instance.instance_id)
|
||||
.map(|value| value != &route_signature)
|
||||
.unwrap_or(true);
|
||||
if !should_send {
|
||||
continue;
|
||||
}
|
||||
let payload = TunRequestPayload {
|
||||
config_id: instance.config_id.clone(),
|
||||
instance_id: instance.instance_id.clone(),
|
||||
display_name: instance.display_name.clone(),
|
||||
virtual_ipv4,
|
||||
virtual_ipv4_cidr,
|
||||
aggregated_routes,
|
||||
magic_dns_enabled: instance.magic_dns_enabled,
|
||||
need_exit_node: instance.need_exit_node,
|
||||
};
|
||||
let payload_json = match serde_json::to_string(&payload) {
|
||||
Ok(json) => json,
|
||||
Err(err) => {
|
||||
ohrs_log_error!("[Rust] serialize tun request failed: {}", err);
|
||||
continue;
|
||||
}
|
||||
if virtual_ipv4.is_none() || virtual_ipv4_cidr.is_none() {
|
||||
continue;
|
||||
}
|
||||
let aggregated_routes = aggregate_tun_routes(instance);
|
||||
let route_signature = serde_json::to_string(&(
|
||||
&virtual_ipv4,
|
||||
&virtual_ipv4_cidr,
|
||||
&aggregated_routes,
|
||||
instance.magic_dns_enabled,
|
||||
instance.need_exit_node,
|
||||
))
|
||||
.unwrap_or_else(|_| "[]".to_string());
|
||||
let should_send = accepted_client
|
||||
|| !delivered_tun_requests.contains(&instance.instance_id)
|
||||
|| last_tun_route_signatures
|
||||
.get(&instance.instance_id)
|
||||
.map(|value| value != &route_signature)
|
||||
.unwrap_or(true);
|
||||
if !should_send {
|
||||
continue;
|
||||
}
|
||||
let payload = TunRequestPayload {
|
||||
config_id: instance.config_id.clone(),
|
||||
instance_id: instance.instance_id.clone(),
|
||||
display_name: instance.display_name.clone(),
|
||||
virtual_ipv4,
|
||||
virtual_ipv4_cidr,
|
||||
aggregated_routes,
|
||||
magic_dns_enabled: instance.magic_dns_enabled,
|
||||
need_exit_node: instance.need_exit_node,
|
||||
};
|
||||
let payload_json = match serde_json::to_string(&payload) {
|
||||
Ok(json) => json,
|
||||
Err(err) => {
|
||||
ohrs_log_error!("[Rust] serialize tun request failed: {}", err);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
if broadcast_local_socket_message(&mut clients, "tun_request", &payload_json) {
|
||||
delivered_tun_requests.insert(instance.instance_id.clone());
|
||||
last_tun_route_signatures
|
||||
.insert(instance.instance_id.clone(), route_signature);
|
||||
}
|
||||
} else {
|
||||
delivered_tun_requests.remove(&instance.instance_id);
|
||||
last_tun_route_signatures.remove(&instance.instance_id);
|
||||
};
|
||||
if broadcast_local_socket_message(&mut clients, "tun_request", &payload_json) {
|
||||
delivered_tun_requests.insert(instance.instance_id.clone());
|
||||
last_tun_route_signatures.insert(instance.instance_id.clone(), route_signature);
|
||||
}
|
||||
}
|
||||
if !snapshot_enabled
|
||||
&& (!delivered_tun_requests.is_empty()
|
||||
|| (saw_running_instance && !saw_tun_candidate)
|
||||
|| now >= tun_fast_until)
|
||||
if !delivered_tun_requests.is_empty()
|
||||
|| (saw_running_instance && !saw_tun_candidate)
|
||||
|| now >= tun_fast_until
|
||||
{
|
||||
tun_bootstrap_done = true;
|
||||
}
|
||||
|
||||
@@ -63,7 +63,7 @@ use easytier::proto::api::manage::NetworkConfig;
|
||||
use easytier::proto::api::manage::NetworkingMethod;
|
||||
use easytier::web_client::{WebClient, WebClientHooks, run_web_client};
|
||||
use kernel_bridge::{
|
||||
set_snapshot_broadcast_enabled, start_local_socket_server as start_local_socket_server_inner,
|
||||
start_local_socket_server as start_local_socket_server_inner,
|
||||
stop_local_socket_server as stop_local_socket_server_inner,
|
||||
};
|
||||
use napi_derive_ohos::napi;
|
||||
@@ -517,18 +517,8 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[napi]
|
||||
pub fn get_runtime_snapshot() -> RuntimeAggregateState {
|
||||
exports::runtime_api::get_runtime_snapshot()
|
||||
}
|
||||
|
||||
#[napi]
|
||||
pub fn set_kernel_snapshot_enabled(enabled: bool) {
|
||||
set_snapshot_broadcast_enabled(enabled);
|
||||
}
|
||||
|
||||
pub(crate) fn get_runtime_snapshot_inner() -> RuntimeAggregateState {
|
||||
exports::runtime_api::get_runtime_snapshot_inner()
|
||||
pub(crate) fn collect_runtime_state_inner() -> RuntimeAggregateState {
|
||||
exports::runtime_api::collect_runtime_state()
|
||||
}
|
||||
|
||||
#[napi]
|
||||
|
||||
@@ -324,7 +324,7 @@ fn route_to_view(route: api::instance::Route) -> RouteView {
|
||||
}
|
||||
}
|
||||
|
||||
fn peer_conn_to_view(conn: api::instance::PeerConnInfo) -> PeerConnInfo {
|
||||
pub(crate) fn peer_conn_to_view(conn: api::instance::PeerConnInfo) -> PeerConnInfo {
|
||||
let stats = conn.stats.map(|stats| PeerConnStats {
|
||||
rx_bytes: stats.rx_bytes as i64,
|
||||
tx_bytes: stats.tx_bytes as i64,
|
||||
|
||||
+12
-4
@@ -51,7 +51,7 @@ time = "0.3"
|
||||
toml = "0.8.12"
|
||||
chrono = { version = "0.4.37", features = ["serde"] }
|
||||
|
||||
guarden = "0.1"
|
||||
guarden = "0.2"
|
||||
|
||||
delegate = "0.13.5"
|
||||
|
||||
@@ -82,7 +82,8 @@ pin-project-lite = "0.2.13"
|
||||
atomic_refcell = "0.1.13"
|
||||
|
||||
quinn = { version = "0.11.8", optional = true, features = ["ring"] }
|
||||
quinn-plaintext = { version = "0.3.0", optional = true }
|
||||
quinn-proto = { version = "0.11.12", optional = true }
|
||||
seahash = { version = "4.1.0", optional = true }
|
||||
|
||||
rustls = { version = "0.23.0", features = [
|
||||
"ring", "tls12"
|
||||
@@ -90,7 +91,7 @@ rustls = { version = "0.23.0", features = [
|
||||
rcgen = { version = "0.12.1", optional = true }
|
||||
|
||||
# for websocket
|
||||
tokio-websockets = { version = "0.13.2", optional = true, features = [
|
||||
tokio-websockets = { version = "0.13.2", git = "https://github.com/EasyTier/tokio-websockets", optional = true, features = [
|
||||
"rustls-webpki-roots",
|
||||
"client",
|
||||
"server",
|
||||
@@ -133,6 +134,7 @@ prost-wkt-types = "0.7.1"
|
||||
pbjson = "0.9.0"
|
||||
|
||||
anyhow = "1.0"
|
||||
ariadne = "0.5"
|
||||
|
||||
url = { version = "2.5", features = ["serde"] }
|
||||
percent-encoding = "2.3.1"
|
||||
@@ -213,6 +215,7 @@ smoltcp = { git = "https://github.com/smoltcp-rs/smoltcp.git", rev = "0a926767a6
|
||||
"async",
|
||||
] }
|
||||
parking_lot = { version = "0.12.0" }
|
||||
fastant = "0.1"
|
||||
|
||||
wildmatch = "2.3.4"
|
||||
|
||||
@@ -339,6 +342,11 @@ futures-util = "0.3.31"
|
||||
maplit = "1.0.2"
|
||||
tempfile = "3.22.0"
|
||||
ctor = "0.8.0"
|
||||
criterion = { version = "0.5", features = ["html_reports"] }
|
||||
|
||||
[[bench]]
|
||||
name = "counter_contention"
|
||||
harness = false
|
||||
|
||||
[target.'cfg(target_os = "linux")'.dev-dependencies]
|
||||
defguard_wireguard_rs = "0.4.2"
|
||||
@@ -373,7 +381,7 @@ full = [
|
||||
"zstd",
|
||||
]
|
||||
wireguard = ["dep:boringtun", "dep:ring"]
|
||||
quic = ["dep:quinn", "dep:quinn-plaintext", "dep:rustls", "dep:rcgen"]
|
||||
quic = ["dep:quinn", "dep:quinn-proto", "dep:seahash", "dep:rustls", "dep:rcgen"]
|
||||
kcp = ["dep:kcp-sys"]
|
||||
mimalloc = ["dep:mimalloc"]
|
||||
aes-gcm = ["dep:aes-gcm"]
|
||||
|
||||
@@ -0,0 +1,443 @@
|
||||
//! Compare counter implementations under tokio-task contention.
|
||||
//!
|
||||
//! Groups:
|
||||
//! - `contention_scaling` : N tokio tasks share one counter, total work fixed.
|
||||
//! Variants: `single_atomic`, `cas_saturating`, `sharded_atomic`,
|
||||
//! `thread_local_cell`, `unsafe_cell` (unsound, for reference only).
|
||||
//! - `single_thread_write`: per-`add` cost with no contention (floor cost).
|
||||
//! - `read_cost` : per-`get()` cost.
|
||||
//! - `counter_handle` : the REAL production hot path. Measures the actual
|
||||
//! `stats_manager::CounterHandle::add` (single-atomic `fetch_add` + lock-free
|
||||
//! fastant `touch`) against reconstructed baselines:
|
||||
//! * `prod` - real `CounterHandle` (this code's version)
|
||||
//! * `baseline_cas_mutex` - pre-optimization: single `AtomicU64` with `fetch_update` (CAS) + `Mutex<Instant>` touch
|
||||
//! * `baseline_fetchadd_mutex` - `fetch_add` + `Mutex<Instant>` touch (isolates the lock-free fastant touch)
|
||||
//!
|
||||
//! Run: `cargo bench -p easytier --bench counter_contention`
|
||||
|
||||
use std::cell::{Cell, UnsafeCell};
|
||||
use std::hint::black_box;
|
||||
use std::sync::Arc;
|
||||
use std::sync::LazyLock;
|
||||
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
|
||||
use std::thread::available_parallelism;
|
||||
use std::time::Instant;
|
||||
|
||||
use criterion::{
|
||||
BenchmarkId, Criterion, Throughput, criterion_group, criterion_main, measurement::WallTime,
|
||||
};
|
||||
use easytier::common::stats_manager::{CounterHandle, MetricName, StatsManager};
|
||||
use parking_lot::Mutex;
|
||||
|
||||
const COUNTER_SHARDS: usize = 16;
|
||||
const TOTAL_WORK: u64 = 8_000_000;
|
||||
// The handle path does a counter update plus a timestamp `touch` per `add`,
|
||||
// so it is heavier per op than the counter-only groups; use a smaller total to
|
||||
// keep the bench fast.
|
||||
const HANDLE_TOTAL_WORK: u64 = 2_000_000;
|
||||
const TASK_COUNTS: &[usize] = &[1, 2, 4, 8, 16, 32];
|
||||
|
||||
trait Counter: Send + Sync {
|
||||
fn add(&self, delta: u64);
|
||||
fn get(&self) -> u64;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 1. SingleAtomic: one atomic, fetch_add. Baseline; contends across cores.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
struct SingleAtomic(AtomicU64);
|
||||
|
||||
impl Default for SingleAtomic {
|
||||
fn default() -> Self {
|
||||
Self(AtomicU64::new(0))
|
||||
}
|
||||
}
|
||||
|
||||
impl Counter for SingleAtomic {
|
||||
#[inline(always)]
|
||||
fn add(&self, delta: u64) {
|
||||
self.0.fetch_add(delta, Ordering::Relaxed);
|
||||
}
|
||||
#[inline(always)]
|
||||
fn get(&self) -> u64 {
|
||||
self.0.load(Ordering::Relaxed)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 2. CasSaturating: fetch_update with saturating_add (the original PR `add`).
|
||||
// A CAS loop that can retry under contention.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
struct CasSaturating(AtomicU64);
|
||||
|
||||
impl Default for CasSaturating {
|
||||
fn default() -> Self {
|
||||
Self(AtomicU64::new(0))
|
||||
}
|
||||
}
|
||||
|
||||
impl Counter for CasSaturating {
|
||||
#[inline(always)]
|
||||
fn add(&self, delta: u64) {
|
||||
let _ = self
|
||||
.0
|
||||
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| {
|
||||
Some(c.saturating_add(delta))
|
||||
});
|
||||
}
|
||||
#[inline(always)]
|
||||
fn get(&self) -> u64 {
|
||||
self.0.load(Ordering::Relaxed)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 3. ShardedAtomic: 16 cache-aligned shards + per-thread shard index.
|
||||
// Comparison-only variant (production `stats_manager::Counter` is
|
||||
// single-atomic; sharding was evaluated and dropped as no benefit for the
|
||||
// default 1-16 worker deployments).
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
thread_local! {
|
||||
static SHARD_IDX: Cell<usize> = Cell::new({
|
||||
static NEXT: AtomicUsize = AtomicUsize::new(0);
|
||||
NEXT.fetch_add(1, Ordering::Relaxed) % COUNTER_SHARDS
|
||||
});
|
||||
}
|
||||
|
||||
#[repr(align(64))]
|
||||
struct Shard {
|
||||
value: AtomicU64,
|
||||
}
|
||||
|
||||
struct ShardedAtomic {
|
||||
shards: Box<[Shard]>,
|
||||
}
|
||||
|
||||
impl Default for ShardedAtomic {
|
||||
fn default() -> Self {
|
||||
let mut shards = Vec::with_capacity(COUNTER_SHARDS);
|
||||
for _ in 0..COUNTER_SHARDS {
|
||||
shards.push(Shard {
|
||||
value: AtomicU64::new(0),
|
||||
});
|
||||
}
|
||||
Self {
|
||||
shards: shards.into_boxed_slice(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Counter for ShardedAtomic {
|
||||
#[inline(always)]
|
||||
fn add(&self, delta: u64) {
|
||||
let i = SHARD_IDX.with(|c| c.get());
|
||||
self.shards[i].value.fetch_add(delta, Ordering::Relaxed);
|
||||
}
|
||||
#[inline(always)]
|
||||
fn get(&self) -> u64 {
|
||||
self.shards
|
||||
.iter()
|
||||
.map(|s| s.value.load(Ordering::Relaxed))
|
||||
.sum()
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 4. ThreadLocalCell: per-thread Cell<u64> accumulation. Zero-atomic writes.
|
||||
// `get()` flushes the caller thread's local into a shared aggregate, so the
|
||||
// measured read cost reflects a flush-based read. Exact totals would require
|
||||
// flushing every thread (not modeled here).
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
thread_local! {
|
||||
static TLS_DELTA: Cell<u64> = const { Cell::new(0) };
|
||||
}
|
||||
|
||||
struct ThreadLocalCell {
|
||||
shared: AtomicU64,
|
||||
}
|
||||
|
||||
impl Default for ThreadLocalCell {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
shared: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Counter for ThreadLocalCell {
|
||||
#[inline(always)]
|
||||
fn add(&self, delta: u64) {
|
||||
TLS_DELTA.with(|c| c.set(c.get() + delta));
|
||||
}
|
||||
#[inline(always)]
|
||||
fn get(&self) -> u64 {
|
||||
let local = TLS_DELTA.with(|c| c.replace(0));
|
||||
self.shared.fetch_add(local, Ordering::Relaxed) + local
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 5. UnsafeCellCounter: a plain u64 mutated through UnsafeCell with manual
|
||||
// `unsafe impl Send/Sync`. This is UNSOUND under concurrent access (data
|
||||
// race / UB) and is exactly what the original code did "for speed". It is
|
||||
// included only to measure the speed ceiling the author was chasing, and to
|
||||
// show that its `get()` returns wrong totals under contention (lost updates).
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
struct UnsafeCellCounter(UnsafeCell<u64>);
|
||||
|
||||
// SAFETY: deliberately unsound; see above.
|
||||
unsafe impl Send for UnsafeCellCounter {}
|
||||
unsafe impl Sync for UnsafeCellCounter {}
|
||||
|
||||
impl Default for UnsafeCellCounter {
|
||||
fn default() -> Self {
|
||||
Self(UnsafeCell::new(0))
|
||||
}
|
||||
}
|
||||
|
||||
impl Counter for UnsafeCellCounter {
|
||||
#[inline(always)]
|
||||
fn add(&self, delta: u64) {
|
||||
// SAFETY: UNSOUND under concurrent access (data race).
|
||||
unsafe {
|
||||
*self.0.get() += delta;
|
||||
}
|
||||
}
|
||||
#[inline(always)]
|
||||
fn get(&self) -> u64 {
|
||||
// SAFETY: UNSOUND under concurrent writers (data race).
|
||||
unsafe { *self.0.get() }
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Production counter handle + reconstructed baselines for the `counter_handle`
|
||||
// group. These measure the full hot path (`add` = counter update + `touch`
|
||||
// timestamp), which is what actually runs per packet in `peer_manager`.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// The real production `CounterHandle`. `CounterHandle::add` does a single
|
||||
// `AtomicU64::fetch_add` then a lock-free fastant `touch`.
|
||||
impl Counter for CounterHandle {
|
||||
#[inline(always)]
|
||||
fn add(&self, delta: u64) {
|
||||
// Fully-qualified to avoid infinite recursion through the trait method.
|
||||
CounterHandle::add(self, delta);
|
||||
}
|
||||
#[inline(always)]
|
||||
fn get(&self) -> u64 {
|
||||
CounterHandle::get(self)
|
||||
}
|
||||
}
|
||||
|
||||
// A faithful replica of the pre-optimization design: a SINGLE `AtomicU64`
|
||||
// (unsharded) plus a `Mutex<Instant>` timestamp. `use_cas` selects whether the
|
||||
// counter write is a `fetch_update` saturating CAS (the original PR `add`) or a
|
||||
// plain `fetch_add`.
|
||||
struct BaselineHandle {
|
||||
counter: AtomicU64,
|
||||
last_updated: Mutex<Instant>,
|
||||
use_cas: bool,
|
||||
}
|
||||
|
||||
impl Default for BaselineHandle {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
counter: AtomicU64::new(0),
|
||||
last_updated: Mutex::new(Instant::now()),
|
||||
use_cas: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Counter for BaselineHandle {
|
||||
#[inline(always)]
|
||||
fn add(&self, delta: u64) {
|
||||
if self.use_cas {
|
||||
let _ = self
|
||||
.counter
|
||||
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| {
|
||||
Some(c.saturating_add(delta))
|
||||
});
|
||||
} else {
|
||||
self.counter.fetch_add(delta, Ordering::Relaxed);
|
||||
}
|
||||
*self.last_updated.lock() = Instant::now();
|
||||
}
|
||||
#[inline(always)]
|
||||
fn get(&self) -> u64 {
|
||||
self.counter.load(Ordering::Relaxed)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Harness: a shared multi-thread tokio runtime sized to the host's parallelism.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
|
||||
let workers = available_parallelism().map(|n| n.get()).unwrap_or(1);
|
||||
tokio::runtime::Builder::new_multi_thread()
|
||||
.worker_threads(workers)
|
||||
.enable_all()
|
||||
.build()
|
||||
.expect("failed to build tokio runtime")
|
||||
});
|
||||
|
||||
fn bench_contention<C: Counter + Default + 'static>(
|
||||
group: &mut criterion::BenchmarkGroup<'_, WallTime>,
|
||||
name: &str,
|
||||
n_tasks: usize,
|
||||
per_task: u64,
|
||||
) {
|
||||
let counter: Arc<C> = Arc::new(C::default());
|
||||
group.bench_with_input(BenchmarkId::new(name, n_tasks), &n_tasks, |b, &n| {
|
||||
b.iter(|| {
|
||||
let counter = counter.clone();
|
||||
RUNTIME.block_on(async move {
|
||||
let mut handles = Vec::with_capacity(n);
|
||||
for _ in 0..n {
|
||||
let c = counter.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
for _ in 0..per_task {
|
||||
c.add(black_box(1));
|
||||
}
|
||||
}));
|
||||
}
|
||||
for handle in handles {
|
||||
let _ = handle.await;
|
||||
}
|
||||
black_box(counter.get());
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
fn contention_scaling(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("contention_scaling");
|
||||
group.throughput(Throughput::Elements(TOTAL_WORK));
|
||||
for &n in TASK_COUNTS {
|
||||
let per = TOTAL_WORK / n as u64;
|
||||
bench_contention::<SingleAtomic>(&mut group, "single_atomic", n, per);
|
||||
bench_contention::<CasSaturating>(&mut group, "cas_saturating", n, per);
|
||||
bench_contention::<ShardedAtomic>(&mut group, "sharded_atomic", n, per);
|
||||
bench_contention::<ThreadLocalCell>(&mut group, "thread_local_cell", n, per);
|
||||
bench_contention::<UnsafeCellCounter>(&mut group, "unsafe_cell", n, per);
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn single_thread_write<C: Counter + Default>(
|
||||
group: &mut criterion::BenchmarkGroup<'_, WallTime>,
|
||||
name: &str,
|
||||
) {
|
||||
let counter = C::default();
|
||||
group.bench_function(name, |b| {
|
||||
b.iter(|| {
|
||||
counter.add(black_box(1));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
fn single_thread_write_group(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("single_thread_write");
|
||||
group.throughput(Throughput::Elements(1));
|
||||
single_thread_write::<SingleAtomic>(&mut group, "single_atomic");
|
||||
single_thread_write::<CasSaturating>(&mut group, "cas_saturating");
|
||||
single_thread_write::<ShardedAtomic>(&mut group, "sharded_atomic");
|
||||
single_thread_write::<ThreadLocalCell>(&mut group, "thread_local_cell");
|
||||
single_thread_write::<UnsafeCellCounter>(&mut group, "unsafe_cell");
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn read_cost<C: Counter + Default>(
|
||||
group: &mut criterion::BenchmarkGroup<'_, WallTime>,
|
||||
name: &str,
|
||||
) {
|
||||
let counter = C::default();
|
||||
counter.add(1000);
|
||||
group.bench_function(name, |b| {
|
||||
b.iter(|| black_box(counter.get()));
|
||||
});
|
||||
}
|
||||
|
||||
fn read_cost_group(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("read_cost");
|
||||
read_cost::<SingleAtomic>(&mut group, "single_atomic");
|
||||
read_cost::<CasSaturating>(&mut group, "cas_saturating");
|
||||
read_cost::<ShardedAtomic>(&mut group, "sharded_atomic");
|
||||
read_cost::<ThreadLocalCell>(&mut group, "thread_local_cell");
|
||||
read_cost::<UnsafeCellCounter>(&mut group, "unsafe_cell");
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn bench_handle(
|
||||
group: &mut criterion::BenchmarkGroup<'_, WallTime>,
|
||||
name: &str,
|
||||
n_tasks: usize,
|
||||
per_task: u64,
|
||||
counter: Arc<dyn Counter>,
|
||||
) {
|
||||
group.bench_with_input(BenchmarkId::new(name, n_tasks), &n_tasks, |b, &n| {
|
||||
b.iter(|| {
|
||||
let counter = counter.clone();
|
||||
RUNTIME.block_on(async move {
|
||||
let mut handles = Vec::with_capacity(n);
|
||||
for _ in 0..n {
|
||||
let c = counter.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
for _ in 0..per_task {
|
||||
c.add(black_box(1));
|
||||
}
|
||||
}));
|
||||
}
|
||||
for handle in handles {
|
||||
let _ = handle.await;
|
||||
}
|
||||
black_box(counter.get());
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
fn counter_handle(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("counter_handle");
|
||||
group.throughput(Throughput::Elements(HANDLE_TOTAL_WORK));
|
||||
|
||||
// StatsManager::new() spawns a background cleanup task, which needs a tokio
|
||||
// runtime context; bind it to our shared RUNTIME for the lifetime of the
|
||||
// group.
|
||||
let _rt_guard = RUNTIME.enter();
|
||||
let stats = StatsManager::new();
|
||||
|
||||
let prod: Arc<dyn Counter> = Arc::new(stats.get_simple_counter(MetricName::TrafficBytesTx));
|
||||
let cas: Arc<dyn Counter> = Arc::new(BaselineHandle {
|
||||
use_cas: true,
|
||||
..Default::default()
|
||||
});
|
||||
let fam: Arc<dyn Counter> = Arc::new(BaselineHandle {
|
||||
use_cas: false,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
for &n in TASK_COUNTS {
|
||||
let per = HANDLE_TOTAL_WORK / n as u64;
|
||||
bench_handle(&mut group, "prod", n, per, prod.clone());
|
||||
bench_handle(&mut group, "baseline_cas_mutex", n, per, cas.clone());
|
||||
bench_handle(&mut group, "baseline_fetchadd_mutex", n, per, fam.clone());
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
// Keep the default measurement config; pass CLI flags to speed up a run, e.g.
|
||||
// `-- --measurement-time 2 --sample-size 30 --warm-up-time 500`.
|
||||
criterion_group! {
|
||||
name = benches;
|
||||
config = Criterion::default();
|
||||
targets = contention_scaling, single_thread_write_group, read_cost_group, counter_handle
|
||||
}
|
||||
criterion_main!(benches);
|
||||
@@ -114,11 +114,11 @@ core_clap:
|
||||
en: "encryption algorithm to use, supported: '', 'xor', 'chacha20', 'aes-gcm', 'aes-gcm-256', 'openssl-aes128-gcm', 'openssl-aes256-gcm', 'openssl-chacha20'. Empty string means default (aes-gcm)"
|
||||
zh-CN: "要使用的加密算法，支持：''（默认aes-gcm）、'xor'、'chacha20'、'aes-gcm'、'aes-gcm-256'、'openssl-aes128-gcm'、'openssl-aes256-gcm'、'openssl-chacha20'"
|
||||
multi_thread:
|
||||
en: "use multi-thread runtime, default is single-thread"
|
||||
zh-CN: "使用多线程运行时，默认为单线程"
|
||||
en: "multi-thread tokio runtime (default on). Only affects launcher-based deployments (GUI/mobile/web/Windows service); the easytier-core CLI always runs single-threaded."
|
||||
zh-CN: "多线程 tokio 运行时（默认开启）。仅对 launcher 部署（GUI/移动端/web/Windows 服务）生效；easytier-core CLI 始终为单线程。"
|
||||
multi_thread_count:
|
||||
en: "the number of threads to use, default is 2, only effective when multi-thread is enabled, must be greater than 2"
|
||||
zh-CN: "使用的线程数，默认为2，仅在多线程模式下有效。取值必须大于2"
|
||||
en: "the number of worker threads, default 2, only effective when multi-thread is enabled, minimum 2"
|
||||
zh-CN: "worker 线程数，默认 2，仅在启用多线程时生效，最小为 2"
|
||||
disable_ipv6:
|
||||
en: "do not use ipv6"
|
||||
zh-CN: "不使用IPv6"
|
||||
|
||||
+158
-17
@@ -6,6 +6,7 @@ use std::{
|
||||
};
|
||||
|
||||
use anyhow::Context;
|
||||
use ariadne::{CharSet, Config as AriadneConfig, IndexType, Label, Report, ReportKind, Source};
|
||||
use base64::{Engine as _, prelude::BASE64_STANDARD};
|
||||
use clap::ValueEnum;
|
||||
use clap::builder::PossibleValue;
|
||||
@@ -569,6 +570,35 @@ struct Config {
|
||||
source: Option<ConfigSourceConfig>,
|
||||
}
|
||||
|
||||
fn format_toml_parse_error(source_name: &str, config_str: &str, error: &toml::de::Error) -> String {
|
||||
let message = format!("failed to parse config TOML from {source_name}");
|
||||
|
||||
let Some(span) = error.span() else {
|
||||
return format!("{message}\ndetail: {error}");
|
||||
};
|
||||
|
||||
let mut output = Vec::new();
|
||||
let report = Report::build(ReportKind::Error, (source_name, span.clone()))
|
||||
.with_config(
|
||||
AriadneConfig::default()
|
||||
.with_color(false)
|
||||
.with_char_set(CharSet::Ascii)
|
||||
.with_index_type(IndexType::Byte),
|
||||
)
|
||||
.with_message(&message)
|
||||
.with_label(Label::new((source_name, span)).with_message(error.message()))
|
||||
.finish();
|
||||
|
||||
if report
|
||||
.write((source_name, Source::from(config_str)), &mut output)
|
||||
.is_ok()
|
||||
{
|
||||
String::from_utf8_lossy(&output).into_owned()
|
||||
} else {
|
||||
format!("{message}\ndetail: {error}")
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TomlConfigLoader {
|
||||
config: Arc<Mutex<Config>>,
|
||||
@@ -591,12 +621,39 @@ impl TomlConfigLoader {
|
||||
}
|
||||
|
||||
pub fn new_from_str(config_str: &str) -> Result<Self, anyhow::Error> {
|
||||
let mut config = toml::de::from_str::<Config>(config_str)
|
||||
.with_context(|| format!("failed to parse config file: {}", config_str))?;
|
||||
Self::new_from_str_with_source("inline config", config_str)
|
||||
}
|
||||
|
||||
pub fn new(config_path: &PathBuf) -> Result<Self, anyhow::Error> {
|
||||
let config_str = std::fs::read_to_string(config_path)
|
||||
.with_context(|| format!("failed to read config file: {}", config_path.display()))?;
|
||||
|
||||
let source_name = config_path.display().to_string();
|
||||
Self::new_from_str_with_source(&source_name, &config_str)
|
||||
}
|
||||
|
||||
pub(crate) fn new_from_str_with_source(
|
||||
source_name: &str,
|
||||
config_str: &str,
|
||||
) -> Result<Self, anyhow::Error> {
|
||||
let mut config = toml::de::from_str::<Config>(config_str).map_err(|err| {
|
||||
let message = format_toml_parse_error(source_name, config_str, &err);
|
||||
anyhow::Error::new(err).context(message)
|
||||
})?;
|
||||
|
||||
Self::normalize_config_source(&mut config);
|
||||
|
||||
config.flags_struct = Some(Self::gen_flags(config.flags.clone().unwrap_or_default()));
|
||||
Self::new_from_config(config).map_err(|err| {
|
||||
let message = format!("failed to load config from {source_name}: {err}");
|
||||
err.context(message)
|
||||
})
|
||||
}
|
||||
|
||||
fn new_from_config(mut config: Config) -> Result<Self, anyhow::Error> {
|
||||
config.flags_struct = Some(
|
||||
Self::gen_flags(config.flags.clone().unwrap_or_default())
|
||||
.context("failed to parse flags")?,
|
||||
);
|
||||
let has_network_identity = config.network_identity.is_some();
|
||||
|
||||
let config = TomlConfigLoader {
|
||||
@@ -628,21 +685,15 @@ impl TomlConfigLoader {
|
||||
Ok(config)
|
||||
}
|
||||
|
||||
pub fn new(config_path: &PathBuf) -> Result<Self, anyhow::Error> {
|
||||
let config_str = std::fs::read_to_string(config_path)
|
||||
.with_context(|| format!("failed to read config file: {:?}", config_path))?;
|
||||
let ret = Self::new_from_str(&config_str)?;
|
||||
|
||||
Ok(ret)
|
||||
}
|
||||
|
||||
fn gen_flags(flags_hashmap: serde_json::Map<String, serde_json::Value>) -> Flags {
|
||||
fn gen_flags(
|
||||
flags_hashmap: serde_json::Map<String, serde_json::Value>,
|
||||
) -> serde_json::Result<Flags> {
|
||||
let mut merged_hashmap = match serde_json::to_value(gen_default_flags()) {
|
||||
Ok(serde_json::Value::Object(map)) => map,
|
||||
_ => serde_json::Map::new(),
|
||||
};
|
||||
merged_hashmap.extend(flags_hashmap);
|
||||
serde_json::from_value(serde_json::Value::Object(merged_hashmap)).unwrap()
|
||||
serde_json::from_value(serde_json::Value::Object(merged_hashmap))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1199,13 +1250,13 @@ pub async fn load_config_from_file(
|
||||
.read_to_string(&mut stdin)
|
||||
.await
|
||||
.context("failed to read config from stdin")?;
|
||||
let config = TomlConfigLoader::new_from_str(&stdin)?;
|
||||
let config = TomlConfigLoader::new_from_str_with_source("stdin", &stdin)?;
|
||||
return Ok((config, ConfigFileControl::STATIC_CONFIG));
|
||||
}
|
||||
|
||||
let config_str = tokio::fs::read_to_string(config_file)
|
||||
.await
|
||||
.with_context(|| format!("failed to read config file: {:?}", config_file))?;
|
||||
.with_context(|| format!("failed to read config file: {}", config_file.display()))?;
|
||||
|
||||
let (expanded_config_str, uses_env_vars) = if disable_env_parsing {
|
||||
(config_str.clone(), false)
|
||||
@@ -1227,8 +1278,8 @@ pub async fn load_config_from_file(
|
||||
);
|
||||
}
|
||||
|
||||
let config = TomlConfigLoader::new_from_str(&expanded_config_str)
|
||||
.with_context(|| format!("failed to load config file: {:?}", config_file))?;
|
||||
let source_name = config_file.display().to_string();
|
||||
let config = TomlConfigLoader::new_from_str_with_source(&source_name, &expanded_config_str)?;
|
||||
|
||||
let mut control = ConfigFileControl::from_path(config_file.clone()).await;
|
||||
|
||||
@@ -1268,6 +1319,96 @@ pub mod tests {
|
||||
use std::path::PathBuf;
|
||||
use tempfile::NamedTempFile;
|
||||
|
||||
#[test]
|
||||
fn invalid_toml_error_includes_location_and_source_line() {
|
||||
let error = TomlConfigLoader::new_from_str("dhcp = \"yes\"").unwrap_err();
|
||||
let display = error.to_string();
|
||||
|
||||
assert!(display.contains("failed to parse config TOML"));
|
||||
assert!(display.contains("inline config"));
|
||||
assert!(display.contains("dhcp = \"yes\""));
|
||||
assert!(display.contains("^"));
|
||||
assert!(display.contains("invalid type: string"));
|
||||
assert!(!display.contains("<unknown>"));
|
||||
assert!(
|
||||
error
|
||||
.chain()
|
||||
.any(|err| err.downcast_ref::<toml::de::Error>().is_some())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_file_toml_error_includes_config_source() {
|
||||
let mut config_file = NamedTempFile::new().unwrap();
|
||||
writeln!(config_file, "dhcp = \"yes\"").unwrap();
|
||||
|
||||
let error = TomlConfigLoader::new(&config_file.path().to_path_buf()).unwrap_err();
|
||||
let error = error.to_string();
|
||||
|
||||
assert!(error.contains(config_file.path().to_string_lossy().as_ref()));
|
||||
assert!(error.contains("failed to parse config TOML"));
|
||||
assert!(error.contains("dhcp = \"yes\""));
|
||||
assert!(error.contains("^"));
|
||||
assert!(error.contains("invalid type: string"));
|
||||
assert!(!error.contains("<unknown>"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_stdin_toml_error_includes_config_source_in_display() {
|
||||
let error = TomlConfigLoader::new_from_str_with_source("stdin", "dhcp = \"yes\"")
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
|
||||
assert!(error.contains("stdin"));
|
||||
assert!(error.contains("failed to parse config TOML"));
|
||||
assert!(error.contains("dhcp = \"yes\""));
|
||||
assert!(error.contains("^"));
|
||||
assert!(error.contains("invalid type: string"));
|
||||
assert!(!error.contains("<unknown>"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_toml_error_handles_non_ascii_before_error() {
|
||||
let error = TomlConfigLoader::new_from_str("hostname = \"节点\"\ndhcp = \"yes\"")
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
|
||||
assert!(error.contains("dhcp = \"yes\""));
|
||||
assert!(error.contains("^"));
|
||||
assert!(error.contains("invalid type: string"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_toml_error_handles_non_ascii_before_error_on_same_line() {
|
||||
let error = TomlConfigLoader::new_from_str("hostname = \"节点\" dhcp = \"yes\"")
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
|
||||
assert!(error.contains("failed to parse config TOML"));
|
||||
assert!(error.contains("inline config:1:"));
|
||||
assert!(error.contains("hostname = \"节点\" dhcp = \"yes\""));
|
||||
assert!(error.contains("expected newline"));
|
||||
assert!(!error.contains("<unknown>"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_file_flags_error_includes_config_source_in_display() {
|
||||
let mut config_file = NamedTempFile::new().unwrap();
|
||||
writeln!(config_file, "[flags]").unwrap();
|
||||
writeln!(config_file, "socket_mark = \"bad\"").unwrap();
|
||||
|
||||
let error = TomlConfigLoader::new(&config_file.path().to_path_buf()).unwrap_err();
|
||||
|
||||
let display = error.to_string();
|
||||
assert!(display.contains(config_file.path().to_string_lossy().as_ref()));
|
||||
assert!(display.contains("failed to load config"));
|
||||
assert!(display.contains("failed to parse flags"));
|
||||
|
||||
// with_context preserves the cause chain so callers can inspect the root reason.
|
||||
let chain: Vec<String> = error.chain().map(|e| e.to_string()).collect();
|
||||
assert!(chain.iter().any(|m| m.contains("failed to parse flags")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn socket_mark_config_file_roundtrip_none_some_and_zero() {
|
||||
// Omitting the flag leaves socket_mark unset (None) -> SO_MARK untouched.
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
use dashmap::DashMap;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::cell::UnsafeCell;
|
||||
use std::fmt;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
use std::sync::OnceLock;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::Duration;
|
||||
use tokio::time::interval;
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
|
||||
@@ -374,136 +375,106 @@ impl Default for LabelSet {
|
||||
}
|
||||
}
|
||||
|
||||
/// UnsafeCounter provides a high-performance counter using UnsafeCell
|
||||
/// Counter provides a high-performance atomic counter
|
||||
#[derive(Debug)]
|
||||
pub struct UnsafeCounter {
|
||||
value: UnsafeCell<u64>,
|
||||
pub struct Counter {
|
||||
value: AtomicU64,
|
||||
}
|
||||
|
||||
impl Default for UnsafeCounter {
|
||||
impl Default for Counter {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl UnsafeCounter {
|
||||
impl Counter {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
value: UnsafeCell::new(0),
|
||||
value: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new_with_value(initial: u64) -> Self {
|
||||
Self {
|
||||
value: UnsafeCell::new(initial),
|
||||
value: AtomicU64::new(initial),
|
||||
}
|
||||
}
|
||||
|
||||
/// Increment the counter by the given amount
|
||||
/// # Safety
|
||||
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
|
||||
/// that no other thread is accessing this counter simultaneously.
|
||||
pub unsafe fn add(&self, delta: u64) {
|
||||
let ptr = self.value.get();
|
||||
unsafe {
|
||||
*ptr = (*ptr).saturating_add(delta);
|
||||
}
|
||||
pub fn add(&self, delta: u64) {
|
||||
self.value.fetch_add(delta, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Increment the counter by 1
|
||||
/// # Safety
|
||||
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
|
||||
/// that no other thread is accessing this counter simultaneously.
|
||||
pub unsafe fn inc(&self) {
|
||||
unsafe {
|
||||
self.add(1);
|
||||
}
|
||||
pub fn inc(&self) {
|
||||
self.add(1);
|
||||
}
|
||||
|
||||
/// Get the current value of the counter
|
||||
/// # Safety
|
||||
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
|
||||
/// that no other thread is modifying this counter simultaneously.
|
||||
pub unsafe fn get(&self) -> u64 {
|
||||
let ptr = self.value.get();
|
||||
unsafe { *ptr }
|
||||
pub fn get(&self) -> u64 {
|
||||
self.value.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
/// Reset the counter to zero
|
||||
/// # Safety
|
||||
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
|
||||
/// that no other thread is accessing this counter simultaneously.
|
||||
pub unsafe fn reset(&self) {
|
||||
let ptr = self.value.get();
|
||||
unsafe {
|
||||
*ptr = 0;
|
||||
}
|
||||
pub fn reset(&self) {
|
||||
self.value.store(0, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Set the counter to a specific value
|
||||
/// # Safety
|
||||
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
|
||||
/// that no other thread is accessing this counter simultaneously.
|
||||
pub unsafe fn set(&self, value: u64) {
|
||||
let ptr = self.value.get();
|
||||
unsafe {
|
||||
*ptr = value;
|
||||
}
|
||||
pub fn set(&self, value: u64) {
|
||||
self.value.store(value, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
// UnsafeCounter is Send + Sync because the safety is guaranteed by the caller
|
||||
unsafe impl Send for UnsafeCounter {}
|
||||
unsafe impl Sync for UnsafeCounter {}
|
||||
/// Epoch used to convert a monotonic clock reading into a storable `u64`
|
||||
/// millisecond count for `MetricData::last_updated`. Lazily initialized on first
|
||||
/// use. Backed by `fastant`, which uses the TSC on x86_64 Linux (and falls back
|
||||
/// to `std::time::Instant` elsewhere), making `now_millis()` cheap enough to
|
||||
/// call per packet.
|
||||
fn time_base() -> fastant::Instant {
|
||||
static BASE: OnceLock<fastant::Instant> = OnceLock::new();
|
||||
*BASE.get_or_init(fastant::Instant::now)
|
||||
}
|
||||
|
||||
fn now_millis() -> u64 {
|
||||
fastant::Instant::now()
|
||||
.saturating_duration_since(time_base())
|
||||
.as_millis() as u64
|
||||
}
|
||||
|
||||
/// MetricData contains both the counter and last update timestamp
|
||||
/// Uses UnsafeCell for lock-free access
|
||||
#[derive(Debug)]
|
||||
struct MetricData {
|
||||
counter: UnsafeCounter,
|
||||
last_updated: UnsafeCell<Instant>,
|
||||
counter: Counter,
|
||||
last_updated: AtomicU64,
|
||||
}
|
||||
|
||||
impl MetricData {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
counter: UnsafeCounter::new(),
|
||||
last_updated: UnsafeCell::new(Instant::now()),
|
||||
counter: Counter::new(),
|
||||
last_updated: AtomicU64::new(now_millis()),
|
||||
}
|
||||
}
|
||||
|
||||
fn new_with_value(initial: u64) -> Self {
|
||||
Self {
|
||||
counter: UnsafeCounter::new_with_value(initial),
|
||||
last_updated: UnsafeCell::new(Instant::now()),
|
||||
counter: Counter::new_with_value(initial),
|
||||
last_updated: AtomicU64::new(now_millis()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Update the last_updated timestamp
|
||||
/// # Safety
|
||||
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
|
||||
/// that no other thread is accessing this timestamp simultaneously.
|
||||
unsafe fn touch(&self) {
|
||||
let ptr = self.last_updated.get();
|
||||
unsafe {
|
||||
*ptr = Instant::now();
|
||||
}
|
||||
/// Update the last_updated timestamp. Lock-free.
|
||||
fn touch(&self) {
|
||||
self.last_updated.store(now_millis(), Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Get the last updated timestamp
|
||||
/// # Safety
|
||||
/// This method is unsafe because it uses UnsafeCell. The caller must ensure
|
||||
/// that no other thread is modifying this timestamp simultaneously.
|
||||
unsafe fn get_last_updated(&self) -> Instant {
|
||||
let ptr = self.last_updated.get();
|
||||
unsafe { *ptr }
|
||||
/// Last update time as milliseconds since `time_base()`.
|
||||
fn last_updated_millis(&self) -> u64 {
|
||||
self.last_updated.load(Ordering::Relaxed)
|
||||
}
|
||||
}
|
||||
|
||||
// MetricData is Send + Sync because the safety is guaranteed by the caller
|
||||
unsafe impl Send for MetricData {}
|
||||
unsafe impl Sync for MetricData {}
|
||||
|
||||
/// MetricKey uniquely identifies a metric with its name and labels
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
|
||||
struct MetricKey {
|
||||
@@ -546,39 +517,31 @@ impl CounterHandle {
|
||||
|
||||
/// Increment the counter by the given amount
|
||||
pub fn add(&self, delta: u64) {
|
||||
unsafe {
|
||||
self.metric_data.counter.add(delta);
|
||||
self.metric_data.touch();
|
||||
}
|
||||
self.metric_data.counter.add(delta);
|
||||
self.metric_data.touch();
|
||||
}
|
||||
|
||||
/// Increment the counter by 1
|
||||
pub fn inc(&self) {
|
||||
unsafe {
|
||||
self.metric_data.counter.inc();
|
||||
self.metric_data.touch();
|
||||
}
|
||||
self.metric_data.counter.inc();
|
||||
self.metric_data.touch();
|
||||
}
|
||||
|
||||
/// Get the current value of the counter
|
||||
pub fn get(&self) -> u64 {
|
||||
unsafe { self.metric_data.counter.get() }
|
||||
self.metric_data.counter.get()
|
||||
}
|
||||
|
||||
/// Reset the counter to zero
|
||||
pub fn reset(&self) {
|
||||
unsafe {
|
||||
self.metric_data.counter.reset();
|
||||
self.metric_data.touch();
|
||||
}
|
||||
self.metric_data.counter.reset();
|
||||
self.metric_data.touch();
|
||||
}
|
||||
|
||||
/// Set the counter to a specific value
|
||||
pub fn set(&self, value: u64) {
|
||||
unsafe {
|
||||
self.metric_data.counter.set(value);
|
||||
self.metric_data.touch();
|
||||
}
|
||||
self.metric_data.counter.set(value);
|
||||
self.metric_data.touch();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -614,9 +577,15 @@ impl StatsManager {
|
||||
loop {
|
||||
interval.tick().await;
|
||||
|
||||
let Some(cutoff_time) = Instant::now().checked_sub(Duration::from_secs(180)) else {
|
||||
continue;
|
||||
};
|
||||
// Drop metrics untouched for 180s and with no live handles.
|
||||
// Compare in the millis-since-base domain so neither the hot
|
||||
// path nor GC reconstructs an `Instant` or locks.
|
||||
//
|
||||
// Use an age-based check (`now - last < STALE`) rather than
|
||||
// `last > now - STALE`: early in process life `now_millis()` is
|
||||
// tiny, so `now - STALE` saturates to 0 and a metric stamped at
|
||||
// 0 would fail a strict `> 0` test and be wrongly evicted.
|
||||
let now = now_millis();
|
||||
|
||||
let Some(counters) = counters_clone.upgrade() else {
|
||||
break;
|
||||
@@ -624,7 +593,7 @@ impl StatsManager {
|
||||
|
||||
counters.retain(|_, metric_data: &mut Arc<MetricData>| {
|
||||
Arc::strong_count(metric_data) > 1
|
||||
|| unsafe { metric_data.get_last_updated() > cutoff_time }
|
||||
|| now.saturating_sub(metric_data.last_updated_millis()) < 180_000
|
||||
});
|
||||
counters.shrink_to_fit();
|
||||
}
|
||||
@@ -662,7 +631,7 @@ impl StatsManager {
|
||||
let key = entry.key();
|
||||
let metric_data = entry.value();
|
||||
|
||||
let value = unsafe { metric_data.counter.get() };
|
||||
let value = metric_data.counter.get();
|
||||
|
||||
metrics.push(MetricSnapshot {
|
||||
name: key.name,
|
||||
@@ -695,7 +664,7 @@ impl StatsManager {
|
||||
let key = MetricKey::new(name, labels.clone());
|
||||
|
||||
if let Some(metric_data) = self.counters.get(&key) {
|
||||
let value = unsafe { metric_data.counter.get() };
|
||||
let value = metric_data.counter.get();
|
||||
Some(MetricSnapshot {
|
||||
name,
|
||||
labels: labels.clone(),
|
||||
@@ -793,20 +762,18 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_unsafe_counter() {
|
||||
let counter = UnsafeCounter::new();
|
||||
async fn test_counter() {
|
||||
let counter = Counter::new();
|
||||
|
||||
unsafe {
|
||||
assert_eq!(counter.get(), 0);
|
||||
counter.inc();
|
||||
assert_eq!(counter.get(), 1);
|
||||
counter.add(5);
|
||||
assert_eq!(counter.get(), 6);
|
||||
counter.set(10);
|
||||
assert_eq!(counter.get(), 10);
|
||||
counter.reset();
|
||||
assert_eq!(counter.get(), 0);
|
||||
}
|
||||
assert_eq!(counter.get(), 0);
|
||||
counter.inc();
|
||||
assert_eq!(counter.get(), 1);
|
||||
counter.add(5);
|
||||
assert_eq!(counter.get(), 6);
|
||||
counter.set(10);
|
||||
assert_eq!(counter.get(), 10);
|
||||
counter.reset();
|
||||
assert_eq!(counter.get(), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -947,12 +914,14 @@ mod tests {
|
||||
let counter = stats.get_simple_counter(MetricName::TrafficBytesForwarded);
|
||||
counter.set(1);
|
||||
|
||||
let cutoff_time = Instant::now().checked_add(Duration::from_secs(1)).unwrap();
|
||||
// Cutoff 1s in the future, so every metric is stale by timestamp; only
|
||||
// a live handle keeps a metric.
|
||||
let cutoff_millis = now_millis() + 1_000;
|
||||
stats
|
||||
.counters
|
||||
.retain(|_, metric_data: &mut Arc<MetricData>| {
|
||||
Arc::strong_count(metric_data) > 1
|
||||
|| unsafe { metric_data.get_last_updated() > cutoff_time }
|
||||
|| metric_data.last_updated_millis() > cutoff_millis
|
||||
});
|
||||
|
||||
assert_eq!(stats.metric_count(), 1);
|
||||
@@ -963,11 +932,33 @@ mod tests {
|
||||
.counters
|
||||
.retain(|_, metric_data: &mut Arc<MetricData>| {
|
||||
Arc::strong_count(metric_data) > 1
|
||||
|| unsafe { metric_data.get_last_updated() > cutoff_time }
|
||||
|| metric_data.last_updated_millis() > cutoff_millis
|
||||
});
|
||||
assert_eq!(stats.metric_count(), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_counter_handle_concurrent_increment() {
|
||||
const THREADS: usize = 8;
|
||||
const INCREMENTS_PER_THREAD: usize = 10_000;
|
||||
|
||||
let stats = StatsManager::new();
|
||||
let counter = stats.get_simple_counter(MetricName::TrafficPacketsForwarded);
|
||||
|
||||
std::thread::scope(|scope| {
|
||||
for _ in 0..THREADS {
|
||||
let counter = counter.clone();
|
||||
scope.spawn(move || {
|
||||
for _ in 0..INCREMENTS_PER_THREAD {
|
||||
counter.inc();
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
assert_eq!(counter.get(), (THREADS * INCREMENTS_PER_THREAD) as u64);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_stats_rpc_data_structures() {
|
||||
// Test GetStatsRequest
|
||||
|
||||
@@ -1614,7 +1614,7 @@ pub async fn main() -> ExitCode {
|
||||
// Verify configurations
|
||||
if cli.check_config {
|
||||
if let Err(error) = validate_config(&cli).await {
|
||||
log::error!(?error, "Config validation failed");
|
||||
log::error!(%error, "Config validation failed");
|
||||
return ExitCode::FAILURE;
|
||||
} else {
|
||||
return ExitCode::SUCCESS;
|
||||
@@ -1624,7 +1624,7 @@ pub async fn main() -> ExitCode {
|
||||
let mut ret_code = 0;
|
||||
|
||||
if let Err(error) = run_main(cli).await {
|
||||
log::error!(?error);
|
||||
log::error!(%error);
|
||||
ret_code = 1;
|
||||
}
|
||||
|
||||
@@ -1644,12 +1644,13 @@ async fn validate_config(cli: &Cli) -> anyhow::Result<()> {
|
||||
for config_file in config_files {
|
||||
if config_file == &PathBuf::from("-") {
|
||||
let mut stdin = String::new();
|
||||
_ = tokio::io::stdin().read_to_string(&mut stdin).await?;
|
||||
TomlConfigLoader::new_from_str(stdin.as_str())
|
||||
.with_context(|| "config source: stdin")?;
|
||||
_ = tokio::io::stdin()
|
||||
.read_to_string(&mut stdin)
|
||||
.await
|
||||
.context("failed to read config from stdin")?;
|
||||
TomlConfigLoader::new_from_str_with_source("stdin", stdin.as_str())?;
|
||||
} else {
|
||||
TomlConfigLoader::new(config_file)
|
||||
.with_context(|| format!("config source: {:?}", config_file))?;
|
||||
TomlConfigLoader::new(config_file)?;
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -23,6 +23,9 @@ pub static malloc_conf: &[u8] = b"retain:false\0";
|
||||
|
||||
rust_i18n::i18n!("locales", fallback = "en");
|
||||
|
||||
// The easytier-core CLI intentionally uses a single-thread runtime. The
|
||||
// `multi_thread` flag only affects launcher-based deployments (GUI / mobile /
|
||||
// web / Windows service); see launcher.rs:223.
|
||||
#[tokio::main(flavor = "current_thread")]
|
||||
async fn main() -> std::process::ExitCode {
|
||||
core::main().await
|
||||
|
||||
@@ -14,7 +14,7 @@ use std::{
|
||||
};
|
||||
|
||||
use dashmap::{DashMap, DashSet};
|
||||
use guarden::defer;
|
||||
use guarden::{Guard, defer};
|
||||
use tokio::{
|
||||
sync::{
|
||||
Mutex,
|
||||
|
||||
@@ -33,7 +33,6 @@ use super::{
|
||||
peer_session::{PeerSession, PeerSessionAction},
|
||||
traffic_metrics::AggregateTrafficMetrics,
|
||||
};
|
||||
use crate::utils::BoxExt;
|
||||
use crate::{
|
||||
common::{
|
||||
PeerId,
|
||||
@@ -380,9 +379,9 @@ impl PeerConn {
|
||||
session_filter,
|
||||
noise_handshake_result: None,
|
||||
|
||||
tunnel: Arc::new(Mutex::new(
|
||||
guard!([mut mpsc_tunnel] mpsc_tunnel.close()).boxed(),
|
||||
)),
|
||||
tunnel: Arc::new(Mutex::new(Box::new(
|
||||
guard!([mut mpsc_tunnel] mpsc_tunnel.close()),
|
||||
))),
|
||||
sink,
|
||||
recv: Mutex::new(Some(recv)),
|
||||
tunnel_info,
|
||||
|
||||
@@ -1533,9 +1533,22 @@ impl PeerManager {
|
||||
) -> Result<(), Error> {
|
||||
let policy =
|
||||
Self::get_next_hop_policy(msg.peer_manager_header().unwrap().is_latency_first());
|
||||
let is_latency_first = msg.peer_manager_header().unwrap().is_latency_first();
|
||||
let packet_type = msg.peer_manager_header().unwrap().packet_type;
|
||||
let msg_len = msg.buf_len() as u64;
|
||||
let send_result = if peers.has_peer(dst_peer_id) {
|
||||
let latency_first_gateway = if is_latency_first {
|
||||
peers
|
||||
.get_gateway_peer_id(dst_peer_id, policy.clone())
|
||||
.await
|
||||
.filter(|gateway| *gateway != dst_peer_id)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let send_result = if let Some(gateway) = latency_first_gateway
|
||||
&& (peers.has_peer(gateway) || foreign_network_client.has_next_hop(gateway))
|
||||
{
|
||||
relay_peer_map.send_msg(msg, dst_peer_id, policy).await
|
||||
} else if peers.has_peer(dst_peer_id) {
|
||||
peers.send_msg_directly(msg, dst_peer_id).await
|
||||
} else if foreign_network_client.has_next_hop(dst_peer_id) {
|
||||
foreign_network_client.send_msg(msg, dst_peer_id).await
|
||||
@@ -2185,6 +2198,7 @@ impl PeerManager {
|
||||
mod tests {
|
||||
use base64::Engine;
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
fmt::Debug,
|
||||
sync::Arc,
|
||||
time::{Duration, Instant},
|
||||
@@ -2192,6 +2206,7 @@ mod tests {
|
||||
|
||||
use crate::{
|
||||
common::{
|
||||
PeerId,
|
||||
config::Flags,
|
||||
global_ctx::{NetworkIdentity, tests::get_mock_global_ctx},
|
||||
stats_manager::{LabelSet, LabelType, MetricName},
|
||||
@@ -2206,7 +2221,7 @@ mod tests {
|
||||
peer_conn::tests::set_secure_mode_cfg,
|
||||
peer_manager::RouteAlgoType,
|
||||
peer_rpc::tests::register_service,
|
||||
route_trait::NextHopPolicy,
|
||||
route_trait::{NextHopPolicy, RouteCostCalculatorInterface},
|
||||
tests::{
|
||||
connect_peer_manager, create_mock_peer_manager_with_name, wait_route_appear,
|
||||
wait_route_appear_with_cost,
|
||||
@@ -2250,6 +2265,16 @@ mod tests {
|
||||
))
|
||||
}
|
||||
|
||||
struct TestCostCalculator {
|
||||
costs: HashMap<(PeerId, PeerId), i32>,
|
||||
}
|
||||
|
||||
impl RouteCostCalculatorInterface for TestCostCalculator {
|
||||
fn calculate_cost(&self, src: PeerId, dst: PeerId) -> i32 {
|
||||
*self.costs.get(&(src, dst)).unwrap_or(&1)
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recent_traffic_fanout_policy_only_marks_single_peer() {
|
||||
assert!(PeerManager::should_mark_recent_traffic_for_fanout(0));
|
||||
@@ -2657,6 +2682,109 @@ mod tests {
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn send_msg_internal_uses_latency_first_gateway_for_direct_peer() {
|
||||
let peer_mgr_a = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
|
||||
let peer_mgr_b = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
|
||||
let peer_mgr_c = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
|
||||
|
||||
connect_peer_manager(peer_mgr_a.clone(), peer_mgr_b.clone()).await;
|
||||
connect_peer_manager(peer_mgr_b.clone(), peer_mgr_c.clone()).await;
|
||||
connect_peer_manager(peer_mgr_a.clone(), peer_mgr_c.clone()).await;
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_b.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
wait_route_appear(peer_mgr_b.clone(), peer_mgr_c.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
wait_route_appear(peer_mgr_a.clone(), peer_mgr_c.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
peer_mgr_a
|
||||
.get_route()
|
||||
.set_route_cost_fn(Box::new(TestCostCalculator {
|
||||
costs: HashMap::from([
|
||||
((peer_mgr_a.my_peer_id(), peer_mgr_c.my_peer_id()), 100),
|
||||
((peer_mgr_a.my_peer_id(), peer_mgr_b.my_peer_id()), 1),
|
||||
((peer_mgr_b.my_peer_id(), peer_mgr_c.my_peer_id()), 1),
|
||||
]),
|
||||
}))
|
||||
.await;
|
||||
|
||||
wait_for_condition(
|
||||
|| {
|
||||
let peer_mgr_a = peer_mgr_a.clone();
|
||||
let peer_mgr_b = peer_mgr_b.clone();
|
||||
let peer_mgr_c = peer_mgr_c.clone();
|
||||
async move {
|
||||
peer_mgr_a
|
||||
.get_route()
|
||||
.get_next_hop_with_policy(peer_mgr_c.my_peer_id(), NextHopPolicy::LeastCost)
|
||||
.await
|
||||
== Some(peer_mgr_b.my_peer_id())
|
||||
}
|
||||
},
|
||||
Duration::from_secs(5),
|
||||
)
|
||||
.await;
|
||||
|
||||
let b_network_labels = network_labels(&peer_mgr_b);
|
||||
let forwarded_bytes_before = metric_value(
|
||||
&peer_mgr_b,
|
||||
MetricName::TrafficBytesForwarded,
|
||||
&b_network_labels,
|
||||
);
|
||||
let forwarded_packets_before = metric_value(
|
||||
&peer_mgr_b,
|
||||
MetricName::TrafficPacketsForwarded,
|
||||
&b_network_labels,
|
||||
);
|
||||
|
||||
let mut pkt = ZCPacket::new_with_payload(b"latency-first");
|
||||
pkt.fill_peer_manager_hdr(
|
||||
peer_mgr_a.my_peer_id(),
|
||||
peer_mgr_c.my_peer_id(),
|
||||
PacketType::Data as u8,
|
||||
);
|
||||
pkt.mut_peer_manager_header()
|
||||
.unwrap()
|
||||
.set_latency_first(true);
|
||||
let pkt_len = pkt.buf_len() as u64;
|
||||
|
||||
PeerManager::send_msg_internal(
|
||||
&peer_mgr_a.peers,
|
||||
&peer_mgr_a.foreign_network_client,
|
||||
&peer_mgr_a.relay_peer_map,
|
||||
Some(&peer_mgr_a.traffic_metrics),
|
||||
pkt,
|
||||
peer_mgr_c.my_peer_id(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
wait_for_condition(
|
||||
|| {
|
||||
let peer_mgr_b = peer_mgr_b.clone();
|
||||
let b_network_labels = b_network_labels.clone();
|
||||
async move {
|
||||
metric_value(
|
||||
&peer_mgr_b,
|
||||
MetricName::TrafficBytesForwarded,
|
||||
&b_network_labels,
|
||||
) >= forwarded_bytes_before + pkt_len
|
||||
&& metric_value(
|
||||
&peer_mgr_b,
|
||||
MetricName::TrafficPacketsForwarded,
|
||||
&b_network_labels,
|
||||
) > forwarded_packets_before
|
||||
}
|
||||
},
|
||||
Duration::from_secs(5),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn send_msg_internal_records_control_metrics_for_direct_peer() {
|
||||
let peer_mgr_a = create_mock_peer_manager_with_mock_stun(NatType::Unknown).await;
|
||||
|
||||
@@ -12,7 +12,7 @@ use std::{
|
||||
sync::Arc,
|
||||
task::{Context as TaskContext, Poll},
|
||||
};
|
||||
use tokio::{io::AsyncReadExt, net::TcpStream, sync::Mutex};
|
||||
use tokio::{io::AsyncReadExt, net::TcpStream};
|
||||
|
||||
use crate::tunnel::{
|
||||
FromUrl, IpVersion, SinkError, SinkItem, StreamItem, Tunnel, TunnelConnector, TunnelError,
|
||||
@@ -85,7 +85,7 @@ pub struct FakeTcpTunnelListener {
|
||||
addr: url::Url,
|
||||
os_listener: Option<tokio::net::TcpListener>,
|
||||
// interface_name -> fake tcp stack
|
||||
stack_map: DashMap<String, Arc<Mutex<stack::Stack>>>,
|
||||
stack_map: DashMap<String, Arc<stack::Stack>>,
|
||||
// a cache from ip addr to interface name
|
||||
ip_to_ifname: IpToIfNameCache,
|
||||
}
|
||||
@@ -148,7 +148,7 @@ impl FakeTcpTunnelListener {
|
||||
async fn get_stack(
|
||||
&self,
|
||||
accept_result: &AcceptResult,
|
||||
) -> Result<Arc<Mutex<stack::Stack>>, TunnelError> {
|
||||
) -> Result<Arc<stack::Stack>, TunnelError> {
|
||||
let local_socket_addr = accept_result.local_addr;
|
||||
|
||||
let interface_name = &accept_result.interface_name;
|
||||
@@ -158,29 +158,38 @@ impl FakeTcpTunnelListener {
|
||||
IpAddr::V6(ip) => (None, Some(ip)),
|
||||
};
|
||||
|
||||
let ret = match self.stack_map.entry(interface_name.to_string()) {
|
||||
dashmap::Entry::Occupied(entry) => entry.get().clone(),
|
||||
dashmap::Entry::Vacant(entry) => {
|
||||
let tun =
|
||||
create_tun_off_runtime(interface_name.to_string(), None, local_socket_addr)
|
||||
.await?;
|
||||
tracing::info!(
|
||||
?local_socket_addr,
|
||||
"create new stack with interface_name: {:?}",
|
||||
interface_name
|
||||
);
|
||||
let stack = Arc::new(Mutex::new(stack::Stack::new(
|
||||
tun,
|
||||
local_ip.unwrap_or(Ipv4Addr::UNSPECIFIED),
|
||||
local_ip6,
|
||||
accept_result.mac,
|
||||
)));
|
||||
entry.insert(stack.clone());
|
||||
stack
|
||||
}
|
||||
};
|
||||
if let Some(entry) = self.stack_map.get(interface_name) {
|
||||
let stack = entry.clone();
|
||||
drop(entry);
|
||||
|
||||
Ok(ret)
|
||||
if !stack.is_closed() {
|
||||
return Ok(stack);
|
||||
}
|
||||
|
||||
tracing::warn!(
|
||||
interface_name,
|
||||
"fake_tcp stack reader_task finished, recreating stack"
|
||||
);
|
||||
self.stack_map.remove(interface_name);
|
||||
}
|
||||
|
||||
let tun =
|
||||
create_tun_off_runtime(interface_name.to_string(), None, local_socket_addr).await?;
|
||||
tracing::info!(
|
||||
?local_socket_addr,
|
||||
"create new stack with interface_name: {:?}",
|
||||
interface_name
|
||||
);
|
||||
let stack = Arc::new(stack::Stack::new(
|
||||
tun,
|
||||
local_ip.unwrap_or(Ipv4Addr::UNSPECIFIED),
|
||||
local_ip6,
|
||||
accept_result.mac,
|
||||
));
|
||||
self.stack_map
|
||||
.insert(interface_name.to_string(), stack.clone());
|
||||
|
||||
Ok(stack)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -215,19 +224,29 @@ impl TunnelListener for FakeTcpTunnelListener {
|
||||
let os_listener = tokio::net::TcpListener::bind(bind_addr).await?;
|
||||
tracing::info!(port, "FakeTcpTunnelListener listening");
|
||||
self.os_listener = Some(os_listener);
|
||||
// self.stack.lock().await.listen(port);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn accept(&mut self) -> Result<Box<dyn Tunnel>, TunnelError> {
|
||||
tracing::debug!("FakeTcpTunnelListener waiting for accept");
|
||||
let res = self.do_accept().await?;
|
||||
let stack = self.get_stack(&res).await?;
|
||||
let socket = stack
|
||||
.lock()
|
||||
.await
|
||||
.alloc_established_socket(res.local_addr, res.remote_addr, stack::State::Established)
|
||||
.await;
|
||||
let (res, stack, socket) = loop {
|
||||
let res = self.do_accept().await?;
|
||||
let stack = self.get_stack(&res).await?;
|
||||
let socket = stack.try_alloc_established_socket(
|
||||
res.local_addr,
|
||||
res.remote_addr,
|
||||
stack::State::Established,
|
||||
);
|
||||
let Some(socket) = socket else {
|
||||
tracing::warn!(
|
||||
interface_name = res.interface_name,
|
||||
"fake_tcp stack closed while accepting connection, dropping accepted socket"
|
||||
);
|
||||
self.stack_map.remove(&res.interface_name);
|
||||
continue;
|
||||
};
|
||||
break (res, stack, socket);
|
||||
};
|
||||
|
||||
tracing::info!(
|
||||
?res,
|
||||
@@ -236,7 +255,7 @@ impl TunnelListener for FakeTcpTunnelListener {
|
||||
);
|
||||
|
||||
let info = TunnelInfo {
|
||||
tunnel_type: get_faketcp_tunnel_type_str(stack.lock().await.driver_type()),
|
||||
tunnel_type: get_faketcp_tunnel_type_str(stack.driver_type()),
|
||||
local_addr: Some(self.local_url().into()),
|
||||
remote_addr: Some(
|
||||
crate::tunnel::build_url_from_socket_addr(
|
||||
@@ -354,12 +373,14 @@ impl TunnelConnector for FakeTcpTunnelConnector {
|
||||
let tun =
|
||||
create_tun_off_runtime(interface_name.clone(), Some(remote_addr), local_addr).await?;
|
||||
let local_ip = local_ip.unwrap_or("0.0.0.0".parse().unwrap());
|
||||
let mut stack = stack::Stack::new(tun, local_ip, local_ip6, mac);
|
||||
let stack = stack::Stack::new(tun, local_ip, local_ip6, mac);
|
||||
let driver_type = stack.driver_type();
|
||||
|
||||
let socket = stack
|
||||
.alloc_established_socket(local_addr, remote_addr, stack::State::SynSent)
|
||||
.await;
|
||||
.try_alloc_established_socket(local_addr, remote_addr, stack::State::SynSent)
|
||||
.ok_or(TunnelError::InternalError(
|
||||
"FakeTCP stack closed while allocating socket".into(),
|
||||
))?;
|
||||
|
||||
let os_stream = os_socket.connect(remote_addr).await?;
|
||||
|
||||
|
||||
@@ -54,7 +54,7 @@ use std::sync::{
|
||||
use tokio::sync::broadcast;
|
||||
use tokio::time;
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
use tracing::{info, trace, warn};
|
||||
use tracing::{error, info, trace, warn};
|
||||
|
||||
const TIMEOUT: time::Duration = time::Duration::from_secs(1);
|
||||
const RETRIES: usize = 6;
|
||||
@@ -83,13 +83,33 @@ impl AddrTuple {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct StackState {
|
||||
tuples: HashMap<AddrTuple, flume::Sender<Bytes>>,
|
||||
closed: bool,
|
||||
}
|
||||
|
||||
struct Shared {
|
||||
tuples: RwLock<HashMap<AddrTuple, flume::Sender<Bytes>>>,
|
||||
state: RwLock<StackState>,
|
||||
listening: RwLock<HashSet<u16>>,
|
||||
tun: Arc<dyn Tun>,
|
||||
tuples_purge: broadcast::Sender<AddrTuple>,
|
||||
}
|
||||
|
||||
impl Shared {
|
||||
fn is_closed(&self) -> bool {
|
||||
self.state.read().unwrap().closed
|
||||
}
|
||||
|
||||
fn mark_closed_and_clear_tuples(&self) -> usize {
|
||||
let mut state = self.state.write().unwrap();
|
||||
state.closed = true;
|
||||
let len = state.tuples.len();
|
||||
state.tuples.clear();
|
||||
len
|
||||
}
|
||||
}
|
||||
|
||||
pub struct Stack {
|
||||
shared: Arc<Shared>,
|
||||
local_ip: Ipv4Addr,
|
||||
@@ -353,7 +373,17 @@ impl Drop for Socket {
|
||||
fn drop(&mut self) {
|
||||
let tuple = AddrTuple::new(self.local_addr, self.remote_addr);
|
||||
// dissociates ourself from the dispatch map
|
||||
assert!(self.shared.tuples.write().unwrap().remove(&tuple).is_some());
|
||||
let (removed, closed) = {
|
||||
let mut state = self.shared.state.write().unwrap();
|
||||
(state.tuples.remove(&tuple).is_some(), state.closed)
|
||||
};
|
||||
if !removed {
|
||||
if closed {
|
||||
trace!(?tuple, "Fake TCP tuple already removed after stack closed");
|
||||
} else {
|
||||
warn!(?tuple, "Fake TCP tuple missing while dropping socket");
|
||||
}
|
||||
}
|
||||
// purge cache
|
||||
let _ = self.shared.tuples_purge.send(tuple);
|
||||
|
||||
@@ -400,7 +430,7 @@ impl Stack {
|
||||
) -> Stack {
|
||||
let (tuples_purge_tx, _tuples_purge_rx) = broadcast::channel(16);
|
||||
let shared = Arc::new(Shared {
|
||||
tuples: RwLock::new(HashMap::new()),
|
||||
state: RwLock::new(StackState::default()),
|
||||
tun: tun.clone(),
|
||||
listening: RwLock::new(HashSet::new()),
|
||||
tuples_purge: tuples_purge_tx.clone(),
|
||||
@@ -426,19 +456,31 @@ impl Stack {
|
||||
self.shared.tun.driver_type()
|
||||
}
|
||||
|
||||
pub fn is_closed(&self) -> bool {
|
||||
self.shared.is_closed() || self.reader_task.is_finished()
|
||||
}
|
||||
|
||||
/// Listens for incoming connections on the given `port`.
|
||||
pub fn listen(&mut self, port: u16) {
|
||||
assert!(self.shared.listening.write().unwrap().insert(port));
|
||||
}
|
||||
|
||||
pub async fn alloc_established_socket(
|
||||
&mut self,
|
||||
pub fn try_alloc_established_socket(
|
||||
&self,
|
||||
local_addr: SocketAddr,
|
||||
remote_addr: SocketAddr,
|
||||
state: State,
|
||||
) -> Socket {
|
||||
) -> Option<Socket> {
|
||||
let tuple = AddrTuple::new(local_addr, remote_addr);
|
||||
let mut tuples = self.shared.tuples.write().unwrap();
|
||||
let mut stack_state = self.shared.state.write().unwrap();
|
||||
if stack_state.closed || self.reader_task.is_finished() {
|
||||
stack_state.closed = true;
|
||||
warn!(
|
||||
?tuple,
|
||||
"fake_tcp stack is closed, refusing to allocate socket"
|
||||
);
|
||||
return None;
|
||||
}
|
||||
let (sock, incoming) = Socket::new(
|
||||
self.shared.clone(),
|
||||
// self.shared.tun.choose(&mut rng).unwrap().clone(),
|
||||
@@ -450,8 +492,8 @@ impl Stack {
|
||||
Some(0), // Initial ACK
|
||||
state,
|
||||
);
|
||||
assert!(tuples.insert(tuple, incoming).is_none());
|
||||
sock
|
||||
assert!(stack_state.tuples.insert(tuple, incoming).is_none());
|
||||
Some(sock)
|
||||
}
|
||||
|
||||
async fn reader_task(
|
||||
@@ -466,7 +508,22 @@ impl Stack {
|
||||
|
||||
tokio::select! {
|
||||
size = tun.recv(&mut buf) => {
|
||||
let size = size.unwrap();
|
||||
let size = match size {
|
||||
Ok(size) => size,
|
||||
Err(e) => {
|
||||
let shared_tuple_count = shared.mark_closed_and_clear_tuples();
|
||||
let cached_tuple_count = tuples.len();
|
||||
tuples.clear();
|
||||
error!(
|
||||
?e,
|
||||
driver_type = tun.driver_type(),
|
||||
shared_tuple_count,
|
||||
cached_tuple_count,
|
||||
"fake_tcp tun recv failed, reader_task exiting"
|
||||
);
|
||||
break;
|
||||
}
|
||||
};
|
||||
tracing::trace!(len = size, ?buf, "PnetTun received packet");
|
||||
let buf = buf.split().freeze();
|
||||
|
||||
@@ -494,8 +551,8 @@ impl Stack {
|
||||
} else {
|
||||
trace!("Cache miss, checking the shared tuples table for connection");
|
||||
let sender = {
|
||||
let tuples = shared.tuples.read().unwrap();
|
||||
tuples.get(&tuple).cloned()
|
||||
let state = shared.state.read().unwrap();
|
||||
state.tuples.get(&tuple).cloned()
|
||||
};
|
||||
|
||||
if let Some(c) = sender {
|
||||
@@ -532,11 +589,107 @@ impl Stack {
|
||||
}
|
||||
},
|
||||
tuple = tuples_purge.recv() => {
|
||||
let tuple = tuple.unwrap();
|
||||
tuples.remove(&tuple);
|
||||
trace!("Removed cached tuple: {:?}", tuple);
|
||||
match tuple {
|
||||
Ok(tuple) => {
|
||||
tuples.remove(&tuple);
|
||||
trace!("Removed cached tuple: {:?}", tuple);
|
||||
}
|
||||
Err(broadcast::error::RecvError::Lagged(skipped)) => {
|
||||
let cached_tuple_count = tuples.len();
|
||||
tuples.clear();
|
||||
warn!(
|
||||
skipped,
|
||||
cached_tuple_count,
|
||||
"fake_tcp tuples purge receiver lagged, cleared local cache"
|
||||
);
|
||||
}
|
||||
Err(broadcast::error::RecvError::Closed) => {
|
||||
let shared_tuple_count = shared.mark_closed_and_clear_tuples();
|
||||
let cached_tuple_count = tuples.len();
|
||||
tuples.clear();
|
||||
warn!(
|
||||
shared_tuple_count,
|
||||
cached_tuple_count,
|
||||
"fake_tcp tuples purge channel closed, reader_task exiting"
|
||||
);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::io;
|
||||
use tokio::{
|
||||
sync::Notify,
|
||||
time::{Duration, timeout},
|
||||
};
|
||||
|
||||
#[derive(Default)]
|
||||
struct FailingTun {
|
||||
fail: Notify,
|
||||
}
|
||||
|
||||
impl FailingTun {
|
||||
fn fail(&self) {
|
||||
self.fail.notify_one();
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Tun for FailingTun {
|
||||
async fn recv(&self, _packet: &mut BytesMut) -> Result<usize, io::Error> {
|
||||
self.fail.notified().await;
|
||||
Err(io::Error::new(io::ErrorKind::BrokenPipe, "test tun closed"))
|
||||
}
|
||||
|
||||
fn try_send(&self, _packet: &Bytes) -> Result<(), io::Error> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn driver_type(&self) -> &'static str {
|
||||
"test"
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reader_task_closes_sockets_on_tun_recv_error() {
|
||||
let tun = Arc::new(FailingTun::default());
|
||||
let mut stack = Stack::new(tun.clone(), Ipv4Addr::LOCALHOST, None, None);
|
||||
let socket = stack
|
||||
.try_alloc_established_socket(
|
||||
SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 10_000),
|
||||
SocketAddr::new(Ipv4Addr::new(192, 0, 2, 1).into(), 20_000),
|
||||
State::Established,
|
||||
)
|
||||
.expect("socket allocation should succeed before tun failure");
|
||||
|
||||
tun.fail();
|
||||
|
||||
let join_result = timeout(Duration::from_secs(1), &mut stack.reader_task)
|
||||
.await
|
||||
.expect("reader task should exit after tun recv error");
|
||||
assert!(join_result.is_ok());
|
||||
assert!(stack.is_closed());
|
||||
|
||||
let mut buf = BytesMut::new();
|
||||
let recv_result = timeout(Duration::from_secs(1), socket.recv(&mut buf))
|
||||
.await
|
||||
.expect("socket recv should not hang after reader task exits");
|
||||
assert_eq!(recv_result, None);
|
||||
|
||||
let new_socket = stack.try_alloc_established_socket(
|
||||
SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 10_001),
|
||||
SocketAddr::new(Ipv4Addr::new(192, 0, 2, 1).into(), 20_001),
|
||||
State::Established,
|
||||
);
|
||||
assert!(new_socket.is_none());
|
||||
|
||||
drop(socket);
|
||||
}
|
||||
}
|
||||
|
||||
+246
-2
@@ -23,6 +23,250 @@ use std::{net::SocketAddr, sync::Arc, time::Duration};
|
||||
use tokio::net::UdpSocket;
|
||||
|
||||
// region config
|
||||
mod crypto {
|
||||
use crate::utils::BoxExt;
|
||||
use bytes::{Buf, BytesMut};
|
||||
use quinn_proto::crypto::{
|
||||
ClientConfig, ExportKeyingMaterialError, KeyPair, Keys, ServerConfig, Session,
|
||||
UnsupportedVersion,
|
||||
};
|
||||
use quinn_proto::transport_parameters::TransportParameters;
|
||||
use quinn_proto::{
|
||||
ConnectError, ConnectionId, Side, TransportError,
|
||||
crypto::{CryptoError, HeaderKey, PacketKey},
|
||||
};
|
||||
use seahash::SeaHasher;
|
||||
use std::any::Any;
|
||||
use std::{hash::Hasher, sync::Arc};
|
||||
use tracing::{error, instrument, trace};
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
struct CryptoKey;
|
||||
|
||||
impl CryptoKey {
|
||||
fn header(self) -> KeyPair<Box<dyn HeaderKey>> {
|
||||
KeyPair {
|
||||
local: Box::new(self),
|
||||
remote: Box::new(self),
|
||||
}
|
||||
}
|
||||
|
||||
fn packet(self) -> KeyPair<Box<dyn PacketKey>> {
|
||||
KeyPair {
|
||||
local: Box::new(self),
|
||||
remote: Box::new(self),
|
||||
}
|
||||
}
|
||||
|
||||
fn keys(self) -> Keys {
|
||||
Keys {
|
||||
header: self.header(),
|
||||
packet: self.packet(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl HeaderKey for CryptoKey {
|
||||
fn decrypt(&self, _: usize, _: &mut [u8]) {}
|
||||
fn encrypt(&self, _: usize, _: &mut [u8]) {}
|
||||
fn sample_size(&self) -> usize {
|
||||
0
|
||||
}
|
||||
}
|
||||
|
||||
impl CryptoKey {
|
||||
fn checksum(slices: &[&[u8]]) -> u64 {
|
||||
let mut hasher = SeaHasher::default();
|
||||
for slice in slices {
|
||||
hasher.write(&(slice.len() as u64).to_le_bytes());
|
||||
hasher.write(slice);
|
||||
}
|
||||
hasher.finish()
|
||||
}
|
||||
}
|
||||
|
||||
impl PacketKey for CryptoKey {
|
||||
#[instrument(level = "trace")]
|
||||
fn encrypt(&self, packet: u64, buf: &mut [u8], header_len: usize) {
|
||||
let (header, rest) = buf.split_at_mut(header_len);
|
||||
let (payload, tag) = rest.split_at_mut(rest.len() - self.tag_len());
|
||||
let checksum = Self::checksum(&[header, payload]);
|
||||
tag.copy_from_slice(&checksum.to_be_bytes());
|
||||
trace!(checksum, ?header, ?payload, ?tag);
|
||||
}
|
||||
|
||||
#[instrument(level = "trace")]
|
||||
fn decrypt(
|
||||
&self,
|
||||
packet: u64,
|
||||
header: &[u8],
|
||||
payload: &mut BytesMut,
|
||||
) -> Result<(), CryptoError> {
|
||||
let tag = payload.split_off(payload.len() - self.tag_len()).get_u64();
|
||||
trace!(tag, ?payload);
|
||||
let checksum = Self::checksum(&[header, payload]);
|
||||
if checksum != tag {
|
||||
error!(tag, checksum, "checksum mismatch");
|
||||
return Err(CryptoError);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn tag_len(&self) -> usize {
|
||||
8
|
||||
}
|
||||
|
||||
fn confidentiality_limit(&self) -> u64 {
|
||||
u64::MAX
|
||||
}
|
||||
|
||||
fn integrity_limit(&self) -> u64 {
|
||||
1 << 36
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum HandshakeState {
|
||||
EmitInitial,
|
||||
EmitHandshake,
|
||||
Done,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct QuicSession {
|
||||
side: Side,
|
||||
state: HandshakeState,
|
||||
local: TransportParameters,
|
||||
remote: Option<TransportParameters>,
|
||||
}
|
||||
|
||||
impl QuicSession {
|
||||
fn new(side: Side, params: TransportParameters) -> Self {
|
||||
Self {
|
||||
side,
|
||||
state: HandshakeState::EmitInitial,
|
||||
local: params,
|
||||
remote: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Session for QuicSession {
|
||||
fn initial_keys(&self, _: &ConnectionId, _: Side) -> Keys {
|
||||
CryptoKey.keys()
|
||||
}
|
||||
|
||||
fn handshake_data(&self) -> Option<Box<dyn Any>> {
|
||||
self.remote.map(|params| params.boxed() as _)
|
||||
}
|
||||
|
||||
fn peer_identity(&self) -> Option<Box<dyn Any>> {
|
||||
None
|
||||
}
|
||||
|
||||
fn early_crypto(&self) -> Option<(Box<dyn HeaderKey>, Box<dyn PacketKey>)> {
|
||||
None
|
||||
}
|
||||
|
||||
fn early_data_accepted(&self) -> Option<bool> {
|
||||
Some(false)
|
||||
}
|
||||
|
||||
#[instrument(level = "trace")]
|
||||
fn is_handshaking(&self) -> bool {
|
||||
self.remote.is_none() || self.state != HandshakeState::Done
|
||||
}
|
||||
|
||||
#[instrument(level = "trace")]
|
||||
fn read_handshake(&mut self, mut buf: &[u8]) -> Result<bool, TransportError> {
|
||||
if self.remote.is_none() {
|
||||
self.remote = Some(
|
||||
TransportParameters::read(self.side, &mut buf)
|
||||
.expect("failed to read transport parameters"),
|
||||
);
|
||||
}
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
#[instrument(level = "trace")]
|
||||
fn transport_parameters(&self) -> Result<Option<TransportParameters>, TransportError> {
|
||||
Ok(self.remote)
|
||||
}
|
||||
|
||||
#[instrument(level = "trace")]
|
||||
fn write_handshake(&mut self, buf: &mut Vec<u8>) -> Option<Keys> {
|
||||
match self.state {
|
||||
HandshakeState::EmitInitial => {
|
||||
if self.side.is_client() {
|
||||
self.local.write(buf);
|
||||
}
|
||||
self.state = HandshakeState::EmitHandshake;
|
||||
Some(CryptoKey.keys())
|
||||
}
|
||||
HandshakeState::EmitHandshake => {
|
||||
if self.side.is_server() {
|
||||
self.local.write(buf);
|
||||
}
|
||||
self.state = HandshakeState::Done;
|
||||
Some(CryptoKey.keys())
|
||||
}
|
||||
HandshakeState::Done => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn next_1rtt_keys(&mut self) -> Option<KeyPair<Box<dyn PacketKey>>> {
|
||||
Some(CryptoKey.packet())
|
||||
}
|
||||
|
||||
fn is_valid_retry(&self, _: &ConnectionId, _: &[u8], _: &[u8]) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn export_keying_material(
|
||||
&self,
|
||||
_: &mut [u8],
|
||||
_: &[u8],
|
||||
_: &[u8],
|
||||
) -> Result<(), ExportKeyingMaterialError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct CryptoConfig;
|
||||
|
||||
impl ClientConfig for CryptoConfig {
|
||||
#[instrument(level = "trace")]
|
||||
fn start_session(
|
||||
self: Arc<Self>,
|
||||
version: u32,
|
||||
server_name: &str,
|
||||
params: &TransportParameters,
|
||||
) -> Result<Box<dyn Session>, ConnectError> {
|
||||
Ok(Box::new(QuicSession::new(Side::Client, *params)))
|
||||
}
|
||||
}
|
||||
|
||||
impl ServerConfig for CryptoConfig {
|
||||
fn initial_keys(&self, _: u32, _: &ConnectionId) -> Result<Keys, UnsupportedVersion> {
|
||||
Ok(CryptoKey.keys())
|
||||
}
|
||||
|
||||
fn retry_tag(&self, _: u32, _: &ConnectionId, _: &[u8]) -> [u8; 16] {
|
||||
[0u8; 16]
|
||||
}
|
||||
|
||||
#[instrument(level = "trace")]
|
||||
fn start_session(
|
||||
self: Arc<Self>,
|
||||
version: u32,
|
||||
params: &TransportParameters,
|
||||
) -> Box<dyn Session> {
|
||||
Box::new(QuicSession::new(Side::Server, *params))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn transport_config() -> Arc<TransportConfig> {
|
||||
let mut config = TransportConfig::default();
|
||||
|
||||
@@ -39,13 +283,13 @@ pub fn transport_config() -> Arc<TransportConfig> {
|
||||
}
|
||||
|
||||
pub fn server_config() -> ServerConfig {
|
||||
let mut config = quinn_plaintext::server_config();
|
||||
let mut config = ServerConfig::with_crypto(Arc::new(crypto::CryptoConfig));
|
||||
config.transport_config(transport_config());
|
||||
config
|
||||
}
|
||||
|
||||
pub fn client_config() -> ClientConfig {
|
||||
let mut config = quinn_plaintext::client_config();
|
||||
let mut config = ClientConfig::new(Arc::new(crypto::CryptoConfig));
|
||||
config.transport_config(transport_config());
|
||||
config
|
||||
}
|
||||
|
||||
@@ -1,7 +1,4 @@
|
||||
use std::{
|
||||
cell::UnsafeCell,
|
||||
sync::atomic::{AtomicU32, Ordering::Relaxed},
|
||||
};
|
||||
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering::Relaxed};
|
||||
|
||||
pub struct WindowLatency {
|
||||
latency_us_window: Vec<AtomicU32>,
|
||||
@@ -63,34 +60,30 @@ impl WindowLatency {
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct Throughput {
|
||||
tx_bytes: UnsafeCell<u64>,
|
||||
rx_bytes: UnsafeCell<u64>,
|
||||
tx_packets: UnsafeCell<u64>,
|
||||
rx_packets: UnsafeCell<u64>,
|
||||
tx_bytes: AtomicU64,
|
||||
rx_bytes: AtomicU64,
|
||||
tx_packets: AtomicU64,
|
||||
rx_packets: AtomicU64,
|
||||
}
|
||||
|
||||
impl Clone for Throughput {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
tx_bytes: UnsafeCell::new(unsafe { *self.tx_bytes.get() }),
|
||||
rx_bytes: UnsafeCell::new(unsafe { *self.rx_bytes.get() }),
|
||||
tx_packets: UnsafeCell::new(unsafe { *self.tx_packets.get() }),
|
||||
rx_packets: UnsafeCell::new(unsafe { *self.rx_packets.get() }),
|
||||
tx_bytes: AtomicU64::new(self.tx_bytes()),
|
||||
rx_bytes: AtomicU64::new(self.rx_bytes()),
|
||||
tx_packets: AtomicU64::new(self.tx_packets()),
|
||||
rx_packets: AtomicU64::new(self.rx_packets()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// add sync::Send and sync::Sync traits to Throughput
|
||||
unsafe impl Send for Throughput {}
|
||||
unsafe impl Sync for Throughput {}
|
||||
|
||||
impl Default for Throughput {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
tx_bytes: UnsafeCell::new(0),
|
||||
rx_bytes: UnsafeCell::new(0),
|
||||
tx_packets: UnsafeCell::new(0),
|
||||
rx_packets: UnsafeCell::new(0),
|
||||
tx_bytes: AtomicU64::new(0),
|
||||
rx_bytes: AtomicU64::new(0),
|
||||
tx_packets: AtomicU64::new(0),
|
||||
rx_packets: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -101,32 +94,68 @@ impl Throughput {
|
||||
}
|
||||
|
||||
pub fn tx_bytes(&self) -> u64 {
|
||||
unsafe { *self.tx_bytes.get() }
|
||||
self.tx_bytes.load(Relaxed)
|
||||
}
|
||||
|
||||
pub fn rx_bytes(&self) -> u64 {
|
||||
unsafe { *self.rx_bytes.get() }
|
||||
self.rx_bytes.load(Relaxed)
|
||||
}
|
||||
|
||||
pub fn tx_packets(&self) -> u64 {
|
||||
unsafe { *self.tx_packets.get() }
|
||||
self.tx_packets.load(Relaxed)
|
||||
}
|
||||
|
||||
pub fn rx_packets(&self) -> u64 {
|
||||
unsafe { *self.rx_packets.get() }
|
||||
self.rx_packets.load(Relaxed)
|
||||
}
|
||||
|
||||
pub fn record_tx_bytes(&self, bytes: u64) {
|
||||
unsafe {
|
||||
*self.tx_bytes.get() += bytes;
|
||||
*self.tx_packets.get() += 1;
|
||||
}
|
||||
self.tx_bytes.fetch_add(bytes, Relaxed);
|
||||
self.tx_packets.fetch_add(1, Relaxed);
|
||||
}
|
||||
|
||||
pub fn record_rx_bytes(&self, bytes: u64) {
|
||||
unsafe {
|
||||
*self.rx_bytes.get() += bytes;
|
||||
*self.rx_packets.get() += 1;
|
||||
}
|
||||
self.rx_bytes.fetch_add(bytes, Relaxed);
|
||||
self.rx_packets.fetch_add(1, Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::Throughput;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[test]
|
||||
fn throughput_records_concurrent_tx_and_rx() {
|
||||
const THREADS: usize = 8;
|
||||
const RECORDS_PER_THREAD: usize = 10_000;
|
||||
const TX_BYTES_PER_RECORD: u64 = 3;
|
||||
const RX_BYTES_PER_RECORD: u64 = 7;
|
||||
|
||||
let throughput = Arc::new(Throughput::new());
|
||||
|
||||
std::thread::scope(|scope| {
|
||||
for _ in 0..THREADS {
|
||||
let throughput = Arc::clone(&throughput);
|
||||
scope.spawn(move || {
|
||||
for _ in 0..RECORDS_PER_THREAD {
|
||||
throughput.record_tx_bytes(TX_BYTES_PER_RECORD);
|
||||
throughput.record_rx_bytes(RX_BYTES_PER_RECORD);
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
let expected_packets = (THREADS * RECORDS_PER_THREAD) as u64;
|
||||
assert_eq!(throughput.tx_packets(), expected_packets);
|
||||
assert_eq!(throughput.rx_packets(), expected_packets);
|
||||
assert_eq!(
|
||||
throughput.tx_bytes(),
|
||||
expected_packets * TX_BYTES_PER_RECORD
|
||||
);
|
||||
assert_eq!(
|
||||
throughput.rx_bytes(),
|
||||
expected_packets * RX_BYTES_PER_RECORD
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -118,6 +118,7 @@ impl WsTunnelListener {
|
||||
|
||||
let (request, stream) = ServerBuilder::new()
|
||||
.limits(Limits::unlimited())
|
||||
.max_headers(128)
|
||||
.accept(stream)
|
||||
.await?;
|
||||
|
||||
@@ -252,7 +253,8 @@ impl WsTunnelConnector {
|
||||
),
|
||||
};
|
||||
|
||||
let c = ClientBuilder::from_uri(http::Uri::try_from(addr.to_string()).unwrap());
|
||||
let c = ClientBuilder::from_uri(http::Uri::try_from(addr.to_string()).unwrap())
|
||||
.max_headers(128);
|
||||
let stream: MaybeTlsStream<TcpStream> = if is_wss {
|
||||
init_crypto_provider();
|
||||
let tls_conn =
|
||||
|
||||
Reference in New Issue
Block a user