From abd74fd5a481d67af1414960be7b7b19c9ee7e82 Mon Sep 17 00:00:00 2001
From: Patrick Barron <barronpm@gmail.com>
Date: Tue, 31 Oct 2023 11:12:09 -0400
Subject: Move TranscodingJobDto and TranscodingThrottler to Controller

---
 .../MediaEncoding/TranscodingJobDto.cs             | 282 +++++++++++++++++++++
 .../MediaEncoding/TranscodingThrottler.cs          | 218 ++++++++++++++++
 2 files changed, 500 insertions(+)
 create mode 100644 MediaBrowser.Controller/MediaEncoding/TranscodingJobDto.cs
 create mode 100644 MediaBrowser.Controller/MediaEncoding/TranscodingThrottler.cs

(limited to 'MediaBrowser.Controller/MediaEncoding')

diff --git a/MediaBrowser.Controller/MediaEncoding/TranscodingJobDto.cs b/MediaBrowser.Controller/MediaEncoding/TranscodingJobDto.cs
new file mode 100644
index 000000000..6f929204f
--- /dev/null
+++ b/MediaBrowser.Controller/MediaEncoding/TranscodingJobDto.cs
@@ -0,0 +1,282 @@
+using System;
+using System.Diagnostics;
+using System.Diagnostics.CodeAnalysis;
+using System.Threading;
+using MediaBrowser.Model.Dto;
+using Microsoft.Extensions.Logging;
+
+namespace MediaBrowser.Controller.MediaEncoding;
+
+/// <summary>
+/// Class TranscodingJob.
+/// </summary>
+public class TranscodingJobDto : IDisposable
+{
+    /// <summary>
+    /// The process lock.
+    /// </summary>
+    [SuppressMessage("Microsoft.Performance", "CA1051:NoVisibleInstanceFields", MessageId = "ProcessLock", Justification = "Imported from ServiceStack")]
+    [SuppressMessage("Microsoft.Performance", "SA1401:PrivateField", MessageId = "ProcessLock", Justification = "Imported from ServiceStack")]
+    public readonly object ProcessLock = new object();
+
+    /// <summary>
+    /// Timer lock.
+    /// </summary>
+    private readonly object _timerLock = new object();
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="TranscodingJobDto"/> class.
+    /// </summary>
+    /// <param name="logger">Instance of the <see cref="ILogger{TranscodingJobDto}"/> interface.</param>
+    public TranscodingJobDto(ILogger<TranscodingJobDto> logger)
+    {
+        Logger = logger;
+    }
+
+    /// <summary>
+    /// Gets or sets the play session identifier.
+    /// </summary>
+    /// <value>The play session identifier.</value>
+    public string? PlaySessionId { get; set; }
+
+    /// <summary>
+    /// Gets or sets the live stream identifier.
+    /// </summary>
+    /// <value>The live stream identifier.</value>
+    public string? LiveStreamId { get; set; }
+
+    /// <summary>
+    /// Gets or sets a value indicating whether is live output.
+    /// </summary>
+    public bool IsLiveOutput { get; set; }
+
+    /// <summary>
+    /// Gets or sets the path.
+    /// </summary>
+    /// <value>The path.</value>
+    public MediaSourceInfo? MediaSource { get; set; }
+
+    /// <summary>
+    /// Gets or sets path.
+    /// </summary>
+    public string? Path { get; set; }
+
+    /// <summary>
+    /// Gets or sets the type.
+    /// </summary>
+    /// <value>The type.</value>
+    public TranscodingJobType Type { get; set; }
+
+    /// <summary>
+    /// Gets or sets the process.
+    /// </summary>
+    /// <value>The process.</value>
+    public Process? Process { get; set; }
+
+    /// <summary>
+    /// Gets logger.
+    /// </summary>
+    public ILogger<TranscodingJobDto> Logger { get; private set; }
+
+    /// <summary>
+    /// Gets or sets the active request count.
+    /// </summary>
+    /// <value>The active request count.</value>
+    public int ActiveRequestCount { get; set; }
+
+    /// <summary>
+    /// Gets or sets the kill timer.
+    /// </summary>
+    /// <value>The kill timer.</value>
+    private Timer? KillTimer { get; set; }
+
+    /// <summary>
+    /// Gets or sets device id.
+    /// </summary>
+    public string? DeviceId { get; set; }
+
+    /// <summary>
+    /// Gets or sets cancellation token source.
+    /// </summary>
+    public CancellationTokenSource? CancellationTokenSource { get; set; }
+
+    /// <summary>
+    /// Gets or sets a value indicating whether has exited.
+    /// </summary>
+    public bool HasExited { get; set; }
+
+    /// <summary>
+    /// Gets or sets exit code.
+    /// </summary>
+    public int ExitCode { get; set; }
+
+    /// <summary>
+    /// Gets or sets a value indicating whether is user paused.
+    /// </summary>
+    public bool IsUserPaused { get; set; }
+
+    /// <summary>
+    /// Gets or sets id.
+    /// </summary>
+    public string? Id { get; set; }
+
+    /// <summary>
+    /// Gets or sets framerate.
+    /// </summary>
+    public float? Framerate { get; set; }
+
+    /// <summary>
+    /// Gets or sets completion percentage.
+    /// </summary>
+    public double? CompletionPercentage { get; set; }
+
+    /// <summary>
+    /// Gets or sets bytes downloaded.
+    /// </summary>
+    public long BytesDownloaded { get; set; }
+
+    /// <summary>
+    /// Gets or sets bytes transcoded.
+    /// </summary>
+    public long? BytesTranscoded { get; set; }
+
+    /// <summary>
+    /// Gets or sets bit rate.
+    /// </summary>
+    public int? BitRate { get; set; }
+
+    /// <summary>
+    /// Gets or sets transcoding position ticks.
+    /// </summary>
+    public long? TranscodingPositionTicks { get; set; }
+
+    /// <summary>
+    /// Gets or sets download position ticks.
+    /// </summary>
+    public long? DownloadPositionTicks { get; set; }
+
+    /// <summary>
+    /// Gets or sets transcoding throttler.
+    /// </summary>
+    public TranscodingThrottler? TranscodingThrottler { get; set; }
+
+    /// <summary>
+    /// Gets or sets last ping date.
+    /// </summary>
+    public DateTime LastPingDate { get; set; }
+
+    /// <summary>
+    /// Gets or sets ping timeout.
+    /// </summary>
+    public int PingTimeout { get; set; }
+
+    /// <summary>
+    /// Stop kill timer.
+    /// </summary>
+    public void StopKillTimer()
+    {
+        lock (_timerLock)
+        {
+            KillTimer?.Change(Timeout.Infinite, Timeout.Infinite);
+        }
+    }
+
+    /// <summary>
+    /// Dispose kill timer.
+    /// </summary>
+    public void DisposeKillTimer()
+    {
+        lock (_timerLock)
+        {
+            if (KillTimer is not null)
+            {
+                KillTimer.Dispose();
+                KillTimer = null;
+            }
+        }
+    }
+
+    /// <summary>
+    /// Start kill timer.
+    /// </summary>
+    /// <param name="callback">Callback action.</param>
+    public void StartKillTimer(Action<object?> callback)
+    {
+        StartKillTimer(callback, PingTimeout);
+    }
+
+    /// <summary>
+    /// Start kill timer.
+    /// </summary>
+    /// <param name="callback">Callback action.</param>
+    /// <param name="intervalMs">Callback interval.</param>
+    public void StartKillTimer(Action<object?> callback, int intervalMs)
+    {
+        if (HasExited)
+        {
+            return;
+        }
+
+        lock (_timerLock)
+        {
+            if (KillTimer is null)
+            {
+                Logger.LogDebug("Starting kill timer at {0}ms. JobId {1} PlaySessionId {2}", intervalMs, Id, PlaySessionId);
+                KillTimer = new Timer(new TimerCallback(callback), this, intervalMs, Timeout.Infinite);
+            }
+            else
+            {
+                Logger.LogDebug("Changing kill timer to {0}ms. JobId {1} PlaySessionId {2}", intervalMs, Id, PlaySessionId);
+                KillTimer.Change(intervalMs, Timeout.Infinite);
+            }
+        }
+    }
+
+    /// <summary>
+    /// Change kill timer if started.
+    /// </summary>
+    public void ChangeKillTimerIfStarted()
+    {
+        if (HasExited)
+        {
+            return;
+        }
+
+        lock (_timerLock)
+        {
+            if (KillTimer is not null)
+            {
+                var intervalMs = PingTimeout;
+
+                Logger.LogDebug("Changing kill timer to {0}ms. JobId {1} PlaySessionId {2}", intervalMs, Id, PlaySessionId);
+                KillTimer.Change(intervalMs, Timeout.Infinite);
+            }
+        }
+    }
+
+    /// <inheritdoc />
+    public void Dispose()
+    {
+        Dispose(true);
+        GC.SuppressFinalize(this);
+    }
+
+    /// <summary>
+    /// Dispose all resources.
+    /// </summary>
+    /// <param name="disposing">Whether to dispose all resources.</param>
+    protected virtual void Dispose(bool disposing)
+    {
+        if (disposing)
+        {
+            Process?.Dispose();
+            Process = null;
+            KillTimer?.Dispose();
+            KillTimer = null;
+            CancellationTokenSource?.Dispose();
+            CancellationTokenSource = null;
+            TranscodingThrottler?.Dispose();
+            TranscodingThrottler = null;
+        }
+    }
+}
diff --git a/MediaBrowser.Controller/MediaEncoding/TranscodingThrottler.cs b/MediaBrowser.Controller/MediaEncoding/TranscodingThrottler.cs
new file mode 100644
index 000000000..aa08af54f
--- /dev/null
+++ b/MediaBrowser.Controller/MediaEncoding/TranscodingThrottler.cs
@@ -0,0 +1,218 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using MediaBrowser.Common.Configuration;
+using MediaBrowser.Model.Configuration;
+using MediaBrowser.Model.IO;
+using Microsoft.Extensions.Logging;
+
+namespace MediaBrowser.Controller.MediaEncoding;
+
+/// <summary>
+/// Transcoding throttler.
+/// </summary>
+public class TranscodingThrottler : IDisposable
+{
+    private readonly TranscodingJobDto _job;
+    private readonly ILogger<TranscodingThrottler> _logger;
+    private readonly IConfigurationManager _config;
+    private readonly IFileSystem _fileSystem;
+    private readonly IMediaEncoder _mediaEncoder;
+    private Timer? _timer;
+    private bool _isPaused;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="TranscodingThrottler"/> class.
+    /// </summary>
+    /// <param name="job">Transcoding job dto.</param>
+    /// <param name="logger">Instance of the <see cref="ILogger{TranscodingThrottler}"/> interface.</param>
+    /// <param name="config">Instance of the <see cref="IConfigurationManager"/> interface.</param>
+    /// <param name="fileSystem">Instance of the <see cref="IFileSystem"/> interface.</param>
+    /// <param name="mediaEncoder">Instance of the <see cref="IMediaEncoder"/> interface.</param>
+    public TranscodingThrottler(TranscodingJobDto job, ILogger<TranscodingThrottler> logger, IConfigurationManager config, IFileSystem fileSystem, IMediaEncoder mediaEncoder)
+    {
+        _job = job;
+        _logger = logger;
+        _config = config;
+        _fileSystem = fileSystem;
+        _mediaEncoder = mediaEncoder;
+    }
+
+    /// <summary>
+    /// Start timer.
+    /// </summary>
+    public void Start()
+    {
+        _timer = new Timer(TimerCallback, null, 5000, 5000);
+    }
+
+    /// <summary>
+    /// Unpause transcoding.
+    /// </summary>
+    /// <returns>A <see cref="Task"/>.</returns>
+    public async Task UnpauseTranscoding()
+    {
+        if (_isPaused)
+        {
+            _logger.LogDebug("Sending resume command to ffmpeg");
+
+            try
+            {
+                var resumeKey = _mediaEncoder.IsPkeyPauseSupported ? "u" : Environment.NewLine;
+                await _job.Process!.StandardInput.WriteAsync(resumeKey).ConfigureAwait(false);
+                _isPaused = false;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error resuming transcoding");
+            }
+        }
+    }
+
+    /// <summary>
+    /// Stop throttler.
+    /// </summary>
+    /// <returns>A <see cref="Task"/>.</returns>
+    public async Task Stop()
+    {
+        DisposeTimer();
+        await UnpauseTranscoding().ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// Dispose throttler.
+    /// </summary>
+    public void Dispose()
+    {
+        Dispose(true);
+        GC.SuppressFinalize(this);
+    }
+
+    /// <summary>
+    /// Dispose throttler.
+    /// </summary>
+    /// <param name="disposing">Disposing.</param>
+    protected virtual void Dispose(bool disposing)
+    {
+        if (disposing)
+        {
+            DisposeTimer();
+        }
+    }
+
+    private EncodingOptions GetOptions()
+    {
+        return _config.GetEncodingOptions();
+    }
+
+    private async void TimerCallback(object? state)
+    {
+        if (_job.HasExited)
+        {
+            DisposeTimer();
+            return;
+        }
+
+        var options = GetOptions();
+
+        if (options.EnableThrottling && IsThrottleAllowed(_job, options.ThrottleDelaySeconds))
+        {
+            await PauseTranscoding().ConfigureAwait(false);
+        }
+        else
+        {
+            await UnpauseTranscoding().ConfigureAwait(false);
+        }
+    }
+
+    private async Task PauseTranscoding()
+    {
+        if (!_isPaused)
+        {
+            var pauseKey = _mediaEncoder.IsPkeyPauseSupported ? "p" : "c";
+
+            _logger.LogDebug("Sending pause command [{Key}] to ffmpeg", pauseKey);
+
+            try
+            {
+                await _job.Process!.StandardInput.WriteAsync(pauseKey).ConfigureAwait(false);
+                _isPaused = true;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error pausing transcoding");
+            }
+        }
+    }
+
+    private bool IsThrottleAllowed(TranscodingJobDto job, int thresholdSeconds)
+    {
+        var bytesDownloaded = job.BytesDownloaded;
+        var transcodingPositionTicks = job.TranscodingPositionTicks ?? 0;
+        var downloadPositionTicks = job.DownloadPositionTicks ?? 0;
+
+        var path = job.Path ?? throw new ArgumentException("Path can't be null.");
+
+        var gapLengthInTicks = TimeSpan.FromSeconds(thresholdSeconds).Ticks;
+
+        if (downloadPositionTicks > 0 && transcodingPositionTicks > 0)
+        {
+            // HLS - time-based consideration
+
+            var targetGap = gapLengthInTicks;
+            var gap = transcodingPositionTicks - downloadPositionTicks;
+
+            if (gap < targetGap)
+            {
+                _logger.LogDebug("Not throttling transcoder gap {0} target gap {1}", gap, targetGap);
+                return false;
+            }
+
+            _logger.LogDebug("Throttling transcoder gap {0} target gap {1}", gap, targetGap);
+            return true;
+        }
+
+        if (bytesDownloaded > 0 && transcodingPositionTicks > 0)
+        {
+            // Progressive Streaming - byte-based consideration
+
+            try
+            {
+                var bytesTranscoded = job.BytesTranscoded ?? _fileSystem.GetFileInfo(path).Length;
+
+                // Estimate the bytes the transcoder should be ahead
+                double gapFactor = gapLengthInTicks;
+                gapFactor /= transcodingPositionTicks;
+                var targetGap = bytesTranscoded * gapFactor;
+
+                var gap = bytesTranscoded - bytesDownloaded;
+
+                if (gap < targetGap)
+                {
+                    _logger.LogDebug("Not throttling transcoder gap {0} target gap {1} bytes downloaded {2}", gap, targetGap, bytesDownloaded);
+                    return false;
+                }
+
+                _logger.LogDebug("Throttling transcoder gap {0} target gap {1} bytes downloaded {2}", gap, targetGap, bytesDownloaded);
+                return true;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error getting output size");
+                return false;
+            }
+        }
+
+        _logger.LogDebug("No throttle data for {Path}", path);
+        return false;
+    }
+
+    private void DisposeTimer()
+    {
+        if (_timer is not null)
+        {
+            _timer.Dispose();
+            _timer = null;
+        }
+    }
+}
-- 
cgit v1.2.3