Merge pull request #17536 from gnattu/fix-concurrent-racing

Fix concurrent ffmpeg segment racing

Jellyfin.Api/Controllers/DynamicHlsController.cs

@@ -1456,22 +1456,16 @@ public class DynamicHlsController : BaseJellyfinApiController
 
         var segmentExtension = EncodingHelper.GetSegmentFileExtension(state.Request.SegmentContainer);
 
-        TranscodingJob? job;
-
-        if (System.IO.File.Exists(segmentPath))
-        {
-            job = _transcodeManager.OnTranscodeBeginRequest(playlistPath, TranscodingJobType);
-            _logger.LogDebug("returning {0} [it exists, try 1]", segmentPath);
-            return await GetSegmentResult(state, playlistPath, segmentPath, segmentExtension, segmentId, job, cancellationToken).ConfigureAwait(false);
-        }
-
+        // Keep segment selection and transcoding replacement under the same playlist lock.
+        // An out-of-order request must not replace a job while another request is using its output.
         using (await _transcodeManager.LockAsync(playlistPath, cancellationToken).ConfigureAwait(false))
         {
+            TranscodingJob? job;
             var startTranscoding = false;
             if (System.IO.File.Exists(segmentPath))
             {
                 job = _transcodeManager.OnTranscodeBeginRequest(playlistPath, TranscodingJobType);
-                _logger.LogDebug("returning {0} [it exists, try 2]", segmentPath);
+                _logger.LogDebug("returning {0} [it exists]", segmentPath);
                 return await GetSegmentResult(state, playlistPath, segmentPath, segmentExtension, segmentId, job, cancellationToken).ConfigureAwait(false);
             }
 
@@ -1505,6 +1499,9 @@ public class DynamicHlsController : BaseJellyfinApiController
                 // If the playlist doesn't already exist, startup ffmpeg
                 try
                 {
+                    var currentJob = _transcodeManager.GetTranscodingJob(playlistPath, TranscodingJobType);
+                    await WaitForActiveTranscodingRequests(currentJob, cancellationToken).ConfigureAwait(false);
+
                     await _transcodeManager.KillTranscodingJobs(streamingRequest.DeviceId, streamingRequest.PlaySessionId, p => false)
                         .ConfigureAwait(false);
 
@@ -1540,11 +1537,19 @@ public class DynamicHlsController : BaseJellyfinApiController
                     await job.TranscodingThrottler.UnpauseTranscoding().ConfigureAwait(false);
                 }
             }
-        }
 
-        _logger.LogDebug("returning {0} [general case]", segmentPath);
-        job ??= _transcodeManager.OnTranscodeBeginRequest(playlistPath, TranscodingJobType);
-        return await GetSegmentResult(state, playlistPath, segmentPath, segmentExtension, segmentId, job, cancellationToken).ConfigureAwait(false);
+            _logger.LogDebug("returning {0} [general case]", segmentPath);
+            job ??= _transcodeManager.OnTranscodeBeginRequest(playlistPath, TranscodingJobType);
+            return await GetSegmentResult(state, playlistPath, segmentPath, segmentExtension, segmentId, job, cancellationToken).ConfigureAwait(false);
+        }
+    }
+
+    internal static async Task WaitForActiveTranscodingRequests(TranscodingJob? job, CancellationToken cancellationToken)
+    {
+        while (job?.ActiveRequestCount > 0)
+        {
+            await Task.Delay(100, cancellationToken).ConfigureAwait(false);
+        }
     }
 
     private static double[] GetSegmentLengths(StreamState state)

MediaBrowser.Controller/MediaEncoding/TranscodingJob.cs

@@ -15,6 +15,7 @@ public sealed class TranscodingJob : IDisposable
     private readonly Lock _processLock = new();
     private readonly Lock _timerLock = new();
 
+    private int _activeRequestCount;
     private Timer? _killTimer;
 
     /// <summary>
@@ -64,7 +65,11 @@ public sealed class TranscodingJob : IDisposable
     /// <summary>
     /// Gets or sets the active request count.
     /// </summary>
-    public int ActiveRequestCount { get; set; }
+    public int ActiveRequestCount
+    {
+        get => Volatile.Read(ref _activeRequestCount);
+        set => Volatile.Write(ref _activeRequestCount, value);
+    }
 
     /// <summary>
     /// Gets or sets device id.
@@ -151,6 +156,20 @@ public sealed class TranscodingJob : IDisposable
     /// </summary>
     public int PingTimeout { get; set; }
 
+    /// <summary>
+    /// Increments the active request count.
+    /// </summary>
+    /// <returns>The incremented count.</returns>
+    public int IncrementActiveRequestCount()
+        => Interlocked.Increment(ref _activeRequestCount);
+
+    /// <summary>
+    /// Decrements the active request count.
+    /// </summary>
+    /// <returns>The decremented count.</returns>
+    public int DecrementActiveRequestCount()
+        => Interlocked.Decrement(ref _activeRequestCount);
+
     /// <summary>
     /// Stop kill timer.
     /// </summary>

