Joseph Doherty
d8a000a409
v2-ci / unit-tests (push) Has been cancelled
Profiled the remaining lmxopcua#789 misses on windev, a VMware VM, using per-thread CPU and context-switch accounting plus ETW CPU sampling. - One blocking hand-off costs ~26 us of latency and ~18 us of CPU there, about 10x macOS. - An idle evaluation makes four hand-offs in a row: host writer -> worker reader -> evaluation thread -> host reader -> dispatcher. - 48% of host CPU at 10k/s is in the kernel: the per-RunAsync NtWriteFile, thread-pool wakes, reads and waits. Spin-before-block was measured at every hand-off and is not kept. It meets 150 us idle p50 (118 us) only by keeping about two host cores and four worker cores busy for one stream. At 10k/s it reads +250 points. Harness: new "paced 10,000/s, batched" row. It uses the same rate and keys, with each 1 ms tick submitted as one RunBatch call, which is the production seams' shape. It runs after the original four phases, so their rows stay comparable. On windev it halves host CPU, to +36..+45 points against +71..+84 unbatched. That is still above 25. Design doc: - 12.8: the record. - 14 #15: the owner's decision on the Windows budgets. Recommended: measure on bare-metal Windows, then state the two budgets per platform, with the batched row as the gate. The alternative is a shared-memory transport. Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi
699 lines
39 KiB
C#
699 lines
39 KiB
C#
using System.Diagnostics;
|
||
using System.Text;
|
||
using Shouldly;
|
||
using Xunit;
|
||
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;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
|
||
|
||
/// <summary>
|
||
/// lmxopcua#780 design §8 perf harness (§13 "Performance"): a Stopwatch-based measurement of the in-process and
|
||
/// out-of-process runners against the accepted gate of <b>5,000 compiled evaluations per second per node</b> and the
|
||
/// §8.2 budgets. Opt-in and never part of a default run: every case skips unless <c>OTOPCUA_SCRIPT_PERF=1</c>.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>Run: <c>OTOPCUA_SCRIPT_PERF=1 dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests -c Release
|
||
/// --filter Category=Perf --logger "console;verbosity=detailed"</c>. <c>OTOPCUA_SCRIPT_PERF_SECONDS</c> sets each
|
||
/// load phase's length (default 10) and <c>OTOPCUA_SCRIPT_PERF_OUT</c> appends the report to a file.</para>
|
||
/// <para>What the numbers are: the runners are driven directly (no OPC UA, no actors), so they are the runner's own
|
||
/// cost — the "added latency" the design budgets. Host CPU is this test process's
|
||
/// <see cref="Process.TotalProcessorTime"/> over the phase, as a percentage of one core; for the in-process runner
|
||
/// that includes the scripts themselves, for the out-of-process one it is IPC + bookkeeping only (the scripts run in
|
||
/// the worker). The script is the trivial one-input script the budgets are written for.</para>
|
||
/// <para>Only the accepted gate (≥ 5,000 evals/s at K=2) is asserted; the other §8.2 budgets are reported
|
||
/// PASS/MISS so a run on a slow box still produces its whole table.</para>
|
||
/// </remarks>
|
||
[Collection(ScriptWorkerCollection.Name)]
|
||
[Trait("Category", "Perf")]
|
||
public sealed class ScriptRunnerPerfTests
|
||
{
|
||
private const string OptInVariable = "OTOPCUA_SCRIPT_PERF";
|
||
private const int AcceptedGateEvalsPerSecond = 5_000;
|
||
|
||
/// <summary>
|
||
/// §8.2 "Worker idle RSS ≤ 200 MB + scripts", per shard (revised by the owner 2026-11-11, lmxopcua#789: compiles
|
||
/// run in the worker, so Roslyn stays resident; the original 80 MB [A-3] is only met by a warm-PE-cache worker).
|
||
/// </summary>
|
||
private const int WorkerIdleRssBudgetMb = 200;
|
||
private const int Keys = 2_000;
|
||
private const int DistinctSources = 50;
|
||
|
||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||
|
||
/// <summary>The runners' own log (worker starts, exits and their reasons), quoted when a phase fails.</summary>
|
||
private readonly CapturingSink _runnerLog = new();
|
||
|
||
private string RunnerLog() => string.Join(Environment.NewLine,
|
||
_runnerLog.Events.Where(e => e.Level >= Serilog.Events.LogEventLevel.Warning).TakeLast(8).Select(e => e.RenderMessage()));
|
||
|
||
private static TimeSpan Phase => TimeSpan.FromSeconds(
|
||
int.TryParse(Environment.GetEnvironmentVariable("OTOPCUA_SCRIPT_PERF_SECONDS"), out var s) && s > 0 ? s : 10);
|
||
|
||
private static void RequireOptIn()
|
||
{
|
||
if (Environment.GetEnvironmentVariable(OptInVariable) != "1")
|
||
Assert.Skip($"perf harness: set {OptInVariable}=1 to run (design §8; not part of a default run)");
|
||
}
|
||
|
||
private static string Source(int i) => $"return (int)ctx.GetTag(\"x\").Value + {i % DistinctSources};";
|
||
|
||
private static ScriptInvocation Invocation(int key, int x)
|
||
=> new("perf-" + key, ScriptContextKind.VirtualTag, Source(key), Scripts.Input("x", x), TimeSpan.FromSeconds(2));
|
||
|
||
private static void Report(StringBuilder report)
|
||
{
|
||
var text = report.ToString();
|
||
TestContext.Current.TestOutputHelper?.WriteLine(text);
|
||
if (Environment.GetEnvironmentVariable("OTOPCUA_SCRIPT_PERF_OUT") is { Length: > 0 } path)
|
||
File.AppendAllText(path, text + Environment.NewLine);
|
||
}
|
||
|
||
private static string Env()
|
||
=> $"{Environment.OSVersion.Platform} {System.Runtime.InteropServices.RuntimeInformation.OSArchitecture}, " +
|
||
$"{Environment.ProcessorCount} logical CPUs, .NET {Environment.Version}, phase {Phase.TotalSeconds:F0} s";
|
||
|
||
private sealed record PhaseResult(double EvalsPerSecond, double P50Us, double P99Us, double HostCpuPercent,
|
||
long Ok, long NotOk, string? FirstFailure);
|
||
|
||
private static double Percentile(List<double> sorted, double p)
|
||
=> sorted.Count == 0 ? double.NaN : sorted[Math.Min(sorted.Count - 1, (int)Math.Ceiling(p * sorted.Count) - 1)];
|
||
|
||
private async Task WarmAsync(IScriptRunner runner)
|
||
{
|
||
// Compile every distinct source and touch every key once, so the phases measure steady state.
|
||
var tasks = Enumerable.Range(0, Keys).Select(k => runner.RunAsync(Invocation(k, 1), Ct)).ToArray();
|
||
var outcomes = await Task.WhenAll(tasks);
|
||
foreach (var o in outcomes) o.Status.ShouldBe(ScriptRunStatus.Ok, $"{o.Reason}{Environment.NewLine}{RunnerLog()}");
|
||
}
|
||
|
||
/// <summary>Sequential evaluations of one key on an idle runner: the per-call added latency.</summary>
|
||
private static async Task<PhaseResult> IdleLatencyAsync(IScriptRunner runner, int count = 5_000)
|
||
{
|
||
var samples = new List<double>(count);
|
||
var cpu0 = Process.GetCurrentProcess().TotalProcessorTime;
|
||
var wall = Stopwatch.StartNew();
|
||
long ok = 0;
|
||
string? firstFailure = null;
|
||
for (var i = 0; i < count; i++)
|
||
{
|
||
var t0 = Stopwatch.GetTimestamp();
|
||
var o = await runner.RunAsync(Invocation(7, i), Ct);
|
||
samples.Add(Stopwatch.GetElapsedTime(t0).TotalMicroseconds);
|
||
if (o.Status == ScriptRunStatus.Ok && Equals(o.Value, i + 7)) ok++;
|
||
else firstFailure ??= $"#{i} {o.Status}: {o.Reason} (value {o.Value})";
|
||
}
|
||
var cpu = (Process.GetCurrentProcess().TotalProcessorTime - cpu0).TotalMilliseconds / wall.Elapsed.TotalMilliseconds * 100;
|
||
samples.Sort();
|
||
return new PhaseResult(count / wall.Elapsed.TotalSeconds, Percentile(samples, 0.50), Percentile(samples, 0.99), cpu, ok, count - ok, firstFailure);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Closed loop: <paramref name="concurrency"/> submitters, each cycling over its own slice of the keys (so
|
||
/// single flight per key never serialises them), for one phase. Measures the maximum sustainable rate.
|
||
/// </summary>
|
||
private static async Task<PhaseResult> ClosedLoopAsync(IScriptRunner runner, int concurrency)
|
||
{
|
||
var deadline = Stopwatch.GetTimestamp() + (long)(Phase.TotalSeconds * Stopwatch.Frequency);
|
||
var perLoop = new List<double>[concurrency];
|
||
long ok = 0, notOk = 0;
|
||
string? firstFailure = null;
|
||
var cpu0 = Process.GetCurrentProcess().TotalProcessorTime;
|
||
var wall = Stopwatch.StartNew();
|
||
await Task.WhenAll(Enumerable.Range(0, concurrency).Select(loop => Task.Run(async () =>
|
||
{
|
||
var samples = perLoop[loop] = new List<double>(200_000);
|
||
var keysPerLoop = Keys / concurrency;
|
||
var n = 0;
|
||
while (Stopwatch.GetTimestamp() < deadline)
|
||
{
|
||
var key = loop * keysPerLoop + n % keysPerLoop;
|
||
var t0 = Stopwatch.GetTimestamp();
|
||
var o = await runner.RunAsync(Invocation(key, n), Ct);
|
||
samples.Add(Stopwatch.GetElapsedTime(t0).TotalMicroseconds);
|
||
if (o.Status == ScriptRunStatus.Ok && Equals(o.Value, n + key % DistinctSources)) Interlocked.Increment(ref ok);
|
||
else
|
||
{
|
||
Interlocked.Increment(ref notOk);
|
||
Interlocked.CompareExchange(ref firstFailure, $"{o.Status}: {o.Reason} (value {o.Value})", null);
|
||
}
|
||
n++;
|
||
}
|
||
}, Ct)));
|
||
var elapsed = wall.Elapsed;
|
||
var cpu = (Process.GetCurrentProcess().TotalProcessorTime - cpu0).TotalMilliseconds / elapsed.TotalMilliseconds * 100;
|
||
var all = perLoop.SelectMany(s => s).ToList();
|
||
all.Sort();
|
||
return new PhaseResult((ok + notOk) / elapsed.TotalSeconds, Percentile(all, 0.50), Percentile(all, 0.99), cpu, ok, notOk, firstFailure);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Open loop at a fixed offered rate: every millisecond tick issues the evaluations due by then (round robin over
|
||
/// the keys, skipping a key still in flight), and latency is measured per call. This is the "p99 at N evals/s" of
|
||
/// the §8.2 table.
|
||
/// </summary>
|
||
/// <param name="runner">The runner.</param>
|
||
/// <param name="ratePerSecond">The offered rate.</param>
|
||
/// <param name="batched">
|
||
/// Submit each tick's evaluations as ONE <see cref="IScriptRunner.RunBatch"/> call instead of one
|
||
/// <see cref="IScriptRunner.RunAsync"/> each — the shape of the production seams, whose Calculation pump, alarm
|
||
/// engine and virtual-tag coalescer hand everything that became due together to the runner at once (§4.5,
|
||
/// lmxopcua#789). Same offered rate, keys, per-result continuation and measurement as the unbatched phase.
|
||
/// </param>
|
||
private static async Task<PhaseResult> PacedAsync(IScriptRunner runner, int ratePerSecond, bool batched = false)
|
||
{
|
||
var samples = new List<double>(ratePerSecond * (int)Phase.TotalSeconds + 1024);
|
||
var gate = new object();
|
||
var inFlight = new bool[Keys];
|
||
long ok = 0, notOk = 0, issued = 0;
|
||
string? firstFailure = null;
|
||
var pending = new List<Task>(ratePerSecond * (int)Phase.TotalSeconds + 1024);
|
||
var cpu0 = Process.GetCurrentProcess().TotalProcessorTime;
|
||
var wall = Stopwatch.StartNew();
|
||
var next = 0;
|
||
while (wall.Elapsed < Phase)
|
||
{
|
||
var due = (long)(wall.Elapsed.TotalSeconds * ratePerSecond);
|
||
List<(int Key, int X)>? tick = batched ? [] : null;
|
||
while (issued < due)
|
||
{
|
||
var key = next++ % Keys;
|
||
if (Volatile.Read(ref inFlight[key])) continue;
|
||
Volatile.Write(ref inFlight[key], true);
|
||
var x = (int)issued++;
|
||
if (tick is not null) tick.Add((key, x));
|
||
else Track(key, x, Stopwatch.GetTimestamp(), runner.RunAsync(Invocation(key, x), Ct));
|
||
}
|
||
if (tick is { Count: > 0 })
|
||
{
|
||
var t0 = Stopwatch.GetTimestamp();
|
||
var outcomes = runner.RunBatch(tick.Select(i => Invocation(i.Key, i.X)).ToArray(), Ct);
|
||
for (var i = 0; i < tick.Count; i++) Track(tick[i].Key, tick[i].X, t0, outcomes[i]);
|
||
}
|
||
await Task.Delay(1, Ct);
|
||
}
|
||
await Task.WhenAll(pending);
|
||
var elapsed = wall.Elapsed;
|
||
var cpu = (Process.GetCurrentProcess().TotalProcessorTime - cpu0).TotalMilliseconds / elapsed.TotalMilliseconds * 100;
|
||
samples.Sort();
|
||
return new PhaseResult(issued / elapsed.TotalSeconds, Percentile(samples, 0.50), Percentile(samples, 0.99), cpu, ok, notOk, firstFailure);
|
||
|
||
void Track(int key, int x, long t0, Task<ScriptRunOutcome> run)
|
||
=> pending.Add(run.ContinueWith(t =>
|
||
{
|
||
var us = Stopwatch.GetElapsedTime(t0).TotalMicroseconds;
|
||
Volatile.Write(ref inFlight[key], false);
|
||
var o = t.Result;
|
||
lock (gate) samples.Add(us);
|
||
if (o.Status == ScriptRunStatus.Ok && Equals(o.Value, x + key % DistinctSources)) Interlocked.Increment(ref ok);
|
||
else
|
||
{
|
||
Interlocked.Increment(ref notOk);
|
||
Interlocked.CompareExchange(ref firstFailure, $"{o.Status}: {o.Reason}", null);
|
||
}
|
||
}, TaskScheduler.Default));
|
||
}
|
||
|
||
private static string Row(string runner, string phase, PhaseResult r, PhaseResult? inProcess = null)
|
||
=> $"| {runner} | {phase} | {r.EvalsPerSecond,9:N0} | {r.P50Us,8:N0} | {r.P99Us,9:N0} | {r.HostCpuPercent,6:N0} % | " +
|
||
(inProcess is null ? "— |" : $"{r.HostCpuPercent - inProcess.HostCpuPercent,6:+0;-0;0} % |") +
|
||
$" {r.Ok:N0} / {r.NotOk:N0} |" +
|
||
(r.FirstFailure is null ? "" : $" first failure: {r.FirstFailure}");
|
||
|
||
/// <summary>
|
||
/// Throughput and latency, InProcess vs OutOfProcess at K = 1 / 2 / 4. Asserts the accepted gate: the
|
||
/// out-of-process runner at K = 2 sustains ≥ 5,000 trivial evals/s with every result correct.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Throughput_and_latency_against_the_accepted_gate()
|
||
{
|
||
RequireOptIn();
|
||
var report = new StringBuilder()
|
||
.AppendLine($"### Script runner perf ({Env()})")
|
||
.AppendLine()
|
||
.AppendLine("| Runner | Phase | evals/s | p50 µs | p99 µs | host CPU | vs InProcess | ok / not ok |")
|
||
.AppendLine("|---|---|---:|---:|---:|---:|---:|---|");
|
||
PhaseResult? k2Closed = null;
|
||
// The in-process run of each phase is the harness's own floor (its pacing loop, its continuations, the pool):
|
||
// the "vs InProcess" column is what the out-of-process runner adds on top — the §8.2 "host CPU for IPC".
|
||
var baseline = new Dictionary<string, PhaseResult>(StringComparer.Ordinal);
|
||
|
||
var configs = new (string Name, Func<IScriptRunner> Create)[]
|
||
{
|
||
("InProcess", () => new InProcessScriptRunner(new ScriptingOptions())),
|
||
("OutOfProcess K=1", () => Scripts.Runner(PerfOptions(1), _runnerLog.CreateLogger())),
|
||
("OutOfProcess K=2", () => Scripts.Runner(PerfOptions(2), _runnerLog.CreateLogger())),
|
||
("OutOfProcess K=4", () => Scripts.Runner(PerfOptions(4), _runnerLog.CreateLogger())),
|
||
};
|
||
foreach (var (name, create) in configs)
|
||
{
|
||
var runner = create();
|
||
try
|
||
{
|
||
await WarmAsync(runner);
|
||
void Add(string phase, PhaseResult r)
|
||
{
|
||
if (name == "InProcess") baseline[phase] = r;
|
||
report.AppendLine(Row(name, phase, r, name == "InProcess" ? null : baseline.GetValueOrDefault(phase)));
|
||
}
|
||
Add("idle, sequential", await IdleLatencyAsync(runner));
|
||
Add("paced 5,000/s", await PacedAsync(runner, 5_000));
|
||
Add("paced 10,000/s", await PacedAsync(runner, 10_000));
|
||
var closed = await ClosedLoopAsync(runner, concurrency: 64);
|
||
Add("closed loop ×64", closed);
|
||
// After the original four phases, so their rows stay comparable with every earlier record (lmxopcua#789).
|
||
Add("paced 10,000/s, batched", await PacedAsync(runner, 10_000, batched: true));
|
||
if (name == "OutOfProcess K=2") k2Closed = closed;
|
||
}
|
||
finally
|
||
{
|
||
switch (runner)
|
||
{
|
||
case IAsyncDisposable a: await a.DisposeAsync(); break;
|
||
case IDisposable d: d.Dispose(); break;
|
||
}
|
||
}
|
||
}
|
||
|
||
report.AppendLine()
|
||
.AppendLine("§8.2 budgets: added latency p50 ≤ 150 µs (idle), p99 ≤ 2 ms at 10k/s, ≥ 10,000 evals/s at K=2, " +
|
||
"host CPU for IPC (the \"vs InProcess\" column) ≤ 25 % of one core at 10k/s. " +
|
||
"\"batched\": each 1 ms tick's evaluations go in one RunBatch call, as the production seams submit. " +
|
||
"Accepted gate (owner, §14): ≥ 5,000 evals/s/node.");
|
||
Report(report);
|
||
|
||
k2Closed.ShouldNotBeNull();
|
||
k2Closed.NotOk.ShouldBe(0, k2Closed.FirstFailure);
|
||
k2Closed.EvalsPerSecond.ShouldBeGreaterThanOrEqualTo(AcceptedGateEvalsPerSecond, "the accepted 5k evals/s/node gate at K=2");
|
||
}
|
||
|
||
/// <summary>
|
||
/// lmxopcua#780 batch submit (design §4.5): a closed loop of one batch at a time, <paramref name="batchSize"/>
|
||
/// distinct loaded keys per batch, either through <see cref="IScriptRunner.RunBatch"/> or as the same evaluations
|
||
/// submitted as concurrent <see cref="IScriptRunner.RunAsync"/> calls. Latency is per batch (submit → last result).
|
||
/// </summary>
|
||
private static async Task<PhaseResult> BatchLoopAsync(IScriptRunner runner, int batchSize, bool viaBatch)
|
||
{
|
||
var deadline = Stopwatch.GetTimestamp() + (long)(Phase.TotalSeconds * Stopwatch.Frequency);
|
||
var samples = new List<double>(100_000);
|
||
long ok = 0, notOk = 0;
|
||
string? firstFailure = null;
|
||
var offset = 0;
|
||
var x = 0;
|
||
var cpu0 = Process.GetCurrentProcess().TotalProcessorTime;
|
||
var wall = Stopwatch.StartNew();
|
||
while (Stopwatch.GetTimestamp() < deadline)
|
||
{
|
||
var invocations = new ScriptInvocation[batchSize];
|
||
var keys = new int[batchSize];
|
||
for (var i = 0; i < batchSize; i++)
|
||
{
|
||
keys[i] = (offset + i) % Keys;
|
||
invocations[i] = Invocation(keys[i], x);
|
||
}
|
||
offset = (offset + batchSize) % Keys;
|
||
var t0 = Stopwatch.GetTimestamp();
|
||
var outcomes = await Task.WhenAll(viaBatch
|
||
? runner.RunBatch(invocations, Ct)
|
||
: invocations.Select(inv => runner.RunAsync(inv, Ct)).ToArray());
|
||
samples.Add(Stopwatch.GetElapsedTime(t0).TotalMicroseconds);
|
||
for (var i = 0; i < batchSize; i++)
|
||
{
|
||
var o = outcomes[i];
|
||
if (o.Status == ScriptRunStatus.Ok && Equals(o.Value, x + keys[i] % DistinctSources)) ok++;
|
||
else
|
||
{
|
||
notOk++;
|
||
firstFailure ??= $"{o.Status}: {o.Reason} (value {o.Value})";
|
||
}
|
||
}
|
||
x++;
|
||
}
|
||
var elapsed = wall.Elapsed;
|
||
var cpu = (Process.GetCurrentProcess().TotalProcessorTime - cpu0).TotalMilliseconds / elapsed.TotalMilliseconds * 100;
|
||
samples.Sort();
|
||
return new PhaseResult((ok + notOk) / elapsed.TotalSeconds, Percentile(samples, 0.50), Percentile(samples, 0.99), cpu, ok, notOk, firstFailure);
|
||
}
|
||
|
||
/// <summary>
|
||
/// A runner that turns a batch back into one evaluation after another — the serial pass the alarm engine and the
|
||
/// Calc pump made before batch submit, which design §8.2 warns costs N round trips (the comparison row).
|
||
/// </summary>
|
||
private sealed class SerialRunner(IScriptRunner inner) : IScriptRunner
|
||
{
|
||
public Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default) => inner.RunAsync(invocation, ct);
|
||
|
||
public IReadOnlyList<Task<ScriptRunOutcome>> RunBatch(IReadOnlyList<ScriptInvocation> invocations, CancellationToken ct = default)
|
||
{
|
||
var results = new Task<ScriptRunOutcome>[invocations.Count];
|
||
Task previous = Task.CompletedTask;
|
||
for (var i = 0; i < results.Length; i++) previous = results[i] = After(previous, invocations[i], ct);
|
||
return results;
|
||
}
|
||
|
||
private async Task<ScriptRunOutcome> After(Task previous, ScriptInvocation invocation, CancellationToken ct)
|
||
{
|
||
await previous.ConfigureAwait(false);
|
||
return await inner.RunAsync(invocation, ct).ConfigureAwait(false);
|
||
}
|
||
|
||
public void Reset(ScriptContextKind kind) => inner.Reset(kind);
|
||
|
||
public IReadOnlyCollection<string> QuarantinedKeys(ScriptContextKind kind) => inner.QuarantinedKeys(kind);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Scripted-alarm pass latency: <paramref name="alarms"/> alarms on one input, every push flips all of them, and a
|
||
/// sample is the time from the push to the last of the N transitions the pass emits (the whole dirty set through
|
||
/// the engine's gate). The design's alarm budget is 250 ms (§1.1; the host now passes <c>Scripting:RunTimeout</c>,
|
||
/// 2 s by default, lmxopcua#779), so the pass is judged against the stricter 250 ms.
|
||
/// </summary>
|
||
private static async Task<(double P50Ms, double P99Ms, double MaxMs, double CpuPercent, int NotCompleted)> AlarmPassLatencyAsync(
|
||
IScriptRunner? runner, int alarms, int pushes)
|
||
{
|
||
var up = new TestUpstream();
|
||
up.Set("x", 0);
|
||
var logger = new Serilog.LoggerConfiguration().CreateLogger();
|
||
using var engine = new ScriptedAlarmEngine(up, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger,
|
||
scriptTimeout: TimeSpan.FromMilliseconds(250), scriptRunner: runner);
|
||
var transitions = 0;
|
||
var notCompleted = 0;
|
||
var target = int.MaxValue;
|
||
var done = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
engine.OnEvent += (_, e) =>
|
||
{
|
||
// A Bad annotation means a predicate did not complete in that pass (timed out, refused): counted, reported.
|
||
if (e.Emission == EmissionKind.QualityChanged && e.WorstInputStatusCode != 0) Interlocked.Increment(ref notCompleted);
|
||
if (e.Emission is not (EmissionKind.Activated or EmissionKind.Cleared)) return;
|
||
if (Interlocked.Increment(ref transitions) == Volatile.Read(ref target)) done.TrySetResult();
|
||
};
|
||
await engine.LoadAsync(Enumerable.Range(0, alarms).Select(i => new ScriptedAlarmDefinition(
|
||
$"alarm-{i}", "Plant/Perf", $"alarm-{i}", AlarmKind.AlarmCondition, AlarmSeverity.High, "perf",
|
||
$"return (int)ctx.GetTag(\"x\").Value > {-(i % DistinctSources)};")).ToArray(), Ct);
|
||
var samples = new List<double>(pushes);
|
||
var cpu0 = Process.GetCurrentProcess().TotalProcessorTime;
|
||
var wall = Stopwatch.StartNew();
|
||
for (var p = 0; p < pushes + 2; p++)
|
||
{
|
||
// Alternate x = -1000 (every predicate false) and x = 1000 (every predicate true). The pass is expected to
|
||
// transition exactly the alarms not already in the target state — counted from the engine, so an alarm
|
||
// that did not complete an earlier pass cannot leave the count short. The first two passes are warm-up.
|
||
var activate = p % 2 == 1;
|
||
var expected = engine.GetAllStates().Count(c => (c.Active == AlarmActiveState.Active) != activate);
|
||
Interlocked.Exchange(ref transitions, 0);
|
||
done = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
Volatile.Write(ref target, expected);
|
||
var t0 = Stopwatch.GetTimestamp();
|
||
up.Push("x", activate ? 1000 : -1000);
|
||
if (expected == 0) continue;
|
||
try
|
||
{
|
||
await done.Task.WaitAsync(TimeSpan.FromSeconds(60), Ct);
|
||
}
|
||
catch (TimeoutException)
|
||
{
|
||
throw new TimeoutException($"alarm pass {p} ({(runner is null ? "in process" : runner.GetType().Name)}, {alarms} alarms): " +
|
||
$"{Volatile.Read(ref transitions)} of {expected} transitions in 60 s, " +
|
||
$"{Volatile.Read(ref notCompleted)} predicate(s) did not complete so far");
|
||
}
|
||
if (p >= 2) samples.Add(Stopwatch.GetElapsedTime(t0).TotalMilliseconds);
|
||
}
|
||
var cpu = (Process.GetCurrentProcess().TotalProcessorTime - cpu0).TotalMilliseconds / wall.Elapsed.TotalMilliseconds * 100;
|
||
samples.Sort();
|
||
return (Percentile(samples, 0.50), Percentile(samples, 0.99), samples[^1], cpu, notCompleted);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Batch throughput (RunBatch vs the same evaluations as concurrent RunAsync calls, K = 2) and scripted-alarm pass
|
||
/// latency (in process, out of process batched, and out of process serial). Report-only, except that every
|
||
/// result must be correct.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Batch_throughput_and_alarm_pass_latency()
|
||
{
|
||
RequireOptIn();
|
||
var report = new StringBuilder()
|
||
.AppendLine($"### Batch submit throughput ({Env()})")
|
||
.AppendLine()
|
||
.AppendLine("| Runner | batch | submitted as | evals/s | batch p50 µs | batch p99 µs | host CPU | ok / not ok |")
|
||
.AppendLine("|---|---:|---|---:|---:|---:|---:|---|");
|
||
|
||
foreach (var (name, create) in new (string, Func<IScriptRunner>)[]
|
||
{
|
||
("InProcess", () => new InProcessScriptRunner(new ScriptingOptions())),
|
||
("OutOfProcess K=2", () => Scripts.Runner(PerfOptions(2), _runnerLog.CreateLogger())),
|
||
})
|
||
{
|
||
var runner = create();
|
||
try
|
||
{
|
||
await WarmAsync(runner);
|
||
foreach (var size in new[] { 1, 10, 100, 500 })
|
||
{
|
||
foreach (var viaBatch in new[] { true, false })
|
||
{
|
||
var r = await BatchLoopAsync(runner, size, viaBatch);
|
||
r.NotOk.ShouldBe(0, $"{name} batch {size}: {r.FirstFailure}{Environment.NewLine}{RunnerLog()}");
|
||
report.AppendLine($"| {name} | {size} | {(viaBatch ? "RunBatch" : "RunAsync ×n")} | {r.EvalsPerSecond,9:N0} | " +
|
||
$"{r.P50Us,8:N0} | {r.P99Us,9:N0} | {r.HostCpuPercent,6:N0} % | {r.Ok:N0} / {r.NotOk:N0} |");
|
||
}
|
||
}
|
||
}
|
||
finally
|
||
{
|
||
switch (runner)
|
||
{
|
||
case IAsyncDisposable a: await a.DisposeAsync(); break;
|
||
case IDisposable d: d.Dispose(); break;
|
||
}
|
||
}
|
||
}
|
||
|
||
report.AppendLine()
|
||
.AppendLine($"### Scripted-alarm pass latency, push → last transition ({Env()})")
|
||
.AppendLine()
|
||
.AppendLine("| Engine | alarms | pass p50 ms | pass p99 ms | max ms | per alarm (p50) µs | host CPU | not completed | 250 ms |")
|
||
.AppendLine("|---|---:|---:|---:|---:|---:|---:|---:|---|");
|
||
const int pushes = 40;
|
||
await using (var shared = Scripts.Runner(PerfOptions(2), _runnerLog.CreateLogger()))
|
||
{
|
||
foreach (var alarms in new[] { 100, 500, 1000 })
|
||
{
|
||
foreach (var (name, runner) in new (string, IScriptRunner?)[]
|
||
{
|
||
("InProcess (serial)", null),
|
||
("OutOfProcess K=2, batched", shared),
|
||
("OutOfProcess K=2, serial", new SerialRunner(shared)),
|
||
})
|
||
{
|
||
var (p50, p99, max, cpu, notCompleted) = await AlarmPassLatencyAsync(runner, alarms, pushes);
|
||
report.AppendLine($"| {name} | {alarms:N0} | {p50,7:N1} | {p99,7:N1} | {max,7:N1} | {p50 * 1000 / alarms,6:N0} | " +
|
||
$"{cpu,5:N0} % | {notCompleted:N0} | {(p99 <= 250 ? "PASS" : "MISS")} |");
|
||
}
|
||
}
|
||
}
|
||
report.AppendLine()
|
||
.AppendLine("Each push flips every alarm; a pass is the engine's whole dirty set under its gate. \"serial\" replays the " +
|
||
"pre-batch one-predicate-at-a-time pass over the same runner.");
|
||
Report(report);
|
||
}
|
||
|
||
private static ScriptingOptions PerfOptions(int shards, string? peCacheDir = null)
|
||
{
|
||
var options = Scripts.Options(shards: shards, hardGrace: TimeSpan.FromSeconds(2), peCacheDir: peCacheDir,
|
||
heapLimit: 512L * 1024 * 1024);
|
||
return options;
|
||
}
|
||
|
||
/// <summary>
|
||
/// §8.2 "Shard restart → first result, 1,000 cached scripts ≤ 2 s", "Worker idle RSS ≤ 200 MB + scripts" and
|
||
/// "Kill latency after T+G ≤ 100 ms". Report-only.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Restart_rss_and_kill_latency()
|
||
{
|
||
RequireOptIn();
|
||
const int scripts = 1_000;
|
||
using var dir = new Scripts.TempDir();
|
||
var cacheDir = Path.Combine(dir.Path, "pe-cache");
|
||
static string Distinct(int i) => $"return (int)ctx.GetTag(\"x\").Value * {i} + 1;";
|
||
ScriptInvocation Inv(int i) => new("restart-" + i, ScriptContextKind.VirtualTag, Distinct(i), Scripts.Input("x", 2), TimeSpan.FromSeconds(30));
|
||
|
||
async Task<TimeSpan> AllFirstResults(OutOfProcessScriptRunner runner)
|
||
{
|
||
var sw = Stopwatch.StartNew();
|
||
var outcomes = await Task.WhenAll(Enumerable.Range(0, scripts).Select(i => runner.RunAsync(Inv(i), Ct)));
|
||
var elapsed = sw.Elapsed;
|
||
for (var i = 0; i < scripts; i++)
|
||
{
|
||
outcomes[i].Status.ShouldBe(ScriptRunStatus.Ok, $"{outcomes[i].Reason}{Environment.NewLine}{RunnerLog()}");
|
||
outcomes[i].Value.ShouldBe(2 * i + 1);
|
||
}
|
||
return elapsed;
|
||
}
|
||
|
||
var report = new StringBuilder().AppendLine($"### Script worker restart / RSS / kill latency ({Env()})").AppendLine();
|
||
TimeSpan cold, warmFresh, killRestart;
|
||
long rssIdle, rssLoaded, rssWarmIdle, rssWarmLoaded;
|
||
await using (var runner = Scripts.Runner(PerfOptions(2, cacheDir), _runnerLog.CreateLogger()))
|
||
{
|
||
(await runner.RunAsync(Inv(0), Ct)).Status.ShouldBe(ScriptRunStatus.Ok);
|
||
await Task.Delay(500, Ct);
|
||
rssIdle = runner.Shards.Max(s => s.Current is { } c ? Process.GetProcessById(c.Pid).WorkingSet64 : 0);
|
||
cold = await AllFirstResults(runner);
|
||
rssLoaded = runner.Shards.Max(s => s.Current is { } c ? Process.GetProcessById(c.Pid).WorkingSet64 : 0);
|
||
}
|
||
await using (var runner = Scripts.Runner(PerfOptions(2, cacheDir), _runnerLog.CreateLogger()))
|
||
{
|
||
// A worker that has only loaded verified PEs from the cache: Roslyn is never entered in it.
|
||
(await runner.RunAsync(Inv(0), Ct)).Status.ShouldBe(ScriptRunStatus.Ok);
|
||
await Task.Delay(500, Ct);
|
||
rssWarmIdle = runner.Shards.Max(s => s.Current is { } c ? Process.GetProcessById(c.Pid).WorkingSet64 : 0);
|
||
runner.PeCacheCounts.Misses.ShouldBe(0, "every load came from the cache");
|
||
}
|
||
await using (var runner = Scripts.Runner(PerfOptions(2, cacheDir), _runnerLog.CreateLogger()))
|
||
{
|
||
warmFresh = await AllFirstResults(runner);
|
||
rssWarmLoaded = runner.Shards.Max(s => s.Current is { } c ? Process.GetProcessById(c.Pid).WorkingSet64 : 0);
|
||
// A kill of both shard workers: time from the kill to every script's next correct result (includes the
|
||
// 250 ms first restart backoff and the cache load).
|
||
foreach (var shard in runner.Shards) Process.GetProcessById(shard.Current!.Pid).Kill();
|
||
var sw = Stopwatch.StartNew();
|
||
await Scripts.Eventually(() => runner.Shards.All(s => s.Current is null || s.Current.Exited.IsCompleted) ? true : false,
|
||
TimeSpan.FromSeconds(5), "both kills observed");
|
||
await Scripts.Eventually(() => runner.State == ScriptRunnerState.Ready, TimeSpan.FromSeconds(30), "both shards back");
|
||
killRestart = sw.Elapsed + await AllFirstResults(runner);
|
||
}
|
||
|
||
// Kill latency: soft T = 200 ms, G = 1 s; time from the soft-timeout answer to the worker's exit.
|
||
var killLatencies = new List<double>();
|
||
await using (var runner = Scripts.Runner(Scripts.Options(hardGrace: TimeSpan.FromSeconds(1), quarantineAfterKills: 100)))
|
||
{
|
||
for (var i = 0; i < 5; i++)
|
||
{
|
||
(await runner.RunAsync(Scripts.Vt("warm", "return 1;"), Ct)).Status.ShouldBe(ScriptRunStatus.Ok);
|
||
var worker = runner.Shards[0].Current!;
|
||
// Timed from the soft-timeout answer (T after the submit), not from the call: the call also compiles
|
||
// the cold runaway key, which on the Windows VM alone is 100–200 ms (lmxopcua#789 — the first
|
||
// harness counted it as kill latency). Kill latency = exit − (answer + G).
|
||
var outcome = await runner.RunAsync(Scripts.Vt("runaway-" + i, "while (true) { }", TimeSpan.FromMilliseconds(200)), Ct);
|
||
var sw = Stopwatch.StartNew();
|
||
outcome.Status.ShouldBe(ScriptRunStatus.Failed);
|
||
await worker.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||
killLatencies.Add((sw.Elapsed - TimeSpan.FromMilliseconds(1000)).TotalMilliseconds);
|
||
}
|
||
}
|
||
|
||
static string RssCell(long bytes)
|
||
{
|
||
var mb = bytes / 1048576.0;
|
||
return $"{mb:N0} MB {(mb <= WorkerIdleRssBudgetMb ? "PASS" : "MISS")}";
|
||
}
|
||
|
||
report.AppendLine("| Measure | Result | §8.2 budget |")
|
||
.AppendLine("|---|---:|---|")
|
||
.AppendLine($"| {scripts} scripts, empty cache, all first results (compile + write) | {cold.TotalMilliseconds:N0} ms | — |")
|
||
.AppendLine($"| {scripts} scripts, fresh runner on a warm cache, all first results | {warmFresh.TotalMilliseconds:N0} ms | ≤ 2,000 ms |")
|
||
.AppendLine($"| kill both shards → every script's next correct result | {killRestart.TotalMilliseconds:N0} ms | ≤ 2,000 ms |")
|
||
.AppendLine($"| worker RSS after first eval, compiled (max of 2 shards) | {RssCell(rssIdle)} | ≤ {WorkerIdleRssBudgetMb} MB + scripts |")
|
||
.AppendLine($"| worker RSS with {scripts / 2} scripts compiled per shard (max) | {rssLoaded / 1048576.0:N0} MB | — |")
|
||
.AppendLine($"| worker RSS after first eval, warm PE cache, no compile (max) | {RssCell(rssWarmIdle)} | ≤ {WorkerIdleRssBudgetMb} MB + scripts |")
|
||
.AppendLine($"| worker RSS with {scripts / 2} scripts loaded from the cache per shard (max) | {rssWarmLoaded / 1048576.0:N0} MB | — |")
|
||
.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";
|
||
}
|
||
}
|