From a2e7e6df1e06ffd7f1f61292b91ebb2029183f47 Mon Sep 17 00:00:00 2001 From: Jason Dove Date: Sat, 5 Mar 2022 11:26:29 -0600 Subject: [PATCH] fix thread sync bug in hls segmenter (#676) --- CHANGELOG.md | 1 + .../Commands/StartFFmpegSessionHandler.cs | 26 +++-- .../Streaming/HlsSessionWorker.cs | 97 ++++++++++++++----- .../Interfaces/FFmpeg/IHlsSessionWorker.cs | 5 +- ErsatzTV/Controllers/IptvController.cs | 15 +-- 5 files changed, 96 insertions(+), 48 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b14b6339..d9aee728f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). - Fix adding songs to collections from search results page - Fix bug scheduling mid-roll filler with content that contains one chapter - No mid-roll filler will be inserted for content with zero or one chapters +- Fix thread sync bug with `HLS Segmenter` (and `MPEG-TS`) streaming modes ### Added - Add automated error reporting via Bugsnag diff --git a/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs b/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs index 52bb5a4d3..64f721db5 100644 --- a/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs +++ b/ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs @@ -37,13 +37,13 @@ public class StartFFmpegSessionHandler : MediatR.IRequestHandler> Handle(StartFFmpegSession request, CancellationToken cancellationToken) => Validate(request) - .MapT(_ => StartProcess(request)) + .MapT(_ => StartProcess(request, cancellationToken)) // this weirdness is needed to maintain the error type (.ToEitherAsync() just gives BaseError) #pragma warning disable VSTHRD103 .Bind(v => v.ToEither().MapLeft(seq => seq.Head()).MapAsync, Unit>(identity)); #pragma warning restore VSTHRD103 - private async Task StartProcess(StartFFmpegSession request) + private async Task StartProcess(StartFFmpegSession request, CancellationToken cancellationToken) { TimeSpan idleTimeout = await _configElementRepository .GetValue(ConfigElementKey.FFmpegSegmenterTimeout) @@ -70,27 +70,33 @@ public class StartFFmpegSessionHandler : MediatR.IRequestHandler(ConfigElementKey.FFmpegInitialSegmentCount) .Map(maybeCount => maybeCount.Match(identity, () => 1)); - await WaitForPlaylistSegments(playlistFileName, initialSegmentCount, worker); + await WaitForPlaylistSegments(playlistFileName, initialSegmentCount, worker, cancellationToken); return Unit.Default; } - private async Task WaitForPlaylistSegments(string playlistFileName, int initialSegmentCount, IHlsSessionWorker worker) + private async Task WaitForPlaylistSegments( + string playlistFileName, + int initialSegmentCount, + IHlsSessionWorker worker, + CancellationToken cancellationToken) { while (!File.Exists(playlistFileName)) { - await Task.Delay(TimeSpan.FromMilliseconds(100)); + await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken); } var segmentCount = 0; while (segmentCount < initialSegmentCount) { - await Task.Delay(TimeSpan.FromMilliseconds(200)); - + await Task.Delay(TimeSpan.FromMilliseconds(200), cancellationToken); + DateTimeOffset now = DateTimeOffset.Now.AddSeconds(-30); - string[] input = await File.ReadAllLinesAsync(playlistFileName); - TrimPlaylistResult result = _hlsPlaylistFilter.TrimPlaylist(worker.PlaylistStart, now, input); - segmentCount = result.SegmentCount; + Option maybeResult = await worker.TrimPlaylist(now, cancellationToken); + foreach (TrimPlaylistResult result in maybeResult) + { + segmentCount = result.SegmentCount; + } } } diff --git a/ErsatzTV.Application/Streaming/HlsSessionWorker.cs b/ErsatzTV.Application/Streaming/HlsSessionWorker.cs index e9170eafd..6619b226f 100644 --- a/ErsatzTV.Application/Streaming/HlsSessionWorker.cs +++ b/ErsatzTV.Application/Streaming/HlsSessionWorker.cs @@ -15,6 +15,7 @@ namespace ErsatzTV.Application.Streaming; public class HlsSessionWorker : IHlsSessionWorker { + private static readonly SemaphoreSlim Slim = new(1, 1); private static int _workAheadCount; private readonly IHlsPlaylistFilter _hlsPlaylistFilter; private readonly IServiceScopeFactory _serviceScopeFactory; @@ -25,8 +26,12 @@ public class HlsSessionWorker : IHlsSessionWorker private readonly object _sync = new(); private DateTimeOffset _playlistStart; private Option _targetFramerate; + private string _channelNumber; - public HlsSessionWorker(IHlsPlaylistFilter hlsPlaylistFilter, IServiceScopeFactory serviceScopeFactory, ILogger logger) + public HlsSessionWorker( + IHlsPlaylistFilter hlsPlaylistFilter, + IServiceScopeFactory serviceScopeFactory, + ILogger logger) { _hlsPlaylistFilter = hlsPlaylistFilter; _serviceScopeFactory = serviceScopeFactory; @@ -46,13 +51,23 @@ public class HlsSessionWorker : IHlsSessionWorker } } + public async Task> TrimPlaylist( + DateTimeOffset filterBefore, + CancellationToken cancellationToken) + { + Option maybeLines = await ReadPlaylistLines(cancellationToken); + return maybeLines.Map(input => _hlsPlaylistFilter.TrimPlaylist(PlaylistStart, filterBefore, input)); + } + public async Task Run(string channelNumber, TimeSpan idleTimeout) { var cts = new CancellationTokenSource(); void Cancel(object o, ElapsedEventArgs e) => cts.Cancel(); - + try { + _channelNumber = channelNumber; + lock (_sync) { _timer = new Timer(idleTimeout.TotalMilliseconds) { AutoReset = false }; @@ -75,7 +90,7 @@ public class HlsSessionWorker : IHlsSessionWorker _playlistStart = _transcodedUntil; bool initialWorkAhead = Volatile.Read(ref _workAheadCount) < await GetWorkAheadLimit(); - if (!await Transcode(channelNumber, true, !initialWorkAhead, cancellationToken)) + if (!await Transcode(true, !initialWorkAhead, cancellationToken)) { return; } @@ -96,14 +111,14 @@ public class HlsSessionWorker : IHlsSessionWorker bool realtime = transcodedBuffer >= TimeSpan.FromSeconds(30); bool subsequentWorkAhead = !realtime && Volatile.Read(ref _workAheadCount) < await GetWorkAheadLimit(); - if (!await Transcode(channelNumber, false, !subsequentWorkAhead, cancellationToken)) + if (!await Transcode(false, !subsequentWorkAhead, cancellationToken)) { return; } } else { - await TrimAndDelete(channelNumber, cancellationToken); + await TrimAndDelete(cancellationToken); await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken); } } @@ -118,7 +133,6 @@ public class HlsSessionWorker : IHlsSessionWorker } private async Task Transcode( - string channelNumber, bool firstProcess, bool realtime, CancellationToken cancellationToken) @@ -130,22 +144,22 @@ public class HlsSessionWorker : IHlsSessionWorker if (!realtime) { Interlocked.Increment(ref _workAheadCount); - _logger.LogInformation("HLS segmenter will work ahead for channel {Channel}", channelNumber); + _logger.LogInformation("HLS segmenter will work ahead for channel {Channel}", _channelNumber); } else { _logger.LogInformation( "HLS segmenter will NOT work ahead for channel {Channel}", - channelNumber); + _channelNumber); } IMediator mediator = scope.ServiceProvider.GetRequiredService(); - long ptsOffset = await GetPtsOffset(mediator, channelNumber, cancellationToken); + long ptsOffset = await GetPtsOffset(mediator, _channelNumber, cancellationToken); // _logger.LogInformation("PTS offset: {PtsOffset}", ptsOffset); var request = new GetPlayoutItemProcessByChannelNumber( - channelNumber, + _channelNumber, "segmenter", firstProcess ? DateTimeOffset.Now : _transcodedUntil.AddSeconds(1), !firstProcess, @@ -163,7 +177,7 @@ public class HlsSessionWorker : IHlsSessionWorker { _logger.LogWarning( "Failed to create process for HLS session on channel {Channel}: {Error}", - channelNumber, + _channelNumber, error.ToString()); return false; @@ -171,7 +185,7 @@ public class HlsSessionWorker : IHlsSessionWorker foreach (PlayoutItemProcessModel processModel in result.RightAsEnumerable()) { - await TrimAndDelete(channelNumber, cancellationToken); + await TrimAndDelete(cancellationToken); Process process = processModel.Process; @@ -187,21 +201,21 @@ public class HlsSessionWorker : IHlsSessionWorker } catch (TaskCanceledException) { - _logger.LogInformation("Terminating HLS process for channel {Channel}", channelNumber); + _logger.LogInformation("Terminating HLS process for channel {Channel}", _channelNumber); process.Kill(); process.WaitForExit(); return false; } - _logger.LogInformation("HLS process has completed for channel {Channel}", channelNumber); + _logger.LogInformation("HLS process has completed for channel {Channel}", _channelNumber); _transcodedUntil = processModel.Until; } } catch (Exception ex) { - _logger.LogError(ex, "Error transcoding channel {Channel}", channelNumber); + _logger.LogError(ex, "Error transcoding channel {Channel}", _channelNumber); try { @@ -223,26 +237,21 @@ public class HlsSessionWorker : IHlsSessionWorker return true; } - private async Task TrimAndDelete(string channelNumber, CancellationToken cancellationToken) + private async Task TrimAndDelete(CancellationToken cancellationToken) { - string playlistFileName = Path.Combine( - FileSystemLayout.TranscodeFolder, - channelNumber, - "live.m3u8"); - - if (File.Exists(playlistFileName)) + Option maybeLines = await ReadPlaylistLines(cancellationToken); + foreach (string[] lines in maybeLines) { // trim playlist and insert discontinuity before appending with new ffmpeg process - string[] lines = await File.ReadAllLinesAsync(playlistFileName, cancellationToken); TrimPlaylistResult trimResult = _hlsPlaylistFilter.TrimPlaylistWithDiscontinuity( _playlistStart, DateTimeOffset.Now.AddMinutes(-1), lines); - await File.WriteAllTextAsync(playlistFileName, trimResult.Playlist, cancellationToken); + await WritePlaylist(trimResult.Playlist, cancellationToken); // delete old segments var allSegments = Directory.GetFiles( - Path.Combine(FileSystemLayout.TranscodeFolder, channelNumber), + Path.Combine(FileSystemLayout.TranscodeFolder, _channelNumber), "live*.ts") .Map( file => @@ -302,5 +311,43 @@ public class HlsSessionWorker : IHlsSessionWorker .Map(maybeCount => maybeCount.Match(identity, () => 1)); } + private async Task> ReadPlaylistLines(CancellationToken cancellationToken) + { + await Slim.WaitAsync(cancellationToken); + try + { + string fileName = PlaylistFileName(); + if (File.Exists(fileName)) + { + return await File.ReadAllLinesAsync(fileName, cancellationToken); + } + + return None; + } + finally + { + Slim.Release(); + } + } + + private async Task WritePlaylist(string playlist, CancellationToken cancellationToken) + { + await Slim.WaitAsync(cancellationToken); + try + { + string fileName = PlaylistFileName(); + await File.WriteAllTextAsync(fileName, playlist, cancellationToken); + } + finally + { + Slim.Release(); + } + } + + private string PlaylistFileName() => Path.Combine( + FileSystemLayout.TranscodeFolder, + _channelNumber, + "live.m3u8"); + private record Segment(string File, int SequenceNumber); } \ No newline at end of file diff --git a/ErsatzTV.Core/Interfaces/FFmpeg/IHlsSessionWorker.cs b/ErsatzTV.Core/Interfaces/FFmpeg/IHlsSessionWorker.cs index 406c157fa..08fcb0711 100644 --- a/ErsatzTV.Core/Interfaces/FFmpeg/IHlsSessionWorker.cs +++ b/ErsatzTV.Core/Interfaces/FFmpeg/IHlsSessionWorker.cs @@ -1,7 +1,10 @@ -namespace ErsatzTV.Core.Interfaces.FFmpeg; +using ErsatzTV.Core.FFmpeg; + +namespace ErsatzTV.Core.Interfaces.FFmpeg; public interface IHlsSessionWorker { DateTimeOffset PlaylistStart { get; } void Touch(); + Task> TrimPlaylist(DateTimeOffset filterBefore, CancellationToken cancellationToken); } \ No newline at end of file diff --git a/ErsatzTV/Controllers/IptvController.cs b/ErsatzTV/Controllers/IptvController.cs index 5e86a6a73..a5fd9a90e 100644 --- a/ErsatzTV/Controllers/IptvController.cs +++ b/ErsatzTV/Controllers/IptvController.cs @@ -76,23 +76,14 @@ public class IptvController : ControllerBase } [HttpGet("iptv/session/{channelNumber}/hls.m3u8")] - public async Task GetLivePlaylist(string channelNumber) + public async Task GetLivePlaylist(string channelNumber, CancellationToken cancellationToken) { if (_ffmpegSegmenterService.SessionWorkers.TryGetValue(channelNumber, out IHlsSessionWorker worker)) { DateTimeOffset now = DateTimeOffset.Now.AddSeconds(-30); - - string fileName = Path.Combine(FileSystemLayout.TranscodeFolder, channelNumber, "live.m3u8"); - if (System.IO.File.Exists(fileName)) + Option maybePlaylist = await worker.TrimPlaylist(now, cancellationToken); + foreach (TrimPlaylistResult result in maybePlaylist) { - string[] input = await System.IO.File.ReadAllLinesAsync(fileName); - - // _logger.LogInformation( - // "Trimming playlist: {PlaylistStart} {FilterBefore}", - // worker.PlaylistStart, - // now); - - TrimPlaylistResult result = _hlsPlaylistFilter.TrimPlaylist(worker.PlaylistStart, now, input); return Content(result.Playlist, "application/vnd.apple.mpegurl"); } }