feat(core): add an opt-in shared runtime for native instances (#2757)

* feat(core): share an instance runtime with CLI worker control

Add --shared-runtime to easytier-core and reuse --multi-thread-count.
Default shared workers to available CPUs while retaining dedicated
runtime defaults. Consume shared runtime flags at process scope so
config files remain writable and keep their own thread settings.

Share executor ownership across the factory and its instances, keeping
normal instance capabilities and independent shutdown paths. Cover
worker selection, instance deletion, hot addition, and final teardown.

* docs: document shared native runtime ownership

Describe the opt-in shared executor alongside the default per-instance model, including worker selection and last-owner shutdown semantics.

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

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"

easytier/src/core.rs

@@ -132,6 +132,13 @@ struct Cli {
     )]
     config_dir: Option<PathBuf>,
 
+    #[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<Item = String> {
         let dynamic = addr.port() == 0;
         IpScheme::VARIANTS.iter().map(move |proto| {
@@ -1210,7 +1237,7 @@ fn win_service_main(arg: Vec<std::ffi::OsString>) {
     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;

easytier/src/instance/composition.rs

@@ -52,7 +52,7 @@ pub(super) fn compose_native_core_instance(
     process_runtime: Arc<CoreProcessRuntime>,
     #[cfg(feature = "tun")] shared_virtual_nic_registry: ArcSharedVirtualNicRegistry,
     compact_runtime: bool,
-    executor: Option<NativeInstanceExecutor>,
+    executor: Option<Arc<NativeInstanceExecutor>>,
 ) -> anyhow::Result<Arc<NativeCoreInstance>> {
     let host_config = if compact_runtime {
         compact_runtime_core_host_config()

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<NativeInstanceManager> {
     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<Arc<NativeCoreInstance>> {
@@ -111,6 +115,7 @@ pub struct NativeInstanceFactory {
     #[cfg(feature = "tun")]
     shared_virtual_nic_registry: ArcSharedVirtualNicRegistry,
     runtime_handle: Option<tokio::runtime::Handle>,
+    shared_executor: Option<Arc<NativeInstanceExecutor>>,
     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<Arc<Self::Instance>, 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] {

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),
         ] {

easytier/src/instance/runtime_host.rs

@@ -41,7 +41,7 @@ pub(crate) struct NativeInstanceRuntimeHost {
     cancel: CancellationToken,
     event_journal: EventJournal,
     tun: NativeTunRuntime,
-    executor: Option<NativeInstanceExecutor>,
+    executor: Option<Arc<NativeInstanceExecutor>>,
 }
 
 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<NativeInstanceExecutor>,
+        executor: Option<Arc<NativeInstanceExecutor>>,
     ) -> Arc<Self> {
         let cancel = CancellationToken::new();
         let tun = NativeTunRuntime::new(