dohertj2/lmxopcua
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(s7): retire the bespoke poll fork onto PollGroupEngine v2 (CONV-1; fixes PERF-3 + STAB-6-S7; deletes the fork's bare-OCE poll-death — cross-ref plan R2-01)

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-13 12:07:39 -04:00
parent 0d879363f8
commit 85ef74ea86
3 changed files with 43 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json
+1 -1
View File
@@ -103,7 +103,7 @@
{
"id": "B3.6",
"subject": "B3: retire the S7 poll fork onto PollGroupEngine (delete PollLoopAsync/PollOnceAsync/HandlePollFailure/SubscriptionState; ComputeBackoffDelay -> ConnectionBackoff) \u2014 CONV-1; cross-ref R2-01",
"status": "pending",
"status": "completed",
"blockedBy": [
"B3.3",
"B3.4",
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs
+38 -201
View File
@@ -53,11 +53,21 @@ public sealed class S7Driver
_resolver = new EquipmentTagRefResolver<S7TagDefinition>(
r => _tagsByName.TryGetValue(r, out var t) ? t : null,
r => S7EquipmentTagParser.TryParse(r, out var d) ? d : null);
// CONV-1: the polling overlay runs on the shared PollGroupEngine (structural array diff +
// drain-before-dispose + capped-exponential failure backoff + caller-token-filtered OCE),
// retiring the bespoke S7 poll fork. minInterval 100 ms matches the S7 scan-mailbox floor;
// backoffCap 30 s matches the retired fork's PollBackoffCap.
_poll = new PollGroupEngine(
reader: ReadAsync,
onChange: (handle, tagRef, snapshot) =>
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)),
minInterval: TimeSpan.FromMilliseconds(100),
onError: HandlePollError,
backoffCap: TimeSpan.FromSeconds(30));
}
// ---- ISubscribable + IHostConnectivityProbe state ----
private readonly ConcurrentDictionary<long, SubscriptionState> _subscriptions = new();
private long _nextSubscriptionId;
private readonly PollGroupEngine _poll;
private readonly object _probeLock = new();
private HostState _hostState = HostState.Unknown;
private DateTime _hostStateChangedUtc = DateTime.UtcNow;
@@ -269,38 +279,28 @@ public sealed class S7Driver
// then await all of them with a bounded timeout BEFORE disposing the shared semaphore
// and CTS objects. Without the drain, a loop iteration mid-_gate would call Release()
// on (or WaitAsync against) a disposed semaphore.
var drain = new List<Task>();
// Drain the polling overlay first — the engine cancels each loop and awaits it before
// disposing its CTS (drain-before-dispose, STAB-6-S7), so no poll iteration touches _gate
// after this returns.
await _poll.DisposeAsync().ConfigureAwait(false);
var probeCts = _probeCts;
var probeTask = _probeTask;
try { probeCts?.Cancel(); } catch { }
if (probeTask is not null) drain.Add(probeTask);
var subscriptions = _subscriptions.Values.ToArray();
_subscriptions.Clear();
foreach (var state in subscriptions)
{
try { state.Cts.Cancel(); } catch { }
drain.Add(state.PollTask);
}
if (drain.Count > 0)
if (probeTask is not null)
{
try
{
await Task.WhenAll(drain).WaitAsync(DrainTimeout, CancellationToken.None)
.ConfigureAwait(false);
await probeTask.WaitAsync(DrainTimeout, CancellationToken.None).ConfigureAwait(false);
}
catch (TimeoutException) { /* a wedged loop — proceed; better than leaking the teardown */ }
catch (TimeoutException) { /* a wedged probe loop — proceed; better than leaking the teardown */ }
catch { /* loop faults are already surfaced via health; teardown continues */ }
}
// Loops have now observed cancellation and released _gate — safe to dispose the CTSs.
// The probe loop has now observed cancellation and released _gate — safe to dispose the CTS.
probeCts?.Dispose();
_probeCts = null;
_probeTask = null;
foreach (var state in subscriptions)
state.Cts.Dispose();
_initialized = false;
_plcDead = false;
@@ -1397,180 +1397,29 @@ public sealed class S7Driver
/// <inheritdoc />
public Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
var id = Interlocked.Increment(ref _nextSubscriptionId);
var cts = new CancellationTokenSource();
// Floor at 100 ms — S7 CPUs scan 2-10 ms but the comms mailbox is processed at most
// once per scan; sub-100 ms polling just queues wire-side with worse latency.
var interval = publishingInterval < TimeSpan.FromMilliseconds(100)
? TimeSpan.FromMilliseconds(100)
: publishingInterval;
var handle = new S7SubscriptionHandle(id);
var state = new SubscriptionState(handle, [.. fullReferences], interval, cts);
_subscriptions[id] = state;
// Track the poll Task so ShutdownAsync can await it after cancelling — a poll
// iteration mid-_gate would otherwise race the semaphore's disposal.
state.PollTask = Task.Run(() => PollLoopAsync(state, cts.Token), CancellationToken.None);
return Task.FromResult<ISubscriptionHandle>(handle);
}
=> Task.FromResult(_poll.Subscribe(fullReferences, publishingInterval));
/// <inheritdoc />
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
{
if (handle is S7SubscriptionHandle h && _subscriptions.TryRemove(h.Id, out var state))
{
state.Cts.Cancel();
state.Cts.Dispose();
}
_poll.Unsubscribe(handle);
return Task.CompletedTask;
}
/// <summary>
/// Upper bound on the poll-loop backoff window. After enough consecutive failures the
/// loop waits this long between retries instead of <see cref="SubscriptionState.Interval"/>,
/// so a subscription against a dropped / uninitialised driver doesn't spin.
/// 05/STAB-9 — routes a poll-loop reader failure (surfaced by the shared
/// <see cref="PollGroupEngine"/>'s onError sink) to the driver health surface: logs it and
/// degrades to <see cref="DriverState.Degraded"/> preserving <c>LastSuccessfulRead</c>. Never
/// downgrades a <see cref="DriverState.Faulted"/> state (e.g. PUT/GET-denied set by ReadAsync)
/// — Faulted is the stronger, permanent-config signal. Replaces the retired fork's
/// <c>HandlePollFailure</c>; the engine now owns the consecutive-failure count + backoff.
/// </summary>
private static readonly TimeSpan PollBackoffCap = TimeSpan.FromSeconds(30);
/// <summary>
/// Number of consecutive poll failures before the loop transitions the driver's
/// health to <see cref="DriverState.Degraded"/>. One stray failure can be transient;
/// a sustained run indicates the operator should see it. Threshold of 1 because the
/// first failure already lives in the LastError surface.
/// </summary>
private const int PollFailureHealthThreshold = 1;
private async Task PollLoopAsync(SubscriptionState state, CancellationToken ct)
/// <param name="ex">The exception caught by the poll engine.</param>
private void HandlePollError(Exception ex)
{
var consecutiveFailures = 0;
// Initial-data push per OPC UA Part 4 convention.
try
{
await PollOnceAsync(state, forceRaise: true, ct).ConfigureAwait(false);
consecutiveFailures = 0;
}
// Only a teardown OCE (the loop's own token) exits the loop; any other OCE (e.g. a
// connect-timeout that slipped past the wrapper filter) falls to the backoff path below
// so the loop lives (STAB-14 defense-in-depth).
catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }
catch (Exception ex)
{
// First-read error — polling continues; log so the operator has an event trail.
consecutiveFailures++;
HandlePollFailure(ex, consecutiveFailures, initial: true);
}
while (!ct.IsCancellationRequested)
{
// Capped exponential backoff: Interval, 2×, 4×, ... up to PollBackoffCap. Healthy
// ticks reset consecutiveFailures back to 0 so the cadence snaps back to Interval.
var delay = ComputeBackoffDelay(state.Interval, consecutiveFailures);
try { await Task.Delay(delay, ct).ConfigureAwait(false); }
// Task.Delay can only observe ct-triggered OCE, so this filter is purely for uniformity.
catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }
try
{
await PollOnceAsync(state, forceRaise: false, ct).ConfigureAwait(false);
consecutiveFailures = 0;
}
// Only teardown exits; any non-teardown OCE (e.g. a slipped-through connect timeout)
// backs off instead of killing the loop (STAB-14 defense-in-depth).
catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }
catch (Exception ex)
{
// Sustained polling error — loop continues with backoff; log + update health.
consecutiveFailures++;
HandlePollFailure(ex, consecutiveFailures, initial: false);
}
}
}
/// <summary>
/// Logs the swallowed poll exception and, once <see cref="PollFailureHealthThreshold"/>
/// consecutive failures have accumulated, degrades the driver health so the failure
/// surfaces on the dashboard. The probe loop owns Running/Stopped
/// transitions for the host-connectivity surface, so we touch <see cref="_health"/>
/// rather than the probe state.
/// </summary>
private void HandlePollFailure(Exception ex, int consecutiveFailures, bool initial)
{
if (initial)
_logger.LogWarning(ex, "S7 poll initial-read failed. Driver={DriverInstanceId} ConsecutiveFailures={Count}",
_driverInstanceId, consecutiveFailures);
else
_logger.LogWarning(ex, "S7 poll tick failed. Driver={DriverInstanceId} ConsecutiveFailures={Count}",
_driverInstanceId, consecutiveFailures);
if (consecutiveFailures >= PollFailureHealthThreshold)
{
// Don't downgrade a Faulted state (e.g. PUT/GET-denied set by ReadAsync) — Faulted
// is a stronger signal than Degraded and is reserved for permanent config faults.
if (_health.State != DriverState.Faulted)
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
}
}
/// <summary>
/// Capped exponential backoff. <c>consecutiveFailures == 0</c> returns the configured
/// <paramref name="interval"/>; each subsequent failure doubles the wait up to
/// <see cref="PollBackoffCap"/>. Computed in ticks to avoid overflow at large counts.
/// </summary>
/// <param name="interval">Base polling interval.</param>
/// <param name="consecutiveFailures">Number of consecutive failures.</param>
/// <returns>The computed backoff delay.</returns>
internal static TimeSpan ComputeBackoffDelay(TimeSpan interval, int consecutiveFailures)
{
if (consecutiveFailures <= 0) return interval;
// Cap the shift to avoid overflow — at 30 the result already saturates PollBackoffCap
// for any reasonable Interval.
var shift = Math.Min(consecutiveFailures - 1, 30);
var ticks = interval.Ticks << shift;
if (ticks <= 0 || ticks > PollBackoffCap.Ticks) return PollBackoffCap;
return TimeSpan.FromTicks(ticks);
}
private async Task PollOnceAsync(SubscriptionState state, bool forceRaise, CancellationToken ct)
{
#pragma warning disable OTOPCUA0001 // Driver-INTERNAL self-call: S7Driver's own poll loop reading via its own ReadAsync; CapabilityInvoker wraps the driver from the dispatch layer OUTWARD, so the driver's internal poll must NOT re-wrap. Intentional, not a dispatch gap.
var snapshots = await ReadAsync(state.TagReferences, ct).ConfigureAwait(false);
#pragma warning restore OTOPCUA0001
for (var i = 0; i < state.TagReferences.Count; i++)
{
var tagRef = state.TagReferences[i];
var current = snapshots[i];
var lastSeen = state.LastValues.TryGetValue(tagRef, out var prev) ? prev : default;
if (forceRaise || !Equals(lastSeen?.Value, current.Value) || lastSeen?.StatusCode != current.StatusCode)
{
state.LastValues[tagRef] = current;
OnDataChange?.Invoke(this, new DataChangeEventArgs(state.Handle, tagRef, current));
}
}
}
private sealed record SubscriptionState(
S7SubscriptionHandle Handle,
IReadOnlyList<string> TagReferences,
TimeSpan Interval,
CancellationTokenSource Cts)
{
/// <summary>Gets the last known values for subscribed tags.</summary>
public ConcurrentDictionary<string, DataValueSnapshot> LastValues { get; }
= new(StringComparer.OrdinalIgnoreCase);
/// <summary>
/// Handle to this subscription's poll loop. Tracked so <see cref="ShutdownAsync"/>
/// can await it after cancelling.
/// </summary>
public Task PollTask { get; set; } = Task.CompletedTask;
}
private sealed record S7SubscriptionHandle(long Id) : ISubscriptionHandle
{
/// <inheritdoc />
public string DiagnosticId => $"s7-sub-{Id}";
_logger.LogWarning(ex, "S7 poll reader failed. Driver={DriverInstanceId}", _driverInstanceId);
if (_health.State != DriverState.Faulted)
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
}
// ---- IHostConnectivityProbe ----
@@ -1693,34 +1542,22 @@ public sealed class S7Driver
/// </summary>
private void SynchronousTeardown()
{
var drain = new List<Task>();
// Drain the polling overlay (drain-before-dispose, STAB-6-S7).
try { _poll.DisposeAsync().AsTask().Wait(DrainTimeout); }
catch { /* timeouts/loop faults are tolerated — teardown continues */ }
var probeCts = _probeCts;
var probeTask = _probeTask;
try { probeCts?.Cancel(); } catch { }
if (probeTask is not null) drain.Add(probeTask);
var subscriptions = _subscriptions.Values.ToArray();
_subscriptions.Clear();
foreach (var state in subscriptions)
if (probeTask is not null)
{
try { state.Cts.Cancel(); } catch { }
drain.Add(state.PollTask);
}
if (drain.Count > 0)
{
try { Task.WhenAll(drain).Wait(DrainTimeout); }
try { probeTask.Wait(DrainTimeout); }
catch { /* timeouts/loop faults are tolerated — teardown continues */ }
}
probeCts?.Dispose();
_probeCts = null;
_probeTask = null;
foreach (var state in subscriptions)
{
try { state.Cts.Dispose(); } catch { }
}
_initialized = false;
_plcDead = false;
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DiscoveryAndSubscribeTests.cs
+4 -2
View File
@@ -91,8 +91,10 @@ public sealed class S7DiscoveryAndSubscribeTests
var h1 = await drv.SubscribeAsync(["T1"], TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
var h2 = await drv.SubscribeAsync(["T2"], TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
h1.DiagnosticId.ShouldStartWith("s7-sub-");
h2.DiagnosticId.ShouldStartWith("s7-sub-");
// Post poll-fork retirement (CONV-1) the subscription handle is the shared engine's
// PollSubscriptionHandle (`poll-sub-<id>`), not the retired S7SubscriptionHandle (`s7-sub-`).
h1.DiagnosticId.ShouldStartWith("poll-sub-");
h2.DiagnosticId.ShouldStartWith("poll-sub-");
h1.DiagnosticId.ShouldNotBe(h2.DiagnosticId);
await drv.UnsubscribeAsync(h1, TestContext.Current.CancellationToken);