feat(scripting): self-contained script worker publish, docker-dev drill, §8 perf harness (lmxopcua#780 P1)
v2-ci / unit-tests (push) Waiting to run
v2-ci / unit-tests (push) Waiting to run
- A self-contained Host publish (-r win-x64 --self-contained, the Windows installs) now publishes the ScriptWorker self-contained for the same RID into script-worker/. The launcher starts its apphost directly when runtimeconfig.json lists includedFrameworks. A framework-dependent worker (docker-dev) keeps running through the dotnet muxer. - docker-dev/docker-compose.script-runner.yml: OutOfProcess for the central pair's VT + Calc seams. Live drill recorded in design §12.1 (runaway kill + quarantine, kill -9 recovery, open circuit with /health/ready still 200, metrics). - ScriptRunnerPerfTests (Category=Perf, opt-in via OTOPCUA_SCRIPT_PERF=1): InProcess vs OutOfProcess at K=1/2/4. The accepted 5k evals/s/node gate passes on macOS and windev. The host-CPU, RSS and Windows latency budget misses are tracked as #789. - Fix: the single-flight slot is released before the awaiting caller resumes. Before, a caller that started its next run right after the previous one returned could be refused with "previous evaluation still running", which the seams publish as Bad. - Fix: the worker compiles on dedicated threads, not the thread pool. A burst of cold loads starved the heartbeat, and the host killed the shard (found on windev). Load-timeout queueing is #790. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
@@ -0,0 +1,337 @@
|
||||
using System.Diagnostics;
|
||||
using System.Text;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
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)
|
||||
=> $"| {runner} | {phase} | {r.EvalsPerSecond,9:N0} | {r.P50Us,8:N0} | {r.P99Us,9:N0} | {r.HostCpuPercent,6:N0} % | {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 | ok / not ok |")
|
||||
.AppendLine("|---|---|---:|---:|---:|---:|---|");
|
||||
PhaseResult? k2Closed = null;
|
||||
|
||||
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);
|
||||
report.AppendLine(Row(name, "idle, sequential", await IdleLatencyAsync(runner)));
|
||||
report.AppendLine(Row(name, "paced 5,000/s", await PacedAsync(runner, 5_000)));
|
||||
report.AppendLine(Row(name, "paced 10,000/s", await PacedAsync(runner, 10_000)));
|
||||
var closed = await ClosedLoopAsync(runner, concurrency: 64);
|
||||
report.AppendLine(Row(name, "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 ≤ 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");
|
||||
}
|
||||
|
||||
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;
|
||||
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()))
|
||||
{
|
||||
warmFresh = await AllFirstResults(runner);
|
||||
// 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!;
|
||||
var sw = Stopwatch.StartNew();
|
||||
var outcome = await runner.RunAsync(Scripts.Vt("runaway-" + i, "while (true) { }", TimeSpan.FromMilliseconds(200)), Ct);
|
||||
outcome.Status.ShouldBe(ScriptRunStatus.Failed);
|
||||
await worker.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||||
killLatencies.Add((sw.Elapsed - TimeSpan.FromMilliseconds(1200)).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 (max of 2 shards) | {rssIdle / 1048576.0:N0} MB | ≤ 80 MB + scripts |")
|
||||
.AppendLine($"| worker RSS with {scripts / 2} scripts loaded per shard (max) | {rssLoaded / 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);
|
||||
}
|
||||
}
|
||||
@@ -100,6 +100,55 @@ public sealed class SupervisionTests
|
||||
attributed.ShouldBeTrue("the runner's verdict is recorded before the callback returns");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A burst of cold compiles must not starve the worker's heartbeat. Found on windev by the §8 perf harness: 500
|
||||
/// loads at once, each compiled synchronously on its own thread-pool work item, queued the heartbeat timer and
|
||||
/// the pipe writer behind them; the host saw no frame for 3 s, killed the shard (KilledHeartbeat) and made all 500
|
||||
/// keys suspects. Here a 100 ms heartbeat (kill after 300 ms of silence) and 400 distinct cold scripts at once:
|
||||
/// every script answers and the shard's worker is the one that started. Mutation check: compile on the thread
|
||||
/// pool again and the worker is killed for silence.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_burst_of_cold_compiles_does_not_starve_the_heartbeat()
|
||||
{
|
||||
var sink = new CapturingSink();
|
||||
await using var runner = new OutOfProcessScriptRunner(Scripts.Options(), sink.CreateLogger(), null,
|
||||
heartbeatInterval: TimeSpan.FromMilliseconds(50), loadTimeout: TimeSpan.FromSeconds(120));
|
||||
(await runner.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7);
|
||||
var worker = runner.Shards[0].Current.ShouldNotBeNull();
|
||||
|
||||
const int count = 1500;
|
||||
var outcomes = await Task.WhenAll(Enumerable.Range(0, count).Select(i => runner.RunAsync(
|
||||
Scripts.Vt("burst-" + i, $"return (int)ctx.GetTag(\"x\").Value * {i} + 1;", TimeSpan.FromSeconds(60),
|
||||
Scripts.Input("x", 2)), Ct)));
|
||||
|
||||
var log = string.Join(Environment.NewLine, sink.Rendered.Where(l => l.Contains("ScriptWorker", StringComparison.Ordinal)).Take(5));
|
||||
for (var i = 0; i < count; i++) outcomes[i].Value.ShouldBe(2 * i + 1, $"{outcomes[i].Status} {outcomes[i].Reason}\n{log}");
|
||||
runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, log);
|
||||
worker.Exited.IsCompleted.ShouldBeFalse(log);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Single flight must not refuse a caller's NEXT run because of the run it already awaited: the slot is freed
|
||||
/// before the awaiting caller resumes. The §8 perf harness found the slot freed by a continuation unordered
|
||||
/// against the caller's await, so 32 of 200 back-to-back runs of one key came back "previous evaluation still
|
||||
/// running" — which the VirtualTag and Calculation seams publish as Bad. Mutation check: drop the inline
|
||||
/// release after the await and this goes red.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_caller_that_awaited_its_run_is_never_refused_by_it()
|
||||
{
|
||||
await using var runner = Scripts.Runner(Scripts.Options());
|
||||
var refused = new List<string>();
|
||||
for (var i = 0; i < 1_000; i++)
|
||||
{
|
||||
var outcome = await runner.RunAsync(
|
||||
Scripts.Vt("seq", "return (int)ctx.GetTag(\"x\").Value + 7;", inputs: Scripts.Input("x", i)), Ct);
|
||||
if (outcome.Status != ScriptRunStatus.Ok || !Equals(outcome.Value, i + 7)) refused.Add($"#{i}: {outcome.Reason}");
|
||||
}
|
||||
refused.ShouldBeEmpty();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The race itself, end to end: with QuarantineAfterKills = 1, the moment the killed script's single-flight slot is
|
||||
/// released the quarantine is already in place — so no trigger can re-run it in the restart gap, and the next
|
||||
|
||||
@@ -35,6 +35,16 @@ internal static class Scripts
|
||||
IReadOnlyDictionary<string, DataValueSnapshot>? inputs = null, ILogger? logger = null)
|
||||
=> new(key, ScriptContextKind.VirtualTag, source, inputs ?? NoInputs, timeout ?? TimeSpan.FromSeconds(5), logger);
|
||||
|
||||
/// <summary>
|
||||
/// Name of the environment variable that points the suite at a different worker entry than the one built beside
|
||||
/// the tests — e.g. the apphost of a self-contained <c>script-worker/</c> from a <c>-r win-x64</c> Host publish.
|
||||
/// </summary>
|
||||
public const string WorkerUnderTestVariable = "OTOPCUA_SCRIPTWORKER_TEST_PATH";
|
||||
|
||||
/// <summary>The worker entry named by <see cref="WorkerUnderTestVariable"/>, or null for the built one.</summary>
|
||||
public static string? WorkerUnderTest
|
||||
=> Environment.GetEnvironmentVariable(WorkerUnderTestVariable) is { Length: > 0 } path ? path : null;
|
||||
|
||||
/// <summary>Options for a single-shard runner with short budgets, so the containment tests run in seconds.</summary>
|
||||
public static ScriptingOptions Options(int shards = 1, TimeSpan? hardGrace = null, int quarantineAfterKills = 2,
|
||||
int probation = 20, TimeSpan? unavailableGrace = null, string? workerPath = null, long heapLimit = 256L * 1024 * 1024,
|
||||
@@ -50,7 +60,7 @@ internal static class Scripts
|
||||
QuarantineAfterKills = quarantineAfterKills,
|
||||
IsolationProbationRuns = probation,
|
||||
UnavailableGrace = unavailableGrace ?? TimeSpan.FromSeconds(30),
|
||||
ExecutablePath = workerPath,
|
||||
ExecutablePath = workerPath ?? WorkerUnderTest,
|
||||
HeapLimitBytes = heapLimit,
|
||||
PeCacheDir = peCacheDir,
|
||||
IsolationIdleTimeout = isolationIdle ?? ScriptWorkerOptions.DefaultIsolationIdleTimeout,
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
using System.Diagnostics;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
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: the Windows installs have no .NET runtime, so a self-contained Host publish ships a self-contained
|
||||
/// worker that the launcher must start through its apphost, while a framework-dependent worker (docker-dev, the
|
||||
/// tests) keeps going through the dotnet muxer.
|
||||
/// </summary>
|
||||
[Collection(ScriptWorkerCollection.Name)]
|
||||
[Trait("Category", "ScriptWorker")]
|
||||
public sealed class WorkerEntryTests
|
||||
{
|
||||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||||
|
||||
private const string SelfContainedConfig = """
|
||||
{ "runtimeOptions": { "tfm": "net10.0", "includedFrameworks": [ { "name": "Microsoft.NETCore.App", "version": "10.0.0" } ] } }
|
||||
""";
|
||||
|
||||
private const string FrameworkDependentConfig = """
|
||||
{ "runtimeOptions": { "tfm": "net10.0", "framework": { "name": "Microsoft.NETCore.App", "version": "10.0.0" } } }
|
||||
""";
|
||||
|
||||
private static string ConfigPath(string folder) => Path.Combine(folder, "ZB.MOM.WW.OtOpcUa.ScriptWorker.runtimeconfig.json");
|
||||
|
||||
private static void Touch(string path) => File.WriteAllText(path, "");
|
||||
|
||||
[Fact]
|
||||
public void A_self_contained_worker_is_started_through_its_apphost()
|
||||
{
|
||||
using var dir = new Scripts.TempDir();
|
||||
Touch(Path.Combine(dir.Path, OutOfProcessScriptRunner.WorkerAppHostFileName));
|
||||
Touch(Path.Combine(dir.Path, OutOfProcessScriptRunner.WorkerAssemblyFileName));
|
||||
File.WriteAllText(ConfigPath(dir.Path), SelfContainedConfig);
|
||||
|
||||
OutOfProcessScriptRunner.IsSelfContainedWorker(dir.Path).ShouldBeTrue();
|
||||
OutOfProcessScriptRunner.ResolveWorkerEntry(dir.Path)
|
||||
.ShouldBe(Path.Combine(Path.GetFullPath(dir.Path), OutOfProcessScriptRunner.WorkerAppHostFileName));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_framework_dependent_worker_keeps_the_muxer_even_with_an_apphost()
|
||||
{
|
||||
using var dir = new Scripts.TempDir();
|
||||
Touch(Path.Combine(dir.Path, OutOfProcessScriptRunner.WorkerAppHostFileName));
|
||||
Touch(Path.Combine(dir.Path, OutOfProcessScriptRunner.WorkerAssemblyFileName));
|
||||
File.WriteAllText(ConfigPath(dir.Path), FrameworkDependentConfig);
|
||||
|
||||
OutOfProcessScriptRunner.IsSelfContainedWorker(dir.Path).ShouldBeFalse();
|
||||
OutOfProcessScriptRunner.ResolveWorkerEntry(dir.Path).ShouldEndWith(OutOfProcessScriptRunner.WorkerAssemblyFileName);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(true, null)] // self-contained config but no apphost on disk
|
||||
[InlineData(false, "{ broken")] // apphost present, unreadable config
|
||||
[InlineData(false, null)] // apphost present, no config at all
|
||||
public void Anything_short_of_a_self_contained_apphost_falls_back_to_the_entry_assembly(bool omitAppHost, string? config)
|
||||
{
|
||||
using var dir = new Scripts.TempDir();
|
||||
if (!omitAppHost) Touch(Path.Combine(dir.Path, OutOfProcessScriptRunner.WorkerAppHostFileName));
|
||||
if (omitAppHost) File.WriteAllText(ConfigPath(dir.Path), SelfContainedConfig);
|
||||
else if (config is not null) File.WriteAllText(ConfigPath(dir.Path), config);
|
||||
|
||||
OutOfProcessScriptRunner.ResolveWorkerEntry(dir.Path).ShouldEndWith(OutOfProcessScriptRunner.WorkerAssemblyFileName);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void An_explicit_executable_path_wins()
|
||||
{
|
||||
var path = Path.Combine(Path.GetTempPath(), "custom", "worker.exe");
|
||||
OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = path }).ShouldBe(Path.GetFullPath(path));
|
||||
}
|
||||
|
||||
/// <summary>The worker built beside the tests is framework-dependent, so the default stays the muxer path.</summary>
|
||||
[Fact]
|
||||
public void The_built_worker_resolves_to_the_entry_assembly()
|
||||
{
|
||||
OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions())
|
||||
.ShouldEndWith(Path.Combine(OutOfProcessScriptRunner.WorkerFolderName, OutOfProcessScriptRunner.WorkerAssemblyFileName));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Runs only when <see cref="Scripts.WorkerUnderTestVariable"/> names a self-contained worker's apphost (the
|
||||
/// windev gate over a <c>-r win-x64 --self-contained</c> Host publish): the worker process IS the apphost — no
|
||||
/// dotnet muxer anywhere — and it evaluates.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_published_self_contained_worker_runs_as_its_own_process()
|
||||
{
|
||||
var path = Scripts.WorkerUnderTest;
|
||||
if (path is null || !path.EndsWith(OutOfProcessScriptRunner.WorkerAppHostFileName, StringComparison.OrdinalIgnoreCase))
|
||||
Assert.Skip($"set {Scripts.WorkerUnderTestVariable} to a self-contained worker apphost to run this gate");
|
||||
OutOfProcessScriptRunner.IsSelfContainedWorker(Path.GetDirectoryName(path)!).ShouldBeTrue("the published folder is self-contained");
|
||||
OutOfProcessScriptRunner.ResolveWorkerEntry(Path.GetDirectoryName(path)!).ShouldBe(Path.GetFullPath(path));
|
||||
|
||||
await using var runner = Scripts.Runner(Scripts.Options(workerPath: path));
|
||||
(await runner.RunAsync(Scripts.Vt("vt", "return System.Math.Abs(-3);"), Ct)).Value.ShouldBe(3);
|
||||
using var worker = Process.GetProcessById(runner.Shards[0].Current!.Pid);
|
||||
Path.GetFullPath(worker.MainModule!.FileName).ShouldBe(Path.GetFullPath(path), StringCompareShould.IgnoreCase);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user