diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index c043110..a776161 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -295,6 +295,14 @@ cache lock) so a cross-thread producer wakes the waiter immediately; in the live worker the update arrives on the STA from inside `pumpStep` itself, and the message wake is what carries it. +`MWMO_INPUTAVAILABLE` makes the drain contract load-bearing: the wait wakes on +input that is merely *present*, including input an earlier `PeekMessage` saw but +did not remove. A `pumpStep` that drains only part of the queue — or a no-op one +— therefore leaves a message that satisfies the wake condition forever, and the +loop spins at 100% CPU until its deadline (deadline and reply shape still hold; +it is a CPU fault, not a correctness one). Every `pumpStep` must drain to empty, +as `StaRuntime.PumpPendingMessages` does. + The wait slice is capped at 50 ms so `pumpStep` runs periodically even when nothing wakes the wait — a process with no STA message queue (unit tests drive these caches from ordinary threads, standing in for the STA by updating the diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs index 6b2ef48..bad8580 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs @@ -179,27 +179,35 @@ public sealed class MxAccessValueCacheTests } /// - /// Verifies the wait is message/signal-driven rather than sleep-polled: - /// with a poll interval far longer than the deadline, a value set from - /// another thread still wakes the wait and is returned well inside the - /// deadline. The sleep-polled loop this replaced would have sat blind for - /// the whole 10 s interval — on the STA that means 10 s with no Windows + /// Verifies the wait slice stays bounded no matter what poll interval the + /// caller asks for: with a 10 s interval and a 3 s deadline, a value set + /// 50 ms in is still returned in time, because every slice is capped at + /// the 50 ms fallback tick. The blind Thread.Sleep this replaced + /// would have slept the full interval — on the STA, 10 s with no Windows /// messages dispatched, so the OnDataChange being waited for could not /// have arrived at all — and then reported a timeout. /// + /// + /// Scope: this proves the cadence bound, not the signal path. The clamp + /// alone would satisfy it, so it stays green if the Set-side + /// signal is deleted. StaWaitHelperTests owns the signal path, + /// where a five-second wait with no clamp in play makes the handle the + /// only thing that can end it early. + /// + /// A task that represents the asynchronous operation. [Fact] - public void TryWaitForUpdate_WakesOnCrossThreadSet_DespiteLongPollInterval() + public async Task TryWaitForUpdate_CompletesWithinDeadline_WhenPollIntervalExceedsTheFallbackTick() { MxAccessValueCache cache = new(); Timestamp sourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow); using ManualResetEventSlim waitEntered = new(false); - Task setter = Task.Run(() => + Task setter = Task.Run(async () => { // Handshake so the value cannot land before the wait starts — // otherwise the first check would satisfy it and prove nothing. waitEntered.Wait(TimeSpan.FromSeconds(5)); - Thread.Sleep(50); + await Task.Delay(50, CancellationToken.None); cache.Set(7, 21, BuildEvent(7, 21, intValue: 8080, quality: 192, sourceTimestamp)); }); @@ -213,13 +221,13 @@ public sealed class MxAccessValueCacheTests out MxAccessValueCache.CachedValue value, pollIntervalMs: 10_000); elapsed.Stop(); - setter.Wait(TimeSpan.FromSeconds(5)); + await setter; Assert.True(found); Assert.Equal(8080, value.Value.Int32Value); Assert.True( elapsed.Elapsed < TimeSpan.FromSeconds(2), - $"The wait should have woken on the cached value, not slept out the poll interval; took {elapsed.ElapsedMilliseconds} ms."); + $"The wait should have re-checked within the fallback tick, not slept out the poll interval; took {elapsed.ElapsedMilliseconds} ms."); } /// diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs index fd817f1..8d4b1c8 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs @@ -113,25 +113,33 @@ public sealed class MxAccessWriteCompletionCacheTests } /// - /// Verifies the wait is message/signal-driven rather than sleep-polled: - /// with a poll interval far longer than the deadline, a completion - /// recorded from another thread still wakes the wait and is returned well - /// inside the deadline. The sleep-polled loop this replaced would have sat - /// blind for the whole 10 s interval — on the STA that means 10 s with no + /// Verifies the wait slice stays bounded no matter what poll interval the + /// caller asks for: with a 10 s interval and a 3 s deadline, a completion + /// recorded 50 ms in is still returned in time, because every slice is + /// capped at the 50 ms fallback tick. The blind Thread.Sleep this + /// replaced would have slept the full interval — on the STA, 10 s with no /// Windows messages dispatched — and then reported a timeout. /// + /// + /// Scope: this proves the cadence bound, not the signal path. The clamp + /// alone would satisfy it, so it stays green if the Record-side + /// Set() is deleted. StaWaitHelperTests owns the signal + /// path, where a five-second wait with no clamp in play makes the handle + /// the only thing that can end it early. + /// + /// A task that represents the asynchronous operation. [Fact] - public void TryWaitForCompletion_WakesOnCrossThreadRecord_DespiteLongPollInterval() + public async Task TryWaitForCompletion_CompletesWithinDeadline_WhenPollIntervalExceedsTheFallbackTick() { MxAccessWriteCompletionCache cache = new(); using ManualResetEventSlim waitEntered = new(false); - Task recorder = Task.Run(() => + Task recorder = Task.Run(async () => { // Handshake so the completion cannot land before the wait starts — // otherwise the first check would satisfy it and prove nothing. waitEntered.Wait(TimeSpan.FromSeconds(5)); - Thread.Sleep(50); + await Task.Delay(50, CancellationToken.None); cache.Record(7, 21, BuildStatuses(detail: 8080)); }); @@ -145,13 +153,13 @@ public sealed class MxAccessWriteCompletionCacheTests out RepeatedField statuses, pollIntervalMs: 10_000); elapsed.Stop(); - recorder.Wait(TimeSpan.FromSeconds(5)); + await recorder; Assert.True(found); Assert.Equal(8080, Assert.Single(statuses).Detail); Assert.True( elapsed.Elapsed < TimeSpan.FromSeconds(2), - $"The wait should have woken on the recorded completion, not slept out the poll interval; took {elapsed.ElapsedMilliseconds} ms."); + $"The wait should have re-checked within the fallback tick, not slept out the poll interval; took {elapsed.ElapsedMilliseconds} ms."); } /// Verifies that Record stores an independent clone of the caller's rows. diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaWaitHelperTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaWaitHelperTests.cs new file mode 100644 index 0000000..925f86c --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Sta/StaWaitHelperTests.cs @@ -0,0 +1,235 @@ +using System; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using ZB.MOM.WW.MxGateway.Worker.Sta; + +namespace ZB.MOM.WW.MxGateway.Worker.Tests.Sta; + +/// +/// Tests for , the message-aware wait behind the +/// write-completion and ReadBulk value waits. +/// +/// +/// +/// These exist because the cache-level tests cannot guard the signal +/// path. Every cache wait clamps its slice to +/// , so a wait +/// woken by the handle and a wait that merely timed out and re-checked +/// are milliseconds apart — no wall-clock assertion up there can tell +/// them apart, and deleting the caches' Set() calls would leave +/// them green. Calling the helper directly with a five-second timeout +/// removes the clamp from the picture: only the handle can end that wait +/// early, and the test asserts the handle was the thing consumed. +/// +/// +/// Boundary: the WAIT_FAILED branch is not exercised, for the same +/// reason gives — forcing +/// MsgWaitForMultipleObjectsEx to fail means handing it a +/// deliberately invalid native handle, which is unsafe to construct in a +/// managed test. +/// +/// +public sealed class StaWaitHelperTests +{ + /// Verifies that a null wait handle is rejected. + [Fact] + public void WaitForSignalOrMessages_NullSignal_ThrowsArgumentNullException() + { + ArgumentNullException exception = Assert.Throws( + () => StaWaitHelper.WaitForSignalOrMessages(null!, 10)); + + Assert.Equal("signal", exception.ParamName); + } + + /// + /// Verifies the wait ends on a cross-thread signal rather than on its + /// timeout: the handle is set ~50 ms into a five-second wait, and the + /// wait must return two orders of magnitude before that timeout. The + /// post-condition that the handle was consumed is what makes this a + /// discriminator — an auto-reset event that our wait did not take would + /// still be signalled afterwards, which would mean something other than + /// the signal ended the wait. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WaitForSignalOrMessages_SignalSetDuringWait_ReturnsLongBeforeTheTimeout() + { + using AutoResetEvent signal = new(initialState: false); + + // Drain anything already queued for this thread so the wait cannot be + // woken by stale input instead of by the signal. + new StaMessagePump().PumpPendingMessages(); + + Task setter = Task.Run(async () => + { + await Task.Delay(50, CancellationToken.None); + signal.Set(); + }); + + Stopwatch elapsed = Stopwatch.StartNew(); + StaWaitHelper.WaitForSignalOrMessages(signal, 5000); + elapsed.Stop(); + await setter; + + Assert.True( + elapsed.Elapsed < TimeSpan.FromMilliseconds(500), + $"Wait took {elapsed.ElapsedMilliseconds} ms of its 5000 ms timeout; the signal should have ended it."); + Assert.False( + signal.WaitOne(0), + "The wait did not consume the signal, so something other than the signal ended it."); + } + + /// + /// Verifies an already-signalled handle ends the wait immediately, which + /// is the case that keeps a completion recorded between the caller's + /// check and its wait from being missed. + /// + [Fact] + public void WaitForSignalOrMessages_PreSignalledHandle_ReturnsImmediately() + { + using AutoResetEvent signal = new(initialState: true); + + Stopwatch elapsed = Stopwatch.StartNew(); + StaWaitHelper.WaitForSignalOrMessages(signal, 30_000); + elapsed.Stop(); + + Assert.True( + elapsed.Elapsed < TimeSpan.FromSeconds(5), + $"Wait took {elapsed.ElapsedMilliseconds} ms; a pre-signalled handle must return at once."); + Assert.False(signal.WaitOne(0), "The wait should have consumed the signal."); + } + + /// + /// Verifies the wait actually blocks for its timeout when nothing signals + /// it. The lower bound is the point: a wait that returned instantly would + /// turn every caller's poll loop into a spin, which is exactly the + /// failure mode the queue must be drained to avoid. + /// + [Fact] + public void WaitForSignalOrMessages_NeverSignalled_BlocksUntilTheTimeout() + { + using AutoResetEvent signal = new(initialState: false); + + // The wait wakes on input that is merely present, so drain first — + // otherwise a stale message would end the wait early and the lower + // bound below would be measuring the wrong thing. + new StaMessagePump().PumpPendingMessages(); + + Stopwatch elapsed = Stopwatch.StartNew(); + StaWaitHelper.WaitForSignalOrMessages(signal, 200); + elapsed.Stop(); + + // Loose lower bound: the OS may return slightly early on a coarse timer + // tick, so this proves "it blocked", not "it blocked for exactly 200 ms". + Assert.True( + elapsed.Elapsed >= TimeSpan.FromMilliseconds(100), + $"Wait returned after {elapsed.ElapsedMilliseconds} ms; a 200 ms wait must not return instantly."); + Assert.True( + elapsed.Elapsed < TimeSpan.FromSeconds(5), + $"Wait took {elapsed.ElapsedMilliseconds} ms; a 200 ms timeout must end it."); + } + + /// + /// Verifies a non-positive timeout returns without entering the wait at + /// all — proven by the signal still being set afterwards, since a wait + /// that ran would have consumed it. + /// + /// The non-positive timeout under test. + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void WaitForSignalOrMessages_NonPositiveTimeout_DoesNotEnterTheWait(int timeoutMilliseconds) + { + using AutoResetEvent signal = new(initialState: true); + + Stopwatch elapsed = Stopwatch.StartNew(); + StaWaitHelper.WaitForSignalOrMessages(signal, timeoutMilliseconds); + elapsed.Stop(); + + Assert.True( + elapsed.Elapsed < TimeSpan.FromSeconds(2), + $"Wait took {elapsed.ElapsedMilliseconds} ms; a non-positive timeout must not block."); + Assert.True(signal.WaitOne(0), "The signal must be untouched when the wait is skipped."); + } + + /// + /// Boundary table for the slice math. Every caller's wait length comes + /// from here, so each clause is pinned: a poll interval below one clamps + /// up to 1 ms, an expired deadline yields 0 (the caller returns without + /// waiting), a deadline nearer than the tick rounds up so the wait never + /// ends before the deadline, and any interval above the fallback ceiling + /// clamps down to it. + /// + /// Time left before the caller's deadline. + /// The caller's requested pump cadence. + /// The expected wait slice in milliseconds. + [Theory] + // pollInterval < 1 clamps up to 1 ms — never a zero-length spin. + [InlineData(1000d, 0, 1)] + [InlineData(1000d, -5, 1)] + // remaining <= 0 yields 0; the caller has already returned on its deadline. + [InlineData(0d, 5, 0)] + [InlineData(-10d, 5, 0)] + // remaining < tick shortens the wait to the deadline. + [InlineData(3d, 5, 3)] + [InlineData(1d, 50, 1)] + // remaining == tick is not "less than", so the tick stands. + [InlineData(5d, 5, 5)] + // The caller's interval wins while it is under the ceiling. + [InlineData(1000d, 5, 5)] + [InlineData(1000d, 49, 49)] + // Anything at or above the ceiling clamps to it — this is what bounds the + // unconditional pumpStep cadence regardless of what a caller asks for. + [InlineData(1000d, 50, 50)] + [InlineData(1000d, 10_000, 50)] + [InlineData(50d, 10_000, 50)] + public void ClampWaitMilliseconds_BoundaryTable( + double remainingMilliseconds, + int pollIntervalMilliseconds, + int expected) + { + int actual = StaWaitHelper.ClampWaitMilliseconds( + TimeSpan.FromMilliseconds(remainingMilliseconds), + pollIntervalMilliseconds); + + Assert.Equal(expected, actual); + } + + /// + /// Verifies a sub-millisecond remainder rounds up rather than down: a + /// zero-length wait would spin, and truncating would end the wait before + /// the caller's deadline. Built from ticks because + /// rounds its argument to + /// whole milliseconds and so cannot express these values at all. + /// + [Fact] + public void ClampWaitMilliseconds_SubMillisecondRemaining_RoundsUp() + { + // 10_000 ticks == 1 ms. + Assert.Equal(1, StaWaitHelper.ClampWaitMilliseconds(TimeSpan.FromTicks(4_000), 5)); + Assert.Equal(3, StaWaitHelper.ClampWaitMilliseconds(TimeSpan.FromTicks(25_000), 5)); + Assert.Equal(1, StaWaitHelper.ClampWaitMilliseconds(TimeSpan.FromTicks(1), 5)); + } + + /// + /// Verifies the clamp never exceeds the time actually left, which is what + /// keeps a wait from overrunning its caller's deadline. + /// + [Fact] + public void ClampWaitMilliseconds_NeverExceedsRemainingOrTheCeiling() + { + for (int remaining = 1; remaining <= 200; remaining++) + { + int slice = StaWaitHelper.ClampWaitMilliseconds( + TimeSpan.FromMilliseconds(remaining), + pollIntervalMilliseconds: 10_000); + + Assert.True(slice >= 1, $"Slice for {remaining} ms remaining was {slice}; a positive deadline must wait."); + Assert.True(slice <= remaining, $"Slice {slice} overran the {remaining} ms left before the deadline."); + Assert.True( + slice <= StaWaitHelper.MaxFallbackTickMilliseconds, + $"Slice {slice} exceeded the {StaWaitHelper.MaxFallbackTickMilliseconds} ms fallback ceiling."); + } + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs index 9c36df2..15019f4 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs @@ -100,6 +100,20 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor this.variantConverter = variantConverter ?? throw new ArgumentNullException(nameof(variantConverter)); this.statusProxyConverter = new MxStatusProxyConverter(); this.alarmCommandHandler = alarmCommandHandler; + + // The no-op fallback is a TEST affordance only — never pass null on a + // real STA. The cache wait loops this pump step feeds + // (MxAccessValueCache.TryWaitForUpdate, + // MxAccessWriteCompletionCache.TryWaitForCompletion) block in + // StaWaitHelper.WaitForSignalOrMessages, whose contract is that the pump + // step drains the message queue to empty on every call: the wait wakes on + // input that is merely present, so a pump step that removes nothing makes + // every wait return instantly and turns the loop into a busy spin at 100% + // CPU until its deadline (deadline and reply shape still hold — it is a + // CPU fault, not a correctness one). It is harmless here only because the + // fakes that pass null pre-populate the caches and run on threads with no + // windows and no pending input, so there is never a message left behind + // for the wait to keep waking on. this.pumpStep = pumpStep ?? (static () => { }); this.writeCompletionTimeout = writeCompletionTimeout ?? DefaultWriteCompletionTimeout; } diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs index abe133f..b6ecd3a 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs @@ -32,11 +32,26 @@ public sealed class MxAccessValueCache // the value is recorded from inside the waiter's own pumpStep (COM // dispatches OnDataChange on the STA), so the message wake is what does the // work there; this handle covers the cross-thread producers — the tests, and - // any future off-STA writer. Not IDisposable on purpose: the cache is - // co-owned by MxAccessBaseEventSink and MxAccessSession with no single - // teardown owner, and a late COM callback must never Set a disposed handle. - // One kernel event per session (one session per worker process), released by - // the handle's finalizer. + // any future off-STA writer. + // + // Single-waiter invariant: an AutoResetEvent releases exactly one waiter per + // Set. Today the sole waiter is the STA (one worker per session, and ReadBulk + // holds the apartment while it waits), so no wake is ever lost. A second + // concurrent waiter would not hang — the loser still re-checks the entry + // every fallback tick and still honors its deadline — it would just fall + // back to 50 ms latency for that iteration. + // + // NOT IDisposable, and that is one invariant with the wait's use of + // DangerousGetHandle: the cache is co-owned by MxAccessBaseEventSink and + // MxAccessSession with no single teardown owner, so a late COM callback + // could otherwise Set a disposed handle. StaWaitHelper deliberately does not + // DangerousAddRef the handle; GC.KeepAlive there closes the *finalization* + // race but not a dispose race — if this field were ever disposed while the + // STA sat inside the wait, the OS could recycle that handle value and the + // wait would silently attach to an unrelated object. Making these caches + // IDisposable is therefore not a free refactor: it must add DangerousAddRef/ + // DangerousRelease around the wait in the same change. One kernel event per + // session (one session per worker process), released by the finalizer. private readonly AutoResetEvent updateSignal = new(false); /// Records a fresh OnDataChange payload for the given handle pair. diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs index bfbfeaa..78a4114 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs @@ -31,11 +31,26 @@ public sealed class MxAccessWriteCompletionCache // worker the completion is recorded from inside the waiter's own pumpStep // (COM dispatches OnWriteComplete on the STA), so the message wake is what // does the work there; this handle covers the cross-thread producers — the - // tests, and any future off-STA recorder. Not IDisposable on purpose: the - // cache is co-owned by MxAccessBaseEventSink and MxAccessSession with no - // single teardown owner, and a late COM callback must never Set a disposed - // handle. One kernel event per session (one session per worker process), - // released by the handle's finalizer. + // tests, and any future off-STA recorder. + // + // Single-waiter invariant: an AutoResetEvent releases exactly one waiter per + // Set. Today the sole waiter is the STA (one worker per session, and the + // write path holds the apartment while it waits), so no wake is ever lost. A + // second concurrent waiter would not hang — the loser still re-checks the + // entry every fallback tick and still honors its deadline — it would just + // fall back to 50 ms latency for that iteration. + // + // NOT IDisposable, and that is one invariant with the wait's use of + // DangerousGetHandle: the cache is co-owned by MxAccessBaseEventSink and + // MxAccessSession with no single teardown owner, so a late COM callback + // could otherwise Set a disposed handle. StaWaitHelper deliberately does not + // DangerousAddRef the handle; GC.KeepAlive there closes the *finalization* + // race but not a dispose race — if this field were ever disposed while the + // STA sat inside the wait, the OS could recycle that handle value and the + // wait would silently attach to an unrelated object. Making these caches + // IDisposable is therefore not a free refactor: it must add DangerousAddRef/ + // DangerousRelease around the wait in the same change. One kernel event per + // session (one session per worker process), released by the finalizer. private readonly AutoResetEvent completionSignal = new(false); /// Records the status rows of a fresh OnWriteComplete callback for the given handle pair. diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaMessagePump.cs b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaMessagePump.cs index 74d31b7..535eec3 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaMessagePump.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaMessagePump.cs @@ -9,10 +9,7 @@ namespace ZB.MOM.WW.MxGateway.Worker.Sta; public sealed class StaMessagePump { private const uint Infinite = 0xFFFFFFFF; - private const uint MsgWaitFailed = 0xFFFFFFFF; - private const uint MwmoInputAvailable = 0x0004; private const uint PmRemove = 0x0001; - private const uint QsAllInput = 0x04FF; /// Waits for a command wake event or Windows messages, pumping any pending messages. /// Event to signal when work is available. @@ -28,14 +25,14 @@ public sealed class StaMessagePump SafeWaitHandle safeHandle = commandWakeEvent.SafeWaitHandle; IntPtr[] handles = [safeHandle.DangerousGetHandle()]; - uint result = MsgWaitForMultipleObjectsEx( + uint result = StaNativeMethods.MsgWaitForMultipleObjectsEx( (uint)handles.Length, handles, timeoutMilliseconds, - QsAllInput, - MwmoInputAvailable); + StaNativeMethods.QsAllInput, + StaNativeMethods.MwmoInputAvailable); - if (result == MsgWaitFailed) + if (result == StaNativeMethods.MsgWaitFailed) { throw new InvalidOperationException( "The worker STA message pump failed while waiting for command work or Windows messages."); @@ -75,14 +72,6 @@ public sealed class StaMessagePump : (uint)Math.Ceiling(timeout.TotalMilliseconds); } - [DllImport("user32.dll", SetLastError = true)] - private static extern uint MsgWaitForMultipleObjectsEx( - uint count, - IntPtr[] handles, - uint milliseconds, - uint wakeMask, - uint flags); - [DllImport("user32.dll", SetLastError = true)] private static extern bool PeekMessage( out NativeMessage message, diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaNativeMethods.cs b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaNativeMethods.cs new file mode 100644 index 0000000..032af95 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaNativeMethods.cs @@ -0,0 +1,58 @@ +using System; +using System.Runtime.InteropServices; + +namespace ZB.MOM.WW.MxGateway.Worker.Sta; + +/// +/// The single declaration of the Win32 wait the worker's STA is built on. +/// Both waiters use it: for the runtime's outer +/// idle wait, and for the inner completion waits +/// a command performs while it already holds the apartment. +/// +/// +/// Deliberately one copy. This is the subtlest Win32 contract in the worker — +/// the wake mask decides which events can wake the STA, and +/// decides whether a message already seen by +/// a PeekMessage still counts as a wake. Two copies of it could drift +/// apart silently, and a divergence would show up only as COM events +/// arriving late on a live provider. +/// +internal static class StaNativeMethods +{ + /// Return value of a failed wait (WAIT_FAILED). + internal const uint MsgWaitFailed = 0xFFFFFFFF; + + /// + /// Wake even when the pending input has already been seen (but not + /// removed) by an earlier PeekMessage. Both callers peek before + /// they wait, so without this flag a message that arrived during the peek + /// would not wake the wait that follows it. + /// + internal const uint MwmoInputAvailable = 0x0004; + + /// Wake on any queued input, including sent and posted messages. + internal const uint QsAllInput = 0x04FF; + + /// + /// Waits for the given handles, for queued input matching + /// , or for the timeout — whichever comes + /// first. + /// + /// Number of handles in . + /// Handles to wait on; marshaled as a pointer to the array. + /// Timeout in milliseconds; 0xFFFFFFFF means INFINITE. + /// Input types that may wake the wait (). + /// Wait flags (). + /// + /// WAIT_OBJECT_0 + i for a signaled handle, WAIT_OBJECT_0 + count + /// for available input, WAIT_TIMEOUT (258), or + /// . + /// + [DllImport("user32.dll", SetLastError = true)] + internal static extern uint MsgWaitForMultipleObjectsEx( + uint count, + IntPtr[] handles, + uint milliseconds, + uint wakeMask, + uint flags); +} diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaWaitHelper.cs b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaWaitHelper.cs index 7549d29..1171f7a 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaWaitHelper.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Sta/StaWaitHelper.cs @@ -1,4 +1,5 @@ using System; +using System.Diagnostics; using System.Runtime.InteropServices; using System.Threading; using Microsoft.Win32.SafeHandles; @@ -17,23 +18,22 @@ namespace ZB.MOM.WW.MxGateway.Worker.Sta; /// Deliberately separate from rather than a new /// method on it: the two caches that use this are plain per-session objects /// with no pump dependency (they take the pump step as an injected delegate -/// so tests can drive them off the STA). Keeping the helper in this folder -/// keeps every user32 wait signature in the worker reviewable in one -/// place; the declaration below mirrors the pump's exactly. +/// so tests can drive them off the STA). The wait itself is not duplicated — +/// both this and the pump call the single declaration in +/// . /// internal static class StaWaitHelper { /// - /// Ceiling on the blind fallback tick. A caller's poll interval is - /// clamped to this so no wait can go longer than 50 ms without calling - /// its pump step, even when the process has no message queue to wake it - /// (unit-test threads) and the wait handle is never signaled. + /// Ceiling on the fallback tick — the longest a wait may block when + /// nothing wakes it. A caller's poll interval is clamped to this, so a + /// caller's pump step keeps running at least this often no matter what + /// interval it asked for, even on a thread that never receives a message + /// and whose signal is never set. /// internal const int MaxFallbackTickMilliseconds = 50; - private const uint MsgWaitFailed = 0xFFFFFFFF; - private const uint MwmoInputAvailable = 0x0004; - private const uint QsAllInput = 0x04FF; + private static int waitFailureReported; /// /// Blocks the calling thread until is set, a @@ -42,6 +42,29 @@ internal static class StaWaitHelper /// never blinds the thread to inbound COM event messages: the wait /// returns the instant one arrives, so the caller can dispatch it. /// + /// + /// + /// Caller contract: the pump step must drain the message queue + /// to empty on every call. The wait uses + /// , which reports + /// input that is merely present — including input an earlier + /// PeekMessage has already seen but not removed. A caller + /// whose pump step drains only part of the queue, or does nothing at + /// all, therefore leaves a message that satisfies the wake condition + /// forever: every wait returns immediately, and the caller's poll + /// loop degenerates into a busy loop that spins the thread at 100% + /// until its deadline. That is a CPU fault, not a correctness one — + /// the loop still honors its deadline and return shape — but it must + /// not ship. Drain with PeekMessage(..., PM_REMOVE) in a loop + /// until it returns false, exactly as + /// does. + /// + /// + /// The wait does not DangerousAddRef the handle. That is + /// sound only because the callers never dispose their wait handles — + /// see the coupled invariant documented on the caches' signal fields. + /// + /// /// Wait handle set by whoever produces the awaited result. /// Maximum wait; values at or below zero return immediately. internal static void WaitForSignalOrMessages( @@ -60,20 +83,28 @@ internal static class StaWaitHelper SafeWaitHandle safeHandle = signal.SafeWaitHandle; IntPtr[] handles = [safeHandle.DangerousGetHandle()]; - uint result = MsgWaitForMultipleObjectsEx( + uint result = StaNativeMethods.MsgWaitForMultipleObjectsEx( (uint)handles.Length, handles, (uint)timeoutMilliseconds, - QsAllInput, - MwmoInputAvailable); + StaNativeMethods.QsAllInput, + StaNativeMethods.MwmoInputAvailable); + + // Read before anything else can run another P/Invoke and overwrite the + // thread's last-error slot. + int lastError = result == StaNativeMethods.MsgWaitFailed + ? Marshal.GetLastWin32Error() + : 0; // The handle is a field of a live caller object, so it cannot be // collected mid-call; KeepAlive states that dependency explicitly // rather than relying on it. GC.KeepAlive(signal); - if (result == MsgWaitFailed) + if (result == StaNativeMethods.MsgWaitFailed) { + ReportWaitFailureOnce(lastError); + // Degrade to the pre-existing blind sleep instead of throwing or // hot-spinning: the caller is a bounded, deadline-driven poll loop, // and a failed wait must not turn a timed-out write into a fault. @@ -108,11 +139,30 @@ internal static class StaWaitHelper : tick; } - [DllImport("user32.dll", SetLastError = true)] - private static extern uint MsgWaitForMultipleObjectsEx( - uint count, - IntPtr[] handles, - uint milliseconds, - uint wakeMask, - uint flags); + /// + /// Traces the first wait failure of the process and stays silent after + /// that. + /// + /// + /// The degrade path is silent by design — a failed wait still honors the + /// caller's deadline — but a permanently failing wait means every + /// completion wait has quietly reverted to the sleep-polled behavior this + /// class exists to replace, and nothing else would ever say so. One trace + /// makes that discoverable; latching it keeps a repeating failure from + /// flooding the trace listener from inside a poll loop. + /// + /// The Win32 error captured immediately after the failed wait. + private static void ReportWaitFailureOnce(int lastError) + { + if (Interlocked.Exchange(ref waitFailureReported, 1) != 0) + { + return; + } + + Trace.TraceWarning( + "MsgWaitForMultipleObjectsEx failed in the worker STA completion wait (Win32 error {0}). " + + "The wait has degraded to blind sleeping: deadlines are still honored, but inbound " + + "MXAccess COM events will only dispatch at poll-tick granularity. Reported once per process.", + lastError); + } }