diff options
4 files changed, 118 insertions, 18 deletions
diff --git a/Jellyfin.Api/Controllers/DynamicHlsController.cs b/Jellyfin.Api/Controllers/DynamicHlsController.cs index 4aa728b5bf..a6555a2beb 100644 --- a/Jellyfin.Api/Controllers/DynamicHlsController.cs +++ b/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) diff --git a/MediaBrowser.Controller/MediaEncoding/TranscodingJob.cs b/MediaBrowser.Controller/MediaEncoding/TranscodingJob.cs index 56990d0b82..5045030b9b 100644 --- a/MediaBrowser.Controller/MediaEncoding/TranscodingJob.cs +++ b/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. @@ -152,6 +157,20 @@ public sealed class TranscodingJob : IDisposable 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> public void StopKillTimer() diff --git a/MediaBrowser.MediaEncoding/Transcoding/TranscodeManager.cs b/MediaBrowser.MediaEncoding/Transcoding/TranscodeManager.cs index 78bb881ec2..dfc057f611 100644 --- a/MediaBrowser.MediaEncoding/Transcoding/TranscodeManager.cs +++ b/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(); diff --git a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsControllerTests.cs b/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsControllerTests.cs index 1f06e8fde6..5f5f273f12 100644 --- a/tests/Jellyfin.Api.Tests/Controllers/DynamicHlsControllerTests.cs +++ b/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); + } } } |