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

perf(scripting): cut out-of-process runner host CPU; compile timeout from compile start; quiet start-failure logs (lmxopcua#789, #790, #788)
v2-ci / unit-tests (push) Has been cancelled
Details

Browse Source 
#789 (partial, further tuning deferred): combining inline frame writer, buffered
frame reader, one timer per evaluation for T and T+G, slot release on the reader
before the caller resumes, a non-spinning per-connection result dispatcher, and
blocking pipe I/O on Unix (both ends) so the socket never enters the socket
engine; TerminateJobObject kill on Windows. macOS host CPU at 10k evals/s K=2:
228-240 % -> 79 % (InProcess 69 %, IPC share +10 vs budget 25); windev 153 % ->
123 % (+115, still a miss). Harness reports the "vs InProcess" share, warm-cache
worker RSS, and times kill latency from the soft-timeout answer.

#790: the worker sends LoadStarted; the compile timeout is armed then, so a load
queued behind other compiles is never killed or struck.

#788: one Error with stack per down period, rate-limited Warnings with attempt
count, one Information on recovery.

Design doc §8.2 notes + §12.2 record.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-11-10 04:25:08 -05:00
parent 963b0d66fc
commit 8582ce91bc
14 changed files with 1576 additions and 317 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/2026-11-09-out-of-process-script-runner-design.md
+92 -2
View File
@@ -533,10 +533,10 @@ size for 2× headroom (10k/s). The passthrough fast path is excluded because it
| Added latency per evaluation, p50 | ≤ 150 µs | Host enqueue → reply decoded, trivial script, idle system |
| Added latency per evaluation, p99 | ≤ 2 ms | At 10k evals/s. Compare the 250 ms alarm budget: under 1% of it |
| Throughput | ≥ 10,000 trivial evals/s/node with K=2 | With opportunistic batching |
| Host CPU for IPC | ≤ 25% of one core at 10k evals/s | Serialisation + pipe I/O, measured by an opt-in perf test |
| Host CPU for IPC | ≤ 25% of one core at 10k evals/s | Serialisation + pipe I/O, measured by an opt-in perf test. Since lmxopcua#789 the harness reports it as the out-of-process runner's host CPU minus the in-process runner's in the same phase ("vs InProcess"): the raw figure includes the harness's own pacing loop and pool, which InProcess pays too (§12.2) |
| Bytes per eval on the wire | ≤ 64 B + 24 B per input | Positional slots, §4.4 |
| Shard restart → first result, 1,000 cached scripts | ≤ 2 s | §7 lazy PE load |
| Worker idle RSS | ≤ 80 MB + scripts **[A-3]** | Per shard |
| Worker idle RSS | ≤ 80 MB + scripts **[A-3]** | Per shard. **Proposed revision (lmxopcua#789, §12.2):** ≤ 80 MB + scripts for a worker that has only loaded cached PEs (measured 36–58 MB); ≤ 200 MB + scripts once Roslyn has run in it (measured 159–187 MB), because compiling in the worker (§14 #7) keeps Roslyn's code and metadata resident and no GC setting moves it |
| Kill latency after `T+G` | ≤ 100 ms | Host timer resolution plus `Process.Kill` |
**Comparison with today:** the in-process cost per compiled run is one `Task.Run` hop plus `WaitAsync`, which
@@ -873,6 +873,96 @@ Both misses → lmxopcua#789 (the §4.5 batching and per-eval allocation work).
The host load timeout still counts the time a load waits behind other compiles → lmxopcua#790.
### 12.2 lmxopcua#789 / #790 / #788 record (2026-11-10)
**What cost the host CPU (#789).** With `DOTNET_ThreadPool_UnfairSemaphoreSpinLimit=0` a scratch probe showed the
out-of-process runner at 33 % of a core at 10k evals/s against 25 % in process: the IPC work itself was about
10 µs per evaluation. The rest was thread-pool spin-waiting, triggered by several pool work items per evaluation, and
on Unix by the socket engine. Changes:
- **Combining frame writer** (`ScriptWorkerFrameWriter`): the thread that queues a frame writes it itself when no
write is in progress. This replaces the channel + writer task (one pool wake-up per submit). Consecutive
evaluations or results are merged into one batch, frame order is preserved, and there is one write per pass. The
old loop also called `PipeStream.FlushAsync`, which PipeStream does not override, so it queued a pool work item
per pass.
- **Buffered frame reader** (`ScriptWorkerFrameReader`): one read pulls in every frame available, and frames are
parsed from the buffer. It replaces two reads per frame plus a header allocation.
- **One timer per evaluation** for T and T+G. It replaces the `WaitAsync(timeout)` timer, the hard-deadline
`CancellationTokenSource` and their two continuations.
- **Slot release and deadline cancel run on the reader, before the caller is resumed.** The release no longer
depends on a continuation that is not ordered against the caller.
- **Result dispatcher** (`ResultDispatcher`, one thread per connection, blocking with `Monitor.Wait`, no spin):
callers resume there, not on the pool. They do not resume on the reader either, because the Calculation pump
publishes to OPC UA subscribers after its await. On the reader, a slow consumer would hold up heartbeats and get a
healthy worker killed. That is pinned by a test, and a mutation that resumed callers on the reader was red.
- **Blocking I/O on Unix** (`ScriptWorkerFrameCodec.UseBlockingIo`):
- The socket under a Unix .NET pipe turns non-blocking for good after one async call. After that even a
"blocking" read goes through the epoll thread plus a spinning wait.
- Both ends now connect, handshake, read (on a dedicated thread) and write synchronously. This was the largest
single gain: sequential host CPU per evaluation fell from about 85 µs to about 13 µs.
- A full pipe blocks only the one thread draining the writer; every other producer just queues. A peer that stops
reading is killed by the heartbeat check within 3 s, which fails that write.
- Windows keeps overlapped async I/O, because a synchronous full-duplex handle serialises reads against writes.
- **Kill via `TerminateJobObject`** on Windows. The job admits one process, so no system-wide snapshot is needed.
- Measured and **not** kept for memory, because none moved the compiled-worker RSS:
- ReadyToRun worker publish: 195 MB;
- `GCgen0size` 4 MB, `GCConserveMemory`, `TieredPGO=0`;
- an aggressive compacting GC after a compile burst.
The memory is Roslyn's JIT'd code and metadata, not the heap.
**Before / after (§8 harness, `OTOPCUA_SCRIPT_PERF=1`; macOS figures under 5–25 load average from other agents on
the box).**
| Budget | macOS before | macOS after | windev before | windev after |
|---|---:|---:|---:|---:|
| Host CPU at 10k/s, K=2, raw (InProcess same phase) | 228–240 % (71 %) | 79 % (69 %) | 153 % (6 %) | 123 % (7 %) |
| … vs InProcess (budget ≤ 25 %) | +157 | **+10 ✓** | +147 | +115 ✗ |
| Host CPU, idle sequential, K=2 | 258 % | 45 % | 80 % | 63 % |
| Idle p50 (≤ 150 µs) | 41–43 µs | 23 µs ✓ | 795 µs | 507 µs ✗ |
| p99 at 10k/s, K=2 (≤ 2 ms) | 720–796 µs | 1,024 µs ✓ | 1,916 µs | 1,676 µs ✓ |
| Closed loop ×64, K=2 (gate ≥ 5k/s) | 342k/s | 442k/s ✓ | 51k/s | 62k/s ✓ |
| Worker RSS after a compile (≤ 80 MB) | 188 MB | 187 MB ✗ | 158 MB | 159–161 MB ✗ |
| Worker RSS, warm PE cache, no compile | not measured | 58 MB ✓ | not measured | 36 MB ✓ |
| Warm restart, 1,000 cached (≤ 2 s) | 245 ms | 206–223 ms ✓ | 1,715 ms | 1,778–1,921 ms ✓ |
| Kill both shards → every next result (≤ 2 s) | 512 ms | 479–502 ms ✓ | 2,028 ms | 2,004–2,087 ms (borderline) |
| Kill latency after T+G (≤ 100 ms) | 31–35 ms | 33–43 ms ✓ | ~210 ms | **10–21 ms ✓** |
Additional measurements:
- In a scratch probe with a steady 10k/s load, K=2, before vs after: 116–158 % → 49–52 % of a core (InProcess
60–65 %). Sequential runs: about 25k → 46k evals/s.
- The first harness timed the kill from before the runaway's call, so it also counted the cold compile of each
distinct runaway key (100–200 ms on the VM). The harness now times from the soft-timeout answer. The windev
"before" figure of about 210 ms is therefore mostly compile time, not kill time.
- macOS K=1 p99 at 10k/s read 2.3 ms in one loaded run (budget 2 ms). K=2 is the configuration the budget names.
**Still missing after this slice (lmxopcua#789 stays open, further tuning deferred by the owner):**
- Windows host CPU for IPC: +115 points at 10k/s on the VM. That path still uses overlapped async I/O, so it pays a
completion-port and pool hand-off per frame. The next step is either two simplex pipes with synchronous handles and
dedicated threads, or a measurement on a non-VM box.
- Windows idle p50: 507 µs. The VM's in-process baseline is already 92 µs.
- Worker RSS after a compile. The proposed revised budget is in §8.2. An alternative is to recycle a worker from the
PE cache after a compile burst (blue/green); that is not built.
- Windows kill-both-shards: about 2.0 s, at the budget line.
**#790 — load timeout from compile start.**
- The worker sends `LoadStarted` when a compile thread picks a load up. The host arms that load's timer then,
instead of each caller timing from when it sent the load.
- A queued load is never blamed. A compile still running `LoadTimeout` after its start kills the worker and strikes
that key only (`LoadEntry.TimedOut`).
- `Hello.compile_threads` is a test knob; 0 keeps the worker's default.
- Tests:
- `Compiles_queued_behind_others_longer_than_the_load_timeout_are_not_killed_or_struck`: one compile thread, and
a burst grown until it outlasts three timeouts (asserted). Red with the timer armed at send.
- `A_compile_still_running_a_load_timeout_after_it_started_is_killed_and_struck`: the positive control, with two
kills leading to quarantine and the healthy key unblamed. Red with the timer never armed.
**#788 — start-failure logging.**
- The first failed start of a down period is one Error with the exception.
- Later retries are a Warning without a stack, carrying the attempt count, at most one per 60 s per shard; the rest
log at Debug.
- The start that ends the outage is one Information line: `started … after N failed start attempt(s) over T`.
- Tests: `A_shard_that_cannot_start_logs_one_error_then_concise_rate_limited_retries_then_its_recovery`, with two
intervals. Red when every failure is logged at Error with the exception.
---
## 13. Test plan
src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs
+90 -71
View File
@@ -136,6 +136,7 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
private readonly string _markerDirectory;
private readonly string? _peCacheDirectory;
private readonly byte[]? _peCacheKey;
private readonly int _compileThreads;
private long _peCacheHits;
private long _peCacheMisses;
private long _peCacheRejected;
@@ -153,8 +154,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)
TimeSpan heartbeatInterval, TimeSpan loadTimeout, TimeSpan? maintenanceInterval = null, string? markerDirectory = null,
int compileThreads = 0)
{
_compileThreads = compileThreads;
ArgumentNullException.ThrowIfNull(options);
_options = options.Sanitized(out _);
_worker = _options.Worker;
@@ -218,7 +221,10 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
_keys.Count(kv => kv.Value.QuarantineReason is not null),
_peCacheDirectory is not null);
/// <summary>How long a compile in the worker may take before the worker is killed.</summary>
/// <summary>
/// How long a compile in the worker may run before the worker is killed, counted from the moment a compile thread
/// starts it — never the time it spent queued behind other compiles (lmxopcua#790).
/// </summary>
internal TimeSpan LoadTimeout { get; }
/// <summary>The shard workers (tests).</summary>
@@ -340,12 +346,12 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
if (Volatile.Read(ref _disposed) != 0) return ScriptRunOutcome.Unavailable(ScriptRunOutcome.RunnerUnavailableReason);
var key = InProcessScriptRunner.ScopedKey(invocation.Kind, invocation.Key);
var state = Admit(key, invocation.Source);
var state = Admit(key, invocation.Kind, invocation.Source);
if (state.QuarantineReason is { } quarantined) return ScriptRunOutcome.Failed(quarantined);
var sha = SHA256.HashData(Encoding.UTF8.GetBytes(invocation.Source));
var failureKey = InProcessScriptRunner.ScopedKey(invocation.Kind, Convert.ToHexString(sha));
if (_compileFailures.TryGetValue(failureKey, out var knownFailure))
// The source hash and its negative-cache key are per source, so they live on the key's state (computed once per
// source, not per evaluation, lmxopcua#789).
if (_compileFailures.TryGetValue(state.FailureKey, out var knownFailure))
{
CompileCounter.Add(1, new KeyValuePair<string, object?>("cache", "negative"));
return ScriptRunOutcome.Failed(knownFailure);
@@ -364,37 +370,29 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
try
{
state.ScriptLogger = invocation.ScriptLogger;
// Circuit open (design §5.3.6): answer at once instead of making every trigger wait out the grace. A zero
// grace still kicks a start that is due, so the shards keep retrying at the backoff cap.
var grace = State == ScriptRunnerState.Unavailable ? TimeSpan.Zero : _worker.UnavailableGrace;
if (shard.IsolationGate is { } isolationGate)
{
if (!await isolationGate.WaitAsync(grace, ct).ConfigureAwait(false))
if (!await isolationGate.WaitAsync(Grace(), ct).ConfigureAwait(false))
return Unavailable(ScriptRunOutcome.RunnerUnavailableReason, shard, null);
gate = isolationGate;
}
connection = await shard.GetReadyAsync(grace, ct).ConfigureAwait(false);
// Fast path: the shard's worker is up (no state computation, no async hop).
connection = shard.Current ?? await shard.GetReadyAsync(Grace(), ct).ConfigureAwait(false);
if (connection is null) return Unavailable(ScriptRunOutcome.RunnerUnavailableReason, shard, null);
var paths = invocation.Inputs.Keys.Order(StringComparer.Ordinal).ToArray();
WorkerConnection.LoadEntry? entry;
try
var paths = SlotOrder(state, invocation.Inputs);
var entry = await connection
.EnsureLoadedAsync(key, WorkerShard.ToWire(invocation.Kind), invocation.Source, state.Sha, state.ShaHex, paths)
.WaitAsync(ct).ConfigureAwait(false);
if (entry.TimedOut)
{
entry = await connection
.EnsureLoadedAsync(key, WorkerShard.ToWire(invocation.Kind), invocation.Source, sha, paths)
.WaitAsync(LoadTimeout, ct).ConfigureAwait(false);
// A compile that kept a compile thread busy for LoadTimeout after it STARTED is the script's fault (a
// pathological source): the connection killed the worker and the exit handler struck the key. Time
// spent queued behind other compiles is not counted (lmxopcua#790).
return ScriptRunOutcome.Failed($"compile failure: compile did not finish within {connection.LoadTimeout.TotalSeconds:F0} s");
}
catch (TimeoutException)
{
// A compile that never finishes is the script's fault (a pathological source): the worker dies, the
// key takes a kill strike through the exit handler.
_log.Warning("Script {ScriptKey}: compile in ScriptWorker[{Shard}] exceeded {Timeout} s; killing the worker",
key, shard.Label, LoadTimeout.TotalSeconds);
connection.Kill(WorkerExitReason.KilledLoadTimeout, key);
return ScriptRunOutcome.Failed($"compile failure: compile did not finish within {LoadTimeout.TotalSeconds:F0} s");
}
if (entry?.Outcome is not { } outcome) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection);
if (entry.Outcome is not { } outcome) return Unavailable(ScriptRunOutcome.WorkerRestartedReason, RouteFor(key, state), connection);
NoteLoad(entry);
if (!outcome.Ok)
{
@@ -405,19 +403,17 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
_ => $"compile failure: {outcome.Error?.Message}",
};
// Remembered until the next deploy boundary, so a broken script never re-enters Roslyn (design §7).
_compileFailures.TryAdd(failureKey, reason);
_compileFailures.TryAdd(state.FailureKey, reason);
return ScriptRunOutcome.Failed(reason);
}
var submitted = Stopwatch.GetTimestamp();
var pending = connection.Submit(key, BuildRequest(key, entry, invocation));
ArmHardDeadline(connection, pending, invocation.Timeout + _worker.EffectiveHardGrace(invocation.Timeout));
// From here the in-flight slot (and the isolation gate) belong to the evaluation itself: they are released
// when the worker answers or dies — not when this caller stops waiting at the soft deadline. That keeps a
// key single-flight between T and T + G, and the isolation worker strictly one-at-a-time. On a worker
// death, attribution (strike, quarantine, suspect) has already run when this continuation fires.
handedOff = true;
// key single-flight between T and T + G, and the isolation worker strictly one-at-a-time. The release runs
// inside the connection BEFORE the caller's await resumes, so a caller that awaited one run and starts the
// next is never refused by the run it already has (the §8 harness found 32 of 200 refused when the release
// was an unordered continuation). On a worker death, attribution (strike, quarantine, suspect) has already
// run by then.
var heldGate = gate;
var released = 0;
void ReleaseSlot()
@@ -427,20 +423,16 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
Interlocked.Exchange(ref state.InFlight, 0);
heldGate?.Release();
}
_ = pending.Completion.ContinueWith(_ => ReleaseSlot(),
CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
EvalResult? result;
try
{
result = await pending.Completion.WaitAsync(invocation.Timeout, ct).ConfigureAwait(false);
// The worker has answered (or died, with attribution already run): free the slot HERE, before this
// caller resumes. The continuation above is not ordered against this await, so relying on it alone
// let a caller that awaited one run and then started the next see "previous evaluation still running"
// (found by the §8 perf harness: 32 of 200 sequential same-key runs refused). Idempotent.
ReleaseSlot();
}
catch (TimeoutException)
var submitted = Stopwatch.GetTimestamp();
var pending = connection.Submit(key, BuildRequest(key, entry, invocation), invocation.Timeout,
_worker.EffectiveHardGrace(invocation.Timeout), ReleaseSlot);
handedOff = true;
var result = ct.CanBeCanceled
? await pending.Outcome.WaitAsync(ct).ConfigureAwait(false)
: await pending.Outcome.ConfigureAwait(false);
if (ReferenceEquals(result, PendingEval.SoftTimedOut))
{
// As in process: the run token is cancelled only once the caller has its timeout, so a polling script
// ends itself (and frees its slot) without the worker being killed.
@@ -468,6 +460,32 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
gate?.Release();
}
}
// Circuit open (design §5.3.6): answer at once instead of making every trigger wait out the grace. A zero grace
// still kicks a start that is due, so the shards keep retrying at the backoff cap. Only computed when the
// shard has no live worker (or for the isolation gate).
TimeSpan Grace() => State == ScriptRunnerState.Unavailable ? TimeSpan.Zero : _worker.UnavailableGrace;
}
/// <summary>
/// The slot order for this evaluation's inputs: the key's previous order when the input set is the same (the
/// steady state — no sort, no allocation, lmxopcua#789), otherwise the ordinal sort of the new set.
/// </summary>
private static string[] SlotOrder(KeyState state, IReadOnlyDictionary<string, Core.Abstractions.DataValueSnapshot> inputs)
{
var cached = state.Paths;
if (cached is not null && cached.Length == inputs.Count)
{
var same = true;
foreach (var path in cached)
{
if (!inputs.ContainsKey(path)) { same = false; break; }
}
if (same) return cached;
}
var fresh = inputs.Keys.Order(StringComparer.Ordinal).ToArray();
state.Paths = fresh;
return fresh;
}
/// <inheritdoc />
@@ -555,7 +573,9 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
HandshakeTimeout: TimeSpan.FromSeconds(15),
PeCacheDirectory: _peCacheDirectory,
PeCacheKey: _peCacheKey,
MarkerDirectory: _markerDirectory);
MarkerDirectory: _markerDirectory,
LoadTimeout: LoadTimeout,
CompileThreads: _compileThreads);
return new WorkerShard(label, isolation,
(onExiting, ct) => WorkerConnection.StartAsync(spec, _log, OnLogs, onExiting, ct),
OnWorkerExit, _log);
@@ -591,15 +611,15 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
}
}
private KeyState Admit(string key, string source)
private KeyState Admit(string key, ScriptContextKind kind, string source)
{
while (true)
{
var state = _keys.GetOrAdd(key, static (_, src) => new KeyState(src), source);
if (string.Equals(state.Source, source, StringComparison.Ordinal)) return state;
var state = _keys.GetOrAdd(key, static (_, arg) => new KeyState(arg.source, arg.kind), (source, kind));
if (ReferenceEquals(state.Source, source) || string.Equals(state.Source, source, StringComparison.Ordinal)) return state;
// The script changed under this key: a clean slate, exactly as the in-process guard does. An evaluation
// of the old source still in flight releases the OLD state's slot when it ends.
if (_keys.TryUpdate(key, new KeyState(source), state)) continue;
if (_keys.TryUpdate(key, new KeyState(source, kind), state)) continue;
}
}
@@ -638,23 +658,6 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
return request;
}
private void ArmHardDeadline(WorkerConnection connection, PendingEval pending, TimeSpan after)
{
var cts = new CancellationTokenSource(after);
var registration = cts.Token.Register(() =>
{
if (pending.Completion.IsCompleted) return;
_log.Warning("Script {ScriptKey} exceeded its hard deadline ({Deadline} ms); killing ScriptWorker[{Shard}] pid {Pid}",
pending.Key, (long)after.TotalMilliseconds, connection.Shard, connection.Pid);
connection.Kill(WorkerExitReason.KilledDeadline, pending.Key);
});
_ = pending.Completion.ContinueWith(_ =>
{
registration.Dispose();
cts.Dispose();
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}
private ScriptRunOutcome Complete(string key, KeyState state, EvalResult result, WorkerShard shard, WorkerConnection connection)
{
// The script returned control (a value or a throw): the soft-timeout streak resets, as in process.
@@ -813,9 +816,25 @@ public sealed class OutOfProcessScriptRunner : IScriptRunner, IAsyncDisposable,
=> LiveRunners.Keys.SelectMany(r => r.AllShards.Select(s => s.Status()))
.Select(s => new Measurement<int>(value(s), new KeyValuePair<string, object?>("shard", s.Label)));
private sealed class KeyState(string source)
private sealed class KeyState(string source, ScriptContextKind kind)
{
private byte[]? _sha;
private string? _shaHex;
private string? _failureKey;
public readonly string Source = source;
/// <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));
/// <summary>Its upper-case hex form.</summary>
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);
/// <summary>The slot order last used for this key (see <see cref="SlotOrder"/>).</summary>
public string[]? Paths;
public readonly List<DateTime> KillStrikes = [];
public int InFlight;
public int ConsecutiveTimeouts;
src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/ResultDispatcher.cs
+90
View File
@@ -0,0 +1,90 @@
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess;
/// <summary>
/// The thread that resumes the callers of one worker's evaluations (lmxopcua#789).
/// </summary>
/// <remarks>
/// <para>Why not the thread pool: a result that wakes a pool worker makes that worker spin-wait for more work when it
/// is done, and results arrive one at a time, spread out, so every evaluation paid a full spin. The §8 harness
/// measured that spin as most of the host's CPU at 10k evals/s (with
/// <c>DOTNET_ThreadPool_UnfairSemaphoreSpinLimit=0</c> the out-of-process runner cost only ~10 points of a core more
/// than in-process). This thread blocks without spinning (<see cref="Monitor.Wait(object)"/>) and runs the
/// continuations of every result that is ready in one wake-up.</para>
/// <para>Why not the reader thread: the continuation is the caller's code (the Calculation pump publishes to OPC UA
/// subscribers after its await). On the reader it could hold up heartbeats and results, and a slow consumer could
/// get a healthy worker killed for silence. Here a slow consumer delays only the resumption of other callers; the
/// reader has already settled each result — its deadline timer cancelled, its key's slot released — before handing
/// it over, so no deadline can fire on an evaluation that has answered.</para>
/// <para>A caller that blocks on another evaluation's result from inside its continuation (sync over async) would
/// wait on this very thread; no seam does, and <see cref="IScriptRunner.RunAsync"/> is async-only for that reason.</para>
/// </remarks>
internal sealed class ResultDispatcher
{
private readonly object _gate = new();
private readonly Queue<(PendingEval Pending, EvalResult? Result)> _queue = new();
private bool _waiting;
private bool _stopped;
/// <summary>Starts the thread.</summary>
/// <param name="name">The thread name.</param>
public ResultDispatcher(string name)
{
new Thread(Run) { IsBackground = true, Name = name }.Start();
}
/// <summary>Queues a caller's resumption. After <see cref="Stop"/> it runs on the calling thread instead.</summary>
/// <param name="pending">The evaluation.</param>
/// <param name="result">Its result (null: the worker ended).</param>
public void Post(PendingEval pending, EvalResult? result)
{
lock (_gate)
{
if (!_stopped)
{
_queue.Enqueue((pending, result));
if (_waiting) Monitor.Pulse(_gate);
return;
}
}
pending.SignalOutcome(result);
}
/// <summary>Lets the thread drain what is queued and end.</summary>
public void Stop()
{
lock (_gate)
{
_stopped = true;
Monitor.Pulse(_gate);
}
}
private void Run()
{
var batch = new List<(PendingEval Pending, EvalResult? Result)>();
while (true)
{
lock (_gate)
{
while (_queue.Count == 0)
{
if (_stopped) return;
_waiting = true;
Monitor.Wait(_gate);
_waiting = false;
}
while (_queue.TryDequeue(out var item)) batch.Add(item);
}
foreach (var (pending, result) in batch)
{
// TrySetResult runs the caller's continuation here; an await or ContinueWith captures its own
// exceptions, so nothing should escape — but this thread must survive a misbehaving one regardless.
try { pending.SignalOutcome(result); }
catch (Exception) { /* the caller's own task carries its failure */ }
}
batch.Clear();
}
}
}
src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WindowsJobObject.cs
+13 -1
View File
@@ -59,6 +59,14 @@ internal sealed class WindowsJobObject : IDisposable
}
}
/// <summary>
/// Kills every process in the job at once (lmxopcua#789). The job admits exactly one process, so this is the
/// worker's whole tree without <c>Process.Kill(entireProcessTree: true)</c>'s system-wide process snapshot,
/// which cost about 100–200 ms of the Windows kill latency.
/// </summary>
/// <returns>False when the job could not be terminated (the caller falls back to a process kill).</returns>
public bool Terminate() => !_handle.IsClosed && TerminateJobObject(_handle, unchecked((uint)-1));
/// <summary>Closes the job handle, which kills every process still in it.</summary>
public void Dispose() => _handle.Dispose();
@@ -98,7 +106,7 @@ internal sealed class WindowsJobObject : IDisposable
public UIntPtr PeakJobMemoryUsed;
}
#pragma warning disable SYSLIB1054 // DllImport kept: LibraryImport would need AllowUnsafeBlocks for three calls.
#pragma warning disable SYSLIB1054 // DllImport kept: LibraryImport would need AllowUnsafeBlocks for four calls.
[DllImport("kernel32.dll", SetLastError = true, CharSet = CharSet.Unicode)]
private static extern SafeFileHandle CreateJobObjectW(IntPtr securityAttributes, string? name);
@@ -110,5 +118,9 @@ internal sealed class WindowsJobObject : IDisposable
[DllImport("kernel32.dll", SetLastError = true)]
[return: MarshalAs(UnmanagedType.Bool)]
private static extern bool AssignProcessToJobObject(SafeFileHandle job, IntPtr process);
[DllImport("kernel32.dll", SetLastError = true)]
[return: MarshalAs(UnmanagedType.Bool)]
private static extern bool TerminateJobObject(SafeFileHandle job, uint exitCode);
#pragma warning restore SYSLIB1054
}
src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs
+357 -125
View File
@@ -3,7 +3,6 @@ using System.Diagnostics;
using System.IO.Pipes;
using System.Security.Cryptography;
using System.Text;
using System.Threading.Channels;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
@@ -65,6 +64,11 @@ internal sealed record WorkerExit(WorkerExitReason Reason, string? CulpritKey, i
/// <param name="PeCacheDirectory">The compiled-PE cache directory handed to the worker; null = no cache.</param>
/// <param name="PeCacheKey">The cache's HMAC key (with <paramref name="PeCacheDirectory"/>).</param>
/// <param name="MarkerDirectory">Where the launch marker for the orphan sweep is written; null = none.</param>
/// <param name="LoadTimeout">
/// How long one compile may run, counted from the worker's <c>LoadStarted</c> for it (not from when the host sent
/// it, lmxopcua#790), before the worker is killed and the key struck; null = 60 s.
/// </param>
/// <param name="CompileThreads">The worker's dedicated compile threads; 0 = its default.</param>
internal sealed record WorkerLaunchSpec(
string Shard,
string WorkerPath,
@@ -75,7 +79,16 @@ internal sealed record WorkerLaunchSpec(
TimeSpan HandshakeTimeout,
string? PeCacheDirectory = null,
byte[]? PeCacheKey = null,
string? MarkerDirectory = null);
string? MarkerDirectory = null,
TimeSpan? LoadTimeout = null,
int CompileThreads = 0)
{
/// <summary>The default compile timeout.</summary>
public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60);
/// <summary><see cref="LoadTimeout"/> or its default.</summary>
public TimeSpan EffectiveLoadTimeout => LoadTimeout ?? DefaultLoadTimeout;
}
/// <summary>
/// One live ScriptWorker process and its pipe (lmxopcua#780, design §3–§6): launch with a cleared, allow-listed
@@ -101,13 +114,15 @@ internal sealed class WorkerConnection : IAsyncDisposable
private readonly string _workDir;
private readonly ILogger _log;
private readonly Action<LogBatch> _onLogs;
private readonly Channel<Envelope> _outbound = Channel.CreateUnbounded<Envelope>(new UnboundedChannelOptions { SingleReader = true });
private readonly ScriptWorkerFrameWriter _writer;
private readonly ResultDispatcher _dispatcher;
private readonly ConcurrentDictionary<ulong, PendingEval> _pending = new();
private readonly ConcurrentDictionary<ulong, (string Key, TaskCompletionSource<LoadResult> Completion)> _loads = new();
private readonly ConcurrentDictionary<ulong, LoadInFlight> _loads = new();
private readonly ConcurrentDictionary<string, LoadEntry> _loaded = new(StringComparer.Ordinal);
private readonly TaskCompletionSource<WorkerExit> _exited = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly CancellationTokenSource _stopping = new();
private readonly TimeSpan _heartbeatInterval;
private TimeSpan _loadTimeout;
private readonly Action<WorkerExit> _onExiting;
private readonly string? _markerPath;
private TaskCompletionSource<DiagnosticsReply>? _diag;
@@ -116,15 +131,15 @@ internal sealed class WorkerConnection : IAsyncDisposable
private int _finished;
private (WorkerExitReason Reason, string? Culprit)? _killedFor;
private long _lastFrameTimestamp;
private long _seq;
private long _nextEvalId;
private long _nextLoadId;
private int _maxFrameBytes = ScriptWorkerFrameCodec.DefaultMaxFrameBytes;
private int _disposed;
private WorkerConnection(string shard, Process process, NamedPipeServerStream pipe, IDisposable? job, string workDir,
ILogger log, Action<LogBatch> onLogs, TimeSpan heartbeatInterval, Action<WorkerExit> onExiting, string? markerPath)
ILogger log, Action<LogBatch> onLogs, TimeSpan heartbeatInterval, TimeSpan loadTimeout, Action<WorkerExit> onExiting,
string? markerPath)
{
_loadTimeout = loadTimeout;
_onExiting = onExiting;
_markerPath = markerPath;
Shard = shard;
@@ -136,6 +151,10 @@ internal sealed class WorkerConnection : IAsyncDisposable
_onLogs = onLogs;
_heartbeatInterval = heartbeatInterval;
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,
synchronous: ScriptWorkerFrameCodec.UseBlockingIo);
_dispatcher = new ResultDispatcher($"ScriptWorker[{shard}].Results");
}
/// <summary>The shard label.</summary>
@@ -156,6 +175,16 @@ internal sealed class WorkerConnection : IAsyncDisposable
/// <summary>Evaluations currently outstanding on this worker.</summary>
public int InFlightCount => _pending.Count;
/// <summary>
/// The compile timeout applied to loads that START from now on (test hook: a test warms the worker's compiler
/// under the launch value, then shortens it).
/// </summary>
internal TimeSpan LoadTimeout
{
get => _loadTimeout;
set => _loadTimeout = value;
}
/// <summary>Compiles currently outstanding on this worker.</summary>
public int LoadingCount => _loads.Count;
@@ -234,16 +263,20 @@ internal sealed class WorkerConnection : IAsyncDisposable
HardGraceMs = (uint)Math.Clamp(spec.HardGrace.TotalMilliseconds, 1, uint.MaxValue),
HeartbeatIntervalMs = (uint)spec.HeartbeatInterval.TotalMilliseconds,
MaxLogLines = 1000,
CompileThreads = (uint)Math.Max(0, spec.CompileThreads),
};
if (spec.PeCacheDirectory is { } cacheDir && spec.PeCacheKey is { } cacheKey)
{
hello.PeCacheDir = cacheDir;
hello.PeCacheKey = Google.Protobuf.ByteString.CopyFrom(cacheKey);
}
await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false);
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 connection = new WorkerConnection(spec.Shard, process, pipe, job, workDir, log, onLogs, spec.HeartbeatInterval,
onExiting, markerPath);
spec.EffectiveLoadTimeout, onExiting, markerPath);
connection.Begin();
return connection;
}
@@ -280,8 +313,41 @@ internal sealed class WorkerConnection : IAsyncDisposable
}, ScriptWorkerFrameCodec.DefaultMaxFrameBytes, ct).ConfigureAwait(false);
await stream.FlushAsync(ct).ConfigureAwait(false);
var reply = await ScriptWorkerFrameCodec.ReadAsync(stream, ScriptWorkerFrameCodec.DefaultMaxFrameBytes, ct).ConfigureAwait(false)
?? throw new ScriptWorkerProtocolException("the worker closed the pipe during the handshake");
var reply = await ScriptWorkerFrameCodec.ReadAsync(stream, ScriptWorkerFrameCodec.DefaultMaxFrameBytes, ct).ConfigureAwait(false);
return CheckReady(reply, expectedNonce, expectedPid);
}
/// <summary>
/// As <see cref="HandshakeAsync"/> with blocking I/O only (<see cref="ScriptWorkerFrameCodec.UseBlockingIo"/>):
/// one asynchronous call would turn the socket under a Unix pipe non-blocking for the connection's whole life.
/// The blocking read runs off the caller and is abandoned at <paramref name="ct"/>; the caller then kills the
/// worker and disposes the pipe, which ends it.
/// </summary>
/// <param name="stream">The accepted pipe.</param>
/// <param name="hello">The Hello to send.</param>
/// <param name="expectedNonce">The nonce put in the worker's environment.</param>
/// <param name="expectedPid">The launched process id; null skips the check.</param>
/// <param name="ct">Cancellation (the handshake timeout).</param>
/// <returns>The worker's Ready.</returns>
internal static async Task<Ready> HandshakeBlockingAsync(Stream stream, Hello hello, string expectedNonce, int? expectedPid,
CancellationToken ct)
{
var reply = await Task.Run(() =>
{
ScriptWorkerFrameCodec.Write(stream, new Envelope
{
ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
Seq = 1,
Hello = hello,
}, ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
return ScriptWorkerFrameCodec.Read(stream, ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
}, CancellationToken.None).WaitAsync(ct).ConfigureAwait(false);
return CheckReady(reply, expectedNonce, expectedPid);
}
private static Ready CheckReady(Envelope? reply, string expectedNonce, int? expectedPid)
{
if (reply is null) throw new ScriptWorkerProtocolException("the worker closed the pipe during the handshake");
if (reply.BodyCase != Envelope.BodyOneofCase.Ready)
throw new ScriptWorkerProtocolException($"expected Ready, got {reply.BodyCase}");
var expected = Encoding.UTF8.GetBytes(expectedNonce);
@@ -361,8 +427,8 @@ internal sealed class WorkerConnection : IAsyncDisposable
ReadyTimestamp = Stopwatch.GetTimestamp();
Volatile.Write(ref _lastFrameTimestamp, ReadyTimestamp);
Interlocked.Exchange(ref _lastUsedTimestamp, ReadyTimestamp);
_ = Task.Run(ReadLoopAsync);
_ = Task.Run(WriteLoopAsync);
if (!ScriptWorkerFrameCodec.UseBlockingIo) _ = Task.Run(ReadLoopAsync);
else new Thread(ReadLoopBlocking) { IsBackground = true, Name = $"ScriptWorker[{Shard}].Reader" }.Start();
_ = Task.Run(HeartbeatMonitorAsync);
_ = Task.Run(WatchProcessAsync);
}
@@ -373,39 +439,39 @@ internal sealed class WorkerConnection : IAsyncDisposable
/// <param name="source">The source.</param>
/// <param name="sourceHash">SHA-256 of the source.</param>
/// <param name="inputPaths">The slot order.</param>
/// <returns>The load outcome and the slot map; null when the worker ended first.</returns>
public Task<LoadEntry?> EnsureLoadedAsync(string key, ContextKind kind, string source, byte[] sourceHash, IReadOnlyList<string> inputPaths)
/// <returns>The entry once the load has an outcome (see <see cref="LoadEntry.Outcome"/>).</returns>
public Task<LoadEntry> EnsureLoadedAsync(string key, ContextKind kind, string source, byte[] sourceHash, IReadOnlyList<string> inputPaths)
=> EnsureLoadedAsync(key, kind, source, sourceHash, Convert.ToHexString(sourceHash), inputPaths);
/// <summary>As the other overload, with the hash's hex form precomputed (the runner caches it per key, lmxopcua#789).</summary>
/// <param name="key">The scoped key.</param>
/// <param name="kind">The context kind.</param>
/// <param name="source">The source.</param>
/// <param name="sourceHash">SHA-256 of the source.</param>
/// <param name="sourceHashHex">Its upper-case hex form.</param>
/// <param name="inputPaths">The slot order.</param>
/// <returns>
/// The entry once the load has an outcome: <see cref="LoadEntry.Outcome"/> is the worker's verdict, or null when
/// the worker ended first; <see cref="LoadEntry.TimedOut"/> marks the compile this connection killed the worker
/// for.
/// </returns>
public Task<LoadEntry> EnsureLoadedAsync(string key, ContextKind kind, string source, byte[] sourceHash, string sourceHashHex,
IReadOnlyList<string> inputPaths)
{
var hashHex = Convert.ToHexString(sourceHash);
if (_loaded.TryGetValue(key, out var existing) && existing.SourceHash == hashHex && existing.SameSlots(inputPaths))
if (_loaded.TryGetValue(key, out var existing) && existing.SourceHash == sourceHashHex && existing.SameSlots(inputPaths))
return existing.Ready;
var entry = new LoadEntry(hashHex, inputPaths);
var entry = new LoadEntry(sourceHashHex, inputPaths);
_loaded[key] = entry;
Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp());
var loadId = (ulong)Interlocked.Increment(ref _nextLoadId);
var tcs = new TaskCompletionSource<LoadResult>(TaskCreationOptions.RunContinuationsAsynchronously);
_loads[loadId] = (key, tcs);
var load = new LoadScripts { LoadId = loadId };
var script = new ScriptSource { Key = key, Kind = kind, Source = source, SourceSha256 = Google.Protobuf.ByteString.CopyFrom(sourceHash) };
script.InputPaths.AddRange(inputPaths);
load.Scripts.Add(script);
if (!Send(new Envelope { Load = load })) { _loads.TryRemove(loadId, out _); entry.Complete(null); return entry.Ready; }
_ = tcs.Task.ContinueWith(t =>
{
if (t.IsCompletedSuccessfully)
{
var outcome = t.Result.Outcomes.FirstOrDefault(o => o.Key == key);
entry.Complete(outcome ?? new LoadOutcome { Key = key, Ok = false, Error = new ScriptError { Kind = ScriptError.Types.Kind.Threw, Message = "no load outcome" } });
if (outcome is { Ok: false }) _loaded.TryRemove(new KeyValuePair<string, LoadEntry>(key, entry));
}
else
{
entry.Complete(null);
_loaded.TryRemove(new KeyValuePair<string, LoadEntry>(key, entry));
}
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
var inFlight = new LoadInFlight(loadId, key, entry, this);
_loads[loadId] = inFlight;
if (!Send(new Envelope { Load = load }) && _loads.TryRemove(loadId, out _)) inFlight.Finish(null);
return entry.Ready;
}
@@ -423,11 +489,29 @@ internal sealed class WorkerConnection : IAsyncDisposable
/// <param name="key">The scoped key.</param>
/// <param name="request">The request (its id is assigned here).</param>
/// <returns>The pending evaluation.</returns>
public PendingEval Submit(string key, EvalRequest request)
public PendingEval Submit(string key, EvalRequest request) => Submit(key, request, null, TimeSpan.Zero, null);
/// <summary>
/// Submits one evaluation with its two deadlines (design §5.1) on ONE timer (lmxopcua#789): at
/// <paramref name="softTimeout"/> <see cref="PendingEval.Outcome"/> yields <see cref="PendingEval.SoftTimedOut"/>,
/// and <paramref name="hardGrace"/> later, if the worker still has not answered, the worker is killed with this
/// key as the culprit.
/// </summary>
/// <param name="key">The scoped key.</param>
/// <param name="request">The request (its id is assigned here).</param>
/// <param name="softTimeout">T; null = no deadlines.</param>
/// <param name="hardGrace">G.</param>
/// <param name="onFinished">
/// Runs once, synchronously, when the worker has answered or ended — BEFORE <see cref="PendingEval.Outcome"/>
/// completes — so a caller that resumes from its await never finds its own run still holding the key's slot.
/// </param>
/// <returns>The pending evaluation.</returns>
public PendingEval Submit(string key, EvalRequest request, TimeSpan? softTimeout, TimeSpan hardGrace, Action? onFinished)
{
request.EvalId = (ulong)Interlocked.Increment(ref _nextEvalId);
Interlocked.Exchange(ref _lastUsedTimestamp, Stopwatch.GetTimestamp());
var pending = new PendingEval(key, request.EvalId);
var pending = new PendingEval(key, request.EvalId, onFinished, _dispatcher);
if (softTimeout is { } soft) pending.Arm(soft, hardGrace, OnHardDeadline);
_pending[request.EvalId] = pending;
var batch = new EvalBatch();
batch.Items.Add(request);
@@ -438,6 +522,13 @@ internal sealed class WorkerConnection : IAsyncDisposable
return pending;
}
private void OnHardDeadline(PendingEval pending)
{
_log.Warning("Script {ScriptKey} exceeded its hard deadline ({Deadline} ms); killing ScriptWorker[{Shard}] pid {Pid}",
pending.Key, (long)pending.HardDeadline.TotalMilliseconds, Shard, Pid);
Kill(WorkerExitReason.KilledDeadline, pending.Key);
}
/// <summary>Cancels a run's <c>ctx.CancellationToken</c> in the worker (sent at the soft deadline).</summary>
/// <param name="pending">The evaluation.</param>
public void Cancel(PendingEval pending)
@@ -467,6 +558,8 @@ internal sealed class WorkerConnection : IAsyncDisposable
}
try
{
// The job holds exactly this one process (ACTIVE_PROCESS = 1), so terminating it is the whole tree, at once.
if (OperatingSystem.IsWindows() && _job is WindowsJobObject job && job.Terminate()) return;
if (!_process.HasExited) _process.Kill(entireProcessTree: true);
}
catch (InvalidOperationException) { /* already gone */ }
@@ -502,7 +595,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
try { await _exited.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); } catch (TimeoutException) { }
}
_stopping.Cancel();
_outbound.Writer.TryComplete();
_writer.Complete();
await _pipe.DisposeAsync().ConfigureAwait(false);
_job?.Dispose();
_process.Dispose();
@@ -513,106 +606,112 @@ internal sealed class WorkerConnection : IAsyncDisposable
/// <summary>Test hook: closes the host end of the pipe without touching the process, as a dying host would.</summary>
internal void SeverPipeForTest() => _pipe.Dispose();
private bool Send(Envelope envelope)
private bool Send(Envelope envelope) => _writer.TryWrite(envelope);
private void OnWriteFault(Exception ex)
{
envelope.ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion;
return _outbound.Writer.TryWrite(envelope);
if (!_stopping.IsCancellationRequested)
_log.Debug(ex, "ScriptWorker[{Shard}] pipe write ended", Shard);
Kill(WorkerExitReason.Crash, null);
}
private async Task WriteLoopAsync()
/// <summary>
/// The reader, on a dedicated thread with blocking reads of a socket that never turns non-blocking
/// (<see cref="ScriptWorkerFrameCodec.UseBlockingIo"/>, lmxopcua#789): a frame wakes this thread straight from
/// the kernel, instead of the socket engine's epoll thread queueing a thread-pool work item per read. Not on
/// Windows, where a synchronous read of an overlapped pipe is itself completed through the I/O completion port;
/// the async loop is the cheaper path there.
/// </summary>
private void ReadLoopBlocking()
{
var reader = _outbound.Reader;
WorkerExitReason? fault = null;
var reader = new ScriptWorkerFrameReader(_pipe);
try
{
while (await reader.WaitToReadAsync(_stopping.Token).ConfigureAwait(false))
{
// Opportunistic batching (design §4.5): every evaluation queued right now goes out as ONE EvalBatch.
EvalBatch? batch = null;
while (reader.TryRead(out var envelope))
{
if (envelope.BodyCase == Envelope.BodyOneofCase.Eval)
{
(batch ??= new EvalBatch()).Items.AddRange(envelope.Eval.Items);
continue;
}
envelope.Seq = (ulong)Interlocked.Increment(ref _seq);
await ScriptWorkerFrameCodec.WriteAsync(_pipe, envelope, _maxFrameBytes, _stopping.Token).ConfigureAwait(false);
}
if (batch is not null)
{
await ScriptWorkerFrameCodec.WriteAsync(_pipe, new Envelope
{
ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion,
Seq = (ulong)Interlocked.Increment(ref _seq),
Eval = batch,
}, _maxFrameBytes, _stopping.Token).ConfigureAwait(false);
}
await _pipe.FlushAsync(_stopping.Token).ConfigureAwait(false);
}
while (reader.Read() is { } frame) HandleFrame(frame);
}
catch (OperationCanceledException) { }
catch (Exception ex) when (ex is IOException or ObjectDisposedException or ScriptWorkerProtocolException)
catch (ScriptWorkerProtocolException ex)
{
if (!_stopping.IsCancellationRequested)
_log.Debug(ex, "ScriptWorker[{Shard}] pipe write ended", Shard);
Kill(WorkerExitReason.Crash, null);
fault = ProtocolFault(ex);
}
catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException
or System.Net.Sockets.SocketException)
{
}
_ = PipeEndedAsync(fault);
}
private async Task ReadLoopAsync()
{
WorkerExitReason? fault = null;
var reader = new ScriptWorkerFrameReader(_pipe);
try
{
while (true)
{
var frame = await ScriptWorkerFrameCodec.ReadAsync(_pipe, _maxFrameBytes, _stopping.Token).ConfigureAwait(false);
if (frame is null) break;
Volatile.Write(ref _lastFrameTimestamp, Stopwatch.GetTimestamp());
switch (frame.BodyCase)
{
case Envelope.BodyOneofCase.Results:
foreach (var result in frame.Results.Items)
{
// An eval id with no outstanding request is a protocol violation (design §6 "what the host
// trusts"), except for one that was already completed by a kill race — ignored then.
if (_pending.TryRemove(result.EvalId, out var pending)) pending.Complete(result);
else if (result.EvalId == 0 || result.EvalId > (ulong)Interlocked.Read(ref _nextEvalId))
throw new ScriptWorkerProtocolException($"result for unknown eval id {result.EvalId}");
}
break;
case Envelope.BodyOneofCase.LoadResult:
if (_loads.TryRemove(frame.LoadResult.LoadId, out var load)) load.Completion.TrySetResult(frame.LoadResult);
else throw new ScriptWorkerProtocolException($"load result for unknown load id {frame.LoadResult.LoadId}");
break;
case Envelope.BodyOneofCase.Logs:
try { _onLogs(frame.Logs); }
catch (Exception ex) { _log.Warning(ex, "ScriptWorker[{Shard}] log forwarding threw", Shard); }
break;
case Envelope.BodyOneofCase.Heartbeat:
Volatile.Write(ref _lastHeartbeat, frame.Heartbeat);
break;
case Envelope.BodyOneofCase.DiagReply:
Volatile.Read(ref _diag)?.TrySetResult(frame.DiagReply);
break;
case Envelope.BodyOneofCase.Fault:
_log.Warning("ScriptWorker[{Shard}] reported a fault: {Fault}", Shard, frame.Fault.Message);
break;
default:
throw new ScriptWorkerProtocolException($"unexpected frame {frame.BodyCase} from the worker");
}
}
while (await reader.ReadAsync(_stopping.Token).ConfigureAwait(false) is { } frame) HandleFrame(frame);
}
catch (OperationCanceledException) { }
catch (ScriptWorkerProtocolException ex)
{
_log.Warning("ScriptWorker[{Shard}] protocol violation: {Message}; killing pid {Pid}", Shard, ex.Message, Pid);
fault = WorkerExitReason.ProtocolFault;
fault = ProtocolFault(ex);
}
catch (Exception ex) when (ex is IOException or ObjectDisposedException)
{
}
await PipeEndedAsync(fault).ConfigureAwait(false);
}
private WorkerExitReason ProtocolFault(ScriptWorkerProtocolException ex)
{
_log.Warning("ScriptWorker[{Shard}] protocol violation: {Message}; killing pid {Pid}", Shard, ex.Message, Pid);
return WorkerExitReason.ProtocolFault;
}
private void HandleFrame(Envelope frame)
{
Volatile.Write(ref _lastFrameTimestamp, Stopwatch.GetTimestamp());
switch (frame.BodyCase)
{
case Envelope.BodyOneofCase.Results:
foreach (var result in frame.Results.Items)
{
// An eval id with no outstanding request is a protocol violation (design §6 "what the host
// trusts"), except for one that was already completed by a kill race — ignored then.
if (_pending.TryRemove(result.EvalId, out var pending)) pending.Complete(result);
else if (result.EvalId == 0 || result.EvalId > (ulong)Interlocked.Read(ref _nextEvalId))
throw new ScriptWorkerProtocolException($"result for unknown eval id {result.EvalId}");
}
break;
case Envelope.BodyOneofCase.LoadStarted:
// The compile timeout starts now (lmxopcua#790). An unknown id is a load already completed by a
// kill race (ignored) or one never issued (a protocol violation).
if (_loads.TryGetValue(frame.LoadStarted.LoadId, out var starting)) starting.Started(_loadTimeout);
else if (frame.LoadStarted.LoadId == 0 || frame.LoadStarted.LoadId > (ulong)Interlocked.Read(ref _nextLoadId))
throw new ScriptWorkerProtocolException($"load start for unknown load id {frame.LoadStarted.LoadId}");
break;
case Envelope.BodyOneofCase.LoadResult:
if (_loads.TryRemove(frame.LoadResult.LoadId, out var load)) load.Finish(frame.LoadResult);
else throw new ScriptWorkerProtocolException($"load result for unknown load id {frame.LoadResult.LoadId}");
break;
case Envelope.BodyOneofCase.Logs:
try { _onLogs(frame.Logs); }
catch (Exception ex) { _log.Warning(ex, "ScriptWorker[{Shard}] log forwarding threw", Shard); }
break;
case Envelope.BodyOneofCase.Heartbeat:
Volatile.Write(ref _lastHeartbeat, frame.Heartbeat);
break;
case Envelope.BodyOneofCase.DiagReply:
Volatile.Read(ref _diag)?.TrySetResult(frame.DiagReply);
break;
case Envelope.BodyOneofCase.Fault:
_log.Warning("ScriptWorker[{Shard}] reported a fault: {Fault}", Shard, frame.Fault.Message);
break;
default:
throw new ScriptWorkerProtocolException($"unexpected frame {frame.BodyCase} from the worker");
}
}
private async Task PipeEndedAsync(WorkerExitReason? fault)
{
// The pipe is gone: the worker is dead or dying. Make sure, then report.
try
{
@@ -670,7 +769,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
if (Interlocked.Exchange(ref _finished, 1) != 0) return;
// No new frame can be queued from here on, so every Submit / load either fails its Send (and completes itself)
// or was registered before this point and is completed below.
_outbound.Writer.TryComplete();
_writer.Complete();
int? code = null;
try { if (_process.HasExited) code = _process.ExitCode; } catch (InvalidOperationException) { }
@@ -692,9 +791,20 @@ internal sealed class WorkerConnection : IAsyncDisposable
foreach (var id in _pending.Keys.ToArray())
if (_pending.TryRemove(id, out var pending)) pending.Complete(null);
foreach (var id in _loads.Keys.ToArray())
if (_loads.TryRemove(id, out var load)) load.Completion.TrySetCanceled();
if (_loads.TryRemove(id, out var load)) load.Finish(null);
Volatile.Read(ref _diag)?.TrySetCanceled();
_outbound.Writer.TryComplete();
// Resumes the callers just completed above, then ends; a later Post runs inline.
_dispatcher.Stop();
}
/// <summary>The compile timer of a started load fired: kill the worker, blaming this key alone (lmxopcua#790).</summary>
private void OnLoadTimeout(LoadInFlight load)
{
if (!_loads.ContainsKey(load.Id)) return;
load.Entry.TimedOut = true;
_log.Warning("Script {ScriptKey}: compile in ScriptWorker[{Shard}] has run for {Timeout} s since it started; killing the worker",
load.Key, Shard, _loadTimeout.TotalSeconds);
Kill(WorkerExitReason.KilledLoadTimeout, load.Key);
}
private static string SafeExitCode(Process process)
@@ -708,10 +818,52 @@ internal sealed class WorkerConnection : IAsyncDisposable
try { if (Directory.Exists(dir)) Directory.Delete(dir, recursive: true); } catch { /* temp dir; best effort */ }
}
/// <summary>One load sent to the worker and not yet answered: its compile timer runs from <c>LoadStarted</c>.</summary>
private sealed class LoadInFlight(ulong id, string key, LoadEntry entry, WorkerConnection owner)
{
private Timer? _timer;
private int _finished;
public ulong Id { get; } = id;
public string Key { get; } = key;
public LoadEntry Entry { get; } = entry;
/// <summary>The worker has started compiling it: arm the compile timeout (once).</summary>
public void Started(TimeSpan timeout)
{
if (Volatile.Read(ref _finished) != 0 || _timer is not null) return;
var timer = new Timer(static s => { var l = (LoadInFlight)s!; l.Owner.OnLoadTimeout(l); }, this, Timeout.Infinite, Timeout.Infinite);
if (Interlocked.CompareExchange(ref _timer, timer, null) is not null) { timer.Dispose(); return; }
timer.Change(timeout, Timeout.InfiniteTimeSpan);
if (Volatile.Read(ref _finished) != 0) timer.Dispose();
}
private WorkerConnection Owner { get; } = owner;
/// <summary>The worker answered (<paramref name="result"/>) or ended (null).</summary>
public void Finish(LoadResult? result)
{
if (Interlocked.Exchange(ref _finished, 1) != 0) return;
Volatile.Read(ref _timer)?.Dispose();
if (result is null)
{
Owner._loaded.TryRemove(new KeyValuePair<string, LoadEntry>(Key, Entry));
Entry.Complete(null);
return;
}
var outcome = result.Outcomes.FirstOrDefault(o => o.Key == Key)
?? new LoadOutcome { Key = Key, Ok = false, Error = new ScriptError { Kind = ScriptError.Types.Kind.Threw, Message = "no load outcome" } };
if (!outcome.Ok) Owner._loaded.TryRemove(new KeyValuePair<string, LoadEntry>(Key, Entry));
Entry.Complete(outcome);
}
}
/// <summary>A key's compiled form in this worker.</summary>
internal sealed class LoadEntry
{
private readonly TaskCompletionSource<LoadEntry?> _ready = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly TaskCompletionSource<LoadEntry> _ready = new(TaskCreationOptions.RunContinuationsAsynchronously);
/// <summary>Set once the runner has counted this load in its metrics.</summary>
public int Counted;
@@ -732,10 +884,17 @@ internal sealed class WorkerConnection : IAsyncDisposable
/// <summary>Path → slot in the registered order.</summary>
public IReadOnlyDictionary<string, uint> Slots { get; }
/// <summary>The worker's verdict (null until it arrives).</summary>
/// <summary>The worker's verdict; null until it arrives, and null for good when the worker ended first.</summary>
public LoadOutcome? Outcome { get; private set; }
public Task<LoadEntry?> Ready => _ready.Task;
/// <summary>
/// Set when this connection killed the worker because THIS compile overran its timeout from its start
/// (lmxopcua#790); the waiters then report a compile timeout instead of a worker restart.
/// </summary>
public volatile bool TimedOut;
/// <summary>Completes with this entry once the load has an outcome (or the worker ended).</summary>
public Task<LoadEntry> Ready => _ready.Task;
public bool SameSlots(IReadOnlyList<string> inputPaths)
{
@@ -748,20 +907,42 @@ internal sealed class WorkerConnection : IAsyncDisposable
public void Complete(LoadOutcome? outcome)
{
Outcome = outcome;
_ready.TrySetResult(outcome is null ? null : this);
_ready.TrySetResult(this);
}
}
}
/// <summary>One evaluation outstanding on a worker.</summary>
/// <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.
/// </summary>
internal sealed class PendingEval
{
private readonly TaskCompletionSource<EvalResult?> _completion = new(TaskCreationOptions.RunContinuationsAsynchronously);
/// <summary>What <see cref="Outcome"/> yields when the soft deadline passed first. Never on the wire.</summary>
internal static readonly EvalResult SoftTimedOut = new();
public PendingEval(string key, ulong evalId)
private readonly TaskCompletionSource<EvalResult?> _completion = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly TaskCompletionSource<EvalResult?> _outcome;
private readonly ResultDispatcher? _dispatcher;
private readonly Action? _onFinished;
private Timer? _timer;
private Action<PendingEval>? _onHardDeadline;
private TimeSpan _grace;
private int _phase;
private int _done;
public PendingEval(string key, ulong evalId, Action? onFinished = null, ResultDispatcher? dispatcher = null)
{
Key = key;
EvalId = evalId;
_onFinished = onFinished;
_dispatcher = dispatcher;
// With a dispatcher the caller resumes ON the dispatcher's thread (synchronous continuation, no pool hop);
// without one, on the thread pool as usual.
_outcome = new TaskCompletionSource<EvalResult?>(dispatcher is null
? TaskCreationOptions.RunContinuationsAsynchronously
: TaskCreationOptions.None);
}
/// <summary>The scoped key.</summary>
@@ -770,8 +951,59 @@ internal sealed class PendingEval
/// <summary>The eval id on the wire.</summary>
public ulong EvalId { get; }
/// <summary>T + G, for the kill log line.</summary>
public TimeSpan HardDeadline { get; private set; }
/// <summary>The worker's result, or null when the worker ended first.</summary>
public Task<EvalResult?> Completion => _completion.Task;
public void Complete(EvalResult? result) => _completion.TrySetResult(result);
/// <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).
/// </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)
{
_grace = grace;
_onHardDeadline = onHardDeadline;
HardDeadline = soft + grace;
// 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);
}
private void OnTimer()
{
if (Volatile.Read(ref _done) != 0) return;
if (Interlocked.CompareExchange(ref _phase, 1, 0) == 0)
{
_outcome.TrySetResult(SoftTimedOut);
try { _timer?.Change(_grace, Timeout.InfiniteTimeSpan); }
catch (ObjectDisposedException) { /* answered meanwhile */ }
return;
}
if (Interlocked.CompareExchange(ref _phase, 2, 1) == 1 && Volatile.Read(ref _done) == 0) _onHardDeadline?.Invoke(this);
}
/// <summary>
/// The worker answered (or ended, with null): cancel the deadlines and release the key's slot HERE, on the
/// reader, then hand the caller's resumption to the dispatcher (or the pool).
/// </summary>
public void Complete(EvalResult? result)
{
if (Interlocked.Exchange(ref _done, 1) != 0) return;
_timer?.Dispose();
try { _onFinished?.Invoke(); }
finally
{
_completion.TrySetResult(result);
if (_dispatcher is { } dispatcher) dispatcher.Post(this, result);
else _outcome.TrySetResult(result);
}
}
/// <summary>Resumes the caller (on the dispatcher's thread). A soft timeout may already have done so.</summary>
internal void SignalOutcome(EvalResult? result) => _outcome.TrySetResult(result);
}
src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerShard.cs
+48 -3
View File
@@ -45,6 +45,12 @@ internal sealed class WorkerShard : IAsyncDisposable
internal static readonly TimeSpan MaxBackoff = TimeSpan.FromSeconds(30);
internal static readonly TimeSpan HealthyResetAfter = TimeSpan.FromSeconds(60);
/// <summary>
/// While a shard keeps failing to start, at most one Warning per this interval (lmxopcua#788); the other retries
/// log at Debug. The first failure of a down period is the one Error (with the exception).
/// </summary>
internal static readonly TimeSpan DefaultStartFailureWarningInterval = TimeSpan.FromSeconds(60);
private static readonly Counter<long> StartsCounter = OtOpcUaTelemetry.Meter.CreateCounter<long>(
"otopcua.script.worker.starts", unit: "{process}", description: "ScriptWorker processes started (tag: shard).");
@@ -58,6 +64,7 @@ internal sealed class WorkerShard : IAsyncDisposable
private readonly object _gate = new();
private readonly Random _jitter = new();
private readonly CancellationTokenSource _disposing = new();
private readonly TimeSpan _startFailureWarningInterval;
private WorkerConnection? _current;
private Task<WorkerConnection?>? _starting;
private TaskCompletionSource? _recovered;
@@ -67,9 +74,15 @@ internal sealed class WorkerShard : IAsyncDisposable
private int _consecutiveFailures;
private bool _disposed;
// The current run of failed starts (lmxopcua#788): how many, since when, and when the last Warning went out.
private int _startFailures;
private long _startFailingSince;
private long _lastStartFailureWarning;
public WorkerShard(string label, bool isolation, Func<Action<WorkerExit>, CancellationToken, Task<WorkerConnection>> launch,
Action<WorkerShard, WorkerExit> onExit, ILogger log)
Action<WorkerShard, WorkerExit> onExit, ILogger log, TimeSpan? startFailureWarningInterval = null)
{
_startFailureWarningInterval = startFailureWarningInterval ?? DefaultStartFailureWarningInterval;
Label = label;
IsIsolation = isolation;
_launch = launch;
@@ -298,7 +311,21 @@ internal sealed class WorkerShard : IAsyncDisposable
{
connection = await _launch(Attribute, CancellationToken.None).ConfigureAwait(false);
StartsCounter.Add(1, new KeyValuePair<string, object?>("shard", Label));
_log.Information("ScriptWorker[{Shard}] started, pid {Pid}", Label, connection.Pid);
int failedBefore;
long failingSince;
lock (_gate)
{
failedBefore = _startFailures;
failingSince = _startFailingSince;
_startFailures = 0;
_startFailingSince = 0;
_lastStartFailureWarning = 0;
}
if (failedBefore == 0)
_log.Information("ScriptWorker[{Shard}] started, pid {Pid}", Label, connection.Pid);
else
_log.Information("ScriptWorker[{Shard}] started, pid {Pid}, after {Attempts} failed start attempt(s) over {Outage}",
Label, connection.Pid, failedBefore, Stopwatch.GetElapsedTime(failingSince));
var started = connection;
_ = started.Exited.ContinueWith(t => OnExited(started, t.Result), CancellationToken.None,
TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnRanToCompletion, TaskScheduler.Default);
@@ -320,13 +347,31 @@ internal sealed class WorkerShard : IAsyncDisposable
}
catch (Exception ex)
{
_log.Error(ex, "ScriptWorker[{Shard}] failed to start; scripts on it report runner unavailable until it does", Label);
int attempt;
TimeSpan failingFor;
bool warn;
lock (_gate)
{
_starting = null;
MarkDownLocked();
ScheduleBackoffLocked();
attempt = ++_startFailures;
var now = Stopwatch.GetTimestamp();
if (attempt == 1) _startFailingSince = now;
failingFor = Stopwatch.GetElapsedTime(_startFailingSince, now);
warn = attempt > 1 && (_lastStartFailureWarning == 0
|| Stopwatch.GetElapsedTime(_lastStartFailureWarning, now) >= _startFailureWarningInterval);
if (attempt == 1 || warn) _lastStartFailureWarning = now;
}
// lmxopcua#788: one Error with the stack per down period; the retries (every backoff, capped at 30 s, per
// shard) are a rate-limited Warning without it, and the rest Debug. Recovery is one Information line above.
if (attempt == 1)
_log.Error(ex, "ScriptWorker[{Shard}] failed to start; scripts on it report runner unavailable until it does (retrying with backoff)", Label);
else if (warn)
_log.Warning("ScriptWorker[{Shard}] still failing to start: {Attempts} failed attempt(s) over {Outage}; last error: {Error}",
Label, attempt, failingFor, ex.Message);
else
_log.Debug("ScriptWorker[{Shard}] start attempt {Attempts} failed: {Error}", Label, attempt, ex.Message);
if (connection is not null) await connection.DisposeAsync().ConfigureAwait(false);
return null;
}
src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto
+9
View File
@@ -26,6 +26,7 @@ message Envelope {
DiagnosticsRequest diag_request = 21; // host -> worker (tests / ops)
DiagnosticsReply diag_reply = 22; // worker -> host
Cancel cancel = 23; // host -> worker, at the soft deadline
LoadStarted load_started = 24; // worker -> host, when a compile thread picks a load up (lmxopcua#790)
}
}
@@ -39,6 +40,7 @@ message Hello {
uint32 max_log_lines = 7; // outbound log buffer cap
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)))
}
message Ready {
@@ -81,6 +83,13 @@ enum PeCache {
PE_CACHE_REJECTED = 3; // an entry failed verification, was deleted, and the source was compiled
}
// A compile thread has taken the load off the worker's queue and is about to compile (or PE-load) it. The host's load
// timeout runs from here, not from when the host sent the load: a load queued behind other compiles is not stuck and
// must not be blamed for the queue (lmxopcua#790).
message LoadStarted {
uint64 load_id = 1;
}
message LoadResult {
uint64 load_id = 1;
repeated LoadOutcome outcomes = 2;
src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFrameCodec.cs
+100 -18
View File
@@ -62,30 +62,14 @@ public static class ScriptWorkerFrameCodec
if (got < 4) throw new ScriptWorkerProtocolException("truncated frame header");
var size = BinaryPrimitives.ReadInt32LittleEndian(header);
if (size == 0) throw new ScriptWorkerProtocolException("zero-length frame");
if (size < 0 || size > maxFrameBytes)
throw new ScriptWorkerProtocolException($"frame length {size} is outside 1..{maxFrameBytes}");
CheckLength(size, maxFrameBytes);
var body = ArrayPool<byte>.Shared.Rent(size);
try
{
var read = await ReadFullyAsync(stream, body.AsMemory(0, size), ct).ConfigureAwait(false);
if (read < size) throw new ScriptWorkerProtocolException($"truncated frame: {read} of {size} bytes");
Envelope envelope;
try
{
envelope = Envelope.Parser.ParseFrom(body, 0, size);
}
catch (InvalidProtocolBufferException ex)
{
throw new ScriptWorkerProtocolException($"unparseable frame: {ex.Message}", ex);
}
if (envelope.ProtocolVersion != ProtocolVersion)
throw new ScriptWorkerProtocolException(
$"protocol version {envelope.ProtocolVersion} is not the supported {ProtocolVersion}");
if (envelope.BodyCase == Envelope.BodyOneofCase.None)
throw new ScriptWorkerProtocolException("frame carries no body");
return envelope;
return ParseBody(body.AsSpan(0, size));
}
finally
{
@@ -93,6 +77,104 @@ public static class ScriptWorkerFrameCodec
}
}
/// <summary>
/// Whether this platform's pipe should be driven with blocking I/O (lmxopcua#789). On Unix a .NET pipe is a
/// socket, and ONE asynchronous operation on it switches it to non-blocking for good, after which even a
/// "blocking" read goes through the socket engine (an epoll-thread hand-off and a spinning wait per frame) —
/// measured as most of the host's per-evaluation CPU. Kept blocking, a read is a plain kernel wait on a
/// dedicated thread. Not on Windows, where an overlapped pipe handle is needed for full duplex and its
/// asynchronous path is the efficient one.
/// </summary>
public static bool UseBlockingIo => !OperatingSystem.IsWindows();
/// <summary>Writes one frame synchronously (the blocking-I/O handshake). Not thread-safe.</summary>
/// <param name="stream">The pipe.</param>
/// <param name="envelope">The envelope.</param>
/// <param name="maxFrameBytes">The cap.</param>
public static void Write(Stream stream, Envelope envelope, int maxFrameBytes)
{
ArgumentNullException.ThrowIfNull(stream);
ArgumentNullException.ThrowIfNull(envelope);
var size = envelope.CalculateSize();
if (size == 0) throw new ScriptWorkerProtocolException("refusing to write an empty frame");
if (size > maxFrameBytes)
throw new ScriptWorkerProtocolException($"frame of {size} bytes exceeds the {maxFrameBytes}-byte cap");
var buffer = ArrayPool<byte>.Shared.Rent(size + 4);
try
{
BinaryPrimitives.WriteInt32LittleEndian(buffer, size);
envelope.WriteTo(new Span<byte>(buffer, 4, size));
stream.Write(buffer, 0, size + 4);
}
finally
{
ArrayPool<byte>.Shared.Return(buffer);
}
}
/// <summary>
/// Reads exactly one frame synchronously, never past it (the blocking-I/O handshake: what follows belongs to the
/// connection's own reader).
/// </summary>
/// <param name="stream">The pipe.</param>
/// <param name="maxFrameBytes">The cap.</param>
/// <returns>The envelope, or null on a clean end of stream at a frame boundary.</returns>
public static Envelope? Read(Stream stream, int maxFrameBytes)
{
ArgumentNullException.ThrowIfNull(stream);
Span<byte> header = stackalloc byte[4];
var got = stream.ReadAtLeast(header, 4, throwOnEndOfStream: false);
if (got == 0) return null;
if (got < 4) throw new ScriptWorkerProtocolException("truncated frame header");
var size = BinaryPrimitives.ReadInt32LittleEndian(header);
CheckLength(size, maxFrameBytes);
var body = ArrayPool<byte>.Shared.Rent(size);
try
{
var read = stream.ReadAtLeast(body.AsSpan(0, size), size, throwOnEndOfStream: false);
if (read < size) throw new ScriptWorkerProtocolException($"truncated frame: {read} of {size} bytes");
return ParseBody(body.AsSpan(0, size));
}
finally
{
ArrayPool<byte>.Shared.Return(body);
}
}
/// <summary>Validates a frame's length prefix against the cap.</summary>
/// <param name="size">The decoded length.</param>
/// <param name="maxFrameBytes">The negotiated cap.</param>
/// <exception cref="ScriptWorkerProtocolException">A zero, negative or oversize length.</exception>
internal static void CheckLength(int size, int maxFrameBytes)
{
if (size == 0) throw new ScriptWorkerProtocolException("zero-length frame");
if (size < 0 || size > maxFrameBytes)
throw new ScriptWorkerProtocolException($"frame length {size} is outside 1..{maxFrameBytes}");
}
/// <summary>Parses and validates one frame body (everything after the length prefix).</summary>
/// <param name="body">The body bytes.</param>
/// <returns>The envelope.</returns>
/// <exception cref="ScriptWorkerProtocolException">Unparseable, the wrong protocol version, or no body.</exception>
internal static Envelope ParseBody(ReadOnlySpan<byte> body)
{
Envelope envelope;
try
{
envelope = Envelope.Parser.ParseFrom(body);
}
catch (InvalidProtocolBufferException ex)
{
throw new ScriptWorkerProtocolException($"unparseable frame: {ex.Message}", ex);
}
if (envelope.ProtocolVersion != ProtocolVersion)
throw new ScriptWorkerProtocolException(
$"protocol version {envelope.ProtocolVersion} is not the supported {ProtocolVersion}");
if (envelope.BodyCase == Envelope.BodyOneofCase.None)
throw new ScriptWorkerProtocolException("frame carries no body");
return envelope;
}
private static async ValueTask<int> ReadFullyAsync(Stream stream, Memory<byte> buffer, CancellationToken ct)
{
var total = 0;
src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs
+262
View File
@@ -0,0 +1,262 @@
using System.Buffers;
using System.Buffers.Binary;
using System.Collections.Concurrent;
using Google.Protobuf;
using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1;
namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
/// <summary>
/// A buffered reader of <see cref="ScriptWorkerFrameCodec"/> frames (lmxopcua#789). One read from the pipe pulls in
/// every frame the peer has written so far, and they are then parsed from the buffer without another read, so a
/// batch of results costs one wake-up instead of two reads per frame. Same validation as
/// <see cref="ScriptWorkerFrameCodec.ReadAsync"/>: zero-length, oversize, truncated, unparseable, wrong-version and
/// bodiless frames are a <see cref="ScriptWorkerProtocolException"/>. Not thread-safe: one reader per stream.
/// </summary>
public sealed class ScriptWorkerFrameReader
{
private const int InitialBuffer = 64 * 1024;
private readonly Stream _stream;
private byte[] _buffer = new byte[InitialBuffer];
private int _start;
private int _end;
/// <summary>Initializes the reader.</summary>
/// <param name="stream">The pipe.</param>
/// <param name="maxFrameBytes">The frame cap.</param>
public ScriptWorkerFrameReader(Stream stream, int maxFrameBytes = ScriptWorkerFrameCodec.DefaultMaxFrameBytes)
{
ArgumentNullException.ThrowIfNull(stream);
_stream = stream;
MaxFrameBytes = maxFrameBytes;
}
/// <summary>The negotiated frame cap.</summary>
public int MaxFrameBytes { get; set; }
/// <summary>Reads the next frame.</summary>
/// <param name="ct">Cancellation.</param>
/// <returns>The envelope, or null on a clean end of stream at a frame boundary.</returns>
public async ValueTask<Envelope?> ReadAsync(CancellationToken ct = default)
{
while (true)
{
if (TryTake(out var envelope, out var need)) return envelope;
Prepare(need);
var n = await _stream.ReadAsync(_buffer.AsMemory(_end), ct).ConfigureAwait(false);
if (n == 0) return EndOfStream();
_end += n;
}
}
/// <summary>Reads the next frame, blocking (for a dedicated reader thread).</summary>
/// <returns>The envelope, or null on a clean end of stream at a frame boundary.</returns>
public Envelope? Read()
{
while (true)
{
if (TryTake(out var envelope, out var need)) return envelope;
Prepare(need);
var n = _stream.Read(_buffer, _end, _buffer.Length - _end);
if (n == 0) return EndOfStream();
_end += n;
}
}
private bool TryTake(out Envelope? envelope, out int need)
{
envelope = null;
var available = _end - _start;
if (available < 4)
{
need = 4;
return false;
}
var size = BinaryPrimitives.ReadInt32LittleEndian(_buffer.AsSpan(_start, 4));
ScriptWorkerFrameCodec.CheckLength(size, MaxFrameBytes);
need = size + 4;
if (available < need) return false;
envelope = ScriptWorkerFrameCodec.ParseBody(_buffer.AsSpan(_start + 4, size));
_start += need;
if (_start == _end) _start = _end = 0;
return true;
}
/// <summary>Makes room for a frame of <paramref name="need"/> bytes (prefix included) starting at <c>_start</c>.</summary>
private void Prepare(int need)
{
if (_start + need <= _buffer.Length && _end < _buffer.Length) return;
var pending = _end - _start;
if (need > _buffer.Length)
{
var bigger = new byte[Math.Max(need, _buffer.Length * 2)];
Buffer.BlockCopy(_buffer, _start, bigger, 0, pending);
_buffer = bigger;
}
else if (_start > 0)
{
Buffer.BlockCopy(_buffer, _start, _buffer, 0, pending);
}
_start = 0;
_end = pending;
}
private Envelope? EndOfStream()
{
var pending = _end - _start;
if (pending == 0) return null;
if (pending < 4) throw new ScriptWorkerProtocolException("truncated frame header");
var size = BinaryPrimitives.ReadInt32LittleEndian(_buffer.AsSpan(_start, 4));
throw new ScriptWorkerProtocolException($"truncated frame: {pending - 4} of {size} bytes");
}
}
/// <summary>
/// A combining frame writer (lmxopcua#789): the thread that queues a frame writes it itself when no write is in
/// progress, so the common case needs no hand-off to a writer task (no thread-pool wake-up, no spin-wait), and
/// frames queued while a write is in progress are written by that writer in the same pass. Within a pass consecutive
/// <see cref="EvalBatch"/> items are merged into one batch and consecutive <see cref="EvalResults"/> items into one
/// (design §4.5 opportunistic batching), every frame is serialised into one buffer, and the buffer goes out in one
/// write. Frame order is preserved.
/// </summary>
/// <remarks>
/// <para>Asynchronous mode: a write that cannot complete at once (a full pipe) continues on the I/O completion, and
/// frames queued meanwhile wait for it. Synchronous mode (<see cref="ScriptWorkerFrameCodec.UseBlockingIo"/>, so
/// the socket under a Unix pipe never turns non-blocking): the draining thread blocks while the pipe is full. Only
/// one thread drains at a time — every other producer just queues and returns — and a peer that stops reading is
/// killed by its supervisor (the host's heartbeat check), which fails the write and releases that thread.</para>
/// <para>A failed write completes the writer and reports the exception once through <c>onFault</c>.</para>
/// </remarks>
public sealed class ScriptWorkerFrameWriter
{
private readonly Stream _stream;
private readonly Action<Exception> _onFault;
private readonly ConcurrentQueue<Envelope> _queue = new();
private readonly ArrayBufferWriter<byte> _pass = new(16 * 1024);
private readonly bool _synchronous;
private int _draining;
private int _completed;
private int _faulted;
private long _seq;
/// <summary>Initializes the writer.</summary>
/// <param name="stream">The pipe.</param>
/// <param name="onFault">Called once when a write fails (or a frame is refused); the writer is then completed.</param>
/// <param name="maxFrameBytes">The frame cap.</param>
/// <param name="firstSeq">The last sequence number already used (the handshake's).</param>
/// <param name="synchronous">Write with blocking calls (see the remarks).</param>
public ScriptWorkerFrameWriter(Stream stream, Action<Exception> onFault,
int maxFrameBytes = ScriptWorkerFrameCodec.DefaultMaxFrameBytes, long firstSeq = 0, bool synchronous = false)
{
_synchronous = synchronous;
ArgumentNullException.ThrowIfNull(stream);
ArgumentNullException.ThrowIfNull(onFault);
_stream = stream;
_onFault = onFault;
MaxFrameBytes = maxFrameBytes;
_seq = firstSeq;
}
/// <summary>The negotiated frame cap.</summary>
public int MaxFrameBytes { get; set; }
/// <summary>Whether <see cref="Complete"/> has been called or a write failed.</summary>
public bool IsCompleted => Volatile.Read(ref _completed) != 0;
/// <summary>
/// Queues a frame and, when no write is in progress, writes it on the calling thread. Returns false once the
/// writer is completed; the frame is then dropped.
/// </summary>
/// <param name="envelope">The frame (its version and sequence number are set here).</param>
/// <returns>Whether the frame was accepted.</returns>
public bool TryWrite(Envelope envelope)
{
ArgumentNullException.ThrowIfNull(envelope);
// A full fence before the completion check, so a caller that registered state (a pending evaluation) before
// calling here and a completer that completes and then sweeps that state cannot both miss each other.
Interlocked.MemoryBarrier();
if (Volatile.Read(ref _completed) != 0) return false;
envelope.ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion;
_queue.Enqueue(envelope);
if (Interlocked.CompareExchange(ref _draining, 1, 0) == 0) _ = DrainAsync();
return true;
}
/// <summary>Stops accepting frames. Frames already queued are still written by a pass in progress.</summary>
public void Complete() => Interlocked.Exchange(ref _completed, 1);
private async Task DrainAsync()
{
try
{
while (true)
{
FillPass();
if (_pass.WrittenCount > 0)
{
if (_synchronous) _stream.Write(_pass.WrittenSpan);
else await _stream.WriteAsync(_pass.WrittenMemory).ConfigureAwait(false);
_pass.ResetWrittenCount();
// Synchronous on purpose: a pipe has no write buffer to flush, and PipeStream does not override
// FlushAsync, so the base Stream.FlushAsync would queue a thread-pool work item per pass.
_stream.Flush();
}
Volatile.Write(ref _draining, 0);
// A frame queued after the last TryDequeue but before the flag dropped saw _draining == 1 and left it
// to us: take the flag back and go round again, unless another thread already has.
if (_queue.IsEmpty || Interlocked.CompareExchange(ref _draining, 1, 0) != 0) return;
}
}
catch (Exception ex)
{
Interlocked.Exchange(ref _completed, 1);
_pass.ResetWrittenCount();
while (_queue.TryDequeue(out _)) { }
Volatile.Write(ref _draining, 0);
if (Interlocked.Exchange(ref _faulted, 1) == 0) _onFault(ex);
}
}
private void FillPass()
{
EvalBatch? evals = null;
EvalResults? results = null;
while (_queue.TryDequeue(out var envelope))
{
switch (envelope.BodyCase)
{
case Envelope.BodyOneofCase.Eval:
if (results is not null) { Append(new Envelope { Results = results }); results = null; }
(evals ??= new EvalBatch()).Items.AddRange(envelope.Eval.Items);
break;
case Envelope.BodyOneofCase.Results:
if (evals is not null) { Append(new Envelope { Eval = evals }); evals = null; }
(results ??= new EvalResults()).Items.AddRange(envelope.Results.Items);
break;
default:
if (evals is not null) { Append(new Envelope { Eval = evals }); evals = null; }
if (results is not null) { Append(new Envelope { Results = results }); results = null; }
Append(envelope);
break;
}
}
if (evals is not null) Append(new Envelope { Eval = evals });
if (results is not null) Append(new Envelope { Results = results });
}
private void Append(Envelope envelope)
{
envelope.ProtocolVersion = ScriptWorkerFrameCodec.ProtocolVersion;
envelope.Seq = (ulong)Interlocked.Increment(ref _seq);
var size = envelope.CalculateSize();
if (size == 0) throw new ScriptWorkerProtocolException("refusing to write an empty frame");
if (size > MaxFrameBytes)
throw new ScriptWorkerProtocolException($"frame of {size} bytes exceeds the {MaxFrameBytes}-byte cap");
var span = _pass.GetSpan(size + 4);
BinaryPrimitives.WriteInt32LittleEndian(span, size);
envelope.WriteTo(span.Slice(4, size));
_pass.Advance(size + 4);
}
}
src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/Program.cs
+3 -1
View File
@@ -22,7 +22,9 @@ internal static class Program
PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
try
{
await pipe.ConnectAsync(TimeSpan.FromSeconds(10), CancellationToken.None);
// Synchronous on Unix so the socket under the pipe stays blocking (ScriptWorkerFrameCodec.UseBlockingIo).
if (ScriptWorkerFrameCodec.UseBlockingIo) pipe.Connect(TimeSpan.FromSeconds(10));
else await pipe.ConnectAsync(TimeSpan.FromSeconds(10), CancellationToken.None);
}
catch (Exception ex)
{
src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs
+109 -80
View File
@@ -2,7 +2,6 @@ using System.Collections;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.IO.Pipes;
using System.Threading.Channels;
using Serilog;
using Serilog.Core;
using Serilog.Events;
@@ -33,8 +32,6 @@ internal sealed class WorkerHost
private readonly NamedPipeClientStream _pipe;
private readonly string _nonce;
private readonly int _parentPid;
private readonly Channel<Envelope> _outbound = Channel.CreateUnbounded<Envelope>(
new UnboundedChannelOptions { SingleReader = true });
private readonly ConcurrentDictionary<string, LoadedScript> _scripts = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<long, RunningEval> _running = new();
private readonly BlockingCollection<EvalRequest> _queue = new(new ConcurrentQueue<EvalRequest>());
@@ -47,8 +44,16 @@ internal sealed class WorkerHost
/// </summary>
private readonly BlockingCollection<LoadScripts> _loads = new(new ConcurrentQueue<LoadScripts>());
/// <summary>Dedicated compile threads: half the processors, at least 1, at most 8.</summary>
internal static int CompileThreads => Math.Clamp(Environment.ProcessorCount / 2, 1, 8);
/// <summary>Default dedicated compile threads: half the processors, at least 1, at most 8.</summary>
internal static int DefaultCompileThreads => Math.Clamp(Environment.ProcessorCount / 2, 1, 8);
/// <summary>
/// Outbound frames (lmxopcua#789): the thread that produces a frame — an evaluation thread with its result, a
/// compile thread with its load result — writes it itself when no write is in progress, instead of handing it to
/// a writer task through a channel (one thread-pool wake-up and spin per result).
/// </summary>
private ScriptWorkerFrameWriter? _writer;
private int _compileThreads = DefaultCompileThreads;
private readonly ForwardingLogSink _logSink;
private readonly ILogger _scriptLogger;
private int _maxFrameBytes = ScriptWorkerFrameCodec.DefaultMaxFrameBytes;
@@ -71,7 +76,11 @@ internal sealed class WorkerHost
{
try
{
var hello = await ScriptWorkerFrameCodec.ReadAsync(_pipe, ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
// Blocking I/O on Unix from the first byte (lmxopcua#789): one async call would turn the socket under the
// pipe non-blocking for good, and every later read would go through the socket engine.
var hello = ScriptWorkerFrameCodec.UseBlockingIo
? ScriptWorkerFrameCodec.Read(_pipe, ScriptWorkerFrameCodec.DefaultMaxFrameBytes)
: await ScriptWorkerFrameCodec.ReadAsync(_pipe, ScriptWorkerFrameCodec.DefaultMaxFrameBytes);
if (hello is null) return ScriptWorkerEnvironment.ExitPipeClosed;
if (hello.BodyCase != Envelope.BodyOneofCase.Hello)
throw new ScriptWorkerProtocolException($"expected Hello, got {hello.BodyCase}");
@@ -79,7 +88,7 @@ internal sealed class WorkerHost
// The nonce is echoed, not trusted: the host compares it with the one it put in OUR environment, which
// proves the process on the other end of its pipe is the one it launched.
await ScriptWorkerFrameCodec.WriteAsync(_pipe, Wrap(new Envelope
var ready = Wrap(new Envelope
{
Ready = new Ready
{
@@ -87,14 +96,16 @@ internal sealed class WorkerHost
Pid = Environment.ProcessId,
RuntimeVersion = Environment.Version.ToString(),
},
}), _maxFrameBytes);
});
if (ScriptWorkerFrameCodec.UseBlockingIo) ScriptWorkerFrameCodec.Write(_pipe, ready, _maxFrameBytes);
else await ScriptWorkerFrameCodec.WriteAsync(_pipe, ready, _maxFrameBytes);
_logSink.Capacity = (int)Math.Max(1, hello.Hello.MaxLogLines == 0 ? 1000 : hello.Hello.MaxLogLines);
_writer = new ScriptWorkerFrameWriter(_pipe, OnWriteFault, _maxFrameBytes, firstSeq: Interlocked.Read(ref _seq),
synchronous: ScriptWorkerFrameCodec.UseBlockingIo);
StartBackground(Math.Max(1, (int)hello.Hello.EvalThreads));
var writer = Task.Run(WriteLoopAsync);
var exit = await ReadLoopAsync();
_outbound.Writer.TryComplete();
try { await writer.WaitAsync(TimeSpan.FromSeconds(1)); } catch { /* exiting regardless */ }
var exit = ScriptWorkerFrameCodec.UseBlockingIo ? await ReadOnDedicatedThread() : await ReadLoopAsync();
_writer.Complete();
return exit;
}
catch (ScriptWorkerProtocolException ex)
@@ -114,6 +125,7 @@ internal sealed class WorkerHost
if (hello.QueueCapacity > 0) _queueCapacity = (int)hello.QueueCapacity;
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);
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.
@@ -127,7 +139,7 @@ internal sealed class WorkerHost
{
new Thread(EvalThreadLoop) { IsBackground = true, Name = $"ScriptWorker.Eval{i}" }.Start();
}
for (var i = 0; i < CompileThreads; i++)
for (var i = 0; i < _compileThreads; i++)
{
new Thread(CompileThreadLoop) { IsBackground = true, Name = $"ScriptWorker.Compile{i}" }.Start();
}
@@ -137,14 +149,50 @@ internal sealed class WorkerHost
if (_parentPid > 0) _ = Task.Run(ParentWatchLoopAsync);
}
/// <summary>The blocking read loop on its own thread (Unix, lmxopcua#789); completes with the exit code.</summary>
private Task<int> ReadOnDedicatedThread()
{
var done = new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
new Thread(() =>
{
var reader = new ScriptWorkerFrameReader(_pipe, _maxFrameBytes);
while (true)
{
Envelope? frame;
try
{
frame = reader.Read();
}
catch (ScriptWorkerProtocolException ex)
{
Console.Error.WriteLine($"ScriptWorker: protocol fault: {ex.Message}");
done.TrySetResult(ScriptWorkerEnvironment.ExitProtocolFault);
return;
}
catch (Exception ex) when (ex is IOException or ObjectDisposedException or System.Net.Sockets.SocketException)
{
done.TrySetResult(ScriptWorkerEnvironment.ExitPipeClosed);
return;
}
if (HandleFrame(frame) is { } exit)
{
done.TrySetResult(exit);
return;
}
}
}) { IsBackground = true, Name = "ScriptWorker.Reader" }.Start();
return done.Task;
}
private async Task<int> ReadLoopAsync()
{
var reader = new ScriptWorkerFrameReader(_pipe, _maxFrameBytes);
while (true)
{
Envelope? frame;
try
{
frame = await ScriptWorkerFrameCodec.ReadAsync(_pipe, _maxFrameBytes);
frame = await reader.ReadAsync();
}
catch (ScriptWorkerProtocolException ex)
{
@@ -155,42 +203,47 @@ internal sealed class WorkerHost
{
return ScriptWorkerEnvironment.ExitPipeClosed;
}
if (frame is null) return ScriptWorkerEnvironment.ExitPipeClosed; // host gone: fatal by design
if (HandleFrame(frame) is { } exit) return exit;
}
}
switch (frame.BodyCase)
{
case Envelope.BodyOneofCase.Load:
/// <summary>Acts on one frame from the host.</summary>
/// <param name="frame">The frame; null = end of stream.</param>
/// <returns>The exit code when the worker must end, otherwise null.</returns>
private int? HandleFrame(Envelope? frame)
{
if (frame is null) return ScriptWorkerEnvironment.ExitPipeClosed; // host gone: fatal by design
switch (frame.BodyCase)
{
case Envelope.BodyOneofCase.Load:
// Compile off the reader, so evaluations for already-loaded keys keep flowing during a load, and off
// the thread pool, so the heartbeat and the writer keep flowing during a burst of them.
_loads.Add(frame.Load);
return null;
case Envelope.BodyOneofCase.Unload:
foreach (var key in frame.Unload.Keys)
if (_scripts.TryRemove(key, out var gone)) gone.Dispose();
return null;
case Envelope.BodyOneofCase.Eval:
foreach (var item in frame.Eval.Items) Admit(item);
return null;
case Envelope.BodyOneofCase.Cancel:
if (_running.TryGetValue((long)frame.Cancel.EvalId, out var toCancel))
{
// Compile off the reader, so evaluations for already-loaded keys keep flowing during a load, and
// off the thread pool, so the heartbeat and the writer keep flowing during a burst of them.
_loads.Add(frame.Load);
break;
try { toCancel.Cancellation.Cancel(); } catch (ObjectDisposedException) { /* just finished */ }
}
case Envelope.BodyOneofCase.Unload:
foreach (var key in frame.Unload.Keys)
if (_scripts.TryRemove(key, out var gone)) gone.Dispose();
break;
case Envelope.BodyOneofCase.Eval:
foreach (var item in frame.Eval.Items) Admit(item);
break;
case Envelope.BodyOneofCase.Cancel:
if (_running.TryGetValue((long)frame.Cancel.EvalId, out var toCancel))
{
try { toCancel.Cancellation.Cancel(); } catch (ObjectDisposedException) { /* just finished */ }
}
break;
case Envelope.BodyOneofCase.DiagRequest:
Send(new Envelope { DiagReply = Diagnostics() });
break;
case Envelope.BodyOneofCase.Shutdown:
return 0;
case Envelope.BodyOneofCase.Fault:
await Console.Error.WriteLineAsync($"ScriptWorker: host fault: {frame.Fault.Message}");
return ScriptWorkerEnvironment.ExitProtocolFault;
default:
await Console.Error.WriteLineAsync($"ScriptWorker: unexpected frame {frame.BodyCase}");
return ScriptWorkerEnvironment.ExitProtocolFault;
}
return null;
case Envelope.BodyOneofCase.DiagRequest:
Send(new Envelope { DiagReply = Diagnostics() });
return null;
case Envelope.BodyOneofCase.Shutdown:
return 0;
case Envelope.BodyOneofCase.Fault:
Console.Error.WriteLine($"ScriptWorker: host fault: {frame.Fault.Message}");
return ScriptWorkerEnvironment.ExitProtocolFault;
default:
Console.Error.WriteLine($"ScriptWorker: unexpected frame {frame.BodyCase}");
return ScriptWorkerEnvironment.ExitProtocolFault;
}
}
@@ -201,6 +254,8 @@ internal sealed class WorkerHost
private void Load(LoadScripts load)
{
// The host's compile timeout runs from here, not from when it sent the load (lmxopcua#790).
Send(new Envelope { LoadStarted = new LoadStarted { LoadId = load.LoadId } });
var result = new LoadResult { LoadId = load.LoadId };
foreach (var script in load.Scripts)
{
@@ -423,41 +478,15 @@ internal sealed class WorkerHost
}
}
private async Task WriteLoopAsync()
private static void OnWriteFault(Exception ex)
{
var reader = _outbound.Reader;
var pendingResults = new EvalResults();
try
// The host is gone (or a frame broke the protocol): fatal by design.
if (ex is ScriptWorkerProtocolException)
{
while (await reader.WaitToReadAsync())
{
// Opportunistic coalescing (design §4.5): drain whatever is queued right now, merging every completed
// evaluation into one EvalResults frame; no timer, so an idle pipe adds no latency.
while (reader.TryRead(out var envelope))
{
if (envelope.BodyCase == Envelope.BodyOneofCase.Results)
{
pendingResults.Items.AddRange(envelope.Results.Items);
continue;
}
await ScriptWorkerFrameCodec.WriteAsync(_pipe, envelope, _maxFrameBytes);
}
if (pendingResults.Items.Count > 0)
{
await ScriptWorkerFrameCodec.WriteAsync(_pipe, Wrap(new Envelope { Results = pendingResults }), _maxFrameBytes);
pendingResults = new EvalResults();
}
await _pipe.FlushAsync();
}
}
catch (IOException)
{
Environment.Exit(ScriptWorkerEnvironment.ExitPipeClosed);
}
catch (ObjectDisposedException)
{
Environment.Exit(ScriptWorkerEnvironment.ExitPipeClosed);
Console.Error.WriteLine($"ScriptWorker: protocol fault writing a frame: {ex.Message}");
Environment.Exit(ScriptWorkerEnvironment.ExitProtocolFault);
}
Environment.Exit(ScriptWorkerEnvironment.ExitPipeClosed);
}
private void SendResult(EvalResult result)
@@ -467,7 +496,7 @@ internal sealed class WorkerHost
Send(new Envelope { Results = results });
}
private void Send(Envelope envelope) => _outbound.Writer.TryWrite(Wrap(envelope));
private void Send(Envelope envelope) => _writer?.TryWrite(envelope);
private Envelope Wrap(Envelope envelope)
{
tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ProtocolTests.cs
+151
View File
@@ -173,6 +173,157 @@ public sealed class ProtocolTests
}
}
// ---------------------------------------------------------------- buffered reader / combining writer (lmxopcua#789)
/// <summary>Hands out at most <c>chunk</c> bytes per read, so frames straddle reads at every possible offset.</summary>
private sealed class ChunkedStream(byte[] data, int chunk) : MemoryStream(data)
{
public override int Read(byte[] buffer, int offset, int count) => base.Read(buffer, offset, Math.Min(count, chunk));
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
=> base.ReadAsync(buffer[..Math.Min(buffer.Length, chunk)], cancellationToken);
}
private static byte[] Frames(int count, int bigEvery = 0)
{
using var stream = new MemoryStream();
for (var i = 0; i < count; i++)
{
var env = new Envelope { ProtocolVersion = 1, Seq = (ulong)i + 1, Heartbeat = new Heartbeat { Inflight = (uint)i } };
if (bigEvery > 0 && i % bigEvery == 0)
env = new Envelope { ProtocolVersion = 1, Seq = (ulong)i + 1, Fault = new Fault { Message = new string('x', 100_000 + i) } };
ScriptWorkerFrameCodec.WriteAsync(stream, env, ScriptWorkerFrameCodec.DefaultMaxFrameBytes).AsTask().GetAwaiter().GetResult();
}
return stream.ToArray();
}
/// <summary>
/// The buffered reader yields exactly the frames the per-frame codec wrote, in order, however the bytes are
/// split across reads — including frames larger than its initial 64 KiB buffer — through both its async and its
/// blocking read, and ends cleanly at a frame boundary.
/// </summary>
[Theory]
[InlineData(1, false)]
[InlineData(3, true)]
[InlineData(7, false)]
[InlineData(65_536, true)]
[InlineData(1_000_000, false)]
public async Task The_buffered_reader_reassembles_frames_split_at_any_offset(int chunk, bool blocking)
{
var bytes = Frames(200, bigEvery: 37);
var reader = new ScriptWorkerFrameReader(new ChunkedStream(bytes, chunk));
for (var i = 0; i < 200; i++)
{
var env = (blocking ? reader.Read() : await reader.ReadAsync(TestContext.Current.CancellationToken)).ShouldNotBeNull($"frame {i}");
env.Seq.ShouldBe((ulong)i + 1);
if (i % 37 == 0) env.Fault.Message.Length.ShouldBe(100_000 + i);
else env.Heartbeat.Inflight.ShouldBe((uint)i);
}
(blocking ? reader.Read() : await reader.ReadAsync(TestContext.Current.CancellationToken)).ShouldBeNull("a clean end of stream");
}
/// <summary>The buffered reader rejects exactly what the codec rejects.</summary>
[Theory]
[MemberData(nameof(BadFrames))]
public async Task The_buffered_reader_rejects_malformed_frames_like_the_codec(string why, byte[] bytes)
{
await Should.ThrowAsync<ScriptWorkerProtocolException>(async () =>
await new ScriptWorkerFrameReader(new MemoryStream(bytes), 1024).ReadAsync(TestContext.Current.CancellationToken), why);
Should.Throw<ScriptWorkerProtocolException>(() => new ScriptWorkerFrameReader(new ChunkedStream(bytes, 1), 1024).Read(), why);
}
/// <summary>Random bytes never escape the buffered reader as anything but a protocol fault (or a clean end).</summary>
[Fact]
public void Fuzzed_input_only_ever_yields_a_protocol_fault_from_the_buffered_reader()
{
var random = new Random(789);
for (var i = 0; i < 3000; i++)
{
var bytes = new byte[random.Next(0, 64)];
random.NextBytes(bytes);
if (bytes.Length >= 4 && random.Next(2) == 0)
BinaryPrimitives.WriteInt32LittleEndian(bytes, random.Next(1, bytes.Length));
var reader = new ScriptWorkerFrameReader(new ChunkedStream(bytes, random.Next(1, 9)), 1024);
try { while (reader.Read() is not null) { } }
catch (ScriptWorkerProtocolException) { }
}
}
/// <summary>A stream whose first write waits on a gate, so frames queue up behind a write in progress.</summary>
private sealed class GatedStream : MemoryStream
{
public readonly TaskCompletionSource Gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
public int Writes;
public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
{
if (Interlocked.Increment(ref Writes) == 1) await Gate.Task;
await base.WriteAsync(buffer, cancellationToken);
}
}
private static List<Envelope> ReadAll(byte[] bytes)
{
var reader = new ScriptWorkerFrameReader(new MemoryStream(bytes));
var frames = new List<Envelope>();
while (reader.Read() is { } env) frames.Add(env);
return frames;
}
private static Envelope Eval(ulong id) => new() { Eval = new EvalBatch { Items = { new EvalRequest { EvalId = id, Key = "k" } } } };
/// <summary>
/// The combining writer (lmxopcua#789): frames queued while a write is in progress go out in the next pass, with
/// consecutive evaluations merged into one batch, every other frame kept in its place (a Cancel is never
/// reordered ahead of the evaluation it cancels), and sequence numbers increasing.
/// </summary>
[Fact]
public async Task The_combining_writer_batches_consecutive_evaluations_and_keeps_frame_order()
{
var stream = new GatedStream();
Exception? fault = null;
var writer = new ScriptWorkerFrameWriter(stream, ex => fault = ex, firstSeq: 1);
writer.TryWrite(Eval(1)).ShouldBeTrue(); // starts a pass that blocks on the gate
writer.TryWrite(Eval(2)).ShouldBeTrue();
writer.TryWrite(Eval(3)).ShouldBeTrue();
writer.TryWrite(new Envelope { Cancel = new Cancel { EvalId = 2 } }).ShouldBeTrue();
writer.TryWrite(Eval(4)).ShouldBeTrue();
stream.Gate.SetResult();
await Scripts.Eventually(() =>
{
try { return ReadAll(stream.ToArray()).Count == 4; }
catch (ScriptWorkerProtocolException) { return false; } // a pass caught mid-copy
}, TimeSpan.FromSeconds(5), "both passes written");
fault.ShouldBeNull();
var frames = ReadAll(stream.ToArray());
frames.Select(f => f.BodyCase).ShouldBe([Envelope.BodyOneofCase.Eval, Envelope.BodyOneofCase.Eval,
Envelope.BodyOneofCase.Cancel, Envelope.BodyOneofCase.Eval]);
frames[0].Eval.Items.Select(e => e.EvalId).ShouldBe([1UL]);
frames[1].Eval.Items.Select(e => e.EvalId).ShouldBe([2UL, 3UL], "queued behind the write in progress: one batch");
frames[2].Cancel.EvalId.ShouldBe(2UL);
frames[3].Eval.Items.Select(e => e.EvalId).ShouldBe([4UL]);
frames.Select(f => f.Seq).ShouldBe([2UL, 3UL, 4UL, 5UL]);
frames.ShouldAllBe(f => f.ProtocolVersion == ScriptWorkerFrameCodec.ProtocolVersion);
stream.Writes.ShouldBe(2, "one write per pass");
}
/// <summary>A failed write completes the writer, reports once, and later frames are refused rather than lost silently.</summary>
[Fact]
public void A_failed_write_completes_the_writer_and_reports_once()
{
var faults = 0;
var stream = new MemoryStream();
stream.Dispose();
var writer = new ScriptWorkerFrameWriter(stream, _ => Interlocked.Increment(ref faults));
writer.TryWrite(Eval(1)).ShouldBeTrue();
faults.ShouldBe(1);
writer.IsCompleted.ShouldBeTrue();
writer.TryWrite(Eval(2)).ShouldBeFalse();
faults.ShouldBe(1);
}
[Fact]
public void Backoff_doubles_from_250ms_to_a_30s_cap_with_20_percent_jitter()
{
tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ScriptRunnerPerfTests.cs
+39 -14
View File
@@ -191,8 +191,10 @@ public sealed class ScriptRunnerPerfTests
return new PhaseResult(issued / elapsed.TotalSeconds, Percentile(samples, 0.50), Percentile(samples, 0.99), cpu, ok, notOk, firstFailure);
}
private static string Row(string runner, string phase, PhaseResult r)
=> $"| {runner} | {phase} | {r.EvalsPerSecond,9:N0} | {r.P50Us,8:N0} | {r.P99Us,9:N0} | {r.HostCpuPercent,6:N0} % | {r.Ok:N0} / {r.NotOk:N0} |" +
private static string Row(string runner, string phase, PhaseResult r, PhaseResult? inProcess = null)
=> $"| {runner} | {phase} | {r.EvalsPerSecond,9:N0} | {r.P50Us,8:N0} | {r.P99Us,9:N0} | {r.HostCpuPercent,6:N0} % | " +
(inProcess is null ? "— |" : $"{r.HostCpuPercent - inProcess.HostCpuPercent,6:+0;-0;0} % |") +
$" {r.Ok:N0} / {r.NotOk:N0} |" +
(r.FirstFailure is null ? "" : $" first failure: {r.FirstFailure}");
/// <summary>
@@ -206,9 +208,12 @@ public sealed class ScriptRunnerPerfTests
var report = new StringBuilder()
.AppendLine($"### Script runner perf ({Env()})")
.AppendLine()
.AppendLine("| Runner | Phase | evals/s | p50 µs | p99 µs | host CPU | ok / not ok |")
.AppendLine("|---|---|---:|---:|---:|---:|---|");
.AppendLine("| Runner | Phase | evals/s | p50 µs | p99 µs | host CPU | vs InProcess | ok / not ok |")
.AppendLine("|---|---|---:|---:|---:|---:|---:|---|");
PhaseResult? k2Closed = null;
// The in-process run of each phase is the harness's own floor (its pacing loop, its continuations, the pool):
// the "vs InProcess" column is what the out-of-process runner adds on top — the §8.2 "host CPU for IPC".
var baseline = new Dictionary<string, PhaseResult>(StringComparer.Ordinal);
var configs = new (string Name, Func<IScriptRunner> Create)[]
{
@@ -223,11 +228,16 @@ public sealed class ScriptRunnerPerfTests
try
{
await WarmAsync(runner);
report.AppendLine(Row(name, "idle, sequential", await IdleLatencyAsync(runner)));
report.AppendLine(Row(name, "paced 5,000/s", await PacedAsync(runner, 5_000)));
report.AppendLine(Row(name, "paced 10,000/s", await PacedAsync(runner, 10_000)));
void Add(string phase, PhaseResult r)
{
if (name == "InProcess") baseline[phase] = r;
report.AppendLine(Row(name, phase, r, name == "InProcess" ? null : baseline.GetValueOrDefault(phase)));
}
Add("idle, sequential", await IdleLatencyAsync(runner));
Add("paced 5,000/s", await PacedAsync(runner, 5_000));
Add("paced 10,000/s", await PacedAsync(runner, 10_000));
var closed = await ClosedLoopAsync(runner, concurrency: 64);
report.AppendLine(Row(name, "closed loop ×64", closed));
Add("closed loop ×64", closed);
if (name == "OutOfProcess K=2") k2Closed = closed;
}
finally
@@ -242,7 +252,8 @@ public sealed class ScriptRunnerPerfTests
report.AppendLine()
.AppendLine("§8.2 budgets: added latency p50 ≤ 150 µs (idle), p99 ≤ 2 ms at 10k/s, ≥ 10,000 evals/s at K=2, " +
"host CPU ≤ 25 % of one core at 10k/s. Accepted gate (owner, §14): ≥ 5,000 evals/s/node.");
"host CPU for IPC (the \"vs InProcess\" column) ≤ 25 % of one core at 10k/s. " +
"Accepted gate (owner, §14): ≥ 5,000 evals/s/node.");
Report(report);
k2Closed.ShouldNotBeNull();
@@ -286,7 +297,7 @@ public sealed class ScriptRunnerPerfTests
var report = new StringBuilder().AppendLine($"### Script worker restart / RSS / kill latency ({Env()})").AppendLine();
TimeSpan cold, warmFresh, killRestart;
long rssIdle, rssLoaded;
long rssIdle, rssLoaded, rssWarmIdle, rssWarmLoaded;
await using (var runner = Scripts.Runner(PerfOptions(2, cacheDir), _runnerLog.CreateLogger()))
{
(await runner.RunAsync(Inv(0), Ct)).Status.ShouldBe(ScriptRunStatus.Ok);
@@ -296,8 +307,17 @@ public sealed class ScriptRunnerPerfTests
rssLoaded = runner.Shards.Max(s => s.Current is { } c ? Process.GetProcessById(c.Pid).WorkingSet64 : 0);
}
await using (var runner = Scripts.Runner(PerfOptions(2, cacheDir), _runnerLog.CreateLogger()))
{
// A worker that has only loaded verified PEs from the cache: Roslyn is never entered in it.
(await runner.RunAsync(Inv(0), Ct)).Status.ShouldBe(ScriptRunStatus.Ok);
await Task.Delay(500, Ct);
rssWarmIdle = runner.Shards.Max(s => s.Current is { } c ? Process.GetProcessById(c.Pid).WorkingSet64 : 0);
runner.PeCacheCounts.Misses.ShouldBe(0, "every load came from the cache");
}
await using (var runner = Scripts.Runner(PerfOptions(2, cacheDir), _runnerLog.CreateLogger()))
{
warmFresh = await AllFirstResults(runner);
rssWarmLoaded = runner.Shards.Max(s => s.Current is { } c ? Process.GetProcessById(c.Pid).WorkingSet64 : 0);
// A kill of both shard workers: time from the kill to every script's next correct result (includes the
// 250 ms first restart backoff and the cache load).
foreach (var shard in runner.Shards) Process.GetProcessById(shard.Current!.Pid).Kill();
@@ -316,11 +336,14 @@ public sealed class ScriptRunnerPerfTests
{
(await runner.RunAsync(Scripts.Vt("warm", "return 1;"), Ct)).Status.ShouldBe(ScriptRunStatus.Ok);
var worker = runner.Shards[0].Current!;
var sw = Stopwatch.StartNew();
// Timed from the soft-timeout answer (T after the submit), not from the call: the call also compiles
// the cold runaway key, which on the Windows VM alone is 100–200 ms (lmxopcua#789 — the first
// harness counted it as kill latency). Kill latency = exit − (answer + G).
var outcome = await runner.RunAsync(Scripts.Vt("runaway-" + i, "while (true) { }", TimeSpan.FromMilliseconds(200)), Ct);
var sw = Stopwatch.StartNew();
outcome.Status.ShouldBe(ScriptRunStatus.Failed);
await worker.Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
killLatencies.Add((sw.Elapsed - TimeSpan.FromMilliseconds(1200)).TotalMilliseconds);
killLatencies.Add((sw.Elapsed - TimeSpan.FromMilliseconds(1000)).TotalMilliseconds);
}
}
@@ -329,8 +352,10 @@ public sealed class ScriptRunnerPerfTests
.AppendLine($"| {scripts} scripts, empty cache, all first results (compile + write) | {cold.TotalMilliseconds:N0} ms | — |")
.AppendLine($"| {scripts} scripts, fresh runner on a warm cache, all first results | {warmFresh.TotalMilliseconds:N0} ms | ≤ 2,000 ms |")
.AppendLine($"| kill both shards → every script's next correct result | {killRestart.TotalMilliseconds:N0} ms | ≤ 2,000 ms |")
.AppendLine($"| worker RSS after first eval (max of 2 shards) | {rssIdle / 1048576.0:N0} MB | ≤ 80 MB + scripts |")
.AppendLine($"| worker RSS with {scripts / 2} scripts loaded per shard (max) | {rssLoaded / 1048576.0:N0} MB | — |")
.AppendLine($"| worker RSS after first eval, compiled (max of 2 shards) | {rssIdle / 1048576.0:N0} MB | ≤ 80 MB + scripts |")
.AppendLine($"| worker RSS with {scripts / 2} scripts compiled per shard (max) | {rssLoaded / 1048576.0:N0} MB | — |")
.AppendLine($"| worker RSS after first eval, warm PE cache, no compile (max) | {rssWarmIdle / 1048576.0:N0} MB | ≤ 80 MB + scripts |")
.AppendLine($"| worker RSS with {scripts / 2} scripts loaded from the cache per shard (max) | {rssWarmLoaded / 1048576.0:N0} MB | — |")
.AppendLine($"| kill latency after T+G (5 runs, ms) | {string.Join(" / ", killLatencies.Select(l => l.ToString("N0")))} | ≤ 100 ms |");
Report(report);
}
tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs
+213 -2
View File
@@ -132,8 +132,11 @@ public sealed class SupervisionTests
/// Single flight must not refuse a caller's NEXT run because of the run it already awaited: the slot is freed
/// before the awaiting caller resumes. The §8 perf harness found the slot freed by a continuation unordered
/// against the caller's await, so 32 of 200 back-to-back runs of one key came back "previous evaluation still
/// running" — which the VirtualTag and Calculation seams publish as Bad. Mutation check: drop the inline
/// release after the await and this goes red.
/// running" — which the VirtualTag and Calculation seams publish as Bad. Since lmxopcua#789 the release runs in
/// the connection, on the reader, before the caller's resumption is handed to the result dispatcher; the second
/// phase makes every release slow (the test hook sleeps inside it) so the ordering, not luck, is what keeps the
/// caller from being refused. Mutation check: release after handing off the resumption and the second phase goes
/// red.
/// </summary>
[Fact]
public async Task A_caller_that_awaited_its_run_is_never_refused_by_it()
@@ -147,6 +150,15 @@ public sealed class SupervisionTests
if (outcome.Status != ScriptRunStatus.Ok || !Equals(outcome.Value, i + 7)) refused.Add($"#{i}: {outcome.Reason}");
}
refused.ShouldBeEmpty();
runner.SlotReleasingForTest = _ => Thread.Sleep(5);
for (var i = 0; i < 50; i++)
{
var outcome = await runner.RunAsync(
Scripts.Vt("seq", "return (int)ctx.GetTag(\"x\").Value + 7;", inputs: Scripts.Input("x", i)), Ct);
if (outcome.Status != ScriptRunStatus.Ok || !Equals(outcome.Value, i + 7)) refused.Add($"slow release #{i}: {outcome.Reason}");
}
refused.ShouldBeEmpty();
}
/// <summary>
@@ -487,4 +499,203 @@ public sealed class SupervisionTests
status.State.ShouldBe(ScriptRunnerState.Ready);
status.Shards.Single(s => s.Label == "0").Pid.ShouldBe(runner.Shards[0].Current!.Pid);
}
// ---------------------------------------------------------------- lmxopcua#789: who resumes the caller
/// <summary>
/// lmxopcua#789: callers resume on the connection's result dispatcher, never on the pipe reader — so a consumer
/// that blocks after its await (the Calculation pump publishes to OPC UA subscribers there) cannot stop the reader
/// from taking heartbeats and get a healthy worker killed for silence. Here a 100 ms heartbeat (kill after 300 ms
/// of silence) and a continuation that blocks for 1.5 s. Mutation check: resume callers inline on the reader (a
/// synchronous-continuation TaskCompletionSource completed by the reader, no dispatcher) and the worker is killed
/// (KilledHeartbeat).
/// </summary>
[Fact]
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();
string? resumedOn = null;
await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct).ContinueWith(t =>
{
// Runs wherever RunAsync's own continuation completed it: the thread that resumed the caller.
resumedOn = Thread.CurrentThread.Name;
t.Result.Value.ShouldBe(7);
Thread.Sleep(1_500);
}, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
// Behaviour first: the worker was not killed for silence while the caller blocked.
(await Task.WhenAny(worker.Exited, Task.Delay(1_000, Ct))).ShouldNotBe(worker.Exited,
string.Join(Environment.NewLine, sink.Rendered.Where(l => l.Contains("ScriptWorker", StringComparison.Ordinal))));
var log = string.Join(Environment.NewLine, sink.Rendered.Where(l => l.Contains("ScriptWorker", StringComparison.Ordinal)));
resumedOn.ShouldBe("ScriptWorker[0].Results", log);
(await runner.RunAsync(Scripts.Vt("warm", Healthy), Ct)).Value.ShouldBe(7);
runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, log);
}
// ---------------------------------------------------------------- lmxopcua#788: start-failure logging
/// <summary>
/// lmxopcua#788: a shard that keeps failing to start logs ONE Error with the exception for the down period; its
/// retries are a Warning without a stack, carrying the attempt count, rate-limited (the rest at Debug); and the
/// start that ends the outage is one Information line naming how many attempts failed. The docker-dev drill logged
/// 28 Error+stack entries in two minutes on one node before this. Mutation check: log every failure at Error with
/// the exception again and the single-Error assertion goes red (3 Errors).
/// </summary>
[Theory]
[InlineData(0, 2, 0)]
[InlineData(3_600_000, 0, 2)]
public async Task A_shard_that_cannot_start_logs_one_error_then_concise_rate_limited_retries_then_its_recovery(
int warnEveryMs, int expectedWarnings, int expectedDebugs)
{
var sink = new CapturingSink();
var log = sink.CreateLogger();
var spec = new WorkerLaunchSpec("t", OutOfProcessScriptRunner.ResolveWorkerPath(new ScriptWorkerOptions { ExecutablePath = Scripts.WorkerUnderTest }),
256L * 1024 * 1024, EvalThreads: 2, HardGrace: TimeSpan.FromSeconds(5), HeartbeatInterval: TimeSpan.FromSeconds(1),
HandshakeTimeout: TimeSpan.FromSeconds(15));
var attempts = 0;
await using var shard = new WorkerShard("t", isolation: false,
(onExiting, ct) => Interlocked.Increment(ref attempts) <= 3
? throw new InvalidOperationException("the worker exited (code 1) before connecting to its pipe")
: WorkerConnection.StartAsync(spec, log, _ => { }, onExiting, ct),
(_, _) => { }, log, startFailureWarningInterval: TimeSpan.FromMilliseconds(warnEveryMs));
// The background recovery retries at the backoff (250 ms, 500 ms, 1 s, ±20 %) until the 4th attempt starts.
await shard.WhenRecoveredAsync(null).WaitAsync(TimeSpan.FromSeconds(30), Ct);
shard.Current.ShouldNotBeNull();
attempts.ShouldBe(4);
var events = sink.Events.Where(e => e.MessageTemplate.Text.StartsWith("ScriptWorker[{Shard}]", StringComparison.Ordinal)).ToList();
string Dump() => string.Join(Environment.NewLine, events.Select(e => $"{e.Level} {e.RenderMessage()} {(e.Exception is null ? "" : "+exception")}"));
static int Attempts(Serilog.Events.LogEvent e) => int.Parse(e.Properties["Attempts"].ToString(), System.Globalization.CultureInfo.InvariantCulture);
var errors = events.Where(e => e.Level == Serilog.Events.LogEventLevel.Error).ToList();
errors.Count.ShouldBe(1, Dump());
errors[0].Exception.ShouldNotBeNull("the first failure of the outage carries the stack");
var warnings = events.Where(e => e.Level == Serilog.Events.LogEventLevel.Warning).ToList();
warnings.Count.ShouldBe(expectedWarnings, Dump());
warnings.ShouldAllBe(e => e.Exception == null);
warnings.Select(Attempts).ShouldBe(expectedWarnings == 0 ? [] : [2, 3]);
events.Count(e => e.Level == Serilog.Events.LogEventLevel.Debug && e.Properties.ContainsKey("Attempts")).ShouldBe(expectedDebugs, Dump());
var recovered = events.Single(e => e.Level == Serilog.Events.LogEventLevel.Information && e.Properties.ContainsKey("Attempts"));
Attempts(recovered).ShouldBe(3);
recovered.RenderMessage().ShouldContain("after 3 failed start attempt(s)");
}
// ---------------------------------------------------------------- lmxopcua#790: compile timeout from compile start
/// <summary>A script of <paramref name="lines"/> statements: Roslyn needs time roughly proportional to it.</summary>
private static string Lines(int lines, int salt)
{
var body = new StringBuilder("var a = ").Append(salt).Append(";\n");
for (var i = 0; i < lines; i++) body.Append("a += (int)ctx.GetTag(\"x\").Value * ").Append(i % 97).Append(";\n");
return body.Append("return a;").ToString();
}
private static int LinesValue(int lines, int salt) => salt + Enumerable.Range(0, lines).Sum(i => i % 97);
/// <summary>
/// lmxopcua#790: the compile timeout runs from when a compile thread STARTS a load, so a queue of compiles that
/// takes several load timeouts to drain kills nothing and strikes nobody. One compile thread; the compiler is
/// warmed under a generous timeout (a cold Roslyn start alone takes seconds on the Windows VM), then the live
/// connection's timeout drops to four times one measured compile (at least 3 s; a 500-statement script takes
/// ~0.1 s on an M-series Mac and ~1.5 s on the Windows VM) and a burst of such scripts, sized from that and grown
/// until it does, queues for over three timeouts — asserted, so the test cannot pass vacuously
/// — while each compile stays far inside the timeout. Mutation check: time the load from when the host SENT it
/// (the old <c>WaitAsync(LoadTimeout)</c> around the load) and the worker is killed and the queued keys are
/// struck.
/// </summary>
[Fact]
public async Task Compiles_queued_behind_others_longer_than_the_load_timeout_are_not_killed_or_struck()
{
const int lines = 500;
var sink = new CapturingSink();
await using var runner = new OutOfProcessScriptRunner(Scripts.Options(), sink.CreateLogger(), null,
heartbeatInterval: TimeSpan.FromSeconds(1), loadTimeout: TimeSpan.FromSeconds(120), compileThreads: 1);
(await runner.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7);
var worker = runner.Shards[0].Current.ShouldNotBeNull();
ScriptInvocation Script(string prefix, int i)
=> Scripts.Vt($"{prefix}-{i}", Lines(lines, i), TimeSpan.FromSeconds(300), Scripts.Input("x", 1));
// Warm the compiler, then calibrate: the compile time of one such script on this box.
for (var i = 0; i < 3; i++) (await runner.RunAsync(Script("warm", i), Ct)).Value.ShouldBe(LinesValue(lines, i));
var probe = Stopwatch.StartNew();
for (var i = 0; i < 3; i++) (await runner.RunAsync(Script("probe", i), Ct)).Value.ShouldBe(LinesValue(lines, i));
var perCompile = probe.Elapsed / 3;
var loadTimeout = TimeSpan.FromTicks(Math.Max(TimeSpan.FromSeconds(3).Ticks, perCompile.Ticks * 4));
worker.LoadTimeout = loadTimeout;
var count = (int)Math.Clamp(Math.Ceiling(7 * loadTimeout / perCompile), 10, 300);
// A burst compiles faster per script than the sequential probe (the compiler keeps warming up), so grow the burst
// until it really outlasts three load timeouts; each burst must pass the same no-kill checks.
ScriptRunOutcome[] outcomes = [];
var elapsed = TimeSpan.Zero;
for (var round = 0; round < 4; round++)
{
runner.Reset(ScriptContextKind.VirtualTag); // unload the previous round's scripts (worker heap)
var sw = Stopwatch.StartNew();
var prefix = "queued" + round;
outcomes = await Task.WhenAll(Enumerable.Range(0, count).Select(i => runner.RunAsync(Script(prefix, i), Ct)));
elapsed = sw.Elapsed;
for (var i = 0; i < count; i++) outcomes[i].Value.ShouldBe(LinesValue(lines, i), $"round {round}: {outcomes[i].Status} {outcomes[i].Reason}");
if (elapsed > loadTimeout * 3) break;
count = Math.Min(1_200, count * 2);
}
var log = string.Join(Environment.NewLine, sink.Rendered.Where(l => l.Contains("ScriptWorker", StringComparison.Ordinal)).Take(5));
elapsed.ShouldBeGreaterThan(loadTimeout * 3,
$"precondition: {count} compiles (probe {perCompile.TotalMilliseconds:F0} ms each) must queue for several load timeouts, or this proves nothing");
worker.Exited.IsCompleted.ShouldBeFalse(log);
runner.Shards[0].Current.ShouldNotBeNull().Pid.ShouldBe(worker.Pid, log);
runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBeEmpty();
runner.SuspectKeys.ShouldBeEmpty();
sink.Rendered.ShouldNotContain(l => l.Contains("killing the worker", StringComparison.Ordinal));
}
/// <summary>
/// The positive control for the test above: a compile that is still running <c>LoadTimeout</c> after it STARTED is
/// the script's fault — the worker is killed (<c>KilledLoadTimeout</c>), the caller gets the compile-timeout
/// failure, the key takes a kill strike (the second quarantines it), and the healthy key on the same shard is
/// not blamed. The healthy key comes from a primed PE cache, so a fresh worker's cold Roslyn start never counts
/// against the short timeout. Mutation check: never arm the timer on <c>LoadStarted</c> and the huge compile
/// simply finishes (status Ok, red).
/// </summary>
[Fact]
public async Task A_compile_still_running_a_load_timeout_after_it_started_is_killed_and_struck()
{
using var dir = new Scripts.TempDir();
var cache = Path.Combine(dir.Path, "pe-cache");
await using (var primer = Scripts.Runner(Scripts.Options(peCacheDir: cache)))
(await primer.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7);
var sink = new CapturingSink();
await using var runner = new OutOfProcessScriptRunner(Scripts.Options(quarantineAfterKills: 2, peCacheDir: cache),
sink.CreateLogger(), null, heartbeatInterval: TimeSpan.FromSeconds(1), loadTimeout: TimeSpan.FromSeconds(2), compileThreads: 1);
(await runner.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7);
// 20,000 statements: about 4 s of Roslyn on an M-series Mac, longer elsewhere.
var huge = Scripts.Vt("huge", Lines(20_000, 0), TimeSpan.FromSeconds(30), Scripts.Input("x", 1));
for (var strike = 1; strike <= 2; strike++)
{
var before = runner.Shards[0].Current ?? await runner.Shards[0].GetReadyAsync(TimeSpan.FromSeconds(30), Ct);
var outcome = await runner.RunAsync(huge, Ct).WaitAsync(TimeSpan.FromSeconds(60), Ct);
outcome.Status.ShouldBe(ScriptRunStatus.Failed, $"strike {strike}");
outcome.Reason.ShouldNotBeNull().ShouldContain("compile did not finish");
var exit = await before.ShouldNotBeNull().Exited.WaitAsync(TimeSpan.FromSeconds(10), Ct);
exit.Reason.ShouldBe(WorkerExitReason.KilledLoadTimeout);
exit.CulpritKey.ShouldBe(VtKey("huge"));
}
var third = await runner.RunAsync(huge, Ct);
third.Reason.ShouldNotBeNull().ShouldStartWith(ScriptRunGuard.QuarantinedReasonPrefix);
runner.QuarantinedKeys(ScriptContextKind.VirtualTag).ShouldBe(["huge"]);
(await runner.RunAsync(Scripts.Vt("first", Healthy), Ct)).Value.ShouldBe(7, "the healthy key on the same shard is not blamed");
}
}