CorePool.cs

using System.Diagnostics;
using System.Threading.Channels;

namespace CSharpener;

public sealed class CorePool
{
    public static int DetectedCores => Environment.ProcessorCount;

    private readonly CoreSlot[] _slots;
    private readonly SemaphoreSlim _leadCapacity;
    private readonly object _lock = new();

    public CorePool(int? coreCount = null, int workersPerCore = 4)
    {
        var cores = coreCount ?? DetectedCores;
        _leadCapacity = new SemaphoreSlim(cores * workersPerCore, cores * workersPerCore);

        _slots = [.. Enumerable.Range(0, cores).Select(
            id => new CoreSlot(id, workersPerCore, _leadCapacity))];
    }

    public async Task<int> ReserveLeadAsync(CancellationToken ct = default)
    {
        await _leadCapacity.WaitAsync(ct);
        lock (_lock)
        {
            var bestIdx   = -1;
            var bestLeads = int.MaxValue;
            for (var i = 0; i < _slots.Length; i++)
            {
                if (!_slots[i].CanAcceptLead || _slots[i].TotalLeads >= bestLeads) continue;
                bestLeads = _slots[i].TotalLeads;
                bestIdx   = i;
            }
            _slots[bestIdx].ReserveLeadSlot();
            return bestIdx;
        }
    }

    public void Route(IWorkItem item, int coreId, CancellationToken ct)
    {
        _slots[coreId].Enqueue(item, ct);
    }

    public Task StartAsync(CancellationToken ct)
    {
        return Task.WhenAll(_slots.Select(s => s.RunAsync(ct)));
    }

    public string StatusLine()
    {
        lock (_lock)
            return string.Join("  |  ", _slots.Select(s =>
                $"Core{s.Id}: {s.Inflight} inflight / {s.TotalLeads} leads"));
    }
}

// ── CoreSlot ──────────────────────────────────────────────────────────────────

/// <summary>
/// Two-stage pipeline:  _intake → StagingLoop (CHECKED) → _ready → WorkerLoop (STARTED)
///
/// CHECKED gate rules (StagingLoop):
///   Children  : always pass to _ready immediately.
///   S (Token) leads  : always pass; register as pending S so C leads defer.
///   C (Ticket) leads : defer to _recheck while pendingS > 0 (unless starved out).
///
/// STARTED guard (WorkerLoop):
///   S leads : always pass — they ARE the preference.
///   C leads : if within MinDeltaMs of the last STARTED lead → defer to _recheck.
///             When S lead passes the guard, _pendingS decrements.
///
/// Starvation prevention: OriginalCheckedAt anchors the C lead's age.
/// After MinDeltaMs × 4, C forces through regardless of pendingS.
/// </summary>
public sealed class CoreSlot(int id, int workerCount, SemaphoreSlim leadCapacity)
{
    public int Id { get; } = id;
    private int WorkerCount { get; } = workerCount;
    public int Inflight => _inflight;
    public int TotalLeads => _reservedLeads + _inflightLeads;
    public bool CanAcceptLead => TotalLeads < WorkerCount;
    private long _lastLeadStartedTicks;

    private const double MinDeltaMs = 1.0;
    private const double StarveAfterMs = MinDeltaMs * 4;

    // Primary intake from AdmissionQueue (ADMITTED items)
    private readonly Channel<(IWorkItem item, CancellationToken ct)> _intake =
        Channel.CreateUnbounded<(IWorkItem, CancellationToken)>(
            new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });

    // Priority re-entry lane (CHECKED items deferred from staging or STARTED guard)
    private readonly Channel<(IWorkItem item, CancellationToken ct, int reentries)> _recheck =
        Channel.CreateUnbounded<(IWorkItem, CancellationToken, int)>(
            new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });

    // Cleared items waiting for workers
    private readonly Channel<(IWorkItem item, CancellationToken ct)> _ready =
        Channel.CreateUnbounded<(IWorkItem, CancellationToken)>(
            new UnboundedChannelOptions { SingleReader = false, SingleWriter = false });

    // Comparand state
    private int _pendingS; // S leads registered in staging, not yet STARTED
    private int _inflight;
    private int _reservedLeads;
    private int _inflightLeads;

    public void ReserveLeadSlot() => Interlocked.Increment(ref _reservedLeads);

    public void Enqueue(IWorkItem item, CancellationToken ct)
        => _intake.Writer.TryWrite((item, ct));

    public Task RunAsync(CancellationToken ct)
    {
        var tasks = new List<Task> { StagingLoop(ct) };
        for (var i = 0; i < WorkerCount; i++)
            tasks.Add(WorkerLoop(i, ct));
        return Task.WhenAll(tasks);
    }

    // ── Staging loop ──────────────────────────────────────────────────────────

    private async Task StagingLoop(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            var processed = false;

            // Priority: drain re-entries before new intake
            while (_recheck.Reader.TryRead(out var re))
            {
                Route(re.item, re.reentries, re.ct);
                processed = true;
            }

            while (_intake.Reader.TryRead(out var entry))
            {
                Route(entry.item, 0, entry.ct);
                processed = true;
            }

            if (!processed)
            {
                // Wait for either channel to produce
                var waitIntake = _intake.Reader.WaitToReadAsync(ct).AsTask();
                var waitRecheck = _recheck.Reader.WaitToReadAsync(ct).AsTask();
                await Task.WhenAny(waitIntake, waitRecheck);
            }
        }
    }

    private void Route(IWorkItem item, int reentries, CancellationToken ct)
    {
        var isLead = item.Tag.Role == OperationRole.Lead;
        var isS = item is Token; // S = Storage/IO

        // Stamp: first entry transitions ADMITTED→CHECKED and locks OriginalCheckedAt.
        // Re-entry (already CHECKED): just update CheckedAt for the fresh comparand stamp.
        if (item.State == WorkState.Admitted)
        {
            switch (item)
            {
                case Token t: t.Check(); break;
                case Ticket k: k.Check(); break;
            }

            if (isS && isLead)
                Interlocked.Increment(ref _pendingS); // count S leads entering staging
        }
        else // WorkState.CHECKED
        {
            switch (item)
            {
                case Token t: t.Restamp(); break;
                case Ticket k: k.Restamp(); break;
            }
        }

        // Children: no comparand, pass directly
        if (!isLead)
        {
            _ready.Writer.TryWrite((item, ct));
            return;
        }

        // S leads: preferred type, pass immediately
        if (isS)
        {
            _ready.Writer.TryWrite((item, ct));
            return;
        }

        // C leads: defer while S tasks are pending (unless starved out)
        var origin = item switch { Token t => t.OriginalCheckedAt, Ticket k => k.OriginalCheckedAt, _ => null };
        var starvedOut = origin.HasValue &&
                         (DateTime.UtcNow - origin.Value).TotalMilliseconds > StarveAfterMs;

        var pending = Volatile.Read(ref _pendingS);
        if (pending > 0 && !starvedOut)
            _recheck.Writer.TryWrite((item, ct, reentries + 1)); // defer
        else
            _ready.Writer.TryWrite((item, ct)); // pass
    }

    // ── Worker loop ───────────────────────────────────────────────────────────

    private async Task WorkerLoop(int workerId, CancellationToken ct)
    {
        await foreach (var (item, itemCt) in _ready.Reader.ReadAllAsync(ct))
        {
            var isLead = item.Tag.Role == OperationRole.Lead;
            var isS = item is Token;
            
            if (isLead && !isS)
            {
                var lastTicks = Interlocked.Read(ref _lastLeadStartedTicks);
                if (lastTicks > 0)
                {
                    var msSinceLast = (DateTime.UtcNow - new DateTime(lastTicks, DateTimeKind.Utc)).TotalMilliseconds;
                    if (msSinceLast < MinDeltaMs)
                    {
                        _recheck.Writer.TryWrite((item, itemCt, 0));
                        continue;
                    }
                }
            }

            // Record the STARTED timestamp for future guard checks
            if (isLead)
                Interlocked.Exchange(ref _lastLeadStartedTicks, DateTime.UtcNow.Ticks);

            // S lead passing the guard: signals its slot is consumed
            if (isLead && isS)
                Interlocked.Decrement(ref _pendingS);

            // Execute
            Interlocked.Increment(ref _inflight);
            if (isLead)
            {
                Interlocked.Decrement(ref _reservedLeads);
                Interlocked.Increment(ref _inflightLeads);
            }

            try
            {
                await item.ExecuteAsync(itemCt);
            }
            finally
            {
                if (isLead)
                {
                    Interlocked.Decrement(ref _inflightLeads);
                    leadCapacity.Release();
                }

                Interlocked.Decrement(ref _inflight);
            }
        }
    }
}