using System.IO.Pipelines; using AyCode.Services.SignalRs; using Microsoft.AspNetCore.SignalR.Protocol; namespace AyCode.Services.Server.Tests.SignalRs; /// /// Thread-pool starvation regression tests for the AsyncSegment paths /// (ACCORE-SBP-I-T3W9: blocking waits on .NET thread-pool threads deadlock the pool under /// concurrent load + real network latency — production-only, "stuck loading → 60 s timeout"). /// /// Deser (receive) side — fixed. The streaming deser blocks in /// ManualResetEventSlim.Wait() between network-delivered chunks. It now runs off the .NET /// thread pool on a never-queue + reuse dedicated executor () so /// the blocking wait never consumes a pool thread (which the producer's receive-continuations need /// → circular dependency). These tests assert that executor contract directly and should all pass. /// /// Ser (send) side — TBD. The per-chunk SyncFlush in WriteMessageChunked /// blocks a dispatch (pool) thread under client backpressure. The send-side test below asserts the /// DESIRED non-blocking behaviour and is [Ignore]d until that fix lands. /// [TestClass] public class AcBinaryProtocolThreadPoolTests { // --------------------------------------------------------------------------------------------- // Deser (receive) side — AcBinaryDeserExecutor contract. Expected: all pass. // --------------------------------------------------------------------------------------------- [TestMethod] public void DeserExecutor_RunsOffThreadPool() { var onPool = (bool)AcBinaryDeserExecutor.Run(() => (object?)Thread.CurrentThread.IsThreadPoolThread).GetAwaiter().GetResult()!; Assert.IsFalse(onPool, "Streaming deser must run OFF the .NET thread pool (dedicated thread) — the blocking " + "MRES.Wait() would otherwise pin a pool thread the producer's receive-continuation needs " + "→ circular-dependency starvation (ACCORE-SBP-I-T3W9)."); } [TestMethod] public void DeserExecutor_NeverQueues_AllConcurrentWorkStartsImmediately() { const int n = 32; using var allStarted = new CountdownEvent(n); using var release = new ManualResetEventSlim(false); var tasks = new Task[n]; for (var i = 0; i < n; i++) { tasks[i] = AcBinaryDeserExecutor.Run(() => { allStarted.Signal(); release.Wait(); return (object?)null; }); } // Every item must START even though none has finished — proves the executor NEVER queues: // each submit immediately gets a thread (reuse idle or grow). A bounded-queue executor // would stall here (excess items wait behind busy workers) and the countdown never reaches 0. // 5 s is generous for 32 thread spawns (~ms) yet fails fast on a queue regression. Assert.IsTrue(allStarted.Wait(TimeSpan.FromSeconds(5)), $"Only {n - allStarted.CurrentCount}/{n} work items started — executor queued instead of " + "growing on demand (would reintroduce the [202] GetResult-on-queued-deser coupling)."); release.Set(); Task.WaitAll(tasks); } [TestMethod] public void DeserExecutor_ReusesThreads_AcrossSequentialWork() { var ids = new HashSet(); for (var i = 0; i < 50; i++) { var id = (int)AcBinaryDeserExecutor.Run(() => (object?)Environment.CurrentManagedThreadId).GetAwaiter().GetResult()!; ids.Add(id); } // Sequential work: each completes before the next is submitted → the same idle worker is // reused (no per-message thread churn). The worker pushes itself to idle right after // SetResult (1 instruction) while the test thread runs ~10+ (incl. a TCS alloc) before the // next TryPop, so the worker almost always wins the race → 1-2 distinct threads in practice. // <= 4 leaves a small margin for the rare preempt-in-window case; more than that is suspect. Assert.IsTrue(ids.Count <= 4, $"Expected thread reuse across sequential work, got {ids.Count} distinct threads / 50 — churn instead of reuse."); } [TestMethod] [DoNotParallelize] // stresses the shared process-wide thread pool public void DeserExecutor_CompletesUnderThreadPoolPressure() { // Occupy many .NET thread-pool threads with blocking work, then run a deser-style item. // Because the executor is off-pool it completes regardless of pool pressure — whereas the // old `Task.Run` deser would queue behind the busy pool and the [202] GetResult would block // (ACCORE-SBP-I-T3W9). This is the end-to-end resilience guard. using var release = new ManualResetEventSlim(false); var hogCount = Environment.ProcessorCount * 4; var hogs = new Task[hogCount]; for (var i = 0; i < hogCount; i++) hogs[i] = Task.Run(() => release.Wait()); try { var done = AcBinaryDeserExecutor.Run(() => (object?)42); Assert.IsTrue(done.Wait(TimeSpan.FromSeconds(5)), "Off-pool deser did not complete under thread-pool pressure — possible pool coupling."); Assert.AreEqual(42, (int)done.Result!); } finally { release.Set(); Task.WaitAll(hogs); } } // --------------------------------------------------------------------------------------------- // Ser (send) side — known-open. Asserts DESIRED behaviour; [Ignore]d until the send-side fix. // --------------------------------------------------------------------------------------------- [TestMethod] [Ignore("Send-side fix TBD — ACCORE-SBP-I-T3W9. Asserts the DESIRED non-blocking behaviour; " + "today WriteMessageChunked blocks the calling (dispatch) thread under client backpressure " + "via the per-chunk SyncFlush. Un-ignore when the send-side fix lands.")] public void WriteMessageChunked_UnderBackpressure_DoesNotBlockCallingThread() { // A Pipe with a small pause threshold and NO reader → FlushAsync backpressures immediately // (simulates a stuck / slow mobile client). The chunked send's per-chunk SyncFlush then // blocks the calling thread. Desired post-fix behaviour: the send does not pin the caller. var pipe = new Pipe(new PipeOptions(pauseWriterThreshold: 4096, resumeWriterThreshold: 2048)); var opts = new AcBinaryHubProtocolOptions { ProtocolMode = BinaryProtocolMode.AsyncSegment, FlushTimeout = TimeSpan.FromSeconds(2), // bound so a stuck flush can't hang the test forever }; var protocol = new AyCodeBinaryHubProtocol(opts); // Large non-byte[] data arg → the chunked path engages and produces > pauseThreshold bytes. // The 4-arg shape mirrors the OnReceiveMessage(tag, requestId, SignalParams, data) convention // the other SignalR tests use — SignalParams here is convention/realism, not functionally // required (the large `bigData` arg alone triggers HasStreamableArgs → chunked send). var bigData = new string('x', 200_000); var msg = new InvocationMessage("OnReceiveMessage", new object?[] { 1, (int?)1, new SignalParams(), bigData }); var writeThread = new Thread(() => { try { protocol.WriteMessage(msg, pipe.Writer); } catch { /* TimeoutException on the bounded flush is acceptable for this probe */ } }) { IsBackground = true }; writeThread.Start(); // 2 s margin so a non-blocking send (returns in ms once it no longer waits on the flush) is // not falsely flagged on a slow CI; a blocking send stays pinned well past this. var finishedPromptly = writeThread.Join(TimeSpan.FromSeconds(2)); Assert.IsTrue(finishedPromptly, "WriteMessageChunked blocked the calling thread under client backpressure (per-chunk " + "SyncFlush) — the send-side thread-pool starvation source (ACCORE-SBP-I-T3W9)."); } }