diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs index ff3582fd..b8950db9 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs @@ -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. /// +/// +/// Upstream pushes coalesce through a dirty set. 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. 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: +/// RefillReadCache always reads , 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 was busy". +/// +/// +/// deliberately stays INSIDE the gate, before +/// the in-memory update. 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 +/// Store_save_failure_leaves_in_memory_state_unchanged. 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. +/// /// 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 _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 _dirtyAlarmIds = new(StringComparer.Ordinal); + private readonly object _dirtyLock = new(); + private bool _drainScheduled; // guarded by _dirtyLock + /// /// Initializes a new ScriptedAlarmEngine with the provided dependencies. /// @@ -497,10 +529,12 @@ public sealed class ScriptedAlarmEngine : IDisposable } /// - /// 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 - /// 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 can drain it. /// /// The upstream tag path that changed. /// The new data value snapshot. @@ -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)); + } + + /// + /// The single drainer for the upstream-push dirty set. Swaps the accumulated + /// alarm ids out under _dirtyLock and re-evaluates that batch, looping + /// until the set is empty — so every push that landed while the previous pass + /// held _evalGate is served by the next pass rather than by a queue of + /// redundant per-push evaluations. Exactly one instance runs at a time + /// (_drainScheduled), 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". + /// + 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. diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineTests.cs index 12786bb0..85f419d7 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineTests.cs @@ -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. + // ------------------------------------------------------------------------- + + /// 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. + [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); + } + + /// 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. + [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"); + } + + /// + /// 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 + /// ctx.Logger) — the store's save count cannot serve as that probe, because a + /// redundant re-evaluation produces no transition and therefore no save. + /// + private sealed class CountingSink : Serilog.Core.ILogEventSink + { + private readonly string _marker; + private int _count; + + /// Initializes the sink with the marker substring it counts. + /// Substring the message template must contain to be counted. + public CountingSink(string marker) => _marker = marker; + + /// Gets the number of matching events emitted since the last . + public int Count => Volatile.Read(ref _count); + + /// Resets the counter to zero. + public void Reset() => Interlocked.Exchange(ref _count, 0); + + /// Counts the event if its message template carries the marker. + /// The Serilog event being emitted. + public void Emit(Serilog.Events.LogEvent logEvent) + { + if (logEvent.MessageTemplate.Text.Contains(_marker, StringComparison.Ordinal)) + Interlocked.Increment(ref _count); + } + } + + /// + /// 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 _evalGate, then count what the released drain actually did). + /// + private sealed class CountingAlarmStateStore : IAlarmStateStore + { + private readonly InMemoryAlarmStateStore _inner = new(); + private int _saveCount; + + /// Gets the number of SaveAsync calls since the last . + public int SaveCount => Volatile.Read(ref _saveCount); + + /// Resets the save counter to zero. + public void ResetCount() => Interlocked.Exchange(ref _saveCount, 0); + + /// Gets or sets a TaskCompletionSource that, when set, blocks the next SaveAsync until signalled. + public TaskCompletionSource? BlockNextSave { get; set; } + + /// Gets a value indicating whether a SaveAsync call is currently blocked waiting on BlockNextSave. + public bool SaveInProgress { get; private set; } + + /// Loads an alarm condition state by ID from the inner store. + /// The ID of the alarm condition state to load. + /// A cancellation token. + public Task LoadAsync(string alarmId, CancellationToken ct) + => _inner.LoadAsync(alarmId, ct); + + /// Loads all alarm condition states from the inner store. + /// A cancellation token. + public Task> LoadAllAsync(CancellationToken ct) + => _inner.LoadAllAsync(ct); + + /// Counts the save, optionally blocking on the BlockNextSave gate, then delegates. + /// The alarm condition state to save. + /// A cancellation token. + 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); + } + + /// Removes an alarm condition state by ID from the inner store. + /// The ID of the alarm condition state to remove. + /// A cancellation token. + public Task RemoveAsync(string alarmId, CancellationToken ct) + => _inner.RemoveAsync(alarmId, ct); + } }