Readme

Author: soenneker

README.md

@@ -5,8 +5,111 @@
 # ![](https://user-images.githubusercontent.com/4441470/224455560-91ed3ee7-f510-4041-a8d2-3fc093025112.png) Soenneker.Utils.BackgroundQueue
 ### A high-performance background Task/ValueTask queue
 
+## Overview
+`BackgroundQueue` provides an efficient way to manage background task execution in .NET applications. It helps prevent application overload by processing tasks in a controlled, asynchronous manner.
+
+## Features
+- Supports both `ValueTask` and `Task` types.
+- Configurable queue size to limit resource usage.
+- Built-in tracking of running and pending tasks.
+- Extension methods for easy setup and management.
+- Includes a hosted service for automatic background processing.
+
 ## Installation
 
-```
+```sh
 dotnet add package Soenneker.Utils.BackgroundQueue
 ```
+
+Register the `BackgroundQueue`:
+```csharp
+void ConfigureServices(IServiceCollection services)
+{
+    services.AddBackgroundQueueAsSingleton();
+}
+```
+
+### Starting
+```csharp
+await serviceProvider.WarmupAndStartBackgroundQueue(cancellationToken);
+```
+
+For synchronous start:
+```csharp
+serviceProvider.WarmupAndStartBackgroundQueueSync(cancellationToken);
+```
+
+### Stopping
+To stop the service:
+```csharp
+await serviceProvider.StopBackgroundQueue(cancellationToken);
+```
+
+For synchronous stop:
+```csharp
+serviceProvider.StopBackgroundQueueSync(cancellationToken);
+```
+
+## Configuration
+Configure the queue length and task tracking settings in your application:
+
+```json
+{
+  "Background": {
+    "QueueLength": 5000,
+    "LockCounts": false,
+    "Log": false
+  }
+}
+```
+
+- `QueueLength`: Defines the maximum number of tasks in the queue.
+- `LockCounts`: Enables thread-safe counting of running tasks.
+- `Log`: Outputs task tracking information to `ILogger`
+
+## Initializing the Queue
+To use `BackgroundQueue`, you probably want to inject it via your constructor:
+
+```csharp
+IBackgroundQueue _queue;
+
+void MyClass(IBackgroundQueue queue)
+{
+    _queue = queue;
+}
+```
+
+## Queueing Tasks
+
+### Queuing a `ValueTask`
+Rather than wrapping the task, you can elide it directly to avoid an extra state machine:
+```csharp
+await _queue.QueueValueTask(_ => someValueTask(), cancellationToken);
+```
+
+### Queuing a `Task`
+Similarly, for `Task`:
+```csharp
+await _queue.QueueTask(_ => someTask(), cancellationToken);
+```
+
+## Waiting for Queue to Empty
+To ensure all queued tasks finish before proceeding:
+```csharp
+await queue.WaitUntilEmpty(cancellationToken);
+```
+
+## Task Tracking
+The queue tracks:
+- The number of active `ValueTask` and `Task` instances.
+- Whether any tasks are still processing.
+
+To check if tasks are running:
+```csharp
+bool isProcessing = await queueInformationUtil.IsProcessing(cancellationToken);
+```
+
+To get current task counts:
+```csharp
+var (taskCount, valueTaskCount) = await queueInformationUtil.GetCountsOfProcessing(cancellationToken);
+```

src/Abstract/IBackgroundQueue.cs

@@ -41,4 +41,6 @@ public interface IBackgroundQueue
     /// <param name="cancellationToken">An optional token to cancel the operation.</param>
     /// <returns>A <see cref="ValueTask"/> containing a function that represents the dequeued work item, which accepts a <see cref="CancellationToken"/> and returns a <see cref="Task"/>.</returns>
     ValueTask<Func<CancellationToken, Task>> DequeueTask(CancellationToken cancellationToken = default);
+
+    ValueTask WaitUntilEmpty(CancellationToken cancellationToken = default);
 }

src/BackgroundQueue.cs

@@ -6,6 +6,7 @@
 using Microsoft.Extensions.Logging;
 using Soenneker.Extensions.Double;
 using Soenneker.Extensions.MethodInfo;
+using Soenneker.Extensions.Task;
 using Soenneker.Extensions.ValueTask;
 using Soenneker.Utils.BackgroundQueue.Abstract;
 
@@ -21,14 +22,14 @@ public class BackgroundQueue : IBackgroundQueue
     private readonly int _queueWarning;
 
     private readonly ILogger<BackgroundQueue> _logger;
-    private readonly IQueueInformationUtil _informationUtil;
+    private readonly IQueueInformationUtil _queueInformationUtil;
 
     private readonly bool _log;
 
-    public BackgroundQueue(IConfiguration config, ILogger<BackgroundQueue> logger, IQueueInformationUtil informationUtil)
+    public BackgroundQueue(IConfiguration config, ILogger<BackgroundQueue> logger, IQueueInformationUtil queueInformationUtil)
     {
         _logger = logger;
-        _informationUtil = informationUtil;
+        _queueInformationUtil = queueInformationUtil;
 
         var configQueueLength = config.GetValue<int>("Background:QueueLength");
         _log = config.GetValue<bool>("Background:Log");
@@ -60,7 +61,7 @@ public async ValueTask QueueValueTask(Func<CancellationToken, ValueTask> workIte
     {
         // TODO: need to redo this, we're going to get too many warnings
 
-        int count = await _informationUtil.IncrementValueTaskCounter(cancellationToken).NoSync();
+        int count = await _queueInformationUtil.IncrementValueTaskCounter(cancellationToken).NoSync();
 
         if (count > _queueWarning)
         {
@@ -76,7 +77,7 @@ public async ValueTask QueueValueTask(Func<CancellationToken, ValueTask> workIte
 
     public async ValueTask QueueTask(Func<CancellationToken, Task> workItem, CancellationToken cancellationToken = default)
     {
-        int count = await _informationUtil.IncrementTaskCounter(cancellationToken).NoSync();
+        int count = await _queueInformationUtil.IncrementTaskCounter(cancellationToken).NoSync();
 
         if (count > _queueWarning)
         {
@@ -99,4 +100,30 @@ public ValueTask<Func<CancellationToken, Task>> DequeueTask(CancellationToken ca
     {
         return _taskChannel.Reader.ReadAsync(cancellationToken);
     }
+
+    public async ValueTask WaitUntilEmpty(CancellationToken cancellationToken = default)
+    {
+        const int delayMs = 500;
+
+        bool isProcessing;
+
+        do
+        {
+            isProcessing = await _queueInformationUtil.IsProcessing(cancellationToken).ConfigureAwait(false);
+
+            if (isProcessing)
+            {
+                if (_log)
+                {
+                    _logger.LogDebug("Delaying for {ms}ms (Background queue emptying...)...", delayMs);
+                }
+
+                await Task.Delay(delayMs, cancellationToken).NoSync();
+            }
+            else
+            {
+                _logger.LogDebug("Background queue is empty; continuing");
+            }
+        } while (isProcessing);
+    }
 }

src/QueuedHostedService.cs

@@ -43,36 +43,36 @@ public override Task StartAsync(CancellationToken cancellationToken)
     /// <summary>
     /// Triggered when the application host is ready to start the service.
     /// </summary>
-    protected override Task ExecuteAsync(CancellationToken stoppingToken)
+    protected override Task ExecuteAsync(CancellationToken cancellationToken)
     {
         if (_log)
             _logger.LogDebug("~~ QueuedHostedService: Executing...");
 
-        Task valueTaskProcessing = ValueTaskProcessing(stoppingToken);
-        Task taskProcessing = TaskProcessing(stoppingToken);
+        Task valueTaskProcessing = ValueTaskProcessing(cancellationToken);
+        Task taskProcessing = TaskProcessing(cancellationToken);
 
         return Task.WhenAll(valueTaskProcessing, taskProcessing);
     }
 
-    private async Task TaskProcessing(CancellationToken stoppingToken)
+    private async Task TaskProcessing(CancellationToken cancellationToken)
     {
-        while (!stoppingToken.IsCancellationRequested)
+        while (!cancellationToken.IsCancellationRequested)
         {
             Func<CancellationToken, Task>? workItem = null;
 
             try
             {
                 string? workItemName = null;
 
-                workItem = await _queue.DequeueTask(stoppingToken).NoSync();
+                workItem = await _queue.DequeueTask(cancellationToken).NoSync();
 
                 if (_log)
                 {
                     workItemName = workItem.Method.GetSignature();
                     _logger.LogDebug("~~ QueuedHostedService: Starting Task: {item}", workItemName);
                 }
 
-                await workItem(stoppingToken).NoSync();
+                await workItem(cancellationToken).NoSync();
 
                 if (_log)
                     _logger.LogDebug("~~ QueuedHostedService: Completed Task: {item}", workItemName);
@@ -83,30 +83,30 @@ private async Task TaskProcessing(CancellationToken stoppingToken)
             }
             finally
             {
-                await _queueInformationUtil.DecrementTaskCounter(stoppingToken).NoSync();
+                await _queueInformationUtil.DecrementTaskCounter(cancellationToken).NoSync();
             }
         }
     }
 
-    private async Task ValueTaskProcessing(CancellationToken stoppingToken)
+    private async Task ValueTaskProcessing(CancellationToken cancellationToken)
     {
-        while (!stoppingToken.IsCancellationRequested)
+        while (!cancellationToken.IsCancellationRequested)
         {
             Func<CancellationToken, ValueTask>? workItem = null;
 
             try
             {
                 string? workItemName = null;
 
-                workItem = await _queue.DequeueValueTask(stoppingToken).NoSync();
+                workItem = await _queue.DequeueValueTask(cancellationToken).NoSync();
 
                 if (_log)
                 {
                     workItemName = workItem.Method.GetSignature();
                     _logger.LogDebug("~~ QueuedHostedService: Starting ValueTask: {item}", workItemName);
                 }
 
-                await workItem(stoppingToken).NoSync();
+                await workItem(cancellationToken).NoSync();
 
                 if (_log)
                     _logger.LogDebug("~~ QueuedHostedService: Completed ValueTask: {item}", workItemName);
@@ -117,7 +117,7 @@ private async Task ValueTaskProcessing(CancellationToken stoppingToken)
             }
             finally
             {
-                await _queueInformationUtil.DecrementValueTaskCounter(stoppingToken).NoSync();
+                await _queueInformationUtil.DecrementValueTaskCounter(cancellationToken).NoSync();
             }
         }
     }