AyCode.Core/AyCode.Services/SignalRs/AcBinaryDeserExecutor.cs

using System.Collections.Concurrent;

namespace AyCode.Services.SignalRs;

/// <summary>
/// Dedicated off-pool executor for the AsyncSegment streaming deserialization (see
/// <c>SIGNALR_BINARY_PROTOCOL_ISSUES.md#accore-sbp-i-t3w9</c>).
///
/// <para>The streaming deser blocks in <c>AsyncPipeReaderInput.TryAdvanceSegment</c> →
/// <c>ManualResetEventSlim.Wait()</c> between network-delivered chunks. Running that blocking
/// wait on a .NET thread-pool thread (plain <c>Task.Run</c>) starves the pool: the signal that
/// unblocks it (the producer <c>Feed()</c>+<c>Set()</c>) runs on the SignalR receive-continuation,
/// which ALSO needs a pool thread → circular dependency → deadlock under concurrent load + real
/// network latency.</para>
///
/// <para><b>Design — never-queue, grow-on-demand, reuse-forever:</b> every <see cref="Run"/>
/// immediately hands the work to an idle worker (reuse) or spawns a new dedicated background
/// thread (grow). It <b>never queues</b>. Consequences:</para>
/// <list type="bullet">
///   <item>The deser always starts immediately on its own off-pool thread → the <c>[202]</c>
///   <c>GetResult()</c> always waits on a <i>running</i> deser (no queue-coupling).</item>
///   <item>The blocking <c>Wait()</c> never consumes a .NET thread-pool thread → the producer's
///   receive-continuations always get pool threads → no circular dependency, no starvation.</item>
///   <item>Threads are reused (no per-message churn). Thread count is bounded in practice by peak
///   concurrent chunked receives — one in-flight deser per connection (the <c>[202]</c>
///   <c>GetResult</c> drains it before the next message on that connection).</item>
/// </list>
///
/// <para>This is <c>TaskCreationOptions.LongRunning</c>'s unconditional safety (never-queue,
/// off-pool) plus thread reuse. Workers live forever — no idle-timeout, hence no idle-vs-assign
/// race (the simplest provably-correct variant). An idle-timeout reclaim can be added later if the
/// held-thread memory at peak concurrency becomes a concern; it must use a CAS idle→dead transition.</para>
///
/// <para><b>Rejected:</b> a bounded-queue executor reintroduces the <c>GetResult</c>-on-queued-deser
/// coupling and drops requests on overload.</para>
/// </summary>
internal static class AcBinaryDeserExecutor
{
    private static readonly ConcurrentStack<Worker> _idle = new();

    // Monotonic id for distinct worker-thread names — eases dump / profiler / debug correlation.
    // Incremented only on grow (new thread), never per message.
    private static int _threadId;

    /// <summary>
    /// Runs <paramref name="work"/> on a dedicated off-pool thread (reused idle worker, or a new
    /// one) — never queues. Returns a task completing with the work's result.
    /// </summary>
    public static Task<object?> Run(Func<object?> work)
    {
        var tcs = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);

        if (_idle.TryPop(out var worker))
            worker.Assign(work, tcs);          // reuse an idle worker
        else
            new Worker().Start(work, tcs);     // grow: spawn a new dedicated thread

        return tcs.Task;
    }

    private sealed class Worker
    {
        // Single-permit park signal. Released exactly once per park (Assign), so the count
        // never exceeds 1 → no SemaphoreFullException. Release/Wait provide the memory barrier
        // that publishes the _work/_tcs writes to the worker thread.
        private readonly SemaphoreSlim _signal = new(0, 1);
        private Func<object?>? _work;
        private TaskCompletionSource<object?>? _tcs;

        public void Start(Func<object?> work, TaskCompletionSource<object?> tcs)
        {
            _work = work;
            _tcs = tcs;
            new Thread(Loop) { IsBackground = true, Name = $"AcBinaryDeser-{Interlocked.Increment(ref _threadId)}" }.Start();
        }

        // Called by Run only when this worker is parked in _signal.Wait() (it was popped from
        // _idle, which it entered only AFTER finishing the previous work). So _work/_tcs are
        // written here strictly while the worker is idle — no race with the worker's active read.
        public void Assign(Func<object?> work, TaskCompletionSource<object?> tcs)
        {
            _work = work;
            _tcs = tcs;
            _signal.Release();
        }

        private void Loop()
        {
            while (true)
            {
                var work = _work ?? throw new InvalidOperationException("AcBinaryDeserExecutor worker woke without work — invariant violated");
                var tcs = _tcs ?? throw new InvalidOperationException("AcBinaryDeserExecutor worker woke without tcs — invariant violated");
                _work = null;
                _tcs = null;

                try { tcs.SetResult(work()); }
                catch (Exception ex) { tcs.SetException(ex); }

                _idle.Push(this);   // become reusable
                _signal.Wait();     // park until the next Assign
            }
        }
    }
}