mirror of
https://github.com/EasyTier/EasyTier.git
synced 2026-10-29 16:05:35 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c5041aec2d | ||
|
|
7fc954a6dc | ||
|
|
cf3087d7ca | ||
|
|
7a8c7356fa | ||
|
|
f24735a86f |
Generated
-146
@@ -263,12 +263,6 @@ version = "0.7.6"
|
|||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
|
checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "ascii"
|
|
||||||
version = "1.1.0"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "d92bec98840b8f03a5ff5413de5293bfcd8bf96467cf5452609f939ec6f5de16"
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "async-broadcast"
|
name = "async-broadcast"
|
||||||
version = "0.7.2"
|
version = "0.7.2"
|
||||||
@@ -1266,12 +1260,6 @@ dependencies = [
|
|||||||
"windows-targets 0.52.6",
|
"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]]
|
[[package]]
|
||||||
name = "ciborium"
|
name = "ciborium"
|
||||||
version = "0.2.2"
|
version = "0.2.2"
|
||||||
@@ -2175,16 +2163,6 @@ dependencies = [
|
|||||||
"dirs-sys 0.5.0",
|
"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]]
|
[[package]]
|
||||||
name = "dirs-sys"
|
name = "dirs-sys"
|
||||||
version = "0.3.7"
|
version = "0.3.7"
|
||||||
@@ -2208,17 +2186,6 @@ dependencies = [
|
|||||||
"windows-sys 0.61.2",
|
"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]]
|
[[package]]
|
||||||
name = "dispatch2"
|
name = "dispatch2"
|
||||||
version = "0.3.1"
|
version = "0.3.1"
|
||||||
@@ -2398,7 +2365,6 @@ dependencies = [
|
|||||||
"hickory-resolver",
|
"hickory-resolver",
|
||||||
"hickory-server",
|
"hickory-server",
|
||||||
"hmac",
|
"hmac",
|
||||||
"hotpath",
|
|
||||||
"http",
|
"http",
|
||||||
"http_req",
|
"http_req",
|
||||||
"humansize",
|
"humansize",
|
||||||
@@ -2741,12 +2707,6 @@ version = "1.2.2"
|
|||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "4ef6b89e5b37196644d8796de5268852ff179b44e96276cf4290264843743bb7"
|
checksum = "4ef6b89e5b37196644d8796de5268852ff179b44e96276cf4290264843743bb7"
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "encode_unicode"
|
|
||||||
version = "1.0.0"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0"
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "encoding"
|
name = "encoding"
|
||||||
version = "0.2.33"
|
version = "0.2.33"
|
||||||
@@ -4005,61 +3965,6 @@ dependencies = [
|
|||||||
"windows-sys 0.52.0",
|
"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]]
|
[[package]]
|
||||||
name = "html5ever"
|
name = "html5ever"
|
||||||
version = "0.29.1"
|
version = "0.29.1"
|
||||||
@@ -6028,15 +5933,6 @@ dependencies = [
|
|||||||
"objc2-foundation",
|
"objc2-foundation",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "object"
|
|
||||||
version = "0.36.7"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "62948e14d923ea95ea2c7c86c71013138b66525b86bdc08d2dcc262bdb497b87"
|
|
||||||
dependencies = [
|
|
||||||
"memchr",
|
|
||||||
]
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "once_cell"
|
name = "once_cell"
|
||||||
version = "1.21.3"
|
version = "1.21.3"
|
||||||
@@ -6941,19 +6837,6 @@ dependencies = [
|
|||||||
"syn 2.0.117",
|
"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]]
|
[[package]]
|
||||||
name = "primeorder"
|
name = "primeorder"
|
||||||
version = "0.13.6"
|
version = "0.13.6"
|
||||||
@@ -8013,12 +7896,6 @@ dependencies = [
|
|||||||
"wasm-bindgen",
|
"wasm-bindgen",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "rustc-demangle"
|
|
||||||
version = "0.1.27"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "b50b8869d9fc858ce7266cce0194bd74df58b9d0e3f6df3a9fc8eb470d95c09d"
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "rustc-hash"
|
name = "rustc-hash"
|
||||||
version = "2.1.0"
|
version = "2.1.0"
|
||||||
@@ -9943,17 +9820,6 @@ dependencies = [
|
|||||||
"utf-8",
|
"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]]
|
[[package]]
|
||||||
name = "terminal_size"
|
name = "terminal_size"
|
||||||
version = "0.4.1"
|
version = "0.4.1"
|
||||||
@@ -10109,18 +9975,6 @@ version = "1.0.1"
|
|||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "825f6c8a18bc36d56a62f66af7296385b628c9c5543a8663d4c217fc920bfefd"
|
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]]
|
[[package]]
|
||||||
name = "tinystr"
|
name = "tinystr"
|
||||||
version = "0.7.6"
|
version = "0.7.6"
|
||||||
|
|||||||
@@ -11,8 +11,17 @@ const props = defineProps<{
|
|||||||
|
|
||||||
const list = defineModel<string[]>({ required: true })
|
const list = defineModel<string[]>({ required: true })
|
||||||
|
|
||||||
|
const fallbackUrl = () => {
|
||||||
|
const protoKeys = Object.keys(props.protos)
|
||||||
|
const defaultProto = protoKeys.includes('tcp')
|
||||||
|
? 'tcp'
|
||||||
|
: (protoKeys[0] ?? 'tcp')
|
||||||
|
const defaultPort = props.protos[defaultProto] ?? 11010
|
||||||
|
return `${defaultProto}://0.0.0.0:${defaultPort}`
|
||||||
|
}
|
||||||
|
|
||||||
const addUrl = () => {
|
const addUrl = () => {
|
||||||
list.value.push(props.defaultUrl || 'tcp://0.0.0.0:11010')
|
list.value.push(props.defaultUrl || fallbackUrl())
|
||||||
}
|
}
|
||||||
|
|
||||||
const removeUrl = (index: number) => {
|
const removeUrl = (index: number) => {
|
||||||
|
|||||||
@@ -0,0 +1,93 @@
|
|||||||
|
import { mount } from '@vue/test-utils'
|
||||||
|
import { describe, expect, it } from 'vitest'
|
||||||
|
import { defineComponent, h, nextTick, ref } from 'vue'
|
||||||
|
import UrlListInput from '../src/components/UrlListInput.vue'
|
||||||
|
|
||||||
|
const ButtonStub = defineComponent({
|
||||||
|
name: 'Button',
|
||||||
|
emits: ['click'],
|
||||||
|
setup(_, { slots, emit }) {
|
||||||
|
return () => h('button', { onClick: (event: MouseEvent) => emit('click', event) }, slots.default?.())
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
const UrlInputStub = defineComponent({
|
||||||
|
name: 'UrlInput',
|
||||||
|
setup(_, { slots }) {
|
||||||
|
return () => h('div', slots.actions?.())
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
function mountUrlListInput(protos: Record<string, number>, defaultUrl?: string) {
|
||||||
|
const urls = ref<string[]>([])
|
||||||
|
const wrapper = mount(defineComponent({
|
||||||
|
components: { UrlListInput },
|
||||||
|
setup() {
|
||||||
|
return { urls, protos, defaultUrl }
|
||||||
|
},
|
||||||
|
template: `
|
||||||
|
<UrlListInput
|
||||||
|
v-model="urls"
|
||||||
|
:protos="protos"
|
||||||
|
:default-url="defaultUrl"
|
||||||
|
add-label="add_url"
|
||||||
|
/>
|
||||||
|
`,
|
||||||
|
}), {
|
||||||
|
global: {
|
||||||
|
stubs: {
|
||||||
|
Button: ButtonStub,
|
||||||
|
UrlInput: UrlInputStub,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
return { wrapper, urls }
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('UrlListInput.vue add fallback', () => {
|
||||||
|
it('derives the fallback URL from protos when defaultUrl is not provided', async () => {
|
||||||
|
const { wrapper, urls } = mountUrlListInput({ tcp: 11010, udp: 11010 })
|
||||||
|
|
||||||
|
await wrapper.find('.cursor-pointer').trigger('click')
|
||||||
|
await nextTick()
|
||||||
|
|
||||||
|
expect(urls.value).toEqual(['tcp://0.0.0.0:11010'])
|
||||||
|
})
|
||||||
|
|
||||||
|
it('falls back to the first available protocol when tcp is not present', async () => {
|
||||||
|
const { wrapper, urls } = mountUrlListInput({ udp: 22000 })
|
||||||
|
|
||||||
|
await wrapper.find('.cursor-pointer').trigger('click')
|
||||||
|
await nextTick()
|
||||||
|
|
||||||
|
expect(urls.value).toEqual(['udp://0.0.0.0:22000'])
|
||||||
|
})
|
||||||
|
|
||||||
|
it('falls back to tcp default port when protos is empty', async () => {
|
||||||
|
const { wrapper, urls } = mountUrlListInput({})
|
||||||
|
|
||||||
|
await wrapper.find('.cursor-pointer').trigger('click')
|
||||||
|
await nextTick()
|
||||||
|
|
||||||
|
expect(urls.value).toEqual(['tcp://0.0.0.0:11010'])
|
||||||
|
})
|
||||||
|
|
||||||
|
it('supports port-zero fallback from protos', async () => {
|
||||||
|
const { wrapper, urls } = mountUrlListInput({ tcp: 0, udp: 0 })
|
||||||
|
|
||||||
|
await wrapper.find('.cursor-pointer').trigger('click')
|
||||||
|
await nextTick()
|
||||||
|
|
||||||
|
expect(urls.value).toEqual(['tcp://0.0.0.0:0'])
|
||||||
|
})
|
||||||
|
|
||||||
|
it('uses defaultUrl when provided', async () => {
|
||||||
|
const { wrapper, urls } = mountUrlListInput({ tcp: 11010 }, 'udp://0.0.0.0:22000')
|
||||||
|
|
||||||
|
await wrapper.find('.cursor-pointer').trigger('click')
|
||||||
|
await nextTick()
|
||||||
|
|
||||||
|
expect(urls.value).toEqual(['udp://0.0.0.0:22000'])
|
||||||
|
})
|
||||||
|
})
|
||||||
+10
-11
@@ -32,6 +32,10 @@ path = "src/lib.rs"
|
|||||||
name = "tx_throughput"
|
name = "tx_throughput"
|
||||||
harness = false
|
harness = false
|
||||||
|
|
||||||
|
[[bench]]
|
||||||
|
name = "packet_bytes_extraction"
|
||||||
|
harness = false
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
git-version = "0.3.9"
|
git-version = "0.3.9"
|
||||||
|
|
||||||
@@ -58,8 +62,6 @@ chrono = { version = "0.4.37", features = ["serde"] }
|
|||||||
guarden = "0.2"
|
guarden = "0.2"
|
||||||
quanta = "0.12"
|
quanta = "0.12"
|
||||||
|
|
||||||
hotpath = { version = "0.21", default-features = false, optional = true }
|
|
||||||
|
|
||||||
delegate = "0.13.5"
|
delegate = "0.13.5"
|
||||||
|
|
||||||
itertools = "0.14.0"
|
itertools = "0.14.0"
|
||||||
@@ -412,14 +414,11 @@ tracing = ["tokio/tracing", "dep:console-subscriber"]
|
|||||||
magic-dns = ["dep:hickory-client", "dep:hickory-server"]
|
magic-dns = ["dep:hickory-client", "dep:hickory-server"]
|
||||||
faketcp = ["dep:flume"]
|
faketcp = ["dep:flume"]
|
||||||
zstd = ["dep:zstd"]
|
zstd = ["dep:zstd"]
|
||||||
hotpath = [
|
# Deprecated: hotpath profiling has been removed. These feature aliases are
|
||||||
"dep:hotpath",
|
# retained as no-ops so existing build scripts using `--features hotpath*`
|
||||||
"hotpath/hotpath",
|
# continue to work without pulling in any dependencies.
|
||||||
"hotpath/tokio",
|
hotpath = []
|
||||||
"hotpath/parking_lot",
|
hotpath-cpu = ["hotpath"]
|
||||||
"hotpath/flume",
|
hotpath-alloc = ["hotpath"]
|
||||||
]
|
|
||||||
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
|
|
||||||
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
|
|
||||||
# For Network Extension on macOS
|
# For Network Extension on macOS
|
||||||
macos-ne = []
|
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`).
|
Criterion benchmark for EasyTier's TX injection path (`peer_manager::send_msg_by_ip`).
|
||||||
|
|
||||||
|
|||||||
@@ -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);
|
||||||
@@ -24,16 +24,6 @@ pub static malloc_conf: &[u8] = b"retain:false\0";
|
|||||||
rust_i18n::i18n!("locales", fallback = "en");
|
rust_i18n::i18n!("locales", fallback = "en");
|
||||||
|
|
||||||
#[tokio::main(flavor = "current_thread")]
|
#[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 {
|
async fn main() -> std::process::ExitCode {
|
||||||
core::main().await
|
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 byteorder::WriteBytesExt as _;
|
||||||
use bytes::{BufMut, BytesMut};
|
use bytes::{Buf, BufMut, BytesMut};
|
||||||
use cidr::{Ipv4Inet, Ipv6Inet};
|
use cidr::{Ipv4Inet, Ipv6Inet};
|
||||||
use futures::{SinkExt, Stream, StreamExt, lock::BiLock, ready};
|
use futures::{SinkExt, Stream, StreamExt, lock::BiLock, ready};
|
||||||
use pin_project_lite::pin_project;
|
use pin_project_lite::pin_project;
|
||||||
@@ -180,12 +180,13 @@ impl ZCPacketToBytes for TunZCPacketToBytes {
|
|||||||
assert!(payload_offset >= 4);
|
assert!(payload_offset >= 4);
|
||||||
|
|
||||||
let ret = if self.has_packet_info {
|
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..]);
|
let proto = infer_proto(&inner[4..]);
|
||||||
self.fill_packet_info(&mut inner[0..4], proto)?;
|
self.fill_packet_info(&mut inner[0..4], proto)?;
|
||||||
inner
|
inner
|
||||||
} else {
|
} else {
|
||||||
inner.split_off(payload_offset)
|
inner.advance(payload_offset);
|
||||||
|
inner
|
||||||
};
|
};
|
||||||
|
|
||||||
tracing::debug!(?ret, ?payload_offset, "convert zc packet to tun packet");
|
tracing::debug!(?ret, ?payload_offset, "convert zc packet to tun packet");
|
||||||
|
|||||||
@@ -5,22 +5,6 @@ use std::io;
|
|||||||
use clap::Command;
|
use clap::Command;
|
||||||
use clap_complete::{Generator, Shell};
|
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
|
// 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
|
// (e.g. `Route::get_peer_info_last_update_time`) reference a deliberate
|
||||||
// public type rather than leaking an inaccessible one.
|
// 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) {
|
// pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
|
||||||
// tachyonix::channel(128)
|
// tachyonix::channel(128)
|
||||||
// }
|
// }
|
||||||
pub type PacketRecvChan = hotpath::wrap::tokio::sync::mpsc::Sender<ZCPacket>;
|
pub type PacketRecvChan = tokio::sync::mpsc::Sender<ZCPacket>;
|
||||||
pub type PacketRecvChanReceiver = hotpath::wrap::tokio::sync::mpsc::Receiver<ZCPacket>;
|
pub type PacketRecvChanReceiver = tokio::sync::mpsc::Receiver<ZCPacket>;
|
||||||
pub fn create_packet_recv_chan() -> (PacketRecvChan, PacketRecvChanReceiver) {
|
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(
|
pub async fn recv_packet_from_chan(
|
||||||
packet_recv_chan_receiver: &mut PacketRecvChanReceiver,
|
packet_recv_chan_receiver: &mut PacketRecvChanReceiver,
|
||||||
) -> Result<ZCPacket, anyhow::Error> {
|
) -> Result<ZCPacket, anyhow::Error> {
|
||||||
|
|||||||
@@ -2,9 +2,6 @@ use std::sync::Arc;
|
|||||||
|
|
||||||
use crossbeam::atomic::AtomicCell;
|
use crossbeam::atomic::AtomicCell;
|
||||||
use dashmap::{DashMap, DashSet};
|
use dashmap::{DashMap, DashSet};
|
||||||
#[cfg(feature = "hotpath")]
|
|
||||||
use hotpath::wrap::parking_lot::RwLock;
|
|
||||||
#[cfg(not(feature = "hotpath"))]
|
|
||||||
use parking_lot::RwLock;
|
use parking_lot::RwLock;
|
||||||
|
|
||||||
use tokio::{select, sync::mpsc};
|
use tokio::{select, sync::mpsc};
|
||||||
@@ -59,7 +56,7 @@ impl Peer {
|
|||||||
let shutdown_notifier = Arc::new(tokio::sync::Notify::new());
|
let shutdown_notifier = Arc::new(tokio::sync::Notify::new());
|
||||||
let peer_identity_type = Arc::new(AtomicCell::new(None));
|
let peer_identity_type = Arc::new(AtomicCell::new(None));
|
||||||
let peer_identity_type_copy = peer_identity_type.clone();
|
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 peer_public_key_copy = peer_public_key.clone();
|
||||||
|
|
||||||
let conns_copy = conns.clone();
|
let conns_copy = conns.clone();
|
||||||
@@ -210,7 +207,6 @@ impl Peer {
|
|||||||
.map(|conn| conn.clone())
|
.map(|conn| conn.clone())
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "Peer"))]
|
|
||||||
pub async fn send_msg(&self, msg: ZCPacket) -> Result<(), Error> {
|
pub async fn send_msg(&self, msg: ZCPacket) -> Result<(), Error> {
|
||||||
let Some(conn) = self.select_conn().await else {
|
let Some(conn) = self.select_conn().await else {
|
||||||
return Err(Error::PeerNoConnectionError(self.peer_node_id));
|
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 tokio::sync::Mutex;
|
||||||
|
|
||||||
use base64::Engine as _;
|
use base64::Engine as _;
|
||||||
@@ -384,12 +381,12 @@ impl PeerConn {
|
|||||||
session_filter,
|
session_filter,
|
||||||
noise_handshake_result: None,
|
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()))
|
Box::new(guard!([mut mpsc_tunnel] mpsc_tunnel.close()))
|
||||||
as Box<dyn Any + Send + 'static>,
|
as Box<dyn Any + Send + 'static>,
|
||||||
))),
|
)),
|
||||||
sink,
|
sink,
|
||||||
recv: hotpath::mutex!(tokio::sync::Mutex::new(Some(recv))),
|
recv: Mutex::new(Some(recv)),
|
||||||
tunnel_info,
|
tunnel_info,
|
||||||
|
|
||||||
tasks: JoinSet::new(),
|
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> {
|
pub async fn send_msg(&self, msg: ZCPacket) -> Result<(), Error> {
|
||||||
Ok(self.sink.send(msg).await?)
|
Ok(self.sink.send(msg).await?)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -11,9 +11,6 @@ use std::{
|
|||||||
time::{Duration, SystemTime},
|
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::{Mutex, RwLock};
|
||||||
use tokio::{
|
use tokio::{
|
||||||
sync::mpsc::{self, UnboundedReceiver, UnboundedSender},
|
sync::mpsc::{self, UnboundedReceiver, UnboundedSender},
|
||||||
@@ -279,8 +276,8 @@ impl PeerManager {
|
|||||||
let rpc_tspt = Arc::new(RpcTransport {
|
let rpc_tspt = Arc::new(RpcTransport {
|
||||||
my_peer_id,
|
my_peer_id,
|
||||||
peers: Arc::downgrade(&peers),
|
peers: Arc::downgrade(&peers),
|
||||||
foreign_peers: hotpath::mutex!(tokio::sync::Mutex::new(None)),
|
foreign_peers: Mutex::new(None),
|
||||||
packet_recv: hotpath::mutex!(tokio::sync::Mutex::new(peer_rpc_tspt_recv)),
|
packet_recv: Mutex::new(peer_rpc_tspt_recv),
|
||||||
peer_rpc_tspt_sender,
|
peer_rpc_tspt_sender,
|
||||||
encryptor: encryptor.clone(),
|
encryptor: encryptor.clone(),
|
||||||
is_secure_mode_enabled,
|
is_secure_mode_enabled,
|
||||||
@@ -412,21 +409,17 @@ impl PeerManager {
|
|||||||
global_ctx,
|
global_ctx,
|
||||||
nic_channel,
|
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,
|
peers,
|
||||||
|
|
||||||
peer_rpc_mgr,
|
peer_rpc_mgr,
|
||||||
peer_rpc_tspt: rpc_tspt,
|
peer_rpc_tspt: rpc_tspt,
|
||||||
|
|
||||||
peer_packet_process_pipeline: Arc::new(hotpath::rw_lock!(tokio::sync::RwLock::new(
|
peer_packet_process_pipeline: Arc::new(RwLock::new(Vec::new())),
|
||||||
Vec::new()
|
nic_packet_process_pipeline: Arc::new(RwLock::new(Vec::new())),
|
||||||
))),
|
|
||||||
nic_packet_process_pipeline: Arc::new(hotpath::rw_lock!(tokio::sync::RwLock::new(
|
|
||||||
Vec::new()
|
|
||||||
))),
|
|
||||||
|
|
||||||
route_algo_inst,
|
route_algo_inst,
|
||||||
|
|
||||||
@@ -437,7 +430,7 @@ impl PeerManager {
|
|||||||
encryptor,
|
encryptor,
|
||||||
data_compress_algo,
|
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(),
|
reserved_my_peer_id_map: DashMap::new(),
|
||||||
recent_have_traffic: Arc::new(DashMap::new()),
|
recent_have_traffic: Arc::new(DashMap::new()),
|
||||||
@@ -963,7 +956,6 @@ impl PeerManager {
|
|||||||
Self::is_relay_data_packet(hdr.packet_type)
|
Self::is_relay_data_packet(hdr.packet_type)
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
|
|
||||||
async fn start_peer_recv(&self) {
|
async fn start_peer_recv(&self) {
|
||||||
let mut recv = self.packet_recv.lock().await.take().unwrap();
|
let mut recv = self.packet_recv.lock().await.take().unwrap();
|
||||||
let my_peer_id = self.my_peer_id;
|
let my_peer_id = self.my_peer_id;
|
||||||
@@ -1445,7 +1437,6 @@ impl PeerManager {
|
|||||||
self.get_route().get_foreign_network_summary().await
|
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 {
|
async fn run_nic_packet_process_pipeline(&self, data: &mut ZCPacket) -> bool {
|
||||||
// Enforce ACL for outbound (NIC-originated) packets. If ACL denies, stop processing.
|
// Enforce ACL for outbound (NIC-originated) packets. If ACL denies, stop processing.
|
||||||
if !self.global_ctx.get_acl_filter().process_packet_with_acl(
|
if !self.global_ctx.get_acl_filter().process_packet_with_acl(
|
||||||
@@ -1531,7 +1522,6 @@ impl PeerManager {
|
|||||||
result
|
result
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
|
|
||||||
async fn send_msg_internal(
|
async fn send_msg_internal(
|
||||||
peers: &Arc<PeerMap>,
|
peers: &Arc<PeerMap>,
|
||||||
foreign_network_client: &Arc<ForeignNetworkClient>,
|
foreign_network_client: &Arc<ForeignNetworkClient>,
|
||||||
@@ -1698,7 +1688,6 @@ impl PeerManager {
|
|||||||
(dst_peers, is_exit_node)
|
(dst_peers, is_exit_node)
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerManager"))]
|
|
||||||
pub async fn try_compress_and_encrypt(
|
pub async fn try_compress_and_encrypt(
|
||||||
compress_algo: CompressorAlgo,
|
compress_algo: CompressorAlgo,
|
||||||
encryptor: &Arc<dyn Encryptor + 'static>,
|
encryptor: &Arc<dyn Encryptor + 'static>,
|
||||||
|
|||||||
@@ -6,9 +6,6 @@ use std::{
|
|||||||
use anyhow::Context;
|
use anyhow::Context;
|
||||||
use dashmap::{DashMap, DashSet};
|
use dashmap::{DashMap, DashSet};
|
||||||
use parking_lot::Mutex;
|
use parking_lot::Mutex;
|
||||||
#[cfg(feature = "hotpath")]
|
|
||||||
use hotpath::wrap::tokio::sync::RwLock;
|
|
||||||
#[cfg(not(feature = "hotpath"))]
|
|
||||||
use tokio::sync::RwLock;
|
use tokio::sync::RwLock;
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
@@ -48,7 +45,7 @@ impl PeerMap {
|
|||||||
my_peer_id,
|
my_peer_id,
|
||||||
peer_map: DashMap::new(),
|
peer_map: DashMap::new(),
|
||||||
packet_send,
|
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())),
|
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)
|
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> {
|
pub async fn send_msg_directly(&self, msg: ZCPacket, dst_peer_id: PeerId) -> Result<(), Error> {
|
||||||
if dst_peer_id == self.my_peer_id {
|
if dst_peer_id == self.my_peer_id {
|
||||||
let packet_send = self.packet_send.clone();
|
let packet_send = self.packet_send.clone();
|
||||||
@@ -167,7 +163,6 @@ impl PeerMap {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerMap"))]
|
|
||||||
pub async fn get_gateway_peer_id(
|
pub async fn get_gateway_peer_id(
|
||||||
&self,
|
&self,
|
||||||
dst_peer_id: PeerId,
|
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> {
|
fn get_next_hop(&self, dst_peer_id: PeerId) -> Option<NextHopInfo> {
|
||||||
if self.suppressed_peer_ids.contains_key(&dst_peer_id) {
|
if self.suppressed_peer_ids.contains_key(&dst_peer_id) {
|
||||||
return None;
|
return None;
|
||||||
@@ -1402,7 +1401,6 @@ impl RouteTable {
|
|||||||
self.get_topology_next_hop(dst_peer_id)
|
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> {
|
fn get_topology_next_hop(&self, dst_peer_id: PeerId) -> Option<NextHopInfo> {
|
||||||
let cur_version = self.next_hop_map_version.get();
|
let cur_version = self.next_hop_map_version.get();
|
||||||
self.next_hop_map.get(&dst_peer_id).and_then(|x| {
|
self.next_hop_map.get(&dst_peer_id).and_then(|x| {
|
||||||
|
|||||||
@@ -1,11 +1,7 @@
|
|||||||
use std::sync::{
|
use std::sync::{
|
||||||
Arc,
|
Arc, RwLock,
|
||||||
atomic::{AtomicBool, Ordering},
|
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 std::time::Duration;
|
||||||
|
|
||||||
use anyhow::anyhow;
|
use anyhow::anyhow;
|
||||||
@@ -266,7 +262,7 @@ impl std::fmt::Debug for PeerSession {
|
|||||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
f.debug_struct("PeerSession")
|
f.debug_struct("PeerSession")
|
||||||
.field("peer_id", &self.peer_id)
|
.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)
|
.field("datagram", &self.datagram)
|
||||||
.finish()
|
.finish()
|
||||||
}
|
}
|
||||||
@@ -286,7 +282,7 @@ impl PeerSession {
|
|||||||
) -> Self {
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
peer_id,
|
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(
|
datagram: SecureDatagramSession::new(
|
||||||
root_key,
|
root_key,
|
||||||
session_generation,
|
session_generation,
|
||||||
@@ -380,7 +376,6 @@ impl PeerSession {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "PeerSession"))]
|
|
||||||
pub fn encrypt_payload(
|
pub fn encrypt_payload(
|
||||||
&self,
|
&self,
|
||||||
sender_peer_id: PeerId,
|
sender_peer_id: PeerId,
|
||||||
@@ -394,7 +389,6 @@ impl PeerSession {
|
|||||||
.encrypt_payload(Self::dir_for_sender(sender_peer_id, receiver_peer_id), pkt)
|
.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(
|
pub fn decrypt_payload(
|
||||||
&self,
|
&self,
|
||||||
sender_peer_id: PeerId,
|
sender_peer_id: PeerId,
|
||||||
|
|||||||
@@ -144,7 +144,6 @@ impl RelayPeerMap {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
|
|
||||||
async fn send_via_next_hop(
|
async fn send_via_next_hop(
|
||||||
&self,
|
&self,
|
||||||
msg: ZCPacket,
|
msg: ZCPacket,
|
||||||
@@ -167,7 +166,6 @@ impl RelayPeerMap {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "RelayPeerMap"))]
|
|
||||||
pub async fn send_msg(
|
pub async fn send_msg(
|
||||||
self: &Arc<Self>,
|
self: &Arc<Self>,
|
||||||
mut msg: ZCPacket,
|
mut msg: ZCPacket,
|
||||||
@@ -615,7 +613,6 @@ impl RelayPeerMap {
|
|||||||
Ok(())
|
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> {
|
pub async fn decrypt_if_needed(self: &Arc<Self>, packet: &mut ZCPacket) -> Result<bool, Error> {
|
||||||
if !self.is_secure_mode_enabled() {
|
if !self.is_secure_mode_enabled() {
|
||||||
return Ok(false);
|
return Ok(false);
|
||||||
|
|||||||
@@ -1,16 +1,11 @@
|
|||||||
use std::{
|
use std::{
|
||||||
sync::{
|
sync::{
|
||||||
|
Arc, Mutex, RwLock,
|
||||||
atomic::{AtomicBool, AtomicU32, Ordering},
|
atomic::{AtomicBool, AtomicU32, Ordering},
|
||||||
Arc,
|
|
||||||
},
|
},
|
||||||
time::{SystemTime, UNIX_EPOCH},
|
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 anyhow::anyhow;
|
||||||
use atomic_shim::AtomicU64;
|
use atomic_shim::AtomicU64;
|
||||||
use hmac::{Hmac, Mac as _};
|
use hmac::{Hmac, Mac as _};
|
||||||
@@ -19,7 +14,7 @@ use sha2::Sha256;
|
|||||||
use zerocopy::FromBytes;
|
use zerocopy::FromBytes;
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
peers::encrypt::{create_encryptor, Encryptor},
|
peers::encrypt::{Encryptor, create_encryptor},
|
||||||
tunnel::packet_def::{StandardAeadTail, ZCPacket},
|
tunnel::packet_def::{StandardAeadTail, ZCPacket},
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -233,15 +228,15 @@ pub struct SecureDatagramSession {
|
|||||||
impl std::fmt::Debug for SecureDatagramSession {
|
impl std::fmt::Debug for SecureDatagramSession {
|
||||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
f.debug_struct("SecureDatagramSession")
|
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("session_generation", &self.session_generation)
|
||||||
.field("send_epoch", &self.send_epoch)
|
.field("send_epoch", &self.send_epoch)
|
||||||
.field("send_seq", &self.send_seq)
|
.field("send_seq", &self.send_seq)
|
||||||
.field("send_epoch_started_ms", &self.send_epoch_started_ms)
|
.field("send_epoch_started_ms", &self.send_epoch_started_ms)
|
||||||
.field("send_packets_since_epoch", &self.send_packets_since_epoch)
|
.field("send_packets_since_epoch", &self.send_packets_since_epoch)
|
||||||
.field("rx_slots", &*self.rx_slots.lock().unwrap())
|
.field("rx_slots", &self.rx_slots)
|
||||||
.field("key_cache", &*self.key_cache.lock().unwrap())
|
.field("key_cache", &self.key_cache)
|
||||||
.field("sync_rx_grace", &*self.sync_rx_grace.lock().unwrap())
|
.field("sync_rx_grace", &self.sync_rx_grace)
|
||||||
.field(
|
.field(
|
||||||
"sync_rx_grace_expires_at_ms",
|
"sync_rx_grace_expires_at_ms",
|
||||||
&self.sync_rx_grace_expires_at_ms,
|
&self.sync_rx_grace_expires_at_ms,
|
||||||
@@ -277,15 +272,15 @@ impl SecureDatagramSession {
|
|||||||
];
|
];
|
||||||
let now_ms = now_ms();
|
let now_ms = now_ms();
|
||||||
Self {
|
Self {
|
||||||
root_key: hotpath::rw_lock!(std::sync::RwLock::new(root_key)),
|
root_key: RwLock::new(root_key),
|
||||||
session_generation: AtomicU32::new(session_generation),
|
session_generation: AtomicU32::new(session_generation),
|
||||||
send_epoch: AtomicU32::new(initial_epoch),
|
send_epoch: AtomicU32::new(initial_epoch),
|
||||||
send_seq: [AtomicU64::new(0), AtomicU64::new(0)],
|
send_seq: [AtomicU64::new(0), AtomicU64::new(0)],
|
||||||
send_epoch_started_ms: AtomicU64::new(now_ms),
|
send_epoch_started_ms: AtomicU64::new(now_ms),
|
||||||
send_packets_since_epoch: AtomicU64::new(0),
|
send_packets_since_epoch: AtomicU64::new(0),
|
||||||
rx_slots: hotpath::mutex!(std::sync::Mutex::new(rx_slots)),
|
rx_slots: Mutex::new(rx_slots),
|
||||||
key_cache: hotpath::mutex!(std::sync::Mutex::new(key_cache)),
|
key_cache: Mutex::new(key_cache),
|
||||||
sync_rx_grace: hotpath::mutex!(std::sync::Mutex::new(SyncRxGrace::default())),
|
sync_rx_grace: Mutex::new(SyncRxGrace::default()),
|
||||||
sync_rx_grace_expires_at_ms: AtomicU64::new(0),
|
sync_rx_grace_expires_at_ms: AtomicU64::new(0),
|
||||||
send_cipher_algorithm,
|
send_cipher_algorithm,
|
||||||
recv_cipher_algorithm,
|
recv_cipher_algorithm,
|
||||||
@@ -706,10 +701,6 @@ impl SecureDatagramSession {
|
|||||||
false
|
false
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(
|
|
||||||
feature = "hotpath",
|
|
||||||
hotpath::measure(impl_type = "SecureDatagramSession")
|
|
||||||
)]
|
|
||||||
pub fn encrypt_payload(
|
pub fn encrypt_payload(
|
||||||
&self,
|
&self,
|
||||||
dir: SecureDatagramDirection,
|
dir: SecureDatagramDirection,
|
||||||
@@ -728,10 +719,6 @@ impl SecureDatagramSession {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(
|
|
||||||
feature = "hotpath",
|
|
||||||
hotpath::measure(impl_type = "SecureDatagramSession")
|
|
||||||
)]
|
|
||||||
pub fn decrypt_payload(
|
pub fn decrypt_payload(
|
||||||
&self,
|
&self,
|
||||||
dir: SecureDatagramDirection,
|
dir: SecureDatagramDirection,
|
||||||
@@ -897,9 +884,11 @@ mod tests {
|
|||||||
let nonce_offset = payload.len() - StandardAeadTail::NONCE_SIZE;
|
let nonce_offset = payload.len() - StandardAeadTail::NONCE_SIZE;
|
||||||
payload[nonce_offset..].copy_from_slice(&poisoned_nonce);
|
payload[nonce_offset..].copy_from_slice(&poisoned_nonce);
|
||||||
|
|
||||||
assert!(receiver
|
assert!(
|
||||||
.decrypt_payload(SecureDatagramDirection::AToB, &mut forged)
|
receiver
|
||||||
.is_err());
|
.decrypt_payload(SecureDatagramDirection::AToB, &mut forged)
|
||||||
|
.is_err()
|
||||||
|
);
|
||||||
|
|
||||||
let plaintext = b"pkt2";
|
let plaintext = b"pkt2";
|
||||||
let mut pkt2 = ZCPacket::new_with_payload(plaintext);
|
let mut pkt2 = ZCPacket::new_with_payload(plaintext);
|
||||||
|
|||||||
@@ -49,7 +49,7 @@ async fn test_route_peer_info_ipv6() {
|
|||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_peer_manager_ipv6() {
|
async fn test_peer_manager_ipv6() {
|
||||||
let global_ctx = get_mock_global_ctx();
|
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(
|
let peer_mgr = crate::peers::peer_manager::PeerManager::new(
|
||||||
RouteAlgoType::Ospf,
|
RouteAlgoType::Ospf,
|
||||||
global_ctx.clone(),
|
global_ctx.clone(),
|
||||||
|
|||||||
@@ -48,13 +48,9 @@ use std::collections::{HashMap, HashSet};
|
|||||||
use std::fmt;
|
use std::fmt;
|
||||||
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
|
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
|
||||||
use std::sync::{
|
use std::sync::{
|
||||||
Arc,
|
Arc, RwLock,
|
||||||
atomic::{AtomicU32, Ordering},
|
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::sync::broadcast;
|
||||||
use tokio::time;
|
use tokio::time;
|
||||||
use tokio_util::task::AbortOnDropHandle;
|
use tokio_util::task::AbortOnDropHandle;
|
||||||
@@ -89,7 +85,7 @@ impl AddrTuple {
|
|||||||
|
|
||||||
#[derive(Default)]
|
#[derive(Default)]
|
||||||
struct StackState {
|
struct StackState {
|
||||||
tuples: HashMap<AddrTuple, hotpath::wrap::flume::Sender<Bytes>>,
|
tuples: HashMap<AddrTuple, flume::Sender<Bytes>>,
|
||||||
closed: bool,
|
closed: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -133,7 +129,7 @@ pub enum State {
|
|||||||
pub struct Socket {
|
pub struct Socket {
|
||||||
shared: Arc<Shared>,
|
shared: Arc<Shared>,
|
||||||
tun: Arc<dyn Tun>,
|
tun: Arc<dyn Tun>,
|
||||||
incoming: hotpath::wrap::flume::Receiver<Bytes>,
|
incoming: flume::Receiver<Bytes>,
|
||||||
local_addr: SocketAddr,
|
local_addr: SocketAddr,
|
||||||
remote_addr: SocketAddr,
|
remote_addr: SocketAddr,
|
||||||
local_mac: MacAddr,
|
local_mac: MacAddr,
|
||||||
@@ -162,8 +158,8 @@ impl Socket {
|
|||||||
remote_mac: Option<MacAddr>,
|
remote_mac: Option<MacAddr>,
|
||||||
ack: Option<u32>,
|
ack: Option<u32>,
|
||||||
state: State,
|
state: State,
|
||||||
) -> (Socket, hotpath::wrap::flume::Sender<Bytes>) {
|
) -> (Socket, flume::Sender<Bytes>) {
|
||||||
let (incoming_tx, incoming_rx) = hotpath::channel!(flume::bounded(MPMC_BUFFER_LEN));
|
let (incoming_tx, incoming_rx) = flume::bounded(MPMC_BUFFER_LEN);
|
||||||
|
|
||||||
(
|
(
|
||||||
Socket {
|
Socket {
|
||||||
@@ -434,9 +430,9 @@ impl Stack {
|
|||||||
) -> Stack {
|
) -> Stack {
|
||||||
let (tuples_purge_tx, _tuples_purge_rx) = broadcast::channel(16);
|
let (tuples_purge_tx, _tuples_purge_rx) = broadcast::channel(16);
|
||||||
let shared = Arc::new(Shared {
|
let shared = Arc::new(Shared {
|
||||||
state: hotpath::rw_lock!(std::sync::RwLock::new(StackState::default())),
|
state: RwLock::new(StackState::default()),
|
||||||
tun: tun.clone(),
|
tun: tun.clone(),
|
||||||
listening: hotpath::rw_lock!(std::sync::RwLock::new(HashSet::new())),
|
listening: RwLock::new(HashSet::new()),
|
||||||
tuples_purge: tuples_purge_tx.clone(),
|
tuples_purge: tuples_purge_tx.clone(),
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -505,7 +501,7 @@ impl Stack {
|
|||||||
shared: Arc<Shared>,
|
shared: Arc<Shared>,
|
||||||
mut tuples_purge: broadcast::Receiver<AddrTuple>,
|
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 {
|
loop {
|
||||||
let mut buf = BytesMut::new();
|
let mut buf = BytesMut::new();
|
||||||
|
|||||||
@@ -9,8 +9,7 @@ use crate::proto::common::TunnelInfo;
|
|||||||
|
|
||||||
use super::{Tunnel, TunnelError, ZCPacketSink, ZCPacketStream, packet_def::ZCPacket};
|
use super::{Tunnel, TunnelError, ZCPacketSink, ZCPacketStream, packet_def::ZCPacket};
|
||||||
|
|
||||||
use hotpath::wrap::tokio::sync::mpsc::{Receiver, Sender};
|
use tokio::sync::mpsc::{Receiver, Sender, channel, error::TrySendError};
|
||||||
use tokio::sync::mpsc::{channel, error::TrySendError};
|
|
||||||
use tokio_util::task::AbortOnDropHandle;
|
use tokio_util::task::AbortOnDropHandle;
|
||||||
// use tachyonix::{channel, Receiver, Sender, TrySendError};
|
// use tachyonix::{channel, Receiver, Sender, TrySendError};
|
||||||
|
|
||||||
@@ -20,7 +19,6 @@ use futures::SinkExt;
|
|||||||
pub struct MpscTunnelSender(Sender<ZCPacket>);
|
pub struct MpscTunnelSender(Sender<ZCPacket>);
|
||||||
|
|
||||||
impl MpscTunnelSender {
|
impl MpscTunnelSender {
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnelSender"))]
|
|
||||||
pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
|
pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
|
||||||
self.0.send(item).await.with_context(|| "send error")?;
|
self.0.send(item).await.with_context(|| "send error")?;
|
||||||
Ok(())
|
Ok(())
|
||||||
@@ -45,7 +43,7 @@ pub struct MpscTunnel<T> {
|
|||||||
|
|
||||||
impl<T: Tunnel> MpscTunnel<T> {
|
impl<T: Tunnel> MpscTunnel<T> {
|
||||||
pub fn new(tunnel: T, send_timeout: Option<Duration>) -> Self {
|
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 (stream, mut sink) = tunnel.split();
|
||||||
|
|
||||||
let task = tokio::spawn(async move {
|
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(
|
async fn forward_one_round(
|
||||||
rx: &mut Receiver<ZCPacket>,
|
rx: &mut Receiver<ZCPacket>,
|
||||||
sink: &mut Pin<Box<dyn ZCPacketSink>>,
|
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(
|
async fn forward_one_round_no_timeout(
|
||||||
rx: &mut Receiver<ZCPacket>,
|
rx: &mut Receiver<ZCPacket>,
|
||||||
sink: &mut Pin<Box<dyn ZCPacketSink>>,
|
sink: &mut Pin<Box<dyn ZCPacketSink>>,
|
||||||
@@ -100,7 +96,6 @@ impl<T: Tunnel> MpscTunnel<T> {
|
|||||||
sink.flush().await
|
sink.flush().await
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "MpscTunnel"))]
|
|
||||||
async fn forward_one_round_with_timeout(
|
async fn forward_one_round_with_timeout(
|
||||||
rx: &mut Receiver<ZCPacket>,
|
rx: &mut Receiver<ZCPacket>,
|
||||||
sink: &mut Pin<Box<dyn ZCPacketSink>>,
|
sink: &mut Pin<Box<dyn ZCPacketSink>>,
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
use bytes::Buf;
|
||||||
use bytes::Bytes;
|
use bytes::Bytes;
|
||||||
use bytes::BytesMut;
|
use bytes::BytesMut;
|
||||||
use zerocopy::AsBytes;
|
use zerocopy::AsBytes;
|
||||||
@@ -486,7 +487,7 @@ impl ZCPacket {
|
|||||||
let total_len = payload_off + payload.len();
|
let total_len = payload_off + payload.len();
|
||||||
ret.inner.reserve(total_len);
|
ret.inner.reserve(total_len);
|
||||||
unsafe { ret.inner.set_len(total_len) };
|
unsafe { ret.inner.set_len(total_len) };
|
||||||
ret.mut_payload()[..payload.len()].copy_from_slice(payload);
|
ret.mut_payload().copy_from_slice(payload);
|
||||||
ret
|
ret
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -587,7 +588,8 @@ impl ZCPacket {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn payload_bytes(mut self) -> BytesMut {
|
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> {
|
pub fn peer_manager_header(&self) -> Option<&PeerManagerHeader> {
|
||||||
@@ -652,11 +654,12 @@ impl ZCPacket {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn tunnel_payload_bytes(mut self) -> BytesMut {
|
pub fn tunnel_payload_bytes(mut self) -> BytesMut {
|
||||||
self.inner.split_off(
|
self.inner.advance(
|
||||||
self.packet_type
|
self.packet_type
|
||||||
.get_packet_offsets()
|
.get_packet_offsets()
|
||||||
.peer_manager_header_offset,
|
.peer_manager_header_offset,
|
||||||
)
|
);
|
||||||
|
self.inner
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn convert_type(mut self, target_packet_type: ZCPacketType) -> Self {
|
pub fn convert_type(mut self, target_packet_type: ZCPacketType) -> Self {
|
||||||
@@ -702,7 +705,8 @@ impl ZCPacket {
|
|||||||
return Self::new_from_buf(buf, target_packet_type);
|
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 {
|
pub fn into_bytes(self) -> Bytes {
|
||||||
@@ -748,8 +752,10 @@ impl ZCPacket {
|
|||||||
let foreign_hdr_len = hdr.get_header_len();
|
let foreign_hdr_len = hdr.get_header_len();
|
||||||
|
|
||||||
Self::new_from_buf(
|
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,
|
ZCPacketType::DummyTunnel,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -12,9 +12,6 @@ use crate::tunnel::{
|
|||||||
use anyhow::Context;
|
use anyhow::Context;
|
||||||
use derivative::Derivative;
|
use derivative::Derivative;
|
||||||
use derive_more::{Deref, DerefMut};
|
use derive_more::{Deref, DerefMut};
|
||||||
#[cfg(feature = "hotpath")]
|
|
||||||
use hotpath::wrap::parking_lot::RwLock;
|
|
||||||
#[cfg(not(feature = "hotpath"))]
|
|
||||||
use parking_lot::RwLock;
|
use parking_lot::RwLock;
|
||||||
use quinn::{
|
use quinn::{
|
||||||
ClientConfig, ConnectError, Connection, Endpoint, EndpointConfig, ServerConfig,
|
ClientConfig, ConnectError, Connection, Endpoint, EndpointConfig, ServerConfig,
|
||||||
@@ -315,25 +312,18 @@ struct RwPoolInner<Item> {
|
|||||||
enabled: bool,
|
enabled: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
struct RwPool<Item> {
|
struct RwPool<Item> {
|
||||||
ephemeral: RwLock<RwPoolInner<Item>>,
|
ephemeral: RwLock<RwPoolInner<Item>>,
|
||||||
persistent: RwLock<RwPoolInner<Item>>,
|
persistent: RwLock<RwPoolInner<Item>>,
|
||||||
capacity: usize,
|
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> {
|
impl<Item> RwPool<Item> {
|
||||||
fn new(capacity: usize) -> Self {
|
fn new(capacity: usize) -> Self {
|
||||||
Self {
|
Self {
|
||||||
ephemeral: hotpath::rw_lock!(parking_lot::RwLock::new(RwPoolInner::default())),
|
ephemeral: RwLock::new(RwPoolInner::default()),
|
||||||
persistent: hotpath::rw_lock!(parking_lot::RwLock::new(RwPoolInner::default())),
|
persistent: RwLock::new(RwPoolInner::default()),
|
||||||
capacity,
|
capacity,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -11,8 +11,7 @@ use async_trait::async_trait;
|
|||||||
use futures::{Sink, SinkExt, Stream, StreamExt};
|
use futures::{Sink, SinkExt, Stream, StreamExt};
|
||||||
use once_cell::sync::Lazy;
|
use once_cell::sync::Lazy;
|
||||||
|
|
||||||
use hotpath::wrap::tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
|
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel};
|
||||||
use tokio::sync::mpsc::unbounded_channel;
|
|
||||||
|
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
@@ -197,7 +196,7 @@ pub struct RingTunnelListener {
|
|||||||
|
|
||||||
impl RingTunnelListener {
|
impl RingTunnelListener {
|
||||||
pub fn new(key: url::Url) -> Self {
|
pub fn new(key: url::Url) -> Self {
|
||||||
let (conn_sender, conn_receiver) = hotpath::channel!(unbounded_channel());
|
let (conn_sender, conn_receiver) = unbounded_channel();
|
||||||
RingTunnelListener {
|
RingTunnelListener {
|
||||||
listener_addr: key,
|
listener_addr: key,
|
||||||
conn_sender,
|
conn_sender,
|
||||||
|
|||||||
@@ -13,10 +13,11 @@ use futures::{StreamExt, stream::FuturesUnordered};
|
|||||||
use rand::{Rng, SeedableRng};
|
use rand::{Rng, SeedableRng};
|
||||||
use zerocopy::{AsBytes, FromBytes};
|
use zerocopy::{AsBytes, FromBytes};
|
||||||
|
|
||||||
use hotpath::wrap::tokio::sync::mpsc::{Receiver, Sender, UnboundedReceiver, UnboundedSender};
|
|
||||||
use tokio::{
|
use tokio::{
|
||||||
net::UdpSocket,
|
net::UdpSocket,
|
||||||
sync::mpsc::{channel, unbounded_channel},
|
sync::mpsc::{
|
||||||
|
Receiver, Sender, UnboundedReceiver, UnboundedSender, channel, unbounded_channel,
|
||||||
|
},
|
||||||
task::JoinSet,
|
task::JoinSet,
|
||||||
};
|
};
|
||||||
use tokio_util::task::AbortOnDropHandle;
|
use tokio_util::task::AbortOnDropHandle;
|
||||||
@@ -292,7 +293,6 @@ fn get_zcpacket_from_buf(buf: BytesMut, allow_stun: bool) -> Result<ZCPacket, Tu
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[instrument]
|
#[instrument]
|
||||||
#[cfg_attr(feature = "hotpath", hotpath::measure())]
|
|
||||||
async fn forward_from_ring_to_udp(
|
async fn forward_from_ring_to_udp(
|
||||||
mut ring_recv: RingStream,
|
mut ring_recv: RingStream,
|
||||||
socket: &Arc<UdpSocket>,
|
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(
|
async fn udp_recv_from_socket_forward_task(
|
||||||
socket: &UdpSocket,
|
socket: &UdpSocket,
|
||||||
buf: &mut BytesMut,
|
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> {
|
pub fn handle_packet_from_remote(&mut self, zc_packet: ZCPacket) -> Result<(), TunnelError> {
|
||||||
let header = zc_packet.udp_tunnel_header().unwrap();
|
let header = zc_packet.udp_tunnel_header().unwrap();
|
||||||
let conn_id = header.conn_id.get();
|
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) {
|
fn do_forward_one_packet_to_conn(&self, zc_packet: ZCPacket, addr: SocketAddr) {
|
||||||
let header = zc_packet.udp_tunnel_header().unwrap();
|
let header = zc_packet.udp_tunnel_header().unwrap();
|
||||||
if header.msg_type == UdpPacketType::Syn as u8 {
|
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) {
|
async fn do_forward_task(self) {
|
||||||
let socket = self.socket.as_ref().unwrap().clone();
|
let socket = self.socket.as_ref().unwrap().clone();
|
||||||
let mut buf = BytesMut::new();
|
let mut buf = BytesMut::new();
|
||||||
@@ -679,8 +675,8 @@ pub struct UdpTunnelListener {
|
|||||||
|
|
||||||
impl UdpTunnelListener {
|
impl UdpTunnelListener {
|
||||||
pub fn new(addr: url::Url) -> Self {
|
pub fn new(addr: url::Url) -> Self {
|
||||||
let (close_event_send, close_event_recv) = hotpath::channel!(unbounded_channel());
|
let (close_event_send, close_event_recv) = unbounded_channel();
|
||||||
let (conn_send, conn_recv) = hotpath::channel!(channel(100));
|
let (conn_send, conn_recv) = channel(100);
|
||||||
Self {
|
Self {
|
||||||
addr: addr.clone(),
|
addr: addr.clone(),
|
||||||
socket: None,
|
socket: None,
|
||||||
@@ -920,8 +916,7 @@ impl UdpTunnelConnector {
|
|||||||
"udp build tunnel for connector"
|
"udp build tunnel for connector"
|
||||||
);
|
);
|
||||||
|
|
||||||
let (close_event_sender, mut close_event_recv) =
|
let (close_event_sender, mut close_event_recv) = unbounded_channel();
|
||||||
hotpath::channel!(unbounded_channel());
|
|
||||||
|
|
||||||
let ring_recv = RingStream::new(ring_for_send_udp.clone());
|
let ring_recv = RingStream::new(ring_for_send_udp.clone());
|
||||||
let ring_sender = RingSink::new(ring_for_recv_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 dst_addr = "127.0.0.1:1".parse().unwrap();
|
||||||
let ring_for_send_udp = Arc::new(RingTunnel::new(8));
|
let ring_for_send_udp = Arc::new(RingTunnel::new(8));
|
||||||
let ring_for_recv_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(
|
let mut conn = UdpConnection::new(
|
||||||
socket,
|
socket,
|
||||||
7,
|
7,
|
||||||
|
|||||||
@@ -37,17 +37,9 @@ use crossbeam::atomic::AtomicCell;
|
|||||||
use dashmap::DashMap;
|
use dashmap::DashMap;
|
||||||
use futures::{SinkExt, StreamExt, stream::FuturesUnordered};
|
use futures::{SinkExt, StreamExt, stream::FuturesUnordered};
|
||||||
use rand::RngCore;
|
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::{
|
use tokio::{
|
||||||
net::UdpSocket,
|
net::UdpSocket,
|
||||||
sync::mpsc::unbounded_channel,
|
sync::{Mutex, mpsc::unbounded_channel},
|
||||||
task::JoinSet,
|
task::JoinSet,
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -355,7 +347,7 @@ struct WgPeer {
|
|||||||
config: WgConfig,
|
config: WgConfig,
|
||||||
endpoint: SocketAddr,
|
endpoint: SocketAddr,
|
||||||
|
|
||||||
sink: StdMutex<Option<Pin<Box<dyn ZCPacketSink>>>>,
|
sink: std::sync::Mutex<Option<Pin<Box<dyn ZCPacketSink>>>>,
|
||||||
|
|
||||||
data: Option<WgPeerData>,
|
data: Option<WgPeerData>,
|
||||||
tasks: JoinSet<()>,
|
tasks: JoinSet<()>,
|
||||||
@@ -366,19 +358,19 @@ struct WgPeer {
|
|||||||
impl WgPeer {
|
impl WgPeer {
|
||||||
fn new(udp: Arc<UdpSocket>, config: WgConfig, endpoint: SocketAddr) -> Self {
|
fn new(udp: Arc<UdpSocket>, config: WgConfig, endpoint: SocketAddr) -> Self {
|
||||||
WgPeer {
|
WgPeer {
|
||||||
tunn: Some(hotpath::mutex!(tokio::sync::Mutex::new(Tunn::new(
|
tunn: Some(Mutex::new(Tunn::new(
|
||||||
config.my_secret_key.clone(),
|
config.my_secret_key.clone(),
|
||||||
config.peer_public_key,
|
config.peer_public_key,
|
||||||
None,
|
None,
|
||||||
None,
|
None,
|
||||||
rand::thread_rng().next_u32(),
|
rand::thread_rng().next_u32(),
|
||||||
None,
|
None,
|
||||||
)))),
|
))),
|
||||||
|
|
||||||
udp,
|
udp,
|
||||||
config,
|
config,
|
||||||
endpoint,
|
endpoint,
|
||||||
sink: hotpath::mutex!(std::sync::Mutex::new(None)),
|
sink: std::sync::Mutex::new(None),
|
||||||
|
|
||||||
data: None,
|
data: None,
|
||||||
tasks: JoinSet::new(),
|
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) {
|
async fn handle_packet_from_me<S: ZCPacketStream + Unpin>(mut stream: S, data: WgPeerData) {
|
||||||
while let Some(Ok(packet)) = stream.next().await {
|
while let Some(Ok(packet)) = stream.next().await {
|
||||||
let ret = data.handle_one_packet_from_me(packet).await;
|
let ret = data.handle_one_packet_from_me(packet).await;
|
||||||
@@ -399,7 +390,6 @@ impl WgPeer {
|
|||||||
.store(true, std::sync::atomic::Ordering::Relaxed);
|
.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]) {
|
async fn handle_packet_from_peer(&self, packet: &[u8]) {
|
||||||
self.access_time.store(Instant::now());
|
self.access_time.store(Instant::now());
|
||||||
tracing::trace!("Received {} bytes from peer", packet.len());
|
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 ConnSender = tokio::sync::mpsc::UnboundedSender<Box<dyn Tunnel>>;
|
||||||
type ConnReceiver = hotpath::wrap::tokio::sync::mpsc::UnboundedReceiver<Box<dyn Tunnel>>;
|
type ConnReceiver = tokio::sync::mpsc::UnboundedReceiver<Box<dyn Tunnel>>;
|
||||||
|
|
||||||
pub struct WgTunnelListener {
|
pub struct WgTunnelListener {
|
||||||
addr: url::Url,
|
addr: url::Url,
|
||||||
@@ -484,7 +474,7 @@ pub struct WgTunnelListener {
|
|||||||
|
|
||||||
impl WgTunnelListener {
|
impl WgTunnelListener {
|
||||||
pub fn new(addr: url::Url, config: WgConfig) -> Self {
|
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 {
|
WgTunnelListener {
|
||||||
addr,
|
addr,
|
||||||
config,
|
config,
|
||||||
|
|||||||
Reference in New Issue
Block a user