My project I am working on requires the following functionality for the processed messages:
- each message has a priority
- messages with higher priorities should have precedence over the ones with lower priority
- if a burst of high priority messages is sent, they should not complete starve lower priority messages (in this regard "priority" is more a QoS level)
I have implemented a simple version that does the following:
- all incoming tasks are spread on a pool of
BlockingCollections (queues) - a task will be scheduled to the first empty queue, if any
- if no queues are empty, the scheduler picks the queue with the lowest count (least amount of queued tasks) based on the priority. Lowest count considers only queues between the first one and total number of queues * some fraction based on priority. High priorities lead to scanning more queue counts, but not 100%.
My implementation is as follows:
Task Scheduler factory and Task Scheduler
public interface IPriorityTaskSchedulerFactory
{
IPriorityTaskScheduler Create();
IPriorityTaskScheduler Create(PriorityTaskSchedulerConfig config);
}
public class PriorityTaskSchedulerFactory : IPriorityTaskSchedulerFactory
{
public const int DefaultMaxPriority = 5;
public const int DefaultMaxParallelism = 50;
public const int DefaultMaxQueueSize = 5;
public const int DefaultBlockingCollectionWaitTimeout = 100;
public static readonly int[] DefaultPrioritySearchableSlotRangePercs = new int[] { 30, 45, 60, 75, 90 };
//TODO: add an identifier (name) for the scheduler
public IPriorityTaskScheduler Create()
{
return new PriorityTaskScheduler(new PriorityTaskSchedulerConfig
{
MaxPriority = DefaultMaxPriority,
MaxParallelism = DefaultMaxParallelism,
MaxQueueSize = DefaultMaxQueueSize,
BlockingCollectionWaitTimeout = DefaultBlockingCollectionWaitTimeout,
PrioritySearchableSlotRangePercs = DefaultPrioritySearchableSlotRangePercs
});
}
public IPriorityTaskScheduler Create(PriorityTaskSchedulerConfig config)
{
return new PriorityTaskScheduler(config);
}
}
public interface IPriorityTaskScheduler
{
void Start();
void ScheduleMessageProcessing(PriorityTaskSchedulerItem item);
public bool DiagnosticMode { get; set; }
}
public class PriorityTaskScheduler : IPriorityTaskScheduler
{
private readonly int _maxPriority;
private readonly int _maxParallelism;
private readonly int _maxQueueSize;
private readonly int _blockingCollectionWaitTimeout;
private BlockingCollection<PriorityTaskSchedulerItem>[] _taskQueues = default!;
private readonly SemaphoreSlim _semaphore = new(1, 1);
private int _totalInProcess = 0;
private int _maxTotalInProcess = 0;
private readonly Random random = new();
public double MaxLockWait { get; private set; } = 0.0;
public double TotalLockWait { get; private set; } = 0.0;
/// <summary>
/// specifies range in % of the total slots are searchable for current priority
/// if all slots have some load on them
/// </summary>
/// <remarks>index-0 correspond to priority=1</remarks>
private readonly int[] _prioritySearchableSlotRangePercs;
/// <summary>
/// specifies range in number of slots that are searchable for current priority
/// if all slots have some load on them
/// </summary>
private int[] _prioritySearchableSlotRange = default!;
/// <summary>
/// this should be set only for local dev testing
/// </summary>
public bool DiagnosticMode { get; set; } = false;
public PriorityTaskScheduler(PriorityTaskSchedulerConfig config)
{
_maxPriority = config.MaxPriority;
_maxParallelism = config.MaxParallelism;
_maxQueueSize = config.MaxQueueSize;
_blockingCollectionWaitTimeout = config.BlockingCollectionWaitTimeout;
_prioritySearchableSlotRangePercs = config.PrioritySearchableSlotRangePercs;
Init();
}
private void Init()
{
_taskQueues = new BlockingCollection<PriorityTaskSchedulerItem>[_maxParallelism];
_prioritySearchableSlotRange = new int[_maxPriority];
for (int i = 0; i < _maxPriority; i++)
{
_prioritySearchableSlotRange[i] = (int)(_maxParallelism * _prioritySearchableSlotRangePercs[i] / 100.0);
}
for (int i = 0; i < _maxParallelism; i++)
{
_taskQueues[i] = new BlockingCollection<PriorityTaskSchedulerItem>(_maxQueueSize);
}
}
public void Start()
{
for (int i = 0; i < _maxParallelism; i++)
{
int queueIndex = i; // local copy for closure
Task.Run(async () => await ProcessTasks(queueIndex));
}
}
public void ScheduleMessageProcessing(PriorityTaskSchedulerItem item)
{
bool addedOk = false;
while (!addedOk)
{
var queue = GetQueueToUse(item.PriorityIndex);
addedOk = queue.TryAdd(item, _blockingCollectionWaitTimeout);
}
}
private BlockingCollection<PriorityTaskSchedulerItem> GetQueueToUse(byte priorityIndex)
{
var emptyQueues = Array.FindAll(_taskQueues, q => q.Count == 0);
if (emptyQueues.Length > 0)
{
int queueIndex = random.Next(0, emptyQueues.Length);
return emptyQueues[queueIndex];
}
Stopwatch sw = Stopwatch.StartNew();
_semaphore.Wait();
try
{
var queueToUse = GetQueueToUseWhenAllLoaded(priorityIndex);
if (queueToUse != null)
return queueToUse;
}
finally
{
var elapsed = sw.Elapsed.TotalMilliseconds;
TotalLockWait += elapsed;
if (elapsed > MaxLockWait)
MaxLockWait = elapsed;
_semaphore.Release();
}
// this should not happen, but ensure the item gets a queue
int randomIndex = (byte)random.Next(0, _maxParallelism);
return _taskQueues[randomIndex];
}
private BlockingCollection<PriorityTaskSchedulerItem>? GetQueueToUseWhenAllLoaded(byte priorityIndex)
{
int prioritySearchableSlotRange = _prioritySearchableSlotRange[priorityIndex];
BlockingCollection<PriorityTaskSchedulerItem>? queueWithMinimumLoad = null;
int minLoad = int.MaxValue;
for (int i = 0; i < prioritySearchableSlotRange; i++)
{
int currentLoad = _taskQueues[i].Count;
if (currentLoad < minLoad)
{
minLoad = currentLoad;
queueWithMinimumLoad = _taskQueues[i];
}
}
return queueWithMinimumLoad;
}
private async Task ProcessTasks(int queueIndex)
{
foreach (var schedulerItem in _taskQueues[queueIndex].GetConsumingEnumerable())
{
try
{
Interlocked.Increment(ref _totalInProcess);
if (_totalInProcess > _maxTotalInProcess)
_maxTotalInProcess = _totalInProcess;
if (DiagnosticMode)
LogSchedulerInfo();
var task = schedulerItem.ProcessingTask();
if (task != null)
await task;
}
catch (Exception ex)
{
//TODO: do proper handling here
Console.WriteLine($"An error occurred in a task: {ex.Message}");
}
finally
{
Interlocked.Decrement(ref _totalInProcess);
if (DiagnosticMode)
LogSchedulerInfo();
}
}
}
private void LogSchedulerInfo()
{
string inProcessStr = string.Join(",", _taskQueues.Select(q => q.Count));
Console.WriteLine($"{_totalInProcess}/{_maxTotalInProcess} in queues: {inProcessStr}");
}
}
Related DTOs
public class PriorityTaskSchedulerConfig
{
public int MaxPriority { get; set; }
public int MaxParallelism { get; set; }
public int MaxQueueSize { get; set; }
public int BlockingCollectionWaitTimeout { get; set; }
public int[] PrioritySearchableSlotRangePercs { get; set; } = default!;
}
/// <summary>
/// specifies a task to be run by the PriorityTaskScheduler with a priority (QoS)
/// </summary>
public class PriorityTaskSchedulerItem
{
/// <summary>
/// 0-based priority (QoS level) of the task. The higher, the more important
/// </summary>
public byte PriorityIndex { get; set; }
/// <summary>
/// Task to be run when the item is to be processed
/// </summary>
public Func<Task> ProcessingTask { get; set; } = default!;
/// <summary>
/// cancellation token
/// </summary>
public CancellationToken Ct { get; set; } = default!;
}
Usage example
One possible usage is to process messages coming from RabbitMQ which has per priority queues. This means that several consumer can share a priority task scheduler:
public void StartConsumingWithPriorities(IPriorityTaskScheduler priorityTaskScheduler)
{
_priorityTaskScheduler = priorityTaskScheduler;
//TODO: error handling and logging
var queue = Bus.QueueDeclare(QueueName);
_ = Bus.Consume(queue, (body, properties, info, ct) => Task.Run(() =>
{
// not copying into a byte[] may lead memory corruption and deserialization failure
var bodyCopy = new byte[body.Length];
body.Span.CopyTo(bodyCopy);
ScheduleMessageProcessing(bodyCopy, properties, info, ct);
}, ct),
cfg =>
{
if (PrefetchCount.HasValue)
cfg.WithPrefetchCount(PrefetchCount.Value);
});
}
private void ScheduleMessageProcessing(byte[] bodyCopy, MessageProperties properties, MessageReceivedInfo receivedInfo, CancellationToken ct)
{
byte priorityIndex = (byte)(properties.Priority - 1);
_priorityTaskScheduler.ScheduleMessageProcessing(new PriorityTaskSchedulerItem
{
PriorityIndex = priorityIndex,
ProcessingTask = async () =>
{
await ProcessMessage(bodyCopy, properties, receivedInfo, ct);
},
Ct = ct
});
}
Review goals
I would like for the task scheduler to be reviewed with focus on the following:
- potential issues in the future, as I want for it to be message broker agnostic
- multithreading best practices
- naming
