From 97246bcbc38420da67e76bbe46a0dbf38f424463 Mon Sep 17 00:00:00 2001 From: twotalesanimation <80506065+twotalesanimation@users.noreply.github.com> Date: Fri, 18 Sep 2026 04:21:54 +0200 Subject: [PATCH] fix(worker): Blender jobs always failed with "Collection was modified" LogTail was read by the progress loop while the output pumps appended to it. Blender rewrites its status line many times a second, so the first progress report of every job threw; aerender hit the same race only rarely. LogTail now locks internally. The exception also escaped BlenderRunner without ending the process, leaving blender.exe rendering unsupervised on the node on every failed attempt. Any unexpected exception in the monitor loop now kills the process tree. Bump worker to 1.6.2. Co-Authored-By: Claude Opus 5 --- .../ProgressParserTests.cs | 17 +++ .../VFXReviewWorker/AerenderRunner.cs | 22 ++- RenderWorker/VFXReviewWorker/BlenderRunner.cs | 130 ++++++++++-------- .../VFXReviewWorker/VFXReviewWorker.csproj | 2 +- 4 files changed, 109 insertions(+), 62 deletions(-) diff --git a/RenderWorker/VFXReviewWorker.Tests/ProgressParserTests.cs b/RenderWorker/VFXReviewWorker.Tests/ProgressParserTests.cs index 2ad8a5f..37387ae 100644 --- a/RenderWorker/VFXReviewWorker.Tests/ProgressParserTests.cs +++ b/RenderWorker/VFXReviewWorker.Tests/ProgressParserTests.cs @@ -117,4 +117,21 @@ public class LogTailTests for (var i = 1; i <= 5; i++) tail.Add($"line {i}"); Assert.Equal("line 3\nline 4\nline 5", tail.ToString()); } + + [Fact] + public async Task ReadingWhileAnotherThreadWritesDoesNotThrow() + { + // The Blender runner's output pumps add lines while its progress loop + // reads the tail; unsynchronised this threw "Collection was modified". + var tail = new LogTail(200); + using var stop = new CancellationTokenSource(TimeSpan.FromSeconds(1)); + var writer = Task.Run(() => + { + var i = 0; + while (!stop.IsCancellationRequested) tail.Add($"Fra:1001 Sample {i++}/128"); + }); + + while (!stop.IsCancellationRequested) _ = tail.ToString(); + await writer; + } } diff --git a/RenderWorker/VFXReviewWorker/AerenderRunner.cs b/RenderWorker/VFXReviewWorker/AerenderRunner.cs index 92df34d..5da43c2 100644 --- a/RenderWorker/VFXReviewWorker/AerenderRunner.cs +++ b/RenderWorker/VFXReviewWorker/AerenderRunner.cs @@ -280,11 +280,19 @@ public sealed class AerenderRunner } } -/// Ring buffer of the last N output lines (§5.3 logTail). +/// +/// Ring buffer of the last N output lines (§5.3 logTail). +/// +/// Thread-safe: output readers add lines while the progress loop reads the +/// tail for its report. Unsynchronised, ToString enumerates the queue while a +/// reader enqueues and throws "Collection was modified" — Blender, which +/// writes a status line many times a second, hit this on every job. +/// public sealed class LogTail { private readonly Queue _lines; private readonly int _capacity; + private readonly object _gate = new(); public LogTail(int capacity) { @@ -294,9 +302,15 @@ public sealed class LogTail public void Add(string line) { - _lines.Enqueue(line); - while (_lines.Count > _capacity) _lines.Dequeue(); + lock (_gate) + { + _lines.Enqueue(line); + while (_lines.Count > _capacity) _lines.Dequeue(); + } } - public override string ToString() => string.Join('\n', _lines); + public override string ToString() + { + lock (_gate) return string.Join('\n', _lines); + } } diff --git a/RenderWorker/VFXReviewWorker/BlenderRunner.cs b/RenderWorker/VFXReviewWorker/BlenderRunner.cs index 7fa310f..f7b1733 100644 --- a/RenderWorker/VFXReviewWorker/BlenderRunner.cs +++ b/RenderWorker/VFXReviewWorker/BlenderRunner.cs @@ -190,75 +190,91 @@ public sealed class BlenderRunner PumpAsync(process.StandardError, line => HandleLine(line, isStderr: true)), }; - while (!process.HasExited) + try { - await Task.Delay(TimeSpan.FromSeconds(2), CancellationToken.None); + return await MonitorAsync(); + } + catch + { + // Disposing a Process does not end it: without this, any exception + // in the monitor loop leaves Blender rendering on the node, + // unsupervised, while the job is retried alongside it. + KillTree(process); + throw; + } - if (cancelJob.IsCancellationRequested || shutdown.IsCancellationRequested) + async Task MonitorAsync() + { + while (!process.HasExited) { - _log.LogWarning("Cancellation requested — killing Blender tree for job {JobId}", jobId); - KillTree(process); - cancelledByServer = cancelJob.IsCancellationRequested; - break; + await Task.Delay(TimeSpan.FromSeconds(2), CancellationToken.None); + + if (cancelJob.IsCancellationRequested || shutdown.IsCancellationRequested) + { + _log.LogWarning("Cancellation requested — killing Blender tree for job {JobId}", jobId); + KillTree(process); + cancelledByServer = cancelJob.IsCancellationRequested; + break; + } + + if ((DateTimeOffset.UtcNow - lastProgressAt).TotalSeconds > config.StallTimeoutSeconds) + { + _log.LogError("Blender stalled ({Sec}s without output) — killing job {JobId}", config.StallTimeoutSeconds, jobId); + KillTree(process); + await WaitForExitSafe(process); + return new RenderResult(false, -2, Retryable: true, + $"Blender produced no output for {config.StallTimeoutSeconds}s (stall timeout)", + tail.ToString(), fullLogPath, sw.Elapsed.TotalSeconds); + } + + if ((DateTimeOffset.UtcNow - lastReportAt).TotalSeconds >= 10) + { + lastReportAt = DateTimeOffset.UtcNow; + var progress = totalFrames > 0 ? Math.Min(1.0, (double)framesSaved / totalFrames) : 0; + var resp = await _api.ProgressAsync(jobId, machineId, progress, currentFrame, totalFrames, + eta.EstimateSeconds(totalFrames - framesSaved), tail.ToString(), shutdown); + if (resp?.CancelRequested == true) + { + _log.LogWarning("Server requested cancel for job {JobId}", jobId); + KillTree(process); + cancelledByServer = true; + break; + } + } } - if ((DateTimeOffset.UtcNow - lastProgressAt).TotalSeconds > config.StallTimeoutSeconds) + await WaitForExitSafe(process); + // Let the readers drain what is still buffered before judging the run — + // the last frame's "Saved:" often arrives after the process exits. + try { await Task.WhenAll(pumps).WaitAsync(TimeSpan.FromSeconds(10)); } catch { } + sw.Stop(); + lock (logLock) { fullLog.Flush(); } + + if (cancelledByServer) { - _log.LogError("Blender stalled ({Sec}s without output) — killing job {JobId}", config.StallTimeoutSeconds, jobId); - KillTree(process); - await WaitForExitSafe(process); - return new RenderResult(false, -2, Retryable: true, - $"Blender produced no output for {config.StallTimeoutSeconds}s (stall timeout)", + return new RenderResult(false, -3, Retryable: false, "Cancelled", tail.ToString(), fullLogPath, sw.Elapsed.TotalSeconds); + } + + var exitCode = process.ExitCode; + if (exitCode != 0) + { + return new RenderResult(false, exitCode, Retryable: !nonRetryable, + firstError ?? $"Blender exited with code {exitCode}", tail.ToString(), fullLogPath, sw.Elapsed.TotalSeconds); } - if ((DateTimeOffset.UtcNow - lastReportAt).TotalSeconds >= 10) + // Blender exits 0 having rendered nothing if the range is empty or the + // output path is unwritable, so a silent success is treated as failure. + if (framesSaved == 0) { - lastReportAt = DateTimeOffset.UtcNow; - var progress = totalFrames > 0 ? Math.Min(1.0, (double)framesSaved / totalFrames) : 0; - var resp = await _api.ProgressAsync(jobId, machineId, progress, currentFrame, totalFrames, - eta.EstimateSeconds(totalFrames - framesSaved), tail.ToString(), shutdown); - if (resp?.CancelRequested == true) - { - _log.LogWarning("Server requested cancel for job {JobId}", jobId); - KillTree(process); - cancelledByServer = true; - break; - } + return new RenderResult(false, 0, Retryable: false, + "Blender exited cleanly but saved no frames — check the output path and frame range in the .blend", + tail.ToString(), fullLogPath, sw.Elapsed.TotalSeconds); } + + _log.LogInformation("Job {JobId}: {Saved}/{Total} frames in {Sec:F0}s", jobId, framesSaved, totalFrames, sw.Elapsed.TotalSeconds); + return new RenderResult(true, 0, true, null, tail.ToString(), fullLogPath, sw.Elapsed.TotalSeconds); } - - await WaitForExitSafe(process); - // Let the readers drain what is still buffered before judging the run — - // the last frame's "Saved:" often arrives after the process exits. - try { await Task.WhenAll(pumps).WaitAsync(TimeSpan.FromSeconds(10)); } catch { } - sw.Stop(); - lock (logLock) { fullLog.Flush(); } - - if (cancelledByServer) - { - return new RenderResult(false, -3, Retryable: false, "Cancelled", tail.ToString(), fullLogPath, sw.Elapsed.TotalSeconds); - } - - var exitCode = process.ExitCode; - if (exitCode != 0) - { - return new RenderResult(false, exitCode, Retryable: !nonRetryable, - firstError ?? $"Blender exited with code {exitCode}", - tail.ToString(), fullLogPath, sw.Elapsed.TotalSeconds); - } - - // Blender exits 0 having rendered nothing if the range is empty or the - // output path is unwritable, so a silent success is treated as failure. - if (framesSaved == 0) - { - return new RenderResult(false, 0, Retryable: false, - "Blender exited cleanly but saved no frames — check the output path and frame range in the .blend", - tail.ToString(), fullLogPath, sw.Elapsed.TotalSeconds); - } - - _log.LogInformation("Job {JobId}: {Saved}/{Total} frames in {Sec:F0}s", jobId, framesSaved, totalFrames, sw.Elapsed.TotalSeconds); - return new RenderResult(true, 0, true, null, tail.ToString(), fullLogPath, sw.Elapsed.TotalSeconds); } /// diff --git a/RenderWorker/VFXReviewWorker/VFXReviewWorker.csproj b/RenderWorker/VFXReviewWorker/VFXReviewWorker.csproj index 7ea8ca5..862fbf0 100644 --- a/RenderWorker/VFXReviewWorker/VFXReviewWorker.csproj +++ b/RenderWorker/VFXReviewWorker/VFXReviewWorker.csproj @@ -15,7 +15,7 @@ every change that gets published. publish.ps1 appends the git SHA via SourceRevisionId, giving "1.1.0+feb1c7b". --> - 1.6.1 + 1.6.2