Merge fix/827-worker-deadline
This commit is contained in:
@@ -398,7 +398,7 @@ lists `ScriptedAlarm` — never by default) in `ZB.MOM.WW.OtOpcUa.ScriptWorker`
|
||||
(`src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess`, protocol in `Core.Scripting.Protocol`), shipped in `script-worker/`
|
||||
beside the host (self-contained + started via its apphost on a `-r win-x64 --self-contained` publish, through the muxer otherwise); a runner-side failure publishes `BadResourceUnavailable` (ScadaBridge-visible, ScadaBridge#112) and the seam re-evaluates
|
||||
when the worker is back. `IScriptRunner.RunBatch` sends ready evaluations as one frame per worker (Calc pump, VT coalescer,
|
||||
alarm dirty set); a run queued behind busy worker threads has its deadlines restarted when it starts, so it is never blamed for a sibling. Optional `Scripting:Worker:PeCacheDir` = HMAC'd compiled-PE cache (restart without Roslyn); optional (Windows) `Scripting:Worker:Identity` = `NT SERVICE\OtOpcUaScriptWorker` runs workers under a restricted token, set up by `scripts/install/Install-ScriptWorkerIsolation.ps1` (ACLs + outbound firewall block; `docs/ServiceHosting.md`); the `script-runner`
|
||||
alarm dirty set); a run queued behind busy worker threads has its deadlines restarted when it starts, so it is never blamed for a sibling. The soft deadline is the **worker's** clock from the run's start (#827, negotiated via `Hello`/`Ready.worker_soft_clock`); wall clock from there, so a pause during the run still counts. The host keeps a T + G transport deadline (passing it is a runner fault: `script worker unresponsive` → `BadResourceUnavailable`, Shadow `shadow_unavailable`; a later soft notice still cancels and counts the timeout) and a progress ceiling (`killed_unresponsive`) so it never depends on the worker's notices. Optional `Scripting:Worker:PeCacheDir` = HMAC'd compiled-PE cache (restart without Roslyn); optional (Windows) `Scripting:Worker:Identity` = `NT SERVICE\OtOpcUaScriptWorker` runs workers under a restricted token, set up by `scripts/install/Install-ScriptWorkerIsolation.ps1` (ACLs + outbound firewall block; `docs/ServiceHosting.md`); the `script-runner`
|
||||
readiness check is at worst Degraded (never fails ready). The worker tests
|
||||
are `tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests` (`Category=ScriptWorker`, real processes; `Category=Perf` harness opt-in via
|
||||
`OTOPCUA_SCRIPT_PERF=1`). docker-dev overlay: `docker-dev/docker-compose.script-runner.yml`. See `docs/VirtualTags.md`
|
||||
|
||||
+24
-2
@@ -58,6 +58,27 @@ processes:
|
||||
`script timed out after …` failure and the run token is cancelled inside the worker; at `T + G`
|
||||
(`Scripting:Worker:HardGrace`, default `max(1 s, T)`) the worker is **killed** (`Process.Kill(entireProcessTree)`,
|
||||
plus a worker-side fail-fast watchdog) — the thread, its memory and its ALC go with it.
|
||||
- **`T` starts when the run starts (lmxopcua#827).** As in process (where the guard's clock starts when the thread
|
||||
pool picks the run up), `T` starts when a worker evaluation thread starts the run — not at submit — so the pipe
|
||||
round trip and the time a starved worker takes to **start** the run never count against the script. From the start
|
||||
on it is wall-clock time, exactly as in process: a GC pause, a CPU-starved or suspended worker **during** the run
|
||||
still counts against `T` (not separated, and not tested beyond that equivalence). The worker runs that clock and
|
||||
reports the crossing (`EvalResults.soft_expired_eval_ids`); the host resumes its caller on the result dispatcher
|
||||
(never on the pipe reader), sends the cancel, and arms the kill `G` after the notice.
|
||||
- **Transport deadline.** What the host still times on its own is `T + G` from submit (restarted at a queued run's
|
||||
start): no result and no soft notice by then is the runner's fault — the tag publishes **`BadResourceUnavailable`**
|
||||
(`"script worker unresponsive"`), nothing is struck, the worker is not killed, and the seam re-evaluates when the
|
||||
worker answers. A re-trigger while that run is still unanswered gets the same answer, not
|
||||
`previous run still running`. If the worker then reports the script over `T` after all (slow **and** late — a
|
||||
notice delayed past `T + G` lands here too), the run still gets its cancel and counts as a timeout. Counted as
|
||||
`otopcua.script.ipc.timeouts{kind, answer = stalled | retrigger}`.
|
||||
- **Progress ceiling.** The host does not depend on the worker's notices to end a run: while evaluations are
|
||||
outstanding, a worker that keeps heartbeating but produces no result, start or soft notice for
|
||||
`3 × the largest outstanding T + G + 3 heartbeat intervals` is killed (`killed_unresponsive`, no culprit,
|
||||
attributed like a heartbeat kill). A worker that stops heartbeating is killed by the silence rule as before.
|
||||
- **Negotiated.** The clock is agreed in the handshake (`Hello`/`Ready.worker_soft_clock`): a worker that does not
|
||||
echo it keeps the host clock from submit (the pre-#827 contract), so a mismatched host/worker pair degrades to the
|
||||
old behaviour instead of losing its deadlines.
|
||||
- **Attribution and quarantine** live in the host and survive restarts: a deadline kill strikes exactly its key
|
||||
(other evaluations in flight are victims, not strikes); a crash with no deadline crossed makes every key in flight
|
||||
a suspect that must complete `Scripting:Worker:IsolationProbationRuns` (20) clean runs in the isolation worker,
|
||||
@@ -108,7 +129,7 @@ processes:
|
||||
"every process with this name", never a live host's worker (mxaccessgw's `OrphanWorkerTerminator` lesson).
|
||||
- **Observability.** On `OtOpcUaTelemetry.Meter`: `otopcua.script.runner.state` (0 ready / 1 degraded / 2 circuit
|
||||
open), `otopcua.script.worker.up` / `.inflight` / `.queue_depth` / `.heap_bytes` (tag `shard`; queue and heap from
|
||||
the worker heartbeat), `otopcua.script.worker.starts` / `.exits` (`reason` adds `idle`), `otopcua.script.ipc.roundtrip`
|
||||
the worker heartbeat), `otopcua.script.worker.starts` / `.exits` (`reason` adds `idle`), `otopcua.script.ipc.roundtrip`, `otopcua.script.ipc.timeouts` (transport deadline, lmxopcua#827)
|
||||
and `otopcua.script.eval.duration` (µs, tag `kind`; no per-key tag), `otopcua.script.compile` (`cache` = `hit` |
|
||||
`miss` | `rejected` | `uncached` | `negative`) and `otopcua.script.isolation.suspects`. The `script-runner` health check
|
||||
is on the **ready** tier and is at worst **Degraded** (still 200): a sick runner never fails readiness, so a pair
|
||||
@@ -168,7 +189,8 @@ what is published** — and each compiled evaluation is then mirrored to the `Sc
|
||||
|
||||
- **Compared:** value (structural for arrays), exact CLR type, status (Ok vs failed) and failure class (`compile`,
|
||||
`budget` = timed out / killed / refused / quarantined, `threw` with its message, `unsupported_type`). A shadow outcome
|
||||
that is the runner's own fault (worker restarted, none ready, worker busy) is **not compared**
|
||||
that is the runner's own fault (worker restarted, none ready, worker busy, worker unresponsive — a stalled or starved
|
||||
worker past the transport deadline, lmxopcua#827) is **not compared**
|
||||
(`result = shadow_unavailable`). An input or result the protocol cannot carry (design §14 #11) counts as a mismatch
|
||||
with reason `unsupported_type` — out of process that evaluation would publish Bad.
|
||||
- **Never reaches a published value or readiness.** The mirror copies the inputs and queues the shadow run on the pool;
|
||||
|
||||
+71
-11
@@ -100,6 +100,10 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
|
||||
"otopcua.script.eval.duration", unit: "us",
|
||||
description: "Worker-measured run time of one out-of-process evaluation (tag: kind).");
|
||||
|
||||
private static readonly Counter<long> TransportTimeoutCounter = OtOpcUaTelemetry.Meter.CreateCounter<long>(
|
||||
"otopcua.script.ipc.timeouts", unit: "{eval}",
|
||||
description: "Out-of-process evaluations answered 'script worker unresponsive' (lmxopcua#827; tags: kind, answer = stalled | retrigger). stalled: no result and no soft-deadline notice from the worker within T + G of submit — a starved or stalled worker, not a slow script; retrigger: a later trigger of the same script refused while that stalled run is still unanswered.");
|
||||
|
||||
static OutOfProcessScriptRunner()
|
||||
{
|
||||
// Gauges observe every live runner (one per node in production; tests make several).
|
||||
@@ -137,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;
|
||||
@@ -155,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;
|
||||
@@ -372,11 +380,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
|
||||
return ScriptRunOutcome.Failed(knownFailure);
|
||||
}
|
||||
|
||||
if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0)
|
||||
{
|
||||
ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair<string, object?>("script", key));
|
||||
return ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason);
|
||||
}
|
||||
if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0) return Refused(key, state);
|
||||
|
||||
var handedOff = false;
|
||||
SemaphoreSlim? gate = null;
|
||||
@@ -533,8 +537,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
|
||||
|
||||
if (Interlocked.CompareExchange(ref state.InFlight, 1, 0) != 0)
|
||||
{
|
||||
ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair<string, object?>("script", key));
|
||||
answered = ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason);
|
||||
answered = Refused(key, state);
|
||||
return BatchPreparation.Answered;
|
||||
}
|
||||
state.ScriptLogger = invocation.ScriptLogger;
|
||||
@@ -559,6 +562,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
|
||||
{
|
||||
if (Interlocked.Exchange(ref released, 1) != 0) return;
|
||||
SlotReleasingForTest?.Invoke(key);
|
||||
Volatile.Write(ref state.StalledRun, null);
|
||||
Interlocked.Exchange(ref state.InFlight, 0);
|
||||
heldGate?.Release();
|
||||
};
|
||||
@@ -583,6 +587,33 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
|
||||
return ScriptRunOutcome.Failed($"{ScriptEvalFailureReasons.TimedOutPrefix}{invocation.Timeout.TotalSeconds:F1}s");
|
||||
}
|
||||
|
||||
if (ReferenceEquals(result, PendingEval.TransportTimedOut))
|
||||
{
|
||||
// lmxopcua#827: T + G passed with neither a result nor a soft-deadline notice. The worker runs the soft
|
||||
// clock from the run's start, so this is the pipe or a starved / stalled worker — the runner's fault,
|
||||
// not (yet) the script's: no timeout counted, no Cancel (the run may not even have started), and the
|
||||
// slot stays with the run until the worker answers. The seam re-evaluates when it does (or when the
|
||||
// worker is replaced, if the heartbeat monitor kills it).
|
||||
TransportTimeoutCounter.Add(1, new KeyValuePair<string, object?>("kind", KindTag(invocation.Kind)),
|
||||
new KeyValuePair<string, object?>("answer", "stalled"));
|
||||
// …unless the worker then reports the script itself over T (slow AND late — a notice delayed past
|
||||
// T + G by the watchdog poll and the pipe lands here too): it still gets its Cancel, so a polling
|
||||
// script ends itself instead of costing the whole worker at notice + G, and it counts as a timeout
|
||||
// (#827 review F2). Off the reader: the completion source runs this asynchronously.
|
||||
var timeout = invocation.Timeout;
|
||||
pending.LateSoftExpiry?.ContinueWith(_ =>
|
||||
{
|
||||
connection.Cancel(pending);
|
||||
NoteTimeout(key, state, timeout);
|
||||
}, CancellationToken.None, TaskContinuationOptions.OnlyOnRanToCompletion | TaskContinuationOptions.ExecuteSynchronously,
|
||||
TaskScheduler.Default);
|
||||
// Read by a refusal only while it is incomplete, so a write that loses a race with the slot release
|
||||
// is harmless.
|
||||
Volatile.Write(ref state.StalledRun, pending.Completion);
|
||||
if (ct.IsCancellationRequested) return ScriptRunOutcome.Failed("evaluation cancelled");
|
||||
return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason) with { Recovery = pending.Completion };
|
||||
}
|
||||
|
||||
if (result is null) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection);
|
||||
var kindTag = new KeyValuePair<string, object?>("kind", KindTag(invocation.Kind));
|
||||
RoundTripHistogram.Record(Stopwatch.GetElapsedTime(submitted).TotalMicroseconds, kindTag);
|
||||
@@ -704,7 +735,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);
|
||||
@@ -717,6 +749,25 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
|
||||
private static ScriptRunOutcome Unavailable(string reason, WorkerShard shard, WorkerConnection? failed)
|
||||
=> ScriptRunOutcome.Unavailable(reason) with { Recovery = shard.WhenRecoveredAsync(failed) };
|
||||
|
||||
/// <summary>
|
||||
/// The answer to an evaluation whose key's previous run still holds its single-flight slot: the usual
|
||||
/// script-attributable refusal — unless that run was answered "worker unresponsive" (lmxopcua#827) and the
|
||||
/// worker still has not answered it, in which case nothing says the script is slow, and the refusal is the
|
||||
/// runner's fault too (re-evaluated when the stalled run completes).
|
||||
/// </summary>
|
||||
private static ScriptRunOutcome Refused(string key, KeyState state)
|
||||
{
|
||||
if (Volatile.Read(ref state.StalledRun) is { IsCompleted: false } stalled)
|
||||
{
|
||||
// Counted, like the run it waits on (#827 review F5): it is not a script refusal, so not ScriptRunGuard's.
|
||||
TransportTimeoutCounter.Add(1, new KeyValuePair<string, object?>("kind", KindTag(state.Kind)),
|
||||
new KeyValuePair<string, object?>("answer", "retrigger"));
|
||||
return ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason) with { Recovery = stalled };
|
||||
}
|
||||
ScriptRunGuard.RefusedCounter.Add(1, new KeyValuePair<string, object?>("script", key));
|
||||
return ScriptRunOutcome.Failed(ScriptRunGuard.PreviousRunStillRunningReason);
|
||||
}
|
||||
|
||||
private void NoteLoad(WorkerConnection.LoadEntry entry)
|
||||
{
|
||||
// Counted once per compile, by the caller that waited for it (an already-loaded key is not a load).
|
||||
@@ -864,7 +915,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
|
||||
case WorkerExitReason.KilledLoadTimeout when exit.CulpritKey is { } culprit:
|
||||
KillStrike(culprit, "its compile did not finish and its worker was killed", "killed");
|
||||
break;
|
||||
case WorkerExitReason.Crash or WorkerExitReason.KilledHeartbeat or WorkerExitReason.ProtocolFault:
|
||||
case WorkerExitReason.Crash or WorkerExitReason.KilledHeartbeat or WorkerExitReason.KilledUnresponsive or WorkerExitReason.ProtocolFault:
|
||||
// Compiles count as well as evaluations: a source that crashes Roslyn (or the load) must be found too.
|
||||
var implicated = exit.InFlightKeys.Concat(exit.LoadingKeys).Distinct(StringComparer.Ordinal).ToArray();
|
||||
if (shard.IsIsolation)
|
||||
@@ -968,6 +1019,9 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
|
||||
|
||||
public readonly string Source = source;
|
||||
|
||||
/// <summary>The seam (metric tags).</summary>
|
||||
public readonly ScriptContextKind Kind = kind;
|
||||
|
||||
/// <summary>SHA-256 of <see cref="Source"/> (computed on first use; a benign race computes the same bytes).</summary>
|
||||
public byte[] Sha => _sha ??= SHA256.HashData(Encoding.UTF8.GetBytes(Source));
|
||||
|
||||
@@ -975,7 +1029,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
|
||||
public string ShaHex => _shaHex ??= Convert.ToHexString(Sha);
|
||||
|
||||
/// <summary>The negative compile cache's key for this source (kind-scoped).</summary>
|
||||
public string FailureKey => _failureKey ??= InProcessScriptRunner.ScopedKey(kind, ShaHex);
|
||||
public string FailureKey => _failureKey ??= InProcessScriptRunner.ScopedKey(Kind, ShaHex);
|
||||
|
||||
/// <summary>The slot order last used for this key (see <see cref="SlotOrder"/>).</summary>
|
||||
public string[]? Paths;
|
||||
@@ -986,5 +1040,11 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
|
||||
public volatile bool Suspect;
|
||||
public string? QuarantineReason;
|
||||
public ILogger? ScriptLogger;
|
||||
|
||||
/// <summary>
|
||||
/// The completion of this key's in-flight run while it is answered "worker unresponsive" and not yet answered
|
||||
/// by the worker (lmxopcua#827); null otherwise. Cleared by the slot release.
|
||||
/// </summary>
|
||||
public Task? StalledRun;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,6 +21,12 @@ internal enum WorkerExitReason
|
||||
/// <summary>The host killed it: no frame for three heartbeat intervals.</summary>
|
||||
KilledHeartbeat,
|
||||
|
||||
/// <summary>
|
||||
/// The host killed it: it kept heartbeating but answered nothing — no result, no start, no soft-deadline notice —
|
||||
/// for the progress ceiling while evaluations were outstanding (#827 review F3). Attributed like a heartbeat kill.
|
||||
/// </summary>
|
||||
KilledUnresponsive,
|
||||
|
||||
/// <summary>The host killed it: a compile did not finish in time (<see cref="WorkerExit.CulpritKey"/>).</summary>
|
||||
KilledLoadTimeout,
|
||||
|
||||
@@ -81,6 +87,14 @@ 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.
|
||||
/// </param>
|
||||
internal sealed record WorkerLaunchSpec(
|
||||
string Shard,
|
||||
string WorkerPath,
|
||||
@@ -94,7 +108,9 @@ internal sealed record WorkerLaunchSpec(
|
||||
string? MarkerDirectory = null,
|
||||
TimeSpan? LoadTimeout = null,
|
||||
int CompileThreads = 0,
|
||||
string? WorkerSid = null)
|
||||
string? WorkerSid = null,
|
||||
bool WorkerSoftClock = true,
|
||||
TimeSpan? HeartbeatSilenceLimit = null)
|
||||
{
|
||||
/// <summary>The default compile timeout.</summary>
|
||||
public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60);
|
||||
@@ -145,14 +161,21 @@ internal sealed class WorkerConnection : IAsyncDisposable
|
||||
private int _finished;
|
||||
private (WorkerExitReason Reason, string? Culprit)? _killedFor;
|
||||
private long _lastFrameTimestamp;
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="Stopwatch.GetTimestamp"/> of the last evaluation progress — a Results frame of any kind — or of the
|
||||
/// submit that made the worker busy again. See <see cref="ProgressCeiling"/>.
|
||||
/// </summary>
|
||||
private long _lastProgressTimestamp;
|
||||
private long _nextEvalId;
|
||||
private long _nextLoadId;
|
||||
private int _disposed;
|
||||
|
||||
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)
|
||||
Action<WorkerExit> onExiting, string? markerPath, bool workerSoftClock, TimeSpan? heartbeatSilenceLimit)
|
||||
{
|
||||
WorkerSoftClock = workerSoftClock;
|
||||
_stdio = stdio;
|
||||
_loadTimeout = loadTimeout;
|
||||
_onExiting = onExiting;
|
||||
@@ -165,6 +188,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,
|
||||
@@ -178,6 +202,16 @@ internal sealed class WorkerConnection : IAsyncDisposable
|
||||
/// <summary>The worker's process id.</summary>
|
||||
public int Pid { get; }
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#827: the worker runs each evaluation's soft clock from the moment an evaluation thread starts it and
|
||||
/// reports the crossing, so the soft deadline no longer counts the pipe or the time a starved worker takes to
|
||||
/// START the run. From the start it is wall-clock time, as in process: a pause during the run still counts. The
|
||||
/// host keeps a transport deadline of its own instead (see
|
||||
/// <see cref="PendingEval"/>). Negotiated: asked for in <c>Hello</c>, true only when the worker echoed it in
|
||||
/// <c>Ready</c>; otherwise the host's clock, started at submit, is the soft deadline (the pre-#827 contract).
|
||||
/// </summary>
|
||||
public bool WorkerSoftClock { get; }
|
||||
|
||||
/// <summary>Completes when the worker has ended, with why.</summary>
|
||||
public Task<WorkerExit> Exited => _exited.Task;
|
||||
|
||||
@@ -200,6 +234,46 @@ 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>
|
||||
/// The host's own bound on a worker that heartbeats but makes no evaluation progress (#827 review F3). With the
|
||||
/// worker soft clock the host's per-run timer is only the transport deadline, and the kill G is armed by the
|
||||
/// worker's soft notice — so a worker whose watchdog never sends one (a bug; the design treats the worker as
|
||||
/// untrusted) would hold a key's slot, and its tag at BadResourceUnavailable, forever; a deferred run has no host
|
||||
/// clock at all. So: while evaluations are outstanding and none of them has produced a result, start or soft
|
||||
/// notice for 3 × the largest outstanding T + G plus the silence limit, the worker is killed
|
||||
/// (<see cref="WorkerExitReason.KilledUnresponsive"/>, no culprit: attributed like a heartbeat kill, so no key is
|
||||
/// struck on a shard). A legitimate run never gets near it: by T + G it has answered or been killed. Null = that
|
||||
/// rule; set = an override (tests).
|
||||
/// </summary>
|
||||
internal TimeSpan? ProgressCeilingOverride { get; set; }
|
||||
|
||||
/// <summary>Test seam: ignore the worker's soft-deadline notices, as a worker whose watchdog never sends them would.</summary>
|
||||
internal bool DropSoftNoticesForTest { get; set; }
|
||||
|
||||
/// <summary>The progress ceiling now (see <see cref="ProgressCeilingOverride"/>); null = no outstanding deadline.</summary>
|
||||
private TimeSpan? ProgressCeiling()
|
||||
{
|
||||
if (ProgressCeilingOverride is { } fixedCeiling) return fixedCeiling;
|
||||
var largest = TimeSpan.Zero;
|
||||
foreach (var (_, pending) in _pending)
|
||||
if (pending.HardDeadline > largest) largest = pending.HardDeadline;
|
||||
return largest == TimeSpan.Zero ? null : (largest * 3) + SilenceLimit;
|
||||
}
|
||||
|
||||
/// <summary>Compiles currently outstanding on this worker.</summary>
|
||||
public int LoadingCount => _loads.Count;
|
||||
|
||||
@@ -300,19 +374,20 @@ internal sealed class WorkerConnection : IAsyncDisposable
|
||||
HeartbeatIntervalMs = (uint)spec.HeartbeatInterval.TotalMilliseconds,
|
||||
MaxLogLines = 1000,
|
||||
CompileThreads = (uint)Math.Max(0, spec.CompileThreads),
|
||||
WorkerSoftClock = spec.WorkerSoftClock,
|
||||
};
|
||||
if (spec.PeCacheDirectory is { } cacheDir && spec.PeCacheKey is { } cacheKey)
|
||||
{
|
||||
hello.PeCacheDir = cacheDir;
|
||||
hello.PeCacheKey = Google.Protobuf.ByteString.CopyFrom(cacheKey);
|
||||
}
|
||||
if (ScriptWorkerFrameCodec.UseBlockingIo)
|
||||
await HandshakeBlockingAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false);
|
||||
else
|
||||
await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false);
|
||||
var ready = ScriptWorkerFrameCodec.UseBlockingIo
|
||||
? await HandshakeBlockingAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false)
|
||||
: 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);
|
||||
spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock,
|
||||
heartbeatSilenceLimit: spec.HeartbeatSilenceLimit);
|
||||
connection.Begin();
|
||||
return connection;
|
||||
}
|
||||
@@ -556,8 +631,9 @@ internal sealed class WorkerConnection : IAsyncDisposable
|
||||
{
|
||||
request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId);
|
||||
Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp());
|
||||
if (_pending.IsEmpty) Interlocked.Exchange(ref _lastProgressTimestamp, Stopwatch.GetTimestamp());
|
||||
var pending = new PendingEval(key, request.EvalId, onFinished, _dispatcher);
|
||||
if (softTimeout is { } soft) pending.Arm(soft, hardGrace, OnHardDeadline);
|
||||
if (softTimeout is { } soft) pending.Arm(soft, hardGrace, OnHardDeadline, WorkerSoftClock);
|
||||
_pending[request.EvalId] = pending;
|
||||
var batch = new EvalBatch();
|
||||
batch.Items.Add(request);
|
||||
@@ -581,12 +657,13 @@ internal sealed class WorkerConnection : IAsyncDisposable
|
||||
var pendings = new PendingEval[items.Count];
|
||||
var batch = new EvalBatch();
|
||||
Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp());
|
||||
if (_pending.IsEmpty) Interlocked.Exchange(ref _lastProgressTimestamp, Stopwatch.GetTimestamp());
|
||||
for (var i = 0; i < items.Count; i++)
|
||||
{
|
||||
var item = items[i];
|
||||
item.Request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId);
|
||||
var pending = new PendingEval(item.Key, item.Request.EvalId, item.OnFinished, _dispatcher);
|
||||
pending.Arm(item.SoftTimeout, item.HardGrace, OnHardDeadline);
|
||||
pending.Arm(item.SoftTimeout, item.HardGrace, OnHardDeadline, WorkerSoftClock);
|
||||
_pending[item.Request.EvalId] = pending;
|
||||
batch.Items.Add(item.Request);
|
||||
pendings[i] = pending;
|
||||
@@ -764,6 +841,18 @@ internal sealed class WorkerConnection : IAsyncDisposable
|
||||
switch (frame.BodyCase)
|
||||
{
|
||||
case Envelope.BodyOneofCase.Results:
|
||||
Volatile.Write(ref _lastProgressTimestamp, Stopwatch.GetTimestamp());
|
||||
if (DropSoftNoticesForTest) frame.Results.SoftExpiredEvalIds.Clear();
|
||||
// Worker soft clock (lmxopcua#827): the script itself has run for T. Only a worker that agreed to run
|
||||
// the clock may send these. Before the results: the writer coalesces frames, and a notice the worker
|
||||
// sent ahead of the same run's result must still answer the caller with its timeout, as in process.
|
||||
if (frame.Results.SoftExpiredEvalIds.Count > 0 && !WorkerSoftClock)
|
||||
throw new ScriptWorkerProtocolException("soft-deadline notice from a worker that does not run the soft clock");
|
||||
foreach (var id in frame.Results.SoftExpiredEvalIds)
|
||||
{
|
||||
if (_pending.TryGetValue(id, out var expired)) expired.SoftExpired();
|
||||
else CheckIssued(id, "soft-deadline notice");
|
||||
}
|
||||
foreach (var result in frame.Results.Items)
|
||||
{
|
||||
// An eval id with no outstanding request is a protocol violation (design §6 "what the host
|
||||
@@ -863,12 +952,23 @@ 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);
|
||||
return;
|
||||
}
|
||||
if (!_pending.IsEmpty && ProgressCeiling() is { } ceiling)
|
||||
{
|
||||
var stuck = Stopwatch.GetElapsedTime(Volatile.Read(ref _lastProgressTimestamp));
|
||||
if (stuck > ceiling)
|
||||
{
|
||||
_log.Warning("ScriptWorker[{Shard}] pid {Pid} has {Count} evaluation(s) outstanding and made no progress for {Stuck} ms (ceiling {Ceiling} ms) while still heartbeating; killing it",
|
||||
Shard, Pid, _pending.Count, (long)stuck.TotalMilliseconds, (long)ceiling.TotalMilliseconds);
|
||||
Kill(WorkerExitReason.KilledUnresponsive, null);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) { }
|
||||
@@ -1026,18 +1126,45 @@ internal sealed class WorkerConnection : IAsyncDisposable
|
||||
/// <summary>
|
||||
/// One evaluation outstanding on a worker. Its deadlines share ONE timer (lmxopcua#789: the per-eval
|
||||
/// <c>WaitAsync(timeout)</c> timer, the hard-deadline <c>CancellationTokenSource</c> and their continuations were a
|
||||
/// measurable share of the host's CPU): it fires at the soft deadline T and is re-armed for the grace G.
|
||||
/// measurable share of the host's CPU).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The deadlines are absolute timestamps the timer callback re-checks, so a deferral (the worker queued the run behind
|
||||
/// busy evaluation threads, lmxopcua#780 batch submit) can stop the clock and a start notice can restart it — the full
|
||||
/// T, then G, from the start — without a stale tick firing early.
|
||||
/// <para>
|
||||
/// <b>Worker soft clock</b> (lmxopcua#827, <see cref="WorkerConnection.WorkerSoftClock"/>): the soft deadline T is
|
||||
/// the worker's — it reports, through <see cref="SoftExpired"/>, that the script has itself run for T, and the
|
||||
/// hard deadline G is armed from that notice. The host's timer is then only a TRANSPORT deadline: T + G from
|
||||
/// submit (restarted at a deferred run's start). Passing it with neither a result nor a soft notice means the
|
||||
/// pipe or the worker process is late, not the script — <see cref="Outcome"/> yields
|
||||
/// <see cref="TransportTimedOut"/> (a runner fault: nothing is struck and the worker is not killed; a worker that
|
||||
/// is really hung goes silent and the heartbeat monitor kills it). So a starved worker can no longer make a
|
||||
/// trivial script report "timed out", which in Shadow mode read as a script mismatch.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Host clock</b> (a worker that did not agree to the above): the timer fires at the soft deadline T, counted
|
||||
/// from submit, and is re-armed for the grace G.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The deadlines are absolute timestamps the timer callback re-checks, so a deferral (the worker queued the run
|
||||
/// behind busy evaluation threads, lmxopcua#780 batch submit) can stop the clock and a start notice can restart
|
||||
/// it — the full T (or transport deadline), then G, from the start — without a stale tick firing early.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal sealed class PendingEval
|
||||
{
|
||||
/// <summary>What <see cref="Outcome"/> yields when the soft deadline passed first. Never on the wire.</summary>
|
||||
internal static readonly EvalResult SoftTimedOut = new();
|
||||
|
||||
/// <summary>
|
||||
/// What <see cref="Outcome"/> yields when the transport deadline passed with no word from the worker (worker soft
|
||||
/// clock only, lmxopcua#827): a runner fault, not the script's. Never on the wire.
|
||||
/// </summary>
|
||||
internal static readonly EvalResult TransportTimedOut = new();
|
||||
|
||||
private const int Waiting = 0;
|
||||
private const int SoftReported = 1;
|
||||
private const int HardFired = 2;
|
||||
private const int TransportReported = 3;
|
||||
|
||||
private readonly TaskCompletionSource<EvalResult?> _completion = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private readonly TaskCompletionSource<EvalResult?> _outcome;
|
||||
private readonly ResultDispatcher? _dispatcher;
|
||||
@@ -1047,12 +1174,15 @@ internal sealed class PendingEval
|
||||
private Action<PendingEval>? _onHardDeadline;
|
||||
private TimeSpan _soft;
|
||||
private TimeSpan _grace;
|
||||
private bool _workerClock;
|
||||
private long _softDue;
|
||||
private long _hardDue;
|
||||
private long _transportDue;
|
||||
private bool _deferred;
|
||||
private bool _started;
|
||||
private int _phase;
|
||||
private int _done;
|
||||
private TaskCompletionSource? _lateSoftExpiry;
|
||||
|
||||
public PendingEval(string key, ulong evalId, Action? onFinished = null, ResultDispatcher? dispatcher = null)
|
||||
{
|
||||
@@ -1086,24 +1216,41 @@ internal sealed class PendingEval
|
||||
public Task<EvalResult?> Completion => _completion.Task;
|
||||
|
||||
/// <summary>
|
||||
/// What the caller waits on: the worker's result, null when the worker ended first, or
|
||||
/// <see cref="SoftTimedOut"/> at the soft deadline (the run itself is still outstanding then).
|
||||
/// After <see cref="TransportTimedOut"/> (null before): completes when the worker then reports the script itself
|
||||
/// over T (slow AND late: it still gets its Cancel and counts as a timeout, #827 review F2), and is cancelled
|
||||
/// when the worker answers or ends first. Continuations run asynchronously, never on the reader.
|
||||
/// </summary>
|
||||
internal Task? LateSoftExpiry
|
||||
{
|
||||
get { lock (_clock) return _lateSoftExpiry?.Task; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// What the caller waits on: the worker's result, null when the worker ended first, <see cref="SoftTimedOut"/> at
|
||||
/// the soft deadline, or <see cref="TransportTimedOut"/> at the transport deadline (the run itself is still
|
||||
/// outstanding in both of the last two cases).
|
||||
/// </summary>
|
||||
public Task<EvalResult?> Outcome => _outcome.Task;
|
||||
|
||||
/// <summary>Starts the deadline timer. Called once, before the request is sent.</summary>
|
||||
internal void Arm(TimeSpan soft, TimeSpan grace, Action<PendingEval> onHardDeadline)
|
||||
/// <param name="soft">T.</param>
|
||||
/// <param name="grace">G.</param>
|
||||
/// <param name="onHardDeadline">The kill.</param>
|
||||
/// <param name="workerClock">The worker runs the soft clock (lmxopcua#827): this timer is the transport deadline.</param>
|
||||
internal void Arm(TimeSpan soft, TimeSpan grace, Action<PendingEval> onHardDeadline, bool workerClock = false)
|
||||
{
|
||||
_soft = soft;
|
||||
_grace = grace;
|
||||
_onHardDeadline = onHardDeadline;
|
||||
_workerClock = workerClock;
|
||||
HardDeadline = soft + grace;
|
||||
var now = Stopwatch.GetTimestamp();
|
||||
_softDue = now + Ticks(soft);
|
||||
_hardDue = _softDue + Ticks(grace);
|
||||
_transportDue = _hardDue;
|
||||
// Created disarmed and assigned before it can fire, so the soft tick can always re-arm it for the grace.
|
||||
_timer = new Timer(static s => ((PendingEval)s!).OnTimer(), this, Timeout.Infinite, Timeout.Infinite);
|
||||
_timer.Change(soft, Timeout.InfiniteTimeSpan);
|
||||
_timer.Change(workerClock ? soft + grace : soft, Timeout.InfiniteTimeSpan);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -1122,7 +1269,8 @@ internal sealed class PendingEval
|
||||
|
||||
/// <summary>
|
||||
/// The worker started this run: a deferred run's deadlines restart from now (T, or T + G when its soft deadline
|
||||
/// had already been reported). A run that was never deferred keeps the clock it has had since submit.
|
||||
/// had already been reported; with the worker soft clock, the transport deadline T + G). A run that was never
|
||||
/// deferred keeps the clock it has had since submit.
|
||||
/// </summary>
|
||||
internal void Start()
|
||||
{
|
||||
@@ -1132,7 +1280,16 @@ internal sealed class PendingEval
|
||||
if (_timer is null || !_deferred || Volatile.Read(ref _done) != 0) return;
|
||||
_deferred = false;
|
||||
var now = Stopwatch.GetTimestamp();
|
||||
if (Volatile.Read(ref _phase) == 0)
|
||||
if (_workerClock)
|
||||
{
|
||||
// Its soft deadline is the worker's from here; the transport deadline restarts. Already answered (a
|
||||
// deferral notice that arrived after the transport deadline): nothing to re-arm — a soft notice or the
|
||||
// result is what comes next.
|
||||
if (Volatile.Read(ref _phase) != Waiting) return;
|
||||
_transportDue = now + Ticks(_soft + _grace);
|
||||
TryChange(_soft + _grace);
|
||||
}
|
||||
else if (Volatile.Read(ref _phase) == Waiting)
|
||||
{
|
||||
_softDue = now + Ticks(_soft);
|
||||
_hardDue = _softDue + Ticks(_grace);
|
||||
@@ -1148,31 +1305,75 @@ internal sealed class PendingEval
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The worker reports that the script has itself run for T (worker soft clock, lmxopcua#827): the caller gets its
|
||||
/// soft timeout now — unless the transport deadline already answered it — and the hard deadline is G from here.
|
||||
/// </summary>
|
||||
internal void SoftExpired()
|
||||
{
|
||||
bool signal, late;
|
||||
lock (_clock)
|
||||
{
|
||||
if (_timer is null || !_workerClock || Volatile.Read(ref _done) != 0) return;
|
||||
var phase = Volatile.Read(ref _phase);
|
||||
if (phase is SoftReported or HardFired) return;
|
||||
_started = true;
|
||||
_deferred = false;
|
||||
signal = phase == Waiting;
|
||||
late = phase == TransportReported;
|
||||
Interlocked.Exchange(ref _phase, SoftReported);
|
||||
_hardDue = Stopwatch.GetTimestamp() + Ticks(_grace);
|
||||
TryChange(_grace);
|
||||
}
|
||||
// This runs on the pipe READER. The caller's resumption is the caller's code (the seam's publish, the runner's
|
||||
// Cancel write): it goes through the dispatcher exactly like a result (ResultDispatcher remarks), never inline
|
||||
// here, or a slow consumer would hold up heartbeats and results (#827 review F1).
|
||||
if (signal)
|
||||
{
|
||||
if (_dispatcher is { } dispatcher) dispatcher.Post(this, SoftTimedOut);
|
||||
else _outcome.TrySetResult(SoftTimedOut);
|
||||
}
|
||||
// The caller was already answered "worker unresponsive"; the script has now itself run for T. Its watcher
|
||||
// (the runner) cancels the run's token and counts the timeout, off this thread: the completion source runs its
|
||||
// continuations asynchronously (#827 review F2).
|
||||
if (late) _lateSoftExpiry?.TrySetResult();
|
||||
}
|
||||
|
||||
private void OnTimer()
|
||||
{
|
||||
if (Volatile.Read(ref _done) != 0) return;
|
||||
bool soft = false, hard = false;
|
||||
bool soft = false, hard = false, transport = false;
|
||||
lock (_clock)
|
||||
{
|
||||
if (_deferred || Volatile.Read(ref _done) != 0) return;
|
||||
var now = Stopwatch.GetTimestamp();
|
||||
if (Volatile.Read(ref _phase) == 0)
|
||||
var phase = Volatile.Read(ref _phase);
|
||||
if (phase == Waiting && _workerClock)
|
||||
{
|
||||
// A tick armed before a deferral/start can land early: re-arm for what is left.
|
||||
if (now < _transportDue) { TryChange(Stopwatch.GetElapsedTime(now, _transportDue)); return; }
|
||||
Interlocked.Exchange(ref _phase, TransportReported);
|
||||
_lateSoftExpiry = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
transport = true;
|
||||
}
|
||||
else if (phase == Waiting)
|
||||
{
|
||||
// A tick armed before a deferral/start can land early: re-arm for what is left.
|
||||
if (now < _softDue) { TryChange(Stopwatch.GetElapsedTime(now, _softDue)); return; }
|
||||
Interlocked.Exchange(ref _phase, 1);
|
||||
Interlocked.Exchange(ref _phase, SoftReported);
|
||||
soft = true;
|
||||
TryChange(_hardDue > now ? Stopwatch.GetElapsedTime(now, _hardDue) : TimeSpan.Zero);
|
||||
}
|
||||
else if (Volatile.Read(ref _phase) == 1)
|
||||
else if (phase == SoftReported)
|
||||
{
|
||||
if (now < _hardDue) { TryChange(Stopwatch.GetElapsedTime(now, _hardDue)); return; }
|
||||
Interlocked.Exchange(ref _phase, 2);
|
||||
Interlocked.Exchange(ref _phase, HardFired);
|
||||
hard = true;
|
||||
}
|
||||
}
|
||||
// Outside the lock: a caller with a dispatcher resumes synchronously inside TrySetResult.
|
||||
if (soft) _outcome.TrySetResult(SoftTimedOut);
|
||||
if (transport) _outcome.TrySetResult(TransportTimedOut);
|
||||
if (hard && Volatile.Read(ref _done) == 0) _onHardDeadline?.Invoke(this);
|
||||
}
|
||||
|
||||
@@ -1192,6 +1393,7 @@ internal sealed class PendingEval
|
||||
{
|
||||
if (Interlocked.Exchange(ref _done, 1) != 0) return;
|
||||
_timer?.Dispose();
|
||||
lock (_clock) _lateSoftExpiry?.TrySetCanceled();
|
||||
try { _onFinished?.Invoke(); }
|
||||
finally
|
||||
{
|
||||
|
||||
@@ -56,7 +56,7 @@ internal sealed class WorkerShard : IAsyncDisposable
|
||||
|
||||
private static readonly Counter<long> ExitsCounter = OtOpcUaTelemetry.Meter.CreateCounter<long>(
|
||||
"otopcua.script.worker.exits", unit: "{process}",
|
||||
description: "ScriptWorker processes ended (tags: shard, reason = clean | killed_deadline | killed_heartbeat | killed_load | crash | protocol | idle | disposed).");
|
||||
description: "ScriptWorker processes ended (tags: shard, reason = clean | killed_deadline | killed_heartbeat | killed_unresponsive | killed_load | crash | protocol | idle | disposed).");
|
||||
|
||||
private readonly Func<Action<WorkerExit>, CancellationToken, Task<WorkerConnection>> _launch;
|
||||
private readonly Action<WorkerShard, WorkerExit> _onExit;
|
||||
@@ -439,6 +439,7 @@ internal sealed class WorkerShard : IAsyncDisposable
|
||||
WorkerExitReason.Clean => "clean",
|
||||
WorkerExitReason.KilledDeadline => "killed_deadline",
|
||||
WorkerExitReason.KilledHeartbeat => "killed_heartbeat",
|
||||
WorkerExitReason.KilledUnresponsive => "killed_unresponsive",
|
||||
WorkerExitReason.KilledLoadTimeout => "killed_load",
|
||||
WorkerExitReason.ProtocolFault => "protocol",
|
||||
WorkerExitReason.Disposed => "disposed",
|
||||
|
||||
@@ -41,12 +41,17 @@ message Hello {
|
||||
string pe_cache_dir = 8; // the node's compiled-PE cache (design §7); empty = no cache
|
||||
bytes pe_cache_key = 9; // its HMAC key; the worker loads only entries that verify under it
|
||||
uint32 compile_threads = 10; // dedicated compile threads; 0 = the worker's default (min(8, max(1, CPUs/2)))
|
||||
// lmxopcua#827: the host asks the worker to run each evaluation's SOFT clock itself, from when an evaluation thread
|
||||
// starts the run, and to report the crossing in EvalResults.soft_expired_eval_ids. The host then keeps only a
|
||||
// transport deadline of its own (see EvalResults). Honoured only when the worker echoes Ready.worker_soft_clock.
|
||||
bool worker_soft_clock = 11;
|
||||
}
|
||||
|
||||
message Ready {
|
||||
string nonce = 1;
|
||||
int32 pid = 2;
|
||||
string runtime_version = 3;
|
||||
bool worker_soft_clock = 4; // lmxopcua#827: this worker runs the soft clock (Hello.worker_soft_clock was set)
|
||||
}
|
||||
|
||||
enum ContextKind {
|
||||
@@ -141,10 +146,23 @@ message EvalResult {
|
||||
// when the evaluation appears in `started_eval_ids`, which an evaluation thread sends alongside the result of the run it
|
||||
// just finished. So a run waiting behind a sibling is never timed out, struck or killed for the sibling's time.
|
||||
// Either list may be the only content of a frame. A host that predates these fields ignores them.
|
||||
//
|
||||
// Worker soft clock (lmxopcua#827, negotiated by Hello/Ready.worker_soft_clock): the soft deadline T starts when an
|
||||
// evaluation thread starts the run (not at submit) and is wall-clock time from there, as in process — a pause or
|
||||
// starvation DURING the run still counts. The worker reports the crossing in `soft_expired_eval_ids` (it does not
|
||||
// cancel the run itself: the host answers its caller first, then sends Cancel, keeping the in-process ordering). The
|
||||
// host arms the hard deadline G from that notice. What the host keeps on its own clock is a TRANSPORT deadline, T + G
|
||||
// from submit (restarted at a deferred run's start): no result and no soft notice by then means the pipe or the worker
|
||||
// process is late — the caller is answered as a runner fault ("script worker unresponsive", BadResourceUnavailable),
|
||||
// not as a script timeout, and nothing is struck (a soft notice that arrives later still cancels the run and counts as
|
||||
// a timeout). A worker that starts runs late therefore cannot make a trivial script look over budget; one that stops
|
||||
// heartbeating is killed for silence, and one that heartbeats without progress is killed at the host's progress
|
||||
// ceiling.
|
||||
message EvalResults {
|
||||
repeated EvalResult items = 1;
|
||||
repeated uint64 started_eval_ids = 2;
|
||||
repeated uint64 deferred_eval_ids = 3;
|
||||
repeated uint64 soft_expired_eval_ids = 4;
|
||||
}
|
||||
|
||||
message ScriptError {
|
||||
|
||||
@@ -237,6 +237,7 @@ public sealed class ScriptWorkerFrameWriter
|
||||
results.Items.AddRange(envelope.Results.Items);
|
||||
results.StartedEvalIds.AddRange(envelope.Results.StartedEvalIds);
|
||||
results.DeferredEvalIds.AddRange(envelope.Results.DeferredEvalIds);
|
||||
results.SoftExpiredEvalIds.AddRange(envelope.Results.SoftExpiredEvalIds);
|
||||
break;
|
||||
default:
|
||||
if (evals is not null) { Append(new Envelope { Eval = evals }); evals = null; }
|
||||
|
||||
@@ -125,6 +125,12 @@ public sealed record ScriptRunOutcome(ScriptRunStatus Status, object? Value, str
|
||||
/// <summary>Reason when the worker could not take more work (the out-of-process runner's back-pressure answer).</summary>
|
||||
public const string WorkerBusyReason = "script worker busy";
|
||||
|
||||
/// <summary>
|
||||
/// Reason when the worker neither answered nor reported the script over its budget within T + G of submit — the
|
||||
/// pipe or a starved / stalled worker process, not the script (lmxopcua#827).
|
||||
/// </summary>
|
||||
public const string WorkerUnresponsiveReason = "script worker unresponsive";
|
||||
|
||||
/// <summary>
|
||||
/// For <see cref="ScriptRunStatus.Unavailable"/>: completes when the runner can run this script again (its worker
|
||||
/// is back), so the seam re-evaluates it then with its latest inputs (design §9, "re-evaluated on the shard's
|
||||
|
||||
@@ -47,7 +47,7 @@ public enum ScriptShadowVerdict
|
||||
/// <summary>The shadow run disagreed; <see cref="ScriptShadowComparison.Reason"/> names how.</summary>
|
||||
Mismatch = 1,
|
||||
|
||||
/// <summary>The shadow runner could not run the script (worker restarted, none ready, worker busy): not compared.</summary>
|
||||
/// <summary>The shadow runner could not run the script (worker restarted, none ready, worker busy, worker unresponsive): not compared.</summary>
|
||||
ShadowUnavailable = 2,
|
||||
}
|
||||
|
||||
|
||||
@@ -491,6 +491,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// (<see cref="DriverInstanceActor.WriteAttribute.IssuedAt"/>) means the same thing on both sides.</summary>
|
||||
private readonly TimeProvider _writeClock;
|
||||
|
||||
/// <summary>
|
||||
/// Test seam: the PostStop shutdown budget every real (non-stub) child is spawned with; null =
|
||||
/// <see cref=DriverInstanceActor.DefaultShutdownTimeout/>. A test that parks a stopped child in its
|
||||
/// PostStop to observe the retired-but-not-terminated window raises it, so the window cannot close under
|
||||
/// load before the test has looked (lmxopcua#828).
|
||||
/// </summary>
|
||||
private readonly TimeSpan? _childShutdownTimeout;
|
||||
|
||||
/// <summary>Whether any redundancy snapshot has reached this host — the HistoryUpdate gate stays
|
||||
/// closed until one has (see <see cref="PrimaryGatePolicy.ShouldServeHistoryUpdate"/>).</summary>
|
||||
private bool _redundancySnapshotReceived;
|
||||
@@ -738,6 +746,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// Ask is derived (<see cref="DriverWriteBounds.HostAsk"/>). Null ⇒ <see cref="DriverWriteBounds.Default"/>.</param>
|
||||
/// <param name="writeClock">The clock the write bounds (and the host's write Ask) run on, shared with every
|
||||
/// child. Null ⇒ <see cref="TimeProvider.System"/>.</param>
|
||||
/// <param name="childShutdownTimeout">Test seam: each real child's PostStop shutdown budget; null ⇒
|
||||
/// <see cref="DriverInstanceActor.DefaultShutdownTimeout"/> (lmxopcua#828).</param>
|
||||
/// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
|
||||
public static Props Props(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
|
||||
@@ -776,7 +786,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
NativeAlarmAckPreflightView? ackPreflightView = null,
|
||||
DriverWriteBounds? writeBounds = null,
|
||||
TimeProvider? writeClock = null,
|
||||
IScriptRunner? scriptedAlarmRunner = null) =>
|
||||
IScriptRunner? scriptedAlarmRunner = null,
|
||||
TimeSpan? childShutdownTimeout = null) =>
|
||||
// WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an
|
||||
// expression tree. Six IActorRef? parameters and several interface-typed ones mean a
|
||||
// mis-ordered argument is usually type-compatible and therefore compiles clean, then binds
|
||||
@@ -790,7 +801,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView,
|
||||
sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay,
|
||||
sharesAlarmQueueWithLocal, applyOutcomeTimeout, captureWhilePrimaryUnreachable, historyUpdateRoleView,
|
||||
scripting, ackPreflightView, writeBounds, writeClock, scriptedAlarmRunner));
|
||||
scripting, ackPreflightView, writeBounds, writeClock, scriptedAlarmRunner, childShutdownTimeout));
|
||||
|
||||
/// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
|
||||
/// <param name="dbFactory">Database context factory for configuration database access.</param>
|
||||
@@ -885,6 +896,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
/// Ask is derived (<see cref="DriverWriteBounds.HostAsk"/>). Null ⇒ <see cref="DriverWriteBounds.Default"/>.</param>
|
||||
/// <param name="writeClock">The clock the write bounds (and the host's write Ask) run on, shared with every
|
||||
/// child. Null ⇒ <see cref="TimeProvider.System"/>.</param>
|
||||
/// <param name="childShutdownTimeout">Test seam: each real child's PostStop shutdown budget; null ⇒
|
||||
/// <see cref="DriverInstanceActor.DefaultShutdownTimeout"/> (lmxopcua#828).</param>
|
||||
public DriverHostActor(
|
||||
IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
|
||||
CommonsNodeId localNode,
|
||||
@@ -922,8 +935,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
NativeAlarmAckPreflightView? ackPreflightView = null,
|
||||
DriverWriteBounds? writeBounds = null,
|
||||
TimeProvider? writeClock = null,
|
||||
IScriptRunner? scriptedAlarmRunner = null)
|
||||
IScriptRunner? scriptedAlarmRunner = null,
|
||||
TimeSpan? childShutdownTimeout = null)
|
||||
{
|
||||
_childShutdownTimeout = childShutdownTimeout;
|
||||
_scriptedAlarmRunner = scriptedAlarmRunner;
|
||||
_writeBounds = writeBounds ?? DriverWriteBounds.Default;
|
||||
_writeClock = writeClock ?? TimeProvider.System;
|
||||
@@ -3957,7 +3972,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
clusterId: clusterId,
|
||||
invoker: invoker,
|
||||
writeTimeout: _writeBounds.Timeout,
|
||||
writeClock: _writeClock),
|
||||
writeClock: _writeClock,
|
||||
shutdownTimeout: _childShutdownTimeout),
|
||||
actorName);
|
||||
child.Tell(new DriverInstanceActor.InitializeRequested(spec.DriverConfig));
|
||||
}
|
||||
|
||||
@@ -29,6 +29,9 @@ internal sealed class WorkerHost
|
||||
/// <summary>The watchdog fails fast this long after the host's own hard deadline, so the host normally wins.</summary>
|
||||
private static readonly TimeSpan WatchdogMargin = TimeSpan.FromMilliseconds(500);
|
||||
|
||||
/// <summary>The watchdog's longest sleep between passes.</summary>
|
||||
private static readonly TimeSpan WatchdogPoll = TimeSpan.FromMilliseconds(50);
|
||||
|
||||
private readonly NamedPipeClientStream _pipe;
|
||||
private readonly string _nonce;
|
||||
private readonly int _parentPid;
|
||||
@@ -75,6 +78,19 @@ internal sealed class WorkerHost
|
||||
private TimeSpan _hardGrace = TimeSpan.FromSeconds(1);
|
||||
private TimeSpan _heartbeatInterval = TimeSpan.FromSeconds(1);
|
||||
private ScriptPeCacheStore? _peCache;
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#827: this worker runs each evaluation's soft clock from the run's start and reports the crossing
|
||||
/// (<c>EvalResults.soft_expired_eval_ids</c>); the host asked for it in <c>Hello</c> and we echoed it in
|
||||
/// <c>Ready</c>. Off (an older host): the host's own clock, started at submit, is the soft deadline.
|
||||
/// </summary>
|
||||
private bool _workerSoftClock;
|
||||
|
||||
/// <summary>Wakes the watchdog early when a run's soft deadline is sooner than its next planned pass.</summary>
|
||||
private readonly AutoResetEvent _watchdogWake = new(false);
|
||||
|
||||
/// <summary><see cref="Stopwatch.GetTimestamp"/> of the watchdog's next planned pass.</summary>
|
||||
private long _watchdogNextPass;
|
||||
private long _seq;
|
||||
|
||||
public WorkerHost(NamedPipeClientStream pipe, string nonce, int parentPid)
|
||||
@@ -109,6 +125,7 @@ internal sealed class WorkerHost
|
||||
Nonce = _nonce,
|
||||
Pid = Environment.ProcessId,
|
||||
RuntimeVersion = Environment.Version.ToString(),
|
||||
WorkerSoftClock = _workerSoftClock,
|
||||
},
|
||||
});
|
||||
if (ScriptWorkerFrameCodec.UseBlockingIo) ScriptWorkerFrameCodec.Write(_pipe, ready, _maxFrameBytes);
|
||||
@@ -140,6 +157,7 @@ internal sealed class WorkerHost
|
||||
if (hello.HardGraceMs > 0) _hardGrace = TimeSpan.FromMilliseconds(hello.HardGraceMs);
|
||||
if (hello.HeartbeatIntervalMs > 0) _heartbeatInterval = TimeSpan.FromMilliseconds(hello.HeartbeatIntervalMs);
|
||||
if (hello.CompileThreads > 0) _compileThreads = (int)Math.Min(hello.CompileThreads, 64);
|
||||
_workerSoftClock = hello.WorkerSoftClock;
|
||||
if (!string.IsNullOrEmpty(hello.PeCacheDir) && hello.PeCacheKey.Length == ScriptPeCacheStore.KeyLength)
|
||||
{
|
||||
// Design §7: the host owns the key; it reaches us only here, never through the environment or a file.
|
||||
@@ -425,11 +443,17 @@ internal sealed class WorkerHost
|
||||
var logger = _scriptLogger.ForContext(ForwardingLogSink.KeyProperty, request.Key);
|
||||
var soft = TimeSpan.FromMilliseconds(Math.Max(1, request.SoftTimeoutMs));
|
||||
|
||||
// Cancelled by the host's Cancel frame at the soft deadline, never by our own clock (see the proto).
|
||||
// Cancelled by the host's Cancel frame at the soft deadline, never by our own clock (see the proto): with the
|
||||
// worker soft clock (lmxopcua#827) the watchdog only REPORTS the crossing; the host answers its caller, then
|
||||
// cancels.
|
||||
using var cts = new CancellationTokenSource();
|
||||
var grace = request.HardGraceMs > 0 ? TimeSpan.FromMilliseconds(request.HardGraceMs) : _hardGrace;
|
||||
var running = new RunningEval(request.Key, Stopwatch.GetTimestamp(), soft + grace + WatchdogMargin, cts);
|
||||
var started = Stopwatch.GetTimestamp();
|
||||
var running = new RunningEval(request.EvalId, request.Key, started, soft, grace, cts);
|
||||
_running[(long)request.EvalId] = running;
|
||||
// The watchdog plans its next pass at most WatchdogPoll ahead; a budget shorter than that wakes it now, so the
|
||||
// soft notice is not late by up to a poll interval.
|
||||
if (_workerSoftClock && running.SoftDue < Interlocked.Read(ref _watchdogNextPass)) _watchdogWake.Set();
|
||||
var before = GC.GetAllocatedBytesForCurrentThread();
|
||||
var sw = Stopwatch.StartNew();
|
||||
try
|
||||
@@ -465,22 +489,68 @@ internal sealed class WorkerHost
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The watchdog: reports soft-deadline crossings (worker soft clock, lmxopcua#827) and fails the process fast for
|
||||
/// a run past its hard deadline that the host has not killed.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The hard deadline it enforces is the host's plus <see cref="WatchdogMargin"/>, from the same anchor the host
|
||||
/// uses: the run's start (T + G) on the host clock, or — with the worker soft clock — the soft notice it sent
|
||||
/// (G), which is where the host arms its kill. So the host normally wins and the exit names the culprit.
|
||||
/// </remarks>
|
||||
private void WatchdogLoop()
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
Thread.Sleep(50);
|
||||
var now = Stopwatch.GetTimestamp();
|
||||
var nextPass = now + Ticks(WatchdogPoll);
|
||||
List<ulong>? expired = null;
|
||||
foreach (var (_, run) in _running)
|
||||
{
|
||||
if (Stopwatch.GetElapsedTime(run.StartedTimestamp) > run.FailFastAfter)
|
||||
if (_workerSoftClock)
|
||||
{
|
||||
// The host should have killed us already; this names the culprit in the exit/dump.
|
||||
Environment.FailFast($"ScriptWorker: script '{run.Key}' exceeded its hard deadline ({run.FailFastAfter.TotalMilliseconds:F0} ms)");
|
||||
if (run.SoftReportedAt == 0)
|
||||
{
|
||||
if (now >= run.SoftDue)
|
||||
{
|
||||
run.SoftReportedAt = now;
|
||||
(expired ??= []).Add(run.EvalId);
|
||||
}
|
||||
else if (run.SoftDue < nextPass)
|
||||
{
|
||||
nextPass = run.SoftDue;
|
||||
}
|
||||
}
|
||||
else if (now - run.SoftReportedAt > Ticks(run.Grace + WatchdogMargin))
|
||||
{
|
||||
FailFast(run, run.Grace + WatchdogMargin, "its soft deadline");
|
||||
}
|
||||
}
|
||||
else if (now - run.StartedTimestamp > Ticks(run.Soft + run.Grace + WatchdogMargin))
|
||||
{
|
||||
FailFast(run, run.Soft + run.Grace + WatchdogMargin, "its start");
|
||||
}
|
||||
}
|
||||
if (expired is not null)
|
||||
{
|
||||
var notice = new EvalResults();
|
||||
notice.SoftExpiredEvalIds.AddRange(expired);
|
||||
Send(new Envelope { Results = notice });
|
||||
}
|
||||
Interlocked.Exchange(ref _watchdogNextPass, nextPass);
|
||||
var wait = Stopwatch.GetElapsedTime(Stopwatch.GetTimestamp(), nextPass);
|
||||
if (wait > TimeSpan.Zero) _watchdogWake.WaitOne(wait);
|
||||
}
|
||||
|
||||
static void FailFast(RunningEval run, TimeSpan after, string anchor)
|
||||
{
|
||||
// The host should have killed us already; this names the culprit in the exit/dump.
|
||||
Environment.FailFast($"ScriptWorker: script '{run.Key}' exceeded its hard deadline ({after.TotalMilliseconds:F0} ms after {anchor})");
|
||||
}
|
||||
}
|
||||
|
||||
private static long Ticks(TimeSpan span) => (long)(span.TotalSeconds * Stopwatch.Frequency);
|
||||
|
||||
private async Task HeartbeatLoopAsync()
|
||||
{
|
||||
using var timer = new PeriodicTimer(_heartbeatInterval);
|
||||
@@ -609,7 +679,30 @@ internal sealed class WorkerHost
|
||||
Environment.Exit(ScriptWorkerEnvironment.ExitForbiddenAssembly);
|
||||
}
|
||||
|
||||
private sealed record RunningEval(string Key, long StartedTimestamp, TimeSpan FailFastAfter, CancellationTokenSource Cancellation);
|
||||
/// <summary>One evaluation on an evaluation thread, as the watchdog sees it.</summary>
|
||||
private sealed class RunningEval(ulong evalId, string key, long startedTimestamp, TimeSpan soft, TimeSpan grace,
|
||||
CancellationTokenSource cancellation)
|
||||
{
|
||||
public ulong EvalId { get; } = evalId;
|
||||
|
||||
public string Key { get; } = key;
|
||||
|
||||
/// <summary><see cref="Stopwatch.GetTimestamp"/> when the evaluation thread started the run.</summary>
|
||||
public long StartedTimestamp { get; } = startedTimestamp;
|
||||
|
||||
/// <summary>T.</summary>
|
||||
public TimeSpan Soft { get; } = soft;
|
||||
|
||||
/// <summary>The run's start + T.</summary>
|
||||
public long SoftDue { get; } = startedTimestamp + Ticks(soft);
|
||||
|
||||
public TimeSpan Grace { get; } = grace;
|
||||
|
||||
public CancellationTokenSource Cancellation { get; } = cancellation;
|
||||
|
||||
/// <summary>When the watchdog reported the soft crossing to the host; 0 = not yet. Watchdog thread only.</summary>
|
||||
public long SoftReportedAt { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>One loaded script: its compiled evaluator and its input slot order.</summary>
|
||||
private sealed class LoadedScript : IDisposable
|
||||
|
||||
@@ -83,6 +83,8 @@ public sealed class ShadowScriptRunnerTests
|
||||
// The runner's fault, not the script's: not compared.
|
||||
{ ScriptRunOutcome.Ok(42), ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason), ScriptShadowVerdict.ShadowUnavailable, null },
|
||||
{ ScriptRunOutcome.Ok(42), ScriptRunOutcome.Failed(ScriptRunOutcome.WorkerBusyReason), ScriptShadowVerdict.ShadowUnavailable, null },
|
||||
// lmxopcua#827: a stalled / starved worker past the transport deadline is the runner's fault, not a mismatch.
|
||||
{ ScriptRunOutcome.Ok(42), ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerUnresponsiveReason), ScriptShadowVerdict.ShadowUnavailable, null },
|
||||
};
|
||||
|
||||
[Theory]
|
||||
|
||||
+39
-14
@@ -47,7 +47,8 @@ public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase
|
||||
|
||||
/// <summary>Spawns a host over a real mux with two enabled drivers, waits for both children to subscribe,
|
||||
/// registers a replay-opting holder (so the mux retains both refs) and publishes one value per driver.</summary>
|
||||
private void ArrangeTwoDriversWithRetainedValues()
|
||||
/// <param name="childShutdownTimeout">Each child's PostStop shutdown budget (null = the production 5 s).</param>
|
||||
private void ArrangeTwoDriversWithRetainedValues(TimeSpan? childShutdownTimeout = null)
|
||||
{
|
||||
_db = NewInMemoryDbFactory();
|
||||
var revA = V3ArtifactSeeder.SeedV3Deployment(_db, RevA, BothDrivers, NoRefs, enabled: true);
|
||||
@@ -61,7 +62,8 @@ public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase
|
||||
localRoles: new HashSet<string> { "driver" },
|
||||
dependencyMux: _mux,
|
||||
virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
|
||||
virtualTagHostOverride: CreateTestProbe().Ref));
|
||||
virtualTagHostOverride: CreateTestProbe().Ref,
|
||||
childShutdownTimeout: childShutdownTimeout));
|
||||
V3ArtifactSeeder.Dispatch(_host, _coordinator, revA, RevA);
|
||||
AwaitAssert(() => _factory.Created.Count.ShouldBe(2), PresenceBudget);
|
||||
// The spawn hands a child the PREVIOUS apply's set (empty on the first deploy) and the bulk push then
|
||||
@@ -137,28 +139,51 @@ public sealed class DriverHostActorSourceRemovedTests : RuntimeActorTestBase
|
||||
|
||||
/// <summary>Review I-1: a burst the removed child sent after the eviction (it stops asynchronously, so it keeps
|
||||
/// publishing through the rest of the apply) must be dropped by the host, not forwarded to re-retain the value.</summary>
|
||||
/// <remarks>
|
||||
/// lmxopcua#828: the window this test needs — the child RETIRED by the host but not yet TERMINATED — is held
|
||||
/// open by parking the child in its PostStop on <see cref="RecordingSubscribableDriver.ShutdownGate"/>. PostStop
|
||||
/// abandons a shutdown after its budget (5 s in production), the child then terminates, the host forgets it on
|
||||
/// <c>Terminated</c>, and a burst sent AS it afterwards is forwarded — so under a loaded suite run (more than
|
||||
/// 5 s from the stop to the host processing the burst) the test failed for a reason that is not a production
|
||||
/// path (a terminated actor cannot send). Reproduced deterministically by sleeping 6 s before the burst. The
|
||||
/// budget is therefore raised for this test, and the premise is a checked positive control: the child must
|
||||
/// still be alive when the fence has passed, or the absence assertion below would mean nothing.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public void A_late_burst_from_a_removed_child_does_not_re_retain_its_value()
|
||||
{
|
||||
ArrangeTwoDriversWithRetainedValues();
|
||||
ArrangeTwoDriversWithRetainedValues(childShutdownTimeout: TimeSpan.FromMinutes(5));
|
||||
var oldChild = ChildOf("DI-1");
|
||||
var watcher = CreateTestProbe();
|
||||
watcher.Watch(oldChild);
|
||||
// Hold the removed child in its PostStop so it has NOT terminated when the stale burst lands; a terminated
|
||||
// actor cannot send, and the host forgets a retired child on Terminated.
|
||||
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
Driver("DI-1").ShutdownGate = gate;
|
||||
try
|
||||
{
|
||||
var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true);
|
||||
V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB);
|
||||
// The child really was stopped (it is parked in PostStop now), so the burst below is a RETIRED child's.
|
||||
AwaitAssert(() => Driver("DI-1").ShutdownCount.ShouldBe(1), PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
|
||||
|
||||
var revB = V3ArtifactSeeder.SeedV3Deployment(_db, RevB, [("DI-2", "Modbus2", "dev2", "flow", "Double", false)], NoRefs, enabled: true);
|
||||
V3ArtifactSeeder.Dispatch(_host, _coordinator, revB, RevB);
|
||||
// The stale burst, sent AS the stopped child.
|
||||
_host.Tell(ValuePublishes.Single("DI-1", Path1, 11.0, OpcUaStatus.Good, DateTime.UtcNow), oldChild);
|
||||
// Positive control / ordering fence: a live driver's value sent after it is delivered, so the stale burst
|
||||
// has been processed by the host by the time we look.
|
||||
Driver("DI-2").FireDataChange(Path2, 23.0);
|
||||
AwaitHolder(Path2, 23.0);
|
||||
// Premise: the burst came from a retired child that had NOT terminated (it is still parked on the gate).
|
||||
watcher.HasMessages.ShouldBeFalse("the removed child terminated before the fence; the test premise is void");
|
||||
|
||||
// The stale burst, sent AS the stopped child.
|
||||
_host.Tell(ValuePublishes.Single("DI-1", Path1, 11.0, OpcUaStatus.Good, DateTime.UtcNow), oldChild);
|
||||
// Positive control / ordering fence: a live driver's value sent after it is delivered, so the stale burst
|
||||
// has been processed by the host by the time we look.
|
||||
Driver("DI-2").FireDataChange(Path2, 23.0);
|
||||
AwaitHolder(Path2, 23.0);
|
||||
|
||||
Replayed().Keys.ToArray().ShouldBe(new[] { Path2 }, "the removed child's late burst re-retained its last Good");
|
||||
gate.TrySetResult();
|
||||
Replayed().Keys.ToArray().ShouldBe(new[] { Path2 }, "the removed child's late burst re-retained its last Good");
|
||||
}
|
||||
finally
|
||||
{
|
||||
gate.TrySetResult();
|
||||
}
|
||||
// And the gate really was what held it: released, the child terminates.
|
||||
watcher.ExpectTerminated(oldChild, PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>Review m-3: a config edit respawns the driver (stop + spawn); the OLD object's value is evicted, and
|
||||
|
||||
@@ -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 =>
|
||||
|
||||
@@ -0,0 +1,492 @@
|
||||
using System.Diagnostics;
|
||||
using System.Runtime.InteropServices;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
using Serilog;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// lmxopcua#827: the soft deadline T starts when an evaluation thread starts the run — the worker runs that clock —
|
||||
/// so the pipe round trip and the time a starved worker takes to START the run no longer count against the script.
|
||||
/// From the start it is wall-clock time, as in process: a pause DURING the run still counts, and is not tested
|
||||
/// here (a mid-run freeze races the script's own return against the watchdog's notice). The host keeps a transport deadline (T + G from submit) and reports passing
|
||||
/// it as a runner fault (<see cref="ScriptRunOutcome.WorkerUnresponsiveReason"/>, BadResourceUnavailable), never as a
|
||||
/// script timeout. The starved worker is made real here: its process is suspended (SIGSTOP / NtSuspendProcess)
|
||||
/// while a trivial script is submitted.
|
||||
/// </summary>
|
||||
[Collection(ScriptWorkerCollection.Name)]
|
||||
[Trait("Category", "ScriptWorker")]
|
||||
public sealed class WorkerSoftClockTests
|
||||
{
|
||||
private const string Trivial = "return 7;";
|
||||
private const string Runaway = "while (true) { }";
|
||||
|
||||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||||
|
||||
/// <summary>
|
||||
/// A runner whose worker heartbeats every 2 s, so the host's silence kill (3 intervals) stays well clear of the
|
||||
/// suspensions below: what is measured is the deadline, not the heartbeat monitor.
|
||||
/// </summary>
|
||||
private static OutOfProcessScriptRunner Runner(TimeSpan hardGrace, ILogger? log = null)
|
||||
=> new(Scripts.Options(hardGrace: hardGrace, quarantineAfterKills: 1), log ?? new LoggerConfiguration().CreateLogger(), null,
|
||||
heartbeatInterval: TimeSpan.FromSeconds(2), loadTimeout: TimeSpan.FromSeconds(60));
|
||||
|
||||
private static async Task<WorkerConnection> WarmAsync(OutOfProcessScriptRunner runner, string key)
|
||||
{
|
||||
var warm = await runner.RunAsync(Scripts.Vt(key, Trivial, TimeSpan.FromSeconds(30)), Ct);
|
||||
warm.Status.ShouldBe(ScriptRunStatus.Ok, warm.Reason);
|
||||
var worker = runner.Shards[0].Current.ShouldNotBeNull();
|
||||
worker.WorkerSoftClock.ShouldBeTrue("the worker negotiated the soft clock");
|
||||
return worker;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The issue's headline: a worker that cannot be scheduled for longer than T, but shorter than the transport
|
||||
/// deadline, answers a trivial script with its value. Before #827 the host's clock, started at submit, answered
|
||||
/// <c>script timed out after 0.3s</c> for <c>return 7;</c>.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_worker_stalled_for_longer_than_T_still_answers_a_trivial_script_with_its_value()
|
||||
{
|
||||
var softT = TimeSpan.FromMilliseconds(300);
|
||||
await using var runner = Runner(hardGrace: TimeSpan.FromSeconds(3));
|
||||
var worker = await WarmAsync(runner, "trivial");
|
||||
|
||||
Task<ScriptRunOutcome> run;
|
||||
using (ProcessFreezer.Freeze(worker.Pid))
|
||||
{
|
||||
run = runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct);
|
||||
await Task.Delay(softT * 4, Ct); // 1.2 s: well past T on the host's clock
|
||||
run.IsCompleted.ShouldBeFalse("nothing can answer while the worker is suspended");
|
||||
}
|
||||
|
||||
var outcome = await run.WaitAsync(TimeSpan.FromSeconds(30), Ct);
|
||||
outcome.Status.ShouldBe(ScriptRunStatus.Ok, outcome.Reason);
|
||||
outcome.Value.ShouldBe(7);
|
||||
runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, "a stall is not a deadline kill");
|
||||
runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBeEmpty();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A stall past the transport deadline (T + G) is the runner's fault: the caller gets
|
||||
/// <see cref="ScriptRunOutcome.WorkerUnresponsiveReason"/> as <see cref="ScriptRunStatus.Unavailable"/> (published
|
||||
/// BadResourceUnavailable) with a recovery that completes when the worker answers; a re-trigger meanwhile is the
|
||||
/// runner's fault too (not "previous run still running"); the worker is not killed and the key is not struck.
|
||||
/// Before #827 this was a script timeout at T and a deadline kill at T + G that struck the key — with
|
||||
/// QuarantineAfterKills = 1, a quarantine for <c>return 7;</c>.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_worker_stalled_past_the_transport_deadline_is_a_runner_fault_and_nothing_is_struck()
|
||||
{
|
||||
var softT = TimeSpan.FromMilliseconds(200);
|
||||
var grace = TimeSpan.FromMilliseconds(400);
|
||||
await using var runner = Runner(grace);
|
||||
var worker = await WarmAsync(runner, "trivial");
|
||||
// Review F5: both answers are counted (otopcua.script.ipc.timeouts, answer = stalled | retrigger).
|
||||
var answers = new System.Collections.Concurrent.ConcurrentQueue<string>();
|
||||
using var listener = new System.Diagnostics.Metrics.MeterListener();
|
||||
listener.InstrumentPublished = (instrument, l) =>
|
||||
{
|
||||
if (instrument.Name == "otopcua.script.ipc.timeouts") l.EnableMeasurementEvents(instrument);
|
||||
};
|
||||
listener.SetMeasurementEventCallback<long>((_, _, tags, _) =>
|
||||
{
|
||||
foreach (var tag in tags)
|
||||
if (tag.Key == "answer") answers.Enqueue((string)tag.Value!);
|
||||
});
|
||||
listener.Start();
|
||||
|
||||
ScriptRunOutcome stalled, retriggered;
|
||||
using (ProcessFreezer.Freeze(worker.Pid))
|
||||
{
|
||||
stalled = await runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct).WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||||
retriggered = await runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct).WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||||
stalled.Recovery.ShouldNotBeNull().IsCompleted.ShouldBeFalse("the worker has not answered yet");
|
||||
}
|
||||
|
||||
stalled.Status.ShouldBe(ScriptRunStatus.Unavailable, stalled.Reason);
|
||||
stalled.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason);
|
||||
retriggered.Status.ShouldBe(ScriptRunStatus.Unavailable, retriggered.Reason);
|
||||
retriggered.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason);
|
||||
stalled.ToVirtualTagEvalResult().StatusCode.ShouldBe(Commons.OpcUa.OpcUaStatus.BadResourceUnavailable);
|
||||
answers.ToArray().ShouldBe(["stalled", "retrigger"], ignoreOrder: true);
|
||||
|
||||
await stalled.Recovery!.WaitAsync(TimeSpan.FromSeconds(30), Ct);
|
||||
var next = await runner.RunAsync(Scripts.Vt("trivial", Trivial, softT), Ct);
|
||||
next.Status.ShouldBe(ScriptRunStatus.Ok, next.Reason);
|
||||
next.Value.ShouldBe(7);
|
||||
runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, "a stall is not a deadline kill");
|
||||
runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBeEmpty();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The other side of the contract is unchanged: a script that really runs past T is answered at T from its own
|
||||
/// start and killed G after that, even with the worker stalled before it started. Positive control for the two
|
||||
/// tests above — they would also pass if the runner simply never timed anything out.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_runaway_that_starts_late_is_still_timed_out_at_T_from_its_start_and_killed_at_G()
|
||||
{
|
||||
var softT = TimeSpan.FromMilliseconds(300);
|
||||
var grace = TimeSpan.FromMilliseconds(600);
|
||||
var stall = TimeSpan.FromMilliseconds(500);
|
||||
await using var runner = Runner(grace);
|
||||
var worker = await WarmAsync(runner, "warm");
|
||||
var load = await runner.RunAsync(Scripts.Vt("runaway", "return 0;", TimeSpan.FromSeconds(30)), Ct);
|
||||
load.Status.ShouldBe(ScriptRunStatus.Ok, load.Reason);
|
||||
|
||||
var sw = Stopwatch.StartNew();
|
||||
Task<ScriptRunOutcome> run;
|
||||
using (ProcessFreezer.Freeze(worker.Pid))
|
||||
{
|
||||
run = runner.RunAsync(Scripts.Vt("runaway", Runaway, softT), Ct);
|
||||
await Task.Delay(stall, Ct);
|
||||
}
|
||||
var thawed = sw.Elapsed;
|
||||
var outcome = await run.WaitAsync(TimeSpan.FromSeconds(30), Ct);
|
||||
var answered = sw.Elapsed;
|
||||
|
||||
outcome.Status.ShouldBe(ScriptRunStatus.Failed);
|
||||
outcome.Reason.ShouldNotBeNull().ShouldStartWith("script timed out after ");
|
||||
(answered - thawed).ShouldBeGreaterThanOrEqualTo(softT - TimeSpan.FromMilliseconds(50), "T counts from the run's start, after the stall");
|
||||
|
||||
var exit = await worker.Exited.WaitAsync(TimeSpan.FromSeconds(30), Ct);
|
||||
exit.Reason.ShouldBe(WorkerExitReason.KilledDeadline);
|
||||
exit.CulpritKey.ShouldBe(InProcessScriptRunner.ScopedKey(ScriptContextKind.VirtualTag, "runaway"));
|
||||
(sw.Elapsed - thawed).ShouldBeGreaterThanOrEqualTo(softT + grace - TimeSpan.FromMilliseconds(50), "G counts from the soft deadline");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Shadow (lmxopcua#780 P2) compares the shadow pool with the published in-process result; a starved worker used to
|
||||
/// show up there as a script mismatch (published Ok, shadow "timed out") and skew the census for the OutOfProcess
|
||||
/// flip. Now a stall shorter than the transport deadline is a match, and a longer one is <c>shadow_unavailable</c>
|
||||
/// — never a mismatch — including the re-trigger that finds the stalled run still holding its slot.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task In_Shadow_a_stalled_worker_is_a_match_or_shadow_unavailable_never_a_mismatch()
|
||||
{
|
||||
var softT = TimeSpan.FromMilliseconds(300);
|
||||
var grace = TimeSpan.FromSeconds(1);
|
||||
var options = Scripts.Options(hardGrace: grace, quarantineAfterKills: 1);
|
||||
options.Runner = ScriptRunnerMode.Shadow;
|
||||
await using var pool = Runner(grace);
|
||||
using var shadow = new ShadowScriptRunner(pool, options, new LoggerConfiguration().CreateLogger());
|
||||
var worker = await WarmAsync(pool, "trivial");
|
||||
var published = ScriptRunOutcome.Ok(7);
|
||||
|
||||
// Stall past T, short of T + G: compared, and a match.
|
||||
using (ProcessFreezer.Freeze(worker.Pid))
|
||||
{
|
||||
shadow.Mirror(Scripts.Vt("trivial", Trivial, softT), published);
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(900), Ct);
|
||||
}
|
||||
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
|
||||
shadow.Stats().Matched.ShouldBe(1, Summary(shadow));
|
||||
|
||||
// Stall past T + G: not compared. The second mirror arrives while the first still holds the key's slot.
|
||||
using (ProcessFreezer.Freeze(worker.Pid))
|
||||
{
|
||||
shadow.Mirror(Scripts.Vt("trivial", Trivial, softT), published);
|
||||
await Scripts.Eventually(() => shadow.Stats().ShadowUnavailable == 1, TimeSpan.FromSeconds(10), "the transport deadline");
|
||||
shadow.Mirror(Scripts.Vt("trivial", Trivial, softT), published);
|
||||
await Scripts.Eventually(() => shadow.Stats().ShadowUnavailable == 2, TimeSpan.FromSeconds(10), "the stalled re-trigger");
|
||||
}
|
||||
(await shadow.WaitIdleAsync(TimeSpan.FromSeconds(30))).ShouldBeTrue();
|
||||
|
||||
var stats = shadow.Stats();
|
||||
stats.Mismatched.ShouldBe(0, Summary(shadow));
|
||||
stats.ShadowUnavailable.ShouldBe(2, Summary(shadow));
|
||||
stats.Matched.ShouldBe(1);
|
||||
pool.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Protocol compatibility: the soft clock is negotiated (<c>Hello.worker_soft_clock</c> → <c>Ready.worker_soft_clock</c>).
|
||||
/// A host that does not ask for it keeps the pre-#827 contract — the soft deadline is the host's, from submit,
|
||||
/// and the worker sends no soft notices — and a runaway is still answered at T and killed at T + G.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData(true)]
|
||||
[InlineData(false)]
|
||||
public async Task The_soft_clock_is_negotiated_and_both_clocks_enforce_T_and_G(bool workerSoftClock)
|
||||
{
|
||||
var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }),
|
||||
256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(1), HeartbeatInterval: TimeSpan.FromSeconds(1),
|
||||
HandshakeTimeout: TimeSpan.FromSeconds(15), WorkerSoftClock: workerSoftClock);
|
||||
var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct);
|
||||
await using var _ = connection;
|
||||
connection.WorkerSoftClock.ShouldBe(workerSoftClock);
|
||||
|
||||
var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Runaway,
|
||||
SHA256.HashData(Encoding.UTF8.GetBytes(Runaway)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct);
|
||||
entry.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue();
|
||||
|
||||
var softT = TimeSpan.FromMilliseconds(300);
|
||||
var grace = TimeSpan.FromMilliseconds(500);
|
||||
var sw = Stopwatch.StartNew();
|
||||
var pending = connection.Submit("k",
|
||||
new EvalRequest { Key = "k", SoftTimeoutMs = (uint)softT.TotalMilliseconds, HardGraceMs = (uint)grace.TotalMilliseconds, NowUtcTicks = DateTime.UtcNow.Ticks },
|
||||
softT, grace, null);
|
||||
(await pending.Outcome.WaitAsync(TimeSpan.FromSeconds(10), Ct)).ShouldBeSameAs(PendingEval.SoftTimedOut);
|
||||
sw.Elapsed.ShouldBeGreaterThanOrEqualTo(softT - TimeSpan.FromMilliseconds(20));
|
||||
connection.Cancel(pending);
|
||||
|
||||
var exit = await connection.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||||
exit.Reason.ShouldBe(WorkerExitReason.KilledDeadline);
|
||||
sw.Elapsed.ShouldBeGreaterThanOrEqualTo(softT + grace - TimeSpan.FromMilliseconds(50));
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------- review findings (#827)
|
||||
|
||||
/// <summary>
|
||||
/// Review F1: the soft notice is handled on the pipe READER, and the caller's resumption is the caller's code. It
|
||||
/// must go through the result dispatcher like a result does: a continuation that blocks for 1.5 s must not hold up
|
||||
/// the reader, or the next heartbeats sit unread and a healthy worker is killed for silence (300 ms here). Before
|
||||
/// the fix the continuation ran inline on the reader and the worker was killed (KilledHeartbeat).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_slow_continuation_of_a_soft_timeout_does_not_stall_the_reader()
|
||||
{
|
||||
var heartbeat = TimeSpan.FromMilliseconds(100);
|
||||
var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }),
|
||||
256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(60), HeartbeatInterval: heartbeat,
|
||||
HandshakeTimeout: TimeSpan.FromSeconds(15), HeartbeatSilenceLimit: TimeSpan.FromSeconds(30));
|
||||
var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct);
|
||||
await using var _ = connection;
|
||||
var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Runaway,
|
||||
SHA256.HashData(Encoding.UTF8.GetBytes(Runaway)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct);
|
||||
entry.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue();
|
||||
connection.SilenceLimit = heartbeat * 3; // armed only once the cold start is over (see SupervisionTests)
|
||||
|
||||
var softT = TimeSpan.FromMilliseconds(200);
|
||||
var pending = connection.Submit("k",
|
||||
new EvalRequest { Key = "k", SoftTimeoutMs = (uint)softT.TotalMilliseconds, HardGraceMs = 60_000, NowUtcTicks = DateTime.UtcNow.Ticks },
|
||||
softT, TimeSpan.FromSeconds(60), null);
|
||||
string? resumedOn = null;
|
||||
var slowConsumer = pending.Outcome.ContinueWith(t =>
|
||||
{
|
||||
resumedOn = Thread.CurrentThread.Name;
|
||||
t.Result.ShouldBeSameAs(PendingEval.SoftTimedOut);
|
||||
Thread.Sleep(1_500);
|
||||
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
|
||||
|
||||
await slowConsumer.WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||||
resumedOn.ShouldBe("ScriptWorker[t].Results", "the caller resumes on the dispatcher, never on the reader");
|
||||
connection.Exited.IsCompleted.ShouldBeFalse("the reader kept reading heartbeats while the consumer was slow");
|
||||
connection.IsAlive.ShouldBeTrue();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Review F2: a stall past T + G answers "worker unresponsive"; when the worker then runs the script and it really
|
||||
/// overruns T, the late soft notice must still cancel its token — a polling script ends itself instead of the whole
|
||||
/// worker being killed G later — and the run counts as a timeout (QuarantineAfterConsecutiveTimeouts = 1 here).
|
||||
/// Before the fix nobody sent the Cancel: the worker was killed at notice + G (KilledDeadline, a kill strike).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_slow_script_whose_notice_arrives_after_the_transport_deadline_is_cancelled_and_counted_not_killed()
|
||||
{
|
||||
var softT = TimeSpan.FromMilliseconds(200);
|
||||
var grace = TimeSpan.FromMilliseconds(400);
|
||||
var options = Scripts.Options(hardGrace: grace, quarantineAfterKills: 1);
|
||||
options.QuarantineAfterConsecutiveTimeouts = 1;
|
||||
await using var runner = new OutOfProcessScriptRunner(options, new LoggerConfiguration().CreateLogger(), null,
|
||||
heartbeatInterval: TimeSpan.FromSeconds(2), loadTimeout: TimeSpan.FromSeconds(60));
|
||||
const string polite = "if ((int)ctx.GetTag(\"spin\").Value == 1) { while (!ctx.CancellationToken.IsCancellationRequested) { } } return 1;";
|
||||
(await runner.RunAsync(Scripts.Vt("polite", polite, TimeSpan.FromSeconds(30), Scripts.Input("spin", 0)), Ct)).Value.ShouldBe(1);
|
||||
var worker = runner.Shards[0].Current.ShouldNotBeNull();
|
||||
|
||||
ScriptRunOutcome stalled;
|
||||
using (ProcessFreezer.Freeze(worker.Pid))
|
||||
{
|
||||
stalled = await runner.RunAsync(Scripts.Vt("polite", polite, softT, Scripts.Input("spin", 1)), Ct).WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||||
}
|
||||
stalled.Reason.ShouldBe(ScriptRunOutcome.WorkerUnresponsiveReason);
|
||||
|
||||
// Thawed: the script starts, spins past T, the late notice cancels it and it returns. The worker survives G.
|
||||
await stalled.Recovery.ShouldNotBeNull().WaitAsync(TimeSpan.FromSeconds(10), Ct);
|
||||
await Task.Delay(softT + grace + TimeSpan.FromMilliseconds(500), Ct);
|
||||
worker.IsAlive.ShouldBeTrue("a cancelled polling script must not cost its worker");
|
||||
runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid);
|
||||
|
||||
// …and it counted as the timeout it was (threshold 1), not as a kill.
|
||||
await Scripts.Eventually(() => runner.QuarantinedKeys(ScriptContextKind.VirtualTag).Count == 1, TimeSpan.FromSeconds(5), "the timeout quarantine");
|
||||
var refused = await runner.RunAsync(Scripts.Vt("polite", polite, softT, Scripts.Input("spin", 0)), Ct);
|
||||
refused.Reason.ShouldNotBeNull().ShouldContain("timed out");
|
||||
refused.Reason.ShouldNotContain("hard deadline");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Review F3: the host keeps its own bound. A worker that heartbeats but never sends a soft notice (a bug, or a
|
||||
/// compromised worker — simulated by the host dropping its notices) used to hold the run, the key's slot and the
|
||||
/// tag's BadResourceUnavailable forever: the transport deadline answers the caller once and the kill G is armed
|
||||
/// only by a notice. Now no progress for the ceiling, with evaluations outstanding, kills the worker
|
||||
/// (<see cref="WorkerExitReason.KilledUnresponsive"/>, no culprit) and the run completes. G is 60 s here, so the
|
||||
/// worker's own fail-fast watchdog (notice + G + 0.5 s) cannot be what ends it.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_worker_that_heartbeats_but_never_reports_progress_is_killed_at_the_host_ceiling()
|
||||
{
|
||||
var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }),
|
||||
256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(60), HeartbeatInterval: TimeSpan.FromMilliseconds(200),
|
||||
HandshakeTimeout: TimeSpan.FromSeconds(15), HeartbeatSilenceLimit: TimeSpan.FromSeconds(30));
|
||||
var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct);
|
||||
await using var _ = connection;
|
||||
var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, Runaway,
|
||||
SHA256.HashData(Encoding.UTF8.GetBytes(Runaway)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct);
|
||||
entry.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue();
|
||||
connection.DropSoftNoticesForTest = true;
|
||||
connection.ProgressCeilingOverride = TimeSpan.FromSeconds(1.5);
|
||||
|
||||
var softT = TimeSpan.FromMilliseconds(200);
|
||||
var sw = Stopwatch.StartNew();
|
||||
var pending = connection.Submit("k",
|
||||
new EvalRequest { Key = "k", SoftTimeoutMs = (uint)softT.TotalMilliseconds, HardGraceMs = 60_000, NowUtcTicks = DateTime.UtcNow.Ticks },
|
||||
softT, TimeSpan.FromSeconds(60), null);
|
||||
|
||||
var exit = await connection.Exited.WaitAsync(TimeSpan.FromSeconds(15), Ct);
|
||||
exit.Reason.ShouldBe(WorkerExitReason.KilledUnresponsive);
|
||||
exit.CulpritKey.ShouldBeNull("nothing says which script, so nothing is struck");
|
||||
exit.InFlightKeys.ShouldBe(["k"]);
|
||||
sw.Elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromSeconds(1.4), "not before the ceiling");
|
||||
(await pending.Completion.WaitAsync(TimeSpan.FromSeconds(5), Ct)).ShouldBeNull("the run is released");
|
||||
}
|
||||
|
||||
/// <summary>Positive control for the ceiling: the default rule leaves a busy, answering worker alone.</summary>
|
||||
[Fact]
|
||||
public async Task The_progress_ceiling_leaves_a_worker_that_keeps_answering_alone()
|
||||
{
|
||||
var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }),
|
||||
256L * 1024 * 1024, EvalThreads: 1, HardGrace: TimeSpan.FromSeconds(60), HeartbeatInterval: TimeSpan.FromMilliseconds(100),
|
||||
HandshakeTimeout: TimeSpan.FromSeconds(15), HeartbeatSilenceLimit: TimeSpan.FromSeconds(30));
|
||||
var connection = await WorkerConnection.StartAsync(spec, new LoggerConfiguration().CreateLogger(), _ => { }, _ => { }, Ct);
|
||||
await using var _ = connection;
|
||||
const string spin = "var end = ctx.Now.AddMilliseconds(150); while (ctx.Now < end) { } return 1;";
|
||||
var entry = await connection.EnsureLoadedAsync("k", ContextKind.VirtualTag, spin,
|
||||
SHA256.HashData(Encoding.UTF8.GetBytes(spin)), []).WaitAsync(TimeSpan.FromSeconds(60), Ct);
|
||||
entry.Outcome.ShouldNotBeNull().Ok.ShouldBeTrue();
|
||||
// Ceiling 1 s; 20 runs of 150 ms queued behind one evaluation thread take ~3 s, but each answers.
|
||||
connection.ProgressCeilingOverride = TimeSpan.FromSeconds(1);
|
||||
var soft = TimeSpan.FromSeconds(30);
|
||||
var pendings = connection.SubmitBatch(Enumerable.Range(0, 20).Select(_ => new BatchSubmission("k",
|
||||
new EvalRequest { Key = "k", SoftTimeoutMs = 30_000, HardGraceMs = 60_000, NowUtcTicks = DateTime.UtcNow.Ticks },
|
||||
soft, TimeSpan.FromSeconds(60), null)).ToArray());
|
||||
foreach (var p in pendings) (await p.Completion.WaitAsync(TimeSpan.FromSeconds(30), Ct)).ShouldNotBeNull();
|
||||
connection.IsAlive.ShouldBeTrue();
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------- PendingEval, no process
|
||||
|
||||
/// <summary>The host side of the worker clock, deterministic: no notice means the transport deadline at T + G.</summary>
|
||||
[Fact]
|
||||
public async Task Worker_clock_with_no_word_from_the_worker_answers_transport_timed_out_at_T_plus_G_and_never_kills()
|
||||
{
|
||||
var killed = 0;
|
||||
var pending = new PendingEval("k", 1);
|
||||
var sw = Stopwatch.StartNew();
|
||||
pending.Arm(TimeSpan.FromMilliseconds(100), TimeSpan.FromMilliseconds(200), _ => Interlocked.Increment(ref killed), workerClock: true);
|
||||
|
||||
(await pending.Outcome.WaitAsync(TimeSpan.FromSeconds(10), Ct)).ShouldBeSameAs(PendingEval.TransportTimedOut);
|
||||
sw.Elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(280));
|
||||
await Task.Delay(500, Ct);
|
||||
killed.ShouldBe(0, "a late worker is not the script's fault");
|
||||
|
||||
// A soft notice after the transport answer still arms the kill: the script itself ran for T.
|
||||
pending.SoftExpired();
|
||||
await Scripts.Eventually(() => Volatile.Read(ref killed) == 1, TimeSpan.FromSeconds(10), "the kill G after the soft notice");
|
||||
}
|
||||
|
||||
/// <summary>The soft notice answers the caller and arms the kill G later; a result before then cancels it.</summary>
|
||||
[Fact]
|
||||
public async Task Worker_clock_soft_notice_answers_soft_timed_out_and_arms_the_kill_G_later()
|
||||
{
|
||||
var killed = 0;
|
||||
var pending = new PendingEval("k", 1);
|
||||
pending.Arm(TimeSpan.FromSeconds(30), TimeSpan.FromMilliseconds(300), _ => Interlocked.Increment(ref killed), workerClock: true);
|
||||
pending.Outcome.IsCompleted.ShouldBeFalse();
|
||||
|
||||
var sw = Stopwatch.StartNew();
|
||||
pending.SoftExpired();
|
||||
(await pending.Outcome.WaitAsync(TimeSpan.FromSeconds(1), Ct)).ShouldBeSameAs(PendingEval.SoftTimedOut);
|
||||
await Scripts.Eventually(() => Volatile.Read(ref killed) == 1, TimeSpan.FromSeconds(10), "the kill");
|
||||
sw.Elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(280));
|
||||
|
||||
var answered = new PendingEval("k2", 2);
|
||||
answered.Arm(TimeSpan.FromSeconds(30), TimeSpan.FromMilliseconds(100), _ => Interlocked.Increment(ref killed), workerClock: true);
|
||||
answered.SoftExpired();
|
||||
answered.Complete(new EvalResult { EvalId = 2 });
|
||||
await Task.Delay(400, Ct);
|
||||
killed.ShouldBe(1, "a run that answered inside G is not killed");
|
||||
}
|
||||
|
||||
/// <summary>Under the host clock a soft notice means nothing (a worker that did not negotiate it cannot send one).</summary>
|
||||
[Fact]
|
||||
public void Host_clock_ignores_a_soft_notice()
|
||||
{
|
||||
var pending = new PendingEval("k", 1);
|
||||
pending.Arm(TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(30), _ => { }, workerClock: false);
|
||||
pending.SoftExpired();
|
||||
pending.Outcome.IsCompleted.ShouldBeFalse();
|
||||
pending.Complete(null);
|
||||
}
|
||||
|
||||
private static string Summary(ShadowScriptRunner shadow)
|
||||
=> $"{shadow.Stats()}; " + string.Join("; ", shadow.Mismatches.Select(m => $"{m.Key}: {m.LastReason} — {m.LastSummary}"));
|
||||
}
|
||||
|
||||
/// <summary>Suspends a process (SIGSTOP, or NtSuspendProcess on Windows) until disposed.</summary>
|
||||
internal sealed class ProcessFreezer : IDisposable
|
||||
{
|
||||
private readonly int _pid;
|
||||
private int _thawed;
|
||||
|
||||
private ProcessFreezer(int pid)
|
||||
{
|
||||
_pid = pid;
|
||||
Signal(pid, stop: true);
|
||||
}
|
||||
|
||||
public static ProcessFreezer Freeze(int pid) => new(pid);
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (Interlocked.Exchange(ref _thawed, 1) == 0) Signal(_pid, stop: false);
|
||||
}
|
||||
|
||||
private static void Signal(int pid, bool stop)
|
||||
{
|
||||
if (OperatingSystem.IsWindows())
|
||||
{
|
||||
using var process = Process.GetProcessById(pid);
|
||||
var status = stop ? NtSuspendProcess(process.Handle) : NtResumeProcess(process.Handle);
|
||||
status.ShouldBe(0, $"Nt{(stop ? "Suspend" : "Resume")}Process");
|
||||
return;
|
||||
}
|
||||
// libc kill(2), not the kill(1) command: a child process's exit is observed through the same SIGCHLD
|
||||
// handling that the suspended worker (another child of this process) just disturbed, and WaitForExit was
|
||||
// seen to take seconds.
|
||||
var signal = stop ? SigStop : SigCont;
|
||||
Kill(pid, signal).ShouldBe(0, $"kill({pid}, {signal}): errno {Marshal.GetLastPInvokeError()}");
|
||||
}
|
||||
|
||||
/// <summary>SIGSTOP / SIGCONT: 17 / 19 on macOS (BSD), 19 / 18 on Linux.</summary>
|
||||
private static int SigStop => OperatingSystem.IsMacOS() ? 17 : 19;
|
||||
|
||||
private static int SigCont => OperatingSystem.IsMacOS() ? 19 : 18;
|
||||
|
||||
#pragma warning disable SYSLIB1054 // DllImport kept: LibraryImport would need AllowUnsafeBlocks for three test-only calls.
|
||||
[DllImport("libc", EntryPoint = "kill", SetLastError = true)]
|
||||
private static extern int Kill(int pid, int signal);
|
||||
|
||||
[DllImport("ntdll.dll")]
|
||||
private static extern int NtSuspendProcess(IntPtr processHandle);
|
||||
|
||||
[DllImport("ntdll.dll")]
|
||||
private static extern int NtResumeProcess(IntPtr processHandle);
|
||||
#pragma warning restore SYSLIB1054
|
||||
}
|
||||
Reference in New Issue
Block a user