Align task scheduler defaults with guidance
This commit is contained in:
1 parent
9cb50d77ca
commit
eda3c2128a
2 files changed
+136
-8
No files matched your search
@@ -0,0 +1,116 @@
|
|||||||
|
using System;
|
||||||
|
using System.Collections.Concurrent;
|
||||||
|
using System.Collections.Generic;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
|
||||||
|
namespace PerfectNumbers.Core;
|
||||||
|
|
||||||
|
internal sealed class TaskThreadPool : IDisposable
|
||||||
|
{
|
||||||
|
private readonly struct WorkItem
|
||||||
|
{
|
||||||
|
public WorkItem(Task task)
|
||||||
|
{
|
||||||
|
Task = task;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Task Task { get; }
|
||||||
|
}
|
||||||
|
|
||||||
|
private readonly ConcurrentQueue<WorkItem> _pendingTasks = new();
|
||||||
|
private readonly SemaphoreSlim _availableTasks = new(0);
|
||||||
|
private readonly Thread[] _threads;
|
||||||
|
private readonly Action<Task> _taskExecutor;
|
||||||
|
private int _disposed;
|
||||||
|
|
||||||
|
public TaskThreadPool(int minimumThreads, Action<Task> taskExecutor)
|
||||||
|
{
|
||||||
|
_taskExecutor = taskExecutor ?? throw new ArgumentNullException(nameof(taskExecutor));
|
||||||
|
_threads = new Thread[minimumThreads];
|
||||||
|
|
||||||
|
for (int i = 0; i < minimumThreads; i++)
|
||||||
|
{
|
||||||
|
Thread thread = new(WorkerLoop)
|
||||||
|
{
|
||||||
|
IsBackground = true,
|
||||||
|
Name = $"{nameof(TaskThreadPool)}-{i + 1}"
|
||||||
|
};
|
||||||
|
|
||||||
|
_threads[i] = thread;
|
||||||
|
thread.Start();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public int ThreadCount => _threads.Length;
|
||||||
|
|
||||||
|
public void Queue(Task task)
|
||||||
|
{
|
||||||
|
if (Volatile.Read(ref _disposed) != 0)
|
||||||
|
{
|
||||||
|
throw new ObjectDisposedException(nameof(TaskThreadPool));
|
||||||
|
}
|
||||||
|
|
||||||
|
_pendingTasks.Enqueue(new WorkItem(task));
|
||||||
|
_availableTasks.Release();
|
||||||
|
}
|
||||||
|
|
||||||
|
public IEnumerable<Task> GetScheduledTasks()
|
||||||
|
{
|
||||||
|
WorkItem[] snapshot = _pendingTasks.ToArray();
|
||||||
|
|
||||||
|
if (snapshot.Length == 0)
|
||||||
|
{
|
||||||
|
return Array.Empty<Task>();
|
||||||
|
}
|
||||||
|
|
||||||
|
Task[] tasks = new Task[snapshot.Length];
|
||||||
|
|
||||||
|
for (int i = 0; i < snapshot.Length; i++)
|
||||||
|
{
|
||||||
|
tasks[i] = snapshot[i].Task;
|
||||||
|
}
|
||||||
|
|
||||||
|
return tasks;
|
||||||
|
}
|
||||||
|
|
||||||
|
public void Dispose()
|
||||||
|
{
|
||||||
|
if (Interlocked.Exchange(ref _disposed, 1) != 0)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
for (int i = 0; i < _threads.Length; i++)
|
||||||
|
{
|
||||||
|
_availableTasks.Release();
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (Thread thread in _threads)
|
||||||
|
{
|
||||||
|
thread.Join();
|
||||||
|
}
|
||||||
|
|
||||||
|
_availableTasks.Dispose();
|
||||||
|
}
|
||||||
|
|
||||||
|
private void WorkerLoop()
|
||||||
|
{
|
||||||
|
while (true)
|
||||||
|
{
|
||||||
|
_availableTasks.Wait();
|
||||||
|
|
||||||
|
if (Volatile.Read(ref _disposed) != 0)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!_pendingTasks.TryDequeue(out WorkItem workItem))
|
||||||
|
{
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
_taskExecutor(workItem.Task);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,27 +1,39 @@
|
|||||||
|
using System;
|
||||||
|
using System.Collections.Generic;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
|
||||||
namespace PerfectNumbers.Core;
|
namespace PerfectNumbers.Core;
|
||||||
|
|
||||||
public sealed class UnboundedTaskScheduler : TaskScheduler
|
public sealed class UnboundedTaskScheduler : TaskScheduler
|
||||||
{
|
{
|
||||||
public static UnboundedTaskScheduler Instance { get; } = new();
|
public static UnboundedTaskScheduler Instance { get; } = new();
|
||||||
|
|
||||||
|
private readonly TaskThreadPool _threadPool;
|
||||||
|
|
||||||
private UnboundedTaskScheduler()
|
private UnboundedTaskScheduler()
|
||||||
{
|
{
|
||||||
ThreadPool.SetMaxThreads(100_000, 100_000);
|
_threadPool = new TaskThreadPool(10_340, ExecuteTask);
|
||||||
ThreadPool.SetMinThreads(10_340, 100);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public override int MaximumConcurrencyLevel => int.MaxValue;
|
public override int MaximumConcurrencyLevel => int.MaxValue;
|
||||||
|
|
||||||
protected override void QueueTask(Task task)
|
protected override void QueueTask(Task task)
|
||||||
{
|
{
|
||||||
ThreadPool.UnsafeQueueUserWorkItem( (state) =>
|
_threadPool.Queue(task);
|
||||||
|
}
|
||||||
|
|
||||||
|
private void ExecuteTask(Task task)
|
||||||
{
|
{
|
||||||
var (scheduler, task) = ((UnboundedTaskScheduler, Task))state!;
|
TryExecuteTask(task);
|
||||||
scheduler.TryExecuteTask(task);
|
|
||||||
}, (this, task));
|
|
||||||
}
|
}
|
||||||
|
|
||||||
protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) => TryExecuteTask(task);
|
protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
|
||||||
|
{
|
||||||
|
return TryExecuteTask(task);
|
||||||
|
}
|
||||||
|
|
||||||
protected override IEnumerable<Task>? GetScheduledTasks() => null;
|
protected override IEnumerable<Task>? GetScheduledTasks()
|
||||||
|
{
|
||||||
|
return _threadPool.GetScheduledTasks();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
Reference in new issue
Block a user