Use proper scheduler that honors the parallel task limit #14281

JPVenson · 2025-06-11T13:13:30Z

Fixes several issues with system overloads when there are multiple folders present in a library as each would spawn its own queue limited to the task limit.

The new scheduler uses a global limit and cleansup tasks afterwards.

crobibero · 2025-06-11T13:58:04Z

I haven't tested this but it looks like this has a cts leak (+ a few other tweaks)

Index: MediaBrowser.Controller/LibraryTaskScheduler/LimitedConcurrencyLibraryScheduler.cs
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/MediaBrowser.Controller/LibraryTaskScheduler/LimitedConcurrencyLibraryScheduler.cs b/MediaBrowser.Controller/LibraryTaskScheduler/LimitedConcurrencyLibraryScheduler.cs
--- a/MediaBrowser.Controller/LibraryTaskScheduler/LimitedConcurrencyLibraryScheduler.cs	(revision a2fb1f63d9d30e09de4244e2b6085bb6ec0e75b6)
+++ b/MediaBrowser.Controller/LibraryTaskScheduler/LimitedConcurrencyLibraryScheduler.cs	(date 1749650016869)
@@ -13,19 +13,16 @@
 /// <summary>
 /// Provides Parallel action interface to process tasks with a set concurrency level.
 /// </summary>
-public class LimitedConcurrencyLibraryScheduler : ILimitedConcurrencyLibraryScheduler
+public sealed class LimitedConcurrencyLibraryScheduler : ILimitedConcurrencyLibraryScheduler, IDisposable
 {
     private readonly IHostApplicationLifetime _hostApplicationLifetime;
     private readonly ILogger<LimitedConcurrencyLibraryScheduler> _logger;
     private readonly Dictionary<CancellationTokenSource, Task> _taskRunners = new();
-
-    private static AsyncLocal<CancellationTokenSource> _deadlockDetector = new();
-
-    /// <summary>
-    /// Gets used to lock all operations on the Tasks queue and creating workers.
-    /// </summary>
+    private readonly BlockingCollection<TaskQueueItem> _tasks = new();
     private readonly Lock _taskLock = new();
 
+    private static readonly AsyncLocal<CancellationTokenSource> _deadlockDetector = new();
+
     private Task? _cleanupTask;
 
     /// <summary>
@@ -39,8 +36,6 @@
         _logger = logger;
     }
 
-    private BlockingCollection<TaskQueueItem> Tasks { get; set; } = new();
-
     private void ScheduleTaskCleanup()
     {
         lock (_taskLock)
@@ -57,13 +52,13 @@
         void RunCleanupTask()
         {
             _cleanupTask = Task.Delay(TimeSpan.FromSeconds(10)).ContinueWith(
-                t =>
+                _ =>
                 {
                     lock (_taskLock)
                     {
-                        if (Tasks.Count > 0)
+                        if (_tasks.Count > 0)
                         {
-                            // tasks are still there so its still in use. Reschedule cleanup task.
+                            // tasks are still there so it is still in use. Reschedule cleanup task.
                             // we cannot just exit here and rely on the other invoker because there is a considerable timeframe where it could have already ended.
                             RunCleanupTask();
                             return;
@@ -88,8 +83,7 @@
             var parallelism = (fanoutConcurrency > 0 ? fanoutConcurrency : Environment.ProcessorCount) - _taskRunners.Count;
             for (int i = 0; i < parallelism; i++)
             {
-                var stopToken = new CancellationTokenSource();
-                var combinedSource = CancellationTokenSource.CreateLinkedTokenSource(stopToken.Token, _hostApplicationLifetime.ApplicationStopping);
+                var combinedSource = CancellationTokenSource.CreateLinkedTokenSource(new CancellationTokenSource().Token, _hostApplicationLifetime.ApplicationStopping);
                 _taskRunners.Add(
                     combinedSource,
                     Task.Factory.StartNew(
@@ -108,19 +102,20 @@
         _deadlockDetector.Value = stopToken;
         try
         {
-            foreach (var item in Tasks.GetConsumingEnumerable(stopToken.Token))
+            foreach (var item in _tasks.GetConsumingEnumerable(stopToken.Token))
             {
                 await ProcessItem(item).ConfigureAwait(false);
             }
         }
         catch (OperationCanceledException) when (stopToken.IsCancellationRequested)
         {
-            // thats how you do it, interupt the waiter thread. There is nothing to do here when it was on purpose.
+            // Task was cancelled, nothing to do.
         }
         finally
         {
-            _deadlockDetector.Value = default!;
+            _deadlockDetector.Value = null!;
             _taskRunners.Remove(stopToken);
+            stopToken.Dispose();
         }
     }
 
@@ -130,15 +125,15 @@
         {
             if (item.CancellationToken.IsCancellationRequested)
             {
-                // if item is cancled, just skip it
+                // if item is cancelled, just skip it
                 return;
             }
 
             await item.Worker(item.Data).ConfigureAwait(true);
         }
-        catch (System.Exception ex)
+        catch (Exception ex)
         {
-            _logger.LogError(ex, "Error while performing a library operation.");
+            _logger.LogError(ex, "Error while performing a library operation");
         }
         finally
         {
@@ -167,7 +162,7 @@
                     {
                         // round the percent and only update progress if it changed to prevent excessive UpdateProgress calls
                         var innerPercentRounded = Math.Round(innerPercent);
-                        if (queueItem!.ProgressValue != innerPercentRounded)
+                        if (queueItem.ProgressValue != innerPercentRounded)
                         {
                             queueItem.ProgressValue = innerPercentRounded;
                             UpdateProgress();
@@ -181,13 +176,13 @@
         for (var i = 0; i < workItems.Length; i++)
         {
             var item = workItems[i]!;
-            Tasks.Add(item, CancellationToken.None);
+            _tasks.Add(item, CancellationToken.None);
         }
 
         if (_deadlockDetector.Value is not null)
         {
-            // we are in a nested loop. There is no reason to spawn a task here as that would just lead to deadlocks and no additional concurrency is achived
-            while (workItems.Any(e => !e.Done.Task.IsCompleted) && Tasks.TryTake(out var item, 0, _deadlockDetector.Value.Token))
+            // we are in a nested loop. There is no reason to spawn a task here as that would just lead to deadlocks and no additional concurrency is achieved
+            while (workItems.Any(e => !e.Done.Task.IsCompleted) && _tasks.TryTake(out var item, 0, _deadlockDetector.Value.Token))
             {
                 await ProcessItem(item).ConfigureAwait(false);
             }
@@ -200,6 +195,13 @@
         }
     }
 
+    /// <inheritdoc />
+    public void Dispose()
+    {
+        _tasks.Dispose();
+        _cleanupTask?.Dispose();
+    }
+
     private class TaskQueueItem
     {
         public required object Data { get; init; }