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..4d3c321c2 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
@@ -138,7 +138,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 +175,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 +196,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");
}
@@ -312,6 +316,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;
@@ -381,11 +386,10 @@ internal sealed class WorkerConnection : IAsyncDisposable
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 +401,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 +435,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 +553,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);
}
@@ -785,11 +789,9 @@ 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()
{
@@ -810,25 +812,6 @@ internal sealed class WorkerConnection : IAsyncDisposable
_ = 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);
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..d7ac54f49 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,8 +122,9 @@ 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.
/// A failed write completes the writer and reports the exception once through onFault.
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..952a7b08e
--- /dev/null
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/WindowsBlockingPipeStream.cs
@@ -0,0 +1,277 @@
+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() { }
+
+ ///
+ 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