diff --git a/ErsatzTV.Application/Maintenance/Commands/ReleaseMemoryHandler.cs b/ErsatzTV.Application/Maintenance/Commands/ReleaseMemoryHandler.cs index 279b056e6..bbdf6cb2d 100644 --- a/ErsatzTV.Application/Maintenance/Commands/ReleaseMemoryHandler.cs +++ b/ErsatzTV.Application/Maintenance/Commands/ReleaseMemoryHandler.cs @@ -27,7 +27,7 @@ public class ReleaseMemoryHandler : IRequestHandler return Task.CompletedTask; } - bool hasActiveWorkers = _ffmpegSegmenterService.Workers.Count >= 0 || FFmpegProcess.ProcessCount > 0; + bool hasActiveWorkers = _ffmpegSegmenterService.Workers.Count > 0 || FFmpegProcess.ProcessCount > 0; if (request.ForceAggressive || !hasActiveWorkers) { _logger.LogDebug("Starting aggressive garbage collection"); diff --git a/ErsatzTV.Application/Streaming/Commands/StartFFmpegNextSessionHandler.cs b/ErsatzTV.Application/Streaming/Commands/StartFFmpegNextSessionHandler.cs index 98b3dd23b..5020b113e 100644 --- a/ErsatzTV.Application/Streaming/Commands/StartFFmpegNextSessionHandler.cs +++ b/ErsatzTV.Application/Streaming/Commands/StartFFmpegNextSessionHandler.cs @@ -7,7 +7,6 @@ using ErsatzTV.Application.Graphics; using ErsatzTV.Application.Maintenance; using ErsatzTV.Core; using ErsatzTV.Core.Domain; -using ErsatzTV.Core.Errors; using ErsatzTV.Core.FFmpeg; using ErsatzTV.Core.Interfaces.FFmpeg; using ErsatzTV.Core.Interfaces.Metadata; @@ -41,15 +40,24 @@ public class StartFFmpegNextSessionHandler( StartFFmpegNextSession request, CancellationToken cancellationToken) { - using IDisposable releaser = - await ffmpegSegmenterService.LockForStart(request.ChannelNumber, cancellationToken); + int initialSegmentCount = await configElementRepository + .GetValue(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken) + .Map(maybeCount => maybeCount.Match(identity, () => 1)); - if (ffmpegSegmenterService.TryGetWorker(request.ChannelNumber, out IHlsSessionWorker existing)) - { - existing.Touch(Option.None); - return new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request)); - } + Either ready = await SessionStartCoordinator.Start( + ffmpegSegmenterService, + request.ChannelNumber, + () => CreateWorker(request, cancellationToken), + initialSegmentCount, + StartDeadline, + cancellationToken); + return await ready.MapAsync(async _ => await GetMultiVariantPlaylist(request)); + } + private async Task> CreateWorker( + StartFFmpegNextSession request, + CancellationToken cancellationToken) + { Validation maybeChannelBinary = await ChannelBinaryMustExist(); if (maybeChannelBinary.IsFail) { @@ -64,10 +72,6 @@ public class StartFFmpegNextSessionHandler( // new GetChannelFramerate(request.ChannelNumber), // cancellationToken); - int initialSegmentCount = await configElementRepository - .GetValue(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken) - .Map(maybeCount => maybeCount.Match(identity, () => 1)); - Option maybeChannel = await mediator.Send(new GetChannelByNumber(request.ChannelNumber), cancellationToken); @@ -113,7 +117,13 @@ public class StartFFmpegNextSessionHandler( if (!ffmpegSegmenterService.TryAddWorker(request.ChannelNumber, worker)) { - return new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request)); + ((IDisposable)worker).Dispose(); + if (ffmpegSegmenterService.TryGetWorker(request.ChannelNumber, out IHlsSessionWorker existing)) + { + return Right(existing); + } + + return new SessionEndedBeforeReady(request.ChannelNumber); } // fire and forget worker @@ -129,14 +139,7 @@ public class StartFFmpegNextSessionHandler( }, TaskScheduler.Default); - Either ready = await SessionStartWait.ForReady( - request.ChannelNumber, - worker, - runTask, - initialSegmentCount, - StartDeadline, - cancellationToken); - return await ready.MapAsync(async _ => await GetMultiVariantPlaylist(request)); + return Right(worker); } private void PrepareTranscodeFolder(string channelNumber) diff --git a/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs b/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs index d646897cb..34fb3f9d1 100644 --- a/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs +++ b/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs @@ -6,7 +6,6 @@ using ErsatzTV.Application.Graphics; using ErsatzTV.Application.Maintenance; using ErsatzTV.Core; using ErsatzTV.Core.Domain; -using ErsatzTV.Core.Errors; using ErsatzTV.Core.FFmpeg; using ErsatzTV.Core.Interfaces.FFmpeg; using ErsatzTV.Core.Interfaces.Metadata; @@ -70,15 +69,24 @@ public class StartFFmpegSessionHandler : IRequestHandler> Handle(StartFFmpegSession request, CancellationToken cancellationToken) { - using IDisposable releaser = - await _ffmpegSegmenterService.LockForStart(request.ChannelNumber, cancellationToken); + int initialSegmentCount = await _configElementRepository + .GetValue(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken) + .Map(maybeCount => maybeCount.Match(identity, () => 1)); - if (_ffmpegSegmenterService.TryGetWorker(request.ChannelNumber, out IHlsSessionWorker existing)) - { - existing.Touch(Option.None); - return new ChannelSessionAlreadyActive(await GetMultiVariantPlaylist(request)); - } + Either ready = await SessionStartCoordinator.Start( + _ffmpegSegmenterService, + request.ChannelNumber, + () => CreateWorker(request, cancellationToken), + initialSegmentCount, + StartDeadline, + cancellationToken); + return await ready.MapAsync(async _ => await GetMultiVariantPlaylist(request)); + } + private async Task> CreateWorker( + StartFFmpegSession request, + CancellationToken cancellationToken) + { Option idleTimeout = await _configElementRepository .GetValue(ConfigElementKey.FFmpegSegmenterTimeout, cancellationToken) .Map(maybeTimeout => maybeTimeout.Match(i => TimeSpan.FromSeconds(i), () => TimeSpan.FromMinutes(1))); @@ -87,10 +95,6 @@ public class StartFFmpegSessionHandler : IRequestHandler(ConfigElementKey.FFmpegInitialSegmentCount, cancellationToken) - .Map(maybeCount => maybeCount.Match(identity, () => 1)); - // disable idle timeout when configured to keep running Option channel = await _mediator.Send(new GetChannelByNumber(request.ChannelNumber), cancellationToken); @@ -106,7 +110,13 @@ public class StartFFmpegSessionHandler : IRequestHandler(existing); + } + + return new SessionEndedBeforeReady(request.ChannelNumber); } // fire and forget worker @@ -122,14 +132,7 @@ public class StartFFmpegSessionHandler : IRequestHandler ready = await SessionStartWait.ForReady( - request.ChannelNumber, - worker, - runTask, - initialSegmentCount, - StartDeadline, - cancellationToken); - return await ready.MapAsync(async _ => await GetMultiVariantPlaylist(request)); + return Right(worker); } private HlsSessionWorker GetSessionWorker(StartFFmpegSession request, Option targetFramerate) => diff --git a/ErsatzTV.Application/Streaming/HlsSessionWorker.cs b/ErsatzTV.Application/Streaming/HlsSessionWorker.cs index 01bc4e741..ac49b756d 100644 --- a/ErsatzTV.Application/Streaming/HlsSessionWorker.cs +++ b/ErsatzTV.Application/Streaming/HlsSessionWorker.cs @@ -319,13 +319,12 @@ public class HlsSessionWorker : IHlsSessionWorker _logger.LogDebug("Playlist exists"); - // start the segment-wait deadline only after the playlist file appears, - // so slow pipeline setup (e.g. h264 profile probing) doesn't consume the budget - DateTimeOffset finish = DateTimeOffset.Now.AddSeconds(8); + // The caller owns the deadline. Only report readiness once playable segments exist. + int requiredSegmentCount = Math.Max(1, initialSegmentCount); var segmentCount = 0; int lastSegmentCount = -1; - while (DateTimeOffset.Now < finish && segmentCount < initialSegmentCount) + while (segmentCount < requiredSegmentCount) { if (segmentCount != lastSegmentCount) { diff --git a/ErsatzTV.Core.Tests/FFmpeg/FFmpegSegmenterServiceTests.cs b/ErsatzTV.Core.Tests/FFmpeg/FFmpegSegmenterServiceTests.cs index 047320568..fcd083718 100644 --- a/ErsatzTV.Core.Tests/FFmpeg/FFmpegSegmenterServiceTests.cs +++ b/ErsatzTV.Core.Tests/FFmpeg/FFmpegSegmenterServiceTests.cs @@ -88,6 +88,69 @@ public class FFmpegSegmenterServiceTests count.ShouldBe(2); } + [TestCase(false)] + [TestCase(true)] + public async Task WaitForReady_Should_Wait_Again_After_Interrupted_Startup(bool cancelRequest) + { + var playlist = new TaskCompletionSource(); + IHlsSessionWorker worker = Substitute.For(); + worker.WaitForPlaylistSegments(Arg.Any(), Arg.Any()) + .Returns(call => playlist.Task.WaitAsync(call.Arg())); + _service.TryAddWorker("1", worker); + + using var cts = new CancellationTokenSource(); + Task> first = _service.WaitForReady( + "1", worker, 1, + cancelRequest ? TimeSpan.FromSeconds(5) : TimeSpan.FromMilliseconds(50), cts.Token); + if (cancelRequest) + { + await cts.CancelAsync(); + await Should.ThrowAsync(() => first); + } + else + { + (await first.WaitAsync(TimeSpan.FromSeconds(5))).IsLeft.ShouldBeTrue(); + } + + _service.TryGetWorker("1", out IHlsSessionWorker existing).ShouldBeTrue(); + Task> retry = _service.WaitForReady( + "1", existing, 1, TimeSpan.FromSeconds(5), CancellationToken.None); + retry.IsCompleted.ShouldBeFalse(); + + playlist.SetResult(); + (await retry.WaitAsync(TimeSpan.FromSeconds(5))).IsRight.ShouldBeTrue(); + + // Established sessions should not poll the playlist again on every tune-in. + worker.ClearReceivedCalls(); + (await _service.WaitForReady("1", worker, 1, TimeSpan.FromSeconds(5), CancellationToken.None)) + .IsRight.ShouldBeTrue(); + await worker.DidNotReceive().WaitForPlaylistSegments(Arg.Any(), Arg.Any()); + } + + [Test] + public async Task WaitForReady_Should_Fail_When_Worker_Is_Removed_And_Not_Reuse_Readiness() + { + var playlist = new TaskCompletionSource(); + IHlsSessionWorker worker = Substitute.For(); + worker.WaitForPlaylistSegments(Arg.Any(), Arg.Any()) + .Returns(call => playlist.Task.WaitAsync(call.Arg())); + _service.TryAddWorker("1", worker); + Task> waiting = _service.WaitForReady( + "1", worker, 1, TimeSpan.FromSeconds(5), CancellationToken.None); + + _service.RemoveWorker("1", worker); + (await waiting.WaitAsync(TimeSpan.FromSeconds(5))).IsLeft.ShouldBeTrue(); + + IHlsSessionWorker replacement = Substitute.For(); + replacement.WaitForPlaylistSegments(Arg.Any(), Arg.Any()) + .Returns(Task.CompletedTask); + _service.TryAddWorker("1", replacement); + (await _service.WaitForReady("1", replacement, 1, TimeSpan.FromSeconds(5), CancellationToken.None)) + .IsRight.ShouldBeTrue(); + (await _service.WaitForReady("1", worker, 1, TimeSpan.FromSeconds(5), CancellationToken.None)) + .IsLeft.ShouldBeTrue(); + } + [Test] public async Task LockForStart_Should_Block_A_Second_Start_On_The_Same_Channel() { diff --git a/ErsatzTV.Core.Tests/FFmpeg/SessionStartCoordinatorTests.cs b/ErsatzTV.Core.Tests/FFmpeg/SessionStartCoordinatorTests.cs new file mode 100644 index 000000000..52bcf732c --- /dev/null +++ b/ErsatzTV.Core.Tests/FFmpeg/SessionStartCoordinatorTests.cs @@ -0,0 +1,156 @@ +using ErsatzTV.Core.FFmpeg; +using ErsatzTV.Core.Interfaces.FFmpeg; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using NUnit.Framework; +using Shouldly; + +namespace ErsatzTV.Core.Tests.FFmpeg; + +[TestFixture] +public class SessionStartCoordinatorTests +{ + private FFmpegSegmenterService _service; + private int _created; + + [SetUp] + public void SetUp() + { + _service = new FFmpegSegmenterService(NullLogger.Instance); + _created = 0; + } + + [Test] + public async Task Viewers_Should_Wait_Concurrently_And_Create_Only_One_Worker() + { + var playlist = new TaskCompletionSource(); + IHlsSessionWorker worker = WaitingWorker(playlist.Task); + Task> first = Start(worker); + Task> second = Start(worker); + + // Both requests must reach the readiness wait before either playlist wait completes. + worker.ReceivedCalls().Count(call => call.GetMethodInfo().Name == nameof(worker.WaitForPlaylistSegments)) + .ShouldBe(2); + _created.ShouldBe(1); + first.IsCompleted.ShouldBeFalse(); + second.IsCompleted.ShouldBeFalse(); + + playlist.SetResult(); + (await first.WaitAsync(TimeSpan.FromSeconds(5))).IsRight.ShouldBeTrue(); + (await second.WaitAsync(TimeSpan.FromSeconds(5))).IsRight.ShouldBeTrue(); + } + + [Test] + public async Task Existing_Worker_Ending_Should_Start_One_Replacement() + { + IHlsSessionWorker existing = WaitingWorker(new TaskCompletionSource().Task); + _service.TryAddWorker("1", existing); + IHlsSessionWorker replacement = WaitingWorker(Task.CompletedTask); + Task> request = Start(replacement); + + _service.RemoveWorker("1", existing); + + (await request.WaitAsync(TimeSpan.FromSeconds(5))).IsRight.ShouldBeTrue(); + _created.ShouldBe(1); + _service.TryGetWorker("1", out IHlsSessionWorker current).ShouldBeTrue(); + current.ShouldBeSameAs(replacement); + } + + [Test] + public async Task Recovery_Should_Use_Another_Viewers_Replacement() + { + IHlsSessionWorker existing = WaitingWorker(new TaskCompletionSource().Task); + _service.TryAddWorker("1", existing); + IHlsSessionWorker replacement = WaitingWorker(Task.CompletedTask); + Task> request = Start(replacement); + + using (await _service.LockForStart("1", CancellationToken.None)) + { + _service.RemoveWorker("1", existing); + _service.TryAddWorker("1", replacement); + } + + (await request.WaitAsync(TimeSpan.FromSeconds(5))).IsRight.ShouldBeTrue(); + _created.ShouldBe(0); + } + + [Test] + public async Task Replacement_Ending_Should_Not_Cause_An_Unbounded_Restart_Loop() + { + IHlsSessionWorker existing = WaitingWorker(new TaskCompletionSource().Task); + _service.TryAddWorker("1", existing); + var replacementWait = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + IHlsSessionWorker replacement = WaitingWorker(new TaskCompletionSource().Task); + replacement.WaitForPlaylistSegments(Arg.Any(), Arg.Any()) + .Returns(call => + { + replacementWait.TrySetResult(); + return Task.Delay(Timeout.Infinite, call.Arg()); + }); + Task> request = Start(replacement); + _service.RemoveWorker("1", existing); + await replacementWait.Task.WaitAsync(TimeSpan.FromSeconds(5)); + _service.RemoveWorker("1", replacement); + + (await request.WaitAsync(TimeSpan.FromSeconds(5))).LeftToSeq().Head() + .ShouldBeOfType(); + _created.ShouldBe(1); + } + + [Test] + public async Task Deadline_Should_Not_Restart_A_Running_Worker() + { + IHlsSessionWorker existing = WaitingWorker(new TaskCompletionSource().Task); + _service.TryAddWorker("1", existing); + + Either result = await Start(existing, deadline: TimeSpan.FromMilliseconds(50)) + .WaitAsync(TimeSpan.FromSeconds(5)); + + result.IsLeft.ShouldBeTrue(); + result.LeftToSeq().Head().ShouldNotBeOfType(); + _created.ShouldBe(0); + _service.IsActive("1").ShouldBeTrue(); + } + + [Test] + public async Task Canceling_One_Viewer_Should_Not_Cancel_Another_Viewers_Wait() + { + var playlist = new TaskCompletionSource(); + IHlsSessionWorker worker = WaitingWorker(playlist.Task); + using var cts = new CancellationTokenSource(); + Task> first = Start(worker, cts.Token); + Task> second = Start(worker); + + await cts.CancelAsync(); + await Should.ThrowAsync(() => first); + second.IsCompleted.ShouldBeFalse(); + playlist.SetResult(); + (await second.WaitAsync(TimeSpan.FromSeconds(5))).IsRight.ShouldBeTrue(); + _created.ShouldBe(1); + } + + private Task> Start( + IHlsSessionWorker worker, + CancellationToken cancellationToken = default, + TimeSpan? deadline = null) => + SessionStartCoordinator.Start( + _service, + "1", + () => + { + _created++; + _service.TryAddWorker("1", worker).ShouldBeTrue(); + return Task.FromResult(Right(worker)); + }, + 1, + deadline ?? TimeSpan.FromSeconds(5), + cancellationToken); + + private static IHlsSessionWorker WaitingWorker(Task playlist) + { + IHlsSessionWorker worker = Substitute.For(); + worker.WaitForPlaylistSegments(Arg.Any(), Arg.Any()) + .Returns(call => playlist.WaitAsync(call.Arg())); + return worker; + } +} diff --git a/ErsatzTV.Core/FFmpeg/FFmpegSegmenterService.cs b/ErsatzTV.Core/FFmpeg/FFmpegSegmenterService.cs index 4d319047f..4a451f551 100644 --- a/ErsatzTV.Core/FFmpeg/FFmpegSegmenterService.cs +++ b/ErsatzTV.Core/FFmpeg/FFmpegSegmenterService.cs @@ -7,11 +7,11 @@ namespace ErsatzTV.Core.FFmpeg; public class FFmpegSegmenterService(ILogger logger) : IFFmpegSegmenterService { private readonly ConcurrentDictionary _startLocks = new(); - private readonly ConcurrentDictionary _sessionWorkers = new(); + private readonly ConcurrentDictionary _sessionWorkers = new(); public event EventHandler OnWorkersChanged; - public ICollection Workers => _sessionWorkers.Values; + public ICollection Workers => _sessionWorkers.Values.Select(session => session.Worker).ToList(); public async Task LockForStart(string channelNumber, CancellationToken cancellationToken) { @@ -20,14 +20,47 @@ public class FFmpegSegmenterService(ILogger logger) : IF return new StartLockReleaser(slim); } - public bool TryGetWorker(string channelNumber, out IHlsSessionWorker worker) => - _sessionWorkers.TryGetValue(channelNumber, out worker); + public bool TryGetWorker(string channelNumber, out IHlsSessionWorker worker) + { + bool found = _sessionWorkers.TryGetValue(channelNumber, out Session session); + worker = session?.Worker; + return found; + } + + public async Task> WaitForReady( + string channelNumber, + IHlsSessionWorker worker, + int initialSegmentCount, + TimeSpan startDeadline, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + if (!_sessionWorkers.TryGetValue(channelNumber, out Session session) || + !ReferenceEquals(session.Worker, worker)) + { + return new SessionEndedBeforeReady(channelNumber); + } + + if (session.IsReady) + { + return Unit.Default; + } + + Either result = await SessionStartWait.ForReady( + channelNumber, worker, session.Ended.Task, initialSegmentCount, startDeadline, cancellationToken); + if (result.IsRight) + { + session.IsReady = true; + } + + return result; + } public bool TryAddWorker(string channelNumber, IHlsSessionWorker worker) { ArgumentNullException.ThrowIfNull(worker); - bool result = _sessionWorkers.TryAdd(channelNumber, worker); + bool result = _sessionWorkers.TryAdd(channelNumber, new Session(worker)); if (result) { OnWorkersChanged?.Invoke(this, EventArgs.Empty); @@ -38,8 +71,11 @@ public class FFmpegSegmenterService(ILogger logger) : IF public void RemoveWorker(string channelNumber, IHlsSessionWorker worker) { - if (_sessionWorkers.TryRemove(new KeyValuePair(channelNumber, worker))) + if (_sessionWorkers.TryGetValue(channelNumber, out Session session) && + ReferenceEquals(session.Worker, worker) && + _sessionWorkers.TryRemove(new KeyValuePair(channelNumber, session))) { + session.Ended.TrySetResult(); OnWorkersChanged?.Invoke(this, EventArgs.Empty); } } @@ -48,7 +84,7 @@ public class FFmpegSegmenterService(ILogger logger) : IF public async Task StopChannel(string channelNumber, CancellationToken cancellationToken) { - if (_sessionWorkers.TryGetValue(channelNumber, out IHlsSessionWorker worker)) + if (TryGetWorker(channelNumber, out IHlsSessionWorker worker)) { await worker.Cancel(cancellationToken); return true; @@ -59,7 +95,7 @@ public class FFmpegSegmenterService(ILogger logger) : IF public void TouchChannel(string channelNumber, string fileName) { - if (_sessionWorkers.TryGetValue(channelNumber, out IHlsSessionWorker worker)) + if (TryGetWorker(channelNumber, out IHlsSessionWorker worker)) { worker.Touch(fileName); } @@ -67,7 +103,7 @@ public class FFmpegSegmenterService(ILogger logger) : IF public void PlayoutUpdated(string channelNumber) { - if (_sessionWorkers.TryGetValue(channelNumber, out IHlsSessionWorker worker)) + if (TryGetWorker(channelNumber, out IHlsSessionWorker worker)) { logger.LogInformation( "Playout has been updated for channel {ChannelNumber}, HLS segmenter will skip ahead to catch up", @@ -76,4 +112,11 @@ public class FFmpegSegmenterService(ILogger logger) : IF worker.PlayoutUpdated(); } } + + private sealed class Session(IHlsSessionWorker worker) + { + public IHlsSessionWorker Worker { get; } = worker; + public TaskCompletionSource Ended { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public volatile bool IsReady; + } } diff --git a/ErsatzTV.Core/FFmpeg/SessionEndedBeforeReady.cs b/ErsatzTV.Core/FFmpeg/SessionEndedBeforeReady.cs new file mode 100644 index 000000000..ae40ac2e1 --- /dev/null +++ b/ErsatzTV.Core/FFmpeg/SessionEndedBeforeReady.cs @@ -0,0 +1,4 @@ +namespace ErsatzTV.Core.FFmpeg; + +public sealed class SessionEndedBeforeReady(string channelNumber) + : BaseError($"Session for channel {channelNumber} ended before the playlist was ready"); diff --git a/ErsatzTV.Core/FFmpeg/SessionStartCoordinator.cs b/ErsatzTV.Core/FFmpeg/SessionStartCoordinator.cs new file mode 100644 index 000000000..01442d906 --- /dev/null +++ b/ErsatzTV.Core/FFmpeg/SessionStartCoordinator.cs @@ -0,0 +1,51 @@ +using ErsatzTV.Core.Interfaces.FFmpeg; + +namespace ErsatzTV.Core.FFmpeg; + +public static class SessionStartCoordinator +{ + public static async Task> Start( + IFFmpegSegmenterService service, + string channelNumber, + Func>> createWorker, + int initialSegmentCount, + TimeSpan startDeadline, + CancellationToken cancellationToken) + { + for (var attempt = 0; ; attempt++) + { + IHlsSessionWorker worker; + bool existing; + using (await service.LockForStart(channelNumber, cancellationToken)) + { + existing = service.TryGetWorker(channelNumber, out worker); + if (existing) + { + worker.Touch(Option.None); + } + else + { + Either created = await createWorker(); + if (created.IsLeft) + { + return created.LeftToSeq().Head(); + } + + worker = created.RightToSeq().Head(); + } + } + + // A slow startup must not queue every viewer behind a separate readiness deadline. + Either ready = await service.WaitForReady( + channelNumber, worker, initialSegmentCount, startDeadline, cancellationToken); + if (attempt == 0 && existing && ready.IsLeft && + ready.LeftToSeq().Head() is SessionEndedBeforeReady) + { + // Reacquire the lock and check for another viewer's replacement before creating one. + continue; + } + + return ready; + } + } +} diff --git a/ErsatzTV.Core/FFmpeg/SessionStartWait.cs b/ErsatzTV.Core/FFmpeg/SessionStartWait.cs index 66b34992e..2a353ce4d 100644 --- a/ErsatzTV.Core/FFmpeg/SessionStartWait.cs +++ b/ErsatzTV.Core/FFmpeg/SessionStartWait.cs @@ -19,13 +19,19 @@ public static class SessionStartWait try { - Task first = await Task.WhenAny(waitTask, runTask); - if (first == runTask) + await Task.WhenAny(waitTask, runTask); + cancellationToken.ThrowIfCancellationRequested(); + if (runTask.IsCompleted) { - return BaseError.New($"Session for channel {channelNumber} ended before the playlist was ready"); + return new SessionEndedBeforeReady(channelNumber); } await waitTask; + timeout.Token.ThrowIfCancellationRequested(); + if (runTask.IsCompleted) + { + return new SessionEndedBeforeReady(channelNumber); + } return Unit.Default; } diff --git a/ErsatzTV.Core/Interfaces/FFmpeg/IFFmpegSegmenterService.cs b/ErsatzTV.Core/Interfaces/FFmpeg/IFFmpegSegmenterService.cs index dcabfec19..f04b1539c 100644 --- a/ErsatzTV.Core/Interfaces/FFmpeg/IFFmpegSegmenterService.cs +++ b/ErsatzTV.Core/Interfaces/FFmpeg/IFFmpegSegmenterService.cs @@ -6,6 +6,12 @@ public interface IFFmpegSegmenterService event EventHandler OnWorkersChanged; Task LockForStart(string channelNumber, CancellationToken cancellationToken); bool TryGetWorker(string channelNumber, out IHlsSessionWorker worker); + Task> WaitForReady( + string channelNumber, + IHlsSessionWorker worker, + int initialSegmentCount, + TimeSpan startDeadline, + CancellationToken cancellationToken); bool TryAddWorker(string channelNumber, IHlsSessionWorker worker); void RemoveWorker(string channelNumber, IHlsSessionWorker worker); bool IsActive(string channelNumber);