diff --git a/docs/core-architecture.md b/docs/core-architecture.md index ecf07bf5..d1e268c9 100644 --- a/docs/core-architecture.md +++ b/docs/core-architecture.md @@ -364,13 +364,18 @@ 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. +By default, 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. The `easytier-core --shared-runtime` option instead shares one +multi-thread executor across the factory and all its instances. Its worker count +defaults to the available CPU count and can be set with the process-level +`--multi-thread-count` option, with a minimum of one; per-instance runtime flags +are ignored in this mode. Construction, start/stop, and FFI data-plane tasks use +the selected executor, independent of the process management runtime. Dropping +the last factory or instance 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 diff --git a/easytier/locales/app.yml b/easytier/locales/app.yml index a88e2a85..69d9704e 100644 --- a/easytier/locales/app.yml +++ b/easytier/locales/app.yml @@ -127,11 +127,14 @@ core_clap: en: "encryption algorithm to use, supported: '', 'xor', 'chacha20', 'aes-gcm', 'aes-gcm-256', 'openssl-aes128-gcm', 'openssl-aes256-gcm', 'openssl-chacha20'. Empty string means default (aes-gcm)" zh-CN: "要使用的加密算法,支持:''(默认aes-gcm)、'xor'、'chacha20'、'aes-gcm'、'aes-gcm-256'、'openssl-aes128-gcm'、'openssl-aes256-gcm'、'openssl-chacha20'" multi_thread: - en: "use multi-thread runtime, default is single-thread" - zh-CN: "使用多线程运行时,默认为单线程" + en: "use a multi-thread runtime for each instance, default is true; ignored with --shared-runtime" + zh-CN: "为每个实例使用多线程运行时,默认开启;启用 --shared-runtime 时忽略此项" multi_thread_count: - en: "the number of threads to use, default is 2, only effective when multi-thread is enabled, must be greater than 2" - zh-CN: "使用的线程数,默认为2,仅在多线程模式下有效。取值必须大于2" + en: "runtime worker count: default/minimum 2 per instance; with --shared-runtime, defaults to available CPUs with a minimum of 1" + zh-CN: "运行时工作线程数:独占多线程模式默认及最小值为 2;启用 --shared-runtime 时默认为可用 CPU 数,最小值为 1" + shared_runtime: + en: "share one multi-thread runtime across all instances; --multi-thread-count sets its worker count independently of instance configuration files" + zh-CN: "所有实例共用一个多线程运行时;通过 --multi-thread-count 设置工作线程数,不受实例配置文件影响" disable_ipv6: en: "do not use ipv6" zh-CN: "不使用IPv6" diff --git a/easytier/src/core.rs b/easytier/src/core.rs index fde9fa58..8e465f72 100644 --- a/easytier/src/core.rs +++ b/easytier/src/core.rs @@ -132,6 +132,13 @@ struct Cli { )] config_dir: Option, + #[arg( + long, + env = "ET_SHARED_RUNTIME", + help = t!("core_clap.shared_runtime").to_string() + )] + shared_runtime: bool, + #[command(flatten)] network_options: NetworkOptions, @@ -506,6 +513,26 @@ struct RpcPortalOptions { } impl Cli { + fn take_runtime_thread_count(&mut self) -> u32 { + let count = self + .network_options + .flags + .multi_thread_count + .unwrap_or_else(|| { + if self.shared_runtime { + std::thread::available_parallelism().map_or(1, |count| count.get() as u32) + } else { + Flags::defaults().multi_thread_count + } + }); + if self.shared_runtime { + // Process runtime options must not make instance configs read-only. + self.network_options.flags.multi_thread = None; + self.network_options.flags.multi_thread_count = None; + } + count + } + fn gen_listeners(addr: SocketAddr) -> impl Iterator { let dynamic = addr.port() == 0; IpScheme::VARIANTS.iter().map(move |proto| { @@ -1210,7 +1237,7 @@ fn win_service_main(arg: Vec) { win_service_event_loop(stop_notify_recv, cli, status_handle); } -async fn run_main(cli: Cli) -> anyhow::Result<()> { +async fn run_main(mut cli: Cli) -> anyhow::Result<()> { defer!(dump_profile(0);); log::init_with_default_console_targets( &cli.logging_options, @@ -1218,7 +1245,10 @@ async fn run_main(cli: Cli) -> anyhow::Result<()> { &["CORE", "easytier_core::instance::manager"], )?; - let manager = Arc::new(native_cli_instance_manager().with_config_path(cli.config_dir.clone())); + let manager = Arc::new( + native_cli_instance_manager(cli.shared_runtime, cli.take_runtime_thread_count())? + .with_config_path(cli.config_dir.clone()), + ); let _rpc_server = ApiRpcServer::new( cli.rpc_portal_options.rpc_portal, @@ -1544,6 +1574,64 @@ mod tests { flags_from(&cli_command().try_get_matches_from(argv)?) } + #[test] + fn shared_runtime_is_process_only_and_reuses_thread_count() { + let cpus = std::thread::available_parallelism().map_or(1, |count| count.get() as u32); + for (args, shared, expected_count) in [ + (vec![], false, Flags::defaults().multi_thread_count), + (vec!["--shared-runtime"], true, cpus), + (vec!["--multi-thread-count", "4"], false, 4), + ( + vec![ + "--shared-runtime", + "--multi-thread-count", + "1", + "--multi-thread", + "false", + ], + true, + 1, + ), + ( + vec![ + "--shared-runtime", + "--multi-thread-count", + "4", + "-c", + "a.toml", + "b.toml", + ], + true, + 4, + ), + ] { + let matches = cli_command() + .try_get_matches_from(std::iter::once("easytier-core").chain(args)) + .unwrap(); + let mut cli = Cli::from_arg_matches(&matches).unwrap(); + cli.network_options.flags = flags_from(&matches).unwrap(); + assert_eq!(cli.shared_runtime, shared); + assert_eq!(cli.take_runtime_thread_count(), expected_count); + + let config = TomlConfigLoader::default(); + if shared { + assert_eq!(cli.network_options.flags.multi_thread, None); + assert_eq!(cli.network_options.flags.multi_thread_count, None); + assert!(!cli.network_options.can_merge( + &config, + ConfigFileSource::CliConfigFile, + 1, + 0, + )); + } + cli.network_options.merge_into(&config).unwrap(); + assert!(!config.dump().contains("shared_runtime")); + if cli.network_options.flags.multi_thread_count.is_none() { + assert_eq!(config.get_flags_patch().multi_thread_count, None); + } + } + } + #[test] fn test_parse_listeners() { type IpSchemeMap = fn(&IpScheme) -> String; diff --git a/easytier/src/instance/composition.rs b/easytier/src/instance/composition.rs index c1bc0078..b9fa01e2 100644 --- a/easytier/src/instance/composition.rs +++ b/easytier/src/instance/composition.rs @@ -52,7 +52,7 @@ pub(super) fn compose_native_core_instance( process_runtime: Arc, #[cfg(feature = "tun")] shared_virtual_nic_registry: ArcSharedVirtualNicRegistry, compact_runtime: bool, - executor: Option, + executor: Option>, ) -> anyhow::Result> { let host_config = if compact_runtime { compact_runtime_core_host_config() diff --git a/easytier/src/instance/factory.rs b/easytier/src/instance/factory.rs index 059f1ea9..dfd5cd31 100644 --- a/easytier/src/instance/factory.rs +++ b/easytier/src/instance/factory.rs @@ -41,12 +41,16 @@ pub fn native_instance_manager() -> NativeInstanceManager { } #[cfg(feature = "management")] -pub fn native_cli_instance_manager() -> NativeInstanceManager { +pub fn native_cli_instance_manager( + shared_runtime: bool, + thread_count: u32, +) -> anyhow::Result { let process_runtime = CoreProcessRuntime::new(); - InstanceManager::new( - NativeInstanceFactory::new(process_runtime).with_cli_event_logging(), - None, - ) + let mut factory = NativeInstanceFactory::new(process_runtime).with_cli_event_logging(); + if shared_runtime { + factory.shared_executor = Some(Arc::new(NativeInstanceExecutor::new(true, thread_count)?)); + } + Ok(InstanceManager::new(factory, None)) } pub fn create_native_instance(config: TomlConfig) -> anyhow::Result> { @@ -111,6 +115,7 @@ pub struct NativeInstanceFactory { #[cfg(feature = "tun")] shared_virtual_nic_registry: ArcSharedVirtualNicRegistry, runtime_handle: Option, + shared_executor: Option>, compact_runtime: bool, #[cfg(feature = "logging")] log_cli_events: bool, @@ -123,6 +128,7 @@ impl NativeInstanceFactory { #[cfg(feature = "tun")] shared_virtual_nic_registry: Arc::new(Mutex::new(SharedVirtualNicRegistry::new())), runtime_handle: None, + shared_executor: None, compact_runtime: false, #[cfg(feature = "logging")] log_cli_events: false, @@ -160,16 +166,18 @@ impl InstanceFactory for NativeInstanceFactory { ) -> Result, Self::Error> { let executor = if self.compact_runtime { None + } else if let Some(executor) = &self.shared_executor { + Some(executor.clone()) } else { let flags = config.get_flags(); - Some(NativeInstanceExecutor::new( + Some(Arc::new(NativeInstanceExecutor::new( flags.multi_thread, - flags.multi_thread_count, - )?) + flags.multi_thread_count.max(2), + )?)) }; let runtime_handle = executor .as_ref() - .map(NativeInstanceExecutor::handle) + .map(|executor| executor.handle()) .or_else(|| self.runtime_handle.clone()); let _runtime = runtime_handle.as_ref().map(tokio::runtime::Handle::enter); let instance = compose_native_core_instance( @@ -231,7 +239,9 @@ mod tests { 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)] { + for (multi_thread, count, workers) in + [(false, 8, 1), (true, 4, 4), (true, 0, 2), (true, 1, 2)] + { let config = isolated_config(multi_thread, count); let id = manager .run_network_instance(config, ConfigFileControl::STATIC_CONFIG) @@ -311,6 +321,89 @@ mod tests { } } + #[cfg(feature = "management")] + #[tokio::test] + async fn shared_runtime_survives_instance_deletion_and_stops_with_last_owner() { + for workers in [1, 4] { + let manager = native_cli_instance_manager(true, workers).unwrap(); + let first = manager.create(isolated_config(false, 8), ()).unwrap(); + let second = manager.create(isolated_config(true, 2), ()).unwrap(); + let runtime = first.runtime_handle().unwrap(); + assert_ne!(runtime.id(), tokio::runtime::Handle::current().id()); + assert_eq!(runtime.id(), second.runtime_handle().unwrap().id()); + assert_eq!(runtime.runtime_flavor(), RuntimeFlavor::MultiThread); + assert_eq!(runtime.metrics().num_workers(), workers as usize); + + first.start().await.unwrap(); + second.start().await.unwrap(); + second + .add_connector(format!("ring://{}", first.instance_id()).parse().unwrap()) + .unwrap(); + tokio::time::timeout(Duration::from_secs(10), async { + while !first.connected_peers().await.contains(&second.peer_id()) + || !second.connected_peers().await.contains(&first.peer_id()) + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("instances must connect on the shared runtime"); + + manager + .delete_network_instances([first.instance_id()]) + .await + .unwrap(); + assert_eq!(first.state(), CoreInstanceState::Stopped); + drop(first); + assert_eq!(second.state(), CoreInstanceState::Running); + tokio::time::timeout( + Duration::from_secs(5), + runtime.spawn(async { + tokio::time::sleep(Duration::from_millis(10)).await; + }), + ) + .await + .unwrap() + .unwrap(); + + let added = manager.create(isolated_config(false, 16), ()).unwrap(); + assert_eq!(runtime.id(), added.runtime_handle().unwrap().id()); + added.start().await.unwrap(); + added + .add_connector(format!("ring://{}", second.instance_id()).parse().unwrap()) + .unwrap(); + tokio::time::timeout(Duration::from_secs(10), async { + while !second.connected_peers().await.contains(&added.peer_id()) + || !added.connected_peers().await.contains(&second.peer_id()) + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("deleting one instance must leave the others functional"); + manager + .delete_network_instances([added.instance_id(), second.instance_id()]) + .await + .unwrap(); + drop(added); + assert_eq!(second.state(), CoreInstanceState::Stopped); + + // A surviving instance owner keeps the executor alive after its manager. + let task = runtime.spawn(pending::<()>()); + drop(manager); + runtime.spawn(async {}).await.unwrap(); + assert!(!task.is_finished()); + drop(second); + assert!( + tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("the last owner must shut down the shared executor") + .unwrap_err() + .is_cancelled() + ); + } + } + #[test] fn direct_instance_outlives_its_callers_runtime() { for multi_thread in [false, true] { diff --git a/easytier/src/instance/runtime_executor.rs b/easytier/src/instance/runtime_executor.rs index bc3284e1..005b4d73 100644 --- a/easytier/src/instance/runtime_executor.rs +++ b/easytier/src/instance/runtime_executor.rs @@ -1,4 +1,4 @@ -//! Native ownership and continuous driving of one instance's Tokio runtime. +//! Native ownership and continuous driving of an instance or shared Tokio runtime. use std::sync::mpsc; @@ -22,7 +22,7 @@ impl NativeInstanceExecutor { .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.worker_threads(1.max(thread_count as usize)); builder } else { Builder::new_current_thread() @@ -67,8 +67,8 @@ mod tests { 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, 0, 1), + (true, 1, 1), (true, 2, 2), (true, 4, 4), ] { diff --git a/easytier/src/instance/runtime_host.rs b/easytier/src/instance/runtime_host.rs index 3d221b5b..8a32be27 100644 --- a/easytier/src/instance/runtime_host.rs +++ b/easytier/src/instance/runtime_host.rs @@ -41,7 +41,7 @@ pub(crate) struct NativeInstanceRuntimeHost { cancel: CancellationToken, event_journal: EventJournal, tun: NativeTunRuntime, - executor: Option, + executor: Option>, } impl NativeInstanceRuntimeHost { @@ -61,7 +61,7 @@ impl NativeInstanceRuntimeHost { pub(super) fn new_with_executor( global_ctx: ArcGlobalCtx, #[cfg(feature = "tun")] shared_virtual_nic_registry: ArcSharedVirtualNicRegistry, - executor: Option, + executor: Option>, ) -> Arc { let cancel = CancellationToken::new(); let tun = NativeTunRuntime::new(