From 8f2fbf7a99c5f67daf2a1589fa12b1287a076bed Mon Sep 17 00:00:00 2001 From: Gary Wilber Date: Thu, 1 Oct 2020 16:24:35 -0700 Subject: [PATCH] Switch to TPL dataflow for subfolder scan --- .../ApplicationHost.cs | 1 - MediaBrowser.Controller/Entities/Folder.cs | 173 +++++++++++------- .../Library/TaskMethods.cs | 133 -------------- .../MediaBrowser.Controller.csproj | 3 +- 4 files changed, 106 insertions(+), 204 deletions(-) delete mode 100644 MediaBrowser.Controller/Library/TaskMethods.cs diff --git a/Emby.Server.Implementations/ApplicationHost.cs b/Emby.Server.Implementations/ApplicationHost.cs index 6d7239f724..7a46fdf2e7 100644 --- a/Emby.Server.Implementations/ApplicationHost.cs +++ b/Emby.Server.Implementations/ApplicationHost.cs @@ -758,7 +758,6 @@ namespace Emby.Server.Implementations BaseItem.FileSystem = _fileSystemManager; BaseItem.UserDataManager = Resolve(); BaseItem.ChannelManager = Resolve(); - TaskMethods.ConfigurationManager = ServerConfigurationManager; Video.LiveTvManager = Resolve(); Folder.UserViewManager = Resolve(); UserView.TVSeriesManager = Resolve(); diff --git a/MediaBrowser.Controller/Entities/Folder.cs b/MediaBrowser.Controller/Entities/Folder.cs index 666455cff4..8cea8755cb 100644 --- a/MediaBrowser.Controller/Entities/Folder.cs +++ b/MediaBrowser.Controller/Entities/Folder.cs @@ -8,6 +8,7 @@ using System.Linq; using System.Text.Json.Serialization; using System.Threading; using System.Threading.Tasks; +using System.Threading.Tasks.Dataflow; using Jellyfin.Data.Entities; using Jellyfin.Data.Enums; using MediaBrowser.Common.Progress; @@ -35,45 +36,16 @@ namespace MediaBrowser.Controller.Entities /// public class Folder : BaseItem { - /// - /// Contains constants used when reporting scan progress. - /// - private static class ProgressHelpers - { - /// - /// Reported after the folders immediate children are retrieved. - /// - public const int RetrievedChildren = 5; + private static Lazy _metadataRefreshThrottler = new Lazy(() => { + var concurrency = ConfigurationManager.Configuration.LibraryMetadataRefreshConcurrency; - /// - /// Reported after add, updating, or deleting child items from the LibraryManager. - /// - public const int UpdatedChildItems = 10; - - /// - /// Reported once subfolders are scanned. - /// When scanning subfolders, the progress will be between [UpdatedItems, ScannedSubfolders]. - /// - public const int ScannedSubfolders = 50; - - /// - /// Reported once metadata is refreshed. - /// When refreshing metadata, the progress will be between [ScannedSubfolders, MetadataRefreshed]. - /// - public const int RefreshedMetadata = 100; - - /// - /// Gets the current progress given the previous step, next step, and progress in between. - /// - /// The previous progress step. - /// The next progress step. - /// The current progress step. - /// The progress. - public static double GetProgress(int previousProgressStep, int nextProgressStep, double currentProgress) + if (concurrency <= 0) { - return previousProgressStep + ((nextProgressStep - previousProgressStep) * (currentProgress / 100)); + concurrency = Environment.ProcessorCount; } - } + + return new SemaphoreSlim(concurrency); + }); public static IUserViewManager UserViewManager { get; set; } @@ -508,19 +480,17 @@ namespace MediaBrowser.Controller.Entities private Task RefreshMetadataRecursive(IList children, MetadataRefreshOptions refreshOptions, bool recursive, IProgress progress, CancellationToken cancellationToken) { - var progressableTasks = children - .Select, Task>>(child => - innerProgress => RefreshChildMetadata(child, refreshOptions, recursive && child.IsFolder, innerProgress, cancellationToken)) - .ToList(); - - return RunTasks(progressableTasks, progress, cancellationToken); + return RunTasks( + (baseItem, innerProgress) => RefreshChildMetadata(baseItem, refreshOptions, recursive && baseItem.IsFolder, innerProgress, cancellationToken), + children, + progress, + cancellationToken); } private async Task RefreshAllMetadataForContainer(IMetadataContainer container, MetadataRefreshOptions refreshOptions, IProgress progress, CancellationToken cancellationToken) { // limit the amount of concurrent metadata refreshes - await TaskMethods.RunThrottled( - TaskMethods.SharedThrottleId.RefreshMetadata, + await RunMetadataRefresh( async () => { var series = container as Series; @@ -547,8 +517,7 @@ namespace MediaBrowser.Controller.Entities if (refreshOptions.RefreshItem(child)) { // limit the amount of concurrent metadata refreshes - await TaskMethods.RunThrottled( - TaskMethods.SharedThrottleId.RefreshMetadata, + await RunMetadataRefresh( async () => await child.RefreshMetadata(refreshOptions, cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false); } @@ -570,38 +539,33 @@ namespace MediaBrowser.Controller.Entities /// Task. private Task ValidateSubFolders(IList children, IDirectoryService directoryService, IProgress progress, CancellationToken cancellationToken) { - var progressableTasks = children - .Select, Task>>(child => - innerProgress => child.ValidateChildrenInternal(innerProgress, cancellationToken, true, false, null, directoryService)) - .ToList(); - - return RunTasks(progressableTasks, progress, cancellationToken); + return RunTasks( + (folder, innerProgress) => folder.ValidateChildrenInternal(innerProgress, cancellationToken, true, false, null, directoryService), + children, + progress, + cancellationToken); } /// - /// Runs a set of tasks concurrently with progress. + /// Runs an action block on a list of children. /// - /// A list of tasks. + /// The task to run for each child. + /// The list of children. /// The progress. /// The cancellation token. /// Task. - private async Task RunTasks(IList, Task>> tasks, IProgress progress, CancellationToken cancellationToken) + private async Task RunTasks(Func, Task> task, IList children, IProgress progress, CancellationToken cancellationToken) { - var childrenCount = tasks.Count; + var childrenCount = children.Count; var childrenProgress = new double[childrenCount]; - var actions = new Func[childrenCount]; void UpdateProgress() { progress.Report(childrenProgress.Average()); } - for (var i = 0; i < childrenCount; i++) - { - var childIndex = i; - var child = tasks[childIndex]; - - actions[childIndex] = async () => + var actionBlock = new ActionBlock( + async i => { var innerProgress = new ActionableProgress(); @@ -609,22 +573,33 @@ namespace MediaBrowser.Controller.Entities { // round the percent and only update progress if it changed to prevent excessive UpdateProgress calls var innerPercentRounded = Math.Round(innerPercent); - if (childrenProgress[childIndex] != innerPercentRounded) + if (childrenProgress[i] != innerPercentRounded) { - childrenProgress[childIndex] = innerPercentRounded; + childrenProgress[i] = innerPercentRounded; UpdateProgress(); } }); - await tasks[childIndex](innerProgress).ConfigureAwait(false); + await task(children[i], innerProgress).ConfigureAwait(false); - childrenProgress[childIndex] = 100; + childrenProgress[i] = 100; UpdateProgress(); - }; + }, + new ExecutionDataflowBlockOptions + { + MaxDegreeOfParallelism = ConfigurationManager.Configuration.LibraryScanFanoutConcurrency, + CancellationToken = cancellationToken, + }); + + for (var i = 0; i < childrenCount; i++) + { + actionBlock.Post(i); } - await TaskMethods.WhenAllThrottled(TaskMethods.SharedThrottleId.ScanFanout, actions, cancellationToken).ConfigureAwait(false); + actionBlock.Complete(); + + await actionBlock.Completion.ConfigureAwait(false); } /// @@ -1272,6 +1247,26 @@ namespace MediaBrowser.Controller.Entities return true; } + /// + /// Runs multiple metadata refreshes concurrently. + /// + /// The action to run. + /// The cancellation token. + /// A representing the result of the asynchronous operation. + private static async Task RunMetadataRefresh(Func action, CancellationToken cancellationToken) + { + await _metadataRefreshThrottler.Value.WaitAsync(cancellationToken).ConfigureAwait(false); + + try + { + await action().ConfigureAwait(false); + } + finally + { + _metadataRefreshThrottler.Value.Release(); + } + } + public List GetChildren(User user, bool includeLinkedChildren) { if (user == null) @@ -1819,5 +1814,45 @@ namespace MediaBrowser.Controller.Entities } } } + + /// + /// Contains constants used when reporting scan progress. + /// + private static class ProgressHelpers + { + /// + /// Reported after the folders immediate children are retrieved. + /// + public const int RetrievedChildren = 5; + + /// + /// Reported after add, updating, or deleting child items from the LibraryManager. + /// + public const int UpdatedChildItems = 10; + + /// + /// Reported once subfolders are scanned. + /// When scanning subfolders, the progress will be between [UpdatedItems, ScannedSubfolders]. + /// + public const int ScannedSubfolders = 50; + + /// + /// Reported once metadata is refreshed. + /// When refreshing metadata, the progress will be between [ScannedSubfolders, MetadataRefreshed]. + /// + public const int RefreshedMetadata = 100; + + /// + /// Gets the current progress given the previous step, next step, and progress in between. + /// + /// The previous progress step. + /// The next progress step. + /// The current progress step. + /// The progress. + public static double GetProgress(int previousProgressStep, int nextProgressStep, double currentProgress) + { + return previousProgressStep + ((nextProgressStep - previousProgressStep) * (currentProgress / 100)); + } + } } } diff --git a/MediaBrowser.Controller/Library/TaskMethods.cs b/MediaBrowser.Controller/Library/TaskMethods.cs deleted file mode 100644 index 66bfbe0d9e..0000000000 --- a/MediaBrowser.Controller/Library/TaskMethods.cs +++ /dev/null @@ -1,133 +0,0 @@ -using System; -using System.Collections.Concurrent; -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; -using MediaBrowser.Controller.Configuration; - -namespace MediaBrowser.Controller.Library -{ - /// - /// Helper methods for running tasks concurrently. - /// - public static class TaskMethods - { - private static readonly int _processorCount = Environment.ProcessorCount; - - private static readonly ConcurrentDictionary _sharedThrottlers = new ConcurrentDictionary(); - - /// - /// Throttle id for sharing a concurrency limit. - /// - public enum SharedThrottleId - { - /// - /// Library scan fan out - /// - ScanFanout, - - /// - /// Refresh metadata - /// - RefreshMetadata, - } - - /// - /// Gets or sets the configuration manager. - /// - public static IServerConfigurationManager ConfigurationManager { get; set; } - - /// - /// Similiar to Task.WhenAll but only allows running a certain amount of tasks at the same time. - /// - /// The throttle id. Multiple calls to this method with the same throttle id will share a concurrency limit. - /// List of actions to run. - /// The cancellation token. - /// A representing the result of the asynchronous operation. - public static async Task WhenAllThrottled(SharedThrottleId throttleId, IEnumerable> actions, CancellationToken cancellationToken) - { - var taskThrottler = throttleId == SharedThrottleId.ScanFanout ? - new SemaphoreSlim(GetConcurrencyLimit(throttleId)) : - _sharedThrottlers.GetOrAdd(throttleId, id => new SemaphoreSlim(GetConcurrencyLimit(id))); - - try - { - var tasks = new List(); - - foreach (var action in actions) - { - await taskThrottler.WaitAsync(cancellationToken).ConfigureAwait(false); - - tasks.Add(Task.Run(async () => - { - try - { - await action().ConfigureAwait(false); - } - finally - { - taskThrottler.Release(); - } - })); - } - - await Task.WhenAll(tasks).ConfigureAwait(false); - } - finally - { - if (throttleId == SharedThrottleId.ScanFanout) - { - taskThrottler.Dispose(); - } - } - } - - /// - /// Runs a task within a given throttler. - /// - /// The throttle id. Multiple calls to this method with the same throttle id will share a concurrency limit. - /// The action to run. - /// The cancellation token. - /// A representing the result of the asynchronous operation. - public static async Task RunThrottled(SharedThrottleId throttleId, Func action, CancellationToken cancellationToken) - { - if (throttleId == SharedThrottleId.ScanFanout) - { - // just await the task instead - throw new InvalidOperationException("Invalid throttle id"); - } - - var taskThrottler = _sharedThrottlers.GetOrAdd(throttleId, id => new SemaphoreSlim(GetConcurrencyLimit(id))); - - await taskThrottler.WaitAsync(cancellationToken).ConfigureAwait(false); - - try - { - await action().ConfigureAwait(false); - } - finally - { - taskThrottler.Release(); - } - } - - /// - /// Get the concurrency limit for the given throttle id. - /// - /// The throttle id. - /// The concurrency limit. - private static int GetConcurrencyLimit(SharedThrottleId throttleId) - { - var concurrency = throttleId == SharedThrottleId.RefreshMetadata ? - ConfigurationManager.Configuration.LibraryMetadataRefreshConcurrency : - ConfigurationManager.Configuration.LibraryScanFanoutConcurrency; - - if (concurrency <= 0) - { - concurrency = _processorCount; - } - - return concurrency; - } - } -} diff --git a/MediaBrowser.Controller/MediaBrowser.Controller.csproj b/MediaBrowser.Controller/MediaBrowser.Controller.csproj index 6544704065..243b8cd02b 100644 --- a/MediaBrowser.Controller/MediaBrowser.Controller.csproj +++ b/MediaBrowser.Controller/MediaBrowser.Controller.csproj @@ -16,7 +16,8 @@ - + +