fix(worker): review 2026-10-21 stream E — cache eviction, phased ReadBulk, poll/STA fault handling
ci / nightly-windev (push) Has been skipped
ci / windows-x86 (push) Failing after 1m3s
ci / java (push) Successful in 2m17s
ci / portable (push) Failing after 4m42s

- RemoveItem/Unregister evict value + write-completion caches (leak under churn)
- ReadBulk snapshot path: add/advise all, one shared deadline, then tear down
- alarm poll faults only after 3 consecutive failures; transient ones logged
- AlarmCommandHandler disposes primary when standby build throws
- StaRuntime pumps between queued work items
- WRK-02: post-start STA loop death surfaces as StaHung WorkerFault
- WRK-09: AppDomain unhandled-exception hook, exit code 7
- RETRIGGER documented as reserved, never synthesized
- TST-05: live facts for Ping/GetSessionState/GetWorkerInfo/DrainEvents/ShutdownWorker
This commit is contained in:
Joseph Doherty
2026-10-22 06:19:32 -04:00
parent 488649ce99
commit 7618cafb6c
22 changed files with 1859 additions and 90 deletions
@@ -246,13 +246,32 @@ public sealed class AlarmCommandHandler : IAlarmCommandHandler
{
IMxAccessAlarmConsumer primary = consumerFactory()
?? throw new InvalidOperationException("Alarm consumer factory returned null.");
IMxAccessAlarmConsumer standby = BuildStandby(watchList);
AlarmFailoverConfig? failoverConfig = command.Failover;
FailoverSettings settings = new FailoverSettings(
failoverConfig?.ConsecutiveFailureThreshold ?? 3,
failoverConfig?.FailbackProbeIntervalSeconds ?? 30,
failoverConfig?.FailbackStableProbes ?? 3);
return new FailoverAlarmConsumer(primary, standby, settings);
IMxAccessAlarmConsumer? standby = null;
try
{
standby = BuildStandby(watchList);
AlarmFailoverConfig? failoverConfig = command.Failover;
FailoverSettings settings = new FailoverSettings(
failoverConfig?.ConsecutiveFailureThreshold ?? 3,
failoverConfig?.FailbackProbeIntervalSeconds ?? 30,
failoverConfig?.FailbackStableProbes ?? 3);
return new FailoverAlarmConsumer(primary, standby, settings);
}
catch
{
// The primary (and a standby that was built before the
// composite constructor threw) is owned by nothing yet: the
// failed Subscribe never reaches the dispatcher that would
// dispose it, so release it here or its wnwrap COM object
// leaks for the life of the worker.
if (standby is not null)
{
try { standby.Dispose(); } catch { /* swallow — preserve the build failure */ }
}
try { primary.Dispose(); } catch { /* swallow — preserve the build failure */ }
throw;
}
}
// Alarmmgr, or Unspecified with an empty watch list — primary only.
@@ -49,6 +49,17 @@ public static class AlarmRecordTransitionMapper
/// <item><description><c>previous=*Rtn</c> + <c>current=*Alm</c> → Raise (re-trigger after clear).</description></item>
/// <item><description>Anything else → Unspecified (no proto kind to emit).</description></item>
/// </list>
/// <para>
/// <see cref="AlarmTransitionKind.Retrigger"/> is never returned. It is
/// reserved in the contract, but neither alarm source gives the worker
/// a native re-alarm signal: the alarmmgr XML carries only the four
/// <c>*_ALM</c>/<c>*_RTN</c> states, so a return-to-alarm is
/// indistinguishable from a fresh raise and is reported as Raise, and
/// the subtag source observes no re-alarm counter. Deliberately not
/// synthesized (the gateway forwards only transitions the source
/// reports — see docs/AlarmClientDiscovery.md, "Retrigger
/// transitions").
/// </para>
/// </summary>
/// <param name="previous">The previous alarm state kind.</param>
/// <param name="current">The current alarm state kind.</param>
@@ -222,6 +222,13 @@ public sealed class MxAccessSession : IDisposable
mxAccessServer.Unregister(serverHandle);
handleRegistry.UnregisterServerHandle(serverHandle);
// Unregister drops every item under the server in one call, so the
// per-item evictions in RemoveItem never run for them. Evict both
// per-handle caches by server so a long-lived worker that registers
// and unregisters clients does not accumulate one entry per item it
// ever wrote or read, and a reused server handle id starts clean.
valueCache.RemoveServer(serverHandle);
writeCompletionCache.RemoveServer(serverHandle);
}
/// <summary>Adds an item to an MXAccess server and returns the item handle.</summary>
@@ -283,6 +290,11 @@ public sealed class MxAccessSession : IDisposable
// same handle id (which MXAccess may reuse) does not serve a stale
// OnDataChange snapshot from the previous lifetime.
valueCache.Remove(serverHandle, itemHandle);
// Same for the OnWriteComplete completion entry: the event sink records
// one per written handle pair, so without this an AddItem -> Write ->
// RemoveItem churn grows the cache by one entry per cycle for the life
// of the worker.
writeCompletionCache.Remove(serverHandle, itemHandle);
}
/// <summary>Advises on item changes with plain subscription.</summary>
@@ -870,14 +882,28 @@ public sealed class MxAccessSession : IDisposable
/// OnDataChange value if the tag is already advised AND a cached value
/// exists (no subscription side effects); otherwise takes the AddItem
/// + Advise + wait + UnAdvise + RemoveItem snapshot lifecycle itself.
/// <paramref name="timeout"/> bounds the wait per-tag in the snapshot
/// case; <paramref name="pumpStep"/> is invoked on every poll
/// iteration so the worker's STA can dispatch the incoming MXAccess
/// message that carries the value.
/// <paramref name="timeout"/> bounds the wait for every snapshot tag;
/// <paramref name="pumpStep"/> is invoked on every poll iteration so the
/// worker's STA can dispatch the incoming MXAccess message that carries
/// the value.
/// </summary>
/// <remarks>
/// The snapshot tags are processed in phases rather than one at a time:
/// every cache probe runs first, then every uncached tag is added and
/// advised, then one wait with a single shared deadline collects the
/// first OnDataChange per handle, then every snapshot handle is torn
/// down. A serial lifecycle held the STA for up to N x timeout when N
/// tags never produced a value; the phased form holds it for at most
/// one timeout. Each tag's result is still its own: a tag that never
/// delivered fails with the same timed-out diagnostic and its own item
/// handle, an AddItem or Advise failure fails only that tag, and results
/// keep request order. The cache probes run before any snapshot AddItem
/// so a later tag can never be served from an earlier tag's own
/// snapshot subscription.
/// </remarks>
/// <param name="serverHandle">The MXAccess server handle.</param>
/// <param name="tagAddresses">The tag addresses to read.</param>
/// <param name="timeout">The timeout per tag.</param>
/// <param name="timeout">The timeout bounding the shared snapshot wait.</param>
/// <param name="pumpStep">Action invoked on each poll iteration.</param>
/// <returns>The per-tag read results, in the order the tags were provided.</returns>
public IReadOnlyList<BulkReadResult> ReadBulk(
@@ -897,89 +923,157 @@ public sealed class MxAccessSession : IDisposable
throw new ArgumentNullException(nameof(pumpStep));
}
List<BulkReadResult> results = new(tagAddresses.Count);
foreach (string? tagAddress in tagAddresses)
int tagCount = tagAddresses.Count;
BulkReadResult?[] results = new BulkReadResult?[tagCount];
List<SnapshotRead> snapshots = new();
// Phase 1: validation and the cached-and-advised fast path. A tag
// already advised by the caller with a cached value is answered without
// touching the existing subscription — the caller didn't create it, so
// ReadBulk must not tear it down.
for (int index = 0; index < tagCount; index++)
{
string? tagAddress = tagAddresses[index];
if (string.IsNullOrWhiteSpace(tagAddress))
{
results.Add(FailedRead(serverHandle, tagAddress ?? string.Empty, itemHandle: 0, wasCached: false, "Tag address is required."));
results[index] = FailedRead(serverHandle, tagAddress ?? string.Empty, itemHandle: 0, wasCached: false, "Tag address is required.");
continue;
}
results.Add(ReadOneTag(serverHandle, tagAddress, timeout, pumpStep));
if (TryGetCachedReadFor(serverHandle, tagAddress!, out int cachedItemHandle, out MxAccessValueCache.CachedValue cachedValue))
{
results[index] = SucceededRead(serverHandle, tagAddress!, cachedItemHandle, wasCached: true, cachedValue);
continue;
}
snapshots.Add(new SnapshotRead(index, tagAddress!));
}
return results;
if (snapshots.Count > 0)
{
try
{
ReadSnapshots(serverHandle, snapshots, results, timeout, pumpStep);
}
finally
{
TearDownSnapshots(serverHandle, snapshots);
}
}
List<BulkReadResult> ordered = new(tagCount);
for (int index = 0; index < tagCount; index++)
{
ordered.Add(results[index]!);
}
return ordered;
}
private BulkReadResult ReadOneTag(
private void ReadSnapshots(
int serverHandle,
string tagAddress,
List<SnapshotRead> snapshots,
BulkReadResult?[] results,
TimeSpan timeout,
Action pumpStep)
{
// 1. Cached-and-advised fast path: scan the registry for a live item
// handle matching this tag and check whether the value cache has a
// payload for it. If so, return the cached value without touching
// the existing subscription — the caller didn't create it, so
// ReadBulk must not tear it down.
if (TryGetCachedReadFor(serverHandle, tagAddress, out int cachedItemHandle, out MxAccessValueCache.CachedValue cachedValue))
// Phase 2: reserve our own item handle and advise for every uncached
// tag. A per-tag AddItem/Advise failure fails only that tag.
List<SnapshotRead> waiting = new(snapshots.Count);
foreach (SnapshotRead snapshot in snapshots)
{
return SucceededRead(
serverHandle,
tagAddress,
cachedItemHandle,
wasCached: true,
cachedValue);
try
{
snapshot.ItemHandle = AddItem(serverHandle, snapshot.TagAddress);
snapshot.Baseline = valueCache.CurrentVersion(serverHandle, snapshot.ItemHandle);
Advise(serverHandle, snapshot.ItemHandle);
snapshot.Advised = true;
waiting.Add(snapshot);
}
catch (Exception exception)
{
results[snapshot.Index] = FailedRead(serverHandle, snapshot.TagAddress, snapshot.ItemHandle, wasCached: false, exception.Message);
}
}
// 2. Snapshot lifecycle. Reserve our own item handle, advise, pump
// until we see a fresh OnDataChange (or the deadline elapses),
// then tear it down.
int itemHandle = 0;
bool advised = false;
if (waiting.Count == 0)
{
return;
}
// Phase 3: one shared deadline. Pump until every advised handle has
// seen its first OnDataChange since its baseline, or the deadline
// elapses.
int[] itemHandles = new int[waiting.Count];
ulong[] baselines = new ulong[waiting.Count];
for (int index = 0; index < waiting.Count; index++)
{
itemHandles[index] = waiting[index].ItemHandle;
baselines[index] = waiting[index].Baseline;
}
bool[] received = new bool[waiting.Count];
MxAccessValueCache.CachedValue[] values = new MxAccessValueCache.CachedValue[waiting.Count];
try
{
itemHandle = AddItem(serverHandle, tagAddress);
ulong baseline = valueCache.CurrentVersion(serverHandle, itemHandle);
Advise(serverHandle, itemHandle);
advised = true;
DateTime deadline = DateTime.UtcNow + timeout;
bool gotValue = valueCache.TryWaitForUpdate(
serverHandle,
itemHandle,
baseline,
deadline,
pumpStep,
out MxAccessValueCache.CachedValue snapshot);
return gotValue
? SucceededRead(serverHandle, tagAddress, itemHandle, wasCached: false, snapshot)
: FailedRead(
serverHandle,
tagAddress,
itemHandle,
wasCached: false,
$"ReadBulk timed out after {timeout.TotalMilliseconds:F0} ms waiting for first OnDataChange.");
valueCache.TryWaitForUpdates(serverHandle, itemHandles, baselines, deadline, pumpStep, received, values);
}
catch (Exception exception)
{
return FailedRead(serverHandle, tagAddress, itemHandle, wasCached: false, exception.Message);
}
finally
{
// Snapshot teardown — best-effort. Errors here are noted on the
// diagnostic message of the original result (above) by appending
// a cleanup suffix; we never re-throw from finally.
if (advised)
// A pump failure ends the wait for every tag still outstanding;
// tags captured before it keep their values.
for (int index = 0; index < waiting.Count; index++)
{
try { UnAdvise(serverHandle, itemHandle); } catch { /* swallow — best effort */ }
if (!received[index])
{
SnapshotRead failed = waiting[index];
results[failed.Index] = FailedRead(serverHandle, failed.TagAddress, failed.ItemHandle, wasCached: false, exception.Message);
}
}
}
for (int index = 0; index < waiting.Count; index++)
{
SnapshotRead snapshot = waiting[index];
if (results[snapshot.Index] is not null)
{
continue;
}
if (itemHandle != 0)
results[snapshot.Index] = received[index]
? SucceededRead(serverHandle, snapshot.TagAddress, snapshot.ItemHandle, wasCached: false, values[index])
: FailedRead(
serverHandle,
snapshot.TagAddress,
snapshot.ItemHandle,
wasCached: false,
$"ReadBulk timed out after {timeout.TotalMilliseconds:F0} ms waiting for first OnDataChange.");
}
}
private void TearDownSnapshots(
int serverHandle,
List<SnapshotRead> snapshots)
{
// Phase 4: snapshot teardown — best-effort, never re-throws. A handle
// is torn down once even if MXAccess handed the same id to two tags in
// the batch.
HashSet<int> unadvised = new();
HashSet<int> removed = new();
foreach (SnapshotRead snapshot in snapshots)
{
if (snapshot.Advised && unadvised.Add(snapshot.ItemHandle))
{
try { RemoveItem(serverHandle, itemHandle); } catch { /* swallow — best effort */ }
try { UnAdvise(serverHandle, snapshot.ItemHandle); } catch { /* swallow — best effort */ }
}
}
foreach (SnapshotRead snapshot in snapshots)
{
if (snapshot.ItemHandle != 0 && removed.Add(snapshot.ItemHandle))
{
try { RemoveItem(serverHandle, snapshot.ItemHandle); } catch { /* swallow — best effort */ }
}
}
}
@@ -1308,6 +1402,35 @@ public sealed class MxAccessSession : IDisposable
}
}
/// <summary>
/// Mutable per-tag state for one ReadBulk snapshot lifecycle. A plain
/// class with get/set properties (no records or init accessors) so it
/// compiles under the worker's net48 target.
/// </summary>
private sealed class SnapshotRead
{
public SnapshotRead(int index, string tagAddress)
{
Index = index;
TagAddress = tagAddress;
}
/// <summary>Gets the position of the tag in the request.</summary>
public int Index { get; }
/// <summary>Gets the requested tag address.</summary>
public string TagAddress { get; }
/// <summary>Gets or sets the snapshot item handle, 0 until AddItem succeeds.</summary>
public int ItemHandle { get; set; }
/// <summary>Gets or sets the value-cache version captured before Advise.</summary>
public ulong Baseline { get; set; }
/// <summary>Gets or sets a value indicating whether Advise succeeded.</summary>
public bool Advised { get; set; }
}
private void ThrowIfDisposed()
{
if (disposed)
@@ -42,7 +42,15 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
/// <summary>Default alarm poll cadence when the environment says nothing usable.</summary>
internal static readonly TimeSpan DefaultAlarmPollInterval = TimeSpan.FromMilliseconds(500);
private readonly TimeSpan alarmPollInterval = ResolveAlarmPollInterval();
/// <summary>
/// Consecutive alarm-poll failures that fault the session. Matches the
/// default of <see cref="FailoverSettings.Threshold"/>
/// (<c>AlarmFailoverConfig.consecutive_failure_threshold</c>, default 3):
/// one transient COM hiccup in <c>PollOnce</c> is logged and retried on
/// the next tick, and only a run of failures with no success in between
/// is treated as a dead subscription.
/// </summary>
internal const int DefaultAlarmPollFailureThreshold = 3;
private readonly IMxAccessComObjectFactory factory;
private readonly IMxAccessEventSink eventSink;
@@ -221,6 +229,28 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
/// </summary>
internal TimeSpan WriteCompletionTimeout { get; set; } = ResolveWriteCompletionTimeout();
/// <summary>
/// Cadence the alarm poll loop waits between polls. Resolved from the
/// launcher environment; internal-settable as a test seam (set before
/// <see cref="StartAsync(string, int, CancellationToken)"/>).
/// </summary>
internal TimeSpan AlarmPollInterval { get; set; } = ResolveAlarmPollInterval();
/// <summary>
/// Consecutive alarm-poll failures that fault the session; clamped to a
/// minimum of 1. Internal-settable as a test seam.
/// </summary>
internal int AlarmPollFailureThreshold { get; set; } = DefaultAlarmPollFailureThreshold;
/// <summary>
/// Sink for the transient alarm-poll failure warning. Defaults to
/// <see cref="Console.Error"/>, the stream <c>WorkerConsoleLogger</c>
/// writes to. The line carries the exception type, HRESULT, and
/// failure counts only — never the exception message, which can echo
/// alarm or tag names. Tests substitute a collector.
/// </summary>
internal Action<string>? AlarmPollWarningSink { get; set; }
/// <summary>
/// Resolves the write-completion wait from the launcher-provided
/// environment variable; a missing or invalid value falls back to
@@ -288,6 +318,13 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
{
staRuntime.Start();
// WRK-02: a post-start STA loop death has no other observer. Record it
// on the event queue so the IPC drain loop sends a WorkerFault to the
// gateway and ends the pipe session (and with it the worker process),
// instead of leaving a live process whose commands queue forever.
staRuntime.RegisterLoopFaultHandler(
exception => eventQueue.RecordFault(CreateStaLoopFault(exception)));
WorkerReady ready = await staRuntime.InvokeAsync(
() =>
{
@@ -345,13 +382,16 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
IAlarmCommandHandler handler,
CancellationToken cancellationToken)
{
TimeSpan pollInterval = AlarmPollInterval;
int failureThreshold = AlarmPollFailureThreshold < 1 ? 1 : AlarmPollFailureThreshold;
return Task.Run(async () =>
{
int consecutiveFailures = 0;
while (!cancellationToken.IsCancellationRequested)
{
try
{
await Task.Delay(alarmPollInterval, cancellationToken).ConfigureAwait(false);
await Task.Delay(pollInterval, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
@@ -384,6 +424,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
}
},
cancellationToken).ConfigureAwait(false);
consecutiveFailures = 0;
}
catch (OperationCanceledException)
{
@@ -413,19 +454,57 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
// A real alarm-poll failure (COMException from
// GetXmlCurrentAlarms2, malformed-XML parse failure, an
// STA-affinity InvalidOperationException from
// EnsureOnAlarmConsumerThread, etc.). Record it as a
// fault on the event queue so a broken alarm subscription
// — or an affinity-invariant regression — becomes
// EnsureOnAlarmConsumerThread, etc.). A single failure is
// logged and retried on the next tick: one transient COM
// hiccup must not tear down the whole session (every data
// subscription with it). Only a run of
// AlarmPollFailureThreshold failures with no success in
// between is recorded as a fault on the event queue, so a
// broken alarm subscription — or an affinity-invariant
// regression, which fails every tick — still becomes
// observable on the IPC fault path instead of silently
// faulting this never-awaited task. The loop then stops —
// the subscription is dead.
eventQueue.RecordFault(CreateAlarmPollFault(exception));
return;
consecutiveFailures++;
if (consecutiveFailures >= failureThreshold)
{
eventQueue.RecordFault(CreateAlarmPollFault(exception));
return;
}
WriteAlarmPollWarning(exception, consecutiveFailures, failureThreshold);
}
}
}, CancellationToken.None);
}
private void WriteAlarmPollWarning(Exception exception, int consecutiveFailures, int failureThreshold)
{
// Same "level=<L> event=<E> k=v" line shape as WorkerConsoleLogger and
// the wnwrap truncation warning. Type and HRESULT only — the exception
// message is deliberately left out (it can carry alarm or tag names).
string hresult = exception is System.Runtime.InteropServices.COMException comException
? "0x" + comException.HResult.ToString("X8", System.Globalization.CultureInfo.InvariantCulture)
: "none";
string message = string.Format(
System.Globalization.CultureInfo.InvariantCulture,
"level=Warning event=AlarmPollFailed exceptionType={0} hresult={1} consecutiveFailures={2} faultThreshold={3}",
exception.GetType().FullName,
hresult,
consecutiveFailures,
failureThreshold);
try
{
Action<string> sink = AlarmPollWarningSink ?? Console.Error.WriteLine;
sink(message);
}
catch
{
// Logging must never end the poll loop.
}
}
private void EnsureOnAlarmConsumerThread()
{
AssertOnAlarmConsumerThread(alarmConsumerThreadId, Environment.CurrentManagedThreadId);
@@ -458,6 +537,36 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
}
}
/// <summary>
/// Builds the fault reported when the STA loop dies after start. Uses
/// <see cref="WorkerFaultCategory.StaHung"/>: the gateway already treats
/// that category as "this worker's apartment can no longer run COM
/// work", which is exactly the state a dead loop leaves behind.
/// </summary>
internal static WorkerFault CreateStaLoopFault(Exception exception)
{
string message =
$"Worker STA loop terminated unexpectedly: {exception.GetType().FullName}: {exception.Message}";
WorkerFault fault = new()
{
Category = WorkerFaultCategory.StaHung,
ExceptionType = exception.GetType().FullName ?? string.Empty,
DiagnosticMessage = message,
ProtocolStatus = new ProtocolStatus
{
Code = ProtocolStatusCode.WorkerUnavailable,
Message = message,
},
};
if (exception is System.Runtime.InteropServices.COMException comException)
{
fault.Hresult = comException.HResult;
}
return fault;
}
private static WorkerFault CreateAlarmPollFault(Exception exception)
{
string message =
@@ -162,6 +162,46 @@ public sealed class MxAccessValueCache
}
}
/// <summary>
/// Removes every cache slot recorded under a server handle. The session
/// calls this when a server handle is unregistered: MXAccess drops all of
/// that server's items with it, so the per-item <see cref="Remove"/>
/// never runs for them and their entries would otherwise outlive the
/// handle (and could be served to a later server that reuses the id).
/// </summary>
/// <param name="serverHandle">MXAccess server handle that was unregistered.</param>
public void RemoveServer(int serverHandle)
{
lock (syncRoot)
{
List<long> stale = new();
foreach (long key in entries.Keys)
{
if ((int)(key >> 32) == serverHandle)
{
stale.Add(key);
}
}
foreach (long key in stale)
{
entries.Remove(key);
}
}
}
/// <summary>Gets the number of handle pairs with a cached value (test seam).</summary>
internal int Count
{
get
{
lock (syncRoot)
{
return entries.Count;
}
}
}
/// <summary>
/// Waits until the cache entry's version exceeds <paramref name="sinceVersion"/>
/// or the deadline elapses, calling <paramref name="pumpStep"/> on every
@@ -231,6 +271,117 @@ public sealed class MxAccessValueCache
}
}
/// <summary>
/// Multi-handle form of <see cref="TryWaitForUpdate"/>: waits until every
/// handle in <paramref name="itemHandles"/> has an entry newer than its
/// matching <paramref name="sinceVersions"/> slot, or the single shared
/// deadline elapses. Each handle is captured the first time an update is
/// observed for it (its first OnDataChange since the baseline), exactly
/// as a dedicated per-handle wait would have captured it, so a bulk
/// snapshot read waits at most one timeout in total rather than one per
/// tag.
/// </summary>
/// <remarks>
/// Pumps unconditionally at the top of every iteration and blocks the
/// idle part of each iteration in
/// <see cref="StaWaitHelper.WaitForSignalOrMessages"/> — the same pump
/// discipline as <see cref="TryWaitForUpdate"/>, for the same reason.
/// </remarks>
/// <param name="serverHandle">MXAccess server handle shared by every item.</param>
/// <param name="itemHandles">Item handles to wait on.</param>
/// <param name="sinceVersions">Per-handle version baselines captured before advising.</param>
/// <param name="deadlineUtc">Absolute UTC deadline shared by every handle.</param>
/// <param name="pumpStep">Action that pumps any pending Windows messages.</param>
/// <param name="received">Output: <see langword="true"/> at index i when handle i received an update before the deadline.</param>
/// <param name="values">Output: the captured value at index i when <paramref name="received"/>[i] is set.</param>
/// <param name="pollIntervalMs">Upper bound on one idle wait slice.</param>
/// <returns>The number of handles that received an update before the deadline.</returns>
public int TryWaitForUpdates(
int serverHandle,
IReadOnlyList<int> itemHandles,
IReadOnlyList<ulong> sinceVersions,
DateTime deadlineUtc,
Action pumpStep,
bool[] received,
CachedValue[] values,
int pollIntervalMs = 5)
{
if (itemHandles is null)
{
throw new ArgumentNullException(nameof(itemHandles));
}
if (sinceVersions is null)
{
throw new ArgumentNullException(nameof(sinceVersions));
}
if (pumpStep is null)
{
throw new ArgumentNullException(nameof(pumpStep));
}
if (received is null)
{
throw new ArgumentNullException(nameof(received));
}
if (values is null)
{
throw new ArgumentNullException(nameof(values));
}
int count = itemHandles.Count;
if (sinceVersions.Count != count || received.Length < count || values.Length < count)
{
throw new ArgumentException("Handle, baseline, and output arrays must have matching lengths.");
}
int receivedCount = 0;
for (int index = 0; index < count; index++)
{
if (received[index])
{
receivedCount++;
}
}
while (true)
{
pumpStep();
for (int index = 0; index < count; index++)
{
if (received[index])
{
continue;
}
if (TryGet(serverHandle, itemHandles[index], out CachedValue value) && value.Version > sinceVersions[index])
{
received[index] = true;
values[index] = value;
receivedCount++;
}
}
if (receivedCount >= count)
{
return receivedCount;
}
TimeSpan remaining = deadlineUtc - DateTime.UtcNow;
if (remaining <= TimeSpan.Zero)
{
return receivedCount;
}
StaWaitHelper.WaitForSignalOrMessages(
updateSignal,
StaWaitHelper.ClampWaitMilliseconds(remaining, pollIntervalMs));
}
}
/// <summary>Returns the current version for a handle pair, or 0 if no entry exists.</summary>
/// <param name="serverHandle">MXAccess server handle.</param>
/// <param name="itemHandle">MXAccess item handle.</param>