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