diff --git a/src-tauri/src/cmd/service.rs b/src-tauri/src/cmd/service.rs index 0b475f99d..862b2cbb3 100644 --- a/src-tauri/src/cmd/service.rs +++ b/src-tauri/src/cmd/service.rs @@ -1,13 +1,11 @@ use super::{CmdResult, StringifyErr as _}; use crate::core::service::{self, SERVICE_MANAGER, ServiceStatus}; -use smartstring::SmartString; async fn execute_service_operation_sync(status: ServiceStatus, op_type: &str) -> CmdResult { - if let Err(e) = SERVICE_MANAGER.lock().await.handle_service_status(&status).await { - let emsg = format!("{} Service failed: {}", op_type, e); - return Err(SmartString::from(emsg)); - } - Ok(()) + SERVICE_MANAGER + .handle_service_status(status) + .await + .map_err(|e| format!("{op_type} Service failed: {e}").into()) } #[tauri::command] diff --git a/src-tauri/src/core/manager/lifecycle.rs b/src-tauri/src/core/manager/lifecycle.rs index 93910162c..49e439260 100644 --- a/src-tauri/src/core/manager/lifecycle.rs +++ b/src-tauri/src/core/manager/lifecycle.rs @@ -12,7 +12,7 @@ use tauri_plugin_clash_verge_sysinfo; impl CoreManager { pub async fn start_core(&self) -> Result<()> { - self.prepare_startup().await?; + self.prepare_startup().await; defer! { self.after_core_process(); } @@ -62,18 +62,13 @@ impl CoreManager { Ok(()) } - async fn prepare_startup(&self) -> Result<()> { + async fn prepare_startup(&self) { #[cfg(target_os = "windows")] self.wait_for_service_if_needed().await; - - let value = SERVICE_MANAGER.lock().await.current(); - let mode = match value { + self.set_running_mode(match SERVICE_MANAGER.current().await { ServiceStatus::Ready => RunningMode::Service, _ => RunningMode::Sidecar, - }; - - self.set_running_mode(mode); - Ok(()) + }); } fn after_core_process(&self) { @@ -98,9 +93,7 @@ impl CoreManager { .with_max_times(max_times as usize); let _ = (|| async { - let mut manager = SERVICE_MANAGER.lock().await; - - if matches!(manager.current(), ServiceStatus::Ready) { + if matches!(SERVICE_MANAGER.current().await, ServiceStatus::Ready) { return Ok(()); } @@ -110,10 +103,10 @@ impl CoreManager { return Err(anyhow::anyhow!("Service IPC not ready")); } - manager.init().await?; - let _ = manager.refresh().await; + SERVICE_MANAGER.init().await?; + let _ = SERVICE_MANAGER.refresh().await; - if matches!(manager.current(), ServiceStatus::Ready) { + if matches!(SERVICE_MANAGER.current().await, ServiceStatus::Ready) { Ok(()) } else { Err(anyhow::anyhow!("Service not ready")) diff --git a/src-tauri/src/core/service.rs b/src-tauri/src/core/service.rs index 5ef7dd3cd..d32b1b2da 100644 --- a/src-tauri/src/core/service.rs +++ b/src-tauri/src/core/service.rs @@ -3,20 +3,24 @@ use crate::{ core::{logger::Logger, tray::Tray}, utils::dirs, }; -use anyhow::{Context as _, Result, anyhow, bail}; +use anyhow::{Context as _, Result, bail}; use backon::{ConstantBuilder, Retryable as _}; -use clash_verge_logging::{Type, logging, logging_error}; +use clash_verge_logging::{Type, logging}; use clash_verge_service_ipc::CoreConfig; use compact_str::CompactString; use once_cell::sync::Lazy; +use parking_lot::Mutex; +use scopeguard::defer; use std::{ borrow::Cow, env::current_exe, + future::Future, path::{Path, PathBuf}, process::Command as StdCommand, + sync::atomic::{AtomicBool, Ordering}, time::Duration, }; -use tokio::sync::Mutex; +use tokio::sync::Notify; #[derive(Debug, Clone, PartialEq, Eq)] pub enum ServiceStatus { @@ -29,8 +33,11 @@ pub enum ServiceStatus { Unavailable(String), } -#[derive(Clone)] -pub struct ServiceManager(ServiceStatus); +pub struct ServiceManager { + status: Mutex, + operation_running: AtomicBool, + operation_done: Notify, +} #[cfg(target_os = "windows")] fn uninstall_service() -> Result<()> { @@ -379,7 +386,7 @@ pub(super) async fn start_with_existing_service(config_file: &PathBuf) -> Result pub(super) async fn run_core_by_service(config_file: &PathBuf) -> Result<()> { logging!(info, Type::Service, "正在尝试通过服务启动核心"); - SERVICE_MANAGER.lock().await.refresh().await?; + SERVICE_MANAGER.refresh().await?; logging!(info, Type::Service, "服务已运行且版本匹配，直接使用"); start_with_existing_service(config_file).await @@ -435,24 +442,7 @@ pub async fn is_service_available() -> Result<()> { Ok(()) } -pub async fn wait_and_check_service_available(status: &mut ServiceManager) -> Result<()> { - wait_for_service_ipc(status, "Waiting for service to be available").await -} - -async fn wait_and_check_service_version(status: &mut ServiceManager) -> Result<()> { - wait_and_check_service_available(status).await?; - - if clash_verge_service_ipc::is_reinstall_service_needed().await { - logging!(info, Type::Service, "服务版本不匹配，执行重装流程"); - reinstall_service()?; - wait_and_check_service_available(status).await?; - } - - Ok(()) -} - -async fn wait_for_service_ipc(status: &mut ServiceManager, reason: &str) -> Result<()> { - status.0 = ServiceStatus::Unavailable(reason.into()); +async fn wait_for_service_ipc(manager: &ServiceManager) -> Result<()> { let config = ServiceManager::config(); let backoff = ConstantBuilder::default() @@ -460,18 +450,18 @@ async fn wait_for_service_ipc(status: &mut ServiceManager, reason: &str) -> Resu .with_max_times(config.max_retries); let result = (|| async { - if Path::new(clash_verge_service_ipc::IPC_PATH).exists() { - clash_verge_service_ipc::connect().await?; - Ok(()) - } else { - Err(anyhow!("IPC path not ready")) + if !is_service_ipc_path_exists() { + bail!("IPC path not ready"); } + clash_verge_service_ipc::connect().await.map(drop) }) .retry(backoff) .await; if result.is_ok() { - status.0 = ServiceStatus::Ready; + manager.set_status(ServiceStatus::Ready); + } else { + manager.set_status(ServiceStatus::Unavailable("Waiting for service to be available".into())); } result @@ -482,10 +472,6 @@ pub fn is_service_ipc_path_exists() -> bool { } impl ServiceManager { - pub fn default() -> Self { - Self(ServiceStatus::Unavailable("Need Checks".into())) - } - pub const fn config() -> clash_verge_service_ipc::IpcConfig { clash_verge_service_ipc::IpcConfig { default_timeout: Duration::from_millis(150), @@ -494,72 +480,109 @@ impl ServiceManager { } } - pub async fn init(&mut self) -> Result<()> { + pub async fn init(&self) -> Result<()> { if let Err(e) = clash_verge_service_ipc::connect().await { - self.0 = ServiceStatus::Unavailable("服务连接失败: {e}".to_string()); + self.set_status(ServiceStatus::Unavailable("服务连接失败: {e}".to_string())); return Err(e); } Ok(()) } - pub fn current(&self) -> ServiceStatus { - self.0.clone() - } - - pub async fn refresh(&mut self) -> Result<()> { - let status = self.check_service_comprehensive().await; - self.0 = status.clone(); - logging_error!(Type::Service, self.handle_service_status(&status).await); - Ok(()) - } - - /// 综合服务状态检查（一次性完成所有检查） - pub async fn check_service_comprehensive(&self) -> ServiceStatus { - if clash_verge_service_ipc::is_reinstall_service_needed().await { - ServiceStatus::NeedsReinstall - } else { - ServiceStatus::Ready + pub async fn current(&self) -> ServiceStatus { + loop { + let notified = self.operation_done.notified(); + if !self.operation_running.load(Ordering::Acquire) { + let status = self.status.lock().clone(); + if !self.operation_running.load(Ordering::Acquire) { + return status; + } + } + notified.await; } } - /// 根据服务状态执行相应操作 - pub async fn handle_service_status(&mut self, status: &ServiceStatus) -> Result<()> { - match status { - ServiceStatus::Ready => { - logging!(info, Type::Service, "服务就绪，直接启动"); - self.0 = ServiceStatus::Ready; + fn set_status(&self, status: ServiceStatus) { + *self.status.lock() = status; + } + + async fn run_operation(&self, operation: impl Future>) -> Result<()> { + { + if self.operation_running.swap(true, Ordering::AcqRel) { + bail!("service operation already running"); } + defer! { + self.operation_running.store(false, Ordering::Release); + self.operation_done.notify_waiters(); + } + + operation.await?; + } + + Tray::global().update_menu().await + } + + pub async fn refresh(&self) -> Result<()> { + self.run_operation(async { + self.apply_service_status(if clash_verge_service_ipc::is_reinstall_service_needed().await { + ServiceStatus::NeedsReinstall + } else { + ServiceStatus::Ready + }) + .await + }) + .await + } + + pub async fn handle_service_status(&self, status: ServiceStatus) -> Result<()> { + self.run_operation(self.apply_service_status(status)).await + } + + async fn apply_service_status(&self, status: ServiceStatus) -> Result<()> { + self.set_status(status.clone()); + match status { + ServiceStatus::Ready => logging!(info, Type::Service, "服务就绪，直接启动"), ServiceStatus::NeedsReinstall | ServiceStatus::ReinstallRequired => { logging!(info, Type::Service, "服务需要重装，执行重装流程"); - reinstall_service()?; - wait_and_check_service_available(self).await?; + run_service_command(reinstall_service, "reinstall service")?; + wait_for_service_ipc(self).await?; } ServiceStatus::ForceReinstallRequired => { logging!(info, Type::Service, "服务需要强制重装，执行强制重装流程"); - force_reinstall_service()?; - wait_and_check_service_available(self).await?; + run_service_command(force_reinstall_service, "force reinstall service")?; + wait_for_service_ipc(self).await?; } ServiceStatus::InstallRequired => { logging!(info, Type::Service, "需要安装服务，执行安装流程"); - install_service()?; - wait_and_check_service_version(self).await?; + run_service_command(install_service, "install service")?; + wait_for_service_ipc(self).await?; + if clash_verge_service_ipc::is_reinstall_service_needed().await { + logging!(info, Type::Service, "服务版本不匹配，执行重装流程"); + self.set_status(ServiceStatus::NeedsReinstall); + run_service_command(reinstall_service, "reinstall service")?; + wait_for_service_ipc(self).await?; + } } ServiceStatus::UninstallRequired => { logging!(info, Type::Service, "服务需要卸载，执行卸载流程"); - uninstall_service()?; - self.0 = ServiceStatus::Unavailable("Service Uninstalled".into()); + run_service_command(uninstall_service, "uninstall service")?; + self.set_status(ServiceStatus::Unavailable("Service Uninstalled".into())); } ServiceStatus::Unavailable(reason) => { logging!(info, Type::Service, "服务不可用: {}，将使用Sidecar模式", reason); - self.0 = ServiceStatus::Unavailable(reason.clone()); - return Err(anyhow::anyhow!("服务不可用: {}", reason)); + bail!("服务不可用: {}", reason); } } - // 防止服务安装成功后，内核未完全启动导致系统托盘无法获取代理节点信息 - Tray::global().update_menu().await?; Ok(()) } } -pub static SERVICE_MANAGER: Lazy> = Lazy::new(|| Mutex::new(ServiceManager::default())); +fn run_service_command(operation: impl FnOnce() -> Result<()>, label: &'static str) -> Result<()> { + tokio::task::block_in_place(operation).with_context(|| format!("{label} failed")) +} + +pub static SERVICE_MANAGER: Lazy = Lazy::new(|| ServiceManager { + status: Mutex::new(ServiceStatus::Unavailable("Need Checks".into())), + operation_running: AtomicBool::new(false), + operation_done: Notify::new(), +}); diff --git a/src-tauri/src/utils/resolve/mod.rs b/src-tauri/src/utils/resolve/mod.rs index f40042f0d..7aac2a551 100644 --- a/src-tauri/src/utils/resolve/mod.rs +++ b/src-tauri/src/utils/resolve/mod.rs @@ -175,14 +175,9 @@ pub(super) async fn init_verge_config() { pub(super) async fn init_service_manager() { clash_verge_service_ipc::set_config(Some(ServiceManager::config())).await; - if !is_service_ipc_path_exists() { - return; + if is_service_ipc_path_exists() && SERVICE_MANAGER.init().await.is_ok() { + logging_error!(Type::Setup, SERVICE_MANAGER.refresh().await); } - let mut manager = SERVICE_MANAGER.lock().await; - if manager.init().await.is_ok() { - logging_error!(Type::Setup, manager.refresh().await); - } - drop(manager); } pub(super) async fn init_core_manager() {