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 f9bd8e713..4e6f19209 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
@@ -965,6 +965,7 @@ on Unix by the socket engine. Changes:
- 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.
+ (Superseded in §12.6: Windows now blocks too, through `WindowsBlockingPipeStream`.)
- **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;
@@ -1299,6 +1300,91 @@ Owner decisions §14 #5 and #6, as built. Operator reference: `docs/ServiceHosti
- **Not verified live:** the installer on a real node. That is the QA deploy gate on lmxopcua#780's follow-up (see
that issue for the checklist).
+### 12.6 lmxopcua#789 host CPU on Windows: blocking pipe I/O off the completion port (2026-11-11)
+
+**Already in place before this slice** (checked against the issue's candidates): one timer per evaluation for T and
+T + G (§12.2); the source SHA-256, its hex form and the negative-cache key are computed once per key's source
+(`KeyState.Sha`), not per `RunAsync`; batch submit (§12.3) is used by the Calculation pump, the VirtualTag coalescer and
+the alarm engine on the out-of-process path. The §8 paced harness phases call `RunAsync` one at a time on purpose (one
+trigger per evaluation); `Batch_throughput_and_alarm_pass_latency` measures the batched path.
+
+**What changed.**
+- **`WindowsBlockingPipeStream`** (Protocol). A .NET pipe opened `Asynchronous` completes even its synchronous
+ `Read`/`Write` through the thread pool's completion port, so every frame on Windows cost a completion packet plus a
+ pool work item. The stream issues its own overlapped `ReadFile`/`WriteFile` on the pipe handle with the
+ `OVERLAPPED.hEvent` low-order bit set, which tells the kernel not to queue a completion packet, and waits on that
+ event on the calling thread. Reads and writes stay concurrent (a handle opened for synchronous I/O would serialise
+ them). Dispose cancels a waiting read or write (`CancelIoEx`) and closes the pipe; a peer that goes away ends a read as
+ end of stream and fails a write with `IOException`, as `PipeStream` does.
+- **`ScriptWorkerFrameCodec.BlockingStream(pipe)`** picks that stream on Windows and the pipe itself on Unix. Both ends
+ now use the dedicated blocking reader thread and the synchronous combining writer on every OS. `UseBlockingIo` and the
+ two async read loops are gone; `HandshakeAsync` stays for its tests.
+- **1 MiB pipe buffer quota each way** (`WorkerPipeSecurity.BufferBytes`). With a quota of 0 an overlapped write stays
+ pending until the peer has read the bytes, so the blocking writer waited out a peer wake-up per frame. The first cut
+ without it halved the windev closed loop (54k → 26–27k evals/s); with it the closed loop is back.
+- Tests: `BlockingPipeStreamTests` (duplex with reads blocked on both ends, a 1 MB frame, peer gone, self-dispose wakes
+ the reader).
+
+**Measured on windev** (same VM as §12.1, idle, Release, `Throughput_and_latency_against_the_accepted_gate`, 10 s
+phases, base = `f109e5f2`, runs alternated base / new):
+
+| K = 2 | base (3 runs) | new (2 runs) | §8.2 budget |
+|---|---:|---:|---|
+| Host CPU at paced 10k/s, raw | 116 / 118 / 120 % | 61 / 84 % | — |
+| … vs InProcess (InProcess 6–9 %) | +108 / +112 / +112 | **+53 / +78** | ≤ 25 ✗ |
+| Host CPU at paced 5k/s, vs InProcess | +35 / +37 / +38 | −26 / −38 | — |
+| Idle p50 (sequential) | 491 / 499 / 560 µs | 363 / 386 µs | ≤ 150 µs ✗ |
+| Idle host CPU (sequential) | 57 / 61 / 66 % | 27 / 37 % | — |
+| p99 at 10k/s | 2,108 / 2,264 / 2,373 µs | 878 / 944 µs | ≤ 2 ms ✓ (was ✗) |
+| Closed loop ×64 | 51k / 54k / 55k evals/s | 54k / 55k evals/s | gate ≥ 5k ✓ |
+
+- The first cut (no buffer quota), two runs: 10k/s raw 118 / 127 %, idle p50 354 / 362 µs, closed loop 26–27k/s.
+- **Attribution of what is left.** With `DOTNET_ThreadPool_UnfairSemaphoreSpinLimit=0` the new build reads 71 % raw at
+ 10k/s (+65) and the base 93 % (+85). So pool spinning is no longer what the remainder is made of: it is about 60 µs of
+ host CPU per evaluation in syscalls and thread wake-ups on this VM (reader wake, dispatcher wake, and the harness's
+ own `ContinueWith` pool hop per result), against about 13 µs on macOS. A non-VM Windows box is still unmeasured.
+
+**macOS** (Unix path unchanged; 2 runs each, alternated, load average 3–14): K = 2 at 10k/s +9 / +12 vs base +16 / +5
+(InProcess 46–57 %); idle p50 26 µs both; closed loop 365–374k vs 364–379k evals/s. No change, as expected.
+
+**Tests.** ScriptWorker `Category=ScriptWorker` on windev: 107 passed, 2 skipped by design (109), including all
+`WorkerSoftClockTests` suspend cases, the deferred-start / runaway-in-batch deadline tests, the kill and quarantine tests
+and the restricted-token worker. On macOS: 104 passed, 5 skipped (109). Unit classes `BlockingPipeStreamTests`,
+`ProtocolTests` and `HandshakeTests`: 67 / 67 on both.
+
+**Review follow-ups (same slice).** Blocking I/O moves two failure modes onto dedicated threads and producers:
+- **Reader faults.** The host reader runs completion work inline (slot release, load bookkeeping). The frame read and
+ its handling are now caught separately: only the read's I/O exceptions mean the pipe ended. Anything the handling
+ throws kills that worker as `ProtocolFault`, so its outstanding runs answer "worker restarted" (BadResourceUnavailable)
+ and the shard restarts. Before, an exception escaping a dedicated thread would end the host process, and an
+ `InvalidOperationException` was swallowed as a clean pipe end, leaving the worker unread until a `Crash` 2 s later.
+ The worker's reader has the same catch-all (exit `ProtocolFault`).
+- **A worker that talks but does not listen.** A full pipe blocks the producer that is draining the writer: a seam
+ submitting, a deploy loading scripts, a Cancel. Heartbeats keep the silence check quiet, and with only loads queued
+ nothing else fires. The heartbeat monitor now kills a worker whose pipe write has been blocked for longer than
+ `max(10 s, silence limit)` (`ScriptWorkerFrameWriter.WriteInProgressFor`, `KilledUnresponsive`). That fails the write
+ and releases the producer. The one known way a healthy worker's reader could wedge is closed too: `Cancel` now runs a
+ script's `ctx.CancellationToken` callbacks on the pool, not on the worker's reader. `StopAsync` sends `Shutdown` off
+ the caller, so its 1 s budget holds on a full pipe.
+- `WindowsBlockingPipeStream` refuses the async surface, because `Stream`'s base versions serialise a write behind a
+ waiting read.
+- Tests: `PipeFaultTests` covers a throwing completion callback (the worker ends `ProtocolFault`, the sibling in flight
+ gets BadResourceUnavailable, and the shard serves again) and a worker that stops reading while it heartbeats (killed
+ `KilledUnresponsive` within the bound, and the blocked deploy producer returns). The second uses the
+ `Hello.test_stall_reader_on_diag` seam. `BlockingPipeStreamTests` adds dispose with a write blocked on a full pipe,
+ and with a read and a write both pending. The protocol, handshake and pipe classes now carry `Category=ScriptWorker`.
+ Mutation checks: with no stall kill the stall test is red (timeout); with a catch-all that does not fault, the
+ callback test is red (`Crash`).
+- Re-run: `Category=ScriptWorker` on windev: 179 passed, 2 skipped by design (181). On macOS: 176 passed, 5 skipped
+ (181). windev perf after the fixes (alternated new / base / new): 10k/s vs InProcess +55 / +68 (base +95 in the same
+ session); idle p50 391 / 365 µs (base 517); closed loop 51k / 54k evals/s (base 57k). No regression against the
+ figures above.
+
+**Still missed:** Windows host CPU vs InProcess (+53 to +78 against ≤ 25) and Windows idle p50 (363–386 µs against
+≤ 150 µs; InProcess alone is about 105 µs on this VM). Next candidates: drop the dispatcher wake when the caller's
+continuation is known to be cheap (needs a seam-level guarantee, the reason it exists is a slow consumer), or measure
+on non-VM Windows hardware before tuning further.
+
---
## 13. Test plan
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 31a128db9..4c5894c97 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
@@ -91,6 +91,10 @@ internal sealed record BatchSubmission(string Key, EvalRequest Request, TimeSpan
/// How long the worker may send nothing before it is killed (); null =
/// three s. Test seam (with ).
///
+///
+/// Test seam: the worker's reader stops reading for good at the first diagnostics request while its heartbeats go on —
+/// a worker that keeps talking but stopped listening (the write-stall check, lmxopcua#789).
+///
///
/// Ask the worker to run the soft clock from each run's start (lmxopcua#827; see ).
/// False (tests only) keeps the pre-#827 host clock, which is also what a worker that does not echo it gets.
@@ -110,7 +114,8 @@ internal sealed record WorkerLaunchSpec(
int CompileThreads = 0,
string? WorkerSid = null,
bool WorkerSoftClock = true,
- TimeSpan? HeartbeatSilenceLimit = null)
+ TimeSpan? HeartbeatSilenceLimit = null,
+ bool TestStallReaderOnDiag = false)
{
/// The default compile timeout.
public static readonly TimeSpan DefaultLoadTimeout = TimeSpan.FromSeconds(60);
@@ -138,7 +143,11 @@ internal sealed class WorkerConnection : IAsyncDisposable
];
private readonly Process _process;
- private readonly NamedPipeServerStream _pipe;
+ ///
+ /// The pipe as both directions drive it: blocking, through
+ /// (the pipe itself on Unix, a over it on Windows). Owns the pipe.
+ ///
+ private readonly Stream _pipe;
private readonly IDisposable? _job;
private readonly IDisposable? _stdio;
private readonly string _workDir;
@@ -171,7 +180,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
private long _nextLoadId;
private int _disposed;
- private WorkerConnection(string shard, Process process, NamedPipeServerStream pipe, IDisposable? job, IDisposable? stdio,
+ private WorkerConnection(string shard, Process process, Stream pipe, IDisposable? job, IDisposable? stdio,
string workDir, ILogger log, Action onLogs, TimeSpan heartbeatInterval, TimeSpan loadTimeout,
Action onExiting, string? markerPath, bool workerSoftClock, TimeSpan? heartbeatSilenceLimit)
{
@@ -192,7 +201,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
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);
+ synchronous: true);
_dispatcher = new ResultDispatcher($"ScriptWorker[{shard}].Results");
}
@@ -261,6 +270,20 @@ internal sealed class WorkerConnection : IAsyncDisposable
///
internal TimeSpan? ProgressCeilingOverride { get; set; }
+ ///
+ /// How long one pipe write may stay blocked before the worker is killed (,
+ /// lmxopcua#789). Writes are blocking, so a full pipe — a worker whose reader stopped while its heartbeats go on —
+ /// would otherwise hold whichever producer is draining the writer (a seam submitting, a deploy loading scripts, a
+ /// Cancel) forever: the silence check never fires, and with only loads queued nothing else does. A worker that is
+ /// reading drains the 1 MiB buffer far faster than this. Null = the larger of 10 s and the silence limit; set = an
+ /// override (tests).
+ ///
+ internal TimeSpan? WriteStallLimitOverride { get; set; }
+
+ /// The write-stall bound now (see ).
+ internal TimeSpan WriteStallLimit
+ => WriteStallLimitOverride ?? (SilenceLimit > TimeSpan.FromSeconds(10) ? SilenceLimit : TimeSpan.FromSeconds(10));
+
/// Test seam: ignore the worker's soft-deadline notices, as a worker whose watchdog never sends them would.
internal bool DropSoftNoticesForTest { get; set; }
@@ -312,6 +335,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
? WorkerPipeSecurity.CreateServer(pipeName, WorkerPipeSecurity.ForCurrentHost(spec.WorkerSid))
: new NamedPipeServerStream(pipeName, PipeDirection.InOut, maxNumberOfServerInstances: 1,
PipeTransmissionMode.Byte, PipeOptions.Asynchronous | PipeOptions.CurrentUserOnly);
+ Stream? io = null;
Process? process = null;
IDisposable? job = null;
IDisposable? stdio = null;
@@ -375,17 +399,17 @@ internal sealed class WorkerConnection : IAsyncDisposable
MaxLogLines = 1000,
CompileThreads = (uint)Math.Max(0, spec.CompileThreads),
WorkerSoftClock = spec.WorkerSoftClock,
+ TestStallReaderOnDiag = spec.TestStallReaderOnDiag,
};
if (spec.PeCacheDirectory is { } cacheDir && spec.PeCacheKey is { } cacheKey)
{
hello.PeCacheDir = cacheDir;
hello.PeCacheKey = Google.Protobuf.ByteString.CopyFrom(cacheKey);
}
- var ready = ScriptWorkerFrameCodec.UseBlockingIo
- ? await HandshakeBlockingAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false)
- : await HandshakeAsync(pipe, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false);
+ io = ScriptWorkerFrameCodec.BlockingStream(pipe);
+ var ready = await HandshakeBlockingAsync(io, hello, nonce, process.Id, handshakeCts.Token).ConfigureAwait(false);
- var connection = new WorkerConnection(spec.Shard, process, pipe, job, stdio, workDir, log, onLogs, spec.HeartbeatInterval,
+ var connection = new WorkerConnection(spec.Shard, process, io, job, stdio, workDir, log, onLogs, spec.HeartbeatInterval,
spec.EffectiveLoadTimeout, onExiting, markerPath, workerSoftClock: spec.WorkerSoftClock && ready.WorkerSoftClock,
heartbeatSilenceLimit: spec.HeartbeatSilenceLimit);
connection.Begin();
@@ -397,7 +421,8 @@ internal sealed class WorkerConnection : IAsyncDisposable
job?.Dispose();
stdio?.Dispose();
process?.Dispose();
- await pipe.DisposeAsync().ConfigureAwait(false);
+ if (io is not null) await io.DisposeAsync().ConfigureAwait(false);
+ else await pipe.DisposeAsync().ConfigureAwait(false);
TryDeleteDirectory(workDir);
WorkerLaunchMarker.TryDelete(markerPath);
throw;
@@ -430,7 +455,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
}
///
- /// As with blocking I/O only ():
+ /// 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.
@@ -548,8 +573,7 @@ internal sealed class WorkerConnection : IAsyncDisposable
ReadyTimestamp = Stopwatch.GetTimestamp();
Volatile.Write(ref _lastFrameTimestamp, ReadyTimestamp);
Interlocked.Exchange(ref _lastUsedTimestamp, ReadyTimestamp);
- if (!ScriptWorkerFrameCodec.UseBlockingIo) _ = Task.Run(ReadLoopAsync);
- else new Thread(ReadLoopBlocking) { IsBackground = true, Name = $"ScriptWorker[{Shard}].Reader" }.Start();
+ new Thread(ReadLoopBlocking) { IsBackground = true, Name = $"ScriptWorker[{Shard}].Reader" }.Start();
_ = Task.Run(HeartbeatMonitorAsync);
_ = Task.Run(WatchProcessAsync);
}
@@ -751,7 +775,9 @@ internal sealed class WorkerConnection : IAsyncDisposable
if (!_exited.Task.IsCompleted)
{
lock (_pending) { _killedFor ??= (reason, null); }
- Send(new Envelope { Shutdown = new Shutdown() });
+ // Off this thread: the send may drain the writer, and a write blocks on a full pipe — the 1 s budget below
+ // must hold regardless (the kill then fails that write).
+ _ = Task.Run(() => Send(new Envelope { Shutdown = new Shutdown() }));
try
{
await _process.WaitForExitAsync().WaitAsync(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
@@ -785,50 +811,56 @@ internal sealed class WorkerConnection : IAsyncDisposable
}
///
- /// 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.
+ /// The reader, on a dedicated thread with blocking reads (,
+ /// lmxopcua#789): a frame wakes this thread straight from the kernel, instead of the socket engine's epoll thread
+ /// (Unix) or the completion port (Windows) queueing a thread-pool work item per read.
///
private void ReadLoopBlocking()
{
WorkerExitReason? fault = null;
var reader = new ScriptWorkerFrameReader(_pipe);
- try
- {
- while (reader.Read() is { } frame) HandleFrame(frame);
- }
- catch (ScriptWorkerProtocolException ex)
- {
- fault = ProtocolFault(ex);
- }
- catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException
- or System.Net.Sockets.SocketException)
+ while (true)
{
+ Envelope? frame;
+ try
+ {
+ frame = reader.Read();
+ }
+ catch (ScriptWorkerProtocolException ex)
+ {
+ fault = ProtocolFault(ex);
+ break;
+ }
+ catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException
+ or System.Net.Sockets.SocketException)
+ {
+ break; // the pipe ended (these come from the read only, never from the frame's handling below)
+ }
+ if (frame is null) break;
+
+ try
+ {
+ HandleFrame(frame);
+ }
+ catch (ScriptWorkerProtocolException ex)
+ {
+ fault = ProtocolFault(ex);
+ break;
+ }
+ catch (Exception ex)
+ {
+ // HandleFrame runs completion work inline (slot releases, load bookkeeping). On this dedicated thread an
+ // exception that escaped would take the whole server process down, and treating it as the pipe ending
+ // would leave the worker running unread; so it ends this worker as a fault: every outstanding evaluation
+ // completes as "worker restarted" (BadResourceUnavailable) and the shard restarts.
+ _log.Error(ex, "ScriptWorker[{Shard}] failed handling a {Frame} frame; killing pid {Pid}", Shard, frame.BodyCase, Pid);
+ fault = WorkerExitReason.ProtocolFault;
+ break;
+ }
}
_ = PipeEndedAsync(fault);
}
- private async Task ReadLoopAsync()
- {
- WorkerExitReason? fault = null;
- var reader = new ScriptWorkerFrameReader(_pipe);
- try
- {
- while (await reader.ReadAsync(_stopping.Token).ConfigureAwait(false) is { } frame) HandleFrame(frame);
- }
- catch (OperationCanceledException) { }
- catch (ScriptWorkerProtocolException ex)
- {
- 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);
@@ -958,6 +990,15 @@ internal sealed class WorkerConnection : IAsyncDisposable
Kill(WorkerExitReason.KilledHeartbeat, null);
return;
}
+ if (_writer.WriteInProgressFor is { } blocked && blocked > WriteStallLimit)
+ {
+ // It talks but does not listen: the pipe is full and a producer is blocked in the write. The kill
+ // fails that write and releases it (lmxopcua#789).
+ _log.Warning("ScriptWorker[{Shard}] pid {Pid} has not read from its pipe for {Blocked} ms (a write is blocked on a full pipe) while still heartbeating; killing it",
+ Shard, Pid, (long)blocked.TotalMilliseconds);
+ Kill(WorkerExitReason.KilledUnresponsive, null);
+ return;
+ }
if (!_pending.IsEmpty && ProgressCeiling() is { } ceiling)
{
var stuck = Stopwatch.GetElapsedTime(Volatile.Read(ref _lastProgressTimestamp));
diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerPipeSecurity.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerPipeSecurity.cs
index 5c491e7ab..f30b819cd 100644
--- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerPipeSecurity.cs
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.OutOfProcess/WorkerPipeSecurity.cs
@@ -50,11 +50,21 @@ internal static class WorkerPipeSecurity
return security;
}
+ ///
+ /// The pipe's buffer quota in each direction (lmxopcua#789). With a quota of 0 an overlapped write stays pending
+ /// until the peer has READ the bytes, so a blocking writer ()
+ /// would wait out a full peer wake-up per frame — measured on windev as half the closed-loop throughput. With a
+ /// quota a write completes as soon as the bytes are buffered; a peer that stops reading still fills it and blocks
+ /// the writer: the host's write-stall check (WorkerConnection.WriteStallLimit) then kills the worker, which
+ /// fails the write and releases the blocked thread.
+ ///
+ internal const int BufferBytes = 1024 * 1024;
+
/// Creates the single-instance server end of a worker pipe with .
/// The random pipe name.
/// The ACL ().
/// The server stream.
public static NamedPipeServerStream CreateServer(string pipeName, PipeSecurity security)
=> NamedPipeServerStreamAcl.Create(pipeName, PipeDirection.InOut, maxNumberOfServerInstances: 1,
- PipeTransmissionMode.Byte, PipeOptions.Asynchronous, inBufferSize: 0, outBufferSize: 0, security);
+ PipeTransmissionMode.Byte, PipeOptions.Asynchronous, inBufferSize: BufferBytes, outBufferSize: BufferBytes, security);
}
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 ed67c6a63..086de3ae7 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
@@ -45,6 +45,9 @@ message Hello {
// starts the run, and to report the crossing in EvalResults.soft_expired_eval_ids. The host then keeps only a
// transport deadline of its own (see EvalResults). Honoured only when the worker echoes Ready.worker_soft_clock.
bool worker_soft_clock = 11;
+ // Test seam (lmxopcua#789): the worker's reader stops reading for good at the first DiagnosticsRequest while its
+ // heartbeats go on, so the host's write-stall check can be exercised against a real worker. Never set in production.
+ bool test_stall_reader_on_diag = 12;
}
message Ready {
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 9291e3ed2..78d97121d 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
@@ -78,14 +78,21 @@ 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.
+ /// The stream both ends drive a connected worker pipe through, with blocking I/O on dedicated threads
+ /// (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; so the pipe itself
+ /// is used, only ever synchronously. On Windows an overlapped pipe completes even its synchronous calls through the
+ /// thread pool's completion port, so the pipe is wrapped in a , which waits
+ /// for its own overlapped I/O without the port and keeps reads and writes concurrent.
///
- public static bool UseBlockingIo => !OperatingSystem.IsWindows();
+ /// The connected pipe (opened with ).
+ /// The stream to read and write frames through; disposing it closes the pipe.
+ public static Stream BlockingStream(System.IO.Pipes.PipeStream pipe)
+ {
+ ArgumentNullException.ThrowIfNull(pipe);
+ return OperatingSystem.IsWindows() ? new WindowsBlockingPipeStream(pipe) : pipe;
+ }
/// Writes one frame synchronously (the blocking-I/O handshake). Not thread-safe.
/// The pipe.
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
index 260be8bbf..a49357986 100644
--- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ScriptWorkerFramePipe.cs
@@ -122,10 +122,12 @@ public sealed class ScriptWorkerFrameReader
///
///
/// 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
+/// frames queued meanwhile wait for it. Synchronous mode (the worker pipe, through
+/// , so no frame needs the socket engine or the completion port):
+/// 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.
+/// killed by its supervisor (the host's write-stall check, ; on the worker side the
+/// host's silence 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
@@ -140,6 +142,9 @@ public sealed class ScriptWorkerFrameWriter
private int _faulted;
private long _seq;
+ /// when the write in progress started; 0 = none.
+ private long _writeStarted;
+
/// Initializes the writer.
/// The pipe.
/// Called once when a write fails (or a frame is refused); the writer is then completed.
@@ -161,6 +166,14 @@ public sealed class ScriptWorkerFrameWriter
/// The negotiated frame cap.
public int MaxFrameBytes { get; set; }
+ ///
+ /// How long the pipe write in progress has been blocked, or null when no write is in progress. A blocking write
+ /// stays in progress while the pipe is full, i.e. while the peer is not reading: the supervisor uses this to end a
+ /// peer that keeps talking but stopped listening (lmxopcua#789), which releases the thread blocked here.
+ ///
+ public TimeSpan? WriteInProgressFor
+ => Volatile.Read(ref _writeStarted) is var started and not 0 ? System.Diagnostics.Stopwatch.GetElapsedTime(started) : null;
+
/// Whether has been called or a write failed.
public bool IsCompleted => Volatile.Read(ref _completed) != 0;
@@ -195,8 +208,16 @@ public sealed class ScriptWorkerFrameWriter
FillPass();
if (_pass.WrittenCount > 0)
{
- if (_synchronous) _stream.Write(_pass.WrittenSpan);
- else await _stream.WriteAsync(_pass.WrittenMemory).ConfigureAwait(false);
+ Volatile.Write(ref _writeStarted, System.Diagnostics.Stopwatch.GetTimestamp());
+ try
+ {
+ if (_synchronous) _stream.Write(_pass.WrittenSpan);
+ else await _stream.WriteAsync(_pass.WrittenMemory).ConfigureAwait(false);
+ }
+ finally
+ {
+ Volatile.Write(ref _writeStarted, 0);
+ }
_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.
diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/WindowsBlockingPipeStream.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/WindowsBlockingPipeStream.cs
new file mode 100644
index 000000000..47ee80d6b
--- /dev/null
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/WindowsBlockingPipeStream.cs
@@ -0,0 +1,307 @@
+using System.ComponentModel;
+using System.IO.Pipes;
+using System.Runtime.InteropServices;
+using System.Runtime.Versioning;
+using Microsoft.Win32.SafeHandles;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol;
+
+///
+/// Blocking, full-duplex I/O on an overlapped Windows pipe handle that never touches the I/O completion port
+/// (lmxopcua#789).
+///
+///
+/// A .NET pipe opened with is bound to the thread pool's completion
+/// port, and even its synchronous Read/Write complete through it: every frame costs a completion-port
+/// packet plus a thread-pool work item, and the pool's spin-waiting between those items was most of the host's CPU
+/// per evaluation on Windows (+115 points over in process at 10k evals/s). A handle opened for synchronous I/O
+/// would avoid the port but serialises a read against a write on the same file object, which a full-duplex pipe
+/// with a blocked reader cannot afford.
+/// So this stream issues its own overlapped ReadFile/WriteFile on the pipe's handle with an
+/// OVERLAPPED.hEvent whose low-order bit is set — which tells the kernel not to queue a completion packet
+/// (documented on GetQueuedCompletionStatus) — and waits on that event on the calling thread. Reads and
+/// writes stay concurrent (one read and one write at a time, which is how the frame reader and the combining writer
+/// use a pipe), and a frame wakes the waiting thread straight from the kernel: the Windows twin of the Unix
+/// blocking socket ().
+/// Disposing cancels any read or write still waiting (CancelIoEx) and then closes the pipe; the waiter
+/// sees end of stream (a read) or an . A peer that goes away completes a
+/// pending read as end of stream and fails a pending write with an , as
+/// does.
+///
+[SupportedOSPlatform("windows")]
+public sealed unsafe class WindowsBlockingPipeStream : Stream
+{
+ private const int ErrorHandleEof = 38;
+ private const int ErrorBrokenPipe = 109;
+ private const int ErrorNoData = 232;
+ private const int ErrorPipeNotConnected = 233;
+ private const int ErrorOperationAborted = 995;
+ private const int ErrorIoPending = 997;
+
+ private const uint StatusSuccess = 0;
+ private const uint StatusEndOfFile = 0xC0000011;
+ private const uint StatusPipeDisconnected = 0xC00000B0;
+ private const uint StatusPipeClosing = 0xC00000B1;
+ private const uint StatusPipeBroken = 0xC000014B;
+ private const uint StatusCancelled = 0xC0000120;
+
+ private readonly PipeStream _pipe;
+ private readonly SafePipeHandle _handle;
+ private readonly Operation _read;
+ private readonly Operation _write;
+
+ /// The owner's reference plus one per operation in progress; the native state is freed when it reaches 0.
+ private int _refs = 1;
+ private int _disposed;
+
+ /// Wraps a connected pipe; the wrapper owns it from here on (disposing the wrapper closes it).
+ /// A connected pipe opened with .
+ public WindowsBlockingPipeStream(PipeStream pipe)
+ {
+ ArgumentNullException.ThrowIfNull(pipe);
+ if (!pipe.IsAsync) throw new ArgumentException("the pipe must be opened for overlapped (asynchronous) I/O", nameof(pipe));
+ _pipe = pipe;
+ _handle = pipe.SafePipeHandle;
+ _read = new Operation();
+ _write = new Operation();
+ }
+
+ ///
+ public override bool CanRead => Volatile.Read(ref _disposed) == 0;
+
+ ///
+ public override bool CanWrite => Volatile.Read(ref _disposed) == 0;
+
+ ///
+ public override bool CanSeek => false;
+
+ ///
+ public override long Length => throw new NotSupportedException();
+
+ ///
+ public override long Position
+ {
+ get => throw new NotSupportedException();
+ set => throw new NotSupportedException();
+ }
+
+ ///
+ public override int Read(byte[] buffer, int offset, int count)
+ {
+ ValidateBufferArguments(buffer, offset, count);
+ return Read(buffer.AsSpan(offset, count));
+ }
+
+ ///
+ /// The bytes read; 0 at end of stream (the peer closed or went away).
+ public override int Read(Span buffer)
+ {
+ if (buffer.IsEmpty) return 0;
+ Enter();
+ try
+ {
+ fixed (byte* p = buffer)
+ {
+ var ov = _read.Prepare();
+ var started = ReadFile(_handle, p, buffer.Length, IntPtr.Zero, ov);
+ var (status, bytes) = _read.Finish(started, out var error);
+ if (error != 0)
+ {
+ return error is ErrorBrokenPipe or ErrorPipeNotConnected or ErrorNoData or ErrorHandleEof
+ ? 0
+ : throw Failure(error, "read");
+ }
+ return status switch
+ {
+ StatusSuccess => bytes,
+ StatusEndOfFile or StatusPipeBroken or StatusPipeDisconnected or StatusPipeClosing => 0,
+ _ => throw StatusFailure(status, "read"),
+ };
+ }
+ }
+ finally
+ {
+ Exit();
+ }
+ }
+
+ ///
+ public override void Write(byte[] buffer, int offset, int count)
+ {
+ ValidateBufferArguments(buffer, offset, count);
+ Write(new ReadOnlySpan(buffer, offset, count));
+ }
+
+ ///
+ public override void Write(ReadOnlySpan buffer)
+ {
+ Enter();
+ try
+ {
+ fixed (byte* start = buffer)
+ {
+ var done = 0;
+ while (done < buffer.Length)
+ {
+ var ov = _write.Prepare();
+ var started = WriteFile(_handle, start + done, buffer.Length - done, IntPtr.Zero, ov);
+ var (status, bytes) = _write.Finish(started, out var error);
+ if (error != 0) throw Failure(error, "write");
+ if (status != StatusSuccess) throw StatusFailure(status, "write");
+ if (bytes <= 0) throw new IOException("pipe write made no progress");
+ done += bytes;
+ }
+ }
+ }
+ finally
+ {
+ Exit();
+ }
+ }
+
+ /// A no-op: a pipe has no write buffer to flush.
+ public override void Flush() { }
+
+ /// A no-op, completed: a pipe has no write buffer to flush.
+ /// Unused.
+ /// A completed task.
+ public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+
+ // The asynchronous surface is refused: Stream's base versions run the blocking call on the pool behind ONE semaphore
+ // shared by reads and writes, which would serialise a write behind a read that is waiting for the peer — the very
+ // deadlock this stream exists to avoid. Callers use the blocking calls on their own threads.
+
+ ///
+ public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => throw AsyncRefused();
+
+ ///
+ public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => throw AsyncRefused();
+
+ ///
+ public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => throw AsyncRefused();
+
+ ///
+ public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) => throw AsyncRefused();
+
+ ///
+ public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) => throw AsyncRefused();
+
+ ///
+ public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) => throw AsyncRefused();
+
+ private static NotSupportedException AsyncRefused()
+ => new("WindowsBlockingPipeStream is blocking only: read and write on dedicated threads (lmxopcua#789)");
+
+ ///
+ public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+
+ ///
+ public override void SetLength(long value) => throw new NotSupportedException();
+
+ ///
+ protected override void Dispose(bool disposing)
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) == 0)
+ {
+ // Wake a reader or writer still waiting on the kernel before the handle goes: it then sees cancellation
+ // (reported as disposed), and the pipe's own close cannot race an operation that is still being issued.
+ try
+ {
+ var add = false;
+ _handle.DangerousAddRef(ref add);
+ try { _ = CancelIoEx(_handle.DangerousGetHandle(), IntPtr.Zero); }
+ finally { if (add) _handle.DangerousRelease(); }
+ }
+ catch (ObjectDisposedException) { }
+ _pipe.Dispose();
+ Exit();
+ }
+ base.Dispose(disposing);
+ }
+
+ private void Enter()
+ {
+ while (true)
+ {
+ var refs = Volatile.Read(ref _refs);
+ if (refs == 0 || Volatile.Read(ref _disposed) != 0) throw new ObjectDisposedException(nameof(WindowsBlockingPipeStream));
+ if (Interlocked.CompareExchange(ref _refs, refs + 1, refs) == refs) return;
+ }
+ }
+
+ private void Exit()
+ {
+ if (Interlocked.Decrement(ref _refs) != 0) return;
+ _read.Free();
+ _write.Free();
+ }
+
+ private Exception Failure(int error, string what)
+ => error == ErrorOperationAborted && Volatile.Read(ref _disposed) != 0
+ ? new ObjectDisposedException(nameof(WindowsBlockingPipeStream))
+ : new IOException($"pipe {what} failed: {new Win32Exception(error).Message}", new Win32Exception(error));
+
+ private Exception StatusFailure(uint status, string what)
+ => status == StatusCancelled && Volatile.Read(ref _disposed) != 0
+ ? new ObjectDisposedException(nameof(WindowsBlockingPipeStream))
+ : new IOException($"pipe {what} failed: NTSTATUS 0x{status:X8}");
+
+ /// One direction's native state: its OVERLAPPED and the manual-reset event it completes on.
+ private sealed class Operation
+ {
+ private readonly ManualResetEvent _done = new(false);
+ private readonly NativeOverlapped* _ov = (NativeOverlapped*)NativeMemory.AllocZeroed((nuint)sizeof(NativeOverlapped));
+ private readonly IntPtr _event;
+
+ public Operation()
+ {
+ // Low-order bit set: the completion is NOT queued to the completion port the pipe is bound to.
+ _event = _done.SafeWaitHandle.DangerousGetHandle() | 1;
+ }
+
+ public NativeOverlapped* Prepare()
+ {
+ *_ov = default;
+ _ov->EventHandle = _event;
+ return _ov;
+ }
+
+ /// Waits for the operation issued and returns its NTSTATUS and byte count.
+ /// What ReadFile / WriteFile returned.
+ /// The Win32 error when the call failed outright (not pending); 0 otherwise.
+ public (uint Status, int Bytes) Finish(bool started, out int error)
+ {
+ error = 0;
+ if (!started)
+ {
+ var last = Marshal.GetLastPInvokeError();
+ if (last != ErrorIoPending)
+ {
+ error = last;
+ return (0, 0);
+ }
+ // The kernel sets the event when the I/O completes (Read/WriteFile reset it when they start it).
+ _done.WaitOne();
+ }
+ return ((uint)(long)_ov->InternalLow, (int)(long)_ov->InternalHigh);
+ }
+
+ public void Free()
+ {
+ NativeMemory.Free(_ov);
+ _done.Dispose();
+ }
+ }
+
+ [DllImport("kernel32.dll", SetLastError = true)]
+ [return: MarshalAs(UnmanagedType.Bool)]
+ private static extern bool ReadFile(SafePipeHandle handle, byte* buffer, int count, IntPtr bytesRead, NativeOverlapped* overlapped);
+
+ [DllImport("kernel32.dll", SetLastError = true)]
+ [return: MarshalAs(UnmanagedType.Bool)]
+ private static extern bool WriteFile(SafePipeHandle handle, byte* buffer, int count, IntPtr bytesWritten, NativeOverlapped* overlapped);
+
+ [DllImport("kernel32.dll", SetLastError = true)]
+ [return: MarshalAs(UnmanagedType.Bool)]
+ private static extern bool CancelIoEx(IntPtr handle, IntPtr overlapped);
+}
diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.csproj b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.csproj
index ddfa49b67..103affaaf 100644
--- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.csproj
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol.csproj
@@ -5,6 +5,8 @@
true
$(NoWarn);CS1591
+
+ true