perf(scripting): #789 Windows worker pipe I/O blocks on dedicated threads, off the completion port

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 (and the pool's spin-wait). New WindowsBlockingPipeStream issues its own overlapped ReadFile/WriteFile with the OVERLAPPED.hEvent low bit set (no completion packet) and waits on that event on the calling thread; reads and writes stay concurrent, unlike a synchronous handle. ScriptWorkerFrameCodec.BlockingStream picks it on Windows and the pipe itself on Unix, so both ends now use the dedicated blocking reader + synchronous combining writer on every OS (UseBlockingIo and the async read loops are gone). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
2026-11-11 19:11:10 -05:00
parent f109e5f29e
commit 3c0e5ea4d7
7 changed files with 328 additions and 83 deletions
@@ -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;
+
+/// <summary>
+///     Blocking, full-duplex I/O on an overlapped Windows pipe handle that never touches the I/O completion port
+///     (lmxopcua#789).
+/// </summary>
+/// <remarks>
+///     <para>A .NET pipe opened with <see cref="PipeOptions.Asynchronous"/> is bound to the thread pool's completion
+///     port, and even its synchronous <c>Read</c>/<c>Write</c> 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.</para>
+///     <para>So this stream issues its own overlapped <c>ReadFile</c>/<c>WriteFile</c> on the pipe's handle with an
+///     <c>OVERLAPPED.hEvent</c> whose low-order bit is set — which tells the kernel not to queue a completion packet
+///     (documented on <c>GetQueuedCompletionStatus</c>) — 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 (<see cref="ScriptWorkerFrameCodec.BlockingStream"/>).</para>
+///     <para>Disposing cancels any read or write still waiting (<c>CancelIoEx</c>) and then closes the pipe; the waiter
+///     sees end of stream (a read) or an <see cref="ObjectDisposedException"/>. A peer that goes away completes a
+///     pending read as end of stream and fails a pending write with an <see cref="IOException"/>, as
+///     <see cref="PipeStream"/> does.</para>
+/// </remarks>
+[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;
+
+    /// <summary>The owner's reference plus one per operation in progress; the native state is freed when it reaches 0.</summary>
+    private int _refs = 1;
+    private int _disposed;
+
+    /// <summary>Wraps a connected pipe; the wrapper owns it from here on (disposing the wrapper closes it).</summary>
+    /// <param name="pipe">A connected pipe opened with <see cref="PipeOptions.Asynchronous"/>.</param>
+    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();
+    }
+
+    /// <inheritdoc />
+    public override bool CanRead => Volatile.Read(ref _disposed) == 0;
+
+    /// <inheritdoc />
+    public override bool CanWrite => Volatile.Read(ref _disposed) == 0;
+
+    /// <inheritdoc />
+    public override bool CanSeek => false;
+
+    /// <inheritdoc />
+    public override long Length => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    public override long Position
+    {
+        get => throw new NotSupportedException();
+        set => throw new NotSupportedException();
+    }
+
+    /// <inheritdoc />
+    public override int Read(byte[] buffer, int offset, int count)
+    {
+        ValidateBufferArguments(buffer, offset, count);
+        return Read(buffer.AsSpan(offset, count));
+    }
+
+    /// <inheritdoc />
+    /// <returns>The bytes read; 0 at end of stream (the peer closed or went away).</returns>
+    public override int Read(Span<byte> 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();
+        }
+    }
+
+    /// <inheritdoc />
+    public override void Write(byte[] buffer, int offset, int count)
+    {
+        ValidateBufferArguments(buffer, offset, count);
+        Write(new ReadOnlySpan<byte>(buffer, offset, count));
+    }
+
+    /// <inheritdoc />
+    public override void Write(ReadOnlySpan<byte> 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();
+        }
+    }
+
+    /// <summary>A no-op: a pipe has no write buffer to flush.</summary>
+    public override void Flush() { }
+
+    /// <inheritdoc />
+    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    public override void SetLength(long value) => throw new NotSupportedException();
+
+    /// <inheritdoc />
+    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}");
+
+    /// <summary>One direction's native state: its OVERLAPPED and the manual-reset event it completes on.</summary>
+    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;
+        }
+
+        /// <summary>Waits for the operation <paramref name="started"/> issued and returns its NTSTATUS and byte count.</summary>
+        /// <param name="started">What ReadFile / WriteFile returned.</param>
+        /// <param name="error">The Win32 error when the call failed outright (not pending); 0 otherwise.</param>
+        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);
+}