Keep TaskThreadPool workers idle until work arrives

This commit is contained in:
Marek Lemańczyk committed 2025-10-20 12:15:31 +02:00
1 parent 9c08cf48a7
commit 5fbbfd7590
1 file changed
+35 -22
+35 -22
View File
@@ -18,15 +18,21 @@ internal sealed class TaskThreadPool : IDisposable
public Task Task { get; } public Task Task { get; }
} }
private readonly ConcurrentQueue<WorkItem> _pendingTasks = new(); private readonly ConcurrentQueue<WorkItem> _pendingTaskQueue = new();
private readonly SemaphoreSlim _availableTasks = new(0); private readonly BlockingCollection<WorkItem> _pendingTasks;
private readonly Thread[] _threads; private readonly Thread[] _threads;
private readonly Action<Task> _taskExecutor; private readonly Action<Task> _taskExecutor;
private int _disposed; private int _disposed;
public TaskThreadPool(int minimumThreads, Action<Task> taskExecutor) public TaskThreadPool(int minimumThreads, Action<Task> taskExecutor)
{ {
if (minimumThreads <= 0)
{
throw new ArgumentOutOfRangeException(nameof(minimumThreads));
}
_taskExecutor = taskExecutor ?? throw new ArgumentNullException(nameof(taskExecutor)); _taskExecutor = taskExecutor ?? throw new ArgumentNullException(nameof(taskExecutor));
_pendingTasks = new BlockingCollection<WorkItem>(_pendingTaskQueue);
_threads = new Thread[minimumThreads]; _threads = new Thread[minimumThreads];
for (int i = 0; i < minimumThreads; i++) for (int i = 0; i < minimumThreads; i++)
@@ -46,18 +52,33 @@ internal sealed class TaskThreadPool : IDisposable
public void Queue(Task task) public void Queue(Task task)
{ {
if (task is null)
{
throw new ArgumentNullException(nameof(task));
}
if (Volatile.Read(ref _disposed) != 0) if (Volatile.Read(ref _disposed) != 0)
{ {
throw new ObjectDisposedException(nameof(TaskThreadPool)); throw new ObjectDisposedException(nameof(TaskThreadPool));
} }
_pendingTasks.Enqueue(new WorkItem(task)); try
_availableTasks.Release(); {
_pendingTasks.Add(new WorkItem(task));
}
catch (InvalidOperationException ex) when (Volatile.Read(ref _disposed) != 0)
{
throw new ObjectDisposedException(nameof(TaskThreadPool), ex);
}
catch (ObjectDisposedException ex)
{
throw new ObjectDisposedException(nameof(TaskThreadPool), ex);
}
} }
public IEnumerable<Task> GetScheduledTasks() public IEnumerable<Task> GetScheduledTasks()
{ {
WorkItem[] snapshot = _pendingTasks.ToArray(); WorkItem[] snapshot = _pendingTaskQueue.ToArray();
if (snapshot.Length == 0) if (snapshot.Length == 0)
{ {
@@ -81,36 +102,28 @@ internal sealed class TaskThreadPool : IDisposable
return; return;
} }
for (int i = 0; i < _threads.Length; i++) _pendingTasks.CompleteAdding();
{
_availableTasks.Release();
}
foreach (Thread thread in _threads) foreach (Thread thread in _threads)
{ {
thread.Join(); thread.Join();
} }
_availableTasks.Dispose(); _pendingTasks.Dispose();
} }
private void WorkerLoop() private void WorkerLoop()
{ {
while (true) try
{ {
_availableTasks.Wait(); foreach (WorkItem workItem in _pendingTasks.GetConsumingEnumerable())
if (Volatile.Read(ref _disposed) != 0)
{ {
return; _taskExecutor(workItem.Task);
} }
}
if (!_pendingTasks.TryDequeue(out WorkItem workItem)) catch (ObjectDisposedException)
{ {
continue; // The collection was disposed while the worker was waiting. Exit gracefully.
}
_taskExecutor(workItem.Task);
} }
} }
} }