diff --git a/build_support/localization.rs b/build_support/localization.rs index 6e3e62f..679c4ab 100644 --- a/build_support/localization.rs +++ b/build_support/localization.rs @@ -118,6 +118,10 @@ const MENU_STATUS_HELP_KEYS: &[(&str, &str)] = &[ "IDM_SERVICE_CONTINUE", "crate::ui::resource_ids::IDM_SERVICE_CONTINUE", ), + ( + "IDM_SERVICE_RESTART", + "crate::ui::resource_ids::IDM_SERVICE_RESTART", + ), ( "IDM_SERVICE_GOTOPROCESS", "crate::ui::resource_ids::IDM_SERVICE_GOTOPROCESS", diff --git a/localization/de.toml b/localization/de.toml index 5bfd680..c2fe8cc 100644 --- a/localization/de.toml +++ b/localization/de.toml @@ -330,6 +330,7 @@ ServicesPageTitle = "Dienste" ServicesMenu = "&Dienste" ServiceStart = "&Starten" ServiceStop = "S&toppen" +ServiceRestart = "&Neu starten" ServicePause = "&Anhalten" ServiceContinue = "&Fortsetzen" ServiceColumnName = "Name" @@ -349,6 +350,7 @@ ServiceStartupManual = "Manuell" ServiceStartupDisabled = "Deaktiviert" ServiceStartFailed = "Der Dienst konnte nicht gestartet werden." ServiceStopFailed = "Der Dienst konnte nicht beendet werden." +ServiceRestartFailed = "Der Dienst konnte nicht neu gestartet werden." ServicePauseFailed = "Der Dienst konnte nicht angehalten werden." ServiceContinueFailed = "Der Dienst konnte nicht fortgesetzt werden." ServiceStateUnchanged = "Der Dienst hat den angeforderten Zustand nicht erreicht." @@ -417,6 +419,7 @@ IDM_TASK_TILEVERT = "Tiles the windows vertically on the desktop" IDM_TASK_BRINGTOFRONT = "Brings the window front, but does not switch to it" IDM_SERVICE_START = "Startet den ausgewählten Dienst" IDM_SERVICE_STOP = "Beendet den ausgewählten Dienst" +IDM_SERVICE_RESTART = "Startet den ausgewählten Dienst neu" IDM_SERVICE_PAUSE = "Hält den ausgewählten Dienst an" IDM_SERVICE_CONTINUE = "Setzt den ausgewählten Dienst fort" IDM_SERVICE_GOTOPROCESS = "Wählt den Prozess des Diensts auf der Seite Prozesse aus" diff --git a/localization/en_us.toml b/localization/en_us.toml index 6f4802a..69d8f59 100644 --- a/localization/en_us.toml +++ b/localization/en_us.toml @@ -337,6 +337,7 @@ ServicesPageTitle = "Services" ServicesMenu = "&Services" ServiceStart = "&Start" ServiceStop = "&Stop" +ServiceRestart = "&Restart" ServicePause = "&Pause" ServiceContinue = "&Continue" ServiceColumnName = "Name" @@ -356,6 +357,7 @@ ServiceStartupManual = "Manual" ServiceStartupDisabled = "Disabled" ServiceStartFailed = "Unable to start the service." ServiceStopFailed = "Unable to stop the service." +ServiceRestartFailed = "Unable to restart the service." ServicePauseFailed = "Unable to pause the service." ServiceContinueFailed = "Unable to continue the service." ServiceStateUnchanged = "The service did not enter the requested state." @@ -424,6 +426,7 @@ IDM_TASK_TILEVERT = "Tiles the windows vertically on the desktop" IDM_TASK_BRINGTOFRONT = "Brings the window front, but does not switch to it" IDM_SERVICE_START = "Starts the selected service" IDM_SERVICE_STOP = "Stops the selected service" +IDM_SERVICE_RESTART = "Restarts the selected service" IDM_SERVICE_PAUSE = "Pauses the selected service" IDM_SERVICE_CONTINUE = "Continues the selected service" IDM_SERVICE_GOTOPROCESS = "Selects the service process on the Processes page" diff --git a/localization/es.toml b/localization/es.toml index 35e19ea..e0fe1f5 100644 --- a/localization/es.toml +++ b/localization/es.toml @@ -330,6 +330,7 @@ ServicesPageTitle = "Servicios" ServicesMenu = "&Servicios" ServiceStart = "&Iniciar" ServiceStop = "&Detener" +ServiceRestart = "&Reiniciar" ServicePause = "&Pausar" ServiceContinue = "&Continuar" ServiceColumnName = "Nombre" @@ -349,6 +350,7 @@ ServiceStartupManual = "Manual" ServiceStartupDisabled = "Deshabilitado" ServiceStartFailed = "No se pudo iniciar el servicio." ServiceStopFailed = "No se pudo detener el servicio." +ServiceRestartFailed = "No se pudo reiniciar el servicio." ServicePauseFailed = "No se pudo pausar el servicio." ServiceContinueFailed = "No se pudo continuar el servicio." ServiceStateUnchanged = "El servicio no entró en el estado solicitado." @@ -417,6 +419,7 @@ IDM_TASK_TILEVERT = "Tiles the windows vertically on the desktop" IDM_TASK_BRINGTOFRONT = "Brings the window front, but does not switch to it" IDM_SERVICE_START = "Inicia el servicio seleccionado" IDM_SERVICE_STOP = "Detiene el servicio seleccionado" +IDM_SERVICE_RESTART = "Reinicia el servicio seleccionado" IDM_SERVICE_PAUSE = "Pausa el servicio seleccionado" IDM_SERVICE_CONTINUE = "Continúa el servicio seleccionado" IDM_SERVICE_GOTOPROCESS = "Selecciona el proceso del servicio en la página Procesos" diff --git a/localization/fr.toml b/localization/fr.toml index c9b8992..b646759 100644 --- a/localization/fr.toml +++ b/localization/fr.toml @@ -330,6 +330,7 @@ ServicesPageTitle = "Services" ServicesMenu = "&Services" ServiceStart = "&Démarrer" ServiceStop = "&Arrêter" +ServiceRestart = "&Redémarrer" ServicePause = "&Suspendre" ServiceContinue = "&Continuer" ServiceColumnName = "Nom" @@ -349,6 +350,7 @@ ServiceStartupManual = "Manuel" ServiceStartupDisabled = "Désactivé" ServiceStartFailed = "Impossible de démarrer le service." ServiceStopFailed = "Impossible d'arrêter le service." +ServiceRestartFailed = "Impossible de redémarrer le service." ServicePauseFailed = "Impossible de suspendre le service." ServiceContinueFailed = "Impossible de continuer le service." ServiceStateUnchanged = "Le service n'a pas atteint l'état demandé." @@ -417,6 +419,7 @@ IDM_TASK_TILEVERT = "Tiles the windows vertically on the desktop" IDM_TASK_BRINGTOFRONT = "Brings the window front, but does not switch to it" IDM_SERVICE_START = "Démarre le service sélectionné" IDM_SERVICE_STOP = "Arrête le service sélectionné" +IDM_SERVICE_RESTART = "Redémarre le service sélectionné" IDM_SERVICE_PAUSE = "Suspend le service sélectionné" IDM_SERVICE_CONTINUE = "Reprend le service sélectionné" IDM_SERVICE_GOTOPROCESS = "Sélectionne le processus du service dans la page Processus" diff --git a/localization/pt.toml b/localization/pt.toml index 66b2d54..79d9e47 100644 --- a/localization/pt.toml +++ b/localization/pt.toml @@ -330,6 +330,7 @@ ServicesPageTitle = "Serviços" ServicesMenu = "&Serviços" ServiceStart = "&Iniciar" ServiceStop = "&Parar" +ServiceRestart = "&Reiniciar" ServicePause = "&Pausar" ServiceContinue = "&Continuar" ServiceColumnName = "Nome" @@ -349,6 +350,7 @@ ServiceStartupManual = "Manual" ServiceStartupDisabled = "Desativado" ServiceStartFailed = "Não foi possível iniciar o serviço." ServiceStopFailed = "Não foi possível parar o serviço." +ServiceRestartFailed = "Não foi possível reiniciar o serviço." ServicePauseFailed = "Não foi possível pausar o serviço." ServiceContinueFailed = "Não foi possível continuar o serviço." ServiceStateUnchanged = "O serviço não entrou no estado solicitado." @@ -417,6 +419,7 @@ IDM_TASK_TILEVERT = "Tiles the windows vertically on the desktop" IDM_TASK_BRINGTOFRONT = "Brings the window front, but does not switch to it" IDM_SERVICE_START = "Inicia o serviço selecionado" IDM_SERVICE_STOP = "Para o serviço selecionado" +IDM_SERVICE_RESTART = "Reinicia o serviço selecionado" IDM_SERVICE_PAUSE = "Pausa o serviço selecionado" IDM_SERVICE_CONTINUE = "Continua o serviço selecionado" IDM_SERVICE_GOTOPROCESS = "Seleciona o processo do serviço na página Processos" diff --git a/localization/ru.toml b/localization/ru.toml index 9845df7..52f9a57 100644 --- a/localization/ru.toml +++ b/localization/ru.toml @@ -330,6 +330,7 @@ ServicesPageTitle = "Службы" ServicesMenu = "&Службы" ServiceStart = "&Запустить" ServiceStop = "&Остановить" +ServiceRestart = "Пере&запустить" ServicePause = "&Приостановить" ServiceContinue = "&Продолжить" ServiceColumnName = "Имя" @@ -349,6 +350,7 @@ ServiceStartupManual = "Вручную" ServiceStartupDisabled = "Отключена" ServiceStartFailed = "Не удалось запустить службу." ServiceStopFailed = "Не удалось остановить службу." +ServiceRestartFailed = "Не удалось перезапустить службу." ServicePauseFailed = "Не удалось приостановить службу." ServiceContinueFailed = "Не удалось возобновить службу." ServiceStateUnchanged = "Служба не перешла в запрошенное состояние." @@ -417,6 +419,7 @@ IDM_TASK_TILEVERT = "Tiles the windows vertically on the desktop" IDM_TASK_BRINGTOFRONT = "Brings the window front, but does not switch to it" IDM_SERVICE_START = "Запускает выбранную службу" IDM_SERVICE_STOP = "Останавливает выбранную службу" +IDM_SERVICE_RESTART = "Перезапускает выбранную службу" IDM_SERVICE_PAUSE = "Приостанавливает выбранную службу" IDM_SERVICE_CONTINUE = "Возобновляет выбранную службу" IDM_SERVICE_GOTOPROCESS = "Выбирает процесс службы на странице процессов" diff --git a/localization/zh_cn.toml b/localization/zh_cn.toml index 72ff3ad..18492a7 100644 --- a/localization/zh_cn.toml +++ b/localization/zh_cn.toml @@ -327,6 +327,7 @@ ServicesPageTitle = "服务" ServicesMenu = "服务(&S)" ServiceStart = "启动(&S)" ServiceStop = "停止(&T)" +ServiceRestart = "重新启动(&R)" ServicePause = "暂停(&P)" ServiceContinue = "继续(&C)" ServiceColumnName = "名称" @@ -346,6 +347,7 @@ ServiceStartupManual = "手动" ServiceStartupDisabled = "禁用" ServiceStartFailed = "无法启动该服务。" ServiceStopFailed = "无法停止该服务。" +ServiceRestartFailed = "无法重新启动该服务。" ServicePauseFailed = "无法暂停该服务。" ServiceContinueFailed = "无法继续该服务。" ServiceStateUnchanged = "该服务没有进入请求的状态。" @@ -414,6 +416,7 @@ IDM_TASK_TILEVERT = "Tiles the windows vertically on the desktop" IDM_TASK_BRINGTOFRONT = "Brings the window front, but does not switch to it" IDM_SERVICE_START = "启动所选服务" IDM_SERVICE_STOP = "停止所选服务" +IDM_SERVICE_RESTART = "重新启动所选服务" IDM_SERVICE_PAUSE = "暂停所选服务" IDM_SERVICE_CONTINUE = "继续所选服务" IDM_SERVICE_GOTOPROCESS = "在进程页选中该服务的进程" diff --git a/localization/zh_tw.toml b/localization/zh_tw.toml index c7dbd0a..1dbb14c 100644 --- a/localization/zh_tw.toml +++ b/localization/zh_tw.toml @@ -327,6 +327,7 @@ ServicesPageTitle = "服務" ServicesMenu = "服務(&S)" ServiceStart = "啟動(&S)" ServiceStop = "停止(&T)" +ServiceRestart = "重新啟動(&R)" ServicePause = "暫停(&P)" ServiceContinue = "繼續(&C)" ServiceColumnName = "名稱" @@ -346,6 +347,7 @@ ServiceStartupManual = "手動" ServiceStartupDisabled = "已停用" ServiceStartFailed = "無法啟動服務。" ServiceStopFailed = "無法停止服務。" +ServiceRestartFailed = "無法重新啟動服務。" ServicePauseFailed = "無法暫停服務。" ServiceContinueFailed = "無法繼續服務。" ServiceStateUnchanged = "服務沒有進入要求的狀態。" @@ -414,6 +416,7 @@ IDM_TASK_TILEVERT = "Tiles the windows vertically on the desktop" IDM_TASK_BRINGTOFRONT = "Brings the window front, but does not switch to it" IDM_SERVICE_START = "啟動選取的服務" IDM_SERVICE_STOP = "停止選取的服務" +IDM_SERVICE_RESTART = "重新啟動選取的服務" IDM_SERVICE_PAUSE = "暫停選取的服務" IDM_SERVICE_CONTINUE = "繼續選取的服務" IDM_SERVICE_GOTOPROCESS = "在處理程序頁選取該服務的處理程序" diff --git a/src/app/mod.rs b/src/app/mod.rs index 0e65c3f..d23243b 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -1180,6 +1180,7 @@ impl App { .unwrap_or_default(); self.enable_menu_item(menu, IDM_SERVICE_START, commands.start); self.enable_menu_item(menu, IDM_SERVICE_STOP, commands.stop); + self.enable_menu_item(menu, IDM_SERVICE_RESTART, commands.restart); self.enable_menu_item(menu, IDM_SERVICE_PAUSE, commands.pause); self.enable_menu_item(menu, IDM_SERVICE_CONTINUE, commands.continue_service); self.enable_menu_item(menu, IDM_SERVICE_GOTOPROCESS, commands.go_to_process); @@ -2306,6 +2307,7 @@ impl App { } IDM_SERVICE_START | IDM_SERVICE_STOP + | IDM_SERVICE_RESTART | IDM_SERVICE_PAUSE | IDM_SERVICE_CONTINUE | IDM_SERVICE_GOTOPROCESS => { diff --git a/src/infrastructure/native/process_architecture.rs b/src/infrastructure/native/process_architecture.rs index 241ced3..a4804ab 100644 --- a/src/infrastructure/native/process_architecture.rs +++ b/src/infrastructure/native/process_architecture.rs @@ -111,10 +111,10 @@ fn classify_machines( let process_machine = match reported_machine { Some(machine) => machine, None if wow_machine != IMAGE_FILE_MACHINE_UNKNOWN => wow_machine, - // On ARM64, both native ARM64 and x86-64/ARM64EC can report UNKNOWN. The old API - // alone cannot distinguish them, including on Windows 10 preview builds. - None if native_machine == IMAGE_FILE_MACHINE_ARM64 => return Err(ERROR_NOT_SUPPORTED), - None => native_machine, + // IsWow64Process2 reports IMAGE_FILE_MACHINE_UNKNOWN for every non-WOW64 + // process. That value was not observed. On ARM64 it cannot separate native + // ARM64 from x64/ARM64EC, and on every other host it is still not native. + None => return Err(ERROR_NOT_SUPPORTED), }; let process_is_32_bit = machine_is_32_bit(process_machine)?; let mode = match (native_machine, process_machine) { @@ -417,11 +417,17 @@ mod tests { } #[test] - fn old_api_keeps_known_bitness_but_never_guesses_arm64_native() { - assert_eq!( - classify_machines(0, IMAGE_FILE_MACHINE_ARM64, None), - Err(ERROR_NOT_SUPPORTED) - ); + fn unknown_process_machine_is_an_error_and_reported_native_stays_native() { + for host in [ + IMAGE_FILE_MACHINE_ARM64, + IMAGE_FILE_MACHINE_AMD64, + IMAGE_FILE_MACHINE_I386, + ] { + assert_eq!( + classify_machines(IMAGE_FILE_MACHINE_UNKNOWN, host, None), + Err(ERROR_NOT_SUPPORTED) + ); + } assert_eq!( classify_machines(IMAGE_FILE_MACHINE_I386, IMAGE_FILE_MACHINE_ARM64, None) .unwrap() @@ -429,19 +435,43 @@ mod tests { ExecutionMode::EmulatedX86 ); assert_eq!( - classify_machines(0, IMAGE_FILE_MACHINE_I386, None) + classify_machines(IMAGE_FILE_MACHINE_I386, IMAGE_FILE_MACHINE_AMD64, None) .unwrap() .mode, - ExecutionMode::Native + ExecutionMode::Compatibility32 ); assert_eq!( - classify_machines(IMAGE_FILE_MACHINE_I386, IMAGE_FILE_MACHINE_AMD64, None) + classify_machines(IMAGE_FILE_MACHINE_AMD64, IMAGE_FILE_MACHINE_AMD64, None) .unwrap() .mode, - ExecutionMode::Compatibility32 + ExecutionMode::Native + ); + assert_eq!( + classify_machines( + IMAGE_FILE_MACHINE_UNKNOWN, + IMAGE_FILE_MACHINE_AMD64, + Some(IMAGE_FILE_MACHINE_AMD64), + ) + .unwrap() + .mode, + ExecutionMode::Native ); assert_eq!( - classify_machines(0, IMAGE_FILE_MACHINE_ARM64, Some(0)), + classify_machines( + IMAGE_FILE_MACHINE_UNKNOWN, + IMAGE_FILE_MACHINE_ARM64, + Some(IMAGE_FILE_MACHINE_ARM64), + ) + .unwrap() + .mode, + ExecutionMode::Native + ); + assert_eq!( + classify_machines( + IMAGE_FILE_MACHINE_UNKNOWN, + IMAGE_FILE_MACHINE_ARM64, + Some(0) + ), Err(ERROR_INVALID_DATA) ); assert_eq!( diff --git a/src/infrastructure/worker.rs b/src/infrastructure/worker.rs index 7b40342..0693959 100644 --- a/src/infrastructure/worker.rs +++ b/src/infrastructure/worker.rs @@ -457,6 +457,18 @@ pub(crate) struct WorkerDrain { type WorkerSpawner = Box Result, u32> + Send + Sync>; +type ExitDrain = Box; +static EXIT_DRAINS: Mutex> = Mutex::new(Vec::new()); + +/// Called after the UI loop and all windows have gone, but before diagnostics/process exit. +/// Accepted writes own their requests here and must finish before process termination. +pub(crate) fn drain_before_process_exit() { + let drains = std::mem::take(&mut *EXIT_DRAINS.lock().unwrap_or_else(|e| e.into_inner())); + for drain in drains { + drain(); + } +} + pub(crate) struct SingleFlightWorker { worker: Option>, restart: WorkerSpawner, @@ -756,6 +768,65 @@ where self.pending.is_some() } + /// Only for final page teardown of a writer: close its notification gate immediately, + /// retain the current request and coalesced commands, then finish after UI teardown. + /// Unlike cancellable sampling, accepted mutations must not be discarded on exit. + pub(crate) fn defer_drain_until_exit(self) { + let drain = self.into_exit_drain(); + EXIT_DRAINS + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(drain); + } + + fn into_exit_drain(mut self) -> ExitDrain { + if let Some(worker) = self.worker.as_mut() { + worker.stop.close_notifications(false); + worker.retire_async = false; + } + Box::new(move || { + loop { + if self.in_flight { + let event = self + .worker + .as_ref() + .and_then(|worker| worker.completion_receiver.as_ref()) + .and_then(|receiver| receiver.recv().ok()); + self.in_flight = false; + match event { + Some(WorkerEvent::Completed(_)) => {} + Some(WorkerEvent::CollectorPanicked { .. }) => { + record_win32_error("exit drain collector panic", ERROR_GEN_FAILURE); + } + None => { + record_win32_error("exit drain disconnected", ERROR_BROKEN_PIPE); + self.worker = None; + } + } + } + let Some(request) = self.pending.take() else { + break; + }; + let operation_id = self + .pending_operation_id + .take() + .unwrap_or_else(diagnostics::next_operation_id); + // No HWND survives into this phase, including when a worker must restart. + if let Err(error) = self.submit_request(request, std::ptr::null_mut(), operation_id) + { + record_win32_error("exit drain pending submission", error.code); + break; + } + self.in_flight = true; + } + // A restarted worker also joins here, before diagnostics and the process exit. + if let Some(worker) = self.worker.as_mut() { + worker.retire_async = false; + worker.stop.close_notifications(false); + } + }) + } + #[cfg(test)] fn disconnect(&self) { self.worker @@ -791,6 +862,49 @@ mod tests { }; use std::time::{Duration, Instant}; + #[test] + fn exit_drain_preserves_unstarted_and_pending_writes_without_waiting_on_ui() { + let (entered_tx, entered_rx) = bounded(1); + let (release_tx, release_rx) = bounded(1); + let (written_tx, written_rx) = bounded(3); + let mut worker = SingleFlightWorker::spawn_cancellable_initialized( + "taskmgr-rs-exit-drain-test", + 0, + |pending: &mut Vec, incoming| pending.extend(incoming), + move |stop| { + entered_tx.send(stop).unwrap(); + release_rx.recv().unwrap(); + let written_tx = written_tx.clone(); + move |commands: Vec| { + for value in commands { + written_tx.send(value).unwrap(); + } + } + }, + ) + .unwrap(); + worker.request(vec![1], null_mut()).unwrap(); + let stop = entered_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + worker.request(vec![2], null_mut()).unwrap(); + worker.request(vec![3], null_mut()).unwrap(); + let started = Instant::now(); + worker.defer_drain_until_exit(); + assert!(started.elapsed() < Duration::from_secs(1)); + assert!(!stop.is_cancelled()); + assert!(*stop.notifications_closed.lock().unwrap()); + + let (finished_tx, finished_rx) = bounded(1); + let drain = thread::spawn(move || { + drain_before_process_exit(); + finished_tx.send(()).unwrap(); + }); + assert!(finished_rx.recv_timeout(Duration::from_millis(50)).is_err()); + release_tx.send(()).unwrap(); + finished_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + drain.join().unwrap(); + assert_eq!(written_rx.try_iter().collect::>(), vec![1, 2, 3]); + } + #[test] fn worker_delivers_completion_even_when_notification_window_is_gone() { let worker = BackgroundWorker::spawn("taskmgr-rs-worker-test", 0, |value: u32| value * 2) @@ -912,7 +1026,14 @@ mod tests { .expect("follow-up request should coalesce"); assert!(worker.has_pending()); - let deadline = Instant::now() + Duration::from_secs(2); + // The collector publishes `CollectorPanicked` only after the panic hook + // flushes diagnostics and the default hook captures a backtrace. CI sets + // `RUST_BACKTRACE=1`, so a loaded Windows runner needs this longer poll + // before that report and the reinitialized follow-up can be observed. + const DRAIN_BUDGET: Duration = Duration::from_secs(30); + const POLL_INTERVAL: Duration = Duration::from_millis(20); + + let deadline = Instant::now() + DRAIN_BUDGET; loop { let drained = worker.drain(null_mut()); if drained.error == Some(ERROR_GEN_FAILURE) { @@ -922,10 +1043,10 @@ mod tests { break; } assert!(Instant::now() < deadline, "panic report timed out"); - thread::yield_now(); + thread::sleep(POLL_INTERVAL); } - let deadline = Instant::now() + Duration::from_secs(2); + let deadline = Instant::now() + DRAIN_BUDGET; loop { let drained = worker.drain(null_mut()); assert_eq!(drained.error, None); @@ -939,7 +1060,7 @@ mod tests { Instant::now() < deadline, "reinitialized collector timed out" ); - thread::yield_now(); + thread::sleep(POLL_INTERVAL); } } diff --git a/src/main.rs b/src/main.rs index d5b7b36..6953245 100644 --- a/src/main.rs +++ b/src/main.rs @@ -29,6 +29,7 @@ fn main() { std::process::exit(exit_code); } let exit_code = app::run(); + infrastructure::worker::drain_before_process_exit(); infrastructure::diagnostics::shutdown(); std::process::exit(exit_code); } diff --git a/src/pages/applications/mod.rs b/src/pages/applications/mod.rs index 229ab57..f2a62e5 100644 --- a/src/pages/applications/mod.rs +++ b/src/pages/applications/mod.rs @@ -27,6 +27,7 @@ use std::collections::{HashMap, HashSet}; // // 线程模型： // 一个单飞 worker 枚举窗口，另一个单飞协调器驱动固定数量的长期图标线程。 +// 结束任务使用第三条长期单飞 worker：UI 线程只提交窗口身份，EndTask 在后台执行。 // UI 线程只提交请求和应用完整结果，采集失败不会覆盖上一轮可信列表。 // // 缓存失效策略： @@ -217,6 +218,7 @@ pub struct TaskPageState { pass_count: u64, snapshot_worker: Option>, icon_worker: Option>, + end_task_worker: Option, Option>>, pending_icon_identities: HashSet, icon_generation: u64, last_refresh_error: Option, @@ -243,6 +245,7 @@ impl Default for TaskPageState { pass_count: 0, snapshot_worker: None, icon_worker: None, + end_task_worker: None, pending_icon_identities: HashSet::new(), icon_generation: 0, last_refresh_error: None, @@ -484,26 +487,7 @@ impl TaskPageState { } } id if id == IDM_TASK_ENDTASK || id == IDC_ENDTASK as u16 => { - let force = (GetKeyState(i32::from(VK_CONTROL)) & (1 << 15)) != 0; - let mut first_error = None; - for hwnd in self.selected_hwnds(true) { - if EndTask(hwnd, 0, if force { 1 } else { 0 }) == 0 { - let error = last_error_or_gen_failure(); - record_win32_error("ending task", error); - first_error.get_or_insert(error); - } - } - if let Some(error) = first_error { - let title = to_wide_null(text(TextKey::WarningTitle)); - let body = - to_wide_null(&format!("{} {error}", text(TextKey::Win32ErrorPrefix))); - MessageBoxW( - self.hwnd_page, - body.as_ptr(), - title.as_ptr(), - MB_OK | MB_ICONERROR, - ); - } + self.request_end_task(); } IDM_TASK_FINDPROCESS => { if let Some(identity) = self.selected_task_identities(true).first().copied() @@ -1003,6 +987,65 @@ impl TaskPageState { pub fn handle_worker_completion(&mut self) { self.drain_worker_results(); self.drain_icon_worker_results(); + self.drain_end_task_results(); + } + + fn request_end_task(&mut self) { + // Ctrl is part of the click. GetKeyState reports this UI thread's message state, so the + // worker receives the flag instead of sampling the keyboard itself. + let force = unsafe { (GetKeyState(i32::from(VK_CONTROL)) & (1 << 15)) != 0 }; + let targets = self.selected_task_identities(true); + if targets.is_empty() { + return; + } + let request = vec![EndTaskRequest { targets, force }]; + let Some(worker) = self.end_task_worker.as_mut() else { + self.report_end_task_failure(windows_sys::Win32::Foundation::ERROR_BROKEN_PIPE); + return; + }; + if let Err(error) = worker.request(request, self.hwnd_page) { + self.report_end_task_failure(error); + } + } + + fn drain_end_task_results(&mut self) { + let drained = match self.end_task_worker.as_mut() { + Some(worker) => worker.drain(self.hwnd_page), + None => return, + }; + let mut end_task_error = None; + for completion in drained.completions { + if end_task_error.is_none() { + end_task_error = completion.value; + } + } + if let Some(error) = drained.error { + record_win32_error("ending task", error); + if end_task_error.is_none() { + end_task_error = Some(error); + } + } + if let Some(error) = end_task_error { + self.show_end_task_warning(error); + } + } + + fn report_end_task_failure(&self, error: u32) { + record_win32_error("ending task", error); + self.show_end_task_warning(error); + } + + fn show_end_task_warning(&self, error: u32) { + let title = to_wide_null(text(TextKey::WarningTitle)); + let body = to_wide_null(&format!("{} {error}", text(TextKey::Win32ErrorPrefix))); + unsafe { + MessageBoxW( + self.hwnd_page, + body.as_ptr(), + title.as_ptr(), + MB_OK | MB_ICONERROR, + ); + } } fn resort_tasks(&mut self) { @@ -1014,7 +1057,10 @@ impl TaskPageState { fn start_worker_thread(&mut self) -> Result<(), u32> { // 顶层窗口枚举可能涉及跨窗口站和桌面切换， // 放到后台线程可以避免主线程在刷新时明显卡顿。 - if self.snapshot_worker.is_some() && self.icon_worker.is_some() { + if self.snapshot_worker.is_some() + && self.icon_worker.is_some() + && self.end_task_worker.is_some() + { return Ok(()); } @@ -1043,12 +1089,23 @@ impl TaskPageState { }, )?); } + if self.end_task_worker.is_none() { + // EndTask can block inside the target window's queue. Retire this worker without + // joining it on the UI thread; a destroyed page drops the completion instead. + self.end_task_worker = Some(SingleFlightWorker::spawn_cancellable_initialized( + "taskmgr-rs-task-end", + PWM_TASK_WORKER_COMPLETE, + queue_end_tasks, + |_stop| run_end_tasks, + )?); + } Ok(()) } fn stop_worker_thread(&mut self) { self.snapshot_worker = None; self.icon_worker = None; + self.end_task_worker = None; self.pending_icon_identities.clear(); } @@ -1481,6 +1538,74 @@ fn last_error_or_gen_failure() -> u32 { } } +struct EndTaskRequest { + targets: Vec, + force: bool, +} + +fn queue_end_tasks(pending: &mut Vec, incoming: Vec) { + // An accepted End Task is not a refresh sample. Append it; do not replace an earlier command. + pending.extend(incoming); +} + +fn run_end_tasks(requests: Vec) -> Option { + run_end_task_queue(&requests, window_matches_identity, |hwnd, force| { + // SAFETY: `window_matches_identity` just accepted this HWND on this thread. EndTask + // does not retain the handle after it returns. + let ended = unsafe { EndTask(hwnd, 0, if force { 1 } else { 0 }) } != 0; + if ended { + Ok(()) + } else { + Err(last_error_or_gen_failure()) + } + }) +} + +fn run_end_task_queue( + requests: &[EndTaskRequest], + mut matches: Match, + mut end_task: End, +) -> Option +where + Match: FnMut(TaskIdentity) -> bool, + End: FnMut(HWND, bool) -> Result<(), u32>, +{ + let mut first_error = None; + for request in requests { + let error = + end_matching_tasks(&request.targets, request.force, &mut matches, &mut end_task); + if first_error.is_none() { + first_error = error; + } + } + first_error +} + +fn end_matching_tasks( + targets: &[TaskIdentity], + force: bool, + matches: &mut Match, + end_task: &mut End, +) -> Option +where + Match: FnMut(TaskIdentity) -> bool, + End: FnMut(HWND, bool) -> Result<(), u32>, +{ + let mut first_error = None; + for identity in targets.iter().copied() { + // Re-check HWND, thread id, and ProcIdentity (pid + creation time) immediately before + // EndTask. An unknown or reused window is skipped and is not reported as success. + if !matches(identity) { + continue; + } + if let Err(error) = end_task(identity.hwnd(), force) { + record_win32_error("ending task", error); + first_error.get_or_insert(error); + } + } + first_error +} + #[cfg(test)] mod tests { use super::*; @@ -1597,6 +1722,85 @@ mod tests { assert!(store.free_slots.is_empty()); } + #[test] + fn end_task_skips_a_mismatched_window_and_reports_only_the_first_failure() { + let skipped = TaskIdentity { + hwnd: 1, + process: ProcIdentity::pid_only(10), + thread_id: 2, + }; + let failed = TaskIdentity { + hwnd: 3, + process: ProcIdentity::new(11, 100), + thread_id: 4, + }; + let succeeded = TaskIdentity { + hwnd: 5, + process: ProcIdentity::new(12, 200), + thread_id: 6, + }; + let failed_again = TaskIdentity { + hwnd: 7, + process: ProcIdentity::new(13, 300), + thread_id: 8, + }; + let mut calls = Vec::new(); + let error = end_matching_tasks( + &[skipped, failed, succeeded, failed_again], + true, + &mut |identity: TaskIdentity| identity.process.is_verified(), + &mut |hwnd, force| { + calls.push((hwnd as isize, force)); + if hwnd as isize == failed.hwnd { + Err(5) + } else if hwnd as isize == failed_again.hwnd { + Err(9) + } else { + Ok(()) + } + }, + ); + assert_eq!( + calls, + vec![ + (failed.hwnd, true), + (succeeded.hwnd, true), + (failed_again.hwnd, true) + ] + ); + assert_eq!(error, Some(5)); + } + + #[test] + fn pending_end_task_commands_run_in_submission_order() { + let command = |hwnd: isize, force: bool| EndTaskRequest { + targets: vec![TaskIdentity { + hwnd, + process: ProcIdentity::new(10, 100), + thread_id: 2, + }], + force, + }; + let mut pending = vec![command(1, false)]; + queue_end_tasks(&mut pending, vec![command(2, true)]); + queue_end_tasks(&mut pending, vec![command(3, false)]); + + let mut calls = Vec::new(); + let error = run_end_task_queue(&pending, |_| true, &mut |hwnd, force| { + calls.push((hwnd as isize, force)); + if hwnd as isize == 2 { + Err(5) + } else if hwnd as isize == 3 { + Err(9) + } else { + Ok(()) + } + }); + + assert_eq!(calls, vec![(1, false), (2, true), (3, false)]); + assert_eq!(error, Some(5)); + } + #[test] fn unverified_process_identity_is_never_actionable() { let identity = TaskIdentity { diff --git a/src/pages/cpu/mod.rs b/src/pages/cpu/mod.rs index 7e4928c..a8b395d 100644 --- a/src/pages/cpu/mod.rs +++ b/src/pages/cpu/mod.rs @@ -555,6 +555,9 @@ impl CpuPageState { self.kernel_history.push(sample.kernel_usage); self.history_valid = true; self.graph_scroll_offset = (self.graph_scroll_offset + 2) % GRAPH_GRID; + } else if self.cpu_usage.is_some() { + // Keep the previous percent, but do not present it as a current measurement. + self.system_sample_stale = true; } match sample.cpu_diagnostics { Ok(diagnostics) => { @@ -1692,7 +1695,13 @@ fn last_error_or_gen_failure() -> u32 { #[cfg(test)] mod tests { + use std::sync::Arc; + + use windows_sys::Win32::System::Threading::{ALL_PROCESSOR_GROUPS, GetActiveProcessorCount}; + use super::*; + use crate::system::cpu_topology::query_processor_topology; + use crate::system::disk::DiskActivityUpdate; #[test] fn queued_refresh_keeps_the_strongest_reason() { @@ -1771,4 +1780,87 @@ mod tests { assert!(component.is_stale()); assert_eq!(component.value, Some(42)); } + + #[test] + fn invalid_cpu_delta_marks_retained_usage_stale_until_measured_zero() { + let processor_count = unsafe { GetActiveProcessorCount(ALL_PROCESSOR_GROUPS) } as usize; + assert!(processor_count > 0); + let topology = Arc::new(query_processor_topology(processor_count)); + assert_eq!(topology.error(), None); + assert!(topology.identity().is_some()); + + let mut page = CpuPageState::new(); + let mut sample = system_sample(Arc::clone(&topology), false, 0, 0); + page.apply_system_sample(&sample, false).unwrap(); + assert_eq!(page.cpu_usage, None); + assert!(!page.history_valid); + assert_eq!(page.graph_scroll_offset, 0); + assert!(!page.system_sample_stale); + + sample = system_sample(Arc::clone(&topology), true, 40, 10); + page.apply_system_sample(&sample, false).unwrap(); + assert_eq!(page.cpu_usage, Some(40)); + assert_eq!(page.total_history.value_at(0), Some(40)); + assert_eq!(page.kernel_history.value_at(0), Some(10)); + assert!(page.history_valid); + assert_eq!(page.graph_scroll_offset, 2); + assert!(!page.system_sample_stale); + + sample = system_sample(topology, false, 0, 0); + page.apply_system_sample(&sample, false).unwrap(); + assert_eq!(page.cpu_usage, Some(40)); + assert_eq!(page.total_history.value_at(0), Some(40)); + assert_eq!(page.kernel_history.value_at(0), Some(10)); + assert_eq!(page.graph_scroll_offset, 2); + assert!(page.system_sample_stale); + assert_eq!(page.derive_status(), CpuPageStatus::Stale); + assert_eq!(page.status, CpuPageStatus::Stale); + + sample.cpu_delta_valid = true; + page.apply_system_sample(&sample, false).unwrap(); + assert_eq!(page.cpu_usage, Some(0)); + assert_eq!(page.total_history.value_at(0), Some(0)); + assert_eq!(page.total_history.value_at(1), Some(40)); + assert_eq!(page.kernel_history.value_at(0), Some(0)); + assert_eq!(page.kernel_history.value_at(1), Some(10)); + assert_eq!(page.graph_scroll_offset, 4); + assert!(!page.system_sample_stale); + assert_ne!(page.status, CpuPageStatus::Stale); + assert_ne!(page.derive_status(), CpuPageStatus::Stale); + } + + fn system_sample( + topology: Arc, + cpu_delta_valid: bool, + cpu_usage: u8, + kernel_usage: u8, + ) -> SystemSample { + let processor_count = topology.processors().len(); + SystemSample { + processor_count, + processor_topology: topology, + cpu_delta_valid, + cpu_usage, + kernel_usage, + processor_cpu_usage: vec![cpu_usage; processor_count], + processor_kernel_usage: vec![kernel_usage; processor_count], + cpu_diagnostics: Ok(CpuDiagnosticSample::default()), + uptime_ms: 1_000, + physical_mem_usage_kb: 0, + physical_mem_limit_kb: 0, + commit_total_kb: 0, + commit_limit_kb: 0, + commit_peak_kb: 0, + total_physical_kb: 0, + avail_physical_kb: 0, + file_cache_kb: 0, + kernel_total_kb: 0, + kernel_paged_kb: 0, + kernel_nonpaged_kb: 0, + handle_count: 0, + thread_count: 0, + process_count: 0, + disk: DiskActivityUpdate::NoRate, + } + } } diff --git a/src/pages/gpu/mod.rs b/src/pages/gpu/mod.rs index 6657b50..26557ea 100644 --- a/src/pages/gpu/mod.rs +++ b/src/pages/gpu/mod.rs @@ -308,14 +308,16 @@ impl AdapterHistory { .all(|engine_id| self.engine_histories.contains_key(engine_id)) ); - if let Some(usage) = sample.dedicated_usage_bytes { - self.dedicated_history - .push(memory_percentage(usage, sample.info.dedicated_limit_bytes)); - } - if let Some(usage) = sample.shared_usage_bytes { - self.shared_history - .push(memory_percentage(usage, sample.info.shared_limit_bytes)); - } + record_memory_history( + &mut self.dedicated_history, + sample.dedicated_usage_bytes, + sample.info.dedicated_limit_bytes, + ); + record_memory_history( + &mut self.shared_history, + sample.shared_usage_bytes, + sample.info.shared_limit_bytes, + ); } fn engine_options(&self) -> Vec<(GpuEngineId, GpuEngineKind)> { @@ -1685,6 +1687,19 @@ fn memory_percentage(usage: u64, limit: Option) -> u8 { rounded.min(100) as u8 } +fn record_memory_history(history: &mut HistoryBuffer, usage: Option, limit: Option) { + // A missing or zero limit makes memory_percentage return 0. That clamp is not a measured + // empty GPU, so this sample must not advance the graph. A later sample with a non-zero + // limit and real 0% usage still advances. + let Some(usage) = usage else { + return; + }; + if limit.filter(|limit| *limit != 0).is_none() { + return; + } + history.push(memory_percentage(usage, limit)); +} + fn status_after_refresh_error(has_snapshot: bool, unsupported: bool) -> GpuPageStatus { if has_snapshot { GpuPageStatus::Stale @@ -1964,6 +1979,65 @@ mod tests { assert_eq!(memory_percentage(1, Some(0)), 0); } + fn adapter_with_limits( + dedicated_limit_bytes: Option, + shared_limit_bytes: Option, + ) -> Arc { + let mut info = (*staged_adapter()).clone(); + info.dedicated_limit_bytes = dedicated_limit_bytes; + info.shared_limit_bytes = shared_limit_bytes; + Arc::new(info) + } + + fn memory_sample( + info: &Arc, + dedicated_usage_bytes: Option, + shared_usage_bytes: Option, + ) -> GpuAdapterSample { + GpuAdapterSample { + info: Arc::clone(info), + overall_utilization_percent: None, + engines: Vec::new(), + dedicated_usage_bytes, + shared_usage_bytes, + temperature_deci_c: None, + row_errors: Vec::new(), + } + } + + fn history_values(history: &HistoryBuffer) -> Vec { + (0..history.len()) + .map(|index| history.value_at(index).unwrap()) + .collect() + } + + #[test] + fn unknown_memory_limit_does_not_advance_history() { + let mut history = AdapterHistory::new(); + let limited = adapter_with_limits(Some(1_000), Some(2_000)); + history.push(&memory_sample(&limited, Some(500), Some(500))); + assert_eq!(history.dedicated_history.newest_value(), 50); + assert_eq!(history.shared_history.newest_value(), 25); + let dedicated_before = history_values(&history.dedicated_history); + let shared_before = history_values(&history.shared_history); + + let missing_limit = adapter_with_limits(None, None); + history.push(&memory_sample(&missing_limit, Some(1), Some(1))); + assert_eq!(history_values(&history.dedicated_history), dedicated_before); + assert_eq!(history_values(&history.shared_history), shared_before); + + let zero_limit = adapter_with_limits(Some(0), Some(0)); + history.push(&memory_sample(&zero_limit, Some(1), Some(1))); + assert_eq!(history_values(&history.dedicated_history), dedicated_before); + assert_eq!(history_values(&history.shared_history), shared_before); + + history.push(&memory_sample(&limited, Some(0), Some(0))); + assert_eq!(history.dedicated_history.newest_value(), 0); + assert_eq!(history.shared_history.newest_value(), 0); + assert_eq!(history.dedicated_history.value_at(1), Some(50)); + assert_eq!(history.shared_history.value_at(1), Some(25)); + } + #[test] fn refresh_errors_only_claim_a_stale_sample_when_one_exists() { assert_eq!( diff --git a/src/pages/network.rs b/src/pages/network.rs index 34c3f75..0cec896 100644 --- a/src/pages/network.rs +++ b/src/pages/network.rs @@ -1051,6 +1051,9 @@ impl NetworkPageState { (titles[2], 90, LVCFMT_RIGHT), (titles[3], 120, LVCFMT_LEFT), (titles[4], 140, LVCFMT_LEFT), + (titles[5], 110, LVCFMT_RIGHT), + (titles[6], 120, LVCFMT_RIGHT), + (titles[7], 110, LVCFMT_RIGHT), ]; for (index, (title, width, fmt)) in columns.iter().enumerate() { @@ -1406,13 +1409,16 @@ fn push_history(history: &mut HistoryBuffer, value: Option) { } } -fn adapter_row_texts(adapter: &NetworkAdapterEntry) -> [&str; 5] { +fn adapter_row_texts(adapter: &NetworkAdapterEntry) -> [&str; 8] { [ &adapter.name, &adapter.utilization, &adapter.link_speed, &adapter.state, &adapter.ipv4, + &adapter.bytes_sent, + &adapter.bytes_received, + &adapter.bytes_total, ] } @@ -2064,7 +2070,20 @@ mod tests { let adapter = test_adapter("Renamed Ethernet"); let columns = adapter_row_texts(&adapter); assert_eq!(columns[0], "Renamed Ethernet"); + assert_eq!(columns[2], "1.0 Gbps"); assert_eq!(columns[4], "192.168.0.2"); + assert_eq!(columns[5], "10"); + assert_eq!(columns[6], "20"); + assert_eq!(columns[7], "30"); + + let mut failed = adapter; + failed.bytes_sent = "-".to_string(); + failed.bytes_received = "-".to_string(); + failed.bytes_total = "-".to_string(); + let failed_columns = adapter_row_texts(&failed); + assert_eq!(failed_columns[5], "-"); + assert_eq!(failed_columns[6], "-"); + assert_eq!(failed_columns[7], "-"); } #[test] diff --git a/src/pages/performance/mod.rs b/src/pages/performance/mod.rs index 828b36a..0cde4ff 100644 --- a/src/pages/performance/mod.rs +++ b/src/pages/performance/mod.rs @@ -372,14 +372,14 @@ impl PerformancePageState { self.averaged_cpu_history.push(sample.cpu_usage); self.averaged_kernel_history.push(sample.kernel_usage); } - let mem_percent = if sample.physical_mem_limit_kb == 0 { - 0 - } else { - ((u128::from(sample.physical_mem_usage_kb) * 100) + // A missing physical limit is not an empty machine. Leave the previous + // history point in place so the graph does not paint that sample as 0%. + if sample.physical_mem_limit_kb != 0 { + let mem_percent = ((u128::from(sample.physical_mem_usage_kb) * 100) / u128::from(sample.physical_mem_limit_kb)) - .min(100) as u8 - }; - self.mem_history.push(mem_percent); + .min(100) as u8; + self.mem_history.push(mem_percent); + } let disk_label_changed = apply_disk_activity( &mut self.disk_history, &mut self.disk_known, @@ -833,11 +833,18 @@ impl PerformancePageState { pub fn draw_mem_meter(&self, hdc: HDC, rect: RECT) { // 内存仪表显示的是已用内存字节量文本，而不是百分比文本。 - let mem_percent = (u128::from(self.physical_mem_usage_kb) * 100) - .checked_div(u128::from(self.physical_mem_limit_kb)) - .unwrap_or(0) + // 缺少物理上限时沿用“不可用”文案，避免把未知样本画成 0% 成功值。 + let (mem_usage_text, mem_percent) = if self.physical_mem_limit_kb == 0 { + (text(TextKey::NotAvailable).to_string(), 0) + } else { + let mem_percent = ((u128::from(self.physical_mem_usage_kb) * 100) + / u128::from(self.physical_mem_limit_kb)) .min(100) as u8; - let mem_usage_text = format_mem_meter_text(self.physical_mem_usage_kb); + ( + format_mem_meter_text(self.physical_mem_usage_kb), + mem_percent, + ) + }; if self.draw_strip_meter(hdc, rect, &mem_usage_text, mem_percent, 0) { return; } @@ -1850,6 +1857,61 @@ mod tests { assert_eq!(page.cpu_usage, 0); assert_eq!(page.kernel_usage, 0); } + + #[test] + fn unknown_memory_limit_keeps_history_and_real_zero_still_records() { + let mut page = PerformancePageState { + no_title: true, + ..Default::default() + }; + let mut sample = SystemSample { + processor_count: 1, + processor_topology: Arc::new(query_processor_topology(1)), + cpu_delta_valid: false, + cpu_usage: 0, + kernel_usage: 0, + processor_cpu_usage: vec![0], + processor_kernel_usage: vec![0], + cpu_diagnostics: Ok(CpuDiagnosticSample::default()), + uptime_ms: 1_000, + physical_mem_usage_kb: 40, + physical_mem_limit_kb: 100, + commit_total_kb: 0, + commit_limit_kb: 0, + commit_peak_kb: 0, + total_physical_kb: 100, + avail_physical_kb: 60, + file_cache_kb: 0, + kernel_total_kb: 0, + kernel_paged_kb: 0, + kernel_nonpaged_kb: 0, + handle_count: 0, + thread_count: 0, + process_count: 0, + disk: DiskActivityUpdate::NoRate, + }; + page.apply_system_sample(null_mut(), &sample, false) + .unwrap(); + assert_eq!(page.mem_history.value_at(0), Some(40)); + assert_eq!(page.mem_history.value_at(1), Some(0)); + + // Non-zero usage with a missing limit must not be stored as 0% or shift history. + sample.physical_mem_usage_kb = 80; + sample.physical_mem_limit_kb = 0; + page.apply_system_sample(null_mut(), &sample, false) + .unwrap(); + assert_eq!(page.physical_mem_limit_kb, 0); + assert_eq!(page.mem_history.value_at(0), Some(40)); + assert_eq!(page.mem_history.value_at(1), Some(0)); + + // A later measured empty machine still advances history with a real 0. + sample.physical_mem_usage_kb = 0; + sample.physical_mem_limit_kb = 100; + page.apply_system_sample(null_mut(), &sample, false) + .unwrap(); + assert_eq!(page.mem_history.value_at(0), Some(0)); + assert_eq!(page.mem_history.value_at(1), Some(40)); + } } fn apply_disk_activity( diff --git a/src/pages/processes/action_worker.rs b/src/pages/processes/action_worker.rs index be0e861..f7da789 100644 --- a/src/pages/processes/action_worker.rs +++ b/src/pages/processes/action_worker.rs @@ -1,6 +1,6 @@ // +------------------------------------------------------------------------- // -// taskmgr-rs - 进程结束与优先级 +// taskmgr-rs - 进程结束、优先级与亲和性 // // 文件: src/pages/processes/action_worker.rs // @@ -9,17 +9,20 @@ // 作者: OpenAI Codex // -------------------------------------------------------------------------- -//! End Process, End Process Tree, and Set Priority after the UI confirmation. +//! End Process, End Process Tree, Set Priority, and Set Affinity after the UI +//! confirmation. //! //! One persistent [`SingleFlightWorker`] owns every confirmed action. The page //! posts no extra window message: completion reuses `PWM_PROC_WORKER_COMPLETE` //! and is drained beside the process sampler. Dropping the worker retires it -//! without joining on the UI thread, including while a tree walk is still -//! inside the collector. +//! without joining on the UI thread, including while a tree walk or an affinity +//! update is still inside the collector. //! //! Each action re-opens its target with `open_process_for_identity` and requires //! both the pid and `creation_time_100ns`. An unknown or mismatched identity -//! returns before `TerminateProcess`, the tree walk, or `SetPriorityClass`. +//! returns before `TerminateProcess`, the tree walk, `SetPriorityClass`, +//! `SetProcessAffinityMask`, or `SetProcessDefaultCpuSets`. A group or mask that +//! the native `KAFFINITY` cannot express fails before those calls. #[cfg(test)] use std::sync::mpsc::{Receiver, Sender}; @@ -34,8 +37,8 @@ use windows_sys::Win32::System::Threading::{ use super::ProcPriority; use super::actions::{ - ProcessTreePrepareError, ProcessTreeTerminationOutcome, prepare_process_tree_termination, - terminate_prepared_process_tree, + ProcessTreePrepareError, ProcessTreeTerminationOutcome, apply_confirmed_affinity, + prepare_process_tree_termination, terminate_prepared_process_tree, }; use crate::infrastructure::native::OwnedHandle; use crate::infrastructure::worker::SingleFlightWorker; @@ -45,6 +48,16 @@ use crate::ui::resource_ids::PWM_PROC_WORKER_COMPLETE; pub(super) type ProcessActionWorker = SingleFlightWorker, Vec>; +/// Group-qualified affinity bits chosen in the dialog. +/// +/// `mask` is the full group mask. A later conversion that would keep only the +/// low bits of `KAFFINITY` is rejected instead. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct AffinityGroupMask { + pub(super) group: u16, + pub(super) mask: u64, +} + pub(super) enum ProcessAction { End(ProcIdentity), EndTree(ProcIdentity), @@ -52,6 +65,10 @@ pub(super) enum ProcessAction { identity: ProcIdentity, priority: ProcPriority, }, + SetAffinity { + identity: ProcIdentity, + masks: Vec, + }, #[cfg(test)] Block { entered: Sender<()>, @@ -72,6 +89,7 @@ pub(super) enum ProcessActionFeedback { None, CantKill(u32), CantChangePriority(u32), + CantSetAffinity(u32), /// `show_failure_message` with the existing tree-failure body. KillTreePrepare(u32), /// Existing warning dialog. This is not a full success. @@ -80,7 +98,8 @@ pub(super) enum ProcessActionFeedback { pub(super) fn spawn_process_action_worker() -> Result { // Asynchronous retirement closes completion posts and does not join a worker - // that is still inside TerminateProcess, Toolhelp, or SetPriorityClass. + // that is still inside TerminateProcess, Toolhelp, SetPriorityClass, or an + // affinity update. SingleFlightWorker::spawn_cancellable_initialized( "taskmgr-rs-process-actions", PWM_PROC_WORKER_COMPLETE, @@ -137,6 +156,7 @@ fn run_process_action(action: ProcessAction) -> ProcessActionReport { ProcessAction::SetPriority { identity, priority } => { execute_set_priority(identity, priority) } + ProcessAction::SetAffinity { identity, masks } => execute_set_affinity(identity, masks), #[cfg(test)] ProcessAction::Block { entered, release } => { let _ = entered.send(()); @@ -179,6 +199,21 @@ fn execute_end_tree(identity: ProcIdentity) -> ProcessActionReport { } } +fn execute_set_affinity( + identity: ProcIdentity, + masks: Vec, +) -> ProcessActionReport { + // Reject an unknown identity before `apply_confirmed_affinity` opens the + // process and reaches either affinity API. + if let Err(error) = require_action_identity(identity) { + return affinity_failure(error); + } + match apply_confirmed_affinity(identity, &masks) { + Ok(()) => succeeded(), + Err(error) => affinity_failure(error), + } +} + fn execute_set_priority(identity: ProcIdentity, priority: ProcPriority) -> ProcessActionReport { let handle = match open_action_process( identity, @@ -227,6 +262,13 @@ const fn priority_failure(error: u32) -> ProcessActionReport { } } +const fn affinity_failure(error: u32) -> ProcessActionReport { + ProcessActionReport { + refresh: false, + feedback: ProcessActionFeedback::CantSetAffinity(error), + } +} + const fn succeeded() -> ProcessActionReport { ProcessActionReport { refresh: true, @@ -238,9 +280,10 @@ const fn succeeded() -> ProcessActionReport { mod tests { use super::super::actions::ProcessTreePrepareError; use super::{ - ProcessAction, ProcessActionFeedback, ProcessTreeTerminationOutcome, - append_process_actions, report_end_tree_outcome, report_end_tree_prepare_error, - require_action_identity, run_process_action, spawn_process_action_worker, + AffinityGroupMask, ProcessAction, ProcessActionFeedback, ProcessTreeTerminationOutcome, + affinity_failure, append_process_actions, report_end_tree_outcome, + report_end_tree_prepare_error, require_action_identity, run_process_action, + spawn_process_action_worker, }; use crate::system::process_identity::ProcIdentity; use std::ptr::null_mut; @@ -271,6 +314,13 @@ mod tests { identity: ProcIdentity::new(0, 50), priority: super::super::ProcPriority::High, }, + ProcessAction::SetAffinity { + identity: ProcIdentity::pid_only(12), + masks: vec![AffinityGroupMask { + group: 0, + mask: 0b1111, + }], + }, ] { let report = run_process_action(action); assert!(!report.refresh); @@ -334,6 +384,19 @@ mod tests { identity: first, priority: super::super::ProcPriority::Normal, }, + ProcessAction::SetAffinity { + identity: second, + masks: vec![ + AffinityGroupMask { + group: 0, + mask: 0b0011, + }, + AffinityGroupMask { + group: 1, + mask: 1u64 << 63, + }, + ], + }, ], ); assert!(matches!(pending[0], ProcessAction::End(identity) if identity == first)); @@ -343,6 +406,24 @@ mod tests { ProcessAction::SetPriority { identity, priority: super::super::ProcPriority::Normal } if identity == first )); + assert!(matches!( + &pending[3], + ProcessAction::SetAffinity { identity, masks } + if *identity == second + && masks.len() == 2 + && masks[0] == AffinityGroupMask { group: 0, mask: 0b0011 } + && masks[1] == AffinityGroupMask { group: 1, mask: 1u64 << 63 } + )); + } + + #[test] + fn affinity_failure_does_not_refresh_or_count_as_success() { + let report = affinity_failure(ERROR_INVALID_PARAMETER); + assert!(!report.refresh); + assert_eq!( + report.feedback, + ProcessActionFeedback::CantSetAffinity(ERROR_INVALID_PARAMETER) + ); } #[test] @@ -372,6 +453,39 @@ mod tests { ); } + #[test] + fn worker_completion_does_not_apply_affinity_for_an_unverified_process() { + let mut worker = spawn_process_action_worker().expect("action worker should start"); + worker + .request( + vec![ProcessAction::SetAffinity { + identity: ProcIdentity::new(0, 80), + masks: vec![AffinityGroupMask { + group: 1, + mask: 1u64 << 63, + }], + }], + null_mut(), + ) + .expect("request should submit"); + let deadline = Instant::now() + Duration::from_secs(2); + let reports = loop { + let drained = worker.drain(null_mut()); + assert_eq!(drained.error, None); + if let Some(completion) = drained.completions.into_iter().next() { + break completion.value; + } + assert!(Instant::now() < deadline, "affinity completion timed out"); + thread::yield_now(); + }; + assert_eq!(reports.len(), 1); + assert!(!reports[0].refresh); + assert_eq!( + reports[0].feedback, + ProcessActionFeedback::CantSetAffinity(ERROR_INVALID_PARAMETER) + ); + } + #[test] fn dropping_a_blocked_action_worker_does_not_join() { let mut worker = spawn_process_action_worker().expect("action worker should start"); diff --git a/src/pages/processes/actions.rs b/src/pages/processes/actions.rs index 2d27767..f2b7379 100644 --- a/src/pages/processes/actions.rs +++ b/src/pages/processes/actions.rs @@ -11,9 +11,9 @@ //! Centralizes destructive and identity-sensitive process operations. //! Every target process is reopened through `ProcIdentity` immediately before use. //! -//! End Process, End Process Tree, and Set Priority confirm on the UI thread, then -//! run on one persistent worker. Debug, Open File Location, and Affinity stay on -//! this thread. +//! End Process, End Process Tree, Set Priority, and Set Affinity confirm on the UI +//! thread, then run on one persistent worker. The affinity dialog stays on this +//! thread. Debug and Open File Location stay on this thread. use std::collections::{HashMap, HashSet}; use std::ffi::c_void; @@ -66,7 +66,8 @@ use windows_sys::Win32::UI::WindowsAndMessaging::{ }; use super::action_worker::{ - ProcessAction, ProcessActionFeedback, ProcessActionReport, spawn_process_action_worker, + AffinityGroupMask, ProcessAction, ProcessActionFeedback, ProcessActionReport, + require_action_identity, spawn_process_action_worker, }; use super::{ProcPriority, ProcessPageState}; use crate::infrastructure::native::{ @@ -87,12 +88,16 @@ struct AffinityDialogContext { topology: CpuSetTopology, selected_masks: Vec, selected_group_index: usize, - original_default_ids: Vec, } const PROCESS_TREE_ACCESS: u32 = PROCESS_TERMINATE | PROCESS_QUERY_LIMITED_INFORMATION | SYNCHRONIZE; +const AFFINITY_QUERY_ACCESS: u32 = PROCESS_QUERY_INFORMATION | PROCESS_QUERY_LIMITED_INFORMATION; + +const AFFINITY_UPDATE_ACCESS: u32 = + AFFINITY_QUERY_ACCESS | PROCESS_SET_INFORMATION | PROCESS_SET_LIMITED_INFORMATION; + // `TerminateProcess` fails once a process has begun to exit, and the process object may not be // signaled in that same instant. This bound only applies after termination has already failed. // The wait returns as soon as the process exits. @@ -501,6 +506,9 @@ impl ProcessPageState { ProcessActionFeedback::CantChangePriority(error) => { self.show_failure_message(&self.strings.cant_change_priority, error); } + ProcessActionFeedback::CantSetAffinity(error) => { + self.show_failure_message(&self.strings.cant_set_affinity, error); + } ProcessActionFeedback::KillTreePrepare(error) => { self.show_failure_message(&self.strings.kill_tree_fail_body, error); } @@ -523,20 +531,26 @@ impl ProcessPageState { } } - // Single-group processes retain the classic hard-affinity API. Multi-group systems use CPU - // Set IDs, whose group-qualified identities do not collapse at the 64-processor boundary. + // 亲和性对话框留在 UI 线程。确认后把身份和各组掩码交给持久 worker。 + // SetProcessAffinityMask 与 SetProcessDefaultCpuSets 在 worker 上执行。 pub(super) fn set_affinity(&mut self, identity: ProcIdentity) -> bool { - let handle = match open_process_for_identity( - identity, - PROCESS_QUERY_INFORMATION - | PROCESS_QUERY_LIMITED_INFORMATION - | PROCESS_SET_INFORMATION - | PROCESS_SET_LIMITED_INFORMATION, - ) { + let Some(masks) = self.confirmed_affinity_masks(identity) else { + return false; + }; + self.submit_process_action(ProcessAction::SetAffinity { identity, masks }) + } + + // Query-only handle for the dialog. It is dropped before the action is queued, + // so the worker re-opens the process before it writes a mask. + fn confirmed_affinity_masks( + &mut self, + identity: ProcIdentity, + ) -> Option> { + let handle = match open_process_for_identity(identity, AFFINITY_QUERY_ACCESS) { Ok(handle) => handle, Err(error) => { self.show_failure_message(&self.strings.cant_set_affinity, error); - return false; + return None; } }; @@ -544,21 +558,21 @@ impl ProcessPageState { Ok(topology) => topology, Err(error) => { self.show_failure_message(&self.strings.cant_set_affinity, error.win32_code()); - return false; + return None; } }; let original_default_ids = match query_process_default_cpu_sets(handle.as_raw()) { Ok(ids) => ids, Err(error) => { self.show_failure_message(&self.strings.cant_set_affinity, error.win32_code()); - return false; + return None; } }; let original_process_groups = match query_process_groups(handle.as_raw()) { Ok(groups) => groups, Err(error) => { self.show_failure_message(&self.strings.cant_set_affinity, error); - return false; + return None; } }; let selected_masks = match initial_affinity_masks( @@ -570,7 +584,7 @@ impl ProcessPageState { Ok(masks) => masks, Err(error) => { self.show_failure_message(&self.strings.cant_set_affinity, error); - return false; + return None; } }; let selected_group_index = selected_masks @@ -582,7 +596,6 @@ impl ProcessPageState { topology, selected_masks, selected_group_index, - original_default_ids, }; match dialog_box( @@ -592,22 +605,17 @@ impl ProcessPageState { Some(affinity_dialog_proc), &mut context as *mut AffinityDialogContext as LPARAM, ) { - Ok(result) if result == IDOK as isize => { - match apply_affinity_selection(handle.as_raw(), identity.pid, &context) { - Ok(()) => { - self.refresh_processes(); - true - } - Err(error) => { - self.show_failure_message(&self.strings.cant_set_affinity, error); - false - } + Ok(result) if result == IDOK as isize => match confirmed_group_masks(&context) { + Ok(masks) => Some(masks), + Err(error) => { + self.show_failure_message(&self.strings.cant_set_affinity, error); + None } - } - Ok(_) => false, + }, + Ok(_) => None, Err(error) => { self.show_failure_message(&self.strings.cant_set_affinity, error); - false + None } } } @@ -616,12 +624,14 @@ impl ProcessPageState { enum ActionFailureKind { Kill, Priority, + Affinity, } fn action_failure_kind(action: &ProcessAction) -> ActionFailureKind { match action { ProcessAction::End(_) | ProcessAction::EndTree(_) => ActionFailureKind::Kill, ProcessAction::SetPriority { .. } => ActionFailureKind::Priority, + ProcessAction::SetAffinity { .. } => ActionFailureKind::Affinity, #[cfg(test)] ProcessAction::Block { .. } => ActionFailureKind::Kill, } @@ -633,10 +643,30 @@ fn failure_report(kind: ActionFailureKind, error: u32) -> ProcessActionReport { feedback: match kind { ActionFailureKind::Kill => ProcessActionFeedback::CantKill(error), ActionFailureKind::Priority => ProcessActionFeedback::CantChangePriority(error), + ActionFailureKind::Affinity => ProcessActionFeedback::CantSetAffinity(error), }, } } +fn confirmed_group_masks(context: &AffinityDialogContext) -> Result, u32> { + context + .topology + .groups() + .iter() + .zip(context.selected_masks.iter().copied()) + .map(|(group, mask)| { + // `u64` can hold every `usize` bit. A failed conversion is reported + // instead of submitting a shorter mask. + u64::try_from(mask) + .map(|mask| AffinityGroupMask { + group: group.number, + mask, + }) + .map_err(|_| ERROR_NOT_SUPPORTED) + }) + .collect() +} + fn initial_affinity_masks( process: HANDLE, topology: &CpuSetTopology, @@ -678,48 +708,167 @@ fn initial_affinity_masks( Ok(masks) } -fn apply_affinity_selection( - process: HANDLE, - process_id: u32, - context: &AffinityDialogContext, -) -> Result<(), u32> { - let selected_ids = context - .topology - .ids_for_masks(&context.selected_masks) - .map_err(|error| error.win32_code())?; - if selected_ids.is_empty() { +/// How a validated selection is written. A multi-group selection stays a list of +/// group masks and is never folded into `SetProcessAffinityMask`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) enum AffinityUpdatePlan { + ProcessMask(usize), + GroupCpuSets(Vec), +} + +pub(super) fn plan_affinity_update(masks: &[usize]) -> Result { + if masks.is_empty() || masks.iter().all(|mask| *mask == 0) { return Err(ERROR_INVALID_PARAMETER); } + if masks.len() == 1 { + Ok(AffinityUpdatePlan::ProcessMask(masks[0])) + } else { + Ok(AffinityUpdatePlan::GroupCpuSets(masks.to_vec())) + } +} - if context.topology.groups().len() == 1 { - set_process_default_cpu_sets(process, &[])?; - if unsafe { SetProcessAffinityMask(process, context.selected_masks[0]) } == 0 { - let error = nonzero_last_error(); - let _ = set_process_default_cpu_sets(process, &context.original_default_ids); - return Err(error); +/// Place each selected mask on the topology group with the same number. +/// +/// A missing group, a duplicated group, a bit outside the assignable mask, or a +/// mask wider than native `KAFFINITY` is an error. Bits are not clipped to the +/// low half of the mask or moved onto another group. +pub(super) fn align_affinity_selection( + topology_groups: &[(u16, usize)], + selected: &[(u16, u64)], +) -> Result, u32> { + if topology_groups.is_empty() || selected.len() != topology_groups.len() { + return Err(ERROR_NOT_SUPPORTED); + } + if duplicate_group_ids(topology_groups.iter().map(|(group, _)| *group)) + || duplicate_group_ids(selected.iter().map(|(group, _)| *group)) + { + return Err(ERROR_INVALID_PARAMETER); + } + + let mut masks = Vec::with_capacity(topology_groups.len()); + for &(group, assignable) in topology_groups { + let Some(&(_, wide_mask)) = selected.iter().find(|(number, _)| *number == group) else { + return Err(ERROR_NOT_SUPPORTED); + }; + let native = usize::try_from(wide_mask).map_err(|_| ERROR_NOT_SUPPORTED)?; + if native & !assignable != 0 { + return Err(ERROR_INVALID_PARAMETER); } - return Ok(()); + masks.push(native); } + if masks.iter().all(|mask| *mask == 0) { + return Err(ERROR_INVALID_PARAMETER); + } + Ok(masks) +} - set_process_default_cpu_sets(process, &selected_ids)?; - let selected_groups = context - .topology +fn duplicate_group_ids(mut groups: impl Iterator) -> bool { + let mut seen = HashSet::::new(); + groups.any(|group| !seen.insert(group)) +} + +pub(super) fn apply_confirmed_affinity( + identity: ProcIdentity, + selected: &[AffinityGroupMask], +) -> Result<(), u32> { + // Unknown and mismatched identities fail here, before either mask API. + require_action_identity(identity)?; + let handle = open_process_for_identity(identity, AFFINITY_UPDATE_ACCESS)?; + let topology = CpuSetTopology::query(handle.as_raw()).map_err(|error| error.win32_code())?; + let original_default_ids = + query_process_default_cpu_sets(handle.as_raw()).map_err(|error| error.win32_code())?; + let topology_groups = topology .groups() .iter() - .zip(context.selected_masks.iter().copied()) - .filter(|(_, mask)| *mask != 0) - .map(|(group, mask)| (group.number, mask)) + .map(|group| (group.number, group.assignable_mask)) .collect::>(); + let selected_pairs = selected + .iter() + .map(|group| (group.group, group.mask)) + .collect::>(); + let masks = align_affinity_selection(&topology_groups, &selected_pairs)?; + apply_affinity_selection( + handle.as_raw(), + identity, + &topology, + &masks, + &original_default_ids, + ) +} - // CPU Sets span groups, but a restrictive thread affinity mask takes precedence over a - // conflicting CPU Set assignment. Apply a group-qualified hard mask to every existing thread - // so an old per-thread affinity cannot keep using a CPU the user just deselected. The process - // default CPU Sets above cover threads created after the verified thread snapshot. - if let Err(error) = apply_thread_group_affinities(process_id, &selected_groups) { - let _ = set_process_default_cpu_sets(process, &context.original_default_ids); - return Err(error); +fn apply_affinity_selection( + process: HANDLE, + identity: ProcIdentity, + topology: &CpuSetTopology, + selected_masks: &[usize], + original_default_ids: &[u32], +) -> Result<(), u32> { + let selected_ids = topology + .ids_for_masks(selected_masks) + .map_err(|error| error.win32_code())?; + if selected_ids.is_empty() { + return Err(ERROR_INVALID_PARAMETER); + } + if selected_masks.len() != topology.groups().len() { + return Err(ERROR_NOT_SUPPORTED); + } + + // Validation is finished. The handle was opened for this identity above. + match plan_affinity_update(selected_masks)? { + AffinityUpdatePlan::ProcessMask(mask) => { + if topology.groups().len() != 1 { + return Err(ERROR_NOT_SUPPORTED); + } + set_process_default_cpu_sets(process, &[])?; + if unsafe { SetProcessAffinityMask(process, mask) } == 0 { + let error = nonzero_last_error(); + let _ = set_process_default_cpu_sets(process, original_default_ids); + // The CPU-set clear already landed. Restoring it does not make the + // failed mask a success. + return Err(error); + } + Ok(()) + } + AffinityUpdatePlan::GroupCpuSets(masks) => { + if masks.len() != topology.groups().len() || topology.groups().len() < 2 { + return Err(ERROR_NOT_SUPPORTED); + } + set_process_default_cpu_sets(process, &selected_ids)?; + let selected_groups = topology + .groups() + .iter() + .zip(masks.iter().copied()) + .filter(|(_, mask)| *mask != 0) + .map(|(group, mask)| (group.number, mask)) + .collect::>(); + // Thread masks are applied by PID. Re-check the identity first so a + // reused PID does not receive this process's group masks. A mismatch + // after the CPU-set update is still a failure. + match query_process_identity_for_pid(identity.pid) { + Ok(current) if current == identity => {} + Ok(_) => { + let _ = set_process_default_cpu_sets(process, original_default_ids); + return Err(ERROR_INVALID_PARAMETER); + } + Err(error) => { + let _ = set_process_default_cpu_sets(process, original_default_ids); + return Err(error); + } + } + + // CPU Sets span groups, but a restrictive thread affinity mask takes + // precedence over a conflicting CPU Set assignment. Apply a + // group-qualified hard mask to every existing thread so an old + // per-thread affinity cannot keep using a CPU the user just + // deselected. The process default CPU Sets above cover threads + // created after the verified thread snapshot. + if let Err(error) = apply_thread_group_affinities(identity.pid, &selected_groups) { + let _ = set_process_default_cpu_sets(process, original_default_ids); + return Err(error); + } + Ok(()) + } } - Ok(()) } fn set_process_default_cpu_sets(process: HANDLE, ids: &[u32]) -> Result<(), u32> { @@ -1943,3 +2092,107 @@ mod tree_identity_tests { )); } } + +#[cfg(test)] +mod affinity_selection_tests { + use super::{ + AffinityGroupMask, AffinityUpdatePlan, align_affinity_selection, apply_confirmed_affinity, + plan_affinity_update, + }; + use crate::system::process_identity::ProcIdentity; + use windows_sys::Win32::Foundation::{ERROR_INVALID_PARAMETER, ERROR_NOT_SUPPORTED}; + + #[test] + fn unverified_identity_does_not_apply_a_mask() { + let masks = [AffinityGroupMask { + group: 0, + mask: 0b1111, + }]; + assert_eq!( + apply_confirmed_affinity(ProcIdentity::pid_only(4), &masks), + Err(ERROR_INVALID_PARAMETER) + ); + assert_eq!( + apply_confirmed_affinity(ProcIdentity::new(0, 100), &masks), + Err(ERROR_INVALID_PARAMETER) + ); + } + + #[test] + fn multigroup_plan_does_not_collapse_to_the_low_group() { + let masks = [0b0011usize, 0b1100]; + assert_eq!( + plan_affinity_update(&masks), + Ok(AffinityUpdatePlan::GroupCpuSets(vec![0b0011, 0b1100])) + ); + assert_ne!( + plan_affinity_update(&masks), + Ok(AffinityUpdatePlan::ProcessMask(0b0011)) + ); + assert_eq!( + plan_affinity_update(&[0b1010]), + Ok(AffinityUpdatePlan::ProcessMask(0b1010)) + ); + } + + #[test] + fn inexpressible_group_or_mask_is_not_truncated() { + let topology = [(0u16, 0b0011usize), (1u16, 0b1100)]; + // Bit 0 of group 0 is expressible. Bit 3 is not, so the mask is not reduced to bit 0. + assert_eq!( + align_affinity_selection(&topology, &[(0, 0b1001), (1, 0b0100)]), + Err(ERROR_INVALID_PARAMETER) + ); + // Group 2 is absent. Its bits are not applied to group 0 or group 1. + assert_eq!( + align_affinity_selection(&topology, &[(0, 0b0001), (2, 0b0100)]), + Err(ERROR_NOT_SUPPORTED) + ); + assert_eq!( + align_affinity_selection(&topology, &[(1, 0b1000), (0, 0b0010)]).unwrap(), + vec![0b0010, 0b1000] + ); + } + + #[test] + fn mask_wider_than_native_kaffinity_is_not_truncated() { + let wide = 1u64 << 63; + let result = align_affinity_selection(&[(0, usize::MAX)], &[(0, wide)]); + if usize::BITS < 64 { + assert_eq!(result, Err(ERROR_NOT_SUPPORTED)); + } else { + assert_eq!(result.unwrap(), vec![usize::try_from(wide).unwrap()]); + } + + // Truncating to 32 bits would leave 0x1. That value must not be applied. + let widened = (1u64 << 32) | 1; + let narrowed = align_affinity_selection(&[(0, usize::MAX)], &[(0, widened)]); + if usize::BITS < 64 { + assert_eq!(narrowed, Err(ERROR_NOT_SUPPORTED)); + } else { + assert_eq!(narrowed.unwrap(), vec![usize::try_from(widened).unwrap()]); + } + } + + #[test] + fn highest_native_affinity_bit_survives_alignment() { + let high = 1usize << (usize::BITS - 1); + assert_eq!( + align_affinity_selection(&[(3, high)], &[(3, u64::try_from(high).unwrap())]).unwrap(), + vec![high] + ); + } + + #[test] + fn empty_or_duplicate_affinity_selection_is_rejected() { + assert_eq!( + align_affinity_selection(&[(0, 0b1)], &[(0, 0)]), + Err(ERROR_INVALID_PARAMETER) + ); + assert_eq!(plan_affinity_update(&[0, 0]), Err(ERROR_INVALID_PARAMETER)); + assert_eq!( + align_affinity_selection(&[(0, 1), (0, 1)], &[(0, 1), (1, 1)]), + Err(ERROR_INVALID_PARAMETER) + ); + } +} diff --git a/src/pages/processes/mod.rs b/src/pages/processes/mod.rs index 0e12ddc..9d27136 100644 --- a/src/pages/processes/mod.rs +++ b/src/pages/processes/mod.rs @@ -26,11 +26,13 @@ mod tree; // 线程模型： // SingleFlightWorker 工作线程在后台执行 collect_process_entries() 采样， // UI 线程只提交刷新意图并异步提交结果；CPU delta 基线由 worker 独占。 -// 结束进程、结束进程树和设置优先级在确认后进入同一个持久 SingleFlightWorker。 -// 确认框留在 UI 线程。TerminateProcess、进程树遍历和 SetPriorityClass 在该 -// worker 上执行，执行前用 open_process_for_identity 核对 pid 与创建时间。 -// 完成通知复用 PWM_PROC_WORKER_COMPLETE，和采样结果一起排空。页面销毁时丢弃 -// 该 worker，不在 UI 线程上 join。 +// 结束进程、结束进程树、设置优先级和设置亲和性在确认后进入同一个持久 +// SingleFlightWorker。确认框和亲和性对话框留在 UI 线程。TerminateProcess、 +// 进程树遍历、SetPriorityClass、SetProcessAffinityMask 和 +// SetProcessDefaultCpuSets 在该 worker 上执行，执行前用 +// open_process_for_identity 核对 pid 与创建时间。完成通知复用 +// PWM_PROC_WORKER_COMPLETE，和采样结果一起排空。页面销毁时丢弃该 worker， +// 不在 UI 线程上 join。 // 创建转储文件使用另一个 worker。保存对话框留在 UI 线程， // MiniDumpWriteDump 在该 worker 上执行，不与采样请求合并。 // diff --git a/src/pages/services/mod.rs b/src/pages/services/mod.rs index b2b2aa9..6707c09 100644 --- a/src/pages/services/mod.rs +++ b/src/pages/services/mod.rs @@ -12,7 +12,7 @@ mod model; mod sampler; -// 服务页列出本机 Win32 服务，并在后台 worker 上启动、停止、暂停或继续服务。 +// 服务页列出本机 Win32 服务，并在后台 worker 上启动、停止、重新启动、暂停或继续服务。 // UI 线程只提交请求、维护列表选择和绘制。刷新保留服务名对应的选择、焦点和滚动位置。 // “转到进程”只使用带创建时间的 ProcIdentity；没有校验过的进程时该命令保持禁用。 @@ -48,8 +48,8 @@ use crate::ui::localization::{TextKey, service_column_titles, text}; use crate::ui::menus::build_popup_menu; use crate::ui::resource_ids::{ IDC_SERVICELIST, IDM_SERVICE_CONTINUE, IDM_SERVICE_GOTOPROCESS, IDM_SERVICE_PAUSE, - IDM_SERVICE_START, IDM_SERVICE_STOP, IDR_SERVICE_CONTEXT, PWM_SERVICE_WORKER_COMPLETE, - WM_FINDPROC, + IDM_SERVICE_RESTART, IDM_SERVICE_START, IDM_SERVICE_STOP, IDR_SERVICE_CONTEXT, + PWM_SERVICE_WORKER_COMPLETE, WM_FINDPROC, }; pub(crate) use self::model::ServiceCommandState; @@ -172,6 +172,7 @@ impl ServicePageState { state.start = false; state.stop = false; } + state.restart = state.stop; state } @@ -185,6 +186,10 @@ impl ServicePageState { self.request_control(ServiceControlAction::Stop); true } + IDM_SERVICE_RESTART => { + self.request_control(ServiceControlAction::Restart); + true + } IDM_SERVICE_PAUSE => { self.request_control(ServiceControlAction::Pause); true @@ -268,6 +273,7 @@ impl ServicePageState { unsafe { enable_command(popup.as_raw(), IDM_SERVICE_START, state.start); enable_command(popup.as_raw(), IDM_SERVICE_STOP, state.stop); + enable_command(popup.as_raw(), IDM_SERVICE_RESTART, state.restart); enable_command(popup.as_raw(), IDM_SERVICE_PAUSE, state.pause); enable_command(popup.as_raw(), IDM_SERVICE_CONTINUE, state.continue_service); enable_command(popup.as_raw(), IDM_SERVICE_GOTOPROCESS, state.go_to_process); @@ -354,11 +360,11 @@ impl ServicePageState { if self.worker.is_some() { return Ok(()); } - self.worker = Some(SingleFlightWorker::spawn( + self.worker = Some(SingleFlightWorker::spawn_cancellable_initialized( "taskmgr-rs-service-sampler", PWM_SERVICE_WORKER_COMPLETE, merge_service_requests, - perform, + |stop| move |request| perform(request, &stop), )?); Ok(()) } @@ -378,6 +384,7 @@ impl ServicePageState { let allowed = match action { ServiceControlAction::Start => state.start, ServiceControlAction::Stop => state.stop, + ServiceControlAction::Restart => state.restart, ServiceControlAction::Pause => state.pause, ServiceControlAction::Continue => state.continue_service, }; @@ -704,6 +711,7 @@ impl ServicePageState { let summary = match action { ServiceControlAction::Start => text(TextKey::ServiceStartFailed), ServiceControlAction::Stop => text(TextKey::ServiceStopFailed), + ServiceControlAction::Restart => text(TextKey::ServiceRestartFailed), ServiceControlAction::Pause => text(TextKey::ServicePauseFailed), ServiceControlAction::Continue => text(TextKey::ServiceContinueFailed), }; @@ -870,6 +878,34 @@ mod tests { assert_eq!(page.last_refresh_error, Some(ERROR_BROKEN_PIPE)); } + #[test] + fn restart_is_enabled_exactly_when_stop_is_and_a_pending_control_disables_it() { + let mut page = ServicePageState { + rows: vec![service_row( + self::sampler::ServiceRecord { + service_name: "test-service".to_string(), + description: String::new(), + status: ServiceRunState::Running, + startup: ServiceStartupType::Manual, + accepts_stop: true, + accepts_pause_continue: false, + process: None, + }, + "PID", + )], + selected_service: Some("test-service".to_string()), + ..ServicePageState::default() + }; + let enabled = page.command_state(); + assert!(enabled.stop); + assert!(enabled.restart); + page.pending_control = Some(ServiceControlAction::Restart); + let pending = page.command_state(); + assert!(!pending.stop); + assert!(!pending.restart); + assert!(!pending.start); + } + #[test] fn a_queued_service_control_cannot_be_overwritten_by_another_selection() { let (started_sender, started_receiver) = crossbeam_channel::bounded(1); diff --git a/src/pages/services/model.rs b/src/pages/services/model.rs index 484d786..190f921 100644 --- a/src/pages/services/model.rs +++ b/src/pages/services/model.rs @@ -16,6 +16,7 @@ //! enable Go to Process. use crate::system::process_identity::ProcIdentity; +use windows_sys::Win32::Foundation::ERROR_SERVICE_REQUEST_TIMEOUT; pub(crate) const SERVICE_STOPPED: u32 = 1; pub(crate) const SERVICE_START_PENDING: u32 = 2; @@ -57,6 +58,7 @@ pub(crate) enum ServiceStartupType { pub(crate) enum ServiceControlAction { Start, Stop, + Restart, Pause, Continue, } @@ -74,6 +76,7 @@ pub(crate) enum ServiceRequest { pub(crate) struct ServiceCommandState { pub(crate) start: bool, pub(crate) stop: bool, + pub(crate) restart: bool, pub(crate) pause: bool, pub(crate) continue_service: bool, pub(crate) go_to_process: bool, @@ -84,6 +87,7 @@ impl ServiceCommandState { Self { start: false, stop: false, + restart: false, pause: false, continue_service: false, go_to_process: false, @@ -145,14 +149,16 @@ pub(crate) fn command_state(selected: Option) -> ServiceCom let Some(selected) = selected else { return ServiceCommandState::disabled(); }; + let stop = selected.accepts_stop + && matches!( + selected.status, + ServiceRunState::Running | ServiceRunState::Paused + ); ServiceCommandState { start: selected.status == ServiceRunState::Stopped && selected.startup != ServiceStartupType::Disabled, - stop: selected.accepts_stop - && matches!( - selected.status, - ServiceRunState::Running | ServiceRunState::Paused - ), + stop, + restart: stop, go_to_process: service_has_valid_process_id(selected.status) && selected.process.is_some_and(ProcIdentity::is_verified), pause: selected.accepts_pause_continue && selected.status == ServiceRunState::Running, @@ -184,6 +190,12 @@ pub(crate) fn control_reached_requested_state( state, ServiceRunState::Stopped | ServiceRunState::StopPending ), + // Restart's start half uses the same accepted states as Start. Stop-pending, + // stopped, and unknown are not a successful restart. + ServiceControlAction::Restart => matches!( + state, + ServiceRunState::Running | ServiceRunState::StartPending + ), ServiceControlAction::Pause => matches!( state, ServiceRunState::Paused | ServiceRunState::PausePending @@ -206,6 +218,24 @@ pub(crate) fn classify_control_result( } } +/// Stop has finished only when the service is stopped. Pending and unknown are not completion. +pub(crate) fn stop_has_completed(state: ServiceRunState) -> bool { + state == ServiceRunState::Stopped +} + +/// Starts a restart only after stop completed. A stop error or any other observed state is +/// rejected and `start` is not called. An old Running state is never restart success. +pub(crate) fn restart_after_stop( + stop: Result, + start: impl FnOnce() -> Result, +) -> Result { + match stop { + Ok(state) if stop_has_completed(state) => start(), + Ok(_) => Err(ERROR_SERVICE_REQUEST_TIMEOUT), + Err(error) => Err(error), + } +} + /// Keeps a pending start or stop ahead of a coalesced refresh. pub(crate) fn merge_service_requests(pending: &mut ServiceRequest, incoming: ServiceRequest) { match (&*pending, &incoming) { @@ -307,6 +337,7 @@ mod tests { ))); assert!(stopped.start); assert!(!stopped.stop); + assert!(!stopped.restart); assert!(!stopped.pause); assert!(!stopped.continue_service); assert!(!stopped.go_to_process); @@ -325,6 +356,7 @@ mod tests { ))); assert!(!running.start); assert!(running.stop); + assert!(running.restart); assert!(running.pause); assert!(!running.continue_service); assert!(!running.go_to_process); @@ -335,6 +367,7 @@ mod tests { Some(ProcIdentity::new(44, 100)), ))); assert!(verified.stop); + assert!(verified.restart); assert!(!verified.pause); assert!(verified.continue_service); assert!(verified.go_to_process); @@ -346,6 +379,7 @@ mod tests { ))); assert!(!pending.start); assert!(!pending.stop); + assert!(!pending.restart); assert!(!pending.pause); assert!(!pending.continue_service); assert!(!pending.go_to_process); @@ -356,9 +390,13 @@ mod tests { for status in [ServiceRunState::Running, ServiceRunState::Paused] { let mut selected = input(status, ServiceStartupType::Automatic, None); selected.accepts_stop = false; - assert!(!command_state(Some(selected)).stop); + let blocked = command_state(Some(selected)); + assert!(!blocked.stop); + assert!(!blocked.restart); selected.accepts_stop = true; - assert!(command_state(Some(selected)).stop); + let allowed = command_state(Some(selected)); + assert!(allowed.stop); + assert!(allowed.restart); } } @@ -477,6 +515,91 @@ mod tests { ); } + #[test] + fn restart_starts_only_after_a_completed_stop_and_unknown_is_not_success() { + assert!(stop_has_completed(ServiceRunState::Stopped)); + assert!(!stop_has_completed(ServiceRunState::StopPending)); + assert!(!stop_has_completed(ServiceRunState::Running)); + assert!(!stop_has_completed(ServiceRunState::Unknown(0))); + + let mut starts = 0u32; + let rejected = restart_after_stop(Err(1060), || { + starts += 1; + Ok(ServiceRunState::Running) + }); + assert_eq!(rejected, Err(1060)); + assert_eq!(starts, 0); + + for state in [ + ServiceRunState::StopPending, + ServiceRunState::Running, + ServiceRunState::Paused, + ServiceRunState::Unknown(7), + ] { + let observed = restart_after_stop(Ok(state), || { + starts += 1; + Ok(ServiceRunState::Running) + }); + assert_eq!(observed, Err(ERROR_SERVICE_REQUEST_TIMEOUT)); + assert!(classify_control_result(ServiceControlAction::Restart, observed).is_err()); + } + assert_eq!(starts, 0); + + let start_error = restart_after_stop(Ok(ServiceRunState::Stopped), || { + starts += 1; + Err(5) + }); + assert_eq!(start_error, Err(5)); + assert_eq!(starts, 1); + assert_eq!( + classify_control_result(ServiceControlAction::Restart, start_error), + Err(ServiceActionFailure::Api(5)) + ); + + assert!(control_reached_requested_state( + ServiceControlAction::Restart, + ServiceRunState::Running + )); + assert!(control_reached_requested_state( + ServiceControlAction::Restart, + ServiceRunState::StartPending + )); + assert!(!control_reached_requested_state( + ServiceControlAction::Restart, + ServiceRunState::Stopped + )); + assert!(!control_reached_requested_state( + ServiceControlAction::Restart, + ServiceRunState::StopPending + )); + assert!(!control_reached_requested_state( + ServiceControlAction::Restart, + ServiceRunState::Unknown(0) + )); + assert_eq!( + classify_control_result( + ServiceControlAction::Restart, + Ok(ServiceRunState::StartPending) + ), + Ok(ServiceRunState::StartPending) + ); + assert_eq!( + classify_control_result( + ServiceControlAction::Restart, + Ok(ServiceRunState::Unknown(9)) + ), + Err(ServiceActionFailure::StateUnchanged { + observed: ServiceRunState::Unknown(9) + }) + ); + assert_eq!( + classify_control_result(ServiceControlAction::Restart, Ok(ServiceRunState::Stopped)), + Err(ServiceActionFailure::StateUnchanged { + observed: ServiceRunState::Stopped + }) + ); + } + #[test] fn refresh_does_not_replace_a_pending_control() { let mut pending = ServiceRequest::Control { diff --git a/src/pages/services/sampler.rs b/src/pages/services/sampler.rs index c46479b..307b66d 100644 --- a/src/pages/services/sampler.rs +++ b/src/pages/services/sampler.rs @@ -9,18 +9,23 @@ // 作者: OpenAI Codex // -------------------------------------------------------------------------- -//! Service Control Manager queries and start, stop, pause, and continue calls. +//! Service Control Manager queries and start, stop, restart, pause, and continue calls. //! //! The page worker owns every SCM handle. Enumeration asks only for connect and enumerate rights. //! Each service is opened with query rights, and a control opens that one service with only the -//! right the action needs plus status query so the result can be checked. +//! rights that action needs plus status query so the result can be checked. Restart opens with +//! stop and start on that same handle. It sends stop, then starts only after stop has completed. +//! Stop completion follows SCM checkpoint progress with an overall deadline. Cancellation is +//! checked between short polls, so page destruction never joins a sleeping SCM worker. +//! Pending and unknown states are not completion. use std::collections::HashMap; use std::mem::{size_of, zeroed}; +use std::time::{Duration, Instant}; use windows_sys::Win32::Foundation::{ - ERROR_GEN_FAILURE, ERROR_INSUFFICIENT_BUFFER, ERROR_INVALID_PARAMETER, ERROR_MORE_DATA, - GetLastError, + ERROR_CANCELLED, ERROR_GEN_FAILURE, ERROR_INSUFFICIENT_BUFFER, ERROR_INVALID_PARAMETER, + ERROR_MORE_DATA, ERROR_SERVICE_REQUEST_TIMEOUT, GetLastError, }; use windows_sys::Win32::System::Services::{ CloseServiceHandle, ControlService, ENUM_SERVICE_STATUS_PROCESSW, EnumServicesStatusExW, @@ -37,18 +42,21 @@ use windows_sys::Win32::System::Threading::{ }; use crate::infrastructure::native::to_wide_null; +use crate::infrastructure::worker::WorkerStopSignal; use crate::system::process_identity::{ ProcIdentity, open_process_for_identity, query_process_identity_for_pid, }; use super::model::{ OwningProcess, ServiceActionFailure, ServiceControlAction, ServiceRequest, ServiceRunState, - ServiceStartupType, classify_control_result, file_name_from_win32_path, - service_has_valid_process_id, startup_from_scm, state_from_scm, + ServiceStartupType, classify_control_result, file_name_from_win32_path, restart_after_stop, + service_has_valid_process_id, startup_from_scm, state_from_scm, stop_has_completed, }; const ENUM_ATTEMPT_LIMIT: usize = 10_000; const CONFIG_GROW_LIMIT: usize = 8; +/// Overall deadline, independent of a service repeatedly advancing its checkpoint. +const SERVICE_STOP_WAIT_CAP_MS: u32 = 30_000; #[derive(Clone, Debug, PartialEq, Eq)] pub(super) struct ServiceRecord { @@ -118,13 +126,13 @@ impl Drop for ScHandle { } } -pub(super) fn perform(request: ServiceRequest) -> ServiceWorkerCompletion { +pub(super) fn perform(request: ServiceRequest, stop: &WorkerStopSignal) -> ServiceWorkerCompletion { match request { ServiceRequest::Refresh => ServiceWorkerCompletion::Snapshot(collect_services()), ServiceRequest::Control { service_name, action, - } => ServiceWorkerCompletion::Action(control_service(&service_name, action)), + } => ServiceWorkerCompletion::Action(control_service(&service_name, action, stop)), } } @@ -466,11 +474,18 @@ fn query_process_file_name(identity: ProcIdentity) -> Result { Ok(file_name_from_win32_path(&path).to_string()) } -fn control_service(service_name: &str, action: ServiceControlAction) -> ServiceActionReport { +fn control_service( + service_name: &str, + action: ServiceControlAction, + stop: &WorkerStopSignal, +) -> ServiceActionReport { let outcome = if service_name.is_empty() { Err(ServiceActionFailure::Api(ERROR_INVALID_PARAMETER)) } else { - classify_control_result(action, control_service_and_read_state(service_name, action)) + classify_control_result( + action, + control_service_and_read_state(service_name, action, stop), + ) }; ServiceActionReport { service_name: service_name.to_string(), @@ -482,7 +497,11 @@ fn control_service(service_name: &str, action: ServiceControlAction) -> ServiceA fn control_service_and_read_state( service_name: &str, action: ServiceControlAction, + stop: &WorkerStopSignal, ) -> Result { + if stop.is_cancelled() { + return Err(ERROR_CANCELLED); + } // SAFETY: null names select the local SCM. Connect is the only manager right required to open // one service; the service handle itself carries the action right. let manager = ScHandle::new(unsafe { @@ -491,39 +510,148 @@ fn control_service_and_read_state( let access = match action { ServiceControlAction::Start => SERVICE_START | SERVICE_QUERY_STATUS, ServiceControlAction::Stop => SERVICE_STOP | SERVICE_QUERY_STATUS, + ServiceControlAction::Restart => SERVICE_STOP | SERVICE_START | SERVICE_QUERY_STATUS, ServiceControlAction::Pause | ServiceControlAction::Continue => { SERVICE_PAUSE_CONTINUE | SERVICE_QUERY_STATUS } }; let service = open_service(manager.as_raw(), service_name, access)?; - let control = match action { + match action { ServiceControlAction::Start => { - // SAFETY: the service is open for SERVICE_START. No start arguments are supplied. - let success = unsafe { StartServiceW(service.as_raw(), 0, std::ptr::null()) }; - if success == 0 { - return Err(last_error()); - } - None + start_open_service(service.as_raw())?; + query_service_state(service.as_raw()) } - ServiceControlAction::Stop => Some(SERVICE_CONTROL_STOP), - ServiceControlAction::Pause => Some(SERVICE_CONTROL_PAUSE), - ServiceControlAction::Continue => Some(SERVICE_CONTROL_CONTINUE), - }; - if let Some(control) = control { - let mut status = unsafe { zeroed::() }; - // SAFETY: the service is open for this one control right. `status` is a writable output - // of the control call and is not used as proof of the final state. - let success = unsafe { ControlService(service.as_raw(), control, &mut status) }; - if success == 0 { - return Err(last_error()); + ServiceControlAction::Stop => { + send_service_control(service.as_raw(), SERVICE_CONTROL_STOP)?; + query_service_state(service.as_raw()) + } + ServiceControlAction::Pause => { + send_service_control(service.as_raw(), SERVICE_CONTROL_PAUSE)?; + query_service_state(service.as_raw()) } + ServiceControlAction::Continue => { + send_service_control(service.as_raw(), SERVICE_CONTROL_CONTINUE)?; + query_service_state(service.as_raw()) + } + ServiceControlAction::Restart => restart_open_service(service.as_raw(), stop), + } +} + +fn restart_open_service( + service: windows_sys::Win32::System::Services::SC_HANDLE, + cancellation: &WorkerStopSignal, +) -> Result { + // Acceptance of STOP is not completion. Only an observed STOPPED state permits START. + let stop = send_service_control(service, SERVICE_CONTROL_STOP) + .and_then(|_| wait_until_service_stopped(service, cancellation)); + restart_after_stop(stop, || { + if cancellation.is_cancelled() { + return Err(ERROR_CANCELLED); + } + start_open_service(service)?; + query_service_state(service) + }) +} + +fn wait_until_service_stopped( + service: windows_sys::Win32::System::Services::SC_HANDLE, + cancellation: &WorkerStopSignal, +) -> Result { + let started = Instant::now(); + poll_service_stop( + || query_service_status(service), + || started.elapsed(), + std::thread::sleep, + || cancellation.is_cancelled(), + ) +} + +fn poll_service_stop( + mut query: impl FnMut() -> Result, + elapsed: impl Fn() -> Duration, + mut wait: impl FnMut(Duration), + cancelled: impl Fn() -> bool, +) -> Result { + let overall_limit = Duration::from_millis(u64::from(SERVICE_STOP_WAIT_CAP_MS)); + let mut progress = None; + let mut progress_at = Duration::ZERO; + let mut progress_limit = overall_limit; + loop { + if cancelled() { + return Err(ERROR_CANCELLED); + } + if elapsed() >= overall_limit { + return Err(ERROR_SERVICE_REQUEST_TIMEOUT); + } + let status = query()?; + if cancelled() { + return Err(ERROR_CANCELLED); + } + let state = state_from_scm(status.dwCurrentState); + if stop_has_completed(state) { + return Ok(state); + } + let now = elapsed(); + // The wait hint bounds the next checkpoint/state change, not the whole stop. + // Increasing a hint without making progress must not extend the deadline. + if progress.is_none_or(|(old_state, checkpoint)| { + status.dwCurrentState != old_state || status.dwCheckPoint > checkpoint + }) { + progress = Some((status.dwCurrentState, status.dwCheckPoint)); + progress_at = now; + progress_limit = Duration::from_millis(u64::from( + status.dwWaitHint.clamp(1_000, SERVICE_STOP_WAIT_CAP_MS), + )); + } + let stalled_for = now.saturating_sub(progress_at); + if now >= overall_limit || stalled_for >= progress_limit { + return Err(ERROR_SERVICE_REQUEST_TIMEOUT); + } + // A short upper bound lets cancellation interrupt even a very large wait hint. + wait( + Duration::from_millis(100) + .min(overall_limit - now) + .min(progress_limit - stalled_for), + ); + } +} + +fn start_open_service(service: windows_sys::Win32::System::Services::SC_HANDLE) -> Result<(), u32> { + // SAFETY: the service is open for SERVICE_START. No start arguments are supplied. + let success = unsafe { StartServiceW(service, 0, std::ptr::null()) }; + if success == 0 { + Err(last_error()) + } else { + Ok(()) + } +} + +fn send_service_control( + service: windows_sys::Win32::System::Services::SC_HANDLE, + control: u32, +) -> Result<(), u32> { + let mut status = unsafe { zeroed::() }; + // SAFETY: the service is open for this one control right. `status` is a writable output of + // the control call and is not used as proof of the final state. + let success = unsafe { ControlService(service, control, &mut status) }; + if success == 0 { + Err(last_error()) + } else { + Ok(()) } - query_service_state(service.as_raw()) } fn query_service_state( service: windows_sys::Win32::System::Services::SC_HANDLE, ) -> Result { + Ok(state_from_scm( + query_service_status(service)?.dwCurrentState, + )) +} + +fn query_service_status( + service: windows_sys::Win32::System::Services::SC_HANDLE, +) -> Result { let mut status = unsafe { zeroed::() }; let mut bytes_needed = 0u32; // SAFETY: the service is open for SERVICE_QUERY_STATUS. The status object is writable and @@ -540,7 +668,7 @@ fn query_service_state( if success == 0 { return Err(last_error()); } - Ok(state_from_scm(status.dwCurrentState)) + Ok(status) } fn open_service( @@ -558,3 +686,133 @@ fn last_error() -> u32 { let error = unsafe { GetLastError() }; if error == 0 { ERROR_GEN_FAILURE } else { error } } + +#[cfg(test)] +mod tests { + use super::*; + use std::cell::Cell; + use windows_sys::Win32::System::Services::{ + SERVICE_RUNNING, SERVICE_STOP_PENDING, SERVICE_STOPPED, + }; + + fn status(state: u32, checkpoint: u32, hint: u32) -> SERVICE_STATUS_PROCESS { + // SAFETY: the SCM status structure contains only integer fields. + let mut value = unsafe { zeroed::() }; + value.dwCurrentState = state; + value.dwCheckPoint = checkpoint; + value.dwWaitHint = hint; + value + } + + #[test] + fn restart_waits_through_old_running_state_and_checkpoint_progress() { + let clock = Cell::new(Duration::ZERO); + let stopped = poll_service_stop( + || { + let ms = clock.get().as_millis(); + Ok(if ms < 200 { + status(SERVICE_RUNNING, 0, 1_000) + } else if ms < 3_500 { + status(SERVICE_STOP_PENDING, (ms / 800) as u32 + 1, 1_000) + } else { + status(SERVICE_STOPPED, 0, 0) + }) + }, + || clock.get(), + |duration| clock.set(clock.get() + duration), + || false, + ); + let starts = Cell::new(0); + assert_eq!( + restart_after_stop(stopped, || { + starts.set(starts.get() + 1); + Ok(ServiceRunState::Running) + }), + Ok(ServiceRunState::Running) + ); + assert_eq!(starts.get(), 1); + assert_eq!(clock.get(), Duration::from_millis(3_500)); + } + + #[test] + fn unchanged_checkpoint_times_out_even_if_wait_hint_grows() { + let clock = Cell::new(Duration::ZERO); + let result = poll_service_stop( + || { + Ok(status( + SERVICE_STOP_PENDING, + 1, + if clock.get().is_zero() { 1_000 } else { 30_000 }, + )) + }, + || clock.get(), + |duration| clock.set(clock.get() + duration), + || false, + ); + assert_eq!(result, Err(ERROR_SERVICE_REQUEST_TIMEOUT)); + assert_eq!(clock.get(), Duration::from_secs(1)); + assert!( + restart_after_stop(result, || panic!("start must not run after a timeout")).is_err() + ); + } + + #[test] + fn continuous_progress_still_obeys_overall_deadline() { + let clock = Cell::new(Duration::ZERO); + let result = poll_service_stop( + || { + Ok(status( + SERVICE_STOP_PENDING, + (clock.get().as_millis() / 100) as u32, + 1_000, + )) + }, + || clock.get(), + |duration| clock.set(clock.get() + duration), + || false, + ); + assert_eq!(result, Err(ERROR_SERVICE_REQUEST_TIMEOUT)); + assert_eq!( + clock.get(), + Duration::from_millis(u64::from(SERVICE_STOP_WAIT_CAP_MS)) + ); + } + + #[test] + fn cancellation_interrupts_large_hint_before_another_query_or_start() { + let stop = WorkerStopSignal::default(); + let queries = Cell::new(0); + let result = poll_service_stop( + || { + queries.set(queries.get() + 1); + Ok(status(SERVICE_STOP_PENDING, 1, u32::MAX)) + }, + || Duration::ZERO, + |duration| { + assert!(duration <= Duration::from_millis(100)); + stop.cancel(); + }, + || stop.is_cancelled(), + ); + assert_eq!(result, Err(ERROR_CANCELLED)); + assert_eq!(queries.get(), 1); + assert_eq!( + restart_after_stop(result, || panic!("cancelled restart must not start")), + Err(ERROR_CANCELLED) + ); + } + + #[test] + fn failed_status_query_is_preserved_without_retry_or_start() { + let result = poll_service_stop( + || Err(5), + || Duration::ZERO, + |_| panic!("failed query must not wait"), + || false, + ); + assert_eq!( + restart_after_stop(result, || panic!("failed stop must not start")), + Err(5) + ); + } +} diff --git a/src/pages/startup/mod.rs b/src/pages/startup/mod.rs index 6cf1847..3b8368a 100644 --- a/src/pages/startup/mod.rs +++ b/src/pages/startup/mod.rs @@ -15,7 +15,10 @@ mod sampler; // 启动项页列出 Run 键和“启动”文件夹中的项目，并在后台 worker 上启用或禁用。 // UI 线程只提交请求、维护列表选择和绘制。刷新保留来源加名称对应的选择、焦点和滚动位置。 // 某个来源枚举失败时保留该来源上一轮的行，并把这些行和页面提示标成未刷新。 +// 启用和禁用按接受顺序进入同一 worker 队列。后到的刷新不会丢掉尚未执行的切换， +// 已在等待的刷新也不会丢掉后到的切换。写入前用页面保存的身份再核对；对不上就跳过并报告原有失败。 // 启用和禁用不删除原项；写入结果读回不一致、项目已消失或 API 失败时立即显示错误。 +// 页面销毁时不在 UI 线程上 join 这个 worker。 use std::mem::zeroed; @@ -159,7 +162,9 @@ impl StartupPageState { } pub fn destroy(&mut self) { - self.worker = None; + if let Some(worker) = self.worker.take() { + worker.defer_drain_until_exit(); + } self.rows.clear(); } @@ -350,6 +355,7 @@ impl StartupPageState { if self.worker.is_some() { return Ok(()); } + // Final teardown closes notifications and drains accepted writes after the UI loop. self.worker = Some(SingleFlightWorker::spawn( "taskmgr-rs-startup-sampler", PWM_STARTUP_WORKER_COMPLETE, @@ -361,7 +367,7 @@ impl StartupPageState { fn refresh(&mut self) { self.drain_worker(); - self.submit(StartupRequest::Refresh); + self.submit(StartupRequest::refresh()); } fn request_toggle(&mut self, toggle: StartupToggle) { @@ -378,14 +384,11 @@ impl StartupPageState { self.beep(); return; } - self.submit(StartupRequest::SetEnabled { id, toggle }); + self.submit(StartupRequest::toggle(id, toggle)); } fn submit(&mut self, request: StartupRequest) { - let toggle = match &request { - StartupRequest::SetEnabled { toggle, .. } => Some(*toggle), - StartupRequest::Refresh => None, - }; + let toggle = request.toggles.first().map(|command| command.toggle); let Some(worker) = self.worker.as_mut() else { self.report_submit_failure(StartupQueryFailure::Win32(ERROR_BROKEN_PIPE), toggle); return; @@ -432,12 +435,17 @@ impl StartupPageState { Ok(collection) => self.apply_collection(collection), Err(failure) => self.apply_refresh_failure(failure), }, - StartupWorkerCompletion::Action(report) => { - match report.outcome { - Ok(state) => self.apply_toggled_state(&report.id, state), - Err(failure) => self.show_action_failure(report.toggle, failure), + StartupWorkerCompletion::Actions(reports) => { + if reports.is_empty() { + return; + } + for report in reports { + match report.outcome { + Ok(state) => self.apply_toggled_state(&report.id, state), + Err(failure) => self.show_action_failure(report.toggle, failure), + } } - self.submit(StartupRequest::Refresh); + self.submit(StartupRequest::refresh()); } } } diff --git a/src/pages/startup/model.rs b/src/pages/startup/model.rs index 9c670ef..b9e2195 100644 --- a/src/pages/startup/model.rs +++ b/src/pages/startup/model.rs @@ -134,12 +134,35 @@ impl StartupToggle { } #[derive(Clone, Debug, PartialEq, Eq)] -pub(crate) enum StartupRequest { - Refresh, - SetEnabled { - id: StartupEntryId, - toggle: StartupToggle, - }, +pub(crate) struct StartupToggleCommand { + pub(crate) id: StartupEntryId, + pub(crate) toggle: StartupToggle, +} + +/// One coalesced worker request. +/// +/// Enable and disable commands stay in [`StartupRequest::toggles`] in acceptance order. +/// `refresh` records that a snapshot is also wanted. Neither field replaces the other. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct StartupRequest { + pub(crate) toggles: Vec, + pub(crate) refresh: bool, +} + +impl StartupRequest { + pub(crate) fn refresh() -> Self { + Self { + toggles: Vec::new(), + refresh: true, + } + } + + pub(crate) fn toggle(id: StartupEntryId, toggle: StartupToggle) -> Self { + Self { + toggles: vec![StartupToggleCommand { id, toggle }], + refresh: false, + } + } } #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] @@ -271,6 +294,22 @@ pub(crate) fn classify_startup_result( } } +/// `stored` is the identity the page already kept for this command. +/// +/// The worker reads the live entry immediately before the registry or startup-folder write and +/// passes that entry here. A missing entry, or one that is no longer the same source and name, +/// is [`StartupActionFailure::Missing`]. The caller skips the write. +pub(crate) fn confirm_startup_identity( + stored: &StartupEntryId, + observed: Option<&StartupEntryId>, +) -> Result<(), StartupActionFailure> { + if observed.is_some_and(|observed| stored.matches(observed)) { + Ok(()) + } else { + Err(StartupActionFailure::Missing) + } +} + pub(crate) fn command_state(selected: Option) -> StartupCommandState { let Some(state) = selected else { return StartupCommandState::disabled(); @@ -281,12 +320,11 @@ pub(crate) fn command_state(selected: Option) -> StartupCom } } -/// Keeps a pending enable or disable ahead of a coalesced refresh. +/// Appends every accepted enable or disable. A refresh does not erase that queue, and a toggle +/// is not dropped when a refresh is already pending. pub(crate) fn merge_startup_requests(pending: &mut StartupRequest, incoming: StartupRequest) { - match (&*pending, &incoming) { - (StartupRequest::SetEnabled { .. }, StartupRequest::Refresh) => {} - _ => *pending = incoming, - } + pending.toggles.extend(incoming.toggles); + pending.refresh = pending.refresh || incoming.refresh; } pub(crate) const fn refresh_scroll_policy() -> StartupScrollPolicy { @@ -616,30 +654,91 @@ mod tests { ); } - #[test] - fn pending_toggle_survives_a_coalesced_refresh() { - let original = StartupRequest::SetEnabled { - id: id(StartupSource::CurrentUserRun, "OneDrive"), - toggle: StartupToggle::Disable, - }; - let mut pending = original.clone(); - merge_startup_requests(&mut pending, StartupRequest::Refresh); - assert_eq!(pending, original); + fn queued_toggle( + source: StartupSource, + name: &str, + toggle: StartupToggle, + ) -> StartupToggleCommand { + StartupToggleCommand { + id: id(source, name), + toggle, + } + } + #[test] + fn two_pending_toggles_survive_a_merge_and_a_refresh() { + let first = queued_toggle( + StartupSource::CurrentUserRun, + "OneDrive", + StartupToggle::Disable, + ); + let second = queued_toggle( + StartupSource::LocalMachineRun, + "SecurityHealth", + StartupToggle::Enable, + ); + let mut pending = StartupRequest::toggle(first.id.clone(), first.toggle); merge_startup_requests( &mut pending, - StartupRequest::SetEnabled { - id: id(StartupSource::LocalMachineRun, "SecurityHealth"), - toggle: StartupToggle::Enable, - }, + StartupRequest::toggle(second.id.clone(), second.toggle), ); - assert!(matches!( + assert_eq!( pending, - StartupRequest::SetEnabled { - toggle: StartupToggle::Enable, - .. + StartupRequest { + toggles: vec![first.clone(), second.clone()], + refresh: false, } - )); + ); + + merge_startup_requests(&mut pending, StartupRequest::refresh()); + assert_eq!( + pending, + StartupRequest { + toggles: vec![first.clone(), second], + refresh: true, + } + ); + + let mut refresh_first = StartupRequest::refresh(); + merge_startup_requests( + &mut refresh_first, + StartupRequest::toggle(first.id.clone(), first.toggle), + ); + assert_eq!( + refresh_first, + StartupRequest { + toggles: vec![first], + refresh: true, + } + ); + } + + #[test] + fn a_changed_startup_identity_is_the_existing_missing_failure() { + let stored = id(StartupSource::CurrentUserRun, "OneDrive"); + assert_eq!(confirm_startup_identity(&stored, Some(&stored)), Ok(())); + assert_eq!( + confirm_startup_identity( + &stored, + Some(&id(StartupSource::CurrentUserRun, "onedrive")) + ), + Ok(()) + ); + assert_eq!( + confirm_startup_identity(&stored, None), + Err(StartupActionFailure::Missing) + ); + assert_eq!( + confirm_startup_identity( + &stored, + Some(&id(StartupSource::LocalMachineRun, "OneDrive")) + ), + Err(StartupActionFailure::Missing) + ); + assert_eq!( + confirm_startup_identity(&stored, Some(&id(StartupSource::CurrentUserRun, "Teams"))), + Err(StartupActionFailure::Missing) + ); } #[test] diff --git a/src/pages/startup/sampler.rs b/src/pages/startup/sampler.rs index 5d2e30c..1d85a1f 100644 --- a/src/pages/startup/sampler.rs +++ b/src/pages/startup/sampler.rs @@ -14,8 +14,10 @@ //! The page worker owns every registry key and find handle. Enumeration reads the native //! 64-bit registry view, including the explicit WOW6432Node Run key. Enable and disable rewrite //! only the 12-byte StartupApproved value and then read it back, keeping the 2/3 family distinct -//! from the 6/7 family. The Run value and the Startup folder file are checked before and after -//! that write so a missing entry is never reported as a successful change. +//! from the 6/7 family. Queued commands run in acceptance order. Immediately before each write, +//! the Run value or startup file is read again and compared with the identity the page stored. +//! A missing or different entry skips that command and is [`StartupActionFailure::Missing`]. +//! The entry is checked again after the write so a disappearance is never a successful change. //! //! A source that fails to enumerate is listed in `omitted_sources` and contributes no rows. //! The page keeps the previous rows for that source. The snapshot is an error only when every @@ -51,9 +53,9 @@ use crate::infrastructure::native::{to_wide_null, widestr_ptr_to_string}; use super::model::{ StartupActionFailure, StartupEnabledState, StartupEntryId, StartupSource, StartupToggle, - approved_blob, approved_flag, classify_startup_result, executable_candidate, - has_version_resource_extension, is_absolute_win32_path, registry_value_name_is_safe, - startup_file_name_is_safe, state_from_approved, + approved_blob, approved_flag, classify_startup_result, confirm_startup_identity, + executable_candidate, has_version_resource_extension, is_absolute_win32_path, + registry_value_name_is_safe, startup_file_name_is_safe, state_from_approved, }; const VALUE_ENUM_LIMIT: u32 = 4_096; @@ -126,7 +128,7 @@ pub(super) struct StartupRefreshFailure { #[derive(Clone, Debug, PartialEq, Eq)] pub(super) enum StartupWorkerCompletion { Snapshot(Result), - Action(StartupActionReport), + Actions(Vec), } struct RegKey { @@ -185,14 +187,22 @@ impl SourcePart { } pub(super) fn perform(request: super::model::StartupRequest) -> StartupWorkerCompletion { - match request { - super::model::StartupRequest::Refresh => { - StartupWorkerCompletion::Snapshot(collect_startup_entries()) - } - super::model::StartupRequest::SetEnabled { id, toggle } => { - StartupWorkerCompletion::Action(set_enabled(&id, toggle)) - } - } + // Commands stay ahead of a coalesced snapshot. Each one re-checks the stored identity + // before it writes. The page requests the snapshot after these reports, so a refresh that + // arrived with the queue still runs, and the queue is not held inside enumeration. + if !request.toggles.is_empty() { + return StartupWorkerCompletion::Actions( + request + .toggles + .into_iter() + .map(|command| set_enabled(&command.id, command.toggle)) + .collect(), + ); + } + if request.refresh { + return StartupWorkerCompletion::Snapshot(collect_startup_entries()); + } + StartupWorkerCompletion::Actions(Vec::new()) } fn collect_startup_entries() -> Result { @@ -424,9 +434,8 @@ fn set_registry_enabled( if !registry_value_name_is_safe(&id.name) { return Err(StartupActionFailure::Win32(ERROR_INVALID_PARAMETER)); } - if !registry_entry_exists(id)? { - return Err(StartupActionFailure::Missing); - } + // The page stored this source and name. Re-read the Run value immediately before the write. + confirm_startup_identity(id, registry_entry_exists(id)?.then_some(id))?; let observed = write_and_read_approved(id, toggle)?; if !registry_entry_exists(id)? { return Err(StartupActionFailure::Missing); @@ -442,9 +451,8 @@ fn set_folder_enabled( return Err(StartupActionFailure::Win32(ERROR_INVALID_PARAMETER)); } let path = startup_file_path(id)?; - if !startup_file_exists(&path)? { - return Err(StartupActionFailure::Missing); - } + // The page stored this source and name. Re-read the startup file immediately before the write. + confirm_startup_identity(id, startup_file_exists(&path)?.then_some(id))?; let observed = write_and_read_approved(id, toggle)?; if !startup_file_exists(&path)? { return Err(StartupActionFailure::Missing); diff --git a/src/pages/users.rs b/src/pages/users.rs index eaf04a1..7c9bc36 100644 --- a/src/pages/users.rs +++ b/src/pages/users.rs @@ -155,6 +155,9 @@ pub struct UserPageState { sort_column: usize, sort_ascending: bool, worker: Option>, + // Session commands accepted by the worker and not yet completed. A repeat of one of + // these is not submitted again while that worker is still busy. + outstanding_session_actions: Vec, last_refresh_error: Option, last_row_errors: Vec, } @@ -221,6 +224,7 @@ impl UserPageState { fn stop_worker_thread(&mut self) { self.worker = None; + self.outstanding_session_actions.clear(); } pub fn size_page(&self) { @@ -442,6 +446,7 @@ impl UserPageState { }); } if let Some(error) = drained.error { + self.outstanding_session_actions.clear(); self.set_refresh_error(error); } } @@ -450,6 +455,10 @@ impl UserPageState { match completion { UserWorkerCompletion::Snapshot(result) => self.apply_user_worker_result(result), UserWorkerCompletion::Actions(reports) => { + finish_outstanding_session_actions( + &mut self.outstanding_session_actions, + reports.len(), + ); let mut refresh = false; for report in reports { if let Some(error) = report.error { @@ -956,7 +965,19 @@ impl UserPageState { fn submit_session_command(&mut self, command: UserSessionCommand) { let kind = command.kind; + let worker_busy = self + .worker + .as_ref() + .is_some_and(|worker| worker.is_in_flight() || worker.has_pending()); + if !queue_session_action( + &mut self.outstanding_session_actions, + command.clone(), + worker_busy, + ) { + return; + } if let Err(error) = self.submit_request(UserWorkerRequest::Actions(vec![command])) { + self.outstanding_session_actions.pop(); self.show_action_error(kind, error); } } @@ -1029,17 +1050,55 @@ enum UserWorkerCompletion { } /// Keeps a confirmed session action ahead of a coalesced refresh, and queues a later action -/// behind one that has not started. +/// behind one that has not started. Duplicate disconnect/logoff commands are coalesced; +/// every confirmed message is retained, including an intentional repeated message. fn merge_user_requests(pending: &mut UserWorkerRequest, incoming: UserWorkerRequest) { match (&mut *pending, incoming) { (UserWorkerRequest::Actions(queued), UserWorkerRequest::Actions(more)) => { - queued.extend(more); + for command in more { + if session_action_already_queued(queued, &command) { + continue; + } + queued.push(command); + } } (UserWorkerRequest::Actions(_), UserWorkerRequest::Refresh) => {} (pending, incoming) => *pending = incoming, } } +fn session_action_already_queued( + queued: &[UserSessionCommand], + command: &UserSessionCommand, +) -> bool { + command.kind != UserSessionKind::SendMessage + && queued + .iter() + .any(|existing| existing.kind == command.kind && existing.sessions == command.sessions) +} + +/// Records a session command unless the worker is busy with the same kind and identities. +/// Returns whether the caller should submit it. An idle worker drops stale outstanding +/// commands so a later action for that session is not blocked. +fn queue_session_action( + outstanding: &mut Vec, + command: UserSessionCommand, + worker_busy: bool, +) -> bool { + if !worker_busy { + outstanding.clear(); + } else if session_action_already_queued(outstanding, &command) { + return false; + } + outstanding.push(command); + true +} + +fn finish_outstanding_session_actions(outstanding: &mut Vec, finished: usize) { + let finished = finished.min(outstanding.len()); + outstanding.drain(..finished).for_each(drop); +} + fn perform_user_request(request: UserWorkerRequest) -> UserWorkerCompletion { match request { UserWorkerRequest::Refresh => UserWorkerCompletion::Snapshot(collect_user_sessions()), @@ -1772,6 +1831,67 @@ mod tests { assert_eq!(pending, action); } + #[test] + fn identical_session_actions_collapse_and_different_sessions_remain() { + let first = session_command(UserSessionKind::Logoff); + let mut other_session = session_command(UserSessionKind::Logoff); + other_session.sessions = vec![identity(2, 100)]; + let mut other_logon = session_command(UserSessionKind::Logoff); + other_logon.sessions = vec![identity(1, 200)]; + let mut pending = UserWorkerRequest::Actions(vec![first.clone()]); + + merge_user_requests( + &mut pending, + UserWorkerRequest::Actions(vec![first.clone(), other_session.clone()]), + ); + assert_eq!( + pending, + UserWorkerRequest::Actions(vec![first.clone(), other_session.clone()]) + ); + + merge_user_requests( + &mut pending, + UserWorkerRequest::Actions(vec![other_logon.clone()]), + ); + assert_eq!( + pending, + UserWorkerRequest::Actions(vec![first, other_session, other_logon]) + ); + } + + #[test] + fn repeated_session_action_is_not_submitted_while_the_worker_is_busy() { + let first = session_command(UserSessionKind::Disconnect); + let mut other_session = session_command(UserSessionKind::Disconnect); + other_session.sessions = vec![identity(2, 100)]; + let mut other_logon = session_command(UserSessionKind::Disconnect); + other_logon.sessions = vec![identity(1, 200)]; + let mut outstanding = Vec::new(); + + assert!(queue_session_action(&mut outstanding, first.clone(), false)); + assert!(!queue_session_action(&mut outstanding, first.clone(), true)); + assert!(queue_session_action( + &mut outstanding, + other_session.clone(), + true + )); + assert!(queue_session_action( + &mut outstanding, + other_logon.clone(), + true + )); + assert_eq!( + outstanding, + vec![first.clone(), other_session.clone(), other_logon.clone()] + ); + + finish_outstanding_session_actions(&mut outstanding, 1); + assert_eq!(outstanding, vec![other_session, other_logon]); + + assert!(queue_session_action(&mut outstanding, first.clone(), false)); + assert_eq!(outstanding, vec![first]); + } + #[test] fn session_action_replaces_a_pending_refresh() { let action = UserWorkerRequest::Actions(vec![session_command(UserSessionKind::Logoff)]); @@ -1780,6 +1900,34 @@ mod tests { assert_eq!(pending, action); } + #[test] + fn every_confirmed_message_survives_busy_and_pending_queues() { + let mut first = session_command(UserSessionKind::SendMessage); + first.title = "First".into(); + first.body = "One".into(); + let mut second = first.clone(); + second.title = "Second".into(); + second.body = "Two".into(); + let commands = vec![first.clone(), second, first]; + let mut outstanding = Vec::new(); + let mut pending = UserWorkerRequest::Refresh; + for command in &commands { + assert!(queue_session_action( + &mut outstanding, + command.clone(), + true + )); + merge_user_requests( + &mut pending, + UserWorkerRequest::Actions(vec![command.clone()]), + ); + } + assert_eq!(outstanding, commands); + assert_eq!(pending, UserWorkerRequest::Actions(commands)); + finish_outstanding_session_actions(&mut outstanding, 3); + assert!(outstanding.is_empty()); + } + #[test] fn a_second_session_action_stays_queued() { let first = session_command(UserSessionKind::Disconnect); diff --git a/src/system/cpu_sampler.rs b/src/system/cpu_sampler.rs index 0e6e724..40def28 100644 --- a/src/system/cpu_sampler.rs +++ b/src/system/cpu_sampler.rs @@ -61,23 +61,69 @@ pub(crate) fn query_processor_performance( ) }; - if status >= 0 { - if returned != 0 { - let returned = returned as usize; - if !returned.is_multiple_of(item_size) || returned > byte_len as usize { - return Err(STATUS_INVALID_BUFFER_SIZE); - } - if returned / item_size != count { - return Err(STATUS_INFO_LENGTH_MISMATCH); - } - } - return Ok(()); + match take_processor_performance_query(output, status, returned, byte_len, count) { + Ok(ProcessorPerformanceQuery::Complete) => return Ok(()), + Ok(ProcessorPerformanceQuery::GrowTo(next_count)) => count = next_count, + Err(error) => return Err(error), } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ProcessorPerformanceQuery { + Complete, + GrowTo(usize), +} - if status != STATUS_INFO_LENGTH_MISMATCH || returned as usize <= byte_len as usize { - return Err(status); +/// Classifies one processor-performance query. +/// +/// A success status with a zero returned length is an error. A nonzero length is +/// accepted only when it is an exact multiple of the struct and matches `count`. +fn decide_processor_performance_query( + status: i32, + returned: u32, + buffer_bytes: u32, + count: usize, +) -> Result { + let item_size = size_of::(); + if status >= 0 { + if returned == 0 { + return Err(STATUS_INVALID_BUFFER_SIZE); + } + let returned_len = returned as usize; + let buffer_len = buffer_bytes as usize; + if !returned_len.is_multiple_of(item_size) || returned_len > buffer_len { + return Err(STATUS_INVALID_BUFFER_SIZE); } - count = (returned as usize).div_ceil(item_size); + if returned_len / item_size != count { + return Err(STATUS_INFO_LENGTH_MISMATCH); + } + return Ok(ProcessorPerformanceQuery::Complete); + } + + if status != STATUS_INFO_LENGTH_MISMATCH || returned as usize <= buffer_bytes as usize { + return Err(status); + } + Ok(ProcessorPerformanceQuery::GrowTo( + (returned as usize).div_ceil(item_size), + )) +} + +fn take_processor_performance_query( + output: &mut Vec, + status: i32, + returned: u32, + buffer_bytes: u32, + count: usize, +) -> Result { + match decide_processor_performance_query(status, returned, buffer_bytes, count) { + // A successful query that wrote nothing leaves the caller's default-zero + // rows in place. Drop them so they cannot be read as idle counters. + Err(error) if status >= 0 && returned == 0 => { + output.clear(); + Err(error) + } + decision => decision, } } @@ -100,8 +146,13 @@ pub(crate) fn checked_summed_processor_times( #[cfg(test)] mod tests { + use std::mem::size_of; + use super::{ - ProcessorPerformance, checked_summed_processor_times, query_processor_performance, + ProcessorPerformance, ProcessorPerformanceQuery, STATUS_INFO_LENGTH_MISMATCH, + STATUS_INVALID_BUFFER_SIZE, checked_summed_processor_times, + decide_processor_performance_query, query_processor_performance, + take_processor_performance_query, }; use windows_sys::Win32::System::Threading::{ALL_PROCESSOR_GROUPS, GetActiveProcessorCount}; @@ -155,4 +206,96 @@ mod tests { Some((15, 70, 27)) ); } + + #[test] + fn success_with_zero_returned_length_is_an_error() { + let bytes = size_of::() as u32; + assert!(decide_processor_performance_query(0, 0, bytes, 1).is_err()); + assert!(decide_processor_performance_query(0, 0, bytes, 0).is_err()); + } + + #[test] + fn success_with_exact_multiple_is_ok() { + let item_size = size_of::(); + let count = 3; + let bytes = u32::try_from(item_size * count).unwrap(); + assert_eq!( + decide_processor_performance_query(0, bytes, bytes, count), + Ok(ProcessorPerformanceQuery::Complete) + ); + } + + #[test] + fn non_multiple_returned_length_is_an_error() { + let bytes = size_of::() as u32; + assert_eq!( + decide_processor_performance_query(0, bytes - 1, bytes, 1), + Err(STATUS_INVALID_BUFFER_SIZE) + ); + } + + #[test] + fn short_buffer_keeps_invalid_buffer_size() { + let one = size_of::() as u32; + assert_eq!( + decide_processor_performance_query(0, one * 2, one, 2), + Err(STATUS_INVALID_BUFFER_SIZE) + ); + } + + #[test] + fn count_mismatch_keeps_length_mismatch() { + let one = size_of::() as u32; + let bytes = one * 2; + assert_eq!( + decide_processor_performance_query(0, bytes, bytes, 1), + Err(STATUS_INFO_LENGTH_MISMATCH) + ); + } + + #[test] + fn length_mismatch_grows_when_the_kernel_reports_a_larger_buffer() { + let one = size_of::() as u32; + assert_eq!( + decide_processor_performance_query(STATUS_INFO_LENGTH_MISMATCH, one * 3, one, 1), + Ok(ProcessorPerformanceQuery::GrowTo(3)) + ); + } + + #[test] + fn length_mismatch_without_a_larger_return_keeps_the_status() { + let bytes = size_of::() as u32; + assert_eq!( + decide_processor_performance_query(STATUS_INFO_LENGTH_MISMATCH, bytes, bytes, 1), + Err(STATUS_INFO_LENGTH_MISMATCH) + ); + } + + #[test] + fn zero_returned_length_drops_default_processor_times() { + let mut processors = vec![ProcessorPerformance::default(); 2]; + let bytes = size_of::() as u32; + assert!(take_processor_performance_query(&mut processors, 0, 0, bytes, 2).is_err()); + assert!(processors.is_empty()); + } + + #[test] + fn failed_zero_return_keeps_existing_rows() { + let mut processors = vec![ProcessorPerformance { + idle_time: 4, + ..ProcessorPerformance::default() + }]; + let bytes = size_of::() as u32; + assert_eq!( + take_processor_performance_query( + &mut processors, + STATUS_INFO_LENGTH_MISMATCH, + 0, + bytes, + 1 + ), + Err(STATUS_INFO_LENGTH_MISMATCH) + ); + assert_eq!(processors[0].idle_time, 4); + } } diff --git a/src/system/cpu_topology.rs b/src/system/cpu_topology.rs index a54508e..a070197 100644 --- a/src/system/cpu_topology.rs +++ b/src/system/cpu_topology.rs @@ -951,6 +951,35 @@ mod tests { ); } + #[test] + fn pointer_width_group_builds_one_bit_per_processor() { + let count = usize::BITS as usize; + let ids = (0..usize::BITS) + .map(|number| id(0, number as u8)) + .collect::>(); + let records = (0..usize::BITS) + .map(|number| { + // A shift of `usize::BITS` is out of range for this pointer width. + let mask = if number < usize::BITS { + 1usize << number + } else { + 0 + }; + record(0, &[(0, mask)]) + }) + .collect::>(); + + let topology = build_topology(&ids, &[usize::BITS], records).unwrap(); + assert!(topology.matches_sample_len(count)); + } + + #[test] + fn active_count_above_pointer_width_is_out_of_range() { + let active_count = usize::BITS + 1; + let topology = build_topology(&[id(0, 0)], &[active_count], vec![record(0, &[(0, 1)])]); + assert_eq!(topology, Err(CpuTopologyError::OutOfRangeAffinityMask)); + } + #[test] fn malformed_variable_records_are_rejected() { let valid = encoded_record(0, &[(0, 1)]); diff --git a/src/system/disk.rs b/src/system/disk.rs index 432abff..00dbe4f 100644 --- a/src/system/disk.rs +++ b/src/system/disk.rs @@ -22,7 +22,8 @@ //! byte and timing counters stay on the worker. A failed read leaves those //! baselines in place. The first sample, a counter reset, and a rate that does //! not fit in `u64` replace the baseline without publishing a point. A zero -//! query interval publishes nothing and keeps the previous baseline. +//! query interval publishes nothing and stores the current counters, so bytes +//! that moved while the clock was frozen are not counted on the next real tick. use std::mem::{offset_of, size_of, zeroed}; use std::ptr::{null, null_mut}; @@ -482,7 +483,9 @@ pub(crate) fn commit_disk_readings( active_percent = active_percent.max(rate.active_percent); rate_indexes.push(index); } - DiskDelta::ZeroInterval => {} + DiskDelta::ZeroInterval => { + baselines[index] = Some(currents[index]); + } DiskDelta::Baseline | DiskDelta::Reset | DiskDelta::Overflow => { replace_baselines(baselines, ¤ts); return DiskActivityUpdate::NoRate; @@ -688,7 +691,20 @@ mod tests { } #[test] - fn zero_interval_keeps_the_baseline_and_counts_bytes_later() { + fn zero_interval_advances_the_baseline_without_publishing_a_rate() { + let previous = counters(0, 0, 0, 0, 1, 1_000); + let mut baselines = [Some(previous), Some(previous)]; + let first = counters(1_000, 250, 0, 0, 1, 1_000); + let second = counters(40, 80, 0, 0, 1, 1_000); + assert_eq!( + commit_disk_readings(&mut baselines, &[Ok(first), Ok(second)]), + DiskActivityUpdate::NoRate + ); + assert_eq!(baselines, [Some(first), Some(second)]); + } + + #[test] + fn zero_interval_bytes_are_excluded_from_the_next_real_interval() { let previous = counters(0, 0, 0, 0, 1, 1_000); let mut baselines = [Some(previous)]; let frozen = counters(1_000, 250, 0, 0, 1, 1_000); @@ -696,17 +712,41 @@ mod tests { commit_disk_readings(&mut baselines, &[Ok(frozen)]), DiskActivityUpdate::NoRate ); - assert_eq!(baselines, [Some(previous)]); + assert_eq!(baselines, [Some(frozen)]); - let next = counters(1_000, 250, 0, 0, 1 + 10_000_000, 1_000 + 10_000_000); + let next = counters( + 1_000 + 5_000_000, + 250 + 2_000_000, + 0, + 0, + 1 + 10_000_000, + 1_000 + 10_000_000, + ); assert_eq!( commit_disk_readings(&mut baselines, &[Ok(next)]), DiskActivityUpdate::Rate(DiskRate { - read_bytes_per_second: 1_000, - write_bytes_per_second: 250, + read_bytes_per_second: 5_000_000, + write_bytes_per_second: 2_000_000, + active_percent: 0, + }) + ); + } + + #[test] + fn zero_interval_bytes_are_omitted_from_a_summed_rate() { + let previous = counters(0, 0, 0, 0, 1, 1_000); + let mut baselines = [Some(previous), Some(previous)]; + let frozen = counters(4_000, 1_000, 0, 0, 1, 1_000); + let moving = counters(3_000_000, 500_000, 0, 0, 1 + 10_000_000, 1_000 + 10_000_000); + assert_eq!( + commit_disk_readings(&mut baselines, &[Ok(frozen), Ok(moving)]), + DiskActivityUpdate::Rate(DiskRate { + read_bytes_per_second: 3_000_000, + write_bytes_per_second: 500_000, active_percent: 0, }) ); + assert_eq!(baselines, [Some(frozen), Some(moving)]); } #[test] diff --git a/src/system/sampler.rs b/src/system/sampler.rs index 66c9b36..cbb08c3 100644 --- a/src/system/sampler.rs +++ b/src/system/sampler.rs @@ -377,6 +377,12 @@ fn calculate_cpu_delta( if current.is_empty() || current.len() != previous.len() { return delta; } + // Detail-counter failures must not skip this check. A missing interval is not a + // measurement, so its processor percents stay unpublished. + let Some(elapsed_ms) = elapsed_ms.filter(|elapsed| *elapsed != 0) else { + delta.diagnostics = Err(CpuDiagnosticError::InvalidElapsedTime); + return delta; + }; let mut idle_sum = 0u128; let mut kernel_sum = 0u128; @@ -471,10 +477,6 @@ fn calculate_cpu_delta( delta.diagnostics = Err(error); return delta; } - let Some(elapsed_ms) = elapsed_ms.filter(|elapsed| *elapsed != 0) else { - delta.diagnostics = Err(CpuDiagnosticError::InvalidElapsedTime); - return delta; - }; let (Some(user_usage), Some(dpc_usage), Some(interrupt_usage)) = ( checked_percent_u128(user_sum, total_sum), checked_percent_u128(dpc_sum, total_sum), @@ -646,13 +648,55 @@ mod tests { let previous = [times(100, 100, 100, 10, 5, 100)]; let current = [times(120, 140, 140, 18, 8, 120)]; let delta = calculate_cpu_delta(¤t, &previous, Some(0)); - assert!(delta.valid); + assert!(!delta.valid); + assert_eq!(delta.cpu_usage, 0); + assert_eq!(delta.processor_cpu_usage, vec![0]); + assert_eq!(delta.processor_kernel_usage, vec![0]); assert_eq!( delta.diagnostics, Err(CpuDiagnosticError::InvalidElapsedTime) ); } + #[test] + fn zero_elapsed_with_regressed_detail_is_not_valid_but_idle_zero_percent_is() { + let previous = [times(100, 100, 100, 10, 5, 100)]; + let regressed = [times(120, 140, 140, 18, 8, 90)]; + let rejected = calculate_cpu_delta(®ressed, &previous, Some(0)); + assert!(!rejected.valid); + assert_eq!(rejected.cpu_usage, 0); + assert_eq!(rejected.processor_cpu_usage, vec![0]); + assert_eq!(rejected.processor_kernel_usage, vec![0]); + assert_eq!( + rejected.diagnostics, + Err(CpuDiagnosticError::InvalidElapsedTime) + ); + + let missing = calculate_cpu_delta(®ressed, &previous, None); + assert!(!missing.valid); + assert_eq!( + missing.diagnostics, + Err(CpuDiagnosticError::InvalidElapsedTime) + ); + + let idle = [times(200, 100, 100, 10, 5, 100)]; + let later = calculate_cpu_delta(&idle, &previous, Some(1000)); + assert!(later.valid); + assert_eq!(later.cpu_usage, 0); + assert_eq!(later.kernel_usage, 0); + assert_eq!(later.processor_cpu_usage, vec![0]); + assert_eq!( + later.diagnostics, + Ok(CpuDiagnosticSample { + user_usage: 0, + kernel_usage: 0, + dpc_usage: 0, + interrupt_usage: 0, + interrupts_per_second: 0, + }) + ); + } + #[test] fn page_conversion_reports_overflow() { assert_eq!( diff --git a/src/ui/localization/text_key.rs b/src/ui/localization/text_key.rs index 1b0fc7d..8fa7147 100644 --- a/src/ui/localization/text_key.rs +++ b/src/ui/localization/text_key.rs @@ -338,6 +338,7 @@ pub enum TextKey { ServicesMenu, ServiceStart, ServiceStop, + ServiceRestart, ServicePause, ServiceContinue, ServiceColumnName, @@ -357,6 +358,7 @@ pub enum TextKey { ServiceStartupDisabled, ServiceStartFailed, ServiceStopFailed, + ServiceRestartFailed, ServicePauseFailed, ServiceContinueFailed, ServiceStateUnchanged, diff --git a/src/ui/menus.rs b/src/ui/menus.rs index 541ae10..650eb64 100644 --- a/src/ui/menus.rs +++ b/src/ui/menus.rs @@ -324,6 +324,7 @@ fn build_service_command_menu() -> MenuResult { let mut menu = PopupMenu::new()?; append_item(&mut menu, IDM_SERVICE_START, TextKey::ServiceStart)?; append_item(&mut menu, IDM_SERVICE_STOP, TextKey::ServiceStop)?; + append_item(&mut menu, IDM_SERVICE_RESTART, TextKey::ServiceRestart)?; append_item(&mut menu, IDM_SERVICE_PAUSE, TextKey::ServicePause)?; append_item(&mut menu, IDM_SERVICE_CONTINUE, TextKey::ServiceContinue)?; append_separator(&mut menu)?; diff --git a/src/ui/resource_ids.rs b/src/ui/resource_ids.rs index e4eb2db..9489400 100644 --- a/src/ui/resource_ids.rs +++ b/src/ui/resource_ids.rs @@ -396,6 +396,7 @@ pub const IDM_SERVICE_STOP: u16 = 40121; pub const IDM_SERVICE_GOTOPROCESS: u16 = 40122; pub const IDM_SERVICE_PAUSE: u16 = 40123; pub const IDM_SERVICE_CONTINUE: u16 = 40124; +pub const IDM_SERVICE_RESTART: u16 = 40125; pub const IDM_PROC_TREE: u16 = 40130; pub const IDM_PROC_EXPAND: u16 = 40131; pub const IDM_PROC_COLLAPSE: u16 = 40132;