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

perf(site-runtime): AlarmActor trigger expressions evaluate on the script scheduler (P1)

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-08 23:59:54 -04:00
parent 6baf5a1d45
commit bd89c1474e
2 changed files with 166 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/AlarmActor.cs
+120 -55
View File
@@ -86,6 +86,12 @@ public class AlarmActor : ReceiveActor
// expression on the hot path.
private readonly Script<object?>? _compiledTriggerExpression;
private readonly Dictionary<string, object?> _attributeSnapshot = new();
// Coalescing guard for off-dispatcher expression evaluation (P1): at most one
// evaluation runs on the script scheduler at a time; a change arriving while
// one is in flight sets _evalPending so exactly one more runs afterwards
// against the latest snapshot. Both flags are touched only on the actor thread.
private bool _evalInFlight;
private bool _evalPending;
/// <summary>
/// The exact dictionary instance this actor was seeded from
@@ -177,6 +183,9 @@ public class AlarmActor : ReceiveActor
// Handle alarm execution completion
Receive<AlarmExecutionCompleted>(_ =>
_logger.LogDebug("Alarm {Alarm} execution completed on {Instance}", _alarmName, _instanceName));
// Handle the off-dispatcher trigger-expression evaluation result (P1).
Receive<ExpressionEvalResult>(HandleExpressionEvalResult);
}
/// <inheritdoc />
@@ -230,52 +239,25 @@ public class AlarmActor : ReceiveActor
return;
}
// Expression triggers evaluate off the dispatcher on the script
// scheduler (P1) — the result comes back as ExpressionEvalResult and
// drives ApplyTriggeredState there. All other trigger types are cheap
// synchronous comparisons and keep the inline path.
if (_triggerType == AlarmTriggerType.Expression)
{
StartExpressionEvaluation();
return;
}
var isTriggered = _triggerType switch
{
AlarmTriggerType.ValueMatch => EvaluateValueMatch(changed.Value),
AlarmTriggerType.RangeViolation => EvaluateRangeViolation(changed.Value),
AlarmTriggerType.RateOfChange => EvaluateRateOfChange(changed.Value, changed.Timestamp),
AlarmTriggerType.Expression => EvaluateExpression(),
_ => false
};
if (isTriggered && _currentState == AlarmState.Normal)
{
// Transition: Normal → Active
_currentState = AlarmState.Active;
_logger.LogInformation(
"Alarm {Alarm} ACTIVATED on instance {Instance}",
_alarmName, _instanceName);
// Notify Instance Actor of alarm state change
var alarmChanged = new AlarmStateChanged(
_instanceName, _alarmName, AlarmState.Active, _priority, DateTimeOffset.UtcNow);
_instanceActor.Tell(alarmChanged);
// Operational `alarm` event — raise. Severity by priority.
LogAlarmEvent(RaiseSeverity(_priority), $"Alarm {_alarmName} activated (priority {_priority})");
// Spawn AlarmExecutionActor if on-trigger script defined
if (_onTriggerCompiledScript != null)
{
SpawnAlarmExecution(AlarmLevel.None, _priority, string.Empty);
}
}
else if (!isTriggered && _currentState == AlarmState.Active)
{
// Transition: Active → Normal (no script on clear)
_currentState = AlarmState.Normal;
_logger.LogInformation(
"Alarm {Alarm} CLEARED on instance {Instance}",
_alarmName, _instanceName);
var alarmChanged = new AlarmStateChanged(
_instanceName, _alarmName, AlarmState.Normal, _priority, DateTimeOffset.UtcNow);
_instanceActor.Tell(alarmChanged);
// Operational `alarm` event — return to normal.
LogAlarmEvent("Info", $"Alarm {_alarmName} cleared");
}
ApplyTriggeredState(isTriggered);
}
catch (Exception ex)
{
@@ -287,6 +269,53 @@ public class AlarmActor : ReceiveActor
}
}
/// <summary>
/// Applies the binary Normal↔Active transition for the current triggered
/// value. Shared by the synchronous trigger types and the asynchronous
/// Expression path (via <see cref="HandleExpressionEvalResult"/>). Edge state
/// (<see cref="_currentState"/>) is touched only on the actor thread.
/// </summary>
private void ApplyTriggeredState(bool isTriggered)
{
if (isTriggered && _currentState == AlarmState.Normal)
{
// Transition: Normal → Active
_currentState = AlarmState.Active;
_logger.LogInformation(
"Alarm {Alarm} ACTIVATED on instance {Instance}",
_alarmName, _instanceName);
// Notify Instance Actor of alarm state change
var alarmChanged = new AlarmStateChanged(
_instanceName, _alarmName, AlarmState.Active, _priority, DateTimeOffset.UtcNow);
_instanceActor.Tell(alarmChanged);
// Operational `alarm` event — raise. Severity by priority.
LogAlarmEvent(RaiseSeverity(_priority), $"Alarm {_alarmName} activated (priority {_priority})");
// Spawn AlarmExecutionActor if on-trigger script defined
if (_onTriggerCompiledScript != null)
{
SpawnAlarmExecution(AlarmLevel.None, _priority, string.Empty);
}
}
else if (!isTriggered && _currentState == AlarmState.Active)
{
// Transition: Active → Normal (no script on clear)
_currentState = AlarmState.Normal;
_logger.LogInformation(
"Alarm {Alarm} CLEARED on instance {Instance}",
_alarmName, _instanceName);
var alarmChanged = new AlarmStateChanged(
_instanceName, _alarmName, AlarmState.Normal, _priority, DateTimeOffset.UtcNow);
_instanceActor.Tell(alarmChanged);
// Operational `alarm` event — return to normal.
LogAlarmEvent("Info", $"Alarm {_alarmName} cleared");
}
}
/// <summary>
/// HiLo state machine: emit an AlarmStateChanged whenever the evaluated
/// level changes. Spawns the on-trigger script only on the Normal→Active
@@ -488,34 +517,63 @@ public class AlarmActor : ReceiveActor
/// as an alarm error) so that the state machine still runs — an Active
/// alarm correctly clears if the expression starts throwing.
/// </summary>
private bool EvaluateExpression()
private void StartExpressionEvaluation()
{
if (_compiledTriggerExpression == null) return false;
if (_compiledTriggerExpression == null) return;
if (_evalInFlight) { _evalPending = true; return; } // coalesce bursts: one in flight, one pending
_evalInFlight = true;
var snapshot = new Dictionary<string, object?>(_attributeSnapshot); // point-in-time copy, actor thread
var expression = _compiledTriggerExpression;
var self = Self;
Task.Factory.StartNew(async () =>
{
try
{
// Bound evaluation with a short timeout. The CancellationToken covers
// cooperative/async cases; a pathological CPU-bound expression is not
// fully interruptible — acceptable because trigger expressions are
// authored by trusted Design-role users and compile-checked pre-deploy.
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2));
var state = await expression.RunAsync(new TriggerExpressionGlobals(snapshot), cancellationToken: cts.Token);
return state.ReturnValue is bool b && b;
}
catch (Exception ex)
{
// OperationCanceledException (timeout) falls through here too and is
// treated as false. _healthCollector (Interlocked) and _logger are
// thread-safe, so this catch is safe off the actor thread.
_healthCollector?.IncrementAlarmError();
_logger.LogError(ex,
"Alarm {Alarm} trigger expression evaluation failed on {Instance}; treated as false",
_alarmName, _instanceName);
return false;
}
}, CancellationToken.None, TaskCreationOptions.DenyChildAttach,
ScriptExecutionScheduler.Shared(_options)).Unwrap().PipeTo(self,
success: r => new ExpressionEvalResult(r));
}
/// <summary>
/// Applies an off-dispatcher expression result on the actor thread: runs the
/// Normal↔Active transition, clears the in-flight flag, and drains a coalesced
/// pending evaluation (against the now-current snapshot) if one was requested.
/// </summary>
private void HandleExpressionEvalResult(ExpressionEvalResult msg)
{
_evalInFlight = false;
try
{
var globals = new TriggerExpressionGlobals(_attributeSnapshot);
// Bound evaluation with a short timeout. The CancellationToken
// covers cooperative/async cases; a pathological CPU-bound
// expression is not fully interruptible. Acceptable because
// trigger expressions are authored by trusted Design-role users
// and are compile-checked pre-deployment.
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2));
var state = _compiledTriggerExpression
.RunAsync(globals, cancellationToken: cts.Token)
.GetAwaiter().GetResult();
return state.ReturnValue is bool b && b;
ApplyTriggeredState(msg.Result);
}
catch (Exception ex)
{
// OperationCanceledException (timeout) falls through here too,
// and is correctly treated as false.
_healthCollector?.IncrementAlarmError();
_logger.LogError(ex,
"Alarm {Alarm} trigger expression evaluation failed on {Instance}; treated as false",
"Alarm {Alarm} state transition error on {Instance}",
_alarmName, _instanceName);
return false;
}
if (_evalPending) { _evalPending = false; StartExpressionEvaluation(); }
}
/// <summary>
@@ -730,6 +788,13 @@ public class AlarmActor : ReceiveActor
// ── Internal messages ──
internal record AlarmExecutionCompleted(string AlarmName, bool Success);
/// <summary>
/// Piped back to self from an off-dispatcher trigger-expression evaluation (P1);
/// carries the boolean truth value so the Normal↔Active transition runs on the
/// actor thread.
/// </summary>
private sealed record ExpressionEvalResult(bool Result);
}
internal enum RateOfChangeDirection { Either, Rising, Falling }
tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/AlarmActorTests.cs
+46
View File
@@ -1,6 +1,8 @@
using Akka.Actor;
using Akka.TestKit;
using Akka.TestKit.Xunit2;
using Microsoft.CodeAnalysis.CSharp.Scripting;
using Microsoft.CodeAnalysis.Scripting;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
@@ -1008,4 +1010,48 @@ public class AlarmActorTests : TestKit, IDisposable
instanceProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
}
// ── P1: Expression-alarm evaluation runs off the dispatcher ──
/// <summary>
/// Compiles a trigger expression WITHOUT the trust validator so a test can use
/// <c>System.Threading.Thread</c> to observe which thread the evaluation runs on.
/// </summary>
private static Script<object?> CompileRawTriggerExpression(string expression)
{
var opts = ScriptOptions.Default
.WithReferences(typeof(object).Assembly, typeof(Enumerable).Assembly)
.WithImports("System", "System.Collections.Generic", "System.Linq", "System.Threading.Tasks");
var s = CSharpScript.Create<object?>(expression, opts, typeof(TriggerExpressionGlobals));
s.Compile();
return s;
}
[Fact]
public void ExpressionAlarm_EvaluatesOnSchedulerThread_AndActivates()
{
// TRUE only when evaluated on a script-execution thread. Before P1 the
// expression ran on the actor dispatcher (name is NOT "script-execution-*")
// → false → alarm never activates. After P1 it runs on the script scheduler.
var expr = CompileRawTriggerExpression(
"System.Threading.Thread.CurrentThread.Name != null && " +
"System.Threading.Thread.CurrentThread.Name.StartsWith(\"script-execution-\")");
var alarmConfig = new ResolvedAlarm
{
CanonicalName = "ExprAlarm",
TriggerType = "Expression",
TriggerConfiguration = "{\"expression\":\"true\"}",
PriorityLevel = 1
};
var instanceProbe = CreateTestProbe();
var alarm = ActorOf(Props.Create(() => new AlarmActor(
"ExprAlarm", "Pump1", instanceProbe.Ref, alarmConfig,
null, _sharedLibrary, _options,
NullLogger<AlarmActor>.Instance, expr)));
alarm.Tell(new AttributeValueChanged("Pump1", "A", "A", 1, "Good", DateTimeOffset.UtcNow));
var change = instanceProbe.ExpectMsg<AlarmStateChanged>(TimeSpan.FromSeconds(10));
Assert.Equal(AlarmState.Active, change.State);
}
}