fix(virtual-tags): evaluate scripts off the Akka dispatcher thread (lmxopcua#779)
VirtualTagActor called the blocking IVirtualTagEvaluator.Evaluate, and RoslynVirtualTagEvaluator blocked on the run guard with .GetAwaiter().GetResult(), so every script run parked a dispatcher thread for up to its 2 s budget (a first-use Roslyn compile did too). This is the #750 P0 refactor for virtual tags. - IVirtualTagEvaluator.EvaluateAsync (default: a completed task over the snapshot overload, so the null evaluator and fakes are unchanged). - RoslynVirtualTagEvaluator implements it without blocking: passthrough and compile failures complete synchronously, a first-use compile goes to the pool, the run is awaited. The blocking overload now delegates to it. - VirtualTagActor: a completed task is applied in the same turn, as before; an incomplete one switches to an "evaluating" behaviour that stashes every dependency/reassert message until a private completion message arrives, then applies the result and unstashes. Each trigger is still evaluated against its own inputs, results are applied/published in trigger order and the dedup/Bad-transition state sees the same sequence. The dependency map is not touched while a run reads it. Calculation and scripted-alarm serialisation is NOT relaxed: neither blocks a thread, and running independent scripts in parallel is not semantics-neutral (pool consumption bound, cross-tag publish order, timer/pump exclusion for Calc; state-machine/persist boundary, transition timestamps and the store-failure path for alarms). Recorded in docs/VirtualTags.md and the #750 design's P0 entry. Tests: VirtualTagActorAsyncEvaluationTests — on a one-thread dispatcher a slow script no longer starves a sibling actor and the slow actor answers Identify (fails when the actor calls the blocking Evaluate); triggers during a run are evaluated afterwards in order against their own inputs; a reassert during a run is applied after it; a faulted evaluation takes the evaluator-threw path (the ordering/reassert tests fail with the stash removed). RoslynVirtualTagEvaluatorAsyncTests — a slow script and a cold compile return an incomplete task (fail without the override); cheap outcomes and timeouts match the blocking overload. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
+3
-1
@@ -96,7 +96,9 @@ Virtual-tag nodes are materialised by `AddressSpaceApplier` and fed by `VirtualT
|
||||
|
||||
`ITagUpstreamSource` and `IHistoryWriter` are the two ports value consumers require from their host. Both live in `Core.VirtualTags`. In the v2 actor system:
|
||||
|
||||
- **Upstream-tag feed.** `DependencyMuxActor` (`src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs`) routes the items of each `DriverInstanceActor.AttributeValueBatchPublished` to the `VirtualTagActor` instances that declared interest in those tag refs. Each `VirtualTagActor` holds the in-memory per-tag dependency map; the `IVirtualTagEvaluator` (`RoslynVirtualTagEvaluator`) receives the dependency snapshot synchronously on the actor message thread. Reads of never-pushed dependency refs return `null` values in the dependency snapshot.
|
||||
- **Upstream-tag feed.** `DependencyMuxActor` (`src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/DependencyMuxActor.cs`) routes the items of each `DriverInstanceActor.AttributeValueBatchPublished` to the `VirtualTagActor` instances that declared interest in those tag refs. Each `VirtualTagActor` holds the in-memory per-tag dependency map; the `IVirtualTagEvaluator` (`RoslynVirtualTagEvaluator`) receives the dependency snapshot through `EvaluateAsync`. Reads of never-pushed dependency refs return `null` values in the dependency snapshot.
|
||||
- **Evaluation is off the dispatcher thread (lmxopcua#779, #750 P0).** The actor used to call the blocking `Evaluate`, which parked an Akka dispatcher thread for the whole run — up to the 2 s budget per evaluation. It now calls `EvaluateAsync`: a passthrough, a compile failure or any synchronous evaluator completes in the same message turn as before, while a real script run (and a first-use compile, which is pushed to the pool) returns an incomplete task. Until it completes the actor **stashes** every further dependency / reassert message, then applies the result and unstashes in arrival order. So each trigger is still evaluated against the inputs as of that trigger, results are published in trigger order, and the dedup / Bad-transition state sees the same sequence — only the thread is freed. Pinned by `VirtualTagActorAsyncEvaluationTests` (a slow script on a one-thread dispatcher no longer starves a sibling actor; ordering, reassert and fault paths) and `RoslynVirtualTagEvaluatorAsyncTests`.
|
||||
- **Calculation tags and scripted alarms still evaluate one script at a time per driver / per engine (deliberately, for now).** Neither blocks a thread — both await the run — but a slow script still delays its siblings. Running independent scripts in parallel was considered for #779 and not done, because it is not semantics-neutral: for Calculation, the single pump is what bounds a driver to one pool thread of script work (per-tag lanes would let N slow tags hold N pool threads, the pool Akka's dispatchers also run on, so it needs a bounded-parallelism knob), it would reorder publishes across tags, and the timer path would need per-tag exclusion with the pump; for scripted alarms, the engine gate is also the consistency boundary of the Part 9 state machine, persist-before-update and ack/shelve/load, a predicate's transition time is sampled just before its own run, and a store failure mid-batch today leaves later predicates un-run (in parallel they would already have run, moving their run-guard strike streaks and writing script logs for results that are then discarded). The out-of-process runner's batch-submit `IScriptRunner` (#750) is where that concurrency belongs, with the runner owning the bound.
|
||||
- **`IHistoryWriter`** — the equipment-namespace path threads `Historize` end-to-end and `VirtualTagHostActor` invokes the injected writer on historized results (H5); the writer is resolved through `DriverHostActor` DI with a `NullHistoryWriter` default. No *durable* writer ships on this seam — durable continuous historization now flows through the separate `ContinuousHistorizationRecorder` → HistorianGateway `WriteLiveValues` path (see the `IHistoryWriter` section above and [Historian.md](Historian.md)).
|
||||
|
||||
## Composition
|
||||
|
||||
@@ -656,6 +656,10 @@ per-shard state, restarts in the last hour and the quarantine list. Quarantines
|
||||
- Gate: every existing Scripting / VirtualTags / Calculation / ScriptedAlarms suite stays green unchanged,
|
||||
plus new ordering tests for batch-then-apply.
|
||||
- This phase has value on its own: it ends the dispatcher-thread blocking.
|
||||
- **Partly landed in lmxopcua#779:** VT evaluation is async (`IVirtualTagEvaluator.EvaluateAsync` + stash in
|
||||
`VirtualTagActor`), so the dispatcher-thread blocking is gone. The Calc pump and the alarm engine still run one
|
||||
script at a time; why that was not relaxed in-process is recorded in `docs/VirtualTags.md` §"Upstream-tag feed".
|
||||
`IScriptRunner` and batch-submit remain to do.
|
||||
2. **P1, worker + protocol behind the flag.**
|
||||
- Ship `ScriptWorker`, `OutOfProcessScriptRunner` and the PE cache.
|
||||
- Default stays `InProcess`. Enable on the docker-dev rig only.
|
||||
|
||||
@@ -47,6 +47,29 @@ public interface IVirtualTagEvaluator
|
||||
foreach (var kv in dependencies) values[kv.Key] = kv.Value?.Value;
|
||||
return Evaluate(virtualTagId, expression, values);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Asynchronous form of the snapshot overload — what <c>VirtualTagActor</c> calls (lmxopcua#779). An
|
||||
/// implementation whose evaluation can take real time (the Roslyn one: a compile on first use, then a
|
||||
/// script run of up to its budget) returns an <b>incomplete</b> task instead of blocking the caller, so
|
||||
/// the actor's dispatcher thread is free while the script runs. The actor applies the result when the
|
||||
/// task completes, in trigger order.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Default-implemented as a synchronously completed task over the snapshot overload, so the null
|
||||
/// evaluator and every test fake keep their exact behaviour: a completed task is applied inline by the
|
||||
/// actor, in the same message turn, as before. Must not fault for script failures — those are
|
||||
/// <see cref="VirtualTagEvalResult.Failure"/>; a faulted task is treated as "evaluator threw".
|
||||
/// <para>The caller must not mutate <paramref name="dependencies"/> until the task completes: the
|
||||
/// implementation reads it on another thread while the script runs.</para>
|
||||
/// </remarks>
|
||||
/// <param name="virtualTagId">The unique identifier of the virtual tag being evaluated.</param>
|
||||
/// <param name="expression">The expression string to evaluate.</param>
|
||||
/// <param name="dependencies">Read-only dictionary of dependency snapshots, keyed by tag path.</param>
|
||||
/// <returns>The evaluation result.</returns>
|
||||
Task<VirtualTagEvalResult> EvaluateAsync(
|
||||
string virtualTagId, string expression, IReadOnlyDictionary<string, DataValueSnapshot> dependencies)
|
||||
=> Task.FromResult(Evaluate(virtualTagId, expression, dependencies));
|
||||
}
|
||||
|
||||
/// <summary>Result of one virtual-tag expression eval. Stash a Reason on every Failure so
|
||||
|
||||
@@ -11,8 +11,9 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Engines;
|
||||
/// F8b — production <see cref="IVirtualTagEvaluator"/> binding. Compiles each unique
|
||||
/// expression once via <see cref="ScriptEvaluator{TContext, TResult}"/> (Roslyn-backed
|
||||
/// sandbox) and caches the resulting evaluator keyed by source. Subsequent evaluations are
|
||||
/// in-process method invocations on the dependency dictionary — fast enough to run inline
|
||||
/// inside the actor's message handler.
|
||||
/// in-process method invocations on the dependency dictionary. The actor calls
|
||||
/// <see cref="EvaluateAsync"/>, which awaits the run instead of blocking its dispatcher thread
|
||||
/// (lmxopcua#779).
|
||||
///
|
||||
/// Single-tag mode: cross-tag <c>ctx.SetVirtualTag</c> writes are dropped (logged) because
|
||||
/// fan-out between actors is owned by <c>DependencyMuxActor</c>, not by the eval engine.
|
||||
@@ -98,8 +99,22 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac
|
||||
// actor holds evaluation entirely while any input is Bad, so what a script observes here is Good
|
||||
// or Uncertain.
|
||||
/// <inheritdoc />
|
||||
/// <remarks>Blocks the caller for up to the run budget. The runtime calls <see cref="EvaluateAsync"/> instead;
|
||||
/// this overload remains for single-shot callers.</remarks>
|
||||
public VirtualTagEvalResult Evaluate(
|
||||
string virtualTagId, string expression, IReadOnlyDictionary<string, DataValueSnapshot> dependencies)
|
||||
=> EvaluateAsync(virtualTagId, expression, dependencies).GetAwaiter().GetResult();
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks>
|
||||
/// lmxopcua#779 / #750 P0 — the virtual-tag actor used to call the synchronous overload, which blocked on the
|
||||
/// run guard with <c>.GetAwaiter().GetResult()</c> and parked an Akka dispatcher thread for up to the whole
|
||||
/// budget (2 s) per evaluation. This path never blocks: the passthrough fast path and compile failures
|
||||
/// complete synchronously (cheap), a first-use compile is pushed to the pool, and the script run is awaited.
|
||||
/// The outcomes are exactly those of the synchronous overload, which now delegates here.
|
||||
/// </remarks>
|
||||
public async Task<VirtualTagEvalResult> EvaluateAsync(
|
||||
string virtualTagId, string expression, IReadOnlyDictionary<string, DataValueSnapshot> dependencies)
|
||||
{
|
||||
if (_disposed) return VirtualTagEvalResult.Failure("evaluator disposed");
|
||||
if (string.IsNullOrWhiteSpace(expression)) return VirtualTagEvalResult.Failure("empty expression");
|
||||
@@ -141,7 +156,11 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac
|
||||
ScriptEvaluator<VirtualTagContext, object?> evaluator;
|
||||
try
|
||||
{
|
||||
evaluator = _cache.GetOrCompile(expression);
|
||||
// A first-use Roslyn compile costs tens to hundreds of milliseconds: never on the caller's thread
|
||||
// (an Akka dispatcher thread). A cached fetch is a dictionary hit and stays inline.
|
||||
evaluator = _cache.Contains(expression)
|
||||
? _cache.GetOrCompile(expression)
|
||||
: await Task.Run(() => _cache.GetOrCompile(expression)).ConfigureAwait(false);
|
||||
}
|
||||
catch (CompilationErrorException ex)
|
||||
{
|
||||
@@ -172,8 +191,8 @@ public sealed class RoslynVirtualTagEvaluator : IVirtualTagEvaluator, IScriptCac
|
||||
// it), so the wall-clock budget is what returns control — and the guard is what stops the
|
||||
// abandoned run from being joined by another on the next trigger (one orphan per tag, then
|
||||
// quarantine), which is the leak the old per-evaluation TimedScriptEvaluator allowed.
|
||||
var raw = _guard.RunAsync(virtualTagId, expression, evaluator, context, _runTimeout)
|
||||
.GetAwaiter().GetResult();
|
||||
var raw = await _guard.RunAsync(virtualTagId, expression, evaluator, context, _runTimeout)
|
||||
.ConfigureAwait(false);
|
||||
return VirtualTagEvalResult.Ok(raw);
|
||||
}
|
||||
catch (ScriptTimeoutException)
|
||||
|
||||
@@ -27,9 +27,29 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
/// multi-dep script does not flash Bad during deploy warm-up. Recovery to a Good value clears the
|
||||
/// Bad state and is force-published even when the value equals the pre-failure value.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Evaluation never blocks the dispatcher (lmxopcua#779, #750 P0).</b> The evaluator is called through
|
||||
/// <see cref="IVirtualTagEvaluator.EvaluateAsync"/>. A synchronously completed result (passthrough, the null
|
||||
/// evaluator, test fakes) is applied inline in the same message turn, exactly as before. An incomplete one — a
|
||||
/// Roslyn script run, up to its 2 s budget — switches the actor into an <i>evaluating</i> behaviour that
|
||||
/// <b>stashes</b> every dependency / reassert message, and the result comes back as a private completion
|
||||
/// message. Applying it unstashes everything in arrival order. So each trigger is still evaluated against
|
||||
/// the dependency map as it stood at that trigger, results are applied and published in trigger order, and
|
||||
/// the dedup / Bad-transition state sees exactly the sequence it saw when the call blocked — only the
|
||||
/// dispatcher thread is no longer held while the script runs. The dependency map is not touched while a run
|
||||
/// is in flight (the messages that would touch it are stashed), which is what makes handing it to the
|
||||
/// script on another thread safe.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class VirtualTagActor : ReceiveActor
|
||||
public sealed class VirtualTagActor : ReceiveActor, IWithUnboundedStash
|
||||
{
|
||||
/// <summary>The asynchronous evaluation started by <see cref="Evaluate"/> finished. Private to the actor.</summary>
|
||||
private sealed record EvaluationCompleted(
|
||||
Task<VirtualTagEvalResult> Evaluation, uint WorstInputStatus, DateTime TimestampUtc);
|
||||
|
||||
/// <inheritdoc />
|
||||
public IStash Stash { get; set; } = null!;
|
||||
|
||||
public const string ScriptLogsTopic = "script-logs";
|
||||
|
||||
/// <summary>A dependency's value changed. <paramref name="StatusCode"/> carries the source
|
||||
@@ -168,6 +188,9 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
Receive<DependencyValueChanged>(OnDependencyChanged);
|
||||
Receive<DependencyValueBatch>(OnDependencyBatch);
|
||||
Receive<ReassertValue>(_ => OnReassertValue());
|
||||
// Only reachable for a completion that outlived its behaviour (a supervision restart mid-evaluation
|
||||
// resets the behaviour stack): the incarnation it belonged to is gone, so it is dropped.
|
||||
Receive<EvaluationCompleted>(_ => { });
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -270,18 +293,66 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
return;
|
||||
}
|
||||
|
||||
VirtualTagEvalResult result;
|
||||
Task<VirtualTagEvalResult> evaluation;
|
||||
try
|
||||
{
|
||||
result = _evaluator.Evaluate(_virtualTagId, _expression, _dependencies);
|
||||
evaluation = _evaluator.EvaluateAsync(_virtualTagId, _expression, _dependencies);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.Warning(ex, "VirtualTag {Id}: evaluator threw", _virtualTagId);
|
||||
OnEvaluationFailed($"evaluator threw: {ex.Message}", "Error", timestampUtc);
|
||||
OnEvaluatorThrew(ex, timestampUtc);
|
||||
return;
|
||||
}
|
||||
|
||||
if (evaluation.IsCompleted)
|
||||
{
|
||||
// Synchronous evaluators (passthrough, null evaluator, fakes) and cheap failures: same turn, as before.
|
||||
ApplyEvaluation(evaluation, worst, timestampUtc);
|
||||
return;
|
||||
}
|
||||
|
||||
// A real script run is in flight: free the dispatcher thread, hold every later input until it lands.
|
||||
var self = Self;
|
||||
evaluation.ContinueWith(
|
||||
t => self.Tell(new EvaluationCompleted(t, worst, timestampUtc)),
|
||||
CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
|
||||
BecomeStacked(Evaluating);
|
||||
}
|
||||
|
||||
/// <summary>Behaviour while an evaluation is in flight: everything but its completion is stashed, so inputs and
|
||||
/// reasserts are processed after it, in arrival order, against the state it leaves behind.</summary>
|
||||
private void Evaluating()
|
||||
{
|
||||
Receive<EvaluationCompleted>(msg =>
|
||||
{
|
||||
UnbecomeStacked();
|
||||
ApplyEvaluation(msg.Evaluation, msg.WorstInputStatus, msg.TimestampUtc);
|
||||
Stash.UnstashAll();
|
||||
});
|
||||
ReceiveAny(_ => Stash.Stash());
|
||||
}
|
||||
|
||||
private void OnEvaluatorThrew(Exception ex, DateTime timestampUtc)
|
||||
{
|
||||
_log.Warning(ex, "VirtualTag {Id}: evaluator threw", _virtualTagId);
|
||||
OnEvaluationFailed($"evaluator threw: {ex.Message}", "Error", timestampUtc);
|
||||
}
|
||||
|
||||
/// <summary>Applies one completed evaluation: the failure tail, or the (value, status) dedup and the publish.</summary>
|
||||
/// <param name="evaluation">The completed evaluation task.</param>
|
||||
/// <param name="worst">The worst input status at the trigger.</param>
|
||||
/// <param name="timestampUtc">The trigger's timestamp.</param>
|
||||
private void ApplyEvaluation(Task<VirtualTagEvalResult> evaluation, uint worst, DateTime timestampUtc)
|
||||
{
|
||||
if (!evaluation.IsCompletedSuccessfully)
|
||||
{
|
||||
var ex = evaluation.Exception?.GetBaseException()
|
||||
?? new OperationCanceledException("virtual-tag evaluation was cancelled");
|
||||
OnEvaluatorThrew(ex, timestampUtc);
|
||||
return;
|
||||
}
|
||||
|
||||
var result = evaluation.Result;
|
||||
if (!result.Success)
|
||||
{
|
||||
OnEvaluationFailed(result.Reason ?? "evaluator failure", "Warning", timestampUtc);
|
||||
|
||||
+91
@@ -0,0 +1,91 @@
|
||||
using System.Diagnostics;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Serilog;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.Engines;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Engines;
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#779 / #750 P0 — <see cref="RoslynVirtualTagEvaluator.EvaluateAsync"/> is what the virtual-tag actor
|
||||
/// calls, and it must hand back an incomplete task for a slow script instead of blocking the caller (an Akka
|
||||
/// dispatcher thread) for the run. The result must be the one the blocking overload produces.
|
||||
/// </summary>
|
||||
public sealed class RoslynVirtualTagEvaluatorAsyncTests
|
||||
{
|
||||
private const string SlowScript =
|
||||
"""var end = System.DateTime.UtcNow.AddMilliseconds(800); while (System.DateTime.UtcNow < end) { } return (int)ctx.GetTag("a").Value + 1;""";
|
||||
|
||||
private static RoslynVirtualTagEvaluator Build() =>
|
||||
new(NullLogger<RoslynVirtualTagEvaluator>.Instance, new ScriptRootLogger(new LoggerConfiguration().CreateLogger()));
|
||||
|
||||
private static Dictionary<string, DataValueSnapshot> Deps(int a) => new(StringComparer.Ordinal)
|
||||
{
|
||||
["a"] = new DataValueSnapshot(a, 0u, DateTime.UtcNow, DateTime.UtcNow),
|
||||
};
|
||||
|
||||
[Fact]
|
||||
public async Task A_slow_script_returns_an_incomplete_task_without_blocking_the_caller()
|
||||
{
|
||||
using var sut = Build();
|
||||
IVirtualTagEvaluator evaluator = sut;
|
||||
sut.Evaluate("warm", SlowScript, Deps(1)).Success.ShouldBeTrue(); // compile + JIT out of the measurement
|
||||
|
||||
var sw = Stopwatch.StartNew();
|
||||
var task = evaluator.EvaluateAsync("vt", SlowScript, Deps(41));
|
||||
var returnedAfter = sw.Elapsed;
|
||||
|
||||
task.IsCompleted.ShouldBeFalse("the run is in flight on the pool, not on the caller's thread");
|
||||
returnedAfter.ShouldBeLessThan(TimeSpan.FromMilliseconds(400));
|
||||
|
||||
var result = await task;
|
||||
result.Success.ShouldBeTrue(result.Reason);
|
||||
result.Value.ShouldBe(42);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_first_use_compile_does_not_run_on_the_caller_thread()
|
||||
{
|
||||
using var sut = Build();
|
||||
IVirtualTagEvaluator evaluator = sut;
|
||||
|
||||
// A never-seen source: the Roslyn compile is tens to hundreds of ms and must not block the caller either.
|
||||
var task = evaluator.EvaluateAsync("vt-cold", """return (int)ctx.GetTag("a").Value * 3;""", Deps(5));
|
||||
task.IsCompleted.ShouldBeFalse();
|
||||
(await task).Value.ShouldBe(15);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Cheap_outcomes_complete_synchronously_with_the_same_result_as_the_blocking_overload()
|
||||
{
|
||||
using var sut = Build();
|
||||
IVirtualTagEvaluator evaluator = sut;
|
||||
|
||||
// Passthrough fast path: no Roslyn, no pool hop.
|
||||
var passthrough = evaluator.EvaluateAsync("vt-p", """return ctx.GetTag("a").Value;""", Deps(7));
|
||||
passthrough.IsCompleted.ShouldBeTrue();
|
||||
(await passthrough).Value.ShouldBe(7);
|
||||
|
||||
// Empty expression: a synchronous Failure, identical text to the blocking overload.
|
||||
var empty = await evaluator.EvaluateAsync("vt-e", " ", Deps(1));
|
||||
empty.ShouldBe(sut.Evaluate("vt-e", " ", Deps(1)));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_timeout_is_reported_exactly_as_the_blocking_overload_reports_it()
|
||||
{
|
||||
using var sut = new RoslynVirtualTagEvaluator(
|
||||
NullLogger<RoslynVirtualTagEvaluator>.Instance, new ScriptRootLogger(new LoggerConfiguration().CreateLogger()),
|
||||
scripting: new ScriptingOptions { RunTimeout = TimeSpan.FromMilliseconds(200) });
|
||||
IVirtualTagEvaluator evaluator = sut;
|
||||
sut.Evaluate("warm", "return 1;", Deps(1));
|
||||
|
||||
var result = await evaluator.EvaluateAsync("vt-slow", SlowScript, Deps(1));
|
||||
result.Success.ShouldBeFalse();
|
||||
result.Reason.ShouldBe($"{ScriptEvalFailureReasons.TimedOutPrefix}0.2s");
|
||||
}
|
||||
}
|
||||
+213
@@ -0,0 +1,213 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Akka.Actor;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.VirtualTags;
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#779 / #750 P0 — a slow virtual-tag script no longer holds an Akka dispatcher thread for its whole
|
||||
/// run (up to the 2 s budget), and moving the run off the thread changes neither the order results are applied
|
||||
/// and published in nor what each evaluation sees.
|
||||
/// </summary>
|
||||
public sealed class VirtualTagActorAsyncEvaluationTests : RuntimeActorTestBase
|
||||
{
|
||||
/// <summary>A dispatcher with exactly ONE thread, shared by every actor deployed on it — so an actor that blocks
|
||||
/// its thread starves its siblings, which is the failure this class pins.</summary>
|
||||
private const string OneThreadDispatcher = "one-thread-dispatcher";
|
||||
|
||||
private const string Hocon = @"
|
||||
one-thread-dispatcher {
|
||||
type = ForkJoinDispatcher
|
||||
throughput = 1
|
||||
dedicated-thread-pool {
|
||||
thread-count = 1
|
||||
threadtype = background
|
||||
}
|
||||
}";
|
||||
|
||||
public VirtualTagActorAsyncEvaluationTests() : base(AkkaTestHocon + Hocon)
|
||||
{
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_slow_script_does_not_starve_a_sibling_on_the_same_dispatcher_thread()
|
||||
{
|
||||
var slow = new GatedEvaluator(value: 1);
|
||||
try
|
||||
{
|
||||
var slowParent = CreateTestProbe();
|
||||
var fastParent = CreateTestProbe();
|
||||
var slowActor = slowParent.ChildActorOf(
|
||||
VirtualTagActor.Props("vt-slow", "slow", evaluator: slow).WithDispatcher(OneThreadDispatcher));
|
||||
var fastActor = fastParent.ChildActorOf(
|
||||
VirtualTagActor.Props("vt-fast", "fast", evaluator: new ConstEvaluator(42)).WithDispatcher(OneThreadDispatcher));
|
||||
|
||||
slowActor.Tell(new VirtualTagActor.DependencyValueChanged("a", 1, DateTime.UtcNow));
|
||||
AwaitCondition(() => slow.Calls == 1, PresenceBudget);
|
||||
|
||||
// The slow script is mid-run. Before the fix it held the ONLY thread of this dispatcher, so the sibling's
|
||||
// evaluation could not even start until the slow one finished.
|
||||
fastActor.Tell(new VirtualTagActor.DependencyValueChanged("b", 1, DateTime.UtcNow));
|
||||
fastParent.ExpectMsg<VirtualTagActor.EvaluationResult>(TimeSpan.FromSeconds(5)).Value.ShouldBe(42);
|
||||
|
||||
// And the slow actor itself still answers (Identify is handled by the actor cell on its own thread).
|
||||
var identity = await slowActor.Ask<ActorIdentity>(new Identify(7), TimeSpan.FromSeconds(5));
|
||||
identity.MessageId.ShouldBe(7);
|
||||
slowParent.ExpectNoMsg(TimeSpan.FromMilliseconds(100));
|
||||
|
||||
// Positive control: the slow evaluation still lands once its script returns.
|
||||
slow.Release();
|
||||
slowParent.ExpectMsg<VirtualTagActor.EvaluationResult>().Value.ShouldBe(1);
|
||||
}
|
||||
finally
|
||||
{
|
||||
slow.Release(); // never leave a dispatcher thread parked if the assertion above failed
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Triggers_during_a_run_are_evaluated_after_it_in_order_each_against_its_own_inputs()
|
||||
{
|
||||
var parent = CreateTestProbe();
|
||||
var evaluator = new ManualEvaluator();
|
||||
var actor = parent.ChildActorOf(VirtualTagActor.Props("vt-1", "expr", evaluator: evaluator));
|
||||
|
||||
var t1 = new DateTime(2026, 11, 9, 0, 0, 1, DateTimeKind.Utc);
|
||||
var t2 = t1.AddSeconds(1);
|
||||
var t3 = t1.AddSeconds(2);
|
||||
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 1, t1));
|
||||
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 2, t2));
|
||||
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 3, t3));
|
||||
|
||||
// One run at a time: the second and third triggers wait behind the first.
|
||||
AwaitCondition(() => evaluator.Calls.Count == 1, PresenceBudget);
|
||||
ExpectNoMsg(TimeSpan.FromMilliseconds(150));
|
||||
evaluator.Calls.Count.ShouldBe(1);
|
||||
|
||||
evaluator.Complete(0, VirtualTagEvalResult.Ok(10));
|
||||
var r1 = parent.ExpectMsg<VirtualTagActor.EvaluationResult>();
|
||||
AwaitCondition(() => evaluator.Calls.Count == 2, PresenceBudget);
|
||||
evaluator.Complete(1, VirtualTagEvalResult.Ok(20));
|
||||
var r2 = parent.ExpectMsg<VirtualTagActor.EvaluationResult>();
|
||||
AwaitCondition(() => evaluator.Calls.Count == 3, PresenceBudget);
|
||||
evaluator.Complete(2, VirtualTagEvalResult.Ok(30));
|
||||
var r3 = parent.ExpectMsg<VirtualTagActor.EvaluationResult>();
|
||||
|
||||
// Each evaluation saw the input as of ITS trigger — not the newest value — exactly as the blocking path did.
|
||||
evaluator.Calls.ShouldBe(new object?[] { 1, 2, 3 });
|
||||
new[] { r1.Value, r2.Value, r3.Value }.ShouldBe(new object?[] { 10, 20, 30 });
|
||||
new[] { r1.TimestampUtc, r2.TimestampUtc, r3.TimestampUtc }.ShouldBe(new[] { t1, t2, t3 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_reassert_during_a_run_is_applied_after_the_run_and_carries_its_result()
|
||||
{
|
||||
var parent = CreateTestProbe();
|
||||
var evaluator = new ManualEvaluator();
|
||||
var actor = parent.ChildActorOf(VirtualTagActor.Props("vt-1", "expr", evaluator: evaluator));
|
||||
|
||||
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 1, DateTime.UtcNow));
|
||||
AwaitCondition(() => evaluator.Calls.Count == 1, PresenceBudget);
|
||||
actor.Tell(VirtualTagActor.ReassertValue.Instance);
|
||||
parent.ExpectNoMsg(TimeSpan.FromMilliseconds(150)); // nothing published yet: the reassert waits its turn
|
||||
|
||||
evaluator.Complete(0, VirtualTagEvalResult.Ok(99));
|
||||
var fresh = parent.ExpectMsg<VirtualTagActor.EvaluationResult>();
|
||||
fresh.Value.ShouldBe(99);
|
||||
fresh.IsReassert.ShouldBeFalse();
|
||||
var reassert = parent.ExpectMsg<VirtualTagActor.EvaluationResult>();
|
||||
reassert.IsReassert.ShouldBeTrue();
|
||||
reassert.Value.ShouldBe(99, "the reassert re-emits what the run produced, as it did when the run blocked");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_faulted_evaluation_takes_the_evaluator_threw_path_and_publishes_Bad()
|
||||
{
|
||||
var parent = CreateTestProbe();
|
||||
var evaluator = new ManualEvaluator();
|
||||
var actor = parent.ChildActorOf(VirtualTagActor.Props("vt-1", "expr", evaluator: evaluator));
|
||||
|
||||
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 1, DateTime.UtcNow));
|
||||
AwaitCondition(() => evaluator.Calls.Count == 1, PresenceBudget);
|
||||
evaluator.Fault(0, new InvalidOperationException("boom"));
|
||||
|
||||
parent.ExpectMsg<VirtualTagActor.EvaluationResult>().StatusCode.ShouldBe(OpcUaStatus.BadInternalError);
|
||||
|
||||
// The actor left the evaluating behaviour: the next trigger is evaluated normally.
|
||||
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 2, DateTime.UtcNow));
|
||||
AwaitCondition(() => evaluator.Calls.Count == 2, PresenceBudget);
|
||||
evaluator.Complete(1, VirtualTagEvalResult.Ok(5));
|
||||
var recovered = parent.ExpectMsg<VirtualTagActor.EvaluationResult>();
|
||||
recovered.StatusCode.ShouldBe(OpcUaStatus.Good);
|
||||
recovered.Value.ShouldBe(5);
|
||||
}
|
||||
|
||||
/// <summary>Models the Roslyn evaluator: the blocking overload parks its caller until the script returns (what
|
||||
/// the actor used to call), the async one returns an incomplete task (what it calls now).</summary>
|
||||
private sealed class GatedEvaluator(object value) : IVirtualTagEvaluator
|
||||
{
|
||||
private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private int _calls;
|
||||
|
||||
public int Calls => Volatile.Read(ref _calls);
|
||||
|
||||
public void Release() => _gate.TrySetResult();
|
||||
|
||||
public VirtualTagEvalResult Evaluate(string virtualTagId, string expression, IReadOnlyDictionary<string, object?> dependencies)
|
||||
{
|
||||
Interlocked.Increment(ref _calls);
|
||||
_gate.Task.Wait(TimeSpan.FromSeconds(30));
|
||||
return VirtualTagEvalResult.Ok(value);
|
||||
}
|
||||
|
||||
public VirtualTagEvalResult Evaluate(
|
||||
string virtualTagId, string expression, IReadOnlyDictionary<string, DataValueSnapshot> dependencies)
|
||||
=> Evaluate(virtualTagId, expression, new Dictionary<string, object?>());
|
||||
|
||||
public Task<VirtualTagEvalResult> EvaluateAsync(
|
||||
string virtualTagId, string expression, IReadOnlyDictionary<string, DataValueSnapshot> dependencies)
|
||||
{
|
||||
Interlocked.Increment(ref _calls);
|
||||
return _gate.Task.ContinueWith(_ => VirtualTagEvalResult.Ok(value), TaskScheduler.Default);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Each call records the value of input <c>a</c> it was given and returns a task the test completes.</summary>
|
||||
private sealed class ManualEvaluator : IVirtualTagEvaluator
|
||||
{
|
||||
private readonly ConcurrentDictionary<int, TaskCompletionSource<VirtualTagEvalResult>> _pending = new();
|
||||
private readonly ConcurrentQueue<object?> _calls = new();
|
||||
|
||||
public IReadOnlyList<object?> Calls => _calls.ToArray();
|
||||
|
||||
public void Complete(int call, VirtualTagEvalResult result) => _pending[call].SetResult(result);
|
||||
|
||||
public void Fault(int call, Exception ex) => _pending[call].SetException(ex);
|
||||
|
||||
public VirtualTagEvalResult Evaluate(string virtualTagId, string expression, IReadOnlyDictionary<string, object?> dependencies)
|
||||
=> throw new InvalidOperationException("the actor must call EvaluateAsync");
|
||||
|
||||
public Task<VirtualTagEvalResult> EvaluateAsync(
|
||||
string virtualTagId, string expression, IReadOnlyDictionary<string, DataValueSnapshot> dependencies)
|
||||
{
|
||||
// Read the input NOW, as the script would at the start of its run.
|
||||
var a = dependencies.TryGetValue("a", out var snap) ? snap.Value : null;
|
||||
var tcs = new TaskCompletionSource<VirtualTagEvalResult>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
_pending[_calls.Count] = tcs;
|
||||
_calls.Enqueue(a);
|
||||
return tcs.Task;
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class ConstEvaluator(object value) : IVirtualTagEvaluator
|
||||
{
|
||||
public VirtualTagEvalResult Evaluate(string virtualTagId, string expression, IReadOnlyDictionary<string, object?> dependencies)
|
||||
=> VirtualTagEvalResult.Ok(value);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user