diff options
| -rw-r--r-- | Emby.Server.Implementations/ApplicationHost.cs | 1 | ||||
| -rw-r--r-- | MediaBrowser.Controller/Entities/Folder.cs | 208 | ||||
| -rw-r--r-- | MediaBrowser.Controller/Library/TaskMethods.cs | 133 | ||||
| -rw-r--r-- | MediaBrowser.Model/Configuration/ServerConfiguration.cs | 12 |
4 files changed, 277 insertions, 77 deletions
diff --git a/Emby.Server.Implementations/ApplicationHost.cs b/Emby.Server.Implementations/ApplicationHost.cs index 7a46fdf2e..6d7239f72 100644 --- a/Emby.Server.Implementations/ApplicationHost.cs +++ b/Emby.Server.Implementations/ApplicationHost.cs @@ -758,6 +758,7 @@ namespace Emby.Server.Implementations BaseItem.FileSystem = _fileSystemManager; BaseItem.UserDataManager = Resolve<IUserDataManager>(); BaseItem.ChannelManager = Resolve<IChannelManager>(); + TaskMethods.ConfigurationManager = ServerConfigurationManager; Video.LiveTvManager = Resolve<ILiveTvManager>(); Folder.UserViewManager = Resolve<IUserViewManager>(); UserView.TVSeriesManager = Resolve<ITVSeriesManager>(); diff --git a/MediaBrowser.Controller/Entities/Folder.cs b/MediaBrowser.Controller/Entities/Folder.cs index 901ea875b..666455cff 100644 --- a/MediaBrowser.Controller/Entities/Folder.cs +++ b/MediaBrowser.Controller/Entities/Folder.cs @@ -35,6 +35,46 @@ namespace MediaBrowser.Controller.Entities /// </summary> public class Folder : BaseItem { + /// <summary> + /// Contains constants used when reporting scan progress. + /// </summary> + private static class ProgressHelpers + { + /// <summary> + /// Reported after the folders immediate children are retrieved. + /// </summary> + public const int RetrievedChildren = 5; + + /// <summary> + /// Reported after add, updating, or deleting child items from the LibraryManager. + /// </summary> + public const int UpdatedChildItems = 10; + + /// <summary> + /// Reported once subfolders are scanned. + /// When scanning subfolders, the progress will be between [UpdatedItems, ScannedSubfolders]. + /// </summary> + public const int ScannedSubfolders = 50; + + /// <summary> + /// Reported once metadata is refreshed. + /// When refreshing metadata, the progress will be between [ScannedSubfolders, MetadataRefreshed]. + /// </summary> + public const int RefreshedMetadata = 100; + + /// <summary> + /// Gets the current progress given the previous step, next step, and progress in between. + /// </summary> + /// <param name="previousProgressStep">The previous progress step.</param> + /// <param name="nextProgressStep">The next progress step.</param> + /// <param name="currentProgress">The current progress step.</param> + /// <returns>The progress.</returns> + public static double GetProgress(int previousProgressStep, int nextProgressStep, double currentProgress) + { + return previousProgressStep + ((nextProgressStep - previousProgressStep) * (currentProgress / 100)); + } + } + public static IUserViewManager UserViewManager { get; set; } /// <summary> @@ -327,11 +367,11 @@ namespace MediaBrowser.Controller.Entities return; } - progress.Report(5); + progress.Report(ProgressHelpers.RetrievedChildren); if (recursive) { - ProviderManager.OnRefreshProgress(this, 5); + ProviderManager.OnRefreshProgress(this, ProgressHelpers.RetrievedChildren); } // Build a dictionary of the current children we have now by Id so we can compare quickly and easily @@ -392,11 +432,11 @@ namespace MediaBrowser.Controller.Entities validChildrenNeedGeneration = true; } - progress.Report(10); + progress.Report(ProgressHelpers.UpdatedChildItems); if (recursive) { - ProviderManager.OnRefreshProgress(this, 10); + ProviderManager.OnRefreshProgress(this, ProgressHelpers.UpdatedChildItems); } cancellationToken.ThrowIfCancellationRequested(); @@ -406,11 +446,13 @@ namespace MediaBrowser.Controller.Entities var innerProgress = new ActionableProgress<double>(); var folder = this; - innerProgress.RegisterAction(p => + innerProgress.RegisterAction(innerPercent => { - double newPct = 0.80 * p + 10; - progress.Report(newPct); - ProviderManager.OnRefreshProgress(folder, newPct); + var percent = ProgressHelpers.GetProgress(ProgressHelpers.UpdatedChildItems, ProgressHelpers.ScannedSubfolders, innerPercent); + + progress.Report(percent); + + ProviderManager.OnRefreshProgress(folder, percent); }); if (validChildrenNeedGeneration) @@ -424,11 +466,11 @@ namespace MediaBrowser.Controller.Entities if (refreshChildMetadata) { - progress.Report(90); + progress.Report(ProgressHelpers.ScannedSubfolders); if (recursive) { - ProviderManager.OnRefreshProgress(this, 90); + ProviderManager.OnRefreshProgress(this, ProgressHelpers.ScannedSubfolders); } var container = this as IMetadataContainer; @@ -436,13 +478,15 @@ namespace MediaBrowser.Controller.Entities var innerProgress = new ActionableProgress<double>(); var folder = this; - innerProgress.RegisterAction(p => + innerProgress.RegisterAction(innerPercent => { - double newPct = 0.10 * p + 90; - progress.Report(newPct); + var percent = ProgressHelpers.GetProgress(ProgressHelpers.ScannedSubfolders, ProgressHelpers.RefreshedMetadata, innerPercent); + + progress.Report(percent); + if (recursive) { - ProviderManager.OnRefreshProgress(folder, newPct); + ProviderManager.OnRefreshProgress(folder, percent); } }); @@ -457,55 +501,37 @@ namespace MediaBrowser.Controller.Entities validChildren = Children.ToList(); } - await RefreshMetadataRecursive(validChildren, refreshOptions, recursive, innerProgress, cancellationToken); + await RefreshMetadataRecursive(validChildren, refreshOptions, recursive, innerProgress, cancellationToken).ConfigureAwait(false); } } } - private async Task RefreshMetadataRecursive(List<BaseItem> children, MetadataRefreshOptions refreshOptions, bool recursive, IProgress<double> progress, CancellationToken cancellationToken) + private Task RefreshMetadataRecursive(IList<BaseItem> children, MetadataRefreshOptions refreshOptions, bool recursive, IProgress<double> progress, CancellationToken cancellationToken) { - var numComplete = 0; - var count = children.Count; - double currentPercent = 0; - - foreach (var child in children) - { - cancellationToken.ThrowIfCancellationRequested(); - - var innerProgress = new ActionableProgress<double>(); - - // Avoid implicitly captured closure - var currentInnerPercent = currentPercent; - - innerProgress.RegisterAction(p => - { - double innerPercent = currentInnerPercent; - innerPercent += p / count; - progress.Report(innerPercent); - }); - - await RefreshChildMetadata(child, refreshOptions, recursive && child.IsFolder, innerProgress, cancellationToken) - .ConfigureAwait(false); - - numComplete++; - double percent = numComplete; - percent /= count; - percent *= 100; - currentPercent = percent; + var progressableTasks = children + .Select<BaseItem, Func<IProgress<double>, Task>>(child => + innerProgress => RefreshChildMetadata(child, refreshOptions, recursive && child.IsFolder, innerProgress, cancellationToken)) + .ToList(); - progress.Report(percent); - } + return RunTasks(progressableTasks, progress, cancellationToken); } private async Task RefreshAllMetadataForContainer(IMetadataContainer container, MetadataRefreshOptions refreshOptions, IProgress<double> progress, CancellationToken cancellationToken) { - var series = container as Series; - if (series != null) - { - await series.RefreshMetadata(refreshOptions, cancellationToken).ConfigureAwait(false); - } + // limit the amount of concurrent metadata refreshes + await TaskMethods.RunThrottled( + TaskMethods.SharedThrottleId.RefreshMetadata, + async () => + { + var series = container as Series; + if (series != null) + { + await series.RefreshMetadata(refreshOptions, cancellationToken).ConfigureAwait(false); + } - await container.RefreshAllMetadata(refreshOptions, progress, cancellationToken).ConfigureAwait(false); + await container.RefreshAllMetadata(refreshOptions, progress, cancellationToken).ConfigureAwait(false); + }, + cancellationToken).ConfigureAwait(false); } private async Task RefreshChildMetadata(BaseItem child, MetadataRefreshOptions refreshOptions, bool recursive, IProgress<double> progress, CancellationToken cancellationToken) @@ -520,12 +546,16 @@ namespace MediaBrowser.Controller.Entities { if (refreshOptions.RefreshItem(child)) { - await child.RefreshMetadata(refreshOptions, cancellationToken).ConfigureAwait(false); + // limit the amount of concurrent metadata refreshes + await TaskMethods.RunThrottled( + TaskMethods.SharedThrottleId.RefreshMetadata, + async () => await child.RefreshMetadata(refreshOptions, cancellationToken).ConfigureAwait(false), + cancellationToken).ConfigureAwait(false); } if (recursive && child is Folder folder) { - await folder.RefreshMetadataRecursive(folder.Children.ToList(), refreshOptions, true, progress, cancellationToken); + await folder.RefreshMetadataRecursive(folder.Children.ToList(), refreshOptions, true, progress, cancellationToken).ConfigureAwait(false); } } } @@ -538,39 +568,63 @@ namespace MediaBrowser.Controller.Entities /// <param name="progress">The progress.</param> /// <param name="cancellationToken">The cancellation token.</param> /// <returns>Task.</returns> - private async Task ValidateSubFolders(IList<Folder> children, IDirectoryService directoryService, IProgress<double> progress, CancellationToken cancellationToken) + private Task ValidateSubFolders(IList<Folder> children, IDirectoryService directoryService, IProgress<double> progress, CancellationToken cancellationToken) { - var numComplete = 0; - var count = children.Count; - double currentPercent = 0; + var progressableTasks = children + .Select<Folder, Func<IProgress<double>, Task>>(child => + innerProgress => child.ValidateChildrenInternal(innerProgress, cancellationToken, true, false, null, directoryService)) + .ToList(); - foreach (var child in children) - { - cancellationToken.ThrowIfCancellationRequested(); + return RunTasks(progressableTasks, progress, cancellationToken); + } - var innerProgress = new ActionableProgress<double>(); + /// <summary> + /// Runs a set of tasks concurrently with progress. + /// </summary> + /// <param name="tasks">A list of tasks.</param> + /// <param name="progress">The progress.</param> + /// <param name="cancellationToken">The cancellation token.</param> + /// <returns>Task.</returns> + private async Task RunTasks(IList<Func<IProgress<double>, Task>> tasks, IProgress<double> progress, CancellationToken cancellationToken) + { + var childrenCount = tasks.Count; + var childrenProgress = new double[childrenCount]; + var actions = new Func<Task>[childrenCount]; + + void UpdateProgress() + { + progress.Report(childrenProgress.Average()); + } - // Avoid implicitly captured closure - var currentInnerPercent = currentPercent; + for (var i = 0; i < childrenCount; i++) + { + var childIndex = i; + var child = tasks[childIndex]; - innerProgress.RegisterAction(p => + actions[childIndex] = async () => { - double innerPercent = currentInnerPercent; - innerPercent += p / count; - progress.Report(innerPercent); - }); + var innerProgress = new ActionableProgress<double>(); + + innerProgress.RegisterAction(innerPercent => + { + // round the percent and only update progress if it changed to prevent excessive UpdateProgress calls + var innerPercentRounded = Math.Round(innerPercent); + if (childrenProgress[childIndex] != innerPercentRounded) + { + childrenProgress[childIndex] = innerPercentRounded; + UpdateProgress(); + } + }); - await child.ValidateChildrenInternal(innerProgress, cancellationToken, true, false, null, directoryService) - .ConfigureAwait(false); + await tasks[childIndex](innerProgress).ConfigureAwait(false); - numComplete++; - double percent = numComplete; - percent /= count; - percent *= 100; - currentPercent = percent; + childrenProgress[childIndex] = 100; - progress.Report(percent); + UpdateProgress(); + }; } + + await TaskMethods.WhenAllThrottled(TaskMethods.SharedThrottleId.ScanFanout, actions, cancellationToken).ConfigureAwait(false); } /// <summary> diff --git a/MediaBrowser.Controller/Library/TaskMethods.cs b/MediaBrowser.Controller/Library/TaskMethods.cs new file mode 100644 index 000000000..66bfbe0d9 --- /dev/null +++ b/MediaBrowser.Controller/Library/TaskMethods.cs @@ -0,0 +1,133 @@ +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 +{ + /// <summary> + /// Helper methods for running tasks concurrently. + /// </summary> + public static class TaskMethods + { + private static readonly int _processorCount = Environment.ProcessorCount; + + private static readonly ConcurrentDictionary<SharedThrottleId, SemaphoreSlim> _sharedThrottlers = new ConcurrentDictionary<SharedThrottleId, SemaphoreSlim>(); + + /// <summary> + /// Throttle id for sharing a concurrency limit. + /// </summary> + public enum SharedThrottleId + { + /// <summary> + /// Library scan fan out + /// </summary> + ScanFanout, + + /// <summary> + /// Refresh metadata + /// </summary> + RefreshMetadata, + } + + /// <summary> + /// Gets or sets the configuration manager. + /// </summary> + public static IServerConfigurationManager ConfigurationManager { get; set; } + + /// <summary> + /// Similiar to Task.WhenAll but only allows running a certain amount of tasks at the same time. + /// </summary> + /// <param name="throttleId">The throttle id. Multiple calls to this method with the same throttle id will share a concurrency limit.</param> + /// <param name="actions">List of actions to run.</param> + /// <param name="cancellationToken">The cancellation token.</param> + /// <returns>A <see cref="Task"/> representing the result of the asynchronous operation.</returns> + public static async Task WhenAllThrottled(SharedThrottleId throttleId, IEnumerable<Func<Task>> 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<Task>(); + + 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(); + } + } + } + + /// <summary> + /// Runs a task within a given throttler. + /// </summary> + /// <param name="throttleId">The throttle id. Multiple calls to this method with the same throttle id will share a concurrency limit.</param> + /// <param name="action">The action to run.</param> + /// <param name="cancellationToken">The cancellation token.</param> + /// <returns>A <see cref="Task"/> representing the result of the asynchronous operation.</returns> + public static async Task RunThrottled(SharedThrottleId throttleId, Func<Task> 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(); + } + } + + /// <summary> + /// Get the concurrency limit for the given throttle id. + /// </summary> + /// <param name="throttleId">The throttle id.</param> + /// <returns>The concurrency limit.</returns> + 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.Model/Configuration/ServerConfiguration.cs b/MediaBrowser.Model/Configuration/ServerConfiguration.cs index 8b78ad842..14bfcbf9e 100644 --- a/MediaBrowser.Model/Configuration/ServerConfiguration.cs +++ b/MediaBrowser.Model/Configuration/ServerConfiguration.cs @@ -272,6 +272,16 @@ namespace MediaBrowser.Model.Configuration public string[] KnownProxies { get; set; } /// <summary> + /// Gets or sets the how the library scan fans out. + /// </summary> + public int LibraryScanFanoutConcurrency { get; set; } + + /// <summary> + /// Gets or sets the how many metadata refreshes can run concurrently. + /// </summary> + public int LibraryMetadataRefreshConcurrency { get; set; } + + /// <summary> /// Initializes a new instance of the <see cref="ServerConfiguration" /> class. /// </summary> public ServerConfiguration() @@ -381,6 +391,8 @@ namespace MediaBrowser.Model.Configuration SlowResponseThresholdMs = 500; CorsHosts = new[] { "*" }; KnownProxies = Array.Empty<string>(); + LibraryMetadataRefreshConcurrency = 0; + LibraryScanFanoutConcurrency = 0; } } |