Added queue polling delay constructor parameter and QueuePollingDelayMilliseconds property

Author: XHunter74

TaskExecutor.Tests/TaskExecutorTests.cs

@@ -5,12 +5,10 @@ public class TaskExecutorTests
     [Fact]
     public async Task EnqueueTask_ShouldExecuteTasksConcurrently()
     {
-        // Arrange
         var cts = new CancellationTokenSource();
         var executor = new TaskExecutor(3, cts.Token);
         int completedTasks = 0;
 
-        // Act
         for (int i = 0; i < 10; i++)
         {
             executor.EnqueueTask(i, async () =>
@@ -20,27 +18,24 @@ public async Task EnqueueTask_ShouldExecuteTasksConcurrently()
             });
         }
 
-        await Task.Delay(500); // Allow time for tasks to start
+        await Task.Delay(500);
 
         while (executor.HasRunningTasks)
         {
             await Task.Delay(100);
         }
         await executor.StopAsync();
 
-        // Assert
         Assert.Equal(10, completedTasks);
     }
 
     [Fact]
     public async Task ChangeConcurrency_ShouldAdjustConcurrencyLevel()
     {
-        // Arrange
         var cts = new CancellationTokenSource();
         var executor = new TaskExecutor(1, cts.Token);
         int runningTasks = 0;
 
-        // Act
         for (int i = 0; i < 5; i++)
         {
             executor.EnqueueTask(i, async () =>
@@ -51,13 +46,12 @@ public async Task ChangeConcurrency_ShouldAdjustConcurrencyLevel()
             });
         }
 
-        await Task.Delay(100); // Allow initial tasks to start
+        await Task.Delay(100);
         Assert.Equal(1, runningTasks);
 
         executor.ChangeConcurrency(3);
-        await Task.Delay(100); // Allow new concurrency level to take effect
+        await Task.Delay(100);
 
-        // Assert
         Assert.True(runningTasks >= 2);
 
         await executor.StopAsync();
@@ -66,27 +60,23 @@ public async Task ChangeConcurrency_ShouldAdjustConcurrencyLevel()
     [Fact]
     public async Task OnTaskError_ShouldCaptureTaskExceptions()
     {
-        // Arrange
         var cts = new CancellationTokenSource();
         var executor = new TaskExecutor(3, cts.Token);
         Exception? capturedException = null;
 
         executor.OnTaskError += (id, ex) => capturedException = ex;
 
-        // Act
         executor.EnqueueTask(1, () => throw new InvalidOperationException("Test exception"));
-        await Task.Delay(200); // Allow time for the task to fail
+        await Task.Delay(200);
         await executor.StopAsync();
 
-        // Assert
         Assert.NotNull(capturedException);
         Assert.IsType<InvalidOperationException>(capturedException);
     }
 
     [Fact]
     public async Task StopAsync_ShouldWaitForRunningTasksToComplete()
     {
-        // Arrange
         var cts = new CancellationTokenSource();
         var executor = new TaskExecutor(3, cts.Token);
         bool taskCompleted = false;
@@ -97,11 +87,37 @@ public async Task StopAsync_ShouldWaitForRunningTasksToComplete()
             taskCompleted = true;
         });
 
-        // Act
-        await Task.Delay(100); // Allow time for the task to start
+        await Task.Delay(100);
         await executor.StopAsync();
 
-        // Assert
         Assert.True(taskCompleted);
     }
+
+    [Fact]
+    public void QueuePollingDelayMilliseconds_Setter_ShouldThrowOnNegative()
+    {
+        var executor = new TaskExecutor(1, CancellationToken.None);
+        Assert.Throws<ArgumentException>(() => executor.QueuePollingDelayMilliseconds = -1);
+    }
+
+    [Fact]
+    public void Constructor_ShouldThrowOnNegativeQueuePollingDelay()
+    {
+        Assert.Throws<ArgumentException>(() => new TaskExecutor(1, -10));
+    }
+
+    [Fact]
+    public void Constructor_ShouldThrowOnZeroOrNegativeConcurrency()
+    {
+        Assert.Throws<ArgumentException>(() => new TaskExecutor(0, CancellationToken.None));
+        Assert.Throws<ArgumentException>(() => new TaskExecutor(-1, CancellationToken.None));
+    }
+
+    [Fact]
+    public void ChangeConcurrency_ShouldThrowOnZeroOrNegative()
+    {
+        var executor = new TaskExecutor(1, CancellationToken.None);
+        Assert.Throws<ArgumentException>(() => executor.ChangeConcurrency(0));
+        Assert.Throws<ArgumentException>(() => executor.ChangeConcurrency(-5));
+    }
 }

