lmxopcua/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Protocol/WindowsBlockingPipeStream.cs

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() { }

    /// <summary>A no-op, completed: a pipe has no write buffer to flush.</summary>
    /// <param name="cancellationToken">Unused.</param>
    /// <returns>A completed task.</returns>
    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.

    /// <inheritdoc />
    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => throw AsyncRefused();

    /// <inheritdoc />
    public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) => throw AsyncRefused();

    /// <inheritdoc />
    public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => throw AsyncRefused();

    /// <inheritdoc />
    public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default) => throw AsyncRefused();

    /// <inheritdoc />
    public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) => throw AsyncRefused();

    /// <inheritdoc />
    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)");

    /// <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);
}