Refactor: remove Server DLLs, unify SignalR references
- Removed all AyCode.Services.Server and Server DLL references from projects and code; now use only AyCode.Services and AyCode.Services.SignalRs. - Updated all documentation links to reference AyCode.Services/docs/SIGNALR_DATASOURCE/README.md. - Enabled AOT compilation in FruitBankHybrid.Web.Client. - Added InternalsVisibleTo for AyCode.Services.Server.Tests to support internal testing. - Introduced AcBinaryProtocolThreadPoolTests for thread-pool regression and contract testing. - Added P2 feature TODO for per-message protocol mode selection in AcBinaryHubProtocol. - Cleaned up using directives and removed obsolete Server-specific references. - Minor formatting and whitespace improvements.
This commit is contained in:
parent
bb1dbd242e
commit
a8e90e5b59
|
|
@ -0,0 +1,169 @@
|
|||
using System.IO.Pipelines;
|
||||
using AyCode.Services.SignalRs;
|
||||
using Microsoft.AspNetCore.SignalR.Protocol;
|
||||
|
||||
namespace AyCode.Services.Server.Tests.SignalRs;
|
||||
|
||||
/// <summary>
|
||||
/// Thread-pool starvation regression tests for the AsyncSegment paths
|
||||
/// (<c>ACCORE-SBP-I-T3W9</c>: blocking waits on .NET thread-pool threads deadlock the pool under
|
||||
/// concurrent load + real network latency — production-only, "stuck loading → 60 s timeout").
|
||||
///
|
||||
/// <para><b>Deser (receive) side — fixed.</b> The streaming deser blocks in
|
||||
/// <c>ManualResetEventSlim.Wait()</c> between network-delivered chunks. It now runs off the .NET
|
||||
/// thread pool on a never-queue + reuse dedicated executor (<see cref="AcBinaryDeserExecutor"/>) 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.</para>
|
||||
///
|
||||
/// <para><b>Ser (send) side — TBD.</b> The per-chunk <c>SyncFlush</c> in <c>WriteMessageChunked</c>
|
||||
/// blocks a dispatch (pool) thread under client backpressure. The send-side test below asserts the
|
||||
/// DESIRED non-blocking behaviour and is <c>[Ignore]</c>d until that fix lands.</para>
|
||||
/// </summary>
|
||||
[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<object?>[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<int>();
|
||||
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).");
|
||||
}
|
||||
}
|
||||
|
|
@ -24,6 +24,12 @@
|
|||
<ProjectReference Include="..\AyCode.Models\AyCode.Models.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<!-- Lets the test project drive the internal AcBinaryDeserExecutor directly
|
||||
(deser-side thread-pool regression tests for ACCORE-SBP-I-T3W9). -->
|
||||
<InternalsVisibleTo Include="AyCode.Services.Server.Tests" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<None Include="docs\**\*.md" />
|
||||
<None Include="**\README.md" Exclude="$(DefaultItemExcludes);docs\**" />
|
||||
|
|
|
|||
|
|
@ -5,6 +5,21 @@
|
|||
|
||||
---
|
||||
|
||||
## ACCORE-SBP-T-Y2R8: Per-message protocol mode selection
|
||||
**Priority:** P2 · **Type:** Feature
|
||||
|
||||
**Problem:** Global `BinaryProtocolMode` config (program.cs) forces all messages through the same path — either Bytes (RAM spike on large payloads, no streaming overlap) or AsyncSegment (dedicated deser thread overhead on small frequent messages). Mixed workloads (small logs/telemetry + large DataSource/exports) cannot optimize per use-case.
|
||||
|
||||
**Solution:** Per-message mode override via invocation headers (`X-Binary-Protocol-Mode: Bytes|AsyncSegment`). Server `AcBinaryHubProtocol.SelectProtocolMode` checks header → explicit mode; fallback → `DefaultProtocolMode`. Client fluent API: `InvokeAsync(..., protocolMode: BinaryProtocolMode.Bytes)`.
|
||||
|
||||
**Use-cases:**
|
||||
- Small frequent (logs, telemetry, CRUD < 4KB): `Bytes` → minimal overhead, inline deser on pool thread
|
||||
- Large payloads (DataSource 10MB, file exports): `AsyncSegment` → streaming overlap + bounded RAM + dedicated deser thread
|
||||
|
||||
**Config:** `AcBinaryHubProtocolOptions.AllowPerMessageProtocolMode` gate (default: true).
|
||||
|
||||
**Non-solution:** Auto-detect size heuristic rejected — wire-format size unknown until serialize (dynamic collections, string interning, ID tracking).
|
||||
|
||||
## ACCORE-SBP-T-P8X9: ~~SegmentBufferReader isolated unit tests~~
|
||||
**Status:** Closed (2026-05-03) — obsoleted by `ACCORE-SBP-T-G7T2` · **Priority:** ~~P1~~ · **Type:** ~~Test coverage~~
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue