fix(worker): StaWaitHelper regression tests + full-drain contract notes; consolidate wait P/Invoke
This commit is contained in:
Joseph Doherty
@@ -179,27 +179,35 @@ public sealed class MxAccessValueCacheTests
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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 <c>Thread.Sleep</c> 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.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Scope: this proves the cadence bound, not the signal path. The clamp
|
||||
/// alone would satisfy it, so it stays green if the <c>Set</c>-side
|
||||
/// signal is deleted. <c>StaWaitHelperTests</c> 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.
|
||||
/// </remarks>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
+18
-10
@@ -113,25 +113,33 @@ public sealed class MxAccessWriteCompletionCacheTests
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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 <c>Thread.Sleep</c> this
|
||||
/// replaced would have slept the full interval — on the STA, 10 s with no
|
||||
/// Windows messages dispatched — and then reported a timeout.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Scope: this proves the cadence bound, not the signal path. The clamp
|
||||
/// alone would satisfy it, so it stays green if the <c>Record</c>-side
|
||||
/// <c>Set()</c> is deleted. <c>StaWaitHelperTests</c> 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.
|
||||
/// </remarks>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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<MxStatusProxy> 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.");
|
||||
}
|
||||
|
||||
/// <summary>Verifies that Record stores an independent clone of the caller's rows.</summary>
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Tests for <see cref="StaWaitHelper"/>, the message-aware wait behind the
|
||||
/// write-completion and ReadBulk value waits.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// These exist because the cache-level tests cannot guard the signal
|
||||
/// path. Every cache wait clamps its slice to
|
||||
/// <see cref="StaWaitHelper.MaxFallbackTickMilliseconds"/>, 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' <c>Set()</c> 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Boundary: the <c>WAIT_FAILED</c> branch is not exercised, for the same
|
||||
/// reason <see cref="StaMessagePumpTests"/> gives — forcing
|
||||
/// <c>MsgWaitForMultipleObjectsEx</c> to fail means handing it a
|
||||
/// deliberately invalid native handle, which is unsafe to construct in a
|
||||
/// managed test.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class StaWaitHelperTests
|
||||
{
|
||||
/// <summary>Verifies that a null wait handle is rejected.</summary>
|
||||
[Fact]
|
||||
public void WaitForSignalOrMessages_NullSignal_ThrowsArgumentNullException()
|
||||
{
|
||||
ArgumentNullException exception = Assert.Throws<ArgumentNullException>(
|
||||
() => StaWaitHelper.WaitForSignalOrMessages(null!, 10));
|
||||
|
||||
Assert.Equal("signal", exception.ParamName);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <param name="timeoutMilliseconds">The non-positive timeout under test.</param>
|
||||
[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.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <param name="remainingMilliseconds">Time left before the caller's deadline.</param>
|
||||
/// <param name="pollIntervalMilliseconds">The caller's requested pump cadence.</param>
|
||||
/// <param name="expected">The expected wait slice in milliseconds.</param>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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
|
||||
/// <see cref="TimeSpan.FromMilliseconds(double)"/> rounds its argument to
|
||||
/// whole milliseconds and so cannot express these values at all.
|
||||
/// </summary>
|
||||
[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));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies the clamp never exceeds the time actually left, which is what
|
||||
/// keeps a wait from overrunning its caller's deadline.
|
||||
/// </summary>
|
||||
[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.");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
/// <summary>Records a fresh OnDataChange payload for the given handle pair.</summary>
|
||||
|
||||
@@ -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);
|
||||
|
||||
/// <summary>Records the status rows of a fresh OnWriteComplete callback for the given handle pair.</summary>
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <summary>Waits for a command wake event or Windows messages, pumping any pending messages.</summary>
|
||||
/// <param name="commandWakeEvent">Event to signal when work is available.</param>
|
||||
@@ -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,
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
using System;
|
||||
using System.Runtime.InteropServices;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Worker.Sta;
|
||||
|
||||
/// <summary>
|
||||
/// The single declaration of the Win32 wait the worker's STA is built on.
|
||||
/// Both waiters use it: <see cref="StaMessagePump"/> for the runtime's outer
|
||||
/// idle wait, and <see cref="StaWaitHelper"/> for the inner completion waits
|
||||
/// a command performs while it already holds the apartment.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Deliberately one copy. This is the subtlest Win32 contract in the worker —
|
||||
/// the wake mask decides which events can wake the STA, and
|
||||
/// <see cref="MwmoInputAvailable"/> decides whether a message already seen by
|
||||
/// a <c>PeekMessage</c> 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.
|
||||
/// </remarks>
|
||||
internal static class StaNativeMethods
|
||||
{
|
||||
/// <summary>Return value of a failed wait (<c>WAIT_FAILED</c>).</summary>
|
||||
internal const uint MsgWaitFailed = 0xFFFFFFFF;
|
||||
|
||||
/// <summary>
|
||||
/// Wake even when the pending input has already been seen (but not
|
||||
/// removed) by an earlier <c>PeekMessage</c>. 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.
|
||||
/// </summary>
|
||||
internal const uint MwmoInputAvailable = 0x0004;
|
||||
|
||||
/// <summary>Wake on any queued input, including sent and posted messages.</summary>
|
||||
internal const uint QsAllInput = 0x04FF;
|
||||
|
||||
/// <summary>
|
||||
/// Waits for the given handles, for queued input matching
|
||||
/// <paramref name="wakeMask"/>, or for the timeout — whichever comes
|
||||
/// first.
|
||||
/// </summary>
|
||||
/// <param name="count">Number of handles in <paramref name="handles"/>.</param>
|
||||
/// <param name="handles">Handles to wait on; marshaled as a pointer to the array.</param>
|
||||
/// <param name="milliseconds">Timeout in milliseconds; <c>0xFFFFFFFF</c> means INFINITE.</param>
|
||||
/// <param name="wakeMask">Input types that may wake the wait (<see cref="QsAllInput"/>).</param>
|
||||
/// <param name="flags">Wait flags (<see cref="MwmoInputAvailable"/>).</param>
|
||||
/// <returns>
|
||||
/// <c>WAIT_OBJECT_0 + i</c> for a signaled handle, <c>WAIT_OBJECT_0 + count</c>
|
||||
/// for available input, <c>WAIT_TIMEOUT</c> (258), or
|
||||
/// <see cref="MsgWaitFailed"/>.
|
||||
/// </returns>
|
||||
[DllImport("user32.dll", SetLastError = true)]
|
||||
internal static extern uint MsgWaitForMultipleObjectsEx(
|
||||
uint count,
|
||||
IntPtr[] handles,
|
||||
uint milliseconds,
|
||||
uint wakeMask,
|
||||
uint flags);
|
||||
}
|
||||
@@ -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 <see cref="StaMessagePump"/> 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 <c>user32</c> 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
|
||||
/// <see cref="StaNativeMethods"/>.
|
||||
/// </remarks>
|
||||
internal static class StaWaitHelper
|
||||
{
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
internal const int MaxFallbackTickMilliseconds = 50;
|
||||
|
||||
private const uint MsgWaitFailed = 0xFFFFFFFF;
|
||||
private const uint MwmoInputAvailable = 0x0004;
|
||||
private const uint QsAllInput = 0x04FF;
|
||||
private static int waitFailureReported;
|
||||
|
||||
/// <summary>
|
||||
/// Blocks the calling thread until <paramref name="signal"/> 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.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <em>Caller contract: the pump step must drain the message queue
|
||||
/// to empty on every call.</em> The wait uses
|
||||
/// <see cref="StaNativeMethods.MwmoInputAvailable"/>, which reports
|
||||
/// input that is merely <em>present</em> — including input an earlier
|
||||
/// <c>PeekMessage</c> 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 <c>PeekMessage(..., PM_REMOVE)</c> in a loop
|
||||
/// until it returns false, exactly as
|
||||
/// <see cref="StaMessagePump.PumpPendingMessages"/> does.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The wait does not <c>DangerousAddRef</c> the handle. That is
|
||||
/// sound only because the callers never dispose their wait handles —
|
||||
/// see the coupled invariant documented on the caches' signal fields.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="signal">Wait handle set by whoever produces the awaited result.</param>
|
||||
/// <param name="timeoutMilliseconds">Maximum wait; values at or below zero return immediately.</param>
|
||||
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);
|
||||
/// <summary>
|
||||
/// Traces the first wait failure of the process and stays silent after
|
||||
/// that.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The degrade path is silent by design — a failed wait still honors the
|
||||
/// caller's deadline — but a <em>permanently</em> 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.
|
||||
/// </remarks>
|
||||
/// <param name="lastError">The Win32 error captured immediately after the failed wait.</param>
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user