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

perf(alarms): dirty-set coalescing — one evaluation per alarm per drain, latest values win

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-16 04:02:20 -04:00
parent 2c708f5560
commit 5df793e79a
2 changed files with 293 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs
+95 -9
View File
@@ -29,6 +29,29 @@ namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
/// Operators investigating a broken predicate shouldn't see a phantom
/// clear-event preceding the failure.
/// </para>
/// <para>
/// <b>Upstream pushes coalesce through a dirty set.</b> A push does not spawn its own
/// re-evaluation: it marks the referencing alarm ids dirty and, if no drainer is already
/// running, starts one. <see cref="DrainDirtyAsync"/> then swaps the set out and evaluates
/// the batch, looping until the set is empty. N pushes arriving during one slow evaluation
/// therefore collapse into ONE further pass instead of N serialized ones that would each
/// re-read the same latest values. The collapse is loss-free by construction:
/// <c>RefillReadCache</c> always reads <see cref="_valueCache"/>, which every push has
/// already updated, so the surviving pass evaluates the latest values — latest-wins, never
/// a lost update. There is no timer and no wall-clock batching window; a push that lands on
/// an idle engine is evaluated immediately, and the coalescing window is exactly "whatever
/// queued while <see cref="_evalGate"/> was busy".
/// </para>
/// <para>
/// <b><see cref="IAlarmStateStore.SaveAsync"/> deliberately stays INSIDE the gate, before
/// the in-memory update.</b> The persist-before-update ordering is a real crash-consistency
/// property — a store failure must leave the in-memory condition and the persisted one both
/// at the prior state rather than diverging — and it is pinned by
/// <c>Store_save_failure_leaves_in_memory_state_unchanged</c>. Post-coalescing a save only
/// happens on a genuine Part 9 transition, which is rare relative to the push rate, so the
/// gate-held store call is no longer on the hot path and a write-behind queue would trade a
/// correctness property for throughput the engine does not need.
/// </para>
/// </remarks>
public sealed class ScriptedAlarmEngine : IDisposable
{
@@ -149,7 +172,7 @@ public sealed class ScriptedAlarmEngine : IDisposable
private bool _disposed;
// Tracks fire-and-forget background work launched by OnUpstreamChange
// (ReevaluateAsync) and RunShelvingCheck (ShelvingCheckAsync). Dispose drains
// (DrainDirtyAsync) and RunShelvingCheck (ShelvingCheckAsync). Dispose drains
// these so a re-evaluation in flight when shutdown begins finishes its
// SaveAsync before the engine returns control to the caller. The HashSet is
// accessed under its own lock — never under _evalGate — so registration /
@@ -157,6 +180,15 @@ public sealed class ScriptedAlarmEngine : IDisposable
private readonly HashSet<Task> _inFlight = [];
private readonly object _inFlightLock = new();
// Coalescing state for the upstream-push path. _dirtyAlarmIds accumulates the
// alarms a push made stale; _drainScheduled says a DrainDirtyAsync loop is
// already running and will pick them up, so a push storm starts exactly one
// drainer. Both are guarded by _dirtyLock — a leaf lock, never taken while
// holding _evalGate or _inFlightLock, and never held across an await.
private readonly HashSet<string> _dirtyAlarmIds = new(StringComparer.Ordinal);
private readonly object _dirtyLock = new();
private bool _drainScheduled; // guarded by _dirtyLock
/// <summary>
/// Initializes a new ScriptedAlarmEngine with the provided dependencies.
/// </summary>
@@ -497,10 +529,12 @@ public sealed class ScriptedAlarmEngine : IDisposable
}
/// <summary>
/// Upstream-change callback. Updates the value cache + enqueues predicate
/// re-evaluation for every alarm referencing the changed path. Fire-and-forget
/// so driver-side dispatch isn't blocked; the background task is tracked so
/// <see cref="Dispose"/> can drain it.
/// Upstream-change callback. Updates the value cache + marks every alarm
/// referencing the changed path dirty, starting the drainer if one isn't
/// already running. Never blocks driver-side dispatch, and never spawns work
/// per push: a push landing while a drain is in flight only grows the dirty
/// set, so a storm collapses into one further evaluation pass over the latest
/// values. The drainer is tracked so <see cref="Dispose"/> can drain it.
/// </summary>
/// <param name="path">The upstream tag path that changed.</param>
/// <param name="value">The new data value snapshot.</param>
@@ -508,9 +542,59 @@ public sealed class ScriptedAlarmEngine : IDisposable
{
_valueCache[path] = value;
if (_disposed) return; // don't queue new work against a disposing engine
if (_alarmsReferencing.TryGetValue(path, out var alarmIds))
if (!_alarmsReferencing.TryGetValue(path, out var alarmIds)) return;
bool startDrain;
lock (_dirtyLock)
{
TrackBackgroundTask(ReevaluateAsync(alarmIds.ToArray(), CancellationToken.None));
// Iterate the referencing set directly — the pre-coalescing path
// allocated a fresh string[] on every single push.
foreach (var id in alarmIds) _dirtyAlarmIds.Add(id);
startDrain = !_drainScheduled;
if (startDrain) _drainScheduled = true;
}
// Started outside the lock: DrainDirtyAsync runs synchronously up to its
// first real await, and nothing it does should happen under _dirtyLock.
if (startDrain) TrackBackgroundTask(DrainDirtyAsync(CancellationToken.None));
}
/// <summary>
/// The single drainer for the upstream-push dirty set. Swaps the accumulated
/// alarm ids out under <c>_dirtyLock</c> and re-evaluates that batch, looping
/// until the set is empty — so every push that landed while the previous pass
/// held <c>_evalGate</c> is served by the next pass rather than by a queue of
/// redundant per-push evaluations. Exactly one instance runs at a time
/// (<c>_drainScheduled</c>), and it clears that flag under the same lock it
/// reads the set under, so the last drainer out and the next push in can never
/// both decide "someone else will handle it".
/// </summary>
private async Task DrainDirtyAsync(CancellationToken ct)
{
while (true)
{
string[] batch;
lock (_dirtyLock)
{
// Bail promptly on a disposing engine, but ALWAYS release
// _drainScheduled on the way out — leaving it stuck true would make a
// hypothetical post-dispose push believe a drainer is coming. (Such
// pushes are already dropped in OnUpstreamChange; this keeps the flag
// honest regardless.)
if (_disposed || _dirtyAlarmIds.Count == 0)
{
_dirtyAlarmIds.Clear();
_drainScheduled = false;
return;
}
batch = [.. _dirtyAlarmIds];
_dirtyAlarmIds.Clear();
}
// Reused as-is: the gate / evaluate / persist / emit-outside-the-gate shape
// is already correct for a batch. Only the spawn-per-push is gone.
await ReevaluateAsync(batch, ct).ConfigureAwait(false);
}
}
@@ -925,9 +1009,11 @@ public sealed class ScriptedAlarmEngine : IDisposable
_shelvingTimer?.Dispose();
UnsubscribeFromUpstream();
// Drain any fire-and-forget background work (ReevaluateAsync from
// Drain any fire-and-forget background work (DrainDirtyAsync from
// OnUpstreamChange + ShelvingCheckAsync from the 5s timer) that started
// before _disposed = true was visible. Without this, a SaveAsync in
// before _disposed = true was visible. The drain loop observes _disposed
// at the top of every iteration and exits, so it cannot spin here. Without
// this, a SaveAsync in
// flight can outlive the engine and write to a (possibly disposed) store
// after Dispose() has returned. The tasks re-check _disposed after
// acquiring the gate and bail out, but the await still has to complete.
tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineTests.cs
+198
View File
@@ -1434,4 +1434,202 @@ public sealed class ScriptedAlarmEngineTests
.GetValue(evaluator)!;
return new WeakReference(del.Method.Module.Assembly);
}
// -------------------------------------------------------------------------
// Dirty-set coalescing (perf roadmap task 15).
//
// Pre-fix, every upstream push spawned its own fire-and-forget
// ReevaluateAsync, and each one queued on the single _evalGate. N pushes
// arriving during one slow evaluation therefore produced N serialized
// re-evaluations of the same alarms, all of them reading the SAME (latest)
// cached values — N-1 of them redundant by construction.
//
// Post-fix a push only marks its alarms dirty; a single drainer swaps the
// set and evaluates it once. The evaluation-count probe below is a
// ctx.Logger line inside the predicate counted by a Serilog sink: the store
// save-count alone cannot tell the two shapes apart, because the redundant
// passes produce no transition and therefore no save.
// -------------------------------------------------------------------------
/// <summary>Verifies that a storm of pushes arriving while an evaluation is in flight collapses
/// into a single further drain pass whose evaluation sees the LATEST pushed value, with a
/// falsification control proving the collapsed count comes from coalescing and not from dedup.</summary>
[Fact]
public async Task A_storm_of_pushes_coalesces_to_one_drain_pass()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
var probe = new CountingSink("EVALPROBE");
var logger = new LoggerConfiguration().WriteTo.Sink(probe).CreateLogger();
var store = new CountingAlarmStateStore();
using var eng = new ScriptedAlarmEngine(up, store, new ScriptLoggerFactory(logger), logger);
await eng.LoadAsync([
Alarm("A", """ctx.Logger.Information("EVALPROBE"); return (int)ctx.GetTag("Temp").Value > 100;"""),
], TestContext.Current.CancellationToken);
// LoadAsync itself evaluates once (startup recovery) — start counting after it.
probe.Reset();
store.ResetCount();
// Block the save the first push's evaluation performs. That evaluation then
// sits INSIDE _evalGate for as long as the test wants, which is the window
// every subsequent push has to coalesce into.
var saveGate = new TaskCompletionSource();
store.BlockNextSave = saveGate;
up.Push("Temp", 150); // evaluation #1 → Activated → blocks in SaveAsync
await WaitForAsync(() => store.SaveInProgress, timeoutMs: 3000);
// The storm: 10 further pushes while the gate is held. The LAST one drops
// back below the threshold, so "the drain evaluated the latest values" is
// observable as the alarm ending Inactive rather than Active.
for (var i = 1; i <= 9; i++) up.Push("Temp", 150 + i);
up.Push("Temp", 50);
saveGate.SetResult();
await WaitForAsync(() => eng.GetState("A")!.Active == AlarmActiveState.Inactive, timeoutMs: 3000);
probe.Count.ShouldBeLessThanOrEqualTo(2,
"11 pushes over one blocked evaluation must produce the in-flight evaluation plus at " +
"most ONE coalesced drain pass — pre-fix this was 11 serialized re-evaluations.");
store.SaveCount.ShouldBe(2,
"only genuine transitions persist: Activated (the blocked save) + Cleared (the drain).");
eng.GetState("A")!.Active.ShouldBe(AlarmActiveState.Inactive,
"the coalesced drain must evaluate the LATEST cached values — latest-wins, never lost-update.");
// --- Falsification control -------------------------------------------
// Spaced pushes (each drain completes before the next push lands) must
// still produce one evaluation EACH. If this collapsed too, the counts
// above would be proving dedup/short-circuiting rather than coalescing.
probe.Reset();
store.ResetCount();
foreach (var v in new[] { 150, 50, 150, 50 })
{
up.Push("Temp", v);
var want = v > 100 ? AlarmActiveState.Active : AlarmActiveState.Inactive;
await WaitForAsync(() => eng.GetState("A")!.Active == want, timeoutMs: 3000);
}
probe.Count.ShouldBe(4,
"four pushes that each land on an idle engine must each be evaluated — the collapsed " +
"count in the storm above comes from coalescing, not from the engine skipping work.");
store.SaveCount.ShouldBe(4);
}
/// <summary>Verifies that a push landing while a drain pass holds the evaluation gate is not
/// dropped — the drainer loops and evaluates it in a following pass.</summary>
[Fact]
public async Task A_push_during_a_drain_is_not_lost()
{
var up = new FakeUpstream();
up.Set("Temp", 50);
var probe = new CountingSink("EVALPROBE");
var logger = new LoggerConfiguration().WriteTo.Sink(probe).CreateLogger();
var store = new CountingAlarmStateStore();
using var eng = new ScriptedAlarmEngine(up, store, new ScriptLoggerFactory(logger), logger);
await eng.LoadAsync([
Alarm("A", """ctx.Logger.Information("EVALPROBE"); return (int)ctx.GetTag("Temp").Value > 100;"""),
], TestContext.Current.CancellationToken);
probe.Reset();
var saveGate = new TaskCompletionSource();
store.BlockNextSave = saveGate;
up.Push("Temp", 150); // drain pass #1 → Activated → blocks in SaveAsync
await WaitForAsync(() => store.SaveInProgress, timeoutMs: 3000);
// Lands while the drainer is inside _evalGate: it must be recorded in the
// dirty set and picked up by the drainer's next loop iteration, NOT dropped
// because "a drain is already scheduled".
up.Push("Temp", 40);
saveGate.SetResult();
await WaitForAsync(() => eng.GetState("A")!.Active == AlarmActiveState.Inactive, timeoutMs: 3000);
probe.Count.ShouldBe(2, "the in-flight pass plus exactly one follow-up pass for the late push");
}
/// <summary>
/// A Serilog sink that counts the events whose message template contains a marker string.
/// Used to count predicate EVALUATIONS (the predicate emits the marker through
/// <c>ctx.Logger</c>) — the store's save count cannot serve as that probe, because a
/// redundant re-evaluation produces no transition and therefore no save.
/// </summary>
private sealed class CountingSink : Serilog.Core.ILogEventSink
{
private readonly string _marker;
private int _count;
/// <summary>Initializes the sink with the marker substring it counts.</summary>
/// <param name="marker">Substring the message template must contain to be counted.</param>
public CountingSink(string marker) => _marker = marker;
/// <summary>Gets the number of matching events emitted since the last <see cref="Reset"/>.</summary>
public int Count => Volatile.Read(ref _count);
/// <summary>Resets the counter to zero.</summary>
public void Reset() => Interlocked.Exchange(ref _count, 0);
/// <summary>Counts the event if its message template carries the marker.</summary>
/// <param name="logEvent">The Serilog event being emitted.</param>
public void Emit(Serilog.Events.LogEvent logEvent)
{
if (logEvent.MessageTemplate.Text.Contains(_marker, StringComparison.Ordinal))
Interlocked.Increment(ref _count);
}
}
/// <summary>
/// Store decorator that counts SaveAsync calls and can hold the next save open until the
/// test releases it — the two capabilities the coalescing tests need together (block one
/// evaluation inside <c>_evalGate</c>, then count what the released drain actually did).
/// </summary>
private sealed class CountingAlarmStateStore : IAlarmStateStore
{
private readonly InMemoryAlarmStateStore _inner = new();
private int _saveCount;
/// <summary>Gets the number of SaveAsync calls since the last <see cref="ResetCount"/>.</summary>
public int SaveCount => Volatile.Read(ref _saveCount);
/// <summary>Resets the save counter to zero.</summary>
public void ResetCount() => Interlocked.Exchange(ref _saveCount, 0);
/// <summary>Gets or sets a TaskCompletionSource that, when set, blocks the next SaveAsync until signalled.</summary>
public TaskCompletionSource? BlockNextSave { get; set; }
/// <summary>Gets a value indicating whether a SaveAsync call is currently blocked waiting on BlockNextSave.</summary>
public bool SaveInProgress { get; private set; }
/// <summary>Loads an alarm condition state by ID from the inner store.</summary>
/// <param name="alarmId">The ID of the alarm condition state to load.</param>
/// <param name="ct">A cancellation token.</param>
public Task<AlarmConditionState?> LoadAsync(string alarmId, CancellationToken ct)
=> _inner.LoadAsync(alarmId, ct);
/// <summary>Loads all alarm condition states from the inner store.</summary>
/// <param name="ct">A cancellation token.</param>
public Task<IReadOnlyList<AlarmConditionState>> LoadAllAsync(CancellationToken ct)
=> _inner.LoadAllAsync(ct);
/// <summary>Counts the save, optionally blocking on the BlockNextSave gate, then delegates.</summary>
/// <param name="state">The alarm condition state to save.</param>
/// <param name="ct">A cancellation token.</param>
public async Task SaveAsync(AlarmConditionState state, CancellationToken ct)
{
Interlocked.Increment(ref _saveCount);
var gate = BlockNextSave;
if (gate is not null)
{
BlockNextSave = null;
SaveInProgress = true;
try { await gate.Task.WaitAsync(ct).ConfigureAwait(false); }
finally { SaveInProgress = false; }
}
await _inner.SaveAsync(state, ct).ConfigureAwait(false);
}
/// <summary>Removes an alarm condition state by ID from the inner store.</summary>
/// <param name="alarmId">The ID of the alarm condition state to remove.</param>
/// <param name="ct">A cancellation token.</param>
public Task RemoveAsync(string alarmId, CancellationToken ct)
=> _inner.RemoveAsync(alarmId, ct);
}
}