From d55e63b88ec5f2fca2b7aeedf4770d4261e812d4 Mon Sep 17 00:00:00 2001 From: KKRainbow <443152178@qq.com> Date: Tue, 28 Jul 2026 00:29:08 +0800 Subject: [PATCH] perf(wasi): optimize data plane and extend host ABI to v3 (#2455) Overhaul the WASI guest data plane for throughput and add the host capabilities it relies on. The externally driven Tokio runtime now runs its timer pre-turn only when a tracked deadline has expired, and all WASI-reachable timers (STUN, port mapping, WebClient, UDP flow cleanup) go through the portable time facade so conditional timer driving cannot starve them. Data plane: - Move read/write deadlines onto TCP and UDP resources with one ABI setter per direction, reuse a single expiration timer per resource, and drop timeout arguments from the four hot data-plane submissions (ABI v3). Checked absolute instants treat unrepresentable finite timeouts as unbounded instead of panicking. - Batch host traffic: vectored TCP frame writes combine queued slices into one host operation, and reads request a bounded 64 KiB while retaining excess bytes in the stream buffer. - Complete TCP writes inside the guest with cancellation-safe writes, reporting the completed prefix before honoring cancellation or timeout so hosts never replay bytes. - Repoll smoltcp egress immediately on zero poll delay, enlarge virtual UDP receive queues to 128 KiB payload with 128 metadata slots, and bound UDP session receive buffers to 8 KiB plus one byte while keeping oversized-datagram detection. Host integration: - Add optional algorithm-neutral AEAD seal/open imports with the ring backend as fallback, and pin the ring AES-128-GCM wire vector so the Go host stays interoperable. - Forward instance events to hosts through one best-effort, synchronous, non-blocking import. - Add a repository-owned build entry point for the Go host artifact: Binaryen 131 at -O4 with cached, SHA-256-verified official archives. --- .github/workflows/gui.yml | 2 +- .github/workflows/mobile.yml | 2 +- .github/workflows/ohos.yml | 2 +- CONTRIBUTING.md | 13 +- .../easytier-ffi/DATA_PLANE_ABI.md | 18 +- .../easytier-ffi/src/data_plane/abi.rs | 43 ++- .../easytier-ffi/src/data_plane/session.rs | 42 +-- easytier-contrib/easytier-ffi/src/lib.rs | 5 +- easytier-contrib/easytier-ffi/src/tests.rs | 4 + easytier-core/Cargo.toml | 1 + .../connectivity/hole_punch/port_mapping.rs | 2 +- easytier-core/src/connectivity/stun/client.rs | 15 +- .../src/connectivity/stun/collector.rs | 6 +- .../src/foundation/operation_broker.rs | 32 +++ easytier-core/src/foundation/time.rs | 6 +- .../src/gateway/dataplane/deadline.rs | 167 ++++++++++- easytier-core/src/gateway/dataplane/mod.rs | 2 +- .../src/gateway/dataplane/session.rs | 260 +++++++++++++++--- easytier-core/src/gateway/dataplane/stack.rs | 2 + easytier-core/src/gateway/dataplane/tests.rs | 23 +- easytier-core/src/gateway/port_forward.rs | 2 +- .../gateway/smoltcp/tokio_smoltcp/reactor.rs | 20 +- easytier-core/src/host/socket/mod.rs | 156 ++++++++--- .../src/management/full/web_client.rs | 17 +- easytier-core/src/socket/udp/layer.rs | 2 +- easytier-core/src/socket/udp/tests.rs | 29 ++ .../src/socket/udp/virtual_socket.rs | 38 ++- easytier-core/src/tunnel/encrypt/mod.rs | 35 ++- easytier-core/src/tunnel/encrypt/ring.rs | 19 ++ easytier-core/src/tunnel/encrypt/wasi_host.rs | 152 ++++++++++ easytier-core/src/wasi/abi.rs | 18 +- easytier-core/src/wasi/adapter/event.rs | 60 ++++ easytier-core/src/wasi/adapter/mod.rs | 1 + easytier-core/src/wasi/imports.rs | 51 ++++ easytier-core/src/wasi/runtime.rs | 29 +- .../src/wasi/runtime/abi/data_plane.rs | 59 ++-- easytier-core/src/wasi/runtime_driver.rs | 73 ++++- easytier-core/src/wasi/time.rs | 2 +- easytier/src/socket/udp.rs | 18 +- script/build-wasi-core.sh | 134 +++++++++ 40 files changed, 1342 insertions(+), 220 deletions(-) create mode 100644 easytier-core/src/tunnel/encrypt/wasi_host.rs create mode 100644 easytier-core/src/wasi/adapter/event.rs create mode 100755 script/build-wasi-core.sh diff --git a/.github/workflows/gui.yml b/.github/workflows/gui.yml index d897fbc5..b053139b 100644 --- a/.github/workflows/gui.yml +++ b/.github/workflows/gui.yml @@ -35,7 +35,7 @@ jobs: concurrent_skipping: 'same_content_newer' skip_after_successful_duplicate: 'true' cancel_others: 'true' - paths: '["Cargo.toml", "Cargo.lock", "easytier/**", "easytier-gui/**", ".github/workflows/gui.yml", ".github/actions/**", "easytier-web/frontend-lib/**"]' + paths: '["Cargo.toml", "Cargo.lock", "easytier/**", "easytier-core/**", "easytier-gui/**", ".github/workflows/gui.yml", ".github/actions/**", "easytier-web/frontend-lib/**"]' build-gui: strategy: fail-fast: true diff --git a/.github/workflows/mobile.yml b/.github/workflows/mobile.yml index e8caa484..1eb5a069 100644 --- a/.github/workflows/mobile.yml +++ b/.github/workflows/mobile.yml @@ -35,7 +35,7 @@ jobs: concurrent_skipping: 'same_content_newer' skip_after_successful_duplicate: 'true' cancel_others: 'true' - paths: '["Cargo.toml", "Cargo.lock", "easytier/**", "easytier-gui/**", "tauri-plugin-vpnservice/**", ".github/workflows/mobile.yml", ".github/actions/**"]' + paths: '["Cargo.toml", "Cargo.lock", "easytier/**", "easytier-core/**", "easytier-gui/**", "tauri-plugin-vpnservice/**", ".github/workflows/mobile.yml", ".github/actions/**"]' build-mobile: strategy: fail-fast: true diff --git a/.github/workflows/ohos.yml b/.github/workflows/ohos.yml index 5c780cc2..4308e0fb 100644 --- a/.github/workflows/ohos.yml +++ b/.github/workflows/ohos.yml @@ -61,7 +61,7 @@ jobs: concurrent_skipping: "same_content_newer" skip_after_successful_duplicate: "true" cancel_others: "true" - paths: '["Cargo.toml", "Cargo.lock", "easytier/**", "easytier-contrib/easytier-ohrs/**", ".github/workflows/ohos.yml", ".github/actions/**"]' + paths: '["Cargo.toml", "Cargo.lock", "easytier/**", "easytier-core/**", "easytier-contrib/easytier-ohrs/**", ".github/workflows/ohos.yml", ".github/actions/**"]' build-ohos: runs-on: ubuntu-latest diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 072505f3..e27aa788 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -113,6 +113,17 @@ cargo build --release --target x86_64-pc-windows-msvc # Windows x86_64 Build artifacts: `target/[target-triple]/release/` +### Building the WASI core + +```bash +script/build-wasi-core.sh +``` + +This builds the `easytier-core` Go-host profile for `wasm32-wasip1`, then +optimizes it with the pinned official Binaryen release. Binaryen is downloaded +once into `target/binaryen/` and verified by SHA-256; set `WASM_OPT` to use an +existing matching binary. + ### Building GUI ```bash @@ -222,4 +233,4 @@ Feel free to: - Join our community discussions - Reach out to maintainers -Thank you for contributing to EasyTier! \ No newline at end of file +Thank you for contributing to EasyTier! diff --git a/easytier-contrib/easytier-ffi/DATA_PLANE_ABI.md b/easytier-contrib/easytier-ffi/DATA_PLANE_ABI.md index ec7b1a8c..3a63568a 100644 --- a/easytier-contrib/easytier-ffi/DATA_PLANE_ABI.md +++ b/easytier-contrib/easytier-ffi/DATA_PLANE_ABI.md @@ -1,4 +1,4 @@ -# Native data-plane ABI v2 +# Native data-plane ABI v3 The native data-plane ABI is a thin adapter over the instance-owned `DataPlaneSession`. It does not own sockets, operation state, completion @@ -13,12 +13,18 @@ queues, routing policy, or timeouts. - `data_plane_completion_drain` returns a non-negative descriptor count or a negative error value. - Handle zero is invalid. -- `timeout_ms == UINT64_MAX` means no deadline. Every other timeout starts when - submission is accepted, including time spent waiting for an I/O direction - lock. +- `timeout_ms == UINT64_MAX` means no deadline. +- TCP connect/bind/accept and UDP bind timeouts start when submission is + accepted. +- TCP streams and UDP sockets have persistent read and write deadlines. + `data_plane_resource_deadline_set` replaces the selected directions' + deadlines immediately, including for active operations. An expired deadline + remains expired until it is replaced or cleared with `UINT64_MAX`. +- Deadline direction `1` selects reads, `2` selects writes, and `3` selects + both. - Request and write bytes are copied before a submit call returns. - Socket-address fields use native-endian integers. Address bytes are in - network order. ABI v2 accepts IPv4 only. + network order. ABI v3 accepts IPv4 only. `DataPlaneSocketAddr` is: @@ -46,6 +52,7 @@ One native session may be open for an EasyTier instance at a time: ```text data_plane_session_open + -> set resource deadlines -> submit operations -> completion_wait -> completion_drain @@ -92,6 +99,7 @@ The exported function families are: - `data_plane_tcp_*_submit` - `data_plane_udp_*_submit` +- `data_plane_resource_deadline_set` - `data_plane_completion_wait` - `data_plane_completion_drain` - `data_plane_*_result_take` diff --git a/easytier-contrib/easytier-ffi/src/data_plane/abi.rs b/easytier-contrib/easytier-ffi/src/data_plane/abi.rs index 71fc1f16..90f9e4bf 100644 --- a/easytier-contrib/easytier-ffi/src/data_plane/abi.rs +++ b/easytier-contrib/easytier-ffi/src/data_plane/abi.rs @@ -13,6 +13,9 @@ use crate::{ types::{DataPlaneCompletion, DataPlaneSocketAddr}, }; +pub const DATA_PLANE_DEADLINE_READ: u32 = 1 << 0; +pub const DATA_PLANE_DEADLINE_WRITE: u32 = 1 << 1; + fn failure(error: NativeDataPlaneError) -> c_int { set_error_msg(&error.message); -(error.kind as c_int) @@ -43,7 +46,7 @@ fn socket_addr(address: DataPlaneSocketAddr) -> NativeDataPlaneResult { return Err(NativeDataPlaneError { kind: DataPlaneErrorKind::AddressFamilyUnsupported, - message: "IPv6 is not supported by data-plane ABI v2".to_string(), + message: "IPv6 is not supported by data-plane ABI v3".to_string(), }); } family => { @@ -210,11 +213,10 @@ pub unsafe extern "C" fn data_plane_tcp_read_submit( session: u64, stream: u64, max_len: u32, - timeout_ms: u64, out_operation: *mut u64, ) -> c_int { write_operation(out_operation, || { - super::session::submit_tcp_read(session, stream, max_len, timeout_ms) + super::session::submit_tcp_read(session, stream, max_len) }) } @@ -229,7 +231,6 @@ pub unsafe extern "C" fn data_plane_tcp_write_submit( stream: u64, data: *const c_uchar, len: u32, - timeout_ms: u64, out_operation: *mut u64, ) -> c_int { let data = match unsafe { copy_input(data, len) } { @@ -237,7 +238,7 @@ pub unsafe extern "C" fn data_plane_tcp_write_submit( Err(error) => return failure(error), }; write_operation(out_operation, || { - super::session::submit_tcp_write(session, stream, data, timeout_ms) + super::session::submit_tcp_write(session, stream, data) }) } @@ -266,11 +267,10 @@ pub unsafe extern "C" fn data_plane_udp_receive_submit( session: u64, socket: u64, max_len: u32, - timeout_ms: u64, out_operation: *mut u64, ) -> c_int { write_operation(out_operation, || { - super::session::submit_udp_receive(session, socket, max_len, timeout_ms) + super::session::submit_udp_receive(session, socket, max_len) }) } @@ -286,7 +286,6 @@ pub unsafe extern "C" fn data_plane_udp_send_submit( peer_addr: DataPlaneSocketAddr, data: *const c_uchar, len: u32, - timeout_ms: u64, out_operation: *mut u64, ) -> c_int { let peer_addr = match socket_addr(peer_addr) { @@ -298,10 +297,27 @@ pub unsafe extern "C" fn data_plane_udp_send_submit( Err(error) => return failure(error), }; write_operation(out_operation, || { - super::session::submit_udp_send(session, socket, peer_addr, data, timeout_ms) + super::session::submit_udp_send(session, socket, peer_addr, data) }) } +#[cfg_attr(feature = "c-abi", unsafe(no_mangle))] +pub extern "C" fn data_plane_resource_deadline_set( + session: u64, + resource: u64, + direction: u32, + timeout_ms: u64, +) -> c_int { + let read = direction & DATA_PLANE_DEADLINE_READ != 0; + let write = direction & DATA_PLANE_DEADLINE_WRITE != 0; + if direction == 0 || direction & !(DATA_PLANE_DEADLINE_READ | DATA_PLANE_DEADLINE_WRITE) != 0 { + return failure(invalid(format!("invalid deadline direction {direction}"))); + } + status(super::session::set_resource_deadline( + session, resource, read, write, timeout_ms, + )) +} + #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub extern "C" fn data_plane_operation_cancel(session: u64, operation: u64) -> c_int { status(super::session::cancel_operation(session, operation)) @@ -628,7 +644,7 @@ mod tests { } #[test] - fn ipv6_is_rejected_by_v2() { + fn ipv6_is_rejected_by_v3() { let error = socket_addr(ffi_socket_addr( "[2001:db8::1]:4321".parse::().unwrap(), )) @@ -646,6 +662,13 @@ mod tests { assert_eq!(error.kind, DataPlaneErrorKind::AddressFamilyUnsupported); } + #[test] + fn invalid_deadline_direction_is_rejected_before_session_lookup() { + let invalid = -(DataPlaneErrorKind::Io as c_int); + assert_eq!(data_plane_resource_deadline_set(u64::MAX, 1, 0, 0), invalid); + assert_eq!(data_plane_resource_deadline_set(u64::MAX, 1, 4, 0), invalid); + } + #[test] fn null_operation_output_does_not_submit() { let submitted = std::cell::Cell::new(false); diff --git a/easytier-contrib/easytier-ffi/src/data_plane/session.rs b/easytier-contrib/easytier-ffi/src/data_plane/session.rs index eafb4fd7..13718c40 100644 --- a/easytier-contrib/easytier-ffi/src/data_plane/session.rs +++ b/easytier-contrib/easytier-ffi/src/data_plane/session.rs @@ -112,10 +112,10 @@ impl NativeDataPlaneSession { self.core.discard_all(); } - fn submit( + fn call( &self, - submit: impl FnOnce(&Arc) -> Result, - ) -> NativeDataPlaneResult { + call: impl FnOnce(&Arc) -> Result, + ) -> NativeDataPlaneResult { let _gate = self .submit_gate .lock() @@ -126,9 +126,14 @@ impl NativeDataPlaneSession { )); } let _runtime = self.runtime.enter(); - submit(&self.core) - .map(DataPlaneOperationId::get) - .map_err(Into::into) + call(&self.core).map_err(Into::into) + } + + fn submit( + &self, + submit: impl FnOnce(&Arc) -> Result, + ) -> NativeDataPlaneResult { + self.call(submit).map(DataPlaneOperationId::get) } } @@ -287,21 +292,18 @@ pub(super) fn submit_tcp_read( session: u64, stream: u64, max_len: u32, - timeout_ms: u64, ) -> NativeDataPlaneResult { let stream = resource_id(stream)?; - get_session(session)? - .submit(|core| core.submit_tcp_read(stream, max_len as usize, timeout(timeout_ms))) + get_session(session)?.submit(|core| core.submit_tcp_read(stream, max_len as usize)) } pub(super) fn submit_tcp_write( session: u64, stream: u64, data: Vec, - timeout_ms: u64, ) -> NativeDataPlaneResult { let stream = resource_id(stream)?; - get_session(session)?.submit(|core| core.submit_tcp_write(stream, data, timeout(timeout_ms))) + get_session(session)?.submit(|core| core.submit_tcp_write(stream, data)) } pub(super) fn submit_udp_bind( @@ -316,11 +318,9 @@ pub(super) fn submit_udp_receive( session: u64, socket: u64, max_len: u32, - timeout_ms: u64, ) -> NativeDataPlaneResult { let socket = resource_id(socket)?; - get_session(session)? - .submit(|core| core.submit_udp_receive(socket, max_len as usize, timeout(timeout_ms))) + get_session(session)?.submit(|core| core.submit_udp_receive(socket, max_len as usize)) } pub(super) fn submit_udp_send( @@ -328,11 +328,21 @@ pub(super) fn submit_udp_send( socket: u64, peer_addr: SocketAddr, data: Vec, - timeout_ms: u64, ) -> NativeDataPlaneResult { let socket = resource_id(socket)?; + get_session(session)?.submit(|core| core.submit_udp_send(socket, peer_addr, data)) +} + +pub(super) fn set_resource_deadline( + session: u64, + resource: u64, + read: bool, + write: bool, + timeout_ms: u64, +) -> NativeDataPlaneResult<()> { + let resource = resource_id(resource)?; get_session(session)? - .submit(|core| core.submit_udp_send(socket, peer_addr, data, timeout(timeout_ms))) + .call(|core| core.set_resource_deadline(resource, read, write, timeout(timeout_ms))) } pub(super) fn cancel_operation(session: u64, operation: u64) -> NativeDataPlaneResult<()> { diff --git a/easytier-contrib/easytier-ffi/src/lib.rs b/easytier-contrib/easytier-ffi/src/lib.rs index b185b602..6c332c89 100644 --- a/easytier-contrib/easytier-ffi/src/lib.rs +++ b/easytier-contrib/easytier-ffi/src/lib.rs @@ -312,8 +312,9 @@ pub extern "C" fn is_config_server_client_connected() -> c_int { #[cfg(feature = "ffi-dataplane")] pub use data_plane::{ - data_plane_completion_drain, data_plane_completion_wait, data_plane_operation_cancel, - data_plane_operation_free, data_plane_resource_close, data_plane_result_size, + DATA_PLANE_DEADLINE_READ, DATA_PLANE_DEADLINE_WRITE, data_plane_completion_drain, + data_plane_completion_wait, data_plane_operation_cancel, data_plane_operation_free, + data_plane_resource_close, data_plane_resource_deadline_set, data_plane_result_size, data_plane_session_close, data_plane_session_open, data_plane_tcp_accept_result_take, data_plane_tcp_accept_submit, data_plane_tcp_bind_result_take, data_plane_tcp_bind_submit, data_plane_tcp_connect_result_take, data_plane_tcp_connect_submit, diff --git a/easytier-contrib/easytier-ffi/src/tests.rs b/easytier-contrib/easytier-ffi/src/tests.rs index 03a78383..c008a66f 100644 --- a/easytier-contrib/easytier-ffi/src/tests.rs +++ b/easytier-contrib/easytier-ffi/src/tests.rs @@ -739,4 +739,8 @@ fn data_plane_invalid_handle_errors_are_stable() { assert_eq!(data_plane_operation_cancel(u64::MAX, 1), closed); assert_eq!(data_plane_operation_free(u64::MAX, 1), closed); assert_eq!(data_plane_resource_close(u64::MAX, 1), closed); + assert_eq!( + data_plane_resource_deadline_set(u64::MAX, 1, DATA_PLANE_DEADLINE_READ, 0), + closed + ); } diff --git a/easytier-core/Cargo.toml b/easytier-core/Cargo.toml index 96d6aca4..c4465008 100644 --- a/easytier-core/Cargo.toml +++ b/easytier-core/Cargo.toml @@ -88,6 +88,7 @@ aes-gcm = ["dep:aes-gcm"] chacha20 = ["dep:chacha20poly1305"] openssl-crypto = ["dep:openssl"] ring-crypto = ["dep:ring"] +wasi-crypto-offload = ["ring-crypto"] config-write = [] endpoint-discovery = [ "dep:http-body-util", diff --git a/easytier-core/src/connectivity/hole_punch/port_mapping.rs b/easytier-core/src/connectivity/hole_punch/port_mapping.rs index 5d78290d..e1bd4653 100644 --- a/easytier-core/src/connectivity/hole_punch/port_mapping.rs +++ b/easytier-core/src/connectivity/hole_punch/port_mapping.rs @@ -258,7 +258,7 @@ async fn run_udp_port_mapping_lifecycle( ) { loop { tokio::select! { - _ = tokio::time::sleep(UPNP_RENEW_INTERVAL) => { + _ = crate::foundation::time::sleep(UPNP_RENEW_INTERVAL) => { if let Err(error) = mapping.renew().await { tracing::warn!( err = ?error, diff --git a/easytier-core/src/connectivity/stun/client.rs b/easytier-core/src/connectivity/stun/client.rs index 5f35e3f4..2987780a 100644 --- a/easytier-core/src/connectivity/stun/client.rs +++ b/easytier-core/src/connectivity/stun/client.rs @@ -218,13 +218,14 @@ where tids: &[u32], stun_host: &SocketAddr, ) -> anyhow::Result<(Message, SocketAddr)> { - let mut now = tokio::time::Instant::now(); + let mut now = crate::foundation::time::Instant::now(); let deadline = now + self.resp_timeout; while now < deadline { let mut receiver = self.stun_packet_receiver.lock().await; - let packet = tokio::time::timeout(deadline - now, receiver.recv()).await??; - now = tokio::time::Instant::now(); + let packet = + crate::foundation::time::timeout(deadline - now, receiver.recv()).await??; + now = crate::foundation::time::Instant::now(); if packet.data.len() < 20 { continue; @@ -773,12 +774,12 @@ where S: AsyncRead + Unpin, { let mut header = [0u8; 20]; - tokio::time::timeout(timeout, stream.read_exact(&mut header)).await??; + crate::foundation::time::timeout(timeout, stream.read_exact(&mut header)).await??; let total_size = Self::message_size_from_header(&header)?; let mut buf = vec![0u8; total_size]; buf[..20].copy_from_slice(&header); if total_size > 20 { - tokio::time::timeout(timeout, stream.read_exact(&mut buf[20..])).await??; + crate::foundation::time::timeout(timeout, stream.read_exact(&mut buf[20..])).await??; } let mut decoder = MessageDecoder::::new(); @@ -808,7 +809,7 @@ where .with_reuse_addr(true) .with_reuse_port(true) .with_only_v6(bind_addr.is_ipv6()); - tokio::time::timeout( + crate::foundation::time::timeout( self.conn_timeout, self.runtime.connect_tcp( TcpConnectOptions::stun_probe(self.stun_server, bind_addr).with_bind(bind), @@ -827,7 +828,7 @@ where let bytes = MessageEncoder::new() .encode_into_bytes(message) .with_context(|| "encode tcp stun message")?; - tokio::time::timeout(self.io_timeout, stream.write_all(&bytes)).await??; + crate::foundation::time::timeout(self.io_timeout, stream.write_all(&bytes)).await??; let now = Instant::now(); let message = Self::tcp_read_stun_message(&mut stream, self.io_timeout).await?; diff --git a/easytier-core/src/connectivity/stun/collector.rs b/easytier-core/src/connectivity/stun/collector.rs index b76b9971..d9704c83 100644 --- a/easytier-core/src/connectivity/stun/collector.rs +++ b/easytier-core/src/connectivity/stun/collector.rs @@ -288,7 +288,7 @@ where tokio::select! { _ = redetect_notify.notified() => {} - _ = tokio::time::sleep(Duration::from_secs(sleep_sec)) => {} + _ = crate::foundation::time::sleep(Duration::from_secs(sleep_sec)) => {} } } }); @@ -325,7 +325,7 @@ where tokio::select! { _ = redetect_notify.notified() => {} - _ = tokio::time::sleep(Duration::from_secs(sleep_sec)) => {} + _ = crate::foundation::time::sleep(Duration::from_secs(sleep_sec)) => {} } } }); @@ -357,7 +357,7 @@ where }; tokio::select! { _ = redetect_notify.notified() => {} - _ = tokio::time::sleep(Duration::from_secs(sleep_sec)) => {} + _ = crate::foundation::time::sleep(Duration::from_secs(sleep_sec)) => {} } } }); diff --git a/easytier-core/src/foundation/operation_broker.rs b/easytier-core/src/foundation/operation_broker.rs index 84f0e796..7c6565fb 100644 --- a/easytier-core/src/foundation/operation_broker.rs +++ b/easytier-core/src/foundation/operation_broker.rs @@ -251,6 +251,23 @@ where !self.completions.is_empty() } + pub(crate) fn pending_kind(&self, operation_id: OperationId) -> Option { + self.operations.get(&operation_id).and_then(|operation| { + matches!(operation.state, OperationState::Pending).then_some(operation.kind) + }) + } + + pub(crate) fn request_cancellation(&self, operation_id: OperationId) -> bool { + let Some(operation) = self.operations.get(&operation_id) else { + return false; + }; + if !matches!(operation.state, OperationState::Pending) { + return false; + } + operation.cancellation.cancel(); + true + } + pub(crate) fn with_drained( &self, operation_id: OperationId, @@ -408,6 +425,21 @@ mod tests { assert!(completions[0].status); } + #[test] + fn requested_cancellation_stays_pending_until_operation_completes() { + let mut broker = OperationBroker::new(4); + let admission = broker.admit(Kind::Write, "metadata").unwrap(); + + assert_eq!(broker.pending_kind(admission.id), Some(Kind::Write)); + assert!(broker.request_cancellation(admission.id)); + assert!(admission.cancellation.is_cancelled()); + assert!(!broker.has_completions()); + + assert!(broker.complete_with(admission.id, |_, _| Ok::<_, &'static str>(7))); + assert!(broker.has_completions()); + assert_eq!(broker.pending_kind(admission.id), None); + } + #[test] fn completion_notification_is_an_empty_to_nonempty_edge() { let mut broker = OperationBroker::new(4); diff --git a/easytier-core/src/foundation/time.rs b/easytier-core/src/foundation/time.rs index 95a8646f..dc4bfabb 100644 --- a/easytier-core/src/foundation/time.rs +++ b/easytier-core/src/foundation/time.rs @@ -5,10 +5,12 @@ //! the guest without polling. #[cfg(not(any(test, target_os = "wasi")))] -pub use tokio::time::{Duration, Instant, Interval, error, interval, sleep, timeout}; +pub use tokio::time::{Duration, Instant, Interval, error, interval, sleep, sleep_until, timeout}; #[cfg(any(test, target_os = "wasi"))] -pub use crate::wasi::time::{Duration, Instant, Interval, error, interval, sleep, timeout}; +pub use crate::wasi::time::{ + Duration, Instant, Interval, error, interval, sleep, sleep_until, timeout, +}; #[cfg(target_os = "wasi")] pub(crate) use crate::wasi::time::{clear_domain, enter_domain, next_deadline_millis}; diff --git a/easytier-core/src/gateway/dataplane/deadline.rs b/easytier-core/src/gateway/dataplane/deadline.rs index a1128b9e..aff2f84f 100644 --- a/easytier-core/src/gateway/dataplane/deadline.rs +++ b/easytier-core/src/gateway/dataplane/deadline.rs @@ -1,8 +1,10 @@ -//! One absolute deadline shared by every stage of a data-plane operation. +//! Deadlines for data-plane control operations and persistent I/O resources. -use std::{future::Future, time::Duration}; +use std::{future::Future, sync::Mutex, time::Duration}; use quanta::Instant; +use tokio::{sync::watch, task::JoinHandle}; +use tokio_util::sync::CancellationToken; use crate::foundation::time; @@ -50,3 +52,164 @@ impl DataPlaneDeadline { } } } + +pub(super) struct DataPlaneIoDeadline { + generation: watch::Sender, + timer: Mutex>>, +} + +impl Default for DataPlaneIoDeadline { + fn default() -> Self { + Self { + generation: watch::channel(CancellationToken::new()).0, + timer: Mutex::new(None), + } + } +} + +impl DataPlaneIoDeadline { + pub(super) fn set_timeout(&self, timeout: Option) { + let mut timer = self.timer.lock().unwrap_or_else(|error| error.into_inner()); + if let Some(timer) = timer.take() { + timer.abort(); + } + + let expired = CancellationToken::new(); + if let Some(timeout) = timeout { + if timeout.is_zero() { + expired.cancel(); + } else { + if let Some(deadline) = time::Instant::now().checked_add(timeout) { + let sleep = time::sleep_until(deadline); + let expiration = expired.clone(); + *timer = Some(tokio::spawn(async move { + sleep.await; + expiration.cancel(); + })); + } + } + } + self.generation.send_replace(expired); + } + + pub(super) async fn run( + &self, + cancel: CancellationToken, + future: impl Future>, + ) -> DataPlaneResult + where + E: Into, + { + let mut deadline = self.generation.subscribe(); + tokio::pin!(future); + loop { + let expired = deadline.borrow_and_update().clone(); + tokio::select! { + biased; + _ = cancel.cancelled() => { + return Err(DataPlaneError::new( + super::DataPlaneErrorKind::Cancelled, + "data-plane operation cancelled", + )); + } + changed = deadline.changed() => { + if changed.is_err() { + return Err(DataPlaneError::new( + super::DataPlaneErrorKind::HandleClosed, + "data-plane resource is closed", + )); + } + } + _ = expired.cancelled() => { + return Err(DataPlaneError::deadline_exceeded()); + } + result = &mut future => return result.map_err(Into::into), + } + } + } +} + +impl Drop for DataPlaneIoDeadline { + fn drop(&mut self) { + if let Some(timer) = self + .timer + .get_mut() + .unwrap_or_else(|error| error.into_inner()) + .take() + { + timer.abort(); + } + } +} + +#[cfg(test)] +mod tests { + use std::{future, sync::Arc}; + + use super::*; + + #[tokio::test] + async fn io_deadline_updates_an_active_operation() { + let deadline = Arc::new(DataPlaneIoDeadline::default()); + let operation = { + let deadline = deadline.clone(); + tokio::spawn(async move { + deadline + .run( + CancellationToken::new(), + future::pending::>(), + ) + .await + }) + }; + tokio::task::yield_now().await; + + deadline.set_timeout(Some(Duration::ZERO)); + + assert_eq!( + operation.await.unwrap().unwrap_err().kind(), + super::super::DataPlaneErrorKind::DeadlineExceeded + ); + } + + #[tokio::test] + async fn io_deadline_expires_future_operations_until_cleared() { + let deadline = DataPlaneIoDeadline::default(); + deadline.set_timeout(Some(Duration::ZERO)); + + assert_eq!( + deadline + .run( + CancellationToken::new(), + future::ready(Ok::<_, DataPlaneError>(())), + ) + .await + .unwrap_err() + .kind(), + super::super::DataPlaneErrorKind::DeadlineExceeded + ); + + deadline.set_timeout(None); + deadline + .run( + CancellationToken::new(), + future::ready(Ok::<_, DataPlaneError>(())), + ) + .await + .unwrap(); + } + + #[tokio::test] + async fn io_deadline_treats_unrepresentable_timeout_as_unbounded() { + let deadline = DataPlaneIoDeadline::default(); + deadline.set_timeout(Some(Duration::from_millis(u64::MAX - 1))); + + deadline + .run( + CancellationToken::new(), + future::ready(Ok::<_, DataPlaneError>(())), + ) + .await + .unwrap(); + } +} diff --git a/easytier-core/src/gateway/dataplane/mod.rs b/easytier-core/src/gateway/dataplane/mod.rs index 661f0430..9fa3dc76 100644 --- a/easytier-core/src/gateway/dataplane/mod.rs +++ b/easytier-core/src/gateway/dataplane/mod.rs @@ -71,7 +71,7 @@ mod tests; mod udp; use self::{ - deadline::DataPlaneDeadline, + deadline::{DataPlaneDeadline, DataPlaneIoDeadline}, error::DataPlaneResult, flow::{FlowKey, FlowKind, FlowLease, FlowTable}, packet::PeerPacketRoute, diff --git a/easytier-core/src/gateway/dataplane/session.rs b/easytier-core/src/gateway/dataplane/session.rs index df9c6047..bc475d37 100644 --- a/easytier-core/src/gateway/dataplane/session.rs +++ b/easytier-core/src/gateway/dataplane/session.rs @@ -25,9 +25,9 @@ use crate::{ }; use super::{ - DataPlaneConsumerLease, DataPlaneDeadline, DataPlaneError, DataPlaneErrorKind, DataPlaneResult, - DataPlaneRuntime, DataPlaneTcpConnectOptions, DataPlaneTcpListener, DataPlaneTcpStream, - DataPlaneUdpSocket, + DataPlaneConsumerLease, DataPlaneDeadline, DataPlaneError, DataPlaneErrorKind, + DataPlaneIoDeadline, DataPlaneResult, DataPlaneRuntime, DataPlaneTcpConnectOptions, + DataPlaneTcpListener, DataPlaneTcpStream, DataPlaneUdpSocket, operation::{ DataPlaneCompletionDescriptor, DataPlaneCompletionStatus, DataPlaneOperationId, DataPlaneOperationKind, DataPlaneOperationOutcome, DataPlaneOperationResult, @@ -62,12 +62,39 @@ impl Default for DataPlaneSessionLimits { struct TcpResource { read: AsyncMutex>, write: AsyncMutex>, + read_deadline: DataPlaneIoDeadline, + write_deadline: DataPlaneIoDeadline, } struct UdpResource { socket: Arc, read: AsyncMutex<()>, write: AsyncMutex<()>, + read_deadline: DataPlaneIoDeadline, + write_deadline: DataPlaneIoDeadline, +} + +async fn write_tcp_payload( + writer: &mut (impl tokio::io::AsyncWrite + Unpin), + data: &[u8], + cancel: &CancellationToken, + deadline: &DataPlaneIoDeadline, +) -> DataPlaneResult { + let mut written = 0; + while written < data.len() { + let result = deadline + .run(cancel.clone(), writer.write(&data[written..])) + .await; + match result { + Ok(0) if written == 0 => { + return Err(std::io::Error::from(std::io::ErrorKind::WriteZero).into()); + } + Ok(0) | Err(_) if written > 0 => return Ok(written), + Ok(len) => written += len, + Err(error) => return Err(error), + } + } + Ok(written) } #[derive(Clone)] @@ -525,11 +552,9 @@ where self: &Arc, stream_id: DataPlaneResourceId, max_len: usize, - timeout: Option, ) -> DataPlaneResult { Self::ensure_executor()?; self.require_read_size(max_len)?; - let deadline = DataPlaneDeadline::from_optional_timeout(timeout); let (stream, operation_id, cancel) = { let mut state = self.lock_state(); let stream = Self::require_tcp(&state, stream_id)?; @@ -553,13 +578,15 @@ where return Ok(operation_id); } self.spawn_operation(operation_id, async move { - let (data, eof) = Self::run_operation(cancel, deadline, async move { - let mut data = vec![0u8; max_len]; - let len = stream.read.lock().await.read(&mut data).await?; - data.truncate(len); - Ok::<_, std::io::Error>((data, len == 0)) - }) - .await?; + let (data, eof) = stream + .read_deadline + .run(cancel, async { + let mut data = vec![0u8; max_len]; + let len = stream.read.lock().await.read(&mut data).await?; + data.truncate(len); + Ok::<_, std::io::Error>((data, len == 0)) + }) + .await?; Ok(PendingOperationResult::TcpRead { data, eof }) }); Ok(operation_id) @@ -569,10 +596,8 @@ where self: &Arc, stream_id: DataPlaneResourceId, data: Vec, - timeout: Option, ) -> DataPlaneResult { Self::ensure_executor()?; - let deadline = DataPlaneDeadline::from_optional_timeout(timeout); let (stream, operation_id, cancel) = { let mut state = self.lock_state(); let stream = Self::require_tcp(&state, stream_id)?; @@ -586,10 +611,14 @@ where (stream, operation_id, cancel) }; self.spawn_operation(operation_id, async move { - let len = Self::run_operation(cancel, deadline, async move { - stream.write.lock().await.write(&data).await - }) - .await?; + let mut writer = stream + .write_deadline + .run(cancel.clone(), async { + Ok::<_, DataPlaneError>(stream.write.lock().await) + }) + .await?; + let len = + write_tcp_payload(&mut *writer, &data, &cancel, &stream.write_deadline).await?; Ok(PendingOperationResult::TcpWritten(len)) }); Ok(operation_id) @@ -623,11 +652,9 @@ where self: &Arc, socket_id: DataPlaneResourceId, max_len: usize, - timeout: Option, ) -> DataPlaneResult { Self::ensure_executor()?; self.require_read_size(max_len)?; - let deadline = DataPlaneDeadline::from_optional_timeout(timeout); let (socket, operation_id, cancel) = { let mut state = self.lock_state(); let socket = Self::require_udp(&state, socket_id)?; @@ -641,11 +668,13 @@ where (socket, operation_id, cancel) }; self.spawn_operation(operation_id, async move { - let (data, peer_addr, truncated) = Self::run_operation(cancel, deadline, async move { - let _read = socket.read.lock().await; - socket.socket.recv_from_limited(max_len).await - }) - .await?; + let (data, peer_addr, truncated) = socket + .read_deadline + .run(cancel, async { + let _read = socket.read.lock().await; + socket.socket.recv_from_limited(max_len).await + }) + .await?; Ok(PendingOperationResult::UdpReceived { data, peer_addr, @@ -660,10 +689,8 @@ where socket_id: DataPlaneResourceId, peer_addr: SocketAddr, data: Vec, - timeout: Option, ) -> DataPlaneResult { Self::ensure_executor()?; - let deadline = DataPlaneDeadline::from_optional_timeout(timeout); let (socket, operation_id, cancel) = { let mut state = self.lock_state(); let socket = Self::require_udp(&state, socket_id)?; @@ -677,16 +704,57 @@ where (socket, operation_id, cancel) }; self.spawn_operation(operation_id, async move { - let len = Self::run_operation(cancel, deadline, async move { - let _write = socket.write.lock().await; - socket.socket.send_to(&data, peer_addr).await - }) - .await?; + let len = socket + .write_deadline + .run(cancel, async { + let _write = socket.write.lock().await; + socket.socket.send_to(&data, peer_addr).await + }) + .await?; Ok(PendingOperationResult::UdpSent(len)) }); Ok(operation_id) } + pub fn set_resource_deadline( + &self, + resource_id: DataPlaneResourceId, + read: bool, + write: bool, + timeout: Option, + ) -> DataPlaneResult<()> { + Self::ensure_executor()?; + let resource = { + let state = self.lock_state(); + Self::resource_io(&state, resource_id)? + }; + match resource { + ResourceIo::Tcp(resource) => { + if read { + resource.read_deadline.set_timeout(timeout); + } + if write { + resource.write_deadline.set_timeout(timeout); + } + } + ResourceIo::Udp(resource) => { + if read { + resource.read_deadline.set_timeout(timeout); + } + if write { + resource.write_deadline.set_timeout(timeout); + } + } + ResourceIo::TcpListener(_) => { + return Err(Self::error( + DataPlaneErrorKind::HandleClosed, + "data-plane resource does not support I/O deadlines", + )); + } + } + Ok(()) + } + fn unlink_target_locked( resources: &mut ResourceTable, operation_id: DataPlaneOperationId, @@ -730,6 +798,8 @@ where io: ResourceIo::Tcp(Arc::new(TcpResource { read: AsyncMutex::new(read), write: AsyncMutex::new(write), + read_deadline: DataPlaneIoDeadline::default(), + write_deadline: DataPlaneIoDeadline::default(), })), pending_operations: HashSet::new(), }, @@ -764,6 +834,8 @@ where socket: Arc::new(socket), read: AsyncMutex::new(()), write: AsyncMutex::new(()), + read_deadline: DataPlaneIoDeadline::default(), + write_deadline: DataPlaneIoDeadline::default(), })), pending_operations: HashSet::new(), }, @@ -938,8 +1010,14 @@ where pub fn cancel_operation(&self, operation_id: DataPlaneOperationId) { let mut state = self.lock_state(); + let broker_id = operation_id.broker_id(); let notify = - Self::queue_error_locked(&mut state, operation_id, DataPlaneErrorKind::Cancelled); + if state.broker.pending_kind(broker_id) == Some(DataPlaneOperationKind::TcpWrite) { + state.broker.request_cancellation(broker_id); + false + } else { + Self::queue_error_locked(&mut state, operation_id, DataPlaneErrorKind::Cancelled) + }; drop(state); if notify { self.notify_completion(); @@ -1209,10 +1287,14 @@ where #[cfg(test)] mod tests { use std::{ + pin::Pin, sync::{Arc, Barrier}, + task::{Context, Poll}, thread, }; + use tokio::io::AsyncWrite; + use crate::host::testkit::TestHost; use super::*; @@ -1231,6 +1313,118 @@ mod tests { Ok(PendingOperationResult::TcpWritten(len)) } + #[tokio::test] + async fn tcp_write_payload_waits_for_the_full_buffer() { + let expected = b"larger than the duplex capacity".to_vec(); + let (mut writer, mut reader) = tokio::io::duplex(4); + let payload = expected.clone(); + let write = tokio::spawn(async move { + let deadline = DataPlaneIoDeadline::default(); + write_tcp_payload(&mut writer, &payload, &CancellationToken::new(), &deadline) + .await + .unwrap() + }); + let mut received = Vec::new(); + + reader.read_to_end(&mut received).await.unwrap(); + + assert_eq!(write.await.unwrap(), expected.len()); + assert_eq!(received, expected); + } + + struct PrefixThenPendingWriter { + first_write: Option>, + prefix_len: usize, + } + + impl AsyncWrite for PrefixThenPendingWriter { + fn poll_write( + mut self: Pin<&mut Self>, + _context: &mut Context<'_>, + data: &[u8], + ) -> Poll> { + let Some(first_write) = self.first_write.take() else { + return Poll::Pending; + }; + let len = self.prefix_len.min(data.len()); + let _ = first_write.send(()); + Poll::Ready(Ok(len)) + } + + fn poll_flush( + self: Pin<&mut Self>, + _context: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown( + self: Pin<&mut Self>, + _context: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + } + + #[tokio::test] + async fn tcp_write_payload_reports_progress_when_cancelled() { + let (first_write_tx, first_write_rx) = tokio::sync::oneshot::channel(); + let mut writer = PrefixThenPendingWriter { + first_write: Some(first_write_tx), + prefix_len: 4, + }; + let cancel = CancellationToken::new(); + let write_cancel = cancel.clone(); + let write = tokio::spawn(async move { + let deadline = DataPlaneIoDeadline::default(); + write_tcp_payload(&mut writer, b"partial payload", &write_cancel, &deadline).await + }); + + first_write_rx.await.unwrap(); + cancel.cancel(); + + assert_eq!(write.await.unwrap().unwrap(), 4); + } + + #[tokio::test] + async fn tcp_write_payload_reports_progress_when_deadline_expires() { + let (first_write_tx, first_write_rx) = tokio::sync::oneshot::channel(); + let mut writer = PrefixThenPendingWriter { + first_write: Some(first_write_tx), + prefix_len: 5, + }; + let write = tokio::spawn(async move { + let deadline = DataPlaneIoDeadline::default(); + deadline.set_timeout(Some(Duration::from_millis(10))); + write_tcp_payload( + &mut writer, + b"partial payload", + &CancellationToken::new(), + &deadline, + ) + .await + }); + + first_write_rx.await.unwrap(); + + assert_eq!(write.await.unwrap().unwrap(), 5); + } + + #[test] + fn tcp_write_cancellation_waits_for_the_progress_outcome() { + let session = session(); + let operation_id = session + .admit_test_operation(DataPlaneOperationKind::TcpWrite, 0) + .unwrap(); + + session.cancel_operation(operation_id); + assert!(session.drain_completions(1).is_empty()); + + session.complete_test_operation(operation_id, successful_write(4)); + let completion = session.drain_completions(1).pop().unwrap(); + assert_eq!(completion.status, DataPlaneCompletionStatus::Success); + } + #[test] fn completion_is_drained_once_and_result_is_taken_once() { let session = session(); diff --git a/easytier-core/src/gateway/dataplane/stack.rs b/easytier-core/src/gateway/dataplane/stack.rs index bc78fe83..bc8a2386 100644 --- a/easytier-core/src/gateway/dataplane/stack.rs +++ b/easytier-core/src/gateway/dataplane/stack.rs @@ -95,6 +95,8 @@ impl SmoltcpPlane { Some(BufferSize { tcp_rx_size: 1024 * 128, tcp_tx_size: 1024 * 128, + udp_rx_size: 1024 * 128, + udp_rx_meta_size: 128, ..Default::default() }), ), diff --git a/easytier-core/src/gateway/dataplane/tests.rs b/easytier-core/src/gateway/dataplane/tests.rs index d56908b0..377abb13 100644 --- a/easytier-core/src/gateway/dataplane/tests.rs +++ b/easytier-core/src/gateway/dataplane/tests.rs @@ -308,11 +308,9 @@ async fn data_plane_sessions_complete_tcp_operations_end_to_end() { .unwrap() .unwrap(); - let read = session_b - .submit_tcp_read(server, 16, Some(Duration::from_secs(10))) - .unwrap(); + let read = session_b.submit_tcp_read(server, 16).unwrap(); let write = session_a - .submit_tcp_write(client, b"ping".to_vec(), Some(Duration::from_secs(10))) + .submit_tcp_write(client, b"ping".to_vec()) .unwrap(); let (write_completion, read_completion) = tokio::join!( wait_for_session_completion(&session_a), @@ -337,7 +335,7 @@ async fn data_plane_sessions_complete_tcp_operations_end_to_end() { assert_eq!(written, 4); assert_eq!(received, b"ping"); - let blocked_read = session_b.submit_tcp_read(server, 16, None).unwrap(); + let blocked_read = session_b.submit_tcp_read(server, 16).unwrap(); session_b.close_resource(server); let close_completion = wait_for_session_completion(&session_b).await; assert_eq!(close_completion.operation_id, blocked_read); @@ -351,7 +349,7 @@ async fn data_plane_sessions_complete_tcp_operations_end_to_end() { .unwrap(); assert_eq!(close_error, DataPlaneErrorKind::HandleClosed); - let stopped_read = session_a.submit_tcp_read(client, 16, None).unwrap(); + let stopped_read = session_a.submit_tcp_read(client, 16).unwrap(); session_a.stop(); let stop_completion = wait_for_session_completion(&session_a).await; assert_eq!(stop_completion.operation_id, stopped_read); @@ -402,7 +400,7 @@ async fn data_plane_sessions_report_udp_truncation() { .unwrap(); let warmup = session_b - .submit_udp_send(socket_b, addr_a, b"warmup".to_vec(), None) + .submit_udp_send(socket_b, addr_a, b"warmup".to_vec()) .unwrap(); wait_for_session_completion(&session_b).await; session_b @@ -413,16 +411,9 @@ async fn data_plane_sessions_report_udp_truncation() { .unwrap() .unwrap(); - let receive = session_b - .submit_udp_receive(socket_b, 2, Some(Duration::from_secs(10))) - .unwrap(); + let receive = session_b.submit_udp_receive(socket_b, 2).unwrap(); let send = session_a - .submit_udp_send( - socket_a, - addr_b, - b"ping".to_vec(), - Some(Duration::from_secs(10)), - ) + .submit_udp_send(socket_a, addr_b, b"ping".to_vec()) .unwrap(); let (send_completion, receive_completion) = tokio::join!( wait_for_session_completion(&session_a), diff --git a/easytier-core/src/gateway/port_forward.rs b/easytier-core/src/gateway/port_forward.rs index 93f4cc54..e58792b3 100644 --- a/easytier-core/src/gateway/port_forward.rs +++ b/easytier-core/src/gateway/port_forward.rs @@ -373,7 +373,7 @@ where let response_tasks = self.udp_response_tasks.clone(); self.tasks.lock().unwrap().spawn(async move { loop { - tokio::time::sleep(Duration::from_secs(30)).await; + crate::foundation::time::sleep(Duration::from_secs(30)).await; let now = Instant::now(); udp_clients.retain(|_, client| { now.duration_since(client.last_active.load()).as_secs() < 600 diff --git a/easytier-core/src/gateway/smoltcp/tokio_smoltcp/reactor.rs b/easytier-core/src/gateway/smoltcp/tokio_smoltcp/reactor.rs index f7371a0e..73c44e44 100644 --- a/easytier-core/src/gateway/smoltcp/tokio_smoltcp/reactor.rs +++ b/easytier-core/src/gateway/smoltcp/tokio_smoltcp/reactor.rs @@ -64,15 +64,17 @@ async fn run( .unwrap_or(default_timeout) }; - timer - .as_mut() - .reset(crate::foundation::time::Instant::now() + deadline.into()); - select! { - _ = &mut timer => {}, - _ = receive(&mut async_iface,&mut recv_buf) => {} - _ = notify.notified() => {} - _ = stopper.notified() => break, - }; + if deadline != Duration::ZERO { + timer + .as_mut() + .reset(crate::foundation::time::Instant::now() + deadline.into()); + select! { + _ = &mut timer => {}, + _ = receive(&mut async_iface,&mut recv_buf) => {} + _ = notify.notified() => {} + _ = stopper.notified() => break, + }; + } while let (true, Some(Ok(p))) = ( recv_buf.len() < max_burst_size, diff --git a/easytier-core/src/host/socket/mod.rs b/easytier-core/src/host/socket/mod.rs index af7f9b34..8e295343 100644 --- a/easytier-core/src/host/socket/mod.rs +++ b/easytier-core/src/host/socket/mod.rs @@ -14,6 +14,7 @@ use std::{ collections::HashMap, fmt, io, + io::IoSlice, net::SocketAddr, pin::Pin, sync::{Arc, LazyLock, Mutex, atomic::Ordering}, @@ -296,6 +297,8 @@ pub struct HostTcpStream { closed: bool, } +const HOST_TCP_READ_CAPACITY: usize = 64 * 1024; + impl HostTcpStream { fn close(&mut self) -> io::Result<()> { if self.closed { @@ -352,6 +355,40 @@ impl HostTcpStream { } } } + + fn poll_submit_write( + &mut self, + context: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + if self.closed { + return Poll::Ready(Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "host TCP stream is closed", + ))); + } + if buffer.is_empty() { + return Poll::Ready(Ok(0)); + } + + match self.poll_write_completion(context) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), + Poll::Ready(Ok(())) => {} + } + + let operation = self.runtime.next_operation(); + if let Err(error) = self.io.submit_write(self.handle, operation, buffer) { + return Poll::Ready(Err(error)); + } + self.write_operation = Some(PendingHostOperation::new( + self.runtime.clone(), + self.io.clone(), + operation, + |io, operation| io.cancel_operation(operation), + )); + Poll::Ready(Ok(buffer.len())) + } } impl fmt::Debug for HostTcpStream { @@ -383,10 +420,11 @@ impl AsyncRead for HostTcpStream { loop { if self.read_operation.is_none() { let operation = self.runtime.next_operation(); - if let Err(error) = self - .io - .submit_read(self.handle, operation, buffer.remaining()) - { + if let Err(error) = self.io.submit_read( + self.handle, + operation, + buffer.remaining().max(HOST_TCP_READ_CAPACITY), + ) { return Poll::Ready(Err(error)); } self.read_operation = Some(PendingHostOperation::new( @@ -431,33 +469,31 @@ impl AsyncWrite for HostTcpStream { context: &mut Context<'_>, buffer: &[u8], ) -> Poll> { - if self.closed { - return Poll::Ready(Err(io::Error::new( - io::ErrorKind::BrokenPipe, - "host TCP stream is closed", - ))); + self.poll_submit_write(context, buffer) + } + + fn poll_write_vectored( + mut self: Pin<&mut Self>, + context: &mut Context<'_>, + buffers: &[IoSlice<'_>], + ) -> Poll> { + if buffers.len() == 1 { + return self.poll_submit_write(context, &buffers[0]); } - if buffer.is_empty() { - return Poll::Ready(Ok(0)); + let length = buffers.iter().map(|buffer| buffer.len()).sum(); + if length == 0 { + return self.poll_submit_write(context, &[]); } - match self.poll_write_completion(context) { - Poll::Pending => return Poll::Pending, - Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), - Poll::Ready(Ok(())) => {} + let mut combined = Vec::with_capacity(length); + for buffer in buffers { + combined.extend_from_slice(buffer); } + self.poll_submit_write(context, &combined) + } - let operation = self.runtime.next_operation(); - if let Err(error) = self.io.submit_write(self.handle, operation, buffer) { - return Poll::Ready(Err(error)); - } - self.write_operation = Some(PendingHostOperation::new( - self.runtime.clone(), - self.io.clone(), - operation, - |io, operation| io.cancel_operation(operation), - )); - Poll::Ready(Ok(buffer.len())) + fn is_write_vectored(&self) -> bool { + true } fn poll_flush(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll> { @@ -513,7 +549,10 @@ mod tests { } enum TestOperation { - Read(Option>>), + Read { + capacity: usize, + result: Option>>, + }, Write { source: Vec, result: Option>, @@ -536,7 +575,7 @@ mod tests { .unwrap() .iter() .find_map(|(id, operation)| match (read, operation) { - (true, TestOperation::Read(_)) | (false, TestOperation::Write { .. }) => { + (true, TestOperation::Read { .. }) | (false, TestOperation::Write { .. }) => { Some(*id) } _ => None, @@ -552,9 +591,17 @@ mod tests { source.clone() } + fn read_capacity(&self, operation: HostOperationId) -> usize { + let operations = self.operations.lock().unwrap(); + let TestOperation::Read { capacity, .. } = operations.get(&operation).unwrap() else { + panic!("operation is not a read"); + }; + *capacity + } + fn complete_read(&self, operation: HostOperationId, data: Vec) { let mut operations = self.operations.lock().unwrap(); - let TestOperation::Read(result) = operations.get_mut(&operation).unwrap() else { + let TestOperation::Read { result, .. } = operations.get_mut(&operation).unwrap() else { panic!("operation is not a read"); }; *result = Some(Ok(data)); @@ -597,12 +644,15 @@ mod tests { &self, _handle: HostSocketHandle, operation: HostOperationId, - _capacity: usize, + capacity: usize, ) -> io::Result<()> { - self.operations - .lock() - .unwrap() - .insert(operation, TestOperation::Read(None)); + self.operations.lock().unwrap().insert( + operation, + TestOperation::Read { + capacity, + result: None, + }, + ); Ok(()) } @@ -617,7 +667,7 @@ mod tests { return Poll::Pending; } let mut operations = self.operations.lock().unwrap(); - let Some(TestOperation::Read(result)) = operations.get_mut(&operation) else { + let Some(TestOperation::Read { result, .. }) = operations.get_mut(&operation) else { return Poll::Ready(Err(io::Error::new( io::ErrorKind::NotFound, "read operation is missing", @@ -714,6 +764,42 @@ mod tests { assert!(io.closed.lock().unwrap().contains(&HostSocketHandle(7))); } + #[tokio::test] + async fn small_reads_submit_one_bounded_read_ahead_operation() { + let io = Arc::new(TestHostIo::default()); + let (runtime, mut stream) = test_stream(io.clone()); + let mut first = [0_u8; 1]; + let mut read = Box::pin(stream.read(&mut first)); + assert!(futures::poll!(&mut read).is_pending()); + + let operation = io.operation(true); + assert_eq!(io.read_capacity(operation), HOST_TCP_READ_CAPACITY); + io.complete_read(operation, b"abc".to_vec()); + runtime.notify_completions(); + assert_eq!(read.await.unwrap(), 1); + assert_eq!(&first, b"a"); + + let mut remainder = [0_u8; 2]; + stream.read_exact(&mut remainder).await.unwrap(); + assert_eq!(&remainder, b"bc"); + } + + #[tokio::test] + async fn vectored_write_submits_one_ordered_host_operation() { + let io = Arc::new(TestHostIo::default()); + let (runtime, mut stream) = test_stream(io.clone()); + assert!(stream.is_write_vectored()); + + let buffers = [IoSlice::new(b"one"), IoSlice::new(b"two")]; + assert_eq!(stream.write_vectored(&buffers).await.unwrap(), 6); + let operation = io.operation(false); + assert_eq!(io.write_source(operation), b"onetwo"); + + io.complete_write(operation); + runtime.notify_completions(); + stream.shutdown().await.unwrap(); + } + #[tokio::test] async fn cancelled_read_keeps_owned_completion_remainder() { let io = Arc::new(TestHostIo::default()); diff --git a/easytier-core/src/management/full/web_client.rs b/easytier-core/src/management/full/web_client.rs index bbdd0baf..89564e36 100644 --- a/easytier-core/src/management/full/web_client.rs +++ b/easytier-core/src/management/full/web_client.rs @@ -10,12 +10,13 @@ use easytier_proto::{ WebServerServiceClientFactory, }, }; -use tokio::{sync::Mutex, task::JoinSet, time::interval}; +use tokio::{sync::Mutex, task::JoinSet}; use tokio_util::task::AbortOnDropHandle; use url::Url; use crate::{ connectivity::protocol::raw::TunnelDialer, + foundation::time, instance::{CoreInstance, CoreInstanceHost, manager::InstanceFactory}, rpc::{bidirect::BidirectRpcManager, service_registry::ServiceRegistry}, tunnel::{Tunnel, web_security}, @@ -171,7 +172,7 @@ where Ok(connection) => connection, Err(error) => { tracing::warn!(%error, "failed to connect to config server; retrying"); - tokio::time::sleep(RETRY_INTERVAL).await; + time::sleep(RETRY_INTERVAL).await; continue; } }; @@ -180,7 +181,7 @@ where tracing::info!(?connection, "connected to config server"); let mut session = WebClientSession::new(connection, controller.clone()); let support_encryption = - match tokio::time::timeout(FEATURE_TIMEOUT, session.get_feature()).await { + match time::timeout(FEATURE_TIMEOUT, session.get_feature()).await { Ok(Ok(feature)) => feature.support_encryption, Ok(Err(error)) => { tracing::warn!(%error, "GetFeature RPC failed; using legacy tunnel"); @@ -199,7 +200,7 @@ where Err(error) => { connected.store(false, Ordering::Release); tracing::warn!(%error, "failed to reconnect secure config-server tunnel"); - tokio::time::sleep(RETRY_INTERVAL).await; + time::sleep(RETRY_INTERVAL).await; continue; } }; @@ -208,7 +209,7 @@ where Err(error) => { connected.store(false, Ordering::Release); tracing::warn!(%error, "config-server secure handshake failed"); - tokio::time::sleep(RETRY_INTERVAL).await; + time::sleep(RETRY_INTERVAL).await; continue; } }; @@ -225,7 +226,7 @@ where tracing::warn!( "secure mode requires web secure-tunnel support in the local build" ); - tokio::time::sleep(RETRY_INTERVAL).await; + time::sleep(RETRY_INTERVAL).await; continue; } tracing::warn!( @@ -235,7 +236,7 @@ where if controller.config.secure_mode { connected.store(false, Ordering::Release); tracing::warn!("secure mode requires config-server encryption support"); - tokio::time::sleep(RETRY_INTERVAL).await; + time::sleep(RETRY_INTERVAL).await; continue; } @@ -302,7 +303,7 @@ where let client = rpc .rpc_client() .scoped_client::>(1, 1, String::new()); - let mut tick = interval(std::time::Duration::from_secs(1)); + let mut tick = time::interval(std::time::Duration::from_secs(1)); tasks.spawn(async move { loop { diff --git a/easytier-core/src/socket/udp/layer.rs b/easytier-core/src/socket/udp/layer.rs index a5279139..ae2ab934 100644 --- a/easytier-core/src/socket/udp/layer.rs +++ b/easytier-core/src/socket/udp/layer.rs @@ -447,7 +447,7 @@ pub(super) async fn udp_session_layer_recv_task( { let control_permits = Arc::new(Semaphore::new(UDP_SESSION_QUEUE_CAPACITY)); loop { - let datagram = match socket.recv_datagram().await { + let datagram = match socket.recv_session_datagram().await { Ok(datagram) => datagram, Err(err) => { tracing::debug!(?err, "udp session recv loop stopped"); diff --git a/easytier-core/src/socket/udp/tests.rs b/easytier-core/src/socket/udp/tests.rs index 301932d5..9be9cc39 100644 --- a/easytier-core/src/socket/udp/tests.rs +++ b/easytier-core/src/socket/udp/tests.rs @@ -301,6 +301,7 @@ async fn udp_session_listener_reports_bound_local_addr_before_accept() { struct MockVirtualUdpSocket { local_addr: SocketAddr, incoming: Mutex, SocketAddr)>>, + recv_capacities: Mutex>, sent: Mutex, SocketAddr)>>, send_attempts: Mutex, SocketAddr, UdpSocketSendMeta)>>, reject_preferred_source: AtomicBool, @@ -311,6 +312,7 @@ impl MockVirtualUdpSocket { Self { local_addr, incoming: Mutex::new(incoming.into()), + recv_capacities: Mutex::new(Vec::new()), sent: Mutex::new(Vec::new()), send_attempts: Mutex::new(Vec::new()), reject_preferred_source: AtomicBool::new(false), @@ -357,6 +359,7 @@ impl VirtualUdpSocket for MockVirtualUdpSocket { } async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { + self.recv_capacities.lock().unwrap().push(buf.len()); let (data, remote_addr) = self.incoming.lock().unwrap().pop_front().ok_or_else(|| { io::Error::new(io::ErrorKind::UnexpectedEof, "no incoming datagram") @@ -367,6 +370,32 @@ impl VirtualUdpSocket for MockVirtualUdpSocket { } } +#[tokio::test] +async fn portable_udp_receive_keeps_general_and_session_capacities_separate() { + let local_addr = SocketAddr::from(([127, 0, 0, 1], 12000)); + let peer_addr = SocketAddr::from(([127, 0, 0, 1], 12001)); + let socket = MockVirtualUdpSocket::new( + local_addr, + vec![ + (b"session".to_vec(), peer_addr), + (b"general".to_vec(), peer_addr), + ], + ); + + assert_eq!( + socket.recv_session_datagram().await.unwrap().payload, + b"session".as_slice() + ); + assert_eq!( + socket.recv_datagram().await.unwrap().payload, + b"general".as_slice() + ); + assert_eq!( + *socket.recv_capacities.lock().unwrap(), + [MAX_UDP_SESSION_DATAGRAM_SIZE + 1, MAX_UDP_DATAGRAM_SIZE] + ); +} + fn easytier_stun_request(change_ip: bool, change_port: bool) -> Vec { let mut request = Message::::new(MessageClass::Request, BINDING, u32_to_tid(7)); if change_ip || change_port { diff --git a/easytier-core/src/socket/udp/virtual_socket.rs b/easytier-core/src/socket/udp/virtual_socket.rs index cb0a9d35..4620f927 100644 --- a/easytier-core/src/socket/udp/virtual_socket.rs +++ b/easytier-core/src/socket/udp/virtual_socket.rs @@ -76,16 +76,36 @@ pub trait VirtualUdpSocket: Send + Sync + 'static { /// override it when their socket API can write directly into owned storage, /// avoiding a second allocation and copy at the Host boundary. async fn recv_datagram(&self) -> std::io::Result { - let mut payload = BytesMut::new(); - payload.resize(MAX_UDP_DATAGRAM_SIZE, 0); - let (len, remote_addr, meta) = self.recv_from_with_meta(&mut payload).await?; - payload.truncate(len); - Ok(UdpSocketDatagram { - payload, - remote_addr, - meta, - }) + recv_portable_datagram(self, MAX_UDP_DATAGRAM_SIZE).await } + + /// Receives one datagram for the UDP session/multiplexer data plane. + /// + /// Portable hosts receive one byte past the session limit so a truncated + /// oversized datagram remains distinguishable from a valid maximum-sized + /// datagram. Native hosts may override this when they can detect truncation + /// without the extra byte. + async fn recv_session_datagram(&self) -> std::io::Result { + recv_portable_datagram(self, MAX_UDP_SESSION_DATAGRAM_SIZE + 1).await + } +} + +async fn recv_portable_datagram( + socket: &S, + capacity: usize, +) -> std::io::Result +where + S: VirtualUdpSocket + ?Sized, +{ + let mut payload = BytesMut::new(); + payload.resize(capacity, 0); + let (len, remote_addr, meta) = socket.recv_from_with_meta(&mut payload).await?; + payload.truncate(len); + Ok(UdpSocketDatagram { + payload, + remote_addr, + meta, + }) } #[async_trait] diff --git a/easytier-core/src/tunnel/encrypt/mod.rs b/easytier-core/src/tunnel/encrypt/mod.rs index 8d7da21c..804ab818 100644 --- a/easytier-core/src/tunnel/encrypt/mod.rs +++ b/easytier-core/src/tunnel/encrypt/mod.rs @@ -17,6 +17,8 @@ pub mod chacha20; mod openssl; #[cfg(all(feature = "ring-crypto", any(not(feature = "openssl-crypto"), test)))] mod ring; +#[cfg(all(target_os = "wasi", feature = "wasi-crypto-offload"))] +mod wasi_host; pub mod xor; @@ -202,7 +204,7 @@ fn preferred_aead_backend(_algorithm: EncryptionAlgorithm) -> Option Arc { - match preferred_aead_backend(EncryptionAlgorithm::AesGcm) { + let fallback = match preferred_aead_backend(EncryptionAlgorithm::AesGcm) { #[cfg(feature = "openssl-crypto")] Some(AeadBackend::OpenSsl) => Arc::new(openssl::OpenSslCipher::new_aes128_gcm(key)), #[cfg(all(not(feature = "openssl-crypto"), feature = "ring-crypto"))] @@ -214,12 +216,13 @@ fn create_aes_128(key: [u8; 16]) -> Arc { ))] Some(AeadBackend::RustCrypto) => Arc::new(aes_gcm::AesGcmCipher::new_128(key)), _ => unavailable_encryptor("aes-gcm"), - } + }; + maybe_offload_aead(EncryptionAlgorithm::AesGcm, &key, fallback) } #[allow(unreachable_patterns)] fn create_aes_256(key: [u8; 32]) -> Arc { - match preferred_aead_backend(EncryptionAlgorithm::Aes256Gcm) { + let fallback = match preferred_aead_backend(EncryptionAlgorithm::Aes256Gcm) { #[cfg(feature = "openssl-crypto")] Some(AeadBackend::OpenSsl) => Arc::new(openssl::OpenSslCipher::new_aes256_gcm(key)), #[cfg(all(not(feature = "openssl-crypto"), feature = "ring-crypto"))] @@ -231,12 +234,13 @@ fn create_aes_256(key: [u8; 32]) -> Arc { ))] Some(AeadBackend::RustCrypto) => Arc::new(aes_gcm::AesGcmCipher::new_256(key)), _ => unavailable_encryptor("aes-256-gcm"), - } + }; + maybe_offload_aead(EncryptionAlgorithm::Aes256Gcm, &key, fallback) } #[allow(unreachable_patterns)] fn create_chacha20(key: [u8; 32]) -> Arc { - match preferred_aead_backend(EncryptionAlgorithm::ChaCha20) { + let fallback = match preferred_aead_backend(EncryptionAlgorithm::ChaCha20) { #[cfg(feature = "openssl-crypto")] Some(AeadBackend::OpenSsl) => Arc::new(openssl::OpenSslCipher::new_chacha20(key)), #[cfg(all(not(feature = "openssl-crypto"), feature = "ring-crypto"))] @@ -248,7 +252,26 @@ fn create_chacha20(key: [u8; 32]) -> Arc { ))] Some(AeadBackend::RustCrypto) => Arc::new(chacha20::ChaCha20Cipher::new(key)), _ => unavailable_encryptor("chacha20"), - } + }; + maybe_offload_aead(EncryptionAlgorithm::ChaCha20, &key, fallback) +} + +#[cfg(all(target_os = "wasi", feature = "wasi-crypto-offload"))] +fn maybe_offload_aead( + algorithm: EncryptionAlgorithm, + key: &[u8], + fallback: Arc, +) -> Arc { + Arc::new(wasi_host::WasiHostAead::new(algorithm, key, fallback)) +} + +#[cfg(not(all(target_os = "wasi", feature = "wasi-crypto-offload")))] +fn maybe_offload_aead( + _algorithm: EncryptionAlgorithm, + _key: &[u8], + fallback: Arc, +) -> Arc { + fallback } pub(crate) fn validate_algorithm(algorithm: &str) -> Result<(), Error> { diff --git a/easytier-core/src/tunnel/encrypt/ring.rs b/easytier-core/src/tunnel/encrypt/ring.rs index f89091d6..519af493 100644 --- a/easytier-core/src/tunnel/encrypt/ring.rs +++ b/easytier-core/src/tunnel/encrypt/ring.rs @@ -172,4 +172,23 @@ mod tests { round_trip(RingCipher::new_aes256_gcm([2; 32])); round_trip(RingCipher::new_chacha20([3; 32])); } + + #[test] + fn aes128_gcm_matches_standard_vector() { + let cipher = RingCipher::new_aes128_gcm([0; 16]); + let mut packet = ZCPacket::new_with_payload(&[0; 16]); + packet.fill_peer_manager_hdr(0, 0, 0); + cipher + .encrypt_with_nonce(&mut packet, Some(&[0; 12])) + .unwrap(); + + assert_eq!( + packet.payload(), + &[ + 0x03, 0x88, 0xda, 0xce, 0x60, 0xb6, 0xa3, 0x92, 0xf3, 0x28, 0xc2, 0xb9, 0x71, 0xb2, + 0xfe, 0x78, 0xab, 0x6e, 0x47, 0xd4, 0x2c, 0xec, 0x13, 0xbd, 0xf5, 0x3a, 0x67, 0xb2, + 0x12, 0x57, 0xbd, 0xdf, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + ] + ); + } } diff --git a/easytier-core/src/tunnel/encrypt/wasi_host.rs b/easytier-core/src/tunnel/encrypt/wasi_host.rs new file mode 100644 index 00000000..92eb5963 --- /dev/null +++ b/easytier-core/src/tunnel/encrypt/wasi_host.rs @@ -0,0 +1,152 @@ +use std::sync::Arc; + +use rand::RngCore as _; +use zerocopy::FromBytes as _; + +use crate::{ + config::EncryptionAlgorithm, + packet::{StandardAeadTail, ZCPacket}, + wasi::{ + abi::{ + AEAD_AES_128_GCM, AEAD_AES_256_GCM, AEAD_CHACHA20_POLY1305, HOST_CRYPTO_AUTH_FAILED, + }, + imports::{crypto_aead_open, crypto_aead_seal}, + }, +}; + +use super::{Encryptor, Error}; + +pub(super) struct WasiHostAead { + algorithm: u32, + key: Box<[u8]>, + fallback: Arc, +} + +impl WasiHostAead { + pub(super) fn new( + algorithm: EncryptionAlgorithm, + key: &[u8], + fallback: Arc, + ) -> Self { + let algorithm = match algorithm { + EncryptionAlgorithm::AesGcm => AEAD_AES_128_GCM, + EncryptionAlgorithm::Aes256Gcm => AEAD_AES_256_GCM, + EncryptionAlgorithm::ChaCha20 => AEAD_CHACHA20_POLY1305, + EncryptionAlgorithm::Xor => unreachable!("XOR is not an AEAD algorithm"), + }; + Self { + algorithm, + key: key.into(), + fallback, + } + } + + fn call( + &self, + open: bool, + nonce: &[u8; StandardAeadTail::NONCE_SIZE], + buffer: &mut [u8], + text_len: usize, + ) -> i32 { + let key_len = u32::try_from(self.key.len()).expect("AEAD keys fit u32"); + let nonce_len = u32::try_from(nonce.len()).expect("AEAD nonces fit u32"); + let text_len = u32::try_from(text_len).expect("packet payloads fit u32"); + let function = if open { + crypto_aead_open + } else { + crypto_aead_seal + }; + unsafe { + function( + self.algorithm, + self.key.as_ptr() as u32, + key_len, + nonce.as_ptr() as u32, + nonce_len, + 0, + 0, + buffer.as_mut_ptr() as u32, + text_len, + ) + } + } +} + +impl Encryptor for WasiHostAead { + fn decrypt(&self, packet: &mut ZCPacket) -> Result<(), Error> { + let header = packet.peer_manager_header().unwrap(); + if !header.is_encrypted() { + return Ok(()); + } + + let payload_len = packet.payload().len(); + if payload_len < StandardAeadTail::SIZE { + return Err(Error::PacketTooShort(payload_len)); + } + + let text_len = payload_len - StandardAeadTail::SIZE; + let tail = StandardAeadTail::ref_from_suffix(packet.payload()) + .unwrap() + .clone(); + let status = self.call( + true, + &tail.nonce, + &mut packet.mut_payload()[..text_len + StandardAeadTail::TAG_SIZE], + text_len, + ); + if status == HOST_CRYPTO_AUTH_FAILED { + return Err(Error::DecryptionFailed); + } + if status != 0 { + return self.fallback.decrypt(packet); + } + + packet + .mut_peer_manager_header() + .unwrap() + .set_encrypted(false); + let old_len = packet.buf_len(); + packet + .mut_inner() + .truncate(old_len - StandardAeadTail::SIZE); + Ok(()) + } + + fn encrypt(&self, packet: &mut ZCPacket) -> Result<(), Error> { + self.encrypt_with_nonce(packet, None) + } + + fn encrypt_with_nonce(&self, packet: &mut ZCPacket, nonce: Option<&[u8]>) -> Result<(), Error> { + let header = packet.peer_manager_header().unwrap(); + if header.is_encrypted() { + tracing::warn!(?packet, "packet is already encrypted"); + return Ok(()); + } + + let mut nonce_bytes = [0; StandardAeadTail::NONCE_SIZE]; + match nonce { + Some(nonce) => { + nonce_bytes = nonce.try_into().map_err(|_| Error::EncryptionFailed)?; + } + None => rand::thread_rng().fill_bytes(&mut nonce_bytes), + } + + let text_len = packet.payload().len(); + let old_len = packet.buf_len(); + packet + .mut_inner() + .extend_from_slice(&[0; StandardAeadTail::TAG_SIZE]); + let status = self.call(false, &nonce_bytes, packet.mut_payload(), text_len); + if status != 0 { + packet.mut_inner().truncate(old_len); + return self.fallback.encrypt_with_nonce(packet, Some(&nonce_bytes)); + } + + packet.mut_inner().extend_from_slice(&nonce_bytes); + packet + .mut_peer_manager_header() + .unwrap() + .set_encrypted(true); + Ok(()) + } +} diff --git a/easytier-core/src/wasi/abi.rs b/easytier-core/src/wasi/abi.rs index 4a1031a0..fe7ea972 100644 --- a/easytier-core/src/wasi/abi.rs +++ b/easytier-core/src/wasi/abi.rs @@ -15,11 +15,22 @@ /// WebAssembly import module a WASI runtime must implement. pub const HOST_IMPORT_MODULE: &str = "easytier_host"; +/// AEAD algorithm identifiers accepted by the optional crypto imports. +pub const AEAD_AES_128_GCM: u32 = 1; +pub const AEAD_AES_256_GCM: u32 = 2; +pub const AEAD_CHACHA20_POLY1305: u32 = 3; + +/// The Host could not authenticate an AEAD record. +/// +/// Unlike other non-zero crypto statuses, this must not fall back to the +/// built-in implementation because an in-place open may have changed bytes. +pub const HOST_CRYPTO_AUTH_FAILED: i32 = -10; + /// Version of the JSON document accepted by `easytier_instance_create`. pub const CORE_INSTANCE_CONFIG_VERSION: u32 = 14; /// Version of the public data-plane guest export contract. -pub const DATA_PLANE_ABI_VERSION: u32 = 2; +pub const DATA_PLANE_ABI_VERSION: u32 = 3; /// The guest exposes an instance-scoped data-plane operation broker. pub const DATA_PLANE_CAPABILITY: u64 = 1 << 0; @@ -27,6 +38,10 @@ pub const DATA_PLANE_CAPABILITY: u64 = 1 << 0; pub const DATA_PLANE_TCP_CAPABILITY: u64 = 1 << 1; /// The guest data plane supports UDP sockets. pub const DATA_PLANE_UDP_CAPABILITY: u64 = 1 << 2; +/// Update the read deadline in `easytier_data_plane_resource_deadline_set`. +pub const DATA_PLANE_DEADLINE_READ: u32 = 1 << 0; +/// Update the write deadline in `easytier_data_plane_resource_deadline_set`. +pub const DATA_PLANE_DEADLINE_WRITE: u32 = 1 << 1; /// Guest exports a WASI runtime calls to manage a core instance. /// @@ -68,6 +83,7 @@ pub const DATA_PLANE_GUEST_EXPORTS: &[&str] = &[ "easytier_data_plane_udp_bind_submit", "easytier_data_plane_udp_receive_submit", "easytier_data_plane_udp_send_submit", + "easytier_data_plane_resource_deadline_set", // Completion, result, and resource lifecycle. "easytier_data_plane_completion_drain", "easytier_data_plane_result_size", diff --git a/easytier-core/src/wasi/adapter/event.rs b/easytier-core/src/wasi/adapter/event.rs new file mode 100644 index 00000000..8b5c1398 --- /dev/null +++ b/easytier-core/src/wasi/adapter/event.rs @@ -0,0 +1,60 @@ +use crate::{ + events::{CoreEvent, CoreEventSink}, + wasi::imports::emit_event, +}; + +#[derive(Debug)] +pub struct WasiHostEventSink { + handle: u64, +} + +impl WasiHostEventSink { + pub fn new(handle: u64) -> Self { + Self { handle } + } +} + +impl CoreEventSink for WasiHostEventSink { + fn emit(&self, event: CoreEvent) { + let kind = event_kind(&event); + let message = format!("{event:?}"); + let _ = unsafe { + emit_event( + self.handle, + kind.as_ptr() as u32, + kind.len() as u32, + message.as_ptr() as u32, + message.len() as u32, + ) + }; + } +} + +fn event_kind(event: &CoreEvent) -> &'static str { + match event { + CoreEvent::PeerAdded(_) => "peer_added", + CoreEvent::PeerRemoved(_) => "peer_removed", + CoreEvent::PeerConnAdded(_) => "peer_connection_added", + CoreEvent::PeerConnRemoved(_) => "peer_connection_removed", + CoreEvent::CredentialChanged => "credential_changed", + CoreEvent::ManualConnecting { .. } => "connecting", + CoreEvent::ManualConnectError { .. } => "connect_error", + CoreEvent::ListenerPlanFailed { .. } => "listener_plan_failed", + CoreEvent::ListenerAdded { .. } => "listener_added", + CoreEvent::ListenerRemoved { .. } => "listener_removed", + CoreEvent::ListenerAddFailed { .. } => "listener_add_failed", + CoreEvent::ListenerAcceptFailed { .. } => "listener_accept_failed", + CoreEvent::ListenerSocketAccepted { .. } => "listener_socket_accepted", + CoreEvent::ListenerAcceptedSocketHandleFailed { .. } => "listener_socket_handle_failed", + CoreEvent::TunnelAccepted { .. } => "tunnel_accepted", + CoreEvent::TunnelAdmissionFailed { .. } => "tunnel_admission_failed", + CoreEvent::UdpPortMappingEstablished { .. } => "udp_port_mapping_established", + CoreEvent::ProxyCidrsUpdated { .. } => "proxy_cidrs_updated", + CoreEvent::PublicIpv6LeaseChanged { .. } => "public_ipv6_lease_changed", + CoreEvent::PublicIpv6RoutesChanged { .. } => "public_ipv6_routes_changed", + CoreEvent::VpnPortalStarted(_) => "vpn_portal_started", + CoreEvent::VpnPortalClientConnected { .. } => "vpn_portal_client_connected", + CoreEvent::VpnPortalClientDisconnected { .. } => "vpn_portal_client_disconnected", + CoreEvent::GatewayPortForwardAdded(_) => "gateway_port_forward_added", + } +} diff --git a/easytier-core/src/wasi/adapter/mod.rs b/easytier-core/src/wasi/adapter/mod.rs index 70008e19..63e2bfdf 100644 --- a/easytier-core/src/wasi/adapter/mod.rs +++ b/easytier-core/src/wasi/adapter/mod.rs @@ -2,5 +2,6 @@ pub mod dns; pub mod environment; +pub mod event; pub mod packet; pub mod socket; diff --git a/easytier-core/src/wasi/imports.rs b/easytier-core/src/wasi/imports.rs index d5fb05bb..c4ce7a00 100644 --- a/easytier-core/src/wasi/imports.rs +++ b/easytier-core/src/wasi/imports.rs @@ -9,6 +9,57 @@ pub(crate) const HOST_WOULD_BLOCK: i32 = -5; #[link(wasm_import_module = "easytier_host")] unsafe extern "C" { + /// Emits one best-effort instance event after the host copies both strings. + /// + /// The host must not block the guest. A non-zero status drops this event + /// without affecting core execution. + pub(crate) fn emit_event( + handle: u64, + kind: u32, + kind_len: u32, + message: u32, + message_len: u32, + ) -> i32; + + /// Encrypts `text_len` bytes in place and writes the AEAD tag immediately + /// after them. + /// + /// The guest reserves the algorithm's tag size in linear memory before + /// calling. Every non-zero result except + /// [`crate::wasi::abi::HOST_CRYPTO_AUTH_FAILED`] must leave the buffer + /// unchanged so the guest can use its built-in implementation. + #[cfg(feature = "wasi-crypto-offload")] + pub(crate) fn crypto_aead_seal( + algorithm: u32, + key: u32, + key_len: u32, + nonce: u32, + nonce_len: u32, + aad: u32, + aad_len: u32, + buffer: u32, + text_len: u32, + ) -> i32; + + /// Authenticates and decrypts `text_len` bytes in place using the AEAD tag + /// immediately after them. + /// + /// Authentication failure may change the buffer and must return + /// [`crate::wasi::abi::HOST_CRYPTO_AUTH_FAILED`]. Every other non-zero + /// result must leave the buffer unchanged so the guest can fall back. + #[cfg(feature = "wasi-crypto-offload")] + pub(crate) fn crypto_aead_open( + algorithm: u32, + key: u32, + key_len: u32, + nonce: u32, + nonce_len: u32, + aad: u32, + aad_len: u32, + buffer: u32, + text_len: u32, + ) -> i32; + /// Starts one TCP read into a host-owned pending operation. /// /// The host records at most `capacity` bytes for `operation` and must not diff --git a/easytier-core/src/wasi/runtime.rs b/easytier-core/src/wasi/runtime.rs index 6bba04fc..9a1badcc 100644 --- a/easytier-core/src/wasi/runtime.rs +++ b/easytier-core/src/wasi/runtime.rs @@ -31,20 +31,18 @@ pub(super) fn new_wasi_core_runtime( process_runtime: std::sync::Arc, environment_snapshot: HostConnectorEnvironmentSnapshot, packet_sink: crate::host::packet::HostPacketSinkHandle, + event_sink: u64, ) -> anyhow::Result { use std::sync::Arc; - use crate::host::{ - dns::HostDnsResolver, - packet::{HostPacket, HostPacketSink}, - socket::HostSocketRuntime, - }; + use crate::host::{dns::HostDnsResolver, packet::HostPacketSink, socket::HostSocketRuntime}; use crate::{ connectivity::connector_host::new_connector_host, instance::{CoreHostAdapters, CoreInstance}, wasi::adapter::{ dns::WasiHostDnsIo, environment::WasiHostConnectorEnvironmentIo, - packet::WasiHostPacketIo, socket::backend::WasiHostSocketBackend, + event::WasiHostEventSink, packet::WasiHostPacketIo, + socket::backend::WasiHostSocketBackend, }, }; @@ -64,7 +62,8 @@ pub(super) fn new_wasi_core_runtime( Arc::new(WasiHostPacketIo), packet_sink, )); - let adapters = CoreHostAdapters::new(host, dns, packet_sink, process_runtime); + let mut adapters = CoreHostAdapters::new(host, dns, packet_sink, process_runtime); + adapters.events = Arc::new(WasiHostEventSink::new(event_sink)); let core = CoreInstance::from_toml(config, adapters)?; Ok(WasiCoreRuntime { @@ -85,7 +84,7 @@ mod abi { use crate::{ config::toml::{ConfigLoader as _, TomlConfig}, foundation::time::{clear_domain, enter_domain, next_deadline_millis}, - host::packet::HostPacketSinkHandle, + host::packet::{HostPacket, HostPacketSinkHandle}, instance::{ CoreInstanceState, manager::{InstanceFactory, ManagedInstance}, @@ -164,6 +163,7 @@ mod abi { domain: u64, environment: crate::connectivity::connector_host::HostConnectorEnvironmentSnapshot, packet_sink: HostPacketSinkHandle, + event_sink: u64, } struct WasiInstance { @@ -220,6 +220,7 @@ mod abi { self.process_runtime.clone(), context.environment, context.packet_sink, + context.event_sink, )? }; @@ -348,8 +349,11 @@ mod abi { fn drive(&self) -> anyhow::Result<()> { let _domain = enter_domain(self.domain); + let advance_timers = next_deadline_millis(self.domain) == Some(0); let mut execution = self.execution.lock().unwrap(); - execution.drive_again = execution.runtime_driver.drive(&execution.runtime) + execution.drive_again = execution + .runtime_driver + .drive(&execution.runtime, advance_timers) == RuntimeDriveOutcome::BudgetExhausted; if execution @@ -561,13 +565,15 @@ mod abi { #[unsafe(no_mangle)] /// Creates one core instance from a versioned envelope containing TOML. /// - /// `config_pointer` must name a live ABI buffer and `packet_sink_handle` - /// identifies the host sink used for locally delivered raw IP packets. + /// `config_pointer` must name a live ABI buffer. `packet_sink_handle` and + /// `event_sink_handle` identify the host sinks used for locally delivered + /// raw IP packets and best-effort instance events. /// Returns zero on failure; retrieve the reason through the error exports. pub extern "C" fn easytier_instance_create( config_pointer: u32, config_length: u32, packet_sink_handle: u64, + event_sink_handle: u64, ) -> u64 { let encoded = match read_guest_buffer(config_pointer, config_length, MAX_CREATE_CONFIG_LEN) { @@ -601,6 +607,7 @@ mod abi { domain: handle, environment: create_config.environment, packet_sink: HostPacketSinkHandle(packet_sink_handle), + event_sink: event_sink_handle, }, ); let instance = match instance { diff --git a/easytier-core/src/wasi/runtime/abi/data_plane.rs b/easytier-core/src/wasi/runtime/abi/data_plane.rs index 02fde008..a4617977 100644 --- a/easytier-core/src/wasi/runtime/abi/data_plane.rs +++ b/easytier-core/src/wasi/runtime/abi/data_plane.rs @@ -9,8 +9,8 @@ use crate::{ }, wasi::{ abi::{ - DATA_PLANE_ABI_VERSION, DATA_PLANE_CAPABILITY, DATA_PLANE_TCP_CAPABILITY, - DATA_PLANE_UDP_CAPABILITY, + DATA_PLANE_ABI_VERSION, DATA_PLANE_CAPABILITY, DATA_PLANE_DEADLINE_READ, + DATA_PLANE_DEADLINE_WRITE, DATA_PLANE_TCP_CAPABILITY, DATA_PLANE_UDP_CAPABILITY, }, wire::{ data_plane::{ @@ -42,12 +42,10 @@ impl WasiInstance { self.core.core().data_plane_session() } - fn submit_data_plane( + fn submit_data_plane( &self, - submit: impl FnOnce( - &std::sync::Arc, - ) -> Result, - ) -> Result { + submit: impl FnOnce(&std::sync::Arc) -> Result, + ) -> Result { let execution = self.execution.lock().unwrap(); let _domain = crate::foundation::time::enter_domain(self.domain); let _runtime = execution.runtime.enter(); @@ -316,7 +314,6 @@ pub extern "C" fn easytier_data_plane_tcp_read_submit( handle: u64, stream: u64, max_len: u32, - timeout_ms: u64, output_operation: u32, ) -> i32 { let stream = match resource_id(stream) { @@ -327,9 +324,7 @@ pub extern "C" fn easytier_data_plane_tcp_read_submit( } }; submit_operation(handle, output_operation, |instance| { - instance.submit_data_plane(|session| { - session.submit_tcp_read(stream, max_len as usize, timeout(timeout_ms)) - }) + instance.submit_data_plane(|session| session.submit_tcp_read(stream, max_len as usize)) }) } @@ -339,7 +334,6 @@ pub extern "C" fn easytier_data_plane_tcp_write_submit( stream: u64, data_pointer: u32, data_length: u32, - timeout_ms: u64, output_operation: u32, ) -> i32 { let stream = match resource_id(stream) { @@ -357,9 +351,7 @@ pub extern "C" fn easytier_data_plane_tcp_write_submit( } }; submit_operation(handle, output_operation, |instance| { - instance.submit_data_plane(|session| { - session.submit_tcp_write(stream, data, timeout(timeout_ms)) - }) + instance.submit_data_plane(|session| session.submit_tcp_write(stream, data)) }) } @@ -388,7 +380,6 @@ pub extern "C" fn easytier_data_plane_udp_receive_submit( handle: u64, socket: u64, max_len: u32, - timeout_ms: u64, output_operation: u32, ) -> i32 { let socket = match resource_id(socket) { @@ -399,9 +390,7 @@ pub extern "C" fn easytier_data_plane_udp_receive_submit( } }; submit_operation(handle, output_operation, |instance| { - instance.submit_data_plane(|session| { - session.submit_udp_receive(socket, max_len as usize, timeout(timeout_ms)) - }) + instance.submit_data_plane(|session| session.submit_udp_receive(socket, max_len as usize)) }) } @@ -412,7 +401,6 @@ pub extern "C" fn easytier_data_plane_udp_send_submit( peer_address: u32, data_pointer: u32, data_length: u32, - timeout_ms: u64, output_operation: u32, ) -> i32 { let socket = match resource_id(socket) { @@ -437,9 +425,36 @@ pub extern "C" fn easytier_data_plane_udp_send_submit( } }; submit_operation(handle, output_operation, |instance| { + instance.submit_data_plane(|session| session.submit_udp_send(socket, peer_address, data)) + }) +} + +#[unsafe(no_mangle)] +pub extern "C" fn easytier_data_plane_resource_deadline_set( + handle: u64, + resource: u64, + direction: u32, + timeout_ms: u64, +) -> i32 { + let resource = match resource_id(resource) { + Ok(resource) => resource, + Err(error) => { + set_instance_error(handle, error.message()); + return error_status(error.kind()); + } + }; + let read = direction & DATA_PLANE_DEADLINE_READ != 0; + let write = direction & DATA_PLANE_DEADLINE_WRITE != 0; + if direction == 0 || direction & !(DATA_PLANE_DEADLINE_READ | DATA_PLANE_DEADLINE_WRITE) != 0 { + let error = invalid_input(format!("invalid deadline direction {direction}")); + set_instance_error(handle, error.message()); + return error_status(error.kind()); + } + data_plane_call(handle, |instance| { instance.submit_data_plane(|session| { - session.submit_udp_send(socket, peer_address, data, timeout(timeout_ms)) - }) + session.set_resource_deadline(resource, read, write, timeout(timeout_ms)) + })?; + Ok(0) }) } diff --git a/easytier-core/src/wasi/runtime_driver.rs b/easytier-core/src/wasi/runtime_driver.rs index 1c821b7a..feb9b22f 100644 --- a/easytier-core/src/wasi/runtime_driver.rs +++ b/easytier-core/src/wasi/runtime_driver.rs @@ -41,12 +41,14 @@ impl RuntimeDriver { } } - pub(super) fn drive(&self, runtime: &Runtime) -> RuntimeDriveOutcome { - // First give the timer driver a non-blocking turn. The quiescence hook - // stays disabled here so an expired timer can wake its task. - runtime.block_on(async { - tokio::time::sleep(Duration::ZERO).await; - }); + pub(super) fn drive(&self, runtime: &Runtime, advance_timers: bool) -> RuntimeDriveOutcome { + if advance_timers { + // Give the timer driver a turn before enabling the quiescence hook + // so an expired timer can wake its task. + runtime.block_on(async { + tokio::time::sleep(Duration::ZERO).await; + }); + } let _active = RuntimeDriverGuard::activate(self.state.as_ref()); runtime.block_on(async { @@ -101,9 +103,10 @@ impl Drop for RuntimeDriverGuard<'_> { mod tests { use std::{future::poll_fn, sync::Arc, task::Poll}; - use tokio::{runtime::Builder, sync::Notify}; + use tokio::{runtime::Builder, sync::Notify, time::Duration}; use super::{RuntimeDriveOutcome, RuntimeDriver}; + use crate::wasi::time::{enter_domain, next_deadline_millis}; fn runtime(driver: &RuntimeDriver) -> tokio::runtime::Runtime { let park_driver = driver.clone(); @@ -124,10 +127,13 @@ mod tests { Poll::<()>::Pending })); - assert_eq!(driver.drive(&runtime), RuntimeDriveOutcome::BudgetExhausted); + assert_eq!( + driver.drive(&runtime, false), + RuntimeDriveOutcome::BudgetExhausted + ); task.abort(); - while driver.drive(&runtime) == RuntimeDriveOutcome::BudgetExhausted {} + while driver.drive(&runtime, false) == RuntimeDriveOutcome::BudgetExhausted {} assert!(task.is_finished()); } @@ -141,11 +147,56 @@ mod tests { task_notify.notified().await; }); - assert_eq!(driver.drive(&runtime), RuntimeDriveOutcome::Quiescent); + assert_eq!( + driver.drive(&runtime, false), + RuntimeDriveOutcome::Quiescent + ); assert!(!task.is_finished()); notify.notify_one(); - while driver.drive(&runtime) == RuntimeDriveOutcome::BudgetExhausted {} + while driver.drive(&runtime, false) == RuntimeDriveOutcome::BudgetExhausted {} assert!(task.is_finished()); } + + #[test] + fn advances_expired_timers_only_when_requested() { + let driver = RuntimeDriver::default(); + let runtime = runtime(&driver); + let task = runtime.spawn(async { + tokio::time::sleep(Duration::ZERO).await; + }); + + assert_eq!( + driver.drive(&runtime, false), + RuntimeDriveOutcome::Quiescent + ); + assert!(!task.is_finished()); + + assert_eq!(driver.drive(&runtime, true), RuntimeDriveOutcome::Quiescent); + assert!(task.is_finished()); + } + + #[test] + fn tracked_expired_timer_requests_advancement() { + let _domain = enter_domain(7); + let driver = RuntimeDriver::default(); + let runtime = runtime(&driver); + let task = runtime.spawn(async { + crate::foundation::time::sleep(Duration::ZERO).await; + }); + + assert_eq!( + driver.drive(&runtime, false), + RuntimeDriveOutcome::Quiescent + ); + assert_eq!(next_deadline_millis(7), Some(0)); + + let advance_timers = next_deadline_millis(7) == Some(0); + assert_eq!( + driver.drive(&runtime, advance_timers), + RuntimeDriveOutcome::Quiescent + ); + assert!(task.is_finished()); + assert_eq!(next_deadline_millis(7), None); + } } diff --git a/easytier-core/src/wasi/time.rs b/easytier-core/src/wasi/time.rs index 7cf1d899..67d1ead4 100644 --- a/easytier-core/src/wasi/time.rs +++ b/easytier-core/src/wasi/time.rs @@ -230,7 +230,7 @@ mod tracked { } } -pub use tracked::{Duration, Instant, Interval, error, interval, sleep, timeout}; +pub use tracked::{Duration, Instant, Interval, error, interval, sleep, sleep_until, timeout}; pub(crate) use tracked::{clear_domain, enter_domain, next_deadline_millis}; diff --git a/easytier/src/socket/udp.rs b/easytier/src/socket/udp.rs index 87a45545..9e2e2cbb 100644 --- a/easytier/src/socket/udp.rs +++ b/easytier/src/socket/udp.rs @@ -12,8 +12,9 @@ use easytier_core::socket::{ use easytier_core::socket::{ SocketContext, udp::{ - MAX_UDP_SESSION_DATAGRAM_SIZE, UdpBindOptions, UdpSocketDatagram, UdpSocketPurpose, - UdpSocketRecvMeta, UdpSocketSendMeta, VirtualUdpSocket, VirtualUdpSocketFactory, + MAX_UDP_DATAGRAM_SIZE, MAX_UDP_SESSION_DATAGRAM_SIZE, UdpBindOptions, UdpSocketDatagram, + UdpSocketPurpose, UdpSocketRecvMeta, UdpSocketSendMeta, VirtualUdpSocket, + VirtualUdpSocketFactory, }, }; use tokio::net::UdpSocket; @@ -111,6 +112,17 @@ impl VirtualUdpSocket for RuntimeUdpSocket { #[cfg(unix)] async fn recv_datagram(&self) -> std::io::Result { + let (payload, remote_addr, dst_ip) = + udp_src::recv_datagram_with_dst_ip(&self.socket, MAX_UDP_DATAGRAM_SIZE).await?; + Ok(UdpSocketDatagram { + payload, + remote_addr, + meta: UdpSocketRecvMeta { dst_ip }, + }) + } + + #[cfg(unix)] + async fn recv_session_datagram(&self) -> std::io::Result { let (payload, remote_addr, dst_ip) = udp_src::recv_datagram_with_dst_ip(&self.socket, MAX_UDP_SESSION_DATAGRAM_SIZE).await?; Ok(UdpSocketDatagram { @@ -265,7 +277,7 @@ mod tests { let datagram = tokio::time::timeout( std::time::Duration::from_secs(1), - runtime_socket.recv_datagram(), + runtime_socket.recv_session_datagram(), ) .await .unwrap() diff --git a/script/build-wasi-core.sh b/script/build-wasi-core.sh new file mode 100755 index 00000000..cb6c37cc --- /dev/null +++ b/script/build-wasi-core.sh @@ -0,0 +1,134 @@ +#!/usr/bin/env bash + +set -Eeuo pipefail + +readonly binaryen_version=131 +readonly binaryen_release="version_${binaryen_version}" +readonly script_dir=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd) +readonly repository_root=$(cd -- "${script_dir}/.." && pwd) +cd "$repository_root" +readonly target_dir="${CARGO_TARGET_DIR:-target}" +readonly raw_artifact="${target_dir}/wasm32-wasip1/release/easytier_core.wasm" +readonly artifact="${target_dir}/wasm32-wasip1/release/easytier_core_go_host.wasm" +readonly core_features="proxy-smoltcp-stack,ring-crypto,wasi-crypto-offload" + +sha256_file() { + if command -v sha256sum >/dev/null; then + sha256sum "$1" | awk '{print $1}' + elif command -v shasum >/dev/null; then + shasum -a 256 "$1" | awk '{print $1}' + else + echo "sha256sum or shasum is required" >&2 + return 1 + fi +} + +resolve_binaryen_asset() { + local operating_system machine + operating_system=$(uname -s) + machine=$(uname -m) + + case "$machine" in + x86_64 | amd64) machine=x86_64 ;; + aarch64 | arm64) machine=arm64 ;; + *) + echo "unsupported Binaryen host architecture: ${machine}" >&2 + return 1 + ;; + esac + + case "$operating_system" in + Linux) + if [[ "$machine" == arm64 ]]; then + binaryen_platform=aarch64-linux + binaryen_sha256=ba991f677edd9a21d2bc96c0144bc8ac5b112d4d98a3eb266e075e22e557df2a + else + binaryen_platform=x86_64-linux + binaryen_sha256=b5bf1f0eaf17c63ee588ff7a5954dc8f6ce2c26989051c66f24dfe9ece3e46db + fi + ;; + Darwin) + binaryen_platform="${machine}-macos" + if [[ "$machine" == arm64 ]]; then + binaryen_sha256=e441b48dc22163d209b4f05e44dc7210909b01237642b6c9ae48fd710a3ef83e + else + binaryen_sha256=d209fadd8a894bdaf3bd3612a23c32a0af184d2f4a979b8c789e6e4f6a4de883 + fi + ;; + MINGW* | MSYS* | CYGWIN*) + binaryen_platform="${machine}-windows" + binaryen_executable=.exe + if [[ "$machine" == arm64 ]]; then + binaryen_sha256=e3eaed3d43bcbba867895e55f5e3e9fcfebf776bc4ed6ee59cae071f083cedb9 + else + binaryen_sha256=2f4edac1703a2f695254d6ff52ede03481e67db1f094915763d863158c17d9bc + fi + ;; + *) + echo "unsupported Binaryen host operating system: ${operating_system}" >&2 + return 1 + ;; + esac +} + +download_wasm_opt() { + local archive archive_name archive_url cache_dir digest temporary + resolve_binaryen_asset + archive_name="binaryen-${binaryen_release}-${binaryen_platform}.tar.gz" + archive_url="https://github.com/WebAssembly/binaryen/releases/download/${binaryen_release}/${archive_name}" + cache_dir="${target_dir}/binaryen/${binaryen_release}-${binaryen_platform}" + archive="${cache_dir}/${archive_name}" + wasm_opt="${cache_dir}/binaryen-${binaryen_release}/bin/wasm-opt${binaryen_executable:-}" + if [[ -x "$wasm_opt" ]]; then + return + fi + + mkdir -p "$cache_dir" + if [[ -f "$archive" ]]; then + digest=$(sha256_file "$archive") + fi + if [[ ${digest:-} != "$binaryen_sha256" ]]; then + command -v curl >/dev/null || { + echo "curl is required to download Binaryen" >&2 + return 1 + } + temporary="${archive}.$$" + trap 'rm -f "${temporary:-}" "${optimized:-}"' EXIT + curl --fail --location --output "$temporary" "$archive_url" + digest=$(sha256_file "$temporary") + if [[ "$digest" != "$binaryen_sha256" ]]; then + echo "Binaryen archive SHA-256 mismatch: ${digest}" >&2 + return 1 + fi + mv "$temporary" "$archive" + fi + tar -xzf "$archive" -C "$cache_dir" + if [[ ! -x "$wasm_opt" ]]; then + echo "wasm-opt is missing from Binaryen archive" >&2 + return 1 + fi +} + +wasm_opt=${WASM_OPT:-} +if [[ -z "$wasm_opt" ]]; then + download_wasm_opt +fi +version=$("$wasm_opt" --version) +if [[ "$version" != *"version ${binaryen_version} "* ]]; then + echo "wasm-opt ${binaryen_version} is required, got: ${version}" >&2 + exit 1 +fi + +cargo build --release --target wasm32-wasip1 -p easytier-core \ + --features "$core_features" + +optimized="${artifact}.wasm-opt.$$" +trap 'rm -f "${temporary:-}" "${optimized:-}"' EXIT +"$wasm_opt" -O4 \ + --enable-bulk-memory \ + --enable-bulk-memory-opt \ + --enable-nontrapping-float-to-int \ + "$raw_artifact" \ + -o "$optimized" +mv "$optimized" "$artifact" +echo "built ${artifact} with ${version}"