mirror of
https://github.com/EasyTier/EasyTier.git
synced 2026-09-20 03:22:05 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 fanyang
|
a7aae67c43 | ||
|
fanyangandGitHub
|
f24735a86f |
@@ -0,0 +1,32 @@
|
||||
# rust-analyzer config
|
||||
# Skip the `easytier-gui` (Tauri) crate because its build scripts require
|
||||
# system libraries webkit2gtk-4.1 / javascriptcoregtk-4.1 that are not
|
||||
# installed on this host. Excluding it keeps rust-analyzer healthy for the
|
||||
# rest of the workspace.
|
||||
|
||||
# Override the command used to run build scripts / collect build data.
|
||||
[cargo.buildScripts]
|
||||
overrideCommand = [
|
||||
"cargo",
|
||||
"check",
|
||||
"--quiet",
|
||||
"--workspace",
|
||||
"--exclude",
|
||||
"easytier-gui",
|
||||
"--message-format=json",
|
||||
"--all-targets",
|
||||
"--keep-going",
|
||||
]
|
||||
|
||||
# Override the command used for diagnostics-on-save.
|
||||
[check]
|
||||
overrideCommand = [
|
||||
"cargo",
|
||||
"check",
|
||||
"--workspace",
|
||||
"--exclude",
|
||||
"easytier-gui",
|
||||
"--message-format=json",
|
||||
"--all-targets",
|
||||
"--keep-going",
|
||||
]
|
||||
Generated
+1
-146
@@ -263,12 +263,6 @@ version = "0.7.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
|
||||
|
||||
[[package]]
|
||||
name = "ascii"
|
||||
version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d92bec98840b8f03a5ff5413de5293bfcd8bf96467cf5452609f939ec6f5de16"
|
||||
|
||||
[[package]]
|
||||
name = "async-broadcast"
|
||||
version = "0.7.2"
|
||||
@@ -1266,12 +1260,6 @@ dependencies = [
|
||||
"windows-targets 0.52.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "chunked_transfer"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901"
|
||||
|
||||
[[package]]
|
||||
name = "ciborium"
|
||||
version = "0.2.2"
|
||||
@@ -2175,16 +2163,6 @@ dependencies = [
|
||||
"dirs-sys 0.5.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dirs-next"
|
||||
version = "2.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b98cf8ebf19c3d1b223e151f99a4f9f0690dca41414773390fc824184ac833e1"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"dirs-sys-next",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dirs-sys"
|
||||
version = "0.3.7"
|
||||
@@ -2208,17 +2186,6 @@ dependencies = [
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dirs-sys-next"
|
||||
version = "0.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4ebda144c4fe02d1f7ea1a7d9641b6fc6b580adcfa024ae48797ecdeb6825b4d"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"redox_users 0.4.5",
|
||||
"winapi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dispatch2"
|
||||
version = "0.3.1"
|
||||
@@ -2398,7 +2365,6 @@ dependencies = [
|
||||
"hickory-resolver",
|
||||
"hickory-server",
|
||||
"hmac",
|
||||
"hotpath",
|
||||
"http",
|
||||
"http_req",
|
||||
"humansize",
|
||||
@@ -2469,6 +2435,7 @@ dependencies = [
|
||||
"tempfile",
|
||||
"terminal_size",
|
||||
"thiserror 1.0.63",
|
||||
"thread_local",
|
||||
"thunk-rs",
|
||||
"tikv-jemalloc-ctl",
|
||||
"tikv-jemalloc-sys",
|
||||
@@ -2741,12 +2708,6 @@ version = "1.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4ef6b89e5b37196644d8796de5268852ff179b44e96276cf4290264843743bb7"
|
||||
|
||||
[[package]]
|
||||
name = "encode_unicode"
|
||||
version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0"
|
||||
|
||||
[[package]]
|
||||
name = "encoding"
|
||||
version = "0.2.33"
|
||||
@@ -4005,61 +3966,6 @@ dependencies = [
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hotpath"
|
||||
version = "0.21.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1ff6b552a6afa29d9e33f8d555bee9093c142dd449501ae128e6494a303f03dc"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"cfg-if",
|
||||
"crossbeam-channel",
|
||||
"flate2",
|
||||
"flume 0.12.0",
|
||||
"futures-util",
|
||||
"hdrhistogram",
|
||||
"hotpath-macros",
|
||||
"hotpath-meta",
|
||||
"libc",
|
||||
"object",
|
||||
"parking_lot",
|
||||
"pin-project-lite",
|
||||
"prettytable-rs",
|
||||
"quanta",
|
||||
"regex",
|
||||
"rustc-demangle",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tiny_http",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hotpath-macros"
|
||||
version = "0.21.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4f15322569d3cfadf84c0de7ef72be435b8f4b4839ee4ace78a7eaca48a87ded"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hotpath-macros-meta"
|
||||
version = "0.21.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b3675e29d16c844ccad12763672b33e51d9a000c346720c4f354f7a3bdc649a8"
|
||||
|
||||
[[package]]
|
||||
name = "hotpath-meta"
|
||||
version = "0.21.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d558d972ddc9483fb4e713af3dd41634edf895c7cb18ed13dc58c56431d42e27"
|
||||
dependencies = [
|
||||
"hotpath-macros-meta",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "html5ever"
|
||||
version = "0.29.1"
|
||||
@@ -6028,15 +5934,6 @@ dependencies = [
|
||||
"objc2-foundation",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "object"
|
||||
version = "0.36.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "62948e14d923ea95ea2c7c86c71013138b66525b86bdc08d2dcc262bdb497b87"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "once_cell"
|
||||
version = "1.21.3"
|
||||
@@ -6941,19 +6838,6 @@ dependencies = [
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prettytable-rs"
|
||||
version = "0.10.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "eea25e07510aa6ab6547308ebe3c036016d162b8da920dbb079e3ba8acf3d95a"
|
||||
dependencies = [
|
||||
"encode_unicode",
|
||||
"is-terminal",
|
||||
"lazy_static",
|
||||
"term",
|
||||
"unicode-width 0.1.11",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "primeorder"
|
||||
version = "0.13.6"
|
||||
@@ -8013,12 +7897,6 @@ dependencies = [
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustc-demangle"
|
||||
version = "0.1.27"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b50b8869d9fc858ce7266cce0194bd74df58b9d0e3f6df3a9fc8eb470d95c09d"
|
||||
|
||||
[[package]]
|
||||
name = "rustc-hash"
|
||||
version = "2.1.0"
|
||||
@@ -9943,17 +9821,6 @@ dependencies = [
|
||||
"utf-8",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "term"
|
||||
version = "0.7.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c59df8ac95d96ff9bede18eb7300b0fda5e5d8d90960e76f8e14ae765eedbf1f"
|
||||
dependencies = [
|
||||
"dirs-next",
|
||||
"rustversion",
|
||||
"winapi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "terminal_size"
|
||||
version = "0.4.1"
|
||||
@@ -10109,18 +9976,6 @@ version = "1.0.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "825f6c8a18bc36d56a62f66af7296385b628c9c5543a8663d4c217fc920bfefd"
|
||||
|
||||
[[package]]
|
||||
name = "tiny_http"
|
||||
version = "0.12.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "389915df6413a2e74fb181895f933386023c71110878cd0825588928e64cdc82"
|
||||
dependencies = [
|
||||
"ascii",
|
||||
"chunked_transfer",
|
||||
"httpdate",
|
||||
"log",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tinystr"
|
||||
version = "0.7.6"
|
||||
|
||||
+20
-12
@@ -32,6 +32,10 @@ path = "src/lib.rs"
|
||||
name = "tx_throughput"
|
||||
harness = false
|
||||
|
||||
[[bench]]
|
||||
name = "packet_bytes_extraction"
|
||||
harness = false
|
||||
|
||||
[dependencies]
|
||||
git-version = "0.3.9"
|
||||
|
||||
@@ -58,8 +62,6 @@ chrono = { version = "0.4.37", features = ["serde"] }
|
||||
guarden = "0.2"
|
||||
quanta = "0.12"
|
||||
|
||||
hotpath = { version = "0.21", default-features = false, optional = true }
|
||||
|
||||
delegate = "0.13.5"
|
||||
|
||||
itertools = "0.14.0"
|
||||
@@ -222,6 +224,7 @@ smoltcp = { git = "https://github.com/smoltcp-rs/smoltcp.git", rev = "0a926767a6
|
||||
"async",
|
||||
] }
|
||||
parking_lot = { version = "0.12.0" }
|
||||
thread_local = "1.1"
|
||||
|
||||
wildmatch = "2.3.4"
|
||||
|
||||
@@ -342,7 +345,7 @@ zip = "4.0.0"
|
||||
|
||||
|
||||
[dev-dependencies]
|
||||
criterion = "0.5.1"
|
||||
criterion = { version = "0.5", features = ["html_reports"] }
|
||||
serial_test = "3.0.0"
|
||||
rstest = "0.25.0"
|
||||
futures-util = "0.3.31"
|
||||
@@ -350,6 +353,14 @@ maplit = "1.0.2"
|
||||
tempfile = "3.22.0"
|
||||
ctor = "0.8.0"
|
||||
|
||||
[[bench]]
|
||||
name = "acl_hotpath"
|
||||
harness = false
|
||||
|
||||
[[bench]]
|
||||
name = "zc_packet"
|
||||
harness = false
|
||||
|
||||
[target.'cfg(target_os = "linux")'.dev-dependencies]
|
||||
defguard_wireguard_rs = "0.4.2"
|
||||
tokio-socks = "0.5.2"
|
||||
@@ -412,14 +423,11 @@ 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",
|
||||
"hotpath/tokio",
|
||||
"hotpath/parking_lot",
|
||||
"hotpath/flume",
|
||||
]
|
||||
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
|
||||
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
|
||||
# Deprecated: hotpath profiling has been removed. These feature aliases are
|
||||
# retained as no-ops so existing build scripts using `--features hotpath*`
|
||||
# continue to work without pulling in any dependencies.
|
||||
hotpath = []
|
||||
hotpath-cpu = ["hotpath"]
|
||||
hotpath-alloc = ["hotpath"]
|
||||
# For Network Extension on macOS
|
||||
macos-ne = []
|
||||
|
||||
@@ -1,4 +1,47 @@
|
||||
# TX Throughput Benchmark
|
||||
# Benchmarks
|
||||
|
||||
Criterion benchmarks for EasyTier hot paths.
|
||||
|
||||
| Bench | What it measures |
|
||||
| --------------------------- | -------------------------------------------------------------------------------- |
|
||||
| `tx_throughput` | End-to-end TX injection path through `peer_manager::send_msg_by_ip` |
|
||||
| `packet_bytes_extraction` | `ZCPacket::payload_bytes` / `tunnel_payload_bytes` extraction (advance hot path) |
|
||||
|
||||
## Packet Bytes Extraction
|
||||
|
||||
Criterion benchmark for `ZCPacket` bytes extraction — the methods touched by the
|
||||
`advance`-based slicing refactor. Measures `payload_bytes` and
|
||||
`tunnel_payload_bytes` at two payload sizes (1280, 4096). Setup
|
||||
(`ZCPacket::new_with_payload`) runs in the benchmark harness's preparation
|
||||
phase and is excluded from the timed region, so the numbers reflect only the
|
||||
extraction call.
|
||||
|
||||
### Quick start
|
||||
|
||||
```bash
|
||||
cargo bench --bench packet_bytes_extraction
|
||||
```
|
||||
|
||||
Smoke run:
|
||||
|
||||
```bash
|
||||
PACKET_BYTES_MEASUREMENT_SECS=2 \
|
||||
PACKET_BYTES_WARMUP_SECS=1 \
|
||||
PACKET_BYTES_SAMPLE_SIZE=10 \
|
||||
cargo bench --bench packet_bytes_extraction -- --quiet
|
||||
```
|
||||
|
||||
### Environment variables
|
||||
|
||||
| Variable | Default | Notes |
|
||||
| ------------------------------- | ------- | ---------------------------- |
|
||||
| `PACKET_BYTES_MEASUREMENT_SECS` | `10` | Criterion `measurement_time` |
|
||||
| `PACKET_BYTES_WARMUP_SECS` | `3` | Criterion `warm_up_time` |
|
||||
| `PACKET_BYTES_SAMPLE_SIZE` | `10` | Criterion `sample_size` (min 10) |
|
||||
|
||||
---
|
||||
|
||||
## TX Throughput Benchmark
|
||||
|
||||
Criterion benchmark for EasyTier's TX injection path (`peer_manager::send_msg_by_ip`).
|
||||
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
|
||||
use easytier::common::acl_processor::{AclProcessor, PacketInfo};
|
||||
use easytier::proto::acl::*;
|
||||
use std::net::{IpAddr, Ipv4Addr};
|
||||
|
||||
fn make_acl_config() -> Acl {
|
||||
let mut acl_config = Acl::default();
|
||||
let mut acl_v1 = AclV1::default();
|
||||
|
||||
let mut chain = Chain {
|
||||
name: "bench_inbound".to_string(),
|
||||
chain_type: ChainType::Inbound as i32,
|
||||
enabled: true,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
chain.rules.push(Rule {
|
||||
name: "allow_all".to_string(),
|
||||
priority: 100,
|
||||
enabled: true,
|
||||
action: Action::Allow as i32,
|
||||
protocol: Protocol::Any as i32,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
acl_v1.chains.push(chain);
|
||||
acl_config.acl_v1 = Some(acl_v1);
|
||||
acl_config
|
||||
}
|
||||
|
||||
fn make_packet_info() -> PacketInfo {
|
||||
PacketInfo {
|
||||
src_ip: IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)),
|
||||
dst_ip: IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)),
|
||||
src_port: Some(12345),
|
||||
dst_port: Some(80),
|
||||
protocol: Protocol::Tcp,
|
||||
packet_size: 1024,
|
||||
src_groups: Arc::new(vec![]),
|
||||
dst_groups: Arc::new(vec![]),
|
||||
}
|
||||
}
|
||||
|
||||
fn bench_cache_hit_single(c: &mut Criterion) {
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
let processor = rt.block_on(async { AclProcessor::new(make_acl_config()) });
|
||||
let packet_info = make_packet_info();
|
||||
|
||||
// Prime the cache
|
||||
let _ = processor.process_packet(&packet_info, ChainType::Inbound);
|
||||
|
||||
c.bench_function("acl_cache_hit_1t", |b| {
|
||||
b.iter(|| {
|
||||
std::hint::black_box(processor.process_packet(&packet_info, ChainType::Inbound));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
fn bench_cache_hit_multi(c: &mut Criterion) {
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
let mut group = c.benchmark_group("acl_cache_hit_multi");
|
||||
for threads in [2, 4, 8] {
|
||||
let processor = Arc::new(rt.block_on(async { AclProcessor::new(make_acl_config()) }));
|
||||
let packet_info = Arc::new(make_packet_info());
|
||||
|
||||
// Prime the cache
|
||||
let _ = processor.process_packet(&packet_info, ChainType::Inbound);
|
||||
|
||||
group.bench_with_input(
|
||||
BenchmarkId::from_parameter(threads),
|
||||
&threads,
|
||||
|b, &threads| {
|
||||
b.iter_custom(|iters| {
|
||||
use std::sync::Barrier;
|
||||
use std::thread;
|
||||
|
||||
let barrier = Arc::new(Barrier::new(threads + 1));
|
||||
let per_thread = (iters / threads as u64) as usize;
|
||||
let mut handles = Vec::with_capacity(threads);
|
||||
|
||||
for _ in 0..threads {
|
||||
let processor = Arc::clone(&processor);
|
||||
let packet_info = Arc::clone(&packet_info);
|
||||
let barrier = Arc::clone(&barrier);
|
||||
handles.push(thread::spawn(move || {
|
||||
barrier.wait();
|
||||
for _ in 0..per_thread {
|
||||
std::hint::black_box(
|
||||
processor.process_packet(&packet_info, ChainType::Inbound),
|
||||
);
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
let start = std::time::Instant::now();
|
||||
barrier.wait();
|
||||
for handle in handles {
|
||||
handle.join().unwrap();
|
||||
}
|
||||
start.elapsed()
|
||||
});
|
||||
},
|
||||
);
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn bench_unique_rule_match(c: &mut Criterion) {
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
let processor = rt.block_on(async { AclProcessor::new(make_acl_config()) });
|
||||
|
||||
c.bench_function("acl_unique_rule_match_1t", |b| {
|
||||
let mut i = 0usize;
|
||||
b.iter(|| {
|
||||
let mut packet_info = make_packet_info();
|
||||
packet_info.src_port = Some((1024 + (i % 60_000)) as u16);
|
||||
packet_info.src_ip = IpAddr::V4(Ipv4Addr::new(
|
||||
10,
|
||||
((i >> 16) & 0xff) as u8,
|
||||
((i >> 8) & 0xff) as u8,
|
||||
(i & 0xff) as u8,
|
||||
));
|
||||
std::hint::black_box(processor.process_packet(&packet_info, ChainType::Inbound));
|
||||
i = i.wrapping_add(1);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
criterion_group!(
|
||||
benches,
|
||||
bench_cache_hit_single,
|
||||
bench_cache_hit_multi,
|
||||
bench_unique_rule_match
|
||||
);
|
||||
criterion_main!(benches);
|
||||
@@ -0,0 +1,65 @@
|
||||
use std::hint::black_box;
|
||||
use std::time::Duration;
|
||||
|
||||
use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main};
|
||||
|
||||
use easytier::tunnel::packet_def::ZCPacket;
|
||||
|
||||
const PAYLOAD_SIZES: &[usize] = &[1280, 4096];
|
||||
|
||||
fn env_parse<T: std::str::FromStr>(key: &str, default: T) -> T {
|
||||
std::env::var(key)
|
||||
.ok()
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(default)
|
||||
}
|
||||
|
||||
fn bench_payload_bytes(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("payload_bytes");
|
||||
for &size in PAYLOAD_SIZES {
|
||||
let data = vec![0u8; size];
|
||||
group.throughput(Throughput::Bytes(size as u64));
|
||||
group.bench_with_input(format!("{size}"), &data, |b, data| {
|
||||
b.iter_batched(
|
||||
|| ZCPacket::new_with_payload(black_box(data)),
|
||||
|p| black_box(p).payload_bytes(),
|
||||
BatchSize::SmallInput,
|
||||
)
|
||||
});
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn bench_tunnel_payload_bytes(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("tunnel_payload_bytes");
|
||||
for &size in PAYLOAD_SIZES {
|
||||
let data = vec![0u8; size];
|
||||
group.throughput(Throughput::Bytes(size as u64));
|
||||
group.bench_with_input(format!("{size}"), &data, |b, data| {
|
||||
b.iter_batched(
|
||||
|| ZCPacket::new_with_payload(black_box(data)),
|
||||
|p| black_box(p).tunnel_payload_bytes(),
|
||||
BatchSize::SmallInput,
|
||||
)
|
||||
});
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn criterion_config() -> Criterion {
|
||||
let measurement_secs = env_parse("PACKET_BYTES_MEASUREMENT_SECS", 10u64);
|
||||
let warmup_secs = env_parse("PACKET_BYTES_WARMUP_SECS", 3u64);
|
||||
let sample_size = env_parse("PACKET_BYTES_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)
|
||||
}
|
||||
|
||||
criterion_group! {
|
||||
name = benches;
|
||||
config = criterion_config();
|
||||
targets = bench_payload_bytes, bench_tunnel_payload_bytes
|
||||
}
|
||||
criterion_main!(benches);
|
||||
@@ -0,0 +1,51 @@
|
||||
use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
|
||||
use easytier::tunnel::packet_def::{ZCPacket, ZCPacketType};
|
||||
|
||||
fn bench_new_with_payload(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("zc_new_with_payload");
|
||||
for size in [64usize, 1500] {
|
||||
let payload = vec![0xabu8; size];
|
||||
group.bench_with_input(BenchmarkId::from_parameter(size), &payload, |b, payload| {
|
||||
b.iter(|| {
|
||||
std::hint::black_box(ZCPacket::new_with_payload(std::hint::black_box(payload)));
|
||||
});
|
||||
});
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn bench_new_for_foreign_network(c: &mut Criterion) {
|
||||
let payload = vec![0xabu8; 64];
|
||||
let foreign_packet = ZCPacket::new_with_payload(&payload);
|
||||
let network_name = "bench-network".to_string();
|
||||
|
||||
c.bench_function("zc_new_for_foreign_network_64b", |b| {
|
||||
b.iter(|| {
|
||||
std::hint::black_box(ZCPacket::new_for_foreign_network(
|
||||
std::hint::black_box(&network_name),
|
||||
42,
|
||||
std::hint::black_box(&foreign_packet),
|
||||
));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
fn bench_convert_type(c: &mut Criterion) {
|
||||
let payload = vec![0xabu8; 64];
|
||||
let packet = ZCPacket::new_with_payload(&payload);
|
||||
|
||||
c.bench_function("zc_convert_type_tcp_64b", |b| {
|
||||
b.iter(|| {
|
||||
let p = std::hint::black_box(packet.clone());
|
||||
std::hint::black_box(p.convert_type(ZCPacketType::TCP));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
criterion_group!(
|
||||
benches,
|
||||
bench_new_with_payload,
|
||||
bench_new_for_foreign_network,
|
||||
bench_convert_type
|
||||
);
|
||||
criterion_main!(benches);
|
||||
@@ -3,7 +3,8 @@ use std::{
|
||||
net::{IpAddr, SocketAddr},
|
||||
str::FromStr as _,
|
||||
sync::Arc,
|
||||
time::{Duration, SystemTime, UNIX_EPOCH},
|
||||
sync::atomic::{AtomicU64, Ordering::Relaxed},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use quanta::Instant;
|
||||
@@ -76,7 +77,41 @@ pub struct FastLookupRule {
|
||||
pub stateful: bool,
|
||||
pub rate_limit: u32,
|
||||
pub burst_limit: u32,
|
||||
pub rule_stats: Arc<RuleStats>,
|
||||
pub rule_stats: Arc<RuleStatsTracker>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct RuleStatsTracker {
|
||||
rule: Option<Rule>,
|
||||
packets: AtomicU64,
|
||||
bytes: AtomicU64,
|
||||
}
|
||||
|
||||
impl RuleStatsTracker {
|
||||
fn new(rule: Option<Rule>) -> Self {
|
||||
Self {
|
||||
rule,
|
||||
packets: AtomicU64::new(0),
|
||||
bytes: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RuleStatsTracker"))]
|
||||
fn increment(&self, packet_size: usize) {
|
||||
self.packets.fetch_add(1, Relaxed);
|
||||
self.bytes.fetch_add(packet_size as u64, Relaxed);
|
||||
}
|
||||
|
||||
fn snapshot(&self) -> RuleStats {
|
||||
RuleStats {
|
||||
rule: self.rule.clone(),
|
||||
stat: Some(StatItem {
|
||||
packet_count: self.packets.load(Relaxed),
|
||||
byte_count: self.bytes.load(Relaxed),
|
||||
}),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Cache key combining packet info and chain type
|
||||
@@ -108,17 +143,17 @@ impl AclCacheKey {
|
||||
}
|
||||
|
||||
// Cache entry with timestamp for LRU cleanup
|
||||
#[derive(Debug, Clone)]
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct AclCacheEntry {
|
||||
pub action: Action,
|
||||
pub matched_rule: RuleId,
|
||||
pub last_access: Instant,
|
||||
pub last_access: AtomicU64,
|
||||
// New fields to track rule characteristics for proper cache behavior
|
||||
pub conn_track_key: Option<String>,
|
||||
pub rate_limit_keys: Vec<RateLimitKey>,
|
||||
pub chain_type: ChainType,
|
||||
pub acl_result: Option<AclResult>,
|
||||
pub rule_stats_vec: Vec<Arc<RuleStats>>,
|
||||
pub rule_stats_vec: Vec<Arc<RuleStatsTracker>>,
|
||||
}
|
||||
|
||||
// Packet info extracted for ACL processing
|
||||
@@ -211,7 +246,7 @@ pub struct AclProcessor {
|
||||
default_outbound_action: Action,
|
||||
default_forward_action: Action,
|
||||
|
||||
default_rule_stats: Arc<RuleStats>,
|
||||
default_rule_stats: Arc<RuleStatsTracker>,
|
||||
|
||||
// Connection tracking table - shared across different processor instances if needed
|
||||
conn_track: Arc<DashMap<String, ConnTrackEntry>>,
|
||||
@@ -224,6 +259,13 @@ pub struct AclProcessor {
|
||||
cache_max_size: usize,
|
||||
cache_cleanup_interval: Duration,
|
||||
|
||||
// Coarse monotonic timestamp updated by the cleanup task, used to avoid
|
||||
// calling Instant::now() on every cache hit.
|
||||
coarse_millis: Arc<AtomicU64>,
|
||||
|
||||
// Hot-path counters that bypass the DashMap stats table
|
||||
cache_hits: AtomicU64,
|
||||
|
||||
// Statistics
|
||||
stats: Arc<DashMap<AclStatKey, u64>>,
|
||||
|
||||
@@ -259,18 +301,14 @@ impl AclProcessor {
|
||||
default_outbound_action,
|
||||
default_forward_action,
|
||||
|
||||
default_rule_stats: Arc::new(RuleStats {
|
||||
rule: None,
|
||||
stat: Some(StatItem {
|
||||
packet_count: 0,
|
||||
byte_count: 0,
|
||||
}),
|
||||
}),
|
||||
default_rule_stats: Arc::new(RuleStatsTracker::new(None)),
|
||||
conn_track: conn_track.unwrap_or_else(|| Arc::new(DashMap::new())),
|
||||
rate_limiters: rate_limiters.unwrap_or_else(|| Arc::new(DashMap::new())),
|
||||
rule_cache: Arc::new(DashMap::new()), // Always start with fresh cache
|
||||
cache_max_size: 1024, // Limit cache to 1k entries
|
||||
cache_cleanup_interval: Duration::from_secs(20), // Cleanup every 5 minutes
|
||||
coarse_millis: Arc::new(AtomicU64::new(0)),
|
||||
cache_hits: AtomicU64::new(0),
|
||||
stats: stats.unwrap_or_else(|| Arc::new(DashMap::new())),
|
||||
tasks,
|
||||
};
|
||||
@@ -374,11 +412,14 @@ impl AclProcessor {
|
||||
let rule_cache = self.rule_cache.clone();
|
||||
let cache_max_size = self.cache_max_size;
|
||||
let cleanup_interval = self.cache_cleanup_interval;
|
||||
let coarse_millis = self.coarse_millis.clone();
|
||||
|
||||
self.tasks.spawn(async move {
|
||||
let mut interval = tokio::time::interval(cleanup_interval);
|
||||
loop {
|
||||
interval.tick().await;
|
||||
let now = crate::common::stats_manager::now_monotonic_millis();
|
||||
coarse_millis.store(now, Relaxed);
|
||||
Self::cleanup_cache(&rule_cache, cache_max_size);
|
||||
rule_cache.shrink_to_fit();
|
||||
|
||||
@@ -401,10 +442,9 @@ impl AclProcessor {
|
||||
/// Clean up cache using LRU strategy
|
||||
fn cleanup_cache(cache: &DashMap<AclCacheKey, AclCacheEntry>, max_size: usize) {
|
||||
// remove cache not be used in last 15 second
|
||||
let expired_timepoint = Instant::now()
|
||||
.checked_sub(Duration::from_secs(15))
|
||||
.unwrap_or(Instant::now());
|
||||
cache.retain(|_, entry| entry.last_access > expired_timepoint);
|
||||
let now = crate::common::stats_manager::now_monotonic_millis();
|
||||
let cutoff = now.saturating_sub(15_000);
|
||||
cache.retain(|_, entry| entry.last_access.load(Relaxed) > cutoff);
|
||||
|
||||
let current_size = cache.len();
|
||||
if current_size <= max_size {
|
||||
@@ -412,9 +452,9 @@ impl AclProcessor {
|
||||
}
|
||||
|
||||
// Remove oldest entries (LRU cleanup)
|
||||
let mut entries: Vec<(AclCacheKey, Instant)> = cache
|
||||
let mut entries: Vec<(AclCacheKey, u64)> = cache
|
||||
.iter()
|
||||
.map(|entry| (entry.key().clone(), entry.value().last_access))
|
||||
.map(|entry| (entry.key().clone(), entry.value().last_access.load(Relaxed)))
|
||||
.collect();
|
||||
|
||||
// Sort by last_access (oldest first)
|
||||
@@ -433,6 +473,7 @@ impl AclProcessor {
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "AclProcessor"))]
|
||||
pub(crate) fn process_packet_with_cache_entry(
|
||||
&self,
|
||||
packet_info: &PacketInfo,
|
||||
@@ -459,42 +500,41 @@ impl AclProcessor {
|
||||
cache_entry.acl_result.clone().unwrap()
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "AclProcessor"))]
|
||||
fn inc_cache_entry_stats(&self, cache_entry: &AclCacheEntry, packet_info: &PacketInfo) {
|
||||
for rule_stats in cache_entry.rule_stats_vec.iter() {
|
||||
// Use unsafe code to mutate the contents behind the Arc
|
||||
let stat_ptr = rule_stats.stat.as_ref().unwrap() as *const StatItem as *mut StatItem;
|
||||
unsafe {
|
||||
(*stat_ptr).packet_count += 1;
|
||||
(*stat_ptr).byte_count += packet_info.packet_size as u64;
|
||||
}
|
||||
rule_stats.increment(packet_info.packet_size);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn get_rules_stats(&self) -> Vec<RuleStats> {
|
||||
let mut stats: Vec<RuleStats> = Vec::new();
|
||||
for rule in self.inbound_rules.iter() {
|
||||
stats.push((*rule.rule_stats).clone());
|
||||
stats.push(rule.rule_stats.snapshot());
|
||||
}
|
||||
for rule in self.outbound_rules.iter() {
|
||||
stats.push((*rule.rule_stats).clone());
|
||||
stats.push(rule.rule_stats.snapshot());
|
||||
}
|
||||
for rule in self.forward_rules.iter() {
|
||||
stats.push((*rule.rule_stats).clone());
|
||||
stats.push(rule.rule_stats.snapshot());
|
||||
}
|
||||
stats
|
||||
}
|
||||
|
||||
/// Process a packet through ACL rules - Now lock-free!
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "AclProcessor"))]
|
||||
pub fn process_packet(&self, packet_info: &PacketInfo, chain_type: ChainType) -> AclResult {
|
||||
// Check cache first for performance
|
||||
let cache_key = AclCacheKey::from_packet_info(packet_info, chain_type);
|
||||
|
||||
// If cache hit and can skip checks, return cached result
|
||||
if let Some(mut cached) = self.rule_cache.get_mut(&cache_key) {
|
||||
// Update last access time for LRU
|
||||
cached.last_access = Instant::now();
|
||||
|
||||
self.increment_stat(AclStatKey::CacheHits);
|
||||
// If cache hit and can skip checks, return cached result.
|
||||
// Use get() (read lock) instead of get_mut() (write lock) and update
|
||||
// last_access via AtomicU64, avoiding expensive Instant::now().
|
||||
if let Some(cached) = self.rule_cache.get(&cache_key) {
|
||||
cached
|
||||
.last_access
|
||||
.store(self.coarse_millis.load(Relaxed), Relaxed);
|
||||
self.cache_hits.fetch_add(1, Relaxed);
|
||||
return self.process_packet_with_cache_entry(packet_info, &cached);
|
||||
}
|
||||
|
||||
@@ -516,7 +556,7 @@ impl AclProcessor {
|
||||
let mut cache_entry = AclCacheEntry {
|
||||
action: Action::Allow,
|
||||
matched_rule: RuleId::Default,
|
||||
last_access: Instant::now(),
|
||||
last_access: AtomicU64::new(self.coarse_millis.load(Relaxed)),
|
||||
conn_track_key: None,
|
||||
rate_limit_keys: vec![],
|
||||
chain_type,
|
||||
@@ -581,8 +621,9 @@ impl AclProcessor {
|
||||
// Cache the result with rule info
|
||||
self.increment_stat(AclStatKey::RuleMatches);
|
||||
self.inc_cache_entry_stats(&cache_entry, packet_info);
|
||||
self.cache_result(&cache_key, cache_entry.clone());
|
||||
return cache_entry.acl_result.clone().unwrap();
|
||||
let result = cache_entry.acl_result.clone().unwrap();
|
||||
self.cache_result(&cache_key, cache_entry);
|
||||
return result;
|
||||
}
|
||||
|
||||
let default_action = match chain_type {
|
||||
@@ -618,8 +659,9 @@ impl AclProcessor {
|
||||
|
||||
// Cache the default result (no rule info)
|
||||
self.inc_cache_entry_stats(&cache_entry, packet_info);
|
||||
self.cache_result(&cache_key, cache_entry.clone());
|
||||
cache_entry.acl_result.clone().unwrap()
|
||||
let result = cache_entry.acl_result.clone().unwrap();
|
||||
self.cache_result(&cache_key, cache_entry);
|
||||
result
|
||||
}
|
||||
|
||||
/// Get shared state for preserving across hot reloads
|
||||
@@ -743,13 +785,11 @@ impl AclProcessor {
|
||||
|
||||
/// Check connection state for stateful rules
|
||||
fn check_connection_state(&self, conn_track_key: &str, packet_info: &PacketInfo) {
|
||||
let now = current_unix_secs();
|
||||
self.conn_track
|
||||
.entry(conn_track_key.to_string())
|
||||
.and_modify(|x| {
|
||||
x.last_seen = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_secs();
|
||||
x.last_seen = now;
|
||||
x.packet_count += 1;
|
||||
x.byte_count += packet_info.packet_size as u64;
|
||||
x.state = ConnState::Established as i32;
|
||||
@@ -763,14 +803,8 @@ impl AclProcessor {
|
||||
),
|
||||
protocol: packet_info.protocol as i32,
|
||||
state: ConnState::New as i32,
|
||||
created_at: SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_secs(),
|
||||
last_seen: SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_secs(),
|
||||
created_at: now,
|
||||
last_seen: now,
|
||||
packet_count: 1,
|
||||
byte_count: packet_info.packet_size as u64,
|
||||
});
|
||||
@@ -862,13 +896,7 @@ impl AclProcessor {
|
||||
stateful: rule.stateful,
|
||||
rate_limit: rule.rate_limit,
|
||||
burst_limit: rule.burst_limit,
|
||||
rule_stats: Arc::new(RuleStats {
|
||||
rule: Some(rule.clone()),
|
||||
stat: Some(StatItem {
|
||||
packet_count: 0,
|
||||
byte_count: 0,
|
||||
}),
|
||||
}),
|
||||
rule_stats: Arc::new(RuleStatsTracker::new(Some(rule.clone()))),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -894,6 +922,10 @@ impl AclProcessor {
|
||||
.collect::<HashMap<_, _>>();
|
||||
|
||||
// Add cache statistics using enum keys
|
||||
stats.insert(
|
||||
AclStatKey::CacheHits.as_str(),
|
||||
self.cache_hits.load(Relaxed),
|
||||
);
|
||||
stats.insert(AclStatKey::CacheSize.as_str(), self.rule_cache.len() as u64);
|
||||
stats.insert(
|
||||
AclStatKey::CacheMaxSize.as_str(),
|
||||
@@ -908,14 +940,11 @@ impl AclProcessor {
|
||||
conn_track: Arc<DashMap<String, ConnTrackEntry>>,
|
||||
timeout_secs: u64,
|
||||
) {
|
||||
let current_time = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_secs();
|
||||
let current_time = current_unix_secs();
|
||||
let keys_to_remove: Vec<String> = conn_track
|
||||
.iter()
|
||||
.filter_map(|entry| {
|
||||
if current_time - entry.last_seen > timeout_secs {
|
||||
if current_time.saturating_sub(entry.last_seen) > timeout_secs {
|
||||
Some(entry.key().clone())
|
||||
} else {
|
||||
None
|
||||
@@ -930,11 +959,7 @@ impl AclProcessor {
|
||||
|
||||
/// Get cache hit rate
|
||||
pub fn get_cache_hit_rate(&self) -> f64 {
|
||||
let cache_hits = self
|
||||
.stats
|
||||
.get(&AclStatKey::CacheHits)
|
||||
.map(|v| *v.value())
|
||||
.unwrap_or(0);
|
||||
let cache_hits = self.cache_hits.load(Relaxed);
|
||||
let total_requests = cache_hits
|
||||
+ self
|
||||
.stats
|
||||
@@ -950,6 +975,13 @@ impl AclProcessor {
|
||||
}
|
||||
}
|
||||
|
||||
fn current_unix_secs() -> u64 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_secs()
|
||||
}
|
||||
|
||||
// 新增辅助函数
|
||||
fn parse_port_start(port_strs: &[String]) -> Option<u16> {
|
||||
port_strs
|
||||
|
||||
@@ -22,6 +22,7 @@ pub mod machine_id;
|
||||
pub mod netns;
|
||||
pub mod network;
|
||||
pub mod os_info;
|
||||
pub mod sharded_counter;
|
||||
pub mod stats_manager;
|
||||
pub mod stun;
|
||||
pub mod stun_codec_ext;
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
|
||||
|
||||
const PUBLISH_INTERVAL: u64 = 256;
|
||||
|
||||
/// Counter optimized for long-lived worker threads.
|
||||
///
|
||||
/// Pending values below `PUBLISH_INTERVAL` live in `ThreadLocal` shards. The
|
||||
/// shards are retained until this counter is dropped, so this is intended for
|
||||
/// tokio workers or similarly long-lived threads rather than high-churn threads.
|
||||
pub struct ShardedCounter {
|
||||
published: AtomicU64,
|
||||
locals: thread_local::ThreadLocal<AtomicU64>,
|
||||
}
|
||||
|
||||
impl ShardedCounter {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
published: AtomicU64::new(0),
|
||||
locals: thread_local::ThreadLocal::new(),
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "ShardedCounter"))]
|
||||
pub fn add(&self, delta: u64) {
|
||||
let local = self.locals.get_or(|| AtomicU64::new(0));
|
||||
let v = local.load(Relaxed).saturating_add(delta);
|
||||
local.store(v, Relaxed);
|
||||
if v >= PUBLISH_INTERVAL {
|
||||
let pending = local.swap(0, Relaxed);
|
||||
if pending > 0 {
|
||||
self.published.fetch_add(pending, Relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn inc(&self) {
|
||||
self.add(1);
|
||||
}
|
||||
|
||||
pub fn get(&self) -> u64 {
|
||||
self.locals
|
||||
.iter()
|
||||
.fold(self.published.load(Relaxed), |total, local| {
|
||||
total.saturating_add(local.load(Relaxed))
|
||||
})
|
||||
}
|
||||
|
||||
pub fn reset(&self) {
|
||||
self.published.store(0, Relaxed);
|
||||
for local in self.locals.iter() {
|
||||
local.store(0, Relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for ShardedCounter {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for ShardedCounter {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("ShardedCounter")
|
||||
.field("value", &self.get())
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::{sync::Arc, thread};
|
||||
|
||||
#[test]
|
||||
fn sharded_counter_get_includes_other_thread_locals() {
|
||||
let counter = Arc::new(ShardedCounter::new());
|
||||
let thread_counter = Arc::clone(&counter);
|
||||
|
||||
thread::spawn(move || {
|
||||
for _ in 0..10 {
|
||||
thread_counter.inc();
|
||||
}
|
||||
})
|
||||
.join()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(counter.get(), 10);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sharded_counter_reset_clears_other_thread_locals() {
|
||||
let counter = Arc::new(ShardedCounter::new());
|
||||
let thread_counter = Arc::clone(&counter);
|
||||
|
||||
thread::spawn(move || {
|
||||
for _ in 0..10 {
|
||||
thread_counter.inc();
|
||||
}
|
||||
})
|
||||
.join()
|
||||
.unwrap();
|
||||
|
||||
counter.reset();
|
||||
|
||||
assert_eq!(counter.get(), 0);
|
||||
}
|
||||
}
|
||||
@@ -1,13 +1,16 @@
|
||||
use crate::common::sharded_counter::ShardedCounter;
|
||||
use dashmap::DashMap;
|
||||
use quanta::Instant;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::cell::UnsafeCell;
|
||||
use std::fmt;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::{Arc, LazyLock};
|
||||
use std::time::Duration;
|
||||
use tokio::time::interval;
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
|
||||
static START_INSTANT: LazyLock<Instant> = LazyLock::new(Instant::now);
|
||||
|
||||
/// Predefined metric names for type safety
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
|
||||
pub enum MetricName {
|
||||
@@ -375,10 +378,10 @@ impl Default for LabelSet {
|
||||
}
|
||||
}
|
||||
|
||||
/// UnsafeCounter provides a high-performance counter using UnsafeCell
|
||||
/// High-performance counter backed by sharded thread-local accumulation.
|
||||
#[derive(Debug)]
|
||||
pub struct UnsafeCounter {
|
||||
value: UnsafeCell<u64>,
|
||||
inner: ShardedCounter,
|
||||
}
|
||||
|
||||
impl Default for UnsafeCounter {
|
||||
@@ -390,121 +393,56 @@ impl Default for UnsafeCounter {
|
||||
impl UnsafeCounter {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
value: UnsafeCell::new(0),
|
||||
inner: ShardedCounter::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new_with_value(initial: u64) -> Self {
|
||||
Self {
|
||||
value: UnsafeCell::new(initial),
|
||||
}
|
||||
pub fn add(&self, delta: u64) {
|
||||
self.inner.add(delta);
|
||||
}
|
||||
|
||||
/// 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 inc(&self) {
|
||||
self.inner.inc();
|
||||
}
|
||||
|
||||
/// 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 get(&self) -> u64 {
|
||||
self.inner.get()
|
||||
}
|
||||
|
||||
/// 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 }
|
||||
}
|
||||
|
||||
/// 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;
|
||||
}
|
||||
}
|
||||
|
||||
/// 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 reset(&self) {
|
||||
self.inner.reset();
|
||||
}
|
||||
}
|
||||
|
||||
// UnsafeCounter is Send + Sync because the safety is guaranteed by the caller
|
||||
unsafe impl Send for UnsafeCounter {}
|
||||
unsafe impl Sync for UnsafeCounter {}
|
||||
|
||||
/// 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>,
|
||||
last_updated: AtomicU64,
|
||||
}
|
||||
|
||||
pub(crate) fn now_monotonic_millis() -> u64 {
|
||||
Instant::now().duration_since(*START_INSTANT).as_millis() as u64
|
||||
}
|
||||
|
||||
impl MetricData {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
counter: UnsafeCounter::new(),
|
||||
last_updated: UnsafeCell::new(Instant::now()),
|
||||
last_updated: AtomicU64::new(now_monotonic_millis()),
|
||||
}
|
||||
}
|
||||
|
||||
fn new_with_value(initial: u64) -> Self {
|
||||
Self {
|
||||
counter: UnsafeCounter::new_with_value(initial),
|
||||
last_updated: UnsafeCell::new(Instant::now()),
|
||||
}
|
||||
fn touch(&self) {
|
||||
self.last_updated
|
||||
.store(now_monotonic_millis(), Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// 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();
|
||||
}
|
||||
}
|
||||
|
||||
/// 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 }
|
||||
fn get_last_updated(&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 {
|
||||
@@ -545,41 +483,23 @@ 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();
|
||||
}
|
||||
}
|
||||
|
||||
/// 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.reset();
|
||||
self.metric_data.touch();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -615,9 +535,7 @@ impl StatsManager {
|
||||
loop {
|
||||
interval.tick().await;
|
||||
|
||||
let Some(cutoff_time) = Instant::now().checked_sub(Duration::from_secs(180)) else {
|
||||
continue;
|
||||
};
|
||||
let cutoff_millis = now_monotonic_millis().saturating_sub(180_000);
|
||||
|
||||
let Some(counters) = counters_clone.upgrade() else {
|
||||
break;
|
||||
@@ -625,7 +543,7 @@ impl StatsManager {
|
||||
|
||||
counters.retain(|_, metric_data: &mut Arc<MetricData>| {
|
||||
Arc::strong_count(metric_data) > 1
|
||||
|| unsafe { metric_data.get_last_updated() > cutoff_time }
|
||||
|| metric_data.get_last_updated() >= cutoff_millis
|
||||
});
|
||||
counters.shrink_to_fit();
|
||||
}
|
||||
@@ -663,7 +581,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,
|
||||
@@ -696,7 +614,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(),
|
||||
@@ -797,17 +715,9 @@ mod tests {
|
||||
async fn test_unsafe_counter() {
|
||||
let counter = UnsafeCounter::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.add(256);
|
||||
assert_eq!(counter.get(), 256);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -853,11 +763,11 @@ mod tests {
|
||||
let stats = StatsManager::new();
|
||||
|
||||
let counter1 = stats.get_simple_counter(MetricName::TrafficBytesTx);
|
||||
counter1.set(100);
|
||||
counter1.add(100);
|
||||
|
||||
let labels = LabelSet::new().with_label("status", "success");
|
||||
let counter2 = stats.get_counter(MetricName::PeerRpcClientTx, labels);
|
||||
counter2.set(50);
|
||||
counter2.add(50);
|
||||
|
||||
let traffic_labels = LabelSet::new()
|
||||
.with_label_type(LabelType::NetworkName("default".to_string()))
|
||||
@@ -865,7 +775,7 @@ mod tests {
|
||||
"87ede5a2-9c3d-492d-9bbe-989b9d07e742".to_string(),
|
||||
));
|
||||
let counter3 = stats.get_counter(MetricName::TrafficBytesTxByInstance, traffic_labels);
|
||||
counter3.set(25);
|
||||
counter3.add(25);
|
||||
|
||||
let prometheus_output = stats.export_prometheus();
|
||||
|
||||
@@ -885,7 +795,7 @@ mod tests {
|
||||
|
||||
let labels = LabelSet::new().with_label("peer", "test");
|
||||
let counter = stats.get_counter(MetricName::PeerRpcClientTx, labels.clone());
|
||||
counter.set(42);
|
||||
counter.add(42);
|
||||
|
||||
let metric = stats
|
||||
.get_metric(MetricName::PeerRpcClientTx, &labels)
|
||||
@@ -902,11 +812,11 @@ mod tests {
|
||||
|
||||
stats
|
||||
.get_simple_counter(MetricName::PeerRpcClientTx)
|
||||
.set(10);
|
||||
stats.get_simple_counter(MetricName::PeerRpcErrors).set(2);
|
||||
.add(10);
|
||||
stats.get_simple_counter(MetricName::PeerRpcErrors).add(2);
|
||||
stats
|
||||
.get_simple_counter(MetricName::TrafficBytesTx)
|
||||
.set(100);
|
||||
.add(100);
|
||||
|
||||
let rpc_metrics = stats.get_metrics_by_prefix("peer_rpc");
|
||||
assert_eq!(rpc_metrics.len(), 2);
|
||||
@@ -921,19 +831,15 @@ mod tests {
|
||||
|
||||
// 创建一些计数器
|
||||
let counter1 = stats.get_simple_counter(MetricName::PeerRpcClientTx);
|
||||
counter1.set(10);
|
||||
counter1.add(10);
|
||||
|
||||
let labels = LabelSet::new().with_label("test", "value");
|
||||
let counter2 = stats.get_counter(MetricName::TrafficBytesTx, labels);
|
||||
counter2.set(20);
|
||||
counter2.add(20);
|
||||
|
||||
// 验证计数器存在
|
||||
assert_eq!(stats.metric_count(), 2);
|
||||
|
||||
// 注意：实际的清理测试需要等待3分钟，这在单元测试中不现实
|
||||
// 这里我们只验证清理机制的基本结构是否正确
|
||||
// 清理逻辑在后台线程中运行，会自动删除超过3分钟未更新的条目
|
||||
|
||||
// 验证计数器仍然可以正常工作
|
||||
counter1.inc();
|
||||
assert_eq!(counter1.get(), 11);
|
||||
@@ -946,14 +852,14 @@ mod tests {
|
||||
async fn test_cleanup_keeps_metrics_with_live_handles() {
|
||||
let stats = StatsManager::new();
|
||||
let counter = stats.get_simple_counter(MetricName::TrafficBytesForwarded);
|
||||
counter.set(1);
|
||||
counter.add(1);
|
||||
|
||||
let cutoff_time = Instant::now().checked_add(Duration::from_secs(1)).unwrap();
|
||||
// Use a future cutoff so last_updated check always fails
|
||||
let future_cutoff = now_monotonic_millis() + 1000;
|
||||
stats
|
||||
.counters
|
||||
.retain(|_, metric_data: &mut Arc<MetricData>| {
|
||||
Arc::strong_count(metric_data) > 1
|
||||
|| unsafe { metric_data.get_last_updated() > cutoff_time }
|
||||
Arc::strong_count(metric_data) > 1 || metric_data.get_last_updated() > future_cutoff
|
||||
});
|
||||
|
||||
assert_eq!(stats.metric_count(), 1);
|
||||
@@ -963,8 +869,7 @@ mod tests {
|
||||
stats
|
||||
.counters
|
||||
.retain(|_, metric_data: &mut Arc<MetricData>| {
|
||||
Arc::strong_count(metric_data) > 1
|
||||
|| unsafe { metric_data.get_last_updated() > cutoff_time }
|
||||
Arc::strong_count(metric_data) > 1 || metric_data.get_last_updated() > future_cutoff
|
||||
});
|
||||
assert_eq!(stats.metric_count(), 0);
|
||||
}
|
||||
|
||||
@@ -24,16 +24,6 @@ pub static malloc_conf: &[u8] = b"retain:false\0";
|
||||
rust_i18n::i18n!("locales", fallback = "en");
|
||||
|
||||
#[tokio::main(flavor = "current_thread")]
|
||||
#[cfg_attr(
|
||||
all(
|
||||
feature = "hotpath",
|
||||
not(all(
|
||||
feature = "hotpath-alloc",
|
||||
any(feature = "jemalloc", feature = "mimalloc")
|
||||
))
|
||||
),
|
||||
hotpath::main
|
||||
)]
|
||||
async fn main() -> std::process::ExitCode {
|
||||
core::main().await
|
||||
}
|
||||
|
||||
@@ -1,52 +0,0 @@
|
||||
//! No-op stand-in for the `hotpath` macros used by this crate, selected when
|
||||
//! the `hotpath` feature is disabled.
|
||||
//!
|
||||
//! Keeping `hotpath` as an optional dependency means default builds do not pull
|
||||
//! the profiler (or any of its transitive dependencies) into the dependency
|
||||
//! graph. These macros expand to their input unchanged, mirroring `hotpath`'s
|
||||
//! own disabled mode so call sites compile identically with or without the
|
||||
//! feature.
|
||||
//!
|
||||
//! The macros are `#[macro_export]`-ed so that `lib.rs`' `extern crate self as
|
||||
//! hotpath` alias exposes them through the same `hotpath::...` paths used when
|
||||
//! the feature is enabled.
|
||||
|
||||
/// No-op mirroring `hotpath::channel!`: returns the channel expression
|
||||
/// unchanged (dropping any optional trailing `label`/`log`/`capacity` args).
|
||||
#[doc(hidden)]
|
||||
#[macro_export]
|
||||
macro_rules! channel {
|
||||
($expr:expr $(, $($rest:tt)*)?) => {
|
||||
$expr
|
||||
};
|
||||
}
|
||||
|
||||
/// No-op mirroring `hotpath::mutex!`: returns the expression unchanged.
|
||||
#[doc(hidden)]
|
||||
#[macro_export]
|
||||
macro_rules! mutex {
|
||||
($expr:expr $(, $($rest:tt)*)?) => {
|
||||
$expr
|
||||
};
|
||||
}
|
||||
|
||||
/// No-op mirroring `hotpath::rw_lock!`: returns the expression unchanged.
|
||||
#[doc(hidden)]
|
||||
#[macro_export]
|
||||
macro_rules! rw_lock {
|
||||
($expr:expr $(, $($rest:tt)*)?) => {
|
||||
$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;
|
||||
}
|
||||
@@ -24,7 +24,7 @@ use crate::{
|
||||
};
|
||||
|
||||
use byteorder::WriteBytesExt as _;
|
||||
use bytes::{BufMut, BytesMut};
|
||||
use bytes::{Buf, BufMut, BytesMut};
|
||||
use cidr::{Ipv4Inet, Ipv6Inet};
|
||||
use futures::{SinkExt, Stream, StreamExt, lock::BiLock, ready};
|
||||
use pin_project_lite::pin_project;
|
||||
@@ -180,12 +180,13 @@ impl ZCPacketToBytes for TunZCPacketToBytes {
|
||||
assert!(payload_offset >= 4);
|
||||
|
||||
let ret = if self.has_packet_info {
|
||||
let mut inner = inner.split_off(payload_offset - 4);
|
||||
inner.advance(payload_offset - 4);
|
||||
let proto = infer_proto(&inner[4..]);
|
||||
self.fill_packet_info(&mut inner[0..4], proto)?;
|
||||
inner
|
||||
} else {
|
||||
inner.split_off(payload_offset)
|
||||
inner.advance(payload_offset);
|
||||
inner
|
||||
};
|
||||
|
||||
tracing::debug!(?ret, ?payload_offset, "convert zc packet to tun packet");
|
||||
|
||||
@@ -5,22 +5,6 @@ use std::io;
|
||||
use clap::Command;
|
||||
use clap_complete::{Generator, Shell};
|
||||
|
||||
// When the `hotpath` feature is off, alias the current crate as `hotpath` so
|
||||
// call sites keep using `hotpath::...` paths, and provide a local no-op shim
|
||||
// for the profiling macros. This keeps `hotpath` an optional dependency: the
|
||||
// profiler is absent from the dependency graph entirely in default builds.
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
extern crate self as hotpath;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
mod hotpath_off;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
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.
|
||||
|
||||
@@ -56,12 +56,11 @@ type BoxNicPacketFilter = Box<dyn NicPacketFilter + Send + Sync>;
|
||||
// pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
|
||||
// tachyonix::channel(128)
|
||||
// }
|
||||
pub type PacketRecvChan = hotpath::wrap::tokio::sync::mpsc::Sender<ZCPacket>;
|
||||
pub type PacketRecvChanReceiver = hotpath::wrap::tokio::sync::mpsc::Receiver<ZCPacket>;
|
||||
pub type PacketRecvChan = tokio::sync::mpsc::Sender<ZCPacket>;
|
||||
pub type PacketRecvChanReceiver = tokio::sync::mpsc::Receiver<ZCPacket>;
|
||||
pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
|
||||
hotpath::channel!(tokio::sync::mpsc::channel(128))
|
||||
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,9 +2,6 @@ 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};
|
||||
@@ -59,7 +56,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(hotpath::rw_lock!(parking_lot::RwLock::new(None)));
|
||||
let peer_public_key = Arc::new(RwLock::new(None));
|
||||
let peer_public_key_copy = peer_public_key.clone();
|
||||
|
||||
let conns_copy = conns.clone();
|
||||
@@ -210,7 +207,6 @@ impl Peer {
|
||||
.map(|conn| conn.clone())
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "Peer"))]
|
||||
pub async fn send_msg(&self, msg: ZCPacket) -> Result<(), Error> {
|
||||
let Some(conn) = self.select_conn().await else {
|
||||
return Err(Error::PeerNoConnectionError(self.peer_node_id));
|
||||
|
||||
@@ -11,9 +11,6 @@ use std::{
|
||||
},
|
||||
};
|
||||
|
||||
#[cfg(feature = "hotpath")]
|
||||
use hotpath::wrap::tokio::sync::Mutex;
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
use base64::Engine as _;
|
||||
@@ -384,12 +381,12 @@ impl PeerConn {
|
||||
session_filter,
|
||||
noise_handshake_result: None,
|
||||
|
||||
tunnel: Arc::new(hotpath::mutex!(tokio::sync::Mutex::new(
|
||||
tunnel: Arc::new(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))),
|
||||
recv: Mutex::new(Some(recv)),
|
||||
tunnel_info,
|
||||
|
||||
tasks: JoinSet::new(),
|
||||
@@ -1466,7 +1463,6 @@ impl PeerConn {
|
||||
});
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerConn"))]
|
||||
pub async fn send_msg(&self, msg: ZCPacket) -> Result<(), Error> {
|
||||
Ok(self.sink.send(msg).await?)
|
||||
}
|
||||
|
||||
@@ -11,9 +11,6 @@ use std::{
|
||||
time::{Duration, SystemTime},
|
||||
};
|
||||
|
||||
#[cfg(feature = "hotpath")]
|
||||
use hotpath::wrap::tokio::sync::{Mutex, RwLock};
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
use tokio::sync::{Mutex, RwLock};
|
||||
use tokio::{
|
||||
sync::mpsc::{self, UnboundedReceiver, UnboundedSender},
|
||||
@@ -279,8 +276,8 @@ impl PeerManager {
|
||||
let rpc_tspt = Arc::new(RpcTransport {
|
||||
my_peer_id,
|
||||
peers: Arc::downgrade(&peers),
|
||||
foreign_peers: hotpath::mutex!(tokio::sync::Mutex::new(None)),
|
||||
packet_recv: hotpath::mutex!(tokio::sync::Mutex::new(peer_rpc_tspt_recv)),
|
||||
foreign_peers: Mutex::new(None),
|
||||
packet_recv: Mutex::new(peer_rpc_tspt_recv),
|
||||
peer_rpc_tspt_sender,
|
||||
encryptor: encryptor.clone(),
|
||||
is_secure_mode_enabled,
|
||||
@@ -412,21 +409,17 @@ impl PeerManager {
|
||||
global_ctx,
|
||||
nic_channel,
|
||||
|
||||
tasks: hotpath::mutex!(tokio::sync::Mutex::new(JoinSet::new())),
|
||||
tasks: Mutex::new(JoinSet::new()),
|
||||
|
||||
packet_recv: Arc::new(hotpath::mutex!(tokio::sync::Mutex::new(Some(packet_recv)))),
|
||||
packet_recv: Arc::new(Mutex::new(Some(packet_recv))),
|
||||
|
||||
peers,
|
||||
|
||||
peer_rpc_mgr,
|
||||
peer_rpc_tspt: rpc_tspt,
|
||||
|
||||
peer_packet_process_pipeline: Arc::new(hotpath::rw_lock!(tokio::sync::RwLock::new(
|
||||
Vec::new()
|
||||
))),
|
||||
nic_packet_process_pipeline: Arc::new(hotpath::rw_lock!(tokio::sync::RwLock::new(
|
||||
Vec::new()
|
||||
))),
|
||||
peer_packet_process_pipeline: Arc::new(RwLock::new(Vec::new())),
|
||||
nic_packet_process_pipeline: Arc::new(RwLock::new(Vec::new())),
|
||||
|
||||
route_algo_inst,
|
||||
|
||||
@@ -437,7 +430,7 @@ impl PeerManager {
|
||||
encryptor,
|
||||
data_compress_algo,
|
||||
|
||||
exit_nodes: hotpath::rw_lock!(tokio::sync::RwLock::new(exit_nodes)),
|
||||
exit_nodes: RwLock::new(exit_nodes),
|
||||
|
||||
reserved_my_peer_id_map: DashMap::new(),
|
||||
recent_have_traffic: Arc::new(DashMap::new()),
|
||||
@@ -963,7 +956,6 @@ 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;
|
||||
@@ -1445,7 +1437,6 @@ impl PeerManager {
|
||||
self.get_route().get_foreign_network_summary().await
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
|
||||
async fn run_nic_packet_process_pipeline(&self, data: &mut ZCPacket) -> bool {
|
||||
// Enforce ACL for outbound (NIC-originated) packets. If ACL denies, stop processing.
|
||||
if !self.global_ctx.get_acl_filter().process_packet_with_acl(
|
||||
@@ -1531,7 +1522,6 @@ impl PeerManager {
|
||||
result
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
|
||||
async fn send_msg_internal(
|
||||
peers: &Arc<PeerMap>,
|
||||
foreign_network_client: &Arc<ForeignNetworkClient>,
|
||||
@@ -1698,7 +1688,6 @@ 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>,
|
||||
|
||||
@@ -6,9 +6,6 @@ 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::{
|
||||
@@ -48,7 +45,7 @@ impl PeerMap {
|
||||
my_peer_id,
|
||||
peer_map: DashMap::new(),
|
||||
packet_send,
|
||||
routes: hotpath::rw_lock!(tokio::sync::RwLock::new(Vec::new())),
|
||||
routes: RwLock::new(Vec::new()),
|
||||
alive_client_urls: Arc::new(Mutex::new(multimap::MultiMap::new())),
|
||||
}
|
||||
}
|
||||
@@ -135,7 +132,6 @@ impl PeerMap {
|
||||
peer_id == self.my_peer_id || self.peer_map.contains_key(&peer_id)
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerMap"))]
|
||||
pub async fn send_msg_directly(&self, msg: ZCPacket, dst_peer_id: PeerId) -> Result<(), Error> {
|
||||
if dst_peer_id == self.my_peer_id {
|
||||
let packet_send = self.packet_send.clone();
|
||||
@@ -167,7 +163,6 @@ impl PeerMap {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerMap"))]
|
||||
pub async fn get_gateway_peer_id(
|
||||
&self,
|
||||
dst_peer_id: PeerId,
|
||||
|
||||
@@ -1394,7 +1394,6 @@ impl RouteTable {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RouteTable"))]
|
||||
fn get_next_hop(&self, dst_peer_id: PeerId) -> Option<NextHopInfo> {
|
||||
if self.suppressed_peer_ids.contains_key(&dst_peer_id) {
|
||||
return None;
|
||||
@@ -1402,7 +1401,6 @@ impl RouteTable {
|
||||
self.get_topology_next_hop(dst_peer_id)
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RouteTable"))]
|
||||
fn get_topology_next_hop(&self, dst_peer_id: PeerId) -> Option<NextHopInfo> {
|
||||
let cur_version = self.next_hop_map_version.get();
|
||||
self.next_hop_map.get(&dst_peer_id).and_then(|x| {
|
||||
|
||||
@@ -1,11 +1,7 @@
|
||||
use std::sync::{
|
||||
Arc,
|
||||
Arc, RwLock,
|
||||
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;
|
||||
@@ -266,7 +262,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.read().unwrap())
|
||||
.field("peer_static_pubkey", &self.peer_static_pubkey)
|
||||
.field("datagram", &self.datagram)
|
||||
.finish()
|
||||
}
|
||||
@@ -286,7 +282,7 @@ impl PeerSession {
|
||||
) -> Self {
|
||||
Self {
|
||||
peer_id,
|
||||
peer_static_pubkey: hotpath::rw_lock!(std::sync::RwLock::new(peer_static_pubkey)),
|
||||
peer_static_pubkey: RwLock::new(peer_static_pubkey),
|
||||
datagram: SecureDatagramSession::new(
|
||||
root_key,
|
||||
session_generation,
|
||||
@@ -380,7 +376,6 @@ impl PeerSession {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerSession"))]
|
||||
pub fn encrypt_payload(
|
||||
&self,
|
||||
sender_peer_id: PeerId,
|
||||
@@ -394,7 +389,6 @@ impl PeerSession {
|
||||
.encrypt_payload(Self::dir_for_sender(sender_peer_id, receiver_peer_id), pkt)
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerSession"))]
|
||||
pub fn decrypt_payload(
|
||||
&self,
|
||||
sender_peer_id: PeerId,
|
||||
|
||||
@@ -144,7 +144,6 @@ impl RelayPeerMap {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
|
||||
async fn send_via_next_hop(
|
||||
&self,
|
||||
msg: ZCPacket,
|
||||
@@ -167,7 +166,6 @@ impl RelayPeerMap {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
|
||||
pub async fn send_msg(
|
||||
self: &Arc<Self>,
|
||||
mut msg: ZCPacket,
|
||||
@@ -615,7 +613,6 @@ 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,16 +1,11 @@
|
||||
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 _};
|
||||
@@ -19,7 +14,7 @@ use sha2::Sha256;
|
||||
use zerocopy::FromBytes;
|
||||
|
||||
use crate::{
|
||||
peers::encrypt::{create_encryptor, Encryptor},
|
||||
peers::encrypt::{Encryptor, create_encryptor},
|
||||
tunnel::packet_def::{StandardAeadTail, ZCPacket},
|
||||
};
|
||||
|
||||
@@ -233,15 +228,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.read().unwrap())
|
||||
.field("root_key", &self.root_key)
|
||||
.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.lock().unwrap())
|
||||
.field("key_cache", &*self.key_cache.lock().unwrap())
|
||||
.field("sync_rx_grace", &*self.sync_rx_grace.lock().unwrap())
|
||||
.field("rx_slots", &self.rx_slots)
|
||||
.field("key_cache", &self.key_cache)
|
||||
.field("sync_rx_grace", &self.sync_rx_grace)
|
||||
.field(
|
||||
"sync_rx_grace_expires_at_ms",
|
||||
&self.sync_rx_grace_expires_at_ms,
|
||||
@@ -277,15 +272,15 @@ impl SecureDatagramSession {
|
||||
];
|
||||
let now_ms = now_ms();
|
||||
Self {
|
||||
root_key: hotpath::rw_lock!(std::sync::RwLock::new(root_key)),
|
||||
root_key: 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: 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())),
|
||||
rx_slots: Mutex::new(rx_slots),
|
||||
key_cache: Mutex::new(key_cache),
|
||||
sync_rx_grace: Mutex::new(SyncRxGrace::default()),
|
||||
sync_rx_grace_expires_at_ms: AtomicU64::new(0),
|
||||
send_cipher_algorithm,
|
||||
recv_cipher_algorithm,
|
||||
@@ -706,10 +701,6 @@ impl SecureDatagramSession {
|
||||
false
|
||||
}
|
||||
|
||||
#[cfg_attr(
|
||||
feature = "hotpath",
|
||||
hotpath::measure(impl_type = "SecureDatagramSession")
|
||||
)]
|
||||
pub fn encrypt_payload(
|
||||
&self,
|
||||
dir: SecureDatagramDirection,
|
||||
@@ -728,10 +719,6 @@ impl SecureDatagramSession {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg_attr(
|
||||
feature = "hotpath",
|
||||
hotpath::measure(impl_type = "SecureDatagramSession")
|
||||
)]
|
||||
pub fn decrypt_payload(
|
||||
&self,
|
||||
dir: SecureDatagramDirection,
|
||||
@@ -897,9 +884,11 @@ 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);
|
||||
@@ -1001,13 +990,9 @@ mod tests {
|
||||
|
||||
s.sync_root_key(root_key, 2, 2, true);
|
||||
assert!(s.check_replay_for_test(2, 0, SecureDatagramDirection::AToB, now + 2));
|
||||
let expires_at = s.sync_rx_grace_expires_at_ms.load(Ordering::Relaxed);
|
||||
|
||||
assert!(!s.check_replay_for_test(
|
||||
0,
|
||||
1,
|
||||
SecureDatagramDirection::AToB,
|
||||
now + SecureDatagramSession::SYNC_RX_GRACE_AFTER_MS + 3
|
||||
));
|
||||
assert!(!s.check_replay_for_test(0, 1, SecureDatagramDirection::AToB, expires_at + 1));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -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) = crate::peers::create_packet_recv_chan();
|
||||
let (packet_sender, _packet_receiver) = tokio::sync::mpsc::channel(100);
|
||||
let peer_mgr = crate::peers::peer_manager::PeerManager::new(
|
||||
RouteAlgoType::Ospf,
|
||||
global_ctx.clone(),
|
||||
|
||||
@@ -48,13 +48,9 @@ use std::collections::{HashMap, HashSet};
|
||||
use std::fmt;
|
||||
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
|
||||
use std::sync::{
|
||||
Arc,
|
||||
Arc, RwLock,
|
||||
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;
|
||||
@@ -89,7 +85,7 @@ impl AddrTuple {
|
||||
|
||||
#[derive(Default)]
|
||||
struct StackState {
|
||||
tuples: HashMap<AddrTuple, hotpath::wrap::flume::Sender<Bytes>>,
|
||||
tuples: HashMap<AddrTuple, flume::Sender<Bytes>>,
|
||||
closed: bool,
|
||||
}
|
||||
|
||||
@@ -133,7 +129,7 @@ pub enum State {
|
||||
pub struct Socket {
|
||||
shared: Arc<Shared>,
|
||||
tun: Arc<dyn Tun>,
|
||||
incoming: hotpath::wrap::flume::Receiver<Bytes>,
|
||||
incoming: flume::Receiver<Bytes>,
|
||||
local_addr: SocketAddr,
|
||||
remote_addr: SocketAddr,
|
||||
local_mac: MacAddr,
|
||||
@@ -162,8 +158,8 @@ impl Socket {
|
||||
remote_mac: Option<MacAddr>,
|
||||
ack: Option<u32>,
|
||||
state: State,
|
||||
) -> (Socket, hotpath::wrap::flume::Sender<Bytes>) {
|
||||
let (incoming_tx, incoming_rx) = hotpath::channel!(flume::bounded(MPMC_BUFFER_LEN));
|
||||
) -> (Socket, flume::Sender<Bytes>) {
|
||||
let (incoming_tx, incoming_rx) = flume::bounded(MPMC_BUFFER_LEN);
|
||||
|
||||
(
|
||||
Socket {
|
||||
@@ -434,9 +430,9 @@ impl Stack {
|
||||
) -> Stack {
|
||||
let (tuples_purge_tx, _tuples_purge_rx) = broadcast::channel(16);
|
||||
let shared = Arc::new(Shared {
|
||||
state: hotpath::rw_lock!(std::sync::RwLock::new(StackState::default())),
|
||||
state: RwLock::new(StackState::default()),
|
||||
tun: tun.clone(),
|
||||
listening: hotpath::rw_lock!(std::sync::RwLock::new(HashSet::new())),
|
||||
listening: RwLock::new(HashSet::new()),
|
||||
tuples_purge: tuples_purge_tx.clone(),
|
||||
});
|
||||
|
||||
@@ -505,7 +501,7 @@ impl Stack {
|
||||
shared: Arc<Shared>,
|
||||
mut tuples_purge: broadcast::Receiver<AddrTuple>,
|
||||
) {
|
||||
let mut tuples: HashMap<AddrTuple, hotpath::wrap::flume::Sender<Bytes>> = HashMap::new();
|
||||
let mut tuples: HashMap<AddrTuple, flume::Sender<Bytes>> = HashMap::new();
|
||||
|
||||
loop {
|
||||
let mut buf = BytesMut::new();
|
||||
|
||||
@@ -9,8 +9,7 @@ use crate::proto::common::TunnelInfo;
|
||||
|
||||
use super::{Tunnel, TunnelError, ZCPacketSink, ZCPacketStream, packet_def::ZCPacket};
|
||||
|
||||
use hotpath::wrap::tokio::sync::mpsc::{Receiver, Sender};
|
||||
use tokio::sync::mpsc::{channel, error::TrySendError};
|
||||
use tokio::sync::mpsc::{Receiver, Sender, channel, error::TrySendError};
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
// use tachyonix::{channel, Receiver, Sender, TrySendError};
|
||||
|
||||
@@ -20,7 +19,6 @@ 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(())
|
||||
@@ -45,7 +43,7 @@ pub struct MpscTunnel<T> {
|
||||
|
||||
impl<T: Tunnel> MpscTunnel<T> {
|
||||
pub fn new(tunnel: T, send_timeout: Option<Duration>) -> Self {
|
||||
let (tx, mut rx) = hotpath::channel!(channel(32));
|
||||
let (tx, mut rx) = channel(32);
|
||||
let (stream, mut sink) = tunnel.split();
|
||||
|
||||
let task = tokio::spawn(async move {
|
||||
@@ -68,7 +66,6 @@ impl<T: Tunnel> MpscTunnel<T> {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))]
|
||||
async fn forward_one_round(
|
||||
rx: &mut Receiver<ZCPacket>,
|
||||
sink: &mut Pin<Box<dyn ZCPacketSink>>,
|
||||
@@ -82,7 +79,6 @@ impl<T: Tunnel> MpscTunnel<T> {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))]
|
||||
async fn forward_one_round_no_timeout(
|
||||
rx: &mut Receiver<ZCPacket>,
|
||||
sink: &mut Pin<Box<dyn ZCPacketSink>>,
|
||||
@@ -100,7 +96,6 @@ impl<T: Tunnel> MpscTunnel<T> {
|
||||
sink.flush().await
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))]
|
||||
async fn forward_one_round_with_timeout(
|
||||
rx: &mut Receiver<ZCPacket>,
|
||||
sink: &mut Pin<Box<dyn ZCPacketSink>>,
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
use bytes::Buf;
|
||||
use bytes::Bytes;
|
||||
use bytes::BytesMut;
|
||||
use zerocopy::byteorder::*;
|
||||
use zerocopy::AsBytes;
|
||||
use zerocopy::FromBytes;
|
||||
use zerocopy::FromZeroes;
|
||||
use zerocopy::byteorder::*;
|
||||
|
||||
type DefaultEndian = LittleEndian;
|
||||
|
||||
@@ -485,8 +486,16 @@ impl ZCPacket {
|
||||
let payload_off = ret.packet_type.get_packet_offsets().payload_offset;
|
||||
let total_len = payload_off + payload.len();
|
||||
ret.inner.reserve(total_len);
|
||||
unsafe { ret.inner.set_len(total_len) };
|
||||
ret.mut_payload()[..payload.len()].copy_from_slice(payload);
|
||||
|
||||
// SAFETY: `reserve` guarantees capacity >= total_len.
|
||||
// We zero the header region and copy payload before advancing length,
|
||||
// so every byte in [0..total_len) is initialized before any read.
|
||||
unsafe {
|
||||
let ptr = ret.inner.as_mut_ptr();
|
||||
std::ptr::write_bytes(ptr, 0, payload_off);
|
||||
std::ptr::copy_nonoverlapping(payload.as_ptr(), ptr.add(payload_off), payload.len());
|
||||
ret.inner.set_len(total_len);
|
||||
}
|
||||
ret
|
||||
}
|
||||
|
||||
@@ -494,12 +503,12 @@ impl ZCPacket {
|
||||
let mut ret = Self::new_nic_packet();
|
||||
ret.inner.reserve(cap);
|
||||
let total_len = ret.packet_type.get_packet_offsets().payload_offset - packet_info_len;
|
||||
unsafe { ret.inner.set_len(total_len) };
|
||||
ret.inner.resize(total_len, 0);
|
||||
ret
|
||||
}
|
||||
|
||||
pub fn new_for_foreign_network(
|
||||
network_name: &String,
|
||||
network_name: &str,
|
||||
dst_peer_id: u32,
|
||||
foreign_zc_packet: &ZCPacket,
|
||||
) -> Self {
|
||||
@@ -508,26 +517,71 @@ impl ZCPacket {
|
||||
foreign_network_hdr.get_header_len() + foreign_zc_packet.tunnel_payload().len();
|
||||
|
||||
let mut ret = Self::new_nic_packet();
|
||||
let payload_off = ret.packet_type.get_packet_offsets().payload_offset;
|
||||
ret.inner.reserve(payload_off + total_payload_len);
|
||||
unsafe { ret.inner.set_len(payload_off + total_payload_len) };
|
||||
let offsets = ret.packet_type.get_packet_offsets();
|
||||
let payload_off = offsets.payload_offset;
|
||||
let pm_hdr_off = offsets.peer_manager_header_offset;
|
||||
let total_len = payload_off + total_payload_len;
|
||||
ret.inner.reserve(total_len);
|
||||
|
||||
let fixed_hdr_len = std::mem::size_of::<ForeignNetworkPacketHeader>();
|
||||
ret.mut_payload()[..fixed_hdr_len].copy_from_slice(foreign_network_hdr.as_bytes());
|
||||
|
||||
let name_offset = foreign_network_hdr.network_name_offset.get() as usize;
|
||||
let name_len = foreign_network_hdr.network_name_len.get() as usize;
|
||||
ret.mut_payload()[name_offset..name_offset + name_len]
|
||||
.copy_from_slice(network_name.as_bytes());
|
||||
let foreign_payload = foreign_zc_packet.tunnel_payload();
|
||||
|
||||
ret.mut_payload()[foreign_network_hdr.get_header_len()..]
|
||||
.copy_from_slice(foreign_zc_packet.tunnel_payload());
|
||||
// Construct the PeerManagerHeader on the stack so we can write it
|
||||
// directly into the buffer, avoiding a separate mut_peer_manager_header()
|
||||
// call after set_len.
|
||||
let pm_hdr = PeerManagerHeader {
|
||||
from_peer_id: 0.into(),
|
||||
to_peer_id: 0.into(),
|
||||
packet_type: PacketType::ForeignNetworkPacket as u8,
|
||||
flags: 0,
|
||||
forward_counter: 0,
|
||||
reserved: 0,
|
||||
len: U32::new(total_payload_len as u32),
|
||||
};
|
||||
|
||||
let hdr = ret.mut_peer_manager_header().unwrap();
|
||||
hdr.from_peer_id = 0.into();
|
||||
hdr.to_peer_id = 0.into();
|
||||
hdr.packet_type = PacketType::ForeignNetworkPacket as u8;
|
||||
hdr.len.set(total_payload_len as u32);
|
||||
// SAFETY: `reserve` guarantees capacity >= total_len.
|
||||
// We zero only the tunnel-header reserved space [0..pm_hdr_off], write
|
||||
// the PeerManagerHeader directly at pm_hdr_off, then copy the foreign
|
||||
// network header, network name, and payload. Every byte in [0..total_len)
|
||||
// is initialized before set_len.
|
||||
unsafe {
|
||||
let ptr = ret.inner.as_mut_ptr();
|
||||
|
||||
// Zero the tunnel header reserved space only (not the PM header region)
|
||||
std::ptr::write_bytes(ptr, 0, pm_hdr_off);
|
||||
|
||||
// Write PeerManagerHeader directly
|
||||
std::ptr::copy_nonoverlapping(
|
||||
pm_hdr.as_bytes().as_ptr(),
|
||||
ptr.add(pm_hdr_off),
|
||||
std::mem::size_of::<PeerManagerHeader>(),
|
||||
);
|
||||
|
||||
// Copy foreign network fixed header
|
||||
std::ptr::copy_nonoverlapping(
|
||||
foreign_network_hdr.as_bytes().as_ptr(),
|
||||
ptr.add(payload_off),
|
||||
fixed_hdr_len,
|
||||
);
|
||||
|
||||
// Copy network name
|
||||
std::ptr::copy_nonoverlapping(
|
||||
network_name.as_ptr(),
|
||||
ptr.add(payload_off + name_offset),
|
||||
name_len,
|
||||
);
|
||||
|
||||
// Copy foreign payload
|
||||
std::ptr::copy_nonoverlapping(
|
||||
foreign_payload.as_ptr(),
|
||||
ptr.add(payload_off + foreign_network_hdr.get_header_len()),
|
||||
foreign_payload.len(),
|
||||
);
|
||||
|
||||
ret.inner.set_len(total_len);
|
||||
}
|
||||
|
||||
ret
|
||||
}
|
||||
@@ -587,7 +641,8 @@ impl ZCPacket {
|
||||
}
|
||||
|
||||
pub fn payload_bytes(mut self) -> BytesMut {
|
||||
self.inner.split_off(self.payload_offset())
|
||||
self.inner.advance(self.payload_offset());
|
||||
self.inner
|
||||
}
|
||||
|
||||
pub fn peer_manager_header(&self) -> Option<&PeerManagerHeader> {
|
||||
@@ -652,11 +707,12 @@ impl ZCPacket {
|
||||
}
|
||||
|
||||
pub fn tunnel_payload_bytes(mut self) -> BytesMut {
|
||||
self.inner.split_off(
|
||||
self.inner.advance(
|
||||
self.packet_type
|
||||
.get_packet_offsets()
|
||||
.peer_manager_header_offset,
|
||||
)
|
||||
);
|
||||
self.inner
|
||||
}
|
||||
|
||||
pub fn convert_type(mut self, target_packet_type: ZCPacketType) -> Self {
|
||||
@@ -697,12 +753,13 @@ impl ZCPacket {
|
||||
.get_packet_offsets()
|
||||
.peer_manager_header_offset;
|
||||
let mut buf = BytesMut::with_capacity(new_pm_offset + tunnel_payload.len());
|
||||
unsafe { buf.set_len(new_pm_offset) };
|
||||
buf.resize(new_pm_offset, 0);
|
||||
buf.extend_from_slice(tunnel_payload);
|
||||
return Self::new_from_buf(buf, target_packet_type);
|
||||
}
|
||||
|
||||
Self::new_from_buf(self.inner.split_off(new_offset), target_packet_type)
|
||||
self.inner.advance(new_offset);
|
||||
Self::new_from_buf(self.inner, target_packet_type)
|
||||
}
|
||||
|
||||
pub fn into_bytes(self) -> Bytes {
|
||||
@@ -748,8 +805,10 @@ impl ZCPacket {
|
||||
let foreign_hdr_len = hdr.get_header_len();
|
||||
|
||||
Self::new_from_buf(
|
||||
self.inner
|
||||
.split_off(foreign_hdr_len + self.payload_offset()),
|
||||
{
|
||||
self.inner.advance(foreign_hdr_len + self.payload_offset());
|
||||
self.inner
|
||||
},
|
||||
ZCPacketType::DummyTunnel,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -12,9 +12,6 @@ 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,
|
||||
@@ -315,25 +312,18 @@ 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: hotpath::rw_lock!(parking_lot::RwLock::new(RwPoolInner::default())),
|
||||
persistent: hotpath::rw_lock!(parking_lot::RwLock::new(RwPoolInner::default())),
|
||||
ephemeral: RwLock::new(RwPoolInner::default()),
|
||||
persistent: RwLock::new(RwPoolInner::default()),
|
||||
capacity,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,8 +11,7 @@ use async_trait::async_trait;
|
||||
use futures::{Sink, SinkExt, Stream, StreamExt};
|
||||
use once_cell::sync::Lazy;
|
||||
|
||||
use hotpath::wrap::tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
|
||||
use tokio::sync::mpsc::unbounded_channel;
|
||||
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel};
|
||||
|
||||
use uuid::Uuid;
|
||||
|
||||
@@ -197,7 +196,7 @@ pub struct RingTunnelListener {
|
||||
|
||||
impl RingTunnelListener {
|
||||
pub fn new(key: url::Url) -> Self {
|
||||
let (conn_sender, conn_receiver) = hotpath::channel!(unbounded_channel());
|
||||
let (conn_sender, conn_receiver) = unbounded_channel();
|
||||
RingTunnelListener {
|
||||
listener_addr: key,
|
||||
conn_sender,
|
||||
|
||||
@@ -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.load(Relaxed)),
|
||||
rx_bytes: AtomicU64::new(self.rx_bytes.load(Relaxed)),
|
||||
tx_packets: AtomicU64::new(self.tx_packets.load(Relaxed)),
|
||||
rx_packets: AtomicU64::new(self.rx_packets.load(Relaxed)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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,28 @@ 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,10 +13,11 @@ 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::{channel, unbounded_channel},
|
||||
sync::mpsc::{
|
||||
Receiver, Sender, UnboundedReceiver, UnboundedSender, channel, unbounded_channel,
|
||||
},
|
||||
task::JoinSet,
|
||||
};
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
@@ -292,7 +293,6 @@ 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>,
|
||||
@@ -327,7 +327,6 @@ 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,
|
||||
@@ -396,7 +395,6 @@ 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();
|
||||
@@ -543,7 +541,6 @@ 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 {
|
||||
@@ -650,7 +647,6 @@ 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();
|
||||
@@ -679,8 +675,8 @@ pub struct UdpTunnelListener {
|
||||
|
||||
impl UdpTunnelListener {
|
||||
pub fn new(addr: url::Url) -> Self {
|
||||
let (close_event_send, close_event_recv) = hotpath::channel!(unbounded_channel());
|
||||
let (conn_send, conn_recv) = hotpath::channel!(channel(100));
|
||||
let (close_event_send, close_event_recv) = unbounded_channel();
|
||||
let (conn_send, conn_recv) = channel(100);
|
||||
Self {
|
||||
addr: addr.clone(),
|
||||
socket: None,
|
||||
@@ -920,8 +916,7 @@ impl UdpTunnelConnector {
|
||||
"udp build tunnel for connector"
|
||||
);
|
||||
|
||||
let (close_event_sender, mut close_event_recv) =
|
||||
hotpath::channel!(unbounded_channel());
|
||||
let (close_event_sender, mut close_event_recv) = unbounded_channel();
|
||||
|
||||
let ring_recv = RingStream::new(ring_for_send_udp.clone());
|
||||
let ring_sender = RingSink::new(ring_for_recv_udp.clone());
|
||||
@@ -1184,7 +1179,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) = hotpath::channel!(tokio::sync::mpsc::unbounded_channel());
|
||||
let (close_event_sender, _close_event_recv) = tokio::sync::mpsc::unbounded_channel();
|
||||
let mut conn = UdpConnection::new(
|
||||
socket,
|
||||
7,
|
||||
|
||||
@@ -37,17 +37,9 @@ use crossbeam::atomic::AtomicCell;
|
||||
use dashmap::DashMap;
|
||||
use futures::{SinkExt, StreamExt, stream::FuturesUnordered};
|
||||
use rand::RngCore;
|
||||
#[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,
|
||||
sync::{Mutex, mpsc::unbounded_channel},
|
||||
task::JoinSet,
|
||||
};
|
||||
|
||||
@@ -355,7 +347,7 @@ struct WgPeer {
|
||||
config: WgConfig,
|
||||
endpoint: SocketAddr,
|
||||
|
||||
sink: StdMutex<Option<Pin<Box<dyn ZCPacketSink>>>>,
|
||||
sink: std::sync::Mutex<Option<Pin<Box<dyn ZCPacketSink>>>>,
|
||||
|
||||
data: Option<WgPeerData>,
|
||||
tasks: JoinSet<()>,
|
||||
@@ -366,19 +358,19 @@ struct WgPeer {
|
||||
impl WgPeer {
|
||||
fn new(udp: Arc<UdpSocket>, config: WgConfig, endpoint: SocketAddr) -> Self {
|
||||
WgPeer {
|
||||
tunn: Some(hotpath::mutex!(tokio::sync::Mutex::new(Tunn::new(
|
||||
tunn: Some(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: hotpath::mutex!(std::sync::Mutex::new(None)),
|
||||
sink: std::sync::Mutex::new(None),
|
||||
|
||||
data: None,
|
||||
tasks: JoinSet::new(),
|
||||
@@ -387,7 +379,6 @@ 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;
|
||||
@@ -399,7 +390,6 @@ 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());
|
||||
@@ -465,8 +455,8 @@ impl WgPeer {
|
||||
}
|
||||
}
|
||||
|
||||
type ConnSender = hotpath::wrap::tokio::sync::mpsc::UnboundedSender<Box<dyn Tunnel>>;
|
||||
type ConnReceiver = hotpath::wrap::tokio::sync::mpsc::UnboundedReceiver<Box<dyn Tunnel>>;
|
||||
type ConnSender = tokio::sync::mpsc::UnboundedSender<Box<dyn Tunnel>>;
|
||||
type ConnReceiver = tokio::sync::mpsc::UnboundedReceiver<Box<dyn Tunnel>>;
|
||||
|
||||
pub struct WgTunnelListener {
|
||||
addr: url::Url,
|
||||
@@ -484,7 +474,7 @@ pub struct WgTunnelListener {
|
||||
|
||||
impl WgTunnelListener {
|
||||
pub fn new(addr: url::Url, config: WgConfig) -> Self {
|
||||
let (conn_send, conn_recv) = hotpath::channel!(unbounded_channel());
|
||||
let (conn_send, conn_recv) = unbounded_channel();
|
||||
WgTunnelListener {
|
||||
addr,
|
||||
config,
|
||||
|
||||
Reference in New Issue
Block a user