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

Merge fix/798-shadow-timing-proof

Browse Source
This commit is contained in:
Joseph Doherty
2026-11-10 21:24:38 -05:00
parent 606a22daf4 304d1f1e6c
commit e6781bc5bc
3 changed files with 214 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ContainmentTests.cs
+7 -3
View File
@@ -61,10 +61,14 @@ public sealed class ContainmentTests
await using var runner = Scripts.Runner(Scripts.Options(hardGrace: grace, quarantineAfterKills: 2));
(await runner.RunAsync(Scripts.Vt("sibling", Healthy), Ct)).Value.ShouldBe(7);
// Compile the runaway's source first (it only spins when told to), so the stopwatch below times the soft answer
// and the kill, not the worker's first compile of new script text (lmxopcua#798's flaw, in its runner form).
const string runaway = "if ((int)ctx.GetTag(\"spin\").Value == 1) { while (true) { } } return 0;";
(await runner.RunAsync(Scripts.Vt("runaway", runaway, softT, Scripts.Input("spin", 0)), Ct)).Value.ShouldBe(0);
var worker = runner.Shards[0].Current.ShouldNotBeNull();
var sw = Stopwatch.StartNew();
var outcome = await runner.RunAsync(Scripts.Vt("runaway", Runaway, softT), Ct);
var outcome = await runner.RunAsync(Scripts.Vt("runaway", runaway, softT, Scripts.Input("spin", 1)), Ct);
var softElapsed = sw.Elapsed;
outcome.Status.ShouldBe(ScriptRunStatus.Failed);
outcome.Reason!.ShouldStartWith("script timed out after ");
@@ -90,12 +94,12 @@ public sealed class ContainmentTests
// A second kill quarantines (QuarantineAfterKills = 2); the third trigger is refused without running.
var second = runner.Shards[0].Current!;
(await runner.RunAsync(Scripts.Vt("runaway", Runaway, softT), Ct)).Status.ShouldBe(ScriptRunStatus.Failed);
(await runner.RunAsync(Scripts.Vt("runaway", runaway, softT, Scripts.Input("spin", 1)), Ct)).Status.ShouldBe(ScriptRunStatus.Failed);
(await second.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct)).Reason.ShouldBe(WorkerExitReason.KilledDeadline);
// The strike is recorded by the shard's exit handler, which runs asynchronously to the Exited completion.
await Scripts.Eventually(() => runner.QuarantinedKeys(ScriptContextKind.VirtualTag).Count == 1,
TimeSpan.FromSeconds(5), "the second kill strike");
var refused = await runner.RunAsync(Scripts.Vt("runaway", Runaway, softT), Ct);
var refused = await runner.RunAsync(Scripts.Vt("runaway", runaway, softT, Scripts.Input("spin", 1)), Ct);
refused.Reason!.ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix);
refused.Reason.ShouldContain("hard deadline");
runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBe(["runaway"]);
tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ScriptRunnerPerfTests.cs
+91
View File
@@ -571,4 +571,95 @@ public sealed class ScriptRunnerPerfTests
.AppendLine($"| kill latency after T+G (5 runs, ms) | {string.Join(" / ", killLatencies.Select(l => l.ToString("N0")))} | ≤ 100 ms |");
Report(report);
}
/// <summary>
/// lmxopcua#798: the wall-clock half of the Shadow non-blocking contract, moved out of the default
/// <see cref="ShadowModeTests"/> run (which now proves it causally with barriers). Published-path latency of a
/// WARMED Calculation evaluator under <c>Scripting:Runner=Shadow</c> while its shadow keeps running away in the
/// worker and being killed, against the same evaluator with no shadow. Compile/JIT is excluded by a warm-up
/// pass. The 250 ms per-evaluation ceiling the default test used to assert is asserted here, opt-in only; the
/// report records hardware and (where the OS exposes it) load average, so a MISS on a loaded box reads as such.
/// </summary>
[Fact]
public async Task Shadow_published_path_latency_while_the_shadow_runs_away()
{
RequireOptIn();
const string inWorker = "(string.Compare(\"a\", \"B\") > 0)"; // see ShadowModeTests.InWorker
const string runaway = $"if ({inWorker}) {{ while (true) {{ }} }} return (int)ctx.GetTag(\"x\").Value * 2;";
const string sibling = "return (int)ctx.GetTag(\"x\").Value + 100;";
const int count = 2_000;
var options = Scripts.Options(hardGrace: TimeSpan.FromMilliseconds(400), quarantineAfterKills: 1_000_000);
options.Runner = ScriptRunnerMode.Shadow;
await using var pool = Scripts.Runner(options, _runnerLog.CreateLogger());
using var shadow = new ShadowScriptRunner(pool, options, _runnerLog.CreateLogger());
Driver.Calculation.CalculationEvaluator Calc(IScriptRunner? runner)
=> new(Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance,
new ScriptRootLogger(new Serilog.LoggerConfiguration().CreateLogger()), TimeSpan.FromMilliseconds(300), scriptRunner: runner);
async Task<List<double>> Measure(Driver.Calculation.CalculationEvaluator evaluator)
{
for (var w = 0; w < 50; w++) // warm-up: compile + JIT both scripts in process
{
(await evaluator.EvaluateAsync("runaway", runaway, Scripts.Input("x", w), Ct)).Value.ShouldBe(w * 2);
(await evaluator.EvaluateAsync("sibling", sibling, Scripts.Input("x", w), Ct)).Value.ShouldBe(w + 100);
}
// A 5 ms trigger cadence for the phase, so shadow runs complete, run away and are killed DURING the
// measurement (back-to-back evaluations would all land behind the first shadow run, key_busy).
var samples = new List<double>(count);
var phaseEnd = Stopwatch.GetTimestamp() + (long)(Phase.TotalSeconds * Stopwatch.Frequency);
for (var n = 0; Stopwatch.GetTimestamp() < phaseEnd; n++)
{
await Task.Delay(5, Ct);
var key = n % 2 == 0 ? "runaway" : "sibling";
var t0 = Stopwatch.GetTimestamp();
var r = await evaluator.EvaluateAsync(key, n % 2 == 0 ? runaway : sibling, Scripts.Input("x", n), Ct);
samples.Add(Stopwatch.GetElapsedTime(t0).TotalMilliseconds);
r.Value.ShouldBe(n % 2 == 0 ? n * 2 : n + 100);
}
samples.Sort();
return samples;
}
List<double> reference, shadowed;
using (var plain = Calc(null)) reference = await Measure(plain);
using (var mirrored = Calc(shadow)) shadowed = await Measure(mirrored);
await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30));
var stats = shadow.Stats();
var load = LoadAverage();
var ceiling = TimeSpan.FromMilliseconds(250).TotalMilliseconds;
var report = new StringBuilder()
.AppendLine($"### Shadow published-path latency ({Env()}; load {load})").AppendLine()
.AppendLine("| Evaluator (warmed) | p50 ms | p99 ms | max ms |")
.AppendLine("|---|---:|---:|---:|")
.AppendLine($"| InProcess (no shadow) | {Percentile(reference, 0.50):F3} | {Percentile(reference, 0.99):F3} | {reference[^1]:F3} |")
.AppendLine($"| Shadow, runaway killed in the worker | {Percentile(shadowed, 0.50):F3} | {Percentile(shadowed, 0.99):F3} | {shadowed[^1]:F3} |")
.AppendLine().AppendLine($"shadow: compared {stats.Compared}, mismatched {stats.Mismatched}, unavailable {stats.ShadowUnavailable}, " +
$"dropped key_busy {stats.DroppedKeyBusy}, dropped behind {stats.DroppedBehind}; ceiling {ceiling:F0} ms per evaluation");
Report(report);
stats.Mismatched.ShouldBeGreaterThan(0, "precondition: the shadow really ran the runaway away (Failed [budget] vs published Ok)");
shadowed[^1].ShouldBeLessThan(ceiling, "a warmed published evaluation never waits for its shadow");
}
/// <summary>The host's 1/5/15-minute load average for the report: <c>/proc/loadavg</c> on Linux, <c>sysctl</c> on macOS.</summary>
private static string LoadAverage()
{
try
{
if (File.Exists("/proc/loadavg")) return File.ReadAllText("/proc/loadavg").Trim();
if (OperatingSystem.IsMacOS())
{
using var p = Process.Start(new ProcessStartInfo("/usr/sbin/sysctl", "-n vm.loadavg") { RedirectStandardOutput = true });
var text = p!.StandardOutput.ReadToEnd().Trim();
p.WaitForExit(2_000);
return text;
}
}
catch (Exception ex) when (ex is IOException or System.ComponentModel.Win32Exception or InvalidOperationException)
{
return "n/a (" + ex.GetType().Name + ")";
}
return "n/a";
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ShadowModeTests.cs
+116 -14
View File
@@ -8,6 +8,7 @@ using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
using ZB.MOM.WW.OtOpcUa.Driver.Calculation;
namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
@@ -130,37 +131,89 @@ public sealed class ShadowModeTests
lines.ShouldAllBe(e => e.Level == Serilog.Events.LogEventLevel.Warning);
}
/// <summary>
/// The non-blocking contract of <see cref="IScriptShadow.Mirror"/>, proven causally rather than by a stopwatch
/// (lmxopcua#798: a 250 ms bound around the first evaluations measured cold Roslyn compile/JIT of the in-process
/// run, not shadow scheduling). <see cref="GatedShadow"/> sits between the REAL <see cref="ShadowScriptRunner"/>
/// and the REAL worker pool and holds the runaway's first shadow run at two test-owned barriers: before it reaches
/// the pool, and — after the worker has run it away, answered the soft timeout and been killed at its hard
/// deadline — before its outcome is handed back for comparison. Every published evaluation made while a barrier
/// is closed completed while its shadow was provably outstanding (only the test opens a barrier), so a published
/// path that waited for its shadow would hang here rather than merely run slow. Mutation check: make
/// <see cref="ShadowScriptRunner.Mirror"/> wait for the shadow run and the first held evaluation never returns.
/// A representative latency number is the opt-in <see cref="ScriptRunnerPerfTests"/> case (Category=Perf).
/// </summary>
[Fact]
public async Task A_runaway_that_runs_away_only_in_the_worker_is_killed_and_quarantined_there_and_never_reaches_a_published_value()
{
var log = new CapturingSink();
await using var pool = Scripts.Runner(ShadowOptions(TimeSpan.FromMilliseconds(400)), log.CreateLogger());
using var shadow = new ShadowScriptRunner(pool, ShadowOptions(), log.CreateLogger());
var gate = new GatedShadow(pool);
using var shadow = new ShadowScriptRunner(gate, ShadowOptions(), log.CreateLogger());
using var shadowed = Calc(shadow, TimeSpan.FromMilliseconds(300));
const string runaway = $"if ({InWorker}) {{ while (true) {{ }} }} return (int)ctx.GetTag(\"i\").Value * 2;";
const string sibling = "return (int)ctx.GetTag(\"i\").Value + 100;";
var slowest = TimeSpan.Zero;
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(90);
var i = 0;
for (; !pool.QuarantinedKeys(ScriptContextKind.Calculation).Contains("runaway"); i++)
// Off the test thread with a liveness bound (not a latency bound): a Mirror that waited for its held shadow would
// otherwise hang the test instead of failing it.
Task<VirtualTagEvalResult> Publish(string key, string source)
=> Task.Run(() => shadowed.EvaluateAsync(key, source, Inputs(i), Ct), Ct).WaitAsync(TimeSpan.FromSeconds(30), Ct);
async Task PublishBoth(string phase)
{
(await Publish("runaway", runaway)).ShouldBe(VirtualTagEvalResult.Ok(i * 2), $"{phase}: published from the in-process run");
(await Publish("sibling", sibling)).ShouldBe(VirtualTagEvalResult.Ok(i + 100), phase);
i++;
}
// Barrier 1: the runaway's first shadow run is dispatched and held before the pool ever sees it.
var held = gate.Hold("runaway");
await PublishBoth("first evaluation");
await held.Entered.WaitAsync(TimeSpan.FromSeconds(30), Ct);
var droppedBefore = shadow.Stats().DroppedKeyBusy;
for (var k = 0; k < 3; k++)
{
await PublishBoth("shadow held before the worker");
held.ReachedPool.ShouldBeFalse("barrier 1 is still closed: this evaluation published while its shadow was blocked");
}
shadow.Stats().DroppedKeyBusy.ShouldBeGreaterThanOrEqualTo(droppedBefore + 3,
"each runaway trigger behind the held shadow run was dropped — not queued, not waited for");
// Open barrier 1: the shadow run really runs away in the worker, the pool answers the soft timeout T and the
// worker is killed at T + G. Barrier 2 still holds the outcome back from the comparison.
var worker = await Scripts.Eventually(() => pool.Shards[0].Current, TimeSpan.FromSeconds(30), "the shadow worker");
held.ReleaseToPool();
var answered = await held.PoolAnswered.WaitAsync(TimeSpan.FromSeconds(30), Ct);
answered.Status.ShouldBe(ScriptRunStatus.Failed, answered.Reason);
answered.Reason!.ShouldStartWith("script timed out after ", Case.Sensitive, "the worker ran it away and the pool answered at T");
await PublishBoth("shadow running away / being killed in the worker");
var exit = await worker.Exited.WaitAsync(TimeSpan.FromSeconds(30), Ct);
exit.Reason.ShouldBe(WorkerExitReason.KilledDeadline);
exit.CulpritKey.ShouldBe(InProcessScriptRunner.ScopedKey(ScriptContextKind.Calculation, "runaway"),
"only the runaway is blamed, never the sibling sharing its shard");
await PublishBoth("worker killed, shadow outcome still held");
shadow.Mismatches.ShouldBeEmpty("nothing about the held shadow run has been compared, and nothing published waited for it");
shadow.Stats().InFlight.ShouldBeGreaterThan(0, "the held shadow run is still outstanding");
// Open barrier 2: the comparison records the divergence; the published side never saw any of it.
held.Deliver();
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
shadow.Mismatches.ShouldHaveSingleItem().Key.ShouldBe("runaway");
// Un-gated from here: keep triggering until the shadow's second kill quarantines it in the pool.
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(90);
while (!pool.QuarantinedKeys(ScriptContextKind.Calculation).Contains("runaway"))
{
if (DateTime.UtcNow > deadline)
throw new TimeoutException("the shadow never quarantined the runaway:" + Environment.NewLine + string.Join(Environment.NewLine, log.Rendered));
var sw = Stopwatch.StartNew();
var a = await shadowed.EvaluateAsync("runaway", runaway, Inputs(i), Ct);
var b = await shadowed.EvaluateAsync("sibling", sibling, Inputs(i), Ct);
if (sw.Elapsed > slowest) slowest = sw.Elapsed;
a.ShouldBe(VirtualTagEvalResult.Ok(i * 2), "published from the in-process run");
b.ShouldBe(VirtualTagEvalResult.Ok(i + 100));
await PublishBoth("until quarantined");
await Task.Delay(100, Ct);
}
// Still publishing correctly after the shadow quarantined it, and never quarantined on the published side.
for (var j = i; j < i + 5; j++)
(await shadowed.EvaluateAsync("runaway", runaway, Inputs(j), Ct)).ShouldBe(VirtualTagEvalResult.Ok(j * 2));
for (var j = 0; j < 5; j++) await PublishBoth("after the shadow quarantine");
shadowed.QuarantinedScripts.ShouldBeEmpty();
slowest.ShouldBeLessThan(TimeSpan.FromMilliseconds(250), "a published evaluation never waits for its shadow");
log.Rendered.ShouldContain(l => l.Contains("exceeded its hard deadline"));
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
@@ -171,6 +224,55 @@ public sealed class ShadowModeTests
shadow.Stats().DroppedKeyBusy.ShouldBeGreaterThan(0, "triggers while its shadow run was still going were dropped, not queued");
}
/// <summary>
/// A pass-through shadow runner with test-owned barriers on one key's next run: it holds that run before it reaches
/// the inner runner, and holds the inner runner's outcome before returning it. Everything else forwards untouched.
/// </summary>
private sealed class GatedShadow(IScriptRunner inner) : IScriptRunner
{
private readonly ConcurrentDictionary<string, Held> _holds = new(StringComparer.Ordinal);
public Held Hold(string key) => _holds[key] = new Held();
public async Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default)
{
if (!_holds.TryRemove(invocation.Key, out var held)) return await inner.RunAsync(invocation, ct);
held.EnteredSource.TrySetResult();
await held.ToPool.Task.WaitAsync(ct);
held.ReachedPool = true;
var outcome = await inner.RunAsync(invocation, ct);
held.Answered.TrySetResult(outcome);
await held.Delivery.Task.WaitAsync(ct);
return outcome;
}
public void Reset(ScriptContextKind kind) => inner.Reset(kind);
public IReadOnlyCollection<string> QuarantinedKeys(ScriptContextKind kind) => inner.QuarantinedKeys(kind);
public sealed class Held
{
internal readonly TaskCompletionSource EnteredSource = new(TaskCreationOptions.RunContinuationsAsynchronously);
internal readonly TaskCompletionSource ToPool = new(TaskCreationOptions.RunContinuationsAsynchronously);
internal readonly TaskCompletionSource<ScriptRunOutcome> Answered = new(TaskCreationOptions.RunContinuationsAsynchronously);
internal readonly TaskCompletionSource Delivery = new(TaskCreationOptions.RunContinuationsAsynchronously);
private volatile bool _reachedPool;
/// <summary>The shadow runner dispatched the held run: it is outstanding from here on.</summary>
public Task Entered => EnteredSource.Task;
/// <summary>Whether the held run has been let through to the inner runner.</summary>
public bool ReachedPool { get => _reachedPool; internal set => _reachedPool = value; }
/// <summary>The inner runner's outcome for the held run, before it is returned to the shadow runner.</summary>
public Task<ScriptRunOutcome> PoolAnswered => Answered.Task;
public void ReleaseToPool() => ToPool.TrySetResult();
public void Deliver() => Delivery.TrySetResult();
}
}
private static ScriptedAlarmDefinition Alarm(string id, string predicate) =>
new(AlarmId: id, EquipmentPath: "Plant/Line1/Reactor", AlarmName: id, Kind: AlarmKind.AlarmCondition,
Severity: AlarmSeverity.High, MessageTemplate: "condition", PredicateScriptSource: predicate);