diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs
index 06b748b4d..1d7556180 100644
--- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs
@@ -141,6 +141,9 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
private readonly string? _peCacheDirectory;
private readonly byte[]? _peCacheKey;
private readonly int _compileThreads;
+
+ /// Test seam: the launched workers' initial silence limit (null = three heartbeat intervals).
+ 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);
diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs
index a9fce4fa4..228068a9d 100644
--- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs
@@ -81,6 +81,10 @@ internal sealed record BatchSubmission(string Key, EvalRequest Request, TimeSpan
/// Windows only: the worker identity's SID (Scripting:Worker:Identity). Set, the worker starts under a restricted
/// token () and the pipe ACL admits it beside the host; null keeps the host's token.
///
+///
+/// How long the worker may send nothing before it is killed (); null =
+/// three s. Test seam (with ).
+///
///
/// Ask the worker to run the soft clock from each run's start (lmxopcua#827; see ).
/// 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)
{
/// The default compile timeout.
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 onLogs, TimeSpan heartbeatInterval, TimeSpan loadTimeout,
- Action onExiting, string? markerPath, bool workerSoftClock)
+ Action 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;
}
+ ///
+ /// 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.
+ ///
+ internal TimeSpan SilenceLimit
+ {
+ get => TimeSpan.FromTicks(Volatile.Read(ref _silenceLimitTicks));
+ set => Volatile.Write(ref _silenceLimitTicks, value.Ticks);
+ }
+
+ private long _silenceLimitTicks;
+
/// Compiles currently outstanding on this worker.
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);
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs
index 5f075fcfa..df214c4b6 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs
@@ -31,6 +31,27 @@ public sealed class SupervisionTests
private static CancellationToken Ct => TestContext.Current.CancellationToken;
+ ///
+ /// A runner whose workers heartbeat every 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). then tightens the limit to three intervals for the
+ /// phase under test.
+ ///
+ private static OutOfProcessScriptRunner TightHeartbeatRunner(TimeSpan heartbeat, TimeSpan loadTimeout, CapturingSink sink)
+ => new(Scripts.Options(), sink.CreateLogger(), null, heartbeatInterval: heartbeat, loadTimeout: loadTimeout,
+ heartbeatSilenceLimit: TimeSpan.FromSeconds(30));
+
+ /// Warms the shard's worker with , then arms the three-interval silence kill.
+ 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 =>