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

test(scriptworker): tight-heartbeat tests warm the worker before arming the 3-interval silence kill (#827 audit)

Browse Source 
A_burst_of_cold_compiles_does_not_starve_the_heartbeat (50 ms heartbeat) and
A_caller_that_blocks_after_its_result_does_not_starve_the_reader (100 ms)
launched their worker under the tight silence kill, so its cold start (launch,
first compile, Roslyn JIT) had to stay within 150/300 ms of silence. Under 28
busy processes on 14 cores that failed 2 in 10 runs in the WARM-UP, killed
for ~165 ms of silence, and replacement workers were killed the same way.

New test seam: WorkerLaunchSpec.HeartbeatSilenceLimit (runner internal ctor
heartbeatSilenceLimit) and WorkerConnection.SilenceLimit. The tests launch
with a 30 s limit, warm, then tighten to three intervals for the phase they
are about; the assertions on that phase are unchanged.
This commit is contained in:
Joseph Doherty
2026-11-11 01:59:11 -05:00
parent f7ca2c5826
commit 30c6c3a0bf
3 changed files with 59 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs
+7 -2
View File
@@ -141,6 +141,9 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
private readonly string? _peCacheDirectory;
private readonly byte[]? _peCacheKey;
private readonly int _compileThreads;
/// <summary>Test seam: the launched workers' initial silence limit (null = three heartbeat intervals).</summary>
private readonly TimeSpan? _heartbeatSilenceLimit;
private long _peCacheHits;
private long _peCacheMisses;
private long _peCacheRejected;
@@ -159,9 +162,10 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
internal OutOfProcessScriptRunner(ScriptingOptions options, ILogger? log, ILogger? scriptRoot,
TimeSpan heartbeatInterval, TimeSpan loadTimeout, TimeSpan? maintenanceInterval = null, string? markerDirectory = null,
int compileThreads = 0, int evalThreads = 0)
int compileThreads = 0, int evalThreads = 0, TimeSpan? heartbeatSilenceLimit = null)
{
_compileThreads = compileThreads;
_heartbeatSilenceLimit = heartbeatSilenceLimit;
ArgumentNullException.ThrowIfNull(options);
_options = options.Sanitized(out _);
_worker = _options.Worker;
@@ -719,7 +723,8 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
MarkerDirectory: _markerDirectory,
LoadTimeout: LoadTimeout,
CompileThreads: _compileThreads,
WorkerSid: WorkerSid);
WorkerSid: WorkerSid,
HeartbeatSilenceLimit: _heartbeatSilenceLimit);
return new WorkerShard(label, isolation,
(onExiting, ct) => WorkerConnection.StartAsync(spec, _log, OnLogs, onExiting, ct),
OnWorkerExit, _log);
src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs
+25 -4
View File
@@ -81,6 +81,10 @@ internal sealed record BatchSubmission(string Key, EvalRequest Request, TimeSpan
/// Windows only: the worker identity's SID (<c>Scripting:Worker:Identity</c>). Set, the worker starts under a restricted
/// token (<see cref="RestrictedWorkerLauncher"/>) and the pipe ACL admits it beside the host; null keeps the host's token.
/// </param>
/// <param name="HeartbeatSilenceLimit">
/// How long the worker may send nothing before it is killed (<see cref="WorkerExitReason.KilledHeartbeat"/>); null =
/// three <paramref name="HeartbeatInterval"/>s. Test seam (with <see cref="WorkerConnection.SilenceLimit"/>).
/// </param>
/// <param name="WorkerSoftClock">
/// Ask the worker to run the soft clock from each run's start (lmxopcua#827; see <see cref="WorkerConnection.WorkerSoftClock"/>).
/// False (tests only) keeps the pre-#827 host clock, which is also what a worker that does not echo it gets.
@@ -99,7 +103,8 @@ internal sealed record WorkerLaunchSpec(
TimeSpan? LoadTimeout = null,
int CompileThreads = 0,
string? WorkerSid = null,
bool WorkerSoftClock = true)
bool WorkerSoftClock = true,
TimeSpan? HeartbeatSilenceLimit = null)
{
/// <summary>The default compile timeout.</summary>
public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60);
@@ -156,7 +161,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
private WorkerConnection(string shard, Process process, NamedPipeServerStream pipe, IDisposable? job, IDisposable? stdio,
string workDir, ILogger log, Action<LogBatch> onLogs, TimeSpan heartbeatInterval, TimeSpan loadTimeout,
Action<WorkerExit> onExiting, string? markerPath, bool workerSoftClock)
Action<WorkerExit> onExiting, string? markerPath, bool workerSoftClock, TimeSpan? heartbeatSilenceLimit)
{
WorkerSoftClock = workerSoftClock;
_stdio = stdio;
@@ -171,6 +176,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
_log = log;
_onLogs = onLogs;
_heartbeatInterval = heartbeatInterval;
_silenceLimitTicks = (heartbeatSilenceLimit ?? heartbeatInterval * 3).Ticks;
Pid = process.Id;
// The handshake used seq 1. A failed write means the pipe is gone: the worker is dead or dying.
_writer = new ScriptWorkerFrameWriter(pipe, OnWriteFault, ScriptWorkerFrameCodec.DefaultMaxFrameBytes, firstSeq: 1,
@@ -215,6 +221,20 @@ internal sealed class WorkerConnection : IAsyncDisposable
set => _loadTimeout = value;
}
/// <summary>
/// How long the worker may send nothing before it is killed for silence: three heartbeat intervals unless the
/// launch spec said otherwise. Test hook: a test with a tight heartbeat launches and warms the worker (its cold
/// start JITs Roslyn, which on a loaded box can pause it for longer than three short intervals) under a generous
/// limit, then tightens it for the phase it is about.
/// </summary>
internal TimeSpan SilenceLimit
{
get => TimeSpan.FromTicks(Volatile.Read(ref _silenceLimitTicks));
set => Volatile.Write(ref _silenceLimitTicks, value.Ticks);
}
private long _silenceLimitTicks;
/// <summary>Compiles currently outstanding on this worker.</summary>
public int LoadingCount => _loads.Count;
@@ -327,7 +347,8 @@ internal sealed class WorkerConnection : IAsyncDisposable
: await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false);
var connection = new WorkerConnection(spec.Shard, process, pipe, job, stdio, workDir, log, onLogs, spec.HeartbeatInterval,
spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock);
spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock,
heartbeatSilenceLimit: spec.HeartbeatSilenceLimit);
connection.Begin();
return connection;
}
@@ -888,7 +909,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
{
if (_exited.Task.IsCompleted) return;
var silent = Stopwatch.GetElapsedTime(Volatile.Read(ref _lastFrameTimestamp));
if (silent > _heartbeatInterval * 3)
if (silent > SilenceLimit)
{
_log.Warning("ScriptWorker[{Shard}] pid {Pid} sent nothing for {Silent} ms; killing it", Shard, Pid, (long)silent.TotalMilliseconds);
Kill(WorkerExitReason.KilledHeartbeat, null);
tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs
+27 -8
View File
@@ -31,6 +31,27 @@ public sealed class SupervisionTests
private static CancellationToken Ct => TestContext.Current.CancellationToken;
/// <summary>
/// A runner whose workers heartbeat every <paramref name="heartbeat"/> but start with a generous silence limit:
/// the cold start (launch, first compile, Roslyn's JIT) is not what the tight-heartbeat tests are about, and on a
/// loaded box it can pause the worker for longer than three short intervals (lmxopcua#827 audit: 2 in 10 runs
/// under 28 busy processes on 14 cores were killed for 160 ms of silence during the warm-up compile, and the
/// replacement workers were too). <see cref="WarmTight"/> then tightens the limit to three intervals for the
/// phase under test.
/// </summary>
private static OutOfProcessScriptRunner TightHeartbeatRunner(TimeSpan heartbeat, TimeSpan loadTimeout, CapturingSink sink)
=> new(Scripts.Options(), sink.CreateLogger(), null, heartbeatInterval: heartbeat, loadTimeout: loadTimeout,
heartbeatSilenceLimit: TimeSpan.FromSeconds(30));
/// <summary>Warms the shard's worker with <paramref name="key"/>, then arms the three-interval silence kill.</summary>
private static WorkerConnection WarmTight(OutOfProcessScriptRunner runner, ScriptRunOutcome warm, TimeSpan heartbeat, CapturingSink sink)
{
warm.Value.ShouldBe(7, $"{warm.Status} {warm.Reason}\n{string.Join(Environment.NewLine, sink.Rendered)}");
var worker = runner.Shards[0].Current.ShouldNotBeNull();
worker.SilenceLimit = heartbeat * 3;
return worker;
}
private static string VtKey(string key) => InProcessScriptRunner.ScopedKey(ScriptContextKind.VirtualTag, key);
// ---------------------------------------------------------------- the quarantine race
@@ -112,10 +133,9 @@ public sealed class SupervisionTests
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();
var heartbeat = TimeSpan.FromMilliseconds(50);
await using var runner = TightHeartbeatRunner(heartbeat, TimeSpan.FromSeconds(120), sink);
var worker = WarmTight(runner, await runner.RunAsync(Scripts.Vt("first", Healthy), Ct), heartbeat, sink);
const int count = 1500;
var outcomes = await Task.WhenAll(Enumerable.Range(0, count).Select(i => runner.RunAsync(
@@ -514,10 +534,9 @@ public sealed class SupervisionTests
public async Task A_caller_that_blocks_after_its_result_does_not_starve_the_reader()
{
var sink = new CapturingSink();
await using var runner = new OutOfProcessScriptRunner(Scripts.Options(), sink.CreateLogger(), null,
heartbeatInterval: TimeSpan.FromMilliseconds(100), loadTimeout: TimeSpan.FromSeconds(60));
(await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct)).Value.ShouldBe(7);
var worker = runner.Shards[0].Current.ShouldNotBeNull();
var heartbeat = TimeSpan.FromMilliseconds(100);
await using var runner = TightHeartbeatRunner(heartbeat, TimeSpan.FromSeconds(60), sink);
var worker = WarmTight(runner, await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct), heartbeat, sink);
string? resumedOn = null;
await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct).ContinueWith(t =>