From 9be20cfaa612bf71c777c7fe4d4c6b7b0e740f11 Mon Sep 17 00:00:00 2001 From: David Date: Thu, 9 Jul 2020 12:44:56 +0200 Subject: [PATCH] Add jellyfin api entry point --- Jellyfin.Api/JellyfinApiEntryPoint.cs | 405 ++++++++++++++++++ .../Models/TranscodingDtos/TranscodingJob.cs | 239 +++++++++++ 2 files changed, 644 insertions(+) create mode 100644 Jellyfin.Api/JellyfinApiEntryPoint.cs create mode 100644 Jellyfin.Api/Models/TranscodingDtos/TranscodingJob.cs diff --git a/Jellyfin.Api/JellyfinApiEntryPoint.cs b/Jellyfin.Api/JellyfinApiEntryPoint.cs new file mode 100644 index 0000000000..7594abc5d5 --- /dev/null +++ b/Jellyfin.Api/JellyfinApiEntryPoint.cs @@ -0,0 +1,405 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Jellyfin.Api.Models.TranscodingDtos; +using MediaBrowser.Common.Configuration; +using MediaBrowser.Controller.Configuration; +using MediaBrowser.Controller.Library; +using MediaBrowser.Controller.MediaEncoding; +using MediaBrowser.Controller.Plugins; +using MediaBrowser.Controller.Session; +using MediaBrowser.Model.IO; +using Microsoft.Extensions.Logging; + +namespace Jellyfin.Api +{ + /// + /// The jellyfin api entry point. + /// + public class JellyfinApiEntryPoint : IServerEntryPoint + { + private readonly ILogger _logger; + private readonly IServerConfigurationManager _serverConfigurationManager; + private readonly ISessionManager _sessionManager; + private readonly IFileSystem _fileSystem; + private readonly IMediaSourceManager _mediaSourceManager; + private readonly List _activeTranscodingJobs; + private readonly Dictionary _transcodingLocks; + private bool _disposed = false; + + /// + /// Initializes a new instance of the class. + /// + /// The logger. + /// The session manager. + /// The configuration. + /// The file system. + /// The media source manager. + public JellyfinApiEntryPoint( + ILogger logger, + ISessionManager sessionManager, + IServerConfigurationManager config, + IFileSystem fileSystem, + IMediaSourceManager mediaSourceManager) + { + _logger = logger; + _sessionManager = sessionManager; + _serverConfigurationManager = config; + _fileSystem = fileSystem; + _mediaSourceManager = mediaSourceManager; + + _activeTranscodingJobs = new List(); + _transcodingLocks = new Dictionary(); + + _sessionManager!.PlaybackProgress += OnPlaybackProgress; + _sessionManager!.PlaybackStart += OnPlaybackProgress; + + Instance = this; + } + + /// + /// Gets the initialized instance of . + /// + public static JellyfinApiEntryPoint? Instance { get; private set; } + + /// + public void Dispose() + { + Dispose(true); + GC.SuppressFinalize(this); + } + + /// + /// Releases unmanaged and - optionally - managed resources. + /// + /// true to release both managed and unmanaged resources; false to release only unmanaged resources. + protected virtual void Dispose(bool dispose) + { + if (_disposed) + { + return; + } + + if (dispose) + { + // TODO: dispose + } + + List jobs; + lock (_activeTranscodingJobs) + { + jobs = _activeTranscodingJobs.ToList(); + } + + var jobCount = jobs.Count; + + IEnumerable GetKillJobs() + { + foreach (var job in jobs) + { + yield return KillTranscodingJob(job, false, path => true); + } + } + + // Wait for all processes to be killed + if (jobCount > 0) + { + Task.WaitAll(GetKillJobs().ToArray()); + } + + lock (_activeTranscodingJobs) + { + _activeTranscodingJobs.Clear(); + } + + lock (_transcodingLocks) + { + _transcodingLocks.Clear(); + } + + _sessionManager.PlaybackProgress -= OnPlaybackProgress; + _sessionManager.PlaybackStart -= OnPlaybackProgress; + + _disposed = true; + } + + /// + public Task RunAsync() + { + try + { + DeleteEncodedMediaCache(); + } + catch (Exception ex) + { + _logger.LogError(ex, "Error deleting encoded media cache"); + } + + return Task.CompletedTask; + } + + private void OnPlaybackProgress(object sender, PlaybackProgressEventArgs e) + { + if (!string.IsNullOrWhiteSpace(e.PlaySessionId)) + { + PingTranscodingJob(e.PlaySessionId, e.IsPaused); + } + } + + /// + /// Deletes the encoded media cache. + /// + private void DeleteEncodedMediaCache() + { + var path = _serverConfigurationManager.GetTranscodePath(); + if (!Directory.Exists(path)) + { + return; + } + + foreach (var file in _fileSystem.GetFilePaths(path, true)) + { + _fileSystem.DeleteFile(file); + } + } + + internal void PingTranscodingJob(string playSessionId, bool? isUserPaused) + { + if (string.IsNullOrEmpty(playSessionId)) + { + throw new ArgumentNullException(nameof(playSessionId)); + } + + _logger.LogDebug("PingTranscodingJob PlaySessionId={0} isUsedPaused: {1}", playSessionId, isUserPaused); + + List jobs; + + lock (_activeTranscodingJobs) + { + // This is really only needed for HLS. + // Progressive streams can stop on their own reliably + jobs = _activeTranscodingJobs.Where(j => string.Equals(playSessionId, j.PlaySessionId, StringComparison.OrdinalIgnoreCase)).ToList(); + } + + foreach (var job in jobs) + { + if (isUserPaused.HasValue) + { + _logger.LogDebug("Setting job.IsUserPaused to {0}. jobId: {1}", isUserPaused, job.Id); + job.IsUserPaused = isUserPaused.Value; + } + + PingTimer(job, true); + } + } + + private void PingTimer(TranscodingJob job, bool isProgressCheckIn) + { + if (job.HasExited) + { + job.StopKillTimer(); + return; + } + + var timerDuration = 10000; + + if (job.Type != TranscodingJobType.Progressive) + { + timerDuration = 60000; + } + + job.PingTimeout = timerDuration; + job.LastPingDate = DateTime.UtcNow; + + // Don't start the timer for playback checkins with progressive streaming + if (job.Type != TranscodingJobType.Progressive || !isProgressCheckIn) + { + job.StartKillTimer(OnTranscodeKillTimerStopped); + } + else + { + job.ChangeKillTimerIfStarted(); + } + } + + /// + /// Called when [transcode kill timer stopped]. + /// + /// The state. + private async void OnTranscodeKillTimerStopped(object state) + { + var job = (TranscodingJob)state; + + if (!job.HasExited && job.Type != TranscodingJobType.Progressive) + { + var timeSinceLastPing = (DateTime.UtcNow - job.LastPingDate).TotalMilliseconds; + + if (timeSinceLastPing < job.PingTimeout) + { + job.StartKillTimer(OnTranscodeKillTimerStopped, job.PingTimeout); + return; + } + } + + _logger.LogInformation("Transcoding kill timer stopped for JobId {0} PlaySessionId {1}. Killing transcoding", job.Id, job.PlaySessionId); + + await KillTranscodingJob(job, true, path => true).ConfigureAwait(false); + } + + /// + /// Kills the transcoding job. + /// + /// The job. + /// if set to true [close live stream]. + /// The delete. + private async Task KillTranscodingJob(TranscodingJob job, bool closeLiveStream, Func delete) + { + job.DisposeKillTimer(); + + _logger.LogDebug("KillTranscodingJob - JobId {0} PlaySessionId {1}. Killing transcoding", job.Id, job.PlaySessionId); + + lock (_activeTranscodingJobs) + { + _activeTranscodingJobs.Remove(job); + + if (!job.CancellationTokenSource!.IsCancellationRequested) + { + job.CancellationTokenSource.Cancel(); + } + } + + lock (_transcodingLocks) + { + _transcodingLocks.Remove(job.Path!); + } + + lock (job) + { + job.TranscodingThrottler?.Stop().GetAwaiter().GetResult(); + + var process = job.Process; + + var hasExited = job.HasExited; + + if (!hasExited) + { + try + { + _logger.LogInformation("Stopping ffmpeg process with q command for {Path}", job.Path); + + process?.StandardInput.WriteLine("q"); + + // Need to wait (an arbitrary amount of time) because killing is asynchronous + if (!process!.WaitForExit(5000)) + { + _logger.LogInformation("Killing ffmpeg process for {Path}", job.Path); + process.Kill(); + } + } + catch (InvalidOperationException) + { + } + } + } + + if (delete(job.Path!)) + { + await DeletePartialStreamFiles(job.Path!, job.Type, 0, 1500).ConfigureAwait(false); + } + + if (closeLiveStream && !string.IsNullOrWhiteSpace(job.LiveStreamId)) + { + try + { + await _mediaSourceManager.CloseLiveStream(job.LiveStreamId).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogError(ex, "Error closing live stream for {Path}", job.Path); + } + } + } + + private async Task DeletePartialStreamFiles(string path, TranscodingJobType jobType, int retryCount, int delayMs) + { + if (retryCount >= 10) + { + return; + } + + _logger.LogInformation("Deleting partial stream file(s) {Path}", path); + + await Task.Delay(delayMs).ConfigureAwait(false); + + try + { + if (jobType == TranscodingJobType.Progressive) + { + DeleteProgressivePartialStreamFiles(path); + } + else + { + DeleteHlsPartialStreamFiles(path); + } + } + catch (IOException ex) + { + _logger.LogError(ex, "Error deleting partial stream file(s) {Path}", path); + + await DeletePartialStreamFiles(path, jobType, retryCount + 1, 500).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogError(ex, "Error deleting partial stream file(s) {Path}", path); + } + } + + /// + /// Deletes the progressive partial stream files. + /// + /// The output file path. + private void DeleteProgressivePartialStreamFiles(string outputFilePath) + { + if (File.Exists(outputFilePath)) + { + _fileSystem.DeleteFile(outputFilePath); + } + } + + /// + /// Deletes the HLS partial stream files. + /// + /// The output file path. + private void DeleteHlsPartialStreamFiles(string outputFilePath) + { + var directory = Path.GetDirectoryName(outputFilePath); + var name = Path.GetFileNameWithoutExtension(outputFilePath); + + var filesToDelete = _fileSystem.GetFilePaths(directory) + .Where(f => f.IndexOf(name, StringComparison.OrdinalIgnoreCase) != -1); + + List? exs = null; + foreach (var file in filesToDelete) + { + try + { + _logger.LogDebug("Deleting HLS file {0}", file); + _fileSystem.DeleteFile(file); + } + catch (IOException ex) + { + (exs ??= new List(4)).Add(ex); + _logger.LogError(ex, "Error deleting HLS file {Path}", file); + } + } + + if (exs != null) + { + throw new AggregateException("Error deleting HLS files", exs); + } + } + } +} diff --git a/Jellyfin.Api/Models/TranscodingDtos/TranscodingJob.cs b/Jellyfin.Api/Models/TranscodingDtos/TranscodingJob.cs new file mode 100644 index 0000000000..9b72107cfc --- /dev/null +++ b/Jellyfin.Api/Models/TranscodingDtos/TranscodingJob.cs @@ -0,0 +1,239 @@ +using System; +using System.Diagnostics; +using System.Threading; +using Jellyfin.Api.Models.PlaybackDtos; +using MediaBrowser.Controller.MediaEncoding; +using MediaBrowser.Model.Dto; +using Microsoft.Extensions.Logging; + +namespace Jellyfin.Api.Models.TranscodingDtos +{ + /// + /// The transcoding job. + /// + public class TranscodingJob + { + private readonly ILogger _logger; + private readonly object _timerLock = new object(); + + /// + /// Initializes a new instance of the class. + /// + /// Instance of the interface. + public TranscodingJob(ILogger logger) + { + _logger = logger; + } + + /// + /// Gets or sets the play session identifier. + /// + /// The play session identifier. + public string? PlaySessionId { get; set; } + + /// + /// Gets or sets the live stream identifier. + /// + /// The live stream identifier. + public string? LiveStreamId { get; set; } + + /// + /// Gets or sets a value indicating whether the transcoding job is a live output. + /// + public bool IsLiveOutput { get; set; } + + /// + /// Gets or sets the path. + /// + /// The path. + public MediaSourceInfo? MediaSource { get; set; } + + /// + /// Gets or sets the transcoding path. + /// + public string? Path { get; set; } + + /// + /// Gets or sets the type. + /// + /// The type. + public TranscodingJobType Type { get; set; } + + /// + /// Gets or sets the process. + /// + /// The process. + public Process? Process { get; set; } + + /// + /// Gets or sets the active request count. + /// + /// The active request count. + public int ActiveRequestCount { get; set; } + + /// + /// Gets or sets the kill timer. + /// + /// The kill timer. + private Timer? KillTimer { get; set; } + + /// + /// Gets or sets the device id. + /// + public string? DeviceId { get; set; } + + /// + /// Gets or sets the cancellation token source. + /// + public CancellationTokenSource? CancellationTokenSource { get; set; } + + /// + /// Gets or sets a value indicating whether the transcoding job has exited. + /// + public bool HasExited { get; set; } + + /// + /// Gets or sets a value indicating whether the user has paused the video. + /// + public bool IsUserPaused { get; set; } + + /// + /// Gets or sets the id. + /// + public string? Id { get; set; } + + /// + /// Gets or sets the framerate. + /// + public float? Framerate { get; set; } + + /// + /// Gets or sets the completion percentage. + /// + public double? CompletionPercentage { get; set; } + + /// + /// Gets or sets the bytes downloaded. + /// + public long? BytesDownloaded { get; set; } + + /// + /// Gets or sets the bytes transcoded. + /// + public long? BytesTranscoded { get; set; } + + /// + /// Gets or sets the bitrate. + /// + public int? BitRate { get; set; } + + /// + /// Gets or sets the transcoding position ticks. + /// + public long? TranscodingPositionTicks { get; set; } + + /// + /// Gets or sets the download position ticks. + /// + public long? DownloadPositionTicks { get; set; } + + /// + /// Gets or sets the transcodign throttler. + /// + public TranscodingThrottler? TranscodingThrottler { get; set; } + + /// + /// Gets or sets the last ping datetime. + /// + public DateTime LastPingDate { get; set; } + + /// + /// Gets or sets the ping timeout. + /// + public int PingTimeout { get; set; } + + /// + /// Stops the kill timer. + /// + public void StopKillTimer() + { + lock (_timerLock) + { + KillTimer?.Change(Timeout.Infinite, Timeout.Infinite); + } + } + + /// + /// Disposes the kill timer. + /// + public void DisposeKillTimer() + { + lock (_timerLock) + { + if (KillTimer != null) + { + KillTimer.Dispose(); + KillTimer = null; + } + } + } + + /// + /// Starts the kill timer. + /// + /// The amount of ms the timer should wait before the transcoding job gets killed. + public void StartKillTimer(Action callback) + { + StartKillTimer(callback, PingTimeout); + } + + /// + /// Starts the kill timer. + /// + /// The to run when the kill timer has finished. + /// The amount of ms the timer should wait before the transcoding job gets killed. + public void StartKillTimer(Action callback, int intervalMs) + { + if (HasExited) + { + return; + } + + lock (_timerLock) + { + if (KillTimer == null) + { + _logger.LogDebug($"Starting kill timer at {intervalMs}ms. JobId {Id} PlaySessionId {PlaySessionId}"); + KillTimer = new Timer(new TimerCallback(callback), this, intervalMs, Timeout.Infinite); + } + else + { + _logger.LogDebug($"Changing kill timer to {intervalMs}ms. JobId {Id} PlaySessionId {PlaySessionId}"); + KillTimer.Change(intervalMs, Timeout.Infinite); + } + } + } + + /// + /// Changes the kill timer if it has started. + /// + public void ChangeKillTimerIfStarted() + { + if (HasExited) + { + return; + } + + lock (_timerLock) + { + if (KillTimer != null) + { + var intervalMs = PingTimeout; + + _logger.LogDebug($"Changing kill timer to {intervalMs}ms. JobId {Id} PlaySessionId {PlaySessionId}"); + KillTimer.Change(intervalMs, Timeout.Infinite); + } + } + } + } +}