diff --git a/docs/core-architecture.md b/docs/core-architecture.md index c89c9136..bcb1a780 100644 --- a/docs/core-architecture.md +++ b/docs/core-architecture.md @@ -349,6 +349,14 @@ A core instance: `CoreProcessRuntime`, plus optional protocol and platform capabilities. The bundle carries capabilities, not preconstructed portable managers. +The standard native factory owns one continuously driven Tokio executor per +instance. `multi_thread` selects a multi-thread or current-thread runtime; +`multi_thread_count` sets the worker count with the historical minimum of two. +Construction, start/stop, and FFI data-plane tasks use that executor, independent +of the process management runtime. Dropping its Host owner shuts down the +executor without blocking a worker on its own termination. Compact and WASI +Hosts retain their externally supplied execution model. + Each Module owns partial-start cleanup for its internal resources. `CoreInstance` has one outer cancellation and recovery path for the complete serial startup. `Running` therefore means the Host runtime and every enabled diff --git a/easytier-core/src/gateway/port_forward.rs b/easytier-core/src/gateway/port_forward.rs index eda8df7c..a3a0a88f 100644 --- a/easytier-core/src/gateway/port_forward.rs +++ b/easytier-core/src/gateway/port_forward.rs @@ -100,6 +100,7 @@ where { operation: Mutex<()>, started: AtomicBool, + runtime_handle: Option, runtime_config: CoreRuntimeConfigStore, data_plane: Arc>, host: Arc, @@ -124,10 +125,12 @@ where host: Arc, socket_context: SocketContext, events: Arc, + runtime_handle: Option, ) -> Arc { Arc::new(Self { operation: Mutex::new(()), started: AtomicBool::new(false), + runtime_handle, runtime_config, data_plane, host, @@ -168,7 +171,22 @@ where Ok(()) } - pub(crate) async fn reload(&self, cfgs: &[PortForwardConfig]) -> anyhow::Result<()> { + pub(crate) async fn reload(self: &Arc, cfgs: &[PortForwardConfig]) -> anyhow::Result<()> { + if let Some(runtime) = &self.runtime_handle { + let adapter = self.clone(); + let cfgs = cfgs.to_vec(); + // Binding and spawning must both happen on the instance executor, + // including reloads reached through management config patches. + return AbortOnDropHandle::new( + runtime.spawn(async move { adapter.reload_on_runtime(&cfgs).await }), + ) + .await + .map_err(|error| anyhow::anyhow!("port-forward reload task failed: {error}"))?; + } + self.reload_on_runtime(cfgs).await + } + + async fn reload_on_runtime(&self, cfgs: &[PortForwardConfig]) -> anyhow::Result<()> { let _operation = self.operation.lock().await; if !self.started.load(Ordering::Acquire) { return Ok(()); diff --git a/easytier-core/src/instance/lifecycle.rs b/easytier-core/src/instance/lifecycle.rs index 18139b39..ca1617e0 100644 --- a/easytier-core/src/instance/lifecycle.rs +++ b/easytier-core/src/instance/lifecycle.rs @@ -216,8 +216,25 @@ where self.packet_egress.stop().await; } + /// Returns the optional Host-owned executor for this instance. + pub fn runtime_handle(&self) -> Option { + self.instance_runtime.runtime_handle() + } + /// Starts the complete instance through one serial composition path. pub async fn start(self: &Arc) -> anyhow::Result<()> { + if let Some(runtime) = self.runtime_handle() { + let instance = self.clone(); + return tokio_util::task::AbortOnDropHandle::new( + runtime.spawn(async move { instance.start_on_runtime().await }), + ) + .await + .map_err(|error| anyhow::anyhow!("instance start task failed: {error}"))?; + } + self.start_on_runtime().await + } + + async fn start_on_runtime(self: &Arc) -> anyhow::Result<()> { let _operation = self.operation.lock().await; let state = self.state(); if state != CoreInstanceState::Created { @@ -260,6 +277,21 @@ where } pub async fn stop(self: &Arc) { + self.cancel.cancel(); + if let Some(runtime) = self.runtime_handle() { + let instance = self.clone(); + // Cleanup must survive cancellation even before its first poll. + // Dropping a plain JoinHandle detaches rather than aborts the task. + runtime + .spawn(async move { instance.stop_on_runtime().await }) + .await + .expect("instance stop task failed"); + } else { + self.stop_on_runtime().await; + } + } + + async fn stop_on_runtime(self: &Arc) { self.cancel.cancel(); let mut recovery = self.recovery_guard(); let _operation = self.operation.lock().await; diff --git a/easytier-core/src/instance/manager.rs b/easytier-core/src/instance/manager.rs index 3351d607..64b47900 100644 --- a/easytier-core/src/instance/manager.rs +++ b/easytier-core/src/instance/manager.rs @@ -479,9 +479,9 @@ where } pub fn data_plane_runtime_handle(&self, instance_id: &Uuid) -> Option { - self.instance(*instance_id)?; - self.runtime_handle - .clone() + self.instance(*instance_id)? + .runtime_handle() + .or_else(|| self.runtime_handle.clone()) .or_else(|| tokio::runtime::Handle::try_current().ok()) } diff --git a/easytier-core/src/instance/mod.rs b/easytier-core/src/instance/mod.rs index 00b299ad..3ae2c028 100644 --- a/easytier-core/src/instance/mod.rs +++ b/easytier-core/src/instance/mod.rs @@ -229,6 +229,12 @@ fn proxy_cidr_snapshot(config: &CoreInstanceRuntimeConfig) -> ProxyCidrSnapshot /// lifetime, such as a native packet interface. #[async_trait::async_trait] pub trait InstanceRuntimeHost: std::any::Any + Send + Sync + 'static { + /// Optional Host-owned executor used for construction, lifecycle, and data-plane tasks. + /// Hosts without a dedicated executor continue to use the caller's runtime. + fn runtime_handle(&self) -> Option { + None + } + async fn prepare( &self, packet_plane: Arc, @@ -783,6 +789,7 @@ where host.clone(), direct_options.tcp_bind.context.clone(), events.clone(), + instance_runtime.runtime_handle(), ); #[cfg(feature = "tcp-hole-punch")] let tcp_hole_punch = stun diff --git a/easytier-core/src/instance/tests.rs b/easytier-core/src/instance/tests.rs index 773993ad..ff00a6ed 100644 --- a/easytier-core/src/instance/tests.rs +++ b/easytier-core/src/instance/tests.rs @@ -1513,6 +1513,162 @@ virtual_ip = "10.82.0.2/24" assert_eq!(runtime_host.shutdown_calls.load(Ordering::Relaxed), 1); } + #[test] + fn host_executor_routes_lifecycle_and_cancelled_start() { + struct ExecutorRuntimeHost { + handle: tokio::runtime::Handle, + block_prepare: bool, + prepare_started: Notify, + shutdown_calls: AtomicUsize, + } + + #[async_trait] + impl InstanceRuntimeHost for ExecutorRuntimeHost { + fn runtime_handle(&self) -> Option { + Some(self.handle.clone()) + } + + async fn prepare( + &self, + _packet_plane: Arc, + ) -> anyhow::Result>> { + assert_eq!(tokio::runtime::Handle::current().id(), self.handle.id()); + self.prepare_started.notify_one(); + if self.block_prepare { + std::future::pending::<()>().await; + } + Ok(None) + } + + async fn shutdown(&self) { + assert_eq!(tokio::runtime::Handle::current().id(), self.handle.id()); + self.shutdown_calls.fetch_add(1, Ordering::Relaxed); + } + } + + for block_prepare in [false, true] { + let executor = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .unwrap(); + let caller = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let runtime_host = Arc::new(ExecutorRuntimeHost { + handle: executor.handle().clone(), + block_prepare, + prepare_started: Notify::new(), + shutdown_calls: AtomicUsize::new(0), + }); + let instance = { + let _runtime = executor.enter(); + let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16); + let mut adapters = adapters(None, Arc::new(packet_sink)); + adapters.instance_runtime = runtime_host.clone(); + CoreInstance::new(test_config("host-executor"), adapters).unwrap() + }; + caller.block_on(async { + if block_prepare { + let start = tokio::spawn({ + let instance = instance.clone(); + async move { instance.start().await } + }); + tokio::time::timeout( + Duration::from_secs(2), + runtime_host.prepare_started.notified(), + ) + .await + .unwrap(); + start.abort(); + assert!(start.await.unwrap_err().is_cancelled()); + } else { + instance.start().await.unwrap(); + assert_eq!(instance.state(), CoreInstanceState::Running); + instance.stop().await; + } + tokio::time::timeout(Duration::from_secs(2), async { + while instance.state() != CoreInstanceState::Stopped { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .expect("lifecycle cleanup must complete on the Host executor"); + assert_eq!(runtime_host.shutdown_calls.load(Ordering::Relaxed), 1); + }); + } + } + + #[test] + fn cancelled_stop_before_executor_poll_still_finishes_cleanup() { + struct ExecutorRuntimeHost { + handle: tokio::runtime::Handle, + shutdown_calls: AtomicUsize, + } + + #[async_trait] + impl InstanceRuntimeHost for ExecutorRuntimeHost { + fn runtime_handle(&self) -> Option { + Some(self.handle.clone()) + } + + async fn prepare( + &self, + _packet_plane: Arc, + ) -> anyhow::Result>> { + Ok(None) + } + + async fn shutdown(&self) { + assert_eq!(tokio::runtime::Handle::current().id(), self.handle.id()); + self.shutdown_calls.fetch_add(1, Ordering::Relaxed); + } + } + + let executor = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let runtime_host = Arc::new(ExecutorRuntimeHost { + handle: executor.handle().clone(), + shutdown_calls: AtomicUsize::new(0), + }); + let instance = { + let _runtime = executor.enter(); + let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16); + let mut adapters = adapters(None, Arc::new(packet_sink)); + adapters.instance_runtime = runtime_host.clone(); + CoreInstance::new(test_config("cancel-stop-before-poll"), adapters).unwrap() + }; + executor.block_on(instance.start()).unwrap(); + assert_eq!(instance.state(), CoreInstanceState::Running); + + let caller = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + caller.block_on(async { + // The instance executor is not driven here, so the dispatched stop + // cannot be polled before its caller is cancelled. + let mut stop = Box::pin(instance.stop()); + assert!(futures::poll!(&mut stop).is_pending()); + drop(stop); + }); + drop(caller); + + executor.block_on(async { + tokio::time::timeout(Duration::from_secs(2), async { + while instance.state() != CoreInstanceState::Stopped { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .expect("cancelled stop must still finish on the instance executor"); + }); + assert_eq!(runtime_host.shutdown_calls.load(Ordering::Relaxed), 1); + } + #[cfg(feature = "management")] #[tokio::test] async fn cancelled_delete_finishes_stop_and_keeps_wait_blocked() { diff --git a/easytier/Cargo.toml b/easytier/Cargo.toml index 02d8d68f..6939019e 100644 --- a/easytier/Cargo.toml +++ b/easytier/Cargo.toml @@ -70,6 +70,7 @@ tokio = { workspace = true, features = [ "net", "process", "rt", + "rt-multi-thread", "signal", "sync", "time", diff --git a/easytier/src/instance/composition.rs b/easytier/src/instance/composition.rs index 05e8ba72..f6cc4a73 100644 --- a/easytier/src/instance/composition.rs +++ b/easytier/src/instance/composition.rs @@ -31,7 +31,10 @@ use crate::{ instance::runtime_host::NativeInstanceRuntimeHost, }; -use super::host::{NativeInstanceHost, native_instance_host}; +use super::{ + host::{NativeInstanceHost, native_instance_host}, + runtime_executor::NativeInstanceExecutor, +}; #[cfg(feature = "kcp")] use crate::gateway::kcp_proxy::KcpProxyService; #[cfg(feature = "quic")] @@ -42,10 +45,11 @@ use easytier_core::gateway::proxy::wrapped_transport::WrappedTransportEngine; pub(crate) type NativeCoreInstance = CoreInstance; -pub(crate) fn compose_native_core_instance( +pub(super) fn compose_native_core_instance( toml_config: TomlConfig, process_runtime: Arc, compact_runtime: bool, + executor: Option, ) -> anyhow::Result> { let host_config = if compact_runtime { compact_runtime_core_host_config() @@ -58,7 +62,7 @@ pub(crate) fn compose_native_core_instance( &normalized, &host_config, )); - let runtime_host = NativeInstanceRuntimeHost::new(global_ctx.clone()); + let runtime_host = NativeInstanceRuntimeHost::new_with_executor(global_ctx.clone(), executor); let mut adapters = runtime_core_host_adapters_with_packet_egress_and_config( global_ctx.clone(), process_runtime, diff --git a/easytier/src/instance/factory.rs b/easytier/src/instance/factory.rs index 67dfb98c..71c341ce 100644 --- a/easytier/src/instance/factory.rs +++ b/easytier/src/instance/factory.rs @@ -5,7 +5,7 @@ use easytier_core::instance::manager::InstanceManager; #[cfg(feature = "management-rpc")] use easytier_core::management::ProcessRuntimeProvider; use easytier_core::{ - config::toml::TomlConfig, + config::toml::{ConfigLoader as _, TomlConfig}, instance::{CoreInstance, manager::InstanceFactory}, process_runtime::CoreProcessRuntime, }; @@ -14,7 +14,7 @@ use crate::common::global_ctx::EventBusSubscriber; use super::{ composition::compose_native_core_instance, host::NativeInstanceHost, - runtime_host::NativeInstanceRuntimeHost, + runtime_executor::NativeInstanceExecutor, runtime_host::NativeInstanceRuntimeHost, }; pub type NativeCoreInstance = CoreInstance; @@ -142,14 +142,25 @@ impl InstanceFactory for NativeInstanceFactory { config: TomlConfig, (): Self::CreateContext, ) -> Result, Self::Error> { - let _runtime = self - .runtime_handle + let executor = if self.compact_runtime { + None + } else { + let flags = config.get_flags(); + Some(NativeInstanceExecutor::new( + flags.multi_thread, + flags.multi_thread_count, + )?) + }; + let runtime_handle = executor .as_ref() - .map(tokio::runtime::Handle::enter); + .map(NativeInstanceExecutor::handle) + .or_else(|| self.runtime_handle.clone()); + let _runtime = runtime_handle.as_ref().map(tokio::runtime::Handle::enter); let instance = compose_native_core_instance( config, self.process_runtime.clone(), self.compact_runtime, + executor, )?; #[cfg(feature = "logging")] if self.log_cli_events { @@ -170,10 +181,297 @@ impl ProcessRuntimeProvider for NativeInstanceFactory { #[cfg(test)] mod tests { - use easytier_core::{config::toml::ConfigLoader as _, instance::CoreInstanceState}; + use std::{future::pending, time::Duration}; + + use easytier_core::instance::{CoreInstanceState, manager::ConfigFileControl}; + use tokio::runtime::RuntimeFlavor; use super::*; + fn isolated_config(multi_thread: bool, count: u32) -> TomlConfig { + let config = TomlConfig::default(); + let mut flags = config.get_flags(); + flags.no_tun = true; + flags.enable_ipv6 = false; + flags.disable_upnp = true; + flags.disable_p2p = true; + flags.disable_tcp_hole_punching = true; + flags.disable_udp_hole_punching = true; + flags.multi_thread = multi_thread; + flags.multi_thread_count = count; + config.set_flags(flags); + config.set_listeners(Vec::new()); + config.set_stun_servers(Some(Vec::new())); + config.set_stun_servers_v6(Some(Vec::new())); + config.set_tcp_stun_servers(Some(Vec::new())); + config + } + + #[tokio::test] + async fn manager_honors_per_instance_runtime_configuration_and_teardown() { + let caller = tokio::runtime::Handle::current(); + let manager = native_instance_manager_with_runtime(caller.clone()); + let mut runtimes = Vec::new(); + let mut ids = Vec::new(); + for (multi_thread, count, workers) in [(false, 8, 1), (true, 4, 4)] { + let config = isolated_config(multi_thread, count); + let id = manager + .run_network_instance(config, ConfigFileControl::STATIC_CONFIG) + .unwrap(); + let runtime = manager.data_plane_runtime_handle(&id).unwrap(); + assert_ne!(runtime.id(), caller.id()); + assert_eq!(runtime.metrics().num_workers(), workers); + assert_eq!( + runtime.runtime_flavor(), + if multi_thread { + RuntimeFlavor::MultiThread + } else { + RuntimeFlavor::CurrentThread + } + ); + ids.push(id); + runtimes.push(runtime); + } + assert_ne!(runtimes[0].id(), runtimes[1].id()); + tokio::time::timeout(Duration::from_secs(5), async { + while manager + .instances() + .iter() + .any(|instance| !instance.is_ready()) + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("instances must start on their own executors"); + + // Shared process resources must continue to work across distinct executors. + let single = manager.instance(ids[0]).unwrap(); + let multi = manager.instance(ids[1]).unwrap(); + multi + .add_connector(format!("ring://{}", ids[0]).parse().unwrap()) + .unwrap(); + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if single.connected_peers().await.contains(&multi.peer_id()) + && multi.connected_peers().await.contains(&single.peer_id()) + { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("single- and multi-thread instances must connect through the shared ring registry"); + drop(single); + drop(multi); + + let instances = manager.instances(); + let tasks: Vec<_> = runtimes + .iter() + .map(|runtime| runtime.spawn(pending::<()>())) + .collect(); + manager.delete_network_instances(ids.clone()).await.unwrap(); + assert!( + instances + .iter() + .all(|instance| instance.state() == CoreInstanceState::Stopped) + ); + assert!( + ids.iter() + .all(|id| manager.data_plane_runtime_handle(id).is_none()) + ); + drop(instances); + for task in tasks { + assert!( + tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("deleting the last instance owner must shut down its executor") + .unwrap_err() + .is_cancelled() + ); + } + } + + #[test] + fn direct_instance_outlives_its_callers_runtime() { + for multi_thread in [false, true] { + let caller = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + // Standard native construction no longer needs an entered caller runtime. + let instance = create_native_instance(isolated_config(multi_thread, 3)).unwrap(); + caller.block_on(instance.start()).unwrap(); + drop(caller); + let executor = instance.runtime_handle().unwrap(); + let caller = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + caller.block_on(async { + tokio::time::timeout( + Duration::from_secs(5), + executor.spawn(async { + tokio::time::sleep(Duration::from_millis(10)).await; + }), + ) + .await + .unwrap() + .unwrap(); + instance.stop().await; + }); + assert_eq!(instance.state(), CoreInstanceState::Stopped); + drop(instance); + } + } + + #[cfg(all(feature = "smoltcp", feature = "management"))] + #[test] + fn hot_added_port_forwards_outlive_the_callers_runtime() { + use easytier_core::{ + config::{gateway::PortForwardConfig, runtime::CoreInstanceRuntimeConfig}, + instance::CoreInstanceConfig, + }; + use easytier_proto::api::config::{ + ConfigPatchAction, InstanceConfigPatch, PortForwardPatch, + }; + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + + for multi_thread in [false, true] { + for via_patch in [false, true] { + let observer = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let (tcp_echo, udp_echo) = observer.block_on(async { + ( + tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(), + tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap(), + ) + }); + let tcp_reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let udp_reservation = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let forwards = vec![ + PortForwardConfig { + bind_addr: tcp_reservation.local_addr().unwrap(), + dst_addr: tcp_echo.local_addr().unwrap(), + proto: "tcp".to_owned(), + }, + PortForwardConfig { + bind_addr: udp_reservation.local_addr().unwrap(), + dst_addr: udp_echo.local_addr().unwrap(), + proto: "udp".to_owned(), + }, + ]; + let config = isolated_config(multi_thread, 2); + // Local virtual-IP forwarding uses native loopback endpoints, + // avoiding TUN devices, network namespaces and remote peers. + config.set_ipv4(Some("127.0.0.1/32".parse().unwrap())); + let instance = create_native_instance(config.clone()).unwrap(); + let caller = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + caller.block_on(async { + instance.start().await.unwrap(); + drop(tcp_reservation); + drop(udp_reservation); + if via_patch { + easytier_core::management::apply_config_patch( + &instance, + InstanceConfigPatch { + port_forwards: forwards + .iter() + .cloned() + .map(|forward| PortForwardPatch { + action: ConfigPatchAction::Add as i32, + cfg: Some(forward.into()), + }) + .collect(), + ..Default::default() + }, + None, + ) + .await + .unwrap(); + } else { + let mut normalized = CoreInstanceConfig::from_toml(&config).unwrap(); + normalized.connectivity.runtime.gateway.port_forwards = forwards.clone(); + instance + .update_runtime_config(CoreInstanceRuntimeConfig { + services: normalized.connectivity.runtime, + peer: Arc::new(normalized.peer.snapshot), + }) + .await + .unwrap(); + } + }); + // No accept/receive task or reactor may depend on this runtime. + drop(caller); + assert_eq!(instance.state(), CoreInstanceState::Running); + + observer.block_on(async { + tokio::time::timeout(Duration::from_secs(5), async { + let tcp_response = tokio::spawn(async move { + let (mut stream, _) = tcp_echo.accept().await.unwrap(); + let mut request = [0; 4]; + stream.read_exact(&mut request).await.unwrap(); + stream.write_all(&request).await.unwrap(); + }); + let udp_response = tokio::spawn(async move { + let mut request = [0; 4]; + let (len, source) = udp_echo.recv_from(&mut request).await.unwrap(); + udp_echo.send_to(&request[..len], source).await.unwrap(); + }); + let mut stream = tokio::net::TcpStream::connect(forwards[0].bind_addr) + .await + .unwrap(); + stream.write_all(b"ping").await.unwrap(); + let mut reply = [0; 4]; + stream.read_exact(&mut reply).await.unwrap(); + assert_eq!(&reply, b"ping"); + tcp_response.await.unwrap(); + + let socket = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap(); + socket + .send_to(b"pong", forwards[1].bind_addr) + .await + .unwrap(); + let (len, _) = socket.recv_from(&mut reply).await.unwrap(); + assert_eq!(&reply[..len], b"pong"); + udp_response.await.unwrap(); + }) + .await + .expect("hot-added TCP/UDP forwarding must survive caller-runtime shutdown"); + instance.stop().await; + }); + assert_eq!(instance.state(), CoreInstanceState::Stopped); + } + } + } + + #[tokio::test] + async fn compact_manager_retains_the_supplied_runtime() { + let caller = tokio::runtime::Handle::current(); + let manager = native_compact_instance_manager_with_runtime(caller.clone()); + let instance = manager.create(isolated_config(true, 8), ()).unwrap(); + assert!(instance.runtime_handle().is_none()); + assert_eq!( + manager + .data_plane_runtime_handle(&instance.instance_id()) + .unwrap() + .id(), + caller.id() + ); + instance.start().await.unwrap(); + manager + .delete_network_instances([instance.instance_id()]) + .await + .unwrap(); + assert_eq!(instance.state(), CoreInstanceState::Stopped); + } + #[tokio::test] async fn core_manager_stores_and_runs_native_core_instance_directly() { let factory = NativeInstanceFactory::new(CoreProcessRuntime::new()); diff --git a/easytier/src/instance/mod.rs b/easytier/src/instance/mod.rs index 39dc3200..4657ba43 100644 --- a/easytier/src/instance/mod.rs +++ b/easytier/src/instance/mod.rs @@ -7,6 +7,7 @@ pub(crate) mod config_storage; pub mod dns_server; pub mod factory; pub mod host; +mod runtime_executor; pub(crate) mod runtime_host; #[cfg(test)] pub(crate) mod test_instance; diff --git a/easytier/src/instance/runtime_executor.rs b/easytier/src/instance/runtime_executor.rs new file mode 100644 index 00000000..bc3284e1 --- /dev/null +++ b/easytier/src/instance/runtime_executor.rs @@ -0,0 +1,109 @@ +//! Native ownership and continuous driving of one instance's Tokio runtime. + +use std::sync::mpsc; + +use tokio::{ + runtime::{Builder, Handle}, + sync::oneshot, +}; + +pub(super) struct NativeInstanceExecutor { + handle: Handle, + // Dropping the sender stops the driver, including on construction failure. + _shutdown: oneshot::Sender<()>, +} + +impl NativeInstanceExecutor { + pub(super) fn new(multi_thread: bool, thread_count: u32) -> anyhow::Result { + let (ready_tx, ready_rx) = mpsc::sync_channel(1); + let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); + std::thread::Builder::new() + .name("easytier-instance".to_owned()) + .spawn(move || { + let mut builder = if multi_thread { + let mut builder = Builder::new_multi_thread(); + builder.worker_threads(2.max(thread_count as usize)); + builder + } else { + Builder::new_current_thread() + }; + let runtime = match builder.enable_all().build() { + Ok(runtime) => runtime, + Err(error) => { + let _ = ready_tx.send(Err(error)); + return; + } + }; + if ready_tx.send(Ok(runtime.handle().clone())).is_ok() { + runtime.block_on(async { + let _ = shutdown_rx.await; + }); + } + // Never block an instance worker waiting for its own shutdown. + // Blocking Host operations may finish after the owner is dropped. + runtime.shutdown_background(); + })?; + let handle = ready_rx.recv()??; + Ok(Self { + handle, + _shutdown: shutdown_tx, + }) + } + + pub(super) fn handle(&self) -> Handle { + self.handle.clone() + } +} + +#[cfg(test)] +mod tests { + use std::{future::pending, time::Duration}; + + use tokio::runtime::RuntimeFlavor; + + use super::*; + + #[tokio::test] + async fn honors_runtime_flavor_and_worker_count() { + for (multi_thread, count, expected_workers) in [ + (false, 8, 1), + (true, 0, 2), + (true, 1, 2), + (true, 2, 2), + (true, 4, 4), + ] { + let executor = NativeInstanceExecutor::new(multi_thread, count).unwrap(); + let handle = executor.handle(); + assert_eq!(handle.metrics().num_workers(), expected_workers); + assert_eq!( + handle.runtime_flavor(), + if multi_thread { + RuntimeFlavor::MultiThread + } else { + RuntimeFlavor::CurrentThread + } + ); + let caller_thread = std::thread::current().id(); + let task_thread = handle + .spawn(async { std::thread::current().id() }) + .await + .unwrap(); + assert_ne!(task_thread, caller_thread); + } + } + + #[tokio::test] + async fn dropping_owner_from_a_worker_shuts_down_tasks() { + for multi_thread in [false, true] { + let executor = NativeInstanceExecutor::new(multi_thread, 2).unwrap(); + let handle = executor.handle(); + let task = handle.spawn(pending::<()>()); + handle.spawn(async move { drop(executor) }); + let error = tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("runtime must shut down without joining its own worker") + .unwrap_err(); + assert!(error.is_cancelled()); + } + } +} diff --git a/easytier/src/instance/runtime_host.rs b/easytier/src/instance/runtime_host.rs index 93c0d7fc..aa22a702 100644 --- a/easytier/src/instance/runtime_host.rs +++ b/easytier/src/instance/runtime_host.rs @@ -25,6 +25,7 @@ mod tun_runtime; #[path = "runtime_host/tun_mobile.rs"] mod tun_runtime; +use super::runtime_executor::NativeInstanceExecutor; use event_journal::EventJournal; #[cfg(feature = "tun")] use magic_dns::MagicDnsRuntime; @@ -36,10 +37,19 @@ pub(crate) struct NativeInstanceRuntimeHost { cancel: CancellationToken, event_journal: EventJournal, tun: NativeTunRuntime, + executor: Option, } impl NativeInstanceRuntimeHost { + #[cfg(test)] pub(crate) fn new(global_ctx: ArcGlobalCtx) -> Arc { + Self::new_with_executor(global_ctx, None) + } + + pub(super) fn new_with_executor( + global_ctx: ArcGlobalCtx, + executor: Option, + ) -> Arc { let cancel = CancellationToken::new(); let tun = NativeTunRuntime::new(global_ctx.clone(), cancel.clone()); let event_journal = EventJournal::new(&global_ctx); @@ -49,6 +59,7 @@ impl NativeInstanceRuntimeHost { operation: Arc::new(Mutex::new(())), cancel, tun, + executor, }) } diff --git a/easytier/src/instance/runtime_host/implementation.rs b/easytier/src/instance/runtime_host/implementation.rs index 9484f3d7..1d5f21e1 100644 --- a/easytier/src/instance/runtime_host/implementation.rs +++ b/easytier/src/instance/runtime_host/implementation.rs @@ -10,6 +10,10 @@ use super::NativeInstanceRuntimeHost; #[async_trait::async_trait] impl InstanceRuntimeHost for NativeInstanceRuntimeHost { + fn runtime_handle(&self) -> Option { + self.executor.as_ref().map(|executor| executor.handle()) + } + async fn prepare( &self, packet_plane: Arc,