fix thread sync bug in hls segmenter (#676)

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

ErsatzTV.Application/Streaming/Commands/StartFFmpegSessionHandler.cs

@@ -37,13 +37,13 @@ public class StartFFmpegSessionHandler : MediatR.IRequestHandler<StartFFmpegSess
 
     public Task<Either<BaseError, Unit>> 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<BaseError, Task<Unit>, Unit>(identity));
 #pragma warning restore VSTHRD103
 
-    private async Task<Unit> StartProcess(StartFFmpegSession request)
+    private async Task<Unit> StartProcess(StartFFmpegSession request, CancellationToken cancellationToken)
     {
         TimeSpan idleTimeout = await _configElementRepository
             .GetValue<int>(ConfigElementKey.FFmpegSegmenterTimeout)
@@ -70,27 +70,33 @@ public class StartFFmpegSessionHandler : MediatR.IRequestHandler<StartFFmpegSess
         int initialSegmentCount = await repo.GetValue<int>(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);
+            Option<TrimPlaylistResult> maybeResult = await worker.TrimPlaylist(now, cancellationToken);
-            TrimPlaylistResult result = _hlsPlaylistFilter.TrimPlaylist(worker.PlaylistStart, now, input);
+            foreach (TrimPlaylistResult result in maybeResult)
-            segmentCount = result.SegmentCount;
+            {
+                segmentCount = result.SegmentCount;
+            }
         }
     }
 

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<int> _targetFramerate;
+    private string _channelNumber;
 
-    public HlsSessionWorker(IHlsPlaylistFilter hlsPlaylistFilter, IServiceScopeFactory serviceScopeFactory, ILogger<HlsSessionWorker> logger)
+    public HlsSessionWorker(
+        IHlsPlaylistFilter hlsPlaylistFilter,
+        IServiceScopeFactory serviceScopeFactory,
+        ILogger<HlsSessionWorker> logger)
     {
         _hlsPlaylistFilter = hlsPlaylistFilter;
         _serviceScopeFactory = serviceScopeFactory;
@@ -46,13 +51,23 @@ public class HlsSessionWorker : IHlsSessionWorker
         }
     }
 
+    public async Task<Option<TrimPlaylistResult>> TrimPlaylist(
+        DateTimeOffset filterBefore,
+        CancellationToken cancellationToken)
+    {
+        Option<string[]> 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<bool> 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<IMediator>();
 
-            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(
+        Option<string[]> maybeLines = await ReadPlaylistLines(cancellationToken);
-            FileSystemLayout.TranscodeFolder,
+        foreach (string[] lines in maybeLines)
-            channelNumber,
-            "live.m3u8");
-
-        if (File.Exists(playlistFileName))
         {
             // 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<Option<string[]>> 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);
 }

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<Option<TrimPlaylistResult>> TrimPlaylist(DateTimeOffset filterBefore, CancellationToken cancellationToken);
 }

ErsatzTV/Controllers/IptvController.cs

@@ -76,23 +76,14 @@ public class IptvController : ControllerBase
     }
 
     [HttpGet("iptv/session/{channelNumber}/hls.m3u8")]
-    public async Task<IActionResult> GetLivePlaylist(string channelNumber)
+    public async Task<IActionResult> GetLivePlaylist(string channelNumber, CancellationToken cancellationToken)
     {
         if (_ffmpegSegmenterService.SessionWorkers.TryGetValue(channelNumber, out IHlsSessionWorker worker))
         {
             DateTimeOffset now = DateTimeOffset.Now.AddSeconds(-30);
-            
+            Option<TrimPlaylistResult> maybePlaylist = await worker.TrimPlaylist(now, cancellationToken);
-            string fileName = Path.Combine(FileSystemLayout.TranscodeFolder, channelNumber, "live.m3u8");
+            foreach (TrimPlaylistResult result in maybePlaylist)
-            if (System.IO.File.Exists(fileName))
             {
-                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");
             }
         }