mirror of
https://github.com/EasyTier/EasyTier.git
synced 2026-09-20 19:22:58 +00:00
Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Paweł Urbanek
|
cb04cefb2a | ||
|
fanyang89
|
20873fc62e | ||
|
fanyang89
|
27f6b90fa1 | ||
|
fanyang89
|
a602125d97 | ||
|
|
7756a15cbe | ||
|
fanyang
|
741460e1e4 | ||
|
fanyang
|
13412895c5 |
Generated
+128
-8
@@ -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"
|
||||
@@ -1145,6 +1151,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"
|
||||
@@ -1260,6 +1272,33 @@ version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901"
|
||||
|
||||
[[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"
|
||||
@@ -1605,6 +1644,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"
|
||||
@@ -2299,6 +2374,7 @@ dependencies = [
|
||||
"clap_complete",
|
||||
"clap_complete_nushell",
|
||||
"console-subscriber",
|
||||
"criterion",
|
||||
"crossbeam",
|
||||
"ctor 0.8.0",
|
||||
"dashmap",
|
||||
@@ -3931,9 +4007,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "hotpath"
|
||||
version = "0.18.0"
|
||||
version = "0.21.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "dc2c28b1fa962e433f800ed1ea0bf53dc028d3745cf2acec6cfd28b65ac96afa"
|
||||
checksum = "1ff6b552a6afa29d9e33f8d555bee9093c142dd449501ae128e6494a303f03dc"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"cfg-if",
|
||||
@@ -3960,9 +4036,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "hotpath-macros"
|
||||
version = "0.18.0"
|
||||
version = "0.21.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a585238d8daf746e27df0f24d1bbdcd2410e9febff63f9a0173f90d7e71c50f6"
|
||||
checksum = "4f15322569d3cfadf84c0de7ef72be435b8f4b4839ee4ace78a7eaca48a87ded"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -3971,15 +4047,15 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "hotpath-macros-meta"
|
||||
version = "0.18.0"
|
||||
version = "0.21.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "309f63c2f755dead454dd4b3ea8ab5c947f14f8ea435fbcd37fa820e17290e80"
|
||||
checksum = "b3675e29d16c844ccad12763672b33e51d9a000c346720c4f354f7a3bdc649a8"
|
||||
|
||||
[[package]]
|
||||
name = "hotpath-meta"
|
||||
version = "0.18.0"
|
||||
version = "0.21.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "68faa91a9e1114dff668cd90560f332da6bbde40dae37ec28ea1c43ca5ce3be3"
|
||||
checksum = "d558d972ddc9483fb4e713af3dd41634edf895c7cb18ed13dc58c56431d42e27"
|
||||
dependencies = [
|
||||
"hotpath-macros-meta",
|
||||
]
|
||||
@@ -5971,6 +6047,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"
|
||||
@@ -6601,6 +6683,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"
|
||||
@@ -10021,6 +10131,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"
|
||||
|
||||
+10
-4
@@ -28,6 +28,10 @@ path = "src/easytier-cli.rs"
|
||||
name = "easytier"
|
||||
path = "src/lib.rs"
|
||||
|
||||
[[bench]]
|
||||
name = "tx_throughput"
|
||||
harness = false
|
||||
|
||||
[dependencies]
|
||||
git-version = "0.3.9"
|
||||
|
||||
@@ -52,9 +56,10 @@ toml = "0.8.12"
|
||||
chrono = { version = "0.4.37", features = ["serde"] }
|
||||
|
||||
guarden = "0.2"
|
||||
hotpath = { version = "0.18", default-features = false, optional = true }
|
||||
quanta = "0.12"
|
||||
|
||||
hotpath = { version = "0.21", default-features = false, optional = true }
|
||||
|
||||
delegate = "0.13.5"
|
||||
|
||||
itertools = "0.14.0"
|
||||
@@ -337,6 +342,7 @@ zip = "4.0.0"
|
||||
|
||||
|
||||
[dev-dependencies]
|
||||
criterion = "0.5.1"
|
||||
serial_test = "3.0.0"
|
||||
rstest = "0.25.0"
|
||||
futures-util = "0.3.31"
|
||||
@@ -403,6 +409,9 @@ jemalloc-prof = [
|
||||
"jemalloc-sys/stats",
|
||||
]
|
||||
tracing = ["tokio/tracing", "dep:console-subscriber"]
|
||||
magic-dns = ["dep:hickory-client", "dep:hickory-server"]
|
||||
faketcp = ["dep:flume"]
|
||||
zstd = ["dep:zstd"]
|
||||
hotpath = [
|
||||
"dep:hotpath",
|
||||
"hotpath/hotpath",
|
||||
@@ -412,8 +421,5 @@ hotpath = [
|
||||
]
|
||||
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
|
||||
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
|
||||
magic-dns = ["dep:hickory-client", "dep:hickory-server"]
|
||||
faketcp = ["dep:flume"]
|
||||
zstd = ["dep:zstd"]
|
||||
# For Network Extension on macOS
|
||||
macos-ne = []
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
# TX Throughput Benchmark
|
||||
|
||||
Criterion benchmark for EasyTier's TX injection path (`peer_manager::send_msg_by_ip`).
|
||||
|
||||
## What it measures
|
||||
|
||||
The benchmark sets up two EasyTier instances (`hot-a` / `hot-b`) and drives
|
||||
packets from `hot-a` to `hot-b` via `peer_manager.send_msg_by_ip`. This is the
|
||||
same entry point `easytier-core` uses for daily forwarded traffic, so the
|
||||
numbers reflect the real TX hot path: NIC pipeline → route lookup →
|
||||
compress/encrypt → peer connection → tunnel send.
|
||||
|
||||
Two variants are reported per tunnel kind:
|
||||
|
||||
| Bench | What it measures |
|
||||
| --------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `tx_throughput/<tunnel>` | Serial baseline. One send in flight at a time. Reports per-packet CPU cost (TX injection latency). |
|
||||
| `tx_throughput/<tunnel>-saturate` | Spawns `TX_THROUGHPUT_INFLIGHT` tokio tasks that independently pump `send_msg_by_ip`. Reports the aggregate throughput ceiling the peer manager + tunnel can sustain across worker threads. |
|
||||
|
||||
> **Out of scope (by design):** TUN read/write (`no_tun = true`), compression
|
||||
> (default `None`), reverse/RX-side measurement, multi-peer fanout. Add
|
||||
> separate benchmarks if you need those.
|
||||
|
||||
## Quick start
|
||||
|
||||
### ring tunnel (no root, fastest)
|
||||
|
||||
```bash
|
||||
cargo bench --bench tx_throughput
|
||||
```
|
||||
|
||||
Smoke run (faster iteration):
|
||||
|
||||
```bash
|
||||
TX_THROUGHPUT_MEASUREMENT_SECS=2 \
|
||||
TX_THROUGHPUT_WARMUP_SECS=1 \
|
||||
TX_THROUGHPUT_SAMPLE_SIZE=10 \
|
||||
cargo bench --bench tx_throughput -- --quiet
|
||||
```
|
||||
|
||||
### tcp / udp tunnels (requires Docker + root)
|
||||
|
||||
The benchmark creates a Docker network and registers each container's netns
|
||||
under `/var/run/netns`, which requires root. Run the whole command under
|
||||
`sudo`:
|
||||
|
||||
```bash
|
||||
sudo TX_THROUGHPUT_TUNNEL=tcp \
|
||||
TX_THROUGHPUT_MEASUREMENT_SECS=5 \
|
||||
TX_THROUGHPUT_WARMUP_SECS=2 \
|
||||
TX_THROUGHPUT_INFLIGHT=64 \
|
||||
cargo bench --bench tx_throughput -- --quiet
|
||||
|
||||
sudo TX_THROUGHPUT_TUNNEL=udp cargo bench --bench tx_throughput -- --quiet
|
||||
```
|
||||
|
||||
> If `sudo` cannot find `cargo`, use `sudo -E` or the absolute path
|
||||
> (`$(which cargo)`).
|
||||
|
||||
## Environment variables
|
||||
|
||||
| Variable | Default | Notes |
|
||||
| -------------------------------- | --------------------- | -------------------------------------- |
|
||||
| `TX_THROUGHPUT_TUNNEL` | `ring` | `ring` / `tcp` / `udp` |
|
||||
| `TX_THROUGHPUT_PKT_SIZE` | `1400` | IP total length in bytes |
|
||||
| `TX_THROUGHPUT_WORKER_THREADS` | `4` | tokio worker threads |
|
||||
| `TX_THROUGHPUT_INFLIGHT` | `64` | saturate-mode concurrency (task count) |
|
||||
| `TX_THROUGHPUT_TUNNEL_PORT` | `35521` | tcp/udp listen port |
|
||||
| `TX_THROUGHPUT_MEASUREMENT_SECS` | `10` | Criterion `measurement_time` |
|
||||
| `TX_THROUGHPUT_WARMUP_SECS` | `3` | Criterion `warm_up_time` |
|
||||
| `TX_THROUGHPUT_SAMPLE_SIZE` | `10` | Criterion `sample_size` (min 10) |
|
||||
| `TX_THROUGHPUT_DOCKER_IMAGE` | `busybox:latest` | tcp/udp only |
|
||||
| `TX_THROUGHPUT_DOCKER_NET` | `easytier-bench-<id>` | auto-generated unique name |
|
||||
| `TX_THROUGHPUT_DOCKER_SUBNET` | `172.31.250.0/24` | |
|
||||
| `TX_THROUGHPUT_DOCKER_IP_A` | `172.31.250.2` | |
|
||||
| `TX_THROUGHPUT_DOCKER_IP_B` | `172.31.250.3` | |
|
||||
|
||||
## Parameter sweeps
|
||||
|
||||
```bash
|
||||
# Packet size
|
||||
for sz in 64 256 1400 9000; do
|
||||
TX_THROUGHPUT_PKT_SIZE=$sz cargo bench --bench tx_throughput -- --quick
|
||||
done
|
||||
|
||||
# Inflight depth (self-check: depth=1 should match serial baseline)
|
||||
for d in 1 4 16 64 256; do
|
||||
TX_THROUGHPUT_INFLIGHT=$d cargo bench --bench tx_throughput -- --quick
|
||||
done
|
||||
|
||||
# Worker threads
|
||||
for w in 1 2 4 8; do
|
||||
TX_THROUGHPUT_WORKER_THREADS=$w cargo bench --bench tx_throughput -- --quick
|
||||
done
|
||||
```
|
||||
|
||||
## Interpreting results
|
||||
|
||||
- **`<tunnel>`** reports per-packet latency. Lower is better. Throughput
|
||||
column here is "what one in-flight sender sustains".
|
||||
- **`<tunnel>-saturate`** reports aggregate throughput across
|
||||
`TX_THROUGHPUT_INFLIGHT` concurrent senders. If this matches the serial
|
||||
baseline, the TX path is bottlenecked on an internal serialization point
|
||||
(lock, single-threaded queue, etc.) rather than CPU or link bandwidth.
|
||||
|
||||
### Known finding (ring, single peer)
|
||||
|
||||
On the ring tunnel with a single destination peer, saturate does **not** beat
|
||||
serial (observed ~277 MiB/s saturate vs ~288 MiB/s serial on a 4-worker
|
||||
runtime). This points to a serialization point inside the peer-connection TX
|
||||
path. Tunnels with real I/O await points (tcp/udp via Docker) are expected to
|
||||
show a saturate > serial gap; verify with the sudo commands above.
|
||||
|
||||
## Output artifacts
|
||||
|
||||
Criterion writes HTML reports + SVG plots under
|
||||
`easytier/target/criterion/`. Open `tx_throughput/<tunnel>/report/index.html`
|
||||
or `.../<tunnel>-saturate/report/index.html` in a browser to inspect
|
||||
distributions and regressions across runs.
|
||||
@@ -0,0 +1,472 @@
|
||||
use std::{
|
||||
net::IpAddr,
|
||||
path::PathBuf,
|
||||
process::{Command, Stdio},
|
||||
str::FromStr,
|
||||
sync::Arc,
|
||||
sync::atomic::{AtomicU64, Ordering},
|
||||
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
use bytes::BytesMut;
|
||||
use criterion::{Criterion, Throughput, criterion_group, criterion_main};
|
||||
|
||||
use easytier::{
|
||||
common::config::{ConfigLoader, TomlConfigLoader},
|
||||
instance::instance::Instance,
|
||||
tunnel::{
|
||||
packet_def::ZCPacket, ring::RingTunnelConnector, tcp::TcpTunnelConnector,
|
||||
udp::UdpTunnelConnector,
|
||||
},
|
||||
};
|
||||
|
||||
const VIRTUAL_IP_A: &str = "10.144.144.1";
|
||||
const VIRTUAL_IP_B: &str = "10.144.144.2";
|
||||
const DEFAULT_DOCKER_SUBNET: &str = "172.31.250.0/24";
|
||||
const DEFAULT_DOCKER_IP_A: &str = "172.31.250.2";
|
||||
const DEFAULT_DOCKER_IP_B: &str = "172.31.250.3";
|
||||
const DEFAULT_TUNNEL_PORT: u16 = 35521;
|
||||
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
enum TunnelKind {
|
||||
Ring,
|
||||
Tcp,
|
||||
Udp,
|
||||
}
|
||||
|
||||
impl TunnelKind {
|
||||
fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
TunnelKind::Ring => "ring",
|
||||
TunnelKind::Tcp => "tcp",
|
||||
TunnelKind::Udp => "udp",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl FromStr for TunnelKind {
|
||||
type Err = String;
|
||||
|
||||
fn from_str(value: &str) -> Result<Self, Self::Err> {
|
||||
match value {
|
||||
"ring" => Ok(TunnelKind::Ring),
|
||||
"tcp" => Ok(TunnelKind::Tcp),
|
||||
"udp" => Ok(TunnelKind::Udp),
|
||||
other => Err(format!(
|
||||
"unsupported TX_THROUGHPUT_TUNNEL={other:?}; expected ring, tcp, or udp"
|
||||
)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct BenchTopology {
|
||||
_docker: Option<DockerNetns>,
|
||||
inst_a: Instance,
|
||||
_inst_b: Instance,
|
||||
dst: IpAddr,
|
||||
packet: ZCPacket,
|
||||
}
|
||||
|
||||
struct DockerNetns {
|
||||
network: String,
|
||||
container_a: String,
|
||||
container_b: String,
|
||||
netns_a: String,
|
||||
netns_b: String,
|
||||
ip_a: String,
|
||||
netns_a_path: PathBuf,
|
||||
netns_b_path: PathBuf,
|
||||
}
|
||||
|
||||
impl DockerNetns {
|
||||
fn create() -> Self {
|
||||
let id = unique_id();
|
||||
let image = env_string("TX_THROUGHPUT_DOCKER_IMAGE", "busybox:latest");
|
||||
let network = env_string("TX_THROUGHPUT_DOCKER_NET", &format!("easytier-bench-{id}"));
|
||||
let subnet = env_string("TX_THROUGHPUT_DOCKER_SUBNET", DEFAULT_DOCKER_SUBNET);
|
||||
let ip_a = env_string("TX_THROUGHPUT_DOCKER_IP_A", DEFAULT_DOCKER_IP_A);
|
||||
let ip_b = env_string("TX_THROUGHPUT_DOCKER_IP_B", DEFAULT_DOCKER_IP_B);
|
||||
let container_a = format!("easytier-bench-a-{id}");
|
||||
let container_b = format!("easytier-bench-b-{id}");
|
||||
let netns_a = format!("easytier-bench-a-{id}");
|
||||
let netns_b = format!("easytier-bench-b-{id}");
|
||||
|
||||
docker(&[
|
||||
"network", "create", "--driver", "bridge", "--subnet", &subnet, &network,
|
||||
]);
|
||||
|
||||
let mut docker_netns = Self {
|
||||
network,
|
||||
container_a,
|
||||
container_b,
|
||||
netns_a,
|
||||
netns_b,
|
||||
ip_a: ip_a.clone(),
|
||||
netns_a_path: PathBuf::new(),
|
||||
netns_b_path: PathBuf::new(),
|
||||
};
|
||||
|
||||
docker_netns.start_container(&docker_netns.container_a, &ip_a, &image);
|
||||
docker_netns.start_container(&docker_netns.container_b, &ip_b, &image);
|
||||
|
||||
let pid_a = docker(&["inspect", "-f", "{{.State.Pid}}", &docker_netns.container_a]);
|
||||
let pid_b = docker(&["inspect", "-f", "{{.State.Pid}}", &docker_netns.container_b]);
|
||||
|
||||
docker_netns.netns_a_path = register_netns(&docker_netns.netns_a, &pid_a);
|
||||
docker_netns.netns_b_path = register_netns(&docker_netns.netns_b, &pid_b);
|
||||
docker_netns
|
||||
}
|
||||
|
||||
fn start_container(&self, name: &str, ip: &str, image: &str) {
|
||||
docker(&[
|
||||
"run",
|
||||
"-d",
|
||||
"--name",
|
||||
name,
|
||||
"--network",
|
||||
&self.network,
|
||||
"--ip",
|
||||
ip,
|
||||
image,
|
||||
"sleep",
|
||||
"3600",
|
||||
]);
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for DockerNetns {
|
||||
fn drop(&mut self) {
|
||||
let _ = std::fs::remove_file(&self.netns_a_path);
|
||||
let _ = std::fs::remove_file(&self.netns_b_path);
|
||||
docker_ignore(&["rm", "-f", &self.container_a, &self.container_b]);
|
||||
docker_ignore(&["network", "rm", &self.network]);
|
||||
}
|
||||
}
|
||||
|
||||
fn bench_tx_throughput(c: &mut Criterion) {
|
||||
let tunnel = env_string("TX_THROUGHPUT_TUNNEL", "ring")
|
||||
.parse::<TunnelKind>()
|
||||
.unwrap_or_else(|err| panic!("{err}"));
|
||||
let packet_size = env_parse("TX_THROUGHPUT_PKT_SIZE", 1400usize);
|
||||
const MIN_PKT_SIZE: usize = 28; // IPv4 (20) + UDP (8) header
|
||||
assert!(
|
||||
packet_size >= MIN_PKT_SIZE,
|
||||
"TX_THROUGHPUT_PKT_SIZE={packet_size} is smaller than the minimum {MIN_PKT_SIZE} (IPv4+UDP headers)"
|
||||
);
|
||||
let worker_threads = env_parse("TX_THROUGHPUT_WORKER_THREADS", 4usize);
|
||||
let inflight_depth = env_parse("TX_THROUGHPUT_INFLIGHT", 64usize).max(1);
|
||||
let runtime = tokio::runtime::Builder::new_multi_thread()
|
||||
.worker_threads(worker_threads)
|
||||
.enable_all()
|
||||
.build()
|
||||
.expect("create tokio runtime");
|
||||
|
||||
let topology = runtime.block_on(setup_topology(tunnel, packet_size));
|
||||
let peer_manager = topology.inst_a.get_peer_manager();
|
||||
let packet = topology.packet.clone();
|
||||
let dst = topology.dst;
|
||||
|
||||
eprintln!(
|
||||
"tx_throughput: tunnel={} inflight={} workers={} pkt_size={}",
|
||||
tunnel.as_str(),
|
||||
inflight_depth.max(1),
|
||||
worker_threads,
|
||||
packet_size
|
||||
);
|
||||
|
||||
let mut group = c.benchmark_group("tx_throughput");
|
||||
group.throughput(Throughput::Bytes(packet_size as u64));
|
||||
|
||||
// Serial baseline: one packet in flight at a time.
|
||||
// Measures per-packet CPU cost (TX injection latency).
|
||||
group.bench_function(tunnel.as_str(), |b| {
|
||||
b.iter_custom(|iterations| {
|
||||
let pm = peer_manager.clone();
|
||||
let pkt = packet.clone();
|
||||
runtime.block_on(async move {
|
||||
let start = Instant::now();
|
||||
for _ in 0..iterations {
|
||||
pm.send_msg_by_ip(pkt.clone(), dst, false)
|
||||
.await
|
||||
.expect("send packet by EasyTier IP");
|
||||
}
|
||||
start.elapsed()
|
||||
})
|
||||
});
|
||||
});
|
||||
|
||||
// Saturate: spawn TX_THROUGHPUT_INFLIGHT worker tasks, each independently
|
||||
// pumping send_msg_by_ip. Work is distributed across tokio worker threads,
|
||||
// exposing the peer manager + tunnel's true aggregate throughput ceiling.
|
||||
// With TX_THROUGHPUT_INFLIGHT=1 it degrades to the serial baseline.
|
||||
group.bench_function(format!("{}-saturate", tunnel.as_str()), |b| {
|
||||
b.iter_custom(|iterations| {
|
||||
let pm = peer_manager.clone();
|
||||
let pkt = packet.clone();
|
||||
let concurrency = inflight_depth.min(iterations as usize).max(1);
|
||||
runtime.block_on(async move {
|
||||
let counter = Arc::new(AtomicU64::new(iterations));
|
||||
let start = Instant::now();
|
||||
let mut handles = Vec::with_capacity(concurrency);
|
||||
for _ in 0..concurrency {
|
||||
let pm = pm.clone();
|
||||
let pkt = pkt.clone();
|
||||
let counter = counter.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
loop {
|
||||
if counter
|
||||
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |cur| {
|
||||
if cur > 0 { Some(cur - 1) } else { None }
|
||||
})
|
||||
.is_err()
|
||||
{
|
||||
return;
|
||||
}
|
||||
pm.send_msg_by_ip(pkt.clone(), dst, false)
|
||||
.await
|
||||
.expect("send packet by EasyTier IP");
|
||||
}
|
||||
}));
|
||||
}
|
||||
for h in handles {
|
||||
h.await.expect("saturate worker task panicked");
|
||||
}
|
||||
start.elapsed()
|
||||
})
|
||||
});
|
||||
});
|
||||
|
||||
group.finish();
|
||||
|
||||
runtime.block_on(async move {
|
||||
drop(topology);
|
||||
});
|
||||
}
|
||||
|
||||
async fn setup_topology(tunnel: TunnelKind, packet_size: usize) -> BenchTopology {
|
||||
let tunnel_port = env_parse("TX_THROUGHPUT_TUNNEL_PORT", DEFAULT_TUNNEL_PORT);
|
||||
let docker = match tunnel {
|
||||
TunnelKind::Ring => None,
|
||||
TunnelKind::Tcp | TunnelKind::Udp => Some(DockerNetns::create()),
|
||||
};
|
||||
|
||||
let (netns_a, netns_b) = match &docker {
|
||||
Some(docker) => (Some(docker.netns_a.clone()), Some(docker.netns_b.clone())),
|
||||
None => (None, None),
|
||||
};
|
||||
let listeners_a = match tunnel {
|
||||
TunnelKind::Ring => Vec::new(),
|
||||
TunnelKind::Tcp | TunnelKind::Udp => vec![
|
||||
format!("{}://0.0.0.0:{}", tunnel.as_str(), tunnel_port)
|
||||
.parse()
|
||||
.unwrap(),
|
||||
],
|
||||
};
|
||||
|
||||
let mut inst_a = Instance::new(no_tun_config("hot-a", VIRTUAL_IP_A, netns_a, listeners_a));
|
||||
let mut inst_b = Instance::new(no_tun_config("hot-b", VIRTUAL_IP_B, netns_b, Vec::new()));
|
||||
|
||||
inst_a.run().await.expect("inst_a run");
|
||||
inst_b.run().await.expect("inst_b run");
|
||||
|
||||
match tunnel {
|
||||
TunnelKind::Ring => inst_b
|
||||
.get_conn_manager()
|
||||
.add_connector(RingTunnelConnector::new(
|
||||
format!("ring://{}", inst_a.id()).parse().unwrap(),
|
||||
)),
|
||||
TunnelKind::Tcp => inst_b
|
||||
.get_conn_manager()
|
||||
.add_connector(TcpTunnelConnector::new(
|
||||
format!(
|
||||
"tcp://{}:{}",
|
||||
docker.as_ref().expect("tcp benchmark needs Docker").ip_a,
|
||||
tunnel_port
|
||||
)
|
||||
.parse()
|
||||
.unwrap(),
|
||||
)),
|
||||
TunnelKind::Udp => inst_b
|
||||
.get_conn_manager()
|
||||
.add_connector(UdpTunnelConnector::new(
|
||||
format!(
|
||||
"udp://{}:{}",
|
||||
docker.as_ref().expect("udp benchmark needs Docker").ip_a,
|
||||
tunnel_port
|
||||
)
|
||||
.parse()
|
||||
.unwrap(),
|
||||
)),
|
||||
}
|
||||
|
||||
wait_for_routes(&inst_a, &inst_b).await;
|
||||
|
||||
BenchTopology {
|
||||
_docker: docker,
|
||||
inst_a,
|
||||
_inst_b: inst_b,
|
||||
dst: VIRTUAL_IP_B.parse().unwrap(),
|
||||
packet: make_data_packet(VIRTUAL_IP_A, VIRTUAL_IP_B, packet_size),
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait_for_routes(inst_a: &Instance, inst_b: &Instance) {
|
||||
tokio::time::timeout(Duration::from_secs(15), async {
|
||||
loop {
|
||||
let routes_a = inst_a.get_peer_manager().list_routes().await;
|
||||
let routes_b = inst_b.get_peer_manager().list_routes().await;
|
||||
if !routes_a.is_empty() && !routes_b.is_empty() {
|
||||
return;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("EasyTier routes did not converge within 15s");
|
||||
}
|
||||
|
||||
fn make_data_packet(src: &str, dst: &str, total_size: usize) -> ZCPacket {
|
||||
use std::net::Ipv4Addr;
|
||||
|
||||
let hdr_len = 28;
|
||||
let payload_len = total_size.saturating_sub(hdr_len);
|
||||
let ip_total_len = (hdr_len + payload_len) as u16;
|
||||
let mut buf = BytesMut::with_capacity(total_size);
|
||||
|
||||
buf.extend_from_slice(&[
|
||||
0x45,
|
||||
0x00,
|
||||
(ip_total_len >> 8) as u8,
|
||||
(ip_total_len & 0xff) as u8,
|
||||
0x00,
|
||||
0x00,
|
||||
0x40,
|
||||
0x00,
|
||||
0x40,
|
||||
0x11,
|
||||
0x00,
|
||||
0x00,
|
||||
]);
|
||||
let src: Ipv4Addr = src.parse().unwrap();
|
||||
buf.extend_from_slice(&src.octets());
|
||||
let dst: Ipv4Addr = dst.parse().unwrap();
|
||||
buf.extend_from_slice(&dst.octets());
|
||||
|
||||
let udp_len = (8 + payload_len) as u16;
|
||||
buf.extend_from_slice(&[
|
||||
0x30,
|
||||
0x39,
|
||||
0xd4,
|
||||
0x31,
|
||||
(udp_len >> 8) as u8,
|
||||
(udp_len & 0xff) as u8,
|
||||
0x00,
|
||||
0x00,
|
||||
]);
|
||||
|
||||
buf.resize(total_size, 0xaa);
|
||||
ZCPacket::new_with_payload(&buf)
|
||||
}
|
||||
|
||||
fn no_tun_config(
|
||||
name: &str,
|
||||
ipv4: &str,
|
||||
netns: Option<String>,
|
||||
listeners: Vec<url::Url>,
|
||||
) -> TomlConfigLoader {
|
||||
let config = TomlConfigLoader::default();
|
||||
config.set_inst_name(name.to_owned());
|
||||
config.set_netns(netns);
|
||||
config.set_ipv4(Some(ipv4.parse().unwrap()));
|
||||
config.set_listeners(listeners);
|
||||
let mut flags = config.get_flags();
|
||||
flags.no_tun = true;
|
||||
config.set_flags(flags);
|
||||
config
|
||||
}
|
||||
|
||||
fn register_netns(name: &str, pid: &str) -> PathBuf {
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
let dir = PathBuf::from("/var/run/netns");
|
||||
std::fs::create_dir_all(&dir).expect("create /var/run/netns");
|
||||
let path = dir.join(name);
|
||||
let _ = std::fs::remove_file(&path);
|
||||
std::os::unix::fs::symlink(format!("/proc/{pid}/ns/net"), &path)
|
||||
.expect("link Docker netns into /var/run/netns");
|
||||
path
|
||||
}
|
||||
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
{
|
||||
let _ = (name, pid);
|
||||
panic!("Docker netns benchmark requires Linux");
|
||||
}
|
||||
}
|
||||
|
||||
fn docker(args: &[&str]) -> String {
|
||||
let output = Command::new("docker")
|
||||
.args(args)
|
||||
.output()
|
||||
.unwrap_or_else(|err| panic!("failed to run docker {args:?}: {err}"));
|
||||
if !output.status.success() {
|
||||
panic!(
|
||||
"docker {:?} failed with status {:?}: {}",
|
||||
args,
|
||||
output.status.code(),
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
}
|
||||
String::from_utf8_lossy(&output.stdout).trim().to_owned()
|
||||
}
|
||||
|
||||
fn docker_ignore(args: &[&str]) {
|
||||
let _ = Command::new("docker")
|
||||
.args(args)
|
||||
.stdout(Stdio::null())
|
||||
.stderr(Stdio::null())
|
||||
.status();
|
||||
}
|
||||
|
||||
fn env_string(name: &str, default: &str) -> String {
|
||||
std::env::var(name).unwrap_or_else(|_| default.to_owned())
|
||||
}
|
||||
|
||||
fn env_parse<T>(name: &str, default: T) -> T
|
||||
where
|
||||
T: FromStr,
|
||||
T::Err: std::fmt::Display,
|
||||
{
|
||||
match std::env::var(name) {
|
||||
Ok(value) => value
|
||||
.parse()
|
||||
.unwrap_or_else(|err| panic!("invalid {name}={value:?}: {err}")),
|
||||
Err(_) => default,
|
||||
}
|
||||
}
|
||||
|
||||
fn criterion_config() -> Criterion {
|
||||
let measurement_secs = env_parse("TX_THROUGHPUT_MEASUREMENT_SECS", 10u64);
|
||||
let warmup_secs = env_parse("TX_THROUGHPUT_WARMUP_SECS", 3u64);
|
||||
let sample_size = env_parse("TX_THROUGHPUT_SAMPLE_SIZE", 10usize).max(10);
|
||||
|
||||
Criterion::default()
|
||||
.measurement_time(Duration::from_secs(measurement_secs))
|
||||
.warm_up_time(Duration::from_secs(warmup_secs))
|
||||
.sample_size(sample_size)
|
||||
}
|
||||
|
||||
fn unique_id() -> String {
|
||||
let nanos = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.expect("system clock before UNIX epoch")
|
||||
.as_nanos();
|
||||
format!("{}-{nanos}", std::process::id())
|
||||
}
|
||||
|
||||
criterion_group! {
|
||||
name = benches;
|
||||
config = criterion_config();
|
||||
targets = bench_tx_throughput
|
||||
}
|
||||
criterion_main!(benches);
|
||||
@@ -6,7 +6,7 @@ use std::{
|
||||
time::{Duration, SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
|
||||
use crate::common::{config::ConfigLoader, global_ctx::ArcGlobalCtx, token_bucket::TokenBucket};
|
||||
use crate::proto::acl::*;
|
||||
|
||||
@@ -219,6 +219,7 @@ pub struct GlobalCtx {
|
||||
|
||||
running_listeners: Mutex<Vec<url::Url>>,
|
||||
advertised_ipv6_public_addr_prefix: Mutex<Option<cidr::Ipv6Cidr>>,
|
||||
tun_device_name: Mutex<Option<String>>,
|
||||
|
||||
flags: ArcSwap<Flags>,
|
||||
|
||||
@@ -336,6 +337,7 @@ impl GlobalCtx {
|
||||
|
||||
running_listeners: Mutex::new(Vec::new()),
|
||||
advertised_ipv6_public_addr_prefix: Mutex::new(None),
|
||||
tun_device_name: Mutex::new(None),
|
||||
|
||||
flags: ArcSwap::new(Arc::new(flags)),
|
||||
|
||||
@@ -370,6 +372,24 @@ impl GlobalCtx {
|
||||
}
|
||||
}
|
||||
|
||||
fn set_tun_device_name(&self, name: Option<String>) {
|
||||
*self.tun_device_name.lock().unwrap() = name;
|
||||
}
|
||||
|
||||
pub(crate) fn set_tun_device_ready(&self, name: String) {
|
||||
self.set_tun_device_name(Some(name.clone()));
|
||||
self.issue_event(GlobalCtxEvent::TunDeviceReady(name));
|
||||
}
|
||||
|
||||
pub(crate) fn set_tun_device_error(&self, error: String) {
|
||||
self.set_tun_device_name(None);
|
||||
self.issue_event(GlobalCtxEvent::TunDeviceError(error));
|
||||
}
|
||||
|
||||
pub fn get_tun_device_name(&self) -> Option<String> {
|
||||
self.tun_device_name.lock().unwrap().clone()
|
||||
}
|
||||
|
||||
pub fn check_network_in_whitelist(&self, network_name: &str) -> Result<(), anyhow::Error> {
|
||||
if self
|
||||
.get_flags()
|
||||
@@ -825,6 +845,36 @@ pub mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_tun_device_name_tracks_explicit_runtime_state() {
|
||||
let config = TomlConfigLoader::default();
|
||||
let global_ctx = GlobalCtx::new(config);
|
||||
|
||||
assert_eq!(global_ctx.get_tun_device_name(), None);
|
||||
|
||||
global_ctx.issue_event(GlobalCtxEvent::TunDeviceReady("ignored".to_string()));
|
||||
assert_eq!(global_ctx.get_tun_device_name(), None);
|
||||
|
||||
let mut subscriber = global_ctx.subscribe();
|
||||
|
||||
global_ctx.set_tun_device_ready("easytier0".to_string());
|
||||
assert_eq!(
|
||||
global_ctx.get_tun_device_name(),
|
||||
Some("easytier0".to_string())
|
||||
);
|
||||
assert_eq!(
|
||||
subscriber.recv().await.unwrap(),
|
||||
GlobalCtxEvent::TunDeviceReady("easytier0".to_string())
|
||||
);
|
||||
|
||||
global_ctx.set_tun_device_error("closed".to_string());
|
||||
assert_eq!(global_ctx.get_tun_device_name(), None);
|
||||
assert_eq!(
|
||||
subscriber.recv().await.unwrap(),
|
||||
GlobalCtxEvent::TunDeviceError("closed".to_string())
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn trusted_key_source_lookup_is_precise() {
|
||||
let config = TomlConfigLoader::default();
|
||||
|
||||
@@ -177,3 +177,20 @@ pub(crate) fn list_ipv6_route_messages()
|
||||
pub(crate) fn get_interface_index(name: &str) -> Result<u32, Error> {
|
||||
netlink::NetlinkIfConfiger::get_interface_index(name)
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
pub(crate) fn add_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
|
||||
netlink::NetlinkIfConfiger::add_ipv6_ndp_proxy(name, address)
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
pub(crate) fn remove_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
|
||||
netlink::NetlinkIfConfiger::remove_ipv6_ndp_proxy(name, address)
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
pub(crate) fn list_ipv6_ndp_proxy(
|
||||
name: &str,
|
||||
) -> Result<std::collections::BTreeSet<Ipv6Addr>, Error> {
|
||||
netlink::NetlinkIfConfiger::list_ipv6_ndp_proxy(name)
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
use std::{
|
||||
collections::BTreeSet,
|
||||
ffi::CString,
|
||||
fmt::Debug,
|
||||
net::{IpAddr, Ipv4Addr, Ipv6Addr},
|
||||
@@ -16,6 +17,10 @@ use netlink_packet_core::{
|
||||
use netlink_packet_route::{
|
||||
AddressFamily, RouteNetlinkMessage,
|
||||
address::{AddressAttribute, AddressMessage},
|
||||
neighbour::{
|
||||
NeighbourAddress, NeighbourAttribute, NeighbourFlags, NeighbourHeader, NeighbourMessage,
|
||||
NeighbourState,
|
||||
},
|
||||
route::{
|
||||
RouteAddress, RouteAttribute, RouteHeader, RouteMessage, RouteProtocol, RouteScope,
|
||||
RouteType,
|
||||
@@ -375,6 +380,105 @@ impl NetlinkIfConfiger {
|
||||
pub(crate) fn list_ipv6_route_messages() -> Result<Vec<RouteMessage>, Error> {
|
||||
Self::list_route_messages(AddressFamily::Inet6)
|
||||
}
|
||||
|
||||
fn ipv6_ndp_proxy_message(name: &str, address: Ipv6Addr) -> Result<NeighbourMessage, Error> {
|
||||
let mut message = NeighbourMessage::default();
|
||||
message.header = NeighbourHeader {
|
||||
family: AddressFamily::Inet6,
|
||||
ifindex: Self::get_interface_index(name)?,
|
||||
state: NeighbourState::Permanent,
|
||||
flags: NeighbourFlags::Proxy,
|
||||
kind: RouteType::Unicast,
|
||||
};
|
||||
message
|
||||
.attributes
|
||||
.push(NeighbourAttribute::Destination(NeighbourAddress::Inet6(
|
||||
address,
|
||||
)));
|
||||
Ok(message)
|
||||
}
|
||||
|
||||
pub(crate) fn add_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
|
||||
send_netlink_req_and_wait_one_resp(
|
||||
RouteNetlinkMessage::NewNeighbour(Self::ipv6_ndp_proxy_message(name, address)?),
|
||||
false,
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn remove_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
|
||||
send_netlink_req_and_wait_one_resp(
|
||||
RouteNetlinkMessage::DelNeighbour(Self::ipv6_ndp_proxy_message(name, address)?),
|
||||
true,
|
||||
)
|
||||
}
|
||||
|
||||
fn list_neighbour_messages(
|
||||
address_family: AddressFamily,
|
||||
) -> Result<Vec<NeighbourMessage>, Error> {
|
||||
let mut message = NeighbourMessage::default();
|
||||
message.header.family = address_family;
|
||||
message.header.flags = NeighbourFlags::Proxy;
|
||||
|
||||
let s = send_netlink_req(
|
||||
RouteNetlinkMessage::GetNeighbour(message),
|
||||
NLM_F_REQUEST | NLM_F_DUMP,
|
||||
)?;
|
||||
|
||||
let mut ret_vec = vec![];
|
||||
let mut resp = Vec::<u8>::new();
|
||||
loop {
|
||||
if resp.is_empty() {
|
||||
let (new_resp, _) = s.recv_from_full()?;
|
||||
resp = new_resp;
|
||||
}
|
||||
|
||||
let ret = NetlinkMessage::<RouteNetlinkMessage>::deserialize(&resp)
|
||||
.with_context(|| "Failed to deserialize netlink neighbour message")?;
|
||||
resp = resp.split_off(ret.buffer_len());
|
||||
|
||||
tracing::debug!("net link response <<< {:?}", ret);
|
||||
|
||||
match ret.payload {
|
||||
NetlinkPayload::Error(e) => {
|
||||
if e.code == NonZero::new(0) {
|
||||
continue;
|
||||
} else {
|
||||
return Err(e.to_io().into());
|
||||
}
|
||||
}
|
||||
NetlinkPayload::InnerMessage(RouteNetlinkMessage::NewNeighbour(m)) => {
|
||||
ret_vec.push(m);
|
||||
}
|
||||
NetlinkPayload::Done(_) => {
|
||||
break;
|
||||
}
|
||||
p => {
|
||||
tracing::error!("Unexpected netlink response: {:?}", p);
|
||||
return Err(anyhow::anyhow!("Unexpected netlink response").into());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(ret_vec)
|
||||
}
|
||||
|
||||
pub(crate) fn list_ipv6_ndp_proxy(name: &str) -> Result<BTreeSet<Ipv6Addr>, Error> {
|
||||
let ifindex = Self::get_interface_index(name)?;
|
||||
|
||||
Ok(Self::list_neighbour_messages(AddressFamily::Inet6)?
|
||||
.into_iter()
|
||||
.filter(|message| {
|
||||
message.header.ifindex == ifindex
|
||||
&& message.header.flags.contains(NeighbourFlags::Proxy)
|
||||
})
|
||||
.filter_map(|message| {
|
||||
message.attributes.into_iter().find_map(|attr| match attr {
|
||||
NeighbourAttribute::Destination(NeighbourAddress::Inet6(addr)) => Some(addr),
|
||||
_ => None,
|
||||
})
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::cell::UnsafeCell;
|
||||
use std::fmt;
|
||||
|
||||
@@ -8,7 +8,7 @@ use crate::proto::common::{NatType, StunInfo};
|
||||
use anyhow::Context;
|
||||
use chrono::Local;
|
||||
use crossbeam::atomic::AtomicCell;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use rand::seq::IteratorRandom;
|
||||
use socket2::{SockAddr, SockRef};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
@@ -11,7 +11,7 @@ use std::{
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
|
||||
use crate::{
|
||||
common::{
|
||||
|
||||
@@ -6,7 +6,7 @@ use std::{
|
||||
};
|
||||
|
||||
use dashmap::DashSet;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use tokio::{sync::mpsc, task::JoinSet, time::timeout};
|
||||
|
||||
use crate::{
|
||||
|
||||
@@ -5,7 +5,7 @@ use std::{
|
||||
};
|
||||
|
||||
use anyhow::{Context, Error};
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use rand::Rng as _;
|
||||
use tokio::task::JoinSet;
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ use std::{
|
||||
};
|
||||
|
||||
use anyhow::Context;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use tokio::sync::Mutex;
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
|
||||
|
||||
@@ -7,7 +7,7 @@ use std::{
|
||||
use crossbeam::atomic::AtomicCell;
|
||||
use dashmap::{DashMap, DashSet};
|
||||
use guarden::defer;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use rand::seq::SliceRandom as _;
|
||||
use tokio::{net::UdpSocket, sync::Mutex, task::JoinSet};
|
||||
use tracing::{Instrument, Level, instrument};
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
use std::{sync::Arc, time::Duration};
|
||||
|
||||
use anyhow::Context;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use tokio::net::UdpSocket;
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
|
||||
|
||||
@@ -8,8 +8,8 @@ use both_easy_sym::{PunchBothEasySymHoleClient, PunchBothEasySymHoleServer};
|
||||
use common::{PunchHoleServerCommon, UdpNatType, UdpPunchClientMethod};
|
||||
use cone::{PunchConeHoleClient, PunchConeHoleServer};
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use once_cell::sync::Lazy;
|
||||
use quanta::Instant;
|
||||
use sym_to_cone::{PunchSymToConeHoleClient, PunchSymToConeHoleServer};
|
||||
use tokio::{sync::Mutex, task::JoinHandle};
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ use std::{
|
||||
|
||||
use anyhow::Context;
|
||||
use guarden::defer;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use rand::{Rng, seq::SliceRandom};
|
||||
use tokio::{net::UdpSocket, sync::RwLock};
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
|
||||
@@ -1,11 +1,5 @@
|
||||
use easytier::core;
|
||||
|
||||
#[cfg(all(
|
||||
feature = "hotpath-alloc",
|
||||
any(feature = "jemalloc", feature = "mimalloc")
|
||||
))]
|
||||
compile_error!("feature `hotpath-alloc` cannot be enabled together with `jemalloc` or `mimalloc`");
|
||||
|
||||
#[cfg(all(feature = "mimalloc", not(feature = "jemalloc")))]
|
||||
use mimalloc::MiMalloc;
|
||||
|
||||
|
||||
@@ -7,13 +7,13 @@ use std::{
|
||||
};
|
||||
|
||||
use anyhow::Context;
|
||||
use hotpath::instant::Instant;
|
||||
use pnet::packet::{
|
||||
Packet,
|
||||
icmp::{self, IcmpCode, IcmpTypes, MutableIcmpPacket, echo_reply::MutableEchoReplyPacket},
|
||||
ip::IpNextHeaderProtocols,
|
||||
ipv4::Ipv4Packet,
|
||||
};
|
||||
use quanta::Instant;
|
||||
use socket2::Socket;
|
||||
use tokio::{
|
||||
sync::{Mutex, mpsc::UnboundedSender},
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use pnet::packet::Packet;
|
||||
use pnet::packet::ip::IpNextHeaderProtocol;
|
||||
use pnet::packet::ipv4::{self, Ipv4Flags, Ipv4Packet, MutableIpv4Packet};
|
||||
use quanta::Instant;
|
||||
use std::net::Ipv4Addr;
|
||||
use std::time::Duration;
|
||||
|
||||
|
||||
@@ -1018,7 +1018,7 @@ impl TcpProxyRpc for QuicProxyDstRpcService {
|
||||
mod tests {
|
||||
use super::*;
|
||||
use bytes::Buf;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
|
||||
/// Helper function: Create a pair of interconnected QuicSockets.
|
||||
/// Data sent by socket_a will enter socket_b's rx, and vice versa.
|
||||
|
||||
@@ -9,9 +9,9 @@ use std::{
|
||||
};
|
||||
|
||||
use crossbeam::atomic::AtomicCell;
|
||||
use hotpath::instant::Instant;
|
||||
#[cfg(feature = "kcp")]
|
||||
use kcp_sys::{endpoint::KcpEndpoint, stream::KcpStream};
|
||||
use quanta::Instant;
|
||||
use tokio_util::sync::{CancellationToken, DropGuard};
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
|
||||
|
||||
@@ -26,7 +26,7 @@ use std::{
|
||||
};
|
||||
|
||||
use anyhow::Context as _;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use tokio::io::{AsyncRead, AsyncWrite};
|
||||
|
||||
use crate::{common::error::Error, gateway::fast_socks5::server::AsyncTcpConnector};
|
||||
|
||||
@@ -3,12 +3,12 @@ use cidr::Ipv4Inet;
|
||||
use core::panic;
|
||||
use crossbeam::atomic::AtomicCell;
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use pnet::packet::MutablePacket;
|
||||
use pnet::packet::Packet;
|
||||
use pnet::packet::ip::IpNextHeaderProtocols;
|
||||
use pnet::packet::ipv4::{Ipv4Packet, MutableIpv4Packet};
|
||||
use pnet::packet::tcp::{MutableTcpPacket, TcpPacket, ipv4_checksum};
|
||||
use quanta::Instant;
|
||||
use socket2::{SockRef, TcpKeepalive};
|
||||
use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU16};
|
||||
|
||||
@@ -8,13 +8,13 @@ use bytes::{BufMut, BytesMut};
|
||||
use cidr::Ipv4Inet;
|
||||
use crossbeam::atomic::AtomicCell;
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use pnet::packet::{
|
||||
Packet,
|
||||
ip::IpNextHeaderProtocols,
|
||||
ipv4::Ipv4Packet,
|
||||
udp::{self, MutableUdpPacket},
|
||||
};
|
||||
use quanta::Instant;
|
||||
use tokio::sync::mpsc::{Receiver, Sender, channel, error::TrySendError};
|
||||
use tokio::{
|
||||
net::UdpSocket,
|
||||
|
||||
@@ -38,3 +38,15 @@ macro_rules! rw_lock {
|
||||
$expr
|
||||
};
|
||||
}
|
||||
|
||||
/// Type-level mirror of `hotpath::wrap` for type positions: with the feature
|
||||
/// off, `channel!` returns the original endpoints, so the wrapped endpoint
|
||||
/// types are the plain channel types.
|
||||
pub(crate) mod wrap {
|
||||
pub(crate) mod tokio {
|
||||
pub(crate) mod sync {
|
||||
pub(crate) use ::tokio::sync::mpsc;
|
||||
}
|
||||
}
|
||||
pub(crate) use ::flume;
|
||||
}
|
||||
|
||||
@@ -65,9 +65,9 @@ use crate::vpn_portal::{self, VpnPortal};
|
||||
use super::dns_server::{MAGIC_DNS_FAKE_IP, runner::DnsRunner};
|
||||
use super::listeners::ListenerManager;
|
||||
use super::public_ipv6_provider::{
|
||||
reconcile_public_ipv6_provider_runtime, run_public_ipv6_provider_reconcile_task,
|
||||
should_run_public_ipv6_provider_reconcile, validate_public_ipv6_config,
|
||||
validate_public_ipv6_config_values,
|
||||
PublicIpv6ProviderReconcileTask, reconcile_public_ipv6_provider_runtime,
|
||||
run_public_ipv6_provider_reconcile_task, should_run_public_ipv6_provider_reconcile,
|
||||
validate_public_ipv6_config, validate_public_ipv6_config_values,
|
||||
};
|
||||
|
||||
#[cfg(feature = "socks5")]
|
||||
@@ -194,6 +194,44 @@ impl NicCtxContainer {
|
||||
|
||||
#[cfg(feature = "tun")]
|
||||
type ArcNicCtx = Arc<Mutex<Option<NicCtxContainer>>>;
|
||||
type ArcPublicIpv6ProviderTaskSlot = Arc<PublicIpv6ProviderTaskSlot>;
|
||||
|
||||
struct PublicIpv6ProviderTaskSlot {
|
||||
task: Mutex<Option<PublicIpv6ProviderReconcileTask>>,
|
||||
closing: AtomicBool,
|
||||
}
|
||||
|
||||
impl PublicIpv6ProviderTaskSlot {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
task: Mutex::new(None),
|
||||
closing: AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
|
||||
async fn ensure_started(&self, global_ctx: &ArcGlobalCtx) {
|
||||
let mut task = self.task.lock().await;
|
||||
if self.closing.load(Ordering::Acquire) || task.is_some() {
|
||||
return;
|
||||
}
|
||||
*task = run_public_ipv6_provider_reconcile_task(global_ctx);
|
||||
}
|
||||
|
||||
async fn shutdown(&self) {
|
||||
self.closing.store(true, Ordering::Release);
|
||||
let task = self.task.lock().await.take();
|
||||
if let Some(task) = task {
|
||||
task.shutdown().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn ensure_public_ipv6_provider_reconcile_task(
|
||||
global_ctx: &ArcGlobalCtx,
|
||||
task_slot: &ArcPublicIpv6ProviderTaskSlot,
|
||||
) {
|
||||
task_slot.ensure_started(global_ctx).await;
|
||||
}
|
||||
|
||||
pub struct InstanceRpcServerHook {
|
||||
rpc_portal_whitelist: Vec<IpCidr>,
|
||||
@@ -254,6 +292,7 @@ pub struct InstanceConfigPatcher {
|
||||
socks5_server: Weak<Socks5Server>,
|
||||
peer_manager: Weak<PeerManager>,
|
||||
conn_manager: Weak<ManualConnectorManager>,
|
||||
public_ipv6_provider_task: ArcPublicIpv6ProviderTaskSlot,
|
||||
}
|
||||
|
||||
impl InstanceConfigPatcher {
|
||||
@@ -324,7 +363,6 @@ impl InstanceConfigPatcher {
|
||||
self.patch_mapped_listeners(patch.mapped_listeners).await?;
|
||||
self.patch_connector(patch.connectors).await?;
|
||||
|
||||
let provider_reconcile_was_running = should_run_public_ipv6_provider_reconcile(&global_ctx);
|
||||
let mut provider_config_changed = false;
|
||||
if let Some(hostname) = patch.hostname {
|
||||
global_ctx.set_hostname(hostname.clone());
|
||||
@@ -362,10 +400,12 @@ impl InstanceConfigPatcher {
|
||||
if provider_config_changed {
|
||||
reconcile_public_ipv6_provider_runtime(&global_ctx).await;
|
||||
|
||||
let provider_reconcile_should_run =
|
||||
should_run_public_ipv6_provider_reconcile(&global_ctx);
|
||||
if !provider_reconcile_was_running && provider_reconcile_should_run {
|
||||
run_public_ipv6_provider_reconcile_task(&global_ctx);
|
||||
if should_run_public_ipv6_provider_reconcile(&global_ctx) {
|
||||
ensure_public_ipv6_provider_reconcile_task(
|
||||
&global_ctx,
|
||||
&self.public_ipv6_provider_task,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -647,6 +687,7 @@ pub struct Instance {
|
||||
socks5_server: Arc<Socks5Server>,
|
||||
|
||||
proxy_cidrs_monitor: Option<AbortOnDropHandle<()>>,
|
||||
public_ipv6_provider_task: ArcPublicIpv6ProviderTaskSlot,
|
||||
|
||||
global_ctx: ArcGlobalCtx,
|
||||
}
|
||||
@@ -734,6 +775,7 @@ impl Instance {
|
||||
socks5_server,
|
||||
|
||||
proxy_cidrs_monitor: None,
|
||||
public_ipv6_provider_task: Arc::new(PublicIpv6ProviderTaskSlot::new()),
|
||||
|
||||
global_ctx,
|
||||
}
|
||||
@@ -1034,7 +1076,11 @@ impl Instance {
|
||||
.await?;
|
||||
self.listener_manager.lock().await.run().await?;
|
||||
self.peer_manager.run().await?;
|
||||
run_public_ipv6_provider_reconcile_task(&self.global_ctx);
|
||||
ensure_public_ipv6_provider_reconcile_task(
|
||||
&self.global_ctx,
|
||||
&self.public_ipv6_provider_task,
|
||||
)
|
||||
.await;
|
||||
|
||||
#[cfg(feature = "tun")]
|
||||
{
|
||||
@@ -1347,6 +1393,7 @@ impl Instance {
|
||||
socks5_server: Arc::downgrade(&self.socks5_server),
|
||||
peer_manager: Arc::downgrade(&self.peer_manager),
|
||||
conn_manager: Arc::downgrade(&self.conn_manager),
|
||||
public_ipv6_provider_task: self.public_ipv6_provider_task.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1602,6 +1649,7 @@ impl Instance {
|
||||
}
|
||||
|
||||
pub async fn clear_resources(&mut self) {
|
||||
self.public_ipv6_provider_task.shutdown().await;
|
||||
self.peer_manager.clear_resources().await;
|
||||
#[cfg(feature = "tun")]
|
||||
let _ = self.nic_ctx.lock().await.take();
|
||||
@@ -1787,6 +1835,21 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn public_ipv6_provider_task_slot_does_not_restart_after_shutdown() {
|
||||
let global_ctx = get_mock_global_ctx();
|
||||
let slot = std::sync::Arc::new(super::PublicIpv6ProviderTaskSlot::new());
|
||||
global_ctx.config.set_ipv6_public_addr_provider(true);
|
||||
global_ctx
|
||||
.config
|
||||
.set_ipv6_public_addr_prefix(Some("2001:db8::/48".parse().unwrap()));
|
||||
|
||||
slot.shutdown().await;
|
||||
super::ensure_public_ipv6_provider_reconcile_task(&global_ctx, &slot).await;
|
||||
|
||||
assert!(slot.task.lock().await.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn validate_public_ipv6_patch_allows_enabling_auto_with_manual_ipv6() {
|
||||
let global_ctx = get_mock_global_ctx();
|
||||
|
||||
@@ -3,7 +3,7 @@ use std::sync::{Arc, Weak};
|
||||
|
||||
use crate::common::global_ctx::{ArcGlobalCtx, GlobalCtxEvent};
|
||||
use crate::peers::peer_manager::PeerManager;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
|
||||
/// ProxyCidrsMonitor monitors changes in proxy CIDRs from peer routes
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1361,12 +1361,11 @@ impl NicCtx {
|
||||
}
|
||||
|
||||
self.global_ctx
|
||||
.issue_event(GlobalCtxEvent::TunDeviceReady(nic.ifname().to_string()));
|
||||
.set_tun_device_ready(nic.ifname().to_string());
|
||||
ret
|
||||
}
|
||||
Err(err) => {
|
||||
self.global_ctx
|
||||
.issue_event(GlobalCtxEvent::TunDeviceError(err.to_string()));
|
||||
self.global_ctx.set_tun_device_error(err.to_string());
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
@@ -1405,12 +1404,11 @@ impl NicCtx {
|
||||
match nic.create_dev_for_mobile(tun_fd).await {
|
||||
Ok(ret) => {
|
||||
self.global_ctx
|
||||
.issue_event(GlobalCtxEvent::TunDeviceReady(nic.ifname().to_string()));
|
||||
.set_tun_device_ready(nic.ifname().to_string());
|
||||
ret
|
||||
}
|
||||
Err(err) => {
|
||||
self.global_ctx
|
||||
.issue_event(GlobalCtxEvent::TunDeviceError(err.to_string()));
|
||||
self.global_ctx.set_tun_device_error(err.to_string());
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -173,7 +173,12 @@ impl EasyTierLauncher {
|
||||
#[cfg(mobile)]
|
||||
Self::run_routine_for_mobile(&instance, &data, &mut tasks).await;
|
||||
|
||||
instance.run().await?;
|
||||
if let Err(err) = instance.run().await {
|
||||
tasks.abort_all();
|
||||
drop(tasks);
|
||||
instance.clear_resources().await;
|
||||
return Err(err.into());
|
||||
}
|
||||
|
||||
#[cfg(feature = "ffi-dataplane")]
|
||||
data.data_plane
|
||||
|
||||
+7
-10
@@ -13,21 +13,18 @@ use clap_complete::{Generator, Shell};
|
||||
extern crate self as hotpath;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
mod hotpath_off;
|
||||
|
||||
// When the `hotpath` feature is off, expose a local `instant` module backed by
|
||||
// `quanta::Instant` so call sites can uniformly write `use hotpath::instant::Instant;`
|
||||
// regardless of whether the feature is enabled. With the feature on, the real
|
||||
// `hotpath` crate provides the same path (also `quanta::Instant` on Linux), so
|
||||
// the two modes resolve to the identical type.
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
pub mod instant {
|
||||
pub type Instant = quanta::Instant;
|
||||
}
|
||||
pub(crate) use hotpath_off::wrap;
|
||||
|
||||
// `hotpath-alloc` registers a global profiling allocator, which is mutually
|
||||
// exclusive with the `jemalloc`/`mimalloc` global allocators.
|
||||
#[cfg(all(feature = "hotpath-alloc", any(feature = "jemalloc", feature = "mimalloc")))]
|
||||
compile_error!("feature `hotpath-alloc` cannot be enabled together with `jemalloc` or `mimalloc`");
|
||||
|
||||
// Re-export `Instant` at the crate root so public APIs that expose it
|
||||
// (e.g. `Route::get_peer_info_last_update_time`) reference a deliberate
|
||||
// public type rather than leaking an inaccessible one.
|
||||
pub use hotpath::instant::Instant;
|
||||
pub use quanta::Instant;
|
||||
|
||||
mod arch;
|
||||
mod gateway;
|
||||
|
||||
@@ -7,11 +7,11 @@ use std::{
|
||||
|
||||
use arc_swap::ArcSwap;
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use pnet::packet::ipv6::Ipv6Packet;
|
||||
use pnet::packet::{
|
||||
Packet as _, ip::IpNextHeaderProtocols, ipv4::Ipv4Packet, tcp::TcpPacket, udp::UdpPacket,
|
||||
};
|
||||
use quanta::Instant;
|
||||
|
||||
use crate::proto::acl::{AclStats, Protocol};
|
||||
use crate::tunnel::packet_def::PacketType;
|
||||
@@ -404,7 +404,7 @@ mod tests {
|
||||
sync::Arc,
|
||||
};
|
||||
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
|
||||
use crate::{
|
||||
common::acl_processor::PacketInfo,
|
||||
|
||||
@@ -56,11 +56,12 @@ type BoxNicPacketFilter = Box<dyn NicPacketFilter + Send + Sync>;
|
||||
// pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
|
||||
// tachyonix::channel(128)
|
||||
// }
|
||||
pub type PacketRecvChan = tokio::sync::mpsc::Sender<ZCPacket>;
|
||||
pub type PacketRecvChanReceiver = tokio::sync::mpsc::Receiver<ZCPacket>;
|
||||
pub type PacketRecvChan = hotpath::wrap::tokio::sync::mpsc::Sender<ZCPacket>;
|
||||
pub type PacketRecvChanReceiver = hotpath::wrap::tokio::sync::mpsc::Receiver<ZCPacket>;
|
||||
pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
|
||||
hotpath::channel!(tokio::sync::mpsc::channel(128))
|
||||
}
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure())]
|
||||
pub async fn recv_packet_from_chan(
|
||||
packet_recv_chan_receiver: &mut PacketRecvChanReceiver,
|
||||
) -> Result<ZCPacket, anyhow::Error> {
|
||||
|
||||
@@ -2,6 +2,9 @@ use std::sync::Arc;
|
||||
|
||||
use crossbeam::atomic::AtomicCell;
|
||||
use dashmap::{DashMap, DashSet};
|
||||
#[cfg(feature = "hotpath")]
|
||||
use hotpath::wrap::parking_lot::RwLock;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
use parking_lot::RwLock;
|
||||
|
||||
use tokio::{select, sync::mpsc};
|
||||
@@ -56,7 +59,7 @@ impl Peer {
|
||||
let shutdown_notifier = Arc::new(tokio::sync::Notify::new());
|
||||
let peer_identity_type = Arc::new(AtomicCell::new(None));
|
||||
let peer_identity_type_copy = peer_identity_type.clone();
|
||||
let peer_public_key = Arc::new(RwLock::new(None));
|
||||
let peer_public_key = Arc::new(hotpath::rw_lock!(parking_lot::RwLock::new(None)));
|
||||
let peer_public_key_copy = peer_public_key.clone();
|
||||
|
||||
let conns_copy = conns.clone();
|
||||
|
||||
@@ -384,10 +384,10 @@ impl PeerConn {
|
||||
session_filter,
|
||||
noise_handshake_result: None,
|
||||
|
||||
tunnel: Arc::new(hotpath::mutex!(tokio::sync::Mutex::new(Box::new(
|
||||
guard!([mut mpsc_tunnel] mpsc_tunnel.close()),
|
||||
)
|
||||
as Box<dyn Any + Send + 'static>))),
|
||||
tunnel: Arc::new(hotpath::mutex!(tokio::sync::Mutex::new(
|
||||
Box::new(guard!([mut mpsc_tunnel] mpsc_tunnel.close()))
|
||||
as Box<dyn Any + Send + 'static>,
|
||||
))),
|
||||
sink,
|
||||
recv: hotpath::mutex!(tokio::sync::Mutex::new(Some(recv))),
|
||||
tunnel_info,
|
||||
|
||||
@@ -6,7 +6,7 @@ use std::{
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use rand::{Rng, thread_rng};
|
||||
use tokio::{
|
||||
sync::broadcast,
|
||||
|
||||
@@ -2,7 +2,7 @@ use anyhow::Context;
|
||||
use async_trait::async_trait;
|
||||
use cidr::{Ipv4Cidr, Ipv6Cidr};
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use std::collections::BTreeSet;
|
||||
use std::{
|
||||
fmt::Debug,
|
||||
@@ -963,6 +963,7 @@ impl PeerManager {
|
||||
Self::is_relay_data_packet(hdr.packet_type)
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
|
||||
async fn start_peer_recv(&self) {
|
||||
let mut recv = self.packet_recv.lock().await.take().unwrap();
|
||||
let my_peer_id = self.my_peer_id;
|
||||
@@ -1697,6 +1698,7 @@ impl PeerManager {
|
||||
(dst_peers, is_exit_node)
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
|
||||
pub async fn try_compress_and_encrypt(
|
||||
compress_algo: CompressorAlgo,
|
||||
encryptor: &Arc<dyn Encryptor + 'static>,
|
||||
@@ -2207,7 +2209,7 @@ mod tests {
|
||||
use base64::Engine;
|
||||
use std::{collections::HashMap, fmt::Debug, sync::Arc, time::Duration};
|
||||
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
|
||||
use crate::{
|
||||
common::{
|
||||
|
||||
@@ -6,6 +6,9 @@ use std::{
|
||||
use anyhow::Context;
|
||||
use dashmap::{DashMap, DashSet};
|
||||
use parking_lot::Mutex;
|
||||
#[cfg(feature = "hotpath")]
|
||||
use hotpath::wrap::tokio::sync::RwLock;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
use crate::{
|
||||
@@ -45,7 +48,7 @@ impl PeerMap {
|
||||
my_peer_id,
|
||||
peer_map: DashMap::new(),
|
||||
packet_send,
|
||||
routes: RwLock::new(Vec::new()),
|
||||
routes: hotpath::rw_lock!(tokio::sync::RwLock::new(Vec::new())),
|
||||
alive_client_urls: Arc::new(Mutex::new(multimap::MultiMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,7 +13,6 @@ use arc_swap::ArcSwap;
|
||||
use cidr::{IpCidr, Ipv4Cidr, Ipv6Cidr, Ipv6Inet};
|
||||
use crossbeam::atomic::AtomicCell;
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use ordered_hash_map::OrderedHashMap;
|
||||
use parking_lot::{RwLock, lock_api::RwLockUpgradableReadGuard};
|
||||
use petgraph::{
|
||||
@@ -25,6 +24,7 @@ use petgraph::{
|
||||
use prefix_trie::PrefixMap;
|
||||
use prost::Message;
|
||||
use prost_reflect::{DynamicMessage, ReflectMessage};
|
||||
use quanta::Instant;
|
||||
use tokio::{
|
||||
select,
|
||||
sync::Mutex,
|
||||
|
||||
@@ -1,13 +1,17 @@
|
||||
use std::sync::{
|
||||
Arc, RwLock,
|
||||
Arc,
|
||||
atomic::{AtomicBool, Ordering},
|
||||
};
|
||||
#[cfg(feature = "hotpath")]
|
||||
use hotpath::wrap::std::sync::RwLock;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
use std::sync::RwLock;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::anyhow;
|
||||
use crossbeam::atomic::AtomicCell;
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
|
||||
use super::secure_datagram::{SecureDatagramDirection, SecureDatagramSession};
|
||||
use crate::{
|
||||
@@ -262,7 +266,7 @@ impl std::fmt::Debug for PeerSession {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("PeerSession")
|
||||
.field("peer_id", &self.peer_id)
|
||||
.field("peer_static_pubkey", &self.peer_static_pubkey)
|
||||
.field("peer_static_pubkey", &*self.peer_static_pubkey.read().unwrap())
|
||||
.field("datagram", &self.datagram)
|
||||
.finish()
|
||||
}
|
||||
@@ -282,7 +286,7 @@ impl PeerSession {
|
||||
) -> Self {
|
||||
Self {
|
||||
peer_id,
|
||||
peer_static_pubkey: RwLock::new(peer_static_pubkey),
|
||||
peer_static_pubkey: hotpath::rw_lock!(std::sync::RwLock::new(peer_static_pubkey)),
|
||||
datagram: SecureDatagramSession::new(
|
||||
root_key,
|
||||
session_generation,
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use prost::Message;
|
||||
use quanta::Instant;
|
||||
use snow::params::NoiseParams;
|
||||
use tokio::sync::{Mutex, OwnedMutexGuard, oneshot};
|
||||
use tokio::time::{Duration, timeout};
|
||||
@@ -144,6 +144,7 @@ impl RelayPeerMap {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
|
||||
async fn send_via_next_hop(
|
||||
&self,
|
||||
msg: ZCPacket,
|
||||
@@ -166,6 +167,7 @@ impl RelayPeerMap {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
|
||||
pub async fn send_msg(
|
||||
self: &Arc<Self>,
|
||||
mut msg: ZCPacket,
|
||||
@@ -613,6 +615,7 @@ impl RelayPeerMap {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
|
||||
pub async fn decrypt_if_needed(self: &Arc<Self>, packet: &mut ZCPacket) -> Result<bool, Error> {
|
||||
if !self.is_secure_mode_enabled() {
|
||||
return Ok(false);
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
use cidr::Ipv6Inet;
|
||||
use cidr::{Ipv4Cidr, Ipv6Cidr};
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
use std::{
|
||||
collections::BTreeSet,
|
||||
net::{Ipv4Addr, Ipv6Addr},
|
||||
|
||||
@@ -1,11 +1,16 @@
|
||||
use std::{
|
||||
sync::{
|
||||
Arc, Mutex, RwLock,
|
||||
atomic::{AtomicBool, AtomicU32, Ordering},
|
||||
Arc,
|
||||
},
|
||||
time::{SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
#[cfg(feature = "hotpath")]
|
||||
use hotpath::wrap::std::sync::{Mutex, RwLock};
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
use std::sync::{Mutex, RwLock};
|
||||
|
||||
use anyhow::anyhow;
|
||||
use atomic_shim::AtomicU64;
|
||||
use hmac::{Hmac, Mac as _};
|
||||
@@ -14,7 +19,7 @@ use sha2::Sha256;
|
||||
use zerocopy::FromBytes;
|
||||
|
||||
use crate::{
|
||||
peers::encrypt::{Encryptor, create_encryptor},
|
||||
peers::encrypt::{create_encryptor, Encryptor},
|
||||
tunnel::packet_def::{StandardAeadTail, ZCPacket},
|
||||
};
|
||||
|
||||
@@ -228,15 +233,15 @@ pub struct SecureDatagramSession {
|
||||
impl std::fmt::Debug for SecureDatagramSession {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("SecureDatagramSession")
|
||||
.field("root_key", &self.root_key)
|
||||
.field("root_key", &*self.root_key.read().unwrap())
|
||||
.field("session_generation", &self.session_generation)
|
||||
.field("send_epoch", &self.send_epoch)
|
||||
.field("send_seq", &self.send_seq)
|
||||
.field("send_epoch_started_ms", &self.send_epoch_started_ms)
|
||||
.field("send_packets_since_epoch", &self.send_packets_since_epoch)
|
||||
.field("rx_slots", &self.rx_slots)
|
||||
.field("key_cache", &self.key_cache)
|
||||
.field("sync_rx_grace", &self.sync_rx_grace)
|
||||
.field("rx_slots", &*self.rx_slots.lock().unwrap())
|
||||
.field("key_cache", &*self.key_cache.lock().unwrap())
|
||||
.field("sync_rx_grace", &*self.sync_rx_grace.lock().unwrap())
|
||||
.field(
|
||||
"sync_rx_grace_expires_at_ms",
|
||||
&self.sync_rx_grace_expires_at_ms,
|
||||
@@ -272,15 +277,15 @@ impl SecureDatagramSession {
|
||||
];
|
||||
let now_ms = now_ms();
|
||||
Self {
|
||||
root_key: RwLock::new(root_key),
|
||||
root_key: hotpath::rw_lock!(std::sync::RwLock::new(root_key)),
|
||||
session_generation: AtomicU32::new(session_generation),
|
||||
send_epoch: AtomicU32::new(initial_epoch),
|
||||
send_seq: [AtomicU64::new(0), AtomicU64::new(0)],
|
||||
send_epoch_started_ms: AtomicU64::new(now_ms),
|
||||
send_packets_since_epoch: AtomicU64::new(0),
|
||||
rx_slots: Mutex::new(rx_slots),
|
||||
key_cache: Mutex::new(key_cache),
|
||||
sync_rx_grace: Mutex::new(SyncRxGrace::default()),
|
||||
rx_slots: hotpath::mutex!(std::sync::Mutex::new(rx_slots)),
|
||||
key_cache: hotpath::mutex!(std::sync::Mutex::new(key_cache)),
|
||||
sync_rx_grace: hotpath::mutex!(std::sync::Mutex::new(SyncRxGrace::default())),
|
||||
sync_rx_grace_expires_at_ms: AtomicU64::new(0),
|
||||
send_cipher_algorithm,
|
||||
recv_cipher_algorithm,
|
||||
@@ -892,11 +897,9 @@ mod tests {
|
||||
let nonce_offset = payload.len() - StandardAeadTail::NONCE_SIZE;
|
||||
payload[nonce_offset..].copy_from_slice(&poisoned_nonce);
|
||||
|
||||
assert!(
|
||||
receiver
|
||||
.decrypt_payload(SecureDatagramDirection::AToB, &mut forged)
|
||||
.is_err()
|
||||
);
|
||||
assert!(receiver
|
||||
.decrypt_payload(SecureDatagramDirection::AToB, &mut forged)
|
||||
.is_err());
|
||||
|
||||
let plaintext = b"pkt2";
|
||||
let mut pkt2 = ZCPacket::new_with_payload(plaintext);
|
||||
|
||||
@@ -5,8 +5,8 @@ use std::sync::{Arc, Mutex};
|
||||
use bytes::Bytes;
|
||||
use dashmap::DashMap;
|
||||
use guarden::defer;
|
||||
use hotpath::instant::Instant;
|
||||
use prost::Message;
|
||||
use quanta::Instant;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::task::JoinSet;
|
||||
use tokio::time::timeout;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use prost::{Message as _, length_delimiter_len};
|
||||
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
|
||||
use crate::{
|
||||
common::{PeerId, compressor::DefaultCompressor},
|
||||
|
||||
@@ -5,8 +5,8 @@ use std::{
|
||||
|
||||
use bytes::Bytes;
|
||||
use dashmap::DashMap;
|
||||
use hotpath::instant::Instant;
|
||||
use prost::Message;
|
||||
use quanta::Instant;
|
||||
use tokio::{task::JoinSet, time::timeout};
|
||||
use tokio_stream::StreamExt;
|
||||
|
||||
|
||||
@@ -49,7 +49,7 @@ async fn test_route_peer_info_ipv6() {
|
||||
#[tokio::test]
|
||||
async fn test_peer_manager_ipv6() {
|
||||
let global_ctx = get_mock_global_ctx();
|
||||
let (packet_sender, _packet_receiver) = tokio::sync::mpsc::channel(100);
|
||||
let (packet_sender, _packet_receiver) = crate::peers::create_packet_recv_chan();
|
||||
let peer_mgr = crate::peers::peer_manager::PeerManager::new(
|
||||
RouteAlgoType::Ospf,
|
||||
global_ctx.clone(),
|
||||
|
||||
@@ -477,6 +477,12 @@ struct PublicIpv6Lab {
|
||||
extra_bridges: [&'static str; 2],
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum PublicIpv6LabTopology {
|
||||
DelegatedPrefix,
|
||||
OnLinkPrefix,
|
||||
}
|
||||
|
||||
impl PublicIpv6Lab {
|
||||
const PROVIDER_NS: &'static str = "net_a";
|
||||
const CLIENT_NS: &'static str = "net_b";
|
||||
@@ -490,11 +496,13 @@ impl PublicIpv6Lab {
|
||||
const PROVIDER_DEFAULT_FROM: &'static str = "2001:db8:100::/64";
|
||||
const PROVIDER_WAN_ADDR: &'static str = "2001:db8:ffff:1::2/64";
|
||||
const UPSTREAM_WAN_ADDR: &'static str = "2001:db8:ffff:1::1/64";
|
||||
const ON_LINK_PROVIDER_WAN_ADDR: &'static str = "2001:db8:100::2/64";
|
||||
const ON_LINK_UPSTREAM_WAN_ADDR: &'static str = "2001:db8:100::1/64";
|
||||
const UPSTREAM_SERVER_ADDR: &'static str = "2001:db8:ffff:2::1/64";
|
||||
const SERVER_ADDR: &'static str = "2001:db8:ffff:2::100/64";
|
||||
const SERVER_IP: &'static str = "2001:db8:ffff:2::100";
|
||||
|
||||
fn setup() -> Self {
|
||||
fn setup_with_topology(topology: PublicIpv6LabTopology) -> Self {
|
||||
prepare_linux_namespaces();
|
||||
|
||||
del_netns(Self::UPSTREAM_NS);
|
||||
@@ -544,13 +552,23 @@ impl PublicIpv6Lab {
|
||||
Self::SERVER_BRIDGE,
|
||||
);
|
||||
|
||||
let (provider_wan_addr, upstream_wan_addr) = match topology {
|
||||
PublicIpv6LabTopology::DelegatedPrefix => {
|
||||
(Self::PROVIDER_WAN_ADDR, Self::UPSTREAM_WAN_ADDR)
|
||||
}
|
||||
PublicIpv6LabTopology::OnLinkPrefix => (
|
||||
Self::ON_LINK_PROVIDER_WAN_ADDR,
|
||||
Self::ON_LINK_UPSTREAM_WAN_ADDR,
|
||||
),
|
||||
};
|
||||
|
||||
run_ip_in_ns(
|
||||
Self::PROVIDER_NS,
|
||||
&["addr", "add", Self::PROVIDER_WAN_ADDR, "dev", "pubwan0"],
|
||||
&["addr", "add", provider_wan_addr, "dev", "pubwan0"],
|
||||
);
|
||||
run_ip_in_ns(
|
||||
Self::UPSTREAM_NS,
|
||||
&["addr", "add", Self::UPSTREAM_WAN_ADDR, "dev", "upwan0"],
|
||||
&["addr", "add", upstream_wan_addr, "dev", "upwan0"],
|
||||
);
|
||||
run_ip_in_ns(
|
||||
Self::UPSTREAM_NS,
|
||||
@@ -561,37 +579,56 @@ impl PublicIpv6Lab {
|
||||
&["addr", "add", Self::SERVER_ADDR, "dev", "srv0"],
|
||||
);
|
||||
|
||||
run_ip_in_ns(
|
||||
Self::PROVIDER_NS,
|
||||
&["link", "add", "pubprefix0", "type", "dummy"],
|
||||
);
|
||||
run_ip_in_ns(Self::PROVIDER_NS, &["link", "set", "pubprefix0", "up"]);
|
||||
run_ip_in_ns(
|
||||
Self::PROVIDER_NS,
|
||||
&[
|
||||
"-6",
|
||||
"route",
|
||||
"add",
|
||||
Self::PROVIDER_PREFIX,
|
||||
"dev",
|
||||
"pubprefix0",
|
||||
],
|
||||
);
|
||||
run_ip_in_ns(
|
||||
Self::PROVIDER_NS,
|
||||
&[
|
||||
"-6",
|
||||
"route",
|
||||
"add",
|
||||
"default",
|
||||
"from",
|
||||
Self::PROVIDER_DEFAULT_FROM,
|
||||
"via",
|
||||
"2001:db8:ffff:1::1",
|
||||
"dev",
|
||||
"pubwan0",
|
||||
],
|
||||
);
|
||||
match topology {
|
||||
PublicIpv6LabTopology::DelegatedPrefix => {
|
||||
run_ip_in_ns(
|
||||
Self::PROVIDER_NS,
|
||||
&["link", "add", "pubprefix0", "type", "dummy"],
|
||||
);
|
||||
run_ip_in_ns(Self::PROVIDER_NS, &["link", "set", "pubprefix0", "up"]);
|
||||
run_ip_in_ns(
|
||||
Self::PROVIDER_NS,
|
||||
&[
|
||||
"-6",
|
||||
"route",
|
||||
"add",
|
||||
Self::PROVIDER_PREFIX,
|
||||
"dev",
|
||||
"pubprefix0",
|
||||
],
|
||||
);
|
||||
run_ip_in_ns(
|
||||
Self::PROVIDER_NS,
|
||||
&[
|
||||
"-6",
|
||||
"route",
|
||||
"add",
|
||||
"default",
|
||||
"from",
|
||||
Self::PROVIDER_DEFAULT_FROM,
|
||||
"via",
|
||||
"2001:db8:ffff:1::1",
|
||||
"dev",
|
||||
"pubwan0",
|
||||
],
|
||||
);
|
||||
}
|
||||
PublicIpv6LabTopology::OnLinkPrefix => {
|
||||
run_ip_in_ns(
|
||||
Self::PROVIDER_NS,
|
||||
&[
|
||||
"-6",
|
||||
"route",
|
||||
"add",
|
||||
"default",
|
||||
"via",
|
||||
"2001:db8:100::1",
|
||||
"dev",
|
||||
"pubwan0",
|
||||
],
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
run_ip_in_ns(
|
||||
Self::SERVER_NS,
|
||||
@@ -606,19 +643,21 @@ impl PublicIpv6Lab {
|
||||
"srv0",
|
||||
],
|
||||
);
|
||||
run_ip_in_ns(
|
||||
Self::UPSTREAM_NS,
|
||||
&[
|
||||
"-6",
|
||||
"route",
|
||||
"add",
|
||||
Self::PROVIDER_PREFIX,
|
||||
"via",
|
||||
"2001:db8:ffff:1::2",
|
||||
"dev",
|
||||
"upwan0",
|
||||
],
|
||||
);
|
||||
if matches!(topology, PublicIpv6LabTopology::DelegatedPrefix) {
|
||||
run_ip_in_ns(
|
||||
Self::UPSTREAM_NS,
|
||||
&[
|
||||
"-6",
|
||||
"route",
|
||||
"add",
|
||||
Self::PROVIDER_PREFIX,
|
||||
"via",
|
||||
"2001:db8:ffff:1::2",
|
||||
"dev",
|
||||
"upwan0",
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
run_sysctl_in_ns(Self::PROVIDER_NS, "net.ipv6.conf.all.forwarding=1");
|
||||
run_sysctl_in_ns(Self::UPSTREAM_NS, "net.ipv6.conf.all.forwarding=1");
|
||||
@@ -672,7 +711,15 @@ fn get_public_ipv6_config(
|
||||
async fn init_public_ipv6_two_node(
|
||||
client_inst_id: uuid::Uuid,
|
||||
) -> (PublicIpv6Lab, Instance, Instance) {
|
||||
let lab = PublicIpv6Lab::setup();
|
||||
init_public_ipv6_two_node_with_topology(client_inst_id, PublicIpv6LabTopology::DelegatedPrefix)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn init_public_ipv6_two_node_with_topology(
|
||||
client_inst_id: uuid::Uuid,
|
||||
topology: PublicIpv6LabTopology,
|
||||
) -> (PublicIpv6Lab, Instance, Instance) {
|
||||
let lab = PublicIpv6Lab::setup_with_topology(topology);
|
||||
|
||||
let provider_cfg = get_public_ipv6_config(
|
||||
"provider_public_ipv6",
|
||||
@@ -756,6 +803,13 @@ fn addr_exists_in_ns(ns: &str, dev: &str, needle: &str) -> bool {
|
||||
run_ip_in_ns_output(ns, &["-6", "addr", "show", "dev", dev]).contains(needle)
|
||||
}
|
||||
|
||||
fn ndp_proxy_exists_in_ns(ns: &str, dev: &str, addr: std::net::Ipv6Addr) -> bool {
|
||||
let addr = addr.to_string();
|
||||
run_ip_in_ns_output(ns, &["-6", "neigh", "show", "proxy", "dev", dev])
|
||||
.lines()
|
||||
.any(|line| line.split_whitespace().next() == Some(addr.as_str()))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
pub async fn public_ipv6_auto_addr_end_to_end() {
|
||||
@@ -878,6 +932,67 @@ pub async fn public_ipv6_auto_addr_end_to_end() {
|
||||
drop_insts(vec![provider, client]).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
pub async fn public_ipv6_auto_addr_on_link_ndp_proxy_end_to_end() {
|
||||
let client_id = uuid::Uuid::parse_str("44444444-4444-4444-4444-444444444444").unwrap();
|
||||
let (_lab, provider, client) =
|
||||
init_public_ipv6_two_node_with_topology(client_id, PublicIpv6LabTopology::OnLinkPrefix)
|
||||
.await;
|
||||
|
||||
wait_for_condition(
|
||||
|| async {
|
||||
provider
|
||||
.get_global_ctx()
|
||||
.get_advertised_ipv6_public_addr_prefix()
|
||||
== Some(PublicIpv6Lab::PROVIDER_PREFIX.parse().unwrap())
|
||||
},
|
||||
Duration::from_secs(10),
|
||||
)
|
||||
.await;
|
||||
|
||||
let leased = wait_for_public_ipv6_addr(&client).await;
|
||||
wait_for_public_ipv6_route(&provider, leased).await;
|
||||
|
||||
wait_for_condition(
|
||||
|| async {
|
||||
addr_exists_in_ns(
|
||||
PublicIpv6Lab::CLIENT_NS,
|
||||
PublicIpv6Lab::CLIENT_TUN,
|
||||
&leased.to_string(),
|
||||
) && route_exists_in_ns(
|
||||
PublicIpv6Lab::PROVIDER_NS,
|
||||
&format!("{} dev {}", leased.address(), PublicIpv6Lab::PROVIDER_TUN),
|
||||
)
|
||||
},
|
||||
Duration::from_secs(10),
|
||||
)
|
||||
.await;
|
||||
|
||||
wait_for_condition(
|
||||
|| async {
|
||||
ndp_proxy_exists_in_ns(PublicIpv6Lab::PROVIDER_NS, "pubwan0", leased.address())
|
||||
},
|
||||
Duration::from_secs(20),
|
||||
)
|
||||
.await;
|
||||
|
||||
wait_for_condition(
|
||||
|| async {
|
||||
ping6_test(
|
||||
PublicIpv6Lab::SERVER_NS,
|
||||
leased.address().to_string().as_str(),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
},
|
||||
Duration::from_secs(20),
|
||||
)
|
||||
.await;
|
||||
|
||||
drop_insts(vec![provider, client]).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
pub async fn public_ipv6_auto_addr_reconnect_reuses_same_address() {
|
||||
|
||||
@@ -48,9 +48,13 @@ use std::collections::{HashMap, HashSet};
|
||||
use std::fmt;
|
||||
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
|
||||
use std::sync::{
|
||||
Arc, RwLock,
|
||||
Arc,
|
||||
atomic::{AtomicU32, Ordering},
|
||||
};
|
||||
#[cfg(feature = "hotpath")]
|
||||
use hotpath::wrap::std::sync::RwLock;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
use std::sync::RwLock;
|
||||
use tokio::sync::broadcast;
|
||||
use tokio::time;
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
@@ -85,7 +89,7 @@ impl AddrTuple {
|
||||
|
||||
#[derive(Default)]
|
||||
struct StackState {
|
||||
tuples: HashMap<AddrTuple, flume::Sender<Bytes>>,
|
||||
tuples: HashMap<AddrTuple, hotpath::wrap::flume::Sender<Bytes>>,
|
||||
closed: bool,
|
||||
}
|
||||
|
||||
@@ -129,7 +133,7 @@ pub enum State {
|
||||
pub struct Socket {
|
||||
shared: Arc<Shared>,
|
||||
tun: Arc<dyn Tun>,
|
||||
incoming: flume::Receiver<Bytes>,
|
||||
incoming: hotpath::wrap::flume::Receiver<Bytes>,
|
||||
local_addr: SocketAddr,
|
||||
remote_addr: SocketAddr,
|
||||
local_mac: MacAddr,
|
||||
@@ -158,7 +162,7 @@ impl Socket {
|
||||
remote_mac: Option<MacAddr>,
|
||||
ack: Option<u32>,
|
||||
state: State,
|
||||
) -> (Socket, flume::Sender<Bytes>) {
|
||||
) -> (Socket, hotpath::wrap::flume::Sender<Bytes>) {
|
||||
let (incoming_tx, incoming_rx) = hotpath::channel!(flume::bounded(MPMC_BUFFER_LEN));
|
||||
|
||||
(
|
||||
@@ -430,9 +434,9 @@ impl Stack {
|
||||
) -> Stack {
|
||||
let (tuples_purge_tx, _tuples_purge_rx) = broadcast::channel(16);
|
||||
let shared = Arc::new(Shared {
|
||||
state: RwLock::new(StackState::default()),
|
||||
state: hotpath::rw_lock!(std::sync::RwLock::new(StackState::default())),
|
||||
tun: tun.clone(),
|
||||
listening: RwLock::new(HashSet::new()),
|
||||
listening: hotpath::rw_lock!(std::sync::RwLock::new(HashSet::new())),
|
||||
tuples_purge: tuples_purge_tx.clone(),
|
||||
});
|
||||
|
||||
@@ -501,7 +505,7 @@ impl Stack {
|
||||
shared: Arc<Shared>,
|
||||
mut tuples_purge: broadcast::Receiver<AddrTuple>,
|
||||
) {
|
||||
let mut tuples: HashMap<AddrTuple, flume::Sender<Bytes>> = HashMap::new();
|
||||
let mut tuples: HashMap<AddrTuple, hotpath::wrap::flume::Sender<Bytes>> = HashMap::new();
|
||||
|
||||
loop {
|
||||
let mut buf = BytesMut::new();
|
||||
|
||||
@@ -9,7 +9,8 @@ use crate::proto::common::TunnelInfo;
|
||||
|
||||
use super::{Tunnel, TunnelError, ZCPacketSink, ZCPacketStream, packet_def::ZCPacket};
|
||||
|
||||
use tokio::sync::mpsc::{Receiver, Sender, channel, error::TrySendError};
|
||||
use hotpath::wrap::tokio::sync::mpsc::{Receiver, Sender};
|
||||
use tokio::sync::mpsc::{channel, error::TrySendError};
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
// use tachyonix::{channel, Receiver, Sender, TrySendError};
|
||||
|
||||
@@ -19,6 +20,7 @@ use futures::SinkExt;
|
||||
pub struct MpscTunnelSender(Sender<ZCPacket>);
|
||||
|
||||
impl MpscTunnelSender {
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnelSender"))]
|
||||
pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
|
||||
self.0.send(item).await.with_context(|| "send error")?;
|
||||
Ok(())
|
||||
|
||||
@@ -12,6 +12,9 @@ use crate::tunnel::{
|
||||
use anyhow::Context;
|
||||
use derivative::Derivative;
|
||||
use derive_more::{Deref, DerefMut};
|
||||
#[cfg(feature = "hotpath")]
|
||||
use hotpath::wrap::parking_lot::RwLock;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
use parking_lot::RwLock;
|
||||
use quinn::{
|
||||
ClientConfig, ConnectError, Connection, Endpoint, EndpointConfig, ServerConfig,
|
||||
@@ -312,18 +315,25 @@ struct RwPoolInner<Item> {
|
||||
enabled: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct RwPool<Item> {
|
||||
ephemeral: RwLock<RwPoolInner<Item>>,
|
||||
persistent: RwLock<RwPoolInner<Item>>,
|
||||
capacity: usize,
|
||||
}
|
||||
|
||||
impl<Item> std::fmt::Debug for RwPool<Item> {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("RwPool")
|
||||
.field("capacity", &self.capacity)
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
impl<Item> RwPool<Item> {
|
||||
fn new(capacity: usize) -> Self {
|
||||
Self {
|
||||
ephemeral: RwLock::new(RwPoolInner::default()),
|
||||
persistent: RwLock::new(RwPoolInner::default()),
|
||||
ephemeral: hotpath::rw_lock!(parking_lot::RwLock::new(RwPoolInner::default())),
|
||||
persistent: hotpath::rw_lock!(parking_lot::RwLock::new(RwPoolInner::default())),
|
||||
capacity,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,7 +11,8 @@ use async_trait::async_trait;
|
||||
use futures::{Sink, SinkExt, Stream, StreamExt};
|
||||
use once_cell::sync::Lazy;
|
||||
|
||||
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
|
||||
use hotpath::wrap::tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
|
||||
use tokio::sync::mpsc::unbounded_channel;
|
||||
|
||||
use uuid::Uuid;
|
||||
|
||||
@@ -196,8 +197,7 @@ pub struct RingTunnelListener {
|
||||
|
||||
impl RingTunnelListener {
|
||||
pub fn new(key: url::Url) -> Self {
|
||||
let (conn_sender, conn_receiver) =
|
||||
hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
|
||||
let (conn_sender, conn_receiver) = hotpath::channel!(unbounded_channel());
|
||||
RingTunnelListener {
|
||||
listener_addr: key,
|
||||
conn_sender,
|
||||
|
||||
@@ -13,9 +13,10 @@ use futures::{StreamExt, stream::FuturesUnordered};
|
||||
use rand::{Rng, SeedableRng};
|
||||
use zerocopy::{AsBytes, FromBytes};
|
||||
|
||||
use hotpath::wrap::tokio::sync::mpsc::{Receiver, Sender, UnboundedReceiver, UnboundedSender};
|
||||
use tokio::{
|
||||
net::UdpSocket,
|
||||
sync::mpsc::{Receiver, Sender, UnboundedReceiver, UnboundedSender},
|
||||
sync::mpsc::{channel, unbounded_channel},
|
||||
task::JoinSet,
|
||||
};
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
@@ -291,6 +292,7 @@ fn get_zcpacket_from_buf(buf: BytesMut, allow_stun: bool) -> Result<ZCPacket, Tu
|
||||
}
|
||||
|
||||
#[instrument]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure())]
|
||||
async fn forward_from_ring_to_udp(
|
||||
mut ring_recv: RingStream,
|
||||
socket: &Arc<UdpSocket>,
|
||||
@@ -325,6 +327,7 @@ async fn forward_from_ring_to_udp(
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure())]
|
||||
async fn udp_recv_from_socket_forward_task(
|
||||
socket: &UdpSocket,
|
||||
buf: &mut BytesMut,
|
||||
@@ -393,6 +396,7 @@ impl UdpConnection {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "UdpTunnel"))]
|
||||
pub fn handle_packet_from_remote(&mut self, zc_packet: ZCPacket) -> Result<(), TunnelError> {
|
||||
let header = zc_packet.udp_tunnel_header().unwrap();
|
||||
let conn_id = header.conn_id.get();
|
||||
@@ -539,6 +543,7 @@ impl UdpTunnelListenerData {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "UdpTunnelListener"))]
|
||||
fn do_forward_one_packet_to_conn(&self, zc_packet: ZCPacket, addr: SocketAddr) {
|
||||
let header = zc_packet.udp_tunnel_header().unwrap();
|
||||
if header.msg_type == UdpPacketType::Syn as u8 {
|
||||
@@ -645,6 +650,7 @@ impl UdpTunnelListenerData {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "UdpTunnelListener"))]
|
||||
async fn do_forward_task(self) {
|
||||
let socket = self.socket.as_ref().unwrap().clone();
|
||||
let mut buf = BytesMut::new();
|
||||
@@ -673,9 +679,8 @@ pub struct UdpTunnelListener {
|
||||
|
||||
impl UdpTunnelListener {
|
||||
pub fn new(addr: url::Url) -> Self {
|
||||
let (close_event_send, close_event_recv) =
|
||||
hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
|
||||
let (conn_send, conn_recv) = hotpath::channel!(tokio::sync::mpsc::channel(100));
|
||||
let (close_event_send, close_event_recv) = hotpath::channel!(unbounded_channel());
|
||||
let (conn_send, conn_recv) = hotpath::channel!(channel(100));
|
||||
Self {
|
||||
addr: addr.clone(),
|
||||
socket: None,
|
||||
@@ -916,7 +921,7 @@ impl UdpTunnelConnector {
|
||||
);
|
||||
|
||||
let (close_event_sender, mut close_event_recv) =
|
||||
hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
|
||||
hotpath::channel!(unbounded_channel());
|
||||
|
||||
let ring_recv = RingStream::new(ring_for_send_udp.clone());
|
||||
let ring_sender = RingSink::new(ring_for_recv_udp.clone());
|
||||
@@ -1179,7 +1184,7 @@ mod tests {
|
||||
let dst_addr = "127.0.0.1:1".parse().unwrap();
|
||||
let ring_for_send_udp = Arc::new(RingTunnel::new(8));
|
||||
let ring_for_recv_udp = Arc::new(RingTunnel::new(8));
|
||||
let (close_event_sender, _close_event_recv) = tokio::sync::mpsc::unbounded_channel();
|
||||
let (close_event_sender, _close_event_recv) = hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
|
||||
let mut conn = UdpConnection::new(
|
||||
socket,
|
||||
7,
|
||||
|
||||
@@ -6,7 +6,7 @@ use std::{
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use hotpath::instant::Instant;
|
||||
use quanta::Instant;
|
||||
|
||||
use super::{
|
||||
FromUrl, IpVersion, Tunnel, TunnelError, TunnelInfo, TunnelListener, TunnelUrl, ZCPacketSink,
|
||||
@@ -37,7 +37,19 @@ use crossbeam::atomic::AtomicCell;
|
||||
use dashmap::DashMap;
|
||||
use futures::{SinkExt, StreamExt, stream::FuturesUnordered};
|
||||
use rand::RngCore;
|
||||
use tokio::{net::UdpSocket, sync::Mutex, task::JoinSet};
|
||||
#[cfg(feature = "hotpath")]
|
||||
use hotpath::wrap::std::sync::Mutex as StdMutex;
|
||||
#[cfg(feature = "hotpath")]
|
||||
use hotpath::wrap::tokio::sync::Mutex;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
use std::sync::Mutex as StdMutex;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
use tokio::sync::Mutex;
|
||||
use tokio::{
|
||||
net::UdpSocket,
|
||||
sync::mpsc::unbounded_channel,
|
||||
task::JoinSet,
|
||||
};
|
||||
|
||||
const MAX_PACKET: usize = 2048;
|
||||
|
||||
@@ -343,7 +355,7 @@ struct WgPeer {
|
||||
config: WgConfig,
|
||||
endpoint: SocketAddr,
|
||||
|
||||
sink: std::sync::Mutex<Option<Pin<Box<dyn ZCPacketSink>>>>,
|
||||
sink: StdMutex<Option<Pin<Box<dyn ZCPacketSink>>>>,
|
||||
|
||||
data: Option<WgPeerData>,
|
||||
tasks: JoinSet<()>,
|
||||
@@ -354,19 +366,19 @@ struct WgPeer {
|
||||
impl WgPeer {
|
||||
fn new(udp: Arc<UdpSocket>, config: WgConfig, endpoint: SocketAddr) -> Self {
|
||||
WgPeer {
|
||||
tunn: Some(Mutex::new(Tunn::new(
|
||||
tunn: Some(hotpath::mutex!(tokio::sync::Mutex::new(Tunn::new(
|
||||
config.my_secret_key.clone(),
|
||||
config.peer_public_key,
|
||||
None,
|
||||
None,
|
||||
rand::thread_rng().next_u32(),
|
||||
None,
|
||||
))),
|
||||
)))),
|
||||
|
||||
udp,
|
||||
config,
|
||||
endpoint,
|
||||
sink: std::sync::Mutex::new(None),
|
||||
sink: hotpath::mutex!(std::sync::Mutex::new(None)),
|
||||
|
||||
data: None,
|
||||
tasks: JoinSet::new(),
|
||||
@@ -375,6 +387,7 @@ impl WgPeer {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "WgTunnel"))]
|
||||
async fn handle_packet_from_me<S: ZCPacketStream + Unpin>(mut stream: S, data: WgPeerData) {
|
||||
while let Some(Ok(packet)) = stream.next().await {
|
||||
let ret = data.handle_one_packet_from_me(packet).await;
|
||||
@@ -386,6 +399,7 @@ impl WgPeer {
|
||||
.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "WgTunnel"))]
|
||||
async fn handle_packet_from_peer(&self, packet: &[u8]) {
|
||||
self.access_time.store(Instant::now());
|
||||
tracing::trace!("Received {} bytes from peer", packet.len());
|
||||
@@ -451,8 +465,8 @@ impl WgPeer {
|
||||
}
|
||||
}
|
||||
|
||||
type ConnSender = tokio::sync::mpsc::UnboundedSender<Box<dyn Tunnel>>;
|
||||
type ConnReceiver = tokio::sync::mpsc::UnboundedReceiver<Box<dyn Tunnel>>;
|
||||
type ConnSender = hotpath::wrap::tokio::sync::mpsc::UnboundedSender<Box<dyn Tunnel>>;
|
||||
type ConnReceiver = hotpath::wrap::tokio::sync::mpsc::UnboundedReceiver<Box<dyn Tunnel>>;
|
||||
|
||||
pub struct WgTunnelListener {
|
||||
addr: url::Url,
|
||||
@@ -470,7 +484,7 @@ pub struct WgTunnelListener {
|
||||
|
||||
impl WgTunnelListener {
|
||||
pub fn new(addr: url::Url, config: WgConfig) -> Self {
|
||||
let (conn_send, conn_recv) = hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
|
||||
let (conn_send, conn_recv) = hotpath::channel!(unbounded_channel());
|
||||
WgTunnelListener {
|
||||
addr,
|
||||
config,
|
||||
|
||||
Reference in New Issue
Block a user