feat(headless): add deterministic multi-session scheduler

src/AcDream.Headless/Hosting/HeadlessProcessScheduler.cs

@@ -0,0 +1,249 @@
+using AcDream.Runtime;
+
+namespace AcDream.Headless.Hosting;
+
+internal sealed class HeadlessProcessScheduler
+{
+    private sealed class Slot
+    {
+        internal Slot(
+            HeadlessSessionHost session,
+            long startTimestamp,
+            long periodTicks)
+        {
+            Session = session;
+            LastTurnTimestamp = startTimestamp;
+            NextDeadline = AddTicks(startTimestamp, periodTicks);
+        }
+
+        internal HeadlessSessionHost Session { get; }
+        internal long LastTurnTimestamp { get; set; }
+        internal long NextDeadline { get; set; }
+    }
+
+    private readonly TimeProvider _timeProvider;
+    private readonly long _periodTicks;
+    private readonly int _maximumCatchUpTurns;
+    private readonly Slot[] _slots;
+    private long _waitCount;
+    private long _turnCount;
+    private long _catchUpCollapseCount;
+
+    internal HeadlessProcessScheduler(
+        IReadOnlyList<HeadlessSessionHost> sessions,
+        TimeProvider? timeProvider = null,
+        TimeSpan? turnPeriod = null,
+        int maximumCatchUpTurns = 8)
+    {
+        ArgumentNullException.ThrowIfNull(sessions);
+        if (sessions.Count == 0)
+        {
+            throw new ArgumentException(
+                "A headless scheduler requires at least one session.",
+                nameof(sessions));
+        }
+        if (maximumCatchUpTurns <= 0)
+        {
+            throw new ArgumentOutOfRangeException(
+                nameof(maximumCatchUpTurns));
+        }
+
+        _timeProvider = timeProvider ?? TimeProvider.System;
+        TimeSpan configuredTurnPeriod =
+            turnPeriod ?? TimeSpan.FromMilliseconds(15);
+        if (configuredTurnPeriod <= TimeSpan.Zero)
+            throw new ArgumentOutOfRangeException(nameof(turnPeriod));
+        _periodTicks = DurationToTimestampTicks(
+            _timeProvider,
+            configuredTurnPeriod);
+        _maximumCatchUpTurns = maximumCatchUpTurns;
+
+        long start = _timeProvider.GetTimestamp();
+        _slots = new Slot[sessions.Count];
+        for (int index = 0; index < sessions.Count; index++)
+        {
+            _slots[index] = new Slot(
+                sessions[index]
+                    ?? throw new ArgumentException(
+                        "A headless scheduler session cannot be null.",
+                        nameof(sessions)),
+                start,
+                _periodTicks);
+        }
+    }
+
+    internal HeadlessSchedulerSnapshot CaptureSnapshot()
+    {
+        int activeSessionCount = 0;
+        for (int index = 0; index < _slots.Length; index++)
+        {
+            if (!_slots[index].Session.IsPolicyComplete)
+                activeSessionCount++;
+        }
+
+        return new HeadlessSchedulerSnapshot(
+            _slots.Length,
+            Interlocked.Read(ref _waitCount),
+            Interlocked.Read(ref _turnCount),
+            Interlocked.Read(ref _catchUpCollapseCount),
+            activeSessionCount,
+            NextDeadline());
+    }
+
+    internal async Task RunAsync(CancellationToken cancellationToken)
+    {
+        while (!cancellationToken.IsCancellationRequested
+            && HasActiveSession())
+        {
+            long now = _timeProvider.GetTimestamp();
+            if (DispatchDue(now))
+                continue;
+
+            long deadline = NextDeadline();
+            TimeSpan delay = deadline <= now
+                ? TimeSpan.Zero
+                : _timeProvider.GetElapsedTime(now, deadline);
+            Interlocked.Increment(ref _waitCount);
+            await Task.Delay(
+                    delay,
+                    _timeProvider,
+                    cancellationToken)
+                .ConfigureAwait(false);
+        }
+    }
+
+    internal bool DispatchDue(long nowTimestamp)
+    {
+        bool dispatched = false;
+        foreach (Slot slot in _slots)
+        {
+            HeadlessSessionHost session = slot.Session;
+            if (session.IsPolicyComplete)
+                continue;
+
+            if (session.IsReconnectPending)
+            {
+                if (nowTimestamp < session.ReconnectDeadline)
+                    continue;
+
+                RuntimeSessionStartResult result =
+                    session.CompletePendingReconnect(nowTimestamp);
+                if (result.Status
+                    is not RuntimeSessionStartStatus.Connected)
+                {
+                    throw result.Error
+                        ?? new InvalidOperationException(
+                            $"Headless session reconnect completed with {result.Status}.");
+                }
+                slot.LastTurnTimestamp = nowTimestamp;
+                slot.NextDeadline = AddTicks(
+                    nowTimestamp,
+                    _periodTicks);
+                dispatched = true;
+                continue;
+            }
+
+            if (nowTimestamp < slot.NextDeadline)
+                continue;
+
+            DispatchSessionDue(slot, nowTimestamp);
+            dispatched = true;
+        }
+        return dispatched;
+    }
+
+    private void DispatchSessionDue(Slot slot, long nowTimestamp)
+    {
+        long dueCount =
+            1L + ((nowTimestamp - slot.NextDeadline) / _periodTicks);
+        int turns = (int)Math.Min(
+            dueCount,
+            _maximumCatchUpTurns);
+        bool collapse = dueCount > _maximumCatchUpTurns;
+
+        for (int turn = 0; turn < turns; turn++)
+        {
+            if (slot.Session.IsPolicyComplete
+                || slot.Session.IsReconnectPending)
+            {
+                break;
+            }
+
+            bool finalCollapsedTurn =
+                collapse && turn == turns - 1;
+            long targetTimestamp = finalCollapsedTurn
+                ? nowTimestamp
+                : slot.NextDeadline;
+            double deltaSeconds = _timeProvider
+                .GetElapsedTime(
+                    slot.LastTurnTimestamp,
+                    targetTimestamp)
+                .TotalSeconds;
+            slot.Session.Tick(deltaSeconds);
+            Interlocked.Increment(ref _turnCount);
+            slot.LastTurnTimestamp = targetTimestamp;
+            slot.NextDeadline = AddTicks(
+                targetTimestamp,
+                _periodTicks);
+        }
+
+        if (collapse)
+            Interlocked.Increment(ref _catchUpCollapseCount);
+    }
+
+    private long NextDeadline()
+    {
+        long next = long.MaxValue;
+        foreach (Slot slot in _slots)
+        {
+            if (slot.Session.IsPolicyComplete)
+                continue;
+            long candidate = slot.Session.IsReconnectPending
+                ? slot.Session.ReconnectDeadline
+                : slot.NextDeadline;
+            if (candidate < next)
+                next = candidate;
+        }
+        return next;
+    }
+
+    private bool HasActiveSession()
+    {
+        for (int index = 0; index < _slots.Length; index++)
+        {
+            if (!_slots[index].Session.IsPolicyComplete)
+                return true;
+        }
+        return false;
+    }
+
+    private static long DurationToTimestampTicks(
+        TimeProvider timeProvider,
+        TimeSpan duration)
+    {
+        double ticks =
+            duration.TotalSeconds * timeProvider.TimestampFrequency;
+        if (!double.IsFinite(ticks)
+            || ticks <= 0d
+            || ticks > long.MaxValue)
+        {
+            throw new ArgumentOutOfRangeException(nameof(duration));
+        }
+        return Math.Max(
+            1L,
+            checked((long)Math.Ceiling(ticks)));
+    }
+
+    private static long AddTicks(long timestamp, long ticks) =>
+        timestamp > long.MaxValue - ticks
+            ? long.MaxValue
+            : timestamp + ticks;
+}
+
+internal readonly record struct HeadlessSchedulerSnapshot(
+    int SessionCount,
+    long WaitCount,
+    long TurnCount,
+    long CatchUpCollapseCount,
+    int ActiveSessionCount,
+    long NextDeadline);