Fixed DataLoader DeadLock in tests. (#6962)

Author: michaelstaib

src/GreenDonut/src/Core/GreenDonut.csproj

@@ -28,4 +28,13 @@
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="6.0.0" />
   </ItemGroup>
 
+  <ItemGroup>
+    <Compile Update="DataLoaderBase.Fetch.cs">
+      <DependentUpon>DataLoaderBase.cs</DependentUpon>
+    </Compile>
+    <Compile Update="DataLoaderBase.IDataLoader.cs">
+      <DependentUpon>DataLoaderBase.cs</DependentUpon>
+    </Compile>
+  </ItemGroup>
+
 </Project>

src/GreenDonut/test/Core.Tests/DataLoaderTests.cs

@@ -3,26 +3,31 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
 using Snapshooter.Xunit;
 using Xunit;
+using Xunit.Abstractions;
 using static GreenDonut.TestHelpers;
 // ReSharper disable CollectionNeverUpdated.Local
 // ReSharper disable InconsistentNaming
 
 namespace GreenDonut;
 
-public class DataLoaderTests
+public class DataLoaderTests(ITestOutputHelper output)
 {
     [Fact(DisplayName = "Clear: Should not throw any exception")]
     public void ClearNoException()
     {
         // arrange
         var fetch = CreateFetch<string, string>();
-        var batchScheduler = new ManualBatchScheduler();
-        var loader = new DataLoader<string, string>(fetch, batchScheduler);
+        var services = new ServiceCollection()
+            .AddScoped<IBatchScheduler, ManualBatchScheduler>()
+            .AddDataLoader(sp => new DataLoader<string, string>(fetch, sp.GetRequiredService<IBatchScheduler>()));
+        var scope = services.BuildServiceProvider().CreateScope();
+        var dataLoader = scope.ServiceProvider.GetRequiredService<DataLoader<string, string>>();
 
         // act
-        void Verify() => loader.Clear();
+        void Verify() => dataLoader.Clear();
 
         // assert
         Assert.Null(Record.Exception(Verify));
@@ -411,11 +416,11 @@ ValueTask Fetch(
     [InlineData(5, 25, 25, 0, true, false)]
     [InlineData(5, 25, 25, 0, false, true)]
     [InlineData(5, 25, 25, 0, false, false)]
-    // [InlineData(100, 1000, 25, 25, true, true)]
-    // [InlineData(100, 1000, 25, 0, true, true)]
-    // [InlineData(100, 1000, 25, 0, true, false)]
-    // [InlineData(100, 1000, 25, 25, false, true)]
-    // [InlineData(100, 1000, 25, 0, false, false)]
+    [InlineData(100, 1000, 25, 25, true, true)]
+    [InlineData(100, 1000, 25, 0, true, true)]
+    [InlineData(100, 1000, 25, 0, true, false)]
+    [InlineData(100, 1000, 25, 25, false, true)]
+    [InlineData(100, 1000, 25, 0, false, false)]
     [Theory(DisplayName = "LoadAsync: Runs integration tests with different settings")]
     public async Task LoadTest(
         int uniqueKeys,
@@ -445,14 +450,16 @@ async ValueTask Wait()
                 => await Task.Delay(random.Next(maxDelay), cancellationToken);
         }
 
+        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+        var ct = cts.Token;
         using var cacheOwner = caching 
             ? new TaskCacheOwner()
             : null;
-
+        
         var options = new DataLoaderOptions
         {
             Cache = cacheOwner?.Cache,
-            CancellationToken = cacheOwner?.CancellationToken ?? default,
+            CancellationToken = ct,
             MaxBatchSize = batching ? 1 : maxBatchSize,
         };
 
@@ -468,26 +475,31 @@ async ValueTask Wait()
         var requests = new Task<int>[maxRequests];
 
         // act
+        output.WriteLine("LoadAsync");
         for (var i = 0; i < maxRequests; i++)
         {
             requests[i] = Task.Factory.StartNew(async () =>
             {
                 var index = random.Next(uniqueKeys);
                 var delay = random.Next(maxDelay);
 
-                await Task.Delay(delay);
+                await Task.Delay(delay, ct);
 
-                return await loader.LoadAsync(keyArray[index]);
+                return await loader.LoadAsync(keyArray[index], ct);
             }, TaskCreationOptions.RunContinuationsAsynchronously).Unwrap();
         }
 
+        output.WriteLine("Start Dispatch");
         while (requests.Any(task => !task.IsCompleted))
         {
-            await Task.Delay(25);
+            output.WriteLine("Wait");
+            await Task.Delay(25, ct);
+            output.WriteLine("Dispatch");
             batchScheduler.Dispatch();
         }
 
         // assert
+        output.WriteLine("Wait for results.");
         var responses = await Task.WhenAll(requests);
 
         foreach (var response in responses)

src/GreenDonut/test/Core.Tests/GreenDonut.Tests.csproj

@@ -11,14 +11,17 @@
 
   <ItemGroup Condition="'$(TargetFramework)' == 'net8.0'">
     <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="8.0.0" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="8.0.0" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'net7.0'">
     <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="7.0.0" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="7.0.0" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'net6.0' OR '$(TargetFramework)' == 'netstandard2.1' OR '$(TargetFramework)' == 'netstandard2.0'">
     <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="6.0.0" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="6.0.0" />
   </ItemGroup>
 
 </Project>

src/GreenDonut/test/Core.Tests/ManualBatchScheduler.cs

@@ -1,30 +1,24 @@
 using System;
 using System.Collections.Concurrent;
+using System.Diagnostics;
 using System.Threading.Tasks;
 
 namespace GreenDonut;
 
 public class ManualBatchScheduler : IBatchScheduler
 {
-    private readonly object _sync = new();
     private readonly ConcurrentQueue<Func<ValueTask>> _queue = new();
 
     public void Dispatch()
     {
-        lock(_sync)
+        while (_queue.TryDequeue(out var dispatch))
         {
-            while (_queue.TryDequeue(out var dispatch))
-            {
-                dispatch();
-            }
+            Task.Run(async () => await dispatch());
         }
     }
 
     public void Schedule(Func<ValueTask> dispatch)
     {
-        lock (_sync)
-        {
-            _queue.Enqueue(dispatch);
-        }
+        _queue.Enqueue(dispatch);
     }
-}
+}