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);