diff --git a/docs/plans/2026-11-09-out-of-process-script-runner-design.md b/docs/plans/2026-11-09-out-of-process-script-runner-design.md index 763a521c..e2b6e0ba 100644 --- a/docs/plans/2026-11-09-out-of-process-script-runner-design.md +++ b/docs/plans/2026-11-09-out-of-process-script-runner-design.md @@ -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 diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs index 776a619e..e4c1f64b 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/OutOfProcessScriptRunner.cs @@ -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); - /// How long a compile in the worker may take before the worker is killed. + /// + /// 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). + /// internal TimeSpan LoadTimeout { get; } /// The shard workers (tests). @@ -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("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; + } + + /// + /// 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. + /// + private static string[] SlotOrder(KeyState state, IReadOnlyDictionary 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; } /// @@ -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(value(s), new KeyValuePair("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; + + /// SHA-256 of (computed on first use; a benign race computes the same bytes). + public byte[] Sha => _sha ??= SHA256.HashData(Encoding.UTF8.GetBytes(Source)); + + /// Its upper-case hex form. + public string ShaHex => _shaHex ??= Convert.ToHexString(Sha); + + /// The negative compile cache's key for this source (kind-scoped). + public string FailureKey => _failureKey ??= InProcessScriptRunner.ScopedKey(kind, ShaHex); + + /// The slot order last used for this key (see ). + public string[]? Paths; public readonly List KillStrikes = []; public int InFlight; public int ConsecutiveTimeouts; diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/ResultDispatcher.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/ResultDispatcher.cs new file mode 100644 index 00000000..12f4a7c0 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/ResultDispatcher.cs @@ -0,0 +1,90 @@ +using ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.V1; + +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess; + +/// +/// The thread that resumes the callers of one worker's evaluations (lmxopcua#789). +/// +/// +/// 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 +/// DOTNET_ThreadPool_UnfairSemaphoreSpinLimit=0 the out-of-process runner cost only ~10 points of a core more +/// than in-process). This thread blocks without spinning () and runs the +/// continuations of every result that is ready in one wake-up. +/// 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. +/// 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 is async-only for that reason. +/// +internal sealed class ResultDispatcher +{ + private readonly object _gate = new(); + private readonly Queue<(PendingEval Pending, EvalResult? Result)> _queue = new(); + private bool _waiting; + private bool _stopped; + + /// Starts the thread. + /// The thread name. + public ResultDispatcher(string name) + { + new Thread(Run) { IsBackground = true, Name = name }.Start(); + } + + /// Queues a caller's resumption. After it runs on the calling thread instead. + /// The evaluation. + /// Its result (null: the worker ended). + public void Post(PendingEval pending, EvalResult? result) + { + lock (_gate) + { + if (!_stopped) + { + _queue.Enqueue((pending, result)); + if (_waiting) Monitor.Pulse(_gate); + return; + } + } + pending.SignalOutcome(result); + } + + /// Lets the thread drain what is queued and end. + 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(); + } + } +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WindowsJobObject.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WindowsJobObject.cs index c44d6829..449deade 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WindowsJobObject.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WindowsJobObject.cs @@ -59,6 +59,14 @@ internal sealed class WindowsJobObject : IDisposable } } + /// + /// 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 Process.Kill(entireProcessTree: true)'s system-wide process snapshot, + /// which cost about 100–200 ms of the Windows kill latency. + /// + /// False when the job could not be terminated (the caller falls back to a process kill). + public bool Terminate() => !_handle.IsClosed && TerminateJobObject(_handle, unchecked((uint)-1)); + /// Closes the job handle, which kills every process still in it. 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 } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs index 063fa6d0..009421e4 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerConnection.cs @@ -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 /// The compiled-PE cache directory handed to the worker; null = no cache. /// The cache's HMAC key (with ). /// Where the launch marker for the orphan sweep is written; null = none. +/// +/// How long one compile may run, counted from the worker's LoadStarted for it (not from when the host sent +/// it, lmxopcua#790), before the worker is killed and the key struck; null = 60 s. +/// +/// The worker's dedicated compile threads; 0 = its default. 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) +{ + /// The default compile timeout. + public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60); + + /// or its default. + public TimeSpan EffectiveLoadTimeout => LoadTimeout ?? DefaultLoadTimeout; +} /// /// 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 _onLogs; - private readonly Channel _outbound = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true }); + private readonly ScriptWorkerFrameWriter _writer; + private readonly ResultDispatcher _dispatcher; private readonly ConcurrentDictionary _pending = new(); - private readonly ConcurrentDictionary Completion)> _loads = new(); + private readonly ConcurrentDictionary _loads = new(); private readonly ConcurrentDictionary _loaded = new(StringComparer.Ordinal); private readonly TaskCompletionSource _exited = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly CancellationTokenSource _stopping = new(); private readonly TimeSpan _heartbeatInterval; + private TimeSpan _loadTimeout; private readonly Action _onExiting; private readonly string? _markerPath; private TaskCompletionSource? _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 onLogs, TimeSpan heartbeatInterval, Action onExiting, string? markerPath) + ILogger log, Action onLogs, TimeSpan heartbeatInterval, TimeSpan loadTimeout, Action 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"); } /// The shard label. @@ -156,6 +175,16 @@ internal sealed class WorkerConnection : IAsyncDisposable /// Evaluations currently outstanding on this worker. public int InFlightCount => _pending.Count; + /// + /// 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). + /// + internal TimeSpan LoadTimeout + { + get => _loadTimeout; + set => _loadTimeout = value; + } + /// Compiles currently outstanding on this worker. 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); + } + + /// + /// As with blocking I/O only (): + /// 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 ; the caller then kills the + /// worker and disposes the pipe, which ends it. + /// + /// The accepted pipe. + /// The Hello to send. + /// The nonce put in the worker's environment. + /// The launched process id; null skips the check. + /// Cancellation (the handshake timeout). + /// The worker's Ready. + internal static async Task 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 /// The source. /// SHA-256 of the source. /// The slot order. - /// The load outcome and the slot map; null when the worker ended first. - public Task EnsureLoadedAsync(string key, ContextKind kind, string source, byte[] sourceHash, IReadOnlyList inputPaths) + /// The entry once the load has an outcome (see ). + public Task EnsureLoadedAsync(string key, ContextKind kind, string source, byte[] sourceHash, IReadOnlyList inputPaths) + => EnsureLoadedAsync(key, kind, source, sourceHash, Convert.ToHexString(sourceHash), inputPaths); + + /// As the other overload, with the hash's hex form precomputed (the runner caches it per key, lmxopcua#789). + /// The scoped key. + /// The context kind. + /// The source. + /// SHA-256 of the source. + /// Its upper-case hex form. + /// The slot order. + /// + /// The entry once the load has an outcome: is the worker's verdict, or null when + /// the worker ended first; marks the compile this connection killed the worker + /// for. + /// + public Task EnsureLoadedAsync(string key, ContextKind kind, string source, byte[] sourceHash, string sourceHashHex, + IReadOnlyList 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(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(key, entry)); - } - else - { - entry.Complete(null); - _loaded.TryRemove(new KeyValuePair(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 /// The scoped key. /// The request (its id is assigned here). /// The pending evaluation. - public PendingEval Submit(string key, EvalRequest request) + public PendingEval Submit(string key, EvalRequest request) => Submit(key, request, null, TimeSpan.Zero, null); + + /// + /// Submits one evaluation with its two deadlines (design §5.1) on ONE timer (lmxopcua#789): at + /// yields , + /// and later, if the worker still has not answered, the worker is killed with this + /// key as the culprit. + /// + /// The scoped key. + /// The request (its id is assigned here). + /// T; null = no deadlines. + /// G. + /// + /// Runs once, synchronously, when the worker has answered or ended — BEFORE + /// completes — so a caller that resumes from its await never finds its own run still holding the key's slot. + /// + /// The pending evaluation. + 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); + } + /// Cancels a run's ctx.CancellationToken in the worker (sent at the soft deadline). /// The evaluation. 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 /// Test hook: closes the host end of the pipe without touching the process, as a dying host would. 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() + /// + /// The reader, on a dedicated thread with blocking reads of a socket that never turns non-blocking + /// (, 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. + /// + 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(); + } + + /// The compile timer of a started load fired: kill the worker, blaming this key alone (lmxopcua#790). + 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 */ } } + /// One load sent to the worker and not yet answered: its compile timer runs from LoadStarted. + 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; + + /// The worker has started compiling it: arm the compile timeout (once). + 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; + + /// The worker answered () or ended (null). + 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(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(Key, Entry)); + Entry.Complete(outcome); + } + } + /// A key's compiled form in this worker. internal sealed class LoadEntry { - private readonly TaskCompletionSource _ready = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _ready = new(TaskCreationOptions.RunContinuationsAsynchronously); /// Set once the runner has counted this load in its metrics. public int Counted; @@ -732,10 +884,17 @@ internal sealed class WorkerConnection : IAsyncDisposable /// Path → slot in the registered order. public IReadOnlyDictionary Slots { get; } - /// The worker's verdict (null until it arrives). + /// The worker's verdict; null until it arrives, and null for good when the worker ended first. public LoadOutcome? Outcome { get; private set; } - public Task Ready => _ready.Task; + /// + /// 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. + /// + public volatile bool TimedOut; + + /// Completes with this entry once the load has an outcome (or the worker ended). + public Task Ready => _ready.Task; public bool SameSlots(IReadOnlyList 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); } } } -/// One evaluation outstanding on a worker. +/// +/// One evaluation outstanding on a worker. Its deadlines share ONE timer (lmxopcua#789: the per-eval +/// WaitAsync(timeout) timer, the hard-deadline CancellationTokenSource 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. +/// internal sealed class PendingEval { - private readonly TaskCompletionSource _completion = new(TaskCreationOptions.RunContinuationsAsynchronously); + /// What yields when the soft deadline passed first. Never on the wire. + internal static readonly EvalResult SoftTimedOut = new(); - public PendingEval(string key, ulong evalId) + private readonly TaskCompletionSource _completion = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _outcome; + private readonly ResultDispatcher? _dispatcher; + private readonly Action? _onFinished; + private Timer? _timer; + private Action? _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(dispatcher is null + ? TaskCreationOptions.RunContinuationsAsynchronously + : TaskCreationOptions.None); } /// The scoped key. @@ -770,8 +951,59 @@ internal sealed class PendingEval /// The eval id on the wire. public ulong EvalId { get; } + /// T + G, for the kill log line. + public TimeSpan HardDeadline { get; private set; } + /// The worker's result, or null when the worker ended first. public Task Completion => _completion.Task; - public void Complete(EvalResult? result) => _completion.TrySetResult(result); + /// + /// What the caller waits on: the worker's result, null when the worker ended first, or + /// at the soft deadline (the run itself is still outstanding then). + /// + public Task Outcome => _outcome.Task; + + /// Starts the deadline timer. Called once, before the request is sent. + internal void Arm(TimeSpan soft, TimeSpan grace, Action 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); + } + + /// + /// 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). + /// + 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); + } + } + + /// Resumes the caller (on the dispatcher's thread). A soft timeout may already have done so. + internal void SignalOutcome(EvalResult? result) => _outcome.TrySetResult(result); } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerShard.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerShard.cs index 7f432eeb..e8359f94 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerShard.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerShard.cs @@ -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); + /// + /// 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). + /// + internal static readonly TimeSpan DefaultStartFailureWarningInterval = TimeSpan.FromSeconds(60); + private static readonly Counter StartsCounter = OtOpcUaTelemetry.Meter.CreateCounter( "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? _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, CancellationToken, Task> launch, - Action onExit, ILogger log) + Action 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("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; } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto index 4c5573f0..2afa1d78 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/Protos/script_worker.proto @@ -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; diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFrameCodec.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFrameCodec.cs index ea54851f..9291e3ed 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFrameCodec.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFrameCodec.cs @@ -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.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 } } + /// + /// 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. + /// + public static bool UseBlockingIo => !OperatingSystem.IsWindows(); + + /// Writes one frame synchronously (the blocking-I/O handshake). Not thread-safe. + /// The pipe. + /// The envelope. + /// The cap. + 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.Shared.Rent(size + 4); + try + { + BinaryPrimitives.WriteInt32LittleEndian(buffer, size); + envelope.WriteTo(new Span(buffer, 4, size)); + stream.Write(buffer, 0, size + 4); + } + finally + { + ArrayPool.Shared.Return(buffer); + } + } + + /// + /// Reads exactly one frame synchronously, never past it (the blocking-I/O handshake: what follows belongs to the + /// connection's own reader). + /// + /// The pipe. + /// The cap. + /// The envelope, or null on a clean end of stream at a frame boundary. + public static Envelope? Read(Stream stream, int maxFrameBytes) + { + ArgumentNullException.ThrowIfNull(stream); + Span 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.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.Shared.Return(body); + } + } + + /// Validates a frame's length prefix against the cap. + /// The decoded length. + /// The negotiated cap. + /// A zero, negative or oversize length. + 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}"); + } + + /// Parses and validates one frame body (everything after the length prefix). + /// The body bytes. + /// The envelope. + /// Unparseable, the wrong protocol version, or no body. + internal static Envelope ParseBody(ReadOnlySpan 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 ReadFullyAsync(Stream stream, Memory buffer, CancellationToken ct) { var total = 0; diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs new file mode 100644 index 00000000..09e6fa16 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs @@ -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; + +/// +/// A buffered reader of 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 +/// : zero-length, oversize, truncated, unparseable, wrong-version and +/// bodiless frames are a . Not thread-safe: one reader per stream. +/// +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; + + /// Initializes the reader. + /// The pipe. + /// The frame cap. + public ScriptWorkerFrameReader(Stream stream, int maxFrameBytes = ScriptWorkerFrameCodec.DefaultMaxFrameBytes) + { + ArgumentNullException.ThrowIfNull(stream); + _stream = stream; + MaxFrameBytes = maxFrameBytes; + } + + /// The negotiated frame cap. + public int MaxFrameBytes { get; set; } + + /// Reads the next frame. + /// Cancellation. + /// The envelope, or null on a clean end of stream at a frame boundary. + public async ValueTask 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; + } + } + + /// Reads the next frame, blocking (for a dedicated reader thread). + /// The envelope, or null on a clean end of stream at a frame boundary. + 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; + } + + /// Makes room for a frame of bytes (prefix included) starting at _start. + 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"); + } +} + +/// +/// 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 +/// items are merged into one batch and consecutive 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. +/// +/// +/// 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 (, 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. +/// A failed write completes the writer and reports the exception once through onFault. +/// +public sealed class ScriptWorkerFrameWriter +{ + private readonly Stream _stream; + private readonly Action _onFault; + private readonly ConcurrentQueue _queue = new(); + private readonly ArrayBufferWriter _pass = new(16 * 1024); + private readonly bool _synchronous; + private int _draining; + private int _completed; + private int _faulted; + private long _seq; + + /// Initializes the writer. + /// The pipe. + /// Called once when a write fails (or a frame is refused); the writer is then completed. + /// The frame cap. + /// The last sequence number already used (the handshake's). + /// Write with blocking calls (see the remarks). + public ScriptWorkerFrameWriter(Stream stream, Action 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; + } + + /// The negotiated frame cap. + public int MaxFrameBytes { get; set; } + + /// Whether has been called or a write failed. + public bool IsCompleted => Volatile.Read(ref _completed) != 0; + + /// + /// 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. + /// + /// The frame (its version and sequence number are set here). + /// Whether the frame was accepted. + 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; + } + + /// Stops accepting frames. Frames already queued are still written by a pass in progress. + 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); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/Program.cs index 6ffa161e..bf598115 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/Program.cs @@ -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) { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs index 31d3734b..fe239452 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker/WorkerHost.cs @@ -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 _outbound = Channel.CreateUnbounded( - new UnboundedChannelOptions { SingleReader = true }); private readonly ConcurrentDictionary _scripts = new(StringComparer.Ordinal); private readonly ConcurrentDictionary _running = new(); private readonly BlockingCollection _queue = new(new ConcurrentQueue()); @@ -47,8 +44,16 @@ internal sealed class WorkerHost /// private readonly BlockingCollection _loads = new(new ConcurrentQueue()); - /// Dedicated compile threads: half the processors, at least 1, at most 8. - internal static int CompileThreads => Math.Clamp(Environment.ProcessorCount / 2, 1, 8); + /// Default dedicated compile threads: half the processors, at least 1, at most 8. + internal static int DefaultCompileThreads => Math.Clamp(Environment.ProcessorCount / 2, 1, 8); + + /// + /// 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). + /// + 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); } + /// The blocking read loop on its own thread (Unix, lmxopcua#789); completes with the exit code. + private Task ReadOnDedicatedThread() + { + var done = new TaskCompletionSource(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 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: + /// Acts on one frame from the host. + /// The frame; null = end of stream. + /// The exit code when the worker must end, otherwise null. + 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) { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ProtocolTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ProtocolTests.cs index f20be772..f8fc959b 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ProtocolTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ProtocolTests.cs @@ -173,6 +173,157 @@ public sealed class ProtocolTests } } + // ---------------------------------------------------------------- buffered reader / combining writer (lmxopcua#789) + + /// Hands out at most chunk bytes per read, so frames straddle reads at every possible offset. + 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 ReadAsync(Memory 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(); + } + + /// + /// 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. + /// + [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"); + } + + /// The buffered reader rejects exactly what the codec rejects. + [Theory] + [MemberData(nameof(BadFrames))] + public async Task The_buffered_reader_rejects_malformed_frames_like_the_codec(string why, byte[] bytes) + { + await Should.ThrowAsync(async () => + await new ScriptWorkerFrameReader(new MemoryStream(bytes), 1024).ReadAsync(TestContext.Current.CancellationToken), why); + Should.Throw(() => new ScriptWorkerFrameReader(new ChunkedStream(bytes, 1), 1024).Read(), why); + } + + /// Random bytes never escape the buffered reader as anything but a protocol fault (or a clean end). + [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) { } + } + } + + /// A stream whose first write waits on a gate, so frames queue up behind a write in progress. + private sealed class GatedStream : MemoryStream + { + public readonly TaskCompletionSource Gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + public int Writes; + + public override async ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + { + if (Interlocked.Increment(ref Writes) == 1) await Gate.Task; + await base.WriteAsync(buffer, cancellationToken); + } + } + + private static List ReadAll(byte[] bytes) + { + var reader = new ScriptWorkerFrameReader(new MemoryStream(bytes)); + var frames = new List(); + 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" } } } }; + + /// + /// 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. + /// + [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"); + } + + /// A failed write completes the writer, reports once, and later frames are refused rather than lost silently. + [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() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ScriptRunnerPerfTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ScriptRunnerPerfTests.cs index cdc3ce6d..3f82550a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ScriptRunnerPerfTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/ScriptRunnerPerfTests.cs @@ -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}"); /// @@ -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(StringComparer.Ordinal); var configs = new (string Name, Func 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); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs index fbfbd6e6..5f075fcf 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ScriptWorker.Tests/SupervisionTests.cs @@ -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. /// [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(); } /// @@ -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 + + /// + /// 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). + /// + [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 + + /// + /// 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). + /// + [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 + + /// A script of statements: Roslyn needs time roughly proportional to it. + 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); + + /// + /// 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 WaitAsync(LoadTimeout) around the load) and the worker is killed and the queued keys are + /// struck. + /// + [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)); + } + + /// + /// The positive control for the test above: a compile that is still running LoadTimeout after it STARTED is + /// the script's fault — the worker is killed (KilledLoadTimeout), 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 LoadStarted and the huge compile + /// simply finishes (status Ok, red). + /// + [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"); + } }