diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs index 7cbff85..4f6703c 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -311,6 +311,61 @@ public sealed class WorkerPipeSessionTests await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); } + /// + /// The event drain loop waits on the runtime's wake signal instead of sleeping a fixed tick, + /// so an event enqueued at an idle worker is framed as soon as it is enqueued rather than up + /// to EventDrainInterval later. The fake's wait honours only the signal here, so the + /// event reaching the pipe is proof the enqueue woke the loop — a poll-driven loop would + /// never run again, and the test would fail on its cancellation deadline instead of passing + /// on a fallback tick that happened to fire. The loop is left parked on that wait before the + /// enqueue, which also makes the recorded fallback ceiling assertable. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAsync_EventAfterIdle_DrainLoopWakesOnSignalNotOnPollTick() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(10)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new() + { + WaitForEventsOnSignalOnly = true, + }; + + // A far-off heartbeat interval keeps the drain loop the only thing that can produce a frame + // after the first beat, so nothing else can mask a drain loop that never woke. + WorkerPipeSession session = CreatePipeSession( + pipePair.WorkerStream, + runtime, + new WorkerPipeSessionOptions + { + HeartbeatInterval = TimeSpan.FromMinutes(5), + HeartbeatGrace = TimeSpan.FromSeconds(30), + }); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); + + // Park the drain loop: it drains empty once and then waits. Enqueuing before it parks would + // let the first drain pass find the event, which proves nothing about the wake. + while (runtime.LastWaitForEventsTimeout is null) + { + await Task.Delay(TimeSpan.FromMilliseconds(5), cancellation.Token); + } + + runtime.EnqueueEvent(CreateWorkerEvent(sequence: 7)); + + WorkerEnvelope workerEvent = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerEvent, + cancellation.Token); + + Assert.Equal(7UL, workerEvent.WorkerEvent.Event.WorkerSequence); + + // The 25 ms survives as the ceiling the loop passes to every wait, not as a poll period. + Assert.Equal(TimeSpan.FromMilliseconds(25), runtime.LastWaitForEventsTimeout); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + /// /// Verifies that a Ping control command is answered on the worker side /// (not dispatched to the STA) with an OK reply that echoes the ping diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs index 1492ec0..e783fde 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs @@ -1,5 +1,7 @@ using System; using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Worker.MxAccess; @@ -300,6 +302,123 @@ public sealed class MxAccessEventQueueTests Assert.Equal(WorkerFaultCategory.MxaccessEventConversionFailed, queue.Fault?.Category); } + // Wake-signal timings. The fallback is far longer than the patience deliberately: every wait + // below asserts "the signal completed this", which is only a faithful claim while the fallback + // timeout cannot have completed it within the patience window. The patience itself is generous + // so a loaded CI box cannot fail a test that is not about latency. + private static readonly TimeSpan WakeFallback = TimeSpan.FromSeconds(30); + private static readonly TimeSpan WakePatience = TimeSpan.FromSeconds(5); + + /// + /// Verifies the queue wakes a parked waiter as soon as an event is enqueued, rather than + /// leaving it to time out. This is what removes the drain loop's latency floor: before the + /// signal existed, an event arriving at an idle queue waited out the loop's whole poll tick. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WaitForEventsAsync_EnqueueAfterIdle_CompletesWithoutWaitingTheFallback() + { + MxAccessEventQueue queue = new(capacity: 4); + + Task wait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None); + Assert.False(wait.IsCompleted); + + queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10)); + + Assert.Same(wait, await Task.WhenAny(wait, Task.Delay(WakePatience))); + await wait; + Assert.Equal(1, queue.Count); + } + + /// + /// Verifies a fault recorded while the waiter is parked wakes it too. The drain loop + /// discovers faults by calling DrainFault() at the top of each pass, so without this + /// signal an overflow or conversion fault would not be reported until the loop's fallback + /// tick expired. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WaitForEventsAsync_RecordFaultWhileParked_WakesTheWaiter() + { + MxAccessEventQueue queue = new(capacity: 4); + + Task wait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None); + Assert.False(wait.IsCompleted); + + queue.RecordFault(new WorkerFault + { + Category = WorkerFaultCategory.MxaccessEventConversionFailed, + }); + + Assert.Same(wait, await Task.WhenAny(wait, Task.Delay(WakePatience))); + await wait; + Assert.NotNull(queue.DrainFault()); + } + + /// + /// Verifies the one-permit cap loses no wakeups. A burst that lands while nobody is waiting + /// leaves exactly one pending wake — the waiter that consumes it drains the whole burst, so + /// coalescing costs nothing — and that consumed wake is not replayed: the next wait parks + /// until a new enqueue signals it. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WaitForEventsAsync_BurstWhileNoWaiter_CoalescesToOneWakeThatLosesNothing() + { + MxAccessEventQueue queue = new(capacity: 16); + for (int itemHandle = 0; itemHandle < 5; itemHandle++) + { + queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle)); + } + + Task firstWait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None); + Assert.Same(firstWait, await Task.WhenAny(firstWait, Task.Delay(WakePatience))); + await firstWait; + + // One wake, the whole burst: the waiter re-drains everything queued, which is why capping + // the signal at a single permit cannot drop an event. + Assert.Equal(5, queue.Drain(maxEvents: 0).Count); + + Task secondWait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None); + Assert.False(secondWait.IsCompleted); + + queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 99)); + + Assert.Same(secondWait, await Task.WhenAny(secondWait, Task.Delay(WakePatience))); + await secondWait; + } + + /// + /// Verifies the timeout still bounds an unsignalled wait: the fallback survives as a ceiling + /// on how long a caller may sleep, so a state change reached by some future path that does + /// not signal is still observed on the next pass rather than never. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WaitForEventsAsync_WithNoSignal_CompletesAtTheFallbackTimeout() + { + MxAccessEventQueue queue = new(capacity: 4); + + Task wait = queue.WaitForEventsAsync(TimeSpan.FromMilliseconds(25), CancellationToken.None); + + Assert.Same(wait, await Task.WhenAny(wait, Task.Delay(WakePatience))); + await wait; + } + + /// Verifies a cancelled wait unwinds instead of hanging until the fallback expires. + /// A task that represents the asynchronous operation. + [Fact] + public async Task WaitForEventsAsync_WhenCancelled_Throws() + { + MxAccessEventQueue queue = new(capacity: 4); + using CancellationTokenSource cancellation = new(); + + Task wait = queue.WaitForEventsAsync(WakeFallback, cancellation.Token); + cancellation.Cancel(); + + await Assert.ThrowsAnyAsync(async () => await wait); + } + // Mirrors MxAccessEventQueue's per-event repeated-field allowance. Kept local rather than made // public on the queue: the byte-budget tests state their budgets in units of that charge, so a // change to it should surface here as a failing bound instead of silently moving with the code. diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs index a992b12..84edc37 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs @@ -24,6 +24,13 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession private readonly object gate = new(); private readonly Queue events = new(); private readonly List cancelledCorrelationIds = new(); + + // Mirrors MxAccessEventQueue's coalesced wake signal so the drain loop under test is driven the + // same way it is in production: EnqueueEvent(s) releases one permit, WaitForEventsAsync consumes + // it. Never disposed — the drain loop can still be parked on it while Dispose runs, and a + // disposed SemaphoreSlim would turn that shutdown into an ObjectDisposedException. + private readonly SemaphoreSlim eventSignal = new(0, 1); + private TimeSpan? lastWaitForEventsTimeout; private WorkerRuntimeHeartbeatSnapshot snapshot = new( DateTimeOffset.UtcNow, pendingCommandCount: 0, @@ -263,6 +270,52 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession } } + /// + /// When set, honours only the wake signal and cancellation, + /// never the fallback timeout. A drain loop that ships an event while this is set can only + /// have been woken by the enqueue signal, which is what makes "the drain is signal-driven, + /// not poll-driven" assertable without racing the 25 ms fallback tick. + /// + public bool WaitForEventsOnSignalOnly { get; set; } + + /// + /// The timeout argument of the most recent call, so + /// a test can assert the drain loop still passes its fallback ceiling. + /// + public TimeSpan? LastWaitForEventsTimeout + { + get + { + lock (gate) + { + return lastWaitForEventsTimeout; + } + } + } + + /// + public Task WaitForEventsAsync(TimeSpan timeout, CancellationToken cancellationToken) + { + lock (gate) + { + lastWaitForEventsTimeout = timeout; + } + + if (BackingQueue is not null) + { + // Tests that drive a real queue enqueue into it directly, so the real queue owns the + // wake signal too. + return BackingQueue.WaitForEventsAsync(timeout, cancellationToken); + } + + if (WaitForEventsOnSignalOnly) + { + return eventSignal.WaitAsync(cancellationToken); + } + + return eventSignal.WaitAsync(timeout, cancellationToken); + } + /// public WorkerFault? DrainFault() { @@ -370,6 +423,8 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession { events.Enqueue(workerEvent); } + + SignalWake(); } /// @@ -387,6 +442,26 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession events.Enqueue(workerEvent); } } + + SignalWake(); + } + + // Coalesced wake, released outside the gate exactly as MxAccessEventQueue does. + private void SignalWake() + { + if (eventSignal.CurrentCount > 0) + { + return; + } + + try + { + eventSignal.Release(); + } + catch (SemaphoreFullException) + { + // A concurrent enqueue already published the pending wake this call wanted. + } } /// diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs index e91442a..8934946 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs @@ -15,6 +15,11 @@ namespace ZB.MOM.WW.MxGateway.Worker.Ipc; public sealed class WorkerPipeSession { + // Fallback ceiling for the event drain loop's wait — not a poll period. MxAccessEventQueue + // signals on enqueue (and on a recorded fault), so the loop wakes as soon as there is something + // to ship instead of paying up to this interval of latency on every burst from idle, and an + // idle worker parks instead of waking 40x/s. The interval survives only as the bound on how + // long the loop may sleep unsignalled, which keeps its DrainFault() poll on a known cadence. private static readonly TimeSpan EventDrainInterval = TimeSpan.FromMilliseconds(25); private static readonly TimeSpan BackgroundTaskStopTimeout = TimeSpan.FromSeconds(1); private const uint EventDrainBatchSize = 128; @@ -367,7 +372,15 @@ public sealed class WorkerPipeSession IReadOnlyList events = runtimeSession.DrainEvents(EventDrainBatchSize); if (events.Count == 0) { - await Task.Delay(EventDrainInterval, cancellationToken).ConfigureAwait(false); + // Wait on the queue's wake signal rather than sleeping a fixed tick: an event + // enqueued by the STA completes this immediately, so the first event of a burst is + // framed at signal latency instead of waiting out EventDrainInterval, and a session + // with no traffic stops waking at all. The wait's outcome is intentionally ignored — + // whether a signal or the fallback ended it, the next pass re-checks DrainFault() + // and re-drains, which is also why one coalesced wake for many enqueues is safe. + await runtimeSession + .WaitForEventsAsync(EventDrainInterval, cancellationToken) + .ConfigureAwait(false); continue; } diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWorkerRuntimeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWorkerRuntimeSession.cs index bf1f252..cabfd82 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWorkerRuntimeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWorkerRuntimeSession.cs @@ -57,6 +57,23 @@ public interface IWorkerRuntimeSession : IDisposable /// The drained events and the truncation facts describing what stayed queued. WorkerEventDrainResult DrainEvents(uint maxEvents, int maxTotalBytes); + /// + /// Waits until the outbound event queue has something for the caller to look at (an + /// enqueued event or a recorded fault), the fallback timeout elapses, or the token is + /// cancelled. Lets a drain loop be signal-driven instead of polling. + /// + /// + /// Declared on the interface because the pipe session only ever sees an + /// , never the queue behind it; .NET Framework 4.8 has + /// no default interface members, so every implementation supplies it. The wait's outcome is + /// not surfaced: the caller re-drains and re-checks after every + /// wait, because the signal is coalesced and the timeout is a ceiling, not a poll period. + /// + /// Maximum time to wait before the wait completes unsignalled. + /// Token cancelling the wait. + /// A task that completes when the queue is signalled or the timeout elapses. + Task WaitForEventsAsync(TimeSpan timeout, CancellationToken cancellationToken); + /// /// Drains a pending fault from the queue, if any. /// diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs index ed27a67..17f1d67 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs @@ -1,5 +1,7 @@ using System; using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; using Google.Protobuf.WellKnownTypes; using ZB.MOM.WW.MxGateway.Contracts.Proto; @@ -38,6 +40,17 @@ public sealed class MxAccessEventQueue private readonly int capacity; private readonly Queue events; private readonly object syncRoot = new(); + + // Wake signal for the worker's event drain loop (see WorkerPipeSession.RunEventDrainLoopAsync). + // Capped at one permit, so it is a level ("there is something to look at"), never a count: a + // burst of enqueues coalesces into one wake and the waiter re-drains everything that arrived. + // Never released while holding syncRoot — the STA thread enqueues here, so its path must stay a + // lock acquire plus a non-blocking Release. Deliberately not disposed: the queue outlives its + // waiter, no wait handle is ever materialized (nothing touches AvailableWaitHandle), and making + // the queue IDisposable would only add a lifecycle race between the STA's Enqueue and the + // drain loop's wait. WorkerFrameWriter's _writeLock is held the same way. + private readonly SemaphoreSlim eventSignal = new SemaphoreSlim(initialCount: 0, maxCount: 1); + private ulong lastEventSequence; private WorkerFault? fault; private bool faultDrained; @@ -142,33 +155,67 @@ public sealed class MxAccessEventQueue throw new ArgumentNullException(nameof(mxEvent)); } - lock (syncRoot) + try { - if (fault is not null) + lock (syncRoot) { - throw new InvalidOperationException("MXAccess outbound event queue is faulted."); + if (fault is not null) + { + throw new InvalidOperationException("MXAccess outbound event queue is faulted."); + } + + if (events.Count >= capacity) + { + fault = CreateOverflowFault(); + throw new MxAccessEventQueueOverflowException(capacity); + } + + // WRK-11: stamp the sequence/timestamp on the caller's own event and + // enqueue that same instance under the lock instead of cloning. See + // the ownership invariant on the class summary — the caller hands the + // event over exclusively, so a defensive Clone() here is pure + // overhead on the hottest path. + mxEvent.WorkerSequence = ++lastEventSequence; + mxEvent.WorkerTimestamp = Timestamp.FromDateTime(DateTime.UtcNow); + + WorkerEvent workerEvent = new() + { + Event = mxEvent, + }; + events.Enqueue(workerEvent); } - - if (events.Count >= capacity) - { - fault = CreateOverflowFault(); - throw new MxAccessEventQueueOverflowException(capacity); - } - - // WRK-11: stamp the sequence/timestamp on the caller's own event and - // enqueue that same instance under the lock instead of cloning. See - // the ownership invariant on the class summary — the caller hands the - // event over exclusively, so a defensive Clone() here is pure - // overhead on the hottest path. - mxEvent.WorkerSequence = ++lastEventSequence; - mxEvent.WorkerTimestamp = Timestamp.FromDateTime(DateTime.UtcNow); - - WorkerEvent workerEvent = new() - { - Event = mxEvent, - }; - events.Enqueue(workerEvent); } + finally + { + // Wake the drain loop outside the lock, on every exit path: the success path has an + // event to ship, and the overflow path has just self-recorded the fault the loop + // reports through DrainFault(). Signalling in a finally also keeps the already-faulted + // rejection harmless — a spurious wake only costs the waiter one empty re-drain. + SignalWake(); + } + } + + /// + /// Waits for a wake signal, the fallback timeout, or cancellation, so the worker's event + /// drain loop is signal-driven instead of polling. and + /// both signal, so an event arriving at an idle queue is picked + /// up immediately rather than at the caller's next tick, and an idle queue costs no wakeups + /// at all. + /// + /// + /// The wait's outcome is deliberately not surfaced: the caller must re-drain (and re-check + /// ) after every wait regardless of whether a signal or the timeout + /// ended it, because the signal is coalesced — one pending wake can stand for any number of + /// enqueues. The timeout is therefore a ceiling on how long the caller may sleep, not a + /// poll period: it bounds how late a state change reaches the caller if some future path + /// mutates the queue without signalling. + /// + /// Maximum time to wait before the wait completes unsignalled. + /// Token cancelling the wait. + /// A task that completes when the queue is signalled or the timeout elapses. + public Task WaitForEventsAsync(TimeSpan timeout, CancellationToken cancellationToken) + { + return eventSignal.WaitAsync(timeout, cancellationToken); } /// @@ -291,6 +338,11 @@ public sealed class MxAccessEventQueue { fault ??= workerFault.Clone(); } + + // The drain loop discovers faults by polling DrainFault() at the top of each pass, so a + // fault recorded while it is parked must wake it — otherwise the fault would be reported + // no sooner than the loop's fallback tick. Signalled outside the lock, as Enqueue does. + SignalWake(); } /// @@ -311,6 +363,28 @@ public sealed class MxAccessEventQueue } } + private void SignalWake() + { + // Fast path: a wake is already pending and no waiter has consumed it yet, so releasing + // again would only throw. This keeps a burst of enqueues on the STA at a single volatile + // read once the first event has signalled. + if (eventSignal.CurrentCount > 0) + { + return; + } + + try + { + eventSignal.Release(); + } + catch (SemaphoreFullException) + { + // A concurrent producer won the race between the check above and this Release. The + // pending wake it published is exactly the wake this call wanted, so there is nothing + // to do — the signal is a level, not a count. + } + } + private WorkerFault CreateOverflowFault() { string message = $"MXAccess outbound event queue reached capacity {capacity}."; diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs index 90e622f..03ddf16 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs @@ -455,6 +455,12 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession return eventQueue.Drain(maxEvents, maxTotalBytes); } + /// + public Task WaitForEventsAsync(TimeSpan timeout, CancellationToken cancellationToken) + { + return eventQueue.WaitForEventsAsync(timeout, cancellationToken); + } + /// public WorkerFault? DrainFault() {