Teach Service to run more than one task concurrently

DalSoft · Sep 3, 2024 · 9c0ffb1 · 9c0ffb1
1 parent 8c51724
commit 9c0ffb1
Show file tree

Hide file tree

Showing 13 changed files with 184 additions and 39 deletions.
diff --git a/.gitignore b/.gitignore
@@ -34,4 +34,7 @@ _ReSharper.*
 *.ncrunch*
 
 # NUNIT
-TestResult.xml
+TestResult.xml
+
+# Rider
+.idea
diff --git a/DalSoft.Hosting.BackgroundQueue.Test/DalSoft.Hosting.BackgroundQueue.Test.csproj b/DalSoft.Hosting.BackgroundQueue.Test/DalSoft.Hosting.BackgroundQueue.Test.csproj
@@ -0,0 +1,29 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+
+    <IsPackable>false</IsPackable>
+
+    <RootNamespace>DalSoft.Hosting.BackgroundQueue.Test</RootNamespace>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="FluentAssertions" Version="5.10.3" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.8.3" />
+    <PackageReference Include="xunit" Version="2.4.1" />
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.3">
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+      <PrivateAssets>all</PrivateAssets>
+    </PackageReference>
+    <PackageReference Include="coverlet.collector" Version="1.3.0">
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+      <PrivateAssets>all</PrivateAssets>
+    </PackageReference>
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\DalSoft.Hosting.BackgroundQueue\DalSoft.Hosting.BackgroundQueue.csproj" />
+  </ItemGroup>
+
+</Project>
diff --git a/DalSoft.Hosting.BackgroundQueue.Test/MaxConcurrentCountTest.cs b/DalSoft.Hosting.BackgroundQueue.Test/MaxConcurrentCountTest.cs
@@ -0,0 +1,43 @@
+using FluentAssertions;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace DalSoft.Hosting.BackgroundQueue.Test
+{
+    public class MaxConcurrentCountTest
+    {
+        [Fact]
+        public async Task ServiceShouldRunMaxConcurrentCountTaskWhenExistInQueue()
+        {
+            var tokenSource = new CancellationTokenSource();
+            var queue = new BackgroundQueue(ex => throw ex, 10, 10); ;
+            var queueService = new BackgroundQueueService(queue);
+            var highwaterMark = 0;
+
+
+            // queue background tasks
+            for (var i = 0; i < 20; i++)
+            {
+                queue.Enqueue(async ct =>
+                {
+                    highwaterMark = Math.Max(queue.ConcurrentCount, highwaterMark);
+                    await Task.Delay(5, ct);
+                });
+            }
+
+            // process background tasks
+            var runningService = Task.Run(async () => await queueService.StartAsync(tokenSource.Token), tokenSource.Token);
+
+            // wait for all tasks to be processed
+            while(queue.Count > 0)
+            {
+                await Task.Delay(20, tokenSource.Token);
+            }
+
+            // Check that tasks run concurrently up to the maxConcurrentCount.
+            highwaterMark.Should().Be(11);
+        }
+    }
+}
diff --git a/DalSoft.Hosting.BackgroundQueue.sln b/DalSoft.Hosting.BackgroundQueue.sln
@@ -5,6 +5,8 @@ VisualStudioVersion = 15.0.27004.2000
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DalSoft.Hosting.BackgroundQueue", "DalSoft.Hosting.BackgroundQueue\DalSoft.Hosting.BackgroundQueue.csproj", "{CC5BCB5D-506A-4AC2-BD71-9FF329B17230}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DalSoft.Hosting.BackgroundQueue.Test", "DalSoft.Hosting.BackgroundQueue.Test\DalSoft.Hosting.BackgroundQueue.Test.csproj", "{91F4AEC4-5818-441C-8C47-92F64C9CF1C3}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -15,6 +17,10 @@ Global
 		{CC5BCB5D-506A-4AC2-BD71-9FF329B17230}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{CC5BCB5D-506A-4AC2-BD71-9FF329B17230}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{CC5BCB5D-506A-4AC2-BD71-9FF329B17230}.Release|Any CPU.Build.0 = Release|Any CPU
+		{91F4AEC4-5818-441C-8C47-92F64C9CF1C3}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{91F4AEC4-5818-441C-8C47-92F64C9CF1C3}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{91F4AEC4-5818-441C-8C47-92F64C9CF1C3}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{91F4AEC4-5818-441C-8C47-92F64C9CF1C3}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

diff --git a/DalSoft.Hosting.BackgroundQueue/Assets/LICENSE b/DalSoft.Hosting.BackgroundQueue/Assets/LICENSE
@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2014 Darran
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/DalSoft.Hosting.BackgroundQueue/Assets/icon.png b/DalSoft.Hosting.BackgroundQueue/Assets/icon.png
diff --git a/DalSoft.Hosting.BackgroundQueue/BackgroundQueue.cs b/DalSoft.Hosting.BackgroundQueue/BackgroundQueue.cs
@@ -5,14 +5,19 @@
 
 namespace DalSoft.Hosting.BackgroundQueue
 {
-    public class BackgroundQueue
+    public class BackgroundQueue : IBackgroundQueue
     {
         private readonly Action<Exception> _onException;
 
-        internal readonly ConcurrentQueue<Func<CancellationToken, Task>> TaskQueue = new ConcurrentQueue<Func<CancellationToken, Task>>();
-        internal readonly int MaxConcurrentCount;
-        internal readonly int MillisecondsToWaitBeforePickingUpTask;
-        internal int ConcurrentCount;
+        private readonly ConcurrentQueue<Func<CancellationToken, Task>> _taskQueue = new ConcurrentQueue<Func<CancellationToken, Task>>();
+        public int MaxConcurrentCount { get; }
+        public int MillisecondsToWaitBeforePickingUpTask { get; }
+
+        public int Count => _taskQueue.Count;
+
+        public int ConcurrentCount => _concurrentCount;
+
+        private int _concurrentCount;
 
         public BackgroundQueue(Action<Exception> onException, int maxConcurrentCount, int millisecondsToWaitBeforePickingUpTask)
         {
@@ -26,14 +31,14 @@ public BackgroundQueue(Action<Exception> onException, int maxConcurrentCount, in
 
         public void Enqueue(Func<CancellationToken, Task> task)
         {
-            TaskQueue.Enqueue(task);
+            _taskQueue.Enqueue(task);
         }
 
-        internal async Task Dequeue(CancellationToken cancellationToken)
+        public async Task Dequeue(CancellationToken cancellationToken)
         {
-            if (TaskQueue.TryDequeue(out var nextTaskAction))
+            if (_taskQueue.TryDequeue(out var nextTaskAction))
             {
-                Interlocked.Increment(ref ConcurrentCount);
+                Interlocked.Increment(ref _concurrentCount);
                 try
                 {
                     await nextTaskAction(cancellationToken);
@@ -44,7 +49,7 @@ internal async Task Dequeue(CancellationToken cancellationToken)
                 }
                 finally
                 {
-                    Interlocked.Decrement(ref ConcurrentCount);
+                    Interlocked.Decrement(ref _concurrentCount);
                 }
             }
 

diff --git a/DalSoft.Hosting.BackgroundQueue/BackgroundQueueService.cs b/DalSoft.Hosting.BackgroundQueue/BackgroundQueueService.cs
@@ -1,13 +1,14 @@
-using System.Threading;
+using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace DalSoft.Hosting.BackgroundQueue
 {
     public class BackgroundQueueService : HostedService
     {
-        private readonly BackgroundQueue _backgroundQueue;
+        private readonly IBackgroundQueue _backgroundQueue;
 
-        public BackgroundQueueService(BackgroundQueue backgroundQueue)
+        public BackgroundQueueService(IBackgroundQueue backgroundQueue)
         {
             _backgroundQueue = backgroundQueue;
         }
@@ -16,10 +17,19 @@ protected override async Task ExecuteAsync(CancellationToken cancellationToken)
         {
             while (!cancellationToken.IsCancellationRequested)
             {
-                if (_backgroundQueue.TaskQueue.Count == 0 || _backgroundQueue.ConcurrentCount > _backgroundQueue.MaxConcurrentCount)
+                if (_backgroundQueue.Count == 0 || _backgroundQueue.ConcurrentCount > _backgroundQueue.MaxConcurrentCount)
+                {
                     await Task.Delay(_backgroundQueue.MillisecondsToWaitBeforePickingUpTask, cancellationToken);
+                }
                 else
-                    await _backgroundQueue.Dequeue(cancellationToken);
+                {
+                    var concurrentTasks = new List<Task>();
+                    while (_backgroundQueue.Count > 0 && _backgroundQueue.ConcurrentCount <= _backgroundQueue.MaxConcurrentCount)
+                    {
+                        concurrentTasks.Add(_backgroundQueue.Dequeue(cancellationToken));
+                    }
+                    await Task.WhenAll(concurrentTasks);
+                }
             }
         }
     }

diff --git a/DalSoft.Hosting.BackgroundQueue/DalSoft.Hosting.BackgroundQueue.csproj b/DalSoft.Hosting.BackgroundQueue/DalSoft.Hosting.BackgroundQueue.csproj
@@ -1,26 +1,35 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
+    <VersionPrefix>1.1.0</VersionPrefix>
     <TargetFramework>netstandard2.0</TargetFramework>
     <AssemblyName>DalSoft.Hosting.BackgroundQueue</AssemblyName>
-    <IsPackable>true</IsPackable>
     <PackageId>DalSoft.Hosting.BackgroundQueue</PackageId>
-    <PackageVersion>1.0.4</PackageVersion>
-    <PackageReleaseNotes>Removed superfluous Task.Run and now pass the cancellation token to the background task.</PackageReleaseNotes>
-    <Authors>DalSoft</Authors>
+    <Title>DalSoft.Hosting.BackgroundQueue - lightweight .NET Core replacement for HostingEnvironment.QueueBackgroundWorkItem</Title>
     <Description>DalSoft.Hosting.BackgroundQueue is a very lightweight .NET Core replacement for HostingEnvironment.QueueBackgroundWorkItem using IHostedService</Description>
+    <Copyright>DalSoft Ltd</Copyright>
     <PackageTags>asp.net core webbackgrounder environment.queuebackgroundworkitem IHostedService IHost IWebHost</PackageTags>
-    <Copyright></Copyright>
-    <RepositoryUrl>https://github.com/DalSoft/DalSoft.Hosting.BackgroundQueue</RepositoryUrl>
-    <PackageLicenseUrl>https://github.com/DalSoft/DalSoft.Hosting.BackgroundQueue/blob/master/LICENSE</PackageLicenseUrl>
+    <Owners>DalSoft</Owners>
+    <Authors>DalSoft, JonasBr68</Authors>
+    <PackageReleaseNotes>Added fix to run more than one task concurrently.</PackageReleaseNotes>
+    <PackageIcon>icon.png</PackageIcon>
+    <PackageLicenseFile>LICENSE</PackageLicenseFile>
+    <PackageReadmeFile>README.md</PackageReadmeFile>
     <PackageProjectUrl>https://github.com/DalSoft/DalSoft.Hosting.BackgroundQueue</PackageProjectUrl>
-    <PackageIconUrl>http://dalsoft.co.uk/images/icon.png</PackageIconUrl>
+    <RepositoryUrl>https://github.com/DalSoft/DalSoft.Hosting.BackgroundQueue</RepositoryUrl>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
+    <GenerateAssemblyInfo>false</GenerateAssemblyInfo>
   </PropertyGroup>
 
   <ItemGroup>
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="2.0.0" />
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.0.0" />
+    <None Include="Assets\icon.png" Pack="true" PackagePath="\"/>
+    <None Include="Assets\LICENSE" Pack="true" PackagePath="\"/>
+    <None Include="../README.md" Pack="true" PackagePath="\"/>
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleToAttribute">
+      <_Parameter1>DalSoft.Hosting.BackgroundQueue.Test</_Parameter1>
+    </AssemblyAttribute>
   </ItemGroup>
 
 </Project>
diff --git a/DalSoft.Hosting.BackgroundQueue/DependencyInjection/Extensions.cs b/DalSoft.Hosting.BackgroundQueue/DependencyInjection/Extensions.cs
@@ -8,7 +8,7 @@ public static class Extensions
     {
         public static void AddBackgroundQueue(this IServiceCollection services, Action<Exception> onException, int maxConcurrentCount=1, int millisecondsToWaitBeforePickingUpTask = 1000)
         {
-            services.AddSingleton(new BackgroundQueue(onException, maxConcurrentCount, millisecondsToWaitBeforePickingUpTask));
+            services.AddSingleton<IBackgroundQueue>(new BackgroundQueue(onException, maxConcurrentCount, millisecondsToWaitBeforePickingUpTask));
             services.AddSingleton<IHostedService, BackgroundQueueService>();
         }
     }

diff --git a/DalSoft.Hosting.BackgroundQueue/HostedService.cs b/DalSoft.Hosting.BackgroundQueue/HostedService.cs
@@ -12,19 +12,19 @@ public abstract class HostedService : IHostedService
 
         public Task StartAsync(CancellationToken cancellationToken)
         {
-            // Create a linked token so we can trigger cancellation outside of this token's cancellation
+            // Create a linked token, so we can trigger cancellation outside of this token's cancellation
             _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 
             // Store the task we're executing
             _executingTask = ExecuteAsync(_cancellationTokenSource.Token);
 
-            // If the task is completed then return it, otherwise it's running
+            // If the task is completed, then return it, otherwise it's running
             return _executingTask.IsCompleted ? _executingTask : Task.CompletedTask;
         }
 
         public async Task StopAsync(CancellationToken cancellationToken)
         {
-            // Stop called without start
+            // Stop called without a start
             if (_executingTask == null)
             {
                 return;
@@ -40,7 +40,7 @@ public async Task StopAsync(CancellationToken cancellationToken)
             cancellationToken.ThrowIfCancellationRequested();
         }
 
-        // Derived classes should override this and execute a long running method until 
+        // Derived classes should override this and execute a long-running method until 
         // cancellation is requested
         protected abstract Task ExecuteAsync(CancellationToken cancellationToken);
     }

diff --git a/DalSoft.Hosting.BackgroundQueue/IBackgroundQueue.cs b/DalSoft.Hosting.BackgroundQueue/IBackgroundQueue.cs
@@ -0,0 +1,19 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DalSoft.Hosting.BackgroundQueue
+{
+    public interface IBackgroundQueue
+    {
+        int Count { get; }
+        int ConcurrentCount { get; }
+        int MaxConcurrentCount { get; }
+
+        int MillisecondsToWaitBeforePickingUpTask { get; }
+
+        Task Dequeue(CancellationToken cancellationToken);
+
+        void Enqueue(Func<CancellationToken, Task> task);
+    }
+}
diff --git a/README.md b/README.md
@@ -1,23 +1,23 @@
-### `If you find this repo / package useful all I ask is you please star it.`
+### `If you find this repo / package useful all I ask is you please star it ⭐`
 
 > **Update August 2021** Although [Microsoft.NET.Sdk.Worker](https://docs.microsoft.com/en-us/aspnet/core/fundamentals/host/hosted-services?view=aspnetcore-5.0&tabs=visual-studio) works well, you end up with a lot of bolierplate code and have to solve things like exception handling and concurrency. [MS are leaving it up to the end user](https://github.com/dotnet/extensions/issues/805) to decide how to implement (which makes sense rather than trying to implement every scenario). 
 For me I need something simple akin to HostingEnvironment.QueueBackgroundWorkItem, so I will continue to support and improve this package.
 
 # DalSoft.Hosting.BackgroundQueue
 
-> This is used in production environments, however the test coverage isn't where it needs to be, should you run into a problem please raise an issue
+> This is used in production environments however, the test coverage isn't where it needs to be. Should you run into a problem please raise an issue.
 
-DalSoft.Hosting.BackgroundQueue is a very lightweight .NET Core replacement for [HostingEnvironment.QueueBackgroundWorkItem](https://www.hanselman.com/blog/HowToRunBackgroundTasksInASPNET.aspx) it has no extra dependancies!
+DalSoft.Hosting.BackgroundQueue is a very lightweight .NET Core replacement for [HostingEnvironment.QueueBackgroundWorkItem](https://www.hanselman.com/blog/HowToRunBackgroundTasksInASPNET.aspx) it has no extra dependencies!
 
-For those of you that haven't used HostingEnvironment.QueueBackgroundWorkItem it was a simple way in .NET 4.5.2 to safely run a background task on a webhost, for example sending an email when a user registers. 
+For those of you that haven't used HostingEnvironment.QueueBackgroundWorkItem it was a simple way in .NET 4.5.2 to safely run a background task on a webhost, for example, sending an email when a user registers. 
 
-Yes there are loads of  good options (hangfire, Azure Web Jobs/Functions) for doing this, but nothing in ASP.NET Core to replace the simplicity of the classic one liner ```HostingEnvironment.QueueBackgroundWorkItem(cancellationToken => DoWork())```.
+Yes there are loads of good options (hangfire, Azure Web Jobs/Functions) for doing this, but nothing in ASP.NET Core to replace the simplicity of the classic one-liner ```HostingEnvironment.QueueBackgroundWorkItem(cancellationToken => DoWork())```.
 
 ## Supported Platforms
 
-DalSoft.Hosting.BackgroundQueue uses [IHostedService](https://blogs.msdn.microsoft.com/cesardelatorre/2017/11/18/implementing-background-tasks-in-microservices-with-ihostedservice-and-the-backgroundservice-class-net-core-2-x/) and works with any .NET Core 2.0 IWebHost i.e. a server that supports ASP.NET Core.
+DalSoft.Hosting.BackgroundQueue uses [IHostedService](https://blogs.msdn.microsoft.com/cesardelatorre/2017/11/18/implementing-background-tasks-in-microservices-with-ihostedservice-and-the-backgroundservice-class-net-core-2-x/) and works with any .NET Core 2.0 or higher IWebHost i.e. a server that supports ASP.NET Core.
 
-DalSoft.Hosting.BackgroundQueue also works with .NET Core's 2.1 lighter-weight [IHost](https://blogs.msdn.microsoft.com/cesardelatorre/2017/11/18/implementing-background-tasks-in-microservices-with-ihostedservice-and-the-backgroundservice-class-net-core-2-x/) - i.e. just services no ASP.NET Core, ideal for microservices.
+DalSoft.Hosting.BackgroundQueue also works with .NET Core's lighter-weight [IHost](https://blogs.msdn.microsoft.com/cesardelatorre/2017/11/18/implementing-background-tasks-in-microservices-with-ihostedservice-and-the-backgroundservice-class-net-core-2-x/) - i.e. just services no ASP.NET Core, ideal for microservices.
 
 ## Getting Started
 ```dos
@@ -34,7 +34,7 @@ public void ConfigureServices(IServiceCollection services)
       });
 }
 ```
-> This setups DalSoft.Hosting.BackgroundQueue using .NET Core's DI container. If your using a different DI container you need to register BackgroundQueue and BackgroundQueueService as singletons.
+> This setups DalSoft.Hosting.BackgroundQueue using .NET Core's DI container. If you're using a different DI container, you need to register IBackgroundQueue and BackgroundQueueService as singletons.
  
 **maxConcurrentCount (optional)**
 maxConcurrentCount is the number of Tasks allowed to run in the background concurrently. maxConcurrentCount defaults to 1.
@@ -49,10 +49,10 @@ You are running tasks in the background on a different thread you need to know w
 
 ## Queuing a Background Task
 
-To queue a background Task just add ```BackgroundQueue``` to your controller's constructor and call ```Enqueue```.
+To queue a background Task just add ```IBackgroundQueue``` to your controller's constructor and call ```Enqueue```.
 
 ```cs
-public EmailController(BackgroundQueue backgroundQueue)
+public EmailController(IBackgroundQueue backgroundQueue)
 {
    _backgroundQueue = backgroundQueue;
 }