Joseph Doherty
0796928504
v2-ci / unit-tests (push) Waiting to run
- IScriptRunner.RunBatch: one EvalBatch frame per worker for every evaluation ready to run; per-item, streamed results. The default (InProcessScriptRunner) is RunAsync per item. Used by the Calculation pump (dequeue all, split at a repeated tag, publish in dequeue order), the VirtualTag seam (ScriptBatchCoalescer, no timer) and the scripted-alarm engine (whole dirty set). In process the pump and the engine stay one at a time, unchanged. - Exact per-item deadlines: the worker hands work to an idle evaluation thread or defers it (EvalResults deferred_eval_ids); the host stops that run's clock and restarts T then G when it starts (started_eval_ids). A run queued behind a runaway is neither timed out nor struck and is a victim, not the culprit, at the kill. - Scripted-alarm seam behind Scripting:Runner:Seams; unset Seams now means VirtualTag + Calculation, so alarms run out of process only when listed. Load still compiles each predicate (to PE bytes only) so a broken one fails the load; passes batch and apply in the serial order under the gate. Script faults freeze the condition BadInternalError; runner faults freeze it BadResourceUnavailable and re-evaluate on recovery; a dead worker never clears an alarm. IScriptRunnerProvider carries the runner to DriverHostActor. - Perf harness: batch throughput (RunBatch vs RunAsync x n) and alarm-pass latency (in process / batched / serial); design §12.3 records macOS + windev. On windev the serial out-of-process alarm pass misses 250 ms from ~500 alarms; batched p99 <= 53 ms at 1,000. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
575 lines
31 KiB
C#
575 lines
31 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;
|
||
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>
|
||
private static async Task<PhaseResult> PacedAsync(IScriptRunner runner, int ratePerSecond)
|
||
{
|
||
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);
|
||
while (issued < due)
|
||
{
|
||
var key = next++ % Keys;
|
||
if (Volatile.Read(ref inFlight[key])) continue;
|
||
Volatile.Write(ref inFlight[key], true);
|
||
var x = (int)issued++;
|
||
var t0 = Stopwatch.GetTimestamp();
|
||
pending.Add(runner.RunAsync(Invocation(key, x), Ct).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));
|
||
}
|
||
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);
|
||
}
|
||
|
||
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);
|
||
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. " +
|
||
"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 ≤ 80 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);
|
||
}
|
||
}
|
||
|
||
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) | {rssIdle / 1048576.0:N0} MB | ≤ 80 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) | {rssWarmIdle / 1048576.0:N0} MB | ≤ 80 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);
|
||
}
|
||
}
|