michael.heier/jellyfin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rename and clean up TranscodingJob

Browse Source
This commit is contained in:
Patrick Barron 2023-10-31 11:31:09 -04:00
parent abd74fd5a4
commit c2081955c8
7 changed files with 97 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
Jellyfin.Api/Controllers/DynamicHlsController.cs
View File

@ -287,7 +287,7 @@ public class DynamicHlsController : BaseJellyfinApiController
cancellationToken)
.ConfigureAwait(false);
TranscodingJobDto? job = null;
TranscodingJob? job = null;
var playlistPath = Path.ChangeExtension(state.OutputFilePath, ".m3u8");
if (!System.IO.File.Exists(playlistPath))
@ -1431,7 +1431,7 @@ public class DynamicHlsController : BaseJellyfinApiController
var segmentExtension = EncodingHelper.GetSegmentFileExtension(state.Request.SegmentContainer);
TranscodingJobDto? job;
TranscodingJob? job;
if (System.IO.File.Exists(segmentPath))
{
@ -1921,7 +1921,7 @@ public class DynamicHlsController : BaseJellyfinApiController
string segmentPath,
string segmentExtension,
int segmentIndex,
TranscodingJobDto? transcodingJob,
TranscodingJob? transcodingJob,
CancellationToken cancellationToken)
{
var segmentExists = System.IO.File.Exists(segmentPath);
@ -1990,7 +1990,7 @@ public class DynamicHlsController : BaseJellyfinApiController
return GetSegmentResult(state, segmentPath, transcodingJob);
}
private ActionResult GetSegmentResult(StreamState state, string segmentPath, TranscodingJobDto? transcodingJob)
private ActionResult GetSegmentResult(StreamState state, string segmentPath, TranscodingJob? transcodingJob)
{
var segmentEndingPositionTicks = state.Request.CurrentRuntimeTicks + state.Request.ActualSegmentLengthTicks;

2
Jellyfin.Api/Helpers/FileStreamResponseHelpers.cs
View File

@ -96,7 +96,7 @@ public static class FileStreamResponseHelpers
await transcodingLock.WaitAsync(cancellationTokenSource.Token).ConfigureAwait(false);
try
{
TranscodingJobDto? job;
TranscodingJob? job;
if (!File.Exists(outputPath))
{
job = await transcodingJobHelper.StartFfMpeg(state, outputPath, ffmpegCommandLineArguments, httpContext.Request, transcodingJobType, cancellationTokenSource).ConfigureAwait(false);

4
Jellyfin.Api/Helpers/ProgressiveFileStream.cs
View File

@ -14,7 +14,7 @@ namespace Jellyfin.Api.Helpers;
public class ProgressiveFileStream : Stream
{
private readonly Stream _stream;
private readonly TranscodingJobDto? _job;
private readonly TranscodingJob? _job;
private readonly TranscodingJobHelper? _transcodingJobHelper;
private readonly int _timeoutMs;
private bool _disposed;
@ -26,7 +26,7 @@ public class ProgressiveFileStream : Stream
/// <param name="job">The transcoding job information.</param>
/// <param name="transcodingJobHelper">The transcoding job helper.</param>
/// <param name="timeoutMs">The timeout duration in milliseconds.</param>
public ProgressiveFileStream(string filePath, TranscodingJobDto? job, TranscodingJobHelper transcodingJobHelper, int timeoutMs = 30000)
public ProgressiveFileStream(string filePath, TranscodingJob? job, TranscodingJobHelper transcodingJobHelper, int timeoutMs = 30000)
{
_job = job;
_transcodingJobHelper = transcodingJobHelper;

71
Jellyfin.Api/Helpers/TranscodingJobHelper.cs
View File

@ -36,7 +36,7 @@ public class TranscodingJobHelper : IDisposable
/// <summary>
/// The active transcoding jobs.
/// </summary>
private static readonly List<TranscodingJobDto> _activeTranscodingJobs = new List<TranscodingJobDto>();
private static readonly List<TranscodingJob> _activeTranscodingJobs = new List<TranscodingJob>();
/// <summary>
/// The transcoding locks.
@ -105,7 +105,7 @@ public class TranscodingJobHelper : IDisposable
/// </summary>
/// <param name="playSessionId">Playback session id.</param>
/// <returns>The transcoding job.</returns>
public TranscodingJobDto? GetTranscodingJob(string playSessionId)
public TranscodingJob? GetTranscodingJob(string playSessionId)
{
lock (_activeTranscodingJobs)
{
@ -119,7 +119,7 @@ public class TranscodingJobHelper : IDisposable
/// <param name="path">Path to the transcoding file.</param>
/// <param name="type">The <see cref="TranscodingJobType"/>.</param>
/// <returns>The transcoding job.</returns>
public TranscodingJobDto? GetTranscodingJob(string path, TranscodingJobType type)
public TranscodingJob? GetTranscodingJob(string path, TranscodingJobType type)
{
lock (_activeTranscodingJobs)
{
@ -139,7 +139,7 @@ public class TranscodingJobHelper : IDisposable
_logger.LogDebug("PingTranscodingJob PlaySessionId={0} isUsedPaused: {1}", playSessionId, isUserPaused);
List<TranscodingJobDto> jobs;
List<TranscodingJob> jobs;
lock (_activeTranscodingJobs)
{
@ -160,7 +160,7 @@ public class TranscodingJobHelper : IDisposable
}
}
private void PingTimer(TranscodingJobDto job, bool isProgressCheckIn)
private void PingTimer(TranscodingJob job, bool isProgressCheckIn)
{
if (job.HasExited)
{
@ -195,7 +195,7 @@ public class TranscodingJobHelper : IDisposable
/// <param name="state">The state.</param>
private async void OnTranscodeKillTimerStopped(object? state)
{
var job = state as TranscodingJobDto ?? throw new ArgumentException($"{nameof(state)} is not of type {nameof(TranscodingJobDto)}", nameof(state));
var job = state as TranscodingJob ?? throw new ArgumentException($"{nameof(state)} is not of type {nameof(TranscodingJob)}", nameof(state));
if (!job.HasExited && job.Type != TranscodingJobType.Progressive)
{
var timeSinceLastPing = (DateTime.UtcNow - job.LastPingDate).TotalMilliseconds;
@ -234,9 +234,9 @@ public class TranscodingJobHelper : IDisposable
/// <param name="killJob">The kill job.</param>
/// <param name="deleteFiles">The delete files.</param>
/// <returns>Task.</returns>
private Task KillTranscodingJobs(Func<TranscodingJobDto, bool> killJob, Func<string, bool> deleteFiles)
private Task KillTranscodingJobs(Func<TranscodingJob, bool> killJob, Func<string, bool> deleteFiles)
{
var jobs = new List<TranscodingJobDto>();
var jobs = new List<TranscodingJob>();
lock (_activeTranscodingJobs)
{
@ -267,7 +267,7 @@ public class TranscodingJobHelper : IDisposable
/// <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(TranscodingJobDto job, bool closeLiveStream, Func<string, bool> delete)
private async Task KillTranscodingJob(TranscodingJob job, bool closeLiveStream, Func<string, bool> delete)
{
job.DisposeKillTimer();
@ -281,6 +281,7 @@ public class TranscodingJobHelper : IDisposable
{
#pragma warning disable CA1849 // Can't await in lock block
job.CancellationTokenSource.Cancel();
#pragma warning restore CA1849
}
}
@ -289,35 +290,7 @@ public class TranscodingJobHelper : IDisposable
_transcodingLocks.Remove(job.Path!);
}
lock (job.ProcessLock!)
{
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 because killing is asynchronous.
if (!process.WaitForExit(5000))
{
_logger.LogInformation("Killing FFmpeg process for {Path}", job.Path);
process.Kill();
}
}
catch (InvalidOperationException)
{
}
}
#pragma warning restore CA1849
}
job.Stop();
if (delete(job.Path!))
{
@ -430,7 +403,7 @@ public class TranscodingJobHelper : IDisposable
/// <summary>
/// Report the transcoding progress to the session manager.
/// </summary>
/// <param name="job">The <see cref="TranscodingJobDto"/> of which the progress will be reported.</param>
/// <param name="job">The <see cref="TranscodingJob"/> of which the progress will be reported.</param>
/// <param name="state">The <see cref="StreamState"/> of the current transcoding job.</param>
/// <param name="transcodingPosition">The current transcoding position.</param>
/// <param name="framerate">The framerate of the transcoding job.</param>
@ -438,7 +411,7 @@ public class TranscodingJobHelper : IDisposable
/// <param name="bytesTranscoded">The number of bytes transcoded.</param>
/// <param name="bitRate">The bitrate of the transcoding job.</param>
public void ReportTranscodingProgress(
TranscodingJobDto job,
TranscodingJob job,
StreamState state,
TimeSpan? transcodingPosition,
float? framerate,
@ -500,7 +473,7 @@ public class TranscodingJobHelper : IDisposable
/// <param name="cancellationTokenSource">The cancellation token source.</param>
/// <param name="workingDirectory">The working directory.</param>
/// <returns>Task.</returns>
public async Task<TranscodingJobDto> StartFfMpeg(
public async Task<TranscodingJob> StartFfMpeg(
StreamState state,
string outputPath,
string commandLineArguments,
@ -655,7 +628,7 @@ public class TranscodingJobHelper : IDisposable
return transcodingJob;
}
private void StartThrottler(StreamState state, TranscodingJobDto transcodingJob)
private void StartThrottler(StreamState state, TranscodingJob transcodingJob)
{
if (EnableThrottling(state))
{
@ -688,7 +661,7 @@ public class TranscodingJobHelper : IDisposable
/// <param name="state">The state.</param>
/// <param name="cancellationTokenSource">The cancellation token source.</param>
/// <returns>TranscodingJob.</returns>
public TranscodingJobDto OnTranscodeBeginning(
public TranscodingJob OnTranscodeBeginning(
string path,
string? playSessionId,
string? liveStreamId,
@ -701,7 +674,7 @@ public class TranscodingJobHelper : IDisposable
{
lock (_activeTranscodingJobs)
{
var job = new TranscodingJobDto(_loggerFactory.CreateLogger<TranscodingJobDto>())
var job = new TranscodingJob(_loggerFactory.CreateLogger<TranscodingJob>())
{
Type = type,
Path = path,
@ -727,7 +700,7 @@ public class TranscodingJobHelper : IDisposable
/// Called when [transcode end].
/// </summary>
/// <param name="job">The transcode job.</param>
public void OnTranscodeEndRequest(TranscodingJobDto job)
public void OnTranscodeEndRequest(TranscodingJob job)
{
job.ActiveRequestCount--;
_logger.LogDebug("OnTranscodeEndRequest job.ActiveRequestCount={ActiveRequestCount}", job.ActiveRequestCount);
@ -775,7 +748,7 @@ public class TranscodingJobHelper : IDisposable
/// <param name="process">The process.</param>
/// <param name="job">The job.</param>
/// <param name="state">The state.</param>
private void OnFfMpegProcessExited(Process process, TranscodingJobDto job, StreamState state)
private void OnFfMpegProcessExited(Process process, TranscodingJob job, StreamState state)
{
job.HasExited = true;
job.ExitCode = process.ExitCode;
@ -826,8 +799,8 @@ public class TranscodingJobHelper : IDisposable
/// </summary>
/// <param name="path">The path.</param>
/// <param name="type">The type.</param>
/// <returns>The <see cref="TranscodingJobDto"/>.</returns>
public TranscodingJobDto? OnTranscodeBeginRequest(string path, TranscodingJobType type)
/// <returns>The <see cref="TranscodingJob"/>.</returns>
public TranscodingJob? OnTranscodeBeginRequest(string path, TranscodingJobType type)
{
lock (_activeTranscodingJobs)
{
@ -844,7 +817,7 @@ public class TranscodingJobHelper : IDisposable
}
}
private void OnTranscodeBeginRequest(TranscodingJobDto job)
private void OnTranscodeBeginRequest(TranscodingJob job)
{
job.ActiveRequestCount++;

2
Jellyfin.Api/Models/StreamingDtos/StreamState.cs
View File

@ -140,7 +140,7 @@ public class StreamState : EncodingJobInfo, IDisposable
/// <summary>
/// Gets or sets the transcoding job.
/// </summary>
public TranscodingJobDto? TranscodingJob { get; set; }
public TranscodingJob? TranscodingJob { get; set; }
/// <inheritdoc />
public void Dispose()

130
MediaBrowser.Controller/MediaEncoding/TranscodingJobDto.cs → MediaBrowser.Controller/MediaEncoding/TranscodingJob.cs
View File

@ -1,6 +1,5 @@
using System;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Threading;
using MediaBrowser.Model.Dto;
using Microsoft.Extensions.Logging;
@ -10,39 +9,31 @@ namespace MediaBrowser.Controller.MediaEncoding;
/// <summary>
/// Class TranscodingJob.
/// </summary>
public class TranscodingJobDto : IDisposable
public sealed class TranscodingJob : 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();
private readonly ILogger<TranscodingJob> _logger;
private readonly object _processLock = new();
private readonly object _timerLock = new();
private Timer? _killTimer;
/// <summary>
/// Timer lock.
/// </summary>
private readonly object _timerLock = new object();
/// <summary>
/// Initializes a new instance of the <see cref="TranscodingJobDto"/> class.
/// Initializes a new instance of the <see cref="TranscodingJob"/> class.
/// </summary>
/// <param name="logger">Instance of the <see cref="ILogger{TranscodingJobDto}"/> interface.</param>
public TranscodingJobDto(ILogger<TranscodingJobDto> logger)
public TranscodingJob(ILogger<TranscodingJob> logger)
{
Logger = 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>
@ -53,7 +44,6 @@ public class TranscodingJobDto : IDisposable
/// <summary>
/// Gets or sets the path.
/// </summary>
/// <value>The path.</value>
public MediaSourceInfo? MediaSource { get; set; }
/// <summary>
@ -64,32 +54,18 @@ public class TranscodingJobDto : IDisposable
/// <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>
@ -177,7 +153,7 @@ public class TranscodingJobDto : IDisposable
{
lock (_timerLock)
{
KillTimer?.Change(Timeout.Infinite, Timeout.Infinite);
_killTimer?.Change(Timeout.Infinite, Timeout.Infinite);
}
}
@ -188,10 +164,10 @@ public class TranscodingJobDto : IDisposable
{
lock (_timerLock)
{
if (KillTimer is not null)
if (_killTimer is not null)
{
KillTimer.Dispose();
KillTimer = null;
_killTimer.Dispose();
_killTimer = null;
}
}
}
@ -219,15 +195,15 @@ public class TranscodingJobDto : IDisposable
lock (_timerLock)
{
if (KillTimer is null)
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);
_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);
_logger.LogDebug("Changing kill timer to {0}ms. JobId {1} PlaySessionId {2}", intervalMs, Id, PlaySessionId);
_killTimer.Change(intervalMs, Timeout.Infinite);
}
}
}
@ -244,39 +220,61 @@ public class TranscodingJobDto : IDisposable
lock (_timerLock)
{
if (KillTimer is not null)
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);
_logger.LogDebug("Changing kill timer to {0}ms. JobId {1} PlaySessionId {2}", intervalMs, Id, PlaySessionId);
_killTimer.Change(intervalMs, Timeout.Infinite);
}
}
}
/// <summary>
/// Stops the transcoding job.
/// </summary>
public void Stop()
{
lock (_processLock)
{
#pragma warning disable CA1849 // Can't await in lock block
TranscodingThrottler?.Stop().GetAwaiter().GetResult();
var process = Process;
if (!HasExited)
{
try
{
_logger.LogInformation("Stopping ffmpeg process with q command for {Path}", Path);
process!.StandardInput.WriteLine("q");
// Need to wait because killing is asynchronous.
if (!process.WaitForExit(5000))
{
_logger.LogInformation("Killing FFmpeg process for {Path}", Path);
process.Kill();
}
}
catch (InvalidOperationException)
{
}
}
#pragma warning restore CA1849
}
}
/// <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;
}
Process?.Dispose();
Process = null;
_killTimer?.Dispose();
_killTimer = null;
CancellationTokenSource?.Dispose();
CancellationTokenSource = null;
TranscodingThrottler?.Dispose();
TranscodingThrottler = null;
}
}

6
MediaBrowser.Controller/MediaEncoding/TranscodingThrottler.cs
View File

@ -13,7 +13,7 @@ namespace MediaBrowser.Controller.MediaEncoding;
/// </summary>
public class TranscodingThrottler : IDisposable
{
private readonly TranscodingJobDto _job;
private readonly TranscodingJob _job;
private readonly ILogger<TranscodingThrottler> _logger;
private readonly IConfigurationManager _config;
private readonly IFileSystem _fileSystem;
@ -29,7 +29,7 @@ public class TranscodingThrottler : IDisposable
/// <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)
public TranscodingThrottler(TranscodingJob job, ILogger<TranscodingThrottler> logger, IConfigurationManager config, IFileSystem fileSystem, IMediaEncoder mediaEncoder)
{
_job = job;
_logger = logger;
@ -145,7 +145,7 @@ public class TranscodingThrottler : IDisposable
}
}
private bool IsThrottleAllowed(TranscodingJobDto job, int thresholdSeconds)
private bool IsThrottleAllowed(TranscodingJob job, int thresholdSeconds)
{
var bytesDownloaded = job.BytesDownloaded;
var transcodingPositionTicks = job.TranscodingPositionTicks ?? 0;