MediaBrowser.MediaEncoding/Transcoding/TranscodeManager.cs

@@ -612,9 +612,9 @@ public sealed class TranscodeManager : ITranscodeManager, IDisposable
     /// <inheritdoc />
     public void OnTranscodeEndRequest(TranscodingJob job)
     {
-        job.ActiveRequestCount--;
-        _logger.LogDebug("OnTranscodeEndRequest job.ActiveRequestCount={ActiveRequestCount}", job.ActiveRequestCount);
-        if (job.ActiveRequestCount <= 0)
+        var activeRequestCount = job.DecrementActiveRequestCount();
+        _logger.LogDebug("OnTranscodeEndRequest job.ActiveRequestCount={ActiveRequestCount}", activeRequestCount);
+        if (activeRequestCount <= 0)
         {
             PingTimer(job, false);
         }
@@ -697,7 +697,7 @@ public sealed class TranscodeManager : ITranscodeManager, IDisposable
                 return null;
             }
 
-            job.ActiveRequestCount++;
+            job.IncrementActiveRequestCount();
             if (string.IsNullOrWhiteSpace(job.PlaySessionId) || job.Type == TranscodingJobType.Progressive)
             {
                 job.StopKillTimer();

tests/Jellyfin.Api.Tests/Controllers/DynamicHlsControllerTests.cs

@@ -1,5 +1,9 @@
 using System;
+using System.Threading;
+using System.Threading.Tasks;
 using Jellyfin.Api.Controllers;
+using MediaBrowser.Controller.MediaEncoding;
+using Microsoft.Extensions.Logging.Abstractions;
 using Xunit;
 
 namespace Jellyfin.Api.Tests.Controllers
@@ -41,5 +45,77 @@ namespace Jellyfin.Api.Tests.Controllers
 
             return data;
         }
+
+        [Fact]
+        public async Task WaitForActiveTranscodingRequests_WaitsUntilRequestCompletes()
+        {
+            var job = new TranscodingJob(NullLogger<TranscodingJob>.Instance)
+            {
+                ActiveRequestCount = 1
+            };
+
+            var waitTask = DynamicHlsController.WaitForActiveTranscodingRequests(job, CancellationToken.None);
+            Assert.False(waitTask.IsCompleted);
+
+            job.DecrementActiveRequestCount();
+
+            await waitTask;
+        }
+
+        [Fact]
+        public async Task WaitForActiveTranscodingRequests_WaitsForEveryRequest()
+        {
+            var job = new TranscodingJob(NullLogger<TranscodingJob>.Instance)
+            {
+                ActiveRequestCount = 2
+            };
+
+            var waitTask = DynamicHlsController.WaitForActiveTranscodingRequests(job, CancellationToken.None);
+            job.DecrementActiveRequestCount();
+
+            await Task.Delay(150, TestContext.Current.CancellationToken);
+            Assert.False(waitTask.IsCompleted);
+
+            job.DecrementActiveRequestCount();
+
+            await waitTask;
+        }
+
+        [Fact]
+        public async Task WaitForActiveTranscodingRequests_ReturnsWithoutAnActiveRequest()
+        {
+            var job = new TranscodingJob(NullLogger<TranscodingJob>.Instance);
+
+            await DynamicHlsController.WaitForActiveTranscodingRequests(job, CancellationToken.None);
+            await DynamicHlsController.WaitForActiveTranscodingRequests(null, CancellationToken.None);
+        }
+
+        [Fact]
+        public async Task WaitForActiveTranscodingRequests_ObservesCancellation()
+        {
+            var job = new TranscodingJob(NullLogger<TranscodingJob>.Instance)
+            {
+                ActiveRequestCount = 1
+            };
+            using var cancellationTokenSource = new CancellationTokenSource();
+
+            var waitTask = DynamicHlsController.WaitForActiveTranscodingRequests(job, cancellationTokenSource.Token);
+            await cancellationTokenSource.CancelAsync();
+
+            await Assert.ThrowsAnyAsync<OperationCanceledException>(() => waitTask);
+        }
+
+        [Fact]
+        public async Task ActiveRequestCount_UpdatesAtomically()
+        {
+            const int RequestCount = 1000;
+            var job = new TranscodingJob(NullLogger<TranscodingJob>.Instance);
+
+            await Task.WhenAll(
+                Task.Run(() => Parallel.For(0, RequestCount, _ => job.IncrementActiveRequestCount())),
+                Task.Run(() => Parallel.For(0, RequestCount, _ => job.DecrementActiveRequestCount())));
+
+            Assert.Equal(0, job.ActiveRequestCount);
+        }
     }
 }