Extremely fast Queue

SchedulerQueue.cs

using System.Diagnostics.Metrics;
using System.Runtime.CompilerServices;

using Microsoft.Extensions.Logging;

/// <summary>
/// Encapsulates a priority-based work scheduling system to execute actions at specified times.
/// </summary>
public sealed class SchedulerQueue : IDisposable
{
    private readonly string? _name;
    private readonly ILogger? _logger;

    private readonly PriorityQueue<ScheduledAction, long> _queue;

    private readonly object _lock = new();
    private readonly Thread _schedulerThread;
    private readonly CancellationTokenSource _cts;
    private bool _isDisposed;

    private long _nextWakeupTicks = long.MaxValue;

    private readonly Meter? s_meter;
    private readonly Counter<long>? s_scheduledCounter;
    private readonly Counter<long>? s_executedCounter;
    private readonly Histogram<double>? s_driftHistogram;

    private const int MaxWaitMs = int.MaxValue;

    /// <summary>
    /// Encapsulates a priority-based work scheduling system.
    /// </summary>
    /// <param name="name">The name of the scheduler thread. The preferred naming convention is <c>scheduler-{queueName}</c>.</param>
    /// <param name="logger">A logger to use for logging.</param>
    public SchedulerQueue(string? name, ILogger? logger)
    {
        _name = name;
        _logger = logger;
        _queue = new PriorityQueue<ScheduledAction, long>();
        _cts = new CancellationTokenSource();

        if (name != null)
        {
            s_meter = new Meter(name);
            s_scheduledCounter = s_meter.CreateCounter<long>($"{name}.actions.scheduled");
            s_executedCounter = s_meter.CreateCounter<long>($"{name}.actions.executed");
            s_driftHistogram = s_meter.CreateHistogram<double>($"{name}.actions.drift_ms", "ms", "Action execution latency relative to schedule time");
        }

        _schedulerThread = new Thread(SchedulerLoop)
        {
            Name = name,
            IsBackground = true,
            Priority = ThreadPriority.AboveNormal,
        };
        _schedulerThread.Start();
    }

    /// <summary>
    /// Schedules an action to be executed at a specified time.
    /// </summary>
    /// <param name="action">The action to execute at the specified time. Must not be null.</param>
    /// <param name="invocationTime">The time at which the action should be executed.</param>
    public void Schedule(Action action, DateTime invocationTime)
    {
        ArgumentNullException.ThrowIfNull(action);
        if (_isDisposed) ThrowObjectDisposed();

        var invocationTimeUtc = invocationTime.Kind switch
        {
            DateTimeKind.Utc => invocationTime,
            DateTimeKind.Local => invocationTime.ToUniversalTime(),
            DateTimeKind.Unspecified => DateTime.SpecifyKind(invocationTime, DateTimeKind.Utc),
            _ => throw new ArgumentException("Invalid DateTimeKind for invocationTime. Must be Utc, Local, or Unspecified.", nameof(invocationTime)),
        };
        var dueTicks = invocationTimeUtc.Ticks;
        s_scheduledCounter?.Add(1);

        lock (_lock)
        {
            var scheduledAction = new ScheduledAction(action, dueTicks);
            _queue.Enqueue(scheduledAction, dueTicks);

            if (dueTicks < _nextWakeupTicks)
            {
                Monitor.Pulse(_lock);
            }
        }
    }

    private void SchedulerLoop()
    {
        try
        {
            while (!_cts.IsCancellationRequested)
            {
                ScheduledAction actionToRun = default;
                int waitTimeMs;

                lock (_lock)
                {
                    _nextWakeupTicks = long.MaxValue;

                    if (_isDisposed) break;

                    if (_queue.TryPeek(out _, out var nextDueTicks))
                    {
                        var nowTicks = DateTime.UtcNow.Ticks;
                        var ticksToWait = nextDueTicks - nowTicks;

                        if (ticksToWait <= 0)
                        {
                            actionToRun = _queue.Dequeue();
                            waitTimeMs = 0;
                        }
                        else
                        {
                            _nextWakeupTicks = nextDueTicks;
                            var msToWait = ticksToWait / TimeSpan.TicksPerMillisecond;
                            waitTimeMs = msToWait switch
                            {
                                <= 0 => 1,
                                > MaxWaitMs => MaxWaitMs,
                                _ => (int)msToWait,
                            };
                        }
                    }
                    else
                    {
                        waitTimeMs = Timeout.Infinite;
                    }

                    if (actionToRun.Action == null)
                    {
                        Monitor.Wait(_lock, waitTimeMs);
                        continue;
                    }
                }

                if (actionToRun.Action != null)
                {
                    DispatchAction(actionToRun);
                }
            }
        }
        catch (Exception ex)
        {
            _logger?.LogCritical(ex, "Work Scheduler {Name} loop crashed.", _name);
        }
    }

    private void DispatchAction(ScheduledAction scheduledAction)
    {
        var nowTicks = DateTime.UtcNow.Ticks;
        var driftTicks = nowTicks - scheduledAction.ScheduledTicks;
        var driftMs = driftTicks / (double)TimeSpan.TicksPerMillisecond;

        s_driftHistogram?.Record(Math.Abs(driftMs));
        s_executedCounter?.Add(1);

        ThreadPool.UnsafeQueueUserWorkItem(static state =>
        {
            var (scheduledAction, log) = state;
            try
            {
                scheduledAction.Action();
            }
            catch (Exception ex)
            {
                log.LogError(ex, "Error executing scheduled action in {Name}", nameof(SchedulerQueue));
            }
        }, (scheduledAction, _logger), preferLocal: false);
    }

    public void Dispose()
    {
        if (_isDisposed) return;

        lock (_lock)
        {
            _isDisposed = true;
            _cts.Cancel();
            Monitor.PulseAll(_lock);
        }

        _schedulerThread.Join(TimeSpan.FromSeconds(5));
        _cts.Dispose();
    }

    [MethodImpl(MethodImplOptions.AggressiveInlining)]
    private static void ThrowObjectDisposed() => throw new ObjectDisposedException(nameof(SchedulerQueue));

    private readonly struct ScheduledAction
    {
        public readonly Action? Action;
        public readonly long ScheduledTicks;

        public ScheduledAction(Action action, long ticks)
        {
            Action = action;
            ScheduledTicks = ticks;
        }
    }
}