David
5 years ago
parent
2cb6eb984b
commit
9be20cfaa6
@ -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
|
||||
{
|
||||
/// <summary>
|
||||
/// The jellyfin api entry point.
|
||||
/// </summary>
|
||||
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<TranscodingJob> _activeTranscodingJobs;
|
||||
private readonly Dictionary<string, SemaphoreSlim> _transcodingLocks;
|
||||
private bool _disposed = false;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="JellyfinApiEntryPoint" /> class.
|
||||
/// </summary>
|
||||
/// <param name="logger">The logger.</param>
|
||||
/// <param name="sessionManager">The session manager.</param>
|
||||
/// <param name="config">The configuration.</param>
|
||||
/// <param name="fileSystem">The file system.</param>
|
||||
/// <param name="mediaSourceManager">The media source manager.</param>
|
||||
public JellyfinApiEntryPoint(
|
||||
ILogger<JellyfinApiEntryPoint> logger,
|
||||
ISessionManager sessionManager,
|
||||
IServerConfigurationManager config,
|
||||
IFileSystem fileSystem,
|
||||
IMediaSourceManager mediaSourceManager)
|
||||
{
|
||||
_logger = logger;
|
||||
_sessionManager = sessionManager;
|
||||
_serverConfigurationManager = config;
|
||||
_fileSystem = fileSystem;
|
||||
_mediaSourceManager = mediaSourceManager;
|
||||
|
||||
_activeTranscodingJobs = new List<TranscodingJob>();
|
||||
_transcodingLocks = new Dictionary<string, SemaphoreSlim>();
|
||||
|
||||
_sessionManager!.PlaybackProgress += OnPlaybackProgress;
|
||||
_sessionManager!.PlaybackStart += OnPlaybackProgress;
|
||||
|
||||
Instance = this;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the initialized instance of <see cref="JellyfinApiEntryPoint"/>.
|
||||
/// </summary>
|
||||
public static JellyfinApiEntryPoint? Instance { get; private set; }
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose()
|
||||
{
|
||||
Dispose(true);
|
||||
GC.SuppressFinalize(this);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Releases unmanaged and - optionally - managed resources.
|
||||
/// </summary>
|
||||
/// <param name="dispose"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release only unmanaged resources.</param>
|
||||
protected virtual void Dispose(bool dispose)
|
||||
{
|
||||
if (_disposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (dispose)
|
||||
{
|
||||
// TODO: dispose
|
||||
}
|
||||
|
||||
List<TranscodingJob> jobs;
|
||||
lock (_activeTranscodingJobs)
|
||||
{
|
||||
jobs = _activeTranscodingJobs.ToList();
|
||||
}
|
||||
|
||||
var jobCount = jobs.Count;
|
||||
|
||||
IEnumerable<Task> 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;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Deletes the encoded media cache.
|
||||
/// </summary>
|
||||
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<TranscodingJob> 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();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Called when [transcode kill timer stopped].
|
||||
/// </summary>
|
||||
/// <param name="state">The state.</param>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Kills the transcoding job.
|
||||
/// </summary>
|
||||
/// <param name="job">The job.</param>
|
||||
/// <param name="closeLiveStream">if set to <c>true</c> [close live stream].</param>
|
||||
/// <param name="delete">The delete.</param>
|
||||
private async Task KillTranscodingJob(TranscodingJob job, bool closeLiveStream, Func<string, bool> 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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Deletes the progressive partial stream files.
|
||||
/// </summary>
|
||||
/// <param name="outputFilePath">The output file path.</param>
|
||||
private void DeleteProgressivePartialStreamFiles(string outputFilePath)
|
||||
{
|
||||
if (File.Exists(outputFilePath))
|
||||
{
|
||||
_fileSystem.DeleteFile(outputFilePath);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Deletes the HLS partial stream files.
|
||||
/// </summary>
|
||||
/// <param name="outputFilePath">The output file path.</param>
|
||||
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<Exception>? exs = null;
|
||||
foreach (var file in filesToDelete)
|
||||
{
|
||||
try
|
||||
{
|
||||
_logger.LogDebug("Deleting HLS file {0}", file);
|
||||
_fileSystem.DeleteFile(file);
|
||||
}
|
||||
catch (IOException ex)
|
||||
{
|
||||
(exs ??= new List<Exception>(4)).Add(ex);
|
||||
_logger.LogError(ex, "Error deleting HLS file {Path}", file);
|
||||
}
|
||||
}
|
||||
|
||||
if (exs != null)
|
||||
{
|
||||
throw new AggregateException("Error deleting HLS files", exs);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -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
|
||||
{
|
||||
/// <summary>
|
||||
/// The transcoding job.
|
||||
/// </summary>
|
||||
public class TranscodingJob
|
||||
{
|
||||
private readonly ILogger _logger;
|
||||
private readonly object _timerLock = new object();
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="TranscodingJob"/> class.
|
||||
/// </summary>
|
||||
/// <param name="logger">Instance of the <see cref="ILogger{TranscodingJob}"/> interface.</param>
|
||||
public TranscodingJob(ILogger<TranscodingJob> 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 the transcoding job is a 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 the transcoding 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 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 the device id.
|
||||
/// </summary>
|
||||
public string? DeviceId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the cancellation token source.
|
||||
/// </summary>
|
||||
public CancellationTokenSource? CancellationTokenSource { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets a value indicating whether the transcoding job has exited.
|
||||
/// </summary>
|
||||
public bool HasExited { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets a value indicating whether the user has paused the video.
|
||||
/// </summary>
|
||||
public bool IsUserPaused { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the id.
|
||||
/// </summary>
|
||||
public string? Id { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the framerate.
|
||||
/// </summary>
|
||||
public float? Framerate { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the completion percentage.
|
||||
/// </summary>
|
||||
public double? CompletionPercentage { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the bytes downloaded.
|
||||
/// </summary>
|
||||
public long? BytesDownloaded { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the bytes transcoded.
|
||||
/// </summary>
|
||||
public long? BytesTranscoded { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the bitrate.
|
||||
/// </summary>
|
||||
public int? BitRate { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the transcoding position ticks.
|
||||
/// </summary>
|
||||
public long? TranscodingPositionTicks { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the download position ticks.
|
||||
/// </summary>
|
||||
public long? DownloadPositionTicks { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the transcodign throttler.
|
||||
/// </summary>
|
||||
public TranscodingThrottler? TranscodingThrottler { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the last ping datetime.
|
||||
/// </summary>
|
||||
public DateTime LastPingDate { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the ping timeout.
|
||||
/// </summary>
|
||||
public int PingTimeout { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Stops the kill timer.
|
||||
/// </summary>
|
||||
public void StopKillTimer()
|
||||
{
|
||||
lock (_timerLock)
|
||||
{
|
||||
KillTimer?.Change(Timeout.Infinite, Timeout.Infinite);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Disposes the kill timer.
|
||||
/// </summary>
|
||||
public void DisposeKillTimer()
|
||||
{
|
||||
lock (_timerLock)
|
||||
{
|
||||
if (KillTimer != null)
|
||||
{
|
||||
KillTimer.Dispose();
|
||||
KillTimer = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts the kill timer.
|
||||
/// </summary>
|
||||
/// <param name="callback">The amount of ms the timer should wait before the transcoding job gets killed.</param>
|
||||
public void StartKillTimer(Action<object> callback)
|
||||
{
|
||||
StartKillTimer(callback, PingTimeout);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts the kill timer.
|
||||
/// </summary>
|
||||
/// <param name="callback">The <see cref="Action"/> to run when the kill timer has finished.</param>
|
||||
/// <param name="intervalMs">The amount of ms the timer should wait before the transcoding job gets killed.</param>
|
||||
public void StartKillTimer(Action<object> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Changes the kill timer if it has started.
|
||||
/// </summary>
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue