From 0d9c9c9eccf2f547f824ad47f2038c378930fdd2 Mon Sep 17 00:00:00 2001 From: Shadowghost Date: Fri, 4 Sep 2026 19:23:29 +0200 Subject: Fix the library scheduler never retiring its runners --- .../LimitedConcurrencyLibrarySchedulerTests.cs | 213 +++++++++++++++++++++ 1 file changed, 213 insertions(+) create mode 100644 tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs (limited to 'tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs') diff --git a/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs b/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs new file mode 100644 index 0000000000..21a719a78f --- /dev/null +++ b/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs @@ -0,0 +1,213 @@ +using System; +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using MediaBrowser.Controller.Configuration; +using MediaBrowser.Controller.LibraryTaskScheduler; +using MediaBrowser.Model.Configuration; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace Jellyfin.Controller.Tests.LibraryTaskScheduler +{ + public class LimitedConcurrencyLibrarySchedulerTests + { + private static readonly TimeSpan _shortGracePeriod = TimeSpan.FromMilliseconds(50); + + // Generous, because these only ever wait for something that should already have happened. + private static readonly TimeSpan _timeout = TimeSpan.FromSeconds(10); + + [Fact] + public async Task Enqueue_ProcessesEveryItem() + { + using var appStopping = new CancellationTokenSource(); + var scheduler = CreateScheduler(appStopping); + await using (scheduler) + { + var data = Enumerable.Range(0, 100).ToArray(); + var processed = new ConcurrentBag(); + + await scheduler.Enqueue( + data, + (item, _) => + { + processed.Add(item); + return Task.CompletedTask; + }, + new Progress(), + CancellationToken.None); + + Assert.Equal(data, processed.Order()); + } + } + + [Fact] + public async Task Enqueue_WithFailingWorker_StillCompletes() + { + using var appStopping = new CancellationTokenSource(); + var scheduler = CreateScheduler(appStopping); + await using (scheduler) + { + await scheduler.Enqueue( + Enumerable.Range(0, 20).ToArray(), + (item, _) => item % 2 == 0 ? throw new InvalidOperationException("boom") : Task.CompletedTask, + new Progress(), + CancellationToken.None); + } + } + + /// + /// The runners wait on a source linked to , + /// so a shutdown has to reach them. It does not travel from the linked source back to the one + /// the cleanup cancels, which is what made them immortal. + /// + [Fact] + public async Task ApplicationStopping_RetiresRunners() + { + using var appStopping = new CancellationTokenSource(); + + // Long enough that the cleanup cannot be what retires them. + var scheduler = CreateScheduler(appStopping, gracePeriod: TimeSpan.FromMinutes(5)); + await using (scheduler) + { + await RunOneOperation(scheduler); + Assert.True(scheduler.ActiveRunnerCount > 0); + + await appStopping.CancelAsync(); + + await WaitForAsync(() => scheduler.ActiveRunnerCount == 0); + } + } + + /// + /// The cleanup used to be a one shot: it never released the scheduling slot it took, so + /// every runner spawned after the first pass stayed around for the lifetime of the server. + /// + [Fact] + public async Task Enqueue_RetiresIdleRunnersAfterEveryOperation() + { + using var appStopping = new CancellationTokenSource(); + var scheduler = CreateScheduler(appStopping); + await using (scheduler) + { + for (var round = 0; round < 3; round++) + { + await RunOneOperation(scheduler); + Assert.True(scheduler.ActiveRunnerCount > 0, $"no runner spawned in round {round}"); + + await WaitForAsync(() => scheduler.ActiveRunnerCount == 0); + } + } + } + + /// + /// Disposing used to sit out the rest of the cleanup grace period, holding up shutdown for + /// up to a minute. + /// + [Fact] + public async Task DisposeAsync_DoesNotWaitOutTheGracePeriod() + { + using var appStopping = new CancellationTokenSource(); + var scheduler = CreateScheduler(appStopping, gracePeriod: TimeSpan.FromMinutes(5)); + + await RunOneOperation(scheduler); + + var stopwatch = Stopwatch.StartNew(); + await scheduler.DisposeAsync(); + + Assert.True(stopwatch.Elapsed < _timeout, $"disposing took {stopwatch.Elapsed}"); + } + + [Fact] + public async Task Enqueue_AfterDispose_DoesNothing() + { + using var appStopping = new CancellationTokenSource(); + var scheduler = CreateScheduler(appStopping); + await scheduler.DisposeAsync(); + + var processed = 0; + await scheduler.Enqueue( + Enumerable.Range(0, 10).ToArray(), + (_, _) => + { + Interlocked.Increment(ref processed); + return Task.CompletedTask; + }, + new Progress(), + CancellationToken.None); + + Assert.Equal(0, processed); + } + + [Theory] + [InlineData(1)] + [InlineData(4)] + public async Task Enqueue_FromWithinAWorker_DoesNotDeadlock(int fanout) + { + using var appStopping = new CancellationTokenSource(); + var scheduler = CreateScheduler(appStopping, fanout: fanout); + await using (scheduler) + { + var inner = 0; + + var outer = scheduler.Enqueue( + Enumerable.Range(0, 8).ToArray(), + (_, _) => scheduler.Enqueue( + Enumerable.Range(0, 4).ToArray(), + (_, _) => + { + Interlocked.Increment(ref inner); + return Task.CompletedTask; + }, + new Progress(), + CancellationToken.None), + new Progress(), + CancellationToken.None); + + await outer.WaitAsync(_timeout, TestContext.Current.CancellationToken); + + Assert.Equal(32, inner); + } + } + + private static LimitedConcurrencyLibraryScheduler CreateScheduler( + CancellationTokenSource appStopping, + int fanout = 4, + TimeSpan? gracePeriod = null) + { + var lifetime = new Mock(); + lifetime.SetupGet(x => x.ApplicationStopping).Returns(() => appStopping.Token); + + var configurationManager = new Mock(); + configurationManager.SetupGet(x => x.Configuration) + .Returns(new ServerConfiguration { LibraryScanFanoutConcurrency = fanout }); + + return new LimitedConcurrencyLibraryScheduler( + lifetime.Object, + NullLogger.Instance, + configurationManager.Object, + gracePeriod ?? _shortGracePeriod); + } + + private static Task RunOneOperation(LimitedConcurrencyLibraryScheduler scheduler) + => scheduler.Enqueue( + Enumerable.Range(0, 8).ToArray(), + (_, _) => Task.CompletedTask, + new Progress(), + CancellationToken.None); + + private static async Task WaitForAsync(Func condition) + { + var stopwatch = Stopwatch.StartNew(); + while (!condition()) + { + Assert.True(stopwatch.Elapsed < _timeout, "timed out waiting for the scheduler to settle"); + await Task.Delay(20, TestContext.Current.CancellationToken); + } + } + } +} -- cgit v1.2.3 From e5cd3381acefbe13aa36a26059c28629e23e1944 Mon Sep 17 00:00:00 2001 From: Shadowghost Date: Sat, 5 Sep 2026 07:08:14 +0200 Subject: Limit cache size again Co-Authored-By: Cody Robibero --- .../Providers/DirectoryService.cs | 91 ++++++++++++++++++---- .../DirectoryServiceTests.cs | 73 +++++++++++++++++ .../LimitedConcurrencyLibrarySchedulerTests.cs | 3 + 3 files changed, 152 insertions(+), 15 deletions(-) (limited to 'tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs') diff --git a/MediaBrowser.Controller/Providers/DirectoryService.cs b/MediaBrowser.Controller/Providers/DirectoryService.cs index 684e247f86..38872d3cbe 100644 --- a/MediaBrowser.Controller/Providers/DirectoryService.cs +++ b/MediaBrowser.Controller/Providers/DirectoryService.cs @@ -5,13 +5,19 @@ using System.Collections.Concurrent; using System.Collections.Generic; using System.IO; using System.Linq; +using System.Threading; using MediaBrowser.Model.IO; namespace MediaBrowser.Controller.Providers { public class DirectoryService : IDirectoryService { - // TODO make static and switch to FastConcurrentLru. + // TODO replace with one shared bounded cache. + private const int MaxCachedRecords = 100_000; + private const int AccessIntervalMs = 1_000; + // Timeout cache if no access for 5 minutes. + private const int IdleTimeoutMs = 5 * 60 * 1_000; + private readonly ConcurrentDictionary _cache = new(StringComparer.Ordinal); private readonly ConcurrentDictionary _fileCache = new(StringComparer.Ordinal); @@ -20,6 +26,12 @@ namespace MediaBrowser.Controller.Providers private readonly IFileSystem _fileSystem; + // ConcurrentDictionary.Count locks the dictionary, so keep an estimated counter. + // Concurrent factory runs can overcount and a clear racing an add can undercount, + // it only has to be roughly right. + private int _recordCount; + private long _lastAccess = Environment.TickCount64; + public DirectoryService(IFileSystem fileSystem) { _fileSystem = fileSystem; @@ -27,20 +39,26 @@ namespace MediaBrowser.Controller.Providers public FileSystemMetadata[] GetFileSystemEntries(string path) { + DropCacheIfIdleOrFull(); + return _cache.GetOrAdd( path, - static (p, fileSystem) => + static (p, state) => { + FileSystemMetadata[] entries; try { - return fileSystem.GetFileSystemEntries(p).ToArray(); + entries = state.FileSystem.GetFileSystemEntries(p).ToArray(); } catch (DirectoryNotFoundException) { - return []; + entries = []; } + + Interlocked.Add(ref state.Service._recordCount, entries.Length + 1); + return entries; }, - _fileSystem); + (FileSystem: _fileSystem, Service: this)); } public List GetDirectories(string path) @@ -89,13 +107,18 @@ namespace MediaBrowser.Controller.Providers public FileSystemMetadata? GetFileSystemEntry(string path) { + DropCacheIfIdleOrFull(); + if (!_fileCache.TryGetValue(path, out var result)) { var file = _fileSystem.GetFileSystemInfo(path); if (file?.Exists ?? false) { result = file; - _fileCache.TryAdd(path, result); + if (_fileCache.TryAdd(path, result)) + { + Interlocked.Increment(ref _recordCount); + } } } @@ -107,25 +130,31 @@ namespace MediaBrowser.Controller.Providers public IReadOnlyList GetFilePaths(string path, bool clearCache) { - if (clearCache) + if (clearCache && _filePathCache.TryRemove(path, out var cached)) { - _filePathCache.TryRemove(path, out _); + Interlocked.Add(ref _recordCount, -(cached.Count + 1)); } + DropCacheIfIdleOrFull(); + var filePaths = _filePathCache.GetOrAdd( path, - static (p, fileSystem) => + static (p, state) => { + List filePaths; try { - return fileSystem.GetFilePaths(p).OrderBy(x => x).ToList(); + filePaths = state.FileSystem.GetFilePaths(p).OrderBy(x => x).ToList(); } catch (DirectoryNotFoundException) { - return []; + filePaths = []; } + + Interlocked.Add(ref state.Service._recordCount, filePaths.Count + 1); + return filePaths; }, - _fileSystem); + (FileSystem: _fileSystem, Service: this)); return filePaths; } @@ -146,11 +175,43 @@ namespace MediaBrowser.Controller.Providers return _fileSystem.GetFileSystemEntryPaths(path).Any(); } + private void DropCacheIfIdleOrFull() + { + var nowMs = Environment.TickCount64; + var idleMs = nowMs - Volatile.Read(ref _lastAccess); + + if (idleMs >= IdleTimeoutMs || Volatile.Read(ref _recordCount) >= MaxCachedRecords) + { + _cache.Clear(); + _fileCache.Clear(); + _filePathCache.Clear(); + Volatile.Write(ref _recordCount, 0); + Volatile.Write(ref _lastAccess, nowMs); + return; + } + + if (idleMs >= AccessIntervalMs) + { + Volatile.Write(ref _lastAccess, nowMs); + } + } + private void Forget(string path) { - _cache.TryRemove(path, out _); - _fileCache.TryRemove(path, out _); - _filePathCache.TryRemove(path, out _); + if (_cache.TryRemove(path, out var entries)) + { + Interlocked.Add(ref _recordCount, -(entries.Length + 1)); + } + + if (_fileCache.TryRemove(path, out _)) + { + Interlocked.Decrement(ref _recordCount); + } + + if (_filePathCache.TryRemove(path, out var filePaths)) + { + Interlocked.Add(ref _recordCount, -(filePaths.Count + 1)); + } } } } diff --git a/tests/Jellyfin.Controller.Tests/DirectoryServiceTests.cs b/tests/Jellyfin.Controller.Tests/DirectoryServiceTests.cs index 7c275b78cc..e57fbfe473 100644 --- a/tests/Jellyfin.Controller.Tests/DirectoryServiceTests.cs +++ b/tests/Jellyfin.Controller.Tests/DirectoryServiceTests.cs @@ -1,3 +1,4 @@ +using System.Globalization; using System.IO; using System.Linq; using MediaBrowser.Controller.Providers; @@ -326,6 +327,78 @@ namespace Jellyfin.Controller.Tests fileSystemMock.Verify(f => f.GetFileSystemEntries(parentPath), Times.Once); } + [Fact] + public void GetFileSystemEntries_MoreRecordsThanTheCeiling_DropsCache() + { + // Charged by the files in a listing, not the number of listings, so a few big folders + // reach the limit where a lot of small ones would not. + const int FolderCount = 60; + var bigListing = new FileSystemMetadata[5000]; + for (var i = 0; i < bigListing.Length; i++) + { + bigListing[i] = new FileSystemMetadata + { + FullName = "/music/track" + i.ToString(CultureInfo.InvariantCulture), + IsDirectory = false + }; + } + + var fileSystemMock = new Mock(); + fileSystemMock.Setup(f => f.GetFileSystemEntries(It.IsAny())) + .Returns(bigListing); + + var directoryService = new DirectoryService(fileSystemMock.Object); + + const string FirstPath = "/music/artist0"; + directoryService.GetFileSystemEntries(FirstPath); + + for (var i = 1; i < FolderCount; i++) + { + directoryService.GetFileSystemEntries("/music/artist" + i.ToString(CultureInfo.InvariantCulture)); + } + + directoryService.GetFileSystemEntries(FirstPath); + + fileSystemMock.Verify(f => f.GetFileSystemEntries(FirstPath), Times.Exactly(2)); + } + + [Fact] + public void GetFileSystemEntries_RepeatedlyInvalidatedFolder_KeepsUnrelatedEntriesCached() + { + // Invalidating gives the records back, so churning one folder must not add up to the + // ceiling and drop everything else with it. + const int ChurnCount = 50; + var bigListing = new FileSystemMetadata[5000]; + for (var i = 0; i < bigListing.Length; i++) + { + bigListing[i] = new FileSystemMetadata + { + FullName = "/music/track" + i.ToString(CultureInfo.InvariantCulture), + IsDirectory = false + }; + } + + var fileSystemMock = new Mock(); + fileSystemMock.Setup(f => f.GetFileSystemEntries(It.IsAny())) + .Returns(bigListing); + + var directoryService = new DirectoryService(fileSystemMock.Object); + + const string ChurnedPath = "/music/watched"; + const string StablePath = "/music/untouched"; + directoryService.GetFileSystemEntries(StablePath); + + for (var i = 0; i < ChurnCount; i++) + { + directoryService.GetFileSystemEntries(ChurnedPath); + directoryService.Invalidate(ChurnedPath); + } + + directoryService.GetFileSystemEntries(StablePath); + + fileSystemMock.Verify(f => f.GetFileSystemEntries(StablePath), Times.Once); + } + [Fact] public void GetFileSystemEntry_MissingPath_IsNotRemembered() { diff --git a/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs b/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs index 21a719a78f..f776e893a0 100644 --- a/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs +++ b/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs @@ -65,6 +65,7 @@ namespace Jellyfin.Controller.Tests.LibraryTaskScheduler /// so a shutdown has to reach them. It does not travel from the linked source back to the one /// the cleanup cancels, which is what made them immortal. /// + /// A representing the asynchronous unit test. [Fact] public async Task ApplicationStopping_RetiresRunners() { @@ -87,6 +88,7 @@ namespace Jellyfin.Controller.Tests.LibraryTaskScheduler /// The cleanup used to be a one shot: it never released the scheduling slot it took, so /// every runner spawned after the first pass stayed around for the lifetime of the server. /// + /// A representing the asynchronous unit test. [Fact] public async Task Enqueue_RetiresIdleRunnersAfterEveryOperation() { @@ -108,6 +110,7 @@ namespace Jellyfin.Controller.Tests.LibraryTaskScheduler /// Disposing used to sit out the rest of the cleanup grace period, holding up shutdown for /// up to a minute. /// + /// A representing the asynchronous unit test. [Fact] public async Task DisposeAsync_DoesNotWaitOutTheGracePeriod() { -- cgit v1.2.3 From f8470630be0f3fa7b8052ebd822f23531d30da2f Mon Sep 17 00:00:00 2001 From: Shadowghost Date: Sat, 5 Sep 2026 09:58:04 +0200 Subject: Retire runners that are cancelled before they start --- .../LibraryTaskScheduler/LimitedConcurrencyLibraryScheduler.cs | 4 +++- .../LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs | 6 +++--- 2 files changed, 6 insertions(+), 4 deletions(-) (limited to 'tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs') diff --git a/MediaBrowser.Controller/LibraryTaskScheduler/LimitedConcurrencyLibraryScheduler.cs b/MediaBrowser.Controller/LibraryTaskScheduler/LimitedConcurrencyLibraryScheduler.cs index 3c6ffe3cbd..be75117b6f 100644 --- a/MediaBrowser.Controller/LibraryTaskScheduler/LimitedConcurrencyLibraryScheduler.cs +++ b/MediaBrowser.Controller/LibraryTaskScheduler/LimitedConcurrencyLibraryScheduler.cs @@ -191,12 +191,14 @@ public sealed class LimitedConcurrencyLibraryScheduler : ILimitedConcurrencyLibr // Keyed on its own stop source, because cancelling that is what reaches the linked // source the runner waits on. Cancellation does not travel the other way. + // Started without the runner's own token: a task cancelled before it is scheduled + // never runs its body, so it would never take itself out of _taskRunners again. _taskRunners.Add( stopToken, Task.Factory.StartNew( ItemWorker, (stopToken, combinedSource), - combinedSource.Token, + CancellationToken.None, TaskCreationOptions.PreferFairness, TaskScheduler.Default)); } diff --git a/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs b/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs index f776e893a0..686d839f4f 100644 --- a/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs +++ b/tests/Jellyfin.Controller.Tests/LibraryTaskScheduler/LimitedConcurrencyLibrarySchedulerTests.cs @@ -65,7 +65,7 @@ namespace Jellyfin.Controller.Tests.LibraryTaskScheduler /// so a shutdown has to reach them. It does not travel from the linked source back to the one /// the cleanup cancels, which is what made them immortal. /// - /// A representing the asynchronous unit test. + /// A representing the asynchronous unit test. [Fact] public async Task ApplicationStopping_RetiresRunners() { @@ -88,7 +88,7 @@ namespace Jellyfin.Controller.Tests.LibraryTaskScheduler /// The cleanup used to be a one shot: it never released the scheduling slot it took, so /// every runner spawned after the first pass stayed around for the lifetime of the server. /// - /// A representing the asynchronous unit test. + /// A representing the asynchronous unit test. [Fact] public async Task Enqueue_RetiresIdleRunnersAfterEveryOperation() { @@ -110,7 +110,7 @@ namespace Jellyfin.Controller.Tests.LibraryTaskScheduler /// Disposing used to sit out the rest of the cleanup grace period, holding up shutdown for /// up to a minute. /// - /// A representing the asynchronous unit test. + /// A representing the asynchronous unit test. [Fact] public async Task DisposeAsync_DoesNotWaitOutTheGracePeriod() { -- cgit v1.2.3