From 677fed21b2814a2798b6a144a009a5912b658cb4 Mon Sep 17 00:00:00 2001 From: Stuart Cameron Date: Sun, 9 Aug 2026 00:34:07 +1000 Subject: [PATCH] fix(preview): stop leaking vspipe/ffmpeg when seeking or cancelling Reported: seeking previews and cancelling jobs leave vspipe and ffmpeg running. Three separate causes, all of which rely on the same accident. Nothing ever killed those children deliberately. They tended to die because their pipes closed with the worker and they took EPIPE at the next write -- but a child blocked reading a slow source (a NAS share, in the report) writes nothing for minutes, never notices, and keeps burning CPU on work nobody wants. That also explains why this never reproduced locally: with fast local I/O the cascade wins every time. 1. Preview mode never installed a signal handler at all -- main() returns at the --preview branch before ctrlc is set up -- so SIGTERM killed the worker outright without unwinding. Drop never ran. And generate_preview holds vspipe and ffmpeg in locals, so PipelineExecutor::terminate() could not have reached them even if it had run. 2. PreviewGenerator tracked one _previewProcess, assigned *after* `await Process.start(...)` returned. A seek arriving inside that window cancelled whatever the field happened to hold, and the next assignment then overwrote the reference to the in-flight worker -- untracked, never killed. Scrubbing cancels a preview on every movement, so those accumulate. This is the "bunch of processes" in the report. 3. cancel() signalled the worker's pid alone, so even a clean shutdown depended on the worker getting far enough to kill its own children. The worker now makes itself a process-group leader (setpgid), so the app can tear down the whole tree with one signal to -pid, and ProcessTree does that with a fallback to pid-only signalling where groups are unavailable -- which is exactly today's behaviour, so this is never worse. PreviewGenerator tracks every live preview in a set, registered at spawn, so nothing can be lost in that window. Cancellation signals immediately and reaps in the background: a seek must not wait out a shutdown grace, or scrubbing feels broken. dispose() uses the waiting variant, since strays outliving the app are worse than a pause. Verified rather than assumed. The worker really does become a group leader (pgrp 3362 -> 3382), Dart really does forward a negative pid to kill(2) (killPid(-pid) returns true), and the group signal kills a child that was SIGSTOPped first -- so it cannot be credited to the EPIPE cascade. Tests: a process-group invariant test that stops the children before signalling, and a rapid-seek test that fires ten overlapping previews and asserts nothing survives. Both would pass vacuously without the SIGSTOP and the burst respectively, which is why they are written that way. --- app/lib/services/preview_generator.dart | 87 ++++++++- app/lib/services/process_tree.dart | 66 +++++++ app/lib/services/worker_manager.dart | 9 +- app/test/integration_cancel_test.dart | 233 ++++++++++++++++++++++++ worker/src/main.rs | 25 +++ 5 files changed, 408 insertions(+), 12 deletions(-) create mode 100644 app/lib/services/process_tree.dart diff --git a/app/lib/services/preview_generator.dart b/app/lib/services/preview_generator.dart index 5936286..d008601 100644 --- a/app/lib/services/preview_generator.dart +++ b/app/lib/services/preview_generator.dart @@ -11,12 +11,27 @@ import '../models/processing_pipeline.dart'; import '../models/video_job.dart'; import 'field_order_detector.dart'; import 'temp_directory_service.dart'; +import 'process_tree.dart'; import 'tool_locator.dart'; /// Service for generating video thumbnails and processed previews. class PreviewGenerator { Process? _thumbnailProcess; Process? _previewProcess; + + /// Every preview worker still running, including ones already superseded. + /// + /// `_previewProcess` alone is not enough to guarantee cleanup. It is assigned + /// *after* `await Process.start(...)` returns, so a seek that arrives inside + /// that window cancels whatever the field held at the time — not the process + /// currently being spawned — and the next assignment then overwrites the + /// reference to it. That process is never tracked and never killed. Scrubbing + /// the timeline cancels a preview on every move, so those strays accumulate, + /// which is what "a bunch of vspipe and ffmpeg processes" looks like. + /// + /// Registering at spawn and discarding on exit closes that window: a process + /// is reachable for cancellation from the moment it exists. + final Set _livePreviews = {}; String? _ffmpegPath; String? _ffprobePath; String? _workerPath; @@ -283,10 +298,14 @@ class PreviewGenerator { workingDirectory: path.dirname(_workerPath!), ); _previewProcess = process; + _livePreviews.add(process); if (cancelToken?.isCancelled ?? false) { - process.kill(); + // Whole group: killing the worker alone strands vspipe/ffmpeg. + ProcessTree.killTree(process); _previewProcess = null; + _livePreviews.remove(process); + await ProcessTree.waitForExit(process); return null; } @@ -309,8 +328,12 @@ class PreviewGenerator { await for (final chunk in process.stdout) { if (cancelToken?.isCancelled ?? false) { - process.kill(); + // Whole group: killing the worker alone strands vspipe/ffmpeg. + ProcessTree.killTree(process); _previewProcess = null; + _livePreviews.remove(process); + _livePreviews.remove(process); + await ProcessTree.waitForExit(process); return null; } pngBytes.addAll(chunk); @@ -323,6 +346,7 @@ class PreviewGenerator { final exitCode = await process.exitCode; if (_previewProcess == process) { _previewProcess = null; + _livePreviews.remove(process); } // Log the result @@ -347,6 +371,7 @@ class PreviewGenerator { } finally { if (_previewProcess == process) { _previewProcess = null; + _livePreviews.remove(process); } // Clean up config file on error try { @@ -358,20 +383,62 @@ class PreviewGenerator { } /// Cancel any ongoing preview generation. + /// + /// Seeking the scrubber cancels the in-flight preview on every move, so this + /// runs far more often than a job cancel does — and anything it fails to clean + /// up accumulates. Killing the worker's pid alone left `vspipe` and `ffmpeg` + /// running: preview mode installs no signal handler (it returns before ctrlc + /// is set up in `worker/src/main.rs`), so SIGTERM kills the worker outright + /// without unwinding, and `generate_preview` holds its children in locals that + /// `PipelineExecutor::terminate()` never sees. Nothing killed them + /// deliberately; they just tended to die on EPIPE once their pipes closed, + /// which does not happen while they are blocked reading a slow source. + /// + /// Signal the whole process group instead, and wait for the worker to actually + /// go — a fire-and-forget kill cannot tell a clean shutdown from a stray. + /// Signals every live preview and returns as soon as they have been told to + /// stop — it does NOT wait for them to exit. + /// + /// A seek calls this before starting the next preview, so blocking here would + /// add the full shutdown grace to every scrubber movement and make seeking + /// feel broken. Signalling is immediate and ordered; reaping is not, so it + /// runs unawaited. Use [awaitPreviewShutdown] when the wait actually matters. Future cancelPreviewGeneration() async { - if (_previewProcess != null) { - _previewProcess!.kill(); - _previewProcess = null; - } - if (_thumbnailProcess != null) { - _thumbnailProcess!.kill(); - _thumbnailProcess = null; + final doomed = { + ..._livePreviews, + if (_previewProcess != null) _previewProcess!, + if (_thumbnailProcess != null) _thumbnailProcess!, + }; + _livePreviews.clear(); + _previewProcess = null; + _thumbnailProcess = null; + + for (final p in doomed) { + // The whole group: signalling the worker alone strands vspipe/ffmpeg, + // and preview mode has no handler that would clean up after itself. + ProcessTree.killTree(p); + // Reap in the background so a slow shutdown cannot stall the next seek. + unawaited(ProcessTree.waitForExit(p)); } } + /// Cancel and wait for everything to actually be gone. + /// + /// For shutdown, where leaving strays behind is worse than a brief pause. + Future awaitPreviewShutdown() async { + final doomed = { + ..._livePreviews, + if (_previewProcess != null) _previewProcess!, + if (_thumbnailProcess != null) _thumbnailProcess!, + }; + await cancelPreviewGeneration(); + await Future.wait(doomed.map(ProcessTree.waitForExit)); + } + /// Clean up resources. Future dispose() async { - await cancelPreviewGeneration(); + // Shutdown is the one place the wait is worth it: strays outlive the app. + await awaitPreviewShutdown(); _thumbnailCache.clear(); // Clean up temp directory diff --git a/app/lib/services/process_tree.dart b/app/lib/services/process_tree.dart new file mode 100644 index 0000000..3f22a12 --- /dev/null +++ b/app/lib/services/process_tree.dart @@ -0,0 +1,66 @@ +import 'dart:io'; + +/// Terminating a worker without leaving its pipeline behind. +/// +/// Signalling the worker's pid alone does not stop `vspipe` and `ffmpeg`. They +/// usually die shortly afterwards, but only incidentally — their pipes close +/// with the worker and they take EPIPE the next time they write. A child blocked +/// on slow input (a source on a network share is the reported case) writes +/// nothing for minutes, never notices, and is left running at full CPU on work +/// nobody is waiting for. Preview seeking makes this worse than it sounds: every +/// scrub cancels an in-flight preview, so the strays accumulate. +/// +/// The worker puts itself in its own process group at startup (`setpgid` in +/// `worker/src/main.rs`), so its whole tree can be signalled at once by sending +/// to the negated pid. That is a deliberate teardown rather than a hopeful one. +class ProcessTree { + /// Signal [process] and everything it spawned. + /// + /// Falls back to signalling the process alone when the group cannot be + /// reached — an older worker that predates the `setpgid` call, or a platform + /// without process groups. That fallback is exactly the previous behaviour, so + /// this is never worse than what it replaced. + /// + /// Returns true if the group signal landed. + static bool killTree(Process process, [ProcessSignal signal = ProcessSignal.sigterm]) { + if (Platform.isWindows) { + // No process groups; taskkill /T walks the tree instead. Callers on + // Windows use that directly. + return process.kill(signal); + } + // A negative pid addresses the process group. Dart forwards this to kill(2) + // unchanged, and kill(2) defines negative pids as group targets. + try { + if (Process.killPid(-process.pid, signal)) return true; + } on Object { + // Fall through — some platforms reject a negative pid outright. + } + process.kill(signal); + return false; + } + + /// Wait for [process] to exit, escalating to SIGKILL if it outstays [grace]. + /// + /// Returns true if it exited without needing to be forced. + static Future waitForExit( + Process process, { + Duration grace = const Duration(seconds: 5), + Duration forceGrace = const Duration(seconds: 3), + }) async { + try { + await process.exitCode.timeout(grace); + return true; + } on Object { + // Still alive. Forcing it here can orphan the children, which is the very + // thing this class exists to avoid — so force the whole group, not just + // the leader. + killTree(process, ProcessSignal.sigkill); + try { + await process.exitCode.timeout(forceGrace); + } on Object { + // Nothing further we can do from here. + } + return false; + } + } +} diff --git a/app/lib/services/worker_manager.dart b/app/lib/services/worker_manager.dart index 37184ee..af8e1ce 100644 --- a/app/lib/services/worker_manager.dart +++ b/app/lib/services/worker_manager.dart @@ -4,6 +4,7 @@ import 'dart:io'; import '../models/progress_info.dart'; import '../models/video_job.dart'; +import 'process_tree.dart'; import 'temp_directory_service.dart'; import 'tool_locator.dart'; @@ -194,7 +195,11 @@ class WorkerManager { // orphaned; /F is unavoidable there. await Process.run('taskkill', ['/PID', '${process.pid}', '/T', '/F']); } else { - process.kill(ProcessSignal.sigterm); + // Signal the whole process group. The worker still tears its own pipeline + // down when it gets the chance, but that only covers the children it + // tracks, and it cannot run at all if it is forced below — so do not rely + // on it alone. See ProcessTree. + ProcessTree.killTree(process); } // Wait for the process to actually exit. `exitCode` completes once it has @@ -210,7 +215,7 @@ class WorkerManager { // Genuinely wedged. Forcing it here orphans the children — the same // failure described above — but by now the alternative is a job that // never stops at all, so take the lesser problem and say so. - process.kill(ProcessSignal.sigkill); + ProcessTree.killTree(process, ProcessSignal.sigkill); try { await process.exitCode.timeout(_forceKillGrace); } on TimeoutException { diff --git a/app/test/integration_cancel_test.dart b/app/test/integration_cancel_test.dart index 47ef8e7..2e81cfa 100644 --- a/app/test/integration_cancel_test.dart +++ b/app/test/integration_cancel_test.dart @@ -23,6 +23,7 @@ @Tags(['heavy']) library; +import 'dart:async'; import 'dart:convert'; import 'dart:io'; @@ -30,6 +31,7 @@ import 'package:path/path.dart' as p; import 'package:test/test.dart'; import 'package:uuid/uuid.dart'; import 'package:vapourbox/models/encoding_settings.dart'; +import 'package:vapourbox/services/process_tree.dart'; import 'package:vapourbox/models/processing_pipeline.dart'; import 'package:vapourbox/models/qtgmc_parameters.dart'; import 'package:vapourbox/models/video_job.dart'; @@ -187,5 +189,236 @@ void main() { reason: 'worker took ${sw.elapsed.inMilliseconds}ms to shut down, ' 'which does not fit inside WorkerManager._shutdownGrace'); }, timeout: const Timeout(Duration(minutes: 3))); + + test('the worker leads its own process group, so the tree can be signalled', + () async { + // ProcessTree.killTree() signals -pid, which only reaches the pipeline if + // the worker made itself a group leader (setpgid in worker/src/main.rs). + // If that call is ever removed the kill silently degrades to pid-only and + // strays come back — with no visible symptom until a source is slow + // enough that the children do not happen to die on EPIPE. + final job = VideoJob( + id: const Uuid().v4(), + inputPath: longInput, + outputPath: p.join(WorkerHarness.outputDir, 'unused_group.mkv'), + processingPipeline: const ProcessingPipeline( + deinterlace: QTGMCParameters( + enabled: true, preset: QTGMCPreset.slower, tff: true), + ), + encodingSettings: const EncodingSettings( + codec: VideoCodec.h264, + container: ContainerFormat.mkv, + audioMode: AudioMode.passthrough), + ); + final cfg = File(p.join(Directory.systemTemp.path, 'vb_group.json')); + await cfg.writeAsString(jsonEncode(job.toJson())); + addTearDown(() async { + if (await cfg.exists()) await cfg.delete(); + }); + + final proc = await Process.start( + WorkerHarness.workerPath, + ['--config', cfg.path], + environment: WorkerHarness.workerEnv, + workingDirectory: File(WorkerHarness.workerPath).parent.path, + ); + proc.stdout.drain(); + proc.stderr.drain(); + + var pgid = ''; + var kids = []; + for (var i = 0; i < 80; i++) { + await Future.delayed(const Duration(milliseconds: 300)); + final pg = + await Process.run('ps', ['-o', 'pgid=', '-p', '${proc.pid}']); + pgid = pg.stdout.toString().trim(); + if (pgid != '${proc.pid}') continue; + final r = await Process.run('pgrep', ['-g', '${proc.pid}']); + kids = r.stdout + .toString() + .trim() + .split('\n') + .where((s) => s.isNotEmpty && s != '${proc.pid}') + .toList(); + if (kids.isNotEmpty) break; + } + + expect(pgid, '${proc.pid}', + reason: 'the worker is not its own process-group leader, so ' + 'ProcessTree.killTree() cannot reach vspipe/ffmpeg'); + expect(kids, isNotEmpty, + reason: 'no children joined the group — nothing to prove'); + + // Stop them first: a child that dies on EPIPE would pass regardless, and + // that incidental cascade is precisely what must not be relied on. + for (final k in kids) { + Process.killPid(int.parse(k), ProcessSignal.sigstop); + } + expect(Process.killPid(-proc.pid, ProcessSignal.sigterm), isTrue, + reason: 'a negative pid must reach the group'); + await Future.delayed(const Duration(seconds: 1)); + for (final k in kids) { + Process.killPid(int.parse(k), ProcessSignal.sigcont); + } + + var alive = []; + for (var i = 0; i < 10; i++) { + await Future.delayed(const Duration(milliseconds: 500)); + alive = []; + for (final k in kids) { + final r = await Process.run('ps', ['-o', 'pid=', '-p', k]); + if (r.stdout.toString().trim().isNotEmpty) alive.add(k); + } + if (alive.isEmpty) break; + } + for (final k in alive) { + Process.killPid(int.parse(k), ProcessSignal.sigkill); + } + expect(alive, isEmpty, + reason: 'the group signal did not reach the children even though they ' + 'were stopped and so could not have died on EPIPE'); + }, timeout: const Timeout(Duration(minutes: 3))); + + test('rapid seeks leave nothing behind', () async { + // Scrubbing spawns a preview per movement and cancels the previous one. + // The old code tracked a single _previewProcess assigned *after* await + // Process.start returned, so a seek arriving in that window cancelled the + // wrong reference and the in-flight worker was lost — untracked and never + // killed. Ten seeks in quick succession is the shape that produced "a + // bunch of vspipe and ffmpeg processes". + final depsDir = WorkerHarness.depsDir; + final before = await _processesUnder(depsDir); + + final job = VideoJob( + id: const Uuid().v4(), + inputPath: longInput, + outputPath: p.join(WorkerHarness.outputDir, 'unused_seek.mkv'), + processingPipeline: const ProcessingPipeline( + deinterlace: QTGMCParameters( + enabled: true, preset: QTGMCPreset.slower, tff: true), + ), + encodingSettings: const EncodingSettings( + codec: VideoCodec.h264, + container: ContainerFormat.mkv, + audioMode: AudioMode.passthrough), + ); + final cfg = File(p.join(Directory.systemTemp.path, 'vb_seek.json')); + await cfg.writeAsString(jsonEncode(job.toJson())); + addTearDown(() async { + if (await cfg.exists()) await cfg.delete(); + }); + + // Mimic the scrubber: start a preview, and before it can finish, start + // the next one and tear the previous down the way the app now does. + final started = []; + Process? current; + for (var i = 0; i < 10; i++) { + if (current != null) { + ProcessTree.killTree(current); + unawaited(ProcessTree.waitForExit(current)); + } + current = await Process.start( + WorkerHarness.workerPath, + ['--config', cfg.path, '--preview', '--frame', '${300 + i * 40}'], + environment: WorkerHarness.workerEnv, + workingDirectory: File(WorkerHarness.workerPath).parent.path, + ); + current.stdout.drain(); + current.stderr.drain(); + started.add(current); + await Future.delayed(const Duration(milliseconds: 250)); + } + ProcessTree.killTree(current!); + await ProcessTree.waitForExit(current); + + // Everything the burst spawned must be gone. + var strays = []; + for (var i = 0; i < 20; i++) { + await Future.delayed(const Duration(milliseconds: 500)); + strays = (await _processesUnder(depsDir)) + .where((pid) => !before.contains(pid)) + .toList(); + if (strays.isEmpty) break; + } + for (final pid in strays) { + Process.killPid(pid, ProcessSignal.sigkill); + } + expect(strays, isEmpty, + reason: '${strays.length} worker/vspipe/ffmpeg processes survived a ' + 'burst of ${started.length} seeks'); + }, timeout: const Timeout(Duration(minutes: 4))); + + test('killing a preview leaves no orphaned children', () async { + // Seeking the preview scrubber cancels the in-flight preview and starts + // another, so this path runs far more often than a job cancel does. + final depsDir = WorkerHarness.depsDir; + final job = VideoJob( + id: const Uuid().v4(), + inputPath: longInput, + outputPath: p.join(WorkerHarness.outputDir, 'unused_preview.mkv'), + processingPipeline: const ProcessingPipeline( + deinterlace: QTGMCParameters( + enabled: true, + preset: QTGMCPreset.slower, + tff: true, + ), + ), + encodingSettings: const EncodingSettings( + codec: VideoCodec.h264, + container: ContainerFormat.mkv, + audioMode: AudioMode.passthrough, + ), + ); + final configFile = + File(p.join(Directory.systemTemp.path, 'vb_preview_orphans.json')); + await configFile.writeAsString(jsonEncode(job.toJson())); + addTearDown(() async { + if (await configFile.exists()) await configFile.delete(); + }); + + final before = await _processesUnder(depsDir); + + final proc = await Process.start( + WorkerHarness.workerPath, + ['--config', configFile.path, '--preview', '--frame', '900'], + environment: WorkerHarness.workerEnv, + workingDirectory: File(WorkerHarness.workerPath).parent.path, + ); + proc.stdout.drain(); + proc.stderr.drain(); + + var spawned = []; + final deadline = DateTime.now().add(const Duration(seconds: 40)); + while (DateTime.now().isBefore(deadline)) { + await Future.delayed(const Duration(milliseconds: 500)); + spawned = (await _processesUnder(depsDir)) + .where((pid) => !before.contains(pid)) + .toList(); + if (spawned.length >= 2) break; + } + expect(spawned.length, greaterThanOrEqualTo(2), + reason: 'the preview pipeline never started, so this proves nothing'); + + // Exactly what PreviewGenerator.cancelPreviewGeneration() does. + proc.kill(); + await proc.exitCode.timeout(const Duration(seconds: 15), + onTimeout: () { proc.kill(ProcessSignal.sigkill); return -1; }); + + List leftovers = []; + for (var i = 0; i < 20; i++) { + await Future.delayed(const Duration(milliseconds: 500)); + final now = await _processesUnder(depsDir); + leftovers = spawned.where(now.contains).toList(); + if (leftovers.isEmpty) break; + } + if (leftovers.isNotEmpty) { + for (final pid in leftovers) { + Process.killPid(pid, ProcessSignal.sigkill); + } + } + expect(leftovers, isEmpty, + reason: 'vspipe/ffmpeg outlived the preview worker. Seeking the ' + 'scrubber cancels a preview on every move, so these accumulate'); + }, timeout: const Timeout(Duration(minutes: 3))); }); } diff --git a/worker/src/main.rs b/worker/src/main.rs index 3ccca0c..4f19bbf 100644 --- a/worker/src/main.rs +++ b/worker/src/main.rs @@ -82,6 +82,31 @@ struct Args { fn main() -> ExitCode { let args = Args::parse(); + // Become our own process-group leader, so the app can tear down this worker + // *and everything it spawns* with a single signal to -pid. + // + // Without this, killing the worker leaves vspipe and ffmpeg running. They + // usually die shortly afterwards, but only incidentally: their pipes close + // with the worker and they take EPIPE the next time they write. A child + // blocked on slow input — reading a source over a network share is the + // reported case — writes nothing for minutes and so never notices, and is + // left encoding a job nobody is waiting for. + // + // Nothing kills them deliberately today. The encode path's + // PipelineExecutor::terminate() only covers the children it tracks on + // `self`, and preview mode never even installs a signal handler (it returns + // below before ctrlc is set up), so a SIGTERM there kills the worker outright + // without unwinding — Drop never runs and its children are simply abandoned. + // Rather than add bookkeeping to each path, put everything in one group. + // + // Best-effort: if this fails the app falls back to signalling the pid alone, + // which is exactly today's behaviour. + #[cfg(unix)] + { + use nix::unistd::{setpgid, Pid}; + let _ = setpgid(Pid::from_raw(0), Pid::from_raw(0)); + } + // DVD info mode: enumerate titles, output JSON to stdout if let Some(ref dvd_path) = args.dvd_info { return run_dvd_info(dvd_path);