TaskExecutor/TaskExecutor.cs

@@ -1,5 +1,4 @@
 using System.Collections.Concurrent;
-using System.Threading.Tasks;
 
 namespace TaskExecutor;
 
@@ -8,11 +7,13 @@ namespace TaskExecutor;
 /// </summary>
 public class TaskExecutor : IDisposable
 {
+    private const int DefaultQueuePollingDelayMilliseconds = 50;
     private readonly ConcurrentQueue<TaskForExecute> _taskQueue = new();
     private readonly CancellationTokenSource _internalCts = new();
     private readonly ActiveTaskRegistry _taskRegistry = new();
     private readonly CancellationToken _externalCancellationToken;
     private readonly SemaphoreSlim _semaphore;
+    private int _queuePollingDelayMilliseconds;
     private int _maxConcurrency;
     private bool _disposed;
 
@@ -33,6 +34,20 @@ public bool HasRunningTasks
         }
     }
 
+    /// <summary>
+    /// Gets or sets the delay, in milliseconds, between polling attempts for the queue.
+    /// </summary>
+    public int QueuePollingDelayMilliseconds
+    {
+        get => _queuePollingDelayMilliseconds;
+        set
+        {
+            if (value < 0)
+                throw new ArgumentException("Queue polling delay must be non-negative");
+            _queuePollingDelayMilliseconds = value;
+        }
+    }
+
     /// <summary>
     /// Constructor for the TaskExecutor class.
     /// </summary>
@@ -60,6 +75,33 @@ public TaskExecutor(int initialConcurrency, CancellationToken cancellationToken
         StartProcessing();
     }
 
+    /// <summary>
+    /// Constructor for the TaskExecutor class.
+    /// </summary>
+    /// <param name="initialConcurrency">
+    /// Specifies the initial level of concurrency for the TaskExecutor. 
+    /// This determines how many tasks can run concurrently. 
+    /// Must be greater than 0.
+    /// </param>
+    /// <param name="queuePollingDelayMilliseconds">
+    /// Specifies the delay, in milliseconds, between polling attempts when the task queue is empty.
+    /// Must be non-negative. A higher value reduces CPU usage but may increase task start latency.
+    /// </param>
+    /// </param>
+    /// <param name="cancellationToken">
+    /// An optional CancellationToken that can be used to signal cancellation 
+    /// of task processing from an external source.
+    /// </param>
+    /// <exception cref="ArgumentException">
+    /// Thrown when the initialConcurrency is less than or equal to 0.
+    /// </exception>
+    public TaskExecutor(int initialConcurrency, int queuePollingDelayMilliseconds = DefaultQueuePollingDelayMilliseconds, CancellationToken cancellationToken = default) : this(initialConcurrency, cancellationToken)
+    {
+        if (queuePollingDelayMilliseconds < 0)
+            throw new ArgumentException("Queue polling delay must be non-negative");
+        _queuePollingDelayMilliseconds = queuePollingDelayMilliseconds;
+    }
+
     /// <summary>
     /// Enqueues a task for execution.
     /// </summary>
@@ -129,7 +171,7 @@ private void StartProcessing()
                     }
                     else
                     {
-                        await Task.Delay(50, _internalCts.Token).ConfigureAwait(false);
+                        await Task.Delay(_queuePollingDelayMilliseconds, _internalCts.Token).ConfigureAwait(false);
                     }
                 }
             }

release-notes.txt

@@ -1,3 +1,4 @@
+2025-05-30 Added queue polling delay constructor parameter and QueuePollingDelayMilliseconds property
 2025-05-29 Refactored code. Improved stability.
 2025-04-29 Added .ConfigureAwait(false) to all asynchronous calls in TaskExecutor to improve performance and avoid deadlocks.
 2025-04-27 Initial version